use std::{borrow::Cow, str::FromStr, time::Instant};

use actix_web::ResponseError;
use api_models::admin;
use base64::Engine;
use common_enums::{
    connector_enums::Connector, AttemptStatus, CallConnectorAction, ConnectorIntegrationType,
    ExecutionMode, ExecutionPath, GatewaySystem, PaymentMethodType, UcsAvailability,
};
use common_utils::{
    consts::BASE64_ENGINE,
    errors::{CustomResult, ErrorSwitch},
    ext_traits::ValueExt,
    id_type,
    request::Method,
    ucs_types,
};
use diesel_models::types::FeatureMetadata;
use error_stack::ResultExt;
use external_services::grpc_client::{
    unified_connector_service::{ConnectorAuthMetadata, UnifiedConnectorServiceError},
    LineageIds,
};
use hyperswitch_connectors::utils::CardData;
#[cfg(feature = "v2")]
use hyperswitch_domain_models::merchant_connector_account::MerchantConnectorAccountTypeDetails;
use hyperswitch_domain_models::{
    merchant_connector_account::ExternalVaultConnectorMetadata,
    platform::Processor,
    router_data::{AccessToken, ConnectorAuthType, ErrorResponse, PaymentMethodToken, RouterData},
    router_flow_types::refunds,
    router_request_types::RefundsData,
    router_response_types::{PaymentsResponseData, PayoutsResponseData, RefundsResponseData},
};
use hyperswitch_masking::{ExposeInterface, PeekInterface, Secret};
use router_env::{instrument, logger, tracing};
use unified_connector_service_cards::CardNumber;
use unified_connector_service_client::payments::{
    self as payments_grpc, payment_method::PaymentMethod, CardDetails, CardDetailsWithNoCvc,
    ClassicReward, CryptoCurrency, EVoucher, OpenBanking, PaymentServiceAuthorizeResponse,
};

use crate::{
    consts,
    core::{
        errors::{self, RouterResult},
        payments::{
            helpers::{
                is_googlepay_predecrypted_flow_supported, is_ucs_enabled,
                should_execute_based_on_rollout, should_execute_based_on_rollout_with_precedence,
                MerchantConnectorAccountType, ProxyOverride, WebhookRolloutConfig,
                WebhookRolloutExecutionResult,
            },
            OperationSessionGetters, OperationSessionSetters,
        },
        utils::get_flow_name,
        webhooks::types::FeatureTrackingData,
    },
    events::connector_api_logs::ConnectorEvent,
    routes::SessionState,
    types::{
        api::enums as api_enums,
        domain::Event,
        transformers::{ForeignFrom, ForeignTryFrom},
        UcsPaymentAuthorizeResponseData, UcsPaymentCaptureResponseData,
        UcsPaymentSetupRecurringResponseData, UcsRecurringPaymentChargeResponseData,
    },
};

pub mod connector_config;
pub mod transformers;

/// Returns Apple Pay data from payment method token when it has decrypt data,
/// otherwise returns the original payment data.
fn get_apple_pay_payment_data(
    payment_data: &common_types::payments::ApplePayPaymentData,
    payment_method_token: Option<&PaymentMethodToken>,
) -> common_types::payments::ApplePayPaymentData {
    match (payment_data, payment_method_token) {
        (
            common_types::payments::ApplePayPaymentData::Encrypted(_),
            Some(PaymentMethodToken::ApplePayDecrypt(apple_pay_decrypt_data)),
        ) => common_types::payments::ApplePayPaymentData::Decrypted(
            apple_pay_decrypt_data.as_ref().clone(),
        ),
        // Fallback: keep original payload for non-matching token cases.
        _ => payment_data.clone(),
    }
}

fn get_google_pay_tokenization_data(
    tokenization_data: &common_types::payments::GpayTokenizationData,
    payment_method_token: Option<&PaymentMethodToken>,
    connector_meta_data: Option<&common_utils::pii::SecretSerdeValue>,
) -> common_types::payments::GpayTokenizationData {
    if !is_googlepay_predecrypted_flow_supported(connector_meta_data.cloned()) {
        return tokenization_data.clone();
    }
    match (tokenization_data, payment_method_token) {
        (
            common_types::payments::GpayTokenizationData::Encrypted(_),
            Some(PaymentMethodToken::GooglePayDecrypt(google_pay_decrypt_data)),
        ) => common_types::payments::GpayTokenizationData::Decrypted(
            google_pay_decrypt_data.as_ref().clone(),
        ),
        // Fallback: keep original payload for non-matching token cases.
        _ => tokenization_data.clone(),
    }
}

/// Returns Paze data from payment method token when it has decrypt data,
/// otherwise returns data from wallet payload.
fn get_paze_wallet_data(
    wallet_data: &hyperswitch_domain_models::payment_method_data::PazeWalletData,
    payment_method_token: Option<&PaymentMethodToken>,
) -> CustomResult<payments_grpc::paze_wallet::PazeData, UnifiedConnectorServiceError> {
    match payment_method_token {
        Some(PaymentMethodToken::PazeDecrypt(paze_decrypted_data)) => {
            Ok(payments_grpc::paze_wallet::PazeData::DecryptedData(
                payments_grpc::PazeDecryptedData::foreign_try_from(paze_decrypted_data.as_ref())?,
            ))
        }
        _ => payments_grpc::paze_wallet::PazeData::foreign_try_from(wallet_data),
    }
}

pub async fn get_access_token_from_ucs_response(
    session_state: &SessionState,
    processor: &Processor,
    connector_name: &str,
    merchant_connector_id: Option<&id_type::MerchantConnectorAccountId>,
    creds_identifier: Option<String>,
    ucs_state: Option<&unified_connector_service_client::payments::ConnectorState>,
) -> CustomResult<Option<AccessToken>, UnifiedConnectorServiceError> {
    let ucs_access_token = ucs_state
        .and_then(|state| state.access_token.as_ref())
        .map(AccessToken::foreign_try_from)
        .transpose()
        .inspect_err(|err| {
            logger::warn!(
                error = ?err,
                "Invalid access token received from UCS connector state"
            )
        })?;

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

    let merchant_connector_id_or_connector_name = merchant_connector_id
        .map(|mca_id| mca_id.get_string_repr().to_string())
        .or(creds_identifier.map(|id| id.to_string()))
        .unwrap_or(connector_name.to_string());

    let key = common_utils::access_token::get_default_access_token_key(
        merchant_id,
        merchant_connector_id_or_connector_name,
    );

    if let Ok(Some(cached_token)) = session_state.store.get_access_token(key).await {
        if let Some(ref ucs_token) = ucs_access_token {
            if cached_token.token.peek() == ucs_token.token.peek() {
                return Ok(None);
            }
        }
    }

    Ok(ucs_access_token)
}

pub async fn set_access_token_for_ucs(
    state: &SessionState,
    processor: &Processor,
    connector_name: &str,
    access_token: AccessToken,
    merchant_connector_id: Option<&id_type::MerchantConnectorAccountId>,
    creds_identifier: Option<String>,
) -> Result<(), errors::StorageError> {
    let merchant_id = processor.get_account().get_id();

    let merchant_connector_id_or_connector_name = merchant_connector_id
        .map(|mca_id| mca_id.get_string_repr().to_string())
        .or(creds_identifier.map(|id| id.to_string()))
        .unwrap_or(connector_name.to_string());

    let key = common_utils::access_token::get_default_access_token_key(
        merchant_id,
        &merchant_connector_id_or_connector_name,
    );

    let modified_access_token = AccessToken {
        expires: access_token
            .expires
            .saturating_sub(consts::REDUCE_ACCESS_TOKEN_EXPIRY_TIME.into()),
        ..access_token
    };

    logger::debug!(
        access_token_expiry_after_modification = modified_access_token.expires,
        merchant_id = ?merchant_id,
        connector_name = connector_name,
        merchant_connector_id_or_connector_name = merchant_connector_id_or_connector_name
    );

    if let Err(access_token_set_error) = state
        .store
        .set_access_token(key, modified_access_token)
        .await
    {
        // If we are not able to set the access token in redis, the error should just be logged and proceed with the payment
        // Payments should not fail, once the access token is successfully created
        // The next request will create new access token, if required
        logger::error!(access_token_set_error=?access_token_set_error, "Failed to store UCS access token");
    }

    Ok(())
}

/// Type alias for return type used by unified connector service response handlers
type UnifiedConnectorServiceResult = CustomResult<
    (
        Result<(PaymentsResponseData, AttemptStatus), ErrorResponse>,
        u16,
    ),
    UnifiedConnectorServiceError,
>;

/// Type alias for return type used by unified connector service refund response handlers
type UnifiedConnectorServiceRefundResult =
    CustomResult<(Result<RefundsResponseData, ErrorResponse>, u16), UnifiedConnectorServiceError>;

type UnifiedConnectorServiceCreateOrderResult = CustomResult<
    (
        Result<(PaymentsResponseData, AttemptStatus), ErrorResponse>,
        u16,
    ),
    UnifiedConnectorServiceError,
>;

/// Checks if the Unified Connector Service (UCS) is available for use
async fn check_ucs_availability(state: &SessionState) -> UcsAvailability {
    let is_client_available = state.grpc_client.unified_connector_service_client.is_some();

    let is_enabled = is_ucs_enabled(state, consts::UCS_ENABLED).await;

    match (is_client_available, is_enabled) {
        (true, true) => {
            router_env::logger::debug!("UCS is available and enabled");
            UcsAvailability::Enabled
        }
        _ => {
            router_env::logger::debug!(
                "UCS client is {} and UCS is {} in configuration",
                if is_client_available {
                    "available"
                } else {
                    "not available"
                },
                if is_enabled { "enabled" } else { "not enabled" }
            );
            UcsAvailability::Disabled
        }
    }
}

/// Determines the connector integration type based on UCS configuration or on both
pub async fn determine_connector_integration_type(
    state: &SessionState,
    connector: Connector,
) -> RouterResult<ConnectorIntegrationType> {
    match state.conf.grpc_client.unified_connector_service.as_ref() {
        Some(ucs_config) => {
            let is_ucs_only = ucs_config.ucs_only_connectors.contains(&connector);

            if is_ucs_only {
                router_env::logger::debug!(
                    connector = ?connector,
                    ucs_only_list = is_ucs_only,
                    "Using UcsConnector"
                );
                Ok(ConnectorIntegrationType::UcsConnector)
            } else {
                router_env::logger::debug!(
                    connector = ?connector,
                    "Using DirectandUCSConnector - not in ucs_only_list"
                );
                Ok(ConnectorIntegrationType::DirectandUCSConnector)
            }
        }
        None => {
            router_env::logger::debug!(
                connector = ?connector,
                "UCS config not present, using DirectandUCSConnector"
            );
            Ok(ConnectorIntegrationType::DirectandUCSConnector)
        }
    }
}

pub async fn should_call_unified_connector_service<F: Clone, T, R>(
    state: &SessionState,
    processor: &Processor,
    router_data: &RouterData<F, T, R>,
    previous_gateway: Option<GatewaySystem>,
    call_connector_action: CallConnectorAction,
    shadow_ucs_call_connector_action: Option<CallConnectorAction>,
    transaction_type: common_enums::TransactionType,
) -> RouterResult<(ExecutionPath, SessionState)>
where
    R: Send + Sync + Clone,
{
    // Extract context information
    let merchant_id = processor.get_account().get_id().get_string_repr();
    let org_id = processor.get_account().get_org_id().get_string_repr();

    let connector_name = &router_data.connector;
    let connector_enum = Connector::from_str(connector_name)
        .change_context(errors::ApiErrorResponse::IncorrectConnectorNameGiven)
        .attach_printable_lazy(|| format!("Failed to parse connector name: {connector_name}"))?;

    let flow_name = get_flow_name::<F>()?;

    // Check UCS availability using idiomatic helper
    let ucs_availability = check_ucs_availability(state).await;

    // Build rollout keys in ascending precedence order.
    // Payments use org-level hierarchy; payouts use a single merchant-level key for now.
    let rollout_keys = match transaction_type {
        common_enums::TransactionType::Payment
        | common_enums::TransactionType::ThreeDsAuthentication => build_rollout_keys(
            org_id,
            merchant_id,
            connector_name,
            &flow_name,
            router_data.payment_method,
            router_data.payment_method_type,
        ),
        common_enums::TransactionType::Payout => build_rollout_keys_for_payouts(
            merchant_id,
            connector_name,
            &flow_name,
            router_data.payment_method_type,
        ),
    };

    // Determine connector integration type
    let connector_integration_type =
        determine_connector_integration_type(state, connector_enum).await?;

    // Try keys highest → lowest precedence, use first match found
    let rollout_result =
        should_execute_based_on_rollout_with_precedence(state, &rollout_keys).await?;

    // Single decision point using pattern matching
    let (gateway_system, mut execution_path) = if ucs_availability == UcsAvailability::Disabled {
        match call_connector_action {
            CallConnectorAction::UCSConsumeResponse(_) => {
                Err(errors::ApiErrorResponse::InternalServerError)
                    .attach_printable("CallConnectorAction UCSConsumeResponse received but UCS is disabled. This action is only valid in UCS gateway")?
            }
            CallConnectorAction::Avoid
            | CallConnectorAction::Trigger
            | CallConnectorAction::HandleResponse { .. }
            | CallConnectorAction::HandleResponseWithoutBuildRequest
            | CallConnectorAction::StatusUpdate { .. } => {
                router_env::logger::debug!("UCS is disabled, using Direct gateway");
                (GatewaySystem::Direct, ExecutionPath::Direct)
            }
        }
    } else {
        match call_connector_action {
            CallConnectorAction::UCSConsumeResponse(_) => {
                router_env::logger::info!(
                    "CallConnectorAction UCSConsumeResponse received, using UCS gateway"
                );
                (
                    GatewaySystem::UnifiedConnectorService,
                    ExecutionPath::UnifiedConnectorService,
                )
            }
            CallConnectorAction::HandleResponse { .. } => {
                router_env::logger::info!(
                    "CallConnectorAction HandleResponse received, using Direct gateway"
                );
                if shadow_ucs_call_connector_action.is_some() {
                    (
                        GatewaySystem::Direct,
                        ExecutionPath::ShadowUnifiedConnectorService,
                    )
                } else {
                    (GatewaySystem::Direct, ExecutionPath::Direct)
                }
            }
            CallConnectorAction::Trigger
            | CallConnectorAction::HandleResponseWithoutBuildRequest
            | CallConnectorAction::Avoid
            | CallConnectorAction::StatusUpdate { .. } => {
                // If a rollout config key exists use its execution mode,
                // otherwise default to Shadow so all traffic mirrors through UCS.
                let execution_mode = if rollout_result.should_execute {
                    rollout_result.execution_mode
                } else {
                    ExecutionMode::Shadow
                };
                decide_execution_path(connector_integration_type, previous_gateway, execution_mode)?
            }
        }
    };

    // Handle proxy configuration for Shadow UCS flows
    let session_state = match execution_path {
        ExecutionPath::ShadowUnifiedConnectorService => {
            // For shadow UCS, use rollout_result for proxy configuration since it takes priority
            match &rollout_result.proxy_override {
                Some(proxy_override) => {
                    router_env::logger::debug!(
                        proxy_override = ?proxy_override,
                        "Creating updated session state with proxy configuration for Shadow UCS"
                    );
                    create_updated_session_state_with_proxy(state.clone(), proxy_override)
                }
                None => {
                    router_env::logger::debug!(
                        "No proxy override available for Shadow UCS, Using the Original State and Sending Request Directly"
                    );
                    execution_path = ExecutionPath::Direct;
                    state.clone()
                }
            }
        }
        ExecutionPath::Direct | ExecutionPath::UnifiedConnectorService => {
            // For Direct and UCS flows, use original state
            state.clone()
        }
    };

    router_env::logger::info!(
        "Payment gateway decision: gateway={:?}, execution_path={:?} - merchant_id={}, connector={}, flow={}",
        gateway_system,
        execution_path,
        merchant_id,
        connector_name,
        flow_name
    );

    Ok((execution_path, session_state))
}

/// Creates a new SessionState with proxy configuration updated from the override
fn create_updated_session_state_with_proxy(
    state: SessionState,
    proxy_override: &ProxyOverride,
) -> SessionState {
    let mut updated_state = state;

    // Create updated configuration with proxy overrides
    let mut updated_conf = (*updated_state.conf).clone();

    // Update proxy URLs with overrides, falling back to existing values
    if let Some(ref http_url) = proxy_override.http_url {
        updated_conf.proxy.http_url = Some(http_url.clone());
    }
    if let Some(ref https_url) = proxy_override.https_url {
        updated_conf.proxy.https_url = Some(https_url.clone());
    }

    updated_state.conf = std::sync::Arc::new(updated_conf);

    updated_state
}

fn decide_execution_path(
    connector_type: ConnectorIntegrationType,
    previous_gateway: Option<GatewaySystem>,
    execution_mode: ExecutionMode,
) -> RouterResult<(GatewaySystem, ExecutionPath)> {
    match connector_type {
        // UCS-only connectors always use UCS
        ConnectorIntegrationType::UcsConnector => Ok((
            GatewaySystem::UnifiedConnectorService,
            ExecutionPath::UnifiedConnectorService,
        )),
        ConnectorIntegrationType::DirectandUCSConnector => {
            match (previous_gateway, execution_mode) {
                (Some(GatewaySystem::Direct), ExecutionMode::NotApplicable) => {
                    // Previous gateway was Direct, continue using Direct
                    Ok((GatewaySystem::Direct, ExecutionPath::Direct))
                }
                (Some(GatewaySystem::Direct), ExecutionMode::Primary) => {
                    // Previous gateway was Direct, continue using Direct
                    Ok((GatewaySystem::Direct, ExecutionPath::Direct))
                }
                (Some(GatewaySystem::Direct), ExecutionMode::Shadow) => {
                    // Previous gateway was Direct, but now UCS is in shadow mode for comparison
                    Ok((
                        GatewaySystem::Direct,
                        ExecutionPath::ShadowUnifiedConnectorService,
                    ))
                }
                (Some(GatewaySystem::UnifiedConnectorService), ExecutionMode::NotApplicable) => {
                    // Previous gateway was UCS, continue using Direct as the config key has notapplicable execution mode
                    Ok((GatewaySystem::Direct, ExecutionPath::Direct))
                }
                (Some(GatewaySystem::UnifiedConnectorService), ExecutionMode::Primary) => {
                    // previous gateway was UCS, and config key has execution mode primary - continue using UCS
                    Ok((
                        GatewaySystem::UnifiedConnectorService,
                        ExecutionPath::UnifiedConnectorService,
                    ))
                }
                (Some(GatewaySystem::UnifiedConnectorService), ExecutionMode::Shadow) => {
                    // previous gateway was UCS, but now UCS is in shadow mode for comparison
                    Ok((
                        GatewaySystem::Direct,
                        ExecutionPath::ShadowUnifiedConnectorService,
                    ))
                }
                (None, ExecutionMode::Primary) => {
                    // Fresh payment for a UCS-enabled connector - use UCS as primary
                    Ok((
                        GatewaySystem::UnifiedConnectorService,
                        ExecutionPath::UnifiedConnectorService,
                    ))
                }
                (None, ExecutionMode::Shadow) => {
                    // Fresh payment for UCS-enabled connector with shadow mode - use shadow UCS
                    Ok((
                        GatewaySystem::Direct,
                        ExecutionPath::ShadowUnifiedConnectorService,
                    ))
                }
                (None, ExecutionMode::NotApplicable) => {
                    // Fresh payment request for direct connector - use direct gateway
                    Ok((GatewaySystem::Direct, ExecutionPath::Direct))
                }
            }
        }
    }
}

/// Builds rollout config keys in ascending precedence order:
/// 1. `ucs_rollout_config_<org_id>`                               — org level (lowest)
/// 2. `ucs_rollout_config_<org_id>_<merchant_id>`                 — org + merchant
/// 3. `ucs_rollout_config_<org_id>_<merchant_id>_<connector>_...` — org + merchant + connector (highest)
///
/// The caller tries keys highest → lowest and uses the first match found.
/// Builds rollout config keys in ascending precedence order (lowest → highest):
/// 1. `ucs_rollout_config_<org_id>`                                         — org level
/// 2. `ucs_rollout_config_<org_id>_<merchant_id>`                           — org + merchant
/// 3. `ucs_rollout_config_<org_id>_<merchant_id>_<connector>_...`           — org + merchant + connector
/// 4. `ucs_rollout_config_<merchant_id>_<connector>_...`                    — merchant + connector (highest)
///
/// The caller iterates highest → lowest and uses the first match found.
fn build_rollout_keys(
    org_id: &str,
    merchant_id: &str,
    connector_name: &str,
    flow_name: &str,
    payment_method: common_enums::PaymentMethod,
    payment_method_type: Option<PaymentMethodType>,
) -> Vec<String> {
    let prefix = consts::UCS_ROLLOUT_PERCENT_CONFIG_PREFIX;
    let is_refund_flow = matches!(flow_name, "Execute" | "RSync");

    let (merchant_connector_key, org_merchant_connector_key) = if is_refund_flow {
        // Refund flows: ucs_rollout_config_<merchant_id>_<connector>_<flow>
        (
            format!("{prefix}_{merchant_id}_{connector_name}_{flow_name}"),
            format!("{prefix}_{org_id}_{merchant_id}_{connector_name}_{flow_name}"),
        )
    } else {
        match payment_method {
            common_enums::PaymentMethod::Wallet
            | common_enums::PaymentMethod::BankRedirect
            | common_enums::PaymentMethod::Voucher
            | common_enums::PaymentMethod::PayLater => {
                let payment_method_str = payment_method.to_string();
                let payment_method_type_str = payment_method_type
                    .map(|pmt| pmt.to_string())
                    .unwrap_or_else(|| "unknown".to_string());
                (
                    format!("{prefix}_{merchant_id}_{connector_name}_{payment_method_str}_{payment_method_type_str}_{flow_name}"),
                    format!("{prefix}_{org_id}_{merchant_id}_{connector_name}_{payment_method_str}_{payment_method_type_str}_{flow_name}"),
                )
            }
            common_enums::PaymentMethod::Card
            | common_enums::PaymentMethod::CardRedirect
            | common_enums::PaymentMethod::Upi
            | common_enums::PaymentMethod::Crypto
            | common_enums::PaymentMethod::Reward
            | common_enums::PaymentMethod::BankDebit
            | common_enums::PaymentMethod::RealTimePayment
            | common_enums::PaymentMethod::BankTransfer
            | common_enums::PaymentMethod::GiftCard
            | common_enums::PaymentMethod::MobilePayment
            | common_enums::PaymentMethod::NetworkToken
            | common_enums::PaymentMethod::OpenBanking => {
                // For other payment methods, use a generic format without specific payment method type details
                let payment_method_str = payment_method.to_string();
                (
                    format!("{prefix}_{merchant_id}_{connector_name}_{payment_method_str}_{flow_name}"),
                    format!("{prefix}_{org_id}_{merchant_id}_{connector_name}_{payment_method_str}_{flow_name}"),
                )
            }
        }
    };

    // Ascending precedence order (lowest first, highest last)
    vec![
        format!("{prefix}_{org_id}"),
        format!("{prefix}_{org_id}_{merchant_id}"),
        org_merchant_connector_key,
        merchant_connector_key,
    ]
}

/// Build rollout key for payouts — single key, org-level precedence not applied to payouts yet.
fn build_rollout_keys_for_payouts(
    merchant_id: &str,
    connector_name: &str,
    flow_name: &str,
    payment_method_type: Option<PaymentMethodType>,
) -> Vec<String> {
    let pmt = payment_method_type
        .map(|t| t.to_string())
        .unwrap_or_else(|| "unknown".to_string());

    vec![format!(
        "{}_{}_{}_{}_{}",
        consts::UCS_ROLLOUT_PERCENT_CONFIG_PREFIX,
        merchant_id,
        connector_name,
        pmt,
        flow_name
    )]
}

/// Extracts the gateway system from the payment intent's feature metadata
/// Returns None if metadata is missing, corrupted, or doesn't contain gateway_system
pub fn extract_gateway_system_from_payment_intent<F: Clone, D>(
    payment_data: &D,
) -> Option<GatewaySystem>
where
    D: OperationSessionGetters<F>,
{
    #[cfg(feature = "v1")]
    {
        payment_data
            .get_payment_intent()
            .feature_metadata
            .as_ref()
            .and_then(|metadata| {
                // Try to parse the JSON value as FeatureMetadata
                // Log errors but don't fail the flow for corrupted metadata
                match serde_json::from_value::<FeatureMetadata>(metadata.clone().expose()) {
                    Ok(feature_metadata) => feature_metadata.gateway_system,
                    Err(err) => {
                        router_env::logger::warn!(
                            "Failed to parse feature_metadata for gateway_system extraction: {}",
                            err
                        );
                        None
                    }
                }
            })
    }
    #[cfg(feature = "v2")]
    {
        None // V2 does not use feature metadata for gateway system tracking
    }
}

/// Updates the payment intent's feature metadata to track the gateway system being used
#[cfg(feature = "v1")]
pub fn update_gateway_system_in_feature_metadata<F: Clone, D>(
    payment_data: &mut D,
    gateway_system: GatewaySystem,
) -> RouterResult<()>
where
    D: OperationSessionGetters<F> + OperationSessionSetters<F>,
{
    let mut payment_intent = payment_data.get_payment_intent().clone();

    let existing_metadata = payment_intent.feature_metadata.as_ref();

    let mut feature_metadata = match existing_metadata {
        Some(metadata) => serde_json::from_value::<FeatureMetadata>(metadata.clone().expose())
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable(
                "Failed to deserialize existing feature metadata while updating gateway system",
            )?,
        None => FeatureMetadata::default(),
    };

    feature_metadata.gateway_system = Some(gateway_system);

    let updated_metadata = serde_json::to_value(feature_metadata)
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to serialize feature metadata")?;

    payment_intent.feature_metadata = Some(Secret::new(updated_metadata));
    payment_data.set_payment_intent(payment_intent);

    Ok(())
}

pub async fn should_call_unified_connector_service_for_webhooks(
    state: &SessionState,
    processor: &Processor,
    connector_name: &str,
    connector_integration_type: ConnectorIntegrationType,
) -> RouterResult<(ExecutionPath, Vec<api_models::webhooks::WebhookFlow>)> {
    // Extract context information
    let merchant_id = processor.get_account().get_id().get_string_repr();

    let flow_name = "Webhooks";

    // Check UCS availability using idiomatic helper
    let ucs_availability = check_ucs_availability(state).await;

    let rollout_key = format!(
        "{}_{}_{}_{}",
        consts::UCS_ROLLOUT_PERCENT_CONFIG_PREFIX,
        merchant_id,
        connector_name,
        flow_name
    );

    // For webhooks, there is no previous gateway system to consider (webhooks are stateless)
    let previous_gateway = None;

    let rollout_result = should_execute_based_on_rollout::<
        WebhookRolloutConfig,
        WebhookRolloutExecutionResult,
    >(state, &rollout_key)
    .await?;

    // Use the same decision logic as payments, with no call_connector_action to consider
    let (gateway_system, execution_path) = if ucs_availability == UcsAvailability::Disabled {
        router_env::logger::debug!("UCS is disabled for webhooks, using Direct gateway");
        (GatewaySystem::Direct, ExecutionPath::Direct)
    } else {
        // UCS is enabled, use decide function with no previous gateway for webhooks
        decide_execution_path(
            connector_integration_type,
            previous_gateway,
            rollout_result.rollout_execution_result.execution_mode,
        )?
    };

    router_env::logger::info!(
        "Webhook gateway decision: gateway={:?}, execution_path={:?} - merchant_id={}, connector={}, flow={}",
        gateway_system,
        execution_path,
        merchant_id,
        connector_name,
        flow_name,
    );

    Ok((execution_path, rollout_result.webhook_flows))
}

/// On a redirect-complete callback the inbound request carries only redirect params, so
/// `payment_method_data` is `None`. UCS still requires `payment_method` on the authorize call,
/// so rebuild a minimal one from the attempt's payment method type.
pub fn reconstruct_payment_method_data_for_redirect_completion(
    payment_method_type: Option<PaymentMethodType>,
    email: Option<common_utils::pii::Email>,
) -> Option<hyperswitch_domain_models::payment_method_data::PaymentMethodData> {
    match payment_method_type {
        Some(PaymentMethodType::Paypal) => Some(
            hyperswitch_domain_models::payment_method_data::PaymentMethodData::Wallet(
                hyperswitch_domain_models::payment_method_data::WalletData::PaypalRedirect(
                    hyperswitch_domain_models::payment_method_data::PaypalRedirection { email },
                ),
            ),
        ),
        _ => None,
    }
}

pub fn build_unified_connector_service_payment_method(
    payment_method_data: hyperswitch_domain_models::payment_method_data::PaymentMethodData,
    payment_method_type: Option<PaymentMethodType>,
    payment_method_token: Option<&PaymentMethodToken>,
    connector_meta_data: Option<&common_utils::pii::SecretSerdeValue>,
) -> CustomResult<payments_grpc::PaymentMethod, UnifiedConnectorServiceError> {
    // A connector tokenization token settles for any payment method, including wallets.
    if let Some(PaymentMethodToken::Token(token)) = payment_method_token {
        return Ok(payments_grpc::PaymentMethod {
            payment_method: Some(PaymentMethod::Token(
                payments_grpc::TokenPaymentMethodType {
                    token: Some(token.clone()),
                },
            )),
        });
    }
    match payment_method_data {
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::Card(card) => {
            let card_exp_month = card
                .get_card_expiry_month_2_digit()
                .attach_printable("Failed to extract 2-digit expiry month from card")
                .change_context(UnifiedConnectorServiceError::InvalidDataFormat {
                    field_name: "card_exp_month",
                })?
                .peek()
                .to_string();

            let card_network = card
                .card_network
                .clone()
                .map(payments_grpc::CardNetwork::foreign_from);

            let card_details = CardDetails {
                card_number: Some(
                    CardNumber::from_str(&card.card_number.get_card_no()).change_context(
                        UnifiedConnectorServiceError::RequestEncodingFailedWithReason(
                            "Failed to parse card number".to_string(),
                        ),
                    )?,
                ),
                card_exp_month: Some(card_exp_month.into()),
                card_exp_year: Some(card.card_exp_year.expose().into()),
                card_cvc: Some(card.card_cvc.expose().into()),
                card_holder_name: card.card_holder_name.map(|name| name.expose().into()),
                card_issuer: card.card_issuer.clone(),
                card_network: card_network.map(|card_network| card_network.into()),
                card_type: card.card_type.clone(),
                bank_code: card.bank_code.clone(),
                nick_name: card.nick_name.map(|n| n.expose()),
                card_issuing_country_alpha2: card.card_issuing_country.clone(),
            };

            Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::Card(card_details)),
            })
        }
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::CardWithOptionalCVC(
            card,
        ) => {
            let card_exp_month = card
                .get_card_expiry_month_2_digit()
                .attach_printable("Failed to extract 2-digit expiry month from card")
                .change_context(UnifiedConnectorServiceError::InvalidDataFormat {
                    field_name: "card_exp_month",
                })?
                .peek()
                .to_string();

            let card_network = card
                .card_network
                .clone()
                .map(payments_grpc::CardNetwork::foreign_from);

            let card_details = CardDetailsWithNoCvc {
                card_number: Some(
                    CardNumber::from_str(&card.card_number.get_card_no()).change_context(
                        UnifiedConnectorServiceError::RequestEncodingFailedWithReason(
                            "Failed to parse card number".to_string(),
                        ),
                    )?,
                ),
                card_exp_month: Some(card_exp_month.into()),
                card_exp_year: Some(card.card_exp_year.expose().into()),
                card_holder_name: card.card_holder_name.map(|name| name.expose().into()),
                card_issuer: card.card_issuer.clone(),
                card_network: card_network.map(|card_network| card_network.into()),
                card_type: card.card_type.clone(),
                bank_code: card.bank_code.clone(),
                nick_name: card.nick_name.map(|n| n.expose()),
                card_issuing_country_alpha2: card.card_issuing_country.clone(),
            };

            Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::CardWithNoCvc(card_details)),
            })
        }
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::CardRedirect(
            card_redirect_data,
        ) => {
            let card_redirect_type = match card_redirect_data {
                hyperswitch_domain_models::payment_method_data::CardRedirectData::Knet {} => {
                    payments_grpc::card_redirect::CardRedirectType::Knet
                }
                hyperswitch_domain_models::payment_method_data::CardRedirectData::Benefit {} => {
                    payments_grpc::card_redirect::CardRedirectType::Benefit
                }
                hyperswitch_domain_models::payment_method_data::CardRedirectData::MomoAtm {} => {
                    payments_grpc::card_redirect::CardRedirectType::MomoAtm
                }
                hyperswitch_domain_models::payment_method_data::CardRedirectData::CardRedirect {} => {
                    payments_grpc::card_redirect::CardRedirectType::CardRedirect
                }
            };

            Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::CardRedirect(payments_grpc::CardRedirect {
                    r#type: card_redirect_type.into(),
                })),
            })
        }
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::Upi(upi_data) => {
            let upi_type = match upi_data {
                hyperswitch_domain_models::payment_method_data::UpiData::UpiCollect(
                    upi_collect_data,
                ) => {
                    let upi_details = payments_grpc::UpiCollect {
                        vpa_id: upi_collect_data.vpa_id.map(|vpa| vpa.expose().into()),
                        upi_source: upi_collect_data
                            .upi_source
                            .map(payments_grpc::UpiSource::foreign_try_from)
                            .transpose()?
                            .map(|upi_source| upi_source.into()),
                    };
                    PaymentMethod::UpiCollect(upi_details)
                }
                hyperswitch_domain_models::payment_method_data::UpiData::UpiIntent(upi_intent_data) => {
                    let upi_details = payments_grpc::UpiIntent {
                        app_name: upi_intent_data.app_name,
                        upi_source: upi_intent_data
                            .upi_source
                            .map(payments_grpc::UpiSource::foreign_try_from)
                            .transpose()?
                            .map(|upi_source| upi_source.into()),
                    };
                    PaymentMethod::UpiIntent(upi_details)
                }
                hyperswitch_domain_models::payment_method_data::UpiData::UpiQr(upi_qr_data) => {
                    let upi_details = payments_grpc::UpiQr {
                        upi_source: upi_qr_data
                            .upi_source
                            .map(payments_grpc::UpiSource::foreign_try_from)
                            .transpose()?
                            .map(|upi_source| upi_source.into()),
                    };
                    PaymentMethod::UpiQr(upi_details)
                }
            };

            Ok(payments_grpc::PaymentMethod {
                payment_method: Some(upi_type),
            })
        }
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::BankRedirect(
            bank_redirect_data,
        ) => match bank_redirect_data {
            hyperswitch_domain_models::payment_method_data::BankRedirectData::OpenBankingUk {
                issuer,
                country,
            } => {
                let open_banking_uk = payments_grpc::OpenBankingUk {
                    issuer: issuer.map(|issuer| issuer.to_string()),
                    country: country.map(|country| country.to_string()),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::OpenBankingUk(open_banking_uk)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::OpenBanking {} =>
                Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::OpenBanking(OpenBanking {})),
                    }),
            hyperswitch_domain_models::payment_method_data::BankRedirectData::Ideal { bank_name } => {
                let ideal = payments_grpc::Ideal {
                    bank_name: bank_name.map(payments_grpc::BankNames::foreign_try_from)
                    .transpose()?
                    .map(|b| b.into()),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::Ideal(ideal)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::Giropay {
                bank_account_bic,
                bank_account_iban,
                country,
            } => {
                let giropay = payments_grpc::Giropay {
                    bank_account_bic: bank_account_bic.map(|v| v.expose().into()),
                    bank_account_iban: bank_account_iban.map(|v| v.expose().into()),
                    country: country.and_then(|c| payments_grpc::CountryAlpha2::from_str_name(&c.to_string()))
                    .map(|country| country.into()),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::Giropay(giropay)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::Eps { bank_name, country } => {
                let eps = payments_grpc::Eps {
                    bank_name: bank_name.map(payments_grpc::BankNames::foreign_try_from)
                    .transpose()?
                    .map(|b| b.into()),
                    country: country.and_then(|c| payments_grpc::CountryAlpha2::from_str_name(&c.to_string()))
                    .map(|country| country.into()),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::Eps(eps)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::Sofort { country, preferred_language } => {
                let sofort = payments_grpc::Sofort {
                    country: country.and_then(|c| payments_grpc::CountryAlpha2::from_str_name(&c.to_string()))
                    .map(|country| country.into()),
                    preferred_language,
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::Sofort(sofort)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::Przelewy24 { bank_name } => {
                let p24 = payments_grpc::Przelewy24 {
                    bank_name: bank_name.map(payments_grpc::BankNames::foreign_try_from)
                    .transpose()?
                    .map(|b| b.into()),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::Przelewy24(p24)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::Blik { blik_code}  => {
                let blik = payments_grpc::Blik {
                    blik_code,
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::Blik(blik)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::Trustly { country } => {
                let trustly = payments_grpc::Trustly {
                    country: country.and_then(|c| payments_grpc::CountryAlpha2::from_str_name(&c.to_string())).map(|c| c.into()),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::Trustly(trustly)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::Interac {
                country,
                email,
            } => {
                let interac = payments_grpc::Interac {
                    country: country
                        .and_then(|c| payments_grpc::CountryAlpha2::from_str_name(&c.to_string()))
                        .map(|c| c.into()),
                    email: email.map(|e| e.expose().expose().into()),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::Interac(interac)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::Bizum { } => {
                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::Bizum(payments_grpc::Bizum{})),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::LocalBankRedirect { } => {
                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::LocalBankRedirect(payments_grpc::LocalBankRedirect{})),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::OnlineBankingCzechRepublic {
                issuer
            } => {
                let online_banking_czech_republic = payments_grpc::OnlineBankingCzechRepublic {
                    issuer: payments_grpc::BankNames::foreign_try_from(issuer)?.into(),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::OnlineBankingCzechRepublic(online_banking_czech_republic)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::OnlineBankingFinland {
                email
            } => {
                let online_banking_finland = payments_grpc::OnlineBankingFinland {
                    email: email.map(|e| e.expose().expose().into()),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::OnlineBankingFinland(online_banking_finland)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::OnlineBankingPoland {
                issuer
            } => {
                let online_banking_poland = payments_grpc::OnlineBankingPoland {
                    issuer: payments_grpc::BankNames::foreign_try_from(issuer)?.into(),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::OnlineBankingPoland(online_banking_poland)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::OnlineBankingSlovakia {
                issuer
            } => {
                let online_banking_slovakia = payments_grpc::OnlineBankingSlovakia {
                    issuer: payments_grpc::BankNames::foreign_try_from(issuer)?.into(),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::OnlineBankingSlovakia(online_banking_slovakia)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::OnlineBankingThailand {
                issuer
            } => {
                let online_banking_thailand = payments_grpc::OnlineBankingThailand {
                    issuer: payments_grpc::BankNames::foreign_try_from(issuer)?.into(),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::OnlineBankingThailand(online_banking_thailand)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::Eft {
                provider,
            } => {
                let eft_bank_redirect = payments_grpc::EftBankRedirect { provider };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::EftBankRedirect(eft_bank_redirect)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::BancontactCard {
                card_number,
                card_exp_month,
                card_exp_year,
                card_holder_name,
            } => {
                let bancontact = payments_grpc::BancontactCard {
                    card_number: card_number.map(|v| CardNumber::from_str(&v.get_card_no())).transpose()
                        .change_context(
                        UnifiedConnectorServiceError::RequestEncodingFailedWithReason(
                            "Failed to parse card number".to_string(),
                        ),
                    )?,
                    card_exp_month: card_exp_month.map(|v| v.expose().into()),
                    card_exp_year: card_exp_year.map(|v| v.expose().into()),
                    card_holder_name: card_holder_name.map(|v| v.expose().into()),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::BancontactCard(bancontact)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankRedirectData::OnlineBankingFpx { issuer }  => {
                let online_banking_fpx = payments_grpc::OnlineBankingFpx {
                    issuer: payments_grpc::BankNames::foreign_try_from(issuer)?.into(),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::OnlineBankingFpx(online_banking_fpx)),
                })
            }
        },
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::RealTimePayment(
            bank_transfer_data,
        ) => match *bank_transfer_data {
            hyperswitch_domain_models::payment_method_data::RealTimePaymentData::DuitNow {  } => Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::DuitNow(
                    payments_grpc::DuitNow {  }
                )),
            }),
            _ => Err(UnifiedConnectorServiceError::NotImplemented(format!(
                "Unimplemented payment method subtype: {payment_method_type:?}"
            ))
            .into()),
        }
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::BankTransfer(
            bank_transfer_data,
        ) => match *bank_transfer_data {
            hyperswitch_domain_models::payment_method_data::BankTransferData::AchBankTransfer {  } => Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::AchBankTransfer(
                    payments_grpc::AchBankTransfer {  }
                )),
            }),
            hyperswitch_domain_models::payment_method_data::BankTransferData::BacsBankTransfer {  } => Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::BacsBankTransfer(
                    payments_grpc::BacsBankTransfer {  }
                )),
            }),
            hyperswitch_domain_models::payment_method_data::BankTransferData::SepaBankTransfer {  } => Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::SepaBankTransfer(
                    payments_grpc::SepaBankTransfer {  }
                )),
            }),
            hyperswitch_domain_models::payment_method_data::BankTransferData::MultibancoBankTransfer {  } => Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::MultibancoBankTransfer(
                    payments_grpc::MultibancoBankTransfer {  }
                )),
            }),
            hyperswitch_domain_models::payment_method_data::BankTransferData::Pix {
                pix_key,
                cpf,
                cnpj,
                source_bank_account_id,
                destination_bank_account_id,
                expiry_date,
            } => Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::Pix(payments_grpc::PixPayment {
                    pix_key: pix_key.map(|v| v.expose().into()),
                    cpf: cpf.map(|v| v.expose().into()),
                    cnpj: cnpj.map(|v| v.expose().into()),
                    source_bank_account_id: source_bank_account_id.map(|v| v.expose_inner()),
                    destination_bank_account_id: destination_bank_account_id.map(|v| v.expose_inner()),
                    expiry_date: expiry_date.map(|dt| {
                        dt.format(&time::format_description::well_known::Iso8601::DEFAULT)
                            .unwrap_or_default()
                    }),
                })),
            }),
            hyperswitch_domain_models::payment_method_data::BankTransferData::PixAutomaticoPush { .. }
            | hyperswitch_domain_models::payment_method_data::BankTransferData::PixAutomaticoQr {}
            | hyperswitch_domain_models::payment_method_data::BankTransferData::PixEmv {}
            | hyperswitch_domain_models::payment_method_data::BankTransferData::PixQr {} => {
                Err(UnifiedConnectorServiceError::NotImplemented(format!(
                    "Unimplemented payment method subtype: {payment_method_type:?}"
                ))
                .into())
            }
            hyperswitch_domain_models::payment_method_data::BankTransferData::PermataBankTransfer {} => {
                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::PermataBankTransfer(
                        payments_grpc::PermataBankTransfer {},
                    )),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankTransferData::BcaBankTransfer {} => {
                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::BcaBankTransfer(
                        payments_grpc::BcaBankTransfer {},
                    )),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankTransferData::BniVaBankTransfer {} => {
                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::BniVaBankTransfer(
                        payments_grpc::BniVaBankTransfer {},
                    )),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankTransferData::BriVaBankTransfer {} => {
                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::BriVaBankTransfer(
                        payments_grpc::BriVaBankTransfer {},
                    )),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankTransferData::CimbVaBankTransfer {} => {
                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::CimbVaBankTransfer(
                        payments_grpc::CimbVaBankTransfer {},
                    )),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankTransferData::DanamonVaBankTransfer {} => {
                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::DanamonVaBankTransfer(
                        payments_grpc::DanamonVaBankTransfer {},
                    )),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankTransferData::MandiriVaBankTransfer {} => {
                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::MandiriVaBankTransfer(
                        payments_grpc::MandiriVaBankTransfer {},
                    )),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankTransferData::LocalBankTransfer {bank_code} =>
                Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::LocalBankTransfer(payments_grpc::LocalBankTransfer {
                            bank_code,
                        })),
                    }),
            hyperswitch_domain_models::payment_method_data::BankTransferData::IndonesianBankTransfer {bank_name} =>
                Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::IndonesianBankTransfer(payments_grpc::IndonesianBankTransfer {
                            bank_name: bank_name.map(payments_grpc::BankNames::foreign_try_from).transpose()?.map(|b| b.into()),
                        })),
                    }),
            hyperswitch_domain_models::payment_method_data::BankTransferData::Pse {} =>
                Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::Pse(payments_grpc::Pse {})),
                    }),
            hyperswitch_domain_models::payment_method_data::BankTransferData::InstantBankTransfer {} =>
                Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::InstantBankTransfer(payments_grpc::InstantBankTransfer {})),
                    }),
            hyperswitch_domain_models::payment_method_data::BankTransferData::InstantBankTransferFinland {} =>
                Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::InstantBankTransferFinland(payments_grpc::InstantBankTransferFinland {})),
                    }),
            hyperswitch_domain_models::payment_method_data::BankTransferData::InstantBankTransferPoland {} =>
                Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::InstantBankTransferPoland(payments_grpc::InstantBankTransferPoland {})),
                    }),
        },
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::PayLater(
            pay_later_data,
        ) => match pay_later_data {
            hyperswitch_domain_models::payment_method_data::PayLaterData::KlarnaRedirect {  } => Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::Klarna(
                    payments_grpc::Klarna {  }
                )),
            }),
            hyperswitch_domain_models::payment_method_data::PayLaterData::AffirmRedirect {  } => Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::Affirm(
                    payments_grpc::Affirm {  }
                )),
            }),
            hyperswitch_domain_models::payment_method_data::PayLaterData::AfterpayClearpayRedirect {  } => Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::AfterpayClearpay(
                    payments_grpc::AfterpayClearpay {  }
                )),
            }),
            _ => Err(UnifiedConnectorServiceError::NotImplemented(format!(
                "Unimplemented payment method subtype: {payment_method_type:?}"
            ))
            .into()),
        },
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::Reward => {
            match payment_method_type {
                Some(PaymentMethodType::ClassicReward) => Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::ClassicReward(ClassicReward {})),
                }),
                Some(PaymentMethodType::Evoucher) => Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::EVoucher(EVoucher {})),
                }),
                None | Some(_) => Err(UnifiedConnectorServiceError::NotImplemented(format!(
                    "Unimplemented payment method subtype: {payment_method_type:?}"
                ))
                .into()),
            }
        }
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::Wallet(wallet_data) => {
            match wallet_data {
                hyperswitch_domain_models::payment_method_data::WalletData::Mifinity(
                    mifinity_data,
                ) => Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::MifinityRedirect(payments_grpc::MifinityRedirectWallet {
                        date_of_birth: Some(mifinity_data.date_of_birth.peek().to_string().into()),
                        language_preference: mifinity_data.language_preference,
                    })),
                }),
                hyperswitch_domain_models::payment_method_data::WalletData::ApplePay(
                    apple_pay_wallet_data
                ) => match payment_method_token {
                    // A connector token minted by the PaymentMethodToken pre-step (e.g. a
                    // Paysafe payment handle) supersedes the wallet payload: the settle
                    // leg must reference the handle, not re-send the PKPaymentToken.
                    // Mirrors the Card arm above.
                    Some(PaymentMethodToken::Token(token)) => {
                        let token_payment_method = payments_grpc::TokenPaymentMethodType {
                            token: Some(token.clone()),
                        };
                        Ok(payments_grpc::PaymentMethod {
                            payment_method: Some(PaymentMethod::Token(token_payment_method)),
                        })
                    }
                    _ => {
                        let apple_pay_payment_data = get_apple_pay_payment_data(
                            &apple_pay_wallet_data.payment_data,
                            payment_method_token,
                        );

                        let payment_data =
                            payments_grpc::apple_wallet::payment_data::PaymentData::foreign_try_from(
                                &apple_pay_payment_data,
                            )?;

                        Ok(payments_grpc::PaymentMethod {
                            payment_method: Some(PaymentMethod::ApplePaySdk(payments_grpc::AppleWallet {
                                payment_data: Some(payments_grpc::apple_wallet::PaymentData {
                                    payment_data: Some(payment_data),
                                }),
                                payment_method: Some(payments_grpc::apple_wallet::PaymentMethod {
                                    display_name: apple_pay_wallet_data.payment_method.display_name,
                                    network: apple_pay_wallet_data.payment_method.network,
                                    r#type: apple_pay_wallet_data.payment_method.pm_type,
                                }),
                                transaction_identifier: apple_pay_wallet_data.transaction_identifier,
                            })),
                        })
                    }
                },
                hyperswitch_domain_models::payment_method_data::WalletData::GooglePay(
                    google_pay_wallet_data,
                ) => {
                    let google_pay_tokenization_data = get_google_pay_tokenization_data(
                        &google_pay_wallet_data.tokenization_data,
                        payment_method_token,
                        connector_meta_data,
                    );

                    let tokenization_data =
                        payments_grpc::google_wallet::tokenization_data::TokenizationData::foreign_try_from(
                            &google_pay_tokenization_data,
                        )?;

                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::GooglePaySdk(payments_grpc::GoogleWallet {
                            r#type: google_pay_wallet_data.pm_type,
                            description: google_pay_wallet_data.description,
                            info: Some(payments_grpc::google_wallet::PaymentMethodInfo {
                                card_network: google_pay_wallet_data.info.card_network,
                                card_details: google_pay_wallet_data.info.card_details,
                                assurance_details: google_pay_wallet_data.info.assurance_details.map(|details| {
                                    payments_grpc::google_wallet::payment_method_info::AssuranceDetails {
                                        card_holder_authenticated: details.card_holder_authenticated,
                                        account_verified: details.account_verified,
                                    }
                                }),
                            }),
                            tokenization_data: Some(payments_grpc::google_wallet::TokenizationData {
                                tokenization_data: Some(tokenization_data),
                            }),
                        })),
                    })
                }
                hyperswitch_domain_models::payment_method_data::WalletData::GooglePayThirdPartySdk(
                    google_pay_sdk_data,
                ) => {
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::GooglePayThirdPartySdk(
                            payments_grpc::GooglePayThirdPartySdkWallet {
                                token: google_pay_sdk_data.token.map(|t| t.expose().into()),
                            }
                        )),
                    })
                },
                hyperswitch_domain_models::payment_method_data::WalletData::ApplePayThirdPartySdk(
                    apple_pay_sdk_data,
                ) => {
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::ApplePayThirdPartySdk(
                            payments_grpc::ApplePayThirdPartySdkWallet {
                                token: apple_pay_sdk_data.token.map(|t| t.expose().into()),
                            }
                        )),
                    })
                },
                hyperswitch_domain_models::payment_method_data::WalletData::PaypalSdk(
                    paypal_sdk_data,
                ) => {
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::PaypalSdk(
                            payments_grpc::PaypalSdkWallet {
                                token: Some(paypal_sdk_data.token.into()),
                            }
                        )),
                    })
                },
                hyperswitch_domain_models::payment_method_data::WalletData::AmazonPayRedirect(
                    _,
                ) => Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::AmazonPayRedirect(
                        payments_grpc::AmazonPayRedirectWallet {  }
                    )),
                }),
                hyperswitch_domain_models::payment_method_data::WalletData::CashappQr(
                    _,
                ) => Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::CashappQr(
                        payments_grpc::CashappQrWallet {  }
                    )),
                }),
                hyperswitch_domain_models::payment_method_data::WalletData::WeChatPayQr(
                    _,
                ) => Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::WeChatPayQr(
                        payments_grpc::WeChatPayQrWallet {  }
                    )),
                }),
                hyperswitch_domain_models::payment_method_data::WalletData::AliPayRedirect(
                    _,
                ) => Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::AliPayRedirect(
                        payments_grpc::AliPayRedirectWallet {  }
                    )),
                }),
                hyperswitch_domain_models::payment_method_data::WalletData::RevolutPay(
                    _,
                ) => Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::RevolutPay(
                        payments_grpc::RevolutPayWallet {  }
                    )),
                }),
                hyperswitch_domain_models::payment_method_data::WalletData::BluecodeRedirect {} => Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::BluecodeRedirect(
                        payments_grpc::BluecodeRedirectWallet {  }
                    )),
                }),
                hyperswitch_domain_models::payment_method_data::WalletData::PaypalRedirect(
                    paypal_redirection,
                ) => Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::PaypalRedirect(
                        payments_grpc::PaypalRedirectWallet {
                            email: paypal_redirection.email.map(|e| e.expose().expose().into()),
                        }
                    )),
                }),
                hyperswitch_domain_models::payment_method_data::WalletData::Paze(paze_data) => {
                    let paze_wallet_data = get_paze_wallet_data(&paze_data, payment_method_token)?;
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::PazeSdk(payments_grpc::PazeWallet {
                            paze_data: Some(paze_wallet_data),
                        })),
                    })
                }
                hyperswitch_domain_models::payment_method_data::WalletData::GcashRedirect(_) => {
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::GcashRedirect(
                            payments_grpc::GcashRedirectWallet {},
                        )),
                    })
                }
                hyperswitch_domain_models::payment_method_data::WalletData::Skrill(_) => {
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::SkrillRedirect(
                            payments_grpc::SkrillRedirectWallet {},
                        )),
                    })
                }
                _ => Err(UnifiedConnectorServiceError::NotImplemented(format!(
                    "Unimplemented payment method subtype: {payment_method_type:?}"
                ))
                .into()),
            }
        }
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::Crypto(crypto_data) => {
            Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::Crypto(CryptoCurrency {
                    pay_currency: crypto_data.pay_currency.clone(),
                    network: crypto_data.network.clone(),
                })),
            })
        }
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::CardDetailsForNetworkTransactionId(card_nti_data) => {
            let card_details_for_nti = payments_grpc::CardDetailsForNetworkTransactionId::foreign_try_from(card_nti_data)?;

            Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::CardDetailsForNetworkTransactionId(card_details_for_nti)),
            })
        }
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::NetworkToken(network_token_data) => {
            let network_token = payments_grpc::NetworkTokenData::foreign_try_from(network_token_data)?;

            Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::NetworkToken(network_token)),
            })
        }
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::DecryptedWalletTokenDetailsForNetworkTransactionId(wallet_token_data) => {
            let decrypted_wallet_token_details = payments_grpc::DecryptedWalletTokenDetailsForNetworkTransactionId::foreign_try_from(wallet_token_data)?;

            Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::DecryptedWalletTokenDetailsForNetworkTransactionId(decrypted_wallet_token_details)),
            })
        }
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::BankDebit(
            bank_debit_data,
        ) => match bank_debit_data {
            hyperswitch_domain_models::payment_method_data::BankDebitData::AchBankDebit {
                account_number,
                routing_number,
                bank_account_holder_name,
                bank_name,
                bank_type,
                bank_holder_type,
            } => {
                let bank_name = bank_name
                    .map(payments_grpc::BankNames::foreign_try_from)
                    .transpose()?;
                let bank_type = bank_type
                    .map(payments_grpc::BankType::foreign_try_from)
                    .transpose()?;
                let bank_holder_type = bank_holder_type
                    .map(payments_grpc::BankHolderType::foreign_try_from)
                    .transpose()?;

                let ach = payments_grpc::Ach {
                    account_number: Some(account_number.expose().into()),
                    routing_number: Some(routing_number.expose().into()),
                    card_holder_name: None,
                    bank_account_holder_name: bank_account_holder_name
                        .map(|name| name.expose().into()),
                    bank_name: bank_name.map(Into::into).unwrap_or_default(),
                    bank_type: bank_type.map(Into::into).unwrap_or_default(),
                    bank_holder_type: bank_holder_type.map(Into::into).unwrap_or_default(),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::Ach(ach)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankDebitData::SepaBankDebit {
                iban,
                bank_account_holder_name,
            } => {
                let sepa = payments_grpc::Sepa {
                    iban: Some(iban.expose().into()),
                    bank_account_holder_name: bank_account_holder_name
                        .map(|name| name.expose().into()),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::Sepa(sepa)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankDebitData::BecsBankDebit {
                account_number,
                bsb_number,
                bank_account_holder_name,
            } => {
                let becs = payments_grpc::Becs {
                    account_number: Some(account_number.expose().into()),
                    bsb_number: Some(bsb_number.expose().into()),
                    bank_account_holder_name: bank_account_holder_name
                        .map(|name| name.expose().into()),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::Becs(becs)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankDebitData::BacsBankDebit {
                account_number,
                sort_code,
                bank_account_holder_name,
            } => {
                let bacs = payments_grpc::Bacs {
                    account_number: Some(account_number.expose().into()),
                    sort_code: Some(sort_code.expose().into()),
                    bank_account_holder_name: bank_account_holder_name
                        .map(|name| name.expose().into()),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::Bacs(bacs)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankDebitData::SepaGuarenteedBankDebit {
                iban,
                bank_account_holder_name,
            } => {
                let sepa_guaranteed_debit = payments_grpc::SepaGuaranteedDebit {
                    iban: Some(iban.expose().into()),
                    bank_account_holder_name: bank_account_holder_name
                        .map(|name| name.expose().into()),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::SepaGuaranteedDebit(sepa_guaranteed_debit)),
                })
            }
            hyperswitch_domain_models::payment_method_data::BankDebitData::EftDebitOrder {
                account_number,
                branch_code,
                bank_account_holder_name,
                bank_name,
                bank_type,
            } => {
                let bank_name = bank_name
                    .map(payments_grpc::BankNames::foreign_try_from)
                    .transpose()?;
                let bank_type = bank_type
                    .map(payments_grpc::BankType::foreign_try_from)
                    .transpose()?;

                let eft = payments_grpc::Eft {
                    account_number: Some(account_number.expose().into()),
                    branch_code: branch_code.map(|code| code.expose().into()),
                    bank_account_holder_name: bank_account_holder_name
                        .map(|name| name.expose().into()),
                    bank_name: bank_name.map(Into::into).unwrap_or_default(),
                    bank_type: bank_type.map(Into::into).unwrap_or_default(),
                };

                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::Eft(eft)),
                })
            }
        },
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::Voucher(voucher_data) => {
            match voucher_data {
                hyperswitch_domain_models::payment_method_data::VoucherData::Boleto(boleto_data) => {
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::Boleto(
                            payments_grpc::Boleto {
                                social_security_number: boleto_data
                                    .social_security_number
                                    .map(|ssn| ssn.expose()),
                                expiration_date: boleto_data.due_date.clone(),
                            },
                        )),
                    })
                }
                hyperswitch_domain_models::payment_method_data::VoucherData::Alfamart(_) => {
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::Alfamart(
                            payments_grpc::Alfamart {},
                        )),
                    })
                }
                hyperswitch_domain_models::payment_method_data::VoucherData::Indomaret(_) => {
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::Indomaret(
                            payments_grpc::Indomaret {},
                        )),
                    })
                }
                hyperswitch_domain_models::payment_method_data::VoucherData::Oxxo => {
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::Oxxo(
                            payments_grpc::Oxxo {},
                        )),
                    })
                }
                hyperswitch_domain_models::payment_method_data::VoucherData::SevenEleven(_) => {
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::SevenEleven(
                            payments_grpc::SevenEleven {},
                        )),
                    })
                }
                hyperswitch_domain_models::payment_method_data::VoucherData::Lawson(_) => {
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::Lawson(
                            payments_grpc::Lawson {},
                        )),
                    })
                }
                hyperswitch_domain_models::payment_method_data::VoucherData::MiniStop(_) => {
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::MiniStop(
                            payments_grpc::MiniStop {},
                        )),
                    })
                }
                hyperswitch_domain_models::payment_method_data::VoucherData::FamilyMart(_) => {
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::FamilyMart(
                            payments_grpc::FamilyMart {},
                        )),
                    })
                }
                hyperswitch_domain_models::payment_method_data::VoucherData::Seicomart(_) => {
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::Seicomart(
                            payments_grpc::Seicomart {},
                        )),
                    })
                }
                hyperswitch_domain_models::payment_method_data::VoucherData::PayEasy(_) => {
                    Ok(payments_grpc::PaymentMethod {
                        payment_method: Some(PaymentMethod::PayEasy(
                            payments_grpc::PayEasy {},
                        )),
                    })
                }
                 _ => Err(UnifiedConnectorServiceError::NotImplemented(format!(
                "Unimplemented payment method subtype: {payment_method_type:?}"
            ))
            .into()),
            }
        }
        hyperswitch_domain_models::payment_method_data::PaymentMethodData::GiftCard(
            gift_card_data,
        ) => match *gift_card_data {
            hyperswitch_domain_models::payment_method_data::GiftCardData::PaySafeCard {} => {
                Ok(payments_grpc::PaymentMethod {
                    payment_method: Some(PaymentMethod::PaySafeCard(
                        payments_grpc::PaySafeCard {},
                    )),
                })
            }
            _ => Err(UnifiedConnectorServiceError::NotImplemented(format!(
                "Unimplemented payment method subtype: {payment_method_type:?}"
            ))
            .into()),
        },
        _ => Err(UnifiedConnectorServiceError::NotImplemented(format!(
            "Unimplemented payment method: {payment_method_data:?}"
        ))
        .into()),
    }
}

pub fn build_unified_connector_service_payment_method_for_external_proxy(
    payment_method_data: hyperswitch_domain_models::payment_method_data::ExternalVaultPaymentMethodData,
    payment_method_type: Option<PaymentMethodType>,
) -> CustomResult<payments_grpc::PaymentMethod, UnifiedConnectorServiceError> {
    match payment_method_data {
        hyperswitch_domain_models::payment_method_data::ExternalVaultPaymentMethodData::Card(
            external_vault_card,
        ) => {
            let card_network = external_vault_card
                .card_network
                .clone()
                .map(payments_grpc::CardNetwork::foreign_from);
            let card_details = payments_grpc::ProxyCardDetails {
                card_number: Some(external_vault_card.card_number.expose().into()),
                card_exp_month: Some(external_vault_card.card_exp_month.expose().into()),
                card_exp_year: Some(external_vault_card.card_exp_year.expose().into()),
                card_cvc: Some(external_vault_card.card_cvc.expose().into()),
                card_holder_name: external_vault_card.card_holder_name.map(|name| name.expose().into()),
                card_issuer: external_vault_card.card_issuer.clone(),
                card_network: card_network.map(|card_network| card_network.into()),
                card_type: external_vault_card.card_type.clone(),
                bank_code: external_vault_card.bank_code.clone(),
                nick_name: external_vault_card.nick_name.map(|n| n.expose()),
                card_issuing_country_alpha2: external_vault_card.card_issuing_country.clone(),
            };
            Ok(payments_grpc::PaymentMethod {
                payment_method: Some(PaymentMethod::CardProxy(card_details)),
            })
        }
        hyperswitch_domain_models::payment_method_data::ExternalVaultPaymentMethodData::VaultToken(_) => {
            Err(UnifiedConnectorServiceError::NotImplemented(format!(
                        "Unimplemented payment method subtype: {payment_method_type:?}"
            ))
            .into())
        }
    }
}

/// Gets the UCS client from session state
fn get_ucs_client(
    state: &SessionState,
) -> RouterResult<
    &external_services::grpc_client::unified_connector_service::UnifiedConnectorServiceClient,
> {
    state
        .grpc_client
        .unified_connector_service_client
        .as_ref()
        .ok_or_else(|| {
            error_stack::report!(errors::ApiErrorResponse::InternalServerError)
                .attach_printable("UCS client is not available")
        })
}

fn as_header_value(value: &Secret<String>) -> Option<Secret<String>> {
    http::HeaderValue::from_str(value.peek())
        .is_ok()
        .then(|| value.clone())
}

pub fn build_unified_connector_service_auth_metadata(
    #[cfg(feature = "v1")] merchant_connector_account: MerchantConnectorAccountType,
    #[cfg(feature = "v2")] merchant_connector_account: MerchantConnectorAccountTypeDetails,
    processor_merchant_id: &id_type::MerchantId,
    connector_name: String,
) -> CustomResult<ConnectorAuthMetadata, UnifiedConnectorServiceError> {
    #[cfg(feature = "v1")]
    let auth_type: ConnectorAuthType = merchant_connector_account
        .get_connector_account_details()
        .parse_value("ConnectorAuthType")
        .change_context(UnifiedConnectorServiceError::FailedToObtainAuthType)
        .attach_printable("Failed while parsing value for ConnectorAuthType")?;

    #[cfg(feature = "v2")]
    let auth_type: ConnectorAuthType = merchant_connector_account
        .get_connector_account_details()
        .change_context(UnifiedConnectorServiceError::FailedToObtainAuthType)
        .attach_printable("Failed to obtain ConnectorAuthType")?;
    let merchant_id = processor_merchant_id.get_string_repr();
    // Extract connector metadata from MCA for connector-specific config
    let merchant_account_metadata = merchant_connector_account.get_metadata();
    let merchant_account_metadata_value = merchant_account_metadata
        .as_ref()
        .and_then(|m| serde_json::to_value(m.clone().expose()).ok());
    // Build connector-specific config for supported connectors
    let connector_config = connector_config::build_connector_config_header(
        &connector_name,
        &auth_type,
        merchant_account_metadata_value.as_ref(),
    )
    .change_context(UnifiedConnectorServiceError::FailedToObtainAuthType)
    .attach_printable("Failed to build connector config header")?
    .map(Secret::new);

    match &auth_type {
        ConnectorAuthType::SignatureKey {
            api_key,
            key1,
            api_secret,
        } => Ok(ConnectorAuthMetadata {
            connector_name,
            auth_type: consts::UCS_AUTH_SIGNATURE_KEY.to_string(),
            api_key: Some(api_key.clone()),
            key1: Some(key1.clone()),
            key2: None,
            api_secret: Some(api_secret.clone()),
            auth_key_map: None,
            merchant_id: Secret::new(merchant_id.to_string()),
            connector_config,
        }),
        ConnectorAuthType::BodyKey { api_key, key1 } => Ok(ConnectorAuthMetadata {
            connector_name,
            auth_type: consts::UCS_AUTH_BODY_KEY.to_string(),
            api_key: Some(api_key.clone()),
            key1: Some(key1.clone()),
            key2: None,
            api_secret: None,
            auth_key_map: None,
            merchant_id: Secret::new(merchant_id.to_string()),
            connector_config,
        }),
        ConnectorAuthType::HeaderKey { api_key } => Ok(ConnectorAuthMetadata {
            connector_name,
            auth_type: consts::UCS_AUTH_HEADER_KEY.to_string(),
            api_key: Some(api_key.clone()),
            key1: None,
            key2: None,
            api_secret: None,
            auth_key_map: None,
            merchant_id: Secret::new(merchant_id.to_string()),
            connector_config,
        }),
        ConnectorAuthType::CurrencyAuthKey { auth_key_map } => Ok(ConnectorAuthMetadata {
            connector_name,
            auth_type: consts::UCS_AUTH_CURRENCY_AUTH_KEY.to_string(),
            api_key: None,
            key1: None,
            key2: None,
            api_secret: None,
            auth_key_map: Some(auth_key_map.clone()),
            merchant_id: Secret::new(merchant_id.to_string()),
            connector_config,
        }),
        ConnectorAuthType::MultiAuthKey {
            api_key,
            key1,
            api_secret,
            key2,
        } => Ok(ConnectorAuthMetadata {
            connector_name,
            auth_type: consts::UCS_AUTH_MULTI_KEY.to_string(),
            api_key: as_header_value(api_key),
            key1: as_header_value(key1),
            key2: as_header_value(key2),
            api_secret: as_header_value(api_secret),
            auth_key_map: None,
            merchant_id: Secret::new(merchant_id.to_string()),
            connector_config,
        }),
        ConnectorAuthType::CertificateAuth {
            certificate,
            private_key,
        } => Ok(ConnectorAuthMetadata {
            connector_name,
            auth_type: consts::UCS_AUTH_MULTI_KEY.to_string(),
            api_key: Some(certificate.clone()),
            key1: Some(private_key.clone()),
            key2: None,
            api_secret: None,
            auth_key_map: None,
            merchant_id: Secret::new(merchant_id.to_string()),
            connector_config,
        }),
        ConnectorAuthType::TemporaryAuth | ConnectorAuthType::NoKey => {
            Err(UnifiedConnectorServiceError::FailedToObtainAuthType)
                .attach_printable("Unsupported ConnectorAuthType for header injection")
        }
    }
}

pub fn parse_merchant_reference_id(id: &str) -> Option<id_type::PaymentReferenceId> {
    id_type::PaymentReferenceId::from_str(id)
        .inspect_err(|err| {
            logger::warn!(
                error = ?err,
                "Invalid Merchant ReferenceId found"
            )
        })
        .ok()
}

pub fn parse_merchant_payout_reference_id(id: &str) -> Option<id_type::PayoutReferenceId> {
    id_type::PayoutReferenceId::from_str(id)
        .inspect_err(|err| {
            logger::warn!(
                error = ?err,
                "Invalid Merchant Payout ReferenceId found"
            )
        })
        .ok()
}

#[cfg(feature = "v1")]
pub fn build_unified_connector_service_external_vault_proxy_metadata_v1(
    external_vault_merchant_connector_account: MerchantConnectorAccountType,
    connectors: &hyperswitch_domain_models::connector_endpoints::Connectors,
) -> CustomResult<String, UnifiedConnectorServiceError> {
    let connector_name = external_vault_merchant_connector_account
        .get_connector_name()
        .ok_or(UnifiedConnectorServiceError::InvalidConnectorName)
        .attach_printable("Connector name not found in MerchantConnectorAccountType")?;

    let external_vault_connector = api_enums::VaultConnectors::try_from(connector_name.clone())
        .map_err(|err| {
            error_stack::report!(UnifiedConnectorServiceError::InvalidConnectorName)
                .attach_printable(format!("Failed to parse Vault connector: {err}"))
        })?;

    let external_vault_proxy_config = match external_vault_connector {
        api_enums::VaultConnectors::Vgs => {
            let external_vault_metadata = external_vault_merchant_connector_account
                .get_metadata()
                .ok_or(UnifiedConnectorServiceError::ParsingFailed)
                .attach_printable("Metadata is required for VGS vault connector")?;

            let external_vault_metadata_parsed: ExternalVaultConnectorMetadata =
                external_vault_metadata
                    .expose()
                    .parse_value("ExternalVaultConnectorMetadata")
                    .change_context(UnifiedConnectorServiceError::ParsingFailed)
                    .attach_printable("Failed to parse external vault connector metadata")?;

            external_services::grpc_client::unified_connector_service::ExternalVaultProxyConfig {
                vault_connector_type:
                    external_services::grpc_client::unified_connector_service::VaultConnectorType::Proxy,
                vault_connector_id: Some("vgs".to_string()),
                metadata: external_services::grpc_client::unified_connector_service::ExternalVaultProxyMetadata::VgsMetadata(
                    external_services::grpc_client::unified_connector_service::VgsMetadata {
                        proxy_url: external_vault_metadata_parsed.proxy_url,
                        certificate: external_vault_metadata_parsed.certificate,
                    },
                ),
            }
        }
        api_enums::VaultConnectors::HyperswitchVault => {
            let base = &connectors.hyperswitch_vault.base_url;
            let vault_endpoint_url = format!("{}/proxy", base)
                .parse::<url::Url>()
                .map(common_utils::types::Url::wrap)
                .change_context(UnifiedConnectorServiceError::ParsingFailed)
                .attach_printable("Failed to parse hyperswitch_vault endpoint URL")?;

            let auth_type: ConnectorAuthType = external_vault_merchant_connector_account
                .get_connector_account_details()
                .parse_value("ConnectorAuthType")
                .change_context(UnifiedConnectorServiceError::FailedToObtainAuthType)
                .attach_printable("Failed to obtain ConnectorAuthType for HyperswitchVault")?;

            let (api_key, profile_id) = match &auth_type {
                ConnectorAuthType::SignatureKey {
                    api_key,
                    api_secret,
                    ..
                } => (api_key.clone(), api_secret.clone()),
                _ => {
                    return Err(error_stack::report!(
                        UnifiedConnectorServiceError::FailedToObtainAuthType
                    )
                    .attach_printable("HyperswitchVault requires SignatureKey auth type"))
                }
            };

            external_services::grpc_client::unified_connector_service::ExternalVaultProxyConfig {
                vault_connector_type:
                    external_services::grpc_client::unified_connector_service::VaultConnectorType::Transformation,
                vault_connector_id: Some("hyperswitch_vault".to_string()),
                metadata: external_services::grpc_client::unified_connector_service::ExternalVaultProxyMetadata::HyperswitchVaultMetadata(
                    external_services::grpc_client::unified_connector_service::HyperswitchVaultMetadata {
                        vault_endpoint: vault_endpoint_url,
                        vault_auth_data: external_services::grpc_client::unified_connector_service::VaultConnectorAuth {
                            api_key,
                            profile_id,
                        },
                    },
                ),
            }
        }
        api_enums::VaultConnectors::Tokenex => Err(error_stack::report!(
            UnifiedConnectorServiceError::NotImplemented(
                "External vault proxy metadata is not supported for Tokenex".to_string(),
            )
        ))?,
    };

    logger::info!(external_vault_proxy_config = ?external_vault_proxy_config, "Built ExternalVaultProxyConfig (v1)");

    let external_vault_config_bytes = serde_json::to_vec(&external_vault_proxy_config)
        .change_context(UnifiedConnectorServiceError::ParsingFailed)
        .attach_printable("Failed to serialize ExternalVaultProxyConfig to bytes")?;

    Ok(BASE64_ENGINE.encode(&external_vault_config_bytes))
}

#[cfg(feature = "v2")]
pub fn build_unified_connector_service_external_vault_proxy_metadata(
    external_vault_merchant_connector_account: MerchantConnectorAccountTypeDetails,
    connectors: &hyperswitch_domain_models::connector_endpoints::Connectors,
) -> CustomResult<String, UnifiedConnectorServiceError> {
    let connector = external_vault_merchant_connector_account.get_connector_name();

    let external_vault_connector =
        api_enums::VaultConnectors::try_from(connector).map_err(|err| {
            error_stack::report!(UnifiedConnectorServiceError::InvalidConnectorName)
                .attach_printable(format!("Failed to parse Vault connector: {err}"))
        })?;

    let external_vault_proxy_config = match external_vault_connector {
        api_enums::VaultConnectors::Vgs => {
            let external_vault_metadata = external_vault_merchant_connector_account
                .get_metadata()
                .ok_or(UnifiedConnectorServiceError::ParsingFailed)
                .attach_printable("Metadata is required for VGS vault connector")?;

            let external_vault_metadata_parsed: ExternalVaultConnectorMetadata =
                external_vault_metadata
                    .expose()
                    .parse_value("ExternalVaultConnectorMetadata")
                    .change_context(UnifiedConnectorServiceError::ParsingFailed)
                    .attach_printable("Failed to parse external vault connector metadata")?;

            external_services::grpc_client::unified_connector_service::ExternalVaultProxyConfig {
                vault_connector_type:
                    external_services::grpc_client::unified_connector_service::VaultConnectorType::Proxy,
                vault_connector_id: Some("vgs".to_string()),
                metadata: external_services::grpc_client::unified_connector_service::ExternalVaultProxyMetadata::VgsMetadata(
                    external_services::grpc_client::unified_connector_service::VgsMetadata {
                        proxy_url: external_vault_metadata_parsed.proxy_url,
                        certificate: external_vault_metadata_parsed.certificate,
                    },
                ),
            }
        }
        api_enums::VaultConnectors::HyperswitchVault => {
            let base = &connectors.hyperswitch_vault.base_url;
            let vault_endpoint_url = format!("{}/proxy", base)
                .parse::<url::Url>()
                .map(common_utils::types::Url::wrap)
                .change_context(UnifiedConnectorServiceError::ParsingFailed)
                .attach_printable("Failed to parse hyperswitch_vault endpoint URL")?;

            let auth_type: ConnectorAuthType = external_vault_merchant_connector_account
                .get_connector_account_details()
                .change_context(UnifiedConnectorServiceError::FailedToObtainAuthType)
                .attach_printable("Failed to obtain ConnectorAuthType for HyperswitchVault")?;

            let (api_key, profile_id) = match &auth_type {
                ConnectorAuthType::SignatureKey {
                    api_key,
                    api_secret,
                    ..
                } => (api_key.clone(), api_secret.clone()),
                _ => {
                    return Err(error_stack::report!(
                        UnifiedConnectorServiceError::FailedToObtainAuthType
                    )
                    .attach_printable("HyperswitchVault requires SignatureKey auth type"))
                }
            };

            external_services::grpc_client::unified_connector_service::ExternalVaultProxyConfig {
                vault_connector_type:
                    external_services::grpc_client::unified_connector_service::VaultConnectorType::Transformation,
                vault_connector_id: Some("hyperswitch_vault".to_string()),
                metadata: external_services::grpc_client::unified_connector_service::ExternalVaultProxyMetadata::HyperswitchVaultMetadata(
                    external_services::grpc_client::unified_connector_service::HyperswitchVaultMetadata {
                        vault_endpoint: vault_endpoint_url,
                        vault_auth_data: external_services::grpc_client::unified_connector_service::VaultConnectorAuth {
                            api_key,
                            profile_id,
                        },
                    },
                ),
            }
        }
        api_enums::VaultConnectors::Tokenex => Err(error_stack::report!(
            UnifiedConnectorServiceError::NotImplemented(
                "External vault proxy metadata is not supported for Tokenex".to_string(),
            )
        ))?,
    };

    logger::info!(external_vault_proxy_config = ?external_vault_proxy_config, "Built ExternalVaultProxyConfig (v2)");

    let external_vault_config_bytes = serde_json::to_vec(&external_vault_proxy_config)
        .change_context(UnifiedConnectorServiceError::ParsingFailed)
        .attach_printable("Failed to serialize ExternalVaultProxyConfig to bytes")?;

    Ok(BASE64_ENGINE.encode(&external_vault_config_bytes))
}

pub fn handle_unified_connector_service_response_for_payment_authorize(
    response: PaymentServiceAuthorizeResponse,
    prev_status: AttemptStatus,
) -> CustomResult<UcsPaymentAuthorizeResponseData, UnifiedConnectorServiceError> {
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let router_data_response =
        Result::<(PaymentsResponseData, AttemptStatus), ErrorResponse>::foreign_try_from((
            response.clone(),
            prev_status,
        ))?;

    let connector_customer_id =
        extract_connector_customer_id_from_ucs_state(response.state.as_ref());
    let connector_response =
        extract_connector_response_from_ucs(response.connector_response.as_ref());

    Ok(UcsPaymentAuthorizeResponseData {
        router_data_response,
        status_code,
        connector_customer_id,
        connector_response,
    })
}

pub fn handle_unified_connector_service_response_for_create_connector_customer(
    response: payments_grpc::CustomerServiceCreateResponse,
) -> CustomResult<(Result<PaymentsResponseData, ErrorResponse>, u16), UnifiedConnectorServiceError>
{
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let connector_customer_result =
        Result::<PaymentsResponseData, ErrorResponse>::foreign_try_from(response)?;

    Ok((connector_customer_result, status_code))
}

pub fn handle_unified_connector_service_response_for_payment_create_order(
    response: payments_grpc::PaymentServiceCreateOrderResponse,
    prev_status: AttemptStatus,
) -> UnifiedConnectorServiceCreateOrderResult {
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let router_data_response =
        Result::<(PaymentsResponseData, AttemptStatus), ErrorResponse>::foreign_try_from((
            response,
            prev_status,
        ))?;

    Ok((router_data_response, status_code))
}

pub fn handle_unified_connector_service_response_for_payment_post_authenticate(
    response: payments_grpc::PaymentMethodAuthenticationServicePostAuthenticateResponse,
    prev_status: AttemptStatus,
) -> UnifiedConnectorServiceResult {
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let router_data_response =
        Result::<(PaymentsResponseData, AttemptStatus), ErrorResponse>::foreign_try_from((
            response,
            prev_status,
        ))?;

    Ok((router_data_response, status_code))
}

pub fn handle_unified_connector_service_response_for_payment_method_tokenize(
    response: payments_grpc::PaymentMethodServiceTokenizeResponse,
) -> CustomResult<(Result<PaymentsResponseData, ErrorResponse>, u16), UnifiedConnectorServiceError>
{
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let router_data_response =
        Result::<PaymentsResponseData, ErrorResponse>::foreign_try_from(response)?;

    Ok((router_data_response, status_code))
}

pub fn handle_unified_connector_service_response_for_create_sdk_session_token(
    response: payments_grpc::MerchantAuthenticationServiceCreateClientAuthenticationTokenResponse,
) -> CustomResult<(Result<PaymentsResponseData, ErrorResponse>, u16), UnifiedConnectorServiceError>
{
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let router_data_response =
        Result::<PaymentsResponseData, ErrorResponse>::foreign_try_from(response)?;

    Ok((router_data_response, status_code))
}

pub fn handle_unified_connector_service_response_for_payment_incremental_authorization(
    response: payments_grpc::PaymentServiceIncrementalAuthorizationResponse,
) -> CustomResult<(Result<PaymentsResponseData, ErrorResponse>, u16), UnifiedConnectorServiceError>
{
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let router_data_response =
        Result::<PaymentsResponseData, ErrorResponse>::foreign_try_from(response)?;

    Ok((router_data_response, status_code))
}

pub fn handle_unified_connector_service_response_for_payment_authenticate(
    response: payments_grpc::PaymentMethodAuthenticationServiceAuthenticateResponse,
    prev_status: AttemptStatus,
) -> UnifiedConnectorServiceResult {
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let router_data_response =
        Result::<(PaymentsResponseData, AttemptStatus), ErrorResponse>::foreign_try_from((
            response,
            prev_status,
        ))?;

    Ok((router_data_response, status_code))
}

pub fn handle_unified_connector_service_response_for_payment_pre_authenticate(
    response: payments_grpc::PaymentMethodAuthenticationServicePreAuthenticateResponse,
    prev_status: AttemptStatus,
) -> UnifiedConnectorServiceResult {
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let router_data_response =
        Result::<(PaymentsResponseData, AttemptStatus), ErrorResponse>::foreign_try_from((
            response,
            prev_status,
        ))?;

    Ok((router_data_response, status_code))
}

pub fn handle_unified_connector_service_response_for_payment_capture(
    response: payments_grpc::PaymentServiceCaptureResponse,
    prev_status: AttemptStatus,
) -> CustomResult<UcsPaymentCaptureResponseData, UnifiedConnectorServiceError> {
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let router_data_response =
        Result::<(PaymentsResponseData, AttemptStatus), ErrorResponse>::foreign_try_from((
            response.clone(),
            prev_status,
        ))?;

    let connector_response =
        extract_connector_response_from_ucs(response.connector_response.as_ref());

    Ok(UcsPaymentCaptureResponseData {
        router_data_response,
        status_code,
        connector_response,
    })
}

pub fn handle_unified_connector_service_response_for_payment_setup_recurring(
    response: payments_grpc::PaymentServiceSetupRecurringResponse,
    prev_status: AttemptStatus,
) -> CustomResult<UcsPaymentSetupRecurringResponseData, UnifiedConnectorServiceError> {
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let router_data_response =
        Result::<(PaymentsResponseData, AttemptStatus), ErrorResponse>::foreign_try_from((
            response.clone(),
            prev_status,
        ))?;

    let connector_customer_id =
        extract_connector_customer_id_from_ucs_state(response.state.as_ref());
    let connector_response =
        extract_connector_response_from_ucs(response.connector_response.as_ref());

    Ok(UcsPaymentSetupRecurringResponseData {
        router_data_response,
        status_code,
        connector_customer_id,
        connector_response,
        amount_captured: response.captured_amount,
        minor_amount_captured: response
            .captured_amount
            .map(common_utils::types::MinorUnit::new),
    })
}

pub fn handle_unified_connector_service_response_for_create_session_token(
    response: payments_grpc::MerchantAuthenticationServiceCreateServerSessionAuthenticationTokenResponse,
) -> CustomResult<(Result<PaymentsResponseData, ErrorResponse>, u16), UnifiedConnectorServiceError>
{
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let router_data_response =
        Result::<PaymentsResponseData, ErrorResponse>::foreign_try_from(response)?;

    Ok((router_data_response, status_code))
}

pub fn handle_unified_connector_service_response_for_recurring_payment_charge(
    response: payments_grpc::RecurringPaymentServiceChargeResponse,
    prev_status: AttemptStatus,
) -> CustomResult<UcsRecurringPaymentChargeResponseData, UnifiedConnectorServiceError> {
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let router_data_response =
        Result::<(PaymentsResponseData, AttemptStatus), ErrorResponse>::foreign_try_from((
            response.clone(),
            prev_status,
        ))?;

    let connector_customer_id =
        extract_connector_customer_id_from_ucs_state(response.state.as_ref());
    let connector_response =
        extract_connector_response_from_ucs(response.connector_response.as_ref());

    Ok(UcsRecurringPaymentChargeResponseData {
        router_data_response,
        status_code,
        connector_customer_id,
        connector_response,
    })
}

/// Extracts connector_customer_id from UCS state
pub fn extract_connector_customer_id_from_ucs_state(
    ucs_state: Option<&payments_grpc::ConnectorState>,
) -> Option<String> {
    ucs_state.and_then(|state| {
        state
            .connector_customer_id
            .as_ref()
            .map(|id| id.to_string())
    })
}

/// Extracts connector_response from UCS response
pub fn extract_connector_response_from_ucs(
    connector_response: Option<&payments_grpc::ConnectorResponseData>,
) -> Option<hyperswitch_domain_models::router_data::ConnectorResponseData> {
    connector_response.and_then(|data| {
        <hyperswitch_domain_models::router_data::ConnectorResponseData as hyperswitch_interfaces::helpers::ForeignTryFrom<payments_grpc::ConnectorResponseData>>::foreign_try_from(data.clone())
            .map_err(|e| {
                logger::warn!(
                    error=?e,
                    "Failed to deserialize connector_response from UCS"
                );
                e
            })
            .ok()
    })
}

pub fn handle_unified_connector_service_response_for_payment_refund(
    response: payments_grpc::RefundResponse,
    prev_status: common_enums::RefundStatus,
) -> UnifiedConnectorServiceRefundResult {
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let router_data_response: Result<RefundsResponseData, ErrorResponse> =
        Result::<RefundsResponseData, ErrorResponse>::foreign_try_from((response, prev_status))?;

    Ok((router_data_response, status_code))
}

pub fn handle_unified_connector_service_response_for_refund_get(
    response: payments_grpc::RefundResponse,
    prev_status: common_enums::RefundStatus,
) -> UnifiedConnectorServiceRefundResult {
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let router_data_response =
        Result::<RefundsResponseData, ErrorResponse>::foreign_try_from((response, prev_status))?;

    Ok((router_data_response, status_code))
}

pub fn handle_unified_connector_service_response_for_payment_cancel(
    response: payments_grpc::PaymentServiceVoidResponse,
    prev_status: AttemptStatus,
) -> UnifiedConnectorServiceResult {
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let router_data_response =
        Result::<(PaymentsResponseData, AttemptStatus), ErrorResponse>::foreign_try_from((
            response,
            prev_status,
        ))?;

    Ok((router_data_response, status_code))
}

/// Handles the unified connector service response for create access token
pub fn handle_unified_connector_service_response_for_create_access_token(
    response: payments_grpc::MerchantAuthenticationServiceCreateServerAuthenticationTokenResponse,
) -> CustomResult<(Result<AccessToken, ErrorResponse>, u16), UnifiedConnectorServiceError> {
    let status_code = transformers::convert_connector_service_status_code(response.status_code)?;

    let access_token_result = Result::<AccessToken, ErrorResponse>::foreign_try_from(response)?;

    Ok((access_token_result, status_code))
}

/// Handle UCS payout create response and transform it to router data format
impl
    ForeignTryFrom<(
        payments_grpc::PayoutServiceCreateResponse,
        common_enums::PayoutStatus,
    )> for crate::types::UcsPayoutCreateResponseData
{
    type Error = error_stack::Report<UnifiedConnectorServiceError>;

    fn foreign_try_from(
        value: (
            payments_grpc::PayoutServiceCreateResponse,
            common_enums::PayoutStatus,
        ),
    ) -> Result<Self, Self::Error> {
        let (response, prev_status) = value;

        let status_code =
            transformers::convert_connector_service_status_code(response.status_code)?;

        let router_data_response = Result::<PayoutsResponseData, ErrorResponse>::foreign_try_from(
            (response.clone(), prev_status),
        )?;

        Ok(Self {
            router_data_response,
            status_code,
        })
    }
}

impl
    ForeignTryFrom<(
        payments_grpc::PayoutServiceTransferResponse,
        common_enums::PayoutStatus,
    )> for crate::types::UcsPayoutTransferResponseData
{
    type Error = error_stack::Report<UnifiedConnectorServiceError>;

    fn foreign_try_from(
        value: (
            payments_grpc::PayoutServiceTransferResponse,
            common_enums::PayoutStatus,
        ),
    ) -> Result<Self, Self::Error> {
        let (response, prev_status) = value;

        let status_code =
            transformers::convert_connector_service_status_code(response.status_code)?;

        let router_data_response = Result::<PayoutsResponseData, ErrorResponse>::foreign_try_from(
            (response.clone(), prev_status),
        )?;

        Ok(Self {
            router_data_response,
            status_code,
        })
    }
}

impl
    ForeignTryFrom<(
        payments_grpc::PayoutMethodEligibilityResponse,
        common_enums::PayoutStatus,
    )> for crate::types::UcsPayoutEligibilityResponseData
{
    type Error = error_stack::Report<UnifiedConnectorServiceError>;

    fn foreign_try_from(
        value: (
            payments_grpc::PayoutMethodEligibilityResponse,
            common_enums::PayoutStatus,
        ),
    ) -> Result<Self, Self::Error> {
        let (response, prev_status) = value;

        let status_code =
            transformers::convert_connector_service_status_code(response.status_code)?;

        let router_data_response = Result::<PayoutsResponseData, ErrorResponse>::foreign_try_from(
            (response.clone(), prev_status),
        )?;

        Ok(Self {
            router_data_response,
            status_code,
        })
    }
}

impl
    ForeignTryFrom<(
        payments_grpc::PayoutServiceGetResponse,
        common_enums::PayoutStatus,
    )> for crate::types::UcsPayoutGetResponseData
{
    type Error = error_stack::Report<UnifiedConnectorServiceError>;

    fn foreign_try_from(
        value: (
            payments_grpc::PayoutServiceGetResponse,
            common_enums::PayoutStatus,
        ),
    ) -> Result<Self, Self::Error> {
        let (response, prev_status) = value;

        let status_code =
            transformers::convert_connector_service_status_code(response.status_code)?;

        let router_data_response = Result::<PayoutsResponseData, ErrorResponse>::foreign_try_from(
            (response.clone(), prev_status),
        )?;

        Ok(Self {
            router_data_response,
            status_code,
        })
    }
}

impl
    ForeignTryFrom<(
        payments_grpc::PayoutServiceVoidResponse,
        common_enums::PayoutStatus,
    )> for crate::types::UcsPayoutVoidResponseData
{
    type Error = error_stack::Report<UnifiedConnectorServiceError>;

    fn foreign_try_from(
        value: (
            payments_grpc::PayoutServiceVoidResponse,
            common_enums::PayoutStatus,
        ),
    ) -> Result<Self, Self::Error> {
        let (response, prev_status) = value;

        let status_code =
            transformers::convert_connector_service_status_code(response.status_code)?;

        let router_data_response = Result::<PayoutsResponseData, ErrorResponse>::foreign_try_from(
            (response.clone(), prev_status),
        )?;

        Ok(Self {
            router_data_response,
            status_code,
        })
    }
}

impl
    ForeignTryFrom<(
        payments_grpc::PayoutServiceStageResponse,
        common_enums::PayoutStatus,
    )> for crate::types::UcsPayoutStageResponseData
{
    type Error = error_stack::Report<UnifiedConnectorServiceError>;

    fn foreign_try_from(
        value: (
            payments_grpc::PayoutServiceStageResponse,
            common_enums::PayoutStatus,
        ),
    ) -> Result<Self, Self::Error> {
        let (response, prev_status) = value;

        let status_code =
            transformers::convert_connector_service_status_code(response.status_code)?;

        let router_data_response = Result::<PayoutsResponseData, ErrorResponse>::foreign_try_from(
            (response.clone(), prev_status),
        )?;

        Ok(Self {
            router_data_response,
            status_code,
        })
    }
}

impl
    ForeignTryFrom<(
        payments_grpc::PayoutServiceCreateRecipientResponse,
        common_enums::PayoutStatus,
    )> for crate::types::UcsPayoutCreateRecipientResponseData
{
    type Error = error_stack::Report<UnifiedConnectorServiceError>;

    fn foreign_try_from(
        value: (
            payments_grpc::PayoutServiceCreateRecipientResponse,
            common_enums::PayoutStatus,
        ),
    ) -> Result<Self, Self::Error> {
        let (response, prev_status) = value;

        let status_code =
            transformers::convert_connector_service_status_code(response.status_code)?;

        let router_data_response = Result::<PayoutsResponseData, ErrorResponse>::foreign_try_from(
            (response.clone(), prev_status),
        )?;

        Ok(Self {
            router_data_response,
            status_code,
        })
    }
}

impl
    ForeignTryFrom<(
        payments_grpc::PayoutServiceEnrollDisburseAccountResponse,
        common_enums::PayoutStatus,
    )> for crate::types::UcsPayoutEnrollDisburseAccountResponseData
{
    type Error = error_stack::Report<UnifiedConnectorServiceError>;

    fn foreign_try_from(
        value: (
            payments_grpc::PayoutServiceEnrollDisburseAccountResponse,
            common_enums::PayoutStatus,
        ),
    ) -> Result<Self, Self::Error> {
        let (response, prev_status) = value;

        let status_code =
            transformers::convert_connector_service_status_code(response.status_code)?;

        let router_data_response = Result::<PayoutsResponseData, ErrorResponse>::foreign_try_from(
            (response.clone(), prev_status),
        )?;

        Ok(Self {
            router_data_response,
            status_code,
        })
    }
}

pub fn build_webhook_secrets_from_merchant_connector_account(
    #[cfg(feature = "v1")] merchant_connector_account: &MerchantConnectorAccountType,
    #[cfg(feature = "v2")] merchant_connector_account: &MerchantConnectorAccountTypeDetails,
) -> CustomResult<Option<payments_grpc::WebhookSecrets>, UnifiedConnectorServiceError> {
    // Extract webhook credentials from merchant connector account
    // This depends on how webhook secrets are stored in the merchant connector account

    #[cfg(feature = "v1")]
    let webhook_details = merchant_connector_account
        .get_webhook_details()
        .map_err(|_| UnifiedConnectorServiceError::FailedToObtainAuthType)?;

    #[cfg(feature = "v2")]
    let webhook_details = match merchant_connector_account {
        MerchantConnectorAccountTypeDetails::MerchantConnectorAccount(mca) => {
            mca.connector_webhook_details.as_ref()
        }
        MerchantConnectorAccountTypeDetails::MerchantConnectorDetails(_) => None,
    };

    match webhook_details {
        Some(details) => {
            // Parse the webhook details JSON to extract secrets
            let webhook_details: admin::MerchantConnectorWebhookDetails = details
                .clone()
                .parse_value("MerchantConnectorWebhookDetails")
                .change_context(UnifiedConnectorServiceError::FailedToObtainAuthType)
                .attach_printable("Failed to parse MerchantConnectorWebhookDetails")?;

            // Build gRPC WebhookSecrets from parsed details
            Ok(Some(payments_grpc::WebhookSecrets {
                secret: webhook_details.merchant_secret.expose().to_string(),
                additional_secret: webhook_details
                    .additional_secret
                    .map(|secret| secret.expose().to_string()),
            }))
        }
        None => Ok(None),
    }
}

/// Emit the connector event for a Hyperswitch -> UCS gRPC call, tagged
/// `destination = unified_connector_service` (same `connector_events` stream as a direct call).
#[allow(clippy::too_many_arguments)]
fn emit_ucs_connector_event(
    state: &SessionState,
    flow_type: &'static str,
    connector_name: String,
    payment_id: String,
    merchant_id: id_type::MerchantId,
    refund_id: Option<String>,
    dispute_id: Option<String>,
    payout_id: Option<String>,
    grpc_request_body: serde_json::Value,
    status_code: u16,
    response_body: Option<serde_json::Value>,
    external_latency: u128,
    execution_mode: ExecutionMode,
) {
    let mut connector_event = ConnectorEvent::new(
        state.tenant.tenant_id.clone(),
        connector_name,
        flow_type,
        grpc_request_body,
        "grpc://unified-connector-service".to_string(),
        Method::Post,
        payment_id,
        merchant_id,
        state.request_id.as_ref(),
        external_latency,
        refund_id,
        dispute_id,
        payout_id,
        status_code,
        common_enums::EventDestination::UnifiedConnectorService,
        common_enums::EventExecutionMode::from(execution_mode),
    );

    if let Some(body) = response_body {
        match status_code {
            400..=599 => connector_event.set_error_response_body(&body),
            _ => connector_event.set_response_body(&body),
        }
    }

    state.event_handler.log_event(&connector_event);
}

/// UCS Event Logging Wrapper Function
/// This function wraps UCS calls with comprehensive event logging.
/// It logs the actual gRPC request/response data, timing, and error information.
#[instrument(skip_all, fields(connector_name, flow_type, payment_id))]
pub async fn ucs_logging_wrapper<T, F, Fut, Req, Resp, GrpcReq, GrpcResp, FlowOutput>(
    router_data: RouterData<T, Req, Resp>,
    state: &SessionState,
    grpc_request: GrpcReq,
    grpc_header_builder: external_services::grpc_client::GrpcHeadersUcsBuilderFinal,
    execution_mode: ExecutionMode,
    handler: F,
) -> RouterResult<(RouterData<T, Req, Resp>, FlowOutput)>
where
    T: std::fmt::Debug + Clone + Send + 'static,
    Req: std::fmt::Debug + Clone + Send + Sync + 'static,
    Resp: std::fmt::Debug + Clone + Send + Sync + 'static,
    GrpcReq: serde::Serialize,
    GrpcResp: serde::Serialize,
    F: FnOnce(
            RouterData<T, Req, Resp>,
            GrpcReq,
            external_services::grpc_client::GrpcHeadersUcs,
        ) -> Fut
        + Send,
    Fut: std::future::Future<Output = RouterResult<(RouterData<T, Req, Resp>, FlowOutput, GrpcResp)>>
        + Send,
{
    tracing::Span::current().record("connector_name", &router_data.connector);
    tracing::Span::current().record("flow_type", std::any::type_name::<T>());
    tracing::Span::current().record("payment_id", &router_data.payment_id);

    // Capture request data for logging
    let connector_name = router_data.connector.clone();
    let payment_id = router_data.payment_id.clone();
    let merchant_id = router_data.merchant_id.clone();
    let refund_id = router_data.refund_id.clone();
    let dispute_id = router_data.dispute_id.clone();
    let payout_id = router_data.payout_id.clone();
    let grpc_header = grpc_header_builder.build();
    // Log the actual gRPC request with masking
    let grpc_request_body = hyperswitch_masking::masked_serialize(&grpc_request)
        .unwrap_or_else(|_| serde_json::json!({"error": "failed_to_serialize_grpc_request"}));

    // Update connector call count metrics for UCS operations
    crate::routes::metrics::CONNECTOR_CALL_COUNT.add(
        1,
        router_env::metric_attributes!(
            ("connector", connector_name.clone()),
            (
                "flow",
                std::any::type_name::<T>()
                    .split("::")
                    .last()
                    .unwrap_or_default()
            ),
        ),
    );

    // Execute UCS function and measure timing
    let start_time = Instant::now();
    let result = handler(router_data, grpc_request, grpc_header).await;
    let external_latency = start_time.elapsed().as_millis();

    // Create and emit connector event after UCS call
    let (status_code, response_body, router_result) = match result {
        Ok((updated_router_data, flow_output, grpc_response)) => {
            let status = updated_router_data
                .connector_http_status_code
                .unwrap_or(200);

            // Log the actual gRPC response with masking
            let grpc_response_body = hyperswitch_masking::masked_serialize(&grpc_response)
                .unwrap_or_else(
                    |_| serde_json::json!({"error": "failed_to_serialize_grpc_response"}),
                );

            (
                status,
                Some(grpc_response_body),
                Ok((updated_router_data, flow_output)),
            )
        }
        Err(error) => {
            // Update error metrics for UCS calls
            crate::routes::metrics::CONNECTOR_ERROR_RESPONSE_COUNT.add(
                1,
                router_env::metric_attributes!(("connector", connector_name.clone(),)),
            );

            let error_body = serde_json::json!({
                "error": error.to_string(),
                "error_type": "ucs_call_failed"
            });
            (
                error.current_context().status_code().as_u16(),
                Some(error_body),
                Err(error),
            )
        }
    };

    if let ExecutionMode::Primary = execution_mode {
        emit_ucs_connector_event(
            state,
            std::any::type_name::<T>(),
            connector_name,
            payment_id,
            merchant_id,
            refund_id,
            dispute_id,
            payout_id,
            grpc_request_body,
            status_code,
            response_body,
            external_latency,
            execution_mode,
        );
    }

    // Set external latency on router data
    router_result.map(|mut router_data| {
        router_data.0.external_latency =
            Some(router_data.0.external_latency.unwrap_or(0) + external_latency);
        router_data
    })
}

/// new UCS Event Logging Wrapper Function with UCS error response
/// This function wraps UCS calls with comprehensive event logging.
/// It logs the actual gRPC request/response data, timing, and error information.
#[instrument(skip_all, fields(connector_name, flow_type, payment_id))]
pub async fn ucs_logging_wrapper_granular<T, F, Fut, Req, Resp, GrpcReq, FlowOutput, GrpcResp>(
    router_data: RouterData<T, Req, Resp>,
    state: &SessionState,
    grpc_request: GrpcReq,
    grpc_header_builder: external_services::grpc_client::GrpcHeadersUcsBuilderFinal,
    execution_mode: ExecutionMode,
    handler: F,
) -> CustomResult<(RouterData<T, Req, Resp>, FlowOutput), UnifiedConnectorServiceError>
where
    T: std::fmt::Debug + Clone + Send + 'static,
    Req: std::fmt::Debug + Clone + Send + Sync + 'static,
    Resp: std::fmt::Debug + Clone + Send + Sync + 'static,
    GrpcReq: serde::Serialize,
    GrpcResp: serde::Serialize,
    F: FnOnce(
            RouterData<T, Req, Resp>,
            GrpcReq,
            external_services::grpc_client::GrpcHeadersUcs,
        ) -> Fut
        + Send,
    Fut: std::future::Future<
            Output = CustomResult<
                (RouterData<T, Req, Resp>, FlowOutput, GrpcResp),
                UnifiedConnectorServiceError,
            >,
        > + Send,
{
    tracing::Span::current().record("connector_name", &router_data.connector);
    tracing::Span::current().record("flow_type", std::any::type_name::<T>());
    tracing::Span::current().record("payment_id", &router_data.payment_id);

    // Capture request data for logging
    let connector_name = router_data.connector.clone();
    let payment_id = router_data.payment_id.clone();
    let merchant_id = router_data.merchant_id.clone();
    let refund_id = router_data.refund_id.clone();
    let dispute_id = router_data.dispute_id.clone();
    let payout_id = router_data.payout_id.clone();
    let grpc_header = grpc_header_builder.build();
    // Log the actual gRPC request with masking
    let grpc_request_body = hyperswitch_masking::masked_serialize(&grpc_request)
        .unwrap_or_else(|_| serde_json::json!({"error": "failed_to_serialize_grpc_request"}));

    // Update connector call count metrics for UCS operations
    crate::routes::metrics::CONNECTOR_CALL_COUNT.add(
        1,
        router_env::metric_attributes!(
            ("connector", connector_name.clone()),
            (
                "flow",
                std::any::type_name::<T>()
                    .split("::")
                    .last()
                    .unwrap_or_default()
            ),
        ),
    );

    // Execute UCS function and measure timing
    let start_time = Instant::now();
    let result = handler(router_data, grpc_request, grpc_header).await;
    let external_latency = start_time.elapsed().as_millis();

    // Create and emit connector event after UCS call
    let (status_code, response_body, router_result) = match result {
        Ok((updated_router_data, flow_output, grpc_response)) => {
            let status = updated_router_data
                .connector_http_status_code
                .unwrap_or(200);

            // Log the actual gRPC response
            let grpc_response_body = hyperswitch_masking::masked_serialize(&grpc_response)
                .unwrap_or_else(
                    |_| serde_json::json!({"error": "failed_to_serialize_grpc_response"}),
                );

            (
                status,
                Some(grpc_response_body),
                Ok((updated_router_data, flow_output)),
            )
        }
        Err(error) => {
            // Update error metrics for UCS calls
            crate::routes::metrics::CONNECTOR_ERROR_RESPONSE_COUNT.add(
                1,
                router_env::metric_attributes!(("connector", connector_name.clone(),)),
            );

            let error_body = serde_json::json!({
                "error": error.to_string(),
                "error_type": "ucs_call_failed"
            });
            (
                error.current_context().http_status(),
                Some(error_body),
                Err(error),
            )
        }
    };

    if let ExecutionMode::Primary = execution_mode {
        emit_ucs_connector_event(
            state,
            std::any::type_name::<T>(),
            connector_name,
            payment_id,
            merchant_id,
            refund_id,
            dispute_id,
            payout_id,
            grpc_request_body,
            status_code,
            response_body,
            external_latency,
            execution_mode,
        );
    }

    // Set external latency on router data
    router_result.map(|mut router_data| {
        router_data.0.external_latency =
            Some(router_data.0.external_latency.unwrap_or(0) + external_latency);
        router_data
    })
}

/// UCS wrapper for webhook flows. It centralizes header building and handler
/// invocation plus masked request/response logging, and takes no
/// `RouterData` — incoming webhooks don't have one at the UCS call site. `flow` is
/// the gRPC method label (e.g. `"EventServiceParseEvent"`).
#[allow(clippy::too_many_arguments)]
#[instrument(skip_all, fields(connector_name, flow_type))]
pub async fn ucs_webhook_logging_wrapper<F, Fut, GrpcReq, GrpcResp>(
    _state: &SessionState,
    connector_name: String,
    flow: &'static str,
    grpc_request: GrpcReq,
    grpc_header_builder: external_services::grpc_client::GrpcHeadersUcsBuilderFinal,
    handler: F,
) -> RouterResult<GrpcResp>
where
    GrpcReq: serde::Serialize,
    GrpcResp: serde::Serialize,
    F: FnOnce(GrpcReq, external_services::grpc_client::GrpcHeadersUcs) -> Fut + Send,
    Fut: std::future::Future<Output = RouterResult<GrpcResp>> + Send,
{
    tracing::Span::current().record("connector_name", &connector_name);
    tracing::Span::current().record("flow_type", flow);

    let grpc_header = grpc_header_builder.build();
    let grpc_request_body =
        hyperswitch_masking::masked_serialize(&grpc_request).unwrap_or_else(|error| {
            logger::warn!(
                ?error,
                "Failed to mask-serialize UCS webhook gRPC request for logging"
            );
            serde_json::json!({"error": "failed_to_serialize_grpc_request"})
        });
    logger::info!(
        flow,
        ucs_webhook_grpc_request = ?grpc_request_body,
        "UCS webhook gRPC request"
    );

    let start_time = Instant::now();
    let result = handler(grpc_request, grpc_header).await;
    let external_latency = start_time.elapsed().as_millis();

    match result {
        Ok(grpc_response) => {
            let grpc_response_body = hyperswitch_masking::masked_serialize(&grpc_response)
                .unwrap_or_else(|error| {
                    logger::warn!(
                        ?error,
                        "Failed to mask-serialize UCS webhook gRPC response for logging"
                    );
                    serde_json::json!({"error": "failed_to_serialize_grpc_response"})
                });
            logger::info!(
                flow,
                external_latency,
                ucs_webhook_grpc_response = ?grpc_response_body,
                "UCS webhook gRPC response"
            );
            Ok(grpc_response)
        }
        Err(error) => {
            logger::warn!(
                flow,
                external_latency,
                ?error,
                "UCS webhook gRPC call failed"
            );
            Err(error)
        }
    }
}

// ============================================================================
// REFUND UCS FUNCTIONS
// ============================================================================

/// Execute UCS refund request using PaymentService.Refund gRPC method
#[instrument(skip_all)]
pub async fn call_unified_connector_service_for_refund_execute(
    state: &SessionState,
    processor: &Processor,
    router_data: RouterData<refunds::Execute, RefundsData, RefundsResponseData>,
    execution_mode: ExecutionMode,
    #[cfg(feature = "v1")] merchant_connector_account: MerchantConnectorAccountType,
    #[cfg(feature = "v2")] merchant_connector_account: MerchantConnectorAccountTypeDetails,
) -> RouterResult<RouterData<refunds::Execute, RefundsData, RefundsResponseData>> {
    // Get UCS client
    let ucs_client = get_ucs_client(state)?;

    // Build auth metadata using standard UCS function
    let connector_auth_metadata = build_unified_connector_service_auth_metadata(
        merchant_connector_account,
        processor.get_account().get_id(),
        router_data.connector.clone(),
    )
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to build UCS auth metadata for refund execute")?;

    // Transform router data to UCS refund request
    let ucs_refund_request =
        payments_grpc::PaymentServiceRefundRequest::foreign_try_from(&router_data)
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to transform router data to UCS refund request")?;

    // Build gRPC headers
    // Use merchant_id as profile_id fallback since RouterData doesn't have profile_id field
    let merchant_id = processor.get_account().get_id().clone();
    let profile_id = id_type::ProfileId::from_str(merchant_id.get_string_repr())
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to convert merchant_id to profile_id for UCS refund")?;
    let lineage_ids = LineageIds::new(merchant_id, profile_id);
    let merchant_reference_id =
        id_type::PaymentReferenceId::from_str(router_data.payment_id.as_str())
            .inspect_err(|err| logger::warn!(error=?err, "Invalid PaymentId for UCS reference id"))
            .ok()
            .map(ucs_types::UcsReferenceId::Payment);
    let resource_id = router_data
        .refund_id
        .as_ref()
        .and_then(|refund_id| {
            id_type::RefundReferenceId::try_from(Cow::Owned(refund_id.to_string()))
                .inspect_err(
                    |err| logger::warn!(error=?err, "Invalid RefundId for UCS resource id"),
                )
                .ok()
        })
        .map(ucs_types::UcsResourceId::Refund);
    let grpc_header_builder = state
        .get_grpc_headers_ucs(execution_mode)
        .lineage_ids(lineage_ids)
        .external_vault_proxy_metadata(None)
        .merchant_reference_id(merchant_reference_id)
        .resource_id(resource_id);

    let prev_refund_status = router_data.request.refund_status;

    // Make UCS refund call with logging wrapper
    Box::pin(ucs_logging_wrapper(
        router_data,
        state,
        ucs_refund_request,
        grpc_header_builder,
        execution_mode,
        |mut router_data, grpc_request, grpc_headers| async move {
            // Call UCS payment_refund method
            let response = match ucs_client
                .payment_refund(grpc_request, connector_auth_metadata, grpc_headers)
                .await
            {
                Ok(resp) => resp,
                Err(report) => {
                    if let UnifiedConnectorServiceError::ConnectorError(inner) =
                        report.current_context()
                    {
                        let (code, message, status_code, reason, connector,
                             network_decline_code, network_advice_code, network_error_message) = (
                            &inner.code, &inner.message, inner.status_code, &inner.reason,
                            &inner.connector, &inner.network_decline_code,
                            &inner.network_advice_code, &inner.network_error_message,
                        );
                        logger::info!(
                            "Connector error via UCS for refund execute (connector {}, status {}): {} - {}",
                            connector,
                            status_code,
                            code,
                            message
                        );
                        router_data.response = Err(ErrorResponse {
                            code: code.clone(),
                            message: message.clone(),
                            reason: reason.clone(),
                            status_code,
                            attempt_status: None,
                            connector_transaction_id: inner.connector_transaction_id.clone(),
                            connector_response_reference_id: None,
                            network_decline_code: network_decline_code.clone(),
                            network_advice_code: network_advice_code.clone(),
                            network_error_message: network_error_message.clone(),
                            connector_metadata: None,
                        });
                        router_data.connector_http_status_code = Some(status_code);
                        return Ok((router_data, (), payments_grpc::RefundResponse::default()));
                    }
                    let api_error = report.current_context().switch();
                    return Err(report
                        .change_context(api_error)
                        .attach_printable("UCS refund execution failed"));
                }
            };

            let grpc_response = response.into_inner();

            // Transform UCS response back to RouterData
            let (refund_response_data, status_code) =
                handle_unified_connector_service_response_for_payment_refund(
                    grpc_response.clone(),
                    prev_refund_status,
                )
                .change_context(errors::ApiErrorResponse::InternalServerError)
                .attach_printable("Failed to transform UCS refund response")?;

            router_data.response = refund_response_data;
            router_data.connector_http_status_code = Some(status_code);

            Ok((router_data, (), grpc_response))
        },
    ))
    .await
    .map(|(router_data, _flow_response)| router_data)
}

/// Execute UCS refund sync request using RefundService.Get gRPC method
#[instrument(skip_all)]
pub async fn call_unified_connector_service_for_refund_sync(
    state: &SessionState,
    processor: &Processor,
    router_data: RouterData<refunds::RSync, RefundsData, RefundsResponseData>,
    execution_mode: ExecutionMode,
    #[cfg(feature = "v1")] merchant_connector_account: MerchantConnectorAccountType,
    #[cfg(feature = "v2")] merchant_connector_account: MerchantConnectorAccountTypeDetails,
) -> RouterResult<RouterData<refunds::RSync, RefundsData, RefundsResponseData>> {
    // Get UCS client
    let ucs_client = get_ucs_client(state)?;

    // Build auth metadata using standard UCS function
    let connector_auth_metadata = build_unified_connector_service_auth_metadata(
        merchant_connector_account,
        processor.get_account().get_id(),
        router_data.connector.clone(),
    )
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to build UCS auth metadata for refund sync")?;

    // Transform router data to UCS refund sync request
    let ucs_refund_sync_request =
        payments_grpc::RefundServiceGetRequest::foreign_try_from(&router_data)
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Failed to transform router data to UCS refund sync request")?;

    // Build gRPC headers
    // Use merchant_id as profile_id fallback since RouterData doesn't have profile_id field
    let merchant_id = processor.get_account().get_id().clone();
    let profile_id = id_type::ProfileId::from_str(merchant_id.get_string_repr())
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to convert merchant_id to profile_id for UCS refund")?;
    let lineage_ids = LineageIds::new(merchant_id, profile_id);
    let merchant_reference_id =
        id_type::PaymentReferenceId::from_str(router_data.payment_id.as_str())
            .inspect_err(|err| logger::warn!(error=?err, "Invalid PaymentId for UCS reference id"))
            .ok()
            .map(ucs_types::UcsReferenceId::Payment);
    let resource_id = router_data
        .refund_id
        .as_ref()
        .and_then(|refund_id| {
            id_type::RefundReferenceId::try_from(Cow::Owned(refund_id.to_string()))
                .inspect_err(
                    |err| logger::warn!(error=?err, "Invalid RefundId for UCS resource id"),
                )
                .ok()
        })
        .map(ucs_types::UcsResourceId::Refund);

    let grpc_header_builder = state
        .get_grpc_headers_ucs(execution_mode)
        .lineage_ids(lineage_ids)
        .external_vault_proxy_metadata(None)
        .merchant_reference_id(merchant_reference_id)
        .resource_id(resource_id);

    let prev_refund_status = router_data.request.refund_status;

    // Make UCS refund sync call with logging wrapper
    Box::pin(ucs_logging_wrapper(
        router_data,
        state,
        ucs_refund_sync_request,
        grpc_header_builder,
        execution_mode,
        |mut router_data, grpc_request, grpc_headers| async move {
            // Call UCS refund_sync method
            let response = match ucs_client
                .refund_get(grpc_request, connector_auth_metadata, grpc_headers)
                .await
            {
                Ok(resp) => resp,
                Err(report) => {
                    if let UnifiedConnectorServiceError::ConnectorError(inner) =
                        report.current_context()
                    {
                        let (code, message, status_code, reason, connector,
                             network_decline_code, network_advice_code, network_error_message) = (
                            &inner.code, &inner.message, inner.status_code, &inner.reason,
                            &inner.connector, &inner.network_decline_code,
                            &inner.network_advice_code, &inner.network_error_message,
                        );
                        logger::info!(
                            "Connector error via UCS for refund sync (connector {}, status {}): {} - {}",
                            connector,
                            status_code,
                            code,
                            message
                        );
                        router_data.response = Err(ErrorResponse {
                            code: code.clone(),
                            message: message.clone(),
                            reason: reason.clone(),
                            status_code,
                            attempt_status: None,
                            connector_transaction_id: inner.connector_transaction_id.clone(),
                            connector_response_reference_id: None,
                            network_decline_code: network_decline_code.clone(),
                            network_advice_code: network_advice_code.clone(),
                            network_error_message: network_error_message.clone(),
                            connector_metadata: None,
                        });
                        router_data.connector_http_status_code = Some(status_code);
                        return Ok((router_data, (), payments_grpc::RefundResponse::default()));
                    }
                    let api_error = report.current_context().switch();
                    return Err(report
                        .change_context(api_error)
                        .attach_printable("UCS refund sync execution failed"));
                }
            };

            let grpc_response = response.into_inner();

            // Transform UCS response back to RouterData
            let (refund_response_data, status_code) =
                handle_unified_connector_service_response_for_refund_get(
                    grpc_response.clone(),
                    prev_refund_status,
                )
                .change_context(errors::ApiErrorResponse::InternalServerError)
                .attach_printable("Failed to transform UCS refund get response")?;

            router_data.response = refund_response_data;
            router_data.connector_http_status_code = Some(status_code);

            Ok((router_data, (), grpc_response))
        },
    ))
    .await
    .map(|(router_data, _flow_response)| router_data)
}

/// Execute a surcharge calculation call via UCS SurchargeService.Calculate.
///
/// This is a lightweight helper — it does not go through the full RouterData
/// machinery. It is called from `payments_submit_eligibility` after eligibility
/// checks pass.
#[cfg(all(feature = "oltp", feature = "v1"))]
#[instrument(skip_all, fields(connector_name))]
pub async fn call_unified_connector_service_for_surcharge_calculate(
    state: &SessionState,
    processor: &Processor,
    #[cfg(feature = "v1")] merchant_connector_account: MerchantConnectorAccountType,
    payments_surcharge_calculate_data: hyperswitch_domain_models::router_request_types::PaymentsSurchargeCalculationData,
    payment_id: &id_type::PaymentId,
    profile_id: &id_type::ProfileId,
    connector_name: String,
) -> RouterResult<hyperswitch_domain_models::router_response_types::SurchargeCalculationResponseData>
{
    let ucs_client = get_ucs_client(state)?;

    let connector_auth_metadata = build_unified_connector_service_auth_metadata(
        merchant_connector_account,
        processor.get_account().get_id(),
        connector_name.clone(),
    )
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to build UCS auth metadata for surcharge calculate")?;

    // Build the gRPC request directly
    let amount = payments_grpc::Money {
        minor_amount: payments_surcharge_calculate_data.amount.get_amount_as_i64(),
        currency: payments_grpc::Currency::foreign_try_from(
            payments_surcharge_calculate_data.currency,
        )
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to convert currency for surcharge gRPC request")?
        .into(),
    };
    let surcharge_request = payments_grpc::SurchargeServiceCalculateRequest {
        merchant_surcharge_id: Some(payment_id.get_string_repr().to_owned()),
        amount: Some(amount),
        card_bin: payments_surcharge_calculate_data.card_iin.clone(),
        postal_code: payments_surcharge_calculate_data.postal_code.clone(),
        previous_connector_surcharge_id: payments_surcharge_calculate_data
            .previous_connector_surcharge_id
            .clone(),
        country: payments_surcharge_calculate_data
            .country
            .as_ref()
            .and_then(|c| payments_grpc::CountryAlpha2::from_str_name(&c.to_string()))
            .map(|c| c.into()),
        surcharge_strategy: payments_surcharge_calculate_data
            .external_surcharge_strategy
            .as_ref()
            .map(|s| payments_grpc::SurchargeStrategy::foreign_from(*s).into()),
    };

    // Build gRPC headers
    let lineage_ids = LineageIds::new(processor.get_account().get_id().clone(), profile_id.clone());
    let merchant_reference_id = id_type::PaymentReferenceId::from_str(payment_id.get_string_repr())
        .inspect_err(
            |err| logger::warn!(error=?err, "Invalid PaymentId for surcharge UCS reference id"),
        )
        .ok()
        .map(ucs_types::UcsReferenceId::Payment);

    let grpc_header_builder = state
        .get_grpc_headers_ucs(ExecutionMode::Primary)
        .lineage_ids(lineage_ids)
        .external_vault_proxy_metadata(None)
        .merchant_reference_id(merchant_reference_id)
        .resource_id(None);
    let grpc_headers = grpc_header_builder.build();

    let response = ucs_client
        .surcharge_calculate(surcharge_request, connector_auth_metadata, grpc_headers)
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("UCS surcharge_calculate gRPC call failed")?;

    hyperswitch_domain_models::router_response_types::SurchargeCalculationResponseData::foreign_try_from(
        response.into_inner(),
    )
    .change_context(errors::ApiErrorResponse::InternalServerError)
    .attach_printable("Failed to parse UCS surcharge calculate response")
}

#[cfg(feature = "v1")]
fn extract_notify_connector_content_from_request(
    feature_data: FeatureTrackingData,
) -> CustomResult<payments_grpc::NotifyConnectorContent, errors::ApiErrorResponse> {
    match feature_data {
        FeatureTrackingData::SurchargeDetails(details) => {
            Ok(payments_grpc::NotifyConnectorContent {
                content: Some(
                    payments_grpc::notify_connector_content::Content::SurchargeContent(
                        payments_grpc::SurchargeContent {
                            connector_surcharge_id: details.external_surcharge_id.clone(),
                        },
                    ),
                ),
            })
        }
    }
}

#[cfg(all(feature = "oltp", feature = "v1"))]
#[instrument(skip_all, fields(connector_name))]
pub fn build_notify_connector_request(
    event: &Event,
    merchant_id: &id_type::MerchantId,
    merchant_connector_account: MerchantConnectorAccountType,
    connector_name: String,
    notify_feature_data: FeatureTrackingData,
) -> RouterResult<(
    payments_grpc::NotifyConnectorRequest,
    ConnectorAuthMetadata,
    payments_grpc::NotifyEventType,
)> {
    let connector_auth_metadata: ConnectorAuthMetadata =
        build_unified_connector_service_auth_metadata(
            merchant_connector_account,
            merchant_id,
            connector_name.clone(),
        )
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to build UCS auth metadata for surcharge calculate")?;

    let notify_event_type = match event.event_type {
        api_models::enums::EventType::SurchargePaymentSucceeded => {
            payments_grpc::NotifyEventType::SurchargePaymentSucceeded
        }
        api_models::enums::EventType::SurchargeRefundSucceeded => {
            payments_grpc::NotifyEventType::SurchargeRefundSucceeded
        }
        _ => {
            return Err(errors::ApiErrorResponse::InternalServerError).attach_printable(format!(
                "Unsupported event type for connector notification: {:?}",
                event.event_type
            ))
        }
    };

    // Build NotifyConnectorRequest
    Ok((
        payments_grpc::NotifyConnectorRequest {
            event_id: event.event_id.clone(),
            event_type: notify_event_type.into(),
            content: Some(extract_notify_connector_content_from_request(
                notify_feature_data,
            )?),
            timestamp: event.created_at.assume_utc().unix_timestamp(),
            state: None,
        },
        connector_auth_metadata,
        notify_event_type,
    ))
}

/// Execute a notify connector call via UCS EventService.NotifyConnector.
///
/// This sends event notifications (e.g. surcharge payment succeeded) to the
/// unified connector service so that downstream connectors can act on them.
/// It does not go through the full RouterData machinery.
#[cfg(all(feature = "oltp", feature = "v1"))]
#[instrument(skip_all, fields(connector_name))]
pub async fn call_unified_connector_service_for_notify_connector(
    state: &SessionState,
    event: &Event,
    connector_auth_metadata: ConnectorAuthMetadata,
    notify_request: payments_grpc::NotifyConnectorRequest,
    notify_event_type: payments_grpc::NotifyEventType,
    merchant_id: &id_type::MerchantId,
    profile_id: &id_type::ProfileId,
) -> CustomResult<
    hyperswitch_domain_models::router_response_types::NotifyConnectorResponseData,
    errors::ApiClientError,
> {
    let lineage_ids = LineageIds::new(merchant_id.clone(), profile_id.clone());
    let merchant_reference_id = id_type::PaymentReferenceId::from_str(&event.primary_object_id)
        .inspect_err(
            |err| logger::warn!(error=?err, "Invalid primary_object_id for UCS reference id"),
        )
        .ok()
        .map(ucs_types::UcsReferenceId::Payment);

    let grpc_header_builder = state
        .get_grpc_headers_ucs(ExecutionMode::Primary)
        .lineage_ids(lineage_ids)
        .external_vault_proxy_metadata(None)
        .merchant_reference_id(merchant_reference_id)
        .resource_id(None);

    let grpc_headers = grpc_header_builder.build();
    let ucs_client =
        get_ucs_client(state).change_context(errors::ApiClientError::ClientConstructionFailed)?;

    let response = ucs_client
        .notify_connector(
            notify_request,
            connector_auth_metadata,
            grpc_headers,
            notify_event_type,
        )
        .await
        .change_context(errors::ApiClientError::UnexpectedServerResponse)
        .attach_printable("UCS notify_connector gRPC call failed")?;

    let notify_connector_response_data =
        hyperswitch_domain_models::router_response_types::NotifyConnectorResponseData::foreign_try_from(
            response.into_inner(),
        )
        .change_context(errors::ApiClientError::ResponseDecodingFailed)
        .attach_printable("Failed to parse UCS notify_connector response")?;

    Ok(notify_connector_response_data)
}
