forge_workspace/store/
gotify.rs

1//! Gotify subscription CRUD on the redb `subscriptions` table.
2//!
3//! The whole [`GotifySubscription`] record is stored as serde-json
4//! keyed by its uuid bytes - no field schema on disk, so a v1->v2 type
5//! change needs no table migration.
6
7use anyhow::Context;
8use forge_primitives::GotifySubscription;
9use redb::{ReadableTable, TableDefinition};
10use uuid::Uuid;
11
12use super::Db;
13
14const SUBS: TableDefinition<[u8; 16], &[u8]> = TableDefinition::new("subscriptions");
15
16/// Persist a subscription, replacing any prior record with the same id.
17pub fn insert(db: &Db, sub: &GotifySubscription) -> anyhow::Result<()> {
18    let value = serde_json::to_vec(sub).context("serialize subscription")?;
19    let txn = db.database().begin_write()?;
20    {
21        let mut table = txn.open_table(SUBS)?;
22        table.insert(sub.id.as_bytes(), value.as_slice())?;
23    }
24    txn.commit()?;
25    Ok(())
26}
27
28/// Every persisted subscription.
29pub fn list(db: &Db) -> anyhow::Result<Vec<GotifySubscription>> {
30    read_all(db)
31}
32
33/// Delete a subscription by id. Returns whether a record existed.
34pub fn remove(db: &Db, id: Uuid) -> anyhow::Result<bool> {
35    let txn = db.database().begin_write()?;
36    let existed = {
37        let mut table = txn.open_table(SUBS)?;
38        table.remove(id.as_bytes())?.is_some()
39    };
40    txn.commit()?;
41    Ok(existed)
42}
43
44fn read_all(db: &Db) -> anyhow::Result<Vec<GotifySubscription>> {
45    let txn = db.database().begin_read()?;
46    let table = match txn.open_table(SUBS) {
47        Ok(t) => t,
48        // A fresh database has no table until the first write; an absent
49        // table is an empty list, not an error.
50        Err(redb::TableError::TableDoesNotExist(_)) => return Ok(Vec::new()),
51        Err(e) => return Err(e.into()),
52    };
53    let mut out = Vec::new();
54    for entry in table.iter()? {
55        let (id, value) = entry?;
56        match serde_json::from_slice::<GotifySubscription>(value.value()) {
57            Ok(sub) => out.push(sub),
58            // One undecodable record (schema drift, a corrupt blob) must not
59            // wipe the rest of the durable set - skip it and warn.
60            Err(err) => tracing::warn!(
61                target: "forge_workspace::store::gotify",
62                id = %Uuid::from_bytes(id.value()),
63                error = %err,
64                "skipping Gotify subscription record that failed to decode",
65            ),
66        }
67    }
68    Ok(out)
69}
70
71#[cfg(test)]
72mod tests {
73    use super::*;
74    use std::time::SystemTime;
75    use tempfile::tempdir;
76
77    fn sub(project: &str) -> GotifySubscription {
78        GotifySubscription {
79            id: Uuid::new_v4(),
80            project: project.to_owned(),
81            team_role: None,
82            applications: vec![],
83            min_priority: None,
84            created_at: SystemTime::UNIX_EPOCH,
85        }
86    }
87
88    #[test]
89    fn subscription_store_round_trip() {
90        let dir = tempdir().expect("tempdir");
91        let db = Db::open(&dir.path().join("db.redb")).expect("open db");
92
93        let s1 = sub("p1");
94        let s2 = sub("p2");
95        insert(&db, &s1).expect("insert s1");
96        insert(&db, &s2).expect("insert s2");
97
98        let all = list(&db).expect("list");
99        assert_eq!(all.len(), 2);
100        assert!(all.iter().any(|s| s.id == s1.id) && all.iter().any(|s| s.id == s2.id));
101
102        assert!(remove(&db, s1.id).expect("remove"), "an existing id removes to true");
103        assert!(!remove(&db, s1.id).expect("remove again"), "an absent id removes to false");
104        assert_eq!(list(&db).expect("list").len(), 1);
105    }
106
107    #[test]
108    fn dynamic_worker_subscription_survives_db_reopen() {
109        let dir = tempdir().expect("tempdir");
110        let path = dir.path().join("db.redb");
111
112        let mut worker_sub = sub("forge");
113        worker_sub.team_role = Some("scratch".to_owned());
114        let id = worker_sub.id;
115        {
116            let db = Db::open(&path).expect("open db");
117            insert(&db, &worker_sub).expect("insert");
118        }
119
120        // Reopen the store the way a forge restart does; the boot path in
121        // workspace.rs rebuilds the active set from `list`.
122        let db = Db::open(&path).expect("reopen db");
123        let restored = list(&db).expect("list after reopen");
124        assert_eq!(restored.len(), 1);
125        assert_eq!(restored[0].id, id);
126        assert_eq!(
127            restored[0].team_role.as_deref(),
128            Some("scratch"),
129            "a dynamic worker's durable sub survives a restart with its label intact",
130        );
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 = sub("p1");
139        insert(&db, &good).expect("insert good");
140
141        // A blob that isn't a valid subscription must not poison the load.
142        let txn = db.database().begin_write().expect("begin");
143        {
144            let mut table = txn.open_table(SUBS).expect("open table");
145            table.insert([0u8; 16], "not a subscription".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}
forge_workspace/store/gotify.rs

forge_workspace/store/
gotify.rs
