fslite_sqlite/
db.rs

1use std::path::{Path, PathBuf};
2use std::time::{Duration, UNIX_EPOCH};
3
4use fslite_core::{FsError, FsResult};
5use rusqlite::{Connection as RusqliteConnection, OptionalExtension};
6use tokio_rusqlite::Connection;
7
8const MIGRATIONS: &[(i64, &str)] = &[(1, include_str!("../migrations/0001_initial.sql"))];
9
10fn latest_schema_version() -> i64 {
11    MIGRATIONS.last().map(|(version, _)| *version).unwrap_or(0)
12}
13
14/// Options controlling how a SQLite-backed filesystem opens its database connection.
15#[non_exhaustive]
16#[derive(Clone, Copy, Debug)]
17pub struct ConnectOptions {
18    /// How long a connection waits on a locked database before failing.
19    pub busy_timeout: Duration,
20}
21
22impl Default for ConnectOptions {
23    fn default() -> Self {
24        Self {
25            busy_timeout: Duration::from_secs(5),
26        }
27    }
28}
29
30/// Returns the current Unix timestamp in milliseconds.
31pub(crate) fn now_ms() -> i64 {
32    UNIX_EPOCH
33        .elapsed()
34        .expect("system clock is before the Unix epoch")
35        .as_millis() as i64
36}
37
38pub(crate) fn map_call_error(err: tokio_rusqlite::Error) -> FsError {
39    FsError::internal_storage_failure(err)
40}
41
42pub(crate) async fn open_file(path: &Path, options: ConnectOptions) -> FsResult<Connection> {
43    let owned_path: PathBuf = path.to_owned();
44    let conn = Connection::open(owned_path).await.map_err(map_call_error)?;
45    initialize(&conn, options).await?;
46    Ok(conn)
47}
48
49pub(crate) async fn open_memory(options: ConnectOptions) -> FsResult<Connection> {
50    let conn = Connection::open_in_memory().await.map_err(map_call_error)?;
51    initialize(&conn, options).await?;
52    Ok(conn)
53}
54
55async fn initialize(conn: &Connection, options: ConnectOptions) -> FsResult<()> {
56    let current_version = conn
57        .call(move |conn| {
58            conn.pragma_update(None, "foreign_keys", "ON")?;
59            conn.pragma_update(None, "journal_mode", "WAL")?;
60            conn.pragma_update(None, "synchronous", "NORMAL")?;
61            conn.busy_timeout(options.busy_timeout)?;
62            Ok(read_current_version(conn)?)
63        })
64        .await
65        .map_err(map_call_error)?;
66
67    if current_version > latest_schema_version() {
68        return Err(FsError::internal_storage_failure(format!(
69            "database schema version {current_version} is newer than the supported version {}",
70            latest_schema_version()
71        )));
72    }
73
74    conn.call(move |conn| Ok(apply_migrations(conn, current_version)?))
75        .await
76        .map_err(map_call_error)
77}
78
79fn read_current_version(conn: &RusqliteConnection) -> rusqlite::Result<i64> {
80    let table_exists: i64 = conn.query_row(
81        "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'schema_migrations'",
82        [],
83        |row| row.get(0),
84    )?;
85
86    if table_exists == 0 {
87        return Ok(0);
88    }
89
90    conn.query_row(
91        "SELECT COALESCE(MAX(version), 0) FROM schema_migrations",
92        [],
93        |row| row.get(0),
94    )
95    .optional()
96    .map(|value| value.unwrap_or(0))
97}
98
99fn apply_migrations(conn: &mut RusqliteConnection, current_version: i64) -> rusqlite::Result<()> {
100    for (version, sql) in MIGRATIONS {
101        if *version <= current_version {
102            continue;
103        }
104
105        let tx = conn.transaction()?;
106        tx.execute_batch(sql)?;
107        tx.execute(
108            "INSERT INTO schema_migrations(version, applied_at_ms) VALUES (?1, ?2)",
109            rusqlite::params![version, now_ms()],
110        )?;
111        tx.commit()?;
112    }
113
114    Ok(())
115}
fslite_sqlite/db.rs

fslite_sqlite/
db.rs
