valence_backend_postgres/
backend.rs

1//! Postgres storage engine.
2
3use sqlx::postgres::PgPool;
4
5use valence_backend_sql::{
6    apply_ttl_policy_postgres, create_record_postgres, define_unique_index_postgres,
7    delete_record_postgres, ensure_edges_postgres, ensure_table_postgres,
8    ensure_typed_table_postgres, execute_select_postgres, get_edge_sources_postgres,
9    get_edge_targets_postgres, get_record_postgres, inspect_typed_layout_postgres,
10    merge_record_postgres, read_schema_version_postgres, relate_edge_postgres, sql_capabilities,
11    sync_typed_table_postgres, ttl_deferred, unrelate_edge_postgres, update_record_postgres,
12    write_schema_version_postgres, WriteEnsureCache,
13};
14use valence_core::backend::DatabaseBackend;
15use valence_core::compiled_query::CompiledQuery;
16use valence_core::error::{Error, Result};
17use valence_core::record_id::RecordId;
18use valence_core::ttl::SchemaTtlPolicy;
19use valence_core::{Database, DatabaseFromEngine, KnownEngines};
20
21/// Stable engine slug for router keys (`postgres:logical_name`).
22pub const ENGINE_ID: &str = KnownEngines::POSTGRES;
23
24/// Schema evaluator const for `database:` routing.
25pub const PRIMARY: DatabaseFromEngine = Database::from_engine("primary", ENGINE_ID);
26
27/// Postgres-backed [`DatabaseBackend`] using typed columns (JSONB cells for `Json` fields).
28///
29/// # Examples
30///
31/// ```ignore
32/// use std::sync::Arc;
33/// use valence::{
34///     valence_schema, Database, DatabaseFromEngine, FieldType, PostgresBackend, Valence,
35///     POSTGRES_ENGINE_ID,
36/// };
37///
38/// const COUNTER_DB: DatabaseFromEngine =
39///     Database::from_engine("default", POSTGRES_ENGINE_ID);
40///
41/// valence_schema! {
42///     Counter {
43///         table: "counter",
44///         version: "0.1.0",
45///         database: COUNTER_DB,
46///         fields: [
47///             id: { r#type: FieldType::String, primary_key: true, required: true },
48///             value: { r#type: FieldType::Integer, required: true },
49///         ],
50///     }
51/// }
52///
53/// // Reads DATABASE_URL.
54/// let backend = PostgresBackend::from_env().await?;
55/// let valence = Valence::builder()
56///     .add_backend("default", Arc::new(backend))
57///     .build()?;
58/// assert_eq!(
59///     valence.backend_for_table("counter")?.engine_id(),
60///     POSTGRES_ENGINE_ID
61/// );
62/// # Ok::<(), valence::Error>(())
63/// ```
64#[derive(Debug, Clone)]
65pub struct PostgresBackend {
66    pool: PgPool,
67    layout_ensured: WriteEnsureCache,
68}
69
70impl PostgresBackend {
71    /// Start a builder for explicit host wiring.
72    pub fn builder() -> crate::config::PostgresBackendBuilder {
73        crate::config::PostgresBackendBuilder::new()
74    }
75
76    /// Connect using env defaults via builder (shorthand).
77    ///
78    /// # Errors
79    ///
80    /// Returns an error if `DATABASE_URL` is missing or the connection fails.
81    pub async fn from_env() -> Result<Self> {
82        Self::builder().from_env_defaults().build().await
83    }
84
85    /// Connect using a Postgres connection URL.
86    ///
87    /// # Errors
88    ///
89    /// Returns [`Error::Database`] if connecting or ensuring the edges schema fails.
90    pub async fn connect(url: &str) -> Result<Self> {
91        let pool = PgPool::connect(url)
92            .await
93            .map_err(|e| Error::database(e.to_string()))?;
94        ensure_edges_postgres(&pool).await?;
95        Ok(Self {
96            pool,
97            layout_ensured: WriteEnsureCache::new(),
98        })
99    }
100
101    /// Borrow the underlying pool.
102    pub fn pool(&self) -> &PgPool {
103        &self.pool
104    }
105}
106
107#[async_trait::async_trait]
108impl DatabaseBackend for PostgresBackend {
109    fn engine_id(&self) -> &'static str {
110        ENGINE_ID
111    }
112
113    fn capabilities(&self) -> valence_core::BackendCapabilities {
114        sql_capabilities("postgres")
115    }
116
117    async fn execute_compiled_query(
118        &self,
119        compiled: &CompiledQuery,
120    ) -> Result<Vec<serde_json::Value>> {
121        execute_select_postgres(&self.pool, compiled, "").await
122    }
123
124    async fn ensure_schemaless_table(&self, table: &str) -> Result<()> {
125        ensure_table_postgres(&self.pool, table).await
126    }
127
128    async fn inspect_typed_layout(
129        &self,
130        table: &str,
131    ) -> Result<Option<valence_core::storage_layout::StorageLayout>> {
132        inspect_typed_layout_postgres(&self.pool, table).await
133    }
134
135    async fn ensure_typed_table(
136        &self,
137        layout: &valence_core::storage_layout::StorageLayout,
138    ) -> Result<()> {
139        ensure_typed_table_postgres(&self.pool, layout).await
140    }
141
142    async fn sync_typed_table(
143        &self,
144        layout: &valence_core::storage_layout::StorageLayout,
145    ) -> Result<()> {
146        sync_typed_table_postgres(&self.pool, layout).await
147    }
148
149    async fn read_schema_version(&self, table: &str) -> Result<Option<String>> {
150        read_schema_version_postgres(&self.pool, table).await
151    }
152
153    async fn write_schema_version(&self, table: &str, version: &str) -> Result<()> {
154        write_schema_version_postgres(&self.pool, table, version).await
155    }
156
157    async fn get_record(&self, table: &str, id: &str) -> Result<Option<serde_json::Value>> {
158        get_record_postgres(&self.pool, table, id).await
159    }
160
161    async fn create_record(
162        &self,
163        table: &str,
164        content: serde_json::Value,
165    ) -> Result<serde_json::Value> {
166        create_record_postgres(&self.pool, table, content, &self.layout_ensured).await
167    }
168
169    async fn update_record(
170        &self,
171        table: &str,
172        id: &str,
173        content: serde_json::Value,
174    ) -> Result<serde_json::Value> {
175        update_record_postgres(&self.pool, table, id, content, &self.layout_ensured).await
176    }
177
178    async fn merge_record(
179        &self,
180        table: &str,
181        id: &str,
182        patch: serde_json::Value,
183    ) -> Result<serde_json::Value> {
184        merge_record_postgres(&self.pool, table, id, patch, &self.layout_ensured).await
185    }
186
187    async fn upsert_record(
188        &self,
189        table: &str,
190        id: &str,
191        content: serde_json::Value,
192    ) -> Result<serde_json::Value> {
193        if self.get_record(table, id).await?.is_some() {
194            self.update_record(table, id, content).await
195        } else {
196            let mut c = content;
197            if let Some(obj) = c.as_object_mut() {
198                obj.insert("id".into(), serde_json::json!({"table": table, "id": id}));
199            }
200            self.create_record(table, c).await
201        }
202    }
203
204    async fn delete_record(&self, table: &str, id: &str) -> Result<()> {
205        delete_record_postgres(&self.pool, table, id).await
206    }
207
208    async fn relate_edge(&self, from: &RecordId, edge_table: &str, to: &RecordId) -> Result<()> {
209        relate_edge_postgres(&self.pool, from, edge_table, to).await
210    }
211
212    async fn unrelate_edge(&self, from: &RecordId, edge_table: &str, to: &RecordId) -> Result<()> {
213        unrelate_edge_postgres(&self.pool, from, edge_table, to).await
214    }
215
216    async fn get_edge_targets(&self, from: &RecordId, edge_table: &str) -> Result<Vec<RecordId>> {
217        get_edge_targets_postgres(&self.pool, from, edge_table).await
218    }
219
220    async fn get_edge_sources(&self, to: &RecordId, edge_table: &str) -> Result<Vec<RecordId>> {
221        get_edge_sources_postgres(&self.pool, to, edge_table).await
222    }
223
224    async fn define_unique_index(&self, table: &str, field: &str) -> Result<()> {
225        define_unique_index_postgres(&self.pool, table, field, &self.layout_ensured).await
226    }
227
228    fn ttl_capability(&self) -> valence_core::ttl::BackendTtlCapability {
229        ttl_deferred()
230    }
231
232    async fn apply_ttl_policy(&self, table: &str, policy: &SchemaTtlPolicy) -> Result<()> {
233        apply_ttl_policy_postgres(&self.pool, table, policy, &self.layout_ensured).await
234    }
235}
valence_backend_postgres/backend.rs

valence_backend_postgres/
backend.rs
