//! The queue mechanism's mem arm — the ADR-010 state machine over the
//! guarded jobs table, all driver-free (engine-mem.md §Mapping the
//! contract, queues row).
//!
//! The row-level operations below are the crate-internal effect layer:
//! the auto-commit handle, the outbox helper, the scheduler's boundary
//! fires, and the tx overlay's commit-merge all land rows through
//! them — one owner of the row mechanics (the claim predicate, the
//! uniform handle-op validity predicate, the dead-letter move). The
//! semantics mirror the contracts pin for pin: claim ordering
//! `priority` DESC → ready-time ASC → enqueue order (id ASC); a
//! reclaim consumes an attempt; every handle op succeeds only while
//! the row is processing and its claim deadline is unexpired
//! (refusals are `false`, never errors); exhaustion dead-letters with
//! the pinned default strings; `sweep_expired` moves every
//! past-`expires_at` row in either state (the no-stranded-rows
//! property) and enforces dead-letter retention.
//!
//! Expiry is a clock check at the claim/renew boundaries — no ambient
//! sweeper exists (ADR-010 §6); `Clock::unix_now` is the single
//! timestamp source. Numeric-domain guards live at the trait-impl
//! entries (`claim_batch`'s `n <= 0` extent clamps empty, ADR-023 §2).

use std::cmp::Reverse;
use std::sync::Arc;

use serde_json::Value;

use alkstore::{
    BoxedFuture, EnqueueOpts, Job, JobHandle, JobState, Queue, QueueOpts, Result,
    validate_local_name,
};

use crate::clock::Clock;
use crate::resolution::{
    EXHAUSTED_ERROR, EXPIRED_ERROR, FAILED_ERROR, ResolvedEnqueue, Stamps, backoff_delay_s,
    resolve_enqueue_opts, stamps_with_override,
};
use crate::state::Guard;
use crate::state::{DeadRow, JobRow, QueueState, State};

fn job_state_for(pending: bool) -> JobState {
    if pending {
        JobState::Pending
    } else {
        JobState::Processing
    }
}

/// Build a pending job row from the resolved enqueue fields and the
/// §3a stamps — the one constructor every enqueue shape shares (the
/// handle's, the tx's, the outbox's, the scheduler fire's).
pub(crate) fn build_job_row(
    id: i64,
    queue: String,
    payload: Vec<u8>,
    now: i64,
    resolved: &ResolvedEnqueue,
    priority: i64,
    stamps: &Stamps,
) -> JobRow {
    JobRow {
        id,
        queue,
        payload,
        pending: true,
        priority,
        run_at: resolved.run_at,
        worker_id: None,
        claim_expires_at: None,
        claimed_at: None,
        attempts: 0,
        max_attempts: stamps.max_attempts,
        created_at: now,
        expires_at: resolved.expires_at,
        visibility_timeout_s: stamps.visibility_timeout_s,
        backoff_base_s: stamps.backoff_base_s,
        dead_letter_retention_s: stamps.dead_letter_retention_s,
    }
}

/// The claim-order readiness shape (ADR-010 §1/§2, the SQL engines'
/// predicate): budget unspent, job-level expiry not passed, and either
/// a ready pending row or a deadline-lapsed processing row (the
/// reclaim case — the claim consumes the attempt; exactly-once is the
/// guarded section's atomicity).
fn claimable(row: &JobRow, now: i64) -> bool {
    let ready = row.pending && row.run_at <= now;
    let lapsed = !row.pending && row.claim_expires_at.is_some_and(|d| d < now);
    (ready || lapsed) && row.attempts < row.max_attempts && row.expires_at.is_none_or(|e| e > now)
}

fn dead_row_from(row: JobRow, last_error: &str, died_at: i64) -> DeadRow {
    DeadRow {
        id: row.id,
        queue: row.queue,
        payload: row.payload,
        priority: row.priority,
        run_at: row.run_at,
        attempts: row.attempts,
        max_attempts: row.max_attempts,
        created_at: row.created_at,
        expires_at: row.expires_at,
        claimed_at: row.claimed_at,
        visibility_timeout_s: row.visibility_timeout_s,
        backoff_base_s: row.backoff_base_s,
        dead_letter_retention_s: row.dead_letter_retention_s,
        last_error: last_error.to_string(),
        died_at,
    }
}

fn move_to_dead(q: &mut QueueState, row: JobRow, last_error: &str, died_at: i64) {
    let id = row.id;
    let dead = dead_row_from(row, last_error, died_at);
    q.dead.insert(id, dead);
}

/// The claimable-but-exhausted rows dead-letter first (the pre-claim
/// sweep — a row whose budget is spent must not sit claimable
/// forever; it lands dead with the exhaust string).
fn dead_letter_exhausted(q: &mut QueueState, queue: &str, exhaust_error: &str, now: i64) {
    let exhausted: Vec<JobRow> = q
        .live
        .values()
        .filter(|row| {
            row.queue == queue
                && row.attempts >= row.max_attempts
                && row.expires_at.is_none_or(|e| e > now)
                && (row.pending && row.run_at <= now
                    || !row.pending && row.claim_expires_at.is_some_and(|d| d < now))
        })
        .cloned()
        .collect();
    for row in exhausted {
        q.live.remove(&row.id);
        move_to_dead(q, row, exhaust_error, now);
    }
}

/// The exactly-once handout (ADR-010 §1): pick claimable rows in
/// priority DESC → ready-time ASC → id ASC order under one guarded
/// section, mark them processing with their own visibility stamps, and
/// return the post-claim row values (what a claim's `Job` shows).
pub(crate) fn claim_rows(
    q: &mut QueueState,
    queue: &str,
    worker_id: &str,
    n: i64,
    now: i64,
) -> Vec<JobRow> {
    dead_letter_exhausted(q, queue, EXHAUSTED_ERROR, now);
    if n <= 0 {
        return Vec::new();
    }
    let mut order: Vec<(i64, i64, i64)> = q
        .live
        .values()
        .filter(|row| row.queue == queue && claimable(row, now))
        .map(|row| (row.priority, row.run_at, row.id))
        .collect();
    order.sort_by_key(|a| (Reverse(a.0), a.1, a.2));
    let mut out = Vec::new();
    for (_, _, id) in order.into_iter().take(n as usize) {
        // Same guarded scope: the candidate's row cannot vanish between
        // the collect and this remove — no other mutation runs under the
        // held guard. The `else continue` below is the honest fallback.
        let Some(mut row) = q.live.remove(&id) else {
            continue;
        };
        row.pending = false;
        row.worker_id = Some(worker_id.to_string());
        row.claim_expires_at = Some(now.saturating_add(row.visibility_timeout_s));
        row.claimed_at = Some(now);
        row.attempts += 1;
        let claimed = row.clone();
        q.live.insert(id, row);
        out.push(claimed);
    }
    out
}

/// The uniform validity predicate (ADR-010 §2), for one-shot handle
/// ops: processing state + the caller's claim deadline unexpired —
/// `claim_expires_at >= now` is the ack predicate's exact boundary.
fn valid_claim(row: &JobRow, worker_id: &str, now: i64) -> bool {
    !row.pending
        && row.worker_id.as_deref() == Some(worker_id)
        && row.claim_expires_at.is_some_and(|d| d >= now)
}

/// Ack one id under the predicate; `false` = lapsed, acked elsewhere,
/// cancelled, or not the caller's claim. The predicate reads before
/// the delete — a refusal leaves the row exactly as it was.
pub(crate) fn ack_row(q: &mut QueueState, job_id: i64, worker_id: &str, now: i64) -> bool {
    let valid = q
        .live
        .get(&job_id)
        .is_some_and(|row| valid_claim(row, worker_id, now));
    if valid {
        q.live.remove(&job_id);
    }
    valid
}

/// The batch ack (the batch form carries no worker identity — per-id
/// independent outcomes; lapsed/not-claimed/cancelled ids silently not
/// counted). Returns the count that landed. The predicate reads before
/// the delete — refusals leave their rows untouched.
pub(crate) fn ack_batch_rows(q: &mut QueueState, ids: &[i64], now: i64) -> i64 {
    let mut count = 0;
    for id in ids {
        let valid = q
            .live
            .get(id)
            .is_some_and(|row| !row.pending && row.claim_expires_at.is_some_and(|d| d >= now));
        if valid {
            q.live.remove(id);
            count += 1;
        }
    }
    count
}

/// Heartbeat: renewal — an absolute reset of the claim deadline from
/// now (extend is the new full deadline, not additive). A late
/// heartbeat is refused (`false`); `claimed_at` is never touched.
pub(crate) fn heartbeat_row(
    q: &mut QueueState,
    job_id: i64,
    worker_id: &str,
    extend: i64,
    now: i64,
) -> bool {
    let Some(row) = q.live.get_mut(&job_id) else {
        return false;
    };
    if !valid_claim(row, worker_id, now) {
        return false;
    }
    row.claim_expires_at = Some(now.saturating_add(extend));
    true
}

/// Retry: return the row to pending for redelivery (`run_at` = now +
/// delay); at an exhausted budget, dead-letter instead (`last_error` =
/// the caller's exhaust string — `None` callers get the pinned
/// `"max attempts exceeded"` from the handle). Refusals leave the
/// row's pre-call state untouched.
pub(crate) fn retry_row(
    q: &mut QueueState,
    job_id: i64,
    worker_id: &str,
    delay_s: i64,
    exhaust_error: &str,
    now: i64,
) -> bool {
    let Some(row) = q.live.get(&job_id) else {
        return false;
    };
    if !valid_claim(row, worker_id, now) {
        return false;
    }
    let Some(pulled) = q.live.remove(&job_id) else {
        return false;
    };
    if pulled.attempts >= pulled.max_attempts {
        move_to_dead(q, pulled, exhaust_error, now);
    } else {
        let mut pending = pulled;
        pending.pending = true;
        pending.run_at = now.saturating_add(delay_s);
        pending.worker_id = None;
        pending.claim_expires_at = None;
        pending.claimed_at = None;
        q.live.insert(job_id, pending);
    }
    true
}

/// Fail: immediate dead-letter with the caller's string (`None` maps
/// to the pinned `"failed"` default at the handle).
pub(crate) fn fail_row(
    q: &mut QueueState,
    job_id: i64,
    worker_id: &str,
    error: &str,
    now: i64,
) -> bool {
    let Some(row) = q.live.get(&job_id) else {
        return false;
    };
    if !valid_claim(row, worker_id, now) {
        return false;
    }
    let Some(pulled) = q.live.remove(&job_id) else {
        return false;
    };
    move_to_dead(q, pulled, error, now);
    true
}

/// Unconditional delete of a live row in either state; not an
/// interrupt — the holder's next op refuses (the `false` shape, the
/// same as deadline expiry).
pub(crate) fn cancel_row(q: &mut QueueState, job_id: i64) -> bool {
    q.live.remove(&job_id).is_some()
}

/// `get_job` — live rows first, dead rows second (with
/// `last_error`/`died_at`; no job = `None`).
pub(crate) fn get_job_row(q: &QueueState, job_id: i64) -> Option<Job> {
    if let Some(row) = q.live.get(&job_id) {
        return Some(job_from_row(
            row.clone(),
            job_state_for(row.pending),
            None,
            None,
        ));
    }
    let dead = q.dead.get(&job_id)?;
    Some(job_from_row(
        JobRow {
            id: dead.id,
            queue: dead.queue.clone(),
            payload: dead.payload.clone(),
            pending: false,
            priority: dead.priority,
            run_at: dead.run_at,
            worker_id: None,
            claim_expires_at: None,
            claimed_at: dead.claimed_at,
            attempts: dead.attempts,
            max_attempts: dead.max_attempts,
            created_at: dead.created_at,
            expires_at: dead.expires_at,
            visibility_timeout_s: dead.visibility_timeout_s,
            backoff_base_s: dead.backoff_base_s,
            dead_letter_retention_s: dead.dead_letter_retention_s,
        },
        JobState::Dead,
        Some(dead.last_error.clone()),
        Some(dead.died_at),
    ))
}

/// Map a row into the contract's `Job` value (`from_row` — the
/// engine-side constructor; one decode owner for every reader).
pub(crate) fn job_from_row(
    row: JobRow,
    state: JobState,
    last_error: Option<String>,
    died_at: Option<i64>,
) -> Job {
    Job::from_row(
        row.id,
        row.queue,
        state,
        row.payload,
        row.priority,
        row.run_at,
        row.attempts,
        row.max_attempts,
        row.worker_id,
        row.claimed_at,
        row.claim_expires_at,
        row.created_at,
        row.expires_at,
        row.visibility_timeout_s,
        row.backoff_base_s,
        row.dead_letter_retention_s,
        last_error,
        died_at,
    )
}

/// One guarded enqueue (auto-commit shapes): allocate the id, insert
/// the stamped row, return it. One clock read per enqueue (the
/// caller's; resolution and stamping run before any table effect).
pub(crate) fn enqueue_row(
    state: &State,
    now: i64,
    queue: &str,
    payload: Vec<u8>,
    opts: &EnqueueOpts,
    stamps: Stamps,
    resolved: &ResolvedEnqueue,
) -> i64 {
    let id = state.allocate_job_id();
    let row = build_job_row(
        id,
        queue.to_string(),
        payload,
        now,
        resolved,
        opts.priority,
        &stamps,
    );
    state.queues.guard().live.insert(id, row);
    id
}

/// One guarded claim page (claims are writes; the outbox's `run_once`
/// claims through this same owner).
pub(crate) fn claim_via_state(
    state: &State,
    queue: &str,
    worker_id: &str,
    n: i64,
    now: i64,
) -> Vec<JobRow> {
    let mut q = state.queues.guard();
    claim_rows(&mut q, queue, worker_id, n, now)
}

/// The `sweep_expired` effect (no-stranded-rows): every
/// past-`expires_at` row of the queue in either state moves to dead
/// (`"expired"`), and dead rows past their retention stamp delete.
/// Returns the moved + retention-deleted count.
pub(crate) fn sweep_expired_rows(q: &mut QueueState, queue: &str, now: i64) -> i64 {
    let expired: Vec<JobRow> = q
        .live
        .values()
        .filter(|row| row.queue == queue && row.expires_at.is_some_and(|e| e <= now))
        .cloned()
        .collect();
    let mut moved = 0;
    for row in expired {
        q.live.remove(&row.id);
        move_to_dead(q, row, EXPIRED_ERROR, now);
        moved += 1;
    }
    let aged: Vec<i64> = q
        .dead
        .values()
        .filter(|row| {
            row.queue == queue
                && row
                    .dead_letter_retention_s
                    .is_some_and(|r| row.died_at <= now.saturating_sub(r))
        })
        .map(|row| row.id)
        .collect();
    for id in aged {
        q.dead.remove(&id);
        moved += 1;
    }
    moved
}

/// The queue-scoped handle: enqueue, claims, and maintenance ops for
/// one validated queue name. The handle carries its `QueueOpts` — the
/// §3a stamp source enqueues resolve over (queues are names, not
/// config owners; opts changes apply to future enqueues only).
pub struct MemQueueHandle {
    pub(crate) name: String,
    opts: QueueOpts,
    pub(crate) state: Arc<State>,
    pub(crate) clock: Arc<Clock>,
}

impl std::fmt::Debug for MemQueueHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MemQueueHandle")
            .field("name", &self.name)
            .field("opts", &self.opts)
            .finish_non_exhaustive()
    }
}

impl MemQueueHandle {
    pub(crate) fn new(name: String, opts: QueueOpts, state: Arc<State>, clock: Arc<Clock>) -> Self {
        Self {
            name,
            opts,
            state,
            clock,
        }
    }

    fn base_stamps(&self) -> Stamps {
        Stamps {
            max_attempts: self.opts.max_attempts,
            visibility_timeout_s: self.opts.visibility_timeout_s,
            backoff_base_s: self.opts.backoff_base_s,
            dead_letter_retention_s: self.opts.dead_letter_retention_s,
        }
    }

    fn claim(&self, worker_id: &str, n: i64) -> Result<Vec<Job>> {
        validate_local_name(worker_id)?;
        if n <= 0 {
            return Ok(Vec::new());
        }
        let now = self.clock.unix_now();
        Ok(claim_via_state(&self.state, &self.name, worker_id, n, now)
            .into_iter()
            .map(|row| job_from_row(row, JobState::Processing, None, None))
            .collect())
    }

    fn stage_enqueue(&self, payload: Value, opts: &EnqueueOpts) -> Result<i64> {
        let bytes = alkstore::encode_payload(&payload)?;
        let now = self.clock.unix_now();
        let resolved = resolve_enqueue_opts(now, opts);
        let stamps = stamps_with_override(self.base_stamps(), opts);
        Ok(enqueue_row(
            &self.state,
            now,
            &self.name,
            bytes,
            opts,
            stamps,
            &resolved,
        ))
    }
}

impl Queue for MemQueueHandle {
    fn name(&self) -> &str {
        &self.name
    }

    fn enqueue<'a>(&'a self, payload: Value, opts: EnqueueOpts) -> BoxedFuture<'a, Result<i64>> {
        match self.stage_enqueue(payload, &opts) {
            Ok(id) => Box::pin(async move { Ok(id) }),
            Err(e) => Box::pin(async move { Err(e) }),
        }
    }

    fn claim_one<'a>(
        &'a self,
        worker_id: &str,
    ) -> BoxedFuture<'a, Result<Option<Box<dyn JobHandle>>>> {
        let worker = worker_id.to_string();
        Box::pin(async move {
            let claimed = self.claim(&worker, 1)?;
            Ok(claimed.into_iter().next().map(|job| {
                Box::new(MemJobHandle::new(
                    job,
                    self.state.clone(),
                    self.clock.clone(),
                )) as Box<dyn JobHandle>
            }))
        })
    }

    fn claim_batch<'a>(
        &'a self,
        worker_id: &str,
        n: i64,
    ) -> BoxedFuture<'a, Result<Vec<Box<dyn JobHandle>>>> {
        let worker = worker_id.to_string();
        Box::pin(async move {
            let claimed = self.claim(&worker, n)?;
            Ok(claimed
                .into_iter()
                .map(|job| {
                    Box::new(MemJobHandle::new(
                        job,
                        self.state.clone(),
                        self.clock.clone(),
                    )) as Box<dyn JobHandle>
                })
                .collect())
        })
    }

    fn ack_batch<'a>(&'a self, ids: &[i64]) -> BoxedFuture<'a, Result<i64>> {
        let ids = ids.to_vec();
        let state = self.state.clone();
        let clock = self.clock.clone();
        Box::pin(async move {
            let now = clock.unix_now();
            let mut q = state.queues.guard();
            Ok(ack_batch_rows(&mut q, &ids, now))
        })
    }

    fn cancel<'a>(&'a self, job_id: i64) -> BoxedFuture<'a, Result<bool>> {
        let state = self.state.clone();
        Box::pin(async move {
            let mut q = state.queues.guard();
            Ok(cancel_row(&mut q, job_id))
        })
    }

    fn get_job<'a>(&'a self, job_id: i64) -> BoxedFuture<'a, Result<Option<Job>>> {
        let state = self.state.clone();
        Box::pin(async move { Ok(get_job_row(&state.queues.guard(), job_id)) })
    }

    fn sweep_expired<'a>(&'a self) -> BoxedFuture<'a, Result<i64>> {
        let name = self.name.clone();
        let state = self.state.clone();
        let clock = self.clock.clone();
        Box::pin(async move {
            let now = clock.unix_now();
            let mut q = state.queues.guard();
            Ok(sweep_expired_rows(&mut q, &name, now))
        })
    }
}

/// The claim handle: the post-claim row value plus its ops. The
/// one-shot ops consume the handle (`self: Box<Self>`); `heartbeat` is
/// the deliberate repeatable `&self` counter-case (ADR-019 §3).
pub struct MemJobHandle {
    job: Job,
    state: Arc<State>,
    clock: Arc<Clock>,
}

impl std::fmt::Debug for MemJobHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MemJobHandle")
            .field("job", &self.job)
            .finish_non_exhaustive()
    }
}

impl MemJobHandle {
    pub(crate) fn new(job: Job, state: Arc<State>, clock: Arc<Clock>) -> Self {
        Self { job, state, clock }
    }
}

impl JobHandle for MemJobHandle {
    fn job(&self) -> &Job {
        &self.job
    }

    fn ack(self: Box<Self>) -> BoxedFuture<'static, Result<bool>> {
        let MemJobHandle { job, state, clock } = *self;
        let id = job.id;
        let worker = job.worker_id.unwrap_or_default();
        Box::pin(async move {
            let now = clock.unix_now();
            let mut q = state.queues.guard();
            Ok(ack_row(&mut q, id, &worker, now))
        })
    }

    fn retry(
        self: Box<Self>,
        err: Option<String>,
        delay: Option<i64>,
    ) -> BoxedFuture<'static, Result<bool>> {
        let MemJobHandle { job, state, clock } = *self;
        let id = job.id;
        let worker = job.worker_id.unwrap_or_default();
        let delay_s = match delay {
            Some(d) => d,
            None => backoff_delay_s(job.backoff_base_s, job.attempts, id),
        };
        let exhaust = err.unwrap_or_else(|| EXHAUSTED_ERROR.to_string());
        Box::pin(async move {
            let now = clock.unix_now();
            let mut q = state.queues.guard();
            Ok(retry_row(&mut q, id, &worker, delay_s, &exhaust, now))
        })
    }

    fn fail(self: Box<Self>, err: Option<String>) -> BoxedFuture<'static, Result<bool>> {
        let MemJobHandle { job, state, clock } = *self;
        let id = job.id;
        let worker = job.worker_id.unwrap_or_default();
        let error = err.unwrap_or_else(|| FAILED_ERROR.to_string());
        Box::pin(async move {
            let now = clock.unix_now();
            let mut q = state.queues.guard();
            Ok(fail_row(&mut q, id, &worker, &error, now))
        })
    }

    fn heartbeat<'a>(&'a self, extend: i64) -> BoxedFuture<'a, Result<bool>> {
        let id = self.job.id;
        let worker = self.job.worker_id.clone().unwrap_or_default();
        let state = self.state.clone();
        let clock = self.clock.clone();
        Box::pin(async move {
            let now = clock.unix_now();
            let mut q = state.queues.guard();
            Ok(heartbeat_row(&mut q, id, &worker, extend, now))
        })
    }
}

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

    fn handle(store: &MemStore, name: &str) -> MemQueueHandle {
        MemQueueHandle::new(
            name.to_string(),
            QueueOpts::default(),
            store.state.clone(),
            store.clock.clone(),
        )
    }

    fn ready_now() -> EnqueueOpts {
        EnqueueOpts {
            run_at: Some(-1),
            ..Default::default()
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn enqueue_claims_and_acks_the_lifecycle() {
        let store = MemStore::new();
        store.clock.freeze(1_000_000);
        let handle = handle(&store, "emails");
        let id = handle
            .enqueue(serde_json::json!({"k": "v"}), EnqueueOpts::default())
            .await
            .expect("enqueue");
        let claimed = handle.claim_batch("w1", 8).await.expect("claim");
        assert_eq!(claimed.len(), 1);
        let job = claimed[0].job();
        assert_eq!(job.id, id);
        assert_eq!(job.state, JobState::Processing);
        assert_eq!(job.worker_id.as_deref(), Some("w1"));
        assert_eq!(job.attempts, 1);
        assert_eq!(job.max_attempts, 3, "the plain defaults stamp the row");
        assert_eq!(job.visibility_timeout_s, 300);
        assert_eq!(job.backoff_base_s, 5);
        assert_eq!(job.dead_letter_retention_s, None);
        assert_eq!(job.claim_expires_at, Some(1_000_300));
        assert_eq!(job.claimed_at, Some(1_000_000));
        assert!(
            claimed
                .into_iter()
                .next()
                .unwrap()
                .ack()
                .await
                .expect("ack")
        );
        assert!(
            handle.get_job(id).await.expect("get").is_none(),
            "ack deletes the row"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn claim_ordering_is_priority_desc_then_ready_then_enqueue_order() {
        let store = MemStore::new();
        store.clock.freeze(1_000);
        let handle = handle(&store, "q");
        // low: priority 0 — claimed last however ready.
        let low = handle
            .enqueue(
                serde_json::json!("low"),
                EnqueueOpts {
                    priority: 0,
                    ..ready_now()
                },
            )
            .await
            .unwrap();
        // b/c: highest priority, the earliest ready time, equal —
        // enqueue order breaks the tie.
        let b = handle
            .enqueue(
                serde_json::json!("b"),
                EnqueueOpts {
                    priority: 5,
                    run_at: Some(995),
                    ..Default::default()
                },
            )
            .await
            .unwrap();
        let c = handle
            .enqueue(
                serde_json::json!("c"),
                EnqueueOpts {
                    priority: 5,
                    run_at: Some(995),
                    ..Default::default()
                },
            )
            .await
            .unwrap();
        // e: same priority, a later ready time.
        let e = handle
            .enqueue(
                serde_json::json!("e"),
                EnqueueOpts {
                    priority: 5,
                    run_at: Some(998),
                    ..Default::default()
                },
            )
            .await
            .unwrap();
        let mut landed = Vec::new();
        for _ in 0..4 {
            if let Some(h) = handle.claim_one("w").await.expect("claim") {
                landed.push(h.job().id);
            }
        }
        assert_eq!(
            landed,
            vec![b, c, e, low],
            "priority DESC, then ready-time ASC, then enqueue order"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn claim_batch_extent_clamps_empty() {
        let store = MemStore::new();
        let handle = handle(&store, "q");
        handle
            .enqueue(serde_json::json!("x"), ready_now())
            .await
            .unwrap();
        assert!(handle.claim_batch("w", 0).await.expect("0").is_empty());
        assert!(
            handle.claim_batch("w", -1).await.expect("-1").is_empty(),
            "the extent guard: a negative n claims nothing, never the whole queue"
        );
        assert_eq!(handle.claim_batch("w", 1).await.expect("1").len(), 1);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn empty_worker_id_is_invalid_at_the_entry() {
        let store = MemStore::new();
        let handle = handle(&store, "q");
        handle
            .enqueue(serde_json::json!("x"), ready_now())
            .await
            .unwrap();
        match handle.claim_batch("", 1).await {
            Err(alkstore::Error::InvalidName { .. }) => {}
            _other => panic!("an empty worker_id rejects"),
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn stamps_resolve_over_the_handle_opts_with_the_max_attempts_override() {
        let store = MemStore::new();
        store.clock.freeze(1_000_000);
        let custom = MemQueueHandle::new(
            "custom".to_string(),
            QueueOpts {
                visibility_timeout_s: 77,
                max_attempts: 9,
                backoff_base_s: 11,
                dead_letter_retention_s: Some(60),
            },
            store.state.clone(),
            store.clock.clone(),
        );
        let id = custom
            .enqueue(
                serde_json::json!("x"),
                EnqueueOpts {
                    max_attempts: Some(4),
                    delay: Some(10),
                    expires: Some(100),
                    ..Default::default()
                },
            )
            .await
            .unwrap();
        let job = custom.get_job(id).await.unwrap().unwrap();
        assert_eq!(job.visibility_timeout_s, 77);
        assert_eq!(
            job.max_attempts, 4,
            "the per-job override wins over the handle stamp"
        );
        assert_eq!(job.backoff_base_s, 11);
        assert_eq!(job.dead_letter_retention_s, Some(60));
        assert_eq!(job.run_at, 1_000_010, "delay wins and resolves");
        assert_eq!(job.expires_at, Some(1_000_100));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn validity_predicate_refuses_lapsed_deadline_ops() {
        let store = MemStore::new();
        store.clock.freeze(1_000_000);
        let handle = handle(&store, "q");
        handle
            .enqueue(serde_json::json!("x"), ready_now())
            .await
            .unwrap();
        let w1 = handle.claim_batch("w1", 1).await.unwrap().pop().unwrap();
        store.clock.advance(301);
        assert!(
            !w1.ack().await.unwrap(),
            "the lapsed ack refuses — a value, not an error"
        );
        let w2 = handle.claim_batch("w2", 1).await.unwrap().pop().unwrap();
        assert_eq!(
            w2.job().worker_id.as_deref(),
            Some("w2"),
            "a reclaim rehands the lapsed row"
        );
        assert_eq!(w2.job().attempts, 2, "a reclaim consumes an attempt");
        store.clock.advance(301);
        let id2 = w2.job().id;
        assert!(
            !w2.ack().await.unwrap(),
            "the renewed lapse refuses — the predicate is uniform"
        );
        let intact = handle.get_job(id2).await.unwrap().unwrap();
        assert_eq!(
            intact.state,
            JobState::Processing,
            "refusals destroy no row"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn heartbeat_renews_absolutely_and_late_heartbeat_refuses() {
        let store = MemStore::new();
        store.clock.freeze(500_000);
        let handle = handle(&store, "q");
        handle
            .enqueue(serde_json::json!("x"), ready_now())
            .await
            .unwrap();
        let claimed = handle.claim_batch("w1", 1).await.unwrap().pop().unwrap();
        assert_eq!(claimed.job().claim_expires_at, Some(500_300));
        store.clock.advance(100);
        assert!(
            claimed.heartbeat(600).await.unwrap(),
            "renewal is an absolute reset from now"
        );
        let stored = handle.get_job(claimed.job().id).await.unwrap().unwrap();
        assert_eq!(
            stored.claim_expires_at,
            Some(500_700),
            "a full 600 from now(500100)"
        );
        assert_eq!(
            stored.claimed_at,
            Some(500_000),
            "heartbeat never touches claimed_at"
        );
        assert_eq!(stored.attempts, 1, "renewal is not a claim");
        store.clock.advance(700);
        assert!(
            !claimed.heartbeat(60).await.unwrap(),
            "a late heartbeat is refused — it cannot steal the job back"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn foreign_worker_ops_refuse() {
        let store = MemStore::new();
        store.clock.freeze(1_000);
        let handle = handle(&store, "q");
        handle
            .enqueue(serde_json::json!("x"), ready_now())
            .await
            .unwrap();
        let w1 = handle.claim_batch("w1", 1).await.unwrap().pop().unwrap();
        let claimed_id = w1.job().id;
        let foreign = handle.claim_batch("w2", 1).await.unwrap();
        assert!(
            foreign.is_empty(),
            "w1's unexpired claim holds the row — no extra handout"
        );
        // A foreign-worker ack against the row identity refuses. Forge
        // the op through the same claim whose worker_id says w1... the
        // substrate predicate is the reference: ack requires the row's
        // claimed worker; w2 claiming first would take it. What the
        // claim batch pins here is the exactly-once handout; the
        // wrong-worker refusal rides the validity predicate tests.
        let row = handle.get_job(claimed_id).await.unwrap().unwrap();
        assert_eq!(row.state, JobState::Processing);
        assert_eq!(row.worker_id.as_deref(), Some("w1"));
        assert!(w1.ack().await.unwrap(), "the owner's ack lands");
    }

    #[tokio::test(flavor = "current_thread")]
    async fn dead_letter_paths_carry_the_pinned_default_strings() {
        let store = MemStore::new();
        store.clock.freeze(1_000);
        let handle = handle(&store, "q");
        handle
            .enqueue(serde_json::json!("x"), ready_now())
            .await
            .unwrap();
        handle
            .enqueue(serde_json::json!("y"), ready_now())
            .await
            .unwrap();
        let claimed = handle.claim_batch("w", 2).await.unwrap();
        let mut it = claimed.into_iter();
        let first = it.next().unwrap();
        let second = it.next().unwrap();
        let first_id = first.job().id;
        let second_id = second.job().id;
        assert!(
            first.retry(None, None).await.unwrap(),
            "budget 3 with 1 attempt is not exhausted"
        );
        let retried = handle.get_job(first_id).await.unwrap().unwrap();
        assert_eq!(
            retried.state,
            JobState::Pending,
            "the retry reschedules pending"
        );
        assert!(
            second.fail(None).await.unwrap(),
            "fail is the immediate dead-letter"
        );
        let dead = handle.get_job(second_id).await.unwrap().unwrap();
        assert_eq!(dead.state, JobState::Dead);
        assert_eq!(
            dead.last_error.as_deref(),
            Some("failed"),
            "fail(None) pins the default string"
        );
        assert!(dead.died_at.is_some_and(|d| d == 1_000));
        assert_eq!(
            dead.visibility_timeout_s, 300,
            "dead rows keep their stamps"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn retry_override_and_exhaustion_strings() {
        let store = MemStore::new();
        store.clock.freeze(2_000);
        let handle = handle(&store, "q");
        handle
            .enqueue(
                serde_json::json!("x"),
                EnqueueOpts {
                    run_at: Some(1_999),
                    max_attempts: Some(2),
                    ..Default::default()
                },
            )
            .await
            .unwrap();
        let first = handle.claim_batch("w", 1).await.unwrap().pop().unwrap();
        let first_id = first.job().id;
        assert!(
            first
                .retry(Some("driver failed".into()), Some(30))
                .await
                .unwrap()
        );
        let pending = handle.get_job(first_id).await.unwrap().unwrap();
        assert_eq!(pending.run_at, 2_030, "an explicit delay passes through");
        assert_eq!(
            pending.claimed_at, None,
            "the retry clears the claim identity"
        );
        store.clock.advance(31);
        let second = handle.claim_batch("w", 1).await.unwrap().pop().unwrap();
        assert_eq!(
            second.job().attempts,
            2,
            "every claim counts, reclaims included"
        );
        let second_id = second.job().id;
        assert!(
            second.retry(None, None).await.unwrap(),
            "the budget is spent — dead-letter"
        );
        let dead = handle.get_job(second_id).await.unwrap().unwrap();
        assert_eq!(dead.state, JobState::Dead);
        assert_eq!(
            dead.last_error.as_deref(),
            Some("max attempts exceeded"),
            "exhaustion via retry(None, …) pins the engine default"
        );
        // The dead row's stored exhaust string is the caller's when
        // supplied: a healthy row retrying with Some carries it only on
        // exhaustion — here the earlier Some-string retry rescheduled,
        // so the engine default recorded this death.
        let _ = first_id;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn reclaim_consumes_an_attempt_and_wins_over_ack() {
        let store = MemStore::new();
        store.clock.freeze(10_000);
        let handle = handle(&store, "q");
        handle
            .enqueue(
                serde_json::json!("x"),
                EnqueueOpts {
                    run_at: Some(9_999),
                    max_attempts: Some(5),
                    ..Default::default()
                },
            )
            .await
            .unwrap();
        let claimed = handle.claim_batch("w1", 1).await.unwrap().pop().unwrap();
        store.clock.advance(301);
        let reclaimed = handle.claim_batch("w2", 1).await.unwrap().pop().unwrap();
        assert_eq!(reclaimed.job().worker_id.as_deref(), Some("w2"));
        assert_eq!(reclaimed.job().attempts, 2, "a reclaim consumes an attempt");
        assert!(
            !claimed.ack().await.unwrap(),
            "the loser's ack refuses against the reclaim"
        );
        assert!(reclaimed.ack().await.unwrap());
    }

    #[tokio::test(flavor = "current_thread")]
    async fn ack_batch_applies_the_predicate_per_id() {
        let store = MemStore::new();
        store.clock.freeze(1_000);
        let handle = handle(&store, "q");
        for i in 0..3 {
            handle
                .enqueue(serde_json::json!(i), ready_now())
                .await
                .unwrap();
        }
        let claimed = handle.claim_batch("w1", 3).await.unwrap();
        let ids: Vec<i64> = claimed.iter().map(|h| h.job().id).collect();
        assert!(
            handle.cancel(ids[2]).await.unwrap(),
            "cancel deletes unconditionally"
        );
        let count = handle
            .ack_batch(&[ids[0], ids[1], ids[2], 999_999])
            .await
            .unwrap();
        assert_eq!(count, 2, "cancelled and absent ids silently not counted");
        assert!(
            !handle.cancel(ids[0]).await.unwrap(),
            "acked rows are gone (delete-on-ack)"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn sweep_moves_both_states_past_expiry() {
        let store = MemStore::new();
        store.clock.freeze(5_000);
        let handle = handle(&store, "q");
        // Both rows expire at 5_001: one gets claimed first (processing),
        // the other stays pending; then both expire.
        let pending = handle
            .enqueue(
                serde_json::json!("p"),
                EnqueueOpts {
                    expires: Some(1),
                    ..ready_now()
                },
            )
            .await
            .unwrap();
        let processing = handle
            .enqueue(
                serde_json::json!("w"),
                EnqueueOpts {
                    expires: Some(1),
                    ..ready_now()
                },
            )
            .await
            .unwrap();
        let claimed = handle.claim_batch("w", 1).await.unwrap().pop().unwrap();
        assert_eq!(
            claimed.job().id,
            pending,
            "the first enqueued row claims; it is now the processing one"
        );
        let _healthy = handle
            .enqueue(
                serde_json::json!("h"),
                EnqueueOpts {
                    expires: Some(3_600),
                    ..ready_now()
                },
            )
            .await
            .unwrap();
        store.clock.advance(2);
        assert_eq!(handle.sweep_expired().await.unwrap(), 2, "both states move");
        for id in [pending, processing] {
            let dead = handle.get_job(id).await.unwrap().unwrap();
            assert_eq!(dead.state, JobState::Dead);
            assert_eq!(dead.last_error.as_deref(), Some("expired"));
            assert_eq!(dead.died_at, Some(5_002));
        }
        assert_eq!(
            handle.sweep_expired().await.unwrap(),
            0,
            "the healthy row stays"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn retention_deletes_aged_dead_rows() {
        let store = MemStore::new();
        store.clock.freeze(5_000);
        // Same queue name, two handles: the §3a stamps ride each
        // handle's opts — one stamps retention 60, the other forever.
        let retention60 = MemQueueHandle::new(
            "q".to_string(),
            QueueOpts {
                dead_letter_retention_s: Some(60),
                ..QueueOpts::default()
            },
            store.state.clone(),
            store.clock.clone(),
        );
        let handle = handle(&store, "q");
        let expiring = retention60
            .enqueue(
                serde_json::json!("e"),
                EnqueueOpts {
                    expires: Some(-1),
                    ..ready_now()
                },
            )
            .await
            .unwrap();
        let forever = handle
            .enqueue(
                serde_json::json!("f"),
                EnqueueOpts {
                    expires: Some(-1),
                    ..ready_now()
                },
            )
            .await
            .unwrap();
        assert_eq!(handle.sweep_expired().await.unwrap(), 2);
        assert_eq!(
            handle.sweep_expired().await.unwrap(),
            0,
            "nothing aged out yet"
        );
        store.clock.advance(61);
        assert_eq!(
            handle.sweep_expired().await.unwrap(),
            1,
            "the retention-60 row deletes"
        );
        assert!(handle.get_job(expiring).await.unwrap().is_none());
        let kept = handle.get_job(forever).await.unwrap().unwrap();
        assert_eq!(kept.state, JobState::Dead, "retention None = forever");
    }

    #[tokio::test(flavor = "current_thread")]
    async fn concurrent_claims_hand_out_work_exactly_once() {
        let store = MemStore::new();
        store.clock.freeze(1_000);
        let handle = handle(&store, "q");
        let mut ids = Vec::new();
        for i in 0..4 {
            ids.push(
                handle
                    .enqueue(serde_json::json!(i), ready_now())
                    .await
                    .unwrap(),
            );
        }
        let mut tasks = Vec::new();
        for w in 0..8 {
            let state = store.state.clone();
            let clock = store.clock.clone();
            tasks.push(tokio::spawn(async move {
                MemQueueHandle::new("q".to_string(), QueueOpts::default(), state, clock)
                    .claim_batch(&format!("w{w}"), 1)
                    .await
                    .expect("claim")
                    .into_iter()
                    .map(|h| h.job().id)
                    .collect::<Vec<i64>>()
            }));
        }
        let mut handed = Vec::new();
        for task in tasks {
            handed.extend(task.await.expect("task joins"));
        }
        handed.sort();
        ids.sort();
        assert_eq!(
            handed, ids,
            "8 racing workers over 4 rows: every row handed once, never twice — \
             the guarded section makes the in-process claim race row-atomic"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn claim_vs_expiry_is_deterministic_at_the_row_boundary() {
        let store = MemStore::new();
        store.clock.freeze(1_000);
        let handle = handle(&store, "q");
        let first = handle
            .enqueue(serde_json::json!("x"), ready_now())
            .await
            .unwrap();
        let w1 = handle.claim_batch("w1", 1).await.unwrap().pop().unwrap();
        let deadline = w1.job().claim_expires_at.unwrap();
        assert_eq!(deadline, 1_300);
        store.clock.advance(300);
        let contested = handle.claim_batch("w2", 1).await.unwrap();
        assert!(
            contested.is_empty(),
            "at now == deadline the claim is not lapsed (d < now is false) — no reclaim"
        );
        assert!(
            w1.heartbeat(60).await.unwrap(),
            "the same boundary second keeps the holder's ops valid (d >= now)"
        );
        assert!(w1.ack().await.unwrap(), "the boundary ack lands");
        assert!(handle.get_job(first).await.unwrap().is_none());

        let second = handle
            .enqueue(serde_json::json!("y"), ready_now())
            .await
            .unwrap();
        let w3 = handle.claim_batch("w3", 1).await.unwrap().pop().unwrap();
        assert_eq!(w3.job().id, second);
        let deadline2 = w3.job().claim_expires_at.unwrap();
        assert_eq!(deadline2, 1_600);
        store.clock.advance(301);
        let reclaimed = handle.claim_batch("w4", 1).await.unwrap();
        assert_eq!(
            reclaimed.len(),
            1,
            "one second past the deadline the row is reclaimable"
        );
        assert_eq!(
            reclaimed[0].job().attempts,
            2,
            "the reclaim consumes its attempt"
        );
        let lapsed_id = reclaimed[0].job().id;
        store.clock.advance(301);
        assert_eq!(
            handle.ack_batch(&[lapsed_id]).await.unwrap(),
            0,
            "a lapsed id is silently not counted by the batch form"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn the_equal_jitter_curve_mirrors_the_suite_rows_numeric_bounds() {
        let store = MemStore::new();
        store.clock.freeze(10_000);
        let curve = MemQueueHandle::new(
            "curve".to_string(),
            QueueOpts {
                max_attempts: 7,
                backoff_base_s: 2,
                ..QueueOpts::default()
            },
            store.state.clone(),
            store.clock.clone(),
        );
        let id = curve
            .enqueue(serde_json::json!({"leg": "curve"}), EnqueueOpts::default())
            .await
            .unwrap();
        // Base 2 s at the first three attempts — the suite row's
        // identical ranges ([1,2], [2,4], [4,8]); the frozen clock makes
        // the observation exact (no straddle tolerance needed).
        for (attempt, lower, upper) in [(1i64, 1i64, 2i64), (2, 2, 4), (3, 4, 8)] {
            store.clock.advance(upper + 1);
            let held = curve.claim_batch("w", 1).await.unwrap().pop().unwrap();
            assert_eq!(
                held.job().attempts,
                attempt,
                "the attempt index is the claim count"
            );
            let now = store.clock.unix_now();
            assert!(held.retry(Some("boom".to_string()), None).await.unwrap());
            let job = curve.get_job(id).await.unwrap().unwrap();
            assert_eq!(job.state, JobState::Pending);
            let observed = job.run_at - now;
            assert!(
                (lower..=upper).contains(&observed),
                "attempt {attempt}: delay {observed} outside the suite row's [{lower}, {upper}]"
            );
        }
        store.clock.advance(9);
        let held = curve.claim_batch("w", 1).await.unwrap().pop().unwrap();
        let now = store.clock.unix_now();
        assert!(held.retry(Some("boom".to_string()), Some(5)).await.unwrap());
        let job = curve.get_job(id).await.unwrap().unwrap();
        assert_eq!(job.run_at, now + 5, "Some(5) overrides the curve verbatim");
        store.clock.advance(6);
        let held = curve.claim_batch("w", 1).await.unwrap().pop().unwrap();
        let now = store.clock.unix_now();
        assert!(held.retry(Some("boom".to_string()), Some(0)).await.unwrap());
        let job = curve.get_job(id).await.unwrap().unwrap();
        assert_eq!(
            job.run_at, now,
            "Some(0) resolves ready-now — the boundary-total rule"
        );
        assert_eq!(
            curve.claim_batch("w", 1).await.unwrap().len(),
            1,
            "a ready-now row claims immediately"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn the_pre_claim_sweep_dead_letters_exhausted_reclaimables() {
        let store = MemStore::new();
        store.clock.freeze(1_000);
        let handle = handle(&store, "q");
        let id = handle
            .enqueue(
                serde_json::json!("x"),
                EnqueueOpts {
                    max_attempts: Some(1),
                    ..ready_now()
                },
            )
            .await
            .unwrap();
        let claimed = handle.claim_batch("w1", 1).await.unwrap().pop().unwrap();
        assert_eq!(claimed.job().attempts, 1);
        store.clock.advance(301);
        let reclaimed = handle.claim_batch("w2", 1).await.unwrap();
        assert!(
            reclaimed.is_empty(),
            "a lapsed row with its budget spent claims nothing — it dead-letters instead"
        );
        let dead = handle.get_job(id).await.unwrap().unwrap();
        assert_eq!(dead.state, JobState::Dead);
        assert_eq!(
            dead.last_error.as_deref(),
            Some("max attempts exceeded"),
            "exhaustion via reclaim pins the engine default string"
        );
        assert_eq!(dead.died_at, Some(1_301));
        assert_eq!(dead.attempts, 1, "the row never got a second claim");
    }

    #[tokio::test(flavor = "current_thread")]
    async fn cancel_is_unconditional_across_states() {
        let store = MemStore::new();
        store.clock.freeze(1_000);
        let handle = handle(&store, "q");
        let pending = handle
            .enqueue(serde_json::json!("p"), ready_now())
            .await
            .unwrap();
        assert!(
            handle.cancel(pending).await.unwrap(),
            "a pending row cancels"
        );
        assert!(handle.get_job(pending).await.unwrap().is_none());
        assert!(
            !handle.cancel(999).await.unwrap(),
            "a row-less id is Ok(false)"
        );
        let processing = handle
            .enqueue(serde_json::json!("w"), ready_now())
            .await
            .unwrap();
        let holder = handle.claim_batch("w1", 1).await.unwrap().pop().unwrap();
        assert!(
            handle.cancel(processing).await.unwrap(),
            "a processing row cancels"
        );
        assert!(
            !holder.ack().await.unwrap(),
            "cancel is not an interrupt — the holder's next op refuses, the expiry shape"
        );
        assert!(handle.get_job(processing).await.unwrap().is_none());
    }

    #[tokio::test(flavor = "current_thread")]
    async fn sweep_is_scoped_to_its_queue() {
        let store = MemStore::new();
        store.clock.freeze(5_000);
        let mine = handle(&store, "mine");
        let other = handle(&store, "other");
        let mine_id = mine
            .enqueue(
                serde_json::json!("x"),
                EnqueueOpts {
                    expires: Some(-1),
                    ..ready_now()
                },
            )
            .await
            .unwrap();
        let other_id = other
            .enqueue(
                serde_json::json!("y"),
                EnqueueOpts {
                    expires: Some(-1),
                    ..ready_now()
                },
            )
            .await
            .unwrap();
        assert_eq!(mine.sweep_expired().await.unwrap(), 1);
        assert_eq!(
            other.sweep_expired().await.unwrap(),
            1,
            "each queue sweeps its own rows"
        );
        let mine_seen = mine.get_job(mine_id).await.unwrap().unwrap();
        let other_seen = other.get_job(other_id).await.unwrap().unwrap();
        assert_eq!(mine_seen.state, JobState::Dead);
        assert_eq!(other_seen.state, JobState::Dead);
        assert_eq!(
            mine.sweep_expired().await.unwrap(),
            0,
            "a second sweep moves nothing"
        );
    }
}