forge_workspace/store/
cron.rs

1//! Durable forge-cron persistence on the redb `crons` table.
2//!
3//! The whole [`CronEntry`] record is stored as serde-json keyed by its
4//! `CronId` string - no field schema on disk, so a v1->v2 type change
5//! needs no table migration. Machine-local: crons are per-machine, not
6//! synced across the user's Macs.
7
8use anyhow::Context;
9use forge_primitives::cron::{CronEntry, CronId};
10use redb::{ReadableTable, TableDefinition};
11
12use super::Db;
13
14const CRONS: TableDefinition<&str, &[u8]> = TableDefinition::new("crons");
15
16/// Every persisted cron.
17pub fn list(db: &Db) -> anyhow::Result<Vec<CronEntry>> {
18    let txn = db.database().begin_read()?;
19    let table = match txn.open_table(CRONS) {
20        Ok(t) => t,
21        // A fresh database has no table until the first write; an absent
22        // table is an empty list, not an error.
23        Err(redb::TableError::TableDoesNotExist(_)) => return Ok(Vec::new()),
24        Err(e) => return Err(e.into()),
25    };
26    let mut out = Vec::new();
27    for entry in table.iter()? {
28        let (id, value) = entry?;
29        match serde_json::from_slice::<CronEntry>(value.value()) {
30            Ok(cron) => out.push(cron),
31            // One undecodable record (schema drift, a corrupt blob) must not
32            // wipe the rest of the durable set - skip it and warn.
33            Err(err) => tracing::warn!(
34                target: "forge_workspace::store::cron",
35                id = %id.value(),
36                error = %err,
37                "skipping cron record that failed to decode",
38            ),
39        }
40    }
41    Ok(out)
42}
43
44/// Persist one cron, replacing any prior record with the same id. One
45/// write txn per row, so a write can only touch the row it names.
46pub fn upsert(db: &Db, cron: &CronEntry) -> anyhow::Result<()> {
47    let value = serde_json::to_vec(cron).context("serialize cron")?;
48    let txn = db.database().begin_write()?;
49    {
50        let mut table = txn.open_table(CRONS)?;
51        table.insert(cron.id.as_str(), value.as_slice())?;
52    }
53    txn.commit()?;
54    Ok(())
55}
56
57/// Delete a cron by id. Returns whether a record existed.
58pub fn delete(db: &Db, id: &CronId) -> anyhow::Result<bool> {
59    let txn = db.database().begin_write()?;
60    let existed = {
61        let mut table = txn.open_table(CRONS)?;
62        table.remove(id.as_str())?.is_some()
63    };
64    txn.commit()?;
65    Ok(existed)
66}
67
68#[cfg(test)]
69mod tests {
70    use super::*;
71    use forge_primitives::cron::{CronId, CronKind};
72    use std::time::{Duration, SystemTime};
73    use tempfile::tempdir;
74
75    fn epoch(secs: u64) -> SystemTime {
76        SystemTime::UNIX_EPOCH + Duration::from_secs(secs)
77    }
78
79    fn cron(id: &str) -> CronEntry {
80        CronEntry {
81            id: CronId::from(id),
82            project_name: "forge".to_owned(),
83            kind: CronKind::Recurring("0 9 * * *".to_owned()),
84            prompt: "stand-up".to_owned(),
85            created_at: epoch(1_700_000_000),
86            description: None,
87            last_fire: None,
88            next_fire: epoch(1_700_032_400),
89            team_role: None,
90            wake: true,
91        }
92    }
93
94    #[test]
95    fn cron_survives_db_reopen() {
96        let dir = tempdir().expect("tempdir");
97        let path = dir.path().join("db.redb");
98        let entry = cron("persist-me");
99        {
100            let db = Db::open(&path).expect("open db");
101            upsert(&db, &entry).expect("write");
102        }
103        let db = Db::open(&path).expect("reopen db");
104        let restored = list(&db).expect("list after reopen");
105        assert_eq!(restored, vec![entry], "a durable cron survives a restart intact");
106    }
107
108    #[test]
109    fn upsert_replaces_one_row_and_leaves_its_neighbours_alone() {
110        let dir = tempdir().expect("tempdir");
111        let db = Db::open(&dir.path().join("db.redb")).expect("open db");
112
113        upsert(&db, &cron("a")).expect("insert a");
114        upsert(&db, &cron("b")).expect("insert b");
115
116        let mut moved = cron("a");
117        moved.prompt = "moved".to_owned();
118        upsert(&db, &moved).expect("replace a");
119
120        let all = list(&db).expect("list");
121        assert_eq!(all.len(), 2, "a second upsert adds a row without clearing the table");
122        let a = all.iter().find(|c| c.id == CronId::from("a")).expect("a is there");
123        assert_eq!(a.prompt, "moved", "and the row it names is the one replaced");
124        assert!(all.iter().any(|c| c.id == CronId::from("b")), "the neighbour is untouched");
125
126        assert!(delete(&db, &CronId::from("a")).expect("delete"), "an existing id deletes to true");
127        assert!(!delete(&db, &CronId::from("a")).expect("delete again"), "and reports a miss");
128        let left = list(&db).expect("list");
129        assert_eq!(left.len(), 1, "only the named row is gone");
130        assert_eq!(left[0].id, CronId::from("b"));
131    }
132
133    #[test]
134    fn corrupt_record_is_skipped_not_fatal() {
135        let dir = tempdir().expect("tempdir");
136        let db = Db::open(&dir.path().join("db.redb")).expect("open db");
137
138        let good = cron("good");
139        upsert(&db, &good).expect("write good");
140
141        // A blob that isn't a valid cron must not poison the load.
142        let txn = db.database().begin_write().expect("begin");
143        {
144            let mut table = txn.open_table(CRONS).expect("open table");
145            table.insert("corrupt", "not a cron".as_bytes()).expect("insert corrupt");
146        }
147        txn.commit().expect("commit");
148
149        let all = list(&db).expect("list tolerates the corrupt blob");
150        assert_eq!(all.len(), 1, "the good record survives a corrupt sibling");
151        assert_eq!(all[0].id, good.id);
152    }
153
154    #[test]
155    fn a_field_absent_team_role_deserializes_to_none() {
156        // A record an old forge wrote (before team_role existed) has no
157        // team_role key at all. The decode relies on it defaulting to a
158        // lead cron; the round-trip tests only prove None -> no-key -> None,
159        // so pin the field-absent -> None direction against a hand-authored
160        // blob that never carried the key.
161        let blob = serde_json::json!({
162            "id": "legacy",
163            "project_name": "forge",
164            "kind": { "Recurring": "0 9 * * *" },
165            "prompt": "p",
166            "created_at": { "secs_since_epoch": 0, "nanos_since_epoch": 0 },
167            "next_fire": { "secs_since_epoch": 0, "nanos_since_epoch": 0 },
168        });
169        let entry: CronEntry =
170            serde_json::from_value(blob).expect("a field-absent record deserializes");
171        assert_eq!(entry.id, CronId::from("legacy"));
172        assert_eq!(entry.team_role, None, "a field-absent team_role is a lead cron");
173    }
174}
forge_workspace/store/cron.rs

forge_workspace/store/
cron.rs
