//! DB-backed integration tests for the export epoch state machine (migration 014).
//!
//! These run against a REAL Postgres: `#[sqlx::test]` creates a throwaway database per test and
//! runs `backend/migrations/` into it, so the schema, the enums, the `UNIQUE (event_id, type)`
//! constraint and the `export_current` view are the production ones — not a mock.
//!
//! THE INVARIANT, from migration 014:
//!
//!     An export is downloadable IFF
//!         event.export_released_at IS NOT NULL
//!     AND export_job.epoch = event.export_epoch
//!     AND export_job.status = 'done'
//!
//! Readiness is DERIVED (the `export_current` view), never stored. Every test below pins one leg of
//! that invariant with the exact SQL `src/` executes (see `tests/common/mod.rs`).

mod common;

use common::*;
use sqlx::PgPool;

// ─────────────────────────────────────────────────────────────────────────────
// 1. Epoch monotonicity
// ─────────────────────────────────────────────────────────────────────────────

/// Release, reopen and re-release each bump `export_epoch`, and `RETURNING export_epoch` hands the
/// caller the POST-increment value.
///
/// PREVENTS: a worker born with the PRE-increment epoch. It would be inert from the instant it
/// started — every one of its writes is `epoch`-guarded, so `claim_job`/`finalize_job` would match
/// nothing, the job row would sit at `pending` 0% forever with no live worker, and the host's
/// download button would spin and then 404. The keepsake would never be built at all.
#[sqlx::test]
async fn release_returns_post_increment_epoch(pool: PgPool) {
    let event_id = seed_event(&pool, "wedding").await;
    assert_eq!(event_epoch(&pool, event_id).await, 0, "a fresh event starts at epoch 0");

    let released = release_gallery(&pool, "wedding").await.expect("release claims the event");
    assert_eq!(released, 1, "RETURNING must give the epoch AFTER the +1, not before");
    assert_eq!(event_epoch(&pool, event_id).await, released, "worker's epoch == event's epoch");

    // The jobs armed by the release carry exactly that epoch — this is what makes the worker's
    // guarded writes match.
    for t in ["zip", "html"] {
        let (status, epoch, _) = job_row(&pool, event_id, t).await.expect("job armed");
        assert_eq!(status, "pending");
        assert_eq!(epoch, released, "{t} job must be armed at the epoch the worker was born with");
    }
}

/// Epoch is strictly monotonic across the whole release/reopen/re-release cycle, and a second
/// release attempt while already released is rejected WITHOUT bumping.
///
/// PREVENTS: epoch reuse. If a reopen could return the event to an epoch some old `done` row still
/// carries, a retired keepsake — one that a guest asked to be taken down from — would silently
/// become downloadable again.
#[sqlx::test]
async fn epoch_is_strictly_monotonic_across_reopen(pool: PgPool) {
    let event_id = seed_event(&pool, "wedding").await;

    assert_eq!(release_gallery(&pool, "wedding").await, Some(1));

    // A duplicate release is a no-op (`WHERE export_released_at IS NULL`) and must NOT bump.
    assert_eq!(release_gallery(&pool, "wedding").await, None, "already released");
    assert_eq!(event_epoch(&pool, event_id).await, 1, "a rejected release must not move the epoch");

    // Reopen retires the generation with ONE write.
    assert_eq!(open_event(&pool, "wedding").await, 1);
    assert_eq!(event_epoch(&pool, event_id).await, 2, "reopen bumps");

    // And re-releasing bumps again — never back to 1.
    assert_eq!(release_gallery(&pool, "wedding").await, Some(3), "re-release bumps again");
    assert_eq!(event_epoch(&pool, event_id).await, 3);
}

// ─────────────────────────────────────────────────────────────────────────────
// 2. A retired-epoch worker is inert
// ─────────────────────────────────────────────────────────────────────────────

/// A worker holding a retired epoch cannot write anything anybody can see: once the rows have been
/// re-armed at a newer epoch, its `update_progress` and `finalize_job` both match 0 rows, and
/// `export_current` never exposes its output.
///
/// PREVENTS: the classic lost race — a slow worker from BEFORE a takedown finishing afterwards and
/// publishing an archive that still contains the photo a guest asked to have removed. "Please take
/// my photo out" is the one request that most needs to reach the keepsake, and the keepsake is the
/// artifact people keep forever.
#[sqlx::test]
async fn retired_epoch_worker_writes_are_no_ops(pool: PgPool) {
    let event_id = seed_event(&pool, "wedding").await;
    let old_epoch = release_gallery(&pool, "wedding").await.unwrap();

    // Worker A is born at epoch 1 and claims the ZIP.
    assert!(claim_job(&pool, event_id, "zip", old_epoch).await, "worker A wins its claim");
    assert!(update_progress(&pool, event_id, "zip", old_epoch, 40).await, "still live at 40%");

    // ── A takedown lands mid-export: `invalidate_and_arm(Affects::Both)` bumps and re-arms. ──
    let (_, _, new_epoch) = bump_epoch(&pool, "wedding").await.expect("bump on a released event");
    assert_eq!(new_epoch, old_epoch + 1);
    let mut conn = pool.acquire().await.unwrap();
    enqueue_types_at_epoch(&mut conn, event_id, new_epoch, &["zip", "html"]).await;
    drop(conn);

    // Worker A is now INERT BY CONSTRUCTION. Every write is guarded on its own birth epoch.
    assert!(
        !update_progress(&pool, event_id, "zip", old_epoch, 90).await,
        "the liveness check must report `false` so worker A stops grinding through the gallery"
    );
    assert!(
        !finalize_job(&pool, event_id, "zip", old_epoch, "exports/Gallery.1.zip").await,
        "worker A's finalize MUST affect 0 rows — this is the write that would have published a \
         keepsake still containing the taken-down photo"
    );

    // The re-armed row is untouched by the loser: still pending at the LIVE epoch, waiting for the
    // fresh worker. (If worker A had won, this row would read `done` at epoch 1.)
    let (status, epoch, file_path) = job_row(&pool, event_id, "zip").await.unwrap();
    assert_eq!((status.as_str(), epoch), ("pending", new_epoch));
    assert_eq!(file_path, None, "the loser's file_path must never be recorded");

    // And nothing is downloadable — not the stale archive, not anything.
    assert_eq!(downloadable(&pool, event_id, "zip").await, None);
}

/// The documented, deliberate nuance in `claim_job`: after a bare `open_event` (which writes
/// NOTHING to `export_job` — that is the point of the design), a worker at the old epoch still WINS
/// its claim and can still write `done`. That is wasted work, not incorrectness: retirement is
/// enforced at READ time. `export_current` must refuse to expose the row.
///
/// PREVENTS: someone "optimising" `claim_job` into a cross-table `EXISTS (SELECT ... FROM event)`
/// guard — the exact unsound guard migration 014 removed (under READ COMMITTED, a blocked UPDATE
/// re-evaluates same-row predicates but answers other-table subqueries from a stale snapshot).
/// This test pins the read-time enforcement so the write-time guard is never re-added.
#[sqlx::test]
async fn reopen_retires_at_read_time_not_write_time(pool: PgPool) {
    let event_id = seed_event(&pool, "wedding").await;
    let epoch = release_gallery(&pool, "wedding").await.unwrap();

    assert!(claim_job(&pool, event_id, "zip", epoch).await);

    // Host reopens uploads. No export_job row is touched.
    assert_eq!(open_event(&pool, "wedding").await, 1);

    // The in-flight worker's row-local writes still match — it was never told to stop.
    assert!(
        finalize_job(&pool, event_id, "zip", epoch, "exports/Gallery.1.zip").await,
        "documented: the claim/finalize is guarded on the JOB row's epoch, not the event's"
    );
    let (status, _, _) = job_row(&pool, event_id, "zip").await.unwrap();
    assert_eq!(status, "done", "the row really does say done");

    // …and yet it is invisible. `export_current` requires the event to be released AND the epochs to
    // match; the reopen broke both. A worker at a dead epoch writes a row nobody can see.
    assert!(!in_export_current(&pool, event_id, "zip").await);
    assert_eq!(
        downloadable(&pool, event_id, "zip").await,
        None,
        "a reopened event must serve NO keepsake, however finished the job row looks"
    );
}

// ─────────────────────────────────────────────────────────────────────────────
// 3. `export_current` exactness (table-driven)
// ─────────────────────────────────────────────────────────────────────────────

/// The view is the ONE definition of "downloadable". Released + `done` + matching epoch ⇒ present;
/// break ANY single leg ⇒ absent. Nothing else may make it appear or disappear.
///
/// PREVENTS, leg by leg:
///  * `released` — serving a keepsake for an event whose uploads are still open, i.e. an archive
///    missing every photo taken after the snapshot.
///  * `done` — handing out a half-written ZIP (a corrupt keepsake, downloaded once, kept forever).
///  * `epoch` — the retired-generation download: the 404-forever keepsake, or worse, the archive
///    still containing content that was taken down.
#[sqlx::test]
async fn export_current_is_exactly_the_invariant(pool: PgPool) {
    // (name, released?, status, job epoch offset from the event epoch, expected visible)
    let cases: &[(&str, bool, &str, i64, bool)] = &[
        ("released + done + current epoch", true, "done", 0, true),
        ("NOT released (done, epoch matches)", false, "done", 0, false),
        ("NOT done: pending", true, "pending", 0, false),
        ("NOT done: running", true, "running", 0, false),
        ("NOT done: failed", true, "failed", 0, false),
        ("stale epoch (done, released)", true, "done", -1, false),
        ("future epoch (done, released)", true, "done", 1, false),
        ("migration-014 retired sentinel epoch -1", true, "done", -2, false),
    ];

    for (i, (name, released, status, offset, expect_visible)) in cases.iter().enumerate() {
        let slug = format!("case{i}");
        let event_id = seed_event(&pool, &slug).await;

        // Get the event to a known epoch (1) either by releasing it, or — for the unreleased case —
        // by releasing and reopening, which leaves it unreleased at a non-zero epoch.
        let event_epoch_now = if *released {
            release_gallery(&pool, &slug).await.unwrap()
        } else {
            release_gallery(&pool, &slug).await.unwrap();
            open_event(&pool, &slug).await;
            event_epoch(&pool, event_id).await
        };

        // Plant a single ZIP job row in the exact state under test. `-2` encodes "the sentinel the
        // migration stamps on retired rows", which must never equal a non-negative event epoch.
        // (Clear the rows the release armed first — `UNIQUE (event_id, type)`.)
        sqlx::query("DELETE FROM export_job WHERE event_id = $1")
            .bind(event_id)
            .execute(&pool)
            .await
            .expect("clear armed jobs");

        let job_epoch = if *offset == -2 { -1 } else { event_epoch_now + offset };
        sqlx::query(
            "INSERT INTO export_job (event_id, type, status, progress_pct, epoch, file_path)
             VALUES ($1, 'zip', $2::export_status, 100, $3, 'exports/Gallery.zip')",
        )
        .bind(event_id)
        .bind(*status)
        .bind(job_epoch)
        .execute(&pool)
        .await
        .expect("plant job row");

        let visible = downloadable(&pool, event_id, "zip").await.is_some();
        assert_eq!(
            visible, *expect_visible,
            "export_current exactness violated for case: {name} \
             (released={released}, status={status}, job_epoch={job_epoch}, event_epoch={event_epoch_now})"
        );
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// 4. The ViewerOnly ZIP carry-forward
// ─────────────────────────────────────────────────────────────────────────────

/// Branch A — the ZIP is `done` at the outgoing epoch: the carry-forward re-stamps it to the new
/// epoch (rows_affected = 1), so only the HTML viewer is rebuilt and the finished ZIP stays
/// downloadable throughout.
///
/// PREVENTS: rebuilding a multi-GB archive because someone deleted a comment. The ZIP holds media,
/// not comments — a needless rebuild would 404 the photo download for minutes to change nothing
/// inside it.
#[sqlx::test]
async fn viewer_only_carries_a_done_zip_forward(pool: PgPool) {
    let event_id = seed_event(&pool, "wedding").await;
    let e1 = release_gallery(&pool, "wedding").await.unwrap();

    // Both halves finish at epoch 1 — the keepsake is live.
    for t in ["zip", "html"] {
        assert!(claim_job(&pool, event_id, t, e1).await);
        assert!(finalize_job(&pool, event_id, t, e1, &format!("exports/{t}.{e1}.zip")).await);
    }
    let zip_file = downloadable(&pool, event_id, "zip").await.expect("zip is live");

    // ── A comment is moderated: invalidate_and_arm(Affects::ViewerOnly). ──
    let (_, _, e2) = bump_epoch(&pool, "wedding").await.unwrap();
    let carried = carry_zip_forward(&pool, event_id, e2).await;
    assert!(carried, "a `done` ZIP at epoch-1 MUST be carried forward (rows_affected == 1)");

    // Only the viewer is re-armed…
    let mut conn = pool.acquire().await.unwrap();
    enqueue_types_at_epoch(&mut conn, event_id, e2, &["html"]).await;
    drop(conn);

    // …and the ZIP is STILL DOWNLOADABLE, at the new epoch, pointing at the same, unrenamed file.
    let (status, epoch, _) = job_row(&pool, event_id, "zip").await.unwrap();
    assert_eq!((status.as_str(), epoch), ("done", e2), "the ZIP row rode the epoch bump");
    assert_eq!(
        downloadable(&pool, event_id, "zip").await,
        Some(zip_file),
        "the carried archive must never stop being served — same file, new epoch"
    );

    // The viewer, meanwhile, is correctly retired and pending a rebuild.
    assert_eq!(job_row(&pool, event_id, "html").await.unwrap().0, "pending");
    assert_eq!(downloadable(&pool, event_id, "html").await, None);
}

/// Branch B — THE BUG WE JUST FIXED. If the ZIP is still `pending`/`running` when the comment is
/// moderated (which is MINUTES for a real multi-GB gallery, and deleting a comment right after
/// release is an utterly ordinary thing to do), the carry-forward matches NOTHING
/// (rows_affected = 0) — so the caller must NOT assume it carried, and must re-arm the ZIP too.
///
/// PREVENTS: the stranded ZIP. Blindly re-arming only the viewer would leave the ZIP row at the
/// retired epoch; the in-flight worker then finishes and writes `done` at an epoch `export_current`
/// no longer matches, nothing ever re-arms it, and `GET /export/zip` 404s FOREVER — a keepsake the
/// couple paid for that simply never appears, short of a reboot.
#[sqlx::test]
async fn viewer_only_carry_forward_matches_nothing_when_zip_unfinished(pool: PgPool) {
    for zip_state in ["pending", "running"] {
        let slug = format!("wedding-{zip_state}");
        let event_id = seed_event(&pool, &slug).await;
        let e1 = release_gallery(&pool, &slug).await.unwrap();

        // The ZIP worker is still going; only the viewer has finished.
        if zip_state == "running" {
            assert!(claim_job(&pool, event_id, "zip", e1).await);
        }
        assert!(claim_job(&pool, event_id, "html", e1).await);
        assert!(finalize_job(&pool, event_id, "html", e1, "exports/Memories.1.zip").await);

        // ── The comment is moderated. ──
        let (_, _, e2) = bump_epoch(&pool, &slug).await.unwrap();
        let carried = carry_zip_forward(&pool, event_id, e2).await;

        assert!(
            !carried,
            "a {zip_state} ZIP has nothing to carry forward — the UPDATE must affect 0 rows \
             (its `status = 'done'` predicate is the whole precondition)"
        );

        // The carry-forward's OWN result decides. It didn't match ⇒ rebuild the ZIP as well.
        let types: &[&str] = if carried { &["html"] } else { &["zip", "html"] };
        let mut conn = pool.acquire().await.unwrap();
        enqueue_types_at_epoch(&mut conn, event_id, e2, types).await;
        drop(conn);

        // THE ASSERTION THAT WOULD HAVE CAUGHT THE BUG: the ZIP must not be stranded at the dead
        // epoch. It is re-armed at the live one, so a fresh worker will actually build it.
        let (status, epoch, _) = job_row(&pool, event_id, "zip").await.unwrap();
        assert_eq!(
            (status.as_str(), epoch),
            ("pending", e2),
            "the unfinished ZIP MUST be re-armed at the new epoch, not left stranded at {e1}"
        );

        // Even if the old in-flight worker now "finishes", it is inert and cannot resurrect itself.
        assert!(!finalize_job(&pool, event_id, "zip", e1, "exports/Gallery.1.zip").await);
        assert_eq!(downloadable(&pool, event_id, "zip").await, None);
    }
}

/// The re-arm upsert must never clobber the archive it just carried forward.
///
/// `enqueue_types_at_epoch`'s `WHERE export_job.status <> 'done' OR export_job.epoch <> EXCLUDED.epoch`
/// is the "startup recovery must not clobber a good half" rule, expressed as the readiness predicate
/// itself. PREVENTS: boot recovery resetting a perfectly good, downloadable ZIP back to `pending`
/// and making the keepsake 404 while it needlessly rebuilds.
#[sqlx::test]
async fn enqueue_preserves_a_done_half_at_the_current_epoch(pool: PgPool) {
    let event_id = seed_event(&pool, "wedding").await;
    let e1 = release_gallery(&pool, "wedding").await.unwrap();

    // The ZIP finished; the HTML worker was killed mid-flight (crash) and sits at `running`.
    assert!(claim_job(&pool, event_id, "zip", e1).await);
    assert!(finalize_job(&pool, event_id, "zip", e1, "exports/Gallery.1.zip").await);
    assert!(claim_job(&pool, event_id, "html", e1).await);

    // Boot recovery re-arms both types at the SAME epoch.
    let mut conn = pool.acquire().await.unwrap();
    enqueue_types_at_epoch(&mut conn, event_id, e1, &["zip", "html"]).await;
    drop(conn);

    // The good half survives untouched…
    let (status, epoch, file_path) = job_row(&pool, event_id, "zip").await.unwrap();
    assert_eq!((status.as_str(), epoch), ("done", e1), "a done half at the live epoch is preserved");
    assert_eq!(file_path.as_deref(), Some("exports/Gallery.1.zip"), "file_path not nulled");
    assert!(downloadable(&pool, event_id, "zip").await.is_some());

    // …and only the missing half is re-armed.
    assert_eq!(job_row(&pool, event_id, "html").await.unwrap().0, "pending");
}