zeph_scheduler/
store.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4use zeph_db::DbPool;
5#[cfg(any(feature = "sqlite", feature = "postgres"))]
6use zeph_db::sql;
7
8use crate::error::SchedulerError;
9use crate::task::CronExpr;
10
11/// A scheduled task row returned by [`JobStore::list_jobs`].
12///
13/// Replaces the previous `(String, String, String, String)` tuple to eliminate
14/// positional destructuring bugs. Fields map 1-to-1 to the SQL columns in the
15/// same order as the query: `name`, `kind`, `task_mode`, and the coalesced
16/// `next_run`.
17///
18/// # Examples
19///
20/// ```rust,no_run
21/// use zeph_scheduler::JobStore;
22///
23/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
24/// let store = JobStore::open("sqlite:scheduler.db").await?;
25/// store.init().await?;
26///
27/// for job in store.list_jobs().await? {
28///     println!("{}: {} ({}) → {}", job.name, job.kind, job.task_mode, job.next_run);
29/// }
30/// # Ok(())
31/// # }
32/// ```
33#[derive(Debug, Clone)]
34pub struct ScheduledTaskRecord {
35    /// Unique task name (primary key in the `scheduled_jobs` table).
36    pub name: String,
37    /// Serialised [`crate::TaskKind`] string (e.g. `"health_check"`).
38    pub kind: String,
39    /// Execution mode: `"periodic"` or `"oneshot"`.
40    pub task_mode: String,
41    /// Next scheduled run time as an ISO 8601 / RFC 3339 string.
42    ///
43    /// Falls back to `run_at` for one-shot jobs that have not yet computed a
44    /// `next_run`. Empty string when neither field is set.
45    pub next_run: String,
46}
47
48/// Full details for a scheduled task, returned by [`JobStore::list_jobs_full`].
49///
50/// Intended for display in the TUI or CLI task list. All string fields are UTF-8
51/// and come directly from the `scheduled_jobs` `SQLite` table.
52#[derive(Debug, Clone)]
53pub struct ScheduledTaskInfo {
54    /// Unique task name (primary key in the `scheduled_jobs` table).
55    pub name: String,
56    /// Serialised [`crate::TaskKind`] string (e.g. `"health_check"`).
57    pub kind: String,
58    /// Execution mode: `"periodic"` or `"oneshot"`.
59    pub task_mode: String,
60    /// Validated cron expression for periodic tasks; `None` for one-shot tasks or rows
61    /// whose stored expression is invalid (those are marked `status = "error"` in the DB).
62    pub cron_expr: Option<CronExpr>,
63    /// Next scheduled run time as an ISO 8601 / RFC 3339 string, or empty if unknown.
64    pub next_run: String,
65    /// Stored task prompt for custom tasks; empty for config-driven built-in tasks.
66    pub task_data: String,
67    /// Current job status: `"pending"`, `"completed"`, `"done"`, or `"error"`.
68    pub status: String,
69    /// RTW-A provenance tag: `"static"`, `"user_added"`, or `"external"`.
70    pub provenance: String,
71    /// Last recorded run time (RFC 3339), or `None` if the task has never run.
72    pub last_run: Option<String>,
73    /// Raw `run_at` column value (one-shot tasks only), independent of the `next_run`
74    /// coalescing applied to [`ScheduledTaskInfo::next_run`]. `Scheduler::init()` hydration
75    /// reads this field (rather than the coalesced `next_run`) to reconstruct one-shot rows
76    /// unambiguously (#6361).
77    pub run_at: Option<String>,
78}
79
80/// A single row returned by [`JobStore::list_recent_runs`], ordered by recency.
81#[derive(Debug, Clone)]
82pub struct RecentRunRecord {
83    /// Unique task name.
84    pub name: String,
85    /// Execution mode: `"periodic"` or `"oneshot"`.
86    pub task_mode: String,
87    /// Last recorded run time (RFC 3339), or `None` if the task has never run.
88    pub last_run: Option<String>,
89    /// Next scheduled run time as an ISO 8601 / RFC 3339 string, or empty if unknown.
90    pub next_run: String,
91    /// Current job status: `"pending"`, `"completed"`, `"done"`, or `"error"`.
92    pub status: String,
93}
94
95/// Persistent storage layer for scheduled jobs.
96///
97/// All job definitions and run history are stored in a `SQLite` database managed by
98/// `zeph-db` migrations. The `scheduled_jobs` table schema is defined in migration
99/// `051_scheduler_jobs.sql`.
100///
101/// # Examples
102///
103/// ```rust,no_run
104/// use zeph_scheduler::JobStore;
105///
106/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
107/// // Open from a file path.
108/// let store = JobStore::open("sqlite:scheduler.db").await?;
109/// store.init().await?;
110///
111/// // Query job list.
112/// let jobs = store.list_jobs().await?;
113/// for job in &jobs {
114///     println!("{}: {} ({}) → {}", job.name, job.kind, job.task_mode, job.next_run);
115/// }
116/// # Ok(())
117/// # }
118/// ```
119#[derive(Debug)]
120pub struct JobStore {
121    pool: DbPool,
122}
123
124impl JobStore {
125    /// Create a `JobStore` from an existing [`zeph_db::DbPool`].
126    ///
127    /// You must call [`JobStore::init`] before any other operation to ensure the
128    /// schema migrations have been applied.
129    #[must_use]
130    pub fn new(pool: DbPool) -> Self {
131        Self { pool }
132    }
133
134    /// Open (or create) a `JobStore` from a `SQLite` file path.
135    ///
136    /// # Errors
137    ///
138    /// Returns [`SchedulerError::Db`] if the connection cannot be established.
139    #[tracing::instrument(name = "sched.store.open", skip_all, fields(path = %path), err)]
140    pub async fn open(path: &str) -> Result<Self, SchedulerError> {
141        let pool = zeph_db::DbConfig {
142            url: path.to_string(),
143            pool_size: 5,
144        }
145        .connect()
146        .await?;
147        Ok(Self { pool })
148    }
149
150    /// Run all pending migrations on the underlying pool.
151    ///
152    /// Replaces the former inline `CREATE TABLE IF NOT EXISTS` DDL. The
153    /// `scheduled_jobs` schema is now managed by migration
154    /// `051_scheduler_jobs.sql` in `zeph-db`.
155    ///
156    /// # Errors
157    ///
158    /// Returns an error if any migration fails.
159    #[tracing::instrument(name = "sched.store.init", skip_all, err)]
160    pub async fn init(&self) -> Result<(), SchedulerError> {
161        zeph_db::run_migrations(&self.pool).await?;
162        Ok(())
163    }
164
165    /// Upsert a job definition.
166    ///
167    /// # Errors
168    ///
169    /// Returns an error if the SQL statement fails.
170    #[tracing::instrument(name = "sched.store.upsert_job", skip_all, fields(task = %name), err)]
171    pub async fn upsert_job(
172        &self,
173        name: &str,
174        cron_expr: &str,
175        kind: &str,
176    ) -> Result<(), SchedulerError> {
177        self.upsert_job_with_mode(name, cron_expr, kind, "periodic", None, "")
178            .await
179    }
180
181    /// Upsert a job definition with explicit `task_mode`, optional `run_at`, and `task_data`.
182    ///
183    /// # Errors
184    ///
185    /// Returns an error if the SQL statement fails.
186    #[tracing::instrument(name = "sched.store.upsert_job_with_mode", skip_all, fields(task = %name), err)]
187    pub async fn upsert_job_with_mode(
188        &self,
189        name: &str,
190        cron_expr: &str,
191        kind: &str,
192        task_mode: &str,
193        run_at: Option<&str>,
194        task_data: &str,
195    ) -> Result<(), SchedulerError> {
196        self.upsert_job_with_provenance(
197            name, cron_expr, kind, task_mode, run_at, task_data, "static",
198        )
199        .await
200    }
201
202    /// Upsert a job definition with explicit `task_mode`, `task_data`, and RTW-A `provenance`.
203    ///
204    /// # Errors
205    ///
206    /// Returns an error if the SQL statement fails.
207    #[allow(clippy::too_many_arguments)]
208    #[tracing::instrument(name = "sched.store.upsert_job_with_provenance", skip_all, fields(task = %name), err)]
209    pub async fn upsert_job_with_provenance(
210        &self,
211        name: &str,
212        cron_expr: &str,
213        kind: &str,
214        task_mode: &str,
215        run_at: Option<&str>,
216        task_data: &str,
217        provenance: &str,
218    ) -> Result<(), SchedulerError> {
219        zeph_db::query(sql!(
220            "INSERT INTO scheduled_jobs (name, cron_expr, kind, task_mode, run_at, task_data, provenance)
221             VALUES (?, ?, ?, ?, ?, ?, ?)
222             ON CONFLICT(name) DO UPDATE SET
223               cron_expr = excluded.cron_expr,
224               kind = excluded.kind,
225               task_mode = excluded.task_mode,
226               run_at = excluded.run_at,
227               task_data = excluded.task_data,
228               provenance = excluded.provenance"
229        ))
230        .bind(name)
231        .bind(cron_expr)
232        .bind(kind)
233        .bind(task_mode)
234        .bind(run_at)
235        .bind(task_data)
236        .bind(provenance)
237        .execute(&self.pool)
238        .await?;
239        Ok(())
240    }
241
242    /// Insert a new job. Returns [`SchedulerError::DuplicateJob`] if a job with this name exists.
243    ///
244    /// This is the CLI write path (`zeph schedule add`, both the cron and `--run-at` forms), so
245    /// the row is stored with [`crate::task::TaskProvenance::UserAdded`] provenance — reflecting
246    /// its genuine CLI/user origin — rather than relying on the `provenance` column's DB default.
247    /// `Scheduler::init()` hydration still force-downgrades out-of-process rows to
248    /// [`crate::task::TaskProvenance::External`] regardless of the stored value (#6114), so this
249    /// does not change runtime trust decisions; it keeps the stored label accurate for direct
250    /// inspection (e.g. a raw DB dump) — `zeph schedule list`/`show` do not currently surface
251    /// `provenance`.
252    ///
253    /// # Errors
254    ///
255    /// Returns [`SchedulerError::DuplicateJob`] on unique constraint violation,
256    /// or [`SchedulerError::Database`] on other SQL errors.
257    #[tracing::instrument(name = "sched.store.insert_job", skip_all, fields(task = %name), err)]
258    pub async fn insert_job(
259        &self,
260        name: &str,
261        cron_expr: &str,
262        kind: &str,
263        task_mode: &str,
264        run_at: Option<&str>,
265        task_data: &str,
266    ) -> Result<(), SchedulerError> {
267        let result = zeph_db::query(sql!(
268            "INSERT INTO scheduled_jobs (name, cron_expr, kind, task_mode, run_at, task_data, provenance)
269             VALUES (?, ?, ?, ?, ?, ?, ?)"
270        ))
271        .bind(name)
272        .bind(cron_expr)
273        .bind(kind)
274        .bind(task_mode)
275        .bind(run_at)
276        .bind(task_data)
277        .bind(crate::task::TaskProvenance::UserAdded.as_str())
278        .execute(&self.pool)
279        .await;
280        match result {
281            Ok(_) => Ok(()),
282            Err(zeph_db::SqlxError::Database(db_err))
283                if db_err.message().contains("UNIQUE constraint failed")
284                    || db_err.code().as_deref() == Some("23505") =>
285            {
286                Err(SchedulerError::DuplicateJob(name.to_string()))
287            }
288            Err(e) => Err(SchedulerError::Database(e)),
289        }
290    }
291
292    /// Record a job execution and persist the next scheduled run time.
293    ///
294    /// # Errors
295    ///
296    /// Returns an error if the SQL statement fails.
297    #[tracing::instrument(name = "sched.store.record_run", skip_all, fields(task = %name), err)]
298    pub async fn record_run(
299        &self,
300        name: &str,
301        timestamp: &str,
302        next_run: &str,
303    ) -> Result<(), SchedulerError> {
304        zeph_db::query(
305            sql!("UPDATE scheduled_jobs SET last_run = ?, next_run = ?, status = 'completed' WHERE name = ?"),
306        )
307        .bind(timestamp)
308        .bind(next_run)
309        .bind(name)
310        .execute(&self.pool)
311        .await?;
312        Ok(())
313    }
314
315    /// Mark a one-shot job as done.
316    ///
317    /// # Errors
318    ///
319    /// Returns an error if the SQL statement fails.
320    #[tracing::instrument(name = "sched.store.mark_done", skip_all, fields(task = %name), err)]
321    pub async fn mark_done(&self, name: &str) -> Result<(), SchedulerError> {
322        zeph_db::query(sql!(
323            "UPDATE scheduled_jobs SET status = 'done', last_run = CURRENT_TIMESTAMP WHERE name = ?"
324        ))
325        .bind(name)
326        .execute(&self.pool)
327        .await?;
328        Ok(())
329    }
330
331    /// Mark a job as permanently errored (e.g. invalid cron expression on hydration).
332    ///
333    /// The job remains visible in [`JobStore::list_jobs_full`] with `status = "error"` so
334    /// operators can identify it via `zeph scheduler list` without reading debug logs.
335    ///
336    /// # Errors
337    ///
338    /// Returns an error if the SQL statement fails.
339    #[tracing::instrument(name = "sched.store.mark_error", skip_all, fields(task = %name), err)]
340    pub async fn mark_error(&self, name: &str) -> Result<(), SchedulerError> {
341        zeph_db::query(sql!(
342            "UPDATE scheduled_jobs SET status = 'error' WHERE name = ?"
343        ))
344        .bind(name)
345        .execute(&self.pool)
346        .await?;
347        Ok(())
348    }
349
350    /// Delete a job by name.
351    ///
352    /// # Errors
353    ///
354    /// Returns an error if the SQL statement fails.
355    #[tracing::instrument(name = "sched.store.delete_job", skip_all, fields(task = %name), err)]
356    pub async fn delete_job(&self, name: &str) -> Result<bool, SchedulerError> {
357        let result = zeph_db::query(sql!("DELETE FROM scheduled_jobs WHERE name = ?"))
358            .bind(name)
359            .execute(&self.pool)
360            .await?;
361        Ok(result.rows_affected() > 0)
362    }
363
364    /// Check if a job with the given name exists.
365    ///
366    /// # Errors
367    ///
368    /// Returns an error if the SQL query fails.
369    #[tracing::instrument(name = "sched.store.job_exists", skip_all, fields(task = %name), err)]
370    pub async fn job_exists(&self, name: &str) -> Result<bool, SchedulerError> {
371        let row: Option<(i64,)> =
372            zeph_db::query_as(sql!("SELECT 1 FROM scheduled_jobs WHERE name = ?"))
373                .bind(name)
374                .fetch_optional(&self.pool)
375                .await?;
376        Ok(row.is_some())
377    }
378
379    /// Persist the next scheduled run time for a job (used during init).
380    ///
381    /// # Errors
382    ///
383    /// Returns an error if the SQL statement fails.
384    #[tracing::instrument(name = "sched.store.set_next_run", skip_all, fields(task = %name), err)]
385    pub async fn set_next_run(&self, name: &str, next_run: &str) -> Result<(), SchedulerError> {
386        zeph_db::query(sql!(
387            "UPDATE scheduled_jobs SET next_run = ? WHERE name = ?"
388        ))
389        .bind(next_run)
390        .bind(name)
391        .execute(&self.pool)
392        .await?;
393        Ok(())
394    }
395
396    /// Get the persisted next run timestamp for a job.
397    ///
398    /// # Errors
399    ///
400    /// Returns an error if the SQL query fails.
401    #[tracing::instrument(name = "sched.store.get_next_run", skip_all, fields(task = %name), err)]
402    pub async fn get_next_run(&self, name: &str) -> Result<Option<String>, SchedulerError> {
403        let row: Option<(Option<String>,)> =
404            zeph_db::query_as(sql!("SELECT next_run FROM scheduled_jobs WHERE name = ?"))
405                .bind(name)
406                .fetch_optional(&self.pool)
407                .await?;
408        Ok(row.and_then(|r| r.0))
409    }
410
411    /// List all active (non-done) jobs.
412    ///
413    /// Returns a [`ScheduledTaskRecord`] per active job, ordered by name.
414    /// One-shot jobs without a computed `next_run` fall back to their `run_at` value;
415    /// if neither is set the field is an empty string.
416    ///
417    /// # Errors
418    ///
419    /// Returns an error if the SQL query fails.
420    #[tracing::instrument(name = "sched.store.list_jobs", skip_all, err)]
421    pub async fn list_jobs(&self) -> Result<Vec<ScheduledTaskRecord>, SchedulerError> {
422        let rows: Vec<(String, String, String, Option<String>)> = zeph_db::query_as(
423            sql!("SELECT name, kind, task_mode, COALESCE(next_run, run_at) FROM scheduled_jobs WHERE status != 'done' ORDER BY name"),
424        )
425        .fetch_all(&self.pool)
426        .await?;
427        Ok(rows
428            .into_iter()
429            .map(|(name, kind, task_mode, next_run)| ScheduledTaskRecord {
430                name,
431                kind,
432                task_mode,
433                next_run: next_run.unwrap_or_default(),
434            })
435            .collect())
436    }
437
438    /// List all active (non-done) jobs with full details for display.
439    ///
440    /// The `cron_expr` field of each returned [`ScheduledTaskInfo`] is `Some` for periodic tasks
441    /// with a valid expression and `None` for one-shot tasks or rows whose stored expression
442    /// failed validation. Invalid rows are not filtered out — callers can check `status` to
443    /// distinguish error rows.
444    ///
445    /// # Errors
446    ///
447    /// Returns an error if the SQL query fails.
448    #[tracing::instrument(name = "sched.store.list_jobs_full", skip_all, err)]
449    pub async fn list_jobs_full(&self) -> Result<Vec<ScheduledTaskInfo>, SchedulerError> {
450        #[allow(clippy::type_complexity)]
451        let rows: Vec<(
452            String,
453            String,
454            String,
455            String,
456            Option<String>,
457            String,
458            String,
459            String,
460            Option<String>,
461            Option<String>,
462        )> = zeph_db::query_as(sql!(
463            "SELECT name, kind, task_mode, cron_expr, COALESCE(next_run, run_at), task_data, status, provenance, last_run, run_at \
464             FROM scheduled_jobs WHERE status != 'done' ORDER BY name"
465        ))
466        .fetch_all(&self.pool)
467        .await?;
468        Ok(rows
469            .into_iter()
470            .map(
471                |(
472                    name,
473                    kind,
474                    task_mode,
475                    raw_cron,
476                    next_run,
477                    task_data,
478                    status,
479                    provenance,
480                    last_run,
481                    run_at,
482                )| {
483                    // Empty string = oneshot task (no cron). Non-empty = validate eagerly.
484                    let cron_expr = if raw_cron.is_empty() {
485                        None
486                    } else {
487                        match CronExpr::try_from(raw_cron.as_str()) {
488                            Ok(expr) => Some(expr),
489                            Err(e) => {
490                                tracing::warn!(
491                                    task = %name,
492                                    cron_expr = %raw_cron,
493                                    "invalid cron expression in DB row: {e}"
494                                );
495                                None
496                            }
497                        }
498                    };
499                    ScheduledTaskInfo {
500                        name,
501                        kind,
502                        task_mode,
503                        cron_expr,
504                        next_run: next_run.unwrap_or_default(),
505                        task_data,
506                        status,
507                        provenance,
508                        last_run,
509                        run_at,
510                    }
511                },
512            )
513            .collect())
514    }
515
516    /// Count active (non-done) jobs.
517    ///
518    /// # Errors
519    ///
520    /// Returns an error if the SQL query fails.
521    #[tracing::instrument(name = "sched.store.count_active_jobs", skip_all, err)]
522    pub async fn count_active_jobs(&self) -> Result<usize, SchedulerError> {
523        let (count,): (i64,) = zeph_db::query_as(sql!(
524            "SELECT COUNT(*) FROM scheduled_jobs WHERE status != 'done'"
525        ))
526        .fetch_one(&self.pool)
527        .await?;
528        Ok(usize::try_from(count).unwrap_or(0))
529    }
530
531    /// List the `n` most recently run (non-done) jobs, ordered by `last_run` descending.
532    ///
533    /// Never-run jobs (`last_run IS NULL`) sort last. Ordering and truncation are pushed into
534    /// SQL via `ORDER BY last_run IS NULL, last_run DESC LIMIT ?` rather than fetched-then-sorted
535    /// in Rust: `last_run IS NULL` evaluates to a sortable `0`/`1` (`SQLite`) or `false`/`true`
536    /// (`PostgreSQL`) value, giving "`NULLS LAST`" semantics on both backends without relying on
537    /// `PostgreSQL`-only `NULLS LAST` syntax, which `SQLite` does not support.
538    ///
539    /// # Errors
540    ///
541    /// Returns an error if the SQL query fails.
542    #[tracing::instrument(name = "sched.store.list_recent_runs", skip_all, err)]
543    pub async fn list_recent_runs(&self, n: usize) -> Result<Vec<RecentRunRecord>, SchedulerError> {
544        #[allow(clippy::type_complexity)]
545        let rows: Vec<(String, String, Option<String>, Option<String>, String)> =
546            zeph_db::query_as(sql!(
547                "SELECT name, task_mode, last_run, COALESCE(next_run, run_at), status \
548                 FROM scheduled_jobs WHERE status != 'done' \
549                 ORDER BY last_run IS NULL, last_run DESC LIMIT ?"
550            ))
551            .bind(i64::try_from(n).unwrap_or(i64::MAX))
552            .fetch_all(&self.pool)
553            .await?;
554        Ok(rows
555            .into_iter()
556            .map(
557                |(name, task_mode, last_run, next_run, status)| RecentRunRecord {
558                    name,
559                    task_mode,
560                    last_run,
561                    next_run: next_run.unwrap_or_default(),
562                    status,
563                },
564            )
565            .collect())
566    }
567
568    /// Returns a reference to the underlying connection pool.
569    ///
570    /// Primarily used in tests that need to execute raw SQL against the same database.
571    #[must_use]
572    pub fn pool(&self) -> &DbPool {
573        &self.pool
574    }
575}
576
577// Every test in this module opens a real connection pool via the `:memory:` sentinel, which is
578// SQLite-specific: under `--features postgres`, `DbConfig::connect()` takes cfg-priority and routes
579// `:memory:` into `connect_postgres`, which fails to parse it as a Postgres URL. See #5608.
580#[cfg(all(test, feature = "sqlite"))]
581mod tests {
582    use super::*;
583
584    async fn test_pool() -> DbPool {
585        zeph_db::DbConfig {
586            url: ":memory:".to_string(),
587            pool_size: 5,
588        }
589        .connect()
590        .await
591        .unwrap()
592    }
593
594    #[tokio::test]
595    async fn init_creates_table() {
596        let pool = test_pool().await;
597        let store = JobStore::new(pool);
598        assert!(store.init().await.is_ok());
599    }
600
601    #[tokio::test]
602    async fn upsert_and_query() {
603        let pool = test_pool().await;
604        let store = JobStore::new(pool);
605        store.init().await.unwrap();
606
607        store
608            .upsert_job("test_job", "0 * * * * *", "health_check")
609            .await
610            .unwrap();
611        assert!(store.get_next_run("test_job").await.unwrap().is_none());
612
613        store
614            .record_run("test_job", "2026-01-01T00:00:00Z", "2026-01-01T00:01:00Z")
615            .await
616            .unwrap();
617        assert_eq!(
618            store.get_next_run("test_job").await.unwrap().as_deref(),
619            Some("2026-01-01T00:01:00Z")
620        );
621    }
622
623    #[tokio::test]
624    async fn upsert_updates_existing() {
625        let pool = test_pool().await;
626        let store = JobStore::new(pool);
627        store.init().await.unwrap();
628
629        store
630            .upsert_job("job1", "0 * * * * *", "health_check")
631            .await
632            .unwrap();
633        store
634            .upsert_job("job1", "0 0 * * * *", "memory_cleanup")
635            .await
636            .unwrap();
637
638        let row: (String,) =
639            zeph_db::query_as(sql!("SELECT kind FROM scheduled_jobs WHERE name = 'job1'"))
640                .fetch_one(store.pool())
641                .await
642                .unwrap();
643        assert_eq!(row.0, "memory_cleanup");
644    }
645
646    #[tokio::test]
647    async fn next_run_nonexistent_job() {
648        let pool = test_pool().await;
649        let store = JobStore::new(pool);
650        store.init().await.unwrap();
651        assert!(store.get_next_run("no_such_job").await.unwrap().is_none());
652    }
653
654    #[tokio::test]
655    async fn job_exists_returns_true_for_existing() {
656        let pool = test_pool().await;
657        let store = JobStore::new(pool);
658        store.init().await.unwrap();
659        store
660            .upsert_job("exists_job", "0 * * * * *", "health_check")
661            .await
662            .unwrap();
663        assert!(store.job_exists("exists_job").await.unwrap());
664        assert!(!store.job_exists("missing").await.unwrap());
665    }
666
667    #[tokio::test]
668    async fn delete_job_removes_row() {
669        let pool = test_pool().await;
670        let store = JobStore::new(pool);
671        store.init().await.unwrap();
672        store
673            .upsert_job("del_job", "0 * * * * *", "health_check")
674            .await
675            .unwrap();
676        assert!(store.delete_job("del_job").await.unwrap());
677        assert!(!store.job_exists("del_job").await.unwrap());
678        assert!(!store.delete_job("del_job").await.unwrap());
679    }
680
681    #[tokio::test]
682    async fn mark_done_sets_status() {
683        let pool = test_pool().await;
684        let store = JobStore::new(pool);
685        store.init().await.unwrap();
686        store
687            .upsert_job_with_mode(
688                "os_job",
689                "",
690                "health_check",
691                "oneshot",
692                Some("2026-01-01T01:00:00Z"),
693                "",
694            )
695            .await
696            .unwrap();
697        store.mark_done("os_job").await.unwrap();
698        let row: (String,) = zeph_db::query_as(sql!(
699            "SELECT status FROM scheduled_jobs WHERE name = 'os_job'"
700        ))
701        .fetch_one(store.pool())
702        .await
703        .unwrap();
704        assert_eq!(row.0, "done");
705    }
706
707    #[tokio::test]
708    async fn list_jobs_excludes_done_jobs() {
709        let pool = test_pool().await;
710        let store = JobStore::new(pool);
711        store.init().await.unwrap();
712        store
713            .upsert_job_with_mode(
714                "done_job",
715                "",
716                "health_check",
717                "oneshot",
718                Some("2026-01-01T01:00:00Z"),
719                "",
720            )
721            .await
722            .unwrap();
723        store.mark_done("done_job").await.unwrap();
724        let jobs = store.list_jobs().await.unwrap();
725        assert!(
726            jobs.iter().all(|j| j.name != "done_job"),
727            "list_jobs must not return done jobs"
728        );
729    }
730
731    #[tokio::test]
732    async fn list_jobs_uses_run_at_for_oneshot_when_next_run_is_null() {
733        let pool = test_pool().await;
734        let store = JobStore::new(pool);
735        store.init().await.unwrap();
736        store
737            .upsert_job_with_mode(
738                "oneshot_job",
739                "",
740                "custom",
741                "oneshot",
742                Some("2026-06-01T10:00:00Z"),
743                "",
744            )
745            .await
746            .unwrap();
747        let jobs = store.list_jobs().await.unwrap();
748        let job = jobs.iter().find(|j| j.name == "oneshot_job").unwrap();
749        assert_eq!(
750            job.next_run, "2026-06-01T10:00:00Z",
751            "run_at must be shown as next_run for oneshot jobs"
752        );
753    }
754
755    #[tokio::test]
756    async fn list_jobs_full_returns_correct_fields() {
757        let pool = test_pool().await;
758        let store = JobStore::new(pool);
759        store.init().await.unwrap();
760        store
761            .upsert_job("periodic_job", "0 0 3 * * *", "memory_cleanup")
762            .await
763            .unwrap();
764        store
765            .upsert_job_with_mode(
766                "oneshot_job",
767                "",
768                "custom",
769                "oneshot",
770                Some("2030-01-01T10:00:00Z"),
771                "",
772            )
773            .await
774            .unwrap();
775
776        store
777            .record_run(
778                "periodic_job",
779                "2026-01-01T00:00:00Z",
780                "2026-01-02T00:00:00Z",
781            )
782            .await
783            .unwrap();
784
785        let jobs = store.list_jobs_full().await.unwrap();
786        assert_eq!(jobs.len(), 2);
787
788        let periodic = jobs.iter().find(|j| j.name == "periodic_job").unwrap();
789        assert_eq!(periodic.kind, "memory_cleanup");
790        assert_eq!(periodic.task_mode, "periodic");
791        assert_eq!(
792            periodic.cron_expr.as_ref().map(CronExpr::as_str),
793            Some("0 0 3 * * *")
794        );
795        assert_eq!(
796            periodic.last_run.as_deref(),
797            Some("2026-01-01T00:00:00Z"),
798            "last_run must reflect the timestamp recorded by record_run"
799        );
800
801        let oneshot = jobs.iter().find(|j| j.name == "oneshot_job").unwrap();
802        assert_eq!(oneshot.kind, "custom");
803        assert_eq!(oneshot.task_mode, "oneshot");
804        assert!(oneshot.cron_expr.is_none());
805        assert_eq!(oneshot.next_run, "2030-01-01T10:00:00Z");
806        assert_eq!(
807            oneshot.run_at.as_deref(),
808            Some("2030-01-01T10:00:00Z"),
809            "run_at must be the raw column value, independent of the next_run coalescing (#6361 C1)"
810        );
811        assert_eq!(
812            oneshot.last_run, None,
813            "a task that has never run must report last_run = None"
814        );
815    }
816
817    #[tokio::test]
818    async fn list_jobs_full_excludes_done_jobs() {
819        let pool = test_pool().await;
820        let store = JobStore::new(pool);
821        store.init().await.unwrap();
822        store
823            .upsert_job_with_mode(
824                "done_job",
825                "",
826                "custom",
827                "oneshot",
828                Some("2026-01-01T01:00:00Z"),
829                "",
830            )
831            .await
832            .unwrap();
833        store.mark_done("done_job").await.unwrap();
834        let jobs = store.list_jobs_full().await.unwrap();
835        assert!(jobs.iter().all(|j| j.name != "done_job"));
836    }
837
838    #[tokio::test]
839    async fn duplicate_name_detected() {
840        let pool = test_pool().await;
841        let store = JobStore::new(pool);
842        store.init().await.unwrap();
843        store
844            .upsert_job("dup", "0 * * * * *", "health_check")
845            .await
846            .unwrap();
847        assert!(store.job_exists("dup").await.unwrap());
848    }
849
850    #[tokio::test]
851    async fn insert_job_success() {
852        let pool = test_pool().await;
853        let store = JobStore::new(pool);
854        store.init().await.unwrap();
855        store
856            .insert_job(
857                "new_job",
858                "0 * * * * *",
859                "custom",
860                "periodic",
861                None,
862                "run daily report",
863            )
864            .await
865            .unwrap();
866        assert!(store.job_exists("new_job").await.unwrap());
867    }
868
869    /// `insert_job` is the CLI write path (`zeph schedule add`), so it must stamp
870    /// `provenance = "user_added"` rather than falling through to the `scheduled_jobs.provenance`
871    /// column's schema default. Regression test for #6442.
872    #[tokio::test]
873    async fn insert_job_stores_user_added_provenance() {
874        let pool = test_pool().await;
875        let store = JobStore::new(pool);
876        store.init().await.unwrap();
877        store
878            .insert_job(
879                "cli_added_job",
880                "0 * * * * *",
881                "custom",
882                "periodic",
883                None,
884                "run daily report",
885            )
886            .await
887            .unwrap();
888        let jobs = store.list_jobs_full().await.unwrap();
889        let job = jobs.iter().find(|j| j.name == "cli_added_job").unwrap();
890        assert_eq!(
891            job.provenance, "user_added",
892            "insert_job (the CLI write path) must stamp provenance = user_added, not the DB default"
893        );
894    }
895
896    #[tokio::test]
897    async fn insert_job_duplicate_returns_error() {
898        let pool = test_pool().await;
899        let store = JobStore::new(pool);
900        store.init().await.unwrap();
901        store
902            .insert_job(
903                "dup_job",
904                "0 * * * * *",
905                "custom",
906                "periodic",
907                None,
908                "first",
909            )
910            .await
911            .unwrap();
912        let result = store
913            .insert_job(
914                "dup_job",
915                "0 0 * * * *",
916                "custom",
917                "periodic",
918                None,
919                "second",
920            )
921            .await;
922        assert!(
923            matches!(result, Err(SchedulerError::DuplicateJob(ref n)) if n == "dup_job"),
924            "expected DuplicateJob, got {result:?}"
925        );
926    }
927
928    /// `list_recent_runs` orders by `last_run` descending with never-run jobs sorted last, and
929    /// respects the `n` limit — verifies the SQL `ORDER BY last_run IS NULL, last_run DESC LIMIT ?`
930    /// approach (#6115) matches the previous fetch-then-sort-in-Rust behavior.
931    #[tokio::test]
932    async fn list_recent_runs_orders_by_recency_with_nulls_last_and_respects_limit() {
933        let pool = test_pool().await;
934        let store = JobStore::new(pool);
935        store.init().await.unwrap();
936
937        let now = chrono::Utc::now();
938        let older_ts = (now - chrono::Duration::days(200)).to_rfc3339();
939        let newer_ts = (now - chrono::Duration::days(10)).to_rfc3339();
940
941        store
942            .upsert_job("older", "0 * * * * *", "health_check")
943            .await
944            .unwrap();
945        store
946            .record_run("older", &older_ts, "2026-01-02T00:00:00Z")
947            .await
948            .unwrap();
949
950        store
951            .upsert_job("newer", "0 * * * * *", "health_check")
952            .await
953            .unwrap();
954        store
955            .record_run("newer", &newer_ts, "2026-06-02T00:00:00Z")
956            .await
957            .unwrap();
958
959        store
960            .upsert_job("never_run", "0 * * * * *", "health_check")
961            .await
962            .unwrap();
963
964        assert_eq!(store.count_active_jobs().await.unwrap(), 3);
965
966        let recent = store.list_recent_runs(10).await.unwrap();
967        let names: Vec<&str> = recent.iter().map(|r| r.name.as_str()).collect();
968        assert_eq!(
969            names,
970            vec!["newer", "older", "never_run"],
971            "list_recent_runs must order by last_run descending, never-run last"
972        );
973        assert_eq!(recent[0].last_run.as_deref(), Some(newer_ts.as_str()));
974        assert_eq!(recent[1].last_run.as_deref(), Some(older_ts.as_str()));
975        assert_eq!(recent[2].last_run, None);
976
977        let limited = store.list_recent_runs(1).await.unwrap();
978        assert_eq!(limited.len(), 1, "limit must be applied in SQL");
979        assert_eq!(limited[0].name, "newer");
980    }
981
982    #[tokio::test]
983    async fn list_jobs_full_includes_task_data() {
984        let pool = test_pool().await;
985        let store = JobStore::new(pool);
986        store.init().await.unwrap();
987        store
988            .insert_job(
989                "task_job",
990                "0 * * * * *",
991                "custom",
992                "periodic",
993                None,
994                "my prompt",
995            )
996            .await
997            .unwrap();
998        let jobs = store.list_jobs_full().await.unwrap();
999        let job = jobs.iter().find(|j| j.name == "task_job").unwrap();
1000        assert_eq!(job.task_data, "my prompt");
1001    }
1002}
zeph_scheduler/store.rs

zeph_scheduler/
store.rs
