ling_http/
db.rs

1use r2d2::Pool;
2use std::path::Path;
3
4use crate::error::{HttpError, Result};
5use crate::pool::SqliteConnectionManager;
6
7pub type SqlitePool = Pool<SqliteConnectionManager>;
8
9/// A pooled SQLite connection. Cheap to clone (shares the pool handle).
10/// Queries run on rusqlite (blocking) under `spawn_blocking` via
11/// [`Db::with_conn`] — a deliberate choice, not a stopgap: builtins called
12/// from the (synchronous) Ling interpreter dispatch this way too, so one
13/// blocking DB layer serves both the async HTTP side and the interpreter
14/// bridge without two separate drivers.
15#[derive(Clone)]
16pub struct Db(pub SqlitePool);
17
18impl Db {
19    /// Opens (creating if needed) a SQLite database file and returns a pool.
20    pub async fn connect(path: impl AsRef<Path>) -> anyhow::Result<Self> {
21        let path = path.as_ref().to_path_buf();
22        let pool = tokio::task::spawn_blocking(move || -> anyhow::Result<SqlitePool> {
23            let manager = SqliteConnectionManager::file(path);
24            let pool = Pool::builder().max_size(8).build(manager)?;
25            pool.get()?.execute_batch("PRAGMA journal_mode = WAL;")?;
26            Ok(pool)
27        })
28        .await??;
29        Ok(Self(pool))
30    }
31
32    /// A private, shared-cache in-memory database — useful for tests and
33    /// examples. All connections in the pool see the same data.
34    pub async fn connect_memory() -> anyhow::Result<Self> {
35        let manager =
36            SqliteConnectionManager::shared_memory("file:ling_http_mem?mode=memory&cache=shared");
37        let pool = tokio::task::spawn_blocking(move || -> anyhow::Result<SqlitePool> {
38            // A single connection for the shared in-memory DB: SQLite drops
39            // shared-cache memory data once its last connection closes, so
40            // keeping exactly one alive in the pool for the process
41            // lifetime is what makes data survive across pool checkouts.
42            Ok(Pool::builder().max_size(1).min_idle(Some(1)).build(manager)?)
43        })
44        .await??;
45        Ok(Self(pool))
46    }
47
48    pub fn pool(&self) -> &SqlitePool {
49        &self.0
50    }
51
52    /// Runs `f` against a pooled connection on a blocking thread, mapping
53    /// pool/query errors into [`HttpError`] (so handlers can just `?` it).
54    pub async fn with_conn<T, F>(&self, f: F) -> Result<T>
55    where
56        T: Send + 'static,
57        F: FnOnce(&rusqlite::Connection) -> rusqlite::Result<T> + Send + 'static,
58    {
59        let pool = self.0.clone();
60        tokio::task::spawn_blocking(move || {
61            let conn = pool.get().map_err(HttpError::Pool)?;
62            f(&conn).map_err(HttpError::Database)
63        })
64        .await
65        .map_err(|e| HttpError::Internal(anyhow::anyhow!("db task panicked: {e}")))?
66    }
67
68    /// Applies `(name, sql)` migrations in order, skipping ones already
69    /// recorded as applied in a `_migrations` bookkeeping table.
70    pub async fn run_migrations(&self, migrations: &'static [(&'static str, &'static str)]) -> anyhow::Result<()> {
71        let pool = self.0.clone();
72        tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
73            let mut conn = pool.get()?;
74            conn.execute_batch(
75                "CREATE TABLE IF NOT EXISTS _migrations (\
76                   name TEXT PRIMARY KEY, \
77                   applied_at TEXT NOT NULL DEFAULT (datetime('now'))\
78                 )",
79            )?;
80            let tx = conn.transaction()?;
81            for (name, sql) in migrations {
82                let already: bool = tx.query_row(
83                    "SELECT EXISTS(SELECT 1 FROM _migrations WHERE name = ?1)",
84                    [name],
85                    |row| row.get(0),
86                )?;
87                if already {
88                    continue;
89                }
90                tx.execute_batch(sql)?;
91                tx.execute("INSERT INTO _migrations (name) VALUES (?1)", [name])?;
92            }
93            tx.commit()?;
94            Ok(())
95        })
96        .await?
97    }
98}
ling_http/db.rs

ling_http/
db.rs
