use chrono::{DateTime, Duration, Utc};
use cron::Schedule as CronSchedule;
use serde::{Deserialize, Serialize};
use std::str::FromStr;
use super::JobError;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum JobSchedule {
Cron {
expression: String,
#[serde(skip)]
schedule: Option<Box<CronSchedule>>,
},
Delayed {
#[serde(with = "duration_ms")]
delay: std::time::Duration,
},
Recurring {
#[serde(with = "duration_ms")]
interval: std::time::Duration,
max_executions: Option<u64>,
},
}
impl JobSchedule {
pub fn cron(expression: &str) -> Result<Self, JobError> {
let schedule = CronSchedule::from_str(expression)
.map_err(|e| JobError::Other(format!("Invalid cron expression: {e}")))?;
Ok(Self::Cron {
expression: expression.to_string(),
schedule: Some(Box::new(schedule)),
})
}
#[must_use]
pub const fn after(delay: std::time::Duration) -> Self {
Self::Delayed { delay }
}
#[must_use]
pub const fn every(interval: std::time::Duration) -> Self {
Self::Recurring {
interval,
max_executions: None,
}
}
#[must_use]
pub fn with_max_executions(self, max: u64) -> Self {
match self {
Self::Recurring {
interval,
max_executions: _,
} => Self::Recurring {
interval,
max_executions: Some(max),
},
other => other,
}
}
#[must_use]
pub fn next_execution(&self, from: DateTime<Utc>) -> Option<DateTime<Utc>> {
match self {
Self::Cron {
expression: _,
schedule,
} => {
let sched = schedule.as_ref()?;
sched.after(&from).next()
}
Self::Delayed { delay } => {
let duration = Duration::from_std(*delay).ok()?;
Some(from + duration)
}
Self::Recurring {
interval,
max_executions: _,
} => {
let duration = Duration::from_std(*interval).ok()?;
Some(from + duration)
}
}
}
#[must_use]
pub const fn has_more_executions(&self, executions: u64) -> bool {
match self {
Self::Cron { .. } => true,
Self::Delayed { .. } => executions == 0,
Self::Recurring {
max_executions,
interval: _,
} => {
if let Some(max) = max_executions {
executions < *max
} else {
true
}
}
}
}
#[must_use]
pub fn description(&self) -> String {
match self {
Self::Cron { expression, .. } => format!("cron: {expression}"),
Self::Delayed { delay } => {
format!("delayed: {}s", delay.as_secs())
}
Self::Recurring {
interval,
max_executions,
} => max_executions.as_ref().map_or_else(
|| format!("every {}s", interval.as_secs()),
|max| format!("every {}s (max {max} times)", interval.as_secs()),
),
}
}
}
mod duration_ms {
use serde::{Deserialize, Deserializer, Serializer};
use std::time::Duration;
#[allow(clippy::trivially_copy_pass_by_ref)]
pub fn serialize<S>(duration: &Duration, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
serializer.serialize_u64(duration.as_millis().try_into().unwrap_or(u64::MAX))
}
pub fn deserialize<'de, D>(deserializer: D) -> Result<Duration, D::Error>
where
D: Deserializer<'de>,
{
let ms = u64::deserialize(deserializer)?;
Ok(Duration::from_millis(ms))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_cron_schedule_parsing() {
assert!(JobSchedule::cron("0 0 0 * * *").is_ok()); assert!(JobSchedule::cron("0 */15 * * * *").is_ok()); assert!(JobSchedule::cron("0 0 9 * * 1-5").is_ok());
assert!(JobSchedule::cron("invalid").is_err());
}
#[test]
fn test_delayed_schedule() {
let delay = std::time::Duration::from_secs(3600);
let schedule = JobSchedule::after(delay);
let now = Utc::now();
let next = schedule.next_execution(now).unwrap();
let diff = next.signed_duration_since(now);
assert!((diff.num_seconds() - 3600).abs() < 1);
}
#[test]
fn test_recurring_schedule() {
let interval = std::time::Duration::from_secs(60);
let schedule = JobSchedule::every(interval);
let now = Utc::now();
let next = schedule.next_execution(now).unwrap();
let diff = next.signed_duration_since(now);
assert!((diff.num_seconds() - 60).abs() < 1);
}
#[test]
fn test_recurring_with_max_executions() {
let schedule = JobSchedule::every(std::time::Duration::from_secs(60)).with_max_executions(5);
assert!(schedule.has_more_executions(0));
assert!(schedule.has_more_executions(4));
assert!(!schedule.has_more_executions(5));
assert!(!schedule.has_more_executions(10));
}
#[test]
fn test_delayed_single_execution() {
let schedule = JobSchedule::after(std::time::Duration::from_secs(60));
assert!(schedule.has_more_executions(0));
assert!(!schedule.has_more_executions(1));
}
#[test]
fn test_cron_next_execution() {
let schedule = JobSchedule::cron("0 0 0 * * *").unwrap();
let now = Utc::now();
let next = schedule.next_execution(now).unwrap();
assert!(next > now);
}
#[test]
fn test_schedule_description() {
let cron = JobSchedule::cron("0 0 0 * * *").unwrap();
assert!(cron.description().contains("cron"));
let delayed = JobSchedule::after(std::time::Duration::from_secs(3600));
assert!(delayed.description().contains("3600"));
let recurring = JobSchedule::every(std::time::Duration::from_secs(60));
assert!(recurring.description().contains("60"));
}
#[test]
fn test_serialization() {
let schedule = JobSchedule::every(std::time::Duration::from_secs(60)).with_max_executions(5);
let json = serde_json::to_string(&schedule).unwrap();
let deserialized: JobSchedule = serde_json::from_str(&json).unwrap();
match deserialized {
JobSchedule::Recurring {
interval,
max_executions,
} => {
assert_eq!(interval.as_secs(), 60);
assert_eq!(max_executions, Some(5));
}
_ => panic!("Expected Recurring schedule"),
}
}
}