fix(db): migrator introspects only the target schema

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 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed 2026-09-28 17:15:04 +08:00
1 parent 082c556a67
commit f6063af12d
3 files changed
+70 -2

No files matched your search

+1
View File
@@ -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,
+48 -2
View File
@@ -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 } : {}),
});
+21
View File
@@ -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);
});