//! Negative paths: what the server does when object storage fails.
//!
//! Negative paths are split by the dependency that fails, which is also how you
//! look these up: anything about S3 being down lands here.
//!
//! 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 makenotwork::storage::StorageBackend;

// The durable S3 deletion queue

/// Count rows still queued for deletion of `key`.
async fn queued_deletions(h: &TestHarness, key: &str) -> i64 {
    sqlx::query_scalar("SELECT COUNT(*) FROM pending_s3_deletions WHERE s3_key = $1")
        .bind(key)
        .fetch_one(&h.db)
        .await
        .unwrap()
}

/// A delete that fails must leave the row queued. Dequeuing it would orphan the
/// S3 object with no durable record, which is the leak the queue exists to
/// prevent.
#[tokio::test]
async fn s3_delete_failure_keeps_the_row_queued_for_retry() {
    let h = TestHarness::with_storage().await;
    let storage = h.storage.clone().expect("with_storage provides a backend");
    let key = "test/orphan-retry.bin";

    storage.put(key, b"payload".to_vec());
    db::pending_s3_deletions::enqueue_deletions(
        &h.db,
        &[(key.to_string(), "main".to_string())],
        "test_failure_path",
    )
    .await
    .unwrap();
    assert_eq!(queued_deletions(&h, key).await, 1, "row starts queued");

    storage
        .faults()
        .fail_always("delete_object", storage_unavailable);
    let deleted = h.drain_s3_deletions().await;

    assert_eq!(deleted, 0, "a failing backend deletes nothing");
    assert_eq!(
        queued_deletions(&h, key).await,
        1,
        "the row must survive a failed delete, dropping it would orphan the object"
    );
    assert!(
        storage.object_exists(key).await.unwrap(),
        "the object is still there, which is why the row must be"
    );
    assert_eq!(
        storage.faults().calls("delete_object"),
        1,
        "the drain attempted the delete exactly once"
    );
}

/// The point of keeping the row: a later drain finishes the job. This is the
/// whole contract of the durable queue and nothing asserted it before.
#[tokio::test]
async fn s3_delete_queue_recovers_when_the_backend_comes_back() {
    let h = TestHarness::with_storage().await;
    let storage = h.storage.clone().expect("with_storage provides a backend");
    let key = "test/orphan-recovers.bin";

    storage.put(key, b"payload".to_vec());
    db::pending_s3_deletions::enqueue_deletions(
        &h.db,
        &[(key.to_string(), "main".to_string())],
        "test_failure_path",
    )
    .await
    .unwrap();

    // Down for the first attempt, up for the second.
    storage
        .faults()
        .fail_until("delete_object", 2, storage_unavailable);

    assert_eq!(h.drain_s3_deletions().await, 0, "first drain fails");
    assert_eq!(queued_deletions(&h, key).await, 1, "still queued");

    assert_eq!(h.drain_s3_deletions().await, 1, "second drain succeeds");
    assert_eq!(
        queued_deletions(&h, key).await,
        0,
        "a completed delete is dequeued"
    );
    assert!(
        !storage.object_exists(key).await.unwrap(),
        "the object is gone"
    );
}

// The orphaned-upload reaper

/// Insert a pending upload that is already old enough for the reaper, with the
/// object present in storage. Returns the key.
async fn stale_pending_upload(h: &TestHarness, user_id: db::UserId, key: &str) -> String {
    h.storage.as_ref().unwrap().put(key, b"orphan".to_vec());
    sqlx::query(
        "INSERT INTO pending_uploads (user_id, s3_key, bucket, created_at)
         VALUES ($1, $2, 'main', NOW() - INTERVAL '48 hours')",
    )
    .bind(user_id)
    .bind(key)
    .execute(&h.db)
    .await
    .unwrap();
    key.to_string()
}

async fn pending_upload_rows(h: &TestHarness, key: &str) -> i64 {
    sqlx::query_scalar("SELECT COUNT(*) FROM pending_uploads WHERE s3_key = $1")
        .bind(key)
        .fetch_one(&h.db)
        .await
        .unwrap()
}

/// The happy path, asserted here so the failure path below is a contrast rather
/// than the only thing observed: a reaped orphan is deleted, its tracking row is
/// cleared, and nothing is handed to the durable queue.
#[tokio::test]
async fn the_reaper_deletes_an_orphan_and_clears_its_row() {
    let mut h = TestHarness::with_storage().await;
    let user_id = h.signup("reap1", "reap1@test.com", "pass1234").await;
    let key = stale_pending_upload(&h, user_id, "staging/reaped.bin").await;
    let storage = h.storage.clone().unwrap();

    h.run_orphan_upload_reaper().await;

    assert!(
        !storage.object_exists(&key).await.unwrap(),
        "the orphan object is deleted"
    );
    assert_eq!(
        pending_upload_rows(&h, &key).await,
        0,
        "tracking row cleared"
    );
    assert_eq!(
        queued_deletions(&h, &key).await,
        0,
        "a successful delete must not also enqueue, that would double-handle the key"
    );
}

/// A transient S3 failure must hand the key to the durable deletion queue
/// BEFORE the tracking row is cleared. Clearing the row on a transient failure
/// drops the only record of the object and leaks it permanently.
#[tokio::test]
async fn a_transient_delete_failure_hands_the_orphan_to_the_durable_queue() {
    let mut h = TestHarness::with_storage().await;
    let user_id = h.signup("reap2", "reap2@test.com", "pass1234").await;
    let key = stale_pending_upload(&h, user_id, "staging/handed-off.bin").await;
    let storage = h.storage.clone().unwrap();

    storage
        .faults()
        .fail_always("delete_object", storage_unavailable);
    h.run_orphan_upload_reaper().await;

    assert!(
        storage.object_exists(&key).await.unwrap(),
        "the delete failed, so the object is still there"
    );
    assert_eq!(
        queued_deletions(&h, &key).await,
        1,
        "the key must be queued for retry; without this the object leaks"
    );
    assert_eq!(
        pending_upload_rows(&h, &key).await,
        0,
        "the tracking row is cleared only because the durable queue now owns the key"
    );

    // The handoff is worth nothing if the queue cannot then finish the job.
    storage.faults().clear("delete_object");
    assert_eq!(h.drain_s3_deletions().await, 1, "the retry completes it");
    assert!(!storage.object_exists(&key).await.unwrap(), "object gone");
}

/// Aborting orphaned multipart sessions is documented best-effort: it must not
/// block the object delete. A failing abort that stranded the delete would leave
/// the orphan in place every tick forever, and the tracking row with it.
#[tokio::test]
async fn a_failed_multipart_abort_does_not_block_the_orphan_delete() {
    let mut h = TestHarness::with_storage().await;
    let user_id = h.signup("reap3", "reap3@test.com", "pass1234").await;
    let key = stale_pending_upload(&h, user_id, "staging/abort-fails.bin").await;
    let storage = h.storage.clone().unwrap();

    storage
        .faults()
        .fail_always("list_multipart_uploads_for_key", storage_unavailable);
    h.run_orphan_upload_reaper().await;

    assert!(
        !storage.object_exists(&key).await.unwrap(),
        "a failed abort is best-effort and must not stop the delete"
    );
    assert_eq!(
        pending_upload_rows(&h, &key).await,
        0,
        "and the tracking row is still cleared"
    );
}