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

minco_sqlx_sqlite/
lib.rs
