dbnexus 0.6.0-rc.4

// Copyright (c) 2026 Kirky.X
// SPDX-License-Identifier: MIT
//! 迁移执行器
//!
//! 负责执行数据库迁移操作

use super::differ::SqlGenerator;
use super::schema::*;
use crate::foundation::DatabaseType;
use crate::foundation::DbError;
use sea_orm::{ConnectionTrait, TransactionTrait};
use std::path::PathBuf;

/// 迁移执行器
///
/// 负责执行数据库迁移操作，内部字段已封装以防止未授权访问
pub struct MigrationExecutor {
    /// 数据库连接
    pub connection: sea_orm::DatabaseConnection,
    /// SQL 生成器
    pub(crate) sql_generator: SqlGenerator,
    /// 迁移历史记录
    pub(crate) history: MigrationHistory,
}

fn build_placeholder_list(backend: sea_orm::DbBackend, count: usize) -> String {
    match backend {
        sea_orm::DbBackend::Postgres => (1..=count)
            .map(|index| format!("${}", index))
            .collect::<Vec<_>>()
            .join(", "),
        _ => std::iter::repeat_n("?", count)
            .collect::<Vec<_>>()
            .join(", "),
    }
}

fn build_migration_insert_sql(backend: sea_orm::DbBackend) -> String {
    match backend {
        sea_orm::DbBackend::Postgres => {
            "INSERT INTO dbnexus_migrations (version, description, applied_at, file_path) VALUES ($1, $2, CAST($3 AS TIMESTAMP), $4)".to_string()
        }
        _ => format!(
            "INSERT INTO dbnexus_migrations (version, description, applied_at, file_path) VALUES ({})",
            build_placeholder_list(backend, 4)
        ),
    }
}

fn sql_escape_single_quotes(s: &str) -> String {
    s.replace('\'', "''")
}

fn format_mysql_applied_at(applied_at: time::OffsetDateTime) -> String {
    let applied_at = applied_at.to_offset(time::UtcOffset::UTC);
    #[allow(deprecated)]
    match time::format_description::parse("[year]-[month]-[day] [hour]:[minute]:[second]") {
        Ok(format) => applied_at
            .format(&format)
            .unwrap_or_else(|_| applied_at.to_string()),
        Err(_) => applied_at.to_string(),
    }
}

fn format_applied_at_for_backend(
    backend: sea_orm::DbBackend,
    applied_at: time::OffsetDateTime,
) -> String {
    match backend {
        sea_orm::DbBackend::MySql => format_mysql_applied_at(applied_at),
        _ => applied_at.to_string(),
    }
}

fn parse_mysql_applied_at(value: &str) -> Option<time::OffsetDateTime> {
    #[allow(deprecated)]
    let format_with_subseconds = time::format_description::parse(
        "[year]-[month]-[day] [hour]:[minute]:[second].[subsecond]",
    )
    .ok();
    if let Some(format) = format_with_subseconds
        && let Ok(dt) = time::PrimitiveDateTime::parse(value, &format)
    {
        return Some(dt.assume_utc());
    }

    #[allow(deprecated)]
    let format_without_subseconds =
        time::format_description::parse("[year]-[month]-[day] [hour]:[minute]:[second]").ok();
    if let Some(format) = format_without_subseconds
        && let Ok(dt) = time::PrimitiveDateTime::parse(value, &format)
    {
        return Some(dt.assume_utc());
    }

    None
}

fn parse_applied_at_for_db(db_type: DatabaseType, value: &str) -> Option<time::OffsetDateTime> {
    match db_type {
        DatabaseType::MySql => parse_mysql_applied_at(value),
        _ => {
            time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339).ok()
        }
    }
}

impl MigrationExecutor {
    /// 创建新的迁移执行器
    pub fn new(connection: sea_orm::DatabaseConnection, db_type: DatabaseType) -> Self {
        Self {
            connection,
            sql_generator: SqlGenerator::new(db_type),
            history: MigrationHistory::new(),
        }
    }

    /// 构建迁移历史插入语句（原始 SQL 字符串）
    ///
    /// 根据数据库后端格式化 `applied_at` 并转义文本字段，返回可直接执行的 INSERT 语句。
    #[deprecated(
        since = "0.2.0",
        note = "Use MigrationExecutor::apply_migration_file_public with MigrationFile::new instead"
    )]
    pub fn build_history_insert_sql_raw(
        &self,
        version: u32,
        description: &str,
        applied_at: time::OffsetDateTime,
        file_path: &str,
    ) -> String {
        let backend = match self.sql_generator.db_type {
            DatabaseType::Postgres => sea_orm::DbBackend::Postgres,
            DatabaseType::MySql => sea_orm::DbBackend::MySql,
            DatabaseType::Sqlite => sea_orm::DbBackend::Sqlite,
            // 不可达：DuckDB 连接在 as_sea_orm 处被拒绝，走不到 MigrationExecutor，仅编译兜底
            DatabaseType::DuckDb => sea_orm::DbBackend::Postgres,
            DatabaseType::Ladybug | DatabaseType::Neo4j => {
                panic!("Graph databases do not participate in relational migrations")
            }
        };

        let applied_at_value = format_applied_at_for_backend(backend, applied_at);
        let desc_esc = sql_escape_single_quotes(description);
        let path_esc = sql_escape_single_quotes(file_path);

        match backend {
            sea_orm::DbBackend::Postgres => format!(
                "INSERT INTO dbnexus_migrations (version, description, applied_at, file_path) VALUES ({}, '{}', CAST('{}' AS TIMESTAMP), '{}')",
                version, desc_esc, applied_at_value, path_esc
            ),
            _ => format!(
                "INSERT INTO dbnexus_migrations (version, description, applied_at, file_path) VALUES ({}, '{}', '{}', '{}')",
                version, desc_esc, applied_at_value, path_esc
            ),
        }
    }

    /// 获取迁移历史的不可变引用
    ///
    /// 返回迁移历史的只读引用，用于查看已应用的迁移
    pub fn history(&self) -> &MigrationHistory {
        &self.history
    }

    /// 读取数据库中的迁移历史
    pub async fn load_history(&mut self) -> Result<(), DbError> {
        // 确保迁移历史表存在
        self.ensure_migration_table_exists().await?;

        let rows = {
            use sea_orm::sea_query::{Alias, Expr, Order, Query};

            let mut query = Query::select();
            query.from(Alias::new("dbnexus_migrations"));
            query.column(Alias::new("version"));
            query.column(Alias::new("description"));
            query.column(Alias::new("file_path"));

            match self.sql_generator.db_type {
                DatabaseType::Postgres => {
                    query.expr_as(Expr::cust("applied_at::text"), Alias::new("applied_at"));
                }
                DatabaseType::MySql => {
                    query.expr_as(
                        Expr::cust("CAST(applied_at AS CHAR)"),
                        Alias::new("applied_at"),
                    );
                }
                DatabaseType::Sqlite => {
                    query.column(Alias::new("applied_at"));
                }
                DatabaseType::DuckDb => {
                    query.column(Alias::new("applied_at"));
                }
                DatabaseType::Ladybug | DatabaseType::Neo4j => {
                    panic!("Graph databases do not participate in relational migrations")
                }
            }

            query.order_by(Alias::new("version"), Order::Asc);

            self.connection
                .query_all(&query)
                .await
                .map_err(DbError::Connection)?
        };

        let mut history = MigrationHistory::new();
        for row in rows {
            // 使用更安全的错误处理方式
            // PostgreSQL 的 sqlx 驱动严格要求类型匹配：INTEGER 列必须用 i32 读取
            let version: Result<i32, _> = row.try_get("", "version");
            let version_val = match version {
                Ok(v) => v,
                Err(_e) => {
                    continue;
                }
            };
            let Ok(version) = u32::try_from(version_val) else {
                continue;
            };

            let description: String = row.try_get("", "description").unwrap_or_default();

            let applied_at_str: String = row.try_get("", "applied_at").unwrap_or_default();
            let applied_at = if applied_at_str.is_empty() {
                time::OffsetDateTime::now_utc()
            } else {
                match parse_applied_at_for_db(self.sql_generator.db_type, &applied_at_str) {
                    Some(dt) => dt,
                    None => time::OffsetDateTime::now_utc(),
                }
            };

            let file_path: String = row.try_get("", "file_path").unwrap_or_default();

            history.add_migration(MigrationVersion {
                version,
                description,
                applied_at,
                file_path,
            });
        }
        self.history = history;

        Ok(())
    }

    /// 确保迁移历史表存在
    ///
    /// PostgreSQL 已知问题：并发 `CREATE TABLE IF NOT EXISTS` 可能因 `pg_type` 类型注册冲突
    /// 失败，表现为两类错误：
    /// - SQLSTATE 23505（`pg_type_typname_nsp_index`）：并发类型注册唯一约束冲突；
    /// - SQLSTATE 42710（`type "..." already exists`）：另一会话已提交同名复合类型。
    ///
    /// `IF NOT EXISTS` 仅跳过"关系已存在"的情况，不保护并发类型注册，也不覆盖"类型已存在"。
    /// 由于 `CREATE TABLE` 与其行类型原子绑定（类型存在即表存在），上述冲突均可安全视为
    /// "表已由并发会话创建"。此方法在捕获这两类错误后等待 50ms 再重试，确保另一会话的
    /// `CREATE TABLE` 已提交，使重试的 `IF NOT EXISTS` 成为真正的 no-op。
    async fn ensure_migration_table_exists(&self) -> Result<(), DbError> {
        let create_table_sql = match self.sql_generator.db_type {
            DatabaseType::Postgres => {
                "CREATE TABLE IF NOT EXISTS dbnexus_migrations (
                    version INTEGER PRIMARY KEY,
                    description TEXT NOT NULL,
                    applied_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
                    file_path TEXT
                );"
            }
            DatabaseType::MySql => {
                "CREATE TABLE IF NOT EXISTS dbnexus_migrations (
                    version INT PRIMARY KEY,
                    description TEXT NOT NULL,
                    applied_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
                    file_path TEXT
                );"
            }
            DatabaseType::Sqlite => {
                "CREATE TABLE IF NOT EXISTS dbnexus_migrations (
                    version INTEGER PRIMARY KEY,
                    description TEXT NOT NULL,
                    applied_at TEXT NOT NULL DEFAULT (datetime('now')),
                    file_path TEXT
                );"
            }
            DatabaseType::DuckDb => {
                "CREATE TABLE IF NOT EXISTS dbnexus_migrations (
                    version INTEGER PRIMARY KEY,
                    description TEXT NOT NULL,
                    applied_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
                    file_path TEXT
                );"
            }
            DatabaseType::Ladybug | DatabaseType::Neo4j => {
                panic!("Graph databases do not participate in relational migrations")
            }
        };

        match self.connection.execute_unprepared(create_table_sql).await {
            Ok(_) => Ok(()),
            Err(e) => {
                let err_str = e.to_string();
                // 并发 CREATE TABLE 或历史残留触发两类"已存在"冲突：
                //   - pg_type_typname_nsp_index（SQLSTATE 23505）：并发类型注册唯一约束冲突
                //   - "type ... already exists"（SQLSTATE 42710）：另一会话已提交同名复合类型
                let is_creation_conflict = err_str.contains("pg_type_typname_nsp_index")
                    || err_str.contains("already exists");
                if is_creation_conflict {
                    // 等待并发 CREATE TABLE 提交后重试，使 IF NOT EXISTS 成为 no-op
                    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
                    match self.connection.execute_unprepared(create_table_sql).await {
                        Ok(_) => Ok(()),
                        Err(e2) => {
                            let err_str2 = e2.to_string();
                            // 重试后仍为 pg_type 冲突或表已存在，视为成功
                            if err_str2.contains("pg_type_typname_nsp_index")
                                || err_str2.contains("already exists")
                            {
                                Ok(())
                            } else {
                                Err(DbError::Connection(e2))
                            }
                        }
                    }
                } else {
                    Err(DbError::Connection(e))
                }
            }
        }
    }

    /// 应用单个迁移
    pub async fn apply_migration(&mut self, migration: &Migration) -> Result<(), DbError> {
        // 确保迁移历史表存在
        self.ensure_migration_table_exists().await?;

        // 生成迁移 SQL
        let sql = self.sql_generator.generate_migration_sql(migration)?;

        // 开始事务
        let txn = self.connection.begin().await.map_err(DbError::Connection)?;

        // 执行迁移 SQL
        if !sql.is_empty() {
            txn.execute_unprepared(&sql)
                .await
                .map_err(DbError::Connection)?;
        }

        // 记录迁移历史
        let version_record = MigrationVersion {
            version: migration.version,
            description: migration.description.clone(),
            applied_at: migration
                .timestamp
                .unwrap_or_else(time::OffsetDateTime::now_utc),
            file_path: format!("migration_v{}.sql", migration.version),
        };

        // 插入到迁移历史表（使用参数化查询防止 SQL 注入）
        // 使用 Statement::from_sql_and_values 进行参数化查询
        let backend = match self.sql_generator.db_type {
            DatabaseType::Postgres => sea_orm::DbBackend::Postgres,
            DatabaseType::MySql => sea_orm::DbBackend::MySql,
            DatabaseType::Sqlite => sea_orm::DbBackend::Sqlite,
            // 不可达：DuckDB 连接在 as_sea_orm 处被拒绝，走不到 MigrationExecutor，仅编译兜底
            DatabaseType::DuckDb => sea_orm::DbBackend::Postgres,
            DatabaseType::Ladybug | DatabaseType::Neo4j => {
                panic!("Graph databases do not participate in relational migrations")
            }
        };

        let insert_sql = build_migration_insert_sql(backend);
        let applied_at_value = format_applied_at_for_backend(backend, version_record.applied_at);

        let stmt = sea_orm::Statement::from_sql_and_values(
            backend,
            insert_sql.to_string(),
            vec![
                migration.version.into(),
                migration.description.clone().into(),
                applied_at_value.into(),
                version_record.file_path.clone().into(),
            ],
        );

        txn.execute_raw(stmt).await.map_err(DbError::Connection)?;
        // 提交事务
        txn.commit().await.map_err(DbError::Connection)?;

        self.history.add_migration(version_record);

        Ok(())
    }

    /// 获取待应用的迁移
    pub async fn get_pending_migrations<'a>(
        &'a mut self,
        all_migrations: &'a [Migration],
    ) -> Vec<&'a Migration> {
        // 重新加载历史记录以获取最新状态
        if self.load_history().await.is_ok() {
            self.history.get_pending_migrations(all_migrations)
        } else {
            // 如果加载失败，返回所有迁移（保守处理）
            all_migrations.iter().collect()
        }
    }

    /// 获取所有迁移的版本号
    pub fn get_all_versions(&self) -> Vec<u32> {
        self.history
            .applied_migrations
            .iter()
            .map(|m| m.version)
            .collect()
    }

    /// 获取最新应用的迁移
    pub fn get_latest_migration(&self) -> Option<&MigrationVersion> {
        self.history.applied_migrations.last()
    }

    /// 检查是否所有迁移都已应用
    pub fn is_fully_migrated(&self, total_migrations: usize) -> bool {
        self.history.applied_migrations.len() == total_migrations
    }
}

/// 迁移文件信息
///
/// 存储迁移文件的基本信息，用于扫描和管理迁移文件
#[derive(Debug, Clone)]
pub struct MigrationFile {
    /// 版本号
    pub(crate) version: u32,
    /// 描述
    pub(crate) description: String,
    /// 文件路径
    pub(crate) file_path: PathBuf,
    /// 文件内容
    pub(crate) content: String,
}

impl MigrationFile {
    /// 创建新的迁移文件信息
    ///
    /// # Arguments
    ///
    /// * `version` - 迁移版本号
    /// * `description` - 迁移描述
    /// * `file_path` - 迁移文件路径
    /// * `content` - 迁移文件内容（SQL 语句）
    pub fn new(version: u32, description: String, file_path: PathBuf, content: String) -> Self {
        Self {
            version,
            description,
            file_path,
            content,
        }
    }

    /// 获取迁移版本号
    pub fn version(&self) -> u32 {
        self.version
    }

    /// 获取迁移描述
    pub fn description(&self) -> &str {
        &self.description
    }

    /// 获取文件路径
    pub fn file_path(&self) -> &PathBuf {
        &self.file_path
    }

    /// 获取文件内容
    pub fn content(&self) -> &str {
        &self.content
    }
}

/// UP / DOWN 标记集合（大小写不敏感）
///
/// 供标记行查找（`find_marker_line`）与迁移文件校验（`MigrationFileParser`）共用，
/// 保证"验证"与"提取"对标记的判定口径一致，故放在 `auto-migrate` 门控之外。
const UP_MARKERS: [&str; 6] = ["-- UP:", "-- up:", "-- UP", "-- up", "UP:", "UP"];
const DOWN_MARKERS: [&str; 6] = [
    "-- DOWN:", "-- down:", "-- DOWN", "-- down", "DOWN:", "DOWN",
];

/// 判断一行（trim 后）是否以某个标记开头（大小写不敏感）
///
/// - 带冒号（`-- UP:`、`UP:`）与注释形（`-- UP`）标记自带边界，做纯前缀匹配；
/// - 裸词标记（`UP` / `DOWN`）额外要求标记后是行尾或非单词字符，
///   避免把 `updated_at`、`download_url` 这类以 up/down 开头的标识符误判为标记行。
fn line_matches_marker(line: &str, markers: &[&str]) -> bool {
    let trimmed = line.trim();
    markers.iter().any(|marker| {
        // get 保证按字符边界取前缀，避免多字节字符下的切片 panic
        let Some(prefix) = trimmed.get(..marker.len()) else {
            return false;
        };
        if !prefix.eq_ignore_ascii_case(marker) {
            return false;
        }
        if !marker.starts_with("--") && !marker.ends_with(':') {
            // 裸词标记要求词边界：标记后不能紧跟字母/数字/下划线
            return trimmed[marker.len()..]
                .chars()
                .next()
                .is_none_or(|next| !next.is_alphanumeric() && next != '_');
        }
        true
    })
}

/// 在内容中查找标记所在的行
///
/// 逐行扫描：对每行 trim 后做大小写不敏感的"以 marker 开头"判断，命中即返回
/// 该行边界 `(行起始偏移, 行结束偏移)`（行结束偏移包含换行符）。
///
/// 相比旧的全文子串搜索：
/// - `-- Down:` 等混合大小写标记可被识别（旧实现按字节精确匹配会漏判）；
/// - 裸标记 `DOWN` 只在行首（trim 后）匹配，不再命中行中间的单词。
fn find_marker_line(content: &str, markers: &[&str]) -> Option<(usize, usize)> {
    let mut rest = content;
    let mut line_start = 0usize;
    while let Some(nl) = rest.find('\n') {
        if line_matches_marker(&rest[..nl], markers) {
            return Some((line_start, line_start + nl + 1));
        }
        line_start += nl + 1;
        rest = &rest[nl + 1..];
    }
    // 末行（不以换行符结尾）
    if line_matches_marker(rest, markers) {
        return Some((line_start, content.len()));
    }
    None
}

/// 自动迁移执行器
#[cfg(feature = "auto-migrate")]
impl MigrationExecutor {
    /// 扫描指定目录中的迁移文件
    ///
    /// 迁移文件命名格式: `{version}_{description}.sql`
    ///
    /// # Arguments
    ///
    /// * `dir` - 迁移文件目录路径
    ///
    /// # Returns
    ///
    /// 扫描到的迁移文件列表（按版本号排序）
    pub fn scan_migrations(&self, dir: &std::path::Path) -> Result<Vec<MigrationFile>, DbError> {
        let mut migrations = Vec::new();

        if !dir.exists() {
            return Ok(migrations);
        }

        let entries = std::fs::read_dir(dir)
            .map_err(|e| DbError::Config(format!("Failed to read migration directory: {}", e)))?;

        for entry in entries {
            let entry = entry
                .map_err(|e| DbError::Config(format!("Failed to read migration entry: {}", e)))?;
            let path = entry.path();

            if path.is_file()
                && path.extension().map(|e| e == "sql").unwrap_or(false)
                && let Some(filename) = path.file_name().and_then(|n| n.to_str())
                && let Some((version, description)) = Self::parse_filename(filename)
            {
                let content = std::fs::read_to_string(&path).map_err(|e| {
                    DbError::Config(format!("Failed to read migration file: {}", e))
                })?;

                migrations.push(MigrationFile {
                    version,
                    description,
                    file_path: path,
                    content,
                });
            }
        }

        // 按版本号排序
        migrations.sort_by_key(|m| m.version);

        Ok(migrations)
    }

    /// 解析迁移文件名
    pub(crate) fn parse_filename(filename: &str) -> Option<(u32, String)> {
        let parts: Vec<&str> = filename.split('_').collect();
        if parts.is_empty() {
            return None;
        }

        let version = parts[0].parse::<u32>().ok()?;
        let description = parts[1..].join("_").replace(".sql", "");

        Some((version, description))
    }

    /// 运行所有待应用的迁移
    ///
    /// # Arguments
    ///
    /// * `dir` - 迁移文件目录路径
    ///
    /// # Returns
    ///
    /// 成功应用的迁移数量
    pub async fn run_migrations(&mut self, dir: &std::path::Path) -> Result<u32, DbError> {
        // 扫描迁移文件
        let migration_files = self.scan_migrations(dir)?;

        // 批量加载所有已应用的版本（消除 N+1 查询）
        let applied_versions = self.load_applied_versions().await?;

        let pending: Vec<_> = migration_files
            .into_iter()
            .filter(|m| !applied_versions.contains(&m.version))
            .collect();

        if pending.is_empty() {
            return Ok(0);
        }

        // 应用迁移
        let mut applied_count = 0;
        for migration_file in &pending {
            match self.apply_migration_file(migration_file).await {
                Ok(_) => {
                    applied_count += 1;
                }
                Err(e) => {
                    return Err(e);
                }
            }
        }

        Ok(applied_count)
    }

    /// 批量加载所有已应用的迁移版本（消除 N+1 查询）
    ///
    /// 在 `run_migrations` 开始时调用，一次性加载所有已应用的版本。
    /// 避免对每个迁移文件单独调用 `is_migration_applied` 导致的 N+1 查询问题。
    async fn load_applied_versions(&self) -> Result<std::collections::HashSet<u32>, DbError> {
        // 先确保迁移历史表存在
        self.ensure_migration_table_exists().await?;

        use sea_orm::sea_query::{Alias, Query};

        let mut query = Query::select();
        query.column(Alias::new("version"));
        query.from(Alias::new("dbnexus_migrations"));

        let rows = self
            .connection
            .query_all(&query)
            .await
            .map_err(DbError::Connection)?;

        let mut applied_versions = std::collections::HashSet::new();
        for row in rows {
            // PostgreSQL 的 sqlx 驱动严格要求类型匹配：INTEGER 列必须用 i32 读取
            if let Ok(version) = row.try_get::<i32>("", "version") {
                applied_versions.insert(version as u32);
            }
        }

        Ok(applied_versions)
    }

    /// 应用单个迁移文件
    async fn apply_migration_file(
        &mut self,
        migration_file: &MigrationFile,
    ) -> Result<(), DbError> {
        // 解析迁移文件内容
        let sql = Self::extract_up_sql(&migration_file.content);

        // 开始事务
        let txn = self.connection.begin().await.map_err(DbError::Connection)?;

        // 执行迁移 SQL
        if !sql.is_empty() {
            txn.execute_unprepared(sql)
                .await
                .map_err(DbError::Connection)?;
        }

        // 记录迁移历史（使用参数化查询防止 SQL 注入）
        let applied_at = time::OffsetDateTime::now_utc();

        let backend = match self.sql_generator.db_type {
            DatabaseType::Postgres => sea_orm::DbBackend::Postgres,
            DatabaseType::MySql => sea_orm::DbBackend::MySql,
            DatabaseType::Sqlite => sea_orm::DbBackend::Sqlite,
            // 不可达：DuckDB 连接在 as_sea_orm 处被拒绝，走不到 MigrationExecutor，仅编译兜底
            DatabaseType::DuckDb => sea_orm::DbBackend::Postgres,
            DatabaseType::Ladybug | DatabaseType::Neo4j => {
                panic!("Graph databases do not participate in relational migrations")
            }
        };
        let insert_sql = build_migration_insert_sql(backend);
        let applied_at_value = format_applied_at_for_backend(backend, applied_at);
        let stmt = sea_orm::Statement::from_sql_and_values(
            backend,
            insert_sql.to_string(),
            vec![
                migration_file.version.into(),
                migration_file.description.clone().into(),
                applied_at_value.into(),
                migration_file.file_path.to_string_lossy().into(),
            ],
        );

        txn.execute_raw(stmt).await.map_err(DbError::Connection)?;

        // 提交事务
        txn.commit().await.map_err(DbError::Connection)?;

        // 添加到历史记录
        self.history.add_migration(MigrationVersion {
            version: migration_file.version,
            description: migration_file.description.clone(),
            applied_at,
            file_path: migration_file.file_path.to_string_lossy().to_string(),
        });

        Ok(())
    }

    #[allow(missing_docs)]
    pub async fn apply_migration_file_public(
        &mut self,
        migration_file: &MigrationFile,
    ) -> Result<(), DbError> {
        self.apply_migration_file(migration_file).await
    }

    /// 从迁移文件中提取 UP SQL
    fn extract_up_sql(content: &str) -> &str {
        let up_marker = find_marker_line(content, &UP_MARKERS);
        let down_marker = find_marker_line(content, &DOWN_MARKERS);

        match (up_marker, down_marker) {
            (Some((_, up_end)), Some((down_start, _))) if down_start > up_end => {
                &content[up_end..down_start]
            }
            (Some((_, up_end)), _) => &content[up_end..],
            (None, Some((down_start, _))) => &content[..down_start],
            (None, None) => content,
        }
        .trim()
    }

    /// 从迁移文件中提取 DOWN SQL
    ///
    /// 与 `extract_up_sql` 对称：存在 DOWN 标记时返回标记行之后的内容（可能为空）；
    /// 返回 `None` 表示迁移文件没有 DOWN 标记（即无可回滚部分）。
    ///
    /// 注意：仅含注释/空白的 DOWN 段同样按原文返回（`Some`，语义不变）；
    /// 拒绝"假回滚"由 `rollback_version` 负责（剥离注释后无可执行语句即报错）。
    pub fn extract_down_sql(content: &str) -> Option<&str> {
        find_marker_line(content, &DOWN_MARKERS).map(|(_, down_end)| content[down_end..].trim())
    }

    /// 判断 DOWN 段剥离 `--` 行注释与空白后是否仍含可执行内容
    ///
    /// 逐行检查：跳过空行与以 `--` 开头的注释行，其余行视为可执行内容。
    /// 这是保守的语法级判断（不校验 SQL 语句本身的合法性），块注释 `/* */` 不在处理范围。
    fn down_has_executable_sql(down_sql: &str) -> bool {
        down_sql
            .lines()
            .map(str::trim)
            .any(|line| !line.is_empty() && !line.starts_with("--"))
    }

    /// 回滚单个迁移文件
    ///
    /// 在同一事务内执行迁移文件的 DOWN SQL，成功后删除 `dbnexus_migrations`
    /// 表中对应版本的历史行；DOWN SQL 执行失败时整体回滚，历史记录保持不变。
    ///
    /// # Arguments
    ///
    /// * `version` - 待回滚的迁移版本号（与历史表中的 version 一致）
    /// * `migration_file` - 迁移文件（用于提取 DOWN SQL）
    ///
    /// # Errors
    ///
    /// - 迁移文件没有 DOWN 标记时返回 `DbError::Migration`（不删除历史记录）；
    /// - DOWN 标记存在但剥离 `--` 行注释与空白后不含可执行语句（如模板占位未填写）
    ///   时返回 `DbError::Migration`（不删除历史记录，避免"假回滚"）；
    /// - DOWN SQL 或删除历史行执行失败时返回 `DbError::Connection`（事务回滚）。
    pub async fn rollback_version(
        &self,
        version: u32,
        migration_file: &MigrationFile,
    ) -> Result<(), DbError> {
        // 无 DOWN 段的迁移无法回滚，提前返回明确错误（不删除历史记录）
        let down_sql = Self::extract_down_sql(&migration_file.content).ok_or_else(|| {
            DbError::Migration(format!(
                "迁移 v{} ({}) 无可回滚的 DOWN 部分",
                version, migration_file.description
            ))
        })?;

        // 模板生成的迁移常带未填写的 DOWN 段（仅 `--` 注释/空白）。execute_unprepared
        // 执行纯注释 SQL 会"成功"（0 行受影响），若据此删除历史行会造成"假回滚"：
        // DOWN 实际未执行，版本却被标记为已回滚。因此剥离注释与空白后必须仍存在
        // 可执行内容，否则拒绝回滚（历史行保留）。
        if !Self::down_has_executable_sql(down_sql) {
            return Err(DbError::Migration(
                "DOWN 段存在但不含可执行语句，拒绝回滚以避免假回滚".to_string(),
            ));
        }

        let backend = match self.sql_generator.db_type {
            DatabaseType::Postgres => sea_orm::DbBackend::Postgres,
            DatabaseType::MySql => sea_orm::DbBackend::MySql,
            DatabaseType::Sqlite => sea_orm::DbBackend::Sqlite,
            // DuckDB 连接在 as_sea_orm 处被拒绝，走不到 MigrationExecutor；
            // 显式报错，避免生成误导性的 Postgres 占位符 SQL
            DatabaseType::DuckDb => {
                return Err(DbError::Config(
                    "DuckDB connections do not support SeaORM-based migration rollback".to_string(),
                ));
            }
            DatabaseType::Ladybug | DatabaseType::Neo4j => {
                return Err(DbError::Config(
                    "Graph databases do not participate in relational migrations".to_string(),
                ));
            }
        };

        // 开始事务：DOWN SQL 与历史行删除原子提交
        let txn = self.connection.begin().await.map_err(DbError::Connection)?;

        if !down_sql.is_empty() {
            txn.execute_unprepared(down_sql)
                .await
                .map_err(DbError::Connection)?;
        }

        // 删除迁移历史记录（使用参数化查询防止 SQL 注入）
        let delete_sql = format!(
            "DELETE FROM dbnexus_migrations WHERE version = {}",
            build_placeholder_list(backend, 1)
        );
        let stmt =
            sea_orm::Statement::from_sql_and_values(backend, delete_sql, vec![version.into()]);

        txn.execute_raw(stmt).await.map_err(DbError::Connection)?;

        // 提交事务
        txn.commit().await.map_err(DbError::Connection)?;

        Ok(())
    }
}

/// 迁移文件解析器
pub struct MigrationFileParser;

impl MigrationFileParser {
    /// 解析迁移文件内容
    pub fn parse_migration_file(content: &str) -> Result<(String, String), String> {
        // 提取迁移描述
        let description = Self::extract_description(content);

        // 验证SQL语法（简单验证）
        Self::validate_sql_syntax(content)?;

        Ok((description, content.to_string()))
    }

    /// 从迁移文件中提取描述
    fn extract_description(content: &str) -> String {
        // 尝试从注释中提取描述
        for line in content.lines() {
            let trimmed = line.trim();
            if let Some(stripped) = trimmed.strip_prefix("-- Migration:") {
                return stripped.trim().to_string();
            } else if trimmed.starts_with("/*") || trimmed.starts_with("--") {
                continue; // 跳过其他注释行
            } else {
                break; // 遇到非注释行则停止
            }
        }
        "Migration".to_string()
    }

    /// 验证SQL语法（基本验证）
    fn validate_sql_syntax(content: &str) -> Result<(), String> {
        // UP/DOWN 标记检测复用 find_marker_line，与 extract_up_sql / extract_down_sql
        // 的提取口径保持一致（按行、大小写不敏感前缀匹配），
        // 避免"验证认定有标记、提取却找不到"的口径分裂
        let has_up = find_marker_line(content, &UP_MARKERS).is_some();
        let has_down = find_marker_line(content, &DOWN_MARKERS).is_some();

        if !has_up && !has_down {
            // 如果没有UP/DOWN标记，只要包含SQL语句即可
            let sql_statements = ["CREATE", "ALTER", "DROP", "INSERT", "UPDATE", "DELETE"];
            let contains_sql = sql_statements
                .iter()
                .any(|stmt| content.to_uppercase().contains(stmt));

            if !contains_sql {
                return Err(
                    "Migration file does not contain recognizable SQL statements".to_string(),
                );
            }
        }

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    // TableChange 仅被下方 sqlite 门控的迁移落地测试使用
    #[cfg(all(feature = "sqlite", feature = "runtime-tokio-rustls"))]
    use crate::domain::TableChange;

    // =====================================================================
    // build_placeholder_list
    // =====================================================================

    #[test]
    fn test_build_placeholder_list_postgres() {
        let result = build_placeholder_list(sea_orm::DbBackend::Postgres, 3);
        assert_eq!(result, "$1, $2, $3");
    }

    #[test]
    fn test_build_placeholder_list_postgres_single() {
        let result = build_placeholder_list(sea_orm::DbBackend::Postgres, 1);
        assert_eq!(result, "$1");
    }

    #[test]
    fn test_build_placeholder_list_sqlite() {
        let result = build_placeholder_list(sea_orm::DbBackend::Sqlite, 4);
        assert_eq!(result, "?, ?, ?, ?");
    }

    #[test]
    fn test_build_placeholder_list_mysql() {
        let result = build_placeholder_list(sea_orm::DbBackend::MySql, 2);
        assert_eq!(result, "?, ?");
    }

    #[test]
    fn test_build_placeholder_list_zero() {
        assert_eq!(build_placeholder_list(sea_orm::DbBackend::Postgres, 0), "");
        assert_eq!(build_placeholder_list(sea_orm::DbBackend::Sqlite, 0), "");
    }

    // =====================================================================
    // build_migration_insert_sql
    // =====================================================================

    #[test]
    fn test_build_migration_insert_sql_postgres() {
        let sql = build_migration_insert_sql(sea_orm::DbBackend::Postgres);
        assert!(sql.contains("INSERT INTO dbnexus_migrations"));
        assert!(sql.contains("$1, $2, CAST($3 AS TIMESTAMP), $4"));
    }

    #[test]
    fn test_build_migration_insert_sql_sqlite() {
        let sql = build_migration_insert_sql(sea_orm::DbBackend::Sqlite);
        assert!(sql.contains("INSERT INTO dbnexus_migrations"));
        assert!(sql.contains("?, ?, ?, ?"));
    }

    #[test]
    fn test_build_migration_insert_sql_mysql() {
        let sql = build_migration_insert_sql(sea_orm::DbBackend::MySql);
        assert!(sql.contains("INSERT INTO dbnexus_migrations"));
        assert!(sql.contains("?, ?, ?, ?"));
    }

    // =====================================================================
    // sql_escape_single_quotes
    // =====================================================================

    #[test]
    fn test_sql_escape_single_quotes_no_quotes() {
        assert_eq!(sql_escape_single_quotes("hello world"), "hello world");
    }

    #[test]
    fn test_sql_escape_single_quotes_single_quote() {
        assert_eq!(sql_escape_single_quotes("it's"), "it''s");
    }

    #[test]
    fn test_sql_escape_single_quotes_multiple_quotes() {
        assert_eq!(sql_escape_single_quotes("'a'b'"), "''a''b''");
    }

    #[test]
    fn test_sql_escape_single_quotes_empty() {
        assert_eq!(sql_escape_single_quotes(""), "");
    }

    // =====================================================================
    // format_mysql_applied_at
    // =====================================================================

    #[test]
    fn test_format_mysql_applied_at() {
        let dt = time::Date::from_calendar_date(2026, time::Month::June, 25)
            .unwrap()
            .with_hms(12, 30, 45)
            .unwrap()
            .assume_utc();
        let result = format_mysql_applied_at(dt);
        assert_eq!(result, "2026-06-25 12:30:45");
    }

    // =====================================================================
    // format_applied_at_for_backend
    // =====================================================================

    #[test]
    fn test_format_applied_at_for_backend_mysql() {
        let dt = time::Date::from_calendar_date(2026, time::Month::January, 1)
            .unwrap()
            .with_hms(0, 0, 0)
            .unwrap()
            .assume_utc();
        let result = format_applied_at_for_backend(sea_orm::DbBackend::MySql, dt);
        assert_eq!(result, "2026-01-01 00:00:00");
    }

    #[test]
    fn test_format_applied_at_for_backend_non_mysql() {
        let dt = time::Date::from_calendar_date(2026, time::Month::January, 1)
            .unwrap()
            .with_hms(0, 0, 0)
            .unwrap()
            .assume_utc();
        let result = format_applied_at_for_backend(sea_orm::DbBackend::Sqlite, dt);
        // 非 MySQL 使用 OffsetDateTime::to_string()（Rfc3339 格式）
        assert!(result.contains("2026-01-01"));
    }

    // =====================================================================
    // parse_mysql_applied_at
    // =====================================================================

    #[test]
    fn test_parse_mysql_applied_at_without_subseconds() {
        let result = parse_mysql_applied_at("2026-06-25 12:30:45");
        assert!(result.is_some());
        let dt = result.unwrap();
        assert_eq!(dt.year(), 2026);
        assert_eq!(dt.month(), time::Month::June);
        assert_eq!(dt.day(), 25);
    }

    #[test]
    fn test_parse_mysql_applied_at_with_subseconds() {
        let result = parse_mysql_applied_at("2026-06-25 12:30:45.123");
        assert!(result.is_some());
    }

    #[test]
    fn test_parse_mysql_applied_at_invalid() {
        assert!(parse_mysql_applied_at("not a date").is_none());
        assert!(parse_mysql_applied_at("").is_none());
    }

    // =====================================================================
    // parse_applied_at_for_db
    // =====================================================================

    #[test]
    fn test_parse_applied_at_for_db_mysql() {
        let result = parse_applied_at_for_db(DatabaseType::MySql, "2026-06-25 12:30:45");
        assert!(result.is_some());
    }

    #[test]
    fn test_parse_applied_at_for_db_sqlite_rfc3339() {
        let result = parse_applied_at_for_db(DatabaseType::Sqlite, "2026-06-25T12:30:45Z");
        assert!(result.is_some());
    }

    #[test]
    fn test_parse_applied_at_for_db_invalid() {
        assert!(parse_applied_at_for_db(DatabaseType::Sqlite, "invalid").is_none());
        assert!(parse_applied_at_for_db(DatabaseType::MySql, "invalid").is_none());
    }

    // =====================================================================
    // MigrationFile
    // =====================================================================

    #[test]
    fn test_migration_file_new_and_getters() {
        let file = MigrationFile::new(
            1,
            "create_users".to_string(),
            PathBuf::from("/migrations/001_create_users.sql"),
            "CREATE TABLE users (id INTEGER);".to_string(),
        );
        assert_eq!(file.version(), 1);
        assert_eq!(file.description(), "create_users");
        assert_eq!(
            file.file_path(),
            &PathBuf::from("/migrations/001_create_users.sql")
        );
        assert_eq!(file.content(), "CREATE TABLE users (id INTEGER);");
    }

    // =====================================================================
    // MigrationFileParser
    // =====================================================================

    #[test]
    fn test_migration_file_parser_valid_with_up_down() {
        let content = "-- Migration: create users table\n-- UP:\nCREATE TABLE users (id INTEGER);\n-- DOWN:\nDROP TABLE users;\n";
        let result = MigrationFileParser::parse_migration_file(content);
        assert!(result.is_ok());
        let (desc, _) = result.unwrap();
        assert_eq!(desc, "create users table");
    }

    #[test]
    fn test_migration_file_parser_valid_with_create() {
        let content = "CREATE TABLE users (id INTEGER);";
        let result = MigrationFileParser::parse_migration_file(content);
        assert!(result.is_ok());
        let (desc, _) = result.unwrap();
        // 无 "-- Migration:" 注释时返回默认描述
        assert_eq!(desc, "Migration");
    }

    #[test]
    fn test_migration_file_parser_invalid_no_sql() {
        let content = "-- just a comment\n-- nothing else\n";
        let result = MigrationFileParser::parse_migration_file(content);
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(err.contains("does not contain recognizable SQL statements"));
    }

    #[test]
    fn test_migration_file_parser_extract_description_with_marker() {
        let content =
            "-- Migration: add index on users\nCREATE INDEX idx_users_email ON users(email);";
        let result = MigrationFileParser::parse_migration_file(content);
        assert!(result.is_ok());
        let (desc, _) = result.unwrap();
        assert_eq!(desc, "add index on users");
    }

    #[test]
    fn test_migration_file_parser_extract_description_default() {
        let content = "CREATE TABLE t (id INTEGER);";
        let result = MigrationFileParser::parse_migration_file(content);
        assert!(result.is_ok());
        let (desc, _) = result.unwrap();
        assert_eq!(desc, "Migration");
    }

    #[test]
    fn test_migration_file_parser_validate_sql_with_alter() {
        let content = "ALTER TABLE users ADD COLUMN name TEXT;";
        let result = MigrationFileParser::parse_migration_file(content);
        assert!(result.is_ok());
    }

    #[test]
    fn test_migration_file_parser_validate_sql_with_drop() {
        let content = "DROP TABLE old_table;";
        let result = MigrationFileParser::parse_migration_file(content);
        assert!(result.is_ok());
    }

    // =====================================================================
    // MigrationExecutor - non-database methods (需要 sqlite 以构造执行器)
    // =====================================================================

    #[cfg(all(feature = "sqlite", feature = "runtime-tokio-rustls"))]
    #[tokio::test]
    async fn test_migration_executor_get_all_versions_empty() {
        let executor = create_sqlite_executor().await;
        assert!(executor.get_all_versions().is_empty());
    }

    #[cfg(all(feature = "sqlite", feature = "runtime-tokio-rustls"))]
    #[tokio::test]
    async fn test_migration_executor_get_latest_migration_empty() {
        let executor = create_sqlite_executor().await;
        assert!(executor.get_latest_migration().is_none());
    }

    #[cfg(all(feature = "sqlite", feature = "runtime-tokio-rustls"))]
    #[tokio::test]
    async fn test_migration_executor_is_fully_migrated_empty() {
        let executor = create_sqlite_executor().await;
        // 0 applied == 0 total → fully migrated
        assert!(executor.is_fully_migrated(0));
        assert!(!executor.is_fully_migrated(1));
    }

    #[cfg(all(feature = "sqlite", feature = "runtime-tokio-rustls"))]
    #[tokio::test]
    async fn test_migration_executor_history_empty() {
        let executor = create_sqlite_executor().await;
        assert!(executor.history().applied_migrations.is_empty());
    }

    #[cfg(all(feature = "sqlite", feature = "runtime-tokio-rustls"))]
    #[tokio::test]
    async fn test_migration_executor_build_history_insert_sql_raw_sqlite() {
        let executor = create_sqlite_executor().await;
        let dt = time::OffsetDateTime::now_utc();
        #[allow(deprecated)]
        let sql =
            executor.build_history_insert_sql_raw(1, "test migration", dt, "/path/to/file.sql");
        assert!(sql.contains("INSERT INTO dbnexus_migrations"));
        assert!(sql.contains("1"));
        assert!(sql.contains("test migration"));
        assert!(sql.contains("/path/to/file.sql"));
    }

    #[cfg(all(feature = "sqlite", feature = "runtime-tokio-rustls"))]
    #[tokio::test]
    async fn test_migration_executor_build_history_insert_sql_raw_escapes_quotes() {
        let executor = create_sqlite_executor().await;
        let dt = time::OffsetDateTime::now_utc();
        #[allow(deprecated)]
        let sql =
            executor.build_history_insert_sql_raw(1, "it's a 'test'", dt, "/path/to/file.sql");
        // 单引号应被转义为 ''
        assert!(sql.contains("it''s a ''test''"));
    }

    // =====================================================================
    // MigrationExecutor - auto-migrate feature methods
    // =====================================================================

    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_parse_filename_valid() {
        let result = MigrationExecutor::parse_filename("001_create_users.sql");
        assert_eq!(result, Some((1, "create_users".to_string())));
    }

    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_parse_filename_multi_part() {
        let result = MigrationExecutor::parse_filename("002_add_index_to_users_table.sql");
        assert_eq!(result, Some((2, "add_index_to_users_table".to_string())));
    }

    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_parse_filename_invalid_version() {
        let result = MigrationExecutor::parse_filename("abc_create_users.sql");
        assert!(result.is_none());
    }

    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_parse_filename_no_underscore() {
        // "123.sql" split('_') = ["123.sql"]，parts[0]="123.sql" 无法 parse::<u32>()
        // 因为 "123.sql" 不是纯数字
        let result = MigrationExecutor::parse_filename("123.sql");
        assert!(result.is_none());
    }

    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_extract_up_sql_with_up_and_down() {
        let content = "-- UP:\nCREATE TABLE users (id INTEGER);\n-- DOWN:\nDROP TABLE users;\n";
        let result = MigrationExecutor::extract_up_sql(content);
        assert!(result.contains("CREATE TABLE users"));
        assert!(!result.contains("DROP TABLE"));
    }

    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_extract_up_sql_only_up() {
        let content = "-- UP:\nCREATE TABLE users (id INTEGER);\n";
        let result = MigrationExecutor::extract_up_sql(content);
        assert!(result.contains("CREATE TABLE users"));
    }

    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_extract_up_sql_no_markers() {
        let content = "CREATE TABLE users (id INTEGER);";
        let result = MigrationExecutor::extract_up_sql(content);
        // 无标记时返回整个内容
        assert!(result.contains("CREATE TABLE users"));
    }

    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_extract_up_sql_case_insensitive_markers() {
        let content = "-- up:\nCREATE TABLE t (id INTEGER);\n-- down:\nDROP TABLE t;\n";
        let result = MigrationExecutor::extract_up_sql(content);
        assert!(result.contains("CREATE TABLE t"));
        assert!(!result.contains("DROP TABLE"));
    }

    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_extract_up_sql_only_down() {
        let content = "-- DOWN:\nDROP TABLE users;\n";
        let result = MigrationExecutor::extract_up_sql(content);
        // 只有 DOWN 标记时，UP 部分为 DOWN 之前的内容（空）
        assert!(result.is_empty());
    }

    // =====================================================================
    // extract_down_sql
    // =====================================================================

    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_extract_down_sql_with_up_and_down() {
        let content = "-- UP:\nCREATE TABLE users (id INTEGER);\n-- DOWN:\nDROP TABLE users;\n";
        let result = MigrationExecutor::extract_down_sql(content);
        let down = result.expect("应提取到 DOWN SQL");
        assert_eq!(down, "DROP TABLE users;");
    }

    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_extract_down_sql_case_insensitive_markers() {
        let content = "-- up:\nCREATE TABLE t (id INTEGER);\n-- down:\nDROP TABLE t;\n";
        let result = MigrationExecutor::extract_down_sql(content);
        let down = result.expect("应提取到 DOWN SQL");
        assert_eq!(down, "DROP TABLE t;");
    }

    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_extract_down_sql_only_down() {
        let content = "-- DOWN:\nDROP TABLE users;\n";
        let result = MigrationExecutor::extract_down_sql(content);
        let down = result.expect("应提取到 DOWN SQL");
        assert_eq!(down, "DROP TABLE users;");
    }

    /// DOWN 段仅注释/空白（模板占位未填写）时拒绝回滚，且历史行保留——不发生"假回滚"
    ///
    /// 旧契约（已废弃）：仅注释的 DOWN 段会被当作 SQL 执行（0 行"成功"）后删除历史行，
    /// 用模板生成迁移的用户会得到假回滚；现改为显式报错。
    #[cfg(all(
        feature = "sqlite",
        feature = "runtime-tokio-rustls",
        feature = "auto-migrate"
    ))]
    #[tokio::test]
    async fn test_rollback_version_down_only_comments_rejected() {
        let mut executor = create_sqlite_executor().await;
        executor.load_history().await.unwrap();

        // 典型模板生成的迁移：DOWN 标记后只有注释占位，未填写真实回滚语句
        let file = MigrationFile::new(
            1,
            "template_down".to_string(),
            PathBuf::from("/migrations/001_template_down.sql"),
            "-- UP:\nCREATE TABLE template_down_test (id INTEGER);\n-- DOWN: Rollback migration\n-- Reversal of migration SQL goes here\n"
                .to_string(),
        );

        executor.apply_migration_file_public(&file).await.unwrap();
        assert_eq!(executor.get_all_versions(), vec![1]);

        let result = executor.rollback_version(1, &file).await;
        assert!(result.is_err(), "仅注释的 DOWN 段应拒绝回滚");
        let err = result.unwrap_err().to_string();
        assert!(err.contains("不含可执行语句"), "实际错误: {}", err);

        // 历史行未被删除：未发生假回滚
        executor.load_history().await.unwrap();
        assert_eq!(executor.get_all_versions(), vec![1]);
    }

    /// 对照：DOWN 段为注释 + 真实语句时正常回滚（注释不阻碍回滚）
    #[cfg(all(
        feature = "sqlite",
        feature = "runtime-tokio-rustls",
        feature = "auto-migrate"
    ))]
    #[tokio::test]
    async fn test_rollback_version_down_comments_plus_statement_succeeds() {
        let mut executor = create_sqlite_executor().await;
        executor.load_history().await.unwrap();

        let file = MigrationFile::new(
            1,
            "down_with_comments".to_string(),
            PathBuf::from("/migrations/001_down_with_comments.sql"),
            "-- UP:\nCREATE TABLE down_comment_test (id INTEGER);\n-- DOWN:\n-- 模板注释：回滚说明\nDROP TABLE down_comment_test;\n"
                .to_string(),
        );

        executor.apply_migration_file_public(&file).await.unwrap();
        assert_eq!(executor.get_all_versions(), vec![1]);

        executor.rollback_version(1, &file).await.unwrap();

        // 历史行已删除
        executor.load_history().await.unwrap();
        assert!(executor.get_all_versions().is_empty());
        // DOWN SQL 已执行：表已不存在
        let dropped = executor
            .connection
            .execute_unprepared("DROP TABLE down_comment_test")
            .await;
        assert!(
            dropped.is_err(),
            "DOWN SQL 未执行，down_comment_test 仍存在"
        );
    }

    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_extract_down_sql_missing_marker_returns_none() {
        let content = "-- UP:\nCREATE TABLE users (id INTEGER);\n";
        let result = MigrationExecutor::extract_down_sql(content);
        // 无 DOWN 标记时返回 None，表示无可回滚部分
        assert!(result.is_none());
    }

    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_extract_down_sql_no_markers_returns_none() {
        let content = "CREATE TABLE users (id INTEGER);";
        let result = MigrationExecutor::extract_down_sql(content);
        assert!(result.is_none());
    }

    // =====================================================================
    // find_marker_line 按行、大小写不敏感前缀匹配
    // =====================================================================

    /// find_marker_line 命中时返回整行边界（含换行符），未命中返回 None
    #[test]
    fn test_find_marker_line_returns_line_bounds() {
        let content = "CREATE TABLE t (id INTEGER);\n-- DOWN: 模板注释\nDROP TABLE t;\n";
        let (start, end) = find_marker_line(content, &DOWN_MARKERS).unwrap();
        assert_eq!(&content[start..end], "-- DOWN: 模板注释\n");
        assert!(find_marker_line("no markers here", &UP_MARKERS).is_none());
    }

    /// 混合大小写 `-- Down:`：旧全文精确搜索匹配不到，现应能识别
    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_extract_down_sql_mixed_case_down_marker() {
        let content = "-- up:\nCREATE TABLE t (id INTEGER);\n-- Down:\nDROP TABLE t;\n";
        let down = MigrationExecutor::extract_down_sql(content).expect("应识别 -- Down: 标记");
        assert_eq!(down, "DROP TABLE t;");
    }

    /// 裸标记 `Up:` / `DOWN`（大小写不敏感，行首匹配）
    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_extract_up_sql_bare_mixed_case_markers() {
        let content = "Up:\nCREATE TABLE t (id INTEGER);\nDOWN\nDROP TABLE t;\n";
        let up = MigrationExecutor::extract_up_sql(content);
        assert!(up.contains("CREATE TABLE t"));
        assert!(!up.contains("DROP TABLE"));
        let down = MigrationExecutor::extract_down_sql(content).expect("应识别裸 DOWN 标记");
        assert_eq!(down, "DROP TABLE t;");
    }

    /// 回归：UP 段内以 up/down 开头的标识符（如列名 updated_at / download_url）
    /// 不得被裸标记 UP / DOWN 误判为标记行而截断 UP 段（词边界守卫）
    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_extract_up_sql_not_truncated_by_up_down_prefixed_identifiers() {
        let content = "-- UP:\nCREATE TABLE posts (\n    id INTEGER PRIMARY KEY,\n    updated_at TIMESTAMP,\n    download_url TEXT\n);\n-- DOWN:\nDROP TABLE posts;\n";
        let up = MigrationExecutor::extract_up_sql(content);
        assert!(up.contains("updated_at"), "UP 段不应被 updated_at 截断");
        assert!(up.contains("download_url"), "UP 段不应被 download_url 截断");
        assert!(up.contains(");"));
        assert!(!up.contains("DROP TABLE"));
        let down = MigrationExecutor::extract_down_sql(content).unwrap();
        assert_eq!(down, "DROP TABLE posts;");
    }

    /// validate 与提取共用 find_marker_line 后口径一致：
    /// 纯文本中的 "down" 单词不再让无标记、无 SQL 的文件绕过校验
    #[test]
    fn test_migration_file_parser_marker_detection_line_based() {
        // 旧实现 contains("down") 会把纯文本误判为含 DOWN 标记而放行；
        // 现按行匹配，无标记且无 SQL 关键字时应报错
        let content = "download the file here\n";
        assert!(MigrationFileParser::parse_migration_file(content).is_err());
        // 标记行大小写不敏感：`-- Up:` / `-- Down:` 可通过校验
        assert!(MigrationFileParser::parse_migration_file("-- Up:\n-- Down:\n").is_ok());
    }

    // =====================================================================
    // down_has_executable_sql
    // =====================================================================

    #[cfg(feature = "auto-migrate")]
    #[test]
    fn test_down_has_executable_sql() {
        // 空与纯空白
        assert!(!MigrationExecutor::down_has_executable_sql(""));
        assert!(!MigrationExecutor::down_has_executable_sql("  \n\t\n"));
        // 模板占位：仅注释
        assert!(!MigrationExecutor::down_has_executable_sql(
            "-- DOWN: Rollback migration\n-- Reversal of migration SQL goes here\n"
        ));
        // 真实语句
        assert!(MigrationExecutor::down_has_executable_sql("DROP TABLE t;"));
        // 注释 + 真实语句 → 仍可执行
        assert!(MigrationExecutor::down_has_executable_sql(
            "-- 说明\nDROP TABLE t;"
        ));
        // 语句行尾注释仍视为可执行
        assert!(MigrationExecutor::down_has_executable_sql(
            "DROP TABLE t; -- 说明"
        ));
    }

    // =====================================================================
    // MigrationExecutor - scan_migrations (需要 auto-migrate)
    // =====================================================================

    #[cfg(all(
        feature = "auto-migrate",
        feature = "sqlite",
        feature = "runtime-tokio-rustls"
    ))]
    #[tokio::test]
    async fn test_scan_migrations_empty_dir() {
        let executor = create_sqlite_executor().await;
        let dir = tempfile::tempdir().unwrap();
        let result = executor.scan_migrations(dir.path());
        assert!(result.is_ok());
        assert!(result.unwrap().is_empty());
    }

    #[cfg(all(
        feature = "auto-migrate",
        feature = "sqlite",
        feature = "runtime-tokio-rustls"
    ))]
    #[tokio::test]
    async fn test_scan_migrations_nonexistent_dir() {
        let executor = create_sqlite_executor().await;
        let result = executor.scan_migrations(std::path::Path::new("/nonexistent/path"));
        // 不存在的目录返回空 Vec
        assert!(result.is_ok());
        assert!(result.unwrap().is_empty());
    }

    #[cfg(all(
        feature = "auto-migrate",
        feature = "sqlite",
        feature = "runtime-tokio-rustls"
    ))]
    #[tokio::test]
    async fn test_scan_migrations_with_files() {
        let executor = create_sqlite_executor().await;
        let dir = tempfile::tempdir().unwrap();

        // 创建迁移文件
        std::fs::write(
            dir.path().join("002_add_column.sql"),
            "ALTER TABLE t ADD COLUMN c TEXT;",
        )
        .unwrap();
        std::fs::write(
            dir.path().join("001_create_table.sql"),
            "CREATE TABLE t (id INTEGER);",
        )
        .unwrap();
        // 非SQL文件应被忽略
        std::fs::write(dir.path().join("readme.txt"), "not a migration").unwrap();
        // 无效文件名应被忽略（无法解析版本号）
        std::fs::write(
            dir.path().join("invalid.sql"),
            "CREATE TABLE t (id INTEGER);",
        )
        .unwrap();

        let result = executor.scan_migrations(dir.path());
        assert!(result.is_ok());
        let files = result.unwrap();
        assert_eq!(files.len(), 2);
        // 按版本号排序
        assert_eq!(files[0].version(), 1);
        assert_eq!(files[0].description(), "create_table");
        assert_eq!(files[1].version(), 2);
        assert_eq!(files[1].description(), "add_column");
    }

    // =====================================================================
    // MigrationExecutor - 数据库测试 (需要 sqlite feature)
    // =====================================================================

    #[cfg(all(feature = "sqlite", feature = "runtime-tokio-rustls"))]
    #[tokio::test]
    async fn test_ensure_migration_table_exists() {
        let mut executor = create_sqlite_executor().await;
        // 调用 load_history 会先 ensure_migration_table_exists
        let result = executor.load_history().await;
        assert!(result.is_ok());
        // 历史应为空
        assert!(executor.history().applied_migrations.is_empty());
    }

    #[cfg(all(feature = "sqlite", feature = "runtime-tokio-rustls"))]
    #[tokio::test]
    async fn test_apply_migration_creates_table() {
        let mut executor = create_sqlite_executor().await;
        // 先确保迁移历史表存在
        executor.load_history().await.unwrap();

        let mut migration = Migration::new(1, "create_users".into());
        migration.add_table_change(TableChange::CreateTable(Table {
            name: "users".into(),
            columns: vec![
                Column {
                    name: "id".into(),
                    column_type: ColumnType::Integer,
                    is_primary_key: true,
                    is_nullable: false,
                    has_default: false,
                    default_value: None,
                    is_auto_increment: false,
                    comment: None,
                },
                Column {
                    name: "name".into(),
                    column_type: ColumnType::String(Some(255)),
                    is_primary_key: false,
                    is_nullable: false,
                    has_default: false,
                    default_value: None,
                    is_auto_increment: false,
                    comment: None,
                },
            ],
            primary_key_columns: vec!["id".into()],
            indexes: vec![],
            foreign_keys: vec![],
            comment: None,
        }));

        let result = executor.apply_migration(&migration).await;
        assert!(result.is_ok(), "apply_migration failed: {:?}", result.err());

        // 验证历史记录
        assert_eq!(executor.get_all_versions(), vec![1]);
        assert!(executor.get_latest_migration().is_some());
        assert_eq!(executor.get_latest_migration().unwrap().version, 1);
    }

    #[cfg(all(feature = "sqlite", feature = "runtime-tokio-rustls"))]
    #[tokio::test]
    async fn test_apply_migration_multiple_versions() {
        let mut executor = create_sqlite_executor().await;
        executor.load_history().await.unwrap();

        // 应用第一个迁移
        let mut m1 = Migration::new(1, "create_table".into());
        m1.add_table_change(TableChange::CreateTable(Table {
            name: "t1".into(),
            columns: vec![Column {
                name: "id".into(),
                column_type: ColumnType::Integer,
                is_primary_key: true,
                is_nullable: false,
                has_default: false,
                default_value: None,
                is_auto_increment: false,
                comment: None,
            }],
            primary_key_columns: vec!["id".into()],
            indexes: vec![],
            foreign_keys: vec![],
            comment: None,
        }));
        executor.apply_migration(&m1).await.unwrap();

        // 应用第二个迁移
        let mut m2 = Migration::new(2, "add_column".into());
        m2.add_table_change(TableChange::AlterTable {
            table_name: "t1".into(),
            column_changes: vec![],
            added_columns: vec![Column {
                name: "name".into(),
                column_type: ColumnType::Text,
                is_primary_key: false,
                is_nullable: true,
                has_default: false,
                default_value: None,
                is_auto_increment: false,
                comment: None,
            }],
            removed_columns: vec![],
            added_indexes: vec![],
            removed_indexes: vec![],
            added_foreign_keys: vec![],
            removed_foreign_keys: vec![],
        });
        executor.apply_migration(&m2).await.unwrap();

        // 验证
        assert_eq!(executor.get_all_versions(), vec![1, 2]);
        assert_eq!(executor.get_latest_migration().unwrap().version, 2);
        assert!(executor.is_fully_migrated(2));
        assert!(!executor.is_fully_migrated(3));
    }

    #[cfg(all(feature = "sqlite", feature = "runtime-tokio-rustls"))]
    #[tokio::test]
    async fn test_load_history_after_apply() {
        let mut executor = create_sqlite_executor().await;
        executor.load_history().await.unwrap();

        // 应用迁移
        let mut m = Migration::new(1, "test".into());
        m.add_table_change(TableChange::CreateTable(Table {
            name: "test_table".into(),
            columns: vec![Column {
                name: "id".into(),
                column_type: ColumnType::Integer,
                is_primary_key: true,
                is_nullable: false,
                has_default: false,
                default_value: None,
                is_auto_increment: false,
                comment: None,
            }],
            primary_key_columns: vec!["id".into()],
            indexes: vec![],
            foreign_keys: vec![],
            comment: None,
        }));
        executor.apply_migration(&m).await.unwrap();

        // 创建新的执行器（模拟重启），从数据库加载历史
        let connection = executor.connection.clone();
        let mut executor2 = MigrationExecutor::new(connection, DatabaseType::Sqlite);
        let result = executor2.load_history().await;
        assert!(result.is_ok());
        assert_eq!(executor2.get_all_versions(), vec![1]);
        let latest = executor2.get_latest_migration().unwrap();
        assert_eq!(latest.version, 1);
        assert_eq!(latest.description, "test");
    }

    #[cfg(all(
        feature = "sqlite",
        feature = "runtime-tokio-rustls",
        feature = "auto-migrate"
    ))]
    #[tokio::test]
    async fn test_run_migrations_from_files() {
        let mut executor = create_sqlite_executor().await;
        let dir = tempfile::tempdir().unwrap();

        // 创建迁移文件
        let sql = "-- UP:\nCREATE TABLE test_table (id INTEGER PRIMARY KEY);\n";
        std::fs::write(dir.path().join("001_create_test_table.sql"), sql).unwrap();

        let result = executor.run_migrations(dir.path()).await;
        assert!(result.is_ok(), "run_migrations failed: {:?}", result.err());
        assert_eq!(result.unwrap(), 1); // 1 个迁移被应用

        // 再次运行应返回 0（已应用）
        let result = executor.run_migrations(dir.path()).await;
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), 0);
    }

    #[cfg(all(
        feature = "sqlite",
        feature = "runtime-tokio-rustls",
        feature = "auto-migrate"
    ))]
    #[tokio::test]
    async fn test_run_migrations_empty_dir() {
        let mut executor = create_sqlite_executor().await;
        let dir = tempfile::tempdir().unwrap();

        let result = executor.run_migrations(dir.path()).await;
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), 0);
    }

    #[cfg(all(
        feature = "sqlite",
        feature = "runtime-tokio-rustls",
        feature = "auto-migrate"
    ))]
    #[tokio::test]
    async fn test_get_pending_migrations() {
        let mut executor = create_sqlite_executor().await;
        executor.load_history().await.unwrap();

        // 先应用版本 1
        let mut m1 = Migration::new(1, "first".into());
        m1.add_table_change(TableChange::CreateTable(Table {
            name: "t1".into(),
            columns: vec![Column {
                name: "id".into(),
                column_type: ColumnType::Integer,
                is_primary_key: true,
                is_nullable: false,
                has_default: false,
                default_value: None,
                is_auto_increment: false,
                comment: None,
            }],
            primary_key_columns: vec!["id".into()],
            indexes: vec![],
            foreign_keys: vec![],
            comment: None,
        }));
        executor.apply_migration(&m1).await.unwrap();

        // 检查待应用迁移
        let all = vec![
            m1,
            Migration::new(2, "second".into()),
            Migration::new(3, "third".into()),
        ];
        let pending = executor.get_pending_migrations(&all).await;
        assert_eq!(pending.len(), 2);
        assert_eq!(pending[0].version, 2);
        assert_eq!(pending[1].version, 3);
    }

    #[cfg(all(
        feature = "sqlite",
        feature = "runtime-tokio-rustls",
        feature = "auto-migrate"
    ))]
    #[tokio::test]
    async fn test_apply_migration_file_public() {
        let mut executor = create_sqlite_executor().await;
        executor.load_history().await.unwrap();

        let file = MigrationFile::new(
            1,
            "create_test".to_string(),
            PathBuf::from("/migrations/001_create_test.sql"),
            "CREATE TABLE test_table (id INTEGER PRIMARY KEY);".to_string(),
        );

        let result = executor.apply_migration_file_public(&file).await;
        assert!(
            result.is_ok(),
            "apply_migration_file_public failed: {:?}",
            result.err()
        );
        assert_eq!(executor.get_all_versions(), vec![1]);
    }

    // =====================================================================
    // rollback_version
    // =====================================================================

    /// 正常回滚：DOWN SQL 与历史行删除在同一事务内完成
    #[cfg(all(
        feature = "sqlite",
        feature = "runtime-tokio-rustls",
        feature = "auto-migrate"
    ))]
    #[tokio::test]
    async fn test_rollback_version_applies_down_and_deletes_history() {
        let mut executor = create_sqlite_executor().await;
        executor.load_history().await.unwrap();

        let file = MigrationFile::new(
            1,
            "create_rollback_test".to_string(),
            PathBuf::from("/migrations/001_create_rollback_test.sql"),
            "-- UP:\nCREATE TABLE rollback_test (id INTEGER);\n-- DOWN:\nDROP TABLE rollback_test;\n"
                .to_string(),
        );

        executor.apply_migration_file_public(&file).await.unwrap();
        assert_eq!(executor.get_all_versions(), vec![1]);

        executor.rollback_version(1, &file).await.unwrap();

        // 历史行已被删除（用新执行器从数据库重新加载验证）
        let connection = executor.connection.clone();
        let mut executor2 = MigrationExecutor::new(connection, DatabaseType::Sqlite);
        executor2.load_history().await.unwrap();
        assert!(executor2.get_all_versions().is_empty());

        // DOWN SQL 已执行：表已被删除，再次 DROP 应失败
        let dropped = executor
            .connection
            .execute_unprepared("DROP TABLE rollback_test")
            .await;
        assert!(dropped.is_err(), "DOWN SQL 未执行，rollback_test 仍存在");
    }

    /// 无 DOWN 段时返回明确错误，且不删除历史记录
    #[cfg(all(
        feature = "sqlite",
        feature = "runtime-tokio-rustls",
        feature = "auto-migrate"
    ))]
    #[tokio::test]
    async fn test_rollback_version_missing_down_section() {
        let mut executor = create_sqlite_executor().await;
        executor.load_history().await.unwrap();

        let file = MigrationFile::new(
            1,
            "no_down".to_string(),
            PathBuf::from("/migrations/001_no_down.sql"),
            "-- UP:\nCREATE TABLE no_down_test (id INTEGER);\n".to_string(),
        );

        executor.apply_migration_file_public(&file).await.unwrap();
        assert_eq!(executor.get_all_versions(), vec![1]);

        let result = executor.rollback_version(1, &file).await;
        assert!(result.is_err(), "无 DOWN 段的迁移应回滚失败");
        let err = result.unwrap_err().to_string();
        assert!(err.contains("无可回滚的 DOWN 部分"), "实际错误: {}", err);

        // 历史记录保持不变
        executor.load_history().await.unwrap();
        assert_eq!(executor.get_all_versions(), vec![1]);
    }

    /// DOWN SQL 执行失败时整体回滚，历史记录保留
    #[cfg(all(
        feature = "sqlite",
        feature = "runtime-tokio-rustls",
        feature = "auto-migrate"
    ))]
    #[tokio::test]
    async fn test_rollback_version_down_failure_keeps_history() {
        let mut executor = create_sqlite_executor().await;
        executor.load_history().await.unwrap();

        let file = MigrationFile::new(
            1,
            "bad_down".to_string(),
            PathBuf::from("/migrations/001_bad_down.sql"),
            "-- UP:\nCREATE TABLE bad_down_test (id INTEGER);\n-- DOWN:\nTHIS IS NOT VALID SQL;\n"
                .to_string(),
        );

        executor.apply_migration_file_public(&file).await.unwrap();
        assert_eq!(executor.get_all_versions(), vec![1]);

        let result = executor.rollback_version(1, &file).await;
        assert!(result.is_err(), "非法 DOWN SQL 应导致回滚失败");

        // 事务回滚后历史记录保留
        executor.load_history().await.unwrap();
        assert_eq!(executor.get_all_versions(), vec![1]);
    }

    // =====================================================================
    // 辅助函数
    // =====================================================================

    /// 创建基于 SQLite 内存数据库的 MigrationExecutor
    #[cfg(all(feature = "sqlite", feature = "runtime-tokio-rustls"))]
    async fn create_sqlite_executor() -> MigrationExecutor {
        let connection = sea_orm::Database::connect("sqlite::memory:").await.unwrap();
        MigrationExecutor::new(connection, DatabaseType::Sqlite)
    }
}