chronon_scheduler/
job_builder.rs

1//! Fluent builder for constructing [`Job`](chronon_core::Job) rows (preferred schedule entry point).
2//!
3//! Owns cron/run-once/manual scheduling, params, pool/region, retry/misfire policy, and optional
4//! opaque [`Job::actor_json`](chronon_core::Job::actor_json). Does **not** own Valence identity or
5//! persistence — hosts snapshot actors and upsert via `CoordinatorService`,
6//! `RemoteCoordinatorClient`, or an L1 coordinator backend.
7//!
8//! Prefer this over seeding with [`ScriptHandle::job`](chronon_core::ScriptHandle::job) /
9//! [`job_with_params`](chronon_core::ScriptHandle::job_with_params) and mutating schedule fields
10//! by hand.
11//!
12//! # Examples
13//!
14//! ```
15//! use chronon_core::ScriptHandle;
16//! use chronon_scheduler::JobBuilder;
17//! use serde::Serialize;
18//!
19//! #[derive(Serialize)]
20//! struct NightlyParams {
21//!     retention_days: u32,
22//! }
23//!
24//! let handle = ScriptHandle::<NightlyParams>::new("nightly_cleanup");
25//! let job = JobBuilder::new(&handle)
26//!     .name("nightly-cleanup")
27//!     .cron("0 0 * * * *")?
28//!     .timezone("UTC")
29//!     .params(NightlyParams {
30//!         retention_days: 7,
31//!     })
32//!     .with_actor_json(serde_json::json!({ "Service": { "name": "ops" } }))
33//!     .build()?;
34//!
35//! assert_eq!(job.script_name, "nightly_cleanup");
36//! assert!(job.next_run_at.is_some());
37//! # Ok::<(), chronon_core::ChrononError>(())
38//! ```
39
40use chrono::{DateTime, Utc};
41use chronon_core::{
42    ChrononError, Job, MisfirePolicy, Result, RetryPolicy, ScheduleKind, ScriptHandle,
43};
44use serde::Serialize;
45use serde_json::Value;
46
47use crate::CronExpr;
48
49/// Fluent builder for a scheduled [`Job`].
50///
51/// Requires [`Self::name`] before [`Self::build`]. Identity is optional via
52/// [`Self::with_actor_json`] (default remains [`Job::new`](chronon_core::Job::new)'s `Null`).
53///
54/// # Errors
55///
56/// Fallible setters and [`Self::build`] return [`ChrononError`] variants documented on each method.
57/// Error messages must not embed full `actor_json` or `params_json` payloads.
58#[must_use = "build the Job with JobBuilder::build"]
59pub struct JobBuilder<P> {
60    script_name: &'static str,
61    job_name: Option<String>,
62    actor_json: Option<Value>,
63    params: Option<P>,
64    cron_expr: Option<String>,
65    timezone: Option<String>,
66    run_once_at: Option<DateTime<Utc>>,
67    schedule_kind: ScheduleKind,
68    enabled: bool,
69    pool: Option<String>,
70    region: Option<String>,
71    concurrency: i32,
72    timeout_ms: Option<i64>,
73    retry_policy: RetryPolicy,
74    misfire_policy: MisfirePolicy,
75}
76
77impl<P> JobBuilder<P>
78where
79    P: Serialize,
80{
81    /// Create a new job builder for the given script handle.
82    pub fn new(handle: &ScriptHandle<P>) -> Self {
83        Self {
84            script_name: handle.name(),
85            job_name: None,
86            actor_json: None,
87            params: None,
88            cron_expr: None,
89            timezone: None,
90            run_once_at: None,
91            schedule_kind: ScheduleKind::Cron,
92            enabled: true,
93            pool: None,
94            region: None,
95            concurrency: 1,
96            timeout_ms: None,
97            retry_policy: RetryPolicy::default(),
98            misfire_policy: MisfirePolicy::default(),
99        }
100    }
101
102    /// Set the job name (unique per deployment).
103    pub fn name(mut self, name: impl Into<String>) -> Self {
104        self.job_name = Some(name.into());
105        self
106    }
107
108    /// Set opaque identity JSON persisted on [`Job::actor_json`](chronon_core::Job::actor_json).
109    ///
110    /// Use an identity snapshot only — do not store secrets. Unified Field hosts with Valence
111    /// should prefer L1 `chronon_coordinator::JobBuilder::with_valence` instead.
112    pub fn with_actor_json(mut self, actor_json: Value) -> Self {
113        self.actor_json = Some(actor_json);
114        self
115    }
116
117    /// Set the cron schedule.
118    ///
119    /// # Errors
120    ///
121    /// Returns [`ChrononError::InvalidCron`] when `expr` is not a valid cron expression.
122    pub fn cron(mut self, expr: &str) -> Result<Self> {
123        CronExpr::parse(expr, None)?;
124        self.cron_expr = Some(expr.to_string());
125        self.schedule_kind = ScheduleKind::Cron;
126        Ok(self)
127    }
128
129    /// Set the timezone for cron evaluation.
130    pub fn timezone(mut self, tz: impl Into<String>) -> Self {
131        self.timezone = Some(tz.into());
132        self
133    }
134
135    /// Schedule a one-time execution at the specified time.
136    pub const fn run_once_at(mut self, at: DateTime<Utc>) -> Self {
137        self.run_once_at = Some(at);
138        self.schedule_kind = ScheduleKind::RunOnce;
139        self
140    }
141
142    /// Set the job to manual-only (no automatic scheduling).
143    pub const fn manual(mut self) -> Self {
144        self.schedule_kind = ScheduleKind::Manual;
145        self
146    }
147
148    /// Set the script parameters.
149    pub fn params(mut self, params: P) -> Self {
150        self.params = Some(params);
151        self
152    }
153
154    /// Set the execution pool (for distributed mode).
155    pub fn pool(mut self, pool: impl Into<String>) -> Self {
156        self.pool = Some(pool.into());
157        self
158    }
159
160    /// Set the target region (for distributed mode).
161    pub fn region(mut self, region: impl Into<String>) -> Self {
162        self.region = Some(region.into());
163        self
164    }
165
166    /// Set maximum concurrent runs.
167    pub const fn concurrency(mut self, max: i32) -> Self {
168        self.concurrency = max;
169        self
170    }
171
172    /// Set execution timeout in milliseconds.
173    pub const fn timeout_ms(mut self, ms: i64) -> Self {
174        self.timeout_ms = Some(ms);
175        self
176    }
177
178    /// Set the retry policy for failed runs.
179    pub const fn retry_policy(mut self, policy: RetryPolicy) -> Self {
180        self.retry_policy = policy;
181        self
182    }
183
184    /// Set the misfire policy for missed runs.
185    pub const fn misfire_policy(mut self, policy: MisfirePolicy) -> Self {
186        self.misfire_policy = policy;
187        self
188    }
189
190    /// Disable the job (it won't run until enabled).
191    pub const fn disabled(mut self) -> Self {
192        self.enabled = false;
193        self
194    }
195
196    /// Build the final [`Job`] payload.
197    ///
198    /// # Errors
199    ///
200    /// - [`ChrononError::ParamError`] when [`Self::name`] was not set
201    /// - [`ChrononError::InvalidCron`] / [`ChrononError::InvalidTimezone`] when cron fields are invalid
202    /// - [`ChrononError::ParamError`] when params fail to serialize
203    pub fn build(self) -> Result<Job> {
204        let job_name = self
205            .job_name
206            .ok_or_else(|| ChrononError::ParamError("job name is required".to_string()))?;
207
208        let params_json = match self.params {
209            Some(p) => serde_json::to_value(&p)?,
210            None => Value::Object(serde_json::Map::default()),
211        };
212
213        let actor_json = self.actor_json.unwrap_or(Value::Null);
214
215        let cron_expr = match self.schedule_kind {
216            ScheduleKind::Cron => self
217                .cron_expr
218                .as_deref()
219                .map(|expr| CronExpr::parse(expr, self.timezone.as_deref()))
220                .transpose()?,
221            ScheduleKind::Manual | ScheduleKind::RunOnce => None,
222        };
223
224        let next_run_at = match self.schedule_kind {
225            ScheduleKind::Cron => cron_expr.as_ref().and_then(CronExpr::next_from_now),
226            ScheduleKind::RunOnce => self.run_once_at,
227            ScheduleKind::Manual => None,
228        };
229
230        let mut job = Job::new(&job_name, self.script_name);
231        job.enabled = self.enabled;
232        job.schedule_kind = self.schedule_kind;
233        job.cron_expr = cron_expr.map(|cron| cron.expression().to_string());
234        job.timezone = self.timezone;
235        job.run_once_at = self.run_once_at;
236        job.pool = self.pool;
237        job.region = self.region;
238        job.actor_json = actor_json;
239        job.params_json = params_json;
240        job.concurrency = self.concurrency;
241        job.timeout_ms = self.timeout_ms;
242        job.retry_policy_json = serde_json::to_value(&self.retry_policy)?;
243        job.misfire_policy_json = serde_json::to_value(&self.misfire_policy)?;
244        job.next_run_at = next_run_at;
245        job.current_revision = 1;
246
247        Ok(job)
248    }
249}
250
251#[cfg(test)]
252mod tests {
253    use super::*;
254    use chrono::TimeZone;
255
256    #[test]
257    fn build_cron_sets_schedule_and_next_run() {
258        let handle = ScriptHandle::<()>::new("nightly_cleanup");
259        let job = JobBuilder::new(&handle)
260            .name("nightly-cleanup")
261            .cron("0 0 * * * *")
262            .expect("cron")
263            .timezone("UTC")
264            .build()
265            .expect("build");
266        assert_eq!(job.script_name, "nightly_cleanup");
267        assert_eq!(job.job_name, "nightly-cleanup");
268        assert_eq!(job.schedule_kind, ScheduleKind::Cron);
269        assert_eq!(job.cron_expr.as_deref(), Some("0 0 * * * *"));
270        assert!(job.next_run_at.is_some());
271        assert!(job.actor_json.is_null());
272    }
273
274    #[test]
275    fn build_run_once_sets_next_run_at() {
276        let at = Utc.with_ymd_and_hms(2030, 1, 1, 0, 0, 0).unwrap();
277        let handle = ScriptHandle::<()>::new("once");
278        let job = JobBuilder::new(&handle)
279            .name("once-job")
280            .run_once_at(at)
281            .build()
282            .expect("build");
283        assert_eq!(job.schedule_kind, ScheduleKind::RunOnce);
284        assert_eq!(job.next_run_at, Some(at));
285        assert!(job.cron_expr.is_none());
286    }
287
288    #[test]
289    fn build_manual_clears_automatic_schedule() {
290        let handle = ScriptHandle::<()>::new("manual");
291        let job = JobBuilder::new(&handle)
292            .name("manual-job")
293            .manual()
294            .build()
295            .expect("build");
296        assert_eq!(job.schedule_kind, ScheduleKind::Manual);
297        assert!(job.next_run_at.is_none());
298        assert!(job.cron_expr.is_none());
299    }
300
301    #[test]
302    fn build_params_serialized() {
303        #[derive(Serialize)]
304        struct Params {
305            n: u32,
306        }
307        let handle = ScriptHandle::<Params>::new("demo");
308        let job = JobBuilder::new(&handle)
309            .name("demo-job")
310            .manual()
311            .params(Params { n: 3 })
312            .build()
313            .expect("build");
314        assert_eq!(job.params_json["n"], 3);
315    }
316
317    #[test]
318    fn build_with_actor_json_round_trip() {
319        let actor = serde_json::json!({ "Service": { "name": "ops" } });
320        let handle = ScriptHandle::<()>::new("probe");
321        let job = JobBuilder::new(&handle)
322            .name("probe")
323            .manual()
324            .with_actor_json(actor.clone())
325            .build()
326            .expect("build");
327        assert_eq!(job.actor_json, actor);
328    }
329
330    #[test]
331    fn build_missing_name_is_param_error() {
332        let handle = ScriptHandle::<()>::new("probe");
333        let err = JobBuilder::new(&handle).manual().build().unwrap_err();
334        match err {
335            ChrononError::ParamError(msg) => assert!(msg.contains("job name")),
336            other => panic!("expected ParamError, got {other}"),
337        }
338    }
339
340    #[test]
341    fn cron_invalid_expr_is_invalid_cron() {
342        let handle = ScriptHandle::<()>::new("probe");
343        let Err(err) = JobBuilder::new(&handle).cron("not-a-cron") else {
344            panic!("expected InvalidCron");
345        };
346        assert!(matches!(err, ChrononError::InvalidCron(_)));
347    }
348}
chronon_scheduler/job_builder.rs

chronon_scheduler/
job_builder.rs
