prospero_core/
postgres_config_store.rs

1//! sqlx-backed Postgres [`ConfigStore`] — the clustered-tier config backend.
2//!
3//! Mirrors [`crate::config_store::SqliteConfigStore`] with Postgres dialect.
4//! Runs the same conformance battery, gated on `DATABASE_URL`.
5
6use async_trait::async_trait;
7use sqlx::Row;
8use sqlx::postgres::PgPool;
9
10use crate::config_store::ConfigStore;
11use crate::error::{CoreError, Result};
12use crate::registry::RegisteredWorkspace;
13
14const SCHEMA: &str = "CREATE TABLE IF NOT EXISTS repos (\
15    name   TEXT PRIMARY KEY,\
16    root   TEXT NOT NULL,\
17    config TEXT NOT NULL\
18)";
19
20/// sqlx/Postgres-backed config store (clustered tier).
21pub struct PostgresConfigStore {
22    pool: PgPool,
23}
24
25impl PostgresConfigStore {
26    /// Connect to Postgres at `url` and ensure the schema exists.
27    pub async fn connect(url: &str) -> Result<Self> {
28        let pool = crate::pg::connect(url).await?;
29        crate::pg::ensure_schema(&pool, SCHEMA, "repos table").await?;
30        Ok(Self { pool })
31    }
32
33    /// Truncate all repos. Test-only.
34    #[cfg(any(test, feature = "testkit"))]
35    pub async fn reset_for_tests(&self) -> Result<()> {
36        sqlx::query("TRUNCATE repos")
37            .execute(&self.pool)
38            .await
39            .map_err(|e| CoreError::Store(format!("reset: {e}")))?;
40        Ok(())
41    }
42}
43
44#[async_trait]
45impl ConfigStore for PostgresConfigStore {
46    async fn list_repos(&self) -> Result<Vec<RegisteredWorkspace>> {
47        let rows = sqlx::query("SELECT name, root, config FROM repos ORDER BY name")
48            .fetch_all(&self.pool)
49            .await
50            .map_err(|e| CoreError::Store(format!("list_repos: {e}")))?;
51        let mut repos = Vec::with_capacity(rows.len());
52        for row in rows {
53            let decode = |e: sqlx::Error| CoreError::Store(format!("list_repos decode: {e}"));
54            let name: String = row.try_get("name").map_err(decode)?;
55            let root: String = row.try_get("root").map_err(decode)?;
56            let config_json: String = row.try_get("config").map_err(decode)?;
57            repos.push(RegisteredWorkspace {
58                name,
59                root: root.into(),
60                config: serde_json::from_str(&config_json)?,
61            });
62        }
63        Ok(repos)
64    }
65
66    async fn upsert_repo(&self, repo: &RegisteredWorkspace) -> Result<()> {
67        let config = serde_json::to_string(&repo.config)?;
68        let root = repo
69            .root
70            .to_str()
71            .ok_or_else(|| CoreError::Store(format!("non-UTF8 repo root path: {:?}", repo.root)))?;
72        sqlx::query(
73            "INSERT INTO repos (name, root, config) VALUES ($1, $2, $3) \
74             ON CONFLICT (name) DO UPDATE SET root = excluded.root, config = excluded.config",
75        )
76        .bind(&repo.name)
77        .bind(root)
78        .bind(config)
79        .execute(&self.pool)
80        .await
81        .map_err(|e| CoreError::Store(format!("upsert_repo: {e}")))?;
82        Ok(())
83    }
84
85    async fn delete_repo(&self, name: &str) -> Result<bool> {
86        let res = sqlx::query("DELETE FROM repos WHERE name = $1")
87            .bind(name)
88            .execute(&self.pool)
89            .await
90            .map_err(|e| CoreError::Store(format!("delete_repo: {e}")))?;
91        Ok(res.rows_affected() > 0)
92    }
93}
94
95#[cfg(test)]
96mod tests {
97    use super::*;
98
99    #[tokio::test]
100    async fn postgres_config_store_satisfies_conformance() {
101        let Ok(url) = std::env::var("DATABASE_URL") else {
102            eprintln!("SKIP postgres_config_store_satisfies_conformance: DATABASE_URL unset");
103            return;
104        };
105        let store = PostgresConfigStore::connect(&url).await.unwrap();
106        store.reset_for_tests().await.unwrap();
107        crate::testkit::config_store_conformance(&store).await;
108    }
109}
prospero_core/postgres_config_store.rs

prospero_core/
postgres_config_store.rs
