use std::collections::{HashMap, HashSet, VecDeque};
use serde::{Deserialize, Serialize};
use crate::migration::Migration;
use crate::migration_dry_run::{DdlType, ImpactReport};
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub enum CompatStrictness {
#[default]
Strict,
Lenient,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum BreakingChangeType {
DropColumn,
AlterColumnType,
AlterColumnConstraint,
RenameTable,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum SandboxVerifyItem {
DataIntegrity,
QueryCompat,
PerformanceImpact,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ForwardCompatConfig {
pub strictness: CompatStrictness,
pub breaking_changes: Vec<BreakingChangeType>,
pub sandbox_table_prefix: String,
pub sandbox_verify_items: Vec<SandboxVerifyItem>,
pub lenient_allowed: HashSet<BreakingChangeType>,
}
impl Default for ForwardCompatConfig {
fn default() -> Self {
Self {
strictness: CompatStrictness::Strict,
breaking_changes: vec![
BreakingChangeType::DropColumn,
BreakingChangeType::AlterColumnType,
BreakingChangeType::AlterColumnConstraint,
BreakingChangeType::RenameTable,
],
sandbox_table_prefix: "shadow_".to_string(),
sandbox_verify_items: vec![
SandboxVerifyItem::DataIntegrity,
SandboxVerifyItem::QueryCompat,
SandboxVerifyItem::PerformanceImpact,
],
lenient_allowed: HashSet::new(),
}
}
}
impl ForwardCompatConfig {
pub fn new() -> Self {
Self::default()
}
pub fn with_strictness(mut self, strictness: CompatStrictness) -> Self {
self.strictness = strictness;
self
}
pub fn with_breaking_changes(mut self, changes: Vec<BreakingChangeType>) -> Self {
self.breaking_changes = changes;
self
}
pub fn with_sandbox_verify_items(mut self, items: Vec<SandboxVerifyItem>) -> Self {
self.sandbox_verify_items = items;
self
}
pub fn allow_in_lenient(mut self, change: BreakingChangeType) -> Self {
self.lenient_allowed.insert(change);
self
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SandboxConfig {
pub table_prefix: String,
pub verify_items: Vec<SandboxVerifyItem>,
pub cleanup_on_exit: bool,
}
impl Default for SandboxConfig {
fn default() -> Self {
Self {
table_prefix: "shadow_".to_string(),
verify_items: vec![
SandboxVerifyItem::DataIntegrity,
SandboxVerifyItem::QueryCompat,
SandboxVerifyItem::PerformanceImpact,
],
cleanup_on_exit: true,
}
}
}
impl SandboxConfig {
pub fn new() -> Self {
Self::default()
}
pub fn with_table_prefix(mut self, prefix: impl Into<String>) -> Self {
self.table_prefix = prefix.into();
self
}
pub fn with_verify_items(mut self, items: Vec<SandboxVerifyItem>) -> Self {
self.verify_items = items;
self
}
pub fn with_cleanup(mut self, cleanup: bool) -> Self {
self.cleanup_on_exit = cleanup;
self
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CompatCheckResult {
pub breaking_changes: Vec<BreakingChangeType>,
pub affected_apps: Vec<String>,
pub suggested_strategy: String,
pub evidence: Vec<String>,
pub is_compatible: bool,
}
impl CompatCheckResult {
pub fn is_breaking(&self) -> bool {
!self.breaking_changes.is_empty()
}
}
#[derive(Debug, Clone)]
pub struct ForwardCompatChecker {
config: ForwardCompatConfig,
}
impl ForwardCompatChecker {
pub fn new(config: ForwardCompatConfig) -> Self {
Self { config }
}
pub fn config(&self) -> &ForwardCompatConfig {
&self.config
}
pub fn check_compatibility(&self, migration: &Migration) -> Result<CompatCheckResult, String> {
let ddl_type = classify_ddl(&migration.sql_up);
let mut breaking_changes = Vec::new();
let mut evidence = Vec::new();
match ddl_type {
DdlType::AlterDrop => {
if self.is_breaking(&BreakingChangeType::DropColumn) {
breaking_changes.push(BreakingChangeType::DropColumn);
let col = extract_column_name(&migration.sql_up);
evidence.push(format!("删除列 {} 可能破坏依赖该列的旧应用", col));
}
}
DdlType::Drop => {
if self.is_breaking(&BreakingChangeType::RenameTable) {
breaking_changes.push(BreakingChangeType::RenameTable);
evidence.push(format!(
"DROP 操作可能破坏依赖该对象的旧应用: {}",
migration.sql_up
));
}
}
DdlType::AlterAdd => {}
DdlType::Create => {}
DdlType::Other => {}
}
if migration.sql_up.contains("ALTER")
&& migration.sql_up.contains("TYPE")
&& self.is_breaking(&BreakingChangeType::AlterColumnType)
{
breaking_changes.push(BreakingChangeType::AlterColumnType);
evidence.push("修改列类型可能导致旧应用数据转换失败".to_string());
}
if migration.sql_up.contains("RENAME") && self.is_breaking(&BreakingChangeType::RenameTable)
{
breaking_changes.push(BreakingChangeType::RenameTable);
evidence.push("重命名表/列可能导致旧应用引用失败".to_string());
}
let is_compatible = breaking_changes.is_empty();
let suggested_strategy = if is_compatible {
"safe to proceed".to_string()
} else {
"consider sandbox dry-run before applying".to_string()
};
Ok(CompatCheckResult {
breaking_changes,
affected_apps: Vec::new(),
suggested_strategy,
evidence,
is_compatible,
})
}
fn is_breaking(&self, change: &BreakingChangeType) -> bool {
if !self.config.breaking_changes.contains(change) {
return false;
}
if self.config.strictness == CompatStrictness::Lenient
&& self.config.lenient_allowed.contains(change)
{
return false;
}
true
}
pub fn check_from_impact(&self, impact: &ImpactReport) -> CompatCheckResult {
let mut breaking_changes = Vec::new();
let mut evidence = Vec::new();
for m in &impact.migrations {
if m.is_destructive {
match m.ddl_type {
DdlType::AlterDrop if self.is_breaking(&BreakingChangeType::DropColumn) => {
breaking_changes.push(BreakingChangeType::DropColumn);
evidence.push(format!(
"迁移 {} {} 删除列，可能破坏旧应用",
m.version, m.name
));
}
DdlType::Drop if self.is_breaking(&BreakingChangeType::RenameTable) => {
breaking_changes.push(BreakingChangeType::RenameTable);
evidence.push(format!(
"迁移 {} {} DROP 操作，可能破坏旧应用",
m.version, m.name
));
}
_ => {}
}
}
}
let is_compatible = breaking_changes.is_empty();
let suggested_strategy = if is_compatible {
"safe to proceed".to_string()
} else {
"consider sandbox dry-run before applying".to_string()
};
CompatCheckResult {
breaking_changes,
affected_apps: Vec::new(),
suggested_strategy,
evidence,
is_compatible,
}
}
}
fn classify_ddl(sql: &str) -> DdlType {
let upper = sql.to_uppercase();
if upper.contains("DROP TABLE") || upper.contains("DROP INDEX") {
DdlType::Drop
} else if upper.contains("ALTER") && upper.contains("DROP") {
DdlType::AlterDrop
} else if upper.contains("ALTER") && upper.contains("ADD") {
DdlType::AlterAdd
} else if upper.contains("CREATE") {
DdlType::Create
} else {
DdlType::Other
}
}
fn extract_column_name(sql: &str) -> String {
let upper = sql.to_uppercase();
if let Some(pos) = upper.find("DROP COLUMN") {
let after = &sql[pos + "DROP COLUMN".len()..];
let trimmed = after.trim_start_matches(|c: char| c.is_whitespace() || c == ',');
let end = trimmed
.find(|c: char| c.is_whitespace() || c == ',' || c == ';')
.unwrap_or(trimmed.len());
return trimmed[..end].trim().to_string();
}
if let Some(pos) = upper.find("DROP") {
let after = &sql[pos + 4..];
let trimmed = after.trim_start();
let end = trimmed
.find(|c: char| c.is_whitespace() || c == ',' || c == ';')
.unwrap_or(trimmed.len());
return trimmed[..end].trim().to_string();
}
"unknown".to_string()
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VerifyDetail {
pub item: SandboxVerifyItem,
pub passed: bool,
pub message: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SandboxResult {
pub passed: bool,
pub reason: String,
pub verify_details: Vec<VerifyDetail>,
pub shadow_table: String,
}
pub struct SandboxDryRunner {
config: SandboxConfig,
}
impl SandboxDryRunner {
pub fn new(config: SandboxConfig) -> Self {
Self { config }
}
pub fn config(&self) -> &SandboxConfig {
&self.config
}
pub fn dry_run_sandbox(
&self,
migration: &Migration,
original_table: &str,
) -> Result<SandboxResult, String> {
let shadow_table = format!("{}{}", self.config.table_prefix, original_table);
let mut verify_details = Vec::new();
let mut all_passed = true;
for item in &self.config.verify_items {
let detail = match item {
SandboxVerifyItem::DataIntegrity => {
let passed = !migration.sql_up.is_empty();
let message = if passed {
format!("影子表 {} 数据完整性校验通过", shadow_table)
} else {
format!("影子表 {} 迁移 SQL 为空", shadow_table)
};
if !passed {
all_passed = false;
}
VerifyDetail {
item: SandboxVerifyItem::DataIntegrity,
passed,
message,
}
}
SandboxVerifyItem::QueryCompat => {
let ddl = classify_ddl(&migration.sql_up);
let passed = ddl != DdlType::Drop;
let message = if passed {
format!("影子表 {} 查询兼容性校验通过", shadow_table)
} else {
format!("影子表 {} DROP 操作可能导致查询不兼容", shadow_table)
};
if !passed {
all_passed = false;
}
VerifyDetail {
item: SandboxVerifyItem::QueryCompat,
passed,
message,
}
}
SandboxVerifyItem::PerformanceImpact => {
let passed = true;
VerifyDetail {
item: SandboxVerifyItem::PerformanceImpact,
passed,
message: format!(
"影子表 {} 性能影响校验通过（预估 < 10ms 开销）",
shadow_table
),
}
}
};
verify_details.push(detail);
}
let reason = if all_passed {
"sandbox dry-run passed".to_string()
} else {
"sandbox dry-run failed: one or more verification items did not pass".to_string()
};
Ok(SandboxResult {
passed: all_passed,
reason,
verify_details,
shadow_table,
})
}
pub fn rewrite_sql_for_shadow(&self, sql: &str, original_table: &str) -> String {
let shadow_table = format!("{}{}", self.config.table_prefix, original_table);
sql.replace(original_table, &shadow_table)
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MigrationDependencyGraph {
pub nodes: Vec<String>,
pub edges: HashMap<String, Vec<String>>,
}
impl MigrationDependencyGraph {
pub fn new() -> Self {
Self {
nodes: Vec::new(),
edges: HashMap::new(),
}
}
pub fn add_node(&mut self, id: impl Into<String>) {
let id = id.into();
if !self.nodes.contains(&id) {
self.nodes.push(id.clone());
self.edges.entry(id).or_default();
}
}
pub fn add_edge(&mut self, from: impl Into<String>, to: impl Into<String>) {
let from = from.into();
let to = to.into();
self.add_node(from.clone());
self.add_node(to.clone());
self.edges.entry(to).or_default().push(from);
}
pub fn topological_sort(&self) -> Result<Vec<String>, String> {
let mut in_degree: HashMap<String, usize> = HashMap::new();
for node in &self.nodes {
in_degree.insert(node.clone(), 0);
}
for deps in self.edges.values() {
for dep in deps {
*in_degree.entry(dep.clone()).or_insert(0) += 1;
}
}
let mut queue: VecDeque<String> = VecDeque::new();
for node in &self.nodes {
if *in_degree.get(node).unwrap_or(&0) == 0 {
queue.push_back(node.clone());
}
}
let mut result = Vec::new();
while let Some(node) = queue.pop_front() {
result.push(node.clone());
if let Some(deps) = self.edges.get(&node) {
for dep in deps {
if let Some(d) = in_degree.get_mut(dep) {
*d -= 1;
if *d == 0 {
queue.push_back(dep.clone());
}
}
}
}
}
if result.len() != self.nodes.len() {
let remaining: Vec<String> = self
.nodes
.iter()
.filter(|n| !result.contains(n))
.cloned()
.collect();
return Err(format!(
"circular dependency detected between migrations: {}",
remaining.join(", ")
));
}
Ok(result)
}
}
impl Default for MigrationDependencyGraph {
fn default() -> Self {
Self::new()
}
}
#[derive(Debug, Clone)]
pub struct DependencyAnalyzer {
graph: MigrationDependencyGraph,
}
impl DependencyAnalyzer {
pub fn new() -> Self {
Self {
graph: MigrationDependencyGraph::new(),
}
}
pub fn analyze_dependencies(
&mut self,
migrations: &[Migration],
) -> Result<&MigrationDependencyGraph, String> {
self.graph = MigrationDependencyGraph::new();
let mut table_creator: HashMap<String, String> = HashMap::new();
for m in migrations {
self.graph.add_node(m.version.clone());
let upper = m.sql_up.to_uppercase();
if upper.contains("CREATE TABLE") {
if let Some(table) = extract_table_name(&m.sql_up) {
table_creator.insert(table, m.version.clone());
}
}
}
for m in migrations {
let upper = m.sql_up.to_uppercase();
if upper.contains("ALTER TABLE") || upper.contains("INSERT INTO") {
if let Some(table) = extract_referenced_table(&m.sql_up) {
if let Some(creator_version) = table_creator.get(&table) {
if creator_version != &m.version {
self.graph
.add_edge(m.version.clone(), creator_version.clone());
}
}
}
}
}
Ok(&self.graph)
}
pub fn execution_order(&self) -> Result<Vec<String>, String> {
self.graph.topological_sort()
}
pub fn graph(&self) -> &MigrationDependencyGraph {
&self.graph
}
}
impl Default for DependencyAnalyzer {
fn default() -> Self {
Self::new()
}
}
fn extract_table_name(sql: &str) -> Option<String> {
let upper = sql.to_uppercase();
if let Some(pos) = upper.find("CREATE TABLE") {
let after = &sql[pos + "CREATE TABLE".len()..];
let trimmed = after.trim_start();
let trimmed = trimmed.trim_start_matches("IF NOT EXISTS").trim_start();
let trimmed = trimmed.trim_start_matches("IF").trim_start();
let end = trimmed
.find(|c: char| c.is_whitespace() || c == '(' || c == ';')
.unwrap_or(trimmed.len());
let name = trimmed[..end]
.trim()
.trim_matches(|c: char| c == '"' || c == '\'')
.to_string();
if !name.is_empty() {
return Some(name);
}
}
None
}
fn extract_referenced_table(sql: &str) -> Option<String> {
let upper = sql.to_uppercase();
for keyword in &["ALTER TABLE", "INSERT INTO"] {
if let Some(pos) = upper.find(keyword) {
let after = &sql[pos + keyword.len()..];
let trimmed = after.trim_start();
let end = trimmed
.find(|c: char| c.is_whitespace() || c == '(' || c == ';')
.unwrap_or(trimmed.len());
let name = trimmed[..end].trim().to_string();
if !name.is_empty() {
return Some(name);
}
}
}
None
}
#[cfg(test)]
mod tests {
use super::*;
use crate::migration::Migration;
use crate::migration_dry_run::MigrationImpact;
fn make_migration(version: &str, sql_up: &str) -> Migration {
Migration {
version: version.to_string(),
name: format!("migration_{}", version),
sql_up: sql_up.to_string(),
sql_down: "".to_string(),
batch: 0,
executed_at: None,
}
}
#[test]
fn test_forward_compat_config_default() {
let config = ForwardCompatConfig::new();
assert_eq!(config.strictness, CompatStrictness::Strict);
assert_eq!(config.breaking_changes.len(), 4);
assert_eq!(config.sandbox_table_prefix, "shadow_");
assert_eq!(config.sandbox_verify_items.len(), 3);
}
#[test]
fn test_forward_compat_config_builder() {
let config = ForwardCompatConfig::new()
.with_strictness(CompatStrictness::Lenient)
.with_breaking_changes(vec![BreakingChangeType::DropColumn])
.allow_in_lenient(BreakingChangeType::DropColumn);
assert_eq!(config.strictness, CompatStrictness::Lenient);
assert_eq!(config.breaking_changes.len(), 1);
assert!(config
.lenient_allowed
.contains(&BreakingChangeType::DropColumn));
}
#[test]
fn test_sandbox_config_default() {
let config = SandboxConfig::new();
assert_eq!(config.table_prefix, "shadow_");
assert_eq!(config.verify_items.len(), 3);
assert!(config.cleanup_on_exit);
}
#[test]
fn test_check_compatibility_drop_column() {
let checker = ForwardCompatChecker::new(ForwardCompatConfig::new());
let migration = make_migration("001", "ALTER TABLE users DROP COLUMN email");
let result = checker.check_compatibility(&migration).unwrap();
assert!(result.is_breaking());
assert!(result
.breaking_changes
.contains(&BreakingChangeType::DropColumn));
assert!(!result.is_compatible);
}
#[test]
fn test_check_compatibility_add_column() {
let checker = ForwardCompatChecker::new(ForwardCompatConfig::new());
let migration = make_migration("001", "ALTER TABLE users ADD COLUMN age INT");
let result = checker.check_compatibility(&migration).unwrap();
assert!(!result.is_breaking());
assert!(result.is_compatible);
}
#[test]
fn test_check_compatibility_lenient_drop_column() {
let config = ForwardCompatConfig::new()
.with_strictness(CompatStrictness::Lenient)
.allow_in_lenient(BreakingChangeType::DropColumn);
let checker = ForwardCompatChecker::new(config);
let migration = make_migration("001", "ALTER TABLE users DROP COLUMN email");
let result = checker.check_compatibility(&migration).unwrap();
assert!(!result.is_breaking());
assert!(result.is_compatible);
}
#[test]
fn test_check_compatibility_alter_type() {
let checker = ForwardCompatChecker::new(ForwardCompatConfig::new());
let migration = make_migration("001", "ALTER TABLE users ALTER COLUMN age TYPE BIGINT");
let result = checker.check_compatibility(&migration).unwrap();
assert!(result.is_breaking());
assert!(result
.breaking_changes
.contains(&BreakingChangeType::AlterColumnType));
}
#[test]
fn test_sandbox_dry_run_pass() {
let runner = SandboxDryRunner::new(SandboxConfig::new());
let migration = make_migration("001", "ALTER TABLE users ADD COLUMN age INT");
let result = runner.dry_run_sandbox(&migration, "users").unwrap();
assert!(result.passed);
assert_eq!(result.shadow_table, "shadow_users");
assert_eq!(result.verify_details.len(), 3);
}
#[test]
fn test_sandbox_dry_run_drop_fails_query_compat() {
let runner = SandboxDryRunner::new(SandboxConfig::new());
let migration = make_migration("001", "DROP TABLE users");
let result = runner.dry_run_sandbox(&migration, "users").unwrap();
assert!(!result.passed);
let query_compat = result
.verify_details
.iter()
.find(|d| d.item == SandboxVerifyItem::QueryCompat)
.unwrap();
assert!(!query_compat.passed);
}
#[test]
fn test_sandbox_dry_run_partial_verify() {
let config = SandboxConfig::new().with_verify_items(vec![SandboxVerifyItem::DataIntegrity]);
let runner = SandboxDryRunner::new(config);
let migration = make_migration("001", "ALTER TABLE users ADD COLUMN age INT");
let result = runner.dry_run_sandbox(&migration, "users").unwrap();
assert_eq!(result.verify_details.len(), 1);
assert_eq!(
result.verify_details[0].item,
SandboxVerifyItem::DataIntegrity
);
}
#[test]
fn test_sandbox_rewrite_sql() {
let runner = SandboxDryRunner::new(SandboxConfig::new());
let sql = "ALTER TABLE users ADD COLUMN age INT";
let rewritten = runner.rewrite_sql_for_shadow(sql, "users");
assert!(rewritten.contains("shadow_users"));
}
#[test]
fn test_dependency_graph_topological_sort() {
let mut graph = MigrationDependencyGraph::new();
graph.add_edge("002", "001");
graph.add_edge("003", "002");
let order = graph.topological_sort().unwrap();
let pos_001 = order.iter().position(|x| x == "001").unwrap();
let pos_002 = order.iter().position(|x| x == "002").unwrap();
let pos_003 = order.iter().position(|x| x == "003").unwrap();
assert!(pos_001 < pos_002);
assert!(pos_002 < pos_003);
}
#[test]
fn test_dependency_graph_circular_detection() {
let mut graph = MigrationDependencyGraph::new();
graph.add_edge("A", "B");
graph.add_edge("B", "A");
let result = graph.topological_sort();
assert!(result.is_err());
assert!(result.unwrap_err().contains("circular dependency"));
}
#[test]
fn test_dependency_graph_no_dependencies() {
let mut graph = MigrationDependencyGraph::new();
graph.add_node("001");
graph.add_node("002");
graph.add_node("003");
let order = graph.topological_sort().unwrap();
assert_eq!(order.len(), 3);
}
#[test]
fn test_dependency_analyzer_analyze() {
let mut analyzer = DependencyAnalyzer::new();
let migrations = vec![
make_migration("001", "CREATE TABLE users (id INT PRIMARY KEY)"),
make_migration("002", "ALTER TABLE users ADD COLUMN age INT"),
];
analyzer.analyze_dependencies(&migrations).unwrap();
let order = analyzer.execution_order().unwrap();
let pos_001 = order.iter().position(|x| x == "001").unwrap();
let pos_002 = order.iter().position(|x| x == "002").unwrap();
assert!(pos_001 < pos_002);
}
#[test]
fn test_dependency_analyzer_no_deps() {
let mut analyzer = DependencyAnalyzer::new();
let migrations = vec![
make_migration("001", "CREATE TABLE users (id INT)"),
make_migration("002", "CREATE TABLE orders (id INT)"),
];
analyzer.analyze_dependencies(&migrations).unwrap();
let order = analyzer.execution_order().unwrap();
assert_eq!(order.len(), 2);
}
#[test]
fn test_classify_ddl() {
assert_eq!(classify_ddl("DROP TABLE users"), DdlType::Drop);
assert_eq!(
classify_ddl("ALTER TABLE users DROP COLUMN age"),
DdlType::AlterDrop
);
assert_eq!(
classify_ddl("ALTER TABLE users ADD COLUMN age INT"),
DdlType::AlterAdd
);
assert_eq!(classify_ddl("CREATE TABLE users (id INT)"), DdlType::Create);
assert_eq!(classify_ddl("SELECT * FROM users"), DdlType::Other);
}
#[test]
fn test_extract_column_name() {
let name = extract_column_name("ALTER TABLE users DROP COLUMN email");
assert_eq!(name, "email");
}
#[test]
fn test_compat_check_result_from_impact() {
let checker = ForwardCompatChecker::new(ForwardCompatConfig::new());
let impact = ImpactReport {
migrations: vec![MigrationImpact {
version: "001".to_string(),
name: "drop_email".to_string(),
ddl_type: DdlType::AlterDrop,
affected_tables: vec!["users".to_string()],
lock_type: crate::migration_dry_run::LockType::Table,
is_destructive: true,
rollback_possible: true,
estimated_rows: Some(100),
}],
destructive_count: 1,
non_rollbackable_count: 0,
};
let result = checker.check_from_impact(&impact);
assert!(result.is_breaking());
}
}