feat: add PostgreSQL backend

Mirrors the node:sqlite sub-backend structure with full migration support. Uses native pg types (UUID, JSONB, TIMESTAMPTZ) and $1/$2 parameterisation via internal ? placeholders converted at execution time. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-20 02:00:10 +00:00
parent dd8d31633f
commit 4521274a27
22 changed files with 1273 additions and 1 deletions
@@ -0,0 +1,126 @@
+import type {
+  AppliedMigration,
+  AvailableMigration,
+  Migrator,
+} from "../types/Migrator.js";
+import { init, migrations } from "./migrations/index.js";
+import { pgToDate } from "./cast.js";
+import { WrappedPool } from "./pool.js";
+import { sql, toPositional } from "./sql.js";
+
+export class PostgresMigrator implements Migrator {
+  #pool: WrappedPool;
+
+  constructor(pool: WrappedPool) {
+    this.#pool = pool;
+  }
+
+  async #listAppliedMigrations(): Promise<
+    { id: number; name: string; applied_at: string }[] | null
+  > {
+    const client = await this.#pool._pool.connect();
+    try {
+      const existsResult = await client.query<{ exists: boolean }>(
+        `SELECT EXISTS (
+          SELECT 1 FROM information_schema.tables
+          WHERE table_schema = 'public' AND table_name = '_migrations'
+        ) AS "exists"`
+      );
+      if (!existsResult.rows[0]?.exists) return null;
+
+      const result = await client.query<{
+        id: number;
+        name: string;
+        applied_at: Date;
+      }>(
+        `SELECT "id", "name", "applied_at" FROM "_migrations" ORDER BY "id" ASC`
+      );
+      return result.rows.map((x) => ({
+        ...x,
+        applied_at: pgToDate(x.applied_at),
+      }));
+    } finally {
+      client.release();
+    }
+  }
+
+  async listAppliedMigrations(): Promise<AppliedMigration[]> {
+    return (await this.#listAppliedMigrations()) ?? [];
+  }
+
+  listAvailableMigrations(): AvailableMigration[] {
+    return migrations.map((m) => ({ id: m.id, name: m.name }));
+  }
+
+  async migrateTo(id: number): Promise<void> {
+    const target = migrations.find((x) => x.id === id);
+    if (!target) throw new Error(`No known migration with id ${id}`);
+
+    let current = await this.#listAppliedMigrations();
+    if (!current) {
+      const client = await this.#pool._pool.connect();
+      try {
+        await client.query(init);
+      } finally {
+        client.release();
+      }
+      current = [];
+    }
+
+    for (const { id, name } of current) {
+      const migration = migrations.find((x) => x.id === id);
+      if (!migration)
+        throw new Error(`Applied migration ${id} (${name}) not known`);
+      if (migration.name !== name)
+        throw new Error(
+          `Applied migration ${id} (${name}) has a different name from expected (${migration.name})`
+        );
+    }
+
+    const start =
+      migrations.findIndex((x) => x.id === current.at(-1)?.id) + 1;
+    const end = migrations.indexOf(target) + 1;
+
+    if (end < start) {
+      throw new Error(
+        `Cannot migrate backward, at ${current.at(-1)?.id ?? 0}, going to ${target.id}`
+      );
+    }
+
+    const toApply = migrations.slice(start, end);
+
+    if (!toApply.length) {
+      console.log("Nothing to do");
+      return;
+    }
+
+    for (const migration of toApply) {
+      console.log(`Applying migration ${migration.id} (${migration.name})`);
+      const client = await this.#pool._pool.connect();
+      try {
+        await client.query("BEGIN");
+        for (const part of migration.parts) {
+          console.log(`- Applying part ${part.id} (${part.name})`);
+          if ("sql" in part) {
+            await client.query(part.sql);
+          } else {
+            await part.apply(this.#pool);
+          }
+        }
+        const recordSql = sql`
+          INSERT INTO "_migrations"("id", "name")
+          VALUES (${{ int: migration.id }}, ${{ text: migration.name }})
+        `;
+        await client.query(toPositional(recordSql._sql), recordSql._vars);
+        await client.query("COMMIT");
+      } catch (e) {
+        await client.query("ROLLBACK");
+        throw e;
+      } finally {
+        client.release();
+      }
+    }
+
+    console.log("Done migrating database");
+  }
+}