//! Stripe v2 thin event webhook handler.
//!
//! Stripe's v2 event system sends "thin" events that contain only a reference
//! to the affected object, not the full snapshot.  The handler verifies the
//! signature, parses the event type, fetches the full object via the API, and
//! delegates to the same business logic used by the v1 handler.

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

use crate::{
    Billing, Integrations, db,
    error::{AppError, Result},
    payments::{self, ThinEvent},
    wam_client::WamClient,
};

/// POST /stripe/webhook/v2: Handle Stripe v2 thin events
#[tracing::instrument(skip_all, name = "stripe::webhook_v2")]
pub(super) async fn webhook_v2(
    State(db): State<PgPool>,
    State(integrations): State<Integrations>,
    State(payments): State<Billing>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<impl IntoResponse> {
    let stripe = payments
        .stripe
        .as_ref()
        .ok_or_else(|| AppError::BadRequest("Stripe is not configured".to_string()))?;

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

    let payload = std::str::from_utf8(&body)
        .map_err(|_| AppError::BadRequest("Invalid payload encoding".to_string()))?;

    // Verify signature and parse JSON
    let body_json = stripe.verify_webhook_v2(payload, signature)?;

    // Parse the thin event
    let thin: ThinEvent = serde_json::from_value(body_json).map_err(|e| {
        tracing::warn!(error = ?e, "failed to parse v2 thin event");
        AppError::BadRequest("Invalid v2 event format".to_string())
    })?;

    tracing::info!(event_type = %thin.event_type, event_id = %thin.id, "received v2 thin event");

    // Serialize concurrent redeliveries of this event id (see
    // `db::webhook_events::try_lock_event`). Held across dedup-read -> process ->
    // mark, it makes the check-then-act read below race-free. We *try* the lock
    // rather than block: a same-event delivery arriving mid-flight gets `None` and
    // returns 503 immediately instead of parking a pooled connection (Run 23
    // Conc/Perf). Dropping `_event_lock` on any return releases the lock.
    let _event_lock = match db::webhook_events::try_lock_event(&db, &thin.id).await {
        Ok(Some(tx)) => tx,
        Ok(None) => {
            tracing::info!(event_id = %thin.id, "concurrent delivery of this v2 event is in flight; returning 503 for redelivery");
            return Ok(StatusCode::SERVICE_UNAVAILABLE);
        }
        Err(e) => {
            tracing::error!(event_id = %thin.id, error = ?e, "failed to acquire v2 webhook event lock, returning 503 for retry");
            return Ok(StatusCode::SERVICE_UNAVAILABLE);
        }
    };

    // Deduplicate: skip if this event was already processed. Read-only; the
    // "processed" row is written only after the handler succeeds (below), so a
    // crash mid-processing leaves no marker and Stripe redelivers (the handler
    // is idempotent, it re-fetches and re-applies account state).
    match db::webhook_events::is_event_processed(&db, &thin.id).await {
        Ok(true) => {
            tracing::debug!(event_id = %thin.id, "v2 event already processed, skipping");
            return Ok(StatusCode::OK);
        }
        Err(e) => {
            // Return 503 so Stripe retries later (matching v1 webhook behavior)
            tracing::error!(event_id = %thin.id, error = ?e, "v2 dedup check failed, returning 503 for retry");
            return Ok(StatusCode::SERVICE_UNAVAILABLE);
        }
        Ok(false) => {} // first time, proceed
    }

    if let Err(e) =
        process_v2_thin_event(&db, integrations.wam.as_ref(), stripe.as_ref(), &thin).await
    {
        // Persist to the local retry queue (backoff + dead-letter WAM via the
        // scheduler), matching v1's failure path rather than relying solely on
        // Stripe's redelivery window. Not marked processed, so a Stripe
        // redelivery also still re-runs the idempotent handler.
        //
        // Deliberate trade (fuzz 2026-07-06, Payments): returning 200 below ACKs
        // the event to Stripe, so Stripe will not redeliver it on its own 3-day
        // schedule, the in-house queue + scheduler owns retry from here (richer:
        // local backoff + dead-lettering). The operational consequence is that
        // for a persistently-failing money event, **scheduler liveness is the
        // only retry backstop**; if the scheduler is down, a failed webhook has no
        // external redelivery. The `insert_failed_event` guard below returns 503
        // (inviting Stripe redelivery) if even the queue insert fails, so the net
        // is never "silently dropped".
        tracing::warn!(event_id = %thin.id, error = ?e, "v2 event processing failed; queueing for retry");
        if let Err(queue_err) = db::webhook_events::insert_failed_event(
            &db,
            "stripe_v2",
            &thin.event_type,
            payload,
            Some(signature),
            &format!("{e:?}"),
        )
        .await
        {
            tracing::error!(event_id = %thin.id, error = ?queue_err, "failed to queue v2 event for retry; returning 503 for Stripe redelivery");
            return Ok(StatusCode::SERVICE_UNAVAILABLE);
        }
        return Ok(StatusCode::OK);
    }

    // Succeeded, record it so a redelivery short-circuits.
    if let Err(e) = db::webhook_events::mark_event_processed(&db, &thin.id).await {
        tracing::error!(event_id = %thin.id, error = ?e, "failed to record processed v2 event; returning 503 for redelivery");
        return Ok(StatusCode::SERVICE_UNAVAILABLE);
    }

    Ok(StatusCode::OK)
}

/// Route a verified v2 thin event to its handler. Shared by the live webhook and
/// the scheduler retry worker (which re-parses the stored payload, the
/// signature was already verified when the event was first received).
pub(crate) async fn process_v2_thin_event(
    db: &PgPool,
    wam: Option<&WamClient>,
    stripe: &dyn payments::PaymentProvider,
    thin: &ThinEvent,
) -> Result<()> {
    if thin.event_type.starts_with("v2.core.account") {
        handle_account_thin_event(db, wam, stripe, thin).await
    } else {
        tracing::debug!(event_type = %thin.event_type, "unhandled v2 event type");
        Ok(())
    }
}

/// Fetch the full account object and delegate to the shared account-updated handler.
async fn handle_account_thin_event(
    db: &PgPool,
    wam: Option<&WamClient>,
    stripe: &dyn payments::PaymentProvider,
    thin: &ThinEvent,
) -> Result<()> {
    let account_id = match &thin.related_object {
        Some(obj) => &obj.id,
        None => {
            tracing::warn!(event_id = %thin.id, "v2 account event missing related_object");
            return Ok(()); // nothing to fetch, acknowledge
        }
    };

    let update = stripe.fetch_account(account_id).await.map_err(|e| {
        tracing::warn!(account_id = %account_id, error = ?e, "failed to fetch account for v2 event");
        e
    })?;

    super::webhook::handle_account_updated_from_v2(db, wam, &update).await.map_err(|e| {
        tracing::warn!(account_id = %account_id, error = ?e, "failed to process account update from v2 event");
        e
    })?;

    Ok(())
}