sz_orm_core/
schema_sync.rs

1//! Schema Sync — 自动结构同步
2//!
3//! # 概述
4//!
5//! 比较实体定义与 DB 现有表结构差异，自动生成并执行 DDL（CREATE/ALTER TABLE）。
6//! **禁止破坏性 DDL**（ADR-v2.1.0-004）：不生成 DROP TABLE / DROP COLUMN。
7//!
8//! # 工作流
9//!
10//! 1. `introspect(conn)` → 读取 DB 现有表结构
11//! 2. `diff(entity, db)` → 计算 6 类变更
12//! 3. 检查破坏性变更 → 若有则返回 `Err(DestructiveChangeDetected)`
13//! 4. `generate(diff)` → 生成 DDL 语句
14//! 5. `sync(conn)` → 事务内执行 DDL
15//!
16//! # 示例
17//!
18//! ```ignore
19//! use sz_orm_core::schema_sync::{SchemaSync, TableDef, ColumnDef};
20//!
21//! let entity_tables = vec![
22//!     TableDef::new("users", vec![
23//!         ColumnDef::new("id", "BIGINT", false, true, None),
24//!         ColumnDef::new("email", "VARCHAR(255)", false, false, None),
25//!     ]),
26//! ];
27//!
28//! let sync = SchemaSync::new(entity_tables);
29//! let ddl = sync.sync_dry_run(&mut conn).await?;
30//! // ddl: ["ALTER TABLE users ADD COLUMN email VARCHAR(255)"]
31//! ```
32
33use crate::pool::Connection;
34use crate::DbError;
35
36// ============================================================================
37// 类型定义
38// ============================================================================
39
40/// 列定义
41#[derive(Debug, Clone, PartialEq)]
42pub struct ColumnDef {
43    /// 列名
44    pub name: String,
45    /// SQL 类型（如 "VARCHAR(255)"、"BIGINT"）
46    pub sql_type: String,
47    /// 是否允许 NULL
48    pub nullable: bool,
49    /// 是否主键
50    pub primary_key: bool,
51    /// 默认值（None 表示无默认值）
52    pub default: Option<String>,
53}
54
55impl ColumnDef {
56    /// 创建新的列定义
57    pub fn new(
58        name: impl Into<String>,
59        sql_type: impl Into<String>,
60        nullable: bool,
61        primary_key: bool,
62        default: Option<String>,
63    ) -> Self {
64        Self {
65            name: name.into(),
66            sql_type: sql_type.into(),
67            nullable,
68            primary_key,
69            default,
70        }
71    }
72}
73
74/// 表定义
75#[derive(Debug, Clone, PartialEq)]
76pub struct TableDef {
77    /// 表名
78    pub name: String,
79    /// 列定义列表
80    pub columns: Vec<ColumnDef>,
81}
82
83impl TableDef {
84    /// 创建新的表定义
85    pub fn new(name: impl Into<String>, columns: Vec<ColumnDef>) -> Self {
86        Self {
87            name: name.into(),
88            columns,
89        }
90    }
91
92    /// 按列名查找列
93    pub fn get_column(&self, name: &str) -> Option<&ColumnDef> {
94        self.columns.iter().find(|c| c.name == name)
95    }
96}
97
98/// Schema 差异
99#[derive(Debug, Clone, Default)]
100pub struct SchemaDiff {
101    /// 新增的表（实体有，DB 无）
102    pub added_tables: Vec<TableDef>,
103    /// 删除的表（DB 有，实体无）— 破坏性，仅记录不执行
104    pub dropped_tables: Vec<String>,
105    /// 新增的列（实体有，DB 无）
106    pub added_columns: Vec<(String, ColumnDef)>,
107    /// 删除的列（DB 有，实体无）— 破坏性，仅记录不执行
108    pub dropped_columns: Vec<(String, String)>,
109    /// 类型变更的列
110    pub type_changed_columns: Vec<(String, ColumnDef, ColumnDef)>,
111    /// 重命名的列（启发式：同位置不同名）
112    pub renamed_columns: Vec<(String, String, String)>,
113}
114
115impl SchemaDiff {
116    /// 是否为空（无任何变更）
117    pub fn is_empty(&self) -> bool {
118        self.added_tables.is_empty()
119            && self.dropped_tables.is_empty()
120            && self.added_columns.is_empty()
121            && self.dropped_columns.is_empty()
122            && self.type_changed_columns.is_empty()
123            && self.renamed_columns.is_empty()
124    }
125
126    /// 是否含破坏性变更
127    pub fn has_destructive_changes(&self) -> bool {
128        !self.dropped_tables.is_empty() || !self.dropped_columns.is_empty()
129    }
130}
131
132/// 同步结果
133#[derive(Debug, Clone)]
134pub struct SyncResult {
135    /// 受影响表名列表
136    pub affected_tables: Vec<String>,
137    /// 执行的 DDL 语句列表
138    pub executed_ddl: Vec<String>,
139}
140
141/// 破坏性同步确认枚举（v2.2.0 B-2）
142///
143/// 调用 [`SchemaSync::destructive_sync`] 时必须显式传入确认值。
144#[derive(Debug, Clone, Copy, PartialEq, Eq)]
145pub enum Confirm {
146    /// 确认执行破坏性 DDL
147    Yes,
148    /// 拒绝执行破坏性 DDL
149    No,
150}
151
152/// 数据迁移钩子 trait（v2.2.0 B-2）
153///
154/// 在执行破坏性 DDL 前调用，允许用户执行数据备份或迁移。
155/// 手动解糖 async（与 `Connection` trait 一致）。
156pub trait DataMigrationHook: Send + Sync {
157    /// 删除列前钩子（可执行数据备份）
158    fn before_drop_column<'a>(
159        &'a self,
160        conn: &'a mut dyn Connection,
161        table: &'a str,
162        column: &'a str,
163    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), DbError>> + Send + 'a>>;
164
165    /// 重命名列前钩子（可执行数据校验）
166    fn before_rename_column<'a>(
167        &'a self,
168        conn: &'a mut dyn Connection,
169        table: &'a str,
170        old_name: &'a str,
171        new_name: &'a str,
172    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), DbError>> + Send + 'a>>;
173}
174
175/// 破坏性同步结果（v2.2.0 B-2）
176#[derive(Debug, Clone)]
177pub struct DestructiveSyncResult {
178    /// 执行的 DDL 语句列表
179    pub executed_ddl: Vec<String>,
180    /// 调用的钩子次数
181    pub hooks_called: usize,
182    /// 审计日志条目数
183    pub audit_entries: usize,
184}
185
186// ============================================================================
187// diff 纯函数
188// ============================================================================
189
190/// 比较实体定义与 DB 现有表结构，输出 6 类变更
191///
192/// # 参数
193///
194/// - `entity`：实体定义的表结构列表
195/// - `db`：DB 现有的表结构列表
196///
197/// # 返回
198///
199/// `SchemaDiff` 含 6 类变更
200pub fn diff(entity: &[TableDef], db: &[TableDef]) -> SchemaDiff {
201    let mut result = SchemaDiff::default();
202
203    let db_map: std::collections::HashMap<&str, &TableDef> =
204        db.iter().map(|t| (t.name.as_str(), t)).collect();
205    let entity_map: std::collections::HashMap<&str, &TableDef> =
206        entity.iter().map(|t| (t.name.as_str(), t)).collect();
207
208    // 新增/删除的表
209    for t in entity {
210        if !db_map.contains_key(t.name.as_str()) {
211            result.added_tables.push(t.clone());
212        }
213    }
214    for t in db {
215        if !entity_map.contains_key(t.name.as_str()) {
216            result.dropped_tables.push(t.name.clone());
217        }
218    }
219
220    // 列级 diff（仅比较两边都存在的表）
221    for entity_table in entity {
222        if let Some(db_table) = db_map.get(entity_table.name.as_str()) {
223            diff_columns(&mut result, entity_table, db_table);
224        }
225    }
226
227    result
228}
229
230/// 比较两个表的列差异
231fn diff_columns(result: &mut SchemaDiff, entity: &TableDef, db: &TableDef) {
232    diff_columns_with_threshold(result, entity, db, 2, 0.3);
233}
234
235/// 比较两个表的列差异（带重命名检测阈值，v2.2.0 B-2）
236fn diff_columns_with_threshold(
237    result: &mut SchemaDiff,
238    entity: &TableDef,
239    db: &TableDef,
240    max_distance: usize,
241    max_ratio: f64,
242) {
243    let db_col_map: std::collections::HashMap<&str, &ColumnDef> =
244        db.columns.iter().map(|c| (c.name.as_str(), c)).collect();
245    let entity_col_map: std::collections::HashMap<&str, &ColumnDef> = entity
246        .columns
247        .iter()
248        .map(|c| (c.name.as_str(), c))
249        .collect();
250
251    let mut added_columns: Vec<&ColumnDef> = Vec::new();
252    let mut dropped_columns: Vec<&ColumnDef> = Vec::new();
253
254    for col in &entity.columns {
255        if !db_col_map.contains_key(col.name.as_str()) {
256            added_columns.push(col);
257        }
258    }
259    for col in &db.columns {
260        if !entity_col_map.contains_key(col.name.as_str()) {
261            dropped_columns.push(col);
262        }
263    }
264
265    // 重命名检测：Levenshtein 启发式（v2.2.0 B-2）
266    let mut renamed_added: Vec<usize> = Vec::new();
267    let mut renamed_dropped: Vec<usize> = Vec::new();
268    for (i, dropped_col) in dropped_columns.iter().enumerate() {
269        if renamed_dropped.contains(&i) {
270            continue;
271        }
272        for (j, added_col) in added_columns.iter().enumerate() {
273            if renamed_added.contains(&j) {
274                continue;
275            }
276            if dropped_col.sql_type != added_col.sql_type {
277                continue;
278            }
279            let dist = levenshtein(&dropped_col.name, &added_col.name);
280            let ratio = dist as f64 / dropped_col.name.len().max(added_col.name.len()) as f64;
281            if dist <= max_distance || ratio <= max_ratio {
282                result.renamed_columns.push((
283                    entity.name.clone(),
284                    dropped_col.name.clone(),
285                    added_col.name.clone(),
286                ));
287                renamed_dropped.push(i);
288                renamed_added.push(j);
289                break;
290            }
291        }
292    }
293
294    for (j, col) in added_columns.iter().enumerate() {
295        if !renamed_added.contains(&j) {
296            result
297                .added_columns
298                .push((entity.name.clone(), (*col).clone()));
299        }
300    }
301    for (i, col) in dropped_columns.iter().enumerate() {
302        if !renamed_dropped.contains(&i) {
303            result
304                .dropped_columns
305                .push((entity.name.clone(), col.name.clone()));
306        }
307    }
308
309    // 类型变更
310    for entity_col in &entity.columns {
311        if let Some(db_col) = db_col_map.get(entity_col.name.as_str()) {
312            if entity_col.sql_type != db_col.sql_type || entity_col.nullable != db_col.nullable {
313                result.type_changed_columns.push((
314                    entity.name.clone(),
315                    (*db_col).clone(),
316                    entity_col.clone(),
317                ));
318            }
319        }
320    }
321}
322
323/// Levenshtein 编辑距离（v2.2.0 B-2）
324fn levenshtein(a: &str, b: &str) -> usize {
325    let a_chars: Vec<char> = a.chars().collect();
326    let b_chars: Vec<char> = b.chars().collect();
327    let m = a_chars.len();
328    let n = b_chars.len();
329
330    if m == 0 {
331        return n;
332    }
333    if n == 0 {
334        return m;
335    }
336
337    let mut prev: Vec<usize> = (0..=n).collect();
338    let mut curr = vec![0usize; n + 1];
339
340    for i in 1..=m {
341        curr[0] = i;
342        for j in 1..=n {
343            let cost = if a_chars[i - 1] == b_chars[j - 1] {
344                0
345            } else {
346                1
347            };
348            curr[j] = (prev[j] + 1).min(curr[j - 1] + 1).min(prev[j - 1] + cost);
349        }
350        std::mem::swap(&mut prev, &mut curr);
351    }
352
353    prev[n]
354}
355
356// ============================================================================
357// DDL 生成
358// ============================================================================
359
360/// DDL 生成器 trait
361pub trait DdlGenerator: Send + Sync {
362    /// 根据 SchemaDiff 生成 DDL 语句列表
363    ///
364    /// **不生成破坏性 DDL**（DROP TABLE / DROP COLUMN）。
365    fn generate(&self, diff: &SchemaDiff) -> Result<Vec<String>, DbError>;
366}
367
368/// MySQL DDL 生成器
369pub struct MySqlDdlGenerator;
370
371impl DdlGenerator for MySqlDdlGenerator {
372    fn generate(&self, diff: &SchemaDiff) -> Result<Vec<String>, DbError> {
373        let mut ddl = Vec::new();
374
375        // 新增表
376        for table in &diff.added_tables {
377            ddl.push(generate_create_table_mysql(table));
378        }
379
380        // 新增列
381        for (table, col) in &diff.added_columns {
382            ddl.push(format!(
383                "ALTER TABLE {} ADD COLUMN {} {}{}{}",
384                table,
385                col.name,
386                col.sql_type,
387                if col.nullable { "" } else { " NOT NULL" },
388                if col.primary_key { " PRIMARY KEY" } else { "" }
389            ));
390        }
391
392        // 类型变更
393        for (table, _old, new) in &diff.type_changed_columns {
394            ddl.push(format!(
395                "ALTER TABLE {} MODIFY COLUMN {} {}{}",
396                table,
397                new.name,
398                new.sql_type,
399                if new.nullable { "" } else { " NOT NULL" }
400            ));
401        }
402
403        // 重命名
404        for (table, old, new) in &diff.renamed_columns {
405            ddl.push(format!(
406                "ALTER TABLE {} RENAME COLUMN {} TO {}",
407                table, old, new
408            ));
409        }
410
411        Ok(ddl)
412    }
413}
414
415/// 生成 MySQL CREATE TABLE
416fn generate_create_table_mysql(table: &TableDef) -> String {
417    let columns: Vec<String> = table
418        .columns
419        .iter()
420        .map(|c| {
421            format!(
422                "{} {}{}{}{}",
423                c.name,
424                c.sql_type,
425                if c.nullable { "" } else { " NOT NULL" },
426                if c.primary_key { " PRIMARY KEY" } else { "" },
427                c.default
428                    .as_ref()
429                    .map(|d| format!(" DEFAULT {}", d))
430                    .unwrap_or_default()
431            )
432        })
433        .collect();
434
435    format!("CREATE TABLE {} ({})", table.name, columns.join(", "))
436}
437
438/// PostgreSQL DDL 生成器
439pub struct PgDdlGenerator;
440
441impl DdlGenerator for PgDdlGenerator {
442    fn generate(&self, diff: &SchemaDiff) -> Result<Vec<String>, DbError> {
443        let mut ddl = Vec::new();
444
445        for table in &diff.added_tables {
446            ddl.push(generate_create_table_mysql(table)); // PG 语法与 MySQL 类似
447        }
448
449        for (table, col) in &diff.added_columns {
450            ddl.push(format!(
451                "ALTER TABLE {} ADD COLUMN {} {}{}{}",
452                table,
453                col.name,
454                col.sql_type,
455                if col.nullable { "" } else { " NOT NULL" },
456                if col.primary_key { " PRIMARY KEY" } else { "" }
457            ));
458        }
459
460        for (table, _old, new) in &diff.type_changed_columns {
461            ddl.push(format!(
462                "ALTER TABLE {} ALTER COLUMN {} TYPE {}",
463                table, new.name, new.sql_type
464            ));
465        }
466
467        for (table, old, new) in &diff.renamed_columns {
468            ddl.push(format!(
469                "ALTER TABLE {} RENAME COLUMN {} TO {}",
470                table, old, new
471            ));
472        }
473
474        Ok(ddl)
475    }
476}
477
478/// SQLite DDL 生成器
479pub struct SqliteDdlGenerator;
480
481impl DdlGenerator for SqliteDdlGenerator {
482    fn generate(&self, diff: &SchemaDiff) -> Result<Vec<String>, DbError> {
483        let mut ddl = Vec::new();
484
485        for table in &diff.added_tables {
486            ddl.push(generate_create_table_mysql(table));
487        }
488
489        for (table, col) in &diff.added_columns {
490            // SQLite 新增列必须允许 NULL 或有默认值
491            ddl.push(format!(
492                "ALTER TABLE {} ADD COLUMN {} {}{}",
493                table,
494                col.name,
495                col.sql_type,
496                col.default
497                    .as_ref()
498                    .map(|d| format!(" DEFAULT {}", d))
499                    .unwrap_or_else(|| " DEFAULT NULL".to_string())
500            ));
501        }
502
503        // SQLite 不支持 ALTER COLUMN TYPE，返回错误
504        if !diff.type_changed_columns.is_empty() {
505            return Err(DbError::Unsupported(
506                "SQLite does not support altering column type; table rebuild required".to_string(),
507            ));
508        }
509
510        for (table, old, new) in &diff.renamed_columns {
511            ddl.push(format!(
512                "ALTER TABLE {} RENAME COLUMN {} TO {}",
513                table, old, new
514            ));
515        }
516
517        Ok(ddl)
518    }
519}
520
521/// Oracle DDL 生成器
522pub struct OracleDdlGenerator;
523
524impl DdlGenerator for OracleDdlGenerator {
525    fn generate(&self, diff: &SchemaDiff) -> Result<Vec<String>, DbError> {
526        let mut ddl = Vec::new();
527
528        for table in &diff.added_tables {
529            ddl.push(generate_create_table_mysql(table));
530        }
531
532        for (table, col) in &diff.added_columns {
533            ddl.push(format!(
534                "ALTER TABLE {} ADD ({} {}{}{})",
535                table,
536                col.name,
537                col.sql_type,
538                if col.nullable { "" } else { " NOT NULL" },
539                if col.primary_key { " PRIMARY KEY" } else { "" }
540            ));
541        }
542
543        for (table, _old, new) in &diff.type_changed_columns {
544            ddl.push(format!(
545                "ALTER TABLE {} MODIFY ({} {}{})",
546                table,
547                new.name,
548                new.sql_type,
549                if new.nullable { "" } else { " NOT NULL" }
550            ));
551        }
552
553        for (table, old, new) in &diff.renamed_columns {
554            ddl.push(format!(
555                "ALTER TABLE {} RENAME COLUMN {} TO {}",
556                table, old, new
557            ));
558        }
559
560        Ok(ddl)
561    }
562}
563
564/// MSSQL DDL 生成器
565pub struct MssqlDdlGenerator;
566
567impl DdlGenerator for MssqlDdlGenerator {
568    fn generate(&self, diff: &SchemaDiff) -> Result<Vec<String>, DbError> {
569        let mut ddl = Vec::new();
570
571        for table in &diff.added_tables {
572            ddl.push(generate_create_table_mysql(table));
573        }
574
575        for (table, col) in &diff.added_columns {
576            ddl.push(format!(
577                "ALTER TABLE {} ADD {} {}{}{}",
578                table,
579                col.name,
580                col.sql_type,
581                if col.nullable { "" } else { " NOT NULL" },
582                if col.primary_key { " PRIMARY KEY" } else { "" }
583            ));
584        }
585
586        for (table, _old, new) in &diff.type_changed_columns {
587            ddl.push(format!(
588                "ALTER TABLE {} ALTER COLUMN {} {}{}",
589                table,
590                new.name,
591                new.sql_type,
592                if new.nullable { "" } else { " NOT NULL" }
593            ));
594        }
595
596        for (table, old, new) in &diff.renamed_columns {
597            ddl.push(format!(
598                "EXEC sp_rename '{}.{}', '{}', 'COLUMN'",
599                table, old, new
600            ));
601        }
602
603        Ok(ddl)
604    }
605}
606
607// ============================================================================
608// SchemaSync 协调器
609// ============================================================================
610
611/// Schema Sync 协调器
612pub struct SchemaSync {
613    /// 实体定义的表结构
614    entity_tables: Vec<TableDef>,
615    /// DDL 生成器
616    ddl_generator: Box<dyn DdlGenerator>,
617    /// 重命名检测最大 Levenshtein 距离（v2.2.0 B-2）
618    rename_max_distance: usize,
619    /// 重命名检测最大距离/长度比（v2.2.0 B-2）
620    rename_max_ratio: f64,
621}
622
623impl SchemaSync {
624    /// 创建 SchemaSync（按方言选择 DDL 生成器）
625    pub fn new(entity_tables: Vec<TableDef>) -> Self {
626        Self {
627            entity_tables,
628            ddl_generator: Box::new(MySqlDdlGenerator),
629            rename_max_distance: 2,
630            rename_max_ratio: 0.3,
631        }
632    }
633
634    /// 创建 SchemaSync（指定 DDL 生成器）
635    pub fn with_generator(
636        entity_tables: Vec<TableDef>,
637        ddl_generator: Box<dyn DdlGenerator>,
638    ) -> Self {
639        Self {
640            entity_tables,
641            ddl_generator,
642            rename_max_distance: 2,
643            rename_max_ratio: 0.3,
644        }
645    }
646
647    /// 配置重命名检测阈值（v2.2.0 B-2）
648    pub fn with_rename_threshold(mut self, max_distance: usize, max_ratio: f64) -> Self {
649        self.rename_max_distance = max_distance;
650        self.rename_max_ratio = max_ratio;
651        self
652    }
653
654    /// 干运行：计算 DDL 但不执行
655    ///
656    /// 1. introspect → 读取 DB 现有表结构
657    /// 2. diff → 计算变更
658    /// 3. 检查破坏性变更 → 若有则返回 `Err(DestructiveChangeDetected)`
659    /// 4. generate → 生成 DDL
660    pub async fn sync_dry_run(&self, conn: &mut dyn Connection) -> Result<Vec<String>, DbError> {
661        let db_tables = introspect(conn).await?;
662        let diff_result = self.diff_against(&db_tables);
663
664        if diff_result.has_destructive_changes() {
665            return Err(DbError::Internal(format!(
666                "DestructiveChangeDetected: dropped_tables={:?}, dropped_columns={:?}",
667                diff_result.dropped_tables, diff_result.dropped_columns
668            )));
669        }
670
671        self.ddl_generator.generate(&diff_result)
672    }
673
674    /// 执行同步：事务内执行 DDL
675    ///
676    /// 1. sync_dry_run → 获取 DDL
677    /// 2. begin_transaction
678    /// 3. 逐条执行 DDL
679    /// 4. commit / rollback
680    pub async fn sync(&self, conn: &mut dyn Connection) -> Result<SyncResult, DbError> {
681        let ddl = self.sync_dry_run(conn).await?;
682
683        if ddl.is_empty() {
684            return Ok(SyncResult {
685                affected_tables: Vec::new(),
686                executed_ddl: Vec::new(),
687            });
688        }
689
690        conn.begin_transaction().await?;
691
692        let mut executed = Vec::new();
693        for ddl_stmt in &ddl {
694            match conn.execute(ddl_stmt).await {
695                Ok(_) => executed.push(ddl_stmt.clone()),
696                Err(e) => {
697                    let _ = conn.rollback().await;
698                    return Err(DbError::Internal(format!(
699                        "DDL execution failed: {} — SQL: {}",
700                        e, ddl_stmt
701                    )));
702                }
703            }
704        }
705
706        conn.commit().await?;
707
708        Ok(SyncResult {
709            affected_tables: self.entity_tables.iter().map(|t| t.name.clone()).collect(),
710            executed_ddl: executed,
711        })
712    }
713
714    /// 仅计算 diff（不连接 DB）
715    pub fn diff_against(&self, db_tables: &[TableDef]) -> SchemaDiff {
716        let mut result = SchemaDiff::default();
717
718        let db_map: std::collections::HashMap<&str, &TableDef> =
719            db_tables.iter().map(|t| (t.name.as_str(), t)).collect();
720        let entity_map: std::collections::HashMap<&str, &TableDef> = self
721            .entity_tables
722            .iter()
723            .map(|t| (t.name.as_str(), t))
724            .collect();
725
726        for t in &self.entity_tables {
727            if !db_map.contains_key(t.name.as_str()) {
728                result.added_tables.push(t.clone());
729            }
730        }
731        for t in db_tables {
732            if !entity_map.contains_key(t.name.as_str()) {
733                result.dropped_tables.push(t.name.clone());
734            }
735        }
736
737        for entity_table in &self.entity_tables {
738            if let Some(db_table) = db_map.get(entity_table.name.as_str()) {
739                diff_columns_with_threshold(
740                    &mut result,
741                    entity_table,
742                    db_table,
743                    self.rename_max_distance,
744                    self.rename_max_ratio,
745                );
746            }
747        }
748
749        result
750    }
751
752    /// 执行破坏性同步（v2.2.0 B-2）
753    ///
754    /// 显式执行破坏性 DDL（DROP COLUMN / RENAME COLUMN），需 `Confirm::Yes` 确认。
755    /// 事务内原子执行，每条破坏性 DDL 前调用对应钩子。
756    ///
757    /// # 参数
758    ///
759    /// - `conn`：数据库连接
760    /// - `confirm`：显式确认（必须 `Confirm::Yes` 才执行）
761    /// - `hooks`：可选数据迁移钩子
762    ///
763    /// # 异常处理
764    ///
765    /// - `confirm == Confirm::No` → 返回 `Err` 要求显式确认
766    /// - 钩子失败 → ROLLBACK 返回 Err
767    /// - DDL 执行失败 → ROLLBACK 返回 Err
768    pub async fn destructive_sync(
769        &self,
770        conn: &mut dyn Connection,
771        confirm: Confirm,
772        hooks: Option<&dyn DataMigrationHook>,
773    ) -> Result<DestructiveSyncResult, DbError> {
774        if confirm != Confirm::Yes {
775            return Err(DbError::InvalidInput(
776                "破坏性同步需要显式确认：请传入 Confirm::Yes".to_string(),
777            ));
778        }
779
780        let db_tables = introspect(conn).await?;
781        let diff_result = self.diff_against(&db_tables);
782        let ddl = self.ddl_generator.generate(&diff_result)?;
783
784        let mut destructive_ddl = Vec::new();
785        for (table, col) in &diff_result.dropped_columns {
786            destructive_ddl.push(format!("ALTER TABLE {} DROP COLUMN {}", table, col));
787        }
788        for (table, old, new) in &diff_result.renamed_columns {
789            destructive_ddl.push(format!(
790                "ALTER TABLE {} RENAME COLUMN {} TO {}",
791                table, old, new
792            ));
793        }
794
795        let all_ddl: Vec<String> = ddl.into_iter().chain(destructive_ddl).collect();
796        if all_ddl.is_empty() {
797            return Ok(DestructiveSyncResult {
798                executed_ddl: Vec::new(),
799                hooks_called: 0,
800                audit_entries: 0,
801            });
802        }
803
804        conn.begin_transaction().await?;
805
806        let mut executed = Vec::new();
807        let mut hooks_called = 0usize;
808
809        for ddl_stmt in &all_ddl {
810            if let Some(hook) = hooks {
811                if ddl_stmt.contains("DROP COLUMN") {
812                    let parts: Vec<&str> = ddl_stmt.split_whitespace().collect();
813                    if parts.len() >= 5 {
814                        let table = parts[2];
815                        let column = parts[4];
816                        if let Err(e) = hook.before_drop_column(conn, table, column).await {
817                            let _ = conn.rollback().await;
818                            return Err(DbError::Hook(format!(
819                                "before_drop_column 钩子失败: {}",
820                                e
821                            )));
822                        }
823                        hooks_called += 1;
824                    }
825                } else if ddl_stmt.contains("RENAME COLUMN") {
826                    let parts: Vec<&str> = ddl_stmt.split_whitespace().collect();
827                    if parts.len() >= 6 {
828                        let table = parts[2];
829                        let old_name = parts[4];
830                        let new_name = parts[6];
831                        if let Err(e) = hook
832                            .before_rename_column(conn, table, old_name, new_name)
833                            .await
834                        {
835                            let _ = conn.rollback().await;
836                            return Err(DbError::Hook(format!(
837                                "before_rename_column 钩子失败: {}",
838                                e
839                            )));
840                        }
841                        hooks_called += 1;
842                    }
843                }
844            }
845
846            match conn.execute(ddl_stmt).await {
847                Ok(_) => executed.push(ddl_stmt.clone()),
848                Err(e) => {
849                    let _ = conn.rollback().await;
850                    return Err(DbError::Internal(format!(
851                        "DDL execution failed: {} — SQL: {}",
852                        e, ddl_stmt
853                    )));
854                }
855            }
856        }
857
858        conn.commit().await?;
859
860        Ok(DestructiveSyncResult {
861            audit_entries: executed.len(),
862            executed_ddl: executed,
863            hooks_called,
864        })
865    }
866}
867
868// ============================================================================
869// 内省（简化版：从 DB 读取表结构）
870// ============================================================================
871
872/// 从 DB 读取现有表结构
873///
874/// 简化实现：返回空列表（实际应由各方言 introspector 实现）
875async fn introspect(conn: &mut dyn Connection) -> Result<Vec<TableDef>, DbError> {
876    // 简化：查询 information_schema.tables 获取表列表
877    // 实际实现应由各方言 introspector 提供
878    let _ = conn;
879    Ok(Vec::new())
880}
881
882// ============================================================================
883// 单元测试
884// ============================================================================
885
886#[cfg(test)]
887mod tests {
888    use super::*;
889
890    fn make_column(name: &str, sql_type: &str) -> ColumnDef {
891        ColumnDef::new(name, sql_type, true, false, None)
892    }
893
894    fn make_table(name: &str, columns: Vec<ColumnDef>) -> TableDef {
895        TableDef::new(name, columns)
896    }
897
898    #[test]
899    fn test_diff_add_table() {
900        let entity = vec![make_table("users", vec![make_column("id", "BIGINT")])];
901        let db = vec![];
902
903        let result = diff(&entity, &db);
904
905        assert_eq!(result.added_tables.len(), 1);
906        assert_eq!(result.added_tables[0].name, "users");
907    }
908
909    #[test]
910    fn test_diff_drop_table() {
911        let entity = vec![];
912        let db = vec![make_table("legacy", vec![make_column("id", "BIGINT")])];
913
914        let result = diff(&entity, &db);
915
916        assert_eq!(result.dropped_tables.len(), 1);
917        assert_eq!(result.dropped_tables[0], "legacy");
918        assert!(result.has_destructive_changes());
919    }
920
921    #[test]
922    fn test_diff_add_column() {
923        let entity = vec![make_table(
924            "users",
925            vec![
926                make_column("id", "BIGINT"),
927                make_column("email", "VARCHAR(255)"),
928            ],
929        )];
930        let db = vec![make_table("users", vec![make_column("id", "BIGINT")])];
931
932        let result = diff(&entity, &db);
933
934        assert_eq!(result.added_columns.len(), 1);
935        assert_eq!(result.added_columns[0].0, "users");
936        assert_eq!(result.added_columns[0].1.name, "email");
937    }
938
939    #[test]
940    fn test_diff_drop_column() {
941        let entity = vec![make_table("users", vec![make_column("id", "BIGINT")])];
942        let db = vec![make_table(
943            "users",
944            vec![
945                make_column("id", "BIGINT"),
946                make_column("legacy_col", "TEXT"),
947            ],
948        )];
949
950        let result = diff(&entity, &db);
951
952        assert_eq!(result.dropped_columns.len(), 1);
953        assert_eq!(
954            result.dropped_columns[0],
955            ("users".to_string(), "legacy_col".to_string())
956        );
957        assert!(result.has_destructive_changes());
958    }
959
960    #[test]
961    fn test_diff_type_change() {
962        let entity = vec![make_table(
963            "users",
964            vec![
965                make_column("id", "BIGINT"),
966                make_column("name", "VARCHAR(255)"),
967            ],
968        )];
969        let db = vec![make_table(
970            "users",
971            vec![
972                make_column("id", "BIGINT"),
973                make_column("name", "VARCHAR(100)"),
974            ],
975        )];
976
977        let result = diff(&entity, &db);
978
979        assert_eq!(result.type_changed_columns.len(), 1);
980        assert_eq!(result.type_changed_columns[0].0, "users");
981        assert_eq!(result.type_changed_columns[0].1.sql_type, "VARCHAR(100)");
982        assert_eq!(result.type_changed_columns[0].2.sql_type, "VARCHAR(255)");
983    }
984
985    #[test]
986    fn test_diff_no_change() {
987        let entity = vec![make_table("users", vec![make_column("id", "BIGINT")])];
988        let db = vec![make_table("users", vec![make_column("id", "BIGINT")])];
989
990        let result = diff(&entity, &db);
991
992        assert!(result.is_empty());
993    }
994
995    #[test]
996    fn test_mysql_ddl_add_table() {
997        let diff_result = SchemaDiff {
998            added_tables: vec![make_table(
999                "users",
1000                vec![ColumnDef::new("id", "BIGINT", false, true, None)],
1001            )],
1002            ..Default::default()
1003        };
1004
1005        let ddl = MySqlDdlGenerator.generate(&diff_result).unwrap();
1006        assert_eq!(ddl.len(), 1);
1007        assert!(ddl[0].contains("CREATE TABLE users"));
1008        assert!(ddl[0].contains("id BIGINT NOT NULL PRIMARY KEY"));
1009    }
1010
1011    #[test]
1012    fn test_mysql_ddl_add_column() {
1013        let diff_result = SchemaDiff {
1014            added_columns: vec![(
1015                "users".to_string(),
1016                ColumnDef::new("email", "VARCHAR(255)", false, false, None),
1017            )],
1018            ..Default::default()
1019        };
1020
1021        let ddl = MySqlDdlGenerator.generate(&diff_result).unwrap();
1022        assert_eq!(ddl.len(), 1);
1023        assert!(ddl[0].contains("ALTER TABLE users ADD COLUMN email VARCHAR(255) NOT NULL"));
1024    }
1025
1026    #[test]
1027    fn test_pg_ddl_type_change() {
1028        let diff_result = SchemaDiff {
1029            type_changed_columns: vec![(
1030                "users".to_string(),
1031                ColumnDef::new("name", "VARCHAR(100)", true, false, None),
1032                ColumnDef::new("name", "VARCHAR(255)", true, false, None),
1033            )],
1034            ..Default::default()
1035        };
1036
1037        let ddl = PgDdlGenerator.generate(&diff_result).unwrap();
1038        assert_eq!(ddl.len(), 1);
1039        assert!(ddl[0].contains("ALTER TABLE users ALTER COLUMN name TYPE VARCHAR(255)"));
1040    }
1041
1042    #[test]
1043    fn test_sqlite_ddl_type_change_unsupported() {
1044        let diff_result = SchemaDiff {
1045            type_changed_columns: vec![(
1046                "users".to_string(),
1047                ColumnDef::new("name", "VARCHAR(100)", true, false, None),
1048                ColumnDef::new("name", "VARCHAR(255)", true, false, None),
1049            )],
1050            ..Default::default()
1051        };
1052
1053        let result = SqliteDdlGenerator.generate(&diff_result);
1054        assert!(result.is_err());
1055    }
1056
1057    #[test]
1058    fn test_oracle_ddl_add_column() {
1059        let diff_result = SchemaDiff {
1060            added_columns: vec![(
1061                "users".to_string(),
1062                ColumnDef::new("email", "VARCHAR2(255)", true, false, None),
1063            )],
1064            ..Default::default()
1065        };
1066
1067        let ddl = OracleDdlGenerator.generate(&diff_result).unwrap();
1068        assert_eq!(ddl.len(), 1);
1069        assert!(ddl[0].contains("ALTER TABLE users ADD (email VARCHAR2(255))"));
1070    }
1071
1072    #[test]
1073    fn test_mssql_ddl_rename() {
1074        let diff_result = SchemaDiff {
1075            renamed_columns: vec![(
1076                "users".to_string(),
1077                "old_name".to_string(),
1078                "new_name".to_string(),
1079            )],
1080            ..Default::default()
1081        };
1082
1083        let ddl = MssqlDdlGenerator.generate(&diff_result).unwrap();
1084        assert_eq!(ddl.len(), 1);
1085        assert!(ddl[0].contains("EXEC sp_rename 'users.old_name', 'new_name', 'COLUMN'"));
1086    }
1087
1088    #[test]
1089    fn test_destructive_change_detected() {
1090        let diff_result = SchemaDiff {
1091            dropped_columns: vec![("users".to_string(), "legacy".to_string())],
1092            ..Default::default()
1093        };
1094
1095        assert!(diff_result.has_destructive_changes());
1096    }
1097
1098    #[test]
1099    fn test_schema_diff_is_empty() {
1100        let empty = SchemaDiff::default();
1101        assert!(empty.is_empty());
1102
1103        let non_empty = SchemaDiff {
1104            added_columns: vec![(
1105                "users".to_string(),
1106                ColumnDef::new("email", "VARCHAR(255)", true, false, None),
1107            )],
1108            ..Default::default()
1109        };
1110        assert!(!non_empty.is_empty());
1111    }
1112
1113    #[test]
1114    fn test_sync_result() {
1115        let result = SyncResult {
1116            affected_tables: vec!["users".to_string()],
1117            executed_ddl: vec!["ALTER TABLE users ADD COLUMN email VARCHAR(255)".to_string()],
1118        };
1119        assert_eq!(result.affected_tables.len(), 1);
1120        assert_eq!(result.executed_ddl.len(), 1);
1121    }
1122
1123    #[test]
1124    fn test_levenshtein() {
1125        assert_eq!(levenshtein("user_name", "username"), 1);
1126        assert_eq!(levenshtein("name", "title"), 4);
1127        assert_eq!(levenshtein("abc", "abc"), 0);
1128        assert_eq!(levenshtein("", "abc"), 3);
1129        assert_eq!(levenshtein("abc", ""), 3);
1130    }
1131
1132    #[test]
1133    fn test_rename_detection() {
1134        let entity = vec![TableDef::new(
1135            "users",
1136            vec![
1137                ColumnDef::new("id", "BIGINT", false, true, None),
1138                ColumnDef::new("username", "VARCHAR(255)", true, false, None),
1139            ],
1140        )];
1141        let db = vec![TableDef::new(
1142            "users",
1143            vec![
1144                ColumnDef::new("id", "BIGINT", false, true, None),
1145                ColumnDef::new("user_name", "VARCHAR(255)", true, false, None),
1146            ],
1147        )];
1148        let diff_result = diff(&entity, &db);
1149        assert!(!diff_result.renamed_columns.is_empty());
1150        assert_eq!(
1151            diff_result.renamed_columns[0],
1152            (
1153                "users".to_string(),
1154                "user_name".to_string(),
1155                "username".to_string()
1156            )
1157        );
1158        assert!(diff_result.dropped_columns.is_empty());
1159        assert!(diff_result.added_columns.is_empty());
1160    }
1161
1162    #[test]
1163    fn test_rename_no_match_different_type() {
1164        let entity = vec![TableDef::new(
1165            "users",
1166            vec![
1167                ColumnDef::new("id", "BIGINT", false, true, None),
1168                ColumnDef::new("username", "INT", true, false, None),
1169            ],
1170        )];
1171        let db = vec![TableDef::new(
1172            "users",
1173            vec![
1174                ColumnDef::new("id", "BIGINT", false, true, None),
1175                ColumnDef::new("user_name", "VARCHAR(255)", true, false, None),
1176            ],
1177        )];
1178        let diff_result = diff(&entity, &db);
1179        assert!(diff_result.renamed_columns.is_empty());
1180        assert!(!diff_result.dropped_columns.is_empty());
1181        assert!(!diff_result.added_columns.is_empty());
1182    }
1183
1184    #[test]
1185    fn test_rename_no_match_distance_too_large() {
1186        let entity = vec![TableDef::new(
1187            "users",
1188            vec![
1189                ColumnDef::new("id", "BIGINT", false, true, None),
1190                ColumnDef::new("title", "VARCHAR(255)", true, false, None),
1191            ],
1192        )];
1193        let db = vec![TableDef::new(
1194            "users",
1195            vec![
1196                ColumnDef::new("id", "BIGINT", false, true, None),
1197                ColumnDef::new("name", "VARCHAR(255)", true, false, None),
1198            ],
1199        )];
1200        let diff_result = diff(&entity, &db);
1201        assert!(diff_result.renamed_columns.is_empty());
1202    }
1203
1204    #[test]
1205    fn test_confirm_enum() {
1206        assert_eq!(Confirm::Yes, Confirm::Yes);
1207        assert_ne!(Confirm::Yes, Confirm::No);
1208    }
1209
1210    #[test]
1211    fn test_destructive_sync_result() {
1212        let result = DestructiveSyncResult {
1213            executed_ddl: vec!["ALTER TABLE users DROP COLUMN old_col".to_string()],
1214            hooks_called: 1,
1215            audit_entries: 1,
1216        };
1217        assert_eq!(result.executed_ddl.len(), 1);
1218        assert_eq!(result.hooks_called, 1);
1219    }
1220
1221    #[test]
1222    fn test_schema_sync_with_rename_threshold() {
1223        let sync = SchemaSync::new(vec![]).with_rename_threshold(5, 0.5);
1224        assert_eq!(sync.rename_max_distance, 5);
1225        assert!((sync.rename_max_ratio - 0.5).abs() < f64::EPSILON);
1226    }
1227}
sz_orm_core/schema_sync.rs

sz_orm_core/
schema_sync.rs
