use std::str::FromStr;

#[cfg(feature = "payouts")]
use api_models::payouts as payout_models;
use api_models::{
    enums::Connector,
    webhooks::{self, WebhookResponseTracker},
};
pub use common_enums::{connector_enums::InvoiceStatus, enums::ProcessTrackerRunner};
use common_utils::{
    errors::ReportSwitchExt,
    events::ApiEventsType,
    ext_traits::{AsyncExt, ByteSliceExt},
    types::{AmountConvertor, StringMinorUnitForConnector},
};
use diesel_models::{refund as diesel_refund, ConnectorMandateReferenceId};
use error_stack::{report, ResultExt};
use hyperswitch_connectors::connectors::unified_authentication_service::transformers::WebhookResponse as UasWebhookResponse;
#[cfg(feature = "payouts")]
use hyperswitch_domain_models::payouts::payouts::PayoutsUpdate;
use hyperswitch_domain_models::{
    api::{IncomingWebhookEventMetadata, WebhookResponse},
    mandates::CommonMandateReference,
    payments::{payment_attempt::PaymentAttempt, HeaderPayload},
    router_request_types::unified_authentication_service::UasAuthenticationResponseData,
};
use hyperswitch_interfaces::webhooks::{
    IncomingWebhookFlowError, IncomingWebhookRequestDetails, WebhookContext, WebhookResourceData,
};
use hyperswitch_masking::{ExposeInterface, PeekInterface, Secret};
use router_env::{instrument, tracing};

use super::{types, MERCHANT_ID};
use crate::{
    consts,
    core::{
        api_locking,
        configs::dimension_state,
        errors::{self, CustomResult, RouterResponse, StorageErrorExt},
        metrics,
        payment_methods::{self, cards},
        payments::{self, tokenization, PaymentIntentStateMetadataExt},
        refunds, relay,
        unified_authentication_service::{
            types::UNIFIED_AUTHENTICATION_SERVICE, utils as uas_utils,
        },
        unified_connector_service, utils as core_utils,
        webhooks::{network_tokenization_incoming, utils},
    },
    logger,
    routes::{app::ReqState, lock_utils, SessionState},
    services::{self, connector_integration_interface::ConnectorEnum, ConnectorValidation},
    types::{
        api::{
            self, mandates::MandateResponseExt, ConnectorCommon, ConnectorData, GetToken,
            IncomingWebhook,
        },
        domain,
        storage::{self, enums},
        transformers::{ForeignFrom, ForeignInto, ForeignTryFrom},
    },
    utils::{self as helper_utils, ext_traits::OptionExt, generate_id},
};
#[cfg(feature = "payouts")]
use crate::{core::payouts, types::storage::PayoutAttemptUpdate};

#[allow(clippy::too_many_arguments)]
pub async fn incoming_webhooks_wrapper<W: types::OutgoingWebhookType>(
    state: SessionState,
    req_state: ReqState,
    req: &actix_web::HttpRequest,
    platform: domain::Platform,
    connector_name_or_mca_id: &str,
    body: actix_web::web::Bytes,
    is_relay_webhook: bool,
) -> RouterResponse<serde_json::Value> {
    let (webhook_response, webhooks_response_tracker, serialized_req) =
        Box::pin(incoming_webhooks_core::<W>(
            state.clone(),
            req_state,
            req,
            platform.clone(),
            connector_name_or_mca_id,
            body.clone(),
            is_relay_webhook,
        ))
        .await?;

    logger::info!(incoming_webhook_payload = ?serialized_req.peek());

    let metadata = IncomingWebhookEventMetadata {
        event_type: ApiEventsType::Webhooks {
            connector: connector_name_or_mca_id.to_string(),
            payment_id: webhooks_response_tracker.get_payment_id(),
            refund_id: webhooks_response_tracker.get_refund_id(),
        },
        serialized_request: serialized_req,
        webhook_tracker_data: serde_json::to_value(&webhooks_response_tracker)
            .inspect_err(
                |err| logger::error!(error = ?err, "Could not convert webhook effect to string"),
            )
            .ok(),
    };

    Ok(services::ApplicationResponse::IncomingWebhookEvent {
        response: Box::new(webhook_response),
        metadata,
    })
}

#[cfg(feature = "v1")]
pub async fn network_token_incoming_webhooks_wrapper<W: types::OutgoingWebhookType>(
    state: SessionState,
    req: &actix_web::HttpRequest,
    body: actix_web::web::Bytes,
) -> RouterResponse<serde_json::Value> {
    let request_details: IncomingWebhookRequestDetails<'_> = IncomingWebhookRequestDetails {
        method: req.method().clone(),
        uri: req.uri().clone(),
        headers: req.headers(),
        query_params: req.query_string().to_string(),
        body: &body,
    };

    let (webhook_response, webhooks_response_tracker, serialized_req, _merchant_id) = Box::pin(
        network_token_incoming_webhooks_core::<W>(&state, request_details),
    )
    .await?;

    logger::info!(incoming_webhook_payload = ?serialized_req);

    let metadata = IncomingWebhookEventMetadata {
        event_type: ApiEventsType::NetworkTokenWebhook {
            payment_method_id: webhooks_response_tracker.get_payment_method_id(),
        },
        serialized_request: Secret::new(serialized_req),
        webhook_tracker_data: serde_json::to_value(&webhooks_response_tracker)
            .inspect_err(
                |err| logger::error!(error = ?err, "Could not convert webhook effect to string"),
            )
            .ok(),
    };

    Ok(services::ApplicationResponse::IncomingWebhookEvent {
        response: Box::new(webhook_response),
        metadata,
    })
}

fn confirm_path_for_flow(
    candidate_path: common_enums::ExecutionPath,
    flow: Option<api::WebhookFlow>,
    enabled_flows: &[api::WebhookFlow],
    connector_integration_type: common_enums::ConnectorIntegrationType,
) -> common_enums::ExecutionPath {
    match (flow, connector_integration_type) {
        (_, common_enums::ConnectorIntegrationType::UcsConnector) => {
            common_enums::ExecutionPath::UnifiedConnectorService
        }
        (Some(f), _) if enabled_flows.contains(&f) => candidate_path,
        _ => common_enums::ExecutionPath::Direct,
    }
}

#[allow(clippy::too_many_arguments)]
#[instrument(skip_all)]
async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
    state: SessionState,
    req_state: ReqState,
    req: &actix_web::HttpRequest,
    platform: domain::Platform,
    connector_name_or_mca_id: &str,
    body: actix_web::web::Bytes,
    is_relay_webhook: bool,
) -> errors::RouterResult<(
    WebhookResponse<serde_json::Value>,
    WebhookResponseTracker,
    common_utils::pii::SecretSerdeValue,
)> {
    // Initial setup and metrics
    metrics::WEBHOOK_INCOMING_COUNT.add(
        1,
        router_env::metric_attributes!((
            MERCHANT_ID,
            platform.get_processor().get_account().get_id().clone()
        )),
    );
    let dimensions = dimension_state::Dimensions::new()
        .with_processor_merchant_id(platform.get_processor().get_processor_merchant_id())
        .with_provider_merchant_id(platform.get_provider().get_provider_merchant_id())
        .with_organization_id(platform.get_processor().get_account().get_org_id().clone());
    let request_details = IncomingWebhookRequestDetails {
        method: req.method().clone(),
        uri: req.uri().clone(),
        headers: req.headers(),
        query_params: req.query_string().to_string(),
        body: &body,
    };

    // Fetch Threeds Execution Path to decode webhook Body
    // Checks if the merchant and connector is eligible for authentication service
    // If Yes, then it will return UnifiedAuthenticationService
    // If No, then it will return Direct
    // Direct Signifies its not a authentication connector
    let three_ds_execution_path =
        fetch_three_ds_execution_path(&platform, connector_name_or_mca_id, &state, &dimensions)
            .await?;

    // Each arm resolves its path (UAS vs Direct/UCS) and returns a
    // `WebhookOutcome` plus the connector/name used for acks, errors, and
    // logging. Downstream filter interception and business-logic dispatch are
    // identical for both and run once after the match.
    let (connector, connector_name, outcome) = match three_ds_execution_path {
        ThreeDsProcessingMode::UnifiedAuthenticationService(ref mca_data) => {
            let connector_name = mca_data.connector_name.clone();
            let (uas_connector, _) =
                get_connector_by_connector_name(&state, UNIFIED_AUTHENTICATION_SERVICE, None)?;

            let outcome: errors::RouterResult<super::gateway::WebhookOutcome> =
                Box::pin(process_uas_incoming_webhook(
                    &state,
                    &platform,
                    &request_details,
                    connector_name.clone(),
                    uas_connector.clone(),
                ))
                .await;

            (uas_connector, connector_name, outcome)
        }
        ThreeDsProcessingMode::Direct(ref mca_data) => {
            let connector = mca_data.connector.clone();
            let connector_name = mca_data.connector_name.clone();
            let mca_ref = mca_data.merchant_connector_account.as_ref();

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

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

            let (candidate_path, enabled_flows) =
                unified_connector_service::should_call_unified_connector_service_for_webhooks(
                    &state,
                    platform.get_processor(),
                    &connector_name,
                    connector_integration_type,
                )
                .await?;

            logger::info!(
                connector = %connector_name,
                ?candidate_path,
                "Selected webhook execution path for event parsing"
            );

            let (ucs_event_reference, ucs_event_type) = match candidate_path {
                common_enums::ExecutionPath::UnifiedConnectorService
                | common_enums::ExecutionPath::ShadowUnifiedConnectorService => {
                    super::gateway::get_webhook_event_details_from_ucs(
                        &state,
                        &platform,
                        connector.clone(),
                        &connector_name,
                        mca_ref,
                        &request_details,
                    )
                    .await
                }
                _ => (None, None),
            };

            let flow = ucs_event_type.map(api::WebhookFlow::from);

            let execution_path = confirm_path_for_flow(
                candidate_path,
                flow,
                &enabled_flows,
                connector_integration_type,
            );

            let execution_mode = match execution_path {
                common_enums::ExecutionPath::UnifiedConnectorService => {
                    common_enums::ExecutionMode::Primary
                }
                common_enums::ExecutionPath::ShadowUnifiedConnectorService => {
                    common_enums::ExecutionMode::Shadow
                }
                common_enums::ExecutionPath::Direct => common_enums::ExecutionMode::NotApplicable,
            };
            logger::info!(?execution_path, "Selected webhook execution path");

            let ctx = super::gateway::WebhookGatewayContext {
                state: state.clone(),
                platform: platform.clone(),
                connector: connector.clone(),
                connector_name: connector_name.clone(),
                merchant_connector_account: mca_ref.cloned(),
                execution_path,
                execution_mode,
                ucs_reference: ucs_event_reference,
                ucs_event_type,
            };

            let outcome =
                super::gateway::execute_incoming_webhook_gateway(&ctx, &request_details).await;

            (connector, connector_name, outcome)
        }
    };

    if matches!(
        &outcome,
        Ok(outcome) if outcome.event_type() == webhooks::IncomingWebhookEvent::SetupWebhook
    ) {
        return Ok((
            WebhookResponse::StatusOk,
            WebhookResponseTracker::NoEffect,
            Secret::new(serde_json::Value::default()),
        ));
    }

    let (ack_response, webhook_effect, serialized_body) = match outcome {
        Ok(super::gateway::WebhookOutcome::Skipped {
            event_type,
            ack_response,
            ..
        }) => {
            logger::info!(?event_type, "Webhook filtered before business logic");
            metrics::WEBHOOK_INCOMING_FILTERED_COUNT.add(
                1,
                router_env::metric_attributes!((
                    MERCHANT_ID,
                    platform.get_processor().get_account().get_id().clone()
                )),
            );
            (
                ack_response,
                WebhookResponseTracker::NoEffect,
                Secret::new(serde_json::Value::Null),
            )
        }
        Ok(super::gateway::WebhookOutcome::Processed {
            reference,
            event_type,
            source_verified,
            content,
            decoded_body,
            webhook_resource_data,
            masked_log_payload,
            merchant_connector_account: processed_mca,
            ack_response,
        }) => {
            // Reconstruct request_details with the decoded body when available
            // (Direct and UAS paths), so downstream connector methods that parse
            // request_details.body (get_dispute_details, get_webhook_mandate_details,
            // etc.) receive the structured payload they expect.
            // UCS path leaves decoded_body as None and uses the raw request body.
            let decoded_request_details;
            let request_for_biz: &IncomingWebhookRequestDetails<'_> = match decoded_body.as_deref()
            {
                Some(body) => {
                    decoded_request_details = IncomingWebhookRequestDetails {
                        method: request_details.method.clone(),
                        uri: request_details.uri.clone(),
                        headers: request_details.headers,
                        query_params: request_details.query_params.clone(),
                        body,
                    };
                    &decoded_request_details
                }
                None => &request_details,
            };

            let effect = match Box::pin(process_webhook_business_logic(
                &state,
                req_state,
                &platform,
                &connector,
                &connector_name,
                event_type,
                source_verified,
                content,
                request_for_biz,
                is_relay_webhook,
                reference,
                *processed_mca,
                *webhook_resource_data,
            ))
            .await
            {
                Ok(effect) => effect,
                Err(error) => match handle_incoming_webhook_error(
                    error,
                    &connector,
                    connector_name.as_str(),
                    &request_details,
                    platform.get_processor().get_account().get_id(),
                ) {
                    Ok((_, tracker, _)) => tracker,
                    Err(e) => return Err(e),
                },
            };
            (ack_response, effect, masked_log_payload)
        }
        Err(error) => {
            return handle_incoming_webhook_error(
                error,
                &connector,
                connector_name.as_str(),
                &request_details,
                platform.get_processor().get_account().get_id(),
            );
        }
    };

    Ok((ack_response, webhook_effect, serialized_body))
}

async fn fetch_three_ds_execution_path(
    platform: &domain::Platform,
    connector_name_or_mca_id: &str,
    state: &SessionState,
    _dimensions: &dimension_state::DimensionsWithProcessorAndProviderMerchantIdAndOrgId,
) -> errors::RouterResult<ThreeDsProcessingMode> {
    let (merchant_connector_account, connector, connector_name) =
        fetch_optional_mca_and_connector(state, platform, connector_name_or_mca_id).await?;

    let mca_details = MerchantConnectorDetails {
        merchant_connector_account,
        connector,
        connector_name: connector_name.clone(),
    };

    let eligible_connector_list = state
        .conf
        .authentication_service_enabled_connectors
        .connector_list
        .clone();
    let connector_enum = Connector::from_str(&connector_name)
        .change_context(errors::ApiErrorResponse::InvalidDataValue {
            field_name: "connector",
        })
        .attach_printable_lazy(|| format!("unable to parse connector name {connector_name:?}"))?;
    let is_merchant_eligible_for_uas =
        payments::helpers::is_merchant_eligible_authentication_service(
            platform.get_processor(),
            state,
        )
        .await?;

    if is_merchant_eligible_for_uas && eligible_connector_list.contains(&connector_enum) {
        Ok(ThreeDsProcessingMode::UnifiedAuthenticationService(
            mca_details,
        ))
    } else {
        Ok(ThreeDsProcessingMode::Direct(mca_details))
    }
}

#[derive(Clone, Debug)]
pub struct MerchantConnectorDetails {
    /// Merchant Connector Account
    pub merchant_connector_account:
        Option<hyperswitch_domain_models::merchant_connector_account::MerchantConnectorAccount>,
    /// Connector
    pub connector: ConnectorEnum,
    pub connector_name: String,
}

#[derive(Clone, Debug)]
/// Indicates the execution path through which the authentication is processed.
pub enum ThreeDsProcessingMode {
    Direct(MerchantConnectorDetails),
    UnifiedAuthenticationService(MerchantConnectorDetails),
}

/// Runs the post-gateway business logic. The gateway has already resolved the
/// merchant-connector-account, verified source, and produced the webhook
/// payload. This function enforces mandatory-verification policy, picks the
/// entity-specific flow, and hands it a ready-to-consume payload.
#[allow(clippy::too_many_arguments)]
async fn process_webhook_business_logic(
    state: &SessionState,
    req_state: ReqState,
    platform: &domain::Platform,
    connector: &ConnectorEnum,
    connector_name: &str,
    event_type: webhooks::IncomingWebhookEvent,
    source_verified: bool,
    content: super::gateway::WebhookContent,
    request_details: &IncomingWebhookRequestDetails<'_>,
    is_relay_webhook: bool,
    object_reference_id: api::ObjectReferenceId,
    merchant_connector_account: domain::MerchantConnectorAccount,
    webhook_resource_data: Option<WebhookResourceData>,
) -> errors::RouterResult<WebhookResponseTracker> {
    let object_ref_id = object_reference_id;

    if source_verified {
        metrics::WEBHOOK_SOURCE_VERIFIED_COUNT.add(
            1,
            router_env::metric_attributes!((
                MERCHANT_ID,
                platform.get_processor().get_account().get_id().clone()
            )),
        );
    } else if connector.is_webhook_source_verification_mandatory() {
        // if webhook consumption is mandatory for connector, fail webhook
        // so that merchant can retrigger it after updating merchant_secret
        return Err(errors::ApiErrorResponse::WebhookAuthenticationFailed.into());
    }

    logger::info!(source_verified=?source_verified);

    let webhook_details = api::IncomingWebhookDetails {
        object_reference_id: object_ref_id.clone(),
        resource_object: content.bytes().to_vec(),
    };

    let profile_id = &merchant_connector_account.profile_id;

    let business_profile = state
        .store
        .find_business_profile_by_profile_id(platform.get_processor().get_key_store(), profile_id)
        .await
        .to_not_found_response(errors::ApiErrorResponse::ProfileNotFound {
            id: profile_id.get_string_repr().to_owned(),
        })?;

    // If the incoming webhook is a relay webhook, then we need to trigger the relay webhook flow
    let result_response = if is_relay_webhook {
        let relay_webhook_response = Box::pin(relay_incoming_webhook_flow(
            state.clone(),
            platform.clone(),
            business_profile,
            webhook_details,
            event_type,
            source_verified,
        ))
        .await
        .attach_printable("Incoming webhook flow for relay failed");

        // Using early return ensures unsupported webhooks are acknowledged to the connector
        if let Some(errors::ApiErrorResponse::NotSupported { .. }) = relay_webhook_response
            .as_ref()
            .err()
            .map(|a| a.current_context())
        {
            logger::error!(
                webhook_payload =? request_details.body,
                "Failed while identifying the event type",
            );

            let _response = connector
                    .get_webhook_api_response(request_details, None, Some(merchant_connector_account.connector_account_details))
                    .switch()
                    .attach_printable(
                        "Failed while early return in case of not supported event type in relay webhooks",
                    )?;

            return Ok(WebhookResponseTracker::NoEffect);
        };

        relay_webhook_response
    } else {
        let flow_type: api::WebhookFlow = event_type.into();
        match flow_type {
            api::WebhookFlow::Payment => Box::pin(payments_incoming_webhook_flow(
                state.clone(),
                req_state,
                platform.clone(),
                business_profile,
                webhook_details,
                source_verified,
                connector,
                request_details,
                event_type,
                &content,
                webhook_resource_data,
            ))
            .await
            .attach_printable("Incoming webhook flow for payments failed"),

            api::WebhookFlow::Refund => Box::pin(refunds_incoming_webhook_flow(
                state.clone(),
                platform.clone(),
                business_profile,
                webhook_details,
                connector_name,
                source_verified,
                event_type,
                webhook_resource_data,
            ))
            .await
            .attach_printable("Incoming webhook flow for refunds failed"),

            api::WebhookFlow::Dispute => Box::pin(disputes_incoming_webhook_flow(
                state.clone(),
                platform.clone(),
                business_profile,
                webhook_details,
                source_verified,
                connector,
                request_details,
                event_type,
                webhook_resource_data,
            ))
            .await
            .attach_printable("Incoming webhook flow for disputes failed"),

            api::WebhookFlow::BankTransfer => Box::pin(bank_transfer_webhook_flow(
                state.clone(),
                req_state,
                platform.clone(),
                business_profile,
                webhook_details,
                source_verified,
            ))
            .await
            .attach_printable("Incoming bank-transfer webhook flow failed"),

            api::WebhookFlow::ReturnResponse => Ok(WebhookResponseTracker::NoEffect),

            api::WebhookFlow::Mandate => Box::pin(mandates_incoming_webhook_flow(
                state.clone(),
                platform.clone(),
                business_profile,
                webhook_details,
                source_verified,
                event_type,
            ))
            .await
            .attach_printable("Incoming webhook flow for mandates failed"),

            api::WebhookFlow::ExternalAuthentication => {
                Box::pin(external_authentication_incoming_webhook_flow(
                    state.clone(),
                    req_state,
                    platform.clone(),
                    source_verified,
                    event_type,
                    request_details,
                    connector,
                    object_ref_id,
                    business_profile,
                    merchant_connector_account,
                ))
                .await
                .attach_printable("Incoming webhook flow for external authentication failed")
            }
            api::WebhookFlow::FraudCheck => Box::pin(frm_incoming_webhook_flow(
                state.clone(),
                req_state,
                platform.clone(),
                source_verified,
                event_type,
                object_ref_id,
                business_profile,
            ))
            .await
            .attach_printable("Incoming webhook flow for fraud check failed"),

            #[cfg(feature = "payouts")]
            api::WebhookFlow::Payout => Box::pin(payouts_incoming_webhook_flow(
                state.clone(),
                platform.clone(),
                business_profile,
                webhook_details,
                event_type,
                source_verified,
                request_details,
                connector,
            ))
            .await
            .attach_printable("Incoming webhook flow for payouts failed"),

            api::WebhookFlow::Subscription => {
                Box::pin(subscriptions::webhooks::incoming_webhook_flow(
                    state.clone().into(),
                    platform.clone(),
                    business_profile,
                    webhook_details,
                    source_verified,
                    connector,
                    request_details,
                    event_type,
                    merchant_connector_account,
                ))
                .await
                .attach_printable("Incoming webhook flow for subscription failed")
            }

            _ => Err(errors::ApiErrorResponse::InternalServerError)
                .attach_printable("Unsupported Flow Type received in incoming webhooks"),
        }
    };

    match result_response {
        Ok(response) => Ok(response),
        Err(error) => {
            let result = handle_incoming_webhook_error(
                error,
                connector,
                connector_name,
                request_details,
                platform.get_processor().get_account().get_id(),
            );
            match result {
                Ok((_, webhook_tracker, _)) => Ok(webhook_tracker),
                Err(e) => Err(e),
            }
        }
    }
}

fn handle_incoming_webhook_error(
    error: error_stack::Report<errors::ApiErrorResponse>,
    connector: &ConnectorEnum,
    connector_name: &str,
    request_details: &IncomingWebhookRequestDetails<'_>,
    merchant_id: &common_utils::id_type::MerchantId,
) -> errors::RouterResult<(
    WebhookResponse<serde_json::Value>,
    WebhookResponseTracker,
    common_utils::pii::SecretSerdeValue,
)> {
    logger::error!(?error, "Incoming webhook flow failed");

    // fetch the connector enum from the connector name
    let connector_enum = Connector::from_str(connector_name)
        .change_context(errors::ApiErrorResponse::InvalidDataValue {
            field_name: "connector",
        })
        .attach_printable_lazy(|| format!("unable to parse connector name {connector_name:?}"))?;

    // get the error response from the connector
    if connector_enum.should_acknowledge_webhook_for_resource_not_found_errors() {
        metrics::WEBHOOK_FLOW_FAILED_BUT_ACKNOWLEDGED.add(
            1,
            router_env::metric_attributes!(
                ("connector", connector_name.to_string()),
                ("merchant_id", merchant_id.get_string_repr().to_string())
            ),
        );
        let response = connector
            .get_webhook_api_response(
                request_details,
                Some(IncomingWebhookFlowError::from(error.current_context())),
                None,
            )
            .switch()
            .attach_printable("Failed to get incoming webhook api response from connector")?;
        Ok((
            response,
            WebhookResponseTracker::NoEffect,
            Secret::new(serde_json::Value::Null),
        ))
    } else {
        Err(error)
    }
}

#[instrument(skip_all)]
#[cfg(feature = "v1")]
async fn network_token_incoming_webhooks_core<W: types::OutgoingWebhookType>(
    state: &SessionState,
    request_details: IncomingWebhookRequestDetails<'_>,
) -> errors::RouterResult<(
    WebhookResponse<serde_json::Value>,
    WebhookResponseTracker,
    serde_json::Value,
    common_utils::id_type::MerchantId,
)> {
    let serialized_request =
        network_tokenization_incoming::get_network_token_resource_object(&request_details)
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Network Token Requestor Webhook deserialization failed")?
            .masked_serialize()
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Could not convert webhook effect to string")?;

    let network_tokenization_service = &state
        .conf
        .network_tokenization_service
        .as_ref()
        .ok_or(errors::NetworkTokenizationError::NetworkTokenizationServiceNotConfigured)
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Network Tokenization Service not configured")?;

    //source verification
    network_tokenization_incoming::Authorization::new(request_details.headers.get("Authorization"))
        .verify_webhook_source(network_tokenization_service.get_inner())
        .await?;

    let response: network_tokenization_incoming::NetworkTokenWebhookResponse = request_details
        .body
        .parse_struct("NetworkTokenWebhookResponse")
        .change_context(errors::ApiErrorResponse::WebhookUnprocessableEntity)?;

    let (merchant_id, payment_method_id, _customer_id) = response
        .fetch_merchant_id_payment_method_id_customer_id_from_callback_mapper(state)
        .await?;

    metrics::WEBHOOK_SOURCE_VERIFIED_COUNT.add(
        1,
        router_env::metric_attributes!((MERCHANT_ID, merchant_id.clone())),
    );

    let platform =
        network_tokenization_incoming::fetch_merchant_account_for_network_token_webhooks(
            state,
            &merchant_id,
        )
        .await?;
    let payment_method =
        network_tokenization_incoming::fetch_payment_method_for_network_token_webhooks(
            state,
            platform.get_provider().get_account(),
            platform.get_provider().get_key_store(),
            &payment_method_id,
        )
        .await?;

    let response_data = response.get_response_data();

    let webhook_resp_tracker = response_data
        .update_payment_method(state, &payment_method, &platform)
        .await?;

    Ok((
        WebhookResponse::StatusOk,
        webhook_resp_tracker,
        serialized_request,
        merchant_id.clone(),
    ))
}

#[allow(clippy::too_many_arguments)]
#[instrument(skip_all)]
async fn payments_incoming_webhook_flow(
    state: SessionState,
    req_state: ReqState,
    platform: domain::Platform,
    business_profile: domain::Profile,
    webhook_details: api::IncomingWebhookDetails,
    source_verified: bool,
    connector: &ConnectorEnum,
    request_details: &IncomingWebhookRequestDetails<'_>,
    event_type: webhooks::IncomingWebhookEvent,
    content: &super::gateway::WebhookContent,
    webhook_resource_data: Option<WebhookResourceData>,
) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
    let consume_or_trigger_flow = if source_verified {
        let resource_object = webhook_details.resource_object.clone();
        match content {
            super::gateway::WebhookContent::Direct(_) => {
                payments::CallConnectorAction::HandleResponse {
                    resource_object,
                    event_type: Some(event_type.into()),
                }
            }
            super::gateway::WebhookContent::UnifiedConnectorService(_) => {
                payments::CallConnectorAction::UCSConsumeResponse(resource_object)
            }
        }
    } else {
        payments::CallConnectorAction::Trigger
    };

    let shadow_ucs_call_connector_action: Option<payments::CallConnectorAction> = None;

    // Reuse the pre-fetched payment attempt when available; otherwise fetch now.
    let payment_attempt =
        if let Some(WebhookResourceData::Payment { payment_attempt }) = webhook_resource_data {
            payment_attempt
        } else {
            get_payment_attempt_from_object_reference_id(
                &state,
                webhook_details.object_reference_id.clone(),
                platform.get_processor(),
            )
            .await?
        };

    let payments_response = match webhook_details.object_reference_id {
        webhooks::ObjectReferenceId::PaymentId(ref id) => {
            let payment_id = payment_attempt.payment_id.clone();

            let lock_action = api_locking::LockAction::Hold {
                input: api_locking::LockingInput {
                    unique_locking_key: payment_id.get_string_repr().to_owned(),
                    api_identifier: lock_utils::ApiIdentifier::Payments,
                    override_lock_retries: None,
                },
            };

            lock_action
                .clone()
                .perform_locking_action(
                    &state,
                    platform.get_processor().get_account().get_id().to_owned(),
                )
                .await?;

            let response = Box::pin(payments::payments_core::<
                api::PSync,
                api::PaymentsResponse,
                _,
                _,
                _,
                payments::PaymentData<api::PSync>,
            >(
                state.clone(),
                req_state,
                platform.clone(),
                None,
                payments::operations::PaymentStatus,
                api::PaymentsRetrieveRequest {
                    resource_id: id.clone(),
                    merchant_id: Some(platform.get_processor().get_account().get_id().clone()),
                    force_sync: true,
                    connector: None,
                    param: None,
                    merchant_connector_details: None,
                    client_secret: None,
                    expand_attempts: None,
                    expand_captures: None,
                    all_keys_required: None,
                },
                services::AuthFlow::Merchant,
                consume_or_trigger_flow.clone(),
                shadow_ucs_call_connector_action,
                None,
                HeaderPayload::default(),
                None,
            ))
            .await;
            // When mandate details are present in successful webhooks, and consuming webhooks are skipped during payment sync if the payment status is already updated to charged, this function is used to update the connector mandate details.
            if should_update_connector_mandate_details(source_verified, event_type) {
                update_connector_mandate_details(
                    &state,
                    &platform,
                    webhook_details.object_reference_id.clone(),
                    connector,
                    request_details,
                )
                .await?
            };

            lock_action
                .free_lock_action(
                    &state,
                    platform.get_processor().get_account().get_id().to_owned(),
                )
                .await?;

            match response {
                Ok(value) => value,
                Err(err)
                    if matches!(
                        err.current_context(),
                        &errors::ApiErrorResponse::PaymentNotFound
                    ) && state
                        .conf
                        .webhooks
                        .ignore_error
                        .payment_not_found
                        .unwrap_or(true) =>
                {
                    metrics::WEBHOOK_PAYMENT_NOT_FOUND.add(
                        1,
                        router_env::metric_attributes!((
                            "merchant_id",
                            platform.get_processor().get_account().get_id().clone()
                        )),
                    );
                    return Ok(WebhookResponseTracker::NoEffect);
                }
                error @ Err(_) => error?,
            }
        }
        _ => Err(errors::ApiErrorResponse::WebhookProcessingFailure).attach_printable(
            "Did not get payment id as object reference id in webhook payments flow",
        )?,
    };

    match payments_response {
        services::ApplicationResponse::JsonWithHeaders((payments_response, _)) => {
            let payment_id = payments_response.payment_id.clone();
            let payment_method_id_opt = payments_response.payment_method_id.clone();

            match payment_method_id_opt {
                Some(payment_method_id) => {
                    if let Err(e) = update_additional_payment_method_data(
                        &state,
                        &platform,
                        connector,
                        request_details,
                        payment_method_id,
                    )
                    .await
                    {
                        logger::warn!(?e, "Failed to update additional payment method data");
                    }
                }
                None => {
                    logger::warn!("Failed to find payment_method_id in payments response");
                }
            }

            let status = payments_response.status;

            let event_type: Option<enums::EventType> = payments_response.status.into();

            // If event is NOT an UnsupportedEvent, trigger Outgoing Webhook
            if let Some(outgoing_event_type) = event_type {
                let primary_object_created_at = payments_response.created;
                let webhook_recipient = utils::resolve_webhook_recipient_from_created_by(
                    &state,
                    &platform,
                    &business_profile,
                    payment_attempt.created_by.as_ref(),
                )
                .await?;
                Box::pin(super::create_event_and_trigger_outgoing_webhook(
                    state,
                    platform,
                    outgoing_event_type,
                    enums::EventClass::Payments,
                    payment_id.get_string_repr().to_owned(),
                    enums::EventObjectType::PaymentDetails,
                    api::OutgoingWebhookContent::PaymentDetails(Box::new(payments_response)),
                    primary_object_created_at,
                    webhook_recipient,
                    Some(WebhookResourceData::Payment { payment_attempt }),
                    business_profile,
                ))
                .await?;
            };

            let response = WebhookResponseTracker::Payment { payment_id, status };

            Ok(response)
        }

        _ => Err(errors::ApiErrorResponse::WebhookProcessingFailure)
            .attach_printable("received non-json response from payments core")?,
    }
}

#[cfg(feature = "payouts")]
#[instrument(skip_all)]
#[allow(clippy::too_many_arguments)]
async fn payouts_incoming_webhook_flow(
    state: SessionState,
    platform: domain::Platform,
    business_profile: domain::Profile,
    webhook_details: api::IncomingWebhookDetails,
    event_type: webhooks::IncomingWebhookEvent,
    source_verified: bool,
    request_details: &IncomingWebhookRequestDetails<'_>,
    connector: &ConnectorEnum,
) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
    let payout_id =
        get_payout_id_from_object_reference_id(&state, &platform, &webhook_details).await?;

    let lock_action = api_locking::LockAction::Hold {
        input: api_locking::LockingInput {
            unique_locking_key: payout_id.get_string_repr().to_owned(),
            api_identifier: lock_utils::ApiIdentifier::Payouts,
            override_lock_retries: None,
        },
    };

    lock_action
        .clone()
        .perform_locking_action(
            &state,
            platform.get_processor().get_account().get_id().to_owned(),
        )
        .await?;

    let payout_response = Box::pin(process_payout_incoming_webhook(
        state.clone(),
        platform.clone(),
        business_profile,
        event_type,
        payout_id,
        source_verified,
        request_details,
        connector,
    ))
    .await;

    lock_action
        .free_lock_action(
            &state,
            platform.get_processor().get_account().get_id().to_owned(),
        )
        .await?;

    payout_response
}

#[cfg(feature = "payouts")]
struct PayoutWebhookPreCondition {
    is_non_terminal_status: bool,
    is_source_verified: bool,
}

#[cfg(feature = "payouts")]
enum PaoyoutWebhookAction {
    UpdateStatus,
    RetrieveStatus,
    NoAction,
}

#[cfg(feature = "payouts")]
impl PayoutWebhookPreCondition {
    pub fn new(is_non_terminal_status: bool, is_source_verified: bool) -> Self {
        Self {
            is_non_terminal_status,
            is_source_verified,
        }
    }

    // only update if the payout is in non-terminal status
    // if source verified, update the payout attempt and trigger outgoing webhook
    // if not source verified, do a payout retrieve call and update the status
    pub fn get_payout_webhook_action(&self) -> PaoyoutWebhookAction {
        match (self.is_non_terminal_status, self.is_source_verified) {
            (true, true) => PaoyoutWebhookAction::UpdateStatus,
            (true, false) => PaoyoutWebhookAction::RetrieveStatus,
            (false, true) | (false, false) => PaoyoutWebhookAction::NoAction,
        }
    }
}

#[cfg(feature = "payouts")]
async fn get_payout_id_from_object_reference_id(
    state: &SessionState,
    platform: &domain::Platform,
    webhook_details: &api::IncomingWebhookDetails,
) -> CustomResult<common_utils::id_type::PayoutId, errors::ApiErrorResponse> {
    let db = &*state.store;
    let payout_attempt = match webhook_details.object_reference_id.clone() {
        webhooks::ObjectReferenceId::PayoutId(payout_id_type) => match payout_id_type {
            webhooks::PayoutIdType::PayoutAttemptId(id) => db
                .find_payout_attempt_by_merchant_id_payout_attempt_id(
                    platform.get_processor().get_account().get_id(),
                    &id,
                    platform.get_processor().get_account().storage_scheme,
                )
                .await
                .change_context(errors::ApiErrorResponse::WebhookResourceNotFound)
                .attach_printable("Failed to fetch the payout attempt")?,
            webhooks::PayoutIdType::ConnectorPayoutId(id) => db
                .find_payout_attempt_by_merchant_id_connector_payout_id(
                    platform.get_processor().get_account().get_id(),
                    &id,
                    platform.get_processor().get_account().storage_scheme,
                )
                .await
                .change_context(errors::ApiErrorResponse::WebhookResourceNotFound)
                .attach_printable("Failed to fetch the payout attempt")?,
        },
        _ => Err(errors::ApiErrorResponse::WebhookProcessingFailure)
            .attach_printable("received a non-payout id when processing payout webhooks")?,
    };

    Ok(payout_attempt.payout_id)
}

#[cfg(feature = "payouts")]
#[instrument(skip_all)]
#[allow(clippy::too_many_arguments)]
async fn process_payout_incoming_webhook(
    state: SessionState,
    platform: domain::Platform,
    business_profile: domain::Profile,
    event_type: webhooks::IncomingWebhookEvent,
    payout_id: common_utils::id_type::PayoutId,
    source_verified: bool,
    request_details: &IncomingWebhookRequestDetails<'_>,
    connector: &ConnectorEnum,
) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
    metrics::INCOMING_PAYOUT_WEBHOOK_METRIC.add(1, &[]);

    let action_req =
        payout_models::PayoutRequest::PayoutRetrieveRequest(payout_models::PayoutRetrieveRequest {
            payout_id,
            force_sync: None,
            merchant_id: Some(platform.get_processor().get_account().get_id().clone()),
            expand_attempts: None,
        });

    let mut payout_data = Box::pin(payouts::make_payout_data(
        &state,
        &platform,
        None,
        &action_req,
        common_utils::consts::DEFAULT_LOCALE,
    ))
    .await?;

    let payout_webhook_action = PayoutWebhookPreCondition::new(
        payout_data.payout_attempt.status.is_non_terminal_status(),
        source_verified,
    )
    .get_payout_webhook_action();
    match payout_webhook_action {
        PaoyoutWebhookAction::UpdateStatus => {
            payout_incoming_webhook_update_status(
                state,
                platform,
                business_profile,
                event_type,
                request_details,
                connector,
                &mut payout_data,
            )
            .await
        }
        PaoyoutWebhookAction::RetrieveStatus => {
            payout_incoming_webhook_retrieve_status(
                state,
                platform,
                business_profile,
                &mut payout_data,
            )
            .await
        }
        PaoyoutWebhookAction::NoAction => Ok(WebhookResponseTracker::Payout {
            payout_id: payout_data.payout_attempt.payout_id,
            status: payout_data.payout_attempt.status,
        }),
    }
}

#[cfg(feature = "payouts")]
#[instrument(skip_all)]
#[allow(clippy::too_many_arguments)]
async fn payout_incoming_webhook_update_status(
    state: SessionState,
    platform: domain::Platform,
    business_profile: domain::Profile,
    event_type: webhooks::IncomingWebhookEvent,
    request_details: &IncomingWebhookRequestDetails<'_>,
    connector: &ConnectorEnum,
    payout_data: &mut payouts::PayoutData,
) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
    let payout_attempt = &payout_data.payout_attempt;
    let db = &*state.store;
    let status = common_enums::PayoutStatus::foreign_try_from(event_type)
        .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
        .attach_printable("failed payout status mapping from event type")?;

    let payout_webhook_details = connector
        .get_payout_webhook_details(request_details)
        .switch()
        .attach_printable("Failed to get error object for payouts")?;

    let payouts_update = PayoutsUpdate::StatusUpdate { status };

    let updated_payouts = db
        .update_payout(
            &payout_data.payouts,
            payouts_update,
            payout_attempt,
            platform.get_processor().get_account().storage_scheme,
        )
        .await
        .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
        .attach_printable_lazy(|| {
            format!(
                "Failed while updating payouts: payout_id: {}",
                payout_data.payouts.payout_id.get_string_repr()
            )
        })?;
    payout_data.payouts = updated_payouts;

    // if status is failure then update the error_code and error_message as well
    let payout_attempt_update = if status.is_payout_failure() {
        PayoutAttemptUpdate::StatusUpdate {
            connector_payout_id: payout_attempt.connector_payout_id.clone(),
            status,
            error_message: payout_webhook_details.error_message,
            error_code: payout_webhook_details.error_code,
            is_eligible: payout_attempt.is_eligible,
            unified_code: None,
            unified_message: None,
            payout_connector_metadata: payout_attempt.payout_connector_metadata.clone(),
        }
    } else {
        PayoutAttemptUpdate::StatusUpdate {
            connector_payout_id: payout_attempt.connector_payout_id.clone(),
            status,
            error_message: None,
            error_code: None,
            is_eligible: payout_data.payout_attempt.is_eligible,
            unified_code: None,
            unified_message: None,
            payout_connector_metadata: payout_attempt.payout_connector_metadata.clone(),
        }
    };

    let updated_payout_attempt = db
        .update_payout_attempt(
            payout_attempt,
            payout_attempt_update,
            &payout_data.payouts,
            platform.get_processor().get_account().storage_scheme,
        )
        .await
        .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
        .attach_printable_lazy(|| {
            format!(
                "Failed while updating payout attempt: payout_attempt_id: {}",
                payout_attempt.payout_attempt_id
            )
        })?;
    payout_data.payout_attempt = updated_payout_attempt;

    let event_type: Option<enums::EventType> = payout_data.payout_attempt.status.into();

    // If event is NOT an UnsupportedEvent, trigger Outgoing Webhook
    if let Some(outgoing_event_type) = event_type {
        let payout_create_response =
            payouts::response_handler(&state, &platform, payout_data).await?;

        let webhook_recipient = utils::resolve_webhook_recipient_from_created_by(
            &state,
            &platform,
            &business_profile,
            payout_data.payouts.created_by.as_ref(),
        )
        .await?;

        Box::pin(super::create_event_and_trigger_outgoing_webhook(
            state,
            platform,
            outgoing_event_type,
            enums::EventClass::Payouts,
            payout_data
                .payout_attempt
                .payout_id
                .get_string_repr()
                .to_string(),
            enums::EventObjectType::PayoutDetails,
            api::OutgoingWebhookContent::PayoutDetails(Box::new(payout_create_response)),
            Some(payout_data.payout_attempt.created_at),
            webhook_recipient,
            None,
            business_profile,
        ))
        .await?;
    }

    Ok(WebhookResponseTracker::Payout {
        payout_id: payout_data.payout_attempt.payout_id.clone(),
        status: payout_data.payout_attempt.status,
    })
}

// source verified = false
#[cfg(feature = "payouts")]
#[instrument(skip_all)]
#[allow(clippy::too_many_arguments)]
async fn payout_incoming_webhook_retrieve_status(
    state: SessionState,
    platform: domain::Platform,
    business_profile: domain::Profile,
    payout_data: &mut payouts::PayoutData,
) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
    metrics::INCOMING_PAYOUT_WEBHOOK_SIGNATURE_FAILURE_METRIC.add(1, &[]);
    // Form connector data
    let connector_data = match &payout_data.payout_attempt.connector {
        Some(connector) => ConnectorData::get_payout_connector_by_name(
            &state.conf.connectors,
            connector,
            GetToken::Connector,
            payout_data.payout_attempt.merchant_connector_id.clone(),
        )
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to get the connector data")?,
        _ => Err(errors::ApplicationError::InvalidConfigurationValueError(
            "Connector not found in payout_attempt - should not reach here.".to_string(),
        ))
        .change_context(errors::ApiErrorResponse::MissingRequiredField {
            field_name: "connector",
        })
        .attach_printable("Connector not found for payout fulfillment")?,
    };

    Box::pin(payouts::create_payout_retrieve(
        &state,
        &platform,
        HeaderPayload::default(),
        &connector_data,
        payout_data,
    ))
    .await
    .attach_printable("Payout retrieval failed for given Payout request")?;

    let event_type: Option<enums::EventType> = payout_data.payout_attempt.status.into();

    // If event is NOT an UnsupportedEvent, trigger Outgoing Webhook
    if let Some(outgoing_event_type) = event_type {
        let payout_response = payouts::response_handler(&state, &platform, payout_data).await?;

        let webhook_recipient = utils::resolve_webhook_recipient_from_created_by(
            &state,
            &platform,
            &business_profile,
            payout_data.payouts.created_by.as_ref(),
        )
        .await?;

        Box::pin(super::create_event_and_trigger_outgoing_webhook(
            state,
            platform,
            outgoing_event_type,
            enums::EventClass::Payouts,
            payout_data
                .payout_attempt
                .payout_id
                .get_string_repr()
                .to_string(),
            enums::EventObjectType::PayoutDetails,
            api::OutgoingWebhookContent::PayoutDetails(Box::new(payout_response)),
            Some(payout_data.payout_attempt.created_at),
            webhook_recipient,
            None,
            business_profile,
        ))
        .await?;
    }

    Ok(WebhookResponseTracker::Payout {
        payout_id: payout_data.payout_attempt.payout_id.clone(),
        status: payout_data.payout_attempt.status,
    })
}

async fn relay_refunds_incoming_webhook_flow(
    state: SessionState,
    platform: domain::Platform,
    business_profile: domain::Profile,
    webhook_details: api::IncomingWebhookDetails,
    event_type: webhooks::IncomingWebhookEvent,
    source_verified: bool,
) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
    let db = &*state.store;

    let relay_record = match webhook_details.object_reference_id {
        webhooks::ObjectReferenceId::RefundId(refund_id_type) => match refund_id_type {
            webhooks::RefundIdType::RefundId(refund_id) => {
                let relay_id = common_utils::id_type::RelayId::from_str(&refund_id)
                    .change_context(errors::ValidationError::IncorrectValueProvided {
                        field_name: "relay_id",
                    })
                    .change_context(errors::ApiErrorResponse::InternalServerError)?;

                db.find_relay_by_id(platform.get_processor().get_key_store(), &relay_id)
                    .await
                    .to_not_found_response(errors::ApiErrorResponse::WebhookResourceNotFound)
                    .attach_printable("Failed to fetch the relay record")?
            }
            webhooks::RefundIdType::ConnectorRefundId(connector_refund_id) => db
                .find_relay_by_profile_id_connector_reference_id(
                    platform.get_processor().get_key_store(),
                    business_profile.get_id(),
                    &connector_refund_id,
                )
                .await
                .to_not_found_response(errors::ApiErrorResponse::WebhookResourceNotFound)
                .attach_printable("Failed to fetch the relay record")?,
        },
        _ => Err(errors::ApiErrorResponse::WebhookProcessingFailure)
            .attach_printable("received a non-refund id when processing relay refund webhooks")?,
    };

    // if source_verified then update relay status else trigger relay force sync
    let relay_response = if source_verified {
        let relay_update = hyperswitch_domain_models::relay::RelayUpdate::StatusUpdate {
            connector_reference_id: None,
            status: common_enums::RelayStatus::foreign_try_from(event_type)
                .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
                .attach_printable("failed relay refund status mapping from event type")?,
        };
        db.update_relay(
            platform.get_processor().get_key_store(),
            relay_record,
            relay_update,
        )
        .await
        .map(api_models::relay::RelayResponse::from)
        .to_not_found_response(errors::ApiErrorResponse::WebhookResourceNotFound)
        .attach_printable("Failed to update relay")?
    } else {
        let relay_retrieve_request = api_models::relay::RelayRetrieveRequest {
            force_sync: true,
            id: relay_record.id,
        };
        let relay_force_sync_response = Box::pin(relay::relay_retrieve(
            state,
            platform,
            Some(business_profile.get_id().clone()),
            relay_retrieve_request,
        ))
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to force sync relay")?;

        if let hyperswitch_domain_models::api::ApplicationResponse::Json(response) =
            relay_force_sync_response
        {
            response
        } else {
            Err(errors::ApiErrorResponse::InternalServerError)
                .attach_printable("Unexpected response from force sync relay")?
        }
    };

    Ok(WebhookResponseTracker::Relay {
        relay_id: relay_response.id,
        status: relay_response.status,
    })
}

#[allow(clippy::too_many_arguments)]
#[instrument(skip_all)]
async fn refunds_incoming_webhook_flow(
    state: SessionState,
    platform: domain::Platform,
    business_profile: domain::Profile,
    webhook_details: api::IncomingWebhookDetails,
    connector_name: &str,
    source_verified: bool,
    event_type: webhooks::IncomingWebhookEvent,
    webhook_resource_data: Option<WebhookResourceData>,
) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
    let db = &*state.store;
    //find refund by connector refund id
    let refund = match webhook_details.object_reference_id {
        webhooks::ObjectReferenceId::RefundId(refund_id_type) => match refund_id_type {
            webhooks::RefundIdType::RefundId(id) => db
                .find_refund_by_processor_merchant_id_refund_id(
                    platform.get_processor().get_account().get_id(),
                    &id,
                    platform.get_processor().get_account().storage_scheme,
                )
                .await
                .change_context(errors::ApiErrorResponse::WebhookResourceNotFound)
                .attach_printable("Failed to fetch the refund")?,
            webhooks::RefundIdType::ConnectorRefundId(id) => db
                .find_refund_by_processor_merchant_id_connector_refund_id_connector(
                    platform.get_processor().get_account().get_id(),
                    &id,
                    connector_name,
                    platform.get_processor().get_account().storage_scheme,
                )
                .await
                .change_context(errors::ApiErrorResponse::WebhookResourceNotFound)
                .attach_printable("Failed to fetch the refund")?,
        },
        _ => Err(errors::ApiErrorResponse::WebhookProcessingFailure)
            .attach_printable("received a non-refund id when processing refund webhooks")?,
    };
    let refund_id = refund.refund_id.to_owned();
    //if source verified then update refund status else trigger refund sync
    let updated_refund = if source_verified {
        let refund_update = diesel_refund::RefundUpdate::StatusUpdate {
            connector_refund_id: None,
            sent_to_gateway: true,
            refund_status: common_enums::RefundStatus::foreign_try_from(event_type)
                .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
                .attach_printable("failed refund status mapping from event type")?,
            updated_by: platform
                .get_processor()
                .get_account()
                .storage_scheme
                .to_string(),
            processor_refund_data: None,
        };
        db.update_refund(
            refund.to_owned(),
            refund_update,
            platform.get_processor().get_account().storage_scheme,
        )
        .await
        .to_not_found_response(errors::ApiErrorResponse::WebhookResourceNotFound)
        .attach_printable_lazy(|| format!("Failed while updating refund: refund_id: {refund_id}"))?
    } else {
        Box::pin(refunds::refund_retrieve_core_with_refund_id(
            state.clone(),
            platform.clone(),
            None,
            api_models::refunds::RefundsRetrieveRequest {
                refund_id: refund_id.to_owned(),
                force_sync: Some(true),
                merchant_connector_details: None,
                all_keys_required: None,
            },
        ))
        .await
        .attach_printable_lazy(|| format!("Failed while updating refund: refund_id: {refund_id}"))?
        .0
    };

    let payment_intent = db
        .find_payment_intent_by_payment_id_processor_merchant_id(
            &refund.payment_id,
            platform.get_processor().get_account().get_id(),
            platform.get_processor().get_key_store(),
            platform.get_processor().get_account().storage_scheme,
        )
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable_lazy(|| {
            format!(
                "Failed to find payment intent for payment_id: {:?}",
                refund.payment_id
            )
        })?;
    let state_task = state.clone();
    let processor_task = platform.get_processor().clone();
    let payment_intent_task = payment_intent.clone();
    tokio::spawn(async move {
        if let Err(err) = PaymentIntentStateMetadataExt::from(
            payment_intent_task
                .state_metadata
                .clone()
                .unwrap_or_default(),
        )
        .update_intent_state_metadata_for_refund(&state_task, &processor_task, payment_intent_task)
        .await
        {
            tracing::error!(?err, "Failed to update intent state metadata for refund");
        }
    });

    let event_type: Option<enums::EventType> = updated_refund.refund_status.into();

    // If event is NOT an UnsupportedEvent, trigger Outgoing Webhook
    if let Some(outgoing_event_type) = event_type {
        let refund_response: api_models::refunds::RefundResponse =
            updated_refund.clone().foreign_into();
        let refund_created_by = updated_refund
            .created_by
            .as_deref()
            .and_then(|created_by| created_by.parse::<common_utils::types::CreatedBy>().ok());
        let webhook_recipient = utils::resolve_webhook_recipient_from_created_by(
            &state,
            &platform,
            &business_profile,
            refund_created_by.as_ref(),
        )
        .await?;
        Box::pin(super::create_event_and_trigger_outgoing_webhook(
            state,
            platform,
            outgoing_event_type,
            enums::EventClass::Refunds,
            refund_id,
            enums::EventObjectType::RefundDetails,
            api::OutgoingWebhookContent::RefundDetails(Box::new(refund_response)),
            Some(updated_refund.created_at),
            webhook_recipient,
            webhook_resource_data,
            business_profile,
        ))
        .await?;
    }

    Ok(WebhookResponseTracker::Refund {
        payment_id: updated_refund.payment_id,
        refund_id: updated_refund.refund_id,
        status: updated_refund.refund_status,
    })
}

async fn relay_incoming_webhook_flow(
    state: SessionState,
    platform: domain::Platform,
    business_profile: domain::Profile,
    webhook_details: api::IncomingWebhookDetails,
    event_type: webhooks::IncomingWebhookEvent,
    source_verified: bool,
) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
    let flow_type: api::WebhookFlow = event_type.into();

    let result_response = match flow_type {
        webhooks::WebhookFlow::Refund => Box::pin(relay_refunds_incoming_webhook_flow(
            state,
            platform,
            business_profile,
            webhook_details,
            event_type,
            source_verified,
        ))
        .await
        .attach_printable("Incoming webhook flow for relay refund failed")?,
        webhooks::WebhookFlow::Payment
        | webhooks::WebhookFlow::Payout
        | webhooks::WebhookFlow::Dispute
        | webhooks::WebhookFlow::Subscription
        | webhooks::WebhookFlow::ReturnResponse
        | webhooks::WebhookFlow::BankTransfer
        | webhooks::WebhookFlow::Mandate
        | webhooks::WebhookFlow::Setup
        | webhooks::WebhookFlow::ExternalAuthentication
        | webhooks::WebhookFlow::FraudCheck => Err(errors::ApiErrorResponse::NotSupported {
            message: "Relay webhook flow types not supported".to_string(),
        })?,
    };
    Ok(result_response)
}

pub async fn get_payment_attempt_from_object_reference_id(
    state: &SessionState,
    object_reference_id: webhooks::ObjectReferenceId,
    processor: &domain::Processor,
) -> CustomResult<PaymentAttempt, errors::ApiErrorResponse> {
    let db = &*state.store;
    match object_reference_id {
        api::ObjectReferenceId::PaymentId(api::PaymentIdType::ConnectorTransactionId(ref id)) => db
            .find_payment_attempt_by_processor_merchant_id_connector_txn_id(
                processor.get_account().get_id(),
                id,
                processor.get_account().storage_scheme,
                processor.get_key_store(),
            )
            .await
            .to_not_found_response(errors::ApiErrorResponse::WebhookResourceNotFound),
        api::ObjectReferenceId::PaymentId(api::PaymentIdType::PaymentAttemptId(ref id)) => db
            .find_payment_attempt_by_attempt_id_processor_merchant_id(
                id,
                processor.get_account().get_id(),
                processor.get_account().storage_scheme,
                processor.get_key_store(),
            )
            .await
            .to_not_found_response(errors::ApiErrorResponse::WebhookResourceNotFound),
        api::ObjectReferenceId::PaymentId(api::PaymentIdType::PreprocessingId(ref id)) => db
            .find_payment_attempt_by_preprocessing_id_processor_merchant_id(
                id,
                processor.get_account().get_id(),
                processor.get_account().storage_scheme,
                processor.get_key_store(),
            )
            .await
            .to_not_found_response(errors::ApiErrorResponse::WebhookResourceNotFound),
        _ => Err(errors::ApiErrorResponse::WebhookProcessingFailure)
            .attach_printable("received a non-payment id for retrieving payment")?,
    }
}

#[allow(clippy::too_many_arguments)]
pub async fn get_or_update_dispute_object(
    state: SessionState,
    option_dispute: Option<diesel_models::dispute::Dispute>,
    dispute_details: api::disputes::DisputePayload,
    platform: &domain::Platform,
    payment_attempt: &PaymentAttempt,
    dispute_status: common_enums::enums::DisputeStatus,
    business_profile: &domain::Profile,
    connector_name: &str,
) -> CustomResult<diesel_models::dispute::Dispute, errors::ApiErrorResponse> {
    let db = &*state.store;
    match option_dispute {
        None => {
            metrics::INCOMING_DISPUTE_WEBHOOK_NEW_RECORD_METRIC.add(1, &[]);
            let dispute_id = generate_id(consts::ID_LENGTH, "dp");
            let new_dispute = diesel_models::dispute::DisputeNew {
                dispute_id,
                amount: dispute_details.amount.clone(),
                currency: dispute_details.currency.to_string(),
                dispute_stage: dispute_details.dispute_stage,
                dispute_status,
                payment_id: payment_attempt.payment_id.to_owned(),
                connector: connector_name.to_owned(),
                attempt_id: payment_attempt.attempt_id.to_owned(),
                merchant_id: platform.get_provider().get_account().get_id().to_owned(),
                connector_status: dispute_details.connector_status,
                connector_dispute_id: dispute_details.connector_dispute_id,
                connector_reason: dispute_details.connector_reason,
                connector_reason_code: dispute_details.connector_reason_code,
                challenge_required_by: dispute_details.challenge_required_by,
                connector_created_at: dispute_details.created_at,
                connector_updated_at: dispute_details.updated_at,
                profile_id: Some(business_profile.get_id().to_owned()),
                evidence: Secret::new(serde_json::json!({})),
                merchant_connector_id: payment_attempt.merchant_connector_id.clone(),
                dispute_amount: StringMinorUnitForConnector::convert_back(
                    &StringMinorUnitForConnector,
                    dispute_details.amount,
                    dispute_details.currency,
                )
                .change_context(
                    errors::ApiErrorResponse::AmountConversionFailed {
                        amount_type: "MinorUnit",
                    },
                )?,
                organization_id: platform
                    .get_processor()
                    .get_account()
                    .organization_id
                    .clone(),
                dispute_currency: Some(dispute_details.currency),
                processor_merchant_id: Some(
                    platform.get_processor().get_account().get_id().to_owned(),
                ),
                created_by: payment_attempt
                    .created_by
                    .as_ref()
                    .map(|created_by| created_by.to_string()),
                created_at: common_utils::date_time::now(),
                modified_at: common_utils::date_time::now(),
            };
            state
                .store
                .insert_dispute(
                    new_dispute.clone(),
                    platform.get_processor().get_account().storage_scheme,
                )
                .await
                .to_not_found_response(errors::ApiErrorResponse::WebhookResourceNotFound)
        }
        Some(dispute) => {
            logger::info!("Dispute Already exists, Updating the dispute details");
            metrics::INCOMING_DISPUTE_WEBHOOK_UPDATE_RECORD_METRIC.add(1, &[]);
            core_utils::validate_dispute_stage_and_dispute_status(
                dispute.dispute_stage,
                dispute.dispute_status,
                dispute_details.dispute_stage,
                dispute_status,
            )
            .change_context(errors::ApiErrorResponse::WebhookBadRequest)
            .attach_printable("dispute stage and status validation failed")?;
            let update_dispute = diesel_models::dispute::DisputeUpdate::Update {
                dispute_stage: dispute_details.dispute_stage,
                dispute_status,
                connector_status: dispute_details.connector_status,
                connector_reason: dispute_details.connector_reason,
                connector_reason_code: dispute_details.connector_reason_code,
                challenge_required_by: dispute_details.challenge_required_by,
                connector_updated_at: dispute_details.updated_at,
            };
            db.update_dispute(
                dispute,
                update_dispute,
                platform.get_processor().get_account().storage_scheme,
            )
            .await
            .to_not_found_response(errors::ApiErrorResponse::WebhookResourceNotFound)
        }
    }
}

#[allow(clippy::too_many_arguments)]
#[instrument(skip_all)]
async fn external_authentication_incoming_webhook_flow(
    state: SessionState,
    req_state: ReqState,
    platform: domain::Platform,
    source_verified: bool,
    event_type: webhooks::IncomingWebhookEvent,
    request_details: &IncomingWebhookRequestDetails<'_>,
    connector: &ConnectorEnum,
    object_ref_id: api::ObjectReferenceId,
    business_profile: domain::Profile,
    merchant_connector_account: domain::MerchantConnectorAccount,
) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
    let key_manager_state = (&state).into();
    if source_verified {
        let authentication_details = connector
            .get_external_authentication_details(request_details)
            .switch()?;
        let trans_status = authentication_details.trans_status;
        let authentication_update = hyperswitch_domain_models::authentication::AuthenticationUpdate::PostAuthenticationUpdate {
            authentication_status: common_enums::AuthenticationStatus::foreign_from(
                trans_status.clone(),
            ),
            trans_status,
            eci: authentication_details.eci,
            challenge_cancel: authentication_details.challenge_cancel,
            challenge_code_reason: authentication_details.challenge_code_reason,
            updated_by: platform.get_processor().get_account().storage_scheme.to_string(),
        };
        let authentication =
            if let webhooks::ObjectReferenceId::ExternalAuthenticationID(authentication_id_type) =
                object_ref_id
            {
                match authentication_id_type {
                    webhooks::AuthenticationIdType::AuthenticationId(authentication_id) => state
                        .store
                        .find_authentication_by_processor_merchant_id_authentication_id(
                            platform.get_processor().get_account().get_id(),
                            &authentication_id,
                            platform.get_processor().get_key_store(),
                            &key_manager_state,
                            platform.get_processor().get_account().storage_scheme,
                        )
                        .await
                        .to_not_found_response(errors::ApiErrorResponse::AuthenticationNotFound {
                            id: authentication_id.get_string_repr().to_string(),
                        })
                        .attach_printable("Error while fetching authentication record"),
                    webhooks::AuthenticationIdType::ConnectorAuthenticationId(
                        connector_authentication_id,
                    ) => state
                        .store
                        .find_authentication_by_processor_merchant_id_connector_authentication_id(
                            platform.get_processor().get_account().get_id().clone(),
                            connector_authentication_id.clone(),
                            platform.get_processor().get_key_store(),
                            &key_manager_state,
                            platform.get_processor().get_account().storage_scheme,
                        )
                        .await
                        .to_not_found_response(errors::ApiErrorResponse::AuthenticationNotFound {
                            id: connector_authentication_id,
                        })
                        .attach_printable("Error while fetching authentication record"),
                }
            } else {
                Err(errors::ApiErrorResponse::WebhookProcessingFailure).attach_printable(
                    "received a non-external-authentication id for retrieving authentication",
                )
            }?;

        // Store the authentication value before publishing the successful authentication state.
        // Otherwise, the browser callback can observe `Success` and attempt authorization before
        // the CAVV is available in Redis.
        authentication_details
            .authentication_value
            .async_map(|auth_val| {
                payment_methods::vault::create_tokenize_without_configurable_expiry(
                    &state,
                    auth_val.expose(),
                    None,
                    authentication
                        .authentication_id
                        .clone()
                        .get_string_repr()
                        .to_string(),
                    platform.get_processor().get_key_store().key.get_inner(),
                )
            })
            .await
            .transpose()?;

        let updated_authentication = state
            .store
            .update_authentication_by_processor_merchant_id_authentication_id(
                authentication,
                authentication_update,
                platform.get_processor().get_key_store(),
                &key_manager_state,
                platform.get_processor().get_account().storage_scheme,
            )
            .await
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Error while updating authentication")?;

        // Check if it's a payment authentication flow, payment_id would be there only for payment authentication flows
        if let Some(payment_id) = updated_authentication.payment_id {
            let is_pull_mechanism_enabled = helper_utils::check_if_pull_mechanism_for_external_3ds_enabled_from_connector_metadata(merchant_connector_account.metadata.map(|metadata| metadata.expose()));
            // Merchant doesn't have pull mechanism enabled and if it's challenge flow, we have to authorize whenever we receive a ARes webhook
            if !is_pull_mechanism_enabled
                && updated_authentication.authentication_type
                    == Some(common_enums::DecoupledAuthenticationType::Challenge)
                && event_type == webhooks::IncomingWebhookEvent::ExternalAuthenticationARes
            {
                let payment_intent = state
                    .store
                    .find_payment_intent_by_payment_id_processor_merchant_id(
                        &payment_id,
                        platform.get_processor().get_account().get_id(),
                        platform.get_processor().get_key_store(),
                        platform.get_processor().get_account().storage_scheme,
                    )
                    .await
                    .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
                let payment_confirm_req = api::PaymentsRequest {
                    payment_id: Some(api_models::payments::PaymentIdType::PaymentIntentId(
                        payment_intent.payment_id.clone(),
                    )),
                    merchant_id: Some(platform.get_processor().get_account().get_id().clone()),
                    ..Default::default()
                };
                let is_setup_mandate = payment_intent.is_setup_mandate();
                let payments_response = if is_setup_mandate {
                    Box::pin(payments::payments_core::<
                        api::SetupMandate,
                        api::PaymentsResponse,
                        _,
                        _,
                        _,
                        payments::PaymentData<api::SetupMandate>,
                    >(
                        state.clone(),
                        req_state,
                        platform.clone(),
                        None,
                        payments::PaymentConfirm,
                        payment_confirm_req,
                        services::api::AuthFlow::Merchant,
                        payments::CallConnectorAction::Trigger,
                        None,
                        None,
                        HeaderPayload::with_source(enums::PaymentSource::ExternalAuthenticator),
                        None,
                    ))
                    .await?
                } else {
                    Box::pin(payments::payments_core::<
                        api::Authorize,
                        api::PaymentsResponse,
                        _,
                        _,
                        _,
                        payments::PaymentData<api::Authorize>,
                    >(
                        state.clone(),
                        req_state,
                        platform.clone(),
                        None,
                        payments::PaymentConfirm,
                        payment_confirm_req,
                        services::api::AuthFlow::Merchant,
                        payments::CallConnectorAction::Trigger,
                        None,
                        None,
                        HeaderPayload::with_source(enums::PaymentSource::ExternalAuthenticator),
                        None,
                    ))
                    .await?
                };
                match payments_response {
                    services::ApplicationResponse::JsonWithHeaders((payments_response, _)) => {
                        let payment_id = payments_response.payment_id.clone();

                        let status = payments_response.status;
                        let event_type: Option<enums::EventType> = payments_response.status.into();
                        // Set poll_id as completed in redis to allow the fetch status of poll through retrieve_poll_status api from client
                        let poll_id = core_utils::get_poll_id(
                            platform.get_processor().get_account().get_id(),
                            core_utils::get_external_authentication_request_poll_id(&payment_id),
                        );
                        let redis_conn = state
                            .store
                            .get_redis_conn()
                            .change_context(errors::ApiErrorResponse::InternalServerError)
                            .attach_printable("Failed to get redis connection")?;
                        redis_conn
                            .set_key_without_modifying_ttl(
                                &poll_id.into(),
                                api_models::poll::PollStatus::Completed.to_string(),
                            )
                            .await
                            .change_context(errors::ApiErrorResponse::InternalServerError)
                            .attach_printable("Failed to add poll_id in redis")?;
                        // If event is NOT an UnsupportedEvent, trigger Outgoing Webhook
                        if let Some(outgoing_event_type) = event_type {
                            let primary_object_created_at = payments_response.created;
                            let webhook_recipient =
                                utils::resolve_webhook_recipient_from_created_by(
                                    &state,
                                    &platform,
                                    &business_profile,
                                    updated_authentication.created_by.as_ref(),
                                )
                                .await?;
                            Box::pin(super::create_event_and_trigger_outgoing_webhook(
                                state,
                                platform,
                                outgoing_event_type,
                                enums::EventClass::Payments,
                                payment_id.get_string_repr().to_owned(),
                                enums::EventObjectType::PaymentDetails,
                                api::OutgoingWebhookContent::PaymentDetails(Box::new(
                                    payments_response,
                                )),
                                primary_object_created_at,
                                webhook_recipient,
                                None,
                                business_profile,
                            ))
                            .await?;
                        };
                        let response = WebhookResponseTracker::Payment { payment_id, status };
                        Ok(response)
                    }
                    _ => Err(errors::ApiErrorResponse::WebhookProcessingFailure).attach_printable(
                        "Did not get payment id as object reference id in webhook payments flow",
                    )?,
                }
            } else {
                Ok(WebhookResponseTracker::NoEffect)
            }
        } else {
            Ok(WebhookResponseTracker::NoEffect)
        }
    } else {
        logger::error!(
            "Webhook source verification failed for external authentication webhook flow"
        );
        Err(report!(
            errors::ApiErrorResponse::WebhookAuthenticationFailed
        ))
    }
}

#[instrument(skip_all)]
async fn mandates_incoming_webhook_flow(
    state: SessionState,
    platform: domain::Platform,
    business_profile: domain::Profile,
    webhook_details: api::IncomingWebhookDetails,
    source_verified: bool,
    event_type: webhooks::IncomingWebhookEvent,
) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
    if source_verified {
        let db = &*state.store;
        let mandate = match webhook_details.object_reference_id {
            webhooks::ObjectReferenceId::MandateId(webhooks::MandateIdType::MandateId(
                mandate_id,
            )) => db
                .find_mandate_by_merchant_id_mandate_id(
                    platform.get_processor().get_account().get_id(),
                    mandate_id.as_str(),
                    platform.get_processor().get_account().storage_scheme,
                )
                .await
                .to_not_found_response(errors::ApiErrorResponse::MandateNotFound)?,
            webhooks::ObjectReferenceId::MandateId(
                webhooks::MandateIdType::ConnectorMandateId(connector_mandate_id),
            ) => db
                .find_mandate_by_merchant_id_connector_mandate_id(
                    platform.get_processor().get_account().get_id(),
                    connector_mandate_id.as_str(),
                    platform.get_processor().get_account().storage_scheme,
                )
                .await
                .to_not_found_response(errors::ApiErrorResponse::MandateNotFound)?,
            _ => Err(errors::ApiErrorResponse::WebhookProcessingFailure)
                .attach_printable("received a non-mandate id for retrieving mandate")?,
        };
        let mandate_status = common_enums::MandateStatus::foreign_try_from(event_type)
            .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
            .attach_printable("event type to mandate status mapping failed")?;
        let mandate_id = mandate.mandate_id.clone();
        let updated_mandate = db
            .update_mandate_by_merchant_id_mandate_id(
                platform.get_processor().get_account().get_id(),
                &mandate_id,
                storage::MandateUpdate::StatusUpdate { mandate_status },
                mandate,
                platform.get_processor().get_account().storage_scheme,
            )
            .await
            .to_not_found_response(errors::ApiErrorResponse::MandateNotFound)?;
        let mandates_response = Box::new(
            api::mandates::MandateResponse::from_db_mandate(
                &state,
                platform.get_processor().get_key_store().clone(),
                updated_mandate.clone(),
                platform.get_processor().get_account(),
            )
            .await?,
        );
        let event_type: Option<enums::EventType> = updated_mandate.mandate_status.into();
        if let Some(outgoing_event_type) = event_type {
            let webhook_recipient = utils::resolve_webhook_recipient_from_created_by(
                &state,
                &platform,
                &business_profile,
                None, // Mandates do not carry created_by, default to processor
            )
            .await?;
            Box::pin(super::create_event_and_trigger_outgoing_webhook(
                state,
                platform,
                outgoing_event_type,
                enums::EventClass::Mandates,
                updated_mandate.mandate_id.clone(),
                enums::EventObjectType::MandateDetails,
                api::OutgoingWebhookContent::MandateDetails(mandates_response),
                Some(updated_mandate.created_at),
                webhook_recipient,
                None,
                business_profile,
            ))
            .await?;
        }
        Ok(WebhookResponseTracker::Mandate {
            mandate_id: updated_mandate.mandate_id,
            status: updated_mandate.mandate_status,
        })
    } else {
        logger::error!("Webhook source verification failed for mandates webhook flow");
        Err(report!(
            errors::ApiErrorResponse::WebhookAuthenticationFailed
        ))
    }
}

#[allow(clippy::too_many_arguments)]
#[instrument(skip_all)]
async fn frm_incoming_webhook_flow(
    state: SessionState,
    req_state: ReqState,
    platform: domain::Platform,
    source_verified: bool,
    event_type: webhooks::IncomingWebhookEvent,
    object_ref_id: api::ObjectReferenceId,
    business_profile: domain::Profile,
) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
    if source_verified {
        let payment_attempt = get_payment_attempt_from_object_reference_id(
            &state,
            object_ref_id,
            platform.get_processor(),
        )
        .await?;
        let payment_response = match event_type {
            webhooks::IncomingWebhookEvent::FrmApproved => {
                Box::pin(payments::payments_core::<
                    api::Capture,
                    api::PaymentsResponse,
                    _,
                    _,
                    _,
                    payments::PaymentData<api::Capture>,
                >(
                    state.clone(),
                    req_state,
                    platform.clone(),
                    None,
                    payments::PaymentApprove,
                    api::PaymentsCaptureRequest {
                        payment_id: payment_attempt.payment_id,
                        amount_to_capture: payment_attempt.amount_to_capture,
                        ..Default::default()
                    },
                    services::api::AuthFlow::Merchant,
                    payments::CallConnectorAction::Trigger,
                    None,
                    None,
                    HeaderPayload::default(),
                    None,
                ))
                .await?
            }
            webhooks::IncomingWebhookEvent::FrmRejected => {
                Box::pin(payments::payments_core::<
                    api::Void,
                    api::PaymentsResponse,
                    _,
                    _,
                    _,
                    payments::PaymentData<api::Void>,
                >(
                    state.clone(),
                    req_state,
                    platform.clone(),
                    None,
                    payments::PaymentReject,
                    api::PaymentsCancelRequest {
                        payment_id: payment_attempt.payment_id.clone(),
                        cancellation_reason: Some(
                            "Rejected by merchant based on FRM decision".to_string(),
                        ),
                        ..Default::default()
                    },
                    services::api::AuthFlow::Merchant,
                    payments::CallConnectorAction::Trigger,
                    None,
                    None,
                    HeaderPayload::default(),
                    None,
                ))
                .await?
            }
            _ => Err(errors::ApiErrorResponse::EventNotFound)?,
        };
        match payment_response {
            services::ApplicationResponse::JsonWithHeaders((payments_response, _)) => {
                let payment_id = payments_response.payment_id.clone();
                let status = payments_response.status;
                let event_type: Option<enums::EventType> = payments_response.status.into();
                if let Some(outgoing_event_type) = event_type {
                    let primary_object_created_at = payments_response.created;
                    let webhook_recipient = utils::resolve_webhook_recipient_from_created_by(
                        &state,
                        &platform,
                        &business_profile,
                        payment_attempt.created_by.as_ref(),
                    )
                    .await?;
                    Box::pin(super::create_event_and_trigger_outgoing_webhook(
                        state,
                        platform,
                        outgoing_event_type,
                        enums::EventClass::Payments,
                        payment_id.get_string_repr().to_owned(),
                        enums::EventObjectType::PaymentDetails,
                        api::OutgoingWebhookContent::PaymentDetails(Box::new(payments_response)),
                        primary_object_created_at,
                        webhook_recipient,
                        None,
                        business_profile,
                    ))
                    .await?;
                };
                let response = WebhookResponseTracker::Payment { payment_id, status };
                Ok(response)
            }
            _ => Err(errors::ApiErrorResponse::WebhookProcessingFailure).attach_printable(
                "Did not get payment id as object reference id in webhook payments flow",
            )?,
        }
    } else {
        logger::error!("Webhook source verification failed for frm webhooks flow");
        Err(report!(
            errors::ApiErrorResponse::WebhookAuthenticationFailed
        ))
    }
}

#[allow(clippy::too_many_arguments)]
#[instrument(skip_all)]
async fn disputes_incoming_webhook_flow(
    state: SessionState,
    platform: domain::Platform,
    business_profile: domain::Profile,
    webhook_details: api::IncomingWebhookDetails,
    source_verified: bool,
    connector: &ConnectorEnum,
    request_details: &IncomingWebhookRequestDetails<'_>,
    event_type: webhooks::IncomingWebhookEvent,
    webhook_resource_data: Option<WebhookResourceData>,
) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
    metrics::INCOMING_DISPUTE_WEBHOOK_METRIC.add(1, &[]);
    if source_verified {
        let db = &*state.store;
        // Reuse the pre-fetched payment attempt when available; otherwise fetch now.
        let payment_attempt =
            if let Some(WebhookResourceData::Payment { payment_attempt }) = webhook_resource_data {
                payment_attempt
            } else {
                get_payment_attempt_from_object_reference_id(
                    &state,
                    webhook_details.object_reference_id.clone(),
                    platform.get_processor(),
                )
                .await?
            };
        let resource_data = WebhookResourceData::Payment {
            payment_attempt: payment_attempt.clone(),
        };
        let dispute_details = connector
            .get_dispute_details(request_details, Some(&WebhookContext::from(&resource_data)))
            .switch()?;

        let option_dispute = db
            .find_by_processor_merchant_id_payment_id_connector_dispute_id(
                platform.get_processor().get_account().get_id(),
                &payment_attempt.payment_id,
                &dispute_details.connector_dispute_id,
                platform.get_processor().get_account().storage_scheme,
            )
            .await
            .to_not_found_response(errors::ApiErrorResponse::WebhookResourceNotFound)?;
        let dispute_status = common_enums::DisputeStatus::foreign_try_from(event_type)
            .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
            .attach_printable("event type to dispute status mapping failed")?;

        let dispute_object = get_or_update_dispute_object(
            state.clone(),
            option_dispute.clone(),
            dispute_details,
            &platform,
            &payment_attempt,
            dispute_status,
            &business_profile,
            connector.id(),
        )
        .await?;

        //Updating Payment Intent State Metadata if Dispute is lost and customer got their funds back
        if diesel_models::dispute::Dispute::is_not_lost_or_none(&option_dispute)
            && dispute_object.dispute_status == common_enums::DisputeStatus::DisputeLost
        {
            tokio::spawn({
                let state = state.clone();
                let platform = platform.clone();
                let payment_intent = db
                    .find_payment_intent_by_payment_id_processor_merchant_id(
                        &payment_attempt.payment_id,
                        platform.get_processor().get_account().get_id(),
                        platform.get_processor().get_key_store(),
                        platform.get_processor().get_account().storage_scheme,
                    )
                    .await
                    .change_context(errors::ApiErrorResponse::InternalServerError)
                    .attach_printable("Failed to fetch payment_intent")?;
                let dispute_object = dispute_object.clone();
                let state_metadata = payment_intent.state_metadata.clone().unwrap_or_default();

                async move {
                    if let Err(err) = PaymentIntentStateMetadataExt::from(state_metadata)
                        .update_intent_state_metadata_for_dispute(
                            &state,
                            &platform,
                            payment_intent,
                            &dispute_object,
                        )
                        .await
                    {
                        logger::error!(
                            ?err,
                            "Failed to update payment intent state metadata after dispute lost"
                        );
                    }
                }
            });
        }

        let disputes_response = Box::new(dispute_object.clone().foreign_into());
        let event_type: enums::EventType = dispute_object.dispute_status.into();

        let webhook_recipient = utils::resolve_webhook_recipient_from_created_by(
            &state,
            &platform,
            &business_profile,
            payment_attempt.created_by.as_ref(),
        )
        .await?;

        Box::pin(super::create_event_and_trigger_outgoing_webhook(
            state,
            platform,
            event_type,
            enums::EventClass::Disputes,
            dispute_object.dispute_id.clone(),
            enums::EventObjectType::DisputeDetails,
            api::OutgoingWebhookContent::DisputeDetails(disputes_response),
            Some(dispute_object.created_at),
            webhook_recipient,
            None,
            business_profile,
        ))
        .await?;
        metrics::INCOMING_DISPUTE_WEBHOOK_MERCHANT_NOTIFIED_METRIC.add(1, &[]);
        Ok(WebhookResponseTracker::Dispute {
            dispute_id: dispute_object.dispute_id,
            payment_id: dispute_object.payment_id,
            status: dispute_object.dispute_status,
        })
    } else {
        metrics::INCOMING_DISPUTE_WEBHOOK_SIGNATURE_FAILURE_METRIC.add(1, &[]);
        Err(report!(
            errors::ApiErrorResponse::WebhookAuthenticationFailed
        ))
    }
}

#[instrument(skip_all)]
async fn bank_transfer_webhook_flow(
    state: SessionState,
    req_state: ReqState,
    platform: domain::Platform,
    business_profile: domain::Profile,
    webhook_details: api::IncomingWebhookDetails,
    source_verified: bool,
) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
    let (response, created_by, payment_attempt) = if source_verified {
        let payment_attempt = get_payment_attempt_from_object_reference_id(
            &state,
            webhook_details.object_reference_id,
            platform.get_processor(),
        )
        .await?;
        let payment_id = payment_attempt.payment_id.clone();
        let created_by = payment_attempt.created_by.clone();
        let request = api::PaymentsRequest {
            payment_id: Some(api_models::payments::PaymentIdType::PaymentIntentId(
                payment_id,
            )),
            payment_token: payment_attempt.payment_token.clone(),
            ..Default::default()
        };
        let response = Box::pin(payments::payments_core::<
            api::Authorize,
            api::PaymentsResponse,
            _,
            _,
            _,
            payments::PaymentData<api::Authorize>,
        >(
            state.clone(),
            req_state,
            platform.to_owned(),
            None,
            payments::PaymentConfirm,
            request,
            services::api::AuthFlow::Merchant,
            payments::CallConnectorAction::Trigger,
            None,
            None,
            HeaderPayload::with_source(common_enums::PaymentSource::Webhook),
            None,
        ))
        .await;
        (response, created_by, Some(payment_attempt.clone()))
    } else {
        (
            Err(report!(
                errors::ApiErrorResponse::WebhookAuthenticationFailed
            )),
            None,
            None,
        )
    };

    match response? {
        services::ApplicationResponse::JsonWithHeaders((payments_response, _)) => {
            let payment_id = payments_response.payment_id.clone();

            let event_type: Option<enums::EventType> = payments_response.status.into();
            let status = payments_response.status;

            // If event is NOT an UnsupportedEvent, trigger Outgoing Webhook
            if let Some(outgoing_event_type) = event_type {
                let primary_object_created_at = payments_response.created;
                let webhook_recipient = utils::resolve_webhook_recipient_from_created_by(
                    &state,
                    &platform,
                    &business_profile,
                    created_by.as_ref(),
                )
                .await?;
                Box::pin(super::create_event_and_trigger_outgoing_webhook(
                    state,
                    platform,
                    outgoing_event_type,
                    enums::EventClass::Payments,
                    payment_id.get_string_repr().to_owned(),
                    enums::EventObjectType::PaymentDetails,
                    api::OutgoingWebhookContent::PaymentDetails(Box::new(payments_response)),
                    primary_object_created_at,
                    webhook_recipient,
                    payment_attempt.map(|pa| WebhookResourceData::Payment {
                        payment_attempt: pa,
                    }),
                    business_profile,
                ))
                .await?;
            }

            Ok(WebhookResponseTracker::Payment { payment_id, status })
        }

        _ => Err(errors::ApiErrorResponse::WebhookProcessingFailure)
            .attach_printable("received non-json response from payments core")?,
    }
}

pub fn get_connector_by_connector_name(
    state: &SessionState,
    connector_name: &str,
    merchant_connector_id: Option<common_utils::id_type::MerchantConnectorAccountId>,
) -> CustomResult<(ConnectorEnum, String), errors::ApiErrorResponse> {
    let authentication_connector =
        api_models::enums::convert_authentication_connector(connector_name);
    #[cfg(feature = "frm")]
    {
        let frm_connector = api_models::enums::convert_frm_connector(connector_name);
        if frm_connector.is_some() {
            let frm_connector_data =
                api::FraudCheckConnectorData::get_connector_by_name(connector_name)?;
            return Ok((
                frm_connector_data.connector,
                frm_connector_data.connector_name.to_string(),
            ));
        }
    }

    let (connector, connector_name) = if authentication_connector.is_some() {
        let authentication_connector_data =
            api::AuthenticationConnectorData::get_connector_by_name(connector_name)?;
        (
            authentication_connector_data.connector,
            authentication_connector_data.connector_name.to_string(),
        )
    } else {
        let connector_data = ConnectorData::get_connector_by_name(
            &state.conf.connectors,
            connector_name,
            GetToken::Connector,
            merchant_connector_id,
        )
        .change_context(errors::ApiErrorResponse::InvalidRequestData {
            message: "invalid connector name received".to_string(),
        })
        .attach_printable("Failed construction of ConnectorData")?;
        (
            connector_data.connector,
            connector_data.connector_name.to_string(),
        )
    };
    Ok((connector, connector_name))
}

/// This function fetches the merchant connector account ( if the url used is /{merchant_connector_id})
/// if merchant connector id is not passed in the request, then this will return None for mca
async fn fetch_optional_mca_and_connector(
    state: &SessionState,
    platform: &domain::Platform,
    connector_name_or_mca_id: &str,
) -> CustomResult<
    (
        Option<domain::MerchantConnectorAccount>,
        ConnectorEnum,
        String,
    ),
    errors::ApiErrorResponse,
> {
    let db = &state.store;
    if connector_name_or_mca_id.starts_with("mca_") {
        #[cfg(feature = "v1")]
        let mca = db
            .find_by_merchant_connector_account_merchant_id_merchant_connector_id(
                platform.get_processor().get_account().get_id(),
                &common_utils::id_type::MerchantConnectorAccountId::wrap(
                    connector_name_or_mca_id.to_owned(),
                )
                .change_context(errors::ApiErrorResponse::InternalServerError)
                .attach_printable(
                    "Error while converting MerchanConnectorAccountId from string
                    ",
                )?,
                platform.get_processor().get_key_store(),
            )
            .await
            .to_not_found_response(errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
                id: connector_name_or_mca_id.to_string(),
            })
            .attach_printable(
                "error while fetching merchant_connector_account from connector_id",
            )?;

        #[cfg(feature = "v2")]
        let mca: domain::MerchantConnectorAccount = {
            let _ = merchant_account;
            let _ = key_store;
            let _ = db;
            todo!()
        };

        let (connector, connector_name) =
            get_connector_by_connector_name(state, &mca.connector_name, Some(mca.get_id()))?;

        Ok((Some(mca), connector, connector_name))
    } else {
        // Merchant connector account is already being queried, it is safe to set connector id as None
        let (connector, connector_name) =
            get_connector_by_connector_name(state, connector_name_or_mca_id, None)?;
        Ok((None, connector, connector_name))
    }
}

fn should_update_connector_mandate_details(
    source_verified: bool,
    event_type: webhooks::IncomingWebhookEvent,
) -> bool {
    source_verified && event_type == webhooks::IncomingWebhookEvent::PaymentIntentSuccess
}

async fn update_additional_payment_method_data(
    state: &SessionState,
    platform: &domain::Platform,
    connector: &ConnectorEnum,
    request_details: &IncomingWebhookRequestDetails<'_>,
    payment_method_id: String,
) -> CustomResult<(), errors::ApiErrorResponse> {
    let payment_method_update = connector
        .get_additional_payment_method_data(request_details)
        .change_context(errors::ApiErrorResponse::InternalServerError)?
        .ok_or(errors::ApiErrorResponse::InternalServerError)?;

    let db = state.store.as_ref();

    let pm = db
        .find_payment_method(
            platform.get_provider().get_key_store(),
            payment_method_id.as_str(),
            platform.get_provider().get_account().storage_scheme,
        )
        .await
        .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;

    if pm.locker_id.is_some() {
        return Err(report!(errors::ApiErrorResponse::NotSupported {
            message: "Cannot proceed to update Payment Method when locker_id is already present"
                .into(),
        }));
    }

    Box::pin(cards::update_customer_payment_method(
        state.clone(),
        platform.get_provider().clone(),
        platform.get_initiator().cloned(),
        payment_method_update,
        &payment_method_id,
        Some(pm),
    ))
    .await?;
    Ok(())
}

async fn update_connector_mandate_details(
    state: &SessionState,
    platform: &domain::Platform,
    object_ref_id: api::ObjectReferenceId,
    connector: &ConnectorEnum,
    request_details: &IncomingWebhookRequestDetails<'_>,
) -> CustomResult<(), errors::ApiErrorResponse> {
    let webhook_connector_mandate_details = connector
        .get_mandate_details(request_details)
        .switch()
        .attach_printable("Could not find connector mandate details in incoming webhook body")?;

    let webhook_connector_network_transaction_id = connector
        .get_network_txn_id(request_details)
        .switch()
        .attach_printable(
            "Could not find connector network transaction id in incoming webhook body",
        )?;

    // Either one OR both of the fields are present
    if webhook_connector_mandate_details.is_some()
        || webhook_connector_network_transaction_id.is_some()
    {
        let payment_attempt = get_payment_attempt_from_object_reference_id(
            state,
            object_ref_id,
            platform.get_processor(),
        )
        .await?;
        if let Some(ref payment_method_id) = payment_attempt.payment_method_id {
            let payment_method_info = state
                .store
                .find_payment_method(
                    platform.get_provider().get_key_store(),
                    payment_method_id,
                    platform.get_provider().get_account().storage_scheme,
                )
                .await
                .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;

            // Update connector's mandate details
            let updated_connector_mandate_details =
                if let Some(webhook_mandate_details) = webhook_connector_mandate_details {
                    let mandate_details = payment_method_info
                        .get_common_mandate_reference()
                        .change_context(errors::ApiErrorResponse::InternalServerError)
                        .attach_printable("Failed to deserialize to Payment Mandate Reference")?;

                    let merchant_connector_account_id = payment_attempt
                        .merchant_connector_id
                        .clone()
                        .get_required_value("merchant_connector_id")?;

                    if mandate_details.payments.as_ref().is_none_or(|payments| {
                        !payments.0.contains_key(&merchant_connector_account_id)
                    }) {
                        // Update the payment attempt to maintain consistency across tables.
                        let (mandate_metadata, connector_mandate_request_reference_id) =
                            payment_attempt
                                .connector_mandate_detail
                                .as_ref()
                                .map(|details| {
                                    (
                                        details.mandate_metadata.clone(),
                                        details.connector_mandate_request_reference_id.clone(),
                                    )
                                })
                                .unwrap_or((None, None));

                        let connector_mandate_reference_id = ConnectorMandateReferenceId {
                            connector_mandate_id: Some(
                                webhook_mandate_details
                                    .connector_mandate_id
                                    .peek()
                                    .to_string(),
                            ),
                            payment_method_id: Some(payment_method_id.to_string()),
                            mandate_metadata,
                            connector_mandate_request_reference_id,
                        };

                        let attempt_update =
                            storage::PaymentAttemptUpdate::ConnectorMandateDetailUpdate {
                                connector_mandate_detail: Some(connector_mandate_reference_id),
                                tokenization: None, // We need to handle tokenization field update in webhooks as well
                                updated_by: platform
                                    .get_processor()
                                    .get_account()
                                    .storage_scheme
                                    .to_string(),
                            };

                        state
                            .store
                            .update_payment_attempt_with_attempt_id(
                                payment_attempt.clone(),
                                attempt_update,
                                platform.get_processor().get_account().storage_scheme,
                                platform.get_processor().get_key_store(),
                            )
                            .await
                            .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;

                        insert_mandate_details(
                            &payment_attempt,
                            &webhook_mandate_details,
                            Some(mandate_details),
                        )?
                    } else {
                        logger::info!(
                        "Skipping connector mandate details update since they are already present."
                    );
                        None
                    }
                } else {
                    None
                };

            let connector_mandate_details_value = updated_connector_mandate_details
                .map(|common_mandate| {
                    common_mandate.get_mandate_details_value().map_err(|err| {
                        router_env::logger::error!(
                            "Failed to get get_mandate_details_value : {:?}",
                            err
                        );
                        errors::ApiErrorResponse::MandateUpdateFailed
                    })
                })
                .transpose()?;

            let pm_update = diesel_models::PaymentMethodUpdate::ConnectorNetworkTransactionIdAndMandateDetailsUpdate {
                connector_mandate_details: connector_mandate_details_value.map(Secret::new),
                network_transaction_id: webhook_connector_network_transaction_id
                    .map(|webhook_network_transaction_id| webhook_network_transaction_id.get_id().clone()),
                last_modified_by: platform.get_initiator().and_then(|initiator| initiator.to_created_by()).map(|last_modified_by| last_modified_by.to_string()),
            };
            let compat_action = payment_methods::payment_method_modular_compat_action(
                state,
                &payment_method_info.merchant_id,
                &platform.get_provider().get_account().organization_id,
                payment_method_info.customer_id.as_ref(),
            )
            .await;

            state
                .store
                .update_payment_method(
                    platform.get_provider().get_key_store(),
                    payment_method_info,
                    pm_update,
                    platform.get_provider().get_account().storage_scheme,
                    compat_action,
                )
                .await
                .change_context(errors::ApiErrorResponse::InternalServerError)
                .attach_printable("Failed to update payment method in db")?;
        }
    }
    Ok(())
}

fn insert_mandate_details(
    payment_attempt: &PaymentAttempt,
    webhook_mandate_details: &hyperswitch_domain_models::router_flow_types::ConnectorMandateDetails,
    payment_method_mandate_details: Option<CommonMandateReference>,
) -> CustomResult<Option<CommonMandateReference>, errors::ApiErrorResponse> {
    let (mandate_metadata, connector_mandate_request_reference_id) = payment_attempt
        .connector_mandate_detail
        .clone()
        .map(|mandate_reference| {
            (
                mandate_reference.mandate_metadata,
                mandate_reference.connector_mandate_request_reference_id,
            )
        })
        .unwrap_or((None, None));
    let connector_mandate_details = tokenization::update_connector_mandate_details(
        payment_method_mandate_details,
        payment_attempt.payment_method_type,
        Some(
            payment_attempt
                .net_amount
                .get_total_amount()
                .get_amount_as_i64(),
        ),
        payment_attempt.currency,
        payment_attempt.merchant_connector_id.clone(),
        Some(
            webhook_mandate_details
                .connector_mandate_id
                .peek()
                .to_string(),
        ),
        mandate_metadata,
        connector_mandate_request_reference_id,
    )?;
    Ok(connector_mandate_details)
}

/// Runs the Unified Authentication Service path and returns `WebhookOutcome`
/// so `incoming_webhooks_core` handles all paths through one
/// `Skipped`/`Processed` call site.
#[cfg(feature = "v1")]
pub async fn process_uas_incoming_webhook<'a>(
    state: &'a SessionState,
    platform: &'a domain::Platform,
    request_details: &'a IncomingWebhookRequestDetails<'a>,
    connector_name: String,
    uas_connector: ConnectorEnum,
) -> errors::RouterResult<super::gateway::WebhookOutcome> {
    use hyperswitch_masking::ErasedMaskSerialize;

    let dimensions = dimension_state::Dimensions::new()
        .with_processor_merchant_id(platform.get_processor().get_processor_merchant_id())
        .with_provider_merchant_id(platform.get_provider().get_provider_merchant_id())
        .with_organization_id(
            platform
                .get_processor()
                .get_account()
                .organization_id
                .clone(),
        );
    let routing_region = uas_utils::fetch_routing_region_for_uas(state, &dimensions).await;
    let webhook_data =
        uas_utils::get_webhook_request_data_for_uas(request_details, Some(routing_region));

    let webhook_router_data: hyperswitch_domain_models::types::UasProcessWebhookRouterData =
        uas_utils::construct_uas_webhook_router_data(
            state,
            connector_name.to_string(),
            webhook_data,
            None,
        )?;

    let response = Box::pin(uas_utils::do_auth_connector_call(
        state,
        UNIFIED_AUTHENTICATION_SERVICE.to_string(),
        webhook_router_data,
    ))
    .await?;

    let response_body = match response.response {
        Ok(resp) => match resp {
            UasAuthenticationResponseData::Webhook {
                trans_status,
                authentication_value,
                eci,
                three_ds_server_transaction_id,
                authentication_id,
                results_request,
                results_response,
            } => Ok(UasWebhookResponse {
                trans_status,
                authentication_value,
                eci,
                three_ds_server_transaction_id,
                authentication_id,
                results_request,
                results_response,
            }),
            _ => {
                router_env::logger::error!("received unknown webhook response from uas");
                Err(errors::ApiErrorResponse::WebhookProcessingFailure)
            }
        },
        Err(err) => {
            router_env::logger::error!("error processing webhook {:?}", err);
            Err(errors::ApiErrorResponse::WebhookProcessingFailure)
        }
    }?;

    // Event type is classified from the inbound request: the UAS connector's
    // parser inspects request headers/body to identify the event kind, before
    // the response body from the gRPC call is consumed for downstream steps.
    let event_type = uas_connector
        .get_webhook_event_type(request_details, None)
        .switch()
        .attach_printable("Could not get uas webhook event type")?;

    let body = response_body
        .to_bytes()
        .inspect_err(|err| {
            router_env::logger::error!("error converting uas webhook response to bytes: {}", err);
        })
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("error converting uas webhook response to bytes")?;
    let body_bytes = body.to_vec();

    let ctx = super::gateway::WebhookGatewayContext {
        state: state.clone(),
        platform: platform.clone(),
        connector: uas_connector.clone(),
        connector_name: connector_name.clone(),
        merchant_connector_account: None,
        execution_path: common_enums::ExecutionPath::Direct,
        execution_mode: common_enums::ExecutionMode::NotApplicable,
        ucs_reference: None,
        ucs_event_type: None,
    };

    // Reuse inbound request metadata, but parse using UAS response body.
    let uas_webhook_request = IncomingWebhookRequestDetails {
        method: request_details.method.clone(),
        uri: request_details.uri.clone(),
        headers: request_details.headers,
        query_params: request_details.query_params.clone(),
        body: &body_bytes,
    };

    if matches!(
        super::gateway::FilterDecision::evaluate(event_type, &ctx).await,
        super::gateway::FilterDecision::Skip
    ) {
        let ack_response = uas_connector
            .get_webhook_api_response(&uas_webhook_request, None, None)
            .switch()
            .attach_printable("Failed to build UAS webhook ack")?;
        return Ok(super::gateway::WebhookOutcome::Skipped {
            reference: None,
            event_type,
            ack_response,
        });
    }

    let reference = uas_connector
        .get_webhook_object_reference_id(&uas_webhook_request)
        .switch()
        .attach_printable("Could not find object reference id in UAS webhook body")?;

    let mca = Box::pin(helper_utils::get_mca_from_object_reference_id(
        state,
        reference.clone(),
        platform,
        connector_name.as_str(),
    ))
    .await?;

    let source_verified =
        super::gateway::verify_webhook_source_via_connector(&ctx, &uas_webhook_request, &mca)
            .await?;

    let resource_object = uas_connector
        .get_webhook_resource_object(&uas_webhook_request)
        .switch()
        .attach_printable("Failed to extract UAS webhook resource object")?;
    let masked_log_payload =
        Secret::new(resource_object.masked_serialize().unwrap_or_else(|error| {
            router_env::logger::warn!(
                ?error,
                "Failed to mask-serialize UAS webhook resource object for logging"
            );
            serde_json::Value::Null
        }));

    let ack_response = uas_connector
        .get_webhook_api_response(
            &uas_webhook_request,
            None,
            Some(mca.connector_account_details.clone()),
        )
        .switch()
        .attach_printable("Failed to build UAS webhook ack")?;

    Ok(super::gateway::WebhookOutcome::Processed {
        reference,
        event_type,
        source_verified,
        content: super::gateway::WebhookContent::Direct(body_bytes.clone()),
        // Carry the UAS response body so downstream connector calls that parse
        // request_details.body receive the correct structured payload.
        decoded_body: Some(body_bytes),
        // UAS path does not pre-fetch resource data here.
        webhook_resource_data: Box::new(None),
        masked_log_payload,
        merchant_connector_account: Box::new(mca),
        ack_response,
    })
}
