shared_framework/job/mod.rs
1//! Background jobs: periodic tasks, daily schedules, or both.
2//!
3//! Implement [`ServiceJob`] for each task and run it through [`JobRegistry`].
4//! [`JobType::Periodic`] runs on an interval via a Tokio task,
5//! [`JobType::Exact`] runs once per day at a [`TimeOfDay`] via the cron
6//! scheduler, and [`JobType::PeriodicAndExact`] runs both legs.
7//!
8//! Key types: [`ServiceJob`] for the task, [`JobRegistry`] for ownership and
9//! lifecycle, [`JobType`] for scheduling behavior, [`TimeOfDay`] for daily times.
10//!
11//! Use this module for maintenance work such as cleanup or aggregation.
12//!
13//! ```ignore
14//! # use std::time::Duration;
15//! # use crate::job::{JobRegistry, JobType, ServiceJob, TimeOfDay};
16//! # use crate::response::ServiceResult;
17//! struct Cleanup;
18//!
19//! #[async_trait::async_trait]
20//! impl ServiceJob for Cleanup {
21//! fn name(&self) -> &str { "cleanup" }
22//! fn job_type(&self) -> JobType { JobType::Periodic }
23//! fn period(&self) -> Option<Duration> { Some(Duration::from_secs(60)) }
24//! fn schedule(&self) -> Option<TimeOfDay> { None }
25//! async fn run(&self) -> anyhow::Result<ServiceResult<serde_json::Value>> {
26//! Ok(ServiceResult::ok("cleaned", serde_json::json!({})))
27//! }
28//! }
29//!
30//! # async fn start() -> anyhow::Result<()> {
31//! let mut registry = JobRegistry::new();
32//! registry.add_job(Cleanup);
33//! registry.start().await?;
34//! # Ok(())
35//! # }
36//! ```
37
38use std::sync::{Arc, Mutex};
39use std::time::Duration;
40use tokio_cron_scheduler::{Job, JobScheduler};
41
42use crate::response::ServiceResult;
43
44/// Scheduling behavior of a [`ServiceJob`].
45#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub enum JobType {
47 /// Runs repeatedly on the [`ServiceJob::period`] interval.
48 Periodic,
49 /// Runs once per day at the [`ServiceJob::schedule`] time.
50 Exact,
51 /// Runs both on the interval and once per day at the scheduled time.
52 PeriodicAndExact,
53}
54
55/// A time of day for [`JobType::Exact`] schedules.
56///
57/// All fields use clock ranges; [`new`](Self::new) rejects out-of-range values.
58#[derive(Debug, Clone)]
59pub struct TimeOfDay {
60 /// Hour of day, 0–23.
61 pub hour: u8,
62 /// Minute of the hour, 0–59.
63 pub minute: u8,
64 /// Second of minute, 0–59.
65 pub second: u8
66}
67
68impl TimeOfDay {
69 /// Creates a time of day, returning an error string for out-of-range fields.
70 pub fn new(hour: u8, minute: u8, second: u8) -> Result<Self, String> {
71 if hour > 23 { return Err("hour >23".into()); }
72 if minute > 59 { return Err("minute >59".into()); }
73 if second > 59 { return Err("second >59".into()); }
74 Ok(Self { hour, minute, second })
75 }
76}
77
78/// A unit of background work managed by [`JobRegistry`].
79///
80/// `period` supplies the repeat interval for `Periodic` jobs (defaulting to
81/// 60 seconds when the job type needs it but `None` is returned), and
82/// `schedule` supplies the daily time for `Exact` jobs.
83#[async_trait::async_trait]
84pub trait ServiceJob: Send + Sync {
85 /// Stable name of the job, used in logs.
86 fn name(&self) -> &str;
87 /// Scheduling behavior of the job.
88 fn job_type(&self) -> JobType;
89 /// Repeat interval for periodic legs; `None` means no interval configured.
90 fn period(&self) -> Option<Duration>;
91 /// Daily time for exact legs; `None` means no daily time configured.
92 fn schedule(&self) -> Option<TimeOfDay>;
93 /// Whether the job is critical. Defaults to `false`.
94 fn is_critical(&self) -> bool { false }
95 /// Executes one run of the job.
96 async fn run(&self) -> anyhow::Result<ServiceResult<serde_json::Value>>;
97 /// Releases job resources on shutdown. Defaults to doing nothing.
98 async fn stop_gracefully(&self) -> anyhow::Result<()> { Ok(()) }
99}
100
101#[derive(Clone)]
102struct JobEntry {
103 job: Arc<dyn ServiceJob>,
104 runs: Arc<Mutex<usize>>,
105 failures: Arc<Mutex<usize>>,
106 successes: Arc<Mutex<usize>>,
107}
108
109/// Owns jobs and drives their schedules.
110///
111/// Add jobs with [`add_job`](Self::add_job), start all schedules with
112/// [`start`](Self::start), and release resources with [`stop`](Self::stop).
113/// `Periodic` legs run as Tokio interval tasks; `Exact` legs are cron entries
114/// (`PeriodicAndExact` daily leg fires on day 1 of each month).
115pub struct JobRegistry {
116 jobs: Vec<JobEntry>,
117 scheduler: Option<JobScheduler>,
118}
119
120impl JobRegistry {
121 /// Creates an empty registry with no scheduler running.
122 pub fn new() -> Self { Self { jobs: vec![], scheduler: None } }
123
124 /// Returns the total number of runs of a job.
125 pub fn get_runs(&self, name: &str) -> Option<usize> {
126 for entry in &self.jobs {
127 if entry.job.name() == name {
128 return Some(entry.runs.lock().unwrap().clone());
129 }
130 }
131 None
132 }
133
134 /// Returns the number of successful runs of a job.
135 pub fn get_failures(&self, name: &str) -> Option<usize> {
136 for entry in &self.jobs {
137 if entry.job.name() == name {
138 return Some(entry.failures.lock().unwrap().clone());
139 }
140 }
141 None
142 }
143
144
145 /// Returns the number of successful runs of a job.
146 pub fn get_successes(&self, name: &str) -> Option<usize> {
147 for entry in &self.jobs {
148 if entry.job.name() == name {
149 return Some(entry.successes.lock().unwrap().clone());
150 }
151 }
152 None
153 }
154
155 /// Adds a job to the registry. The job runs once [`start`](Self::start) is called.
156 ///
157 /// `J` is the concrete [`ServiceJob`] implementation being stored.
158 pub fn add_job<J: ServiceJob + 'static>(&mut self, job: J) {
159 self.jobs.push(JobEntry {
160 job: Arc::new(job),
161 runs: Arc::new(Mutex::new(0)),
162 failures: Arc::new(Mutex::new(0)),
163 successes: Arc::new(Mutex::new(0)),
164 });
165 }
166
167 /// Returns the number of jobs in the registry.
168 pub fn job_count(&self) -> usize { self.jobs.len() }
169
170 /// Starts the scheduler and all configured job legs.
171 ///
172 /// `Periodic` jobs without a period default to 60 seconds; `Exact` legs
173 /// without a schedule are skipped. Failures of individual runs are logged
174 /// and do not stop the schedule. Returns an error if the scheduler cannot start.
175 pub async fn start(&mut self) -> anyhow::Result<()> {
176 let sched = JobScheduler::new().await?;
177 tracing::info!(job_count = self.jobs.len(), "Starting job scheduler");
178 for entry in &self.jobs {
179 let job_clone = entry.job.clone();
180 match job_clone.job_type() {
181 JobType::Periodic => {
182 let period = job_clone.period().unwrap_or(Duration::from_secs(60));
183 tracing::info!(job = %job_clone.name(), schedule = "periodic", interval_secs = period.as_secs(), "Registered job");
184 // cron every N seconds: use Tokio interval instead of cron for simplicity
185 // Spawn periodic task
186 tokio::spawn({
187 let job_clone = job_clone.clone();
188 async move {
189 let mut interval = tokio::time::interval(period);
190 loop {
191 interval.tick().await;
192 if let Err(e) = job_clone.run().await {
193 tracing::error!(job = %job_clone.name(), error = %e, "Periodic job failed");
194 }
195 }
196 }
197 });
198 }
199 JobType::PeriodicAndExact => {
200 let schedule_clone = job_clone.clone();
201 let period_clone = job_clone.clone();
202 if let Some(tod) = schedule_clone.schedule() {
203 let job_name = schedule_clone.name().to_string();
204 // Monthly, on day 1 at the scheduled time.
205 let cron = format!("{} {} {} 1 * *", tod.second, tod.minute, tod.hour);
206 let j = Job::new_async(cron.as_str(), move |_, _| {
207 let jc = schedule_clone.clone();
208 Box::pin(async move {
209 if let Err(e) = jc.run().await {
210 tracing::error!(job = %jc.name(), error = %e, "Scheduled job failed");
211 }
212 })
213 })?;
214 sched.add(j).await?;
215 tracing::info!(job = %job_name, schedule = "exact", "Registered scheduled job");
216 }
217 // Also honor periodic interval if provided (PeriodicAndExact = both)
218 if let Some(period) = period_clone.period() {
219 tracing::info!(job = %period_clone.name(), schedule = "periodic", interval_secs = period.as_secs(), "Registered periodic job leg");
220 let pc = period_clone.clone();
221 tokio::spawn(async move {
222 let mut interval = tokio::time::interval(period);
223 loop {
224 interval.tick().await;
225 if let Err(e) = pc.run().await {
226 tracing::error!(job = %pc.name(), error = %e, "Periodic job leg failed");
227 }
228 }
229 });
230 }
231 }
232 JobType::Exact => {
233 if let Some(tod) = job_clone.schedule() {
234 let job_name = job_clone.name().to_string();
235 let cron = format!("{} {} {} * * *", tod.second, tod.minute, tod.hour);
236 let j = Job::new_async(cron.as_str(), move |_, _| {
237 let jc = job_clone.clone();
238 Box::pin(async move {
239 if let Err(e) = jc.run().await {
240 tracing::error!(job = %jc.name(), error = %e, "Scheduled job failed");
241 }
242 })
243 })?;
244 sched.add(j).await?;
245 tracing::info!(job = %job_name, schedule = "exact", "Registered scheduled job");
246 }
247 }
248 }
249 }
250 sched.start().await?;
251 self.scheduler = Some(sched);
252 tracing::info!("Job scheduler started");
253 Ok(())
254 }
255
256 /// Stops every job gracefully and shuts down the scheduler, if running.
257 ///
258 /// Returns the first error raised by a job's graceful stop or by shutdown.
259 pub async fn stop(&mut self) -> anyhow::Result<()> {
260 tracing::info!(job_count = self.jobs.len(), "Stopping jobs");
261 for entry in &self.jobs {
262 entry.job.stop_gracefully().await?;
263 }
264 if let Some(mut s) = self.scheduler.take() {
265 s.shutdown().await?;
266 }
267 tracing::info!("Jobs stopped");
268 Ok(())
269 }
270}
271
272impl Default for JobRegistry {
273 fn default() -> Self { Self::new() }
274}