use std::collections::{HashMap, HashSet};
use serde::Serialize;
use crate::directive::MigrationDirectives;
use crate::error::{Result, WaypointError};
use crate::migration::{MigrationVersion, ResolvedMigration};
#[allow(deprecated)]
#[cfg(feature = "mysql")]
pub use crate::engines::mysql::migrate::execute as execute_mysql;
#[cfg(feature = "mysql")]
pub use crate::engines::mysql::migrate::execute_with_options as execute_mysql_with_options;
#[allow(deprecated)]
#[cfg(feature = "postgres")]
pub use crate::engines::postgres::migrate::execute;
#[cfg(feature = "postgres")]
pub use crate::engines::postgres::migrate::execute_with_options;
#[derive(Debug, Serialize)]
pub struct MigrateReport {
pub migrations_applied: usize,
pub total_time_ms: i32,
pub details: Vec<MigrateDetail>,
pub hooks_executed: usize,
pub hooks_time_ms: i32,
#[serde(default)]
pub skipped: Vec<SkippedMigration>,
}
#[derive(Debug, Serialize)]
pub struct SkippedMigration {
pub version: Option<String>,
pub script: String,
pub expression: String,
}
#[derive(Debug, Serialize)]
pub struct MigrateDetail {
pub version: Option<String>,
pub description: String,
pub script: String,
pub execution_time_ms: i32,
}
pub(crate) enum GuardAction {
Continue,
Skip(String),
Error(WaypointError),
}
pub(crate) fn classify_require(
outcome: Result<bool>,
expr_str: &str,
script: &str,
on_require_fail: &crate::guard::OnRequireFail,
) -> GuardAction {
match outcome {
Ok(true) => GuardAction::Continue,
Ok(false) => match on_require_fail {
crate::guard::OnRequireFail::Skip => {
log::info!(
"Guard require failed, skipping migration; script={}, expr={}",
script,
expr_str
);
GuardAction::Skip(expr_str.to_string())
}
crate::guard::OnRequireFail::Warn => {
log::warn!(
"Guard require failed (continuing); script={}, expr={}",
script,
expr_str
);
GuardAction::Continue
}
crate::guard::OnRequireFail::Error => GuardAction::Error(WaypointError::GuardFailed {
kind: "require".to_string(),
script: script.to_string(),
expression: expr_str.to_string(),
}),
},
Err(e) => {
log::warn!(
"Guard evaluation error; script={}, expr={}, error={}",
script,
expr_str,
e
);
GuardAction::Error(WaypointError::GuardFailed {
kind: "require".to_string(),
script: script.to_string(),
expression: format!("{} ({})", expr_str, describe_guard_error(&e)),
})
}
}
}
pub(crate) fn classify_ensure(outcome: Result<bool>, expr_str: &str, script: &str) -> Result<()> {
match outcome {
Ok(true) => Ok(()),
Ok(false) => Err(WaypointError::GuardFailed {
kind: "ensure".to_string(),
script: script.to_string(),
expression: expr_str.to_string(),
}),
Err(e) => Err(WaypointError::GuardFailed {
kind: "ensure".to_string(),
script: script.to_string(),
expression: format!("{} ({})", expr_str, describe_guard_error(&e)),
}),
}
}
pub(crate) fn guard_parse_error(
kind: &str,
script: &str,
expr_str: &str,
e: &WaypointError,
) -> WaypointError {
WaypointError::GuardFailed {
kind: kind.to_string(),
script: script.to_string(),
expression: format!("{} (parse error: {})", expr_str, e),
}
}
fn describe_guard_error(e: &WaypointError) -> String {
format!("evaluation error: {}", e)
}
pub(crate) struct PendingCriteria<'a> {
pub effective_versions: &'a HashSet<String>,
pub baseline_version: Option<&'a MigrationVersion>,
pub target: Option<&'a MigrationVersion>,
pub highest_applied: Option<&'a MigrationVersion>,
pub applied_scripts: &'a HashMap<String, Option<i32>>,
pub current_env: Option<&'a str>,
pub out_of_order: bool,
pub dependency_ordering: bool,
}
#[derive(Debug)]
pub(crate) struct PendingSelection<'a> {
pub versioned: Vec<&'a ResolvedMigration>,
pub repeatables: Vec<&'a ResolvedMigration>,
}
pub(crate) fn select_pending<'a>(
resolved: &'a [ResolvedMigration],
criteria: &PendingCriteria<'_>,
) -> Result<PendingSelection<'a>> {
let mut versioned: Vec<&ResolvedMigration> = Vec::new();
for migration in resolved.iter().filter(|m| m.is_versioned()) {
if !should_run_in_environment(&migration.directives, criteria.current_env) {
continue;
}
let version = match migration.version() {
Some(v) => v,
None => continue,
};
if criteria.effective_versions.contains(&version.raw) {
continue;
}
if let Some(baseline) = criteria.baseline_version
&& version <= baseline
{
log::debug!("Skipping {} (below baseline)", migration.script);
continue;
}
if let Some(target) = criteria.target
&& version > target
{
log::debug!("Skipping {} (above target {})", migration.script, target);
continue;
}
if !criteria.out_of_order
&& let Some(highest) = criteria.highest_applied
&& version < highest
{
return Err(WaypointError::OutOfOrder {
version: version.raw.clone(),
highest: highest.raw.clone(),
});
}
versioned.push(migration);
}
if criteria.dependency_ordering {
order_by_dependencies(resolved, &mut versioned)?;
} else {
versioned.sort_by(|a, b| a.version().cmp(&b.version()));
}
let repeatables: Vec<&ResolvedMigration> = resolved
.iter()
.filter(|m| !m.is_versioned() && !m.is_undo())
.filter(|m| should_run_in_environment(&m.directives, criteria.current_env))
.filter(|m| match criteria.applied_scripts.get(&m.script) {
None => true,
Some(applied) => *applied != Some(m.checksum),
})
.collect();
Ok(PendingSelection {
versioned,
repeatables,
})
}
fn order_by_dependencies(
resolved: &[ResolvedMigration],
pending: &mut Vec<&ResolvedMigration>,
) -> Result<()> {
let all: Vec<&ResolvedMigration> = resolved.iter().collect();
let graph = crate::dependency::DependencyGraph::build(&all, true)?;
let order = graph.topological_sort()?;
let rank: HashMap<&str, usize> = order
.iter()
.enumerate()
.map(|(i, v)| (v.as_str(), i))
.collect();
pending.sort_by_key(|m| {
m.version()
.and_then(|v| rank.get(v.raw.as_str()).copied())
.unwrap_or(usize::MAX)
});
Ok(())
}
pub(crate) fn should_run_in_environment(
directives: &MigrationDirectives,
current_env: Option<&str>,
) -> bool {
if directives.env.is_empty() {
return true;
}
let env = match current_env {
Some(e) => e,
None => return true,
};
directives.env.iter().any(|e| e.eq_ignore_ascii_case(env))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_should_run_in_environment_no_directives() {
let directives = MigrationDirectives::default();
assert!(should_run_in_environment(&directives, Some("production")));
assert!(should_run_in_environment(&directives, None));
}
#[test]
fn test_should_run_in_environment_matches() {
let directives = MigrationDirectives {
env: vec!["production".to_string(), "staging".to_string()],
..Default::default()
};
assert!(should_run_in_environment(&directives, Some("production")));
assert!(should_run_in_environment(&directives, Some("staging")));
assert!(!should_run_in_environment(&directives, Some("dev")));
}
#[test]
fn test_should_run_in_environment_case_insensitive() {
let directives = MigrationDirectives {
env: vec!["PROD".to_string()],
..Default::default()
};
assert!(should_run_in_environment(&directives, Some("prod")));
assert!(should_run_in_environment(&directives, Some("PROD")));
assert!(should_run_in_environment(&directives, Some("Prod")));
assert!(!should_run_in_environment(&directives, Some("dev")));
}
#[test]
fn test_should_run_in_environment_no_env_configured() {
let directives = MigrationDirectives {
env: vec!["prod".to_string()],
..Default::default()
};
assert!(should_run_in_environment(&directives, None));
}
use crate::migration::{MigrationKind, MigrationVersion};
fn mig(name: &str, depends: &[&str]) -> ResolvedMigration {
let (kind, description) = crate::migration::parse_migration_filename(name).unwrap();
ResolvedMigration {
kind,
description,
script: name.to_string(),
checksum: 1,
sql: String::new(),
directives: MigrationDirectives {
depends: depends.iter().map(|s| s.to_string()).collect(),
..Default::default()
},
}
}
fn criteria<'a>(
applied: &'a HashSet<String>,
scripts: &'a HashMap<String, Option<i32>>,
highest: Option<&'a MigrationVersion>,
) -> PendingCriteria<'a> {
PendingCriteria {
effective_versions: applied,
baseline_version: None,
target: None,
highest_applied: highest,
applied_scripts: scripts,
current_env: None,
out_of_order: false,
dependency_ordering: false,
}
}
#[test]
fn select_pending_orders_by_version() {
let migs = vec![mig("V10__Ten.sql", &[]), mig("V2__Two.sql", &[])];
let applied = HashSet::new();
let scripts = HashMap::new();
let out = select_pending(&migs, &criteria(&applied, &scripts, None)).unwrap();
let order: Vec<&str> = out.versioned.iter().map(|m| m.script.as_str()).collect();
assert_eq!(order, vec!["V2__Two.sql", "V10__Ten.sql"]);
}
#[test]
fn select_pending_errors_on_out_of_order() {
let migs = vec![mig("V1__One.sql", &[])];
let mut applied = HashSet::new();
applied.insert("5".to_string());
let scripts = HashMap::new();
let highest = MigrationVersion::parse("5").unwrap();
let err = select_pending(&migs, &criteria(&applied, &scripts, Some(&highest))).unwrap_err();
assert!(
matches!(err, WaypointError::OutOfOrder { .. }),
"expected OutOfOrder, got {err:?}"
);
}
#[test]
fn select_pending_allows_out_of_order_when_enabled() {
let migs = vec![mig("V1__One.sql", &[])];
let mut applied = HashSet::new();
applied.insert("5".to_string());
let scripts = HashMap::new();
let highest = MigrationVersion::parse("5").unwrap();
let mut c = criteria(&applied, &scripts, Some(&highest));
c.out_of_order = true;
let out = select_pending(&migs, &c).unwrap();
assert_eq!(out.versioned.len(), 1);
}
#[test]
fn select_pending_honours_depends_directive() {
let migs = vec![
mig("V1__One.sql", &[]),
mig("V2__Two.sql", &["3"]),
mig("V3__Three.sql", &[]),
];
let applied = HashSet::new();
let scripts = HashMap::new();
let mut c = criteria(&applied, &scripts, None);
c.dependency_ordering = true;
let out = select_pending(&migs, &c).unwrap();
let order: Vec<&str> = out.versioned.iter().map(|m| m.script.as_str()).collect();
assert_eq!(
order,
vec!["V1__One.sql", "V3__Three.sql", "V2__Two.sql"],
"V2 depends on V3 so it must follow it"
);
}
#[test]
fn select_pending_dependency_ordering_is_version_order_without_directives() {
let migs = vec![
mig("V1__One.sql", &[]),
mig("V2__Two.sql", &[]),
mig("V3__Three.sql", &[]),
];
let applied = HashSet::new();
let scripts = HashMap::new();
let mut c = criteria(&applied, &scripts, None);
c.dependency_ordering = true;
let out = select_pending(&migs, &c).unwrap();
let order: Vec<&str> = out.versioned.iter().map(|m| m.script.as_str()).collect();
assert_eq!(
order,
vec!["V1__One.sql", "V2__Two.sql", "V3__Three.sql"],
"no directives means dependency ordering degrades to version order"
);
}
#[test]
fn select_pending_repeatable_reruns_on_checksum_change() {
let mut r = mig("V1__One.sql", &[]);
r.kind = MigrationKind::Repeatable;
r.script = "R__View.sql".to_string();
r.checksum = 99;
let migs = vec![r];
let applied = HashSet::new();
let mut scripts = HashMap::new();
scripts.insert("R__View.sql".to_string(), Some(99));
let out = select_pending(&migs, &criteria(&applied, &scripts, None)).unwrap();
assert!(
out.repeatables.is_empty(),
"unchanged checksum must not re-run"
);
let mut scripts = HashMap::new();
scripts.insert("R__View.sql".to_string(), Some(1));
let out = select_pending(&migs, &criteria(&applied, &scripts, None)).unwrap();
assert_eq!(out.repeatables.len(), 1, "changed checksum must re-run");
}
}