//! In-process crawler daemon.
//!
//! Owns a cron task that fires a daily metadata pass and N worker tasks
//! that drain `SyncChapterContent` jobs from `crawler_jobs`. The dispatch
//! seams ([`MetadataPass`], [`ChapterDispatcher`]) are traits so tests can
//! inject stubs without standing up a real Chromium / `Source` impl.
//!
//! ## Cron
//!
//! Each tick:
//!   1. Acquire a Postgres advisory lock on a dedicated pool connection
//!      (multi-replica safety). Skip the tick on contention.
//!   2. Call [`MetadataPass::run`] (typically `pipeline::run_metadata_pass`).
//!   3. Enqueue `SyncChapterContent` jobs for any bookmarked manga whose
//!      chapters still have `page_count = 0`.
//!   4. Reap `done` jobs older than `retention_days`.
//!   5. Persist `last_metadata_tick_at` and release the lock.
//!
//! If the last persisted tick is older than the most recent scheduled slot
//! (e.g. backend was down at midnight), the daemon fires immediately on
//! startup before resuming the regular schedule.
//!
//! ## Workers
//!
//! Each worker leases one chapter-content job at a time, dispatches via the
//! [`ChapterDispatcher`], and acks `done` / `failed` / re-`pending` based on
//! the outcome. A `SessionExpired` outcome flips the sticky
//! `session_expired` flag — all workers idle while it's set (until operator
//! restart with a refreshed PHPSESSID).
//!
//! Worker dispatch is wrapped in `catch_unwind` so a panicking handler
//! marks the job failed instead of taking down the worker task.

use std::panic::AssertUnwindSafe;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use chrono::{DateTime, Datelike, NaiveTime, TimeZone, Timelike, Utc};
use chrono_tz::Tz;
use futures_util::FutureExt;
use serde_json::json;
use sqlx::PgPool;
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;

use crate::crawler::content::SyncOutcome;
use crate::crawler::jobs::{self, JobPayload, Lease, KIND_SYNC_CHAPTER_CONTENT, KIND_SYNC_MANGA};
use crate::crawler::pipeline;
use crate::crawler::status::{Phase, StatusHandle};

/// Fixed `pg_try_advisory_lock` key. ASCII "MANGALRD" interpreted as a
/// big-endian i64. Hardcoded so every replica agrees on the lock identity
/// without consulting config.
pub const CRON_LOCK_KEY: i64 = 0x4D414E47414C5244;

const STATE_KEY_LAST_TICK: &str = "last_metadata_tick_at";

/// Lease window handed to `jobs::lease`. Kept short, but continuously
/// extended by the per-job heartbeat (see [`WorkerContext::process_lease`])
/// so a long-but-healthy job never lapses and gets stolen.
const LEASE_DURATION: Duration = Duration::from_secs(60);

/// How often the heartbeat renews the lease while a job runs. A third of
/// the lease window leaves two missed-beat's slack before expiry.
const LEASE_HEARTBEAT: Duration = Duration::from_secs(20);

#[async_trait]
pub trait MetadataPass: Send + Sync {
    async fn run(&self) -> anyhow::Result<pipeline::MetadataStats>;
}

#[async_trait]
pub trait ChapterDispatcher: Send + Sync {
    async fn dispatch(&self, payload: JobPayload) -> anyhow::Result<SyncOutcome>;
}

/// A full list-only walk that enqueues mangas missing from the DB as
/// `SyncManga` jobs. Triggered on demand from the admin `/reconcile`
/// endpoint (not on the cron). Mirrors [`MetadataPass`] so the endpoint can
/// hold a `dyn` and tests can stub it.
#[async_trait]
pub trait ReconcilePass: Send + Sync {
    async fn run(&self) -> anyhow::Result<crate::crawler::reconcile::ReconcileStats>;
}

/// Configuration for [`spawn`]. Use `None` for `metadata_pass` to disable
/// the cron entirely (worker-pool-only mode — useful when only the
/// bookmark-triggered enqueue path is wanted).
pub struct DaemonConfig {
    pub metadata_pass: Option<Arc<dyn MetadataPass>>,
    pub dispatcher: Arc<dyn ChapterDispatcher>,
    pub chapter_workers: usize,
    pub daily_at: NaiveTime,
    pub tz: Tz,
    pub retention_days: u32,
    pub metrics_retention_days: u32,
    pub session_expired: Arc<AtomicBool>,
    /// Live status surface updated by the cron + workers.
    pub status: StatusHandle,
    /// Hard upper bound on a single job's dispatch. A job that exceeds it
    /// is acked failed (exponential backoff) rather than wedging a worker
    /// forever. Must exceed [`LEASE_HEARTBEAT`] and the realistic
    /// single-job runtime.
    pub job_timeout: Duration,
    /// Tasks that should run alongside the cron + workers and be cancelled
    /// on shutdown. Used to hand the daemon ownership of the browser
    /// manager's idle reaper.
    pub extra_tasks: Vec<tokio::task::JoinHandle<()>>,
}

pub struct DaemonHandle {
    cancel: CancellationToken,
    join: JoinSet<()>,
    extra: Vec<tokio::task::JoinHandle<()>>,
}

impl DaemonHandle {
    /// Trigger shutdown and await all worker / cron / extra tasks.
    pub async fn shutdown(mut self) {
        self.cancel.cancel();
        while self.join.join_next().await.is_some() {}
        for task in self.extra.drain(..) {
            let _ = task.await;
        }
    }

    /// Cancellation token that drives shutdown — exposed so callers
    /// (`app::spawn_crawler_daemon`) can hand the same token to auxiliary
    /// tasks (e.g. the BrowserManager idle reaper) and have them stop on
    /// the daemon's signal.
    pub fn cancel_token(&self) -> CancellationToken {
        self.cancel.clone()
    }
}

/// Spawn the daemon. Returns immediately; tasks run in the background.
/// Pass an external [`CancellationToken`] so auxiliary tasks (e.g. a
/// BrowserManager idle reaper) can share the same shutdown signal —
/// typically created in the caller, cloned into both spawns.
pub fn spawn(pool: PgPool, cancel: CancellationToken, cfg: DaemonConfig) -> DaemonHandle {
    let mut join = JoinSet::new();

    let DaemonConfig {
        metadata_pass,
        dispatcher,
        chapter_workers,
        daily_at,
        tz,
        retention_days,
        metrics_retention_days,
        session_expired,
        status,
        job_timeout,
        extra_tasks,
    } = cfg;

    if let Some(metadata) = metadata_pass {
        let ctx = CronContext {
            pool: pool.clone(),
            cancel: cancel.clone(),
            daily_at,
            tz,
            retention_days,
            metrics_retention_days,
            metadata,
            status: status.clone(),
        };
        join.spawn(async move { ctx.run().await });
    } else {
        tracing::info!("crawler daemon: no metadata_pass — cron disabled");
    }

    for worker_id in 0..chapter_workers.max(1) {
        let ctx = WorkerContext {
            pool: pool.clone(),
            cancel: cancel.clone(),
            dispatcher: Arc::clone(&dispatcher),
            session_expired: Arc::clone(&session_expired),
            status: status.clone(),
            job_timeout,
            id: worker_id,
        };
        join.spawn(async move { ctx.run().await });
    }

    DaemonHandle {
        cancel,
        join,
        extra: extra_tasks,
    }
}

// ---------------------------------------------------------------------------
// Cron
// ---------------------------------------------------------------------------

struct CronContext {
    pool: PgPool,
    cancel: CancellationToken,
    daily_at: NaiveTime,
    tz: Tz,
    retention_days: u32,
    metrics_retention_days: u32,
    metadata: Arc<dyn MetadataPass>,
    status: StatusHandle,
}

impl CronContext {
    async fn run(self) {
        // On startup, fire immediately if the most recent slot has already
        // passed and we never recorded a tick for it.
        let now = Utc::now();
        let mut catchup = match read_last_tick(&self.pool).await {
            Ok(Some(last)) => previous_fire(now, self.daily_at, self.tz) > last,
            Ok(None) => true,
            Err(e) => {
                tracing::warn!(?e, "cron: read_last_tick failed; assuming no catch-up");
                false
            }
        };

        loop {
            if catchup {
                tracing::info!("cron: catch-up tick (missed scheduled slot)");
                self.run_tick().await;
                catchup = false;
                continue;
            }
            // Recompute next-fire from now() each iteration so clock jumps
            // (NTP step, suspend/resume) don't strand us on a stale instant.
            let next = next_fire(Utc::now(), self.daily_at, self.tz);
            let wait = (next - Utc::now()).to_std().unwrap_or(Duration::ZERO);
            self.status
                .set_phase(Phase::Idle {
                    next_fire: Some(next),
                })
                .await;
            tracing::info!(
                next_fire_utc = %next.to_rfc3339(),
                wait_seconds = wait.as_secs(),
                "cron: sleeping until next slot"
            );
            tokio::select! {
                _ = tokio::time::sleep(wait) => {}
                _ = self.cancel.cancelled() => {
                    tracing::info!("cron: shutdown");
                    return;
                }
            }
            self.run_tick().await;
        }
    }

    async fn run_tick(&self) {
        let mut conn = match self.pool.acquire().await {
            Ok(c) => c,
            Err(e) => {
                tracing::error!(?e, "cron: acquire conn failed; skipping tick");
                return;
            }
        };
        // pg_try_advisory_lock is session-scoped — we must hold the same
        // connection for the unlock or the call silently no-ops on a
        // different connection from the pool.
        let acquired: bool = sqlx::query_scalar("SELECT pg_try_advisory_lock($1)")
            .bind(CRON_LOCK_KEY)
            .fetch_one(&mut *conn)
            .await
            .unwrap_or(false);
        if !acquired {
            tracing::info!("cron: tick skipped — another replica holds the lock");
            return;
        }

        // Panic-isolate the tick body the same way `process_lease` does
        // for worker dispatch. Without this, a panic in metadata.run
        // (or any of the follow-on steps) would kill the cron task and
        // no future tick would ever run — workers would keep going but
        // no new metadata work would be scheduled until daemon restart.
        // The advisory unlock below runs unconditionally so a panicked
        // (or cancelled) tick doesn't leave the lock held for another
        // replica.
        let metadata = &self.metadata;
        let pool = &self.pool;
        let retention_days = self.retention_days;
        let metrics_retention_days = self.metrics_retention_days;
        let status = &self.status;
        let body = async move {
            match metadata.run().await {
                Ok(stats) => {
                    status.record_pass(&stats, Utc::now()).await;
                    tracing::info!(?stats, "cron: metadata pass done");
                }
                Err(e) => tracing::error!(?e, "cron: metadata pass failed"),
            }
            match pipeline::enqueue_bookmarked_pending(pool).await {
                Ok(summary) => {
                    tracing::info!(?summary, "cron: enqueued bookmarked-pending");
                }
                Err(e) => tracing::error!(?e, "cron: enqueue_bookmarked_pending failed"),
            }
            match jobs::reap_terminal(pool, retention_days).await {
                Ok(n) => tracing::info!(reaped = n, "cron: terminal-job reaper finished"),
                Err(e) => tracing::error!(?e, "cron: terminal-job reaper failed"),
            }
            match crate::repo::crawl_metrics::reap(pool, metrics_retention_days).await {
                Ok(n) => tracing::info!(reaped = n, "cron: crawl-metrics reaper finished"),
                Err(e) => tracing::error!(?e, "cron: crawl-metrics reaper failed"),
            }
            if let Err(e) = write_last_tick(pool, Utc::now()).await {
                tracing::warn!(?e, "cron: persist last_metadata_tick_at failed");
            }
        };
        // Race the tick body against shutdown. Without this, a long
        // metadata pass on a fresh catalog (minutes) wedges
        // `DaemonHandle::shutdown`, `Supervisors::reload_crawler` (under
        // the supervisor lock), and SIGTERM responsiveness for the
        // entire pass. On cancel, dropping `body` cascades to dropping
        // `metadata.run()`, which is what gives Chromium/lease cleanup
        // their chance via the Browse Manager's idle reaper.
        tokio::select! {
            biased;
            _ = self.cancel.cancelled() => {
                tracing::info!("cron: cancelled mid-tick — abandoning to release advisory lock");
            }
            r = AssertUnwindSafe(body).catch_unwind() => {
                if r.is_err() {
                    tracing::error!("cron: tick body panicked — continuing");
                }
            }
        }

        let _ = sqlx::query("SELECT pg_advisory_unlock($1)")
            .bind(CRON_LOCK_KEY)
            .execute(&mut *conn)
            .await;
        drop(conn);
    }
}

// ---------------------------------------------------------------------------
// Workers
// ---------------------------------------------------------------------------

struct WorkerContext {
    pool: PgPool,
    cancel: CancellationToken,
    dispatcher: Arc<dyn ChapterDispatcher>,
    session_expired: Arc<AtomicBool>,
    status: StatusHandle,
    job_timeout: Duration,
    id: usize,
}

impl WorkerContext {
    async fn run(self) {
        loop {
            if self.cancel.is_cancelled() {
                tracing::info!(worker = self.id, "worker: shutdown");
                return;
            }
            if self.session_expired.load(Ordering::Acquire) {
                tokio::select! {
                    _ = tokio::time::sleep(Duration::from_secs(30)) => continue,
                    _ = self.cancel.cancelled() => return,
                }
            }
            let leases = match jobs::lease_kinds(
                &self.pool,
                &[KIND_SYNC_CHAPTER_CONTENT, KIND_SYNC_MANGA],
                1,
                LEASE_DURATION,
            )
            .await
            {
                Ok(v) => v,
                Err(e) => {
                    tracing::warn!(worker = self.id, ?e, "worker: lease failed");
                    tokio::select! {
                        _ = tokio::time::sleep(Duration::from_secs(5)) => continue,
                        _ = self.cancel.cancelled() => return,
                    }
                }
            };
            let Some(lease) = leases.into_iter().next() else {
                tokio::select! {
                    _ = tokio::time::sleep(Duration::from_secs(1)) => continue,
                    _ = self.cancel.cancelled() => return,
                }
            };
            self.process_lease(lease).await;
        }
    }

    async fn process_lease(&self, lease: Lease) {
        // Consumer-side dedup safety net: if the chapter already has pages
        // (because a force-refetch race or a job that was re-enqueued
        // after a previous one finished), ack done without re-fetching.
        if let JobPayload::SyncChapterContent { chapter_id, .. } = &lease.payload {
            let page_count = crate::repo::chapter::page_count(&self.pool, *chapter_id)
                .await
                .ok()
                .flatten();
            if matches!(page_count, Some(n) if n > 0) {
                let _ = jobs::ack_done(&self.pool, lease.id, lease.lease_generation).await;
                return;
            }
        }

        // Heartbeat: keep the lease fresh while the (potentially long)
        // dispatch runs, so a slow-but-healthy job is never re-leased and
        // never inflates `attempts` toward `max_attempts`. Stops itself
        // once the job is no longer ours (renew returns false).
        let heartbeat = {
            let hb_pool = self.pool.clone();
            let hb_id = lease.id;
            let hb_gen = lease.lease_generation;
            tokio::spawn(async move {
                loop {
                    tokio::time::sleep(LEASE_HEARTBEAT).await;
                    match jobs::renew(&hb_pool, hb_id, hb_gen, LEASE_DURATION).await {
                        Ok(true) => {}
                        Ok(false) => break,
                        Err(e) => {
                            tracing::warn!(lease_id = %hb_id, ?e, "heartbeat renew failed");
                        }
                    }
                }
            })
        };

        // The "currently crawling" chapter (with its live page count) is
        // registered by the dispatcher itself (RealChapterDispatcher) so it
        // carries the manga/chapter identity + page progress and is removed
        // via an RAII guard on every exit path.

        // Outer timeout: a dispatch that exceeds `job_timeout` is acked
        // failed (exponential backoff) rather than wedging the worker.
        let dispatch = AssertUnwindSafe(self.dispatcher.dispatch(lease.payload.clone()))
            .catch_unwind();
        let outcome = tokio::select! {
            // `biased` so a shutdown that arrives while the dispatch is also
            // ready still takes the cancel arm and releases the lease.
            biased;
            _ = self.cancel.cancelled() => {
                // Graceful shutdown mid-dispatch: drop the in-flight dispatch
                // future (cancelling its browser work) and return the job to
                // `pending` WITHOUT burning a retry attempt, so a clean
                // restart doesn't march healthy jobs toward `max_attempts`.
                // `release` refunds the lease's `attempts` increment, mirroring
                // the session-expired path. Previously the worker had no cancel
                // branch here, so a mid-dispatch SIGTERM left the row `running`
                // until lease expiry and cost one attempt.
                heartbeat.abort();
                let _ = jobs::release(&self.pool, lease.id, lease.lease_generation).await;
                tracing::info!(
                    worker = self.id,
                    lease_id = %lease.id,
                    "worker: shutdown mid-dispatch — released lease without burning an attempt"
                );
                return;
            }
            o = tokio::time::timeout(self.job_timeout, dispatch) => o,
        };
        heartbeat.abort();

        let outcome = match outcome {
            Ok(o) => o,
            Err(_elapsed) => {
                tracing::warn!(
                    worker = self.id,
                    lease_id = %lease.id,
                    timeout_secs = self.job_timeout.as_secs(),
                    "worker: dispatch timed out — ack failed"
                );
                let _ = jobs::ack_failed(
                    &self.pool,
                    lease.id,
                    "dispatch timed out",
                    lease.attempts,
                    lease.max_attempts,
                    lease.lease_generation,
                )
                .await;
                return;
            }
        };
        match outcome {
            Ok(Ok(SyncOutcome::Fetched { .. } | SyncOutcome::Skipped)) => {
                let _ = jobs::ack_done(&self.pool, lease.id, lease.lease_generation).await;
            }
            Ok(Ok(SyncOutcome::SessionExpired)) => {
                tracing::error!(
                    worker = self.id,
                    lease_id = %lease.id,
                    "session expired — workers will idle until restart"
                );
                self.session_expired.store(true, Ordering::Release);
                // Push the session-expired flip to live status subscribers.
                self.status.poke();
                let _ = jobs::release(&self.pool, lease.id, lease.lease_generation).await;
            }
            Ok(Err(e)) => {
                tracing::warn!(
                    worker = self.id,
                    lease_id = %lease.id,
                    error = ?e,
                    "worker: dispatch error — ack failed"
                );
                let _ = jobs::ack_failed(
                    &self.pool,
                    lease.id,
                    &format!("{e:#}"),
                    lease.attempts,
                    lease.max_attempts,
                    lease.lease_generation,
                )
                .await;
            }
            Err(_panic) => {
                tracing::error!(
                    worker = self.id,
                    lease_id = %lease.id,
                    "worker: dispatcher panicked — ack failed"
                );
                let _ = jobs::ack_failed(
                    &self.pool,
                    lease.id,
                    "worker panicked",
                    lease.attempts,
                    lease.max_attempts,
                    lease.lease_generation,
                )
                .await;
            }
        }
    }
}

// ---------------------------------------------------------------------------
// Cron timing primitives
// ---------------------------------------------------------------------------

/// Compute the next UTC instant when `daily_at` (interpreted in `tz`) will
/// fire, strictly after `now`. Handles DST gaps (spring-forward) by
/// advancing past the gap; on DST overlap (fall-back) picks the later
/// instant so the job runs once, not twice.
pub fn next_fire(now: DateTime<Utc>, daily_at: NaiveTime, tz: Tz) -> DateTime<Utc> {
    let now_local = now.with_timezone(&tz);
    // Start with today's slot in the local TZ.
    let mut candidate = local_at(now_local.date_naive(), daily_at, tz);
    // If today's slot is in the past (or now), roll forward day-by-day.
    while candidate <= now {
        let next_day = candidate
            .with_timezone(&tz)
            .date_naive()
            .succ_opt()
            .unwrap_or_else(|| {
                // Defensive: succ_opt only fails at chrono's max date.
                chrono::NaiveDate::from_ymd_opt(
                    candidate.year(),
                    candidate.month(),
                    candidate.day(),
                )
                .expect("valid date")
            });
        candidate = local_at(next_day, daily_at, tz);
    }
    candidate
}

/// The most recent fire instant at or before `now`. Used to detect missed
/// slots after a restart.
pub fn previous_fire(now: DateTime<Utc>, daily_at: NaiveTime, tz: Tz) -> DateTime<Utc> {
    let now_local = now.with_timezone(&tz);
    let today = local_at(now_local.date_naive(), daily_at, tz);
    if today <= now {
        return today;
    }
    let yesterday = now_local
        .date_naive()
        .pred_opt()
        .expect("a day before now");
    local_at(yesterday, daily_at, tz)
}

/// Resolve a local date+time to a UTC instant in `tz`, navigating DST
/// edges deterministically:
///   - `LocalResult::Single` → that instant.
///   - `LocalResult::Ambiguous(_, latest)` → the later instant (fall-back
///     hour). Picking latest means a daily job fires once across the
///     repeated hour, not twice.
///   - `LocalResult::None` → spring-forward gap. Advance the local time
///     by 1 minute and try again, repeating up to 120 times (so the worst
///     case is still well inside an hour-long gap).
fn local_at(date: chrono::NaiveDate, time: NaiveTime, tz: Tz) -> DateTime<Utc> {
    use chrono::LocalResult;
    for offset_minutes in 0..120 {
        let mut t = time;
        if offset_minutes > 0 {
            let added = chrono::NaiveTime::from_num_seconds_from_midnight_opt(
                ((time.num_seconds_from_midnight() as i64 + offset_minutes * 60) % 86_400) as u32,
                0,
            )
            .unwrap_or(time);
            t = added;
        }
        let naive = date.and_time(t);
        match tz.from_local_datetime(&naive) {
            LocalResult::Single(dt) => return dt.with_timezone(&Utc),
            LocalResult::Ambiguous(_, latest) => return latest.with_timezone(&Utc),
            LocalResult::None => continue,
        }
    }
    // Should be unreachable — DST gaps are always less than an hour.
    Utc.from_utc_datetime(&date.and_time(time))
}

// ---------------------------------------------------------------------------
// crawler_state I/O
// ---------------------------------------------------------------------------

async fn read_last_tick(pool: &PgPool) -> sqlx::Result<Option<DateTime<Utc>>> {
    let row: Option<serde_json::Value> = sqlx::query_scalar(
        "SELECT value FROM crawler_state WHERE key = $1",
    )
    .bind(STATE_KEY_LAST_TICK)
    .fetch_optional(pool)
    .await?;
    Ok(row.and_then(|v| {
        v.get("at")
            .and_then(|s| s.as_str())
            .and_then(|s| DateTime::parse_from_rfc3339(s).ok())
            .map(|dt| dt.with_timezone(&Utc))
    }))
}

async fn write_last_tick(pool: &PgPool, at: DateTime<Utc>) -> sqlx::Result<()> {
    sqlx::query(
        "INSERT INTO crawler_state (key, value, updated_at) \
         VALUES ($1, $2, now()) \
         ON CONFLICT (key) DO UPDATE \
            SET value = EXCLUDED.value, updated_at = now()",
    )
    .bind(STATE_KEY_LAST_TICK)
    .bind(json!({ "at": at.to_rfc3339() }))
    .execute(pool)
    .await?;
    Ok(())
}

// ---------------------------------------------------------------------------
// Test helpers (not gated on cfg(test) — integration tests in tests/ dir
// need them too).
// ---------------------------------------------------------------------------

pub mod test_support {
    //! Lightweight stubs the daemon tests use. Public because integration
    //! tests live outside this module.
    use super::*;
    use std::sync::atomic::AtomicUsize;

    pub struct CountingMetadataPass {
        pub count: AtomicUsize,
    }

    impl Default for CountingMetadataPass {
        fn default() -> Self {
            Self {
                count: AtomicUsize::new(0),
            }
        }
    }

    #[async_trait]
    impl MetadataPass for CountingMetadataPass {
        async fn run(&self) -> anyhow::Result<pipeline::MetadataStats> {
            self.count.fetch_add(1, Ordering::AcqRel);
            Ok(pipeline::MetadataStats::default())
        }
    }

    /// `MetadataPass` that sleeps for `delay` before returning. Lets a
    /// test trigger shutdown WHILE a tick body is in flight, exercising
    /// the cancellation race in `CronContext::run_tick`.
    pub struct SlowMetadataPass {
        pub delay: std::time::Duration,
        pub started: AtomicUsize,
    }

    impl SlowMetadataPass {
        pub fn new(delay: std::time::Duration) -> Arc<Self> {
            Arc::new(Self {
                delay,
                started: AtomicUsize::new(0),
            })
        }
        pub fn start_count(&self) -> usize {
            self.started.load(Ordering::Acquire)
        }
    }

    #[async_trait]
    impl MetadataPass for SlowMetadataPass {
        async fn run(&self) -> anyhow::Result<pipeline::MetadataStats> {
            self.started.fetch_add(1, Ordering::AcqRel);
            tokio::time::sleep(self.delay).await;
            Ok(pipeline::MetadataStats::default())
        }
    }

    pub type DispatchFn = Arc<
        dyn Fn(JobPayload) -> futures_util::future::BoxFuture<'static, anyhow::Result<SyncOutcome>>
            + Send
            + Sync,
    >;

    pub struct StubDispatcher {
        pub handler: DispatchFn,
    }

    #[async_trait]
    impl ChapterDispatcher for StubDispatcher {
        async fn dispatch(&self, payload: JobPayload) -> anyhow::Result<SyncOutcome> {
            (self.handler)(payload).await
        }
    }

    pub fn always_done() -> Arc<StubDispatcher> {
        Arc::new(StubDispatcher {
            handler: Arc::new(|_| Box::pin(async { Ok(SyncOutcome::Fetched { pages: 1 }) })),
        })
    }

    pub fn panicking_dispatcher() -> Arc<StubDispatcher> {
        Arc::new(StubDispatcher {
            handler: Arc::new(|_| Box::pin(async { panic!("intentional dispatcher panic") })),
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::Duration as ChronoDuration;

    fn dt_utc(y: i32, mo: u32, d: u32, h: u32, mi: u32) -> DateTime<Utc> {
        Utc.with_ymd_and_hms(y, mo, d, h, mi, 0).unwrap()
    }

    #[test]
    fn next_fire_in_utc_at_midnight_advances_one_day() {
        let now = dt_utc(2026, 5, 25, 12, 0); // noon UTC
        let at = NaiveTime::from_hms_opt(0, 0, 0).unwrap();
        let next = next_fire(now, at, Tz::UTC);
        // Next midnight is May 26 00:00 UTC.
        assert_eq!(next, dt_utc(2026, 5, 26, 0, 0));
    }

    #[test]
    fn next_fire_before_today_slot_returns_today() {
        let now = dt_utc(2026, 5, 25, 23, 0); // 23:00 UTC
        let at = NaiveTime::from_hms_opt(23, 30, 0).unwrap();
        let next = next_fire(now, at, Tz::UTC);
        assert_eq!(next, dt_utc(2026, 5, 25, 23, 30));
    }

    #[test]
    fn next_fire_skips_spring_forward_gap_in_europe_berlin() {
        // 2024-03-31: clocks jump 02:00 -> 03:00 in Berlin (CET -> CEST).
        // Asking for daily_at = 02:30 on the morning of the jump should
        // land on the *next valid* local instant past the gap. We test
        // by computing `next_fire` at 2024-03-31 00:30 UTC (= 01:30 CET,
        // i.e. just before the gap). The next 02:30 local does not exist,
        // so the helper advances past it.
        let now = dt_utc(2024, 3, 31, 0, 30); // 01:30 local Berlin (CET = UTC+1)
        let at = NaiveTime::from_hms_opt(2, 30, 0).unwrap();
        let next = next_fire(now, at, Tz::Europe__Berlin);
        // Local Berlin time skips from 02:00 -> 03:00. After the +1 minute
        // search, the first valid slot is 03:00 local on 2024-03-31, which
        // is 01:00 UTC (CEST = UTC+2).
        // We assert the result is strictly between (now) and 1h later
        // and is in UTC — the exact minute depends on how many +1m steps
        // were required.
        assert!(next > now);
        assert!(next < now + ChronoDuration::hours(2));
    }

    #[test]
    fn next_fire_on_fall_back_picks_later_instant() {
        // 2024-10-27: clocks jump 03:00 -> 02:00 (CEST -> CET) in Berlin.
        // 02:30 happens twice on that day. We pick the later one.
        let now = dt_utc(2024, 10, 26, 12, 0); // day before, noon UTC
        let at = NaiveTime::from_hms_opt(2, 30, 0).unwrap();
        let next = next_fire(now, at, Tz::Europe__Berlin);
        // First 02:30 local is 00:30 UTC (CEST = UTC+2).
        // Second 02:30 local is 01:30 UTC (CET = UTC+1).
        // We expect the later instant: 01:30 UTC on 2024-10-27.
        assert_eq!(next, dt_utc(2024, 10, 27, 1, 30));
    }

    #[test]
    fn previous_fire_returns_today_when_now_is_after_slot() {
        let now = dt_utc(2026, 5, 25, 12, 0); // noon UTC
        let at = NaiveTime::from_hms_opt(0, 0, 0).unwrap();
        let prev = previous_fire(now, at, Tz::UTC);
        assert_eq!(prev, dt_utc(2026, 5, 25, 0, 0));
    }

    #[test]
    fn previous_fire_returns_yesterday_when_now_is_before_today_slot() {
        let now = dt_utc(2026, 5, 25, 8, 0); // 08:00 UTC
        let at = NaiveTime::from_hms_opt(23, 30, 0).unwrap();
        let prev = previous_fire(now, at, Tz::UTC);
        assert_eq!(prev, dt_utc(2026, 5, 24, 23, 30));
    }

    /// Documents the panic-isolation pattern `run_tick` now relies on:
    /// `AssertUnwindSafe(...).catch_unwind().await` must yield `Err(_)`
    /// when the wrapped future panics, so the surrounding loop (or in
    /// our case, the unconditional advisory-unlock that follows) keeps
    /// running. The shape of this test mirrors the production callsite.
    #[tokio::test]
    async fn assert_unwind_safe_catches_a_panicking_future() {
        let result = AssertUnwindSafe(async {
            panic!("boom");
        })
        .catch_unwind()
        .await;
        assert!(result.is_err(), "panicking future must yield Err");
    }
}
