//! Pending refunds queue for out-of-order webhook delivery.
//!
//! When a `charge.refunded` webhook arrives before its matching
//! `checkout.session.completed`, the refund data is stored here.
//! The scheduler and checkout handler both check for pending matches.

use sqlx::PgPool;

use super::validated_types::Cents;

use crate::error::Result;

/// Insert a pending refund for later matching. Deduplicates on
/// `payment_intent_id`; if a pending (unmatched) refund already exists
/// for this payment intent, the insert is silently skipped.
pub async fn insert_pending_refund(
    pool: &PgPool,
    payment_intent_id: &str,
    amount: i64,
    amount_refunded: i64,
) -> Result<()> {
    sqlx::query!(
        r#"
        INSERT INTO pending_refunds (payment_intent_id, amount, amount_refunded)
        VALUES ($1, $2, $3)
        ON CONFLICT (payment_intent_id) WHERE matched_at IS NULL DO NOTHING
        "#,
        payment_intent_id,
        amount,
        amount_refunded,
    )
    .execute(pool)
    .await?;

    Ok(())
}

/// Row from the pending_refunds table.
#[derive(Debug, sqlx::FromRow)]
pub struct PendingRefund {
    pub id: uuid::Uuid,
    pub payment_intent_id: String,
    pub amount: Cents,
    pub amount_refunded: Cents,
}

/// Claim a pending refund matching a payment intent ID.
///
/// Atomically marks it as matched (so it is only claimed once) but NOT completed,
/// completion is recorded separately by [`mark_refund_completed`] only after the
/// fallible refund work succeeds. A claim that is never completed (process killed
/// mid-refund) leaves `completed_at IS NULL`, so the stale-refund sweep surfaces it
/// for human escalation (PAY-S1). Returns `None` if no unmatched pending refund exists.
pub async fn claim_pending_refund(
    pool: &PgPool,
    payment_intent_id: &str,
) -> Result<Option<PendingRefund>> {
    let row = sqlx::query_as!(
        PendingRefund,
        r#"
        UPDATE pending_refunds
        SET matched_at = NOW()
        WHERE id = (
            SELECT id FROM pending_refunds
            WHERE payment_intent_id = $1 AND matched_at IS NULL
            LIMIT 1
            FOR UPDATE SKIP LOCKED
        )
        RETURNING id, payment_intent_id, amount AS "amount: Cents", amount_refunded AS "amount_refunded: Cents"
        "#,
        payment_intent_id,
    )
    .fetch_optional(pool)
    .await?;

    Ok(row)
}

/// Record that a claimed pending refund's processing finished successfully.
///
/// Sets `completed_at`; only after this is the row considered fully handled. A
/// claimed row without a `completed_at` (the process died between claim and this
/// call) is surfaced by [`get_stale_refunds`] for manual reconciliation instead of
/// being auto-retried, re-issuing a refund that may already have reached Stripe
/// could double-refund (PAY-S1).
pub async fn mark_refund_completed(pool: &PgPool, id: uuid::Uuid) -> Result<()> {
    sqlx::query!(
        "UPDATE pending_refunds SET completed_at = NOW() WHERE id = $1",
        id
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// Release a claimed pending refund back to the queue (`matched_at` → NULL) after a
/// *graceful* processing failure (a transient error where the handler committed
/// nothing, it is atomic). Releasing re-opens the row so a later webhook delivery
/// can re-claim and retry. A non-graceful failure (process killed) cannot reach
/// here; that row stays matched-but-incomplete and is escalated by the sweep
/// instead (PAY-S1). Idempotent.
pub async fn unclaim_pending_refund(pool: &PgPool, id: uuid::Uuid) -> Result<()> {
    sqlx::query!(
        "UPDATE pending_refunds SET matched_at = NULL WHERE id = $1",
        id
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// Row for stale pending refunds that need escalation.
#[derive(Debug, sqlx::FromRow)]
pub struct StaleRefund {
    pub id: uuid::Uuid,
    pub payment_intent_id: String,
    pub amount: Cents,
    pub amount_refunded: Cents,
    pub created_at: chrono::DateTime<chrono::Utc>,
}

/// Per-tick cap on the stale-refund escalation sweep. It runs every scheduler
/// tick under the tick-wide advisory lock; escalation is idempotent (sets
/// `escalated_at`), so a bound here drains a backlog across ticks instead of
/// letting one unbounded query stall the tick.
pub const STALE_REFUND_BATCH: i64 = 100;

/// Get up to [`STALE_REFUND_BATCH`] pending refunds older than `age` that still
/// need attention and have not been escalated, oldest first. "Need attention" means
/// the refund work never completed: either the row was never matched to a payment
/// (`matched_at IS NULL`), or it was claimed but the process died before recording
/// completion (`matched_at IS NOT NULL AND completed_at IS NULL`), the crash-window
/// case (PAY-S1). Both surface here for human reconciliation.
pub async fn get_stale_refunds(pool: &PgPool, age: chrono::Duration) -> Result<Vec<StaleRefund>> {
    let cutoff = chrono::Utc::now() - age;
    // runtime-checked: binds a chrono `DateTime<Utc>` cutoff (`$1`); a bind
    // parameter's type can't be overridden in the macro when sqlx's `time` and
    // `chrono` features are unified. See db::pending_uploads::get_stale_pending_uploads.
    let rows = sqlx::query_as::<_, StaleRefund>(
        r"
        SELECT id, payment_intent_id, amount, amount_refunded, created_at
        FROM pending_refunds
        WHERE completed_at IS NULL
          AND escalated_at IS NULL
          AND created_at < $1
        ORDER BY created_at
        LIMIT $2
        ",
    )
    .bind(cutoff)
    .bind(STALE_REFUND_BATCH)
    .fetch_all(pool)
    .await?;

    Ok(rows)
}

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