shared-framework 0.0.17

//! Background jobs: periodic tasks, daily schedules, or both.
//!
//! Implement [`ServiceJob`] for each task and run it through [`JobRegistry`].
//! [`JobType::Periodic`] runs on an interval via a Tokio task,
//! [`JobType::Exact`] runs once per day at a [`TimeOfDay`] via the cron
//! scheduler, and [`JobType::PeriodicAndExact`] runs both legs.
//!
//! Key types: [`ServiceJob`] for the task, [`JobRegistry`] for ownership and
//! lifecycle, [`JobType`] for scheduling behavior, [`TimeOfDay`] for daily times.
//!
//! Use this module for maintenance work such as cleanup or aggregation.
//!
//! ```ignore
//! # use std::time::Duration;
//! # use crate::job::{JobRegistry, JobType, ServiceJob, TimeOfDay};
//! # use crate::response::ServiceResult;
//! struct Cleanup;
//!
//! #[async_trait::async_trait]
//! impl ServiceJob for Cleanup {
//!     fn name(&self) -> &str { "cleanup" }
//!     fn job_type(&self) -> JobType { JobType::Periodic }
//!     fn period(&self) -> Option<Duration> { Some(Duration::from_secs(60)) }
//!     fn schedule(&self) -> Option<TimeOfDay> { None }
//!     async fn run(&self) -> anyhow::Result<ServiceResult<serde_json::Value>> {
//!         Ok(ServiceResult::ok("cleaned", serde_json::json!({})))
//!     }
//! }
//!
//! # async fn start() -> anyhow::Result<()> {
//! let mut registry = JobRegistry::new();
//! registry.add_job(Cleanup);
//! registry.start().await?;
//! # Ok(())
//! # }
//! ```

use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio_cron_scheduler::{Job, JobScheduler};

use crate::response::ServiceResult;

/// Scheduling behavior of a [`ServiceJob`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JobType {
    /// Runs repeatedly on the [`ServiceJob::period`] interval.
    Periodic,
    /// Runs once per day at the [`ServiceJob::schedule`] time.
    Exact,
    /// Runs both on the interval and once per day at the scheduled time.
    PeriodicAndExact,
}

/// A time of day for [`JobType::Exact`] schedules.
///
/// All fields use clock ranges; [`new`](Self::new) rejects out-of-range values.
#[derive(Debug, Clone)]
pub struct TimeOfDay {
    /// Hour of day, 0–23.
    pub hour: u8,
    /// Minute of the hour, 0–59.
    pub minute: u8,
    /// Second of minute, 0–59.
    pub second: u8
}

impl TimeOfDay {
    /// Creates a time of day, returning an error string for out-of-range fields.
    pub fn new(hour: u8, minute: u8, second: u8) -> Result<Self, String> {
        if hour > 23 { return Err("hour >23".into()); }
        if minute > 59 { return Err("minute >59".into()); }
        if second > 59 { return Err("second >59".into()); }
        Ok(Self { hour, minute, second })
    }
}

/// A unit of background work managed by [`JobRegistry`].
///
/// `period` supplies the repeat interval for `Periodic` jobs (defaulting to
/// 60 seconds when the job type needs it but `None` is returned), and
/// `schedule` supplies the daily time for `Exact` jobs.
#[async_trait::async_trait]
pub trait ServiceJob: Send + Sync {
    /// Stable name of the job, used in logs.
    fn name(&self) -> &str;
    /// Scheduling behavior of the job.
    fn job_type(&self) -> JobType;
    /// Repeat interval for periodic legs; `None` means no interval configured.
    fn period(&self) -> Option<Duration>;
    /// Daily time for exact legs; `None` means no daily time configured.
    fn schedule(&self) -> Option<TimeOfDay>;
    /// Whether the job is critical. Defaults to `false`.
    fn is_critical(&self) -> bool { false }
    /// Executes one run of the job.
    async fn run(&self) -> anyhow::Result<ServiceResult<serde_json::Value>>;
    /// Releases job resources on shutdown. Defaults to doing nothing.
    async fn stop_gracefully(&self) -> anyhow::Result<()> { Ok(()) }
}

#[derive(Clone)]
struct JobEntry {
    job: Arc<dyn ServiceJob>,
    runs: Arc<Mutex<usize>>,
    failures: Arc<Mutex<usize>>,
    successes: Arc<Mutex<usize>>,
}

/// Owns jobs and drives their schedules.
///
/// Add jobs with [`add_job`](Self::add_job), start all schedules with
/// [`start`](Self::start), and release resources with [`stop`](Self::stop).
/// `Periodic` legs run as Tokio interval tasks; `Exact` legs are cron entries
/// (`PeriodicAndExact` daily leg fires on day 1 of each month).
pub struct JobRegistry {
    jobs: Vec<JobEntry>,
    scheduler: Option<JobScheduler>,
}

impl JobRegistry {
    /// Creates an empty registry with no scheduler running.
    pub fn new() -> Self { Self { jobs: vec![], scheduler: None } }

    /// Returns the total number of runs of a job.
    pub fn get_runs(&self, name: &str) -> Option<usize> {
        for entry in &self.jobs {
            if entry.job.name() == name {
                return Some(entry.runs.lock().unwrap().clone());
            }
        }
        None
    }

    /// Returns the number of successful runs of a job.
    pub fn get_failures(&self, name: &str) -> Option<usize> {
        for entry in &self.jobs {
            if entry.job.name() == name {
                return Some(entry.failures.lock().unwrap().clone());
            }
        }
        None
    }


    /// Returns the number of successful runs of a job.
    pub fn get_successes(&self, name: &str) -> Option<usize> {
        for entry in &self.jobs {
            if entry.job.name() == name {
                return Some(entry.successes.lock().unwrap().clone());
            }
        }
        None
    }

    /// Adds a job to the registry. The job runs once [`start`](Self::start) is called.
    ///
    /// `J` is the concrete [`ServiceJob`] implementation being stored.
    pub fn add_job<J: ServiceJob + 'static>(&mut self, job: J) {
        self.jobs.push(JobEntry {
            job: Arc::new(job),
            runs: Arc::new(Mutex::new(0)),
            failures: Arc::new(Mutex::new(0)),
            successes: Arc::new(Mutex::new(0)),
        });
    }

    /// Returns the number of jobs in the registry.
    pub fn job_count(&self) -> usize { self.jobs.len() }

    /// Starts the scheduler and all configured job legs.
    ///
    /// `Periodic` jobs without a period default to 60 seconds; `Exact` legs
    /// without a schedule are skipped. Failures of individual runs are logged
    /// and do not stop the schedule. Returns an error if the scheduler cannot start.
    pub async fn start(&mut self) -> anyhow::Result<()> {
        let sched = JobScheduler::new().await?;
        tracing::info!(job_count = self.jobs.len(), "Starting job scheduler");
        for entry in &self.jobs {
            let job_clone = entry.job.clone();
            match job_clone.job_type() {
                JobType::Periodic => {
                    let period = job_clone.period().unwrap_or(Duration::from_secs(60));
                    tracing::info!(job = %job_clone.name(), schedule = "periodic", interval_secs = period.as_secs(), "Registered job");
                    // cron every N seconds: use Tokio interval instead of cron for simplicity
                    // Spawn periodic task
                    tokio::spawn({
                        let job_clone = job_clone.clone();
                        async move {
                            let mut interval = tokio::time::interval(period);
                            loop {
                                interval.tick().await;
                                if let Err(e) = job_clone.run().await {
                                    tracing::error!(job = %job_clone.name(), error = %e, "Periodic job failed");
                                }
                            }
                        }
                    });
                }
                JobType::PeriodicAndExact => {
                    let schedule_clone = job_clone.clone();
                    let period_clone = job_clone.clone();
                    if let Some(tod) = schedule_clone.schedule() {
                        let job_name = schedule_clone.name().to_string();
                        // Monthly, on day 1 at the scheduled time.
                        let cron = format!("{} {} {} 1 * *", tod.second, tod.minute, tod.hour);
                        let j = Job::new_async(cron.as_str(), move |_, _| {
                            let jc = schedule_clone.clone();
                            Box::pin(async move {
                                if let Err(e) = jc.run().await {
                                    tracing::error!(job = %jc.name(), error = %e, "Scheduled job failed");
                                }
                            })
                        })?;
                        sched.add(j).await?;
                        tracing::info!(job = %job_name, schedule = "exact", "Registered scheduled job");
                    }
                    // Also honor periodic interval if provided (PeriodicAndExact = both)
                    if let Some(period) = period_clone.period() {
                        tracing::info!(job = %period_clone.name(), schedule = "periodic", interval_secs = period.as_secs(), "Registered periodic job leg");
                        let pc = period_clone.clone();
                        tokio::spawn(async move {
                            let mut interval = tokio::time::interval(period);
                            loop {
                                interval.tick().await;
                                if let Err(e) = pc.run().await {
                                    tracing::error!(job = %pc.name(), error = %e, "Periodic job leg failed");
                                }
                            }
                        });
                    }
                }
                JobType::Exact => {
                    if let Some(tod) = job_clone.schedule() {
                        let job_name = job_clone.name().to_string();
                        let cron = format!("{} {} {} * * *", tod.second, tod.minute, tod.hour);
                        let j = Job::new_async(cron.as_str(), move |_, _| {
                            let jc = job_clone.clone();
                            Box::pin(async move {
                                if let Err(e) = jc.run().await {
                                    tracing::error!(job = %jc.name(), error = %e, "Scheduled job failed");
                                }
                            })
                        })?;
                        sched.add(j).await?;
                        tracing::info!(job = %job_name, schedule = "exact", "Registered scheduled job");
                    }
                }
            }
        }
        sched.start().await?;
        self.scheduler = Some(sched);
        tracing::info!("Job scheduler started");
        Ok(())
    }

    /// Stops every job gracefully and shuts down the scheduler, if running.
    ///
    /// Returns the first error raised by a job's graceful stop or by shutdown.
    pub async fn stop(&mut self) -> anyhow::Result<()> {
        tracing::info!(job_count = self.jobs.len(), "Stopping jobs");
        for entry in &self.jobs {
            entry.job.stop_gracefully().await?;
        }
        if let Some(mut s) = self.scheduler.take() {
            s.shutdown().await?;
        }
        tracing::info!("Jobs stopped");
        Ok(())
    }
}

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