tdm_server_rust/runner/

schema_migration_runner.rs

//! PostgreSQL 启动迁移。
//!
//! 迁移在数据库连接建立后、RSS 和缓存预热启动前执行，避免新字段未创建时业务任务先查询失败。

use sqlx::PgPool;

/// 旧库接入迁移系统时跳过的历史迁移版本上限。
const LEGACY_BOOTSTRAP_CUTOFF_VERSION: &str = "202606181200";
/// 去重迁移之前的已知结构版本，用于兼容没有迁移历史的现有数据库。
const PREVIOUS_MIGRATION_VERSION: &str = "202607171900";
/// 当前编译进服务的最新迁移版本。
const LATEST_MIGRATION_VERSION: &str = "202607190130";

/// 单个内置迁移脚本。
struct Migration {
    /// 迁移版本号，来自文件名前缀。
    version: &'static str,
    /// 迁移名称，来自文件名说明段。
    name: &'static str,
    /// 迁移 SQL 内容。
    sql: &'static str,
}

/// 编译进二进制的 PostgreSQL 迁移列表。
const MIGRATIONS: &[Migration] = &[
    Migration {
        version: "202606170000",
        name: "postgresql_baseline",
        sql: include_str!("../../migrations/V202606170000__postgresql_baseline.sql"),
    },
    Migration {
        version: "202606171200",
        name: "Add_select_query_indexes",
        sql: include_str!("../../migrations/V202606171200__Add_select_query_indexes.sql"),
    },
    Migration {
        version: "202606171400",
        name: "Add_pg_trgm_fuzzy_indexes",
        sql: include_str!("../../migrations/V202606171400__Add_pg_trgm_fuzzy_indexes.sql"),
    },
    Migration {
        version: "202606181200",
        name: "Fix_unpublished_episode_indexes",
        sql: include_str!("../../migrations/V202606181200__Fix_unpublished_episode_indexes.sql"),
    },
    Migration {
        version: "202606201000",
        name: "Add_editor_tables",
        sql: include_str!("../../migrations/V202606201000__Add_editor_tables.sql"),
    },
    Migration {
        version: "202606201300",
        name: "Add_member_avatar_url",
        sql: include_str!("../../migrations/V202606201300__Add_member_avatar_url.sql"),
    },
    Migration {
        version: "202606201500",
        name: "Add_global_search_trgm_indexes",
        sql: include_str!("../../migrations/V202606201500__Add_global_search_trgm_indexes.sql"),
    },
    Migration {
        version: "202606201610",
        name: "Create_episode_role_taker",
        sql: include_str!("../../migrations/V202606201610__Create_episode_role_taker.sql"),
    },
    Migration {
        version: "202606211420",
        name: "Fix_editor_page_id_bigint",
        sql: include_str!("../../migrations/V202606211420__Fix_editor_page_id_bigint.sql"),
    },
    Migration {
        version: "202606211500",
        name: "Fix_editor_timestamp_utc",
        sql: include_str!("../../migrations/V202606211500__Fix_editor_timestamp_utc.sql"),
    },
    Migration {
        version: "202606211510",
        name: "Force_editor_timestamp_utc",
        sql: include_str!("../../migrations/V202606211510__Force_editor_timestamp_utc.sql"),
    },
    Migration {
        version: "202607020900",
        name: "Create_gray_release_tables",
        sql: include_str!("../../migrations/V202607020900__Create_gray_release_tables.sql"),
    },
    Migration {
        version: "202607171900",
        name: "Add_episode_type",
        sql: include_str!("../../migrations/V202607171900__Add_episode_type.sql"),
    },
    Migration {
        version: "202607190130",
        name: "Deduplicate_workreminder",
        sql: include_str!("../../migrations/V202607190130__Deduplicate_workreminder.sql"),
    },
];

/// 执行所有待应用迁移。
pub async fn run(pool: &PgPool) -> anyhow::Result<()> {
    ensure_migration_table(pool).await?;

    bootstrap_existing_database(pool).await?;

    for migration in MIGRATIONS {
        if is_applied(pool, migration.version).await? {
            continue;
        }
        apply_migration(pool, migration).await?;
    }

    Ok(())
}

/// 确保迁移记录表存在。
async fn ensure_migration_table(pool: &PgPool) -> anyhow::Result<()> {
    sqlx::query(
        r#"
        CREATE TABLE IF NOT EXISTS __tdm_schema_migrations (
            version TEXT PRIMARY KEY,
            name TEXT NOT NULL,
            applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
        )
        "#,
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// 为没有迁移历史的既有结构补记版本。
async fn bootstrap_existing_database(pool: &PgPool) -> anyhow::Result<()> {
    let applied_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM __tdm_schema_migrations")
        .fetch_one(pool)
        .await?;

    let membertb: Option<String> =
        sqlx::query_scalar("SELECT to_regclass('public.membertb')::text")
            .fetch_one(pool)
            .await?;
    let core_schema_present = membertb.is_some();

    // Docker 首次初始化会直接执行 migrations/*.sql，但不会创建 Flyway 历史表。
    let previous_schema_present: bool = sqlx::query_scalar(
        r#"
        SELECT EXISTS (
            SELECT 1
            FROM information_schema.columns
            WHERE table_schema = 'public'
              AND table_name = 'mangaepisodetb'
              AND column_name = 'episodeType'
        )
        "#,
    )
    .fetch_one(pool)
    .await?;

    // 最新唯一索引存在时，说明排序在它之前的脚本也已执行，可安全补记全部版本。
    let latest_schema_present: bool =
        sqlx::query_scalar("SELECT to_regclass('public.uk_workreminder_episode_post') IS NOT NULL")
            .fetch_one(pool)
            .await?;

    if let Some(cutoff) = bootstrap_cutoff(
        applied_count,
        core_schema_present,
        previous_schema_present,
        latest_schema_present,
    ) {
        mark_history_through(pool, cutoff).await?;
    }

    Ok(())
}

fn bootstrap_cutoff(
    applied_count: i64,
    core_schema_present: bool,
    previous_schema_present: bool,
    latest_schema_present: bool,
) -> Option<&'static str> {
    if !core_schema_present {
        None
    } else if latest_schema_present {
        Some(LATEST_MIGRATION_VERSION)
    } else if applied_count > 0 {
        None
    } else if previous_schema_present {
        Some(PREVIOUS_MIGRATION_VERSION)
    } else {
        Some(LEGACY_BOOTSTRAP_CUTOFF_VERSION)
    }
}

/// 旧库已有基线表结构，历史基线和索引迁移只登记，不重复执行。
async fn mark_history_through(pool: &PgPool, cutoff_version: &str) -> anyhow::Result<()> {
    for migration in MIGRATIONS {
        if migration.version <= cutoff_version {
            mark_applied(pool, migration).await?;
        }
    }
    tracing::info!("既有数据库迁移记录初始化完成 cutoff={cutoff_version}");
    Ok(())
}

/// 查询迁移是否已应用（同时检查应用内嵌迁移表和 Flyway 历史表）。
async fn is_applied(pool: &PgPool, version: &str) -> anyhow::Result<bool> {
    // 先查应用自己的迁移记录表
    let app_applied: Option<String> =
        sqlx::query_scalar("SELECT version FROM __tdm_schema_migrations WHERE version = $1")
            .bind(version)
            .fetch_optional(pool)
            .await?;
    if app_applied.is_some() {
        return Ok(true);
    }

    let flyway_history_exists: bool =
        sqlx::query_scalar("SELECT to_regclass('flyway_schema_history') IS NOT NULL")
            .fetch_one(pool)
            .await?;
    if !flyway_history_exists {
        return Ok(false);
    }

    // 再查 Flyway 历史表（deploy 脚本通过 Flyway 以 superuser 运行迁移，
    // 应用以 app user 连接时无法 REPLACE superuser 拥有的函数，必须跳过）
    let flyway_applied: Option<String> = sqlx::query_scalar(
        "SELECT version FROM flyway_schema_history WHERE version = $1 AND success = true",
    )
    .bind(version)
    .fetch_optional(pool)
    .await?;
    if flyway_applied.is_some() {
        // 同步到应用迁移表，避免后续重复检查 Flyway
        sqlx::query(
            "INSERT INTO __tdm_schema_migrations(version, name) VALUES ($1, $2) ON CONFLICT (version) DO NOTHING",
        )
        .bind(version)
        .bind("synced_from_flyway")
        .execute(pool)
        .await?;
        return Ok(true);
    }
    Ok(false)
}

/// 执行迁移并记录版本。
async fn apply_migration(pool: &PgPool, migration: &Migration) -> anyhow::Result<()> {
    tracing::info!(
        "开始执行数据库迁移 version={} name={}",
        migration.version,
        migration.name
    );
    let mut tx = pool.begin().await?;
    sqlx::raw_sql(migration.sql).execute(&mut *tx).await?;
    sqlx::query(
        r#"
        INSERT INTO __tdm_schema_migrations(version, name)
        VALUES ($1, $2)
        ON CONFLICT (version) DO NOTHING
        "#,
    )
    .bind(migration.version)
    .bind(migration.name)
    .execute(&mut *tx)
    .await?;
    tx.commit().await?;
    tracing::info!(
        "数据库迁移执行完成 version={} name={}",
        migration.version,
        migration.name
    );
    Ok(())
}

/// 仅记录迁移版本。
async fn mark_applied(pool: &PgPool, migration: &Migration) -> anyhow::Result<()> {
    sqlx::query(
        r#"
        INSERT INTO __tdm_schema_migrations(version, name)
        VALUES ($1, $2)
        ON CONFLICT (version) DO NOTHING
        "#,
    )
    .bind(migration.version)
    .bind(migration.name)
    .execute(pool)
    .await?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{
        bootstrap_cutoff, LATEST_MIGRATION_VERSION, LEGACY_BOOTSTRAP_CUTOFF_VERSION, MIGRATIONS,
        PREVIOUS_MIGRATION_VERSION,
    };

    #[test]
    fn bootstrap_cutoff_matches_database_shape() {
        assert_eq!(bootstrap_cutoff(0, false, false, false), None);
        assert_eq!(
            bootstrap_cutoff(0, true, false, false),
            Some(LEGACY_BOOTSTRAP_CUTOFF_VERSION)
        );
        assert_eq!(
            bootstrap_cutoff(0, true, true, false),
            Some(PREVIOUS_MIGRATION_VERSION)
        );
        assert_eq!(
            bootstrap_cutoff(0, true, true, true),
            Some(LATEST_MIGRATION_VERSION)
        );
        assert_eq!(
            bootstrap_cutoff(4, true, true, true),
            Some(LATEST_MIGRATION_VERSION)
        );
        assert_eq!(bootstrap_cutoff(4, true, true, false), None);
    }

    /// 确保最新去重迁移已编译进启动迁移清单。
    #[test]
    fn latest_migration_version_matches_embedded_list() {
        assert_eq!(LATEST_MIGRATION_VERSION, "202607190130");
        assert_eq!(
            MIGRATIONS.last().map(|migration| migration.version),
            Some("202607190130")
        );
    }

    /// 断言去重迁移在清理和建索引前取得写入阻断锁。
    fn assert_migration_lock_order(sql: &str, lock_marker: &str) {
        let lock = sql.find(lock_marker).expect("迁移必须先取得写入阻断锁");
        let delete = sql.find("DELETE").expect("迁移必须清理历史重复行");
        let index = sql
            .find("CREATE UNIQUE INDEX")
            .expect("迁移必须创建唯一索引");
        assert!(lock < delete, "写入阻断锁必须早于重复清理");
        assert!(delete < index, "重复清理必须早于唯一索引创建");
    }

    /// PostgreSQL 与 MySQL 去重迁移都应关闭清理和建索引之间的并发窗口。
    #[test]
    fn dedup_migrations_lock_before_cleanup_and_unique_index() {
        let postgres = include_str!("../../migrations/V202607190130__Deduplicate_workreminder.sql");
        let mysql =
            include_str!("../../migrations/mysql/V202607190130__Deduplicate_workreminder.sql");

        assert_migration_lock_order(
            postgres,
            "LOCK TABLE workreminder IN SHARE ROW EXCLUSIVE MODE",
        );
        assert_migration_lock_order(mysql, "LOCK TABLES");
        for lock_target in [
            "workreminder WRITE",
            "workreminder AS older WRITE",
            "workreminder AS newer WRITE",
        ] {
            assert!(
                mysql.contains(lock_target),
                "MySQL 缺少锁目标: {lock_target}"
            );
        }
        assert!(
            mysql.find("CREATE UNIQUE INDEX").unwrap() < mysql.find("UNLOCK TABLES").unwrap(),
            "MySQL 应在唯一索引创建后释放表锁"
        );
    }
}