#[cfg(feature = "payouts")]
use api_models::payouts as payout_models;
use api_models::{
    enums::EventType,
    webhook_events::OutgoingWebhookRequestContent,
    webhooks::{OutgoingWebhook, OutgoingWebhookContent},
};
use common_utils::{
    consts::DEFAULT_LOCALE,
    ext_traits::{StringExt, ValueExt},
    id_type,
};
use diesel_models::process_tracker::business_status;
use error_stack::ResultExt;
use hyperswitch_masking::PeekInterface;
use router_env::tracing::{self, instrument};
use scheduler::{
    consumer::{self, workflows::ProcessTrackerWorkflow},
    utils as scheduler_utils,
};
#[cfg(feature = "v1")]
use subscriptions::workflows::invoice_sync;

#[cfg(feature = "payouts")]
use crate::core::payouts;
use crate::{
    core::{
        payments,
        webhooks::{
            self as webhooks_core,
            types::{OutgoingWebhookTrackingData, WebhookRecipientData},
        },
    },
    db::StorageInterface,
    errors, logger,
    routes::{app::ReqState, SessionState},
    types::{domain, storage},
};

pub struct OutgoingWebhookRetryWorkflow;

#[async_trait::async_trait]
impl ProcessTrackerWorkflow<SessionState> for OutgoingWebhookRetryWorkflow {
    #[cfg(feature = "v1")]
    #[instrument(skip_all)]
    async fn execute_workflow<'a>(
        &'a self,
        state: &'a SessionState,
        process: storage::ProcessTracker,
    ) -> Result<(), errors::ProcessTrackerError> {
        let delivery_attempt = storage::enums::WebhookDeliveryAttempt::AutomaticRetry;
        let tracking_data: OutgoingWebhookTrackingData = process
            .tracking_data
            .clone()
            .parse_value("OutgoingWebhookTrackingData")?;

        let db = &*state.store;
        let master_key = &db.get_master_key().to_vec().into();
        let provider_merchant_id = tracking_data.merchant_id.clone();

        // Resolve the webhook recipient's merchant_id for keystore lookup.
        // `initiator_merchant_id` is populated by new deployments; fall back to
        // `merchant_id` for tracking data written by older deployments.
        let webhook_recipient_merchant_id = tracking_data
            .initiator_merchant_id
            .clone()
            .unwrap_or_else(|| tracking_data.merchant_id.clone());
        let processor_merchant_id = tracking_data
            .processor_merchant_id
            .clone()
            .unwrap_or_else(|| tracking_data.merchant_id.clone());

        let recipient_data = tracking_data.recipient_data.clone().unwrap_or_else(|| {
            WebhookRecipientData::Merchant {
                merchant_id: processor_merchant_id.clone(),
            }
        });

        let webhook_key_store = db
            .get_merchant_key_store_by_merchant_id(&webhook_recipient_merchant_id, master_key)
            .await?;
        let business_profile = db
            .find_business_profile_by_profile_id(
                &webhook_key_store,
                &tracking_data.business_profile_id,
            )
            .await?;

        let event_id = webhooks_core::utils::generate_event_id();
        let idempotent_event_id = webhooks_core::utils::get_idempotent_event_id(
            &tracking_data.primary_object_id,
            tracking_data.event_type,
            delivery_attempt,
        )
        .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
        .attach_printable("Failed to generate idempotent event ID")?;

        // Look up the initial event by event_id alone (event_id is globally unique).
        // This works regardless of which merchant owns the event and also handles
        // old tracking data created before `initial_attempt_id` was tracked.
        let initial_attempt_id = tracking_data.initial_attempt_id.clone().unwrap_or_else(|| {
            format!(
                "{}_{}",
                tracking_data.primary_object_id, tracking_data.event_type
            )
        });
        let initial_event = db
            .find_event_by_event_id(&initial_attempt_id, &webhook_key_store)
            .await?;

        let now = common_utils::date_time::now();
        let new_event = domain::Event {
            event_id,
            event_type: initial_event.event_type,
            event_class: initial_event.event_class,
            is_webhook_notified: false,
            primary_object_id: initial_event.primary_object_id,
            primary_object_type: initial_event.primary_object_type,
            created_at: now,
            merchant_id: Some(provider_merchant_id.clone()),
            business_profile_id: Some(business_profile.get_id().to_owned()),
            primary_object_created_at: initial_event.primary_object_created_at,
            idempotent_event_id: Some(idempotent_event_id),
            initial_attempt_id: Some(initial_event.event_id.clone()),
            request: initial_event.request,
            response: None,
            delivery_attempt: Some(delivery_attempt),
            metadata: initial_event.metadata,
            is_overall_delivery_successful: Some(false),
            processor_merchant_id: initial_event
                .processor_merchant_id
                .or(Some(processor_merchant_id.clone())),
            initiator_merchant_id: initial_event
                .initiator_merchant_id
                .or(Some(webhook_key_store.merchant_id.clone())),
            recipient: initial_event.recipient,
        };

        let event = db
            .insert_event(new_event, &webhook_key_store)
            .await
            .inspect_err(|error| {
                logger::error!(?error, "Failed to insert event in events table");
            })?;

        match &event.request {
            Some(request) => {
                let request_content: OutgoingWebhookRequestContent = request
                    .get_inner()
                    .peek()
                    .parse_struct("OutgoingWebhookRequestContent")?;

                Box::pin(webhooks_core::trigger_webhook_and_raise_event(
                    state.clone(),
                    business_profile,
                    &webhook_key_store,
                    provider_merchant_id,
                    processor_merchant_id,
                    event,
                    Some(request_content),
                    delivery_attempt,
                    None,
                    Some(process),
                    recipient_data,
                ))
                .await;
            }

            // Event inserted by an older version of the application, or event created when sending webhooks to a connector
            None => {
                // Fetch provider merchant account and keystore
                let provider_key_store = db
                    .get_merchant_key_store_by_merchant_id(&provider_merchant_id, master_key)
                    .await?;
                let provider_account = db
                    .find_merchant_account_by_merchant_id(
                        &provider_merchant_id,
                        &provider_key_store,
                    )
                    .await?;

                // Fetch processor merchant account and keystore (may be same as provider)
                let (processor_account, processor_key_store) = if provider_merchant_id
                    == processor_merchant_id
                {
                    (provider_account.clone(), provider_key_store.clone())
                } else {
                    let processor_key_store = db
                        .get_merchant_key_store_by_merchant_id(&processor_merchant_id, master_key)
                        .await?;
                    let processor_account = db
                        .find_merchant_account_by_merchant_id(
                            &processor_merchant_id,
                            &processor_key_store,
                        )
                        .await?;
                    (processor_account, processor_key_store)
                };

                let platform = domain::Platform::new(
                    provider_account,
                    provider_key_store,
                    processor_account,
                    processor_key_store,
                    None,
                );

                // TODO: Add request state for the PT flows as well
                // Fetch the current resource information
                let (content, event_type) =
                    if let WebhookRecipientData::Merchant { .. } = recipient_data.clone() {
                        let (content, event_type) =
                            Box::pin(get_outgoing_webhook_content_and_event_type(
                                state.clone(),
                                state.get_req_state(),
                                &platform,
                                &tracking_data,
                            ))
                            .await?;

                        (Some(content), event_type)
                    } else {
                        (None, None)
                    };

                match event_type {
                    // Resource status is same as the event type of the current event
                    Some(event_type) if event_type == tracking_data.event_type => {
                        let outgoing_webhook = content.as_ref().map(|content| OutgoingWebhook {
                            merchant_id: provider_merchant_id.clone(),
                            event_id: event.event_id.clone(),
                            event_type,
                            content: content.clone(),
                            timestamp: event.created_at,
                            processor_merchant_id: Some(processor_merchant_id.clone()),
                        });

                        // Use the webhook recipient's merchant account for request construction.
                        // If the recipient is the provider, use the provider account (platform-initiated);
                        // otherwise use the processor account (connected-merchant-initiated).
                        let webhook_recipient_account =
                            if webhook_recipient_merchant_id == provider_merchant_id {
                                platform.get_provider().get_account()
                            } else {
                                platform.get_processor().get_account()
                            };

                        let request_content = outgoing_webhook
                            .map(|outgoing_webhook_data| {
                                webhooks_core::get_outgoing_webhook_request(
                                    webhook_recipient_account,
                                    outgoing_webhook_data,
                                    &business_profile,
                                )
                            })
                            .transpose()
                            .map_err(|error| {
                                logger::error!(
                                    ?error,
                                    "Failed to obtain outgoing webhook request content"
                                );
                                errors::ProcessTrackerError::EApiErrorResponse
                            })?;

                        Box::pin(webhooks_core::trigger_webhook_and_raise_event(
                            state.clone(),
                            business_profile,
                            &webhook_key_store,
                            provider_merchant_id,
                            processor_merchant_id,
                            event,
                            request_content,
                            delivery_attempt,
                            content,
                            Some(process),
                            recipient_data,
                        ))
                        .await;
                    }
                    // Resource status has changed since the event was created, finish task
                    _ => {
                        logger::warn!(
                            %event.event_id,
                            "The current status of the resource `{:?}` (event type: {:?}) and the status of \
                            the resource when the event was created (event type: {:?}) differ, finishing task",
                            tracking_data.primary_object_id,
                            event_type,
                            tracking_data.event_type
                        );
                        db.as_scheduler()
                            .finish_process_with_business_status(
                                process.clone(),
                                business_status::RESOURCE_STATUS_MISMATCH,
                            )
                            .await?;
                    }
                }
            }
        };

        Ok(())
    }
    #[cfg(feature = "v2")]
    async fn execute_workflow<'a>(
        &'a self,
        _state: &'a SessionState,
        _process: storage::ProcessTracker,
    ) -> Result<(), errors::ProcessTrackerError> {
        todo!()
    }

    #[instrument(skip_all)]
    async fn error_handler<'a>(
        &'a self,
        state: &'a SessionState,
        process: storage::ProcessTracker,
        error: errors::ProcessTrackerError,
    ) -> errors::CustomResult<(), errors::ProcessTrackerError> {
        consumer::consumer_error_handler(state.store.as_scheduler(), process, error).await
    }
}

/// Get the schedule time for the specified retry count.
///
/// The schedule time can be configured in configs with this key: `pt_mapping_outgoing_webhooks`.
///
/// ```json
/// {
///   "default_mapping": {
///     "start_after": 60,
///     "frequency": [300],
///     "count": [5]
///   }
/// }
/// ```
///
/// This configuration value represents:
/// - `default_mapping.start_after`: The first retry attempt should happen after 60 seconds by
///   default.
/// - `default_mapping.frequency` and `count`: The next 5 retries should have an interval of 300
///   seconds between them by default.
#[cfg(feature = "v1")]
#[instrument(skip_all)]
pub(crate) async fn get_webhook_delivery_retry_schedule_time(
    db: &dyn StorageInterface,
    superposition_client: &external_services::superposition::SuperpositionClient,
    dimensions: &crate::core::configs::dimension_state::DimensionsWithProcessorMerchantId,
    retry_count: i32,
) -> Option<time::PrimitiveDateTime> {
    let mapping = dimensions
        .get_pt_mapping_outgoing_webhooks(db, superposition_client, None)
        .await;

    let time_delta =
        scheduler_utils::get_outgoing_webhook_retry_schedule_time(mapping, retry_count);

    scheduler_utils::get_time_from_delta(time_delta)
}

/// This configuration value represents:
/// - `default_mapping.start_after`: The first retry attempt should happen after 60 seconds by
///   default.
/// - `default_mapping.frequency` and `count`: The next 5 retries should have an interval of 300
///   seconds between them by default.
#[cfg(feature = "v1")]
#[instrument(skip_all)]
pub(crate) async fn get_connector_webhook_delivery_retry_schedule_time(
    db: &dyn StorageInterface,
    superposition_client: &external_services::superposition::SuperpositionClient,
    dimensions: &crate::core::configs::dimension_state::DimensionsWithProcessorMerchantIdAndConnector,
    retry_count: i32,
) -> Option<time::PrimitiveDateTime> {
    let mapping = dimensions
        .get_pt_mapping_outgoing_connector_webhooks(db, superposition_client, None)
        .await;

    let time_delta =
        scheduler_utils::get_outgoing_webhook_retry_schedule_time(mapping, retry_count);

    scheduler_utils::get_time_from_delta(time_delta)
}

/// Schedule the webhook delivery task for retry
#[cfg(feature = "v1")]
#[instrument(skip_all)]
pub(crate) async fn retry_webhook_delivery_task(
    db: &dyn StorageInterface,
    merchant_id: &id_type::MerchantId,
    superposition_client: &external_services::superposition::SuperpositionClient,
    process: storage::ProcessTracker,
    recipient_data: WebhookRecipientData,
) -> errors::CustomResult<(), errors::StorageError> {
    let schedule_time = match recipient_data {
        WebhookRecipientData::Merchant { merchant_id } => {
            let dimensions = crate::core::configs::dimension_state::Dimensions::new()
                .with_processor_merchant_id(merchant_id.clone().into());
            get_webhook_delivery_retry_schedule_time(
                db,
                superposition_client,
                &dimensions,
                process.retry_count + 1,
            )
            .await
        }
        WebhookRecipientData::Connector { connector, .. } => {
            let dimensions = crate::core::configs::dimension_state::Dimensions::new()
                .with_processor_merchant_id(merchant_id.clone().into())
                .with_connector(connector);

            get_connector_webhook_delivery_retry_schedule_time(
                db,
                superposition_client,
                &dimensions,
                process.retry_count + 1,
            )
            .await
        }
    };

    match schedule_time {
        Some(schedule_time) => {
            db.as_scheduler()
                .retry_process(process, schedule_time)
                .await
        }
        None => {
            db.as_scheduler()
                .finish_process_with_business_status(process, business_status::RETRIES_EXCEEDED)
                .await
        }
    }
}

#[cfg(feature = "v1")]
#[instrument(skip_all)]
async fn get_outgoing_webhook_content_and_event_type(
    state: SessionState,
    req_state: ReqState,
    platform: &domain::Platform,
    tracking_data: &OutgoingWebhookTrackingData,
) -> Result<(OutgoingWebhookContent, Option<EventType>), errors::ProcessTrackerError> {
    use api_models::{
        disputes::DisputeRetrieveRequest,
        mandates::MandateId,
        payments::{PaymentIdType, PaymentsResponse, PaymentsRetrieveRequest},
        refunds::{RefundResponse, RefundsRetrieveRequest},
    };

    use crate::{
        core::{
            disputes::retrieve_dispute,
            mandate::get_mandate,
            payments::{payments_core, CallConnectorAction, PaymentStatus},
            refunds::refund_retrieve_core_with_refund_id,
        },
        services::{ApplicationResponse, AuthFlow},
        types::{api::PSync, transformers::ForeignFrom},
    };

    match tracking_data.event_class {
        diesel_models::enums::EventClass::Payments => {
            let payment_id = tracking_data.primary_object_id.clone();
            let payment_id = id_type::PaymentId::try_from(std::borrow::Cow::Owned(payment_id))
                .map_err(|payment_id_parsing_error| {
                    logger::error!(
                        ?payment_id_parsing_error,
                        "Failed to parse payment ID from tracking data"
                    );
                    errors::ProcessTrackerError::DeserializationFailed
                })?;
            let request = PaymentsRetrieveRequest {
                resource_id: PaymentIdType::PaymentIntentId(payment_id),
                merchant_id: Some(tracking_data.merchant_id.clone()),
                force_sync: false,
                ..Default::default()
            };

            let payments_response = match Box::pin(payments_core::<
                PSync,
                PaymentsResponse,
                _,
                _,
                _,
                payments::PaymentData<PSync>,
            >(
                state,
                req_state,
                platform.clone(),
                None,
                PaymentStatus,
                request,
                AuthFlow::Client,
                CallConnectorAction::Avoid,
                None,
                None,
                hyperswitch_domain_models::payments::HeaderPayload::default(),
                None,
            ))
            .await?
            {
                ApplicationResponse::Json(payments_response)
                | ApplicationResponse::JsonWithHeaders((payments_response, _)) => {
                    Ok(payments_response)
                }
                ApplicationResponse::StatusOk
                | ApplicationResponse::TextPlain(_)
                | ApplicationResponse::JsonForRedirection(_)
                | ApplicationResponse::Form(_)
                | ApplicationResponse::GenericLinkForm(_)
                | ApplicationResponse::PaymentLinkForm(_)
                | ApplicationResponse::FileData(_)
                | ApplicationResponse::IncomingWebhookEvent { .. } => {
                    Err(errors::ProcessTrackerError::ResourceFetchingFailed {
                        resource_name: tracking_data.primary_object_id.clone(),
                    })
                }
            }?;
            let event_type: Option<EventType> = payments_response.status.into();
            logger::debug!(current_resource_status=%payments_response.status);

            Ok((
                OutgoingWebhookContent::PaymentDetails(Box::new(payments_response)),
                event_type,
            ))
        }

        diesel_models::enums::EventClass::Refunds => {
            let refund_id = tracking_data.primary_object_id.clone();
            let request = RefundsRetrieveRequest {
                refund_id,
                force_sync: Some(false),
                merchant_connector_details: None,
                all_keys_required: None,
            };

            let (refund, _) = Box::pin(refund_retrieve_core_with_refund_id(
                state,
                platform.clone(),
                None,
                request,
            ))
            .await?;
            let event_type: Option<EventType> = refund.refund_status.into();
            logger::debug!(current_resource_status=%refund.refund_status);
            let refund_response = RefundResponse::foreign_from(refund);

            Ok((
                OutgoingWebhookContent::RefundDetails(Box::new(refund_response)),
                event_type,
            ))
        }

        diesel_models::enums::EventClass::Disputes => {
            let dispute_id = tracking_data.primary_object_id.clone();
            let request = DisputeRetrieveRequest {
                dispute_id,
                force_sync: None,
            };

            let dispute_response =
                match Box::pin(retrieve_dispute(state, platform.clone(), None, request)).await? {
                    ApplicationResponse::Json(dispute_response)
                    | ApplicationResponse::JsonWithHeaders((dispute_response, _)) => {
                        Ok(dispute_response)
                    }
                    ApplicationResponse::StatusOk
                    | ApplicationResponse::TextPlain(_)
                    | ApplicationResponse::JsonForRedirection(_)
                    | ApplicationResponse::Form(_)
                    | ApplicationResponse::GenericLinkForm(_)
                    | ApplicationResponse::PaymentLinkForm(_)
                    | ApplicationResponse::FileData(_)
                    | ApplicationResponse::IncomingWebhookEvent { .. } => {
                        Err(errors::ProcessTrackerError::ResourceFetchingFailed {
                            resource_name: tracking_data.primary_object_id.clone(),
                        })
                    }
                }
                .map(Box::new)?;
            let event_type = Some(EventType::from(dispute_response.dispute_status));
            logger::debug!(current_resource_status=%dispute_response.dispute_status);

            Ok((
                OutgoingWebhookContent::DisputeDetails(dispute_response),
                event_type,
            ))
        }

        diesel_models::enums::EventClass::Mandates => {
            let mandate_id = tracking_data.primary_object_id.clone();
            let request = MandateId { mandate_id };

            let mandate_response = match get_mandate(state, platform.clone(), request).await? {
                ApplicationResponse::Json(mandate_response)
                | ApplicationResponse::JsonWithHeaders((mandate_response, _)) => {
                    Ok(mandate_response)
                }
                ApplicationResponse::StatusOk
                | ApplicationResponse::TextPlain(_)
                | ApplicationResponse::JsonForRedirection(_)
                | ApplicationResponse::Form(_)
                | ApplicationResponse::GenericLinkForm(_)
                | ApplicationResponse::PaymentLinkForm(_)
                | ApplicationResponse::FileData(_)
                | ApplicationResponse::IncomingWebhookEvent { .. } => {
                    Err(errors::ProcessTrackerError::ResourceFetchingFailed {
                        resource_name: tracking_data.primary_object_id.clone(),
                    })
                }
            }
            .map(Box::new)?;
            let event_type: Option<EventType> = mandate_response.status.into();
            logger::debug!(current_resource_status=%mandate_response.status);

            Ok((
                OutgoingWebhookContent::MandateDetails(mandate_response),
                event_type,
            ))
        }
        #[cfg(feature = "payouts")]
        diesel_models::enums::EventClass::Payouts => {
            let payout_id = tracking_data.primary_object_id.clone();
            let request = payout_models::PayoutRequest::PayoutActionRequest(
                payout_models::PayoutActionRequest {
                    payout_id: id_type::PayoutId::try_from(std::borrow::Cow::Owned(payout_id))?,
                },
            );

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

            let payout_create_response =
                payouts::response_handler(&state, platform, &payout_data).await?;

            let event_type: Option<EventType> = payout_data.payout_attempt.status.into();
            logger::debug!(current_resource_status=%payout_data.payout_attempt.status);

            Ok((
                OutgoingWebhookContent::PayoutDetails(Box::new(payout_create_response)),
                event_type,
            ))
        }
        diesel_models::enums::EventClass::Subscriptions => {
            let invoice_id = tracking_data.primary_object_id.clone();
            let profile_id = &tracking_data.business_profile_id;

            let response = Box::pin(
                invoice_sync::InvoiceSyncHandler::form_response_for_retry_outgoing_webhook_task(
                    state.clone().into(),
                    platform.get_processor().get_key_store(),
                    invoice_id,
                    profile_id,
                    platform.get_processor().get_account(),
                ),
            )
            .await
            .inspect_err(|e| {
                logger::error!(
                    "Failed to generate response for subscription outgoing webhook: {e:?}"
                );
            })?;

            Ok((
                OutgoingWebhookContent::SubscriptionDetails(Box::new(response)),
                Some(EventType::InvoicePaid),
            ))
        }
    }
}
