Skip to main content

waypoint_core/commands/
undo.rs

1//! Undo applied migrations by executing U{version}__*.sql files,
2//! or auto-generated reversal SQL stored in the history table.
3
4use std::collections::HashMap;
5
6use serde::Serialize;
7
8#[cfg(feature = "postgres")]
9use tokio_postgres::Client;
10
11use crate::config::WaypointConfig;
12#[cfg(feature = "postgres")]
13use crate::db;
14use crate::db::DbClient;
15use crate::dialect::DialectKind;
16use crate::error::{Result, WaypointError};
17use crate::history;
18use crate::migration::{MigrationVersion, ResolvedMigration, scan_migrations};
19use crate::placeholder::{build_placeholders, replace_placeholders};
20
21/// How many / which versions to undo.
22///
23/// `Last` and `Count` walk the currently-applied versions in descending
24/// *version* order, which is not necessarily installation order — with
25/// `out_of_order` enabled a lower version can be installed later. Undo is
26/// defined on version order so that it is the inverse of migrate.
27#[derive(Debug, Clone)]
28pub enum UndoTarget {
29    /// Undo the highest currently-applied version.
30    Last,
31    /// Undo all migrations above this version (the target version itself stays applied).
32    Version(MigrationVersion),
33    /// Undo the N highest currently-applied versions, highest first.
34    Count(usize),
35}
36
37/// Report returned after an undo operation.
38#[derive(Debug, Serialize)]
39pub struct UndoReport {
40    /// Number of migrations that were undone.
41    pub migrations_undone: usize,
42    /// Total execution time of all undo operations in milliseconds.
43    pub total_time_ms: i32,
44    /// Per-migration details for each undone migration.
45    pub details: Vec<UndoDetail>,
46}
47
48/// Details of a single undone migration.
49#[derive(Debug, Serialize)]
50pub struct UndoDetail {
51    /// Version string of the migration that was undone.
52    pub version: String,
53    /// Human-readable description from the undo migration filename.
54    pub description: String,
55    /// Filename of the undo migration script that was executed.
56    pub script: String,
57    /// Execution time of the undo operation in milliseconds.
58    pub execution_time_ms: i32,
59    /// Whether the undo used auto-generated reversal SQL.
60    pub auto_reversal: bool,
61}
62
63/// Execute undo SQL within an atomic transaction (BEGIN/execute/history-insert/COMMIT).
64///
65/// On SQL execution failure, the transaction is rolled back and a best-effort
66/// failure record is inserted into the history table. Returns the execution
67/// time in milliseconds on success.
68#[cfg(feature = "postgres")]
69#[allow(clippy::too_many_arguments)]
70async fn execute_undo_sql(
71    client: &Client,
72    schema: &str,
73    table: &str,
74    version: &str,
75    description: &str,
76    script: &str,
77    checksum: Option<i32>,
78    installed_by: &str,
79    sql: &str,
80) -> Result<i32> {
81    let start = std::time::Instant::now();
82    client.batch_execute("BEGIN").await?;
83
84    match client.batch_execute(sql).await {
85        Ok(()) => {
86            let exec_time = start.elapsed().as_millis() as i32;
87            match history::insert_applied_migration(
88                client,
89                schema,
90                table,
91                Some(version),
92                description,
93                "UNDO_SQL",
94                script,
95                checksum,
96                installed_by,
97                exec_time,
98                true,
99            )
100            .await
101            {
102                Ok(()) => {
103                    client.batch_execute("COMMIT").await?;
104                    Ok(exec_time)
105                }
106                Err(e) => {
107                    if let Err(rb) = client.batch_execute("ROLLBACK").await {
108                        log::error!("Failed to rollback undo transaction: {}", rb);
109                    }
110                    Err(e)
111                }
112            }
113        }
114        Err(e) => {
115            if let Err(rollback_err) = client.batch_execute("ROLLBACK").await {
116                log::error!("Failed to rollback undo transaction: {}", rollback_err);
117            }
118
119            // Record failure — best-effort outside the rolled-back transaction
120            if let Err(record_err) = history::insert_applied_migration(
121                client,
122                schema,
123                table,
124                Some(version),
125                description,
126                "UNDO_SQL",
127                script,
128                checksum,
129                installed_by,
130                0,
131                false,
132            )
133            .await
134            {
135                log::warn!(
136                    "Failed to record undo failure; script={}, error={}",
137                    script,
138                    record_err
139                );
140            }
141
142            let reason = crate::error::format_db_error(&e);
143            Err(WaypointError::UndoFailed {
144                script: script.to_string(),
145                reason,
146            })
147        }
148    }
149}
150
151/// Execute the undo command (PostgreSQL legacy entry).
152#[cfg(feature = "postgres")]
153pub async fn execute(
154    client: &Client,
155    config: &WaypointConfig,
156    target: UndoTarget,
157) -> Result<UndoReport> {
158    let table = &config.migrations.table;
159
160    // Acquire advisory lock
161    db::acquire_advisory_lock(client, table).await?;
162
163    let result = run_undo(client, config, target).await;
164
165    // Always release the advisory lock
166    if let Err(e) = db::release_advisory_lock(client, table).await {
167        log::error!("Failed to release advisory lock: {}", e);
168    }
169
170    match &result {
171        Ok(report) => {
172            log::info!(
173                "Undo completed; migrations_undone={}, total_time_ms={}",
174                report.migrations_undone,
175                report.total_time_ms
176            );
177        }
178        Err(e) => {
179            log::error!("Undo failed: {}", e);
180        }
181    }
182
183    result
184}
185
186#[cfg(feature = "postgres")]
187async fn run_undo(
188    client: &Client,
189    config: &WaypointConfig,
190    target: UndoTarget,
191) -> Result<UndoReport> {
192    let schema = &config.migrations.schema;
193    let table = &config.migrations.table;
194
195    // Create history table if not exists
196    history::create_history_table(client, schema, table).await?;
197
198    // Scan migration files — build map of undo files by version
199    let resolved = scan_migrations(&config.migrations.locations)?;
200    let undo_by_version: HashMap<String, &ResolvedMigration> = resolved
201        .iter()
202        .filter(|m| m.is_undo())
203        .filter_map(|m| m.version().map(|v| (v.raw.clone(), m)))
204        .collect();
205
206    // Get applied history and compute effective set
207    let applied = history::get_applied_migrations(client, schema, table).await?;
208    let effective = history::effective_applied_versions(&applied);
209
210    // Build list of currently-applied versioned migrations, sorted descending by version
211    let mut applied_versions: Vec<MigrationVersion> = effective
212        .iter()
213        .filter_map(|v| MigrationVersion::parse(v).ok())
214        .collect();
215    applied_versions.sort();
216    applied_versions.reverse(); // newest first
217
218    // Determine which versions to undo
219    let versions_to_undo: Vec<MigrationVersion> = match target {
220        UndoTarget::Last => applied_versions.into_iter().take(1).collect(),
221        UndoTarget::Count(n) => applied_versions.into_iter().take(n).collect(),
222        UndoTarget::Version(ref target_ver) => applied_versions
223            .into_iter()
224            .filter(|v| v > target_ver)
225            .collect(),
226    };
227
228    // Get database user info for placeholders
229    let db_user = db::get_current_user(client)
230        .await
231        .unwrap_or_else(|_| "unknown".to_string());
232    let db_name = db::get_current_database(client)
233        .await
234        .unwrap_or_else(|_| "unknown".to_string());
235    let installed_by = config
236        .migrations
237        .installed_by
238        .as_deref()
239        .unwrap_or(&db_user);
240
241    let mut report = UndoReport {
242        migrations_undone: 0,
243        total_time_ms: 0,
244        details: Vec::new(),
245    };
246
247    // Execute undo for each version (newest first)
248    for version in &versions_to_undo {
249        // Try manual U file first, then fall back to auto-generated reversal
250        if let Some(undo_migration) = undo_by_version.get(&version.raw) {
251            // Manual undo file takes precedence
252            log::info!(
253                "Undoing migration (manual); migration={}, schema={}",
254                undo_migration.script,
255                schema
256            );
257
258            let placeholders = build_placeholders(
259                &config.placeholders,
260                schema,
261                &db_user,
262                &db_name,
263                &undo_migration.script,
264            );
265            let sql = replace_placeholders(&undo_migration.sql, &placeholders)?;
266
267            let exec_time = execute_undo_sql(
268                client,
269                schema,
270                table,
271                &version.raw,
272                &undo_migration.description,
273                &undo_migration.script,
274                Some(undo_migration.checksum),
275                installed_by,
276                &sql,
277            )
278            .await?;
279
280            report.migrations_undone += 1;
281            report.total_time_ms += exec_time;
282            report.details.push(UndoDetail {
283                version: version.raw.clone(),
284                description: undo_migration.description.clone(),
285                script: undo_migration.script.clone(),
286                execution_time_ms: exec_time,
287                auto_reversal: false,
288            });
289        } else if config.reversals.enabled {
290            // Fall back to auto-generated reversal SQL from history table
291            match crate::reversal::get_reversal(client, schema, table, &version.raw).await? {
292                Some(reversal_sql) => {
293                    let script = format!("auto-reversal:V{}", version.raw);
294                    log::info!(
295                        "Undoing migration (auto-reversal); version={}, schema={}",
296                        version.raw,
297                        schema
298                    );
299
300                    let exec_time = execute_undo_sql(
301                        client,
302                        schema,
303                        table,
304                        &version.raw,
305                        "Auto-generated reversal",
306                        &script,
307                        None,
308                        installed_by,
309                        &reversal_sql,
310                    )
311                    .await?;
312
313                    report.migrations_undone += 1;
314                    report.total_time_ms += exec_time;
315                    report.details.push(UndoDetail {
316                        version: version.raw.clone(),
317                        description: "Auto-generated reversal".to_string(),
318                        script,
319                        execution_time_ms: exec_time,
320                        auto_reversal: true,
321                    });
322                }
323                None => {
324                    return Err(WaypointError::UndoMissing {
325                        version: version.raw.clone(),
326                    });
327                }
328            }
329        } else {
330            return Err(WaypointError::UndoMissing {
331                version: version.raw.clone(),
332            });
333        }
334    }
335
336    Ok(report)
337}
338
339// ── Dialect-aware entry + MySQL path ─────────────────────────────────────────
340//
341// Both engines resolve an undo the same way: a manual `U{version}__*.sql` file
342// takes precedence, falling back to the auto-generated reversal SQL stored in
343// the history table when `[reversals] enabled` is set.
344
345/// Execute the undo command (dialect-aware entry).
346pub async fn execute_db(
347    client: &DbClient,
348    config: &WaypointConfig,
349    target: UndoTarget,
350) -> Result<UndoReport> {
351    match client.dialect_kind() {
352        #[cfg(feature = "postgres")]
353        DialectKind::Postgres => execute(client.as_postgres()?, config, target).await,
354        #[cfg(not(feature = "postgres"))]
355        DialectKind::Postgres => Err(WaypointError::ConfigError(
356            "PostgreSQL support is not compiled in (enable the `postgres` feature)".into(),
357        )),
358        #[cfg(feature = "mysql")]
359        DialectKind::Mysql => execute_mysql(client, config, target).await,
360        #[cfg(not(feature = "mysql"))]
361        DialectKind::Mysql => Err(WaypointError::ConfigError(
362            "MySQL support is not compiled in (enable the `mysql` feature)".into(),
363        )),
364    }
365}
366
367#[cfg(feature = "mysql")]
368async fn execute_mysql(
369    client: &DbClient,
370    config: &WaypointConfig,
371    target: UndoTarget,
372) -> Result<UndoReport> {
373    let table = &config.migrations.table;
374
375    client.acquire_lock(table).await?;
376
377    let result = run_undo_mysql(client, config, target).await;
378
379    if let Err(e) = client.release_lock(table).await {
380        log::error!("Failed to release advisory lock: {}", e);
381    }
382
383    match &result {
384        Ok(report) => {
385            log::info!(
386                "Undo completed (mysql); migrations_undone={}, total_time_ms={}",
387                report.migrations_undone,
388                report.total_time_ms
389            );
390        }
391        Err(e) => {
392            log::error!("Undo failed (mysql): {}", e);
393        }
394    }
395
396    result
397}
398
399#[cfg(feature = "mysql")]
400async fn run_undo_mysql(
401    client: &DbClient,
402    config: &WaypointConfig,
403    target: UndoTarget,
404) -> Result<UndoReport> {
405    let schema = client.resolve_schema(&config.migrations.schema).await?;
406    let schema = schema.as_str();
407    let table = &config.migrations.table;
408
409    history::create_history_table_db(client, schema, table).await?;
410
411    let resolved = scan_migrations(&config.migrations.locations)?;
412    let undo_by_version: HashMap<String, &ResolvedMigration> = resolved
413        .iter()
414        .filter(|m| m.is_undo())
415        .filter_map(|m| m.version().map(|v| (v.raw.clone(), m)))
416        .collect();
417
418    let applied = history::get_applied_migrations_db(client, schema, table).await?;
419    let effective = history::effective_applied_versions(&applied);
420
421    let mut applied_versions: Vec<MigrationVersion> = effective
422        .iter()
423        .filter_map(|v| MigrationVersion::parse(v).ok())
424        .collect();
425    applied_versions.sort();
426    applied_versions.reverse();
427
428    let versions_to_undo: Vec<MigrationVersion> = match target {
429        UndoTarget::Last => applied_versions.into_iter().take(1).collect(),
430        UndoTarget::Count(n) => applied_versions.into_iter().take(n).collect(),
431        UndoTarget::Version(ref target_ver) => applied_versions
432            .into_iter()
433            .filter(|v| v > target_ver)
434            .collect(),
435    };
436
437    let db_user = client
438        .current_user()
439        .await
440        .unwrap_or_else(|_| "unknown".into());
441    let db_name = client
442        .current_database()
443        .await
444        .unwrap_or_else(|_| "unknown".into());
445    let installed_by = config
446        .migrations
447        .installed_by
448        .as_deref()
449        .unwrap_or(&db_user)
450        .to_string();
451
452    let mut report = UndoReport {
453        migrations_undone: 0,
454        total_time_ms: 0,
455        details: Vec::new(),
456    };
457
458    for version in &versions_to_undo {
459        let (sql, script, description, checksum, auto_reversal) = match undo_by_version
460            .get(&version.raw)
461        {
462            Some(m) => {
463                // Manual U file: highest precedence.
464                let placeholders =
465                    build_placeholders(&config.placeholders, schema, &db_user, &db_name, &m.script);
466                let sql = replace_placeholders(&m.sql, &placeholders)?;
467                log::info!(
468                    "Undoing migration (manual); migration={}, schema={}",
469                    m.script,
470                    schema
471                );
472                (
473                    sql,
474                    m.script.clone(),
475                    m.description.clone(),
476                    Some(m.checksum),
477                    false,
478                )
479            }
480            None if config.reversals.enabled => {
481                // Fall back to auto-generated reversal SQL stored in history.
482                match crate::reversal::get_reversal_db(client, schema, table, &version.raw).await? {
483                    Some(reversal_sql) => {
484                        let script = format!("auto-reversal:V{}", version.raw);
485                        log::info!(
486                            "Undoing migration (auto-reversal); version={}, schema={}",
487                            version.raw,
488                            schema
489                        );
490                        (
491                            reversal_sql,
492                            script,
493                            "Auto-generated reversal".to_string(),
494                            None,
495                            true,
496                        )
497                    }
498                    None => {
499                        return Err(WaypointError::UndoMissing {
500                            version: version.raw.clone(),
501                        });
502                    }
503                }
504            }
505            None => {
506                return Err(WaypointError::UndoMissing {
507                    version: version.raw.clone(),
508                });
509            }
510        };
511
512        let start = std::time::Instant::now();
513        let exec_result = client.execute_raw(&sql).await;
514        let exec_time = start.elapsed().as_millis() as i32;
515
516        match exec_result {
517            Ok(_) => {
518                history::insert_applied_migration_db(
519                    client,
520                    schema,
521                    table,
522                    Some(&version.raw),
523                    &description,
524                    "UNDO_SQL",
525                    &script,
526                    checksum,
527                    &installed_by,
528                    exec_time,
529                    true,
530                )
531                .await?;
532
533                report.migrations_undone += 1;
534                report.total_time_ms += exec_time;
535                report.details.push(UndoDetail {
536                    version: version.raw.clone(),
537                    description: description.clone(),
538                    script: script.clone(),
539                    execution_time_ms: exec_time,
540                    auto_reversal,
541                });
542            }
543            Err(e) => {
544                // MySQL DDL auto-commits so the schema may be in a partially-
545                // undone state; record the failure and surface a clear error.
546                if let Err(record_err) = history::insert_applied_migration_db(
547                    client,
548                    schema,
549                    table,
550                    Some(&version.raw),
551                    &description,
552                    "UNDO_SQL",
553                    &script,
554                    checksum,
555                    &installed_by,
556                    exec_time,
557                    false,
558                )
559                .await
560                {
561                    log::warn!(
562                        "Failed to record undo failure; script={}, error={}",
563                        script,
564                        record_err
565                    );
566                }
567                return Err(WaypointError::UndoFailed {
568                    script: script.clone(),
569                    reason: e.to_string(),
570                });
571            }
572        }
573    }
574
575    Ok(report)
576}