//! Generic scheduled timers: fire MCP calls or prompts at a configured instant, //! with optional recurring re-arm. //! //! Separate from [`crate::service::timer::TimerStore`] which handles story-scoped //! rate-limit retry timers. This module provides a general-purpose scheduling //! primitive with explicit action types and unique IDs. use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use sqlx::SqlitePool; use std::path::{Path, PathBuf}; use std::sync::Mutex; use tokio::sync::mpsc; use uuid::Uuid; // ── Action ──────────────────────────────────────────────────────────────────── /// What to execute when a scheduled timer fires. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(tag = "type", rename_all = "snake_case")] pub enum TimerAction { /// Call an MCP tool by name with JSON arguments. Mcp { /// MCP tool name (e.g. `"start_agent"`, `"create_bug"`). method: String, /// JSON arguments object for the tool call. #[serde(default)] args: serde_json::Value, }, /// Broadcast a reminder text to the server log. Prompt { /// Free-form reminder text logged when the timer fires. text: String, }, } // ── Mode ───────────────────────────────────────────────────────────────────── /// Whether the timer fires once or repeatedly. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(tag = "mode", rename_all = "snake_case")] pub enum TimerMode { /// Fire once, then discard. Once, /// Fire, then re-arm at `fire_at + interval_secs`. Recurring { /// Seconds between firings. interval_secs: u64, }, } // ── Entry ───────────────────────────────────────────────────────────────────── /// A single generic scheduled timer entry. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct ScheduledTimer { /// Unique stable identifier (UUID v4 short form, e.g. `"tm-a3f7b9c2"`). pub id: String, /// Optional human-readable label. pub label: Option, /// UTC instant when this timer should next fire. pub fire_at: DateTime, /// What to execute on fire. pub action: TimerAction, /// One-shot or recurring. pub mode: TimerMode, /// UTC instant when this timer was created. pub created_at: DateTime, } impl ScheduledTimer { /// Generate a fresh timer ID. pub fn new_id() -> String { let id = Uuid::new_v4(); let hex = id.as_simple().to_string(); let prefix: String = hex.chars().take(8).collect(); format!("tm-{prefix}") } } // ── Background writer ───────────────────────────────────────────────────────── enum SchedWriteCmd { Upsert(ScheduledTimer), Delete(String), } fn spawn_sched_writer(pool: SqlitePool, mut rx: mpsc::UnboundedReceiver) { tokio::spawn(async move { while let Some(cmd) = rx.recv().await { match cmd { SchedWriteCmd::Upsert(t) => { let action_json = match serde_json::to_string(&t.action) { Ok(j) => j, Err(e) => { crate::slog!("[scheduled-timer] Serialize action failed: {e}"); continue; } }; let mode_json = match serde_json::to_string(&t.mode) { Ok(j) => j, Err(e) => { crate::slog!("[scheduled-timer] Serialize mode failed: {e}"); continue; } }; let result = sqlx::query( "INSERT INTO scheduled_timers \ (id, label, fire_at, action_json, mode_json, created_at) \ VALUES (?1, ?2, ?3, ?4, ?5, ?6) \ ON CONFLICT(id) DO UPDATE SET \ label = excluded.label, \ fire_at = excluded.fire_at, \ action_json = excluded.action_json, \ mode_json = excluded.mode_json", ) .bind(&t.id) .bind(&t.label) .bind(t.fire_at.to_rfc3339()) .bind(&action_json) .bind(&mode_json) .bind(t.created_at.to_rfc3339()) .execute(&pool) .await; if let Err(e) = result { crate::slog!("[scheduled-timer] DB upsert failed for '{}': {e}", t.id); } } SchedWriteCmd::Delete(id) => { let result = sqlx::query("DELETE FROM scheduled_timers WHERE id = ?1") .bind(&id) .execute(&pool) .await; if let Err(e) = result { crate::slog!("[scheduled-timer] DB delete failed for '{id}': {e}"); } } } } }); } // ── Store ───────────────────────────────────────────────────────────────────── enum SchedPersistence { Sqlite(mpsc::UnboundedSender), Json(PathBuf), } /// Persistent store for generic scheduled timers, backed by SQLite in /// production and a JSON file in tests. pub struct ScheduledTimerStore { timers: Mutex>, persistence: SchedPersistence, } impl ScheduledTimerStore { /// Load from the shared SQLite pool. This is the production constructor. pub async fn from_pool(pool: SqlitePool) -> Result { let rows: Vec<(String, Option, String, String, String, String)> = sqlx::query_as( "SELECT id, label, fire_at, action_json, mode_json, created_at \ FROM scheduled_timers", ) .fetch_all(&pool) .await?; let mut timers = Vec::with_capacity(rows.len()); for (id, label, fire_at_str, action_json, mode_json, created_at_str) in rows { let fire_at = match fire_at_str.parse::>() { Ok(t) => t, Err(e) => { crate::slog!("[scheduled-timer] Bad fire_at for {id}: {e}"); continue; } }; let action: TimerAction = match serde_json::from_str(&action_json) { Ok(a) => a, Err(e) => { crate::slog!("[scheduled-timer] Bad action for {id}: {e}"); continue; } }; let mode: TimerMode = match serde_json::from_str(&mode_json) { Ok(m) => m, Err(e) => { crate::slog!("[scheduled-timer] Bad mode for {id}: {e}"); continue; } }; let created_at = created_at_str .parse::>() .unwrap_or_else(|_| Utc::now()); timers.push(ScheduledTimer { id, label, fire_at, action, mode, created_at, }); } let (tx, rx) = mpsc::unbounded_channel(); spawn_sched_writer(pool, rx); Ok(Self { timers: Mutex::new(timers), persistence: SchedPersistence::Sqlite(tx), }) } /// Load (or create empty) store from a JSON file path. Used by unit tests. pub fn load(path: PathBuf) -> Self { let timers = if path.exists() { std::fs::read_to_string(&path) .ok() .and_then(|s| serde_json::from_str::>(&s).ok()) .unwrap_or_default() } else { Vec::new() }; Self { timers: Mutex::new(timers), persistence: SchedPersistence::Json(path), } } fn save_json(path: &Path, timers: &[ScheduledTimer]) -> Result<(), String> { if let Some(parent) = path.parent() { std::fs::create_dir_all(parent).map_err(|e| format!("mkdir failed: {e}"))?; } let content = serde_json::to_string_pretty(timers).map_err(|e| format!("serialize failed: {e}"))?; std::fs::write(path, content).map_err(|e| format!("write failed: {e}")) } /// Add a timer. Errors if a timer with the same ID already exists. pub fn add(&self, timer: ScheduledTimer) -> Result<(), String> { let mut timers = self.timers.lock().unwrap(); if timers.iter().any(|t| t.id == timer.id) { return Err(format!("Timer with id '{}' already exists", timer.id)); } timers.push(timer.clone()); match &self.persistence { SchedPersistence::Sqlite(tx) => { let _ = tx.send(SchedWriteCmd::Upsert(timer)); Ok(()) } SchedPersistence::Json(path) => Self::save_json(path, &timers), } } /// Remove a timer by ID. Returns `true` if one was removed. pub fn remove_by_id(&self, id: &str) -> bool { let mut timers = self.timers.lock().unwrap(); let before = timers.len(); timers.retain(|t| t.id != id); let removed = timers.len() < before; if removed { match &self.persistence { SchedPersistence::Sqlite(tx) => { let _ = tx.send(SchedWriteCmd::Delete(id.to_string())); } SchedPersistence::Json(path) => { let _ = Self::save_json(path, &timers); } } } removed } /// Return all pending timers (cloned). pub fn list(&self) -> Vec { self.timers.lock().unwrap().clone() } /// Remove and return all timers whose `fire_at` ≤ `now`. pub fn take_due(&self, now: DateTime) -> Vec { let mut timers = self.timers.lock().unwrap(); let mut due = Vec::new(); let mut remaining = Vec::new(); for t in timers.drain(..) { if t.fire_at <= now { due.push(t); } else { remaining.push(t); } } *timers = remaining; if !due.is_empty() { match &self.persistence { SchedPersistence::Sqlite(tx) => { for entry in &due { let _ = tx.send(SchedWriteCmd::Delete(entry.id.clone())); } } SchedPersistence::Json(path) => { let _ = Self::save_json(path, &timers); } } } due } } // ── When parsing ────────────────────────────────────────────────────────────── /// Parse a `when` string into `(fire_at, optional_interval_secs)`. /// /// Accepted forms: /// - Relative: `"in 2 hours"`, `"in 15 minutes"`, `"in 30 seconds"`, /// `"2h"`, `"15m"`, `"30s"` (the `"in "` prefix is optional) /// - Absolute: ISO 8601 / RFC 3339 timestamp (`"2026-05-15T10:00:00Z"`) /// /// Returns the interval in seconds for relative durations so callers can /// set up recurring re-arm intervals. pub fn parse_when_str( when: &str, now: DateTime, ) -> Result<(DateTime, Option), String> { let trimmed = when.trim(); // Strip optional "in " prefix, then attempt interval parse. let candidate = trimmed .strip_prefix("in ") .or_else(|| trimmed.strip_prefix("In ")) .unwrap_or(trimmed); if let Some(secs) = parse_interval_str(candidate) { let fire_at = now + chrono::Duration::seconds(secs as i64); return Ok((fire_at, Some(secs))); } // Try ISO 8601 / RFC 3339 absolute timestamp. if let Ok(dt) = trimmed.parse::>() { return Ok((dt, None)); } if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(trimmed) { return Ok((dt.with_timezone(&Utc), None)); } Err(format!( "Cannot parse 'when': '{when}'. \ Use 'in 2 hours', '15 minutes', '2h', or an ISO 8601 timestamp." )) } /// Parse an interval string like `"2h"`, `"15m"`, `"30s"`, `"2 hours"`, /// `"15 minutes"`, `"30 seconds"` into a number of seconds. pub fn parse_interval_str(s: &str) -> Option { let s = s.trim().to_lowercase(); let (num_str, unit_str) = split_num_unit(&s); let n: u64 = num_str.parse().ok()?; if n == 0 { return None; } match unit_str.trim() { "h" | "hr" | "hrs" | "hour" | "hours" => Some(n * 3600), "m" | "min" | "mins" | "minute" | "minutes" => Some(n * 60), "s" | "sec" | "secs" | "second" | "seconds" => Some(n), _ => None, } } /// Split a string like `"2hours"` or `"15 minutes"` into `("2", "hours")`. fn split_num_unit(s: &str) -> (&str, &str) { let idx = s.find(|c: char| !c.is_ascii_digit()).unwrap_or(s.len()); let (num, unit) = s.split_at(idx); (num, unit.trim_start_matches(' ')) } // ── Tests ───────────────────────────────────────────────────────────────────── #[cfg(test)] mod tests { use super::*; use chrono::TimeZone; use tempfile::TempDir; fn fixed_now() -> DateTime { Utc.with_ymd_and_hms(2026, 5, 15, 10, 0, 0).unwrap() } // ── parse_interval_str ──────────────────────────────────────────────────── #[test] fn parse_interval_hours_long() { assert_eq!(parse_interval_str("2 hours"), Some(7200)); } #[test] fn parse_interval_hours_short() { assert_eq!(parse_interval_str("2h"), Some(7200)); } #[test] fn parse_interval_minutes_long() { assert_eq!(parse_interval_str("15 minutes"), Some(900)); } #[test] fn parse_interval_minutes_short() { assert_eq!(parse_interval_str("15m"), Some(900)); } #[test] fn parse_interval_seconds_long() { assert_eq!(parse_interval_str("30 seconds"), Some(30)); } #[test] fn parse_interval_seconds_short() { assert_eq!(parse_interval_str("30s"), Some(30)); } #[test] fn parse_interval_zero_returns_none() { assert_eq!(parse_interval_str("0 hours"), None); } #[test] fn parse_interval_unknown_unit_returns_none() { assert_eq!(parse_interval_str("5 fortnights"), None); } // ── parse_when_str ──────────────────────────────────────────────────────── #[test] fn parse_when_relative_with_in_prefix() { let now = fixed_now(); let (fire_at, interval) = parse_when_str("in 2 hours", now).unwrap(); assert_eq!(interval, Some(7200)); assert_eq!(fire_at, now + chrono::Duration::seconds(7200)); } #[test] fn parse_when_relative_without_in_prefix() { let now = fixed_now(); let (fire_at, interval) = parse_when_str("15 minutes", now).unwrap(); assert_eq!(interval, Some(900)); assert_eq!(fire_at, now + chrono::Duration::seconds(900)); } #[test] fn parse_when_short_form() { let now = fixed_now(); let (fire_at, interval) = parse_when_str("2h", now).unwrap(); assert_eq!(interval, Some(7200)); assert_eq!(fire_at, now + chrono::Duration::seconds(7200)); } #[test] fn parse_when_iso8601() { let now = fixed_now(); let (fire_at, interval) = parse_when_str("2026-05-15T12:00:00Z", now).unwrap(); assert_eq!(interval, None); let expected = Utc.with_ymd_and_hms(2026, 5, 15, 12, 0, 0).unwrap(); assert_eq!(fire_at, expected); } #[test] fn parse_when_invalid_returns_err() { let now = fixed_now(); assert!(parse_when_str("next Tuesday", now).is_err()); } // ── ScheduledTimerStore ─────────────────────────────────────────────────── fn make_timer(id: &str, fire_at: DateTime) -> ScheduledTimer { ScheduledTimer { id: id.to_string(), label: None, fire_at, action: TimerAction::Prompt { text: "test".to_string(), }, mode: TimerMode::Once, created_at: Utc::now(), } } #[test] fn store_empty_on_missing_file() { let dir = TempDir::new().unwrap(); let store = ScheduledTimerStore::load(dir.path().join("timers.json")); assert!(store.list().is_empty()); } #[test] fn store_add_and_list() { let dir = TempDir::new().unwrap(); let store = ScheduledTimerStore::load(dir.path().join("timers.json")); let t = Utc::now() + chrono::Duration::hours(1); store.add(make_timer("tm-aabbccdd", t)).unwrap(); let list = store.list(); assert_eq!(list.len(), 1); assert_eq!(list[0].id, "tm-aabbccdd"); } #[test] fn store_add_duplicate_id_fails() { let dir = TempDir::new().unwrap(); let store = ScheduledTimerStore::load(dir.path().join("timers.json")); let t = Utc::now() + chrono::Duration::hours(1); store.add(make_timer("tm-aabbccdd", t)).unwrap(); assert!(store.add(make_timer("tm-aabbccdd", t)).is_err()); } #[test] fn store_remove_by_id() { let dir = TempDir::new().unwrap(); let store = ScheduledTimerStore::load(dir.path().join("timers.json")); let t = Utc::now() + chrono::Duration::hours(1); store.add(make_timer("tm-aabbccdd", t)).unwrap(); assert!(store.remove_by_id("tm-aabbccdd")); assert!(!store.remove_by_id("tm-aabbccdd")); assert!(store.list().is_empty()); } #[test] fn store_persists_and_reloads() { let dir = TempDir::new().unwrap(); let path = dir.path().join("timers.json"); let t = Utc::now() + chrono::Duration::hours(2); { let store = ScheduledTimerStore::load(path.clone()); store.add(make_timer("tm-aabbccdd", t)).unwrap(); } let store2 = ScheduledTimerStore::load(path); assert_eq!(store2.list().len(), 1); assert_eq!(store2.list()[0].id, "tm-aabbccdd"); } #[test] fn take_due_returns_only_past_entries() { let dir = TempDir::new().unwrap(); let store = ScheduledTimerStore::load(dir.path().join("timers.json")); let past = Utc::now() - chrono::Duration::minutes(1); let future = Utc::now() + chrono::Duration::hours(1); store.add(make_timer("tm-past", past)).unwrap(); store.add(make_timer("tm-future", future)).unwrap(); let due = store.take_due(Utc::now()); assert_eq!(due.len(), 1); assert_eq!(due[0].id, "tm-past"); assert_eq!(store.list().len(), 1); assert_eq!(store.list()[0].id, "tm-future"); } #[test] fn take_due_with_already_past_fires_immediately() { let dir = TempDir::new().unwrap(); let store = ScheduledTimerStore::load(dir.path().join("timers.json")); // Simulate server restart: timer scheduled in past (catch-up semantics) let way_past = Utc::now() - chrono::Duration::hours(3); store.add(make_timer("tm-catchup", way_past)).unwrap(); let due = store.take_due(Utc::now()); assert_eq!(due.len(), 1, "past timer must fire on next tick"); } #[tokio::test] async fn from_pool_persists_and_reloads() { let dir = TempDir::new().unwrap(); let db_path = dir.path().join("test.db"); let opts = sqlx::sqlite::SqliteConnectOptions::new() .filename(&db_path) .create_if_missing(true); let pool = sqlx::SqlitePool::connect_with(opts).await.unwrap(); sqlx::query( "CREATE TABLE IF NOT EXISTS scheduled_timers \ (id TEXT PRIMARY KEY, label TEXT, fire_at TEXT NOT NULL, \ action_json TEXT NOT NULL, mode_json TEXT NOT NULL, created_at TEXT NOT NULL)", ) .execute(&pool) .await .unwrap(); let t = Utc::now() + chrono::Duration::hours(1); { let store = ScheduledTimerStore::from_pool(pool.clone()).await.unwrap(); store.add(make_timer("tm-aabbccdd", t)).unwrap(); // Give the background writer a moment to flush. tokio::time::sleep(std::time::Duration::from_millis(50)).await; } let store2 = ScheduledTimerStore::from_pool(pool).await.unwrap(); let list = store2.list(); assert_eq!(list.len(), 1); assert_eq!(list[0].id, "tm-aabbccdd"); } #[test] fn new_id_has_tm_prefix() { let id = ScheduledTimer::new_id(); assert!(id.starts_with("tm-"), "expected 'tm-' prefix: {id}"); assert_eq!(id.len(), 11, "expected 'tm-' + 8 hex chars: {id}"); } // ── TimerAction serde ───────────────────────────────────────────────────── #[test] fn timer_action_mcp_round_trips() { let action = TimerAction::Mcp { method: "start_agent".to_string(), args: serde_json::json!({ "story_id": "42_foo" }), }; let s = serde_json::to_string(&action).unwrap(); let back: TimerAction = serde_json::from_str(&s).unwrap(); assert_eq!(action, back); } #[test] fn timer_action_prompt_round_trips() { let action = TimerAction::Prompt { text: "daily standup reminder".to_string(), }; let s = serde_json::to_string(&action).unwrap(); let back: TimerAction = serde_json::from_str(&s).unwrap(); assert_eq!(action, back); } // ── TimerMode serde ─────────────────────────────────────────────────────── #[test] fn timer_mode_once_round_trips() { let mode = TimerMode::Once; let s = serde_json::to_string(&mode).unwrap(); let back: TimerMode = serde_json::from_str(&s).unwrap(); assert_eq!(mode, back); } #[test] fn timer_mode_recurring_round_trips() { let mode = TimerMode::Recurring { interval_secs: 3600, }; let s = serde_json::to_string(&mode).unwrap(); let back: TimerMode = serde_json::from_str(&s).unwrap(); assert_eq!(mode, back); } }