//! Webhook retry with exponential backoff and stale refund escalation.

use crate::AppState;
use crate::db;
use axum::extract::FromRef;

/// Maximum webhook retry attempts before marking as dead letter.
const WEBHOOK_MAX_RETRIES: i32 = 5;

/// Determine whether a webhook retry attempt should be treated as a dead letter.
pub(super) fn is_webhook_dead(attempt: i32) -> bool {
    attempt >= WEBHOOK_MAX_RETRIES
}

/// Retry failed webhook events with exponential backoff.
#[tracing::instrument(skip_all, name = "scheduler::retry_failed_webhooks")]
pub(super) async fn retry_failed_webhooks(state: &AppState) {
    let events = match db::webhook_events::get_retryable_events(&state.db).await {
        Ok(e) if e.is_empty() => return,
        Ok(e) => e,
        Err(e) => {
            tracing::error!(error = ?e, "failed to fetch retryable webhook events");
            return;
        }
    };

    if state.stripe.is_none() {
        return;
    }

    for event in events {
        let attempt = event.attempts + 1;
        tracing::info!(
            event_id = %event.id, source = %event.source, event_type = %event.event_type,
            attempt = attempt, "retrying webhook event"
        );

        // The Stripe event id (from the stored payload) is the key the live
        // webhook handler locks and dedups on. Parse it up front so this retry
        // takes the SAME per-event advisory lock and consults the SAME processed
        // marker, otherwise a queue retry and a live Stripe redelivery of one
        // event can run concurrently, degrading exactly-once dispatch to
        // at-least-once + per-handler idempotency (Run 21 payments).
        let stripe_event_id: Option<String> = if event.source == "stripe" {
            crate::payments::UntypedEvent::from_payload(&event.payload)
                .ok()
                .map(|p| p.id)
        } else if event.source == "stripe_v2" {
            serde_json::from_str::<crate::payments::ThinEvent>(&event.payload)
                .ok()
                .map(|t| t.id)
        } else {
            None
        };

        // Hold the per-event lock across dedup-read -> process -> mark, exactly
        // like the live handler. We *try* the lock: if a live redelivery (or
        // another tick) already holds it, leave this row queued and move on rather
        // than blocking the retry loop on a pooled connection. Named binding (not
        // bare `_`) so the guard lives, and holds the lock, for the whole
        // iteration.
        let _event_lock = match &stripe_event_id {
            Some(eid) => match db::webhook_events::try_lock_event(&state.db, eid).await {
                Ok(Some(tx)) => Some(tx),
                Ok(None) => {
                    tracing::info!(event_id = %event.id, "webhook event locked by another worker; leaving queued for next tick");
                    continue;
                }
                Err(e) => {
                    tracing::error!(event_id = %event.id, error = ?e, "failed to lock webhook event for retry; will retry next tick");
                    continue;
                }
            },
            None => None,
        };

        // If a live redelivery already processed this event while it sat in the
        // queue, don't re-run the handler, just resolve the retry-queue row.
        if let Some(eid) = &stripe_event_id
            && matches!(
                db::webhook_events::is_event_processed(&state.db, eid).await,
                Ok(true)
            )
        {
            tracing::info!(event_id = %event.id, stripe_event_id = %eid, "webhook already processed (live redelivery won the race); resolving retry row");
            if let Err(e) = db::webhook_events::mark_processed(&state.db, event.id).await {
                tracing::error!(error = ?e, "failed to mark webhook event as processed");
            }
            continue;
        }

        // Retry re-runs the full event handler. All handlers must be idempotent
        // (use ON CONFLICT / WHERE status='pending' guards) since steps completed
        // before the original failure are not rolled back.
        let result = if event.source == "stripe" {
            match crate::payments::UntypedEvent::from_payload(&event.payload) {
                Ok(parsed) => {
                    let crate::payments::UntypedEvent {
                        id,
                        type_,
                        data_object,
                    } = parsed;
                    crate::routes::stripe::process_webhook_event(
                        &state.db,
                        &state.bg,
                        &state.email,
                        state.wam.as_ref(),
                        &crate::Billing::from_ref(state),
                        &state.config,
                        &type_,
                        &id,
                        data_object,
                    )
                    .await
                }
                Err(e) => Err(e),
            }
        } else if event.source == "stripe_v2" {
            // v2 thin events: re-parse the stored payload (signature was verified
            // at receive time) and re-route. The handler re-fetches the object
            // from Stripe and re-applies it idempotently.
            match serde_json::from_str::<crate::payments::ThinEvent>(&event.payload) {
                Ok(thin) => match state.stripe.as_ref() {
                    Some(stripe) => {
                        crate::routes::stripe::process_v2_thin_event(
                            &state.db,
                            state.wam.as_ref(),
                            stripe.as_ref(),
                            &state.config.signing_secret,
                            &thin,
                        )
                        .await
                    }
                    None => Err(crate::error::AppError::BadRequest(
                        "Stripe not configured".to_string(),
                    )),
                },
                Err(e) => Err(crate::error::AppError::BadRequest(format!(
                    "failed to parse stored v2 event: {e}"
                ))),
            }
        } else {
            Err(crate::error::AppError::BadRequest(format!(
                "Unknown webhook source: {}",
                event.source
            )))
        };

        match result {
            Ok(()) => {
                tracing::info!(event_id = %event.id, "webhook retry succeeded");
                // Write the shared dedup marker so a later live redelivery of the
                // same event short-circuits (mirrors the live handler's mark).
                if let Some(eid) = &stripe_event_id
                    && let Err(e) = db::webhook_events::mark_event_processed(&state.db, eid).await
                {
                    tracing::error!(event_id = %event.id, error = ?e, "webhook retry succeeded but recording processed-marker failed");
                }
                if let Err(e) = db::webhook_events::mark_processed(&state.db, event.id).await {
                    tracing::error!(error = ?e, "failed to mark webhook event as processed");
                }
            }
            Err(e) => {
                let is_dead = is_webhook_dead(attempt);
                tracing::warn!(
                    event_id = %event.id, attempt = attempt, error = ?e,
                    dead = is_dead,
                    "webhook retry failed"
                );
                if let Err(e) = db::webhook_events::schedule_retry(
                    &state.db,
                    event.id,
                    attempt,
                    &format!("{e:?}"),
                )
                .await
                {
                    tracing::error!(error = ?e, "failed to schedule webhook retry");
                }

                if is_dead && let Some(ref wam) = state.wam {
                    let title = format!("Dead webhook: {} ({})", event.event_type, event.id);
                    let body = format!(
                        "Webhook event exhausted all {} retry attempts.\n\
                         Source: {}\nType: {}\nLast error: {:?}",
                        attempt, event.source, event.event_type, e,
                    );
                    wam.create_ticket(
                        &title,
                        Some(&body),
                        "high",
                        "webhook-dead-letter",
                        Some(&event.id.to_string()),
                    )
                    .await;
                }
            }
        }
    }
}

/// Alert the admin about pending refunds that have gone unmatched for >24 hours.
#[tracing::instrument(skip_all, name = "scheduler::escalate_stale_refunds")]
pub(super) async fn escalate_stale_refunds(state: &AppState) {
    let stale = match db::pending_refunds::get_stale_refunds(&state.db, chrono::Duration::hours(24))
        .await
    {
        Ok(s) if s.is_empty() => return,
        Ok(s) => s,
        Err(e) => {
            tracing::error!(error = ?e, "failed to query stale pending refunds");
            return;
        }
    };

    let alert_email = std::env::var("ALERT_EMAIL").ok();

    for refund in &stale {
        // Mark escalated FIRST to prevent duplicate alerts on retry
        if let Err(e) = db::pending_refunds::mark_escalated(&state.db, refund.id).await {
            tracing::error!(error = ?e, "failed to mark pending refund as escalated, skipping alerts");
            continue;
        }

        tracing::error!(
            payment_intent_id = %refund.payment_intent_id,
            amount = refund.amount.as_i64(),
            amount_refunded = refund.amount_refunded.as_i64(),
            created_at = %refund.created_at,
            "STALE PENDING REFUND: not completed within >24h (unmatched, or claimed but \
             the refund never finished), needs manual investigation"
        );

        if let Some(ref to) = alert_email {
            let subject = format!(
                "Unmatched refund: {} ({}c refunded)",
                refund.payment_intent_id, refund.amount_refunded
            );
            let body = format!(
                "A charge.refunded webhook for payment intent {} has been pending for >24 hours \
                 without its refund completing.\n\n\
                 Amount: {}c\nAmount refunded: {}c\nReceived: {}\n\n\
                 Either the checkout.session.completed webhook was lost (never matched), or the \
                 refund was claimed but the process died before it finished. \
                 Check the Stripe dashboard for whether the refund was actually issued and \
                 reconcile manually.",
                refund.payment_intent_id, refund.amount, refund.amount_refunded, refund.created_at,
            );
            if let Err(e) = state.email.send_alert(to, &subject, &body).await {
                tracing::error!(error = ?e, "failed to send stale refund alert email");
            }
        }

        if let Some(ref wam) = state.wam {
            let title = format!(
                "Unmatched refund: {} ({}c)",
                refund.payment_intent_id, refund.amount_refunded
            );
            let body = format!(
                "charge.refunded webhook pending >24h with no matching completed transaction.\n\
                 Amount: {}c\nRefunded: {}c\nReceived: {}\n\
                 Check Stripe dashboard and reconcile manually.",
                refund.amount, refund.amount_refunded, refund.created_at,
            );
            wam.create_ticket(
                &title,
                Some(&body),
                "critical",
                "refund-escalation",
                Some(&refund.payment_intent_id),
            )
            .await;
        }
    }
}

/// Per-tick cap on platform-credit settlement, so a backlog drains across ticks
/// instead of one unbounded loop stalling the scheduler tick.
const SETTLE_CREDITS_PER_TICK: usize = 50;

/// Settle owed platform-funded credits (Fan+ reimbursements) via platform ->
/// connected transfers, making the creator whole for a credit MNW funded.
///
/// Each credit is claimed (so one worker settles it), the seller's Stripe account
/// resolved, the owed amount transferred with a deterministic idempotency key, then
/// marked settled. A seller not yet payable, or a transient transfer failure,
/// releases the claim for a later retry; a process death between claim and settle
/// leaves the row claimed-but-unsettled and is escalated by
/// [`escalate_stale_platform_credits`] rather than blindly retried.
#[tracing::instrument(skip_all, name = "scheduler::settle_platform_credits")]
pub(super) async fn settle_platform_credits(state: &AppState) {
    let Some(stripe) = state.stripe.as_ref() else {
        return;
    };

    for _ in 0..SETTLE_CREDITS_PER_TICK {
        let credit = match db::platform_credits::claim_unsettled_credit(&state.db).await {
            Ok(Some(c)) => c,
            Ok(None) => break,
            Err(e) => {
                tracing::error!(error = ?e, "failed to claim platform credit for settlement");
                break;
            }
        };

        let account_id = match db::users::get_user_by_id(&state.db, credit.seller_id).await {
            Ok(Some(u)) => u.stripe_account_id,
            Ok(None) => None,
            Err(e) => {
                tracing::error!(error = ?e, transaction_id = %credit.transaction_id, "failed to load seller for platform credit; releasing");
                db::platform_credits::unclaim_credit(&state.db, credit.transaction_id)
                    .await
                    .ok();
                continue;
            }
        };

        let Some(account) = account_id.as_deref() else {
            // Seller not (yet) payable, release for a later retry. The stale sweep
            // escalates any that never become payable.
            db::platform_credits::unclaim_credit(&state.db, credit.transaction_id)
                .await
                .ok();
            continue;
        };

        match stripe
            .create_platform_credit_transfer(
                account,
                credit.amount_cents.as_i64(),
                credit.transaction_id,
                credit.currency,
            )
            .await
        {
            Ok(transfer_id) => {
                if let Err(e) = db::platform_credits::mark_settled(
                    &state.db,
                    credit.transaction_id,
                    &transfer_id,
                )
                .await
                {
                    // The transfer succeeded but the settle write failed. The
                    // deterministic idempotency key makes the next-tick retry safe
                    // (Stripe returns the same transfer, no double-pay).
                    tracing::error!(error = ?e, transaction_id = %credit.transaction_id, "platform credit transferred but marking settled failed");
                }
            }
            Err(e) => {
                tracing::error!(error = ?e, transaction_id = %credit.transaction_id, "platform credit transfer failed; releasing for retry");
                db::platform_credits::unclaim_credit(&state.db, credit.transaction_id)
                    .await
                    .ok();
            }
        }
    }
}

/// Reverse the MNW -> creator transfer for settled platform-funded credits whose
/// sale was later refunded, clawing the reimbursement back so the platform isn't
/// left funding a returned item (Run 21 money-loss finding).
///
/// Only *settled* credits need this: an unsettled credit on a refunded
/// transaction is never paid out (the settle sweep gates on `status = 'completed'`,
/// and refunds flip the row to `refunded`). Each reversal uses a deterministic
/// idempotency key, so a redelivery or retry can't claw back twice. Runs after
/// settlement in the tick and is single-instance (scheduler advisory lock).
#[tracing::instrument(skip_all, name = "scheduler::reverse_refunded_platform_credits")]
pub(super) async fn reverse_refunded_platform_credits(state: &AppState) {
    let Some(stripe) = state.stripe.as_ref() else {
        return;
    };

    let reversible = match db::platform_credits::get_reversible_credits(
        &state.db,
        SETTLE_CREDITS_PER_TICK as i64,
    )
    .await
    {
        Ok(r) => r,
        Err(e) => {
            tracing::error!(error = ?e, "failed to query reversible platform credits");
            return;
        }
    };

    for credit in reversible {
        match stripe
            .create_platform_credit_reversal(
                &credit.transfer_id,
                credit.amount_cents.as_i64(),
                credit.transaction_id,
            )
            .await
        {
            Ok(()) => {
                if let Err(e) =
                    db::platform_credits::mark_reversed(&state.db, credit.transaction_id).await
                {
                    // Reversal succeeded but the mark write failed, the
                    // deterministic idempotency key makes the next-tick retry safe.
                    tracing::error!(error = ?e, transaction_id = %credit.transaction_id, "platform credit reversed but marking reversed failed");
                } else {
                    tracing::info!(transaction_id = %credit.transaction_id, amount = credit.amount_cents.as_i64(), "reversed platform credit on refund");
                }
            }
            Err(e) => {
                // Leave it for the next tick; nothing is marked, so it's retried.
                tracing::error!(error = ?e, transaction_id = %credit.transaction_id, "platform credit reversal failed; will retry next tick");
            }
        }
    }
}

/// Alert the admin about platform-funded credits claimed for settlement but never
/// completed (the crash window between claim and transfer). The transfer's
/// deterministic idempotency key means a human can safely re-trigger it.
#[tracing::instrument(skip_all, name = "scheduler::escalate_stale_platform_credits")]
pub(super) async fn escalate_stale_platform_credits(state: &AppState) {
    let stale =
        match db::platform_credits::get_stale_credits(&state.db, chrono::Duration::hours(24)).await
        {
            Ok(s) if s.is_empty() => return,
            Ok(s) => s,
            Err(e) => {
                tracing::error!(error = ?e, "failed to query stale platform credits");
                return;
            }
        };

    let alert_email = std::env::var("ALERT_EMAIL").ok();

    for credit in &stale {
        if let Err(e) = db::platform_credits::mark_escalated(&state.db, credit.transaction_id).await
        {
            tracing::error!(error = ?e, "failed to mark platform credit escalated, skipping alerts");
            continue;
        }

        tracing::error!(
            transaction_id = %credit.transaction_id,
            seller_id = %credit.seller_id,
            amount = credit.amount_cents.as_i64(),
            "STALE PLATFORM CREDIT: Fan+ reimbursement claimed but not settled within >24h; \
             verify the transfer in Stripe (idempotency key platform-credit-<transaction_id>) and reconcile"
        );

        if let Some(ref to) = alert_email {
            let subject = format!(
                "Unsettled Fan+ credit: transaction {}",
                credit.transaction_id
            );
            let body = format!(
                "A platform-funded (Fan+) credit reimbursement was claimed for settlement but the \
                 transfer never completed within >24h.\n\n\
                 Transaction: {}\nSeller: {}\nAmount owed: {}c\n\n\
                 The process likely died between claim and transfer. The transfer uses the \
                 deterministic idempotency key `platform-credit-{}`, so it is safe to re-trigger \
                 or verify in the Stripe dashboard and reconcile manually.",
                credit.transaction_id,
                credit.seller_id,
                credit.amount_cents.as_i64(),
                credit.transaction_id,
            );
            if let Err(e) = state.email.send_alert(to, &subject, &body).await {
                tracing::error!(error = ?e, "failed to send stale platform credit alert email");
            }
        }

        if let Some(ref wam) = state.wam {
            let title = format!("Unsettled Fan+ credit: {}c", credit.amount_cents.as_i64());
            let body = format!(
                "Platform-funded credit claimed >24h ago without settling.\n\
                 Transaction: {}\nSeller: {}\nAmount: {}c\n\
                 Idempotency key: platform-credit-{}. Verify in Stripe and reconcile.",
                credit.transaction_id,
                credit.seller_id,
                credit.amount_cents.as_i64(),
                credit.transaction_id,
            );
            wam.create_ticket(
                &title,
                Some(&body),
                "critical",
                "platform-credit-escalation",
                Some(&credit.transaction_id.to_string()),
            )
            .await;
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn webhook_not_dead_under_threshold() {
        assert!(!is_webhook_dead(1));
        assert!(!is_webhook_dead(4));
    }

    #[test]
    fn webhook_dead_at_threshold() {
        assert!(is_webhook_dead(5));
    }

    #[test]
    fn webhook_dead_above_threshold() {
        assert!(is_webhook_dead(6));
        assert!(is_webhook_dead(100));
    }
}