use std::collections::HashMap;
use serde::Serialize;
#[cfg(feature = "postgres")]
use tokio_postgres::Client;
use crate::config::WaypointConfig;
#[cfg(feature = "postgres")]
use crate::db;
use crate::db::DbClient;
use crate::dialect::DialectKind;
use crate::error::{Result, WaypointError};
use crate::history;
use crate::migration::{scan_migrations, MigrationVersion, ResolvedMigration};
use crate::placeholder::{build_placeholders, replace_placeholders};
#[derive(Debug, Clone)]
pub enum UndoTarget {
Last,
Version(MigrationVersion),
Count(usize),
}
#[derive(Debug, Serialize)]
pub struct UndoReport {
pub migrations_undone: usize,
pub total_time_ms: i32,
pub details: Vec<UndoDetail>,
}
#[derive(Debug, Serialize)]
pub struct UndoDetail {
pub version: String,
pub description: String,
pub script: String,
pub execution_time_ms: i32,
pub auto_reversal: bool,
}
#[cfg(feature = "postgres")]
#[allow(clippy::too_many_arguments)]
async fn execute_undo_sql(
client: &Client,
schema: &str,
table: &str,
version: &str,
description: &str,
script: &str,
checksum: Option<i32>,
installed_by: &str,
sql: &str,
) -> Result<i32> {
let start = std::time::Instant::now();
client.batch_execute("BEGIN").await?;
match client.batch_execute(sql).await {
Ok(()) => {
let exec_time = start.elapsed().as_millis() as i32;
match history::insert_applied_migration(
client,
schema,
table,
Some(version),
description,
"UNDO_SQL",
script,
checksum,
installed_by,
exec_time,
true,
)
.await
{
Ok(()) => {
client.batch_execute("COMMIT").await?;
Ok(exec_time)
}
Err(e) => {
if let Err(rb) = client.batch_execute("ROLLBACK").await {
log::error!("Failed to rollback undo transaction: {}", rb);
}
Err(e)
}
}
}
Err(e) => {
if let Err(rollback_err) = client.batch_execute("ROLLBACK").await {
log::error!("Failed to rollback undo transaction: {}", rollback_err);
}
if let Err(record_err) = history::insert_applied_migration(
client,
schema,
table,
Some(version),
description,
"UNDO_SQL",
script,
checksum,
installed_by,
0,
false,
)
.await
{
log::warn!(
"Failed to record undo failure; script={}, error={}",
script,
record_err
);
}
let reason = crate::error::format_db_error(&e);
Err(WaypointError::UndoFailed {
script: script.to_string(),
reason,
})
}
}
}
#[cfg(feature = "postgres")]
pub async fn execute(
client: &Client,
config: &WaypointConfig,
target: UndoTarget,
) -> Result<UndoReport> {
let table = &config.migrations.table;
db::acquire_advisory_lock(client, table).await?;
let result = run_undo(client, config, target).await;
if let Err(e) = db::release_advisory_lock(client, table).await {
log::error!("Failed to release advisory lock: {}", e);
}
match &result {
Ok(report) => {
log::info!(
"Undo completed; migrations_undone={}, total_time_ms={}",
report.migrations_undone,
report.total_time_ms
);
}
Err(e) => {
log::error!("Undo failed: {}", e);
}
}
result
}
#[cfg(feature = "postgres")]
async fn run_undo(
client: &Client,
config: &WaypointConfig,
target: UndoTarget,
) -> Result<UndoReport> {
let schema = &config.migrations.schema;
let table = &config.migrations.table;
history::create_history_table(client, schema, table).await?;
let resolved = scan_migrations(&config.migrations.locations)?;
let undo_by_version: HashMap<String, &ResolvedMigration> = resolved
.iter()
.filter(|m| m.is_undo())
.filter_map(|m| m.version().map(|v| (v.raw.clone(), m)))
.collect();
let applied = history::get_applied_migrations(client, schema, table).await?;
let effective = history::effective_applied_versions(&applied);
let mut applied_versions: Vec<MigrationVersion> = effective
.iter()
.filter_map(|v| MigrationVersion::parse(v).ok())
.collect();
applied_versions.sort();
applied_versions.reverse();
let versions_to_undo: Vec<MigrationVersion> = match target {
UndoTarget::Last => applied_versions.into_iter().take(1).collect(),
UndoTarget::Count(n) => applied_versions.into_iter().take(n).collect(),
UndoTarget::Version(ref target_ver) => applied_versions
.into_iter()
.filter(|v| v > target_ver)
.collect(),
};
let db_user = db::get_current_user(client)
.await
.unwrap_or_else(|_| "unknown".to_string());
let db_name = db::get_current_database(client)
.await
.unwrap_or_else(|_| "unknown".to_string());
let installed_by = config
.migrations
.installed_by
.as_deref()
.unwrap_or(&db_user);
let mut report = UndoReport {
migrations_undone: 0,
total_time_ms: 0,
details: Vec::new(),
};
for version in &versions_to_undo {
if let Some(undo_migration) = undo_by_version.get(&version.raw) {
log::info!(
"Undoing migration (manual); migration={}, schema={}",
undo_migration.script,
schema
);
let placeholders = build_placeholders(
&config.placeholders,
schema,
&db_user,
&db_name,
&undo_migration.script,
);
let sql = replace_placeholders(&undo_migration.sql, &placeholders)?;
let exec_time = execute_undo_sql(
client,
schema,
table,
&version.raw,
&undo_migration.description,
&undo_migration.script,
Some(undo_migration.checksum),
installed_by,
&sql,
)
.await?;
report.migrations_undone += 1;
report.total_time_ms += exec_time;
report.details.push(UndoDetail {
version: version.raw.clone(),
description: undo_migration.description.clone(),
script: undo_migration.script.clone(),
execution_time_ms: exec_time,
auto_reversal: false,
});
} else if config.reversals.enabled {
match crate::reversal::get_reversal(client, schema, table, &version.raw).await? {
Some(reversal_sql) => {
let script = format!("auto-reversal:V{}", version.raw);
log::info!(
"Undoing migration (auto-reversal); version={}, schema={}",
version.raw,
schema
);
let exec_time = execute_undo_sql(
client,
schema,
table,
&version.raw,
"Auto-generated reversal",
&script,
None,
installed_by,
&reversal_sql,
)
.await?;
report.migrations_undone += 1;
report.total_time_ms += exec_time;
report.details.push(UndoDetail {
version: version.raw.clone(),
description: "Auto-generated reversal".to_string(),
script,
execution_time_ms: exec_time,
auto_reversal: true,
});
}
None => {
return Err(WaypointError::UndoMissing {
version: version.raw.clone(),
});
}
}
} else {
return Err(WaypointError::UndoMissing {
version: version.raw.clone(),
});
}
}
Ok(report)
}
pub async fn execute_db(
client: &DbClient,
config: &WaypointConfig,
target: UndoTarget,
) -> Result<UndoReport> {
match client.dialect_kind() {
#[cfg(feature = "postgres")]
DialectKind::Postgres => execute(client.as_postgres()?, config, target).await,
#[cfg(not(feature = "postgres"))]
DialectKind::Postgres => Err(WaypointError::ConfigError(
"PostgreSQL support is not compiled in (enable the `postgres` feature)".into(),
)),
#[cfg(feature = "mysql")]
DialectKind::Mysql => execute_mysql(client, config, target).await,
#[cfg(not(feature = "mysql"))]
DialectKind::Mysql => Err(WaypointError::ConfigError(
"MySQL support is not compiled in (enable the `mysql` feature)".into(),
)),
}
}
#[cfg(feature = "mysql")]
async fn execute_mysql(
client: &DbClient,
config: &WaypointConfig,
target: UndoTarget,
) -> Result<UndoReport> {
let table = &config.migrations.table;
client.acquire_lock(table).await?;
let result = run_undo_mysql(client, config, target).await;
if let Err(e) = client.release_lock(table).await {
log::error!("Failed to release advisory lock: {}", e);
}
match &result {
Ok(report) => {
log::info!(
"Undo completed (mysql); migrations_undone={}, total_time_ms={}",
report.migrations_undone,
report.total_time_ms
);
}
Err(e) => {
log::error!("Undo failed (mysql): {}", e);
}
}
result
}
#[cfg(feature = "mysql")]
async fn run_undo_mysql(
client: &DbClient,
config: &WaypointConfig,
target: UndoTarget,
) -> Result<UndoReport> {
let schema = client.resolve_schema(&config.migrations.schema).await?;
let schema = schema.as_str();
let table = &config.migrations.table;
history::create_history_table_db(client, schema, table).await?;
let resolved = scan_migrations(&config.migrations.locations)?;
let undo_by_version: HashMap<String, &ResolvedMigration> = resolved
.iter()
.filter(|m| m.is_undo())
.filter_map(|m| m.version().map(|v| (v.raw.clone(), m)))
.collect();
let applied = history::get_applied_migrations_db(client, schema, table).await?;
let effective = history::effective_applied_versions(&applied);
let mut applied_versions: Vec<MigrationVersion> = effective
.iter()
.filter_map(|v| MigrationVersion::parse(v).ok())
.collect();
applied_versions.sort();
applied_versions.reverse();
let versions_to_undo: Vec<MigrationVersion> = match target {
UndoTarget::Last => applied_versions.into_iter().take(1).collect(),
UndoTarget::Count(n) => applied_versions.into_iter().take(n).collect(),
UndoTarget::Version(ref target_ver) => applied_versions
.into_iter()
.filter(|v| v > target_ver)
.collect(),
};
let db_user = client
.current_user()
.await
.unwrap_or_else(|_| "unknown".into());
let db_name = client
.current_database()
.await
.unwrap_or_else(|_| "unknown".into());
let installed_by = config
.migrations
.installed_by
.as_deref()
.unwrap_or(&db_user)
.to_string();
let mut report = UndoReport {
migrations_undone: 0,
total_time_ms: 0,
details: Vec::new(),
};
for version in &versions_to_undo {
let (sql, script, description, checksum, auto_reversal) = match undo_by_version
.get(&version.raw)
{
Some(m) => {
let placeholders =
build_placeholders(&config.placeholders, schema, &db_user, &db_name, &m.script);
let sql = replace_placeholders(&m.sql, &placeholders)?;
log::info!(
"Undoing migration (manual); migration={}, schema={}",
m.script,
schema
);
(
sql,
m.script.clone(),
m.description.clone(),
Some(m.checksum),
false,
)
}
None if config.reversals.enabled => {
match crate::reversal::get_reversal_db(client, schema, table, &version.raw).await? {
Some(reversal_sql) => {
let script = format!("auto-reversal:V{}", version.raw);
log::info!(
"Undoing migration (auto-reversal); version={}, schema={}",
version.raw,
schema
);
(
reversal_sql,
script,
"Auto-generated reversal".to_string(),
None,
true,
)
}
None => {
return Err(WaypointError::UndoMissing {
version: version.raw.clone(),
});
}
}
}
None => {
return Err(WaypointError::UndoMissing {
version: version.raw.clone(),
});
}
};
let start = std::time::Instant::now();
let exec_result = client.execute_raw(&sql).await;
let exec_time = start.elapsed().as_millis() as i32;
match exec_result {
Ok(_) => {
history::insert_applied_migration_db(
client,
schema,
table,
Some(&version.raw),
&description,
"UNDO_SQL",
&script,
checksum,
&installed_by,
exec_time,
true,
)
.await?;
report.migrations_undone += 1;
report.total_time_ms += exec_time;
report.details.push(UndoDetail {
version: version.raw.clone(),
description: description.clone(),
script: script.clone(),
execution_time_ms: exec_time,
auto_reversal,
});
}
Err(e) => {
if let Err(record_err) = history::insert_applied_migration_db(
client,
schema,
table,
Some(&version.raw),
&description,
"UNDO_SQL",
&script,
checksum,
&installed_by,
exec_time,
false,
)
.await
{
log::warn!(
"Failed to record undo failure; script={}, error={}",
script,
record_err
);
}
return Err(WaypointError::UndoFailed {
script: script.clone(),
reason: e.to_string(),
});
}
}
}
Ok(report)
}