fraiseql-functions 2.16.0

//! Cron-scheduled triggers.
//!
//! Handles `cron:<expression>` triggers that fire on a schedule using POSIX cron expressions.
//!
//! ## Features
//!
//! - **Stateful**: Last execution time is persisted to `_fraiseql_cron_state` table
//! - **Missed Fire Detection**: Detects missed executions on restart
//! - **Configurable Replay**: Optionally replay missed fires on startup
//! - **Timezone Support**: Execute in specific timezone (defaults to UTC)
//!
//! ## Example Expressions
//!
//! - `0 * * * *` - Every hour
//! - `0 2 * * *` - Daily at 2 AM
//! - `*/5 * * * *` - Every 5 minutes
//! - `0 0 1 * *` - First day of every month

use chrono::{Datelike, Timelike};
use serde::{Deserialize, Serialize};

use crate::types::EventPayload;

/// A parsed cron field that can be a number, wildcard, or step value.
#[derive(Debug, Clone)]
enum CronField {
    /// Wildcard (*) matches any value
    Any,
    /// Specific value
    Value(u32),
    /// Step value (*/5 matches every 5th)
    Step { base: Option<u32>, step: u32 },
    /// List of values (1,3,5)
    List(Vec<u32>),
    /// Range (1-5)
    Range { start: u32, end: u32 },
}

impl CronField {
    /// Check if this field matches the given value.
    fn matches(&self, value: u32) -> bool {
        match self {
            CronField::Any => true,
            CronField::Value(v) => *v == value,
            CronField::Step { base, step } => {
                let start = base.unwrap_or(0);
                if value < start {
                    return false;
                }
                (value - start).is_multiple_of(*step)
            },
            CronField::List(values) => values.contains(&value),
            CronField::Range { start, end } => value >= *start && value <= *end,
        }
    }

    /// Parse a cron field from a string.
    fn parse(field: &str) -> Result<Self, String> {
        if field == "*" {
            return Ok(CronField::Any);
        }

        if let Some(step_idx) = field.find('/') {
            let (base_part, step_part) = field.split_at(step_idx);
            let step_str = &step_part[1..]; // Skip the '/'

            let step = step_str
                .parse::<u32>()
                .map_err(|_| format!("Invalid step value: {}", step_str))?;

            let base = if base_part == "*" {
                None
            } else {
                Some(
                    base_part
                        .parse::<u32>()
                        .map_err(|_| format!("Invalid base value: {}", base_part))?,
                )
            };

            return Ok(CronField::Step { base, step });
        }

        if let Some(dash_idx) = field.find('-') {
            let (start_str, end_str) = field.split_at(dash_idx);
            let end_str = &end_str[1..]; // Skip the '-'

            let start = start_str
                .parse::<u32>()
                .map_err(|_| format!("Invalid range start: {}", start_str))?;
            let end =
                end_str.parse::<u32>().map_err(|_| format!("Invalid range end: {}", end_str))?;

            return Ok(CronField::Range { start, end });
        }

        if field.contains(',') {
            let values = field
                .split(',')
                .map(|v| v.parse::<u32>().map_err(|_| format!("Invalid list value: {}", v)))
                .collect::<Result<Vec<u32>, String>>()?;

            return Ok(CronField::List(values));
        }

        let value = field
            .parse::<u32>()
            .map_err(|_| format!("Invalid cron field value: {}", field))?;

        Ok(CronField::Value(value))
    }

    /// Refuse a field that can never match inside `min..=max`, naming the field.
    ///
    /// Each refused shape used to parse and then never match: minute `61`, a step of
    /// `0`, a range whose start is past its end. A schedule built from one never fires,
    /// so a function or source declared on it silently never runs (#1399).
    fn within(self, name: &str, min: u32, max: u32) -> Result<Self, String> {
        let out_of_range = |value: u32| {
            if (min..=max).contains(&value) {
                Ok(())
            } else {
                Err(format!("{name} value {value} is outside {min}-{max}"))
            }
        };
        match &self {
            CronField::Any => {},
            CronField::Value(value) => out_of_range(*value)?,
            CronField::List(values) => values.iter().try_for_each(|value| out_of_range(*value))?,
            CronField::Step { base, step } => {
                if *step == 0 {
                    return Err(format!("{name} step must be at least 1"));
                }
                if let Some(base) = base {
                    out_of_range(*base)?;
                }
            },
            CronField::Range { start, end } => {
                out_of_range(*start)?;
                out_of_range(*end)?;
                if start > end {
                    return Err(format!("{name} range {start}-{end} runs backwards"));
                }
            },
        }
        Ok(self)
    }
}

/// A cron expression with validation.
#[derive(Debug, Clone)]
pub struct CronSchedule {
    /// The raw cron expression (e.g., "0 2 * * *").
    pub expression: String,
    /// Parsed minute field (0-59)
    minute:         CronField,
    /// Parsed hour field (0-23)
    hour:           CronField,
    /// Parsed day-of-month field (1-31)
    day:            CronField,
    /// Parsed month field (1-12)
    month:          CronField,
    /// Parsed day-of-week field (0-7, 0 and 7 = Sunday)
    weekday:        CronField,
}

impl CronSchedule {
    /// Parse and validate a cron expression.
    ///
    /// # Errors
    ///
    /// Returns an error if the cron expression is invalid.
    pub fn parse(expression: &str) -> Result<Self, String> {
        let parts: Vec<&str> = expression.split_whitespace().collect();

        if parts.len() != 5 {
            return Err(format!("Cron expression must have 5 fields, got {}", parts.len()));
        }

        let minute = CronField::parse(parts[0])?.within("minute", 0, 59)?;
        let hour = CronField::parse(parts[1])?.within("hour", 0, 23)?;
        let day = CronField::parse(parts[2])?.within("day-of-month", 1, 31)?;
        let month = CronField::parse(parts[3])?.within("month", 1, 12)?;
        // 7 is POSIX's alternate Sunday (see `matches`).
        let weekday = CronField::parse(parts[4])?.within("day-of-week", 0, 7)?;

        Ok(CronSchedule {
            expression: expression.to_string(),
            minute,
            hour,
            day,
            month,
            weekday,
        })
    }

    /// Check if a given datetime matches this cron schedule.
    #[must_use]
    pub fn matches(&self, datetime: &chrono::DateTime<chrono::Utc>) -> bool {
        let minute = datetime.minute();
        let hour = datetime.hour();
        let day = datetime.day();
        let month = datetime.month();
        // POSIX cron weekdays are 0=Sunday..6=Saturday (#841 shipped
        // `number_from_sunday()`, which is 1..=7, firing everything a day
        // early and making `0` unmatchable). `7` is the POSIX alternate
        // Sunday encoding, so a Sunday also matches a field containing 7.
        let weekday = datetime.weekday().num_days_from_sunday();
        let weekday_matches =
            self.weekday.matches(weekday) || (weekday == 0 && self.weekday.matches(7));

        self.minute.matches(minute)
            && self.hour.matches(hour)
            && self.day.matches(day)
            && self.month.matches(month)
            && weekday_matches
    }
}

/// Why a tick did not fire — or that it should.
///
/// [`CronExecutionState::decide`] returns this so scheduling loops can log the
/// reason for every skipped tick instead of silently `continue`-ing (#796: the
/// inverted window guard suppressed every firing for months with no signal).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CronDecision {
    /// The schedule matches and this window has not fired yet — execute.
    Fire,
    /// The schedule does not match this tick's minute (the normal idle tick).
    NotDue,
    /// The schedule matches but this window already fired (`last_executed >=
    /// window_start`) — the dedup guard, expected at most once per window
    /// (e.g. after a restart restores persisted state mid-window).
    AlreadyFired {
        /// Start of the window containing the tick (its minute, truncated).
        window_start:  chrono::DateTime<chrono::Utc>,
        /// When the window fired.
        last_executed: chrono::DateTime<chrono::Utc>,
    },
}

/// Execution state for a cron trigger (tracks last execution to prevent duplicates).
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct CronExecutionState {
    /// Last time this trigger was executed.
    pub last_executed: Option<chrono::DateTime<chrono::Utc>>,
}

impl CronExecutionState {
    /// Create a new execution state with no prior executions.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Restore state from a persisted last-firing instant (the
    /// `_fraiseql_cron_state` cross-restart guard): a restart inside an
    /// already-fired window must not re-fire it.
    #[must_use]
    pub const fn resuming_from(last_executed: chrono::DateTime<chrono::Utc>) -> Self {
        Self {
            last_executed: Some(last_executed),
        }
    }

    /// Check if the trigger should execute at the given time.
    ///
    /// Returns true iff [`decide`](Self::decide) says [`CronDecision::Fire`]:
    /// the schedule matches `exec_time` and no execution has occurred in the
    /// window containing it.
    #[must_use]
    pub fn should_execute(
        &self,
        schedule: &CronSchedule,
        exec_time: &chrono::DateTime<chrono::Utc>,
    ) -> bool {
        self.decide(schedule, exec_time) == CronDecision::Fire
    }

    /// Decide whether the trigger fires at `exec_time`, with the reason when
    /// it does not (so schedulers can log every skipped tick — see
    /// [`CronDecision`]).
    ///
    /// The invariant: a window fires exactly once, and `last_executed` lies
    /// inside a fired window — so the guard is `last_executed < window_start`,
    /// never the reverse (#796 shipped the negation and every schedule fired
    /// once, then never again).
    #[must_use]
    pub fn decide(
        &self,
        schedule: &CronSchedule,
        exec_time: &chrono::DateTime<chrono::Utc>,
    ) -> CronDecision {
        // Schedule must match the execution time
        if !schedule.matches(exec_time) {
            return CronDecision::NotDue;
        }

        // If no prior execution, always execute
        let Some(last_executed) = self.last_executed else {
            return CronDecision::Fire;
        };

        // The start of the window CONTAINING exec_time (not the previous one).
        let window_start = Self::find_schedule_window(schedule, exec_time);

        if last_executed < window_start {
            CronDecision::Fire
        } else {
            CronDecision::AlreadyFired {
                window_start,
                last_executed,
            }
        }
    }

    /// Find the start of the schedule window **containing** the given time.
    ///
    /// Cron's resolution is one minute, so for a matching instant the window
    /// start is that instant truncated to its minute: for `"0 2 * * *"` the
    /// window for 2024-03-15 02:00:07.312 starts at 2024-03-15 02:00:00.
    ///
    /// Callers must pass a `time` the schedule matches (checked by
    /// `debug_assert`); truncation never moves the instant across a minute
    /// boundary, so the returned window start matches the schedule too.
    fn find_schedule_window(
        schedule: &CronSchedule,
        time: &chrono::DateTime<chrono::Utc>,
    ) -> chrono::DateTime<chrono::Utc> {
        debug_assert!(
            schedule.matches(time),
            "find_schedule_window called for a non-matching instant"
        );
        // Truncating seconds + nanos to zero cannot fail.
        time.with_second(0).and_then(|t| t.with_nanosecond(0)).unwrap_or(*time)
    }

    /// Record an execution at the given time.
    #[allow(clippy::missing_const_for_fn)] // Reason: takes &mut self, cannot be const
    pub fn record_execution(&mut self, time: chrono::DateTime<chrono::Utc>) {
        self.last_executed = Some(time);
    }

    /// Find all missed execution times between two timestamps.
    ///
    /// Finds all times when the schedule would have executed after `since` and before `until`.
    #[must_use]
    pub fn find_missed_executions(
        &self,
        schedule: &CronSchedule,
        since: &chrono::DateTime<chrono::Utc>,
        until: &chrono::DateTime<chrono::Utc>,
    ) -> Vec<chrono::DateTime<chrono::Utc>> {
        let mut missed = Vec::new();
        // Start scanning from the minute AFTER since
        let mut current = *since + chrono::Duration::minutes(1);

        // Scan minute by minute (more fine-grained to catch all matches)
        while current < *until {
            if schedule.matches(&current) {
                missed.push(current);
            }

            // Always advance to next minute
            current += chrono::Duration::minutes(1);
        }

        missed
    }
}

/// A trigger that fires on a cron schedule.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CronTrigger {
    /// Name of the function to invoke.
    pub function_name: String,
    /// Cron expression (e.g., `"0 2 * * *"`).
    pub schedule:      String,
    /// Timezone for schedule evaluation (e.g., `"UTC"`, `"America/New_York"`).
    pub timezone:      String,
}

impl CronTrigger {
    /// Build an `EventPayload` from a cron execution.
    #[must_use]
    pub fn build_payload(&self, exec_time: &chrono::DateTime<chrono::Utc>) -> EventPayload {
        let trigger_type = format!("cron:{}", self.function_name);

        let data = serde_json::json!({
            "schedule": self.schedule,
            "timezone": self.timezone,
            "executed_at": exec_time.to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
        });

        EventPayload {
            trigger_type,
            entity: "cron".to_string(),
            event_kind: "scheduled".to_string(),
            data,
            timestamp: chrono::Utc::now(),
        }
    }
}

// ── CronScheduler ─────────────────────────────────────────────────────────────

/// Service that drives scheduled function execution.
///
/// Each registered [`CronTrigger`] is checked once per minute. When a trigger's
/// schedule matches the current time and it has not already fired in this
/// scheduling window, the function is invoked via the provided
/// [`FunctionObserver`](crate::FunctionObserver).
///
/// # Lifecycle
///
/// 1. Build a scheduler with [`CronScheduler::new`].
/// 2. Call [`CronScheduler::start`] to spawn the background task; this returns a
///    [`CronSchedulerHandle`] that can be used to stop the task.
/// 3. On server shutdown, drop (or explicitly call `stop()` on) the handle.
pub struct CronScheduler {
    /// Triggers paired with their per-trigger execution state.
    triggers: Vec<(CronTrigger, CronExecutionState)>,
}

impl CronScheduler {
    /// Create a new scheduler for the given cron triggers.
    ///
    /// Each trigger is paired with a fresh [`CronExecutionState`] (no prior
    /// executions recorded). Call [`Self::start`] to begin scheduling.
    #[must_use]
    pub fn new(triggers: Vec<CronTrigger>) -> Self {
        let triggers = triggers.into_iter().map(|t| (t, CronExecutionState::new())).collect();
        Self { triggers }
    }

    /// Returns the number of cron triggers registered.
    #[must_use]
    pub const fn trigger_count(&self) -> usize {
        self.triggers.len()
    }

    /// Start the scheduler as a background tokio task.
    ///
    /// Spawns a task that ticks once per minute. On each tick, triggers whose
    /// schedule matches the current time and have not fired in the current
    /// window are dispatched via `observer.invoke()` (fire-and-forget).
    ///
    /// Returns a [`CronSchedulerHandle`] — drop it (or call `stop()`) to
    /// cancel the background task.
    ///
    /// # Panics
    ///
    /// Panics if called outside a tokio runtime context.
    #[must_use]
    pub fn start(
        self,
        observer: std::sync::Arc<crate::observer::FunctionObserver>,
        module_registry: std::collections::HashMap<String, crate::types::FunctionModule>,
    ) -> CronSchedulerHandle {
        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
        tokio::spawn(cron_scheduler_task(self, observer, module_registry, shutdown_rx));
        CronSchedulerHandle {
            shutdown_tx: Some(shutdown_tx),
        }
    }
}

/// Inner async loop for the cron scheduler.
async fn cron_scheduler_task(
    mut scheduler: CronScheduler,
    observer: std::sync::Arc<crate::observer::FunctionObserver>,
    module_registry: std::collections::HashMap<String, crate::types::FunctionModule>,
    mut shutdown_rx: tokio::sync::oneshot::Receiver<()>,
) {
    let mut interval = tokio::time::interval(tokio::time::Duration::from_mins(1));
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    // Skip the initial immediate tick so the scheduler doesn't fire on startup.
    interval.tick().await;

    loop {
        tokio::select! {
            _ = interval.tick() => {
                let now = chrono::Utc::now();
                for (trigger, state) in &mut scheduler.triggers {
                    let schedule = match CronSchedule::parse(&trigger.schedule) {
                        Ok(s) => s,
                        Err(e) => {
                            tracing::warn!(
                                function = %trigger.function_name,
                                expression = %trigger.schedule,
                                error = %e,
                                "Invalid cron expression — skipping trigger"
                            );
                            continue;
                        }
                    };
                    match state.decide(&schedule, &now) {
                        CronDecision::Fire => {},
                        // The normal idle tick — the schedule is simply not due.
                        CronDecision::NotDue => continue,
                        // Never silent (#796): the window guard suppressing a
                        // matching tick is expected at most once per window.
                        CronDecision::AlreadyFired {
                            window_start,
                            last_executed,
                        } => {
                            tracing::warn!(
                                function = %trigger.function_name,
                                expression = %trigger.schedule,
                                %window_start,
                                %last_executed,
                                "cron tick skipped — this window already fired"
                            );
                            continue;
                        },
                    }
                    state.record_execution(now);

                    // Look up the function module; log and skip if not found.
                    let Some(module) = module_registry.get(&trigger.function_name) else {
                        tracing::warn!(
                            function = %trigger.function_name,
                            "Cron trigger fired but function module not found — skipping"
                        );
                        continue;
                    };

                    let payload = trigger.build_payload(&now);
                    let observer_clone = std::sync::Arc::clone(&observer);
                    let module_clone = module.clone();
                    let fn_name = trigger.function_name.clone();

                    // Invoke fire-and-forget; failures are logged but don't stop the scheduler.
                    tokio::spawn(async move {
                        let host = crate::host::NoopHostContext::new(payload.clone());
                        match observer_clone
                            .invoke(
                                &module_clone,
                                payload,
                                &host,
                                crate::types::ResourceLimits::default(),
                            )
                            .await
                        {
                            Ok(_) => {
                                tracing::debug!(function = %fn_name, "Cron function completed");
                            }
                            Err(e) => {
                                // F047: walk the source chain and concatenate the
                                // causes so the log carries the whole `std::error::Error`
                                // hierarchy, not just the top-level Display.
                                let chain = error_source_chain(&e);
                                tracing::error!(
                                    function = %fn_name,
                                    error = %e,
                                    error.debug = ?e,
                                    error.chain = %chain,
                                    "Cron function invocation failed"
                                );
                            }
                        }
                    });
                }
            }
            _ = &mut shutdown_rx => {
                tracing::debug!("Cron scheduler received shutdown signal — stopping");
                break;
            }
        }
    }
}

/// Walk an `std::error::Error`'s source chain and concatenate the causes
/// into a single string. Used by the cron-task error log so that wrapped
/// transport / WASM / engine errors keep their context in observability
/// (F047).
fn error_source_chain(err: &(dyn std::error::Error + 'static)) -> String {
    use std::fmt::Write as _;
    let mut chain = String::new();
    let mut current: Option<&dyn std::error::Error> = err.source();
    while let Some(source) = current {
        if !chain.is_empty() {
            chain.push_str(" → ");
        }
        // `write!` into a `String` is infallible.
        let _ = write!(chain, "{source}");
        current = source.source();
    }
    if chain.is_empty() {
        "<no source>".to_string()
    } else {
        chain
    }
}

/// Handle for a running [`CronScheduler`] background task.
///
/// Drop this handle (or call [`stop`][Self::stop]) to cancel the scheduler.
pub struct CronSchedulerHandle {
    shutdown_tx: Option<tokio::sync::oneshot::Sender<()>>,
}

impl CronSchedulerHandle {
    /// Stop the cron scheduler.
    ///
    /// Sends a shutdown signal to the background task. The task stops at its
    /// next scheduling boundary.
    pub fn stop(mut self) {
        if let Some(tx) = self.shutdown_tx.take() {
            let _ = tx.send(());
        }
    }
}

impl Drop for CronSchedulerHandle {
    fn drop(&mut self) {
        if let Some(tx) = self.shutdown_tx.take() {
            let _ = tx.send(());
        }
    }
}