//! Integration tests for the crawler daemon's cron + worker pool. The
//! daemon's full real path requires Chromium and a live source; here we
//! test the seam (MetadataPass / ChapterDispatcher traits) and the
//! cron/worker control-flow.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

use chrono::NaiveTime;
use chrono_tz::Tz;
use mangalord::crawler::content::SyncOutcome;
use mangalord::crawler::daemon::{
    self,
    test_support::{CountingMetadataPass, SlowMetadataPass},
    ChapterDispatcher, DaemonConfig, MetadataPass, CRON_LOCK_KEY,
};
use mangalord::crawler::jobs::{self, JobPayload};
use mangalord::crawler::pipeline;
use serde_json::json;
use sqlx::PgPool;
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

fn far_future_daily_at() -> NaiveTime {
    // Some time hours from "now" so the scheduler sleeps for the whole test.
    NaiveTime::from_hms_opt(23, 59, 0).unwrap()
}

fn make_cfg(
    metadata_pass: Option<Arc<dyn MetadataPass>>,
    dispatcher: Arc<dyn ChapterDispatcher>,
    session_expired: Arc<std::sync::atomic::AtomicBool>,
    workers: usize,
) -> DaemonConfig {
    DaemonConfig {
        metadata_pass,
        dispatcher,
        chapter_workers: workers,
        daily_at: far_future_daily_at(),
        tz: Tz::UTC,
        retention_days: 7,
        metrics_retention_days: 90,
        session_expired,
        status: mangalord::crawler::status::StatusHandle::new(workers),
        job_timeout: Duration::from_secs(60),
        extra_tasks: Vec::new(),
    }
}

async fn enqueue_chapter_job(pool: &PgPool) -> Uuid {
    let chapter_id = Uuid::new_v4();
    let payload = JobPayload::SyncChapterContent {
        source_id: "target".into(),
        chapter_id,
        source_chapter_key: format!("ch-{chapter_id}"),
    };
    let res = jobs::enqueue(pool, &payload).await.unwrap();
    match res {
        jobs::EnqueueResult::Inserted(_) => chapter_id,
        jobs::EnqueueResult::Skipped => unreachable!("fresh chapter_id"),
    }
}

async fn count_state(pool: &PgPool, state: &str) -> i64 {
    sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM crawler_jobs WHERE state = $1")
        .bind(state)
        .fetch_one(pool)
        .await
        .unwrap()
}

struct AlwaysDoneDispatcher {
    seen: AtomicUsize,
}
#[async_trait::async_trait]
impl ChapterDispatcher for AlwaysDoneDispatcher {
    async fn dispatch(&self, _payload: JobPayload) -> anyhow::Result<SyncOutcome> {
        self.seen.fetch_add(1, Ordering::AcqRel);
        Ok(SyncOutcome::Fetched { pages: 1 })
    }
}

struct PanickingDispatcher {
    seen: AtomicUsize,
}
#[async_trait::async_trait]
impl ChapterDispatcher for PanickingDispatcher {
    async fn dispatch(&self, _payload: JobPayload) -> anyhow::Result<SyncOutcome> {
        self.seen.fetch_add(1, Ordering::AcqRel);
        panic!("intentional dispatcher panic");
    }
}

/// Always returns an error — used to drive a job to the dead-letter state.
struct FailingDispatcher {
    seen: AtomicUsize,
}
#[async_trait::async_trait]
impl ChapterDispatcher for FailingDispatcher {
    async fn dispatch(&self, _payload: JobPayload) -> anyhow::Result<SyncOutcome> {
        self.seen.fetch_add(1, Ordering::AcqRel);
        anyhow::bail!("simulated detail-page failure")
    }
}

/// Always reports the browser unavailable — models a Chromium outage where
/// `acquire()` fails. The job must be returned to `pending` WITHOUT burning an
/// attempt, so an outage doesn't chew the backlog to `dead`.
struct BrowserUnavailableDispatcher {
    seen: AtomicUsize,
}
#[async_trait::async_trait]
impl ChapterDispatcher for BrowserUnavailableDispatcher {
    async fn dispatch(&self, _payload: JobPayload) -> anyhow::Result<SyncOutcome> {
        self.seen.fetch_add(1, Ordering::AcqRel);
        Ok(SyncOutcome::BrowserUnavailable)
    }
}

/// Never completes — used to verify the worker's outer dispatch timeout.
struct HangingDispatcher {
    seen: AtomicUsize,
}
#[async_trait::async_trait]
impl ChapterDispatcher for HangingDispatcher {
    async fn dispatch(&self, _payload: JobPayload) -> anyhow::Result<SyncOutcome> {
        self.seen.fetch_add(1, Ordering::AcqRel);
        std::future::pending::<()>().await;
        unreachable!("hanging dispatcher never resolves");
    }
}

#[sqlx::test(migrations = "./migrations")]
async fn worker_times_out_a_hung_dispatch_and_acks_failed(pool: PgPool) {
    enqueue_chapter_job(&pool).await;
    let dispatcher = Arc::new(HangingDispatcher {
        seen: AtomicUsize::new(0),
    });
    let session_expired = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let cancel = CancellationToken::new();
    let mut cfg = make_cfg(None, dispatcher.clone(), session_expired, 1);
    cfg.job_timeout = Duration::from_millis(300);
    let handle = daemon::spawn(pool.clone(), cancel.clone(), cfg);

    // The hung job should time out and return to pending with backoff
    // (attempts=1 < max=5). Poll for the recorded error.
    let mut timed_out = false;
    for _ in 0..40 {
        let n: i64 = sqlx::query_scalar(
            "SELECT COUNT(*) FROM crawler_jobs WHERE last_error = 'dispatch timed out'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        if n == 1 {
            timed_out = true;
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    handle.shutdown().await;
    assert!(timed_out, "hung dispatch must be acked failed with a timeout error");
    assert!(dispatcher.seen.load(Ordering::Acquire) >= 1);
}

#[sqlx::test(migrations = "./migrations")]
async fn shutdown_mid_dispatch_releases_lease_without_burning_attempt(pool: PgPool) {
    // A worker that is mid-dispatch when shutdown is signalled must return
    // its job to `pending` with the attempt refunded (attempts back to 0),
    // rather than leaving it `running` until lease expiry with a burned
    // attempt. Uses a never-completing dispatcher so the job is provably
    // in-flight when we cancel.
    enqueue_chapter_job(&pool).await;
    let dispatcher = Arc::new(HangingDispatcher {
        seen: AtomicUsize::new(0),
    });
    let session_expired = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let cancel = CancellationToken::new();
    let mut cfg = make_cfg(None, dispatcher.clone(), session_expired, 1);
    // Long timeout so the dispatch can't be acked-failed by the outer
    // timeout before we cancel.
    cfg.job_timeout = Duration::from_secs(60);
    let handle = daemon::spawn(pool.clone(), cancel.clone(), cfg);

    // Wait until the job is actually leased (running) and the dispatch has
    // entered the hanging future.
    let mut entered = false;
    for _ in 0..40 {
        if dispatcher.seen.load(Ordering::Acquire) >= 1 && count_state(&pool, "running").await == 1
        {
            entered = true;
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    assert!(entered, "dispatch must be in-flight (running) before shutdown");

    handle.shutdown().await;

    assert_eq!(
        count_state(&pool, "running").await,
        0,
        "no job left running after graceful shutdown"
    );
    assert_eq!(
        count_state(&pool, "pending").await,
        1,
        "the in-flight job is released back to pending"
    );
    let attempts: i32 = sqlx::query_scalar("SELECT attempts FROM crawler_jobs")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(
        attempts, 0,
        "graceful shutdown must refund the lease attempt (no burn)"
    );
}

#[sqlx::test(migrations = "./migrations")]
async fn browser_unavailable_releases_lease_without_burning_attempt(pool: PgPool) {
    // During a browser outage the dispatcher reports BrowserUnavailable. The
    // worker must return the job to `pending` with the attempt refunded
    // (attempts stays 0) instead of ack-failing it toward `dead`, so the whole
    // pending backlog survives the outage.
    enqueue_chapter_job(&pool).await;
    let dispatcher = Arc::new(BrowserUnavailableDispatcher {
        seen: AtomicUsize::new(0),
    });
    let session_expired = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let cancel = CancellationToken::new();
    let handle = daemon::spawn(
        pool.clone(),
        cancel.clone(),
        make_cfg(None, dispatcher.clone(), session_expired, 1),
    );

    // Wait until the dispatcher has been invoked at least once (the job was
    // leased and deferred).
    let mut dispatched = false;
    for _ in 0..40 {
        if dispatcher.seen.load(Ordering::Acquire) >= 1 {
            dispatched = true;
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    assert!(dispatched, "dispatcher must have been invoked");

    handle.shutdown().await;

    assert_eq!(
        count_state(&pool, "dead").await,
        0,
        "browser outage must not dead-letter the job"
    );
    assert_eq!(
        count_state(&pool, "pending").await,
        1,
        "job returns to pending after a browser-unavailable outcome"
    );
    let attempts: i32 = sqlx::query_scalar("SELECT attempts FROM crawler_jobs")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(
        attempts, 0,
        "browser-unavailable must refund the lease attempt (no burn)"
    );
}

#[sqlx::test(migrations = "./migrations")]
async fn workers_drain_jobs_through_dispatcher(pool: PgPool) {
    enqueue_chapter_job(&pool).await;
    enqueue_chapter_job(&pool).await;
    enqueue_chapter_job(&pool).await;

    let dispatcher = Arc::new(AlwaysDoneDispatcher {
        seen: AtomicUsize::new(0),
    });
    let session_expired = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let cancel = CancellationToken::new();
    let handle = daemon::spawn(
        pool.clone(),
        cancel.clone(),
        make_cfg(None, dispatcher.clone(), session_expired, 2),
    );

    // Wait for the workers to drain all three jobs.
    let dispatcher_seen = || dispatcher.seen.load(Ordering::Acquire);
    for _ in 0..40 {
        if dispatcher_seen() >= 3 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    assert!(
        dispatcher_seen() >= 3,
        "expected at least 3 dispatches, got {}",
        dispatcher_seen()
    );

    handle.shutdown().await;
    assert_eq!(count_state(&pool, "done").await, 3);
}

async fn enqueue_sync_manga_job(pool: &PgPool, key: &str) -> Uuid {
    let res = jobs::enqueue(
        pool,
        &JobPayload::SyncManga {
            source_id: "target".into(),
            source_manga_key: key.into(),
            url: format!("http://x/{key}"),
            title: format!("M {key}"),
        },
    )
    .await
    .unwrap();
    match res {
        jobs::EnqueueResult::Inserted(id) => id,
        jobs::EnqueueResult::Skipped => unreachable!("fresh key"),
    }
}

#[sqlx::test(migrations = "./migrations")]
async fn worker_leases_sync_manga_and_failure_dead_letters(pool: PgPool) {
    // The crawl worker must pick up reconcile-enqueued SyncManga jobs (the
    // lease_kinds change), and a failure must dead-letter (leave-dead). Set
    // max_attempts=1 so a single failure is terminal — no waiting through
    // exponential backoff.
    let id = enqueue_sync_manga_job(&pool, "missing-1").await;
    sqlx::query("UPDATE crawler_jobs SET max_attempts = 1 WHERE id = $1")
        .bind(id)
        .execute(&pool)
        .await
        .unwrap();

    let dispatcher = Arc::new(FailingDispatcher {
        seen: AtomicUsize::new(0),
    });
    let session_expired = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let cancel = CancellationToken::new();
    let handle = daemon::spawn(
        pool.clone(),
        cancel.clone(),
        make_cfg(None, dispatcher.clone(), session_expired, 1),
    );

    for _ in 0..60 {
        if count_state(&pool, "dead").await >= 1 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    handle.shutdown().await;

    assert!(
        dispatcher.seen.load(Ordering::Acquire) >= 1,
        "worker must lease the sync_manga job"
    );
    assert_eq!(
        count_state(&pool, "dead").await,
        1,
        "a failed sync_manga job dead-letters (leave-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!(
        last_error
            .unwrap_or_default()
            .contains("simulated detail-page failure"),
        "dead job records the failure reason"
    );
}

#[sqlx::test(migrations = "./migrations")]
async fn workers_idle_while_session_expired(pool: PgPool) {
    let id = enqueue_chapter_job(&pool).await;
    let dispatcher = Arc::new(AlwaysDoneDispatcher {
        seen: AtomicUsize::new(0),
    });
    let session_expired = Arc::new(std::sync::atomic::AtomicBool::new(true));
    let cancel = CancellationToken::new();
    let handle = daemon::spawn(
        pool.clone(),
        cancel.clone(),
        make_cfg(None, dispatcher.clone(), Arc::clone(&session_expired), 1),
    );

    // Wait long enough that a non-idled worker would have leased and ack'd.
    tokio::time::sleep(Duration::from_millis(800)).await;
    assert_eq!(
        dispatcher.seen.load(Ordering::Acquire),
        0,
        "dispatcher must not be invoked while session_expired flag is set"
    );
    assert_eq!(count_state(&pool, "pending").await, 1);
    let _ = id;

    handle.shutdown().await;
}

#[sqlx::test(migrations = "./migrations")]
async fn dispatcher_panic_is_contained_and_job_is_acked_failed(pool: PgPool) {
    enqueue_chapter_job(&pool).await;
    enqueue_chapter_job(&pool).await;

    let dispatcher = Arc::new(PanickingDispatcher {
        seen: AtomicUsize::new(0),
    });
    let session_expired = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let cancel = CancellationToken::new();
    let handle = daemon::spawn(
        pool.clone(),
        cancel.clone(),
        make_cfg(None, dispatcher.clone(), session_expired, 1),
    );

    // Wait for the worker to handle both panicking jobs.
    for _ in 0..40 {
        if dispatcher.seen.load(Ordering::Acquire) >= 2 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    assert!(
        dispatcher.seen.load(Ordering::Acquire) >= 2,
        "worker must keep going after a panic — handled at least 2 jobs"
    );

    handle.shutdown().await;

    // attempts=1 below max=5, so the panicking jobs go back to pending with
    // backoff and `last_error = "worker panicked"`.
    let last_errors: Vec<String> = sqlx::query_scalar(
        "SELECT last_error FROM crawler_jobs WHERE last_error IS NOT NULL",
    )
    .fetch_all(&pool)
    .await
    .unwrap();
    assert_eq!(last_errors.len(), 2);
    assert!(last_errors.iter().all(|e| e == "worker panicked"));
}

#[sqlx::test(migrations = "./migrations")]
async fn cron_skips_tick_when_advisory_lock_held(pool: PgPool) {
    // With no last_metadata_tick_at row, the daemon does a catch-up tick
    // immediately on spawn. We hold the advisory lock on a separate
    // connection beforehand so the catch-up's pg_try_advisory_lock returns
    // false and the tick must skip without invoking the metadata pass.
    let mut lock_conn = pool.acquire().await.unwrap();
    sqlx::query("SELECT pg_advisory_lock($1)")
        .bind(CRON_LOCK_KEY)
        .execute(&mut *lock_conn)
        .await
        .unwrap();

    let counter = Arc::new(CountingMetadataPass::default());
    let dispatcher = Arc::new(AlwaysDoneDispatcher {
        seen: AtomicUsize::new(0),
    });
    let session_expired = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let cancel = CancellationToken::new();
    // daily_at far in the future so after the (skipped) catch-up the
    // cron sleeps for the rest of the test rather than racing for the lock.
    let cfg = make_cfg(
        Some(counter.clone() as Arc<dyn MetadataPass>),
        dispatcher,
        session_expired,
        1,
    );
    let handle = daemon::spawn(pool.clone(), cancel.clone(), cfg);

    tokio::time::sleep(Duration::from_millis(800)).await;
    assert_eq!(
        counter.count.load(Ordering::Acquire),
        0,
        "cron must skip the catch-up tick while the advisory lock is held"
    );

    sqlx::query("SELECT pg_advisory_unlock($1)")
        .bind(CRON_LOCK_KEY)
        .execute(&mut *lock_conn)
        .await
        .unwrap();
    drop(lock_conn);

    handle.shutdown().await;
}

#[sqlx::test(migrations = "./migrations")]
async fn cron_catches_up_when_last_tick_is_stale(pool: PgPool) {
    // Pre-seed last_metadata_tick_at well in the past so previous_fire(now)
    // > last_tick is trivially true and the daemon catches up immediately.
    sqlx::query(
        "INSERT INTO crawler_state (key, value) VALUES ($1, $2)
         ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value",
    )
    .bind("last_metadata_tick_at")
    .bind(json!({"at": "2020-01-01T00:00:00Z"}))
    .execute(&pool)
    .await
    .unwrap();

    let counter = Arc::new(CountingMetadataPass::default());
    let dispatcher = Arc::new(AlwaysDoneDispatcher {
        seen: AtomicUsize::new(0),
    });
    let session_expired = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let cancel = CancellationToken::new();
    let handle = daemon::spawn(
        pool.clone(),
        cancel.clone(),
        make_cfg(
            Some(counter.clone() as Arc<dyn MetadataPass>),
            dispatcher,
            session_expired,
            1,
        ),
    );

    for _ in 0..40 {
        if counter.count.load(Ordering::Acquire) >= 1 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    assert!(
        counter.count.load(Ordering::Acquire) >= 1,
        "catch-up tick should have fired immediately"
    );

    handle.shutdown().await;
}

#[sqlx::test(migrations = "./migrations")]
async fn enqueue_bookmarked_pending_skips_dropped_sources(pool: PgPool) {
    // Setup: one manga with two chapters (page_count = 0). One has a
    // non-dropped source; the other's source is dropped. A user bookmarks
    // the manga. Expectation: only the non-dropped chapter is enqueued.
    let user_id: Uuid = sqlx::query_scalar(
        "INSERT INTO users (username, password_hash) VALUES ($1, $2) RETURNING id",
    )
    .bind("alice")
    .bind("not-a-real-hash")
    .fetch_one(&pool)
    .await
    .unwrap();
    let manga_id: Uuid = sqlx::query_scalar(
        "INSERT INTO mangas (title) VALUES ($1) RETURNING id",
    )
    .bind("Berserk")
    .fetch_one(&pool)
    .await
    .unwrap();
    sqlx::query("INSERT INTO sources (id, name, base_url) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING")
        .bind("target")
        .bind("Target")
        .bind("https://example.com")
        .execute(&pool)
        .await
        .unwrap();
    let c1: Uuid = sqlx::query_scalar(
        "INSERT INTO chapters (manga_id, number, page_count) VALUES ($1, 1, 0) RETURNING id",
    )
    .bind(manga_id)
    .fetch_one(&pool)
    .await
    .unwrap();
    let c2: Uuid = sqlx::query_scalar(
        "INSERT INTO chapters (manga_id, number, page_count) VALUES ($1, 2, 0) RETURNING id",
    )
    .bind(manga_id)
    .fetch_one(&pool)
    .await
    .unwrap();
    // c1: alive source. c2: dropped source.
    sqlx::query(
        "INSERT INTO chapter_sources (source_id, source_chapter_key, chapter_id, source_url) \
         VALUES ($1, $2, $3, $4)",
    )
    .bind("target")
    .bind("ch1")
    .bind(c1)
    .bind("https://example.com/ch1")
    .execute(&pool)
    .await
    .unwrap();
    sqlx::query(
        "INSERT INTO chapter_sources (source_id, source_chapter_key, chapter_id, source_url, dropped_at) \
         VALUES ($1, $2, $3, $4, now())",
    )
    .bind("target")
    .bind("ch2")
    .bind(c2)
    .bind("https://example.com/ch2")
    .execute(&pool)
    .await
    .unwrap();
    sqlx::query("INSERT INTO bookmarks (user_id, manga_id) VALUES ($1, $2)")
        .bind(user_id)
        .bind(manga_id)
        .execute(&pool)
        .await
        .unwrap();

    let summary = pipeline::enqueue_bookmarked_pending(&pool).await.unwrap();
    assert_eq!(summary.inserted, 1, "only the non-dropped chapter enqueued");
    assert_eq!(summary.skipped, 0);
    let payloads: Vec<serde_json::Value> = sqlx::query_scalar(
        "SELECT payload FROM crawler_jobs WHERE payload->>'kind' = 'sync_chapter_content'",
    )
    .fetch_all(&pool)
    .await
    .unwrap();
    assert_eq!(payloads.len(), 1);
    assert_eq!(
        payloads[0]["chapter_id"].as_str().unwrap(),
        c1.to_string()
    );
}

#[sqlx::test(migrations = "./migrations")]
async fn enqueue_bookmarked_pending_skips_recently_dead_chapters(pool: PgPool) {
    // Setup: a chapter whose last SyncChapterContent job died yesterday.
    // The cron tick must not re-enqueue — without the quarantine, the
    // chapter would spin: re-enqueue → max_attempts retries → dies again
    // → re-enqueue next tick → forever.
    let user_id: Uuid = sqlx::query_scalar(
        "INSERT INTO users (username, password_hash) VALUES ($1, $2) RETURNING id",
    )
    .bind("alice")
    .bind("not-a-real-hash")
    .fetch_one(&pool)
    .await
    .unwrap();
    let manga_id: Uuid =
        sqlx::query_scalar("INSERT INTO mangas (title) VALUES ($1) RETURNING id")
            .bind("Test")
            .fetch_one(&pool)
            .await
            .unwrap();
    sqlx::query(
        "INSERT INTO sources (id, name, base_url) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
    )
    .bind("target")
    .bind("Target")
    .bind("https://example.com")
    .execute(&pool)
    .await
    .unwrap();
    let chapter_id: Uuid = sqlx::query_scalar(
        "INSERT INTO chapters (manga_id, number, page_count) VALUES ($1, 1, 0) RETURNING id",
    )
    .bind(manga_id)
    .fetch_one(&pool)
    .await
    .unwrap();
    sqlx::query(
        "INSERT INTO chapter_sources (source_id, source_chapter_key, chapter_id, source_url) \
         VALUES ($1, $2, $3, $4)",
    )
    .bind("target")
    .bind("ch1")
    .bind(chapter_id)
    .bind("https://example.com/ch1")
    .execute(&pool)
    .await
    .unwrap();
    sqlx::query("INSERT INTO bookmarks (user_id, manga_id) VALUES ($1, $2)")
        .bind(user_id)
        .bind(manga_id)
        .execute(&pool)
        .await
        .unwrap();
    // The dead job from the prior tick, updated 1 day ago (well inside the
    // 7-day quarantine window).
    sqlx::query(
        "INSERT INTO crawler_jobs (payload, state, updated_at) \
         VALUES ($1::jsonb, 'dead', now() - interval '1 day')",
    )
    .bind(serde_json::json!({
        "kind": "sync_chapter_content",
        "source_id": "target",
        "chapter_id": chapter_id.to_string(),
        "source_chapter_key": "ch1",
    }))
    .execute(&pool)
    .await
    .unwrap();

    let summary = pipeline::enqueue_bookmarked_pending(&pool).await.unwrap();
    assert_eq!(summary.inserted, 0, "recently dead chapter is quarantined");
    assert_eq!(summary.skipped, 0);
}

#[sqlx::test(migrations = "./migrations")]
async fn enqueue_bookmarked_pending_resumes_after_quarantine_expires(pool: PgPool) {
    // Same setup as above but the dead job is 10 days old — past the
    // 7-day quarantine. The chapter should be re-enqueued so a once-failed
    // chapter eventually gets a second shot at success.
    let user_id: Uuid = sqlx::query_scalar(
        "INSERT INTO users (username, password_hash) VALUES ($1, $2) RETURNING id",
    )
    .bind("alice")
    .bind("not-a-real-hash")
    .fetch_one(&pool)
    .await
    .unwrap();
    let manga_id: Uuid =
        sqlx::query_scalar("INSERT INTO mangas (title) VALUES ($1) RETURNING id")
            .bind("Test")
            .fetch_one(&pool)
            .await
            .unwrap();
    sqlx::query(
        "INSERT INTO sources (id, name, base_url) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
    )
    .bind("target")
    .bind("Target")
    .bind("https://example.com")
    .execute(&pool)
    .await
    .unwrap();
    let chapter_id: Uuid = sqlx::query_scalar(
        "INSERT INTO chapters (manga_id, number, page_count) VALUES ($1, 1, 0) RETURNING id",
    )
    .bind(manga_id)
    .fetch_one(&pool)
    .await
    .unwrap();
    sqlx::query(
        "INSERT INTO chapter_sources (source_id, source_chapter_key, chapter_id, source_url) \
         VALUES ($1, $2, $3, $4)",
    )
    .bind("target")
    .bind("ch1")
    .bind(chapter_id)
    .bind("https://example.com/ch1")
    .execute(&pool)
    .await
    .unwrap();
    sqlx::query("INSERT INTO bookmarks (user_id, manga_id) VALUES ($1, $2)")
        .bind(user_id)
        .bind(manga_id)
        .execute(&pool)
        .await
        .unwrap();
    sqlx::query(
        "INSERT INTO crawler_jobs (payload, state, updated_at) \
         VALUES ($1::jsonb, 'dead', now() - interval '10 days')",
    )
    .bind(serde_json::json!({
        "kind": "sync_chapter_content",
        "source_id": "target",
        "chapter_id": chapter_id.to_string(),
        "source_chapter_key": "ch1",
    }))
    .execute(&pool)
    .await
    .unwrap();

    let summary = pipeline::enqueue_bookmarked_pending(&pool).await.unwrap();
    assert_eq!(
        summary.inserted, 1,
        "dead chapter is re-enqueued after quarantine expires"
    );
}

/// Helper: insert a chapter with the given `number` and a non-dropped
/// source row, returning the chapter id. Used by the ordering tests so
/// the setup boilerplate doesn't drown the assertion.
async fn insert_pending_chapter(
    pool: &PgPool,
    manga_id: Uuid,
    number: i32,
    source_chapter_key: &str,
) -> Uuid {
    let chapter_id: Uuid = sqlx::query_scalar(
        "INSERT INTO chapters (manga_id, number, page_count) VALUES ($1, $2, 0) RETURNING id",
    )
    .bind(manga_id)
    .bind(number)
    .fetch_one(pool)
    .await
    .unwrap();
    sqlx::query(
        "INSERT INTO chapter_sources (source_id, source_chapter_key, chapter_id, source_url) \
         VALUES ($1, $2, $3, $4)",
    )
    .bind("target")
    .bind(source_chapter_key)
    .bind(chapter_id)
    .bind(format!("https://example.com/{source_chapter_key}"))
    .execute(pool)
    .await
    .unwrap();
    chapter_id
}

#[sqlx::test(migrations = "./migrations")]
async fn enqueue_bookmarked_pending_queues_chapters_in_ascending_number_order(pool: PgPool) {
    // Insert chapters with `number` values 3, 1, 2 in that insertion
    // order — so `created_at` order (the previous tiebreaker) does NOT
    // match number order. After enqueue + lease, the worker should see
    // chapters 1, 2, 3 in that sequence.
    let user_id: Uuid = sqlx::query_scalar(
        "INSERT INTO users (username, password_hash) VALUES ($1, $2) RETURNING id",
    )
    .bind("alice")
    .bind("not-a-real-hash")
    .fetch_one(&pool)
    .await
    .unwrap();
    let manga_id: Uuid = sqlx::query_scalar("INSERT INTO mangas (title) VALUES ($1) RETURNING id")
        .bind("Test")
        .fetch_one(&pool)
        .await
        .unwrap();
    sqlx::query(
        "INSERT INTO sources (id, name, base_url) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
    )
    .bind("target")
    .bind("Target")
    .bind("https://example.com")
    .execute(&pool)
    .await
    .unwrap();
    let c3 = insert_pending_chapter(&pool, manga_id, 3, "ch3").await;
    let c1 = insert_pending_chapter(&pool, manga_id, 1, "ch1").await;
    let c2 = insert_pending_chapter(&pool, manga_id, 2, "ch2").await;
    sqlx::query("INSERT INTO bookmarks (user_id, manga_id) VALUES ($1, $2)")
        .bind(user_id)
        .bind(manga_id)
        .execute(&pool)
        .await
        .unwrap();

    let summary = pipeline::enqueue_bookmarked_pending(&pool).await.unwrap();
    assert_eq!(summary.inserted, 3);

    let leases = jobs::lease(&pool, None, 10, std::time::Duration::from_secs(60))
        .await
        .unwrap();
    let leased_chapter_ids: Vec<Uuid> = leases
        .iter()
        .map(|l| match &l.payload {
            JobPayload::SyncChapterContent { chapter_id, .. } => *chapter_id,
            other => panic!("unexpected payload kind: {other:?}"),
        })
        .collect();
    assert_eq!(
        leased_chapter_ids,
        vec![c1, c2, c3],
        "chapters must be leased in ascending chapter-number order, not insertion order"
    );
}

#[sqlx::test(migrations = "./migrations")]
async fn enqueue_pending_for_manga_queues_chapters_in_ascending_number_order(pool: PgPool) {
    // Same scenario as above but exercising the bookmark-create hook path
    // (`enqueue_pending_for_manga`) which has its own ORDER BY.
    let manga_id: Uuid = sqlx::query_scalar("INSERT INTO mangas (title) VALUES ($1) RETURNING id")
        .bind("Test")
        .fetch_one(&pool)
        .await
        .unwrap();
    sqlx::query(
        "INSERT INTO sources (id, name, base_url) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
    )
    .bind("target")
    .bind("Target")
    .bind("https://example.com")
    .execute(&pool)
    .await
    .unwrap();
    let c3 = insert_pending_chapter(&pool, manga_id, 3, "ch3").await;
    let c1 = insert_pending_chapter(&pool, manga_id, 1, "ch1").await;
    let c2 = insert_pending_chapter(&pool, manga_id, 2, "ch2").await;

    let summary = pipeline::enqueue_pending_for_manga(&pool, manga_id)
        .await
        .unwrap();
    assert_eq!(summary.inserted, 3);

    let leases = jobs::lease(&pool, None, 10, std::time::Duration::from_secs(60))
        .await
        .unwrap();
    let leased_chapter_ids: Vec<Uuid> = leases
        .iter()
        .map(|l| match &l.payload {
            JobPayload::SyncChapterContent { chapter_id, .. } => *chapter_id,
            other => panic!("unexpected payload kind: {other:?}"),
        })
        .collect();
    assert_eq!(leased_chapter_ids, vec![c1, c2, c3]);
}


/// Shutdown must race the in-flight metadata pass against cancellation,
/// not wait for the pass to drain. Without this, a long catalog walk
/// (minutes) wedges `DaemonHandle::shutdown`, `Supervisors::reload_crawler`
/// (under the supervisor lock), and SIGTERM responsiveness.
#[sqlx::test(migrations = "./migrations")]
async fn cron_shutdown_does_not_block_on_in_flight_metadata_pass(pool: PgPool) {
    // Pre-seed a stale tick so the daemon does its catch-up tick at spawn
    // — that immediately drops us into `metadata.run().await`, which our
    // SlowMetadataPass will then sit on for 30s.
    sqlx::query(
        "INSERT INTO crawler_state (key, value) VALUES ($1, $2)
         ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value",
    )
    .bind("last_metadata_tick_at")
    .bind(json!({"at": "2020-01-01T00:00:00Z"}))
    .execute(&pool)
    .await
    .unwrap();

    let slow = SlowMetadataPass::new(Duration::from_secs(30));
    let dispatcher = Arc::new(AlwaysDoneDispatcher {
        seen: AtomicUsize::new(0),
    });
    let session_expired = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let cancel = CancellationToken::new();
    let handle = daemon::spawn(
        pool.clone(),
        cancel.clone(),
        make_cfg(
            Some(slow.clone() as Arc<dyn MetadataPass>),
            dispatcher,
            session_expired,
            1,
        ),
    );

    // Wait for the catch-up to enter `metadata.run` so we're sure to be
    // observing cancellation mid-pass.
    let deadline = std::time::Instant::now() + Duration::from_secs(3);
    while slow.start_count() == 0 && std::time::Instant::now() < deadline {
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    assert_eq!(
        slow.start_count(),
        1,
        "metadata pass should be in flight when we cancel"
    );

    // Cancellation must drop the pass future and return promptly — under
    // the 30s slow-pass deadline by a wide margin.
    let started = std::time::Instant::now();
    tokio::time::timeout(Duration::from_secs(5), handle.shutdown())
        .await
        .expect("shutdown must observe cancel and return promptly");
    assert!(
        started.elapsed() < Duration::from_secs(5),
        "shutdown took too long; cancel didn't propagate into the cron tick body"
    );

    // The advisory lock must be released so another replica's cron can
    // pick up the slot. The 0.87.4 fix runs `pg_advisory_unlock`
    // UNCONDITIONALLY below the `tokio::select!` — both the
    // cancel-arm and the body-completed paths converge there. Without
    // this assertion, a refactor that moved the unlock INTO the
    // body-completed arm only (skipping the cancel path) would ship
    // green under shutdown-latency-only coverage. `pg_try_advisory_lock`
    // returns true iff the lock is free, so we use it as the probe.
    let acquired: bool = sqlx::query_scalar("SELECT pg_try_advisory_lock($1)")
        .bind(CRON_LOCK_KEY)
        .fetch_one(&pool)
        .await
        .unwrap();
    assert!(
        acquired,
        "advisory lock must be released after cancel-during-tick \
         (otherwise multi-replica safety breaks)"
    );
    // Be a good citizen for whatever sqlx test runs next.
    let _ = sqlx::query("SELECT pg_advisory_unlock($1)")
        .bind(CRON_LOCK_KEY)
        .execute(&pool)
        .await;
}