cratefield_adapter_postgres/
migrate.rs

1//! The Postgres migration runner (issue #18): applies module migration
2//! sets in lock order, tracked in `harness_migrations(id, applied_at)`.
3//!
4//! The runner's own bookkeeping statements are built with sea-query and
5//! rendered with [`sea_query::PostgresQueryBuilder`] (ADR 0004); module
6//! migration files are plain SQL applied verbatim via the simple query
7//! protocol (they may hold several statements).
8
9use crate::convert::{bind_values, rebind_placeholders};
10use cratefield_core::{Database, DbError, Harness, Migrations, SqlMigration, Statement};
11use sea_query::{Alias, ColumnDef, PostgresQueryBuilder, Query, Table};
12use sqlx::postgres::{PgArguments, PgConnection};
13use std::collections::HashMap;
14
15/// Selects the migration set a module contributes to a Postgres
16/// deployment: its `postgres` set when it ships one, else its `sqlite`
17/// set — but only when every file passes the portable-SQL lint
18/// ([`cratefield_core::lint_portable_sql`], the same predicate `fz doctor`
19/// enforces, so the two can never drift).
20///
21/// # Errors
22///
23/// A human-readable message naming the offending migration file and token
24/// when the fallback `sqlite` set fails the lint.
25pub fn select_set(migrations: &Migrations) -> Result<&'static [SqlMigration], String> {
26    if !migrations.postgres.is_empty() {
27        return Ok(migrations.postgres);
28    }
29    for migration in migrations.sqlite {
30        let hits = cratefield_core::lint_portable_sql(migration.sql);
31        if let Some((token, explanation)) = hits.first() {
32            return Err(format!(
33                "sqlite migration {}/{} fails the portable-SQL lint (found {token:?}: \
34                 {explanation}); ship a migrations/postgres override instead",
35                migration.id, migration.name
36            ));
37        }
38    }
39    Ok(migrations.sqlite)
40}
41
42fn tracking_table_ddl() -> Statement {
43    let sql = Table::create()
44        .table(Alias::new("harness_migrations"))
45        .if_not_exists()
46        .col(ColumnDef::new(Alias::new("id")).text().primary_key())
47        .col(ColumnDef::new(Alias::new("applied_at")).text().not_null())
48        .col(ColumnDef::new(Alias::new("checksum")).text())
49        .build(PostgresQueryBuilder);
50    Statement::new(sql)
51}
52
53/// Databases migrated before checksums were recorded have the
54/// two-column table. Their existing rows stay NULL, which reads as
55/// "applied, cannot verify" rather than as a mismatch.
56fn tracking_table_backfill_ddl() -> Statement {
57    Statement::new(
58        "ALTER TABLE harness_migrations ADD COLUMN IF NOT EXISTS checksum text".to_owned(),
59    )
60}
61
62fn applied_query() -> Statement {
63    let (sql, values) = Query::select()
64        .column(Alias::new("id"))
65        .column(Alias::new("checksum"))
66        .from(Alias::new("harness_migrations"))
67        .build(PostgresQueryBuilder);
68    debug_assert!(values.0.is_empty(), "the applied query binds nothing");
69    Statement::new(sql)
70}
71
72/// Renders `INSERT INTO harness_migrations (id, applied_at)`
73/// `VALUES ($1, $2)` with the values as binds (`PostgresQueryBuilder`).
74fn record_insert(id: &str, applied_at: &str, checksum: &str) -> Result<Statement, DbError> {
75    let (sql, values) = Query::insert()
76        .into_table(Alias::new("harness_migrations"))
77        .columns([
78            Alias::new("id"),
79            Alias::new("applied_at"),
80            Alias::new("checksum"),
81        ])
82        .values([id.into(), applied_at.into(), checksum.into()])
83        .map_err(|err| DbError::Batch(err.to_string()))?
84        .build(PostgresQueryBuilder);
85    Ok(Statement::with_values(sql, values.0))
86}
87
88/// Executes one port statement on an open connection (pool connection or
89/// migration transaction). Statements rendered by
90/// `PostgresQueryBuilder` already carry `$n` placeholders, so
91/// [`rebind_placeholders`] is a no-op for them and only translates
92/// portable `?` statements.
93async fn exec_on(conn: &mut PgConnection, stmt: &Statement) -> Result<(), DbError> {
94    let sql = rebind_placeholders(&stmt.sql);
95    let mut args = PgArguments::default();
96    bind_values(&mut args, &stmt.values.0)?;
97    sqlx::query_with(sql.as_str(), args)
98        .execute(conn)
99        .await
100        .map_err(|err| DbError::Batch(first_line(&err.to_string()).to_owned()))?;
101    Ok(())
102}
103
104/// sqlx error strings can embed the whole offending statement; keep the
105/// first non-empty line for the error message.
106fn first_line(message: &str) -> &str {
107    message
108        .lines()
109        .find(|line| !line.is_empty())
110        .unwrap_or("unknown error")
111}
112
113fn iso_now() -> String {
114    time::OffsetDateTime::now_utc()
115        .format(&time::format_description::well_known::Rfc3339)
116        .unwrap_or_default()
117}
118
119impl crate::Postgres {
120    /// Applies a module's unapplied migrations in `id` order, tracked
121    /// under `<module>/<id>` in a `harness_migrations` table; each
122    /// migration runs in its own transaction so a failure rolls back
123    /// atomically and re-running is idempotent. Mirrors
124    /// `SqliteDatabase::apply_migrations` exactly.
125    ///
126    /// # Errors
127    ///
128    /// [`DbError::Batch`] when the tracking DDL or a migration's SQL
129    /// fails, or when the bookkeeping insert cannot be rendered.
130    pub async fn apply_migrations(
131        &self,
132        module: &str,
133        migrations: &[SqlMigration],
134    ) -> Result<(), DbError> {
135        self.execute(&tracking_table_ddl()).await?;
136        self.execute(&tracking_table_backfill_ddl()).await?;
137
138        let applied: HashMap<String, Option<String>> = self
139            .query(&applied_query())
140            .await?
141            .rows
142            .iter()
143            .filter_map(|row| Some((row.get::<String>("id")?, row.get::<String>("checksum"))))
144            .collect();
145
146        // Lock order: ids are zero-padded, so lexical order is apply order
147        // (the same discipline `fz migrations collect` numbers files by).
148        let mut ordered: Vec<&SqlMigration> = migrations.iter().collect();
149        ordered.sort_by_key(|migration| migration.id);
150
151        for migration in ordered {
152            let key = format!("{module}/{}", migration.id);
153            let checksum = cratefield_core::migration_checksum(migration.sql);
154            if let Some(recorded) = applied.get(&key) {
155                if let Some(recorded) = recorded.as_ref().filter(|hash| *hash != &checksum) {
156                    return Err(DbError::Batch(cratefield_core::migration_edited(
157                        &key, recorded, &checksum,
158                    )));
159                }
160                continue;
161            }
162            let mut tx = self
163                .pool
164                .begin()
165                .await
166                .map_err(|err| DbError::Batch(err.to_string()))?;
167            // Migration files may hold several statements; the simple
168            // query protocol (raw_sql) is the only multi-statement path.
169            sqlx::raw_sql(migration.sql)
170                .execute(&mut *tx)
171                .await
172                .map_err(|err| {
173                    DbError::Batch(format!(
174                        "migration {key} failed: {}",
175                        first_line(&err.to_string())
176                    ))
177                })?;
178            let record = record_insert(&key, &iso_now(), &checksum)?;
179            exec_on(&mut tx, &record).await?;
180            tx.commit()
181                .await
182                .map_err(|err| DbError::Batch(err.to_string()))?;
183        }
184        Ok(())
185    }
186
187    /// Applies every module's migrations for a whole harness, in lock
188    /// order: modules in harness (config) order, each set by zero-padded
189    /// migration id — the order `fz migrations collect` pins in
190    /// `.harness-lock.json`. For each module the `postgres` set applies
191    /// when shipped, else the `sqlite` set when it passes the portable
192    /// lint ([`select_set`]).
193    ///
194    /// # Errors
195    ///
196    /// [`DbError::Batch`] when a module's sqlite set fails the lint or any
197    /// migration fails to apply.
198    pub async fn apply_harness_migrations(&self, harness: &Harness) -> Result<(), DbError> {
199        for module in harness.modules() {
200            let set = select_set(&module.migrations())
201                .map_err(|reason| DbError::Batch(format!("module {}: {reason}", module.name())))?;
202            self.apply_migrations(module.name(), set).await?;
203        }
204        Ok(())
205    }
206}
207
208#[cfg(test)]
209mod tests {
210    use super::select_set;
211    use cratefield_core::{Migrations, SqlMigration};
212
213    const PORTABLE: SqlMigration = SqlMigration {
214        id: "0001",
215        name: "init",
216        sql: "CREATE TABLE t (id TEXT PRIMARY KEY, n INTEGER NOT NULL DEFAULT 0);",
217    };
218
219    #[test]
220    fn prefers_the_postgres_set_when_shipped() {
221        const PG: SqlMigration = SqlMigration {
222            id: "0001",
223            name: "init",
224            sql: "CREATE TABLE t (id TEXT PRIMARY KEY);",
225        };
226        let migrations = Migrations {
227            sqlite: &[PORTABLE],
228            postgres: &[PG],
229        };
230        let selected = select_set(&migrations).expect("postgres set wins");
231        assert_eq!(selected.len(), 1);
232        assert!(selected[0].sql.contains("TEXT"));
233    }
234
235    #[test]
236    fn falls_back_to_sqlite_that_passes_the_lint() {
237        let migrations = Migrations::sqlite(&[PORTABLE]);
238        let selected = select_set(&migrations).expect("portable sqlite applies");
239        assert_eq!(selected.len(), 1);
240        assert!(selected[0].sql.contains("TEXT"));
241    }
242
243    #[test]
244    fn rejects_sqlite_that_fails_the_lint() {
245        const NOT_PORTABLE: SqlMigration = SqlMigration {
246            id: "0002",
247            name: "oops",
248            sql: "CREATE TABLE t (id SERIAL PRIMARY KEY);",
249        };
250        let migrations = Migrations::sqlite(&[PORTABLE, NOT_PORTABLE]);
251        let err = select_set(&migrations).expect_err("SERIAL is not portable");
252        assert!(err.contains("0002/oops"), "names the file: {err}");
253        assert!(err.contains("SERIAL"), "names the token: {err}");
254    }
255}
cratefield_adapter_postgres/migrate.rs

cratefield_adapter_postgres/
migrate.rs
