yugendb_sqlite/
driver.rs

1//! SQLite driver implementation.
2//!
3//! SQLite stores yugendb records through the shared storage-table model:
4//! `namespace + collection + key -> serialised typed value`.
5//! Applications use the same yugendb store and collection API across backends.
6
7use async_trait::async_trait;
8use chrono::{DateTime, Utc};
9use parking_lot::Mutex;
10use rusqlite::{params, Connection, OptionalExtension};
11use yugendb_core::{
12    Batch, BatchOperation, BatchResult, Capabilities, CollectionName, DeleteOptions, Driver, Key,
13    Namespace, ReadOptions, Result, ScanOptions, StoredValue, ValueBytes, ValueMetadata,
14    WriteOptions, YugenDbError,
15};
16
17use crate::schema::{schema_statements, SqliteStorageSchema};
18
19/// Options used to configure the SQLite driver.
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub struct SqliteDriverOptions {
22    /// Filesystem path or SQLite connection target.
23    pub path: String,
24    /// Whether the database file should be created if it is missing.
25    pub create_if_missing: bool,
26}
27
28impl SqliteDriverOptions {
29    /// Creates default SQLite options for a path.
30    #[must_use]
31    pub fn new(path: impl Into<String>) -> Self {
32        Self {
33            path: path.into(),
34            create_if_missing: true,
35        }
36    }
37}
38
39/// SQLite driver for durable local storage.
40#[derive(Debug)]
41pub struct SqliteDriver {
42    options: SqliteDriverOptions,
43    connection: Mutex<Option<Connection>>,
44}
45
46impl SqliteDriver {
47    /// Creates a new SQLite driver using default options.
48    #[must_use]
49    pub fn new(path: impl Into<String>) -> Self {
50        Self::with_options(SqliteDriverOptions::new(path))
51    }
52
53    /// Creates a new SQLite driver with explicit options.
54    #[must_use]
55    pub fn with_options(options: SqliteDriverOptions) -> Self {
56        Self {
57            options,
58            connection: Mutex::new(None),
59        }
60    }
61
62    /// Returns the configured options.
63    #[must_use]
64    pub const fn options(&self) -> &SqliteDriverOptions {
65        &self.options
66    }
67
68    /// Returns the SQLite storage schema description.
69    #[must_use]
70    pub const fn schema(&self) -> SqliteStorageSchema {
71        SqliteStorageSchema::current()
72    }
73
74    fn map_error(error: rusqlite::Error) -> YugenDbError {
75        YugenDbError::driver_error("sqlite", error.to_string())
76    }
77
78    fn with_connection<T>(&self, operation: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
79        let guard = self.connection.lock();
80        let Some(connection) = guard.as_ref() else {
81            return Err(YugenDbError::connection_error(
82                "SQLite driver has not been initialised",
83            ));
84        };
85        operation(connection)
86    }
87
88    fn parse_timestamp(value: String) -> Result<DateTime<Utc>> {
89        DateTime::parse_from_rfc3339(&value)
90            .map(|value| value.with_timezone(&Utc))
91            .map_err(|error| YugenDbError::driver_error("sqlite", error.to_string()))
92    }
93
94    fn parse_optional_timestamp(value: Option<String>) -> Result<Option<DateTime<Utc>>> {
95        value.map(Self::parse_timestamp).transpose()
96    }
97
98    fn stored_value_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<StoredValue> {
99        let bytes: Vec<u8> = row.get("value")?;
100        let codec: String = row.get("codec")?;
101        let created_at: String = row.get("created_at")?;
102        let updated_at: String = row.get("updated_at")?;
103        let expires_at: Option<String> = row.get("expires_at")?;
104
105        let created_at = Self::parse_timestamp(created_at).map_err(|error| {
106            rusqlite::Error::FromSqlConversionFailure(
107                0,
108                rusqlite::types::Type::Text,
109                Box::new(error),
110            )
111        })?;
112        let updated_at = Self::parse_timestamp(updated_at).map_err(|error| {
113            rusqlite::Error::FromSqlConversionFailure(
114                0,
115                rusqlite::types::Type::Text,
116                Box::new(error),
117            )
118        })?;
119        let expires_at = Self::parse_optional_timestamp(expires_at).map_err(|error| {
120            rusqlite::Error::FromSqlConversionFailure(
121                0,
122                rusqlite::types::Type::Text,
123                Box::new(error),
124            )
125        })?;
126
127        Ok(StoredValue::new(
128            ValueBytes::from(bytes),
129            ValueMetadata {
130                codec,
131                created_at,
132                updated_at,
133                expires_at,
134            },
135        ))
136    }
137
138    fn insert_value(
139        connection: &Connection,
140        namespace: &Namespace,
141        collection: &CollectionName,
142        key: &Key,
143        value: &StoredValue,
144    ) -> Result<()> {
145        connection
146            .execute(
147                r#"INSERT INTO yugendb_store
148(namespace, collection, key, value, codec, created_at, updated_at, expires_at)
149VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
150ON CONFLICT(namespace, collection, key) DO UPDATE SET
151    value = excluded.value,
152    codec = excluded.codec,
153    updated_at = excluded.updated_at,
154    expires_at = excluded.expires_at"#,
155                params![
156                    namespace.as_str(),
157                    collection.as_str(),
158                    key.as_str(),
159                    value.bytes.as_slice(),
160                    value.metadata.codec.as_str(),
161                    value.metadata.created_at.to_rfc3339(),
162                    value.metadata.updated_at.to_rfc3339(),
163                    value
164                        .metadata
165                        .expires_at
166                        .as_ref()
167                        .map(|expires_at| expires_at.to_rfc3339()),
168                ],
169            )
170            .map_err(Self::map_error)?;
171        Ok(())
172    }
173}
174
175impl Clone for SqliteDriver {
176    fn clone(&self) -> Self {
177        // Cloned handles are configured the same way but own a separate lazy connection.
178        Self::with_options(self.options.clone())
179    }
180}
181
182/// Capabilities for the implemented SQLite driver.
183#[must_use]
184pub const fn sqlite_capabilities() -> Capabilities {
185    Capabilities {
186        transactions: true,
187        ttl: true,
188        prefix_scan: true,
189        atomic_increment: false,
190        batch_write: true,
191        raw_sql: false,
192        document_query: false,
193        json_query: false,
194        migrations: true,
195        connection_pooling: false,
196        watch: false,
197        backup: false,
198    }
199}
200
201
202/// Convenience constructor for the SQLite driver.
203#[must_use]
204pub fn sqlite(path: impl Into<String>) -> SqliteDriver {
205    SqliteDriver::new(path)
206}
207
208#[async_trait]
209impl Driver for SqliteDriver {
210    fn name(&self) -> &'static str {
211        "sqlite"
212    }
213
214    fn capabilities(&self) -> Capabilities {
215        sqlite_capabilities()
216    }
217
218    async fn initialise(&self) -> Result<()> {
219        let mut guard = self.connection.lock();
220        if guard.is_some() {
221            return Ok(());
222        }
223
224        let connection = if self.options.create_if_missing {
225            Connection::open(&self.options.path)
226        } else {
227            Connection::open_with_flags(
228                &self.options.path,
229                rusqlite::OpenFlags::SQLITE_OPEN_READ_WRITE,
230            )
231        }
232        .map_err(Self::map_error)?;
233
234        for statement in schema_statements() {
235            connection
236                .execute_batch(statement)
237                .map_err(Self::map_error)?;
238        }
239
240        *guard = Some(connection);
241        Ok(())
242    }
243
244    async fn finalise(&self) -> Result<()> {
245        let mut guard = self.connection.lock();
246        *guard = None;
247        Ok(())
248    }
249
250    async fn get(
251        &self,
252        namespace: &Namespace,
253        collection: &CollectionName,
254        key: &Key,
255        options: &ReadOptions,
256    ) -> Result<Option<StoredValue>> {
257        self.with_connection(|connection| {
258            let value = connection
259                .query_row(
260                    "SELECT value, codec, created_at, updated_at, expires_at FROM yugendb_store WHERE namespace = ?1 AND collection = ?2 AND key = ?3",
261                    params![namespace.as_str(), collection.as_str(), key.as_str()],
262                    Self::stored_value_from_row,
263                )
264                .optional()
265                .map_err(Self::map_error)?;
266
267            Ok(value.filter(|value| options.allow_expired || !value.is_expired()))
268        })
269    }
270
271    async fn set(
272        &self,
273        namespace: &Namespace,
274        collection: &CollectionName,
275        key: &Key,
276        value: StoredValue,
277        options: &WriteOptions,
278    ) -> Result<()> {
279        self.with_connection(|connection| {
280            if !options.overwrite {
281                let existing = connection
282                    .query_row(
283                        "SELECT value, codec, created_at, updated_at, expires_at FROM yugendb_store WHERE namespace = ?1 AND collection = ?2 AND key = ?3",
284                        params![namespace.as_str(), collection.as_str(), key.as_str()],
285                        Self::stored_value_from_row,
286                    )
287                    .optional()
288                    .map_err(Self::map_error)?;
289
290                if existing.is_some_and(|existing| !existing.is_expired()) {
291                    return Err(YugenDbError::conflict(
292                        "SQLite driver refused to overwrite an existing value",
293                    ));
294                }
295            }
296
297            Self::insert_value(connection, namespace, collection, key, &value)
298        })
299    }
300
301    async fn delete(
302        &self,
303        namespace: &Namespace,
304        collection: &CollectionName,
305        key: &Key,
306        options: &DeleteOptions,
307    ) -> Result<bool> {
308        self.with_connection(|connection| {
309            let removed = connection
310                .execute(
311                    "DELETE FROM yugendb_store WHERE namespace = ?1 AND collection = ?2 AND key = ?3",
312                    params![namespace.as_str(), collection.as_str(), key.as_str()],
313                )
314                .map_err(Self::map_error)?
315                > 0;
316
317            if options.must_exist && !removed {
318                return Err(YugenDbError::not_found(
319                    "SQLite driver could not delete a missing value",
320                ));
321            }
322
323            Ok(removed)
324        })
325    }
326
327    async fn exists(
328        &self,
329        namespace: &Namespace,
330        collection: &CollectionName,
331        key: &Key,
332    ) -> Result<bool> {
333        Ok(self
334            .get(namespace, collection, key, &ReadOptions::default())
335            .await?
336            .is_some())
337    }
338
339    async fn scan_prefix(
340        &self,
341        namespace: &Namespace,
342        collection: &CollectionName,
343        options: &ScanOptions,
344    ) -> Result<Vec<(Key, StoredValue)>> {
345        self.with_connection(|connection| {
346            let prefix = options.prefix.as_deref().unwrap_or("");
347            let mut statement = connection
348                .prepare(
349                    "SELECT key, value, codec, created_at, updated_at, expires_at FROM yugendb_store WHERE namespace = ?1 AND collection = ?2 AND key LIKE ?3 ORDER BY key ASC",
350                )
351                .map_err(Self::map_error)?;
352            let rows = statement
353                .query_map(
354                    params![namespace.as_str(), collection.as_str(), format!("{prefix}%")],
355                    |row| {
356                        let key: String = row.get("key")?;
357                        let value = Self::stored_value_from_row(row)?;
358                        Ok((key, value))
359                    },
360                )
361                .map_err(Self::map_error)?;
362
363            let mut values = Vec::new();
364            for row in rows {
365                let (key, value) = row.map_err(Self::map_error)?;
366                if value.is_expired() && !options.include_expired {
367                    continue;
368                }
369                values.push((Key::try_from(key)?, value));
370                if let Some(limit) = options.limit {
371                    if values.len() >= limit {
372                        break;
373                    }
374                }
375            }
376            Ok(values)
377        })
378    }
379
380    async fn batch(&self, batch: Batch) -> Result<BatchResult> {
381        self.with_connection(|connection| {
382            let transaction = connection.unchecked_transaction().map_err(Self::map_error)?;
383            let mut set_count = 0;
384            let mut delete_count = 0;
385
386            for operation in batch.into_operations() {
387                match operation {
388                    BatchOperation::Set {
389                        namespace,
390                        collection,
391                        key,
392                        value,
393                    } => {
394                        Self::insert_value(&transaction, &namespace, &collection, &key, &value)?;
395                        set_count += 1;
396                    }
397                    BatchOperation::Delete {
398                        namespace,
399                        collection,
400                        key,
401                    } => {
402                        let removed = transaction
403                            .execute(
404                                "DELETE FROM yugendb_store WHERE namespace = ?1 AND collection = ?2 AND key = ?3",
405                                params![namespace.as_str(), collection.as_str(), key.as_str()],
406                            )
407                            .map_err(Self::map_error)?;
408                        if removed > 0 {
409                            delete_count += 1;
410                        }
411                    }
412                }
413            }
414
415            transaction.commit().map_err(Self::map_error)?;
416            Ok(BatchResult::new(set_count, delete_count))
417        })
418    }
419}
yugendb_sqlite/driver.rs

yugendb_sqlite/
driver.rs
