//! DB-layer contract tests for `db::pending_uploads`, the tracking table that
//! decides whether an abandoned upload's object is reclaimed or leaked.
//!
//! Every function in the module is `pub(crate)`, so its reclaim/expiry half is
//! reached the way production reaches it, through the orphan-upload reaper
//! (`TestHarness::run_orphan_upload_reaper`), which calls
//! `get_stale_pending_uploads` and `delete_pending_uploads` and nothing else.
//! Those two carry the retry machinery this table exists for, and they are what
//! the assertions below are about: the age cutoff on both sides, oldest-first
//! draining under the `STALE_UPLOAD_BATCH` cap, the `(s3_key, bucket)` pair
//! scoping on the delete, and the reclaim case where a live row took the key
//! back and the object must survive while the tracking row goes.
//!
//! Not asserted here, and deliberately not claimed: `record_pending_upload`,
//! `record_pending_multipart_upload`, `declared_size`, `touch_pending_upload`,
//! `remove_pending_upload` and `is_owned`, which an integration test can only
//! reach through the presign handlers.
//!
//! Delete this file and a reaper that either deletes a reclaimed object or
//! leaves an orphan behind forever passes silently.

use crate::harness::{TestHarness, seed_project};
use makenotwork::db::items::update_item_file_cas;
use makenotwork::db::{ItemId, ProjectId, UserId};
use makenotwork::storage::{FileType, StorageBackend};
// ── helpers ──────────────────────────────────────────────────────────────────

async fn seed_item(pool: &sqlx::PgPool, project: ProjectId, slug: &str) -> ItemId {
    sqlx::query_scalar::<_, ItemId>(
        "INSERT INTO items (project_id, title, item_type, price_cents, slug)
         VALUES ($1, $2, 'digital', 1000, $3) RETURNING id",
    )
    .bind(project)
    .bind(format!("Item {slug}"))
    .bind(slug)
    .fetch_one(pool)
    .await
    .expect("seed item")
}

/// The audio pair as stored.
async fn audio_cols(pool: &sqlx::PgPool, item: ItemId) -> (Option<String>, Option<i64>) {
    sqlx::query_as("SELECT audio_s3_key, audio_file_size_bytes FROM items WHERE id = $1")
        .bind(item)
        .fetch_one(pool)
        .await
        .expect("read audio columns")
}

// ── db::pending_uploads: the expiry and reclaim half ─────────────────────────
//
// Reached through the reaper, which is the only production caller of
// `get_stale_pending_uploads` and `delete_pending_uploads`.

/// Insert a pending upload aged `hours` old, with its object staged in storage.
async fn pending_upload(h: &TestHarness, user: UserId, key: &str, bucket: &str, hours: i32) {
    h.storage
        .as_ref()
        .expect("with_storage provides a backend")
        .put(key, b"orphan".to_vec());
    sqlx::query(
        "INSERT INTO pending_uploads (user_id, s3_key, bucket, created_at)
         VALUES ($1, $2, $3, NOW() - make_interval(hours => $4))",
    )
    .bind(user)
    .bind(key)
    .bind(bucket)
    .bind(hours)
    .execute(&h.db)
    .await
    .expect("insert pending upload");
}

async fn pending_rows(h: &TestHarness, key: &str, bucket: &str) -> i64 {
    sqlx::query_scalar("SELECT COUNT(*) FROM pending_uploads WHERE s3_key = $1 AND bucket = $2")
        .bind(key)
        .bind(bucket)
        .fetch_one(&h.db)
        .await
        .expect("count pending rows")
}

/// The cutoff is 24 hours and it is a strict age test, so both sides of it are
/// asserted in one run: a 25-hour-old upload is reaped and a 23-hour-old one is
/// left completely alone. A test that only checked the old row could not tell
/// "older than 24h" from "every row".
#[tokio::test]
async fn the_reaper_takes_uploads_past_the_cutoff_and_leaves_younger_ones_untouched() {
    let mut h = TestHarness::with_storage().await;
    let user = h
        .signup("reaper_cutoff", "reaper_cutoff@test.com", "pass1234")
        .await;
    let storage = h.storage.clone().expect("with_storage provides a backend");

    pending_upload(&h, user, "staging/too-old.bin", "main", 25).await;
    pending_upload(&h, user, "staging/still-young.bin", "main", 23).await;

    h.run_orphan_upload_reaper().await;

    assert_eq!(
        pending_rows(&h, "staging/too-old.bin", "main").await,
        0,
        "an upload older than the 24h cutoff must lose its tracking row"
    );
    assert!(
        !storage.object_exists("staging/too-old.bin").await.unwrap(),
        "and its object must be deleted"
    );

    assert_eq!(
        pending_rows(&h, "staging/still-young.bin", "main").await,
        1,
        "an upload younger than the cutoff is still in flight and must be kept"
    );
    assert!(
        storage
            .object_exists("staging/still-young.bin")
            .await
            .unwrap(),
        "deleting a young upload's object would destroy a transfer in progress"
    );
}

/// The reaper is capped at `STALE_UPLOAD_BATCH` (200) rows per tick and drains
/// oldest first, so a backlog cannot wedge the tick. 201 rows put the cap on both
/// sides in one run: exactly one row survives, and it is the youngest, which is
/// what distinguishes oldest-first from an unordered or newest-first scan.
#[tokio::test]
async fn the_reaper_drains_the_oldest_two_hundred_and_leaves_the_rest_for_the_next_tick() {
    let mut h = TestHarness::with_storage().await;
    let user = h
        .signup("reaper_batch", "reaper_batch@test.com", "pass1234")
        .await;
    let storage = h.storage.clone().expect("with_storage provides a backend");

    // 201 stale rows, ages 48h down to about 44h40m, so the ordering is total and
    // every one of them is past the 24h cutoff.
    for i in 0..201i32 {
        let key = format!("staging/batch-{i:03}.bin");
        storage.put(&key, b"orphan".to_vec());
    }
    sqlx::query(
        "INSERT INTO pending_uploads (user_id, s3_key, bucket, created_at)
         SELECT $1, 'staging/batch-' || to_char(i, 'FM000') || '.bin', 'main',
                NOW() - INTERVAL '48 hours' + make_interval(mins => i)
         FROM generate_series(0, 200) AS i",
    )
    .bind(user)
    .execute(&h.db)
    .await
    .expect("insert 201 stale uploads");

    h.run_orphan_upload_reaper().await;

    let remaining: Vec<String> =
        sqlx::query_scalar("SELECT s3_key FROM pending_uploads ORDER BY s3_key")
            .fetch_all(&h.db)
            .await
            .expect("read remaining rows");
    assert_eq!(
        remaining,
        vec!["staging/batch-200.bin".to_string()],
        "exactly the youngest row is left for the next tick, got {remaining:?}"
    );
    assert!(
        storage
            .object_exists("staging/batch-200.bin")
            .await
            .unwrap(),
        "the row that was not reaped keeps its object"
    );
    assert!(
        !storage
            .object_exists("staging/batch-000.bin")
            .await
            .unwrap(),
        "the oldest row is the first one drained"
    );

    // The next tick finishes the backlog, which is what makes the cap a drain
    // rather than a permanent leak.
    h.run_orphan_upload_reaper().await;
    let left: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pending_uploads")
        .fetch_one(&h.db)
        .await
        .expect("count remaining");
    assert_eq!(left, 0, "the second tick drains what the cap held back");
}

/// The uniqueness key is `(s3_key, bucket)`, so the same key can be pending in
/// two buckets at once and the delete must match the pair. Only the main-bucket
/// row is stale here: a delete keyed on `s3_key` alone would take the synckit
/// row with it and the reaper would forget an upload that is still in flight.
#[tokio::test]
async fn reaping_one_bucket_leaves_another_buckets_row_for_the_same_key() {
    let mut h = TestHarness::with_storage().await;
    let user = h
        .signup("reaper_bucket", "reaper_bucket@test.com", "pass1234")
        .await;

    pending_upload(&h, user, "staging/shared-key.bin", "main", 30).await;
    // Young, and in a different bucket: not eligible for this reap at all.
    sqlx::query(
        "INSERT INTO pending_uploads (user_id, s3_key, bucket, created_at)
         VALUES ($1, 'staging/shared-key.bin', 'synckit', NOW())",
    )
    .bind(user)
    .execute(&h.db)
    .await
    .expect("insert synckit pending row");

    h.run_orphan_upload_reaper().await;

    assert_eq!(
        pending_rows(&h, "staging/shared-key.bin", "main").await,
        0,
        "the stale main-bucket row is reaped"
    );
    assert_eq!(
        pending_rows(&h, "staging/shared-key.bin", "synckit").await,
        1,
        "the synckit row shares the key but not the bucket and must survive"
    );
}

/// The reclaim path, which is why the reaper routes through the live-key guard.
/// A confirm can take a staging key over onto a live item row before the reaper
/// reaches the stale pending record. The object now belongs to that item, so it
/// must survive; only the tracking row goes.
#[tokio::test]
async fn a_key_reclaimed_by_a_live_item_keeps_its_object_and_only_loses_the_pending_row() {
    let mut h = TestHarness::with_storage().await;
    let user = h
        .signup("reaper_reclaim", "reaper_reclaim@test.com", "pass1234")
        .await;
    let storage = h.storage.clone().expect("with_storage provides a backend");

    let project = seed_project(&h.db, user, "reclaim-proj").await;
    let item = seed_item(&h.db, project, "reclaim-item").await;
    pending_upload(&h, user, "staging/reclaimed.bin", "main", 40).await;

    // The confirm that beat the reaper: the key is now the item's live audio.
    update_item_file_cas(
        &h.db,
        item,
        user,
        FileType::Audio,
        None,
        "staging/reclaimed.bin",
        3_300_000,
    )
    .await
    .expect("confirm reclaims the key");

    h.run_orphan_upload_reaper().await;

    assert!(
        storage
            .object_exists("staging/reclaimed.bin")
            .await
            .unwrap(),
        "the object is a live item file now; deleting it would destroy a paid-for upload"
    );
    assert_eq!(
        pending_rows(&h, "staging/reclaimed.bin", "main").await,
        0,
        "the stale tracking row is still cleared, or the reaper retries it forever"
    );
    let queued: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM pending_s3_deletions WHERE s3_key = $1")
            .bind("staging/reclaimed.bin")
            .fetch_one(&h.db)
            .await
            .expect("count queued deletions");
    assert_eq!(
        queued, 0,
        "a live key must not be handed to the deletion queue either"
    );
    let audio = audio_cols(&h.db, item).await;
    assert_eq!(
        audio,
        (Some("staging/reclaimed.bin".to_string()), Some(3_300_000)),
        "and the item still points at it, got {audio:?}"
    );
}

/// Running the reaper twice over the same backlog must delete each object once.
/// The tracking rows are gone after the first pass, so the second finds nothing:
/// this is the replay guard on the reap itself.
#[tokio::test]
async fn a_second_reaper_pass_over_a_drained_backlog_deletes_nothing_again() {
    let mut h = TestHarness::with_storage().await;
    let user = h
        .signup("reaper_twice", "reaper_twice@test.com", "pass1234")
        .await;
    let storage = h.storage.clone().expect("with_storage provides a backend");

    pending_upload(&h, user, "staging/once-a.bin", "main", 26).await;
    pending_upload(&h, user, "staging/once-b.bin", "main", 27).await;

    h.run_orphan_upload_reaper().await;
    let after_first = storage.faults().calls("delete_object");
    assert_eq!(
        after_first, 2,
        "the first pass deletes each of the two orphans exactly once"
    );

    h.run_orphan_upload_reaper().await;
    assert_eq!(
        storage.faults().calls("delete_object"),
        after_first,
        "the second pass has no rows to act on and must issue no further deletes"
    );
    let left: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pending_uploads")
        .fetch_one(&h.db)
        .await
        .expect("count remaining");
    assert_eq!(left, 0, "nothing is left pending");
}