Skip to main content

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::recover::{
17    allowed, is_unreadable, quarantine_database_journaled, unfinished_quarantine, RecoveryPolicy,
18    QUARANTINE_JOURNAL_SUFFIX,
19};
20use crate::{DbError, DbOptions, Result};
21use rusqlite::{Connection, ErrorCode, OptionalExtension};
22use std::path::{Path, PathBuf};
23use std::time::{SystemTime, UNIX_EPOCH};
24
25pub enum Step {
26    Sql(&'static str),
27    /// Rust step (data backfills, table rebuilds); runs inside the step's transaction.
28    Code(fn(&Connection) -> rusqlite::Result<()>),
29}
30
31pub struct Migration {
32    pub version: i64,
33    pub name: &'static str,
34    pub up: Step,
35    /// Reverse SQL; `None` makes this step unrevertable (rollback refuses to guess).
36    pub down: Option<&'static str>,
37}
38
39impl Migration {
40    pub const fn sql(version: i64, name: &'static str, up: &'static str) -> Self {
41        Self {
42            version,
43            name,
44            up: Step::Sql(up),
45            down: None,
46        }
47    }
48
49    pub const fn with_down(mut self, down: &'static str) -> Self {
50        self.down = Some(down);
51        self
52    }
53
54    pub const fn code(
55        version: i64,
56        name: &'static str,
57        up: fn(&Connection) -> rusqlite::Result<()>,
58    ) -> Self {
59        Self {
60            version,
61            name,
62            up: Step::Code(up),
63            down: None,
64        }
65    }
66}
67
68/// Where migration versions are stored. Table storage uses Blueprint-compatible
69/// `key TEXT PRIMARY KEY, value TEXT NOT NULL` rows, leaving `user_version` untouched.
70#[derive(Clone, Debug, Default, PartialEq, Eq)]
71pub enum VersionStorage {
72    #[default]
73    UserVersion,
74    /// Table names must contain only ASCII letters, digits, or underscores.
75    /// Keys are bound parameters and may contain arbitrary text.
76    Table { table: String, key: String },
77    /// One row per applied version, shaped `"<version_column>" INTEGER PRIMARY KEY,
78    /// "<applied_at_column>" INTEGER NOT NULL` (ViewRight's `_migrations` shape). The applied
79    /// version is the highest stored `version_column`; a rollback deletes the top row. The table is
80    /// created on the first write when absent. Identifiers must be ASCII letters, digits, or
81    /// underscores. `applied_at` stores unix seconds.
82    VersionTable {
83        table: String,
84        version_column: String,
85        applied_at_column: String,
86    },
87}
88
89impl VersionStorage {
90    /// [`VersionStorage::VersionTable`] with the `version` and `applied_at` column names.
91    pub fn version_table(table: impl Into<String>) -> Self {
92        Self::VersionTable {
93            table: table.into(),
94            version_column: "version".into(),
95            applied_at_column: "applied_at".into(),
96        }
97    }
98}
99
100/// An app's existing one-row-per-version table, read once to adopt its applied version (see
101/// [`MigrateOptions::import_legacy_versions`]). Only the highest `version_column` value is used.
102#[derive(Clone, Debug, PartialEq, Eq)]
103pub struct LegacyVersionTable {
104    pub table: String,
105    pub version_column: String,
106}
107
108/// Checks a pre-migration backup before the first step runs; `Err(reason)` stops the migration.
109pub type VerifyBackup = fn(&Path) -> std::result::Result<(), String>;
110#[derive(Clone)]
111pub struct MigrateOptions {
112    /// Version location shared by reads and transactional migration writes.
113    pub version_storage: VersionStorage,
114    /// Take a pre-migration backup when migrating an existing (version > 0) on-disk database.
115    pub backup: bool,
116    /// Key source for the backup (SQLCipher); `DbOptions::open_writer_migrated` fills this in.
117    pub backup_options: Option<DbOptions>,
118    /// For a database newer than this build: `true` means the subset this build reads/writes is
119    /// intact, so open it read-write without migrating.
120    pub forward_compatible: Option<fn(&Connection) -> bool>,
121    /// Refresh planner statistics after migrating (`PRAGMA optimize`).
122    pub optimize_after: bool,
123    /// Adopt an existing app version table (see [`MigrateOptions::import_legacy_versions`]). Default none.
124    pub legacy_versions: Option<LegacyVersionTable>,
125    /// Name of the pre-migration backup, given the current and target versions. Default
126    /// `<db>.pre-v<target>.bak` beside the database. Must be a plain file name that is not the
127    /// database's own name (a name with a path separator or NUL is refused as
128    /// [`DbError::Backup`]).
129    pub backup_name: Option<fn(current: i64, target: i64) -> String>,
130    /// Checks the pre-migration backup before the first step runs. `Err(reason)` stops the
131    /// migration with [`DbError::Backup`]. The live database is untouched and the backup is kept.
132    pub verify_backup: Option<VerifyBackup>,
133    /// Keeps only the newest `n` pre-migration backups of this database, by modification time:
134    /// after each backup, older files named `<db>.pre-*.bak` in the same directory are removed.
135    /// The backup just taken is always kept. Removal is best effort and never fails the migration.
136    /// Default `None`: every backup is kept.
137    pub backup_retention: Option<usize>,
138}
139
140impl Default for MigrateOptions {
141    fn default() -> Self {
142        Self {
143            version_storage: VersionStorage::default(),
144            backup: true,
145            backup_options: None,
146            forward_compatible: None,
147            optimize_after: true,
148            legacy_versions: None,
149            backup_name: None,
150            verify_backup: None,
151            backup_retention: None,
152        }
153    }
154}
155
156impl MigrateOptions {
157    /// Adopt the applied version of an app's existing version table, once, so its migrations are
158    /// not re-run. Applies only when the configured storage reads version 0 and the legacy table's
159    /// highest version is above zero. The configured storage is then seeded to that version (a
160    /// `VersionTable` gets one row per version up to it), and the migrations up to it are skipped.
161    /// The legacy table is only read, never written. Idempotent: once seeded, the configured
162    /// storage is above zero, so later opens do not consult the legacy table. Rolling back to
163    /// version 0 while the legacy table still exists re-adopts it on the next open, so drop or
164    /// rename that table after adoption. A legacy version above this build's newest migration is
165    /// refused as [`DbError::NewerDatabase`] and nothing is written.
166    pub fn import_legacy_versions(mut self, legacy: LegacyVersionTable) -> Self {
167        self.legacy_versions = Some(legacy);
168        self
169    }
170}
171
172#[derive(Clone, Debug, Default, PartialEq, Eq)]
173pub struct MigrationReport {
174    pub from: i64,
175    pub to: i64,
176    pub applied: Vec<i64>,
177    pub backup: Option<PathBuf>,
178    /// Opened a newer-but-compatible database without migrating.
179    pub forward_compatible: bool,
180    /// A previous batch started but never finished (crash or failed step).
181    pub interrupted_previous: Option<PendingMarker>,
182}
183
184#[derive(Clone, Debug, PartialEq, Eq)]
185pub struct PendingMarker {
186    pub from: i64,
187    pub to: i64,
188    pub pid: u32,
189}
190
191pub fn user_version(conn: &Connection) -> Result<i64> {
192    Ok(conn.query_row("PRAGMA user_version", [], |row| row.get(0))?)
193}
194
195fn set_user_version(conn: &Connection, version: i64) -> rusqlite::Result<()> {
196    // `user_version` cannot be bound as a parameter; the value is an i64, never text.
197    conn.execute_batch(&format!("PRAGMA user_version = {version};"))
198}
199
200/// Read the selected migration version without creating a table or row.
201/// Missing table/key means zero; malformed or negative versions are refused.
202pub fn migration_version(conn: &Connection, storage: &VersionStorage) -> Result<i64> {
203    match storage {
204        VersionStorage::UserVersion => user_version(conn),
205        VersionStorage::Table { table, key } => key_value_version(conn, table, key),
206        VersionStorage::VersionTable {
207            table,
208            version_column,
209            ..
210        } => version_row_max(conn, table, version_column),
211    }
212}
213
214fn key_value_version(conn: &Connection, table: &str, key: &str) -> Result<i64> {
215    check_ident(table)?;
216    let exists: bool = conn.query_row(
217        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1 COLLATE NOCASE)",
218        [table],
219        |row| row.get(0),
220    )?;
221    if !exists {
222        return Ok(0);
223    }
224    let value: Option<String> = conn
225        .query_row(
226            &format!("SELECT CAST(\"value\" AS TEXT) FROM \"{table}\" WHERE \"key\"=?1"),
227            [key],
228            |row| row.get(0),
229        )
230        .optional()?;
231    match value {
232        None => Ok(0),
233        Some(value) => value
234            .parse::<i64>()
235            .ok()
236            .filter(|version| *version >= 0)
237            .ok_or_else(|| DbError::InvalidMigrations(format!("invalid stored version {value:?}"))),
238    }
239}
240
241// Called only inside the migration/down transaction, after identifier validation.
242fn set_key_value_version(
243    conn: &Connection,
244    storage: &VersionStorage,
245    version: i64,
246) -> rusqlite::Result<()> {
247    match storage {
248        VersionStorage::UserVersion => set_user_version(conn, version),
249        VersionStorage::Table { table, key } => {
250            conn.execute_batch(&format!(
251                "CREATE TABLE IF NOT EXISTS \"{table}\" (\"key\" TEXT PRIMARY KEY, \"value\" TEXT NOT NULL);"
252            ))?;
253            conn.execute(
254                &format!(
255                    "INSERT INTO \"{table}\"(\"key\",\"value\") VALUES(?1,?2) \
256                     ON CONFLICT(\"key\") DO UPDATE SET \"value\"=excluded.\"value\""
257                ),
258                rusqlite::params![key, version.to_string()],
259            )?;
260            Ok(())
261        }
262        // Routed to `advance_version` / `retreat_version`, never reached through them.
263        VersionStorage::VersionTable { .. } => Err(rusqlite::Error::InvalidQuery),
264    }
265}
266
267/// Make `version` the applied version: a forward step, inside its transaction.
268fn advance_version(
269    conn: &Connection,
270    storage: &VersionStorage,
271    version: i64,
272) -> rusqlite::Result<()> {
273    match storage {
274        VersionStorage::VersionTable {
275            table,
276            version_column,
277            applied_at_column,
278        } => {
279            conn.execute_batch(&format!(
280                "CREATE TABLE IF NOT EXISTS \"{table}\" (\"{version_column}\" INTEGER PRIMARY KEY, \"{applied_at_column}\" INTEGER NOT NULL);"
281            ))?;
282            conn.execute(
283                &format!("DELETE FROM \"{table}\" WHERE \"{version_column}\" = ?1"),
284                [version],
285            )?;
286            conn.execute(
287                &format!(
288                    "INSERT INTO \"{table}\" (\"{version_column}\", \"{applied_at_column}\") VALUES (?1, ?2)"
289                ),
290                rusqlite::params![version, now_secs()],
291            )?;
292            Ok(())
293        }
294        _ => set_key_value_version(conn, storage, version),
295    }
296}
297
298/// Un-apply `version`: a rollback step, so the applied version becomes `version - 1`.
299fn retreat_version(
300    conn: &Connection,
301    storage: &VersionStorage,
302    version: i64,
303) -> rusqlite::Result<()> {
304    match storage {
305        VersionStorage::VersionTable {
306            table,
307            version_column,
308            ..
309        } => {
310            conn.execute(
311                &format!("DELETE FROM \"{table}\" WHERE \"{version_column}\" = ?1"),
312                [version],
313            )?;
314            Ok(())
315        }
316        _ => set_key_value_version(conn, storage, version - 1),
317    }
318}
319
320/// Seed a database at version 0 with `through` as its applied version (legacy adoption). A
321/// `VersionTable` gets one row per version; the other storages take the single value.
322fn adopt_legacy_version(conn: &Connection, storage: &VersionStorage, through: i64) -> Result<()> {
323    conn.execute_batch("BEGIN IMMEDIATE;")?;
324    let seeded = match storage {
325        VersionStorage::VersionTable { .. } => {
326            (1..=through).try_for_each(|version| advance_version(conn, storage, version))
327        }
328        _ => advance_version(conn, storage, through),
329    };
330    match seeded {
331        Ok(()) => {
332            conn.execute_batch("COMMIT;")?;
333            Ok(())
334        }
335        Err(error) => {
336            let _ = conn.execute_batch("ROLLBACK;");
337            Err(DbError::Sqlite(error))
338        }
339    }
340}
341
342/// Highest version stored in a one-row-per-version table; zero when the table is absent or empty.
343fn version_row_max(conn: &Connection, table: &str, version_column: &str) -> Result<i64> {
344    check_ident(table)?;
345    check_ident(version_column)?;
346    let exists: bool = conn.query_row(
347        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1 COLLATE NOCASE)",
348        [table],
349        |row| row.get(0),
350    )?;
351    if !exists {
352        return Ok(0);
353    }
354    let highest: Option<i64> = conn.query_row(
355        &format!("SELECT MAX(CAST(\"{version_column}\" AS INTEGER)) FROM \"{table}\""),
356        [],
357        |row| row.get(0),
358    )?;
359    let version = highest.unwrap_or(0);
360    if version < 0 {
361        return Err(DbError::InvalidMigrations(format!(
362            "invalid stored version {version}"
363        )));
364    }
365    Ok(version)
366}
367
368/// Refuses any identifier that is not a plain ASCII name, before SQL is built from it.
369fn check_storage(storage: &VersionStorage) -> Result<()> {
370    match storage {
371        VersionStorage::UserVersion => Ok(()),
372        VersionStorage::Table { table, .. } => check_ident(table),
373        VersionStorage::VersionTable {
374            table,
375            version_column,
376            applied_at_column,
377        } => {
378            check_ident(table)?;
379            check_ident(version_column)?;
380            check_ident(applied_at_column)
381        }
382    }
383}
384
385/// Where the pre-migration backup goes: `<db>.pre-v<target>.bak`, or the name from
386/// [`MigrateOptions::backup_name`]. The name must be a plain file name other than the database's.
387fn backup_path(db: &Path, options: &MigrateOptions, current: i64, target: i64) -> Result<PathBuf> {
388    let name = db
389        .file_name()
390        .ok_or(DbError::InvalidDatabasePath)?
391        .to_string_lossy()
392        .into_owned();
393    let file = match options.backup_name {
394        Some(namer) => namer(current, target),
395        None => format!("{name}.pre-v{target}.bak"),
396    };
397    if file.is_empty() || file.contains(['/', '\\', '\0']) || file == "." || file == ".." {
398        return Err(DbError::Backup(format!(
399            "backup name {file:?} must be a plain file name other than the database's"
400        )));
401    }
402    // The backup is published with a replacing rename, so it must never land on the database,
403    // a live sidecar, or a recovery file this crate keeps beside the database. Compared
404    // case-insensitively: a case-insensitive filesystem treats those names as the same file.
405    let lowered = file.to_lowercase();
406    let reserved = [
407        "",
408        "-wal",
409        "-shm",
410        "-journal",
411        ".migrating",
412        ".restore-journal",
413        ".store-lock",
414        ".transition-journal",
415        QUARANTINE_JOURNAL_SUFFIX,
416    ]
417    .iter()
418    .any(|suffix| lowered == format!("{name}{suffix}").to_lowercase());
419    if reserved {
420        return Err(DbError::Backup(format!(
421            "backup name {file:?} is the database, one of its sidecars or a recovery file; choose another name"
422        )));
423    }
424    Ok(db.with_file_name(file))
425}
426
427/// Keeps the `keep` newest pre-migration backups of `db`, counting `newest` (the backup just
428/// taken), and removes the older `<db>.pre-*.bak` files beside it. Best effort: a file that
429/// cannot be removed is left in place.
430fn prune_backups(db: &Path, newest: &Path, keep: usize) {
431    let Some(name) = db.file_name().map(|n| n.to_string_lossy().into_owned()) else {
432        return;
433    };
434    let dir = db
435        .parent()
436        .filter(|p| !p.as_os_str().is_empty())
437        .unwrap_or(Path::new("."));
438    let prefix = format!("{name}.pre-");
439    let Ok(entries) = std::fs::read_dir(dir) else {
440        return;
441    };
442    let mut older: Vec<(SystemTime, PathBuf)> = entries
443        .filter_map(|entry| entry.ok().map(|entry| entry.path()))
444        .filter(|path| path.as_path() != newest)
445        .filter(|path| {
446            path.file_name().is_some_and(|n| {
447                let n = n.to_string_lossy();
448                n.starts_with(&prefix) && n.ends_with(".bak")
449            })
450        })
451        .filter_map(|path| {
452            let modified = std::fs::metadata(&path).and_then(|m| m.modified()).ok()?;
453            Some((modified, path))
454        })
455        .collect();
456    // Newest first; the just-taken backup already fills one of the `keep` places.
457    older.sort_by_key(|(modified, _)| std::cmp::Reverse(*modified));
458    for (_, path) in older.into_iter().skip(keep.saturating_sub(1)) {
459        let _ = std::fs::remove_file(path);
460    }
461}
462
463fn now_secs() -> i64 {
464    SystemTime::now()
465        .duration_since(UNIX_EPOCH)
466        .map(|d| d.as_secs() as i64)
467        .unwrap_or(0)
468}
469
470pub fn validate(migrations: &[Migration]) -> Result<()> {
471    for (index, migration) in migrations.iter().enumerate() {
472        let expected = index as i64 + 1;
473        if migration.version != expected {
474            return Err(DbError::InvalidMigrations(format!(
475                "expected version {expected} at position {index}, found {} ({})",
476                migration.version, migration.name
477            )));
478        }
479    }
480    Ok(())
481}
482
483fn marker_path(db: &Path) -> PathBuf {
484    let mut name = db.as_os_str().to_os_string();
485    name.push(".migrating");
486    PathBuf::from(name)
487}
488
489pub fn pending_marker(db: &Path) -> Option<PendingMarker> {
490    let text = std::fs::read_to_string(marker_path(db)).ok()?;
491    let mut marker = PendingMarker {
492        from: 0,
493        to: 0,
494        pid: 0,
495    };
496    for line in text.lines() {
497        match line.split_once('=')? {
498            ("from", v) => marker.from = v.parse().ok()?,
499            ("to", v) => marker.to = v.parse().ok()?,
500            ("pid", v) => marker.pid = v.parse().ok()?,
501            _ => {}
502        }
503    }
504    Some(marker)
505}
506
507/// Unix seconds when the pending `<db>.migrating` batch started, `0` for a marker written
508/// before the start time was recorded, `None` without a readable marker. A separate reader keeps
509/// [`PendingMarker`] constructible as before.
510pub fn pending_marker_started_at(db: &Path) -> Option<u64> {
511    pending_marker(db)?;
512    let text = std::fs::read_to_string(marker_path(db)).ok()?;
513    Some(
514        text.lines()
515            .find_map(|line| line.strip_prefix("started="))
516            .and_then(|v| v.parse().ok())
517            .unwrap_or(0),
518    )
519}
520
521/// Writes the `<db>.migrating` marker for a batch that moves the schema from version `from` to
522/// `to`. The write is atomic and records the writing process id and the start time (unix
523/// seconds). [`migrate`] writes this same marker; [`pending_marker`] reads it back. Use it to
524/// simulate an interrupted migration in a crash test, instead of a private copy of the format.
525pub fn write_pending_marker(db: &Path, from: i64, to: i64) -> Result<()> {
526    let body = format!(
527        "from={from}\nto={to}\npid={}\nstarted={}\n",
528        std::process::id(),
529        now_secs().max(0)
530    );
531    rightkit_fs::atomic_write(&marker_path(db), body.as_bytes())?;
532    Ok(())
533}
534
535/// True when the version stored under `storage` is above the newest of `migrations`, which means
536/// a newer build wrote this database. It only reads: nothing is migrated, created or written. It
537/// does not consult [`MigrateOptions::import_legacy_versions`].
538pub fn stored_version_is_newer(
539    conn: &Connection,
540    storage: &VersionStorage,
541    migrations: &[Migration],
542) -> Result<bool> {
543    validate(migrations)?;
544    check_storage(storage)?;
545    let max = migrations.last().map_or(0, |m| m.version);
546    Ok(migration_version(conn, storage)? > max)
547}
548
549pub fn migrate(
550    conn: &Connection,
551    db_path: Option<&Path>,
552    migrations: &[Migration],
553    options: &MigrateOptions,
554) -> Result<MigrationReport> {
555    validate(migrations)?;
556    check_storage(&options.version_storage)?;
557    let mut current = migration_version(conn, &options.version_storage)?;
558    let max = migrations.last().map(|m| m.version).unwrap_or(0);
559    let mut report = MigrationReport {
560        from: current,
561        to: current,
562        ..Default::default()
563    };
564    report.interrupted_previous = db_path.and_then(pending_marker);
565
566    // Legacy adoption happens only from version 0 (see `MigrateOptions::import_legacy_versions`).
567    if let Some(legacy) = options.legacy_versions.as_ref().filter(|_| current == 0) {
568        let legacy_version = version_row_max(conn, &legacy.table, &legacy.version_column)?;
569        if legacy_version > 0 {
570            if legacy_version <= max {
571                adopt_legacy_version(conn, &options.version_storage, legacy_version)?;
572            }
573            current = legacy_version;
574            report.from = current;
575            report.to = current;
576        }
577    }
578
579    if current > max {
580        if options.forward_compatible.is_some_and(|probe| probe(conn)) {
581            report.forward_compatible = true;
582            return Ok(report);
583        }
584        return Err(DbError::NewerDatabase { current, max });
585    }
586    let pending: Vec<&Migration> = migrations.iter().filter(|m| m.version > current).collect();
587    if pending.is_empty() {
588        return Ok(report);
589    }
590
591    if current > 0 && options.backup {
592        if let Some(path) = db_path {
593            let backup = backup_path(path, options, current, max)?;
594            let backup_options = options
595                .backup_options
596                .clone()
597                .unwrap_or_else(|| DbOptions::new(path));
598            crate::online_backup_to(&backup_options, conn, &backup, None)?;
599            if let Some(verify) = options.verify_backup {
600                verify(&backup).map_err(|reason| {
601                    DbError::Backup(format!("pre-migration backup failed its check: {reason}"))
602                })?;
603            }
604            if let Some(keep) = options.backup_retention {
605                prune_backups(path, &backup, keep);
606            }
607            report.backup = Some(backup);
608        }
609    }
610
611    if let Some(path) = db_path {
612        write_pending_marker(path, current, max)?;
613    }
614
615    for migration in pending {
616        conn.execute_batch("BEGIN IMMEDIATE;")?;
617        let applied = match &migration.up {
618            Step::Sql(sql) => conn.execute_batch(sql),
619            Step::Code(run) => run(conn),
620        }
621        .and_then(|()| advance_version(conn, &options.version_storage, migration.version));
622        match applied {
623            Ok(()) => {
624                conn.execute_batch("COMMIT;")?;
625                report.applied.push(migration.version);
626                report.to = migration.version;
627            }
628            Err(source) => {
629                let _ = conn.execute_batch("ROLLBACK;");
630                return Err(DbError::MigrationFailed {
631                    version: migration.version,
632                    name: migration.name,
633                    source,
634                });
635            }
636        }
637    }
638
639    if let Some(path) = db_path {
640        let _ = std::fs::remove_file(marker_path(path));
641    }
642    if options.optimize_after {
643        crate::optimize_boundary(conn)?;
644    }
645    Ok(report)
646}
647
648/// Revert to `target` by replaying each step's `down`, highest first, one transaction per step.
649/// Refuses before touching anything unless every step in the range has a `down`.
650pub fn rollback_to(conn: &Connection, migrations: &[Migration], target: i64) -> Result<Vec<i64>> {
651    rollback_to_with_storage(conn, migrations, target, &VersionStorage::UserVersion)
652}
653
654/// Revert using the same version storage selected for [`migrate`].
655pub fn rollback_to_with_storage(
656    conn: &Connection,
657    migrations: &[Migration],
658    target: i64,
659    storage: &VersionStorage,
660) -> Result<Vec<i64>> {
661    validate(migrations)?;
662    check_storage(storage)?;
663    if target < 0 {
664        return Err(DbError::Rollback(
665            "target version must be nonnegative".into(),
666        ));
667    }
668    let current = migration_version(conn, storage)?;
669    if current <= target {
670        return Err(DbError::Rollback(format!(
671            "already at v{current}, at or below target v{target}"
672        )));
673    }
674    let mut plan = Vec::new();
675    for version in ((target + 1)..=current).rev() {
676        let migration = migrations
677            .iter()
678            .find(|m| m.version == version)
679            .ok_or_else(|| DbError::Rollback(format!("v{version} is not a known migration")))?;
680        let down = migration.down.ok_or_else(|| {
681            DbError::Rollback(format!(
682                "v{version} ({}) has no down migration",
683                migration.name
684            ))
685        })?;
686        plan.push((version, down));
687    }
688    let mut reverted = Vec::new();
689    for (version, down) in plan {
690        conn.execute_batch("BEGIN IMMEDIATE;")?;
691        let result = conn
692            .execute_batch(down)
693            .and_then(|()| retreat_version(conn, storage, version));
694        match result {
695            Ok(()) => {
696                conn.execute_batch("COMMIT;")?;
697                reverted.push(version);
698            }
699            Err(error) => {
700                let _ = conn.execute_batch("ROLLBACK;");
701                return Err(DbError::Rollback(format!(
702                    "down v{version} failed and was rolled back: {error}"
703                )));
704            }
705        }
706    }
707    Ok(reverted)
708}
709
710/// True if `table` has `column`, including STORED/VIRTUAL generated columns (`table_xinfo`;
711/// `table_info` omits them, which made MailRight re-run an ALTER on every open).
712pub fn column_exists(conn: &Connection, table: &str, column: &str) -> Result<bool> {
713    check_ident(table)?;
714    let mut statement = conn.prepare(&format!("PRAGMA table_xinfo({table})"))?;
715    let names = statement.query_map([], |row| row.get::<_, String>(1))?;
716    for name in names {
717        if name? == column {
718            return Ok(true);
719        }
720    }
721    Ok(false)
722}
723
724/// Idempotent `ALTER TABLE .. ADD COLUMN` (SQLite has no `IF NOT EXISTS` for columns).
725/// `table`, `column` and `decl` must be literals from the app, never user input.
726pub fn add_column_if_missing(
727    conn: &Connection,
728    table: &str,
729    column: &str,
730    decl: &str,
731) -> Result<()> {
732    check_ident(table)?;
733    check_ident(column)?;
734    if !column_exists(conn, table, column)? {
735        conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {column} {decl};"))?;
736    }
737    Ok(())
738}
739
740fn check_ident(ident: &str) -> Result<()> {
741    if !ident.is_empty()
742        && ident
743            .bytes()
744            .all(|b| b.is_ascii_alphanumeric() || b == b'_')
745    {
746        Ok(())
747    } else {
748        Err(DbError::InvalidMigrations(format!(
749            "unsafe SQL identifier {ident:?}"
750        )))
751    }
752}
753
754/// What [`DbOptions::open_writer_migrated_or_degrade`] returns.
755#[derive(Debug)]
756pub struct OpenOrDegrade {
757    /// The writer to use: the database at the path, or a private in-memory store when degraded.
758    pub conn: Connection,
759    /// The migration report for the connection in `conn`.
760    pub report: MigrationReport,
761    /// `Some` when `conn` is not the database as it was stored: it was quarantined and recreated,
762    /// or the app runs on an in-memory store. The app should show it to the user.
763    pub degraded: Option<Degraded>,
764}
765
766/// Why [`DbOptions::open_writer_migrated_or_degrade`] did not use the stored database as-is.
767#[derive(Clone, Debug, PartialEq, Eq)]
768pub struct Degraded {
769    /// The failure that made the stored database unusable.
770    pub reason: String,
771    /// Where the unusable file went, with its sidecars, when it was quarantined. Also set when a
772    /// quarantine stopped part way and could not be undone (see `quarantine_incomplete`): then
773    /// it is where the main file is now.
774    pub quarantined: Option<PathBuf>,
775    /// `true` when a quarantine moved part of the stored database and neither finished nor rolled
776    /// back. The files are split between the original path and `quarantined`; a
777    /// `<db>.quarantine-journal` file records the move and makes later opens through
778    /// [`DbOptions::open_writer_migrated_or_degrade`] stay in memory instead of creating a
779    /// replacement over the split files. A person must reunite or remove them.
780    pub quarantine_incomplete: bool,
781    /// `true` when the connection is a private in-memory store, not a file at the path.
782    pub in_memory: bool,
783}
784
785impl DbOptions {
786    /// Open the writer and bring the schema up to date (see [`migrate`]); the pre-migration
787    /// backup is keyed like the database itself.
788    pub fn open_writer_migrated(
789        &self,
790        migrations: &[Migration],
791        options: &MigrateOptions,
792    ) -> Result<(Connection, MigrationReport)> {
793        let conn = self.open_writer()?;
794        let mut options = options.clone();
795        options.backup_options.get_or_insert_with(|| self.clone());
796        let report = match migrate(&conn, Some(&self.path), migrations, &options) {
797            Ok(report) => report,
798            Err(err) => {
799                // Corruption can first surface here, after `open_at` re-enabled the close
800                // checkpoint. Closing this connection must not checkpoint or delete the `-wal`
801                // before `open_writer_migrated_or_degrade` moves the store aside (rightkit#932).
802                let _ = conn.set_db_config(
803                    rusqlite::config::DbConfig::SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE,
804                    true,
805                );
806                return Err(err);
807            }
808        };
809        Ok((conn, report))
810    }
811
812    /// Open-or-degrade: [`Self::open_writer_migrated`], with recovery for a stored database that
813    /// is corrupt, not a database, or fails its integrity check, at open or during migration.
814    ///
815    /// For such a failure, `policy` decides as in [`Self::open_writer_recovering`]: when it allows
816    /// (`quarantine_corrupt` for a plain database, `quarantine_keyed` for a keyed one), the file
817    /// and its sidecars move aside under the policy's stamp and sidecar options (never deleted by
818    /// default), and the database is created and migrated again. When it does not allow, the file
819    /// is left untouched. Either way, if the store still cannot be opened, the connection is a
820    /// private in-memory store that is migrated the same way only when
821    /// `policy.allow_in_memory_fallback` explicitly acknowledges it (default refuse: the
822    /// original error is returned). When acknowledged, [`OpenOrDegrade::degraded`] says so
823    /// with `in_memory: true`, and the configured key remains applied.
824    ///
825    /// Any other failure (busy, permission, I/O, a newer database, an invalid migration list) is
826    /// returned as an error, because it does not show the stored data is damaged.
827    pub fn open_writer_migrated_or_degrade(
828        &self,
829        migrations: &[Migration],
830        options: &MigrateOptions,
831        policy: RecoveryPolicy,
832    ) -> Result<OpenOrDegrade> {
833        // A quarantine an earlier run could not finish or undo: never create a replacement over
834        // the split files.
835        if let Some(journal) = unfinished_quarantine(&self.path) {
836            let reason = format!(
837                "an earlier quarantine of {} stopped part way and was not undone; see {}",
838                self.path.display(),
839                journal.display()
840            );
841            let refuse = DbError::Fs(rightkit_fs::FsError::io(
842                &journal,
843                std::io::Error::other(reason.clone()),
844            ));
845            return self.degrade_in_memory(migrations, options, policy, reason, None, true, refuse);
846        }
847        let error = match self.open_writer_migrated(migrations, options) {
848            Ok((conn, report)) => {
849                return Ok(OpenOrDegrade {
850                    conn,
851                    report,
852                    degraded: None,
853                })
854            }
855            Err(error) if is_recoverable_damage(&error) => error,
856            Err(error) => return Err(error),
857        };
858        let reason = error.to_string();
859        if !allowed(self, policy) {
860            return self.degrade_in_memory(migrations, options, policy, reason, None, false, error);
861        }
862        let quarantined = match quarantine_database_journaled(
863            &self.path,
864            policy.quarantine_stamp,
865            policy.sidecars,
866        ) {
867            Ok(moved) => moved,
868            Err(failure) => {
869                // Nothing moved (or every move was undone): the store is untouched. Otherwise
870                // the split stays on disk, journaled, and the result says so.
871                let reason = format!("{reason}; quarantine failed: {}", failure.error);
872                let incomplete = failure.stranded_at.is_some();
873                return self.degrade_in_memory(
874                    migrations,
875                    options,
876                    policy,
877                    reason,
878                    failure.stranded_at,
879                    incomplete,
880                    failure.error,
881                );
882            }
883        };
884        match self.open_writer_migrated(migrations, options) {
885            Ok((conn, report)) => Ok(OpenOrDegrade {
886                conn,
887                report,
888                degraded: Some(Degraded {
889                    reason,
890                    quarantined: Some(quarantined),
891                    quarantine_incomplete: false,
892                    in_memory: false,
893                }),
894            }),
895            Err(second) => {
896                let reason = format!("{reason}; reopen after quarantine failed: {second}");
897                self.degrade_in_memory(
898                    migrations,
899                    options,
900                    policy,
901                    reason,
902                    Some(quarantined),
903                    false,
904                    second,
905                )
906            }
907        }
908    }
909
910    #[allow(clippy::too_many_arguments)]
911    fn degrade_in_memory(
912        &self,
913        migrations: &[Migration],
914        options: &MigrateOptions,
915        policy: RecoveryPolicy,
916        reason: String,
917        quarantined: Option<PathBuf>,
918        quarantine_incomplete: bool,
919        refuse: DbError,
920    ) -> Result<OpenOrDegrade> {
921        // In-memory recovery is never automatic: without explicit acknowledgement the
922        // caller gets the original failure, not a store that will not persist.
923        if !policy.allow_in_memory_fallback {
924            return Err(refuse);
925        }
926        let conn = self.open_in_memory()?;
927        let report = migrate(&conn, None, migrations, options)?;
928        Ok(OpenOrDegrade {
929            conn,
930            report,
931            degraded: Some(Degraded {
932                reason,
933                quarantined,
934                quarantine_incomplete,
935                in_memory: true,
936            }),
937        })
938    }
939}
940
941/// Damage that recovery may act on: an unreadable or failed-integrity open (see
942/// [`is_unreadable`]), or a migration step that SQLite reports as not a database or corrupt.
943fn is_recoverable_damage(error: &DbError) -> bool {
944    match error {
945        DbError::MigrationFailed {
946            source: rusqlite::Error::SqliteFailure(failure, _),
947            ..
948        } => matches!(
949            failure.code,
950            ErrorCode::NotADatabase | ErrorCode::DatabaseCorrupt
951        ),
952        other => is_unreadable(other),
953    }
954}