refactor(core): add error handling wrapper to wehbook (#6636)

crates/api_models/src/connector_enums.rs

@@ -282,6 +282,9 @@ impl Connector {
     pub fn is_pre_processing_required_before_authorize(&self) -> bool {
         matches!(self, Self::Airwallex)
     }
+    pub fn should_acknowledge_webhook_for_resource_not_found_errors(&self) -> bool {
+        matches!(self, Self::Adyenplatform)
+    }
     #[cfg(feature = "dummy_connector")]
     pub fn validate_dummy_connector_enabled(
         &self,

crates/hyperswitch_connectors/src/connectors/cashtocode.rs

@@ -30,7 +30,7 @@ use hyperswitch_interfaces::{
     errors,
     events::connector_api_logs::ConnectorEvent,
     types::{PaymentsAuthorizeType, Response},
-    webhooks,
+    webhooks::{self, IncomingWebhookFlowError},
 };
 use masking::{Mask, PeekInterface, Secret};
 use transformers as cashtocode;
@@ -420,6 +420,7 @@ impl webhooks::IncomingWebhook for Cashtocode {
     fn get_webhook_api_response(
         &self,
         request: &webhooks::IncomingWebhookRequestDetails<'_>,
+        _error_kind: Option<IncomingWebhookFlowError>,
     ) -> CustomResult<ApplicationResponse<serde_json::Value>, errors::ConnectorError> {
         let status = "EXECUTED".to_string();
         let obj: transformers::CashtocodePaymentsSyncResponse = request

crates/hyperswitch_connectors/src/connectors/worldline.rs

@@ -41,7 +41,7 @@ use hyperswitch_interfaces::{
         PaymentsAuthorizeType, PaymentsCaptureType, PaymentsSyncType, PaymentsVoidType,
         RefundExecuteType, RefundSyncType, Response,
     },
-    webhooks,
+    webhooks::{self, IncomingWebhookFlowError},
 };
 use masking::{ExposeInterface, Mask, PeekInterface};
 use ring::hmac;
@@ -814,6 +814,7 @@ impl webhooks::IncomingWebhook for Worldline {
     fn get_webhook_api_response(
         &self,
         request: &webhooks::IncomingWebhookRequestDetails<'_>,
+        _error_kind: Option<IncomingWebhookFlowError>,
     ) -> CustomResult<
         hyperswitch_domain_models::api::ApplicationResponse<serde_json::Value>,
         errors::ConnectorError,

crates/hyperswitch_connectors/src/connectors/zen.rs

@@ -41,7 +41,7 @@ use hyperswitch_interfaces::{
     errors,
     events::connector_api_logs::ConnectorEvent,
     types::{PaymentsAuthorizeType, PaymentsSyncType, RefundExecuteType, RefundSyncType, Response},
-    webhooks::{IncomingWebhook, IncomingWebhookRequestDetails},
+    webhooks::{IncomingWebhook, IncomingWebhookFlowError, IncomingWebhookRequestDetails},
 };
 use masking::{Mask, PeekInterface, Secret};
 use transformers::{self as zen, ZenPaymentStatus, ZenWebhookTxnType};
@@ -671,6 +671,7 @@ impl IncomingWebhook for Zen {
     fn get_webhook_api_response(
         &self,
         _request: &IncomingWebhookRequestDetails<'_>,
+        _error_kind: Option<IncomingWebhookFlowError>,
     ) -> CustomResult<ApplicationResponse<serde_json::Value>, errors::ConnectorError> {
         Ok(ApplicationResponse::Json(serde_json::json!({
             "status": "ok"

crates/hyperswitch_connectors/src/connectors/zsl.rs

@@ -36,7 +36,7 @@ use hyperswitch_interfaces::{
     errors,
     events::connector_api_logs::ConnectorEvent,
     types::{self, Response},
-    webhooks::{IncomingWebhook, IncomingWebhookRequestDetails},
+    webhooks::{IncomingWebhook, IncomingWebhookFlowError, IncomingWebhookRequestDetails},
 };
 use masking::{ExposeInterface, Secret};
 use transformers::{self as zsl, get_status};
@@ -442,6 +442,7 @@ impl IncomingWebhook for Zsl {
     fn get_webhook_api_response(
         &self,
         _request: &IncomingWebhookRequestDetails<'_>,
+        _error_kind: Option<IncomingWebhookFlowError>,
     ) -> CustomResult<ApplicationResponse<serde_json::Value>, errors::ConnectorError> {
         Ok(ApplicationResponse::TextPlain("CALLBACK-OK".to_string()))
     }

crates/hyperswitch_interfaces/src/webhooks.rs

@@ -2,7 +2,9 @@
 
 use common_utils::{crypto, errors::CustomResult, ext_traits::ValueExt};
 use error_stack::ResultExt;
-use hyperswitch_domain_models::api::ApplicationResponse;
+use hyperswitch_domain_models::{
+    api::ApplicationResponse, errors::api_error_response::ApiErrorResponse,
+};
 use masking::{ExposeInterface, Secret};
 
 use crate::{api::ConnectorCommon, errors};
@@ -22,6 +24,30 @@ pub struct IncomingWebhookRequestDetails<'a> {
     pub query_params: String,
 }
 
+/// IncomingWebhookFlowError enum defining the error type for incoming webhook
+#[derive(Debug)]
+pub enum IncomingWebhookFlowError {
+    /// Resource not found for the webhook
+    ResourceNotFound,
+    /// Internal error for the webhook
+    InternalError,
+}
+
+impl From<&ApiErrorResponse> for IncomingWebhookFlowError {
+    fn from(api_error_response: &ApiErrorResponse) -> Self {
+        match api_error_response {
+            ApiErrorResponse::WebhookResourceNotFound
+            | ApiErrorResponse::DisputeNotFound { .. }
+            | ApiErrorResponse::PayoutNotFound
+            | ApiErrorResponse::MandateNotFound
+            | ApiErrorResponse::PaymentNotFound
+            | ApiErrorResponse::RefundNotFound
+            | ApiErrorResponse::AuthenticationNotFound { .. } => Self::ResourceNotFound,
+            _ => Self::InternalError,
+        }
+    }
+}
+
 /// Trait defining incoming webhook
 #[async_trait::async_trait]
 pub trait IncomingWebhook: ConnectorCommon + Sync {
@@ -203,6 +229,7 @@ pub trait IncomingWebhook: ConnectorCommon + Sync {
     fn get_webhook_api_response(
         &self,
         _request: &IncomingWebhookRequestDetails<'_>,
+        _error_kind: Option<IncomingWebhookFlowError>,
     ) -> CustomResult<ApplicationResponse<serde_json::Value>, errors::ConnectorError> {
         Ok(ApplicationResponse::StatusOk)
     }

crates/router/src/connector/adyen.rs

@@ -7,6 +7,7 @@ use common_utils::{
 };
 use diesel_models::{enums as storage_enums, enums};
 use error_stack::{report, ResultExt};
+use hyperswitch_interfaces::webhooks::IncomingWebhookFlowError;
 use masking::{ExposeInterface, Secret};
 use ring::hmac;
 use router_env::{instrument, tracing};
@@ -1880,6 +1881,7 @@ impl api::IncomingWebhook for Adyen {
     fn get_webhook_api_response(
         &self,
         _request: &api::IncomingWebhookRequestDetails<'_>,
+        _error_kind: Option<IncomingWebhookFlowError>,
     ) -> CustomResult<services::api::ApplicationResponse<serde_json::Value>, errors::ConnectorError>
     {
         Ok(services::api::ApplicationResponse::TextPlain(

crates/router/src/connector/adyenplatform.rs

@@ -13,6 +13,8 @@ use error_stack::report;
 use error_stack::ResultExt;
 #[cfg(feature = "payouts")]
 use http::HeaderName;
+use hyperswitch_interfaces::webhooks::IncomingWebhookFlowError;
+use masking::Maskable;
 #[cfg(feature = "payouts")]
 use masking::Secret;
 #[cfg(feature = "payouts")]
@@ -27,11 +29,7 @@ use crate::{
     configs::settings,
     core::errors::{self, CustomResult},
     headers,
-    services::{
+    services::{self, request::Mask, ConnectorValidation},
-        self,
-        request::{self, Mask},
-        ConnectorValidation,
-    },
     types::{
         self,
         api::{self, ConnectorCommon},
@@ -67,7 +65,7 @@ impl ConnectorCommon for Adyenplatform {
     fn get_auth_header(
         &self,
         auth_type: &types::ConnectorAuthType,
-    ) -> CustomResult<Vec<(String, request::Maskable<String>)>, errors::ConnectorError> {
+    ) -> CustomResult<Vec<(String, Maskable<String>)>, errors::ConnectorError> {
         let auth = adyenplatform::AdyenplatformAuthType::try_from(auth_type)
             .change_context(errors::ConnectorError::FailedToObtainAuthType)?;
         Ok(vec![(
@@ -209,7 +207,7 @@ impl services::ConnectorIntegration<api::PoFulfill, types::PayoutsData, types::P
         &self,
         req: &types::PayoutsRouterData<api::PoFulfill>,
         _connectors: &settings::Connectors,
-    ) -> CustomResult<Vec<(String, request::Maskable<String>)>, errors::ConnectorError> {
+    ) -> CustomResult<Vec<(String, Maskable<String>)>, errors::ConnectorError> {
         let mut header = vec![(
             headers::CONTENT_TYPE.to_string(),
             types::PayoutFulfillType::get_content_type(self)
@@ -401,6 +399,25 @@ impl api::IncomingWebhook for Adyenplatform {
         }
     }
 
+    fn get_webhook_api_response(
+        &self,
+        _request: &api::IncomingWebhookRequestDetails<'_>,
+        error_kind: Option<IncomingWebhookFlowError>,
+    ) -> CustomResult<services::api::ApplicationResponse<serde_json::Value>, errors::ConnectorError>
+    {
+        if error_kind.is_some() {
+            Ok(services::api::ApplicationResponse::JsonWithHeaders((
+                serde_json::Value::Null,
+                vec![(
+                    "x-http-code".to_string(),
+                    Maskable::Masked(Secret::new("404".to_string())),
+                )],
+            )))
+        } else {
+            Ok(services::api::ApplicationResponse::StatusOk)
+        }
+    }
+
     fn get_webhook_event_type(
         &self,
         #[cfg(feature = "payouts")] request: &api::IncomingWebhookRequestDetails<'_>,

crates/router/src/connector/braintree.rs

@@ -10,6 +10,7 @@ use common_utils::{
 };
 use diesel_models::enums;
 use error_stack::{report, Report, ResultExt};
+use hyperswitch_interfaces::webhooks::IncomingWebhookFlowError;
 use masking::{ExposeInterface, PeekInterface, Secret};
 use ring::hmac;
 use sha1::{Digest, Sha1};
@@ -980,6 +981,7 @@ impl api::IncomingWebhook for Braintree {
     fn get_webhook_api_response(
         &self,
         _request: &api::IncomingWebhookRequestDetails<'_>,
+        _error_kind: Option<IncomingWebhookFlowError>,
     ) -> CustomResult<services::api::ApplicationResponse<serde_json::Value>, errors::ConnectorError>
     {
         Ok(services::api::ApplicationResponse::TextPlain(

crates/router/src/core/webhooks/incoming.rs

@@ -12,7 +12,7 @@ use hyperswitch_domain_models::{
     router_request_types::VerifyWebhookSourceRequestData,
     router_response_types::{VerifyWebhookSourceResponseData, VerifyWebhookStatus},
 };
-use hyperswitch_interfaces::webhooks::IncomingWebhookRequestDetails;
+use hyperswitch_interfaces::webhooks::{IncomingWebhookFlowError, IncomingWebhookRequestDetails};
 use masking::{ExposeInterface, PeekInterface};
 use router_env::{instrument, metrics::add_attributes, tracing, tracing_actix_web::RequestId};
 
@@ -209,7 +209,7 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
             );
 
             let response = connector
-                .get_webhook_api_response(&request_details)
+                .get_webhook_api_response(&request_details, None)
                 .switch()
                 .attach_printable("Failed while early return in case of event type parsing")?;
 
@@ -260,14 +260,25 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
         let merchant_connector_account = match merchant_connector_account {
             Some(merchant_connector_account) => merchant_connector_account,
             None => {
-                Box::pin(helper_utils::get_mca_from_object_reference_id(
+                match Box::pin(helper_utils::get_mca_from_object_reference_id(
                     &state,
                     object_ref_id.clone(),
                     &merchant_account,
                     &connector_name,
                     &key_store,
                 ))
-                .await?
+                .await
+                {
+                    Ok(mca) => mca,
+                    Err(error) => {
+                        return handle_incoming_webhook_error(
+                            error,
+                            &connector,
+                            connector_name.as_str(),
+                            &request_details,
+                        );
+                    }
+                }
             }
         };
 
@@ -358,7 +369,7 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
                 id: profile_id.get_string_repr().to_owned(),
             })?;
 
-        match flow_type {
+        let result_response = match flow_type {
             api::WebhookFlow::Payment => Box::pin(payments_incoming_webhook_flow(
                 state.clone(),
                 req_state,
@@ -372,7 +383,7 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
                 event_type,
             ))
             .await
-            .attach_printable("Incoming webhook flow for payments failed")?,
+            .attach_printable("Incoming webhook flow for payments failed"),
 
             api::WebhookFlow::Refund => Box::pin(refunds_incoming_webhook_flow(
                 state.clone(),
@@ -385,7 +396,7 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
                 event_type,
             ))
             .await
-            .attach_printable("Incoming webhook flow for refunds failed")?,
+            .attach_printable("Incoming webhook flow for refunds failed"),
 
             api::WebhookFlow::Dispute => Box::pin(disputes_incoming_webhook_flow(
                 state.clone(),
@@ -399,7 +410,7 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
                 event_type,
             ))
             .await
-            .attach_printable("Incoming webhook flow for disputes failed")?,
+            .attach_printable("Incoming webhook flow for disputes failed"),
 
             api::WebhookFlow::BankTransfer => Box::pin(bank_transfer_webhook_flow(
                 state.clone(),
@@ -411,9 +422,9 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
                 source_verified,
             ))
             .await
-            .attach_printable("Incoming bank-transfer webhook flow failed")?,
+            .attach_printable("Incoming bank-transfer webhook flow failed"),
 
-            api::WebhookFlow::ReturnResponse => WebhookResponseTracker::NoEffect,
+            api::WebhookFlow::ReturnResponse => Ok(WebhookResponseTracker::NoEffect),
 
             api::WebhookFlow::Mandate => Box::pin(mandates_incoming_webhook_flow(
                 state.clone(),
@@ -425,7 +436,7 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
                 event_type,
             ))
             .await
-            .attach_printable("Incoming webhook flow for mandates failed")?,
+            .attach_printable("Incoming webhook flow for mandates failed"),
 
             api::WebhookFlow::ExternalAuthentication => {
                 Box::pin(external_authentication_incoming_webhook_flow(
@@ -442,7 +453,7 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
                     merchant_connector_account,
                 ))
                 .await
-                .attach_printable("Incoming webhook flow for external authentication failed")?
+                .attach_printable("Incoming webhook flow for external authentication failed")
             }
             api::WebhookFlow::FraudCheck => Box::pin(frm_incoming_webhook_flow(
                 state.clone(),
@@ -455,7 +466,7 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
                 business_profile,
             ))
             .await
-            .attach_printable("Incoming webhook flow for fraud check failed")?,
+            .attach_printable("Incoming webhook flow for fraud check failed"),
 
             #[cfg(feature = "payouts")]
             api::WebhookFlow::Payout => Box::pin(payouts_incoming_webhook_flow(
@@ -468,10 +479,22 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
                 source_verified,
             ))
             .await
-            .attach_printable("Incoming webhook flow for payouts failed")?,
+            .attach_printable("Incoming webhook flow for payouts failed"),
 
             _ => Err(errors::ApiErrorResponse::InternalServerError)
-                .attach_printable("Unsupported Flow Type received in incoming webhooks")?,
+                .attach_printable("Unsupported Flow Type received in incoming webhooks"),
+        };
+
+        match result_response {
+            Ok(response) => response,
+            Err(error) => {
+                return handle_incoming_webhook_error(
+                    error,
+                    &connector,
+                    connector_name.as_str(),
+                    &request_details,
+                );
+            }
         }
     } else {
         metrics::WEBHOOK_INCOMING_FILTERED_COUNT.add(
@@ -486,7 +509,7 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
     };
 
     let response = connector
-        .get_webhook_api_response(&request_details)
+        .get_webhook_api_response(&request_details, None)
         .switch()
         .attach_printable("Could not get incoming webhook api response from connector")?;
 
@@ -497,6 +520,44 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
     Ok((response, webhook_effect, serialized_request))
 }
 
+fn handle_incoming_webhook_error(
+    error: error_stack::Report<errors::ApiErrorResponse>,
+    connector: &ConnectorEnum,
+    connector_name: &str,
+    request_details: &IncomingWebhookRequestDetails<'_>,
+) -> errors::RouterResult<(
+    services::ApplicationResponse<serde_json::Value>,
+    WebhookResponseTracker,
+    serde_json::Value,
+)> {
+    logger::error!(?error, "Incoming webhook flow failed");
+
+    // fetch the connector enum from the connector name
+    let connector_enum = api_models::connector_enums::Connector::from_str(connector_name)
+        .change_context(errors::ApiErrorResponse::InvalidDataValue {
+            field_name: "connector",
+        })
+        .attach_printable_lazy(|| format!("unable to parse connector name {connector_name:?}"))?;
+
+    // get the error response from the connector
+    if connector_enum.should_acknowledge_webhook_for_resource_not_found_errors() {
+        let response = connector
+            .get_webhook_api_response(
+                request_details,
+                Some(IncomingWebhookFlowError::from(error.current_context())),
+            )
+            .switch()
+            .attach_printable("Failed to get incoming webhook api response from connector")?;
+        Ok((
+            response,
+            WebhookResponseTracker::NoEffect,
+            serde_json::Value::Null,
+        ))
+    } else {
+        Err(error)
+    }
+}
+
 #[allow(clippy::too_many_arguments)]
 #[instrument(skip_all)]
 async fn payments_incoming_webhook_flow(

crates/router/src/core/webhooks/incoming_v2.rs

@@ -192,7 +192,7 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
             );
 
             let response = connector
-                .get_webhook_api_response(&request_details)
+                .get_webhook_api_response(&request_details, None)
                 .switch()
                 .attach_printable("Failed while early return in case of event type parsing")?;
 
@@ -367,7 +367,7 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
     };
 
     let response = connector
-        .get_webhook_api_response(&request_details)
+        .get_webhook_api_response(&request_details, None)
         .switch()
         .attach_printable("Could not get incoming webhook api response from connector")?;
 

crates/router/src/services/connector_integration_interface.rs

@@ -1,7 +1,8 @@
 use common_utils::{crypto, errors::CustomResult, request::Request};
 use hyperswitch_domain_models::{router_data::RouterData, router_data_v2::RouterDataV2};
 use hyperswitch_interfaces::{
-    authentication::ExternalAuthenticationPayload, connector_integration_v2::ConnectorIntegrationV2,
+    authentication::ExternalAuthenticationPayload,
+    connector_integration_v2::ConnectorIntegrationV2, webhooks::IncomingWebhookFlowError,
 };
 
 use super::{BoxedConnectorIntegrationV2, ConnectorValidation};
@@ -279,11 +280,12 @@ impl api::IncomingWebhook for ConnectorEnum {
     fn get_webhook_api_response(
         &self,
         request: &IncomingWebhookRequestDetails<'_>,
+        error_kind: Option<IncomingWebhookFlowError>,
     ) -> CustomResult<services_api::ApplicationResponse<serde_json::Value>, errors::ConnectorError>
     {
         match self {
-            Self::Old(connector) => connector.get_webhook_api_response(request),
+            Self::Old(connector) => connector.get_webhook_api_response(request, error_kind),
-            Self::New(connector) => connector.get_webhook_api_response(request),
+            Self::New(connector) => connector.get_webhook_api_response(request, error_kind),
         }
     }