//! Webhook handlers for checkout.session.completed events.

use crate::{
    Billing,
    config::Config,
    db,
    email::EmailClient,
    error::{AppError, Result, ResultExt},
    helpers,
    payments::{
        CheckoutMetadata, CreatorTierCheckoutMetadata, FanPlusCheckoutMetadata, MnwEventName,
        SubscriptionCheckoutMetadata, SynckitAppSubCheckoutMetadata, TipCheckoutMetadata,
    },
    wam_client::WamClient,
};
use sqlx::PgPool;

use super::checkout_helpers::{
    check_pending_refund, finalize_guest_transaction, finalize_purchase_transaction,
    record_tip_splits, send_tip_email,
};

/// A `checkout.session.completed` produced no transaction to complete. Tell a
/// benign duplicate webhook (rows already completed) apart from an ORPHANED paid
/// session (rows never created, buyer charged, got nothing) and escalate the
/// latter to WAM instead of silently logging "duplicate" (Run #2 Payments
/// SERIOUS). Shared by the purchase, cart, and guest completion handlers so the
/// three can't drift.
async fn escalate_if_orphaned_session(
    db: &PgPool,
    bg: &crate::background::BackgroundTx,
    wam: Option<&WamClient>,
    session_id: &str,
    payment_intent_id: &str,
    label: &str,
) -> Result<()> {
    let exists = db::transactions::transaction_exists_for_checkout_session(db, session_id)
        .await
        .context("check session transaction existence")?;
    if exists {
        tracing::info!(session_id = %session_id, "{label} already completed, ignoring duplicate webhook");
    } else {
        tracing::error!(
            session_id = %session_id, payment_intent_id = %payment_intent_id,
            "orphaned paid session ({label}): payment completed but no transaction exists, manual reconciliation required"
        );
        if let Some(wam) = wam.cloned() {
            let sid = session_id.to_string();
            let pi = payment_intent_id.to_string();
            let label = label.to_string();
            // Route through the shutdown-drained background pool, NOT a raw
            // tokio::spawn: this ticket is the only thing that pages a human
            // about a charged-but-undelivered buyer, so it must not be dropped
            // when the process is restarted mid-flight during a deploy (Run 9
            // pattern; audit Run 22).
            bg.spawn("stripe orphaned-session ticket", async move {
                let body = format!(
                    "Checkout session {sid} (payment_intent {pi}, {label}) completed at Stripe but has \
                     NO transaction, the pending row(s) were never created. The buyer was charged and \
                     received nothing. Reconcile manually: refund the payment or recreate the order."
                );
                wam.create_ticket("Orphaned paid session", Some(&body), "high", "stripe-orphaned-session", Some(&sid)).await;
            });
        }
    }
    Ok(())
}

/// Defense-in-depth reconciliation of a completed checkout against Stripe's
/// reported session totals. Our line items are server-built, so the credited
/// total should equal Stripe's pre-tax subtotal and the session should be USD.
/// A currency mismatch or an amount mismatch is logged loudly and escalated to
/// WAM via the shutdown-drained background pool (so it never runs inside an open
/// DB transaction, yet survives a mid-deploy restart rather than being dropped
/// like a raw tokio::spawn, audit Run 22); the server-recorded amount stays
/// authoritative either way. Shared by the purchase, cart, and guest completion
/// handlers so the three can't drift.
fn reconcile_checkout_amount(
    bg: &crate::background::BackgroundTx,
    wam: Option<&WamClient>,
    session_id: &str,
    session: &crate::payments::CheckoutCompletion,
    credited_cents: i64,
    label: &str,
) {
    // Currency guard: a non-USD session makes the integer-cents subtotal
    // comparison meaningless and should never happen (sessions are built USD).
    if let Some(currency) = session.currency.as_deref()
        && !currency.eq_ignore_ascii_case("usd")
    {
        tracing::error!(
            session_id = %session_id, currency = %currency,
            "checkout session currency is not USD ({label}); integer-cents reconciliation skipped"
        );
        if let Some(wam) = wam.cloned() {
            let session_id = session_id.to_string();
            let currency = currency.to_string();
            let label = label.to_string();
            bg.spawn("stripe non-usd-session ticket", async move {
                let body = format!(
                    "Checkout session {session_id} ({label}) settled in {currency}, not USD. The \
                     server credits its own USD amount, but investigate how a non-USD session was created."
                );
                wam.create_ticket("Non-USD checkout session", Some(&body), "high", "stripe-non-usd-session", Some(&session_id)).await;
            });
        }
        return;
    }

    if let Some(subtotal) = session.amount_subtotal
        && subtotal != credited_cents
    {
        tracing::error!(
            session_id = %session_id, credited_cents = %credited_cents, stripe_subtotal_cents = %subtotal,
            "checkout amount mismatch ({label}): credited amount differs from Stripe session subtotal"
        );
        if let Some(wam) = wam.cloned() {
            let session_id = session_id.to_string();
            let label = label.to_string();
            bg.spawn("stripe amount-mismatch ticket", async move {
                let body = format!(
                    "Credited amount {credited_cents} cents != Stripe session subtotal {subtotal} cents \
                     (session {session_id}, {label}). The server amount is authoritative; investigate a \
                     price-edit / Stripe Tax / currency edge."
                );
                wam.create_ticket("Checkout amount mismatch", Some(&body), "high", "stripe-amount-mismatch", Some(&session_id)).await;
            });
        }
    }
}

/// Handle checkout.session.completed for one-time purchases
#[tracing::instrument(skip_all, name = "stripe::handle_purchase_checkout")]
pub(super) async fn handle_purchase_checkout_completed(
    db: &PgPool,
    bg: &crate::background::BackgroundTx,
    email: &EmailClient,
    wam: Option<&WamClient>,
    config: &Config,
    session: &crate::payments::CheckoutCompletion,
    event_id: &str,
) -> Result<()> {
    let session_id = session.session_id.clone();

    tracing::info!(session_id = %session_id, "processing completed purchase checkout");

    // Extract metadata (already typed IDs from CheckoutMetadata)
    let raw_metadata = CheckoutMetadata::from_metadata(session.metadata.as_ref())?;
    let buyer_id = raw_metadata.buyer_id;
    let seller_id = raw_metadata.seller_id;
    let item_id = raw_metadata.item_id;

    let item_id_display = item_id.map_or_else(|| "project".to_string(), |id| id.to_string());

    // Get the payment intent ID
    // Display/logging copy only; the DB write below passes `session.payment_intent_id`
    // directly so a PI-less session stores NULL, not a literal "unknown" that would
    // collide with other PI-less rows in the money-keyed lookup column (Run 9).
    let payment_intent_id = session.payment_intent_id.clone().unwrap_or_default();

    // Complete the transaction (idempotent - returns None if already completed).
    // Steps 1-3 (complete_transaction, increment_sales_count, discount code increment)
    // are wrapped in a single DB transaction to prevent inconsistent state if any step fails.
    let mut db_tx = db
        .begin()
        .await
        .context("begin purchase webhook transaction")?;

    // Both halves or neither, matching the column pair's CHECK: a currency with
    // no amount is not something we can put on a receipt.
    let presentment = session
        .presentment
        .as_ref()
        .and_then(|p| Some((p.amount?, p.currency.as_deref()?)));

    match db::transactions::complete_transaction(
        &mut *db_tx,
        &session_id,
        session.payment_intent_id.as_deref(),
        presentment,
    )
    .await
    {
        Ok(Some(tx)) => {
            tracing::info!(
                buyer_id = %buyer_id, seller_id = %seller_id, item_id = %item_id_display, amount_cents = %tx.amount_cents,
                "transaction completed"
            );

            // Defense-in-depth reconciliation (currency + subtotal) against the
            // server-authoritative credited amount.
            reconcile_checkout_amount(
                bg,
                wam,
                &session_id,
                session,
                i64::from(tx.amount_cents),
                "purchase",
            );

            // Increment denormalized sales_count (inside transaction)
            if let Some(iid) = item_id {
                db::items::increment_sales_count(&mut *db_tx, iid)
                    .await
                    .with_context(|| format!("increment sales count for item {iid}"))?;
            }

            // Promo code use_count is reserved at checkout time (not here) to prevent
            // concurrent checkouts from exceeding max_uses. No increment needed in webhook.

            // Commit the critical data integrity operations
            db_tx
                .commit()
                .await
                .context("commit purchase webhook transaction")?;

            // --- Secondary effects below (outside transaction) ---
            // Consolidated in one re-runnable finalizer so the purchase and cart
            // paths can't drift and a crash-recovery redelivery re-runs safely.
            finalize_purchase_transaction(db, bg, email, wam, config, &tx, buyer_id, seller_id)
                .await;

            if let Err(e) = db::subscriptions::log_subscription_event(
                db,
                None,
                event_id,
                MnwEventName::CheckoutCompletedPurchase,
                &serde_json::json!({"session_id": session_id}),
            )
            .await
            {
                tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
            }

            // Check for a pending refund that arrived before this payment webhook.
            // If found, process it now that the transaction is completed.
            check_pending_refund(db, &payment_intent_id).await;
        }
        Ok(None) => {
            // No row flipped to completed. Either this is a benign duplicate of
            // an already-finalized purchase, OR the first attempt crashed AFTER
            // committing the completed status but BEFORE running finalize, in
            // which case the buyer holds a completed purchase with no license
            // key / no splits. Re-fetch completed rows for the session and
            // re-run the idempotent finalizer; only escalate if none exist
            // (genuinely orphaned: payment took, rows never created).
            let completed =
                db::transactions::get_completed_transactions_for_session(db, &session_id)
                    .await
                    .context("re-fetch completed transactions for crash recovery")?;
            if completed.is_empty() {
                escalate_if_orphaned_session(
                    db,
                    bg,
                    wam,
                    &session_id,
                    &payment_intent_id,
                    "transaction",
                )
                .await?;
            } else {
                for tx in &completed {
                    tracing::info!(
                        session_id = %session_id, transaction_id = %tx.id,
                        "crash-recovery: re-running finalize for already-completed session"
                    );
                    finalize_purchase_transaction(
                        db, bg, email, wam, config, tx, buyer_id, seller_id,
                    )
                    .await;
                }
            }
        }
        Err(e) => {
            tracing::error!(session_id = %session_id, error = ?e, "failed to complete transaction");
            return Err(e);
        }
    }

    Ok(())
}

/// Handle checkout.session.completed for cart (multi-item) purchases
#[tracing::instrument(skip_all, name = "stripe::handle_cart_checkout")]
pub(super) async fn handle_cart_checkout_completed(
    db: &PgPool,
    bg: &crate::background::BackgroundTx,
    email: &EmailClient,
    wam: Option<&WamClient>,
    config: &Config,
    session: &crate::payments::CheckoutCompletion,
    event_id: &str,
) -> Result<()> {
    let session_id = session.session_id.clone();
    tracing::info!(session_id = %session_id, "processing completed cart checkout");

    let meta = crate::payments::CartCheckoutMetadata::from_metadata(session.metadata.as_ref())?;
    let buyer_id = meta.buyer_id;
    let seller_id = meta.seller_id;

    // Display/logging copy only; the DB write below passes `session.payment_intent_id`
    // directly so a PI-less session stores NULL, not a literal "unknown" that would
    // collide with other PI-less rows in the money-keyed lookup column (Run 9).
    let payment_intent_id = session.payment_intent_id.clone().unwrap_or_default();

    // Complete ALL pending transactions for this session in a single DB transaction
    let mut db_tx = db.begin().await.context("begin cart webhook transaction")?;

    let completed_txs = db::transactions::complete_cart_transactions(
        &mut *db_tx,
        &session_id,
        session.payment_intent_id.as_deref(),
    )
    .await
    .context("complete cart transactions")?;

    if completed_txs.is_empty() {
        // Same crash-recovery shape as the single-item handler: re-fetch
        // completed rows for the session and re-run the idempotent finalizer
        // before falling through to orphan escalation.
        db_tx.commit().await.ok();
        let completed = db::transactions::get_completed_transactions_for_session(db, &session_id)
            .await
            .context("re-fetch completed cart transactions for crash recovery")?;
        if completed.is_empty() {
            escalate_if_orphaned_session(
                db,
                bg,
                wam,
                &session_id,
                &payment_intent_id,
                "cart transactions",
            )
            .await?;
        } else {
            for tx in &completed {
                tracing::info!(
                    session_id = %session_id, transaction_id = %tx.id,
                    "crash-recovery: re-running finalize for already-completed cart session"
                );
                finalize_purchase_transaction(db, bg, email, wam, config, tx, buyer_id, seller_id)
                    .await;
            }
        }
        return Ok(());
    }

    tracing::info!(
        session_id = %session_id, buyer_id = %buyer_id, seller_id = %seller_id,
        count = completed_txs.len(), "cart transactions completed"
    );

    // Defense-in-depth reconciliation (currency + subtotal): the sum of the
    // credited transactions should equal Stripe's pre-tax subtotal.
    let cart_credited: i64 = completed_txs
        .iter()
        .map(|tx| i64::from(tx.amount_cents))
        .sum();
    reconcile_checkout_amount(bg, wam, &session_id, session, cart_credited, "cart");

    // Increment sales count for each item
    for tx in &completed_txs {
        if let Some(item_id) = tx.item_id {
            db::items::increment_sales_count(&mut *db_tx, item_id)
                .await
                .with_context(|| format!("increment sales count for item {item_id}"))?;
        }
    }

    db_tx
        .commit()
        .await
        .context("commit cart webhook transaction")?;

    // Remove purchased items from cart (items stay in cart until payment succeeds,
    // so cancelled checkouts don't lose cart contents)
    db::cart::remove_seller_items_from_cart(db, buyer_id, seller_id)
        .await
        .context("remove cart items after successful payment")?;

    // --- Secondary effects (outside transaction) ---
    // One re-runnable finalizer per transaction; shared with the single-item
    // path so the effect blocks can't drift. Idempotent on a crash-recovery
    // redelivery. (clear_contact_revocation runs per-tx that opted in, which is
    // a no-op once already cleared.)
    for tx in &completed_txs {
        finalize_purchase_transaction(db, bg, email, wam, config, tx, buyer_id, seller_id).await;
    }

    if let Err(e) = db::subscriptions::log_subscription_event(
        db,
        None,
        event_id,
        MnwEventName::CheckoutCompletedCart,
        &serde_json::json!({"session_id": session_id, "item_count": completed_txs.len()}),
    )
    .await
    {
        tracing::warn!(event_id = %event_id, error = ?e, "failed to log cart checkout event");
    }

    // Check for pending refund
    check_pending_refund(db, &payment_intent_id).await;

    Ok(())
}

/// Handle checkout.session.completed for subscriptions
#[tracing::instrument(skip_all, name = "stripe::handle_subscription_checkout")]
pub(super) async fn handle_subscription_checkout_completed(
    db: &PgPool,
    bg: &crate::background::BackgroundTx,
    email: &EmailClient,
    session: &crate::payments::CheckoutCompletion,
    event_id: &str,
) -> Result<()> {
    let session_id = session.session_id.clone();
    tracing::info!(session_id = %session_id, "processing completed subscription checkout");

    // Extract subscription-specific metadata (already typed IDs)
    let raw_metadata = SubscriptionCheckoutMetadata::from_metadata(session.metadata.as_ref())?;
    let subscriber_id = raw_metadata.subscriber_id;
    let project_id = raw_metadata.project_id;
    let tier_id = raw_metadata.tier_id;

    // Get the Stripe subscription ID from the session
    let stripe_subscription_id = session.subscription_id.clone().ok_or_else(|| {
        tracing::error!("Subscription checkout completed but no subscription ID on session");
        AppError::BadRequest("Missing subscription ID on session".to_string())
    })?;

    // Get the Stripe customer ID from the session
    let stripe_customer_id = session.customer_id.clone().ok_or_else(|| {
        tracing::error!("Subscription checkout completed but no customer ID on session");
        AppError::BadRequest("Missing customer ID on session".to_string())
    })?;

    // Create the subscription record + increment promo code in a single transaction.
    let mut tx = db
        .begin()
        .await
        .context("begin subscription webhook transaction")?;

    let Some(sub) = db::subscriptions::create_subscription(
        &mut tx,
        subscriber_id,
        tier_id,
        project_id,
        &stripe_subscription_id,
        &stripe_customer_id,
    )
    .await
    .context("create subscription record")?
    else {
        tracing::info!(
            subscriber_id = %subscriber_id, project_id = %project_id,
            "subscription already exists, ignoring duplicate"
        );
        return Ok(());
    };

    // Promo code use_count is reserved at checkout time (not here) to prevent
    // concurrent checkouts from exceeding max_uses. No increment needed in webhook.

    // Delete the pending promo-hold transaction (created at checkout time so
    // cleanup_stale_pending_transactions can release the code if abandoned).
    db::transactions::delete_subscription_pending_transaction(&mut *tx, &session_id)
        .await
        .context("delete subscription pending promo-hold transaction")?;

    tx.commit()
        .await
        .context("commit subscription webhook transaction")?;

    tracing::info!(
        subscription_id = %sub.id, subscriber_id = %subscriber_id, project_id = %project_id, tier_id = %tier_id,
        "subscription created"
    );

    // Send subscription started email (fire-and-forget)
    if let (Ok(Some(subscriber)), Ok(Some(tier)), Ok(Some(project))) = (
        db::users::get_user_by_id(db, subscriber_id).await,
        db::subscriptions::get_subscription_tier_by_id(db, tier_id).await,
        db::projects::get_project_by_id(db, project_id).await,
    ) {
        // The tier was priced by the project's owner, in the owner's currency.
        let creator_currency = db::users::get_user_by_id(db, project.user_id)
            .await
            .ok()
            .flatten()
            .map(|u| u.settlement_currency)
            .unwrap_or_default();
        let price = helpers::format_price(tier.price_cents, creator_currency);
        let sub_email = subscriber.email.clone();
        let sub_name = subscriber.display_name;
        let tier_name = tier.name;
        let project_title = project.title;
        let email = email.clone();
        bg.spawn("subscription started", async move {
            if let Err(e) = email
                .send_subscription_started(
                    &sub_email,
                    sub_name.as_deref(),
                    &tier_name,
                    &project_title,
                    &price,
                )
                .await
            {
                tracing::error!(error = ?e, "failed to send subscription started");
            }
        });
    }

    if let Err(e) = db::subscriptions::log_subscription_event(
        db, Some(sub.id), event_id, MnwEventName::CheckoutCompletedSubscription,
        &serde_json::json!({"session_id": session_id, "stripe_subscription_id": stripe_subscription_id}),
    ).await {
        tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
    }

    Ok(())
}

/// Handle checkout.session.completed for Fan+ subscriptions
#[tracing::instrument(skip_all, name = "stripe::handle_fan_plus_checkout")]
pub(super) async fn handle_fan_plus_checkout_completed(
    db: &PgPool,
    bg: &crate::background::BackgroundTx,
    email: &EmailClient,
    session: &crate::payments::CheckoutCompletion,
    event_id: &str,
) -> Result<()> {
    let session_id = session.session_id.clone();
    tracing::info!(session_id = %session_id, "processing completed Fan+ checkout");

    let metadata = FanPlusCheckoutMetadata::from_metadata(session.metadata.as_ref())?;
    let user_id = metadata.user_id;

    // Get the Stripe subscription ID from the session
    let stripe_subscription_id = session.subscription_id.clone().ok_or_else(|| {
        tracing::error!("Fan+ checkout completed but no subscription ID on session");
        AppError::BadRequest("Missing subscription ID on session".to_string())
    })?;

    // Get the Stripe customer ID from the session
    let stripe_customer_id = session.customer_id.clone().ok_or_else(|| {
        tracing::error!("Fan+ checkout completed but no customer ID on session");
        AppError::BadRequest("Missing customer ID on session".to_string())
    })?;

    // Create the subscription record. Idempotent via ON CONFLICT (user_id) DO
    // UPDATE with a guard WHERE: a duplicate webhook for an unchanged row updates
    // nothing and RETURNING yields no row (-> None below, "already exists"); a
    // genuine re-subscribe updates in place.
    let Some(sub) = db::fan_plus::create_fan_plus_subscription(
        db,
        user_id,
        &stripe_subscription_id,
        &stripe_customer_id,
    )
    .await
    .with_context(|| format!("create Fan+ subscription for user {user_id}"))?
    else {
        tracing::info!(user_id = %user_id, "Fan+ subscription already exists, ignoring duplicate");
        return Ok(());
    };

    tracing::info!(
        subscription_id = %sub.id, user_id = %user_id,
        "Fan+ subscription created"
    );

    // Send welcome email (fire-and-forget)
    if let Ok(Some(user)) = db::users::get_user_by_id(db, user_id).await {
        let user_email = user.email.clone();
        let user_name = user.display_name;
        let email = email.clone();
        bg.spawn("Fan+ welcome", async move {
            if let Err(e) = email
                .send_fan_plus_welcome(&user_email, user_name.as_deref())
                .await
            {
                tracing::error!(error = ?e, "failed to send Fan+ welcome");
            }
        });
    }

    if let Err(e) = db::subscriptions::log_subscription_event(
        db, None, event_id, MnwEventName::CheckoutCompletedFanPlus,
        &serde_json::json!({"session_id": session_id, "stripe_subscription_id": stripe_subscription_id}),
    ).await {
        tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
    }

    Ok(())
}

/// Handle checkout.session.completed for creator tier subscriptions
#[tracing::instrument(skip_all, name = "stripe::handle_creator_tier_checkout")]
pub(super) async fn handle_creator_tier_checkout_completed(
    db: &PgPool,
    bg: &crate::background::BackgroundTx,
    wam: Option<&WamClient>,
    payments: &Billing,
    session: &crate::payments::CheckoutCompletion,
    event_id: &str,
) -> Result<()> {
    let session_id = session.session_id.clone();
    tracing::info!(session_id = %session_id, "processing completed creator tier checkout");

    let metadata = CreatorTierCheckoutMetadata::from_metadata(session.metadata.as_ref())?;
    let user_id = metadata.user_id;
    let tier: db::CreatorTier = metadata
        .tier
        .parse()
        .map_err(|_| AppError::BadRequest(format!("Invalid tier: {}", metadata.tier)))?;

    // Get the Stripe subscription ID from the session
    let stripe_subscription_id = session.subscription_id.clone().ok_or_else(|| {
        tracing::error!("Creator tier checkout completed but no subscription ID on session");
        AppError::BadRequest("Missing subscription ID on session".to_string())
    })?;

    // Get the Stripe customer ID from the session
    let stripe_customer_id = session.customer_id.clone().ok_or_else(|| {
        tracing::error!("Creator tier checkout completed but no customer ID on session");
        AppError::BadRequest("Missing customer ID on session".to_string())
    })?;

    // Create the subscription record. Idempotent via ON CONFLICT (user_id) DO
    // UPDATE with a guard WHERE (`stripe_subscription_id != EXCLUDED OR status
    // != 'active'`): a duplicate webhook updates nothing and RETURNING yields no
    // row (-> None below); a genuine tier-switch or re-subscribe overwrites the
    // row with the new subscription id and flips it active.
    let Some(sub) = db::creator_tiers::create_creator_subscription(
        db,
        user_id,
        &stripe_subscription_id,
        &stripe_customer_id,
        tier,
    )
    .await
    .with_context(|| format!("create creator tier subscription for user {user_id}"))?
    else {
        tracing::info!(user_id = %user_id, "Creator tier subscription already exists, ignoring duplicate");
        return Ok(());
    };

    // Sync the denormalized creator_tier column on users
    db::creator_tiers::sync_user_creator_tier(db, user_id)
        .await
        .with_context(|| format!("sync creator tier for user {user_id}"))?;

    // Auto-unhide: restore items hidden by post-grace enforcement
    match db::items::unhide_all_items_for_user(db, user_id).await {
        Ok(count) if count > 0 => {
            tracing::info!(user_id = %user_id, items_unhidden = count, "auto-unhidden items after tier re-subscription");
        }
        Err(e) => {
            tracing::warn!(user_id = %user_id, error = ?e, "failed to unhide items after tier re-subscription");
        }
        _ => {}
    }

    // Auto-unpause: if this creator was paused and just re-subscribed, clear the pause
    // and un-cancel any fan subscriptions that haven't expired yet.
    if let Ok(Some(db_user)) = db::users::get_user_by_id(db, user_id).await
        && db_user.is_creator_paused()
    {
        db::users::unpause_creator(db, user_id)
            .await
            .with_context(|| format!("unpause creator {user_id}"))?;

        // Un-cancel active fan subscriptions (clear cancel_at_period_end). Fanned
        // out on the background queue: doing it inline here let a creator with
        // many fans stall the webhook past Stripe's delivery timeout, which
        // triggers a retry that re-runs the whole loop.
        if let (Some(stripe), Some(stripe_account_id)) =
            (&payments.stripe, &db_user.stripe_account_id)
        {
            let fan_subs = db::subscriptions::get_active_subscriptions_by_creator(db, user_id)
                .await
                .with_context(|| format!("fetch active fan subs for unpause {user_id}"))?;
            let ids = fan_subs
                .into_iter()
                .map(|s| s.stripe_subscription_id)
                .collect();
            crate::payments::fan_ops::spawn_fan_sub_fanout(
                bg,
                std::sync::Arc::clone(stripe),
                stripe_account_id.clone(),
                ids,
                crate::payments::fan_ops::FanSubOp::CancelAtPeriodEnd(false),
                wam.cloned(),
            );
        }

        tracing::info!(user_id = %user_id, "creator auto-unpaused after re-subscribing to tier");
    }

    tracing::info!(
        user_id = %user_id, tier = %tier,
        "creator tier subscription created"
    );

    if let Err(e) = db::subscriptions::log_subscription_event(
        db,
        None,
        event_id,
        MnwEventName::CheckoutCompletedCreatorTier,
        &serde_json::json!({
            "session_id": session_id,
            "stripe_subscription_id": stripe_subscription_id,
            "tier": sub.tier,
        }),
    )
    .await
    {
        tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
    }

    Ok(())
}

/// Handle checkout.session.completed for tips
#[tracing::instrument(skip_all, name = "stripe::handle_tip_checkout")]
pub(super) async fn handle_tip_checkout_completed(
    db: &PgPool,
    bg: &crate::background::BackgroundTx,
    email: &EmailClient,
    wam: Option<&WamClient>,
    config: &Config,
    session: &crate::payments::CheckoutCompletion,
    event_id: &str,
) -> Result<()> {
    let session_id = session.session_id.clone();
    tracing::info!(session_id = %session_id, "processing completed tip checkout");

    let metadata = TipCheckoutMetadata::from_metadata(session.metadata.as_ref())?;
    let tipper_id = metadata.tipper_id;
    let recipient_id = metadata.recipient_id;

    // Complete the tip (idempotent). A PI-less session stores NULL (not a literal
    // "unknown") in the money-keyed lookup column (Run 9).
    match db::tips::complete_tip(db, &session_id, session.payment_intent_id.as_deref())
        .await
        .context("complete tip")?
    {
        Some(tip) => {
            tracing::info!(
                tip_id = %tip.id, tipper_id = %tipper_id, recipient_id = %recipient_id,
                amount_cents = %tip.amount_cents, "tip completed"
            );

            // Log the event by id for audit parity with the other checkout
            // handlers (record_tip_splits mints split revenue, so an event-id
            // ledger entry matters for reconciliation). MINOR, Run #2 Payments.
            if let Err(e) = db::subscriptions::log_subscription_event(
                db,
                None,
                event_id,
                MnwEventName::CheckoutCompletedTip,
                &serde_json::json!({"session_id": session_id, "tip_id": tip.id}),
            )
            .await
            {
                tracing::warn!(event_id = %event_id, error = ?e, "failed to log tip event");
            }

            // Record revenue splits if the tip's project has members
            if let Some(project_id) = tip.project_id {
                record_tip_splits(db, tip.id, project_id, tip.amount_cents).await;
            }

            // Send tip notification email (fire-and-forget)
            send_tip_email(db, bg, email, config, &tip, tipper_id, recipient_id);
        }
        None => {
            // No pending row flipped to completed. Either a benign duplicate of
            // an already-finalized tip, OR the first delivery crashed AFTER
            // completing the tip but BEFORE recording splits, in which case
            // collaborators hold a completed tip with no split rows. Re-fetch the
            // tip and re-run the idempotent split write (ON CONFLICT DO NOTHING,
            // migration 163); a genuine duplicate is a no-op. Run 20 Payments.
            let recovered = db::tips::get_tip_by_session(db, &session_id)
                .await
                .context("re-fetch tip for crash recovery")?;
            match recovered {
                // A tip row exists (get_tip_by_session is status-agnostic), so
                // this is either crash-recovery (re-run the idempotent splits) or
                // a benign duplicate.
                Some(tip) => {
                    if let Some(project_id) = tip.project_id {
                        tracing::info!(
                            session_id = %session_id, tip_id = %tip.id,
                            "crash-recovery: re-running tip splits for already-completed tip"
                        );
                        record_tip_splits(db, tip.id, project_id, tip.amount_cents).await;
                    } else {
                        tracing::info!(session_id = %session_id, "tip already completed, ignoring duplicate webhook");
                    }
                }
                // No tip row AT ALL for a paid session: the checkout session was
                // created but the pending_tip insert never landed, so the tipper
                // was charged and the recipient got nothing. Escalate like the
                // purchase/cart/guest orphan path instead of logging a benign
                // "duplicate", tips were the one checkout family without orphan
                // escalation (Run 21 payments).
                None => {
                    let pi = session.payment_intent_id.as_deref().unwrap_or("");
                    tracing::error!(
                        session_id = %session_id, payment_intent_id = %pi,
                        "orphaned paid session (tip): payment completed but no tip row exists, manual reconciliation required"
                    );
                    if let Some(wam) = wam.cloned() {
                        let sid = session_id.clone();
                        let pi = pi.to_string();
                        // Drained background pool, not a raw tokio::spawn: a
                        // charged-but-undelivered tip ticket must survive a
                        // mid-deploy restart (audit Run 22).
                        bg.spawn("stripe orphaned-tip ticket", async move {
                            let body = format!(
                                "Tip checkout session {sid} (payment_intent {pi}) completed at Stripe but has \
                                 NO tip row, the pending tip was never created. The tipper was charged and the \
                                 recipient received nothing. Reconcile manually: refund the payment or recreate the tip."
                            );
                            wam.create_ticket("Orphaned paid tip session", Some(&body), "high", "stripe-orphaned-tip", Some(&sid)).await;
                        });
                    }
                }
            }
        }
    }

    Ok(())
}

/// Handle checkout.session.completed for guest purchases (no MNW account).
///
/// Extracts the buyer's email from Stripe, completes the transaction, and
/// auto-attaches to an existing account if the email matches.
#[tracing::instrument(skip_all, name = "stripe::handle_guest_checkout")]
pub(super) async fn handle_guest_checkout_completed(
    db: &PgPool,
    bg: &crate::background::BackgroundTx,
    email: &EmailClient,
    wam: Option<&WamClient>,
    config: &Config,
    session: &crate::payments::CheckoutCompletion,
    _event_id: &str,
) -> Result<()> {
    use crate::payments::GuestCheckoutMetadata;

    let session_id = session.session_id.clone();
    tracing::info!(session_id = %session_id, "processing completed guest checkout");

    let meta = GuestCheckoutMetadata::from_metadata(session.metadata.as_ref())?;

    // Buyer email as Stripe collected it, flattened out of customer_details
    // during normalization.
    let guest_email = session
        .customer_email
        .as_deref()
        .unwrap_or("unknown@guest")
        .to_string();

    // Display/logging copy only; the DB write below passes `session.payment_intent_id`
    // directly so a PI-less session stores NULL, not a literal "unknown" that would
    // collide with other PI-less rows in the money-keyed lookup column (Run 9).
    let payment_intent_id = session.payment_intent_id.clone().unwrap_or_default();

    // Complete the guest transaction and increment sales count in a single DB transaction
    // (matching the non-guest path pattern to prevent counter drift on partial failure)
    let mut db_tx = db
        .begin()
        .await
        .context("begin guest checkout webhook transaction")?;

    // Do NOT auto-attach on an email match. Stripe collects the buyer's email
    // but does not prove the guest controls it, so matching alone would let
    // someone drop a purchase into a stranger's verified library. The buyer
    // always claims via the emailed claim link, which authenticates the
    // recipient; the license key (if any) is minted at claim time
    // (Run #21 Payments MINOR-1 / Max's call 2026-06-15).
    match db::transactions::complete_guest_transaction(
        &mut *db_tx,
        &session_id,
        session.payment_intent_id.as_deref(),
        &guest_email,
    )
    .await?
    {
        Some(tx) => {
            tracing::info!(
                session_id = %session_id,
                guest_email = %guest_email,
                item_id = %meta.item_id,
                "guest transaction completed"
            );

            // Defense-in-depth reconciliation (currency + subtotal), mirroring
            // the single-item and cart paths.
            reconcile_checkout_amount(
                bg,
                wam,
                &session_id,
                session,
                i64::from(tx.amount_cents),
                "guest",
            );

            // Increment sales count inside transaction
            db::items::increment_sales_count(&mut *db_tx, meta.item_id)
                .await
                .with_context(|| {
                    format!("increment sales count for guest item {}", meta.item_id)
                })?;

            db_tx
                .commit()
                .await
                .context("commit guest checkout webhook transaction")?;

            // --- Secondary effects below (outside transaction) ---
            // One re-runnable finalizer (splits + confirmation + sale email).
            finalize_guest_transaction(
                db,
                bg,
                email,
                config,
                &tx,
                &guest_email,
                meta.item_id,
                meta.seller_id,
            );
        }
        None => {
            db_tx.commit().await.ok();
            // Crash-recovery: re-fetch completed guest rows for the session and
            // re-run the idempotent finalizer; only escalate if none exist.
            let completed =
                db::transactions::get_completed_transactions_for_session(db, &session_id)
                    .await
                    .context("re-fetch completed guest transactions for crash recovery")?;
            if completed.is_empty() {
                escalate_if_orphaned_session(
                    db,
                    bg,
                    wam,
                    &session_id,
                    &payment_intent_id,
                    "guest transaction",
                )
                .await?;
            } else {
                for tx in &completed {
                    let guest_email = tx
                        .guest_email
                        .clone()
                        .unwrap_or_else(|| guest_email.clone());
                    tracing::info!(
                        session_id = %session_id, transaction_id = %tx.id,
                        "crash-recovery: re-running finalize for already-completed guest session"
                    );
                    if let Some(item_id) = tx.item_id {
                        finalize_guest_transaction(
                            db,
                            bg,
                            email,
                            config,
                            tx,
                            &guest_email,
                            item_id,
                            meta.seller_id,
                        );
                    }
                }
            }
        }
    }

    Ok(())
}

/// Handle checkout.session.completed for an end-user SyncKit app subscription.
/// Inserts the `app_sync_subscriptions` row; subsequent
/// `customer.subscription.updated/.deleted` events keep it in sync.
#[tracing::instrument(skip_all, name = "stripe::handle_synckit_app_sub_checkout")]
pub(super) async fn handle_synckit_app_sub_checkout_completed(
    db: &PgPool,
    session: &crate::payments::CheckoutCompletion,
    event_id: &str,
) -> Result<()> {
    let session_id = session.session_id.clone();
    tracing::info!(session_id = %session_id, "processing completed SyncKit app subscription checkout");

    let meta = SynckitAppSubCheckoutMetadata::from_metadata(session.metadata.as_ref())?;

    let stripe_subscription_id = session
        .subscription_id
        .clone()
        .ok_or_else(|| AppError::BadRequest("Missing subscription ID on session".to_string()))?;
    let stripe_customer_id = session
        .customer_id
        .clone()
        .ok_or_else(|| AppError::BadRequest("Missing customer ID on session".to_string()))?;

    let inserted = db::synckit::create_app_sync_subscription(
        db,
        &db::synckit::NewAppSyncSubscription {
            user_id: meta.user_id,
            app_id: meta.app_id,
            stripe_subscription_id: &stripe_subscription_id,
            stripe_customer_id: &stripe_customer_id,
            interval: &meta.interval,
            storage_limit_bytes: meta.storage_limit_bytes.unwrap_or(0),
        },
    )
    .await
    .with_context(|| {
        format!(
            "create app sync subscription user={} app={}",
            meta.user_id, meta.app_id
        )
    })?;

    if !inserted {
        tracing::info!(
            user_id = %meta.user_id,
            app_id = %meta.app_id,
            "SyncKit app subscription already exists, ignoring duplicate webhook"
        );
    }

    let _ = event_id;
    Ok(())
}