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

minco_sqlx_sqlite/
lib.rs
