use crate::{
    core::{
        api_locking::{self, GetLockingInput},
        payments::operations::Operation,
    },
    services::authorization::permissions::Permission,
};
pub mod helpers;
use actix_web::{web, Responder};
use error_stack::report;
use hyperswitch_domain_models::{ext_traits::OptionExt, payments::HeaderPayload};
#[cfg(feature = "v1")]
use hyperswitch_interfaces::api::ConnectorSpecifications;
use hyperswitch_masking::{PeekInterface, Secret};
use router_env::{env, instrument, logger, tracing, types, Flow};

use super::app::ReqState;
#[cfg(feature = "v2")]
use crate::core::payment_method_balance;
#[cfg(feature = "v2")]
use crate::core::revenue_recovery::api as recovery;
#[cfg(feature = "v1")]
use crate::routes::payments::payments::operations::payment_recurrence::PaymentRecurrence;
use crate::{
    self as app,
    core::{
        configs::dimension_state,
        errors::{self, http_not_implemented},
        payments::{self, transformers::ToResponse, OperationSessionGetters, PaymentRedirectFlow},
    },
    routes::lock_utils,
    services::{api, authentication as auth},
    types::{
        api::{
            self as api_types, enums as api_enums,
            payments::{self as payment_types, PaymentIdTypeExt},
            ConnectorData, GetToken,
        },
        domain,
        transformers::{ForeignTryFrom, ForeignTryInto},
    },
};

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsCreate, payment_id))]
pub async fn payments_create(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsRequest>,
) -> impl Responder {
    let flow = Flow::PaymentsCreate;
    let mut payload = json_payload.into_inner();
    if let Err(err) = payload
        .validate()
        .map_err(|message| errors::ApiErrorResponse::InvalidRequestData { message })
    {
        return api::log_and_return_error_response(err.into());
    };

    if let Err(err) = payload
        .payment_link_config
        .as_ref()
        .map(|cfg| {
            cfg.theme_config
                .validate()
                .map_err(|message| errors::ApiErrorResponse::InvalidRequestData { message })
        })
        .transpose()
    {
        return api::log_and_return_error_response(err.into());
    };

    if let Some(api_enums::CaptureMethod::Scheduled) = payload.capture_method {
        return http_not_implemented();
    };

    if let Err(err) = get_or_generate_payment_id(&mut payload) {
        return api::log_and_return_error_response(err);
    }

    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    tracing::Span::current().record(
        "payment_id",
        payload
            .payment_id
            .as_ref()
            .map(|payment_id_type| payment_id_type.get_payment_intent_id())
            .transpose()
            .unwrap_or_default()
            .as_ref()
            .map(|id| id.get_string_repr())
            .unwrap_or_default(),
    );

    let locking_action = payload.get_locking_input(flow.clone());

    let auth_type = match env::which() {
        env::Env::Production => {
            &auth::InternalMerchantIdProfileIdAuth(auth::HeaderAuth(auth::ApiKeyAuth {
                allow_connected_scope_operation: true,
                allow_platform_self_operation: false,
            }))
        }
        _ => auth::auth_type(
            &auth::InternalMerchantIdProfileIdAuth(auth::HeaderAuth(auth::ApiKeyAuth {
                allow_connected_scope_operation: true,
                allow_platform_self_operation: false,
            })),
            &auth::InternalMerchantIdProfileIdAuth(auth::JWTAuth {
                permission: Permission::ProfilePaymentWrite,
                allow_connected: true,
                allow_platform: false,
            }),
            req.headers(),
        ),
    };

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            authorize_verify_select::<_>(
                payments::PaymentCreate,
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                header_payload.clone(),
                req,
                api::AuthFlow::Client,
            )
        },
        auth_type,
        locking_action,
    ))
    .await
}

#[cfg(feature = "v2")]
pub async fn recovery_payments_create(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::RecoveryPaymentsCreate>,
) -> impl Responder {
    let flow = Flow::RecoveryPaymentsCreate;
    let mut payload = json_payload.into_inner();
    Box::pin(api::server_wrap(
        flow,
        state,
        &req.clone(),
        payload,
        |state, auth: auth::AuthenticationData, req_payload, req_state| {
            recovery::custom_revenue_recovery_core(
                state.to_owned(),
                req_state,
                auth.platform,
                auth.profile,
                req_payload,
            )
        },
        &auth::V2ApiKeyAuth {
            allow_connected_scope_operation: false,
            allow_platform_self_operation: false,
        },
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsCreateIntent, payment_id))]
pub async fn payments_create_intent(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsCreateIntentRequest>,
) -> impl Responder {
    use hyperswitch_domain_models::payments::PaymentIntentData;

    let flow = Flow::PaymentsCreateIntent;
    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };
    let global_payment_id =
        common_utils::id_type::GlobalPaymentId::generate(&state.conf.cell_information.id);

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        json_payload.into_inner(),
        |state, auth: auth::AuthenticationData, req, req_state| {
            payments::payments_intent_core::<
                api_types::PaymentCreateIntent,
                payment_types::PaymentsIntentResponse,
                _,
                _,
                PaymentIntentData<api_types::PaymentCreateIntent>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile,
                payments::operations::PaymentIntentCreate,
                req,
                global_payment_id.clone(),
                header_payload.clone(),
            )
        },
        match env::which() {
            env::Env::Production => &auth::V2ApiKeyAuth {
                allow_connected_scope_operation: false,
                allow_platform_self_operation: false,
            },
            _ => auth::auth_type(
                &auth::V2ApiKeyAuth {
                    allow_connected_scope_operation: false,
                    allow_platform_self_operation: false,
                },
                &auth::JWTAuth {
                    permission: Permission::ProfilePaymentWrite,
                    allow_connected: false,
                    allow_platform: false,
                },
                req.headers(),
            ),
        },
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsGetIntent, payment_id))]
pub async fn payments_get_intent(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    path: web::Path<common_utils::id_type::GlobalPaymentId>,
) -> impl Responder {
    use api_models::payments::PaymentsGetIntentRequest;
    use hyperswitch_domain_models::payments::PaymentIntentData;

    let flow = Flow::PaymentsGetIntent;
    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    let payload = PaymentsGetIntentRequest {
        id: path.into_inner(),
    };

    let global_payment_id = payload.id.clone();

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            payments::payments_intent_core::<
                api_types::PaymentGetIntent,
                payment_types::PaymentsIntentResponse,
                _,
                _,
                PaymentIntentData<api_types::PaymentGetIntent>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile,
                payments::operations::PaymentGetIntent,
                req,
                global_payment_id.clone(),
                header_payload.clone(),
            )
        },
        auth::api_or_client_or_jwt_auth(
            &auth::V2ApiKeyAuth {
                allow_connected_scope_operation: false,
                allow_platform_self_operation: false,
            },
            &auth::V2ClientAuth(common_utils::types::authentication::ResourceId::Payment(
                global_payment_id.clone(),
            )),
            &auth::JWTAuth {
                permission: Permission::ProfileRevenueRecoveryRead,
                allow_connected: false,
                allow_platform: false,
            },
            req.headers(),
        ),
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(all(feature = "v2", feature = "olap"))]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsGetIntent, payment_id))]
pub async fn revenue_recovery_get_intent(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    path: web::Path<common_utils::id_type::GlobalPaymentId>,
) -> impl Responder {
    use api_models::payments::PaymentsGetIntentRequest;
    use hyperswitch_domain_models::payments::PaymentIntentData;

    let flow = Flow::PaymentsGetIntent;
    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    let payload = PaymentsGetIntentRequest {
        id: path.into_inner(),
    };

    let global_payment_id = payload.id.clone();

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            payments::revenue_recovery_get_intent_core(
                state,
                req_state,
                auth.platform,
                auth.profile,
                req,
                global_payment_id.clone(),
                header_payload.clone(),
            )
        },
        auth::api_or_client_or_jwt_auth(
            &auth::V2ApiKeyAuth {
                allow_connected_scope_operation: false,
                allow_platform_self_operation: false,
            },
            &auth::V2ClientAuth(common_utils::types::authentication::ResourceId::Payment(
                global_payment_id.clone(),
            )),
            &auth::JWTAuth {
                permission: Permission::ProfileRevenueRecoveryRead,
                allow_connected: false,
                allow_platform: false,
            },
            req.headers(),
        ),
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentAttemptsList, payment_id,))]
pub async fn list_payment_attempts(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    path: web::Path<common_utils::id_type::GlobalPaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentAttemptsList;
    let payment_intent_id = path.into_inner();

    let payload = api_models::payments::PaymentAttemptListRequest {
        payment_intent_id: payment_intent_id.clone(),
    };

    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload.clone(),
        |session_state, auth, req_payload, req_state| {
            payments::payments_list_attempts_using_payment_intent_id::<
                payments::operations::PaymentGetListAttempts,
                api_models::payments::PaymentAttemptListResponse,
                api_models::payments::PaymentAttemptListRequest,
                payments::operations::payment_attempt_list::PaymentGetListAttempts,
                hyperswitch_domain_models::payments::PaymentAttemptListData<
                    payments::operations::PaymentGetListAttempts,
                >,
            >(
                session_state,
                req_state,
                auth.platform,
                auth.profile,
                payments::operations::PaymentGetListAttempts,
                payload.clone(),
                req_payload.payment_intent_id,
                header_payload.clone(),
            )
        },
        auth::auth_type(
            &auth::V2ApiKeyAuth {
                allow_connected_scope_operation: false,
                allow_platform_self_operation: false,
            },
            &auth::JWTAuth {
                permission: Permission::ProfilePaymentRead,
                allow_connected: false,
                allow_platform: false,
            },
            req.headers(),
        ),
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsCreateAndConfirmIntent, payment_id))]
pub async fn payments_create_and_confirm_intent(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsRequest>,
) -> impl Responder {
    let flow = Flow::PaymentsCreateAndConfirmIntent;
    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    let auth_type = if state.conf.merchant_id_auth.merchant_id_auth_enabled {
        &auth::MerchantIdAuth
    } else {
        match env::which() {
            env::Env::Production => &auth::V2ApiKeyAuth {
                allow_connected_scope_operation: false,
                allow_platform_self_operation: false,
            },
            _ => auth::auth_type(
                &auth::V2ApiKeyAuth {
                    allow_connected_scope_operation: false,
                    allow_platform_self_operation: false,
                },
                &auth::JWTAuth {
                    permission: Permission::ProfilePaymentWrite,
                    allow_connected: false,
                    allow_platform: false,
                },
                req.headers(),
            ),
        }
    };

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        json_payload.into_inner(),
        |state, auth: auth::AuthenticationData, request, req_state| {
            payments::payments_create_and_confirm_intent(
                state,
                req_state,
                auth.platform,
                auth.profile,
                request,
                header_payload.clone(),
            )
        },
        auth_type,
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsUpdateIntent, payment_id))]
pub async fn payments_update_intent(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsUpdateIntentRequest>,
    path: web::Path<common_utils::id_type::GlobalPaymentId>,
) -> impl Responder {
    use hyperswitch_domain_models::payments::PaymentIntentData;

    let flow = Flow::PaymentsUpdateIntent;
    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    let internal_payload = internal_payload_types::PaymentsGenericRequestWithResourceId {
        global_payment_id: path.into_inner(),
        payload: json_payload.into_inner(),
    };

    let global_payment_id = internal_payload.global_payment_id.clone();

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        internal_payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            payments::payments_intent_core::<
                api_types::PaymentUpdateIntent,
                payment_types::PaymentsIntentResponse,
                _,
                _,
                PaymentIntentData<api_types::PaymentUpdateIntent>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile,
                payments::operations::PaymentUpdateIntent,
                req.payload,
                global_payment_id.clone(),
                header_payload.clone(),
            )
        },
        &auth::V2ApiKeyAuth {
            allow_connected_scope_operation: false,
            allow_platform_self_operation: false,
        },
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip(state, req), fields(flow = ?Flow::PaymentsStart, payment_id))]
pub async fn payments_start(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    path: web::Path<(
        common_utils::id_type::PaymentId,
        common_utils::id_type::MerchantId,
        String,
    )>,
) -> impl Responder {
    let flow = Flow::PaymentsStart;
    let (payment_id, merchant_id, attempt_id) = path.into_inner();
    let payload = payment_types::PaymentsStartRequest {
        payment_id: payment_id.clone(),
        merchant_id: merchant_id.clone(),
        attempt_id: attempt_id.clone(),
    };

    let locking_action = payload.get_locking_input(flow.clone());
    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            payments::payments_core::<
                api_types::Authorize,
                payment_types::PaymentsResponse,
                _,
                _,
                _,
                payments::PaymentData<api_types::Authorize>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::operations::PaymentStart,
                req,
                api::AuthFlow::Client,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                HeaderPayload::default(),
                None,
            )
        },
        &auth::MerchantIdAuth(merchant_id),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip(state, req), fields(flow, payment_id))]
pub async fn payments_retrieve(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    path: web::Path<common_utils::id_type::PaymentId>,
    json_payload: web::Query<payment_types::PaymentRetrieveBody>,
) -> impl Responder {
    let flow = match json_payload.force_sync {
        Some(true) => Flow::PaymentsRetrieveForceSync,
        _ => Flow::PaymentsRetrieve,
    };
    let payment_id = path.into_inner();
    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    let payload = payment_types::PaymentsRetrieveRequest {
        resource_id: payment_types::PaymentIdType::PaymentIntentId(payment_id),
        merchant_id: json_payload.merchant_id.clone(),
        force_sync: json_payload.force_sync.unwrap_or(false),
        client_secret: json_payload.client_secret.clone(),
        expand_attempts: json_payload.expand_attempts,
        expand_captures: json_payload.expand_captures,
        all_keys_required: json_payload.all_keys_required,
        ..Default::default()
    };
    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    tracing::Span::current().record("flow", flow.to_string());

    let api_auth = auth::ApiKeyAuth {
        allow_connected_scope_operation: true,
        allow_platform_self_operation: false,
    };

    // checking or validating
    let (auth_type, auth_flow) = match auth::check_internal_api_key_auth(
        req.headers(),
        &payload,
        api_auth,
        state.conf.internal_merchant_id_profile_id_auth.clone(),
    ) {
        Ok(auth) => auth,
        Err(err) => return api::log_and_return_error_response(report!(err)),
    };

    let locking_action = payload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth, mut req, req_state| {
            // If client_secret is provided via SDK authorization header, use it
            if let Some(client_secret) = auth.client_secret {
                req.client_secret = Some(client_secret);
            }

            payments::payments_core::<
                api_types::PSync,
                payment_types::PaymentsResponse,
                _,
                _,
                _,
                payments::PaymentData<api_types::PSync>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::PaymentStatus,
                req,
                auth_flow,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                header_payload.clone(),
                None,
            )
        },
        auth::auth_type(
            &*auth_type,
            &auth::JWTAuth {
                permission: Permission::ProfilePaymentRead,
                allow_connected: true,
                allow_platform: false,
            },
            req.headers(),
        ),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip(state, req), fields(flow, payment_id))]
pub async fn payments_retrieve_with_gateway_creds(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentRetrieveBodyWithCredentials>,
) -> impl Responder {
    let api_auth = auth::ApiKeyAuth {
        allow_connected_scope_operation: true,
        allow_platform_self_operation: false,
    };

    let (auth_type, _auth_flow) =
        match auth::check_authorization_header_or_get_auth(req.headers(), api_auth) {
            Ok(auth) => auth,
            Err(err) => return api::log_and_return_error_response(report!(err)),
        };

    tracing::Span::current().record("payment_id", json_payload.payment_id.get_string_repr());

    let payload = payment_types::PaymentsRetrieveRequest {
        resource_id: payment_types::PaymentIdType::PaymentIntentId(json_payload.payment_id.clone()),
        merchant_id: json_payload.merchant_id.clone(),
        force_sync: json_payload.force_sync.unwrap_or(false),
        merchant_connector_details: json_payload.merchant_connector_details.clone(),
        ..Default::default()
    };

    let flow = match json_payload.force_sync {
        Some(true) => Flow::PaymentsRetrieveForceSync,
        _ => Flow::PaymentsRetrieve,
    };

    tracing::Span::current().record("flow", flow.to_string());

    let locking_action = payload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth, mut req, req_state| {
            if let Some(client_secret) = auth.client_secret {
                req.client_secret = Some(client_secret);
            }
            payments::payments_core::<
                api_types::PSync,
                payment_types::PaymentsResponse,
                _,
                _,
                _,
                payments::PaymentData<api_types::PSync>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::PaymentStatus,
                req,
                api::AuthFlow::Merchant,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                HeaderPayload::default(),
                None,
            )
        },
        &*auth_type,
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsUpdate, payment_id))]
pub async fn payments_update(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsUpdate;
    let mut payload = json_payload.into_inner();
    if let Err(err) = payload
        .validate()
        .map_err(|message| errors::ApiErrorResponse::InvalidRequestData { message })
    {
        return api::log_and_return_error_response(err.into());
    };

    if let Some(api_enums::CaptureMethod::Scheduled) = payload.capture_method {
        return http_not_implemented();
    };

    let payment_id = path.into_inner();

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    payload.payment_id = Some(payment_types::PaymentIdType::PaymentIntentId(payment_id));
    let api_auth = auth::ApiKeyAuth {
        allow_connected_scope_operation: true,
        allow_platform_self_operation: false,
    };
    let (auth_type, auth_flow) = match auth::check_internal_api_key_auth_no_client_secret(
        req.headers(),
        api_auth,
        state.conf.internal_merchant_id_profile_id_auth.clone(),
    ) {
        Ok(auth) => auth,
        Err(err) => return api::log_and_return_error_response(report!(err)),
    };

    let locking_action = payload.get_locking_input(flow.clone());

    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            payments::payments_core::<
                api_types::UpdatePostConfirm,
                payment_types::PaymentsResponse,
                _,
                _,
                _,
                payments::PaymentData<api_types::UpdatePostConfirm>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::PaymentUpdate,
                req,
                auth_flow,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                header_payload.clone(),
                None,
            )
        },
        &*auth_type,
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsPostSessionTokens, payment_id))]
pub async fn payments_post_session_tokens(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsPostSessionTokensRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsPostSessionTokens;

    let payment_id = path.into_inner();
    let payload = payment_types::PaymentsPostSessionTokensRequest {
        payment_id,
        ..json_payload.into_inner()
    };
    tracing::Span::current().record("payment_id", payload.payment_id.get_string_repr());
    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    let locking_action = payload.get_locking_input(flow.clone());

    // Determine auth type based on Authorization header presence
    let auth: Box<dyn auth::AuthenticateAndFetch<auth::AuthenticationData, _>> =
        match req.headers().get(actix_web::http::header::AUTHORIZATION) {
            // If Authorization header is present, use SdkAuthorizationAuth
            Some(_) => Box::new(auth::SdkAuthorizationAuth {
                allow_connected_scope_operation: true,
                allow_platform_self_operation: false,
            }),
            // If Authorization header is not present, use PublishableKeyAuth
            None => {
                // For PublishableKeyAuth, client_secret is mandatory
                match payload
                    .client_secret
                    .as_ref()
                    .map(|client_secret| client_secret.peek())
                    .check_value_present("client_secret")
                    .map_err(|_| errors::ApiErrorResponse::MissingRequiredField {
                        field_name: "client_secret",
                    }) {
                    Ok(_) => {}
                    Err(err) => return api::log_and_return_error_response(report!(err)),
                }

                Box::new(auth::HeaderAuth(auth::PublishableKeyAuth {
                    allow_connected_scope_operation: true,
                    allow_platform_self_operation: false,
                }))
            }
        };

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, mut req, req_state| {
            // If client_secret is provided via SDK authorization header, use it
            if let Some(client_secret) = auth.client_secret {
                req.client_secret = Some(Secret::new(client_secret));
            }

            payments::payments_core::<
                api_types::PostSessionTokens,
                payment_types::PaymentsPostSessionTokensResponse,
                _,
                _,
                _,
                payments::PaymentData<api_types::PostSessionTokens>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::PaymentPostSessionTokens,
                req,
                api::AuthFlow::Client,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                header_payload.clone(),
                None,
            )
        },
        &*auth,
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsUpdateMetadata, payment_id))]
pub async fn payments_update_metadata(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsUpdateMetadataRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsUpdateMetadata;

    let payment_id = path.into_inner();
    let payload = payment_types::PaymentsUpdateMetadataRequest {
        payment_id,
        ..json_payload.into_inner()
    };
    tracing::Span::current().record("payment_id", payload.payment_id.get_string_repr());
    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    let locking_action = payload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            payments::payments_core::<
                api_types::UpdateMetadata,
                payment_types::PaymentsUpdateMetadataResponse,
                _,
                _,
                _,
                payments::PaymentData<api_types::UpdateMetadata>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::PaymentUpdateMetadata,
                req,
                api::AuthFlow::Client,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                header_payload.clone(),
                None,
            )
        },
        &auth::HeaderAuth(auth::ApiKeyAuth {
            allow_connected_scope_operation: true,
            allow_platform_self_operation: false,
        }),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsConfirm, payment_id))]
pub async fn payments_confirm(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsConfirm;
    let mut payload = json_payload.into_inner();
    if let Err(err) = payload
        .validate()
        .map_err(|message| errors::ApiErrorResponse::InvalidRequestData { message })
    {
        return api::log_and_return_error_response(err.into());
    };

    if let Some(api_enums::CaptureMethod::Scheduled) = payload.capture_method {
        return http_not_implemented();
    };

    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    if let Err(err) = helpers::populate_browser_info(&req, &mut payload, &header_payload) {
        return api::log_and_return_error_response(err);
    }

    let payment_id = path.into_inner();
    tracing::Span::current().record("payment_id", payment_id.get_string_repr());
    payload.payment_id = Some(payment_types::PaymentIdType::PaymentIntentId(payment_id));
    payload.confirm = Some(true);

    let api_auth = auth::ApiKeyAuth {
        allow_connected_scope_operation: true,
        allow_platform_self_operation: false,
    };

    let (auth_type, auth_flow) = match auth::check_internal_api_key_auth(
        req.headers(),
        &payload,
        api_auth,
        state.conf.internal_merchant_id_profile_id_auth.clone(),
    ) {
        Ok(auth) => auth,
        Err(e) => return api::log_and_return_error_response(e),
    };

    let locking_action = payload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, mut req, req_state| {
            // If client_secret is provided via SDK authorization header, use it
            if let Some(client_secret) = auth.client_secret {
                req.client_secret = Some(client_secret);
            }

            authorize_verify_select::<_>(
                payments::PaymentConfirm,
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                header_payload.clone(),
                req,
                auth_flow,
            )
        },
        &*auth_type,
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsCapture, payment_id))]
pub async fn payments_capture(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsCaptureRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let payment_id = path.into_inner();
    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    let flow = Flow::PaymentsCapture;
    let payload = payment_types::PaymentsCaptureRequest {
        payment_id,
        ..json_payload.into_inner()
    };

    let locking_action = payload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, payload, req_state| {
            payments::payments_core::<
                api_types::Capture,
                payment_types::PaymentsResponse,
                _,
                _,
                _,
                payments::PaymentData<api_types::Capture>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::PaymentCapture,
                payload,
                api::AuthFlow::Merchant,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                HeaderPayload::default(),
                None,
            )
        },
        auth::auth_type(
            &auth::HeaderAuth(auth::ApiKeyAuth {
                allow_connected_scope_operation: true,
                allow_platform_self_operation: false,
            }),
            &auth::JWTAuth {
                permission: Permission::ProfilePaymentWrite,
                allow_connected: true,
                allow_platform: false,
            },
            req.headers(),
        ),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::SessionUpdateTaxCalculation, payment_id))]
pub async fn payments_dynamic_tax_calculation(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsDynamicTaxCalculationRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::SessionUpdateTaxCalculation;
    let payment_id = path.into_inner();
    let payload = payment_types::PaymentsDynamicTaxCalculationRequest {
        payment_id,
        ..json_payload.into_inner()
    };
    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(error) => {
            logger::error!(
                ?error,
                "Failed to get headers in payments_connector_session"
            );
            HeaderPayload::default()
        }
    };
    tracing::Span::current().record("payment_id", payload.payment_id.get_string_repr());
    let locking_action = payload.get_locking_input(flow.clone());

    // Determine auth type based on Authorization header presence
    let auth: Box<dyn auth::AuthenticateAndFetch<auth::AuthenticationData, _>> =
        match req.headers().get(actix_web::http::header::AUTHORIZATION) {
            // If Authorization header is present, use SdkAuthorizationAuth
            Some(_) => Box::new(auth::SdkAuthorizationAuth {
                allow_connected_scope_operation: true,
                allow_platform_self_operation: false,
            }),
            // If Authorization header is not present, use PublishableKeyAuth
            None => {
                // For PublishableKeyAuth, client_secret is mandatory
                match payload
                    .client_secret
                    .as_ref()
                    .map(|client_secret| client_secret.peek())
                    .check_value_present("client_secret")
                    .map_err(|_| errors::ApiErrorResponse::MissingRequiredField {
                        field_name: "client_secret",
                    }) {
                    Ok(_) => {}
                    Err(err) => return api::log_and_return_error_response(report!(err)),
                }

                Box::new(auth::HeaderAuth(auth::PublishableKeyAuth {
                    allow_connected_scope_operation: true,
                    allow_platform_self_operation: false,
                }))
            }
        };

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth, mut payload, req_state| {
            // If client_secret is provided via SDK authorization header, use it
            if let Some(client_secret) = auth.client_secret {
                payload.client_secret = Some(Secret::new(client_secret));
            }

            payments::payments_core::<
                api_types::SdkSessionUpdate,
                payment_types::PaymentsDynamicTaxCalculationResponse,
                _,
                _,
                _,
                _,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::PaymentSessionUpdate,
                payload,
                api::AuthFlow::Client,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                header_payload.clone(),
                None,
            )
        },
        &*auth,
        locking_action,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsSessionToken, payment_id))]
pub async fn payments_connector_session(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsSessionRequest>,
    path: web::Path<common_utils::id_type::GlobalPaymentId>,
) -> impl Responder {
    use hyperswitch_domain_models::payments::PaymentIntentData;
    let flow = Flow::PaymentsSessionToken;

    let global_payment_id = path.into_inner();
    tracing::Span::current().record("payment_id", global_payment_id.get_string_repr());

    let internal_payload = internal_payload_types::PaymentsGenericRequestWithResourceId {
        global_payment_id: global_payment_id.clone(),
        payload: json_payload.into_inner(),
    };

    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    let locking_action = internal_payload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        internal_payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            let payment_id = req.global_payment_id;
            let request = req.payload;
            let operation = payments::operations::PaymentSessionIntent;
            payments::payments_session_core::<
                api_types::Session,
                payment_types::PaymentsSessionResponse,
                _,
                _,
                _,
                PaymentIntentData<api_types::Session>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile,
                operation,
                request,
                payment_id,
                payments::CallConnectorAction::Trigger,
                header_payload.clone(),
            )
        },
        &auth::V2ClientAuth(common_utils::types::authentication::ResourceId::Payment(
            global_payment_id,
        )),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsSessionToken, payment_id))]
pub async fn payments_connector_session(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsSessionRequest>,
) -> impl Responder {
    let flow = Flow::PaymentsSessionToken;
    let payload = json_payload.into_inner();

    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(error) => {
            logger::error!(
                ?error,
                "Failed to get headers in payments_connector_session"
            );
            HeaderPayload::default()
        }
    };

    tracing::Span::current().record("payment_id", payload.payment_id.get_string_repr());

    let locking_action = payload.get_locking_input(flow.clone());

    // Determine auth type based on Authorization header presence
    let auth: Box<dyn auth::AuthenticateAndFetch<auth::AuthenticationData, _>> =
        match req.headers().get(actix_web::http::header::AUTHORIZATION) {
            // If Authorization header is present, use SdkAuthorizationAuth
            Some(_) => Box::new(auth::SdkAuthorizationAuth {
                allow_connected_scope_operation: true,
                allow_platform_self_operation: false,
            }),
            // If Authorization header is not present, use PublishableKeyAuth
            None => {
                // For PublishableKeyAuth, client_secret is mandatory
                match payload
                    .client_secret
                    .check_value_present("client_secret")
                    .map_err(|_| errors::ApiErrorResponse::MissingRequiredField {
                        field_name: "client_secret",
                    }) {
                    Ok(_) => {}
                    Err(err) => return api::log_and_return_error_response(report!(err)),
                }

                Box::new(auth::HeaderAuth(auth::PublishableKeyAuth {
                    allow_connected_scope_operation: true,
                    allow_platform_self_operation: false,
                }))
            }
        };

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, mut payload, req_state| {
            // If client_secret is provided via SDK authorization header, use it
            if let Some(client_secret) = auth.client_secret {
                payload.client_secret = Some(client_secret);
            }
            payments::payments_core::<
                api_types::Session,
                payment_types::PaymentsSessionResponse,
                _,
                _,
                _,
                payments::PaymentData<api_types::Session>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::PaymentSession,
                payload,
                api::AuthFlow::Client,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                header_payload.clone(),
                None,
            )
        },
        &*auth,
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsRedirect, payment_id))]
pub async fn payments_redirect_response(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: Option<web::Form<serde_json::Value>>,
    path: web::Path<(
        common_utils::id_type::PaymentId,
        common_utils::id_type::MerchantId,
        String,
    )>,
) -> impl Responder {
    let flow = Flow::PaymentsRedirect;
    let (payment_id, merchant_id, connector) = path.into_inner();
    let param_string = req.query_string();

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    let payload = payments::PaymentsRedirectResponseData {
        resource_id: payment_types::PaymentIdType::PaymentIntentId(payment_id),
        merchant_id: Some(merchant_id.clone()),
        force_sync: true,
        json_payload: json_payload.map(|payload| payload.0),
        param: Some(param_string.to_string()),
        connector: Some(connector),
        creds_identifier: None,
    };
    let locking_action = payload.get_locking_input(flow.clone());
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            <payments::PaymentRedirectSync as PaymentRedirectFlow>::handle_payments_redirect_response(
                &payments::PaymentRedirectSync {},
                state,
                req_state,
                auth.platform,
                req,
            )
        },
        &auth::MerchantIdAuth(merchant_id),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsRedirect, payment_id))]
pub async fn payments_redirect_response_with_creds_identifier(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    path: web::Path<(
        common_utils::id_type::PaymentId,
        common_utils::id_type::MerchantId,
        String,
        String,
    )>,
) -> impl Responder {
    let (payment_id, merchant_id, connector, creds_identifier) = path.into_inner();
    let param_string = req.query_string();

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    let payload = payments::PaymentsRedirectResponseData {
        resource_id: payment_types::PaymentIdType::PaymentIntentId(payment_id),
        merchant_id: Some(merchant_id.clone()),
        force_sync: true,
        json_payload: None,
        param: Some(param_string.to_string()),
        connector: Some(connector),
        creds_identifier: Some(creds_identifier),
    };
    let flow = Flow::PaymentsRedirect;
    let locking_action = payload.get_locking_input(flow.clone());
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
           <payments::PaymentRedirectSync as PaymentRedirectFlow>::handle_payments_redirect_response(
                &payments::PaymentRedirectSync {},
                state,
                req_state,
                auth.platform,
                req,
            )
        },
        &auth::MerchantIdAuth(merchant_id),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow =? Flow::PaymentsRedirect, payment_id))]
pub async fn payments_complete_authorize_redirect(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: Option<web::Form<serde_json::Value>>,
    path: web::Path<(
        common_utils::id_type::PaymentId,
        common_utils::id_type::MerchantId,
        String,
    )>,
) -> impl Responder {
    let flow = Flow::PaymentsRedirect;
    let (payment_id, merchant_id, connector) = path.into_inner();
    let param_string = req.query_string();

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    let payload = payments::PaymentsRedirectResponseData {
        resource_id: payment_types::PaymentIdType::PaymentIntentId(payment_id),
        merchant_id: Some(merchant_id.clone()),
        param: Some(param_string.to_string()),
        json_payload: json_payload.map(|s| s.0),
        force_sync: false,
        connector: Some(connector),
        creds_identifier: None,
    };
    let locking_action = payload.get_locking_input(flow.clone());
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            <payments::PaymentRedirectCompleteAuthorize as PaymentRedirectFlow>::handle_payments_redirect_response(
                &payments::PaymentRedirectCompleteAuthorize {},
                state,
                req_state,
                auth.platform,
                req,
            )
        },
        &auth::MerchantIdAuth(merchant_id),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow =? Flow::PaymentsRedirect, payment_id))]
pub async fn payments_complete_authorize_redirect_with_creds_identifier(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: Option<web::Form<serde_json::Value>>,
    path: web::Path<(
        common_utils::id_type::PaymentId,
        common_utils::id_type::MerchantId,
        String,
        String,
    )>,
) -> impl Responder {
    let flow = Flow::PaymentsRedirect;
    let (payment_id, merchant_id, connector, creds_identifier) = path.into_inner();
    let param_string = req.query_string();

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    let payload = payments::PaymentsRedirectResponseData {
        resource_id: payment_types::PaymentIdType::PaymentIntentId(payment_id),
        merchant_id: Some(merchant_id.clone()),
        param: Some(param_string.to_string()),
        json_payload: json_payload.map(|s| s.0),
        force_sync: false,
        connector: Some(connector),
        creds_identifier: Some(creds_identifier),
    };
    let locking_action = payload.get_locking_input(flow.clone());
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            <payments::PaymentRedirectCompleteAuthorize as PaymentRedirectFlow>::handle_payments_redirect_response(
                &payments::PaymentRedirectCompleteAuthorize {},
                state,
                req_state,
                auth.platform,
                req,
            )
        },
        &auth::MerchantIdAuth(merchant_id),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow =? Flow::PaymentsCompleteAuthorize, payment_id))]
pub async fn payments_complete_authorize(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsCompleteAuthorizeRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsCompleteAuthorize;
    let mut payload = json_payload.into_inner();

    let payment_id = path.into_inner();
    payload.payment_id.clone_from(&payment_id);

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    let payment_confirm_req = payment_types::PaymentsRequest {
        payment_id: Some(payment_types::PaymentIdType::PaymentIntentId(
            payment_id.clone(),
        )),
        shipping: payload.shipping.clone(),
        client_secret: payload
            .client_secret
            .clone()
            .map(|client_secret| client_secret.peek().clone()),
        threeds_method_comp_ind: payload.threeds_method_comp_ind.clone(),
        ..Default::default()
    };

    let api_auth = auth::ApiKeyAuth {
        allow_connected_scope_operation: true,
        allow_platform_self_operation: false,
    };

    let (auth_type, auth_flow) =
        match auth::check_sdk_auth_and_get_auth(req.headers(), &payment_confirm_req, api_auth) {
            Ok(auth) => auth,
            Err(err) => return api::log_and_return_error_response(report!(err)),
        };

    let locking_action = payload.get_locking_input(flow.clone());
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, mut req, req_state| {
            // If client_secret is provided via SDK authorization header, use it
            if let Some(client_secret) = auth.client_secret {
                req.client_secret = Some(Secret::new(client_secret));
            }
            payments::payments_core::<
                api_types::CompleteAuthorize,
                payment_types::PaymentsResponse,
                _,
                _,
                _,
                payments::PaymentData<api_types::CompleteAuthorize>,
            >(
                state.clone(),
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::operations::payment_complete_authorize::CompleteAuthorize,
                payment_confirm_req.clone(),
                auth_flow,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                HeaderPayload::default(),
                None,
            )
        },
        &*auth_type,
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsCancel, payment_id))]
pub async fn payments_cancel(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsCancelRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsCancel;
    let mut payload = json_payload.into_inner();
    let payment_id = path.into_inner();

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    payload.payment_id = payment_id;

    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    let locking_action = payload.get_locking_input(flow.clone());
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            let header_payload = header_payload.clone();
            async move {
                let operation = payments::PaymentCancel;
                let payment_id_type =
                    payment_types::PaymentIdType::PaymentIntentId(req.payment_id.clone());

                // Getting the intent status to determine which flow to use
                let payment_pre_fetched_info = {
                    let preliminary_dimensions = dimension_state::Dimensions::new()
                        .with_processor_merchant_id(
                            auth.platform.get_processor().get_processor_merchant_id(),
                        )
                        .with_provider_merchant_id(
                            auth.platform.get_provider().get_provider_merchant_id(),
                        );
                    let tracker_response = operation
                        .to_get_tracker()?
                        .get_trackers(
                            &state,
                            &payment_id_type,
                            &req,
                            &auth.platform,
                            api::AuthFlow::Merchant,
                            payments::operations::PaymentFlowKind::Standard,
                            &header_payload,
                            crate::core::payment_methods::transformers::PaymentMethodFetchData::default(),
                            &preliminary_dimensions,
                            None
                        )
                        .await?;

                    let get_tracker_payment_data: payments::PaymentData<api_types::Void> =
                        tracker_response.payment_data;

                    // Create PaymentPreFetchedInformation to pass to payments_core
                        payments::operations::PaymentPreFetchedInformation {
                            payment_intent: get_tracker_payment_data.payment_intent.clone(),
                            payment_attempt: get_tracker_payment_data.payment_attempt.clone(),
                        }
                };

                // Check if the payment status is RequiresCustomerAction and connector supports pre-authorize cancel
                let supports_pre_authorize_cancel =
                    if let Some(ref connector_name_str) = payment_pre_fetched_info.payment_attempt.connector {
                        ConnectorData::get_connector_by_name(
                            &state.conf.connectors,
                            connector_name_str,
                            GetToken::Connector,
                            None,
                        )
                        .map(|connector_data| {
                            connector_data.connector.is_pre_authorize_cancel_supported(payment_pre_fetched_info.payment_attempt.payment_method_type)
                        })
                        .unwrap_or(false)
                    } else {
                        false
                    };

                if payment_pre_fetched_info.payment_intent.status == api_enums::IntentStatus::RequiresCustomerAction
                    && supports_pre_authorize_cancel
                {
                    Box::pin(payments::payments_core::<
                        api_types::PreAuthorizeVoid,
                        payment_types::PaymentsResponse,
                        _,
                        _,
                        _,
                        payments::PaymentData<api_types::PreAuthorizeVoid>,
                    >(
                        state,
                        req_state,
                        auth.platform,
                        auth.profile.map(|profile| profile.get_id().clone()),
                        operation,
                        req,
                        api::AuthFlow::Merchant,
                        payments::CallConnectorAction::Trigger,
                        None,
                        None,
                        header_payload.clone(),
                        Some(payment_pre_fetched_info),))
                    .await
                } else {
                    Box::pin(payments::payments_core::<
                        api_types::Void,
                        payment_types::PaymentsResponse,
                        _,
                        _,
                        _,
                        payments::PaymentData<api_types::Void>,
                    >(
                        state,
                        req_state,
                        auth.platform,
                        auth.profile.map(|profile| profile.get_id().clone()),
                        operation,
                        req,
                        api::AuthFlow::Merchant,
                        payments::CallConnectorAction::Trigger,
                        None,
                        None,
                        header_payload.clone(),
                        Some(payment_pre_fetched_info),))
                    .await
                }
            }
        },
        auth::auth_type(
            &auth::HeaderAuth(auth::ApiKeyAuth {
                allow_connected_scope_operation: true,
                allow_platform_self_operation: false,
            }),
            &auth::JWTAuth {
                permission: Permission::ProfilePaymentWrite,
                allow_connected: true,
                allow_platform: false,
            },
            req.headers(),
        ),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsCancel, payment_id))]
pub async fn payments_cancel(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsCancelRequest>,
    path: web::Path<common_utils::id_type::GlobalPaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsCancel;
    let global_payment_id = path.into_inner();

    tracing::Span::current().record("payment_id", global_payment_id.get_string_repr());

    let internal_payload = internal_payload_types::PaymentsGenericRequestWithResourceId {
        global_payment_id: global_payment_id.clone(),
        payload: json_payload.into_inner(),
    };

    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    let locking_action = internal_payload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        internal_payload,
        |state, auth: auth::AuthenticationData, req, req_state| async {
            let payment_id = req.global_payment_id;
            let request = req.payload;

            let operation = payments::operations::payment_cancel_v2::PaymentsCancel;

            Box::pin(payments::payments_core::<
                hyperswitch_domain_models::router_flow_types::Void,
                api_models::payments::PaymentsCancelResponse,
                _,
                _,
                _,
                hyperswitch_domain_models::payments::PaymentCancelData<
                    hyperswitch_domain_models::router_flow_types::Void,
                >,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile,
                operation,
                request,
                payment_id,
                payments::CallConnectorAction::Trigger,
                header_payload.clone(),
            ))
            .await
        },
        auth::auth_type(
            &auth::V2ApiKeyAuth {
                allow_connected_scope_operation: false,
                allow_platform_self_operation: false,
            },
            &auth::JWTAuth {
                permission: Permission::ProfilePaymentWrite,
                allow_connected: false,
                allow_platform: false,
            },
            req.headers(),
        ),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsCancelPostCapture, payment_id))]
pub async fn payments_cancel_post_capture(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsCancelPostCaptureRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsCancelPostCapture;
    let mut payload = json_payload.into_inner();
    let payment_id = path.into_inner();

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    payload.payment_id = payment_id;
    let locking_action = payload.get_locking_input(flow.clone());
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            payments::payments_core::<
                api_types::PostCaptureVoid,
                payment_types::PaymentsResponse,
                _,
                _,
                _,
                payments::PaymentData<api_types::PostCaptureVoid>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::PaymentCancelPostCapture,
                req,
                api::AuthFlow::Merchant,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                HeaderPayload::default(),
                None,
            )
        },
        &auth::HeaderAuth(auth::ApiKeyAuth {
            allow_connected_scope_operation: true,
            allow_platform_self_operation: false,
        }),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsCancelPostCaptureSync, payment_id))]
pub async fn payments_cancel_post_capture_retrieve(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsCancelPostCaptureSync;
    let payment_id = path.into_inner();

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    let locking_action = payment_id.get_locking_input(flow.clone());
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payment_id,
        |state, auth: auth::AuthenticationData, req, req_state| {
            payments::payments_core::<
                api_types::PostCaptureVoidSync,
                payment_types::PaymentsResponse,
                _,
                _,
                _,
                payments::PaymentData<api_types::PostCaptureVoidSync>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::PaymentCancelPostCaptureSync,
                req,
                api::AuthFlow::Merchant,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                HeaderPayload::default(),
                None,
            )
        },
        &auth::HeaderAuth(auth::ApiKeyAuth {
            allow_connected_scope_operation: true,
            allow_platform_self_operation: false,
        }),
        locking_action,
    ))
    .await
}

#[instrument(skip_all, fields(flow = ?Flow::PaymentsList))]
#[cfg(all(feature = "olap", feature = "v1"))]
pub async fn payments_list(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    payload: web::Query<payment_types::PaymentListConstraints>,
) -> impl Responder {
    let flow = Flow::PaymentsList;
    let payload = payload.into_inner();
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, _| {
            payments::list_payments(state, auth.platform, None, req)
        },
        auth::auth_type(
            &auth::HeaderAuth(auth::ApiKeyAuth {
                allow_connected_scope_operation: true,
                allow_platform_self_operation: false,
            }),
            &auth::JWTAuth {
                permission: Permission::MerchantPaymentRead,
                allow_connected: true,
                allow_platform: false,
            },
            req.headers(),
        ),
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[instrument(skip_all, fields(flow = ?Flow::PaymentsList))]
#[cfg(all(feature = "olap", feature = "v2"))]
pub async fn revenue_recovery_invoices_list(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    payload: web::Query<payment_types::PaymentListConstraints>,
) -> impl Responder {
    let flow = Flow::PaymentsList;
    let payload = payload.into_inner();
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, _| {
            payments::revenue_recovery_list_payments(state, auth.platform, req)
        },
        auth::auth_type(
            &auth::V2ApiKeyAuth {
                allow_connected_scope_operation: false,
                allow_platform_self_operation: false,
            },
            &auth::JWTAuth {
                permission: Permission::MerchantPaymentRead,
                allow_connected: false,
                allow_platform: false,
            },
            req.headers(),
        ),
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(feature = "v2")]
pub async fn payments_list(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    payload: web::Query<payment_types::PaymentListConstraints>,
) -> impl Responder {
    let flow = Flow::PaymentsList;
    let payload = payload.into_inner();
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, _| {
            payments::list_payments(state, auth.platform, req)
        },
        auth::auth_type(
            &auth::V2ApiKeyAuth {
                allow_connected_scope_operation: false,
                allow_platform_self_operation: false,
            },
            &auth::JWTAuth {
                permission: Permission::MerchantPaymentRead,
                allow_connected: true,
                allow_platform: false,
            },
            req.headers(),
        ),
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[instrument(skip_all, fields(flow = ?Flow::PaymentsList))]
#[cfg(all(feature = "olap", feature = "v1"))]
pub async fn profile_payments_list(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    payload: web::Query<payment_types::PaymentListConstraints>,
) -> impl Responder {
    let flow = Flow::PaymentsList;
    let payload = payload.into_inner();
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, _| {
            payments::list_payments(
                state,
                auth.platform,
                auth.profile.map(|profile| vec![profile.get_id().clone()]),
                req,
            )
        },
        auth::auth_type(
            &auth::HeaderAuth(auth::ApiKeyAuth {
                allow_connected_scope_operation: true,
                allow_platform_self_operation: false,
            }),
            &auth::JWTAuth {
                permission: Permission::ProfilePaymentRead,
                allow_connected: true,
                allow_platform: false,
            },
            req.headers(),
        ),
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[instrument(skip_all, fields(flow = ?Flow::PaymentsList))]
#[cfg(all(feature = "olap", feature = "v1"))]
pub async fn payments_list_by_filter(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    payload: web::Json<payment_types::PaymentListFilterConstraints>,
) -> impl Responder {
    let flow = Flow::PaymentsList;
    let payload = payload.into_inner();
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, _| {
            payments::apply_filters_on_payments(state, auth.platform, None, req)
        },
        &auth::JWTAuth {
            permission: Permission::MerchantPaymentRead,
            allow_connected: true,
            allow_platform: false,
        },
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[instrument(skip_all, fields(flow = ?Flow::PaymentsList))]
#[cfg(all(feature = "olap", feature = "v1"))]
pub async fn profile_payments_list_by_filter(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    payload: web::Json<payment_types::PaymentListFilterConstraints>,
) -> impl Responder {
    let flow = Flow::PaymentsList;
    let payload = payload.into_inner();
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, _| {
            payments::apply_filters_on_payments(
                state,
                auth.platform,
                auth.profile.map(|profile| vec![profile.get_id().clone()]),
                req,
            )
        },
        &auth::JWTAuth {
            permission: Permission::ProfilePaymentRead,
            allow_connected: true,
            allow_platform: false,
        },
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[instrument(skip_all, fields(flow = ?Flow::PaymentsList))]
#[cfg(all(feature = "olap", feature = "v1"))]
pub async fn get_filters_for_payments(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    payload: web::Json<common_utils::types::TimeRange>,
) -> impl Responder {
    let flow = Flow::PaymentsList;
    let payload = payload.into_inner();
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, _| {
            payments::get_filters_for_payments(state, auth.platform, req)
        },
        &auth::JWTAuth {
            permission: Permission::MerchantPaymentRead,
            allow_connected: true,
            allow_platform: false,
        },
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[instrument(skip_all, fields(flow = ?Flow::PaymentsFilters))]
#[cfg(feature = "olap")]
pub async fn get_payment_filters(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
) -> impl Responder {
    let flow = Flow::PaymentsFilters;
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        (),
        |state, auth: auth::AuthenticationData, _, _| {
            payments::get_payment_filters(state, auth.platform, None)
        },
        &auth::JWTAuth {
            permission: Permission::MerchantPaymentRead,
            allow_connected: true,
            allow_platform: false,
        },
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[instrument(skip_all, fields(flow = ?Flow::PaymentsFilters))]
#[cfg(all(feature = "olap", feature = "v2"))]
pub async fn get_payment_filters_profile(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
) -> impl Responder {
    let flow = Flow::PaymentsFilters;
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        (),
        |state, auth: auth::AuthenticationData, _, _| {
            payments::get_payment_filters(
                state,
                auth.platform,
                Some(vec![auth.profile.get_id().clone()]),
            )
        },
        &auth::JWTAuth {
            permission: Permission::ProfilePaymentRead,
            allow_connected: true,
            allow_platform: false,
        },
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[instrument(skip_all, fields(flow = ?Flow::PaymentsFilters))]
#[cfg(all(feature = "olap", feature = "v1"))]
pub async fn get_payment_filters_profile(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
) -> impl Responder {
    let flow = Flow::PaymentsFilters;
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        (),
        |state, auth: auth::AuthenticationData, _, _| {
            payments::get_payment_filters(
                state,
                auth.platform,
                auth.profile.map(|profile| vec![profile.get_id().clone()]),
            )
        },
        &auth::JWTAuth {
            permission: Permission::ProfilePaymentRead,
            allow_connected: true,
            allow_platform: false,
        },
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[instrument(skip_all, fields(flow = ?Flow::PaymentsAggregate))]
#[cfg(feature = "olap")]
pub async fn get_payments_aggregates(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    payload: web::Query<common_utils::types::TimeRange>,
) -> impl Responder {
    let flow = Flow::PaymentsAggregate;
    let payload = payload.into_inner();
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, _| {
            payments::get_aggregates_for_payments(state, auth.platform, None, req)
        },
        &auth::JWTAuth {
            permission: Permission::MerchantPaymentRead,
            allow_connected: true,
            allow_platform: false,
        },
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(all(feature = "oltp", feature = "v1"))]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsApprove, payment_id))]
pub async fn payments_approve(
    state: web::Data<app::AppState>,
    http_req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsApproveRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let mut payload = json_payload.into_inner();
    let payment_id = path.into_inner();

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    payload.payment_id = payment_id;
    let flow = Flow::PaymentsApprove;
    let fpayload = FPaymentsApproveRequest(&payload);
    let locking_action = fpayload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow.clone(),
        state,
        &http_req,
        payload.clone(),
        |state, auth: auth::AuthenticationData, req, req_state| {
            payments::payments_core::<
                api_types::Capture,
                payment_types::PaymentsResponse,
                _,
                _,
                _,
                payments::PaymentData<api_types::Capture>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::PaymentApprove,
                payment_types::PaymentsCaptureRequest {
                    payment_id: req.payment_id,
                    ..Default::default()
                },
                api::AuthFlow::Merchant,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                HeaderPayload::default(),
                None,
            )
        },
        match env::which() {
            env::Env::Production => &auth::HeaderAuth(auth::ApiKeyAuth {
                allow_connected_scope_operation: true,
                allow_platform_self_operation: false,
            }),
            _ => auth::auth_type(
                &auth::HeaderAuth(auth::ApiKeyAuth {
                    allow_connected_scope_operation: true,
                    allow_platform_self_operation: false,
                }),
                &auth::JWTAuth {
                    permission: Permission::ProfilePaymentWrite,
                    allow_connected: true,
                    allow_platform: false,
                },
                http_req.headers(),
            ),
        },
        locking_action,
    ))
    .await
}

#[cfg(all(feature = "oltp", feature = "v1"))]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsReject, payment_id))]
pub async fn payments_reject(
    state: web::Data<app::AppState>,
    http_req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsRejectRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let mut payload = json_payload.into_inner();
    let payment_id = path.into_inner();

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    payload.payment_id = payment_id;
    let flow = Flow::PaymentsReject;
    let fpayload = FPaymentsRejectRequest(&payload);
    let locking_action = fpayload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow.clone(),
        state,
        &http_req,
        payload.clone(),
        |state, auth: auth::AuthenticationData, req, req_state| {
            payments::payments_core::<
                api_types::Void,
                payment_types::PaymentsResponse,
                _,
                _,
                _,
                payments::PaymentData<api_types::Void>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::PaymentReject,
                payment_types::PaymentsCancelRequest {
                    payment_id: req.payment_id,
                    cancellation_reason: Some("Rejected by merchant".to_string()),
                    ..Default::default()
                },
                api::AuthFlow::Merchant,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                HeaderPayload::default(),
                None,
            )
        },
        match env::which() {
            env::Env::Production => &auth::HeaderAuth(auth::ApiKeyAuth {
                allow_connected_scope_operation: true,
                allow_platform_self_operation: false,
            }),
            _ => auth::auth_type(
                &auth::HeaderAuth(auth::ApiKeyAuth {
                    allow_connected_scope_operation: true,
                    allow_platform_self_operation: false,
                }),
                &auth::JWTAuth {
                    permission: Permission::ProfilePaymentWrite,
                    allow_connected: true,
                    allow_platform: false,
                },
                http_req.headers(),
            ),
        },
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[allow(clippy::too_many_arguments)]
async fn authorize_verify_select<Op>(
    operation: Op,
    state: app::SessionState,
    req_state: ReqState,
    platform: domain::Platform,
    profile_id: Option<common_utils::id_type::ProfileId>,
    header_payload: HeaderPayload,
    req: api_models::payments::PaymentsRequest,
    auth_flow: api::AuthFlow,
) -> errors::RouterResponse<api_models::payments::PaymentsResponse>
where
    Op: Sync
        + Clone
        + std::fmt::Debug
        + Operation<
            api_types::Authorize,
            api_models::payments::PaymentsRequest,
            Data = payments::PaymentData<api_types::Authorize>,
        > + Operation<
            api_types::SetupMandate,
            api_models::payments::PaymentsRequest,
            Data = payments::PaymentData<api_types::SetupMandate>,
        >,
{
    // TODO: Change for making it possible for the flow to be inferred internally or through validation layer
    // This is a temporary fix.
    // After analyzing the code structure,
    // the operation are flow agnostic, and the flow is only required in the post_update_tracker
    // Thus the flow can be generated just before calling the connector instead of explicitly passing it here.

    let should_call_proxy_for_payments_core =
        helpers::should_call_proxy_for_payments_core(req.clone());

    if should_call_proxy_for_payments_core {
        // no list of eligible connectors will be passed in the confirm call
        logger::debug!("Authorize call for NTI and Card Details flow");
        payments::proxy_for_payments_core::<
            api_types::Authorize,
            payment_types::PaymentsResponse,
            _,
            _,
            _,
            payments::PaymentData<api_types::Authorize>,
        >(
            state,
            req_state,
            platform,
            profile_id,
            operation,
            req.clone(),
            auth_flow,
            payments::CallConnectorAction::Trigger,
            header_payload,
            req.all_keys_required,
        )
        .await
    } else {
        let eligible_connectors = req.connector.clone();
        let eligible_routable_connectors = eligible_connectors.clone().map(|connectors| {
            connectors
                .into_iter()
                .flat_map(|c| c.foreign_try_into())
                .collect()
        });
        let dimensions = dimension_state::Dimensions::new()
            .with_processor_merchant_id(platform.get_processor().get_processor_merchant_id())
            .with_provider_merchant_id(platform.get_provider().get_provider_merchant_id());
        match req.payment_type.unwrap_or_default() {
            api_models::enums::PaymentType::Normal
            | api_models::enums::PaymentType::RecurringMandate
            | api_models::enums::PaymentType::NewMandate
            | api_models::enums::PaymentType::Installment => {
                // The external vault proxy flow is non-PCI: the card is vaulted in an external
                // vault. Two confirm shapes route here:
                //   - `VaultDataCard`: inline vault card data.
                //   - `VaultCardTokenData`: a saved card referenced by the top-level
                //     `payment_token`; its vault tokens are retrieved from the modular PM service.
                // In both cases the confirm request itself is used to call the proxy core
                // directly — no conversion to a dedicated proxy request.
                let should_call_external_vault_proxy = req
                    .payment_method_data
                    .as_ref()
                    .and_then(|pmd| pmd.payment_method_data.as_ref())
                    .map(|data| {
                        matches!(
                            data,
                            api_models::payments::PaymentMethodData::ProxyCard(_)
                                | api_models::payments::PaymentMethodData::VaultCardTokenData(_)
                        )
                    })
                    .unwrap_or(false);

                if should_call_external_vault_proxy {
                    // A single-call create+confirm (the create endpoint with `confirm = true`)
                    // reaches here before any intent exists, so drive the proxy core with
                    // `PaymentCreate`: it persists the intent/attempt and swaps to
                    // `PaymentExternalVaultProxyConfirm` for the downstream connector call. The
                    // standalone confirm endpoint already has an intent, so it runs the proxy
                    // confirm operation directly.
                    let is_payment_create = format!("{operation:?}") == "PaymentCreate";
                    if is_payment_create {
                        Box::pin(payments::external_vault_proxy_for_payments_core::<
                            api_types::ExternalVaultProxy,
                            payment_types::PaymentsResponse,
                            _,
                            _,
                            _,
                            payments::PaymentData<api_types::ExternalVaultProxy>,
                        >(
                            state,
                            req_state,
                            platform,
                            profile_id,
                            payments::PaymentCreate,
                            req,
                            auth_flow,
                            payments::CallConnectorAction::Trigger,
                            header_payload,
                            None,
                        ))
                        .await
                    } else {
                        Box::pin(payments::external_vault_proxy_for_payments_core::<
                            api_types::ExternalVaultProxy,
                            payment_types::PaymentsResponse,
                            _,
                            _,
                            _,
                            payments::PaymentData<api_types::ExternalVaultProxy>,
                        >(
                            state,
                            req_state,
                            platform,
                            profile_id,
                            payments::PaymentExternalVaultProxyConfirm,
                            req,
                            auth_flow,
                            payments::CallConnectorAction::Trigger,
                            header_payload,
                            None,
                        ))
                        .await
                    }
                } else {
                    let (payment_data, _req, connector_http_status_code, external_latency) =
                        Box::pin(payments::payments_operation_core::<
                            api_types::Authorize,
                            _,
                            _,
                            _,
                            payments::PaymentData<api_types::Authorize>,
                        >(
                            &state,
                            req_state.clone(),
                            &platform,
                            profile_id.clone(),
                            operation.clone(),
                            req.clone(),
                            payments::CallConnectorAction::Trigger,
                            None,
                            auth_flow,
                            eligible_routable_connectors.clone(),
                            header_payload.clone(),
                            &dimensions,
                            None,
                        ))
                        .await?;

                    let connector = payment_data.get_payment_attempt_connector();

                    if let Some(connector_name) = connector {
                        let connector_data = ConnectorData::get_connector_by_name(
                            &state.conf.connectors,
                            connector_name,
                            GetToken::Connector,
                            None,
                        )?;
                        let should_continue_further = connector_data
                            .connector
                            .is_payment_recurrence_operation_needed(
                                &payment_data.payment_intent.clone(),
                            )
                            .unwrap_or(false);
                        if should_continue_further {
                            logger::info!(
                            "Re-invoking payments_operation_core | should_continue_further: {} | payment_id: {:?}",
                            should_continue_further,
                            payment_data.get_payment_intent().payment_id,
                        );
                            let (pd, _req, connector_status_code, ext_latency) =
                                Box::pin(payments::payments_operation_core::<
                                    api_types::SetupMandate,
                                    _,
                                    _,
                                    _,
                                    payments::PaymentData<api_types::SetupMandate>,
                                >(
                                    &state,
                                    req_state,
                                    &platform,
                                    profile_id,
                                    PaymentRecurrence,
                                    req,
                                    payments::CallConnectorAction::Trigger,
                                    None,
                                    auth_flow,
                                    eligible_routable_connectors,
                                    header_payload.clone(),
                                    &dimensions,
                                    None,
                                ))
                                .await?;
                            let total_ext_latency = match (external_latency, ext_latency) {
                                (Some(l1), Some(l2)) => Some(l1 + l2),
                                (Some(l), None) | (None, Some(l)) => Some(l),
                                (None, None) => None,
                            };
                            return payment_types::PaymentsResponse::generate_response(
                                pd,
                                auth_flow,
                                &state.base_url,
                                operation,
                                &state.conf.connector_request_reference_id_config,
                                connector_status_code,
                                total_ext_latency,
                                header_payload.x_hs_latency,
                                &platform,
                            );
                        }
                    }

                    payment_types::PaymentsResponse::generate_response(
                        payment_data,
                        auth_flow,
                        &state.base_url,
                        operation,
                        &state.conf.connector_request_reference_id_config,
                        connector_http_status_code,
                        external_latency,
                        header_payload.x_hs_latency,
                        &platform,
                    )
                }
            }
            api_models::enums::PaymentType::SetupMandate => {
                payments::payments_core::<
                    api_types::SetupMandate,
                    payment_types::PaymentsResponse,
                    _,
                    _,
                    _,
                    payments::PaymentData<api_types::SetupMandate>,
                >(
                    state,
                    req_state,
                    platform,
                    profile_id,
                    operation,
                    req,
                    auth_flow,
                    payments::CallConnectorAction::Trigger,
                    None,
                    eligible_connectors,
                    header_payload,
                    None,
                )
                .await
            }
        }
    }
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsIncrementalAuthorization, payment_id))]
pub async fn payments_incremental_authorization(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsIncrementalAuthorizationRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsIncrementalAuthorization;
    let mut payload = json_payload.into_inner();
    let payment_id = path.into_inner();

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    payload.payment_id = payment_id;
    let locking_action = payload.get_locking_input(flow.clone());
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            payments::payments_core::<
                api_types::IncrementalAuthorization,
                payment_types::PaymentsResponse,
                _,
                _,
                _,
                payments::PaymentData<api_types::IncrementalAuthorization>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::PaymentIncrementalAuthorization,
                req,
                api::AuthFlow::Merchant,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                HeaderPayload::default(),
                None,
            )
        },
        &auth::HeaderAuth(auth::ApiKeyAuth {
            allow_connected_scope_operation: true,
            allow_platform_self_operation: false,
        }),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsExtendAuthorization, payment_id))]
pub async fn payments_extend_authorization(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsExtendAuthorization;
    let payment_id = path.into_inner();

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());
    let payload = payment_types::PaymentsExtendAuthorizationRequest { payment_id };

    let locking_action = payload.get_locking_input(flow.clone());
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            payments::payments_core::<
                api_types::ExtendAuthorization,
                payment_types::PaymentsResponse,
                _,
                _,
                _,
                payments::PaymentData<api_types::ExtendAuthorization>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile.map(|profile| profile.get_id().clone()),
                payments::PaymentExtendAuthorization,
                req,
                api::AuthFlow::Merchant,
                payments::CallConnectorAction::Trigger,
                None,
                None,
                HeaderPayload::default(),
                None,
            )
        },
        &auth::HeaderAuth(auth::ApiKeyAuth {
            allow_connected_scope_operation: true,
            allow_platform_self_operation: false,
        }),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsExternalAuthentication, payment_id))]
pub async fn payments_external_authentication(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsExternalAuthenticationRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsExternalAuthentication;
    let mut payload = json_payload.into_inner();
    let payment_id = path.into_inner();

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    payload.payment_id = payment_id;
    let locking_action = payload.get_locking_input(flow.clone());

    // Determine auth type based on Authorization header presence
    let auth: Box<dyn auth::AuthenticateAndFetch<auth::AuthenticationData, _>> =
        match req.headers().get(actix_web::http::header::AUTHORIZATION) {
            // If Authorization header is present, use SdkAuthorizationAuth
            Some(_) => Box::new(auth::SdkAuthorizationAuth {
                allow_connected_scope_operation: true,
                allow_platform_self_operation: false,
            }),
            // If Authorization header is not present, use PublishableKeyAuth
            None => {
                // For PublishableKeyAuth, client_secret is mandatory
                match payload
                    .client_secret
                    .as_ref()
                    .map(|client_secret| client_secret.peek())
                    .check_value_present("client_secret")
                    .map_err(|_| errors::ApiErrorResponse::MissingRequiredField {
                        field_name: "client_secret",
                    }) {
                    Ok(_) => {}
                    Err(err) => return api::log_and_return_error_response(report!(err)),
                }

                Box::new(auth::HeaderAuth(auth::PublishableKeyAuth {
                    allow_connected_scope_operation: true,
                    allow_platform_self_operation: false,
                }))
            }
        };

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, mut req, _| {
            // If client_secret is provided via SDK authorization header, use it
            if let Some(client_secret) = auth.client_secret {
                req.client_secret = Some(Secret::new(client_secret));
            }

            payments::payment_external_authentication::<
                hyperswitch_domain_models::router_flow_types::Authenticate,
            >(state, auth.platform, req)
        },
        &*auth,
        locking_action,
    ))
    .await
}

#[cfg(feature = "v1")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsAuthorize, payment_id))]
pub async fn post_3ds_payments_authorize(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: Option<web::Form<serde_json::Value>>,
    path: web::Path<(
        common_utils::id_type::PaymentId,
        common_utils::id_type::MerchantId,
        String,
    )>,
) -> impl Responder {
    let flow = Flow::PaymentsAuthorize;

    let (payment_id, merchant_id, connector) = path.into_inner();
    tracing::Span::current().record("payment_id", payment_id.get_string_repr());
    let param_string = req.query_string();
    let payload = payments::PaymentsRedirectResponseData {
        resource_id: payment_types::PaymentIdType::PaymentIntentId(payment_id),
        merchant_id: Some(merchant_id.clone()),
        force_sync: true,
        json_payload: json_payload.map(|payload| payload.0),
        param: Some(param_string.to_string()),
        connector: Some(connector),
        creds_identifier: None,
    };

    let locking_action = payload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            <payments::PaymentAuthenticateCompleteAuthorize as PaymentRedirectFlow>::handle_payments_redirect_response(
                &payments::PaymentAuthenticateCompleteAuthorize {},
                state,
                req_state,
                auth.platform,
                req,
            )
        },
        &auth::MerchantIdAuth(merchant_id),
        locking_action,
    ))
    .await
}

#[cfg(all(feature = "olap", feature = "v1"))]
pub async fn payments_manual_update(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsManualUpdateRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsManualUpdate;
    let mut payload = json_payload.into_inner();
    let payment_id = path.into_inner();

    let locking_action = payload.get_locking_input(flow.clone());

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    payload.payment_id = payment_id;

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, _auth, req, _req_state| payments::payments_manual_update(state, req),
        &auth::AdminApiAuthWithMerchantIdFromHeader,
        locking_action,
    ))
    .await
}

#[cfg(all(feature = "olap", feature = "v1"))]
/// Manually update payment status from Review to Succeeded or Failed (Dashboard API with JWT auth)
#[instrument(skip_all, fields(flow = ?Flow::PaymentsManualStatusUpdate, payment_id))]
pub async fn payments_manual_status_update(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<api_models::payments::PaymentsManualStatusUpdateRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsManualStatusUpdate;
    let payload = json_payload.into_inner();
    let payment_id = path.into_inner();

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, _req_state| {
            payments::payments_manual_status_update(state, auth.platform, payment_id.clone(), req)
        },
        &auth::JWTAuth {
            permission: Permission::ProfilePaymentWrite,
            allow_connected: true,
            allow_platform: true,
        },
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(feature = "v1")]
/// Retrieve endpoint for merchant to fetch the encrypted customer payment method data
#[instrument(skip_all, fields(flow = ?Flow::GetExtendedCardInfo, payment_id))]
pub async fn retrieve_extended_card_info(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::GetExtendedCardInfo;
    let payment_id = path.into_inner();

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payment_id,
        |state, auth: auth::AuthenticationData, payment_id, _| {
            payments::get_extended_card_info(
                state,
                auth.platform
                    .get_provider()
                    .get_account()
                    .get_id()
                    .to_owned(),
                payment_id,
            )
        },
        &auth::HeaderAuth(auth::ApiKeyAuth {
            allow_connected_scope_operation: true,
            allow_platform_self_operation: false,
        }),
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(all(feature = "oltp", feature = "v1"))]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsSubmitCheckEligibility, payment_id))]
pub async fn payments_submit_eligibility_check(
    state: web::Data<app::AppState>,
    http_req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsEligibilityCheckRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsSubmitCheckEligibility;
    let payment_id = path.into_inner();
    let mut payload = json_payload.into_inner();
    payload.payment_id = payment_id.clone();

    let api_auth = auth::ApiKeyAuth {
        allow_connected_scope_operation: true,
        allow_platform_self_operation: false,
    };

    let (auth_type, _auth_flow) =
        match auth::check_sdk_auth_and_get_auth(http_req.headers(), &payload, api_auth) {
            Ok(auth) => auth,
            Err(err) => return api::log_and_return_error_response(report!(err)),
        };

    Box::pin(api::server_wrap(
        flow,
        state,
        &http_req,
        payment_id,
        |state, auth: auth::AuthenticationData, payment_id, _| {
            let mut payload = payload.clone();
            if let Some(client_secret) = auth.client_secret {
                payload.client_secret = Some(Secret::new(client_secret));
            }
            payments::payments_submit_eligibility_check(
                state,
                auth.platform,
                payload.clone(),
                payment_id,
            )
        },
        &*auth_type,
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(all(feature = "oltp", feature = "v1"))]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsSubmitEligibility, payment_id))]
pub async fn payments_submit_eligibility(
    state: web::Data<app::AppState>,
    http_req: actix_web::HttpRequest,
    json_payload: web::Json<payment_types::PaymentsEligibilityRequest>,
    path: web::Path<common_utils::id_type::PaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentsSubmitEligibility;
    let payment_id = path.into_inner();
    let mut payload = json_payload.into_inner();
    payload.payment_id = payment_id.clone();

    let api_auth = auth::ApiKeyAuth {
        allow_connected_scope_operation: true,
        allow_platform_self_operation: false,
    };

    let (auth_type, _auth_flow) =
        match auth::check_sdk_auth_and_get_auth(http_req.headers(), &payload, api_auth) {
            Ok(auth) => auth,
            Err(err) => return api::log_and_return_error_response(report!(err)),
        };

    Box::pin(api::server_wrap(
        flow,
        state,
        &http_req,
        payment_id,
        |state, auth: auth::AuthenticationData, payment_id, _| {
            let mut payload = payload.clone();
            if let Some(client_secret) = auth.client_secret {
                payload.client_secret = Some(Secret::new(client_secret));
            }
            payments::payments_submit_eligibility(state, auth.platform, payload.clone(), payment_id)
        },
        &*auth_type,
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(feature = "v1")]
pub fn get_or_generate_payment_id(
    payload: &mut payment_types::PaymentsRequest,
) -> errors::RouterResult<()> {
    let given_payment_id = payload
        .payment_id
        .clone()
        .map(|payment_id| {
            payment_id
                .get_payment_intent_id()
                .map_err(|err| err.change_context(errors::ApiErrorResponse::PaymentNotFound))
        })
        .transpose()?;

    let payment_id = given_payment_id.unwrap_or(common_utils::id_type::PaymentId::default());

    payload.is_payment_id_from_merchant = matches!(
        &payload.payment_id,
        Some(payment_types::PaymentIdType::PaymentIntentId(_))
    );

    payload.payment_id = Some(api_models::payments::PaymentIdType::PaymentIntentId(
        payment_id,
    ));

    Ok(())
}

#[cfg(feature = "v1")]
impl GetLockingInput for payment_types::PaymentsRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        match self.payment_id {
            Some(payment_types::PaymentIdType::PaymentIntentId(ref id)) => {
                let api_identifier = lock_utils::ApiIdentifier::from(flow);
                let intent_id_locking_input = api_locking::LockingInput {
                    unique_locking_key: id.get_string_repr().to_owned(),
                    api_identifier: api_identifier.clone(),
                    override_lock_retries: None,
                };
                if let Some(customer_id) = self.customer_id.as_ref().or_else(|| {
                    self.customer
                        .as_ref()
                        .and_then(|customer| customer.id.as_ref())
                }) {
                    api_locking::LockAction::HoldMultiple {
                        inputs: vec![
                            intent_id_locking_input,
                            api_locking::LockingInput {
                                unique_locking_key: customer_id.get_string_repr().to_owned(),
                                api_identifier,
                                override_lock_retries: None,
                            },
                        ],
                    }
                } else {
                    api_locking::LockAction::Hold {
                        input: intent_id_locking_input,
                    }
                }
            }
            _ => api_locking::LockAction::NotApplicable,
        }
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for payment_types::PaymentsStartRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for payment_types::PaymentsRetrieveRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        match self.resource_id {
            payment_types::PaymentIdType::PaymentIntentId(ref id) if self.force_sync => {
                api_locking::LockAction::Hold {
                    input: api_locking::LockingInput {
                        unique_locking_key: id.get_string_repr().to_owned(),
                        api_identifier: lock_utils::ApiIdentifier::from(flow),
                        override_lock_retries: None,
                    },
                }
            }
            _ => api_locking::LockAction::NotApplicable,
        }
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for payment_types::PaymentsSessionRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "v2")]
impl GetLockingInput for payment_types::PaymentsSessionRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
    {
        api_locking::LockAction::NotApplicable
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for payment_types::PaymentsDynamicTaxCalculationRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for payment_types::PaymentsPostSessionTokensRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for payment_types::PaymentsUpdateMetadataRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for payments::PaymentsRedirectResponseData {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        match self.resource_id {
            payment_types::PaymentIdType::PaymentIntentId(ref id) => {
                api_locking::LockAction::Hold {
                    input: api_locking::LockingInput {
                        unique_locking_key: id.get_string_repr().to_owned(),
                        api_identifier: lock_utils::ApiIdentifier::from(flow),
                        override_lock_retries: None,
                    },
                }
            }
            _ => api_locking::LockAction::NotApplicable,
        }
    }
}

#[cfg(feature = "v2")]
impl GetLockingInput for payments::PaymentsRedirectResponseData {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for payment_types::PaymentsCompleteAuthorizeRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for payment_types::PaymentsCancelRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for payment_types::PaymentsCancelPostCaptureRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for common_utils::id_type::PaymentId {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for payment_types::PaymentsExtendAuthorizationRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for payment_types::PaymentsCaptureRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "oltp")]
struct FPaymentsApproveRequest<'a>(&'a payment_types::PaymentsApproveRequest);

#[cfg(feature = "oltp")]
impl GetLockingInput for FPaymentsApproveRequest<'_> {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.0.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "oltp")]
struct FPaymentsRejectRequest<'a>(&'a payment_types::PaymentsRejectRequest);

#[cfg(feature = "oltp")]
impl GetLockingInput for FPaymentsRejectRequest<'_> {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.0.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for payment_types::PaymentsIncrementalAuthorizationRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for payment_types::PaymentsExternalAuthenticationRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[cfg(feature = "v1")]
impl GetLockingInput for payment_types::PaymentsManualUpdateRequest {
    fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
    where
        F: types::FlowMetric,
        lock_utils::ApiIdentifier: From<F>,
    {
        api_locking::LockAction::Hold {
            input: api_locking::LockingInput {
                unique_locking_key: self.payment_id.get_string_repr().to_owned(),
                api_identifier: lock_utils::ApiIdentifier::from(flow),
                override_lock_retries: None,
            },
        }
    }
}

#[instrument(skip_all, fields(flow = ?Flow::PaymentsAggregate))]
#[cfg(all(feature = "olap", feature = "v1"))]
pub async fn get_payments_aggregates_profile(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    payload: web::Query<common_utils::types::TimeRange>,
) -> impl Responder {
    let flow = Flow::PaymentsAggregate;
    let payload = payload.into_inner();
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, _| {
            payments::get_aggregates_for_payments(
                state,
                auth.platform,
                auth.profile.map(|profile| vec![profile.get_id().clone()]),
                req,
            )
        },
        &auth::JWTAuth {
            permission: Permission::ProfilePaymentRead,
            allow_connected: true,
            allow_platform: false,
        },
        api_locking::LockAction::NotApplicable,
    ))
    .await
}
#[instrument(skip_all, fields(flow = ?Flow::PaymentsAggregate))]
#[cfg(all(feature = "olap", feature = "v2"))]
pub async fn get_payments_aggregates_profile(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    payload: web::Query<common_utils::types::TimeRange>,
) -> impl Responder {
    let flow = Flow::PaymentsAggregate;
    let payload = payload.into_inner();
    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth: auth::AuthenticationData, req, _| {
            payments::get_aggregates_for_payments(
                state,
                auth.platform,
                Some(vec![auth.profile.get_id().clone()]),
                req,
            )
        },
        &auth::JWTAuth {
            permission: Permission::ProfilePaymentRead,
            allow_connected: true,
            allow_platform: false,
        },
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(feature = "v2")]
/// A private module to hold internal types to be used in route handlers.
/// This is because we will need to implement certain traits on these types which will have the resource id
/// But the api payload will not contain the resource id
/// So these types can hold the resource id along with actual api payload, on which api event and locking action traits can be implemented
mod internal_payload_types {
    use super::*;

    // Serialize is implemented because of api events
    #[derive(Debug, serde::Serialize)]
    pub struct PaymentsGenericRequestWithResourceId<T: serde::Serialize> {
        pub global_payment_id: common_utils::id_type::GlobalPaymentId,
        #[serde(flatten)]
        pub payload: T,
    }

    impl<T: serde::Serialize> GetLockingInput for PaymentsGenericRequestWithResourceId<T> {
        fn get_locking_input<F>(&self, flow: F) -> api_locking::LockAction
        where
            F: types::FlowMetric,
            lock_utils::ApiIdentifier: From<F>,
        {
            api_locking::LockAction::Hold {
                input: api_locking::LockingInput {
                    unique_locking_key: self.global_payment_id.get_string_repr().to_owned(),
                    api_identifier: lock_utils::ApiIdentifier::from(flow),
                    override_lock_retries: None,
                },
            }
        }
    }

    impl<T: serde::Serialize> common_utils::events::ApiEventMetric
        for PaymentsGenericRequestWithResourceId<T>
    {
        fn get_api_event_type(&self) -> Option<common_utils::events::ApiEventsType> {
            Some(common_utils::events::ApiEventsType::Payment {
                payment_id: self.global_payment_id.clone(),
            })
        }
    }
}

#[cfg(feature = "v2")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentStartRedirection, payment_id))]
pub async fn payments_start_redirection(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    payload: web::Query<api_models::payments::PaymentStartRedirectionParams>,
    path: web::Path<common_utils::id_type::GlobalPaymentId>,
) -> impl Responder {
    let flow = Flow::PaymentStartRedirection;

    let global_payment_id = path.into_inner();
    tracing::Span::current().record("payment_id", global_payment_id.get_string_repr());

    let publishable_key = &payload.publishable_key;
    let profile_id = &payload.profile_id;

    let payment_start_redirection_request = api_models::payments::PaymentStartRedirectionRequest {
        id: global_payment_id.clone(),
    };

    let internal_payload = internal_payload_types::PaymentsGenericRequestWithResourceId {
        global_payment_id: global_payment_id.clone(),
        payload: payment_start_redirection_request.clone(),
    };

    let locking_action = internal_payload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payment_start_redirection_request.clone(),
        |state, auth: auth::AuthenticationData, _req, req_state| async {
            payments::payment_start_redirection(
                state,
                auth.platform,
                payment_start_redirection_request.clone(),
            )
            .await
        },
        &auth::PublishableKeyAndProfileIdAuth {
            publishable_key: publishable_key.clone(),
            profile_id: profile_id.clone(),
        },
        locking_action,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsConfirmIntent, payment_id))]
pub async fn payment_confirm_intent(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<api_models::payments::PaymentsConfirmIntentRequest>,
    path: web::Path<common_utils::id_type::GlobalPaymentId>,
) -> impl Responder {
    use hyperswitch_domain_models::payments::PaymentConfirmData;

    let flow = Flow::PaymentsConfirmIntent;

    // TODO: Populate browser information into the payload
    // if let Err(err) = helpers::populate_ip_into_browser_info(&req, &mut payload) {
    //     return api::log_and_return_error_response(err);
    // }

    let global_payment_id = path.into_inner();
    tracing::Span::current().record("payment_id", global_payment_id.get_string_repr());

    let internal_payload = internal_payload_types::PaymentsGenericRequestWithResourceId {
        global_payment_id: global_payment_id.clone(),
        payload: json_payload.into_inner(),
    };

    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    let locking_action = internal_payload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        internal_payload,
        |state, auth: auth::AuthenticationData, req, req_state| async {
            let payment_id = req.global_payment_id;
            let request = req.payload;

            Box::pin(payments::payments_execute_wrapper(
                state,
                req_state,
                auth.platform,
                auth.profile,
                request,
                header_payload.clone(),
                payment_id,
            ))
            .await
        },
        auth::api_or_client_auth(
            &auth::V2ApiKeyAuth {
                allow_connected_scope_operation: false,
                allow_platform_self_operation: false,
            },
            &auth::V2ClientAuth(common_utils::types::authentication::ResourceId::Payment(
                global_payment_id,
            )),
            req.headers(),
        ),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip_all, fields(flow = ?Flow::ApplyPaymentMethodData, payment_id))]
pub async fn payments_apply_pm_data(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<api_models::payments::ApplyPaymentMethodDataRequest>,
    path: web::Path<common_utils::id_type::GlobalPaymentId>,
) -> impl Responder {
    let flow = Flow::ApplyPaymentMethodData;

    let global_payment_id = path.into_inner();
    tracing::Span::current().record("payment_id", global_payment_id.get_string_repr());

    let internal_payload = internal_payload_types::PaymentsGenericRequestWithResourceId {
        global_payment_id: global_payment_id.clone(),
        payload: json_payload.into_inner(),
    };

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        internal_payload,
        |state, auth: auth::AuthenticationData, req, req_state| async {
            let payment_id = req.global_payment_id;
            let request = req.payload;

            Box::pin(
                payment_method_balance::payments_check_and_apply_pm_data_core(
                    state,
                    auth.platform,
                    auth.profile,
                    req_state,
                    request,
                    payment_id,
                ),
            )
            .await
        },
        auth::api_or_client_auth(
            &auth::V2ApiKeyAuth {
                allow_connected_scope_operation: false,
                allow_platform_self_operation: false,
            },
            &auth::V2ClientAuth(common_utils::types::authentication::ResourceId::Payment(
                global_payment_id,
            )),
            req.headers(),
        ),
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip_all, fields(flow = ?Flow::ProxyConfirmIntent, payment_id))]
pub async fn proxy_confirm_intent(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<api_models::payments::ProxyPaymentsRequest>,
    path: web::Path<common_utils::id_type::GlobalPaymentId>,
) -> impl Responder {
    use hyperswitch_domain_models::payments::PaymentConfirmData;

    let flow = Flow::ProxyConfirmIntent;

    // Extract the payment ID from the path
    let global_payment_id = path.into_inner();
    tracing::Span::current().record("payment_id", global_payment_id.get_string_repr());

    // Parse and validate headers
    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    // Prepare the internal payload
    let internal_payload = internal_payload_types::PaymentsGenericRequestWithResourceId {
        global_payment_id,
        payload: json_payload.into_inner(),
    };

    // Determine the locking action, if required
    let locking_action = internal_payload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        internal_payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            let payment_id = req.global_payment_id;
            let request = req.payload;

            // Define the operation for proxy payments intent
            let operation = payments::operations::proxy_payments_intent::PaymentProxyIntent;

            // Call the core proxy logic
            Box::pin(payments::proxy_for_payments_core::<
                api_types::Authorize,
                api_models::payments::PaymentsResponse,
                _,
                _,
                _,
                PaymentConfirmData<api_types::Authorize>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile,
                operation,
                request,
                payment_id,
                payments::CallConnectorAction::Trigger,
                header_payload.clone(),
                None,
            ))
        },
        &auth::V2ApiKeyAuth {
            allow_connected_scope_operation: false,
            allow_platform_self_operation: false,
        },
        locking_action,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip_all, fields(flow = ?Flow::ProxyConfirmIntent, payment_id))]
pub async fn confirm_intent_with_external_vault_proxy(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: web::Json<api_models::payments::ExternalVaultProxyPaymentsRequest>,
    path: web::Path<common_utils::id_type::GlobalPaymentId>,
) -> impl Responder {
    use hyperswitch_domain_models::payments::PaymentConfirmData;

    let flow = Flow::ProxyConfirmIntent;

    // Extract the payment ID from the path
    let global_payment_id = path.into_inner();
    tracing::Span::current().record("payment_id", global_payment_id.get_string_repr());

    // Parse and validate headers
    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    // Prepare the internal payload
    let internal_payload = internal_payload_types::PaymentsGenericRequestWithResourceId {
        global_payment_id: global_payment_id.clone(),
        payload: json_payload.into_inner(),
    };

    // Determine the locking action, if required
    let locking_action = internal_payload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        internal_payload,
        |state, auth: auth::AuthenticationData, req, req_state| {
            let payment_id = req.global_payment_id;
            let request = req.payload;

            // Define the operation for proxy payments intent
            let operation = payments::operations::external_vault_proxy_payment_intent::ExternalVaultProxyPaymentIntent;

            // Call the core proxy logic
            Box::pin(payments::external_vault_proxy_for_payments_core::<
                api_types::ExternalVaultProxy,
                api_models::payments::PaymentsResponse,
                _,
                _,
                _,
                PaymentConfirmData<api_types::ExternalVaultProxy>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile,
                operation,
                request,
                payment_id,
                payments::CallConnectorAction::Trigger,
                header_payload.clone(),
                None,))
        },
        auth::api_or_client_auth(
            &auth::V2ApiKeyAuth {
                allow_connected_scope_operation: false,
                allow_platform_self_operation: false,
            },
            &auth::V2ClientAuth(common_utils::types::authentication::ResourceId::Payment(
                global_payment_id,
            )),
            req.headers(),
        ),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip(state, req), fields(flow, payment_id))]
pub async fn payment_status(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    payload: web::Query<api_models::payments::PaymentsStatusRequest>,
    path: web::Path<common_utils::id_type::GlobalPaymentId>,
) -> impl Responder {
    use hyperswitch_domain_models::payments::PaymentStatusData;

    let flow = match payload.force_sync {
        true => Flow::PaymentsRetrieveForceSync,
        false => Flow::PaymentsRetrieve,
    };

    let global_payment_id = path.into_inner();
    tracing::Span::current().record("payment_id", global_payment_id.get_string_repr());

    let payload = payment_types::PaymentsRetrieveRequest {
        force_sync: payload.force_sync,
        expand_attempts: payload.expand_attempts,
        param: payload.param.clone(),
        return_raw_connector_response: payload.return_raw_connector_response,
        merchant_connector_details: None,
    };

    let internal_payload = internal_payload_types::PaymentsGenericRequestWithResourceId {
        global_payment_id,
        payload,
    };

    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    let locking_action = internal_payload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        internal_payload,
        |state, auth: auth::AuthenticationData, req, req_state| async {
            let payment_id = req.global_payment_id;
            let request = req.payload;

            let operation = payments::operations::PaymentGet;

            Box::pin(payments::payments_core::<
                api_types::PSync,
                api_models::payments::PaymentsResponse,
                _,
                _,
                _,
                PaymentStatusData<api_types::PSync>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile,
                operation,
                request,
                payment_id,
                payments::CallConnectorAction::Trigger,
                header_payload.clone(),
            ))
            .await
        },
        auth::auth_type(
            &auth::V2ApiKeyAuth {
                allow_connected_scope_operation: false,
                allow_platform_self_operation: false,
            },
            &auth::JWTAuth {
                permission: Permission::ProfilePaymentRead,
                allow_connected: false,
                allow_platform: false,
            },
            req.headers(),
        ),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip(state, req), fields(flow, payment_id))]
pub async fn payments_status_with_gateway_creds(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    payload: web::Json<api_models::payments::PaymentsRetrieveRequest>,
    path: web::Path<common_utils::id_type::GlobalPaymentId>,
) -> impl Responder {
    use hyperswitch_domain_models::payments::PaymentStatusData;

    let flow = match payload.force_sync {
        true => Flow::PaymentsRetrieveForceSync,
        false => Flow::PaymentsRetrieve,
    };
    tracing::Span::current().record("flow", flow.to_string());

    let global_payment_id = path.into_inner();
    tracing::Span::current().record("payment_id", global_payment_id.get_string_repr());

    let internal_payload = internal_payload_types::PaymentsGenericRequestWithResourceId {
        global_payment_id,
        payload: payload.into_inner(),
    };

    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    let locking_action = internal_payload.get_locking_input(flow.clone());

    let auth_type = if state.conf.merchant_id_auth.merchant_id_auth_enabled {
        &auth::MerchantIdAuth
    } else {
        match env::which() {
            env::Env::Production => &auth::V2ApiKeyAuth {
                allow_connected_scope_operation: false,
                allow_platform_self_operation: false,
            },
            _ => auth::auth_type(
                &auth::V2ApiKeyAuth {
                    allow_connected_scope_operation: false,
                    allow_platform_self_operation: false,
                },
                &auth::JWTAuth {
                    permission: Permission::ProfilePaymentWrite,
                    allow_connected: false,
                    allow_platform: false,
                },
                req.headers(),
            ),
        }
    };

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        internal_payload,
        |state, auth: auth::AuthenticationData, req, req_state| async {
            let payment_id = req.global_payment_id;
            let request = req.payload;

            let operation = payments::operations::PaymentGet;

            Box::pin(payments::payments_core::<
                api_types::PSync,
                api_models::payments::PaymentsResponse,
                _,
                _,
                _,
                PaymentStatusData<api_types::PSync>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile,
                operation,
                request,
                payment_id,
                payments::CallConnectorAction::Trigger,
                header_payload.clone(),
            ))
            .await
        },
        auth_type,
        locking_action,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip(state, req), fields(flow, payment_id))]
pub async fn payment_get_intent_using_merchant_reference_id(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    path: web::Path<common_utils::id_type::PaymentReferenceId>,
) -> impl Responder {
    let flow = Flow::PaymentsRetrieveUsingMerchantReferenceId;
    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    let merchant_reference_id = path.into_inner();

    let auth_type: &dyn auth::AuthenticateAndFetch<_, _> =
        if state.conf.merchant_id_auth.merchant_id_auth_enabled {
            &auth::MerchantIdAuth
        } else {
            &auth::V2ApiKeyAuth {
                allow_connected_scope_operation: false,
                allow_platform_self_operation: false,
            }
        };

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        (),
        |state, auth: auth::AuthenticationData, _, req_state| async {
            Box::pin(payments::payments_get_intent_using_merchant_reference(
                state,
                auth.platform,
                auth.profile,
                req_state,
                &merchant_reference_id,
                header_payload.clone(),
            ))
            .await
        },
        auth_type,
        api_locking::LockAction::NotApplicable,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentsRedirect, payment_id))]
pub async fn payments_finish_redirection(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    json_payload: Option<web::Form<serde_json::Value>>,
    path: web::Path<(
        common_utils::id_type::GlobalPaymentId,
        String,
        common_utils::id_type::ProfileId,
    )>,
) -> impl Responder {
    let flow = Flow::PaymentsRedirect;
    let (payment_id, publishable_key, profile_id) = path.into_inner();
    let param_string = req.query_string();

    tracing::Span::current().record("payment_id", payment_id.get_string_repr());

    let payload = payments::PaymentsRedirectResponseData {
        payment_id,
        json_payload: json_payload.map(|payload| payload.0),
        query_params: param_string.to_string(),
    };

    let locking_action = payload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        payload,
        |state, auth, req, req_state| {
            <payments::PaymentRedirectSync as PaymentRedirectFlow>::handle_payments_redirect_response(
                &payments::PaymentRedirectSync {},
                state,
                req_state,
                auth.platform,
                auth.profile,
                req,
            )
        },
        &auth::PublishableKeyAndProfileIdAuth {
            publishable_key: publishable_key.clone(),
            profile_id: profile_id.clone(),
        },
        locking_action,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip(state, req), fields(flow, payment_id))]
pub async fn payments_capture(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    payload: web::Json<api_models::payments::PaymentsCaptureRequest>,
    path: web::Path<common_utils::id_type::GlobalPaymentId>,
) -> impl Responder {
    use hyperswitch_domain_models::payments::PaymentCaptureData;
    let flow = Flow::PaymentsCapture;

    let global_payment_id = path.into_inner();
    tracing::Span::current().record("payment_id", global_payment_id.get_string_repr());

    let internal_payload = internal_payload_types::PaymentsGenericRequestWithResourceId {
        global_payment_id,
        payload: payload.into_inner(),
    };

    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    let locking_action = internal_payload.get_locking_input(flow.clone());

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        internal_payload,
        |state, auth: auth::AuthenticationData, req, req_state| async {
            let payment_id = req.global_payment_id;
            let request = req.payload;

            let operation = payments::operations::payment_capture_v2::PaymentsCapture;

            Box::pin(payments::payments_core::<
                api_types::Capture,
                api_models::payments::PaymentsCaptureResponse,
                _,
                _,
                _,
                PaymentCaptureData<api_types::Capture>,
            >(
                state,
                req_state,
                auth.platform,
                auth.profile,
                operation,
                request,
                payment_id,
                payments::CallConnectorAction::Trigger,
                header_payload.clone(),
            ))
            .await
        },
        auth::auth_type(
            &auth::V2ApiKeyAuth {
                allow_connected_scope_operation: false,
                allow_platform_self_operation: false,
            },
            &auth::JWTAuth {
                permission: Permission::ProfileAccountWrite,
                allow_connected: false,
                allow_platform: false,
            },
            req.headers(),
        ),
        locking_action,
    ))
    .await
}

#[cfg(feature = "v2")]
#[instrument(skip_all, fields(flow = ?Flow::PaymentMethodsList))]
pub async fn list_payment_methods(
    state: web::Data<app::AppState>,
    req: actix_web::HttpRequest,
    path: web::Path<common_utils::id_type::GlobalPaymentId>,
    query_payload: web::Query<api_models::payments::ListMethodsForPaymentsRequest>,
) -> impl Responder {
    let flow = Flow::PaymentMethodsList;
    let payload = query_payload.into_inner();
    let global_payment_id = path.into_inner();

    tracing::Span::current().record("payment_id", global_payment_id.get_string_repr());

    let internal_payload = internal_payload_types::PaymentsGenericRequestWithResourceId {
        global_payment_id: global_payment_id.clone(),
        payload,
    };

    let header_payload = match HeaderPayload::foreign_try_from(req.headers()) {
        Ok(headers) => headers,
        Err(err) => {
            return api::log_and_return_error_response(err);
        }
    };

    Box::pin(api::server_wrap(
        flow,
        state,
        &req,
        internal_payload,
        |state, auth, req, _| {
            payments::payment_methods::list_payment_methods(
                state,
                auth.platform,
                auth.profile,
                req.global_payment_id,
                req.payload,
                &header_payload,
            )
        },
        &auth::V2ClientAuth(common_utils::types::authentication::ResourceId::Payment(
            global_payment_id,
        )),
        api_locking::LockAction::NotApplicable,
    ))
    .await
}
