use std::marker::PhantomData;

use base64::Engine;
use common_utils::{
    consts,
    crypto::{self, GenerateDigest},
    errors::CustomResult,
    ext_traits::ValueExt,
    fp_utils,
};
use error_stack::{Report, ResultExt};
use hyperswitch_domain_models::{
    router_request_types::VerifyWebhookSourceRequestData,
    router_response_types::{VerifyWebhookSourceResponseData, VerifyWebhookStatus},
};
use hyperswitch_interfaces::webhooks::IncomingWebhook;
use redis_interface as redis;
use router_env::tracing;

use super::{types as webhook_types, MERCHANT_ID};
use crate::{
    core::{
        configs::dimension_state,
        errors::{self},
        metrics,
        payments::{self, helpers},
    },
    errors::RouterResult,
    routes::app::SessionStateInfo,
    services::{self, connector_integration_interface::ConnectorEnum, logger},
    types::{
        self,
        api::{self, ConnectorData, GetToken},
        domain, PaymentAddress,
    },
    SessionState,
};

const IRRELEVANT_ATTEMPT_ID_IN_SOURCE_VERIFICATION_FLOW: &str =
    "irrelevant_attempt_id_in_source_verification_flow";
const IRRELEVANT_CONNECTOR_REQUEST_REFERENCE_ID_IN_SOURCE_VERIFICATION_FLOW: &str =
    "irrelevant_connector_request_reference_id_in_source_verification_flow";

/// Check whether the merchant has configured to disable the webhook `event` for the `connector`
/// Uses superposition with dimensions [merchant_id, connector, incoming_webhook_events]
pub async fn is_webhook_event_disabled(
    state: &SessionState,
    connector: common_enums::connector_enums::Connector,
    dimensions: &dimension_state::DimensionsWithProcessorAndProviderMerchantId,
    event: &api::IncomingWebhookEvent,
) -> bool {
    let dimensions = dimensions
        .with_connector(connector)
        .with_incoming_webhook_event(*event);

    dimensions
        .get_incoming_webhook_disabled_events(
            state.store.as_ref(),
            state.superposition_service.as_ref(),
            None,
        )
        .await
}

pub async fn construct_webhook_router_data(
    state: &SessionState,
    connector_name: &str,
    merchant_connector_account: domain::MerchantConnectorAccount,
    platform: &domain::Platform,
    connector_wh_secrets: &api_models::webhooks::ConnectorWebhookSecrets,
    request_details: &api::IncomingWebhookRequestDetails<'_>,
) -> CustomResult<types::VerifyWebhookSourceRouterData, errors::ApiErrorResponse> {
    let auth_type: types::ConnectorAuthType =
        helpers::MerchantConnectorAccountType::DbVal(Box::new(merchant_connector_account.clone()))
            .get_connector_account_details()
            .parse_value("ConnectorAuthType")
            .change_context(errors::ApiErrorResponse::InternalServerError)?;

    let router_data = types::RouterData {
        flow: PhantomData,
        merchant_id: platform.get_processor().get_account().get_id().clone(),
        connector: connector_name.to_string(),
        customer_id: None,
        tenant_id: state.tenant.tenant_id.clone(),
        payment_id: common_utils::id_type::PaymentId::get_irrelevant_id("source_verification_flow")
            .get_string_repr()
            .to_owned(),
        attempt_id: IRRELEVANT_ATTEMPT_ID_IN_SOURCE_VERIFICATION_FLOW.to_string(),
        status: diesel_models::enums::AttemptStatus::default(),
        payment_method: diesel_models::enums::PaymentMethod::default(),
        payment_method_type: None,
        connector_auth_type: auth_type,
        description: None,
        address: PaymentAddress::default(),
        auth_type: diesel_models::enums::AuthenticationType::default(),
        connector_meta_data: None,
        connector_wallets_details: None,
        amount_captured: None,
        minor_amount_captured: None,
        request: VerifyWebhookSourceRequestData {
            webhook_headers: request_details.headers.clone(),
            webhook_body: request_details.body.to_vec().clone(),
            merchant_secret: connector_wh_secrets.to_owned(),
            webhook_uri: request_details.uri.clone(),
        },
        response: Err(types::ErrorResponse::default()),
        access_token: None,
        session_token: None,
        reference_id: None,
        payment_method_token: None,
        connector_customer: None,
        recurring_mandate_payment_data: None,
        preprocessing_id: None,
        connector_request_reference_id:
            IRRELEVANT_CONNECTOR_REQUEST_REFERENCE_ID_IN_SOURCE_VERIFICATION_FLOW.to_string(),
        #[cfg(feature = "payouts")]
        payout_method_data: None,
        #[cfg(feature = "payouts")]
        quote_id: None,
        test_mode: None,
        payment_method_balance: None,
        payment_method_status: None,
        connector_api_version: None,
        connector_http_status_code: None,
        external_latency: None,
        apple_pay_flow: None,
        frm_metadata: None,
        refund_id: None,
        dispute_id: None,
        payout_id: None,
        connector_response: None,
        integrity_check: Ok(()),
        additional_merchant_data: None,
        header_payload: None,
        connector_mandate_request_reference_id: None,
        authentication_id: None,
        psd2_sca_exemption_type: None,
        raw_connector_response: None,
        is_payment_id_from_merchant: None,
        l2_l3_data: None,
        minor_amount_capturable: None,
        authorized_amount: None,
        customer_document_details: None,
        feature_data: None,
        sender_payment_instrument_id: None,
    };
    Ok(router_data)
}

/// Makes an outbound HTTP call to the connector's source-verification endpoint.
/// Used when the connector requires a callback-style verification rather than a
/// local HMAC check. Only called for connectors listed in
/// `webhook_source_verification_call.connectors_with_webhook_source_verification_call`.
pub(super) async fn verify_webhook_source_verification_call(
    connector: ConnectorEnum,
    state: &SessionState,
    platform: &domain::Platform,
    merchant_connector_account: domain::MerchantConnectorAccount,
    connector_name: &str,
    request_details: &api::IncomingWebhookRequestDetails<'_>,
) -> CustomResult<bool, errors::ConnectorError> {
    let connector_data = ConnectorData::get_connector_by_name(
        &state.conf.connectors,
        connector_name,
        GetToken::Connector,
        None,
    )
    .change_context(errors::ConnectorError::WebhookSourceVerificationFailed)
    .attach_printable("invalid connector name received in payment attempt")?;
    let connector_integration: services::BoxedWebhookSourceVerificationConnectorIntegrationInterface<
        hyperswitch_domain_models::router_flow_types::VerifyWebhookSource,
        VerifyWebhookSourceRequestData,
        VerifyWebhookSourceResponseData,
    > = connector_data.connector.get_connector_integration();
    let connector_webhook_secrets = connector
        .get_webhook_source_verification_merchant_secret(
            platform.get_processor().get_account().get_id(),
            connector_name,
            merchant_connector_account.connector_webhook_details.clone(),
        )
        .await
        .change_context(errors::ConnectorError::WebhookSourceVerificationFailed)?;

    let router_data = construct_webhook_router_data(
        state,
        connector_name,
        merchant_connector_account,
        platform,
        &connector_webhook_secrets,
        request_details,
    )
    .await
    .change_context(errors::ConnectorError::WebhookSourceVerificationFailed)
    .attach_printable("Failed while constructing webhook router data")?;

    let response = services::execute_connector_processing_step(
        state,
        connector_integration,
        &router_data,
        payments::CallConnectorAction::Trigger,
        None,
        None,
    )
    .await?;

    let verification_result = response
        .response
        .map(|response| response.verify_webhook_status);
    match verification_result {
        Ok(VerifyWebhookStatus::SourceVerified) => Ok(true),
        Ok(VerifyWebhookStatus::SourceNotVerified) => Ok(false),
        Err(err) => {
            tracing::error!(?err, "Webhook source verification failed");
            Ok(false)
        }
    }
}

#[inline]
pub(crate) fn get_idempotent_event_id(
    primary_object_id: &str,
    event_type: types::storage::enums::EventType,
    delivery_attempt: types::storage::enums::WebhookDeliveryAttempt,
) -> Result<String, Report<errors::WebhooksFlowError>> {
    use crate::types::storage::enums::WebhookDeliveryAttempt;

    const EVENT_ID_SUFFIX_LENGTH: usize = 8;

    let common_prefix = format!("{primary_object_id}_{event_type}");

    // Hash the common prefix with SHA256 and encode with URL-safe base64 without padding
    let digest = crypto::Sha256
        .generate_digest(common_prefix.as_bytes())
        .change_context(errors::WebhooksFlowError::IdGenerationFailed)
        .attach_printable("Failed to generate idempotent event ID")?;
    let base_encoded = consts::BASE64_ENGINE_URL_SAFE_NO_PAD.encode(digest);

    let result = match delivery_attempt {
        WebhookDeliveryAttempt::InitialAttempt => base_encoded,
        WebhookDeliveryAttempt::AutomaticRetry | WebhookDeliveryAttempt::ManualRetry => {
            common_utils::generate_id(EVENT_ID_SUFFIX_LENGTH, &base_encoded)
        }
    };

    Ok(result)
}

#[inline]
pub(crate) fn generate_event_id() -> String {
    common_utils::generate_time_ordered_id("evt")
}

pub(crate) fn increment_webhook_outgoing_received_count(
    recipient_data: &webhook_types::WebhookRecipientData,
) {
    let attributes = match recipient_data {
        webhook_types::WebhookRecipientData::Connector {
            merchant_connector_id,
            ..
        } => router_env::metric_attributes!((
            crate::core::webhooks::MERCHANT_CONNECTOR_ACCOUNT_ID,
            merchant_connector_id.clone()
        )),

        webhook_types::WebhookRecipientData::Merchant { merchant_id, .. } => {
            router_env::metric_attributes!((MERCHANT_ID, merchant_id.clone()))
        }
    };

    metrics::WEBHOOK_OUTGOING_RECEIVED_COUNT.add(1, attributes);
}

pub(crate) fn increment_webhook_outgoing_not_received_count(
    recipient_data: &webhook_types::WebhookRecipientData,
) {
    let attributes = match recipient_data {
        webhook_types::WebhookRecipientData::Connector {
            merchant_connector_id,
            ..
        } => router_env::metric_attributes!((
            crate::core::webhooks::MERCHANT_CONNECTOR_ACCOUNT_ID,
            merchant_connector_id.clone()
        )),

        webhook_types::WebhookRecipientData::Merchant { merchant_id } => {
            router_env::metric_attributes!((MERCHANT_ID, merchant_id.clone()))
        }
    };
    metrics::WEBHOOK_OUTGOING_NOT_RECEIVED_COUNT.add(1, attributes);
}

pub fn is_outgoing_webhook_disabled(
    state: &SessionState,
    webhook_url_result: &Result<String, Report<errors::WebhooksFlowError>>,
    business_profile: &domain::Profile,
    idempotent_event_id: &str,
) -> bool {
    if !state.conf.webhooks.outgoing_enabled
        || webhook_url_result.is_err()
        || webhook_url_result.as_ref().is_ok_and(String::is_empty)
    {
        logger::debug!(
            business_profile_id=?business_profile.get_id(),
            %idempotent_event_id,
            "Outgoing webhooks are disabled in application configuration, or merchant webhook URL \
             could not be obtained; skipping outgoing webhooks for event"
        );
        return true;
    }
    false
}

/// Context resolved for outgoing webhook delivery, containing the recipient's
/// merchant account, keystore, and business profile.
pub(crate) struct WebhookRecipientContext {
    /// The webhook recipient's merchant account.
    pub merchant_account: domain::MerchantAccount,
    /// The webhook recipient's keystore (used for encryption/decryption).
    pub key_store: domain::MerchantKeyStore,
    /// The business profile from which the webhook URL and config are read.
    pub profile: domain::Profile,
}

/// Resolves the webhook recipient from the `created_by` field on the resource.
///
/// Used in the incoming webhook flow where the Platform struct has no initiator populated.
/// Falls back to processor/connected merchant when `created_by` is absent.
///
/// Only `CreatedBy::Api` is considered for platform initiation — `EmbeddedToken`
/// and `Jwt` variants do not carry platform context.
pub(crate) async fn resolve_webhook_recipient_from_created_by(
    state: &SessionState,
    platform: &domain::Platform,
    profile: &domain::Profile,
    created_by: Option<&common_utils::types::CreatedBy>,
) -> CustomResult<WebhookRecipientContext, errors::ApiErrorResponse> {
    let provider_merchant_id = platform.get_provider().get_account().get_id();
    let processor_merchant_id = platform.get_processor().get_account().get_id();

    let is_platform_initiated = if provider_merchant_id == processor_merchant_id {
        // Standard merchant setup — provider and processor are the same entity.
        false
    } else {
        // Platform setup — check if the creator is the provider (platform merchant).
        created_by
            .map(|created_by| created_by.is_provider_initiated(provider_merchant_id))
            .unwrap_or_default()
    };

    logger::debug!(
        ?created_by,
        provider_merchant_id=?provider_merchant_id,
        processor_merchant_id=?processor_merchant_id,
        is_platform_initiated,
        "Resolving webhook recipient context from created_by field"
    );
    resolve_webhook_recipient_inner(state, platform, profile, is_platform_initiated).await
}

/// Shared resolution logic: fetch the correct account, keystore, and profile
/// based on whether the operation was platform-initiated.
async fn resolve_webhook_recipient_inner(
    state: &SessionState,
    platform: &domain::Platform,
    profile: &domain::Profile,
    is_platform_initiated: bool,
) -> CustomResult<WebhookRecipientContext, errors::ApiErrorResponse> {
    let (merchant_account, key_store, profile) = if is_platform_initiated {
        let provider = platform.get_provider();
        let provider_profile = get_default_profile_for_platform_merchant(state, provider).await?;

        (
            provider.get_account().clone(),
            provider.get_key_store().clone(),
            provider_profile,
        )
    } else {
        let processor = platform.get_processor();

        (
            processor.get_account().clone(),
            processor.get_key_store().clone(),
            profile.clone(),
        )
    };

    Ok(WebhookRecipientContext {
        merchant_account,
        key_store,
        profile,
    })
}

/// Fetches the default business profile for a platform merchant.
///
/// Platform merchants are restricted to a single profile. This function must
/// only be called when the provider is confirmed to be a platform merchant.
///
/// In V1, this uses the `default_profile` field on the merchant account.
/// In V2, this lists all profiles and returns the single configured profile.
#[cfg(feature = "v1")]
async fn get_default_profile_for_platform_merchant(
    state: &SessionState,
    provider: &domain::Provider,
) -> CustomResult<domain::Profile, errors::ApiErrorResponse> {
    // Validate that merchant account type is platform
    fp_utils::when(!provider.get_account().is_platform_account(), || {
        Err(error_stack::report!(
            errors::ApiErrorResponse::InternalServerError
        ))
        .attach_printable("Expected provider merchant to be a platform account")
    })?;

    let profile_id = provider
        .get_account()
        .default_profile
        .as_ref()
        .ok_or(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Platform merchant does not have a default profile configured")?;

    state
        .store
        .find_business_profile_by_profile_id(provider.get_key_store(), profile_id)
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to fetch platform merchant's default business profile")
}

#[cfg(feature = "v2")]
async fn get_default_profile_for_platform_merchant(
    state: &SessionState,
    provider: &domain::Provider,
) -> CustomResult<domain::Profile, errors::ApiErrorResponse> {
    // Validate that merchant account type is platform
    fp_utils::when(!provider.get_account().is_platform_account(), || {
        Err(error_stack::report!(
            errors::ApiErrorResponse::InternalServerError
        ))
        .attach_printable("Expected provider merchant to be a platform account")
    })?;

    let profiles = state
        .store
        .list_profile_by_merchant_id(provider.get_key_store(), provider.get_account().get_id())
        .await
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Failed to list business profiles for platform merchant")?;

    profiles
        .into_iter()
        .next()
        .ok_or(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Platform merchant has no business profiles configured")
}

const WEBHOOK_LOCK_PREFIX: &str = "WEBHOOK_LOCK";

pub(super) async fn perform_redis_lock<A>(
    state: &A,
    unique_locking_key: &str,
    merchant_id: common_utils::id_type::MerchantId,
) -> RouterResult<Option<String>>
where
    A: SessionStateInfo,
{
    let lock_value: String = uuid::Uuid::new_v4().to_string();
    let redis_conn = state
        .store()
        .get_redis_conn()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Error connecting to redis")?;

    let redis_locking_key = format!(
        "{}_{}_{}",
        WEBHOOK_LOCK_PREFIX,
        merchant_id.get_string_repr(),
        unique_locking_key
    );
    let redis_lock_expiry_seconds = state.conf().webhooks.redis_lock_expiry_seconds;

    let redis_lock_result = redis_conn
        .set_key_if_not_exists_with_expiry(
            &redis_locking_key.as_str().into(),
            lock_value.clone(),
            Some(i64::from(redis_lock_expiry_seconds)),
        )
        .await;

    match redis_lock_result {
        Ok(redis::SetnxReply::KeySet) => {
            logger::info!("Lock acquired for for {redis_locking_key}");
            Ok(Some(lock_value))
        }
        Ok(redis::SetnxReply::KeyNotSet) => {
            logger::info!("Lock already held for {redis_locking_key}");
            Ok(None)
        }
        Err(err) => Err(err
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Error acquiring redis lock")),
    }
}

pub(super) async fn free_redis_lock<A>(
    state: &A,
    unique_locking_key: &str,
    merchant_id: common_utils::id_type::MerchantId,
    lock_value: Option<String>,
) -> RouterResult<()>
where
    A: SessionStateInfo,
{
    let redis_conn = state
        .store()
        .get_redis_conn()
        .change_context(errors::ApiErrorResponse::InternalServerError)
        .attach_printable("Error connecting to redis")?;

    let redis_locking_key = format!(
        "{}_{}_{}",
        WEBHOOK_LOCK_PREFIX,
        merchant_id.get_string_repr(),
        unique_locking_key
    );
    match redis_conn
        .get_key::<Option<String>>(&redis_locking_key.as_str().into())
        .await
    {
        Ok(val) => {
            if val == lock_value {
                match redis_conn
                    .delete_key(&redis_locking_key.as_str().into())
                    .await
                {
                    Ok(redis::types::DelReply::KeyDeleted) => {
                        logger::info!("Lock freed {redis_locking_key}");
                        tracing::Span::current().record("redis_lock_released", redis_locking_key);
                        Ok(())
                    }
                    Ok(redis::types::DelReply::KeyNotDeleted) => Err(
                        errors::ApiErrorResponse::InternalServerError,
                    )
                    .attach_printable("Status release lock called but key is not found in redis"),
                    Err(error) => Err(error)
                        .change_context(errors::ApiErrorResponse::InternalServerError)
                        .attach_printable("Error while deleting redis key"),
                }
            } else {
                Err(errors::ApiErrorResponse::InternalServerError)
                            .attach_printable("The redis value which acquired the lock is not equal to the redis value requesting for releasing the lock")
            }
        }
        Err(error) => Err(error)
            .change_context(errors::ApiErrorResponse::InternalServerError)
            .attach_printable("Error while deleting redis key"),
    }
}
