feat(router): added new webhook URL to support merchant_connector_id (#2006)

Co-authored-by: Sahkal Poddar <sahkal.poddar@juspay.in>
Co-authored-by: Narayan Bhat <narayan.bhat@juspay.in>

crates/api_models/src/admin.rs

@@ -631,7 +631,6 @@ pub struct MerchantConnectorCreate {
         }
     }))]
     pub connector_webhook_details: Option<MerchantConnectorWebhookDetails>,
-
     /// Identifier for the business profile, if not provided default will be chosen from merchant account
     pub profile_id: Option<String>,
 }

crates/router/src/connector/adyen.rs

@@ -18,7 +18,6 @@ use crate::{
         self,
         errors::{self, CustomResult},
     },
-    db::StorageInterface,
     headers, logger, routes,
     services::{
         self,
@@ -1398,23 +1397,19 @@ impl api::IncomingWebhook for Adyen {
 
     async fn verify_webhook_source(
         &self,
-        db: &dyn StorageInterface,
         request: &api::IncomingWebhookRequestDetails<'_>,
         merchant_account: &domain::MerchantAccount,
+        merchant_connector_account: domain::MerchantConnectorAccount,
         connector_label: &str,
-        key_store: &domain::MerchantKeyStore,
-        object_reference_id: api_models::webhooks::ObjectReferenceId,
     ) -> CustomResult<bool, errors::ConnectorError> {
         let signature = self
             .get_webhook_source_verification_signature(request)
             .change_context(errors::ConnectorError::WebhookSourceVerificationFailed)?;
         let connector_webhook_secrets = self
             .get_webhook_source_verification_merchant_secret(
-                db,
                 merchant_account,
                 connector_label,
-                key_store,
+                merchant_connector_account,
-                object_reference_id,
             )
             .await
             .change_context(errors::ConnectorError::WebhookSourceVerificationFailed)?;

crates/router/src/connector/cashtocode.rs

@@ -12,7 +12,6 @@ use crate::{
     configs::settings::{self},
     connector::{utils as connector_utils, utils as conn_utils},
     core::errors::{self, CustomResult},
-    db::StorageInterface,
     headers,
     services::{
         self,
@@ -331,23 +330,19 @@ impl api::IncomingWebhook for Cashtocode {
 
     async fn verify_webhook_source(
         &self,
-        db: &dyn StorageInterface,
         request: &api::IncomingWebhookRequestDetails<'_>,
         merchant_account: &domain::MerchantAccount,
+        merchant_connector_account: domain::MerchantConnectorAccount,
         connector_label: &str,
-        key_store: &domain::MerchantKeyStore,
-        object_reference_id: api_models::webhooks::ObjectReferenceId,
     ) -> CustomResult<bool, errors::ConnectorError> {
         let signature = self
             .get_webhook_source_verification_signature(request)
             .change_context(errors::ConnectorError::WebhookSourceVerificationFailed)?;
         let connector_webhook_secrets = self
             .get_webhook_source_verification_merchant_secret(
-                db,
                 merchant_account,
                 connector_label,
-                key_store,
+                merchant_connector_account,
-                object_reference_id,
             )
             .await
             .change_context(errors::ConnectorError::WebhookSourceVerificationFailed)?;

crates/router/src/connector/payme.rs

@@ -16,12 +16,12 @@ use crate::{
         errors::{self, CustomResult},
         payments,
     },
-    db, headers,
+    headers,
     services::{self, request, ConnectorIntegration, ConnectorValidation},
     types::{
         self,
         api::{self, ConnectorCommon, ConnectorCommonExt},
-        ErrorResponse, Response,
+        domain, ErrorResponse, Response,
     },
     utils::{self, BytesExt},
 };
@@ -927,12 +927,10 @@ impl api::IncomingWebhook for Payme {
 
     async fn verify_webhook_source(
         &self,
-        db: &dyn db::StorageInterface,
         request: &api::IncomingWebhookRequestDetails<'_>,
-        merchant_account: &types::domain::MerchantAccount,
+        merchant_account: &domain::MerchantAccount,
+        merchant_connector_account: domain::MerchantConnectorAccount,
         connector_label: &str,
-        key_store: &types::domain::MerchantKeyStore,
-        object_reference_id: api_models::webhooks::ObjectReferenceId,
     ) -> CustomResult<bool, errors::ConnectorError> {
         let algorithm = self
             .get_webhook_source_verification_algorithm(request)
@@ -944,11 +942,9 @@ impl api::IncomingWebhook for Payme {
 
         let connector_webhook_secrets = self
             .get_webhook_source_verification_merchant_secret(
-                db,
                 merchant_account,
                 connector_label,
-                key_store,
+                merchant_connector_account,
-                object_reference_id,
             )
             .await
             .change_context(errors::ConnectorError::WebhookSourceVerificationFailed)?;

crates/router/src/connector/paypal.rs

@@ -20,7 +20,6 @@ use crate::{
         errors::{self, CustomResult},
         payments,
     },
-    db::StorageInterface,
     headers,
     services::{
         self,
@@ -915,12 +914,10 @@ impl ConnectorIntegration<api::RSync, types::RefundsData, types::RefundsResponse
 impl api::IncomingWebhook for Paypal {
     async fn verify_webhook_source(
         &self,
-        _db: &dyn StorageInterface,
         _request: &api::IncomingWebhookRequestDetails<'_>,
         _merchant_account: &domain::MerchantAccount,
+        _merchant_connector_account: domain::MerchantConnectorAccount,
         _connector_label: &str,
-        _key_store: &domain::MerchantKeyStore,
-        _object_reference_id: api_models::webhooks::ObjectReferenceId,
     ) -> CustomResult<bool, errors::ConnectorError> {
         Ok(false) // Verify webhook source is not implemented for Paypal it requires additional apicall this function needs to be modified once we have a way to verify webhook source
     }

crates/router/src/connector/rapyd.rs

@@ -15,7 +15,6 @@ use crate::{
     connector::{utils as connector_utils, utils as conn_utils},
     consts,
     core::errors::{self, CustomResult},
-    db::StorageInterface,
     headers, logger,
     services::{
         self,
@@ -762,23 +761,19 @@ impl api::IncomingWebhook for Rapyd {
 
     async fn verify_webhook_source(
         &self,
-        db: &dyn StorageInterface,
         request: &api::IncomingWebhookRequestDetails<'_>,
         merchant_account: &domain::MerchantAccount,
+        merchant_connector_account: domain::MerchantConnectorAccount,
         connector_label: &str,
-        key_store: &domain::MerchantKeyStore,
-        object_reference_id: api_models::webhooks::ObjectReferenceId,
     ) -> CustomResult<bool, errors::ConnectorError> {
         let signature = self
             .get_webhook_source_verification_signature(request)
             .change_context(errors::ConnectorError::WebhookSourceVerificationFailed)?;
         let connector_webhook_secrets = self
             .get_webhook_source_verification_merchant_secret(
-                db,
                 merchant_account,
                 connector_label,
-                key_store,
+                merchant_connector_account,
-                object_reference_id,
             )
             .await
             .change_context(errors::ConnectorError::WebhookSourceVerificationFailed)?;

crates/router/src/connector/stax.rs

@@ -14,7 +14,6 @@ use crate::{
     configs::settings,
     consts,
     core::errors::{self, CustomResult},
-    db::StorageInterface,
     headers,
     services::{
         self,
@@ -773,12 +772,10 @@ impl ConnectorIntegration<api::RSync, types::RefundsData, types::RefundsResponse
 impl api::IncomingWebhook for Stax {
     async fn verify_webhook_source(
         &self,
-        _db: &dyn StorageInterface,
         _request: &api::IncomingWebhookRequestDetails<'_>,
         _merchant_account: &domain::MerchantAccount,
+        _merchant_connector_account: domain::MerchantConnectorAccount,
         _connector_label: &str,
-        _key_store: &domain::MerchantKeyStore,
-        _object_reference_id: api_models::webhooks::ObjectReferenceId,
     ) -> CustomResult<bool, errors::ConnectorError> {
         Ok(false)
     }

crates/router/src/connector/zen.rs

@@ -16,7 +16,6 @@ use crate::{
         errors::{self, CustomResult},
         payments,
     },
-    db::StorageInterface,
     headers,
     services::{
         self,
@@ -563,23 +562,19 @@ impl api::IncomingWebhook for Zen {
 
     async fn verify_webhook_source(
         &self,
-        db: &dyn StorageInterface,
         request: &api::IncomingWebhookRequestDetails<'_>,
         merchant_account: &domain::MerchantAccount,
+        merchant_connector_account: domain::MerchantConnectorAccount,
         connector_label: &str,
-        key_store: &domain::MerchantKeyStore,
-        object_reference_id: api_models::webhooks::ObjectReferenceId,
     ) -> CustomResult<bool, errors::ConnectorError> {
         let algorithm = self.get_webhook_source_verification_algorithm(request)?;
 
         let signature = self.get_webhook_source_verification_signature(request)?;
         let mut connector_webhook_secrets = self
             .get_webhook_source_verification_merchant_secret(
-                db,
                 merchant_account,
                 connector_label,
-                key_store,
+                merchant_connector_account,
-                object_reference_id,
             )
             .await?;
         let mut message = self.get_webhook_source_verification_message(

crates/router/src/core/webhooks.rs

@@ -24,7 +24,7 @@ use crate::{
         storage::{self, enums},
         transformers::{ForeignInto, ForeignTryInto},
     },
-    utils::{generate_id, Encode, OptionExt, ValueExt},
+    utils::{self as helper_utils, generate_id, Encode, OptionExt, ValueExt},
 };
 
 const OUTGOING_WEBHOOK_TIMEOUT_SECS: u64 = 5;
@@ -182,21 +182,19 @@ pub async fn refunds_incoming_webhook_flow<W: types::OutgoingWebhookType>(
                 .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
                 .attach_printable("failed refund status mapping from event type")?,
         };
-        state
+        db.update_refund(
-            .store
+            refund.to_owned(),
-            .update_refund(
+            refund_update,
-                refund.to_owned(),
+            merchant_account.storage_scheme,
-                refund_update,
+        )
-                merchant_account.storage_scheme,
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::WebhookResourceNotFound)
+        .attach_printable_lazy(|| {
+            format!(
+                "Failed while updating refund: refund_id: {}",
+                refund_id.to_owned()
             )
-            .await
+        })?
-            .to_not_found_response(errors::ApiErrorResponse::WebhookResourceNotFound)
-            .attach_printable_lazy(|| {
-                format!(
-                    "Failed while updating refund: refund_id: {}",
-                    refund_id.to_owned()
-                )
-            })?
     } else {
         refunds::refund_retrieve_core(
             &state,
@@ -707,7 +705,7 @@ pub async fn webhooks_core<W: types::OutgoingWebhookType>(
     req: &actix_web::HttpRequest,
     merchant_account: domain::MerchantAccount,
     key_store: domain::MerchantKeyStore,
-    connector_name: &str,
+    connector_name_or_mca_id: &str,
     body: actix_web::web::Bytes,
 ) -> RouterResponse<serde_json::Value> {
     metrics::WEBHOOK_INCOMING_COUNT.add(
@@ -718,18 +716,6 @@ pub async fn webhooks_core<W: types::OutgoingWebhookType>(
             merchant_account.merchant_id.clone(),
         )],
     );
-
-    let connector = api::ConnectorData::get_connector_by_name(
-        &state.conf.connectors,
-        connector_name,
-        api::GetToken::Connector,
-    )
-    .change_context(errors::ApiErrorResponse::InvalidRequestData {
-        message: "invalid connnector name received".to_string(),
-    })
-    .attach_printable("Failed construction of ConnectorData")?;
-
-    let connector = connector.connector;
     let mut request_details = api::IncomingWebhookRequestDetails {
         method: req.method().clone(),
         uri: req.uri().clone(),
@@ -738,6 +724,19 @@ pub async fn webhooks_core<W: types::OutgoingWebhookType>(
         body: &body,
     };
 
+    let (merchant_connector_account, connector) = fetch_mca_and_connector(
+        state,
+        &merchant_account,
+        connector_name_or_mca_id,
+        &key_store,
+        &request_details,
+    )
+    .await?;
+
+    let connector_name = merchant_connector_account.clone().connector_name;
+
+    let connector = connector.connector;
+
     let decoded_body = connector
         .decode_webhook_body(
             &*state.store,
@@ -784,7 +783,7 @@ pub async fn webhooks_core<W: types::OutgoingWebhookType>(
 
     let process_webhook_further = utils::lookup_webhook_event(
         &*state.store,
-        connector_name,
+        connector_name.as_str(),
         &merchant_account.merchant_id,
         &event_type,
     )
@@ -802,12 +801,10 @@ pub async fn webhooks_core<W: types::OutgoingWebhookType>(
 
         let source_verified = connector
             .verify_webhook_source(
-                &*state.store,
                 &request_details,
                 &merchant_account,
-                connector_name,
+                merchant_connector_account.clone(),
-                &key_store,
+                connector_name.as_str(),
-                object_ref_id.clone(),
             )
             .await
             .or_else(|error| match error.current_context() {
@@ -863,7 +860,7 @@ pub async fn webhooks_core<W: types::OutgoingWebhookType>(
                 merchant_account,
                 key_store,
                 webhook_details,
-                connector_name,
+                connector_name.as_str(),
                 source_verified,
                 event_type,
             )
@@ -916,3 +913,85 @@ pub async fn webhooks_core<W: types::OutgoingWebhookType>(
 
     Ok(response)
 }
+
+async fn fetch_mca_and_connector(
+    state: &AppState,
+    merchant_account: &domain::MerchantAccount,
+    connector_name_or_mca_id: &str,
+    key_store: &domain::MerchantKeyStore,
+    request_details: &api::IncomingWebhookRequestDetails<'_>,
+) -> CustomResult<(domain::MerchantConnectorAccount, api::ConnectorData), errors::ApiErrorResponse>
+{
+    let db = &state.store;
+    if connector_name_or_mca_id.starts_with("mca_") {
+        let mca = db
+            .find_by_merchant_connector_account_merchant_id_merchant_connector_id(
+                &merchant_account.merchant_id,
+                connector_name_or_mca_id,
+                key_store,
+            )
+            .await
+            .to_not_found_response(errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+                id: connector_name_or_mca_id.to_string(),
+            })
+            .attach_printable(
+                "error while fetching merchant_connector_account from connector_id",
+            )?;
+
+        let connector = api::ConnectorData::get_connector_by_name(
+            &state.conf.connectors,
+            &mca.connector_name,
+            api::GetToken::Connector,
+        )
+        .change_context(errors::ApiErrorResponse::InvalidRequestData {
+            message: "invalid connector name received".to_string(),
+        })
+        .attach_printable("Failed construction of ConnectorData")?;
+
+        Ok((mca, connector))
+    } else {
+        let connector = api::ConnectorData::get_connector_by_name(
+            &state.conf.connectors,
+            connector_name_or_mca_id,
+            api::GetToken::Connector,
+        )
+        .change_context(errors::ApiErrorResponse::InvalidRequestData {
+            message: "invalid connector name received".to_string(),
+        })
+        .attach_printable("Failed construction of ConnectorData")?;
+
+        let object_ref_id = connector
+            .connector
+            .get_webhook_object_reference_id(request_details)
+            .switch()
+            .attach_printable("Could not find object reference id in incoming webhook body")?;
+
+        let profile_id = helper_utils::get_profile_id_using_object_reference_id(
+            &*state.store,
+            object_ref_id,
+            merchant_account,
+            connector_name_or_mca_id,
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::InvalidDataValue {
+            field_name: "object reference id",
+        })
+        .attach_printable("Could not find profile id from object reference id")?;
+
+        let mca = db
+            .find_merchant_connector_account_by_profile_id_connector_name(
+                &profile_id,
+                connector_name_or_mca_id,
+                key_store,
+            )
+            .await
+            .to_not_found_response(errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+                id: format!(
+                    "profile_id {profile_id} and connector name {connector_name_or_mca_id}"
+                ),
+            })
+            .attach_printable("error while fetching merchant_connector_account from profile_id")?;
+
+        Ok((mca, connector))
+    }
+}

crates/router/src/routes/app.rs

@@ -452,7 +452,7 @@ impl Webhooks {
         web::scope("/webhooks")
             .app_data(web::Data::new(config))
             .service(
-                web::resource("/{merchant_id}/{connector_name}")
+                web::resource("/{merchant_id}/{connector_id_or_name}")
                     .route(
                         web::post().to(receive_incoming_webhook::<webhook_type::OutgoingWebhook>),
                     )

crates/router/src/routes/webhooks.rs

@@ -15,7 +15,7 @@ pub async fn receive_incoming_webhook<W: types::OutgoingWebhookType>(
     path: web::Path<(String, String)>,
 ) -> impl Responder {
     let flow = Flow::IncomingWebhookReceive;
-    let (merchant_id, connector_name) = path.into_inner();
+    let (merchant_id, connector_id_or_name) = path.into_inner();
 
     api::server_wrap(
         flow,
@@ -28,7 +28,7 @@ pub async fn receive_incoming_webhook<W: types::OutgoingWebhookType>(
                 &req,
                 auth.merchant_account,
                 auth.key_store,
-                &connector_name,
+                &connector_id_or_name,
                 body,
             )
         },

crates/router/src/types/api/webhooks.rs

@@ -11,9 +11,9 @@ use super::ConnectorCommon;
 use crate::{
     core::errors::{self, CustomResult},
     db::StorageInterface,
-    logger, services,
+    services,
     types::domain,
-    utils::{self, crypto},
+    utils::crypto,
 };
 
 pub struct IncomingWebhookRequestDetails<'a> {
@@ -78,11 +78,9 @@ pub trait IncomingWebhook: ConnectorCommon + Sync {
 
     async fn get_webhook_source_verification_merchant_secret(
         &self,
-        db: &dyn StorageInterface,
         merchant_account: &domain::MerchantAccount,
         connector_name: &str,
-        key_store: &domain::MerchantKeyStore,
+        merchant_connector_account: domain::MerchantConnectorAccount,
-        object_reference_id: ObjectReferenceId,
     ) -> CustomResult<api_models::webhooks::ConnectorWebhookSecrets, errors::ConnectorError> {
         let merchant_id = merchant_account.merchant_id.as_str();
         let debug_suffix = format!(
@@ -90,71 +88,39 @@ pub trait IncomingWebhook: ConnectorCommon + Sync {
             merchant_id, connector_name
         );
         let default_secret = "default_secret".to_string();
-        let profile_id = utils::get_profile_id_using_object_reference_id(
+        let merchant_secret = match merchant_connector_account.connector_webhook_details {
-            db,
+            Some(merchant_connector_webhook_details) => {
-            object_reference_id,
+                let connector_webhook_details = merchant_connector_webhook_details
-            merchant_account,
+                    .parse_value::<MerchantConnectorWebhookDetails>(
-            connector_name,
+                        "MerchantConnectorWebhookDetails",
-        )
+                    )
-        .await
+                    .change_context_lazy(|| errors::ConnectorError::WebhookSourceVerificationFailed)
-        .change_context(errors::ConnectorError::WebhookSourceVerificationFailed)
+                    .attach_printable_lazy(|| {
-        .attach_printable("Error while fetching business_profile")?;
+                        format!(
-
+                            "Deserializing MerchantConnectorWebhookDetails failed {}",
-        let merchant_connector_account_result = db
+                            debug_suffix
-            .find_merchant_connector_account_by_profile_id_connector_name(
-                &profile_id,
-                connector_name,
-                key_store,
-            )
-            .await;
-
-        let connector_webhook_secrets = match merchant_connector_account_result {
-            Ok(mca) => match mca.connector_webhook_details {
-                Some(merchant_connector_webhook_details) => {
-                    let connector_webhook_details = merchant_connector_webhook_details
-                        .parse_value::<MerchantConnectorWebhookDetails>(
-                            "MerchantConnectorWebhookDetails",
                         )
-                        .change_context_lazy(|| {
+                    })?;
-                            errors::ConnectorError::WebhookSourceVerificationFailed
-                        })
-                        .attach_printable_lazy(|| {
-                            format!(
-                                "Deserializing MerchantConnectorWebhookDetails failed {}",
-                                debug_suffix
-                            )
-                        })?;
-                    api_models::webhooks::ConnectorWebhookSecrets {
-                        secret: connector_webhook_details
-                            .merchant_secret
-                            .expose()
-                            .into_bytes(),
-                        additional_secret: connector_webhook_details.additional_secret,
-                    }
-                }
-                None => api_models::webhooks::ConnectorWebhookSecrets {
-                    secret: default_secret.into_bytes(),
-                    additional_secret: None,
-                },
-            },
-            Err(err) => {
-                logger::error!(
-                    "Failed to fetch merchant_secret for source verification {}",
-                    debug_suffix
-                );
-                logger::error!("DB error = {:?}", err);
                 api_models::webhooks::ConnectorWebhookSecrets {
-                    secret: default_secret.into_bytes(),
+                    secret: connector_webhook_details
-                    additional_secret: None,
+                        .merchant_secret
+                        .expose()
+                        .into_bytes(),
+                    additional_secret: connector_webhook_details.additional_secret,
                 }
             }
+
+            None => api_models::webhooks::ConnectorWebhookSecrets {
+                secret: default_secret.into_bytes(),
+                additional_secret: None,
+            },
         };
 
         //need to fetch merchant secret from config table with caching in future for enhanced performance
 
         //If merchant has not set the secret for webhook source verification, "default_secret" is returned.
         //So it will fail during verification step and goes to psync flow.
-        Ok(connector_webhook_secrets)
+        Ok(merchant_secret)
     }
 
     fn get_webhook_source_verification_signature(
@@ -175,12 +141,10 @@ pub trait IncomingWebhook: ConnectorCommon + Sync {
 
     async fn verify_webhook_source(
         &self,
-        db: &dyn StorageInterface,
         request: &IncomingWebhookRequestDetails<'_>,
         merchant_account: &domain::MerchantAccount,
+        merchant_connector_account: domain::MerchantConnectorAccount,
         connector_label: &str,
-        key_store: &domain::MerchantKeyStore,
-        object_reference_id: ObjectReferenceId,
     ) -> CustomResult<bool, errors::ConnectorError> {
         let algorithm = self
             .get_webhook_source_verification_algorithm(request)
@@ -191,14 +155,13 @@ pub trait IncomingWebhook: ConnectorCommon + Sync {
             .change_context(errors::ConnectorError::WebhookSourceVerificationFailed)?;
         let connector_webhook_secrets = self
             .get_webhook_source_verification_merchant_secret(
-                db,
                 merchant_account,
                 connector_label,
-                key_store,
+                merchant_connector_account,
-                object_reference_id,
             )
             .await
             .change_context(errors::ConnectorError::WebhookSourceVerificationFailed)?;
+
         let message = self
             .get_webhook_source_verification_message(
                 request,