valence_backend_postgres/
backend.rs

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

valence_backend_postgres/
backend.rs
