later 0.0.48

use std::{fmt::Display, str::FromStr, time::Duration};

use anyhow::Context;

use crate::{retry::RetryBackoff, JobId, RecurringJobId, UtcDateTime};

#[derive(serde::Serialize, serde::Deserialize, Debug)]
#[serde(rename_all = "snake_case", tag = "ty")]
pub(crate) enum AmqpCommand {
    ExecuteJob(JobAmqp),
    /// A topic partition has new or newly-ready work. Carries no ownership
    /// claim: any process that receives it just wakes its own partition
    /// pollers to scan sooner than their idle backoff, the same SQL lease
    /// that already guards `claim_partition_head` decides who actually runs
    /// anything. Safe to lose, duplicate, or deliver to a process that ends
    /// up not owning the partition.
    PartitionReady {
        topic: String,
    },
}

#[derive(serde::Serialize, serde::Deserialize, Debug)]
pub struct JobAmqp {
    pub payload_type: String,
    pub id: JobId,
}

impl From<Job> for JobAmqp {
    fn from(j: Job) -> Self {
        Self {
            payload_type: j.payload_type,
            id: j.id,
        }
    }
}

/// A periodic maintenance operation, driven entirely in-process (see
/// `bg_job_server::start_bg_worker_system_ops_inproc_cmd_to_amqp_cmd`) -
/// not routed through [`AmqpCommand`] and the shared delivery queue the way
/// [`AmqpCommand::ExecuteJob`] is, since these are cheap, idempotent, and
/// safe to run redundantly from every server instance sharing a database;
/// going through the shared queue would mean a deep `ExecuteJob` backlog
/// (FIFO delivery order) could starve them indefinitely.
#[derive(Debug, Clone, Copy)]
#[allow(clippy::enum_variant_names)]
pub(crate) enum ChannelCommand {
    PollDelayedJobs,
    PollRequeuedJobs,
    /// Proactively checks every registered recurring job definition and
    /// enqueues any occurrence that's due. Runs independently of any
    /// occurrence's own execution, so a crashed or never-run occurrence
    /// cannot silently stop the schedule.
    PollRecurringJobs,
    /// Sweeps a bounded batch of already-expired rows out of `later_storage`
    /// and `later_storage_range`. Those tables only ever *filter* expired
    /// rows at read time (a `date_expire` in the past just makes a row
    /// invisible) - something still has to physically delete them, or the
    /// database only ever grows. `later_storage_range`'s reads already do
    /// this opportunistically for whichever range key they touch, but a key
    /// nobody reads for a long time (a burst of stage transitions while the
    /// dashboard sits closed) can build a huge backlog that turns the
    /// eventual sweep into a multi-second write-locking `DELETE`. This poll
    /// keeps both tables trimmed continuously in small batches instead.
    PollExpiredStorage,
    /// Reclaims regular (non-partitioned) jobs whose execution lease
    /// (`later_job_execution`) expired without the job ever reaching a
    /// terminal stage - the worker holding it crashed or was killed mid
    /// handler. Without this, such a job stays `Running` forever: its
    /// delivery was already consumed by the attempt that died, and nothing
    /// else revisits a job by stage the way `PollDelayedJobs`/
    /// `PollRequeuedJobs` do for their stages. Partitioned/sequential jobs
    /// don't need this - a stuck partition head already recovers via its
    /// own partition-lease expiry.
    PollStuckJobs,
    /// Reclaims disk space `PollExpiredStorage` has already freed for reuse
    /// but never returned to the filesystem - see
    /// [`crate::Config::vacuum_interval`]. Unlike every other variant here,
    /// this one does not re-queue itself on completion (see
    /// `bg_job_server::start_bg_worker_system_ops_inproc_cmd_to_amqp_cmd`) -
    /// it only ever runs when this interval-gated backstop re-enqueues it,
    /// since a plain `VACUUM` is far too expensive to run on the same ~10s
    /// cadence as the other polls.
    VacuumDatabase,
}

impl Display for ChannelCommand {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(match self {
            ChannelCommand::PollDelayedJobs => "PollDelayedJobs",
            ChannelCommand::PollRequeuedJobs => "PollRequeuedJobs",
            ChannelCommand::PollRecurringJobs => "PollRecurringJobs",
            ChannelCommand::PollExpiredStorage => "PollExpiredStorage",
            ChannelCommand::PollStuckJobs => "PollStuckJobs",
            ChannelCommand::VacuumDatabase => "VacuumDatabase",
        })
    }
}

#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
#[serde(rename_all = "snake_case")]
pub(crate) struct Job {
    pub id: JobId,

    pub payload_type: String,
    pub payload: Vec<u8>,

    pub config: JobConfig,
    pub stage: Stage,
    pub previous_stages: Vec<Stage>,
    pub recurring_job_id: Option<RecurringJobId>,

    /// Topic this job orders under, when it was enqueued with partition
    /// ordering. `None` keeps the default unordered, competing-worker path.
    #[serde(default)]
    pub topic: Option<String>,
    /// Partition within `topic` that must process this job in order.
    #[serde(default)]
    pub partition: Option<u32>,
    /// Monotonic sequence assigned within `(topic, partition)` at enqueue.
    #[serde(default)]
    pub sequence: Option<i64>,
}

impl Job {
    /// Returns the topic and partition this job orders under, if any.
    pub(crate) fn topic_partition(&self) -> Option<(&str, u32)> {
        match (&self.topic, self.partition) {
            (Some(topic), Some(partition)) => Some((topic.as_str(), partition)),
            _ => None,
        }
    }
}

/// Whether a recurring job's occurrences may run concurrently.
#[derive(serde::Serialize, serde::Deserialize, Debug, Clone, Copy, Default, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub(crate) enum RecurringMode {
    /// Occurrences run on schedule and may overlap. Default, matches the
    /// behavior this feature has always had.
    #[default]
    Normal,
    /// No two instances of this recurring job — including any jobs chained
    /// as continuations of an occurrence — ever run concurrently; they run
    /// in strict order.
    Sequential,
}

#[derive(serde::Serialize, serde::Deserialize, Debug)]
#[serde(rename_all = "snake_case")]
pub(crate) struct RecurringJob {
    pub id: RecurringJobId,

    pub payload_type: String,
    pub payload: Vec<u8>,

    pub cron_schedule: String,
    pub date_added: UtcDateTime,
    pub config: JobConfig,

    /// Concurrency mode for this recurring job's occurrences.
    #[serde(default)]
    pub mode: RecurringMode,
    /// Bookmark used by the proactive recurring-job poller to know when the
    /// next occurrence is due, without recomputing from `date_added` on
    /// every tick. `None` until the poller first computes it.
    #[serde(default)]
    pub next_run_at: Option<UtcDateTime>,
}

impl RecurringJob {
    /// Computes the next `not_before` time from `cron_schedule`, relative to
    /// `after`.
    pub(crate) fn next_occurrence_after(&self, after: UtcDateTime) -> anyhow::Result<UtcDateTime> {
        let cron_schedule = cron::Schedule::from_str(&self.cron_schedule)
            .context("error parsing cron expression")?;
        cron_schedule.after(&after).next().ok_or(anyhow::anyhow!(
            "unable to determine next schedule from this cron expression"
        ))
    }

    /// Builds the `Job` for one occurrence of this recurring job, to run at
    /// or after `not_before`.
    ///
    /// `topic_partition` places the occurrence in a topic partition — used
    /// for [`RecurringMode::Sequential`], where every occurrence and any
    /// continuation chained off one share the same partition so they never
    /// overlap and always run in strict order. `None` keeps the default
    /// unordered path used by [`RecurringMode::Normal`].
    pub(crate) fn occurrence_job(
        &self,
        not_before: UtcDateTime,
        topic_partition: Option<(String, u32)>,
    ) -> Job {
        let (topic, partition) = match topic_partition {
            Some((topic, partition)) => (Some(topic), Some(partition)),
            None => (None, None),
        };
        Job {
            id: JobId(format!("{}R", super::generate_id())),
            payload_type: self.payload_type.clone(),
            payload: self.payload.clone(),
            config: self.config.clone(),
            stage: Stage::Delayed(DelayedStage {
                date: chrono::Utc::now(),
                not_before,
            }),
            previous_stages: Vec::default(),
            recurring_job_id: Some(self.id.clone()),
            topic,
            partition,
            sequence: None,
        }
    }
}

impl TryFrom<RecurringJob> for Job {
    type Error = anyhow::Error;

    fn try_from(value: RecurringJob) -> Result<Self, Self::Error> {
        let delay_until = value.next_occurrence_after(chrono::Utc::now())?;
        Ok(value.occurrence_job(delay_until, None))
    }
}

#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
#[serde(rename_all = "snake_case")]
pub(crate) struct JobConfig {
    pub total_retries: usize,
    #[serde(default)]
    retry_backoff: StoredRetryBackoff,
    #[serde(default = "stored_retry_policy_is_resolved")]
    retry_policy_resolved: bool,
}

fn stored_retry_policy_is_resolved() -> bool {
    true
}

impl Default for JobConfig {
    fn default() -> Self {
        Self {
            total_retries: 6,
            retry_backoff: StoredRetryBackoff::Immediate,
            retry_policy_resolved: false,
        }
    }
}

impl JobConfig {
    pub fn from_policy(policy: &crate::retry::RetryPolicy) -> Self {
        Self {
            total_retries: policy.max_retries(),
            retry_backoff: policy.backoff().into(),
            retry_policy_resolved: false,
        }
    }

    pub fn resolve_retry_policy(&mut self, policy: Option<&crate::retry::RetryPolicy>) {
        if self.retry_policy_resolved {
            return;
        }
        if let Some(policy) = policy {
            self.total_retries = policy.max_retries();
            self.retry_backoff = policy.backoff().into();
        }
        self.retry_policy_resolved = true;
    }

    pub fn needs_retry_policy(&self) -> bool {
        !self.retry_policy_resolved
    }

    fn retry_delay(&self, job_key: &str, retry_number: usize) -> Duration {
        self.retry_backoff.as_public().delay(job_key, retry_number)
    }
}

#[derive(serde::Serialize, serde::Deserialize, Debug, Clone, Default)]
#[serde(rename_all = "snake_case", tag = "type")]
enum StoredRetryBackoff {
    #[default]
    Immediate,
    Fixed {
        delay: Duration,
    },
    Exponential {
        initial_delay: Duration,
        max_delay: Duration,
    },
    ExponentialWithJitter {
        initial_delay: Duration,
        max_delay: Duration,
        max_jitter: Duration,
    },
}

impl From<&RetryBackoff> for StoredRetryBackoff {
    fn from(backoff: &RetryBackoff) -> Self {
        match backoff {
            RetryBackoff::Immediate => Self::Immediate,
            RetryBackoff::Fixed { delay } => Self::Fixed { delay: *delay },
            RetryBackoff::Exponential {
                initial_delay,
                max_delay,
            } => Self::Exponential {
                initial_delay: *initial_delay,
                max_delay: *max_delay,
            },
            RetryBackoff::ExponentialWithJitter {
                initial_delay,
                max_delay,
                max_jitter,
            } => Self::ExponentialWithJitter {
                initial_delay: *initial_delay,
                max_delay: *max_delay,
                max_jitter: *max_jitter,
            },
        }
    }
}

impl StoredRetryBackoff {
    fn as_public(&self) -> RetryBackoff {
        match self {
            Self::Immediate => RetryBackoff::Immediate,
            Self::Fixed { delay } => RetryBackoff::Fixed { delay: *delay },
            Self::Exponential {
                initial_delay,
                max_delay,
            } => RetryBackoff::Exponential {
                initial_delay: *initial_delay,
                max_delay: *max_delay,
            },
            Self::ExponentialWithJitter {
                initial_delay,
                max_delay,
                max_jitter,
            } => RetryBackoff::ExponentialWithJitter {
                initial_delay: *initial_delay,
                max_delay: *max_delay,
                max_jitter: *max_jitter,
            },
        }
    }
}

#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Eq, PartialEq)]
#[serde(rename_all = "snake_case", tag = "type")]
pub(crate) enum Stage {
    /// Scheduled for later or waiting for
    Delayed(DelayedStage),
    Waiting(WaitingStage),
    Enqueued(EnqueuedStage),
    Running(RunningStage),
    Requeued(RequeuedStage),
    Success(SuccessStage),
    Failed(FailedStage),
}

pub trait StageName {
    fn get_name() -> String;
}

impl StageName for EnqueuedStage {
    fn get_name() -> String {
        "enqueued".into()
    }
}

impl StageName for RunningStage {
    fn get_name() -> String {
        "running".into()
    }
}

impl StageName for DelayedStage {
    fn get_name() -> String {
        "delayed".into()
    }
}

impl StageName for WaitingStage {
    fn get_name() -> String {
        "waiting".into()
    }
}

impl StageName for RequeuedStage {
    fn get_name() -> String {
        "requeued".into()
    }
}

impl StageName for SuccessStage {
    fn get_name() -> String {
        "success".into()
    }
}

impl StageName for FailedStage {
    fn get_name() -> String {
        "failed".into()
    }
}

impl DelayedStage {
    pub fn is_time(&self) -> bool {
        chrono::Utc::now() >= self.not_before
    }
}

#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub(crate) struct DelayedStage {
    pub date: UtcDateTime,

    pub not_before: UtcDateTime,
}

#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub(crate) struct WaitingStage {
    pub date: UtcDateTime,

    pub parent_id: JobId,
}

#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub(crate) struct EnqueuedStage {
    pub date: UtcDateTime,
}

#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub(crate) struct RunningStage {
    pub date: UtcDateTime,
}

#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub(crate) struct SuccessStage {
    pub date: UtcDateTime,
}

#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub(crate) struct FailedStage {
    pub date: UtcDateTime,
    pub reason: String,
}

#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub(crate) struct RequeuedStage {
    pub date: UtcDateTime,
    pub requeue_count: usize,
    #[serde(default)]
    pub not_before: Option<UtcDateTime>,
}

impl RequeuedStage {
    pub fn is_ready(&self) -> bool {
        self.not_before
            .is_none_or(|not_before| not_before <= chrono::Utc::now())
    }
}

pub(crate) enum FailureTransition {
    Retry(Job),
    Failed(Job),
}

impl Job {
    #[tracing::instrument]
    pub fn transition(self) -> Job {
        let next_stage = self.stage.clone().transition();
        tracing::debug!(
            "Transition job {}: {} -> {}",
            self.id,
            self.stage.get_name(),
            next_stage.get_name()
        );
        self.transition_to(next_stage)
    }

    fn transition_to(self, next_stage: Stage) -> Job {
        let last_stage = self.stage.clone();
        let mut job = Job {
            stage: next_stage,
            ..self
        };
        job.previous_stages.push(last_stage);
        job
    }

    pub fn transition_failure(self, reason: String) -> anyhow::Result<FailureTransition> {
        let retry_count = self
            .previous_stages
            .iter()
            .filter(|stage| stage.is_req())
            .count()
            + 1;
        if retry_count > self.config.total_retries {
            return self
                .transition_to_terminal_stage(Stage::Failed(FailedStage {
                    date: chrono::Utc::now(),
                    reason,
                }))
                .map(FailureTransition::Failed);
        }

        let delay = self.config.retry_delay(&self.id.to_string(), retry_count);
        let date = chrono::Utc::now();
        let not_before = if delay.is_zero() {
            None
        } else {
            chrono::Duration::from_std(delay)
                .ok()
                .and_then(|delay| date.checked_add_signed(delay))
                .or(Some(chrono::DateTime::<chrono::Utc>::MAX_UTC))
        };
        self.transition_to_terminal_stage(Stage::Requeued(RequeuedStage {
            date,
            requeue_count: retry_count,
            not_before,
        }))
        .map(FailureTransition::Retry)
    }

    pub fn transition_success(self) -> anyhow::Result<Job> {
        self.transition_to_terminal_stage(Stage::Success(SuccessStage {
            date: chrono::Utc::now(),
        }))
    }

    fn transition_to_terminal_stage(self, next_stage: Stage) -> anyhow::Result<Job> {
        if self.stage.is_terminal() {
            return Err(anyhow::anyhow!(
                "Can not transition as job is already is at terminal stage."
            ));
        }
        if let Stage::Running(_) = self.stage {
            return Ok(self.transition_to(next_stage));
        }
        Err(anyhow::anyhow!(
            "Job is not in correct stage to transition to terminal state"
        ))
    }
}

impl Stage {
    #[cfg(any(feature = "dashboard", feature = "prometheus"))]
    pub fn date(&self) -> &UtcDateTime {
        match self {
            Stage::Delayed(stage) => &stage.date,
            Stage::Waiting(stage) => &stage.date,
            Stage::Enqueued(stage) => &stage.date,
            Stage::Running(stage) => &stage.date,
            Stage::Requeued(stage) => &stage.date,
            Stage::Success(stage) => &stage.date,
            Stage::Failed(stage) => &stage.date,
        }
    }

    pub fn get_name(&self) -> String {
        match self {
            Stage::Delayed(_) => DelayedStage::get_name(),
            Stage::Waiting(_) => WaitingStage::get_name(),
            Stage::Enqueued(_) => EnqueuedStage::get_name(),
            Stage::Running(_) => RunningStage::get_name(),
            Stage::Requeued(_) => RequeuedStage::get_name(),
            Stage::Success(_) => SuccessStage::get_name(),
            Stage::Failed(_) => FailedStage::get_name(),
        }
    }

    #[cfg(feature = "dashboard")]
    pub fn get_all_stage_names() -> Vec<String> {
        vec![
            DelayedStage::get_name(),
            WaitingStage::get_name(),
            EnqueuedStage::get_name(),
            RunningStage::get_name(),
            RequeuedStage::get_name(),
            SuccessStage::get_name(),
            FailedStage::get_name(),
        ]
    }

    /// ## Before running
    /// * Delayed -> Scheduled for later
    /// * Waiting -> Waiting for parent job to complete
    ///
    /// ## Running
    /// * Enqueued -> Published
    /// * Running -> A worker accepted the job and running
    ///
    /// ## After running for at least once
    /// * Requeued -> Job failed and retried ... (Next: Enqueued)
    /// * Success -> Job is successful
    pub fn transition(self) -> Stage {
        match self {
            Stage::Delayed(_) => Stage::Enqueued(EnqueuedStage {
                date: chrono::Utc::now(),
            }),
            Stage::Waiting(_) => Stage::Enqueued(EnqueuedStage {
                date: chrono::Utc::now(),
            }),
            Stage::Enqueued(_) => Stage::Running(RunningStage {
                date: chrono::Utc::now(),
            }),
            Stage::Running(_) => self,
            Stage::Requeued(_) => Stage::Enqueued(EnqueuedStage {
                date: chrono::Utc::now(),
            }),
            Stage::Success(_) => self, /* Terminal */
            Stage::Failed(_) => self,  /* Terminal */
        }
    }

    pub fn is_req(&self) -> bool {
        match self {
            Stage::Requeued(_) => true,
            Stage::Delayed(_)
            | Stage::Waiting(_)
            | Stage::Enqueued(_)
            | Stage::Running(_)
            | Stage::Success(_)
            | Stage::Failed(_) => false,
        }
    }

    pub fn is_terminal(&self) -> bool {
        match self {
            Stage::Success(_) | Stage::Failed(_) => true,

            Stage::Delayed(_)
            | Stage::Waiting(_)
            | Stage::Enqueued(_)
            | Stage::Running(_)
            | Stage::Requeued(_) => false,
        }
    }

    pub fn is_success(&self) -> bool {
        match self {
            Stage::Success(_) => true,
            Stage::Delayed(_)
            | Stage::Waiting(_)
            | Stage::Enqueued(_)
            | Stage::Running(_)
            | Stage::Requeued(_)
            | Stage::Failed(_) => false,
        }
    }
}

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

    #[test]
    fn every_stage_has_its_own_name() {
        assert_eq!(
            vec![
                DelayedStage::get_name(),
                WaitingStage::get_name(),
                EnqueuedStage::get_name(),
                RunningStage::get_name(),
                RequeuedStage::get_name(),
                SuccessStage::get_name(),
                FailedStage::get_name(),
            ],
            vec!["delayed", "waiting", "enqueued", "running", "requeued", "success", "failed"]
        );
    }

    #[test]
    fn old_job_without_topic_partition_or_sequence_fields_remains_readable() -> anyhow::Result<()> {
        #[derive(serde::Serialize)]
        #[serde(rename_all = "snake_case")]
        struct OldJob {
            id: JobId,
            payload_type: String,
            payload: Vec<u8>,
            config: JobConfig,
            stage: Stage,
            previous_stages: Vec<Stage>,
            recurring_job_id: Option<RecurringJobId>,
        }

        let old = OldJob {
            id: JobId("old-job".to_string()),
            payload_type: "test".to_string(),
            payload: vec![1, 2, 3],
            config: JobConfig::default(),
            stage: Stage::Enqueued(EnqueuedStage {
                date: chrono::Utc::now(),
            }),
            previous_stages: Vec::new(),
            recurring_job_id: None,
        };

        let job: Job = crate::encoder::decode(&crate::encoder::encode(old)?)?;

        assert_eq!(job.id, JobId("old-job".to_string()));
        assert_eq!(job.topic, None);
        assert_eq!(job.partition, None);
        assert_eq!(job.sequence, None);
        assert_eq!(job.topic_partition(), None);
        Ok(())
    }

    #[test]
    fn old_recurring_job_without_mode_or_next_run_at_remains_readable() -> anyhow::Result<()> {
        #[derive(serde::Serialize)]
        #[serde(rename_all = "snake_case")]
        struct OldRecurringJob {
            id: RecurringJobId,
            payload_type: String,
            payload: Vec<u8>,
            cron_schedule: String,
            date_added: UtcDateTime,
            config: JobConfig,
        }

        let old = OldRecurringJob {
            id: RecurringJobId("old-recurring".to_string()),
            payload_type: "test".to_string(),
            payload: vec![1, 2, 3],
            cron_schedule: "0 * * * * *".to_string(),
            date_added: chrono::Utc::now(),
            config: JobConfig::default(),
        };

        let recurring: RecurringJob = crate::encoder::decode(&crate::encoder::encode(old)?)?;

        assert_eq!(recurring.mode, RecurringMode::Normal);
        assert_eq!(recurring.next_run_at, None);
        Ok(())
    }

    #[test]
    fn old_job_config_and_requeued_stage_data_remain_readable() -> anyhow::Result<()> {
        #[derive(serde::Serialize)]
        struct OldJobConfig {
            total_retries: usize,
        }
        #[derive(serde::Serialize)]
        struct OldRequeuedStage {
            date: UtcDateTime,
            requeue_count: usize,
        }

        let config: JobConfig =
            crate::encoder::decode(&crate::encoder::encode(OldJobConfig { total_retries: 4 })?)?;
        let stage: RequeuedStage =
            crate::encoder::decode(&crate::encoder::encode(OldRequeuedStage {
                date: chrono::Utc::now(),
                requeue_count: 2,
            })?)?;

        assert_eq!(config.total_retries, 4);
        assert!(!config.needs_retry_policy());
        assert_eq!(config.retry_delay("old-job", 1), Duration::ZERO);
        assert_eq!(stage.requeue_count, 2);
        assert_eq!(stage.not_before, None);
        assert!(stage.is_ready());
        Ok(())
    }

    #[test]
    fn retry_policy_is_resolved_once() {
        let mut config = JobConfig::from_policy(&crate::retry::RetryPolicy::no_retries());
        let first = crate::retry::RetryPolicy::fixed(2, Duration::from_secs(5));
        let later = crate::retry::RetryPolicy::immediate(9);

        config.resolve_retry_policy(Some(&first));
        config.resolve_retry_policy(Some(&later));

        assert!(!config.needs_retry_policy());
        assert_eq!(config.total_retries, 2);
        assert_eq!(config.retry_delay("job", 1), Duration::from_secs(5));
    }

    #[test]
    fn failure_transition_retries_then_marks_the_job_failed() -> anyhow::Result<()> {
        let policy = crate::retry::RetryPolicy::fixed(1, Duration::from_secs(10));
        let job = Job {
            id: JobId("retry-test".to_string()),
            payload_type: "test".to_string(),
            payload: Vec::new(),
            config: JobConfig::from_policy(&policy),
            stage: Stage::Running(RunningStage {
                date: chrono::Utc::now(),
            }),
            previous_stages: Vec::new(),
            recurring_job_id: None,
            topic: None,
            partition: None,
            sequence: None,
        };

        let retried = match job.transition_failure("first".to_string())? {
            FailureTransition::Retry(job) => job,
            FailureTransition::Failed(_) => return Err(anyhow::anyhow!("first failure was final")),
        };
        let Stage::Requeued(requeued) = &retried.stage else {
            return Err(anyhow::anyhow!("first failure was not requeued"));
        };
        let retry_delay = requeued
            .not_before
            .ok_or_else(|| anyhow::anyhow!("retry had no delay"))?
            - requeued.date;
        assert_eq!(retry_delay, chrono::Duration::seconds(10));

        let running_again = retried.transition().transition();
        let failed = match running_again.transition_failure("second".to_string())? {
            FailureTransition::Retry(_) => {
                return Err(anyhow::anyhow!("retry limit was not applied"));
            }
            FailureTransition::Failed(job) => job,
        };
        let Stage::Failed(failed_stage) = failed.stage else {
            return Err(anyhow::anyhow!("exhausted job was not failed"));
        };
        assert_eq!(failed_stage.reason, "second");
        Ok(())
    }
}