valence_backend_sqlite/
backend.rs

1//! SQLite storage engine.
2
3use sqlx::sqlite::{SqliteConnectOptions, SqlitePool, SqlitePoolOptions};
4use std::str::FromStr;
5
6use valence_backend_sql::{
7    apply_ttl_policy_sqlite, create_record_sqlite, define_unique_index_sqlite,
8    delete_record_sqlite, ensure_table_sqlite, ensure_typed_table_sqlite, execute_select_sqlite,
9    get_edge_sources_sqlite, get_edge_targets_sqlite, get_record_sqlite,
10    inspect_typed_layout_sqlite, merge_record_sqlite, read_schema_version_sqlite,
11    relate_edge_sqlite, sql_capabilities, sync_typed_table_sqlite, ttl_deferred,
12    unrelate_edge_sqlite, update_record_sqlite, write_schema_version_sqlite, 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 (`sqlite:logical_name`).
22pub const ENGINE_ID: &str = KnownEngines::SQLITE;
23
24/// Schema evaluator const for `database:` routing.
25pub const PRIMARY: DatabaseFromEngine = Database::from_engine("primary", ENGINE_ID);
26
27/// SQLite-backed [`DatabaseBackend`] using typed columns from schema layout.
28///
29/// # Examples
30///
31/// ```ignore
32/// use std::sync::Arc;
33/// use valence::{
34///     valence_schema, Database, DatabaseFromEngine, FieldType, SqliteBackend, Valence,
35///     SQLITE_ENGINE_ID,
36/// };
37///
38/// const COUNTER_DB: DatabaseFromEngine =
39///     Database::from_engine("default", SQLITE_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/// let backend = SqliteBackend::connect_memory().await?;
54/// let valence = Valence::builder()
55///     .add_backend("default", Arc::new(backend))
56///     .build()?;
57/// assert_eq!(
58///     valence.backend_for_table("counter")?.engine_id(),
59///     SQLITE_ENGINE_ID
60/// );
61/// # Ok::<(), valence::Error>(())
62/// ```
63#[derive(Debug, Clone)]
64pub struct SqliteBackend {
65    pool: SqlitePool,
66    layout_ensured: WriteEnsureCache,
67}
68
69impl SqliteBackend {
70    /// Connect to an in-memory SQLite database.
71    ///
72    /// # Errors
73    ///
74    /// Returns an error if the in-memory connection or edges schema setup fails.
75    pub async fn connect_memory() -> Result<Self> {
76        Self::connect(":memory:").await
77    }
78
79    /// Connect to a SQLite database at `path` (`:memory:` for ephemeral).
80    ///
81    /// In-memory URLs (`:memory:` or `mode=memory`) use a single-connection pool so every
82    /// checkout sees the same store. sqlx's default pool size otherwise opens isolated
83    /// private databases per connection for bare `:memory:`.
84    ///
85    /// # Errors
86    ///
87    /// Returns [`Error::Database`] if connecting or ensuring the edges schema fails.
88    pub async fn connect(path: &str) -> Result<Self> {
89        let options = SqliteConnectOptions::from_str(path)
90            .or_else(|_| SqliteConnectOptions::from_str(&format!("sqlite:{path}")))
91            .map_err(|e| Error::database(e.to_string()))?
92            .create_if_missing(true);
93        let memory =
94            path.contains(":memory:") || path.contains("mode=memory") || path == ":memory:";
95        let mut pool_opts = SqlitePoolOptions::new();
96        if memory {
97            pool_opts = pool_opts.max_connections(1);
98        }
99        let pool = pool_opts
100            .connect_with(options)
101            .await
102            .map_err(|e| Error::database(e.to_string()))?;
103        valence_backend_sql::ensure_edges_sqlite(&pool).await?;
104        Ok(Self {
105            pool,
106            layout_ensured: WriteEnsureCache::new(),
107        })
108    }
109
110    /// Borrow the underlying pool.
111    pub fn pool(&self) -> &SqlitePool {
112        &self.pool
113    }
114}
115
116#[async_trait::async_trait]
117impl DatabaseBackend for SqliteBackend {
118    fn engine_id(&self) -> &'static str {
119        ENGINE_ID
120    }
121
122    fn capabilities(&self) -> valence_core::BackendCapabilities {
123        sql_capabilities("sqlite")
124    }
125
126    async fn execute_compiled_query(
127        &self,
128        compiled: &CompiledQuery,
129    ) -> Result<Vec<serde_json::Value>> {
130        execute_select_sqlite(&self.pool, compiled, "").await
131    }
132
133    async fn ensure_schemaless_table(&self, table: &str) -> Result<()> {
134        ensure_table_sqlite(&self.pool, table).await
135    }
136
137    async fn inspect_typed_layout(
138        &self,
139        table: &str,
140    ) -> Result<Option<valence_core::storage_layout::StorageLayout>> {
141        inspect_typed_layout_sqlite(&self.pool, table).await
142    }
143
144    async fn ensure_typed_table(
145        &self,
146        layout: &valence_core::storage_layout::StorageLayout,
147    ) -> Result<()> {
148        ensure_typed_table_sqlite(&self.pool, layout).await
149    }
150
151    async fn sync_typed_table(
152        &self,
153        layout: &valence_core::storage_layout::StorageLayout,
154    ) -> Result<()> {
155        sync_typed_table_sqlite(&self.pool, layout).await
156    }
157
158    async fn read_schema_version(&self, table: &str) -> Result<Option<String>> {
159        read_schema_version_sqlite(&self.pool, table).await
160    }
161
162    async fn write_schema_version(&self, table: &str, version: &str) -> Result<()> {
163        write_schema_version_sqlite(&self.pool, table, version).await
164    }
165
166    async fn get_record(&self, table: &str, id: &str) -> Result<Option<serde_json::Value>> {
167        get_record_sqlite(&self.pool, table, id).await
168    }
169
170    async fn create_record(
171        &self,
172        table: &str,
173        content: serde_json::Value,
174    ) -> Result<serde_json::Value> {
175        create_record_sqlite(&self.pool, table, content, &self.layout_ensured).await
176    }
177
178    async fn update_record(
179        &self,
180        table: &str,
181        id: &str,
182        content: serde_json::Value,
183    ) -> Result<serde_json::Value> {
184        update_record_sqlite(&self.pool, table, id, content, &self.layout_ensured).await
185    }
186
187    async fn merge_record(
188        &self,
189        table: &str,
190        id: &str,
191        patch: serde_json::Value,
192    ) -> Result<serde_json::Value> {
193        merge_record_sqlite(&self.pool, table, id, patch, &self.layout_ensured).await
194    }
195
196    async fn upsert_record(
197        &self,
198        table: &str,
199        id: &str,
200        content: serde_json::Value,
201    ) -> Result<serde_json::Value> {
202        if self.get_record(table, id).await?.is_some() {
203            self.update_record(table, id, content).await
204        } else {
205            let mut c = content;
206            if let Some(obj) = c.as_object_mut() {
207                obj.insert("id".into(), serde_json::json!({"table": table, "id": id}));
208            }
209            self.create_record(table, c).await
210        }
211    }
212
213    async fn delete_record(&self, table: &str, id: &str) -> Result<()> {
214        delete_record_sqlite(&self.pool, table, id).await
215    }
216
217    async fn relate_edge(&self, from: &RecordId, edge_table: &str, to: &RecordId) -> Result<()> {
218        relate_edge_sqlite(&self.pool, from, edge_table, to).await
219    }
220
221    async fn unrelate_edge(&self, from: &RecordId, edge_table: &str, to: &RecordId) -> Result<()> {
222        unrelate_edge_sqlite(&self.pool, from, edge_table, to).await
223    }
224
225    async fn get_edge_targets(&self, from: &RecordId, edge_table: &str) -> Result<Vec<RecordId>> {
226        get_edge_targets_sqlite(&self.pool, from, edge_table).await
227    }
228
229    async fn get_edge_sources(&self, to: &RecordId, edge_table: &str) -> Result<Vec<RecordId>> {
230        get_edge_sources_sqlite(&self.pool, to, edge_table).await
231    }
232
233    async fn define_unique_index(&self, table: &str, field: &str) -> Result<()> {
234        define_unique_index_sqlite(&self.pool, table, field, &self.layout_ensured).await
235    }
236
237    fn ttl_capability(&self) -> valence_core::ttl::BackendTtlCapability {
238        ttl_deferred()
239    }
240
241    async fn apply_ttl_policy(&self, table: &str, policy: &SchemaTtlPolicy) -> Result<()> {
242        apply_ttl_policy_sqlite(&self.pool, table, policy, &self.layout_ensured).await
243    }
244}
valence_backend_sqlite/backend.rs

valence_backend_sqlite/
backend.rs
