stasis/domain/runtime/
recurring.rs

1use chrono::{DateTime, Utc};
2use chrono_tz::Tz;
3use cron::Schedule;
4use std::str::FromStr;
5
6use crate::domain::errors::{Result, StasisError};
7
8#[derive(Clone, Debug)]
9pub struct RecurringDefinition {
10    pub id: String,
11    pub queue: String,
12    pub job_type: String,
13    pub payload_template_ref: String,
14    pub cron_expr: String,
15    pub timezone: String,
16    pub jitter_seconds: i64,
17    pub enabled: bool,
18    pub max_attempts: u32,
19    pub next_run_at: DateTime<Utc>,
20    pub last_run_at: Option<DateTime<Utc>>,
21    pub lease_owner: Option<String>,
22    pub lease_expires_at: Option<DateTime<Utc>>,
23}
24
25impl RecurringDefinition {
26    pub fn compute_next_run_at(&self, from: DateTime<Utc>) -> Result<DateTime<Utc>> {
27        let schedule = Schedule::from_str(&self.cron_expr).map_err(|e| {
28            StasisError::PortFailure(format!(
29                "invalid cron expression for recurring_id={}: {}",
30                self.id, e
31            ))
32        })?;
33
34        let tz: Tz = self.timezone.parse().map_err(|e| {
35            StasisError::PortFailure(format!(
36                "invalid timezone for recurring_id={}: {}",
37                self.id, e
38            ))
39        })?;
40
41        let local_from = from.with_timezone(&tz);
42        let next_local = schedule.after(&local_from).next().ok_or_else(|| {
43            StasisError::PortFailure(format!(
44                "could not compute next run for recurring_id={}",
45                self.id
46            ))
47        })?;
48
49        Ok(next_local.with_timezone(&Utc))
50    }
51}
stasis/domain/runtime/recurring.rs

stasis/domain/runtime/
recurring.rs
