feat(core): Add support for v2 payments get intent using merchant reference id (#7123)

Co-authored-by: Chikke Srujan <chikke.srujan@Chikke-Srujan-N7WRTY72X7.local>
Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>

crates/diesel_models/src/query/payment_intent.rs

@@ -87,6 +87,23 @@ impl PaymentIntent {
         .await
     }
 
+    // This query should be removed in the future because direct queries to the intent table without an intent ID are not allowed.
+    // In an active-active setup, a lookup table should be implemented, and the merchant reference ID will serve as the idempotency key.
+    #[cfg(feature = "v2")]
+    pub async fn find_by_merchant_reference_id_profile_id(
+        conn: &PgPooledConn,
+        merchant_reference_id: &common_utils::id_type::PaymentReferenceId,
+        profile_id: &common_utils::id_type::ProfileId,
+    ) -> StorageResult<Self> {
+        generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::profile_id
+                .eq(profile_id.to_owned())
+                .and(dsl::merchant_reference_id.eq(merchant_reference_id.to_owned())),
+        )
+        .await
+    }
+
     #[cfg(feature = "v1")]
     pub async fn find_by_payment_id_merchant_id(
         conn: &PgPooledConn,

crates/hyperswitch_domain_models/src/payments/payment_intent.rs

@@ -63,6 +63,15 @@ pub trait PaymentIntentInterface {
         merchant_key_store: &MerchantKeyStore,
         storage_scheme: common_enums::MerchantStorageScheme,
     ) -> error_stack::Result<PaymentIntent, errors::StorageError>;
+    #[cfg(feature = "v2")]
+    async fn find_payment_intent_by_merchant_reference_id_profile_id(
+        &self,
+        state: &KeyManagerState,
+        merchant_reference_id: &id_type::PaymentReferenceId,
+        profile_id: &id_type::ProfileId,
+        merchant_key_store: &MerchantKeyStore,
+        storage_scheme: &common_enums::MerchantStorageScheme,
+    ) -> error_stack::Result<PaymentIntent, errors::StorageError>;
 
     #[cfg(feature = "v2")]
     async fn find_payment_intent_by_id(

crates/router/src/core/payments.rs

@@ -1525,6 +1525,70 @@ where
     )
 }
 
+#[cfg(feature = "v2")]
+#[allow(clippy::too_many_arguments)]
+pub async fn payments_get_intent_using_merchant_reference(
+    state: SessionState,
+    merchant_account: domain::MerchantAccount,
+    profile: domain::Profile,
+    key_store: domain::MerchantKeyStore,
+    req_state: ReqState,
+    merchant_reference_id: &id_type::PaymentReferenceId,
+    header_payload: HeaderPayload,
+    platform_merchant_account: Option<domain::MerchantAccount>,
+) -> RouterResponse<api::PaymentsIntentResponse> {
+    let db = state.store.as_ref();
+    let storage_scheme = merchant_account.storage_scheme;
+    let key_manager_state = &(&state).into();
+    let payment_intent = db
+        .find_payment_intent_by_merchant_reference_id_profile_id(
+            key_manager_state,
+            merchant_reference_id,
+            profile.get_id(),
+            &key_store,
+            &storage_scheme,
+        )
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
+
+    let (payment_data, _req, customer) = Box::pin(payments_intent_operation_core::<
+        api::PaymentGetIntent,
+        _,
+        _,
+        PaymentIntentData<api::PaymentGetIntent>,
+    >(
+        &state,
+        req_state,
+        merchant_account.clone(),
+        profile.clone(),
+        key_store.clone(),
+        operations::PaymentGetIntent,
+        api_models::payments::PaymentsGetIntentRequest {
+            id: payment_intent.get_id().clone(),
+        },
+        payment_intent.get_id().clone(),
+        header_payload.clone(),
+        platform_merchant_account,
+    ))
+    .await?;
+
+    transformers::ToResponse::<
+        api::PaymentGetIntent,
+        PaymentIntentData<api::PaymentGetIntent>,
+        operations::PaymentGetIntent,
+    >::generate_response(
+        payment_data,
+        customer,
+        &state.base_url,
+        operations::PaymentGetIntent,
+        &state.conf.connector_request_reference_id_config,
+        None,
+        None,
+        header_payload.x_hs_latency,
+        &merchant_account,
+    )
+}
+
 #[cfg(feature = "v2")]
 #[allow(clippy::too_many_arguments)]
 pub async fn payments_core<F, Res, Req, Op, FData, D>(

crates/router/src/db/kafka_store.rs

@@ -1904,6 +1904,29 @@ impl PaymentIntentInterface for KafkaStore {
             )
             .await
     }
+
+    #[cfg(feature = "v2")]
+    async fn find_payment_intent_by_merchant_reference_id_profile_id(
+        &self,
+        state: &KeyManagerState,
+        merchant_reference_id: &id_type::PaymentReferenceId,
+        profile_id: &id_type::ProfileId,
+        merchant_key_store: &domain::MerchantKeyStore,
+        storage_scheme: &MerchantStorageScheme,
+    ) -> error_stack::Result<
+        hyperswitch_domain_models::payments::PaymentIntent,
+        errors::DataStorageError,
+    > {
+        self.diesel_store
+            .find_payment_intent_by_merchant_reference_id_profile_id(
+                state,
+                merchant_reference_id,
+                profile_id,
+                merchant_key_store,
+                storage_scheme,
+            )
+            .await
+    }
 }
 
 #[async_trait::async_trait]

crates/router/src/routes/app.rs

@@ -561,6 +561,12 @@ impl Payments {
                     .route(web::post().to(payments::payments_create_intent)),
             );
 
+        route =
+            route
+                .service(web::resource("/ref/{merchant_reference_id}").route(
+                    web::get().to(payments::payment_get_intent_using_merchant_reference_id),
+                ));
+
         route = route.service(
             web::scope("/{payment_id}")
                 .service(

crates/router/src/routes/lock_utils.rs

@@ -147,7 +147,8 @@ impl From<Flow> for ApiIdentifier {
             | Flow::PaymentsPostSessionTokens
             | Flow::PaymentsUpdateIntent
             | Flow::PaymentsCreateAndConfirmIntent
-            | Flow::PaymentStartRedirection => Self::Payments,
+            | Flow::PaymentStartRedirection
+            | Flow::PaymentsRetrieveUsingMerchantReferenceId => Self::Payments,
 
             Flow::PayoutsCreate
             | Flow::PayoutsRetrieve

crates/router/src/routes/payments.rs

@@ -2470,6 +2470,47 @@ pub async fn payment_status(
     .await
 }
 
+#[cfg(feature = "v2")]
+#[instrument(skip(state, req), fields(flow, payment_id))]
+pub async fn payment_get_intent_using_merchant_reference_id(
+    state: web::Data<app::AppState>,
+    req: actix_web::HttpRequest,
+    path: web::Path<common_utils::id_type::PaymentReferenceId>,
+) -> impl Responder {
+    let flow = Flow::PaymentsRetrieveUsingMerchantReferenceId;
+    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
+        Ok(headers) => headers,
+        Err(err) => {
+            return api::log_and_return_error_response(err);
+        }
+    };
+
+    let merchant_reference_id = path.into_inner();
+
+    Box::pin(api::server_wrap(
+        flow,
+        state,
+        &req,
+        (),
+        |state, auth: auth::AuthenticationData, _, req_state| async {
+            Box::pin(payments::payments_get_intent_using_merchant_reference(
+                state,
+                auth.merchant_account,
+                auth.profile,
+                auth.key_store,
+                req_state,
+                &merchant_reference_id,
+                header_payload.clone(),
+                auth.platform_merchant_account,
+            ))
+            .await
+        },
+        &auth::HeaderAuth(auth::ApiKeyAuth),
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
+
 #[cfg(feature = "v2")]
 #[instrument(skip_all, fields(flow = ?Flow::PaymentsRedirect, payment_id))]
 pub async fn payments_finish_redirection(

crates/router_env/src/logger/types.rs

@@ -144,6 +144,8 @@ pub enum Flow {
     PaymentsRetrieve,
     /// Payments Retrieve force sync flow.
     PaymentsRetrieveForceSync,
+    /// Payments Retrieve using merchant reference id
+    PaymentsRetrieveUsingMerchantReferenceId,
     /// Payments update flow.
     PaymentsUpdate,
     /// Payments confirm flow.

crates/storage_impl/src/mock_db/payment_intent.rs

@@ -185,4 +185,26 @@ impl PaymentIntentInterface for MockDb {
 
         Ok(payment_intent.clone())
     }
+    #[cfg(feature = "v2")]
+    async fn find_payment_intent_by_merchant_reference_id_profile_id(
+        &self,
+        _state: &KeyManagerState,
+        merchant_reference_id: &common_utils::id_type::PaymentReferenceId,
+        profile_id: &common_utils::id_type::ProfileId,
+        _merchant_key_store: &MerchantKeyStore,
+        _storage_scheme: &common_enums::MerchantStorageScheme,
+    ) -> error_stack::Result<PaymentIntent, StorageError> {
+        let payment_intents = self.payment_intents.lock().await;
+        let payment_intent = payment_intents
+            .iter()
+            .find(|payment_intent| {
+                payment_intent.merchant_reference_id.as_ref() == Some(merchant_reference_id)
+                    && payment_intent.profile_id.eq(profile_id)
+            })
+            .ok_or(StorageError::ValueNotFound(
+                "PaymentIntent not found".to_string(),
+            ))?;
+
+        Ok(payment_intent.clone())
+    }
 }

crates/storage_impl/src/payments/payment_intent.rs

@@ -459,6 +459,32 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
             )
             .await
     }
+    #[cfg(feature = "v2")]
+    async fn find_payment_intent_by_merchant_reference_id_profile_id(
+        &self,
+        state: &KeyManagerState,
+        merchant_reference_id: &common_utils::id_type::PaymentReferenceId,
+        profile_id: &common_utils::id_type::ProfileId,
+        merchant_key_store: &MerchantKeyStore,
+        storage_scheme: &MerchantStorageScheme,
+    ) -> error_stack::Result<PaymentIntent, StorageError> {
+        match storage_scheme {
+            MerchantStorageScheme::PostgresOnly => {
+                self.router_store
+                    .find_payment_intent_by_merchant_reference_id_profile_id(
+                        state,
+                        merchant_reference_id,
+                        profile_id,
+                        merchant_key_store,
+                        storage_scheme,
+                    )
+                    .await
+            }
+            MerchantStorageScheme::RedisKv => {
+                todo!()
+            }
+        }
+    }
 }
 
 #[async_trait::async_trait]
@@ -622,6 +648,39 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
         .change_context(StorageError::DecryptionError)
     }
 
+    #[cfg(feature = "v2")]
+    #[instrument(skip_all)]
+    async fn find_payment_intent_by_merchant_reference_id_profile_id(
+        &self,
+        state: &KeyManagerState,
+        merchant_reference_id: &common_utils::id_type::PaymentReferenceId,
+        profile_id: &common_utils::id_type::ProfileId,
+        merchant_key_store: &MerchantKeyStore,
+        _storage_scheme: &MerchantStorageScheme,
+    ) -> error_stack::Result<PaymentIntent, StorageError> {
+        let conn = pg_connection_read(self).await?;
+        let diesel_payment_intent = DieselPaymentIntent::find_by_merchant_reference_id_profile_id(
+            &conn,
+            merchant_reference_id,
+            profile_id,
+        )
+        .await
+        .map_err(|er| {
+            let new_err = diesel_error_to_data_error(*er.current_context());
+            er.change_context(new_err)
+        })?;
+        let merchant_id = diesel_payment_intent.merchant_id.clone();
+
+        PaymentIntent::convert_back(
+            state,
+            diesel_payment_intent,
+            merchant_key_store.key.get_inner(),
+            merchant_id.to_owned().into(),
+        )
+        .await
+        .change_context(StorageError::DecryptionError)
+    }
+
     #[cfg(all(feature = "v1", feature = "olap"))]
     #[instrument(skip_all)]
     async fn filter_payment_intent_by_constraints(