//! DB-layer contract tests for `db/scanning.rs`: the quarantine purge, the
//! read-side status stamp, and their symmetry across every CDN-served image
//! surface.
//!
//! Audit A5 / B-Sto1: these functions had no direct coverage (only indirect
//! exercise through the end-to-end scan worker), and the purge silently omitted
//! the `projects` cover, so a quarantined project cover stayed rendered AND its
//! (public-bucket) object stayed un-reapable behind the `is_s3_key_live` guard,
//! while the symmetric stamp DID touch projects. These pin that the purge and
//! the stamp cover the SAME five surfaces, and that `quarantined_s3_keys`
//! filters by verdict.

use crate::harness::TestHarness;
use makenotwork::db::FileScanStatus;
use makenotwork::db::scanning;

/// Seed a project + item cover and one gallery image per parent, all pointing at
/// distinct known keys. Returns the (project_id, item_id) as strings.
async fn seed_cdn_image_surfaces(h: &mut TestHarness) -> (String, String) {
    let setup = h.create_creator_with_item("scanlayer", "digital", 0).await;

    // Item + project covers.
    sqlx::query("UPDATE items SET cover_s3_key = 'k/itemcover', cover_image_url = 'https://cdn/x/itemcover' WHERE id = $1::uuid")
        .bind(&setup.item_id)
        .execute(&h.db)
        .await
        .unwrap();
    sqlx::query("UPDATE projects SET cover_s3_key = 'k/projcover', cover_image_url = 'https://cdn/x/projcover' WHERE id = $1::uuid")
        .bind(&setup.project_id)
        .execute(&h.db)
        .await
        .unwrap();

    // Gallery images (item_images / project_images).
    sqlx::query(
        "INSERT INTO item_images (item_id, s3_key, image_url, alt, position, file_size_bytes) \
         VALUES ($1::uuid, 'k/itemimg', 'https://cdn/x/itemimg', '', 0, 10)",
    )
    .bind(&setup.item_id)
    .execute(&h.db)
    .await
    .unwrap();
    sqlx::query(
        "INSERT INTO project_images (project_id, s3_key, image_url, alt, position, file_size_bytes) \
         VALUES ($1::uuid, 'k/projimg', 'https://cdn/x/projimg', '', 0, 10)",
    )
    .bind(&setup.project_id)
    .execute(&h.db)
    .await
    .unwrap();

    (setup.project_id, setup.item_id)
}

/// Whether any live row still references `key` across the CDN image surfaces.
async fn key_still_referenced(h: &TestHarness, key: &str) -> bool {
    let n: i64 = sqlx::query_scalar(
        "SELECT \
           (SELECT COUNT(*) FROM item_images WHERE s3_key = $1) \
         + (SELECT COUNT(*) FROM project_images WHERE s3_key = $1) \
         + (SELECT COUNT(*) FROM items WHERE cover_s3_key = $1) \
         + (SELECT COUNT(*) FROM projects WHERE cover_s3_key = $1)",
    )
    .bind(key)
    .fetch_one(&h.db)
    .await
    .unwrap();
    n > 0
}

#[tokio::test]
async fn purge_clears_every_cdn_image_surface_including_project_cover() {
    let mut h = TestHarness::new().await;
    let _ = seed_cdn_image_surfaces(&mut h).await;

    // Each surface's key must be fully de-referenced by the purge, otherwise the
    // quarantined object stays rendered and un-reapable.
    for key in ["k/itemcover", "k/projcover", "k/itemimg", "k/projimg"] {
        assert!(
            key_still_referenced(&h, key).await,
            "precondition: {key} is referenced"
        );
        let removed = scanning::purge_cdn_image_rows_by_key(&h.db, key)
            .await
            .unwrap();
        assert_eq!(
            removed, 1,
            "purge of {key} should clear exactly one surface"
        );
        assert!(
            !key_still_referenced(&h, key).await,
            "purge must fully de-reference {key}"
        );
    }
}

#[tokio::test]
async fn purge_keeps_the_item_and_project_rows() {
    let mut h = TestHarness::new().await;
    let (project_id, item_id) = seed_cdn_image_surfaces(&mut h).await;

    scanning::purge_cdn_image_rows_by_key(&h.db, "k/itemcover")
        .await
        .unwrap();
    scanning::purge_cdn_image_rows_by_key(&h.db, "k/projcover")
        .await
        .unwrap();

    // Purging a quarantined cover must NULL the columns, never delete the row.
    let item_exists: bool = sqlx::query_scalar(
        "SELECT EXISTS(SELECT 1 FROM items WHERE id = $1::uuid AND cover_s3_key IS NULL)",
    )
    .bind(&item_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert!(
        item_exists,
        "item row must survive cover purge with a NULL cover"
    );
    let project_exists: bool = sqlx::query_scalar(
        "SELECT EXISTS(SELECT 1 FROM projects WHERE id = $1::uuid AND cover_s3_key IS NULL)",
    )
    .bind(&project_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert!(
        project_exists,
        "project row must survive cover purge with a NULL cover"
    );
}

#[tokio::test]
async fn stamp_and_purge_cover_the_same_surfaces() {
    let mut h = TestHarness::new().await;
    let _ = seed_cdn_image_surfaces(&mut h).await;

    // The read-side stamp must touch each surface the purge touches (symmetry):
    // one row affected per key, for all four.
    for key in ["k/itemcover", "k/projcover", "k/itemimg", "k/projimg"] {
        let stamped = scanning::set_cdn_image_scan_status_by_key(&h.db, key, FileScanStatus::Clean)
            .await
            .unwrap();
        assert_eq!(
            stamped, 1,
            "stamp of {key} should affect exactly one surface"
        );
    }
}

#[tokio::test]
async fn quarantined_s3_keys_filters_by_verdict() {
    let mut h = TestHarness::new().await;
    let _ = seed_cdn_image_surfaces(&mut h).await;

    // Two scan results: one quarantined, one clean.
    for (key, status) in [("k/itemcover", "quarantined"), ("k/itemimg", "clean")] {
        sqlx::query(
            "INSERT INTO file_scan_results (s3_key, scan_status, scan_layers) \
             VALUES ($1, $2, '[]'::jsonb)",
        )
        .bind(key)
        .bind(status)
        .execute(&h.db)
        .await
        .unwrap();
    }

    let quarantined = scanning::quarantined_s3_keys(
        &h.db,
        &[
            "k/itemcover".to_string(),
            "k/itemimg".to_string(),
            "k/never-scanned".to_string(),
        ],
    )
    .await
    .unwrap();

    assert!(
        quarantined.contains("k/itemcover"),
        "quarantined key must be returned"
    );
    assert!(
        !quarantined.contains("k/itemimg"),
        "clean key must not be returned"
    );
    assert!(
        !quarantined.contains("k/never-scanned"),
        "unscanned key must not be returned"
    );
    assert_eq!(quarantined.len(), 1);
}