Skip to main content

awa_model/
cron.rs

1//! Periodic/cron job types and database operations.
2//!
3//! Schedules are defined in application code, synced to `awa.cron_jobs` via UPSERT,
4//! and evaluated by the leader to atomically enqueue jobs.
5
6use crate::error::AwaError;
7use crate::job::JobRow;
8use chrono::{DateTime, Utc};
9use croner::Cron;
10use serde::Serialize;
11use sqlx::PgExecutor;
12
13/// A periodic job schedule definition.
14///
15/// Created via `PeriodicJob::builder(name, cron_expr).build(args)`.
16#[derive(Debug, Clone)]
17pub struct PeriodicJob {
18    /// Unique name identifying this schedule (e.g., "daily_report").
19    pub name: String,
20    /// Cron expression (e.g., "0 9 * * *").
21    pub cron_expr: String,
22    /// IANA timezone (e.g., "Pacific/Auckland"). Defaults to "UTC".
23    pub timezone: String,
24    /// Job kind (derived from JobArgs trait).
25    pub kind: String,
26    /// Target queue. Defaults to "default".
27    pub queue: String,
28    /// Serialized job arguments.
29    pub args: serde_json::Value,
30    /// Job priority (1-4). Defaults to 2.
31    pub priority: i16,
32    /// Max retry attempts. Defaults to 25.
33    pub max_attempts: i16,
34    /// Tags attached to created jobs.
35    pub tags: Vec<String>,
36    /// Extra metadata merged into created jobs.
37    pub metadata: serde_json::Value,
38}
39
40impl PeriodicJob {
41    /// Start building a periodic job with a name and cron expression.
42    ///
43    /// The cron expression is validated eagerly — invalid expressions
44    /// cause `build()` to return an error.
45    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    /// Compute the latest fire time <= `now` that is strictly after `after`.
59    ///
60    /// Returns `None` if no fire time exists in the range (after, now].
61    /// This handles both "first registration" (after=None → find latest past fire)
62    /// and "regular evaluation" (after=Some(last_enqueued_at)).
63    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        // Walk backwards from now to find the most recent fire time.
81        // croner doesn't have a "previous" iterator, so we find the fire time
82        // by iterating forward from a start point.
83        let search_start = match after {
84            Some(after_time) => after_time.with_timezone(&tz),
85            // For first registration, search from 24h ago to avoid unbounded iteration
86            None => now_tz - chrono::Duration::hours(24),
87        };
88
89        let mut latest_fire: Option<DateTime<Utc>> = None;
90
91        // Iterate forward from search_start, collecting fire times <= now
92        for fire_time in cron.clone().iter_from(search_start) {
93            let fire_utc = fire_time.with_timezone(&Utc);
94
95            // Stop once we've passed now
96            if fire_utc > now {
97                break;
98            }
99
100            // Skip fires at or before the `after` boundary
101            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/// Builder for `PeriodicJob`.
115#[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    /// Set the IANA timezone (e.g., "Pacific/Auckland").
129    pub fn timezone(mut self, timezone: impl Into<String>) -> Self {
130        self.timezone = timezone.into();
131        self
132    }
133
134    /// Set the target queue.
135    pub fn queue(mut self, queue: impl Into<String>) -> Self {
136        self.queue = queue.into();
137        self
138    }
139
140    /// Set the job priority (1-4).
141    pub fn priority(mut self, priority: i16) -> Self {
142        self.priority = priority;
143        self
144    }
145
146    /// Set the max retry attempts.
147    pub fn max_attempts(mut self, max_attempts: i16) -> Self {
148        self.max_attempts = max_attempts;
149        self
150    }
151
152    /// Set tags for created jobs.
153    pub fn tags(mut self, tags: Vec<String>) -> Self {
154        self.tags = tags;
155        self
156    }
157
158    /// Set extra metadata for created jobs.
159    pub fn metadata(mut self, metadata: serde_json::Value) -> Self {
160        self.metadata = metadata;
161        self
162    }
163
164    /// Build the periodic job, validating the cron expression and timezone.
165    ///
166    /// The `args` parameter must implement `JobArgs` — the kind is derived
167    /// from the type and args are serialized to JSON.
168    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    /// Build from raw kind and args JSON (used by Python bindings).
173    pub fn build_raw(self, kind: String, args: serde_json::Value) -> Result<PeriodicJob, AwaError> {
174        // Validate cron expression
175        Cron::new(&self.cron_expr)
176            .with_seconds_optional()
177            .parse()
178            .map_err(|err| AwaError::Validation(format!("invalid cron expression: {err}")))?;
179
180        // Validate timezone
181        self.timezone
182            .parse::<chrono_tz::Tz>()
183            .map_err(|err| AwaError::Validation(format!("invalid timezone: {err}")))?;
184
185        // Validate priority
186        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        // Validate max_attempts
194        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/// A row from the `awa.cron_jobs` table.
217#[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
234/// Upsert a periodic job schedule into `awa.cron_jobs`.
235///
236/// Additive only — never deletes rows not in the input set.
237pub 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
274/// Compute the next fire time for a cron expression after now.
275///
276/// Returns `None` if the expression or timezone is invalid.
277pub fn next_fire_time(cron_expr: &str, timezone: &str) -> Option<DateTime<Utc>> {
278    next_fire_time_after(cron_expr, timezone, Utc::now())
279}
280
281/// Compute the next fire time for a cron expression after a given timestamp.
282///
283/// Testable variant — accepts an explicit `after` time instead of using the clock.
284/// Returns `None` if the expression or timezone is invalid.
285pub 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
297/// Load all cron job rows from `awa.cron_jobs`.
298pub 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
308/// Delete a cron job schedule by name.
309pub 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
320/// Atomically mark a cron job as enqueued AND insert the resulting job.
321///
322/// Uses a single CTE so that both the UPDATE and INSERT happen in one
323/// atomic operation. If the process crashes mid-transaction, Postgres
324/// rolls back both. If another leader already claimed this fire time
325/// (last_enqueued_at no longer matches), the UPDATE matches 0 rows
326/// and the INSERT produces nothing.
327///
328/// Returns the inserted job row, or `None` if the fire was already claimed.
329pub 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
373/// Trigger an immediate run of a cron job without updating last_enqueued_at.
374///
375/// Reads the cron job config from `awa.cron_jobs` and inserts a new job
376/// directly. Does NOT update `last_enqueued_at` so the normal schedule
377/// is unaffected.
378pub 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        // Every hour at :00
499        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        // Every hour at :00
513        let pj = make_periodic("0 * * * *", "UTC");
514        let now = Utc.with_ymd_and_hms(2025, 6, 15, 14, 35, 0).unwrap();
515        // Already fired at 14:00
516        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        // Every hour at :00, registered at 14:35 with no previous fire
525        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        // Every minute, last enqueued 1 hour ago
539        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        // Should return 15:00, not 14:01 — only the latest missed fire
545        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        // 9 AM daily in Auckland timezone
554        let pj = make_periodic("0 9 * * *", "Pacific/Auckland");
555        // It's 2025-06-15 21:30 UTC = 2025-06-16 09:30 NZST
556        // So 09:00 NZST on June 16 = 21:00 UTC on June 15
557        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        // 09:00 NZST on June 16 = 21:00 UTC on June 15
562        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        // 2:30 AM US/Eastern on March 9 2025 — clocks spring forward from 2:00 to 3:00
571        // Schedule at 2:30 AM should fire once (the 2:30 time doesn't exist, so croner
572        // should skip it or fire at the next valid time)
573        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        // On spring-forward day, 2:30 AM doesn't exist. croner may skip it entirely
579        // or map it to 3:30 AM. Either way, we should get at most one fire.
580        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        // 1:30 AM US/Eastern on Nov 2 2025 — clocks fall back from 2:00 to 1:00
590        // 1:30 AM happens twice. Should fire exactly once.
591        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        // Verify it's only one fire by checking that after this fire, no more fires exist
599        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        // At 14:35 UTC, next fire for "every hour at :00" should be 15:00
626        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        // At 2025-06-15 20:00 UTC, next "9 AM daily" in Auckland should be
637        // 2025-06-15 21:00 UTC (= 2025-06-16 09:00 NZST, UTC+12 in June)
638        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        // Same time, UTC schedule — next 9 AM UTC is the next day
646        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        // US/Eastern spring-forward: 2025-03-09 at 2:00 AM clocks jump to 3:00 AM
656        // At 1:30 AM EST (06:30 UTC), next "every hour at :00" should skip 2:00 AM
657        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        // The 2:00 AM hour doesn't exist; croner should give us 3:00 AM EDT (07:00 UTC)
661        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    // ── 6-field cron (seconds precision) ──────────────────────────
676
677    #[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        // "*/15 * * * * *" = every 15 seconds
687        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        // "30 0 9 * * *" = daily at 09:00:30 UTC
695        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        // "*/15 * * * * *" at 14:35:07 → next fire at 14:35:15
709        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        // Ensure 5-field expressions continue to work as before
721        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        // Verify latest_fire_time also handles 6-field
736        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        // "0 */5 * * * *" = at second 0 of every 5th minute
747        // At 12:07:00, latest fire was 12:05:00
748        assert_eq!(
749            latest.unwrap(),
750            Utc.with_ymd_and_hms(2025, 6, 15, 12, 5, 0).unwrap()
751        );
752    }
753}