//! Persistent job queue and its job kinds.
//!
//! Backed by Postgres (the `crawler_jobs` table). Workers lease rows
//! with `SELECT ... FOR UPDATE SKIP LOCKED`, heartbeat via
//! `leased_until`, and ack by transitioning to `done` (or backoff /
//! `dead`). Handlers are idempotent so a crash mid-run is recoverable
//! by replay.

use std::time::Duration;

use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use uuid::Uuid;

#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum JobPayload {
    /// Fetch one manga's detail page, upsert metadata, sync its chapter
    /// list. `url` and `title` are carried from the list ref so the worker
    /// can reconstruct a `SourceMangaRef` without re-walking, and so the
    /// dead-jobs/history UI can render a label even before any manga row
    /// exists. `#[serde(default)]` keeps older/hand-inserted rows decodable.
    SyncManga {
        source_id: String,
        source_manga_key: String,
        #[serde(default)]
        url: String,
        #[serde(default)]
        title: String,
    },
    /// Diff the chapter list, enqueue `SyncChapterContent` for new
    /// chapters, soft-drop vanished ones.
    SyncChapterList {
        source_id: String,
        manga_id: Uuid,
        source_manga_key: String,
    },
    /// Download a single chapter's page images into storage.
    SyncChapterContent {
        source_id: String,
        chapter_id: Uuid,
        source_chapter_key: String,
    },
    /// Run AI content-analysis (OCR, auto-tags, scene description, NSFW
    /// moderation) on a single page image via the local vision model.
    /// `force` re-analyzes a page that is already `done` (manual admin
    /// re-trigger); otherwise the worker skips it.
    AnalyzePage {
        page_id: Uuid,
        #[serde(default)]
        force: bool,
    },
}

#[derive(Clone, Copy, Debug, sqlx::Type, Serialize, Deserialize)]
#[sqlx(type_name = "text", rename_all = "snake_case")]
#[serde(rename_all = "snake_case")]
pub enum JobState {
    Pending,
    Running,
    Done,
    Failed,
    Dead,
}

/// Kind discriminator stored in `payload->>'kind'`. Public so callers
/// (daemon worker, bookmark hook) can filter `lease()` to a single kind
/// without re-spelling the literal.
pub const KIND_SYNC_CHAPTER_CONTENT: &str = "sync_chapter_content";

/// Kind discriminator for manga-detail sync jobs (used by the reconcile
/// pass to enqueue missing mangas). The crawl worker leases this alongside
/// `KIND_SYNC_CHAPTER_CONTENT`; both serialize on the single browser.
pub const KIND_SYNC_MANGA: &str = "sync_manga";

/// Kind discriminator for AI page-analysis jobs. The analysis daemon
/// leases with this filter so it never contends with crawl jobs.
pub const KIND_ANALYZE_PAGE: &str = "analyze_page";

#[derive(Debug)]
pub enum EnqueueResult {
    Inserted(Uuid),
    Skipped,
}

#[derive(Debug, Clone)]
pub struct Lease {
    pub id: Uuid,
    pub payload: JobPayload,
    pub attempts: i32,
    pub max_attempts: i32,
    /// Strictly-increasing token bumped by every `lease`, `release` /
    /// `release_unowned`, and `reclaim_orphaned`. `ack_done` /
    /// `ack_failed` / `renew` match on `(id, lease_generation)` so a
    /// late ack from a dead-but-still-dispatching lease cannot clobber
    /// a successor that has re-leased the same row. See migration
    /// 0032 for the column and the `ack_done_from_dead_lease_*` tests
    /// in `crawler_jobs.rs` for the race the token closes.
    pub lease_generation: i64,
}

/// Deterministic exponential backoff base for `ack_failed` retries.
/// `attempts` is the post-increment value reported by `lease()` (so the
/// first failure has `attempts == 1` and waits 60s, the second 120s,
/// etc.). Capped at 1h to avoid runaway long sleeps that would outlive
/// the daemon process. Jitter is applied separately by [`apply_jitter`].
fn backoff_base(attempts: i32) -> Duration {
    let shift = attempts.saturating_sub(1).clamp(0, 20) as u32;
    let secs = 60u64.saturating_mul(1u64 << shift);
    Duration::from_secs(secs.min(3600))
}

/// Apply ±20% jitter to a backoff duration. `jitter` is a fraction in
/// `[0.0, 1.0)` (e.g. `rand::random::<f64>()`), mapped to a multiplier in
/// `[0.8, 1.2)`. Pure so the bounds stay unit-testable. Spreading retries
/// avoids a thundering herd when a source outage fails many jobs at once.
fn apply_jitter(base: Duration, jitter: f64) -> Duration {
    let frac = jitter.clamp(0.0, 1.0);
    let mult = 0.8 + 0.4 * frac; // [0.8, 1.2)
    Duration::from_secs((base.as_secs_f64() * mult).round() as u64)
}

/// Jittered exponential backoff for `ack_failed`. Wraps [`backoff_base`]
/// with a random ±20% spread.
fn backoff_for(attempts: i32) -> Duration {
    apply_jitter(backoff_base(attempts), rand::random::<f64>())
}

/// Insert a new pending job. For `SyncChapterContent` payloads the
/// partial unique index `crawler_jobs_chapter_content_dedup_idx` blocks
/// a second `(pending|running)` insert per chapter_id, returning
/// `Skipped`. The slot frees again once the previous job leaves the
/// in-flight states (done/failed/dead), so a re-enqueue after a force
/// refetch succeeds.
pub async fn enqueue(pool: &PgPool, payload: &JobPayload) -> sqlx::Result<EnqueueResult> {
    let json = serde_json::to_value(payload).expect("JobPayload is always serializable");
    let id: Option<Uuid> = sqlx::query_scalar(
        "INSERT INTO crawler_jobs (payload) VALUES ($1) \
         ON CONFLICT DO NOTHING RETURNING id",
    )
    .bind(json)
    .fetch_optional(pool)
    .await?;
    Ok(match id {
        Some(id) => EnqueueResult::Inserted(id),
        None => EnqueueResult::Skipped,
    })
}

/// Lease up to `max` rows whose `state` is `pending`, or `running` with
/// an expired `leased_until` (the crashed-worker recovery path). The
/// inner CTE uses `FOR UPDATE SKIP LOCKED` so concurrent leasers don't
/// block each other and each row is handed to exactly one worker.
///
/// `kind_filter` matches against `payload->>'kind'`; `None` means
/// any kind.
///
/// Ties on `scheduled_at` (the common case: a cron batch enqueues
/// everything with the same default `now()`) break by `created_at`, so
/// jobs come off the queue in insertion order. The enqueue paths insert
/// chapter-content jobs in ascending `chapters.number` order, so this
/// tiebreaker is what propagates that intent through to dequeue.
pub async fn lease(
    pool: &PgPool,
    kind_filter: Option<&str>,
    max: i64,
    lease_duration: Duration,
) -> sqlx::Result<Vec<Lease>> {
    let lease_ms: i64 = lease_duration.as_millis().min(i64::MAX as u128) as i64;
    let rows: Vec<(Uuid, serde_json::Value, i32, i32, i64)> = sqlx::query_as(
        r#"
        WITH leased AS (
            SELECT id FROM crawler_jobs
             WHERE (state = 'pending' OR (state = 'running' AND leased_until < now()))
               AND scheduled_at <= now()
               AND ($1::text IS NULL OR payload->>'kind' = $1)
             ORDER BY scheduled_at, created_at
             LIMIT $2
             FOR UPDATE SKIP LOCKED
        )
        UPDATE crawler_jobs j
           SET state = 'running',
               attempts = j.attempts + 1,
               lease_generation = j.lease_generation + 1,
               leased_until = now() + ($3::bigint || ' milliseconds')::interval,
               updated_at = now()
          FROM leased l
         WHERE j.id = l.id
        RETURNING j.id, j.payload, j.attempts, j.max_attempts, j.lease_generation
        "#,
    )
    .bind(kind_filter)
    .bind(max)
    .bind(lease_ms)
    .fetch_all(pool)
    .await?;

    decode_leases(rows)
}

/// Like [`lease`] but matches any of several `payload->>'kind'` values. The
/// crawl worker uses this to drain both `sync_chapter_content` and
/// `sync_manga` from one loop (both serialize on the single browser); the
/// analysis daemon keeps using single-kind [`lease`] so it never contends.
pub async fn lease_kinds(
    pool: &PgPool,
    kinds: &[&str],
    max: i64,
    lease_duration: Duration,
) -> sqlx::Result<Vec<Lease>> {
    let lease_ms: i64 = lease_duration.as_millis().min(i64::MAX as u128) as i64;
    let kinds_vec: Vec<String> = kinds.iter().map(|s| s.to_string()).collect();
    let rows: Vec<(Uuid, serde_json::Value, i32, i32, i64)> = sqlx::query_as(
        r#"
        WITH leased AS (
            SELECT id FROM crawler_jobs
             WHERE (state = 'pending' OR (state = 'running' AND leased_until < now()))
               AND scheduled_at <= now()
               AND payload->>'kind' = ANY($1)
             ORDER BY scheduled_at, created_at
             LIMIT $2
             FOR UPDATE SKIP LOCKED
        )
        UPDATE crawler_jobs j
           SET state = 'running',
               attempts = j.attempts + 1,
               lease_generation = j.lease_generation + 1,
               leased_until = now() + ($3::bigint || ' milliseconds')::interval,
               updated_at = now()
          FROM leased l
         WHERE j.id = l.id
        RETURNING j.id, j.payload, j.attempts, j.max_attempts, j.lease_generation
        "#,
    )
    .bind(&kinds_vec)
    .bind(max)
    .bind(lease_ms)
    .fetch_all(pool)
    .await?;
    decode_leases(rows)
}

fn decode_leases(
    rows: Vec<(Uuid, serde_json::Value, i32, i32, i64)>,
) -> sqlx::Result<Vec<Lease>> {
    let mut leases = Vec::with_capacity(rows.len());
    for (id, payload_json, attempts, max_attempts, lease_generation) in rows {
        let payload: JobPayload = serde_json::from_value(payload_json).map_err(|e| {
            sqlx::Error::Decode(format!("invalid JobPayload JSON for job {id}: {e}").into())
        })?;
        leases.push(Lease {
            id,
            payload,
            attempts,
            max_attempts,
            lease_generation,
        });
    }
    Ok(leases)
}

/// Extend the lease on a still-owned `running` job. Returns `true` if the
/// row was updated (we still hold the lease), `false` if the job is no
/// longer `running` (re-leased after a missed heartbeat, or already
/// acked) — the caller's heartbeat loop should stop. The `state =
/// 'running'` guard mirrors [`ack_done`]'s rationale.
///
/// This is the heartbeat primitive: a worker renews periodically while a
/// long-but-healthy job runs so `leased_until` never lapses, which would
/// otherwise let another worker steal the in-flight job and spuriously
/// inflate `attempts` toward `max_attempts`.
pub async fn renew(
    pool: &PgPool,
    lease_id: Uuid,
    lease_generation: i64,
    lease_duration: Duration,
) -> sqlx::Result<bool> {
    let lease_ms: i64 = lease_duration.as_millis().min(i64::MAX as u128) as i64;
    let res = sqlx::query(
        "UPDATE crawler_jobs \
            SET leased_until = now() + ($3::bigint || ' milliseconds')::interval, \
                updated_at = now() \
          WHERE id = $1 AND lease_generation = $2 AND state = 'running'",
    )
    .bind(lease_id)
    .bind(lease_generation)
    .bind(lease_ms)
    .execute(pool)
    .await?;
    Ok(res.rows_affected() > 0)
}

/// Mark a leased job as successfully completed. The
/// `(lease_generation, state = 'running')` predicate guards against a
/// late ack from a *specific* dead lease — the case where the worker's
/// lease was released or expired and a successor re-leased the same id.
/// Before migration 0032 the guard was state-only, so a late ack from
/// the original could still match (state was running again, owned by the
/// successor) and clobber. Scoping to the original's `lease_generation`
/// makes the match fail cleanly. `rows_affected == 0` is surfaced as a
/// warn — the successor is doing real work; the late ack steps aside.
pub async fn ack_done(
    pool: &PgPool,
    lease_id: Uuid,
    lease_generation: i64,
) -> sqlx::Result<()> {
    let res = sqlx::query(
        "UPDATE crawler_jobs \
            SET state = 'done', leased_until = NULL, updated_at = now() \
          WHERE id = $1 AND lease_generation = $2 AND state = 'running'",
    )
    .bind(lease_id)
    .bind(lease_generation)
    .execute(pool)
    .await?;
    if res.rows_affected() == 0 {
        tracing::warn!(
            %lease_id,
            lease_generation,
            "ack_done: lease no longer current — generation mismatch (force-analyzed, reclaimed, or re-leased); skipping update"
        );
    }
    Ok(())
}

/// Mark a leased job as failed. If the current attempt count has reached
/// `max_attempts` the job is terminally dead and stops retrying;
/// otherwise it goes back to `pending` with `scheduled_at` pushed into
/// the future by the exponential backoff. See [`ack_done`] for the
/// `state = 'running'` guard rationale.
pub async fn ack_failed(
    pool: &PgPool,
    lease_id: Uuid,
    error: &str,
    attempts: i32,
    max_attempts: i32,
    lease_generation: i64,
) -> sqlx::Result<()> {
    let res = if attempts >= max_attempts {
        sqlx::query(
            "UPDATE crawler_jobs \
                SET state = 'dead', last_error = $2, leased_until = NULL, updated_at = now() \
              WHERE id = $1 AND lease_generation = $3 AND state = 'running'",
        )
        .bind(lease_id)
        .bind(error)
        .bind(lease_generation)
        .execute(pool)
        .await?
    } else {
        let backoff_ms: i64 = backoff_for(attempts).as_millis().min(i64::MAX as u128) as i64;
        sqlx::query(
            "UPDATE crawler_jobs \
                SET state = 'pending', last_error = $2, leased_until = NULL, \
                    scheduled_at = now() + ($3::bigint || ' milliseconds')::interval, \
                    updated_at = now() \
              WHERE id = $1 AND lease_generation = $4 AND state = 'running'",
        )
        .bind(lease_id)
        .bind(error)
        .bind(backoff_ms)
        .bind(lease_generation)
        .execute(pool)
        .await?
    };
    if res.rows_affected() == 0 {
        tracing::warn!(
            %lease_id,
            lease_generation,
            "ack_failed: lease no longer current — generation mismatch (force-analyzed, reclaimed, or re-leased); skipping update"
        );
    }
    Ok(())
}

/// Return a leased job to `pending` without burning a retry attempt.
/// Used by workers on graceful shutdown and on session-expired aborts
/// where the failure isn't the job's fault. The
/// `(lease_generation, state = 'running')` guard scopes the release to
/// the *specific* lease the caller was holding — otherwise a late
/// release could clobber a successor's lease (drop the row back to
/// pending mid-dispatch). Also bumps `lease_generation` so any
/// subsequent ack from this same lease finds nothing.
///
/// For force-analyze and other external paths that release a
/// running lease without holding a [`Lease`] struct, use
/// [`release_unowned`] instead.
pub async fn release(
    pool: &PgPool,
    lease_id: Uuid,
    lease_generation: i64,
) -> sqlx::Result<()> {
    let res = sqlx::query(
        "UPDATE crawler_jobs \
            SET state = 'pending', leased_until = NULL, \
                attempts = GREATEST(0, attempts - 1), \
                lease_generation = lease_generation + 1, \
                updated_at = now() \
          WHERE id = $1 AND lease_generation = $2 AND state = 'running'",
    )
    .bind(lease_id)
    .bind(lease_generation)
    .execute(pool)
    .await?;
    if res.rows_affected() == 0 {
        tracing::warn!(
            %lease_id,
            lease_generation,
            "release: lease no longer current — generation mismatch (force-analyzed, reclaimed, or re-leased); skipping update"
        );
    }
    Ok(())
}

/// Release a `running` lease without owning it: caller has only a
/// job id, not a [`Lease`] struct. Used by force-analyze (admin
/// click that drops an in-flight lease so a refreshed payload is
/// picked up) and ops tools.
///
/// Unconditionally matches the row by id and bumps
/// `lease_generation` so the dropped lease's pending ack from
/// the original worker becomes a no-op. Returns the attempt to
/// `pending` and refunds the retry attempt (the operator click
/// isn't a job-level failure).
pub async fn release_unowned(pool: &PgPool, lease_id: Uuid) -> sqlx::Result<()> {
    let res = sqlx::query(
        "UPDATE crawler_jobs \
            SET state = 'pending', leased_until = NULL, \
                attempts = GREATEST(0, attempts - 1), \
                lease_generation = lease_generation + 1, \
                updated_at = now() \
          WHERE id = $1 AND state = 'running'",
    )
    .bind(lease_id)
    .execute(pool)
    .await?;
    if res.rows_affected() == 0 {
        tracing::warn!(
            %lease_id,
            "release_unowned: row not running — likely already completed or never leased; skipping"
        );
    }
    Ok(())
}

/// Reclaim jobs orphaned by a crashed/killed worker: those still `running`
/// whose `leased_until` has already lapsed. Each is returned to `pending`
/// with its lease cleared and the lease's `attempts` increment refunded
/// (`GREATEST(0, attempts - 1)`, mirroring [`release`]) — a crash that
/// interrupted an attempt mid-flight shouldn't count against `max_attempts`.
/// Returns the number reclaimed.
///
/// Intended to run once at daemon startup so crash recovery is immediate
/// rather than waiting up to a full lease window for `lease`'s expiry clause
/// to re-pick the row. It is safe under multi-replica deployment precisely
/// because it only touches **already-expired** leases: a healthy peer
/// heartbeats (`renew`) every ~20s, keeping its in-flight jobs' `leased_until`
/// in the future, so this never steals live work — it only does eagerly what
/// `lease` would do lazily, minus the attempt burn.
///
/// Trade-off: a job whose dispatch reliably hard-kills the process (e.g. OOM
/// on a pathological payload) is refunded each boot and could loop without
/// dead-lettering. That window is bounded in practice by the per-image size
/// cap and the worker's `job_timeout` (a hang is acked-failed normally, which
/// *does* burn an attempt); only a true process-killer evades it, which is an
/// infrastructure signal worth surfacing rather than silently dead-lettering
/// everyone's chapters during a crash loop.
pub async fn reclaim_orphaned(pool: &PgPool) -> sqlx::Result<u64> {
    let result = sqlx::query(
        "UPDATE crawler_jobs \
            SET state = 'pending', leased_until = NULL, \
                attempts = GREATEST(0, attempts - 1), \
                lease_generation = lease_generation + 1, \
                updated_at = now() \
          WHERE state = 'running' AND leased_until < now()",
    )
    .execute(pool)
    .await?;
    Ok(result.rows_affected())
}

/// Delete `done` jobs whose `updated_at` is older than `retention_days`
/// days. `0` disables the reaper without touching the table. Returns the
/// number of rows removed.
pub async fn reap_done(pool: &PgPool, retention_days: u32) -> sqlx::Result<u64> {
    if retention_days == 0 {
        return Ok(0);
    }
    let result = sqlx::query(
        "DELETE FROM crawler_jobs \
          WHERE state = 'done' \
            AND updated_at < now() - ($1::bigint || ' days')::interval",
    )
    .bind(retention_days as i64)
    .execute(pool)
    .await?;
    Ok(result.rows_affected())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn analyze_page_payload_round_trips_through_tagged_json() {
        let page_id = Uuid::new_v4();
        let payload = JobPayload::AnalyzePage {
            page_id,
            force: true,
        };
        let json = serde_json::to_value(&payload).unwrap();
        assert_eq!(json["kind"], "analyze_page");
        assert_eq!(json["page_id"], page_id.to_string());
        assert_eq!(json["force"], true);

        // `force` defaults to false when an older enqueue omitted it.
        let legacy = serde_json::json!({ "kind": "analyze_page", "page_id": page_id });
        match serde_json::from_value::<JobPayload>(legacy).unwrap() {
            JobPayload::AnalyzePage { page_id: pid, force } => {
                assert_eq!(pid, page_id);
                assert!(!force);
            }
            other => panic!("expected AnalyzePage, got {other:?}"),
        }
    }

    #[test]
    fn sync_manga_payload_round_trips_with_url_and_title() {
        let payload = JobPayload::SyncManga {
            source_id: "target".into(),
            source_manga_key: "foo".into(),
            url: "https://target.example/manga/foo".into(),
            title: "Foo Title".into(),
        };
        let json = serde_json::to_value(&payload).unwrap();
        assert_eq!(json["kind"], KIND_SYNC_MANGA);
        assert_eq!(json["source_id"], "target");
        assert_eq!(json["source_manga_key"], "foo");
        assert_eq!(json["url"], "https://target.example/manga/foo");
        assert_eq!(json["title"], "Foo Title");

        match serde_json::from_value::<JobPayload>(json).unwrap() {
            JobPayload::SyncManga {
                source_id,
                source_manga_key,
                url,
                title,
            } => {
                assert_eq!(source_id, "target");
                assert_eq!(source_manga_key, "foo");
                assert_eq!(url, "https://target.example/manga/foo");
                assert_eq!(title, "Foo Title");
            }
            other => panic!("expected SyncManga, got {other:?}"),
        }
    }

    #[test]
    fn sync_manga_payload_defaults_missing_url_and_title() {
        // A row that predates the url/title fields must still decode.
        let legacy = serde_json::json!({
            "kind": "sync_manga",
            "source_id": "target",
            "source_manga_key": "foo",
        });
        match serde_json::from_value::<JobPayload>(legacy).unwrap() {
            JobPayload::SyncManga {
                url,
                title,
                source_manga_key,
                ..
            } => {
                assert_eq!(source_manga_key, "foo");
                assert_eq!(url, "");
                assert_eq!(title, "");
            }
            other => panic!("expected SyncManga, got {other:?}"),
        }
    }

    #[test]
    fn backoff_base_grows_exponentially_and_caps_at_one_hour() {
        // attempts == 1 → 60s, doubling each step.
        assert_eq!(backoff_base(1), Duration::from_secs(60));
        assert_eq!(backoff_base(2), Duration::from_secs(120));
        assert_eq!(backoff_base(3), Duration::from_secs(240));
        assert_eq!(backoff_base(4), Duration::from_secs(480));
        assert_eq!(backoff_base(5), Duration::from_secs(960));
        assert_eq!(backoff_base(6), Duration::from_secs(1920));
        // 7th: 60 * 64 = 3840 → capped to 3600.
        assert_eq!(backoff_base(7), Duration::from_secs(3600));
        assert_eq!(backoff_base(20), Duration::from_secs(3600));
        // Garbage / zero / negatives stay sane.
        assert_eq!(backoff_base(0), Duration::from_secs(60));
        assert_eq!(backoff_base(-5), Duration::from_secs(60));
    }

    #[test]
    fn apply_jitter_stays_within_plus_minus_twenty_percent() {
        let base = Duration::from_secs(100);
        // Lower bound (jitter = 0.0) → 0.8x.
        assert_eq!(apply_jitter(base, 0.0), Duration::from_secs(80));
        // Midpoint (jitter = 0.5) → 1.0x.
        assert_eq!(apply_jitter(base, 0.5), Duration::from_secs(100));
        // Upper end (jitter → 1.0) → ~1.2x.
        assert_eq!(apply_jitter(base, 1.0), Duration::from_secs(120));
        // Out-of-range inputs are clamped, never panic.
        assert_eq!(apply_jitter(base, -3.0), Duration::from_secs(80));
        assert_eq!(apply_jitter(base, 9.0), Duration::from_secs(120));
    }

    #[test]
    fn backoff_for_random_jitter_stays_in_band() {
        // The production wrapper draws its own randomness; assert the
        // result for a mid-range attempt always lands within the jitter
        // band of the base, across many draws.
        let base = backoff_base(3).as_secs_f64(); // 240s
        for _ in 0..1000 {
            let v = backoff_for(3).as_secs_f64();
            assert!(
                v >= base * 0.8 - 1.0 && v <= base * 1.2 + 1.0,
                "jittered backoff {v} outside band of base {base}"
            );
        }
    }
}