use hyperswitch_domain_models::mandates;
pub mod access_token;
#[cfg(feature = "v2")]
mod backward_compat;
#[cfg(feature = "v1")]
pub mod batch_retrieve;
pub mod cards;
#[cfg(feature = "v1")]
pub mod client;
pub mod migration;
pub mod network_tokenization;
pub mod surcharge_decision_configs;
#[cfg(feature = "v1")]
pub mod tokenize;
pub mod transformers;
pub mod utils;
mod validator;
pub mod vault;
use std::borrow::Cow;
#[cfg(feature = "v1")]
use std::collections::HashSet;
#[cfg(feature = "v2")]
use std::str::FromStr;
#[cfg(feature = "v2")]
use std::sync::Arc;

#[cfg(feature = "v2")]
pub use api_models::enums as api_enums;
pub use api_models::enums::Connector;
use api_models::payment_methods;
#[cfg(feature = "payouts")]
pub use api_models::{enums::PayoutConnectors, payouts as payout_types};
#[cfg(feature = "v2")]
use async_trait::async_trait;
#[cfg(feature = "v1")]
use common_utils::{consts::DEFAULT_LOCALE, ext_traits::OptionExt};
#[cfg(feature = "v2")]
use common_utils::{
    crypto::{EncodeMessage, Encryptable, GcmAes256},
    encryption::Encryption,
    errors::CustomResult,
    ext_traits::{AsyncExt, BytesExt, StringExt, ValueExt},
    fp_utils::when,
    generate_id, types as util_types,
};
use common_utils::{ext_traits::Encode, id_type};
use diesel_models::{
    enums, GenericLinkNew, PaymentMethodCollectLink, PaymentMethodCollectLinkData,
};
use error_stack::{report, ResultExt};
#[cfg(feature = "v2")]
use futures::TryStreamExt;
#[cfg(feature = "v1")]
use hyperswitch_domain_models::api::{GenericLinks, GenericLinksData};
#[cfg(feature = "v2")]
use hyperswitch_domain_models::behaviour::Conversion;
#[cfg(feature = "v2")]
use hyperswitch_domain_models::payment_methods::PaymentMethodUpdate as DomainPaymentMethodUpdate;
use hyperswitch_domain_models::{
    payment_method_data::BankDebitData,
    payments::{payment_attempt::PaymentAttempt, PaymentIntent, VaultData},
    platform::ProviderMerchantId,
    router_data_v2::flow_common_types::VaultConnectorFlowData,
    router_flow_types::ExternalVaultInsertFlow,
    types::VaultRouterData,
};
use hyperswitch_interfaces::connector_integration_interface::RouterDataConversion;
#[cfg(feature = "v2")]
use hyperswitch_masking::ExposeInterface;
use hyperswitch_masking::{PeekInterface, Secret};
use router_env::{instrument, tracing};
use time::Duration;

#[cfg(feature = "v2")]
use super::payments::tokenization;
use super::{
    configs::dimension_state,
    errors::{RouterResponse, StorageErrorExt},
    pm_auth,
};
#[cfg(feature = "v2")]
use crate::{
    configs::settings,
    core::{
        payment_methods::{transformers as pm_transforms, utils as payment_method_utils},
        tokenization as tokenization_core,
    },
    headers,
    routes::{self, payment_methods as pm_routes},
    services::encryption,
    types::{
        api::{PaymentMethodCreateExt, PaymentMethodSessionExt},
        domain::types as domain_types,
        storage::{ephemeral_key, PaymentMethodListContext},
        transformers::{ForeignFrom, ForeignTryFrom},
        Tokenizable,
    },
    utils::ext_traits::OptionExt,
};
use crate::{
    consts,
    core::{
        errors::{ProcessTrackerError, RouterResult},
        payments::{self as payments_core, helpers as payment_helpers},
        utils as core_utils,
    },
    db::errors::ConnectorErrorExt,
    errors, logger,
    routes::{app::StorageInterface, SessionState},
    services,
    types::{
        self, api, domain, payment_methods as pm_types,
        storage::{self, enums as storage_enums},
    },
};

const PAYMENT_METHOD_STATUS_UPDATE_TASK: &str = "PAYMENT_METHOD_STATUS_UPDATE";
const PAYMENT_METHOD_STATUS_TAG: &str = "PAYMENT_METHOD_STATUS";
const PAYMENT_METHOD_MODULAR_FORWARD_COMPAT_TASK: &str = "PM_MOD_FORWARD_COMPAT";
const PAYMENT_METHOD_MODULAR_FORWARD_COMPAT_TAG: &str = "PM_MOD_FORWARD_COMPAT";
#[cfg(feature = "v2")]
const PAYMENT_METHOD_MODULAR_BACKWARD_COMPAT_TASK: &str = "PM_MOD_BACK_COMPAT";
#[cfg(feature = "v2")]
const PAYMENT_METHOD_MODULAR_BACKWARD_COMPAT_TAG: &str = "PM_MOD_BACK_COMPAT";
const PAYMENT_METHOD_MODULAR_COMPAT_PROCESS_TRACKER_ID_MAX_LENGTH: usize = 126;
const PAYMENT_METHOD_MODULAR_COMPAT_PROCESS_TRACKER_ID_SUFFIX_LENGTH: usize = 8;
#[cfg(feature = "v2")]
const PAYMENT_METHOD_REDACTED_FINGERPRINT_ID: &str = "FINGERPRINT_ID_REDACTED";
/// `process_tracker.id` is a `varchar(127)`; ids are built to stay within it.
const NETWORK_TOKENIZATION_PROCESS_TRACKER_ID_MAX_LENGTH: usize = 126;
const NETWORK_TOKENIZATION_PROCESS_TRACKER_ID_SUFFIX_LENGTH: usize = 8;
const NETWORK_TOKENIZATION_TASK: &str = "NETWORK_TOKENIZATION";
const NETWORK_TOKENIZATION_TAG: &str = "NETWORK_TOKENIZATION";

#[instrument(skip_all)]
pub async fn retrieve_payment_method_core(
    pm_data: &Option<domain::PaymentMethodData>,
    state: &SessionState,
    payment_intent: &PaymentIntent,
    payment_attempt: &PaymentAttempt,
    provider: &domain::Provider,
    business_profile: Option<&domain::Profile>,
) -> RouterResult<(Option<domain::PaymentMethodData>, Option<String>)> {
    match pm_data {
        pm_opt @ Some(pm @ domain::PaymentMethodData::Card(_)) => {
            let payment_token = payment_helpers::store_payment_method_data_in_vault(
                state,
                payment_attempt,
                payment_intent,
                enums::PaymentMethod::Card,
                pm,
                provider.get_key_store(),
                business_profile,
            )
            .await?;
            Ok((pm_opt.to_owned(), payment_token))
        }
        pm_opt @ Some(domain::PaymentMethodData::CardWithOptionalCVC(_)) => {
            Ok((pm_opt.to_owned(), None))
        }
        pm_opt @ Some(pm @ domain::PaymentMethodData::BankDebit(_)) => {
            let payment_token = payment_helpers::store_payment_method_data_in_vault(
                state,
                payment_attempt,
                payment_intent,
                enums::PaymentMethod::BankDebit,
                pm,
                provider.get_key_store(),
                business_profile,
            )
            .await?;

            Ok((pm_opt.to_owned(), payment_token))
        }
        pm @ Some(domain::PaymentMethodData::PayLater(_)) => Ok((pm.to_owned(), None)),
        pm @ Some(domain::PaymentMethodData::Crypto(_)) => Ok((pm.to_owned(), None)),
        pm @ Some(domain::PaymentMethodData::Upi(_)) => Ok((pm.to_owned(), None)),
        pm @ Some(domain::PaymentMethodData::Voucher(_)) => Ok((pm.to_owned(), None)),
        pm @ Some(domain::PaymentMethodData::Reward) => Ok((pm.to_owned(), None)),
        pm @ Some(domain::PaymentMethodData::RealTimePayment(_)) => Ok((pm.to_owned(), None)),
        pm @ Some(domain::PaymentMethodData::CardRedirect(_)) => Ok((pm.to_owned(), None)),
        pm @ Some(domain::PaymentMethodData::GiftCard(_)) => Ok((pm.to_owned(), None)),
        pm @ Some(domain::PaymentMethodData::OpenBanking(_)) => Ok((pm.to_owned(), None)),
        pm @ Some(domain::PaymentMethodData::MobilePayment(_)) => Ok((pm.to_owned(), None)),
        pm @ Some(domain::PaymentMethodData::NetworkToken(_)) => Ok((pm.to_owned(), None)),
        pm_opt @ Some(pm @ domain::PaymentMethodData::BankTransfer(_)) => {
            let payment_token = payment_helpers::store_payment_method_data_in_vault(
                state,
                payment_attempt,
                payment_intent,
                enums::PaymentMethod::BankTransfer,
                pm,
                provider.get_key_store(),
                business_profile,
            )
            .await?;

            Ok((pm_opt.to_owned(), payment_token))
        }
        pm_opt @ Some(pm @ domain::PaymentMethodData::Wallet(_)) => {
            let payment_token = payment_helpers::store_payment_method_data_in_vault(
                state,
                payment_attempt,
                payment_intent,
                enums::PaymentMethod::Wallet,
                pm,
                provider.get_key_store(),
                business_profile,
            )
            .await?;

            Ok((pm_opt.to_owned(), payment_token))
        }
        pm_opt @ Some(pm @ domain::PaymentMethodData::BankRedirect(_)) => {
            let payment_token = payment_helpers::store_payment_method_data_in_vault(
                state,
                payment_attempt,
                payment_intent,
                enums::PaymentMethod::BankRedirect,
                pm,
                provider.get_key_store(),
                business_profile,
            )
            .await?;

            Ok((pm_opt.to_owned(), payment_token))
        }
        _ => Ok((None, None)),
    }
}

pub async fn initiate_pm_collect_link(
    state: SessionState,
    platform: domain::Platform,
    req: payment_methods::PaymentMethodCollectLinkRequest,
) -> RouterResponse<payment_methods::PaymentMethodCollectLinkResponse> {
    // Validate request and initiate flow
    let pm_collect_link_data =
        validator::validate_request_and_initiate_payment_method_collect_link(
            &state, &platform, &req,
        )
        .await?;

    // Create DB entry
    let pm_collect_link = create_pm_collect_db_entry(
        &state,
        &platform,
        &pm_collect_link_data,
        req.return_url.clone(),
    )
    .await?;
    let customer_id = id_type::CustomerId::try_from(Cow::from(pm_collect_link.primary_reference))
        .change_context(errors::ApiErrorResponse::InvalidDataValue {
        field_name: "customer_id",
    })?;

    // Return response
    let url = pm_collect_link.url.peek();
    let response = payment_methods::PaymentMethodCollectLinkResponse {
        pm_collect_link_id: pm_collect_link.link_id,
        customer_id,
        expiry: pm_collect_link.expiry,
        link: url::Url::parse(url)
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable_lazy(|| {
                format!("Failed to parse the payment method collect link - {url}")
            })?
            .into(),
        return_url: pm_collect_link.return_url,
        ui_config: pm_collect_link.link_data.ui_config,
        enabled_payment_methods: pm_collect_link.link_data.enabled_payment_methods,
    };
    Ok(services::ApplicationResponse::Json(response))
}

pub async fn create_pm_collect_db_entry(
    state: &SessionState,
    platform: &domain::Platform,
    pm_collect_link_data: &PaymentMethodCollectLinkData,
    return_url: Option<String>,
) -> RouterResult<PaymentMethodCollectLink> {
    let db: &dyn StorageInterface = &*state.store;

    let link_data = serde_json::to_value(pm_collect_link_data)
        .map_err(|_| report!(errors::ApiErrorResponse::InternalServerError))
        .attach_printable("Failed to convert PaymentMethodCollectLinkData to Value")?;

    let pm_collect_link = GenericLinkNew {
        link_id: pm_collect_link_data.pm_collect_link_id.to_string(),
        primary_reference: pm_collect_link_data
            .customer_id
            .get_string_repr()
            .to_string(),
        merchant_id: platform.get_processor().get_account().get_id().to_owned(),
        link_type: common_enums::GenericLinkType::PaymentMethodCollect,
        link_data,
        url: pm_collect_link_data.link.clone(),
        return_url,
        expiry: common_utils::date_time::now()
            + Duration::seconds(pm_collect_link_data.session_expiry.into()),
        ..Default::default()
    };

    db.insert_pm_collect_link(pm_collect_link)
        .await
        .to_duplicate_response(errors::ApiErrorResponse::GenericDuplicateError {
            message: "payment method collect link already exists".to_string(),
        })
}

#[cfg(feature = "v2")]
pub async fn render_pm_collect_link(
    _state: SessionState,
    _platform: domain::Platform,
    _req: payment_methods::PaymentMethodCollectLinkRenderRequest,
) -> RouterResponse<services::GenericLinkFormData> {
    todo!()
}

#[cfg(feature = "v1")]
pub async fn render_pm_collect_link(
    state: SessionState,
    provider: domain::Provider,
    req: payment_methods::PaymentMethodCollectLinkRenderRequest,
) -> RouterResponse<services::GenericLinkFormData> {
    let db: &dyn StorageInterface = &*state.store;

    // Fetch pm collect link
    let pm_collect_link = db
        .find_pm_collect_link_by_link_id(&req.pm_collect_link_id)
        .await
        .to_not_found_response(errors::ApiErrorResponse::GenericNotFoundError {
            message: "payment method collect link not found".to_string(),
        })?;

    // Check status and return form data accordingly
    let has_expired = common_utils::date_time::now() > pm_collect_link.expiry;
    let status = pm_collect_link.link_status;
    let link_data = pm_collect_link.link_data;
    let default_config = &state.conf.generic_link.payment_method_collect;
    let default_ui_config = default_config.ui_config.clone();
    let ui_config_data = common_utils::link_utils::GenericLinkUiConfigFormData {
        merchant_name: link_data
            .ui_config
            .merchant_name
            .unwrap_or(default_ui_config.merchant_name),
        logo: link_data.ui_config.logo.unwrap_or(default_ui_config.logo),
        theme: link_data
            .ui_config
            .theme
            .clone()
            .unwrap_or(default_ui_config.theme.clone()),
    };
    match status {
        common_utils::link_utils::PaymentMethodCollectStatus::Initiated => {
            // if expired, send back expired status page
            if has_expired {
                let expired_link_data = services::GenericExpiredLinkData {
                    title: "Payment collect link has expired".to_string(),
                    message: "This payment collect link has expired.".to_string(),
                    theme: link_data.ui_config.theme.unwrap_or(default_ui_config.theme),
                };
                Ok(services::ApplicationResponse::GenericLinkForm(Box::new(
                    GenericLinks {
                        allowed_domains: HashSet::from([]),
                        data: GenericLinksData::ExpiredLink(expired_link_data),
                        locale: DEFAULT_LOCALE.to_string(),
                    },
                )))

            // else, send back form link
            } else {
                let customer_id = id_type::CustomerId::try_from(Cow::from(
                    pm_collect_link.primary_reference.clone(),
                ))
                .change_context(errors::ApiErrorResponse::InvalidDataValue {
                    field_name: "customer_id",
                })?;
                // Fetch customer

                let customer = db
                    .find_customer_by_customer_id_merchant_id(
                        &customer_id,
                        &req.merchant_id,
                        provider.get_key_store(),
                        provider.get_account().storage_scheme,
                    )
                    .await
                    .change_context(errors::ApiErrorResponse::InvalidRequestData {
                        message: format!(
                            "Customer [{}] not found for link_id - {}",
                            pm_collect_link.primary_reference, pm_collect_link.link_id
                        ),
                    })
                    .attach_printable(format!(
                        "customer [{}] not found",
                        pm_collect_link.primary_reference
                    ))?;

                let js_data = payment_methods::PaymentMethodCollectLinkDetails {
                    publishable_key: Secret::new(provider.get_account().clone().publishable_key),
                    client_secret: link_data.client_secret.clone(),
                    pm_collect_link_id: pm_collect_link.link_id,
                    customer_id: customer.get_id().clone(),
                    session_expiry: pm_collect_link.expiry,
                    return_url: pm_collect_link.return_url,
                    ui_config: ui_config_data,
                    enabled_payment_methods: link_data.enabled_payment_methods,
                };

                let serialized_css_content = String::new();

                let serialized_js_content = format!(
                    "window.__PM_COLLECT_DETAILS = {}",
                    js_data
                        .encode_to_string_of_json()
                        .change_context(errors::ApiErrorResponse::InternalServerError)
                        .attach_printable("Failed to serialize PaymentMethodCollectLinkDetails")?
                );

                let generic_form_data = services::GenericLinkFormData {
                    js_data: serialized_js_content,
                    css_data: serialized_css_content,
                    sdk_url: default_config.sdk_url.clone(),
                    html_meta_tags: String::new(),
                };
                Ok(services::ApplicationResponse::GenericLinkForm(Box::new(
                    GenericLinks {
                        allowed_domains: HashSet::from([]),
                        data: GenericLinksData::PaymentMethodCollect(generic_form_data),
                        locale: DEFAULT_LOCALE.to_string(),
                    },
                )))
            }
        }

        // Send back status page
        status => {
            let js_data = payment_methods::PaymentMethodCollectLinkStatusDetails {
                pm_collect_link_id: pm_collect_link.link_id,
                customer_id: link_data.customer_id,
                session_expiry: pm_collect_link.expiry,
                return_url: pm_collect_link
                    .return_url
                    .as_ref()
                    .map(|url| url::Url::parse(url))
                    .transpose()
                    .change_context(errors::ApiErrorResponse::InternalServerError)
                    .attach_printable(
                        "Failed to parse return URL for payment method collect's status link",
                    )?,
                ui_config: ui_config_data,
                status,
            };

            let serialized_css_content = String::new();

            let serialized_js_content = format!(
                "window.__PM_COLLECT_DETAILS = {}",
                js_data
                    .encode_to_string_of_json()
                    .change_context(errors::ApiErrorResponse::InternalServerError)
                    .attach_printable(
                        "Failed to serialize PaymentMethodCollectLinkStatusDetails"
                    )?
            );

            let generic_status_data = services::GenericLinkStatusData {
                js_data: serialized_js_content,
                css_data: serialized_css_content,
            };
            Ok(services::ApplicationResponse::GenericLinkForm(Box::new(
                GenericLinks {
                    allowed_domains: HashSet::from([]),
                    data: GenericLinksData::PaymentMethodCollectStatus(generic_status_data),
                    locale: DEFAULT_LOCALE.to_string(),
                },
            )))
        }
    }
}

fn generate_task_id_for_payment_method_status_update_workflow(
    key_id: &str,
    runner: storage::ProcessTrackerRunner,
    task: &str,
) -> String {
    format!("{runner}_{task}_{key_id}")
}

fn generate_task_id_for_payment_method_modular_compat_workflow(
    key_id: &str,
    runner: storage::ProcessTrackerRunner,
    task: &str,
) -> String {
    let suffix = common_utils::generate_id_with_len(
        PAYMENT_METHOD_MODULAR_COMPAT_PROCESS_TRACKER_ID_SUFFIX_LENGTH,
    );
    let prefix = format!("{runner}_{task}_");
    let key_id_len = PAYMENT_METHOD_MODULAR_COMPAT_PROCESS_TRACKER_ID_MAX_LENGTH
        .saturating_sub(prefix.len() + suffix.len() + 1);
    let key_id = &key_id[..key_id.len().min(key_id_len)];
    format!("{prefix}{key_id}_{suffix}")
}

/// Builds the `process_tracker` id for a `NetworkTokenizationWorkflow` task.
///
/// The random suffix keeps the id unique so that a payment method whose earlier task already
/// reached a terminal state can be scheduled again — finished rows are retained, so a
/// deterministic id would make every later attempt a silent no-op. The payment method id is
/// truncated to keep the result within the `varchar(127)` `process_tracker.id` column.
fn generate_task_id_for_network_tokenization_workflow(
    payment_method_id: &str,
    runner: storage::ProcessTrackerRunner,
    task: &str,
) -> String {
    let suffix =
        common_utils::generate_id_with_len(NETWORK_TOKENIZATION_PROCESS_TRACKER_ID_SUFFIX_LENGTH);
    let prefix = format!("{runner}_{task}_");
    let payment_method_id_len = NETWORK_TOKENIZATION_PROCESS_TRACKER_ID_MAX_LENGTH
        .saturating_sub(prefix.len() + suffix.len() + 1);
    let payment_method_id =
        &payment_method_id[..payment_method_id.len().min(payment_method_id_len)];
    format!("{prefix}{payment_method_id}_{suffix}")
}

fn payment_method_compat_modifier(payment_method: &domain::PaymentMethod) -> Option<String> {
    payment_method
        .last_modified_by
        .as_ref()
        .or(payment_method.created_by.as_ref())
        .map(ToString::to_string)
}

#[cfg(feature = "v1")]
pub async fn payment_method_modular_forward_compat_action(
    state: &SessionState,
    merchant_id: &id_type::MerchantId,
    organization_id: &id_type::OrganizationId,
    customer_id: Option<&id_type::CustomerId>,
) -> Option<domain::PaymentMethodCompatAction> {
    let dimensions = dimension_state::Dimensions::new()
        .with_provider_merchant_id(ProviderMerchantId::new(merchant_id.clone()))
        .with_organization_id(organization_id.clone());
    let should_schedule_modular_forward_compat =
        utils::get_should_schedule_modular_forward_compat(state, &dimensions, customer_id).await;

    let organization_id = organization_id.clone();
    should_schedule_modular_forward_compat.then(|| {
        let state = state.clone();
        domain::PaymentMethodCompatAction::new(move |payment_method| {
            let state = state.clone();
            let organization_id = organization_id.clone();
            Box::pin(async move {
                let res = add_payment_method_modular_forward_compat_task(
                    &*state.store,
                    payment_method,
                    &payment_method.merchant_id,
                    organization_id,
                    state.conf.application_source,
                    payment_method_compat_modifier(payment_method),
                )
                .await;

                if let Err(err) = res {
                    logger::error!(
                        ?err,
                        payment_method_id=%payment_method.get_id(),
                        merchant_id=%payment_method.merchant_id.get_string_repr(),
                        "Failed to schedule modular forward compatibility PT after payment method DB write"
                    );
                }
            })
        })
    })
}

#[cfg(feature = "v2")]
pub fn payment_method_modular_backward_compat_action(
    state: &SessionState,
    organization_id: &id_type::OrganizationId,
) -> domain::PaymentMethodCompatAction {
    let state = state.clone();
    let organization_id = organization_id.clone();
    domain::PaymentMethodCompatAction::new(move |payment_method| {
        let state = state.clone();
        let organization_id = organization_id.clone();
        Box::pin(async move {
            backward_compat::trigger_payment_method_modular_backward_compat(
                &state,
                payment_method,
                &organization_id,
                payment_method_compat_modifier(payment_method),
            )
            .await;
        })
    })
}

#[cfg(feature = "v1")]
pub async fn payment_method_modular_compat_action(
    state: &SessionState,
    merchant_id: &id_type::MerchantId,
    organization_id: &id_type::OrganizationId,
    customer_id: Option<&id_type::CustomerId>,
) -> Option<domain::PaymentMethodCompatAction> {
    payment_method_modular_forward_compat_action(state, merchant_id, organization_id, customer_id)
        .await
}

#[cfg(feature = "v2")]
pub async fn payment_method_modular_compat_action(
    state: &SessionState,
    _merchant_id: &id_type::MerchantId,
    organization_id: &id_type::OrganizationId,
    _customer_id: Option<&id_type::GlobalCustomerId>,
) -> Option<domain::PaymentMethodCompatAction> {
    Some(payment_method_modular_backward_compat_action(
        state,
        organization_id,
    ))
}

#[cfg(feature = "v1")]
pub async fn add_payment_method_status_update_task(
    db: &dyn StorageInterface,
    payment_method: &domain::PaymentMethod,
    prev_status: enums::PaymentMethodStatus,
    curr_status: enums::PaymentMethodStatus,
    merchant_id: &id_type::MerchantId,
    application_source: common_enums::ApplicationSource,
    initiator: Option<&hyperswitch_domain_models::platform::Initiator>,
) -> Result<(), ProcessTrackerError> {
    let created_at = payment_method.created_at;
    let schedule_time =
        created_at.saturating_add(Duration::seconds(consts::DEFAULT_SESSION_EXPIRY));

    let tracking_data = storage::PaymentMethodStatusTrackingData {
        payment_method_id: payment_method.get_id().clone(),
        prev_status,
        curr_status,
        merchant_id: merchant_id.to_owned(),
        last_modified_by: initiator
            .and_then(|initiator| initiator.to_created_by())
            .map(|last_modified_by| last_modified_by.to_string()),
    };

    let runner = storage::ProcessTrackerRunner::PaymentMethodStatusUpdateWorkflow;
    let task = PAYMENT_METHOD_STATUS_UPDATE_TASK;
    let tag = [PAYMENT_METHOD_STATUS_TAG];

    let process_tracker_id = generate_task_id_for_payment_method_status_update_workflow(
        payment_method.get_id().as_str(),
        runner,
        task,
    );
    let process_tracker_entry = storage::ProcessTrackerNew::new(
        process_tracker_id,
        task,
        runner,
        tag,
        tracking_data,
        None,
        schedule_time,
        common_types::consts::API_VERSION,
        application_source,
    )
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to construct PAYMENT_METHOD_STATUS_UPDATE process tracker task")?;

    db
        .insert_process(process_tracker_entry)
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable_lazy(|| {
            format!(
                "Failed while inserting PAYMENT_METHOD_STATUS_UPDATE reminder to process_tracker for payment_method_id: {}",
                payment_method.get_id().clone()
            )
        })?;

    Ok(())
}

#[cfg(feature = "v1")]
pub async fn add_payment_method_modular_forward_compat_task(
    db: &dyn StorageInterface,
    payment_method: &domain::PaymentMethod,
    merchant_id: &id_type::MerchantId,
    organization_id: id_type::OrganizationId,
    application_source: common_enums::ApplicationSource,
    last_modified_by: Option<String>,
) -> Result<(), ProcessTrackerError> {
    let tracking_data = storage::PaymentMethodModularCompatTrackingData {
        payment_method_id: payment_method.payment_method_id.clone(),
        merchant_id: merchant_id.to_owned(),
        organization_id,
        last_modified_by,
    };

    let runner = storage::ProcessTrackerRunner::PaymentMethodModularForwardCompatWorkflow;
    let task = PAYMENT_METHOD_MODULAR_FORWARD_COMPAT_TASK;
    let tag = [PAYMENT_METHOD_MODULAR_FORWARD_COMPAT_TAG];
    let process_tracker_id = generate_task_id_for_payment_method_modular_compat_workflow(
        payment_method.payment_method_id.as_str(),
        runner,
        task,
    );

    let process_tracker_entry = storage::ProcessTrackerNew::new(
        process_tracker_id,
        task,
        runner,
        tag,
        tracking_data,
        None,
        common_utils::date_time::now(),
        common_enums::ApiVersion::V1,
        application_source,
    )
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable(
        "Failed to construct PAYMENT_METHOD_MODULAR_FORWARD_COMPAT process tracker task",
    )?;

    db.insert_process(process_tracker_entry)
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable_lazy(|| {
            format!(
                "Failed while inserting PAYMENT_METHOD_MODULAR_FORWARD_COMPAT task to process_tracker for payment_method_id: {}",
                payment_method.payment_method_id
            )
        })?;

    Ok(())
}

#[cfg(feature = "v1")]
pub async fn add_network_tokenization_task(
    db: &dyn StorageInterface,
    tracking_data: storage::NetworkTokenizationTrackingData,
    application_source: common_enums::ApplicationSource,
) -> Result<(), ProcessTrackerError> {
    let payment_method_id = tracking_data.payment_method_id.clone();

    let runner = storage::ProcessTrackerRunner::NetworkTokenizationWorkflow;
    let task = NETWORK_TOKENIZATION_TASK;
    let tag = [NETWORK_TOKENIZATION_TAG];
    // The task id carries a unique suffix so that a payment method whose earlier task already
    // finished (completed, retries exceeded or skipped) can be scheduled again; runs against an
    // already-tokenized payment method finish as a no-op in the workflow itself.
    let process_tracker_id =
        generate_task_id_for_network_tokenization_workflow(&payment_method_id, runner, task);

    let process_tracker_entry = storage::ProcessTrackerNew::new(
        process_tracker_id,
        task,
        runner,
        tag,
        tracking_data,
        None,
        common_utils::date_time::now(),
        common_enums::ApiVersion::V1,
        application_source,
    )
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to construct NETWORK_TOKENIZATION process tracker task")?;

    db.insert_process(process_tracker_entry)
        .await
        .map(|_| ())
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable_lazy(|| {
            format!(
                "Failed while inserting NETWORK_TOKENIZATION task to process_tracker for payment_method_id: {payment_method_id}"
            )
        })?;

    Ok(())
}

#[cfg(feature = "v2")]
pub async fn add_network_tokenization_task(
    db: &dyn StorageInterface,
    tracking_data: storage::NetworkTokenizationTrackingData,
    application_source: common_enums::ApplicationSource,
) -> Result<(), ProcessTrackerError> {
    let payment_method_id = tracking_data.payment_method_id.get_string_repr().to_owned();

    let runner = storage::ProcessTrackerRunner::NetworkTokenizationWorkflow;
    let task = NETWORK_TOKENIZATION_TASK;
    let tag = [NETWORK_TOKENIZATION_TAG];
    // The task id carries a unique suffix so that a payment method whose earlier task already
    // finished (completed, retries exceeded or skipped) can be scheduled again; runs against an
    // already-tokenized payment method finish as a no-op in the workflow itself.
    let process_tracker_id =
        generate_task_id_for_network_tokenization_workflow(&payment_method_id, runner, task);

    let process_tracker_entry = storage::ProcessTrackerNew::new(
        process_tracker_id,
        task,
        runner,
        tag,
        tracking_data,
        None,
        common_utils::date_time::now(),
        common_enums::ApiVersion::V2,
        application_source,
    )
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to construct NETWORK_TOKENIZATION process tracker task")?;

    db.insert_process(process_tracker_entry)
        .await
        .map(|_| ())
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable_lazy(|| {
            format!(
                "Failed while inserting NETWORK_TOKENIZATION task to process_tracker for payment_method_id: {payment_method_id}"
            )
        })?;

    Ok(())
}

/// Schedules the async `NetworkTokenizationWorkflow` task for a payment method that was saved
/// without a network token.
///
/// Scheduling is best effort: the payment method create/update it belongs to has already been
/// committed by the time this runs, so a failure to enqueue is logged rather than propagated to
/// the caller.
#[cfg(feature = "v2")]
async fn schedule_network_tokenization_task(
    state: &SessionState,
    platform: &domain::Platform,
    profile: &domain::Profile,
    payment_method_id: &id_type::GlobalPaymentMethodId,
    customer_id: id_type::GlobalCustomerId,
) {
    let tracking_data = storage::NetworkTokenizationTrackingData {
        payment_method_id: payment_method_id.clone(),
        merchant_id: platform.get_provider().get_account().get_id().clone(),
        profile_id: profile.get_id().clone(),
        customer_id,
    };

    match add_network_tokenization_task(&*state.store, tracking_data, state.conf.application_source)
        .await
    {
        Ok(()) => logger::info!(
            payment_method_id=%payment_method_id.get_string_repr(),
            "Scheduled NetworkTokenizationWorkflow process tracker task"
        ),
        Err(err) => logger::error!(
            payment_method_id=%payment_method_id.get_string_repr(),
            ?err,
            "Failed to schedule NetworkTokenizationWorkflow process tracker task"
        ),
    }
}

/// Whether the async `NetworkTokenizationWorkflow` task should be scheduled for a payment method
/// that is being created or updated.
///
/// The network token columns on the payment method are expected to be empty at this point — the
/// token is generated by the scheduled job, not inline — so the profile-level setting, the
/// per-request opt-in and the payment method being a card are considered here.
#[cfg(feature = "v2")]
fn should_schedule_network_tokenization(
    profile: &domain::Profile,
    payment_method_type: common_enums::PaymentMethod,
    network_tokenization: Option<&common_types::payment_methods::NetworkTokenization>,
) -> bool {
    profile.is_network_tokenization_enabled
        && payment_method_type == common_enums::PaymentMethod::Card
        && network_tokenization
            .map(|nt| matches!(nt.enable, common_enums::NetworkTokenizationToggle::Enable))
            .unwrap_or(false)
}

#[cfg(feature = "v2")]
pub async fn add_payment_method_modular_backward_compat_task(
    db: &dyn StorageInterface,
    payment_method: &domain::PaymentMethod,
    merchant_id: &id_type::MerchantId,
    organization_id: id_type::OrganizationId,
    application_source: common_enums::ApplicationSource,
    last_modified_by: Option<String>,
) -> Result<(), ProcessTrackerError> {
    let payment_method_id = payment_method.get_id().get_string_repr().to_owned();

    let tracking_data = storage::PaymentMethodModularCompatTrackingData {
        payment_method_id: payment_method_id.clone(),
        merchant_id: merchant_id.to_owned(),
        organization_id,
        last_modified_by,
    };

    let runner = storage::ProcessTrackerRunner::PaymentMethodModularBackwardCompatWorkflow;
    let task = PAYMENT_METHOD_MODULAR_BACKWARD_COMPAT_TASK;
    let tag = [PAYMENT_METHOD_MODULAR_BACKWARD_COMPAT_TAG];
    let process_tracker_id = generate_task_id_for_payment_method_modular_compat_workflow(
        &payment_method_id,
        runner,
        task,
    );

    let process_tracker_entry = storage::ProcessTrackerNew::new(
        process_tracker_id,
        task,
        runner,
        tag,
        tracking_data,
        None,
        common_utils::date_time::now(),
        common_enums::ApiVersion::V1,
        application_source,
    )
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable(
        "Failed to construct PAYMENT_METHOD_MODULAR_BACKWARD_COMPAT process tracker task",
    )?;

    db.insert_process(process_tracker_entry)
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable_lazy(|| {
            format!(
                "Failed while inserting PAYMENT_METHOD_MODULAR_BACKWARD_COMPAT task to process_tracker for payment_method_id: {}",
                payment_method_id
            )
        })?;

    Ok(())
}

#[cfg(feature = "v2")]
#[allow(clippy::too_many_arguments)]
#[instrument(skip_all)]
pub async fn retrieve_payment_method_with_token(
    _state: &SessionState,
    _merchant_key_store: &domain::MerchantKeyStore,
    _token_data: &storage::PaymentTokenData,
    _payment_intent: &PaymentIntent,
    _card_token_data: Option<&domain::CardToken>,
    _customer: &Option<domain::Customer>,
    _storage_scheme: common_enums::enums::MerchantStorageScheme,
    _mandate_id: Option<mandates::MandateIds>,
    _payment_method_info: Option<domain::PaymentMethod>,
    _business_profile: &domain::Profile,
) -> RouterResult<storage::PaymentMethodDataWithId> {
    todo!()
}

#[cfg(feature = "v1")]
#[instrument(skip_all)]
#[allow(clippy::too_many_arguments)]
pub async fn retrieve_payment_method_with_token(
    state: &SessionState,
    platform: &domain::Platform,
    token_data: &storage::PaymentTokenData,
    payment_intent: &PaymentIntent,
    payment_attempt: &PaymentAttempt,
    card_token_data: Option<&domain::CardToken>,
    storage_scheme: common_enums::enums::MerchantStorageScheme,
    mandate_id: Option<mandates::MandateIds>,
    payment_method_info: Option<domain::PaymentMethod>,
    business_profile: &domain::Profile,
    should_retry_with_pan: bool,
    vault_data: Option<&VaultData>,
) -> RouterResult<storage::PaymentMethodDataWithId> {
    let token = match token_data {
        storage::PaymentTokenData::TemporaryGeneric(generic_token) => {
            payment_helpers::retrieve_payment_method_with_temporary_token(
                state,
                &generic_token.token,
                payment_intent,
                payment_attempt,
                platform.get_provider().get_key_store(),
                card_token_data,
            )
            .await?
            .map(
                |(payment_method_data, payment_method)| storage::PaymentMethodDataWithId {
                    payment_method_data: Some(payment_method_data),
                    payment_method: Some(payment_method),
                    payment_method_id: None,
                },
            )
            .unwrap_or_default()
        }

        storage::PaymentTokenData::Temporary(generic_token) => {
            payment_helpers::retrieve_payment_method_with_temporary_token(
                state,
                &generic_token.token,
                payment_intent,
                payment_attempt,
                platform.get_provider().get_key_store(),
                card_token_data,
            )
            .await?
            .map(
                |(payment_method_data, payment_method)| storage::PaymentMethodDataWithId {
                    payment_method_data: Some(payment_method_data),
                    payment_method: Some(payment_method),
                    payment_method_id: None,
                },
            )
            .unwrap_or_default()
        }

        storage::PaymentTokenData::Permanent(card_token) => Box::pin(
            payment_helpers::retrieve_payment_method_data_with_permanent_token(
                state,
                card_token.locker_id.as_ref().unwrap_or(&card_token.token),
                card_token
                    .payment_method_id
                    .as_ref()
                    .unwrap_or(&card_token.token),
                payment_intent,
                card_token_data,
                platform,
                storage_scheme,
                mandate_id,
                payment_method_info
                    .get_required_value("PaymentMethod")
                    .change_context(errors::ApiErrorResponse::InternalServerError)
                    .attach_printable("PaymentMethod not found")?,
                business_profile,
                payment_attempt.connector.clone(),
                should_retry_with_pan,
                vault_data,
            ),
        )
        .await
        .map(|card| Some((card, enums::PaymentMethod::Card)))?
        .map(
            |(payment_method_data, payment_method)| storage::PaymentMethodDataWithId {
                payment_method_data: Some(payment_method_data),
                payment_method: Some(payment_method),
                payment_method_id: Some(
                    card_token
                        .payment_method_id
                        .as_ref()
                        .unwrap_or(&card_token.token)
                        .to_string(),
                ),
            },
        )
        .unwrap_or_default(),

        storage::PaymentTokenData::PermanentCard(card_token) => Box::pin(
            payment_helpers::retrieve_payment_method_data_with_permanent_token(
                state,
                card_token.locker_id.as_ref().unwrap_or(&card_token.token),
                card_token
                    .payment_method_id
                    .as_ref()
                    .unwrap_or(&card_token.token),
                payment_intent,
                card_token_data,
                platform,
                storage_scheme,
                mandate_id,
                payment_method_info
                    .get_required_value("PaymentMethod")
                    .change_context(errors::ApiErrorResponse::InternalServerError)
                    .attach_printable("PaymentMethod not found")?,
                business_profile,
                payment_attempt.connector.clone(),
                should_retry_with_pan,
                vault_data,
            ),
        )
        .await
        .map(|card| Some((card, enums::PaymentMethod::Card)))?
        .map(
            |(payment_method_data, payment_method)| storage::PaymentMethodDataWithId {
                payment_method_data: Some(payment_method_data),
                payment_method: Some(payment_method),
                payment_method_id: Some(
                    card_token
                        .payment_method_id
                        .as_ref()
                        .unwrap_or(&card_token.token)
                        .to_string(),
                ),
            },
        )
        .unwrap_or_default(),

        storage::PaymentTokenData::AuthBankDebit(auth_token) => {
            pm_auth::retrieve_payment_method_from_auth_service(
                state,
                platform.get_processor(),
                auth_token,
                payment_intent,
            )
            .await?
            .map(
                |(payment_method_data, payment_method)| storage::PaymentMethodDataWithId {
                    payment_method_data: Some(payment_method_data),
                    payment_method: Some(payment_method),
                    payment_method_id: None,
                },
            )
            .unwrap_or_default()
        }

        storage::PaymentTokenData::WalletToken(_) => storage::PaymentMethodDataWithId {
            payment_method: None,
            payment_method_data: None,
            payment_method_id: None,
        },
        storage::PaymentTokenData::BankDebit(bank_debit) => {
            let customer_id = payment_intent.customer_id.as_ref().ok_or(
                errors::ApiErrorResponse::MissingRequiredField {
                    field_name: "customer",
                },
            )?;

            let locker_id = bank_debit.locker_id.as_ref().ok_or(
                errors::ApiErrorResponse::MissingRequiredField {
                    field_name: "locker_id",
                },
            )?;

            let bank_debit_detail = cards::get_bank_debit_from_hs_locker(
                state,
                platform.get_provider(),
                customer_id,
                locker_id,
            )
            .await?;

            let (
                account_number,
                routing_number,
                vault_bank_account_holder_name,
                vault_bank_type,
                vault_bank_holder_type,
            ) = match bank_debit_detail {
                domain::BankDebitDetail::Ach {
                    account_number,
                    routing_number,
                    bank_account_holder_name,
                    bank_type,
                    bank_holder_type,
                } => (
                    account_number,
                    routing_number,
                    bank_account_holder_name,
                    bank_type,
                    bank_holder_type,
                ),
            };

            let payment_method_data = payment_method_info
                .get_required_value("PaymentMethod")
                .change_context(errors::ApiErrorResponse::InternalServerError)
                .attach_printable("PaymentMethod not found")?;

            let (bank_account_holder_name, bank_name, bank_type, bank_holder_type) =
                if let Some(domain::PaymentMethodsData::BankDebit(bank_debit_data)) =
                    payment_method_data.get_payment_methods_data()
                {
                    let domain::BankDebitDetailsPaymentMethod::AchBankDebit {
                        masked_account_number: _,
                        masked_routing_number: _,
                        bank_account_holder_name,
                        bank_name,
                        bank_type,
                        bank_holder_type,
                    } = bank_debit_data;
                    (
                        bank_account_holder_name.clone(),
                        bank_name,
                        bank_type,
                        bank_holder_type,
                    )
                } else {
                    return Err(report!(errors::ApiErrorResponse::InternalServerError)
                        .attach_printable("Payment method data is not bank debit"));
                };

            storage::PaymentMethodDataWithId {
                payment_method: Some(enums::PaymentMethod::BankDebit),
                payment_method_data: Some(domain::PaymentMethodData::BankDebit(
                    BankDebitData::AchBankDebit {
                        account_number,
                        routing_number,
                        bank_account_holder_name: vault_bank_account_holder_name
                            .or(bank_account_holder_name),
                        bank_name,
                        bank_type: vault_bank_type.or(bank_type),
                        bank_holder_type: vault_bank_holder_type.or(bank_holder_type),
                    },
                )),

                payment_method_id: Some(bank_debit.payment_method_id.clone()),
            }
        }
    };
    Ok(token)
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub(crate) fn get_payment_method_create_request(
    payment_method_data: &api_models::payments::PaymentMethodData,
    payment_method_type: storage_enums::PaymentMethod,
    payment_method_subtype: Option<storage_enums::PaymentMethodType>,
    customer_id: Option<id_type::GlobalCustomerId>,
    billing_address: Option<&api_models::payments::Address>,
    payment_method_session: Option<&domain::payment_methods::PaymentMethodSession>,
    storage_type: common_enums::StorageType,
) -> RouterResult<payment_methods::PaymentMethodCreate> {
    match payment_method_data {
        api_models::payments::PaymentMethodData::Card(card) => {
            payments_core::helpers::validate_card_expiry(
                &card.card_exp_month,
                &card.card_exp_year,
            )?;

            let card_detail = payment_methods::CardDetail {
                card_number: card.card_number.clone(),
                card_exp_month: card.card_exp_month.clone(),
                card_exp_year: card.card_exp_year.clone(),
                card_holder_name: card.card_holder_name.clone(),
                nick_name: card.nick_name.clone(),
                card_issuing_country: card
                    .card_issuing_country
                    .as_ref()
                    .map(|c| api_enums::CountryAlpha2::from_str(c))
                    .transpose()
                    .ok()
                    .flatten(),
                card_network: card.card_network.clone(),
                card_issuer: card.card_issuer.clone(),
                card_type: card
                    .card_type
                    .as_ref()
                    .map(|c| payment_methods::CardType::from_str(c))
                    .transpose()
                    .ok()
                    .flatten(),
                card_cvc: Some(card.card_cvc.clone()),
            };
            let payment_method_request = payment_methods::PaymentMethodCreate {
                payment_method_type,
                payment_method_subtype,
                metadata: None,
                customer_id,
                payment_method_data: payment_methods::PaymentMethodCreateData::Card(card_detail),
                billing: billing_address.map(ToOwned::to_owned),
                network_tokenization: payment_method_session
                    .and_then(|pm_session| pm_session.network_tokenization.clone()),
                storage_type,
            };
            Ok(payment_method_request)
        }
        api_models::payments::PaymentMethodData::BankDebit(bank_debit) => {
            let bank_debit_detail = match bank_debit {
                api_models::payments::BankDebitData::AchBankDebit {
                    account_number,
                    routing_number,
                    bank_account_holder_name,
                    bank_name,
                    bank_type,
                    bank_holder_type,
                    ..
                } => payment_methods::BankDebitDetail::Ach {
                    account_number: account_number.clone(),
                    routing_number: routing_number.clone(),
                    bank_account_holder_name: bank_account_holder_name.clone(),
                    bank_name: *bank_name,
                    bank_type: *bank_type,
                    bank_holder_type: *bank_holder_type,
                },
                _ => {
                    return Err(report!(errors::ApiErrorResponse::UnprocessableEntity {
                        message: "unsupported bank debit type in payment method session"
                            .to_string()
                    })
                    .attach_printable(
                        "Only ACH bank debit is supported in payment method sessions",
                    ))
                }
            };

            let payment_method_request = payment_methods::PaymentMethodCreate {
                payment_method_type,
                payment_method_subtype,
                metadata: None,
                customer_id,
                payment_method_data: payment_methods::PaymentMethodCreateData::BankDebit(
                    bank_debit_detail,
                ),
                billing: billing_address.map(ToOwned::to_owned),
                network_tokenization: None,
                storage_type,
            };
            Ok(payment_method_request)
        }
        _ => Err(report!(errors::ApiErrorResponse::UnprocessableEntity {
            message: "payment method type is not supported in payment method session".to_string()
        })
        .attach_printable("Payment method data is incorrect")),
    }
}

#[cfg(feature = "v1")]
#[instrument(skip_all)]
pub(crate) async fn get_payment_method_create_request(
    payment_method_data: Option<&domain::PaymentMethodData>,
    payment_method: Option<storage_enums::PaymentMethod>,
    payment_method_type: Option<storage_enums::PaymentMethodType>,
    customer_id: &Option<id_type::CustomerId>,
    billing_name: Option<Secret<String>>,
    payment_method_billing_address: Option<&hyperswitch_domain_models::address::Address>,
) -> RouterResult<payment_methods::PaymentMethodCreate> {
    match payment_method_data {
        Some(pm_data) => match payment_method {
            Some(payment_method) => {
                let default_payment_method_request = payment_methods::PaymentMethodCreate {
                    payment_method: Some(payment_method),
                    payment_method_type,
                    payment_method_issuer: None,
                    payment_method_issuer_code: None,
                    #[cfg(feature = "payouts")]
                    bank_transfer: None,
                    #[cfg(feature = "payouts")]
                    bank_transfer_data: None,
                    #[cfg(feature = "payouts")]
                    wallet: None,
                    card: None,
                    metadata: None,
                    customer_id: customer_id.clone(),
                    card_network: None,
                    client_secret: None,
                    payment_method_data: None,
                    billing: None,
                    connector_mandate_details: None,
                    network_transaction_id: None,
                };
                match pm_data {
                    domain::PaymentMethodData::Card(card) => {
                        let card_network = get_card_network_with_us_local_debit_network_override(
                            card.card_network.clone(),
                            card.co_badged_card_data.as_ref(),
                        );

                        let card_detail = payment_methods::CardDetail {
                            card_number: card.card_number.clone(),
                            card_exp_month: card.card_exp_month.clone(),
                            card_exp_year: card.card_exp_year.clone(),
                            card_holder_name: billing_name,
                            nick_name: card.nick_name.clone(),
                            card_issuing_country: card.card_issuing_country.clone(),
                            card_issuing_country_code: card.card_issuing_country_code.clone(),
                            card_network: card_network.clone(),
                            card_issuer: card.card_issuer.clone(),
                            card_type: card.card_type.clone(),
                            card_cvc: None, // DO NOT POPULATE CVC FOR ADDITIONAL PAYMENT METHOD DATA
                        };
                        let payment_method_request = payment_methods::PaymentMethodCreate {
                            payment_method: Some(payment_method),
                            payment_method_type,
                            payment_method_issuer: card.card_issuer.clone(),
                            payment_method_issuer_code: None,
                            #[cfg(feature = "payouts")]
                            bank_transfer: None,
                            #[cfg(feature = "payouts")]
                            bank_transfer_data: None,
                            #[cfg(feature = "payouts")]
                            wallet: None,
                            card: Some(card_detail),
                            metadata: None,
                            customer_id: customer_id.clone(),
                            card_network: card_network
                                .clone()
                                .as_ref()
                                .map(|card_network| card_network.to_string()),
                            client_secret: None,
                            payment_method_data: None,
                            //TODO: why are we using api model in router internally
                            billing: payment_method_billing_address.cloned().map(From::from),
                            connector_mandate_details: None,
                            network_transaction_id: None,
                        };
                        Ok(payment_method_request)
                    }
                    domain::PaymentMethodData::CardWithOptionalCVC(card) => {
                        let card_network = get_card_network_with_us_local_debit_network_override(
                            card.card_network.clone(),
                            card.co_badged_card_data.as_ref(),
                        );

                        let card_detail = payment_methods::CardDetail {
                            card_number: card.card_number.clone(),
                            card_exp_month: card.card_exp_month.clone(),
                            card_exp_year: card.card_exp_year.clone(),
                            card_holder_name: billing_name,
                            nick_name: card.nick_name.clone(),
                            card_issuing_country: card.card_issuing_country.clone(),
                            card_issuing_country_code: card.card_issuing_country_code.clone(),
                            card_network: card_network.clone(),
                            card_issuer: card.card_issuer.clone(),
                            card_type: card.card_type.clone(),
                            card_cvc: None, // DO NOT POPULATE CVC FOR ADDITIONAL PAYMENT METHOD DATA
                        };
                        let payment_method_request = payment_methods::PaymentMethodCreate {
                            payment_method: Some(payment_method),
                            payment_method_type,
                            payment_method_issuer: card.card_issuer.clone(),
                            payment_method_issuer_code: None,
                            #[cfg(feature = "payouts")]
                            bank_transfer: None,
                            #[cfg(feature = "payouts")]
                            bank_transfer_data: None,
                            #[cfg(feature = "payouts")]
                            wallet: None,
                            card: Some(card_detail),
                            metadata: None,
                            customer_id: customer_id.clone(),
                            card_network: card_network
                                .clone()
                                .as_ref()
                                .map(|card_network| card_network.to_string()),
                            client_secret: None,
                            payment_method_data: None,
                            //TODO: why are we using api model in router internally
                            billing: payment_method_billing_address.cloned().map(From::from),
                            connector_mandate_details: None,
                            network_transaction_id: None,
                        };
                        Ok(payment_method_request)
                    }
                    domain::PaymentMethodData::BankDebit(BankDebitData::AchBankDebit {
                        account_number,
                        routing_number,
                        bank_account_holder_name,
                        bank_name: _,
                        bank_type,
                        bank_holder_type,
                    }) => {
                        let payment_method_request = payment_methods::PaymentMethodCreate {
                            payment_method: Some(payment_method),
                            payment_method_type,
                            payment_method_issuer: None,
                            payment_method_issuer_code: None,
                            #[cfg(feature = "payouts")]
                            bank_transfer: None,
                            #[cfg(feature = "payouts")]
                            bank_transfer_data: None,
                            #[cfg(feature = "payouts")]
                            wallet: None,
                            card: None,
                            metadata: None,
                            customer_id: customer_id.clone(),
                            card_network: None,
                            client_secret: None,
                            payment_method_data: Some(
                                payment_methods::PaymentMethodCreateData::BankDebit(
                                    payment_methods::BankDebitDetail::Ach {
                                        account_number: account_number.to_owned(),
                                        routing_number: routing_number.to_owned(),
                                        bank_account_holder_name: bank_account_holder_name.clone(),
                                        bank_type: *bank_type,
                                        bank_holder_type: *bank_holder_type,
                                    },
                                ),
                            ),
                            billing: payment_method_billing_address.cloned().map(From::from),
                            connector_mandate_details: None,
                            network_transaction_id: None,
                        };
                        Ok(payment_method_request)
                    }
                    domain::PaymentMethodData::Wallet(domain::WalletData::ApplePay(
                        apple_pay_data,
                    )) => {
                        if let Some(decrypted_data) = apple_pay_data
                            .payment_data
                            .get_decrypted_apple_pay_payment_data_optional()
                        {
                            Ok(payment_methods::PaymentMethodCreate {
                                payment_method: Some(payment_method),
                                payment_method_type,
                                payment_method_issuer: None,
                                payment_method_issuer_code: None,
                                #[cfg(feature = "payouts")]
                                bank_transfer: None,
                                #[cfg(feature = "payouts")]
                                bank_transfer_data: None,
                                #[cfg(feature = "payouts")]
                                wallet: None,
                                card: None,
                                metadata: None,
                                customer_id: customer_id.clone(),
                                card_network: None,
                                client_secret: None,
                                payment_method_data: Some(
                                    payment_methods::PaymentMethodCreateData::Wallet(
                                        decrypted_data.clone().into(),
                                    ),
                                ),
                                billing: payment_method_billing_address.cloned().map(From::from),
                                connector_mandate_details: None,
                                network_transaction_id: None,
                            })
                        } else {
                            Ok(default_payment_method_request)
                        }
                    }
                    domain::PaymentMethodData::Wallet(domain::WalletData::GooglePay(
                        google_pay_data,
                    )) => {
                        if let Some(decrypted_data) = google_pay_data
                            .tokenization_data
                            .get_decrypted_google_pay_payment_data_optional()
                        {
                            Ok(payment_methods::PaymentMethodCreate {
                                payment_method: Some(payment_method),
                                payment_method_type,
                                payment_method_issuer: None,
                                payment_method_issuer_code: None,
                                #[cfg(feature = "payouts")]
                                bank_transfer: None,
                                #[cfg(feature = "payouts")]
                                bank_transfer_data: None,
                                #[cfg(feature = "payouts")]
                                wallet: None,
                                card: None,
                                metadata: None,
                                customer_id: customer_id.clone(),
                                card_network: None,
                                client_secret: None,
                                payment_method_data: Some(
                                    payment_methods::PaymentMethodCreateData::Wallet(
                                        decrypted_data.clone().into(),
                                    ),
                                ),
                                billing: payment_method_billing_address.cloned().map(From::from),
                                connector_mandate_details: None,
                                network_transaction_id: None,
                            })
                        } else {
                            Ok(default_payment_method_request)
                        }
                    }

                    _ => Ok(default_payment_method_request),
                }
            }
            None => Err(report!(errors::ApiErrorResponse::MissingRequiredField {
                field_name: "payment_method_type"
            })
            .attach_printable("PaymentMethodType Required")),
        },
        None => Err(report!(errors::ApiErrorResponse::MissingRequiredField {
            field_name: "payment_method_data"
        })
        .attach_printable("PaymentMethodData required Or Card is already saved")),
    }
}

/// Determines the appropriate card network to to be stored.
///
/// If the provided card network is a US local network, this function attempts to
/// override it with the first global network from the co-badged card data, if available.
/// Otherwise, it returns the original card network as-is.
///
fn get_card_network_with_us_local_debit_network_override(
    card_network: Option<common_enums::CardNetwork>,
    co_badged_card_data: Option<&payment_methods::CoBadgedCardData>,
) -> Option<common_enums::CardNetwork> {
    if let Some(true) = card_network
        .as_ref()
        .map(|network| network.is_us_local_network())
    {
        services::logger::debug!("Card network is a US local network, checking for global network in co-badged card data");
        let info: Option<api_models::open_router::CoBadgedCardNetworksInfo> = co_badged_card_data
            .and_then(|data| {
                data.co_badged_card_networks_info
                    .0
                    .iter()
                    .find(|info| info.network.is_signature_network())
                    .cloned()
            });
        info.map(|data| data.network)
    } else {
        card_network
    }
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn get_card_nt_eligibility(
    state: &SessionState,
    req: api::NetworkTokenEligibilityRequest,
    _platform: &domain::Platform,
    profile: &domain::Profile,
) -> RouterResponse<api::GetNetworkTokenEiligibilityResponse> {
    when(!profile.is_network_tokenization_enabled, || {
        Err(report!(errors::ApiErrorResponse::NotSupported {
            message: "Network tokenization is not enabled for this profile".to_string()
        }))
    })?;

    let response = network_tokenization::make_nt_eligibility_call(state, req)
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to fetch network token eligibility from tokenization service")?;

    Ok(services::ApplicationResponse::Json(
        api::GetNetworkTokenEiligibilityResponse {
            eligible_for_network_tokenization: response.tokenize_support,
        },
    ))
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn create_payment_method(
    state: &SessionState,
    request_state: &routes::app::ReqState,
    req: api::PaymentMethodCreate,
    platform: &domain::Platform,
    profile: &domain::Profile,
) -> RouterResponse<api::PaymentMethodResponse> {
    // payment_method is for internal use, can never be populated in response
    let (response, _payment_method) = Box::pin(create_payment_method_core(
        state,
        request_state,
        req,
        platform,
        profile,
    ))
    .await?;

    Ok(services::ApplicationResponse::Json(response))
}
#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn create_payment_method_core(
    state: &SessionState,
    _request_state: &routes::app::ReqState,
    req: api::PaymentMethodCreate,
    platform: &domain::Platform,
    profile: &domain::Profile,
) -> RouterResult<(api::PaymentMethodResponse, domain::PaymentMethod)> {
    match req.storage_type {
        common_enums::StorageType::Volatile => {
            create_volatile_payment_method_core(state, _request_state, req, platform, profile).await
        }
        common_enums::StorageType::Persistent => {
            create_persistent_payment_method_core(state, _request_state, req, platform, profile)
                .await
        }
    }
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn create_persistent_payment_method_core(
    state: &SessionState,
    _request_state: &routes::app::ReqState,
    req: api::PaymentMethodCreate,
    platform: &domain::Platform,
    profile: &domain::Profile,
) -> RouterResult<(api::PaymentMethodResponse, domain::PaymentMethod)> {
    req.validate()?;

    let db = &*state.store;
    let merchant_id = platform.get_provider().get_account().get_id();
    let customer_id = req
        .customer_id
        .to_owned()
        .get_required_value("customer_id")?;
    let key_manager_state = &(state).into();

    db.find_customer_by_global_id_merchant_id_without_encrypted(
        &customer_id,
        platform.get_provider().get_account().get_id(),
        platform.get_provider().get_account().storage_scheme,
    )
    .await
    .to_not_found_response(errors::ApiErrorResponse::CustomerNotFound)
    .attach_printable("Customer not found for the payment method")?;

    let payment_method_billing_address = req
        .billing
        .clone()
        .async_map(|billing| {
            core_utils::create_encrypted_data(
                key_manager_state,
                platform.get_provider().get_key_store(),
                billing,
                common_utils::type_name!(diesel_models::payment_method::PaymentMethod),
            )
        })
        .await
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to encrypt Payment method billing address")?
        .map(|encoded_address| {
            encoded_address.deserialize_inner_value(|value| value.parse_value("address"))
        })
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to parse Payment method billing address")?;

    let payment_method_id =
        id_type::GlobalPaymentMethodId::generate(&state.conf.cell_information.id);

    match &req.payment_method_data {
        api::PaymentMethodCreateData::Card(_) | api::PaymentMethodCreateData::BankDebit(_) => {
            Box::pin(create_or_fetch_payment_method_core(
                state,
                req,
                platform,
                profile,
                merchant_id,
                &customer_id,
                payment_method_id,
                payment_method_billing_address,
            ))
            .await
        }
        api::PaymentMethodCreateData::ProxyCard(_) => {
            create_payment_method_proxy_card_core(
                state,
                req,
                platform,
                profile,
                merchant_id,
                &customer_id,
                payment_method_id,
                payment_method_billing_address,
            )
            .await
        }
        api::PaymentMethodCreateData::Wallet(wallet_data) => {
            let additional_data = payment_methods::PaymentMethodsData::from(wallet_data.clone());
            create_payment_method_wallet_core(
                state,
                req,
                platform,
                profile,
                merchant_id,
                &customer_id,
                payment_method_id,
                payment_method_billing_address,
                additional_data,
            )
            .await
        }
        api::PaymentMethodCreateData::BankRedirect(_) => {
            create_payment_method_bank_redirect_core(
                state,
                req,
                platform,
                profile,
                merchant_id,
                &customer_id,
                payment_method_id,
                payment_method_billing_address,
            )
            .await
        }
    }
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn create_volatile_payment_method_core(
    state: &SessionState,
    _request_state: &routes::app::ReqState,
    req: api::PaymentMethodCreate,
    platform: &domain::Platform,
    profile: &domain::Profile,
) -> RouterResult<(api::PaymentMethodResponse, domain::PaymentMethod)> {
    req.validate()?;

    let db = &*state.store;
    let merchant_id = platform.get_provider().get_account().get_id();
    let customer_id = req.customer_id.to_owned();
    let key_manager_state = &(state).into();

    if let Some(ref customer_id) = customer_id {
        db.find_customer_by_global_id_merchant_id_without_encrypted(
            customer_id,
            platform.get_provider().get_account().get_id(),
            platform.get_provider().get_account().storage_scheme,
        )
        .await
        .to_not_found_response(errors::ApiErrorResponse::CustomerNotFound)
        .attach_printable("Customer not found for the payment method")?;
    }

    let payment_method_billing_address = req
        .billing
        .clone()
        .async_map(|billing| {
            core_utils::create_encrypted_data(
                key_manager_state,
                platform.get_provider().get_key_store(),
                billing,
                common_utils::type_name!(diesel_models::payment_method::PaymentMethod),
            )
        })
        .await
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to encrypt Payment method billing address")?
        .map(|encoded_address| {
            encoded_address.deserialize_inner_value(|value| value.parse_value("address"))
        })
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to parse Payment method billing address")?;

    let payment_method_id =
        id_type::GlobalPaymentMethodId::generate(&state.conf.cell_information.id);

    match &req.payment_method_data {
        api::PaymentMethodCreateData::Card(_) | api::PaymentMethodCreateData::BankDebit(_) => {
            let payment_method_type = req.payment_method_type.to_string();
            logger::info!("Creating volatile {} payment method", payment_method_type);
            Box::pin(create_generic_volatile_payment_method(
                state,
                req,
                platform,
                profile,
                merchant_id,
                &customer_id,
                payment_method_id,
                payment_method_billing_address,
            ))
            .await
        }
        api::PaymentMethodCreateData::ProxyCard(_) => {
            Err(report!(errors::ApiErrorResponse::UnprocessableEntity {
                message: "Proxy card payment method cannot be created as volatile".to_string()
            }))
        }
        api::PaymentMethodCreateData::Wallet(_) => {
            Err(report!(errors::ApiErrorResponse::UnprocessableEntity {
                message: "Wallet payment method cannot be created as volatile".to_string()
            }))
        }
        api::PaymentMethodCreateData::BankRedirect(_) => {
            Err(report!(errors::ApiErrorResponse::UnprocessableEntity {
                message: "Bank redirect payment method cannot be created as volatile".to_string()
            }))
        }
    }
}

#[cfg(feature = "v2")]
async fn payment_method_resolver(
    state: &SessionState,
    platform: &domain::Platform,
    customer_id: &id_type::GlobalCustomerId,
    req: &api::PaymentMethodCreate,
    payment_method_data: domain::PaymentMethodVaultingData,
) -> RouterResult<PaymentMethodResolver> {
    let locker = LockerType::from_micro_services_config(&state.conf.micro_services);
    locker
        .resolve_payment_method(state, platform, customer_id, req, payment_method_data)
        .await
}

#[cfg(feature = "v2")]
fn resolve_based_on_duplication_status(
    pm: domain::PaymentMethod,
    payment_method_id: id_type::GlobalPaymentMethodId,
    duplication_check: Option<pm_transforms::DataDuplicationCheck>,
    payment_method_data: &domain::PaymentMethodVaultingData,
) -> RouterResult<PaymentMethodResolver> {
    let is_update_eligible = matches!(
        pm.status,
        enums::PaymentMethodStatus::New | enums::PaymentMethodStatus::Inactive
    );

    match duplication_check {
        Some(pm_transforms::DataDuplicationCheck::Duplicated)
        | Some(pm_transforms::DataDuplicationCheck::MetaDataChanged) => {
            if is_update_eligible {
                logger::info!(
                    "Payment method is duplicated with update-eligible status in legacy locker, updating existing payment method"
                );
                Ok(PaymentMethodResolver(PaymentMethodResolution::Update {
                    fingerprint_id: None,
                    payment_method_id,
                    payment_method: Box::new(pm),
                    source_payment_method_data: payment_method_data.clone(),
                }))
            } else if pm.status == enums::PaymentMethodStatus::Active {
                logger::info!("Payment method is duplicated, returning existing payment method");
                Ok(PaymentMethodResolver(PaymentMethodResolution::Get(
                    Box::new(pm),
                )))
            } else {
                logger::info!(
                    "Payment method is in awaiting data or processing state, no existing payment method entry found with locker id"
                );
                Err(
                    report!(errors::ApiErrorResponse::InternalServerError).attach_printable(
                        "Payment method is in invalid state, expected New or Inactive",
                    ),
                )
            }
        }
        None => {
            logger::info!("No duplication check data available from legacy locker");
            Ok(PaymentMethodResolver(PaymentMethodResolution::Get(
                Box::new(pm),
            )))
        }
    }
}

#[cfg(feature = "v2")]
pub enum PaymentMethodResolution {
    Get(Box<domain::PaymentMethod>),
    Update {
        fingerprint_id: Option<String>,
        payment_method_id: id_type::GlobalPaymentMethodId,
        payment_method: Box<domain::PaymentMethod>,
        source_payment_method_data: domain::PaymentMethodVaultingData,
    },
    Create {
        fingerprint_details: Option<FingerprintDetails>,
        payment_method_data: domain::PaymentMethodVaultingData,
        locker_resolver: LockerTypeResolver,
    },
}

#[cfg(feature = "v2")]
pub struct FingerprintDetails {
    pub fingerprint_id: Option<String>,
    pub auxiliary_fingerprint_id: Option<String>,
}

#[cfg(feature = "v2")]
#[derive(Debug, Clone, Copy)]
pub enum LockerType {
    Generic,
    Legacy,
}
#[cfg(feature = "v2")]
pub struct LockerTypeResolver {
    pub locker_type: LockerType,
    pub card_reference: Option<String>,
}

#[cfg(feature = "v2")]
#[allow(clippy::too_many_arguments)]
#[async_trait]
pub trait LockerOperations: Send + Sync {
    /// Retrieves a payment method from the locker
    async fn retrieve_payment_method_from_locker(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        vault_id: &domain::VaultId,
        customer_id: &id_type::GlobalCustomerId,
        payment_method_type: Option<enums::PaymentMethod>,
    ) -> CustomResult<pm_types::VaultRetrieveResponse, errors::VaultError>;

    /// Deletes a payment method from the locker
    async fn delete_payment_method_from_locker(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        vault_id: domain::VaultId,
        customer_id: &id_type::GlobalCustomerId,
    ) -> CustomResult<pm_types::VaultDeleteResponse, errors::VaultError>;

    /// Resolves payment method by checking for duplicates and determining the action to take
    async fn resolve_payment_method(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        customer_id: &id_type::GlobalCustomerId,
        req: &api::PaymentMethodCreate,
        payment_method_data: domain::PaymentMethodVaultingData,
    ) -> RouterResult<PaymentMethodResolver>;

    /// Vaults a new payment method to the locker
    /// For Generic: calls vault_payment_method
    /// For Legacy: constructs AddVaultResponse from card_reference
    async fn vault_new_payment_method(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        profile: &domain::Profile,
        payment_method_data: &domain::PaymentMethodVaultingData,
        fingerprint_id_from_vault: Option<String>,
        customer_id: &id_type::GlobalCustomerId,
        card_reference: Option<String>,
    ) -> RouterResult<(
        pm_types::AddVaultResponse,
        Option<id_type::MerchantConnectorAccountId>,
    )>;

    /// Validates locker operations required for updating an existing payment method
    async fn validate_locker_operations_for_update(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        profile: &domain::Profile,
        payment_method: &domain::PaymentMethod,
        vault_request_data: &domain::PaymentMethodVaultingData,
        is_metadata_changed_for_same_fingerprint: bool,
    ) -> RouterResult<(
        Option<domain::PaymentMethodVaultingData>,
        Option<pm_types::AddVaultResponse>,
    )>;
}

/// Generic locker implementation using the new vault system
#[cfg(feature = "v2")]
pub struct GenericLocker;

/// Legacy locker implementation using the V1 locker system
#[cfg(feature = "v2")]
pub struct LegacyLocker;

#[cfg(feature = "v2")]
#[async_trait]
impl LockerOperations for GenericLocker {
    async fn retrieve_payment_method_from_locker(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        vault_id: &domain::VaultId,
        customer_id: &id_type::GlobalCustomerId,
        _payment_method_type: Option<enums::PaymentMethod>,
    ) -> CustomResult<pm_types::VaultRetrieveResponse, errors::VaultError> {
        let should_trigger_fingerprint_migration =
            payment_method_utils::get_should_trigger_fingerprint_migration(
                state,
                None,
                platform.get_provider().get_provider_merchant_id(),
            )
            .await;

        let payload = cards::encode_vault_retrieve_request(
            should_trigger_fingerprint_migration,
            platform.get_provider().get_account().get_id().clone(),
            customer_id,
            vault_id.get_string_repr(),
        )
        .change_context(errors::VaultError::RequestEncodingFailed)?;

        let resp = vault::call_to_vault::<pm_types::VaultRetrieve>(state, payload, None, None)
            .await
            .change_context(errors::VaultError::VaultAPIError)
            .attach_printable("Call to vault failed")?;

        let stored_pm_resp: pm_types::VaultRetrieveResponse = resp
            .parse_struct("VaultRetrieveResponse")
            .change_context(errors::VaultError::ResponseDeserializationFailed)
            .attach_printable("Failed to parse data into VaultRetrieveResponse")?;

        Ok(stored_pm_resp)
    }

    async fn delete_payment_method_from_locker(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        vault_id: domain::VaultId,
        customer_id: &id_type::GlobalCustomerId,
    ) -> CustomResult<pm_types::VaultDeleteResponse, errors::VaultError> {
        let should_trigger_fingerprint_migration =
            payment_method_utils::get_should_trigger_fingerprint_migration(
                state,
                None,
                platform.get_provider().get_provider_merchant_id(),
            )
            .await;

        let payload = if should_trigger_fingerprint_migration {
            pm_types::VaultDeleteRequestNew {
                entity_id: platform.get_provider().get_account().get_id().clone(),
                vault_id,
            }
            .encode_to_vec()
            .change_context(errors::VaultError::RequestEncodingFailed)
            .attach_printable("Failed to encode VaultDeleteRequestNew")?
        } else {
            pm_types::VaultDeleteRequest {
                entity_id: customer_id.to_owned(),
                vault_id,
            }
            .encode_to_vec()
            .change_context(errors::VaultError::RequestEncodingFailed)
            .attach_printable("Failed to encode VaultDeleteRequest")?
        };

        let resp = vault::call_to_vault::<pm_types::VaultDelete>(state, payload, None, None)
            .await
            .change_context(errors::VaultError::VaultAPIError)
            .attach_printable("Call to vault failed")?;

        let stored_pm_resp: pm_types::VaultDeleteResponse = resp
            .parse_struct("VaultDeleteResponse")
            .change_context(errors::VaultError::ResponseDeserializationFailed)
            .attach_printable("Failed to parse data into VaultDeleteResponse")?;

        Ok(stored_pm_resp)
    }

    async fn resolve_payment_method(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        customer_id: &id_type::GlobalCustomerId,
        _req: &api::PaymentMethodCreate,
        payment_method_data: domain::PaymentMethodVaultingData,
    ) -> RouterResult<PaymentMethodResolver> {
        let db = &*state.store;

        let (fingerprint_id_result, auxiliary_fingerprint_id_result) = tokio::join!(
            vault::get_fingerprint_id_for_payment_method(
                state,
                &payment_method_data,
                customer_id.get_string_repr().to_owned(),
            ),
            vault::get_auxiliary_fingerprint_id_for_payment_method(
                state,
                &payment_method_data,
                customer_id.get_string_repr().to_owned(),
            ),
        );

        let fingerprint_id = fingerprint_id_result
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to get fingerprint_id from vault using generic strategy")?;

        match db
            .find_payment_method_by_fingerprint_id(
                platform.get_provider().get_key_store(),
                &fingerprint_id,
            )
            .await
        {
            Ok(existing_pm) => match existing_pm.status {
                enums::PaymentMethodStatus::New | enums::PaymentMethodStatus::Inactive => {
                    logger::info!(
                        "Payment method is duplicated with update-eligible status, updating existing payment method"
                    );
                    let payment_method_id = existing_pm.id.clone();
                    Ok(PaymentMethodResolver(PaymentMethodResolution::Update {
                        fingerprint_id: Some(fingerprint_id),
                        payment_method_id,
                        payment_method: Box::new(existing_pm),
                        source_payment_method_data: payment_method_data,
                    }))
                }
                enums::PaymentMethodStatus::Active => {
                    logger::info!(
                        "Payment method is duplicated, returning existing payment method"
                    );
                    Ok(PaymentMethodResolver(PaymentMethodResolution::Get(
                        Box::new(existing_pm),
                    )))
                }
                enums::PaymentMethodStatus::AwaitingData
                | enums::PaymentMethodStatus::Processing
                | enums::PaymentMethodStatus::Redacted => {
                    logger::info!(
                        "Payment method is in awaiting data or processing state, no existing payment method entry found with fingerprint id"
                    );
                    Err(
                        report!(errors::ApiErrorResponse::InternalServerError).attach_printable(
                            "Payment method is in invalid state, expected New or Inactive",
                        ),
                    )
                }
            },
            Err(err) => {
                when(!err.current_context().is_db_not_found(), || {
                    Err(err)
                        .change_context(errors::ApiErrorResponse::InternalServerError)
                        .attach_printable("Failed to find payment method by fingerprint id")
                })?;

                logger::debug!("Payment method not found, falling back to creation");

                let auxiliary_fingerprint_id = auxiliary_fingerprint_id_result
                    .change_context(errors::ApiErrorResponse::InternalServerError)
                    .attach_printable(
                        "Failed to get auxiliary fingerprint_id from vault using generic strategy",
                    )?;

                let locker_resolver = LockerTypeResolver {
                    locker_type: LockerType::Generic,
                    card_reference: None,
                };

                let fingerprint_details = FingerprintDetails {
                    fingerprint_id: Some(fingerprint_id),
                    auxiliary_fingerprint_id: Some(auxiliary_fingerprint_id.clone()),
                };

                Ok(PaymentMethodResolver(PaymentMethodResolution::Create {
                    fingerprint_details: Some(fingerprint_details),
                    payment_method_data,
                    locker_resolver,
                }))
            }
        }
    }

    async fn vault_new_payment_method(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        profile: &domain::Profile,
        payment_method_data: &domain::PaymentMethodVaultingData,
        fingerprint_id_from_vault: Option<String>,
        customer_id: &id_type::GlobalCustomerId,
        _card_reference: Option<String>,
    ) -> RouterResult<(
        pm_types::AddVaultResponse,
        Option<id_type::MerchantConnectorAccountId>,
    )> {
        vault_payment_method(
            state,
            payment_method_data,
            platform,
            profile,
            None,
            fingerprint_id_from_vault,
            customer_id,
            Some(pm_types::WriteMode::Insert), // Insert mode for new payment methods
        )
        .await
    }

    async fn validate_locker_operations_for_update(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        profile: &domain::Profile,
        payment_method: &domain::PaymentMethod,
        vault_request_data: &domain::PaymentMethodVaultingData,
        is_metadata_changed_for_same_fingerprint: bool,
    ) -> RouterResult<(
        Option<domain::PaymentMethodVaultingData>,
        Option<pm_types::AddVaultResponse>,
    )> {
        let fingerprint_id_from_vault = vault::get_fingerprint_id_for_payment_method(
            state,
            vault_request_data,
            payment_method
                .customer_id
                .clone()
                .get_required_value("GlobalCustomerId")?
                .get_string_repr()
                .to_owned(),
        )
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to get fingerprint_id from vault")?;

        let current_vault_id = payment_method.locker_id.clone();

        let is_stored_payment_method_same_as_in_request = payment_method
            .locker_fingerprint_id
            .clone()
            .map(|data| data == fingerprint_id_from_vault);

        match is_stored_payment_method_same_as_in_request {
            Some(true) => {
                if is_metadata_changed_for_same_fingerprint {
                    logger::info!(
                        "Payment method fingerprint is same, updating only payment method metadata in db"
                    );
                    let (vault_response, _) = vault_payment_method(
                        state,
                        vault_request_data,
                        platform,
                        profile,
                        current_vault_id,
                        Some(fingerprint_id_from_vault),
                        &payment_method
                            .customer_id
                            .to_owned()
                            .get_required_value("GlobalCustomerId")?,
                        Some(pm_types::WriteMode::Upsert), // Use Upsert mode for updates
                    )
                    .await
                    .attach_printable("Failed to add payment method in vault")?;
                    Ok((Some(vault_request_data.clone()), Some(vault_response)))
                } else {
                    logger::info!(
                        "Payment method vault data is same as in request, skipping vault update"
                    );
                    Ok((None, None))
                }
            }
            Some(false) | None => {
                logger::info!("Payment method vault data is different from request, proceeding with vault update");
                let (vault_response, _) = vault_payment_method(
                    state,
                    vault_request_data,
                    platform,
                    profile,
                    current_vault_id,
                    Some(fingerprint_id_from_vault),
                    &payment_method
                        .customer_id
                        .to_owned()
                        .get_required_value("GlobalCustomerId")?,
                    Some(pm_types::WriteMode::Insert), // Use Insert mode for new payment methods
                )
                .await
                .attach_printable("Failed to add payment method in vault")?;

                Ok((Some(vault_request_data.clone()), Some(vault_response)))
            }
        }
    }
}

#[cfg(feature = "v2")]
#[async_trait]
impl LockerOperations for LegacyLocker {
    async fn retrieve_payment_method_from_locker(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        vault_id: &domain::VaultId,
        customer_id: &id_type::GlobalCustomerId,
        payment_method_type: Option<enums::PaymentMethod>,
    ) -> CustomResult<pm_types::VaultRetrieveResponse, errors::VaultError> {
        let cust_id =
            id_type::CustomerId::try_from(Cow::from(customer_id.get_string_repr().to_owned()))
                .change_context(errors::VaultError::FetchCardFailed)
                .attach_printable("Failed to convert to CustomerId")?;

        let merchant_id = platform.get_provider().get_account().get_id();

        let payload = pm_transforms::CardReqBody {
            merchant_id: merchant_id.clone(),
            merchant_customer_id: cust_id,
            card_reference: vault_id.get_string_repr().to_owned(),
        };

        let get_card_resp: pm_transforms::RetrieveCardResp = pm_transforms::call_vault_api(
            state,
            state.conf.jwekey.get_inner(),
            &state.conf.locker,
            &payload,
            consts::LOCKER_RETRIEVE_CARD_PATH,
            state.tenant.tenant_id.clone(),
            state.request_id.clone(),
        )
        .await
        .change_context(errors::VaultError::FetchCardFailed)
        .attach_printable("Failed to fetch card from legacy locker")?;

        let resp_payload = get_card_resp
            .payload
            .ok_or(errors::VaultError::FetchCardFailed)
            .attach_printable("Empty payload in retrieve card response")?;

        let vaulting_data = match (payment_method_type, resp_payload.card) {
            (Some(enums::PaymentMethod::Card), Some(card)) => card.into(),
            (Some(enums::PaymentMethod::NetworkToken), Some(card)) => {
                domain::PaymentMethodVaultingData::NetworkToken(card.into())
            }
            (_, _) => {
                logger::warn!("Payment method not supported for retrieve from legacy locker");
                return Err(error_stack::report!(errors::VaultError::FetchCardFailed)
                    .attach_printable(
                        "Payment method not supported for retrieve from legacy locker",
                    ));
            }
        };

        Ok(pm_types::VaultRetrieveResponse {
            data: vaulting_data,
        })
    }

    async fn delete_payment_method_from_locker(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        vault_id: domain::VaultId,
        customer_id: &id_type::GlobalCustomerId,
    ) -> CustomResult<pm_types::VaultDeleteResponse, errors::VaultError> {
        let cust_id =
            id_type::CustomerId::try_from(Cow::from(customer_id.get_string_repr().to_owned()))
                .change_context(errors::VaultError::FetchCardFailed)
                .attach_printable("Failed to convert to CustomerId")?;

        let merchant_id = platform.get_provider().get_account().get_id();

        let payload = pm_transforms::CardReqBody {
            merchant_id: merchant_id.clone(),
            merchant_customer_id: cust_id,
            card_reference: vault_id.get_string_repr().to_owned(),
        };

        let legacy_locker_res: pm_transforms::DeleteCardResp = pm_transforms::call_vault_api(
            state,
            state.conf.jwekey.get_inner(),
            &state.conf.locker,
            &payload,
            consts::LOCKER_DELETE_CARD_PATH,
            state.tenant.tenant_id.clone(),
            state.request_id.clone(),
        )
        .await
        .change_context(errors::VaultError::DeleteCardFailed)
        .attach_printable("Failed to delete card from legacy locker")?;

        Ok(pm_types::VaultDeleteResponse {
            vault_id,
            entity_id: customer_id.to_owned(),
        })
    }

    async fn resolve_payment_method(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        customer_id: &id_type::GlobalCustomerId,
        _req: &api::PaymentMethodCreate,
        payment_method_data: domain::PaymentMethodVaultingData,
    ) -> RouterResult<PaymentMethodResolver> {
        let db = &*state.store;

        // Legacy locker does not use fingerprint for deduplication.
        // Dedup is handled by V1 locker's DataDuplicationCheck in the add response.
        let legacy_locker_res = add_payment_method_to_legacy_locker(
            state,
            platform,
            &payment_method_data,
            None,
            customer_id,
        )
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to add payment method to legacy locker")?;

        let pm_result = db
            .find_payment_method_by_locker_id(
                platform.get_provider().get_key_store(),
                &legacy_locker_res.card_reference,
                platform.get_processor().get_account().storage_scheme,
            )
            .await;

        match pm_result {
            Ok(existing_pm) => {
                let payment_method_id = existing_pm.id.clone();

                resolve_based_on_duplication_status(
                    existing_pm,
                    payment_method_id,
                    legacy_locker_res.duplication_check,
                    &payment_method_data,
                )
            }
            Err(err) => {
                when(!err.current_context().is_db_not_found(), || {
                    Err(err)
                        .change_context(errors::ApiErrorResponse::InternalServerError)
                        .attach_printable("Failed to find payment method by locker id")
                })?;

                logger::debug!("Payment method not found with locker id, falling back to creation");

                let locker_resolver = LockerTypeResolver {
                    locker_type: LockerType::Legacy,
                    card_reference: Some(legacy_locker_res.card_reference),
                };

                Ok(PaymentMethodResolver(PaymentMethodResolution::Create {
                    fingerprint_details: None,
                    payment_method_data: payment_method_data.clone(),
                    locker_resolver,
                }))
            }
        }
    }

    async fn vault_new_payment_method(
        &self,
        _state: &SessionState,
        _platform: &domain::Platform,
        _profile: &domain::Profile,
        _payment_method_data: &domain::PaymentMethodVaultingData,
        _fingerprint_id_from_vault: Option<String>,
        customer_id: &id_type::GlobalCustomerId,
        card_reference: Option<String>,
    ) -> RouterResult<(
        pm_types::AddVaultResponse,
        Option<id_type::MerchantConnectorAccountId>,
    )> {
        let card_reference = card_reference
            .get_required_value("card_reference")
            .attach_printable("Card Reference not found")?;

        let add_vault_response = pm_types::AddVaultResponse {
            entity_id: None,
            vault_id: domain::VaultId::generate(card_reference),
            fingerprint_id: None,
        };
        Ok((add_vault_response, None))
    }

    async fn validate_locker_operations_for_update(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        _profile: &domain::Profile,
        payment_method: &domain::PaymentMethod,
        vault_request_data: &domain::PaymentMethodVaultingData,
        _is_metadata_changed_for_same_fingerprint: bool,
    ) -> RouterResult<(
        Option<domain::PaymentMethodVaultingData>,
        Option<pm_types::AddVaultResponse>,
    )> {
        let customer_id = payment_method
            .customer_id
            .to_owned()
            .get_required_value("GlobalCustomerId")?;

        let existing_locker_id = payment_method.locker_id.clone();

        let legacy_locker_res = add_payment_method_to_legacy_locker(
            state,
            platform,
            vault_request_data,
            existing_locker_id.clone(),
            &customer_id,
        )
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to add payment method to legacy locker")?;

        match legacy_locker_res.duplication_check {
            Some(pm_transforms::DataDuplicationCheck::Duplicated) => {
                logger::info!(
                    "Payment method vault data is same as in request, skipping vault update"
                );
                Ok((None, None))
            }
            Some(pm_transforms::DataDuplicationCheck::MetaDataChanged) => {
                let add_vault_response = update_metadata_changed_payment_method_in_legacy_locker(
                    state,
                    platform,
                    vault_request_data,
                    existing_locker_id,
                    &customer_id,
                    legacy_locker_res.card_reference,
                )
                .await
                .attach_printable(
                    "Failed to update metadata-changed payment method in legacy locker",
                )?;

                logger::info!("Updating only payment method metadata in legacy locker");
                Ok((Some(vault_request_data.clone()), Some(add_vault_response)))
            }
            None => Ok((None, None)),
        }
    }
}

#[cfg(feature = "v2")]
pub async fn update_metadata_changed_payment_method_in_legacy_locker(
    state: &SessionState,
    platform: &domain::Platform,
    vault_request_data: &domain::PaymentMethodVaultingData,
    existing_vault_id: Option<domain::VaultId>,
    customer_id: &id_type::GlobalCustomerId,
    metadata_changed_card_reference: String,
) -> RouterResult<pm_types::AddVaultResponse> {
    LegacyLocker
        .delete_payment_method_from_locker(
            state,
            platform,
            domain::VaultId::generate(metadata_changed_card_reference),
            customer_id,
        )
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to delete payment method from legacy locker")?;

    let legacy_locker_res = add_payment_method_to_legacy_locker(
        state,
        platform,
        vault_request_data,
        existing_vault_id,
        customer_id,
    )
    .await
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to add payment method to legacy locker")?;

    Ok(pm_types::AddVaultResponse {
        entity_id: Some(customer_id.get_string_repr().to_owned()),
        vault_id: domain::VaultId::generate(legacy_locker_res.card_reference),
        fingerprint_id: None,
    })
}

#[cfg(feature = "v2")]
impl LockerType {
    pub fn from_micro_services_config(
        micro_services_config: &settings::MicroServicesConfig,
    ) -> Arc<dyn LockerOperations> {
        if micro_services_config.use_legacy_locker {
            Arc::new(LegacyLocker)
        } else {
            Arc::new(GenericLocker)
        }
    }
}

#[cfg(feature = "v2")]
pub struct PaymentMethodResolver(PaymentMethodResolution);

#[cfg(feature = "v2")]
#[allow(clippy::too_many_arguments)]
#[instrument(skip_all)]
async fn create_or_fetch_payment_method_core(
    state: &SessionState,
    req: api::PaymentMethodCreate,
    platform: &domain::Platform,
    profile: &domain::Profile,
    merchant_id: &id_type::MerchantId,
    customer_id: &id_type::GlobalCustomerId,
    payment_method_id: id_type::GlobalPaymentMethodId,
    billing_address: Option<Encryptable<hyperswitch_domain_models::address::Address>>,
) -> RouterResult<(api::PaymentMethodResponse, domain::PaymentMethod)> {
    let bin_enriched_payment_method_data =
        domain::PaymentMethodVaultingData::try_from(req.payment_method_data.clone())?
            .populate_bin_details_for_payment_method(state)
            .await;

    let payment_method_subtype = bin_enriched_payment_method_data
        .payment_method_subtype
        .or(req.payment_method_subtype);

    let payment_method_data = bin_enriched_payment_method_data.data;

    let resolver =
        payment_method_resolver(state, platform, customer_id, &req, payment_method_data).await?;

    Box::pin(resolver.execute(
        state,
        &req,
        platform,
        profile,
        merchant_id,
        customer_id,
        payment_method_id,
        payment_method_subtype,
        billing_address,
    ))
    .await
}

#[cfg(feature = "v2")]
#[allow(clippy::too_many_arguments)]
pub async fn create_payment_method_bank_redirect_core(
    state: &SessionState,
    req: api::PaymentMethodCreate,
    platform: &domain::Platform,
    profile: &domain::Profile,
    merchant_id: &id_type::MerchantId,
    customer_id: &id_type::GlobalCustomerId,
    payment_method_id: id_type::GlobalPaymentMethodId,
    billing_address: Option<Encryptable<hyperswitch_domain_models::address::Address>>,
) -> RouterResult<(api::PaymentMethodResponse, domain::PaymentMethod)> {
    let db = &*state.store;

    let existing_payment_methods = db
        .find_payment_method_by_global_customer_id_merchant_id_statuses(
            platform.get_provider().get_key_store(),
            customer_id,
            merchant_id,
            vec![
                enums::PaymentMethodStatus::Active,
                enums::PaymentMethodStatus::New,
            ],
            None,
            platform.get_provider().get_account().storage_scheme,
        )
        .await;

    let existing_wallet_pm = existing_payment_methods.ok().and_then(|pms| {
        pms.into_iter()
            .find(|pm| pm.get_payment_method_subtype() == req.payment_method_subtype)
    });

    let payment_method = match existing_wallet_pm {
        Some(existing_pm) => {
            logger::debug!("Payment method is duplicate, returning existing payment method record");
            existing_pm
        }
        None => {
            logger::debug!("Payment method is new, creating a new payment method record");
            let payment_method = create_payment_method_for_confirm(
                state,
                customer_id,
                payment_method_id,
                None,
                merchant_id,
                &platform.get_provider().get_account().organization_id,
                platform.get_provider().get_key_store(),
                platform.get_provider().get_account().storage_scheme,
                req.payment_method_type,
                req.payment_method_subtype,
                billing_address,
                None,
                None,
                None,
                platform.get_initiator(),
                enums::PaymentMethodStatus::New,
            )
            .await?;

            payment_method
        }
    };

    let payment_method_response = pm_transforms::generate_payment_method_response(
        &payment_method,
        &None,
        req.storage_type,
        None,
        req.customer_id,
        None,
        None,
        None,
    )?;

    Ok((payment_method_response, payment_method))
}

#[cfg(feature = "v2")]
impl PaymentMethodResolver {
    #[allow(clippy::too_many_arguments)]
    #[instrument(skip_all)]
    pub async fn execute(
        self,
        state: &SessionState,
        req: &api::PaymentMethodCreate,
        platform: &domain::Platform,
        profile: &domain::Profile,
        merchant_id: &id_type::MerchantId,
        customer_id: &id_type::GlobalCustomerId,
        payment_method_id: id_type::GlobalPaymentMethodId,
        payment_method_subtype: Option<storage_enums::PaymentMethodType>,
        billing_address: Option<Encryptable<hyperswitch_domain_models::address::Address>>,
    ) -> RouterResult<(api::PaymentMethodResponse, domain::PaymentMethod)> {
        let db = &*state.store;
        match self.0 {
            PaymentMethodResolution::Get(existing_pm) => {
                logger::debug!("Payment method is duplicate, found {:?}", existing_pm.id);
                let card_cvc = req
                    .payment_method_data
                    .get_card()
                    .and_then(|card| card.card_cvc.clone());

                let cvc_expiry_details = if let Some(cvc) = card_cvc {
                    logger::debug!("Inserting CVC for payment method {:?}", existing_pm.id);
                    let intent_fulfillment_time =
                        common_utils::consts::DEFAULT_INTENT_FULFILLMENT_TIME;
                    Some(
                        vault::insert_cvc_using_payment_token(
                            state,
                            existing_pm.id.get_string_repr(),
                            cvc,
                            intent_fulfillment_time,
                            platform.get_provider().get_key_store(),
                        )
                        .await?,
                    )
                } else {
                    logger::debug!(
                        "No CVC found in the payment method request, trying to retrieve from redis"
                    );
                    vault::retrieve_key_and_ttl_for_cvc_from_payment_method_id(
                        state,
                        existing_pm.id.to_owned(),
                    )
                    .await
                    .change_context(errors::ApiErrorResponse::InternalServerError)
                    .attach_printable("Failed to retrieve cvc from redis")
                    .ok()
                    .map(|time| {
                        payment_methods::CardCVCTokenStorageDetails::generate_expiry_timestamp(time)
                    })
                };
                let billing = billing_address
                    .clone()
                    .map(|billing| billing.into_inner())
                    .map(From::from);

                let resp = pm_transforms::generate_payment_method_response(
                    &existing_pm,
                    &None,
                    req.storage_type,
                    cvc_expiry_details,
                    req.customer_id.clone(),
                    None,
                    billing,
                    None,
                )?;

                Ok((resp, *existing_pm))
            }

            PaymentMethodResolution::Update {
                fingerprint_id,
                payment_method_id,
                payment_method,
                source_payment_method_data,
            } => {
                logger::debug!(
                    "Updating duplicated payment method {:?} for fingerprint {:?}",
                    payment_method_id,
                    fingerprint_id
                );

                let payment_method = *payment_method;
                let payment_method_has_network_token_data = payment_method
                    .network_token_requestor_reference_id
                    .is_some()
                    || payment_method.network_token_locker_id.is_some()
                    || payment_method.network_token_payment_method_data.is_some();

                // When the re-added card is not yet network tokenized, defer generation to the
                // async `NetworkTokenizationWorkflow` process tracker job (scheduled after the
                // update below) rather than tokenizing inline. The job needs a customer to
                // tokenize against, and it is not changed by this update, so it is read from the
                // existing payment method before it is consumed.
                let network_tokenization_customer_id = (!payment_method_has_network_token_data
                    && should_schedule_network_tokenization(
                        profile,
                        req.payment_method_type,
                        req.network_tokenization.as_ref(),
                    ))
                .then(|| payment_method.customer_id.clone())
                .flatten();

                let mut update_payment_method_data: DomainPaymentMethodUpdate =
                    (req, source_payment_method_data).into();

                // Reset Inactive status to New when customer re-adds the card
                if payment_method.status == enums::PaymentMethodStatus::Inactive {
                    update_payment_method_data.status = Some(enums::PaymentMethodStatus::New);
                }

                let (response, updated_payment_method) = Box::pin(update_payment_method_core(
                    state,
                    platform,
                    profile,
                    update_payment_method_data,
                    &payment_method_id,
                    Some(payment_method),
                    None,
                ))
                .await?;

                if let Some(customer_id) = network_tokenization_customer_id {
                    schedule_network_tokenization_task(
                        state,
                        platform,
                        profile,
                        &updated_payment_method.id,
                        customer_id,
                    )
                    .await;
                }

                Ok((response, updated_payment_method))
            }

            PaymentMethodResolution::Create {
                fingerprint_details,
                payment_method_data,
                locker_resolver,
            } => {
                let fingerprint_id = fingerprint_details
                    .as_ref()
                    .and_then(|details| details.fingerprint_id.clone());

                let auxiliary_fingerprint_id = fingerprint_details
                    .and_then(|details| details.auxiliary_fingerprint_id.clone());

                let payment_method = create_payment_method_for_intent(
                    state,
                    req.metadata.clone(),
                    customer_id,
                    payment_method_id,
                    merchant_id,
                    &platform.get_provider().get_account().organization_id,
                    platform.get_provider().get_key_store(),
                    platform.get_provider().get_account().storage_scheme,
                    billing_address.clone(),
                    platform.get_initiator(),
                    auxiliary_fingerprint_id,
                )
                .await?;
                Box::pin(execute_payment_method_create(
                    state,
                    req,
                    platform,
                    profile,
                    customer_id,
                    payment_method,
                    payment_method_subtype,
                    payment_method_data,
                    fingerprint_id,
                    billing_address,
                    locker_resolver,
                ))
                .await
            }
        }
    }
}

#[cfg(feature = "v2")]
#[allow(clippy::too_many_arguments)]
#[instrument(skip_all)]
async fn execute_payment_method_create(
    state: &SessionState,
    req: &api::PaymentMethodCreate,
    platform: &domain::Platform,
    profile: &domain::Profile,
    customer_id: &id_type::GlobalCustomerId,
    payment_method: domain::PaymentMethod,
    payment_method_subtype: Option<storage_enums::PaymentMethodType>,
    payment_method_data: domain::PaymentMethodVaultingData,
    fingerprint_id_from_vault: Option<String>,
    payment_method_billing_address: Option<
        Encryptable<hyperswitch_domain_models::address::Address>,
    >,
    locker_resolver: LockerTypeResolver,
) -> RouterResult<(api::PaymentMethodResponse, domain::PaymentMethod)> {
    match &req.payment_method_data {
        api::PaymentMethodCreateData::Card(card_data) => {
            payments_core::helpers::validate_card_expiry(
                &card_data.card_exp_month,
                &card_data.card_exp_year,
            )?;
        }
        _ => {
            logger::debug!("Payment method data is not CardDetail");
        }
    }
    let db = &*state.store;

    let locker = LockerType::from_micro_services_config(&state.conf.micro_services);

    let vaulting_result = locker
        .vault_new_payment_method(
            state,
            platform,
            profile,
            &payment_method_data,
            fingerprint_id_from_vault,
            customer_id,
            locker_resolver.card_reference.clone(),
        )
        .await;

    // Network tokenization is generated asynchronously via the `NetworkTokenizationWorkflow`
    // process tracker job (scheduled after the payment method row is created below), so the
    // payment method is initially persisted without network token data.
    let schedule_network_tokenization = should_schedule_network_tokenization(
        profile,
        req.payment_method_type,
        req.network_tokenization.as_ref(),
    );

    let (response, payment_method) = match vaulting_result {
        Ok((
            pm_types::AddVaultResponse {
                vault_id,
                fingerprint_id,
                ..
            },
            external_vault_source,
        )) => {
            let pm_update = create_pm_additional_data_update(
                Some(&payment_method_data),
                state,
                platform.get_provider().get_key_store(),
                Some(vault_id.get_string_repr().clone()),
                fingerprint_id,
                &payment_method,
                None,
                None,
                None,
                Some(req.payment_method_type),
                payment_method_subtype,
                external_vault_source,
                Some(enums::PaymentMethodStatus::New),
                platform.get_initiator(),
            )
            .await
            .attach_printable("unable to create payment method data")?;

            let payment_method = db
                .update_payment_method(
                    platform.get_provider().get_key_store(),
                    payment_method,
                    pm_update,
                    platform.get_provider().get_account().storage_scheme,
                    Some(payment_method_modular_backward_compat_action(
                        state,
                        &platform.get_provider().get_account().organization_id,
                    )),
                )
                .await
                .change_context(errors::ApiErrorResponse::InternalServerError)
                .attach_printable("Failed to update payment method in db")?;

            let intent_fulfillment_time = common_utils::consts::DEFAULT_INTENT_FULFILLMENT_TIME;

            let card_cvc = req
                .payment_method_data
                .get_card()
                .and_then(|card| card.card_cvc.clone());

            let cvc_expiry_details = card_cvc
                .async_map(|cvc| {
                    vault::insert_cvc_using_payment_token(
                        state,
                        payment_method.id.get_string_repr(),
                        cvc,
                        intent_fulfillment_time,
                        platform.get_provider().get_key_store(),
                    )
                })
                .await
                .transpose()?;

            let resp = pm_transforms::generate_payment_method_response(
                &payment_method,
                &None,
                req.storage_type,
                cvc_expiry_details,
                req.customer_id.clone(),
                None,
                payment_method_billing_address.map(|add| add.get_inner().clone().into()),
                None,
            )?;

            Ok((resp, payment_method))
        }
        Err(e) => {
            let pm_update = storage::PaymentMethodUpdate::StatusUpdate {
                status: Some(enums::PaymentMethodStatus::Inactive),
                last_modified_by: platform
                    .get_initiator()
                    .and_then(|initiator| initiator.to_created_by())
                    .map(|last_modified_by| last_modified_by.to_string()),
            };

            db.update_payment_method(
                platform.get_provider().get_key_store(),
                payment_method,
                pm_update,
                platform.get_provider().get_account().storage_scheme,
                // Inactivation on failed PM setup should not enqueue compatibility work.
                None,
            )
            .await
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to update payment method in db")?;

            Err(e)
        }
    }?;

    if schedule_network_tokenization {
        schedule_network_tokenization_task(
            state,
            platform,
            profile,
            &payment_method.id,
            customer_id.clone(),
        )
        .await;
    }

    Ok((response, payment_method))
}

#[cfg(feature = "v2")]
#[allow(clippy::too_many_arguments)]
#[instrument(skip_all)]
pub async fn create_generic_volatile_payment_method(
    state: &SessionState,
    req: api::PaymentMethodCreate,
    platform: &domain::Platform,
    profile: &domain::Profile,
    merchant_id: &id_type::MerchantId,
    customer_id: &Option<id_type::GlobalCustomerId>,
    payment_method_id: id_type::GlobalPaymentMethodId,
    payment_method_billing_address: Option<
        Encryptable<hyperswitch_domain_models::address::Address>,
    >,
) -> RouterResult<(api::PaymentMethodResponse, domain::PaymentMethod)> {
    let db = &*state.store;
    let keymanager_state = &(state).into();
    let merchant_key_store = platform.get_provider().get_key_store();

    let bin_enriched_payment_method_data =
        domain::PaymentMethodVaultingData::try_from(req.payment_method_data.clone())?
            .populate_bin_details_for_payment_method(state)
            .await;
    let payment_method_subtype = bin_enriched_payment_method_data
        .payment_method_subtype
        .or(req.payment_method_subtype);
    let payment_method_data = bin_enriched_payment_method_data.data;

    let vaulting_result = vault_payment_method_in_volatile_storage(
        state,
        &payment_method_data,
        platform,
        profile,
        None,
        customer_id,
    )
    .await;

    let (response, payment_method) = match vaulting_result {
        Ok((
            pm_types::AddVaultResponse {
                vault_id,
                fingerprint_id,
                ..
            },
            external_vault_source,
        )) => {
            let locker_id = Some(vault_id.clone());
            let payment_method = construct_payment_method_object(
                req.metadata.clone(),
                customer_id,
                payment_method_id,
                merchant_id,
                payment_method_billing_address.clone(),
                state,
                platform.get_provider().get_key_store(),
                Some(&payment_method_data),
                Some(req.payment_method_type),
                payment_method_subtype,
                locker_id,
                fingerprint_id,
                external_vault_source,
                platform.get_initiator(),
            )
            .await
            .attach_printable("failed to construct payment method")?
            .convert()
            .await
            .change_context(errors::StorageError::EncryptionError)
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to convert payment method")?; //Convert to storage model payment method to store in redis

            let redis_connection = state
                .store
                .get_redis_conn()
                .map_err(Into::<errors::StorageError>::into)
                .change_context(errors::ApiErrorResponse::InternalServerError)
                .attach_printable("Failed to get redis connection")?;

            logger::info!("Storing payment method id in redis");

            redis_connection
                .serialize_and_set_key_with_expiry(
                    &payment_method.get_id().get_string_repr().to_string().into(),
                    payment_method.clone(),
                    consts::DEFAULT_PAYMENT_METHOD_STORE_TTL,
                )
                .await
                .map_err(Into::<errors::StorageError>::into)
                .change_context(errors::ApiErrorResponse::InternalServerError)
                .attach_printable("Failed to insert payment method id in redis")?;

            let domain_payment_method = domain::PaymentMethod::convert_back(
                keymanager_state,
                payment_method,
                merchant_key_store.key.get_inner(),
                merchant_key_store.merchant_id.clone().into(),
            )
            .await
            .change_context(errors::StorageError::DecryptionError)
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("failed to convert payment method")?;

            let intent_fulfillment_time = common_utils::consts::DEFAULT_INTENT_FULFILLMENT_TIME;

            let card_cvc = req
                .payment_method_data
                .get_card()
                .and_then(|card| card.card_cvc.clone());

            let cvc_expiry_details = card_cvc
                .async_map(|cvc| {
                    vault::insert_cvc_using_payment_token(
                        state,
                        domain_payment_method.id.get_string_repr(),
                        cvc,
                        intent_fulfillment_time,
                        platform.get_provider().get_key_store(),
                    )
                })
                .await
                .transpose()?;

            let resp = pm_transforms::generate_payment_method_response(
                &domain_payment_method,
                &None,
                req.storage_type,
                cvc_expiry_details,
                req.customer_id,
                None,
                None,
                None,
            )?;

            Ok((resp, domain_payment_method))
        }
        Err(e) => Err(e),
    }?;

    Ok((response, payment_method))
}

#[cfg(feature = "v2")]
#[allow(clippy::too_many_arguments)]
#[instrument(skip_all)]
pub async fn create_payment_method_wallet_core(
    state: &SessionState,
    req: api::PaymentMethodCreate,
    platform: &domain::Platform,
    _profile: &domain::Profile,
    merchant_id: &id_type::MerchantId,
    customer_id: &id_type::GlobalCustomerId,
    payment_method_id: id_type::GlobalPaymentMethodId,
    payment_method_billing_address: Option<
        Encryptable<hyperswitch_domain_models::address::Address>,
    >,
    additional_payment_method_data: payment_methods::PaymentMethodsData,
) -> RouterResult<(api::PaymentMethodResponse, domain::PaymentMethod)> {
    use crate::core::payment_methods::cards;

    let db = &*state.store;
    let key_manager_state = &(state).into();

    let encrypted_payment_method_data = Some(additional_payment_method_data)
        .async_map(|payment_method_data| {
            cards::create_encrypted_data(
                key_manager_state,
                platform.get_provider().get_key_store(),
                payment_method_data,
            )
        })
        .await
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to encrypt wallet payment method data")?
        .map(|encoded_pmd| {
            encoded_pmd.deserialize_inner_value(|value| value.parse_value("PaymentMethodsData"))
        })
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to parse wallet payment method data")?;

    let existing_payment_methods = db
        .find_payment_method_by_global_customer_id_merchant_id_statuses(
            platform.get_provider().get_key_store(),
            customer_id,
            merchant_id,
            vec![
                enums::PaymentMethodStatus::Active,
                enums::PaymentMethodStatus::New,
            ],
            None,
            platform.get_provider().get_account().storage_scheme,
        )
        .await;

    let existing_wallet_pm = existing_payment_methods.ok().and_then(|pms| {
        pms.into_iter()
            .find(|pm| pm.get_payment_method_subtype() == req.payment_method_subtype)
    });

    let payment_method = match existing_wallet_pm {
        Some(existing_pm) => {
            let pm_update = storage::PaymentMethodUpdate::PaymentMethodDataUpdate {
                payment_method_data: encrypted_payment_method_data.map(From::from),
                last_modified_by: platform
                    .get_initiator()
                    .and_then(|initiator| initiator.to_created_by())
                    .map(|last_modified_by| last_modified_by.to_string()),
            };

            db.update_payment_method(
                platform.get_provider().get_key_store(),
                existing_pm,
                pm_update,
                platform.get_provider().get_account().storage_scheme,
                Some(payment_method_modular_backward_compat_action(
                    state,
                    &platform.get_provider().get_account().organization_id,
                )),
            )
            .await
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to update existing wallet payment method in db")?
        }
        None => {
            create_payment_method_for_confirm(
                state,
                customer_id,
                payment_method_id,
                None,
                merchant_id,
                &platform.get_provider().get_account().organization_id,
                platform.get_provider().get_key_store(),
                platform.get_provider().get_account().storage_scheme,
                req.payment_method_type,
                req.payment_method_subtype,
                payment_method_billing_address,
                encrypted_payment_method_data,
                None,
                None,
                platform.get_initiator(),
                enums::PaymentMethodStatus::New,
            )
            .await?
        }
    };

    let payment_method_response = pm_transforms::generate_payment_method_response(
        &payment_method,
        &None,
        req.storage_type,
        None,
        req.customer_id,
        None,
        None,
        None,
    )?;

    Ok((payment_method_response, payment_method))
}

// network tokenization and vaulting to locker is not required for proxy card since the card is already tokenized
#[cfg(feature = "v2")]
#[allow(clippy::too_many_arguments)]
#[instrument(skip_all)]
pub async fn create_payment_method_proxy_card_core(
    state: &SessionState,
    req: api::PaymentMethodCreate,
    platform: &domain::Platform,
    profile: &domain::Profile,
    merchant_id: &id_type::MerchantId,
    customer_id: &id_type::GlobalCustomerId,
    payment_method_id: id_type::GlobalPaymentMethodId,
    payment_method_billing_address: Option<
        Encryptable<hyperswitch_domain_models::address::Address>,
    >,
) -> RouterResult<(api::PaymentMethodResponse, domain::PaymentMethod)> {
    use crate::core::payment_methods::vault::Vault;

    let key_manager_state = &(state).into();

    // A proxy card is, by definition, vaulted in an external vault, so the external vault
    // connector must be configured on the resolved profile (the platform merchant's profile in
    // platform flows). Fetch it from the profile and require it — otherwise the payment method
    // would be persisted with a `None` external vault source.
    let external_vault_profile =
        payments_core::helpers::resolve_provider_profile(state, platform, profile).await?;
    let external_vault_source = Some(
        external_vault_profile
            .external_vault_connector_details
            .clone()
            .map(|details| details.vault_connector_id)
            .ok_or_else(|| report!(errors::ApiErrorResponse::InternalServerError))
            .attach_printable(
                "external_vault_connector_details must be configured on the profile to create a proxy card payment method",
            )?,
    );

    let bin_enriched_payment_method_data = req
        .payment_method_data
        .populate_bin_details_for_payment_method(state)
        .await;
    let payment_method_subtype = bin_enriched_payment_method_data
        .payment_method_subtype
        .or(req.payment_method_subtype);
    let additional_payment_method_data = Some(
        bin_enriched_payment_method_data
            .data
            .convert_to_additional_payment_method_data()?,
    );

    let encrypted_payment_method_data = additional_payment_method_data
        .async_map(|payment_method_data| {
            core_utils::create_encrypted_data(
                key_manager_state,
                platform.get_provider().get_key_store(),
                payment_method_data,
                common_utils::type_name!(diesel_models::payment_method::PaymentMethod),
            )
        })
        .await
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to encrypt Payment method data")?
        .map(|encoded_pmd| {
            encoded_pmd.deserialize_inner_value(|value| value.parse_value("PaymentMethodsData"))
        })
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to parse Payment method data")?;

    let external_vault_token_data = req.payment_method_data.get_external_vault_token_data();

    let encrypted_external_vault_token_data = external_vault_token_data
        .async_map(|external_vault_token_data| {
            core_utils::create_encrypted_data(
                key_manager_state,
                platform.get_provider().get_key_store(),
                external_vault_token_data,
                common_utils::type_name!(diesel_models::payment_method::PaymentMethod),
            )
        })
        .await
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to encrypt External vault token data")?
        .map(|encoded_data| {
            encoded_data
                .deserialize_inner_value(|value| value.parse_value("ExternalVaultTokenData"))
        })
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to parse External vault token data")?;

    let vault_type = external_vault_source
        .is_some()
        .then_some(common_enums::VaultType::External);

    let payment_method = create_payment_method_for_confirm(
        state,
        customer_id,
        payment_method_id,
        external_vault_source,
        merchant_id,
        &platform.get_provider().get_account().organization_id,
        platform.get_provider().get_key_store(),
        platform.get_provider().get_account().storage_scheme,
        req.payment_method_type,
        payment_method_subtype,
        payment_method_billing_address,
        encrypted_payment_method_data,
        encrypted_external_vault_token_data,
        vault_type,
        platform.get_initiator(),
        enums::PaymentMethodStatus::Inactive,
    )
    .await?;

    let payment_method_response = pm_transforms::generate_payment_method_response(
        &payment_method,
        &None,
        req.storage_type,
        None,
        req.customer_id,
        None,
        None,
        None,
    )?;

    Ok((payment_method_response, payment_method))
}

#[cfg(feature = "v2")]
#[derive(Clone, Debug)]
pub struct NetworkTokenPaymentMethodDetails {
    network_token_requestor_reference_id: String,
    network_token_locker_id: String,
    network_token_pmd: Encryptable<Secret<serde_json::Value>>,
}

#[cfg(feature = "v2")]
#[derive(Clone, Debug)]
pub struct BinEnriched<T> {
    pub data: T,
    pub payment_method_subtype: Option<storage_enums::PaymentMethodType>,
}

/// Generates a network token for the given card and vaults it.
///
/// Errors are returned to the caller so that callers which can act on a failure (such as the
/// async `NetworkTokenizationWorkflow`, which retries) are able to distinguish a failed
/// tokenization from one that was never applicable.
#[cfg(feature = "v2")]
pub async fn network_tokenize_and_vault_the_pmd(
    state: &SessionState,
    payment_method_data: &domain::PaymentMethodVaultingData,
    platform: &domain::Platform,
    network_tokenization: Option<common_types::payment_methods::NetworkTokenization>,
    network_tokenization_enabled_for_profile: bool,
    customer_id: &id_type::GlobalCustomerId,
) -> CustomResult<NetworkTokenPaymentMethodDetails, errors::NetworkTokenizationError> {
    when(!network_tokenization_enabled_for_profile, || {
        Err(report!(
            errors::NetworkTokenizationError::NetworkTokenizationNotEnabledForProfile
        ))
    })?;

    let is_network_tokenization_enabled_for_pm = network_tokenization
        .as_ref()
        .map(|nt| matches!(nt.enable, common_enums::NetworkTokenizationToggle::Enable))
        .unwrap_or(false);

    let card_data = payment_method_data
        .get_card()
        .and_then(|card| is_network_tokenization_enabled_for_pm.then_some(card))
        .ok_or_else(|| {
            report!(errors::NetworkTokenizationError::NotSupported {
                message: "Payment method".to_string(),
            })
        })?;

    let (resp, network_token_req_ref_id) =
        network_tokenization::make_card_network_tokenization_request(state, card_data, customer_id)
            .await?;

    let network_token_vaulting_data = domain::PaymentMethodVaultingData::NetworkToken(resp);
    let vaulting_resp = vault::add_payment_method_to_vault(
        state,
        platform,
        &network_token_vaulting_data,
        None,
        customer_id,
        Some(pm_types::WriteMode::Insert), // Insert mode for new network tokens
    )
    .await
    .change_context(errors::NetworkTokenizationError::SaveNetworkTokenFailed)
    .attach_printable("Failed to vault network token")?;

    let key_manager_state = &(state).into();
    let network_token_pmd = core_utils::create_encrypted_data(
        key_manager_state,
        platform.get_provider().get_key_store(),
        network_token_vaulting_data.get_payment_methods_data(),
        common_utils::type_name!(diesel_models::payment_method::PaymentMethod),
    )
    .await
    .change_context(errors::NetworkTokenizationError::NetworkTokenDetailsEncryptionFailed)
    .attach_printable("Failed to encrypt PaymentMethodsData")?;

    Ok(NetworkTokenPaymentMethodDetails {
        network_token_requestor_reference_id: network_token_req_ref_id,
        network_token_locker_id: vaulting_resp.vault_id.get_string_repr().clone(),
        network_token_pmd,
    })
}

/// Called from the `NetworkTokenizationWorkflow` process tracker job (v2).
///
/// Re-fetches the card from the vault for the given payment method, generates a network token
/// and persists the network token details on the payment method row. This is the async
/// counterpart of the synchronous network tokenization performed during payment method create
/// and update.
///
/// A failure is returned as an error so that the workflow can retry it. Payment methods that
/// network tokenization does not apply to at all are reported as
/// [`NetworkTokenGenerationOutcome::NotApplicable`], since retrying those cannot succeed.
#[cfg(feature = "v2")]
pub async fn generate_network_token_for_payment_method(
    state: &SessionState,
    platform: &domain::Platform,
    profile: &domain::Profile,
    payment_method: domain::PaymentMethod,
) -> RouterResult<NetworkTokenGenerationOutcome> {
    let db = &*state.store;

    let customer_id = payment_method
        .customer_id
        .clone()
        .get_required_value("GlobalCustomerId")?;

    // Re-fetch the card that was vaulted at create/update time.
    let vault_data = Box::pin(vault::retrieve_payment_method_from_vault(
        state,
        platform,
        profile,
        &payment_method,
    ))
    .await
    .attach_printable("Failed to retrieve payment method from vault for network tokenization")?
    .data;

    // The per-payment-method opt-in was recorded when the task was scheduled and is not persisted
    // on the payment method, so request tokenization with the toggle enabled here. The
    // profile-level flag is still passed through and re-checked by the callee.
    let network_tokenization_resp = network_tokenize_and_vault_the_pmd(
        state,
        &vault_data,
        platform,
        Some(common_types::payment_methods::NetworkTokenization {
            enable: common_enums::NetworkTokenizationToggle::Enable,
        }),
        profile.is_network_tokenization_enabled,
        &customer_id,
    )
    .await;

    match network_tokenization_resp {
        Ok(network_token) => {
            // v2 has no NT-only update variant; `GenericUpdate` is the one that carries the
            // network token columns. Its changeset skips `None` fields, so only the network token
            // columns (and last_modified) are written.
            let pm_update = storage::PaymentMethodUpdate::GenericUpdate {
                payment_method_data: None,
                status: None,
                locker_id: None,
                payment_method_type_v2: None,
                payment_method_subtype: None,
                network_token_requestor_reference_id: Some(
                    network_token.network_token_requestor_reference_id,
                ),
                network_token_locker_id: Some(network_token.network_token_locker_id),
                network_token_payment_method_data: Some(network_token.network_token_pmd.into()),
                locker_fingerprint_id: None,
                connector_mandate_details: Box::new(None),
                external_vault_source: None,
                network_transaction_id: None,
                network_transaction_link_id: None,
                last_modified_by: platform
                    .get_initiator()
                    .and_then(|initiator| initiator.to_created_by())
                    .map(|last_modified_by| last_modified_by.to_string()),
            };

            db.update_payment_method(
                platform.get_provider().get_key_store(),
                payment_method,
                pm_update,
                platform.get_provider().get_account().storage_scheme,
                None,
            )
            .await
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to update payment method with network token details")
            .map(|_| NetworkTokenGenerationOutcome::Generated)
        }

        // Neither of these can be fixed by running the task again, so report them as a terminal
        // outcome instead of an error that burns the retry budget.
        Err(err)
            if matches!(
                err.current_context(),
                errors::NetworkTokenizationError::NotSupported { .. }
                    | errors::NetworkTokenizationError::NetworkTokenizationNotEnabledForProfile
            ) =>
        {
            logger::warn!(
                payment_method_id=%payment_method.id.get_string_repr(),
                ?err,
                "Network tokenization is not applicable for this payment method, not retrying"
            );
            Ok(NetworkTokenGenerationOutcome::NotApplicable)
        }

        Err(err) => Err(err)
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to generate and vault the network token"),
    }
}

/// Terminal outcome of an async network token generation attempt.
#[cfg(feature = "v2")]
#[derive(Clone, Copy, Debug)]
pub enum NetworkTokenGenerationOutcome {
    /// A network token was generated and persisted on the payment method.
    Generated,
    /// Network tokenization does not apply to this payment method; retrying cannot help.
    NotApplicable,
}

#[cfg(feature = "v2")]
#[async_trait::async_trait]
pub trait PaymentMethodExt {
    async fn populate_bin_details_for_payment_method(
        &self,
        state: &SessionState,
    ) -> BinEnriched<Self>
    where
        Self: Sized;

    // convert to data format compatible to save in payment method table
    fn convert_to_additional_payment_method_data(
        &self,
    ) -> RouterResult<payment_methods::PaymentMethodsData>;

    // get tokens generated from external vault
    fn get_external_vault_token_data(&self) -> Option<payment_methods::ExternalVaultTokenData>;
}

#[cfg(feature = "v2")]
#[async_trait::async_trait]
impl PaymentMethodExt for domain::PaymentMethodVaultingData {
    async fn populate_bin_details_for_payment_method(
        &self,
        state: &SessionState,
    ) -> BinEnriched<Self> {
        match self {
            Self::Card(card) => {
                let card_isin = card.card_number.get_card_isin();

                let card_info = state
                    .store
                    .get_card_info(&card_isin)
                    .await
                    .map_err(|error| services::logger::error!(card_info_error=?error))
                    .ok()
                    .flatten();

                let additional_payment_data_from_bin_lookup = card_info.as_ref().map(|val| {
                    api_models::payments::AdditionalPaymentData::Card(Box::new(
                        api_models::payments::AdditionalCardInfo {
                            card_type: val.card_type.clone(),
                            ..Default::default()
                        },
                    ))
                });

                let payment_method_subtype =
                    Option::<storage_enums::PaymentMethodType>::foreign_from((
                        None,
                        additional_payment_data_from_bin_lookup.as_ref(),
                        Some(storage_enums::PaymentMethod::Card),
                    ));

                if card.card_issuer.is_some()
                    && card.card_network.is_some()
                    && card.card_type.is_some()
                    && card.card_issuing_country.is_some()
                {
                    BinEnriched {
                        data: Self::Card(card.clone()),
                        payment_method_subtype,
                    }
                } else {
                    BinEnriched {
                        data: Self::Card(payment_methods::CardDetail {
                            card_number: card.card_number.clone(),
                            card_exp_month: card.card_exp_month.clone(),
                            card_exp_year: card.card_exp_year.clone(),
                            card_holder_name: card.card_holder_name.clone(),
                            nick_name: card.nick_name.clone(),
                            card_issuing_country: card_info.as_ref().and_then(|val| {
                                val.card_issuing_country
                                    .as_ref()
                                    .map(|c| api_enums::CountryAlpha2::from_str(c))
                                    .transpose()
                                    .ok()
                                    .flatten()
                            }),
                            card_network: card_info
                                .as_ref()
                                .and_then(|val| val.card_network.clone()),
                            card_issuer: card_info.as_ref().and_then(|val| val.card_issuer.clone()),
                            card_type: card_info.as_ref().and_then(|val| {
                                val.card_type
                                    .as_ref()
                                    .map(|c| payment_methods::CardType::from_str(c))
                                    .transpose()
                                    .ok()
                                    .flatten()
                            }),
                            card_cvc: card.card_cvc.clone(),
                        }),
                        payment_method_subtype,
                    }
                }
            }
            _ => BinEnriched {
                data: self.clone(),
                payment_method_subtype: None,
            },
        }
    }

    // Not implement because it is saved in locker and not in payment method table
    fn convert_to_additional_payment_method_data(
        &self,
    ) -> RouterResult<payment_methods::PaymentMethodsData> {
        Err(report!(errors::ApiErrorResponse::UnprocessableEntity {
            message: "Payment method data is not supported".to_string()
        })
        .attach_printable("Payment method data is not supported"))
    }

    fn get_external_vault_token_data(&self) -> Option<payment_methods::ExternalVaultTokenData> {
        None
    }
}

#[cfg(feature = "v2")]
#[async_trait::async_trait]
impl PaymentMethodExt for payment_methods::PaymentMethodCreateData {
    async fn populate_bin_details_for_payment_method(
        &self,
        state: &SessionState,
    ) -> BinEnriched<Self> {
        match self {
            Self::ProxyCard(card) => {
                let card_isin = card.bin_number.clone();
                let card_info = if let Some(card_isin) = card_isin.as_ref() {
                    state
                        .store
                        .get_card_info(card_isin)
                        .await
                        .map_err(|error| services::logger::error!(card_info_error=?error))
                        .ok()
                        .flatten()
                } else {
                    None
                };

                let additional_payment_data_from_bin_lookup = card_info.as_ref().map(|val| {
                    api_models::payments::AdditionalPaymentData::Card(Box::new(
                        api_models::payments::AdditionalCardInfo {
                            card_type: val.card_type.clone(),
                            ..Default::default()
                        },
                    ))
                });

                let payment_method_subtype =
                    Option::<storage_enums::PaymentMethodType>::foreign_from((
                        None,
                        additional_payment_data_from_bin_lookup.as_ref(),
                        Some(storage_enums::PaymentMethod::Card),
                    ));

                if card.card_issuer.is_some()
                    && card.card_network.is_some()
                    && card.card_type.is_some()
                    && card.card_issuing_country.is_some()
                {
                    BinEnriched {
                        data: Self::ProxyCard(card.clone()),
                        payment_method_subtype,
                    }
                } else if card_isin.is_some() {
                    BinEnriched {
                        data: Self::ProxyCard(payment_methods::ProxyCardDetails {
                            card_number: card.card_number.clone(),
                            card_exp_month: card.card_exp_month.clone(),
                            card_exp_year: card.card_exp_year.clone(),
                            card_holder_name: card.card_holder_name.clone(),
                            bin_number: card.bin_number.clone(),
                            last_four: card.last_four.clone(),
                            nick_name: card.nick_name.clone(),
                            card_issuing_country: card_info
                                .as_ref()
                                .and_then(|val| val.card_issuing_country.clone()),
                            card_network: card_info
                                .as_ref()
                                .and_then(|val| val.card_network.clone()),
                            card_issuer: card_info.as_ref().and_then(|val| val.card_issuer.clone()),
                            card_type: card_info.as_ref().and_then(|val| val.card_type.clone()),
                            card_cvc: card.card_cvc.clone(),
                        }),
                        payment_method_subtype,
                    }
                } else {
                    BinEnriched {
                        data: Self::ProxyCard(card.clone()),
                        payment_method_subtype,
                    }
                }
            }
            _ => BinEnriched {
                data: self.clone(),
                payment_method_subtype: None,
            },
        }
    }

    fn convert_to_additional_payment_method_data(
        &self,
    ) -> RouterResult<payment_methods::PaymentMethodsData> {
        match self.clone() {
            Self::ProxyCard(card_details) => Ok(payment_methods::PaymentMethodsData::Card(
                payment_methods::CardDetailsPaymentMethod {
                    last4_digits: card_details.last_four,
                    expiry_month: Some(card_details.card_exp_month),
                    expiry_year: Some(card_details.card_exp_year),
                    card_holder_name: card_details.card_holder_name,
                    nick_name: card_details.nick_name,
                    issuer_country: card_details.card_issuing_country,
                    issuer_country_code: None,
                    card_network: card_details.card_network,
                    card_issuer: card_details.card_issuer,
                    card_type: card_details.card_type,
                    card_isin: card_details.bin_number,
                    saved_to_locker: false,
                    co_badged_card_data: None,
                },
            )),
            Self::Card(card_details) => Ok(payment_methods::PaymentMethodsData::Card(
                payment_methods::CardDetailsPaymentMethod {
                    expiry_month: Some(card_details.card_exp_month),
                    expiry_year: Some(card_details.card_exp_year),
                    card_holder_name: card_details.card_holder_name,
                    nick_name: card_details.nick_name,
                    issuer_country: card_details
                        .card_issuing_country
                        .map(|country| country.to_string()),
                    card_network: card_details.card_network,
                    issuer_country_code: None,
                    card_issuer: card_details.card_issuer,
                    card_type: card_details
                        .card_type
                        .map(|card_type| card_type.to_string()),
                    saved_to_locker: false,
                    card_isin: None,
                    last4_digits: None,
                    co_badged_card_data: None,
                },
            )),
            Self::BankDebit(bank_debit_details) => Ok(
                payment_methods::PaymentMethodsData::BankDebit(bank_debit_details.clone().into()),
            ),
            Self::Wallet(wallet_data) => match wallet_data {
                api::WalletPaymentMethodData::ApplePay(data) => {
                    Ok(payment_methods::PaymentMethodsData::WalletDetails(*data))
                }
                api::WalletPaymentMethodData::GooglePay(data) => {
                    Ok(payment_methods::PaymentMethodsData::WalletDetails(*data))
                }
                api::WalletPaymentMethodData::PayPal(data) => {
                    Ok(payment_methods::PaymentMethodsData::WalletDetails(
                        payment_methods::PaymentMethodDataWalletInfo {
                            last4: None,
                            card_network: None,
                            card_type: None,
                            card_exp_month: None,
                            card_exp_year: None,
                            auth_code: None,
                            email: data.email,
                        },
                    ))
                }
            },
            Self::BankRedirect(bank_redirect_data) => {
                Err(report!(errors::ApiErrorResponse::UnprocessableEntity {
                    message: "Bank redirect payment method data is not supported".to_string()
                }))
                .attach_printable("Bank redirect payment method data is not supported")
            }
        }
    }

    fn get_external_vault_token_data(&self) -> Option<payment_methods::ExternalVaultTokenData> {
        match self.clone() {
            Self::ProxyCard(card_details) => Some(payment_methods::ExternalVaultTokenData {
                tokenized_card_number: card_details.card_number,
            }),
            Self::Card(_) | Self::BankDebit(_) | Self::Wallet(_) | Self::BankRedirect(_) => None,
        }
    }
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn payment_method_intent_create(
    state: &SessionState,
    req: api::PaymentMethodIntentCreate,
    provider: domain::Provider,
    initiator: Option<&domain::Initiator>,
) -> RouterResponse<api::PaymentMethodResponse> {
    let db = &*state.store;
    let merchant_id = provider.get_account().get_id();
    let customer_id = req.customer_id.to_owned();
    let key_manager_state = &(state).into();

    db.find_customer_by_global_id_merchant_id_without_encrypted(
        &customer_id,
        provider.get_account().get_id(),
        provider.get_account().storage_scheme,
    )
    .await
    .to_not_found_response(errors::ApiErrorResponse::CustomerNotFound)
    .attach_printable("Customer not found for the payment method")?;

    let payment_method_billing_address = req
        .billing
        .clone()
        .async_map(|billing| {
            core_utils::create_encrypted_data(
                key_manager_state,
                provider.get_key_store(),
                billing,
                common_utils::type_name!(diesel_models::payment_method::PaymentMethod),
            )
        })
        .await
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to encrypt Payment method billing address")?
        .map(|encoded_address| {
            encoded_address.deserialize_inner_value(|value| value.parse_value("Address"))
        })
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to parse Payment method billing address")?;

    // create pm entry

    let payment_method_id =
        id_type::GlobalPaymentMethodId::generate(&state.conf.cell_information.id);

    let payment_method = create_payment_method_for_intent(
        state,
        req.metadata.clone(),
        &customer_id,
        payment_method_id,
        merchant_id,
        &provider.get_account().organization_id,
        provider.get_key_store(),
        provider.get_account().storage_scheme,
        payment_method_billing_address,
        initiator,
        None,
    )
    .await
    .attach_printable("Failed to add Payment method to DB")?;

    let resp = pm_transforms::generate_payment_method_response(
        &payment_method,
        &None,
        common_enums::StorageType::Persistent,
        None,
        Some(customer_id),
        None,
        None,
        None,
    )?;

    Ok(services::ApplicationResponse::Json(resp))
}

#[cfg(feature = "v2")]
trait PerformFilteringOnEnabledPaymentMethods {
    fn perform_filtering(self) -> FilteredPaymentMethodsEnabled;
}

#[cfg(feature = "v2")]
impl PerformFilteringOnEnabledPaymentMethods
    for hyperswitch_domain_models::merchant_connector_account::FlattenedPaymentMethodsEnabled
{
    fn perform_filtering(self) -> FilteredPaymentMethodsEnabled {
        FilteredPaymentMethodsEnabled(self.payment_methods_enabled)
    }
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn list_payment_methods_for_session(
    state: SessionState,
    platform: domain::Platform,
    profile: domain::Profile,
    payment_method_session_id: id_type::GlobalPaymentMethodSessionId,
) -> RouterResponse<api::PaymentMethodListResponseForSession> {
    let db = &*state.store;

    let payment_method_session = db
        .get_payment_methods_session(
            platform.get_provider().get_key_store(),
            &payment_method_session_id,
        )
        .await
        .change_context(errors::ApiErrorResponse::PaymentMethodNotFound)
        .attach_printable("Unable to find payment method")?;

    let payment_connector_accounts = db
        .list_enabled_connector_accounts_by_profile_id(
            profile.get_id(),
            platform.get_processor().get_key_store(),
            common_enums::ConnectorType::PaymentProcessor,
        )
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("error when fetching merchant connector accounts")?;

    let customer_payment_methods = match payment_method_session.customer_id {
        Some(ref customer_id) => {
            list_customer_payment_methods_core(
                &state,
                platform.get_provider(),
                customer_id,
                payment_method_session.keep_alive,
            )
            .await?
        }
        None => Vec::new(),
    };

    // Create associated payment methods from customer payment methods
    let associated_payment_methods_vec: Vec<common_types::payment_methods::AssociatedPaymentMethods> = customer_payment_methods
        .iter()
        .map(|cpm| {
            common_types::payment_methods::AssociatedPaymentMethods {
                payment_method_token: common_types::payment_methods::AssociatedPaymentMethodTokenType::PaymentMethodSessionToken(cpm.payment_method_token.clone()),
            }
        })
        .collect();

    // Update payment method session with associated payment methods
    let update_payment_method_session = hyperswitch_domain_models::payment_methods::PaymentMethodsSessionUpdateEnum::UpdateAssociatedPaymentMethods {
        associated_payment_methods: Some(associated_payment_methods_vec.clone())
    };

    let updated_payment_method_session = db
        .update_payment_method_session(
            platform.get_provider().get_key_store(),
            &payment_method_session_id,
            update_payment_method_session,
            payment_method_session,
        )
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable(
            "Failed to update payment method session with associated payment methods",
        )?;

    let response =
        hyperswitch_domain_models::merchant_connector_account::FlattenedPaymentMethodsEnabled::from_payment_connectors_list(payment_connector_accounts)
            .perform_filtering()
            .get_required_fields(RequiredFieldsInput::new(state.conf.required_fields.clone()))
            .generate_response_for_session(customer_payment_methods);

    Ok(hyperswitch_domain_models::api::ApplicationResponse::Json(
        response,
    ))
}

#[cfg(all(feature = "v2", feature = "olap"))]
#[instrument(skip_all)]
pub async fn list_saved_payment_methods_for_customer(
    state: SessionState,
    provider: domain::Provider,
    customer_id: id_type::GlobalCustomerId,
    include_new: bool,
) -> RouterResponse<payment_methods::CustomerPaymentMethodsListResponse> {
    let customer_payment_methods =
        list_payment_methods_core(&state, &provider, &customer_id, include_new).await?;

    Ok(hyperswitch_domain_models::api::ApplicationResponse::Json(
        customer_payment_methods,
    ))
}

#[cfg(all(feature = "v2", feature = "olap"))]
#[instrument(skip_all)]
pub async fn get_token_data_for_payment_method(
    state: SessionState,
    provider: domain::Provider,
    profile: domain::Profile,
    request: payment_methods::GetTokenDataRequest,
    payment_method_id: id_type::GlobalPaymentMethodId,
) -> RouterResponse<api::TokenDataResponse> {
    let db = &*state.store;

    let payment_method = db
        .find_payment_method(
            provider.get_key_store(),
            &payment_method_id,
            provider.get_account().storage_scheme,
        )
        .await
        .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;

    let token_data_response = Box::pin(generate_token_data_response(
        &state,
        request,
        profile,
        &payment_method,
    ))
    .await?;

    Ok(hyperswitch_domain_models::api::ApplicationResponse::Json(
        token_data_response,
    ))
}

#[cfg(all(feature = "v2", feature = "olap"))]
#[instrument(skip_all)]
pub async fn generate_token_data_response(
    state: &SessionState,
    request: payment_methods::GetTokenDataRequest,
    profile: domain::Profile,
    payment_method: &domain::PaymentMethod,
) -> RouterResult<api::TokenDataResponse> {
    let token_details = match request.token_type {
        common_enums::TokenDataType::NetworkToken => {
            let is_network_tokenization_enabled = profile.is_network_tokenization_enabled;
            if !is_network_tokenization_enabled {
                return Err(errors::ApiErrorResponse::UnprocessableEntity {
                    message: "Network tokenization is not enabled for this profile".to_string(),
                }
                .into());
            }
            let network_token_requestor_ref_id = payment_method
                .network_token_requestor_reference_id
                .clone()
                .ok_or(errors::ApiErrorResponse::GenericNotFoundError {
                    message: "NetworkTokenRequestorReferenceId is not present".to_string(),
                })?;

            let network_token = network_tokenization::get_token_from_tokenization_service(
                state,
                network_token_requestor_ref_id,
                payment_method,
            )
            .await
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("failed to fetch network token data from tokenization service")?;

            api::TokenDetailsResponse::NetworkTokenDetails(api::NetworkTokenDetailsResponse {
                network_token: network_token.network_token,
                network_token_exp_month: network_token.network_token_exp_month,
                network_token_exp_year: network_token.network_token_exp_year,
                cryptogram: network_token.cryptogram,
                card_issuer: network_token.card_issuer,
                card_network: network_token.card_network,
                card_type: network_token.card_type,
                card_issuing_country: network_token.card_issuing_country,
                bank_code: network_token.bank_code,
                card_holder_name: network_token.card_holder_name,
                nick_name: network_token.nick_name,
                eci: network_token.eci,
            })
        }
        common_enums::TokenDataType::SingleUseToken
        | common_enums::TokenDataType::MultiUseToken => {
            return Err(errors::ApiErrorResponse::UnprocessableEntity {
                message: "Token type not supported".to_string(),
            }
            .into());
        }
    };

    Ok(api::TokenDataResponse {
        payment_method_id: payment_method.id.clone(),
        token_type: request.token_type,
        token_details,
    })
}

#[cfg(all(feature = "v2", feature = "olap"))]
#[instrument(skip_all)]
pub async fn get_total_saved_payment_methods_for_merchant(
    state: SessionState,
    provider: domain::Provider,
) -> RouterResponse<api::TotalPaymentMethodCountResponse> {
    let total_payment_method_count = get_total_payment_method_count_core(&state, &provider).await?;

    Ok(hyperswitch_domain_models::api::ApplicationResponse::Json(
        total_payment_method_count,
    ))
}

#[cfg(feature = "v2")]
/// Container for the inputs required for the required fields
struct RequiredFieldsInput {
    required_fields_config: settings::RequiredFields,
}

#[cfg(feature = "v2")]
impl RequiredFieldsInput {
    fn new(required_fields_config: settings::RequiredFields) -> Self {
        Self {
            required_fields_config,
        }
    }
}

#[cfg(feature = "v2")]
/// Container for the filtered payment methods
struct FilteredPaymentMethodsEnabled(
    Vec<hyperswitch_domain_models::merchant_connector_account::PaymentMethodsEnabledForConnector>,
);

#[cfg(feature = "v2")]
trait GetRequiredFields {
    fn get_required_fields(
        &self,
        payment_method_enabled: &hyperswitch_domain_models::merchant_connector_account::PaymentMethodsEnabledForConnector,
    ) -> Option<&settings::RequiredFieldFinal>;
}

#[cfg(feature = "v2")]
impl GetRequiredFields for settings::RequiredFields {
    fn get_required_fields(
        &self,
        payment_method_enabled: &hyperswitch_domain_models::merchant_connector_account::PaymentMethodsEnabledForConnector,
    ) -> Option<&settings::RequiredFieldFinal> {
        self.0
            .get(&payment_method_enabled.payment_method)
            .and_then(|required_fields_for_payment_method| {
                required_fields_for_payment_method.0.get(
                    &payment_method_enabled
                        .payment_methods_enabled
                        .payment_method_subtype,
                )
            })
            .map(|connector_fields| &connector_fields.fields)
            .and_then(|connector_hashmap| connector_hashmap.get(&payment_method_enabled.connector))
    }
}

#[cfg(feature = "v2")]
impl FilteredPaymentMethodsEnabled {
    fn get_required_fields(
        self,
        input: RequiredFieldsInput,
    ) -> RequiredFieldsForEnabledPaymentMethodTypes {
        let required_fields_config = input.required_fields_config;

        let required_fields_info = self
            .0
            .into_iter()
            .map(|payment_methods_enabled| {
                let required_fields =
                    required_fields_config.get_required_fields(&payment_methods_enabled);

                let required_fields = required_fields
                    .map(|required_fields| {
                        let common_required_fields = required_fields
                            .common
                            .iter()
                            .flatten()
                            .map(ToOwned::to_owned);

                        // Collect mandate required fields because this is for zero auth mandates only
                        let mandate_required_fields = required_fields
                            .mandate
                            .iter()
                            .flatten()
                            .map(ToOwned::to_owned);

                        // Combine both common and mandate required fields
                        common_required_fields
                            .chain(mandate_required_fields)
                            .collect::<Vec<_>>()
                    })
                    .unwrap_or_default();

                RequiredFieldsForEnabledPaymentMethod {
                    required_fields,
                    payment_method_type: payment_methods_enabled.payment_method,
                    payment_method_subtype: payment_methods_enabled
                        .payment_methods_enabled
                        .payment_method_subtype,
                }
            })
            .collect();

        RequiredFieldsForEnabledPaymentMethodTypes(required_fields_info)
    }
}

#[cfg(feature = "v2")]
/// Element container to hold the filtered payment methods with required fields
struct RequiredFieldsForEnabledPaymentMethod {
    required_fields: Vec<payment_methods::RequiredFieldInfo>,
    payment_method_subtype: common_enums::PaymentMethodType,
    payment_method_type: common_enums::PaymentMethod,
}

#[cfg(feature = "v2")]
/// Container to hold the filtered payment methods enabled with required fields
struct RequiredFieldsForEnabledPaymentMethodTypes(Vec<RequiredFieldsForEnabledPaymentMethod>);

#[cfg(feature = "v2")]
impl RequiredFieldsForEnabledPaymentMethodTypes {
    fn generate_response_for_session(
        self,
        customer_payment_methods: Vec<payment_methods::CustomerPaymentMethodResponseItem>,
    ) -> payment_methods::PaymentMethodListResponseForSession {
        let response_payment_methods = self
            .0
            .into_iter()
            .map(
                |payment_methods_enabled| payment_methods::ResponsePaymentMethodTypes {
                    payment_method_type: payment_methods_enabled.payment_method_type,
                    payment_method_subtype: payment_methods_enabled.payment_method_subtype,
                    required_fields: payment_methods_enabled.required_fields,
                    extra_information: None,
                },
            )
            .collect();

        payment_methods::PaymentMethodListResponseForSession {
            payment_methods_enabled: response_payment_methods,
            customer_payment_methods,
        }
    }
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
#[allow(clippy::too_many_arguments)]
pub async fn create_payment_method_for_intent(
    state: &SessionState,
    metadata: Option<common_utils::pii::SecretSerdeValue>,
    customer_id: &id_type::GlobalCustomerId,
    payment_method_id: id_type::GlobalPaymentMethodId,
    merchant_id: &id_type::MerchantId,
    organization_id: &id_type::OrganizationId,
    key_store: &domain::MerchantKeyStore,
    storage_scheme: enums::MerchantStorageScheme,
    payment_method_billing_address: Option<
        Encryptable<hyperswitch_domain_models::address::Address>,
    >,
    initiator: Option<&domain::Initiator>,
    auxiliary_fingerprint_id: Option<String>,
) -> CustomResult<domain::PaymentMethod, errors::ApiErrorResponse> {
    use josekit::jwe::zip::deflate::DeflateJweCompression::Def;

    let db = &*state.store;

    let current_time = common_utils::date_time::now();

    let response = db
        .insert_payment_method(
            key_store,
            domain::PaymentMethod {
                customer_id: Some(customer_id.to_owned()),
                merchant_id: merchant_id.to_owned(),
                id: payment_method_id,
                locker_id: None,
                payment_method_type: None,
                payment_method_subtype: None,
                payment_method_data: None,
                connector_mandate_details: None,
                customer_acceptance: None,
                client_secret: None,
                status: enums::PaymentMethodStatus::AwaitingData,
                network_transaction_id: None,
                network_transaction_link_id: None,
                created_at: current_time,
                last_modified: current_time,
                last_used_at: current_time,
                payment_method_billing_address,
                updated_by: None,
                version: common_types::consts::API_VERSION,
                locker_fingerprint_id: None,
                network_token_locker_id: None,
                network_token_payment_method_data: None,
                network_token_requestor_reference_id: None,
                external_vault_source: None,
                external_vault_token_data: None,
                vault_type: None,
                created_by: initiator.and_then(|initiator| initiator.to_created_by()),
                last_modified_by: initiator.and_then(|initiator| initiator.to_created_by()),
                customer_details: None,
                network_tokenization_data: None,
                auxiliary_fingerprint_id,
                compatibility_updated_at: None,
            },
            storage_scheme,
            Some(payment_method_modular_backward_compat_action(
                state,
                organization_id,
            )),
        )
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to add payment method in db")?;

    Ok(response)
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
#[allow(clippy::too_many_arguments)]
pub async fn construct_payment_method_object(
    _metadata: Option<common_utils::pii::SecretSerdeValue>,
    customer_id: &Option<id_type::GlobalCustomerId>,
    payment_method_id: id_type::GlobalPaymentMethodId,
    merchant_id: &id_type::MerchantId,
    payment_method_billing_address: Option<
        Encryptable<hyperswitch_domain_models::address::Address>,
    >,
    state: &SessionState,
    key_store: &domain::MerchantKeyStore,
    payment_method_vaulting_data: Option<&domain::PaymentMethodVaultingData>,
    payment_method_type: Option<common_enums::PaymentMethod>,
    payment_method_subtype: Option<common_enums::PaymentMethodType>,
    locker_id: Option<domain::VaultId>,
    locker_fingerprint_id: Option<String>,
    external_vault_source: Option<id_type::MerchantConnectorAccountId>,
    initiator: Option<&domain::Initiator>,
) -> RouterResult<domain::PaymentMethod> {
    let current_time = common_utils::date_time::now();

    let encrypted_payment_method_data = payment_method_vaulting_data
        .map(|payment_method_vaulting_data| payment_method_vaulting_data.get_payment_methods_data())
        .async_map(|payment_method_details| async {
            let key_manager_state = &(state).into();

            core_utils::create_encrypted_data(
                key_manager_state,
                key_store,
                payment_method_details,
                common_utils::type_name!(diesel_models::payment_method::PaymentMethod),
            )
            .await
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Unable to encrypt Payment method data")
        })
        .await
        .transpose()?
        .map(|encoded_data| {
            encoded_data.deserialize_inner_value(|value| value.parse_value("PaymentMethodsData"))
        })
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to parse Payment method data")?;

    Ok(domain::PaymentMethod {
        customer_id: customer_id.clone(),
        merchant_id: merchant_id.to_owned(),
        id: payment_method_id,
        locker_id,
        payment_method_type,
        payment_method_subtype,
        payment_method_data: encrypted_payment_method_data,
        connector_mandate_details: None,
        customer_acceptance: None,
        client_secret: None,
        status: enums::PaymentMethodStatus::New,
        network_transaction_id: None,
        network_transaction_link_id: None,
        created_at: current_time,
        last_modified: current_time,
        last_used_at: current_time,
        payment_method_billing_address,
        updated_by: None,
        version: common_types::consts::API_VERSION,
        locker_fingerprint_id,
        network_token_locker_id: None,
        network_token_payment_method_data: None,
        network_token_requestor_reference_id: None,
        external_vault_source,
        external_vault_token_data: None,
        vault_type: None,
        created_by: initiator.and_then(|initiator| initiator.to_created_by()),
        last_modified_by: initiator.and_then(|initiator| initiator.to_created_by()),
        customer_details: None,
        network_tokenization_data: None,
        auxiliary_fingerprint_id: None,
        compatibility_updated_at: None,
    })
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
#[allow(clippy::too_many_arguments)]
pub async fn create_payment_method_for_confirm(
    state: &SessionState,
    customer_id: &id_type::GlobalCustomerId,
    payment_method_id: id_type::GlobalPaymentMethodId,
    external_vault_source: Option<id_type::MerchantConnectorAccountId>,
    merchant_id: &id_type::MerchantId,
    organization_id: &id_type::OrganizationId,
    key_store: &domain::MerchantKeyStore,
    storage_scheme: enums::MerchantStorageScheme,
    payment_method_type: storage_enums::PaymentMethod,
    payment_method_subtype: Option<storage_enums::PaymentMethodType>,
    encrypted_payment_method_billing_address: Option<
        Encryptable<hyperswitch_domain_models::address::Address>,
    >,
    encrypted_payment_method_data: Option<
        Encryptable<domain::payment_method_data::PaymentMethodsData>,
    >,
    encrypted_external_vault_token_data: Option<
        Encryptable<payment_methods::ExternalVaultTokenData>,
    >,
    vault_type: Option<common_enums::VaultType>,
    initiator: Option<&domain::Initiator>,
    status: enums::PaymentMethodStatus,
) -> CustomResult<domain::PaymentMethod, errors::ApiErrorResponse> {
    let db = &*state.store;
    let current_time = common_utils::date_time::now();

    let response = db
        .insert_payment_method(
            key_store,
            domain::PaymentMethod {
                customer_id: Some(customer_id.to_owned()),
                merchant_id: merchant_id.to_owned(),
                id: payment_method_id,
                locker_id: None,
                payment_method_type: Some(payment_method_type),
                payment_method_subtype,
                payment_method_data: encrypted_payment_method_data,
                connector_mandate_details: None,
                customer_acceptance: None,
                client_secret: None,
                status: enums::PaymentMethodStatus::New,
                network_transaction_id: None,
                network_transaction_link_id: None,
                created_at: current_time,
                last_modified: current_time,
                last_used_at: current_time,
                payment_method_billing_address: encrypted_payment_method_billing_address,
                updated_by: None,
                version: common_types::consts::API_VERSION,
                locker_fingerprint_id: None,
                network_token_locker_id: None,
                network_token_payment_method_data: None,
                network_token_requestor_reference_id: None,
                external_vault_source,
                external_vault_token_data: encrypted_external_vault_token_data,
                vault_type,
                created_by: initiator.and_then(|initiator| initiator.to_created_by()),
                last_modified_by: initiator.and_then(|initiator| initiator.to_created_by()),
                customer_details: None,
                network_tokenization_data: None,
                auxiliary_fingerprint_id: None,
                compatibility_updated_at: None,
            },
            storage_scheme,
            Some(payment_method_modular_backward_compat_action(
                state,
                organization_id,
            )),
        )
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to add payment method in db")?;

    Ok(response)
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
#[allow(clippy::too_many_arguments)]
pub async fn get_external_vault_token(
    state: &SessionState,
    key_store: &domain::MerchantKeyStore,
    storage_scheme: enums::MerchantStorageScheme,
    payment_token: String,
    vault_token: domain::VaultToken,
    payment_method_type: &storage_enums::PaymentMethod,
) -> CustomResult<domain::ExternalVaultPaymentMethodData, errors::ApiErrorResponse> {
    let db = &*state.store;

    let pm_token_data = utils::retrieve_payment_token_data(state, payment_token).await?;

    let payment_method_id = match pm_token_data {
        storage::PaymentTokenData::PermanentCard(card_token_data) => {
            card_token_data.payment_method_id
        }
        storage::PaymentTokenData::BankDebit(bank_debit_token_data) => {
            bank_debit_token_data.payment_method_id
        }
        storage::PaymentTokenData::TemporaryGeneric(_) => {
            Err(errors::ApiErrorResponse::NotImplemented {
                message: errors::NotImplementedMessage::Reason(
                    "TemporaryGeneric Token not implemented".to_string(),
                ),
            })?
        }
        storage::PaymentTokenData::AuthBankDebit(_) => {
            Err(errors::ApiErrorResponse::NotImplemented {
                message: errors::NotImplementedMessage::Reason(
                    "AuthBankDebit Token not implemented".to_string(),
                ),
            })?
        }
        storage::PaymentTokenData::TemporaryCardToken(_) => {
            Err(errors::ApiErrorResponse::NotImplemented {
                message: errors::NotImplementedMessage::Reason(
                    "TemporaryCardToken does not have a vaulted payment method".to_string(),
                ),
            })?
        }
    };

    let payment_method = db
        .find_payment_method(key_store, &payment_method_id, storage_scheme)
        .await
        .change_context(errors::ApiErrorResponse::PaymentMethodNotFound)
        .attach_printable("Payment method not found")?;

    let external_vault_token_data = payment_method
        .external_vault_token_data
        .clone()
        .map(Encryptable::into_inner)
        .ok_or(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Missing vault token data")?;

    let decrypted_addtional_payment_method_data = payment_method
        .payment_method_data
        .clone()
        .map(Encryptable::into_inner)
        .ok_or(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to convert payment method data")?;

    convert_from_saved_payment_method_data(
        decrypted_addtional_payment_method_data,
        external_vault_token_data,
        vault_token,
    )
    .attach_printable("Failed to convert payment method data")
}

#[cfg(feature = "v2")]
fn convert_from_saved_payment_method_data(
    vault_additional_data: domain::payment_method_data::PaymentMethodsData,
    external_vault_token_data: payment_methods::ExternalVaultTokenData,
    vault_token: domain::VaultToken,
) -> RouterResult<domain::ExternalVaultPaymentMethodData> {
    match vault_additional_data {
        domain::payment_method_data::PaymentMethodsData::Card(card_details) => {
            Ok(domain::ExternalVaultPaymentMethodData::Card(Box::new(
                domain::ExternalVaultCard {
                    card_number: external_vault_token_data.tokenized_card_number,
                    card_exp_month: card_details.expiry_month.ok_or(
                        errors::ApiErrorResponse::MissingRequiredField {
                            field_name: "card_details.expiry_month",
                        },
                    )?,
                    card_exp_year: card_details.expiry_year.ok_or(
                        errors::ApiErrorResponse::MissingRequiredField {
                            field_name: "card_details.expiry_year",
                        },
                    )?,
                    card_holder_name: card_details.card_holder_name,
                    bin_number: card_details.card_isin,
                    last_four: card_details.last4_digits,
                    nick_name: card_details.nick_name,
                    card_issuing_country: card_details.issuer_country,
                    card_network: card_details.card_network,
                    card_issuer: card_details.card_issuer,
                    card_type: card_details.card_type,
                    card_cvc: vault_token.card_cvc,
                    co_badged_card_data: None, // Co-badged data is not supported in external vault
                    bank_code: None,           // Bank code is not stored in external vault
                },
            )))
        }
        domain::payment_method_data::PaymentMethodsData::BankDetails(_)
        | domain::payment_method_data::PaymentMethodsData::BankDebit(_)
        | domain::payment_method_data::PaymentMethodsData::WalletDetails(_)
        | domain::payment_method_data::PaymentMethodsData::NetworkToken(_) => {
            Err(errors::ApiErrorResponse::UnprocessableEntity {
                message: "External vaulting is not supported for this payment method type"
                    .to_string(),
            }
            .into())
        }
    }
}

#[cfg(feature = "v2")]
/// Update the connector_mandate_details of the payment method with
/// new token details for the payment
fn create_connector_token_details_update(
    token_details: payment_methods::ConnectorTokenDetails,
    payment_method: &domain::PaymentMethod,
) -> mandates::CommonMandateReference {
    let connector_id = token_details.connector_id.clone();

    let reference_record = mandates::ConnectorTokenReferenceRecord::foreign_from(token_details);

    let connector_token_details = payment_method.connector_mandate_details.clone();

    match connector_token_details {
        Some(mut connector_mandate_reference) => {
            connector_mandate_reference
                .insert_payment_token_reference_record(&connector_id, reference_record);

            connector_mandate_reference
        }
        None => {
            let reference_record_hash_map =
                std::collections::HashMap::from([(connector_id, reference_record)]);
            let payments_mandate_reference =
                mandates::PaymentsTokenReference(reference_record_hash_map);
            mandates::CommonMandateReference {
                payments: Some(payments_mandate_reference),
                payouts: None,
            }
        }
    }
}

#[cfg(feature = "v2")]
#[allow(clippy::too_many_arguments)]
pub async fn create_pm_additional_data_update(
    pmd: Option<&domain::PaymentMethodVaultingData>,
    state: &SessionState,
    key_store: &domain::MerchantKeyStore,
    vault_id: Option<String>,
    vault_fingerprint_id: Option<String>,
    payment_method: &domain::PaymentMethod,
    connector_token_details: Option<payment_methods::ConnectorTokenDetails>,
    network_transaction_id: Option<Secret<String>>,
    nt_data: Option<NetworkTokenPaymentMethodDetails>,
    payment_method_type: Option<common_enums::PaymentMethod>,
    payment_method_subtype: Option<common_enums::PaymentMethodType>,
    external_vault_source: Option<id_type::MerchantConnectorAccountId>,
    status: Option<storage_enums::PaymentMethodStatus>,
    initiator: Option<&domain::Initiator>,
) -> RouterResult<storage::PaymentMethodUpdate> {
    let encrypted_payment_method_data = pmd
        .map(|payment_method_vaulting_data| payment_method_vaulting_data.get_payment_methods_data())
        .async_map(|payment_method_details| async {
            let key_manager_state = &(state).into();

            core_utils::create_encrypted_data(
                key_manager_state,
                key_store,
                payment_method_details,
                common_utils::type_name!(diesel_models::payment_method::PaymentMethod),
            )
            .await
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Unable to encrypt Payment method data")
        })
        .await
        .transpose()?
        .map(|encoded_data| {
            encoded_data.deserialize_inner_value(|value| {
                value.parse_value::<domain::PaymentMethodsData>("PaymentMethodsData")
            })
        })
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to parse Payment method data")?
        .map(From::from);

    let connector_mandate_details_update = connector_token_details
        .map(|connector_token| {
            create_connector_token_details_update(connector_token, payment_method)
        })
        .map(From::from);

    let pm_update = storage::PaymentMethodUpdate::GenericUpdate {
        // A new payment method is created with inactive state
        // It will be marked active after payment succeeds
        status,
        locker_id: vault_id,
        payment_method_type_v2: payment_method_type,
        payment_method_subtype,
        payment_method_data: encrypted_payment_method_data,
        network_token_requestor_reference_id: nt_data
            .clone()
            .map(|data| data.network_token_requestor_reference_id),
        network_token_locker_id: nt_data.clone().map(|data| data.network_token_locker_id),
        network_token_payment_method_data: nt_data.map(|data| data.network_token_pmd.into()),
        connector_mandate_details: Box::new(connector_mandate_details_update),
        locker_fingerprint_id: vault_fingerprint_id,
        external_vault_source,
        network_transaction_id,
        network_transaction_link_id: None,
        last_modified_by: initiator
            .and_then(|initiator| initiator.to_created_by())
            .map(|last_modified_by| last_modified_by.to_string()),
    };

    Ok(pm_update)
}

#[cfg(feature = "v2")]
async fn build_legacy_locker_add_req(
    pmd: &domain::PaymentMethodVaultingData,
    merchant_id: id_type::MerchantId,
    customer_id: &id_type::CustomerId,
    existing_vault_id: Option<&domain::VaultId>,
    ttl: i64,
) -> RouterResult<pm_transforms::StoreLockerReq> {
    match pmd {
        domain::PaymentMethodVaultingData::Card(card_detail) => Ok(
            pm_transforms::StoreLockerReq::LockerCard(pm_transforms::StoreCardReq {
                merchant_id,
                merchant_customer_id: customer_id.to_owned(),
                requestor_card_reference: existing_vault_id
                    .map(|vid| vid.get_string_repr().to_owned()),
                card: payment_methods::Card {
                    card_number: card_detail.card_number.clone(),
                    name_on_card: card_detail.card_holder_name.clone(),
                    card_exp_month: card_detail.card_exp_month.clone(),
                    card_exp_year: card_detail.card_exp_year.clone(),
                    card_brand: card_detail.card_network.as_ref().map(ToString::to_string),
                    card_isin: None,
                    nick_name: card_detail.nick_name.as_ref().map(Secret::peek).cloned(),
                },
                ttl,
            }),
        ),
        domain::PaymentMethodVaultingData::NetworkToken(network_token) => Ok(
            pm_transforms::StoreLockerReq::LockerCard(pm_transforms::StoreCardReq {
                merchant_id,
                merchant_customer_id: customer_id.to_owned(),
                requestor_card_reference: existing_vault_id
                    .map(|vid| vid.get_string_repr().to_owned()),
                card: payment_methods::Card {
                    card_number: network_token.network_token.clone().into(),
                    name_on_card: network_token.card_holder_name.clone(),
                    card_exp_month: network_token.network_token_exp_month.clone(),
                    card_exp_year: network_token.network_token_exp_year.clone(),
                    card_brand: network_token.card_network.as_ref().map(ToString::to_string),
                    card_isin: None,
                    nick_name: network_token.nick_name.as_ref().map(Secret::peek).cloned(),
                },
                ttl,
            }),
        ),
        domain::PaymentMethodVaultingData::CardNumber(_)
        | domain::PaymentMethodVaultingData::BankDebit(_)
        | domain::PaymentMethodVaultingData::Wallet(_) => {
            Err(report!(errors::ApiErrorResponse::NotSupported {
                message: "Legacy locker only supports Card and NetworkToken".into(),
            }))
        }
    }
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn vault_payment_method_internal(
    state: &SessionState,
    pmd: &domain::PaymentMethodVaultingData,
    platform: &domain::Platform,
    existing_vault_id: Option<domain::VaultId>,
    fingerprint_id_from_vault: Option<String>,
    customer_id: &id_type::GlobalCustomerId,
    write_mode: Option<pm_types::WriteMode>,
) -> RouterResult<pm_types::AddVaultResponse> {
    let db = &*state.store;

    let mut resp_from_vault = vault::add_payment_method_to_vault(
        state,
        platform,
        pmd,
        existing_vault_id,
        customer_id,
        write_mode,
    )
    .await
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to add payment method in vault")?;

    // add fingerprint_id to the response
    resp_from_vault.fingerprint_id = fingerprint_id_from_vault;

    Ok(resp_from_vault)
}

#[cfg(feature = "v2")]
pub async fn add_payment_method_to_legacy_locker(
    state: &SessionState,
    platform: &domain::Platform,
    pmd: &domain::PaymentMethodVaultingData,
    existing_vault_id: Option<domain::VaultId>,
    customer_id: &id_type::GlobalCustomerId,
) -> CustomResult<pm_transforms::StoreCardRespPayload, errors::VaultError> {
    let cust_id =
        id_type::CustomerId::try_from(Cow::from(customer_id.get_string_repr().to_owned()))
            .change_context(errors::VaultError::RequestEncodingFailed)
            .attach_printable("Failed to convert to CustomerId")?;

    let merchant_id = platform.get_provider().get_account().get_id();

    // Transform V2 PaymentMethodVaultingData → V1 StoreLockerReq
    let payload = build_legacy_locker_add_req(
        pmd,
        merchant_id.to_owned(),
        &cust_id,
        existing_vault_id.as_ref(),
        state.conf.locker.ttl_for_storage_in_secs,
    )
    .await
    .change_context(errors::VaultError::RequestEncodingFailed)
    .attach_printable("Failed to build request for legacy locker")?;

    // Call V1 locker
    let v1_resp = cards::add_card_to_vault(state, &payload, &cust_id)
        .await
        .attach_printable("Failed to add payment method to legacy locker")?;

    Ok(v1_resp)
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn vault_payment_method_external(
    state: &SessionState,
    pmd: &domain::PaymentMethodCustomVaultingData,
    merchant_account: &domain::MerchantAccount,
    merchant_connector_account: domain::MerchantConnectorAccountTypeDetails,
) -> RouterResult<pm_types::AddVaultResponse> {
    let merchant_connector_account = match &merchant_connector_account {
        domain::MerchantConnectorAccountTypeDetails::MerchantConnectorAccount(mca) => {
            Ok(mca.as_ref())
        }
        domain::MerchantConnectorAccountTypeDetails::MerchantConnectorDetails(_) => {
            Err(report!(errors::ApiErrorResponse::InternalServerError)
                .attach_printable("MerchantConnectorDetails not supported for vault operations"))
        }
    }?;

    let router_data = core_utils::construct_vault_router_data(
        state,
        merchant_account.get_id(),
        merchant_connector_account,
        Some(pmd.clone()),
        None,
        None,
        None,
        None,
    )
    .await?;

    let mut old_router_data = VaultConnectorFlowData::to_old_router_data(router_data)
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable(
            "Cannot construct router data for making the external vault insert api call",
        )?;

    let connector_name = merchant_connector_account.get_connector_name_as_string(); // always get the connector name from this call

    let connector_data = api::ConnectorData::get_external_vault_connector_by_name(
        &state.conf.connectors,
        connector_name,
        api::GetToken::Connector,
        Some(merchant_connector_account.get_id()),
    )
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to get the connector data")?;

    access_token::create_access_token(
        state,
        &connector_data,
        merchant_account.get_id(),
        &mut old_router_data,
    )
    .await?;

    if old_router_data.response.is_ok() {
        let connector_integration: services::BoxedVaultConnectorIntegrationInterface<
            ExternalVaultInsertFlow,
            types::VaultRequestData,
            types::VaultResponseData,
        > = connector_data.connector.get_connector_integration();

        let router_data_resp = services::execute_connector_processing_step(
            state,
            connector_integration,
            &old_router_data,
            payments_core::CallConnectorAction::Trigger,
            None,
            None,
        )
        .await
        .to_vault_failed_response()?;

        get_vault_response_for_insert_payment_method_data(router_data_resp)
    } else {
        logger::error!(
            "Error vaulting payment method: {:?}",
            old_router_data.response
        );
        Err(report!(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to create access token for external vault"))
    }
}

pub fn get_payment_method_custom_data(
    payment_method_vaulting_data: hyperswitch_domain_models::vault::PaymentMethodVaultingData,
    fields_to_tokenize: Option<Vec<diesel_models::business_profile::VaultTokenField>>,
) -> RouterResult<hyperswitch_domain_models::vault::PaymentMethodCustomVaultingData> {
    match fields_to_tokenize {
        Some(fields) => {
            let keys_set: Vec<String> = fields
                .iter()
                .map(|field| field.token_type.to_string())
                .collect();

            if keys_set.is_empty() {
                // edge case where no token to vault is present
                Ok(payment_method_vaulting_data.try_into()?)
            } else {
                match payment_method_vaulting_data {
                    hyperswitch_domain_models::vault::PaymentMethodVaultingData::Card(card_details) => {
                        let mut json_data = serde_json::to_value(card_details)
                            .map_err(|_| {
                                logger::error!("Error Parsing the CardDetail to Value");
                                errors::ApiErrorResponse::InternalServerError
                            })?
                            .as_object()
                            .cloned()
                            .ok_or(errors::ApiErrorResponse::InternalServerError)
                            .attach_printable("Failed to parse Value to Object")?;

                        json_data.retain(|key, _value| keys_set.contains(key));

                        let custom_card_detail: hyperswitch_domain_models::vault::CardCustomData = serde_json::from_value(
                            serde_json::Value::Object(json_data)
                        )
                            .map_err(|_| {
                                logger::error!("Error Parsing the Value to CardCustomData");
                                errors::ApiErrorResponse::InternalServerError
                            })?;
                        Ok(hyperswitch_domain_models::vault::PaymentMethodCustomVaultingData::CardData(custom_card_detail))
                    }
                    hyperswitch_domain_models::vault::PaymentMethodVaultingData::NetworkToken(network_token_details) => {
                        let mut json_data = serde_json::to_value(network_token_details)
                            .map_err(|_| {
                                logger::error!("Error Parsing the NetworkTokenDetails to Value");
                                errors::ApiErrorResponse::InternalServerError
                            })?
                            .as_object()
                            .cloned()
                            .ok_or(errors::ApiErrorResponse::InternalServerError)
                            .attach_printable("Failed to parse Value to Object")?;

                        json_data.retain(|key, _value| keys_set.contains(key));

                        let custom_network_token_detail: hyperswitch_domain_models::vault::NetworkTokenCustomData = serde_json::from_value(
                            serde_json::Value::Object(json_data)
                        )
                            .map_err(|_| {
                                logger::error!("Error Parsing the Value to NetworkTokenCustomData");
                                errors::ApiErrorResponse::InternalServerError
                            })?;
                        Ok(hyperswitch_domain_models::vault::PaymentMethodCustomVaultingData::NetworkTokenData(custom_network_token_detail))
                    }
                    hyperswitch_domain_models::vault::PaymentMethodVaultingData::CardNumber(_) => {
                        Err(errors::ApiErrorResponse::InternalServerError)
                            .attach_printable("Unexpected Behaviour, Card Number variant is not supported for Custom Tokenization")?
                    }
                    hyperswitch_domain_models::vault::PaymentMethodVaultingData::BankDebit(_) => {
                        Err(errors::ApiErrorResponse::InternalServerError)
                            .attach_printable("Unexpected Behaviour, Bank Debit variant is not supported for Custom Tokenization")?
                    }
                    hyperswitch_domain_models::vault::PaymentMethodVaultingData::Wallet(_) => {
                        Err(errors::ApiErrorResponse::InternalServerError)
                            .attach_printable("Unexpected Behaviour, Wallet variant is not supported for Custom Tokenization")?
                    }
                }
            }
        }
        // default case, populate data one to one
        None => Ok(payment_method_vaulting_data.try_into()?),
    }
}

#[cfg(feature = "v1")]
#[instrument(skip_all)]
pub async fn vault_payment_method_external_v1(
    state: &SessionState,
    pmd: &hyperswitch_domain_models::vault::PaymentMethodCustomVaultingData,
    merchant_account: &domain::MerchantAccount,
    merchant_connector_account: hyperswitch_domain_models::merchant_connector_account::MerchantConnectorAccount,
    should_generate_multiple_tokens: Option<bool>,
) -> RouterResult<pm_types::AddVaultResponse> {
    let router_data = core_utils::construct_vault_router_data(
        state,
        merchant_account.get_id(),
        &merchant_connector_account,
        Some(pmd.clone()),
        None,
        None,
        should_generate_multiple_tokens,
        None,
    )
    .await?;

    let mut old_router_data = VaultConnectorFlowData::to_old_router_data(router_data)
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable(
            "Cannot construct router data for making the external vault insert api call",
        )?;

    let connector_name = merchant_connector_account.get_connector_name_as_string();

    let connector_data = api::ConnectorData::get_external_vault_connector_by_name(
        &state.conf.connectors,
        connector_name,
        api::GetToken::Connector,
        Some(merchant_connector_account.get_id()),
    )
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to get the connector data")?;

    access_token::create_access_token(
        state,
        &connector_data,
        merchant_account.get_id(),
        &mut old_router_data,
    )
    .await?;

    if old_router_data.response.is_ok() {
        let connector_integration: services::BoxedVaultConnectorIntegrationInterface<
            ExternalVaultInsertFlow,
            types::VaultRequestData,
            types::VaultResponseData,
        > = connector_data.connector.get_connector_integration();

        let router_data_resp = services::execute_connector_processing_step(
            state,
            connector_integration,
            &old_router_data,
            payments_core::CallConnectorAction::Trigger,
            None,
            None,
        )
        .await
        .to_vault_failed_response()?;

        get_vault_response_for_insert_payment_method_data(router_data_resp)
    } else {
        logger::error!(
            "Error vaulting payment method: {:?}",
            old_router_data.response
        );
        Err(report!(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to create access token for external vault"))
    }
}

pub fn get_vault_response_for_insert_payment_method_data<F>(
    router_data: VaultRouterData<F>,
) -> RouterResult<pm_types::AddVaultResponse> {
    match router_data.response {
        Ok(response) => match response {
            types::VaultResponseData::ExternalVaultInsertResponse {
                connector_vault_id,
                fingerprint_id,
            } => {
                #[cfg(feature = "v2")]
                let vault_id = domain::VaultId::generate(connector_vault_id.get_single_vault_id()?);
                #[cfg(not(feature = "v2"))]
                let vault_id = connector_vault_id;

                Ok(pm_types::AddVaultResponse {
                    vault_id,
                    fingerprint_id: Some(fingerprint_id),
                    entity_id: None,
                })
            }
            types::VaultResponseData::ExternalVaultRetrieveResponse { .. }
            | types::VaultResponseData::ExternalVaultDeleteResponse { .. }
            | types::VaultResponseData::ExternalVaultCreateResponse { .. } => {
                Err(report!(errors::ApiErrorResponse::InternalServerError)
                    .attach_printable("Invalid Vault Response"))
            }
        },
        Err(err) => {
            logger::error!("Error vaulting payment method: {:?}", err);
            Err(report!(errors::ApiErrorResponse::InternalServerError)
                .attach_printable("Failed to vault payment method"))
        }
    }
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
#[allow(clippy::too_many_arguments)]
pub async fn vault_payment_method(
    state: &SessionState,
    pmd: &domain::PaymentMethodVaultingData,
    platform: &domain::Platform,
    profile: &domain::Profile,
    existing_vault_id: Option<domain::VaultId>,
    fingerprint_id_from_vault: Option<String>,
    customer_id: &id_type::GlobalCustomerId,
    write_mode: Option<pm_types::WriteMode>,
) -> RouterResult<(
    pm_types::AddVaultResponse,
    Option<id_type::MerchantConnectorAccountId>,
)> {
    let external_vault_profile =
        payments_core::helpers::resolve_provider_profile(state, platform, profile).await?;
    let is_external_vault_enabled = external_vault_profile.is_external_vault_enabled();

    match is_external_vault_enabled {
        true => {
            let (external_vault_source, vault_token_selector) = external_vault_profile
                .external_vault_connector_details
                .clone()
                .map(|connector_details| {
                    (
                        connector_details.vault_connector_id.clone(),
                        connector_details.vault_token_selector.clone(),
                    )
                })
                .ok_or(errors::ApiErrorResponse::InternalServerError)
                .attach_printable("mca_id not present for external vault")?;

            let merchant_connector_account =
                domain::MerchantConnectorAccountTypeDetails::MerchantConnectorAccount(Box::new(
                    payments_core::helpers::get_provider_mca_v2(
                        state,
                        platform.get_provider(),
                        Some(&external_vault_source),
                    )
                    .await
                    .attach_printable(
                        "failed to fetch merchant connector account for external vault insert",
                    )?,
                ));

            let payment_method_custom_data =
                get_payment_method_custom_data(pmd.clone(), vault_token_selector)?;

            Box::pin(vault_payment_method_external(
                state,
                &payment_method_custom_data,
                platform.get_provider().get_account(),
                merchant_connector_account,
            ))
            .await
            .map(|value| (value, Some(external_vault_source)))
        }
        false => vault_payment_method_internal(
            state,
            pmd,
            platform,
            existing_vault_id,
            fingerprint_id_from_vault,
            customer_id,
            write_mode,
        )
        .await
        .map(|value| (value, None)),
    }
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn vault_payment_method_in_volatile_storage(
    state: &SessionState,
    pmd: &domain::PaymentMethodVaultingData,
    platform: &domain::Platform,
    _profile: &domain::Profile,
    _existing_vault_id: Option<domain::VaultId>,
    customer_id: &Option<id_type::GlobalCustomerId>,
) -> RouterResult<(
    pm_types::AddVaultResponse,
    Option<id_type::MerchantConnectorAccountId>,
)> {
    let vault_id = domain::VaultId::generate(generate_id(consts::ID_LENGTH, "vault"));
    let merchant_key_store = platform.get_provider().get_key_store();

    let encrypted_payload: Encryption = core_utils::create_encrypted_data(
        &(state).into(),
        merchant_key_store,
        pmd.clone(),
        common_utils::type_name!(diesel_models::payment_method::PaymentMethod),
    )
    .await
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to encrypt Payment method vaulting data")?
    .into();

    let redis_connection = state
        .store
        .get_redis_conn()
        .map_err(Into::<errors::StorageError>::into)
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to get redis connection")?;

    logger::info!("Storing payment method vaulting data in redis");
    redis_connection
        .serialize_and_set_key_with_expiry(
            &vault_id.get_string_repr().into(),
            encrypted_payload,
            consts::DEFAULT_PAYMENT_METHOD_STORE_TTL,
        )
        .await
        .map_err(Into::<errors::StorageError>::into)
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to insert payment method vaulting data to redis")?;

    Ok((
        pm_types::AddVaultResponse {
            entity_id: None,
            vault_id,
            fingerprint_id: None,
        },
        None,
    ))
}

#[cfg(feature = "v2")]
fn get_pm_list_context(
    payment_method_type: enums::PaymentMethod,
    payment_method: &domain::PaymentMethod,
    is_payment_associated: bool,
    storage_type: enums::StorageType,
) -> Result<Option<PaymentMethodListContext>, error_stack::Report<errors::ApiErrorResponse>> {
    let payment_method_data = payment_method
        .payment_method_data
        .clone()
        .map(|payment_method_data| payment_method_data.into_inner());

    let payment_method_retrieval_context = match payment_method_data {
        Some(domain::payment_method_data::PaymentMethodsData::Card(card)) => {
            Some(PaymentMethodListContext::Card {
                card_details: card.to_card_details_from_locker(),
                token_data: is_payment_associated.then_some(
                    storage::PaymentTokenData::permanent_card(
                        payment_method.get_id().clone(),
                        payment_method
                            .locker_id
                            .as_ref()
                            .map(|id| id.get_string_repr().to_owned())
                            .or_else(|| Some(payment_method.get_id().get_string_repr().to_owned())),
                        payment_method
                            .locker_id
                            .as_ref()
                            .map(|id| id.get_string_repr().to_owned())
                            .unwrap_or_else(|| {
                                payment_method.get_id().get_string_repr().to_owned()
                            }),
                        storage_type,
                    ),
                ),
            })
        }
        Some(domain::payment_method_data::PaymentMethodsData::BankDetails(bank_details)) => {
            let get_bank_account_token_data =
                || -> CustomResult<payment_methods::BankAccountTokenData,errors::ApiErrorResponse> {
                    let connector_details = bank_details
                        .connector_details
                        .first()
                        .cloned()
                        .ok_or(errors::ApiErrorResponse::InternalServerError)
                        .attach_printable("Failed to obtain bank account connector details")?;

                    let payment_method_subtype = payment_method
                        .get_payment_method_subtype()
                        .get_required_value("payment_method_subtype")
                        .attach_printable("PaymentMethodType not found")?;

                    Ok(payment_methods::BankAccountTokenData {
                        payment_method_type: payment_method_subtype,
                        payment_method: payment_method_type,
                        connector_details,
                    })
                };

            // Retrieve the pm_auth connector details so that it can be tokenized
            let bank_account_token_data = get_bank_account_token_data()
                .inspect_err(|error| logger::error!(?error))
                .ok();
            bank_account_token_data.map(|data| {
                let token_data = storage::PaymentTokenData::AuthBankDebit(data);

                PaymentMethodListContext::Bank {
                    token_data: is_payment_associated.then_some(token_data),
                }
            })
        }
        Some(domain::payment_method_data::PaymentMethodsData::BankDebit(bank_debit)) => {
            Some(PaymentMethodListContext::BankDebit {
                bank_debit_details: bank_debit.into(),
                token_data: is_payment_associated.then_some(storage::PaymentTokenData::bank_debit(
                    payment_method.get_id().clone(),
                    payment_method
                        .locker_id
                        .as_ref()
                        .map(|id| id.get_string_repr().to_owned()),
                    storage_type,
                )),
            })
        }
        Some(domain::payment_method_data::PaymentMethodsData::WalletDetails(_))
        | Some(domain::payment_method_data::PaymentMethodsData::NetworkToken(_))
        | None => Some(PaymentMethodListContext::TemporaryToken {
            token_data: is_payment_associated.then_some(
                storage::PaymentTokenData::temporary_generic(generate_id(
                    consts::ID_LENGTH,
                    "token",
                )),
            ),
        }),
    };

    Ok(payment_method_retrieval_context)
}

#[cfg(feature = "v2")]
fn get_pm_list_token_data(
    payment_method_type: enums::PaymentMethod,
    payment_method: &domain::PaymentMethod,
    storage_type: enums::StorageType,
) -> Result<Option<storage::PaymentTokenData>, error_stack::Report<errors::ApiErrorResponse>> {
    let pm_list_context =
        get_pm_list_context(payment_method_type, payment_method, true, storage_type)?
            .get_required_value("PaymentMethodListContext")?;

    match pm_list_context {
        PaymentMethodListContext::Card {
            card_details: _,
            token_data,
        } => Ok(token_data),
        PaymentMethodListContext::Bank { token_data } => Ok(token_data),
        PaymentMethodListContext::BankTransfer {
            bank_transfer_details: _,
            token_data,
        } => Ok(token_data),
        PaymentMethodListContext::TemporaryToken { token_data } => Ok(token_data),
        PaymentMethodListContext::BankDebit {
            bank_debit_details: _,
            token_data,
        } => Ok(token_data),
    }
}

#[cfg(all(feature = "v2", feature = "olap"))]
pub async fn list_payment_methods_core(
    state: &SessionState,
    provider: &domain::Provider,
    customer_id: &id_type::GlobalCustomerId,
    include_new: bool,
) -> RouterResult<payment_methods::CustomerPaymentMethodsListResponse> {
    let db = &*state.store;

    let statuses = if include_new {
        vec![
            common_enums::PaymentMethodStatus::Active,
            common_enums::PaymentMethodStatus::New,
        ]
    } else {
        vec![common_enums::PaymentMethodStatus::Active]
    };
    let saved_payment_methods = db
        .find_payment_method_by_global_customer_id_merchant_id_statuses(
            provider.get_key_store(),
            customer_id,
            provider.get_account().get_id(),
            statuses,
            None,
            provider.get_account().storage_scheme,
        )
        .await
        .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;

    let customer = db
        .find_customer_by_global_id_merchant_id_without_encrypted(
            customer_id,
            provider.get_account().get_id(),
            provider.get_account().storage_scheme,
        )
        .await
        .to_not_found_response(errors::ApiErrorResponse::CustomerNotFound)?;

    let default_payment_method_id = customer.default_payment_method_id;

    let customer_payment_methods = saved_payment_methods
        .into_iter()
        .map(|pm| {
            payment_methods::PaymentMethodResponseItem::foreign_try_from((
                pm,
                default_payment_method_id.clone(),
            ))
        })
        .collect::<Result<Vec<payment_methods::PaymentMethodResponseItem>, _>>()
        .change_context(errors::ApiErrorResponse::InternalServerError)?;

    let response = payment_methods::CustomerPaymentMethodsListResponse {
        customer_payment_methods,
    };

    Ok(response)
}

#[cfg(all(feature = "v2", feature = "oltp"))]
pub async fn list_customer_payment_methods_core(
    state: &SessionState,
    provider: &domain::Provider,
    customer_id: &id_type::GlobalCustomerId,
    include_new: bool,
) -> RouterResult<Vec<payment_methods::CustomerPaymentMethodResponseItem>> {
    let db = &*state.store;
    let statuses = if include_new {
        vec![
            common_enums::PaymentMethodStatus::Active,
            common_enums::PaymentMethodStatus::New,
        ]
    } else {
        vec![common_enums::PaymentMethodStatus::Active]
    };
    let saved_payment_methods = db
        .find_payment_method_by_global_customer_id_merchant_id_statuses(
            provider.get_key_store(),
            customer_id,
            provider.get_account().get_id(),
            statuses,
            None,
            provider.get_account().storage_scheme,
        )
        .await
        .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;

    let customer = db
        .find_customer_by_global_id_merchant_id_without_encrypted(
            customer_id,
            provider.get_account().get_id(),
            provider.get_account().storage_scheme,
        )
        .await
        .to_not_found_response(errors::ApiErrorResponse::CustomerNotFound)?;

    let default_payment_method_id = customer.default_payment_method_id;

    let mut customer_payment_methods = Vec::new();

    let payment_method_results: Result<Vec<_>, error_stack::Report<errors::ApiErrorResponse>> =
        saved_payment_methods
            .into_iter()
            .map(|pm| {
                let default_payment_method_id = default_payment_method_id.clone();
                async move {
                    let parent_payment_method_token = generate_id(consts::ID_LENGTH, "token");

                    // For payment methods that are active we should always have the payment method type
                    let payment_method_type = pm
                        .payment_method_type
                        .get_required_value("payment_method_type")?;

                    let intent_fulfillment_time =
                        common_utils::consts::DEFAULT_INTENT_FULFILLMENT_TIME;

                    let token_data = get_pm_list_token_data(
                        payment_method_type,
                        &pm,
                        // storage type is hardcoded to Persistent as we are fetching saved payment methods from the database
                        enums::StorageType::Persistent,
                    )?;

                    if let Some(token_data) = token_data {
                        pm_routes::ParentPaymentMethodToken::create_key_for_token(
                            &parent_payment_method_token,
                        )
                        .insert(intent_fulfillment_time, token_data, state)
                        .await?;

                        let final_pm = api::CustomerPaymentMethodResponseItem::foreign_try_from((
                            pm,
                            parent_payment_method_token,
                            default_payment_method_id,
                        ))
                        .change_context(errors::ApiErrorResponse::InternalServerError)
                        .attach_printable("Failed to convert payment method to response format")?;

                        Ok(Some(final_pm))
                    } else {
                        Ok(None)
                    }
                }
            })
            .collect::<futures::stream::FuturesUnordered<_>>()
            .try_collect::<Vec<_>>()
            .await;

    customer_payment_methods.extend(payment_method_results?.into_iter().flatten());

    Ok(customer_payment_methods)
}

#[cfg(all(feature = "v2", feature = "olap"))]
pub async fn get_total_payment_method_count_core(
    state: &SessionState,
    provider: &domain::Provider,
) -> RouterResult<api::TotalPaymentMethodCountResponse> {
    let db = &*state.store;

    let total_count = db
        .get_payment_method_count_by_merchant_id_status(
            provider.get_account().get_id(),
            common_enums::PaymentMethodStatus::Active,
        )
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to get total payment method count")?;

    let response = api::TotalPaymentMethodCountResponse { total_count };

    Ok(response)
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn retrieve_payment_method(
    state: SessionState,
    pm: api::PaymentMethodId,
    profile: domain::Profile,
    platform: domain::Platform,
    api_key_type: enums::ApiKeyType,
    fetch_raw_detail_query_param: bool,
) -> RouterResponse<api::PaymentMethodResponse> {
    let db = state.store.as_ref();

    let dimensions = dimension_state::Dimensions::new()
        .with_provider_merchant_id(platform.get_provider().get_provider_merchant_id());

    // 1. Resolve parent token (if any) -> storage type & optional token data
    let (storage_type, pm_token_data_opt) =
        resolve_storage_type_from_token(&state, &pm.payment_method_id).await?;

    // 2. Fetch payment method record based on resolved storage type
    let (storage_type, payment_method) = fetch_payment_method_by_storage(
        &state,
        platform.get_provider(),
        &pm,
        storage_type,
        pm_token_data_opt,
    )
    .await
    .change_context(errors::ApiErrorResponse::PaymentMethodNotFound)
    .attach_printable("Failed to fetch payment method by storage")?;

    when(
        matches!(
            payment_method.status,
            common_enums::PaymentMethodStatus::Inactive
                | common_enums::PaymentMethodStatus::Redacted
        ),
        || {
            Err(report!(errors::ApiErrorResponse::PaymentMethodNotFound)
                .attach_printable("Payment method is inactive or redacted"))
        },
    )?;

    let raw_payment_method_fetch_access = get_raw_payment_method_data_fetch_access(
        &state,
        &dimensions,
        api_key_type,
        fetch_raw_detail_query_param,
        payment_method.customer_id.as_ref(),
    )
    .await
    .attach_printable("Failed to get raw payment method fetch access")?;

    let single_use_token_in_cache = get_single_use_token_from_store(
        &state.clone(),
        domain::SingleUseTokenKey::store_key(&payment_method.id.clone()),
    )
    .await
    .unwrap_or_default();

    let card_cvc_expiry = vault::retrieve_key_and_ttl_for_cvc_from_payment_method_id(
        &state,
        payment_method.id.to_owned(),
    )
    .await
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to retrieve cvc from redis")
    .ok();

    let raw_payment_method_data =
        Box::pin(raw_payment_method_fetch_access.get_raw_payment_method_data(
            &state,
            &platform,
            &profile,
            &payment_method,
            storage_type,
        ))
        .await
        .attach_printable("Failed to get raw payment method data")?;

    let raw_network_token_details =
        Box::pin(raw_payment_method_fetch_access.get_raw_network_token_data(
            &state,
            &platform,
            &profile,
            &payment_method,
            storage_type,
        ))
        .await
        .inspect_err(|err| {
            logger::warn!(?err, "Failed to fetch raw network token details");
        })
        .ok();

    let raw_payment_method_data = match (raw_payment_method_data, raw_network_token_details) {
        (
            Some(payment_methods::RawPaymentMethodData::Card(card_details)),
            Some(network_token_details),
        ) => Some(payment_methods::RawPaymentMethodData::CardWithNT(
            payment_methods::RawCardWithNTDetails {
                card_details,
                network_token_details,
            },
        )),
        (raw_payment_method_data, _) => raw_payment_method_data,
    };

    let billing = payment_method
        .payment_method_billing_address
        .clone()
        .map(|billing| billing.into_inner())
        .map(From::from);

    transformers::generate_payment_method_response(
        &payment_method,
        &single_use_token_in_cache,
        storage_type,
        card_cvc_expiry.map(|time| {
            payment_methods::CardCVCTokenStorageDetails::generate_expiry_timestamp(time)
        }),
        payment_method.customer_id.clone(),
        raw_payment_method_data,
        billing,
        None,
    )
    .map(services::ApplicationResponse::Json)
}

#[cfg(all(feature = "v2", feature = "olap"))]
#[instrument(skip_all)]
pub async fn retrieve_payment_method_olap(
    state: SessionState,
    pm: api::PaymentMethodId,
    profile: domain::Profile,
    platform: domain::Platform,
) -> RouterResponse<api::PaymentMethodDetailsResponse> {
    let response = Box::pin(retrieve_payment_method(
        state,
        pm,
        profile,
        platform,
        enums::ApiKeyType::External,
        false,
    ))
    .await?
    .get_json_body()
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to get payment method response body")?;

    Ok(services::ApplicationResponse::Json(
        api::PaymentMethodDetailsResponse::from(response),
    ))
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn fetch_payment_method_by_storage(
    state: &SessionState,
    provider: &domain::Provider,
    pm_incoming: &api::PaymentMethodId,
    storage_type: common_enums::StorageType,
    pm_token_data_opt: Option<storage::PaymentTokenData>,
) -> RouterResult<(common_enums::StorageType, domain::PaymentMethod)> {
    let db = state.store.as_ref();
    match storage_type {
        common_enums::StorageType::Volatile => {
            logger::debug!("Fetching volatile payment method");
            // When it is Volatile storage payment token data will be always present in redis
            let pm_id = pm_token_data_opt
                .and_then(|token| match token {
                    storage::PaymentTokenData::PermanentCard(card) => Some(card.payment_method_id),
                    storage::PaymentTokenData::BankDebit(bank_debit) => {
                        Some(bank_debit.payment_method_id)
                    }
                    _ => None,
                })
                .ok_or_else(|| {
                    report!(errors::ApiErrorResponse::PaymentMethodNotFound)
                        .attach_printable("Failed to get token data for volatile storage")
                })?;

            let volatile_payment_method = fetch_volatile_payment_method_record(
                state,
                provider.get_key_store(),
                pm_id.get_string_repr(),
            )
            .await
            .attach_printable("Failed to get volatile payment method record")?;
            Ok((storage_type, volatile_payment_method))
        }
        common_enums::StorageType::Persistent => {
            logger::debug!("Fetching persistent payment method with fallback");
            // In S2S calls, id passed in the request could be payment method id as well
            // If a temporary token is passed after the redis expiration it will also be treated as
            // a persistent GlobalPaymentMethodId, but for temp tokens GlobalPaymentMethodId will fail
            let pm_id = match pm_token_data_opt {
                Some(storage::PaymentTokenData::PermanentCard(card)) => Ok(card.payment_method_id),
                Some(storage::PaymentTokenData::BankDebit(bank_debit)) => {
                    Ok(bank_debit.payment_method_id)
                }
                Some(_) => Err(report!(errors::ApiErrorResponse::PaymentMethodNotFound)
                    .attach_printable("Unexpected token data variant for payment method fetch")),
                None => Ok(id_type::GlobalPaymentMethodId::new_unchecked(
                    pm_incoming.payment_method_id.clone(),
                )),
            }?;

            fetch_payment_method_with_fallback(state, provider, &pm_id, storage_type)
                .await
                .attach_printable("Failed to get payment method with fallback")
        }
    }
}

#[cfg(feature = "v2")]
pub async fn fetch_payment_method_with_fallback(
    state: &SessionState,
    provider: &domain::Provider,
    pm_id: &id_type::GlobalPaymentMethodId,
    storage_type: common_enums::StorageType,
) -> RouterResult<(common_enums::StorageType, domain::PaymentMethod)> {
    let volatile_payment_method = fetch_volatile_payment_method_record(
        state,
        provider.get_key_store(),
        pm_id.get_string_repr(),
    )
    .await
    .attach_printable("Failed to get volatile payment method record");

    match volatile_payment_method {
        Ok(payment_method) => Ok((common_enums::StorageType::Volatile, payment_method)),
        Err(err) => {
            logger::warn!("Volatile payment method not found, falling back to persistent storage",);

            when(
                !matches!(
                    err.current_context(),
                    errors::ApiErrorResponse::GenericNotFoundError { .. }
                ),
                || Err(err),
            )?;

            logger::debug!("Redis lookup failed, falling back to DB");

            let persistent_payment_method = fetch_payment_method(state, provider, pm_id)
                .await
                .attach_printable("Failed to get payment method record from DB")?;

            Ok((storage_type, persistent_payment_method))
        }
    }
}

#[cfg(feature = "v2")]
async fn fetch_volatile_payment_method_record(
    state: &SessionState,
    key_store: &domain::MerchantKeyStore,
    pm_id: &str,
) -> RouterResult<domain::PaymentMethod> {
    let redis_conn = state
        .store
        .get_redis_conn()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to get redis connection")?;

    let payment_method = redis_conn
        .get_and_deserialize_key::<diesel_models::PaymentMethod>(&pm_id.into(), "PaymentMethod")
        .await
        .map_err(|e| error_stack::report!(storage_impl::StorageError::from(e)))
        .to_not_found_response(errors::ApiErrorResponse::GenericNotFoundError {
            message: "Payment method token either expired or does not exist".to_string(),
        })?;

    let keymanager_state = &state.into();

    let domain_payment_method = domain::PaymentMethod::convert_back(
        keymanager_state,
        payment_method,
        key_store.key.get_inner(),
        key_store.merchant_id.clone().into(),
    )
    .await
    .change_context(errors::StorageError::EncryptionError)
    .change_context(errors::ApiErrorResponse::InternalServerError)?;

    Ok(domain_payment_method)
}

#[cfg(feature = "v2")]
pub async fn resolve_storage_type_from_token(
    state: &SessionState,
    token: &String,
) -> RouterResult<(common_enums::StorageType, Option<storage::PaymentTokenData>)> {
    // Try redis lookup for parent token
    let token_data_res = pm_routes::ParentPaymentMethodToken::create_key_for_token(token)
        .get_data_for_token(state)
        .await;

    match token_data_res {
        Ok(token_data) => {
            // only Permanent/PermanentCard token variants carry storage type info relevant for saved PMs
            match token_data {
                storage::PaymentTokenData::PermanentCard(card) => {
                    let storage_type = card.storage_type;
                    Ok((
                        storage_type,
                        Some(storage::PaymentTokenData::PermanentCard(card)),
                    ))
                }
                storage::PaymentTokenData::BankDebit(bank_debit) => {
                    let storage_type = bank_debit.storage_type;
                    Ok((
                        storage_type,
                        Some(storage::PaymentTokenData::BankDebit(bank_debit)),
                    ))
                }
                // Any token kind other than PermanentCard is invalid for retrieving a saved payment method.
                // Only PermanentCard is currently supported; all other enum variants should be removed.
                storage::PaymentTokenData::TemporaryGeneric(_)
                | storage::PaymentTokenData::AuthBankDebit(_)
                | storage::PaymentTokenData::TemporaryCardToken(_) => {
                    Err(report!(errors::ApiErrorResponse::PaymentMethodNotFound))
                }
            }
        }
        Err(err) => {
            // If token not found in redis treat incoming id as a persistent PM id as it could have been obtained from a S2S payment method retrieve call
            // Or it would be an MIT case
            if matches!(
                err.current_context(),
                errors::ApiErrorResponse::GenericNotFoundError { .. }
            ) {
                Ok((common_enums::StorageType::Persistent, None))
            } else {
                Err(err)
            }
        }
    }
}

#[derive(
    Clone, Copy, Debug, strum::Display, PartialEq, Eq, serde::Serialize, serde::Deserialize,
)]
pub enum RawPaymentMethodFetchAccess {
    Allowed,
    Denied,
}

#[cfg(feature = "v2")]
impl RawPaymentMethodFetchAccess {
    pub async fn get_raw_payment_method_data(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        profile: &domain::Profile,
        payment_method: &domain::PaymentMethod,
        storage_type: common_enums::StorageType,
    ) -> RouterResult<Option<payment_methods::RawPaymentMethodData>> {
        match self {
            Self::Denied => {
                logger::debug!("Raw payment method fetch access denied");
                // When access is denied, check if the payment method has external vault token data.
                // If present, return it as a ProxyCard so non-PCI-compliant merchants can still
                // receive a tokenized card reference in the retrieve response.
                let proxy_card_data = payment_method
                    .external_vault_token_data
                    .clone()
                    .map(|enc| enc.into_inner());
                match proxy_card_data {
                    Some(external_vault_token_data) => {
                        Ok(Some(payment_methods::RawPaymentMethodData::ProxyCard(
                            payment_methods::RawProxyCardDataResponse {
                                card_number: external_vault_token_data.tokenized_card_number,
                                card_exp_year: None,
                                card_exp_month: None,
                            },
                        )))
                    }
                    None => Ok(None),
                }
            }

            Self::Allowed => {
                let should_skip_vault_fetch = matches!(
                    payment_method.payment_method_type,
                    Some(enums::PaymentMethod::Wallet) | Some(enums::PaymentMethod::BankRedirect)
                );

                if should_skip_vault_fetch {
                    logger::debug!("Skipping raw payment method fetch for wallet or bank redirect payment method");
                    Ok(None)
                } else {
                    let vault_data = Box::pin(vault::retrieve_payment_method_data_from_storage(
                        state,
                        platform,
                        profile,
                        payment_method,
                        storage_type,
                    ))
                    .await
                    .change_context(errors::ApiErrorResponse::InternalServerError)
                    .attach_printable("Failed to retrieve payment method from vault")?
                    .data;

                    let payment_method_vault_data = vault_data
                        .populated_payment_methods_data_and_get_payment_method_vaulting_data(
                            payment_method.payment_method_data.as_ref(),
                        )
                        .attach_printable(
                            "Failed to get card details for payment method vaulting data",
                        )?
                        .convert_to_raw_payment_method_data();
                    Ok(payment_method_vault_data)
                }
            }
        }
    }

    pub async fn get_raw_network_token_data(
        &self,
        state: &SessionState,
        platform: &domain::Platform,
        profile: &domain::Profile,
        payment_method: &domain::PaymentMethod,
        storage_type: common_enums::StorageType,
    ) -> RouterResult<payment_methods::CardDetail> {
        match self {
            Self::Denied => Err(report!(errors::ApiErrorResponse::AccessForbidden {
                resource: "raw network token data".to_string()
            }))
            .attach_printable("Raw network token fetch access denied"),
            Self::Allowed => {
                let network_token_locker_id = payment_method
                    .network_token_locker_id
                    .clone()
                    .ok_or(report!(errors::ApiErrorResponse::MissingRequiredField {
                        field_name: "network_token_locker_id"
                    }))?;

                let customer_id = payment_method.customer_id.clone().ok_or(report!(
                    errors::ApiErrorResponse::MissingRequiredField {
                        field_name: "customer_id"
                    }
                ))?;

                // Wrap network_token_locker_id with VaultId for type adaptation.
                // The vault retrieval helper consumes domain::PaymentMethod and reads pm.locker_id: Option<VaultId>,
                // while network token id is stored separately as network_token_locker_id: Option<String>.
                let mut network_token_payment_method = payment_method.clone();
                network_token_payment_method.locker_id =
                    Some(domain::VaultId::generate(network_token_locker_id));
                network_token_payment_method.customer_id = Some(customer_id);

                let network_token_vault_data =
                    Box::pin(vault::retrieve_payment_method_data_from_storage(
                        state,
                        platform,
                        profile,
                        &network_token_payment_method,
                        storage_type,
                    ))
                    .await
                    .change_context(errors::ApiErrorResponse::InternalServerError)
                    .attach_printable("Failed to retrieve network token from vault")?
                    .data;

                let check_token_status_response =
                    network_tokenization::do_status_check_for_network_token(state, payment_method)
                        .await
                        .change_context(errors::ApiErrorResponse::InternalServerError)
                        .attach_printable("Failed to check network token status")?;

                match (
                    check_token_status_response.payload.token_status,
                    network_token_vault_data,
                ) {
                    (
                        pm_types::TokenStatus::Active,
                        hyperswitch_domain_models::vault::PaymentMethodVaultingData::NetworkToken(
                            network_token_details,
                        ),
                    ) => Ok(payment_methods::CardDetail {
                        card_number: network_token_details.network_token.into(),
                        card_exp_month: network_token_details.network_token_exp_month,
                        card_exp_year: network_token_details.network_token_exp_year,
                        card_holder_name: network_token_details.card_holder_name,
                        nick_name: network_token_details.nick_name,
                        card_issuing_country: network_token_details.card_issuing_country,
                        card_network: network_token_details.card_network,
                        card_issuer: network_token_details.card_issuer,
                        card_type: network_token_details.card_type,
                        card_cvc: None,
                    }),
                    _ => Err(report!(errors::ApiErrorResponse::GenericNotFoundError {
                        message: "Network token not found or not active".to_string()
                    })),
                }
            }
        }
    }
}

#[cfg(feature = "v2")]
pub async fn get_raw_payment_method_data_fetch_access(
    state: &SessionState,
    dimensions: &dimension_state::DimensionsWithProviderMerchantId,
    api_key_type: enums::ApiKeyType,
    fetch_raw_detail_query_param: bool,
    customer_id: Option<&id_type::GlobalCustomerId>,
) -> RouterResult<RawPaymentMethodFetchAccess> {
    // If query param not set, never allowed to fetch raw payment method details
    let fetch_access = match fetch_raw_detail_query_param {
        true => RawPaymentMethodFetchAccess::Allowed,
        false => RawPaymentMethodFetchAccess::Denied,
    };

    match api_key_type {
        // Internal API keys always allowed
        enums::ApiKeyType::Internal => Ok(fetch_access),

        // External API keys allowed only via org-level config
        // This supports cases where a PCI-compliant entity needs to retrieve raw payment method details.
        enums::ApiKeyType::External => {
            let allowed = dimensions
                .get_should_return_raw_payment_method_details(
                    state.store.as_ref(),
                    state.superposition_service.as_ref(),
                    customer_id,
                )
                .await;

            match allowed {
                true => Ok(fetch_access),
                false => Ok(RawPaymentMethodFetchAccess::Denied),
            }
        }
    }
}

// TODO: When we separate out microservices, this function will be an endpoint in payment_methods
#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn update_payment_method_status_internal(
    state: &SessionState,
    key_store: &domain::MerchantKeyStore,
    storage_scheme: enums::MerchantStorageScheme,
    status: enums::PaymentMethodStatus,
    payment_method_id: &id_type::GlobalPaymentMethodId,
    initiator: Option<&domain::Initiator>,
) -> RouterResult<domain::PaymentMethod> {
    let db = &*state.store;

    let payment_method = db
        .find_payment_method(key_store, payment_method_id, storage_scheme)
        .await
        .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;

    let pm_update = storage::PaymentMethodUpdate::StatusUpdate {
        status: Some(status),
        last_modified_by: initiator
            .and_then(|initiator| initiator.to_created_by())
            .map(|last_modified_by| last_modified_by.to_string()),
    };

    let updated_pm = db
        .update_payment_method(
            key_store,
            payment_method.clone(),
            pm_update,
            storage_scheme,
            // This internal PM update path is not part of modular compatibility scheduling.
            None,
        )
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to update payment method in db")?;

    Ok(updated_pm)
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn update_payment_method(
    state: SessionState,
    platform: domain::Platform,
    profile: domain::Profile,
    req: api::PaymentMethodUpdate,
    payment_method_id: &id_type::GlobalPaymentMethodId,
) -> RouterResponse<api::PaymentMethodResponse> {
    let update_request = DomainPaymentMethodUpdate::from(req);

    let (response, _updated_payment_method) = Box::pin(update_payment_method_core(
        &state,
        &platform,
        &profile,
        update_request,
        payment_method_id,
        None,
        None,
    ))
    .await?;

    // Network tokenization is intentionally not scheduled here: the update request carries no
    // network tokenization opt-in field, so there is no per-payment-method consent to act on.
    // Scheduling happens only in the create flow, where the request opts in explicitly.

    Ok(services::ApplicationResponse::Json(response))
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn update_payment_method_core(
    state: &SessionState,
    platform: &domain::Platform,
    profile: &domain::Profile,
    request: DomainPaymentMethodUpdate,
    payment_method_id: &id_type::GlobalPaymentMethodId,
    existing_payment_method: Option<domain::PaymentMethod>,
    network_tokenization_resp: Option<NetworkTokenPaymentMethodDetails>,
) -> RouterResult<(api::PaymentMethodResponse, domain::PaymentMethod)> {
    let mut handler = match existing_payment_method {
        Some(payment_method) => {
            let handler = pm_types::PaymentMethodUpdateHandler {
                platform,
                profile,
                request,
                payment_method,
                state,
            };
            handler.validate()?;
            handler
        }
        None => pm_types::PaymentMethodUpdateHandler::generate(
            state,
            platform,
            profile,
            request,
            payment_method_id,
        )
        .await
        .attach_printable("Failed to generate PaymentMethodUpdateHandler")?,
    };

    Box::pin(execute_payment_method_update_handler(
        &mut handler,
        network_tokenization_resp,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
async fn execute_payment_method_update_handler(
    handler: &mut pm_types::PaymentMethodUpdateHandler<'_>,
    network_tokenization_resp: Option<NetworkTokenPaymentMethodDetails>,
) -> RouterResult<(api::PaymentMethodResponse, domain::PaymentMethod)> {
    let card_cvc_details = handler
        .update_cvc_if_required()
        .await
        .attach_printable("Failed to update card cvc")?;

    let (vaulting_data, vaulting_resp) =
        Box::pin(handler.perform_vaulting_operations_if_required())
            .await
            .attach_printable("Failed to perform vaulting operations for payment method update")?;

    handler
        .update_payment_method_if_required(vaulting_data, vaulting_resp, network_tokenization_resp)
        .await
        .attach_printable("Failed to update payment method in db")?;

    let response = handler
        .generate_response(card_cvc_details)
        .attach_printable("Failed to generate response for payment method update")?;

    Ok((response, handler.payment_method.clone()))
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn delete_payment_method(
    state: SessionState,
    pm_id: api::PaymentMethodId,
    platform: domain::Platform,
    profile: domain::Profile,
) -> RouterResponse<api::PaymentMethodDeleteResponse> {
    let pm_id = id_type::GlobalPaymentMethodId::new_unchecked(pm_id.payment_method_id);
    let response = Box::pin(delete_payment_method_core(
        &state, pm_id, &platform, &profile,
    ))
    .await?;

    Ok(services::ApplicationResponse::Json(response))
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn delete_payment_method_core(
    state: &SessionState,
    pm_id: id_type::GlobalPaymentMethodId,
    platform: &domain::Platform,
    profile: &domain::Profile,
) -> RouterResult<api::PaymentMethodDeleteResponse> {
    let db = state.store.as_ref();

    let payment_method = db
        .find_payment_method(
            platform.get_provider().get_key_store(),
            &pm_id,
            platform.get_provider().get_account().storage_scheme,
        )
        .await
        .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;
    let customer_id = &payment_method
        .customer_id
        .clone()
        .get_required_value("GlobalCustomerId")?;

    when(
        payment_method.status == enums::PaymentMethodStatus::Inactive,
        || Err(errors::ApiErrorResponse::PaymentMethodNotFound),
    )?;

    when(
        payment_method.status == enums::PaymentMethodStatus::Redacted,
        || Err(errors::ApiErrorResponse::PaymentMethodRedacted),
    )?;

    let _customer = db
        .find_customer_by_global_id_merchant_id_without_encrypted(
            customer_id,
            platform.get_provider().get_account().get_id(),
            platform.get_provider().get_account().storage_scheme,
        )
        .await
        .to_not_found_response(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Customer not found for the payment method")?;

    Box::pin(delete_payment_method_by_record(
        db,
        state,
        platform,
        profile,
        payment_method,
    ))
    .await?;

    let response = api::PaymentMethodDeleteResponse { id: pm_id };

    Ok(response)
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn delete_payment_method_by_record(
    db: &dyn StorageInterface,
    state: &SessionState,
    platform: &domain::Platform,
    profile: &domain::Profile,
    payment_method: domain::PaymentMethod,
) -> RouterResult<()> {
    // Soft delete - mark as Redacted (terminal state, no transitions allowed)
    let pm_update = storage::PaymentMethodUpdate::StatusAndFingerprintUpdate {
        status: Some(enums::PaymentMethodStatus::Redacted),
        last_modified_by: platform
            .get_initiator()
            .and_then(|initiator| initiator.to_created_by())
            .map(|last_modified_by| last_modified_by.to_string()),
        locker_fingerprint_id: Some(PAYMENT_METHOD_REDACTED_FINGERPRINT_ID.to_string()),
    };

    db.update_payment_method(
        platform.get_provider().get_key_store(),
        payment_method.clone(),
        pm_update,
        platform.get_provider().get_account().storage_scheme,
        // Redaction updates should not trigger PM modular compat scheduling.
        None,
    )
    .await
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to update payment method in db")?;

    vault::delete_payment_method_data_from_vault(state, platform, profile, &payment_method)
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to delete payment method from vault")?;

    Ok(())
}

#[cfg(feature = "v2")]
#[async_trait::async_trait]
trait EncryptableData {
    type Output;
    async fn encrypt_data(
        &self,
        key_manager_state: &common_utils::types::keymanager::KeyManagerState,
        key_store: &domain::MerchantKeyStore,
    ) -> RouterResult<Self::Output>;
}

#[cfg(feature = "v2")]
#[async_trait::async_trait]
impl EncryptableData for payment_methods::PaymentMethodSessionRequest {
    type Output = hyperswitch_domain_models::payment_methods::DecryptedPaymentMethodSession;

    async fn encrypt_data(
        &self,
        key_manager_state: &common_utils::types::keymanager::KeyManagerState,
        key_store: &domain::MerchantKeyStore,
    ) -> RouterResult<Self::Output> {
        use common_utils::types::keymanager::ToEncryptable;

        let encrypted_billing_address = self
            .billing
            .clone()
            .map(|address| address.encode_to_value())
            .transpose()
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to encode billing address")?
            .map(Secret::new);

        let batch_encrypted_data = domain_types::crypto_operation(
            key_manager_state,
            common_utils::type_name!(hyperswitch_domain_models::payment_methods::PaymentMethodSession),
            domain_types::CryptoOperation::BatchEncrypt(
                hyperswitch_domain_models::payment_methods::FromRequestEncryptablePaymentMethodSession::to_encryptable(
                    hyperswitch_domain_models::payment_methods::FromRequestEncryptablePaymentMethodSession {
                       billing: encrypted_billing_address,
                    },
                ),
            ),
            common_utils::types::keymanager::Identifier::Merchant(key_store.merchant_id.clone()),
            key_store.key.peek(),
        )
        .await
        .and_then(|val| val.try_into_batchoperation())
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed while encrypting payment methods session details".to_string())?;

        let encrypted_data =
        hyperswitch_domain_models::payment_methods::FromRequestEncryptablePaymentMethodSession::from_encryptable(
            batch_encrypted_data,
        )
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed while encrypting payment methods session detailss")?;

        Ok(encrypted_data)
    }
}

#[cfg(feature = "v2")]
#[async_trait::async_trait]
impl EncryptableData for payment_methods::PaymentMethodsSessionUpdateRequest {
    type Output = hyperswitch_domain_models::payment_methods::DecryptedPaymentMethodSession;

    async fn encrypt_data(
        &self,
        key_manager_state: &common_utils::types::keymanager::KeyManagerState,
        key_store: &domain::MerchantKeyStore,
    ) -> RouterResult<Self::Output> {
        use common_utils::types::keymanager::ToEncryptable;

        let encrypted_billing_address = self
            .billing
            .clone()
            .map(|address| address.encode_to_value())
            .transpose()
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to encode billing address")?
            .map(Secret::new);

        let batch_encrypted_data = domain_types::crypto_operation(
            key_manager_state,
            common_utils::type_name!(hyperswitch_domain_models::payment_methods::PaymentMethodSession),
            domain_types::CryptoOperation::BatchEncrypt(
                hyperswitch_domain_models::payment_methods::FromRequestEncryptablePaymentMethodSession::to_encryptable(
                    hyperswitch_domain_models::payment_methods::FromRequestEncryptablePaymentMethodSession {
                       billing: encrypted_billing_address,
                    },
                ),
            ),
            common_utils::types::keymanager::Identifier::Merchant(key_store.merchant_id.clone()),
            key_store.key.peek(),
        )
        .await
        .and_then(|val| val.try_into_batchoperation())
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed while encrypting payment methods session details".to_string())?;

        let encrypted_data =
        hyperswitch_domain_models::payment_methods::FromRequestEncryptablePaymentMethodSession::from_encryptable(
            batch_encrypted_data,
        )
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed while encrypting payment methods session detailss")?;

        Ok(encrypted_data)
    }
}

#[cfg(feature = "v2")]
pub async fn payment_methods_session_create(
    state: SessionState,
    platform: domain::Platform,
    profile: domain::Profile,
    request: payment_methods::PaymentMethodSessionRequest,
) -> RouterResponse<payment_methods::PaymentMethodSessionResponse> {
    let db = state.store.as_ref();
    let key_manager_state = &(&state).into();
    let _provider = platform.get_provider();

    let customer = if let Some(customer_id) = &request.customer_id {
        let customer = db
            .find_customer_by_global_id_merchant_id(
                customer_id,
                platform.get_provider().get_account().get_id(),
                platform.get_provider().get_key_store(),
                platform.get_provider().get_account().storage_scheme,
            )
            .await
            .to_not_found_response(errors::ApiErrorResponse::CustomerNotFound)
            .attach_printable("Customer not found for the payment method")?;
        Some(customer)
    } else {
        None
    };

    let payment_methods_session_id =
        id_type::GlobalPaymentMethodSessionId::generate(&state.conf.cell_information.id)
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Unable to generate GlobalPaymentMethodSessionId")?;

    let encrypted_data = request
        .encrypt_data(key_manager_state, platform.get_provider().get_key_store())
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to encrypt payment methods session data")?;

    let billing = encrypted_data
        .billing
        .as_ref()
        .map(|data| {
            data.clone()
                .deserialize_inner_value(|value| value.parse_value("Address"))
        })
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to decode billing address")?;

    // If not passed in the request, use the default value from constants
    let expires_in = request
        .expires_in
        .unwrap_or(consts::DEFAULT_PAYMENT_METHOD_SESSION_EXPIRY)
        .into();

    let expires_at = common_utils::date_time::now().saturating_add(Duration::seconds(expires_in));

    let client_secret = payment_helpers::create_client_secret(
        &state,
        platform.get_provider().get_account().get_id(),
        util_types::authentication::ResourceId::PaymentMethodSession(
            payment_methods_session_id.clone(),
        ),
    )
    .await
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Unable to create client secret")?;

    let payment_method_session_domain_model =
        hyperswitch_domain_models::payment_methods::PaymentMethodSession {
            id: payment_methods_session_id,
            customer_id: request.customer_id.clone(),
            billing,
            psp_tokenization: None,
            network_tokenization: request.network_tokenization,
            tokenization_data: request.tokenization_data,
            expires_at,
            return_url: request.return_url,
            associated_payment_methods: None,
            associated_payment: None,
            associated_token_id: None,
            storage_type: request.storage_type,
            keep_alive: request.keep_alive.unwrap_or_default(),
        };

    db.insert_payment_methods_session(
        platform.get_provider().get_key_store(),
        payment_method_session_domain_model.clone(),
        expires_in,
    )
    .await
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to insert payment methods session in db")?;

    let external_vault_details =
        Box::pin(payments_core::vault_session::fetch_external_vault_details(
            &state,
            &platform,
            &profile,
            &customer,
            payment_method_session_domain_model.storage_type,
        ))
        .await
        .unwrap_or_else(|err| {
            router_env::logger::warn!(
                ?err,
                "Failed to fetch external vault details for payment method session"
            );
            None
        });

    let sdk_authorization = Option::<hyperswitch_domain_models::sdk_auth::SdkAuthorization>::from(
        hyperswitch_domain_models::sdk_auth::SdkAuthorizationContext {
            publishable_key: platform
                .get_processor()
                .get_account()
                .publishable_key
                .clone(),
            platform: platform.clone(),
            profile_id: profile.get_id().clone(),
            client_secret: client_secret.secret.clone().expose(),
            customer_id: payment_method_session_domain_model.customer_id.clone(),
            payment_method_session_id: Some(payment_method_session_domain_model.id.clone()),
        },
    );

    let response = transformers::generate_payment_method_session_response(
        payment_method_session_domain_model,
        client_secret.secret,
        sdk_authorization,
        None,
        None,
        request.storage_type,
        None,
        None,
        None,
        external_vault_details,
    );

    Ok(services::ApplicationResponse::Json(response))
}

#[cfg(feature = "v2")]
pub async fn payment_methods_session_update(
    state: SessionState,
    provider: domain::Provider,
    payment_method_session_id: id_type::GlobalPaymentMethodSessionId,
    request: payment_methods::PaymentMethodsSessionUpdateRequest,
) -> RouterResponse<payment_methods::PaymentMethodSessionResponse> {
    let db = state.store.as_ref();
    let key_manager_state = &(&state).into();

    let existing_payment_method_session_state = db
        .get_payment_methods_session(provider.get_key_store(), &payment_method_session_id)
        .await
        .to_not_found_response(errors::ApiErrorResponse::GenericNotFoundError {
            message: "payment methods session does not exist or has expired".to_string(),
        })
        .attach_printable("Failed to retrieve payment methods session from db")?;

    let encrypted_data = request
        .encrypt_data(key_manager_state, provider.get_key_store())
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to encrypt payment methods session data")?;

    let billing = encrypted_data
        .billing
        .as_ref()
        .map(|data| {
            data.clone()
                .deserialize_inner_value(|value| value.parse_value("Address"))
        })
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to decode billing address")?;

    let payment_method_session_domain_model =
        hyperswitch_domain_models::payment_methods::PaymentMethodsSessionUpdateEnum::GeneralUpdate{
            billing: Box::new(billing),
            psp_tokenization: None,
            network_tokenization: request.network_tokenization,
            tokenization_data: request.tokenization_data,
        };

    let update_state_change = db
        .update_payment_method_session(
            provider.get_key_store(),
            &payment_method_session_id,
            payment_method_session_domain_model,
            existing_payment_method_session_state.clone(),
        )
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to update payment methods session in db")?;

    let response = transformers::generate_payment_method_session_response(
        update_state_change.clone(),
        Secret::new("CLIENT_SECRET_REDACTED".to_string()),
        None, // sdk_authorization is not returned for non-create flows
        None, // TODO: send associated payments response based on the expandable param
        None,
        update_state_change.storage_type,
        None,
        None,
        None,
        None,
    );

    Ok(services::ApplicationResponse::Json(response))
}
#[cfg(feature = "v2")]
pub async fn payment_methods_session_retrieve(
    state: SessionState,
    provider: domain::Provider,
    payment_method_session_id: id_type::GlobalPaymentMethodSessionId,
) -> RouterResponse<payment_methods::PaymentMethodSessionResponse> {
    let db = state.store.as_ref();

    let payment_method_session_domain_model = db
        .get_payment_methods_session(provider.get_key_store(), &payment_method_session_id)
        .await
        .to_not_found_response(errors::ApiErrorResponse::GenericNotFoundError {
            message: "payment methods session does not exist or has expired".to_string(),
        })
        .attach_printable("Failed to retrieve payment methods session from db")?;

    // Only one associated payment method is supported for now
    let associated_pm_token = payment_method_session_domain_model
        .associated_payment_methods
        .as_ref()
        .and_then(|payment_methods| {
            payment_methods.iter().map(|pm| match &pm.payment_method_token {
                common_types::payment_methods::AssociatedPaymentMethodTokenType::PaymentMethodSessionToken(token) => token.clone(),
            }).next()
        });

    let expiry = associated_pm_token
        .async_map(|pm_token| {
            utils::retrieve_payment_method_id_from_payment_method_token_data(&state, pm_token)
        })
        .await
        .transpose()
        .ok()
        .flatten()
        .async_map(|pm_id| {
            vault::retrieve_key_and_ttl_for_cvc_from_payment_method_id(&state, pm_id.to_owned())
        })
        .await
        .transpose()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to retrieve cvc from redis")
        .ok()
        .flatten();

    let response = transformers::generate_payment_method_session_response(
        payment_method_session_domain_model.clone(),
        Secret::new("CLIENT_SECRET_REDACTED".to_string()),
        None, // sdk_authorization is not returned for non-create flows
        None, // TODO: send associated payments response based on the expandable param
        None,
        payment_method_session_domain_model.storage_type,
        expiry.map(|time| {
            payment_methods::CardCVCTokenStorageDetails::generate_expiry_timestamp(time)
        }),
        None,
        None,
        None,
    );

    Ok(services::ApplicationResponse::Json(response))
}

/// Stores `card_cvc` as an encrypted CVC under `pm_token_{token}_hyperswitch_cvc` — the same
/// `_hyperswitch_cvc` mechanism used by the first-time flow — so CVC retrieval (router-side
/// `retrieve_and_delete_cvc_from_payment_token`) resolves it consistently. The card holder name is
/// not vaulted with the CVC token; it is carried by the saved payment method itself.
#[cfg(feature = "v2")]
async fn store_cvc_and_card_holder_name_as_payment_token_in_redis(
    state: &SessionState,
    token: &str,
    card_cvc: Option<Secret<String>>,
    _card_holder_name: Option<Secret<String>>,
    key_store: &domain::MerchantKeyStore,
) -> RouterResult<()> {
    if let Some(card_cvc) = card_cvc {
        vault::insert_cvc_using_payment_token(
            state,
            token,
            card_cvc,
            common_utils::consts::DEFAULT_INTENT_FULFILLMENT_TIME,
            key_store,
        )
        .await?;
    }
    Ok(())
}

/// Extracts `(card_cvc, card_holder_name)` from a `PaymentMethodUpdateData`, returning
/// `(None, None)` for non-card variants.
#[cfg(feature = "v2")]
fn extract_card_cvc_and_holder_name(
    payment_method_data: Option<&payment_methods::PaymentMethodUpdateData>,
) -> (Option<Secret<String>>, Option<Secret<String>>) {
    match payment_method_data {
        Some(payment_methods::PaymentMethodUpdateData::Card(card_update)) => (
            card_update.card_cvc.clone(),
            card_update.card_holder_name.clone(),
        ),
        _ => (None, None),
    }
}

/// Updates the associated payment methods list in the session DB record.
#[cfg(feature = "v2")]
async fn update_associated_payment_methods_in_session(
    db: &dyn StorageInterface,
    key_store: &domain::MerchantKeyStore,
    payment_method_session_id: &id_type::GlobalPaymentMethodSessionId,
    payment_method_session: hyperswitch_domain_models::payment_methods::PaymentMethodSession,
    associated_payment_methods: Option<
        Vec<common_types::payment_methods::AssociatedPaymentMethods>,
    >,
) -> RouterResult<hyperswitch_domain_models::payment_methods::PaymentMethodSession> {
    let update = hyperswitch_domain_models::payment_methods::PaymentMethodsSessionUpdateEnum::UpdateAssociatedPaymentMethods {
        associated_payment_methods,
    };
    db.update_payment_method_session(
        key_store,
        payment_method_session_id,
        update,
        payment_method_session,
    )
    .await
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to update payment method session with associated payment methods")
}

/// Moves `token` to the front of the existing associated payment methods list.
/// Entries with a different token are preserved in their original order after position 0.
#[cfg(feature = "v2")]
fn reorder_token_to_front(
    existing: Option<Vec<common_types::payment_methods::AssociatedPaymentMethods>>,
    token: &str,
) -> Option<Vec<common_types::payment_methods::AssociatedPaymentMethods>> {
    existing.map(|mut tokens| {
        tokens.retain(|pm| match &pm.payment_method_token {
            common_types::payment_methods::AssociatedPaymentMethodTokenType::PaymentMethodSessionToken(t) => t != token,
        });
        tokens.insert(
            0,
            common_types::payment_methods::AssociatedPaymentMethods {
                payment_method_token: common_types::payment_methods::AssociatedPaymentMethodTokenType::PaymentMethodSessionToken(token.to_owned()),
            },
        );
        tokens
    })
}

#[cfg(feature = "v2")]
pub async fn payment_methods_session_update_payment_method(
    state: SessionState,
    platform: domain::Platform,
    profile: domain::Profile,
    payment_method_session_id: id_type::GlobalPaymentMethodSessionId,
    request: payment_methods::PaymentMethodSessionUpdateSavedPaymentMethod,
) -> RouterResponse<payment_methods::PaymentMethodSessionResponse> {
    let db = state.store.as_ref();

    // Validate if the session still exists
    let payment_method_session = db
        .get_payment_methods_session(
            platform.get_provider().get_key_store(),
            &payment_method_session_id,
        )
        .await
        .to_not_found_response(errors::ApiErrorResponse::GenericNotFoundError {
            message: "payment methods session does not exist or has expired".to_string(),
        })
        .attach_printable("Failed to retrieve payment methods session from db")?;

    // Extract card CVC and holder name from the update request (if card update)
    let (card_cvc, card_holder_name) = extract_card_cvc_and_holder_name(
        request
            .payment_method_update_request
            .payment_method_data
            .as_ref(),
    );

    // If only CVC/holder name are provided (no payment_method_token), this is a
    // temporary card token flow: store CVC in Redis and return a new token.
    // If a token is provided, re-use it (update the CVC stored under that token).
    let has_cvc_and_card_holder_name_only_data = card_cvc.is_some() || card_holder_name.is_some();

    match request.payment_method_token.clone() {
        None if has_cvc_and_card_holder_name_only_data => {
            // Generate a new token, store CVC + card_holder_name in Redis
            let parent_payment_method_token = generate_id(consts::ID_LENGTH, "token");

            store_cvc_and_card_holder_name_as_payment_token_in_redis(
                &state,
                &parent_payment_method_token,
                card_cvc,
                card_holder_name,
                platform.get_provider().get_key_store(),
            )
            .await?;

            let associated_payment_methods = Some(vec![
                common_types::payment_methods::AssociatedPaymentMethods {
                    payment_method_token: common_types::payment_methods::AssociatedPaymentMethodTokenType::PaymentMethodSessionToken(
                        parent_payment_method_token,
                    ),
                },
            ]);

            let updated_payment_method_session = update_associated_payment_methods_in_session(
                db,
                platform.get_provider().get_key_store(),
                &payment_method_session_id,
                payment_method_session,
                associated_payment_methods,
            )
            .await?;

            let response = transformers::generate_payment_method_session_response(
                updated_payment_method_session.clone(),
                Secret::new("CLIENT_SECRET_REDACTED".to_string()),
                None,
                None,
                None,
                updated_payment_method_session.storage_type,
                None,
                None,
                None,
                None,
            );

            Ok(services::ApplicationResponse::Json(response))
        }
        token_opt => {
            // Existing saved PM flow: token must be present
            let pm_token = token_opt.ok_or(errors::ApiErrorResponse::MissingRequiredField {
                field_name: "payment_method_token",
            })?;

            // Validate token is associated with this session
            payment_method_session
                .associated_payment_methods
                .as_ref()
                .and_then(|payment_methods| {
                    payment_methods.iter().find(|pm| match &pm.payment_method_token {
                        common_types::payment_methods::AssociatedPaymentMethodTokenType::PaymentMethodSessionToken(token) => token == &pm_token
                    })
                })
                .ok_or(errors::ApiErrorResponse::GenericNotFoundError {
                    message: "No associated payment method found in the session".to_string(),
                })?;

            // If CVC is present, refresh the TemporaryCardToken in Redis under the same token
            if has_cvc_and_card_holder_name_only_data {
                store_cvc_and_card_holder_name_as_payment_token_in_redis(
                    &state,
                    &pm_token,
                    card_cvc,
                    card_holder_name,
                    platform.get_provider().get_key_store(),
                )
                .await?;
            }

            let payment_method_id =
                utils::retrieve_payment_method_id_from_payment_method_token_data(
                    &state,
                    pm_token.clone(),
                )
                .await
                .attach_printable(
                    "Failed to retrieve payment method id from payment method token data",
                )?;

            // Re-order the token to the front of the associated payment methods list
            let tokens = reorder_token_to_front(
                payment_method_session.associated_payment_methods.clone(),
                &pm_token,
            );

            let updated_payment_method_session = update_associated_payment_methods_in_session(
                db,
                platform.get_provider().get_key_store(),
                &payment_method_session_id,
                payment_method_session,
                tokens,
            )
            .await?;

            let update_request =
                DomainPaymentMethodUpdate::from(request.payment_method_update_request);

            let (update_response, _updated_payment_method) = Box::pin(update_payment_method_core(
                &state,
                &platform,
                &profile,
                update_request,
                &payment_method_id,
                None,
                None,
            ))
            .await
            .attach_printable("Failed to update saved payment method")?;

            let response = transformers::generate_payment_method_session_response(
                updated_payment_method_session.clone(),
                Secret::new("CLIENT_SECRET_REDACTED".to_string()),
                None, // sdk_authorization is not returned for non-create flows
                None, // TODO: send associated payments response based on the expandable param
                None,
                updated_payment_method_session.storage_type,
                update_response.card_cvc_token_storage,
                update_response.payment_method_data.clone(),
                None,
                None,
            );

            Ok(services::ApplicationResponse::Json(response))
        }
    }
}

#[cfg(feature = "v2")]
pub async fn payment_methods_session_delete_payment_method(
    state: SessionState,
    platform: domain::Platform,
    profile: domain::Profile,
    pm_token: String,
    payment_method_session_id: id_type::GlobalPaymentMethodSessionId,
) -> RouterResponse<payment_methods::PaymentMethodDeleteSessionResponse> {
    let db = state.store.as_ref();

    // Validate if the session still exists
    let payment_method_session = db
        .get_payment_methods_session(
            platform.get_provider().get_key_store(),
            &payment_method_session_id,
        )
        .await
        .to_not_found_response(errors::ApiErrorResponse::GenericNotFoundError {
            message: "payment methods session does not exist or has expired".to_string(),
        })
        .attach_printable("Failed to retrieve payment methods session from db")?;

    let payment_method_id =
        utils::retrieve_payment_method_id_from_payment_method_token_data(&state, pm_token.clone())
            .await
            .attach_printable(
                "Failed to retrieve payment method id from payment method token data",
            )?;

    Box::pin(delete_payment_method_core(
        &state,
        payment_method_id,
        &platform,
        &profile,
    ))
    .await
    .attach_printable("Failed to delete saved payment method")?;

    let tokens =  payment_method_session
        .associated_payment_methods
        .clone()
        .map(|tokens| tokens.into_iter().filter(|token| match &token.payment_method_token {
            common_types::payment_methods::AssociatedPaymentMethodTokenType::PaymentMethodSessionToken(token) => token != &pm_token
        }).collect::<Vec<_>>());

    // Update payment method session with associated payment methods
    let update_payment_method_session = hyperswitch_domain_models::payment_methods::PaymentMethodsSessionUpdateEnum::UpdateAssociatedPaymentMethods {
        associated_payment_methods: tokens
    };

    db.update_payment_method_session(
        platform.get_provider().get_key_store(),
        &payment_method_session_id,
        update_payment_method_session,
        payment_method_session,
    )
    .await
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to update payment method session with associated payment methods")?;

    Ok(services::ApplicationResponse::Json(
        payment_methods::PaymentMethodDeleteSessionResponse {
            payment_method_token: pm_token,
        },
    ))
}

#[cfg(feature = "v2")]
async fn create_zero_auth_payment(
    state: SessionState,
    req_state: routes::app::ReqState,
    platform: domain::Platform,
    profile: domain::Profile,
    request: api_models::payments::PaymentsRequest,
) -> RouterResult<api_models::payments::PaymentsResponse> {
    let response = Box::pin(payments_core::payments_create_and_confirm_intent(
        state,
        req_state,
        platform,
        profile,
        request,
        hyperswitch_domain_models::payments::HeaderPayload::default(),
    ))
    .await?;

    logger::info!(associated_payments_response=?response);

    response
        .get_json_body()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unexpected response from payments core")
}

#[cfg(feature = "v2")]
pub async fn payment_methods_session_confirm(
    state: SessionState,
    req_state: routes::app::ReqState,
    platform: domain::Platform,
    profile: domain::Profile,
    payment_method_session_id: id_type::GlobalPaymentMethodSessionId,
    request: payment_methods::PaymentMethodSessionConfirmRequest,
) -> RouterResponse<payment_methods::PaymentMethodSessionResponse> {
    let db: &dyn StorageInterface = state.store.as_ref();

    // Validate if the session still exists
    let payment_method_session = db
        .get_payment_methods_session(
            platform.get_provider().get_key_store(),
            &payment_method_session_id,
        )
        .await
        .to_not_found_response(errors::ApiErrorResponse::GenericNotFoundError {
            message: "payment methods session does not exist or has expired".to_string(),
        })
        .attach_printable("Failed to retrieve payment methods session from db")?;

    request.validate(&payment_method_session)?;

    let payment_method_session_billing = payment_method_session
        .billing
        .clone()
        .map(|billing| billing.into_inner())
        .map(From::from);

    // Unify the billing address that we receive from the session and from the confirm request
    let unified_billing_address = request
        .payment_method_data
        .billing
        .clone()
        .map(|payment_method_billing| {
            payment_method_billing.unify_address(payment_method_session_billing.as_ref())
        })
        .or_else(|| payment_method_session_billing.clone());

    let storage_type = payment_method_session.storage_type;

    let create_payment_method_request = get_payment_method_create_request(
        request
            .payment_method_data
            .payment_method_data
            .as_ref()
            .get_required_value("payment_method_data")?,
        request.payment_method_type,
        request.payment_method_subtype,
        payment_method_session.customer_id.clone(),
        unified_billing_address.as_ref(),
        Some(&payment_method_session),
        storage_type,
    )
    .attach_printable("Failed to create payment method request")?;

    let (payment_method_response, payment_method) = Box::pin(create_payment_method_core(
        &state,
        &req_state,
        create_payment_method_request.clone(),
        &platform,
        &profile,
    ))
    .await?;

    let parent_payment_method_token = generate_id(consts::ID_LENGTH, "token");

    let token_data = get_pm_list_token_data(
        request.payment_method_type,
        &payment_method,
        create_payment_method_request.storage_type,
    )?;

    // insert the token data into redis
    if let Some(token_data) = token_data {
        let intent_fulfillment_time = common_utils::consts::DEFAULT_INTENT_FULFILLMENT_TIME;
        pm_routes::ParentPaymentMethodToken::create_key_for_token(&parent_payment_method_token)
            .insert(intent_fulfillment_time, token_data, &state)
            .await?;
    };

    let associated_pm_data = payment_method_response.payment_method_data.clone();

    let associated_payment_methods = common_types::payment_methods::AssociatedPaymentMethods {
        payment_method_token: common_types::payment_methods::AssociatedPaymentMethodTokenType::PaymentMethodSessionToken(parent_payment_method_token),
    };

    let update_payment_method_session = hyperswitch_domain_models::payment_methods::PaymentMethodsSessionUpdateEnum::UpdateAssociatedPaymentMethods {
        associated_payment_methods:  Some(vec![associated_payment_methods])
    };

    let payment_method_session = db
        .update_payment_method_session(
            platform.get_provider().get_key_store(),
            &payment_method_session_id,
            update_payment_method_session,
            payment_method_session,
        )
        .await
        .to_not_found_response(errors::ApiErrorResponse::GenericNotFoundError {
            message: "payment methods session does not exist or has expired".to_string(),
        })
        .attach_printable("Failed to update payment methods session from db")?;

    let payments_response: Option<api_models::payments::PaymentsResponse> = None;

    let tokenization_response = match (
        payment_method_session.tokenization_data.clone(),
        payment_method_session.customer_id.clone(),
    ) {
        (Some(tokenization_data), Some(customer_id)) => {
            let tokenization_response = tokenization_core::create_vault_token_core(
                state.clone(),
                platform.get_provider().clone(),
                api_models::tokenization::GenericTokenizationRequest {
                    customer_id: customer_id.clone(),
                    token_request: tokenization_data,
                },
            )
            .await?;
            let token = match tokenization_response {
                services::ApplicationResponse::Json(response) => Some(response),
                _ => None,
            };
            Some(token)
        }
        _ => None,
    };

    logger::debug!(?tokenization_response, "Tokenization response");

    //TODO: update the payment method session with the payment id and payment method id
    let payment_method_session_response = transformers::generate_payment_method_session_response(
        payment_method_session,
        Secret::new("CLIENT_SECRET_REDACTED".to_string()),
        None, // sdk_authorization is not returned for non-create flows
        payments_response,
        (tokenization_response.flatten()),
        payment_method_response.storage_type,
        payment_method_response.card_cvc_token_storage,
        associated_pm_data,
        payment_method_response.network_token,
        None,
    );

    Ok(services::ApplicationResponse::Json(
        payment_method_session_response,
    ))
}

#[cfg(feature = "v2")]
impl pm_types::SavedPMLPaymentsInfo {
    pub async fn form_payments_info(
        payment_intent: PaymentIntent,
        platform: &domain::Platform,
        profile: domain::Profile,
        db: &dyn StorageInterface,
        key_manager_state: &util_types::keymanager::KeyManagerState,
    ) -> RouterResult<Self> {
        let collect_cvv_during_payment = profile.should_collect_cvv_during_payment;

        let off_session_payment_flag = matches!(
            payment_intent.setup_future_usage,
            common_enums::FutureUsage::OffSession
        );

        let is_connector_agnostic_mit_enabled =
            profile.is_connector_agnostic_mit_enabled.unwrap_or(false);

        Ok(Self {
            payment_intent,
            profile,
            collect_cvv_during_payment,
            off_session_payment_flag,
            is_connector_agnostic_mit_enabled,
        })
    }

    pub async fn perform_payment_ops(
        &self,
        state: &SessionState,
        parent_payment_method_token: Option<String>,
        pma: &api::CustomerPaymentMethodResponseItem,
        pm_list_context: PaymentMethodListContext,
    ) -> RouterResult<()> {
        let token = parent_payment_method_token
            .as_ref()
            .get_required_value("parent_payment_method_token")?;
        let token_data = pm_list_context
            .get_token_data()
            .get_required_value("PaymentTokenData")?;

        let intent_fulfillment_time = self
            .profile
            .get_order_fulfillment_time()
            .unwrap_or(common_utils::consts::DEFAULT_INTENT_FULFILLMENT_TIME);

        pm_routes::ParentPaymentMethodToken::create_key_for_token(token)
            .insert(intent_fulfillment_time, token_data, state)
            .await?;

        Ok(())
    }
}

#[cfg(feature = "v2")]
async fn add_single_use_token_to_store(
    state: &SessionState,
    key: domain::SingleUseTokenKey,
    value: domain::SingleUsePaymentMethodToken,
) -> CustomResult<(), errors::StorageError> {
    let redis_connection = state
        .store
        .get_redis_conn()
        .map_err(Into::<errors::StorageError>::into)?;

    redis_connection
        .serialize_and_set_key_with_expiry(
            &domain::SingleUseTokenKey::get_store_key(&key).into(),
            value,
            consts::DEFAULT_PAYMENT_METHOD_STORE_TTL,
        )
        .await
        .change_context(errors::StorageError::KVError)
        .attach_printable("Failed to insert payment method token to redis")?;
    Ok(())
}

#[cfg(feature = "v2")]
async fn get_single_use_token_from_store(
    state: &SessionState,
    key: domain::SingleUseTokenKey,
) -> CustomResult<Option<domain::SingleUsePaymentMethodToken>, errors::StorageError> {
    let redis_connection = state
        .store
        .get_redis_conn()
        .map_err(Into::<errors::StorageError>::into)?;

    redis_connection
        .get_and_deserialize_key::<Option<domain::SingleUsePaymentMethodToken>>(
            &domain::SingleUseTokenKey::get_store_key(&key).into(),
            "SingleUsePaymentMethodToken",
        )
        .await
        .change_context(errors::StorageError::KVError)
        .attach_printable("Failed to get payment method token from redis")
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
async fn fetch_payment_method(
    state: &SessionState,
    provider: &domain::Provider,
    payment_method_id: &id_type::GlobalPaymentMethodId,
) -> RouterResult<domain::PaymentMethod> {
    let db = &state.store;
    let merchant_account = provider.get_account();
    let key_store = provider.get_key_store();

    db.find_payment_method(
        key_store,
        payment_method_id,
        merchant_account.storage_scheme,
    )
    .await
    .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)
    .attach_printable("Payment method not found for network token status check")
}

#[cfg(feature = "v2")]
pub async fn check_network_token_status(
    state: SessionState,
    provider: domain::Provider,
    payment_method_id: id_type::GlobalPaymentMethodId,
) -> RouterResponse<payment_methods::NetworkTokenStatusCheckResponse> {
    // Retrieve the payment method from the database
    let payment_method = fetch_payment_method(&state, &provider, &payment_method_id).await?;

    // Call the network token status check function
    let network_token_status_check_response = if payment_method.status
        == common_enums::PaymentMethodStatus::Active
    {
        // Check if the payment method has network token data
        when(
            payment_method
                .network_token_requestor_reference_id
                .is_none(),
            || {
                Err(errors::ApiErrorResponse::InvalidDataValue {
                    field_name: "payment_method_id",
                })
            },
        )?;
        match network_tokenization::do_status_check_for_network_token(&state, &payment_method).await
        {
            Ok(network_token_details) => {
                let status = match network_token_details.payload.token_status {
                    pm_types::TokenStatus::Active => api_enums::TokenStatus::Active,
                    pm_types::TokenStatus::Suspended => api_enums::TokenStatus::Suspended,
                    pm_types::TokenStatus::Inactive => api_enums::TokenStatus::Inactive,
                    pm_types::TokenStatus::Expired => api_enums::TokenStatus::Expired,
                    pm_types::TokenStatus::Deleted => api_enums::TokenStatus::Deleted,
                };

                payment_methods::NetworkTokenStatusCheckResponse::SuccessResponse(
                    payment_methods::NetworkTokenStatusCheckSuccessResponse {
                        status,
                        token_expiry_month: network_token_details.payload.token_expiry_month,
                        token_expiry_year: network_token_details.payload.token_expiry_year,
                        card_last_four: network_token_details.payload.card_last_four,
                        card_expiry_month: network_token_details.payload.card_expiry_month,
                        card_expiry_year: network_token_details.payload.card_expiry_year,
                        token_last_four: network_token_details.payload.token_last_four,
                        payment_method_id,
                        customer_id: payment_method
                            .customer_id
                            .clone()
                            .get_required_value("GlobalCustomerId")?,
                    },
                )
            }
            Err(e) => {
                let err_message = e.current_context().to_string();
                logger::debug!("Network token status check failed: {:?}", e);

                payment_methods::NetworkTokenStatusCheckResponse::FailureResponse(
                    payment_methods::NetworkTokenStatusCheckFailureResponse {
                        error_message: err_message,
                    },
                )
            }
        }
    } else {
        let err_message = "Payment Method is not active".to_string();
        logger::debug!("Payment Method is not active");

        payment_methods::NetworkTokenStatusCheckResponse::FailureResponse(
            payment_methods::NetworkTokenStatusCheckFailureResponse {
                error_message: err_message,
            },
        )
    };
    Ok(services::ApplicationResponse::Json(
        network_token_status_check_response,
    ))
}

#[cfg(feature = "v2")]
pub async fn payment_method_get_token_details_core(
    state: SessionState,
    provider: domain::Provider,
    temporary_token: String,
) -> RouterResponse<payment_methods::PaymentMethodGetTokenDetailsResponse> {
    let token_data = pm_routes::ParentPaymentMethodToken::create_key_for_token(&temporary_token)
        .get_data_for_token(&state)
        .await;

    match token_data {
        Ok(storage::PaymentTokenData::PermanentCard(card_token_data)) => {
            Ok(services::ApplicationResponse::Json(
                payment_methods::PaymentMethodGetTokenDetailsResponse {
                    id: card_token_data.payment_method_id,
                    payment_method_token: temporary_token,
                },
            ))
        }
        Ok(storage::PaymentTokenData::BankDebit(bank_debit_token_data)) => {
            Ok(services::ApplicationResponse::Json(
                payment_methods::PaymentMethodGetTokenDetailsResponse {
                    id: bank_debit_token_data.payment_method_id,
                    payment_method_token: temporary_token,
                },
            ))
        }
        Ok(_) => Err(errors::ApiErrorResponse::PaymentMethodNotFound.into()),
        Err(e) => Err(e),
    }
}

#[cfg(feature = "v2")]
impl<'a> pm_types::PaymentMethodUpdateHandler<'a> {
    pub async fn generate(
        state: &'a SessionState,
        platform: &'a domain::Platform,
        profile: &'a domain::Profile,
        request: DomainPaymentMethodUpdate,
        payment_method_id: &'a id_type::GlobalPaymentMethodId,
    ) -> RouterResult<Self> {
        let payment_method = state
            .store
            .find_payment_method(
                platform.get_provider().get_key_store(),
                payment_method_id,
                platform.get_provider().get_account().storage_scheme,
            )
            .await
            .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;

        let handler = Self {
            platform,
            profile,
            request,
            payment_method,
            state,
        };
        handler.validate()?;
        Ok(handler)
    }

    fn validate(&self) -> RouterResult<()> {
        let payment_method = &self.payment_method;
        when(
            payment_method.status == enums::PaymentMethodStatus::Redacted,
            || {
                Err(errors::ApiErrorResponse::InvalidRequestData {
                    message: "Redacted Payment Method cannot be updated".to_string(),
                })
            },
        )?;
        Ok(())
    }

    fn is_pm_metadata_changed_for_same_fingerprint(
        &self,
        vault_request_data: &domain::PaymentMethodVaultingData,
    ) -> bool {
        let updated =
            Self::extract_domain_pm_metadata(vault_request_data.get_payment_methods_data());
        let existing =
            self.payment_method
                .payment_method_data
                .clone()
                .and_then(|payment_method_data| {
                    Self::extract_domain_pm_metadata(payment_method_data.into_inner())
                });

        match (existing, updated) {
            (Some(existing), Some(updated)) => existing != updated,
            (None, Some(_)) => true,
            _ => false,
        }
    }

    fn extract_domain_pm_metadata(
        pm_data: domain::PaymentMethodsData,
    ) -> Option<payment_methods::PaymentMethodUpdateData> {
        match pm_data {
            domain::PaymentMethodsData::Card(card) => Some(
                payment_methods::PaymentMethodUpdateData::Card(payment_methods::CardDetailUpdate {
                    card_holder_name: card.card_holder_name,
                    nick_name: card.nick_name,
                    card_cvc: None,
                }),
            ),
            domain::PaymentMethodsData::BankDebit(bank_debit) => match bank_debit {
                domain::BankDebitDetailsPaymentMethod::AchBankDebit {
                    bank_account_holder_name,
                    ..
                } => Some(payment_methods::PaymentMethodUpdateData::BankDebit(
                    payment_methods::BankDebitDetailUpdate::Ach {
                        bank_account_holder_name,
                    },
                )),
            },
            _ => None,
        }
    }

    pub async fn update_cvc_if_required(
        &self,
    ) -> RouterResult<Option<payment_methods::CardCVCTokenStorageDetails>> {
        let card_cvc = self.request.fetch_card_cvc_update();
        let card_cvc_token_details = card_cvc
            .async_map(|cvc| {
                vault::insert_cvc_using_payment_token(
                    self.state,
                    self.payment_method.get_id().get_string_repr(),
                    cvc,
                    common_utils::consts::DEFAULT_INTENT_FULFILLMENT_TIME,
                    self.platform.get_provider().get_key_store(),
                )
            })
            .await
            .transpose()
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to insert encrypted cvc in redis")?;

        Ok(card_cvc_token_details)
    }

    pub async fn perform_vaulting_operations_if_required(
        &self,
    ) -> RouterResult<(
        Option<domain::PaymentMethodVaultingData>,
        Option<pm_types::AddVaultResponse>,
    )> {
        if !self.request.is_payment_method_metadata_update() {
            return Ok((None, None));
        }

        let pmd: domain::PaymentMethodVaultingData =
            Box::pin(vault::retrieve_payment_method_from_vault(
                self.state,
                self.platform,
                self.profile,
                &self.payment_method,
            ))
            .await
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to retrieve payment method from vault")?
            .data;

        let vault_request_data = self
            .request
            .payment_method_data
            .clone()
            .map(|payment_method_data| {
                pm_transforms::generate_pm_vaulting_req_from_update_request(
                    pmd,
                    payment_method_data,
                )
            })
            .transpose()
            .change_context(errors::ApiErrorResponse::NotSupported {
                message: "Payment method type not supported for update".to_string(),
            })?
            .ok_or(errors::ApiErrorResponse::MissingRequiredField {
                field_name: "payment_method_data",
            })?;

        let is_metadata_changed_for_same_fingerprint =
            self.is_pm_metadata_changed_for_same_fingerprint(&vault_request_data);

        let locker = LockerType::from_micro_services_config(&self.state.conf.micro_services);

        locker
            .validate_locker_operations_for_update(
                self.state,
                self.platform,
                self.profile,
                &self.payment_method,
                &vault_request_data,
                is_metadata_changed_for_same_fingerprint,
            )
            .await
    }

    pub async fn update_payment_method_if_required(
        &mut self,
        vault_request_data: Option<domain::PaymentMethodVaultingData>,
        vault_resp: Option<pm_types::AddVaultResponse>,
        network_tokenization_resp: Option<NetworkTokenPaymentMethodDetails>,
    ) -> RouterResult<()> {
        if !self.request.is_payment_method_update_required() && network_tokenization_resp.is_none()
        {
            return Ok(());
        }

        // Get target status from request
        // Domain model always uses PaymentMethodStatus - transformation from AcknowledgementStatus
        // happens at the API boundary (From<api_models::PaymentMethodUpdate> impl)
        let pm_status = self.request.status;

        // Validate status transition
        // Invalid transitions are silently ignored - status won't be updated
        let pm_status = pm_status
            .filter(|&new_status| self.payment_method.status.can_transition_to(new_status));

        let db = self.state.store.as_ref();

        let pm_update = create_pm_additional_data_update(
            vault_request_data.as_ref(),
            self.state,
            self.platform.get_provider().get_key_store(),
            vault_resp
                .as_ref()
                .map(|resp| resp.vault_id.get_string_repr().to_owned()),
            vault_resp
                .as_ref()
                .and_then(|resp| resp.fingerprint_id.clone()),
            &self.payment_method,
            self.request.connector_token_details.clone(),
            self.request.network_transaction_id.clone(),
            network_tokenization_resp,
            None,
            None,
            None,
            pm_status,
            self.platform.get_initiator(),
        )
        .await
        .attach_printable("Unable to create Payment method data")?;

        let updated_payment_method = db
            .update_payment_method(
                self.platform.get_provider().get_key_store(),
                self.payment_method.clone(),
                pm_update,
                self.platform.get_provider().get_account().storage_scheme,
                Some(payment_method_modular_backward_compat_action(
                    self.state,
                    &self.platform.get_provider().get_account().organization_id,
                )),
            )
            .await
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to update payment method in db")?;

        self.payment_method = updated_payment_method;
        Ok(())
    }

    pub fn generate_response(
        &self,
        card_cvc_token_details: Option<payment_methods::CardCVCTokenStorageDetails>,
    ) -> RouterResult<payment_methods::PaymentMethodResponse> {
        let response = pm_transforms::generate_payment_method_response(
            &self.payment_method,
            &None,
            common_enums::StorageType::Persistent,
            card_cvc_token_details,
            self.payment_method.customer_id.clone(),
            None,
            self.payment_method
                .payment_method_billing_address
                .clone()
                .map(|billing| billing.get_inner().clone().into()),
            self.request.acknowledgement_status,
        )?;

        Ok(response)
    }
}
