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, execute_select_sqlite, get_edge_sources_sqlite,
9    get_edge_targets_sqlite, get_record_sqlite, merge_record_sqlite, relate_edge_sqlite,
10    sql_capabilities, ttl_deferred, unrelate_edge_sqlite, update_record_sqlite,
11};
12use valence_core::backend::DatabaseBackend;
13use valence_core::compiled_query::CompiledQuery;
14use valence_core::error::{Error, Result};
15use valence_core::record_id::RecordId;
16use valence_core::ttl::SchemaTtlPolicy;
17use valence_core::{Database, DatabaseFromEngine, KnownEngines};
18
19/// Stable engine slug for router keys (`sqlite:logical_name`).
20pub const ENGINE_ID: &str = KnownEngines::SQLITE;
21
22/// Schema evaluator const for `database:` routing.
23pub const PRIMARY: DatabaseFromEngine = Database::from_engine("primary", ENGINE_ID);
24
25/// SQLite-backed [`DatabaseBackend`] using JSON document rows.
26///
27/// # Examples
28///
29/// ```ignore
30/// use std::sync::Arc;
31/// use valence::{
32///     valence_schema, Database, DatabaseFromEngine, FieldType, SqliteBackend, Valence,
33///     SQLITE_ENGINE_ID,
34/// };
35///
36/// const COUNTER_DB: DatabaseFromEngine =
37///     Database::from_engine("default", SQLITE_ENGINE_ID);
38///
39/// valence_schema! {
40///     Counter {
41///         table: "counter",
42///         version: "0.1.0",
43///         database: COUNTER_DB,
44///         fields: [
45///             id: { r#type: FieldType::String, primary_key: true, required: true },
46///             value: { r#type: FieldType::Integer, required: true },
47///         ],
48///     }
49/// }
50///
51/// let backend = SqliteBackend::connect_memory().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///     SQLITE_ENGINE_ID
58/// );
59/// # Ok::<(), valence::Error>(())
60/// ```
61#[derive(Debug, Clone)]
62pub struct SqliteBackend {
63    pool: SqlitePool,
64}
65
66impl SqliteBackend {
67    /// Connect to an in-memory SQLite database.
68    ///
69    /// # Errors
70    ///
71    /// Returns an error if the in-memory connection or edges schema setup fails.
72    pub async fn connect_memory() -> Result<Self> {
73        Self::connect(":memory:").await
74    }
75
76    /// Connect to a SQLite database at `path` (`:memory:` for ephemeral).
77    ///
78    /// In-memory URLs (`:memory:` or `mode=memory`) use a single-connection pool so every
79    /// checkout sees the same store. sqlx's default pool size otherwise opens isolated
80    /// private databases per connection for bare `:memory:`.
81    ///
82    /// # Errors
83    ///
84    /// Returns [`Error::Database`] if connecting or ensuring the edges schema fails.
85    pub async fn connect(path: &str) -> Result<Self> {
86        let options = SqliteConnectOptions::from_str(path)
87            .or_else(|_| SqliteConnectOptions::from_str(&format!("sqlite:{path}")))
88            .map_err(|e| Error::database(e.to_string()))?
89            .create_if_missing(true);
90        let memory =
91            path.contains(":memory:") || path.contains("mode=memory") || path == ":memory:";
92        let mut pool_opts = SqlitePoolOptions::new();
93        if memory {
94            pool_opts = pool_opts.max_connections(1);
95        }
96        let pool = pool_opts
97            .connect_with(options)
98            .await
99            .map_err(|e| Error::database(e.to_string()))?;
100        valence_backend_sql::ensure_edges_sqlite(&pool).await?;
101        Ok(Self { pool })
102    }
103
104    /// Borrow the underlying pool.
105    pub fn pool(&self) -> &SqlitePool {
106        &self.pool
107    }
108}
109
110#[async_trait::async_trait]
111impl DatabaseBackend for SqliteBackend {
112    fn engine_id(&self) -> &'static str {
113        ENGINE_ID
114    }
115
116    fn capabilities(&self) -> valence_core::BackendCapabilities {
117        sql_capabilities("sqlite")
118    }
119
120    async fn execute_compiled_query(
121        &self,
122        compiled: &CompiledQuery,
123    ) -> Result<Vec<serde_json::Value>> {
124        execute_select_sqlite(&self.pool, compiled, "").await
125    }
126
127    async fn ensure_schemaless_table(&self, table: &str) -> Result<()> {
128        ensure_table_sqlite(&self.pool, table).await
129    }
130
131    async fn get_record(&self, table: &str, id: &str) -> Result<Option<serde_json::Value>> {
132        get_record_sqlite(&self.pool, table, id).await
133    }
134
135    async fn create_record(
136        &self,
137        table: &str,
138        content: serde_json::Value,
139    ) -> Result<serde_json::Value> {
140        create_record_sqlite(&self.pool, table, content).await
141    }
142
143    async fn update_record(
144        &self,
145        table: &str,
146        id: &str,
147        content: serde_json::Value,
148    ) -> Result<serde_json::Value> {
149        update_record_sqlite(&self.pool, table, id, content).await
150    }
151
152    async fn merge_record(
153        &self,
154        table: &str,
155        id: &str,
156        patch: serde_json::Value,
157    ) -> Result<serde_json::Value> {
158        merge_record_sqlite(&self.pool, table, id, patch).await
159    }
160
161    async fn upsert_record(
162        &self,
163        table: &str,
164        id: &str,
165        content: serde_json::Value,
166    ) -> Result<serde_json::Value> {
167        if self.get_record(table, id).await?.is_some() {
168            self.update_record(table, id, content).await
169        } else {
170            let mut c = content;
171            if let Some(obj) = c.as_object_mut() {
172                obj.insert("id".into(), serde_json::json!({"table": table, "id": id}));
173            }
174            self.create_record(table, c).await
175        }
176    }
177
178    async fn delete_record(&self, table: &str, id: &str) -> Result<()> {
179        delete_record_sqlite(&self.pool, table, id).await
180    }
181
182    async fn relate_edge(&self, from: &RecordId, edge_table: &str, to: &RecordId) -> Result<()> {
183        relate_edge_sqlite(&self.pool, from, edge_table, to).await
184    }
185
186    async fn unrelate_edge(&self, from: &RecordId, edge_table: &str, to: &RecordId) -> Result<()> {
187        unrelate_edge_sqlite(&self.pool, from, edge_table, to).await
188    }
189
190    async fn get_edge_targets(&self, from: &RecordId, edge_table: &str) -> Result<Vec<RecordId>> {
191        get_edge_targets_sqlite(&self.pool, from, edge_table).await
192    }
193
194    async fn get_edge_sources(&self, to: &RecordId, edge_table: &str) -> Result<Vec<RecordId>> {
195        get_edge_sources_sqlite(&self.pool, to, edge_table).await
196    }
197
198    async fn define_unique_index(&self, table: &str, field: &str) -> Result<()> {
199        define_unique_index_sqlite(&self.pool, table, field).await
200    }
201
202    fn ttl_capability(&self) -> valence_core::ttl::BackendTtlCapability {
203        ttl_deferred()
204    }
205
206    async fn apply_ttl_policy(&self, table: &str, policy: &SchemaTtlPolicy) -> Result<()> {
207        apply_ttl_policy_sqlite(&self.pool, table, policy).await
208    }
209}
valence_backend_sqlite/backend.rs

valence_backend_sqlite/
backend.rs
