runledger_runtime/catalog/
schedule_spec.rs

1use std::str::FromStr;
2
3use chrono::{DateTime, Utc};
4use cron::Schedule;
5use runledger_postgres::jobs::JOB_SCHEDULE_MAX_JITTER_SECONDS;
6use serde_json::Value;
7use uuid::Uuid;
8
9/// Catalog-owned cron schedule definition synced to `job_schedules` at startup.
10///
11/// Specs are validated before database writes. Names and cron expressions must
12/// be non-empty without surrounding whitespace, cron expressions must parse as
13/// UTC cron schedules, and `max_jitter_seconds` must be in the inclusive range
14/// `0..=`[`JOB_SCHEDULE_MAX_JITTER_SECONDS`].
15///
16/// # Example
17/// ```rust
18/// use runledger_runtime::catalog::CatalogJobScheduleSpec;
19///
20/// let payload = serde_json::json!({ "kind": "refresh" });
21/// let spec = CatalogJobScheduleSpec {
22///     name: "profiles.refresh.hourly",
23///     job_type: "profiles.refresh",
24///     cron_expr: "0 0 * * * *",
25///     payload_template: &payload,
26///     is_active: true,
27///     organization_id: None,
28///     max_jitter_seconds: 0,
29///     next_fire_at: None,
30/// };
31///
32/// assert_eq!(spec.name, "profiles.refresh.hourly");
33/// ```
34#[derive(Debug, Clone, PartialEq, Eq)]
35pub struct CatalogJobScheduleSpec<'a> {
36    /// Unique schedule name. This is the global `job_schedules.name` key and
37    /// must be non-empty without surrounding whitespace.
38    pub name: &'a str,
39    /// Catalog job type enqueued when the schedule fires. The job type must be
40    /// registered and effectively enabled in the [`super::JobCatalog`] used for
41    /// schedule registration or sync.
42    pub job_type: &'a str,
43    /// UTC cron expression without surrounding whitespace.
44    ///
45    /// On conflict, catalog sync preserves the existing `next_fire_at` cursor
46    /// when this expression is unchanged. When this expression changes, the
47    /// synced `next_fire_at` value becomes the stored cursor.
48    pub cron_expr: &'a str,
49    /// JSON payload template copied into scheduled jobs.
50    pub payload_template: &'a Value,
51    /// Desired active state after sync. Applications can map feature flags to
52    /// this value before calling catalog schedule sync.
53    pub is_active: bool,
54    /// Optional organization scope copied into jobs for a new schedule.
55    pub organization_id: Option<Uuid>,
56    /// Maximum deterministic jitter, in seconds, applied to future fire times.
57    /// Must be between `0` and [`JOB_SCHEDULE_MAX_JITTER_SECONDS`], inclusive.
58    pub max_jitter_seconds: i32,
59    /// Next UTC instant at which the schedule is due. Defaults to `Utc::now()`
60    /// when unset at sync time.
61    ///
62    /// This value is used for new rows and for existing rows whose cron
63    /// expression changed. Existing rows with the same cron expression keep
64    /// their stored cursor.
65    pub next_fire_at: Option<DateTime<Utc>>,
66}
67
68impl<'a> CatalogJobScheduleSpec<'a> {
69    pub(super) fn validate_shape(&self) -> Result<(), &'static str> {
70        if self.name.trim().is_empty() {
71            return Err("name");
72        }
73        if self.name != self.name.trim() {
74            return Err("name");
75        }
76        if self.job_type.trim().is_empty() {
77            return Err("job_type");
78        }
79        if self.cron_expr.trim().is_empty() {
80            return Err("cron_expr");
81        }
82        if self.cron_expr != self.cron_expr.trim() {
83            return Err("cron_expr");
84        }
85        if Schedule::from_str(self.cron_expr).is_err() {
86            return Err("cron_expr");
87        }
88        if self.max_jitter_seconds < 0 {
89            return Err("max_jitter_seconds");
90        }
91        if self.max_jitter_seconds > JOB_SCHEDULE_MAX_JITTER_SECONDS {
92            return Err("max_jitter_seconds");
93        }
94        Ok(())
95    }
96}
97
98/// Owned schedule spec stored on [`super::JobCatalog`].
99#[derive(Debug, Clone, PartialEq, Eq)]
100pub(super) struct StoredCatalogJobScheduleSpec {
101    pub name: String,
102    pub job_type: String,
103    pub cron_expr: String,
104    pub payload_template: Value,
105    pub is_active: bool,
106    pub organization_id: Option<Uuid>,
107    pub max_jitter_seconds: i32,
108    pub next_fire_at: Option<DateTime<Utc>>,
109}
110
111impl<'a> From<&CatalogJobScheduleSpec<'a>> for StoredCatalogJobScheduleSpec {
112    fn from(spec: &CatalogJobScheduleSpec<'a>) -> Self {
113        Self {
114            name: spec.name.to_owned(),
115            job_type: spec.job_type.to_owned(),
116            cron_expr: spec.cron_expr.to_owned(),
117            payload_template: spec.payload_template.clone(),
118            is_active: spec.is_active,
119            organization_id: spec.organization_id,
120            max_jitter_seconds: spec.max_jitter_seconds,
121            next_fire_at: spec.next_fire_at,
122        }
123    }
124}
125
126impl StoredCatalogJobScheduleSpec {
127    pub(super) fn as_spec(&self) -> CatalogJobScheduleSpec<'_> {
128        CatalogJobScheduleSpec {
129            name: &self.name,
130            job_type: &self.job_type,
131            cron_expr: &self.cron_expr,
132            payload_template: &self.payload_template,
133            is_active: self.is_active,
134            organization_id: self.organization_id,
135            max_jitter_seconds: self.max_jitter_seconds,
136            next_fire_at: self.next_fire_at,
137        }
138    }
139}
runledger_runtime/catalog/schedule_spec.rs

runledger_runtime/catalog/
schedule_spec.rs
