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
42pub(crate) fn 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.
56pub(crate) fn 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_error_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.
106pub(crate) fn first_error_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            if !migration.transactional {
163                // Runs alone on the pool (autocommit), then the tracking
164                // row (RECONCILIATION.md §4). The SQL must be idempotent:
165                // a crash between the two steps re-runs it on the next
166                // boot.
167                if !cratefield_core::is_idempotent_sql(migration.sql) {
168                    return Err(DbError::Batch(cratefield_core::migration_missing_guard(
169                        &key,
170                    )));
171                }
172                sqlx::raw_sql(migration.sql)
173                    .execute(&self.pool)
174                    .await
175                    .map_err(|err| {
176                        DbError::Batch(format!(
177                            "migration {key} failed: {}",
178                            first_error_line(&err.to_string())
179                        ))
180                    })?;
181                let record = record_insert(&key, &iso_now(), &checksum)?;
182                self.execute(&record).await?;
183                continue;
184            }
185            let mut tx = self
186                .pool
187                .begin()
188                .await
189                .map_err(|err| DbError::Batch(err.to_string()))?;
190            // Migration files may hold several statements; the simple
191            // query protocol (raw_sql) is the only multi-statement path.
192            sqlx::raw_sql(migration.sql)
193                .execute(&mut *tx)
194                .await
195                .map_err(|err| {
196                    DbError::Batch(format!(
197                        "migration {key} failed: {}",
198                        first_error_line(&err.to_string())
199                    ))
200                })?;
201            let record = record_insert(&key, &iso_now(), &checksum)?;
202            exec_on(&mut tx, &record).await?;
203            tx.commit()
204                .await
205                .map_err(|err| DbError::Batch(err.to_string()))?;
206        }
207        Ok(())
208    }
209
210    /// Applies every module's migrations for a whole harness, in lock
211    /// order: modules in harness (config) order, each set by zero-padded
212    /// migration id — the order `fz migrations collect` pins in
213    /// `.harness-lock.json`. For each module the `postgres` set applies
214    /// when shipped, else the `sqlite` set when it passes the portable
215    /// lint ([`select_set`]).
216    ///
217    /// # Errors
218    ///
219    /// [`DbError::Batch`] when a module's sqlite set fails the lint or any
220    /// migration fails to apply.
221    pub async fn apply_harness_migrations(&self, harness: &Harness) -> Result<(), DbError> {
222        for module in harness.modules() {
223            let set = select_set(&module.migrations())
224                .map_err(|reason| DbError::Batch(format!("module {}: {reason}", module.name())))?;
225            self.apply_migrations(module.name(), set).await?;
226        }
227        Ok(())
228    }
229}
230
231#[cfg(test)]
232mod tests {
233    use super::select_set;
234    use cratefield_core::{Migrations, SqlMigration};
235
236    const PORTABLE: SqlMigration = SqlMigration::new(
237        "0001",
238        "init",
239        "CREATE TABLE t (id TEXT PRIMARY KEY, n INTEGER NOT NULL DEFAULT 0);",
240    );
241
242    #[test]
243    fn prefers_the_postgres_set_when_shipped() {
244        const PG: SqlMigration =
245            SqlMigration::new("0001", "init", "CREATE TABLE t (id TEXT PRIMARY KEY);");
246        let migrations = Migrations {
247            sqlite: &[PORTABLE],
248            postgres: &[PG],
249        };
250        let selected = select_set(&migrations).expect("postgres set wins");
251        assert_eq!(selected.len(), 1);
252        assert!(selected[0].sql.contains("TEXT"));
253    }
254
255    #[test]
256    fn falls_back_to_sqlite_that_passes_the_lint() {
257        let migrations = Migrations::sqlite(&[PORTABLE]);
258        let selected = select_set(&migrations).expect("portable sqlite applies");
259        assert_eq!(selected.len(), 1);
260        assert!(selected[0].sql.contains("TEXT"));
261    }
262
263    #[test]
264    fn rejects_sqlite_that_fails_the_lint() {
265        const NOT_PORTABLE: SqlMigration =
266            SqlMigration::new("0002", "oops", "CREATE TABLE t (id SERIAL PRIMARY KEY);");
267        let migrations = Migrations::sqlite(&[PORTABLE, NOT_PORTABLE]);
268        let err = select_set(&migrations).expect_err("SERIAL is not portable");
269        assert!(err.contains("0002/oops"), "names the file: {err}");
270        assert!(err.contains("SERIAL"), "names the token: {err}");
271    }
272}
cratefield_adapter_postgres/migrate.rs

cratefield_adapter_postgres/
migrate.rs
