arcature 2026.2.1

//! The scheduler — a managed recurring-job enqueuer with lifecycle (A12).
//!
//! The `Scheduler` runs in a managed Tokio task and enqueues jobs on a
//! cadence. It is NOT a detached task: it owns a `CancellationToken` for
//! graceful shutdown, it never spawns unmanaged work, and it has
//! well-defined failure semantics (PROGRAM.md: "Do not create detached
//! unmanaged Tokio tasks").
//!
//! # Lifecycle
//!
//! - [`Scheduler::run`] takes a `CancellationToken`. On cancel, the loop
//!   exits cleanly after any in-progress enqueue completes.
//! - The scheduler does NOT run jobs — it only enqueues them. The
//!   `arcature-jobs` worker claims and runs them. This separation matches
//!   the real subsystem: the queue owns execution; the scheduler owns
//!   cadence.
//!
//! # Cadence
//!
//! Two cadence types are supported (matching the `schedule!` macro
//! vocabulary):
//!
//! - `every "5m"` → [`ScheduleCadence::Every`] with a fixed interval.
//! - `daily "03:00"` → [`ScheduleCadence::Daily`] at a fixed UTC time.
//!
//! Time zone is UTC for A12. The `daily` cadence fires at the specified
//! hour:minute in UTC. A future phase may add timezone support if a real
//! subsystem requires it.
//!
//! # Single-instance behavior
//!
//! The scheduler is designed to run as a single instance. If the
//! application accidentally creates two schedulers with the same entries,
//! duplicate jobs will be enqueued — this is a configuration error. The
//! application lifecycle ensures only one `Scheduler::run` task is spawned.
//!
//! # Overlap behavior
//!
//! The scheduler enqueues on cadence regardless of whether the previous
//! job has completed. The job queue's `SKIP LOCKED` claim mechanism ensures
//! each enqueued job is run by at most one worker. If the application needs
//! single-flight semantics (don't enqueue if the previous job is still
//! running), that is a job-level concern (the handler checks a sentinel).

use std::future::Future;
use std::pin::Pin;
use std::time::Duration;

use chrono::{DateTime, Datelike, TimeZone, Utc};
use tokio_util::sync::CancellationToken;

use crate::dx::graph::{ScheduleBinding, ScheduleCadence};

/// A boxed future returned by a type-erased enqueue closure.
type BoxFuture = Pin<Box<dyn Future<Output = Result<(), SchedulerError>> + Send>>;

/// A type-erased enqueue function. Captures a `Jobs` clone at
/// construction time (in the `schedule!` macro's `build_scheduler`).
type EnqueueFn = Box<dyn Fn() -> BoxFuture + Send + Sync>;

/// A single scheduled entry — cadence, next fire time, and enqueue closure.
struct ScheduleEntry {
    /// The job kind (for logging / inspection).
    kind: &'static str,
    /// The job version (for logging / inspection).
    version: i16,
    /// The cadence (interval or daily time).
    cadence: ScheduleCadence,
    /// The next time this entry should fire.
    next_fire: DateTime<Utc>,
    /// The type-erased enqueue closure.
    fire: EnqueueFn,
}

/// A typed error from the scheduler (A12).
///
/// No raw `String` errors (AGENTS.md §18). Each variant is a failure that
/// can actually happen — no "future-proof" variants.
#[derive(Debug)]
pub enum SchedulerError {
    /// A job could not be enqueued (serialization, size limit, or database
    /// error). The underlying `EnqueueError` is from `arcature-jobs`.
    Enqueue(arcature_jobs::EnqueueError),
}

impl std::fmt::Display for SchedulerError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Enqueue(e) => write!(f, "scheduler enqueue failed: {e}"),
        }
    }
}

impl std::error::Error for SchedulerError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Enqueue(e) => Some(e),
        }
    }
}

/// The recurring-job scheduler (A12).
///
/// Holds type-erased enqueue closures keyed by cadence. The `schedule!`
/// macro generates a `build_scheduler(jobs)` function that constructs a
/// `Scheduler` from the declared entries. The application calls
/// `Scheduler::run(shutdown)` in a managed Tokio task.
///
/// # Example
///
/// ```ignore
/// let scheduler = build_scheduler(jobs.clone());
/// let handle = tokio::spawn(scheduler.run(shutdown.clone()));
/// // On shutdown: shutdown.cancel(); handle.await?;
/// ```
pub struct Scheduler {
    entries: Vec<ScheduleEntry>,
}

impl Scheduler {
    /// Create a new empty scheduler (no entries).
    #[must_use]
    pub fn new() -> Self {
        Self {
            entries: Vec::new(),
        }
    }

    /// Add a scheduled entry. The `fire` closure is called on each cadence
    /// tick and should enqueue the job. The closure captures a `Jobs`
    /// clone (provided by the `schedule!` macro's `build_scheduler`).
    ///
    /// The `next_fire` is computed from the cadence and the current time.
    #[must_use]
    pub fn schedule<F>(mut self, binding: &ScheduleBinding, fire: F) -> Self
    where
        F: Fn() -> BoxFuture + Send + Sync + 'static,
    {
        let next_fire = compute_next_fire(&binding.cadence, Utc::now());
        self.entries.push(ScheduleEntry {
            kind: binding.job,
            version: binding.version,
            cadence: binding.cadence.clone(),
            next_fire,
            fire: Box::new(fire),
        });
        self
    }

    /// Returns the number of scheduled entries.
    #[must_use]
    pub fn len(&self) -> usize {
        self.entries.len()
    }

    /// Returns `true` if no entries are scheduled.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    /// Run the scheduler loop until `shutdown` is cancelled.
    ///
    /// On each iteration, the scheduler:
    /// 1. Finds the earliest `next_fire` across all entries.
    /// 2. Sleeps until then (or shutdown, whichever comes first).
    /// 3. Fires all due entries (enqueues their jobs).
    /// 4. Updates each fired entry's `next_fire`.
    ///
    /// If an enqueue fails, the error is logged and the scheduler
    /// continues — one failed enqueue does not stop the scheduler. The
    /// `next_fire` is still updated so the next tick fires on schedule.
    ///
    /// Returns `Ok(())` on shutdown.
    pub async fn run(mut self, shutdown: CancellationToken) -> Result<(), SchedulerError> {
        if self.entries.is_empty() {
            // Nothing to schedule — wait for shutdown.
            shutdown.cancelled().await;
            return Ok(());
        }

        loop {
            // Find the earliest next fire time.
            let earliest = self
                .entries
                .iter()
                .map(|e| e.next_fire)
                .min()
                .unwrap_or_else(Utc::now);

            let now = Utc::now();
            let sleep_duration = if earliest > now {
                (earliest - now).to_std().unwrap_or(Duration::from_secs(0))
            } else {
                Duration::from_secs(0)
            };

            // Sleep until the earliest fire time, or until shutdown.
            tokio::select! {
                _ = shutdown.cancelled() => return Ok(()),
                _ = tokio::time::sleep(sleep_duration) => {}
            }

            // Fire all due entries.
            let now = Utc::now();
            for entry in &mut self.entries {
                if entry.next_fire <= now {
                    match (entry.fire)().await {
                        Ok(()) => {}
                        Err(e) => {
                            eprintln!(
                                "scheduler enqueue error for {} v{}: {e}",
                                entry.kind, entry.version
                            );
                        }
                    }
                    entry.next_fire = compute_next_fire(&entry.cadence, now);
                }
            }
        }
    }
}

impl Default for Scheduler {
    fn default() -> Self {
        Self::new()
    }
}

impl std::fmt::Debug for Scheduler {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Scheduler")
            .field("entries", &self.entries.len())
            .finish_non_exhaustive()
    }
}

/// Compute the next fire time for a cadence, given the current time.
fn compute_next_fire(cadence: &ScheduleCadence, now: DateTime<Utc>) -> DateTime<Utc> {
    match cadence {
        ScheduleCadence::Every { seconds } => {
            let dur = chrono::Duration::seconds(i64::try_from(*seconds).unwrap_or(i64::MAX));
            now + dur
        }
        ScheduleCadence::Daily { hour, minute } => {
            // Find the next occurrence of today (or tomorrow) at hour:minute UTC.
            let h = u32::from(*hour);
            let m = u32::from(*minute);
            // Try today at hour:minute.
            let today = Utc
                .with_ymd_and_hms(now.year(), now.month(), now.day(), h, m, 0)
                .single();
            match today {
                Some(t) if t > now => t,
                // Today's time has passed (or invalid date) → tomorrow.
                _ => {
                    let tomorrow = now + chrono::Duration::days(1);
                    Utc.with_ymd_and_hms(tomorrow.year(), tomorrow.month(), tomorrow.day(), h, m, 0)
                        .single()
                        .unwrap_or(now)
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn compute_next_fire_every_adds_interval() {
        let now = Utc.with_ymd_and_hms(2026, 1, 1, 12, 0, 0).single().unwrap();
        let cadence = ScheduleCadence::Every { seconds: 300 };
        let next = compute_next_fire(&cadence, now);
        assert_eq!(next, now + chrono::Duration::seconds(300));
    }

    #[test]
    fn compute_next_fire_daily_future_today() {
        // 12:00 now, daily at 15:00 → today 15:00.
        let now = Utc.with_ymd_and_hms(2026, 1, 1, 12, 0, 0).single().unwrap();
        let cadence = ScheduleCadence::Daily {
            hour: 15,
            minute: 0,
        };
        let next = compute_next_fire(&cadence, now);
        let expected = Utc.with_ymd_and_hms(2026, 1, 1, 15, 0, 0).single().unwrap();
        assert_eq!(next, expected);
    }

    #[test]
    fn compute_next_fire_daily_past_today_is_tomorrow() {
        // 15:00 now, daily at 12:00 → tomorrow 12:00.
        let now = Utc.with_ymd_and_hms(2026, 1, 1, 15, 0, 0).single().unwrap();
        let cadence = ScheduleCadence::Daily {
            hour: 12,
            minute: 0,
        };
        let next = compute_next_fire(&cadence, now);
        let expected = Utc.with_ymd_and_hms(2026, 1, 2, 12, 0, 0).single().unwrap();
        assert_eq!(next, expected);
    }

    #[test]
    fn compute_next_fire_daily_exact_time_is_tomorrow() {
        // 12:00 now, daily at 12:00 → tomorrow 12:00 (equal time → tomorrow).
        let now = Utc.with_ymd_and_hms(2026, 1, 1, 12, 0, 0).single().unwrap();
        let cadence = ScheduleCadence::Daily {
            hour: 12,
            minute: 0,
        };
        let next = compute_next_fire(&cadence, now);
        let expected = Utc.with_ymd_and_hms(2026, 1, 2, 12, 0, 0).single().unwrap();
        assert_eq!(next, expected);
    }

    #[test]
    fn scheduler_new_is_empty() {
        let s = Scheduler::new();
        assert!(s.is_empty());
        assert_eq!(s.len(), 0);
    }

    #[tokio::test]
    async fn scheduler_run_empty_waits_for_shutdown() {
        let s = Scheduler::new();
        let shutdown = CancellationToken::new();
        let handle = tokio::spawn(s.run(shutdown.clone()));
        shutdown.cancel();
        let result = handle.await.unwrap();
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn scheduler_shutdown_cancels_during_sleep() {
        // Schedule with a 1-hour interval, cancel after 10ms.
        let binding = ScheduleBinding {
            job: "test_job",
            version: 1,
            cadence: ScheduleCadence::Every { seconds: 3600 },
        };
        let s = Scheduler::new().schedule(&binding, || Box::pin(async { Ok(()) }));
        let shutdown = CancellationToken::new();
        let handle = tokio::spawn(s.run(shutdown.clone()));
        tokio::time::sleep(Duration::from_millis(10)).await;
        shutdown.cancel();
        let result = handle.await.unwrap();
        assert!(result.is_ok());
    }
}