persistant_sqlx/
migrate.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2//! Checksummed, ordered, resumable schema migrations that **keep sqlx's ledger**.
3//!
4//! The ledger is sqlx's own `_sqlx_migrations` table, written by sqlx's own `Migrator`, so its
5//! shape and its checksums (SHA-384 of the migration text) are exactly what `sqlx::migrate!` and
6//! `sqlx migrate run` write and read. That is a constraint, not a preference: Humboldt's transfer
7//! tool compares versions *and* checksums from that table (`docs/DESIGN-KINDS.md` §5), and a
8//! database can be handled by either this crate or sqlx's CLI without a conversion.
9//!
10//! What this layer adds is a typed outcome instead of an sqlx error string, and a report of what it
11//! applied.
12
13use crate::{PgBackend, classify};
14use persistant_relational::RelError;
15use sqlx::SqlSafeStr;
16use sqlx::migrate::{Migration as SqlxMigration, MigrationType, Migrator};
17use std::fmt;
18
19/// One migration: a version, a description and the SQL text. The text is hashed; changing a
20/// migration that has been applied is an error, never a silent re-run.
21#[derive(Debug, Clone, PartialEq, Eq)]
22pub struct Migration {
23    /// Positive, strictly increasing across a program's migrations (sqlx uses the timestamp or
24    /// sequence in the file name).
25    pub version: i64,
26    /// A short description, stored in the ledger.
27    pub description: String,
28    /// The statements, run in one transaction.
29    pub sql: String,
30}
31
32impl Migration {
33    /// A migration.
34    pub fn new(version: i64, description: impl Into<String>, sql: impl Into<String>) -> Self {
35        Migration {
36            version,
37            description: description.into(),
38            sql: sql.into(),
39        }
40    }
41}
42
43/// Why a migration run stopped.
44#[derive(Debug, Clone, PartialEq, Eq)]
45#[non_exhaustive]
46pub enum MigrateError {
47    /// A migration already applied has different text now (its checksum differs).
48    ChecksumMismatch {
49        /// The version whose text changed.
50        version: i64,
51    },
52    /// The database has applied a version this program does not know: it is older than the
53    /// database it is pointed at. The program refuses rather than run against a newer schema.
54    NewerThanProgram {
55        /// The unknown applied version.
56        version: i64,
57    },
58    /// A migration is recorded as partially applied (`success = false`); a person must fix it.
59    Dirty {
60        /// The version.
61        version: i64,
62    },
63    /// A migration's SQL failed; its transaction was rolled back.
64    Failed {
65        /// The version.
66        version: i64,
67        /// The database's classification of the failure.
68        error: RelError,
69    },
70    /// Anything else (connection, ledger access).
71    Backend(String),
72}
73
74impl fmt::Display for MigrateError {
75    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
76        match self {
77            MigrateError::ChecksumMismatch { version } => {
78                write!(f, "migration {version} was applied and has been modified")
79            }
80            MigrateError::NewerThanProgram { version } => write!(
81                f,
82                "the database has applied migration {version}, which this program does not know"
83            ),
84            MigrateError::Dirty { version } => {
85                write!(
86                    f,
87                    "migration {version} is partially applied; fix it by hand"
88                )
89            }
90            MigrateError::Failed { version, error } => {
91                write!(f, "migration {version} failed: {error}")
92            }
93            MigrateError::Backend(m) => write!(f, "migration: {m}"),
94        }
95    }
96}
97
98impl std::error::Error for MigrateError {}
99
100/// What a successful run did.
101#[derive(Debug, Clone, PartialEq, Eq)]
102pub struct MigrateReport {
103    /// Versions applied by this run, ascending.
104    pub applied: Vec<i64>,
105    /// Versions that were already applied and still match.
106    pub already_applied: Vec<i64>,
107}
108
109impl PgBackend {
110    /// Apply `migrations` in version order, skipping those already applied, in the ledger
111    /// `_sqlx_migrations` of the connection's current schema.
112    ///
113    /// Each migration runs in its own transaction. Refuses (applying nothing further) if an applied
114    /// migration's text has changed, if the database has applied a version the program does not
115    /// know, or if a migration is recorded as dirty.
116    pub async fn migrate(&self, migrations: &[Migration]) -> Result<MigrateReport, MigrateError> {
117        let sqlx_migrations: Vec<SqlxMigration> = migrations
118            .iter()
119            .map(|m| {
120                SqlxMigration::new(
121                    m.version,
122                    m.description.clone().into(),
123                    MigrationType::Simple,
124                    sqlx::AssertSqlSafe(m.sql.clone()).into_sql_str(),
125                    false,
126                )
127            })
128            .collect();
129        let before = self.applied_versions().await?;
130        let migrator = Migrator::with_migrations(sqlx_migrations);
131        migrator.run(&self.pool).await.map_err(|e| match e {
132            sqlx::migrate::MigrateError::VersionMismatch(version) => {
133                MigrateError::ChecksumMismatch { version }
134            }
135            sqlx::migrate::MigrateError::VersionMissing(version) => {
136                MigrateError::NewerThanProgram { version }
137            }
138            sqlx::migrate::MigrateError::Dirty(version) => MigrateError::Dirty { version },
139            sqlx::migrate::MigrateError::ExecuteMigration(e, version) => MigrateError::Failed {
140                version,
141                error: classify(e),
142            },
143            other => MigrateError::Backend(other.to_string()),
144        })?;
145        let after = self.applied_versions().await?;
146        let applied = after
147            .iter()
148            .copied()
149            .filter(|v| !before.contains(v))
150            .collect();
151        Ok(MigrateReport {
152            applied,
153            already_applied: before,
154        })
155    }
156
157    /// The versions the ledger says succeeded, ascending; empty if there is no ledger yet.
158    pub async fn applied_versions(&self) -> Result<Vec<i64>, MigrateError> {
159        let exists: bool = sqlx::query_scalar("SELECT to_regclass('_sqlx_migrations') IS NOT NULL")
160            .fetch_one(&self.pool)
161            .await
162            .map_err(|e| MigrateError::Backend(e.to_string()))?;
163        if !exists {
164            return Ok(Vec::new());
165        }
166        sqlx::query_scalar("SELECT version FROM _sqlx_migrations WHERE success ORDER BY version")
167            .fetch_all(&self.pool)
168            .await
169            .map_err(|e| MigrateError::Backend(e.to_string()))
170    }
171}
persistant_sqlx/migrate.rs

persistant_sqlx/
migrate.rs
