hammerwork 2.5.0

//! In-memory test implementation of the job queue for testing purposes.
//!
//! This module provides a `TestQueue` that implements the `DatabaseQueue` trait
//! entirely in memory, making it ideal for unit tests and development without
//! requiring a database connection.
//!
//! # Features
//!
//! - All operations are performed in-memory
//! - Thread-safe with async operations
//! - Supports time manipulation for testing delayed jobs
//! - Full compatibility with the `DatabaseQueue` trait
//! - Deterministic behavior for testing
//!
//! # Examples
//!
//! ```rust
//! use hammerwork::queue::test::TestQueue;
//! use hammerwork::{Job, JobStatus, queue::DatabaseQueue};
//! use serde_json::json;
//!
//! # #[tokio::main]
//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
//! let queue = TestQueue::new();
//!
//! // Enqueue a job
//! let job = Job::new("test_queue".to_string(), json!({"test": true}));
//! let job_id = queue.enqueue(job).await?;
//!
//! // Process the job
//! if let Some(mut job) = queue.dequeue("test_queue").await? {
//!     // Process the job...
//!     queue.complete_job(job.id).await?;
//! }
//! # Ok(())
//! # }
//! ```

use crate::{
    HammerworkError, Result,
    batch::{BatchId, BatchResult, BatchStatus, JobBatch},
    job::{Job, JobId, JobStatus},
    priority::{JobPriority, PriorityWeights},
    queue::{DatabaseQueue, PayloadEncryption, QueuePauseInfo, lifecycle},
    rate_limit::ThrottleConfig,
    stats::{DeadJobSummary, QueueStats},
    workflow::{
        DependencyStatus, FailurePolicy, JobGroup, WorkflowId, WorkflowProgress, WorkflowStatus,
    },
};
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use std::{
    collections::HashMap,
    sync::{Arc, Mutex},
};
use tokio::sync::RwLock;

// We need to use a real sqlx Database type for the trait bound
// This is just a marker - TestQueue doesn't actually use SQL
#[cfg(feature = "postgres")]
type TestDatabaseType = sqlx::Postgres;

#[cfg(all(feature = "mysql", not(feature = "postgres")))]
type TestDatabaseType = sqlx::MySql;

#[cfg(all(not(feature = "postgres"), not(feature = "mysql")))]
type TestDatabaseType = sqlx::Any; // Fallback

/// Mock clock for controlling time in tests.
///
/// The `MockClock` allows you to control time flow in your tests, making it possible
/// to test time-dependent functionality like delayed jobs, cron schedules, and timeouts
/// in a deterministic and fast manner.
///
/// # Examples
///
/// ## Basic time manipulation
///
/// ```rust
/// use hammerwork::queue::test::MockClock;
/// use chrono::Duration;
///
/// let clock = MockClock::new();
/// let initial_time = clock.now();
///
/// // Advance time by 1 hour
/// clock.advance(Duration::hours(1));
///
/// let after_advance = clock.now();
/// assert_eq!((after_advance - initial_time).num_hours(), 1);
/// ```
///
/// ## Testing delayed jobs
///
/// ```rust
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// use hammerwork::queue::test::{MockClock, TestQueue};
/// use hammerwork::{Job, queue::DatabaseQueue};
/// use serde_json::json;
/// use chrono::Duration;
///
/// let clock = MockClock::new();
/// let queue = TestQueue::with_clock(clock.clone());
///
/// // Create a delayed job (runs in 2 hours)
/// let delayed_job = Job::with_delay(
///     "delayed_queue".to_string(),
///     json!({"message": "Hello future!"}),
///     Duration::hours(2)
/// );
///
/// let job_id = queue.enqueue(delayed_job).await?;
///
/// // Job should not be available immediately
/// assert!(queue.dequeue("delayed_queue").await?.is_none());
///
/// // Advance time by 2 hours
/// clock.advance(Duration::hours(2));
///
/// // Now the job should be available
/// let job = queue.dequeue("delayed_queue").await?.unwrap();
/// assert_eq!(job.id, job_id);
/// # Ok(())
/// # }
/// ```
#[derive(Clone, Debug)]
pub struct MockClock {
    current_time: Arc<Mutex<DateTime<Utc>>>,
}

impl MockClock {
    /// Create a new mock clock starting at the current time.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use hammerwork::queue::test::MockClock;
    ///
    /// let clock = MockClock::new();
    /// let now = clock.now();
    /// println!("Mock clock started at: {}", now);
    /// ```
    pub fn new() -> Self {
        Self {
            current_time: Arc::new(Mutex::new(Utc::now())),
        }
    }

    /// Get the current mock time.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use hammerwork::queue::test::MockClock;
    /// use chrono::Duration;
    ///
    /// let clock = MockClock::new();
    /// let time1 = clock.now();
    ///
    /// clock.advance(Duration::minutes(30));
    /// let time2 = clock.now();
    ///
    /// assert_eq!((time2 - time1).num_minutes(), 30);
    /// ```
    pub fn now(&self) -> DateTime<Utc> {
        *self.current_time.lock().unwrap()
    }

    /// Advance the mock time by the given duration.
    ///
    /// This is useful for testing time-dependent functionality without
    /// waiting for real time to pass.
    ///
    /// # Arguments
    ///
    /// * `duration` - The amount of time to advance the clock
    ///
    /// # Examples
    ///
    /// ```rust
    /// use hammerwork::queue::test::MockClock;
    /// use chrono::Duration;
    ///
    /// let clock = MockClock::new();
    /// let start_time = clock.now();
    ///
    /// // Advance by 1 day, 2 hours, and 30 minutes
    /// clock.advance(Duration::days(1));
    /// clock.advance(Duration::hours(2));
    /// clock.advance(Duration::minutes(30));
    ///
    /// let end_time = clock.now();
    /// let total_duration = end_time - start_time;
    ///
    /// assert_eq!(total_duration.num_hours(), 26); // 24 + 2
    /// assert_eq!(total_duration.num_minutes(), 1590); // 26 * 60 + 30
    /// ```
    pub fn advance(&self, duration: chrono::Duration) {
        let mut time = self.current_time.lock().unwrap();
        *time += duration;
    }

    /// Set the mock time to a specific instant.
    ///
    /// This allows you to set the clock to any specific time, which can be
    /// useful for testing scenarios that depend on absolute times.
    ///
    /// # Arguments
    ///
    /// * `time` - The specific time to set the clock to
    ///
    /// # Examples
    ///
    /// ```rust
    /// use hammerwork::queue::test::MockClock;
    /// use chrono::{Utc, TimeZone};
    ///
    /// let clock = MockClock::new();
    ///
    /// // Set to a specific date and time
    /// let specific_time = Utc.with_ymd_and_hms(2024, 1, 1, 12, 0, 0).unwrap();
    /// clock.set_time(specific_time);
    ///
    /// assert_eq!(clock.now(), specific_time);
    /// ```
    pub fn set_time(&self, time: DateTime<Utc>) {
        *self.current_time.lock().unwrap() = time;
    }
}

impl Default for MockClock {
    fn default() -> Self {
        Self::new()
    }
}

/// When a lease of `lease` taken at `now` expires: never for a zero lease, as on the
/// database backends.
fn lease_expiry(now: DateTime<Utc>, lease: std::time::Duration) -> DateTime<Utc> {
    if lease.is_zero() {
        DateTime::<Utc>::MAX_UTC
    } else {
        super::saturating_add_to(now, lease)
    }
}

/// In-memory storage for the test queue
#[derive(Debug)]
struct TestStorage {
    /// All jobs stored by ID
    jobs: HashMap<JobId, Job>,
    /// Jobs organized by queue name and status
    queues: HashMap<String, HashMap<JobStatus, Vec<JobId>>>,
    /// Batch information
    batches: HashMap<BatchId, JobBatch>,
    /// Batch job mappings
    batch_jobs: HashMap<BatchId, Vec<JobId>>,
    /// Workflow information
    workflows: HashMap<WorkflowId, JobGroup>,
    /// Job dependencies (job_id -> depends on these job_ids)
    dependencies: HashMap<JobId, Vec<JobId>>,
    /// Reverse dependencies (job_id -> these jobs depend on it)
    dependents: HashMap<JobId, Vec<JobId>>,
    /// Throttle configurations
    throttle_configs: HashMap<String, ThrottleConfig>,
    /// Job results storage
    job_results: HashMap<JobId, (serde_json::Value, Option<DateTime<Utc>>)>,
    /// Paused queues information
    paused_queues: HashMap<String, QueuePauseInfo>,
    /// Job leases: job_id -> (last heartbeat, lease expiry)
    leases: HashMap<JobId, (DateTime<Utc>, DateTime<Utc>)>,
    /// Mock clock for time control
    clock: MockClock,
}

impl TestStorage {
    fn new(clock: MockClock) -> Self {
        Self {
            jobs: HashMap::new(),
            queues: HashMap::new(),
            batches: HashMap::new(),
            batch_jobs: HashMap::new(),
            workflows: HashMap::new(),
            dependencies: HashMap::new(),
            dependents: HashMap::new(),
            throttle_configs: HashMap::new(),
            job_results: HashMap::new(),
            paused_queues: HashMap::new(),
            leases: HashMap::new(),
            clock,
        }
    }

    /// Add a job to the appropriate queue and status list
    fn add_job_to_queue(&mut self, job: &Job) {
        let queue_jobs = self.queues.entry(job.queue_name.clone()).or_default();
        let status_jobs = queue_jobs.entry(job.status).or_default();
        if !status_jobs.contains(&job.id) {
            status_jobs.push(job.id);
        }
    }

    /// Check that a manual `transition` may start from the job's current status (see
    /// [`JobTransition::allowed_from`](crate::queue::JobTransition::allowed_from)).
    fn check_transition(
        &self,
        job_id: JobId,
        transition: crate::queue::JobTransition,
    ) -> Result<()> {
        let status = self
            .jobs
            .get(&job_id)
            .map(|job| job.status)
            .ok_or_else(|| HammerworkError::JobNotFound {
                id: job_id.to_string(),
            })?;
        if transition.is_allowed_from(status) {
            Ok(())
        } else {
            Err(transition.rejected(job_id, status))
        }
    }

    /// Put a recurring job back to `Pending` for its run at `next_run_at`, in place (the
    /// same id), like the database backends. With `keep_failure` the last run's error and
    /// failure time stay visible until the next run; otherwise they are cleared.
    fn reschedule_in_place(
        &mut self,
        job_id: JobId,
        next_run_at: DateTime<Utc>,
        keep_failure: bool,
    ) -> Result<()> {
        self.update_job_status(job_id, JobStatus::Pending)?;
        self.leases.remove(&job_id);
        if let Some(job) = self.jobs.get_mut(&job_id) {
            job.scheduled_at = next_run_at;
            job.next_run_at = Some(next_run_at);
            job.attempts = 0;
            job.started_at = None;
            job.completed_at = None;
            if !keep_failure {
                job.failed_at = None;
                job.timed_out_at = None;
                job.error_message = None;
            }
        }
        Ok(())
    }

    /// Move a job being enqueued onto the mock clock: `created_at` becomes `now` and
    /// `scheduled_at` keeps its delay from the original `created_at` (an immediate
    /// job, scheduled at its creation time, is scheduled at `now`).
    fn stamp_enqueued(job: &mut Job, now: DateTime<Utc>) {
        let delay = job.scheduled_at - job.created_at;
        job.scheduled_at = now.checked_add_signed(delay).unwrap_or(job.scheduled_at);
        job.created_at = now;
    }

    /// Update a job's status
    fn update_job_status(&mut self, job_id: JobId, new_status: JobStatus) -> Result<()> {
        // Get old status first
        let old_status = self
            .jobs
            .get(&job_id)
            .map(|job| job.status)
            .ok_or_else(|| HammerworkError::JobNotFound {
                id: job_id.to_string(),
            })?;

        // Get the queue name
        let queue_name = self
            .jobs
            .get(&job_id)
            .map(|job| job.queue_name.clone())
            .ok_or_else(|| HammerworkError::JobNotFound {
                id: job_id.to_string(),
            })?;

        // Remove from old status
        if let Some(queue_jobs) = self.queues.get_mut(&queue_name)
            && let Some(status_jobs) = queue_jobs.get_mut(&old_status)
        {
            status_jobs.retain(|id| *id != job_id);
        }

        // Update the job status
        if let Some(job) = self.jobs.get_mut(&job_id) {
            job.status = new_status;
        }

        // Add to new status
        let queue_jobs = self.queues.entry(queue_name).or_default();
        let status_jobs = queue_jobs.entry(new_status).or_default();
        if !status_jobs.contains(&job_id) {
            status_jobs.push(job_id);
        }

        Ok(())
    }

    /// Mark an unfinished job `Failed` with `error`, as a side effect of another job's
    /// failure. A job still waiting on its dependencies records that they failed.
    fn fail_as_side_effect(&mut self, job_id: JobId, error: String, now: DateTime<Utc>) {
        if self.update_job_status(job_id, JobStatus::Failed).is_err() {
            return;
        }
        self.leases.remove(&job_id);
        if let Some(job) = self.jobs.get_mut(&job_id) {
            job.failed_at = Some(now);
            job.error_message = Some(error);
            if job.dependency_status == DependencyStatus::Waiting {
                job.dependency_status = DependencyStatus::Failed;
            }
        }
    }

    /// The workflow `job_id` belongs to: the one its `workflow_id` names, or else the
    /// one listing it among its jobs.
    fn workflow_of(&self, job_id: JobId) -> Option<&JobGroup> {
        self.jobs
            .get(&job_id)
            .and_then(|job| job.workflow_id)
            .and_then(|id| self.workflows.get(&id))
            .or_else(|| {
                self.workflows
                    .values()
                    .find(|w| w.jobs.iter().any(|j| j.id == job_id))
            })
    }

    /// The jobs that depend directly on `job_id`.
    fn direct_dependents(&self, job_id: JobId) -> Vec<JobId> {
        let mut dependents = self.dependents.get(&job_id).cloned().unwrap_or_default();
        for job in self.jobs.values() {
            if job.depends_on.contains(&job_id) && !dependents.contains(&job.id) {
                dependents.push(job.id);
            }
        }
        dependents
    }

    /// Fail every pending job that (transitively) depends on `failed_job_id`.
    fn fail_dependents(&mut self, failed_job_id: JobId, now: DateTime<Utc>) -> Vec<JobId> {
        let mut failed = Vec::new();
        let mut to_check = vec![failed_job_id];
        while let Some(current) = to_check.pop() {
            for id in self.direct_dependents(current) {
                let pending = self.jobs.get(&id).is_some_and(|job| {
                    job.status == JobStatus::Pending
                        && job.dependency_status != DependencyStatus::Failed
                });
                if pending && !failed.contains(&id) {
                    self.fail_as_side_effect(
                        id,
                        lifecycle::dependency_failed_message(current),
                        now,
                    );
                    failed.push(id);
                    to_check.push(id);
                }
            }
        }
        failed
    }

    /// Apply the failure policy of `failed_job_id`'s workflow, as the database
    /// backends do (see [`FailurePolicy`]):
    ///
    /// - `FailFast` fails the workflow's other `Pending` and `Retrying` jobs, then
    ///   every job that (transitively) depends on the failed one;
    /// - `ContinueOnFailure`, and jobs outside any workflow, fail only the jobs that
    ///   (transitively) depend on the failed one;
    /// - `Manual` changes nothing: dependents keep waiting.
    ///
    /// Returns the jobs that were failed.
    fn apply_workflow_failure(&mut self, failed_job_id: JobId, now: DateTime<Utc>) -> Vec<JobId> {
        let workflow = self.workflow_of(failed_job_id).map(|w| {
            (
                w.failure_policy.clone(),
                w.jobs.iter().map(|j| j.id).collect(),
            )
        });
        match workflow {
            Some((FailurePolicy::Manual, _)) => Vec::new(),
            Some((FailurePolicy::FailFast, job_ids)) => {
                let job_ids: Vec<JobId> = job_ids;
                let mut failed = Vec::new();
                for id in job_ids {
                    let unstarted = self.jobs.get(&id).is_some_and(|job| {
                        matches!(job.status, JobStatus::Pending | JobStatus::Retrying)
                    });
                    if id != failed_job_id && unstarted {
                        self.fail_as_side_effect(
                            id,
                            lifecycle::workflow_failed_message(failed_job_id),
                            now,
                        );
                        failed.push(id);
                    }
                }
                for id in self.fail_dependents(failed_job_id, now) {
                    if !failed.contains(&id) {
                        failed.push(id);
                    }
                }
                failed
            }
            Some((FailurePolicy::ContinueOnFailure, _)) | None => {
                self.fail_dependents(failed_job_id, now)
            }
        }
    }

    /// Fail the `Pending` and `Retrying` jobs of a `FailFast` batch after one of its
    /// jobs failed. Returns the jobs that were failed.
    fn apply_batch_failure(&mut self, failed_job_id: JobId, now: DateTime<Utc>) -> Vec<JobId> {
        let Some(batch_id) = self.jobs.get(&failed_job_id).and_then(|job| job.batch_id) else {
            return Vec::new();
        };
        let fail_fast = self
            .batches
            .get(&batch_id)
            .is_some_and(|batch| batch.failure_mode == crate::batch::PartialFailureMode::FailFast);
        if !fail_fast {
            return Vec::new();
        }
        let job_ids = self.batch_jobs.get(&batch_id).cloned().unwrap_or_default();
        let mut failed = Vec::new();
        for id in job_ids {
            let unstarted = self
                .jobs
                .get(&id)
                .is_some_and(|job| matches!(job.status, JobStatus::Pending | JobStatus::Retrying));
            if unstarted {
                self.fail_as_side_effect(id, lifecycle::batch_failed_message(failed_job_id), now);
                failed.push(id);
            }
        }
        failed
    }

    /// Get the next job to dequeue based on priority and scheduled time
    /// The jobs of `queue_name` that can be dequeued now: pending, due and not a
    /// disabled recurring job's held occurrence. None while the queue is paused.
    fn eligible_jobs(&self, queue_name: &str) -> Vec<&Job> {
        // Like the database backends, a paused queue hands out no jobs.
        if self.paused_queues.contains_key(queue_name) {
            return Vec::new();
        }
        let Some(pending_jobs) = self
            .queues
            .get(queue_name)
            .and_then(|queue_jobs| queue_jobs.get(&JobStatus::Pending))
        else {
            return Vec::new();
        };

        let now = self.clock.now();
        pending_jobs
            .iter()
            .filter_map(|id| self.jobs.get(id))
            .filter(|job| job.scheduled_at <= now)
            // Like the database backends, a job waiting on its dependencies is held.
            .filter(|job| {
                matches!(
                    job.dependency_status,
                    DependencyStatus::None | DependencyStatus::Satisfied
                )
            })
            // A disabled recurring job's pending occurrence is held.
            .filter(|job| job.recurring || job.cron_schedule.is_none())
            .collect()
    }

    /// The dependencies of `job_id`: its `depends_on` and those recorded for it by
    /// `enqueue_workflow`.
    fn dependencies_of(&self, job_id: JobId) -> Vec<JobId> {
        let mut deps = self
            .jobs
            .get(&job_id)
            .map(|job| job.depends_on.clone())
            .unwrap_or_default();
        for dep in self.dependencies.get(&job_id).into_iter().flatten() {
            if !deps.contains(dep) {
                deps.push(*dep);
            }
        }
        deps
    }

    /// The statuses of the dependencies `ids` that still exist, for
    /// [`lifecycle::reevaluate_dependencies`]. An archived job keeps its times, which
    /// stand in for the status the database backends keep in the archive.
    fn dependency_states(&self, ids: &[JobId]) -> HashMap<JobId, JobStatus> {
        ids.iter()
            .filter_map(|id| Some((*id, self.status_with_archived(*id)?)))
            .collect()
    }

    /// The status of `job_id`; an archived job reports the status it was archived with,
    /// as the database backends read it from the archive (its times stand in for it).
    fn status_with_archived(&self, job_id: JobId) -> Option<JobStatus> {
        let job = self.jobs.get(&job_id)?;
        Some(match job.status {
            JobStatus::Archived if job.completed_at.is_some() => JobStatus::Completed,
            JobStatus::Archived if job.timed_out_at.is_some() => JobStatus::TimedOut,
            JobStatus::Archived if job.failed_at.is_some() => JobStatus::Failed,
            status => status,
        })
    }

    /// Add the spawned job `job_id` to its workflow `workflow_id`, raising the
    /// workflow's job count, as the database backends do for spawned children.
    fn join_workflow(&mut self, workflow_id: WorkflowId, job_id: JobId) {
        let Some(job) = self.jobs.get(&job_id).cloned() else {
            return;
        };
        if let Some(workflow) = self.workflows.get_mut(&workflow_id)
            && !workflow.jobs.iter().any(|j| j.id == job_id)
        {
            workflow.jobs.push(job);
            workflow.total_jobs += 1;
        }
    }

    /// Whether the dependency `job_id` completed (archived jobs keep their times).
    fn dependency_completed(&self, job_id: JobId) -> bool {
        self.jobs.get(&job_id).is_some_and(|job| match job.status {
            JobStatus::Completed => true,
            JobStatus::Archived => job.completed_at.is_some(),
            _ => false,
        })
    }

    /// Release the pending dependents of `completed_job_id` whose dependencies have all
    /// completed, like the database backends' `resolve_dependents`: their dependency
    /// status becomes `Satisfied` (and a workflow job held by `enqueue_workflow` becomes
    /// due). Returns the released jobs.
    fn resolve_dependents(&mut self, completed_job_id: JobId, now: DateTime<Utc>) -> Vec<JobId> {
        let mut candidates: Vec<JobId> = self
            .jobs
            .values()
            .filter(|job| {
                job.status == JobStatus::Pending
                    && (job.dependency_status == DependencyStatus::Waiting
                        || (job.dependency_status == DependencyStatus::None
                            && job.scheduled_at == DateTime::<Utc>::MAX_UTC))
            })
            .map(|job| job.id)
            .filter(|id| self.dependencies_of(*id).contains(&completed_job_id))
            .collect();
        candidates.sort_unstable();

        let mut resolved = Vec::new();
        for id in candidates {
            let ready = self
                .dependencies_of(id)
                .iter()
                .all(|dep| self.dependency_completed(*dep));
            if !ready {
                continue;
            }
            if let Some(job) = self.jobs.get_mut(&id) {
                if job.dependency_status == DependencyStatus::Waiting {
                    job.dependency_status = DependencyStatus::Satisfied;
                }
                if job.scheduled_at == DateTime::<Utc>::MAX_UTC {
                    job.scheduled_at = now;
                }
                resolved.push(id);
            }
        }
        resolved
    }

    /// Settle the dependency status of `jobs` about to be enqueued against the stored
    /// state of their dependencies, as the database backends do on insert: a job whose
    /// dependencies already completed is runnable, one with a failed dependency is
    /// enqueued as failed. Returns the jobs that were failed.
    fn settle_new_dependents(&self, jobs: &mut [Job], now: DateTime<Utc>) -> Vec<JobId> {
        let mut parents = HashMap::new();
        for parent in lifecycle::external_dependencies(jobs) {
            let Some(job) = self.jobs.get(&parent) else {
                continue;
            };
            let manual = self
                .workflow_of(parent)
                .is_some_and(|w| w.failure_policy == FailurePolicy::Manual);
            let state = match job.status {
                JobStatus::Completed => lifecycle::ParentState::Completed,
                JobStatus::Failed | JobStatus::Dead | JobStatus::TimedOut if !manual => {
                    lifecycle::ParentState::Failed
                }
                JobStatus::Archived if job.completed_at.is_some() => {
                    lifecycle::ParentState::Completed
                }
                JobStatus::Archived if job.failed_at.or(job.timed_out_at).is_some() => {
                    lifecycle::ParentState::Failed
                }
                _ => lifecycle::ParentState::Unfinished,
            };
            parents.insert(parent, state);
        }
        lifecycle::settle_new_dependents(jobs, &parents, now)
    }

    /// Up to `max_jobs` jobs to claim at once, in claim order, chosen like the database
    /// backends' `dequeue_batch_leased`: in strict priority order, or with weighted
    /// `weights` filled from a level picked by weight, then from the other levels.
    fn get_next_jobs(
        &self,
        queue_name: &str,
        weights: Option<&PriorityWeights>,
        max_jobs: usize,
    ) -> Vec<JobId> {
        let mut eligible = self.eligible_jobs(queue_name);
        eligible.sort_by_key(|job| {
            (
                std::cmp::Reverse(job.priority.as_i32()),
                job.scheduled_at,
                job.created_at,
            )
        });
        let weights = match weights {
            Some(weights) if !weights.is_strict() => weights,
            _ => {
                return eligible
                    .into_iter()
                    .take(max_jobs)
                    .map(|job| job.id)
                    .collect();
            }
        };

        let mut candidates: Vec<JobPriority> = Vec::new();
        for job in &eligible {
            if !candidates.contains(&job.priority) {
                candidates.push(job.priority);
            }
        }
        let mut claimed = Vec::new();
        let mut attempt: u64 = 0;
        while claimed.len() < max_jobs {
            use std::hash::{Hash, Hasher};
            let mut hasher = std::collections::hash_map::DefaultHasher::new();
            queue_name.hash(&mut hasher);
            self.clock.now().timestamp_millis().hash(&mut hasher);
            attempt.hash(&mut hasher);
            let Some(priority) =
                super::pick_weighted_priority(&candidates, weights, hasher.finish())
            else {
                break;
            };
            let room = max_jobs - claimed.len();
            claimed.extend(
                eligible
                    .iter()
                    .filter(|job| job.priority == priority)
                    .take(room)
                    .map(|job| job.id),
            );
            candidates.retain(|candidate| *candidate != priority);
            attempt += 1;
        }
        claimed
    }

    fn get_next_job(&self, queue_name: &str, weights: Option<&PriorityWeights>) -> Option<JobId> {
        let eligible_jobs = self.eligible_jobs(queue_name);

        if eligible_jobs.is_empty() {
            return None;
        }

        // If no weights provided, use strict priority ordering
        if weights.is_none() {
            return eligible_jobs
                .into_iter()
                .max_by_key(|job| (job.priority.as_i32(), std::cmp::Reverse(job.scheduled_at)))
                .map(|job| job.id);
        }

        // Use weighted selection
        let weights = weights.unwrap();

        // If strict priority, just get highest priority job
        if weights.is_strict() {
            return eligible_jobs
                .into_iter()
                .max_by_key(|job| (job.priority.as_i32(), std::cmp::Reverse(job.scheduled_at)))
                .map(|job| job.id);
        }

        // Group jobs by priority
        let mut candidates_by_priority: HashMap<JobPriority, Vec<&Job>> = HashMap::new();
        for job in eligible_jobs {
            candidates_by_priority
                .entry(job.priority)
                .or_default()
                .push(job);
        }

        // Pick a level by weight as the database backends do (all-zero weights take
        // the highest level), with a hash-based seed for deterministic behavior.
        use std::collections::hash_map::DefaultHasher;
        use std::hash::{Hash, Hasher};

        let mut candidates: Vec<JobPriority> = candidates_by_priority.keys().copied().collect();
        candidates.sort_unstable_by_key(|priority| std::cmp::Reverse(priority.as_i32()));
        let mut hasher = DefaultHasher::new();
        queue_name.hash(&mut hasher);
        self.clock.now().timestamp_millis().hash(&mut hasher);
        let selected_priority =
            super::pick_weighted_priority(&candidates, weights, hasher.finish())?;

        // Get the oldest job from the selected priority
        candidates_by_priority
            .get(&selected_priority)
            .and_then(|jobs| {
                jobs.iter()
                    .min_by_key(|job| job.scheduled_at)
                    .map(|job| job.id)
            })
    }
}

/// In-memory test implementation of the job queue.
///
/// `TestQueue` provides a complete implementation of the `DatabaseQueue` trait
/// that runs entirely in memory, making it perfect for unit testing your job
/// processing logic without requiring a database connection.
///
/// # Features
///
/// - **Full DatabaseQueue compatibility**: Drop-in replacement for testing
/// - **Time control**: MockClock integration for testing time-dependent features
/// - **Deterministic behavior**: Predictable ordering and timing for reliable tests
/// - **Complete feature support**: Batches, workflows, cron jobs, priorities, and more
/// - **Thread-safe**: Safe to use in concurrent test scenarios
///
/// # Examples
///
/// ## Basic job processing test
///
/// ```rust
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// use hammerwork::queue::test::TestQueue;
/// use hammerwork::{Job, JobStatus, queue::DatabaseQueue};
/// use serde_json::json;
///
/// let queue = TestQueue::new();
///
/// // Enqueue a job
/// let job = Job::new("test_queue".to_string(), json!({"task": "process_data"}));
/// let job_id = queue.enqueue(job).await?;
///
/// // Dequeue and process the job
/// let job = queue.dequeue("test_queue").await?.unwrap();
/// assert_eq!(job.status, JobStatus::Running);
///
/// // Mark job as completed
/// queue.complete_job(job_id).await?;
///
/// // Verify completion
/// let completed_job = queue.get_job(job_id).await?.unwrap();
/// assert_eq!(completed_job.status, JobStatus::Completed);
/// # Ok(())
/// # }
/// ```
///
/// ## Testing job priorities
///
/// ```rust
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// use hammerwork::queue::test::TestQueue;
/// use hammerwork::{Job, JobPriority, queue::DatabaseQueue};
/// use serde_json::json;
///
/// let queue = TestQueue::new();
///
/// // Enqueue jobs with different priorities
/// let low_job = Job::new("priority_queue".to_string(), json!({"priority": "low"}))
///     .as_low_priority();
/// let high_job = Job::new("priority_queue".to_string(), json!({"priority": "high"}))
///     .as_high_priority();
///
/// let low_id = queue.enqueue(low_job).await?;
/// let high_id = queue.enqueue(high_job).await?;
///
/// // High priority job should be dequeued first
/// let first_job = queue.dequeue("priority_queue").await?.unwrap();
/// assert_eq!(first_job.id, high_id);
/// assert_eq!(first_job.priority, JobPriority::High);
///
/// let second_job = queue.dequeue("priority_queue").await?.unwrap();
/// assert_eq!(second_job.id, low_id);
/// assert_eq!(second_job.priority, JobPriority::Low);
/// # Ok(())
/// # }
/// ```
///
/// ## Testing job failures and retries
///
/// ```rust
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// use hammerwork::queue::test::TestQueue;
/// use hammerwork::{Job, JobStatus, queue::DatabaseQueue};
/// use serde_json::json;
///
/// let queue = TestQueue::new();
///
/// // Create a job with custom retry settings
/// let job = Job::new("retry_queue".to_string(), json!({"data": "test"}))
///     .with_max_attempts(3);
/// let job_id = queue.enqueue(job).await?;
///
/// // Each dequeue counts as an attempt. Like a worker, retry the job while it has
/// // attempts left and mark it dead once they are used up.
/// for attempt in 1..=3 {
///     let job = queue.dequeue("retry_queue").await?.unwrap();
///     assert_eq!(job.attempts, attempt);
///     if job.attempts < job.max_attempts {
///         queue.retry_job(job_id, queue.clock().now()).await?;
///     } else {
///         queue.mark_job_dead(job_id, "Still failing").await?;
///     }
/// }
///
/// let dead_job = queue.get_job(job_id).await?.unwrap();
/// assert_eq!(dead_job.status, JobStatus::Dead);
/// assert_eq!(dead_job.attempts, 3);
///
/// // `fail_job` records a terminal failure that is not retried automatically.
/// let job_id = queue
///     .enqueue(Job::new("retry_queue".to_string(), json!({"data": "other"})))
///     .await?;
/// queue.dequeue("retry_queue").await?.unwrap();
/// queue.fail_job(job_id, "Permanent error").await?;
/// assert_eq!(queue.get_job(job_id).await?.unwrap().status, JobStatus::Failed);
/// # Ok(())
/// # }
/// ```
///
/// ## Job results
///
/// `TestQueue` keeps every stored result in memory, whatever the job's
/// [`ResultStorage`](crate::job::ResultStorage): `store_job_result` and
/// `get_job_result` behave the same for `Database` and `Memory` jobs, and results expire
/// at their `expires_at` by the [`MockClock`]. Deleting a job deletes its result.
#[derive(Clone)]
pub struct TestQueue {
    storage: Arc<RwLock<TestStorage>>,
    clock: MockClock,
    encryption: PayloadEncryption,
}

impl TestQueue {
    /// Create a new test queue with a fresh MockClock.
    ///
    /// This is the most common way to create a TestQueue for testing.
    /// Each TestQueue gets its own isolated storage and time control.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use hammerwork::queue::test::TestQueue;
    ///
    /// let queue = TestQueue::new();
    /// // Queue is ready to use for testing
    /// ```
    pub fn new() -> Self {
        let clock = MockClock::new();
        Self {
            storage: Arc::new(RwLock::new(TestStorage::new(clock.clone()))),
            clock,
            encryption: PayloadEncryption::default(),
        }
    }

    /// Create a new test queue with a custom MockClock.
    ///
    /// This allows you to share time control across multiple test components
    /// or start with a specific time state.
    ///
    /// # Arguments
    ///
    /// * `clock` - The MockClock instance to use for time control
    ///
    /// # Examples
    ///
    /// ```rust
    /// use hammerwork::queue::test::{MockClock, TestQueue};
    /// use chrono::{Utc, TimeZone};
    ///
    /// // Create a clock set to a specific time
    /// let clock = MockClock::new();
    /// let specific_time = Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap();
    /// clock.set_time(specific_time);
    ///
    /// // Create queue with this clock
    /// let queue = TestQueue::with_clock(clock);
    ///
    /// // Queue will use the specified time
    /// assert_eq!(queue.clock().now(), specific_time);
    /// ```
    pub fn with_clock(clock: MockClock) -> Self {
        Self {
            storage: Arc::new(RwLock::new(TestStorage::new(clock.clone()))),
            clock,
            encryption: PayloadEncryption::default(),
        }
    }

    /// Get access to the mock clock for time manipulation.
    ///
    /// This allows you to control time flow in your tests, which is essential
    /// for testing time-dependent features like delayed jobs, cron schedules,
    /// and timeouts.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use hammerwork::queue::test::TestQueue;
    /// use chrono::Duration;
    ///
    /// let queue = TestQueue::new();
    /// let initial_time = queue.clock().now();
    ///
    /// // Advance time by 1 hour
    /// queue.clock().advance(Duration::hours(1));
    ///
    /// let later_time = queue.clock().now();
    /// assert_eq!((later_time - initial_time).num_hours(), 1);
    /// ```
    pub fn clock(&self) -> &MockClock {
        &self.clock
    }

    /// Encrypts job payloads with `engine`, like
    /// [`JobQueue::with_encryption`](crate::JobQueue::with_encryption).
    ///
    /// The in-memory queue then applies the same encryption semantics as the database
    /// backends, so unit tests exercise them: jobs with an encryption config are stored
    /// sealed (redacted `payload`, `is_encrypted`, ciphertext in `encrypted_payload`) and
    /// returned that way by `dequeue` and `get_job`; [`TestQueue::decrypt_job`] opens them
    /// like a worker does; enqueueing a job with an encryption config fails without an
    /// engine; and [`DatabaseQueue::purge_expired_encrypted_jobs`] enforces retention
    /// against the queue's [`MockClock`].
    ///
    /// # Examples
    ///
    /// ```rust
    /// # #[cfg(feature = "encryption")]
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// use hammerwork::encryption::{EncryptionAlgorithm, EncryptionConfig, EncryptionEngine, KeySource};
    /// use hammerwork::queue::test::TestQueue;
    /// use hammerwork::{Job, queue::DatabaseQueue};
    /// use serde_json::json;
    ///
    /// let config = EncryptionConfig::new(EncryptionAlgorithm::AES256GCM).with_key_source(
    ///     KeySource::Static("QUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUE=".to_string()),
    /// );
    /// let queue = TestQueue::new().with_encryption(EncryptionEngine::new(config.clone()).await?);
    ///
    /// let job = Job::new("payments".to_string(), json!({"card": "4111", "amount": 10}))
    ///     .with_encryption(config)
    ///     .with_pii_fields(vec!["card"]);
    /// let id = queue.enqueue(job).await?;
    ///
    /// let stored = queue.get_job(id).await?.unwrap();
    /// assert_eq!(stored.payload, json!({"card": "[ENCRYPTED]", "amount": 10}));
    /// let opened = queue.decrypt_job(stored).await?;
    /// assert_eq!(opened.payload, json!({"card": "4111", "amount": 10}));
    /// # Ok(())
    /// # }
    /// # #[cfg(not(feature = "encryption"))]
    /// # fn main() {}
    /// ```
    #[cfg(feature = "encryption")]
    pub fn with_encryption(
        mut self,
        engine: impl Into<Arc<crate::encryption::EncryptionEngine>>,
    ) -> Self {
        self.encryption.engine = Some(engine.into());
        self
    }

    /// Encrypts every job enqueued to `queues` (`"*"` for all), like
    /// [`JobQueue::with_encrypted_queues`](crate::JobQueue::with_encrypted_queues).
    #[cfg(feature = "encryption")]
    pub fn with_encrypted_queues<I, S>(mut self, queues: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.encryption.encrypted_queues = queues.into_iter().map(Into::into).collect();
        self
    }

    /// The engine set with [`TestQueue::with_encryption`], if any.
    #[cfg(feature = "encryption")]
    pub fn encryption_engine(&self) -> Option<&Arc<crate::encryption::EncryptionEngine>> {
        self.encryption.engine.as_ref()
    }

    /// Returns `job` with its payload decrypted, like
    /// [`JobQueue::decrypt_job`](crate::JobQueue::decrypt_job) (which workers call just
    /// before the handler). Unencrypted jobs are returned unchanged.
    pub async fn decrypt_job(&self, job: Job) -> Result<Job> {
        self.encryption.open(job).await
    }

    /// Get the number of jobs in a specific status for a queue.
    ///
    /// This is a testing utility that helps you verify the state of your
    /// job queue during tests.
    ///
    /// # Arguments
    ///
    /// * `queue_name` - The name of the queue to check
    /// * `status` - The job status to count
    ///
    /// # Returns
    ///
    /// The number of jobs in the specified status
    ///
    /// # Examples
    ///
    /// ```rust
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// use hammerwork::queue::test::TestQueue;
    /// use hammerwork::{Job, JobStatus, queue::DatabaseQueue};
    /// use serde_json::json;
    ///
    /// let queue = TestQueue::new();
    ///
    /// // Initially no jobs
    /// assert_eq!(queue.get_job_count("test_queue", &JobStatus::Pending).await, 0);
    ///
    /// // Add some jobs
    /// for i in 0..5 {
    ///     let job = Job::new("test_queue".to_string(), json!({"index": i}));
    ///     queue.enqueue(job).await?;
    /// }
    ///
    /// // Should have 5 pending jobs
    /// assert_eq!(queue.get_job_count("test_queue", &JobStatus::Pending).await, 5);
    ///
    /// // Process one job
    /// let job = queue.dequeue("test_queue").await?.unwrap();
    /// queue.complete_job(job.id).await?;
    ///
    /// // Should have 4 pending and 1 completed
    /// assert_eq!(queue.get_job_count("test_queue", &JobStatus::Pending).await, 4);
    /// assert_eq!(queue.get_job_count("test_queue", &JobStatus::Completed).await, 1);
    /// # Ok(())
    /// # }
    /// ```
    pub async fn get_job_count(&self, queue_name: &str, status: &JobStatus) -> usize {
        let storage = self.storage.read().await;
        storage
            .queues
            .get(queue_name)
            .and_then(|q| q.get(status))
            .map(|jobs| jobs.len())
            .unwrap_or(0)
    }

    /// Get all jobs for testing purposes.
    ///
    /// This returns all jobs stored in the TestQueue, regardless of their
    /// queue name or status. This is useful for comprehensive testing and
    /// debugging.
    ///
    /// # Returns
    ///
    /// A vector containing all jobs in the TestQueue
    ///
    /// # Examples
    ///
    /// ```rust
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// use hammerwork::queue::test::TestQueue;
    /// use hammerwork::{Job, queue::DatabaseQueue};
    /// use serde_json::json;
    ///
    /// let queue = TestQueue::new();
    ///
    /// // Add jobs to different queues
    /// let job1 = Job::new("queue_a".to_string(), json!({"type": "a"}));
    /// let job2 = Job::new("queue_b".to_string(), json!({"type": "b"}));
    ///
    /// queue.enqueue(job1).await?;
    /// queue.enqueue(job2).await?;
    ///
    /// // Get all jobs
    /// let all_jobs = queue.get_all_jobs().await;
    /// assert_eq!(all_jobs.len(), 2);
    ///
    /// // Jobs from different queues are included
    /// let queue_names: Vec<String> = all_jobs.iter()
    ///     .map(|job| job.queue_name.clone())
    ///     .collect();
    /// assert!(queue_names.contains(&"queue_a".to_string()));
    /// assert!(queue_names.contains(&"queue_b".to_string()));
    /// # Ok(())
    /// # }
    /// ```
    pub async fn get_all_jobs(&self) -> Vec<Job> {
        let storage = self.storage.read().await;
        storage.jobs.values().cloned().collect()
    }
}

impl TestQueue {
    /// End `job_id` as `Dead` or `TimedOut` with `error`, like the database backends'
    /// terminal transitions. With `apply_policy`, the failure is applied to its
    /// dependents, its workflow's [`FailurePolicy`] and a fail-fast batch; returns the
    /// jobs failed with it.
    async fn end_with_failure(
        &self,
        job_id: JobId,
        status: JobStatus,
        error: &str,
        apply_policy: bool,
    ) -> Result<Vec<JobId>> {
        let transition = if status == JobStatus::TimedOut {
            crate::queue::JobTransition::MarkTimedOut
        } else {
            crate::queue::JobTransition::MarkDead
        };
        let mut storage = self.storage.write().await;
        storage.check_transition(job_id, transition)?;
        storage.update_job_status(job_id, status)?;
        storage.leases.remove(&job_id);

        let now = storage.clock.now();
        if let Some(job) = storage.jobs.get_mut(&job_id) {
            if status == JobStatus::TimedOut {
                job.timed_out_at = Some(now);
            } else {
                job.failed_at = Some(now);
            }
            job.error_message = Some(error.to_string());
        }

        if !apply_policy {
            return Ok(Vec::new());
        }
        let mut cancelled = storage.apply_workflow_failure(job_id, now);
        for id in storage.apply_batch_failure(job_id, now) {
            if !cancelled.contains(&id) {
                cancelled.push(id);
            }
        }
        Ok(cancelled)
    }

    /// Complete a job and release its dependents, as the database backends do.
    /// Returns the dependents that became runnable.
    async fn complete_and_resolve(&self, job_id: JobId) -> Result<Vec<JobId>> {
        let mut storage = self.storage.write().await;
        storage.check_transition(job_id, crate::queue::JobTransition::Complete)?;
        storage.update_job_status(job_id, JobStatus::Completed)?;
        storage.leases.remove(&job_id);

        let now = storage.clock.now();
        if let Some(job) = storage.jobs.get_mut(&job_id) {
            // Like the database backends, completing a job by hand does not reschedule a
            // recurring job; `finish_job_run` does that for a finished run.
            job.completed_at = Some(now);
        }
        Ok(storage.resolve_dependents(job_id, now))
    }
}

impl Default for TestQueue {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait]
impl DatabaseQueue for TestQueue {
    type Database = TestDatabaseType;

    async fn enqueue(&self, mut job: Job) -> Result<JobId> {
        job.validate_dependencies()?;
        self.encryption.seal(std::slice::from_mut(&mut job)).await?;
        let mut storage = self.storage.write().await;

        let now = storage.clock.now();
        TestStorage::stamp_enqueued(&mut job, now);
        storage.settle_new_dependents(std::slice::from_mut(&mut job), now);

        // Store the job
        storage.jobs.insert(job.id, job.clone());
        storage.add_job_to_queue(&job);

        Ok(job.id)
    }

    async fn dequeue(&self, queue_name: &str) -> Result<Option<Job>> {
        self.dequeue_leased(queue_name, None, super::DEFAULT_LEASE_DURATION)
            .await
    }

    async fn dequeue_with_priority_weights(
        &self,
        queue_name: &str,
        weights: &PriorityWeights,
    ) -> Result<Option<Job>> {
        self.dequeue_leased(queue_name, Some(weights), super::DEFAULT_LEASE_DURATION)
            .await
    }

    async fn dequeue_leased(
        &self,
        queue_name: &str,
        weights: Option<&PriorityWeights>,
        lease: std::time::Duration,
    ) -> Result<Option<Job>> {
        let mut storage = self.storage.write().await;

        let Some(job_id) = storage.get_next_job(queue_name, weights) else {
            return Ok(None);
        };
        storage.update_job_status(job_id, JobStatus::Running)?;

        let now = storage.clock.now();
        // Like the database backends, the claim holds a lease from the start.
        storage
            .leases
            .insert(job_id, (now, lease_expiry(now, lease)));
        Ok(storage.jobs.get_mut(&job_id).map(|job| {
            job.started_at = Some(now);
            // Like the database backends, every dequeue counts as an attempt.
            job.attempts += 1;
            job.clone()
        }))
    }

    async fn dequeue_batch_leased(
        &self,
        queue_name: &str,
        weights: Option<&PriorityWeights>,
        lease: std::time::Duration,
        max_jobs: usize,
    ) -> Result<Vec<Job>> {
        // One lock for the whole batch: concurrent callers never see the same job as
        // claimable, like `FOR UPDATE SKIP LOCKED` on the database backends.
        let mut storage = self.storage.write().await;
        let max_jobs = max_jobs.min(super::MAX_DEQUEUE_BATCH_SIZE);
        let ids = storage.get_next_jobs(queue_name, weights, max_jobs);
        let now = storage.clock.now();
        let mut jobs = Vec::with_capacity(ids.len());
        for job_id in ids {
            storage.update_job_status(job_id, JobStatus::Running)?;
            storage
                .leases
                .insert(job_id, (now, lease_expiry(now, lease)));
            if let Some(job) = storage.jobs.get_mut(&job_id) {
                job.started_at = Some(now);
                job.attempts += 1;
                jobs.push(job.clone());
            }
        }
        Ok(jobs)
    }

    async fn release_job_run(&self, run: &Job) -> Result<bool> {
        let mut storage = self.storage.write().await;
        // Only the run that was dequeued is released, as on the database backends.
        let same_run = storage.jobs.get(&run.id).is_some_and(|job| {
            job.status == JobStatus::Running && lifecycle::is_same_run(job, run)
        });
        if !same_run {
            return Ok(false);
        }
        storage.update_job_status(run.id, JobStatus::Pending)?;
        storage.leases.remove(&run.id);
        if let Some(job) = storage.jobs.get_mut(&run.id) {
            job.started_at = None;
            job.attempts = (job.attempts - 1).max(0);
        }
        Ok(true)
    }

    async fn complete_job(&self, job_id: JobId) -> Result<()> {
        self.complete_and_resolve(job_id).await?;
        Ok(())
    }

    async fn fail_job(&self, job_id: JobId, error_message: &str) -> Result<()> {
        let mut storage = self.storage.write().await;
        storage.check_transition(job_id, crate::queue::JobTransition::Fail)?;

        // A terminal failure, as on the database backends: no automatic retry, and
        // `attempts` is left alone (it counts dequeues).
        storage.update_job_status(job_id, JobStatus::Failed)?;
        storage.leases.remove(&job_id);
        let now = storage.clock.now();
        if let Some(job) = storage.jobs.get_mut(&job_id) {
            job.failed_at = Some(now);
            job.error_message = Some(error_message.to_string());
        }

        storage.apply_workflow_failure(job_id, now);
        storage.apply_batch_failure(job_id, now);
        Ok(())
    }

    async fn retry_job(&self, job_id: JobId, retry_at: DateTime<Utc>) -> Result<()> {
        let mut storage = self.storage.write().await;
        storage.check_transition(job_id, crate::queue::JobTransition::Retry)?;
        // Like the database backends, re-evaluate the dependencies of a job retried by
        // hand, refusing when one of them can no longer complete.
        let dependency_status = match storage.jobs.get(&job_id) {
            Some(job) if !job.depends_on.is_empty() => {
                let states = storage.dependency_states(&job.depends_on);
                Some(lifecycle::reevaluate_dependencies(job, &states, "retried")?)
            }
            _ => None,
        };

        storage.update_job_status(job_id, JobStatus::Pending)?;
        storage.leases.remove(&job_id);

        if let Some(job) = storage.jobs.get_mut(&job_id) {
            job.scheduled_at = retry_at;
            job.started_at = None;
            if let Some(status) = dependency_status {
                job.dependency_status = status;
            }
        }

        Ok(())
    }

    async fn get_job(&self, job_id: JobId) -> Result<Option<Job>> {
        let storage = self.storage.read().await;
        #[cfg_attr(not(feature = "encryption"), allow(unused_mut))]
        let mut job = storage.jobs.get(&job_id).cloned();
        // Like the database backends: the ciphertext of an archived job stays in the
        // archive and is not loaded, so it must be restored to be decrypted.
        #[cfg(feature = "encryption")]
        if let Some(job) = job.as_mut()
            && job.status == JobStatus::Archived
        {
            job.encrypted_payload = None;
        }
        Ok(job)
    }

    async fn delete_job(&self, job_id: JobId) -> Result<()> {
        let mut storage = self.storage.write().await;

        if let Some(job) = storage.jobs.remove(&job_id) {
            // Remove from queue
            if let Some(queue_jobs) = storage.queues.get_mut(&job.queue_name)
                && let Some(status_jobs) = queue_jobs.get_mut(&job.status)
            {
                status_jobs.retain(|id| *id != job_id);
            }

            // Remove from batch if applicable
            if let Some(batch_id) = job.batch_id
                && let Some(batch_jobs) = storage.batch_jobs.get_mut(&batch_id)
            {
                batch_jobs.retain(|id| *id != job_id);
            }

            storage.job_results.remove(&job_id);

            // Clean up dependencies
            storage.dependencies.remove(&job_id);
            for deps in storage.dependents.values_mut() {
                deps.retain(|id| *id != job_id);
            }

            Ok(())
        } else {
            Err(HammerworkError::JobNotFound {
                id: job_id.to_string(),
            })
        }
    }

    // Batch operations
    async fn enqueue_batch(&self, mut batch: JobBatch) -> Result<BatchId> {
        crate::job::validate_dependency_limits(&batch.jobs)?;
        self.encryption.seal(&mut batch.jobs).await?;
        let mut storage = self.storage.write().await;

        batch.created_at = storage.clock.now();
        let batch_id = batch.id;

        // Enqueue all jobs in the batch
        let now = storage.clock.now();
        let mut jobs = Vec::with_capacity(batch.jobs.len());
        for mut job in batch.jobs.iter().cloned() {
            job.batch_id = Some(batch_id);
            TestStorage::stamp_enqueued(&mut job, now);
            jobs.push(job);
        }
        storage.settle_new_dependents(&mut jobs, now);
        let mut job_ids = Vec::with_capacity(jobs.len());
        for job in jobs {
            storage.add_job_to_queue(&job);
            job_ids.push(job.id);
            storage.jobs.insert(job.id, job);
        }

        storage.batch_jobs.insert(batch_id, job_ids);
        storage.batches.insert(batch_id, batch);

        Ok(batch_id)
    }

    async fn get_batch_status(&self, batch_id: BatchId) -> Result<BatchResult> {
        let storage = self.storage.read().await;

        let batch = storage
            .batches
            .get(&batch_id)
            .ok_or_else(|| HammerworkError::JobNotFound {
                id: batch_id.to_string(),
            })?;

        let job_ids =
            storage
                .batch_jobs
                .get(&batch_id)
                .ok_or_else(|| HammerworkError::JobNotFound {
                    id: batch_id.to_string(),
                })?;

        // Like the database backends: tally the batch's jobs, archived ones with the
        // status they were archived with, and collect the errors of its live jobs.
        let mut counts: HashMap<String, i64> = HashMap::new();
        let mut job_errors = HashMap::new();
        for job_id in job_ids {
            if let Some(status) = storage.status_with_archived(*job_id) {
                *counts.entry(format!("{status:?}")).or_default() += 1;
            }
            if let Some(job) = storage.jobs.get(job_id)
                && job.status != JobStatus::Archived
                && let Some(error) = &job.error_message
            {
                job_errors.insert(job.id, error.clone());
            }
        }

        let total_jobs = u32::try_from(job_ids.len()).unwrap_or(u32::MAX);
        let progress = crate::batch::BatchProgress::from_status_counts(
            counts,
            total_jobs,
            &batch.failure_mode,
        );
        // The backends stamp `completed_at` once nothing in the batch is left to run.
        let completed_at = (progress.pending == 0 && progress.status != BatchStatus::Pending)
            .then(|| storage.clock.now());

        Ok(BatchResult {
            batch_id,
            total_jobs,
            completed_jobs: progress.completed,
            failed_jobs: progress.failed,
            pending_jobs: progress.pending,
            status: progress.status,
            created_at: batch.created_at,
            completed_at,
            error_summary: (progress.failed > 0)
                .then(|| format!("{} of {} jobs failed", progress.failed, total_jobs)),
            job_errors,
        })
    }

    async fn get_batch_jobs(&self, batch_id: BatchId) -> Result<Vec<Job>> {
        let storage = self.storage.read().await;

        let job_ids = storage
            .batch_jobs
            .get(&batch_id)
            .ok_or_else(|| HammerworkError::Batch {
                message: format!("Batch {} not found", batch_id),
            })?;

        let jobs: Vec<Job> = job_ids
            .iter()
            .filter_map(|id| storage.jobs.get(id).cloned())
            .collect();

        Ok(jobs)
    }

    async fn delete_batch(&self, batch_id: BatchId) -> Result<()> {
        let mut storage = self.storage.write().await;

        // Delete all jobs in the batch
        if let Some(job_ids) = storage.batch_jobs.remove(&batch_id) {
            for job_id in job_ids {
                if let Some(job) = storage.jobs.remove(&job_id)
                    && let Some(queue_jobs) = storage.queues.get_mut(&job.queue_name)
                    && let Some(status_jobs) = queue_jobs.get_mut(&job.status)
                {
                    status_jobs.retain(|id| *id != job_id);
                }
            }
        }

        storage.batches.remove(&batch_id);
        Ok(())
    }

    // Dead job management
    async fn mark_job_dead(&self, job_id: JobId, error_message: &str) -> Result<()> {
        self.end_with_failure(job_id, JobStatus::Dead, error_message, true)
            .await?;
        Ok(())
    }

    async fn mark_job_timed_out(&self, job_id: JobId, error_message: &str) -> Result<()> {
        self.end_with_failure(job_id, JobStatus::TimedOut, error_message, true)
            .await?;
        Ok(())
    }

    async fn complete_job_run_with_children(
        &self,
        run: &Job,
        children: Vec<Job>,
    ) -> Result<Option<crate::queue::RecordedOutcome>> {
        crate::job::validate_dependency_limits(&children)?;
        {
            let storage = self.storage.read().await;
            match storage.jobs.get(&run.id) {
                Some(current) if lifecycle::Guard::Run(run).admits(current) => {}
                _ => return Ok(None),
            }
        }
        // Like the database backends, children that join a workflow raise its job
        // count, so the workflow does not complete before they finish.
        let mut spawned = Vec::with_capacity(children.len());
        for child in children {
            let workflow_id = child.workflow_id;
            let id = self.enqueue(child).await?;
            if let Some(workflow_id) = workflow_id {
                self.storage.write().await.join_workflow(workflow_id, id);
            }
            spawned.push(id);
        }
        let recorded = self
            .finish_job_run(run, crate::queue::JobOutcome::Completed)
            .await?;
        Ok(recorded.map(|mut recorded| {
            recorded.spawned = spawned;
            recorded
        }))
    }

    async fn finish_job_run(
        &self,
        run: &Job,
        outcome: crate::queue::JobOutcome,
    ) -> Result<Option<crate::queue::RecordedOutcome>> {
        use crate::queue::{JobOutcome, RecordedOutcome};

        // Like the database backends: only the run that was dequeued is recorded, and a
        // recurring job that finished is rescheduled instead of failing its dependents.
        let next_run_at = {
            let storage = self.storage.read().await;
            let Some(current) = storage.jobs.get(&run.id) else {
                return Ok(None);
            };
            if !lifecycle::Guard::Run(run).admits(current) {
                return Ok(None);
            }
            match outcome.terminal_status() {
                Some(_) if current.recurring => {
                    lifecycle::next_cron_run(current, storage.clock.now())
                }
                _ => None,
            }
        };
        let apply_policy = next_run_at.is_none();
        let mut recorded = match outcome {
            JobOutcome::Completed => {
                let mut recorded = RecordedOutcome::new(JobStatus::Completed);
                if next_run_at.is_none() {
                    recorded.unblocked = self.complete_and_resolve(run.id).await?;
                }
                recorded
            }
            JobOutcome::Retry {
                retry_at,
                error,
                timed_out,
            } => {
                // A worker computes `retry_at` as "now + backoff" on the wall clock; keep
                // the backoff and move it onto the mock clock, as the database backends
                // move it onto the database clock.
                let retry_at = {
                    let storage = self.storage.read().await;
                    let delay = retry_at - Utc::now();
                    storage
                        .clock
                        .now()
                        .checked_add_signed(delay)
                        .unwrap_or(retry_at)
                };
                self.retry_job(run.id, retry_at).await?;
                // The run's error (and timeout) is recorded, as on the database backends.
                let mut storage = self.storage.write().await;
                let now = storage.clock.now();
                if let Some(job) = storage.jobs.get_mut(&run.id) {
                    job.error_message = Some(error);
                    if timed_out {
                        job.timed_out_at = Some(now);
                    }
                }
                RecordedOutcome::new(JobStatus::Pending)
            }
            JobOutcome::Dead { error } => {
                let mut recorded = RecordedOutcome::new(JobStatus::Dead);
                recorded.cancelled = self
                    .end_with_failure(run.id, JobStatus::Dead, &error, apply_policy)
                    .await?;
                recorded
            }
            JobOutcome::TimedOut { error } => {
                let mut recorded = RecordedOutcome::new(JobStatus::TimedOut);
                recorded.cancelled = self
                    .end_with_failure(run.id, JobStatus::TimedOut, &error, apply_policy)
                    .await?;
                recorded
            }
        };
        if let Some(next) = next_run_at {
            // The same job (same id) runs again, as on the database backends; a failed
            // run's error stays visible until the next run.
            let keep_failure = recorded.status != JobStatus::Completed;
            self.storage
                .write()
                .await
                .reschedule_in_place(run.id, next, keep_failure)?;
            recorded.status = JobStatus::Pending;
            recorded.next_run_at = Some(next);
        }
        Ok(Some(recorded))
    }

    async fn get_dead_jobs(&self, limit: Option<u32>, offset: Option<u32>) -> Result<Vec<Job>> {
        let storage = self.storage.read().await;

        let mut dead_jobs: Vec<Job> = storage
            .jobs
            .values()
            // Like the database backends: `Dead` only, not `TimedOut`.
            .filter(|job| job.status == JobStatus::Dead)
            .cloned()
            .collect();

        // Sort by failed_at or timed_out_at (newest first)
        dead_jobs.sort_by(|a, b| {
            let a_time = a.failed_at.or(a.timed_out_at).unwrap_or(a.created_at);
            let b_time = b.failed_at.or(b.timed_out_at).unwrap_or(b.created_at);
            b_time.cmp(&a_time)
        });

        // Apply offset and limit
        let offset = offset.unwrap_or(0) as usize;
        let limit = limit.unwrap_or(100) as usize;

        Ok(dead_jobs.into_iter().skip(offset).take(limit).collect())
    }

    async fn get_dead_jobs_by_queue(
        &self,
        queue_name: &str,
        limit: Option<u32>,
        offset: Option<u32>,
    ) -> Result<Vec<Job>> {
        let storage = self.storage.read().await;

        let mut dead_jobs: Vec<Job> = storage
            .jobs
            .values()
            .filter(|job| job.queue_name == queue_name && job.status == JobStatus::Dead)
            .cloned()
            .collect();

        // Sort by failed_at or timed_out_at (newest first)
        dead_jobs.sort_by(|a, b| {
            let a_time = a.failed_at.or(a.timed_out_at).unwrap_or(a.created_at);
            let b_time = b.failed_at.or(b.timed_out_at).unwrap_or(b.created_at);
            b_time.cmp(&a_time)
        });

        // Apply offset and limit
        let offset = offset.unwrap_or(0) as usize;
        let limit = limit.unwrap_or(100) as usize;

        Ok(dead_jobs.into_iter().skip(offset).take(limit).collect())
    }

    async fn retry_dead_job(&self, job_id: JobId) -> Result<()> {
        let mut storage = self.storage.write().await;
        // A job that is not dead (or timed out) is rejected with `InvalidJobTransition`,
        // as on the database backends.
        storage.check_transition(job_id, crate::queue::JobTransition::RetryDead)?;

        storage.update_job_status(job_id, JobStatus::Pending)?;
        storage.leases.remove(&job_id);

        let now = storage.clock.now();
        if let Some(job) = storage.jobs.get_mut(&job_id) {
            job.attempts = 0;
            job.error_message = None;
            job.started_at = None;
            job.failed_at = None;
            job.timed_out_at = None;
            job.scheduled_at = now;
        }

        Ok(())
    }

    async fn purge_dead_jobs(&self, older_than: DateTime<Utc>) -> Result<u64> {
        let mut storage = self.storage.write().await;

        let dead_job_ids: Vec<JobId> = storage
            .jobs
            .iter()
            // Like the database backends: dead jobs that failed before `older_than`.
            .filter(|(_, job)| {
                job.status == JobStatus::Dead
                    && job
                        .failed_at
                        .is_some_and(|failed_at| failed_at < older_than)
            })
            .map(|(id, _)| *id)
            .collect();

        let count = dead_job_ids.len() as u64;

        for job_id in dead_job_ids {
            if let Some(job) = storage.jobs.remove(&job_id)
                && let Some(queue_jobs) = storage.queues.get_mut(&job.queue_name)
                && let Some(status_jobs) = queue_jobs.get_mut(&job.status)
            {
                status_jobs.retain(|id| *id != job_id);
            }
        }

        Ok(count)
    }

    async fn get_dead_job_summary(&self) -> Result<DeadJobSummary> {
        let storage = self.storage.read().await;

        let mut dead_jobs_by_queue: HashMap<String, u64> = HashMap::new();
        let mut error_patterns: HashMap<String, u64> = HashMap::new();
        let mut total_dead_jobs = 0;
        let mut oldest_dead_job: Option<DateTime<Utc>> = None;
        let mut newest_dead_job: Option<DateTime<Utc>> = None;

        for job in storage.jobs.values() {
            if job.status == JobStatus::Dead {
                total_dead_jobs += 1;

                *dead_jobs_by_queue
                    .entry(job.queue_name.clone())
                    .or_insert(0) += 1;

                if let Some(error) = &job.error_message {
                    let error_key = error.split('\n').next().unwrap_or(error).to_string();
                    *error_patterns.entry(error_key).or_insert(0) += 1;
                }

                let job_time = job.failed_at.or(job.timed_out_at).unwrap_or(job.created_at);

                if oldest_dead_job.is_none() || job_time < oldest_dead_job.unwrap() {
                    oldest_dead_job = Some(job_time);
                }

                if newest_dead_job.is_none() || job_time > newest_dead_job.unwrap() {
                    newest_dead_job = Some(job_time);
                }
            }
        }

        Ok(DeadJobSummary {
            total_dead_jobs,
            dead_jobs_by_queue,
            error_patterns,
            oldest_dead_job,
            newest_dead_job,
        })
    }

    // Statistics and monitoring
    async fn get_queue_stats(&self, queue_name: &str) -> Result<QueueStats> {
        let storage = self.storage.read().await;

        // An unknown queue has no jobs: zero counts, as on the database backends.
        let no_jobs = HashMap::new();
        let queue_jobs = storage.queues.get(queue_name).unwrap_or(&no_jobs);

        let pending_count = queue_jobs
            .get(&JobStatus::Pending)
            .map(|j| j.len())
            .unwrap_or(0) as u64;
        let running_count = queue_jobs
            .get(&JobStatus::Running)
            .map(|j| j.len())
            .unwrap_or(0) as u64;
        let completed_count = queue_jobs
            .get(&JobStatus::Completed)
            .map(|j| j.len())
            .unwrap_or(0) as u64;
        let dead_count = queue_jobs
            .get(&JobStatus::Dead)
            .map(|j| j.len())
            .unwrap_or(0) as u64;
        let timed_out_count = queue_jobs
            .get(&JobStatus::TimedOut)
            .map(|j| j.len())
            .unwrap_or(0) as u64;

        // Calculate processing times for completed jobs
        let completed_jobs: Vec<&Job> = queue_jobs
            .get(&JobStatus::Completed)
            .map(|job_ids| {
                job_ids
                    .iter()
                    .filter_map(|id| storage.jobs.get(id))
                    .collect()
            })
            .unwrap_or_default();

        let processing_times: Vec<i64> = completed_jobs
            .iter()
            .filter_map(|job| match (job.started_at, job.completed_at) {
                (Some(start), Some(end)) => Some((end - start).num_milliseconds()),
                _ => None,
            })
            .collect();

        let avg_processing_time_ms = if !processing_times.is_empty() {
            processing_times.iter().sum::<i64>() as f64 / processing_times.len() as f64
        } else {
            0.0
        };

        let min_processing_time_ms = processing_times.iter().min().copied().unwrap_or(0) as u64;
        let max_processing_time_ms = processing_times.iter().max().copied().unwrap_or(0) as u64;

        // Get throughput (jobs completed in last minute)
        let one_minute_ago = storage.clock.now() - chrono::Duration::minutes(1);
        let jobs_per_minute = completed_jobs
            .iter()
            .filter(|job| job.completed_at.unwrap_or(job.created_at) > one_minute_ago)
            .count() as f64;

        let total_processed = completed_count + dead_count + timed_out_count;
        // `Failed` jobs only, as on the database backends: dead jobs are in `dead`.
        let failed_count = queue_jobs
            .get(&JobStatus::Failed)
            .map(|j| j.len())
            .unwrap_or(0) as u64;
        let error_rate = if total_processed > 0 {
            (failed_count + dead_count + timed_out_count) as f64 / total_processed as f64
        } else {
            0.0
        };

        let statistics = crate::stats::JobStatistics {
            total_processed,
            completed: completed_count,
            failed: failed_count,
            dead: dead_count,
            timed_out: timed_out_count,
            running: running_count,
            avg_processing_time_ms,
            min_processing_time_ms,
            max_processing_time_ms,
            throughput_per_minute: jobs_per_minute,
            error_rate,
            priority_stats: None,
            time_window: std::time::Duration::from_secs(3600),
            calculated_at: storage.clock.now(),
        };

        Ok(QueueStats {
            queue_name: queue_name.to_string(),
            pending_count,
            running_count,
            completed_count,
            dead_count,
            timed_out_count,
            statistics,
        })
    }

    async fn get_all_queue_stats(&self) -> Result<Vec<QueueStats>> {
        let storage = self.storage.read().await;
        let queue_names: Vec<String> = storage.queues.keys().cloned().collect();
        drop(storage);

        let mut stats = Vec::new();
        for queue_name in queue_names {
            if let Ok(queue_stats) = self.get_queue_stats(&queue_name).await {
                stats.push(queue_stats);
            }
        }

        Ok(stats)
    }

    async fn get_priority_stats(&self, queue_name: &str) -> Result<crate::priority::PriorityStats> {
        let storage = self.storage.read().await;

        let mut job_counts = std::collections::HashMap::new();
        let mut avg_processing_times = std::collections::HashMap::new();
        let mut recent_throughput = std::collections::HashMap::new();
        let mut priority_distribution = std::collections::HashMap::new();

        // Initialize counts for all priorities
        for priority in [
            crate::priority::JobPriority::Background,
            crate::priority::JobPriority::Low,
            crate::priority::JobPriority::Normal,
            crate::priority::JobPriority::High,
            crate::priority::JobPriority::Critical,
        ] {
            job_counts.insert(priority, 0);
            avg_processing_times.insert(priority, 0.0);
            recent_throughput.insert(priority, 0);
            priority_distribution.insert(priority, 0.0);
        }

        let mut total_jobs = 0u64;

        if let Some(queue_jobs) = storage.queues.get(queue_name) {
            for jobs in queue_jobs.values() {
                for job_id in jobs {
                    if let Some(job) = storage.jobs.get(job_id) {
                        *job_counts.entry(job.priority).or_insert(0) += 1;
                        total_jobs += 1;

                        // Calculate processing time if job is completed
                        if let (Some(started), Some(completed)) = (job.started_at, job.completed_at)
                        {
                            let processing_time = (completed - started).num_milliseconds() as f64;
                            let current_avg =
                                avg_processing_times.get(&job.priority).unwrap_or(&0.0);
                            let current_count = *job_counts.get(&job.priority).unwrap_or(&1);
                            let new_avg = (current_avg * (current_count - 1) as f64
                                + processing_time)
                                / current_count as f64;
                            avg_processing_times.insert(job.priority, new_avg);
                        }
                    }
                }
            }
        }

        // Calculate priority distribution percentages
        for (priority, count) in &job_counts {
            let percentage = if total_jobs > 0 {
                (*count as f32 / total_jobs as f32) * 100.0
            } else {
                0.0
            };
            priority_distribution.insert(*priority, percentage);
        }

        Ok(crate::priority::PriorityStats {
            job_counts,
            avg_processing_times,
            recent_throughput,
            priority_distribution,
        })
    }

    async fn get_job_counts_by_status(&self, queue_name: &str) -> Result<HashMap<String, u64>> {
        let storage = self.storage.read().await;

        let mut counts = HashMap::new();

        if let Some(queue_jobs) = storage.queues.get(queue_name) {
            // Keyed by the stored status name ("Pending", ...) like the database
            // backends, which group by the status column; empty statuses are omitted.
            for (status, jobs) in queue_jobs {
                if !jobs.is_empty() {
                    counts.insert(status.as_str().to_string(), jobs.len() as u64);
                }
            }
        }

        Ok(counts)
    }

    async fn get_processing_times(
        &self,
        queue_name: &str,
        since: DateTime<Utc>,
    ) -> Result<Vec<i64>> {
        let storage = self.storage.read().await;

        let processing_times: Vec<i64> = storage
            .jobs
            .values()
            .filter(|job| {
                job.queue_name == queue_name
                    && job.status == JobStatus::Completed
                    && job.completed_at.unwrap_or(job.created_at) >= since
            })
            .filter_map(|job| match (job.started_at, job.completed_at) {
                (Some(start), Some(end)) => Some((end - start).num_milliseconds()),
                _ => None,
            })
            .collect();

        Ok(processing_times)
    }

    async fn get_error_frequencies(
        &self,
        queue_name: Option<&str>,
        since: DateTime<Utc>,
    ) -> Result<HashMap<String, u64>> {
        let storage = self.storage.read().await;

        let mut frequencies = HashMap::new();

        for job in storage.jobs.values() {
            if let Some(qn) = queue_name
                && job.queue_name != qn
            {
                continue;
            }

            if (job.status == JobStatus::Failed
                || job.status == JobStatus::Dead
                || job.status == JobStatus::TimedOut)
                && job.failed_at.or(job.timed_out_at).unwrap_or(job.created_at) >= since
                && let Some(error) = &job.error_message
            {
                let error_key = error.split('\n').next().unwrap_or(error).to_string();
                *frequencies.entry(error_key).or_insert(0) += 1;
            }
        }

        Ok(frequencies)
    }

    async fn get_jobs_completed_in_range(
        &self,
        queue_name: Option<&str>,
        start_time: DateTime<Utc>,
        end_time: DateTime<Utc>,
        limit: Option<u32>,
    ) -> Result<Vec<Job>> {
        let storage = self.storage.read().await;
        let mut matching_jobs = Vec::new();

        for job in storage.jobs.values() {
            // Filter by queue name if specified
            if let Some(qn) = queue_name
                && job.queue_name != qn
            {
                continue;
            }

            // Check if job is completed and within time range
            if job.status == JobStatus::Completed
                && let Some(completed_at) = job.completed_at
                && completed_at >= start_time
                && completed_at < end_time
            {
                matching_jobs.push(job.clone());
            }
        }

        // Sort by completion time (most recent first)
        matching_jobs.sort_by_key(|j| std::cmp::Reverse(j.completed_at));

        // Apply limit if specified
        if let Some(limit) = limit {
            matching_jobs.truncate(limit as usize);
        }

        Ok(matching_jobs)
    }

    // Cron job management
    async fn enqueue_cron_job(&self, mut job: Job) -> Result<JobId> {
        if job.cron_schedule.is_none() {
            return Err(HammerworkError::Queue {
                message: "Job must have a cron schedule".to_string(),
            });
        }

        job.recurring = true;

        // Calculate initial next_run_at
        if let Ok(schedule) = job
            .cron_schedule
            .as_ref()
            .unwrap()
            .parse::<cron::Schedule>()
        {
            let timezone = job
                .timezone
                .as_ref()
                .and_then(|tz| tz.parse::<chrono_tz::Tz>().ok())
                .unwrap_or(chrono_tz::UTC);

            let now = self.clock.now();
            let now_in_tz = now.with_timezone(&timezone);

            if let Some(next) = schedule.after(&now_in_tz).next() {
                job.next_run_at = Some(next.with_timezone(&Utc));
                job.scheduled_at = job.next_run_at.unwrap();
            }
        } else {
            return Err(HammerworkError::Queue {
                message: format!("Invalid cron schedule: {:?}", job.cron_schedule),
            });
        }

        self.enqueue(job).await
    }

    async fn get_due_cron_jobs(&self, queue_name: Option<&str>) -> Result<Vec<Job>> {
        let storage = self.storage.read().await;
        let now = storage.clock.now();

        let due_jobs: Vec<Job> = storage
            .jobs
            .values()
            .filter(|job| {
                // Like the database backends: a recurring job waiting for its next run
                job.recurring
                    && job.status == JobStatus::Pending
                    && job.next_run_at.is_none_or(|next| next <= now)
                    && (queue_name.is_none() || job.queue_name == queue_name.unwrap())
            })
            .cloned()
            .collect();

        Ok(due_jobs)
    }

    async fn reschedule_cron_job(&self, job_id: JobId, next_run_at: DateTime<Utc>) -> Result<()> {
        let mut storage = self.storage.write().await;
        storage.check_transition(job_id, crate::queue::JobTransition::RescheduleCron)?;
        if !storage.jobs.get(&job_id).is_some_and(|job| job.recurring) {
            return Err(HammerworkError::Queue {
                message: format!("Job {job_id} is not a recurring job"),
            });
        }
        // The job itself is rescheduled (same id), like on the database backends.
        storage.reschedule_in_place(job_id, next_run_at, false)
    }

    async fn get_recurring_jobs(&self, queue_name: &str) -> Result<Vec<Job>> {
        let storage = self.storage.read().await;

        let recurring_jobs: Vec<Job> = storage
            .jobs
            .values()
            .filter(|job| job.queue_name == queue_name && job.recurring)
            .cloned()
            .collect();

        Ok(recurring_jobs)
    }

    async fn disable_recurring_job(&self, job_id: JobId) -> Result<()> {
        let mut storage = self.storage.write().await;

        let job = storage
            .jobs
            .get_mut(&job_id)
            .ok_or_else(|| HammerworkError::JobNotFound {
                id: job_id.to_string(),
            })?;
        if job.cron_schedule.is_none() {
            return Err(super::not_a_cron_job(job_id));
        }

        // A pending occurrence is held by `get_next_job`, like on the database backends.
        job.recurring = false;
        job.next_run_at = None;

        Ok(())
    }

    async fn enable_recurring_job(&self, job_id: JobId) -> Result<()> {
        let mut storage = self.storage.write().await;
        let now = storage.clock.now();
        let job = storage
            .jobs
            .get(&job_id)
            .ok_or_else(|| HammerworkError::JobNotFound {
                id: job_id.to_string(),
            })?;
        let next_run_at = super::enabled_recurring_next_run(job, now)?;

        if next_run_at.is_some() {
            // Not running: resume at the next occurrence, like the database backends.
            storage.update_job_status(job_id, JobStatus::Pending)?;
            storage.leases.remove(&job_id);
        }
        let job = storage
            .jobs
            .get_mut(&job_id)
            .ok_or_else(|| HammerworkError::JobNotFound {
                id: job_id.to_string(),
            })?;
        job.recurring = true;
        if let Some(next_run_at) = next_run_at {
            job.scheduled_at = next_run_at;
            job.next_run_at = Some(next_run_at);
            job.attempts = 0;
            job.started_at = None;
            job.completed_at = None;
            job.failed_at = None;
            job.timed_out_at = None;
            job.error_message = None;
        }

        Ok(())
    }

    // Throttling configuration
    async fn set_throttle_config(&self, queue_name: &str, config: ThrottleConfig) -> Result<()> {
        let mut storage = self.storage.write().await;
        storage
            .throttle_configs
            .insert(queue_name.to_string(), config);
        Ok(())
    }

    async fn get_throttle_config(&self, queue_name: &str) -> Result<Option<ThrottleConfig>> {
        let storage = self.storage.read().await;
        Ok(storage.throttle_configs.get(queue_name).cloned())
    }

    async fn remove_throttle_config(&self, queue_name: &str) -> Result<()> {
        let mut storage = self.storage.write().await;
        storage.throttle_configs.remove(queue_name);
        Ok(())
    }

    async fn get_all_throttle_configs(&self) -> Result<HashMap<String, ThrottleConfig>> {
        let storage = self.storage.read().await;
        Ok(storage.throttle_configs.clone())
    }

    async fn get_queue_depth(&self, queue_name: &str) -> Result<u64> {
        let storage = self.storage.read().await;

        let depth = storage
            .queues
            .get(queue_name)
            .and_then(|q| q.get(&JobStatus::Pending))
            .map(|jobs| jobs.len() as u64)
            .unwrap_or(0);

        Ok(depth)
    }

    // Job result storage
    async fn store_job_result(
        &self,
        job_id: JobId,
        result_data: serde_json::Value,
        expires_at: Option<DateTime<Utc>>,
    ) -> Result<()> {
        let mut storage = self.storage.write().await;

        // Verify job exists
        if !storage.jobs.contains_key(&job_id) {
            return Err(HammerworkError::JobNotFound {
                id: job_id.to_string(),
            });
        }

        storage
            .job_results
            .insert(job_id, (result_data, expires_at));
        Ok(())
    }

    async fn get_job_result(&self, job_id: JobId) -> Result<Option<serde_json::Value>> {
        let storage = self.storage.read().await;

        if let Some((result, expires_at)) = storage.job_results.get(&job_id) {
            // Check if expired
            if let Some(exp) = expires_at
                && *exp <= storage.clock.now()
            {
                return Ok(None);
            }
            Ok(Some(result.clone()))
        } else {
            Ok(None)
        }
    }

    async fn delete_job_result(&self, job_id: JobId) -> Result<()> {
        let mut storage = self.storage.write().await;
        storage.job_results.remove(&job_id);
        Ok(())
    }

    async fn cleanup_expired_results(&self) -> Result<u64> {
        let mut storage = self.storage.write().await;
        let now = storage.clock.now();

        let expired_ids: Vec<JobId> = storage
            .job_results
            .iter()
            .filter(|(_, (_, expires_at))| expires_at.map(|exp| exp <= now).unwrap_or(false))
            .map(|(id, _)| *id)
            .collect();

        let count = expired_ids.len() as u64;

        for id in expired_ids {
            storage.job_results.remove(&id);
        }

        Ok(count)
    }

    // Workflow and dependency management
    async fn enqueue_workflow(&self, mut workflow: JobGroup) -> Result<WorkflowId> {
        crate::job::validate_dependency_limits(&workflow.jobs)?;
        self.encryption.seal(&mut workflow.jobs).await?;
        let mut storage = self.storage.write().await;

        workflow.created_at = storage.clock.now();
        let workflow_id = workflow.id;

        // Store dependencies: those of the workflow's dependency map and, like the
        // database backends, each job's own `depends_on`.
        for job in &workflow.jobs {
            if !job.depends_on.is_empty() {
                let deps = workflow.dependencies.entry(job.id).or_default();
                for dep_id in &job.depends_on {
                    if !deps.contains(dep_id) {
                        deps.push(*dep_id);
                    }
                }
            }
        }
        for (job_id, deps) in &workflow.dependencies {
            storage.dependencies.insert(*job_id, deps.clone());

            // Store reverse dependencies
            for dep_id in deps {
                storage.dependents.entry(*dep_id).or_default().push(*job_id);
            }
        }

        // Enqueue all jobs
        let now = storage.clock.now();
        let mut jobs = Vec::with_capacity(workflow.jobs.len());
        for mut job in workflow.jobs.iter().cloned() {
            TestStorage::stamp_enqueued(&mut job, now);

            // A job with dependencies is not available until they complete.
            let has_deps = workflow
                .dependencies
                .get(&job.id)
                .is_some_and(|deps| !deps.is_empty());
            if has_deps {
                job.scheduled_at = DateTime::<Utc>::MAX_UTC; // Far future
            }
            jobs.push(job);
        }
        storage.settle_new_dependents(&mut jobs, now);
        for job in jobs {
            let mut job = job;
            // Dependencies outside the workflow that already completed release it.
            if job.dependency_status == DependencyStatus::Satisfied
                && job.scheduled_at == DateTime::<Utc>::MAX_UTC
            {
                job.scheduled_at = now;
            }
            storage.add_job_to_queue(&job);
            storage.jobs.insert(job.id, job);
        }

        storage.workflows.insert(workflow_id, workflow);
        Ok(workflow_id)
    }

    async fn get_workflow_status(&self, workflow_id: WorkflowId) -> Result<Option<JobGroup>> {
        let storage = self.storage.read().await;

        if let Some(mut workflow) = storage.workflows.get(&workflow_id).cloned() {
            // Workflow counters and status from its jobs, as the database backends
            // compute them on every terminal transition.
            // Archived jobs count with the status they were archived with.
            let mut progress = WorkflowProgress::default();
            for job_id in workflow.jobs.iter().map(|j| j.id) {
                match storage.status_with_archived(job_id) {
                    Some(JobStatus::Completed) => progress.completed += 1,
                    Some(JobStatus::Failed | JobStatus::Dead | JobStatus::TimedOut) => {
                        progress.failed += 1
                    }
                    _ => {}
                }
            }

            workflow.completed_jobs = progress.completed;
            workflow.failed_jobs = progress.failed;
            workflow.status = progress.status(
                workflow.total_jobs,
                &workflow.failure_policy,
                &workflow.status,
            );
            let now = storage.clock.now();
            match workflow.status {
                WorkflowStatus::Completed => {
                    workflow.completed_at.get_or_insert(now);
                }
                WorkflowStatus::Failed => {
                    workflow.failed_at.get_or_insert(now);
                }
                _ => {}
            }

            Ok(Some(workflow))
        } else {
            Ok(None)
        }
    }

    async fn resolve_job_dependencies(&self, completed_job_id: JobId) -> Result<Vec<JobId>> {
        let mut storage = self.storage.write().await;
        let now = storage.clock.now();
        Ok(storage.resolve_dependents(completed_job_id, now))
    }

    async fn get_ready_jobs(&self, queue_name: &str, limit: u32) -> Result<Vec<Job>> {
        let storage = self.storage.read().await;
        let now = storage.clock.now();

        // As the database backends: pending, due jobs whose dependencies do not hold
        // them back, by priority and then schedule.
        let mut ready_jobs: Vec<Job> = storage
            .jobs
            .values()
            .filter(|job| {
                job.queue_name == queue_name
                    && job.status == JobStatus::Pending
                    && matches!(
                        job.dependency_status,
                        DependencyStatus::None | DependencyStatus::Satisfied
                    )
                    && job.scheduled_at <= now
            })
            .cloned()
            .collect();
        ready_jobs.sort_by(|a, b| {
            b.priority
                .cmp(&a.priority)
                .then(a.scheduled_at.cmp(&b.scheduled_at))
                .then(a.created_at.cmp(&b.created_at))
        });
        ready_jobs.truncate(limit as usize);

        Ok(ready_jobs)
    }

    async fn fail_job_dependencies(&self, failed_job_id: JobId) -> Result<Vec<JobId>> {
        let mut storage = self.storage.write().await;
        let now = storage.clock.now();
        Ok(storage.apply_workflow_failure(failed_job_id, now))
    }

    async fn get_workflow_jobs(&self, workflow_id: WorkflowId) -> Result<Vec<Job>> {
        let storage = self.storage.read().await;

        let workflow =
            storage
                .workflows
                .get(&workflow_id)
                .ok_or_else(|| HammerworkError::Workflow {
                    message: format!("Workflow {} not found", workflow_id),
                })?;

        let jobs: Vec<Job> = workflow
            .jobs
            .iter()
            .filter_map(|job| storage.jobs.get(&job.id).cloned())
            .collect();

        Ok(jobs)
    }

    async fn cancel_workflow(&self, workflow_id: WorkflowId) -> Result<()> {
        let mut storage = self.storage.write().await;

        let workflow = storage
            .workflows
            .get(&workflow_id)
            .ok_or_else(|| HammerworkError::Workflow {
                message: format!("Workflow {} not found", workflow_id),
            })?
            .clone();

        // As the database backends' `cancel_workflow_in_tx`: its unfinished jobs
        // (pending, retrying and running) fail with "Workflow cancelled", and a
        // workflow already failed keeps its `failed_at`.
        let now = storage.clock.now();
        let mut job_ids: Vec<JobId> = workflow.jobs.iter().map(|job| job.id).collect();
        let joined: Vec<JobId> = storage
            .jobs
            .values()
            .filter(|job| job.workflow_id == Some(workflow_id) && !job_ids.contains(&job.id))
            .map(|job| job.id)
            .collect();
        job_ids.extend(joined);
        for job_id in job_ids {
            let unfinished = storage.jobs.get(&job_id).is_some_and(|job| {
                matches!(
                    job.status,
                    JobStatus::Pending | JobStatus::Running | JobStatus::Retrying
                )
            });
            if unfinished {
                storage.fail_as_side_effect(
                    job_id,
                    lifecycle::WORKFLOW_CANCELLED_MESSAGE.to_string(),
                    now,
                );
            }
        }

        if let Some(workflow) = storage.workflows.get_mut(&workflow_id) {
            workflow.status = WorkflowStatus::Cancelled;
            workflow.failed_at.get_or_insert(now);
        }

        Ok(())
    }

    // Archival methods
    async fn archive_jobs(
        &self,
        queue_name: Option<&str>,
        policy: &crate::archive::ArchivalPolicy,
        _config: &crate::archive::ArchivalConfig,
        _reason: crate::archive::ArchivalReason,
        _archived_by: Option<&str>,
    ) -> Result<crate::archive::ArchivalStats> {
        let mut storage = self.storage.write().await;
        let now = storage.clock.now();
        let mut jobs_archived = 0u64;
        let mut bytes_archived = 0u64;

        // Find jobs to archive based on policy
        let mut jobs_to_archive = Vec::new();

        for job in storage.jobs.values() {
            // Skip if queue_name filter doesn't match
            if let Some(queue) = queue_name
                && job.queue_name != queue
            {
                continue;
            }

            // Check if job should be archived based on policy and status
            let age = match job.status {
                JobStatus::Completed => job.completed_at.map(|t| now - t),
                JobStatus::Failed => job.failed_at.map(|t| now - t),
                JobStatus::Dead => job.failed_at.map(|t| now - t), // Use failed_at for dead jobs
                JobStatus::TimedOut => job.timed_out_at.map(|t| now - t),
                _ => None,
            };

            if let Some(age) = age
                && policy.should_archive(&job.status, age)
            {
                jobs_to_archive.push(job.id);
            }
        }

        // Archive jobs
        for job_id in jobs_to_archive.iter().take(policy.batch_size) {
            if let Some(job) = storage.jobs.get_mut(job_id) {
                // Calculate payload size for stats
                let payload_size = serde_json::to_string(&job.payload)
                    .map(|s| s.len() as u64)
                    .unwrap_or(0);

                // Update job status
                job.status = JobStatus::Archived;
                // In a real implementation, we'd move to archive table and compress

                jobs_archived += 1;
                bytes_archived += payload_size;
            }
        }

        // Update queue structure separately to avoid borrow checker issues
        let queue_names: Vec<_> = jobs_to_archive
            .iter()
            .take(policy.batch_size)
            .filter_map(|job_id| {
                storage
                    .jobs
                    .get(job_id)
                    .map(|job| (*job_id, job.queue_name.clone()))
            })
            .collect();

        for (job_id, queue_name) in queue_names {
            if let Some(queue_jobs) = storage.queues.get_mut(&queue_name) {
                // Remove from old status list
                for jobs_list in queue_jobs.values_mut() {
                    jobs_list.retain(|&id| id != job_id);
                }
                // Add to archived list
                queue_jobs
                    .entry(JobStatus::Archived)
                    .or_default()
                    .push(job_id);
            }
        }

        // Calculate compression ratio (mock)
        let compression_ratio = if policy.compress_payloads { 0.7 } else { 1.0 };

        Ok(crate::archive::ArchivalStats {
            jobs_archived,
            jobs_purged: 0,
            bytes_archived,
            bytes_purged: 0,
            compression_ratio,
            last_run_at: now,
            operation_duration: std::time::Duration::from_millis(100), // Mock duration
        })
    }

    async fn restore_archived_job(&self, job_id: JobId) -> Result<Job> {
        let mut storage = self.storage.write().await;
        let now = storage.clock.now();

        // Its dependencies may have finished (or failed) since it was archived.
        let dependency_status = match storage.jobs.get(&job_id) {
            Some(job) if job.status == JobStatus::Archived && !job.depends_on.is_empty() => {
                let states = storage.dependency_states(&job.depends_on);
                Some(lifecycle::reevaluate_dependencies(
                    job, &states, "restored",
                )?)
            }
            _ => None,
        };

        if let Some(job) = storage.jobs.get_mut(&job_id) {
            if job.status == JobStatus::Archived {
                if let Some(status) = dependency_status {
                    job.dependency_status = status;
                }
                let queue_name = job.queue_name.clone(); // Clone to avoid borrow checker issues

                // Reset it as the database backends do, on the mock clock.
                let mut restored = crate::archive::reset_for_restore(job.clone());
                restored.scheduled_at = now;
                *job = restored;
                let restored_job = job.clone();

                // Update queue structure
                if let Some(queue_jobs) = storage.queues.get_mut(&queue_name) {
                    // Remove from archived list
                    if let Some(archived_jobs) = queue_jobs.get_mut(&JobStatus::Archived) {
                        archived_jobs.retain(|&id| id != job_id);
                    }
                    // Add to pending list
                    queue_jobs
                        .entry(JobStatus::Pending)
                        .or_default()
                        .push(job_id);
                }

                Ok(restored_job)
            } else {
                // The database backends look only in the archive table, so a job that is
                // not archived is as absent as an unknown id.
                Err(crate::HammerworkError::JobNotFound {
                    id: job_id.to_string(),
                })
            }
        } else {
            Err(crate::HammerworkError::JobNotFound {
                id: job_id.to_string(),
            })
        }
    }

    async fn list_archived_jobs(
        &self,
        queue_name: Option<&str>,
        limit: Option<u32>,
        offset: Option<u32>,
    ) -> Result<Vec<crate::archive::ArchivedJob>> {
        let storage = self.storage.read().await;
        let now = storage.clock.now();

        let mut archived_jobs = Vec::new();

        for job in storage.jobs.values() {
            if job.status != JobStatus::Archived {
                continue;
            }

            // Filter by queue if specified
            if let Some(queue) = queue_name
                && job.queue_name != queue
            {
                continue;
            }

            let payload_size = serde_json::to_string(&job.payload)
                .map(|s| s.len())
                .unwrap_or(0);

            archived_jobs.push(crate::archive::ArchivedJob {
                id: job.id,
                queue_name: job.queue_name.clone(),
                status: job.status,
                created_at: job.created_at,
                archived_at: now, // Mock archived time
                archival_reason: crate::archive::ArchivalReason::Manual, // Mock reason
                original_payload_size: Some(payload_size),
                payload_compressed: false,             // Mock
                archived_by: Some("test".to_string()), // Mock user
            });
        }

        // Apply pagination
        let offset = offset.unwrap_or(0) as usize;
        let limit = limit.unwrap_or(100) as usize;

        archived_jobs.sort_by_key(|j| std::cmp::Reverse(j.archived_at));

        if offset < archived_jobs.len() {
            let end = std::cmp::min(offset + limit, archived_jobs.len());
            Ok(archived_jobs[offset..end].to_vec())
        } else {
            Ok(Vec::new())
        }
    }

    async fn purge_archived_jobs(&self, older_than: DateTime<Utc>) -> Result<u64> {
        let mut storage = self.storage.write().await;
        let mut purged_count = 0u64;

        // Find archived jobs older than cutoff
        let jobs_to_purge: Vec<JobId> = storage
            .jobs
            .values()
            .filter(|job| {
                job.status == JobStatus::Archived && job.created_at < older_than // Mock: use created_at as archived_at
            })
            .map(|job| job.id)
            .collect();

        // Remove jobs
        for job_id in jobs_to_purge {
            storage.jobs.remove(&job_id);
            purged_count += 1;

            // Clean up from queue structure
            for queue_jobs in storage.queues.values_mut() {
                for jobs_list in queue_jobs.values_mut() {
                    jobs_list.retain(|&id| id != job_id);
                }
            }
        }

        Ok(purged_count)
    }

    async fn purge_archived_jobs_in_queue(
        &self,
        queue_name: &str,
        older_than: DateTime<Utc>,
    ) -> Result<u64> {
        let mut storage = self.storage.write().await;
        // Like `purge_archived_jobs`, created_at stands in for archived_at.
        let jobs_to_purge: Vec<JobId> = storage
            .jobs
            .values()
            .filter(|job| {
                job.status == JobStatus::Archived
                    && job.queue_name == queue_name
                    && job.created_at < older_than
            })
            .map(|job| job.id)
            .collect();
        for job_id in &jobs_to_purge {
            storage.jobs.remove(job_id);
            if let Some(queue_jobs) = storage.queues.get_mut(queue_name) {
                for jobs_list in queue_jobs.values_mut() {
                    jobs_list.retain(|id| id != job_id);
                }
            }
        }
        Ok(jobs_to_purge.len() as u64)
    }

    async fn list_queue_names(&self) -> Result<Vec<String>> {
        let storage = self.storage.read().await;
        let mut names: Vec<String> = storage
            .jobs
            .values()
            .map(|job| job.queue_name.clone())
            .collect();
        names.sort();
        names.dedup();
        Ok(names)
    }

    async fn get_archival_stats(
        &self,
        queue_name: Option<&str>,
    ) -> Result<crate::archive::ArchivalStats> {
        let storage = self.storage.read().await;
        let now = storage.clock.now();

        let mut jobs_archived = 0u64;
        let mut bytes_archived = 0u64;

        for job in storage.jobs.values() {
            if job.status != JobStatus::Archived {
                continue;
            }

            // Filter by queue if specified
            if let Some(queue) = queue_name
                && job.queue_name != queue
            {
                continue;
            }

            jobs_archived += 1;
            bytes_archived += serde_json::to_string(&job.payload)
                .map(|s| s.len() as u64)
                .unwrap_or(0);
        }

        Ok(crate::archive::ArchivalStats {
            jobs_archived,
            jobs_purged: 0,
            bytes_archived,
            bytes_purged: 0,
            compression_ratio: 0.8, // Mock compression ratio
            last_run_at: now,
            operation_duration: std::time::Duration::from_millis(50),
        })
    }

    async fn pause_queue(&self, queue_name: &str, paused_by: Option<&str>) -> Result<()> {
        let mut storage = self.storage.write().await;
        let now = storage.clock.now();

        let pause_info = QueuePauseInfo {
            queue_name: queue_name.to_string(),
            paused_at: now,
            paused_by: paused_by.map(|s| s.to_string()),
            reason: None,
        };

        storage
            .paused_queues
            .insert(queue_name.to_string(), pause_info);
        Ok(())
    }

    async fn resume_queue(&self, queue_name: &str, _resumed_by: Option<&str>) -> Result<()> {
        let mut storage = self.storage.write().await;
        storage.paused_queues.remove(queue_name);
        Ok(())
    }

    async fn is_queue_paused(&self, queue_name: &str) -> Result<bool> {
        let storage = self.storage.read().await;
        Ok(storage.paused_queues.contains_key(queue_name))
    }

    async fn get_queue_pause_info(&self, queue_name: &str) -> Result<Option<QueuePauseInfo>> {
        let storage = self.storage.read().await;
        Ok(storage.paused_queues.get(queue_name).cloned())
    }

    async fn get_paused_queues(&self) -> Result<Vec<QueuePauseInfo>> {
        let storage = self.storage.read().await;
        Ok(storage.paused_queues.values().cloned().collect())
    }

    async fn heartbeat_job(&self, run: &Job, lease: std::time::Duration) -> Result<bool> {
        let mut storage = self.storage.write().await;
        // Only the run that was dequeued is extended, as on the database backends.
        let same_run = storage.jobs.get(&run.id).is_some_and(|job| {
            job.status == JobStatus::Running && lifecycle::is_same_run(job, run)
        });
        if !same_run {
            return Ok(false);
        }
        let now = storage.clock.now();
        storage
            .leases
            .insert(run.id, (now, lease_expiry(now, lease)));
        Ok(true)
    }

    async fn start_claimed_run(&self, run: &mut Job, lease: std::time::Duration) -> Result<bool> {
        let mut storage = self.storage.write().await;
        let now = storage.clock.now();
        let Some(job) = storage
            .jobs
            .get_mut(&run.id)
            .filter(|job| job.status == JobStatus::Running && lifecycle::is_same_run(job, run))
        else {
            return Ok(false);
        };
        job.started_at = Some(now);
        storage
            .leases
            .insert(run.id, (now, lease_expiry(now, lease)));
        run.started_at = Some(now);
        Ok(true)
    }

    async fn requeue_stale_jobs(
        &self,
        older_than: std::time::Duration,
    ) -> Result<super::StaleJobRecovery> {
        let mut storage = self.storage.write().await;
        let now = storage.clock.now();
        let cutoff = super::saturating_sub_from(now, older_than);

        let stale: Vec<(JobId, bool)> = storage
            .jobs
            .values()
            .filter(|job| job.status == JobStatus::Running)
            .filter_map(|job| {
                let started_at = job.started_at?;
                let is_stale = match storage.leases.get(&job.id) {
                    Some((heartbeat, expires)) if *heartbeat >= started_at => *expires < now,
                    _ => started_at < cutoff,
                };
                is_stale.then_some((job.id, job.attempts >= job.max_attempts))
            })
            .collect();

        let mut recovery = super::StaleJobRecovery::default();
        for (job_id, exhausted) in stale {
            storage.leases.remove(&job_id);
            if exhausted {
                storage.update_job_status(job_id, JobStatus::Dead)?;
                if let Some(job) = storage.jobs.get_mut(&job_id) {
                    job.failed_at = Some(now);
                    job.error_message = Some(super::STALE_JOB_ERROR_MESSAGE.to_string());
                }
                recovery.dead.push(job_id);
                // A terminal failure like any other, as on the database backends: a
                // recurring job is rescheduled for its next run, otherwise the failure
                // policies apply.
                let next_run_at = storage
                    .jobs
                    .get(&job_id)
                    .filter(|job| job.recurring)
                    .and_then(|job| lifecycle::next_cron_run(job, now));
                if let Some(next_run_at) = next_run_at {
                    storage.reschedule_in_place(job_id, next_run_at, true)?;
                    continue;
                }
                storage.apply_workflow_failure(job_id, now);
                storage.apply_batch_failure(job_id, now);
            } else {
                storage.update_job_status(job_id, JobStatus::Pending)?;
                if let Some(job) = storage.jobs.get_mut(&job_id) {
                    job.scheduled_at = now;
                    job.started_at = None;
                    job.error_message = Some(super::STALE_JOB_ERROR_MESSAGE.to_string());
                }
                recovery.requeued.push(job_id);
            }
        }

        Ok(recovery)
    }

    async fn purge_expired_encrypted_jobs(&self) -> Result<super::EncryptedJobPurge> {
        #[cfg_attr(not(feature = "encryption"), allow(unused_mut))]
        let mut purge = super::EncryptedJobPurge::default();
        #[cfg(feature = "encryption")]
        {
            let storage = self.storage.read().await;
            let now = storage.clock.now();
            let expired: Vec<(JobId, bool)> = storage
                .jobs
                .values()
                .filter(|job| job.is_encrypted)
                .filter(|job| {
                    matches!(
                        job.status,
                        JobStatus::Completed
                            | JobStatus::Failed
                            | JobStatus::Dead
                            | JobStatus::TimedOut
                            | JobStatus::Archived
                    )
                })
                .filter(|job| {
                    job.encrypted_payload
                        .as_ref()
                        .and_then(|encrypted| encrypted.metadata.delete_at)
                        .is_some_and(|delete_at| delete_at <= now)
                })
                .map(|job| (job.id, job.status == JobStatus::Archived))
                .collect();
            drop(storage);
            for (job_id, archived) in expired {
                self.delete_job(job_id).await?;
                if archived {
                    purge.archived_jobs += 1;
                } else {
                    purge.jobs += 1;
                }
            }
        }
        Ok(purge)
    }
}

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

    #[tokio::test]
    async fn dequeue_batch_claims_in_order_with_leases() {
        let queue = TestQueue::new();
        let low = queue
            .enqueue(Job::new("batch".to_string(), json!({})).as_low_priority())
            .await
            .unwrap();
        let high = queue
            .enqueue(Job::new("batch".to_string(), json!({})).as_high_priority())
            .await
            .unwrap();
        let normal = queue
            .enqueue(Job::new("batch".to_string(), json!({})))
            .await
            .unwrap();

        let claimed = queue.dequeue_batch("batch", 2).await.unwrap();
        let ids: Vec<JobId> = claimed.iter().map(|job| job.id).collect();
        assert_eq!(ids, [high, normal]);
        assert!(claimed.iter().all(|job| job.attempts == 1));
        assert_eq!(queue.get_job_count("batch", &JobStatus::Running).await, 2);
        // Leased: a reaper with any window leaves them alone.
        let recovery = queue
            .requeue_stale_jobs(std::time::Duration::ZERO)
            .await
            .unwrap();
        assert!(recovery.is_empty());

        // Paused queues hand out nothing.
        queue.pause_queue("batch", None).await.unwrap();
        assert!(queue.dequeue_batch("batch", 5).await.unwrap().is_empty());
        queue.resume_queue("batch", None).await.unwrap();
        let rest = queue.dequeue_batch("batch", 5).await.unwrap();
        assert_eq!(rest.len(), 1);
        assert_eq!(rest[0].id, low);
    }

    #[tokio::test]
    async fn release_job_run_returns_the_job_unclaimed() {
        let queue = TestQueue::new();
        let id = queue
            .enqueue(Job::new("release".to_string(), json!({})))
            .await
            .unwrap();
        let run = queue.dequeue_batch("release", 1).await.unwrap().remove(0);
        assert!(queue.release_job_run(&run).await.unwrap());
        let job = queue.get_job(id).await.unwrap().unwrap();
        assert_eq!(job.status, JobStatus::Pending);
        assert_eq!(job.attempts, 0);
        assert_eq!(job.started_at, None);
        assert!(!queue.release_job_run(&run).await.unwrap());
        assert_eq!(queue.dequeue_batch("release", 1).await.unwrap().len(), 1);
    }

    #[tokio::test]
    async fn test_basic_enqueue_dequeue() {
        let queue = TestQueue::new();

        // Enqueue a job
        let job = Job::new("test_queue".to_string(), json!({"test": true}));
        let job_id = queue.enqueue(job).await.unwrap();

        // Verify job count
        assert_eq!(
            queue.get_job_count("test_queue", &JobStatus::Pending).await,
            1
        );

        // Dequeue the job
        let dequeued = queue.dequeue("test_queue").await.unwrap();
        assert!(dequeued.is_some());
        assert_eq!(dequeued.unwrap().id, job_id);

        // Verify job is now running
        assert_eq!(
            queue.get_job_count("test_queue", &JobStatus::Running).await,
            1
        );
        assert_eq!(
            queue.get_job_count("test_queue", &JobStatus::Pending).await,
            0
        );
    }

    #[tokio::test]
    async fn test_paused_queue_is_not_dequeued() {
        let queue = TestQueue::new();
        let job_id = queue
            .enqueue(Job::new("paused".to_string(), json!({})))
            .await
            .unwrap();

        queue.pause_queue("paused", Some("test")).await.unwrap();
        assert!(queue.dequeue("paused").await.unwrap().is_none());
        assert!(
            queue
                .dequeue_with_priority_weights("paused", &PriorityWeights::new())
                .await
                .unwrap()
                .is_none()
        );

        queue.resume_queue("paused", None).await.unwrap();
        assert_eq!(queue.dequeue("paused").await.unwrap().unwrap().id, job_id);
    }

    #[tokio::test]
    async fn test_priority_dequeue() {
        let queue = TestQueue::new();

        // Enqueue jobs with different priorities
        let low_job =
            Job::new("test_queue".to_string(), json!({"priority": "low"})).as_low_priority();
        let high_job =
            Job::new("test_queue".to_string(), json!({"priority": "high"})).as_high_priority();
        let normal_job = Job::new("test_queue".to_string(), json!({"priority": "normal"}));

        let _low_id = queue.enqueue(low_job).await.unwrap();
        let high_id = queue.enqueue(high_job).await.unwrap();
        let _normal_id = queue.enqueue(normal_job).await.unwrap();

        // Dequeue should get the high priority job first
        let dequeued = queue.dequeue("test_queue").await.unwrap();
        assert_eq!(dequeued.unwrap().id, high_id);
    }

    #[tokio::test]
    async fn test_delayed_job() {
        let queue = TestQueue::new();
        let clock = queue.clock();

        // Enqueue a delayed job
        let delay = chrono::Duration::hours(1);
        let job = Job::with_delay("test_queue".to_string(), json!({"delayed": true}), delay);
        let job_id = queue.enqueue(job).await.unwrap();

        // Should not be able to dequeue immediately
        let dequeued = queue.dequeue("test_queue").await.unwrap();
        assert!(dequeued.is_none());

        // Advance time
        clock.advance(chrono::Duration::hours(2));

        // Now should be able to dequeue
        let dequeued = queue.dequeue("test_queue").await.unwrap();
        assert_eq!(dequeued.unwrap().id, job_id);
    }

    #[tokio::test]
    async fn test_job_completion() {
        let queue = TestQueue::new();

        // Enqueue and dequeue a job
        let job = Job::new("test_queue".to_string(), json!({"test": true}));
        let job_id = queue.enqueue(job).await.unwrap();
        let _dequeued = queue.dequeue("test_queue").await.unwrap();

        // Complete the job
        queue.complete_job(job_id).await.unwrap();

        // Verify job is completed
        let completed_job = queue.get_job(job_id).await.unwrap().unwrap();
        assert_eq!(completed_job.status, JobStatus::Completed);
        assert!(completed_job.completed_at.is_some());
    }

    #[tokio::test]
    async fn test_job_failure_and_retry() {
        let queue = TestQueue::new();

        // Enqueue a job with multiple retry attempts
        let job = Job::new("test_queue".to_string(), json!({"test": true})).with_max_attempts(3);
        let job_id = queue.enqueue(job).await.unwrap();

        // Dequeue and fail the job: a terminal failure, not retried automatically.
        let dequeued = queue.dequeue("test_queue").await.unwrap().unwrap();
        assert_eq!(dequeued.attempts, 1);
        queue.fail_job(job_id, "Test error").await.unwrap();

        let job = queue.get_job(job_id).await.unwrap().unwrap();
        assert_eq!(job.status, JobStatus::Failed);
        assert_eq!(job.attempts, 1);
        assert_eq!(job.error_message.as_deref(), Some("Test error"));
        assert!(job.failed_at.is_some());

        // Retry the job
        let retry_at = queue.clock().now() + chrono::Duration::minutes(5);
        queue.retry_job(job_id, retry_at).await.unwrap();

        // Job should be pending again
        let job = queue.get_job(job_id).await.unwrap().unwrap();
        assert_eq!(job.status, JobStatus::Pending);
        assert_eq!(job.scheduled_at, retry_at);
    }

    #[tokio::test]
    async fn test_batch_operations() {
        let queue = TestQueue::new();

        // Create a batch of jobs
        let jobs = vec![
            Job::new("batch_queue".to_string(), json!({"batch": 1})),
            Job::new("batch_queue".to_string(), json!({"batch": 2})),
            Job::new("batch_queue".to_string(), json!({"batch": 3})),
        ];

        let batch = JobBatch::new("test_batch".to_string()).with_jobs(jobs);
        let batch_id = queue.enqueue_batch(batch).await.unwrap();

        // Get batch status
        let status = queue.get_batch_status(batch_id).await.unwrap();
        assert_eq!(status.total_jobs, 3);
        assert_eq!(status.pending_jobs, 3);
        assert_eq!(status.status, BatchStatus::Pending);

        // Process all jobs
        for _ in 0..3 {
            if let Some(job) = queue.dequeue("batch_queue").await.unwrap() {
                queue.complete_job(job.id).await.unwrap();
            }
        }

        // Check batch is completed
        let status = queue.get_batch_status(batch_id).await.unwrap();
        assert_eq!(status.completed_jobs, 3);
        assert_eq!(status.status, BatchStatus::Completed);
    }

    #[tokio::test]
    async fn test_workflow_with_dependencies() {
        let queue = TestQueue::new();

        // Create jobs
        let job1 = Job::new("workflow_queue".to_string(), json!({"step": 1}));
        let job2 = Job::new("workflow_queue".to_string(), json!({"step": 2}));
        let job3 = Job::new("workflow_queue".to_string(), json!({"step": 3}));

        let job1_id = job1.id;
        let job2_id = job2.id;
        let job3_id = job3.id;

        // Create workflow with dependencies: job2 depends on job1, job3 depends on job2
        let mut dependencies = HashMap::new();
        dependencies.insert(job2_id, vec![job1_id]);
        dependencies.insert(job3_id, vec![job2_id]);

        let mut workflow = JobGroup::new("test_workflow".to_string());
        workflow.jobs = vec![job1, job2, job3];
        workflow.total_jobs = 3;
        workflow.dependencies = dependencies;

        let workflow_id = queue.enqueue_workflow(workflow).await.unwrap();

        // Only job1 should be available initially
        let dequeued = queue.dequeue("workflow_queue").await.unwrap();
        assert_eq!(dequeued.unwrap().id, job1_id);

        // Job2 and job3 should not be available
        assert!(queue.dequeue("workflow_queue").await.unwrap().is_none());

        // Complete job1
        queue.complete_job(job1_id).await.unwrap();

        // Now job2 should be available
        let dequeued = queue.dequeue("workflow_queue").await.unwrap();
        assert_eq!(dequeued.unwrap().id, job2_id);

        // Complete job2
        queue.complete_job(job2_id).await.unwrap();

        // Now job3 should be available
        let dequeued = queue.dequeue("workflow_queue").await.unwrap();
        assert_eq!(dequeued.unwrap().id, job3_id);

        // Complete job3
        queue.complete_job(job3_id).await.unwrap();

        // Check workflow status
        let workflow_status = queue
            .get_workflow_status(workflow_id)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(workflow_status.status, WorkflowStatus::Completed);
        assert_eq!(workflow_status.completed_jobs, 3);
    }

    #[tokio::test]
    async fn test_cron_job() {
        let queue = TestQueue::new();
        let clock = queue.clock();

        // Create a cron job that runs every hour
        let mut job = Job::new("cron_queue".to_string(), json!({"cron": true}));
        job.cron_schedule = Some("0 0 * * * *".to_string());

        let job_id = queue.enqueue_cron_job(job).await.unwrap();

        // Get the job
        let job = queue.get_job(job_id).await.unwrap().unwrap();
        assert!(job.recurring);
        assert!(job.next_run_at.is_some());

        // Advance clock to make the cron job ready for execution
        // Since the cron is "0 0 * * * *" (every hour at minute 0),
        // advance to the next hour boundary
        clock.advance(chrono::Duration::hours(1));

        // Now the job should be available for dequeue
        let run = queue.dequeue("cron_queue").await.unwrap().unwrap();
        let recorded = queue
            .finish_job_run(&run, crate::queue::JobOutcome::Completed)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(recorded.status, JobStatus::Pending);

        // The same job is rescheduled for its next run
        let job = queue.get_job(job_id).await.unwrap().unwrap();
        assert_eq!(job.status, JobStatus::Pending);
        assert_eq!(job.next_run_at, recorded.next_run_at);
        assert_eq!(Some(job.scheduled_at), job.next_run_at);
        assert_eq!(job.attempts, 0);

        // Advance time to next run
        clock.advance(chrono::Duration::hours(2));

        // Get due cron jobs
        let due_jobs = queue.get_due_cron_jobs(Some("cron_queue")).await.unwrap();
        assert_eq!(due_jobs.len(), 1);
        assert_eq!(due_jobs[0].id, job_id);

        // Reschedule for next run
        let next_run = clock.now() + chrono::Duration::hours(1);
        queue.reschedule_cron_job(job_id, next_run).await.unwrap();

        // Still one job, rescheduled in place
        assert_eq!(
            queue.get_job_count("cron_queue", &JobStatus::Pending).await,
            1
        );
        let job = queue.get_job(job_id).await.unwrap().unwrap();
        assert_eq!(job.scheduled_at, next_run);
    }

    #[tokio::test]
    async fn test_job_result_storage() {
        let queue = TestQueue::new();
        let clock = queue.clock();

        // Create and enqueue a job
        let job = Job::new("result_queue".to_string(), json!({"test": true}));
        let job_id = queue.enqueue(job).await.unwrap();

        // Store a result
        let result = json!({"success": true, "data": "test result"});
        let expires_at = clock.now() + chrono::Duration::hours(1);
        queue
            .store_job_result(job_id, result.clone(), Some(expires_at))
            .await
            .unwrap();

        // Retrieve the result
        let retrieved = queue.get_job_result(job_id).await.unwrap();
        assert_eq!(retrieved, Some(result));

        // Advance time past expiration
        clock.advance(chrono::Duration::hours(2));

        // Result should be expired
        let retrieved = queue.get_job_result(job_id).await.unwrap();
        assert_eq!(retrieved, None);

        // Cleanup expired results
        let cleaned = queue.cleanup_expired_results().await.unwrap();
        assert_eq!(cleaned, 1);
    }

    #[tokio::test]
    async fn test_memory_result_storage_follows_ttl_and_delete() {
        use crate::job::ResultStorage;

        let queue = TestQueue::new();
        let clock = queue.clock();

        let job = Job::new("memory_results".to_string(), json!({}))
            .with_result_storage(ResultStorage::Memory)
            .with_result_ttl(std::time::Duration::from_secs(60));
        let ttl = job.result_config.ttl.unwrap();
        let job_id = queue.enqueue(job).await.unwrap();
        queue.dequeue("memory_results").await.unwrap().unwrap();

        // What a worker does: store the result with its TTL, then complete the job.
        let expires_at = crate::queue::saturating_add_to(clock.now(), ttl);
        queue
            .store_job_result(job_id, json!({"n": 1}), Some(expires_at))
            .await
            .unwrap();
        queue.complete_job(job_id).await.unwrap();

        assert_eq!(
            queue.get_job_result(job_id).await.unwrap(),
            Some(json!({"n": 1}))
        );
        clock.advance(chrono::Duration::seconds(61));
        assert_eq!(queue.get_job_result(job_id).await.unwrap(), None);

        // Deleting a job deletes its result.
        queue
            .store_job_result(job_id, json!({"n": 2}), None)
            .await
            .unwrap();
        queue.delete_job(job_id).await.unwrap();
        assert_eq!(queue.get_job_result(job_id).await.unwrap(), None);
        assert_eq!(queue.cleanup_expired_results().await.unwrap(), 0);
    }

    #[tokio::test]
    async fn test_queue_stats() {
        let queue = TestQueue::new();

        // Enqueue various jobs
        for i in 0..5 {
            let job = Job::new("stats_queue".to_string(), json!({"index": i}));
            queue.enqueue(job).await.unwrap();
        }

        // Process some jobs
        for _ in 0..2 {
            if let Some(job) = queue.dequeue("stats_queue").await.unwrap() {
                queue.complete_job(job.id).await.unwrap();
            }
        }

        // One job runs out of attempts and dies, as a worker would record it.
        if let Some(job) = queue.dequeue("stats_queue").await.unwrap() {
            queue.mark_job_dead(job.id, "Test failure").await.unwrap();
        }

        // Get stats
        let stats = queue.get_queue_stats("stats_queue").await.unwrap();

        assert_eq!(stats.pending_count, 2);
        assert_eq!(stats.statistics.completed, 2);
        assert_eq!(stats.statistics.dead, 1); // One job marked as dead
        assert_eq!(stats.statistics.total_processed, 3); // 2 completed + 1 dead
    }

    #[tokio::test]
    async fn test_requeue_stale_jobs_uses_leases_and_fallback() {
        let queue = TestQueue::new();
        let clock = queue.clock();
        let lease = std::time::Duration::from_secs(30);

        let leased = queue
            .enqueue(Job::new("stale".to_string(), json!({"n": 1})))
            .await
            .unwrap();
        let unleased = queue
            .enqueue(Job::new("stale".to_string(), json!({"n": 2})))
            .await
            .unwrap();
        let exhausted = queue
            .enqueue(Job::new("stale".to_string(), json!({"n": 3})).with_max_attempts(0))
            .await
            .unwrap();
        let mut runs = HashMap::new();
        for _ in 0..3 {
            let run = queue
                .dequeue_leased("stale", None, lease)
                .await
                .unwrap()
                .unwrap();
            runs.insert(run.id, run);
        }
        // `unleased` stands for a job claimed by a version that wrote no lease.
        queue.storage.write().await.leases.remove(&unleased);

        // Nothing is stale yet.
        let recovery = queue
            .requeue_stale_jobs(std::time::Duration::from_secs(600))
            .await
            .unwrap();
        assert!(recovery.is_empty());

        // A short staleness window does not reclaim leased jobs.
        clock.advance(chrono::Duration::seconds(10));
        assert!(queue.heartbeat_job(&runs[&leased], lease).await.unwrap());
        let recovery = queue
            .requeue_stale_jobs(std::time::Duration::ZERO)
            .await
            .unwrap();
        assert_eq!(recovery.requeued, vec![unleased]);

        // After 60s the leases have expired but the 600s fallback has not.
        clock.advance(chrono::Duration::seconds(60));
        let recovery = queue
            .requeue_stale_jobs(std::time::Duration::from_secs(600))
            .await
            .unwrap();
        assert_eq!(recovery.requeued, vec![leased]);
        assert_eq!(recovery.dead, vec![exhausted]);

        let job = queue.get_job(leased).await.unwrap().unwrap();
        assert_eq!(job.status, JobStatus::Pending);
        assert_eq!(
            queue.get_job(exhausted).await.unwrap().unwrap().status,
            JobStatus::Dead
        );

        // A heartbeat for a job that is no longer Running reports a lost lease.
        assert!(!queue.heartbeat_job(&runs[&exhausted], lease).await.unwrap());
    }

    #[tokio::test]
    async fn test_stale_heartbeat_does_not_extend_a_new_run() {
        let queue = TestQueue::new();
        let clock = queue.clock();
        let lease = std::time::Duration::from_secs(30);
        let id = queue
            .enqueue(Job::new("rerun".to_string(), json!({})))
            .await
            .unwrap();

        let first = queue
            .dequeue_leased("rerun", None, lease)
            .await
            .unwrap()
            .unwrap();
        clock.advance(chrono::Duration::seconds(60));
        let recovery = queue
            .requeue_stale_jobs(std::time::Duration::from_secs(3600))
            .await
            .unwrap();
        assert_eq!(recovery.requeued, vec![id]);

        let second = queue
            .dequeue_leased("rerun", None, lease)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(second.attempts, 2);
        // The first run's worker is still alive and heartbeats: it lost its lease.
        assert!(!queue.heartbeat_job(&first, lease).await.unwrap());
        assert!(queue.heartbeat_job(&second, lease).await.unwrap());

        // Its heartbeats did not keep the second run alive.
        clock.advance(chrono::Duration::seconds(60));
        assert!(!queue.heartbeat_job(&first, lease).await.unwrap());
        let recovery = queue
            .requeue_stale_jobs(std::time::Duration::from_secs(3600))
            .await
            .unwrap();
        assert_eq!(recovery.requeued, vec![id]);
    }

    /// #94: starting a claimed run that waited moves its start time to now; the run is
    /// identified by the new start time from then on.
    #[tokio::test]
    async fn start_claimed_run_records_the_actual_start() {
        let queue = TestQueue::new();
        let clock = queue.clock();
        let lease = std::time::Duration::from_secs(30);
        let id = queue
            .enqueue(Job::new("held".to_string(), json!({})))
            .await
            .unwrap();
        let claimed = queue
            .dequeue_leased("held", None, lease)
            .await
            .unwrap()
            .unwrap();
        clock.advance(chrono::Duration::seconds(20));

        let mut run = claimed.clone();
        assert!(queue.start_claimed_run(&mut run, lease).await.unwrap());
        let started = run.started_at.unwrap();
        assert_eq!(
            started - claimed.started_at.unwrap(),
            chrono::Duration::seconds(20)
        );
        assert_eq!(
            queue.get_job(id).await.unwrap().unwrap().started_at,
            Some(started)
        );
        // The lease was renewed from the new start.
        clock.advance(chrono::Duration::seconds(20));
        let recovery = queue
            .requeue_stale_jobs(std::time::Duration::from_secs(3600))
            .await
            .unwrap();
        assert!(recovery.requeued.is_empty());

        // The claim as first dequeued no longer identifies the run.
        let mut stale = claimed.clone();
        assert!(!queue.start_claimed_run(&mut stale, lease).await.unwrap());
        assert_eq!(stale.started_at, claimed.started_at);
        let recorded = queue
            .finish_job_run(&run, crate::queue::JobOutcome::Completed)
            .await
            .unwrap();
        assert!(recorded.is_some());
    }

    #[tokio::test]
    async fn test_zero_lease_never_expires() {
        let queue = TestQueue::new();
        let id = queue
            .enqueue(Job::new("forever".to_string(), json!({})))
            .await
            .unwrap();
        let run = queue
            .dequeue_leased("forever", None, std::time::Duration::ZERO)
            .await
            .unwrap()
            .unwrap();
        queue.clock().advance(chrono::Duration::days(365));
        let recovery = queue
            .requeue_stale_jobs(std::time::Duration::ZERO)
            .await
            .unwrap();
        assert!(recovery.is_empty(), "{recovery:?}");
        assert!(
            queue
                .heartbeat_job(&run, std::time::Duration::ZERO)
                .await
                .unwrap()
        );
        assert_eq!(
            queue.get_job(id).await.unwrap().unwrap().status,
            JobStatus::Running
        );
    }

    #[tokio::test]
    async fn test_mark_job_dead_and_timed_out_apply_the_failure_policy() {
        use crate::workflow::{FailurePolicy, JobGroup};

        for timed_out in [false, true] {
            let queue = TestQueue::new();
            let a = Job::new("policy".to_string(), json!({"step": "a"}));
            let b = Job::new("policy".to_string(), json!({"step": "b"})).depends_on(&a.id);
            let other = Job::new("policy".to_string(), json!({"step": "other"}));
            let (a_id, b_id, other_id) = (a.id, b.id, other.id);
            queue
                .enqueue_workflow(
                    JobGroup::new("policy")
                        .add_job(a)
                        .add_job(b)
                        .add_job(other)
                        .with_failure_policy(FailurePolicy::FailFast),
                )
                .await
                .unwrap();
            let run = queue.dequeue("policy").await.unwrap().unwrap();
            let failed_id = run.id;
            if timed_out {
                queue
                    .mark_job_timed_out(failed_id, "too slow")
                    .await
                    .unwrap();
            } else {
                queue.mark_job_dead(failed_id, "broken").await.unwrap();
            }
            // Fail-fast: every other unfinished job of the workflow fails with it.
            for id in [a_id, b_id, other_id] {
                if id == failed_id {
                    continue;
                }
                let job = queue.get_job(id).await.unwrap().unwrap();
                assert_eq!(job.status, JobStatus::Failed, "{job:?}");
            }
            // Already applied, as on the database backends.
            assert!(
                queue
                    .fail_job_dependencies(failed_id)
                    .await
                    .unwrap()
                    .is_empty()
            );
        }
    }

    #[tokio::test]
    async fn test_finish_job_run_reports_the_jobs_failed_with_it() {
        use crate::queue::JobOutcome;
        use crate::workflow::{FailurePolicy, JobGroup};

        let queue = TestQueue::new();
        let a = Job::new("finish_policy".to_string(), json!({"step": "a"}));
        let b = Job::new("finish_policy".to_string(), json!({"step": "b"})).depends_on(&a.id);
        let (a_id, b_id) = (a.id, b.id);
        queue
            .enqueue_workflow(
                JobGroup::new("finish_policy")
                    .add_job(a)
                    .add_job(b)
                    .with_failure_policy(FailurePolicy::ContinueOnFailure),
            )
            .await
            .unwrap();
        let run = queue.dequeue("finish_policy").await.unwrap().unwrap();
        assert_eq!(run.id, a_id);
        let recorded = queue
            .finish_job_run(
                &run,
                JobOutcome::Dead {
                    error: "broken".to_string(),
                },
            )
            .await
            .unwrap()
            .expect("the run is current");
        assert_eq!(recorded.status, JobStatus::Dead);
        assert_eq!(recorded.cancelled, vec![b_id]);
        // A second report for the same run is discarded.
        assert!(
            queue
                .finish_job_run(&run, crate::queue::JobOutcome::Completed)
                .await
                .unwrap()
                .is_none()
        );
    }

    #[cfg(feature = "encryption")]
    mod encryption {
        use super::*;
        use crate::archive::{ArchivalConfig, ArchivalPolicy, ArchivalReason};
        use crate::encryption::{
            EncryptionAlgorithm, EncryptionConfig, EncryptionEngine, KeySource, RetentionPolicy,
        };

        const KEY: &str = "QUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUE=";

        fn config() -> EncryptionConfig {
            EncryptionConfig::new(EncryptionAlgorithm::AES256GCM)
                .with_key_id("k1")
                .with_key_source(KeySource::Static(KEY.to_string()))
        }

        async fn encrypted_queue() -> TestQueue {
            TestQueue::new().with_encryption(EncryptionEngine::new(config()).await.unwrap())
        }

        #[tokio::test]
        async fn jobs_are_stored_sealed_and_decrypt() {
            let queue = encrypted_queue().await;
            let payload = json!({"ssn": "123-45-6789", "amount": 5});
            let id = queue
                .enqueue(
                    Job::new("q".into(), payload.clone())
                        .with_encryption(config())
                        .with_pii_fields(vec!["ssn"]),
                )
                .await
                .unwrap();

            let stored = queue.get_job(id).await.unwrap().unwrap();
            assert!(stored.is_encrypted);
            assert_eq!(stored.payload, json!({"ssn": "[ENCRYPTED]", "amount": 5}));
            assert!(stored.encrypted_payload.is_some());

            let dequeued = queue.dequeue("q").await.unwrap().unwrap();
            assert_eq!(
                dequeued.payload, stored.payload,
                "dequeue returns the stored form"
            );
            let opened = queue.decrypt_job(dequeued).await.unwrap();
            assert!(!opened.is_encrypted);
            assert_eq!(opened.payload, payload);

            // Unencrypted jobs pass through
            let plain = queue
                .enqueue(Job::new("q".into(), json!({"a": 1})))
                .await
                .unwrap();
            let plain = queue.get_job(plain).await.unwrap().unwrap();
            assert_eq!(
                queue.decrypt_job(plain).await.unwrap().payload,
                json!({"a": 1})
            );
        }

        #[tokio::test]
        async fn batch_and_workflow_are_sealed() {
            let queue = encrypted_queue().await;
            let job = || Job::new("q".into(), json!({"secret": 1})).with_encryption(config());
            let batch = JobBatch::new("b").with_jobs(vec![job(), job()]);
            let batch_id = queue.enqueue_batch(batch).await.unwrap();
            for stored in queue.get_batch_jobs(batch_id).await.unwrap() {
                assert!(stored.is_encrypted);
                assert_eq!(stored.payload, json!({"encrypted": true}));
            }

            let workflow = JobGroup::new("w").add_job(job());
            let workflow_id = queue.enqueue_workflow(workflow).await.unwrap();
            for stored in queue.get_workflow_jobs(workflow_id).await.unwrap() {
                assert!(stored.is_encrypted);
                assert_eq!(
                    queue.decrypt_job(stored).await.unwrap().payload,
                    json!({"secret": 1})
                );
            }
        }

        #[tokio::test]
        async fn fails_closed_without_an_engine() {
            let queue = TestQueue::new();
            let err = queue
                .enqueue(Job::new("q".into(), json!({"secret": 1})).with_encryption(config()))
                .await
                .unwrap_err();
            assert!(matches!(err, HammerworkError::Encryption { .. }), "{err}");
            assert!(queue.get_all_jobs().await.is_empty(), "nothing is stored");

            // A sealed job cannot be opened by a queue without the engine
            let sealed = encrypted_queue().await;
            let id = sealed
                .enqueue(Job::new("q".into(), json!({"secret": 1})).with_encryption(config()))
                .await
                .unwrap();
            let stored = sealed.get_job(id).await.unwrap().unwrap();
            assert!(queue.decrypt_job(stored).await.is_err());
        }

        #[tokio::test]
        async fn tampering_is_detected() {
            let queue = encrypted_queue().await;
            let a = queue
                .enqueue(Job::new("q".into(), json!({"who": "a"})).with_encryption(config()))
                .await
                .unwrap();
            let b = queue
                .enqueue(Job::new("q".into(), json!({"who": "b"})).with_encryption(config()))
                .await
                .unwrap();
            let mut job_a = queue.get_job(a).await.unwrap().unwrap();
            let job_b = queue.get_job(b).await.unwrap().unwrap();
            job_a.encrypted_payload = job_b.encrypted_payload.clone();
            assert!(queue.decrypt_job(job_a).await.is_err());

            let mut moved = job_b;
            moved.queue_name = "other".into();
            assert!(queue.decrypt_job(moved).await.is_err());
        }

        #[tokio::test]
        async fn encrypted_queues_encrypt_without_a_job_config() {
            let queue = encrypted_queue().await.with_encrypted_queues(["secure"]);
            let id = queue
                .enqueue(Job::new("secure".into(), json!({"secret": 1})))
                .await
                .unwrap();
            let stored = queue.get_job(id).await.unwrap().unwrap();
            assert!(stored.is_encrypted);
            assert_eq!(
                queue.decrypt_job(stored).await.unwrap().payload,
                json!({"secret": 1})
            );

            let other = queue
                .enqueue(Job::new("open".into(), json!({"visible": 1})))
                .await
                .unwrap();
            assert!(!queue.get_job(other).await.unwrap().unwrap().is_encrypted);

            let all = encrypted_queue().await.with_encrypted_queues(["*"]);
            let id = all
                .enqueue(Job::new("any".into(), json!({"ssn": "1"})).with_pii_fields(vec!["ssn"]))
                .await
                .unwrap();
            let stored = all.get_job(id).await.unwrap().unwrap();
            assert_eq!(stored.payload, json!({"ssn": "[ENCRYPTED]"}));
        }

        #[tokio::test]
        async fn archived_jobs_must_be_restored_to_decrypt() {
            let queue = encrypted_queue().await;
            let id = queue
                .enqueue(Job::new("q".into(), json!({"secret": 1})).with_encryption(config()))
                .await
                .unwrap();
            queue.dequeue("q").await.unwrap();
            queue.complete_job(id).await.unwrap();
            queue.clock().advance(chrono::Duration::seconds(10));
            let stats = queue
                .archive_jobs(
                    Some("q"),
                    &ArchivalPolicy::new()
                        .archive_completed_after(chrono::Duration::seconds(0))
                        .enabled(true),
                    &ArchivalConfig::new(),
                    ArchivalReason::Manual,
                    None,
                )
                .await
                .unwrap();
            assert_eq!(stats.jobs_archived, 1);

            let archived = queue.get_job(id).await.unwrap().unwrap();
            assert_eq!(archived.status, JobStatus::Archived);
            assert!(archived.is_encrypted);
            let err = queue.decrypt_job(archived).await.unwrap_err();
            assert!(err.to_string().contains("restore_archived_job"), "{err}");

            queue.restore_archived_job(id).await.unwrap();
            let restored = queue.get_job(id).await.unwrap().unwrap();
            assert_eq!(
                queue.decrypt_job(restored).await.unwrap().payload,
                json!({"secret": 1})
            );
        }

        /// Every read path returns an encrypted job with its ciphertext and PII fields, so
        /// `decrypt_job` works on it (#64 H2).
        #[tokio::test]
        async fn dead_and_listed_jobs_keep_their_ciphertext() {
            let queue = encrypted_queue().await;
            let payload = json!({"ssn": "123-45-6789", "amount": 5});
            let id = queue
                .enqueue(
                    Job::new("dlq".into(), payload.clone())
                        .with_encryption(config())
                        .with_pii_fields(vec!["ssn"]),
                )
                .await
                .unwrap();
            queue.mark_job_dead(id, "boom").await.unwrap();

            let dead = queue.get_dead_jobs(None, None).await.unwrap();
            let by_queue = queue
                .get_dead_jobs_by_queue("dlq", None, None)
                .await
                .unwrap();
            for job in dead.into_iter().chain(by_queue) {
                assert_eq!(job.id, id);
                assert!(job.is_encrypted);
                assert!(job.encrypted_payload.is_some());
                assert_eq!(job.pii_fields, vec!["ssn"]);
                assert_eq!(queue.decrypt_job(job).await.unwrap().payload, payload);
            }
        }

        #[tokio::test]
        async fn retention_purge_uses_the_mock_clock() {
            let queue = encrypted_queue().await;
            let enqueue = |policy: RetentionPolicy| {
                let queue = queue.clone();
                async move {
                    queue
                        .enqueue(
                            Job::new("q".into(), json!({"secret": 1}))
                                .with_encryption(config())
                                .with_retention_policy(policy),
                        )
                        .await
                        .unwrap()
                }
            };
            let hour = RetentionPolicy::DeleteAfter(std::time::Duration::from_secs(3600));
            let done = enqueue(hour.clone()).await;
            let pending = enqueue(hour).await;
            let kept = enqueue(RetentionPolicy::KeepIndefinitely).await;
            for id in [done, kept] {
                queue.complete_job(id).await.unwrap();
            }

            assert_eq!(
                queue.purge_expired_encrypted_jobs().await.unwrap().total(),
                0,
                "nothing has expired yet"
            );
            queue.clock().advance(chrono::Duration::hours(2));
            let purge = queue.purge_expired_encrypted_jobs().await.unwrap();
            assert_eq!(purge.jobs, 1);
            assert!(queue.get_job(done).await.unwrap().is_none());
            assert!(
                queue.get_job(pending).await.unwrap().is_some(),
                "not finished"
            );
            assert!(queue.get_job(kept).await.unwrap().is_some());
        }
    }

    /// Like the database backends, a job enqueued with `depends_on` waits until its
    /// dependencies complete; one whose dependencies already completed runs at once and
    /// one with a failed dependency is enqueued as failed.
    #[tokio::test]
    async fn dependent_jobs_wait_for_their_dependencies() {
        use crate::queue::JobOutcome;
        let queue = TestQueue::new();
        let parent = queue
            .enqueue(Job::new("deps".to_string(), json!({"n": 1})))
            .await
            .unwrap();
        let child = queue
            .enqueue(Job::new("deps".to_string(), json!({"n": 2})).depends_on(&parent))
            .await
            .unwrap();

        let run = queue.dequeue("deps").await.unwrap().unwrap();
        assert_eq!(run.id, parent);
        assert!(
            queue.dequeue("deps").await.unwrap().is_none(),
            "a job ran before its dependency finished"
        );

        let recorded = queue
            .finish_job_run(&run, JobOutcome::Completed)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(recorded.unblocked, vec![child]);
        let run = queue.dequeue("deps").await.unwrap().unwrap();
        assert_eq!(run.id, child);
        assert_eq!(run.dependency_status, DependencyStatus::Satisfied);

        // Enqueued after its dependency completed: runnable at once.
        let late = queue
            .enqueue(Job::new("deps".to_string(), json!({"n": 3})).depends_on(&parent))
            .await
            .unwrap();
        assert_eq!(queue.dequeue("deps").await.unwrap().unwrap().id, late);

        // Enqueued after its dependency failed: failed at once.
        let failed = queue
            .enqueue(Job::new("deps".to_string(), json!({"n": 4})))
            .await
            .unwrap();
        queue.fail_job(failed, "boom").await.unwrap();
        let orphan = queue
            .enqueue(Job::new("deps".to_string(), json!({"n": 5})).depends_on(&failed))
            .await
            .unwrap();
        let orphan = queue.get_job(orphan).await.unwrap().unwrap();
        assert_eq!(orphan.status, JobStatus::Failed);
        assert_eq!(orphan.dependency_status, DependencyStatus::Failed);
    }

    /// A retried run records its error and, if it timed out, when, as on the databases.
    #[tokio::test]
    async fn retried_run_records_its_error_and_timeout() {
        use crate::queue::JobOutcome;
        let queue = TestQueue::new();
        let id = queue
            .enqueue(Job::new("retry_err".to_string(), json!({})))
            .await
            .unwrap();
        let run = queue.dequeue("retry_err").await.unwrap().unwrap();
        let retry_at = Utc::now() + chrono::Duration::seconds(30);
        let recorded = queue
            .finish_job_run(
                &run,
                JobOutcome::Retry {
                    retry_at,
                    error: "boom".to_string(),
                    timed_out: true,
                },
            )
            .await
            .unwrap()
            .unwrap();
        assert_eq!(recorded.status, JobStatus::Pending);
        let job = queue.get_job(id).await.unwrap().unwrap();
        assert_eq!(job.status, JobStatus::Pending);
        // The mock clock has not moved, so the due time is on it (give or take the
        // wall-clock time the test itself took).
        let skew = (job.scheduled_at - retry_at).num_seconds().abs();
        assert!(skew <= 2, "scheduled_at {skew}s off");
        assert_eq!(job.error_message.as_deref(), Some("boom"));
        assert!(job.timed_out_at.is_some());
    }

    /// A worker computes `retry_at` from the wall clock; the retry is due that long
    /// after the mock clock's now, not after the wall clock's.
    #[tokio::test]
    async fn retry_outcome_is_moved_onto_the_mock_clock() {
        use crate::queue::JobOutcome;
        let queue = TestQueue::new();
        let id = queue
            .enqueue(Job::new("retry_clock".to_string(), json!({})))
            .await
            .unwrap();
        let run = queue.dequeue("retry_clock").await.unwrap().unwrap();
        queue.clock().advance(chrono::Duration::hours(2));
        let mock_now = queue.clock().now();
        queue
            .finish_job_run(
                &run,
                JobOutcome::Retry {
                    retry_at: Utc::now() + chrono::Duration::seconds(30),
                    error: "boom".to_string(),
                    timed_out: false,
                },
            )
            .await
            .unwrap()
            .unwrap();
        let job = queue.get_job(id).await.unwrap().unwrap();
        let after = (job.scheduled_at - mock_now).num_seconds();
        assert!((28..=30).contains(&after), "due {after}s after mock now");
        // Not yet due on the mock clock, due once it passes.
        assert!(queue.dequeue("retry_clock").await.unwrap().is_none());
        queue.clock().advance(chrono::Duration::seconds(31));
        assert!(queue.dequeue("retry_clock").await.unwrap().is_some());
    }

    /// Unknown ids and queues fail (or succeed) as on the database backends.
    #[tokio::test]
    async fn unknown_ids_and_queues_match_the_database_backends() {
        let queue = TestQueue::new();
        let id = uuid::Uuid::new_v4();
        assert!(matches!(
            queue.get_batch_status(id).await,
            Err(HammerworkError::JobNotFound { .. })
        ));
        assert!(matches!(
            queue.restore_archived_job(id).await,
            Err(HammerworkError::JobNotFound { .. })
        ));
        let live = queue
            .enqueue(Job::new("live_q".to_string(), json!({})))
            .await
            .unwrap();
        assert!(matches!(
            queue.restore_archived_job(live).await,
            Err(HammerworkError::JobNotFound { .. })
        ));
        let stats = queue.get_queue_stats("no_such_queue").await.unwrap();
        assert_eq!(stats.queue_name, "no_such_queue");
        assert_eq!(stats.pending_count, 0);
        assert_eq!(stats.statistics.total_processed, 0);
    }

    /// Completing a workflow job's run reports the dependents it released.
    #[tokio::test]
    async fn completed_run_reports_unblocked_workflow_jobs() {
        use crate::queue::JobOutcome;
        let queue = TestQueue::new();
        let first = Job::new("wf_unblock".to_string(), json!({"step": 1}));
        let second = Job::new("wf_unblock".to_string(), json!({"step": 2}));
        let (first_id, second_id) = (first.id, second.id);
        let workflow = JobGroup::new("unblock").add_job(first).then(second);
        queue.enqueue_workflow(workflow).await.unwrap();

        let run = queue.dequeue("wf_unblock").await.unwrap().unwrap();
        assert_eq!(run.id, first_id);
        let recorded = queue
            .finish_job_run(&run, JobOutcome::Completed)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(recorded.unblocked, vec![second_id]);
        assert_eq!(
            queue.dequeue("wf_unblock").await.unwrap().unwrap().id,
            second_id
        );
    }

    /// A stale recurring job out of attempts is rescheduled for its next occurrence,
    /// like the database reapers do, instead of staying dead.
    #[tokio::test]
    async fn stale_recurring_job_out_of_attempts_is_rescheduled() {
        let queue = TestQueue::new();
        let clock = queue.clock();
        // Half past the hour, so the run (an hour later) and the reaping five minutes
        // after it stay before the next hourly slot whatever the wall-clock time is.
        clock.set_time(
            chrono::DateTime::parse_from_rfc3339("2026-01-05T10:30:00Z")
                .unwrap()
                .with_timezone(&Utc),
        );
        let mut job = Job::new("stale_cron".to_string(), json!({})).with_max_attempts(1);
        job.cron_schedule = Some("0 0 * * * *".to_string());
        let id = queue.enqueue_cron_job(job).await.unwrap();

        clock.advance(chrono::Duration::hours(1));
        let lease = std::time::Duration::from_secs(30);
        let run = queue
            .dequeue_leased("stale_cron", None, lease)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(run.id, id);

        clock.advance(chrono::Duration::minutes(5));
        let recovery = queue
            .requeue_stale_jobs(std::time::Duration::from_secs(60))
            .await
            .unwrap();
        assert_eq!(recovery.dead, vec![id]);

        let job = queue.get_job(id).await.unwrap().unwrap();
        assert_eq!(job.status, JobStatus::Pending, "the schedule was killed");
        assert_eq!(job.attempts, 0);
        assert!(job.next_run_at.is_some_and(|next| next > clock.now()));
        assert_eq!(Some(job.scheduled_at), job.next_run_at);
    }

    /// `QueueStats::statistics.failed` counts `Failed` jobs only, as on the database
    /// backends; dead jobs are in `dead` (counting them in both made the dashboard count
    /// them twice).
    #[tokio::test]
    async fn queue_stats_failed_counts_failed_jobs_only() {
        let queue = TestQueue::new();
        let failed = queue
            .enqueue(Job::new("stats_q".to_string(), json!({"n": 1})))
            .await
            .unwrap();
        queue.fail_job(failed, "nope").await.unwrap();
        let dead = queue
            .enqueue(Job::new("stats_q".to_string(), json!({"n": 2})))
            .await
            .unwrap();
        queue.mark_job_dead(dead, "broken").await.unwrap();

        let stats = queue.get_queue_stats("stats_q").await.unwrap();
        assert_eq!(stats.statistics.failed, 1);
        assert_eq!(stats.statistics.dead, 1);
        assert_eq!(stats.dead_count, 1);
    }

    /// Only `Dead` jobs are dead jobs: timed-out jobs are not listed, counted or purged
    /// as dead, and the purge cut-off is exclusive, as on the database backends.
    #[tokio::test]
    async fn timed_out_jobs_are_not_dead_jobs() {
        let queue = TestQueue::new();
        let timed_out = queue
            .enqueue(Job::new("dead_q".to_string(), json!({"n": 1})))
            .await
            .unwrap();
        queue.dequeue("dead_q").await.unwrap().unwrap();
        queue.mark_job_timed_out(timed_out, "slow").await.unwrap();
        let dead = queue
            .enqueue(Job::new("dead_q".to_string(), json!({"n": 2})))
            .await
            .unwrap();
        queue.mark_job_dead(dead, "broken").await.unwrap();

        let listed: Vec<JobId> = queue
            .get_dead_jobs(None, None)
            .await
            .unwrap()
            .iter()
            .map(|job| job.id)
            .collect();
        assert_eq!(listed, vec![dead]);
        let by_queue: Vec<JobId> = queue
            .get_dead_jobs_by_queue("dead_q", None, None)
            .await
            .unwrap()
            .iter()
            .map(|job| job.id)
            .collect();
        assert_eq!(by_queue, vec![dead]);
        let summary = queue.get_dead_job_summary().await.unwrap();
        assert_eq!(summary.total_dead_jobs, 1);

        let failed_at = queue
            .get_job(dead)
            .await
            .unwrap()
            .unwrap()
            .failed_at
            .unwrap();
        assert_eq!(queue.purge_dead_jobs(failed_at).await.unwrap(), 0);
        assert_eq!(
            queue
                .purge_dead_jobs(failed_at + chrono::Duration::seconds(1))
                .await
                .unwrap(),
            1
        );
        assert!(queue.get_job(timed_out).await.unwrap().is_some());
    }

    /// With every weight zero, a weighted dequeue takes the highest priority with a job,
    /// as the database backends do, instead of returning nothing.
    #[tokio::test]
    async fn weighted_dequeue_with_zero_weights_takes_the_highest_priority() {
        let queue = TestQueue::new();
        let id = queue
            .enqueue(Job::new("zero_w".to_string(), json!({})))
            .await
            .unwrap();
        let weights = PriorityWeights::new()
            .with_weight(JobPriority::Background, 0)
            .with_weight(JobPriority::Low, 0)
            .with_weight(JobPriority::Normal, 0)
            .with_weight(JobPriority::High, 0)
            .with_weight(JobPriority::Critical, 0);
        let job = queue
            .dequeue_with_priority_weights("zero_w", &weights)
            .await
            .unwrap();
        assert_eq!(job.map(|job| job.id), Some(id));
    }

    /// A huge weight is a probability, not an allocation.
    #[tokio::test]
    async fn weighted_dequeue_with_a_huge_weight_does_not_allocate_per_unit() {
        let queue = TestQueue::new();
        let id = queue
            .enqueue(Job::new("huge_w".to_string(), json!({})))
            .await
            .unwrap();
        let weights = PriorityWeights::new().with_weight(JobPriority::Normal, u32::MAX);
        let job = queue
            .dequeue_with_priority_weights("huge_w", &weights)
            .await
            .unwrap();
        assert_eq!(job.map(|job| job.id), Some(id));
    }

    /// Manual transitions on the wrong status fail with `InvalidJobTransition`, as on the
    /// database backends.
    #[tokio::test]
    async fn retry_dead_job_rejects_a_live_job_as_an_invalid_transition() {
        let queue = TestQueue::new();
        let id = queue
            .enqueue(Job::new("retry_dead".to_string(), json!({})))
            .await
            .unwrap();
        assert!(matches!(
            queue.retry_dead_job(id).await,
            Err(HammerworkError::InvalidJobTransition { .. })
        ));
    }

    /// `cancel_workflow` also fails `Retrying` jobs (a status only older versions
    /// wrote), and a workflow that already failed keeps its `failed_at`, as the
    /// database backends' `COALESCE` does.
    #[tokio::test]
    async fn cancel_workflow_fails_retrying_jobs_and_keeps_failed_at() {
        let queue = TestQueue::new();
        let retrying = Job::new("cancel_retrying".to_string(), json!({}));
        let retrying_id = retrying.id;
        let workflow_id = queue
            .enqueue_workflow(JobGroup::new("cancel_retrying").add_job(retrying))
            .await
            .unwrap();
        let failed_at = queue.clock().now() - chrono::Duration::hours(1);
        {
            let mut storage = queue.storage.write().await;
            storage
                .update_job_status(retrying_id, JobStatus::Retrying)
                .unwrap();
            storage.workflows.get_mut(&workflow_id).unwrap().failed_at = Some(failed_at);
        }

        queue.cancel_workflow(workflow_id).await.unwrap();

        let job = queue.get_job(retrying_id).await.unwrap().unwrap();
        assert_eq!(job.status, JobStatus::Failed);
        assert_eq!(
            job.error_message.as_deref(),
            Some(lifecycle::WORKFLOW_CANCELLED_MESSAGE)
        );
        let workflow = queue
            .get_workflow_status(workflow_id)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(workflow.status, WorkflowStatus::Cancelled);
        assert_eq!(workflow.failed_at, Some(failed_at));
    }

    /// A restored job is scheduled on the mock clock, not the wall clock.
    #[tokio::test]
    async fn restored_job_is_scheduled_on_the_mock_clock() {
        let clock = MockClock::new();
        clock.set_time(Utc::now() - chrono::Duration::days(1));
        let queue = TestQueue::with_clock(clock.clone());
        let id = queue
            .enqueue(Job::new("restore_clock".to_string(), json!({})))
            .await
            .unwrap();
        queue.complete_job(id).await.unwrap();
        let policy = crate::archive::ArchivalPolicy::new()
            .archive_completed_after(chrono::Duration::zero())
            .enabled(true);
        let archived = queue
            .archive_jobs(
                None,
                &policy,
                &crate::archive::ArchivalConfig::new(),
                crate::archive::ArchivalReason::Manual,
                None,
            )
            .await
            .unwrap();
        assert_eq!(archived.jobs_archived, 1);

        let restored = queue.restore_archived_job(id).await.unwrap();
        assert_eq!(restored.scheduled_at, clock.now());
        assert_eq!(
            queue.dequeue("restore_clock").await.unwrap().map(|j| j.id),
            Some(id)
        );
    }
}