Merge branch 'feat/embedded-migrations'

Embed database migrations into the server binary: migrations now run
automatically on startup before the server listens, via a hand-written
runner (~150 LOC) using the existing tokio-postgres/deadpool stack.

- New src/db/migrate.rs: MIGRATIONS constant (include_str!), MigrateError,
  run() with advisory lock + per-migration transactions
- schema_migrations table tracks applied versions
- main.rs drives the async runner via a dedicated transient tokio runtime
- migrate.sh retained as a manual/CI fallback with updated header
- 4 unit tests (sorted, unique, non-empty, files-match-rows)
- Verified end-to-end: cold start, warm restart, rewind, failure path

migrate.sh

@@ -1,4 +1,16 @@
 #!/usr/bin/env bash
+#
+# 数据库迁移脚本（手动 / CI 备用）。
+#
+# 注意：服务器二进制现在会在启动时自动执行迁移（见 src/db/migrate.rs），
+# 正常情况下无需手动运行本脚本。保留它是为了：
+#   - 运维 escape hatch：二进制因迁移失败起不来时，手动救库
+#   - CI/CD 中“先迁移再滚动发布”的工作流
+#
+# 本脚本与服务器内置运行器读取相同的 migrations/*.sql 文件，且这些 SQL
+# 都是幂等的（IF NOT EXISTS / IF EXISTS / ON CONFLICT DO NOTHING），
+# 因此两者混用安全。
+#
 set -euo pipefail
 
 # 从 .env 加载环境变量

src/db/migrate.rs

@@ -0,0 +1,317 @@
+//! 数据库迁移运行器。
+//!
+//! 在服务器启动时（`dioxus::server::serve()` 之前）自动执行迁移。
+//! 设计要点：
+//! - 迁移 SQL 通过 `include_str!` 内联进二进制，部署只需单个二进制。
+//! - `schema_migrations` 表记录已应用版本，避免重复执行。
+//! - 每个迁移在独立事务里执行，失败自动回滚，版本行不会写入。
+//! - 咨询锁（`pg_advisory_lock`）保证多实例启动时只有一个进程执行迁移。
+//!
+//! 仅在 `feature = "server"` 时编译。
+
+use std::collections::HashSet;
+
+/// 咨询锁的固定 key。Postgres 咨询锁是数据库级唯一的；
+/// 这里用一个项目专属的大整数，避免与同库其它应用冲突。
+/// 该值无语义，仅要求唯一性。
+const ADVISORY_LOCK_KEY: i64 = 0x5947_4752_4153_494C;
+
+/// 所有迁移的 (version, sql) 列表，按 version 升序排列。
+///
+/// 新增迁移时：
+/// 1. 在 `migrations/` 下创建 `NNN_描述.sql`
+/// 2. 在本数组末尾追加一行 `("NNN", include_str!("../../migrations/NNN_描述.sql"))`
+///
+/// version 用字符串而非整数，与文件名前缀直接对应，且未来可支持日期戳/语义版本。
+const MIGRATIONS: &[(&str, &str)] = &[
+    ("001", include_str!("../../migrations/001_init.sql")),
+    ("002", include_str!("../../migrations/002_posts.sql")),
+    ("003", include_str!("../../migrations/003_indexes.sql")),
+    ("004", include_str!("../../migrations/004_search_trgm.sql")),
+    ("005", include_str!("../../migrations/005_comments.sql")),
+    ("006", include_str!("../../migrations/006_add_toc_html.sql")),
+    ("007", include_str!("../../migrations/007_settings.sql")),
+    ("008", include_str!("../../migrations/008_comments_cascade.sql")),
+    (
+        "009",
+        include_str!("../../migrations/009_cleanup_duplicate_indexes.sql"),
+    ),
+    (
+        "010",
+        include_str!("../../migrations/010_post_word_counts.sql"),
+    ),
+    ("011", include_str!("../../migrations/011_perf_indexes.sql")),
+    (
+        "012",
+        include_str!("../../migrations/012_session_generation.sql"),
+    ),
+    (
+        "013",
+        include_str!("../../migrations/013_comment_content_hash_index.sql"),
+    ),
+    (
+        "014",
+        include_str!("../../migrations/014_drop_ineffective_trgm_index.sql"),
+    ),
+    // 新增迁移在此追加，同时在 migrations/ 下创建对应 .sql 文件。
+];
+
+/// 迁移执行错误。
+#[derive(Debug)]
+pub enum MigrateError {
+    /// 无法从连接池获取连接。
+    Pool(deadpool_postgres::PoolError),
+    /// 执行查询（建表、查版本、咨询锁等）失败。
+    Query(tokio_postgres::Error),
+    /// 某个具体迁移执行失败（包含版本号便于定位）。
+    Apply {
+        version: String,
+        source: tokio_postgres::Error,
+    },
+}
+
+impl From<deadpool_postgres::PoolError> for MigrateError {
+    fn from(e: deadpool_postgres::PoolError) -> Self {
+        MigrateError::Pool(e)
+    }
+}
+
+impl From<tokio_postgres::Error> for MigrateError {
+    fn from(e: tokio_postgres::Error) -> Self {
+        MigrateError::Query(e)
+    }
+}
+
+impl std::fmt::Display for MigrateError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            MigrateError::Pool(e) => write!(f, "database pool error: {}", e),
+            MigrateError::Query(e) => write!(f, "database query error: {}", e),
+            MigrateError::Apply { version, source } => {
+                write!(f, "migration {} failed: {}", version, source)
+            }
+        }
+    }
+}
+
+impl std::error::Error for MigrateError {}
+
+/// 执行所有未应用的迁移。
+///
+/// 在 `main.rs` 启动时调用一次。流程：
+/// 1. 获取一个独占连接（咨询锁是 session 级，需在同一连接上 lock/unlock）
+/// 2. 抢咨询锁（多实例启动时串行化）
+/// 3. 确保 `schema_migrations` 表存在
+/// 4. 查询已应用版本集合
+/// 5. 按序应用未应用的迁移（每个一个事务）
+/// 6. 释放咨询锁
+///
+/// 失败时返回错误；调用方（`main.rs`）应让进程退出，避免启动半残服务。
+pub async fn run() -> Result<(), MigrateError> {
+    use crate::db::pool::get_conn;
+
+    tracing::info!("running database migrations");
+
+    // 咨询锁是 session 级的，必须在同一连接上 lock/unlock。
+    // 因此整个迁移流程独占一个池连接。
+    let mut conn = get_conn().await?;
+
+    // 抢咨询锁：多实例滚动发布时只有一个进程能进入迁移循环，
+    // 其余实例在此等待；锁释放后它们查版本表发现已全部应用，直接返回。
+    conn.execute("SELECT pg_advisory_lock($1)", &[&ADVISORY_LOCK_KEY])
+        .await?;
+
+    // 在已持锁连接上执行迁移主体逻辑。
+    // 锁释放策略：
+    // - 正常返回 / 返回 Err：下面的显式 pg_advisory_unlock 释放锁。
+    // - panic：panic 会跳过显式 unlock。此时 `conn` 在 unwind 中被 drop，
+    //   但 deadpool 会把连接归还池中复用（不关闭 Postgres 会话），
+    //   所以 session 级咨询锁不会立即释放。安全性依赖调用方（main.rs 的
+    //   `.expect()`）在 panic 时终止进程——进程退出会关闭所有池连接，
+    //   Postgres 检测到会话断开后释放 session 级咨询锁。
+    let result = run_inner(&mut conn).await;
+
+    // 无论成功失败都尝试显式释放锁；释放失败不应掩盖原始错误，仅记录告警。
+    if let Err(unlock_err) = conn
+        .execute("SELECT pg_advisory_unlock($1)", &[&ADVISORY_LOCK_KEY])
+        .await
+    {
+        tracing::warn!("failed to release migration advisory lock: {}", unlock_err);
+    }
+
+    result
+}
+
+/// 在已持有咨询锁的连接上执行迁移主体逻辑。
+async fn run_inner(conn: &mut deadpool_postgres::Object) -> Result<(), MigrateError> {
+    // 确保版本表存在（独立语句，不在事务里，否则建表失败无法记录）。
+    ensure_versions_table(conn).await?;
+
+    // 查询已应用的版本集合。
+    let applied = applied_versions(conn).await?;
+
+    // 按序应用未应用的迁移。
+    let mut applied_count = 0usize;
+    for (version, sql) in MIGRATIONS {
+        if applied.contains(*version) {
+            continue;
+        }
+        tracing::info!("applying migration {}", version);
+        apply_one(conn, version, sql).await?;
+        applied_count += 1;
+    }
+
+    if applied_count == 0 {
+        tracing::info!("database is up to date, 0 migrations applied");
+    } else {
+        tracing::info!("successfully applied {} migration(s)", applied_count);
+    }
+    Ok(())
+}
+
+/// 创建 `schema_migrations` 表（若不存在）。
+async fn ensure_versions_table(conn: &deadpool_postgres::Object) -> Result<(), MigrateError> {
+    conn.batch_execute(
+        "CREATE TABLE IF NOT EXISTS schema_migrations (
+            version    TEXT PRIMARY KEY,
+            applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+        )",
+    )
+    .await?;
+    Ok(())
+}
+
+/// 查询已应用的版本集合。
+async fn applied_versions(
+    conn: &deadpool_postgres::Object,
+) -> Result<HashSet<String>, MigrateError> {
+    let rows = conn.query("SELECT version FROM schema_migrations", &[]).await?;
+    let mut set = HashSet::with_capacity(rows.len());
+    for row in rows {
+        set.insert(row.get::<_, String>(0));
+    }
+    Ok(set)
+}
+
+/// 在一个事务内应用单个迁移：执行 SQL + 写入版本行，失败则回滚。
+async fn apply_one(
+    conn: &mut deadpool_postgres::Object,
+    version: &str,
+    sql: &str,
+) -> Result<(), MigrateError> {
+    let tx = conn.transaction().await.map_err(MigrateError::Query)?;
+
+    // batch_execute 执行整段 SQL（可含多条语句）。
+    if let Err(e) = tx.batch_execute(sql).await {
+        // 显式回滚以尽早释放事务（而非等 Transaction drop 的隐式回滚）；
+        // 回滚本身的错误丢弃，因为已有更具信息量的 apply 错误要上报。
+        let _ = tx.rollback().await;
+        return Err(MigrateError::Apply {
+            version: version.to_string(),
+            source: e,
+        });
+    }
+
+    // 记录版本行。显式构造 Apply 错误（不能用 ?，否则会被 blanket From 映射成 Query）。
+    if let Err(e) = tx
+        .execute(
+            "INSERT INTO schema_migrations (version) VALUES ($1)",
+            &[&version],
+        )
+        .await
+    {
+        let _ = tx.rollback().await;
+        return Err(MigrateError::Apply {
+            version: version.to_string(),
+            source: e,
+        });
+    }
+
+    tx.commit().await.map_err(|e| MigrateError::Apply {
+        version: version.to_string(),
+        source: e,
+    })?;
+    Ok(())
+}
+
+#[cfg(all(test, feature = "server"))]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn migrations_are_sorted_ascending() {
+        let mut sorted = MIGRATIONS.iter().map(|(v, _)| *v).collect::<Vec<_>>();
+        sorted.sort_unstable();
+        let original: Vec<&str> = MIGRATIONS.iter().map(|(v, _)| *v).collect();
+        assert_eq!(original, sorted, "MIGRATIONS must be in ascending version order");
+    }
+
+    #[test]
+    fn migrations_have_unique_versions() {
+        let mut versions: Vec<&str> = MIGRATIONS.iter().map(|(v, _)| *v).collect();
+        let total = versions.len();
+        versions.sort_unstable();
+        versions.dedup();
+        assert_eq!(versions.len(), total, "MIGRATIONS has duplicate version strings");
+    }
+
+    #[test]
+    fn migrations_non_empty() {
+        assert!(!MIGRATIONS.is_empty(), "MIGRATIONS must not be empty");
+    }
+
+    /// 防止"新建了 .sql 但忘记在 MIGRATIONS 加行"的脚枪。
+    /// 扫描 migrations/ 目录，断言每个 .sql 文件都在 MIGRATIONS 里有对应版本行。
+    /// 仅在 server feature + test 下运行（WASM 无文件系统）。
+    #[test]
+    fn migrations_match_files_on_disk() {
+        use std::collections::HashSet;
+        use std::fs;
+
+        // CARGO_MANIFEST_DIR 指向 crate 根目录（yggdrasil/）。
+        let manifest_dir = env!("CARGO_MANIFEST_DIR");
+        let migrations_dir = std::path::Path::new(manifest_dir).join("migrations");
+
+        let mut files_on_disk: HashSet<String> = HashSet::new();
+        for entry in fs::read_dir(&migrations_dir)
+            .unwrap_or_else(|e| panic!("failed to read {}: {}", migrations_dir.display(), e))
+        {
+            let entry = entry.unwrap();
+            let path = entry.path();
+            if path.extension().and_then(|e| e.to_str()) == Some("sql") {
+                let filename = path
+                    .file_name()
+                    .and_then(|n| n.to_str())
+                    .unwrap_or_else(|| panic!("non-utf8 filename: {}", path.display()));
+                // 文件名形如 "001_init.sql"，取前 3 位数字作为 version。
+                let version = filename
+                    .split('_')
+                    .next()
+                    .unwrap_or_else(|| panic!("filename has no '_' separator: {}", filename));
+                files_on_disk.insert(version.to_string());
+            }
+        }
+
+        let versions_in_array: HashSet<String> =
+            MIGRATIONS.iter().map(|(v, _)| v.to_string()).collect();
+
+        // 磁盘上有但数组里没有 → 忘记加行（会静默不执行该迁移）。
+        let missing_in_array: Vec<&String> =
+            files_on_disk.difference(&versions_in_array).collect();
+        assert!(
+            missing_in_array.is_empty(),
+            "migrations/*.sql files not registered in MIGRATIONS: {:?}. \
+             Add a row for each in src/db/migrate.rs.",
+            missing_in_array
+        );
+
+        // 数组里有但磁盘上没有 → include_str! 本就会编译失败，这里只是双保险。
+        let missing_on_disk: Vec<&String> =
+            versions_in_array.difference(&files_on_disk).collect();
+        assert!(
+            missing_on_disk.is_empty(),
+            "MIGRATIONS rows without a corresponding .sql file: {:?}",
+            missing_on_disk
+        );
+    }
+}

src/db/mod.rs

@@ -16,6 +16,10 @@ pub mod pool;
 #[cfg(feature = "server")]
 pub mod retry;
 
+/// 数据库迁移运行器，仅在启用 server feature 时编译。
+#[cfg(feature = "server")]
+pub mod migrate;
+
 /// 占位连接池实现，仅在不启用 server feature 时编译。
 ///
 /// `DummyPool` 是一个最小 stub：它提供与真实连接池相同的公开接口形状

src/main.rs

@@ -206,6 +206,25 @@ fn main() {
             std::process::exit(1);
         }
 
+        // 启动前执行数据库迁移。阻塞：完成前不监听端口。
+        // 失败直接 panic（expect），避免启动一个 schema 不一致的半残服务。
+        // 多实例滚动发布时由咨询锁串行化，详见 src/db/migrate.rs。
+        //
+        // main() 是同步函数，这里用一个独立的多线程 runtime 驱动迁移的异步逻辑，
+        // 完成后再交给 dioxus::server::serve() 启动它自己的 runtime。
+        // 两个 runtime 不重叠，避免与 Dioxus 内部 runtime 产生交互。
+        let migrate_rt = tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .expect("failed to build migration runtime");
+        migrate_rt.block_on(async {
+            db::migrate::run()
+                .await
+                .expect("Database migration failed on startup");
+        });
+        // 迁移 runtime 用完即弃，显式 drop 以在 serve() 前释放其线程资源。
+        drop(migrate_rt);
+
         // 启动 Dioxus 服务端，返回构建好的 Axum Router
         dioxus::server::serve(|| async move {
             use dioxus::server::{axum, DioxusRouterExt, ServeConfig};