//! Settlement of platform-funded credits (Fan+ renewal credit reimbursement).
//!
//! A platform-wide credit reduces what a buyer pays on a creator's Direct Charge;
//! MNW owes the creator the discounted amount so they still net the full price
//! ("0% platform fee, creators keep everything"). The obligation lives on the
//! `transactions` row (`platform_credit_cents`); the scheduler sweep claims it,
//! creates a platform -> connected transfer with a deterministic idempotency key,
//! then stamps `platform_credit_settled_at`.
//!
//! Lifecycle mirrors [`super::pending_refunds`]: claim (set `claimed_at`) -> settle
//! (set `settled_at`). A graceful transfer failure clears `claimed_at` for retry; a
//! process death between claim and settle leaves the row claimed-but-unsettled and
//! is surfaced by [`get_stale_credits`] for human reconciliation rather than blindly
//! retried.

use sqlx::PgPool;

use super::validated_types::Cents;
use super::{TransactionId, UserId};
use crate::error::Result;

/// A completed transaction owing a platform-funded credit to its seller.
#[derive(Debug, sqlx::FromRow)]
pub struct PlatformCredit {
    pub transaction_id: TransactionId,
    pub seller_id: UserId,
    pub amount_cents: Cents,
}

/// Atomically claim the oldest unsettled platform credit (set `claimed_at`), so a
/// single scheduler worker settles it once. Returns `None` when nothing is owed.
/// Settlement is recorded separately by [`mark_settled`] only after the transfer
/// succeeds, a claim that never settles (process killed mid-transfer) is escalated
/// by [`get_stale_credits`], not auto-retried.
pub async fn claim_unsettled_credit(pool: &PgPool) -> Result<Option<PlatformCredit>> {
    let row = sqlx::query_as!(
        PlatformCredit,
        r#"
        UPDATE transactions
        SET platform_credit_claimed_at = NOW()
        WHERE id = (
            SELECT id FROM transactions
            WHERE status = 'completed'
              AND platform_credit_cents > 0
              AND platform_credit_settled_at IS NULL
              AND platform_credit_claimed_at IS NULL
            ORDER BY completed_at
            LIMIT 1
            FOR UPDATE SKIP LOCKED
        )
        RETURNING id AS "transaction_id!: TransactionId",
                  seller_id AS "seller_id!: UserId",
                  platform_credit_cents AS "amount_cents!: Cents"
        "#,
    )
    .fetch_optional(pool)
    .await?;

    Ok(row)
}

/// Record that a claimed credit's transfer succeeded, storing the Stripe
/// transfer id so the credit can be reversed if the sale is later refunded.
/// Idempotent. Runtime-checked query (the `platform_credit_transfer_id` column
/// is newer than the offline sqlx cache), per [`get_stale_credits`].
pub async fn mark_settled(
    pool: &PgPool,
    transaction_id: TransactionId,
    transfer_id: &str,
) -> Result<()> {
    sqlx::query(
        "UPDATE transactions SET platform_credit_settled_at = NOW(), platform_credit_transfer_id = $2 WHERE id = $1",
    )
    .bind(transaction_id)
    .bind(transfer_id)
    .execute(pool)
    .await?;
    Ok(())
}

/// A settled platform credit whose sale was refunded and whose MNW -> creator
/// transfer must be reversed to claw the reimbursement back.
#[derive(Debug, sqlx::FromRow)]
pub struct ReversibleCredit {
    pub transaction_id: TransactionId,
    pub amount_cents: Cents,
    pub transfer_id: String,
}

/// Up to `limit` settled platform credits sitting on refunded transactions that
/// haven't been reversed yet, oldest first. The reversal uses a deterministic
/// idempotency key, so the single-instance scheduler can process these
/// sequentially and safely. Runtime-checked query (new columns).
pub async fn get_reversible_credits(pool: &PgPool, limit: i64) -> Result<Vec<ReversibleCredit>> {
    let rows = sqlx::query_as::<_, ReversibleCredit>(
        r"
        SELECT id AS transaction_id,
               platform_credit_cents AS amount_cents,
               platform_credit_transfer_id AS transfer_id
        FROM transactions
        WHERE status = 'refunded'
          AND platform_credit_cents > 0
          AND platform_credit_settled_at IS NOT NULL
          AND platform_credit_transfer_id IS NOT NULL
          AND platform_credit_reversed_at IS NULL
        ORDER BY completed_at
        LIMIT $1
        ",
    )
    .bind(limit)
    .fetch_all(pool)
    .await?;

    Ok(rows)
}

/// Mark a platform credit reversed (funds clawed back). Idempotent.
/// Runtime-checked query (new column).
pub async fn mark_reversed(pool: &PgPool, transaction_id: TransactionId) -> Result<()> {
    sqlx::query("UPDATE transactions SET platform_credit_reversed_at = NOW() WHERE id = $1")
        .bind(transaction_id)
        .execute(pool)
        .await?;
    Ok(())
}

/// Release a claimed-but-unsettled credit back to the queue after a *graceful*
/// transfer failure (transient error where nothing external happened). A
/// non-graceful failure (process killed) cannot reach here; that row stays
/// claimed-but-unsettled and is escalated by the stale sweep. Idempotent.
pub async fn unclaim_credit(pool: &PgPool, transaction_id: TransactionId) -> Result<()> {
    sqlx::query!(
        "UPDATE transactions SET platform_credit_claimed_at = NULL WHERE id = $1",
        transaction_id as TransactionId,
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// Per-tick cap on the stale-credit escalation sweep (mirrors `STALE_REFUND_BATCH`).
pub const STALE_CREDIT_BATCH: i64 = 100;

/// A stale platform credit needing human reconciliation.
#[derive(Debug, sqlx::FromRow)]
pub struct StaleCredit {
    pub transaction_id: TransactionId,
    pub seller_id: UserId,
    pub amount_cents: Cents,
    pub completed_at: Option<chrono::DateTime<chrono::Utc>>,
}

/// Up to [`STALE_CREDIT_BATCH`] credits claimed more than `age` ago that never
/// settled (the crash-window between claim and transfer), oldest first. The transfer
/// uses a deterministic idempotency key so a human can safely re-trigger it.
pub async fn get_stale_credits(pool: &PgPool, age: chrono::Duration) -> Result<Vec<StaleCredit>> {
    let cutoff = chrono::Utc::now() - age;
    // runtime-checked bind of a chrono cutoff, per db::pending_refunds::get_stale_refunds.
    let rows = sqlx::query_as::<_, StaleCredit>(
        r"
        SELECT id AS transaction_id, seller_id, platform_credit_cents AS amount_cents, completed_at
        FROM transactions
        WHERE status = 'completed'
          AND platform_credit_cents > 0
          AND platform_credit_settled_at IS NULL
          AND platform_credit_claimed_at IS NOT NULL
          AND platform_credit_claimed_at < $1
          AND platform_credit_escalated_at IS NULL
        ORDER BY completed_at
        LIMIT $2
        ",
    )
    .bind(cutoff)
    .bind(STALE_CREDIT_BATCH)
    .fetch_all(pool)
    .await?;

    Ok(rows)
}

/// Mark a stale credit escalated (alert sent, won't be re-alerted).
pub async fn mark_escalated(pool: &PgPool, transaction_id: TransactionId) -> Result<()> {
    sqlx::query!(
        "UPDATE transactions SET platform_credit_escalated_at = NOW() WHERE id = $1",
        transaction_id as TransactionId,
    )
    .execute(pool)
    .await?;
    Ok(())
}