fix(webhooks): add support for updating mandate details in webhooks flow (#6523)

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/payment_attempt.rs

@@ -433,6 +433,10 @@ pub enum PaymentAttemptUpdate {
         payment_method_id: Option<String>,
         updated_by: String,
     },
+    ConnectorMandateDetailUpdate {
+        connector_mandate_detail: Option<ConnectorMandateReferenceId>,
+        updated_by: String,
+    },
     BlocklistUpdate {
         status: storage_enums::AttemptStatus,
         error_code: Option<Option<String>>,
@@ -628,6 +632,10 @@ pub enum PaymentAttemptUpdate {
     //     payment_method_id: Option<String>,
     //     updated_by: String,
     // },
+    // ConnectorMandateDetailUpdate {
+    //     connector_mandate_detail: Option<ConnectorMandateReferenceId>,
+    //     updated_by: String,
+    // }
     // BlocklistUpdate {
     //     status: storage_enums::AttemptStatus,
     //     error_code: Option<Option<String>>,
@@ -1393,7 +1401,63 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
         //         customer_acceptance: None,
         //         card_network: None,
         //     },
-        //     PaymentAttemptUpdate::PaymentMethodDetailsUpdate {
+        //     PaymentAttemptUpdate::ConnectorMandateDetailUpdate {
+        //         connector_mandate_detail,
+        // updated_by,
+        //     } => Self {
+        //         payment_method_id: None,
+        //         modified_at: common_utils::date_time::now(),
+        //         updated_by,
+        //         amount: None,
+        //         net_amount: None,
+        //         currency: None,
+        //         status: None,
+        //         connector_transaction_id: None,
+        //         amount_to_capture: None,
+        //         connector: None,
+        //         authentication_type: None,
+        //         payment_method: None,
+        //         error_message: None,
+        //         cancellation_reason: None,
+        //         mandate_id: None,
+        //         browser_info: None,
+        //         payment_token: None,
+        //         error_code: None,
+        //         connector_metadata: None,
+        //         payment_method_data: None,
+        //         payment_method_type: None,
+        //         payment_experience: None,
+        //         business_sub_label: None,
+        //         straight_through_algorithm: None,
+        //         preprocessing_step_id: None,
+        //         error_reason: None,
+        //         capture_method: None,
+        //         connector_response_reference_id: None,
+        //         multiple_capture_count: None,
+        //         surcharge_amount: None,
+        //         tax_amount: None,
+        //         amount_capturable: None,
+        //         merchant_connector_id: None,
+        //         authentication_data: None,
+        //         encoded_data: None,
+        //         unified_code: None,
+        //         unified_message: None,
+        //         external_three_ds_authentication_attempted: None,
+        //         authentication_connector: None,
+        //         authentication_id: None,
+        //         fingerprint_id: None,
+        //         payment_method_billing_address_id: None,
+        //         charge_id: None,
+        //         client_source: None,
+        //         client_version: None,
+        //         customer_acceptance: None,
+        //         card_network: None,
+        //         shipping_cost: None,
+        //         order_tax_amount: None,
+        //         connector_transaction_data: None,
+        //         connector_mandate_detail,
+        //     },
+        //     PaymentAttemptUpdate::ConnectorMandateDetailUpdate {
         //         payment_method_id,
         //         updated_by,
         //     } => Self {
@@ -2394,6 +2458,62 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 connector_transaction_data: None,
                 connector_mandate_detail: None,
             },
+            PaymentAttemptUpdate::ConnectorMandateDetailUpdate {
+                connector_mandate_detail,
+                updated_by,
+            } => Self {
+                payment_method_id: None,
+                modified_at: common_utils::date_time::now(),
+                updated_by,
+                amount: None,
+                net_amount: None,
+                currency: None,
+                status: None,
+                connector_transaction_id: None,
+                amount_to_capture: None,
+                connector: None,
+                authentication_type: None,
+                payment_method: None,
+                error_message: None,
+                cancellation_reason: None,
+                mandate_id: None,
+                browser_info: None,
+                payment_token: None,
+                error_code: None,
+                connector_metadata: None,
+                payment_method_data: None,
+                payment_method_type: None,
+                payment_experience: None,
+                business_sub_label: None,
+                straight_through_algorithm: None,
+                preprocessing_step_id: None,
+                error_reason: None,
+                capture_method: None,
+                connector_response_reference_id: None,
+                multiple_capture_count: None,
+                surcharge_amount: None,
+                tax_amount: None,
+                amount_capturable: None,
+                merchant_connector_id: None,
+                authentication_data: None,
+                encoded_data: None,
+                unified_code: None,
+                unified_message: None,
+                external_three_ds_authentication_attempted: None,
+                authentication_connector: None,
+                authentication_id: None,
+                fingerprint_id: None,
+                payment_method_billing_address_id: None,
+                charge_id: None,
+                client_source: None,
+                client_version: None,
+                customer_acceptance: None,
+                card_network: None,
+                shipping_cost: None,
+                order_tax_amount: None,
+                connector_transaction_data: None,
+                connector_mandate_detail,
+            },
             PaymentAttemptUpdate::PaymentMethodDetailsUpdate {
                 payment_method_id,
                 updated_by,

crates/hyperswitch_connectors/src/connectors/fiuu.rs

@@ -42,7 +42,7 @@ use masking::{ExposeInterface, PeekInterface, Secret};
 use reqwest::multipart::Form;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
-use transformers::{self as fiuu, FiuuWebhooksResponse};
+use transformers::{self as fiuu, ExtraParameters, FiuuWebhooksResponse};
 
 use crate::{
     constants::headers,
@@ -890,4 +890,36 @@ impl webhooks::IncomingWebhook for Fiuu {
             }
         }
     }
+
+    fn get_mandate_details(
+        &self,
+        request: &webhooks::IncomingWebhookRequestDetails<'_>,
+    ) -> CustomResult<
+        Option<hyperswitch_domain_models::router_flow_types::ConnectorMandateDetails>,
+        errors::ConnectorError,
+    > {
+        let webhook_payment_response: transformers::FiuuWebhooksPaymentResponse =
+            serde_urlencoded::from_bytes::<transformers::FiuuWebhooksPaymentResponse>(request.body)
+                .change_context(errors::ConnectorError::WebhookResourceObjectNotFound)?;
+        let mandate_reference = webhook_payment_response.extra_parameters.as_ref().and_then(|extra_p| {
+                    let mandate_token: Result<ExtraParameters, _> = serde_json::from_str(extra_p);
+                    match mandate_token {
+                        Ok(token) => {
+                            token.token.as_ref().map(|token| hyperswitch_domain_models::router_flow_types::ConnectorMandateDetails {
+                                connector_mandate_id:token.clone(),
+                            })
+                        }
+                        Err(err) => {
+                            router_env::logger::warn!(
+                                "Failed to convert 'extraP' from fiuu webhook response to fiuu::ExtraParameters. \
+                                 Input: '{}', Error: {}",
+                                extra_p,
+                                err
+                            );
+                            None
+                        }
+                    }
+                });
+        Ok(mandate_reference)
+    }
 }

crates/hyperswitch_connectors/src/connectors/fiuu/transformers.rs

@@ -726,7 +726,7 @@ pub struct NonThreeDSResponseData {
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct ExtraParameters {
-    token: Option<Secret<String>>,
+    pub token: Option<Secret<String>>,
 }
 
 impl<F>

crates/hyperswitch_domain_models/src/payments/payment_attempt.rs

@@ -807,6 +807,10 @@ pub enum PaymentAttemptUpdate {
         payment_method_id: Option<String>,
         updated_by: String,
     },
+    ConnectorMandateDetailUpdate {
+        connector_mandate_detail: Option<ConnectorMandateReferenceId>,
+        updated_by: String,
+    },
     VoidUpdate {
         status: storage_enums::AttemptStatus,
         cancellation_reason: Option<String>,
@@ -994,6 +998,13 @@ impl PaymentAttemptUpdate {
                 error_message,
                 updated_by,
             },
+            Self::ConnectorMandateDetailUpdate {
+                connector_mandate_detail,
+                updated_by,
+            } => DieselPaymentAttemptUpdate::ConnectorMandateDetailUpdate {
+                connector_mandate_detail,
+                updated_by,
+            },
             Self::PaymentMethodDetailsUpdate {
                 payment_method_id,
                 updated_by,

crates/hyperswitch_domain_models/src/router_flow_types/webhooks.rs

@@ -1,2 +1,9 @@
+use serde::Serialize;
+
 #[derive(Clone, Debug)]
 pub struct VerifyWebhookSource;
+
+#[derive(Debug, Clone, Serialize)]
+pub struct ConnectorMandateDetails {
+    pub connector_mandate_id: masking::Secret<String>,
+}

crates/hyperswitch_interfaces/src/webhooks.rs

@@ -226,4 +226,15 @@ pub trait IncomingWebhook: ConnectorCommon + Sync {
         )
         .into())
     }
+
+    /// fn get_mandate_details
+    fn get_mandate_details(
+        &self,
+        _request: &IncomingWebhookRequestDetails<'_>,
+    ) -> CustomResult<
+        Option<hyperswitch_domain_models::router_flow_types::ConnectorMandateDetails>,
+        errors::ConnectorError,
+    > {
+        Ok(None)
+    }
 }

crates/router/src/configs/defaults/payment_connector_required_fields.rs

@@ -1424,15 +1424,6 @@ impl Default for settings::RequiredFields {
                                                 value: None,
                                             }
                                         ),
-                                        (
-                                            "billing.address.last_name".to_string(),
-                                            RequiredFieldInfo {
-                                                required_field: "payment_method_data.billing.address.last_name".to_string(),
-                                                display_name: "card_holder_name".to_string(),
-                                                field_type: enums::FieldType::UserFullName,
-                                                value: None,
-                                            }
-                                        ),
                                     ]),
                                     non_mandate: HashMap::new(),
                                     common: HashMap::from(
@@ -4544,15 +4535,6 @@ impl Default for settings::RequiredFields {
                                                 value: None,
                                             }
                                         ),
-                                        (
-                                            "billing.address.last_name".to_string(),
-                                            RequiredFieldInfo {
-                                                required_field: "payment_method_data.billing.address.last_name".to_string(),
-                                                display_name: "card_holder_name".to_string(),
-                                                field_type: enums::FieldType::UserFullName,
-                                                value: None,
-                                            }
-                                        ),
                                     ]),
                                     non_mandate: HashMap::new(),
                                     common: HashMap::from(

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

@@ -4,15 +4,16 @@ use actix_web::FromRequest;
 #[cfg(feature = "payouts")]
 use api_models::payouts as payout_models;
 use api_models::webhooks::{self, WebhookResponseTracker};
-use common_utils::{errors::ReportSwitchExt, events::ApiEventsType};
+use common_utils::{errors::ReportSwitchExt, events::ApiEventsType, ext_traits::ValueExt};
+use diesel_models::ConnectorMandateReferenceId;
 use error_stack::{report, ResultExt};
 use hyperswitch_domain_models::{
-    payments::HeaderPayload,
+    payments::{payment_attempt::PaymentAttempt, HeaderPayload},
     router_request_types::VerifyWebhookSourceRequestData,
     router_response_types::{VerifyWebhookSourceResponseData, VerifyWebhookStatus},
 };
 use hyperswitch_interfaces::webhooks::IncomingWebhookRequestDetails;
-use masking::ExposeInterface;
+use masking::{ExposeInterface, PeekInterface};
 use router_env::{instrument, metrics::add_attributes, tracing, tracing_actix_web::RequestId};
 
 use super::{types, utils, MERCHANT_ID};
@@ -21,7 +22,9 @@ use crate::{
     core::{
         api_locking,
         errors::{self, ConnectorErrorExt, CustomResult, RouterResponse, StorageErrorExt},
-        metrics, payments, refunds, utils as core_utils,
+        metrics, payments,
+        payments::tokenization,
+        refunds, utils as core_utils,
         webhooks::utils::construct_webhook_router_data,
     },
     db::StorageInterface,
@@ -44,7 +47,7 @@ use crate::{
         storage::{self, enums},
         transformers::{ForeignFrom, ForeignInto, ForeignTryFrom},
     },
-    utils::{self as helper_utils, generate_id},
+    utils::{self as helper_utils, ext_traits::OptionExt, generate_id},
 };
 #[cfg(feature = "payouts")]
 use crate::{core::payouts, types::storage::PayoutAttemptUpdate};
@@ -364,6 +367,9 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
                 key_store,
                 webhook_details,
                 source_verified,
+                &connector,
+                &request_details,
+                event_type,
             ))
             .await
             .attach_printable("Incoming webhook flow for payments failed")?,
@@ -491,6 +497,7 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
     Ok((response, webhook_effect, serialized_request))
 }
 
+#[allow(clippy::too_many_arguments)]
 #[instrument(skip_all)]
 async fn payments_incoming_webhook_flow(
     state: SessionState,
@@ -500,6 +507,9 @@ async fn payments_incoming_webhook_flow(
     key_store: domain::MerchantKeyStore,
     webhook_details: api::IncomingWebhookDetails,
     source_verified: bool,
+    connector: &ConnectorEnum,
+    request_details: &IncomingWebhookRequestDetails<'_>,
+    event_type: webhooks::IncomingWebhookEvent,
 ) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
     let consume_or_trigger_flow = if source_verified {
         payments::CallConnectorAction::HandleResponse(webhook_details.resource_object)
@@ -507,10 +517,10 @@ async fn payments_incoming_webhook_flow(
         payments::CallConnectorAction::Trigger
     };
     let payments_response = match webhook_details.object_reference_id {
-        webhooks::ObjectReferenceId::PaymentId(id) => {
+        webhooks::ObjectReferenceId::PaymentId(ref id) => {
             let payment_id = get_payment_id(
                 state.store.as_ref(),
-                &id,
+                id,
                 merchant_account.get_id(),
                 merchant_account.storage_scheme,
             )
@@ -544,7 +554,7 @@ async fn payments_incoming_webhook_flow(
                 key_store.clone(),
                 payments::operations::PaymentStatus,
                 api::PaymentsRetrieveRequest {
-                    resource_id: id,
+                    resource_id: id.clone(),
                     merchant_id: Some(merchant_account.get_id().clone()),
                     force_sync: true,
                     connector: None,
@@ -555,12 +565,23 @@ async fn payments_incoming_webhook_flow(
                     expand_captures: None,
                 },
                 services::AuthFlow::Merchant,
-                consume_or_trigger_flow,
+                consume_or_trigger_flow.clone(),
                 None,
                 HeaderPayload::default(),
             ))
             .await;
-
+            // When mandate details are present in successful webhooks, and consuming webhooks are skipped during payment sync if the payment status is already updated to charged, this function is used to update the connector mandate details.
+            if should_update_connector_mandate_details(source_verified, event_type) {
+                update_connector_mandate_details(
+                    &state,
+                    &merchant_account,
+                    &key_store,
+                    webhook_details.object_reference_id.clone(),
+                    connector,
+                    request_details,
+                )
+                .await?
+            };
             lock_action
                 .free_lock_action(&state, merchant_account.get_id().to_owned())
                 .await?;
@@ -869,10 +890,7 @@ async fn get_payment_attempt_from_object_reference_id(
     state: &SessionState,
     object_reference_id: webhooks::ObjectReferenceId,
     merchant_account: &domain::MerchantAccount,
-) -> CustomResult<
+) -> CustomResult<PaymentAttempt, errors::ApiErrorResponse> {
-    hyperswitch_domain_models::payments::payment_attempt::PaymentAttempt,
-    errors::ApiErrorResponse,
-> {
     let db = &*state.store;
     match object_reference_id {
         api::ObjectReferenceId::PaymentId(api::PaymentIdType::ConnectorTransactionId(ref id)) => db
@@ -911,7 +929,7 @@ async fn get_or_update_dispute_object(
     dispute_details: api::disputes::DisputePayload,
     merchant_id: &common_utils::id_type::MerchantId,
     organization_id: &common_utils::id_type::OrganizationId,
-    payment_attempt: &hyperswitch_domain_models::payments::payment_attempt::PaymentAttempt,
+    payment_attempt: &PaymentAttempt,
     event_type: webhooks::IncomingWebhookEvent,
     business_profile: &domain::Profile,
     connector_name: &str,
@@ -1716,3 +1734,173 @@ async fn fetch_optional_mca_and_connector(
         Ok((None, connector, connector_name))
     }
 }
+
+fn should_update_connector_mandate_details(
+    source_verified: bool,
+    event_type: webhooks::IncomingWebhookEvent,
+) -> bool {
+    source_verified && event_type == webhooks::IncomingWebhookEvent::PaymentIntentSuccess
+}
+
+async fn update_connector_mandate_details(
+    state: &SessionState,
+    merchant_account: &domain::MerchantAccount,
+    key_store: &domain::MerchantKeyStore,
+    object_ref_id: api::ObjectReferenceId,
+    connector: &ConnectorEnum,
+    request_details: &IncomingWebhookRequestDetails<'_>,
+) -> CustomResult<(), errors::ApiErrorResponse> {
+    let webhook_connector_mandate_details = connector
+        .get_mandate_details(request_details)
+        .switch()
+        .attach_printable("Could not find connector mandate details in incoming webhook body")?;
+
+    if let Some(webhook_mandate_details) = webhook_connector_mandate_details {
+        let payment_attempt =
+            get_payment_attempt_from_object_reference_id(state, object_ref_id, merchant_account)
+                .await?;
+
+        if let Some(ref payment_method_id) = payment_attempt.payment_method_id {
+            let key_manager_state = &state.into();
+            let payment_method_info = state
+                .store
+                .find_payment_method(
+                    key_manager_state,
+                    key_store,
+                    payment_method_id,
+                    merchant_account.storage_scheme,
+                )
+                .await
+                .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;
+
+            let mandate_details = payment_method_info
+                .connector_mandate_details
+                .clone()
+                .map(|val| {
+                    val.parse_value::<diesel_models::PaymentsMandateReference>(
+                        "PaymentsMandateReference",
+                    )
+                })
+                .transpose()
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("Failed to deserialize to Payment Mandate Reference")?;
+
+            let merchant_connector_account_id = payment_attempt
+                .merchant_connector_id
+                .clone()
+                .get_required_value("merchant_connector_id")?;
+
+            if mandate_details
+                .as_ref()
+                .map(|details: &diesel_models::PaymentsMandateReference| {
+                    !details.0.contains_key(&merchant_connector_account_id)
+                })
+                .unwrap_or(true)
+            {
+                let updated_connector_mandate_details = insert_mandate_details(
+                    &payment_attempt,
+                    &webhook_mandate_details,
+                    mandate_details,
+                )?;
+                let pm_update = diesel_models::PaymentMethodUpdate::ConnectorMandateDetailsUpdate {
+                    connector_mandate_details: updated_connector_mandate_details,
+                };
+
+                state
+                    .store
+                    .update_payment_method(
+                        key_manager_state,
+                        key_store,
+                        payment_method_info,
+                        pm_update,
+                        merchant_account.storage_scheme,
+                    )
+                    .await
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                    .attach_printable("Failed to update payment method in db")?;
+                // Update the payment attempt to maintain consistency across tables.
+
+                let (mandate_metadata, connector_mandate_request_reference_id) = payment_attempt
+                    .connector_mandate_detail
+                    .as_ref()
+                    .map(|details| {
+                        (
+                            details.mandate_metadata.clone(),
+                            details.connector_mandate_request_reference_id.clone(),
+                        )
+                    })
+                    .unwrap_or((None, None));
+
+                let connector_mandate_reference_id = ConnectorMandateReferenceId {
+                    connector_mandate_id: Some(
+                        webhook_mandate_details
+                            .connector_mandate_id
+                            .peek()
+                            .to_string(),
+                    ),
+                    payment_method_id: Some(payment_method_id.to_string()),
+                    mandate_metadata,
+                    connector_mandate_request_reference_id,
+                };
+
+                let attempt_update = storage::PaymentAttemptUpdate::ConnectorMandateDetailUpdate {
+                    connector_mandate_detail: Some(connector_mandate_reference_id),
+                    updated_by: merchant_account.storage_scheme.to_string(),
+                };
+
+                state
+                    .store
+                    .update_payment_attempt_with_attempt_id(
+                        payment_attempt.clone(),
+                        attempt_update,
+                        merchant_account.storage_scheme,
+                    )
+                    .await
+                    .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
+            } else {
+                logger::info!(
+                    "Skipping connector mandate details update since they are already present."
+                );
+            }
+        }
+    }
+    Ok(())
+}
+
+fn insert_mandate_details(
+    payment_attempt: &PaymentAttempt,
+    webhook_mandate_details: &hyperswitch_domain_models::router_flow_types::ConnectorMandateDetails,
+    payment_method_mandate_details: Option<diesel_models::PaymentsMandateReference>,
+) -> CustomResult<Option<serde_json::Value>, errors::ApiErrorResponse> {
+    let (mandate_metadata, connector_mandate_request_reference_id) = payment_attempt
+        .connector_mandate_detail
+        .clone()
+        .map(|mandate_reference| {
+            (
+                mandate_reference.mandate_metadata,
+                mandate_reference.connector_mandate_request_reference_id,
+            )
+        })
+        .unwrap_or((None, None));
+    let connector_mandate_details = tokenization::update_connector_mandate_details(
+        payment_method_mandate_details,
+        payment_attempt.payment_method_type,
+        Some(
+            payment_attempt
+                .net_amount
+                .get_total_amount()
+                .get_amount_as_i64(),
+        ),
+        payment_attempt.currency,
+        payment_attempt.merchant_connector_id.clone(),
+        Some(
+            webhook_mandate_details
+                .connector_mandate_id
+                .peek()
+                .to_string(),
+        ),
+        mandate_metadata,
+        connector_mandate_request_reference_id,
+    )?;
+    Ok(connector_mandate_details)
+}

crates/router/src/services/connector_integration_interface.rs

@@ -306,6 +306,19 @@ impl api::IncomingWebhook for ConnectorEnum {
             Self::New(connector) => connector.get_external_authentication_details(request),
         }
     }
+
+    fn get_mandate_details(
+        &self,
+        request: &IncomingWebhookRequestDetails<'_>,
+    ) -> CustomResult<
+        Option<hyperswitch_domain_models::router_flow_types::ConnectorMandateDetails>,
+        errors::ConnectorError,
+    > {
+        match self {
+            Self::Old(connector) => connector.get_mandate_details(request),
+            Self::New(connector) => connector.get_mandate_details(request),
+        }
+    }
 }
 
 impl api::ConnectorTransactionId for ConnectorEnum {