feat(payment_v2): implement payments sync (#6464)

Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>

crates/storage_impl/src/mock_db/payment_attempt.rs

@@ -1,6 +1,6 @@
 use common_utils::errors::CustomResult;
 #[cfg(feature = "v2")]
-use common_utils::types::keymanager::KeyManagerState;
+use common_utils::{id_type, types::keymanager::KeyManagerState};
 use diesel_models::enums as storage_enums;
 #[cfg(feature = "v2")]
 use hyperswitch_domain_models::merchant_key_store::MerchantKeyStore;
@@ -74,7 +74,7 @@ impl PaymentAttemptInterface for MockDb {
         &self,
         _key_manager_state: &KeyManagerState,
         _merchant_key_store: &MerchantKeyStore,
-        _attempt_id: &str,
+        _attempt_id: &id_type::GlobalAttemptId,
         _storage_scheme: storage_enums::MerchantStorageScheme,
     ) -> error_stack::Result<PaymentAttempt, StorageError> {
         // [#172]: Implement function for `MockDb`

crates/storage_impl/src/mock_db/payment_intent.rs

@@ -95,6 +95,7 @@ impl PaymentIntentInterface for MockDb {
         Ok(new)
     }
 
+    #[cfg(feature = "v1")]
     // safety: only used for testing
     #[allow(clippy::unwrap_used)]
     async fn update_payment_intent(
@@ -130,6 +131,20 @@ impl PaymentIntentInterface for MockDb {
         Ok(payment_intent.clone())
     }
 
+    #[cfg(feature = "v2")]
+    // safety: only used for testing
+    #[allow(clippy::unwrap_used)]
+    async fn update_payment_intent(
+        &self,
+        state: &KeyManagerState,
+        this: PaymentIntent,
+        update: PaymentIntentUpdate,
+        key_store: &MerchantKeyStore,
+        _storage_scheme: storage_enums::MerchantStorageScheme,
+    ) -> CustomResult<PaymentIntent, StorageError> {
+        todo!()
+    }
+
     #[cfg(feature = "v1")]
     // safety: only used for testing
     #[allow(clippy::unwrap_used)]

crates/storage_impl/src/payments/payment_attempt.rs

@@ -375,7 +375,7 @@ impl<T: DatabaseStore> PaymentAttemptInterface for RouterStore<T> {
         &self,
         key_manager_state: &KeyManagerState,
         merchant_key_store: &MerchantKeyStore,
-        attempt_id: &str,
+        attempt_id: &common_utils::id_type::GlobalAttemptId,
         _storage_scheme: MerchantStorageScheme,
     ) -> error_stack::Result<PaymentAttempt, errors::StorageError> {
         let conn = pg_connection_read(self).await?;
@@ -1132,7 +1132,7 @@ impl<T: DatabaseStore> PaymentAttemptInterface for KVRouterStore<T> {
         &self,
         key_manager_state: &KeyManagerState,
         merchant_key_store: &MerchantKeyStore,
-        attempt_id: &str,
+        attempt_id: &common_utils::id_type::GlobalAttemptId,
         storage_scheme: MerchantStorageScheme,
     ) -> error_stack::Result<PaymentAttempt, errors::StorageError> {
         // Ignoring storage scheme for v2 implementation

crates/storage_impl/src/payments/payment_intent.rs

@@ -495,6 +495,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
         .change_context(StorageError::DecryptionError)
     }
 
+    #[cfg(feature = "v1")]
     #[instrument(skip_all)]
     async fn update_payment_intent(
         &self,
@@ -528,6 +529,41 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
         .change_context(StorageError::DecryptionError)
     }
 
+    #[cfg(feature = "v2")]
+    #[instrument(skip_all)]
+    async fn update_payment_intent(
+        &self,
+        state: &KeyManagerState,
+        this: PaymentIntent,
+        payment_intent: PaymentIntentUpdate,
+        merchant_key_store: &MerchantKeyStore,
+        _storage_scheme: MerchantStorageScheme,
+    ) -> error_stack::Result<PaymentIntent, StorageError> {
+        let conn = pg_connection_write(self).await?;
+        let diesel_payment_intent_update =
+            diesel_models::PaymentIntentUpdateInternal::from(payment_intent);
+
+        let diesel_payment_intent = this
+            .convert()
+            .await
+            .change_context(StorageError::EncryptionError)?
+            .update(&conn, diesel_payment_intent_update)
+            .await
+            .map_err(|er| {
+                let new_err = diesel_error_to_data_error(er.current_context());
+                er.change_context(new_err)
+            })?;
+
+        PaymentIntent::convert_back(
+            state,
+            diesel_payment_intent,
+            merchant_key_store.key.get_inner(),
+            merchant_key_store.merchant_id.clone().into(),
+        )
+        .await
+        .change_context(StorageError::DecryptionError)
+    }
+
     #[cfg(feature = "v1")]
     #[instrument(skip_all)]
     async fn find_payment_intent_by_payment_id_merchant_id(