//! In-process scheduler, preserving APScheduler's Monday=0 weekday convention. use crate::{AppState, backup, models::Settings, routeros}; use anyhow::{Context, Result, bail, ensure}; use chrono::{DateTime, Datelike, Duration, TimeZone, Timelike, Utc}; use chrono_tz::Tz; use serde_json::{Value, json}; use std::{ collections::{HashMap, HashSet}, sync::{Arc, Mutex}, time::Duration as StdDuration, }; use tokio::sync::watch; #[derive(Clone)] struct Field { values: HashSet, } #[derive(Clone)] struct Cron { minute: Field, hour: Field, day: Field, month: Field, weekday: Field, last_day: bool, timezone: Tz, } fn value(raw: &str, minimum: u32, maximum: u32, names: &[&str]) -> Result { let v = if let Ok(number) = raw.parse() { number } else { names .iter() .position(|name| raw.eq_ignore_ascii_case(name)) .map(|i| i as u32 + minimum) .context("Invalid cron field value")? }; ensure!( (minimum..=maximum).contains(&v), "Cron field value is out of range" ); Ok(v) } fn field(raw: &str, minimum: u32, maximum: u32, names: &[&str]) -> Result { let mut values = HashSet::new(); for item in raw.split(',') { let (range, step) = if let Some((range, step)) = item.split_once('/') { (range, step.parse::().context("Invalid cron step")?) } else { (item, 1) }; ensure!( step > 0 && step <= maximum - minimum + 1, "Cron step is out of range" ); let (start, end) = if range == "*" { (minimum, maximum) } else if let Some((a, b)) = range.split_once('-') { ( value(a, minimum, maximum, names)?, value(b, minimum, maximum, names)?, ) } else { let start = value(range, minimum, maximum, names)?; (start, if item.contains('/') { maximum } else { start }) }; ensure!(start <= end, "Cron range start exceeds end"); for v in (start..=end).step_by(step as usize) { values.insert(v); } } ensure!(!values.is_empty(), "Empty cron field"); Ok(Field { values }) } impl Cron { fn parse(expression: &str, timezone: &str) -> Result { let parts: Vec<_> = expression.split_whitespace().collect(); ensure!( parts.len() == 5, "Cron expression must contain exactly 5 fields" ); let last_day = parts[2].eq_ignore_ascii_case("last"); Ok(Self { minute: field(parts[0], 0, 59, &[])?, hour: field(parts[1], 0, 23, &[])?, day: field(if last_day { "*" } else { parts[2] }, 1, 31, &[])?, month: field( parts[3], 1, 12, &[ "jan", "feb", "mar", "apr", "may", "jun", "jul", "aug", "sep", "oct", "nov", "dec", ], )?, weekday: field( parts[4], 0, 6, &["mon", "tue", "wed", "thu", "fri", "sat", "sun"], )?, last_day, timezone: timezone.parse().context("Unknown timezone")?, }) } fn matches(&self, time: DateTime) -> bool { let local = time.with_timezone(&self.timezone); let last = !self.last_day || local .date_naive() .succ_opt() .is_some_and(|next| next.month() != local.month()); self.minute.values.contains(&local.minute()) && self.hour.values.contains(&local.hour()) && self.day.values.contains(&local.day()) && self.month.values.contains(&local.month()) && self .weekday .values .contains(&local.weekday().num_days_from_monday()) && last } fn next(&self, from: DateTime, count: usize) -> Vec> { let Some(mut time) = Utc .timestamp_opt((from.timestamp() / 60 + 1) * 60, 0) .single() else { return Vec::new(); }; let mut runs = Vec::new(); let limit = time + Duration::days(366 * 400); while runs.len() < count && time < limit { let local = time.with_timezone(&self.timezone); if !self.month.values.contains(&local.month()) || !self.day.values.contains(&local.day()) || !self .weekday .values .contains(&local.weekday().num_days_from_monday()) || (self.last_day && local .date_naive() .succ_opt() .is_some_and(|next| next.month() == local.month())) { // Jump to the next local midnight, respecting timezone transitions. if let Some(next_date) = local.date_naive().succ_opt() && let Some(midnight) = next_date.and_hms_opt(0, 0, 0) && let Some(next) = self.timezone.from_local_datetime(&midnight).earliest() { let next = next.with_timezone(&Utc); if next > time { time = next; continue; } } } if self.matches(time) { runs.push(time.with_timezone(&self.timezone)); } time += Duration::minutes(1); } runs } } pub fn validate_cron(expression: &str, timezone: &str) -> Result<()> { Cron::parse(expression, timezone).map(|_| ()) } fn cron_status(key: &str, label: &str, enabled: bool, expression: &str, timezone: &str) -> Value { if !enabled || expression.trim().is_empty() { return json!({"key":key,"label":label,"enabled":false,"cron":if expression.is_empty(){None}else{Some(expression)},"description":"settings.scheduleDisabledHint","description_params":null,"valid":true,"next_runs":[],"error":null}); } match Cron::parse(expression, timezone) { Ok(cron) => { json!({"key":key,"label":label,"enabled":true,"cron":expression,"description":"settings.schedulerCronDescription","description_params":{"description":expression},"valid":true,"next_runs":cron.next(Utc::now(),3),"error":null}) } Err(error) => { json!({"key":key,"label":label,"enabled":true,"cron":expression,"description":"settings.schedulerInvalidCron","description_params":null,"valid":false,"next_runs":[],"error":error.to_string()}) } } } pub fn status(state: &AppState) -> Result { let s = state.db.get_settings()?; let minutes = s.connection_test_interval_minutes; let now = Utc::now(); let runs: Vec<_> = if minutes > 0 { (1..=3) .map(|n| now + Duration::minutes(minutes * n)) .collect() } else { Vec::new() }; Ok( json!({"timezone":state.config.timezone,"running":true,"jobs":[ cron_status("auto_export","settings.schedulerAutoExportLabel",s.enable_auto_export,&s.export_cron,&state.config.timezone), cron_status("auto_binary","settings.schedulerBinaryLabel",!s.binary_cron.is_empty(),&s.binary_cron,&state.config.timezone), cron_status("retention","settings.schedulerRetentionLabel",!s.retention_cron.is_empty(),&s.retention_cron,&state.config.timezone), {"key":"connection_probe","label":"settings.schedulerConnectionLabel","enabled":minutes>0,"cron":null,"description":if minutes>0{"settings.connectionTestsEverySummary"}else{"settings.connectionTestsDisabledHint"},"description_params":{"minutes":minutes},"valid":true,"next_runs":runs,"error":null}, {"key":"log_cleanup","label":"settings.schedulerLogsLabel","enabled":true,"cron":null,"description":"settings.schedulerLogsDescription","description_params":null,"valid":true,"next_runs":[],"error":null} ]}), ) } struct RunningGuard { key: String, set: Arc>>, } impl Drop for RunningGuard { fn drop(&mut self) { if let Ok(mut set) = self.set.lock() { set.remove(&self.key); } } } async fn job(state: &AppState, key: &str) -> Result<()> { match key { "auto_export" => { backup::run_all(state, None, "export").await?; } "auto_binary" => { backup::run_all(state, None, "binary").await?; } "retention" => { backup::cleanup(state).await?; } "connection_probe" => { for router in state.db.routers(None)? { routeros::test_connection(state, &router).await?; } } "log_cleanup" => { let s = state.db.get_settings()?; let cutoff = (Utc::now() - Duration::days(s.log_retention_days.max(0))).to_rfc3339(); let count = state.db.with_conn(|conn| { Ok(conn.execute( "DELETE FROM operation_logs WHERE julianday(timestamp) bail!("Unknown scheduler job"), } Ok(()) } fn signature(s: &Settings) -> String { format!( "{}|{}|{}|{}|{}", s.enable_auto_export, s.export_cron, s.binary_cron, s.retention_cron, s.connection_test_interval_minutes ) } pub fn start(state: AppState, mut shutdown: watch::Receiver) -> tokio::task::JoinHandle<()> { tokio::spawn(async move { let mut tick = tokio::time::interval(StdDuration::from_secs(1)); tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); let running = Arc::new(Mutex::new(HashSet::::new())); let mut tasks = tokio::task::JoinSet::new(); let mut schedules: HashMap)> = HashMap::new(); let mut last_signature = String::new(); let mut next_probe = None; let mut next_logs = Utc::now() + Duration::days(1); loop { tokio::select! { changed=shutdown.changed()=>{if changed.is_err()||*shutdown.borrow(){break;}}, Some(_)=tasks.join_next(),if !tasks.is_empty()=>{}, _=tick.tick()=>{ let now=Utc::now();let settings=match state.db.get_settings(){Ok(s)=>s,Err(error)=>{tracing::error!("Scheduler settings unavailable: {error}");continue;}}; let current_signature=signature(&settings); if current_signature!=last_signature{ schedules.clear(); for (key,enabled,expression) in [("auto_export",settings.enable_auto_export,settings.export_cron.as_str()),("auto_binary",true,settings.binary_cron.as_str()),("retention",true,settings.retention_cron.as_str())]{ if enabled&&!expression.trim().is_empty(){match Cron::parse(expression,&state.config.timezone){Ok(cron)=>{if let Some(next)=cron.next(now,1).first(){schedules.insert(key.to_owned(),(cron.clone(),next.with_timezone(&Utc)));}},Err(error)=>{let _=state.db.log(&format!("Scheduler skipped invalid {key} cron: {error}"));}}} } next_probe=(settings.connection_test_interval_minutes>0).then(||now+Duration::minutes(settings.connection_test_interval_minutes));last_signature=current_signature; } let mut due=Vec::new(); for (key,(cron,next)) in &mut schedules{if now>=*next{if now.signed_duration_since(*next)<=Duration::minutes(5){due.push(key.clone());} if let Some(value)=cron.next(now,1).first(){*next=value.with_timezone(&Utc);}else{*next=now+Duration::days(366*400);}}} if next_probe.is_some_and(|time|now>=time){due.push("connection_probe".into());next_probe=Some(now+Duration::minutes(settings.connection_test_interval_minutes.max(1)));} if now>=next_logs{due.push("log_cleanup".into());next_logs=now+Duration::days(1);} for key in due{let inserted=running.lock().map(|mut set|set.insert(key.clone())).unwrap_or(false);if !inserted{continue;}let state=state.clone();let guard=RunningGuard{key:key.clone(),set:running.clone()};tasks.spawn(async move{let _guard=guard;if let Err(error)=job(&state,&key).await{let _=state.db.log(&format!("Scheduled {key} failed: {error:#}"));}});} } } } // Allow active jobs to finish; individual network operations have hard deadlines. while tasks.join_next().await.is_some() {} }) } #[cfg(test)] mod tests { use super::*; #[test] fn validates_cron_and_ranges() { assert!(validate_cron("15 2 * * *", "Europe/Warsaw").is_ok()); for expression in [ "bad cron", "61 * * * *", "*/0 * * * *", "0 0 * * 7", "0 0 * 13 *", ] { assert!(validate_cron(expression, "Europe/Warsaw").is_err()); } } #[test] fn monday_zero_matches_apscheduler() { let cron = Cron::parse("0 2 * * 0", "UTC").unwrap(); assert!(cron.matches(Utc.with_ymd_and_hms(2026, 10, 26, 2, 0, 0).unwrap())); assert!(!cron.matches(Utc.with_ymd_and_hms(2026, 10, 25, 2, 0, 0).unwrap())); } #[test] fn timezone_and_dst_preview() { let cron = Cron::parse("30 2 * * *", "Europe/Warsaw").unwrap(); let start = Utc.with_ymd_and_hms(2026, 3, 28, 12, 0, 0).unwrap(); let runs = cron.next(start, 2); assert_eq!(runs.len(), 2); assert_eq!(runs[0].day(), 30); assert_eq!(runs[0].hour(), 2); assert!(runs[0] < runs[1]); } #[test] fn named_weekdays_and_last_day() { let cron = Cron::parse("0 0 last * *", "UTC").unwrap(); assert!(cron.matches(Utc.with_ymd_and_hms(2028, 2, 29, 0, 0, 0).unwrap())); assert!(!cron.matches(Utc.with_ymd_and_hms(2028, 2, 28, 0, 0, 0).unwrap())); assert!(validate_cron("*/15 1-6/2 * jan-mar mon-fri", "UTC").is_ok()); } #[test] fn rare_calendar_schedule_is_not_lost() { let cron = Cron::parse("0 0 29 feb mon", "UTC").unwrap(); let runs = cron.next(Utc.with_ymd_and_hms(2026, 10, 27, 0, 0, 0).unwrap(), 1); assert_eq!(runs.len(), 1); assert_eq!(runs[0].year(), 2044); } }