//! DB-layer contract tests for `db::items::media`, the writeback layer every
//! upload confirm lands on.
//!
//! This is the compare-and-swap that decides whether storage may be credited,
//! the cover triple-write (url + key + size), and the ownership-filtered
//! size/metadata writebacks. These pin what each one actually promises: which
//! outcome a stale expectation produces, that a losing write leaves every
//! column untouched, that each file type writes its own column pair and no
//! other, and that a non-owner never lands a byte.
//!
//! The pending-uploads table that used to share this file has its own
//! module, db_pending_uploads_layer.
//!
//! Delete this file and two classes of loss stop being observable: a dropped
//! CAS predicate (double-credited storage and clobbered live objects), and a
//! size or metadata writeback that ignores its ownership filter.

use crate::harness::db::TestDb;
use crate::harness::{seed_project, seed_user};
use makenotwork::db::items::{FileConfirmOutcome, update_item_file_cas};
use makenotwork::db::{ItemId, ProjectId, UserId, items};
use makenotwork::error::AppError;
use makenotwork::storage::FileType;

// ── 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")
}

/// A user, a project and one item, the minimum an item-media writeback needs.
async fn owner_and_item(db: &TestDb, tag: &str) -> (UserId, ItemId) {
    let user = seed_user(&db.pool, &format!("media_{tag}")).await;
    let project = seed_project(&db.pool, user, &format!("media-{tag}")).await;
    let item = seed_item(&db.pool, project, &format!("i-{tag}")).await;
    (user, 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")
}

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

/// The cover triple as stored.
#[allow(clippy::type_complexity)]
async fn cover_cols(
    pool: &sqlx::PgPool,
    item: ItemId,
) -> (Option<String>, Option<String>, Option<i64>) {
    sqlx::query_as(
        "SELECT cover_image_url, cover_s3_key, cover_file_size_bytes FROM items WHERE id = $1",
    )
    .bind(item)
    .fetch_one(pool)
    .await
    .expect("read cover columns")
}

// ── update_item_file_cas: the guarded write behind every storage credit ───────

/// The redelivery case. A confirm that is delivered twice observes the same
/// pre-state twice, so the second call carries the same `expected_old_key`. It
/// must report `LostRace` and leave the row exactly as the winner wrote it,
/// because the caller credits storage on `Committed` and would otherwise be
/// charged twice for one object.
#[tokio::test]
async fn a_replayed_file_confirm_reports_a_lost_race_and_credits_nothing_twice() {
    let db = TestDb::new().await;
    let (owner, item) = owner_and_item(&db, "replay").await;

    let first = update_item_file_cas(
        &db.pool,
        item,
        owner,
        FileType::Audio,
        None,
        "staging/first.mp3",
        7_340_032,
    )
    .await
    .expect("first confirm runs");
    assert_eq!(
        first,
        FileConfirmOutcome::Committed,
        "the first confirm against a NULL column must commit"
    );

    // The exact byte count matters: the caller credits this number against the
    // creator's quota, so an off-by-anything is a billing error.
    let after_first = audio_cols(&db.pool, item).await;
    assert_eq!(
        after_first,
        (Some("staging/first.mp3".to_string()), Some(7_340_032)),
        "committed confirm must store the exact key and size, got {after_first:?}"
    );

    // Same event delivered again: the column no longer holds NULL.
    let replay = update_item_file_cas(
        &db.pool,
        item,
        owner,
        FileType::Audio,
        None,
        "staging/first.mp3",
        7_340_032,
    )
    .await
    .expect("replayed confirm runs");
    assert_eq!(
        replay,
        FileConfirmOutcome::LostRace,
        "a redelivered confirm must lose the CAS, not commit a second time"
    );

    let after_replay = audio_cols(&db.pool, item).await;
    assert_eq!(
        after_replay, after_first,
        "the replay must leave the row byte-identical, got {after_replay:?}"
    );

    // A genuine replace observes the current key and swaps it. The new size
    // REPLACES the old one: 2_097_152, not 9_437_184 (a summing bug) and not
    // 7_340_032 (a write that never landed).
    let replace = update_item_file_cas(
        &db.pool,
        item,
        owner,
        FileType::Audio,
        Some("staging/first.mp3"),
        "staging/second.mp3",
        2_097_152,
    )
    .await
    .expect("replace confirm runs");
    assert_eq!(
        replace,
        FileConfirmOutcome::Committed,
        "a confirm carrying the current key must commit the swap"
    );
    let after_replace = audio_cols(&db.pool, item).await;
    assert_eq!(
        after_replace,
        (Some("staging/second.mp3".to_string()), Some(2_097_152)),
        "the replace must overwrite both columns, got {after_replace:?}"
    );
}

/// The ownership filter is part of the same predicate as the CAS, so a stranger
/// holding the correct expected key still writes nothing. Asserted separately
/// from the CAS because a regression could drop either half alone.
#[tokio::test]
async fn a_non_owner_file_confirm_loses_the_race_and_writes_nothing() {
    let db = TestDb::new().await;
    let (owner, item) = owner_and_item(&db, "owner").await;
    let stranger = seed_user(&db.pool, "media_stranger").await;

    update_item_file_cas(
        &db.pool,
        item,
        owner,
        FileType::Video,
        None,
        "staging/owned.mp4",
        4_500_000,
    )
    .await
    .expect("owner confirm runs");

    let outcome = update_item_file_cas(
        &db.pool,
        item,
        stranger,
        FileType::Video,
        Some("staging/owned.mp4"),
        "staging/stolen.mp4",
        99,
    )
    .await
    .expect("stranger confirm runs");
    assert_eq!(
        outcome,
        FileConfirmOutcome::LostRace,
        "a non-owner must not be able to swap another creator's file"
    );

    let cols = video_cols(&db.pool, item).await;
    assert_eq!(
        cols,
        (Some("staging/owned.mp4".to_string()), Some(4_500_000)),
        "the non-owner write must not land, got {cols:?}"
    );
}

/// Each file type owns exactly one column pair. A mapping that crossed audio and
/// video would still "work" for a single-file item, so both are written on one
/// item with distinct keys and distinct sizes and each pair is read back.
#[tokio::test]
async fn each_file_type_writes_only_its_own_column_pair() {
    let db = TestDb::new().await;
    let (owner, item) = owner_and_item(&db, "cols").await;

    update_item_file_cas(
        &db.pool,
        item,
        owner,
        FileType::Audio,
        None,
        "staging/a.mp3",
        5_000_000,
    )
    .await
    .expect("audio confirm runs");
    update_item_file_cas(
        &db.pool,
        item,
        owner,
        FileType::Video,
        None,
        "staging/v.mp4",
        3_000_000,
    )
    .await
    .expect("video confirm runs");

    let audio = audio_cols(&db.pool, item).await;
    let video = video_cols(&db.pool, item).await;
    assert_eq!(
        audio,
        (Some("staging/a.mp3".to_string()), Some(5_000_000)),
        "audio columns hold the audio confirm, got {audio:?}"
    );
    assert_eq!(
        video,
        (Some("staging/v.mp4".to_string()), Some(3_000_000)),
        "video columns hold the video confirm, got {video:?}"
    );
    // Neither generic confirm touches the cover triple.
    let cover = cover_cols(&db.pool, item).await;
    assert_eq!(
        cover,
        (None, None, None),
        "audio/video confirms must leave the cover columns alone, got {cover:?}"
    );
}

/// Types that need a third column (or another table) are refused rather than
/// half-written. The error names the route the caller should have used, which is
/// what makes the rejection actionable.
#[tokio::test]
async fn a_file_type_with_a_dedicated_route_is_refused_before_any_write() {
    let db = TestDb::new().await;
    let (owner, item) = owner_and_item(&db, "route").await;

    let err = update_item_file_cas(
        &db.pool,
        item,
        owner,
        FileType::Cover,
        None,
        "staging/cover.png",
        640_000,
    )
    .await
    .expect_err("a cover must not be confirmable through the generic writer");
    let message = err.to_string();
    match err {
        AppError::Internal(inner) => {
            let detail = inner.to_string();
            assert!(
                detail.contains("/api/items/image/confirm"),
                "the refusal must name the dedicated route, got {detail}"
            );
            assert!(
                detail.contains("cover"),
                "the refusal must name the offending file type, got {detail}"
            );
        }
        other => panic!("expected AppError::Internal, got {other:?} ({message})"),
    }

    let cover = cover_cols(&db.pool, item).await;
    assert_eq!(
        cover,
        (None, None, None),
        "a refused confirm must not half-write the row, got {cover:?}"
    );
}

/// The function takes any executor so the confirm and the storage credit share
/// one transaction. That is only worth anything if a rollback takes the file
/// writeback with it.
#[tokio::test]
async fn a_file_confirm_rolled_back_with_its_transaction_leaves_no_write() {
    let db = TestDb::new().await;
    let (owner, item) = owner_and_item(&db, "tx").await;

    let mut tx = db.pool.begin().await.expect("begin");
    let outcome = update_item_file_cas(
        &mut *tx,
        item,
        owner,
        FileType::Audio,
        None,
        "staging/rolled-back.mp3",
        8_800_000,
    )
    .await
    .expect("in-transaction confirm runs");
    assert_eq!(
        outcome,
        FileConfirmOutcome::Committed,
        "inside the transaction the CAS matches"
    );
    tx.rollback().await.expect("rollback");

    let cols = audio_cols(&db.pool, item).await;
    assert_eq!(
        cols,
        (None, None),
        "rolling back the storage credit must undo the file writeback too, got {cols:?}"
    );
}

// ── update_item_cover: the three-column write ────────────────────────────────

/// The cover write is the one that must move three columns together, and it
/// carries the same CAS as the audio/video path. A stale expectation must leave
/// all three as the winner left them: a partial write here shows a cover whose
/// url, key and size disagree.
#[tokio::test]
async fn a_cover_write_moves_url_key_and_size_together_and_guards_a_stale_key() {
    let db = TestDb::new().await;
    let (owner, item) = owner_and_item(&db, "cover").await;

    let first = items::update_item_cover(
        &db.pool,
        item,
        owner,
        None,
        "https://cdn.test/cover-one.png",
        "covers/one.png",
        640_000,
    )
    .await
    .expect("first cover write runs");
    assert!(first, "the first cover write against NULL must land");
    let after_first = cover_cols(&db.pool, item).await;
    assert_eq!(
        after_first,
        (
            Some("https://cdn.test/cover-one.png".to_string()),
            Some("covers/one.png".to_string()),
            Some(640_000)
        ),
        "all three cover columns must be written, got {after_first:?}"
    );

    // A second confirm that still believes the cover is unset loses.
    let stale = items::update_item_cover(
        &db.pool,
        item,
        owner,
        None,
        "https://cdn.test/cover-two.png",
        "covers/two.png",
        250_000,
    )
    .await
    .expect("stale cover write runs");
    assert!(
        !stale,
        "a cover confirm carrying a stale expected key must report no rows updated"
    );
    let after_stale = cover_cols(&db.pool, item).await;
    assert_eq!(
        after_stale, after_first,
        "the loser must not overwrite any of the three columns, got {after_stale:?}"
    );

    // The correct expectation replaces all three. 250_000 replaces 640_000; a
    // summing bug would read 890_000 and a dropped write 640_000.
    let replace = items::update_item_cover(
        &db.pool,
        item,
        owner,
        Some("covers/one.png"),
        "https://cdn.test/cover-two.png",
        "covers/two.png",
        250_000,
    )
    .await
    .expect("cover replace runs");
    assert!(
        replace,
        "a cover confirm carrying the current key must land"
    );
    let after_replace = cover_cols(&db.pool, item).await;
    assert_eq!(
        after_replace,
        (
            Some("https://cdn.test/cover-two.png".to_string()),
            Some("covers/two.png".to_string()),
            Some(250_000)
        ),
        "the replace must swap all three columns, got {after_replace:?}"
    );

    // Ownership is the other half of the same predicate.
    let stranger = seed_user(&db.pool, "media_cover_stranger").await;
    let by_stranger = items::update_item_cover(
        &db.pool,
        item,
        stranger,
        Some("covers/two.png"),
        "https://cdn.test/stolen.png",
        "covers/stolen.png",
        11,
    )
    .await
    .expect("stranger cover write runs");
    assert!(
        !by_stranger,
        "a non-owner cover write must report no rows updated"
    );
    let after_stranger = cover_cols(&db.pool, item).await;
    assert_eq!(
        after_stranger, after_replace,
        "the non-owner write must not land, got {after_stranger:?}"
    );
}

// ── size and metadata writebacks ─────────────────────────────────────────────

/// `get_item_file_sizes` feeds the storage decrement on delete, so it must read
/// the three columns into the three fields without crossing them, and a missing
/// item must read as three `None`s rather than an error (the delete path calls
/// it after the row may already be gone).
#[tokio::test]
async fn file_sizes_read_back_per_column_and_a_missing_item_reads_as_none() {
    let db = TestDb::new().await;
    let (_owner, item) = owner_and_item(&db, "sizes").await;

    // Three distinct values, so a column swap changes the answer.
    sqlx::query(
        "UPDATE items SET audio_file_size_bytes = 5000000,
                          cover_file_size_bytes = 250000,
                          video_file_size_bytes = 3000000
         WHERE id = $1",
    )
    .bind(item)
    .execute(&db.pool)
    .await
    .expect("seed the three sizes");

    let sizes = items::get_item_file_sizes(&db.pool, item)
        .await
        .expect("read sizes");
    assert_eq!(
        sizes.audio_file_size_bytes,
        Some(5_000_000),
        "audio size read from the audio column"
    );
    assert_eq!(
        sizes.cover_file_size_bytes,
        Some(250_000),
        "cover size read from the cover column"
    );
    assert_eq!(
        sizes.video_file_size_bytes,
        Some(3_000_000),
        "video size read from the video column"
    );

    let missing = items::get_item_file_sizes(&db.pool, ItemId::new())
        .await
        .expect("a missing item is not an error here");
    assert_eq!(
        (
            missing.audio_file_size_bytes,
            missing.cover_file_size_bytes,
            missing.video_file_size_bytes
        ),
        (None, None, None),
        "a deleted item must decrement nothing, so it reads as three Nones"
    );
}

/// Each size writeback is ownership-filtered and touches exactly one column.
/// Written as one test because the interesting assertion is the cross-check: the
/// other two columns are unchanged after each call.
#[tokio::test]
async fn size_writebacks_are_owner_scoped_and_touch_one_column_each() {
    let db = TestDb::new().await;
    let (owner, item) = owner_and_item(&db, "writeback").await;
    let stranger = seed_user(&db.pool, "media_size_stranger").await;

    items::update_item_audio_file_size(&db.pool, item, owner, 6_200_000)
        .await
        .expect("audio size write");
    items::update_item_cover_file_size(&db.pool, item, owner, 480_000)
        .await
        .expect("cover size write");
    items::update_item_video_file_size(&db.pool, item, owner, 9_100_000)
        .await
        .expect("video size write");

    let sizes = items::get_item_file_sizes(&db.pool, item)
        .await
        .expect("read sizes");
    assert_eq!(
        (
            sizes.audio_file_size_bytes,
            sizes.cover_file_size_bytes,
            sizes.video_file_size_bytes
        ),
        (Some(6_200_000), Some(480_000), Some(9_100_000)),
        "each writeback lands in its own column"
    );

    // A stranger's writeback is a silent no-op: the functions return Ok either
    // way, so the only observable contract is that nothing changed.
    items::update_item_audio_file_size(&db.pool, item, stranger, 17)
        .await
        .expect("stranger audio size write");
    items::update_item_cover_file_size(&db.pool, item, stranger, 19)
        .await
        .expect("stranger cover size write");
    items::update_item_video_file_size(&db.pool, item, stranger, 23)
        .await
        .expect("stranger video size write");

    let after = items::get_item_file_sizes(&db.pool, item)
        .await
        .expect("read sizes again");
    assert_eq!(
        (
            after.audio_file_size_bytes,
            after.cover_file_size_bytes,
            after.video_file_size_bytes
        ),
        (Some(6_200_000), Some(480_000), Some(9_100_000)),
        "a non-owner must not be able to rewrite another creator's quota numbers"
    );
}

/// `update_item_video_s3_key` returns the updated row, and its ownership filter
/// is enforced by the `fetch_one`: a non-owner gets a row-not-found database
/// error rather than a silent success, and the stored key is untouched.
#[tokio::test]
async fn setting_a_video_key_returns_the_row_and_a_non_owner_gets_row_not_found() {
    let db = TestDb::new().await;
    let (owner, item) = owner_and_item(&db, "vkey").await;
    let stranger = seed_user(&db.pool, "media_vkey_stranger").await;

    let updated = items::update_item_video_s3_key(&db.pool, item, owner, "videos/take-one.mp4")
        .await
        .expect("owner video key write");
    assert_eq!(updated.id, item, "the returned row is the item written");
    assert_eq!(
        updated.video_s3_key.as_deref(),
        Some("videos/take-one.mp4"),
        "the returned row carries the new key, got {:?}",
        updated.video_s3_key
    );

    let err = items::update_item_video_s3_key(&db.pool, item, stranger, "videos/stolen.mp4")
        .await
        .expect_err("a non-owner must not set another creator's video key");
    assert!(
        matches!(err, AppError::Database(sqlx::Error::RowNotFound)),
        "the ownership filter matches no row, so the error is RowNotFound, got {err:?}"
    );

    let cols = video_cols(&db.pool, item).await;
    assert_eq!(
        cols.0.as_deref(),
        Some("videos/take-one.mp4"),
        "the non-owner write must not land, got {cols:?}"
    );
}

/// Video metadata is three independent fields written in one statement. The
/// values are deliberately asymmetric so a width/height transposition fails, and
/// the clear-to-null pass pins that `None` writes NULL rather than being skipped.
#[tokio::test]
async fn video_metadata_writes_all_three_fields_is_owner_scoped_and_can_clear_them() {
    let db = TestDb::new().await;
    let (owner, item) = owner_and_item(&db, "vmeta").await;
    let stranger = seed_user(&db.pool, "media_vmeta_stranger").await;

    items::update_item_video_metadata(&db.pool, item, owner, Some(754), Some(1920), Some(1080))
        .await
        .expect("owner metadata write");

    let read = |pool: sqlx::PgPool| async move {
        sqlx::query_as::<_, (Option<i32>, Option<i32>, Option<i32>)>(
            "SELECT video_duration_seconds, video_width, video_height FROM items WHERE id = $1",
        )
        .bind(item)
        .fetch_one(&pool)
        .await
        .expect("read video metadata")
    };

    let after_owner = read(db.pool.clone()).await;
    assert_eq!(
        after_owner,
        (Some(754), Some(1920), Some(1080)),
        "duration, width and height each land in their own column, got {after_owner:?}"
    );

    items::update_item_video_metadata(&db.pool, item, stranger, Some(11), Some(320), Some(240))
        .await
        .expect("stranger metadata write");
    let after_stranger = read(db.pool.clone()).await;
    assert_eq!(
        after_stranger, after_owner,
        "a non-owner must not rewrite the metadata, got {after_stranger:?}"
    );

    items::update_item_video_metadata(&db.pool, item, owner, None, None, None)
        .await
        .expect("owner metadata clear");
    let cleared = read(db.pool.clone()).await;
    assert_eq!(
        cleared,
        (None, None, None),
        "writing None must clear the columns, not leave the old values, got {cleared:?}"
    );
}