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::error::{CoreError, Result};
16use crate::registry::RegisteredWorkspace;
17
18/// Durable, mutable store for the managed-repo registry.
19#[async_trait]
20pub trait ConfigStore: Send + Sync {
21    /// All registered repos, ordered by name.
22    async fn list_repos(&self) -> Result<Vec<RegisteredWorkspace>>;
23    /// Insert or update a repo (keyed by `name`).
24    async fn upsert_repo(&self, repo: &RegisteredWorkspace) -> Result<()>;
25    /// Remove a repo by name. Returns whether a row was deleted.
26    async fn delete_repo(&self, name: &str) -> Result<bool>;
27}
28
29const SCHEMA: &str = "CREATE TABLE IF NOT EXISTS repos (\
30    name   TEXT PRIMARY KEY,\
31    root   TEXT NOT NULL,\
32    config TEXT NOT NULL\
33)";
34
35/// sqlx/sqlite-backed config store — shares `events.db` with the event store.
36pub struct SqliteConfigStore {
37    pool: SqlitePool,
38}
39
40impl SqliteConfigStore {
41    /// Open (creating it + parent dirs if missing) the config store in
42    /// `dir/events.db` (the same file the event store uses).
43    pub async fn open(dir: &Path) -> Result<Self> {
44        std::fs::create_dir_all(dir)?;
45        let path = dir.join("events.db");
46        let opts = SqliteConnectOptions::new()
47            .filename(&path)
48            .create_if_missing(true)
49            .journal_mode(SqliteJournalMode::Wal)
50            .busy_timeout(Duration::from_secs(5));
51        let pool = SqlitePoolOptions::new()
52            .connect_with(opts)
53            .await
54            .map_err(|e| CoreError::Store(format!("opening config store: {e}")))?;
55        sqlx::query(SCHEMA)
56            .execute(&pool)
57            .await
58            .map_err(|e| CoreError::Store(format!("initializing config schema: {e}")))?;
59        Ok(Self { pool })
60    }
61}
62
63#[async_trait]
64impl ConfigStore for SqliteConfigStore {
65    async fn list_repos(&self) -> Result<Vec<RegisteredWorkspace>> {
66        let rows = sqlx::query("SELECT name, root, config FROM repos ORDER BY name")
67            .fetch_all(&self.pool)
68            .await
69            .map_err(|e| CoreError::Store(format!("list_repos: {e}")))?;
70        let mut repos = Vec::with_capacity(rows.len());
71        for row in rows {
72            let decode = |e: sqlx::Error| CoreError::Store(format!("list_repos decode: {e}"));
73            let name: String = row.try_get("name").map_err(decode)?;
74            let root: String = row.try_get("root").map_err(decode)?;
75            let config_json: String = row.try_get("config").map_err(decode)?;
76            repos.push(RegisteredWorkspace {
77                name,
78                root: root.into(),
79                config: serde_json::from_str(&config_json)?,
80            });
81        }
82        Ok(repos)
83    }
84
85    async fn upsert_repo(&self, repo: &RegisteredWorkspace) -> Result<()> {
86        let config = serde_json::to_string(&repo.config)?;
87        // Surface a non-UTF8 root explicitly rather than silently lossy-mangling it.
88        let root = repo
89            .root
90            .to_str()
91            .ok_or_else(|| CoreError::Store(format!("non-UTF8 repo root path: {:?}", repo.root)))?;
92        sqlx::query(
93            "INSERT INTO repos (name, root, config) VALUES (?, ?, ?) \
94             ON CONFLICT(name) DO UPDATE SET root = excluded.root, config = excluded.config",
95        )
96        .bind(&repo.name)
97        .bind(root)
98        .bind(config)
99        .execute(&self.pool)
100        .await
101        .map_err(|e| CoreError::Store(format!("upsert_repo: {e}")))?;
102        Ok(())
103    }
104
105    async fn delete_repo(&self, name: &str) -> Result<bool> {
106        let res = sqlx::query("DELETE FROM repos WHERE name = ?")
107            .bind(name)
108            .execute(&self.pool)
109            .await
110            .map_err(|e| CoreError::Store(format!("delete_repo: {e}")))?;
111        Ok(res.rows_affected() > 0)
112    }
113}
114
115#[cfg(test)]
116mod tests {
117    use super::*;
118
119    #[tokio::test]
120    async fn sqlite_config_store_satisfies_conformance() {
121        let dir = tempfile::tempdir().unwrap();
122        let store = SqliteConfigStore::open(dir.path()).await.unwrap();
123        crate::testkit::config_store_conformance(&store).await;
124    }
125}
prospero_core/config_store.rs

prospero_core/
config_store.rs
