Mangalord/backend/tests/crawler_jobs.rs

//! Integration tests for `crawler::jobs` queue operations.
//!
//! Uses `#[sqlx::test(migrations = "./migrations")]` which provisions a fresh
//! migrated DB per test. No browser, no axum router — these exercise the SQL
//! shape and dedup-index semantics directly against Postgres.

use std::time::Duration;

use mangalord::crawler::jobs::{
    self, EnqueueResult, JobPayload, KIND_ANALYZE_PAGE, KIND_SYNC_CHAPTER_CONTENT, KIND_SYNC_MANGA,
};
use sqlx::PgPool;
use uuid::Uuid;

fn chapter_content_payload(chapter_id: Uuid) -> JobPayload {
    JobPayload::SyncChapterContent {
        source_id: "target".into(),
        chapter_id,
        source_chapter_key: format!("ch-{chapter_id}"),
    }
}

/// A non-`SyncChapterContent` payload, used to assert that only the
/// chapter-content kind is deduplicated by the partial index and that
/// `lease`'s kind filter correctly excludes other kinds.
fn sync_manga_payload(key: &str) -> JobPayload {
    JobPayload::SyncManga {
        source_id: "target".into(),
        source_manga_key: key.into(),
        url: format!("https://target.example/manga/{key}"),
        title: format!("Manga {key}"),
    }
}

async fn job_state(pool: &PgPool, id: Uuid) -> String {
    sqlx::query_scalar::<_, String>("SELECT state FROM crawler_jobs WHERE id = $1")
        .bind(id)
        .fetch_one(pool)
        .await
        .unwrap()
}

async fn job_attempts(pool: &PgPool, id: Uuid) -> i32 {
    sqlx::query_scalar::<_, i32>("SELECT attempts FROM crawler_jobs WHERE id = $1")
        .bind(id)
        .fetch_one(pool)
        .await
        .unwrap()
}

async fn job_count(pool: &PgPool) -> i64 {
    sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM crawler_jobs")
        .fetch_one(pool)
        .await
        .unwrap()
}

#[sqlx::test(migrations = "./migrations")]
async fn enqueue_inserts_pending_row_with_round_trip_payload(pool: PgPool) {
    let chapter_id = Uuid::new_v4();
    let payload = chapter_content_payload(chapter_id);

    let result = jobs::enqueue(&pool, &payload).await.unwrap();
    let id = match result {
        EnqueueResult::Inserted(id) => id,
        EnqueueResult::Skipped => panic!("expected Inserted on first enqueue"),
    };

    assert_eq!(job_state(&pool, id).await, "pending");
    assert_eq!(job_attempts(&pool, id).await, 0);

    let raw_payload: serde_json::Value =
        sqlx::query_scalar("SELECT payload FROM crawler_jobs WHERE id = $1")
            .bind(id)
            .fetch_one(&pool)
            .await
            .unwrap();
    let decoded: JobPayload = serde_json::from_value(raw_payload).unwrap();
    match decoded {
        JobPayload::SyncChapterContent {
            source_id,
            chapter_id: c,
            source_chapter_key,
        } => {
            assert_eq!(source_id, "target");
            assert_eq!(c, chapter_id);
            assert_eq!(source_chapter_key, format!("ch-{chapter_id}"));
        }
        _ => panic!("payload variant mismatch"),
    }
}

#[sqlx::test(migrations = "./migrations")]
async fn duplicate_chapter_content_while_pending_is_skipped(pool: PgPool) {
    let chapter_id = Uuid::new_v4();
    let p = chapter_content_payload(chapter_id);

    let first = jobs::enqueue(&pool, &p).await.unwrap();
    assert!(matches!(first, EnqueueResult::Inserted(_)));

    let second = jobs::enqueue(&pool, &p).await.unwrap();
    assert!(matches!(second, EnqueueResult::Skipped));

    assert_eq!(job_count(&pool).await, 1);
}

#[sqlx::test(migrations = "./migrations")]
async fn duplicate_after_done_releases_dedup_slot(pool: PgPool) {
    let chapter_id = Uuid::new_v4();
    let p = chapter_content_payload(chapter_id);

    let first_id = match jobs::enqueue(&pool, &p).await.unwrap() {
        EnqueueResult::Inserted(id) => id,
        EnqueueResult::Skipped => panic!("first enqueue should insert"),
    };
    // Move the first job out of (pending|running) so the partial index drops it.
    sqlx::query("UPDATE crawler_jobs SET state = 'done' WHERE id = $1")
        .bind(first_id)
        .execute(&pool)
        .await
        .unwrap();

    let second = jobs::enqueue(&pool, &p).await.unwrap();
    assert!(
        matches!(second, EnqueueResult::Inserted(_)),
        "after done the chapter_id slot is free again"
    );
    assert_eq!(job_count(&pool).await, 2);
}

#[sqlx::test(migrations = "./migrations")]
async fn different_chapter_ids_can_coexist(pool: PgPool) {
    let p1 = chapter_content_payload(Uuid::new_v4());
    let p2 = chapter_content_payload(Uuid::new_v4());
    assert!(matches!(
        jobs::enqueue(&pool, &p1).await.unwrap(),
        EnqueueResult::Inserted(_)
    ));
    assert!(matches!(
        jobs::enqueue(&pool, &p2).await.unwrap(),
        EnqueueResult::Inserted(_)
    ));
    assert_eq!(job_count(&pool).await, 2);
}

#[sqlx::test(migrations = "./migrations")]
async fn non_chapter_content_payloads_are_never_deduped(pool: PgPool) {
    let p = sync_manga_payload("foo");
    assert!(matches!(
        jobs::enqueue(&pool, &p).await.unwrap(),
        EnqueueResult::Inserted(_)
    ));
    assert!(matches!(
        jobs::enqueue(&pool, &p).await.unwrap(),
        EnqueueResult::Inserted(_)
    ));
    assert_eq!(job_count(&pool).await, 2);
}

#[sqlx::test(migrations = "./migrations")]
async fn lease_marks_running_and_bumps_attempts_and_sets_leased_until(pool: PgPool) {
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        EnqueueResult::Skipped => unreachable!(),
    };

    let leases = jobs::lease(&pool, None, 10, Duration::from_secs(60))
        .await
        .unwrap();
    assert_eq!(leases.len(), 1);
    let lease = &leases[0];
    assert_eq!(lease.id, id);
    assert_eq!(lease.attempts, 1);

    assert_eq!(job_state(&pool, id).await, "running");

    let leased_until: Option<chrono::DateTime<chrono::Utc>> =
        sqlx::query_scalar("SELECT leased_until FROM crawler_jobs WHERE id = $1")
            .bind(id)
            .fetch_one(&pool)
            .await
            .unwrap();
    let leased_until = leased_until.expect("leased_until set");
    assert!(leased_until > chrono::Utc::now());
}

#[sqlx::test(migrations = "./migrations")]
async fn renew_extends_leased_until_while_running(pool: PgPool) {
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        EnqueueResult::Skipped => unreachable!(),
    };

    // Lease with a short window, then collapse leased_until to the recent
    // past so the renew is unambiguously an extension.
    let leases = jobs::lease(&pool, None, 1, Duration::from_secs(5))
        .await
        .unwrap();
    assert_eq!(leases.len(), 1);
    sqlx::query("UPDATE crawler_jobs SET leased_until = now() - interval '1 second' WHERE id = $1")
        .bind(id)
        .execute(&pool)
        .await
        .unwrap();

    let still_owned = jobs::renew(&pool, id, leases[0].lease_generation, Duration::from_secs(120))
        .await
        .unwrap();
    assert!(still_owned, "renew on a running job returns true");

    let leased_until: chrono::DateTime<chrono::Utc> =
        sqlx::query_scalar("SELECT leased_until FROM crawler_jobs WHERE id = $1")
            .bind(id)
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(
        leased_until > chrono::Utc::now() + chrono::Duration::seconds(60),
        "leased_until pushed ~120s into the future"
    );
    assert_eq!(job_state(&pool, id).await, "running");
}

#[sqlx::test(migrations = "./migrations")]
async fn renew_is_noop_once_job_no_longer_running(pool: PgPool) {
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        EnqueueResult::Skipped => unreachable!(),
    };
    let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
        .await
        .unwrap();
    // Job completes — heartbeat should now see it's no longer ours.
    jobs::ack_done(&pool, leases[0].id, leases[0].lease_generation)
        .await
        .unwrap();

    let still_owned = jobs::renew(&pool, id, leases[0].lease_generation, Duration::from_secs(120))
        .await
        .unwrap();
    assert!(!still_owned, "renew on a non-running job returns false");
    assert_eq!(job_state(&pool, id).await, "done");
}

#[sqlx::test(migrations = "./migrations")]
async fn lease_with_kind_filter_only_matches_that_kind(pool: PgPool) {
    let manga_id = match jobs::enqueue(&pool, &sync_manga_payload("foo"))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let chapter_id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };

    let leases = jobs::lease(
        &pool,
        Some(KIND_SYNC_CHAPTER_CONTENT),
        10,
        Duration::from_secs(60),
    )
    .await
    .unwrap();
    assert_eq!(leases.len(), 1, "only chapter content payload leases");
    assert_eq!(leases[0].id, chapter_id);
    // sync_manga is still pending
    assert_eq!(job_state(&pool, manga_id).await, "pending");
}

#[sqlx::test(migrations = "./migrations")]
async fn lease_kinds_matches_listed_kinds_and_excludes_others(pool: PgPool) {
    // The crawl worker drains both sync_chapter_content and sync_manga, but
    // never analyze_page (owned by the analysis daemon).
    let manga_id = match jobs::enqueue(&pool, &sync_manga_payload("foo"))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let chapter_id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let analyze_id = match jobs::enqueue(
        &pool,
        &JobPayload::AnalyzePage {
            page_id: Uuid::new_v4(),
            force: false,
        },
    )
    .await
    .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };

    let leases = jobs::lease_kinds(
        &pool,
        &[KIND_SYNC_CHAPTER_CONTENT, KIND_SYNC_MANGA],
        10,
        Duration::from_secs(60),
    )
    .await
    .unwrap();
    let leased_ids: std::collections::HashSet<Uuid> = leases.iter().map(|l| l.id).collect();
    assert_eq!(leases.len(), 2, "both crawl kinds lease");
    assert!(leased_ids.contains(&manga_id));
    assert!(leased_ids.contains(&chapter_id));
    // analyze_page stays pending — not in the requested kinds.
    assert_eq!(job_state(&pool, analyze_id).await, "pending");
    // Sanity: KIND_ANALYZE_PAGE alone leases only it.
    let only_analyze = jobs::lease_kinds(&pool, &[KIND_ANALYZE_PAGE], 10, Duration::from_secs(60))
        .await
        .unwrap();
    assert_eq!(only_analyze.len(), 1);
    assert_eq!(only_analyze[0].id, analyze_id);
}

#[sqlx::test(migrations = "./migrations")]
async fn concurrent_leases_under_skip_locked_return_disjoint_ids(pool: PgPool) {
    // 4 pending jobs, two concurrent calls each asking for up to 2.
    let mut ids = Vec::new();
    for _ in 0..4 {
        let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
            .await
            .unwrap()
        {
            EnqueueResult::Inserted(id) => id,
            _ => unreachable!(),
        };
        ids.push(id);
    }

    let (a, b) = tokio::join!(
        jobs::lease(&pool, None, 2, Duration::from_secs(60)),
        jobs::lease(&pool, None, 2, Duration::from_secs(60)),
    );
    let a = a.unwrap();
    let b = b.unwrap();
    let mut seen: Vec<Uuid> = a.iter().chain(b.iter()).map(|l| l.id).collect();
    seen.sort();
    seen.dedup();
    let count = a.len() + b.len();
    assert_eq!(
        seen.len(),
        count,
        "no id appears in both lease results (SKIP LOCKED)"
    );
    assert!(count >= 2, "at least one lease saw work");
    assert!(count <= 4);
}

#[sqlx::test(migrations = "./migrations")]
async fn stale_running_lease_can_be_reclaimed(pool: PgPool) {
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };

    let first = jobs::lease(&pool, None, 1, Duration::from_secs(60))
        .await
        .unwrap();
    assert_eq!(first.len(), 1);
    // Pretend the worker crashed: rewind leased_until into the past.
    sqlx::query("UPDATE crawler_jobs SET leased_until = now() - interval '1 minute' WHERE id = $1")
        .bind(id)
        .execute(&pool)
        .await
        .unwrap();

    let second = jobs::lease(&pool, None, 1, Duration::from_secs(60))
        .await
        .unwrap();
    assert_eq!(second.len(), 1, "stale running row was re-leased");
    assert_eq!(second[0].id, id);
    assert_eq!(second[0].attempts, 2, "attempts bumped again");
}

#[sqlx::test(migrations = "./migrations")]
async fn reclaim_orphaned_resets_only_expired_running_jobs(pool: PgPool) {
    // Three jobs: (a) running with an expired lease — a crash orphan that
    // must be reclaimed; (b) running with a live lease — a healthy peer's
    // in-flight job that must be left untouched; (c) pending — irrelevant,
    // must be untouched. Reclaim must touch only (a), refund its attempt,
    // and report exactly 1.
    let orphan = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let live = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let pending = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };

    // Lease the two that should become running (attempts -> 1 each).
    jobs::lease(&pool, None, 10, Duration::from_secs(60))
        .await
        .unwrap();
    // Leave `pending` pending by returning it (release refunds its attempt).
    // Test only has the id, not a Lease, so use the unowned variant.
    jobs::release_unowned(&pool, pending).await.unwrap();
    assert_eq!(job_state(&pool, pending).await, "pending");

    // Orphan: rewind its lease into the past (worker crashed). Live: keep
    // its lease in the future (peer still heartbeating).
    sqlx::query("UPDATE crawler_jobs SET leased_until = now() - interval '1 minute' WHERE id = $1")
        .bind(orphan)
        .execute(&pool)
        .await
        .unwrap();
    assert_eq!(job_state(&pool, live).await, "running");
    assert_eq!(job_attempts(&pool, live).await, 1);

    let reclaimed = jobs::reclaim_orphaned(&pool).await.unwrap();
    assert_eq!(reclaimed, 1, "only the expired-lease running job is reclaimed");

    // Orphan: back to pending, attempt refunded, lease cleared.
    assert_eq!(job_state(&pool, orphan).await, "pending");
    assert_eq!(
        job_attempts(&pool, orphan).await,
        0,
        "reclaim refunds the crashed attempt"
    );
    let orphan_leased: Option<chrono::DateTime<chrono::Utc>> =
        sqlx::query_scalar("SELECT leased_until FROM crawler_jobs WHERE id = $1")
            .bind(orphan)
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(orphan_leased.is_none(), "reclaim clears the lease");

    // Live peer job: untouched.
    assert_eq!(job_state(&pool, live).await, "running");
    assert_eq!(job_attempts(&pool, live).await, 1);

    // Pending job: untouched.
    assert_eq!(job_state(&pool, pending).await, "pending");
}

#[sqlx::test(migrations = "./migrations")]
async fn ack_done_transitions_state_and_clears_lease(pool: PgPool) {
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
        .await
        .unwrap();
    jobs::ack_done(&pool, leases[0].id, leases[0].lease_generation)
        .await
        .unwrap();

    assert_eq!(job_state(&pool, id).await, "done");
    let leased_until: Option<chrono::DateTime<chrono::Utc>> =
        sqlx::query_scalar("SELECT leased_until FROM crawler_jobs WHERE id = $1")
            .bind(id)
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(leased_until.is_none());
}

#[sqlx::test(migrations = "./migrations")]
async fn ack_failed_under_max_returns_to_pending_with_future_schedule(pool: PgPool) {
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
        .await
        .unwrap();
    let lease = &leases[0];
    jobs::ack_failed(
        &pool,
        lease.id,
        "boom",
        lease.attempts,
        lease.max_attempts,
        lease.lease_generation,
    )
    .await
    .unwrap();

    assert_eq!(job_state(&pool, id).await, "pending");

    let (scheduled_at, last_error): (chrono::DateTime<chrono::Utc>, Option<String>) =
        sqlx::query_as("SELECT scheduled_at, last_error FROM crawler_jobs WHERE id = $1")
            .bind(id)
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(scheduled_at > chrono::Utc::now());
    assert_eq!(last_error.as_deref(), Some("boom"));
}

#[sqlx::test(migrations = "./migrations")]
async fn ack_failed_at_max_marks_dead(pool: PgPool) {
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    // Force a single lease then mark "this was attempt N where N == max_attempts".
    let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
        .await
        .unwrap();
    let lease = &leases[0];
    jobs::ack_failed(
        &pool,
        lease.id,
        "final boom",
        lease.max_attempts,
        lease.max_attempts,
        lease.lease_generation,
    )
    .await
    .unwrap();

    assert_eq!(job_state(&pool, id).await, "dead");
    let last_error: Option<String> =
        sqlx::query_scalar("SELECT last_error FROM crawler_jobs WHERE id = $1")
            .bind(id)
            .fetch_one(&pool)
            .await
            .unwrap();
    assert_eq!(last_error.as_deref(), Some("final boom"));
}

#[sqlx::test(migrations = "./migrations")]
async fn ack_done_no_ops_when_lease_was_stolen(pool: PgPool) {
    // Worker A's lease expires, worker B re-leases the job (state stays
    // 'running' but attempts++ and leased_until refreshed). A late
    // ack_done from worker A must not clobber B's progress.
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    // Worker A grabs the lease, but its lease expires immediately.
    let a_leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
        .await
        .unwrap();
    sqlx::query("UPDATE crawler_jobs SET leased_until = now() - interval '1 minute' WHERE id = $1")
        .bind(id)
        .execute(&pool)
        .await
        .unwrap();
    // Worker B re-leases the expired-but-still-running job.
    let b_leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
        .await
        .unwrap();
    assert_eq!(b_leases.len(), 1);
    assert_eq!(b_leases[0].attempts, 2, "re-lease bumps attempts");
    assert_ne!(
        a_leases[0].lease_generation, b_leases[0].lease_generation,
        "re-lease must bump lease_generation so A's late ack can be distinguished"
    );

    // Worker A's late ack_done — now scoped to A's lease_generation by
    // migration 0032. The row stays `running` with B's progress
    // intact; A's stale ack matches nothing and is a no-op.
    jobs::ack_done(&pool, a_leases[0].id, a_leases[0].lease_generation)
        .await
        .unwrap();
    assert_eq!(
        job_state(&pool, id).await,
        "running",
        "A's stale ack_done must NOT clobber B's running lease"
    );
    // Finalize: worker B acks done with its own generation.
    jobs::ack_done(&pool, b_leases[0].id, b_leases[0].lease_generation)
        .await
        .unwrap();
    assert_eq!(job_state(&pool, id).await, "done");
    assert_eq!(job_attempts(&pool, id).await, 2);
}

#[sqlx::test(migrations = "./migrations")]
async fn ack_failed_no_ops_when_state_is_not_running(pool: PgPool) {
    // After a job transitions to 'done', a stale ack_failed (e.g. a
    // worker that finished work and queued its ack but then handed off
    // before the SQL ran) must not flip the state back to 'pending' or
    // 'dead'. The `state = 'running'` predicate enforces this.
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
        .await
        .unwrap();
    jobs::ack_done(&pool, leases[0].id, leases[0].lease_generation)
        .await
        .unwrap();
    assert_eq!(job_state(&pool, id).await, "done");

    // Late ack_failed arrives. Must be a no-op.
    jobs::ack_failed(&pool, leases[0].id, "late", 1, 5, leases[0].lease_generation)
        .await
        .unwrap();
    assert_eq!(
        job_state(&pool, id).await,
        "done",
        "late ack_failed must not resurrect a done job"
    );
}

#[sqlx::test(migrations = "./migrations")]
async fn release_no_ops_when_state_is_not_running(pool: PgPool) {
    // Mirror of ack_failed_no_ops_when_state_is_not_running. release also
    // decrements `attempts`, which would corrupt a re-leased job's
    // attempt count if the guard were missing.
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
        .await
        .unwrap();
    jobs::ack_done(&pool, leases[0].id, leases[0].lease_generation)
        .await
        .unwrap();
    let attempts_before = job_attempts(&pool, id).await;

    // Late release arrives.
    jobs::release(&pool, leases[0].id, leases[0].lease_generation)
        .await
        .unwrap();
    assert_eq!(
        job_state(&pool, id).await,
        "done",
        "late release must not flip a done job back to pending"
    );
    assert_eq!(
        job_attempts(&pool, id).await,
        attempts_before,
        "late release must not decrement attempts of a non-running job"
    );
}

#[sqlx::test(migrations = "./migrations")]
async fn release_returns_to_pending_and_undoes_attempt_increment(pool: PgPool) {
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
        .await
        .unwrap();
    assert_eq!(leases[0].attempts, 1);
    jobs::release(&pool, leases[0].id, leases[0].lease_generation)
        .await
        .unwrap();

    assert_eq!(job_state(&pool, id).await, "pending");
    assert_eq!(job_attempts(&pool, id).await, 0);
    let leased_until: Option<chrono::DateTime<chrono::Utc>> =
        sqlx::query_scalar("SELECT leased_until FROM crawler_jobs WHERE id = $1")
            .bind(id)
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(leased_until.is_none());
}

#[sqlx::test(migrations = "./migrations")]
async fn reap_done_deletes_old_rows_keeps_fresh(pool: PgPool) {
    // Two done rows: one old (updated_at 10 days ago), one fresh.
    let old_id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let fresh_id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };

    sqlx::query("UPDATE crawler_jobs SET state='done', updated_at = now() - interval '10 days' WHERE id = $1")
        .bind(old_id)
        .execute(&pool)
        .await
        .unwrap();
    sqlx::query("UPDATE crawler_jobs SET state='done' WHERE id = $1")
        .bind(fresh_id)
        .execute(&pool)
        .await
        .unwrap();

    let deleted = jobs::reap_done(&pool, 7).await.unwrap();
    assert_eq!(deleted, 1);

    let remaining: Vec<Uuid> = sqlx::query_scalar("SELECT id FROM crawler_jobs ORDER BY id")
        .fetch_all(&pool)
        .await
        .unwrap();
    assert_eq!(remaining, vec![fresh_id], "only fresh row remains");
}

#[sqlx::test(migrations = "./migrations")]
async fn lease_ties_on_scheduled_at_break_by_created_at(pool: PgPool) {
    // Locks in the tiebreaker that lets enqueue order survive the lease
    // step: when many jobs share `scheduled_at` (the common cron-batch
    // case), the worker must pick the earliest-inserted row, not whatever
    // Postgres returns in heap order. The enqueue path inserts chapters
    // in chapter-number order, so this tiebreaker is what makes "queue
    // in rising order" observable at the dequeue side too.
    let a = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let b = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let c = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };

    // Pin `scheduled_at` to a single literal instant (shared across all
    // three rows — `now()` would yield a different microsecond per UPDATE
    // and make scheduled_at the actual sort key). Reverse `created_at`
    // against insertion order so heap order would give the wrong answer.
    let shared_scheduled = chrono::Utc::now() - chrono::Duration::hours(1);
    sqlx::query(
        "UPDATE crawler_jobs \
            SET scheduled_at = $2, \
                created_at   = $3 \
          WHERE id = $1",
    )
    .bind(a)
    .bind(shared_scheduled)
    .bind(chrono::Utc::now() - chrono::Duration::seconds(10))
    .execute(&pool)
    .await
    .unwrap();
    sqlx::query(
        "UPDATE crawler_jobs \
            SET scheduled_at = $2, \
                created_at   = $3 \
          WHERE id = $1",
    )
    .bind(b)
    .bind(shared_scheduled)
    .bind(chrono::Utc::now() - chrono::Duration::seconds(20))
    .execute(&pool)
    .await
    .unwrap();
    sqlx::query(
        "UPDATE crawler_jobs \
            SET scheduled_at = $2, \
                created_at   = $3 \
          WHERE id = $1",
    )
    .bind(c)
    .bind(shared_scheduled)
    .bind(chrono::Utc::now() - chrono::Duration::seconds(30))
    .execute(&pool)
    .await
    .unwrap();

    let leases = jobs::lease(&pool, None, 10, Duration::from_secs(60))
        .await
        .unwrap();
    let order: Vec<Uuid> = leases.iter().map(|l| l.id).collect();
    assert_eq!(
        order,
        vec![c, b, a],
        "lease must return jobs in created_at order when scheduled_at ties"
    );
}

#[sqlx::test(migrations = "./migrations")]
async fn reap_done_zero_is_a_no_op(pool: PgPool) {
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    sqlx::query("UPDATE crawler_jobs SET state='done', updated_at = now() - interval '999 days' WHERE id = $1")
        .bind(id)
        .execute(&pool)
        .await
        .unwrap();

    let deleted = jobs::reap_done(&pool, 0).await.unwrap();
    assert_eq!(deleted, 0);
    assert_eq!(job_count(&pool).await, 1);
}

/// Migration 0031 ships a pre-dedup step (demote all-but-lowest-id
/// duplicates to `dead`) so it can apply cleanly on dirty production
/// databases where the prior buggy `enqueue_pages` left duplicate
/// `analyze_page` rows in `pending`/`running`. This test simulates the
/// dirty state by dropping the unique index, inserting duplicates, then
/// running the migration's pre-dedup SQL — the same SQL the migration
/// file runs — and confirming the duplicates land in `dead` so the
/// subsequent index creation would succeed.
#[sqlx::test(migrations = "./migrations")]
async fn migration_0031_prededup_demotes_duplicate_analyze_page_rows(pool: PgPool) {
    // 1) Simulate pre-0031 dirty state: drop the dedup index so a second
    //    pending insert for the same page_id is accepted.
    sqlx::query("DROP INDEX IF EXISTS crawler_jobs_analyze_page_dedup_idx")
        .execute(&pool)
        .await
        .unwrap();

    let page_a = Uuid::new_v4();
    let page_b = Uuid::new_v4();
    let make_row = |page_id: Uuid| {
        serde_json::json!({
            "kind": "analyze_page",
            "page_id": page_id,
            "force": false
        })
    };
    // Two pending duplicates for page A, plus one running + one pending
    // for page B (the migration handles all in-flight states).
    let a1: Uuid = sqlx::query_scalar(
        "INSERT INTO crawler_jobs (payload, state) VALUES ($1, 'pending') RETURNING id",
    )
    .bind(make_row(page_a))
    .fetch_one(&pool)
    .await
    .unwrap();
    let a2: Uuid = sqlx::query_scalar(
        "INSERT INTO crawler_jobs (payload, state) VALUES ($1, 'pending') RETURNING id",
    )
    .bind(make_row(page_a))
    .fetch_one(&pool)
    .await
    .unwrap();
    let b1: Uuid = sqlx::query_scalar(
        "INSERT INTO crawler_jobs (payload, state) VALUES ($1, 'running') RETURNING id",
    )
    .bind(make_row(page_b))
    .fetch_one(&pool)
    .await
    .unwrap();
    let b2: Uuid = sqlx::query_scalar(
        "INSERT INTO crawler_jobs (payload, state) VALUES ($1, 'pending') RETURNING id",
    )
    .bind(make_row(page_b))
    .fetch_one(&pool)
    .await
    .unwrap();

    // 2) Run the migration's pre-dedup statement verbatim.
    sqlx::query(
        "UPDATE crawler_jobs \
           SET state = 'dead', \
               last_error = COALESCE(last_error, '') || \
                            CASE WHEN last_error IS NULL OR last_error = '' THEN '' \
                                 ELSE ' | ' END || \
                            'pre-0031 duplicate analyze_page; superseded by earlier sibling', \
               updated_at = now() \
         WHERE id IN ( \
            SELECT id FROM ( \
                SELECT id, \
                       row_number() OVER ( \
                           PARTITION BY payload->>'page_id' \
                           ORDER BY id \
                       ) AS rn \
                  FROM crawler_jobs \
                 WHERE payload->>'kind' = 'analyze_page' \
                   AND state IN ('pending', 'running') \
            ) ranked \
            WHERE rn > 1 \
         )",
    )
    .execute(&pool)
    .await
    .unwrap();

    // 3) The earliest-id row per page_id stays in-flight; later
    //    duplicates moved to `dead` with the curator-recoverable
    //    last_error marker. Compute the expected winner per page by
    //    Postgres' UUID ordering (sql min, not Rust Ord — they differ).
    let winner_a: Uuid =
        sqlx::query_scalar("SELECT LEAST($1::uuid, $2::uuid)")
            .bind(a1)
            .bind(a2)
            .fetch_one(&pool)
            .await
            .unwrap();
    let loser_a = if winner_a == a1 { a2 } else { a1 };
    let winner_b: Uuid =
        sqlx::query_scalar("SELECT LEAST($1::uuid, $2::uuid)")
            .bind(b1)
            .bind(b2)
            .fetch_one(&pool)
            .await
            .unwrap();
    let loser_b = if winner_b == b1 { b2 } else { b1 };

    let state_a_winner: String =
        sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1")
            .bind(winner_a)
            .fetch_one(&pool)
            .await
            .unwrap();
    let state_a_loser: String =
        sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1")
            .bind(loser_a)
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(
        state_a_winner == "pending" || state_a_winner == "running",
        "winner kept in-flight"
    );
    assert_eq!(state_a_loser, "dead", "loser demoted to dead");

    let state_b_winner: String =
        sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1")
            .bind(winner_b)
            .fetch_one(&pool)
            .await
            .unwrap();
    let state_b_loser: String =
        sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1")
            .bind(loser_b)
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(state_b_winner == "pending" || state_b_winner == "running");
    assert_eq!(state_b_loser, "dead");

    // 4) After dedup, the index is creatable. Migration would now succeed.
    sqlx::query(
        "CREATE UNIQUE INDEX crawler_jobs_analyze_page_dedup_idx \
            ON crawler_jobs ((payload->>'page_id')) \
         WHERE state IN ('pending', 'running') \
           AND payload->>'kind' = 'analyze_page'",
    )
    .execute(&pool)
    .await
    .expect("post-dedup index creation must succeed");
}

// ---------------------------------------------------------------------------
// Lease-generation race (0.87.26). Closes the gap that 0.87.20 narrowed:
// `ack_done` / `ack_failed` matched on `(id, state='running')` only, so a
// late ack from a dead lease (id same, generation different) could clobber
// the successor's running row. The new `lease_generation` column scopes
// the guard to the *specific* lease the worker was holding.
// ---------------------------------------------------------------------------

#[sqlx::test(migrations = "./migrations")]
async fn lease_assigns_strictly_increasing_generation_per_release(pool: PgPool) {
    // Each lease (the very first one, AND every re-lease after release /
    // expiry / reclaim) must carry a strictly-increasing generation so
    // downstream guards can distinguish "the lease that was leased" from
    // "the lease that's currently leased."
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let a = jobs::lease(&pool, None, 1, Duration::from_secs(60)).await.unwrap();
    assert_eq!(a.len(), 1);
    assert!(a[0].lease_generation >= 1, "first lease must carry a positive generation");

    jobs::release_unowned(&pool, id).await.unwrap();
    let b = jobs::lease(&pool, None, 1, Duration::from_secs(60)).await.unwrap();
    assert_eq!(b.len(), 1);
    assert!(
        b[0].lease_generation > a[0].lease_generation,
        "successor lease must carry a strictly-higher generation: a={}, b={}",
        a[0].lease_generation,
        b[0].lease_generation
    );
}

#[sqlx::test(migrations = "./migrations")]
async fn ack_done_from_dead_lease_after_release_unowned_is_a_no_op(pool: PgPool) {
    // The race that motivated the lease_generation column. A worker A
    // holds a lease. An external actor (force-analyze, reclaim) calls
    // `release_unowned` on the row, putting it back to `pending`. A
    // worker B re-leases the same id. THEN A's dispatch finally returns
    // and acks done. Before this fix, A's `ack_done` matched
    // `state='running'` (set by B) and clobbered B's lease — A's stale
    // result became the row's `done` payload, B's still-in-flight work
    // was lost. Now A's `ack_done(id, generation=A)` finds nothing
    // because B's lease bumped the generation, so the row stays
    // `running` for B.
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let a = jobs::lease(&pool, None, 1, Duration::from_secs(300))
        .await
        .unwrap()[0]
        .clone();
    assert_eq!(a.id, id);

    // External release: e.g. force-analyze flipping payload.force = true
    // and dropping the in-flight lease so the new payload is picked up.
    jobs::release_unowned(&pool, id).await.unwrap();
    assert_eq!(job_state(&pool, id).await, "pending");

    let b = jobs::lease(&pool, None, 1, Duration::from_secs(300))
        .await
        .unwrap()[0]
        .clone();
    assert_eq!(b.id, id);
    assert_ne!(
        a.lease_generation, b.lease_generation,
        "successor must hold a different lease_generation"
    );
    assert_eq!(job_state(&pool, id).await, "running");

    // A's stale ack arrives. MUST be a no-op — the row stays running
    // with B's progress intact.
    jobs::ack_done(&pool, a.id, a.lease_generation).await.unwrap();
    assert_eq!(
        job_state(&pool, id).await,
        "running",
        "stale ack_done from a dead lease must not clobber the successor's lease"
    );

    // B finishes normally with its own generation.
    jobs::ack_done(&pool, b.id, b.lease_generation).await.unwrap();
    assert_eq!(job_state(&pool, id).await, "done");
}

#[sqlx::test(migrations = "./migrations")]
async fn ack_failed_from_dead_lease_after_release_unowned_is_a_no_op(pool: PgPool) {
    // Mirror of the ack_done case for ack_failed. A late ack_failed
    // from A would otherwise either (a) flip B's running row back to
    // `pending` with a stale error and a stale backoff, or (b) at
    // max_attempts, dead-letter B's in-flight work.
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let a = jobs::lease(&pool, None, 1, Duration::from_secs(300))
        .await
        .unwrap()[0]
        .clone();
    jobs::release_unowned(&pool, id).await.unwrap();
    let b = jobs::lease(&pool, None, 1, Duration::from_secs(300))
        .await
        .unwrap()[0]
        .clone();
    assert_eq!(b.id, id);

    // A's stale ack_failed arrives.
    jobs::ack_failed(&pool, a.id, "A late error", a.attempts, a.max_attempts, a.lease_generation)
        .await
        .unwrap();
    assert_eq!(
        job_state(&pool, id).await,
        "running",
        "stale ack_failed from a dead lease must not clobber the successor"
    );
}

#[sqlx::test(migrations = "./migrations")]
async fn renew_from_dead_lease_is_a_no_op(pool: PgPool) {
    // A's heartbeat must not extend B's lease window after A was
    // released and B re-leased. Without the generation check, an A-side
    // renew would push B's `leased_until` further into the future,
    // delaying reclaim of a B-side crash.
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let a = jobs::lease(&pool, None, 1, Duration::from_secs(60))
        .await
        .unwrap()[0]
        .clone();
    jobs::release_unowned(&pool, id).await.unwrap();
    let _b = jobs::lease(&pool, None, 1, Duration::from_secs(60))
        .await
        .unwrap();

    // A's heartbeat arrives. Should report not-owned and skip.
    let still_owned = jobs::renew(&pool, a.id, a.lease_generation, Duration::from_secs(900))
        .await
        .unwrap();
    assert!(
        !still_owned,
        "renew must report not-owned when the lease_generation no longer matches"
    );
}

#[sqlx::test(migrations = "./migrations")]
async fn release_unowned_bumps_generation_even_when_lease_is_held(pool: PgPool) {
    // Belt-and-braces: release_unowned exists precisely for cases
    // where the caller has no `Lease` in hand (force-analyze, ops
    // tools). It must bump the generation unconditionally so any
    // ack from the in-flight lease is invalidated.
    let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
        .await
        .unwrap()
    {
        EnqueueResult::Inserted(id) => id,
        _ => unreachable!(),
    };
    let a = jobs::lease(&pool, None, 1, Duration::from_secs(300))
        .await
        .unwrap()[0]
        .clone();
    let gen_before: i64 = sqlx::query_scalar(
        "SELECT lease_generation FROM crawler_jobs WHERE id = $1",
    )
    .bind(id)
    .fetch_one(&pool)
    .await
    .unwrap();
    assert_eq!(gen_before, a.lease_generation);

    jobs::release_unowned(&pool, id).await.unwrap();
    let gen_after: i64 = sqlx::query_scalar(
        "SELECT lease_generation FROM crawler_jobs WHERE id = $1",
    )
    .bind(id)
    .fetch_one(&pool)
    .await
    .unwrap();
    assert!(
        gen_after > gen_before,
        "release_unowned must bump lease_generation: before={gen_before}, after={gen_after}"
    );
}