zeph-scheduler 0.22.4

// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
// SPDX-License-Identifier: MIT OR Apache-2.0

use zeph_db::DbPool;
#[cfg(any(feature = "sqlite", feature = "postgres"))]
use zeph_db::sql;

use crate::error::SchedulerError;
use crate::task::CronExpr;

/// A scheduled task row returned by [`JobStore::list_jobs`].
///
/// Replaces the previous `(String, String, String, String)` tuple to eliminate
/// positional destructuring bugs. Fields map 1-to-1 to the SQL columns in the
/// same order as the query: `name`, `kind`, `task_mode`, and the coalesced
/// `next_run`.
///
/// # Examples
///
/// ```rust,no_run
/// use zeph_scheduler::JobStore;
///
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// let store = JobStore::open("sqlite:scheduler.db").await?;
/// store.init().await?;
///
/// for job in store.list_jobs().await? {
///     println!("{}: {} ({}) → {}", job.name, job.kind, job.task_mode, job.next_run);
/// }
/// # Ok(())
/// # }
/// ```
#[derive(Debug, Clone)]
pub struct ScheduledTaskRecord {
    /// Unique task name (primary key in the `scheduled_jobs` table).
    pub name: String,
    /// Serialised [`crate::TaskKind`] string (e.g. `"health_check"`).
    pub kind: String,
    /// Execution mode: `"periodic"` or `"oneshot"`.
    pub task_mode: String,
    /// Next scheduled run time as an ISO 8601 / RFC 3339 string.
    ///
    /// Falls back to `run_at` for one-shot jobs that have not yet computed a
    /// `next_run`. Empty string when neither field is set.
    pub next_run: String,
}

/// Full details for a scheduled task, returned by [`JobStore::list_jobs_full`].
///
/// Intended for display in the TUI or CLI task list. All string fields are UTF-8
/// and come directly from the `scheduled_jobs` `SQLite` table.
#[derive(Debug, Clone)]
pub struct ScheduledTaskInfo {
    /// Unique task name (primary key in the `scheduled_jobs` table).
    pub name: String,
    /// Serialised [`crate::TaskKind`] string (e.g. `"health_check"`).
    pub kind: String,
    /// Execution mode: `"periodic"` or `"oneshot"`.
    pub task_mode: String,
    /// Validated cron expression for periodic tasks; `None` for one-shot tasks or rows
    /// whose stored expression is invalid (those are marked `status = "error"` in the DB).
    pub cron_expr: Option<CronExpr>,
    /// Next scheduled run time as an ISO 8601 / RFC 3339 string, or empty if unknown.
    pub next_run: String,
    /// Stored task prompt for custom tasks; empty for config-driven built-in tasks.
    pub task_data: String,
    /// Current job status: `"pending"`, `"completed"`, `"done"`, or `"error"`.
    pub status: String,
    /// RTW-A provenance tag: `"static"`, `"user_added"`, or `"external"`.
    pub provenance: String,
    /// Last recorded run time (RFC 3339), or `None` if the task has never run.
    pub last_run: Option<String>,
    /// Raw `run_at` column value (one-shot tasks only), independent of the `next_run`
    /// coalescing applied to [`ScheduledTaskInfo::next_run`]. `Scheduler::init()` hydration
    /// reads this field (rather than the coalesced `next_run`) to reconstruct one-shot rows
    /// unambiguously (#6361).
    pub run_at: Option<String>,
}

/// A single row returned by [`JobStore::list_recent_runs`], ordered by recency.
#[derive(Debug, Clone)]
pub struct RecentRunRecord {
    /// Unique task name.
    pub name: String,
    /// Execution mode: `"periodic"` or `"oneshot"`.
    pub task_mode: String,
    /// Last recorded run time (RFC 3339), or `None` if the task has never run.
    pub last_run: Option<String>,
    /// Next scheduled run time as an ISO 8601 / RFC 3339 string, or empty if unknown.
    pub next_run: String,
    /// Current job status: `"pending"`, `"completed"`, `"done"`, or `"error"`.
    pub status: String,
}

/// Persistent storage layer for scheduled jobs.
///
/// All job definitions and run history are stored in a `SQLite` database managed by
/// `zeph-db` migrations. The `scheduled_jobs` table schema is defined in migration
/// `051_scheduler_jobs.sql`.
///
/// # Examples
///
/// ```rust,no_run
/// use zeph_scheduler::JobStore;
///
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// // Open from a file path.
/// let store = JobStore::open("sqlite:scheduler.db").await?;
/// store.init().await?;
///
/// // Query job list.
/// let jobs = store.list_jobs().await?;
/// for job in &jobs {
///     println!("{}: {} ({}) → {}", job.name, job.kind, job.task_mode, job.next_run);
/// }
/// # Ok(())
/// # }
/// ```
#[derive(Debug)]
pub struct JobStore {
    pool: DbPool,
}

impl JobStore {
    /// Create a `JobStore` from an existing [`zeph_db::DbPool`].
    ///
    /// You must call [`JobStore::init`] before any other operation to ensure the
    /// schema migrations have been applied.
    #[must_use]
    pub fn new(pool: DbPool) -> Self {
        Self { pool }
    }

    /// Open (or create) a `JobStore` from a `SQLite` file path.
    ///
    /// # Errors
    ///
    /// Returns [`SchedulerError::Db`] if the connection cannot be established.
    #[tracing::instrument(name = "sched.store.open", skip_all, fields(path = %path), err)]
    pub async fn open(path: &str) -> Result<Self, SchedulerError> {
        let pool = zeph_db::DbConfig {
            url: path.to_string(),
            pool_size: 5,
        }
        .connect()
        .await?;
        Ok(Self { pool })
    }

    /// Run all pending migrations on the underlying pool.
    ///
    /// Replaces the former inline `CREATE TABLE IF NOT EXISTS` DDL. The
    /// `scheduled_jobs` schema is now managed by migration
    /// `051_scheduler_jobs.sql` in `zeph-db`.
    ///
    /// # Errors
    ///
    /// Returns an error if any migration fails.
    #[tracing::instrument(name = "sched.store.init", skip_all, err)]
    pub async fn init(&self) -> Result<(), SchedulerError> {
        zeph_db::run_migrations(&self.pool).await?;
        Ok(())
    }

    /// Upsert a job definition.
    ///
    /// # Errors
    ///
    /// Returns an error if the SQL statement fails.
    #[tracing::instrument(name = "sched.store.upsert_job", skip_all, fields(task = %name), err)]
    pub async fn upsert_job(
        &self,
        name: &str,
        cron_expr: &str,
        kind: &str,
    ) -> Result<(), SchedulerError> {
        self.upsert_job_with_mode(name, cron_expr, kind, "periodic", None, "")
            .await
    }

    /// Upsert a job definition with explicit `task_mode`, optional `run_at`, and `task_data`.
    ///
    /// # Errors
    ///
    /// Returns an error if the SQL statement fails.
    #[tracing::instrument(name = "sched.store.upsert_job_with_mode", skip_all, fields(task = %name), err)]
    pub async fn upsert_job_with_mode(
        &self,
        name: &str,
        cron_expr: &str,
        kind: &str,
        task_mode: &str,
        run_at: Option<&str>,
        task_data: &str,
    ) -> Result<(), SchedulerError> {
        self.upsert_job_with_provenance(
            name, cron_expr, kind, task_mode, run_at, task_data, "static",
        )
        .await
    }

    /// Upsert a job definition with explicit `task_mode`, `task_data`, and RTW-A `provenance`.
    ///
    /// # Errors
    ///
    /// Returns an error if the SQL statement fails.
    #[allow(clippy::too_many_arguments)]
    #[tracing::instrument(name = "sched.store.upsert_job_with_provenance", skip_all, fields(task = %name), err)]
    pub async fn upsert_job_with_provenance(
        &self,
        name: &str,
        cron_expr: &str,
        kind: &str,
        task_mode: &str,
        run_at: Option<&str>,
        task_data: &str,
        provenance: &str,
    ) -> Result<(), SchedulerError> {
        zeph_db::query(sql!(
            "INSERT INTO scheduled_jobs (name, cron_expr, kind, task_mode, run_at, task_data, provenance)
             VALUES (?, ?, ?, ?, ?, ?, ?)
             ON CONFLICT(name) DO UPDATE SET
               cron_expr = excluded.cron_expr,
               kind = excluded.kind,
               task_mode = excluded.task_mode,
               run_at = excluded.run_at,
               task_data = excluded.task_data,
               provenance = excluded.provenance"
        ))
        .bind(name)
        .bind(cron_expr)
        .bind(kind)
        .bind(task_mode)
        .bind(run_at)
        .bind(task_data)
        .bind(provenance)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Insert a new job. Returns [`SchedulerError::DuplicateJob`] if a job with this name exists.
    ///
    /// This is the CLI write path (`zeph schedule add`, both the cron and `--run-at` forms), so
    /// the row is stored with [`crate::task::TaskProvenance::UserAdded`] provenance — reflecting
    /// its genuine CLI/user origin — rather than relying on the `provenance` column's DB default.
    /// `Scheduler::init()` hydration still force-downgrades out-of-process rows to
    /// [`crate::task::TaskProvenance::External`] regardless of the stored value (#6114), so this
    /// does not change runtime trust decisions; it keeps the stored label accurate for direct
    /// inspection (e.g. a raw DB dump) — `zeph schedule list`/`show` do not currently surface
    /// `provenance`.
    ///
    /// # Errors
    ///
    /// Returns [`SchedulerError::DuplicateJob`] on unique constraint violation,
    /// or [`SchedulerError::Database`] on other SQL errors.
    #[tracing::instrument(name = "sched.store.insert_job", skip_all, fields(task = %name), err)]
    pub async fn insert_job(
        &self,
        name: &str,
        cron_expr: &str,
        kind: &str,
        task_mode: &str,
        run_at: Option<&str>,
        task_data: &str,
    ) -> Result<(), SchedulerError> {
        let result = zeph_db::query(sql!(
            "INSERT INTO scheduled_jobs (name, cron_expr, kind, task_mode, run_at, task_data, provenance)
             VALUES (?, ?, ?, ?, ?, ?, ?)"
        ))
        .bind(name)
        .bind(cron_expr)
        .bind(kind)
        .bind(task_mode)
        .bind(run_at)
        .bind(task_data)
        .bind(crate::task::TaskProvenance::UserAdded.as_str())
        .execute(&self.pool)
        .await;
        match result {
            Ok(_) => Ok(()),
            Err(zeph_db::SqlxError::Database(db_err))
                if db_err.message().contains("UNIQUE constraint failed")
                    || db_err.code().as_deref() == Some("23505") =>
            {
                Err(SchedulerError::DuplicateJob(name.to_string()))
            }
            Err(e) => Err(SchedulerError::Database(e)),
        }
    }

    /// Record a job execution and persist the next scheduled run time.
    ///
    /// # Errors
    ///
    /// Returns an error if the SQL statement fails.
    #[tracing::instrument(name = "sched.store.record_run", skip_all, fields(task = %name), err)]
    pub async fn record_run(
        &self,
        name: &str,
        timestamp: &str,
        next_run: &str,
    ) -> Result<(), SchedulerError> {
        zeph_db::query(
            sql!("UPDATE scheduled_jobs SET last_run = ?, next_run = ?, status = 'completed' WHERE name = ?"),
        )
        .bind(timestamp)
        .bind(next_run)
        .bind(name)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Mark a one-shot job as done.
    ///
    /// # Errors
    ///
    /// Returns an error if the SQL statement fails.
    #[tracing::instrument(name = "sched.store.mark_done", skip_all, fields(task = %name), err)]
    pub async fn mark_done(&self, name: &str) -> Result<(), SchedulerError> {
        zeph_db::query(sql!(
            "UPDATE scheduled_jobs SET status = 'done', last_run = CURRENT_TIMESTAMP WHERE name = ?"
        ))
        .bind(name)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Mark a job as permanently errored (e.g. invalid cron expression on hydration).
    ///
    /// The job remains visible in [`JobStore::list_jobs_full`] with `status = "error"` so
    /// operators can identify it via `zeph scheduler list` without reading debug logs.
    ///
    /// # Errors
    ///
    /// Returns an error if the SQL statement fails.
    #[tracing::instrument(name = "sched.store.mark_error", skip_all, fields(task = %name), err)]
    pub async fn mark_error(&self, name: &str) -> Result<(), SchedulerError> {
        zeph_db::query(sql!(
            "UPDATE scheduled_jobs SET status = 'error' WHERE name = ?"
        ))
        .bind(name)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Delete a job by name.
    ///
    /// # Errors
    ///
    /// Returns an error if the SQL statement fails.
    #[tracing::instrument(name = "sched.store.delete_job", skip_all, fields(task = %name), err)]
    pub async fn delete_job(&self, name: &str) -> Result<bool, SchedulerError> {
        let result = zeph_db::query(sql!("DELETE FROM scheduled_jobs WHERE name = ?"))
            .bind(name)
            .execute(&self.pool)
            .await?;
        Ok(result.rows_affected() > 0)
    }

    /// Check if a job with the given name exists.
    ///
    /// # Errors
    ///
    /// Returns an error if the SQL query fails.
    #[tracing::instrument(name = "sched.store.job_exists", skip_all, fields(task = %name), err)]
    pub async fn job_exists(&self, name: &str) -> Result<bool, SchedulerError> {
        let row: Option<(i64,)> =
            zeph_db::query_as(sql!("SELECT 1 FROM scheduled_jobs WHERE name = ?"))
                .bind(name)
                .fetch_optional(&self.pool)
                .await?;
        Ok(row.is_some())
    }

    /// Persist the next scheduled run time for a job (used during init).
    ///
    /// # Errors
    ///
    /// Returns an error if the SQL statement fails.
    #[tracing::instrument(name = "sched.store.set_next_run", skip_all, fields(task = %name), err)]
    pub async fn set_next_run(&self, name: &str, next_run: &str) -> Result<(), SchedulerError> {
        zeph_db::query(sql!(
            "UPDATE scheduled_jobs SET next_run = ? WHERE name = ?"
        ))
        .bind(next_run)
        .bind(name)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Get the persisted next run timestamp for a job.
    ///
    /// # Errors
    ///
    /// Returns an error if the SQL query fails.
    #[tracing::instrument(name = "sched.store.get_next_run", skip_all, fields(task = %name), err)]
    pub async fn get_next_run(&self, name: &str) -> Result<Option<String>, SchedulerError> {
        let row: Option<(Option<String>,)> =
            zeph_db::query_as(sql!("SELECT next_run FROM scheduled_jobs WHERE name = ?"))
                .bind(name)
                .fetch_optional(&self.pool)
                .await?;
        Ok(row.and_then(|r| r.0))
    }

    /// List all active (non-done) jobs.
    ///
    /// Returns a [`ScheduledTaskRecord`] per active job, ordered by name.
    /// One-shot jobs without a computed `next_run` fall back to their `run_at` value;
    /// if neither is set the field is an empty string.
    ///
    /// # Errors
    ///
    /// Returns an error if the SQL query fails.
    #[tracing::instrument(name = "sched.store.list_jobs", skip_all, err)]
    pub async fn list_jobs(&self) -> Result<Vec<ScheduledTaskRecord>, SchedulerError> {
        let rows: Vec<(String, String, String, Option<String>)> = zeph_db::query_as(
            sql!("SELECT name, kind, task_mode, COALESCE(next_run, run_at) FROM scheduled_jobs WHERE status != 'done' ORDER BY name"),
        )
        .fetch_all(&self.pool)
        .await?;
        Ok(rows
            .into_iter()
            .map(|(name, kind, task_mode, next_run)| ScheduledTaskRecord {
                name,
                kind,
                task_mode,
                next_run: next_run.unwrap_or_default(),
            })
            .collect())
    }

    /// List all active (non-done) jobs with full details for display.
    ///
    /// The `cron_expr` field of each returned [`ScheduledTaskInfo`] is `Some` for periodic tasks
    /// with a valid expression and `None` for one-shot tasks or rows whose stored expression
    /// failed validation. Invalid rows are not filtered out — callers can check `status` to
    /// distinguish error rows.
    ///
    /// # Errors
    ///
    /// Returns an error if the SQL query fails.
    #[tracing::instrument(name = "sched.store.list_jobs_full", skip_all, err)]
    pub async fn list_jobs_full(&self) -> Result<Vec<ScheduledTaskInfo>, SchedulerError> {
        #[allow(clippy::type_complexity)]
        let rows: Vec<(
            String,
            String,
            String,
            String,
            Option<String>,
            String,
            String,
            String,
            Option<String>,
            Option<String>,
        )> = zeph_db::query_as(sql!(
            "SELECT name, kind, task_mode, cron_expr, COALESCE(next_run, run_at), task_data, status, provenance, last_run, run_at \
             FROM scheduled_jobs WHERE status != 'done' ORDER BY name"
        ))
        .fetch_all(&self.pool)
        .await?;
        Ok(rows
            .into_iter()
            .map(
                |(
                    name,
                    kind,
                    task_mode,
                    raw_cron,
                    next_run,
                    task_data,
                    status,
                    provenance,
                    last_run,
                    run_at,
                )| {
                    // Empty string = oneshot task (no cron). Non-empty = validate eagerly.
                    let cron_expr = if raw_cron.is_empty() {
                        None
                    } else {
                        match CronExpr::try_from(raw_cron.as_str()) {
                            Ok(expr) => Some(expr),
                            Err(e) => {
                                tracing::warn!(
                                    task = %name,
                                    cron_expr = %raw_cron,
                                    "invalid cron expression in DB row: {e}"
                                );
                                None
                            }
                        }
                    };
                    ScheduledTaskInfo {
                        name,
                        kind,
                        task_mode,
                        cron_expr,
                        next_run: next_run.unwrap_or_default(),
                        task_data,
                        status,
                        provenance,
                        last_run,
                        run_at,
                    }
                },
            )
            .collect())
    }

    /// Count active (non-done) jobs.
    ///
    /// # Errors
    ///
    /// Returns an error if the SQL query fails.
    #[tracing::instrument(name = "sched.store.count_active_jobs", skip_all, err)]
    pub async fn count_active_jobs(&self) -> Result<usize, SchedulerError> {
        let (count,): (i64,) = zeph_db::query_as(sql!(
            "SELECT COUNT(*) FROM scheduled_jobs WHERE status != 'done'"
        ))
        .fetch_one(&self.pool)
        .await?;
        Ok(usize::try_from(count).unwrap_or(0))
    }

    /// List the `n` most recently run (non-done) jobs, ordered by `last_run` descending.
    ///
    /// Never-run jobs (`last_run IS NULL`) sort last. Ordering and truncation are pushed into
    /// SQL via `ORDER BY last_run IS NULL, last_run DESC LIMIT ?` rather than fetched-then-sorted
    /// in Rust: `last_run IS NULL` evaluates to a sortable `0`/`1` (`SQLite`) or `false`/`true`
    /// (`PostgreSQL`) value, giving "`NULLS LAST`" semantics on both backends without relying on
    /// `PostgreSQL`-only `NULLS LAST` syntax, which `SQLite` does not support.
    ///
    /// # Errors
    ///
    /// Returns an error if the SQL query fails.
    #[tracing::instrument(name = "sched.store.list_recent_runs", skip_all, err)]
    pub async fn list_recent_runs(&self, n: usize) -> Result<Vec<RecentRunRecord>, SchedulerError> {
        #[allow(clippy::type_complexity)]
        let rows: Vec<(String, String, Option<String>, Option<String>, String)> =
            zeph_db::query_as(sql!(
                "SELECT name, task_mode, last_run, COALESCE(next_run, run_at), status \
                 FROM scheduled_jobs WHERE status != 'done' \
                 ORDER BY last_run IS NULL, last_run DESC LIMIT ?"
            ))
            .bind(i64::try_from(n).unwrap_or(i64::MAX))
            .fetch_all(&self.pool)
            .await?;
        Ok(rows
            .into_iter()
            .map(
                |(name, task_mode, last_run, next_run, status)| RecentRunRecord {
                    name,
                    task_mode,
                    last_run,
                    next_run: next_run.unwrap_or_default(),
                    status,
                },
            )
            .collect())
    }

    /// Returns a reference to the underlying connection pool.
    ///
    /// Primarily used in tests that need to execute raw SQL against the same database.
    #[must_use]
    pub fn pool(&self) -> &DbPool {
        &self.pool
    }
}

// Every test in this module opens a real connection pool via the `:memory:` sentinel, which is
// SQLite-specific: under `--features postgres`, `DbConfig::connect()` takes cfg-priority and routes
// `:memory:` into `connect_postgres`, which fails to parse it as a Postgres URL. See #5608.
#[cfg(all(test, feature = "sqlite"))]
mod tests {
    use super::*;

    async fn test_pool() -> DbPool {
        zeph_db::DbConfig {
            url: ":memory:".to_string(),
            pool_size: 5,
        }
        .connect()
        .await
        .unwrap()
    }

    #[tokio::test]
    async fn init_creates_table() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        assert!(store.init().await.is_ok());
    }

    #[tokio::test]
    async fn upsert_and_query() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();

        store
            .upsert_job("test_job", "0 * * * * *", "health_check")
            .await
            .unwrap();
        assert!(store.get_next_run("test_job").await.unwrap().is_none());

        store
            .record_run("test_job", "2026-01-01T00:00:00Z", "2026-01-01T00:01:00Z")
            .await
            .unwrap();
        assert_eq!(
            store.get_next_run("test_job").await.unwrap().as_deref(),
            Some("2026-01-01T00:01:00Z")
        );
    }

    #[tokio::test]
    async fn upsert_updates_existing() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();

        store
            .upsert_job("job1", "0 * * * * *", "health_check")
            .await
            .unwrap();
        store
            .upsert_job("job1", "0 0 * * * *", "memory_cleanup")
            .await
            .unwrap();

        let row: (String,) =
            zeph_db::query_as(sql!("SELECT kind FROM scheduled_jobs WHERE name = 'job1'"))
                .fetch_one(store.pool())
                .await
                .unwrap();
        assert_eq!(row.0, "memory_cleanup");
    }

    #[tokio::test]
    async fn next_run_nonexistent_job() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();
        assert!(store.get_next_run("no_such_job").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn job_exists_returns_true_for_existing() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();
        store
            .upsert_job("exists_job", "0 * * * * *", "health_check")
            .await
            .unwrap();
        assert!(store.job_exists("exists_job").await.unwrap());
        assert!(!store.job_exists("missing").await.unwrap());
    }

    #[tokio::test]
    async fn delete_job_removes_row() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();
        store
            .upsert_job("del_job", "0 * * * * *", "health_check")
            .await
            .unwrap();
        assert!(store.delete_job("del_job").await.unwrap());
        assert!(!store.job_exists("del_job").await.unwrap());
        assert!(!store.delete_job("del_job").await.unwrap());
    }

    #[tokio::test]
    async fn mark_done_sets_status() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();
        store
            .upsert_job_with_mode(
                "os_job",
                "",
                "health_check",
                "oneshot",
                Some("2026-01-01T01:00:00Z"),
                "",
            )
            .await
            .unwrap();
        store.mark_done("os_job").await.unwrap();
        let row: (String,) = zeph_db::query_as(sql!(
            "SELECT status FROM scheduled_jobs WHERE name = 'os_job'"
        ))
        .fetch_one(store.pool())
        .await
        .unwrap();
        assert_eq!(row.0, "done");
    }

    #[tokio::test]
    async fn list_jobs_excludes_done_jobs() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();
        store
            .upsert_job_with_mode(
                "done_job",
                "",
                "health_check",
                "oneshot",
                Some("2026-01-01T01:00:00Z"),
                "",
            )
            .await
            .unwrap();
        store.mark_done("done_job").await.unwrap();
        let jobs = store.list_jobs().await.unwrap();
        assert!(
            jobs.iter().all(|j| j.name != "done_job"),
            "list_jobs must not return done jobs"
        );
    }

    #[tokio::test]
    async fn list_jobs_uses_run_at_for_oneshot_when_next_run_is_null() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();
        store
            .upsert_job_with_mode(
                "oneshot_job",
                "",
                "custom",
                "oneshot",
                Some("2026-06-01T10:00:00Z"),
                "",
            )
            .await
            .unwrap();
        let jobs = store.list_jobs().await.unwrap();
        let job = jobs.iter().find(|j| j.name == "oneshot_job").unwrap();
        assert_eq!(
            job.next_run, "2026-06-01T10:00:00Z",
            "run_at must be shown as next_run for oneshot jobs"
        );
    }

    #[tokio::test]
    async fn list_jobs_full_returns_correct_fields() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();
        store
            .upsert_job("periodic_job", "0 0 3 * * *", "memory_cleanup")
            .await
            .unwrap();
        store
            .upsert_job_with_mode(
                "oneshot_job",
                "",
                "custom",
                "oneshot",
                Some("2030-01-01T10:00:00Z"),
                "",
            )
            .await
            .unwrap();

        store
            .record_run(
                "periodic_job",
                "2026-01-01T00:00:00Z",
                "2026-01-02T00:00:00Z",
            )
            .await
            .unwrap();

        let jobs = store.list_jobs_full().await.unwrap();
        assert_eq!(jobs.len(), 2);

        let periodic = jobs.iter().find(|j| j.name == "periodic_job").unwrap();
        assert_eq!(periodic.kind, "memory_cleanup");
        assert_eq!(periodic.task_mode, "periodic");
        assert_eq!(
            periodic.cron_expr.as_ref().map(CronExpr::as_str),
            Some("0 0 3 * * *")
        );
        assert_eq!(
            periodic.last_run.as_deref(),
            Some("2026-01-01T00:00:00Z"),
            "last_run must reflect the timestamp recorded by record_run"
        );

        let oneshot = jobs.iter().find(|j| j.name == "oneshot_job").unwrap();
        assert_eq!(oneshot.kind, "custom");
        assert_eq!(oneshot.task_mode, "oneshot");
        assert!(oneshot.cron_expr.is_none());
        assert_eq!(oneshot.next_run, "2030-01-01T10:00:00Z");
        assert_eq!(
            oneshot.run_at.as_deref(),
            Some("2030-01-01T10:00:00Z"),
            "run_at must be the raw column value, independent of the next_run coalescing (#6361 C1)"
        );
        assert_eq!(
            oneshot.last_run, None,
            "a task that has never run must report last_run = None"
        );
    }

    #[tokio::test]
    async fn list_jobs_full_excludes_done_jobs() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();
        store
            .upsert_job_with_mode(
                "done_job",
                "",
                "custom",
                "oneshot",
                Some("2026-01-01T01:00:00Z"),
                "",
            )
            .await
            .unwrap();
        store.mark_done("done_job").await.unwrap();
        let jobs = store.list_jobs_full().await.unwrap();
        assert!(jobs.iter().all(|j| j.name != "done_job"));
    }

    #[tokio::test]
    async fn duplicate_name_detected() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();
        store
            .upsert_job("dup", "0 * * * * *", "health_check")
            .await
            .unwrap();
        assert!(store.job_exists("dup").await.unwrap());
    }

    #[tokio::test]
    async fn insert_job_success() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();
        store
            .insert_job(
                "new_job",
                "0 * * * * *",
                "custom",
                "periodic",
                None,
                "run daily report",
            )
            .await
            .unwrap();
        assert!(store.job_exists("new_job").await.unwrap());
    }

    /// `insert_job` is the CLI write path (`zeph schedule add`), so it must stamp
    /// `provenance = "user_added"` rather than falling through to the `scheduled_jobs.provenance`
    /// column's schema default. Regression test for #6442.
    #[tokio::test]
    async fn insert_job_stores_user_added_provenance() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();
        store
            .insert_job(
                "cli_added_job",
                "0 * * * * *",
                "custom",
                "periodic",
                None,
                "run daily report",
            )
            .await
            .unwrap();
        let jobs = store.list_jobs_full().await.unwrap();
        let job = jobs.iter().find(|j| j.name == "cli_added_job").unwrap();
        assert_eq!(
            job.provenance, "user_added",
            "insert_job (the CLI write path) must stamp provenance = user_added, not the DB default"
        );
    }

    #[tokio::test]
    async fn insert_job_duplicate_returns_error() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();
        store
            .insert_job(
                "dup_job",
                "0 * * * * *",
                "custom",
                "periodic",
                None,
                "first",
            )
            .await
            .unwrap();
        let result = store
            .insert_job(
                "dup_job",
                "0 0 * * * *",
                "custom",
                "periodic",
                None,
                "second",
            )
            .await;
        assert!(
            matches!(result, Err(SchedulerError::DuplicateJob(ref n)) if n == "dup_job"),
            "expected DuplicateJob, got {result:?}"
        );
    }

    /// `list_recent_runs` orders by `last_run` descending with never-run jobs sorted last, and
    /// respects the `n` limit — verifies the SQL `ORDER BY last_run IS NULL, last_run DESC LIMIT ?`
    /// approach (#6115) matches the previous fetch-then-sort-in-Rust behavior.
    #[tokio::test]
    async fn list_recent_runs_orders_by_recency_with_nulls_last_and_respects_limit() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();

        let now = chrono::Utc::now();
        let older_ts = (now - chrono::Duration::days(200)).to_rfc3339();
        let newer_ts = (now - chrono::Duration::days(10)).to_rfc3339();

        store
            .upsert_job("older", "0 * * * * *", "health_check")
            .await
            .unwrap();
        store
            .record_run("older", &older_ts, "2026-01-02T00:00:00Z")
            .await
            .unwrap();

        store
            .upsert_job("newer", "0 * * * * *", "health_check")
            .await
            .unwrap();
        store
            .record_run("newer", &newer_ts, "2026-06-02T00:00:00Z")
            .await
            .unwrap();

        store
            .upsert_job("never_run", "0 * * * * *", "health_check")
            .await
            .unwrap();

        assert_eq!(store.count_active_jobs().await.unwrap(), 3);

        let recent = store.list_recent_runs(10).await.unwrap();
        let names: Vec<&str> = recent.iter().map(|r| r.name.as_str()).collect();
        assert_eq!(
            names,
            vec!["newer", "older", "never_run"],
            "list_recent_runs must order by last_run descending, never-run last"
        );
        assert_eq!(recent[0].last_run.as_deref(), Some(newer_ts.as_str()));
        assert_eq!(recent[1].last_run.as_deref(), Some(older_ts.as_str()));
        assert_eq!(recent[2].last_run, None);

        let limited = store.list_recent_runs(1).await.unwrap();
        assert_eq!(limited.len(), 1, "limit must be applied in SQL");
        assert_eq!(limited[0].name, "newer");
    }

    #[tokio::test]
    async fn list_jobs_full_includes_task_data() {
        let pool = test_pool().await;
        let store = JobStore::new(pool);
        store.init().await.unwrap();
        store
            .insert_job(
                "task_job",
                "0 * * * * *",
                "custom",
                "periodic",
                None,
                "my prompt",
            )
            .await
            .unwrap();
        let jobs = store.list_jobs_full().await.unwrap();
        let job = jobs.iter().find(|j| j.name == "task_job").unwrap();
        assert_eq!(job.task_data, "my prompt");
    }
}