use api_models::customers::CustomerDocumentDetails;
#[cfg(feature = "v2")]
use api_models::payment_methods::PaymentMethodId;
use common_types::primitive_wrappers::CustomerListLimit;
use common_utils::{
    crypto::Encryptable,
    errors::ReportSwitchExt,
    ext_traits::AsyncExt,
    id_type, pii, type_name,
    types::{
        keymanager::{Identifier, KeyManagerState, ToEncryptable},
        Description,
    },
};
use error_stack::{report, ResultExt};
use hyperswitch_domain_models::{
    payment_methods as payment_methods_domain, type_encryption::AsyncLift,
};
use hyperswitch_masking::{ExposeInterface, Secret, SwitchStrategy};
use payment_methods::controller::PaymentMethodsController;
use router_env::{instrument, tracing};

#[cfg(feature = "v2")]
use crate::core::payment_methods::delete_payment_method_by_record;
#[cfg(feature = "v2")]
use crate::core::utils::create_encrypted_data;
#[cfg(feature = "v1")]
use crate::utils::CustomerAddress;
use crate::{
    core::{
        errors::{self, StorageErrorExt},
        payment_methods::{cards, network_tokenization},
    },
    db::StorageInterface,
    pii::PeekInterface,
    routes::{metrics, SessionState},
    services,
    types::{
        api::customers,
        domain::{
            self,
            types::{self, CryptoOperation},
        },
        storage::{self, enums},
    },
};

pub const REDACTED: &str = "Redacted";

pub fn is_global_customer_id_format(input: &str) -> bool {
    let mut parts = input.split('_');
    match (parts.next(), parts.next(), parts.next(), parts.next()) {
        (Some(cell_id), Some(entity), Some(uuid), None) => {
            !cell_id.is_empty()
                && cell_id
                    .chars()
                    .all(|character| character.is_ascii_lowercase() || character.is_ascii_digit())
                && entity == "cus"
                && uuid.len() == 32
                && uuid::Uuid::parse_str(uuid).is_ok()
        }
        _ => false,
    }
}

pub fn is_customer_id_in_global_format(customer_id: &id_type::CustomerId) -> bool {
    is_global_customer_id_format(customer_id.get_string_repr())
}

#[instrument(skip(state))]
pub async fn create_customer(
    state: SessionState,
    provider: domain::Provider,
    initiator: Option<domain::Initiator>,
    customer_data: customers::CustomerRequest,
    connector_customer_details: Option<Vec<payment_methods_domain::ConnectorCustomerDetails>>,
) -> errors::CustomerResponse<customers::CustomerResponse> {
    customer_data
        .document_details
        .as_ref()
        .map(|doc_details| doc_details.validate())
        .transpose()
        .map_err(|err| errors::CustomersErrorResponse::InvalidRequestData {
            message: err.to_string(),
        })?;

    let db: &dyn StorageInterface = state.store.as_ref();
    let key_manager_state = &(&state).into();

    let merchant_reference_id = customer_data.get_merchant_reference_id();

    let merchant_id = provider.get_account().get_id();

    let merchant_reference_id_customer = MerchantReferenceIdForCustomer {
        merchant_reference_id: merchant_reference_id.as_ref(),
        merchant_id,
        merchant_account: provider.get_account(),
        key_store: provider.get_key_store(),
    };

    // We first need to validate whether the customer with the given customer id already exists
    // this may seem like a redundant db call, as the insert_customer will anyway return this error
    //
    // Consider a scenario where the address is inserted and then when inserting the customer,
    // it errors out, now the address that was inserted is not deleted

    merchant_reference_id_customer
        .verify_if_merchant_reference_not_present_by_optional_merchant_reference_id(db)
        .await?;

    let domain_customer = customer_data
        .create_domain_model_from_request(
            &connector_customer_details,
            db,
            &merchant_reference_id,
            &provider,
            initiator.as_ref(),
            key_manager_state,
            &state,
        )
        .await?;

    let customer = db
        .insert_customer(
            domain_customer,
            provider.get_key_store(),
            provider.get_account().storage_scheme,
        )
        .await
        .to_duplicate_response(errors::CustomersErrorResponse::CustomerAlreadyExists)?;

    customer_data.generate_response(&customer)
}

#[allow(clippy::too_many_arguments)]
#[async_trait::async_trait]
trait CustomerCreateBridge {
    async fn create_domain_model_from_request<'a>(
        &'a self,
        connector_customer_details: &'a Option<
            Vec<payment_methods_domain::ConnectorCustomerDetails>,
        >,
        db: &'a dyn StorageInterface,
        merchant_reference_id: &'a Option<id_type::CustomerId>,
        provider: &'a domain::Provider,
        initiator: Option<&'a domain::Initiator>,
        key_manager_state: &'a KeyManagerState,
        state: &'a SessionState,
    ) -> errors::CustomResult<domain::Customer, errors::CustomersErrorResponse>;

    fn generate_response<'a>(
        &'a self,
        customer: &'a domain::Customer,
    ) -> errors::CustomerResponse<customers::CustomerResponse>;
}

#[cfg(feature = "v1")]
#[async_trait::async_trait]
impl CustomerCreateBridge for customers::CustomerRequest {
    async fn create_domain_model_from_request<'a>(
        &'a self,
        connector_customer_details: &'a Option<
            Vec<payment_methods_domain::ConnectorCustomerDetails>,
        >,
        db: &'a dyn StorageInterface,
        merchant_reference_id: &'a Option<id_type::CustomerId>,
        provider: &'a domain::Provider,
        initiator: Option<&'a domain::Initiator>,
        key_manager_state: &'a KeyManagerState,
        state: &'a SessionState,
    ) -> errors::CustomResult<domain::Customer, errors::CustomersErrorResponse> {
        // Setting default billing address to Db
        let address = self.get_address();
        let merchant_id = provider.get_account().get_id();
        let key = provider.get_key_store().key.get_inner().peek();

        let customer_billing_address_struct = AddressStructForDbEntry {
            address: address.as_ref(),
            customer_data: self,
            merchant_id,
            customer_id: merchant_reference_id.as_ref(),
            storage_scheme: provider.get_account().storage_scheme,
            key_store: provider.get_key_store(),
            state,
        };

        let document_details_encrypted = self
            .document_details
            .clone()
            .async_lift(|inner| async move {
                let encoded_inner = inner
                    .map(|details| CustomerDocumentDetails::to(&details))
                    .transpose()
                    .change_context(errors::CustomersErrorResponse::InternalServerError)
                    .attach_printable(
                        "Failed to encode customer document details for encryption",
                    )?;

                let crypto_result = types::crypto_operation(
                    &state.into(),
                    common_utils::type_name!(domain::Customer),
                    CryptoOperation::EncryptOptional(encoded_inner),
                    Identifier::Merchant(merchant_id.clone()),
                    provider.get_key_store().key.peek(),
                )
                .await
                .change_context(errors::CustomersErrorResponse::InternalServerError)
                .attach_printable("Crypto operation failed during document details encryption")?;

                let final_val = crypto_result
                    .try_into_optionaloperation()
                    .change_context(errors::CustomersErrorResponse::InternalServerError)
                    .attach_printable("Failed to parse encrypted document details")?;

                Ok(final_val)
            })
            .await
            .attach_printable("Unable to encrypt document_details")?;

        let address_from_db = customer_billing_address_struct
            .encrypt_customer_address_and_set_to_db(db)
            .await?;

        let encrypted_data = types::crypto_operation(
            key_manager_state,
            type_name!(domain::Customer),
            CryptoOperation::BatchEncrypt(domain::FromRequestEncryptableCustomer::to_encryptable(
                domain::FromRequestEncryptableCustomer {
                    name: self.name.clone(),
                    email: self.email.clone().map(|a| a.expose().switch_strategy()),
                    phone: self.phone.clone(),
                    tax_registration_id: self.tax_registration_id.clone(),
                },
            )),
            Identifier::Merchant(provider.get_key_store().merchant_id.clone()),
            key,
        )
        .await
        .and_then(|val| val.try_into_batchoperation())
        .switch()
        .attach_printable("Failed while encrypting Customer")?;

        let encryptable_customer =
            domain::FromRequestEncryptableCustomer::from_encryptable(encrypted_data)
                .change_context(errors::CustomersErrorResponse::InternalServerError)?;

        let connector_customer = connector_customer_details.as_ref().map(|details_vec| {
            let mut map = serde_json::Map::new();
            for details in details_vec {
                let merchant_connector_id =
                    details.merchant_connector_id.get_string_repr().to_string();
                let connector_customer_id = details.connector_customer_id.clone();
                map.insert(merchant_connector_id, connector_customer_id.into());
            }
            pii::SecretSerdeValue::new(serde_json::Value::Object(map))
        });

        Ok(domain::Customer::new(
            merchant_reference_id
                .to_owned()
                .ok_or(errors::CustomersErrorResponse::InternalServerError)?,
            merchant_id.to_owned(),
            encryptable_customer.name,
            encryptable_customer.email.map(|email| {
                let encryptable: Encryptable<Secret<String, pii::EmailStrategy>> = Encryptable::new(
                    email.clone().into_inner().switch_strategy(),
                    email.into_encrypted(),
                );
                encryptable
            }),
            encryptable_customer.phone,
            self.phone_country_code.clone(),
            self.description.clone(),
            self.metadata.clone(),
            connector_customer,
            address_from_db.clone().map(|addr| addr.address_id),
            encryptable_customer.tax_registration_id,
            document_details_encrypted,
            initiator.and_then(|initiator| initiator.to_created_by()),
            initiator.and_then(|initiator| initiator.to_created_by()),
            id_type::GlobalCustomerId::generate(&state.conf.cell_information.id),
        ))
    }

    fn generate_response<'a>(
        &'a self,
        customer: &'a domain::Customer,
    ) -> errors::CustomerResponse<customers::CustomerResponse> {
        let address = self.get_address();
        Ok(services::ApplicationResponse::Json(
            customers::CustomerResponse::try_from((customer.clone(), address))
                .change_context(errors::CustomersErrorResponse::InternalServerError)
                .attach_printable("Failed to convert domain customer to CustomerResponse")?,
        ))
    }
}

#[cfg(feature = "v2")]
#[async_trait::async_trait]
impl CustomerCreateBridge for customers::CustomerRequest {
    async fn create_domain_model_from_request<'a>(
        &'a self,
        connector_customer_details: &'a Option<
            Vec<payment_methods_domain::ConnectorCustomerDetails>,
        >,
        _db: &'a dyn StorageInterface,
        merchant_reference_id: &'a Option<id_type::CustomerId>,
        provider: &'a domain::Provider,
        initiator: Option<&'a domain::Initiator>,
        key_state: &'a KeyManagerState,
        state: &'a SessionState,
    ) -> errors::CustomResult<domain::Customer, errors::CustomersErrorResponse> {
        let default_customer_billing_address = self.get_default_customer_billing_address();
        let encrypted_customer_billing_address = default_customer_billing_address
            .async_map(|billing_address| {
                create_encrypted_data(
                    key_state,
                    provider.get_key_store(),
                    billing_address,
                    common_utils::type_name!(diesel_models::payment_method::PaymentMethod),
                )
            })
            .await
            .transpose()
            .change_context(errors::CustomersErrorResponse::InternalServerError)
            .attach_printable("Unable to encrypt default customer billing address")?;
        let default_customer_shipping_address = self.get_default_customer_shipping_address();
        let encrypted_customer_shipping_address = default_customer_shipping_address
            .async_map(|shipping_address| {
                create_encrypted_data(
                    key_state,
                    provider.get_key_store(),
                    shipping_address,
                    common_utils::type_name!(diesel_models::payment_method::PaymentMethod),
                )
            })
            .await
            .transpose()
            .change_context(errors::CustomersErrorResponse::InternalServerError)
            .attach_printable("Unable to encrypt default customer shipping address")?;

        let merchant_id = provider.get_account().get_id().clone();
        let key = provider.get_key_store().key.get_inner().peek();

        let encrypted_data = types::crypto_operation(
            key_state,
            type_name!(domain::Customer),
            types::CryptoOperation::BatchEncrypt(
                domain::FromRequestEncryptableCustomer::to_encryptable(
                    domain::FromRequestEncryptableCustomer {
                        name: Some(self.name.clone()),
                        email: Some(self.email.clone().expose().switch_strategy()),
                        phone: self.phone.clone(),
                        tax_registration_id: self.tax_registration_id.clone(),
                    },
                ),
            ),
            Identifier::Merchant(provider.get_key_store().merchant_id.clone()),
            key,
        )
        .await
        .and_then(|val| val.try_into_batchoperation())
        .switch()
        .attach_printable("Failed while encrypting Customer")?;

        let encryptable_customer =
            domain::FromRequestEncryptableCustomer::from_encryptable(encrypted_data)
                .change_context(errors::CustomersErrorResponse::InternalServerError)?;

        let connector_customer = connector_customer_details.as_ref().map(|details_vec| {
            let map: std::collections::HashMap<_, _> = details_vec
                .iter()
                .map(|details| {
                    (
                        details.merchant_connector_id.clone(),
                        details.connector_customer_id.to_string(),
                    )
                })
                .collect();
            common_types::customers::ConnectorCustomerMap::new(map)
        });

        Ok(domain::Customer {
            id: id_type::GlobalCustomerId::generate(&state.conf.cell_information.id),
            merchant_reference_id: merchant_reference_id.to_owned(),
            merchant_id,
            name: encryptable_customer.name,
            email: encryptable_customer.email.map(|email| {
                let encryptable: Encryptable<Secret<String, pii::EmailStrategy>> = Encryptable::new(
                    email.clone().into_inner().switch_strategy(),
                    email.into_encrypted(),
                );
                encryptable
            }),
            phone: encryptable_customer.phone,
            description: self.description.clone(),
            phone_country_code: self.phone_country_code.clone(),
            metadata: self.metadata.clone(),
            connector_customer,
            created_at: common_utils::date_time::now(),
            modified_at: common_utils::date_time::now(),
            default_payment_method_id: None,
            updated_by: None,
            default_billing_address: encrypted_customer_billing_address,
            default_shipping_address: encrypted_customer_shipping_address,
            version: common_types::consts::API_VERSION,
            status: common_enums::DeleteStatus::Active,
            tax_registration_id: encryptable_customer.tax_registration_id,
            document_details: None,
            created_by: initiator.and_then(|initiator| initiator.to_created_by()),
            last_modified_by: initiator.and_then(|initiator| initiator.to_created_by()),
        })
    }

    fn generate_response<'a>(
        &'a self,
        customer: &'a domain::Customer,
    ) -> errors::CustomerResponse<customers::CustomerResponse> {
        Ok(services::ApplicationResponse::Json(
            customers::CustomerResponse::try_from(customer.clone())
                .change_context(errors::CustomersErrorResponse::InternalServerError)
                .attach_printable("Failed to convert domain customer to CustomerResponse")?,
        ))
    }
}

#[cfg(feature = "v1")]
struct AddressStructForDbEntry<'a> {
    address: Option<&'a api_models::payments::AddressDetails>,
    customer_data: &'a customers::CustomerRequest,
    merchant_id: &'a id_type::MerchantId,
    customer_id: Option<&'a id_type::CustomerId>,
    storage_scheme: common_enums::MerchantStorageScheme,
    key_store: &'a domain::MerchantKeyStore,
    state: &'a SessionState,
}

#[cfg(feature = "v1")]
impl AddressStructForDbEntry<'_> {
    async fn encrypt_customer_address_and_set_to_db(
        &self,
        db: &dyn StorageInterface,
    ) -> errors::CustomResult<Option<domain::Address>, errors::CustomersErrorResponse> {
        let encrypted_customer_address = self
            .address
            .async_map(|addr| async {
                self.customer_data
                    .get_domain_address(
                        self.state,
                        addr.clone(),
                        self.merchant_id,
                        self.customer_id
                            .ok_or(errors::CustomersErrorResponse::InternalServerError)?, // should we raise error since in v1 appilcation is supposed to have this id or generate it at this point.
                        self.key_store.key.get_inner().peek(),
                        self.storage_scheme,
                    )
                    .await
                    .switch()
                    .attach_printable("Failed while encrypting address")
            })
            .await
            .transpose()?;

        encrypted_customer_address
            .async_map(|encrypt_add| async {
                db.insert_address_for_customers(encrypt_add, self.key_store)
                    .await
                    .switch()
                    .attach_printable("Failed while inserting new address")
            })
            .await
            .transpose()
    }
}

struct MerchantReferenceIdForCustomer<'a> {
    merchant_reference_id: Option<&'a id_type::CustomerId>,
    merchant_id: &'a id_type::MerchantId,
    merchant_account: &'a domain::MerchantAccount,
    key_store: &'a domain::MerchantKeyStore,
}

#[cfg(feature = "v1")]
impl<'a> MerchantReferenceIdForCustomer<'a> {
    async fn verify_if_merchant_reference_not_present_by_optional_merchant_reference_id(
        &self,
        db: &dyn StorageInterface,
    ) -> Result<Option<()>, error_stack::Report<errors::CustomersErrorResponse>> {
        self.merchant_reference_id
            .async_map(|cust| async {
                self.verify_if_merchant_reference_not_present_by_merchant_reference_id(cust, db)
                    .await
            })
            .await
            .transpose()
    }

    async fn verify_if_merchant_reference_not_present_by_merchant_reference_id(
        &self,
        cus: &'a id_type::CustomerId,
        db: &dyn StorageInterface,
    ) -> Result<(), error_stack::Report<errors::CustomersErrorResponse>> {
        match db
            .find_customer_by_customer_id_merchant_id(
                cus,
                self.merchant_id,
                self.key_store,
                self.merchant_account.storage_scheme,
            )
            .await
        {
            Err(err) => {
                if !err.current_context().is_db_not_found() {
                    Err(err).switch()
                } else {
                    Ok(())
                }
            }
            Ok(_) => Err(report!(
                errors::CustomersErrorResponse::CustomerAlreadyExists
            )),
        }
    }
}

#[cfg(feature = "v2")]
impl<'a> MerchantReferenceIdForCustomer<'a> {
    async fn verify_if_merchant_reference_not_present_by_optional_merchant_reference_id(
        &self,
        db: &dyn StorageInterface,
    ) -> Result<Option<()>, error_stack::Report<errors::CustomersErrorResponse>> {
        self.merchant_reference_id
            .async_map(|merchant_ref| async {
                self.verify_if_merchant_reference_not_present_by_merchant_reference(
                    merchant_ref,
                    db,
                )
                .await
            })
            .await
            .transpose()
    }

    async fn verify_if_merchant_reference_not_present_by_merchant_reference(
        &self,
        merchant_ref: &'a id_type::CustomerId,
        db: &dyn StorageInterface,
    ) -> Result<(), error_stack::Report<errors::CustomersErrorResponse>> {
        match db
            .find_customer_by_merchant_reference_id_merchant_id(
                merchant_ref,
                self.merchant_id,
                self.key_store,
                self.merchant_account.storage_scheme,
            )
            .await
        {
            Err(err) => {
                if !err.current_context().is_db_not_found() {
                    Err(err).switch()
                } else {
                    Ok(())
                }
            }
            Ok(_) => Err(report!(
                errors::CustomersErrorResponse::CustomerAlreadyExists
            )),
        }
    }
}

#[cfg(feature = "v1")]
#[instrument(skip(state))]
pub async fn retrieve_customer(
    state: SessionState,
    provider: domain::Provider,
    _profile_id: Option<id_type::ProfileId>,
    customer_id: id_type::CustomerId,
) -> errors::CustomerResponse<customers::CustomerResponse> {
    let db = state.store.as_ref();

    let response = db
        .find_customer_optional_with_redacted_customer_details_by_customer_id_merchant_id(
            &customer_id,
            provider.get_account().get_id(),
            provider.get_key_store(),
            provider.get_account().storage_scheme,
        )
        .await
        .switch()?
        .ok_or(errors::CustomersErrorResponse::CustomerNotFound)?;

    let address = match &response.address_id {
        Some(address_id) => Some(api_models::payments::AddressDetails::from(
            db.find_address_by_address_id(address_id, provider.get_key_store())
                .await
                .switch()?,
        )),
        None => None,
    };
    Ok(services::ApplicationResponse::Json(
        customers::CustomerResponse::try_from((response, address))
            .change_context(errors::CustomersErrorResponse::InternalServerError)
            .attach_printable("Failed to convert domain customer to CustomerResponse")?,
    ))
}

#[cfg(feature = "v2")]
#[instrument(skip(state))]
pub async fn retrieve_customer(
    state: SessionState,
    provider: domain::Provider,
    id: id_type::GlobalCustomerId,
) -> errors::CustomerResponse<customers::CustomerResponse> {
    let db = state.store.as_ref();

    let response = db
        .find_customer_by_global_id(
            &id,
            provider.get_key_store(),
            provider.get_account().storage_scheme,
        )
        .await
        .switch()?;

    Ok(services::ApplicationResponse::Json(
        customers::CustomerResponse::try_from(response)
            .change_context(errors::CustomersErrorResponse::InternalServerError)
            .attach_printable("Failed to convert domain customer to CustomerResponse")?,
    ))
}

#[cfg(feature = "v2")]
#[instrument(skip(state))]
pub async fn retrieve_customer_by_merchant_reference_id(
    state: SessionState,
    provider: domain::Provider,
    merchant_reference_id: id_type::CustomerId,
) -> errors::CustomerResponse<customers::CustomerResponse> {
    let db = state.store.as_ref();

    let response = db
        .find_customer_by_merchant_reference_id_merchant_id(
            &merchant_reference_id,
            provider.get_account().get_id(),
            provider.get_key_store(),
            provider.get_account().storage_scheme,
        )
        .await
        .switch()?;

    Ok(services::ApplicationResponse::Json(
        customers::CustomerResponse::try_from(response)
            .change_context(errors::CustomersErrorResponse::InternalServerError)
            .attach_printable("Failed to convert domain customer to CustomerResponse")?,
    ))
}

#[instrument(skip(state))]
pub async fn list_customers(
    state: SessionState,
    provider: domain::Provider,
    _profile_id_list: Option<Vec<id_type::ProfileId>>,
    request: customers::CustomerListRequest,
) -> errors::CustomerResponse<Vec<customers::CustomerResponse>> {
    let db = state.store.as_ref();

    let customer_list_constraints = crate::db::customers::CustomerListConstraints {
        limit: request
            .limit
            .unwrap_or(crate::consts::DEFAULT_LIST_API_LIMIT),
        offset: request.offset,
        customer_id: request.customer_id,
        time_range: None,
    };

    let domain_customers = db
        .list_customers_by_merchant_id(
            provider.get_account().get_id(),
            provider.get_key_store(),
            customer_list_constraints,
        )
        .await
        .switch()?;

    #[cfg(feature = "v1")]
    let customers = domain_customers
        .into_iter()
        .map(|domain_customer| {
            customers::CustomerResponse::try_from((domain_customer, None))
                .change_context(errors::CustomersErrorResponse::InternalServerError)
                .attach_printable("Failed to convert domain customer to CustomerResponse")
        })
        .collect::<Result<_, _>>()?;

    #[cfg(feature = "v2")]
    let customers = domain_customers
        .into_iter()
        .map(customers::CustomerResponse::try_from)
        .collect::<Result<Vec<_>, _>>()
        .change_context(errors::CustomersErrorResponse::InternalServerError)
        .attach_printable("Failed to convert domain customer to CustomerResponse")?;

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

#[instrument(skip(state))]
pub async fn list_customers_with_count(
    state: SessionState,
    provider: domain::Provider,
    request: customers::CustomerListRequestWithConstraints,
) -> errors::CustomerResponse<customers::CustomerListResponse> {
    let db = state.store.as_ref();
    let customer_list_constraints = crate::db::customers::CustomerListConstraints {
        limit: request
            .limit
            .map(|l| *l)
            .unwrap_or_else(|| *CustomerListLimit::default()),
        offset: request.offset,
        customer_id: request.customer_id,
        time_range: request.time_range,
    };

    let domain_customers = db
        .list_customers_by_merchant_id_with_count(
            provider.get_account().get_id(),
            provider.get_key_store(),
            customer_list_constraints,
        )
        .await
        .switch()?;

    #[cfg(feature = "v1")]
    let customers: Vec<customers::CustomerResponse> = domain_customers
        .0
        .into_iter()
        .map(|domain_customer| customers::CustomerResponse::try_from((domain_customer, None)))
        .collect::<Result<Vec<_>, _>>()
        .change_context(errors::CustomersErrorResponse::InternalServerError)
        .attach_printable("Failed to convert domain customer to CustomerResponse")?;

    #[cfg(feature = "v2")]
    let customers = domain_customers
        .0
        .into_iter()
        .map(customers::CustomerResponse::try_from)
        .collect::<Result<Vec<_>, _>>()
        .change_context(errors::CustomersErrorResponse::InternalServerError)
        .attach_printable("Failed to convert domain customer to CustomerResponse")?;

    Ok(services::ApplicationResponse::Json(
        customers::CustomerListResponse {
            data: customers.into_iter().map(|c| c.0).collect(),
            total_count: domain_customers.1,
        },
    ))
}

#[cfg(feature = "v2")]
#[instrument(skip_all)]
pub async fn delete_customer(
    state: SessionState,
    platform: domain::Platform,
    id: id_type::GlobalCustomerId,
    profile: domain::Profile,
) -> errors::CustomerResponse<customers::CustomerDeleteResponse> {
    let db = &*state.store;
    let key_manager_state = &(&state).into();
    id.redact_customer_details_and_generate_response(
        db,
        &platform,
        key_manager_state,
        &state,
        profile,
    )
    .await
}

#[cfg(feature = "v2")]
#[async_trait::async_trait]
impl CustomerDeleteBridge for id_type::GlobalCustomerId {
    async fn redact_customer_details_and_generate_response<'a>(
        &'a self,
        db: &'a dyn StorageInterface,
        platform: &'a domain::Platform,
        key_manager_state: &'a KeyManagerState,
        state: &'a SessionState,
        profile: domain::Profile,
    ) -> errors::CustomerResponse<customers::CustomerDeleteResponse> {
        let provider = platform.get_provider();
        let customer_orig = db
            .find_customer_by_global_id(
                self,
                provider.get_key_store(),
                provider.get_account().storage_scheme,
            )
            .await
            .switch()?;

        let merchant_reference_id = customer_orig.merchant_reference_id.clone();

        let customer_mandates = db.find_mandate_by_global_customer_id(self).await.switch()?;

        for mandate in customer_mandates.into_iter() {
            if mandate.mandate_status == enums::MandateStatus::Active {
                Err(errors::CustomersErrorResponse::MandateActive)?
            }
        }

        match db
            .find_payment_method_list_by_global_customer_id(provider.get_key_store(), self, None)
            .await
        {
            Ok(customer_payment_methods) => {
                for pm in customer_payment_methods.into_iter() {
                    Box::pin(delete_payment_method_by_record(
                        db, state, platform, &profile, pm,
                    ))
                    .await
                    .switch()?;
                }
            }
            Err(error) => {
                if error.current_context().is_db_not_found() {
                    Ok(())
                } else {
                    Err(error)
                        .change_context(errors::CustomersErrorResponse::InternalServerError)
                        .attach_printable(
                            "failed find_payment_method_by_customer_id_merchant_id_list",
                        )
                }?
            }
        };

        let key = provider.get_key_store().key.get_inner().peek();

        let identifier = Identifier::Merchant(provider.get_key_store().merchant_id.clone());
        let redacted_encrypted_value: Encryptable<Secret<_>> = types::crypto_operation(
            key_manager_state,
            type_name!(storage::Address),
            types::CryptoOperation::Encrypt(REDACTED.to_string().into()),
            identifier.clone(),
            key,
        )
        .await
        .and_then(|val| val.try_into_operation())
        .switch()?;

        let redacted_encrypted_email = Encryptable::new(
            redacted_encrypted_value
                .clone()
                .into_inner()
                .switch_strategy(),
            redacted_encrypted_value.clone().into_encrypted(),
        );

        let updated_customer =
            storage::CustomerUpdate::Update(Box::new(storage::CustomerGeneralUpdate {
                name: Some(redacted_encrypted_value.clone()),
                email: Box::new(Some(redacted_encrypted_email)),
                phone: Box::new(Some(redacted_encrypted_value.clone())),
                description: Some(Description::from_str_unchecked(REDACTED)),
                phone_country_code: Some(REDACTED.to_string()),
                metadata: None,
                connector_customer: Box::new(None),
                default_billing_address: None,
                default_shipping_address: None,
                default_payment_method_id: None,
                status: Some(common_enums::DeleteStatus::Redacted),
                tax_registration_id: Some(redacted_encrypted_value.clone()),
                document_details: 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_customer_by_global_id(
            self,
            customer_orig,
            updated_customer,
            provider.get_key_store(),
            provider.get_account().storage_scheme,
        )
        .await
        .switch()?;

        let response = customers::CustomerDeleteResponse {
            id: self.clone(),
            merchant_reference_id,
            customer_deleted: true,
            address_deleted: true,
            payment_methods_deleted: true,
        };
        metrics::CUSTOMER_REDACTED.add(1, &[]);
        Ok(services::ApplicationResponse::Json(response))
    }
}
#[cfg(feature = "v1")]
#[async_trait::async_trait]
trait CustomerDeleteBridge {
    async fn redact_customer_details_and_generate_response<'a>(
        &'a self,
        db: &'a dyn StorageInterface,
        provider: &'a domain::Provider,
        initiator: Option<&'a domain::Initiator>,
        key_manager_state: &'a KeyManagerState,
        state: &'a SessionState,
    ) -> errors::CustomerResponse<customers::CustomerDeleteResponse>;
}
#[cfg(feature = "v2")]
#[async_trait::async_trait]
trait CustomerDeleteBridge {
    async fn redact_customer_details_and_generate_response<'a>(
        &'a self,
        db: &'a dyn StorageInterface,
        platform: &'a domain::Platform,
        key_manager_state: &'a KeyManagerState,
        state: &'a SessionState,
        profile: domain::Profile,
    ) -> errors::CustomerResponse<customers::CustomerDeleteResponse>;
}

#[cfg(feature = "v1")]
#[instrument(skip_all)]
pub async fn delete_customer(
    state: SessionState,
    provider: domain::Provider,
    initiator: Option<domain::Initiator>,
    customer_id: id_type::CustomerId,
) -> errors::CustomerResponse<customers::CustomerDeleteResponse> {
    let db = &*state.store;
    let key_manager_state = &(&state).into();
    customer_id
        .redact_customer_details_and_generate_response(
            db,
            &provider,
            initiator.as_ref(),
            key_manager_state,
            &state,
        )
        .await
}

#[cfg(feature = "v1")]
#[async_trait::async_trait]
impl CustomerDeleteBridge for id_type::CustomerId {
    async fn redact_customer_details_and_generate_response<'a>(
        &'a self,
        db: &'a dyn StorageInterface,
        provider: &'a domain::Provider,
        initiator: Option<&'a domain::Initiator>,
        key_manager_state: &'a KeyManagerState,
        state: &'a SessionState,
    ) -> errors::CustomerResponse<customers::CustomerDeleteResponse> {
        let customer_orig = db
            .find_customer_by_customer_id_merchant_id(
                self,
                provider.get_account().get_id(),
                provider.get_key_store(),
                provider.get_account().storage_scheme,
            )
            .await
            .switch()?;

        let customer_mandates = db
            .find_mandate_by_merchant_id_customer_id(provider.get_account().get_id(), self)
            .await
            .switch()?;

        for mandate in customer_mandates.into_iter() {
            if mandate.mandate_status == enums::MandateStatus::Active {
                Err(errors::CustomersErrorResponse::MandateActive)?
            }
        }

        match db
            .find_payment_method_by_customer_id_merchant_id_list(
                provider.get_key_store(),
                self,
                provider.get_account().get_id(),
                None,
            )
            .await
        {
            // check this in review
            Ok(customer_payment_methods) => {
                for pm in customer_payment_methods.into_iter() {
                    if pm.get_payment_method_type() == Some(enums::PaymentMethod::Card) {
                        cards::PmCards { state, provider }
                            .delete_card_from_locker(
                                self,
                                provider.get_account().get_id(),
                                pm.locker_id.as_ref().unwrap_or(&pm.payment_method_id),
                            )
                            .await
                            .switch()?;

                        if let Some(network_token_ref_id) = pm.network_token_requestor_reference_id
                        {
                            network_tokenization::delete_network_token_from_locker_and_token_service(
                            state,
                            self,
                            provider.get_account().get_id(),
                            pm.payment_method_id.clone(),
                            pm.network_token_locker_id,
                            network_token_ref_id,
                            provider,
                        )
                        .await
                        .switch()?;
                        }
                    }

                    db.delete_payment_method_by_merchant_id_payment_method_id(
                        provider.get_key_store(),
                        provider.get_account().get_id(),
                        &pm.payment_method_id,
                    )
                    .await
                    .change_context(errors::CustomersErrorResponse::InternalServerError)
                    .attach_printable(
                        "failed to delete payment method while redacting customer details",
                    )?;
                }
            }
            Err(error) => {
                if error.current_context().is_db_not_found() {
                    Ok(())
                } else {
                    Err(error)
                        .change_context(errors::CustomersErrorResponse::InternalServerError)
                        .attach_printable(
                            "failed find_payment_method_by_customer_id_merchant_id_list",
                        )
                }?
            }
        };

        let key = provider.get_key_store().key.get_inner().peek();
        let identifier = Identifier::Merchant(provider.get_key_store().merchant_id.clone());
        let redacted_encrypted_value: Encryptable<Secret<_>> = types::crypto_operation(
            key_manager_state,
            type_name!(storage::Address),
            CryptoOperation::Encrypt(REDACTED.to_string().into()),
            identifier.clone(),
            key,
        )
        .await
        .and_then(|val| val.try_into_operation())
        .switch()?;

        let redacted_encrypted_email = Encryptable::new(
            redacted_encrypted_value
                .clone()
                .into_inner()
                .switch_strategy(),
            redacted_encrypted_value.clone().into_encrypted(),
        );

        let update_address = storage::AddressUpdate::Update {
            city: Some(REDACTED.to_string()),
            country: None,
            line1: Some(redacted_encrypted_value.clone()),
            line2: Some(redacted_encrypted_value.clone()),
            line3: Some(redacted_encrypted_value.clone()),
            state: Some(redacted_encrypted_value.clone()),
            zip: Some(redacted_encrypted_value.clone()),
            first_name: Some(redacted_encrypted_value.clone()),
            last_name: Some(redacted_encrypted_value.clone()),
            phone_number: Some(redacted_encrypted_value.clone()),
            country_code: Some(REDACTED.to_string()),
            updated_by: provider.get_account().storage_scheme.to_string(),
            email: Some(redacted_encrypted_email),
            origin_zip: Some(redacted_encrypted_value.clone()),
        };

        match db
            .update_address_by_merchant_id_customer_id(
                self,
                provider.get_account().get_id(),
                update_address,
                provider.get_key_store(),
            )
            .await
        {
            Ok(_) => Ok(()),
            Err(error) => {
                if error.current_context().is_db_not_found() {
                    Ok(())
                } else {
                    Err(error)
                        .change_context(errors::CustomersErrorResponse::InternalServerError)
                        .attach_printable("failed update_address_by_merchant_id_customer_id")
                }
            }
        }?;

        let updated_customer = storage::CustomerUpdate::Update {
            name: Some(redacted_encrypted_value.clone()),
            email: Some(
                types::crypto_operation(
                    key_manager_state,
                    type_name!(storage::Customer),
                    CryptoOperation::Encrypt(REDACTED.to_string().into()),
                    identifier,
                    key,
                )
                .await
                .and_then(|val| val.try_into_operation())
                .switch()?,
            ),
            phone: Box::new(Some(redacted_encrypted_value.clone())),
            description: Some(Description::from_str_unchecked(REDACTED)),
            phone_country_code: Some(REDACTED.to_string()),
            metadata: Box::new(None),
            connector_customer: Box::new(None),
            address_id: None,
            tax_registration_id: Some(redacted_encrypted_value.clone()),
            document_details: Box::new(None),
            last_modified_by: initiator
                .and_then(|initiator| initiator.to_created_by())
                .map(|last_modified_by| last_modified_by.to_string()),
        };

        db.update_customer_by_customer_id_merchant_id(
            self.clone(),
            provider.get_account().get_id().to_owned(),
            customer_orig,
            updated_customer,
            provider.get_key_store(),
            provider.get_account().storage_scheme,
        )
        .await
        .switch()?;

        let response = customers::CustomerDeleteResponse {
            customer_id: self.clone(),
            customer_deleted: true,
            address_deleted: true,
            payment_methods_deleted: true,
        };
        metrics::CUSTOMER_REDACTED.add(1, &[]);
        Ok(services::ApplicationResponse::Json(response))
    }
}

#[instrument(skip(state))]
pub async fn update_customer(
    state: SessionState,
    provider: domain::Provider,
    initiator: Option<domain::Initiator>,
    update_customer: customers::CustomerUpdateRequestInternal,
) -> errors::CustomerResponse<customers::CustomerResponse> {
    update_customer
        .request
        .document_details
        .as_ref()
        .map(|doc| doc.validate())
        .transpose()
        .map_err(|err| errors::CustomersErrorResponse::InvalidRequestData {
            message: err.to_string(),
        })?;

    let db = state.store.as_ref();
    let key_manager_state = &(&state).into();
    //Add this in update call if customer can be updated anywhere else

    #[cfg(feature = "v1")]
    let verify_id_for_update_customer = VerifyIdForUpdateCustomer {
        merchant_reference_id: &update_customer.customer_id,
        merchant_account: provider.get_account(),
        key_store: provider.get_key_store(),
    };

    #[cfg(feature = "v2")]
    let verify_id_for_update_customer = VerifyIdForUpdateCustomer {
        id: &update_customer.id,
        merchant_account: provider.get_account(),
        key_store: provider.get_key_store(),
        key_manager_state,
    };

    let customer = verify_id_for_update_customer
        .verify_id_and_get_customer_object(db)
        .await?;

    let updated_customer = update_customer
        .request
        .create_domain_model_from_request(
            &None,
            db,
            &provider,
            initiator.as_ref(),
            key_manager_state,
            &state,
            &customer,
        )
        .await?;

    update_customer.request.generate_response(&updated_customer)
}

#[allow(clippy::too_many_arguments)]
#[async_trait::async_trait]
trait CustomerUpdateBridge {
    async fn create_domain_model_from_request<'a>(
        &'a self,
        connector_customer_details: &'a Option<
            Vec<payment_methods_domain::ConnectorCustomerDetails>,
        >,
        db: &'a dyn StorageInterface,
        provider: &'a domain::Provider,
        initiator: Option<&'a domain::Initiator>,
        key_manager_state: &'a KeyManagerState,
        state: &'a SessionState,
        domain_customer: &'a domain::Customer,
    ) -> errors::CustomResult<domain::Customer, errors::CustomersErrorResponse>;

    fn generate_response<'a>(
        &'a self,
        customer: &'a domain::Customer,
    ) -> errors::CustomerResponse<customers::CustomerResponse>;
}

#[cfg(feature = "v1")]
struct AddressStructForDbUpdate<'a> {
    update_customer: &'a customers::CustomerUpdateRequest,
    merchant_account: &'a domain::MerchantAccount,
    key_store: &'a domain::MerchantKeyStore,
    state: &'a SessionState,
    domain_customer: &'a domain::Customer,
}

#[cfg(feature = "v1")]
impl AddressStructForDbUpdate<'_> {
    async fn update_address_if_sent(
        &self,
        db: &dyn StorageInterface,
    ) -> errors::CustomResult<Option<domain::Address>, errors::CustomersErrorResponse> {
        let address = if let Some(addr) = &self.update_customer.address {
            match self.domain_customer.address_id.clone() {
                Some(address_id) => {
                    let customer_address: api_models::payments::AddressDetails = addr.clone();
                    let update_address = self
                        .update_customer
                        .get_address_update(
                            self.state,
                            customer_address,
                            self.key_store.key.get_inner().peek(),
                            self.merchant_account.storage_scheme,
                            self.merchant_account.get_id().clone(),
                        )
                        .await
                        .switch()
                        .attach_printable("Failed while encrypting Address while Update")?;
                    Some(
                        db.update_address(address_id, update_address, self.key_store)
                            .await
                            .switch()
                            .attach_printable(format!(
                            "Failed while updating address: merchant_id: {:?}, customer_id: {:?}",
                            self.merchant_account.get_id(),
                            self.domain_customer.get_id()
                        ))?,
                    )
                }
                None => {
                    let customer_address: api_models::payments::AddressDetails = addr.clone();

                    let address = self
                        .update_customer
                        .get_domain_address(
                            self.state,
                            customer_address,
                            self.merchant_account.get_id(),
                            self.domain_customer.get_id(),
                            self.key_store.key.get_inner().peek(),
                            self.merchant_account.storage_scheme,
                        )
                        .await
                        .switch()
                        .attach_printable("Failed while encrypting address")?;
                    Some(
                        db.insert_address_for_customers(address, self.key_store)
                            .await
                            .switch()
                            .attach_printable("Failed while inserting new address")?,
                    )
                }
            }
        } else {
            match &self.domain_customer.address_id {
                Some(address_id) => Some(
                    db.find_address_by_address_id(address_id, self.key_store)
                        .await
                        .switch()?,
                ),
                None => None,
            }
        };
        Ok(address)
    }
}

#[cfg(feature = "v1")]
#[derive(Debug)]
struct VerifyIdForUpdateCustomer<'a> {
    merchant_reference_id: &'a id_type::CustomerId,
    merchant_account: &'a domain::MerchantAccount,
    key_store: &'a domain::MerchantKeyStore,
}

#[cfg(feature = "v2")]
#[derive(Debug)]
struct VerifyIdForUpdateCustomer<'a> {
    id: &'a id_type::GlobalCustomerId,
    merchant_account: &'a domain::MerchantAccount,
    key_store: &'a domain::MerchantKeyStore,
    key_manager_state: &'a KeyManagerState,
}

#[cfg(feature = "v1")]
impl VerifyIdForUpdateCustomer<'_> {
    async fn verify_id_and_get_customer_object(
        &self,
        db: &dyn StorageInterface,
    ) -> Result<domain::Customer, error_stack::Report<errors::CustomersErrorResponse>> {
        let customer = db
            .find_customer_by_customer_id_merchant_id(
                self.merchant_reference_id,
                self.merchant_account.get_id(),
                self.key_store,
                self.merchant_account.storage_scheme,
            )
            .await
            .switch()?;

        Ok(customer)
    }
}

#[cfg(feature = "v2")]
impl VerifyIdForUpdateCustomer<'_> {
    async fn verify_id_and_get_customer_object(
        &self,
        db: &dyn StorageInterface,
    ) -> Result<domain::Customer, error_stack::Report<errors::CustomersErrorResponse>> {
        let customer = db
            .find_customer_by_global_id(
                self.id,
                self.key_store,
                self.merchant_account.storage_scheme,
            )
            .await
            .switch()?;

        Ok(customer)
    }
}

#[cfg(feature = "v1")]
#[async_trait::async_trait]
impl CustomerUpdateBridge for customers::CustomerUpdateRequest {
    async fn create_domain_model_from_request<'a>(
        &'a self,
        _connector_customer_details: &'a Option<
            Vec<payment_methods_domain::ConnectorCustomerDetails>,
        >,
        db: &'a dyn StorageInterface,
        provider: &'a domain::Provider,
        initiator: Option<&'a domain::Initiator>,
        key_manager_state: &'a KeyManagerState,
        state: &'a SessionState,
        domain_customer: &'a domain::Customer,
    ) -> errors::CustomResult<domain::Customer, errors::CustomersErrorResponse> {
        let update_address_for_update_customer = AddressStructForDbUpdate {
            update_customer: self,
            merchant_account: provider.get_account(),
            key_store: provider.get_key_store(),
            state,
            domain_customer,
        };

        let address = update_address_for_update_customer
            .update_address_if_sent(db)
            .await?;

        let key = provider.get_key_store().key.get_inner().peek();

        let encrypted_data = types::crypto_operation(
            key_manager_state,
            type_name!(domain::Customer),
            CryptoOperation::BatchEncrypt(domain::FromRequestEncryptableCustomer::to_encryptable(
                domain::FromRequestEncryptableCustomer {
                    name: self.name.clone(),
                    email: self
                        .email
                        .as_ref()
                        .map(|a| a.clone().expose().switch_strategy()),
                    phone: self.phone.clone(),
                    tax_registration_id: self.tax_registration_id.clone(),
                },
            )),
            Identifier::Merchant(provider.get_key_store().merchant_id.clone()),
            key,
        )
        .await
        .and_then(|val| val.try_into_batchoperation())
        .switch()?;

        let encryptable_customer =
            domain::FromRequestEncryptableCustomer::from_encryptable(encrypted_data)
                .change_context(errors::CustomersErrorResponse::InternalServerError)?;

        let document_details = hyperswitch_domain_models::type_encryption::crypto_operation(
            key_manager_state,
            type_name!(CustomerDocumentDetails),
            CryptoOperation::EncryptOptional(
                self.document_details
                    .as_ref()
                    .map(|details| {
                        details.to().map_err(|e| {
                            error_stack::Report::new(
                                errors::CustomersErrorResponse::InternalServerError,
                            )
                            .attach_printable(format!("Failed to encode details: {:?}", e))
                        })
                    })
                    .transpose()?,
            ),
            Identifier::Merchant(provider.get_account().get_id().clone()),
            key,
        )
        .await
        .map_err(|e| {
            error_stack::Report::new(errors::CustomersErrorResponse::InternalServerError)
                .attach_printable(e)
        })?
        .try_into_optionaloperation()
        .map_err(|e| {
            error_stack::Report::new(errors::CustomersErrorResponse::InternalServerError)
                .attach_printable(e)
        })?;

        let response = db
            .update_customer_by_customer_id_merchant_id(
                domain_customer.get_id().to_owned(),
                provider.get_account().get_id().to_owned(),
                domain_customer.to_owned(),
                storage::CustomerUpdate::Update {
                    name: encryptable_customer.name,
                    email: encryptable_customer.email.map(|email| {
                        let encryptable: Encryptable<Secret<String, pii::EmailStrategy>> =
                            Encryptable::new(
                                email.clone().into_inner().switch_strategy(),
                                email.into_encrypted(),
                            );
                        encryptable
                    }),
                    phone: Box::new(encryptable_customer.phone),
                    tax_registration_id: encryptable_customer.tax_registration_id,
                    document_details: Box::new(document_details),
                    phone_country_code: self.phone_country_code.clone(),
                    metadata: Box::new(self.metadata.clone()),
                    description: self.description.clone(),
                    connector_customer: Box::new(None),
                    address_id: address.clone().map(|addr| addr.address_id),
                    last_modified_by: initiator.and_then(|initiator| {
                        initiator
                            .to_created_by()
                            .map(|last_modified_by| last_modified_by.to_string())
                    }),
                },
                provider.get_key_store(),
                provider.get_account().storage_scheme,
            )
            .await
            .switch()?;

        Ok(response)
    }

    fn generate_response<'a>(
        &'a self,
        customer: &'a domain::Customer,
    ) -> errors::CustomerResponse<customers::CustomerResponse> {
        let address = self.get_address();
        Ok(services::ApplicationResponse::Json(
            customers::CustomerResponse::try_from((customer.clone(), address))
                .change_context(errors::CustomersErrorResponse::InternalServerError)
                .attach_printable("Failed to convert domain customer to CustomerResponse")?,
        ))
    }
}

#[cfg(feature = "v2")]
#[async_trait::async_trait]
impl CustomerUpdateBridge for customers::CustomerUpdateRequest {
    async fn create_domain_model_from_request<'a>(
        &'a self,
        connector_customer_details: &'a Option<
            Vec<payment_methods_domain::ConnectorCustomerDetails>,
        >,
        db: &'a dyn StorageInterface,
        provider: &'a domain::Provider,
        initiator: Option<&'a domain::Initiator>,
        key_manager_state: &'a KeyManagerState,
        state: &'a SessionState,
        domain_customer: &'a domain::Customer,
    ) -> errors::CustomResult<domain::Customer, errors::CustomersErrorResponse> {
        let default_billing_address = self.get_default_customer_billing_address();
        let encrypted_customer_billing_address = default_billing_address
            .async_map(|billing_address| {
                create_encrypted_data(
                    key_manager_state,
                    provider.get_key_store(),
                    billing_address,
                    common_utils::type_name!(diesel_models::payment_method::PaymentMethod),
                )
            })
            .await
            .transpose()
            .change_context(errors::CustomersErrorResponse::InternalServerError)
            .attach_printable("Unable to encrypt default customer billing address")?;

        let default_shipping_address = self.get_default_customer_shipping_address();
        let encrypted_customer_shipping_address = default_shipping_address
            .async_map(|shipping_address| {
                create_encrypted_data(
                    key_manager_state,
                    provider.get_key_store(),
                    shipping_address,
                    common_utils::type_name!(diesel_models::payment_method::PaymentMethod),
                )
            })
            .await
            .transpose()
            .change_context(errors::CustomersErrorResponse::InternalServerError)
            .attach_printable("Unable to encrypt default customer shipping address")?;

        let key = provider.get_key_store().key.get_inner().peek();

        let encrypted_data = types::crypto_operation(
            key_manager_state,
            type_name!(domain::Customer),
            types::CryptoOperation::BatchEncrypt(
                domain::FromRequestEncryptableCustomer::to_encryptable(
                    domain::FromRequestEncryptableCustomer {
                        name: self.name.clone(),
                        email: self
                            .email
                            .as_ref()
                            .map(|a| a.clone().expose().switch_strategy()),
                        phone: self.phone.clone(),
                        tax_registration_id: self.tax_registration_id.clone(),
                    },
                ),
            ),
            Identifier::Merchant(provider.get_key_store().merchant_id.clone()),
            key,
        )
        .await
        .and_then(|val| val.try_into_batchoperation())
        .switch()?;

        let encryptable_customer =
            domain::FromRequestEncryptableCustomer::from_encryptable(encrypted_data)
                .change_context(errors::CustomersErrorResponse::InternalServerError)?;

        let response = db
            .update_customer_by_global_id(
                &domain_customer.id,
                domain_customer.to_owned(),
                storage::CustomerUpdate::Update(Box::new(storage::CustomerGeneralUpdate {
                    name: encryptable_customer.name,
                    email: Box::new(encryptable_customer.email.map(|email| {
                        let encryptable: Encryptable<Secret<String, pii::EmailStrategy>> =
                            Encryptable::new(
                                email.clone().into_inner().switch_strategy(),
                                email.into_encrypted(),
                            );
                        encryptable
                    })),
                    phone: Box::new(encryptable_customer.phone),
                    tax_registration_id: encryptable_customer.tax_registration_id,
                    document_details: None,
                    phone_country_code: self.phone_country_code.clone(),
                    metadata: self.metadata.clone(),
                    description: self.description.clone(),
                    connector_customer: Box::new(None),
                    default_billing_address: encrypted_customer_billing_address,
                    default_shipping_address: encrypted_customer_shipping_address,
                    default_payment_method_id: Some(self.default_payment_method_id.clone()),
                    status: None,
                    last_modified_by: initiator.and_then(|initiator| {
                        initiator
                            .to_created_by()
                            .map(|last_modified_by| last_modified_by.to_string())
                    }),
                })),
                provider.get_key_store(),
                provider.get_account().storage_scheme,
            )
            .await
            .switch()?;
        Ok(response)
    }

    fn generate_response<'a>(
        &'a self,
        customer: &'a domain::Customer,
    ) -> errors::CustomerResponse<customers::CustomerResponse> {
        Ok(services::ApplicationResponse::Json(
            customers::CustomerResponse::try_from(customer.clone())
                .change_context(errors::CustomersErrorResponse::InternalServerError)
                .attach_printable("Failed to convert domain customer to CustomerResponse")?,
        ))
    }
}

pub async fn migrate_customers(
    state: SessionState,
    customers_migration: Vec<payment_methods_domain::PaymentMethodCustomerMigrate>,
    platform: domain::Platform,
) -> errors::CustomerResponse<()> {
    for customer_migration in customers_migration {
        #[cfg(feature = "v1")]
        let customer_id = customer_migration.customer.customer_id.clone();
        #[cfg(feature = "v1")]
        let connector_customer_details = customer_migration.connector_customer_details.clone();
        match create_customer(
            state.clone(),
            platform.get_provider().clone(),
            platform.get_initiator().cloned(),
            customer_migration.customer,
            customer_migration.connector_customer_details,
        )
        .await
        {
            Ok(_) => (),
            Err(e) => match e.current_context() {
                // Customer already exists in Hyperswitch: still merge the migrated
                // connector_customer_id(s) into the existing customer's connector_customer.
                errors::CustomersErrorResponse::CustomerAlreadyExists => {
                    #[cfg(feature = "v1")]
                    sync_connector_customer_for_migrated_customer(
                        &state,
                        &platform,
                        customer_id,
                        connector_customer_details,
                    )
                    .await?;
                }
                _ => return Err(e),
            },
        }
    }
    Ok(services::ApplicationResponse::Json(()))
}

/// Saves the connector's customer id (from the migration CSV) onto a customer that already
/// exists in Hyperswitch.
///
/// Background, in plain terms:
/// - Each customer row has a `connector_customer` field. Think of it as a small lookup table:
///   "for connector account X, this customer is known as Y on that connector's side".
/// - When the migration creates a brand-new customer, that field is filled in right away.
/// - But when the customer was already in Hyperswitch, the create step is skipped — and so the
///   connector customer id from the CSV would be thrown away. This function handles that case:
///   it opens the existing customer and adds the id in.
///
/// We *add to* the lookup table, we don't wipe it: ids saved for other connector accounts stay,
/// and if there was already an id for this connector account it gets replaced with the one from
/// the CSV. If the CSV row had no connector customer id (or nothing to attach it to), we do
/// nothing.
#[cfg(feature = "v1")]
async fn sync_connector_customer_for_migrated_customer(
    state: &SessionState,
    platform: &domain::Platform,
    customer_id: Option<id_type::CustomerId>,
    connector_customer_details: Option<Vec<payment_methods_domain::ConnectorCustomerDetails>>,
) -> errors::CustomResult<(), errors::CustomersErrorResponse> {
    // If the CSV row didn't give us a customer id, or gave no connector customer ids to save,
    // there's nothing to do.
    let Some((customer_id, connector_customer_details)) = customer_id
        .zip(connector_customer_details)
        .filter(|(_, details)| !details.is_empty())
    else {
        return Ok(());
    };

    let db: &dyn StorageInterface = state.store.as_ref();
    let provider = platform.get_provider();
    let merchant_id = provider.get_account().get_id();
    let storage_scheme = provider.get_account().storage_scheme;

    // Load the customer that already exists, so we can add to its current list of connector
    // customer ids instead of replacing it.
    let existing_customer = db
        .find_customer_by_customer_id_merchant_id(
            &customer_id,
            merchant_id,
            provider.get_key_store(),
            storage_scheme,
        )
        .await
        .switch()?;

    // Take the customer's current "connector account -> connector customer id" list (empty if
    // it had none yet) and add/overwrite the entries coming from the migration row.
    let mut connector_customer_map = existing_customer
        .connector_customer
        .as_ref()
        .and_then(|connector_customer| connector_customer.peek().as_object().cloned())
        .unwrap_or_default();
    for details in &connector_customer_details {
        connector_customer_map.insert(
            details.merchant_connector_id.get_string_repr().to_string(),
            details.connector_customer_id.clone().into(),
        );
    }

    // Save the updated list back onto the customer. This only changes the `connector_customer`
    // field (and the "last modified" bookkeeping) — the rest of the customer is untouched.
    db.update_customer_by_customer_id_merchant_id(
        customer_id,
        merchant_id.clone(),
        existing_customer,
        storage::CustomerUpdate::ConnectorCustomer {
            connector_customer: Some(pii::SecretSerdeValue::new(serde_json::Value::Object(
                connector_customer_map,
            ))),
            last_modified_by: platform
                .get_initiator()
                .and_then(|initiator| initiator.to_created_by())
                .map(|last_modified_by| last_modified_by.to_string()),
        },
        provider.get_key_store(),
        storage_scheme,
    )
    .await
    .switch()?;

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_is_customer_id_in_global_format() {
        let test_cases = [
            ("0asbjabjbd", false),
            ("0a_cus_12345678123456781234567812345678", true),
            ("abc12_cus_12345678123456781234567812345678", true),
            ("1b_cus_12iufbeksjeb", false),
            ("efbc2_cus_217846821", false),
            ("0a_pm_12345678123456781234567812345678", false),
        ];

        for (customer_id_str, expected) in test_cases {
            let customer_id =
                id_type::CustomerId::wrap(customer_id_str.to_string()).expect("valid customer id");

            assert_eq!(
                is_customer_id_in_global_format(&customer_id),
                expected,
                "failed for customer_id={customer_id_str}",
            );
        }
    }
}
