dbos 0.6.0

//! Schedules: cron-driven workflows, stored as rows in `workflow_schedules`.
//!
//! **A schedule is a database object, not a decorator.** It names a workflow, a cron expression
//! and a context, and every executor of its application fires it: each computes the next tick,
//! sleeps until then, and enqueues the workflow under an id derived from the schedule and the
//! tick, so the fleet runs each tick once however many executors woke for it.
//!
//! Schedules are managed at runtime, from [`DBOS`] or from a [`Client`](crate::Client), and the
//! running executors notice within a poll — creating, pausing, changing or deleting one needs no
//! restart. See [`scheduler`](crate::scheduler) for the loop that fires them.
//!
//! # A scheduled workflow
//!
//! Any registered workflow whose argument is a [`ScheduledWorkflowInput`]: the tick it is running
//! for and the schedule's context. [`ScheduleSpec::new`] takes the workflow by its
//! [`WorkflowRef`], which is what holds the context's type to the workflow's — a schedule whose
//! context the workflow cannot read does not compile.
//!
//! ```no_run
//! # async fn f(dbos: &dbos::DBOS) -> dbos::Result<()> {
//! use dbos::{ScheduleSpec, ScheduledWorkflowInput};
//!
//! let report = dbos.register_workflow(
//!     "nightly-report",
//!     |input: ScheduledWorkflowInput<String>| async move {
//!         println!("report for {} at {:?}", input.context, input.scheduled_time);
//!         Ok::<(), dbos::Error>(())
//!     },
//! )?;
//! dbos.launch().await?;
//!
//! let mut spec = ScheduleSpec::new("nightly", &report, "0 0 2 * * *", &"emea".to_owned())?;
//! spec.cron_timezone = Some("Europe/Paris".to_owned());
//! dbos.create_schedule(&spec).await?;
//! # Ok(()) }
//! ```
//!
//! # Called from inside a workflow
//!
//! As on the rest of the management surface, the single-schedule operations on [`DBOS`] are
//! recorded as steps — `DBOS.createSchedule`, `DBOS.listSchedules` and so on — so a replay reads
//! back what the first run did. [`apply_schedules`](DBOS::apply_schedules),
//! [`backfill_schedule`](DBOS::backfill_schedule) and [`trigger_schedule`](DBOS::trigger_schedule)
//! are refused inside a workflow instead: each writes a batch with no checkpoint, which a replay
//! would write again.

use std::collections::HashSet;
use std::sync::Arc;
use std::time::SystemTime;

use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};

use crate::checkpoint::{PendingStep, StepPlacement, StepWriter};
use crate::client::Client;
use crate::connection::Connection;
use crate::context::Ctx;
use crate::cron::{CronSchedule, firing_id, trigger_id};
use crate::error::{Error, Result};
use crate::handle::WorkflowHandle;
use crate::instance::DBOS;
use crate::registry::{QualifiedWorkflowName, Snapshot, WorkflowRef};
use crate::serialization::{
    Floats, NonFinite, PayloadEncoding, SerializationFormat, Serializer, decode, encode,
    encode_input,
};
use crate::sysdb::INTERNAL_QUEUE;
use crate::sysdb::types::{
    Change, NewSchedule, NewWorkflow, ScheduleFilter, ScheduleRecord, ScheduleStatus,
    ScheduleUpdate, Timestamp, WorkflowIdReusePolicy, step_names,
};
use crate::workflow::{DuplicationPolicy, QueueRequest, init_or_join, new_workflow};

/// What a scheduled workflow is called with: the tick it is running for, and the schedule's
/// context.
///
/// A struct because a workflow takes one argument. Encoded as `{"scheduled_time": <RFC 3339 text>,
/// "context": <the stored context>}`.
///
/// **The context travels as JSON data**, in whatever format carries the input: it is turned into a
/// `serde_json::Value` on the way in and read back from one, as a schedule's stored context is. A
/// type that serializes differently in a format that is not human-readable — an IP address, a UUID
/// — therefore reads back the same through a binary [`CustomSerializer`](crate::CustomSerializer)
/// as through JSON.
///
/// `scheduled_time` is the **tick**, not the moment the workflow started: a workflow that waited
/// in a queue, or one enqueued by a backfill, still reports the time it was scheduled for. A
/// manual [`trigger`](DBOS::trigger_schedule) reports the time it was triggered.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(bound(serialize = "C: Serialize", deserialize = "C: DeserializeOwned"))]
pub struct ScheduledWorkflowInput<C = ()> {
    /// The cron tick this run is for.
    #[serde(with = "rfc3339")]
    pub scheduled_time: SystemTime,
    /// The schedule's context.
    #[serde(with = "json_data")]
    pub context: C,
}

/// `SystemTime` as RFC 3339 text, rather than serde's `{secs_since_epoch, nanos_since_epoch}`.
///
/// Text so that a reader of the row can make sense of it without arithmetic.
mod rfc3339 {
    use std::time::SystemTime;

    use serde::{Deserialize, Deserializer, Serializer, de};

    pub(super) fn serialize<S: Serializer>(
        time: &SystemTime,
        serializer: S,
    ) -> Result<S::Ok, S::Error> {
        let timestamp = jiff::Timestamp::try_from(*time).map_err(serde::ser::Error::custom)?;
        serializer.collect_str(&timestamp)
    }

    pub(super) fn deserialize<'de, D: Deserializer<'de>>(
        deserializer: D,
    ) -> Result<SystemTime, D::Error> {
        let text = String::deserialize(deserializer)?;
        let timestamp: jiff::Timestamp = text.parse().map_err(de::Error::custom)?;
        Ok(SystemTime::from(timestamp))
    }
}

/// A context as the JSON data it is stored as, whatever format carries it.
///
/// The stored context is a `serde_json::Value`, which a format that is not human-readable carries
/// as JSON data: an `Ipv4Addr` stored as `"10.0.0.1"` stays a string in CBOR. Reading `C` straight
/// from such a format would expect `C`'s binary form — four octets — and fail, so `C` is read from
/// the JSON data instead, and written as it, so that both directions agree.
mod json_data {
    use serde::de::DeserializeOwned;

    use crate::serialization::{Floats, NonFinite};
    use serde::{Deserialize, Deserializer, Serialize, Serializer, de, ser};

    pub(super) fn serialize<C: Serialize, S: Serializer>(
        context: &C,
        serializer: S,
    ) -> Result<S::Ok, S::Error> {
        serde_json::to_value(Floats::new(context, NonFinite::Refuse))
            .map_err(ser::Error::custom)?
            .serialize(serializer)
    }

    pub(super) fn deserialize<'de, C: DeserializeOwned, D: Deserializer<'de>>(
        deserializer: D,
    ) -> Result<C, D::Error> {
        serde_json::from_value(serde_json::Value::deserialize(deserializer)?)
            .map_err(de::Error::custom)
    }
}

/// A schedule's definition: what [`create_schedule`](DBOS::create_schedule) creates and
/// [`apply_schedules`](DBOS::apply_schedules) reconciles.
///
/// The context is turned into a JSON value when the spec is built, so a spec that exists is one
/// whose context could be written. It is encoded with the configured
/// [`Serializer`](crate::Serializer) when the spec is written. The rest are plain fields to set
/// after construction.
///
/// # The cron dialect
///
/// Five or six fields, seconds first when there are six. A five-field pattern fires at second
/// zero.
///
/// - **Nicknames:** `@yearly`, `@annually`, `@monthly`, `@weekly`, `@daily`, `@midnight`,
///   `@hourly`.
/// - **Names:** months and weekdays, three-letter or in full, in any case.
/// - **`?`** in day-of-month or day-of-week, meaning `*`.
/// - **`L`** (the last day of the month), **`LW`** (its last weekday) and **`nW`** (the weekday
///   nearest day *n*) in day-of-month; **`nL`** (the last weekday *n* of the month) and **`n#m`**
///   (the *m*th weekday *n*) in day-of-week. Day-of-week `7` is Sunday, as `0` is.
/// - **Steps** on `*` or on a range: `*/15`, `10-50/10`. A step on a single value (`5/10`) is
///   refused.
/// - **Day-of-month and day-of-week must both match** when both are restricted, so `0 0 13 * 5`
///   is Friday the 13th, not every Friday and every 13th.
/// - **A pattern that can never fire is refused**, such as `0 0 31 2 *`.
///
/// # Not accepted
///
/// `L-n` (the *n*th-from-last day), inverted ranges (`22-2`, `Fri-Mon`), and a seventh field are
/// all refused.
///
/// # Daylight saving
///
/// A wall-clock time skipped at spring-forward fires at the first instant after the gap. At
/// fall-back, a pattern for a fixed time of day fires once, on the first of the two repeated
/// hours, while an interval pattern such as `*/15 * * * *` fires in both, since each is a
/// quarter-hour that happened. A daily job at `02:30` or `01:30` therefore runs exactly once every
/// day of the year.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ScheduleSpec {
    /// The schedule's name, unique across every application sharing the database.
    pub schedule_name: String,
    /// The cron expression, in the dialect described above.
    pub schedule: String,
    /// The workflow each tick enqueues.
    ///
    /// A configured instance cannot be scheduled: the table has no column for one.
    pub workflow: QualifiedWorkflowName,
    /// The context each tick is called with.
    context: serde_json::Value,
    /// Whether an executor picking the schedule up — after a restart, or a deploy — first enqueues
    /// the ticks missed since it last fired.
    pub automatic_backfill: bool,
    /// The IANA timezone the cron expression is read in, or `None` for UTC.
    pub cron_timezone: Option<String>,
    /// The queue each tick is enqueued on, or `None` for the engine's internal queue.
    pub queue_name: Option<String>,
    /// The application that owns the schedule and runs its workflows, or `None` for the caller's.
    ///
    /// A client with no application name and no name here writes an **unclaimed** schedule,
    /// which every application sharing the database fires.
    pub app_name: Option<String>,
}

impl ScheduleSpec {
    /// A schedule enqueuing `workflow` on every tick of `schedule`, called with `context`.
    ///
    /// The workflow's argument type is what fixes `C`, so a context the workflow could not read
    /// is a compile error rather than a run that fails on every tick.
    pub fn new<C, R, E>(
        schedule_name: impl Into<String>,
        workflow: &WorkflowRef<ScheduledWorkflowInput<C>, R, E>,
        schedule: impl Into<String>,
        context: &C,
    ) -> Result<Self>
    where
        C: Serialize,
    {
        Self::for_workflow(schedule_name, workflow.qualified_name(), schedule, context)
    }

    /// [`new`](Self::new) for a workflow named rather than held, which is what a
    /// [`Client`](crate::Client) has.
    ///
    /// Nothing here can hold the context to the workflow's type, so a mismatch surfaces as a
    /// deserialization failure in each run.
    pub fn for_workflow<C>(
        schedule_name: impl Into<String>,
        workflow: impl Into<QualifiedWorkflowName>,
        schedule: impl Into<String>,
        context: &C,
    ) -> Result<Self>
    where
        C: Serialize,
    {
        Ok(Self {
            schedule_name: schedule_name.into(),
            schedule: schedule.into(),
            workflow: workflow.into(),
            context: serde_json::to_value(Floats::new(context, NonFinite::Refuse))
                .map_err(|source| Error::encode_failed("schedule context", source))?,
            automatic_backfill: false,
            cron_timezone: None,
            queue_name: None,
            app_name: None,
        })
    }

    /// The context each tick is called with.
    pub fn context(&self) -> &serde_json::Value {
        &self.context
    }

    /// Refuses what no executor could fire: a malformed cron expression, an unknown timezone, a
    /// configured instance, an empty name.
    fn validate(&self, operation: &'static str) -> Result<()> {
        let refuse = |detail: String| Error::InvalidArgument {
            operation: operation.into(),
            detail: format!("schedule `{}`: {detail}", self.schedule_name),
        };
        if self.schedule_name.is_empty() {
            return Err(Error::InvalidArgument {
                operation: operation.into(),
                detail: "a schedule needs a name".to_owned(),
            });
        }
        if self.workflow.name().is_empty() {
            return Err(refuse(
                "`workflow` must not be empty: nothing could run the schedule".to_owned(),
            ));
        }
        if self.workflow.config_name().is_some() {
            return Err(refuse(format!(
                "the configured instance {} cannot be scheduled",
                self.workflow
            )));
        }
        CronSchedule::parse(&self.schedule, self.cron_timezone.as_deref()).map_err(refuse)?;
        Ok(())
    }

    /// The context, encoded with `configured` as the row stores it.
    fn encode_context(&self, configured: &Serializer) -> Result<String> {
        encode(&self.context, &configured.encoding(), "schedule context")
    }

    /// The row this spec writes, with its context already encoded. `schedule_id` is generated by
    /// the system database.
    fn as_new<'a>(&'a self, context: &'a str) -> NewSchedule<'a> {
        NewSchedule {
            workflow_class_name: self.workflow.class_name(),
            context,
            automatic_backfill: self.automatic_backfill,
            cron_timezone: self.cron_timezone.as_deref(),
            queue_name: self.queue_name.as_deref(),
            application_name: self.app_name.as_deref(),
            ..NewSchedule::new(&self.schedule_name, self.workflow.name(), &self.schedule)
        }
    }
}

/// A change to a schedule's definition, leaving what it does not name.
///
/// The workflow cannot be changed — a schedule for a different workflow is a different schedule —
/// and neither can the status, which is [`pause_schedule`](DBOS::pause_schedule)'s, nor
/// `last_fired_at`, which is the loop's.
///
/// The context is a [`serde_json::Value`] rather than the workflow's type, because nothing here
/// holds the workflow to check it against.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct ScheduleChange<'a> {
    /// The cron expression.
    pub schedule: Change<&'a str>,
    /// The context each tick is called with.
    pub context: Change<serde_json::Value>,
    /// Whether a restart first enqueues the ticks it missed.
    pub automatic_backfill: Change<bool>,
    /// The timezone the expression is read in; `Set(None)` is UTC.
    pub cron_timezone: Change<Option<&'a str>>,
    /// The queue ticks are enqueued on; `Set(None)` is the internal queue.
    pub queue_name: Change<Option<&'a str>>,
}

impl ScheduleChange<'_> {
    /// Refuses a change that would leave a schedule no executor could fire.
    ///
    /// **A new expression is checked against the new timezone or UTC**, not the stored one: the
    /// stored row is not read, and an expression that fires somewhere fires everywhere.
    fn validate(&self, name: &str, operation: &'static str) -> Result<()> {
        let refuse = |detail: String| Error::InvalidArgument {
            operation: operation.into(),
            detail: format!("schedule `{name}`: {detail}"),
        };
        let timezone = match self.cron_timezone {
            Change::Set(timezone) => timezone,
            Change::Leave => None,
        };
        match self.schedule {
            Change::Set(expression) => {
                CronSchedule::parse(expression, timezone).map_err(refuse)?;
            }
            Change::Leave if timezone.is_some() => {
                CronSchedule::parse("* * * * *", timezone).map_err(refuse)?;
            }
            Change::Leave => {}
        }
        Ok(())
    }

    /// The new context, encoded with `configured` as the row stores it, or `None` to leave it.
    fn encode_context(&self, configured: &Serializer) -> Result<Option<String>> {
        match &self.context {
            Change::Set(context) => Ok(Some(encode(
                context,
                &configured.encoding(),
                "schedule context",
            )?)),
            Change::Leave => Ok(None),
        }
    }
}

/// A schedule, as the database holds it.
///
/// What [`get_schedule`](DBOS::get_schedule) and [`list_schedules`](DBOS::list_schedules) return:
/// the definition a [`ScheduleSpec`] wrote, and the state the scheduler keeps beside it — whether
/// it is paused, and when it last fired.
///
/// The context is kept encoded, and [`decode_context`](Self::decode_context) decodes it on request:
/// a schedule written by another SDK, or under another serializer, may hold a context this one
/// cannot read, and listing it should not fail for that.
#[derive(Debug, Clone)]
pub struct Schedule {
    id: String,
    schedule_name: String,
    workflow: QualifiedWorkflowName,
    schedule: String,
    status: ScheduleStatus,
    context: String,
    /// What reads the context: the serializer of the handle that read the row.
    serializer: Serializer,
    last_fired_at: Option<SystemTime>,
    automatic_backfill: bool,
    cron_timezone: Option<String>,
    queue_name: Option<String>,
    app_name: Option<String>,
}

/// Two views are equal when they read the same row: which serializer reads the context is not
/// part of the schedule.
impl PartialEq for Schedule {
    fn eq(&self, other: &Self) -> bool {
        let Schedule {
            id,
            schedule_name,
            workflow,
            schedule,
            status,
            context,
            serializer: _,
            last_fired_at,
            automatic_backfill,
            cron_timezone,
            queue_name,
            app_name,
        } = self;
        *id == other.id
            && *schedule_name == other.schedule_name
            && *workflow == other.workflow
            && *schedule == other.schedule
            && *status == other.status
            && *context == other.context
            && *last_fired_at == other.last_fired_at
            && *automatic_backfill == other.automatic_backfill
            && *cron_timezone == other.cron_timezone
            && *queue_name == other.queue_name
            && *app_name == other.app_name
    }
}

impl Eq for Schedule {}

impl Schedule {
    /// Builds the read view from a row, read by a handle configured with `serializer`.
    pub(crate) fn from_record(record: ScheduleRecord, serializer: &Serializer) -> Self {
        Self {
            workflow: QualifiedWorkflowName::from_row(
                record.workflow_name,
                record.workflow_class_name.as_deref(),
                None,
            ),
            id: record.schedule_id,
            schedule_name: record.schedule_name,
            schedule: record.schedule,
            status: record.status,
            context: record.context,
            serializer: serializer.clone(),
            // A `Timestamp` the clock cannot represent is not one a schedule fired at.
            last_fired_at: record.last_fired_at.and_then(Timestamp::to_system_time),
            automatic_backfill: record.automatic_backfill,
            cron_timezone: record.cron_timezone,
            queue_name: record.queue_name,
            app_name: record.application_name,
        }
    }

    /// The schedule's name, unique across every application sharing the database.
    pub fn schedule_name(&self) -> &str {
        &self.schedule_name
    }

    /// The id the database generated for the schedule. A schedule deleted and created again under
    /// the same name has a new one.
    pub fn id(&self) -> &str {
        &self.id
    }

    /// The workflow each tick enqueues.
    pub fn workflow(&self) -> &QualifiedWorkflowName {
        &self.workflow
    }

    /// The cron expression.
    pub fn schedule(&self) -> &str {
        &self.schedule
    }

    /// Whether the schedule is firing or paused.
    pub fn status(&self) -> ScheduleStatus {
        self.status
    }

    /// The context each tick is called with, encoded.
    pub fn context(&self) -> &str {
        &self.context
    }

    /// The context each tick is called with, decoded as `C`.
    ///
    /// Read with the configured [`Serializer`](crate::Serializer). The column carries no tag, so
    /// a context written under another serializer — including one written before a
    /// [`CustomSerializer`](crate::CustomSerializer) was configured — is read as if this one had
    /// written it: it fails to decode, or decodes to something else.
    pub fn decode_context<C: DeserializeOwned>(&self) -> Result<C> {
        decode_context(&self.context, &self.serializer)
    }

    /// The tick the schedule last fired for, or `None` if it has not fired.
    pub fn last_fired_at(&self) -> Option<SystemTime> {
        self.last_fired_at
    }

    /// Whether an executor picking the schedule up first enqueues the ticks it missed.
    pub fn automatic_backfill(&self) -> bool {
        self.automatic_backfill
    }

    /// The IANA timezone the cron expression is read in, or `None` for UTC.
    pub fn cron_timezone(&self) -> Option<&str> {
        self.cron_timezone.as_deref()
    }

    /// The queue each tick is enqueued on, or `None` for the engine's internal queue.
    pub fn queue_name(&self) -> Option<&str> {
        self.queue_name.as_deref()
    }

    /// The application that owns the schedule, or `None` if it is unclaimed and every
    /// application sharing the database fires it.
    pub fn app_name(&self) -> Option<&str> {
        self.app_name.as_deref()
    }
}

/// [`Error::InsideWorkflow`] for a call made from inside a workflow body or one of its steps.
fn refuse_inside_workflow(operation: &'static str) -> Result<()> {
    match Ctx::current() {
        Some(_) => Err(Error::InsideWorkflow {
            operation: operation.into(),
        }),
        None => Ok(()),
    }
}

impl DBOS {
    /// Creates a schedule, which every executor of its application then fires.
    ///
    /// Refused, before anything is written, if the cron expression or the timezone is invalid, if
    /// the pattern never fires, if this instance has no registration for the workflow, or if the
    /// queue is not registered. A name already taken is an error; to create-or-replace, use
    /// [`apply_schedules`](Self::apply_schedules).
    ///
    /// The executors pick the schedule up within their polling interval — a second, for the first
    /// minute after launch, so a schedule created at startup starts promptly.
    pub fn create_schedule<'a>(&self, spec: &'a ScheduleSpec) -> PendingStep<'a, ()> {
        const OPERATION: &str = "create a schedule";
        let built = self.executor(OPERATION).and_then(|executor| {
            // Checks of the spec alone, which a replay answers the same way, so they can refuse
            // the call before it takes a step id.
            spec.validate(OPERATION)?;
            let placement = StepPlacement::of(executor.connection(), OPERATION)?;
            Ok((executor, placement))
        });
        PendingStep::placed(
            step_names::CREATE_SCHEDULE,
            built,
            move |executor, placement| async move {
                let conn = executor.connection();
                // Checks of the world as it is now — the registry, the queues table — only on the
                // first run. A replay returns what that run recorded, even if the queue has since
                // been deleted.
                if placement
                    .check(conn, step_names::CREATE_SCHEDULE)
                    .await?
                    .is_none()
                {
                    refuse_unregistered(&executor, &spec.workflow, OPERATION)?;
                    refuse_unknown_queue(
                        conn,
                        &spec.schedule_name,
                        spec.queue_name.as_deref(),
                        OPERATION,
                    )
                    .await?;
                }
                conn.create_schedule(spec, placement.step_writer()).await
            },
        )
    }

    /// Creates or replaces a set of schedules, in one transaction.
    ///
    /// The declarative form: an application states the schedules it wants at startup, and running
    /// this again — from the same deploy or the next — converges on them. An existing schedule
    /// keeps its id, its status and when it last fired, so re-applying does not resume a paused
    /// schedule or re-run a tick.
    ///
    /// Every spec is checked as [`create_schedule`](Self::create_schedule) checks one before any
    /// is written. **Refused inside a workflow**: it writes many rows with no checkpoint.
    pub async fn apply_schedules(&self, specs: &[ScheduleSpec]) -> Result<()> {
        const OPERATION: &str = "apply schedules";
        let executor = self.executor(OPERATION)?;
        refuse_inside_workflow(OPERATION)?;
        // Each distinct queue read once: a startup declaring many schedules usually puts them on
        // a few queues.
        let mut checked = HashSet::new();
        for spec in specs {
            spec.validate(OPERATION)?;
            refuse_unregistered(&executor, &spec.workflow, OPERATION)?;
            if checked.insert(spec.queue_name.as_deref()) {
                refuse_unknown_queue(
                    executor.connection(),
                    &spec.schedule_name,
                    spec.queue_name.as_deref(),
                    OPERATION,
                )
                .await?;
            }
        }
        executor.connection().apply_schedules(specs).await
    }

    /// The schedules matching `filter`, by name.
    ///
    /// The default filter lists this application's schedules and the unclaimed ones.
    pub fn list_schedules<'a>(
        &self,
        filter: &'a ScheduleFilter<'a>,
    ) -> PendingStep<'a, Vec<Schedule>> {
        PendingStep::placed(
            step_names::LIST_SCHEDULES,
            self.placed("list schedules"),
            move |executor, placement| async move {
                let records = executor
                    .connection()
                    .list_schedules(filter, placement.step_writer())
                    .await?;
                let serializer = executor.serializer();
                Ok(records
                    .into_iter()
                    .map(|record| Schedule::from_record(record, serializer))
                    .collect())
            },
        )
    }

    /// The schedule with this name, or `None`.
    pub fn get_schedule<'a>(&self, name: &'a str) -> PendingStep<'a, Option<Schedule>> {
        PendingStep::placed(
            step_names::GET_SCHEDULE,
            self.placed("read a schedule"),
            move |executor, placement| async move {
                let record = executor
                    .connection()
                    .get_schedule(name, placement.step_writer())
                    .await?;
                Ok(record.map(|record| Schedule::from_record(record, executor.serializer())))
            },
        )
    }

    /// Changes a schedule's definition, leaving what `change` does not name.
    ///
    /// Running executors restart the schedule with the new definition at their next poll. A name
    /// with no schedule is an error, even for an empty change, and a new queue must be registered,
    /// as [`create_schedule`](Self::create_schedule) requires.
    pub fn update_schedule<'a>(
        &self,
        name: &'a str,
        change: &'a ScheduleChange<'a>,
    ) -> PendingStep<'a, ()> {
        const OPERATION: &str = "update a schedule";
        let built = self.executor(OPERATION).and_then(|executor| {
            change.validate(name, OPERATION)?;
            let context = change.encode_context(executor.serializer())?;
            let placement = StepPlacement::of(executor.connection(), OPERATION)?;
            Ok(((executor, context), placement))
        });
        PendingStep::placed(
            step_names::UPDATE_SCHEDULE,
            built,
            move |(executor, context), placement| async move {
                let conn = executor.connection();
                // As in `create_schedule`: the queue is checked on the first run only.
                if let Change::Set(queue) = change.queue_name
                    && placement
                        .check(conn, step_names::UPDATE_SCHEDULE)
                        .await?
                        .is_none()
                {
                    refuse_unknown_queue(conn, name, queue, OPERATION).await?;
                }
                conn.update_schedule(name, change, context.as_deref(), placement.step_writer())
                    .await
            },
        )
    }

    /// Stops a schedule firing, until [`resume_schedule`](Self::resume_schedule).
    ///
    /// **With [`automatic_backfill`](ScheduleSpec::automatic_backfill), resuming runs the ticks
    /// missed while paused**, because the schedule catches up from when it last fired whenever an
    /// executor starts firing it — and resuming is that.
    /// Without it, ticks that passed during the pause are not run.
    pub fn pause_schedule<'a>(&self, name: &'a str) -> PendingStep<'a, ()> {
        PendingStep::placed(
            step_names::PAUSE_SCHEDULE,
            self.placed("pause a schedule"),
            move |executor, placement| async move {
                executor
                    .connection()
                    .set_schedule_status(name, ScheduleStatus::Paused, placement.step_writer())
                    .await
            },
        )
    }

    /// Lets a paused schedule fire again, from its next tick.
    pub fn resume_schedule<'a>(&self, name: &'a str) -> PendingStep<'a, ()> {
        PendingStep::placed(
            step_names::RESUME_SCHEDULE,
            self.placed("resume a schedule"),
            move |executor, placement| async move {
                executor
                    .connection()
                    .set_schedule_status(name, ScheduleStatus::Active, placement.step_writer())
                    .await
            },
        )
    }

    /// Deletes a schedule. Deleting one that does not exist is not an error.
    ///
    /// Runs the schedule has already enqueued are not touched.
    pub fn delete_schedule<'a>(&self, name: &'a str) -> PendingStep<'a, ()> {
        PendingStep::placed(
            step_names::DELETE_SCHEDULE,
            self.placed("delete a schedule"),
            move |executor, placement| async move {
                executor
                    .connection()
                    .delete_schedule(name, placement.step_writer())
                    .await
            },
        )
    }

    /// Enqueues every tick of a schedule strictly between `start` and `end`, and hands back a
    /// handle to each.
    ///
    /// Each tick is enqueued under the id the loop would have used, so a tick that already ran is
    /// not run again — its handle is returned all the same, so the result covers the whole
    /// window. **Refused inside a workflow**: it enqueues many runs with no checkpoint.
    pub async fn backfill_schedule<R, E>(
        &self,
        name: &str,
        start: SystemTime,
        end: SystemTime,
    ) -> Result<Vec<WorkflowHandle<R, E>>> {
        const OPERATION: &str = "backfill a schedule";
        let executor = self.executor(OPERATION)?;
        refuse_inside_workflow(OPERATION)?;
        executor
            .connection()
            .backfill_schedule(name, start, end, Some(executor.workflows()))
            .await
    }

    /// Enqueues one run of a schedule's workflow now, outside its cron expression.
    ///
    /// The run is called with the time it was triggered as its
    /// [`scheduled_time`](ScheduledWorkflowInput::scheduled_time). **Refused inside a workflow**:
    /// its run id is generated from the clock, so a replay would enqueue a second run.
    pub async fn trigger_schedule<R, E>(&self, name: &str) -> Result<WorkflowHandle<R, E>> {
        const OPERATION: &str = "trigger a schedule";
        let executor = self.executor(OPERATION)?;
        refuse_inside_workflow(OPERATION)?;
        executor
            .connection()
            .trigger_schedule(name, Some(executor.workflows()))
            .await
    }
}

/// [`Error::InvalidArgument`] for a workflow this executor has no registration for.
///
/// On [`DBOS`] only: a [`Client`] runs no workflows, and may well be scheduling one for an
/// application written in another language.
fn refuse_unregistered(
    executor: &crate::Executor,
    workflow: &QualifiedWorkflowName,
    operation: &'static str,
) -> Result<()> {
    if executor.workflows().contains_key(workflow) {
        return Ok(());
    }
    Err(Error::InvalidArgument {
        operation: operation.into(),
        detail: format!("no workflow is registered as {workflow}"),
    })
}

/// [`Error::InvalidArgument`] for a queue nobody has registered.
///
/// Read rather than taken on faith, because a tick enqueued on a queue with no row is never
/// dequeued. On [`DBOS`] only, like [`refuse_unregistered`]: a client may be scheduling onto a
/// queue its application registers later.
async fn refuse_unknown_queue(
    conn: &Connection,
    schedule_name: &str,
    queue: Option<&str>,
    operation: &'static str,
) -> Result<()> {
    let Some(queue) = queue else {
        return Ok(());
    };
    if queue == INTERNAL_QUEUE || conn.queue(queue).await?.is_some() {
        return Ok(());
    }
    Err(Error::InvalidArgument {
        operation: operation.into(),
        detail: format!(
            "schedule `{schedule_name}`: queue `{queue}` is not registered; register it before \
             scheduling onto it"
        ),
    })
}

/// The same surface for a process outside the application.
///
/// [`DBOS`]'s methods with a client's two differences: nothing is checkpointed, and nothing checks
/// the workflow or the queue against a registration, because a client has neither — see
/// [`ScheduleSpec::for_workflow`]. [`apply_schedules`](Self::apply_schedules),
/// [`backfill_schedule`](Self::backfill_schedule) and [`trigger_schedule`](Self::trigger_schedule)
/// are not refused inside a workflow either: a client's call is never recorded, so there is
/// nothing for a replay to disagree with.
impl Client {
    /// Creates a schedule. See [`DBOS::create_schedule`].
    pub async fn create_schedule(&self, spec: &ScheduleSpec) -> Result<()> {
        spec.validate("create a schedule")?;
        self.connection().create_schedule(spec, None).await
    }

    /// Creates or replaces a set of schedules, in one transaction. See [`DBOS::apply_schedules`].
    pub async fn apply_schedules(&self, specs: &[ScheduleSpec]) -> Result<()> {
        for spec in specs {
            spec.validate("apply schedules")?;
        }
        self.connection().apply_schedules(specs).await
    }

    /// The schedules matching `filter`. See [`DBOS::list_schedules`].
    pub async fn list_schedules(&self, filter: &ScheduleFilter<'_>) -> Result<Vec<Schedule>> {
        let conn = self.connection();
        let records = conn.list_schedules(filter, None).await?;
        Ok(records
            .into_iter()
            .map(|record| Schedule::from_record(record, conn.serializer()))
            .collect())
    }

    /// The schedule with this name, or `None`.
    pub async fn get_schedule(&self, name: &str) -> Result<Option<Schedule>> {
        let conn = self.connection();
        let record = conn.get_schedule(name, None).await?;
        Ok(record.map(|record| Schedule::from_record(record, conn.serializer())))
    }

    /// Changes a schedule's definition. See [`DBOS::update_schedule`].
    pub async fn update_schedule(&self, name: &str, change: &ScheduleChange<'_>) -> Result<()> {
        change.validate(name, "update a schedule")?;
        let context = change.encode_context(self.connection().serializer())?;
        self.connection()
            .update_schedule(name, change, context.as_deref(), None)
            .await
    }

    /// Stops a schedule firing. See [`DBOS::pause_schedule`].
    pub async fn pause_schedule(&self, name: &str) -> Result<()> {
        self.connection()
            .set_schedule_status(name, ScheduleStatus::Paused, None)
            .await
    }

    /// Lets a paused schedule fire again.
    pub async fn resume_schedule(&self, name: &str) -> Result<()> {
        self.connection()
            .set_schedule_status(name, ScheduleStatus::Active, None)
            .await
    }

    /// Deletes a schedule. Deleting one that does not exist is not an error.
    pub async fn delete_schedule(&self, name: &str) -> Result<()> {
        self.connection().delete_schedule(name, None).await
    }

    /// Enqueues every tick of a schedule strictly between `start` and `end`. See
    /// [`DBOS::backfill_schedule`].
    pub async fn backfill_schedule<R, E>(
        &self,
        name: &str,
        start: SystemTime,
        end: SystemTime,
    ) -> Result<Vec<WorkflowHandle<R, E>>> {
        self.connection()
            .backfill_schedule(name, start, end, None)
            .await
    }

    /// Enqueues one run of a schedule's workflow now. See [`DBOS::trigger_schedule`].
    pub async fn trigger_schedule<R, E>(&self, name: &str) -> Result<WorkflowHandle<R, E>> {
        self.connection().trigger_schedule(name, None).await
    }
}

impl Connection {
    pub(crate) async fn create_schedule(
        &self,
        spec: &ScheduleSpec,
        caller: Option<StepWriter<'_>>,
    ) -> Result<()> {
        let context = spec.encode_context(self.serializer())?;
        self.sysdb()
            .create_schedule(&spec.as_new(&context), self.step_caller(caller))
            .await
            .map_err(Error::from_sysdb)?;
        tracing::info!(schedule = spec.schedule_name, "created a schedule");
        Ok(())
    }

    pub(crate) async fn apply_schedules(&self, specs: &[ScheduleSpec]) -> Result<()> {
        let contexts = specs
            .iter()
            .map(|spec| spec.encode_context(self.serializer()))
            .collect::<Result<Vec<_>>>()?;
        let rows: Vec<NewSchedule<'_>> = specs
            .iter()
            .zip(&contexts)
            .map(|(spec, context)| spec.as_new(context))
            .collect();
        self.sysdb()
            .apply_schedules(&rows)
            .await
            .map_err(Error::from_sysdb)?;
        tracing::info!(schedules = specs.len(), "applied schedules");
        Ok(())
    }

    pub(crate) async fn list_schedules(
        &self,
        filter: &ScheduleFilter<'_>,
        caller: Option<StepWriter<'_>>,
    ) -> Result<Vec<ScheduleRecord>> {
        self.sysdb()
            .list_schedules(filter, self.step_caller(caller))
            .await
            .map_err(Error::from_sysdb)
    }

    pub(crate) async fn get_schedule(
        &self,
        name: &str,
        caller: Option<StepWriter<'_>>,
    ) -> Result<Option<ScheduleRecord>> {
        self.sysdb()
            .get_schedule(name, self.step_caller(caller))
            .await
            .map_err(Error::from_sysdb)
    }

    /// Writes a validated change. `context` is the change's context, already encoded.
    pub(crate) async fn update_schedule(
        &self,
        name: &str,
        change: &ScheduleChange<'_>,
        context: Option<&str>,
        caller: Option<StepWriter<'_>>,
    ) -> Result<()> {
        let update = ScheduleUpdate {
            schedule: change.schedule,
            context: context.map_or(Change::Leave, Change::Set),
            automatic_backfill: change.automatic_backfill,
            cron_timezone: change.cron_timezone,
            queue_name: change.queue_name,
        };
        self.sysdb()
            .update_schedule(name, &update, self.step_caller(caller))
            .await
            .map_err(Error::from_sysdb)?;
        tracing::info!(schedule = name, "updated a schedule");
        Ok(())
    }

    pub(crate) async fn set_schedule_status(
        &self,
        name: &str,
        status: ScheduleStatus,
        caller: Option<StepWriter<'_>>,
    ) -> Result<()> {
        self.sysdb()
            .set_schedule_status(name, status, self.step_caller(caller))
            .await
            .map_err(Error::from_sysdb)?;
        tracing::info!(
            schedule = name,
            status = status.as_str(),
            "set a schedule's status"
        );
        Ok(())
    }

    pub(crate) async fn delete_schedule(
        &self,
        name: &str,
        caller: Option<StepWriter<'_>>,
    ) -> Result<()> {
        self.sysdb()
            .delete_schedule(name, self.step_caller(caller))
            .await
            .map_err(Error::from_sysdb)?;
        tracing::info!(schedule = name, "deleted a schedule");
        Ok(())
    }

    /// Enqueues every tick strictly between `start` and `end` that has not already run.
    ///
    /// `self: &Arc<Self>` because the handles poll through this connection. `workflows` is the
    /// caller's registry, which [`prepare_firing`](Self::prepare_firing) reads; a client has none.
    pub(crate) async fn backfill_schedule<R, E>(
        self: &Arc<Self>,
        name: &str,
        start: SystemTime,
        end: SystemTime,
        workflows: Option<&Snapshot>,
    ) -> Result<Vec<WorkflowHandle<R, E>>> {
        let record = self.get_schedule_or_not_found(name).await?;
        let cron = CronSchedule::parse(&record.schedule, record.cron_timezone.as_deref()).map_err(
            |detail| Error::InvalidArgument {
                operation: "backfill a schedule".into(),
                detail: format!("schedule `{name}`: {detail}"),
            },
        )?;
        let end = to_jiff(end)?;
        let mut cursor = to_jiff(start)?;
        // Once for the whole window: the owner's version and the context are the same for every
        // tick, and a long window would otherwise read them once per tick.
        let firing = self.prepare_firing(&record, workflows).await?;
        let mut handles = Vec::new();
        let mut first_failure = None;
        while let Some(tick) = cron.next_after(cursor) {
            cursor = tick.timestamp();
            if cursor >= end {
                break;
            }
            let workflow_id = firing_id(name, &tick);
            // **On past a tick that fails**, so one bad tick does not leave the rest of the window
            // unfilled. The call still fails, with the first error, once the window is walked; a
            // second call over the same window enqueues only what is still missing.
            match self
                .fire_unless_fired(&record, &firing, cursor, &workflow_id)
                .await
            {
                Ok(_) => handles.push(WorkflowHandle::polling(Arc::clone(self), workflow_id, true)),
                Err(error) => {
                    tracing::warn!(workflow_id, error = %error, "could not backfill a tick");
                    first_failure.get_or_insert(error);
                }
            }
        }
        tracing::info!(
            schedule = name,
            ticks = handles.len(),
            "backfilled a schedule"
        );
        match first_failure {
            Some(error) => Err(error),
            None => Ok(handles),
        }
    }

    /// Enqueues one run now, under an id of its own.
    pub(crate) async fn trigger_schedule<R, E>(
        self: &Arc<Self>,
        name: &str,
        workflows: Option<&Snapshot>,
    ) -> Result<WorkflowHandle<R, E>> {
        let record = self.get_schedule_or_not_found(name).await?;
        let now = jiff::Timestamp::now();
        let workflow_id = trigger_id(name, now);
        let firing = self.prepare_firing(&record, workflows).await?;
        self.enqueue_run(&record, &firing, SystemTime::from(now), &workflow_id)
            .await?;
        tracing::info!(schedule = name, workflow_id, "triggered a schedule");
        Ok(WorkflowHandle::polling(Arc::clone(self), workflow_id, true))
    }

    /// The schedule, or the system database's `NotRegistered` for a name with none.
    async fn get_schedule_or_not_found(&self, name: &str) -> Result<ScheduleRecord> {
        self.get_schedule(name, None).await?.ok_or_else(|| {
            Error::SystemDatabase(crate::sysdb::Error::NotRegistered {
                kind: "Schedule".into(),
                name: name.to_owned(),
            })
        })
    }

    /// Enqueues the tick at `at` under `workflow_id`, unless a workflow already has that id, and
    /// says whether it did.
    ///
    /// Read first, so a tick another executor already enqueued costs a read rather than an
    /// insert. The insert would be harmless anyway — an `ENQUEUED` row taken again is left as it
    /// was — but an executor fleet all waking for the same tick is the common case, and this keeps
    /// it to one write. Two executors can both read nothing and both insert, so `true` means this
    /// call wrote the row or tied for it.
    pub(crate) async fn fire_unless_fired(
        &self,
        record: &ScheduleRecord,
        firing: &Firing,
        at: jiff::Timestamp,
        workflow_id: &str,
    ) -> Result<bool> {
        let existing = self
            .sysdb()
            .get_workflow(workflow_id)
            .await
            .map_err(Error::from_sysdb)?;
        if existing.is_some() {
            return Ok(false);
        }
        self.enqueue_run(record, firing, SystemTime::from(at), workflow_id)
            .await?;
        Ok(true)
    }

    /// Enqueues one run of a schedule's workflow, called for `scheduled_time`.
    ///
    /// What a scheduled run is written as:
    ///
    /// - **`ENQUEUED`**, on the schedule's queue or the internal one, with nobody claiming it.
    /// - **The owner's latest application version.** A schedule is fired by every executor of its
    ///   application, of whatever version, and the run should go to the newest: an executor still
    ///   on the old code during a deploy enqueues work only the new code dequeues.
    /// - **`application_name`** the schedule's owner, or this handle's for an unclaimed schedule.
    /// - **`schedule_name`**, which is what lists a schedule's runs.
    /// - **No parent and no creator**: nothing called this, so nothing records having called it.
    async fn enqueue_run(
        &self,
        record: &ScheduleRecord,
        firing: &Firing,
        scheduled_time: SystemTime,
        workflow_id: &str,
    ) -> Result<()> {
        // As a start through the workflow's own reference would: its registration's choice, or the
        // configured serializer for one it did not make.
        let encoding = PayloadEncoding::resolve(firing.serialization, self.serializer().encoding());
        let input = encode_input(
            &ScheduledWorkflowInput {
                scheduled_time,
                context: &firing.context,
            },
            &encoding,
        )?;
        let queue = QueueRequest::new(record.queue_name.as_deref().unwrap_or(INTERNAL_QUEUE));
        let new = NewWorkflow {
            name: Some(firing.workflow.name()),
            class_name: firing.workflow.class_name(),
            input: Some(&input),
            serialization: Some(encoding.name()),
            executor_id: None,
            application_name: firing.owner.as_deref(),
            application_version: firing.version.as_deref(),
            schedule_name: Some(&record.schedule_name),
            ..new_workflow(workflow_id, Some(&queue))
        };
        // A tick's id is derived from the schedule and the time, so a tick another executor
        // already enqueued is joined rather than refused.
        init_or_join(
            self,
            &new,
            DuplicationPolicy::Reject,
            WorkflowIdReusePolicy::ReturnExisting,
            None,
        )
        .await?;
        tracing::debug!(
            schedule = record.schedule_name,
            workflow_id,
            "enqueued a scheduled run"
        );
        Ok(())
    }
}

/// What every run of a schedule shares, read once rather than per tick.
pub(crate) struct Firing {
    /// The application the runs belong to: the schedule's owner, or the firing handle's for an
    /// unclaimed schedule.
    owner: Option<String>,
    /// The owner's latest application version.
    version: Option<String>,
    /// The schedule's context, decoded.
    context: serde_json::Value,
    /// How the workflow is registered to encode, or [`SerializationFormat::Default`] where the
    /// firing handle has no registration to read: a client, or an executor without the workflow.
    serialization: SerializationFormat,
    /// The workflow each run starts, read from the schedule's columns, so a class stored as `''`
    /// is absent.
    workflow: QualifiedWorkflowName,
}

impl Connection {
    /// Reads what [`enqueue_run`](Self::enqueue_run) stamps on a run of `record`.
    ///
    /// **The latest version is read at the time of firing**, not at the time the schedule was
    /// loaded: a deploy that lands while a schedule is running should get that schedule's next
    /// tick. The loop refreshes this at most once a second, so a long catch-up does not read it
    /// once per tick either.
    ///
    /// `workflows` is the firing executor's registry, which says how the workflow is registered to
    /// encode; a client passes `None`, and its runs take the configured serializer.
    pub(crate) async fn prepare_firing(
        &self,
        record: &ScheduleRecord,
        workflows: Option<&Snapshot>,
    ) -> Result<Firing> {
        let owner = record
            .application_name
            .as_deref()
            .or(self.app_name())
            .map(str::to_owned);
        // `None` when the owner has registered no version — a schedule created by a client before
        // its application ever launched. An unversioned row is dequeued by whichever executor is
        // on the latest version once there is one, which is what stamping the latest would mean.
        let version = self
            .sysdb()
            .get_latest_application_version(owner.as_deref())
            .await
            .map_err(Error::from_sysdb)?
            .map(|version| version.version_name);
        // Decoded to a value and encoded again inside each input, rather than spliced in as text,
        // so a context that cannot be read fails here rather than in every run.
        let context = decode_context(&record.context, self.serializer())?;
        let workflow = QualifiedWorkflowName::from_row(
            record.workflow_name.as_str(),
            record.workflow_class_name.as_deref(),
            None,
        );
        let serialization = workflows
            .and_then(|workflows| workflows.get(&workflow))
            .map_or(SerializationFormat::Default, |registered| {
                registered.serialization
            });
        Ok(Firing {
            owner,
            version,
            context,
            serialization,
            workflow,
        })
    }
}

/// Decodes a schedule's context with the configured serializer.
///
/// The column carries no tag to say how the context was encoded, so it is read as the configured
/// serializer would have written it, whoever did.
///
/// **Read as the JSON data it was written as**, a `serde_json::Value`, and only then as `C`, so a
/// type whose form depends on whether the format is human-readable reads back as it was written.
fn decode_context<C: DeserializeOwned>(context: &str, configured: &Serializer) -> Result<C> {
    const WHAT: &str = "schedule context";
    let value: serde_json::Value = decode(Some(context), &configured.encoding(), WHAT)?;
    serde_json::from_value(value).map_err(|source| Error::decode_failed(WHAT, source))
}

/// A `SystemTime` as the calendar library's instant.
fn to_jiff(time: SystemTime) -> Result<jiff::Timestamp> {
    jiff::Timestamp::try_from(time).map_err(|error| Error::InvalidArgument {
        operation: "backfill a schedule".into(),
        detail: format!("{time:?} is outside the supported range: {error}"),
    })
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use super::*;
    use crate::error::EngineOnly;

    #[test]
    fn the_input_encodes_its_tick_as_rfc3339_text() {
        let input = ScheduledWorkflowInput {
            scheduled_time: SystemTime::UNIX_EPOCH + Duration::from_secs(1_759_665_600),
            context: "emea",
        };
        let encoded =
            encode::<_, EngineOnly>(&input, &PayloadEncoding::Native, "argument").unwrap();
        assert_eq!(
            encoded,
            r#"{"scheduled_time":"2025-10-05T12:00:00Z","context":"emea"}"#
        );
        let decoded: ScheduledWorkflowInput<String> =
            decode::<_, EngineOnly>(Some(&encoded), &PayloadEncoding::Native, "argument").unwrap();
        assert_eq!(decoded.scheduled_time, input.scheduled_time);
        assert_eq!(decoded.context, "emea");
    }

    /// A schedule's context goes through a `serde_json::Value`, which can't hold a non-finite
    /// float, so one is refused rather than stored as `null`, whichever way the context is written.
    #[test]
    fn a_non_finite_float_in_a_context_is_refused() {
        let key = QualifiedWorkflowName::new("report");
        assert!(ScheduleSpec::for_workflow("nan", key.clone(), "0 * * * * *", &f64::NAN).is_err());
        assert!(ScheduleSpec::for_workflow("ok", key, "0 * * * * *", &1.5f64).is_ok());

        let input = ScheduledWorkflowInput {
            scheduled_time: SystemTime::UNIX_EPOCH,
            context: f64::INFINITY,
        };
        assert!(encode::<_, EngineOnly>(&input, &PayloadEncoding::Native, "argument").is_err());
    }

    #[test]
    fn a_schedule_reads_back_everything_its_row_holds() {
        let schedule = Schedule::from_record(
            ScheduleRecord {
                schedule_id: "id-1".into(),
                schedule_name: "nightly".into(),
                workflow_name: "report".into(),
                workflow_class_name: Some("Reports".into()),
                schedule: "0 0 2 * * *".into(),
                status: ScheduleStatus::Paused,
                context: r#""emea""#.into(),
                last_fired_at: Some(Timestamp::from_epoch_ms(1_759_665_600_000)),
                automatic_backfill: true,
                cron_timezone: Some("Europe/Paris".into()),
                queue_name: Some("reports".into()),
                application_name: Some("app".into()),
            },
            &Serializer::RustSerde,
        );
        assert_eq!(schedule.id(), "id-1");
        assert_eq!(schedule.schedule_name(), "nightly");
        assert_eq!(
            schedule.workflow(),
            &QualifiedWorkflowName::method("report", "Reports")
        );
        assert_eq!(schedule.schedule(), "0 0 2 * * *");
        assert_eq!(schedule.status(), ScheduleStatus::Paused);
        assert_eq!(schedule.context(), r#""emea""#);
        assert_eq!(schedule.decode_context::<String>().unwrap(), "emea");
        assert_eq!(
            schedule.last_fired_at(),
            Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1_759_665_600))
        );
        assert!(schedule.automatic_backfill());
        assert_eq!(schedule.cron_timezone(), Some("Europe/Paris"));
        assert_eq!(schedule.queue_name(), Some("reports"));
        assert_eq!(schedule.app_name(), Some("app"));
    }

    /// JSON, reversed: text only this serializer reads.
    #[derive(Debug)]
    struct Backwards;

    impl crate::CustomSerializer for Backwards {
        fn name(&self) -> &str {
            "backwards_json"
        }

        fn serialize(
            &self,
            value: crate::Payload<'_>,
        ) -> std::result::Result<String, Box<dyn std::error::Error + Send + Sync>> {
            Ok(serde_json::to_string(&value)?.chars().rev().collect())
        }

        fn deserialize(
            &self,
            text: &str,
            into: crate::PayloadSlot<'_>,
        ) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> {
            let json: String = text.chars().rev().collect();
            let mut deserializer = serde_json::Deserializer::from_str(&json);
            into.fill(&mut deserializer)?;
            Ok(deserializer.end()?)
        }
    }

    /// MessagePack, as hex: a format that is not human-readable, as an application's binary
    /// serializer would be.
    #[derive(Debug)]
    struct MessagePack;

    impl crate::CustomSerializer for MessagePack {
        fn name(&self) -> &str {
            "msgpack_hex"
        }

        fn serialize(
            &self,
            value: crate::Payload<'_>,
        ) -> std::result::Result<String, Box<dyn std::error::Error + Send + Sync>> {
            let bytes = rmp_serde::to_vec(&value)?;
            Ok(bytes.iter().map(|byte| format!("{byte:02x}")).collect())
        }

        fn deserialize(
            &self,
            text: &str,
            into: crate::PayloadSlot<'_>,
        ) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> {
            let bytes = (0..text.len())
                .step_by(2)
                .map(|at| u8::from_str_radix(&text[at..at + 2], 16))
                .collect::<std::result::Result<Vec<u8>, _>>()?;
            into.fill(&mut rmp_serde::Deserializer::new(bytes.as_slice()))
        }
    }

    /// An `Ipv4Addr` is a string in JSON and four octets in MessagePack. The context is stored as its
    /// JSON data, so it is read back as that, directly and inside a scheduled run's input, rather
    /// than as the octets a MessagePack reader of `Ipv4Addr` would expect.
    #[test]
    fn a_context_reads_back_through_a_format_that_is_not_human_readable() {
        let binary = Serializer::custom(MessagePack);
        let address = std::net::Ipv4Addr::new(10, 0, 0, 1);
        let spec = ScheduleSpec::for_workflow("s", "w", "* * * * *", &address).unwrap();
        let stored = spec.encode_context(&binary).unwrap();

        // Read as `Ipv4Addr` straight from MessagePack, the stored string is not what it expects.
        assert!(
            decode::<std::net::Ipv4Addr, crate::error::EngineOnly>(
                Some(&stored),
                &binary.encoding(),
                "context"
            )
            .is_err()
        );
        assert_eq!(
            decode_context::<std::net::Ipv4Addr>(&stored, &binary).unwrap(),
            address
        );

        // A run is called with the context the tick read, as a value, encoded in its input.
        let fired: serde_json::Value = decode_context(&stored, &binary).unwrap();
        let input = crate::serialization::encode_input::<_, crate::error::EngineOnly>(
            &ScheduledWorkflowInput {
                scheduled_time: SystemTime::UNIX_EPOCH,
                context: fired,
            },
            &binary.encoding(),
        )
        .unwrap();
        let read: ScheduledWorkflowInput<std::net::Ipv4Addr> =
            crate::serialization::decode_input::<_, crate::error::EngineOnly>(
                Some(&input),
                &binary.encoding(),
            )
            .unwrap();
        assert_eq!(read.context, address);

        // And an input built with the typed context reads back the same.
        let typed = crate::serialization::encode_input::<_, crate::error::EngineOnly>(
            &ScheduledWorkflowInput {
                scheduled_time: SystemTime::UNIX_EPOCH,
                context: address,
            },
            &binary.encoding(),
        )
        .unwrap();
        assert_eq!(typed, input);
    }

    /// A context carries no tag, so it is read only by the serializer that wrote it.
    #[test]
    fn a_context_is_read_by_the_configured_serializer_alone() {
        let custom = Serializer::custom(Backwards);
        let spec = ScheduleSpec::for_workflow("s", "w", "* * * * *", &[1, 2]).unwrap();
        let sealed = spec.encode_context(&custom).unwrap();
        assert_eq!(sealed, "]2,1[");
        let plain = spec.encode_context(&Serializer::RustSerde).unwrap();
        assert_eq!(plain, "[1,2]");

        assert_eq!(decode_context::<Vec<u8>>(&sealed, &custom).unwrap(), [1, 2]);
        assert!(decode_context::<Vec<u8>>(&plain, &custom).is_err());
        assert_eq!(
            decode_context::<Vec<u8>>(&plain, &Serializer::RustSerde).unwrap(),
            [1, 2]
        );
        assert!(decode_context::<Vec<u8>>(&sealed, &Serializer::RustSerde).is_err());
    }

    #[test]
    fn an_empty_class_name_reads_as_none() {
        let schedule = Schedule::from_record(
            ScheduleRecord {
                schedule_id: "id".into(),
                schedule_name: "s".into(),
                workflow_name: "w".into(),
                workflow_class_name: Some(String::new()),
                schedule: "* * * * *".into(),
                status: ScheduleStatus::Active,
                context: "null".into(),
                last_fired_at: None,
                automatic_backfill: false,
                cron_timezone: None,
                queue_name: None,
                application_name: None,
            },
            &Serializer::RustSerde,
        );
        assert_eq!(schedule.workflow(), &QualifiedWorkflowName::new("w"));
        assert_eq!(schedule.last_fired_at(), None);
    }

    /// Two handles, each with its own instance of a serializer, see the same schedule.
    #[test]
    fn views_of_one_row_are_equal_whatever_reads_them() {
        let record = ScheduleRecord {
            schedule_id: "id".into(),
            schedule_name: "s".into(),
            workflow_name: "w".into(),
            workflow_class_name: None,
            schedule: "* * * * *".into(),
            status: ScheduleStatus::Active,
            context: "null".into(),
            last_fired_at: None,
            automatic_backfill: false,
            cron_timezone: None,
            queue_name: None,
            application_name: None,
        };
        let one = Schedule::from_record(record.clone(), &Serializer::custom(Backwards));
        let other = Schedule::from_record(record.clone(), &Serializer::custom(Backwards));
        assert_eq!(one, other);
        assert_eq!(
            one,
            Schedule::from_record(record.clone(), &Serializer::RustSerde)
        );
        let changed = ScheduleRecord {
            context: "1".into(),
            ..record
        };
        assert_ne!(one, Schedule::from_record(changed, &Serializer::RustSerde));
    }

    #[test]
    fn a_spec_refuses_what_no_executor_could_fire() {
        let spec = |cron: &str, timezone: Option<&str>| ScheduleSpec {
            cron_timezone: timezone.map(str::to_owned),
            ..ScheduleSpec::for_workflow("s", "w", cron, &()).unwrap()
        };
        assert!(spec("* * * * *", None).validate("test").is_ok());
        for (cron, timezone, expected) in [
            ("not a cron", None, "invalid cron schedule"),
            ("0 0 31 2 *", None, "never fires"),
            ("* * * * *", Some("Nowhere/Special"), "invalid timezone"),
        ] {
            let error = spec(cron, timezone).validate("test").unwrap_err();
            assert!(error.to_string().contains(expected), "{cron}: {error}");
        }

        let instance = ScheduleSpec::for_workflow(
            "s",
            QualifiedWorkflowName::instance("w", "Class", "config"),
            "* * * * *",
            &(),
        )
        .unwrap();
        assert!(
            instance
                .validate("test")
                .unwrap_err()
                .to_string()
                .contains("configured instance")
        );
    }
}