minco_sqlx_sqlite/
lib.rs

1//! `SQLx` `SQLite` pools with explicit file-backed versus in-memory behavior.
2#![forbid(unsafe_code)]
3
4use minco_db::{
5    AppliedMigration, DatabaseBackend, MigrationSet, SeedPlan, SeedTransaction, SeedVerification,
6    TargetState, resolve_seed_source, validate_seed_plan as validate_seed_model_plan,
7};
8use serde::{Deserialize, Serialize};
9pub use sqlx::SqlitePool;
10use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
11use std::{
12    fs::{File, OpenOptions},
13    path::{Path, PathBuf},
14    str::FromStr,
15    time::Duration,
16};
17use thiserror::Error;
18
19pub mod plugin_adapters;
20
21#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
22pub struct SqlitePoolConfig {
23    pub url: String,
24    pub max_connections: u32,
25    pub acquire_timeout_seconds: u64,
26}
27
28impl std::fmt::Debug for SqlitePoolConfig {
29    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
30        formatter
31            .debug_struct("SqlitePoolConfig")
32            .field("url", &"[REDACTED DATABASE URL]")
33            .field("max_connections", &self.max_connections)
34            .field("acquire_timeout_seconds", &self.acquire_timeout_seconds)
35            .finish()
36    }
37}
38
39impl SqlitePoolConfig {
40    pub fn file(path: impl AsRef<Path>) -> Self {
41        Self {
42            url: format!("sqlite://{}", path.as_ref().display()),
43            max_connections: 4,
44            acquire_timeout_seconds: 5,
45        }
46    }
47    pub fn memory() -> Self {
48        Self {
49            url: "sqlite::memory:".into(),
50            max_connections: 1,
51            acquire_timeout_seconds: 5,
52        }
53    }
54    pub fn is_memory(&self) -> bool {
55        self.url == "sqlite::memory:" || self.url.contains("mode=memory")
56    }
57    pub fn validate(&self) -> Result<(), SqliteError> {
58        if self.url.trim().is_empty() {
59            return Err(SqliteError::InvalidConfig("database URL is empty".into()));
60        }
61        if self.max_connections == 0 {
62            return Err(SqliteError::InvalidConfig(
63                "max_connections must be at least 1".into(),
64            ));
65        }
66        if self.is_memory() && self.max_connections != 1 {
67            return Err(SqliteError::InvalidConfig(
68                "in-memory SQLite requires exactly one pooled connection".into(),
69            ));
70        }
71        Ok(())
72    }
73}
74
75pub async fn connect(config: &SqlitePoolConfig) -> Result<SqlitePool, SqliteError> {
76    config.validate()?;
77    let mut options = SqliteConnectOptions::from_str(&config.url)?
78        .create_if_missing(!config.is_memory())
79        .foreign_keys(true)
80        .busy_timeout(Duration::from_secs(config.acquire_timeout_seconds));
81    if !config.is_memory() {
82        options = options.journal_mode(SqliteJournalMode::Wal);
83    }
84    Ok(SqlitePoolOptions::new()
85        .max_connections(config.max_connections)
86        .acquire_timeout(Duration::from_secs(config.acquire_timeout_seconds))
87        .connect_with(options)
88        .await?)
89}
90
91pub async fn migrate(pool: &SqlitePool, path: impl AsRef<Path>) -> Result<(), SqliteError> {
92    let migrator = sqlx::migrate::Migrator::new(path.as_ref()).await?;
93    migrator.run(pool).await?;
94    Ok(())
95}
96
97pub async fn migrate_with_history_table(
98    pool: &SqlitePool,
99    path: impl AsRef<Path>,
100    history_table: &'static str,
101) -> Result<(), SqliteError> {
102    validate_identifier(history_table, "migration history table")?;
103    let mut migrator = sqlx::migrate::Migrator::new(path.as_ref()).await?;
104    migrator.dangerous_set_table_name(history_table);
105    migrator.run(pool).await?;
106    Ok(())
107}
108
109pub async fn migration_target_state(
110    pool: &SqlitePool,
111    set: &MigrationSet,
112) -> Result<TargetState, SqliteError> {
113    validate_set(set)?;
114    if !table_exists(pool, &set.history_table).await? {
115        return Ok(TargetState::default());
116    }
117    let dirty_query = format!(
118        "SELECT version FROM {} WHERE success = false ORDER BY version LIMIT 1",
119        set.history_table
120    );
121    // The only interpolated token was accepted by `validate_identifier`.
122    let dirty_version = sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(dirty_query))
123        .fetch_optional(pool)
124        .await?;
125    let applied_query = format!(
126        "SELECT version, checksum FROM {} WHERE success = true ORDER BY version",
127        set.history_table
128    );
129    // The only interpolated token was accepted by `validate_identifier`.
130    let applied = sqlx::query_as::<_, (i64, Vec<u8>)>(sqlx::AssertSqlSafe(applied_query))
131        .fetch_all(pool)
132        .await?
133        .into_iter()
134        .map(|(version, checksum)| AppliedMigration {
135            version,
136            sqlx_checksum_sha384: hex(&checksum),
137        })
138        .collect();
139    Ok(TargetState {
140        dirty_version,
141        applied,
142    })
143}
144
145pub async fn verify_migration_tables(
146    pool: &SqlitePool,
147    set: &MigrationSet,
148) -> Result<Vec<String>, SqliteError> {
149    validate_set(set)?;
150    let mut missing = Vec::new();
151    for table in &set.verify_tables {
152        if !table_exists(pool, table).await? {
153            missing.push(table.clone());
154        }
155    }
156    Ok(missing)
157}
158
159pub async fn apply_migration_set(
160    pool: &SqlitePool,
161    config: &SqlitePoolConfig,
162    project_root: &Path,
163    set: &MigrationSet,
164) -> Result<(), SqliteError> {
165    apply_migration_plan(pool, config, project_root, std::slice::from_ref(set)).await
166}
167
168pub async fn apply_migration_plan(
169    pool: &SqlitePool,
170    config: &SqlitePoolConfig,
171    project_root: &Path,
172    sets: &[MigrationSet],
173) -> Result<(), SqliteError> {
174    if sets.is_empty() {
175        return Err(SqliteError::InvalidConfig(
176            "migration plan contains no sets".into(),
177        ));
178    }
179    config.validate()?;
180    let mut migrators = Vec::with_capacity(sets.len());
181    for set in sets {
182        validate_set(set)?;
183        let root = migration_root(project_root, set)?;
184        let mut migrator = sqlx::migrate::Migrator::new(root).await?;
185        verify_resolved_migrations(&migrator, set)?;
186        migrator.dangerous_set_table_name(set.history_table.clone());
187        migrators.push(migrator);
188    }
189    let _lock = acquire_migration_lock(config)?;
190    for migrator in migrators {
191        migrator.run(pool).await?;
192    }
193    Ok(())
194}
195
196pub async fn apply_seed_plan(
197    pool: &SqlitePool,
198    project_root: &Path,
199    plan: &SeedPlan,
200) -> Result<(), SqliteError> {
201    validate_seed_plan(plan)?;
202    let sources = plan
203        .seeds
204        .iter()
205        .map(|seed| resolve_seed_source(project_root, seed))
206        .collect::<Result<Vec<_>, _>>()
207        .map_err(|error| SqliteError::SeedSource(error.to_string()))?;
208    match plan.seeds[0].transaction {
209        SeedTransaction::Required => {
210            let mut transaction = pool.begin().await?;
211            for source in sources {
212                sqlx::raw_sql(sqlx::AssertSqlSafe(source.apply_sql))
213                    .execute(&mut *transaction)
214                    .await?;
215            }
216            transaction.commit().await?;
217        }
218        SeedTransaction::Autocommit => {
219            for source in sources {
220                sqlx::raw_sql(sqlx::AssertSqlSafe(source.apply_sql))
221                    .execute(pool)
222                    .await?;
223            }
224        }
225    }
226    Ok(())
227}
228
229pub async fn verify_seed_plan(
230    pool: &SqlitePool,
231    project_root: &Path,
232    plan: &SeedPlan,
233) -> Result<Vec<SeedVerification>, SqliteError> {
234    validate_seed_plan(plan)?;
235    let mut connection = pool.acquire().await?;
236    // `query_only` is connection-local. Closing this connection on every exit
237    // prevents either a failed verification query or the guard itself from
238    // leaking state back into the pool.
239    connection.close_on_drop();
240    sqlx::query("PRAGMA query_only = ON")
241        .execute(&mut *connection)
242        .await?;
243    let mut verification = Vec::with_capacity(plan.seeds.len());
244    for seed in &plan.seeds {
245        let source = resolve_seed_source(project_root, seed)
246            .map_err(|error| SqliteError::SeedSource(error.to_string()))?;
247        let rows = sqlx::query_scalar::<_, bool>(sqlx::AssertSqlSafe(source.verify_sql))
248            .fetch_all(&mut *connection)
249            .await?;
250        if rows.len() != 1 {
251            return Err(SqliteError::InvalidConfig(format!(
252                "seed {} verification must return exactly one boolean row",
253                seed.id
254            )));
255        }
256        verification.push(SeedVerification {
257            seed_id: seed.id.clone(),
258            verified: rows[0],
259        });
260    }
261    Ok(verification)
262}
263
264pub async fn ready(pool: &SqlitePool) -> bool {
265    matches!(
266        sqlx::query_scalar::<_, i64>("SELECT 1")
267            .fetch_one(pool)
268            .await,
269        Ok(1)
270    )
271}
272
273fn validate_seed_plan(plan: &SeedPlan) -> Result<(), SqliteError> {
274    validate_seed_model_plan(plan).map_err(|error| SqliteError::SeedSource(error.to_string()))?;
275    if plan.seeds.is_empty() {
276        return Err(SqliteError::InvalidConfig(
277            "seed plan contains no seeds".into(),
278        ));
279    }
280    if plan
281        .seeds
282        .iter()
283        .any(|seed| seed.backend != DatabaseBackend::Sqlite)
284    {
285        return Err(SqliteError::InvalidConfig(
286            "seed plan contains a non-SQLite seed".into(),
287        ));
288    }
289    if plan
290        .seeds
291        .iter()
292        .any(|seed| seed.transaction != plan.seeds[0].transaction)
293    {
294        return Err(SqliteError::InvalidConfig(
295            "seed plan mixes transaction behaviors".into(),
296        ));
297    }
298    Ok(())
299}
300
301async fn table_exists(pool: &SqlitePool, table: &str) -> Result<bool, SqliteError> {
302    validate_identifier(table, "table")?;
303    Ok(sqlx::query_scalar::<_, String>(
304        "SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?1",
305    )
306    .bind(table)
307    .fetch_optional(pool)
308    .await?
309    .is_some())
310}
311
312fn validate_set(set: &MigrationSet) -> Result<(), SqliteError> {
313    if set.backend != DatabaseBackend::Sqlite {
314        return Err(SqliteError::InvalidConfig(format!(
315            "migration set {} targets a different database backend",
316            set.id
317        )));
318    }
319    validate_identifier(&set.history_table, "migration history table")?;
320    for table in &set.verify_tables {
321        validate_identifier(table, "verification table")?;
322    }
323    Ok(())
324}
325
326fn migration_root(project_root: &Path, set: &MigrationSet) -> Result<PathBuf, SqliteError> {
327    let project_root = project_root.canonicalize().map_err(SqliteError::Io)?;
328    if set.root.is_absolute() {
329        return Err(SqliteError::InvalidConfig(format!(
330            "migration set {} has an absolute source root",
331            set.id
332        )));
333    }
334    let root = project_root
335        .join(&set.root)
336        .canonicalize()
337        .map_err(SqliteError::Io)?;
338    if !root.starts_with(&project_root) {
339        return Err(SqliteError::InvalidConfig(format!(
340            "migration set {} source root escapes the project",
341            set.id
342        )));
343    }
344    Ok(root)
345}
346
347fn verify_resolved_migrations(
348    migrator: &sqlx::migrate::Migrator,
349    set: &MigrationSet,
350) -> Result<(), SqliteError> {
351    let resolved = migrator.iter().collect::<Vec<_>>();
352    if resolved.len() != set.migrations.len() {
353        return Err(SqliteError::SourceDrift(set.id.clone()));
354    }
355    for (resolved, expected) in resolved.iter().zip(&set.migrations) {
356        if resolved.version != expected.version
357            || hex(resolved.checksum.as_ref()) != expected.sqlx_checksum_sha384
358        {
359            return Err(SqliteError::SourceDrift(set.id.clone()));
360        }
361    }
362    Ok(())
363}
364
365fn acquire_migration_lock(config: &SqlitePoolConfig) -> Result<File, SqliteError> {
366    if config.is_memory() {
367        return Err(SqliteError::InvalidConfig(
368            "migration execution requires file-backed SQLite".into(),
369        ));
370    }
371    let options = SqliteConnectOptions::from_str(&config.url)?;
372    let database = options
373        .get_filename()
374        .canonicalize()
375        .map_err(SqliteError::Io)?;
376    let mut lock_name = database.as_os_str().to_os_string();
377    lock_name.push(".minco-migrate.lock");
378    let lock = OpenOptions::new()
379        .read(true)
380        .write(true)
381        .create(true)
382        .truncate(false)
383        .open(PathBuf::from(lock_name))
384        .map_err(SqliteError::Io)?;
385    match lock.try_lock() {
386        Ok(()) => {}
387        Err(std::fs::TryLockError::WouldBlock) => {
388            return Err(SqliteError::MigrationLockUnavailable);
389        }
390        Err(std::fs::TryLockError::Error(source)) => {
391            return Err(SqliteError::Io(source));
392        }
393    }
394    Ok(lock)
395}
396
397fn hex(bytes: &[u8]) -> String {
398    const DIGITS: &[u8; 16] = b"0123456789abcdef";
399    let mut output = String::with_capacity(bytes.len() * 2);
400    for byte in bytes {
401        output.push(char::from(DIGITS[usize::from(byte >> 4)]));
402        output.push(char::from(DIGITS[usize::from(byte & 0x0f)]));
403    }
404    output
405}
406
407fn validate_identifier(value: &str, description: &str) -> Result<(), SqliteError> {
408    let mut bytes = value.bytes();
409    let valid_start = bytes
410        .next()
411        .is_some_and(|byte| byte.is_ascii_alphabetic() || byte == b'_');
412    if !valid_start
413        || value.len() > 63
414        || !bytes.all(|byte| byte.is_ascii_alphanumeric() || byte == b'_')
415    {
416        return Err(SqliteError::InvalidConfig(format!(
417            "{description} must be a SQLite identifier of at most 63 ASCII characters"
418        )));
419    }
420    Ok(())
421}
422
423#[derive(Debug, Error)]
424pub enum SqliteError {
425    #[error("invalid SQLite configuration: {0}")]
426    InvalidConfig(String),
427    #[error("SQLite error: {0}")]
428    Sqlx(#[from] sqlx::Error),
429    #[error("SQLite migration error: {0}")]
430    Migration(#[from] sqlx::migrate::MigrateError),
431    #[error("SQLite migration source changed after planning for set {0}")]
432    SourceDrift(String),
433    #[error("another SQLite migration process holds the migration lock")]
434    MigrationLockUnavailable,
435    #[error("SQLite migration filesystem operation failed: {0}")]
436    Io(#[from] std::io::Error),
437    #[error("SQLite seed source validation failed: {0}")]
438    SeedSource(String),
439}
440
441#[cfg(test)]
442mod tests {
443    use super::*;
444    use minco_db::{MigrationState, compare_target, load_catalog};
445    use std::fs;
446    use tempfile::TempDir;
447
448    fn lifecycle_fixture() -> (TempDir, minco_db::MigrationSet) {
449        let root = TempDir::new().expect("temporary migration project");
450        let migrations = root.path().join("migrations");
451        fs::create_dir(&migrations).expect("create migration directory");
452        fs::write(
453            migrations.join("0001_example.sql"),
454            "CREATE TABLE example (id INTEGER PRIMARY KEY);\n",
455        )
456        .expect("write migration");
457        fs::write(
458            migrations.join(minco_db::MIGRATION_SET_MANIFEST),
459            concat!(
460                "schema = 1\n",
461                "id = \"test-sqlite\"\n",
462                "owner = \"application:test\"\n",
463                "backend = \"sqlite\"\n",
464                "history_table = \"_minco_test_migrations\"\n",
465                "verify_tables = [\"example\"]\n",
466                "\n",
467                "[[migration]]\n",
468                "version = 1\n",
469                "risk = \"additive\"\n",
470                "reversible = false\n",
471            ),
472        )
473        .expect("write lifecycle manifest");
474        let catalog = load_catalog(root.path(), &[Path::new("migrations").to_path_buf()])
475            .expect("load lifecycle catalog");
476        let set = catalog.sets.into_iter().next().expect("migration set");
477        (root, set)
478    }
479
480    #[test]
481    fn memory_profile_rejects_multiple_connections() {
482        let mut config = SqlitePoolConfig::memory();
483        config.max_connections = 2;
484        assert!(config.validate().is_err());
485    }
486
487    #[test]
488    fn pool_configuration_debug_redacts_database_urls() {
489        let config = SqlitePoolConfig {
490            url: "sqlite://var/app.db?password=secret-password".into(),
491            max_connections: 1,
492            acquire_timeout_seconds: 5,
493        };
494        let debug = format!("{config:?}");
495        assert!(!debug.contains("secret-password"));
496        assert!(!debug.contains("sqlite://"));
497    }
498
499    #[tokio::test]
500    async fn migration_history_table_rejects_dynamic_sql_tokens() {
501        let pool = connect(&SqlitePoolConfig::memory())
502            .await
503            .expect("in-memory pool");
504        let result =
505            migrate_with_history_table(&pool, Path::new("missing"), "_migrations;DROP").await;
506        assert!(matches!(result, Err(SqliteError::InvalidConfig(_))));
507    }
508
509    #[tokio::test]
510    async fn lifecycle_migration_reports_state_and_verifies_expected_tables() {
511        let (project, set) = lifecycle_fixture();
512        let database = project.path().join("test.sqlite");
513        let config = SqlitePoolConfig::file(&database);
514        let pool = connect(&config).await.expect("connect SQLite");
515
516        let before = migration_target_state(&pool, &set)
517            .await
518            .expect("read empty target state");
519        assert!(before.applied.is_empty());
520
521        apply_migration_set(&pool, &config, project.path(), &set)
522            .await
523            .expect("apply migration set");
524
525        let after = migration_target_state(&pool, &set)
526            .await
527            .expect("read applied target state");
528        let status = compare_target(&set, &after);
529        assert_eq!(status.entries[0].state, MigrationState::Applied);
530        assert!(
531            verify_migration_tables(&pool, &set)
532                .await
533                .expect("verify migration tables")
534                .is_empty()
535        );
536    }
537
538    #[tokio::test]
539    async fn lifecycle_migration_fails_closed_when_another_process_holds_the_file_lock() {
540        let (project, set) = lifecycle_fixture();
541        let config = SqlitePoolConfig::file(project.path().join("test.sqlite"));
542        let pool = connect(&config).await.expect("connect SQLite");
543        let _held_lock = acquire_migration_lock(&config).expect("hold migration lock");
544
545        let error = apply_migration_set(&pool, &config, project.path(), &set)
546            .await
547            .expect_err("concurrent migration must fail");
548        assert!(matches!(error, SqliteError::MigrationLockUnavailable));
549    }
550
551    #[cfg(unix)]
552    #[tokio::test]
553    async fn lifecycle_lock_cannot_be_bypassed_with_a_database_symlink() {
554        use std::os::unix::fs::symlink;
555
556        let (project, _) = lifecycle_fixture();
557        let database = project.path().join("test.sqlite");
558        let config = SqlitePoolConfig::file(&database);
559        let pool = connect(&config).await.expect("connect SQLite");
560        pool.close().await;
561        let alias = project.path().join("database-alias.sqlite");
562        symlink(&database, &alias).expect("create database symlink");
563        let alias_config = SqlitePoolConfig::file(alias);
564
565        let _held_lock = acquire_migration_lock(&config).expect("hold canonical migration lock");
566        let error =
567            acquire_migration_lock(&alias_config).expect_err("symlink alias must share the lock");
568        assert!(matches!(error, SqliteError::MigrationLockUnavailable));
569    }
570
571    #[tokio::test]
572    async fn lifecycle_migration_rejects_in_memory_targets() {
573        let (project, set) = lifecycle_fixture();
574        let config = SqlitePoolConfig::memory();
575        let pool = connect(&config).await.expect("connect SQLite");
576
577        let error = apply_migration_set(&pool, &config, project.path(), &set)
578            .await
579            .expect_err("in-memory migration target must fail");
580        assert!(matches!(error, SqliteError::InvalidConfig(_)));
581    }
582}
minco_sqlx_sqlite/lib.rs

minco_sqlx_sqlite/
lib.rs
