Bug: Routing changes to choose between MIT details and raw card



Will have to be extended in future to choose between MIT details vs Network Reference ID as well
diff --git a/config/development.toml b/config/development.toml
index 6d4914af8b4..6e27e110ee9 100644
--- a/config/development.toml
+++ b/config/development.toml
@@ -581,4 +581,4 @@ enabled = true
 file_storage_backend = "file_system"
 
 [unmasked_headers]
-keys = "user-agent"
\ No newline at end of file
+keys = "user-agent"
diff --git a/crates/api_models/src/payments.rs b/crates/api_models/src/payments.rs
index 6f97a87fbe9..33d355d255a 100644
--- a/crates/api_models/src/payments.rs
+++ b/crates/api_models/src/payments.rs
@@ -740,7 +740,7 @@ pub enum MandateTransactionType {
 
 #[derive(Default, Eq, PartialEq, Debug, serde::Deserialize, serde::Serialize, Clone)]
 pub struct MandateIds {
-    pub mandate_id: String,
+    pub mandate_id: Option<String>,
     pub mandate_reference_id: Option<MandateReferenceId>,
 }
 
@@ -767,7 +767,7 @@ pub struct UpdateHistory {
 impl MandateIds {
     pub fn new(mandate_id: String) -> Self {
         Self {
-            mandate_id,
+            mandate_id: Some(mandate_id),
             mandate_reference_id: None,
         }
     }
diff --git a/crates/router/src/connector/dlocal/transformers.rs b/crates/router/src/connector/dlocal/transformers.rs
index 5eb682d4cfa..c1b8731ee89 100644
--- a/crates/router/src/connector/dlocal/transformers.rs
+++ b/crates/router/src/connector/dlocal/transformers.rs
@@ -139,7 +139,7 @@ impl TryFrom<&DlocalRouterData<&types::PaymentsAuthorizeRouterData>> for DlocalP
                             .request
                             .mandate_id
                             .as_ref()
-                            .map(|ids| ids.mandate_id.clone()),
+                            .and_then(|ids| ids.mandate_id.clone()),
                         // [#595[FEATURE] Pass Mandate history information in payment flows/request]
                         installments: item
                             .router_data
diff --git a/crates/router/src/core/mandate.rs b/crates/router/src/core/mandate.rs
index d07a8814c88..2838edfbf91 100644
--- a/crates/router/src/core/mandate.rs
+++ b/crates/router/src/core/mandate.rs
@@ -326,48 +326,52 @@ where
         Err(_) => {}
         Ok(_) => match resp.request.get_mandate_id() {
             Some(mandate_id) => {
-                let mandate_id = &mandate_id.mandate_id;
-                let mandate = state
-                    .store
-                    .find_mandate_by_merchant_id_mandate_id(resp.merchant_id.as_ref(), mandate_id)
-                    .await
-                    .to_not_found_response(errors::ApiErrorResponse::MandateNotFound)?;
-                let mandate = match mandate.mandate_type {
-                    storage_enums::MandateType::SingleUse => state
+                if let Some(ref mandate_id) = mandate_id.mandate_id {
+                    let mandate = state
                         .store
-                        .update_mandate_by_merchant_id_mandate_id(
-                            &resp.merchant_id,
+                        .find_mandate_by_merchant_id_mandate_id(
+                            resp.merchant_id.as_ref(),
                             mandate_id,
-                            storage::MandateUpdate::StatusUpdate {
-                                mandate_status: storage_enums::MandateStatus::Revoked,
-                            },
                         )
                         .await
-                        .change_context(errors::ApiErrorResponse::MandateUpdateFailed),
-                    storage_enums::MandateType::MultiUse => state
-                        .store
-                        .update_mandate_by_merchant_id_mandate_id(
-                            &resp.merchant_id,
-                            mandate_id,
-                            storage::MandateUpdate::CaptureAmountUpdate {
-                                amount_captured: Some(
-                                    mandate.amount_captured.unwrap_or(0)
-                                        + resp.request.get_amount(),
-                                ),
-                            },
-                        )
-                        .await
-                        .change_context(errors::ApiErrorResponse::MandateUpdateFailed),
-                }?;
-                metrics::SUBSEQUENT_MANDATE_PAYMENT.add(
-                    &metrics::CONTEXT,
-                    1,
-                    &[metrics::request::add_attributes(
-                        "connector",
-                        mandate.connector,
-                    )],
-                );
-                resp.payment_method_id = Some(mandate.payment_method_id);
+                        .to_not_found_response(errors::ApiErrorResponse::MandateNotFound)?;
+                    let mandate = match mandate.mandate_type {
+                        storage_enums::MandateType::SingleUse => state
+                            .store
+                            .update_mandate_by_merchant_id_mandate_id(
+                                &resp.merchant_id,
+                                mandate_id,
+                                storage::MandateUpdate::StatusUpdate {
+                                    mandate_status: storage_enums::MandateStatus::Revoked,
+                                },
+                            )
+                            .await
+                            .change_context(errors::ApiErrorResponse::MandateUpdateFailed),
+                        storage_enums::MandateType::MultiUse => state
+                            .store
+                            .update_mandate_by_merchant_id_mandate_id(
+                                &resp.merchant_id,
+                                mandate_id,
+                                storage::MandateUpdate::CaptureAmountUpdate {
+                                    amount_captured: Some(
+                                        mandate.amount_captured.unwrap_or(0)
+                                            + resp.request.get_amount(),
+                                    ),
+                                },
+                            )
+                            .await
+                            .change_context(errors::ApiErrorResponse::MandateUpdateFailed),
+                    }?;
+                    metrics::SUBSEQUENT_MANDATE_PAYMENT.add(
+                        &metrics::CONTEXT,
+                        1,
+                        &[metrics::request::add_attributes(
+                            "connector",
+                            mandate.connector,
+                        )],
+                    );
+                    resp.payment_method_id = Some(mandate.payment_method_id);
+                }
             }
             None => {
                 if resp.request.get_setup_mandate_details().is_some() {
@@ -409,7 +413,7 @@ where
                         logger::debug!("{:?}", new_mandate_data);
                         resp.request
                         .set_mandate_id(Some(api_models::payments::MandateIds {
-                            mandate_id: new_mandate_data.mandate_id.clone(),
+                            mandate_id: Some(new_mandate_data.mandate_id.clone()),
                             mandate_reference_id: new_mandate_data
                                 .connector_mandate_ids
                                 .clone()
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index ca3bbfd0651..2544bffda72 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -13,7 +13,10 @@ pub mod types;
 
 use std::{fmt::Debug, marker::PhantomData, ops::Deref, time::Instant, vec::IntoIter};
 
-use api_models::{self, enums, payments::HeaderPayload};
+use api_models::{
+    self, enums,
+    payments::{self as payments_api, HeaderPayload},
+};
 use common_utils::{ext_traits::AsyncExt, pii, types::Surcharge};
 use data_models::mandates::{CustomerAcceptance, MandateData};
 use diesel_models::{ephemeral_key, fraud_check::FraudCheck};
@@ -2061,9 +2064,11 @@ where
     pub customer_acceptance: Option<CustomerAcceptance>,
     pub address: PaymentAddress,
     pub token: Option<String>,
+    pub token_data: Option<storage::PaymentTokenData>,
     pub confirm: Option<bool>,
     pub force_sync: Option<bool>,
     pub payment_method_data: Option<api::PaymentMethodData>,
+    pub payment_method_info: Option<storage::PaymentMethod>,
     pub refunds: Vec<storage::Refund>,
     pub disputes: Vec<storage::Dispute>,
     pub attempts: Option<Vec<storage::PaymentAttempt>>,
@@ -2652,7 +2657,10 @@ where
         .as_ref()
         .zip(payment_data.payment_attempt.payment_method_type.as_ref())
     {
-        if let Some(choice) = pre_routing_results.get(storage_pm_type) {
+        if let (Some(choice), None) = (
+            pre_routing_results.get(storage_pm_type),
+            &payment_data.token_data,
+        ) {
             let connector_data = api::ConnectorData::get_connector_by_name(
                 &state.conf.connectors,
                 &choice.connector.to_string(),
@@ -2687,6 +2695,12 @@ where
         .attach_printable("Failed execution of straight through routing")?;
 
         if check_eligibility {
+            #[cfg(feature = "business_profile_routing")]
+            let profile_id = payment_data.payment_intent.profile_id.clone();
+
+            #[cfg(not(feature = "business_profile_routing"))]
+            let _profile_id: Option<String> = None;
+
             connectors = routing::perform_eligibility_analysis_with_fallback(
                 &state.clone(),
                 key_store,
@@ -2695,20 +2709,13 @@ where
                 &TransactionData::Payment(payment_data),
                 eligible_connectors,
                 #[cfg(feature = "business_profile_routing")]
-                payment_data.payment_intent.profile_id.clone(),
+                profile_id,
             )
             .await
             .change_context(errors::ApiErrorResponse::InternalServerError)
             .attach_printable("failed eligibility analysis and fallback")?;
         }
 
-        let first_connector_choice = connectors
-            .first()
-            .ok_or(errors::ApiErrorResponse::IncorrectPaymentMethodConfiguration)
-            .into_report()
-            .attach_printable("Empty connector list returned")?
-            .clone();
-
         let connector_data = connectors
             .into_iter()
             .map(|conn| {
@@ -2726,17 +2733,11 @@ where
             .change_context(errors::ApiErrorResponse::InternalServerError)
             .attach_printable("Invalid connector name received")?;
 
-        routing_data.routed_through = Some(first_connector_choice.connector.to_string());
-        #[cfg(feature = "connector_choice_mca_id")]
-        {
-            routing_data.merchant_connector_id = first_connector_choice.merchant_connector_id;
-        }
-        #[cfg(not(feature = "connector_choice_mca_id"))]
-        {
-            routing_data.business_sub_label = first_connector_choice.sub_label.clone();
-        }
-        routing_data.routing_info.algorithm = Some(routing_algorithm);
-        return Ok(api::ConnectorCallType::Retryable(connector_data));
+        return decide_connector_for_token_based_mit_flow(
+            payment_data,
+            routing_data,
+            connector_data,
+        );
     }
 
     if let Some(ref routing_algorithm) = routing_data.routing_info.algorithm {
@@ -2748,6 +2749,12 @@ where
         .attach_printable("Failed execution of straight through routing")?;
 
         if check_eligibility {
+            #[cfg(feature = "business_profile_routing")]
+            let profile_id = payment_data.payment_intent.profile_id.clone();
+
+            #[cfg(not(feature = "business_profile_routing"))]
+            let _profile_id: Option<String> = None;
+
             connectors = routing::perform_eligibility_analysis_with_fallback(
                 &state,
                 key_store,
@@ -2756,20 +2763,13 @@ where
                 &TransactionData::Payment(payment_data),
                 eligible_connectors,
                 #[cfg(feature = "business_profile_routing")]
-                payment_data.payment_intent.profile_id.clone(),
+                profile_id,
             )
             .await
             .change_context(errors::ApiErrorResponse::InternalServerError)
             .attach_printable("failed eligibility analysis and fallback")?;
         }
 
-        let first_connector_choice = connectors
-            .first()
-            .ok_or(errors::ApiErrorResponse::IncorrectPaymentMethodConfiguration)
-            .into_report()
-            .attach_printable("Empty connector list returned")?
-            .clone();
-
         let connector_data = connectors
             .into_iter()
             .map(|conn| {
@@ -2787,16 +2787,11 @@ where
             .change_context(errors::ApiErrorResponse::InternalServerError)
             .attach_printable("Invalid connector name received")?;
 
-        routing_data.routed_through = Some(first_connector_choice.connector.to_string());
-        #[cfg(feature = "connector_choice_mca_id")]
-        {
-            routing_data.merchant_connector_id = first_connector_choice.merchant_connector_id;
-        }
-        #[cfg(not(feature = "connector_choice_mca_id"))]
-        {
-            routing_data.business_sub_label = first_connector_choice.sub_label;
-        }
-        return Ok(api::ConnectorCallType::Retryable(connector_data));
+        return decide_connector_for_token_based_mit_flow(
+            payment_data,
+            routing_data,
+            connector_data,
+        );
     }
 
     route_connector_v1(
@@ -2804,13 +2799,107 @@ where
         merchant_account,
         business_profile,
         key_store,
-        &TransactionData::Payment(payment_data),
+        TransactionData::Payment(payment_data),
         routing_data,
         eligible_connectors,
     )
     .await
 }
 
+pub fn decide_connector_for_token_based_mit_flow<F: Clone>(
+    payment_data: &mut PaymentData<F>,
+    routing_data: &mut storage::RoutingData,
+    connectors: Vec<api::ConnectorData>,
+) -> RouterResult<ConnectorCallType> {
+    if let Some((storage_enums::FutureUsage::OffSession, _)) = payment_data
+        .payment_intent
+        .setup_future_usage
+        .zip(payment_data.token_data.as_ref())
+    {
+        logger::debug!("performing routing for token-based MIT flow");
+
+        let payment_method_info = payment_data
+            .payment_method_info
+            .as_ref()
+            .get_required_value("payment_method_info")?;
+
+        let connector_mandate_details = payment_method_info
+            .connector_mandate_details
+            .clone()
+            .map(|details| {
+                details.parse_value::<storage::PaymentsMandateReference>("connector_mandate_details")
+            })
+            .transpose()
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("unable to deserialize connector mandate details")?
+            .get_required_value("connector_mandate_details")
+            .change_context(errors::ApiErrorResponse::IncorrectPaymentMethodConfiguration)
+            .attach_printable("no eligible connector found for token-based MIT flow since there were no connector mandate details")?;
+
+        let mut connector_choice = None;
+        for connector_data in connectors {
+            if let Some(merchant_connector_id) = connector_data.merchant_connector_id.as_ref() {
+                if let Some(mandate_reference_record) =
+                    connector_mandate_details.get(merchant_connector_id)
+                {
+                    connector_choice = Some((connector_data, mandate_reference_record.clone()));
+                    break;
+                }
+            }
+        }
+
+        let (chosen_connector_data, mandate_reference_record) = connector_choice
+            .get_required_value("connector_choice")
+            .change_context(errors::ApiErrorResponse::IncorrectPaymentMethodConfiguration)
+            .attach_printable("no eligible connector found for token-based MIT payment")?;
+
+        routing_data.routed_through = Some(chosen_connector_data.connector_name.to_string());
+        #[cfg(feature = "connector_choice_mca_id")]
+        {
+            routing_data.merchant_connector_id =
+                chosen_connector_data.merchant_connector_id.clone();
+        }
+
+        payment_data.mandate_id = Some(payments_api::MandateIds {
+            mandate_id: None,
+            mandate_reference_id: Some(payments_api::MandateReferenceId::ConnectorMandateId(
+                payments_api::ConnectorMandateReferenceId {
+                    connector_mandate_id: Some(
+                        mandate_reference_record.connector_mandate_id.clone(),
+                    ),
+                    payment_method_id: Some(payment_method_info.payment_method_id.clone()),
+                    update_history: None,
+                },
+            )),
+        });
+
+        payment_data.recurring_mandate_payment_data = Some(RecurringMandatePaymentData {
+            payment_method_type: mandate_reference_record.payment_method_type,
+            original_payment_authorized_amount: mandate_reference_record
+                .original_payment_authorized_amount,
+            original_payment_authorized_currency: mandate_reference_record
+                .original_payment_authorized_currency,
+        });
+
+        Ok(api::ConnectorCallType::PreDetermined(chosen_connector_data))
+    } else {
+        let first_choice = connectors
+            .first()
+            .ok_or(errors::ApiErrorResponse::IncorrectPaymentMethodConfiguration)
+            .into_report()
+            .attach_printable("no eligible connector found for payment")?
+            .clone();
+
+        routing_data.routed_through = Some(first_choice.connector_name.to_string());
+        #[cfg(feature = "connector_choice_mca_id")]
+        {
+            routing_data.merchant_connector_id = first_choice.merchant_connector_id;
+        }
+
+        Ok(api::ConnectorCallType::Retryable(connectors))
+    }
+}
+
 pub fn should_add_task_to_process_tracker<F: Clone>(payment_data: &PaymentData<F>) -> bool {
     let connector = payment_data.payment_attempt.connector.as_deref();
 
@@ -2940,7 +3029,7 @@ pub async fn route_connector_v1<F>(
     merchant_account: &domain::MerchantAccount,
     business_profile: &storage::business_profile::BusinessProfile,
     key_store: &domain::MerchantKeyStore,
-    transaction_data: &TransactionData<'_, F>,
+    transaction_data: TransactionData<'_, F>,
     routing_data: &mut storage::RoutingData,
     eligible_connectors: Option<Vec<api_models::enums::RoutableConnectors>>,
 ) -> RouterResult<ConnectorCallType>
@@ -2948,7 +3037,7 @@ where
     F: Send + Clone,
 {
     #[allow(unused_variables)]
-    let (profile_id, routing_algorithm) = match transaction_data {
+    let (profile_id, routing_algorithm) = match &transaction_data {
         TransactionData::Payment(payment_data) => {
             if cfg!(feature = "business_profile_routing") {
                 (
@@ -2983,7 +3072,7 @@ where
         state,
         &merchant_account.merchant_id,
         algorithm_ref,
-        transaction_data,
+        &transaction_data,
     )
     .await
     .change_context(errors::ApiErrorResponse::InternalServerError)?;
@@ -2993,7 +3082,7 @@ where
         key_store,
         merchant_account.modified_at.assume_utc().unix_timestamp(),
         connectors,
-        transaction_data,
+        &transaction_data,
         eligible_connectors,
         #[cfg(feature = "business_profile_routing")]
         profile_id,
@@ -3002,6 +3091,7 @@ where
     .change_context(errors::ApiErrorResponse::InternalServerError)
     .attach_printable("failed eligibility analysis and fallback")?;
 
+    #[cfg(feature = "payouts")]
     let first_connector_choice = connectors
         .first()
         .ok_or(errors::ApiErrorResponse::IncorrectPaymentMethodConfiguration)
@@ -3009,17 +3099,6 @@ where
         .attach_printable("Empty connector list returned")?
         .clone();
 
-    routing_data.routed_through = Some(first_connector_choice.connector.to_string());
-
-    #[cfg(feature = "connector_choice_mca_id")]
-    {
-        routing_data.merchant_connector_id = first_connector_choice.merchant_connector_id;
-    }
-    #[cfg(not(feature = "connector_choice_mca_id"))]
-    {
-        routing_data.business_sub_label = first_connector_choice.sub_label;
-    }
-
     let connector_data = connectors
         .into_iter()
         .map(|conn| {
@@ -3037,7 +3116,27 @@ where
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Invalid connector name received")?;
 
-    Ok(ConnectorCallType::Retryable(connector_data))
+    match transaction_data {
+        TransactionData::Payment(payment_data) => {
+            decide_connector_for_token_based_mit_flow(payment_data, routing_data, connector_data)
+        }
+
+        #[cfg(feature = "payouts")]
+        TransactionData::Payout(_) => {
+            routing_data.routed_through = Some(first_connector_choice.connector.to_string());
+
+            #[cfg(feature = "connector_choice_mca_id")]
+            {
+                routing_data.merchant_connector_id = first_connector_choice.merchant_connector_id;
+            }
+            #[cfg(not(feature = "connector_choice_mca_id"))]
+            {
+                routing_data.business_sub_label = first_connector_choice.sub_label;
+            }
+
+            Ok(ConnectorCallType::Retryable(connector_data))
+        }
+    }
 }
 
 #[instrument(skip_all)]
diff --git a/crates/router/src/core/payments/flows/authorize_flow.rs b/crates/router/src/core/payments/flows/authorize_flow.rs
index 212e694ac92..efc0e8852da 100644
--- a/crates/router/src/core/payments/flows/authorize_flow.rs
+++ b/crates/router/src/core/payments/flows/authorize_flow.rs
@@ -100,6 +100,8 @@ impl Feature<api::Authorize, types::PaymentsAuthorizeData> for types::PaymentsAu
                         merchant_account,
                         self.request.payment_method_type,
                         key_store,
+                        Some(resp.request.amount),
+                        Some(resp.request.currency),
                     ))
                     .await?;
 
@@ -134,6 +136,8 @@ impl Feature<api::Authorize, types::PaymentsAuthorizeData> for types::PaymentsAu
                         &merchant_account,
                         self.request.payment_method_type,
                         &key_store,
+                        Some(resp.request.amount),
+                        Some(resp.request.currency),
                     ))
                     .await;
 
diff --git a/crates/router/src/core/payments/flows/setup_mandate_flow.rs b/crates/router/src/core/payments/flows/setup_mandate_flow.rs
index 65342a1f06b..189a8a0a3ed 100644
--- a/crates/router/src/core/payments/flows/setup_mandate_flow.rs
+++ b/crates/router/src/core/payments/flows/setup_mandate_flow.rs
@@ -107,6 +107,8 @@ impl Feature<api::SetupMandate, types::SetupMandateRequestData> for types::Setup
                 merchant_account,
                 self.request.payment_method_type,
                 key_store,
+                resp.request.amount,
+                Some(resp.request.currency),
             ))
             .await?;
 
@@ -244,6 +246,8 @@ impl types::SetupMandateRouterData {
                     merchant_account,
                     payment_method_type,
                     key_store,
+                    resp.request.amount,
+                    Some(resp.request.currency),
                 ))
                 .await?;
 
@@ -330,6 +334,8 @@ impl types::SetupMandateRouterData {
                 merchant_account,
                 self.request.payment_method_type,
                 key_store,
+                resp.request.amount,
+                Some(resp.request.currency),
             ))
             .await?
             .0;
diff --git a/crates/router/src/core/payments/helpers.rs b/crates/router/src/core/payments/helpers.rs
index 76ffed499d1..68821ee5032 100644
--- a/crates/router/src/core/payments/helpers.rs
+++ b/crates/router/src/core/payments/helpers.rs
@@ -1602,6 +1602,90 @@ pub async fn retrieve_card_with_permanent_token(
     Ok(api::PaymentMethodData::Card(api_card))
 }
 
+pub async fn retrieve_payment_method_from_db_with_token_data(
+    state: &AppState,
+    token_data: &storage::PaymentTokenData,
+) -> RouterResult<Option<storage::PaymentMethod>> {
+    match token_data {
+        storage::PaymentTokenData::PermanentCard(data) => {
+            if let Some(ref payment_method_id) = data.payment_method_id {
+                state
+                    .store
+                    .find_payment_method(payment_method_id)
+                    .await
+                    .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)
+                    .attach_printable("error retrieving payment method from DB")
+                    .map(Some)
+            } else {
+                Ok(None)
+            }
+        }
+
+        storage::PaymentTokenData::WalletToken(data) => state
+            .store
+            .find_payment_method(&data.payment_method_id)
+            .await
+            .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)
+            .attach_printable("error retrieveing payment method from DB")
+            .map(Some),
+
+        storage::PaymentTokenData::Temporary(_)
+        | storage::PaymentTokenData::TemporaryGeneric(_)
+        | storage::PaymentTokenData::Permanent(_)
+        | storage::PaymentTokenData::AuthBankDebit(_) => Ok(None),
+    }
+}
+
+pub async fn retrieve_payment_token_data(
+    state: &AppState,
+    token: String,
+    payment_method: Option<storage_enums::PaymentMethod>,
+) -> RouterResult<storage::PaymentTokenData> {
+    let redis_conn = state
+        .store
+        .get_redis_conn()
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to get redis connection")?;
+
+    let key = format!(
+        "pm_token_{}_{}_hyperswitch",
+        token,
+        payment_method.get_required_value("payment_method")?
+    );
+
+    let token_data_string = redis_conn
+        .get_key::<Option<String>>(&key)
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to fetch the token from redis")?
+        .ok_or(error_stack::Report::new(
+            errors::ApiErrorResponse::UnprocessableEntity {
+                message: "Token is invalid or expired".to_owned(),
+            },
+        ))?;
+
+    let token_data_result = token_data_string
+        .clone()
+        .parse_struct("PaymentTokenData")
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("failed to deserialize hyperswitch token data");
+
+    let token_data = match token_data_result {
+        Ok(data) => data,
+        Err(e) => {
+            // The purpose of this logic is backwards compatibility to support tokens
+            // in redis that might be following the old format.
+            if token_data_string.starts_with('{') {
+                return Err(e);
+            } else {
+                storage::PaymentTokenData::temporary_generic(token_data_string)
+            }
+        }
+    };
+
+    Ok(token_data)
+}
+
 pub async fn make_pm_data<'a, F: Clone, R, Ctx: PaymentMethodRetrieve>(
     operation: BoxedOperation<'a, F, R, Ctx>,
     state: &'a AppState,
@@ -1630,72 +1714,13 @@ pub async fn make_pm_data<'a, F: Clone, R, Ctx: PaymentMethodRetrieve>(
         }
     }
 
-    let token = payment_data.token.clone();
-
-    let hyperswitch_token = match payment_data.mandate_id {
-        Some(_) => token.map(storage::PaymentTokenData::temporary_generic),
-        None => {
-            if let Some(token) = token {
-                let redis_conn = state
-                    .store
-                    .get_redis_conn()
-                    .change_context(errors::ApiErrorResponse::InternalServerError)
-                    .attach_printable("Failed to get redis connection")?;
-
-                let key = format!(
-                    "pm_token_{}_{}_hyperswitch",
-                    token,
-                    payment_data
-                        .payment_attempt
-                        .payment_method
-                        .to_owned()
-                        .get_required_value("payment_method")?,
-                );
-
-                let token_data_string = redis_conn
-                    .get_key::<Option<String>>(&key)
-                    .await
-                    .change_context(errors::ApiErrorResponse::InternalServerError)
-                    .attach_printable("Failed to fetch the token from redis")?
-                    .ok_or(error_stack::Report::new(
-                        errors::ApiErrorResponse::UnprocessableEntity {
-                            message: "Token is invalid or expired".to_owned(),
-                        },
-                    ))?;
-
-                let token_data_result = token_data_string
-                    .clone()
-                    .parse_struct("PaymentTokenData")
-                    .change_context(errors::ApiErrorResponse::InternalServerError)
-                    .attach_printable("failed to deserialize hyperswitch token data");
-
-                let token_data = match token_data_result {
-                    Ok(data) => data,
-                    Err(e) => {
-                        // The purpose of this logic is backwards compatibility to support tokens
-                        // in redis that might be following the old format.
-                        if token_data_string.starts_with('{') {
-                            return Err(e);
-                        } else {
-                            storage::PaymentTokenData::temporary_generic(token_data_string)
-                        }
-                    }
-                };
-
-                Some(token_data)
-            } else {
-                None
-            }
-        }
-    };
-
     // TODO: Handle case where payment method and token both are present in request properly.
-    let (payment_method, pm_id) = match (request, hyperswitch_token) {
+    let (payment_method, pm_id) = match (request, payment_data.token_data.as_ref()) {
         (_, Some(hyperswitch_token)) => {
             let pm_data = Ctx::retrieve_payment_method_with_token(
                 state,
                 merchant_key_store,
-                &hyperswitch_token,
+                hyperswitch_token,
                 &payment_data.payment_intent,
                 card_token_data.as_ref(),
                 customer,
diff --git a/crates/router/src/core/payments/operations/payment_approve.rs b/crates/router/src/core/payments/operations/payment_approve.rs
index ca6ed3e4ebd..6ad14d8fc4e 100644
--- a/crates/router/src/core/payments/operations/payment_approve.rs
+++ b/crates/router/src/core/payments/operations/payment_approve.rs
@@ -149,6 +149,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             setup_mandate: None,
             customer_acceptance: None,
             token: None,
+            token_data: None,
             address: PaymentAddress {
                 shipping: shipping_address.as_ref().map(|a| a.into()),
                 billing: billing_address.as_ref().map(|a| a.into()),
@@ -158,6 +159,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             },
             confirm: None,
             payment_method_data: None,
+            payment_method_info: None,
             force_sync: None,
             refunds: vec![],
             disputes: vec![],
diff --git a/crates/router/src/core/payments/operations/payment_cancel.rs b/crates/router/src/core/payments/operations/payment_cancel.rs
index 7d5bf73fc1f..2439646b3da 100644
--- a/crates/router/src/core/payments/operations/payment_cancel.rs
+++ b/crates/router/src/core/payments/operations/payment_cancel.rs
@@ -157,6 +157,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             setup_mandate: None,
             customer_acceptance: None,
             token: None,
+            token_data: None,
             address: PaymentAddress {
                 shipping: shipping_address.as_ref().map(|a| a.into()),
                 billing: billing_address.as_ref().map(|a| a.into()),
@@ -166,6 +167,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             },
             confirm: None,
             payment_method_data: None,
+            payment_method_info: None,
             force_sync: None,
             refunds: vec![],
             disputes: vec![],
diff --git a/crates/router/src/core/payments/operations/payment_capture.rs b/crates/router/src/core/payments/operations/payment_capture.rs
index d67151897aa..d643c86698d 100644
--- a/crates/router/src/core/payments/operations/payment_capture.rs
+++ b/crates/router/src/core/payments/operations/payment_capture.rs
@@ -202,6 +202,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             setup_mandate: None,
             customer_acceptance: None,
             token: None,
+            token_data: None,
             address: payments::PaymentAddress {
                 shipping: shipping_address.as_ref().map(|a| a.into()),
                 billing: billing_address.as_ref().map(|a| a.into()),
@@ -211,6 +212,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             },
             confirm: None,
             payment_method_data: None,
+            payment_method_info: None,
             refunds: vec![],
             disputes: vec![],
             attempts: None,
diff --git a/crates/router/src/core/payments/operations/payment_complete_authorize.rs b/crates/router/src/core/payments/operations/payment_complete_authorize.rs
index 22ba50c07b9..79701db575b 100644
--- a/crates/router/src/core/payments/operations/payment_complete_authorize.rs
+++ b/crates/router/src/core/payments/operations/payment_complete_authorize.rs
@@ -131,6 +131,12 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             }
         }
 
+        let token_data = if let Some(token) = token.clone() {
+            Some(helpers::retrieve_payment_token_data(state, token, payment_method).await?)
+        } else {
+            None
+        };
+
         payment_attempt.payment_method = payment_method.or(payment_attempt.payment_method);
         payment_attempt.browser_info = browser_info;
         payment_attempt.payment_method_type =
@@ -247,6 +253,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             setup_mandate,
             customer_acceptance: None,
             token,
+            token_data,
             address: PaymentAddress {
                 shipping: shipping_address.as_ref().map(|a| a.into()),
                 billing: billing_address.as_ref().map(|a| a.into()),
@@ -259,6 +266,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
                 .payment_method_data
                 .as_ref()
                 .map(|pmd| pmd.payment_method_data.clone()),
+            payment_method_info: None,
             force_sync: None,
             refunds: vec![],
             disputes: vec![],
diff --git a/crates/router/src/core/payments/operations/payment_confirm.rs b/crates/router/src/core/payments/operations/payment_confirm.rs
index 2f548129a65..96de0cc924b 100644
--- a/crates/router/src/core/payments/operations/payment_confirm.rs
+++ b/crates/router/src/core/payments/operations/payment_confirm.rs
@@ -383,6 +383,23 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             &token,
         )?;
 
+        let (token_data, payment_method_info) = if let Some(token) = token.clone() {
+            let token_data = helpers::retrieve_payment_token_data(
+                state,
+                token,
+                payment_method.or(payment_attempt.payment_method),
+            )
+            .await?;
+
+            let payment_method_info =
+                helpers::retrieve_payment_method_from_db_with_token_data(state, &token_data)
+                    .await?;
+
+            (Some(token_data), payment_method_info)
+        } else {
+            (None, None)
+        };
+
         payment_attempt.payment_method = payment_method.or(payment_attempt.payment_method);
         payment_attempt.browser_info = browser_info;
         payment_attempt.payment_method_type =
@@ -548,6 +565,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             setup_mandate,
             customer_acceptance,
             token,
+            token_data,
             address: PaymentAddress {
                 shipping: shipping_address.as_ref().map(|a| a.into()),
                 billing: billing_address.as_ref().map(|a| a.into()),
@@ -557,6 +575,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             },
             confirm: request.confirm,
             payment_method_data: payment_method_data_after_card_bin_call,
+            payment_method_info,
             force_sync: None,
             refunds: vec![],
             disputes: vec![],
diff --git a/crates/router/src/core/payments/operations/payment_create.rs b/crates/router/src/core/payments/operations/payment_create.rs
index b4ae0285dd1..bcc13ea1cef 100644
--- a/crates/router/src/core/payments/operations/payment_create.rs
+++ b/crates/router/src/core/payments/operations/payment_create.rs
@@ -302,7 +302,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
                         mandate_obj.connector_mandate_ids,
                     ) {
                         (Some(network_tx_id), _) => Ok(api_models::payments::MandateIds {
-                            mandate_id: mandate_obj.mandate_id,
+                            mandate_id: Some(mandate_obj.mandate_id),
                             mandate_reference_id: Some(
                                 api_models::payments::MandateReferenceId::NetworkMandateId(
                                     network_tx_id,
@@ -314,7 +314,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
                         .change_context(errors::ApiErrorResponse::MandateNotFound)
                         .map(|connector_id: api_models::payments::ConnectorMandateReferenceId| {
                             api_models::payments::MandateIds {
-                                mandate_id: mandate_obj.mandate_id,
+                                mandate_id: Some(mandate_obj.mandate_id),
                                 mandate_reference_id: Some(api_models::payments::MandateReferenceId::ConnectorMandateId(
                                 api_models::payments::ConnectorMandateReferenceId{
                                     connector_mandate_id: connector_id.connector_mandate_id,
@@ -325,7 +325,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
                             }
                          }),
                         (_, _) => Ok(api_models::payments::MandateIds {
-                            mandate_id: mandate_obj.mandate_id,
+                            mandate_id: Some(mandate_obj.mandate_id),
                             mandate_reference_id: None,
                         }),
                     }
@@ -390,6 +390,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             setup_mandate,
             customer_acceptance,
             token,
+            token_data: None,
             address: PaymentAddress {
                 shipping: shipping_address.as_ref().map(|a| a.into()),
                 billing: billing_address.as_ref().map(|a| a.into()),
@@ -399,6 +400,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             },
             confirm: request.confirm,
             payment_method_data: payment_method_data_after_card_bin_call,
+            payment_method_info: None,
             refunds: vec![],
             disputes: vec![],
             attempts: None,
diff --git a/crates/router/src/core/payments/operations/payment_reject.rs b/crates/router/src/core/payments/operations/payment_reject.rs
index 66bab03d047..36197a9463b 100644
--- a/crates/router/src/core/payments/operations/payment_reject.rs
+++ b/crates/router/src/core/payments/operations/payment_reject.rs
@@ -145,6 +145,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             setup_mandate: None,
             customer_acceptance: None,
             token: None,
+            token_data: None,
             address: PaymentAddress {
                 shipping: shipping_address.as_ref().map(|a| a.into()),
                 billing: billing_address.as_ref().map(|a| a.into()),
@@ -154,6 +155,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             },
             confirm: None,
             payment_method_data: None,
+            payment_method_info: None,
             force_sync: None,
             refunds: vec![],
             disputes: vec![],
diff --git a/crates/router/src/core/payments/operations/payment_response.rs b/crates/router/src/core/payments/operations/payment_response.rs
index 18e568eb9b8..eaded6b5e65 100644
--- a/crates/router/src/core/payments/operations/payment_response.rs
+++ b/crates/router/src/core/payments/operations/payment_response.rs
@@ -654,7 +654,7 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
                                 mandate_id: payment_data
                                     .mandate_id
                                     .clone()
-                                    .map(|mandate| mandate.mandate_id),
+                                    .and_then(|mandate| mandate.mandate_id),
                                 connector_metadata,
                                 payment_token: None,
                                 error_code: error_status.clone(),
@@ -844,7 +844,7 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
             .or(payment_data
                 .mandate_id
                 .clone()
-                .map(|mandate_ids| mandate_ids.mandate_id));
+                .and_then(|mandate_ids| mandate_ids.mandate_id));
     let m_router_data_response = router_data.response.clone();
     let mandate_update_fut = tokio::spawn(
         async move {
diff --git a/crates/router/src/core/payments/operations/payment_session.rs b/crates/router/src/core/payments/operations/payment_session.rs
index e64b41ed383..d498e1baca4 100644
--- a/crates/router/src/core/payments/operations/payment_session.rs
+++ b/crates/router/src/core/payments/operations/payment_session.rs
@@ -169,6 +169,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             mandate_connector: None,
             customer_acceptance: None,
             token: None,
+            token_data: None,
             setup_mandate: None,
             address: payments::PaymentAddress {
                 shipping: shipping_address.as_ref().map(|a| a.into()),
@@ -179,6 +180,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             },
             confirm: None,
             payment_method_data: None,
+            payment_method_info: None,
             force_sync: None,
             refunds: vec![],
             disputes: vec![],
diff --git a/crates/router/src/core/payments/operations/payment_start.rs b/crates/router/src/core/payments/operations/payment_start.rs
index 885d2593ba8..f1264868ff7 100644
--- a/crates/router/src/core/payments/operations/payment_start.rs
+++ b/crates/router/src/core/payments/operations/payment_start.rs
@@ -114,6 +114,15 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
         )
         .await?;
 
+        let token_data = if let Some(token) = payment_attempt.payment_token.clone() {
+            Some(
+                helpers::retrieve_payment_token_data(state, token, payment_attempt.payment_method)
+                    .await?,
+            )
+        } else {
+            None
+        };
+
         payment_intent.shipping_address_id = shipping_address.clone().map(|i| i.address_id);
         payment_intent.billing_address_id = billing_address.clone().map(|i| i.address_id);
 
@@ -147,6 +156,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             setup_mandate: None,
             customer_acceptance: None,
             token: payment_attempt.payment_token.clone(),
+            token_data,
             address: PaymentAddress {
                 shipping: shipping_address.as_ref().map(|a| a.into()),
                 billing: billing_address.as_ref().map(|a| a.into()),
@@ -157,6 +167,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             confirm: Some(payment_attempt.confirm),
             payment_attempt,
             payment_method_data: None,
+            payment_method_info: None,
             force_sync: None,
             refunds: vec![],
             disputes: vec![],
diff --git a/crates/router/src/core/payments/operations/payment_status.rs b/crates/router/src/core/payments/operations/payment_status.rs
index f36a2eea81d..66f0527c35d 100644
--- a/crates/router/src/core/payments/operations/payment_status.rs
+++ b/crates/router/src/core/payments/operations/payment_status.rs
@@ -410,13 +410,14 @@ async fn get_tracker_for_sync<
             .mandate_id
             .clone()
             .map(|id| api_models::payments::MandateIds {
-                mandate_id: id,
+                mandate_id: Some(id),
                 mandate_reference_id: None,
             }),
         mandate_connector: None,
         setup_mandate: None,
         customer_acceptance: None,
         token: None,
+        token_data: None,
         address: PaymentAddress {
             shipping: shipping_address.as_ref().map(|a| a.into()),
             billing: billing_address.as_ref().map(|a| a.into()),
@@ -426,6 +427,7 @@ async fn get_tracker_for_sync<
         },
         confirm: Some(request.force_sync),
         payment_method_data: None,
+        payment_method_info: None,
         force_sync: Some(
             request.force_sync
                 && (helpers::check_force_psync_precondition(&payment_attempt.status)
diff --git a/crates/router/src/core/payments/operations/payment_update.rs b/crates/router/src/core/payments/operations/payment_update.rs
index c5fda22ed87..0ce77dfcd13 100644
--- a/crates/router/src/core/payments/operations/payment_update.rs
+++ b/crates/router/src/core/payments/operations/payment_update.rs
@@ -254,6 +254,12 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             )?;
         }
 
+        let token_data = if let Some(token) = token.clone() {
+            Some(helpers::retrieve_payment_token_data(state, token, payment_method).await?)
+        } else {
+            None
+        };
+
         let mandate_id = request
             .mandate_id
             .as_ref()
@@ -268,7 +274,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
                         mandate_obj.connector_mandate_ids,
                     ) {
                         (Some(network_tx_id), _) => Ok(api_models::payments::MandateIds {
-                            mandate_id: mandate_obj.mandate_id,
+                            mandate_id: Some(mandate_obj.mandate_id),
                             mandate_reference_id: Some(
                                 api_models::payments::MandateReferenceId::NetworkMandateId(
                                     network_tx_id,
@@ -280,14 +286,14 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
                         .change_context(errors::ApiErrorResponse::MandateNotFound)
                         .map(|connector_id: api_models::payments::ConnectorMandateReferenceId| {
                             api_models::payments::MandateIds {
-                                mandate_id: mandate_obj.mandate_id,
+                                mandate_id: Some(mandate_obj.mandate_id),
                                 mandate_reference_id: Some(api_models::payments::MandateReferenceId::ConnectorMandateId(
                                     api_models::payments::ConnectorMandateReferenceId {connector_mandate_id:connector_id.connector_mandate_id,payment_method_id:connector_id.payment_method_id, update_history: None },
                                 ))
                             }
                          }),
                         (_, _) => Ok(api_models::payments::MandateIds {
-                            mandate_id: mandate_obj.mandate_id,
+                            mandate_id: Some(mandate_obj.mandate_id),
                             mandate_reference_id: None,
                         }),
                     }
@@ -383,6 +389,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             mandate_id,
             mandate_connector,
             token,
+            token_data,
             setup_mandate,
             customer_acceptance,
             address: PaymentAddress {
@@ -395,6 +402,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
                 .payment_method_data
                 .as_ref()
                 .map(|pmd| pmd.payment_method_data.clone()),
+            payment_method_info: None,
             force_sync: None,
             refunds: vec![],
             disputes: vec![],
diff --git a/crates/router/src/core/payments/operations/payments_incremental_authorization.rs b/crates/router/src/core/payments/operations/payments_incremental_authorization.rs
index dbdf1c062ac..0157822cdf8 100644
--- a/crates/router/src/core/payments/operations/payments_incremental_authorization.rs
+++ b/crates/router/src/core/payments/operations/payments_incremental_authorization.rs
@@ -122,6 +122,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             setup_mandate: None,
             customer_acceptance: None,
             token: None,
+            token_data: None,
             address: PaymentAddress {
                 billing: None,
                 shipping: None,
@@ -129,6 +130,7 @@ impl<F: Send + Clone, Ctx: PaymentMethodRetrieve>
             },
             confirm: None,
             payment_method_data: None,
+            payment_method_info: None,
             force_sync: None,
             refunds: vec![],
             disputes: vec![],
diff --git a/crates/router/src/core/payments/retry.rs b/crates/router/src/core/payments/retry.rs
index 91b6061c111..2af73f29909 100644
--- a/crates/router/src/core/payments/retry.rs
+++ b/crates/router/src/core/payments/retry.rs
@@ -377,7 +377,7 @@ where
                     mandate_id: payment_data
                         .mandate_id
                         .clone()
-                        .map(|mandate| mandate.mandate_id),
+                        .and_then(|mandate| mandate.mandate_id),
                     connector_metadata,
                     payment_token: None,
                     error_code: None,
diff --git a/crates/router/src/core/payments/tokenization.rs b/crates/router/src/core/payments/tokenization.rs
index 7aaaaed92aa..4628f027a05 100644
--- a/crates/router/src/core/payments/tokenization.rs
+++ b/crates/router/src/core/payments/tokenization.rs
@@ -39,6 +39,8 @@ pub async fn save_payment_method<F: Clone, FData>(
     merchant_account: &domain::MerchantAccount,
     payment_method_type: Option<storage_enums::PaymentMethodType>,
     key_store: &domain::MerchantKeyStore,
+    amount: Option<i64>,
+    currency: Option<storage_enums::Currency>,
 ) -> RouterResult<(Option<String>, Option<common_enums::PaymentMethodStatus>)>
 where
     FData: mandate::MandateBehaviour,
@@ -94,7 +96,13 @@ where
                     .map(|future_usage| future_usage == storage_enums::FutureUsage::OffSession)
                     .unwrap_or(false)
             {
-                add_connector_mandate_details_in_payment_method(responses, connector)
+                add_connector_mandate_details_in_payment_method(
+                    responses,
+                    payment_method_type,
+                    amount,
+                    currency,
+                    connector,
+                )
             } else {
                 None
             }
@@ -648,6 +656,9 @@ pub async fn add_payment_method_token<F: Clone, T: types::Tokenizable + Clone>(
 
 fn add_connector_mandate_details_in_payment_method(
     resp: types::PaymentsResponseData,
+    payment_method_type: Option<storage_enums::PaymentMethodType>,
+    authorized_amount: Option<i64>,
+    authorized_currency: Option<storage_enums::Currency>,
     connector: &api::ConnectorData,
 ) -> Option<storage::PaymentsMandateReference> {
     let mut mandate_details = HashMap::new();
@@ -665,8 +676,20 @@ fn add_connector_mandate_details_in_payment_method(
         _ => None,
     };
 
-    if let Some(mca_id) = connector.merchant_connector_id.clone() {
-        mandate_details.insert(mca_id, connector_mandate_id);
+    if let Some((mca_id, connector_mandate_id)) = connector
+        .merchant_connector_id
+        .clone()
+        .zip(connector_mandate_id)
+    {
+        mandate_details.insert(
+            mca_id,
+            storage::PaymentsMandateReferenceRecord {
+                connector_mandate_id,
+                payment_method_type,
+                original_payment_authorized_amount: authorized_amount,
+                original_payment_authorized_currency: authorized_currency,
+            },
+        );
         Some(storage::PaymentsMandateReference(mandate_details))
     } else {
         None
diff --git a/crates/router/src/core/payments/transformers.rs b/crates/router/src/core/payments/transformers.rs
index 8e5b2c9199e..0ae04aaddce 100644
--- a/crates/router/src/core/payments/transformers.rs
+++ b/crates/router/src/core/payments/transformers.rs
@@ -327,7 +327,9 @@ where
                 phone: customer
                     .as_ref()
                     .and_then(|cus| cus.phone.as_ref().map(|s| s.to_owned())),
-                mandate_id: data.mandate_id.map(|mandate_ids| mandate_ids.mandate_id),
+                mandate_id: data
+                    .mandate_id
+                    .and_then(|mandate_ids| mandate_ids.mandate_id),
                 payment_method: data.payment_attempt.payment_method,
                 payment_method_data: payment_method_data_response,
                 payment_token: data.token,
diff --git a/crates/router/src/core/payouts/helpers.rs b/crates/router/src/core/payouts/helpers.rs
index 9fd7458155a..847c555a86d 100644
--- a/crates/router/src/core/payouts/helpers.rs
+++ b/crates/router/src/core/payouts/helpers.rs
@@ -600,7 +600,7 @@ pub async fn decide_payout_connector(
         merchant_account,
         &payout_data.business_profile,
         key_store,
-        &TransactionData::<()>::Payout(payout_data),
+        TransactionData::<()>::Payout(payout_data),
         routing_data,
         eligible_connectors,
     )
diff --git a/crates/router/src/core/routing.rs b/crates/router/src/core/routing.rs
index 31a2e05dd79..0d9c7f4f62c 100644
--- a/crates/router/src/core/routing.rs
+++ b/crates/router/src/core/routing.rs
@@ -36,12 +36,11 @@ use crate::{core::errors, services::api as service_api, types::storage};
 #[cfg(feature = "business_profile_routing")]
 use crate::{errors, services::api as service_api};
 
-#[derive(Clone)]
 pub enum TransactionData<'a, F>
 where
     F: Clone,
 {
-    Payment(&'a payments::PaymentData<F>),
+    Payment(&'a mut payments::PaymentData<F>),
     #[cfg(feature = "payouts")]
     Payout(&'a payouts::PayoutData),
 }
diff --git a/crates/router/src/types/storage/payment_method.rs b/crates/router/src/types/storage/payment_method.rs
index 8edfb8b36d8..d3339b7c4be 100644
--- a/crates/router/src/types/storage/payment_method.rs
+++ b/crates/router/src/types/storage/payment_method.rs
@@ -1,4 +1,7 @@
-use std::collections::HashMap;
+use std::{
+    collections::HashMap,
+    ops::{Deref, DerefMut},
+};
 
 use api_models::payment_methods;
 use diesel_models::enums;
@@ -40,7 +43,7 @@ pub struct WalletTokenData {
     pub payment_method_id: String,
 }
 
-#[derive(Debug, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 #[serde(tag = "kind", rename_all = "snake_case")]
 pub enum PaymentTokenData {
     // The variants 'Temporary' and 'Permanent' are added for backwards compatibility
@@ -76,4 +79,26 @@ impl PaymentTokenData {
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-pub struct PaymentsMandateReference(pub HashMap<String, Option<String>>);
+pub struct PaymentsMandateReferenceRecord {
+    pub connector_mandate_id: String,
+    pub payment_method_type: Option<common_enums::PaymentMethodType>,
+    pub original_payment_authorized_amount: Option<i64>,
+    pub original_payment_authorized_currency: Option<common_enums::Currency>,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct PaymentsMandateReference(pub HashMap<String, PaymentsMandateReferenceRecord>);
+
+impl Deref for PaymentsMandateReference {
+    type Target = HashMap<String, PaymentsMandateReferenceRecord>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl DerefMut for PaymentsMandateReference {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
+}

## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
This PR adds routing support for token-based MIT payments. Basically, the merchant can have their customers set up their payment method for off session payments with Hyperswitch. Behind the scenes, Hyperswitch sets up an MIT/Mandate with the connector.

Now, when the merchant wants to do off-session payments using the customer's saved payment method by using a token to select the payment method, the backend needs to ensure that the payment goes through only those connectors for which that payment method is set up for MITs. This PR adds this check.

### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->
The merchant may want to do off-session payments by passing a payment method token on behalf of their customers. This would require the smart router to choose an appropriate connector for the payment on the basis of whether the selected payment method is set up with the connector for off-session MIT payments.

## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->
1. Create a Merchant Account, and a Merchant Connector Account (preferably Cybersource)
2. Create and Confirm a `Setup Mandate` request
```bash
curl --location 'http://127.0.0.1:8080/payments' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: dev_OnB9g3l1v3cqPalObMlJ874jZerIsl9hlNCoTdXzNyiB2K7Edrp3wEQYmdmnXFJM' \
--data-raw '{
    "amount": 0,
    "currency": "USD",
    "confirm": true,
    "profile_id": "pro_9ci3SqA4bSAe7g3C7jCp",
    "capture_method": "automatic",
    "customer_id": "MyCustomer123456",
    "email": "guest@example.com",
    "setup_future_usage": "off_session",
    "customer_acceptance": {
        "acceptance_type": "online"
    },
    "payment_method": "card",
    "payment_method_type": "debit",
    "payment_method_data": {
        "card": {
            "card_number": "4242424242424242",
            "card_exp_month": "03",
            "card_exp_year": "2030",
            "card_holder_name": "Test Holder",
            "card_cvc": "737"
        }
    },
    "payment_type": "setup_mandate",
    "billing": {
        "address": {
            "city": "test",
            "country": "US",
            "line1": "here",
            "line2": "there",
            "line3": "anywhere",
            "zip": "560095",
            "state": "Washington",
            "first_name": "One",
            "last_name": "Two"
        },
        "phone": {
            "number": "1234567890",
            "country_code": "+1"
        },
        "email": "guest@example.com"
    },
    "authentication_type": "no_three_ds"
}'
```
this will set the given card up for token-based MIT payments, and associate the connector MIT/Mandate details with the saved payment method in DB as shown below in the `connector_mandate_details` column
<img width="1727" alt="image" src="https://github.com/juspay/hyperswitch/assets/47862918/4c8aefe5-3079-43df-ac13-3226a291f006">


3. Next, to make a token-based MIT payment, Create a payment with `setup_future_usage = off_session`
```bash
curl --location 'http://127.0.0.1:8080/payments' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: dev_hOic0bCxCZdrp9ShPzyl9b80NQgr2m4vy4gBVyKbKbWNNRCiNBm9yCA5D7McybRs' \
--data-raw '{
    "amount": 499,
    "currency": "USD",
    "confirm": false,
    "profile_id": "pro_tsOrBWXnZfzYRJ9Jm7Dk",
    "capture_method": "automatic",
    "customer_id": "MyCustomer123456",
    "email": "guest@example.com",
    "setup_future_usage": "off_session",
    "billing": {
        "address": {
            "city": "test",
            "country": "US",
            "line1": "here",
            "line2": "there",
            "line3": "anywhere",
            "zip": "560095",
            "state": "Washington",
            "first_name": "One",
            "last_name": "Two"
        },
        "phone": {
            "number": "1234567890",
            "country_code": "+1"
        },
        "email": "guest@example.com"
    },
    "authentication_type": "no_three_ds"
}'
```

4. Do a list customer payment methods call and confirm the payment with the token (no need to pass `card_cvc`)
```bash
curl --location 'http://127.0.0.1:8080/payments/pay_5C8NWON1GUFGbMLR3tXx/confirm' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: dev_hOic0bCxCZdrp9ShPzyl9b80NQgr2m4vy4gBVyKbKbWNNRCiNBm9yCA5D7McybRs' \
--data '{
    "payment_token": "token_ZTiG3Qz4ajaZB1MnBcmQ",
    "payment_method": "card"
}'
```
This is treated as a token-based MIT payment since we passed `setup_future_usage = off_session` in create, and are confirming the payment with the token. Routing will detect this, get the `connector_mandate_details` from the payment method object associated with the token, and do a recurring mandate payment using the connector mandate details. Additionally, routing will ensure that the connector will be picked for a token-based MIT payment if and only if it has some connector mandate details saved in the payment method object. Following is the connector request sent by the backend to cybersource.
```json
{"processingInformation":{"actionList":null,"actionTokenTypes":null,"authorizationOptions":{"initiator":null,"merchantIntitiatedTransaction":{"reason":null,"originalAuthorizedAmount":"0.00"}},"commerceIndicator":"internet","capture":true,"captureOptions":null,"paymentSolution":null},"paymentInformation":{"paymentInstrument":{"id":"1323EB5C00917EC2E063A2598D0A69D7"}},"orderInformation":{"amountDetails":{"totalAmount":"2.00","currency":"USD"},"billTo":{"firstName":"One","lastName":"Two","address1":"here","locality":"bengaluru","administrativeArea":"Karnataka","postalCode":"560095","country":"IN","email":"guest@example.com"}},"clientReferenceInformation":{"code":"pay_iCiIdD8UqO3SdfAxGRrI_1"}}
```

We can see that `paymentInstrument` was set to the connector mandate ID by Routing, thus turning it into a recurring MIT payment.

5. To test and ensure that Routing picks a connector for token-based MIT payments if and only if it has connector_mandate_details for that connector, create another merchant connector account with a different connector `Y`.
6. Repeat steps 3 and 4 but in the confirm call, pass a `"routing"` value to force straight through routing to pick `Y`.
```bash
curl --location 'http://127.0.0.1:8080/payments/pay_5C8NWON1GUFGbMLR3tXx/confirm' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: dev_hOic0bCxCZdrp9ShPzyl9b80NQgr2m4vy4gBVyKbKbWNNRCiNBm9yCA5D7McybRs' \
--data '{
    "payment_token": "token_ZTiG3Qz4ajaZB1MnBcmQ",
    "routing": {
        "type": "single",
        "data": {
            "connector": "stripe",
            "merchant_connector_id": "mca_hji9rSpSJPUOYn2A9yL9"
        }
    },
    "payment_method": "card"
}'
```
Even tho we are forcing connector `Y` for the token-based MIT payment, Routing will detect that the chosen payment method does not have any MIT/Mandate set up with `Y` by checking the `connector_mandate_details` field in the payment method record, and automatically fall back to the first connector.
<img width="552" alt="image" src="https://github.com/juspay/hyperswitch/assets/47862918/5eb49d36-0c6b-43d6-8bae-9f9bf45bcd31">


## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
- [ ] I added a [CHANGELOG](/CHANGELOG.md) entry if applicable
