sova_tasks/
job.rs

1//! Job definition: handler + optional schedule / queue / priority.
2
3use crate::{BoxFuture, Handler};
4use cron::Schedule as CronSchedule;
5use sova_tasks_store::Task;
6use serde_json::Value;
7use std::str::FromStr;
8use std::sync::Arc;
9use std::time::Duration;
10
11/// Common priority levels (higher is claimed first).
12pub mod priority {
13    pub const LOW: i32 = -100;
14    pub const NORMAL: i32 = 0;
15    pub const HIGH: i32 = 100;
16}
17
18/// When a [`Job`] should be enqueued by the scheduler.
19#[derive(Clone, Debug)]
20#[allow(clippy::large_enum_variant)] // CronSchedule is large; Every is tiny
21pub enum Schedule {
22    Every(Duration),
23    Cron(CronSchedule),
24}
25
26/// Named handler with optional recurring schedule, queue, and priority.
27pub struct Job {
28    pub(crate) name: String,
29    pub(crate) handler: Handler,
30    pub(crate) schedule: Option<Schedule>,
31    pub(crate) payload: Value,
32    pub(crate) queue: Option<String>,
33    pub(crate) priority: i32,
34}
35
36impl Job {
37    pub fn new<F, Fut>(name: impl Into<String>, f: F) -> Self
38    where
39        F: Fn(Task) -> Fut + Send + Sync + 'static,
40        Fut: std::future::Future<Output = Result<(), String>> + Send + 'static,
41    {
42        Self {
43            name: name.into(),
44            handler: Arc::new(move |t| Box::pin(f(t)) as BoxFuture<Result<(), String>>),
45            schedule: None,
46            payload: Value::Object(Default::default()),
47            queue: None,
48            priority: priority::NORMAL,
49        }
50    }
51
52    /// Named queue for schedule enqueue and default dispatch routing.
53    pub fn queue(mut self, q: impl Into<String>) -> Self {
54        self.queue = Some(q.into());
55        self
56    }
57
58    /// Claim priority within the queue (higher first). Default [`priority::NORMAL`].
59    pub fn priority(mut self, p: i32) -> Self {
60        self.priority = p;
61        self
62    }
63
64    /// Run approximately every `period` (slot-based dedup across instances).
65    pub fn every(mut self, period: Duration) -> Self {
66        assert!(!period.is_zero(), "Job::every period must be non-zero");
67        self.schedule = Some(Schedule::Every(period));
68        self
69    }
70
71    /// Cron expression: 5 fields (`min hour day month dow`) or 6 (`sec min …`).
72    pub fn cron(mut self, expr: impl AsRef<str>) -> Self {
73        let sched = parse_cron(expr.as_ref())
74            .unwrap_or_else(|e| panic!("Job::cron({}): {e}", expr.as_ref()));
75        self.schedule = Some(Schedule::Cron(sched));
76        self
77    }
78
79    /// JSON `data` field used when the scheduler enqueues this job.
80    pub fn payload(mut self, data: Value) -> Self {
81        self.payload = data;
82        self
83    }
84}
85
86/// Pad 5-field crontab to 6-field (seconds = 0) for the `cron` crate.
87pub fn parse_cron(expr: &str) -> Result<CronSchedule, String> {
88    let trimmed = expr.trim();
89    let fields = trimmed.split_whitespace().count();
90    let padded = if fields == 5 {
91        format!("0 {trimmed}")
92    } else {
93        trimmed.to_string()
94    };
95    CronSchedule::from_str(&padded).map_err(|e| e.to_string())
96}
97
98#[cfg(test)]
99mod tests {
100    use super::*;
101
102    #[test]
103    fn pads_five_field_cron() {
104        let s = parse_cron("0 * * * *").unwrap();
105        assert!(s.upcoming(chrono::Utc).next().is_some());
106    }
107
108    #[test]
109    #[should_panic(expected = "Job::cron")]
110    fn bad_cron_panics() {
111        let _ = Job::new("x", |_| async { Ok(()) }).cron("not a cron");
112    }
113}
sova_tasks/job.rs

sova_tasks/
job.rs
