//! Stripe webhook event processing.

mod billing;
mod checkout;
pub(crate) mod checkout_helpers;
mod subscriptions;

use axum::{
    body::Bytes,
    extract::State,
    http::{StatusCode, header::HeaderMap},
    response::IntoResponse,
};
use sqlx::PgPool;

use crate::{
    Billing, Integrations,
    config::Config,
    db,
    email::EmailClient,
    error::{AppError, Result, ResultExt},
    payments::{
        self, AccountUpdate, AccountView, ChargeRefundData, ChargeView, CheckoutSessionView,
        InvoiceView, RefundView, SubscriptionView, UntypedEvent,
    },
    wam_client::WamClient,
};

/// POST /stripe/webhook - Handle Stripe webhook events
#[tracing::instrument(skip_all, name = "stripe::webhook")]
#[allow(clippy::too_many_arguments)]
pub(in crate::routes::stripe) async fn webhook(
    State(db): State<PgPool>,
    State(bg): State<crate::background::BackgroundTx>,
    State(email): State<EmailClient>,
    State(integrations): State<Integrations>,
    State(payments): State<Billing>,
    State(config): State<Config>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<impl IntoResponse> {
    let stripe = payments
        .stripe
        .as_ref()
        .ok_or_else(|| AppError::BadRequest("Stripe is not configured".to_string()))?;

    // Get the signature header
    let signature = headers
        .get("stripe-signature")
        .and_then(|v| v.to_str().ok())
        .ok_or_else(|| AppError::BadRequest("Missing Stripe signature".to_string()))?;

    // Parse and verify the webhook
    let payload = std::str::from_utf8(&body)
        .map_err(|_| AppError::BadRequest("Invalid payload encoding".to_string()))?;

    let event = stripe.verify_webhook(payload, signature)?;
    tracing::info!(event_type = %event.type_, event_id = %event.id, "received webhook event");

    // Serialize concurrent redeliveries of this event id. Held across the whole
    // dedup-read -> process -> mark sequence below, the lock makes the dedup read
    // race-free. We *try* the lock rather than block on it: if a second delivery
    // of the same event arrives while this one is mid-flight, it gets `None` and
    // returns 503 immediately instead of parking a pooled connection for the
    // duration of the in-flight handler (Run 23 Conc/Perf). Stripe redelivers
    // after this one commits its processed-event mark, and the redelivery's dedup
    // read then short-circuits. Dropping `_event_lock` on any return path rolls
    // the (write-free) lock transaction back and releases it, it cannot leak.
    // See `db::webhook_events::try_lock_event`.
    let _event_lock = match db::webhook_events::try_lock_event(&db, &event.id).await {
        Ok(Some(tx)) => tx,
        Ok(None) => {
            tracing::info!(event_id = %event.id, "concurrent delivery of this webhook event is in flight; returning 503 for redelivery");
            return Ok(StatusCode::SERVICE_UNAVAILABLE);
        }
        Err(e) => {
            tracing::error!(event_id = %event.id, error = ?e, "failed to acquire webhook event lock, returning 503 for retry");
            return Ok(StatusCode::SERVICE_UNAVAILABLE);
        }
    };

    // Deduplicate: skip if we already processed this event ID. This is a READ;
    // the "processed" row is written only after the handler succeeds (below), so
    // a crash mid-processing leaves no marker and Stripe's redelivery reprocesses.
    // The `_event_lock` above closes the check-then-act race, so this read now
    // guarantees exactly-once dispatch regardless of handler idempotency; per-
    // handler atomic guards (status-guarded UPDATEs, ON CONFLICT writes, the
    // `FanPlusCreditClaim` witness) remain as defense in depth.
    match db::webhook_events::is_event_processed(&db, &event.id).await {
        Ok(true) => {
            tracing::info!(event_id = %event.id, "duplicate webhook event, skipping");
            return Ok(StatusCode::OK);
        }
        Err(e) => {
            tracing::error!(event_id = %event.id, error = ?e, "webhook dedup check failed, returning 503 for retry");
            return Ok(StatusCode::SERVICE_UNAVAILABLE);
        }
        Ok(false) => {} // First time seeing this event
    }

    // For retry-queue persistence we need id+type after `event` is consumed.
    // Move both out without cloning the underlying allocations.
    let UntypedEvent {
        id: event_id,
        type_: event_type_str,
        data_object,
    } = event;
    let result = process_webhook_event(
        &db,
        &bg,
        &email,
        integrations.wam.as_ref(),
        &payments,
        &config,
        &event_type_str,
        &event_id,
        data_object,
    )
    .await;

    match result {
        Ok(()) => {
            // Work is durably committed, now record the event as processed so a
            // redelivery short-circuits. If this write fails, return 503: Stripe
            // redelivers, the idempotent handler re-runs, and the mark is retried.
            // No event is lost; at worst it is processed twice (safe).
            if let Err(e) = db::webhook_events::mark_event_processed(&db, &event_id).await {
                tracing::error!(event_id = %event_id, error = ?e, "failed to record processed webhook event; returning 503 for redelivery");
                return Ok(StatusCode::SERVICE_UNAVAILABLE);
            }
        }
        Err(ref e) => {
            tracing::error!(
                event_id = %event_id, event_type = %event_type_str,
                error = ?e, "webhook handler failed, queueing for retry"
            );
            // Not marked processed, so redelivery (or the retry worker) reruns it.
            if let Err(queue_err) = db::webhook_events::insert_failed_event(
                &db,
                "stripe",
                &event_type_str,
                payload,
                Some(signature),
                &format!("{e:?}"),
            )
            .await
            {
                tracing::error!(error = ?queue_err, "failed to queue webhook event for retry; returning 503 to trigger Stripe redelivery");
                return Ok(StatusCode::SERVICE_UNAVAILABLE);
            }
        }
    }

    Ok(StatusCode::OK)
}

/// Process a verified Stripe webhook event. Extracted to allow the caller
/// to catch errors and persist to the retry queue. Also called by the
/// scheduler's webhook retry worker.
/// Dispatch a verified Stripe webhook event. Consumes `data_object` exactly
/// once into a typed rc.5 struct based on `event_type`. Shared by the live
/// webhook handler and the scheduler retry worker.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn process_webhook_event(
    db: &PgPool,
    bg: &crate::background::BackgroundTx,
    email: &EmailClient,
    wam: Option<&WamClient>,
    payments: &Billing,
    config: &Config,
    event_type: &str,
    event_id: &str,
    data_object: serde_json::Value,
) -> Result<()> {
    match event_type {
        // Both events route through the same dispatcher. `completed` fires
        // immediately; for asynchronous payment methods (ACH/SEPA/Bacs) it
        // arrives with payment_status="unpaid" and the money-taking handlers
        // defer until `async_payment_succeeded` re-delivers the settled session.
        "checkout.session.completed" | "checkout.session.async_payment_succeeded" => {
            let session: CheckoutSessionView =
                serde_json::from_value(data_object).map_err(|e| {
                    AppError::BadRequest(format!("Failed to parse CheckoutSession: {e}"))
                })?;
            dispatch_checkout_session(db, bg, email, wam, payments, config, &session, event_id)
                .await?;
        }
        // The buyer's async payment (ACH/SEPA/Bacs) never cleared. No funds were
        // captured, so there is nothing to deliver; the pending transaction (and
        // any reserved promo hold) is released by the stale-pending cleanup
        // sweeper. Logged for visibility rather than silently dropped.
        "checkout.session.async_payment_failed" => {
            let session: CheckoutSessionView =
                serde_json::from_value(data_object).map_err(|e| {
                    AppError::BadRequest(format!("Failed to parse CheckoutSession: {e}"))
                })?;
            tracing::warn!(
                session_id = %session.id,
                "checkout async payment failed; no funds captured, pending rows will be released by cleanup"
            );
        }
        "account.updated" => {
            let account: AccountView = serde_json::from_value(data_object)
                .map_err(|e| AppError::BadRequest(format!("Failed to parse Account: {e}")))?;
            handle_account_updated(db, wam, &AccountUpdate::from(account)).await?;
        }
        "charge.refunded" => {
            let charge: ChargeView = serde_json::from_value(data_object)
                .map_err(|e| AppError::BadRequest(format!("Failed to parse Charge: {e}")))?;
            if let Some(refund_data) = ChargeRefundData::from_view(charge) {
                // Direct webhook: queue as pending if the matching payment hasn't
                // landed yet. Out-of-band (dashboard) FULL refunds are handled
                // here; per-line refunds land via refund.created below.
                billing::handle_charge_refunded(db, &refund_data, true).await?;
            }
        }
        "refund.created" | "refund.updated" => {
            // Line-scoped self-service refunds tag the Stripe refund with
            // mnw_transaction_id; this marks/revokes exactly that transaction so
            // a cart line refund leaves the order's other lines untouched
            // (Run #2 Payments SERIOUS). Untagged refunds are no-ops here.
            let refund: RefundView = serde_json::from_value(data_object)
                .map_err(|e| AppError::BadRequest(format!("Failed to parse Refund: {e}")))?;
            billing::handle_refund_created(db, &refund).await?;
        }
        "customer.subscription.updated" => {
            let sub: SubscriptionView = serde_json::from_value(data_object)
                .map_err(|e| AppError::BadRequest(format!("Failed to parse Subscription: {e}")))?;
            subscriptions::handle_subscription_updated(db, &sub, event_id).await?;
        }
        "customer.subscription.deleted" => {
            let sub: SubscriptionView = serde_json::from_value(data_object)
                .map_err(|e| AppError::BadRequest(format!("Failed to parse Subscription: {e}")))?;
            subscriptions::handle_subscription_deleted(db, bg, email, &sub, event_id).await?;
        }
        "invoice.payment_succeeded" => {
            let invoice: InvoiceView = serde_json::from_value(data_object)
                .map_err(|e| AppError::BadRequest(format!("Failed to parse Invoice: {e}")))?;
            billing::handle_invoice_payment_succeeded(db, bg, email, wam, &invoice, event_id)
                .await?;
        }
        "invoice.payment_failed" => {
            let invoice: InvoiceView = serde_json::from_value(data_object)
                .map_err(|e| AppError::BadRequest(format!("Failed to parse Invoice: {e}")))?;
            billing::handle_invoice_payment_failed(db, wam, &invoice, event_id).await?;
        }
        other => {
            tracing::debug!(event_type = %other, "unhandled webhook event type");
        }
    }

    Ok(())
}

/// Route a checkout session to its handler by metadata shape.
///
/// Subscription-mode sessions (Fan+, creator tier, SyncKit app sub, project
/// subscription) capture no funds at checkout, the subscription lifecycle bills
/// separately, so they run unconditionally. One-time payment-mode sessions
/// (tip, guest, cart, single purchase) capture funds now and are therefore
/// gated on `payment_settled()`: an async method that reports `payment_status
/// = "unpaid"` on `checkout.session.completed` is deferred until Stripe
/// re-delivers the settled session via `checkout.session.async_payment_succeeded`.
/// Without this gate, enabling any async payment method on a connected account
/// would mint license keys and grant downloads before funds settle.
#[allow(clippy::too_many_arguments)]
async fn dispatch_checkout_session(
    db: &PgPool,
    bg: &crate::background::BackgroundTx,
    email: &EmailClient,
    wam: Option<&WamClient>,
    payments: &Billing,
    config: &Config,
    session: &CheckoutSessionView,
    event_id: &str,
) -> Result<()> {
    let meta = session.metadata.as_ref();

    // Subscription-mode: no funds captured at checkout, no settlement gate.
    if payments::is_fan_plus_checkout(meta) {
        return checkout::handle_fan_plus_checkout_completed(db, bg, email, session, event_id)
            .await;
    }
    if payments::is_creator_tier_checkout(meta) {
        return checkout::handle_creator_tier_checkout_completed(
            db, bg, wam, payments, session, event_id,
        )
        .await;
    }
    if payments::is_synckit_app_sub_checkout(meta) {
        return checkout::handle_synckit_app_sub_checkout_completed(db, session, event_id).await;
    }
    if payments::is_subscription_checkout(meta) {
        return checkout::handle_subscription_checkout_completed(db, bg, email, session, event_id)
            .await;
    }

    // One-time payment-mode: funds captured now. Deliver only once settled.
    if !session.payment_settled() {
        tracing::info!(
            session_id = %session.id,
            payment_status = ?session.payment_status,
            "one-time checkout not yet settled (async payment); deferring finalize until async_payment_succeeded"
        );
        return Ok(());
    }

    if payments::is_tip_checkout(meta) {
        checkout::handle_tip_checkout_completed(db, bg, email, wam, config, session, event_id).await
    } else if payments::is_guest_checkout(meta) {
        checkout::handle_guest_checkout_completed(db, bg, email, wam, config, session, event_id)
            .await
    } else if payments::is_cart_checkout(meta) {
        checkout::handle_cart_checkout_completed(db, bg, email, wam, config, session, event_id)
            .await
    } else {
        checkout::handle_purchase_checkout_completed(db, bg, email, wam, config, session, event_id)
            .await
    }
}

/// Handle account.updated from the v2 thin event endpoint.
pub(in crate::routes::stripe) async fn handle_account_updated_from_v2(
    db: &PgPool,
    wam: Option<&WamClient>,
    update: &AccountUpdate,
) -> Result<()> {
    handle_account_updated(db, wam, update).await
}

/// Handle account.updated webhook
async fn handle_account_updated(
    db: &PgPool,
    wam: Option<&WamClient>,
    update: &AccountUpdate,
) -> Result<()> {
    tracing::info!(
        account_id = %update.account_id, charges_enabled = %update.charges_enabled,
        payouts_enabled = %update.payouts_enabled, details_submitted = %update.details_submitted,
        "account updated"
    );

    // Update the user's Stripe status
    db::users::update_user_stripe_status(
        db,
        &update.account_id,
        update.details_submitted,
        update.payouts_enabled,
        update.charges_enabled,
    )
    .await
    .with_context(|| format!("update Stripe status for account {}", update.account_id))?;

    // Alert if charges or payouts became disabled (creator can't receive payments)
    if (!update.charges_enabled || !update.payouts_enabled)
        && let Some(wam) = wam
    {
        let title = format!("Stripe Connect degraded: {}", update.account_id);
        let body = format!(
            "charges_enabled: {}\npayouts_enabled: {}\ndetails_submitted: {}",
            update.charges_enabled, update.payouts_enabled, update.details_submitted,
        );
        wam.create_ticket(
            &title,
            Some(&body),
            "high",
            "stripe-connect-degraded",
            Some(&update.account_id),
        )
        .await;
    }

    Ok(())
}