1use crate::error::AwaError;
7use crate::job::JobRow;
8use chrono::{DateTime, Utc};
9use croner::Cron;
10use serde::Serialize;
11use sqlx::PgExecutor;
12
13#[derive(Debug, Clone)]
17pub struct PeriodicJob {
18 pub name: String,
20 pub cron_expr: String,
22 pub timezone: String,
24 pub kind: String,
26 pub queue: String,
28 pub args: serde_json::Value,
30 pub priority: i16,
32 pub max_attempts: i16,
34 pub tags: Vec<String>,
36 pub metadata: serde_json::Value,
38}
39
40impl PeriodicJob {
41 pub fn builder(name: impl Into<String>, cron_expr: impl Into<String>) -> PeriodicJobBuilder {
46 PeriodicJobBuilder {
47 name: name.into(),
48 cron_expr: cron_expr.into(),
49 timezone: "UTC".to_string(),
50 queue: "default".to_string(),
51 priority: 2,
52 max_attempts: 25,
53 tags: Vec::new(),
54 metadata: serde_json::json!({}),
55 }
56 }
57
58 pub fn latest_fire_time(
64 &self,
65 now: DateTime<Utc>,
66 after: Option<DateTime<Utc>>,
67 ) -> Option<DateTime<Utc>> {
68 let cron = Cron::new(&self.cron_expr)
69 .with_seconds_optional()
70 .parse()
71 .expect("cron_expr was validated at build time");
72
73 let tz: chrono_tz::Tz = self
74 .timezone
75 .parse()
76 .expect("timezone was validated at build time");
77
78 let now_tz = now.with_timezone(&tz);
79
80 let search_start = match after {
84 Some(after_time) => after_time.with_timezone(&tz),
85 None => now_tz - chrono::Duration::hours(24),
87 };
88
89 let mut latest_fire: Option<DateTime<Utc>> = None;
90
91 for fire_time in cron.clone().iter_from(search_start) {
93 let fire_utc = fire_time.with_timezone(&Utc);
94
95 if fire_utc > now {
97 break;
98 }
99
100 if let Some(after_time) = after {
102 if fire_utc <= after_time {
103 continue;
104 }
105 }
106
107 latest_fire = Some(fire_utc);
108 }
109
110 latest_fire
111 }
112}
113
114#[derive(Debug, Clone)]
116pub struct PeriodicJobBuilder {
117 name: String,
118 cron_expr: String,
119 timezone: String,
120 queue: String,
121 priority: i16,
122 max_attempts: i16,
123 tags: Vec<String>,
124 metadata: serde_json::Value,
125}
126
127impl PeriodicJobBuilder {
128 pub fn timezone(mut self, timezone: impl Into<String>) -> Self {
130 self.timezone = timezone.into();
131 self
132 }
133
134 pub fn queue(mut self, queue: impl Into<String>) -> Self {
136 self.queue = queue.into();
137 self
138 }
139
140 pub fn priority(mut self, priority: i16) -> Self {
142 self.priority = priority;
143 self
144 }
145
146 pub fn max_attempts(mut self, max_attempts: i16) -> Self {
148 self.max_attempts = max_attempts;
149 self
150 }
151
152 pub fn tags(mut self, tags: Vec<String>) -> Self {
154 self.tags = tags;
155 self
156 }
157
158 pub fn metadata(mut self, metadata: serde_json::Value) -> Self {
160 self.metadata = metadata;
161 self
162 }
163
164 pub fn build(self, args: &impl crate::JobArgs) -> Result<PeriodicJob, AwaError> {
169 self.build_raw(args.kind_str().to_string(), args.to_args()?)
170 }
171
172 pub fn build_raw(self, kind: String, args: serde_json::Value) -> Result<PeriodicJob, AwaError> {
174 Cron::new(&self.cron_expr)
176 .with_seconds_optional()
177 .parse()
178 .map_err(|err| AwaError::Validation(format!("invalid cron expression: {err}")))?;
179
180 self.timezone
182 .parse::<chrono_tz::Tz>()
183 .map_err(|err| AwaError::Validation(format!("invalid timezone: {err}")))?;
184
185 if !(1..=4).contains(&self.priority) {
187 return Err(AwaError::Validation(format!(
188 "priority must be between 1 and 4, got {}",
189 self.priority
190 )));
191 }
192
193 if !(1..=1000).contains(&self.max_attempts) {
195 return Err(AwaError::Validation(format!(
196 "max_attempts must be between 1 and 1000, got {}",
197 self.max_attempts
198 )));
199 }
200
201 Ok(PeriodicJob {
202 name: self.name,
203 cron_expr: self.cron_expr,
204 timezone: self.timezone,
205 kind,
206 queue: self.queue,
207 args,
208 priority: self.priority,
209 max_attempts: self.max_attempts,
210 tags: self.tags,
211 metadata: self.metadata,
212 })
213 }
214}
215
216#[derive(Debug, Clone, sqlx::FromRow, Serialize)]
218pub struct CronJobRow {
219 pub name: String,
220 pub cron_expr: String,
221 pub timezone: String,
222 pub kind: String,
223 pub queue: String,
224 pub args: serde_json::Value,
225 pub priority: i16,
226 pub max_attempts: i16,
227 pub tags: Vec<String>,
228 pub metadata: serde_json::Value,
229 pub last_enqueued_at: Option<DateTime<Utc>>,
230 pub created_at: DateTime<Utc>,
231 pub updated_at: DateTime<Utc>,
232}
233
234pub async fn upsert_cron_job<'e, E>(executor: E, job: &PeriodicJob) -> Result<(), AwaError>
238where
239 E: PgExecutor<'e>,
240{
241 sqlx::query(
242 r#"
243 INSERT INTO awa.cron_jobs (name, cron_expr, timezone, kind, queue, args, priority, max_attempts, tags, metadata)
244 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
245 ON CONFLICT (name) DO UPDATE SET
246 cron_expr = EXCLUDED.cron_expr,
247 timezone = EXCLUDED.timezone,
248 kind = EXCLUDED.kind,
249 queue = EXCLUDED.queue,
250 args = EXCLUDED.args,
251 priority = EXCLUDED.priority,
252 max_attempts = EXCLUDED.max_attempts,
253 tags = EXCLUDED.tags,
254 metadata = EXCLUDED.metadata,
255 updated_at = now()
256 "#,
257 )
258 .bind(&job.name)
259 .bind(&job.cron_expr)
260 .bind(&job.timezone)
261 .bind(&job.kind)
262 .bind(&job.queue)
263 .bind(&job.args)
264 .bind(job.priority)
265 .bind(job.max_attempts)
266 .bind(&job.tags)
267 .bind(&job.metadata)
268 .execute(executor)
269 .await?;
270
271 Ok(())
272}
273
274pub fn next_fire_time(cron_expr: &str, timezone: &str) -> Option<DateTime<Utc>> {
278 next_fire_time_after(cron_expr, timezone, Utc::now())
279}
280
281pub fn next_fire_time_after(
286 cron_expr: &str,
287 timezone: &str,
288 after: DateTime<Utc>,
289) -> Option<DateTime<Utc>> {
290 let cron = Cron::new(cron_expr).with_seconds_optional().parse().ok()?;
291 let tz: chrono_tz::Tz = timezone.parse().ok()?;
292 let after_tz = after.with_timezone(&tz);
293 let next = cron.iter_from(after_tz).next()?;
294 Some(next.with_timezone(&Utc))
295}
296
297pub async fn list_cron_jobs<'e, E>(executor: E) -> Result<Vec<CronJobRow>, AwaError>
299where
300 E: PgExecutor<'e>,
301{
302 let rows = sqlx::query_as::<_, CronJobRow>("SELECT * FROM awa.cron_jobs ORDER BY name")
303 .fetch_all(executor)
304 .await?;
305 Ok(rows)
306}
307
308pub async fn delete_cron_job<'e, E>(executor: E, name: &str) -> Result<bool, AwaError>
310where
311 E: PgExecutor<'e>,
312{
313 let result = sqlx::query("DELETE FROM awa.cron_jobs WHERE name = $1")
314 .bind(name)
315 .execute(executor)
316 .await?;
317 Ok(result.rows_affected() > 0)
318}
319
320pub async fn atomic_enqueue<'e, E>(
330 executor: E,
331 cron_name: &str,
332 fire_time: DateTime<Utc>,
333 previous_enqueued_at: Option<DateTime<Utc>>,
334) -> Result<Option<JobRow>, AwaError>
335where
336 E: PgExecutor<'e>,
337{
338 let row = sqlx::query_as::<_, JobRow>(
339 r#"
340 WITH mark AS (
341 UPDATE awa.cron_jobs
342 SET last_enqueued_at = $2, updated_at = now()
343 WHERE name = $1
344 AND (last_enqueued_at IS NOT DISTINCT FROM $3)
345 RETURNING name, kind, queue, args, priority, max_attempts, tags, metadata
346 )
347 SELECT job.*
348 FROM mark
349 CROSS JOIN LATERAL awa.insert_job_compat(
350 mark.kind,
351 mark.queue,
352 mark.args,
353 'available'::awa.job_state,
354 mark.priority,
355 mark.max_attempts,
356 NULL,
357 mark.metadata || jsonb_build_object('cron_name', mark.name, 'cron_fire_time', $2::text),
358 mark.tags,
359 NULL,
360 NULL
361 ) AS job
362 "#,
363 )
364 .bind(cron_name)
365 .bind(fire_time)
366 .bind(previous_enqueued_at)
367 .fetch_optional(executor)
368 .await?;
369
370 Ok(row)
371}
372
373pub async fn trigger_cron_job<'e, E>(executor: E, name: &str) -> Result<JobRow, AwaError>
379where
380 E: PgExecutor<'e>,
381{
382 let row = sqlx::query_as::<_, JobRow>(
383 r#"
384 WITH cron AS (
385 SELECT name, kind, queue, args, priority, max_attempts, tags, metadata
386 FROM awa.cron_jobs
387 WHERE name = $1
388 )
389 SELECT job.*
390 FROM cron
391 CROSS JOIN LATERAL awa.insert_job_compat(
392 cron.kind,
393 cron.queue,
394 cron.args,
395 'available'::awa.job_state,
396 cron.priority,
397 cron.max_attempts,
398 NULL,
399 cron.metadata || jsonb_build_object('cron_name', cron.name, 'triggered_manually', true),
400 cron.tags,
401 NULL,
402 NULL
403 ) AS job
404 "#,
405 )
406 .bind(name)
407 .fetch_optional(executor)
408 .await?;
409
410 row.ok_or_else(|| AwaError::Validation(format!("cron job not found: {name}")))
411}
412
413#[cfg(test)]
414mod tests {
415 use super::*;
416 use chrono::TimeZone;
417
418 fn make_periodic(cron_expr: &str, timezone: &str) -> PeriodicJob {
419 PeriodicJob {
420 name: "test".to_string(),
421 cron_expr: cron_expr.to_string(),
422 timezone: timezone.to_string(),
423 kind: "test_job".to_string(),
424 queue: "default".to_string(),
425 args: serde_json::json!({}),
426 priority: 2,
427 max_attempts: 25,
428 tags: vec![],
429 metadata: serde_json::json!({}),
430 }
431 }
432
433 #[test]
434 fn test_valid_cron_expression() {
435 let result = PeriodicJob::builder("test", "0 9 * * *")
436 .build_raw("test_job".to_string(), serde_json::json!({}));
437 assert!(result.is_ok());
438 }
439
440 #[test]
441 fn test_invalid_cron_expression() {
442 let result = PeriodicJob::builder("test", "not a cron")
443 .build_raw("test_job".to_string(), serde_json::json!({}));
444 assert!(result.is_err());
445 let err = result.unwrap_err();
446 assert!(
447 err.to_string().contains("invalid cron expression"),
448 "got: {err}"
449 );
450 }
451
452 #[test]
453 fn test_invalid_timezone() {
454 let result = PeriodicJob::builder("test", "0 9 * * *")
455 .timezone("Not/A/Timezone")
456 .build_raw("test_job".to_string(), serde_json::json!({}));
457 assert!(result.is_err());
458 let err = result.unwrap_err();
459 assert!(err.to_string().contains("invalid timezone"), "got: {err}");
460 }
461
462 #[test]
463 fn test_builder_defaults() {
464 let job = PeriodicJob::builder("daily_report", "0 9 * * *")
465 .build_raw(
466 "daily_report".to_string(),
467 serde_json::json!({"format": "pdf"}),
468 )
469 .unwrap();
470 assert_eq!(job.name, "daily_report");
471 assert_eq!(job.timezone, "UTC");
472 assert_eq!(job.queue, "default");
473 assert_eq!(job.priority, 2);
474 assert_eq!(job.max_attempts, 25);
475 assert!(job.tags.is_empty());
476 }
477
478 #[test]
479 fn test_builder_custom_fields() {
480 let job = PeriodicJob::builder("report", "0 9 * * *")
481 .timezone("Pacific/Auckland")
482 .queue("reports")
483 .priority(1)
484 .max_attempts(3)
485 .tags(vec!["important".to_string()])
486 .metadata(serde_json::json!({"source": "cron"}))
487 .build_raw("daily_report".to_string(), serde_json::json!({}))
488 .unwrap();
489 assert_eq!(job.timezone, "Pacific/Auckland");
490 assert_eq!(job.queue, "reports");
491 assert_eq!(job.priority, 1);
492 assert_eq!(job.max_attempts, 3);
493 assert_eq!(job.tags, vec!["important"]);
494 }
495
496 #[test]
497 fn test_latest_fire_time_finds_past_fire() {
498 let pj = make_periodic("0 * * * *", "UTC");
500 let now = Utc.with_ymd_and_hms(2025, 6, 15, 14, 35, 0).unwrap();
501 let after = Some(Utc.with_ymd_and_hms(2025, 6, 15, 13, 0, 0).unwrap());
502
503 let fire = pj.latest_fire_time(now, after);
504 assert_eq!(
505 fire,
506 Some(Utc.with_ymd_and_hms(2025, 6, 15, 14, 0, 0).unwrap())
507 );
508 }
509
510 #[test]
511 fn test_no_fire_when_next_is_future() {
512 let pj = make_periodic("0 * * * *", "UTC");
514 let now = Utc.with_ymd_and_hms(2025, 6, 15, 14, 35, 0).unwrap();
515 let after = Some(Utc.with_ymd_and_hms(2025, 6, 15, 14, 0, 0).unwrap());
517
518 let fire = pj.latest_fire_time(now, after);
519 assert!(fire.is_none(), "Should not fire until 15:00");
520 }
521
522 #[test]
523 fn test_first_registration_null_last_enqueued() {
524 let pj = make_periodic("0 * * * *", "UTC");
526 let now = Utc.with_ymd_and_hms(2025, 6, 15, 14, 35, 0).unwrap();
527
528 let fire = pj.latest_fire_time(now, None);
529 assert_eq!(
530 fire,
531 Some(Utc.with_ymd_and_hms(2025, 6, 15, 14, 0, 0).unwrap()),
532 "Should enqueue the most recent past fire on first registration"
533 );
534 }
535
536 #[test]
537 fn test_no_backfill_only_latest_fire() {
538 let pj = make_periodic("* * * * *", "UTC");
540 let now = Utc.with_ymd_and_hms(2025, 6, 15, 15, 0, 0).unwrap();
541 let after = Some(Utc.with_ymd_and_hms(2025, 6, 15, 14, 0, 0).unwrap());
542
543 let fire = pj.latest_fire_time(now, after);
544 assert_eq!(
546 fire,
547 Some(Utc.with_ymd_and_hms(2025, 6, 15, 15, 0, 0).unwrap())
548 );
549 }
550
551 #[test]
552 fn test_timezone_aware_fire_time() {
553 let pj = make_periodic("0 9 * * *", "Pacific/Auckland");
555 let now = Utc.with_ymd_and_hms(2025, 6, 15, 21, 30, 0).unwrap();
558 let after = Some(Utc.with_ymd_and_hms(2025, 6, 14, 21, 0, 0).unwrap());
559
560 let fire = pj.latest_fire_time(now, after);
561 assert_eq!(
563 fire,
564 Some(Utc.with_ymd_and_hms(2025, 6, 15, 21, 0, 0).unwrap())
565 );
566 }
567
568 #[test]
569 fn test_dst_spring_forward() {
570 let pj = make_periodic("30 2 * * *", "US/Eastern");
574 let now = Utc.with_ymd_and_hms(2025, 3, 9, 12, 0, 0).unwrap();
575 let after = Some(Utc.with_ymd_and_hms(2025, 3, 8, 12, 0, 0).unwrap());
576
577 let fire = pj.latest_fire_time(now, after);
578 let fire_count = if fire.is_some() { 1 } else { 0 };
581 assert!(
582 fire_count <= 1,
583 "Should fire at most once during spring-forward"
584 );
585 }
586
587 #[test]
588 fn test_dst_fall_back() {
589 let pj = make_periodic("30 1 * * *", "US/Eastern");
592 let now = Utc.with_ymd_and_hms(2025, 11, 2, 12, 0, 0).unwrap();
593 let after = Some(Utc.with_ymd_and_hms(2025, 11, 1, 12, 0, 0).unwrap());
594
595 let fire = pj.latest_fire_time(now, after);
596 assert!(fire.is_some(), "Should fire once during fall-back");
597
598 let fire_time = fire.unwrap();
600 let second_fire = pj.latest_fire_time(now, Some(fire_time));
601 assert!(
602 second_fire.is_none(),
603 "Should not fire a second time during fall-back"
604 );
605 }
606
607 #[test]
608 fn test_invalid_priority() {
609 let result = PeriodicJob::builder("test", "0 9 * * *")
610 .priority(5)
611 .build_raw("test_job".to_string(), serde_json::json!({}));
612 assert!(result.is_err());
613 }
614
615 #[test]
616 fn test_invalid_max_attempts() {
617 let result = PeriodicJob::builder("test", "0 9 * * *")
618 .max_attempts(0)
619 .build_raw("test_job".to_string(), serde_json::json!({}));
620 assert!(result.is_err());
621 }
622
623 #[test]
624 fn test_next_fire_time_exact() {
625 let now = Utc.with_ymd_and_hms(2025, 6, 15, 14, 35, 0).unwrap();
627 let next = next_fire_time_after("0 * * * *", "UTC", now);
628 assert_eq!(
629 next,
630 Some(Utc.with_ymd_and_hms(2025, 6, 15, 15, 0, 0).unwrap())
631 );
632 }
633
634 #[test]
635 fn test_next_fire_time_respects_timezone() {
636 let now = Utc.with_ymd_and_hms(2025, 6, 15, 20, 0, 0).unwrap();
639 let next = next_fire_time_after("0 9 * * *", "Pacific/Auckland", now);
640 assert_eq!(
641 next,
642 Some(Utc.with_ymd_and_hms(2025, 6, 15, 21, 0, 0).unwrap())
643 );
644
645 let next_utc = next_fire_time_after("0 9 * * *", "UTC", now);
647 assert_eq!(
648 next_utc,
649 Some(Utc.with_ymd_and_hms(2025, 6, 16, 9, 0, 0).unwrap())
650 );
651 }
652
653 #[test]
654 fn test_next_fire_time_dst_boundary() {
655 let now = Utc.with_ymd_and_hms(2025, 3, 9, 6, 30, 0).unwrap();
658 let next = next_fire_time_after("0 * * * *", "US/Eastern", now);
659 assert!(next.is_some());
660 let next = next.unwrap();
662 assert!(
663 next >= Utc.with_ymd_and_hms(2025, 3, 9, 7, 0, 0).unwrap(),
664 "should skip the non-existent 2:00 AM, got {next}"
665 );
666 }
667
668 #[test]
669 fn test_next_fire_time_invalid_input() {
670 let now = Utc::now();
671 assert!(next_fire_time_after("not a cron", "UTC", now).is_none());
672 assert!(next_fire_time_after("* * * * *", "Not/A/Zone", now).is_none());
673 }
674
675 #[test]
678 fn test_six_field_cron_accepted_by_builder() {
679 let result = PeriodicJob::builder("test", "30 0 9 * * *")
680 .build_raw("test_job".to_string(), serde_json::json!({}));
681 assert!(result.is_ok(), "6-field cron should be accepted");
682 }
683
684 #[test]
685 fn test_six_field_cron_every_15_seconds() {
686 let result = PeriodicJob::builder("fast", "*/15 * * * * *")
688 .build_raw("fast_job".to_string(), serde_json::json!({}));
689 assert!(result.is_ok(), "every-15-seconds cron should be accepted");
690 }
691
692 #[test]
693 fn test_six_field_next_fire_time() {
694 let now = Utc.with_ymd_and_hms(2025, 6, 15, 8, 0, 0).unwrap();
696 let next = next_fire_time_after("30 0 9 * * *", "UTC", now);
697 assert_eq!(
698 next,
699 Some(
700 Utc.with_ymd_and_hms(2025, 6, 15, 9, 0, 0).unwrap() + chrono::Duration::seconds(30)
701 ),
702 "should fire at 09:00:30"
703 );
704 }
705
706 #[test]
707 fn test_six_field_every_15_seconds_next_fire() {
708 let now = Utc.with_ymd_and_hms(2025, 6, 15, 14, 35, 7).unwrap();
710 let next = next_fire_time_after("*/15 * * * * *", "UTC", now);
711 assert_eq!(
712 next,
713 Some(Utc.with_ymd_and_hms(2025, 6, 15, 14, 35, 15).unwrap()),
714 "should fire at next 15-second boundary"
715 );
716 }
717
718 #[test]
719 fn test_five_field_still_works() {
720 let result = PeriodicJob::builder("classic", "0 9 * * *")
722 .build_raw("classic_job".to_string(), serde_json::json!({}));
723 assert!(result.is_ok(), "5-field cron should still be accepted");
724
725 let now = Utc.with_ymd_and_hms(2025, 6, 15, 8, 0, 0).unwrap();
726 let next = next_fire_time_after("0 9 * * *", "UTC", now);
727 assert_eq!(
728 next,
729 Some(Utc.with_ymd_and_hms(2025, 6, 15, 9, 0, 0).unwrap())
730 );
731 }
732
733 #[test]
734 fn test_six_field_latest_fire_time() {
735 let job = PeriodicJob::builder("sec_job", "0 */5 * * * *")
737 .build_raw("sec_job".to_string(), serde_json::json!({}))
738 .unwrap();
739
740 let now = Utc.with_ymd_and_hms(2025, 6, 15, 12, 7, 0).unwrap();
741 let latest = job.latest_fire_time(now, None);
742 assert!(
743 latest.is_some(),
744 "6-field cron should produce a latest fire time"
745 );
746 assert_eq!(
749 latest.unwrap(),
750 Utc.with_ymd_and_hms(2025, 6, 15, 12, 5, 0).unwrap()
751 );
752 }
753}