alkstore/alkstore-sqlite/src/queue.rs

//! The queue mechanism's SQLite arm (`sqlite-engine-queues`) — the
//! auto-commit counterpart to the tx substrate's queue ops:
//!
//! - `queue(name, opts)` — the validated constructor (construction is
//!   free: validation + a closed-store check — queues are names, not
//!   registered objects, ADR-010 §3a); the handle carries the
//!   [`QueueOpts`] stamps future enqueues resolve over.
//! - `enqueue(payload, opts)` — the one handle-carrying shape: stamps
//!   resolve over **this handle's opts** (ADR-020 §3a's handle
//!   annotation — the handle's visibility/backoff/retention stamp the
//!   row directly; `EnqueueOpts::max_attempts` is the one per-job
//!   override), through the same resolution arithmetic the tx path
//!   uses (one clock read per enqueue, delay-over-`run_at`,
//!   relative-`expires`, `encode_payload` at the seam).
//! - `claim_one` / `claim_batch` — through the writer slot (claims
//!   are writes; the substrate's single-statement claim is the
//!   exactly-once handout, ADR-010 §1). The **extent guard (ADR-023
//!   §2)** short-circuits `n <= 0` to the empty `Vec` at the
//!   trait-impl entry, before the substrate — the `LIMIT -1` dialect
//!   artifact dies here (the substrate passes `n` through
//!   contract-blind). `worker_id` is consumer-local, non-empty only
//!   (ADR-019 §2 — `InvalidName`). Claim rows decode into `Job` +
//!   boxed [`JobHandle`]s via the shared `job_from_json` decode owner.
//! - The [`JobHandle`] impl: `job()` (the claimed row value),
//!   `ack`/`retry`/`fail` (one-shot, `self: Box<Self>` — the
//!   consuming shape; `BoxedFuture<'static>`), `heartbeat(extend)`
//!   (repeatable, absolute reset). The uniform validity predicate is
//!   the substrate's (processing + unexpired claim deadline, ADR-010
//!   §2); refusals are `Ok(false)`, never errors.
//! - **The backoff curve — engine-side, one owner (ADR-010 §3)**:
//!   `retry(err, None)` computes the equal-jitter exponential from the
//!   job's stamped `backoff_base_s`: `delay ∈ [base·2^(a−1)/2,
//!   base·2^(a−1)]`, capped at 1 hour; the attempt index `a` is the
//!   row's `attempts` at the retry (the claim-row value — post-claim
//!   count, reclaims included). `Some(d)` passes through (ADR-023 §2's
//!   boundary class — any `i64` is total).
//! - `ack_batch(ids)` — the contract's batch ack carries no
//!   `worker_id` (ADR-019 §1's surface): the substrate's per-id
//!   validity predicate (processing + unexpired deadline) counts
//!   landed ids; partial success is ordinary (ADR-010 §1).
//! - `cancel(job_id)` (unconditional delete), `get_job(job_id)`
//!   (sees dead rows — `last_error`/`died_at` carried), and
//!   `sweep_expired()` (both-states move + retention enforcement; no
//!   leader lock required, ADR-010 §5) round out the maintenance
//!   surface.
//! - Dead-letter strings (row content, not error variants): the
//!   pinned defaults for the engine-triggered paths
//!   (`"max attempts exceeded"` at exhaustion, `"failed"` for
//!   `fail(None)` per ADR-019 §3's annotation, `"expired"` on sweeps,
//!   ADR-010 §4); a caller-supplied `Some(err)` carries the caller's
//!   string on the path it names.

use std::collections::hash_map::RandomState;
use std::hash::BuildHasher;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};

use serde_json::Value;

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

use crate::resolution::{resolve_enqueue_opts, stamps_with_override};
use crate::seam::{closed_store_error, sqlite_error, with_reader, with_writer};
use crate::substrate::{
    Readers, Stamps, Writer, ack, ack_batch, cancel, claim_batch, enqueue, fail, get_job,
    heartbeat, retry, sweep_expired,
};

/// The pinned dead-letter default strings (row content, ADR-010 §4 /
/// ADR-019 §3's annotation — the engine passes them; the substrate
/// stores them verbatim).
pub(crate) const EXHAUSTED_ERROR: &str = "max attempts exceeded";
pub(crate) const FAILED_ERROR: &str = "failed";
pub(crate) const EXPIRED_ERROR: &str = "expired";

/// The backoff cap (ADR-010 §3 — a pinned contract constant, no knob).
const BACKOFF_CAP_S: i64 = 3600;

/// The queue-scoped handle (see the module docs).
pub struct SqliteQueueHandle {
    name: String,
    opts: QueueOpts,
    writer: Arc<Writer>,
    readers: Arc<Readers>,
    closed: Arc<AtomicBool>,
}

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

impl SqliteQueueHandle {
    pub(crate) fn new(
        name: String,
        opts: QueueOpts,
        writer: Arc<Writer>,
        readers: Arc<Readers>,
        closed: Arc<AtomicBool>,
    ) -> Self {
        Self {
            name,
            opts,
            writer,
            readers,
            closed,
        }
    }

    /// The handle's opts, as the substrate stamps them (the
    /// `max_attempts` override resolves per enqueue over these).
    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,
        }
    }
}

/// Decode the substrate's job-row JSON into the contract value
/// (`Job::from_row` — the engine-side constructor, ADR-021 §2's
/// `claimed_at` surfaced). One decode owner: the tx reads (`tx.rs`)
/// and this module's claim/get rows share it; malformed rows map to
/// `Error::Codec` (the decode-side posture guard — the fork's writer
/// always wrote well-formed rows).
pub(crate) fn job_from_json(value: &Value) -> Result<Option<Job>> {
    let state = match value["state"].as_str() {
        Some("pending") => JobState::Pending,
        Some("processing") => JobState::Processing,
        Some("dead") => JobState::Dead,
        other => {
            return Err(Error::Codec(format!(
                "job row state must be pending|processing|dead, got {other:?}"
            )));
        }
    };
    let payload: Vec<u8> = value["payload"]
        .as_str()
        .ok_or_else(|| Error::Codec("job row payload must be a string".to_string()))?
        .as_bytes()
        .to_vec();
    let field = |name: &str| -> Result<i64> {
        serde_json::from_value(value[name].clone())
            .map_err(|e| Error::Codec(format!("job row {name} must be i64: {e}")))
    };
    let opt_field = |name: &str| -> Result<Option<i64>> {
        if value[name].is_null() {
            Ok(None)
        } else {
            serde_json::from_value(value[name].clone())
                .map_err(|e| Error::Codec(format!("job row {name} must be i64: {e}")))
        }
    };
    let opt_str = |name: &str| -> Result<Option<String>> {
        if value[name].is_null() {
            Ok(None)
        } else {
            serde_json::from_value(value[name].clone())
                .map_err(|e| Error::Codec(format!("job row {name} must be a string: {e}")))
        }
    };
    Ok(Some(Job::from_row(
        field("id")?,
        value["queue"]
            .as_str()
            .ok_or_else(|| Error::Codec("job row queue must be a string".to_string()))?
            .to_string(),
        state,
        payload,
        field("priority")?,
        field("run_at")?,
        field("attempts")?,
        field("max_attempts")?,
        opt_str("worker_id")?,
        opt_field("claimed_at")?,
        opt_field("claim_expires_at")?,
        field("created_at")?,
        opt_field("expires_at")?,
        field("visibility_timeout_s")?,
        field("backoff_base_s")?,
        opt_field("dead_letter_retention_s")?,
        opt_str("last_error")?,
        opt_field("died_at")?,
    )))
}

/// Decode a claim page of job-row JSON into `Job` values. Shared with
/// the outbox's `run_once` claim (the same decode owner).
pub(crate) fn jobs_from_json_page(page: &str) -> Result<Vec<Job>> {
    let array: Value = serde_json::from_str(page)
        .map_err(|e| Error::Codec(format!("claim page must be json: {e}")))?;
    let rows = array
        .as_array()
        .ok_or_else(|| Error::Codec("claim page must be a json array".to_string()))?;
    let mut out = Vec::with_capacity(rows.len());
    for row in rows {
        out.push(
            job_from_json(row)?
                .ok_or_else(|| Error::Codec("claim page row must decode to a job".to_string()))?,
        );
    }
    Ok(out)
}

/// The equal-jitter exponential backoff (ADR-010 §3 — engine-side,
/// one owner; wave 4's pg engine re-derives it identically and wave 5
/// pins equivalence): `delay ∈ [base·2^(a−1)/2, base·2^(a−1)]` from
/// the job's stamped `backoff_base_s`, capped at **1 hour**.
///
/// Integerized second-precision: the draw is uniform over the
/// integers in `[ceil(half), cap]` (inclusive both ends) — never
/// below the pinned lower bound, never above the cap. The attempt
/// index `a` is the row's `attempts` at the retry (post-claim count,
/// reclaims included).
///
/// The jitter draw is std-only (no `rand` dependency without a
/// decision — the ADR-005 dependency posture; documented in the task
/// Notes): a fresh [`RandomState`] hasher (per-process-random seeds)
/// over the row id, the attempt count, and the subsecond clock
/// spreads the draw uniformly over the range. Hash quality is only
/// asked to spread the draw; the pinned property is the **range**,
/// what the acceptance test pins statistically.
pub(crate) fn backoff_delay_s(base_s: i64, attempts: i64, id: i64) -> i64 {
    let exponent: u32 = (attempts.max(1) as u32).saturating_sub(1).min(30);
    let doubling: i64 = base_s.saturating_mul(1i64 << exponent).max(1);
    let cap = doubling.min(BACKOFF_CAP_S);
    let half = ((cap + 1) / 2).max(1);
    let span = (cap - half + 1).max(1);
    let now_nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.subsec_nanos())
        .unwrap_or(0);
    let draw = RandomState::new().hash_one((id, attempts, now_nanos));
    half + (draw % span as u64) as i64
}

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

    fn enqueue<'a>(&'a self, payload: Value, opts: EnqueueOpts) -> BoxedFuture<'a, Result<i64>> {
        if self.closed.load(Ordering::Acquire) {
            return Box::pin(async { Err(closed_store_error()) });
        }
        let name = self.name.clone();
        let base = self.base_stamps();
        Box::pin(async move {
            with_writer(self.writer.clone(), move |conn| {
                let bytes = alkstore::encode_payload(&payload)?;
                let resolved = resolve_enqueue_opts(conn, &opts)?;
                let stamps = stamps_with_override(base, &opts);
                enqueue(
                    conn,
                    &name,
                    std::str::from_utf8(&bytes)
                        .map_err(|e| Error::Codec(format!("row payload must be utf-8: {e}")))?,
                    resolved.run_at,
                    opts.priority,
                    resolved.expires_at,
                    stamps,
                )
                .map_err(sqlite_error)
            })
            .await
        })
    }

    fn claim_one<'a>(
        &'a self,
        worker_id: &str,
    ) -> BoxedFuture<'a, Result<Option<Box<dyn JobHandle>>>> {
        let worker_id = worker_id.to_string();
        Box::pin(async move { Ok(self.claim_batch(&worker_id, 1).await?.into_iter().next()) })
    }

    fn claim_batch<'a>(
        &'a self,
        worker_id: &str,
        n: i64,
    ) -> BoxedFuture<'a, Result<Vec<Box<dyn JobHandle>>>> {
        if let Err(e) = validate_local_name(worker_id) {
            return Box::pin(async move { Err(e) });
        }
        if self.closed.load(Ordering::Acquire) {
            return Box::pin(async { Err(closed_store_error()) });
        }
        // The extent guard (ADR-023 §2) — before the substrate, whose
        // `LIMIT ?` passes `n` through contract-blind (SQLite's
        // `LIMIT -1` = unbounded is exactly the artifact that dies
        // here).
        if n <= 0 {
            return Box::pin(async { Ok(Vec::new()) });
        }
        let name = self.name.clone();
        let writer = self.writer.clone();
        let closed = self.closed.clone();
        let worker = worker_id.to_string();
        Box::pin(async move {
            with_writer(writer.clone(), move |conn| {
                let page =
                    claim_batch(conn, &name, &worker, n, EXHAUSTED_ERROR).map_err(sqlite_error)?;
                Ok(jobs_from_json_page(&page)?
                    .into_iter()
                    .map(|job| {
                        Box::new(SqliteJobHandle {
                            job,
                            writer: writer.clone(),
                            closed: closed.clone(),
                        }) as Box<dyn JobHandle>
                    })
                    .collect())
            })
            .await
        })
    }

    fn ack_batch<'a>(&'a self, ids: &[i64]) -> BoxedFuture<'a, Result<i64>> {
        if self.closed.load(Ordering::Acquire) {
            return Box::pin(async { Err(closed_store_error()) });
        }
        let ids_json =
            serde_json::to_string(ids).map_err(|e| Error::Codec(format!("ids must encode: {e}")));
        Box::pin(async move {
            let ids_json = ids_json?;
            with_writer(self.writer.clone(), move |conn| {
                ack_batch(conn, &ids_json).map_err(sqlite_error)
            })
            .await
        })
    }

    fn cancel<'a>(&'a self, job_id: i64) -> BoxedFuture<'a, Result<bool>> {
        if self.closed.load(Ordering::Acquire) {
            return Box::pin(async { Err(closed_store_error()) });
        }
        Box::pin(async move {
            with_writer(self.writer.clone(), move |conn| {
                Ok(cancel(conn, job_id).map_err(sqlite_error)? > 0)
            })
            .await
        })
    }

    fn get_job<'a>(&'a self, job_id: i64) -> BoxedFuture<'a, Result<Option<Job>>> {
        if self.closed.load(Ordering::Acquire) {
            return Box::pin(async { Err(closed_store_error()) });
        }
        let readers = self.readers.clone();
        Box::pin(async move {
            with_reader(readers, move |conn| {
                let row = get_job(conn, job_id).map_err(sqlite_error)?;
                match row {
                    Some(json) => {
                        let value: Value = serde_json::from_str(&json)
                            .map_err(|e| Error::Codec(format!("job row must be json: {e}")))?;
                        job_from_json(&value)
                    }
                    None => Ok(None),
                }
            })
            .await
        })
    }

    fn sweep_expired<'a>(&'a self) -> BoxedFuture<'a, Result<i64>> {
        if self.closed.load(Ordering::Acquire) {
            return Box::pin(async { Err(closed_store_error()) });
        }
        let name = self.name.clone();
        Box::pin(async move {
            with_writer(self.writer.clone(), move |conn| {
                sweep_expired(conn, &name, EXPIRED_ERROR).map_err(sqlite_error)
            })
            .await
        })
    }
}

/// The claim handle (see the module docs). The one-shot ops consume
/// `self: Box<Self>` (the whole handle moves into the `'static`
/// future); `heartbeat` is the deliberate repeatable `&self`
/// counter-case (ADR-019 §3 — do not "fix" the asymmetry).
pub struct SqliteJobHandle {
    job: Job,
    writer: Arc<Writer>,
    closed: Arc<AtomicBool>,
}

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

impl SqliteJobHandle {
    pub(crate) fn new(job: Job, writer: Arc<Writer>, closed: Arc<AtomicBool>) -> Self {
        Self {
            job,
            writer,
            closed,
        }
    }
}

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

    fn ack(self: Box<Self>) -> BoxedFuture<'static, Result<bool>> {
        let SqliteJobHandle { job, writer, .. } = *self;
        let id = job.id;
        let worker_id = job.worker_id.unwrap_or_default();
        Box::pin(async move {
            with_writer(writer, move |conn| {
                Ok(ack(conn, id, &worker_id).map_err(sqlite_error)? > 0)
            })
            .await
        })
    }

    fn retry(
        self: Box<Self>,
        err: Option<String>,
        delay: Option<i64>,
    ) -> BoxedFuture<'static, Result<bool>> {
        let SqliteJobHandle { job, writer, .. } = *self;
        let id = job.id;
        let worker_id = job.worker_id.unwrap_or_default();
        let exhaust_error = err.unwrap_or_else(|| EXHAUSTED_ERROR.to_string());
        let delay_s = match delay {
            Some(d) => d,
            None => backoff_delay_s(job.backoff_base_s, job.attempts, id),
        };
        Box::pin(async move {
            with_writer(writer, move |conn| {
                Ok(retry(conn, id, &worker_id, delay_s, &exhaust_error).map_err(sqlite_error)? > 0)
            })
            .await
        })
    }

    fn fail(self: Box<Self>, err: Option<String>) -> BoxedFuture<'static, Result<bool>> {
        let SqliteJobHandle { job, writer, .. } = *self;
        let id = job.id;
        let worker_id = job.worker_id.unwrap_or_default();
        let error = err.unwrap_or_else(|| FAILED_ERROR.to_string());
        Box::pin(async move {
            with_writer(writer, move |conn| {
                Ok(fail(conn, id, &worker_id, &error).map_err(sqlite_error)? > 0)
            })
            .await
        })
    }

    fn heartbeat<'a>(&'a self, extend: i64) -> BoxedFuture<'a, Result<bool>> {
        if self.closed.load(Ordering::Acquire) {
            return Box::pin(async { Err(closed_store_error()) });
        }
        let writer = self.writer.clone();
        let id = self.job.id;
        let worker_id = self.job.worker_id.clone().unwrap_or_default();
        Box::pin(async move {
            with_writer(writer, move |conn| {
                Ok(heartbeat(conn, id, &worker_id, extend).map_err(sqlite_error)? > 0)
            })
            .await
        })
    }
}