Bug: [FEATURE] Mandates for alternate payment methods via Stripe



### Feature Description

**Scope:**
Implementation of 
- [ ] Applepay recurring payments via Stripe connector
- [ ] Googlepay recurring payments via Stripe connector

### Possible Implementation

Recurring payments for Applepay and Googlepay needs to be built in a processor agnostic manner.
This means that if the first recurring payment is processed via processor A, the second recurring payment could be processed via processor B.

### Have you spent some time to check if this feature request has been raised before?

- [X] I checked and didn't find similar issue

### Have you read the Contributing Guidelines?

- [X] I have read the [Contributing Guidelines](https://github.com/juspay/hyperswitch/blob/main/docs/CONTRIBUTING.md)

### Are you willing to submit a PR?

None
diff --git a/config/development.toml b/config/development.toml
index 4ecd69879cf..b684c497652 100644
--- a/config/development.toml
+++ b/config/development.toml
@@ -225,8 +225,8 @@ credit = {currency = "USD"}
 debit = {currency = "USD"}
 
 [tokenization]
-stripe = { long_lived_token = false, payment_method = "wallet" }
-checkout = { long_lived_token = false, payment_method = "wallet" }
+stripe = { long_lived_token = false, payment_method = "wallet", payment_method_type = { type = "disable_only", list = "google_pay" } }
+checkout = { long_lived_token = false, payment_method = "wallet"}
 
 [connector_customer]
 connector_list = "stripe"
diff --git a/crates/api_models/src/payments.rs b/crates/api_models/src/payments.rs
index a8c69d04e52..58259baab7c 100644
--- a/crates/api_models/src/payments.rs
+++ b/crates/api_models/src/payments.rs
@@ -307,10 +307,16 @@ pub struct MandateIds {
 
 #[derive(Eq, PartialEq, Debug, serde::Deserialize, serde::Serialize, Clone)]
 pub enum MandateReferenceId {
-    ConnectorMandateId(String), // mandate_id send by connector
+    ConnectorMandateId(ConnectorMandateReferenceId), // mandate_id send by connector
     NetworkMandateId(String), // network_txns_id send by Issuer to connector, Used for PG agnostic mandate txns
 }
 
+#[derive(Eq, PartialEq, Debug, serde::Deserialize, serde::Serialize, Clone)]
+pub struct ConnectorMandateReferenceId {
+    pub connector_mandate_id: Option<String>,
+    pub payment_method_id: Option<String>,
+}
+
 impl MandateIds {
     pub fn new(mandate_id: String) -> Self {
         Self {
diff --git a/crates/router/src/configs/settings.rs b/crates/router/src/configs/settings.rs
index 5c276050078..7c173965945 100644
--- a/crates/router/src/configs/settings.rs
+++ b/crates/router/src/configs/settings.rs
@@ -107,9 +107,26 @@ pub struct DummyConnector {
 pub struct PaymentMethodTokenFilter {
     #[serde(deserialize_with = "pm_deser")]
     pub payment_method: HashSet<storage_models::enums::PaymentMethod>,
+    pub payment_method_type: Option<PaymentMethodTypeTokenFilter>,
     pub long_lived_token: bool,
 }
 
+#[derive(Debug, Deserialize, Clone, Default)]
+#[serde(
+    deny_unknown_fields,
+    tag = "type",
+    content = "list",
+    rename_all = "snake_case"
+)]
+pub enum PaymentMethodTypeTokenFilter {
+    #[serde(deserialize_with = "pm_type_deser")]
+    EnableOnly(HashSet<storage_models::enums::PaymentMethodType>),
+    #[serde(deserialize_with = "pm_type_deser")]
+    DisableOnly(HashSet<storage_models::enums::PaymentMethodType>),
+    #[default]
+    AllAccepted,
+}
+
 fn pm_deser<'a, D>(
     deserializer: D,
 ) -> Result<HashSet<storage_models::enums::PaymentMethod>, D::Error>
@@ -125,6 +142,21 @@ where
         .map_err(D::Error::custom)
 }
 
+fn pm_type_deser<'a, D>(
+    deserializer: D,
+) -> Result<HashSet<storage_models::enums::PaymentMethodType>, D::Error>
+where
+    D: Deserializer<'a>,
+{
+    let value = <String>::deserialize(deserializer)?;
+    value
+        .trim()
+        .split(',')
+        .map(storage_models::enums::PaymentMethodType::from_str)
+        .collect::<Result<_, _>>()
+        .map_err(D::Error::custom)
+}
+
 #[derive(Debug, Deserialize, Clone, Default)]
 pub struct BankRedirectConfig(
     pub HashMap<api_models::enums::PaymentMethodType, ConnectorBankNames>,
diff --git a/crates/router/src/connector/adyen/transformers.rs b/crates/router/src/connector/adyen/transformers.rs
index 93993c95877..0d07eb72779 100644
--- a/crates/router/src/connector/adyen/transformers.rs
+++ b/crates/router/src/connector/adyen/transformers.rs
@@ -7,7 +7,9 @@ use serde::{Deserialize, Serialize};
 use time::PrimitiveDateTime;
 
 use crate::{
-    connector::utils::{self, CardData, PaymentsAuthorizeRequestData, RouterData},
+    connector::utils::{
+        self, CardData, MandateReferenceData, PaymentsAuthorizeRequestData, RouterData,
+    },
     consts,
     core::errors,
     pii::{self, Email, Secret},
@@ -1104,10 +1106,10 @@ impl<'a> TryFrom<(&types::PaymentsAuthorizeRouterData, MandateReferenceId)>
         let additional_data = get_additional_data(item);
         let return_url = item.request.get_return_url()?;
         let payment_method = match mandate_ref_id {
-            MandateReferenceId::ConnectorMandateId(connector_mandate_id) => {
+            MandateReferenceId::ConnectorMandateId(connector_mandate_ids) => {
                 let adyen_mandate = AdyenMandate {
                     payment_type: PaymentType::Scheme,
-                    stored_payment_method_id: connector_mandate_id,
+                    stored_payment_method_id: connector_mandate_ids.get_connector_mandate_id()?,
                 };
                 Ok::<AdyenPaymentMethod<'_>, Self::Error>(AdyenPaymentMethod::Mandate(Box::new(
                     adyen_mandate,
@@ -1435,7 +1437,10 @@ pub fn get_adyen_response(
     let mandate_reference = response
         .additional_data
         .as_ref()
-        .and_then(|additional_data| additional_data.recurring_detail_reference.to_owned());
+        .map(|data| types::MandateReference {
+            connector_mandate_id: data.recurring_detail_reference.to_owned(),
+            payment_method_id: None,
+        });
     let network_txn_id = response
         .additional_data
         .and_then(|additional_data| additional_data.network_tx_reference);
diff --git a/crates/router/src/connector/globalpay/transformers.rs b/crates/router/src/connector/globalpay/transformers.rs
index 15f3a299d4c..a0454296572 100644
--- a/crates/router/src/connector/globalpay/transformers.rs
+++ b/crates/router/src/connector/globalpay/transformers.rs
@@ -200,6 +200,10 @@ fn get_payment_response(
         pm.card
             .as_ref()
             .and_then(|card| card.brand_reference.to_owned())
+            .map(|id| types::MandateReference {
+                connector_mandate_id: Some(id),
+                payment_method_id: None,
+            })
     });
     match status {
         enums::AttemptStatus::Failure => Err(ErrorResponse {
@@ -374,8 +378,8 @@ fn get_mandate_details(item: &types::PaymentsAuthorizeRouterData) -> Result<Mand
         let connector_mandate_id = item.request.mandate_id.as_ref().and_then(|mandate_ids| {
             match mandate_ids.mandate_reference_id.clone() {
                 Some(api_models::payments::MandateReferenceId::ConnectorMandateId(
-                    connector_mandate_id,
-                )) => Some(connector_mandate_id),
+                    connector_mandate_ids,
+                )) => connector_mandate_ids.connector_mandate_id,
                 _ => None,
             }
         });
diff --git a/crates/router/src/connector/multisafepay/transformers.rs b/crates/router/src/connector/multisafepay/transformers.rs
index 46fee5118b1..7f60c70c2f5 100644
--- a/crates/router/src/connector/multisafepay/transformers.rs
+++ b/crates/router/src/connector/multisafepay/transformers.rs
@@ -354,8 +354,8 @@ impl TryFrom<&types::PaymentsAuthorizeRouterData> for MultisafepayPaymentsReques
                 .clone()
                 .and_then(|mandate_ids| match mandate_ids.mandate_reference_id {
                     Some(api_models::payments::MandateReferenceId::ConnectorMandateId(
-                        connector_mandate_id,
-                    )) => Some(connector_mandate_id),
+                        connector_mandate_ids,
+                    )) => connector_mandate_ids.connector_mandate_id,
                     _ => None,
                 }),
             days_active: Some(30),
@@ -479,7 +479,11 @@ impl<F, T>
                     .response
                     .data
                     .payment_details
-                    .and_then(|payment_details| payment_details.recurring_id),
+                    .and_then(|payment_details| payment_details.recurring_id)
+                    .map(|id| types::MandateReference {
+                        connector_mandate_id: Some(id),
+                        payment_method_id: None,
+                    }),
                 connector_metadata: None,
                 network_txn_id: None,
             }),
diff --git a/crates/router/src/connector/nexinets/transformers.rs b/crates/router/src/connector/nexinets/transformers.rs
index 7c8db924538..05655430a65 100644
--- a/crates/router/src/connector/nexinets/transformers.rs
+++ b/crates/router/src/connector/nexinets/transformers.rs
@@ -344,7 +344,14 @@ impl<F, T>
             }
             _ => Err(errors::ConnectorError::ResponseHandlingFailed)?,
         };
-        let mandate_reference = item.response.payment_instrument.payment_instrument_id;
+        let mandate_reference = item
+            .response
+            .payment_instrument
+            .payment_instrument_id
+            .map(|id| types::MandateReference {
+                connector_mandate_id: Some(id),
+                payment_method_id: None,
+            });
         Ok(Self {
             status: enums::AttemptStatus::foreign_from((
                 transaction.status.clone(),
diff --git a/crates/router/src/connector/nuvei/transformers.rs b/crates/router/src/connector/nuvei/transformers.rs
index 4a981b272d5..1653ecdc849 100644
--- a/crates/router/src/connector/nuvei/transformers.rs
+++ b/crates/router/src/connector/nuvei/transformers.rs
@@ -1143,7 +1143,11 @@ where
                     redirection_data,
                     mandate_reference: response
                         .payment_option
-                        .and_then(|po| po.user_payment_option_id),
+                        .and_then(|po| po.user_payment_option_id)
+                        .map(|id| types::MandateReference {
+                            connector_mandate_id: Some(id),
+                            payment_method_id: None,
+                        }),
                     // we don't need to save session token for capture, void flow so ignoring if it is not present
                     connector_metadata: if let Some(token) = response.session_token {
                         Some(
diff --git a/crates/router/src/connector/payeezy/transformers.rs b/crates/router/src/connector/payeezy/transformers.rs
index b0cd6beaca2..29c15b56722 100644
--- a/crates/router/src/connector/payeezy/transformers.rs
+++ b/crates/router/src/connector/payeezy/transformers.rs
@@ -130,8 +130,8 @@ fn get_transaction_type_and_stored_creds(
     let connector_mandate_id = item.request.mandate_id.as_ref().and_then(|mandate_ids| {
         match mandate_ids.mandate_reference_id.clone() {
             Some(api_models::payments::MandateReferenceId::ConnectorMandateId(
-                connector_mandate_id,
-            )) => Some(connector_mandate_id),
+                connector_mandate_ids,
+            )) => connector_mandate_ids.connector_mandate_id,
             _ => None,
         }
     });
@@ -341,7 +341,11 @@ impl<F, T>
         let mandate_reference = item
             .response
             .stored_credentials
-            .map(|credentials| credentials.cardbrand_original_transaction_id);
+            .map(|credentials| credentials.cardbrand_original_transaction_id)
+            .map(|id| types::MandateReference {
+                connector_mandate_id: Some(id),
+                payment_method_id: None,
+            });
         let status = enums::AttemptStatus::foreign_from((
             item.response.transaction_status,
             item.response.transaction_type,
diff --git a/crates/router/src/connector/stripe.rs b/crates/router/src/connector/stripe.rs
index 5e6797a0963..60630f595dd 100644
--- a/crates/router/src/connector/stripe.rs
+++ b/crates/router/src/connector/stripe.rs
@@ -489,29 +489,33 @@ impl
         types::PaymentsResponseData: Clone,
     {
         let id = data.request.connector_transaction_id.clone();
-        let response: transformers::PaymentIntentSyncResponse = match id
-            .get_connector_transaction_id()
-        {
-            Ok(x) if x.starts_with("set") => res
-                .response
-                .parse_struct::<transformers::SetupIntentSyncResponse>("SetupIntentSyncResponse")
-                .change_context(errors::ConnectorError::ResponseDeserializationFailed)
-                .map(Into::into),
-            Ok(_) => res
-                .response
-                .parse_struct("PaymentIntentSyncResponse")
-                .change_context(errors::ConnectorError::ResponseDeserializationFailed),
+        match id.get_connector_transaction_id() {
+            Ok(x) if x.starts_with("set") => {
+                let response: stripe::SetupIntentResponse = res
+                    .response
+                    .parse_struct("SetupIntentSyncResponse")
+                    .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+                types::RouterData::try_from(types::ResponseRouterData {
+                    response,
+                    data: data.clone(),
+                    http_code: res.status_code,
+                })
+            }
+            Ok(_) => {
+                let response: stripe::PaymentIntentSyncResponse = res
+                    .response
+                    .parse_struct("PaymentIntentSyncResponse")
+                    .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+                types::RouterData::try_from(types::ResponseRouterData {
+                    response,
+                    data: data.clone(),
+                    http_code: res.status_code,
+                })
+            }
             Err(err) => {
                 Err(err).change_context(errors::ConnectorError::MissingConnectorTransactionID)
             }
-        }?;
-
-        types::RouterData::try_from(types::ResponseRouterData {
-            response,
-            data: data.clone(),
-            http_code: res.status_code,
-        })
-        .change_context(errors::ConnectorError::ResponseHandlingFailed)
+        }
     }
 
     fn get_error_response(
@@ -798,7 +802,8 @@ impl
         &self,
         req: &types::RouterData<api::Verify, types::VerifyRequestData, types::PaymentsResponseData>,
     ) -> CustomResult<Option<String>, errors::ConnectorError> {
-        let stripe_req = utils::Encode::<stripe::SetupIntentRequest>::convert_and_url_encode(req)
+        let req = stripe::SetupIntentRequest::try_from(req)?;
+        let stripe_req = utils::Encode::<stripe::SetupIntentRequest>::url_encode(&req)
             .change_context(errors::ConnectorError::RequestEncodingFailed)?;
         Ok(Some(stripe_req))
     }
diff --git a/crates/router/src/connector/stripe/transformers.rs b/crates/router/src/connector/stripe/transformers.rs
index fe46181684f..62e88e918e6 100644
--- a/crates/router/src/connector/stripe/transformers.rs
+++ b/crates/router/src/connector/stripe/transformers.rs
@@ -1,6 +1,6 @@
 use api_models::{self, enums as api_enums, payments};
 use base64::Engine;
-use common_utils::{errors::CustomResult, pii, pii::Email};
+use common_utils::{errors::CustomResult, ext_traits::ByteSliceExt, pii, pii::Email};
 use error_stack::{IntoReport, ResultExt};
 use masking::{ExposeInterface, ExposeOptionInterface, Secret};
 use serde::{Deserialize, Serialize};
@@ -12,7 +12,7 @@ use crate::{
     collect_missing_value_keys, consts,
     core::errors,
     services,
-    types::{self, api, storage::enums},
+    types::{self, api, storage::enums, transformers::ForeignFrom},
     utils::OptionExt,
 };
 
@@ -108,6 +108,7 @@ pub struct PaymentIntentRequest {
     pub capture_method: StripeCaptureMethod,
     pub payment_method_options: Option<StripePaymentMethodOptions>, // For mandate txns using network_txns_id, needs to be validated
     pub setup_future_usage: Option<enums::FutureUsage>,
+    pub off_session: Option<bool>,
 }
 
 #[derive(Debug, Eq, PartialEq, Serialize)]
@@ -295,6 +296,7 @@ pub enum StripePaymentMethodData {
 #[serde(untagged)]
 pub enum StripeWallet {
     ApplepayToken(StripeApplePay),
+    GooglepayToken(GooglePayToken),
     ApplepayPayment(ApplepayPayment),
 }
 
@@ -306,6 +308,14 @@ pub struct StripeApplePay {
     pub pk_token_transaction_id: String,
 }
 
+#[derive(Debug, Eq, PartialEq, Serialize)]
+pub struct GooglePayToken {
+    #[serde(rename = "payment_method_data[type]")]
+    pub payment_type: StripePaymentMethodType,
+    #[serde(rename = "payment_method_data[card][token]")]
+    pub token: String,
+}
+
 #[derive(Debug, Eq, PartialEq, Serialize)]
 pub struct ApplepayPayment {
     #[serde(rename = "payment_method_data[card][token]")]
@@ -314,6 +324,14 @@ pub struct ApplepayPayment {
     pub payment_method_types: StripePaymentMethodType,
 }
 
+#[derive(Debug, Eq, PartialEq, Serialize)]
+pub struct GooglepayPayment {
+    #[serde(rename = "payment_method_data[card][token]")]
+    pub token: String,
+    #[serde(rename = "payment_method_data[type]")]
+    pub payment_method_types: StripePaymentMethodType,
+}
+
 #[derive(Debug, Eq, PartialEq, Serialize, Clone, Copy)]
 #[serde(rename_all = "snake_case")]
 pub enum StripePaymentMethodType {
@@ -789,10 +807,14 @@ fn create_stripe_payment_method(
                 StripePaymentMethodType::ApplePay,
                 StripeBillingAddress::default(),
             )),
-            _ => Err(errors::ConnectorError::NotImplemented(
-                "This wallet is not implemented for stripe".to_string(),
-            )
-            .into()),
+            payments::WalletData::GooglePay(gpay_data) => Ok((
+                StripePaymentMethodData::try_from(gpay_data)?,
+                StripePaymentMethodType::Card,
+                StripeBillingAddress::default(),
+            )),
+            _ => Err(
+                errors::ConnectorError::NotImplemented("This wallet for stripe".to_string()).into(),
+            ),
         },
         payments::PaymentMethodData::BankDebit(bank_debit_data) => {
             let (pm_type, bank_debit_data, billing_address) = get_bank_debit_data(bank_debit_data);
@@ -805,12 +827,28 @@ fn create_stripe_payment_method(
             Ok((pm_data, pm_type, billing_address))
         }
         _ => Err(errors::ConnectorError::NotImplemented(
-            "stripe does not support this payment method".to_string(),
+            "this payment method for stripe".to_string(),
         )
         .into()),
     }
 }
 
+impl TryFrom<&payments::GooglePayWalletData> for StripePaymentMethodData {
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(gpay_data: &payments::GooglePayWalletData) -> Result<Self, Self::Error> {
+        Ok(Self::Wallet(StripeWallet::GooglepayToken(GooglePayToken {
+            token: gpay_data
+                .tokenization_data
+                .token
+                .as_bytes()
+                .parse_struct::<StripeGpayToken>("StripeGpayToken")
+                .change_context(errors::ConnectorError::RequestEncodingFailed)?
+                .id,
+            payment_type: StripePaymentMethodType::Card,
+        })))
+    }
+}
+
 impl TryFrom<&types::PaymentsAuthorizeRouterData> for PaymentIntentRequest {
     type Error = error_stack::Report<errors::ConnectorError>;
     fn try_from(item: &types::PaymentsAuthorizeRouterData) -> Result<Self, Self::Error> {
@@ -849,16 +887,21 @@ impl TryFrom<&types::PaymentsAuthorizeRouterData> for PaymentIntentRequest {
         };
         let mut payment_method_options = None;
 
-        let (mut payment_data, payment_method, billing_address) = {
+        let (mut payment_data, payment_method, mandate, billing_address) = {
             match item
                 .request
                 .mandate_id
                 .clone()
                 .and_then(|mandate_ids| mandate_ids.mandate_reference_id)
             {
-                Some(api_models::payments::MandateReferenceId::ConnectorMandateId(mandate_id)) => {
-                    (None, Some(mandate_id), StripeBillingAddress::default())
-                }
+                Some(api_models::payments::MandateReferenceId::ConnectorMandateId(
+                    connector_mandate_ids,
+                )) => (
+                    None,
+                    connector_mandate_ids.payment_method_id,
+                    connector_mandate_ids.connector_mandate_id,
+                    StripeBillingAddress::default(),
+                ),
                 Some(api_models::payments::MandateReferenceId::NetworkMandateId(
                     network_transaction_id,
                 )) => {
@@ -869,7 +912,7 @@ impl TryFrom<&types::PaymentsAuthorizeRouterData> for PaymentIntentRequest {
                             network_transaction_id,
                         }),
                     });
-                    (None, None, StripeBillingAddress::default())
+                    (None, None, None, StripeBillingAddress::default())
                 }
                 _ => {
                     let (payment_method_data, payment_method_type, billing_address) =
@@ -885,7 +928,7 @@ impl TryFrom<&types::PaymentsAuthorizeRouterData> for PaymentIntentRequest {
                         &payment_method_type,
                     )?;
 
-                    (Some(payment_method_data), None, billing_address)
+                    (Some(payment_method_data), None, None, billing_address)
                 }
             }
         };
@@ -934,17 +977,17 @@ impl TryFrom<&types::PaymentsAuthorizeRouterData> for PaymentIntentRequest {
                 .clone()
                 .unwrap_or_else(|| "https://juspay.in/".to_string()),
             confirm: true, // Stripe requires confirm to be true if return URL is present
-
             description: item.description.clone(),
             shipping: shipping_address,
             billing: billing_address,
             capture_method: StripeCaptureMethod::from(item.request.capture_method),
             payment_data,
-            mandate: None,
+            mandate,
             payment_method_options,
             payment_method,
             customer: item.connector_customer.to_owned(),
             setup_mandate_details,
+            off_session: item.request.off_session,
             setup_future_usage: item.request.setup_future_usage,
         })
     }
@@ -1167,6 +1210,32 @@ pub struct SetupIntentResponse {
     pub latest_attempt: Option<LatestAttempt>,
 }
 
+impl ForeignFrom<(Option<StripePaymentMethodOptions>, String)> for types::MandateReference {
+    fn foreign_from(
+        (payment_method_options, payment_method_id): (Option<StripePaymentMethodOptions>, String),
+    ) -> Self {
+        Self {
+            connector_mandate_id: payment_method_options.and_then(|options| match options {
+                StripePaymentMethodOptions::Card {
+                    mandate_options, ..
+                } => mandate_options.map(|mandate_options| mandate_options.reference),
+                StripePaymentMethodOptions::Klarna {}
+                | StripePaymentMethodOptions::Affirm {}
+                | StripePaymentMethodOptions::AfterpayClearpay {}
+                | StripePaymentMethodOptions::Eps {}
+                | StripePaymentMethodOptions::Giropay {}
+                | StripePaymentMethodOptions::Ideal {}
+                | StripePaymentMethodOptions::Sofort {}
+                | StripePaymentMethodOptions::Ach {}
+                | StripePaymentMethodOptions::Bacs {}
+                | StripePaymentMethodOptions::Becs {}
+                | StripePaymentMethodOptions::Sepa {} => None,
+            }),
+            payment_method_id: Some(payment_method_id),
+        }
+    }
+}
+
 impl<F, T>
     TryFrom<types::ResponseRouterData<F, PaymentIntentResponse, T, types::PaymentsResponseData>>
     for types::RouterData<F, T, types::PaymentsResponseData>
@@ -1179,19 +1248,13 @@ impl<F, T>
             services::RedirectForm::from((next_action_response.get_url(), services::Method::Get))
         });
 
+        let mandate_reference = item.response.payment_method.map(|pm| {
+            types::MandateReference::foreign_from((item.response.payment_method_options, pm))
+        });
+
         //Note: we might have to call retrieve_setup_intent to get the network_transaction_id in case its not sent in PaymentIntentResponse
         // Or we identify the mandate txns before hand and always call SetupIntent in case of mandate payment call
-        let network_txn_id = item.response.latest_attempt.and_then(|latest_attempt| {
-            latest_attempt
-                .payment_method_options
-                .and_then(|payment_method_options| match payment_method_options {
-                    StripePaymentMethodOptions::Card {
-                        network_transaction_id,
-                        ..
-                    } => network_transaction_id,
-                    _ => None,
-                })
-        });
+        let network_txn_id = Option::foreign_from(item.response.latest_attempt);
 
         Ok(Self {
             status: enums::AttemptStatus::from(item.response.status),
@@ -1202,7 +1265,7 @@ impl<F, T>
             response: Ok(types::PaymentsResponseData::TransactionResponse {
                 resource_id: types::ResponseId::ConnectorTransactionId(item.response.id),
                 redirection_data,
-                mandate_reference: item.response.payment_method,
+                mandate_reference,
                 connector_metadata: None,
                 network_txn_id,
             }),
@@ -1236,27 +1299,12 @@ impl<F, T>
                 ))
             });
 
-        let mandate_reference =
-            item.response
-                .payment_method_options
-                .to_owned()
-                .and_then(|payment_method_options| match payment_method_options {
-                    StripePaymentMethodOptions::Card {
-                        mandate_options, ..
-                    } => mandate_options.map(|mandate_options| mandate_options.reference),
-                    StripePaymentMethodOptions::Klarna {}
-                    | StripePaymentMethodOptions::Affirm {}
-                    | StripePaymentMethodOptions::AfterpayClearpay {}
-                    | StripePaymentMethodOptions::Eps {}
-                    | StripePaymentMethodOptions::Giropay {}
-                    | StripePaymentMethodOptions::Ideal {}
-                    | StripePaymentMethodOptions::Sofort {}
-                    | StripePaymentMethodOptions::Ach {}
-                    | StripePaymentMethodOptions::Bacs {}
-                    | StripePaymentMethodOptions::Becs {}
-                    | StripePaymentMethodOptions::Sepa {} => None,
-                });
-
+        let mandate_reference = item.response.payment_method.clone().map(|pm| {
+            types::MandateReference::foreign_from((
+                item.response.payment_method_options.clone(),
+                pm,
+            ))
+        });
         let error_res =
             item.response
                 .last_payment_error
@@ -1300,16 +1348,8 @@ impl<F, T>
             services::RedirectForm::from((next_action_response.get_url(), services::Method::Get))
         });
 
-        let network_txn_id = item.response.latest_attempt.and_then(|latest_attempt| {
-            latest_attempt
-                .payment_method_options
-                .and_then(|payment_method_options| match payment_method_options {
-                    StripePaymentMethodOptions::Card {
-                        network_transaction_id,
-                        ..
-                    } => network_transaction_id,
-                    _ => None,
-                })
+        let mandate_reference = item.response.payment_method.map(|pm| {
+            types::MandateReference::foreign_from((item.response.payment_method_options, pm))
         });
 
         Ok(Self {
@@ -1317,15 +1357,32 @@ impl<F, T>
             response: Ok(types::PaymentsResponseData::TransactionResponse {
                 resource_id: types::ResponseId::ConnectorTransactionId(item.response.id),
                 redirection_data,
-                mandate_reference: item.response.payment_method,
+                mandate_reference,
                 connector_metadata: None,
-                network_txn_id,
+                network_txn_id: Option::foreign_from(item.response.latest_attempt),
             }),
             ..item.data
         })
     }
 }
 
+impl ForeignFrom<Option<LatestAttempt>> for Option<String> {
+    fn foreign_from(latest_attempt: Option<LatestAttempt>) -> Self {
+        match latest_attempt {
+            Some(LatestAttempt::PaymentIntentAttempt(attempt)) => attempt
+                .payment_method_options
+                .and_then(|payment_method_options| match payment_method_options {
+                    StripePaymentMethodOptions::Card {
+                        network_transaction_id,
+                        ..
+                    } => network_transaction_id,
+                    _ => None,
+                }),
+            _ => None,
+        }
+    }
+}
+
 #[derive(Clone, Debug, Eq, PartialEq, Deserialize)]
 #[serde(rename_all = "snake_case", remote = "Self")]
 pub enum StripeNextActionResponse {
@@ -1581,8 +1638,14 @@ pub struct MitExemption {
     pub network_transaction_id: String,
 }
 
+#[derive(Clone, Debug, Eq, PartialEq, Deserialize)]
+#[serde(untagged)]
+pub enum LatestAttempt {
+    PaymentIntentAttempt(LatestPaymentAttempt),
+    SetupAttempt(String),
+}
 #[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize)]
-pub struct LatestAttempt {
+pub struct LatestPaymentAttempt {
     pub payment_method_options: Option<StripePaymentMethodOptions>,
 }
 // #[derive(Deserialize, Debug, Clone, Eq, PartialEq)]
@@ -1854,6 +1917,7 @@ impl
                     });
                     Ok(Self::Wallet(wallet_info))
                 }
+                payments::WalletData::GooglePay(gpay_data) => Self::try_from(&gpay_data),
                 _ => Err(errors::ConnectorError::InvalidWallet.into()),
             },
             api::PaymentMethodData::BankDebit(bank_debit_data) => {
@@ -1876,6 +1940,10 @@ impl
     }
 }
 
+#[derive(Debug, Deserialize)]
+pub struct StripeGpayToken {
+    pub id: String,
+}
 pub fn construct_file_upload_request(
     file_upload_router_data: types::UploadFileRouterData,
 ) -> CustomResult<reqwest::multipart::Form, errors::ConnectorError> {
diff --git a/crates/router/src/connector/utils.rs b/crates/router/src/connector/utils.rs
index ec1d76ffd67..ca3680c08cb 100644
--- a/crates/router/src/connector/utils.rs
+++ b/crates/router/src/connector/utils.rs
@@ -204,8 +204,8 @@ impl PaymentsAuthorizeRequestData for types::PaymentsAuthorizeData {
             .as_ref()
             .and_then(|mandate_ids| match &mandate_ids.mandate_reference_id {
                 Some(api_models::payments::MandateReferenceId::ConnectorMandateId(
-                    connector_mandate_id,
-                )) => Some(connector_mandate_id.to_string()),
+                    connector_mandate_ids,
+                )) => connector_mandate_ids.connector_mandate_id.clone(),
                 _ => None,
             })
     }
@@ -556,6 +556,18 @@ impl MandateData for payments::MandateAmountData {
     }
 }
 
+pub trait MandateReferenceData {
+    fn get_connector_mandate_id(&self) -> Result<String, Error>;
+}
+
+impl MandateReferenceData for api_models::payments::ConnectorMandateReferenceId {
+    fn get_connector_mandate_id(&self) -> Result<String, Error> {
+        self.connector_mandate_id
+            .clone()
+            .ok_or_else(missing_field_err("mandate_id"))
+    }
+}
+
 pub fn get_header_key_value<'a>(
     key: &str,
     headers: &'a actix_web::http::header::HeaderMap,
diff --git a/crates/router/src/core/errors/utils.rs b/crates/router/src/core/errors/utils.rs
index 0df489edf26..87d9e38a2c0 100644
--- a/crates/router/src/core/errors/utils.rs
+++ b/crates/router/src/core/errors/utils.rs
@@ -119,7 +119,8 @@ impl ConnectorErrorExt for error_stack::Report<errors::ConnectorError> {
     }
 
     fn to_verify_failed_response(self) -> error_stack::Report<errors::ApiErrorResponse> {
-        let data = match self.current_context() {
+        let error = self.current_context();
+        let data = match error {
             errors::ConnectorError::ProcessingStepFailed(Some(bytes)) => {
                 let response_str = std::str::from_utf8(bytes);
                 match response_str {
@@ -132,7 +133,10 @@ impl ConnectorErrorExt for error_stack::Report<errors::ConnectorError> {
                     }
                 }
             }
-            _ => None,
+            _ => {
+                logger::error!(%error,"Verify flow failed");
+                None
+            }
         };
         self.change_context(errors::ApiErrorResponse::PaymentAuthorizationFailed { data })
     }
diff --git a/crates/router/src/core/mandate.rs b/crates/router/src/core/mandate.rs
index 25f7f3ab69a..02269d6707c 100644
--- a/crates/router/src/core/mandate.rs
+++ b/crates/router/src/core/mandate.rs
@@ -1,3 +1,4 @@
+use common_utils::ext_traits::Encode;
 use error_stack::{report, ResultExt};
 use router_env::{instrument, logger, tracing};
 use storage_models::enums as storage_enums;
@@ -158,38 +159,52 @@ where
                     _ => (None, None),
                 };
 
+                let mandate_ids = mandate_reference
+                    .map(|md| {
+                        Encode::<types::MandateReference>::encode_to_value(&md)
+                            .change_context(errors::ApiErrorResponse::MandateNotFound)
+                            .map(masking::Secret::new)
+                    })
+                    .transpose()?;
+
                 if let Some(new_mandate_data) = helpers::generate_mandate(
                     resp.merchant_id.clone(),
                     resp.connector.clone(),
                     resp.request.get_setup_mandate_details().map(Clone::clone),
                     maybe_customer,
                     pm_id.get_required_value("payment_method_id")?,
-                    mandate_reference,
+                    mandate_ids,
                     network_txn_id,
                 ) {
                     let connector = new_mandate_data.connector.clone();
                     logger::debug!("{:?}", new_mandate_data);
                     resp.request
-                        .set_mandate_id(api_models::payments::MandateIds {
+                        .set_mandate_id(Some(api_models::payments::MandateIds {
                             mandate_id: new_mandate_data.mandate_id.clone(),
                             mandate_reference_id: new_mandate_data
-                                .connector_mandate_id
+                                .connector_mandate_ids
                                 .clone()
-                                .map_or(
-                                    new_mandate_data.network_transaction_id.clone().map(|id| {
-                                        api_models::payments::MandateReferenceId::NetworkMandateId(
-                                            id,
-                                        )
-                                    }),
-                                    |connector_id| {
-                                        Some(
-                                    api_models::payments::MandateReferenceId::ConnectorMandateId(
-                                        connector_id,
-                                    ),
-                                )
-                                    },
-                                ),
-                        });
+                            .map(|ids| {
+                                Some(ids)
+                                    .parse_value::<api_models::payments::ConnectorMandateReferenceId>(
+                                        "ConnectorMandateId",
+                                    )
+                                    .change_context(errors::ApiErrorResponse::MandateNotFound)
+                            })
+                            .transpose()?
+                            .map_or(
+                                new_mandate_data.network_transaction_id.clone().map(|id| {
+                                    api_models::payments::MandateReferenceId::NetworkMandateId(
+                                        id,
+                                    )
+                                }),
+                                |connector_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,
+                                    }
+                                )))
+                        }));
                     state
                         .store
                         .insert_mandate(new_mandate_data)
@@ -212,7 +227,7 @@ pub trait MandateBehaviour {
     fn get_amount(&self) -> i64;
     fn get_setup_future_usage(&self) -> Option<storage_models::enums::FutureUsage>;
     fn get_mandate_id(&self) -> Option<&api_models::payments::MandateIds>;
-    fn set_mandate_id(&mut self, new_mandate_id: api_models::payments::MandateIds);
+    fn set_mandate_id(&mut self, new_mandate_id: Option<api_models::payments::MandateIds>);
     fn get_payment_method_data(&self) -> api_models::payments::PaymentMethodData;
     fn get_setup_mandate_details(&self) -> Option<&api_models::payments::MandateData>;
 }
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index a292e4503fe..de4ed1649a6 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -1,7 +1,9 @@
 pub mod access_token;
+pub mod customers;
 pub mod flows;
 pub mod helpers;
 pub mod operations;
+pub mod tokenization;
 pub mod transformers;
 
 use std::{fmt::Debug, marker::PhantomData, time::Instant};
@@ -24,6 +26,7 @@ use self::{
     operations::{payment_complete_authorize, BoxedOperation, Operation},
 };
 use crate::{
+    configs::settings::PaymentMethodTypeTokenFilter,
     core::{
         errors::{self, CustomResult, RouterResponse, RouterResult},
         payment_methods::vault,
@@ -647,14 +650,38 @@ fn is_payment_method_tokenization_enabled_for_connector(
     state: &AppState,
     connector_name: &str,
     payment_method: &storage::enums::PaymentMethod,
+    payment_method_type: &Option<storage::enums::PaymentMethodType>,
 ) -> RouterResult<bool> {
     let connector_tokenization_filter = state.conf.tokenization.0.get(connector_name);
 
     Ok(connector_tokenization_filter
-        .map(|connector_filter| connector_filter.payment_method.contains(payment_method))
+        .map(|connector_filter| {
+            connector_filter
+                .payment_method
+                .clone()
+                .contains(payment_method)
+                && is_payment_method_type_allowed_for_connector(
+                    payment_method_type,
+                    connector_filter.payment_method_type.clone(),
+                )
+        })
         .unwrap_or(false))
 }
 
+fn is_payment_method_type_allowed_for_connector(
+    current_pm_type: &Option<storage::enums::PaymentMethodType>,
+    pm_type_filter: Option<PaymentMethodTypeTokenFilter>,
+) -> bool {
+    match current_pm_type.clone().zip(pm_type_filter) {
+        Some((pm_type, type_filter)) => match type_filter {
+            PaymentMethodTypeTokenFilter::AllAccepted => true,
+            PaymentMethodTypeTokenFilter::EnableOnly(enabled) => enabled.contains(&pm_type),
+            PaymentMethodTypeTokenFilter::DisableOnly(disabled) => !disabled.contains(&pm_type),
+        },
+        None => true, // Allow all types if payment_method_type is not present
+    }
+}
+
 async fn decide_payment_method_tokenize_action(
     state: &AppState,
     connector_name: &str,
@@ -729,12 +756,14 @@ where
                 .payment_attempt
                 .payment_method
                 .get_required_value("payment_method")?;
+            let payment_method_type = &payment_data.payment_attempt.payment_method_type;
 
             let is_connector_tokenization_enabled =
                 is_payment_method_tokenization_enabled_for_connector(
                     state,
                     &connector,
                     payment_method,
+                    payment_method_type,
                 )?;
 
             let payment_method_action = decide_payment_method_tokenize_action(
diff --git a/crates/router/src/core/payments/customers.rs b/crates/router/src/core/payments/customers.rs
new file mode 100644
index 00000000000..890e7bfa4d2
--- /dev/null
+++ b/crates/router/src/core/payments/customers.rs
@@ -0,0 +1,142 @@
+use common_utils::ext_traits::ValueExt;
+use error_stack::{self, ResultExt};
+
+use crate::{
+    core::{
+        errors::{self, ConnectorErrorExt, RouterResult},
+        payments,
+    },
+    logger,
+    routes::AppState,
+    services,
+    types::{self, api, storage},
+};
+
+pub async fn create_connector_customer<F: Clone, T: Clone>(
+    state: &AppState,
+    connector: &api::ConnectorData,
+    customer: &Option<storage::Customer>,
+    router_data: &types::RouterData<F, T, types::PaymentsResponseData>,
+    customer_request_data: types::ConnectorCustomerData,
+) -> RouterResult<(Option<String>, Option<storage::CustomerUpdate>)> {
+    let (is_eligible, connector_customer_id, connector_customer_map) =
+        should_call_connector_create_customer(state, connector, customer)?;
+
+    if is_eligible {
+        let connector_integration: services::BoxedConnectorIntegration<
+            '_,
+            api::CreateConnectorCustomer,
+            types::ConnectorCustomerData,
+            types::PaymentsResponseData,
+        > = connector.connector.get_connector_integration();
+
+        let customer_response_data: Result<types::PaymentsResponseData, types::ErrorResponse> =
+            Err(types::ErrorResponse::default());
+
+        let customer_router_data = payments::helpers::router_data_type_conversion::<
+            _,
+            api::CreateConnectorCustomer,
+            _,
+            _,
+            _,
+            _,
+        >(
+            router_data.clone(),
+            customer_request_data,
+            customer_response_data,
+        );
+
+        let resp = services::execute_connector_processing_step(
+            state,
+            connector_integration,
+            &customer_router_data,
+            payments::CallConnectorAction::Trigger,
+        )
+        .await
+        .map_err(|error| error.to_payment_failed_response())?;
+
+        let connector_customer_id = match resp.response {
+            Ok(response) => match response {
+                types::PaymentsResponseData::ConnectorCustomerResponse {
+                    connector_customer_id,
+                } => Some(connector_customer_id),
+                _ => None,
+            },
+            Err(err) => {
+                logger::debug!(payment_method_tokenization_error=?err);
+                None
+            }
+        };
+        let update_customer = update_connector_customer_in_customers(
+            connector,
+            connector_customer_map,
+            &connector_customer_id,
+        )
+        .await?;
+        Ok((connector_customer_id, update_customer))
+    } else {
+        Ok((connector_customer_id, None))
+    }
+}
+
+type CreateCustomerCheck = (
+    bool,
+    Option<String>,
+    Option<serde_json::Map<String, serde_json::Value>>,
+);
+pub fn should_call_connector_create_customer(
+    state: &AppState,
+    connector: &api::ConnectorData,
+    customer: &Option<storage::Customer>,
+) -> RouterResult<CreateCustomerCheck> {
+    let connector_name = connector.connector_name.to_string();
+    //Check if create customer is required for the connector
+    let connector_customer_filter = state
+        .conf
+        .connector_customer
+        .connector_list
+        .contains(&connector.connector_name);
+    if connector_customer_filter {
+        match customer {
+            Some(customer) => match &customer.connector_customer {
+                Some(connector_customer) => {
+                    let connector_customer_map: serde_json::Map<String, serde_json::Value> =
+                        connector_customer
+                            .clone()
+                            .parse_value("Map<String, Value>")
+                            .change_context(errors::ApiErrorResponse::InternalServerError)
+                            .attach_printable("Failed to deserialize Value to CustomerConnector")?;
+                    let value = connector_customer_map.get(&connector_name); //Check if customer already created for this customer and for this connector
+                    Ok((
+                        value.is_none(),
+                        value.and_then(|val| val.as_str().map(|cust| cust.to_string())),
+                        Some(connector_customer_map),
+                    ))
+                }
+                None => Ok((true, None, None)),
+            },
+            None => Ok((false, None, None)),
+        }
+    } else {
+        Ok((false, None, None))
+    }
+}
+pub async fn update_connector_customer_in_customers(
+    connector: &api::ConnectorData,
+    connector_customer_map: Option<serde_json::Map<String, serde_json::Value>>,
+    connector_cust_id: &Option<String>,
+) -> RouterResult<Option<storage::CustomerUpdate>> {
+    let mut connector_customer = match connector_customer_map {
+        Some(cc) => cc,
+        None => serde_json::Map::new(),
+    };
+    connector_cust_id.clone().map(|cc| {
+        connector_customer.insert(
+            connector.connector_name.to_string(),
+            serde_json::Value::String(cc),
+        )
+    });
+    Ok(Some(storage::CustomerUpdate::ConnectorCustomer {
+        connector_customer: Some(serde_json::Value::Object(connector_customer)),
+    }))
+}
diff --git a/crates/router/src/core/payments/flows/authorize_flow.rs b/crates/router/src/core/payments/flows/authorize_flow.rs
index 8e5093205f8..5ff299761fa 100644
--- a/crates/router/src/core/payments/flows/authorize_flow.rs
+++ b/crates/router/src/core/payments/flows/authorize_flow.rs
@@ -1,24 +1,17 @@
 use async_trait::async_trait;
-use common_utils::{ext_traits::ValueExt, pii};
-use error_stack::{report, ResultExt};
-use masking::ExposeInterface;
+use error_stack;
 
 use super::{ConstructFlowSpecificData, Feature};
 use crate::{
     core::{
         errors::{self, ConnectorErrorExt, RouterResult},
-        mandate, payment_methods,
-        payments::{self, access_token, helpers, transformers, PaymentData},
+        mandate,
+        payments::{self, access_token, customers, tokenization, transformers, PaymentData},
     },
     logger,
     routes::{metrics, AppState},
     services,
-    types::{
-        self,
-        api::{self, PaymentMethodCreateExt},
-        storage,
-    },
-    utils::OptionExt,
+    types::{self, api, storage},
 };
 
 #[async_trait]
@@ -93,7 +86,14 @@ impl Feature<api::Authorize, types::PaymentsAuthorizeData> for types::PaymentsAu
         connector: &api::ConnectorData,
         tokenization_action: &payments::TokenizationAction,
     ) -> RouterResult<Option<String>> {
-        add_payment_method_token(state, connector, tokenization_action, self).await
+        tokenization::add_payment_method_token(
+            state,
+            connector,
+            tokenization_action,
+            self,
+            types::PaymentMethodTokenizationData::try_from(self.request.to_owned())?,
+        )
+        .await
     }
 
     async fn create_connector_customer<'a>(
@@ -102,7 +102,14 @@ impl Feature<api::Authorize, types::PaymentsAuthorizeData> for types::PaymentsAu
         connector: &api::ConnectorData,
         customer: &Option<storage::Customer>,
     ) -> RouterResult<(Option<String>, Option<storage::CustomerUpdate>)> {
-        create_connector_customer(state, connector, customer, self).await
+        customers::create_connector_customer(
+            state,
+            connector,
+            customer,
+            self,
+            types::ConnectorCustomerData::try_from(self.request.to_owned())?,
+        )
+        .await
     }
 }
 
@@ -141,7 +148,7 @@ impl types::PaymentsAuthorizeRouterData {
                     .await
                     .map_err(|error| error.to_payment_failed_response())?;
 
-                    let pm_id = save_payment_method(
+                    let pm_id = tokenization::save_payment_method(
                         state,
                         connector,
                         resp.to_owned(),
@@ -178,178 +185,6 @@ impl types::PaymentsAuthorizeRouterData {
     }
 }
 
-pub async fn save_payment_method<F: Clone, FData>(
-    state: &AppState,
-    connector: &api::ConnectorData,
-    resp: types::RouterData<F, FData, types::PaymentsResponseData>,
-    maybe_customer: &Option<storage::Customer>,
-    merchant_account: &storage::MerchantAccount,
-) -> RouterResult<Option<String>>
-where
-    FData: mandate::MandateBehaviour,
-{
-    let db = &*state.store;
-    let token_store = state
-        .conf
-        .tokenization
-        .0
-        .get(&connector.connector_name.to_string())
-        .map(|token_filter| token_filter.long_lived_token)
-        .unwrap_or(false);
-
-    let connector_token = if token_store {
-        let token = resp
-            .payment_method_token
-            .to_owned()
-            .get_required_value("payment_token")?;
-        Some((connector, token))
-    } else {
-        None
-    };
-
-    let pm_id = if resp.request.get_setup_future_usage().is_some() {
-        let customer = maybe_customer.to_owned().get_required_value("customer")?;
-        let payment_method_create_request = helpers::get_payment_method_create_request(
-            Some(&resp.request.get_payment_method_data()),
-            Some(resp.payment_method),
-            &customer,
-        )
-        .await?;
-        let merchant_id = &merchant_account.merchant_id;
-
-        let locker_response = save_in_locker(
-            state,
-            merchant_account,
-            payment_method_create_request.to_owned(),
-        )
-        .await?;
-        let is_duplicate = locker_response.1;
-
-        if is_duplicate {
-            let existing_pm = db
-                .find_payment_method(&locker_response.0.payment_method_id)
-                .await;
-            match existing_pm {
-                Ok(pm) => {
-                    let pm_metadata =
-                        create_payment_method_metadata(pm.metadata.as_ref(), connector_token)?;
-                    if let Some(metadata) = pm_metadata {
-                        payment_methods::cards::update_payment_method(db, pm, metadata)
-                            .await
-                            .change_context(errors::ApiErrorResponse::InternalServerError)
-                            .attach_printable("Failed to add payment method in db")?;
-                    };
-                }
-                Err(error) => {
-                    match error.current_context() {
-                        errors::StorageError::DatabaseError(err) => match err.current_context() {
-                            storage_models::errors::DatabaseError::NotFound => {
-                                let pm_metadata =
-                                    create_payment_method_metadata(None, connector_token)?;
-                                payment_methods::cards::create_payment_method(
-                                    db,
-                                    &payment_method_create_request,
-                                    &customer.customer_id,
-                                    &locker_response.0.payment_method_id,
-                                    merchant_id,
-                                    pm_metadata,
-                                )
-                                .await
-                                .change_context(errors::ApiErrorResponse::InternalServerError)
-                                .attach_printable("Failed to add payment method in db")
-                            }
-                            _ => Err(report!(errors::ApiErrorResponse::InternalServerError)),
-                        },
-                        _ => Err(report!(errors::ApiErrorResponse::InternalServerError)),
-                    }?;
-                }
-            };
-        } else {
-            let pm_metadata = create_payment_method_metadata(None, connector_token)?;
-            payment_methods::cards::create_payment_method(
-                db,
-                &payment_method_create_request,
-                &customer.customer_id,
-                &locker_response.0.payment_method_id,
-                merchant_id,
-                pm_metadata,
-            )
-            .await
-            .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("Failed to add payment method in db")?;
-        };
-        Some(locker_response.0.payment_method_id)
-    } else {
-        None
-    };
-    Ok(pm_id)
-}
-
-pub async fn save_in_locker(
-    state: &AppState,
-    merchant_account: &storage::MerchantAccount,
-    payment_method_request: api::PaymentMethodCreate,
-) -> RouterResult<(api_models::payment_methods::PaymentMethodResponse, bool)> {
-    payment_method_request.validate()?;
-    let merchant_id = &merchant_account.merchant_id;
-    let customer_id = payment_method_request
-        .customer_id
-        .clone()
-        .get_required_value("customer_id")?;
-    match payment_method_request.card.clone() {
-        Some(card) => payment_methods::cards::add_card_to_locker(
-            state,
-            payment_method_request,
-            card,
-            customer_id,
-            merchant_account,
-        )
-        .await
-        .change_context(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable("Add Card Failed"),
-        None => {
-            let pm_id = common_utils::generate_id(crate::consts::ID_LENGTH, "pm");
-            let payment_method_response = api::PaymentMethodResponse {
-                merchant_id: merchant_id.to_string(),
-                customer_id: Some(customer_id),
-                payment_method_id: pm_id,
-                payment_method: payment_method_request.payment_method,
-                payment_method_type: payment_method_request.payment_method_type,
-                card: None,
-                metadata: None,
-                created: Some(common_utils::date_time::now()),
-                recurring_enabled: false,           //[#219]
-                installment_payment_enabled: false, //[#219]
-                payment_experience: Some(vec![api_models::enums::PaymentExperience::RedirectToUrl]), //[#219]
-            };
-            Ok((payment_method_response, false))
-        }
-    }
-}
-
-pub fn create_payment_method_metadata(
-    metadata: Option<&pii::SecretSerdeValue>,
-    connector_token: Option<(&api::ConnectorData, String)>,
-) -> RouterResult<Option<serde_json::Value>> {
-    let mut meta = match metadata {
-        None => serde_json::Map::new(),
-        Some(meta) => {
-            let metadata = meta.clone().expose();
-            let existing_metadata: serde_json::Map<String, serde_json::Value> = metadata
-                .parse_value("Map<String, Value>")
-                .change_context(errors::ApiErrorResponse::InternalServerError)
-                .attach_printable("Failed to parse the metadata")?;
-            existing_metadata
-        }
-    };
-    Ok(connector_token.and_then(|connector_and_token| {
-        meta.insert(
-            connector_and_token.0.connector_name.to_string(),
-            serde_json::Value::String(connector_and_token.1),
-        )
-    }))
-}
-
 pub enum Action {
     Update,
     Insert,
@@ -373,140 +208,8 @@ impl mandate::MandateBehaviour for types::PaymentsAuthorizeData {
         self.setup_mandate_details.as_ref()
     }
 
-    fn set_mandate_id(&mut self, new_mandate_id: api_models::payments::MandateIds) {
-        self.mandate_id = Some(new_mandate_id);
-    }
-}
-
-pub async fn update_connector_customer_in_customers(
-    connector: &api::ConnectorData,
-    connector_customer_map: Option<serde_json::Map<String, serde_json::Value>>,
-    connector_cust_id: &Option<String>,
-) -> RouterResult<Option<storage::CustomerUpdate>> {
-    let mut connector_customer = match connector_customer_map {
-        Some(cc) => cc,
-        None => serde_json::Map::new(),
-    };
-    connector_cust_id.clone().map(|cc| {
-        connector_customer.insert(
-            connector.connector_name.to_string(),
-            serde_json::Value::String(cc),
-        )
-    });
-    Ok(Some(storage::CustomerUpdate::ConnectorCustomer {
-        connector_customer: Some(serde_json::Value::Object(connector_customer)),
-    }))
-}
-
-type CreateCustomerCheck = (
-    bool,
-    Option<String>,
-    Option<serde_json::Map<String, serde_json::Value>>,
-);
-pub fn should_call_connector_create_customer(
-    state: &AppState,
-    connector: &api::ConnectorData,
-    customer: &Option<storage::Customer>,
-) -> RouterResult<CreateCustomerCheck> {
-    let connector_name = connector.connector_name.to_string();
-    //Check if create customer is required for the connector
-    let connector_customer_filter = state
-        .conf
-        .connector_customer
-        .connector_list
-        .contains(&connector.connector_name);
-    if connector_customer_filter {
-        match customer {
-            Some(customer) => match &customer.connector_customer {
-                Some(connector_customer) => {
-                    let connector_customer_map: serde_json::Map<String, serde_json::Value> =
-                        connector_customer
-                            .clone()
-                            .parse_value("Map<String, Value>")
-                            .change_context(errors::ApiErrorResponse::InternalServerError)
-                            .attach_printable("Failed to deserialize Value to CustomerConnector")?;
-                    let value = connector_customer_map.get(&connector_name); //Check if customer already created for this customer and for this connector
-                    Ok((
-                        value.is_none(),
-                        value.and_then(|val| val.as_str().map(|cust| cust.to_string())),
-                        Some(connector_customer_map),
-                    ))
-                }
-                None => Ok((true, None, None)),
-            },
-            None => Ok((false, None, None)),
-        }
-    } else {
-        Ok((false, None, None))
-    }
-}
-
-pub async fn create_connector_customer<F: Clone>(
-    state: &AppState,
-    connector: &api::ConnectorData,
-    customer: &Option<storage::Customer>,
-    router_data: &types::RouterData<F, types::PaymentsAuthorizeData, types::PaymentsResponseData>,
-) -> RouterResult<(Option<String>, Option<storage::CustomerUpdate>)> {
-    let (is_eligible, connector_customer_id, connector_customer_map) =
-        should_call_connector_create_customer(state, connector, customer)?;
-
-    if is_eligible {
-        let connector_integration: services::BoxedConnectorIntegration<
-            '_,
-            api::CreateConnectorCustomer,
-            types::ConnectorCustomerData,
-            types::PaymentsResponseData,
-        > = connector.connector.get_connector_integration();
-
-        let customer_response_data: Result<types::PaymentsResponseData, types::ErrorResponse> =
-            Err(types::ErrorResponse::default());
-
-        let customer_request_data =
-            types::ConnectorCustomerData::try_from(router_data.request.to_owned())?;
-
-        let customer_router_data = payments::helpers::router_data_type_conversion::<
-            _,
-            api::CreateConnectorCustomer,
-            _,
-            _,
-            _,
-            _,
-        >(
-            router_data.clone(),
-            customer_request_data,
-            customer_response_data,
-        );
-
-        let resp = services::execute_connector_processing_step(
-            state,
-            connector_integration,
-            &customer_router_data,
-            payments::CallConnectorAction::Trigger,
-        )
-        .await
-        .map_err(|error| error.to_payment_failed_response())?;
-
-        let connector_customer_id = match resp.response {
-            Ok(response) => match response {
-                types::PaymentsResponseData::ConnectorCustomerResponse {
-                    connector_customer_id,
-                } => Some(connector_customer_id),
-                _ => None,
-            },
-            Err(err) => {
-                logger::debug!(payment_method_tokenization_error=?err);
-                None
-            }
-        };
-        let update_customer = update_connector_customer_in_customers(
-            connector,
-            connector_customer_map,
-            &connector_customer_id,
-        )
-        .await?;
-        Ok((connector_customer_id, update_customer))
-    } else {
-        Ok((connector_customer_id, None))
+    fn set_mandate_id(&mut self, new_mandate_id: Option<api_models::payments::MandateIds>) {
+        self.mandate_id = new_mandate_id;
     }
 }
 
@@ -523,64 +226,6 @@ impl TryFrom<types::PaymentsAuthorizeData> for types::ConnectorCustomerData {
     }
 }
 
-pub async fn add_payment_method_token<F: Clone>(
-    state: &AppState,
-    connector: &api::ConnectorData,
-    tokenization_action: &payments::TokenizationAction,
-    router_data: &types::RouterData<F, types::PaymentsAuthorizeData, types::PaymentsResponseData>,
-) -> RouterResult<Option<String>> {
-    match tokenization_action {
-        payments::TokenizationAction::TokenizeInConnector => {
-            let connector_integration: services::BoxedConnectorIntegration<
-                '_,
-                api::PaymentMethodToken,
-                types::PaymentMethodTokenizationData,
-                types::PaymentsResponseData,
-            > = connector.connector.get_connector_integration();
-
-            let pm_token_response_data: Result<types::PaymentsResponseData, types::ErrorResponse> =
-                Err(types::ErrorResponse::default());
-
-            let pm_token_request_data =
-                types::PaymentMethodTokenizationData::try_from(router_data.request.to_owned())?;
-
-            let pm_token_router_data = payments::helpers::router_data_type_conversion::<
-                _,
-                api::PaymentMethodToken,
-                _,
-                _,
-                _,
-                _,
-            >(
-                router_data.clone(),
-                pm_token_request_data,
-                pm_token_response_data,
-            );
-            let resp = services::execute_connector_processing_step(
-                state,
-                connector_integration,
-                &pm_token_router_data,
-                payments::CallConnectorAction::Trigger,
-            )
-            .await
-            .map_err(|error| error.to_payment_failed_response())?;
-
-            let pm_token = match resp.response {
-                Ok(response) => match response {
-                    types::PaymentsResponseData::TokenizationResponse { token } => Some(token),
-                    _ => None,
-                },
-                Err(err) => {
-                    logger::debug!(payment_method_tokenization_error=?err);
-                    None
-                }
-            };
-            Ok(pm_token)
-        }
-        _ => Ok(None),
-    }
-}
-
 impl TryFrom<types::PaymentsAuthorizeData> for types::PaymentMethodTokenizationData {
     type Error = error_stack::Report<errors::ApiErrorResponse>;
 
diff --git a/crates/router/src/core/payments/flows/verfiy_flow.rs b/crates/router/src/core/payments/flows/verfiy_flow.rs
index 2a6f38f144b..51745cbde12 100644
--- a/crates/router/src/core/payments/flows/verfiy_flow.rs
+++ b/crates/router/src/core/payments/flows/verfiy_flow.rs
@@ -1,11 +1,11 @@
 use async_trait::async_trait;
 
-use super::{authorize_flow, ConstructFlowSpecificData, Feature};
+use super::{ConstructFlowSpecificData, Feature};
 use crate::{
     core::{
-        errors::{ConnectorErrorExt, RouterResult},
+        errors::{self, ConnectorErrorExt, RouterResult},
         mandate,
-        payments::{self, access_token, transformers, PaymentData},
+        payments::{self, access_token, customers, tokenization, transformers, PaymentData},
     },
     routes::AppState,
     services,
@@ -63,6 +63,50 @@ impl Feature<api::Verify, types::VerifyRequestData> for types::VerifyRouterData
     ) -> RouterResult<types::AddAccessTokenResult> {
         access_token::add_access_token(state, connector, merchant_account, self).await
     }
+
+    async fn add_payment_method_token<'a>(
+        &self,
+        state: &AppState,
+        connector: &api::ConnectorData,
+        tokenization_action: &payments::TokenizationAction,
+    ) -> RouterResult<Option<String>> {
+        tokenization::add_payment_method_token(
+            state,
+            connector,
+            tokenization_action,
+            self,
+            types::PaymentMethodTokenizationData::try_from(self.request.to_owned())?,
+        )
+        .await
+    }
+
+    async fn create_connector_customer<'a>(
+        &self,
+        state: &AppState,
+        connector: &api::ConnectorData,
+        customer: &Option<storage::Customer>,
+    ) -> RouterResult<(Option<String>, Option<storage::CustomerUpdate>)> {
+        customers::create_connector_customer(
+            state,
+            connector,
+            customer,
+            self,
+            types::ConnectorCustomerData::try_from(self.request.to_owned())?,
+        )
+        .await
+    }
+}
+
+impl TryFrom<types::VerifyRequestData> for types::ConnectorCustomerData {
+    type Error = error_stack::Report<errors::ApiErrorResponse>;
+    fn try_from(data: types::VerifyRequestData) -> Result<Self, Self::Error> {
+        Ok(Self {
+            email: data.email,
+            description: None,
+            phone: None,
+            name: None,
+        })
+    }
 }
 
 impl types::VerifyRouterData {
@@ -73,7 +117,7 @@ impl types::VerifyRouterData {
         maybe_customer: &Option<storage::Customer>,
         confirm: Option<bool>,
         call_connector_action: payments::CallConnectorAction,
-        _merchant_account: &storage::MerchantAccount,
+        merchant_account: &storage::MerchantAccount,
     ) -> RouterResult<Self> {
         match confirm {
             Some(true) => {
@@ -92,12 +136,12 @@ impl types::VerifyRouterData {
                 .await
                 .map_err(|err| err.to_verify_failed_response())?;
 
-                let pm_id = authorize_flow::save_payment_method(
+                let pm_id = tokenization::save_payment_method(
                     state,
                     connector,
                     resp.to_owned(),
                     maybe_customer,
-                    _merchant_account,
+                    merchant_account,
                 )
                 .await?;
 
@@ -121,8 +165,8 @@ impl mandate::MandateBehaviour for types::VerifyRequestData {
         self.mandate_id.as_ref()
     }
 
-    fn set_mandate_id(&mut self, new_mandate_id: api_models::payments::MandateIds) {
-        self.mandate_id = Some(new_mandate_id);
+    fn set_mandate_id(&mut self, new_mandate_id: Option<api_models::payments::MandateIds>) {
+        self.mandate_id = new_mandate_id;
     }
 
     fn get_payment_method_data(&self) -> api_models::payments::PaymentMethodData {
@@ -133,3 +177,13 @@ impl mandate::MandateBehaviour for types::VerifyRequestData {
         self.setup_mandate_details.as_ref()
     }
 }
+
+impl TryFrom<types::VerifyRequestData> for types::PaymentMethodTokenizationData {
+    type Error = error_stack::Report<errors::ApiErrorResponse>;
+
+    fn try_from(data: types::VerifyRequestData) -> Result<Self, Self::Error> {
+        Ok(Self {
+            payment_method_data: data.payment_method_data,
+        })
+    }
+}
diff --git a/crates/router/src/core/payments/helpers.rs b/crates/router/src/core/payments/helpers.rs
index d6a3c4da585..97cbea16af7 100644
--- a/crates/router/src/core/payments/helpers.rs
+++ b/crates/router/src/core/payments/helpers.rs
@@ -2,7 +2,7 @@ use std::borrow::Cow;
 
 use common_utils::{
     ext_traits::{AsyncExt, ByteSliceExt, ValueExt},
-    fp_utils,
+    fp_utils, pii,
 };
 // TODO : Evaluate all the helper functions ()
 use error_stack::{report, IntoReport, ResultExt};
@@ -1144,7 +1144,7 @@ pub fn generate_mandate(
     setup_mandate_details: Option<api::MandateData>,
     customer: &Option<storage::Customer>,
     payment_method_id: String,
-    connector_mandate_id: Option<String>,
+    connector_mandate_id: Option<pii::SecretSerdeValue>,
     network_txn_id: Option<String>,
 ) -> Option<storage::MandateNew> {
     match (setup_mandate_details, customer) {
@@ -1161,7 +1161,7 @@ pub fn generate_mandate(
                 .set_payment_method_id(payment_method_id)
                 .set_connector(connector)
                 .set_mandate_status(storage_enums::MandateStatus::Active)
-                .set_connector_mandate_id(connector_mandate_id)
+                .set_connector_mandate_ids(connector_mandate_id)
                 .set_network_transaction_id(network_txn_id)
                 .set_customer_ip_address(
                     data.customer_acceptance
diff --git a/crates/router/src/core/payments/operations/payment_create.rs b/crates/router/src/core/payments/operations/payment_create.rs
index f3f9236a9d4..b4bad575c12 100644
--- a/crates/router/src/core/payments/operations/payment_create.rs
+++ b/crates/router/src/core/payments/operations/payment_create.rs
@@ -1,7 +1,7 @@
 use std::marker::PhantomData;
 
 use async_trait::async_trait;
-use common_utils::ext_traits::{AsyncExt, Encode};
+use common_utils::ext_traits::{AsyncExt, Encode, ValueExt};
 use error_stack::{self, ResultExt};
 use router_derive::PaymentOperation;
 use router_env::{instrument, tracing};
@@ -150,26 +150,38 @@ impl<F: Send + Clone> GetTracker<F, PaymentData<F>, api::PaymentsRequest> for Pa
                     .find_mandate_by_merchant_id_mandate_id(merchant_id, mandate_id)
                     .await
                     .change_context(errors::ApiErrorResponse::MandateNotFound);
-                Some(mandate.map(|mandate_obj| api_models::payments::MandateIds {
-                    mandate_id: mandate_obj.mandate_id,
-                    mandate_reference_id: {
-                        match (
-                            mandate_obj.network_transaction_id,
-                            mandate_obj.connector_mandate_id,
-                        ) {
-                            (Some(network_tx_id), _) => {
-                                Some(api_models::payments::MandateReferenceId::NetworkMandateId(
+                Some(mandate.and_then(|mandate_obj| {
+                    match (
+                        mandate_obj.network_transaction_id,
+                        mandate_obj.connector_mandate_ids,
+                    ) {
+                        (Some(network_tx_id), _) => Ok(api_models::payments::MandateIds {
+                            mandate_id: mandate_obj.mandate_id,
+                            mandate_reference_id: Some(
+                                api_models::payments::MandateReferenceId::NetworkMandateId(
                                     network_tx_id,
-                                ))
-                            }
-                            (_, Some(connector_mandate_id)) => Some(
-                                api_models::payments::MandateReferenceId::ConnectorMandateId(
-                                    connector_mandate_id,
                                 ),
                             ),
-                            (_, _) => None,
-                        }
-                    },
+                        }),
+                        (_, Some(connector_mandate_id)) => connector_mandate_id
+                        .parse_value("ConnectorMandateId")
+                        .change_context(errors::ApiErrorResponse::MandateNotFound)
+                        .map(|connector_id: api_models::payments::ConnectorMandateReferenceId| {
+                            api_models::payments::MandateIds {
+                                mandate_id: 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,
+                                    },
+                                ))
+                            }
+                         }),
+                        (_, _) => Ok(api_models::payments::MandateIds {
+                            mandate_id: mandate_obj.mandate_id,
+                            mandate_reference_id: None,
+                        }),
+                    }
                 }))
             })
             .await
diff --git a/crates/router/src/core/payments/operations/payment_update.rs b/crates/router/src/core/payments/operations/payment_update.rs
index 993979a0f6e..8c691a61948 100644
--- a/crates/router/src/core/payments/operations/payment_update.rs
+++ b/crates/router/src/core/payments/operations/payment_update.rs
@@ -1,7 +1,7 @@
 use std::marker::PhantomData;
 
 use async_trait::async_trait;
-use common_utils::ext_traits::{AsyncExt, Encode};
+use common_utils::ext_traits::{AsyncExt, Encode, ValueExt};
 use error_stack::ResultExt;
 use router_derive::PaymentOperation;
 use router_env::{instrument, tracing};
@@ -186,26 +186,38 @@ impl<F: Send + Clone> GetTracker<F, PaymentData<F>, api::PaymentsRequest> for Pa
                     .find_mandate_by_merchant_id_mandate_id(merchant_id, mandate_id)
                     .await
                     .change_context(errors::ApiErrorResponse::MandateNotFound);
-                Some(mandate.map(|mandate_obj| api_models::payments::MandateIds {
-                    mandate_id: mandate_obj.mandate_id,
-                    mandate_reference_id: {
-                        match (
-                            mandate_obj.network_transaction_id,
-                            mandate_obj.connector_mandate_id,
-                        ) {
-                            (Some(network_tx_id), _) => {
-                                Some(api_models::payments::MandateReferenceId::NetworkMandateId(
+                Some(mandate.and_then(|mandate_obj| {
+                    match (
+                        mandate_obj.network_transaction_id,
+                        mandate_obj.connector_mandate_ids,
+                    ) {
+                        (Some(network_tx_id), _) => Ok(api_models::payments::MandateIds {
+                            mandate_id: mandate_obj.mandate_id,
+                            mandate_reference_id: Some(
+                                api_models::payments::MandateReferenceId::NetworkMandateId(
                                     network_tx_id,
-                                ))
-                            }
-                            (_, Some(connector_mandate_id)) => Some(
-                                api_models::payments::MandateReferenceId::ConnectorMandateId(
-                                    connector_mandate_id,
                                 ),
                             ),
-                            (_, _) => None,
-                        }
-                    },
+                        }),
+                        (_, Some(connector_mandate_id)) => connector_mandate_id
+                        .parse_value("ConnectorMandateId")
+                        .change_context(errors::ApiErrorResponse::MandateNotFound)
+                        .map(|connector_id: api_models::payments::ConnectorMandateReferenceId| {
+                            api_models::payments::MandateIds {
+                                mandate_id: 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,
+                                    },
+                                ))
+                            }
+                         }),
+                        (_, _) => Ok(api_models::payments::MandateIds {
+                            mandate_id: mandate_obj.mandate_id,
+                            mandate_reference_id: None,
+                        }),
+                    }
                 }))
             })
             .await
diff --git a/crates/router/src/core/payments/tokenization.rs b/crates/router/src/core/payments/tokenization.rs
new file mode 100644
index 00000000000..3dec82570c0
--- /dev/null
+++ b/crates/router/src/core/payments/tokenization.rs
@@ -0,0 +1,248 @@
+use common_utils::{ext_traits::ValueExt, pii};
+use error_stack::{report, ResultExt};
+use masking::ExposeInterface;
+
+use super::helpers;
+use crate::{
+    core::{
+        errors::{self, ConnectorErrorExt, RouterResult},
+        mandate, payment_methods, payments,
+    },
+    logger,
+    routes::AppState,
+    services,
+    types::{
+        self,
+        api::{self, PaymentMethodCreateExt},
+        storage,
+    },
+    utils::OptionExt,
+};
+
+pub async fn save_payment_method<F: Clone, FData>(
+    state: &AppState,
+    connector: &api::ConnectorData,
+    resp: types::RouterData<F, FData, types::PaymentsResponseData>,
+    maybe_customer: &Option<storage::Customer>,
+    merchant_account: &storage::MerchantAccount,
+) -> RouterResult<Option<String>>
+where
+    FData: mandate::MandateBehaviour,
+{
+    let db = &*state.store;
+    let token_store = state
+        .conf
+        .tokenization
+        .0
+        .get(&connector.connector_name.to_string())
+        .map(|token_filter| token_filter.long_lived_token)
+        .unwrap_or(false);
+
+    let connector_token = if token_store {
+        let token = resp
+            .payment_method_token
+            .to_owned()
+            .get_required_value("payment_token")?;
+        Some((connector, token))
+    } else {
+        None
+    };
+
+    let pm_id = if resp.request.get_setup_future_usage().is_some() {
+        let customer = maybe_customer.to_owned().get_required_value("customer")?;
+        let payment_method_create_request = helpers::get_payment_method_create_request(
+            Some(&resp.request.get_payment_method_data()),
+            Some(resp.payment_method),
+            &customer,
+        )
+        .await?;
+        let merchant_id = &merchant_account.merchant_id;
+
+        let locker_response = save_in_locker(
+            state,
+            merchant_account,
+            payment_method_create_request.to_owned(),
+        )
+        .await?;
+        let is_duplicate = locker_response.1;
+
+        if is_duplicate {
+            let existing_pm = db
+                .find_payment_method(&locker_response.0.payment_method_id)
+                .await;
+            match existing_pm {
+                Ok(pm) => {
+                    let pm_metadata =
+                        create_payment_method_metadata(pm.metadata.as_ref(), connector_token)?;
+                    if let Some(metadata) = pm_metadata {
+                        payment_methods::cards::update_payment_method(db, pm, metadata)
+                            .await
+                            .change_context(errors::ApiErrorResponse::InternalServerError)
+                            .attach_printable("Failed to add payment method in db")?;
+                    };
+                }
+                Err(error) => {
+                    match error.current_context() {
+                        errors::StorageError::DatabaseError(err) => match err.current_context() {
+                            storage_models::errors::DatabaseError::NotFound => {
+                                let pm_metadata =
+                                    create_payment_method_metadata(None, connector_token)?;
+                                payment_methods::cards::create_payment_method(
+                                    db,
+                                    &payment_method_create_request,
+                                    &customer.customer_id,
+                                    &locker_response.0.payment_method_id,
+                                    merchant_id,
+                                    pm_metadata,
+                                )
+                                .await
+                                .change_context(errors::ApiErrorResponse::InternalServerError)
+                                .attach_printable("Failed to add payment method in db")
+                            }
+                            _ => Err(report!(errors::ApiErrorResponse::InternalServerError)),
+                        },
+                        _ => Err(report!(errors::ApiErrorResponse::InternalServerError)),
+                    }?;
+                }
+            };
+        } else {
+            let pm_metadata = create_payment_method_metadata(None, connector_token)?;
+            payment_methods::cards::create_payment_method(
+                db,
+                &payment_method_create_request,
+                &customer.customer_id,
+                &locker_response.0.payment_method_id,
+                merchant_id,
+                pm_metadata,
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Failed to add payment method in db")?;
+        };
+        Some(locker_response.0.payment_method_id)
+    } else {
+        None
+    };
+    Ok(pm_id)
+}
+
+pub async fn save_in_locker(
+    state: &AppState,
+    merchant_account: &storage::MerchantAccount,
+    payment_method_request: api::PaymentMethodCreate,
+) -> RouterResult<(api_models::payment_methods::PaymentMethodResponse, bool)> {
+    payment_method_request.validate()?;
+    let merchant_id = &merchant_account.merchant_id;
+    let customer_id = payment_method_request
+        .customer_id
+        .clone()
+        .get_required_value("customer_id")?;
+    match payment_method_request.card.clone() {
+        Some(card) => payment_methods::cards::add_card_to_locker(
+            state,
+            payment_method_request,
+            card,
+            customer_id,
+            merchant_account,
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Add Card Failed"),
+        None => {
+            let pm_id = common_utils::generate_id(crate::consts::ID_LENGTH, "pm");
+            let payment_method_response = api::PaymentMethodResponse {
+                merchant_id: merchant_id.to_string(),
+                customer_id: Some(customer_id),
+                payment_method_id: pm_id,
+                payment_method: payment_method_request.payment_method,
+                payment_method_type: payment_method_request.payment_method_type,
+                card: None,
+                metadata: None,
+                created: Some(common_utils::date_time::now()),
+                recurring_enabled: false,           //[#219]
+                installment_payment_enabled: false, //[#219]
+                payment_experience: Some(vec![api_models::enums::PaymentExperience::RedirectToUrl]), //[#219]
+            };
+            Ok((payment_method_response, false))
+        }
+    }
+}
+
+pub fn create_payment_method_metadata(
+    metadata: Option<&pii::SecretSerdeValue>,
+    connector_token: Option<(&api::ConnectorData, String)>,
+) -> RouterResult<Option<serde_json::Value>> {
+    let mut meta = match metadata {
+        None => serde_json::Map::new(),
+        Some(meta) => {
+            let metadata = meta.clone().expose();
+            let existing_metadata: serde_json::Map<String, serde_json::Value> = metadata
+                .parse_value("Map<String, Value>")
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("Failed to parse the metadata")?;
+            existing_metadata
+        }
+    };
+    Ok(connector_token.and_then(|connector_and_token| {
+        meta.insert(
+            connector_and_token.0.connector_name.to_string(),
+            serde_json::Value::String(connector_and_token.1),
+        )
+    }))
+}
+
+pub async fn add_payment_method_token<F: Clone, T: Clone>(
+    state: &AppState,
+    connector: &api::ConnectorData,
+    tokenization_action: &payments::TokenizationAction,
+    router_data: &types::RouterData<F, T, types::PaymentsResponseData>,
+    pm_token_request_data: types::PaymentMethodTokenizationData,
+) -> RouterResult<Option<String>> {
+    match tokenization_action {
+        payments::TokenizationAction::TokenizeInConnector => {
+            let connector_integration: services::BoxedConnectorIntegration<
+                '_,
+                api::PaymentMethodToken,
+                types::PaymentMethodTokenizationData,
+                types::PaymentsResponseData,
+            > = connector.connector.get_connector_integration();
+
+            let pm_token_response_data: Result<types::PaymentsResponseData, types::ErrorResponse> =
+                Err(types::ErrorResponse::default());
+
+            let pm_token_router_data = payments::helpers::router_data_type_conversion::<
+                _,
+                api::PaymentMethodToken,
+                _,
+                _,
+                _,
+                _,
+            >(
+                router_data.clone(),
+                pm_token_request_data,
+                pm_token_response_data,
+            );
+            let resp = services::execute_connector_processing_step(
+                state,
+                connector_integration,
+                &pm_token_router_data,
+                payments::CallConnectorAction::Trigger,
+            )
+            .await
+            .map_err(|error| error.to_payment_failed_response())?;
+
+            let pm_token = match resp.response {
+                Ok(response) => match response {
+                    types::PaymentsResponseData::TokenizationResponse { token } => Some(token),
+                    _ => None,
+                },
+                Err(err) => {
+                    logger::debug!(payment_method_tokenization_error=?err);
+                    None
+                }
+            };
+            Ok(pm_token)
+        }
+        _ => Ok(None),
+    }
+}
diff --git a/crates/router/src/core/payments/transformers.rs b/crates/router/src/core/payments/transformers.rs
index 53948e7af02..301a64308ee 100644
--- a/crates/router/src/core/payments/transformers.rs
+++ b/crates/router/src/core/payments/transformers.rs
@@ -521,13 +521,10 @@ impl<F: Clone> TryFrom<PaymentAdditionalData<'_, F>> for types::PaymentsAuthoriz
         ));
 
         // payment_method_data is not required during recurring mandate payment, in such case keep default PaymentMethodData as MandatePayment
-        let payment_method_data = payment_data.payment_method_data.or_else(|| {
-            if payment_data.mandate_id.is_some() {
-                Some(api_models::payments::PaymentMethodData::MandatePayment)
-            } else {
-                None
-            }
-        });
+        let payment_method_data = payment_data
+            .mandate_id
+            .as_ref()
+            .map(|_| api_models::payments::PaymentMethodData::MandatePayment);
         Ok(Self {
             payment_method_data: payment_method_data.get_required_value("payment_method_data")?,
             setup_future_usage: payment_data.payment_intent.setup_future_usage,
@@ -700,6 +697,7 @@ impl<F: Clone> TryFrom<PaymentAdditionalData<'_, F>> for types::VerifyRequestDat
             off_session: payment_data.mandate_id.as_ref().map(|_| true),
             mandate_id: payment_data.mandate_id.clone(),
             setup_mandate_details: payment_data.setup_mandate,
+            email: payment_data.email,
             return_url: payment_data.payment_intent.return_url,
         })
     }
diff --git a/crates/router/src/types.rs b/crates/router/src/types.rs
index d67dd9f7b4e..f1af905237e 100644
--- a/crates/router/src/types.rs
+++ b/crates/router/src/types.rs
@@ -299,6 +299,7 @@ pub struct VerifyRequestData {
     pub setup_future_usage: Option<storage_enums::FutureUsage>,
     pub off_session: Option<bool>,
     pub setup_mandate_details: Option<payments::MandateData>,
+    pub email: Option<Email>,
     pub return_url: Option<String>,
 }
 
@@ -320,12 +321,18 @@ pub struct AccessToken {
     pub expires: i64,
 }
 
+#[derive(serde::Serialize, Debug, Clone)]
+pub struct MandateReference {
+    pub connector_mandate_id: Option<String>,
+    pub payment_method_id: Option<String>,
+}
+
 #[derive(Debug, Clone)]
 pub enum PaymentsResponseData {
     TransactionResponse {
         resource_id: ResponseId,
         redirection_data: Option<services::RedirectForm>,
-        mandate_reference: Option<String>,
+        mandate_reference: Option<MandateReference>,
         connector_metadata: Option<serde_json::Value>,
         network_txn_id: Option<String>,
     },
diff --git a/crates/router/tests/connectors/main.rs b/crates/router/tests/connectors/main.rs
index 607ca097480..95b79c6166d 100644
--- a/crates/router/tests/connectors/main.rs
+++ b/crates/router/tests/connectors/main.rs
@@ -29,6 +29,7 @@ mod rapyd;
 mod selenium;
 mod shift4;
 mod stripe;
+mod stripe_ui;
 mod trustpay;
 mod utils;
 mod worldline;
diff --git a/crates/router/tests/connectors/nuvei_ui.rs b/crates/router/tests/connectors/nuvei_ui.rs
index 784bc7ff8b6..5bb1fce88e7 100644
--- a/crates/router/tests/connectors/nuvei_ui.rs
+++ b/crates/router/tests/connectors/nuvei_ui.rs
@@ -10,8 +10,8 @@ impl SeleniumTest for NuveiSeleniumTest {}
 async fn should_make_nuvei_3ds_payment(c: WebDriver) -> Result<(), WebDriverError> {
     let conn = NuveiSeleniumTest {};
     conn.make_redirection_payment(c, vec![
-            Event::Trigger(Trigger::Goto("https://hs-payment-tests.w3spaces.com?pay-mode=pm-card&cname=CL-BRW1&ccnum=4000027891380961&expmonth=10&expyear=25&cvv=123&amount=200&country=US&currency=USD")),
-            Event::Assert(Assert::IsPresent("Exp Year")),
+            Event::Trigger(Trigger::Goto(&format!("{CHEKOUT_BASE_URL}/card?cname=CL-BRW1&ccnum=4000027891380961&expmonth=10&expyear=25&cvv=123&amount=200&country=US&currency=USD"))),
+            Event::Assert(Assert::IsPresent("Expiry Year")),
             Event::Trigger(Trigger::Click(By::Id("card-submit-btn"))),
             Event::Trigger(Trigger::Query(By::ClassName("title"))),
             Event::Assert(Assert::Eq(Selector::Title, "ThreeDS ACS Emulator - Challenge Page")),
@@ -27,7 +27,7 @@ async fn should_make_nuvei_3ds_payment(c: WebDriver) -> Result<(), WebDriverErro
 async fn should_make_nuvei_3ds_mandate_payment(c: WebDriver) -> Result<(), WebDriverError> {
     let conn = NuveiSeleniumTest {};
     conn.make_redirection_payment(c, vec![
-            Event::Trigger(Trigger::Goto("https://hs-payment-tests.w3spaces.com?pay-mode=pm-card&cname=CL-BRW1&ccnum=4000027891380961&expmonth=10&expyear=25&cvv=123&amount=200&country=US&currency=USD&setup_future_usage=off_session&mandate_data[customer_acceptance][acceptance_type]=offline&mandate_data[customer_acceptance][accepted_at]=1963-05-03T04:07:52.723Z&mandate_data[customer_acceptance][online][ip_address]=in%20sit&mandate_data[customer_acceptance][online][user_agent]=amet%20irure%20esse&mandate_data[mandate_type][multi_use][amount]=7000&mandate_data[mandate_type][multi_use][currency]=USD&mandate_data[mandate_type][multi_use][start_date]=2022-09-10T00:00:00Z&mandate_data[mandate_type][multi_use][end_date]=2023-09-10T00:00:00Z&mandate_data[mandate_type][multi_use][metadata][frequency]=13")),
+            Event::Trigger(Trigger::Goto(&format!("{CHEKOUT_BASE_URL}/card?cname=CL-BRW1&ccnum=4000027891380961&expmonth=10&expyear=25&cvv=123&amount=200&country=US&currency=USD&setup_future_usage=off_session&mandate_data[customer_acceptance][acceptance_type]=offline&mandate_data[customer_acceptance][accepted_at]=1963-05-03T04:07:52.723Z&mandate_data[customer_acceptance][online][ip_address]=in%20sit&mandate_data[customer_acceptance][online][user_agent]=amet%20irure%20esse&mandate_data[mandate_type][multi_use][amount]=7000&mandate_data[mandate_type][multi_use][currency]=USD&mandate_data[mandate_type][multi_use][start_date]=2022-09-10T00:00:00Z&mandate_data[mandate_type][multi_use][end_date]=2023-09-10T00:00:00Z&mandate_data[mandate_type][multi_use][metadata][frequency]=13"))),
             Event::Trigger(Trigger::Click(By::Id("card-submit-btn"))),
             Event::Trigger(Trigger::Query(By::ClassName("title"))),
             Event::Assert(Assert::Eq(Selector::Title, "ThreeDS ACS Emulator - Challenge Page")),
@@ -35,6 +35,7 @@ async fn should_make_nuvei_3ds_mandate_payment(c: WebDriver) -> Result<(), WebDr
             Event::Trigger(Trigger::Click(By::Id("btn5"))),
             Event::Assert(Assert::IsPresent("Google")),
             Event::Assert(Assert::Contains(Selector::QueryParamStr, "status=succeeded")),
+            Event::Assert(Assert::IsPresent("man_")),//mandate id prefix is present
 
     ]).await?;
     Ok(())
@@ -43,7 +44,7 @@ async fn should_make_nuvei_3ds_mandate_payment(c: WebDriver) -> Result<(), WebDr
 async fn should_make_nuvei_gpay_payment(c: WebDriver) -> Result<(), WebDriverError> {
     let conn = NuveiSeleniumTest {};
     conn.make_gpay_payment(c,
-        "https://hs-payment-tests.w3spaces.com?pay-mode=pm-gpay&gatewayname=nuveidigital&gatewaymerchantid=googletest&amount=10.00&country=IN&currency=USD",
+        &format!("{CHEKOUT_BASE_URL}/gpay?gatewayname=nuveidigital&gatewaymerchantid=googletest&amount=10.00&country=IN&currency=USD"),
         vec![
         Event::Assert(Assert::IsPresent("succeeded")),
     ]).await?;
@@ -52,18 +53,21 @@ async fn should_make_nuvei_gpay_payment(c: WebDriver) -> Result<(), WebDriverErr
 
 async fn should_make_nuvei_pypl_payment(c: WebDriver) -> Result<(), WebDriverError> {
     let conn = NuveiSeleniumTest {};
-    conn.make_paypal_payment(c,
-        "https://hs-payment-tests.w3spaces.com?pay-mode=pypl-redirect&amount=12.00&country=US&currency=USD",
-        vec![
-    Event::Assert(Assert::IsPresent("Your transaction has been successfully executed.")),
-    ]).await?;
+    conn.make_paypal_payment(
+        c,
+        &format!("{CHEKOUT_BASE_URL}/paypal-redirect?amount=12.00&country=US&currency=USD"),
+        vec![Event::Assert(Assert::IsPresent(
+            "Your transaction has been successfully executed.",
+        ))],
+    )
+    .await?;
     Ok(())
 }
 
 async fn should_make_nuvei_giropay_payment(c: WebDriver) -> Result<(), WebDriverError> {
     let conn = NuveiSeleniumTest {};
     conn.make_redirection_payment(c, vec![
-            Event::Trigger(Trigger::Goto("https://hs-payment-tests.w3spaces.com?pay-mode=bank-redirect&amount=1.00&country=DE&currency=EUR&paymentmethod=giropay")),
+            Event::Trigger(Trigger::Goto(&format!("{CHEKOUT_BASE_URL}/bank-redirect?amount=1.00&country=DE&currency=EUR&paymentmethod=giropay"))),
             Event::Trigger(Trigger::Click(By::Id("bank-redirect-btn"))),
             Event::Assert(Assert::IsPresent("You are about to make a payment using the Giropay service.")),
             Event::Trigger(Trigger::Click(By::Id("ctl00_ctl00_mainContent_btnConfirm"))),
@@ -86,7 +90,7 @@ async fn should_make_nuvei_giropay_payment(c: WebDriver) -> Result<(), WebDriver
 async fn should_make_nuvei_ideal_payment(c: WebDriver) -> Result<(), WebDriverError> {
     let conn = NuveiSeleniumTest {};
     conn.make_redirection_payment(c, vec![
-            Event::Trigger(Trigger::Goto("https://hs-payment-tests.w3spaces.com?pay-mode=bank-redirect&amount=10.00&country=NL&currency=EUR&paymentmethod=ideal&processingbank=ing")),
+            Event::Trigger(Trigger::Goto(&format!("{CHEKOUT_BASE_URL}/bank-redirect?amount=10.00&country=NL&currency=EUR&paymentmethod=ideal&processingbank=ing"))),
             Event::Trigger(Trigger::Click(By::Id("bank-redirect-btn"))),
             Event::Assert(Assert::IsPresent("Your account will be debited:")),
             Event::Trigger(Trigger::SelectOption(By::Id("ctl00_ctl00_mainContent_ServiceContent_ddlBanks"), "ING Simulator")),
@@ -102,7 +106,7 @@ async fn should_make_nuvei_ideal_payment(c: WebDriver) -> Result<(), WebDriverEr
 async fn should_make_nuvei_sofort_payment(c: WebDriver) -> Result<(), WebDriverError> {
     let conn = NuveiSeleniumTest {};
     conn.make_redirection_payment(c, vec![
-            Event::Trigger(Trigger::Goto("https://hs-payment-tests.w3spaces.com?pay-mode=bank-redirect&amount=10.00&country=DE&currency=EUR&paymentmethod=sofort")),
+            Event::Trigger(Trigger::Goto(&format!("{CHEKOUT_BASE_URL}/bank-redirect?amount=10.00&country=DE&currency=EUR&paymentmethod=sofort"))),
             Event::Trigger(Trigger::Click(By::Id("bank-redirect-btn"))),
             Event::Assert(Assert::IsPresent("SOFORT")),
             Event::Trigger(Trigger::ChangeQueryParam("sender_holder", "John Doe")),
@@ -115,7 +119,7 @@ async fn should_make_nuvei_sofort_payment(c: WebDriver) -> Result<(), WebDriverE
 async fn should_make_nuvei_eps_payment(c: WebDriver) -> Result<(), WebDriverError> {
     let conn = NuveiSeleniumTest {};
     conn.make_redirection_payment(c, vec![
-            Event::Trigger(Trigger::Goto("https://hs-payment-tests.w3spaces.com?pay-mode=bank-redirect&amount=10.00&country=AT&currency=EUR&paymentmethod=eps&processingbank=ing")),
+            Event::Trigger(Trigger::Goto(&format!("{CHEKOUT_BASE_URL}/bank-redirect?amount=10.00&country=AT&currency=EUR&paymentmethod=eps&processingbank=ing"))),
             Event::Trigger(Trigger::Click(By::Id("bank-redirect-btn"))),
             Event::Assert(Assert::IsPresent("You are about to make a payment using the EPS service.")),
             Event::Trigger(Trigger::SendKeys(By::Id("ctl00_ctl00_mainContent_ServiceContent_txtCustomerName"), "John Doe")),
@@ -131,47 +135,47 @@ async fn should_make_nuvei_eps_payment(c: WebDriver) -> Result<(), WebDriverErro
 #[test]
 #[serial]
 fn should_make_nuvei_3ds_payment_test() {
-    tester!(should_make_nuvei_3ds_payment, "firefox");
+    tester!(should_make_nuvei_3ds_payment);
 }
 
 #[test]
 #[serial]
 fn should_make_nuvei_3ds_mandate_payment_test() {
-    tester!(should_make_nuvei_3ds_mandate_payment, "firefox");
+    tester!(should_make_nuvei_3ds_mandate_payment);
 }
 
 #[test]
 #[serial]
 fn should_make_nuvei_gpay_payment_test() {
-    tester!(should_make_nuvei_gpay_payment, "firefox");
+    tester!(should_make_nuvei_gpay_payment);
 }
 
 #[test]
 #[serial]
 fn should_make_nuvei_pypl_payment_test() {
-    tester!(should_make_nuvei_pypl_payment, "firefox");
+    tester!(should_make_nuvei_pypl_payment);
 }
 
 #[test]
 #[serial]
 fn should_make_nuvei_giropay_payment_test() {
-    tester!(should_make_nuvei_giropay_payment, "firefox");
+    tester!(should_make_nuvei_giropay_payment);
 }
 
 #[test]
 #[serial]
 fn should_make_nuvei_ideal_payment_test() {
-    tester!(should_make_nuvei_ideal_payment, "firefox");
+    tester!(should_make_nuvei_ideal_payment);
 }
 
 #[test]
 #[serial]
 fn should_make_nuvei_sofort_payment_test() {
-    tester!(should_make_nuvei_sofort_payment, "firefox");
+    tester!(should_make_nuvei_sofort_payment);
 }
 
 #[test]
 #[serial]
 fn should_make_nuvei_eps_payment_test() {
-    tester!(should_make_nuvei_eps_payment, "firefox");
+    tester!(should_make_nuvei_eps_payment);
 }
diff --git a/crates/router/tests/connectors/selenium.rs b/crates/router/tests/connectors/selenium.rs
index 0114d510b30..c599965feff 100644
--- a/crates/router/tests/connectors/selenium.rs
+++ b/crates/router/tests/connectors/selenium.rs
@@ -2,7 +2,6 @@ use std::{collections::HashMap, env, path::MAIN_SEPARATOR, time::Duration};
 
 use actix_web::cookie::SameSite;
 use async_trait::async_trait;
-use futures::Future;
 use thirtyfour::{components::SelectElement, prelude::*, WebDriver};
 
 pub enum Event<'a> {
@@ -42,6 +41,8 @@ pub enum Assert<'a> {
     IsPresent(&'a str),
 }
 
+pub static CHEKOUT_BASE_URL: &str = "https://hs-payments-test.netlify.app";
+pub static CHEKOUT_DOMAIN: &str = "hs-payments-test.netlify.app";
 #[async_trait]
 pub trait SeleniumTest {
     async fn complete_actions(
@@ -204,14 +205,6 @@ pub trait SeleniumTest {
         Ok(())
     }
 
-    async fn process_payment<F, Fut>(&self, _f: F) -> Result<(), WebDriverError>
-    where
-        F: FnOnce(WebDriver) -> Fut + Send,
-        Fut: Future<Output = Result<(), WebDriverError>> + Send,
-    {
-        let _browser = env::var("HS_TEST_BROWSER").unwrap_or("chrome".to_string()); //Issue: #924
-        Ok(())
-    }
     async fn make_redirection_payment(
         &self,
         c: WebDriver,
@@ -231,7 +224,7 @@ pub trait SeleniumTest {
         );
         let default_actions = vec![
             Event::Trigger(Trigger::Goto(url)),
-            Event::Trigger(Trigger::Click(By::Css("#gpay-btn button"))),
+            Event::Trigger(Trigger::Click(By::Css(".gpay-button"))),
             Event::Trigger(Trigger::SwitchTab(Position::Next)),
             Event::RunIf(
                 Assert::IsPresent("Sign in"),
@@ -255,9 +248,7 @@ pub trait SeleniumTest {
                     ),
                 ],
             ),
-            Event::Trigger(Trigger::Query(By::ClassName(
-                "bootstrapperIframeContainerElement",
-            ))),
+            Event::Trigger(Trigger::Sleep(5)),
             Event::Trigger(Trigger::SwitchFrame(By::Id("sM432dIframe"))),
             Event::Assert(Assert::IsPresent("Gpay Tester")),
             Event::Trigger(Trigger::Click(By::ClassName("jfk-button-action"))),
@@ -314,7 +305,7 @@ async fn is_text_present(driver: &WebDriver, key: &str) -> WebDriverResult<bool>
 fn new_cookie(name: &str, value: String) -> Cookie<'_> {
     let mut base_url_cookie = Cookie::new(name, value);
     base_url_cookie.set_same_site(Some(SameSite::Lax));
-    base_url_cookie.set_domain("hs-payment-tests.w3spaces.com");
+    base_url_cookie.set_domain(CHEKOUT_DOMAIN);
     base_url_cookie.set_path("/");
     base_url_cookie
 }
@@ -362,15 +353,19 @@ macro_rules! tester_inner {
 
 #[macro_export]
 macro_rules! tester {
-    ($f:ident, $endpoint:expr) => {{
+    ($f:ident) => {{
         use $crate::tester_inner;
-
-        let url = make_url($endpoint);
-        let caps = make_capabilities($endpoint);
+        let browser = get_browser();
+        let url = make_url(&browser);
+        let caps = make_capabilities(&browser);
         tester_inner!($f, WebDriver::new(url, caps));
     }};
 }
 
+pub fn get_browser() -> String {
+    env::var("HS_TEST_BROWSER").unwrap_or("firefox".to_string()) //Issue: #924
+}
+
 pub fn make_capabilities(s: &str) -> Capabilities {
     match s {
         "firefox" => {
diff --git a/crates/router/tests/connectors/stripe_ui.rs b/crates/router/tests/connectors/stripe_ui.rs
new file mode 100644
index 00000000000..ab04fc70ef5
--- /dev/null
+++ b/crates/router/tests/connectors/stripe_ui.rs
@@ -0,0 +1,140 @@
+use serial_test::serial;
+use thirtyfour::{prelude::*, WebDriver};
+
+use crate::{selenium::*, tester};
+
+struct StripeSeleniumTest;
+
+impl SeleniumTest for StripeSeleniumTest {}
+
+async fn should_make_stripe_3ds_payment(c: WebDriver) -> Result<(), WebDriverError> {
+    let conn = StripeSeleniumTest {};
+    conn.make_redirection_payment(c, vec![
+            Event::Trigger(Trigger::Goto(&format!("{CHEKOUT_BASE_URL}/card?cname=CL-BRW1&ccnum=4000000000003063&expmonth=10&expyear=25&cvv=123&amount=100&country=US&currency=USD"))),
+            Event::Trigger(Trigger::Click(By::Id("card-submit-btn"))),
+            Event::Trigger(Trigger::Click(By::Id("test-source-authorize-3ds"))),
+            Event::Assert(Assert::IsPresent("Google")),
+            Event::Assert(Assert::Contains(Selector::QueryParamStr, "status=succeeded")),
+
+    ]).await?;
+    Ok(())
+}
+
+async fn should_make_stripe_3ds_mandate_payment(c: WebDriver) -> Result<(), WebDriverError> {
+    let conn = StripeSeleniumTest {};
+    conn.make_redirection_payment(c, vec![
+            Event::Trigger(Trigger::Goto(&format!("{CHEKOUT_BASE_URL}/card?cname=CL-BRW1&ccnum=4000002500003155&expmonth=10&expyear=25&cvv=123&amount=10&country=US&currency=USD&mandate_data[customer_acceptance][acceptance_type]=offline&mandate_data[customer_acceptance][accepted_at]=1963-05-03T04:07:52.723Z&mandate_data[customer_acceptance][online][ip_address]=127.0.0.1&mandate_data[customer_acceptance][online][user_agent]=amet%20irure%20esse&mandate_data[mandate_type][multi_use][amount]=700&mandate_data[mandate_type][multi_use][currency]=USD&apikey=dev_DREFLPJC5SFpFBupKYovdCfg37xgM20g7oXVLQMHXP3t2kJMRSy6aof1rTe6tyyK&return_url={CHEKOUT_BASE_URL}/payments"))),
+            Event::Trigger(Trigger::Click(By::Id("card-submit-btn"))),
+            Event::Trigger(Trigger::Click(By::Id("test-source-authorize-3ds"))),
+            Event::Assert(Assert::IsPresent("succeeded")),
+            Event::Assert(Assert::IsPresent("Mandate ID")),
+            Event::Assert(Assert::IsPresent("man_")),// mandate id starting with man_
+            Event::Trigger(Trigger::Click(By::Id("pm-mandate-btn"))),
+            Event::Trigger(Trigger::Click(By::Id("pay-with-mandate-btn"))),
+            Event::Assert(Assert::IsPresent("succeeded")),
+
+    ]).await?;
+    Ok(())
+}
+
+async fn should_fail_recurring_payment_due_to_authentication(
+    c: WebDriver,
+) -> Result<(), WebDriverError> {
+    let conn = StripeSeleniumTest {};
+    conn.make_redirection_payment(c, vec![
+            Event::Trigger(Trigger::Goto(&format!("{CHEKOUT_BASE_URL}/card?cname=CL-BRW1&ccnum=4000002760003184&expmonth=10&expyear=25&cvv=123&amount=10&country=US&currency=USD&mandate_data[customer_acceptance][acceptance_type]=offline&mandate_data[customer_acceptance][accepted_at]=1963-05-03T04:07:52.723Z&mandate_data[customer_acceptance][online][ip_address]=127.0.0.1&mandate_data[customer_acceptance][online][user_agent]=amet%20irure%20esse&mandate_data[mandate_type][multi_use][amount]=700&mandate_data[mandate_type][multi_use][currency]=USD&apikey=dev_DREFLPJC5SFpFBupKYovdCfg37xgM20g7oXVLQMHXP3t2kJMRSy6aof1rTe6tyyK&return_url={CHEKOUT_BASE_URL}/payments"))),
+            Event::Trigger(Trigger::Click(By::Id("card-submit-btn"))),
+            Event::Trigger(Trigger::Click(By::Id("test-source-authorize-3ds"))),
+            Event::Assert(Assert::IsPresent("succeeded")),
+            Event::Assert(Assert::IsPresent("Mandate ID")),
+            Event::Assert(Assert::IsPresent("man_")),// mandate id starting with man_
+            Event::Trigger(Trigger::Click(By::Id("pm-mandate-btn"))),
+            Event::Trigger(Trigger::Click(By::Id("pay-with-mandate-btn"))),
+            Event::Assert(Assert::IsPresent("authentication_required: Your card was declined. This transaction requires authentication.")),
+
+    ]).await?;
+    Ok(())
+}
+
+async fn should_make_stripe_3ds_mandate_with_zero_dollar_payment(
+    c: WebDriver,
+) -> Result<(), WebDriverError> {
+    let conn = StripeSeleniumTest {};
+    conn.make_redirection_payment(c, vec![
+            Event::Trigger(Trigger::Goto(&format!("{CHEKOUT_BASE_URL}/card?cname=CL-BRW1&ccnum=4000002500003155&expmonth=10&expyear=25&cvv=123&amount=0&country=US&currency=USD&mandate_data[customer_acceptance][acceptance_type]=offline&mandate_data[customer_acceptance][accepted_at]=1963-05-03T04:07:52.723Z&mandate_data[customer_acceptance][online][ip_address]=127.0.0.1&mandate_data[customer_acceptance][online][user_agent]=amet%20irure%20esse&mandate_data[mandate_type][multi_use][amount]=700&mandate_data[mandate_type][multi_use][currency]=USD&apikey=dev_DREFLPJC5SFpFBupKYovdCfg37xgM20g7oXVLQMHXP3t2kJMRSy6aof1rTe6tyyK&return_url={CHEKOUT_BASE_URL}/payments"))),
+            Event::Trigger(Trigger::Click(By::Id("card-submit-btn"))),
+            Event::Trigger(Trigger::Click(By::Id("test-source-authorize-3ds"))),
+            Event::Assert(Assert::IsPresent("succeeded")),
+            Event::Assert(Assert::IsPresent("Mandate ID")),
+            Event::Assert(Assert::IsPresent("man_")),// mandate id starting with man_
+            Event::Trigger(Trigger::Click(By::Id("pm-mandate-btn"))),
+            Event::Trigger(Trigger::Click(By::Id("pay-with-mandate-btn"))),
+            // Need to be handled as mentioned in https://stripe.com/docs/payments/save-and-reuse?platform=web#charge-saved-payment-method
+            Event::Assert(Assert::IsPresent("succeeded")),
+
+    ]).await?;
+    Ok(())
+}
+
+async fn should_make_stripe_gpay_payment(c: WebDriver) -> Result<(), WebDriverError> {
+    let conn = StripeSeleniumTest {};
+    conn.make_gpay_payment(c,
+        &format!("{CHEKOUT_BASE_URL}/gpay?gatewayname=stripe&gpaycustomfields[stripe:version]=2018-10-31&gpaycustomfields[stripe:publishableKey]=pk_test_51Msk2GAGHc77EJXX78h549SX2uaOnEkUYqBfjcoD05PIpAnDkYxMn8nQ4d19im85NQuX4Z6WDyHaUw2fFTPBWsIY00Wa7oNerO&amount=70.00&country=US&currency=USD"),
+        vec![
+        Event::Assert(Assert::IsPresent("succeeded")),
+    ]).await?;
+    Ok(())
+}
+
+async fn should_make_stripe_gpay_mandate_payment(c: WebDriver) -> Result<(), WebDriverError> {
+    let conn = StripeSeleniumTest {};
+    conn.make_gpay_payment(c,
+        &format!("{CHEKOUT_BASE_URL}/gpay?gatewayname=stripe&gpaycustomfields[stripe:version]=2018-10-31&gpaycustomfields[stripe:publishableKey]=pk_test_51Msk2GAGHc77EJXX78h549SX2uaOnEkUYqBfjcoD05PIpAnDkYxMn8nQ4d19im85NQuX4Z6WDyHaUw2fFTPBWsIY00Wa7oNerO&amount=70.00&country=US&currency=USD&mandate_data[customer_acceptance][acceptance_type]=offline&mandate_data[customer_acceptance][accepted_at]=1963-05-03T04:07:52.723Z&mandate_data[customer_acceptance][online][ip_address]=127.0.0.1&mandate_data[customer_acceptance][online][user_agent]=amet%20irure%20esse&mandate_data[mandate_type][multi_use][amount]=700&mandate_data[mandate_type][multi_use][currency]=USD"),
+        vec![
+        Event::Assert(Assert::IsPresent("succeeded")),
+        Event::Assert(Assert::IsPresent("Mandate ID")),
+        Event::Assert(Assert::IsPresent("man_")),// mandate id starting with man_
+        Event::Trigger(Trigger::Click(By::Id("pm-mandate-btn"))),
+        Event::Trigger(Trigger::Click(By::Id("pay-with-mandate-btn"))),
+        Event::Assert(Assert::IsPresent("succeeded")),
+    ]).await?;
+    Ok(())
+}
+
+//https://stripe.com/docs/testing#regulatory-cards
+
+#[test]
+#[serial]
+fn should_make_stripe_3ds_payment_test() {
+    tester!(should_make_stripe_3ds_payment);
+}
+
+#[test]
+#[serial]
+fn should_make_stripe_3ds_mandate_payment_test() {
+    tester!(should_make_stripe_3ds_mandate_payment);
+}
+
+#[test]
+#[serial]
+fn should_fail_recurring_payment_due_to_authentication_test() {
+    tester!(should_fail_recurring_payment_due_to_authentication);
+}
+
+#[test]
+#[serial]
+fn should_make_stripe_3ds_mandate_with_zero_dollar_payment_test() {
+    tester!(should_make_stripe_3ds_mandate_with_zero_dollar_payment);
+}
+
+#[test]
+#[serial]
+fn should_make_stripe_gpay_payment_test() {
+    tester!(should_make_stripe_gpay_payment);
+}
+
+#[test]
+#[serial]
+fn should_make_stripe_gpay_mandate_payment_test() {
+    tester!(should_make_stripe_gpay_mandate_payment);
+}
diff --git a/crates/storage_models/src/mandate.rs b/crates/storage_models/src/mandate.rs
index 3a1bbc65799..b6c27a4aa12 100644
--- a/crates/storage_models/src/mandate.rs
+++ b/crates/storage_models/src/mandate.rs
@@ -29,6 +29,7 @@ pub struct Mandate {
     pub start_date: Option<PrimitiveDateTime>,
     pub end_date: Option<PrimitiveDateTime>,
     pub metadata: Option<pii::SecretSerdeValue>,
+    pub connector_mandate_ids: Option<pii::SecretSerdeValue>,
 }
 
 #[derive(
@@ -56,6 +57,7 @@ pub struct MandateNew {
     pub start_date: Option<PrimitiveDateTime>,
     pub end_date: Option<PrimitiveDateTime>,
     pub metadata: Option<pii::SecretSerdeValue>,
+    pub connector_mandate_ids: Option<pii::SecretSerdeValue>,
 }
 
 #[derive(Debug)]
@@ -67,7 +69,7 @@ pub enum MandateUpdate {
         amount_captured: Option<i64>,
     },
     ConnectorReferenceUpdate {
-        connector_mandate_id: Option<String>,
+        connector_mandate_ids: Option<pii::SecretSerdeValue>,
     },
 }
 
@@ -82,7 +84,7 @@ pub struct SingleUseMandate {
 pub struct MandateUpdateInternal {
     mandate_status: Option<storage_enums::MandateStatus>,
     amount_captured: Option<i64>,
-    connector_mandate_id: Option<String>,
+    connector_mandate_ids: Option<pii::SecretSerdeValue>,
 }
 
 impl From<MandateUpdate> for MandateUpdateInternal {
@@ -90,18 +92,18 @@ impl From<MandateUpdate> for MandateUpdateInternal {
         match mandate_update {
             MandateUpdate::StatusUpdate { mandate_status } => Self {
                 mandate_status: Some(mandate_status),
-                connector_mandate_id: None,
+                connector_mandate_ids: None,
                 amount_captured: None,
             },
             MandateUpdate::CaptureAmountUpdate { amount_captured } => Self {
                 mandate_status: None,
                 amount_captured,
-                connector_mandate_id: None,
+                connector_mandate_ids: None,
             },
             MandateUpdate::ConnectorReferenceUpdate {
-                connector_mandate_id,
+                connector_mandate_ids: connector_mandate_id,
             } => Self {
-                connector_mandate_id,
+                connector_mandate_ids: connector_mandate_id,
                 ..Default::default()
             },
         }
diff --git a/crates/storage_models/src/schema.rs b/crates/storage_models/src/schema.rs
index 3f50d837322..3ab3793d8d5 100644
--- a/crates/storage_models/src/schema.rs
+++ b/crates/storage_models/src/schema.rs
@@ -221,6 +221,7 @@ diesel::table! {
         start_date -> Nullable<Timestamp>,
         end_date -> Nullable<Timestamp>,
         metadata -> Nullable<Jsonb>,
+        connector_mandate_ids -> Nullable<Jsonb>,
     }
 }
 
diff --git a/migrations/2023-04-20-073704_allow_multiple_mandate_ids/down.sql b/migrations/2023-04-20-073704_allow_multiple_mandate_ids/down.sql
new file mode 100644
index 00000000000..dbc01b39a3c
--- /dev/null
+++ b/migrations/2023-04-20-073704_allow_multiple_mandate_ids/down.sql
@@ -0,0 +1 @@
+ALTER TABLE mandate DROP COLUMN connector_mandate_ids;
\ No newline at end of file
diff --git a/migrations/2023-04-20-073704_allow_multiple_mandate_ids/up.sql b/migrations/2023-04-20-073704_allow_multiple_mandate_ids/up.sql
new file mode 100644
index 00000000000..aded1fcb5bd
--- /dev/null
+++ b/migrations/2023-04-20-073704_allow_multiple_mandate_ids/up.sql
@@ -0,0 +1,6 @@
+ALTER TABLE mandate
+    ADD COLUMN connector_mandate_ids jsonb;
+UPDATE mandate SET connector_mandate_ids = jsonb_build_object(
+            'mandate_id', connector_mandate_id,
+            'payment_method_id', NULL
+        );
\ No newline at end of file

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

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

## Description
<!-- Describe your changes in detail -->
Closes #940  

### 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 its an obvious bug or documentation fix
that will have little conversation).
-->
1. Recurring payments for Applepay and Googlepay in stripe.
2. Refactor tokenization and customer handling part in authorize_flow to a common place so that it can be reused in verify_flow also
3. Make tokenization for payment_method_type configurable to allow granular filtering for tokenization. 
Eg: if you want to ignore tokenization for google_pay you can do like 
 ```payment_method_type = { type = "disable_only", list = "google_pay" }```


## 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)?
-->
<img width="1160" alt="Screen Shot 2023-05-04 at 2 54 18 AM" src="https://user-images.githubusercontent.com/20727598/236053199-e91a6c42-828a-4579-867e-84c9f7ad0ec4.png">



## 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 submitted code
- [X] I added unit tests for my changes where possible
