//! DB-layer contract tests for `db::webhook_events`, the webhook dedup +
//! retry-queue module.
//!
//! Audit Run 16 flagged `webhook_events` as a Testing/Concurrency cold spot
//! (type B+): the dedup marker and the retry-claim query are exercised only
//! indirectly through the Stripe webhook flow, with "no two-concurrent-
//! deliveries test." These call the `db::webhook_events` functions directly so
//! the two race-sensitive invariants live where they belong:
//!
//!   1. `mark_event_processed` is idempotent under concurrent duplicate delivery
//!      (Stripe at-least-once), N racing marks of one event leave exactly one row.
//!   2. `get_retryable_events` claims-and-defers so an overlapping scheduler tick
//!      (or a second replica) can't re-claim the same due events within the window.
//!
//! Plus the backoff/dead-letter progression and the retention prune.

use crate::harness::db::TestDb;
use makenotwork::db::webhook_events;

/// Count rows in `processed_webhook_events` (dedup markers).
async fn processed_count(pool: &sqlx::PgPool) -> i64 {
    sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM processed_webhook_events")
        .fetch_one(pool)
        .await
        .expect("count processed")
}

/// Insert a failed event and return its id, backdating `next_retry_at` into the
/// past so it is immediately due (the column defaults to NOW()+60s).
async fn seed_due_event(pool: &sqlx::PgPool, event_type: &str) -> uuid::Uuid {
    webhook_events::insert_failed_event(pool, "stripe", event_type, "{}", None, "boom")
        .await
        .expect("insert failed event");
    sqlx::query_scalar::<_, uuid::Uuid>(
        "UPDATE webhook_events SET next_retry_at = NOW() - INTERVAL '1 minute' \
         WHERE event_type = $1 RETURNING id",
    )
    .bind(event_type)
    .fetch_one(pool)
    .await
    .expect("backdate next_retry_at")
}

// ── dedup marker (`processed_webhook_events`) ────────────────────────────────

#[tokio::test]
async fn is_event_processed_false_before_mark_true_after() {
    let db = TestDb::new().await;
    let id = "evt_dedup_basic";

    assert!(
        !webhook_events::is_event_processed(&db.pool, id)
            .await
            .unwrap(),
        "an unseen event must not read as processed"
    );

    webhook_events::mark_event_processed(&db.pool, id)
        .await
        .unwrap();

    assert!(
        webhook_events::is_event_processed(&db.pool, id)
            .await
            .unwrap(),
        "a marked event must read as processed"
    );
}

#[tokio::test]
async fn mark_event_processed_is_idempotent() {
    let db = TestDb::new().await;
    let id = "evt_dedup_idem";

    // Serial re-marks (redelivery) must not error and must not duplicate the row.
    webhook_events::mark_event_processed(&db.pool, id)
        .await
        .unwrap();
    webhook_events::mark_event_processed(&db.pool, id)
        .await
        .unwrap();
    webhook_events::mark_event_processed(&db.pool, id)
        .await
        .unwrap();

    assert_eq!(
        processed_count(&db.pool).await,
        1,
        "re-marking must be a no-op"
    );
}

/// The core concurrency invariant: Stripe delivers at-least-once, so two
/// deliveries of the same event can race the post-handler mark. `ON CONFLICT DO
/// NOTHING` must collapse them to exactly one dedup row, never a duplicate-key
/// error, never two rows.
#[tokio::test]
async fn mark_event_processed_concurrent_duplicates_leave_one_row() {
    let db = TestDb::new().await;
    let id = "evt_dedup_race";

    let mut handles = Vec::new();
    for _ in 0..24 {
        let pool = db.pool.clone();
        let id = id.to_string();
        handles.push(tokio::spawn(async move {
            webhook_events::mark_event_processed(&pool, &id).await
        }));
    }
    for h in handles {
        // No task may observe a unique-violation, the ON CONFLICT swallows it.
        h.await
            .expect("task panicked")
            .expect("mark must not error under contention");
    }

    assert_eq!(
        processed_count(&db.pool).await,
        1,
        "concurrent duplicate deliveries must yield exactly one dedup row"
    );
}

/// `try_lock_event` must serialize same-event deliveries without blocking: while
/// one delivery holds the per-event lock, a second delivery of the *same* event
/// gets `None` (→ the handler returns 503 for Stripe to redeliver) rather than
/// parking a connection, while a *different* event acquires its lock freely. When
/// the first guard drops, the same event becomes lockable again (Run 23 Conc/Perf:
/// blocking `pg_advisory_xact_lock` → non-blocking `pg_try_advisory_xact_lock`).
#[tokio::test]
async fn try_lock_event_is_non_blocking_and_per_event() {
    let db = TestDb::new().await;

    // First delivery of event A wins the lock and holds it.
    let held = webhook_events::try_lock_event(&db.pool, "evt_lock_A")
        .await
        .expect("first try_lock must not error")
        .expect("first delivery acquires the lock");

    // A second, concurrent delivery of the SAME event must not acquire it, and
    // must return immediately (None), not block.
    assert!(
        webhook_events::try_lock_event(&db.pool, "evt_lock_A")
            .await
            .expect("contended try_lock must not error")
            .is_none(),
        "a same-event delivery must get None while the lock is held"
    );

    // A DIFFERENT event hashes to a different key and is unaffected.
    assert!(
        webhook_events::try_lock_event(&db.pool, "evt_lock_B")
            .await
            .expect("distinct-event try_lock must not error")
            .is_some(),
        "a distinct event must acquire its own lock"
    );

    // Releasing the first guard frees event A for the next delivery.
    //
    // Rolled back explicitly rather than dropped. Dropping a sqlx `Transaction`
    // does not run the ROLLBACK inline, it queues it onto the connection, to be
    // flushed when that connection is next used or returned to the pool. The
    // lock is therefore released *eventually*, not by the time `drop` returns,
    // and asserting re-acquisition straight after a bare `drop(held)` raced that
    // flush: this test failed roughly one run in three under load, and was the
    // only flake in the suite once Sando started gating on it. Production is
    // unaffected (the loser sheds its connection and Stripe redelivers later),
    // but the deterministic release is what a caller can actually rely on, so it
    // is what the test asserts.
    held.rollback()
        .await
        .expect("releasing the held lock must not error");
    assert!(
        webhook_events::try_lock_event(&db.pool, "evt_lock_A")
            .await
            .expect("post-release try_lock must not error")
            .is_some(),
        "the lock must be re-acquirable once the holder releases it"
    );
}

#[tokio::test]
async fn prune_processed_events_respects_retention_boundary() {
    let db = TestDb::new().await;

    webhook_events::mark_event_processed(&db.pool, "evt_old")
        .await
        .unwrap();
    webhook_events::mark_event_processed(&db.pool, "evt_fresh")
        .await
        .unwrap();
    // Age the old marker past a 30-day retention; leave the fresh one at NOW().
    sqlx::query("UPDATE processed_webhook_events SET processed_at = NOW() - INTERVAL '40 days' WHERE event_id = 'evt_old'")
        .execute(&db.pool)
        .await
        .unwrap();

    let deleted = webhook_events::prune_processed_events(&db.pool, 30)
        .await
        .unwrap();
    assert_eq!(deleted, 1, "exactly the aged marker is pruned");

    assert!(
        !webhook_events::is_event_processed(&db.pool, "evt_old")
            .await
            .unwrap()
    );
    assert!(
        webhook_events::is_event_processed(&db.pool, "evt_fresh")
            .await
            .unwrap(),
        "a within-retention marker must survive the prune"
    );
}

// ── retry queue (`webhook_events`) claim-and-defer ───────────────────────────

/// A due event is claimed once, and the same immediate tick can't re-claim it:
/// `get_retryable_events` pushes `next_retry_at` out by 2 minutes as it selects,
/// so an overlapping scheduler tick (or a second replica) gets nothing. This is
/// the "two-concurrent-deliveries" guard the audit noted was untested.
#[tokio::test]
async fn get_retryable_events_claims_and_defers() {
    let db = TestDb::new().await;
    seed_due_event(&db.pool, "claim.a").await;
    seed_due_event(&db.pool, "claim.b").await;

    let first = webhook_events::get_retryable_events(&db.pool)
        .await
        .unwrap();
    assert_eq!(
        first.len(),
        2,
        "both due events are claimed on the first tick"
    );

    let second = webhook_events::get_retryable_events(&db.pool)
        .await
        .unwrap();
    assert!(
        second.is_empty(),
        "a claimed event's next_retry_at is deferred, so an immediate re-tick claims nothing"
    );
}

/// Two overlapping ticks racing the same due events must partition them, never
/// hand the same event to both callers (would double-run the handler).
#[tokio::test]
async fn get_retryable_events_concurrent_ticks_do_not_double_claim() {
    let db = TestDb::new().await;
    for i in 0..6 {
        seed_due_event(&db.pool, &format!("race.claim.{i}")).await;
    }

    let p1 = db.pool.clone();
    let p2 = db.pool.clone();
    let (a, b) = tokio::join!(
        tokio::spawn(async move { webhook_events::get_retryable_events(&p1).await }),
        tokio::spawn(async move { webhook_events::get_retryable_events(&p2).await }),
    );
    let a = a.unwrap().unwrap();
    let b = b.unwrap().unwrap();

    let mut ids: Vec<_> = a.iter().chain(b.iter()).map(|e| e.id).collect();
    let total = ids.len();
    ids.sort();
    ids.dedup();
    assert_eq!(
        ids.len(),
        total,
        "no event id may be claimed by both concurrent ticks"
    );
    assert_eq!(
        ids.len(),
        6,
        "every due event is claimed exactly once across the two ticks"
    );
}

#[tokio::test]
async fn get_retryable_events_excludes_future_and_exhausted() {
    let db = TestDb::new().await;

    // Due now, eligible.
    seed_due_event(&db.pool, "elig.due").await;
    // Scheduled in the future, not yet eligible.
    webhook_events::insert_failed_event(&db.pool, "stripe", "elig.future", "{}", None, "e")
        .await
        .unwrap();
    // Due but retry-exhausted (attempts >= 5), must be skipped so it can't storm.
    let exhausted = seed_due_event(&db.pool, "elig.exhausted").await;
    sqlx::query("UPDATE webhook_events SET attempts = 5 WHERE id = $1")
        .bind(exhausted)
        .execute(&db.pool)
        .await
        .unwrap();

    let due = webhook_events::get_retryable_events(&db.pool)
        .await
        .unwrap();
    let types: Vec<_> = due.iter().map(|e| e.event_type.as_str()).collect();
    assert_eq!(
        types,
        ["elig.due"],
        "only the due, non-exhausted event is returned"
    );
}

// ── backoff + dead-letter progression ────────────────────────────────────────

#[tokio::test]
async fn schedule_retry_sets_retrying_and_backs_off() {
    let db = TestDb::new().await;
    let id = seed_due_event(&db.pool, "backoff.evt").await;

    // First failed retry: attempt count 1, status -> retrying, deferred forward.
    webhook_events::schedule_retry(&db.pool, id, 1, "still failing")
        .await
        .unwrap();

    let (status, attempts, deferred): (String, i32, bool) = sqlx::query_as(
        "SELECT status, attempts, next_retry_at > NOW() FROM webhook_events WHERE id = $1",
    )
    .bind(id)
    .fetch_one(&db.pool)
    .await
    .unwrap();
    assert_eq!(status, "retrying");
    assert_eq!(attempts, 1);
    assert!(
        deferred,
        "next_retry_at is pushed into the future by the backoff"
    );

    // Immediately after scheduling, it is not due, so a tick skips it.
    assert!(
        webhook_events::get_retryable_events(&db.pool)
            .await
            .unwrap()
            .is_empty(),
        "a just-rescheduled event is not yet due"
    );
}

#[tokio::test]
async fn schedule_retry_dead_letters_at_max_attempts() {
    let db = TestDb::new().await;
    let id = seed_due_event(&db.pool, "dead.evt").await;

    // Reaching the max attempts marks the event dead, off the retry queue,
    // onto the operator dead-letter list.
    webhook_events::schedule_retry(&db.pool, id, 5, "exhausted")
        .await
        .unwrap();

    let status: String = sqlx::query_scalar("SELECT status FROM webhook_events WHERE id = $1")
        .bind(id)
        .fetch_one(&db.pool)
        .await
        .unwrap();
    assert_eq!(status, "dead");

    assert!(
        webhook_events::get_retryable_events(&db.pool)
            .await
            .unwrap()
            .is_empty(),
        "a dead event must never be re-claimed for retry"
    );
    let dead = webhook_events::get_dead_events(&db.pool).await.unwrap();
    assert!(
        dead.iter().any(|e| e.id == id),
        "a dead event surfaces on the operator list"
    );
}

#[tokio::test]
async fn retry_dead_event_resurrects_only_dead_rows() {
    let db = TestDb::new().await;

    // A live (failed) event is not a dead-letter, so resetting it is a no-op.
    let live = seed_due_event(&db.pool, "resurrect.live").await;
    assert!(
        !webhook_events::retry_dead_event(&db.pool, live)
            .await
            .unwrap(),
        "retry_dead_event must only act on status = 'dead'"
    );

    // A genuinely dead event flips back to 'failed' and becomes claimable again.
    let dead = seed_due_event(&db.pool, "resurrect.dead").await;
    webhook_events::schedule_retry(&db.pool, dead, 5, "exhausted")
        .await
        .unwrap();
    assert!(
        webhook_events::retry_dead_event(&db.pool, dead)
            .await
            .unwrap()
    );

    let status: String = sqlx::query_scalar("SELECT status FROM webhook_events WHERE id = $1")
        .bind(dead)
        .fetch_one(&db.pool)
        .await
        .unwrap();
    assert_eq!(
        status, "failed",
        "a resurrected event returns to the retry queue"
    );
}

#[tokio::test]
async fn mark_processed_removes_event_from_queue() {
    let db = TestDb::new().await;
    let id = seed_due_event(&db.pool, "done.evt").await;

    webhook_events::mark_processed(&db.pool, id).await.unwrap();

    let remaining: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM webhook_events WHERE id = $1")
        .bind(id)
        .fetch_one(&db.pool)
        .await
        .unwrap();
    assert_eq!(
        remaining, 0,
        "a successfully processed event is deleted from the queue"
    );
}