synd_persistence/sqlite/
connection.rs

1use std::{
2    path::Path,
3    time::{Duration, Instant},
4};
5
6use sqlx::{
7    Sqlite, SqlitePool, Transaction,
8    migrate::MigrateError as SqlxMigrateError,
9    sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions},
10};
11use synd_registry::RegistryDbError;
12use thiserror::Error;
13use tracing::{debug, info};
14
15const SQLITE_BUSY_TIMEOUT: Duration = Duration::from_secs(5);
16const SQLITE_MAX_CONNECTIONS: u32 = 1;
17
18#[derive(Clone)]
19pub struct SqliteDatabase {
20    pool: SqlitePool,
21}
22
23/// Error returned while migrating a `SQLite` database.
24#[derive(Debug, Error)]
25pub enum MigrationError {
26    #[error(transparent)]
27    Sqlx(#[from] SqlxMigrateError),
28}
29
30#[derive(Clone, Copy)]
31enum FileMode {
32    Existing,
33    CreateIfMissing,
34}
35
36impl SqliteDatabase {
37    pub async fn open(db_path: impl AsRef<Path>) -> Result<Self, RegistryDbError> {
38        Self::open_file(db_path, FileMode::Existing).await
39    }
40
41    pub async fn create_or_open(db_path: impl AsRef<Path>) -> Result<Self, RegistryDbError> {
42        Self::open_file(db_path, FileMode::CreateIfMissing).await
43    }
44
45    pub async fn migrate(&self) -> Result<(), MigrationError> {
46        let migrator = sqlx::migrate!("./migrations");
47        let schema_version = migrator
48            .migrations
49            .last()
50            .map_or(0, |migration| migration.version);
51        let started_at = Instant::now();
52        debug!(schema_version, "Running SQLite migrations");
53        migrator.run(&self.pool).await?;
54        info!(
55            schema_version,
56            duration_ms = started_at.elapsed().as_millis(),
57            "SQLite database ready"
58        );
59
60        Ok(())
61    }
62
63    pub async fn begin(&self) -> Result<Transaction<'_, Sqlite>, RegistryDbError> {
64        self.pool.begin().await.map_err(RegistryDbError::retryable)
65    }
66
67    async fn open_file(db_path: impl AsRef<Path>, mode: FileMode) -> Result<Self, RegistryDbError> {
68        let db_path = db_path.as_ref();
69        debug!(
70            database = %db_path.display(),
71            read_only = false,
72            create_if_missing = matches!(mode, FileMode::CreateIfMissing),
73            journal_mode = "wal",
74            foreign_keys = true,
75            busy_timeout_ms = SQLITE_BUSY_TIMEOUT.as_millis(),
76            max_connections = SQLITE_MAX_CONNECTIONS,
77            "Connecting to SQLite"
78        );
79        let opts = Self::file_options(db_path, mode);
80        Self::build_pool(opts).await
81    }
82
83    fn file_options(db_path: impl AsRef<Path>, mode: FileMode) -> SqliteConnectOptions {
84        Self::common_options(
85            SqliteConnectOptions::new()
86                .filename(db_path)
87                .create_if_missing(matches!(mode, FileMode::CreateIfMissing))
88                .journal_mode(SqliteJournalMode::Wal),
89        )
90    }
91
92    fn common_options(opts: SqliteConnectOptions) -> SqliteConnectOptions {
93        opts.foreign_keys(true).busy_timeout(SQLITE_BUSY_TIMEOUT)
94    }
95
96    async fn build_pool(opts: SqliteConnectOptions) -> Result<Self, RegistryDbError> {
97        let database = opts.get_filename().to_owned();
98        let pool = SqlitePoolOptions::new()
99            .max_connections(SQLITE_MAX_CONNECTIONS)
100            .connect_with(opts)
101            .await
102            .map_err(RegistryDbError::retryable)?;
103        debug!(database = %database.display(), "Connected to SQLite");
104
105        Ok(Self { pool })
106    }
107
108    #[cfg(test)]
109    pub async fn in_memory() -> Result<Self, RegistryDbError> {
110        Self::build_pool(Self::common_options(
111            SqliteConnectOptions::new().in_memory(true),
112        ))
113        .await
114    }
115}
synd_persistence/sqlite/connection.rs

synd_persistence/sqlite/
connection.rs
