fraiseql-server 2.16.0

//! Server-side `cron:` function scheduling (#595).
//!
//! Wires the compiled schema's `cron:` functions into a running server: one
//! [`CronPoller`](crate::cron::CronPoller) per cron function ticks on its schedule and —
//! single-firing across replicas via the sources' advisory lease — invokes the function on the
//! phase-02 I/O-capable host (`fraiseql_query` under the function's `run_as` ceiling).
//!
//! A cron function is "a scheduled [`Source`](crate::sources) without a cursor": the
//! poller mirrors [`SourcePoller`](crate::sources) but drops the durable-cursor
//! binding and instead records each firing to `_fraiseql_cron_state` (observability +
//! a cross-restart "already fired this window" guard). It reuses `CronSchedule` /
//! `CronExecutionState` (schedule logic),
//! [`LeaseGuardedRunner`](fraiseql_observers::LeaseGuardedRunner) (the lease), and the
//! shared [`RunAsQueryExecutor`](crate::query_bridge) (the bridge).
//!
//! **Missed-tick policy: skip.** A server down over a scheduled instant does not
//! replay on next boot; the next matching window fires normally (cron has no
//! backlog/cursor to resume — unlike a source).

use std::{
    sync::Arc,
    time::{Duration, Instant},
};

use chrono::{DateTime, Utc};
use fraiseql_functions::{
    EventPayload, FunctionModule, FunctionObserver, FunctionResult, ResourceLimits,
    host::{
        dyn_context::DynHostContext,
        live::{HostContextConfig, LiveHostContext, QueryExecutor},
    },
    triggers::{CronDecision, CronExecutionState, CronSchedule, CronTrigger},
};
use fraiseql_observers::{
    DispatchSource, LeaseGuardedRunner, RunOutcome, derive_idempotency_token,
};
use tracing::{debug, info, warn};

use crate::{query_bridge::RunAsQueryExecutor, subsystems::BeforeMutationHooks};

/// A minimal Postgres store for `_fraiseql_cron_state` (#595): the durable record of
/// when each cron function last fired and how many times.
#[derive(Clone)]
pub struct PgCronState {
    pool: sqlx::PgPool,
}

impl PgCronState {
    /// Wrap a pool.
    #[must_use]
    pub const fn new(pool: sqlx::PgPool) -> Self {
        Self { pool }
    }

    /// Create the `_fraiseql_cron_state` table (idempotent DDL).
    ///
    /// # Errors
    ///
    /// Returns the sqlx error if the DDL cannot be applied.
    pub async fn init(&self) -> Result<(), sqlx::Error> {
        crate::migration_lock::run_migration(
            &self.pool,
            fraiseql_functions::migrations::cron_migration_sql(),
        )
        .await
    }

    /// Read back the last recorded firing for one `(function, expression)` pair.
    ///
    /// `None` when the function has never fired (or its expression changed —
    /// stale state for a different schedule must not suppress the new one).
    ///
    /// # Errors
    ///
    /// Returns the sqlx error if the read fails.
    pub async fn load_last_fired(
        &self,
        function_name: &str,
        cron_expr: &str,
    ) -> Result<Option<DateTime<Utc>>, sqlx::Error> {
        sqlx::query_scalar(
            "SELECT last_fired_at FROM _fraiseql_cron_state \
             WHERE function_name = $1 AND cron_expr = $2",
        )
        .bind(function_name)
        .bind(cron_expr)
        .fetch_optional(&self.pool)
        .await
    }

    /// Rebuild the in-memory fire-window state from the durable record — the
    /// documented cross-restart "already fired this window" guard (#796: the
    /// table was written on every firing and never read back, so a restart
    /// inside a fired window re-fired it).
    ///
    /// # Errors
    ///
    /// Returns the sqlx error if the read fails. Callers must treat that as
    /// fatal at boot: a guard that cannot load its state is not a guard.
    pub async fn resume_state(
        &self,
        function_name: &str,
        cron_expr: &str,
    ) -> Result<CronExecutionState, sqlx::Error> {
        Ok(self
            .load_last_fired(function_name, cron_expr)
            .await?
            .map_or_else(CronExecutionState::new, CronExecutionState::resuming_from))
    }

    /// Record one firing: upsert `last_fired_at` and bump `fire_count`.
    ///
    /// # Errors
    ///
    /// Returns the sqlx error if the upsert fails.
    pub async fn record_fire(
        &self,
        function_name: &str,
        cron_expr: &str,
        fired_at: DateTime<Utc>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query(
            "INSERT INTO _fraiseql_cron_state \
                 (function_name, cron_expr, last_fired_at, fire_count) \
             VALUES ($1, $2, $3, 1) \
             ON CONFLICT (function_name, cron_expr) DO UPDATE SET \
                 last_fired_at = EXCLUDED.last_fired_at, \
                 fire_count = _fraiseql_cron_state.fire_count + 1, \
                 updated_at = now()",
        )
        .bind(function_name)
        .bind(cron_expr)
        .bind(fired_at)
        .execute(&self.pool)
        .await
        .map(|_| ())
    }
}

/// The per-cron-function scheduler loop (#595) — the cursor-less analogue of
/// [`SourcePoller`](crate::sources).
pub struct CronPoller {
    /// The function name — the advisory-lease key and cron-state row.
    function_name:   String,
    /// The parsed cron schedule.
    schedule:        CronSchedule,
    /// The raw cron expression (recorded in `_fraiseql_cron_state`).
    cron_expr:       String,
    /// The function module to invoke.
    module:          FunctionModule,
    /// The runtime-agnostic function observer.
    observer:        Arc<FunctionObserver>,
    /// The `fraiseql_query` bridge under the function's `run_as` identity (#594).
    executor:        Arc<dyn QueryExecutor>,
    /// The function's `run_as` identity (#803): a cron firing has no request
    /// caller, so the host's `auth_context` reflects the function's own
    /// granted authority — never an anonymous placeholder.
    identity:        fraiseql_core::security::SecurityContext,
    /// The single-firing runner (advisory lease keyed on `cron:<function>`).
    runner:          LeaseGuardedRunner,
    /// Durable fire record.
    cron_state:      PgCronState,
    /// Host config (SSRF allowlist, timeouts).
    host_config:     HostContextConfig,
    /// Guest resource limits.
    limits:          ResourceLimits,
    /// HMAC subkey signing each firing's idempotency token.
    idempotency_key: Option<Arc<[u8]>>,
    /// In-memory fire-window state (deduplicates within a process; the lease
    /// deduplicates across replicas; `_fraiseql_cron_state` records the outcome).
    state:           CronExecutionState,
}

impl CronPoller {
    /// Assemble a poller. `runner`, `executor`, and `cron_state` are all keyed on /
    /// scoped to `function_name`.
    // Reason: a constructor wiring a cron function's fixed runtime collaborators; a
    // params struct would relocate the same fields without reducing coupling.
    #[allow(clippy::too_many_arguments)]
    #[must_use]
    pub fn new(
        trigger: &CronTrigger,
        schedule: CronSchedule,
        module: FunctionModule,
        observer: Arc<FunctionObserver>,
        executor: Arc<dyn QueryExecutor>,
        identity: fraiseql_core::security::SecurityContext,
        runner: LeaseGuardedRunner,
        cron_state: PgCronState,
        host_config: HostContextConfig,
        limits: ResourceLimits,
        idempotency_key: Option<Arc<[u8]>>,
        initial_state: CronExecutionState,
    ) -> Self {
        Self {
            function_name: trigger.function_name.clone(),
            schedule,
            cron_expr: trigger.schedule.clone(),
            module,
            observer,
            executor,
            identity,
            runner,
            cron_state,
            host_config,
            limits,
            idempotency_key,
            state: initial_state,
        }
    }

    /// The per-firing idempotency token — a stable hash of the function identity and
    /// this firing's payload, signed with the server HMAC subkey when configured.
    fn idempotency_token(&self, payload: &EventPayload) -> String {
        derive_idempotency_token(
            self.idempotency_key.as_deref(),
            // No dedicated Cron dispatch source; `Source` is the closest background
            // salt and the function name + trigger disambiguate the token.
            DispatchSource::Source,
            &self.module.name,
            &payload.trigger_type,
            &payload.data,
        )
    }

    /// Build the host for one firing: the phase-02 `run_as` executor + the firing's
    /// idempotency token. No cursor (cron functions have no watermark).
    fn build_host(
        &self,
        payload: EventPayload,
        idempotency_token: &str,
    ) -> Arc<dyn DynHostContext> {
        Arc::new(
            LiveHostContext::new(payload, self.host_config.clone())
                .with_executor(Arc::clone(&self.executor))
                .with_idempotency_token(idempotency_token.to_string())
                // #803: the host's auth_context reflects the function's own
                // run_as identity (a cron firing has no request caller).
                .with_security_context(self.identity.clone()),
        )
    }

    /// Fire the cron function once, under the lease. Returns the outcome: skipped
    /// (another replica leads) or ran (with the guest's result). On a successful run
    /// the firing is recorded to `_fraiseql_cron_state`.
    async fn fire_once(
        &self,
        now: DateTime<Utc>,
    ) -> RunOutcome<fraiseql_error::Result<FunctionResult>> {
        let trigger = CronTrigger {
            function_name: self.function_name.clone(),
            schedule:      self.cron_expr.clone(),
            timezone:      "UTC".to_string(),
        };
        let payload = trigger.build_payload(&now);
        let token = self.idempotency_token(&payload);
        let started = Instant::now();
        let attempt = self
            .runner
            .run(|| async {
                let host = self.build_host(payload.clone(), &token);
                self.observer
                    .invoke_with_context(&self.module, payload.clone(), host, self.limits.clone())
                    .await
            })
            .await;
        let elapsed = started.elapsed().as_secs_f64();
        match attempt {
            Ok(RunOutcome::Ran(result)) => {
                let metric_result = match &result {
                    Ok(_) => crate::function_metrics::RESULT_OK,
                    // A failed cron firing is re-runnable (the next window retries,
                    // like a source) — never dead-lettered, so it is `error`.
                    Err(_) => crate::function_metrics::RESULT_ERROR,
                };
                crate::function_metrics::record_dispatch(
                    &self.function_name,
                    crate::function_metrics::KIND_CRON,
                    metric_result,
                    elapsed,
                );
                match &result {
                    Ok(_) => {
                        info!(
                            function = %self.function_name,
                            idempotency_token = %token,
                            duration_ms = elapsed * 1000.0,
                            "cron function fired"
                        );
                        // Record the firing durably (best-effort — a record failure
                        // must not crash the scheduler; the function already ran).
                        if let Err(error) = self
                            .cron_state
                            .record_fire(&self.function_name, &self.cron_expr, now)
                            .await
                        {
                            warn!(
                                function = %self.function_name,
                                %error,
                                "cron state record failed — the function ran but the fire was not persisted"
                            );
                        }
                    },
                    Err(error) => warn!(
                        function = %self.function_name,
                        idempotency_token = %token,
                        %error,
                        "cron function invocation failed"
                    ),
                }
                RunOutcome::Ran(result)
            },
            Ok(RunOutcome::SkippedNotLeader) => {
                debug!(
                    function = %self.function_name,
                    "cron function skipped — another replica leads"
                );
                RunOutcome::SkippedNotLeader
            },
            Err(error) => {
                warn!(
                    function = %self.function_name,
                    %error,
                    "cron lease acquire failed — skipping tick"
                );
                RunOutcome::SkippedNotLeader
            },
        }
    }

    /// Run forever: tick once a minute, fire when the schedule window opens. Shutdown
    /// is by task abort — the lifecycle drives the poller on its `JoinSet`.
    pub async fn run_forever(mut self) {
        let mut ticker = tokio::time::interval(Duration::from_mins(1));
        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        // Skip the immediate initial tick so a cron function does not fire on startup.
        ticker.tick().await;
        info!(
            function = %self.function_name,
            schedule = %self.cron_expr,
            "cron scheduler started"
        );
        loop {
            ticker.tick().await;
            let now = Utc::now();
            match self.state.decide(&self.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 (restart mid-window).
                CronDecision::AlreadyFired {
                    window_start,
                    last_executed,
                } => {
                    warn!(
                        function = %self.function_name,
                        schedule = %self.cron_expr,
                        %window_start,
                        %last_executed,
                        "cron tick skipped — this window already fired"
                    );
                    continue;
                },
            }
            self.state.record_execution(now);
            let _ = self.fire_once(now).await;
        }
    }
}

/// Build one [`CronPoller`] per `cron:` function in the compiled schema (#595).
///
/// Each poller runs under the function's `run_as` identity (via
/// [`RunAsQueryExecutor`] over the hot-reloadable `executor`) and single-fires across
/// replicas on a PostgreSQL advisory lease keyed on the function name. A cron trigger
/// whose module never loaded, or whose expression does not parse, is skipped with a
/// warning. The `_fraiseql_cron_state` table must already be initialized
/// ([`PgCronState::init`]); each poller's fire-window state is **read back** from it
/// ([`PgCronState::resume_state`]) so a restart inside an already-fired window does
/// not re-fire it (#796).
///
/// # Errors
///
/// Returns the sqlx error if reading `_fraiseql_cron_state` fails — fatal at boot: a
/// cross-restart guard that cannot load its state is not a guard.
pub async fn build_cron_pollers(
    db_pool: &sqlx::PgPool,
    executor: &Arc<arc_swap::ArcSwap<fraiseql_core::runtime::Executor>>,
    hooks: &BeforeMutationHooks,
    host_config: &HostContextConfig,
    limits: &ResourceLimits,
) -> Result<Vec<CronPoller>, sqlx::Error> {
    let cron_state = PgCronState::new(db_pool.clone());
    let mut pollers = Vec::new();
    for trigger in &hooks.trigger_registry.cron_triggers {
        let schedule = match CronSchedule::parse(&trigger.schedule) {
            Ok(schedule) => schedule,
            Err(error) => {
                warn!(
                    function = %trigger.function_name,
                    expression = %trigger.schedule,
                    %error,
                    "invalid cron schedule — skipping function"
                );
                continue;
            },
        };
        let Some(module) = hooks.module_registry.get(&trigger.function_name).cloned() else {
            continue;
        };

        // The function's mutations run under its `run_as` ceiling (fail-closed
        // when absent); the request-id correlates the function in the audit
        // envelope, matching the sources pattern.
        let run_as = hooks.run_as.get(&trigger.function_name).cloned().unwrap_or_default();
        let identity = run_as.identity(&trigger.function_name, &trigger.function_name);
        let query_executor: Arc<dyn QueryExecutor> =
            Arc::new(RunAsQueryExecutor::new(Arc::clone(executor), identity.clone()));

        // The cross-restart guard: resume this function's window state from the
        // durable record before the first tick.
        let initial_state =
            cron_state.resume_state(&trigger.function_name, &trigger.schedule).await?;

        pollers.push(CronPoller::new(
            trigger,
            schedule,
            module,
            Arc::clone(&hooks.observer),
            query_executor,
            identity,
            LeaseGuardedRunner::postgres(
                db_pool.clone(),
                format!("cron:{}", trigger.function_name),
            ),
            cron_state.clone(),
            host_config.clone(),
            limits.clone(),
            hooks.idempotency_key.clone(),
            initial_state,
        ));
    }
    Ok(pollers)
}

#[cfg(test)]
mod tests;