1use 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#[derive(Debug, Clone)]
28pub enum UndoTarget {
29 Last,
31 Version(MigrationVersion),
33 Count(usize),
35}
36
37#[derive(Debug, Serialize)]
39pub struct UndoReport {
40 pub migrations_undone: usize,
42 pub total_time_ms: i32,
44 pub details: Vec<UndoDetail>,
46}
47
48#[derive(Debug, Serialize)]
50pub struct UndoDetail {
51 pub version: String,
53 pub description: String,
55 pub script: String,
57 pub execution_time_ms: i32,
59 pub auto_reversal: bool,
61}
62
63#[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 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#[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 db::acquire_advisory_lock(client, table).await?;
162
163 let result = run_undo(client, config, target).await;
164
165 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 history::create_history_table(client, schema, table).await?;
197
198 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 let applied = history::get_applied_migrations(client, schema, table).await?;
208 let effective = history::effective_applied_versions(&applied);
209
210 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(); 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 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 for version in &versions_to_undo {
249 if let Some(undo_migration) = undo_by_version.get(&version.raw) {
251 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 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
339pub 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 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 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 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}