use std::marker::PhantomData;

use api_models::{
    enums::FrmSuggestion, mandates::RecurringDetails, payment_methods::PaymentMethodsData,
    payments::GetAddressFromPaymentMethodData,
};
use async_trait::async_trait;
use common_types::payments as common_payments_types;
use common_utils::{
    ext_traits::{AsyncExt, Encode, ValueExt},
    type_name,
    types::{
        keymanager::{Identifier, KeyManagerState, ToEncryptable},
        MinorUnit,
    },
};
use diesel_models::payment_attempt::ConnectorMandateReferenceId as DieselConnectorMandateReferenceId;
use error_stack::{self, ResultExt};
use hyperswitch_domain_models::{
    mandates,
    mandates::{MandateDetails, MandateTransactionType},
    payment_method_data::RecurringDetails as domain_recurring_details,
    payment_methods::PaymentMethodWithRawData,
    payments::{payment_attempt::PaymentAttempt, FromRequestEncryptablePaymentIntent},
};
use hyperswitch_masking::{ExposeInterface, PeekInterface, Secret};
use router_derive::PaymentOperation;
use router_env::{instrument, logger, tracing};
use storage_impl::platform_wrapper;
use time::PrimitiveDateTime;

use super::{BoxedOperation, Domain, GetTracker, Operation, UpdateTracker, ValidateRequest};
use crate::{
    consts,
    core::{
        configs::dimension_state,
        errors::{self, CustomResult, RouterResult, StorageErrorExt},
        mandate::helpers as m_helpers,
        payment_link,
        payment_methods::transformers as pm_transformers,
        payments::{
            self, client_session::ClientSessionManager, helpers, operations, CustomerDetails,
            OperationSessionGetters, OperationSessionSetters, PaymentAddress, PaymentData,
        },
        utils as core_utils,
    },
    db::StorageInterface,
    events::audit_events::{AuditEvent, AuditEventType},
    routes::{app::ReqState, SessionState},
    services,
    types::{
        self,
        api::{self, ConnectorCallType, PaymentIdTypeExt},
        domain,
        storage::{
            self,
            enums::{self, IntentStatus},
        },
        transformers::ForeignFrom,
    },
    utils::{self, OptionExt},
};

#[derive(Debug, Clone, Copy, PaymentOperation)]
#[operation(operations = "all", flow = "authorize")]
pub struct PaymentCreate;

type PaymentCreateOperation<'a, F> = BoxedOperation<'a, F, api::PaymentsRequest, PaymentData<F>>;

/// The `get_trackers` function for `PaymentsCreate` is an entrypoint for new payments
/// This will create all the entities required for a new payment from the request
#[async_trait]
impl<F: Send + Clone + Sync> GetTracker<F, PaymentData<F>, api::PaymentsRequest> for PaymentCreate {
    #[instrument(skip_all)]
    async fn get_trackers<'a>(
        &'a self,
        state: &'a SessionState,
        payment_id: &api::PaymentIdType,
        request: &api::PaymentsRequest,
        platform: &domain::Platform,
        _auth_flow: services::AuthFlow,
        flow_kind: operations::PaymentFlowKind,
        header_payload: &hyperswitch_domain_models::payments::HeaderPayload,
        payment_method_fetch_data: operations::PaymentMethodFetchData,
        dimensions: &dimension_state::DimensionsWithProcessorAndProviderMerchantId,
        _payment_pre_fetched_info: Option<operations::PaymentPreFetchedInformation>,
    ) -> RouterResult<operations::GetTrackerResponse<'a, F, api::PaymentsRequest, PaymentData<F>>>
    {
        let operations::PaymentMethodFetchData {
            payment_method_info: prefetched_payment_method_info,
            payment_method_with_raw_data,
            ..
        } = payment_method_fetch_data;
        let db = &*state.store;
        let money @ (amount, currency) = payments_create_request_validation(request)?;

        let payment_id = payment_id
            .get_payment_intent_id()
            .change_context(errors::ApiErrorResponse::PaymentNotFound)?;

        #[cfg(feature = "v1")]
        helpers::validate_business_details(
            request.business_country,
            request.business_label.as_ref(),
            platform.get_processor(),
        )?;

        // If profile id is not passed, get it from the business_country and business_label
        #[cfg(feature = "v1")]
        let profile_id = core_utils::get_profile_id_from_business_details(
            request.business_country,
            request.business_label.as_ref(),
            platform.get_processor(),
            request.profile_id.as_ref(),
            &*state.store,
            true,
        )
        .await?;

        // Profile id will be mandatory in v2 in the request / headers
        #[cfg(feature = "v2")]
        let profile_id = request
            .profile_id
            .clone()
            .get_required_value("profile_id")
            .attach_printable("Profile id is a mandatory parameter")?;

        // TODO: eliminate a redundant db call to fetch the business profile
        // Validate whether profile_id passed in request is valid and is linked to the merchant
        let business_profile = if let Some(business_profile) =
            core_utils::validate_and_get_business_profile(
                db,
                platform.get_processor(),
                Some(&profile_id),
            )
            .await?
        {
            business_profile
        } else {
            platform_wrapper::business_profile::find_business_profile_by_profile_id(
                state.store.as_ref(),
                platform.get_processor(),
                &profile_id,
            )
            .await
            .to_not_found_response(errors::ApiErrorResponse::ProfileNotFound {
                id: profile_id.get_string_repr().to_owned(),
            })?
        };
        let customer_acceptance = request.customer_acceptance.clone();

        let recurring_details = request.recurring_details.clone();

        let mandate_type = m_helpers::get_mandate_type(
            request.mandate_data.clone(),
            request.off_session,
            request.setup_future_usage,
            request.customer_acceptance.clone(),
            request.payment_token.clone(),
            request.payment_method,
        )
        .change_context(errors::ApiErrorResponse::MandateValidationFailed {
            reason: "Expected one out of recurring_details and mandate_data but got both".into(),
        })?;

        let m_helpers::MandateGenericData {
            token,
            payment_method,
            payment_method_type,
            mandate_data,
            recurring_mandate_payment_data,
            mandate_connector,
            payment_method_info,
        } = helpers::get_token_pm_type_mandate_details(
            state,
            request,
            mandate_type,
            platform,
            None,
            None,
            prefetched_payment_method_info,
        )
        .await?;

        helpers::validate_allowed_payment_method_types_request(
            state,
            &profile_id,
            platform.get_processor(),
            request.allowed_payment_method_types.clone(),
        )
        .await?;

        let customer_details = helpers::get_customer_details_from_request_or_pm_table(
            request,
            None,
            mandate_type.as_ref(),
        )?;

        let shipping_address = helpers::create_or_find_address_for_payment_by_request(
            state,
            request.shipping.as_ref(),
            None,
            platform.get_processor().get_account().get_id(),
            customer_details.customer_id.as_ref(),
            platform.get_processor().get_key_store(),
            &payment_id,
            platform.get_processor().get_account().storage_scheme,
        )
        .await?;

        let billing_address = helpers::create_or_find_address_for_payment_by_request(
            state,
            request.billing.as_ref(),
            None,
            platform.get_processor().get_account().get_id(),
            customer_details.customer_id.as_ref(),
            platform.get_processor().get_key_store(),
            &payment_id,
            platform.get_processor().get_account().storage_scheme,
        )
        .await?;

        let payment_method_data_billing = request
            .payment_method_data
            .as_ref()
            .and_then(|pmd| pmd.billing.clone())
            .or(payment_method_with_raw_data.as_ref().and_then(|pm| {
                pm.payment_method
                    .payment_method_billing_address
                    .clone()
                    .map(|decrypted_data| decrypted_data.into_inner().expose())
                    .and_then(|decrypted_value| {
                        decrypted_value
                            .parse_value("payment method billing address")
                            .ok()
                    })
            }));

        let payment_method_billing_address =
            helpers::create_or_find_address_for_payment_by_request(
                state,
                payment_method_data_billing.as_ref(),
                None,
                platform.get_processor().get_account().get_id(),
                customer_details.customer_id.as_ref(),
                platform.get_processor().get_key_store(),
                &payment_id,
                platform.get_processor().get_account().storage_scheme,
            )
            .await?;

        let browser_info = request
            .browser_info
            .clone()
            .as_ref()
            .map(Encode::encode_to_value)
            .transpose()
            .change_context(errors::ApiErrorResponse::InvalidDataValue {
                field_name: "browser_info",
            })?;

        let attempt_id = if core_utils::is_merchant_enabled_for_payment_id_as_connector_request_id(
            &state.conf,
            platform.get_processor(),
        ) {
            payment_id.get_string_repr().to_string()
        } else {
            payment_id.get_attempt_id(1)
        };

        let session_expiry =
            common_utils::date_time::now().saturating_add(time::Duration::seconds(
                request.session_expiry.map(i64::from).unwrap_or(
                    business_profile
                        .session_expiry
                        .unwrap_or(consts::DEFAULT_SESSION_EXPIRY),
                ),
            ));

        let payment_link_data = match request.payment_link {
            Some(true) => {
                let default_domain_name = state.base_url.clone();

                let (payment_link_config, domain_name) =
                    payment_link::get_payment_link_config_based_on_priority(
                        request.payment_link_config.clone(),
                        business_profile.payment_link_config.clone(),
                        platform.get_processor(),
                        default_domain_name,
                        request.payment_link_config_id.clone(),
                    )?;

                create_payment_link(
                    request,
                    payment_link_config,
                    platform,
                    payment_id.clone(),
                    db,
                    amount,
                    request.description.clone(),
                    profile_id.clone(),
                    domain_name,
                    session_expiry,
                    header_payload.locale.clone(),
                )
                .await?
            }
            _ => None,
        };

        let payment_intent_new = Self::make_payment_intent(
            state,
            &payment_id,
            platform,
            money,
            request,
            shipping_address
                .as_ref()
                .map(|address| address.address_id.clone()),
            payment_link_data.clone(),
            billing_address
                .as_ref()
                .map(|address| address.address_id.clone()),
            attempt_id,
            profile_id.clone(),
            session_expiry,
            &business_profile,
            request.is_payment_id_from_merchant,
            payment_method_info.clone(),
        )
        .await?;

        let (mandate_reference_id_from_recurring_details, payment_method_recurring_details) =
            match recurring_details
                .clone()
                .map(domain_recurring_details::from)
                .and_then(|details| {
                    details.get_mandate_reference_id_and_payment_method_data_for_proxy_flow()
                }) {
                Some((mandate_reference_id, payment_method_recurring_details)) => (
                    Some(mandate_reference_id),
                    Some(payment_method_recurring_details),
                ),
                None => (None, None),
            };

        let (payment_attempt_new, additional_payment_data) = Self::make_payment_attempt(
            &payment_id,
            platform,
            money,
            payment_method,
            payment_method_type,
            request,
            browser_info,
            state,
            payment_method_billing_address
                .as_ref()
                .map(|address| address.address_id.clone()),
            &payment_method_info,
            profile_id.clone(),
            &customer_acceptance,
            payment_method_recurring_details.clone(),
            customer_details.customer_id.as_ref(),
            dimensions,
        )
        .await?;

        let payment_intent = platform_wrapper::payment_intent::insert_payment_intent(
            state.store.as_ref(),
            platform.get_processor(),
            payment_intent_new,
        )
        .await
        .to_duplicate_response(errors::ApiErrorResponse::DuplicatePayment {
            payment_id: payment_id.clone(),
        })?;

        if let Some(order_details) = &request.order_details {
            helpers::validate_order_details_amount(
                order_details.to_owned(),
                payment_intent.amount,
                false,
            )?;
        }

        #[cfg(feature = "v1")]
        let mut payment_attempt = platform_wrapper::payment_attempt::insert_payment_attempt(
            state.store.as_ref(),
            platform.get_processor(),
            payment_attempt_new,
        )
        .await
        .to_duplicate_response(errors::ApiErrorResponse::DuplicatePayment {
            payment_id: payment_id.clone(),
        })?;

        #[cfg(feature = "v2")]
        let payment_attempt = platform_wrapper::payment_attempt::insert_payment_attempt(
            state.store.as_ref(),
            platform.get_processor(),
            payment_attempt_new,
        )
        .await
        .to_duplicate_response(errors::ApiErrorResponse::DuplicatePayment {
            payment_id: payment_id.clone(),
        })?;

        let mandate_details_present = payment_attempt.mandate_details.is_some();

        helpers::validate_mandate_data_and_future_usage(
            request.setup_future_usage,
            mandate_details_present,
        )?;
        // connector mandate reference update history
        let mandate_id = request
            .mandate_id
            .as_ref()
            .or_else(|| {
                request
                    .recurring_details
                    .as_ref()
                    .and_then(|recurring_details| match recurring_details {
                        RecurringDetails::MandateId(id) => Some(id),
                        _ => None,
                    })
            })
            .async_and_then(|mandate_id| async {
                let mandate = db
                    .find_mandate_by_merchant_id_mandate_id(
                        platform.get_processor().get_account().get_id(),
                        mandate_id,
                        platform.get_processor().get_account().storage_scheme,
                    )
                    .await
                    .to_not_found_response(errors::ApiErrorResponse::MandateNotFound);
                Some(mandate.and_then(|mandate_obj| {
                    match (
                        mandate_obj.network_transaction_id,
                        mandate_obj.connector_mandate_ids,
                    ) {
                        (_, Some(connector_mandate_id)) => connector_mandate_id
                            .parse_value("ConnectorMandateId")
                            .change_context(errors::ApiErrorResponse::MandateNotFound)
                            .map(|connector_id: mandates::ConnectorMandateReferenceId| {
                                mandates::MandateIds {
                                    mandate_id: Some(mandate_obj.mandate_id),
                                    mandate_reference_id: Some(
                                        mandates::MandateReferenceId::ConnectorMandateId(
                                            mandates::ConnectorMandateReferenceId::new(
                                                connector_id.get_connector_mandate_id(),
                                                connector_id.get_payment_method_id(),
                                                None,
                                                connector_id.get_mandate_metadata(),
                                                connector_id
                                                    .get_connector_mandate_request_reference_id(),
                                                None,
                                            ),
                                        ),
                                    ),
                                }
                            }),
                        (Some(network_tx_id), _) => Ok(mandates::MandateIds {
                            mandate_id: Some(mandate_obj.mandate_id),
                            mandate_reference_id: Some(
                                mandates::MandateReferenceId::NetworkMandateId(
                                    mandates::NetworkMandateIdRef {
                                        network_transaction_id: network_tx_id,
                                        transaction_link_id: mandate_obj
                                            .network_transaction_link_id,
                                    },
                                ),
                            ),
                        }),
                        (_, _) => Ok(mandates::MandateIds {
                            mandate_id: Some(mandate_obj.mandate_id),
                            mandate_reference_id: None,
                        }),
                    }
                }))
            })
            .await
            .transpose()?;

        let mandate_id = if mandate_id.is_none() {
            request
                .recurring_details
                .as_ref()
                .and_then(|recurring_details| match recurring_details {
                    RecurringDetails::ProcessorPaymentToken(token) => Some(mandates::MandateIds {
                        mandate_id: None,
                        mandate_reference_id: Some(
                            mandates::MandateReferenceId::ConnectorMandateId(
                                mandates::ConnectorMandateReferenceId::new(
                                    Some(token.processor_payment_token.clone()),
                                    None,
                                    None,
                                    None,
                                    None,
                                    None,
                                ),
                            ),
                        ),
                    }),
                    RecurringDetails::CardWithLimitedData(_)
                    | RecurringDetails::NetworkTransactionIdAndNetworkTokenDetails(_)
                    | RecurringDetails::NetworkTransactionIdAndDecryptedWalletTokenDetails(_)
                    | RecurringDetails::NetworkTransactionIdAndCardDetails(_) => {
                        Some(mandates::MandateIds {
                            mandate_id: None,
                            mandate_reference_id: mandate_reference_id_from_recurring_details,
                        })
                    }
                    RecurringDetails::PaymentMethodId(_) | RecurringDetails::MandateId(_) => None,
                })
        } else {
            mandate_id
        };

        let operation = payments::if_not_create_change_operation::<_, F>(
            payment_intent.status,
            request.confirm,
            flow_kind,
            self,
        );

        let creds_identifier = request
            .merchant_connector_details
            .as_ref()
            .map(|mcd| mcd.creds_identifier.to_owned());
        request
            .merchant_connector_details
            .to_owned()
            .async_map(|mcd| async {
                helpers::insert_merchant_connector_creds_to_config(
                    db,
                    platform.get_processor(),
                    mcd,
                )
                .await
            })
            .await
            .transpose()?;

        // The operation merges mandate data from both request and payment_attempt
        let setup_mandate = mandate_data;

        let surcharge_details = request.surcharge_details.map(|request_surcharge_details| {
            payments::types::SurchargeDetails::from((&request_surcharge_details, &payment_attempt))
        });

        let payment_method_data_from_request =
            request
                .payment_method_data
                .as_ref()
                .and_then(|request_payment_method_data| {
                    request_payment_method_data.payment_method_data.clone()
                });

        let payment_method_data = payment_method_with_raw_data
            .clone()
            .and_then(|pm| pm.raw_payment_method_data)
            .or(payment_method_data_from_request.map(Into::into))
            .or(payment_method_recurring_details.clone())
            .map(|payment_method_data| {
                if let Some(additional_pm_data) = additional_payment_data {
                    payment_method_data.apply_additional_payment_data(additional_pm_data)
                } else {
                    payment_method_data
                }
            });

        let additional_pm_data_from_locker = if let Some(ref pm) = payment_method_info {
            let card_detail_from_locker: Option<api::CardDetailFromLocker> = pm
                .payment_method_data
                .clone()
                .map(|x| x.into_inner().expose())
                .and_then(|v| {
                    v.parse_value("PaymentMethodsData")
                        .map_err(|err| {
                            router_env::logger::info!(
                                "PaymentMethodsData deserialization failed: {:?}",
                                err
                            )
                        })
                        .ok()
                })
                .and_then(|pmd| match pmd {
                    PaymentMethodsData::Card(crd) => Some(api::CardDetailFromLocker::from(crd)),
                    _ => None,
                });

            card_detail_from_locker.map(|card_details| {
                let additional_data = card_details.into();
                api_models::payments::AdditionalPaymentData::Card(Box::new(additional_data))
            })
        } else {
            None
        };
        // Only set `payment_attempt.payment_method_data` if `additional_pm_data_from_locker` is not None
        if let Some(additional_pm_data) = additional_pm_data_from_locker.as_ref() {
            payment_attempt.payment_method_data = Some(
                Encode::encode_to_value(additional_pm_data)
                    .change_context(errors::ApiErrorResponse::InternalServerError)
                    .attach_printable("Failed to encode additional pm data")?,
            );
        }
        let amount = payment_attempt.get_total_amount().into();

        payment_attempt.connector_mandate_detail =
            Some(DieselConnectorMandateReferenceId::foreign_from(
                mandates::ConnectorMandateReferenceId::new(
                    None,
                    None,
                    None, // update_history
                    None, // mandate_metadata
                    Some(common_utils::generate_id_with_len(
                        consts::CONNECTOR_MANDATE_REQUEST_REFERENCE_ID_LENGTH,
                    )), // connector_mandate_request_reference_id
                    None,
                ),
            ));

        let address = PaymentAddress::new(
            shipping_address.as_ref().map(From::from),
            billing_address.as_ref().map(From::from),
            payment_method_billing_address.as_ref().map(From::from),
            business_profile.use_billing_as_payment_method_billing,
        );

        let payment_method_data_billing = request
            .payment_method_data
            .as_ref()
            .and_then(|pmd| pmd.payment_method_data.as_ref())
            .and_then(|payment_method_data_billing| {
                payment_method_data_billing.get_billing_address()
            })
            .map(From::from);
        let pm_pmd_billing = payment_method_with_raw_data.as_ref().and_then(|pm| {
            pm.payment_method
                .payment_method_billing_address
                .clone()
                .map(|decrypted_data| decrypted_data.into_inner().expose())
                .and_then(|decrypted_value| {
                    decrypted_value
                        .parse_value::<hyperswitch_domain_models::address::Address>(
                            "payment method billing address",
                        )
                        .ok()
                })
        });

        let add = payment_method_data_billing.or(pm_pmd_billing);

        let unified_address = address.unify_with_payment_method_data_billing(add);

        // Check if client session validation is enabled
        let dimensions = dimension_state::Dimensions::new()
            .with_provider_merchant_id(platform.get_provider().get_provider_merchant_id())
            .with_processor_merchant_id(platform.get_processor().get_processor_merchant_id());

        let session_validation_enabled = dimensions
            .get_client_session_validation_enabled(
                state.store.as_ref(),
                state.superposition_service.as_ref(),
                None,
            )
            .await;

        // Generate client session for the payment only if validation is enabled
        let client_session_id = if session_validation_enabled {
            Some(
                ClientSessionManager::create_session(
                    state,
                    platform.get_processor().get_account().get_id(),
                    &payment_id,
                    Some(session_expiry),
                )
                .await
                .change_context(errors::ApiErrorResponse::InternalServerError)
                .attach_printable("Failed to create client session")?,
            )
        } else {
            None
        };

        // For a single-call create+confirm routed through the external vault proxy core, the
        // operation has been swapped to `PaymentExternalVaultProxyConfirm` above. That confirm
        // op normally builds `external_vault_pmd` in its own `get_trackers`, which is skipped
        // here — so build it now from the same request + vault tokens, mirroring the two-step
        // confirm flow. The downstream connector call requires this to be populated.
        let external_vault_pmd = (flow_kind == operations::PaymentFlowKind::ExternalVaultProxy
            && request.confirm.unwrap_or(false))
        .then(|| {
            super::payment_confirm_external_vault_proxy::build_external_vault_payment_method_data(
                request,
                payment_method_with_raw_data.as_ref(),
            )
        })
        .transpose()?
        .flatten();

        let payment_data = PaymentData {
            flow: PhantomData,
            payment_intent,
            payment_attempt,
            currency,
            amount,
            mandate_id: mandate_id.clone(),
            mandate_connector,
            setup_mandate,
            customer_acceptance,
            token,
            address: unified_address,
            token_data: None,
            confirm: request.confirm,
            payment_method_data,
            payment_method_token: None,
            payment_method_info,
            refunds: vec![],
            disputes: vec![],
            attempts: None,
            force_sync: None,
            all_keys_required: None,
            sessions_token: vec![],
            card_cvc: request.card_cvc.clone(),
            creds_identifier,
            pm_token: None,
            connector_customer_id: None,
            recurring_mandate_payment_data,
            multiple_capture_data: None,
            redirect_response: None,
            surcharge_details,
            frm_message: None,
            payment_link_data,
            incremental_authorization_details: None,
            authorizations: vec![],
            authentication: None,
            recurring_details,
            poll_config: None,
            tax_data: None,
            session_id: None,
            service_details: None,
            card_testing_guard_data: None,
            vault_operation: None,
            threeds_method_comp_ind: None,
            whole_connector_response: None,
            is_manual_retry_enabled: None,
            is_l2_l3_enabled: business_profile.is_l2_l3_enabled,
            external_authentication_data: request.three_ds_data.clone(),
            client_session_id,
            vault_session_details: None,
            external_vault_pmd,
            update_request_fields: None,
        };

        let get_trackers_response = operations::GetTrackerResponse {
            operation,
            customer_details: Some(customer_details),
            payment_data,
            business_profile,
            mandate_type,
        };

        Ok(get_trackers_response)
    }
}

#[async_trait]
impl<F: Clone + Send + Sync> Domain<F, api::PaymentsRequest, PaymentData<F>> for PaymentCreate {
    #[instrument(skip_all)]
    async fn get_or_create_customer_details<'a>(
        &'a self,
        state: &SessionState,
        payment_data: &mut PaymentData<F>,
        request: Option<CustomerDetails>,
        provider: &domain::Provider,
        initiator: Option<&domain::Initiator>,
        dimensions: &dimension_state::DimensionsWithProcessorAndProviderMerchantIdAndProfileId,
        mandate_type: Option<MandateTransactionType>,
    ) -> CustomResult<(PaymentCreateOperation<'a, F>, Option<domain::Customer>), errors::StorageError>
    {
        let (operation, customer) = match provider.get_account().merchant_account_type {
            common_enums::MerchantAccountType::Standard => {
                helpers::create_customer_if_not_exist(
                    state,
                    Box::new(self),
                    payment_data,
                    request.clone(),
                    provider,
                    initiator,
                    dimensions,
                )
                .await
            }
            common_enums::MerchantAccountType::Platform => {
                let customer = helpers::get_customer_if_exists(
                    state,
                    request.as_ref().and_then(|r| r.customer_id.as_ref()),
                    payment_data.payment_intent.customer_id.as_ref(),
                    provider,
                )
                .await?
                .inspect(|cust| {
                    payment_data.payment_intent.customer_id = Some(cust.get_id().clone());
                });
                let op: PaymentCreateOperation<'a, F> = Box::new(self);
                Ok((op, customer))
            }
            common_enums::MerchantAccountType::Connected => {
                Err(errors::StorageError::ValueNotFound(
                    "Connected merchant cannot be a provider".to_string(),
                )
                .into())
            }
        }?;
        // When no document details is passed in request but customer_id is passed, use that customers document details from customers table
        if let Some(doc_details) = request
            .filter(|req| req.customer_id.is_some() && req.document_details.is_none())
            .and_then(|_| customer.as_ref().and_then(|c| c.document_details.clone()))
        {
            let encrypted_customer_details = core_utils::update_intent_customer_documents(
                payment_data.get_payment_intent(),
                Some(doc_details),
                mandate_type,
                state,
                provider.get_key_store(),
                payment_data
                    .payment_method_info
                    .as_ref()
                    .and_then(|pm| pm.customer_details.clone()),
            )
            .await
            .change_context(errors::StorageError::EncryptionError)?;
            payment_data.set_document_details_in_intent(encrypted_customer_details);
        }
        Ok((operation, customer))
    }

    async fn payments_dynamic_tax_calculation<'a>(
        &'a self,
        state: &SessionState,
        payment_data: &mut PaymentData<F>,
        _connector_call_type: &ConnectorCallType,
        business_profile: &domain::Profile,
        processor: &domain::Processor,
    ) -> CustomResult<(), errors::ApiErrorResponse> {
        let is_tax_connector_enabled = business_profile.get_is_tax_connector_enabled();
        let skip_external_tax_calculation = payment_data
            .payment_intent
            .skip_external_tax_calculation
            .unwrap_or(false);
        if is_tax_connector_enabled && !skip_external_tax_calculation {
            let db = state.store.as_ref();

            let merchant_connector_id = business_profile
                .tax_connector_id
                .as_ref()
                .get_required_value("business_profile.tax_connector_id")?;

            #[cfg(feature = "v1")]
            let mca = db
                .find_by_merchant_connector_account_merchant_id_merchant_connector_id(
                    &business_profile.merchant_id,
                    merchant_connector_id,
                    processor.get_key_store(),
                )
                .await
                .to_not_found_response(
                    errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
                        id: merchant_connector_id.get_string_repr().to_string(),
                    },
                )?;

            #[cfg(feature = "v2")]
            let mca = db
                .find_merchant_connector_account_by_id(merchant_connector_id, key_store)
                .await
                .to_not_found_response(
                    errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
                        id: merchant_connector_id.get_string_repr().to_string(),
                    },
                )?;

            let connector_data =
                api::TaxCalculateConnectorData::get_connector_by_name(&mca.connector_name)?;

            let router_data = core_utils::construct_payments_dynamic_tax_calculation_router_data(
                state,
                processor,
                payment_data,
                &mca,
            )
            .await?;
            let connector_integration: services::BoxedPaymentConnectorIntegrationInterface<
                api::CalculateTax,
                types::PaymentsTaxCalculationData,
                types::TaxCalculationResponseData,
            > = connector_data.connector.get_connector_integration();

            let response = services::execute_connector_processing_step(
                state,
                connector_integration,
                &router_data,
                payments::CallConnectorAction::Trigger,
                None,
                None,
            )
            .await
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Tax connector Response Failed")?;

            let tax_response = response.response.map_err(|err| {
                errors::ApiErrorResponse::ExternalConnectorError {
                    code: err.code,
                    message: err.message,
                    connector: connector_data.connector_name.clone().to_string(),
                    status_code: err.status_code,
                    reason: err.reason,
                }
            })?;

            payment_data.payment_intent.tax_details = Some(diesel_models::TaxDetails {
                default: Some(diesel_models::DefaultTax {
                    order_tax_amount: tax_response.order_tax_amount,
                }),
                payment_method_type: None,
            });

            Ok(())
        } else {
            Ok(())
        }
    }

    #[instrument(skip_all)]
    async fn fetch_payment_method(
        &self,
        state: &SessionState,
        req: &api::PaymentsRequest,
        platform: &domain::Platform,
        feature_config: &core_utils::FeatureConfig,
    ) -> RouterResult<operations::PaymentMethodFetchData> {
        let payment_method_fetch_data = if feature_config.is_payment_method_modular_allowed {
            logger::debug!("PM Modular Service branch selected for payment method fetch");
            utils::when(
                req.off_session == Some(true) && req.recurring_details.is_none(),
                || {
                    Err(error_stack::report!(
                        errors::ApiErrorResponse::PreconditionFailed {
                            message: "off_session requires recurring_details".into(),
                        }
                    ))
                },
            )?;

            if let Some(payment_method_ref) = self.get_payment_method_reference(req) {
                let pm_info = self
                    .fetch_payment_method_from_modular_service(
                        state,
                        req,
                        platform,
                        payment_method_ref,
                    )
                    .await?;

                operations::PaymentMethodFetchData::from_modular(pm_info)
            } else {
                logger::debug!("No payment method reference found, skipping payment method fetch");
                operations::PaymentMethodFetchData::default()
            }
        } else {
            let payment_method_info = if let Some(payment_method_id) =
                self.get_recurring_payment_method_id(req)
            {
                logger::debug!("Non-modular payment method DB lookup branch selected");
                Some(
                    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)?,
                )
            } else {
                logger::debug!(
                    "No recurring payment method id found, skipping payment method DB lookup"
                );
                None
            };

            if let Some(payment_method) = payment_method_info {
                if feature_config.should_use_modular_pm_path(
                    Some(payment_method.version),
                    payment_method.compatibility_updated_at,
                    Some(payment_method.last_modified),
                ) {
                    logger::debug!(
                        "Payment method version is V2, fetching payment method from PM Modular Service"
                    );
                    let payment_method_id = payment_method.get_id().to_owned();
                    let pm_info = self
                        .fetch_payment_method_from_modular_service(
                            state,
                            req,
                            platform,
                            &payment_method_id,
                        )
                        .await?;

                    operations::PaymentMethodFetchData::from_modular(pm_info)
                } else {
                    logger::debug!("Using legacy DB payment method branch");
                    operations::PaymentMethodFetchData::from_legacy(payment_method, None)
                }
            } else {
                operations::PaymentMethodFetchData::default()
            }
        };

        Ok(payment_method_fetch_data)
    }

    #[instrument(skip_all)]
    async fn make_pm_data<'a>(
        &'a self,
        state: &'a SessionState,
        payment_data: &mut PaymentData<F>,
        storage_scheme: enums::MerchantStorageScheme,
        platform: &domain::Platform,
        business_profile: &domain::Profile,
        should_retry_with_pan: bool,
    ) -> RouterResult<(
        PaymentCreateOperation<'a, F>,
        Option<domain::PaymentMethodData>,
        Option<String>,
    )> {
        Box::pin(helpers::make_pm_data(
            Box::new(self),
            state,
            payment_data,
            platform,
            storage_scheme,
            business_profile,
            should_retry_with_pan,
        ))
        .await
    }

    #[instrument(skip_all)]
    async fn add_task_to_process_tracker<'a>(
        &'a self,
        _state: &'a SessionState,
        _payment_attempt: &PaymentAttempt,
        _requeue: bool,
        _schedule_time: Option<PrimitiveDateTime>,
    ) -> CustomResult<(), errors::ApiErrorResponse> {
        Ok(())
    }

    async fn get_connector<'a>(
        &'a self,
        _processor: &domain::Processor,
        state: &SessionState,
        request: &api::PaymentsRequest,
        _payment_intent: &storage::PaymentIntent,
    ) -> CustomResult<api::ConnectorChoice, errors::ApiErrorResponse> {
        helpers::get_connector_default(state, request.routing.clone()).await
    }

    #[instrument(skip_all)]
    async fn guard_payment_against_blocklist<'a>(
        &'a self,
        _state: &SessionState,
        _processor: &domain::Processor,
        _dimensions: &dimension_state::DimensionsWithProcessorAndProviderMerchantId,
        _payment_data: &mut PaymentData<F>,
        _business_profile: &domain::Profile,
    ) -> CustomResult<bool, errors::ApiErrorResponse> {
        Ok(false)
    }
}

#[async_trait]
impl<F: Clone + Sync> UpdateTracker<F, PaymentData<F>, api::PaymentsRequest> for PaymentCreate {
    #[instrument(skip_all)]
    async fn update_trackers<'b>(
        &'b self,
        state: &'b SessionState,
        req_state: ReqState,
        processor: &domain::Processor,
        mut payment_data: PaymentData<F>,
        _frm_suggestion: Option<FrmSuggestion>,
        _header_payload: hyperswitch_domain_models::payments::HeaderPayload,
        _dimensions: &dimension_state::DimensionsWithProcessorAndProviderMerchantId,
    ) -> RouterResult<(PaymentCreateOperation<'b, F>, PaymentData<F>)>
    where
        F: 'b + Send,
    {
        let storage_scheme = processor.get_account().storage_scheme;
        let key_store = processor.get_key_store();
        let status = match payment_data.payment_intent.status {
            IntentStatus::RequiresPaymentMethod => match payment_data.payment_method_data {
                Some(_) => Some(IntentStatus::RequiresConfirmation),
                _ => None,
            },
            IntentStatus::RequiresConfirmation => {
                if let Some(true) = payment_data.confirm {
                    //TODO: do this later, request validation should happen before
                    Some(IntentStatus::Processing)
                } else {
                    None
                }
            }
            _ => None,
        };

        let payment_token = payment_data.token.clone();
        let connector = payment_data.payment_attempt.connector.clone();
        let straight_through_algorithm = payment_data
            .payment_attempt
            .straight_through_algorithm
            .clone();
        let authorized_amount = payment_data.payment_attempt.get_total_amount();
        let merchant_connector_id = payment_data.payment_attempt.merchant_connector_id.clone();

        let surcharge_amount = payment_data
            .surcharge_details
            .as_ref()
            .map(|surcharge_details| surcharge_details.surcharge_amount);
        let tax_amount = payment_data
            .surcharge_details
            .as_ref()
            .map(|surcharge_details| surcharge_details.tax_on_surcharge_amount);

        let routing_approach = payment_data.payment_attempt.routing_approach.clone();
        let is_stored_credential = helpers::is_stored_credential(
            &payment_data.recurring_details,
            &payment_data.pm_token,
            payment_data.mandate_id.is_some(),
            payment_data.payment_attempt.is_stored_credential,
        );
        payment_data.payment_attempt = state
            .store
            .update_payment_attempt_with_attempt_id(
                payment_data.payment_attempt,
                storage::PaymentAttemptUpdate::UpdateTrackers {
                    payment_token,
                    connector,
                    straight_through_algorithm,
                    amount_capturable: match payment_data.confirm.unwrap_or(true) {
                        true => Some(authorized_amount),
                        false => None,
                    },
                    surcharge_amount,
                    tax_amount,
                    updated_by: storage_scheme.to_string(),
                    merchant_connector_id,
                    routing_approach,
                    is_stored_credential,
                },
                storage_scheme,
                key_store,
            )
            .await
            .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;

        let customer_id = payment_data.payment_intent.customer_id.clone();

        let customer_details = payment_data.payment_intent.customer_details.clone();

        payment_data.payment_intent = state
            .store
            .update_payment_intent(
                payment_data.payment_intent,
                storage::PaymentIntentUpdate::PaymentCreateUpdate {
                    return_url: None,
                    status,
                    customer_id,
                    shipping_address_id: None,
                    billing_address_id: None,
                    customer_details,
                    updated_by: storage_scheme.to_string(),
                },
                key_store,
                storage_scheme,
            )
            .await
            .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
        req_state
            .event_context
            .event(AuditEvent::new(AuditEventType::PaymentCreate))
            .with(payment_data.to_event())
            .emit();

        // payment_data.mandate_id = response.and_then(|router_data| router_data.request.mandate_id);
        Ok((
            payments::is_confirm(self, payment_data.confirm),
            payment_data,
        ))
    }
}

impl<F: Send + Clone + Sync> ValidateRequest<F, api::PaymentsRequest, PaymentData<F>>
    for PaymentCreate
{
    #[instrument(skip_all)]
    fn validate_request<'a, 'b>(
        &'b self,
        request: &api::PaymentsRequest,
        processor: &'a domain::Processor,
    ) -> RouterResult<(PaymentCreateOperation<'b, F>, operations::ValidateResult)> {
        helpers::validate_customer_information(request)?;

        if let Some(amount) = request.amount {
            helpers::validate_max_amount(amount)?;
        }
        if let Some(session_expiry) = &request.session_expiry {
            helpers::validate_session_expiry(session_expiry.to_owned())?;
        }

        if let Some(payment_link) = &request.payment_link {
            if *payment_link {
                helpers::validate_payment_link_request(request)?;
            }
        };

        let payment_id = request.payment_id.clone().ok_or(error_stack::report!(
            errors::ApiErrorResponse::PaymentNotFound
        ))?;

        let request_merchant_id = request.merchant_id.as_ref();
        helpers::validate_merchant_id(processor.get_account().get_id(), request_merchant_id)
            .change_context(errors::ApiErrorResponse::MerchantAccountNotFound)?;

        helpers::validate_request_amount_and_amount_to_capture(
            request.amount,
            request.amount_to_capture,
            request.surcharge_details,
        )
        .change_context(errors::ApiErrorResponse::InvalidDataFormat {
            field_name: "amount_to_capture".to_string(),
            expected_format: "amount_to_capture lesser than amount".to_string(),
        })?;

        helpers::validate_amount_to_capture_and_capture_method(None, request)?;
        helpers::validate_card_data(
            request
                .payment_method_data
                .as_ref()
                .and_then(|pmd| pmd.payment_method_data.clone()),
        )?;

        helpers::validate_payment_method_fields_present(request)?;

        let mandate_type =
            helpers::validate_mandate(request, payments::is_operation_confirm(self))?;

        helpers::validate_recurring_details_and_token(
            &request.recurring_details,
            &request.payment_token,
            &request.mandate_id,
        )?;

        request.validate_stored_credential().change_context(
            errors::ApiErrorResponse::InvalidRequestData {
                message:
                    "is_stored_credential should be true when reusing stored payment method data"
                        .to_string(),
            },
        )?;

        helpers::validate_overcapture_request(
            &request.enable_overcapture,
            &request.capture_method,
        )?;

        request.validate_installment_options().map_err(|err| {
            let message = format!("invalid installment options: {err}");
            err.change_context(errors::ApiErrorResponse::InvalidRequestData { message })
        })?;

        if request.confirm.unwrap_or(false) {
            helpers::validate_installment_data_in_create(
                &request.installment_options,
                &request.installment_data,
            )?;
        }

        if request.confirm.unwrap_or(false) {
            helpers::validate_pm_or_token_given(
                &request.payment_method,
                &request
                    .payment_method_data
                    .as_ref()
                    .and_then(|pmd| pmd.payment_method_data.clone()),
                &request.payment_method_type,
                &mandate_type,
                &request.payment_token,
                &request.ctp_service_details,
            )?;

            helpers::validate_customer_id_mandatory_cases(
                request.setup_future_usage.is_some(),
                request.customer_id.as_ref().or(request
                    .customer
                    .as_ref()
                    .and_then(|customer| customer.id.as_ref())),
            )?;
        }

        if request.split_payments.is_some() {
            let amount = request.amount.get_required_value("amount")?;
            helpers::validate_platform_request_for_marketplace(
                amount,
                request.split_payments.clone(),
            )?;
        };

        let _request_straight_through: Option<api::routing::StraightThroughAlgorithm> = request
            .routing
            .clone()
            .map(|val| val.parse_value("RoutingAlgorithm"))
            .transpose()
            .change_context(errors::ApiErrorResponse::InvalidRequestData {
                message: "Invalid straight through routing rules format".to_string(),
            })
            .attach_printable("Invalid straight through routing rules format")?;

        Ok((
            Box::new(self),
            operations::ValidateResult {
                merchant_id: processor.get_account().get_id().to_owned(),
                payment_id,
                storage_scheme: processor.get_account().storage_scheme,
                requeue: matches!(
                    request.retry_action,
                    Some(api_models::enums::RetryAction::Requeue)
                ),
            },
        ))
    }
}

impl PaymentCreate {
    async fn fetch_payment_method_from_modular_service(
        &self,
        state: &SessionState,
        req: &api::PaymentsRequest,
        platform: &domain::Platform,
        payment_method_ref: &str,
    ) -> RouterResult<PaymentMethodWithRawData> {
        let profile_id = req
            .profile_id
            .clone()
            .or(platform
                .get_processor()
                .get_account()
                .get_default_profile()
                .clone())
            .get_required_value("profile_id")?;

        let pm_info = pm_transformers::fetch_payment_method_from_modular_service(
            state,
            platform,
            &profile_id,
            payment_method_ref,
            None, // CVC token data is not passed in create api
            true, // fetch raw card detail from the internal vault
        )
        .await?;
        logger::info!("Payment method fetched from PM Modular Service.");

        utils::when(
            req.get_customer_id().is_some_and(|customer_id| {
                pm_info.payment_method.customer_id.as_ref() != Some(customer_id)
            }),
            || {
                logger::info!(
                    "Payment method id does not belong to the customer id provided in the request."
                );
                Err(errors::ApiErrorResponse::PaymentMethodNotFound)
            },
        )?;

        Ok(pm_info)
    }

    /// Determines the payment method reference for modular payment flows.
    fn get_payment_method_reference(self, req: &api::PaymentsRequest) -> Option<&str> {
        self.get_recurring_payment_method_id(req)
            .or(req.payment_token.as_deref())
    }

    fn get_recurring_payment_method_id(self, req: &api::PaymentsRequest) -> Option<&str> {
        if let Some(RecurringDetails::PaymentMethodId(payment_method_id)) =
            req.recurring_details.as_ref()
        {
            Some(payment_method_id.as_str())
        } else {
            None
        }
    }

    #[cfg(feature = "v2")]
    #[instrument(skip_all)]
    #[allow(clippy::too_many_arguments)]
    pub async fn make_payment_attempt(
        payment_id: &common_utils::id_type::PaymentId,
        merchant_id: &common_utils::id_type::MerchantId,
        organization_id: &common_utils::id_type::OrganizationId,
        money: (api::Amount, enums::Currency),
        payment_method: Option<enums::PaymentMethod>,
        payment_method_type: Option<enums::PaymentMethodType>,
        request: &api::PaymentsRequest,
        browser_info: Option<serde_json::Value>,
        state: &SessionState,
        payment_method_billing_address_id: Option<String>,
        payment_method_info: &Option<domain::PaymentMethod>,
        _key_store: &domain::MerchantKeyStore,
        profile_id: common_utils::id_type::ProfileId,
        customer_acceptance: &Option<payments::CustomerAcceptance>,
        storage_scheme: enums::MerchantStorageScheme,
    ) -> RouterResult<(
        storage::PaymentAttemptNew,
        Option<api_models::payments::AdditionalPaymentData>,
    )> {
        todo!()
    }

    #[cfg(feature = "v1")]
    #[instrument(skip_all)]
    #[allow(clippy::too_many_arguments)]
    pub async fn make_payment_attempt(
        payment_id: &common_utils::id_type::PaymentId,
        platform: &domain::Platform,
        money: (api::Amount, enums::Currency),
        payment_method: Option<enums::PaymentMethod>,
        payment_method_type: Option<enums::PaymentMethodType>,
        request: &api::PaymentsRequest,
        browser_info: Option<serde_json::Value>,
        state: &SessionState,
        optional_payment_method_billing_address_id: Option<String>,
        payment_method_info: &Option<domain::PaymentMethod>,
        profile_id: common_utils::id_type::ProfileId,
        customer_acceptance: &Option<common_payments_types::CustomerAcceptance>,
        payment_method_recurring_details: Option<domain::PaymentMethodData>,
        customer_id: Option<&common_utils::id_type::CustomerId>,
        dimensions: &dimension_state::DimensionsWithProcessorAndProviderMerchantId,
    ) -> RouterResult<(
        PaymentAttempt,
        Option<api_models::payments::AdditionalPaymentData>,
    )> {
        let dimensions = dimensions.with_profile_id(profile_id.clone());
        let payment_method_data =
            request
                .payment_method_data
                .as_ref()
                .and_then(|payment_method_data_request| {
                    payment_method_data_request.payment_method_data.as_ref()
                });

        let created_at @ modified_at @ last_synced = common_utils::date_time::now();
        let status = helpers::payment_attempt_status_fsm(payment_method_data, request.confirm);
        let (amount, currency) = (money.0, Some(money.1));

        let mut additional_pm_data = request
            .payment_method_data
            .as_ref()
            .and_then(|payment_method_data_request| {
                payment_method_data_request.payment_method_data.clone()
            })
            .map(domain::PaymentMethodData::from)
            .or(payment_method_recurring_details)
            .zip(Some(profile_id.clone())) // since data is consumed by async move, profile_id needs to be send separately
            .async_map(|(payment_method_data, _)| async move {
                helpers::get_additional_payment_data(
                    &payment_method_data,
                    &*state.store,
                    state.superposition_service.as_ref(),
                    &dimensions,
                    customer_id,
                    None,
                )
                .await
            })
            .await
            .transpose()?
            .flatten();

        if additional_pm_data.is_none() {
            // If recurring payment is made using payment_method_id, then fetch payment_method_data from retrieved payment_method object
            additional_pm_data = payment_method_info.as_ref().and_then(|pm_info| {
                pm_info
                    .payment_method_data
                    .clone()
                    .map(|x| x.into_inner().expose())
                    .and_then(|v| {
                        serde_json::from_value::<PaymentMethodsData>(v)
                            .map_err(|err| {
                                logger::error!(
                                    "Unable to deserialize payment methods data: {:?}",
                                    err
                                )
                            })
                            .ok()
                    })
                    .and_then(|pmd| match pmd {
                        PaymentMethodsData::Card(card) => {
                            Some(api_models::payments::AdditionalPaymentData::Card(Box::new(
                                api::CardDetailFromLocker::from(card).into(),
                            )))
                        }
                        PaymentMethodsData::WalletDetails(wallet) => match payment_method_type {
                            Some(enums::PaymentMethodType::ApplePay) => {
                                Some(api_models::payments::AdditionalPaymentData::Wallet {
                                    apple_pay: Box::<api_models::payments::ApplepayPaymentMethod>::try_from(
                                        wallet,
                                    )
                                    .inspect_err(|err| {
                                        logger::error!(
                                            "Unable to transform PaymentMethodDataWalletInfo to ApplepayPaymentMethod: {:?}",
                                            err
                                        )
                                    })
                                    .ok(),
                                    google_pay: None,
                                    samsung_pay: None,
                                })
                            }
                            Some(enums::PaymentMethodType::GooglePay) => {
                                Some(api_models::payments::AdditionalPaymentData::Wallet {
                                    apple_pay: None,
                                    google_pay: Some(Box::new(wallet.into())),
                                    samsung_pay: None,
                                })
                            }
                            Some(enums::PaymentMethodType::SamsungPay) => {
                                Some(api_models::payments::AdditionalPaymentData::Wallet {
                                    apple_pay: None,
                                    google_pay: None,
                                    samsung_pay: Some(Box::new(wallet.into())),
                                })
                            }
                            _ => None,
                        },
                        _ => None,
                    })
                    .or_else(|| match payment_method_type {
                        Some(enums::PaymentMethodType::Paypal) => {
                            Some(api_models::payments::AdditionalPaymentData::Wallet {
                                apple_pay: None,
                                google_pay: None,
                                samsung_pay: None,
                            })
                        }
                        _ => None,
                    })
            });
        };

        let additional_pm_data_value = additional_pm_data
            .as_ref()
            .map(Encode::encode_to_value)
            .transpose()
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to encode additional pm data")?;
        let attempt_id = if core_utils::is_merchant_enabled_for_payment_id_as_connector_request_id(
            &state.conf,
            platform.get_processor(),
        ) {
            payment_id.get_string_repr().to_owned()
        } else {
            payment_id.get_attempt_id(1)
        };

        if request.mandate_data.as_ref().is_some_and(|mandate_data| {
            mandate_data.update_mandate_id.is_some() && mandate_data.mandate_type.is_some()
        }) {
            Err(errors::ApiErrorResponse::InvalidRequestData {message:"Only one field out of 'mandate_type' and 'update_mandate_id' was expected, found both".to_string()})?
        }

        let mandate_data = if let Some(update_id) = request
            .mandate_data
            .as_ref()
            .and_then(|inner| inner.update_mandate_id.clone())
        {
            let mandate_details = MandateDetails {
                update_mandate_id: Some(update_id),
            };
            Some(mandate_details)
        } else {
            None
        };

        let payment_method_type = Option::<enums::PaymentMethodType>::foreign_from((
            payment_method_type,
            additional_pm_data.as_ref(),
            payment_method,
        ));

        // TODO: remove once https://github.com/juspay/hyperswitch/issues/7421 is fixed
        let payment_method_billing_address_id = match optional_payment_method_billing_address_id {
            None => payment_method_info
                .as_ref()
                .and_then(|pm_info| pm_info.payment_method_billing_address.as_ref())
                .map(|address| {
                    address.clone().deserialize_inner_value(|value| {
                        value.parse_value::<api_models::payments::Address>("Address")
                    })
                })
                .transpose()
                .change_context(errors::ApiErrorResponse::InternalServerError)
                .ok()
                .flatten()
                .async_map(|addr| async move {
                    helpers::create_or_find_address_for_payment_by_request(
                        state,
                        Some(addr.get_inner()),
                        None,
                        platform.get_processor().get_account().get_id(),
                        payment_method_info
                            .as_ref()
                            .and_then(|pmd_info| pmd_info.customer_id.as_ref()),
                        platform.get_processor().get_key_store(),
                        payment_id,
                        platform.get_processor().get_account().storage_scheme,
                    )
                    .await
                })
                .await
                .transpose()
                .change_context(errors::ApiErrorResponse::InternalServerError)?
                .flatten()
                .map(|address| address.address_id),
            address_id => address_id,
        };

        let is_stored_credential = helpers::is_stored_credential(
            &request.recurring_details,
            &request.payment_token,
            request.mandate_id.is_some(),
            request.is_stored_credential,
        );
        Ok((
            PaymentAttempt {
                payment_id: payment_id.to_owned(),
                merchant_id: platform.get_provider().get_account().get_id().clone(),
                attempt_id,
                status,
                currency,
                payment_method,
                capture_method: request.capture_method,
                capture_on: request.capture_on,
                confirm: request.confirm.unwrap_or(false),
                created_at,
                modified_at,
                last_synced: Some(last_synced),
                authentication_type: request.authentication_type,
                browser_info,
                payment_experience: request.payment_experience,
                payment_method_type,
                payment_method_data: additional_pm_data_value,
                amount_to_capture: request.amount_to_capture,
                payment_token: request.payment_token.clone(),
                mandate_id: request.mandate_id.clone(),
                business_sub_label: request.business_sub_label.clone(),
                mandate_details: request
                    .mandate_data
                    .as_ref()
                    .and_then(|inner| inner.mandate_type.clone().map(Into::into)),
                external_three_ds_authentication_attempted: None,
                external_threeds_authentication_type: None,
                mandate_data,
                payment_method_billing_address_id,
                net_amount: hyperswitch_domain_models::payments::payment_attempt::NetAmount::from_payments_request(
                    request,
                    MinorUnit::from(amount),
                ),
                save_to_locker: None,
                connector: None,
                error_message: None,
                offer_amount: None,
                payment_method_id: payment_method_info
                    .as_ref()
                    .map(|pm_info| pm_info.get_id().clone()),
                cancellation_reason: None,
                error_code: None,
                connector_metadata: None,
                straight_through_algorithm: None,
                preprocessing_step_id: None,
                error_reason: None,
                connector_response_reference_id: None,
                multiple_capture_count: None,
                amount_capturable: MinorUnit::new(i64::default()),
                updated_by: String::default(),
                authentication_data: None,
                encoded_data: None,
                merchant_connector_id: None,
                unified_code: None,
                unified_message: None,
                fingerprint_id: None,
                authentication_connector: None,
                authentication_id: None,
                client_source: None,
                client_version: None,
                customer_acceptance: customer_acceptance
                    .clone()
                    .map(|customer_acceptance| customer_acceptance.encode_to_value())
                    .transpose()
                    .change_context(errors::ApiErrorResponse::InternalServerError)
                    .attach_printable("Failed to serialize customer_acceptance")?
                    .map(Secret::new),
                organization_id: platform.get_processor().get_account().organization_id.clone(),
                profile_id,
                connector_mandate_detail: None,
                request_extended_authorization: None,
                extended_authorization_applied: None,
                extended_authorization_last_applied_at: None,
                capture_before: None,
                card_discovery: None,
                processor_merchant_id: platform.get_processor().get_account().get_id().clone(),
                created_by: platform.get_initiator().and_then(|initiator| initiator.to_created_by()),
                setup_future_usage_applied: request.setup_future_usage,
                routing_approach: Some(common_enums::RoutingApproach::default()),
                connector_request_reference_id: None,
                network_transaction_id: None,
                network_transaction_link_id: None,
                network_details: None,
                is_stored_credential,
                authorized_amount: None,
                tokenization:request.tokenization,
                connector_transaction_id: None,
                charge_id: None,
                charges: None,
                issuer_error_code: None,
                issuer_error_message: None,
                debit_routing_savings: None,
                is_overcapture_enabled: None,
                encrypted_payment_method_data: None,
                error_details: None,
                retry_type: None,
                installment_data: None,
                external_surcharge_details: None,
                sender_payment_instrument_id: None,
            },
            additional_pm_data,

        ))
    }

    #[instrument(skip_all)]
    #[allow(clippy::too_many_arguments)]
    async fn make_payment_intent(
        state: &SessionState,
        payment_id: &common_utils::id_type::PaymentId,
        platform: &domain::Platform,
        money: (api::Amount, enums::Currency),
        request: &api::PaymentsRequest,
        shipping_address_id: Option<String>,
        payment_link_data: Option<api_models::payments::PaymentLinkResponse>,
        billing_address_id: Option<String>,
        active_attempt_id: String,
        profile_id: common_utils::id_type::ProfileId,
        session_expiry: PrimitiveDateTime,
        business_profile: &domain::Profile,
        is_payment_id_from_merchant: bool,
        payment_method_info: Option<domain::PaymentMethod>,
    ) -> RouterResult<storage::PaymentIntent> {
        let created_at @ modified_at @ last_synced = common_utils::date_time::now();

        let status = helpers::payment_intent_status_fsm(
            request
                .payment_method_data
                .as_ref()
                .and_then(|request_payment_method_data| {
                    request_payment_method_data.payment_method_data.as_ref()
                }),
            request.confirm,
        );
        let client_secret = payment_id.generate_client_secret();
        let (amount, currency) = (money.0, Some(money.1));

        let order_details = request
            .get_order_details_as_value()
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to convert order details to value")?;

        let allowed_payment_method_types = request
            .get_allowed_payment_method_types_as_value()
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Error converting allowed_payment_types to Value")?;

        let connector_metadata = request
            .get_connector_metadata_as_value()
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Error converting connector_metadata to Value")?;

        let feature_metadata = request
            .get_feature_metadata_as_value()
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Error converting feature_metadata to Value")?;

        let payment_link_id = payment_link_data.map(|pl_data| pl_data.payment_link_id);

        let request_incremental_authorization =
            core_utils::get_request_incremental_authorization_value(
                request.request_incremental_authorization,
                request.capture_method,
            )?;

        let split_payments = request.split_payments.clone();

        let mandate_type = m_helpers::get_mandate_type(
            request.mandate_data.clone(),
            request.off_session,
            request.setup_future_usage,
            request.customer_acceptance.clone(),
            request.payment_token.clone(),
            request.payment_method,
        )
        .change_context(errors::ApiErrorResponse::MandateValidationFailed {
            reason: "Expected one out of recurring_details and mandate_data but got both".into(),
        })?;

        // Derivation of directly supplied Customer data in our Payment Create Request
        let raw_customer_details = helpers::get_customer_details_from_request_or_pm_table(
            request,
            payment_method_info.as_ref(),
            mandate_type.as_ref(),
        )?
        .get_customer_data();
        let is_payment_processor_token_flow = request.recurring_details.as_ref().and_then(
            |recurring_details| match recurring_details {
                RecurringDetails::ProcessorPaymentToken(_) => Some(true),
                _ => None,
            },
        );

        let key = platform
            .get_processor()
            .get_key_store()
            .key
            .get_inner()
            .peek();

        let identifier =
            Identifier::Merchant(platform.get_processor().get_key_store().merchant_id.clone());
        let key_manager_state: KeyManagerState = state.into();

        let shipping_details_encoded = request
            .shipping
            .clone()
            .map(|shipping| Encode::encode_to_value(&shipping).map(Secret::new))
            .transpose()
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Unable to encode billing details to serde_json::Value")?;

        let billing_details_encoded = request
            .billing
            .clone()
            .map(|billing| Encode::encode_to_value(&billing).map(Secret::new))
            .transpose()
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Unable to encode billing details to serde_json::Value")?;

        let customer_details_encoded = raw_customer_details
            .map(|customer| Encode::encode_to_value(&customer).map(Secret::new))
            .transpose()
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Unable to encode shipping details to serde_json::Value")?;

        let encrypted_data = domain::types::crypto_operation(
            &key_manager_state,
            type_name!(storage::PaymentIntent),
            domain::types::CryptoOperation::BatchEncrypt(
                FromRequestEncryptablePaymentIntent::to_encryptable(
                    FromRequestEncryptablePaymentIntent {
                        shipping_details: shipping_details_encoded,
                        billing_details: billing_details_encoded,
                        customer_details: customer_details_encoded,
                    },
                ),
            ),
            identifier.clone(),
            key,
        )
        .await
        .and_then(|val| val.try_into_batchoperation())
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Unable to encrypt data")?;

        let encrypted_data = FromRequestEncryptablePaymentIntent::from_encryptable(encrypted_data)
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Unable to encrypt the payment intent data")?;

        let skip_external_tax_calculation = request.skip_external_tax_calculation;

        let tax_details = request
            .order_tax_amount
            .map(|tax_amount| diesel_models::TaxDetails {
                default: Some(diesel_models::DefaultTax {
                    order_tax_amount: tax_amount,
                }),
                payment_method_type: None,
            });
        let force_3ds_challenge_trigger = request
            .force_3ds_challenge
            .unwrap_or(business_profile.force_3ds_challenge);

        Ok(storage::PaymentIntent {
            payment_id: payment_id.to_owned(),
            merchant_id: platform.get_provider().get_account().get_id().to_owned(),
            status,
            amount: MinorUnit::from(amount),
            currency,
            description: request.description.clone(),
            created_at,
            modified_at,
            last_synced: Some(last_synced),
            client_secret: Some(client_secret),
            setup_future_usage: request.setup_future_usage,
            off_session: request.off_session,
            return_url: request.return_url.as_ref().map(|a| a.to_string()),
            shipping_address_id,
            billing_address_id,
            statement_descriptor_name: request.statement_descriptor_name.clone(),
            statement_descriptor_suffix: request.statement_descriptor_suffix.clone(),
            metadata: request.metadata.clone(),
            business_country: request.business_country,
            business_label: request.business_label.clone(),
            active_attempt: hyperswitch_domain_models::RemoteStorageObject::ForeignID(
                active_attempt_id,
            ),
            order_details,
            amount_captured: None,
            customer_id: request.get_customer_id().cloned(),
            connector_id: None,
            allowed_payment_method_types,
            connector_metadata,
            feature_metadata: feature_metadata.map(Secret::new),
            attempt_count: 1,
            profile_id: Some(profile_id),
            merchant_decision: None,
            payment_link_id,
            payment_confirm_source: None,
            surcharge_applicable: None,
            updated_by: platform
                .get_processor()
                .get_account()
                .storage_scheme
                .to_string(),
            request_incremental_authorization,
            incremental_authorization_allowed: None,
            authorization_count: None,
            fingerprint_id: None,
            session_expiry: Some(session_expiry),
            request_external_three_ds_authentication: request
                .request_external_three_ds_authentication,
            split_payments,
            frm_metadata: request.frm_metadata.clone(),
            billing_details: encrypted_data.billing_details,
            customer_details: encrypted_data.customer_details,
            merchant_order_reference_id: request.merchant_order_reference_id.clone(),
            shipping_details: encrypted_data.shipping_details,
            is_payment_processor_token_flow,
            organization_id: platform
                .get_processor()
                .get_account()
                .organization_id
                .clone(),
            shipping_cost: request.shipping_cost,
            tax_details,
            skip_external_tax_calculation,
            request_extended_authorization: request.request_extended_authorization,
            psd2_sca_exemption_type: request.psd2_sca_exemption_type,
            processor_merchant_id: platform.get_processor().get_account().get_id().to_owned(),
            created_by: platform
                .get_initiator()
                .and_then(|initiator| initiator.to_created_by()),
            force_3ds_challenge: request.force_3ds_challenge,
            force_3ds_challenge_trigger: Some(force_3ds_challenge_trigger),
            is_iframe_redirection_enabled: request
                .is_iframe_redirection_enabled
                .or(business_profile.is_iframe_redirection_enabled),
            is_payment_id_from_merchant: Some(is_payment_id_from_merchant),
            payment_channel: request.payment_channel.clone(),
            order_date: request.order_date,
            discount_amount: request.discount_amount,
            duty_amount: request.duty_amount,
            tax_status: request.tax_status,
            shipping_amount_tax: request.shipping_amount_tax,
            enable_partial_authorization: request.enable_partial_authorization,
            enable_overcapture: request.enable_overcapture,
            mit_category: request.mit_category,
            billing_descriptor: request.billing_descriptor.clone(),
            tokenization: request.tokenization,
            partner_merchant_identifier_details: request
                .partner_merchant_identifier_details
                .clone(),
            state_metadata: None,
            installment_options: request.installment_options.clone(),
            profile_acquirer_id: request.profile_acquirer_id.clone(),
            external_surcharge_strategy: request.external_surcharge_strategy,
            external_surcharge_applicable: None,
        })
    }
}

#[instrument(skip_all)]
pub fn payments_create_request_validation(
    req: &api::PaymentsRequest,
) -> RouterResult<(api::Amount, enums::Currency)> {
    let currency = req.currency.get_required_value("currency")?;
    let amount = req.amount.get_required_value("amount")?;
    Ok((amount, currency))
}

#[allow(clippy::too_many_arguments)]
async fn create_payment_link(
    request: &api::PaymentsRequest,
    payment_link_config: api_models::admin::PaymentLinkConfig,
    platform: &domain::Platform,
    payment_id: common_utils::id_type::PaymentId,
    db: &dyn StorageInterface,
    amount: api::Amount,
    description: Option<String>,
    profile_id: common_utils::id_type::ProfileId,
    domain_name: String,
    session_expiry: PrimitiveDateTime,
    locale: Option<String>,
) -> RouterResult<Option<api_models::payments::PaymentLinkResponse>> {
    let created_at @ last_modified_at = Some(common_utils::date_time::now());
    let payment_link_id = utils::generate_id(consts::ID_LENGTH, "plink");
    let locale_str = locale.unwrap_or("en".to_owned());
    let processor_merchant_id = platform.get_processor().get_account().get_id();
    let open_payment_link = format!(
        "{}/payment_link/{}/{}?locale={}",
        domain_name,
        processor_merchant_id.get_string_repr(),
        payment_id.get_string_repr(),
        locale_str.clone(),
    );

    let secure_link = payment_link_config.allowed_domains.as_ref().map(|_| {
        format!(
            "{}/payment_link/s/{}/{}?locale={}",
            domain_name,
            processor_merchant_id.get_string_repr(),
            payment_id.get_string_repr(),
            locale_str,
        )
    });

    let payment_link_config_encoded_value = payment_link_config.encode_to_value().change_context(
        errors::ApiErrorResponse::InvalidDataValue {
            field_name: "payment_link_config",
        },
    )?;

    let payment_link_req = storage::PaymentLinkNew {
        payment_link_id: payment_link_id.clone(),
        payment_id: payment_id.clone(),
        merchant_id: platform.get_provider().get_account().get_id().clone(),
        link_to_pay: open_payment_link.clone(),
        amount: MinorUnit::from(amount),
        currency: request.currency,
        created_at,
        last_modified_at,
        fulfilment_time: Some(session_expiry),
        custom_merchant_name: Some(payment_link_config.seller_name),
        description,
        payment_link_config: Some(payment_link_config_encoded_value),
        profile_id: Some(profile_id),
        secure_link,
        processor_merchant_id: Some(processor_merchant_id.clone()),
        created_by: platform
            .get_initiator()
            .and_then(|initiator| initiator.to_created_by())
            .map(|created_by| created_by.to_string()),
    };
    let payment_link_db = db
        .insert_payment_link(payment_link_req)
        .await
        .to_duplicate_response(errors::ApiErrorResponse::GenericDuplicateError {
            message: "payment link already exists!".to_string(),
        })?;

    Ok(Some(api_models::payments::PaymentLinkResponse {
        link: payment_link_db.link_to_pay.clone(),
        secure_link: payment_link_db.secure_link,
        payment_link_id: payment_link_db.payment_link_id,
    }))
}
