//! Negative paths: what the server does when the scanner cannot fetch bytes.
//!
//! Negative paths are split by the dependency that fails, which is also how you
//! look these
//! up: the scan-job retry budget and its parking behaviour.
//!
//! These tests exist because an infallible mock leaves the retry and
//! compensation machinery the server carries with no test that can reach it.
//! Retry logic no test can enter is worse than none, because it reads as
//! handled. Each test installs a failure policy on a mock (see
//! `harness::faults`) and asserts the compensating behaviour, not just that the
//! request failed.
//!
//! Rationale: wiki `testing-posture`, the "absent oracle" section.

use crate::harness::TestHarness;
use crate::harness::faults::storage_unavailable;
use makenotwork::db;
use serde_json::Value;

// The scan-job retry budget

/// Set up a trusted creator with an audio item, presign an upload, put the
/// bytes, and confirm it, leaving exactly one queued scan job. Returns the item
/// id and the staging key the job will try to download.
async fn queue_one_scan_job(h: &mut TestHarness) -> (String, String) {
    let setup = h.create_creator_with_item("fpscan", "audio", 0).await;
    h.trust_user(setup.user_id).await;
    h.grant_tier(setup.user_id, "small_files").await;

    let body = serde_json::json!({
        "item_id": setup.item_id,
        "file_type": "audio",
        "file_name": "held.mp3",
        "content_type": "audio/mpeg",
    });
    let resp = h
        .client
        .post_json("/api/upload/presign", &body.to_string())
        .await;
    assert_eq!(resp.status, 200, "presign failed: {}", resp.text);
    let s3_key = resp.json::<Value>()["s3_key"]
        .as_str()
        .expect("presign returns s3_key")
        .to_string();

    let mut mp3 = b"ID3".to_vec();
    mp3.extend_from_slice(&[0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]);
    mp3.extend_from_slice(&[0u8; 100]);
    h.storage.as_ref().unwrap().put(&s3_key, mp3);

    let body = serde_json::json!({
        "item_id": setup.item_id,
        "file_type": "audio",
        "s3_key": s3_key,
    });
    let resp = h
        .client
        .post_json("/api/upload/confirm", &body.to_string())
        .await;
    assert_eq!(resp.status, 200, "confirm failed: {}", resp.text);

    (setup.item_id, s3_key)
}

async fn job_row(h: &TestHarness, item_id: &str) -> (String, i32, Option<String>) {
    sqlx::query_as("SELECT status, attempts, last_error FROM scan_jobs WHERE target_id = $1::uuid")
        .bind(item_id)
        .fetch_one(&h.db)
        .await
        .unwrap()
}

/// A scan whose download fails must record the failure and park the entity at
/// `held_for_review`. Leaving it at `scanning` is the production regression the
/// reset in `process_job` exists to prevent: the file is invisible to the buyer
/// and invisible to the admin queue, so nothing ever resolves it.
#[tokio::test]
async fn scan_download_failure_marks_the_job_failed_and_holds_the_entity() {
    let mut h = TestHarness::with_storage_and_scanner().await;
    let (item_id, _key) = queue_one_scan_job(&mut h).await;
    let storage = h.storage.clone().expect("scanner harness provides storage");

    // Both scanner read paths (`download_object_buf_capped` for small files,
    // `download_stream` for spooled ones) bottom out in `download_stream`, so
    // one rule covers the branch either size takes.
    storage
        .faults()
        .fail_always("download_stream", storage_unavailable);

    let err = h
        .try_process_one_scan_job()
        .await
        .expect_err("a failing download must surface as a job error");

    let (status, attempts, last_error) = job_row(&h, &item_id).await;
    assert_eq!(status, "failed", "the job records its own failure");
    assert_eq!(attempts, 1, "the claim consumed exactly one attempt");
    assert!(
        last_error.is_some_and(|e| !e.is_empty()),
        "last_error is what an admin has to work from"
    );

    let scan_status: String =
        sqlx::query_scalar("SELECT scan_status FROM items WHERE id = $1::uuid")
            .bind(&item_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        scan_status, "held_for_review",
        "a failed scan must not leave the entity stuck at 'scanning'"
    );
    assert!(
        err.contains("S3") || err.contains("torage"),
        "the error should name the failing dependency, got: {err}"
    );
}

/// A worker that dies mid-scan leaves its row `running` forever; `reap_stuck` is
/// what returns it to the queue. Below the attempt ceiling that is a requeue,
/// and the retry then succeeds once storage is back. Nothing asserted the
/// recovery half before, which is the half the budget exists for.
#[tokio::test]
async fn a_reaped_scan_job_is_requeued_and_succeeds_when_storage_recovers() {
    let mut h = TestHarness::with_storage_and_scanner().await;
    let (item_id, _key) = queue_one_scan_job(&mut h).await;
    let storage = h.storage.clone().expect("scanner harness provides storage");

    // Claim the job the way a worker would, then abandon it: no mark_done, no
    // mark_failed, exactly what a killed process leaves behind.
    let job = db::scan_jobs::claim_next(&h.db)
        .await
        .unwrap()
        .expect("the confirm queued a job");
    sqlx::query("UPDATE scan_jobs SET heartbeat_at = NOW() - INTERVAL '1 hour' WHERE id = $1")
        .bind(job.id)
        .execute(&h.db)
        .await
        .unwrap();

    let reaped = db::scan_jobs::reap_stuck(&h.db, 60).await.unwrap();
    assert_eq!(reaped, 1, "the stale heartbeat is what the reaper keys on");

    let (status, attempts, _) = job_row(&h, &item_id).await;
    assert_eq!(
        status, "queued",
        "below the ceiling a reaped job goes back to the queue, not to failed"
    );
    assert_eq!(attempts, 1, "the abandoned attempt is still spent");

    // Storage is healthy again; the retry must complete the job.
    assert!(
        storage.faults().calls("download_stream") == 0,
        "no fault installed, the first attempt never reached the backend"
    );
    h.drain_scan_jobs().await;

    let (status, attempts, _) = job_row(&h, &item_id).await;
    assert_eq!(status, "done", "the retry completes the job");
    assert_eq!(attempts, 2, "the retry consumed a second attempt");
}

/// The ceiling is what stops a job that reliably kills its worker from being
/// re-attempted forever. At `MAX_SCAN_ATTEMPTS` the reaper retires the row to
/// `failed` rather than requeueing it, and `claim_next` will not hand it out
/// again.
#[tokio::test]
async fn a_scan_job_at_its_attempt_ceiling_is_retired_not_requeued() {
    let mut h = TestHarness::with_storage_and_scanner().await;
    let (item_id, _key) = queue_one_scan_job(&mut h).await;

    // Spend the budget down to its last attempt, then claim, which takes it.
    sqlx::query("UPDATE scan_jobs SET attempts = $1 WHERE target_id = $2::uuid")
        .bind(db::scan_jobs::MAX_SCAN_ATTEMPTS - 1)
        .bind(&item_id)
        .execute(&h.db)
        .await
        .unwrap();
    let job = db::scan_jobs::claim_next(&h.db)
        .await
        .unwrap()
        .expect("a job one under the ceiling is still claimable");
    assert_eq!(job.attempts, db::scan_jobs::MAX_SCAN_ATTEMPTS);

    sqlx::query("UPDATE scan_jobs SET heartbeat_at = NOW() - INTERVAL '1 hour' WHERE id = $1")
        .bind(job.id)
        .execute(&h.db)
        .await
        .unwrap();
    assert_eq!(db::scan_jobs::reap_stuck(&h.db, 60).await.unwrap(), 1);

    let (status, _, last_error) = job_row(&h, &item_id).await;
    assert_eq!(
        status, "failed",
        "at the ceiling the reaper retires the job instead of requeueing it"
    );
    assert!(
        last_error.is_some_and(|e| e.contains("max scan attempts")),
        "the retirement reason must be legible to an admin"
    );

    assert!(
        db::scan_jobs::claim_next(&h.db).await.unwrap().is_none(),
        "a retired job must never be claimed again"
    );
}