feat: db pool + migration runner with idempotency

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

lib/db/migrate.js

@@ -0,0 +1,49 @@
+import fs from 'node:fs/promises';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { pool } from './pool.js';
+import { log } from '../log.js';
+
+const __dirname = path.dirname(fileURLToPath(import.meta.url));
+const MIG_DIR = path.join(__dirname, 'migrations');
+
+export async function migrateUp() {
+  const client = await pool.connect();
+  try {
+    await client.query(`
+      CREATE TABLE IF NOT EXISTS schema_migrations (
+        name text PRIMARY KEY,
+        applied_at timestamptz NOT NULL DEFAULT now()
+      );
+    `);
+
+    const files = (await fs.readdir(MIG_DIR)).filter(f => f.endsWith('.sql')).sort();
+    const { rows } = await client.query('SELECT name FROM schema_migrations');
+    const applied = new Set(rows.map(r => r.name));
+
+    for (const file of files) {
+      if (applied.has(file)) continue;
+      const sql = await fs.readFile(path.join(MIG_DIR, file), 'utf8');
+      log.info({ migration: file }, 'applying migration');
+      await client.query('BEGIN');
+      try {
+        await client.query(sql);
+        await client.query(
+          'INSERT INTO schema_migrations(name) VALUES ($1)', [file]
+        );
+        await client.query('COMMIT');
+      } catch (err) {
+        await client.query('ROLLBACK');
+        throw err;
+      }
+    }
+  } finally {
+    client.release();
+  }
+}
+
+if (process.argv[2] === 'up') {
+  migrateUp().then(() => process.exit(0)).catch((e) => {
+    log.error(e); process.exit(1);
+  });
+}

lib/db/pool.js

@@ -0,0 +1,8 @@
+import pg from 'pg';
+import 'dotenv/config';
+
+export const pool = new pg.Pool({
+  connectionString: process.env.DATABASE_URL,
+  max: 10,
+  idleTimeoutMillis: 30_000
+});

tests/db/migrate.test.js

@@ -0,0 +1,28 @@
+import { describe, it, expect, beforeAll } from 'vitest';
+import { resetDb, withClient } from '../helpers/db.js';
+import { migrateUp } from '../../lib/db/migrate.js';
+
+describe('migrate', () => {
+  beforeAll(async () => { await resetDb(); });
+
+  it('creates schema_migrations table on first run', async () => {
+    await migrateUp();
+    await withClient(async (c) => {
+      const { rows } = await c.query(
+        `SELECT to_regclass('public.schema_migrations') AS t;`
+      );
+      expect(rows[0].t).toBe('schema_migrations');
+    });
+  });
+
+  it('is idempotent — second run is a no-op', async () => {
+    await migrateUp();
+    await migrateUp();
+    await withClient(async (c) => {
+      const { rows } = await c.query(
+        `SELECT count(*)::int AS n FROM schema_migrations;`
+      );
+      expect(rows[0].n).toBeGreaterThanOrEqual(0);
+    });
+  });
+});

tests/helpers/db.js

@@ -0,0 +1,16 @@
+import { pool } from '../../lib/db/pool.js';
+
+export async function resetDb() {
+  await pool.query(`
+    DROP SCHEMA IF EXISTS public CASCADE;
+    CREATE SCHEMA public;
+    CREATE EXTENSION IF NOT EXISTS pgcrypto;
+    CREATE EXTENSION IF NOT EXISTS vector;
+  `);
+}
+
+export async function withClient(fn) {
+  const client = await pool.connect();
+  try { await fn(client); }
+  finally { client.release(); }
+}

tests/helpers/setup.js

@@ -0,0 +1 @@
+import 'dotenv/config';