siderite_migrations/
executor.rs

1//! Apply and reverse migrations against a [`siderite_orm::Db`].
2
3use crate::error::MigrationError;
4use crate::loader::{MigrationGraph, is_applied};
5use crate::migration::Migration;
6use crate::operation::Operation;
7use crate::registry::MigrationRegistry;
8use crate::schema_editor;
9use crate::state::ProjectState;
10use chrono::Utc;
11use siderite_orm::types::TIMESTAMP_FORMAT;
12use siderite_orm::{BackendKind, Db, TransactionSupport, Value};
13use std::collections::{HashMap, HashSet};
14use std::future::Future;
15
16mod sqlite;
17use sqlite::{
18    load_sqlite_rebuild_extras, reject_lossy_sqlite_cast, reject_not_null_violations,
19    restore_sqlite_rebuild_extras,
20};
21mod operations;
22use operations::{apply_ops_in_order, unapply_ops_in_order};
23mod progress;
24use progress::{
25    PROGRESS_APPLY, PROGRESS_UNAPPLY, Progress, clear_progress, load_progress,
26    reset_progress_direction, save_progress,
27};
28mod catalog;
29mod intents;
30pub use intents::MigrationIntent;
31mod recovery;
32pub use recovery::{RecoveryEntry, RecoveryReport};
33mod holder;
34pub use holder::LockHolder;
35mod locks;
36use locks::release_migrate_lock;
37
38#[cfg(test)]
39mod intent_tests;
40#[cfg(test)]
41mod tests;
42
43/// History table written by the migrator.
44pub const HISTORY_TABLE: &str = "siderite_migrations";
45
46/// Outcome of [`Migrator::migrate`] or [`Migrator::rollback`].
47#[derive(Debug, Clone, PartialEq, Eq, Default)]
48pub struct Report {
49    /// Migration ids that would be (or were) applied / unapplied, in order.
50    pub planned: Vec<String>,
51    /// Generated SQL, including the history-table DDL when it would be created.
52    pub sql: Vec<String>,
53    /// Ids actually written to (or deleted from) the history table.
54    pub applied: Vec<String>,
55    /// When true, no statement was executed.
56    pub dry_run: bool,
57}
58
59/// Applies JSON migrations through [`Db`].
60pub struct Migrator<'a> {
61    db: &'a Db,
62    graph: &'a MigrationGraph,
63    registry: MigrationRegistry,
64    lock_timeout: std::time::Duration,
65}
66
67impl<'a> Migrator<'a> {
68    /// Plan and execute using `graph`.
69    pub fn new(db: &'a Db, graph: &'a MigrationGraph) -> Self {
70        Self {
71            db,
72            graph,
73            registry: MigrationRegistry::new(),
74            lock_timeout: std::time::Duration::from_secs(30),
75        }
76    }
77
78    /// Register `RunRust` functions used by the planned migrations.
79    #[must_use]
80    pub fn with_registry(mut self, registry: MigrationRegistry) -> Self {
81        self.registry = registry;
82        self
83    }
84
85    /// Set the advisory lock acquisition deadline (default: 30 seconds).
86    ///
87    /// Args:
88    ///     timeout: Positive, representable PostgreSQL/MySQL lock wait.
89    ///
90    /// Returns:
91    ///     This migrator with the new acquisition policy.
92    ///
93    /// Pool acquisition and SQLite BEGIN use backend deadlines. This limit
94    /// does not bound migration execution after the lock is held.
95    ///
96    /// # Errors
97    /// Zero or a duration that overflows the runtime clock.
98    pub fn with_lock_timeout(
99        mut self,
100        timeout: std::time::Duration,
101    ) -> Result<Self, MigrationError> {
102        if timeout.is_zero() || tokio::time::Instant::now().checked_add(timeout).is_none() {
103            return Err(MigrationError::usage("invalid migration lock timeout"));
104        }
105        self.lock_timeout = timeout;
106        Ok(self)
107    }
108
109    /// Ids that would be applied to reach `target` (`None` = all remaining).
110    ///
111    /// # Errors
112    /// Checksum mismatch, unknown target, graph/state errors, or backend errors
113    /// while reading the history table.
114    pub async fn plan_forward(&self, target: Option<&str>) -> Result<Vec<String>, MigrationError> {
115        let history = self.load_history().await?;
116        self.verify_checksums(&history)?;
117        self.forward_plan(&history, target)
118    }
119
120    /// Apply unapplied migrations up to `target` (`None` = all).
121    ///
122    /// `dry_run` returns the SQL without executing it (history is still read).
123    ///
124    /// # Errors
125    /// See [`Migrator::plan_forward`]; also statement execution errors.
126    pub async fn migrate(
127        &self,
128        target: Option<&str>,
129        dry_run: bool,
130    ) -> Result<Report, MigrationError> {
131        self.require_backend()?;
132        if dry_run {
133            let history = self.load_history().await?;
134            self.verify_checksums(&history)?;
135            let plan = self.forward_plan(&history, target)?;
136            let (sql, _) = self.plan_sql(&history, &plan)?;
137            return Ok(Report {
138                planned: plan,
139                sql,
140                applied: Vec::new(),
141                dry_run: true,
142            });
143        }
144        if self.db.is_scoped() {
145            return Err(MigrationError::state(
146                "migration execution requires a pool handle; \
147                 caller-owned transactions and connections are unsupported",
148            ));
149        }
150        self.with_lock(|session| {
151            let this = self;
152            async move {
153                let created_history = if !this.history_table_exists_on(&session).await? {
154                    session
155                        .execute_script(&create_history_sql(this.kind()))
156                        .await?;
157                    true
158                } else {
159                    false
160                };
161                let history = this.load_history_from(&session).await?;
162                this.verify_checksums(&history)?;
163                let plan = this.forward_plan(&history, target)?;
164                // Render under the lock, from the plan re-read here:
165                // `Report.sql` must not disagree with `Report.planned`
166                // about what actually ran.
167                let (mut sql, mut state) = this.plan_sql(&history, &plan)?;
168                if created_history {
169                    sql.insert(0, create_history_sql(this.kind()));
170                }
171                let mut applied = Vec::new();
172                for id in &plan {
173                    let migration = this.require_migration(id)?.clone();
174                    this.apply_one(&session, &migration, &mut state).await?;
175                    applied.push(id.clone());
176                }
177                Ok(Report {
178                    planned: plan,
179                    sql,
180                    applied,
181                    dry_run: false,
182                })
183            }
184        })
185        .await
186    }
187
188    /// Unapply migrations. `target` is the id that should remain the head
189    /// (everything after it is rolled back). `steps` unapplies that many of
190    /// the most recently applied visible migrations. Exactly one of the two
191    /// must be set unless `steps` is `Some(n)`.
192    ///
193    /// # Errors
194    /// Irreversible migrations, checksum mismatch, unknown target, backend
195    /// errors.
196    pub async fn rollback(
197        &self,
198        target: Option<&str>,
199        steps: Option<u32>,
200        dry_run: bool,
201    ) -> Result<Report, MigrationError> {
202        self.require_backend()?;
203        if dry_run {
204            let history = self.load_history().await?;
205            self.verify_checksums(&history)?;
206            let plan = self.rollback_plan(&history, target, steps)?;
207            let sql = self.rollback_sql(&history, &plan)?;
208            return Ok(Report {
209                planned: plan,
210                sql,
211                applied: Vec::new(),
212                dry_run: true,
213            });
214        }
215        if self.db.is_scoped() {
216            return Err(MigrationError::state(
217                "migration execution requires a pool handle; \
218                 caller-owned transactions and connections are unsupported",
219            ));
220        }
221        self.with_lock(|session| {
222            let this = self;
223            async move {
224                let created_history = if !this.history_table_exists_on(&session).await? {
225                    session
226                        .execute_script(&create_history_sql(this.kind()))
227                        .await?;
228                    true
229                } else {
230                    false
231                };
232                let history = this.load_history_from(&session).await?;
233                this.verify_checksums(&history)?;
234                let plan = this.rollback_plan(&history, target, steps)?;
235                let mut sql = this.rollback_sql(&history, &plan)?;
236                if created_history {
237                    sql.insert(0, create_history_sql(this.kind()));
238                }
239                let mut unapplied = Vec::new();
240                for id in &plan {
241                    let migration = this.require_migration(id)?.clone();
242                    let state_before = state_before_migration(this.graph, &history, id)?;
243                    this.unapply_one(&session, &migration, &state_before)
244                        .await?;
245                    unapplied.push(id.clone());
246                }
247                Ok(Report {
248                    planned: plan,
249                    sql,
250                    applied: unapplied,
251                    dry_run: false,
252                })
253            }
254        })
255        .await
256    }
257
258    /// Every loaded migration with its applied flag (including replaced ids).
259    ///
260    /// # Errors
261    /// History-table or checksum errors.
262    pub async fn show(&self) -> Result<Vec<(String, bool)>, MigrationError> {
263        let history = self.load_history().await?;
264        self.verify_checksums(&history)?;
265        let mut rows = Vec::new();
266        let mut seen = HashSet::new();
267        for id in &self.graph.order {
268            if let Some(m) = self.graph.get(id) {
269                rows.push((id.clone(), is_applied(m, &history.ids)));
270                seen.insert(id.clone());
271            }
272        }
273        let mut rest: Vec<&Migration> = self
274            .graph
275            .all
276            .iter()
277            .filter(|m| !seen.contains(&m.id))
278            .collect();
279        rest.sort_by(|a, b| a.id.cmp(&b.id));
280        for m in rest {
281            rows.push((m.id.clone(), is_applied(m, &history.ids)));
282        }
283        Ok(rows)
284    }
285
286    fn kind(&self) -> BackendKind {
287        self.db.capabilities().kind
288    }
289
290    async fn with_lock<F, Fut>(&self, f: F) -> Result<Report, MigrationError>
291    where
292        F: FnOnce(Db) -> Fut,
293        Fut: Future<Output = Result<Report, MigrationError>>,
294    {
295        // SQLite holds a single `BEGIN IMMEDIATE` transaction for the whole
296        // run (the transaction *is* the lock): `migrate`/`rollback` there
297        // are all-or-nothing, and the per-migration `atomic` flag has no
298        // effect. PostgreSQL and MySQL hold a session lock on a
299        // non-transactional connection instead, and each migration commits
300        // on its own.
301        let transactional = self.kind() == BackendKind::Sqlite;
302        self.db
303            .schema_change(transactional, |session| async move {
304                locks::acquire(&session, self.lock_timeout).await?;
305                let result = f(session.clone()).await;
306                let unlocked = release_migrate_lock(&session).await;
307                match (result, unlocked) {
308                    (Ok(v), Ok(())) => Ok(v),
309                    (Err(e), _) => Err(e),
310                    // Work finished but ownership is unconfirmed; say so
311                    // instead of surfacing a bare driver error.
312                    (Ok(report), Err(e)) => Err(uncertain(report, e)),
313                }
314            })
315            .await
316    }
317
318    fn require_backend(&self) -> Result<(), MigrationError> {
319        match self.kind() {
320            BackendKind::Postgres | BackendKind::Sqlite | BackendKind::MySql => Ok(()),
321            other => Err(MigrationError::UnsupportedBackend(other)),
322        }
323    }
324
325    fn require_migration(&self, id: &str) -> Result<&Migration, MigrationError> {
326        self.graph
327            .get(id)
328            .ok_or_else(|| MigrationError::usage(format!("unknown migration `{id}`")))
329    }
330
331    /// Render `plan` to SQL, replaying project state from `history`.
332    ///
333    /// Returns the statements and the pre-plan state (callers feed it to
334    /// `apply_one`, which advances it migration by migration so rendering
335    /// and execution agree).
336    fn plan_sql(
337        &self,
338        history: &History,
339        plan: &[String],
340    ) -> Result<(Vec<String>, ProjectState), MigrationError> {
341        let state = replay(self.graph, history)?;
342        let mut render = state.clone();
343        let mut sql = Vec::new();
344        if !history.table_exists {
345            sql.push(create_history_sql(self.kind()));
346        }
347        for id in plan {
348            let migration = self.require_migration(id)?;
349            sql.extend(schema_editor::statements(
350                self.kind(),
351                &render,
352                &migration.operations,
353            )?);
354            for op in &migration.operations {
355                op.apply_to_state(&mut render)?;
356            }
357        }
358        Ok((sql, state))
359    }
360
361    /// Render reverse SQL for `plan` from `history`.
362    fn rollback_sql(
363        &self,
364        history: &History,
365        plan: &[String],
366    ) -> Result<Vec<String>, MigrationError> {
367        let mut sql = Vec::new();
368        for id in plan {
369            let migration = self.require_migration(id)?;
370            let state_before = state_before_migration(self.graph, history, id)?;
371            sql.extend(reverse_sql(self.kind(), migration, &state_before)?);
372        }
373        Ok(sql)
374    }
375
376    fn forward_plan(
377        &self,
378        history: &History,
379        target: Option<&str>,
380    ) -> Result<Vec<String>, MigrationError> {
381        if let Some(id) = target
382            && self.graph.get(id).is_none()
383        {
384            return Err(MigrationError::usage(format!("unknown migration `{id}`")));
385        }
386        let mut plan = Vec::new();
387        for id in &self.graph.order {
388            let migration = self.require_migration(id)?;
389            if is_applied(migration, &history.ids) {
390                if target == Some(id.as_str()) {
391                    return Ok(plan);
392                }
393                continue;
394            }
395            plan.push(id.clone());
396            if target == Some(id.as_str()) {
397                break;
398            }
399        }
400        if let Some(id) = target
401            && !plan.iter().any(|p| p == id)
402            && !history.ids.contains(id)
403            && !self
404                .graph
405                .get(id)
406                .is_some_and(|m| is_applied(m, &history.ids))
407        {
408            // Target is a replaced id: applying the squash (already in plan or applied)
409            // is enough; if the squash is not in the visible graph, error.
410            if self.graph.replaced_by.contains_key(id) {
411                return Ok(plan);
412            }
413            return Err(MigrationError::usage(format!(
414                "migration `{id}` is not reachable from the current graph"
415            )));
416        }
417        Ok(plan)
418    }
419
420    fn rollback_plan(
421        &self,
422        history: &History,
423        target: Option<&str>,
424        steps: Option<u32>,
425    ) -> Result<Vec<String>, MigrationError> {
426        let applied_visible: Vec<String> = self
427            .graph
428            .order
429            .iter()
430            .filter(|id| {
431                self.graph
432                    .get(id)
433                    .is_some_and(|m| is_applied(m, &history.ids))
434            })
435            .cloned()
436            .collect();
437        if let Some(n) = steps {
438            let n = n as usize;
439            let start = applied_visible.len().saturating_sub(n);
440            let mut plan = applied_visible[start..].to_vec();
441            plan.reverse();
442            return Ok(plan);
443        }
444        let Some(target) = target else {
445            return Err(MigrationError::usage(
446                "rollback requires --steps N or a target id",
447            ));
448        };
449        if !applied_visible.iter().any(|id| id == target) && !history.ids.contains(target) {
450            return Err(MigrationError::usage(format!(
451                "migration `{target}` is not applied"
452            )));
453        }
454        // Unapply everything after `target` (target stays applied).
455        let mut plan = Vec::new();
456        let mut after = false;
457        for id in &applied_visible {
458            if after {
459                plan.push(id.clone());
460            }
461            if id == target {
462                after = true;
463            }
464        }
465        plan.reverse();
466        Ok(plan)
467    }
468
469    fn verify_checksums(&self, history: &History) -> Result<(), MigrationError> {
470        for migration in &self.graph.all {
471            if let Some(stored) = history.checksums.get(&migration.id)
472                && stored != &migration.checksum
473            {
474                return Err(MigrationError::ChecksumMismatch {
475                    id: migration.id.clone(),
476                    history: stored.clone(),
477                    file: migration.checksum.clone(),
478                });
479            }
480        }
481        Ok(())
482    }
483
484    async fn load_history(&self) -> Result<History, MigrationError> {
485        self.load_history_from(self.db).await
486    }
487
488    async fn load_history_from(&self, db: &Db) -> Result<History, MigrationError> {
489        if !self.history_table_exists_on(db).await? {
490            return Ok(History {
491                ids: HashSet::new(),
492                checksums: HashMap::new(),
493                table_exists: false,
494            });
495        }
496        // Any failure past this point (permissions, a dropped connection, a
497        // damaged table) must surface: treating it as an empty history would
498        // replay every migration against a live database.
499        let result = db
500            .raw_sql(
501                &format!(
502                    "SELECT {} as id, {} as checksum FROM {}",
503                    quote_star(self.kind(), "id"),
504                    quote_star(self.kind(), "checksum"),
505                    quote_star(self.kind(), HISTORY_TABLE)
506                ),
507                Vec::new(),
508            )
509            .await?;
510        let mut ids = HashSet::new();
511        let mut checksums = HashMap::new();
512        for row in result.rows {
513            let Some(Value::Text(id)) = row.get("id") else {
514                return Err(MigrationError::state(format!(
515                    "`{HISTORY_TABLE}` has a row without a text `id`"
516                )));
517            };
518            let checksum = match row.get("checksum") {
519                Some(Value::Text(s)) => s.clone(),
520                _ => String::new(),
521            };
522            ids.insert(id.clone());
523            checksums.insert(id.clone(), checksum);
524        }
525        Ok(History {
526            ids,
527            checksums,
528            table_exists: true,
529        })
530    }
531
532    /// Ask the catalog whether the history table exists, so a failing read
533    /// of the table itself is never mistaken for a fresh database.
534    async fn history_table_exists_on(&self, db: &Db) -> Result<bool, MigrationError> {
535        self.table_exists_on(db, HISTORY_TABLE).await
536    }
537
538    async fn table_exists_on(&self, db: &Db, table: &str) -> Result<bool, MigrationError> {
539        catalog::table_exists(db, table).await
540    }
541
542    async fn apply_one(
543        &self,
544        db: &Db,
545        migration: &Migration,
546        state: &mut ProjectState,
547    ) -> Result<(), MigrationError> {
548        let kind = self.kind();
549        let migration = migration.clone();
550        let start_state = state.clone();
551        let run = |db: Db| {
552            let migration = migration.clone();
553            let registry = self.registry.clone();
554            let start_state = start_state.clone();
555            async move {
556                apply_ops_in_order(&db, kind, &migration, &start_state, &registry).await?;
557                Ok::<_, MigrationError>(())
558            }
559        };
560        // Migrations always run under `with_lock`. On SQLite that holds one
561        // transaction for the whole run (the lock), so this branch always
562        // hits; on other backends each migration gets its own transaction
563        // when `runs_in_transaction`.
564        if db.in_transaction() {
565            run(db.clone()).await?;
566        } else if runs_in_transaction(db, &migration) {
567            db.transaction(run).await?;
568        } else {
569            run(db.clone()).await?;
570        }
571        for op in &migration.operations {
572            op.apply_to_state(state)?;
573        }
574        Ok(())
575    }
576
577    async fn unapply_one(
578        &self,
579        db: &Db,
580        migration: &Migration,
581        state_before: &ProjectState,
582    ) -> Result<(), MigrationError> {
583        let kind = self.kind();
584        // See `apply_one`: the whole migration travels into the run closure
585        // because the reverse direction needs its id, checksum and `atomic`
586        // flag for the progress row alongside its operations.
587        let migration = migration.clone();
588        let start_state = state_before.clone();
589        let run = |db: Db| {
590            let migration = migration.clone();
591            let registry = self.registry.clone();
592            let start_state = start_state.clone();
593            async move {
594                unapply_ops_in_order(&db, kind, &migration, &start_state, &registry).await?;
595                Ok::<_, MigrationError>(())
596            }
597        };
598        // See `apply_one`: SQLite always arrives inside the run-wide
599        // transaction held by `with_lock`.
600        if db.in_transaction() {
601            run(db.clone()).await?;
602        } else if runs_in_transaction(db, &migration) {
603            db.transaction(run).await?;
604        } else {
605            run(db.clone()).await?;
606        }
607        Ok(())
608    }
609}
610
611struct History {
612    ids: HashSet<String>,
613    checksums: HashMap<String, String>,
614    table_exists: bool,
615}
616
617fn quote_star(kind: BackendKind, ident: &str) -> String {
618    schema_editor::quote_identifier(kind, ident)
619}
620
621/// Whether one migration can be wrapped in a transaction that also undoes
622/// its DDL. MySQL commits implicitly on every DDL statement, so wrapping it
623/// would only suggest an atomicity the database cannot give.
624fn runs_in_transaction(db: &Db, migration: &Migration) -> bool {
625    let caps = db.capabilities();
626    migration.atomic
627        && caps.transactions > TransactionSupport::None
628        && caps.kind != BackendKind::MySql
629}
630
631/// What a failed migration leaves behind on the server.
632///
633/// Decides both whether the executor records per-statement progress (only a
634/// run whose earlier statements survive a failure can be resumed) and what a
635/// failure reports: MySQL commits DDL implicitly but rolls DML back with the
636/// surrounding statement, while a PostgreSQL migration with `atomic: false`
637/// runs outside a transaction, where *every* statement autocommits.
638#[derive(Debug, Clone, Copy, PartialEq, Eq)]
639enum OnFailure {
640    /// A transaction wraps the run: a failure rolls every statement back.
641    Rollback,
642    /// MySQL: only implicit-commit DDL survives a failure.
643    ImplicitCommit,
644    /// PostgreSQL with `atomic: false`: no transaction, so every statement
645    /// autocommits (DML included).
646    Autocommit,
647}
648
649impl OnFailure {
650    /// Partial-failure behaviour of one migration on `kind`.
651    ///
652    /// MySQL is always partial, whatever `atomic` says. PostgreSQL is partial
653    /// only without the transaction, i.e. when the migration opted out with
654    /// `atomic: false` (what `CREATE INDEX CONCURRENTLY` needs). SQLite never
655    /// is: the whole `migrate` run is the lock's transaction.
656    fn of(kind: BackendKind, atomic: bool) -> Self {
657        match kind {
658            BackendKind::MySql => Self::ImplicitCommit,
659            BackendKind::Postgres if !atomic => Self::Autocommit,
660            _ => Self::Rollback,
661        }
662    }
663
664    /// Whether per-statement progress is recorded for this kind of run.
665    const fn resumes(self) -> bool {
666        !matches!(self, Self::Rollback)
667    }
668
669    /// Whether an earlier statement of this run is committed for good.
670    ///
671    /// MySQL counts only implicit-commit DDL, so a failure after mere DML
672    /// still reports the plain error; a non-transactional PostgreSQL run
673    /// counts anything, because outside a transaction every statement
674    /// autocommits.
675    fn prior_committed(self, committed: bool, ddl: bool) -> bool {
676        match self {
677            Self::Rollback => false,
678            Self::ImplicitCommit => ddl,
679            Self::Autocommit => committed,
680        }
681    }
682}
683
684/// [`MigrationError`] for a `RunRust` failure after earlier statements
685/// committed. The operation's own error is kept as the source.
686fn op_partial(
687    failure: OnFailure,
688    id: &str,
689    index: usize,
690    total: usize,
691    summary: String,
692    source: MigrationError,
693) -> MigrationError {
694    let id = id.to_owned();
695    match failure {
696        OnFailure::ImplicitCommit => MigrationError::MysqlOpPartial {
697            id,
698            index,
699            total,
700            summary,
701            source: Box::new(source),
702        },
703        OnFailure::Autocommit => MigrationError::PostgresOpPartial {
704            id,
705            index,
706            total,
707            summary,
708            source: Box::new(source),
709        },
710        OnFailure::Rollback => source,
711    }
712}
713
714fn create_history_sql(kind: BackendKind) -> String {
715    // MySQL cannot key a TEXT column without a prefix length.
716    let key_type = if kind == BackendKind::MySql {
717        "VARCHAR(255)"
718    } else {
719        "TEXT"
720    };
721    format!(
722        "CREATE TABLE IF NOT EXISTS {} (\n  {} {key_type} PRIMARY KEY,\n  {} TEXT NOT NULL,\n  {} TEXT NOT NULL\n)",
723        quote_star(kind, HISTORY_TABLE),
724        quote_star(kind, "id"),
725        quote_star(kind, "checksum"),
726        quote_star(kind, "applied_at"),
727    )
728}
729
730async fn execute_one_sql(
731    db: &Db,
732    failure: OnFailure,
733    migration_id: &str,
734    index: usize,
735    total: usize,
736    prior_committed: bool,
737    sql: &str,
738) -> Result<(), MigrationError> {
739    if let Err(err) = db.execute_script(sql).await {
740        // With earlier statements committed for good, a re-run replays them
741        // and gets stuck, so name the failed statement and what is left
742        // behind. Without them (a first-statement failure, or only rolled-back
743        // DML before it) the plain error is accurate.
744        if prior_committed {
745            return Err(match failure {
746                OnFailure::ImplicitCommit => MigrationError::MysqlPartial {
747                    id: migration_id.to_owned(),
748                    index,
749                    total,
750                    source: err,
751                },
752                OnFailure::Autocommit => MigrationError::PostgresPartial {
753                    id: migration_id.to_owned(),
754                    index,
755                    total,
756                    source: err,
757                },
758                OnFailure::Rollback => err.into(),
759            });
760        }
761        return Err(err.into());
762    }
763    Ok(())
764}
765
766/// Whether `sql` is (probably) DDL that MySQL commits implicitly, so a
767/// later failure leaves it applied. Matches the statement's first keyword,
768/// skipping whitespace and `--` / `/* */` comments.
769fn is_implicit_commit(sql: &str) -> bool {
770    matches!(
771        first_keyword(sql).as_str(),
772        "CREATE" | "ALTER" | "DROP" | "TRUNCATE" | "RENAME"
773    )
774}
775
776fn first_keyword(sql: &str) -> String {
777    let mut rest = sql.trim_start();
778    loop {
779        if let Some(body) = rest.strip_prefix("--") {
780            rest = body
781                .split_once('\n')
782                .map(|(_, tail)| tail)
783                .unwrap_or("")
784                .trim_start();
785        } else if let Some(body) = rest.strip_prefix("/*") {
786            rest = body
787                .split_once("*/")
788                .map(|(_, tail)| tail)
789                .unwrap_or("")
790                .trim_start();
791        } else {
792            break;
793        }
794    }
795    rest.split(|c: char| !c.is_ascii_alphabetic())
796        .next()
797        .unwrap_or("")
798        .to_ascii_uppercase()
799}
800
801fn placeholder(kind: BackendKind, index: usize) -> String {
802    match kind {
803        BackendKind::Postgres => format!("${index}"),
804        _ => "?".to_owned(),
805    }
806}
807
808async fn record_history(
809    db: &Db,
810    kind: BackendKind,
811    migration: &Migration,
812) -> Result<(), MigrationError> {
813    let sql = format!(
814        "INSERT INTO {} ({}, {}, {}) VALUES ({}, {}, {})",
815        quote_star(kind, HISTORY_TABLE),
816        quote_star(kind, "id"),
817        quote_star(kind, "checksum"),
818        quote_star(kind, "applied_at"),
819        placeholder(kind, 1),
820        placeholder(kind, 2),
821        placeholder(kind, 3),
822    );
823    let applied_at = Utc::now().format(TIMESTAMP_FORMAT).to_string();
824    db.raw_execute(
825        &sql,
826        vec![
827            Value::Text(migration.id.clone()),
828            Value::Text(migration.checksum.clone()),
829            Value::Text(applied_at),
830        ],
831    )
832    .await?;
833    Ok(())
834}
835
836async fn delete_history(db: &Db, kind: BackendKind, id: &str) -> Result<(), MigrationError> {
837    let sql = format!(
838        "DELETE FROM {} WHERE {} = {}",
839        quote_star(kind, HISTORY_TABLE),
840        quote_star(kind, "id"),
841        placeholder(kind, 1),
842    );
843    db.raw_execute(&sql, vec![Value::Text(id.to_owned())])
844        .await?;
845    Ok(())
846}
847
848fn replay(graph: &MigrationGraph, history: &History) -> Result<ProjectState, MigrationError> {
849    let mut state = ProjectState::new();
850    for id in &graph.order {
851        let Some(migration) = graph.get(id) else {
852            continue;
853        };
854        if !is_applied(migration, &history.ids) {
855            continue;
856        }
857        for op in &migration.operations {
858            op.apply_to_state(&mut state)?;
859        }
860    }
861    Ok(state)
862}
863
864fn state_before_migration(
865    graph: &MigrationGraph,
866    history: &History,
867    id: &str,
868) -> Result<ProjectState, MigrationError> {
869    let mut state = ProjectState::new();
870    for vis in &graph.order {
871        if vis == id {
872            break;
873        }
874        let Some(migration) = graph.get(vis) else {
875            continue;
876        };
877        if !is_applied(migration, &history.ids) {
878            continue;
879        }
880        for op in &migration.operations {
881            op.apply_to_state(&mut state)?;
882        }
883    }
884    Ok(state)
885}
886
887fn reverse_sql(
888    kind: BackendKind,
889    migration: &Migration,
890    state_before: &ProjectState,
891) -> Result<Vec<String>, MigrationError> {
892    let mut snapshots = vec![state_before.clone()];
893    let mut current = state_before.clone();
894    for op in &migration.operations {
895        op.apply_to_state(&mut current)?;
896        snapshots.push(current.clone());
897    }
898    let mut sql = Vec::new();
899    current = snapshots
900        .last()
901        .cloned()
902        .unwrap_or_else(|| state_before.clone());
903    for i in (0..migration.operations.len()).rev() {
904        let op = &migration.operations[i];
905        let Some(rev) = op.reverse(&snapshots[i]) else {
906            return Err(MigrationError::Irreversible {
907                id: migration.id.clone(),
908                reason: format!("{} has no reverse", op.summary()),
909            });
910        };
911        sql.extend(schema_editor::render(kind, &rev, &current)?);
912        rev.apply_to_state(&mut current)?;
913    }
914    Ok(sql)
915}
916
917fn uncertain(report: Report, source: MigrationError) -> MigrationError {
918    MigrationError::LockReleaseUncertain {
919        applied: report.applied,
920        source: Box::new(source),
921    }
922}
siderite_migrations/executor.rs

siderite_migrations/
executor.rs
