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::CratestackError;
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, Default)]
22pub struct Migration {
23    /// Sortable id, conventionally `YYYYMMDDHHMMSS_<slug>`.
24    pub id: String,
25    pub description: String,
26    /// Preparatory SQL run immediately before [`Self::up`], in the
27    /// *same* transaction — the `up.pre.sql` half of a migration
28    /// directory, scaffolded by `cratestack migrate diff` whenever it
29    /// emits a blocking op and filled in by the operator.
30    ///
31    /// A separate field rather than text prepended to `up` so ownership
32    /// stays clean: `up.sql` is wholly generated, `up.pre.sql` wholly
33    /// hand-authored.
34    pub up_pre: Option<String>,
35    pub up: String,
36    pub down: Option<String>,
37}
38
39impl Migration {
40    pub fn checksum(&self) -> [u8; 32] {
41        let mut hasher = Sha256::new();
42        hasher.update(self.id.as_bytes());
43        hasher.update(b"\0");
44        hasher.update(self.description.as_bytes());
45        hasher.update(b"\0");
46        hasher.update(self.up.as_bytes());
47        // Mixed in only when present, so a migration without an
48        // `up.pre.sql` hashes byte-identically to how it did before
49        // `up_pre` existed. Hashing `None` as (say) an empty string
50        // plus a separator would change every checksum already
51        // recorded in `cratestack_migrations`, and every deployment
52        // upgrading to this version would see its entire applied
53        // history as `ChecksumMismatch` — drift where nothing drifted.
54        if let Some(up_pre) = &self.up_pre {
55            hasher.update(b"\0");
56            hasher.update(up_pre.as_bytes());
57        }
58        hasher.finalize().into()
59    }
60}
61
62#[derive(Debug, Clone, Copy, PartialEq, Eq)]
63pub enum MigrationStatus {
64    Pending,
65    Applied,
66    ChecksumMismatch,
67}
68
69#[derive(Debug, Clone)]
70pub struct MigrationState {
71    pub id: String,
72    pub status: MigrationStatus,
73}
74
75pub async fn ensure_migrations_table(pool: &sqlx::PgPool) -> Result<(), CratestackError> {
76    // `raw_sql` sends the whole DDL block as one round-trip over PG's
77    // simple-query protocol, which understands `;`-separated statements
78    // (and dollar-quoting) natively — no client-side splitting needed.
79    sqlx::raw_sql(MIGRATIONS_TABLE_DDL)
80        .execute(pool)
81        .await
82        .map_err(|error| CratestackError::Database(error.to_string()))?;
83    Ok(())
84}
85
86/// Inspect each migration in `migrations` against `cratestack_migrations`
87/// and report which are pending / applied / drifted. Use before `apply` to
88/// surface drift to the operator without changing state.
89pub async fn status(
90    pool: &sqlx::PgPool,
91    migrations: &[Migration],
92) -> Result<Vec<MigrationState>, CratestackError> {
93    ensure_migrations_table(pool).await?;
94    let rows = sqlx::query_as::<_, (String, Vec<u8>)>(
95        "SELECT id, checksum FROM cratestack_migrations ORDER BY id",
96    )
97    .fetch_all(pool)
98    .await
99    .map_err(|error| CratestackError::Database(error.to_string()))?;
100
101    let mut applied: std::collections::HashMap<String, Vec<u8>> = std::collections::HashMap::new();
102    for (id, checksum) in rows {
103        applied.insert(id, checksum);
104    }
105
106    Ok(migrations
107        .iter()
108        .map(|m| {
109            let id = m.id.clone();
110            match applied.get(&id) {
111                Some(stored) if stored.as_slice() == m.checksum().as_slice() => MigrationState {
112                    id,
113                    status: MigrationStatus::Applied,
114                },
115                Some(_) => MigrationState {
116                    id,
117                    status: MigrationStatus::ChecksumMismatch,
118                },
119                None => MigrationState {
120                    id,
121                    status: MigrationStatus::Pending,
122                },
123            }
124        })
125        .collect())
126}
127
128/// Apply every pending migration in the input slice in order. Each
129/// runs in its own transaction — [`Migration::up_pre`] then
130/// [`Migration::up`], both inside it — and checksum drift aborts the
131/// whole apply (banks treat drift as a release-process failure for
132/// humans, not a silent overwrite).
133pub async fn apply_pending(
134    pool: &sqlx::PgPool,
135    migrations: &[Migration],
136) -> Result<Vec<String>, CratestackError> {
137    let states = status(pool, migrations).await?;
138    for (state, migration) in states.iter().zip(migrations) {
139        if state.status == MigrationStatus::ChecksumMismatch {
140            return Err(CratestackError::Internal(format!(
141                "migration `{}` is recorded as applied but its SQL has changed; \
142                 resolve drift before continuing",
143                migration.id
144            )));
145        }
146    }
147
148    let mut applied = Vec::new();
149    for (state, migration) in states.iter().zip(migrations) {
150        if state.status != MigrationStatus::Pending {
151            continue;
152        }
153        let mut tx = pool
154            .begin()
155            .await
156            .map_err(|error| CratestackError::Database(error.to_string()))?;
157        // `up.pre.sql` first, in this same transaction. Its purpose is to
158        // make `up`'s blocking statement succeed, so a commit boundary
159        // between them would defeat it: that window is exactly when a
160        // concurrent INSERT could reintroduce the NULL a backfill just
161        // removed. Both halves land or neither does.
162        if let Some(up_pre) = &migration.up_pre {
163            sqlx::raw_sql(sqlx::AssertSqlSafe(up_pre.clone()))
164                .execute(&mut *tx)
165                .await
166                .map_err(|error| CratestackError::Database(error.to_string()))?;
167        }
168        // `raw_sql` sends the whole `up` script as one batch over PG's
169        // simple-query protocol inside this transaction, so a mid-script
170        // failure can't leave partial state (and dollar-quoted PL/pgSQL
171        // bodies survive intact — no client-side `;` splitting, which
172        // would cut inside a `$$...$$` block).
173        // `AssertSqlSafe`: `migration.up` *is* SQL by construction — the text
174        // of a migration file the operator ships. There is no bind-parameter
175        // alternative for a DDL batch (sqlx 0.9's `SqlSafeStr` bound).
176        sqlx::raw_sql(sqlx::AssertSqlSafe(migration.up.clone()))
177            .execute(&mut *tx)
178            .await
179            .map_err(|error| CratestackError::Database(error.to_string()))?;
180        sqlx::query(
181            "INSERT INTO cratestack_migrations (id, description, checksum) VALUES ($1, $2, $3)",
182        )
183        .bind(&migration.id)
184        .bind(&migration.description)
185        .bind(migration.checksum().as_slice())
186        .execute(&mut *tx)
187        .await
188        .map_err(|error| CratestackError::Database(error.to_string()))?;
189        tx.commit()
190            .await
191            .map_err(|error| CratestackError::Database(error.to_string()))?;
192        applied.push(migration.id.clone());
193    }
194
195    Ok(applied)
196}
197
198#[cfg(test)]
199mod tests;
cratestack_sqlx/migrations.rs

cratestack_sqlx/
migrations.rs
