//! SQLite persistence for the memory-tree job queue.
//!
//! Producers call [`enqueue`] inside their own writes (or with a fresh tx)
//! to atomically commit the side-effect plus its follow-up job. The worker
//! pool calls [`claim_next`] to lease a job, [`mark_done`] / [`mark_failed`]
//! to settle it, and [`recover_stale_locks`] on startup to flip rows whose
//! `locked_until_ms` expired without a settle.
//!
//! Concurrency:
//! - The dedupe key is enforced by a partial `UNIQUE` index that only
//!   covers `status IN ('ready', 'running')`. Producers use `INSERT OR
//!   IGNORE` so a duplicate enqueue while a job is in flight or queued is
//!   a silent no-op; a duplicate enqueue after the first completes is
//!   accepted and creates a fresh row.
//! - `claim_next` is one statement: `UPDATE … WHERE id = (SELECT … LIMIT 1)
//!   RETURNING …`. SQLite serialises writes, so no two workers can claim
//!   the same row.

use anyhow::{Context, Result};
use chrono::Utc;
use rusqlite::{params, Connection, OptionalExtension, Transaction};
use uuid::Uuid;

use crate::openhuman::config::Config;
use crate::openhuman::memory::tree::jobs::redact::scrub_for_log;
use crate::openhuman::memory::tree::jobs::types::{Job, JobKind, JobStatus, NewJob};
use crate::openhuman::memory::tree::store::with_connection;

/// Default visibility lock — a worker that crashes mid-job will have its
/// row recovered after this window. 5 min is comfortably larger than any
/// expected single-job runtime (LLM extract or summarise) without leaving
/// real failures stuck for hours.
pub const DEFAULT_LOCK_DURATION_MS: i64 = 5 * 60 * 1_000;

/// Backoff math for retry. Returns `now + min(base * 2^attempts, cap)`.
const RETRY_BASE_MS: i64 = 60 * 1_000;
const RETRY_CAP_MS: i64 = 60 * 60 * 1_000;
const DEFAULT_MAX_ATTEMPTS: u32 = 5;

/// Enqueue one job. Idempotent on `dedupe_key` while another active row
/// (status `ready`/`running`) shares it. Returns `Some(id)` if the row
/// was inserted, `None` if a duplicate was suppressed.
pub fn enqueue(config: &Config, job: &NewJob) -> Result<Option<String>> {
    with_connection(config, |conn| enqueue_conn(conn, job))
}

/// Enqueue inside a caller-owned transaction. Use this when the producer
/// is already mid-tx (e.g. `ingest::persist` writing chunks + jobs in one
/// commit) so the queue insert lands atomically with the side-effect.
/// `Transaction` derefs to `Connection`, so callers just pass `&tx`.
pub fn enqueue_tx(tx: &Transaction<'_>, job: &NewJob) -> Result<Option<String>> {
    enqueue_conn(tx, job)
}

pub(crate) fn enqueue_conn(conn: &Connection, job: &NewJob) -> Result<Option<String>> {
    let id = format!("job:{}", Uuid::new_v4());
    let now_ms = Utc::now().timestamp_millis();
    let available_at = job.available_at_ms.unwrap_or(now_ms);
    let max_attempts = job.max_attempts.unwrap_or(DEFAULT_MAX_ATTEMPTS) as i64;

    let inserted = conn.execute(
        "INSERT OR IGNORE INTO mem_tree_jobs (
            id, kind, payload_json, dedupe_key, status, attempts, max_attempts,
            available_at_ms, locked_until_ms, last_error,
            created_at_ms, started_at_ms, completed_at_ms
        ) VALUES (?1, ?2, ?3, ?4, 'ready', 0, ?5, ?6, NULL, NULL, ?7, NULL, NULL)",
        params![
            id,
            job.kind.as_str(),
            job.payload_json,
            job.dedupe_key,
            max_attempts,
            available_at,
            now_ms,
        ],
    )?;

    if inserted == 0 {
        log::debug!(
            "[memory_tree::jobs] enqueue suppressed by dedupe kind={} key={:?}",
            job.kind.as_str(),
            job.dedupe_key
        );
        return Ok(None);
    }
    log::debug!(
        "[memory_tree::jobs] enqueued id={} kind={} avail_at_ms={} dedupe={:?}",
        id,
        job.kind.as_str(),
        available_at,
        job.dedupe_key
    );
    Ok(Some(id))
}

/// Atomically claim the next ready job whose `available_at_ms` has come
/// due. Sets `status=running`, bumps `attempts`, stamps `started_at_ms`
/// and `locked_until_ms`. Returns `None` when the queue is empty / not
/// yet due.
pub fn claim_next(config: &Config, lock_duration_ms: i64) -> Result<Option<Job>> {
    with_connection(config, |conn| {
        let now_ms = Utc::now().timestamp_millis();
        let lock_until = now_ms.saturating_add(lock_duration_ms);

        let row = conn
            .query_row(
                // Drain forward, don't widen. Most-downstream kinds run
                // first so a slow LLM-bound `extract_chunk` can't starve
                // the routing/seal/digest pipeline behind it.
                "UPDATE mem_tree_jobs
                    SET status = 'running',
                        attempts = attempts + 1,
                        started_at_ms = ?1,
                        locked_until_ms = ?2,
                        last_error = NULL
                  WHERE id = (
                      SELECT id FROM mem_tree_jobs
                       WHERE status = 'ready'
                         AND available_at_ms <= ?1
                       ORDER BY
                         CASE kind
                           WHEN 'digest_daily'  THEN 0
                           WHEN 'seal'          THEN 1
                           WHEN 'flush_stale'   THEN 2
                           WHEN 'topic_route'   THEN 3
                           WHEN 'append_buffer' THEN 4
                           ELSE 5
                         END ASC,
                         available_at_ms ASC
                       LIMIT 1
                  )
              RETURNING id, kind, payload_json, dedupe_key, status, attempts,
                        max_attempts, available_at_ms, locked_until_ms, last_error,
                        created_at_ms, started_at_ms, completed_at_ms",
                params![now_ms, lock_until],
                row_to_job,
            )
            .optional()
            .context("Failed to claim next mem_tree_jobs row")?;
        if let Some(j) = &row {
            log::debug!(
                "[memory_tree::jobs] claimed id={} kind={} attempt={}/{}",
                j.id,
                j.kind.as_str(),
                j.attempts,
                j.max_attempts
            );
        }
        Ok(row)
    })
}

/// Mark a claimed job as `done`. Clears the lock and stamps `completed_at_ms`.
///
/// The UPDATE is gated on `attempts` and `started_at_ms` matching the values
/// in `job` (the snapshot returned by [`claim_next`]). If the lease expired
/// and another worker re-claimed the row, `rows_affected` will be 0 — the
/// stale worker's settlement is a silent no-op rather than clobbering the new
/// lessee's state.
pub fn mark_done(config: &Config, job: &Job) -> Result<()> {
    let job_id = &job.id;
    let claim_attempts = job.attempts as i64;
    let claim_started_at = job.started_at_ms;
    with_connection(config, |conn| {
        let now_ms = Utc::now().timestamp_millis();
        let n = conn.execute(
            "UPDATE mem_tree_jobs
                SET status = 'done',
                    completed_at_ms = ?1,
                    locked_until_ms = NULL,
                    last_error = NULL
              WHERE id = ?2
                AND attempts = ?3
                AND started_at_ms IS ?4",
            params![now_ms, job_id, claim_attempts, claim_started_at],
        )?;
        if n == 0 {
            // Either the job row was deleted (shouldn't happen) or the lease
            // expired and a second worker re-claimed the row. Log and move on —
            // this is a known race outcome, not a bug in the current worker.
            log::warn!(
                "[memory_tree::jobs] mark_done id={job_id} was a no-op \
                 (stale lease: attempts={claim_attempts} started_at_ms={claim_started_at:?})"
            );
        }
        Ok(())
    })
}

/// Settle a failed job. If `attempts < max_attempts`, the row goes back
/// to `ready` with an exponential-backoff `available_at_ms`. Otherwise
/// it terminates as `failed`. Either way `last_error` is recorded.
///
/// Like [`mark_done`], the UPDATE is gated on the claim-token
/// (`attempts` + `started_at_ms`) so a stale worker's failure settlement
/// cannot clobber an active lessee's row — rows_affected == 0 is a silent
/// no-op.
pub fn mark_failed(config: &Config, job: &Job, error: &str) -> Result<()> {
    let job_id = &job.id;
    let attempts = job.attempts as i64;
    let max_attempts = job.max_attempts as i64;
    let claim_started_at = job.started_at_ms;
    with_connection(config, |conn| {
        let now_ms = Utc::now().timestamp_millis();

        // `error` is free-form (anyhow chain or handler-supplied text) and
        // may carry credential-shaped substrings; scrub before logging,
        // but keep the original in the DB column for diagnostics.
        let error_for_log = scrub_for_log(error);
        if attempts >= max_attempts {
            log::warn!(
                "[memory_tree::jobs] terminal failure id={job_id} \
                 attempts={attempts}/{max_attempts} err={error_for_log}"
            );
            let n = conn.execute(
                "UPDATE mem_tree_jobs
                    SET status = 'failed',
                        completed_at_ms = ?1,
                        locked_until_ms = NULL,
                        last_error = ?2
                  WHERE id = ?3
                    AND attempts = ?4
                    AND started_at_ms IS ?5",
                params![now_ms, error, job_id, attempts, claim_started_at],
            )?;
            if n == 0 {
                log::warn!(
                    "[memory_tree::jobs] mark_failed(terminal) id={job_id} was a no-op \
                     (stale lease: attempts={attempts} started_at_ms={claim_started_at:?})"
                );
            }
        } else {
            let backoff = backoff_ms(attempts as u32);
            let next_at = now_ms.saturating_add(backoff);
            log::info!(
                "[memory_tree::jobs] retry id={job_id} attempt={attempts}/{max_attempts} \
                 next_at_ms={next_at} err={error_for_log}"
            );
            let n = conn.execute(
                "UPDATE mem_tree_jobs
                    SET status = 'ready',
                        available_at_ms = ?1,
                        locked_until_ms = NULL,
                        last_error = ?2
                  WHERE id = ?3
                    AND attempts = ?4
                    AND started_at_ms IS ?5",
                params![next_at, error, job_id, attempts, claim_started_at],
            )?;
            if n == 0 {
                log::warn!(
                    "[memory_tree::jobs] mark_failed(retry) id={job_id} was a no-op \
                     (stale lease: attempts={attempts} started_at_ms={claim_started_at:?})"
                );
            }
        }
        Ok(())
    })
}

/// Mark a claimed job as deferred: put it back to `ready` with
/// `available_at_ms = until_ms` so [`claim_next`] will re-pick it once the
/// wake-up time has passed. The handler ran successfully but chose not to
/// make progress (cloud rate-limited, dependency unavailable, model
/// warming up), so this path **does not** burn the failure budget — the
/// `attempts` bump that [`claim_next`] applied at claim time is reverted.
///
/// `reason` is recorded in `last_error` for visibility and `started_at_ms`
/// is cleared (mirroring the retry branch of [`mark_failed`]) so the next
/// claim stamps a fresh start time.
///
/// Like [`mark_done`] / [`mark_failed`], the UPDATE is gated on the
/// claim-token (`attempts` + `started_at_ms`) so a stale lessee's
/// settlement is a silent no-op rather than clobbering an active lessee's
/// row.
pub fn mark_deferred(config: &Config, job: &Job, until_ms: i64, reason: &str) -> Result<()> {
    let job_id = &job.id;
    let claim_attempts = job.attempts as i64;
    let pre_claim_attempts = claim_attempts.saturating_sub(1);
    let claim_started_at = job.started_at_ms;
    with_connection(config, |conn| {
        let n = conn.execute(
            "UPDATE mem_tree_jobs
                SET status = 'ready',
                    attempts = ?1,
                    available_at_ms = ?2,
                    locked_until_ms = NULL,
                    started_at_ms = NULL,
                    last_error = ?3
              WHERE id = ?4
                AND attempts = ?5
                AND started_at_ms IS ?6",
            params![
                pre_claim_attempts,
                until_ms,
                reason,
                job_id,
                claim_attempts,
                claim_started_at,
            ],
        )?;
        if n == 0 {
            log::warn!(
                "[memory_tree::jobs] mark_deferred id={job_id} was a no-op \
                 (stale lease: attempts={claim_attempts} started_at_ms={claim_started_at:?})"
            );
        }
        Ok(())
    })
}

/// Flip any `running` row whose `locked_until_ms` has expired back to
/// `ready`. Called once at worker startup so a process crash mid-job
/// doesn't leave work stranded. Returns the number of rows recovered.
pub fn recover_stale_locks(config: &Config) -> Result<usize> {
    with_connection(config, |conn| {
        let now_ms = Utc::now().timestamp_millis();
        let n = conn.execute(
            "UPDATE mem_tree_jobs
                SET status = 'ready',
                    last_error = COALESCE(last_error, 'recovered_from_stale_lock')
              WHERE status = 'running'
                AND locked_until_ms IS NOT NULL
                AND locked_until_ms < ?1",
            params![now_ms],
        )?;
        if n > 0 {
            log::warn!("[memory_tree::jobs] recovered {n} stale-locked job(s) at startup");
        }
        Ok(n)
    })
}

/// Quick count helper for tests / diagnostics.
pub fn count_by_status(config: &Config, status: JobStatus) -> Result<u64> {
    with_connection(config, |conn| {
        let n: i64 = conn.query_row(
            "SELECT COUNT(*) FROM mem_tree_jobs WHERE status = ?1",
            params![status.as_str()],
            |r| r.get(0),
        )?;
        Ok(n.max(0) as u64)
    })
}

/// Total count regardless of status — handy for assertions.
pub fn count_total(config: &Config) -> Result<u64> {
    with_connection(config, |conn| {
        let n: i64 = conn.query_row("SELECT COUNT(*) FROM mem_tree_jobs", [], |r| r.get(0))?;
        Ok(n.max(0) as u64)
    })
}

/// Fetch one job by id (test/diagnostic helper).
pub fn get_job(config: &Config, id: &str) -> Result<Option<Job>> {
    with_connection(config, |conn| {
        let job = conn
            .query_row(
                "SELECT id, kind, payload_json, dedupe_key, status, attempts, max_attempts,
                        available_at_ms, locked_until_ms, last_error,
                        created_at_ms, started_at_ms, completed_at_ms
                   FROM mem_tree_jobs WHERE id = ?1",
                params![id],
                row_to_job,
            )
            .optional()?;
        Ok(job)
    })
}

fn row_to_job(row: &rusqlite::Row<'_>) -> rusqlite::Result<Job> {
    let id: String = row.get(0)?;
    let kind_s: String = row.get(1)?;
    let payload_json: String = row.get(2)?;
    let dedupe_key: Option<String> = row.get(3)?;
    let status_s: String = row.get(4)?;
    let attempts: i64 = row.get(5)?;
    let max_attempts: i64 = row.get(6)?;
    let available_at_ms: i64 = row.get(7)?;
    let locked_until_ms: Option<i64> = row.get(8)?;
    let last_error: Option<String> = row.get(9)?;
    let created_at_ms: i64 = row.get(10)?;
    let started_at_ms: Option<i64> = row.get(11)?;
    let completed_at_ms: Option<i64> = row.get(12)?;

    let kind = JobKind::parse(&kind_s).map_err(|e| {
        rusqlite::Error::FromSqlConversionFailure(1, rusqlite::types::Type::Text, e.into())
    })?;
    let status = JobStatus::parse(&status_s).map_err(|e| {
        rusqlite::Error::FromSqlConversionFailure(4, rusqlite::types::Type::Text, e.into())
    })?;

    Ok(Job {
        id,
        kind,
        payload_json,
        dedupe_key,
        status,
        attempts: attempts.max(0) as u32,
        max_attempts: max_attempts.max(0) as u32,
        available_at_ms,
        locked_until_ms,
        last_error,
        created_at_ms,
        started_at_ms,
        completed_at_ms,
    })
}

/// Exponential backoff: attempt 1 → 60s, 2 → 120s, 3 → 240s, capped at 1h.
fn backoff_ms(attempts_so_far: u32) -> i64 {
    // attempts_so_far is the count BEFORE the next retry's attempt — so the
    // first retry uses attempts_so_far=1, giving base*2^0 = 60s.
    let exp = attempts_so_far.saturating_sub(1).min(20); // cap shift
    let mult = 1i64 << exp; // 1, 2, 4, …
    let raw = RETRY_BASE_MS.saturating_mul(mult);
    raw.min(RETRY_CAP_MS)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::openhuman::memory::tree::jobs::types::{
        AppendBufferPayload, AppendTarget, ExtractChunkPayload, NodeRef,
    };
    use tempfile::TempDir;

    fn test_config() -> (TempDir, Config) {
        let tmp = TempDir::new().unwrap();
        let mut cfg = Config::default();
        cfg.workspace_dir = tmp.path().to_path_buf();
        cfg.memory_tree.embedding_endpoint = None;
        cfg.memory_tree.embedding_model = None;
        cfg.memory_tree.embedding_strict = false;
        (tmp, cfg)
    }

    #[test]
    fn enqueue_and_claim_roundtrip() {
        let (_tmp, cfg) = test_config();
        let payload = ExtractChunkPayload {
            chunk_id: "c1".into(),
        };
        let nj = NewJob::extract_chunk(&payload).unwrap();
        let id = enqueue(&cfg, &nj).unwrap().expect("inserted");

        let claimed = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        assert_eq!(claimed.id, id);
        assert_eq!(claimed.status, JobStatus::Running);
        assert_eq!(claimed.attempts, 1);
        assert!(claimed.locked_until_ms.is_some());

        // Second claim should see no eligible row (the only one is now running).
        let again = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap();
        assert!(again.is_none());

        mark_done(&cfg, &claimed).unwrap();
        let row = get_job(&cfg, &id).unwrap().unwrap();
        assert_eq!(row.status, JobStatus::Done);
        assert!(row.completed_at_ms.is_some());
        assert!(row.locked_until_ms.is_none());
    }

    #[test]
    fn enqueue_dedupes_active_jobs() {
        let (_tmp, cfg) = test_config();
        let payload = ExtractChunkPayload {
            chunk_id: "c1".into(),
        };
        let nj = NewJob::extract_chunk(&payload).unwrap();
        let id1 = enqueue(&cfg, &nj).unwrap();
        let id2 = enqueue(&cfg, &nj).unwrap();
        assert!(id1.is_some());
        assert!(id2.is_none(), "duplicate should be suppressed while ready");
        assert_eq!(count_total(&cfg).unwrap(), 1);
    }

    #[test]
    fn enqueue_after_done_creates_fresh_row() {
        let (_tmp, cfg) = test_config();
        let payload = ExtractChunkPayload {
            chunk_id: "c1".into(),
        };
        let nj = NewJob::extract_chunk(&payload).unwrap();
        let id1 = enqueue(&cfg, &nj).unwrap().unwrap();
        let claimed = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        assert_eq!(claimed.id, id1);
        mark_done(&cfg, &claimed).unwrap();

        // Now the dedupe key is free (partial index excludes 'done').
        let id2 = enqueue(&cfg, &nj).unwrap();
        assert!(id2.is_some());
        assert_ne!(id2.unwrap(), id1);
        assert_eq!(count_total(&cfg).unwrap(), 2);
    }

    #[test]
    fn mark_failed_retries_then_terminates() {
        let (_tmp, cfg) = test_config();
        let payload = AppendBufferPayload {
            node: NodeRef::Leaf {
                chunk_id: "c1".into(),
            },
            target: AppendTarget::Source {
                source_id: "slack:#x".into(),
            },
        };
        let mut nj = NewJob::append_buffer(&payload).unwrap();
        nj.max_attempts = Some(2);
        let id = enqueue(&cfg, &nj).unwrap().unwrap();

        // Fail #1 — should bounce back to 'ready' with future available_at.
        let attempt1 = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        mark_failed(&cfg, &attempt1, "boom").unwrap();
        let row = get_job(&cfg, &id).unwrap().unwrap();
        assert_eq!(row.status, JobStatus::Ready);
        assert!(row.available_at_ms > Utc::now().timestamp_millis());
        assert_eq!(row.last_error.as_deref(), Some("boom"));

        // Force the row available again so the test doesn't hinge on sleep.
        with_connection(&cfg, |c| {
            c.execute(
                "UPDATE mem_tree_jobs SET available_at_ms = 0 WHERE id = ?1",
                params![id],
            )?;
            Ok(())
        })
        .unwrap();

        // Fail #2 — exceeds max_attempts → terminal 'failed'.
        let attempt2 = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        mark_failed(&cfg, &attempt2, "fatal").unwrap();
        let row = get_job(&cfg, &id).unwrap().unwrap();
        assert_eq!(row.status, JobStatus::Failed);
        assert_eq!(row.last_error.as_deref(), Some("fatal"));
        assert!(row.completed_at_ms.is_some());
    }

    /// `mark_failed` scrubs only the log emission, not the persisted
    /// `last_error` column. A reader of `mem_tree_jobs` should still see
    /// the full anyhow / handler-supplied chain so they can root-cause
    /// the failure; the scrub is a defense-in-depth for the log sink only.
    #[test]
    fn mark_failed_persists_full_error_unredacted() {
        let (_tmp, cfg) = test_config();
        let payload = AppendBufferPayload {
            node: NodeRef::Leaf {
                chunk_id: "c1".into(),
            },
            target: AppendTarget::Source {
                source_id: "slack:#x".into(),
            },
        };
        let mut nj = NewJob::append_buffer(&payload).unwrap();
        nj.max_attempts = Some(1);
        let id = enqueue(&cfg, &nj).unwrap().unwrap();

        let claim = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        let raw = "upstream returned 401: Bearer abc123.def-456 token rejected";
        mark_failed(&cfg, &claim, raw).unwrap();

        let row = get_job(&cfg, &id).unwrap().unwrap();
        assert_eq!(row.status, JobStatus::Failed);
        // The persisted column keeps the full original — the scrub is
        // applied at log emission only.
        assert_eq!(row.last_error.as_deref(), Some(raw));
    }

    /// Same contract for `mark_deferred`: the log line is scrubbed in
    /// `worker::run_once`, but the persisted `last_error` keeps the
    /// full handler-supplied reason for diagnostics.
    #[test]
    fn mark_deferred_persists_full_reason_unredacted() {
        let (_tmp, cfg) = test_config();
        let payload = AppendBufferPayload {
            node: NodeRef::Leaf {
                chunk_id: "c2".into(),
            },
            target: AppendTarget::Source {
                source_id: "slack:#y".into(),
            },
        };
        let nj = NewJob::append_buffer(&payload).unwrap();
        let id = enqueue(&cfg, &nj).unwrap().unwrap();

        let claim = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        let raw = "rate_limited by api_key=hunter2; retry after 30s";
        mark_deferred(&cfg, &claim, 0, raw).unwrap();

        let row = get_job(&cfg, &id).unwrap().unwrap();
        assert_eq!(row.last_error.as_deref(), Some(raw));
    }

    #[test]
    fn recover_stale_locks_resets_running_rows() {
        let (_tmp, cfg) = test_config();
        let payload = ExtractChunkPayload {
            chunk_id: "c1".into(),
        };
        let nj = NewJob::extract_chunk(&payload).unwrap();
        let id = enqueue(&cfg, &nj).unwrap().unwrap();

        // Claim with a lock window that's already in the past so recovery
        // sees it as expired.
        let _ = claim_next(&cfg, -1).unwrap().unwrap();

        let recovered = recover_stale_locks(&cfg).unwrap();
        assert_eq!(recovered, 1);
        let row = get_job(&cfg, &id).unwrap().unwrap();
        assert_eq!(row.status, JobStatus::Ready);
    }

    /// Happy path: a non-stale settlement still succeeds after the claim-token
    /// check is applied. Regression guard so the common case isn't broken.
    #[test]
    fn mark_done_succeeds_for_current_lessee() {
        let (_tmp, cfg) = test_config();
        let payload = ExtractChunkPayload {
            chunk_id: "c-happy".into(),
        };
        let nj = NewJob::extract_chunk(&payload).unwrap();
        let id = enqueue(&cfg, &nj).unwrap().expect("inserted");

        let claimed = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        assert_eq!(claimed.id, id);

        // Current lessee should settle successfully.
        mark_done(&cfg, &claimed).unwrap();
        let row = get_job(&cfg, &id).unwrap().unwrap();
        assert_eq!(row.status, JobStatus::Done);
        assert!(row.completed_at_ms.is_some());
        assert!(row.locked_until_ms.is_none());
    }

    /// Stale-worker settlement is a no-op: after a lock expires and a second
    /// worker re-claims the job, the first worker's `mark_done` must not
    /// clobber the new lessee's row.
    #[test]
    fn stale_worker_settlement_is_noop() {
        let (_tmp, cfg) = test_config();
        let payload = ExtractChunkPayload {
            chunk_id: "c-stale".into(),
        };
        let nj = NewJob::extract_chunk(&payload).unwrap();
        let id = enqueue(&cfg, &nj).unwrap().expect("inserted");

        // Worker A claims with a lock that's already expired (negative window).
        let worker_a_job = claim_next(&cfg, -1).unwrap().unwrap();
        assert_eq!(worker_a_job.id, id);
        assert_eq!(worker_a_job.attempts, 1);

        // Simulate lease expiry: recover_stale_locks resets the row to 'ready'.
        let recovered = recover_stale_locks(&cfg).unwrap();
        assert_eq!(recovered, 1);

        // Worker B claims the reset row — different lease token (attempts=2).
        let worker_b_job = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        assert_eq!(worker_b_job.id, id);
        assert_eq!(worker_b_job.attempts, 2);

        // Worker A (stale) tries to mark done using its old claim snapshot.
        mark_done(&cfg, &worker_a_job).unwrap(); // must NOT return Err

        // Worker B's row must be untouched — still 'running' with attempts=2.
        let row = get_job(&cfg, &id).unwrap().unwrap();
        assert_eq!(
            row.status,
            JobStatus::Running,
            "stale settlement must not clobber Worker B's running row"
        );
        assert_eq!(
            row.attempts, 2,
            "attempts must still reflect Worker B's claim"
        );
    }

    /// Same contract as stale_worker_settlement_is_noop but for mark_failed.
    #[test]
    fn stale_worker_mark_failed_is_noop() {
        let (_tmp, cfg) = test_config();
        let payload = ExtractChunkPayload {
            chunk_id: "c-stale-fail".into(),
        };
        let nj = NewJob::extract_chunk(&payload).unwrap();
        let id = enqueue(&cfg, &nj).unwrap().expect("inserted");

        // Worker A claims with an already-expired lock.
        let worker_a_job = claim_next(&cfg, -1).unwrap().unwrap();
        assert_eq!(worker_a_job.attempts, 1);

        // Lease expires, recovered, Worker B re-claims.
        let recovered = recover_stale_locks(&cfg).unwrap();
        assert_eq!(recovered, 1);
        let worker_b_job = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        assert_eq!(worker_b_job.attempts, 2);

        // Worker A (stale) tries to record a failure — must be a no-op.
        mark_failed(&cfg, &worker_a_job, "stale error").unwrap();

        // Worker B's row must be untouched.
        let row = get_job(&cfg, &id).unwrap().unwrap();
        assert_eq!(
            row.status,
            JobStatus::Running,
            "stale mark_failed must not clobber Worker B's running row"
        );
        assert_ne!(
            row.last_error.as_deref(),
            Some("stale error"),
            "stale error must not be written to the row"
        );
        assert_eq!(row.attempts, 2);
    }

    #[test]
    fn backoff_grows_then_caps() {
        assert_eq!(backoff_ms(1), 60_000);
        assert_eq!(backoff_ms(2), 120_000);
        assert_eq!(backoff_ms(3), 240_000);
        // Eventually clamps at the cap.
        assert_eq!(backoff_ms(20), RETRY_CAP_MS);
        assert_eq!(backoff_ms(99), RETRY_CAP_MS);
    }

    #[test]
    fn count_by_status_reports_each_state() {
        let (_tmp, cfg) = test_config();
        for i in 0..3 {
            let p = ExtractChunkPayload {
                chunk_id: format!("c{i}"),
            };
            let nj = NewJob::extract_chunk(&p).unwrap();
            enqueue(&cfg, &nj).unwrap();
        }
        assert_eq!(count_by_status(&cfg, JobStatus::Ready).unwrap(), 3);
        let claimed = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        mark_done(&cfg, &claimed).unwrap();
        assert_eq!(count_by_status(&cfg, JobStatus::Done).unwrap(), 1);
        assert_eq!(count_by_status(&cfg, JobStatus::Ready).unwrap(), 2);
    }

    /// Defer must NOT advance the failure-attempt counter. After a claim
    /// (which bumps attempts to 1) and a deferral, the next claim should
    /// see attempts==2 (just the second claim's bump) — proving the
    /// transient deferral did not burn a slot from the row's failure
    /// budget.
    #[test]
    fn mark_deferred_does_not_increment_attempts() {
        let (_tmp, cfg) = test_config();
        let payload = ExtractChunkPayload {
            chunk_id: "c-defer-1".into(),
        };
        let nj = NewJob::extract_chunk(&payload).unwrap();
        let id = enqueue(&cfg, &nj).unwrap().expect("inserted");

        let claim1 = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        assert_eq!(claim1.id, id);
        assert_eq!(claim1.attempts, 1, "first claim bumps attempts to 1");

        // Defer with a wake-up time already in the past so the next
        // claim_next is immediately eligible without sleeping.
        mark_deferred(&cfg, &claim1, 0, "rate_limited").unwrap();

        let row = get_job(&cfg, &id).unwrap().unwrap();
        assert_eq!(row.status, JobStatus::Ready);
        assert_eq!(
            row.attempts, 0,
            "Defer must revert the claim's attempts bump (back to pre-claim 0)"
        );
        assert_eq!(row.last_error.as_deref(), Some("rate_limited"));
        assert_eq!(row.available_at_ms, 0);
        assert!(row.locked_until_ms.is_none());
        assert!(row.started_at_ms.is_none());

        let claim2 = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        assert_eq!(claim2.id, id);
        assert_eq!(
            claim2.attempts, 1,
            "second claim bumps attempts to 1 again (Defer didn't count)"
        );
    }

    /// A row deferred to a future `until_ms` must not be claimable until
    /// the system clock crosses that threshold.
    #[test]
    fn deferred_row_not_claimable_until_until_ms() {
        let (_tmp, cfg) = test_config();
        let payload = ExtractChunkPayload {
            chunk_id: "c-defer-2".into(),
        };
        let nj = NewJob::extract_chunk(&payload).unwrap();
        let id = enqueue(&cfg, &nj).unwrap().expect("inserted");

        let claimed = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        let future_ms = Utc::now().timestamp_millis() + 60_000;
        mark_deferred(&cfg, &claimed, future_ms, "warming_up").unwrap();

        // Right now: not yet eligible.
        let none = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap();
        assert!(
            none.is_none(),
            "deferred row must not be claimable before until_ms"
        );

        // Force the row available again (proxy for "wall clock advanced
        // past until_ms") and confirm claim_next picks it up.
        with_connection(&cfg, |c| {
            c.execute(
                "UPDATE mem_tree_jobs SET available_at_ms = 0 WHERE id = ?1",
                params![id],
            )?;
            Ok(())
        })
        .unwrap();
        let claim2 = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        assert_eq!(claim2.id, id);
        assert_eq!(claim2.attempts, 1, "Defer left attempts at pre-claim 0");
    }

    /// Three rows with three different terminal verbs: Done, Failed (with
    /// retry left, so it bounces back to ready with bumped attempts), and
    /// Defer. After processing, each row's terminal state must reflect its
    /// settlement verb. Critically, the Defer row keeps its pre-claim
    /// `attempts` value while the failed row bumps.
    #[test]
    fn mixed_outcomes_stress() {
        let (_tmp, cfg) = test_config();
        let p_a = ExtractChunkPayload {
            chunk_id: "c-mix-a".into(),
        };
        let p_b = ExtractChunkPayload {
            chunk_id: "c-mix-b".into(),
        };
        let p_c = ExtractChunkPayload {
            chunk_id: "c-mix-c".into(),
        };
        let id_a = enqueue(&cfg, &NewJob::extract_chunk(&p_a).unwrap())
            .unwrap()
            .unwrap();
        let id_b = enqueue(&cfg, &NewJob::extract_chunk(&p_b).unwrap())
            .unwrap()
            .unwrap();
        let id_c = enqueue(&cfg, &NewJob::extract_chunk(&p_c).unwrap())
            .unwrap()
            .unwrap();

        // Claim the three rows in turn (FIFO within same kind/priority).
        let claim_a = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        let claim_b = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        let claim_c = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        // Sanity: three distinct ids covering A/B/C.
        let mut got: Vec<&str> = vec![&claim_a.id, &claim_b.id, &claim_c.id];
        got.sort();
        let mut want = vec![id_a.as_str(), id_b.as_str(), id_c.as_str()];
        want.sort();
        assert_eq!(got, want);

        let until_ms = Utc::now().timestamp_millis() + 30_000;

        // Settle: A=done, B=err (retry path), C=defer.
        mark_done(&cfg, &claim_a).unwrap();
        mark_failed(&cfg, &claim_b, "transient_error").unwrap();
        mark_deferred(&cfg, &claim_c, until_ms, "rate_limited").unwrap();

        // A: terminal done.
        let row_a = get_job(&cfg, &id_a).unwrap().unwrap();
        assert_eq!(row_a.status, JobStatus::Done);
        assert!(row_a.completed_at_ms.is_some());

        // B: retry path — back to ready with bumped attempts (1) and a
        // future available_at_ms from exponential backoff.
        let row_b = get_job(&cfg, &id_b).unwrap().unwrap();
        assert_eq!(row_b.status, JobStatus::Ready);
        assert_eq!(
            row_b.attempts, 1,
            "Err settlement keeps the claim's attempts bump"
        );
        assert!(row_b.available_at_ms > Utc::now().timestamp_millis());
        assert_eq!(row_b.last_error.as_deref(), Some("transient_error"));

        // C: deferred — back to ready with attempts reverted and
        // available_at_ms == until_ms.
        let row_c = get_job(&cfg, &id_c).unwrap().unwrap();
        assert_eq!(row_c.status, JobStatus::Ready);
        assert_eq!(
            row_c.attempts, 0,
            "Defer settlement reverts the claim's attempts bump"
        );
        assert_eq!(row_c.available_at_ms, until_ms);
        assert_eq!(row_c.last_error.as_deref(), Some("rate_limited"));
        assert!(row_c.locked_until_ms.is_none());
        assert!(row_c.started_at_ms.is_none());
    }

    /// Stale-worker `mark_deferred` must be a no-op — same lease-token
    /// gating as `mark_done` / `mark_failed`. After Worker A's claim
    /// expires and Worker B re-claims, Worker A's stale Defer must not
    /// clobber B's running row.
    #[test]
    fn mark_deferred_stale_lease_is_noop() {
        let (_tmp, cfg) = test_config();
        let payload = ExtractChunkPayload {
            chunk_id: "c-defer-stale".into(),
        };
        let nj = NewJob::extract_chunk(&payload).unwrap();
        let id = enqueue(&cfg, &nj).unwrap().expect("inserted");

        // Worker A claims with already-expired lock.
        let worker_a_job = claim_next(&cfg, -1).unwrap().unwrap();
        assert_eq!(worker_a_job.attempts, 1);

        // Lease expires; Worker B re-claims.
        let recovered = recover_stale_locks(&cfg).unwrap();
        assert_eq!(recovered, 1);
        let worker_b_job = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
        assert_eq!(worker_b_job.attempts, 2);

        // Worker A (stale) tries to defer using its old snapshot — must
        // be a silent no-op.
        let stale_until_ms = Utc::now().timestamp_millis() + 999_000;
        mark_deferred(&cfg, &worker_a_job, stale_until_ms, "stale_defer").unwrap();

        // Worker B's row must be untouched: still 'running' with
        // attempts=2 and no stale_defer side effect.
        let row = get_job(&cfg, &id).unwrap().unwrap();
        assert_eq!(
            row.status,
            JobStatus::Running,
            "stale mark_deferred must not clobber Worker B's running row"
        );
        assert_eq!(row.attempts, 2);
        assert_ne!(
            row.last_error.as_deref(),
            Some("stale_defer"),
            "stale defer reason must not be persisted"
        );
        assert_ne!(
            row.available_at_ms, stale_until_ms,
            "stale until_ms must not be persisted"
        );
    }
}
