cja/cron/
registry.rs

1use std::{collections::HashMap, error::Error, future::Future, pin::Pin, time::Duration};
2
3use chrono::Utc;
4use chrono_tz::Tz;
5
6use crate::app_state::AppState as AS;
7#[cfg(feature = "jobs")]
8use crate::jobs::Job;
9
10pub struct CronRegistry<AppState: AS> {
11    pub(super) jobs: HashMap<&'static str, CronJob<AppState>>,
12}
13
14#[async_trait::async_trait]
15pub trait CronFn<AppState: AS> {
16    // This collapses the error type to a string, right now thats because thats
17    // what the only consumer really needs. As we add more error debugging we'll
18    // need to change this.
19    async fn run(&self, app_state: AppState, context: String) -> Result<(), String>;
20}
21
22pub struct CronFnClosure<
23    AppState: AS,
24    FnError: Error + Send + Sync + 'static,
25    F: Fn(AppState, String) -> Pin<Box<dyn Future<Output = Result<(), FnError>> + Send>>
26        + Send
27        + Sync
28        + 'static,
29> {
30    pub(super) func: F,
31    _marker: std::marker::PhantomData<AppState>,
32}
33
34#[async_trait::async_trait]
35impl<
36    AppState: AS,
37    FnError: Error + Send + Sync + 'static,
38    F: Fn(AppState, String) -> Pin<Box<dyn Future<Output = Result<(), FnError>> + Send>>
39        + Send
40        + Sync
41        + 'static,
42> CronFn<AppState> for CronFnClosure<AppState, FnError, F>
43{
44    async fn run(&self, app_state: AppState, context: String) -> Result<(), String> {
45        (self.func)(app_state, context)
46            .await
47            .map_err(|err| format!("{err:?}"))
48    }
49}
50
51#[derive(Clone, Debug)]
52pub struct IntervalSchedule(pub Duration);
53
54impl IntervalSchedule {
55    fn should_run(
56        &self,
57        last_run: Option<&chrono::DateTime<Utc>>,
58        now: chrono::DateTime<Utc>,
59        _worker_started_at: chrono::DateTime<Utc>,
60        _timezone: Tz,
61    ) -> bool {
62        if let Some(last_run) = last_run {
63            let elapsed = now - last_run;
64            // If elapsed is negative, return false (don't run)
65            if elapsed < chrono::Duration::zero() {
66                return false;
67            }
68            // Safe to convert since we checked it's non-negative
69            let elapsed = elapsed.to_std().unwrap_or(Duration::ZERO);
70            elapsed > self.0
71        } else {
72            true
73        }
74    }
75
76    pub fn next_run(
77        &self,
78        last_run: Option<&chrono::DateTime<Utc>>,
79        now: chrono::DateTime<Utc>,
80        timezone: Tz,
81    ) -> chrono::DateTime<Tz> {
82        let last_run = last_run.unwrap_or(&now);
83        let last_run_tz = last_run.with_timezone(&timezone);
84        let duration: chrono::Duration = chrono::Duration::from_std(self.0).unwrap();
85        let next_run = last_run_tz.checked_add_signed(duration).unwrap();
86        next_run.with_timezone(&timezone)
87    }
88}
89
90#[derive(Clone, Debug)]
91pub struct CronSchedule(pub Box<cron::Schedule>);
92
93impl CronSchedule {
94    fn should_run(
95        &self,
96        last_run: Option<&chrono::DateTime<Utc>>,
97        now: chrono::DateTime<Utc>,
98        worker_started_at: chrono::DateTime<Utc>,
99        timezone: Tz,
100    ) -> bool {
101        // Use last run time if available, otherwise use worker start time
102        let last_run = last_run.unwrap_or(&worker_started_at);
103        let last_run_tz = last_run.with_timezone(&timezone);
104
105        if let Some(next_run) = self.0.after(&last_run_tz).next() {
106            let now_tz = now.with_timezone(&timezone);
107            now_tz >= next_run
108        } else {
109            false
110        }
111    }
112
113    pub fn next_run(
114        &self,
115        last_run: Option<&chrono::DateTime<Utc>>,
116        now: chrono::DateTime<Utc>,
117        timezone: Tz,
118    ) -> chrono::DateTime<Tz> {
119        let last_run = last_run.unwrap_or(&now);
120        let last_run_tz = last_run.with_timezone(&timezone);
121        self.0.after(&last_run_tz).next().unwrap()
122    }
123}
124
125#[derive(Clone, Debug)]
126pub enum Schedule {
127    Interval(IntervalSchedule),
128    Cron(CronSchedule),
129}
130
131impl Schedule {
132    pub fn should_run(
133        &self,
134        last_run: Option<&chrono::DateTime<Utc>>,
135        now: chrono::DateTime<Utc>,
136        worker_started_at: chrono::DateTime<Utc>,
137        timezone: Tz,
138    ) -> bool {
139        match self {
140            Schedule::Interval(interval) => {
141                interval.should_run(last_run, now, worker_started_at, timezone)
142            }
143            Schedule::Cron(cron) => cron.should_run(last_run, now, worker_started_at, timezone),
144        }
145    }
146
147    pub fn next_run(
148        &self,
149        last_run: Option<&chrono::DateTime<Utc>>,
150        now: chrono::DateTime<Utc>,
151        timezone: Tz,
152    ) -> chrono::DateTime<Tz> {
153        match self {
154            Schedule::Interval(interval) => interval.next_run(last_run, now, timezone),
155            Schedule::Cron(cron) => cron.next_run(last_run, now, timezone),
156        }
157    }
158}
159
160#[allow(clippy::type_complexity)]
161pub struct CronJob<AppState: AS> {
162    pub name: &'static str,
163    pub description: Option<&'static str>,
164    func: Box<dyn CronFn<AppState> + Send + Sync + 'static>,
165    pub schedule: Schedule,
166}
167
168#[derive(Debug, thiserror::Error)]
169#[error("TickError: {0}")]
170pub enum TickError {
171    JobError(String),
172    SqlxError(sqlx::Error),
173}
174
175impl<AppState: AS> CronJob<AppState> {
176    #[tracing::instrument(
177        name = "cron_job.tick",
178        level = "trace",
179        skip_all,
180        fields(
181            cron_job.name = self.name,
182            cron_job.schedule = ?self.schedule
183        )
184    )]
185    pub(crate) async fn tick(
186        &self,
187        app_state: AppState,
188        last_enqueue_map: &HashMap<String, chrono::DateTime<Utc>>,
189        worker_started_at: chrono::DateTime<Utc>,
190        timezone: Tz,
191    ) -> Result<(), TickError> {
192        let last_enqueue = last_enqueue_map.get(self.name);
193        let context = format!("Cron@{}", app_state.version());
194        let now = Utc::now();
195
196        let should_run = self
197            .schedule
198            .should_run(last_enqueue, now, worker_started_at, timezone);
199
200        if should_run {
201            tracing::info!(
202                task_name = self.name,
203                last_run = ?last_enqueue,
204                "Enqueuing Task"
205            );
206            (self.func)
207                .run(app_state.clone(), context)
208                .await
209                .map_err(TickError::JobError)?;
210
211            sqlx::query!(
212                "INSERT INTO Crons (cron_id, name, last_run_at, created_at, updated_at)
213                VALUES ($1, $2, $3, $4, $5)
214                ON CONFLICT (name)
215                DO UPDATE SET
216                last_run_at = $3",
217                uuid::Uuid::new_v4(),
218                self.name,
219                now,
220                now,
221                now
222            )
223            .execute(app_state.db())
224            .await
225            .map_err(TickError::SqlxError)?;
226        }
227
228        Ok(())
229    }
230
231    pub async fn run(&self, app_state: AppState, context: String) -> Result<(), String> {
232        (self.func).run(app_state, context).await
233    }
234}
235
236impl<AppState: AS> CronRegistry<AppState> {
237    pub fn new() -> Self {
238        Self {
239            jobs: HashMap::new(),
240        }
241    }
242
243    #[tracing::instrument(name = "cron.register", skip_all, fields(cron_job.name = name, cron_job.interval = ?interval))]
244    pub fn register<FnError: Error + Send + Sync + 'static>(
245        &mut self,
246        name: &'static str,
247        description: Option<&'static str>,
248        interval: Duration,
249        job: impl Fn(AppState, String) -> Pin<Box<dyn Future<Output = Result<(), FnError>> + Send>>
250        + Send
251        + Sync
252        + 'static,
253    ) {
254        let cron_job = CronJob {
255            name,
256            description,
257            func: Box::new(CronFnClosure {
258                func: job,
259                _marker: std::marker::PhantomData,
260            }),
261            schedule: Schedule::Interval(IntervalSchedule(interval)),
262        };
263        self.jobs.insert(name, cron_job);
264    }
265
266    #[tracing::instrument(name = "cron.register_with_cron", skip_all, fields(cron_job.name = name, cron_job.cron = cron_expr))]
267    pub fn register_with_cron<FnError: Error + Send + Sync + 'static>(
268        &mut self,
269        name: &'static str,
270        description: Option<&'static str>,
271        cron_expr: &str,
272        job: impl Fn(AppState, String) -> Pin<Box<dyn Future<Output = Result<(), FnError>> + Send>>
273        + Send
274        + Sync
275        + 'static,
276    ) -> Result<(), cron::error::Error> {
277        let cron_schedule = cron_expr.parse::<cron::Schedule>()?;
278        let cron_job = CronJob {
279            name,
280            description,
281            func: Box::new(CronFnClosure {
282                func: job,
283                _marker: std::marker::PhantomData,
284            }),
285            schedule: Schedule::Cron(CronSchedule(Box::new(cron_schedule))),
286        };
287        self.jobs.insert(name, cron_job);
288        Ok(())
289    }
290
291    #[cfg(feature = "jobs")]
292    #[tracing::instrument(name = "cron.register_job", skip_all, fields(cron_job.name = J::NAME, cron_job.interval = ?interval))]
293    pub fn register_job<J: Job<AppState>>(
294        &mut self,
295        job: J,
296        description: Option<&'static str>,
297        interval: Duration,
298    ) {
299        self.register(J::NAME, description, interval, move |app_state, context| {
300            J::enqueue(job.clone(), app_state, context, None)
301        });
302    }
303
304    #[cfg(feature = "jobs")]
305    #[tracing::instrument(name = "cron.register_job_with_cron", skip_all, fields(cron_job.name = J::NAME, cron_job.cron = cron_expr))]
306    pub fn register_job_with_cron<J: Job<AppState>>(
307        &mut self,
308        job: J,
309        description: Option<&'static str>,
310        cron_expr: &str,
311    ) -> Result<(), cron::error::Error> {
312        self.register_with_cron(
313            J::NAME,
314            description,
315            cron_expr,
316            move |app_state, context| J::enqueue(job.clone(), app_state, context, None),
317        )
318    }
319
320    #[cfg(feature = "jobs")]
321    #[tracing::instrument(name = "cron.get", skip_all, fields(cron_job.name = name))]
322    pub fn get(&self, name: &str) -> Option<&CronJob<AppState>> {
323        self.jobs.get(name)
324    }
325
326    /// Returns a reference to all registered cron jobs.
327    pub fn jobs(&self) -> &HashMap<&'static str, CronJob<AppState>> {
328        &self.jobs
329    }
330
331    /// Returns each registered cron job's name and schedule string, sorted by name.
332    ///
333    /// Interval schedules use the `Duration` `Debug` form (e.g. `"300s"`),
334    /// matching what the `cron.register` span records. Cron-expression
335    /// schedules use the original expression (e.g. `"0 0 * * * * *"`).
336    #[must_use]
337    pub fn entries(&self) -> Vec<(&'static str, String)> {
338        let mut entries: Vec<(&'static str, String)> = self
339            .jobs
340            .values()
341            .map(|job| {
342                let schedule = match &job.schedule {
343                    Schedule::Interval(IntervalSchedule(duration)) => format!("{duration:?}"),
344                    Schedule::Cron(CronSchedule(schedule)) => schedule.to_string(),
345                };
346                (job.name, schedule)
347            })
348            .collect();
349        entries.sort_unstable_by_key(|(name, _)| *name);
350        entries
351    }
352}
353
354impl<AppState: AS> Default for CronRegistry<AppState> {
355    fn default() -> Self {
356        Self::new()
357    }
358}
359
360#[cfg(test)]
361mod test {
362    use crate::app_state::AppState;
363    use crate::server::cookies::CookieKey;
364
365    use super::*;
366
367    #[derive(Clone)]
368    struct TestAppState {
369        db: sqlx::PgPool,
370        cookie_key: CookieKey,
371    }
372
373    impl AppState for TestAppState {
374        fn db(&self) -> &sqlx::PgPool {
375            &self.db
376        }
377
378        fn version(&self) -> &'static str {
379            "test"
380        }
381
382        fn cookie_key(&self) -> &CookieKey {
383            &self.cookie_key
384        }
385    }
386
387    #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
388    struct TestJob;
389
390    #[async_trait::async_trait]
391    impl Job<TestAppState> for TestJob {
392        const NAME: &'static str = "test_job";
393
394        async fn run(&self, _app_state: TestAppState) -> color_eyre::Result<()> {
395            Ok(())
396        }
397    }
398
399    #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
400    struct FailingJob;
401
402    #[async_trait::async_trait]
403    impl Job<TestAppState> for FailingJob {
404        const NAME: &'static str = "failing_job";
405
406        async fn run(&self, _app_state: TestAppState) -> color_eyre::Result<()> {
407            Err(color_eyre::eyre::eyre!("Test error"))
408        }
409    }
410
411    #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
412    struct SecondTestJob;
413
414    #[async_trait::async_trait]
415    impl Job<TestAppState> for SecondTestJob {
416        const NAME: &'static str = "second_test_job";
417
418        async fn run(&self, _app_state: TestAppState) -> color_eyre::Result<()> {
419            Ok(())
420        }
421    }
422
423    #[sqlx::test]
424    async fn test_tick_creates_new_cron_record(db: sqlx::PgPool) {
425        let app_state = TestAppState {
426            db: db.clone(),
427            cookie_key: CookieKey::generate(),
428        };
429        let mut registry = CronRegistry::new();
430        registry.register_job(TestJob, None, Duration::from_secs(1));
431
432        let cron_job = registry.jobs.get(TestJob::NAME).unwrap();
433        assert_eq!(cron_job.name, TestJob::NAME);
434        assert!(
435            matches!(cron_job.schedule, Schedule::Interval(IntervalSchedule(d)) if d == Duration::from_secs(1))
436        );
437
438        let worker = crate::cron::Worker::new(app_state.clone(), registry);
439
440        let existing_record =
441            sqlx::query!("SELECT cron_id FROM Crons where name = $1", TestJob::NAME)
442                .fetch_optional(&app_state.db)
443                .await
444                .unwrap();
445        assert!(
446            existing_record.is_none(),
447            "Record should not exist {}",
448            existing_record.unwrap().cron_id
449        );
450
451        worker.tick().await.unwrap();
452
453        let last_run = sqlx::query!(
454            "SELECT last_run_at FROM Crons WHERE name = $1",
455            TestJob::NAME
456        )
457        .fetch_one(&app_state.db)
458        .await
459        .unwrap();
460
461        let now = Utc::now();
462        let last_run_at = last_run.last_run_at;
463        let diff = now.signed_duration_since(last_run_at);
464        assert!(diff.num_milliseconds() < 1000);
465    }
466
467    #[sqlx::test]
468    async fn test_tick_skips_updating_existing_cron_record(db: sqlx::PgPool) {
469        let app_state = TestAppState {
470            db: db.clone(),
471            cookie_key: CookieKey::generate(),
472        };
473        let mut registry = CronRegistry::new();
474        registry.register_job(TestJob, None, Duration::from_mins(1));
475        let worker = crate::cron::Worker::new(app_state.clone(), registry);
476
477        let previously = Utc::now();
478        sqlx::query!(
479            "INSERT INTO Crons (cron_id, name, last_run_at, created_at, updated_at)
480            VALUES ($1, $2, $3, $3, $3)
481            ON CONFLICT (name)
482            DO UPDATE SET
483            last_run_at = $3",
484            uuid::Uuid::new_v4(),
485            TestJob::NAME,
486            previously
487        )
488        .execute(&app_state.db)
489        .await
490        .unwrap();
491
492        worker.tick().await.unwrap();
493
494        let last_run = sqlx::query!(
495            "SELECT last_run_at FROM Crons WHERE name = $1",
496            TestJob::NAME
497        )
498        .fetch_one(&app_state.db)
499        .await
500        .unwrap();
501
502        let diff = last_run.last_run_at.signed_duration_since(previously);
503        assert!(diff.num_milliseconds() < 50);
504    }
505
506    #[sqlx::test]
507    async fn test_tick_updates_cron_record_when_interval_elapsed(db: sqlx::PgPool) {
508        let app_state = TestAppState {
509            db: db.clone(),
510            cookie_key: CookieKey::generate(),
511        };
512        let mut registry = CronRegistry::new();
513        registry.register_job(TestJob, None, Duration::from_secs(1));
514        let worker = crate::cron::Worker::new(app_state.clone(), registry);
515
516        let two_seconds_ago = Utc::now() - chrono::Duration::seconds(2);
517        sqlx::query!(
518            "INSERT INTO Crons (cron_id, name, last_run_at, created_at, updated_at)
519            VALUES ($1, $2, $3, $3, $3)",
520            uuid::Uuid::new_v4(),
521            TestJob::NAME,
522            two_seconds_ago
523        )
524        .execute(&app_state.db)
525        .await
526        .unwrap();
527
528        worker.tick().await.unwrap();
529
530        let last_run = sqlx::query!(
531            "SELECT last_run_at FROM Crons WHERE name = $1",
532            TestJob::NAME
533        )
534        .fetch_one(&app_state.db)
535        .await
536        .unwrap();
537
538        assert!(last_run.last_run_at > two_seconds_ago);
539        let diff = Utc::now().signed_duration_since(last_run.last_run_at);
540        assert!(diff.num_milliseconds() < 1000);
541    }
542
543    #[sqlx::test]
544    async fn test_tick_enqueues_failing_job_successfully(db: sqlx::PgPool) {
545        let app_state = TestAppState {
546            db: db.clone(),
547            cookie_key: CookieKey::generate(),
548        };
549        let mut registry = CronRegistry::new();
550        registry.register_job(FailingJob, None, Duration::from_secs(1));
551
552        let cron_job = registry.jobs.get(FailingJob::NAME).unwrap();
553        let last_enqueue_map = HashMap::new();
554        let worker_started_at = Utc::now();
555
556        let result = cron_job
557            .tick(
558                app_state.clone(),
559                &last_enqueue_map,
560                worker_started_at,
561                chrono_tz::UTC,
562            )
563            .await;
564        assert!(result.is_ok());
565
566        let cron_record = sqlx::query!(
567            "SELECT last_run_at FROM Crons WHERE name = $1",
568            FailingJob::NAME
569        )
570        .fetch_one(&app_state.db)
571        .await
572        .unwrap();
573
574        let now = Utc::now();
575        let diff = now.signed_duration_since(cron_record.last_run_at);
576        assert!(diff.num_milliseconds() < 1000);
577
578        let job_count = sqlx::query!(
579            "SELECT COUNT(*) as count FROM jobs WHERE name = $1",
580            FailingJob::NAME
581        )
582        .fetch_one(&app_state.db)
583        .await
584        .unwrap();
585        assert_eq!(job_count.count.unwrap(), 1);
586    }
587
588    #[sqlx::test]
589    async fn test_tick_with_custom_function_error(db: sqlx::PgPool) {
590        let app_state = TestAppState {
591            db: db.clone(),
592            cookie_key: CookieKey::generate(),
593        };
594        #[derive(Debug, thiserror::Error)]
595        #[error("Custom function error")]
596        struct CustomError;
597
598        let mut registry = CronRegistry::new();
599        registry.register(
600            "custom_failing",
601            None,
602            Duration::from_secs(1),
603            |_app_state, _context| Box::pin(async { Err(CustomError) }),
604        );
605
606        let cron_job = registry.jobs.get("custom_failing").unwrap();
607        let last_enqueue_map = HashMap::new();
608        let worker_started_at = Utc::now();
609
610        let result = cron_job
611            .tick(
612                app_state.clone(),
613                &last_enqueue_map,
614                worker_started_at,
615                chrono_tz::UTC,
616            )
617            .await;
618        assert!(result.is_err());
619        match result.unwrap_err() {
620            TickError::JobError(err) => {
621                assert!(err.contains("CustomError"));
622            }
623            TickError::SqlxError(_) => panic!("Expected JobError"),
624        }
625
626        let cron_count = sqlx::query!(
627            "SELECT COUNT(*) as count FROM Crons WHERE name = $1",
628            "custom_failing"
629        )
630        .fetch_one(&app_state.db)
631        .await
632        .unwrap();
633        assert_eq!(cron_count.count.unwrap(), 0);
634    }
635
636    #[sqlx::test]
637    async fn test_worker_tick_with_multiple_jobs(db: sqlx::PgPool) {
638        let app_state = TestAppState {
639            db: db.clone(),
640            cookie_key: CookieKey::generate(),
641        };
642        let mut registry = CronRegistry::new();
643        registry.register_job(TestJob, None, Duration::from_secs(1));
644        registry.register_job(SecondTestJob, None, Duration::from_secs(1));
645
646        assert_eq!(registry.jobs.len(), 2);
647
648        let worker = crate::cron::Worker::new(app_state.clone(), registry);
649
650        worker.tick().await.unwrap();
651
652        let test_job_record = sqlx::query!(
653            "SELECT last_run_at FROM Crons WHERE name = $1",
654            TestJob::NAME
655        )
656        .fetch_one(&app_state.db)
657        .await
658        .unwrap();
659
660        let second_job_record = sqlx::query!(
661            "SELECT last_run_at FROM Crons WHERE name = $1",
662            SecondTestJob::NAME
663        )
664        .fetch_one(&app_state.db)
665        .await
666        .unwrap();
667
668        let now = Utc::now();
669        let diff1 = now.signed_duration_since(test_job_record.last_run_at);
670        let diff2 = now.signed_duration_since(second_job_record.last_run_at);
671
672        assert!(diff1.num_milliseconds() < 1000);
673        assert!(diff2.num_milliseconds() < 1000);
674    }
675
676    #[sqlx::test]
677    async fn test_worker_respects_existing_last_run_times(db: sqlx::PgPool) {
678        let app_state = TestAppState {
679            db: db.clone(),
680            cookie_key: CookieKey::generate(),
681        };
682        let mut registry = CronRegistry::new();
683        registry.register_job(TestJob, None, Duration::from_secs(10));
684        registry.register_job(SecondTestJob, None, Duration::from_secs(5));
685
686        let worker = crate::cron::Worker::new(app_state.clone(), registry);
687
688        let recent_time = Utc::now() - chrono::Duration::seconds(3);
689        let old_time = Utc::now() - chrono::Duration::seconds(10);
690
691        sqlx::query!(
692            "INSERT INTO Crons (cron_id, name, last_run_at, created_at, updated_at)
693                VALUES ($1, $2, $3, $3, $3)",
694            uuid::Uuid::new_v4(),
695            TestJob::NAME,
696            recent_time
697        )
698        .execute(&app_state.db)
699        .await
700        .unwrap();
701
702        sqlx::query!(
703            "INSERT INTO Crons (cron_id, name, last_run_at, created_at, updated_at)
704                VALUES ($1, $2, $3, $3, $3)",
705            uuid::Uuid::new_v4(),
706            SecondTestJob::NAME,
707            old_time
708        )
709        .execute(&app_state.db)
710        .await
711        .unwrap();
712
713        worker.tick().await.unwrap();
714
715        let test_job_record = sqlx::query!(
716            "SELECT last_run_at FROM Crons WHERE name = $1",
717            TestJob::NAME
718        )
719        .fetch_one(&app_state.db)
720        .await
721        .unwrap();
722
723        let second_job_record = sqlx::query!(
724            "SELECT last_run_at FROM Crons WHERE name = $1",
725            SecondTestJob::NAME
726        )
727        .fetch_one(&app_state.db)
728        .await
729        .unwrap();
730
731        let recent_diff = test_job_record
732            .last_run_at
733            .signed_duration_since(recent_time);
734        assert!(recent_diff.num_milliseconds() < 100);
735        assert!(second_job_record.last_run_at > old_time);
736    }
737
738    #[sqlx::test]
739    async fn test_tick_handles_future_last_run_time(db: sqlx::PgPool) {
740        let app_state = TestAppState {
741            db: db.clone(),
742            cookie_key: CookieKey::generate(),
743        };
744        let mut registry = CronRegistry::new();
745        registry.register_job(TestJob, None, Duration::from_secs(1));
746
747        let cron_job = registry.jobs.get(TestJob::NAME).unwrap();
748
749        let future_time = Utc::now() + chrono::Duration::hours(1);
750        let mut last_enqueue_map = HashMap::new();
751        last_enqueue_map.insert(TestJob::NAME.to_string(), future_time);
752        let worker_started_at = Utc::now();
753
754        let result = cron_job
755            .tick(
756                app_state.clone(),
757                &last_enqueue_map,
758                worker_started_at,
759                chrono_tz::UTC,
760            )
761            .await;
762
763        // With future last_run time, the job should not run
764        assert!(result.is_ok());
765
766        let cron_count = sqlx::query!(
767            "SELECT COUNT(*) as count FROM Crons WHERE name = $1",
768            TestJob::NAME
769        )
770        .fetch_one(&app_state.db)
771        .await
772        .unwrap();
773        assert_eq!(
774            cron_count.count.unwrap(),
775            0,
776            "Should not have created cron record with future last_run time"
777        );
778    }
779
780    #[sqlx::test]
781    async fn test_cron_expression_scheduling(db: sqlx::PgPool) {
782        let app_state = TestAppState {
783            db: db.clone(),
784            cookie_key: CookieKey::generate(),
785        };
786        let mut registry = CronRegistry::new();
787
788        // Register with cron expression that runs every minute
789        registry
790            .register_job_with_cron(TestJob, None, "0 * * * * * *")
791            .unwrap();
792
793        let cron_job = registry.jobs.get(TestJob::NAME).unwrap();
794        assert!(matches!(cron_job.schedule, Schedule::Cron(_)));
795
796        // First run should wait for the next scheduled time based on worker start
797        let last_enqueue_map = HashMap::new();
798        // Use a time in the past to ensure the cron will trigger
799        let worker_started_at = Utc::now() - chrono::Duration::minutes(2);
800        let result = cron_job
801            .tick(
802                app_state.clone(),
803                &last_enqueue_map,
804                worker_started_at,
805                chrono_tz::UTC,
806            )
807            .await;
808        assert!(result.is_ok());
809
810        // Verify cron record was created (if it ran)
811        let cron_record = sqlx::query!(
812            "SELECT last_run_at FROM Crons WHERE name = $1",
813            TestJob::NAME
814        )
815        .fetch_optional(&app_state.db)
816        .await
817        .unwrap();
818
819        // The job should have run since we used a worker start time 2 minutes ago
820        assert!(cron_record.is_some());
821        if let Some(record) = cron_record {
822            let now = Utc::now();
823            let diff = now.signed_duration_since(record.last_run_at);
824            assert!(diff.num_milliseconds() < 1000);
825        }
826    }
827
828    #[sqlx::test]
829    async fn test_cron_expression_respects_schedule(db: sqlx::PgPool) {
830        let app_state = TestAppState {
831            db: db.clone(),
832            cookie_key: CookieKey::generate(),
833        };
834        let mut registry = CronRegistry::new();
835
836        // Register with cron expression that runs at specific minute (e.g., minute 30)
837        registry
838            .register_job_with_cron(TestJob, None, "0 30 * * * * *")
839            .unwrap();
840
841        let cron_job = registry.jobs.get(TestJob::NAME).unwrap();
842
843        // Set last run to 29 minutes ago (shouldn't trigger if we're not at minute 30)
844        let last_run = Utc::now() - chrono::Duration::minutes(29);
845        let mut last_enqueue_map = HashMap::new();
846        last_enqueue_map.insert(TestJob::NAME.to_string(), last_run);
847
848        // This might or might not run depending on current minute
849        // We'll just verify it doesn't error
850        let worker_started_at = Utc::now();
851        let result = cron_job
852            .tick(
853                app_state.clone(),
854                &last_enqueue_map,
855                worker_started_at,
856                chrono_tz::UTC,
857            )
858            .await;
859        assert!(result.is_ok());
860    }
861
862    #[test]
863    fn test_invalid_cron_expression() {
864        let mut registry: CronRegistry<TestAppState> = CronRegistry::new();
865
866        // Invalid cron expression should return error
867        let result = registry.register_with_cron(
868            "invalid_cron",
869            None,
870            "invalid expression",
871            |_app_state, _context| Box::pin(async { Ok::<(), std::io::Error>(()) }),
872        );
873
874        assert!(result.is_err());
875    }
876
877    #[sqlx::test]
878    async fn test_mixed_interval_and_cron_jobs(db: sqlx::PgPool) {
879        let app_state = TestAppState {
880            db: db.clone(),
881            cookie_key: CookieKey::generate(),
882        };
883        let mut registry = CronRegistry::new();
884
885        // Register one job with interval
886        registry.register_job(TestJob, None, Duration::from_secs(1));
887
888        // Register another job with cron expression (every second)
889        registry
890            .register_job_with_cron(SecondTestJob, None, "* * * * * * *")
891            .unwrap();
892
893        assert_eq!(registry.jobs.len(), 2);
894
895        let interval_job = registry.jobs.get(TestJob::NAME).unwrap();
896        assert!(matches!(interval_job.schedule, Schedule::Interval(_)));
897
898        let cron_job = registry.jobs.get(SecondTestJob::NAME).unwrap();
899        assert!(matches!(cron_job.schedule, Schedule::Cron(_)));
900
901        // Create worker with start time in past to ensure both jobs run
902        let mut worker = crate::cron::Worker::new(app_state.clone(), registry);
903        worker.started_at = Utc::now() - chrono::Duration::seconds(2);
904
905        // Both jobs should run on first tick
906        worker.tick().await.unwrap();
907
908        let test_job_count = sqlx::query!(
909            "SELECT COUNT(*) as count FROM Crons WHERE name = $1",
910            TestJob::NAME
911        )
912        .fetch_one(&app_state.db)
913        .await
914        .unwrap();
915
916        let second_job_count = sqlx::query!(
917            "SELECT COUNT(*) as count FROM Crons WHERE name = $1",
918            SecondTestJob::NAME
919        )
920        .fetch_one(&app_state.db)
921        .await
922        .unwrap();
923
924        assert_eq!(test_job_count.count.unwrap(), 1);
925        assert_eq!(second_job_count.count.unwrap(), 1);
926    }
927
928    #[test]
929    fn test_entries_lists_names_and_schedule_strings() {
930        let mut registry: CronRegistry<TestAppState> = CronRegistry::new();
931
932        registry.register(
933            "interval_job",
934            None,
935            Duration::from_mins(5),
936            |_app_state, _context| Box::pin(async { Ok::<(), std::io::Error>(()) }),
937        );
938        registry
939            .register_with_cron(
940                "cron_expr_job",
941                None,
942                "0 0 9 * * * *",
943                |_app_state, _context| Box::pin(async { Ok::<(), std::io::Error>(()) }),
944            )
945            .unwrap();
946
947        let entries = registry.entries();
948        assert_eq!(
949            entries,
950            vec![
951                ("cron_expr_job", "0 0 9 * * * *".to_string()),
952                ("interval_job", "300s".to_string()),
953            ]
954        );
955    }
956
957    /// Regression test for the cron starvation bug fixed in 4f29b39 ("Fix Cron (#5)"):
958    /// `last_run_at` used to be bumped on every worker tick (even when the job did
959    /// not fire), so any interval longer than the tick cadence never fired again
960    /// after its first run. This verifies a >60s interval survives multiple
961    /// non-firing ticks with `last_run_at` untouched, then fires once due.
962    #[sqlx::test]
963    async fn test_long_interval_last_run_only_advances_on_fire(db: sqlx::PgPool) {
964        let app_state = TestAppState {
965            db: db.clone(),
966            cookie_key: CookieKey::generate(),
967        };
968        let mut registry = CronRegistry::new();
969        // 5-minute interval: much longer than the worker's ~60s tick cadence
970        registry.register_job(TestJob, None, Duration::from_mins(5));
971        let worker = crate::cron::Worker::new(app_state.clone(), registry);
972
973        // Simulate a fire 2 minutes ago — the interval has NOT yet elapsed
974        let two_minutes_ago = Utc::now() - chrono::Duration::minutes(2);
975        sqlx::query!(
976            "INSERT INTO Crons (cron_id, name, last_run_at, created_at, updated_at)
977            VALUES ($1, $2, $3, $3, $3)",
978            uuid::Uuid::new_v4(),
979            TestJob::NAME,
980            two_minutes_ago
981        )
982        .execute(&app_state.db)
983        .await
984        .unwrap();
985
986        // Multiple simulated worker ticks while the interval has not elapsed
987        for _ in 0..3 {
988            worker.tick().await.unwrap();
989
990            let last_run = sqlx::query!(
991                "SELECT last_run_at FROM Crons WHERE name = $1",
992                TestJob::NAME
993            )
994            .fetch_one(&app_state.db)
995            .await
996            .unwrap();
997
998            let drift = last_run
999                .last_run_at
1000                .signed_duration_since(two_minutes_ago)
1001                .num_milliseconds()
1002                .abs();
1003            assert!(
1004                drift < 50,
1005                "last_run_at must not advance on non-firing ticks (drifted {drift}ms)"
1006            );
1007        }
1008
1009        // No job should have been enqueued by the non-firing ticks
1010        let job_count = sqlx::query!(
1011            "SELECT COUNT(*) as count FROM jobs WHERE name = $1",
1012            TestJob::NAME
1013        )
1014        .fetch_one(&app_state.db)
1015        .await
1016        .unwrap();
1017        assert_eq!(job_count.count.unwrap(), 0);
1018
1019        // Now simulate the interval having elapsed: last fire 6 minutes ago
1020        let six_minutes_ago = Utc::now() - chrono::Duration::minutes(6);
1021        sqlx::query!(
1022            "UPDATE Crons SET last_run_at = $1 WHERE name = $2",
1023            six_minutes_ago,
1024            TestJob::NAME
1025        )
1026        .execute(&app_state.db)
1027        .await
1028        .unwrap();
1029
1030        worker.tick().await.unwrap();
1031
1032        let last_run = sqlx::query!(
1033            "SELECT last_run_at FROM Crons WHERE name = $1",
1034            TestJob::NAME
1035        )
1036        .fetch_one(&app_state.db)
1037        .await
1038        .unwrap();
1039        assert!(
1040            last_run.last_run_at > six_minutes_ago,
1041            "cron should fire once the interval has elapsed"
1042        );
1043        let diff = Utc::now().signed_duration_since(last_run.last_run_at);
1044        assert!(diff.num_milliseconds() < 1000);
1045
1046        let job_count = sqlx::query!(
1047            "SELECT COUNT(*) as count FROM jobs WHERE name = $1",
1048            TestJob::NAME
1049        )
1050        .fetch_one(&app_state.db)
1051        .await
1052        .unwrap();
1053        assert_eq!(job_count.count.unwrap(), 1);
1054    }
1055}
cja/cron/registry.rs

cja/cron/
registry.rs
