origin_storage_sqlite/
lib.rs

1//! SQLite implementation of [`origin_storage::Storage`].
2//!
3//! SQLite holds cache, read models, sync metadata and settings — never credentials
4//! (ADR-0008). Deleting the database file must cost the user nothing but a resync.
5//!
6//! `rusqlite` is blocking, so every call runs on the blocking pool rather than
7//! stalling the async runtime.
8
9mod schema;
10
11use async_trait::async_trait;
12use origin_domain::{AppError, Result};
13use origin_storage::{Record, Storage, StorageKey};
14use rusqlite::{Connection, OptionalExtension, params};
15use std::path::Path;
16use std::sync::{Arc, Mutex};
17use time::OffsetDateTime;
18use time::format_description::well_known::Rfc3339;
19
20#[derive(Debug, Clone)]
21pub struct SqliteStorage {
22    connection: Arc<Mutex<Connection>>,
23}
24
25impl SqliteStorage {
26    /// Open (and create if needed) the database at `path`.
27    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
28        let connection = Connection::open(path.as_ref()).map_err(to_storage_error)?;
29        Self::from_connection(connection)
30    }
31
32    /// A private in-memory database. Useful for tests and for a `--no-persist` run.
33    pub fn in_memory() -> Result<Self> {
34        let connection = Connection::open_in_memory().map_err(to_storage_error)?;
35        Self::from_connection(connection)
36    }
37
38    fn from_connection(connection: Connection) -> Result<Self> {
39        schema::apply(&connection)?;
40        Ok(Self {
41            connection: Arc::new(Mutex::new(connection)),
42        })
43    }
44
45    /// Delete every record whose expiry has passed.
46    ///
47    /// Reads already ignore expired records; this only reclaims disk space, so it is
48    /// safe to call on a schedule or never.
49    pub async fn prune_expired(&self, now: OffsetDateTime) -> Result<usize> {
50        let now = encode_time(now)?;
51        self.with_connection(move |connection| {
52            let removed = connection
53                .execute(
54                    "DELETE FROM records WHERE expires_at IS NOT NULL AND expires_at <= ?1",
55                    params![now],
56                )
57                .map_err(to_storage_error)?;
58            Ok(removed)
59        })
60        .await
61    }
62
63    /// Run a blocking database operation on the blocking pool.
64    async fn with_connection<T, F>(&self, operation: F) -> Result<T>
65    where
66        T: Send + 'static,
67        F: FnOnce(&Connection) -> Result<T> + Send + 'static,
68    {
69        let connection = self.connection.clone();
70        tokio::task::spawn_blocking(move || {
71            let guard = connection
72                .lock()
73                .map_err(|_| AppError::storage("sqlite connection poisoned"))?;
74            operation(&guard)
75        })
76        .await
77        .map_err(|error| AppError::storage(format!("storage task failed: {error}")))?
78    }
79}
80
81#[async_trait]
82impl Storage for SqliteStorage {
83    async fn get(&self, key: &StorageKey) -> Result<Option<Record>> {
84        let (namespace, name) = split(key);
85        self.with_connection(move |connection| {
86            connection
87                .query_row(
88                    "SELECT value, stored_at, expires_at FROM records \
89                     WHERE namespace = ?1 AND key = ?2",
90                    params![namespace, name],
91                    |row| {
92                        Ok((
93                            row.get::<_, String>(0)?,
94                            row.get::<_, String>(1)?,
95                            row.get::<_, Option<String>>(2)?,
96                        ))
97                    },
98                )
99                .optional()
100                .map_err(to_storage_error)?
101                .map(|(value, stored_at, expires_at)| {
102                    let mut record = Record::new(value, decode_time(&stored_at)?);
103                    record.expires_at = expires_at.as_deref().map(decode_time).transpose()?;
104                    Ok(record)
105                })
106                .transpose()
107        })
108        .await
109    }
110
111    async fn put(&self, key: &StorageKey, record: Record) -> Result<()> {
112        let (namespace, name) = split(key);
113        let stored_at = encode_time(record.stored_at)?;
114        let expires_at = record.expires_at.map(encode_time).transpose()?;
115        let value = record.value;
116
117        self.with_connection(move |connection| {
118            connection
119                .execute(
120                    "INSERT INTO records (namespace, key, value, stored_at, expires_at) \
121                     VALUES (?1, ?2, ?3, ?4, ?5) \
122                     ON CONFLICT(namespace, key) DO UPDATE SET \
123                       value = excluded.value, \
124                       stored_at = excluded.stored_at, \
125                       expires_at = excluded.expires_at",
126                    params![namespace, name, value, stored_at, expires_at],
127                )
128                .map_err(to_storage_error)?;
129            Ok(())
130        })
131        .await
132    }
133
134    async fn delete(&self, key: &StorageKey) -> Result<()> {
135        let (namespace, name) = split(key);
136        self.with_connection(move |connection| {
137            connection
138                .execute(
139                    "DELETE FROM records WHERE namespace = ?1 AND key = ?2",
140                    params![namespace, name],
141                )
142                .map_err(to_storage_error)?;
143            Ok(())
144        })
145        .await
146    }
147
148    async fn keys(&self, namespace: &str) -> Result<Vec<StorageKey>> {
149        let namespace = namespace.to_owned();
150        self.with_connection(move |connection| {
151            let mut statement = connection
152                .prepare("SELECT key FROM records WHERE namespace = ?1")
153                .map_err(to_storage_error)?;
154
155            let keys = statement
156                .query_map(params![namespace], |row| row.get::<_, String>(0))
157                .map_err(to_storage_error)?
158                .collect::<std::result::Result<Vec<_>, _>>()
159                .map_err(to_storage_error)?
160                .into_iter()
161                .map(|key| StorageKey::new(&namespace, key))
162                .collect();
163
164            Ok(keys)
165        })
166        .await
167    }
168
169    async fn clear(&self, namespace: &str) -> Result<()> {
170        let namespace = namespace.to_owned();
171        self.with_connection(move |connection| {
172            connection
173                .execute(
174                    "DELETE FROM records WHERE namespace = ?1",
175                    params![namespace],
176                )
177                .map_err(to_storage_error)?;
178            Ok(())
179        })
180        .await
181    }
182
183    async fn clear_prefix(&self, prefix: &str) -> Result<usize> {
184        // GLOB rather than LIKE: LIKE would treat `_` and `%` in a connector or
185        // account id as wildcards, so `acct.gh_x.` would also delete `acct.ghax.`.
186        let pattern = format!("{}*", glob_escape(prefix));
187
188        self.with_connection(move |connection| {
189            let removed = connection
190                .execute(
191                    "DELETE FROM records WHERE namespace GLOB ?1",
192                    params![pattern],
193                )
194                .map_err(to_storage_error)?;
195            Ok(removed)
196        })
197        .await
198    }
199}
200
201/// Escape the characters GLOB treats specially, so a prefix is matched literally.
202fn glob_escape(value: &str) -> String {
203    let mut escaped = String::with_capacity(value.len());
204    for character in value.chars() {
205        match character {
206            '*' | '?' | '[' | ']' => {
207                escaped.push('[');
208                escaped.push(character);
209                escaped.push(']');
210            }
211            other => escaped.push(other),
212        }
213    }
214    escaped
215}
216
217fn split(key: &StorageKey) -> (String, String) {
218    (key.namespace().to_owned(), key.key().to_owned())
219}
220
221fn encode_time(value: OffsetDateTime) -> Result<String> {
222    value
223        .format(&Rfc3339)
224        .map_err(|error| AppError::storage(format!("cannot format timestamp: {error}")))
225}
226
227fn decode_time(value: &str) -> Result<OffsetDateTime> {
228    OffsetDateTime::parse(value, &Rfc3339)
229        .map_err(|error| AppError::storage(format!("cannot parse timestamp {value:?}: {error}")))
230}
231
232/// The single place where a `rusqlite` error becomes an Origin error. Nothing
233/// database-specific travels further up (ADR-0002).
234fn to_storage_error(error: rusqlite::Error) -> AppError {
235    AppError::storage(error.to_string())
236}
origin_storage_sqlite/lib.rs

origin_storage_sqlite/
lib.rs
