feat(appkit): add appkit db introspect

packages/appkit/src/database/migrations/apply.ts

@@ -147,7 +147,7 @@ export class MigrationBaselineMismatchError extends Error {
   constructor(schemaName: string, problems: readonly string[]) {
     super(
       `Cannot adopt ${BASELINE_TAG}: schema "${schemaName}" has no recorded migrations and only part of the baseline: ${problems.join("; ")}. ` +
-        "Create the missing tables and columns (or drop the partial ones), or regenerate the baseline against this database.",
+        "Create the missing tables and columns (or drop the partial ones), or regenerate the baseline with `appkit db introspect` against this database.",
     );
     this.name = "MigrationBaselineMismatchError";
     this.problems = Object.freeze([...problems]);
@@ -575,6 +575,124 @@ function duration(ms: number): string {
   return `${ms}ms`;
 }
 
+/** How the branch's recorded migrations differ from the committed ones. */
+interface MigrationDrift {
+  /** Committed, not recorded, newer than every recorded one. */
+  readonly pending: readonly string[];
+  /** Committed, not recorded, older than the last recorded one. */
+  readonly stranded: readonly string[];
+  /** Recorded rows no committed migration matches: the branch is ahead of the checkout. */
+  readonly unknown: number;
+}
+
+function driftOf(
+  history: ReturnType<typeof reconcileMigrations>,
+): MigrationDrift {
+  return {
+    pending: tags(history.pending),
+    stranded: tags(history.stranded),
+    unknown: history.unknown.length,
+  };
+}
+
+function committedMigrations(folder: string): MigrationFile[] {
+  const journal = readJournal(folder);
+  return journal ? readMigrations(folder, journal) : [];
+}
+
+/** Read without the lock; markMigrationsApplied checks again under it. */
+export async function readMigrationDrift(
+  client: QueryClient,
+  options: ApplyMigrationsOptions,
+): Promise<MigrationDrift> {
+  return driftOf(
+    reconcileMigrations(
+      committedMigrations(options.migrationsFolder),
+      await readRecordedMigrations(client, options.schemaName),
+    ),
+  );
+}
+
+/** Why the branch's history is not the committed one plus toRecord; undefined when it is. */
+function driftReason(
+  drift: MigrationDrift,
+  toRecord: readonly string[],
+): string | undefined {
+  const notApplied = drift.pending.filter((tag) => !toRecord.includes(tag));
+  const recorded = toRecord.filter((tag) => !drift.pending.includes(tag));
+  const reasons = [
+    drift.unknown > 0
+      ? `it records ${drift.unknown} migration${drift.unknown === 1 ? "" : "s"} this checkout does not have`
+      : "",
+    drift.stranded.length > 0 ? `it skipped ${drift.stranded.join(", ")}` : "",
+    notApplied.length > 0 ? `it has not applied ${notApplied.join(", ")}` : "",
+    recorded.length > 0 ? `it already records ${recorded.join(", ")}` : "",
+  ].filter(Boolean);
+  return reasons.length > 0 ? reasons.join("; ") : undefined;
+}
+
+/**
+ * Record migrations as applied without running them: the branch already holds
+ * what they describe. Under the migration lock, and only when they are exactly
+ * the branch's pending migrations, so DDL the branch never ran is not recorded.
+ */
+export async function markMigrationsApplied(
+  pool: Pick<Pool, "connect">,
+  options: ApplyMigrationsOptions,
+  toRecord: readonly string[],
+): Promise<void> {
+  const files = committedMigrations(options.migrationsFolder);
+  const missing = toRecord.filter(
+    (tag) => !files.some((file) => file.tag === tag),
+  );
+  if (missing.length > 0) {
+    throw new MigrationFilesError(
+      `Cannot record ${missing.join(", ")} as applied: not in ${options.migrationsFolder}/meta/_journal.json.`,
+    );
+  }
+  const { schemaName } = options;
+  const client = await pool.connect();
+  const reset: string[] = [];
+  let locked = false;
+  try {
+    await prepareMigrationSession(client, reset);
+    await acquireMigrationLock(
+      client,
+      schemaName,
+      options.lockWaitMs ?? LOCK_WAIT_MS,
+    );
+    locked = true;
+    const history = reconcileMigrations(
+      files,
+      await readRecordedMigrations(client, schemaName),
+    );
+    const reason = driftReason(driftOf(history), toRecord);
+    if (reason) {
+      throw new Error(
+        `Schema "${schemaName}" changed while this ran (${reason}), so nothing was recorded.`,
+      );
+    }
+    await inTransaction(client, async () => {
+      await ensureJournalTable(client, schemaName);
+      for (const file of history.pending) {
+        await recordMigration(client, schemaName, file);
+      }
+    });
+  } catch (error) {
+    if (!(error instanceof MigrationLockWaitError)) throw error;
+    throw new Error(
+      `${error.holder ? `Session ${error.holder}` : "Another session"} held the migration lock of schema "${schemaName}" for over ${duration(options.lockWaitMs ?? LOCK_WAIT_MS)}, so nothing was recorded. Rerun once it finishes.`,
+      { cause: error },
+    );
+  } finally {
+    await releaseMigrationClient(
+      client,
+      locked ? migrationLockKey(schemaName) : undefined,
+      reset,
+    );
+  }
+}
+
 /**
  * Apply every committed migration the database has not recorded yet.
  *

packages/appkit/src/database/migrations/tests/apply.pg.test.ts

@@ -16,8 +16,11 @@ import {
   applyPendingMigrations,
   MigrationApplyError,
   MigrationHistoryError,
+  markMigrationsApplied,
   migrationLockKey,
+  readMigrationDrift,
 } from "../apply";
+import { MigrationFilesError } from "../paths";
 import { hashOf, removeMigrationFolders, writeMigrations } from "./fixtures";
 
 // A real server, for what a fake cannot show: transactions, enum rules, locks
@@ -405,4 +408,130 @@ describe.skipIf(!url)("applyPendingMigrations against Postgres", () => {
     });
     expect(await columns("notes")).toEqual(["id", "body"]);
   });
+
+  describe("recording migrations the branch already has", () => {
+    const withTags = () =>
+      writeMigrations([
+        {
+          tag: "0000_notes",
+          when: 1000,
+          sql: [`CREATE TABLE ${q("notes")} ("id" integer PRIMARY KEY);`],
+        },
+        {
+          tag: "0001_introspect",
+          when: 3000,
+          sql: [`ALTER TABLE ${q("notes")} ADD COLUMN "body" text;`],
+        },
+      ]);
+    /** The branch applied 0000_notes and recorded a 2000 migration this checkout lacks. */
+    const aheadOfCheckout = async (folder: string) => {
+      await applyPendingMigrations(newPool(), {
+        ...options(folder),
+        migrationsFolder: notes(),
+      });
+      await admin.query(
+        `delete from ${q("__drizzle_migrations")} where created_at <> 1000`,
+      );
+      await admin.query(
+        `insert into ${q("__drizzle_migrations")} (hash, created_at) values ('teammate', 2000)`,
+      );
+    };
+
+    test("the drift lists unrecorded committed migrations and recorded ones the checkout lacks", async () => {
+      const folder = withTags();
+      expect(await readMigrationDrift(newPool(), options(folder))).toEqual({
+        pending: ["0000_notes", "0001_introspect"],
+        stranded: [],
+        unknown: 0,
+      });
+
+      await aheadOfCheckout(folder);
+
+      expect(await readMigrationDrift(newPool(), options(folder))).toEqual({
+        pending: ["0001_introspect"],
+        stranded: [],
+        unknown: 1,
+      });
+    });
+
+    test("records exactly the pending migrations, without running them", async () => {
+      await admin.query(`create schema "${schema}"`);
+      const folder = withTags();
+
+      await markMigrationsApplied(newPool(), options(folder), [
+        "0000_notes",
+        "0001_introspect",
+      ]);
+
+      expect(await recorded()).toEqual([
+        { hash: hashOf(folder, "0000_notes"), createdAt: 1000 },
+        { hash: hashOf(folder, "0001_introspect"), createdAt: 3000 },
+      ]);
+      expect(await columns("notes")).toEqual([]);
+    });
+
+    test.each([
+      [
+        "records a migration the checkout lacks",
+        aheadOfCheckout,
+        ["0001_introspect"],
+        "it records 1 migration this checkout does not have",
+      ],
+      [
+        "has not applied an earlier one",
+        async () => {
+          await admin.query(`create schema "${schema}"`);
+        },
+        ["0001_introspect"],
+        "it has not applied 0000_notes",
+      ],
+    ])(
+      "refuses when the branch %s, and records nothing",
+      async (_label, arrange, toRecord, reason) => {
+        const folder = withTags();
+        await arrange(folder);
+        const before = await recorded().catch(() => []);
+
+        await expect(
+          markMigrationsApplied(newPool(), options(folder), toRecord),
+        ).rejects.toThrow(
+          `Schema "${schema}" changed while this ran (${reason}), so nothing was recorded.`,
+        );
+        expect(await recorded().catch(() => [])).toEqual(before);
+      },
+    );
+
+    test("a tag missing from the journal fails before connecting", async () => {
+      const pool = newPool();
+      const connect = vi.spyOn(pool, "connect");
+
+      await expect(
+        markMigrationsApplied(pool, options(withTags()), ["0002_introspect"]),
+      ).rejects.toBeInstanceOf(MigrationFilesError);
+      expect(connect).not.toHaveBeenCalled();
+    });
+
+    test("the wait for the migration lock has a limit, and nothing is recorded", async () => {
+      await admin.query(`create schema "${schema}"`);
+      const holder = await admin.connect();
+      held.push(holder);
+      await holder.query("set application_name = 'another-deploy'");
+      await holder.query("select pg_advisory_lock($1::bigint)", [
+        migrationLockKey(schema),
+      ]);
+
+      await expect(
+        markMigrationsApplied(
+          newPool(),
+          { ...options(withTags()), lockWaitMs: 300 },
+          ["0000_notes", "0001_introspect"],
+        ),
+      ).rejects.toThrow(
+        new RegExp(
+          `^Session pid \\d+ \\(another-deploy, idle\\) held the migration lock of schema "${schema}" for over 300ms, so nothing was recorded\\. Rerun once it finishes\\.$`,
+        ),
+      );
+      expect(await recorded().catch(() => [])).toEqual([]);
+    });
+  });
 });

packages/appkit/src/database/schema-builder/define-schema.ts

@@ -72,9 +72,27 @@ function assertName(value: string, label: string): void {
   }
 }
 
+/** Table names defineSchema() refuses. */
+export function isReservedTableName(name: string): boolean {
+  return (
+    !name ||
+    RESERVED_OBJECT_KEYS.has(name) ||
+    RESERVED_DATABASE_EXPORT_KEYS.has(name)
+  );
+}
+
+/** Column keys that collide with the runtime metadata on a table handle. */
+function collidesWithMetadata(key: string): boolean {
+  return key.startsWith("$") || key === "and" || key === "or";
+}
+
+/** Column names defineSchema() refuses. */
+export function isReservedColumnName(name: string): boolean {
+  return !name || RESERVED_OBJECT_KEYS.has(name) || collidesWithMetadata(name);
+}
+
 function assertTableName(value: string): void {
-  assertName(value, "Table name");
-  if (RESERVED_DATABASE_EXPORT_KEYS.has(value)) {
+  if (isReservedTableName(value)) {
     throw new SchemaBuildError(`Table name "${value}" is reserved`);
   }
 }
@@ -120,7 +138,7 @@ function declareTable<C extends Record<string, ColumnBuilder>>(
   const handle = Object.create(null) as Record<string, ColumnRef>;
   for (const [key, column] of Object.entries(columns)) {
     assertName(key, `Column name on table "${name}"`);
-    if (key.startsWith("$") || key === "and" || key === "or") {
+    if (collidesWithMetadata(key)) {
       throw new SchemaBuildError(
         `Column "${name}.${key}" collides with AppKit runtime metadata`,
       );
@@ -344,7 +362,7 @@ export function assertFinalizedSchema(value: unknown): asserts value is Schema {
  * the deployed app's service principal can create and own a schema of its
  * own, but has no CREATE on `public`.
  */
-const DEFAULT_SCHEMA_NAME = "app";
+export const DEFAULT_SCHEMA_NAME = "app";
 
 /**
  * Compile one declared schema. The returned type keeps the table names the

packages/appkit/src/database/tooling/index.ts

@@ -7,4 +7,9 @@
  * the same AppKit release should import it.
  */
 export { createDbContext, type DbContext } from "./context";
+export {
+  type IntrospectOptions,
+  type IntrospectReport,
+  runIntrospect,
+} from "./introspect";
 export { type MigrateOptions, type MigrateReport, runMigrate } from "./migrate";