refactor(迁移): 换掉 drizzle 的执行器，一条迁移一个事务

drizzle 的 migrate()（pg-core/dialect.js）把所有待执行的迁移塞进同一个 session.transaction()，两个后果：第 3 条失败会把第 1、2 条一起回滚，和 Django migrate 的逐条提交语义不一样；以及 CREATE INDEX CONCURRENTLY 一律跑不了，没有任何开关。现在自己按 journal 逐条执行，一条一个事务。文件第一行写 `-- oj2:no-transaction` 的迁移走裸执行（简单查询协议——扩展协议会把语句包进隐式事务块，CONCURRENTLY 照样被拒），代价是没有回滚，所以这类迁移一个文件只放一条语句。记账行的写法和 drizzle 完全一致（hash = 整文件 sha256，created_at = journal 的 when），migrator 又只比 created_at、不校验 hash，两套执行器可以互换。顺带：日志和报错说得出迁移文件名了（readMigrationFiles 不返回 tag，自己读一遍 journal），失败时打印的是 Postgres 那句话而不是 postgres.js 的内部堆栈，新增退出码 5。实跑验证（本机 Docker，生产 schema 灌进探针库）：CONCURRENTLY 带标记成功、 indisvalid = t，去掉标记复现 "cannot run inside a transaction block"； 0003 成功 + 0004 失败时，0003 留下、0004 的首条合法语句没留下。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-26 09:54:01 -06:00
parent eabb61189c
commit 2e871cb5b0
2 changed files with 141 additions and 10 deletions
--- a/apps/api/src/db/migrate.ts
+++ b/apps/api/src/db/migrate.ts
@@ -1,6 +1,6 @@
+import { readFileSync } from "node:fs"
+
 import { readMigrationFiles } from "drizzle-orm/migrator"
-import { drizzle } from "drizzle-orm/postgres-js"
-import { migrate as drizzleMigrate } from "drizzle-orm/postgres-js/migrator"
 import postgres from "postgres"

 import { migrationsDir } from "../runtime"
@@ -61,6 +61,10 @@ export async function runMigrations() {
    process.exit(2)
  }

+  // readMigrationFiles 只返回 { sql, hash, folderMillis }，不给文件名。日志和报错里
+  // 说「0002_drop_django_leftovers」比说「1787740469403」有用得多，所以自己读一遍 journal。
+  const tags = readMigrationTags()
+
  // max: 1 —— advisory lock 是会话级的，多连接会让锁挂在另一条连接上，等于没锁
  const client = postgres(url, { max: 1, onnotice: () => {} })

@@ -114,13 +118,16 @@ export async function runMigrations() {
    }

    const blocked = pending
-      .map((f) => destructiveReasons(f.sql.join("\n")))
-      .filter((reasons) => reasons.length > 0)
+      .map((f) => ({
+        tag: tags.get(f.folderMillis) ?? String(f.folderMillis),
+        reasons: destructiveReasons(f.sql.join("\n")),
+      }))
+      .filter(({ reasons }) => reasons.length > 0)

    if (blocked.length > 0 && process.env.OJ2_ALLOW_DESTRUCTIVE !== "1") {
      console.error(
        "待执行的迁移里有破坏性语句，已停下：\n" +
-          blocked.map((reasons) => `  · ${reasons.join(" / ")}`).join("\n") +
+          blocked.map(({ tag, reasons }) => `  · ${tag}：${reasons.join(" / ")}`).join("\n") +
          "\n\n这类改动不可逆，不该在一次日常部署里顺手执行。" +
          "\n确认已经做过备份之后，用这个显式放行：\n\n" +
          "  OJ2_ALLOW_DESTRUCTIVE=1 docker/deploy.sh\n",
@@ -129,7 +136,30 @@ export async function runMigrations() {
    }

    console.log(`待执行 ${pending.length} 条迁移，开始。`)
-    await drizzleMigrate(drizzle(client), { migrationsFolder: migrationsDir })
+    let done = 0
+    for (const file of pending) {
+      const tag = tags.get(file.folderMillis) ?? String(file.folderMillis)
+      try {
+        await applyMigration(client, file, tag)
+      } catch (error) {
+        // 裸抛的话看到的是 postgres.js 内部的堆栈，真正的原因（那一行 PostgresError）
+        // 被埋在中间。这里只留有用的部分。
+        const detail = error instanceof Error ? error.message : String(error)
+        const naked = NO_TRANSACTION_MARKER.test(file.sql[0] ?? "")
+        console.error(
+          `\n迁移 ${tag} 失败：${detail}\n\n` +
+            (naked
+              ? "这条迁移标了 oj2:no-transaction，**没有事务保护** —— 失败点之前的语句已经生效。\n" +
+                "如果失败的是 CREATE INDEX CONCURRENTLY，库里多半留下了一个 INVALID 索引，\n" +
+                "先 `DROP INDEX <名字>` 再重来（`select indexrelid::regclass from pg_index where not indisvalid` 能找出来）。\n"
+              : "这条迁移已整体回滚，库里没有留下它的任何改动。\n") +
+            `本次已经成功执行的 ${done} 条不会被回滚 —— 每条迁移各自一个事务。`,
+        )
+        process.exit(5)
+      }
+      done++
+      console.log(`  ✓ ${tag}`)
+    }
    console.log("迁移完成。")
  } finally {
    await client.end()
@@ -140,3 +170,77 @@ function destructiveReasons(sql: string) {
  const bare = stripComments(sql)
  return DESTRUCTIVE_PATTERNS.filter(([re]) => re.test(bare)).map(([, label]) => label)
 }
+
+/**
+ * 从 `meta/_journal.json` 读出 `when → tag` 的对应。`readMigrationFiles` 不返回文件名，
+ * 但日志和报错里说得出「0002_drop_django_leftovers」比说「1787740469403」有用得多。
+ *
+ * 读不到就返回空表 —— 到这一步 `readMigrationFiles` 已经成功读过同一个文件了，
+ * 真读不到也只是日志退化成时间戳，不该因此中止一次迁移。
+ */
+function readMigrationTags(): Map<number, string> {
+  try {
+    const journal = JSON.parse(readFileSync(`${migrationsDir}/meta/_journal.json`, "utf8")) as {
+      entries?: Array<{ when: number; tag: string }>
+    }
+    return new Map((journal.entries ?? []).map((e) => [e.when, e.tag]))
+  } catch {
+    return new Map()
+  }
+}
+
+/**
+ * 写在迁移文件**开头**的这行标记，表示这条迁移不能包在事务里跑。
+ *
+ * 唯一的用途是 `CREATE INDEX CONCURRENTLY` —— Postgres 明确禁止它出现在事务块里，
+ * 而大表加索引又常常不能接受 `CREATE INDEX` 那段锁写窗口。
+ *
+ * 代价要清楚：**没有回滚**。中途失败时前面的语句已经生效，而且 CONCURRENTLY 失败还会
+ * 在库里留下一个 INVALID 索引，得手工 `DROP INDEX` 之后重来。所以这种迁移**一个文件
+ * 只放一条语句**，别图省事把几条塞一起。
+ */
+const NO_TRANSACTION_MARKER = /^[ \t]*--[ \t]*oj2:no-transaction\b/m
+
+/**
+ * 执行一条迁移。
+ *
+ * 这里没有用 drizzle 自带的 `migrate()`，原因有两条，都在 `pg-core/dialect.js` 里摆着：
+ *
+ * 1. 它把**所有**待执行的迁移塞进同一个 `session.transaction()`。于是第 3 条失败会把
+ *    第 1、2 条一起回滚 —— 和 Django `migrate` 的逐条提交语义不一样，排查时也更难判断
+ *    库到底停在哪儿。这里改成一条一个事务。
+ * 2. 正因为全都在事务里，`CREATE INDEX CONCURRENTLY` 一律跑不了，没有任何开关。
+ *
+ * 记账行（`drizzle.__drizzle_migrations`）的写法和 drizzle 保持一致：`hash` 是整个文件的
+ * sha256，`created_at` 是 journal 里的 `when`。migrator 只比 `created_at`、不校验 hash，
+ * 所以两套执行器可以互换着用，不会互相看不懂对方写的记录。
+ */
+async function applyMigration(
+  client: postgres.Sql,
+  migration: ReturnType<typeof readMigrationFiles>[number],
+  tag: string,
+) {
+  // 只留有可执行内容的段。`readMigrationFiles` 按 `--> statement-breakpoint` 切开后
+  // 保留原文，所以纯注释段（比如 0002 开头那一大段说明）会自成一段。
+  const statements = migration.sql.filter((stmt) => stripComments(stmt).trim() !== "")
+  if (statements.length === 0) {
+    // 上游已经拦过一次（那条兜底检查），走到这里说明拦漏了，宁可响一声也别静默跳过
+    throw new Error(`${tag} 没有任何可执行语句`)
+  }
+
+  const record = (exec: postgres.Sql | postgres.TransactionSql) =>
+    exec`insert into drizzle.__drizzle_migrations ("hash", "created_at")
+         values (${migration.hash}, ${migration.folderMillis})`
+
+  if (NO_TRANSACTION_MARKER.test(migration.sql[0] ?? "")) {
+    // 走简单查询协议：扩展协议会把语句包进一个隐式事务块，CONCURRENTLY 照样被拒。
+    for (const stmt of statements) await client.unsafe(stmt).simple()
+    await record(client)
+    return
+  }
+
+  await client.begin(async (tx) => {
+    for (const stmt of statements) await tx.unsafe(stmt)
+    await record(tx)
+  })
+}