assay_auth/store/
sqlite.rs

1//! SQLite implementations of [`UserStore`] and [`SessionStore`].
2//!
3//! Tables live in the attached `auth` database (engine boot ATTACHes
4//! `data/auth.db` AS `auth` before [`crate::schema::migrate_sqlite`]
5//! creates them). Queries are schema-qualified (`auth.users`, …) so the
6//! syntax matches the PG store exactly — the `auth.` prefix resolves
7//! against the ATTACH alias on SQLite and against the schema on PG.
8
9use std::sync::Arc;
10
11use anyhow::{Context, Result};
12use sqlx::{Row, SqlitePool};
13
14use super::types::{PasskeyCred, Session, User};
15use super::{SessionStore, UserStore};
16
17/// User store backed by a shared `SqlitePool`. Mirrors
18/// [`super::postgres::PostgresUserStore`] in shape so callers swap one
19/// for the other based on the engine's selected backend.
20#[derive(Clone)]
21pub struct SqliteUserStore {
22    pool: SqlitePool,
23}
24
25impl SqliteUserStore {
26    pub fn new(pool: SqlitePool) -> Self {
27        Self { pool }
28    }
29
30    /// Wrap into an `Arc<dyn UserStore>` for [`crate::ctx::AuthCtx`].
31    pub fn into_dyn(self) -> Arc<dyn UserStore> {
32        Arc::new(self)
33    }
34}
35
36#[async_trait::async_trait]
37impl UserStore for SqliteUserStore {
38    async fn create_user(&self, user: &User) -> Result<()> {
39        sqlx::query(
40            "INSERT INTO auth.users
41                 (id, email, email_verified, display_name, password_hash, created_at)
42             VALUES (?, ?, ?, ?, NULL, ?)",
43        )
44        .bind(&user.id)
45        .bind(&user.email)
46        .bind(if user.email_verified { 1i64 } else { 0i64 })
47        .bind(&user.display_name)
48        .bind(user.created_at)
49        .execute(&self.pool)
50        .await
51        .context("auth.users insert")?;
52        Ok(())
53    }
54
55    async fn get_user_by_id(&self, id: &str) -> Result<Option<User>> {
56        let row = sqlx::query(
57            "SELECT id, email, email_verified, display_name, created_at
58             FROM auth.users WHERE id = ?",
59        )
60        .bind(id)
61        .fetch_optional(&self.pool)
62        .await
63        .context("auth.users select by id")?;
64        Ok(row.map(map_user_row_sqlite))
65    }
66
67    async fn get_user_by_email(&self, email: &str) -> Result<Option<User>> {
68        let row = sqlx::query(
69            "SELECT id, email, email_verified, display_name, created_at
70             FROM auth.users WHERE email = ?",
71        )
72        .bind(email)
73        .fetch_optional(&self.pool)
74        .await
75        .context("auth.users select by email")?;
76        Ok(row.map(map_user_row_sqlite))
77    }
78
79    async fn update_user(&self, user: &User) -> Result<()> {
80        sqlx::query(
81            "UPDATE auth.users
82             SET email = ?,
83                 email_verified = ?,
84                 display_name = ?
85             WHERE id = ?",
86        )
87        .bind(&user.email)
88        .bind(if user.email_verified { 1i64 } else { 0i64 })
89        .bind(&user.display_name)
90        .bind(&user.id)
91        .execute(&self.pool)
92        .await
93        .context("auth.users update")?;
94        Ok(())
95    }
96
97    async fn set_password_hash(&self, user_id: &str, hash: &str) -> Result<()> {
98        sqlx::query("UPDATE auth.users SET password_hash = ? WHERE id = ?")
99            .bind(hash)
100            .bind(user_id)
101            .execute(&self.pool)
102            .await
103            .context("auth.users set password_hash")?;
104        Ok(())
105    }
106
107    async fn get_password_hash(&self, user_id: &str) -> Result<Option<String>> {
108        let row: Option<(Option<String>,)> =
109            sqlx::query_as("SELECT password_hash FROM auth.users WHERE id = ?")
110                .bind(user_id)
111                .fetch_optional(&self.pool)
112                .await
113                .context("auth.users select password_hash")?;
114        Ok(row.and_then(|r| r.0))
115    }
116
117    async fn list_passkeys(&self, user_id: &str) -> Result<Vec<PasskeyCred>> {
118        let rows = sqlx::query(
119            "SELECT credential_id, public_key, sign_count, transports, created_at
120             FROM auth.passkeys WHERE user_id = ?
121             ORDER BY created_at",
122        )
123        .bind(user_id)
124        .fetch_all(&self.pool)
125        .await
126        .context("auth.passkeys list")?;
127        Ok(rows.into_iter().map(map_passkey_row_sqlite).collect())
128    }
129
130    async fn add_passkey(&self, user_id: &str, cred: &PasskeyCred) -> Result<()> {
131        sqlx::query(
132            "INSERT INTO auth.passkeys
133                 (credential_id, user_id, public_key, sign_count, transports, created_at)
134             VALUES (?, ?, ?, ?, ?, ?)",
135        )
136        .bind(&cred.credential_id)
137        .bind(user_id)
138        .bind(&cred.public_key)
139        .bind(cred.sign_count as i64)
140        .bind(cred.transports.join(","))
141        .bind(cred.created_at)
142        .execute(&self.pool)
143        .await
144        .context("auth.passkeys insert")?;
145        Ok(())
146    }
147
148    async fn remove_passkey(&self, credential_id: &[u8]) -> Result<bool> {
149        let res = sqlx::query("DELETE FROM auth.passkeys WHERE credential_id = ?")
150            .bind(credential_id)
151            .execute(&self.pool)
152            .await
153            .context("auth.passkeys delete")?;
154        Ok(res.rows_affected() > 0)
155    }
156
157    async fn link_upstream(&self, user_id: &str, provider: &str, subject: &str) -> Result<()> {
158        sqlx::query(
159            "INSERT INTO auth.user_upstream (provider, subject, user_id)
160             VALUES (?, ?, ?)
161             ON CONFLICT (provider, subject) DO UPDATE SET user_id = excluded.user_id",
162        )
163        .bind(provider)
164        .bind(subject)
165        .bind(user_id)
166        .execute(&self.pool)
167        .await
168        .context("auth.user_upstream upsert")?;
169        Ok(())
170    }
171
172    async fn get_user_by_upstream(&self, provider: &str, subject: &str) -> Result<Option<User>> {
173        let row = sqlx::query(
174            "SELECT u.id, u.email, u.email_verified, u.display_name, u.created_at
175             FROM auth.users u
176             JOIN auth.user_upstream l ON l.user_id = u.id
177             WHERE l.provider = ? AND l.subject = ?",
178        )
179        .bind(provider)
180        .bind(subject)
181        .fetch_optional(&self.pool)
182        .await
183        .context("auth.user_upstream lookup")?;
184        Ok(row.map(map_user_row_sqlite))
185    }
186
187    async fn list_users(&self, limit: i64, offset: i64, search: Option<&str>) -> Result<Vec<User>> {
188        let lim = limit.clamp(1, 500);
189        let off = offset.max(0);
190        let rows = if let Some(needle) = search {
191            let pat = format!("%{}%", needle.to_lowercase());
192            sqlx::query(
193                "SELECT id, email, email_verified, display_name, created_at
194                 FROM auth.users
195                 WHERE LOWER(COALESCE(email, '')) LIKE ?
196                    OR LOWER(COALESCE(display_name, '')) LIKE ?
197                 ORDER BY created_at DESC
198                 LIMIT ? OFFSET ?",
199            )
200            .bind(&pat)
201            .bind(&pat)
202            .bind(lim)
203            .bind(off)
204            .fetch_all(&self.pool)
205            .await
206            .context("auth.users list (search)")?
207        } else {
208            sqlx::query(
209                "SELECT id, email, email_verified, display_name, created_at
210                 FROM auth.users
211                 ORDER BY created_at DESC
212                 LIMIT ? OFFSET ?",
213            )
214            .bind(lim)
215            .bind(off)
216            .fetch_all(&self.pool)
217            .await
218            .context("auth.users list")?
219        };
220        Ok(rows.into_iter().map(map_user_row_sqlite).collect())
221    }
222
223    async fn count_users(&self, search: Option<&str>) -> Result<i64> {
224        let row: (i64,) = if let Some(needle) = search {
225            let pat = format!("%{}%", needle.to_lowercase());
226            sqlx::query_as(
227                "SELECT COUNT(*) FROM auth.users
228                 WHERE LOWER(COALESCE(email, '')) LIKE ?
229                    OR LOWER(COALESCE(display_name, '')) LIKE ?",
230            )
231            .bind(&pat)
232            .bind(&pat)
233            .fetch_one(&self.pool)
234            .await
235            .context("auth.users count (search)")?
236        } else {
237            sqlx::query_as("SELECT COUNT(*) FROM auth.users")
238                .fetch_one(&self.pool)
239                .await
240                .context("auth.users count")?
241        };
242        Ok(row.0)
243    }
244
245    async fn delete_user(&self, id: &str) -> Result<bool> {
246        let res = sqlx::query("DELETE FROM auth.users WHERE id = ?")
247            .bind(id)
248            .execute(&self.pool)
249            .await
250            .context("auth.users delete")?;
251        Ok(res.rows_affected() > 0)
252    }
253
254    async fn list_upstream_for_user(&self, user_id: &str) -> Result<Vec<(String, String)>> {
255        let rows = sqlx::query(
256            "SELECT provider, subject FROM auth.user_upstream
257             WHERE user_id = ? ORDER BY provider, subject",
258        )
259        .bind(user_id)
260        .fetch_all(&self.pool)
261        .await
262        .context("auth.user_upstream list")?;
263        Ok(rows
264            .into_iter()
265            .map(|r| {
266                (
267                    r.get::<String, _>("provider"),
268                    r.get::<String, _>("subject"),
269                )
270            })
271            .collect())
272    }
273}
274
275/// Session store backed by `auth.sessions`. Independent struct from
276/// [`SqliteUserStore`] because they're independently mockable in
277/// tests and the engine may swap one without the other.
278#[derive(Clone)]
279pub struct SqliteSessionStore {
280    pool: SqlitePool,
281}
282
283impl SqliteSessionStore {
284    pub fn new(pool: SqlitePool) -> Self {
285        Self { pool }
286    }
287
288    pub fn into_dyn(self) -> Arc<dyn SessionStore> {
289        Arc::new(self)
290    }
291}
292
293#[async_trait::async_trait]
294impl SessionStore for SqliteSessionStore {
295    async fn create(&self, session: &Session) -> Result<()> {
296        sqlx::query(
297            "INSERT INTO auth.sessions
298                 (id, user_id, csrf_token, created_at, expires_at, ip_hash, user_agent_hash)
299             VALUES (?, ?, ?, ?, ?, ?, ?)",
300        )
301        .bind(&session.id)
302        .bind(&session.user_id)
303        .bind(&session.csrf_token)
304        .bind(session.created_at)
305        .bind(session.expires_at)
306        .bind(&session.ip_hash)
307        .bind(&session.user_agent_hash)
308        .execute(&self.pool)
309        .await
310        .context("auth.sessions insert")?;
311        Ok(())
312    }
313
314    async fn get(&self, id: &str) -> Result<Option<Session>> {
315        let row = sqlx::query(
316            "SELECT id, user_id, csrf_token, created_at, expires_at, ip_hash, user_agent_hash
317             FROM auth.sessions WHERE id = ?",
318        )
319        .bind(id)
320        .fetch_optional(&self.pool)
321        .await
322        .context("auth.sessions select")?;
323        Ok(row.map(map_session_row_sqlite))
324    }
325
326    async fn delete(&self, id: &str) -> Result<bool> {
327        let res = sqlx::query("DELETE FROM auth.sessions WHERE id = ?")
328            .bind(id)
329            .execute(&self.pool)
330            .await
331            .context("auth.sessions delete")?;
332        Ok(res.rows_affected() > 0)
333    }
334
335    async fn list_for_user(&self, user_id: &str) -> Result<Vec<Session>> {
336        let rows = sqlx::query(
337            "SELECT id, user_id, csrf_token, created_at, expires_at, ip_hash, user_agent_hash
338             FROM auth.sessions WHERE user_id = ? ORDER BY created_at DESC",
339        )
340        .bind(user_id)
341        .fetch_all(&self.pool)
342        .await
343        .context("auth.sessions list_for_user")?;
344        Ok(rows.into_iter().map(map_session_row_sqlite).collect())
345    }
346
347    async fn delete_for_user(&self, user_id: &str) -> Result<u64> {
348        let res = sqlx::query("DELETE FROM auth.sessions WHERE user_id = ?")
349            .bind(user_id)
350            .execute(&self.pool)
351            .await
352            .context("auth.sessions delete_for_user")?;
353        Ok(res.rows_affected())
354    }
355
356    async fn purge_expired(&self, now: f64) -> Result<u64> {
357        let res = sqlx::query("DELETE FROM auth.sessions WHERE expires_at <= ?")
358            .bind(now)
359            .execute(&self.pool)
360            .await
361            .context("auth.sessions purge_expired")?;
362        Ok(res.rows_affected())
363    }
364
365    async fn list_all(
366        &self,
367        limit: i64,
368        offset: i64,
369        user_filter: Option<&str>,
370    ) -> Result<Vec<Session>> {
371        let lim = limit.clamp(1, 500);
372        let off = offset.max(0);
373        let rows = if let Some(uid) = user_filter {
374            sqlx::query(
375                "SELECT id, user_id, csrf_token, created_at, expires_at, ip_hash, user_agent_hash
376                 FROM auth.sessions WHERE user_id = ?
377                 ORDER BY created_at DESC
378                 LIMIT ? OFFSET ?",
379            )
380            .bind(uid)
381            .bind(lim)
382            .bind(off)
383            .fetch_all(&self.pool)
384            .await
385            .context("auth.sessions list_all (user filter)")?
386        } else {
387            sqlx::query(
388                "SELECT id, user_id, csrf_token, created_at, expires_at, ip_hash, user_agent_hash
389                 FROM auth.sessions
390                 ORDER BY created_at DESC
391                 LIMIT ? OFFSET ?",
392            )
393            .bind(lim)
394            .bind(off)
395            .fetch_all(&self.pool)
396            .await
397            .context("auth.sessions list_all")?
398        };
399        Ok(rows.into_iter().map(map_session_row_sqlite).collect())
400    }
401
402    async fn count_all(&self, user_filter: Option<&str>) -> Result<i64> {
403        let row: (i64,) = if let Some(uid) = user_filter {
404            sqlx::query_as("SELECT COUNT(*) FROM auth.sessions WHERE user_id = ?")
405                .bind(uid)
406                .fetch_one(&self.pool)
407                .await
408                .context("auth.sessions count_all (user filter)")?
409        } else {
410            sqlx::query_as("SELECT COUNT(*) FROM auth.sessions")
411                .fetch_one(&self.pool)
412                .await
413                .context("auth.sessions count_all")?
414        };
415        Ok(row.0)
416    }
417}
418
419fn map_user_row_sqlite(row: sqlx::sqlite::SqliteRow) -> User {
420    let email_verified: i64 = row.get("email_verified");
421    User {
422        id: row.get("id"),
423        email: row.get("email"),
424        email_verified: email_verified != 0,
425        display_name: row.get("display_name"),
426        created_at: row.get("created_at"),
427    }
428}
429
430fn map_session_row_sqlite(row: sqlx::sqlite::SqliteRow) -> Session {
431    Session {
432        id: row.get("id"),
433        user_id: row.get("user_id"),
434        csrf_token: row.get("csrf_token"),
435        created_at: row.get("created_at"),
436        expires_at: row.get("expires_at"),
437        ip_hash: row.get("ip_hash"),
438        user_agent_hash: row.get("user_agent_hash"),
439    }
440}
441
442fn map_passkey_row_sqlite(row: sqlx::sqlite::SqliteRow) -> PasskeyCred {
443    let transports: String = row.get("transports");
444    let sign_count: i64 = row.get("sign_count");
445    PasskeyCred {
446        credential_id: row.get("credential_id"),
447        public_key: row.get("public_key"),
448        sign_count: sign_count.max(0) as u32,
449        transports: if transports.is_empty() {
450            Vec::new()
451        } else {
452            transports.split(',').map(|s| s.to_string()).collect()
453        },
454        created_at: row.get("created_at"),
455    }
456}
assay_auth/store/sqlite.rs

assay_auth/store/
sqlite.rs
