rightkit_db/
migrate.rs

1//! Ordered, transactional, crash-aware schema migrations.
2//!
3//! The library owns the *mechanism*; apps own the schemas. Rules, taken from the strongest
4//! donors (CodeRight `storage::migrations`, MailRight `db::migrations`, WorkRight `local-store`):
5//!
6//! - versions are contiguous from 1 and default to `PRAGMA user_version`;
7//! - each step runs in its own `BEGIN IMMEDIATE` transaction together with its version bump, so
8//!   a failing step rolls back to the last good version;
9//! - before the first pending step on a non-empty database an online backup is taken to
10//!   `<db>.pre-v<target>.bak` (WAL-inclusive, atomic, owner-only);
11//! - a `<db>.migrating` marker brackets the batch so a crash mid-batch is reported on next open;
12//! - a database newer than this build is refused unless the app's probe says the subset it
13//!   needs is still present (additive forward compatibility);
14//! - rollback replays `down` steps and refuses when any step in the range lacks one.
15
16use crate::{DbError, DbOptions, Result};
17use rusqlite::{Connection, OptionalExtension};
18use std::path::{Path, PathBuf};
19use std::time::{SystemTime, UNIX_EPOCH};
20
21pub enum Step {
22    Sql(&'static str),
23    /// Rust step (data backfills, table rebuilds); runs inside the step's transaction.
24    Code(fn(&Connection) -> rusqlite::Result<()>),
25}
26
27pub struct Migration {
28    pub version: i64,
29    pub name: &'static str,
30    pub up: Step,
31    /// Reverse SQL; `None` makes this step unrevertable (rollback refuses to guess).
32    pub down: Option<&'static str>,
33}
34
35impl Migration {
36    pub const fn sql(version: i64, name: &'static str, up: &'static str) -> Self {
37        Self {
38            version,
39            name,
40            up: Step::Sql(up),
41            down: None,
42        }
43    }
44
45    pub const fn with_down(mut self, down: &'static str) -> Self {
46        self.down = Some(down);
47        self
48    }
49
50    pub const fn code(
51        version: i64,
52        name: &'static str,
53        up: fn(&Connection) -> rusqlite::Result<()>,
54    ) -> Self {
55        Self {
56            version,
57            name,
58            up: Step::Code(up),
59            down: None,
60        }
61    }
62}
63
64/// Where migration versions are stored. Table storage uses Blueprint-compatible
65/// `key TEXT PRIMARY KEY, value TEXT NOT NULL` rows, leaving `user_version` untouched.
66#[derive(Clone, Debug, Default, PartialEq, Eq)]
67pub enum VersionStorage {
68    #[default]
69    UserVersion,
70    /// Table names must contain only ASCII letters, digits, or underscores.
71    /// Keys are bound parameters and may contain arbitrary text.
72    Table { table: String, key: String },
73    /// One row per applied version, shaped `"<version_column>" INTEGER PRIMARY KEY,
74    /// "<applied_at_column>" INTEGER NOT NULL` (ViewRight's `_migrations` shape). The applied
75    /// version is the highest stored `version_column`; a rollback deletes the top row. The table is
76    /// created on the first write when absent. Identifiers must be ASCII letters, digits, or
77    /// underscores. `applied_at` stores unix seconds.
78    VersionTable {
79        table: String,
80        version_column: String,
81        applied_at_column: String,
82    },
83}
84
85impl VersionStorage {
86    /// [`VersionStorage::VersionTable`] with the `version` and `applied_at` column names.
87    pub fn version_table(table: impl Into<String>) -> Self {
88        Self::VersionTable {
89            table: table.into(),
90            version_column: "version".into(),
91            applied_at_column: "applied_at".into(),
92        }
93    }
94}
95
96/// An app's existing one-row-per-version table, read once to adopt its applied version (see
97/// [`MigrateOptions::import_legacy_versions`]). Only the highest `version_column` value is used.
98#[derive(Clone, Debug, PartialEq, Eq)]
99pub struct LegacyVersionTable {
100    pub table: String,
101    pub version_column: String,
102}
103
104#[derive(Clone)]
105pub struct MigrateOptions {
106    /// Version location shared by reads and transactional migration writes.
107    pub version_storage: VersionStorage,
108    /// Take a pre-migration backup when migrating an existing (version > 0) on-disk database.
109    pub backup: bool,
110    /// Key source for the backup (SQLCipher); `DbOptions::open_writer_migrated` fills this in.
111    pub backup_options: Option<DbOptions>,
112    /// For a database newer than this build: `true` means the subset this build reads/writes is
113    /// intact, so open it read-write without migrating.
114    pub forward_compatible: Option<fn(&Connection) -> bool>,
115    /// Refresh planner statistics after migrating (`PRAGMA optimize`).
116    pub optimize_after: bool,
117    /// Adopt an existing app version table (see [`MigrateOptions::import_legacy_versions`]). Default none.
118    pub legacy_versions: Option<LegacyVersionTable>,
119}
120
121impl Default for MigrateOptions {
122    fn default() -> Self {
123        Self {
124            version_storage: VersionStorage::default(),
125            backup: true,
126            backup_options: None,
127            forward_compatible: None,
128            optimize_after: true,
129            legacy_versions: None,
130        }
131    }
132}
133
134impl MigrateOptions {
135    /// Adopt the applied version of an app's existing version table, once, so its migrations are
136    /// not re-run. Applies only when the configured storage reads version 0 and the legacy table's
137    /// highest version is above zero. The configured storage is then seeded to that version (a
138    /// `VersionTable` gets one row per version up to it), and the migrations up to it are skipped.
139    /// The legacy table is only read, never written. Idempotent: once seeded, the configured
140    /// storage is above zero, so later opens do not consult the legacy table. Rolling back to
141    /// version 0 while the legacy table still exists re-adopts it on the next open, so drop or
142    /// rename that table after adoption. A legacy version above this build's newest migration is
143    /// refused as [`DbError::NewerDatabase`] and nothing is written.
144    pub fn import_legacy_versions(mut self, legacy: LegacyVersionTable) -> Self {
145        self.legacy_versions = Some(legacy);
146        self
147    }
148}
149
150#[derive(Clone, Debug, Default, PartialEq, Eq)]
151pub struct MigrationReport {
152    pub from: i64,
153    pub to: i64,
154    pub applied: Vec<i64>,
155    pub backup: Option<PathBuf>,
156    /// Opened a newer-but-compatible database without migrating.
157    pub forward_compatible: bool,
158    /// A previous batch started but never finished (crash or failed step).
159    pub interrupted_previous: Option<PendingMarker>,
160}
161
162#[derive(Clone, Debug, PartialEq, Eq)]
163pub struct PendingMarker {
164    pub from: i64,
165    pub to: i64,
166    pub pid: u32,
167}
168
169pub fn user_version(conn: &Connection) -> Result<i64> {
170    Ok(conn.query_row("PRAGMA user_version", [], |row| row.get(0))?)
171}
172
173fn set_user_version(conn: &Connection, version: i64) -> rusqlite::Result<()> {
174    // `user_version` cannot be bound as a parameter; the value is an i64, never text.
175    conn.execute_batch(&format!("PRAGMA user_version = {version};"))
176}
177
178/// Read the selected migration version without creating a table or row.
179/// Missing table/key means zero; malformed or negative versions are refused.
180pub fn migration_version(conn: &Connection, storage: &VersionStorage) -> Result<i64> {
181    match storage {
182        VersionStorage::UserVersion => user_version(conn),
183        VersionStorage::Table { table, key } => key_value_version(conn, table, key),
184        VersionStorage::VersionTable {
185            table,
186            version_column,
187            ..
188        } => version_row_max(conn, table, version_column),
189    }
190}
191
192fn key_value_version(conn: &Connection, table: &str, key: &str) -> Result<i64> {
193    check_ident(table)?;
194    let exists: bool = conn.query_row(
195        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1 COLLATE NOCASE)",
196        [table],
197        |row| row.get(0),
198    )?;
199    if !exists {
200        return Ok(0);
201    }
202    let value: Option<String> = conn
203        .query_row(
204            &format!("SELECT CAST(\"value\" AS TEXT) FROM \"{table}\" WHERE \"key\"=?1"),
205            [key],
206            |row| row.get(0),
207        )
208        .optional()?;
209    match value {
210        None => Ok(0),
211        Some(value) => value
212            .parse::<i64>()
213            .ok()
214            .filter(|version| *version >= 0)
215            .ok_or_else(|| DbError::InvalidMigrations(format!("invalid stored version {value:?}"))),
216    }
217}
218
219// Called only inside the migration/down transaction, after identifier validation.
220fn set_key_value_version(
221    conn: &Connection,
222    storage: &VersionStorage,
223    version: i64,
224) -> rusqlite::Result<()> {
225    match storage {
226        VersionStorage::UserVersion => set_user_version(conn, version),
227        VersionStorage::Table { table, key } => {
228            conn.execute_batch(&format!(
229                "CREATE TABLE IF NOT EXISTS \"{table}\" (\"key\" TEXT PRIMARY KEY, \"value\" TEXT NOT NULL);"
230            ))?;
231            conn.execute(
232                &format!(
233                    "INSERT INTO \"{table}\"(\"key\",\"value\") VALUES(?1,?2) \
234                     ON CONFLICT(\"key\") DO UPDATE SET \"value\"=excluded.\"value\""
235                ),
236                rusqlite::params![key, version.to_string()],
237            )?;
238            Ok(())
239        }
240        // Routed to `advance_version` / `retreat_version`, never reached through them.
241        VersionStorage::VersionTable { .. } => Err(rusqlite::Error::InvalidQuery),
242    }
243}
244
245/// Make `version` the applied version: a forward step, inside its transaction.
246fn advance_version(
247    conn: &Connection,
248    storage: &VersionStorage,
249    version: i64,
250) -> rusqlite::Result<()> {
251    match storage {
252        VersionStorage::VersionTable {
253            table,
254            version_column,
255            applied_at_column,
256        } => {
257            conn.execute_batch(&format!(
258                "CREATE TABLE IF NOT EXISTS \"{table}\" (\"{version_column}\" INTEGER PRIMARY KEY, \"{applied_at_column}\" INTEGER NOT NULL);"
259            ))?;
260            conn.execute(
261                &format!("DELETE FROM \"{table}\" WHERE \"{version_column}\" = ?1"),
262                [version],
263            )?;
264            conn.execute(
265                &format!(
266                    "INSERT INTO \"{table}\" (\"{version_column}\", \"{applied_at_column}\") VALUES (?1, ?2)"
267                ),
268                rusqlite::params![version, now_secs()],
269            )?;
270            Ok(())
271        }
272        _ => set_key_value_version(conn, storage, version),
273    }
274}
275
276/// Un-apply `version`: a rollback step, so the applied version becomes `version - 1`.
277fn retreat_version(
278    conn: &Connection,
279    storage: &VersionStorage,
280    version: i64,
281) -> rusqlite::Result<()> {
282    match storage {
283        VersionStorage::VersionTable {
284            table,
285            version_column,
286            ..
287        } => {
288            conn.execute(
289                &format!("DELETE FROM \"{table}\" WHERE \"{version_column}\" = ?1"),
290                [version],
291            )?;
292            Ok(())
293        }
294        _ => set_key_value_version(conn, storage, version - 1),
295    }
296}
297
298/// Seed a database at version 0 with `through` as its applied version (legacy adoption). A
299/// `VersionTable` gets one row per version; the other storages take the single value.
300fn adopt_legacy_version(conn: &Connection, storage: &VersionStorage, through: i64) -> Result<()> {
301    conn.execute_batch("BEGIN IMMEDIATE;")?;
302    let seeded = match storage {
303        VersionStorage::VersionTable { .. } => {
304            (1..=through).try_for_each(|version| advance_version(conn, storage, version))
305        }
306        _ => advance_version(conn, storage, through),
307    };
308    match seeded {
309        Ok(()) => {
310            conn.execute_batch("COMMIT;")?;
311            Ok(())
312        }
313        Err(error) => {
314            let _ = conn.execute_batch("ROLLBACK;");
315            Err(DbError::Sqlite(error))
316        }
317    }
318}
319
320/// Highest version stored in a one-row-per-version table; zero when the table is absent or empty.
321fn version_row_max(conn: &Connection, table: &str, version_column: &str) -> Result<i64> {
322    check_ident(table)?;
323    check_ident(version_column)?;
324    let exists: bool = conn.query_row(
325        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1 COLLATE NOCASE)",
326        [table],
327        |row| row.get(0),
328    )?;
329    if !exists {
330        return Ok(0);
331    }
332    let highest: Option<i64> = conn.query_row(
333        &format!("SELECT MAX(CAST(\"{version_column}\" AS INTEGER)) FROM \"{table}\""),
334        [],
335        |row| row.get(0),
336    )?;
337    let version = highest.unwrap_or(0);
338    if version < 0 {
339        return Err(DbError::InvalidMigrations(format!(
340            "invalid stored version {version}"
341        )));
342    }
343    Ok(version)
344}
345
346/// Refuses any identifier that is not a plain ASCII name, before SQL is built from it.
347fn check_storage(storage: &VersionStorage) -> Result<()> {
348    match storage {
349        VersionStorage::UserVersion => Ok(()),
350        VersionStorage::Table { table, .. } => check_ident(table),
351        VersionStorage::VersionTable {
352            table,
353            version_column,
354            applied_at_column,
355        } => {
356            check_ident(table)?;
357            check_ident(version_column)?;
358            check_ident(applied_at_column)
359        }
360    }
361}
362
363fn now_secs() -> i64 {
364    SystemTime::now()
365        .duration_since(UNIX_EPOCH)
366        .map(|d| d.as_secs() as i64)
367        .unwrap_or(0)
368}
369
370pub fn validate(migrations: &[Migration]) -> Result<()> {
371    for (index, migration) in migrations.iter().enumerate() {
372        let expected = index as i64 + 1;
373        if migration.version != expected {
374            return Err(DbError::InvalidMigrations(format!(
375                "expected version {expected} at position {index}, found {} ({})",
376                migration.version, migration.name
377            )));
378        }
379    }
380    Ok(())
381}
382
383fn marker_path(db: &Path) -> PathBuf {
384    let mut name = db.as_os_str().to_os_string();
385    name.push(".migrating");
386    PathBuf::from(name)
387}
388
389pub fn pending_marker(db: &Path) -> Option<PendingMarker> {
390    let text = std::fs::read_to_string(marker_path(db)).ok()?;
391    let mut marker = PendingMarker {
392        from: 0,
393        to: 0,
394        pid: 0,
395    };
396    for line in text.lines() {
397        match line.split_once('=')? {
398            ("from", v) => marker.from = v.parse().ok()?,
399            ("to", v) => marker.to = v.parse().ok()?,
400            ("pid", v) => marker.pid = v.parse().ok()?,
401            _ => {}
402        }
403    }
404    Some(marker)
405}
406
407pub fn migrate(
408    conn: &Connection,
409    db_path: Option<&Path>,
410    migrations: &[Migration],
411    options: &MigrateOptions,
412) -> Result<MigrationReport> {
413    validate(migrations)?;
414    check_storage(&options.version_storage)?;
415    let mut current = migration_version(conn, &options.version_storage)?;
416    let max = migrations.last().map(|m| m.version).unwrap_or(0);
417    let mut report = MigrationReport {
418        from: current,
419        to: current,
420        ..Default::default()
421    };
422    report.interrupted_previous = db_path.and_then(pending_marker);
423
424    // Legacy adoption happens only from version 0 (see `MigrateOptions::import_legacy_versions`).
425    if let Some(legacy) = options.legacy_versions.as_ref().filter(|_| current == 0) {
426        let legacy_version = version_row_max(conn, &legacy.table, &legacy.version_column)?;
427        if legacy_version > 0 {
428            if legacy_version <= max {
429                adopt_legacy_version(conn, &options.version_storage, legacy_version)?;
430            }
431            current = legacy_version;
432            report.from = current;
433            report.to = current;
434        }
435    }
436
437    if current > max {
438        if options.forward_compatible.is_some_and(|probe| probe(conn)) {
439            report.forward_compatible = true;
440            return Ok(report);
441        }
442        return Err(DbError::NewerDatabase { current, max });
443    }
444    let pending: Vec<&Migration> = migrations.iter().filter(|m| m.version > current).collect();
445    if pending.is_empty() {
446        return Ok(report);
447    }
448
449    if current > 0 && options.backup {
450        if let Some(path) = db_path {
451            let name = path
452                .file_name()
453                .ok_or(DbError::InvalidDatabasePath)?
454                .to_string_lossy();
455            let backup = path.with_file_name(format!("{name}.pre-v{max}.bak"));
456            let backup_options = options
457                .backup_options
458                .clone()
459                .unwrap_or_else(|| DbOptions::new(path));
460            crate::online_backup_to(&backup_options, conn, &backup, None)?;
461            report.backup = Some(backup);
462        }
463    }
464
465    if let Some(path) = db_path {
466        let body = format!("from={current}\nto={max}\npid={}\n", std::process::id());
467        rightkit_fs::atomic_write(&marker_path(path), body.as_bytes())?;
468    }
469
470    for migration in pending {
471        conn.execute_batch("BEGIN IMMEDIATE;")?;
472        let applied = match &migration.up {
473            Step::Sql(sql) => conn.execute_batch(sql),
474            Step::Code(run) => run(conn),
475        }
476        .and_then(|()| advance_version(conn, &options.version_storage, migration.version));
477        match applied {
478            Ok(()) => {
479                conn.execute_batch("COMMIT;")?;
480                report.applied.push(migration.version);
481                report.to = migration.version;
482            }
483            Err(source) => {
484                let _ = conn.execute_batch("ROLLBACK;");
485                return Err(DbError::MigrationFailed {
486                    version: migration.version,
487                    name: migration.name,
488                    source,
489                });
490            }
491        }
492    }
493
494    if let Some(path) = db_path {
495        let _ = std::fs::remove_file(marker_path(path));
496    }
497    if options.optimize_after {
498        crate::optimize_boundary(conn)?;
499    }
500    Ok(report)
501}
502
503/// Revert to `target` by replaying each step's `down`, highest first, one transaction per step.
504/// Refuses before touching anything unless every step in the range has a `down`.
505pub fn rollback_to(conn: &Connection, migrations: &[Migration], target: i64) -> Result<Vec<i64>> {
506    rollback_to_with_storage(conn, migrations, target, &VersionStorage::UserVersion)
507}
508
509/// Revert using the same version storage selected for [`migrate`].
510pub fn rollback_to_with_storage(
511    conn: &Connection,
512    migrations: &[Migration],
513    target: i64,
514    storage: &VersionStorage,
515) -> Result<Vec<i64>> {
516    validate(migrations)?;
517    check_storage(storage)?;
518    if target < 0 {
519        return Err(DbError::Rollback(
520            "target version must be nonnegative".into(),
521        ));
522    }
523    let current = migration_version(conn, storage)?;
524    if current <= target {
525        return Err(DbError::Rollback(format!(
526            "already at v{current}, at or below target v{target}"
527        )));
528    }
529    let mut plan = Vec::new();
530    for version in ((target + 1)..=current).rev() {
531        let migration = migrations
532            .iter()
533            .find(|m| m.version == version)
534            .ok_or_else(|| DbError::Rollback(format!("v{version} is not a known migration")))?;
535        let down = migration.down.ok_or_else(|| {
536            DbError::Rollback(format!(
537                "v{version} ({}) has no down migration",
538                migration.name
539            ))
540        })?;
541        plan.push((version, down));
542    }
543    let mut reverted = Vec::new();
544    for (version, down) in plan {
545        conn.execute_batch("BEGIN IMMEDIATE;")?;
546        let result = conn
547            .execute_batch(down)
548            .and_then(|()| retreat_version(conn, storage, version));
549        match result {
550            Ok(()) => {
551                conn.execute_batch("COMMIT;")?;
552                reverted.push(version);
553            }
554            Err(error) => {
555                let _ = conn.execute_batch("ROLLBACK;");
556                return Err(DbError::Rollback(format!(
557                    "down v{version} failed and was rolled back: {error}"
558                )));
559            }
560        }
561    }
562    Ok(reverted)
563}
564
565/// True if `table` has `column`, including STORED/VIRTUAL generated columns (`table_xinfo`;
566/// `table_info` omits them, which made MailRight re-run an ALTER on every open).
567pub fn column_exists(conn: &Connection, table: &str, column: &str) -> Result<bool> {
568    check_ident(table)?;
569    let mut statement = conn.prepare(&format!("PRAGMA table_xinfo({table})"))?;
570    let names = statement.query_map([], |row| row.get::<_, String>(1))?;
571    for name in names {
572        if name? == column {
573            return Ok(true);
574        }
575    }
576    Ok(false)
577}
578
579/// Idempotent `ALTER TABLE .. ADD COLUMN` (SQLite has no `IF NOT EXISTS` for columns).
580/// `table`, `column` and `decl` must be literals from the app, never user input.
581pub fn add_column_if_missing(
582    conn: &Connection,
583    table: &str,
584    column: &str,
585    decl: &str,
586) -> Result<()> {
587    check_ident(table)?;
588    check_ident(column)?;
589    if !column_exists(conn, table, column)? {
590        conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {column} {decl};"))?;
591    }
592    Ok(())
593}
594
595fn check_ident(ident: &str) -> Result<()> {
596    if !ident.is_empty()
597        && ident
598            .bytes()
599            .all(|b| b.is_ascii_alphanumeric() || b == b'_')
600    {
601        Ok(())
602    } else {
603        Err(DbError::InvalidMigrations(format!(
604            "unsafe SQL identifier {ident:?}"
605        )))
606    }
607}
608
609impl DbOptions {
610    /// Open the writer and bring the schema up to date (see [`migrate`]); the pre-migration
611    /// backup is keyed like the database itself.
612    pub fn open_writer_migrated(
613        &self,
614        migrations: &[Migration],
615        options: &MigrateOptions,
616    ) -> Result<(Connection, MigrationReport)> {
617        let conn = self.open_writer()?;
618        let mut options = options.clone();
619        options.backup_options.get_or_insert_with(|| self.clone());
620        let report = migrate(&conn, Some(&self.path), migrations, &options)?;
621        Ok((conn, report))
622    }
623}
rightkit_db/migrate.rs

rightkit_db/
migrate.rs
