cratestack_sqlx/
migrations.rs

1//! Forward-only migration runner with a checksum guard against drift.
2//! Banks write migrations by hand (the contract under regulation is "the
3//! change is reviewable as a SQL diff").
4
5use crate::sqlx;
6use cratestack_core::CoolError;
7use sha2::{Digest, Sha256};
8
9pub const MIGRATIONS_TABLE_DDL: &str = r#"
10CREATE TABLE IF NOT EXISTS cratestack_migrations (
11    id TEXT PRIMARY KEY,
12    description TEXT NOT NULL,
13    checksum BYTEA NOT NULL,
14    applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
15);
16"#;
17
18/// A single migration step. The runner applies any rows not yet
19/// present in `cratestack_migrations`. `down` is recorded but never
20/// called — irreversible-by-default is the safe banking posture.
21#[derive(Debug, Clone)]
22pub struct Migration {
23    /// Sortable id, conventionally `YYYYMMDDHHMMSS_<slug>`.
24    pub id: String,
25    pub description: String,
26    pub up: String,
27    pub down: Option<String>,
28}
29
30impl Migration {
31    pub fn checksum(&self) -> [u8; 32] {
32        let mut hasher = Sha256::new();
33        hasher.update(self.id.as_bytes());
34        hasher.update(b"\0");
35        hasher.update(self.description.as_bytes());
36        hasher.update(b"\0");
37        hasher.update(self.up.as_bytes());
38        hasher.finalize().into()
39    }
40}
41
42#[derive(Debug, Clone, Copy, PartialEq, Eq)]
43pub enum MigrationStatus {
44    Pending,
45    Applied,
46    ChecksumMismatch,
47}
48
49#[derive(Debug, Clone)]
50pub struct MigrationState {
51    pub id: String,
52    pub status: MigrationStatus,
53}
54
55pub async fn ensure_migrations_table(pool: &sqlx::PgPool) -> Result<(), CoolError> {
56    // `raw_sql` sends the whole DDL block as one round-trip over PG's
57    // simple-query protocol, which understands `;`-separated statements
58    // (and dollar-quoting) natively — no client-side splitting needed.
59    sqlx::raw_sql(MIGRATIONS_TABLE_DDL)
60        .execute(pool)
61        .await
62        .map_err(|error| CoolError::Database(error.to_string()))?;
63    Ok(())
64}
65
66/// Inspect each migration in `migrations` against `cratestack_migrations`
67/// and report which are pending / applied / drifted. Use before `apply` to
68/// surface drift to the operator without changing state.
69pub async fn status(
70    pool: &sqlx::PgPool,
71    migrations: &[Migration],
72) -> Result<Vec<MigrationState>, CoolError> {
73    ensure_migrations_table(pool).await?;
74    let rows = sqlx::query_as::<_, (String, Vec<u8>)>(
75        "SELECT id, checksum FROM cratestack_migrations ORDER BY id",
76    )
77    .fetch_all(pool)
78    .await
79    .map_err(|error| CoolError::Database(error.to_string()))?;
80
81    let mut applied: std::collections::HashMap<String, Vec<u8>> = std::collections::HashMap::new();
82    for (id, checksum) in rows {
83        applied.insert(id, checksum);
84    }
85
86    Ok(migrations
87        .iter()
88        .map(|m| {
89            let id = m.id.clone();
90            match applied.get(&id) {
91                Some(stored) if stored.as_slice() == m.checksum().as_slice() => MigrationState {
92                    id,
93                    status: MigrationStatus::Applied,
94                },
95                Some(_) => MigrationState {
96                    id,
97                    status: MigrationStatus::ChecksumMismatch,
98                },
99                None => MigrationState {
100                    id,
101                    status: MigrationStatus::Pending,
102                },
103            }
104        })
105        .collect())
106}
107
108/// Apply every pending migration in the input slice in order. Each
109/// runs in its own transaction; checksum drift aborts the whole apply
110/// (banks treat drift as a release-process failure for humans, not a
111/// silent overwrite).
112pub async fn apply_pending(
113    pool: &sqlx::PgPool,
114    migrations: &[Migration],
115) -> Result<Vec<String>, CoolError> {
116    let states = status(pool, migrations).await?;
117    for (state, migration) in states.iter().zip(migrations) {
118        if state.status == MigrationStatus::ChecksumMismatch {
119            return Err(CoolError::Internal(format!(
120                "migration `{}` is recorded as applied but its SQL has changed; \
121                 resolve drift before continuing",
122                migration.id
123            )));
124        }
125    }
126
127    let mut applied = Vec::new();
128    for (state, migration) in states.iter().zip(migrations) {
129        if state.status != MigrationStatus::Pending {
130            continue;
131        }
132        let mut tx = pool
133            .begin()
134            .await
135            .map_err(|error| CoolError::Database(error.to_string()))?;
136        // `raw_sql` sends the whole `up` script as one batch over PG's
137        // simple-query protocol inside this transaction, so a mid-script
138        // failure can't leave partial state (and dollar-quoted PL/pgSQL
139        // bodies survive intact — no client-side `;` splitting, which
140        // would cut inside a `$$...$$` block).
141        sqlx::raw_sql(&migration.up)
142            .execute(&mut *tx)
143            .await
144            .map_err(|error| CoolError::Database(error.to_string()))?;
145        sqlx::query(
146            "INSERT INTO cratestack_migrations (id, description, checksum) VALUES ($1, $2, $3)",
147        )
148        .bind(&migration.id)
149        .bind(&migration.description)
150        .bind(migration.checksum().as_slice())
151        .execute(&mut *tx)
152        .await
153        .map_err(|error| CoolError::Database(error.to_string()))?;
154        tx.commit()
155            .await
156            .map_err(|error| CoolError::Database(error.to_string()))?;
157        applied.push(migration.id.clone());
158    }
159
160    Ok(applied)
161}
162
163#[cfg(test)]
164mod tests {
165    use super::*;
166
167    fn migration(id: &str, up: &str) -> Migration {
168        Migration {
169            id: id.to_owned(),
170            description: format!("migration {id}"),
171            up: up.to_owned(),
172            down: None,
173        }
174    }
175
176    #[test]
177    fn checksum_changes_when_up_sql_changes() {
178        let a = migration("20260101000000_init", "CREATE TABLE a (id INT);");
179        let mut b = a.clone();
180        b.up = "CREATE TABLE a (id BIGINT);".to_owned();
181        assert_ne!(a.checksum(), b.checksum());
182    }
183
184    #[test]
185    fn checksum_is_stable_for_same_inputs() {
186        let a = migration("20260101000000_init", "CREATE TABLE a (id INT);");
187        let b = a.clone();
188        assert_eq!(a.checksum(), b.checksum());
189    }
190}
cratestack_sqlx/migrations.rs

cratestack_sqlx/
migrations.rs
