prospero_core/
config_store.rs

1//! Mutable config records (the managed-repo registry) on the shared DB.
2//!
3//! Distinct from [`crate::store::Store`] because the access pattern is key-value
4//! upsert/read, not append/replay. Standalone uses [`SqliteConfigStore`] in the
5//! same `events.db`; a Postgres-backed impl drops in behind the trait in the
6//! clustered tier (Phase 2). See the topology design spec §3.4.
7
8use std::path::Path;
9use std::time::Duration;
10
11use async_trait::async_trait;
12use sqlx::Row;
13use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePool, SqlitePoolOptions};
14
15use crate::automation::{AutomationRun, RUN_HISTORY_LIMIT, RunSource, StoredAutomation};
16use crate::error::{CoreError, Result};
17use crate::registry::RegisteredWorkspace;
18
19/// Durable, mutable store for the managed-repo registry.
20#[async_trait]
21pub trait ConfigStore: Send + Sync {
22    /// All registered repos, ordered by name.
23    async fn list_repos(&self) -> Result<Vec<RegisteredWorkspace>>;
24    /// Insert or update a repo (keyed by `name`).
25    async fn upsert_repo(&self, repo: &RegisteredWorkspace) -> Result<()>;
26    /// Remove a repo by name. Returns whether a row was deleted.
27    async fn delete_repo(&self, name: &str) -> Result<bool>;
28
29    /// All automations, ordered by id (#220).
30    async fn list_automations(&self) -> Result<Vec<StoredAutomation>>;
31
32    /// Insert or update an automation (keyed by `id`).
33    ///
34    /// Does **not** write `last_fired_at`: that column is owned by
35    /// [`Self::claim_automation_fire`], and letting an ordinary edit move it
36    /// would hand an operator a way to replay or skip a tick.
37    async fn upsert_automation(&self, automation: &StoredAutomation) -> Result<()>;
38
39    /// Remove an automation and its run history. Returns whether it existed.
40    async fn delete_automation(&self, id: &str) -> Result<bool>;
41
42    /// Claim `fire_at` for `id`, returning whether **this** caller won it.
43    ///
44    /// The exactly-once primitive behind scheduled spawns: a conditional
45    /// `UPDATE … WHERE last_fired_at < fire_at` that the database serializes,
46    /// so of N replicas noticing the same due tick, exactly one is told to act.
47    ///
48    /// This is the guarantee, not the lifecycle lease. A lease makes two
49    /// replicas *unlikely* to act on one tick; it cannot rule it out, because
50    /// ownership can change hands between the moment a replica checks and the
51    /// moment it spawns. The compare-and-set closes that window, and keeps
52    /// holding when a lease is handed over mid-tick.
53    async fn claim_automation_fire(&self, id: &str, fire_at: &str) -> Result<bool>;
54
55    /// Append a run to an automation's history, trimming to
56    /// [`RUN_HISTORY_LIMIT`].
57    async fn record_run(&self, run: &AutomationRun) -> Result<()>;
58
59    /// An automation's runs, newest first, at most `limit`.
60    async fn list_runs(&self, automation_id: &str, limit: usize) -> Result<Vec<AutomationRun>>;
61}
62
63const SCHEMA: &str = "CREATE TABLE IF NOT EXISTS repos (\
64    name   TEXT PRIMARY KEY,\
65    root   TEXT NOT NULL,\
66    config TEXT NOT NULL\
67)";
68
69/// `last_fired_at` is a column rather than a field inside `spec` so the claim
70/// statement can compare it in SQL — a JSON round trip could not be made
71/// atomic across replicas. `spec` holds the rest of the [`Automation`].
72const AUTOMATIONS_SCHEMA: &str = "CREATE TABLE IF NOT EXISTS automations (\
73    id             TEXT PRIMARY KEY,\
74    spec           TEXT NOT NULL,\
75    webhook_secret TEXT,\
76    last_fired_at  TEXT\
77)";
78
79const RUNS_SCHEMA: &str = "CREATE TABLE IF NOT EXISTS automation_runs (\
80    automation_id TEXT NOT NULL,\
81    fired_at      TEXT NOT NULL,\
82    source        TEXT NOT NULL,\
83    agent_id      TEXT,\
84    error         TEXT\
85)";
86
87const RUNS_INDEX: &str = "CREATE INDEX IF NOT EXISTS automation_runs_by_time ON automation_runs (automation_id, fired_at DESC)";
88
89/// Serialize an automation's `spec` column.
90///
91/// `last_fired_at` is cleared first: the column of the same name is the single
92/// source of truth (only the claim statement may move it), and a second copy
93/// inside the JSON would be a stale value waiting to be read by mistake.
94pub(crate) fn encode_spec(automation: &StoredAutomation) -> Result<String> {
95    let mut spec = automation.automation.clone();
96    spec.last_fired_at = None;
97    Ok(serde_json::to_string(&spec)?)
98}
99
100/// Render a [`RunSource`] for its column via serde, so the stored spelling and
101/// the wire spelling cannot drift apart.
102pub(crate) fn encode_source(source: RunSource) -> Result<String> {
103    serde_json::to_value(source)?
104        .as_str()
105        .map(str::to_string)
106        .ok_or_else(|| CoreError::Store("run source did not serialize as a string".into()))
107}
108
109/// Rebuild a [`StoredAutomation`] from a row of `spec`, `webhook_secret` and
110/// `last_fired_at`. Shared by both backends — one decoder, one behaviour.
111pub(crate) fn decode_automation<R>(row: &R) -> Result<StoredAutomation>
112where
113    R: Row,
114    for<'a> String: sqlx::Decode<'a, R::Database> + sqlx::Type<R::Database>,
115    for<'a> Option<String>: sqlx::Decode<'a, R::Database> + sqlx::Type<R::Database>,
116    for<'a> &'a str: sqlx::ColumnIndex<R>,
117{
118    let decode = |e: sqlx::Error| CoreError::Store(format!("automation decode: {e}"));
119    let spec: String = row.try_get("spec").map_err(decode)?;
120    let webhook_secret: Option<String> = row.try_get("webhook_secret").map_err(decode)?;
121    let last_fired_at: Option<String> = row.try_get("last_fired_at").map_err(decode)?;
122    let mut automation: crate::automation::Automation = serde_json::from_str(&spec)?;
123    automation.last_fired_at = last_fired_at;
124    Ok(StoredAutomation {
125        automation,
126        webhook_secret,
127    })
128}
129
130/// Rebuild an [`AutomationRun`] from its row.
131pub(crate) fn decode_run<R>(row: &R) -> Result<AutomationRun>
132where
133    R: Row,
134    for<'a> String: sqlx::Decode<'a, R::Database> + sqlx::Type<R::Database>,
135    for<'a> Option<String>: sqlx::Decode<'a, R::Database> + sqlx::Type<R::Database>,
136    for<'a> &'a str: sqlx::ColumnIndex<R>,
137{
138    let decode = |e: sqlx::Error| CoreError::Store(format!("run decode: {e}"));
139    let source: String = row.try_get("source").map_err(decode)?;
140    Ok(AutomationRun {
141        automation_id: row.try_get("automation_id").map_err(decode)?,
142        fired_at: row.try_get("fired_at").map_err(decode)?,
143        source: serde_json::from_value(serde_json::Value::String(source))?,
144        agent_id: row.try_get("agent_id").map_err(decode)?,
145        error: row.try_get("error").map_err(decode)?,
146    })
147}
148
149/// sqlx/sqlite-backed config store — shares `events.db` with the event store.
150pub struct SqliteConfigStore {
151    pool: SqlitePool,
152}
153
154impl SqliteConfigStore {
155    /// Open (creating it + parent dirs if missing) the config store in
156    /// `dir/events.db` (the same file the event store uses).
157    pub async fn open(dir: &Path) -> Result<Self> {
158        std::fs::create_dir_all(dir)?;
159        let path = dir.join("events.db");
160        let opts = SqliteConnectOptions::new()
161            .filename(&path)
162            .create_if_missing(true)
163            .journal_mode(SqliteJournalMode::Wal)
164            .busy_timeout(Duration::from_secs(5));
165        let pool = SqlitePoolOptions::new()
166            .connect_with(opts)
167            .await
168            .map_err(|e| CoreError::Store(format!("opening config store: {e}")))?;
169        for ddl in [SCHEMA, AUTOMATIONS_SCHEMA, RUNS_SCHEMA, RUNS_INDEX] {
170            sqlx::query(ddl)
171                .execute(&pool)
172                .await
173                .map_err(|e| CoreError::Store(format!("initializing config schema: {e}")))?;
174        }
175        Ok(Self { pool })
176    }
177}
178
179#[async_trait]
180impl ConfigStore for SqliteConfigStore {
181    async fn list_repos(&self) -> Result<Vec<RegisteredWorkspace>> {
182        let rows = sqlx::query("SELECT name, root, config FROM repos ORDER BY name")
183            .fetch_all(&self.pool)
184            .await
185            .map_err(|e| CoreError::Store(format!("list_repos: {e}")))?;
186        let mut repos = Vec::with_capacity(rows.len());
187        for row in rows {
188            let decode = |e: sqlx::Error| CoreError::Store(format!("list_repos decode: {e}"));
189            let name: String = row.try_get("name").map_err(decode)?;
190            let root: String = row.try_get("root").map_err(decode)?;
191            let config_json: String = row.try_get("config").map_err(decode)?;
192            repos.push(RegisteredWorkspace {
193                name,
194                root: root.into(),
195                config: serde_json::from_str(&config_json)?,
196            });
197        }
198        Ok(repos)
199    }
200
201    async fn upsert_repo(&self, repo: &RegisteredWorkspace) -> Result<()> {
202        let config = serde_json::to_string(&repo.config)?;
203        // Surface a non-UTF8 root explicitly rather than silently lossy-mangling it.
204        let root = repo
205            .root
206            .to_str()
207            .ok_or_else(|| CoreError::Store(format!("non-UTF8 repo root path: {:?}", repo.root)))?;
208        sqlx::query(
209            "INSERT INTO repos (name, root, config) VALUES (?, ?, ?) \
210             ON CONFLICT(name) DO UPDATE SET root = excluded.root, config = excluded.config",
211        )
212        .bind(&repo.name)
213        .bind(root)
214        .bind(config)
215        .execute(&self.pool)
216        .await
217        .map_err(|e| CoreError::Store(format!("upsert_repo: {e}")))?;
218        Ok(())
219    }
220
221    async fn delete_repo(&self, name: &str) -> Result<bool> {
222        let res = sqlx::query("DELETE FROM repos WHERE name = ?")
223            .bind(name)
224            .execute(&self.pool)
225            .await
226            .map_err(|e| CoreError::Store(format!("delete_repo: {e}")))?;
227        Ok(res.rows_affected() > 0)
228    }
229
230    async fn list_automations(&self) -> Result<Vec<StoredAutomation>> {
231        let rows =
232            sqlx::query("SELECT spec, webhook_secret, last_fired_at FROM automations ORDER BY id")
233                .fetch_all(&self.pool)
234                .await
235                .map_err(|e| CoreError::Store(format!("list_automations: {e}")))?;
236        rows.iter().map(decode_automation).collect()
237    }
238
239    async fn upsert_automation(&self, automation: &StoredAutomation) -> Result<()> {
240        let spec = encode_spec(automation)?;
241        sqlx::query(
242            "INSERT INTO automations (id, spec, webhook_secret, last_fired_at) \
243             VALUES (?, ?, ?, NULL) \
244             ON CONFLICT(id) DO UPDATE SET spec = excluded.spec, \
245             webhook_secret = excluded.webhook_secret",
246        )
247        .bind(&automation.automation.id)
248        .bind(spec)
249        .bind(&automation.webhook_secret)
250        .execute(&self.pool)
251        .await
252        .map_err(|e| CoreError::Store(format!("upsert_automation: {e}")))?;
253        Ok(())
254    }
255
256    async fn delete_automation(&self, id: &str) -> Result<bool> {
257        sqlx::query("DELETE FROM automation_runs WHERE automation_id = ?")
258            .bind(id)
259            .execute(&self.pool)
260            .await
261            .map_err(|e| CoreError::Store(format!("delete_automation runs: {e}")))?;
262        let res = sqlx::query("DELETE FROM automations WHERE id = ?")
263            .bind(id)
264            .execute(&self.pool)
265            .await
266            .map_err(|e| CoreError::Store(format!("delete_automation: {e}")))?;
267        Ok(res.rows_affected() > 0)
268    }
269
270    async fn claim_automation_fire(&self, id: &str, fire_at: &str) -> Result<bool> {
271        let res = sqlx::query(
272            "UPDATE automations SET last_fired_at = ? \
273             WHERE id = ? AND (last_fired_at IS NULL OR last_fired_at < ?)",
274        )
275        .bind(fire_at)
276        .bind(id)
277        .bind(fire_at)
278        .execute(&self.pool)
279        .await
280        .map_err(|e| CoreError::Store(format!("claim_automation_fire: {e}")))?;
281        Ok(res.rows_affected() == 1)
282    }
283
284    async fn record_run(&self, run: &AutomationRun) -> Result<()> {
285        sqlx::query(
286            "INSERT INTO automation_runs (automation_id, fired_at, source, agent_id, error) \
287             VALUES (?, ?, ?, ?, ?)",
288        )
289        .bind(&run.automation_id)
290        .bind(&run.fired_at)
291        .bind(encode_source(run.source)?)
292        .bind(&run.agent_id)
293        .bind(&run.error)
294        .execute(&self.pool)
295        .await
296        .map_err(|e| CoreError::Store(format!("record_run: {e}")))?;
297
298        // Trim in the same call that appends, so history stays bounded even if
299        // nothing ever reads it.
300        sqlx::query(
301            "DELETE FROM automation_runs WHERE automation_id = ? AND rowid NOT IN \
302             (SELECT rowid FROM automation_runs WHERE automation_id = ? \
303              ORDER BY fired_at DESC LIMIT ?)",
304        )
305        .bind(&run.automation_id)
306        .bind(&run.automation_id)
307        .bind(RUN_HISTORY_LIMIT as i64)
308        .execute(&self.pool)
309        .await
310        .map_err(|e| CoreError::Store(format!("trim runs: {e}")))?;
311        Ok(())
312    }
313
314    async fn list_runs(&self, automation_id: &str, limit: usize) -> Result<Vec<AutomationRun>> {
315        let rows = sqlx::query(
316            "SELECT automation_id, fired_at, source, agent_id, error FROM automation_runs \
317             WHERE automation_id = ? ORDER BY fired_at DESC LIMIT ?",
318        )
319        .bind(automation_id)
320        .bind(limit as i64)
321        .fetch_all(&self.pool)
322        .await
323        .map_err(|e| CoreError::Store(format!("list_runs: {e}")))?;
324        rows.iter().map(decode_run).collect()
325    }
326}
327
328#[cfg(test)]
329mod tests {
330    use super::*;
331
332    #[tokio::test]
333    async fn sqlite_config_store_satisfies_conformance() {
334        let dir = tempfile::tempdir().unwrap();
335        let store = SqliteConfigStore::open(dir.path()).await.unwrap();
336        crate::testkit::config_store_conformance(&store).await;
337    }
338
339    #[tokio::test]
340    async fn sqlite_config_store_satisfies_automation_conformance() {
341        let dir = tempfile::tempdir().unwrap();
342        let store = SqliteConfigStore::open(dir.path()).await.unwrap();
343        crate::testkit::automation_store_conformance(&store).await;
344    }
345
346    #[tokio::test]
347    async fn sqlite_config_store_claims_a_tick_once() {
348        let dir = tempfile::tempdir().unwrap();
349        let store = SqliteConfigStore::open(dir.path()).await.unwrap();
350        crate::testkit::automation_claim_conformance(&store).await;
351    }
352}
prospero_core/config_store.rs

prospero_core/
config_store.rs
