Repository Public
1/** Applies db/migrations/*.sql in filename order, once each, under a session advisory lock. */2import fs from 'node:fs';3import path from 'node:path';4import { fileURLToPath } from 'node:url';5import pg from 'pg';6import { sha256Hex } from './canonical.js';7import { LOCK_CLASS, LOCK_MIGRATE } from './db.js';8import { KernelError, KernelStorageError } from './errors.js';910/** Finds `db/migrations` by walking up from this file (works from src/ and dist/); env AOS_MIGRATIONS_DIR wins. */11export function findMigrationsDir(): string {12 if (process.env.AOS_MIGRATIONS_DIR) return path.resolve(process.env.AOS_MIGRATIONS_DIR);13 let dir = path.dirname(fileURLToPath(import.meta.url));14 for (let i = 0; i < 8; i++) {15 const cand = path.join(dir, 'db', 'migrations');16 if (fs.existsSync(cand)) return cand;17 const up = path.dirname(dir);18 if (up === dir) break;19 dir = up;20 }21 throw new KernelError('migrate: db/migrations not found (set AOS_MIGRATIONS_DIR)');22}2324export interface MigrateResult {25 applied: string[];26 skipped: string[];27}2829export async function migrate(databaseUrl?: string, opts: { dir?: string } = {}): Promise<MigrateResult> {30 const url = databaseUrl ?? process.env.DATABASE_URL;31 if (!url) throw new KernelError('migrate: no databaseUrl / DATABASE_URL');32 const dir = opts.dir ?? findMigrationsDir();33 const files = fs.readdirSync(dir).filter((f) => /^\d+_.*\.sql$/.test(f)).sort();34 const client = new pg.Client({ connectionString: url });35 const res: MigrateResult = { applied: [], skipped: [] };36 try {37 await client.connect();38 // Session lock: concurrent migrators (server + worker + tests) queue here.39 await client.query('SELECT pg_advisory_lock($1, $2)', [LOCK_CLASS, LOCK_MIGRATE]);40 await client.query(41 `CREATE TABLE IF NOT EXISTS schema_migrations (name text PRIMARY KEY, sha256 text NOT NULL, applied_at timestamptz NOT NULL DEFAULT now())`,42 );43 const done = new Set((await client.query('SELECT name FROM schema_migrations')).rows.map((r: { name: string }) => r.name));44 for (const f of files) {45 if (done.has(f)) {46 res.skipped.push(f);47 continue;48 }49 const sql = fs.readFileSync(path.join(dir, f), 'utf8');50 const sha = sha256Hex(sql);51 await client.query('BEGIN');52 try {53 await client.query(sql);54 await client.query('INSERT INTO schema_migrations (name, sha256) VALUES ($1, $2)', [f, sha]);55 await client.query('COMMIT');56 } catch (err) {57 await client.query('ROLLBACK').catch(() => undefined);58 throw new KernelStorageError(`migration ${f} failed: ${(err as Error).message}`, { cause: err });59 }60 res.applied.push(f);61 }62 return res;63 } catch (err) {64 if (err instanceof KernelError) throw err;65 throw new KernelStorageError(`migrate failed: ${(err as Error).message}`, { cause: err });66 } finally {67 // Closing the session releases the advisory lock.68 await client.end().catch(() => undefined);69 }70}71