feat: periodic task CRUD (list/delete/pause/resume)

crates/taskito-core/src/storage/mod.rs

@@ -306,6 +306,21 @@ macro_rules! impl_storage {
             ) -> $crate::error::Result<()> {
                 self.update_periodic_schedule(name, last_run, next_run)
             }
+            fn list_periodic(
+                &self,
+            ) -> $crate::error::Result<Vec<$crate::storage::models::PeriodicTaskRow>> {
+                self.list_periodic()
+            }
+            fn delete_periodic(&self, name: &str) -> $crate::error::Result<bool> {
+                self.delete_periodic(name)
+            }
+            fn set_periodic_enabled(
+                &self,
+                name: &str,
+                enabled: bool,
+            ) -> $crate::error::Result<bool> {
+                self.set_periodic_enabled(name, enabled)
+            }
             fn record_metric(
                 &self,
                 task_name: &str,
@@ -800,6 +815,15 @@ impl Storage for StorageBackend {
     fn update_periodic_schedule(&self, name: &str, last_run: i64, next_run: i64) -> Result<()> {
         delegate!(self, update_periodic_schedule, name, last_run, next_run)
     }
+    fn list_periodic(&self) -> Result<Vec<models::PeriodicTaskRow>> {
+        delegate!(self, list_periodic)
+    }
+    fn delete_periodic(&self, name: &str) -> Result<bool> {
+        delegate!(self, delete_periodic, name)
+    }
+    fn set_periodic_enabled(&self, name: &str, enabled: bool) -> Result<bool> {
+        delegate!(self, set_periodic_enabled, name, enabled)
+    }
     fn record_metric(
         &self,
         task_name: &str,

crates/taskito-core/src/storage/postgres/periodic.rs

@@ -46,4 +46,36 @@ impl PostgresStorage {
 
         Ok(())
     }
+
+    /// List all registered periodic tasks, enabled or paused.
+    pub fn list_periodic(&self) -> Result<Vec<PeriodicTaskRow>> {
+        let mut conn = self.conn()?;
+
+        let rows = periodic_tasks::table
+            .select(PeriodicTaskRow::as_select())
+            .load(&mut conn)?;
+
+        Ok(rows)
+    }
+
+    /// Remove a periodic task. Returns false if no task had that name.
+    pub fn delete_periodic(&self, name: &str) -> Result<bool> {
+        let mut conn = self.conn()?;
+
+        let affected = diesel::delete(periodic_tasks::table.find(name)).execute(&mut conn)?;
+
+        Ok(affected > 0)
+    }
+
+    /// Pause (false) or resume (true) a periodic task by toggling `enabled`.
+    /// Returns false if no task had that name.
+    pub fn set_periodic_enabled(&self, name: &str, enabled: bool) -> Result<bool> {
+        let mut conn = self.conn()?;
+
+        let affected = diesel::update(periodic_tasks::table.find(name))
+            .set(periodic_tasks::enabled.eq(enabled))
+            .execute(&mut conn)?;
+
+        Ok(affected > 0)
+    }
 }

crates/taskito-core/src/storage/redis_backend/periodic.rs

@@ -115,4 +115,92 @@ impl RedisStorage {
 
         Ok(())
     }
+
+    /// List all registered periodic tasks, enabled or paused. Scans the
+    /// `periodic:*` keyspace (like `list_dead`/`list_workers`) so there is no
+    /// secondary index to keep consistent — tasks registered by any version are
+    /// listed.
+    pub fn list_periodic(&self) -> Result<Vec<PeriodicTaskRow>> {
+        let mut conn = self.conn()?;
+        let pattern = self.key(&["periodic", "*"]);
+        // The due sorted-set shares the `periodic:` namespace but is not a task
+        // JSON blob, so skip it (a `GET` on it would be a WRONGTYPE error).
+        let due_key = self.key(&["periodic", "due"]);
+
+        let mut rows = Vec::new();
+        let mut cursor: u64 = 0;
+        loop {
+            let (next_cursor, keys): (u64, Vec<String>) = redis::cmd("SCAN")
+                .arg(cursor)
+                .arg("MATCH")
+                .arg(&pattern)
+                .arg("COUNT")
+                .arg(100)
+                .query(&mut conn)
+                .map_err(map_err)?;
+
+            for key in keys {
+                if key == due_key {
+                    continue;
+                }
+                let data: Option<String> = conn.get(&key).map_err(map_err)?;
+                if let Some(d) = data {
+                    let entry: PeriodicEntry =
+                        serde_json::from_str(&d).map_err(|e| QueueError::Other(e.to_string()))?;
+                    rows.push(PeriodicTaskRow::from(entry));
+                }
+            }
+
+            cursor = next_cursor;
+            if cursor == 0 {
+                break;
+            }
+        }
+
+        Ok(rows)
+    }
+
+    /// Remove a periodic task. Returns false if no task had that name.
+    pub fn delete_periodic(&self, name: &str) -> Result<bool> {
+        let mut conn = self.conn()?;
+        let pkey = self.key(&["periodic", name]);
+        let due_key = self.key(&["periodic", "due"]);
+
+        let pipe = &mut redis::pipe();
+        pipe.del(&pkey);
+        pipe.zrem(&due_key, name);
+        let (deleted, _zrem): (i64, i64) = pipe.query(&mut conn).map_err(map_err)?;
+
+        Ok(deleted > 0)
+    }
+
+    /// Pause (false) or resume (true) a periodic task by toggling `enabled`.
+    /// Returns false if no task had that name.
+    pub fn set_periodic_enabled(&self, name: &str, enabled: bool) -> Result<bool> {
+        let mut conn = self.conn()?;
+        let pkey = self.key(&["periodic", name]);
+
+        let data: Option<String> = conn.get(&pkey).map_err(map_err)?;
+        let Some(d) = data else {
+            return Ok(false);
+        };
+
+        let mut entry: PeriodicEntry =
+            serde_json::from_str(&d).map_err(|e| QueueError::Other(e.to_string()))?;
+        entry.enabled = enabled;
+
+        let json = serde_json::to_string(&entry).map_err(|e| QueueError::Other(e.to_string()))?;
+        let due_key = self.key(&["periodic", "due"]);
+
+        let pipe = &mut redis::pipe();
+        pipe.set(&pkey, &json);
+        if enabled {
+            pipe.zadd(&due_key, name, entry.next_run as f64);
+        } else {
+            pipe.zrem(&due_key, name);
+        }
+        pipe.query::<()>(&mut conn).map_err(map_err)?;
+
+        Ok(true)
+    }
 }

crates/taskito-core/src/storage/sqlite/periodic.rs

@@ -43,4 +43,36 @@ impl SqliteStorage {
 
         Ok(())
     }
+
+    /// List all registered periodic tasks, enabled or paused.
+    pub fn list_periodic(&self) -> Result<Vec<PeriodicTaskRow>> {
+        let mut conn = self.conn()?;
+
+        let rows = periodic_tasks::table
+            .select(PeriodicTaskRow::as_select())
+            .load(&mut conn)?;
+
+        Ok(rows)
+    }
+
+    /// Remove a periodic task. Returns false if no task had that name.
+    pub fn delete_periodic(&self, name: &str) -> Result<bool> {
+        let mut conn = self.conn()?;
+
+        let affected = diesel::delete(periodic_tasks::table.find(name)).execute(&mut conn)?;
+
+        Ok(affected > 0)
+    }
+
+    /// Pause (false) or resume (true) a periodic task by toggling `enabled`.
+    /// Returns false if no task had that name.
+    pub fn set_periodic_enabled(&self, name: &str, enabled: bool) -> Result<bool> {
+        let mut conn = self.conn()?;
+
+        let affected = diesel::update(periodic_tasks::table.find(name))
+            .set(periodic_tasks::enabled.eq(enabled))
+            .execute(&mut conn)?;
+
+        Ok(affected > 0)
+    }
 }

crates/taskito-core/src/storage/sqlite/tests.rs

@@ -1241,3 +1241,68 @@ fn test_dlq_retry_count_round_trip() {
     let dead2 = storage.list_dead(10, 0).unwrap();
     assert_eq!(dead2[0].dlq_retry_count, 1);
 }
+
+fn make_periodic(name: &str, next_run: i64) -> crate::storage::models::NewPeriodicTaskRow<'_> {
+    crate::storage::models::NewPeriodicTaskRow {
+        name,
+        task_name: "periodic_task",
+        cron_expr: "* * * * *",
+        args: None,
+        kwargs: None,
+        queue: "default",
+        enabled: true,
+        next_run,
+        timezone: None,
+    }
+}
+
+#[test]
+fn test_periodic_pause_resume_and_delete() {
+    let storage = test_storage();
+    let now = now_millis();
+    let past = now - 1000;
+
+    storage
+        .register_periodic(&make_periodic("alpha", past))
+        .unwrap();
+    storage
+        .register_periodic(&make_periodic("beta", past))
+        .unwrap();
+
+    // Both registered tasks are listed.
+    let listed = storage.list_periodic().unwrap();
+    assert_eq!(listed.len(), 2);
+    assert!(listed.iter().all(|row| row.enabled));
+
+    // Both are due and fire.
+    assert_eq!(storage.get_due_periodic(now).unwrap().len(), 2);
+
+    // Pausing "alpha" toggles enabled off and stops it firing.
+    assert!(storage.set_periodic_enabled("alpha", false).unwrap());
+    let due_names: Vec<String> = storage
+        .get_due_periodic(now)
+        .unwrap()
+        .into_iter()
+        .map(|row| row.name)
+        .collect();
+    assert_eq!(due_names, vec!["beta".to_string()]);
+
+    // But it is still listed (paused, not removed).
+    assert_eq!(storage.list_periodic().unwrap().len(), 2);
+
+    // Resuming brings it back into the due set.
+    assert!(storage.set_periodic_enabled("alpha", true).unwrap());
+    assert_eq!(storage.get_due_periodic(now).unwrap().len(), 2);
+
+    // Deleting removes it from the listing.
+    assert!(storage.delete_periodic("alpha").unwrap());
+    let remaining = storage.list_periodic().unwrap();
+    assert_eq!(remaining.len(), 1);
+    assert_eq!(remaining[0].name, "beta");
+
+    // A second delete reports nothing removed.
+    assert!(!storage.delete_periodic("alpha").unwrap());
+
+    // Toggling an unknown task reports nothing changed.
+    assert!(!storage.set_periodic_enabled("ghost", false).unwrap());
+}

crates/taskito-core/src/storage/traits.rs

@@ -102,6 +102,13 @@ pub trait Storage: Send + Sync + Clone {
     fn register_periodic(&self, task: &NewPeriodicTaskRow) -> Result<()>;
     fn get_due_periodic(&self, now: i64) -> Result<Vec<PeriodicTaskRow>>;
     fn update_periodic_schedule(&self, name: &str, last_run: i64, next_run: i64) -> Result<()>;
+    /// All registered periodic tasks, enabled or paused.
+    fn list_periodic(&self) -> Result<Vec<PeriodicTaskRow>>;
+    /// Remove a periodic task. Returns false if no task had that name.
+    fn delete_periodic(&self, name: &str) -> Result<bool>;
+    /// Pause (false) or resume (true) a periodic task by toggling `enabled`.
+    /// Returns false if no task had that name.
+    fn set_periodic_enabled(&self, name: &str, enabled: bool) -> Result<bool>;
 
     // ── Metrics operations ──────────────────────────────────────────
 

crates/taskito-core/tests/rust/storage_tests.rs

@@ -626,6 +626,57 @@ fn test_archived_job_payload_resolves(s: &impl Storage) {
     assert_eq!(row.result, Some(vec![0x11, 0x22]));
 }
 
+fn due_periodic_names(s: &impl Storage) -> Vec<String> {
+    s.get_due_periodic(now_millis())
+        .unwrap()
+        .into_iter()
+        .map(|p| p.name)
+        .collect()
+}
+
+fn test_periodic_crud(s: &impl Storage) {
+    use taskito_core::storage::models::NewPeriodicTaskRow;
+    let past = now_millis() - 1_000;
+    let row = |name: &'static str| NewPeriodicTaskRow {
+        name,
+        task_name: "periodic-task",
+        cron_expr: "* * * * *",
+        args: None,
+        kwargs: None,
+        queue: "default",
+        enabled: true,
+        next_run: past,
+        timezone: None,
+    };
+    s.register_periodic(&row("pc-a")).unwrap();
+    s.register_periodic(&row("pc-b")).unwrap();
+
+    // list_periodic returns every registered task.
+    let listed: Vec<String> = s
+        .list_periodic()
+        .unwrap()
+        .into_iter()
+        .map(|p| p.name)
+        .collect();
+    assert!(listed.contains(&"pc-a".to_string()) && listed.contains(&"pc-b".to_string()));
+
+    // Pausing drops it from the due set but keeps it in the catalog.
+    assert!(s.set_periodic_enabled("pc-a", false).unwrap());
+    assert!(!due_periodic_names(s).contains(&"pc-a".to_string()));
+    assert!(s.list_periodic().unwrap().iter().any(|p| p.name == "pc-a"));
+
+    // Resuming makes it due again.
+    assert!(s.set_periodic_enabled("pc-a", true).unwrap());
+    assert!(due_periodic_names(s).contains(&"pc-a".to_string()));
+
+    // Toggling or deleting an unknown task reports "not found".
+    assert!(!s.set_periodic_enabled("pc-missing", true).unwrap());
+
+    assert!(s.delete_periodic("pc-a").unwrap());
+    assert!(!s.list_periodic().unwrap().iter().any(|p| p.name == "pc-a"));
+    assert!(!s.delete_periodic("pc-a").unwrap());
+}
+
 fn run_storage_tests(s: &impl Storage) {
     test_enqueue_and_get(s);
     test_dequeue(s);
@@ -647,6 +698,7 @@ fn run_storage_tests(s: &impl Storage) {
     test_record_and_get_errors(s);
     test_workers(s);
     test_pause_resume_queue(s);
+    test_periodic_crud(s);
     test_circuit_breakers(s);
     test_execution_claims_purge(s);
     test_dashboard_settings(s);

crates/taskito-java/src/convert.rs

@@ -6,7 +6,7 @@
 use serde::{Deserialize, Serialize};
 use taskito_core::job::{now_millis, Job, NewJob};
 use taskito_core::storage::models::{
-    JobErrorRow, LockInfoRow, TaskLogRow, TaskMetricRow, WorkerRow,
+    JobErrorRow, LockInfoRow, PeriodicTaskRow, TaskLogRow, TaskMetricRow, WorkerRow,
 };
 use taskito_core::storage::{DeadJob, QueueStats};
 
@@ -203,6 +203,35 @@ pub fn status_code(status: &str) -> Option<i32> {
     }
 }
 
+/// Periodic-task view (omits the opaque args/kwargs payloads).
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PeriodicTaskView<'a> {
+    pub name: &'a str,
+    pub task_name: &'a str,
+    pub cron_expr: &'a str,
+    pub queue: &'a str,
+    pub enabled: bool,
+    pub last_run: Option<i64>,
+    pub next_run: i64,
+    pub timezone: Option<&'a str>,
+}
+
+impl<'a> From<&'a PeriodicTaskRow> for PeriodicTaskView<'a> {
+    fn from(p: &'a PeriodicTaskRow) -> Self {
+        Self {
+            name: &p.name,
+            task_name: &p.task_name,
+            cron_expr: &p.cron_expr,
+            queue: &p.queue,
+            enabled: p.enabled,
+            last_run: p.last_run,
+            next_run: p.next_run,
+            timezone: p.timezone.as_deref(),
+        }
+    }
+}
+
 /// Dead-letter entry view (omits the opaque payload).
 #[derive(Serialize)]
 #[serde(rename_all = "camelCase")]