From f6063af12dc1c2845d791a2b2aa1e90d2960ea3f Mon Sep 17 00:00:00 2001
From: root <root@localhost>
Date: Mon, 28 Sep 2026 17:15:04 +0800
Subject: [PATCH] fix(db): migrator introspects only the target schema
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Kysely's Migrator checks for its tables via introspection.getTables(),
which reflects every schema and calls has_schema_privilege /
pg_get_serial_sequence per schema. When another test file drops its schema
at the same moment those functions fail with "schema … does not exist" —
the intermittent server PG test failures on the shared dev database.
createMigrator now hands the Migrator a proxy whose introspection only
looks at the migration schema (or current_schema()). A regression test
churns schemas while migrating; it fails on the old migrator.

Also samples EngineRunQuery's comma-separated status/triggeredBy in the
contracts schema test.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 packages/contracts/test/schemas.test.ts |  1 +
 packages/db/src/migrator.ts             | 50 ++++++++++++++++++++++++-
 packages/db/test/schema.test.ts         | 21 +++++++++++
 3 files changed, 70 insertions(+), 2 deletions(-)

diff --git a/packages/contracts/test/schemas.test.ts b/packages/contracts/test/schemas.test.ts
index 5db34409..c80b776f 100644
--- a/packages/contracts/test/schemas.test.ts
+++ b/packages/contracts/test/schemas.test.ts
@@ -96,6 +96,7 @@ const OVERRIDES: Record<string, (v: Record<string, unknown>) => Record<string, u
   PathBindingCreate: (v) => ({ ...v, bindingKind: "season", seasonNumber: 1 }),
   OrganizeRequest: (v) => ({ ...v, sourcePaths: undefined }),
   FilesListQuery: (v) => ({ ...v, filter: "video,unidentified" }),
+  EngineRunQuery: (v) => ({ ...v, status: "running,pending", triggeredBy: "user,api_key" }),
   CatalogLookupInput: (v) => v,
   IdentityResolveInput: (v) => v,
   AssetLocationCreate: (v) => v,
diff --git a/packages/db/src/migrator.ts b/packages/db/src/migrator.ts
index 4a8e0a8e..ec026ad0 100644
--- a/packages/db/src/migrator.ts
+++ b/packages/db/src/migrator.ts
@@ -1,7 +1,7 @@
 import { promises as fs } from "node:fs";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
-import { type Kysely, sql } from "kysely";
+import { type DatabaseIntrospector, type Kysely, sql, type TableMetadata } from "kysely";
 import { FileMigrationProvider, type MigrationResultSet, Migrator } from "kysely/migration";
 
 /** 迁移目录:packages/db/migrations,文件名 NNNN_name.ts,导出 up/down */
@@ -16,10 +16,56 @@ export interface MigratorOptions {
   migrationTableSchema?: string;
 }
 
+/**
+ * 只看一个 schema 的内省器。Kysely 迁移器判断迁移表是否存在时会调用 introspection.getTables(),
+ * 默认实现反射全库所有 schema,并对每个 schema 调 has_schema_privilege / pg_get_serial_sequence;
+ * 同一个库里别的进程恰好 drop schema 时,这两个函数会报 "schema … does not exist"(并行测试共用开发库时的偶发失败)。
+ * 迁移器只需要知道迁移表 / 锁表在不在目标 schema,这里只查目标 schema(未指定时取 current_schema())的表名。
+ */
+// biome-ignore lint/suspicious/noExplicitAny: 同上
+function scopedIntrospector(db: Kysely<any>, schema: string | undefined): DatabaseIntrospector {
+  const getTables = async (): Promise<TableMetadata[]> => {
+    const rows = await sql<{ name: string; schema: string; kind: string }>`
+      select c.relname as name, n.nspname as schema, c.relkind as kind
+      from pg_catalog.pg_class c join pg_catalog.pg_namespace n on n.oid = c.relnamespace
+      where n.nspname = coalesce(${schema ?? null}::text, current_schema())
+        and c.relkind in ('r', 'v', 'p', 'f')`.execute(db);
+    return rows.rows.map((r) => ({
+      name: r.name,
+      schema: r.schema,
+      isView: r.kind === "v",
+      isForeign: r.kind === "f",
+      columns: [],
+    }));
+  };
+  return {
+    async getSchemas() {
+      const rows = await sql<{ name: string }>`
+        select nspname as name from pg_catalog.pg_namespace
+        where nspname = coalesce(${schema ?? null}::text, current_schema())`.execute(db);
+      return rows.rows;
+    },
+    getTables,
+  };
+}
+
+/** 把 db.introspection 换成只看目标 schema 的实现,其余成员原样转发(方法绑回原实例,Kysely 用了私有字段) */
+// biome-ignore lint/suspicious/noExplicitAny: 同上
+function withScopedIntrospection(db: Kysely<any>, schema: string | undefined): Kysely<any> {
+  const introspection = scopedIntrospector(db, schema);
+  return new Proxy(db, {
+    get(target, prop) {
+      if (prop === "introspection") return introspection;
+      const v = Reflect.get(target, prop, target);
+      return typeof v === "function" ? v.bind(target) : v;
+    },
+  });
+}
+
 // biome-ignore lint/suspicious/noExplicitAny: 迁移器不关心具体表类型
 export function createMigrator(db: Kysely<any>, opts: MigratorOptions = {}): Migrator {
   return new Migrator({
-    db,
+    db: withScopedIntrospection(db, opts.migrationTableSchema),
     provider: new FileMigrationProvider({ fs, path, migrationFolder: MIGRATIONS_DIR }),
     ...(opts.migrationTableSchema ? { migrationTableSchema: opts.migrationTableSchema } : {}),
   });
diff --git a/packages/db/test/schema.test.ts b/packages/db/test/schema.test.ts
index 836b3bdb..c6462e7e 100644
--- a/packages/db/test/schema.test.ts
+++ b/packages/db/test/schema.test.ts
@@ -384,4 +384,25 @@ describe.skipIf(skip)("schema migrations (real PG)", () => {
     expect(up.error).toBeUndefined();
     expect(await tables()).toEqual(EXPECTED_TABLES);
   }, 60_000);
+
+  it("迁移器只内省目标 schema:别处同时建删 schema 不影响(并行测试共用库)", async () => {
+    const migrator = createMigrator(db, { migrationTableSchema: schema });
+    let stop = false;
+    const churn = (async () => {
+      for (let i = 0; !stop && i < 200; i++) {
+        const tmp = `${schema}_churn_${i}`;
+        await sql.raw(`create schema ${tmp}; create table ${tmp}.t (id serial primary key)`).execute(admin);
+        await sql.raw(`drop schema ${tmp} cascade`).execute(admin);
+      }
+    })();
+    try {
+      for (let i = 0; i < 20; i++) {
+        const r = await migrator.migrateToLatest();
+        expect(r.error).toBeUndefined();
+      }
+    } finally {
+      stop = true;
+      await churn;
+    }
+  }, 60_000);
 });