pn-new-crm/scripts/db-migrate.ts

/**
 * Production migration runner.
 *
 * Why this exists (and why `drizzle-kit migrate` isn't enough):
 *
 *   - Drizzle's bundled `migrate()` wraps every migration in a single
 *     transaction. Postgres forbids `CREATE INDEX CONCURRENTLY` inside
 *     a transaction (raises 25001) — so any migration containing
 *     CONCURRENTLY silently aborts or, worse, leaves the migration
 *     marked applied with the index missing. `0052_audit_critical_fixes.sql`
 *     ships six CONCURRENTLY composite indexes today and they never
 *     landed in prod.
 *
 *   - `drizzle-kit push` skips DDL the kit can't infer from the schema —
 *     e.g. CHECK constraints, partial unique indexes, the berth-pdf
 *     circular FK. push-only deployments diverge from migration-tracked
 *     truth.
 *
 * This script:
 *   1. Reads migrations in journal order from `src/lib/db/migrations`.
 *   2. Tracks applied state in `drizzle.__drizzle_migrations` (matching
 *      Drizzle's schema so other tooling sees the same source of truth).
 *   3. For each pending migration: splits on `--> statement-breakpoint`,
 *      classifies each statement as concurrency-safe (CREATE INDEX
 *      CONCURRENTLY / REINDEX CONCURRENTLY → outside tx) or
 *      transactional (everything else → batched in one tx per migration).
 *   4. Records hash + when-applied so re-runs are no-ops.
 *
 * Modes:
 *   `pnpm db:migrate`           — apply pending migrations
 *   `pnpm db:migrate:status`    — show pending vs applied without applying
 *   `pnpm db:migrate:baseline`  — mark every migration as applied without
 *                                 running it. Use ONCE per environment when
 *                                 the schema was bootstrapped via `db:push`
 *                                 (dev + the original prod cutover). After
 *                                 baseline, all future migrations go through
 *                                 `db:migrate` and are tracked in
 *                                 `__drizzle_migrations`.
 */

import 'dotenv/config';
import { createHash } from 'node:crypto';
import { readFileSync, readdirSync } from 'node:fs';
import { join } from 'node:path';
import postgres from 'postgres';

const STATEMENT_BREAKPOINT = '--> statement-breakpoint';
const MIGRATIONS_DIR = join(process.cwd(), 'src/lib/db/migrations');
const SCHEMA_NAME = 'drizzle';
const TABLE_NAME = '__drizzle_migrations';

interface JournalEntry {
  idx: number;
  version: string;
  when: number;
  tag: string;
  breakpoints: boolean;
}

interface Journal {
  version: string;
  dialect: string;
  entries: JournalEntry[];
}

interface MigrationFile {
  tag: string;
  /** Folder millis from journal `when` — Drizzle uses this as the
   *  primary key in `__drizzle_migrations`. */
  folderMillis: number;
  /** Full file contents. */
  sql: string;
  /** SHA-256 hex of the raw file for re-application detection. */
  hash: string;
}

interface Statement {
  /** Raw SQL text (trimmed). */
  sql: string;
  /** True when the statement must execute outside a transaction. */
  needsAutocommit: boolean;
}

function isConcurrencyDDL(sql: string): boolean {
  const head = sql
    .replace(/^\s*--.*$/gm, '')
    .trim()
    .toUpperCase();
  return (
    /\bCREATE\s+INDEX\s+CONCURRENTLY\b/.test(head) ||
    /\bREINDEX\s+\w*\s*CONCURRENTLY\b/.test(head) ||
    /\bDROP\s+INDEX\s+CONCURRENTLY\b/.test(head)
  );
}

function readMigrations(): MigrationFile[] {
  const journal = JSON.parse(
    readFileSync(join(MIGRATIONS_DIR, 'meta', '_journal.json'), 'utf8'),
  ) as Journal;

  const files = readdirSync(MIGRATIONS_DIR).filter((f) => f.endsWith('.sql'));
  const byTag = new Map(files.map((f) => [f.replace(/\.sql$/, ''), f]));

  return journal.entries.map((entry) => {
    const filename = byTag.get(entry.tag);
    if (!filename) {
      throw new Error(`Migration ${entry.tag} in journal but ${entry.tag}.sql not on disk`);
    }
    const sql = readFileSync(join(MIGRATIONS_DIR, filename), 'utf8');
    const hash = createHash('sha256').update(sql).digest('hex');
    return { tag: entry.tag, folderMillis: entry.when, sql, hash };
  });
}

function splitStatements(sql: string): Statement[] {
  // Drizzle inserts `--> statement-breakpoint` between every statement
  // when `breakpoints: true` in drizzle.config. We split on those AND
  // strip trailing semicolons. Anything before the first breakpoint
  // counts too.
  const parts = sql.split(STATEMENT_BREAKPOINT);
  const out: Statement[] = [];
  for (const part of parts) {
    const trimmed = part.trim();
    if (!trimmed || trimmed.startsWith('--')) {
      // Comment-only chunks (pre-breakpoint header etc.) — skip if
      // they have no executable SQL.
      const nonComment = trimmed
        .split('\n')
        .filter((line) => !line.trim().startsWith('--') && line.trim().length > 0);
      if (nonComment.length === 0) continue;
    }
    out.push({ sql: trimmed, needsAutocommit: isConcurrencyDDL(trimmed) });
  }
  return out;
}

async function ensureMigrationsTable(sql: postgres.Sql): Promise<void> {
  await sql.unsafe(`CREATE SCHEMA IF NOT EXISTS "${SCHEMA_NAME}"`);
  await sql.unsafe(`
    CREATE TABLE IF NOT EXISTS "${SCHEMA_NAME}"."${TABLE_NAME}" (
      id SERIAL PRIMARY KEY,
      hash text NOT NULL,
      created_at bigint
    )
  `);
}

async function getAppliedHashes(sql: postgres.Sql): Promise<Set<string>> {
  const rows = await sql.unsafe<{ hash: string }[]>(
    `SELECT hash FROM "${SCHEMA_NAME}"."${TABLE_NAME}"`,
  );
  return new Set(rows.map((r) => r.hash));
}

async function applyMigration(sql: postgres.Sql, migration: MigrationFile): Promise<void> {
  const statements = splitStatements(migration.sql);
  if (statements.length === 0) {
    console.log(`  [${migration.tag}] no executable statements, skipping`);
    return;
  }

  const autocommit = statements.filter((s) => s.needsAutocommit);
  const transactional = statements.filter((s) => !s.needsAutocommit);

  // Transactional batch first — schema changes that CONCURRENTLY ops
  // depend on (e.g. column adds before CREATE INDEX) need to exist
  // before the index build runs. Drizzle migrations are written in
  // this order; we preserve it within each phase.
  if (transactional.length > 0) {
    await sql.begin(async (tx) => {
      for (const stmt of transactional) {
        await tx.unsafe(stmt.sql);
      }
    });
  }

  // CONCURRENTLY ops run one at a time, each as its own implicit tx.
  // No `BEGIN`/`COMMIT` wrapping — postgres-js's `sql.unsafe` runs
  // each call as an independent transaction.
  for (const stmt of autocommit) {
    await sql.unsafe(stmt.sql);
  }

  // Record the migration as applied. created_at mirrors Drizzle's own
  // schema so `drizzle-kit migrate` (if ever invoked) sees the same
  // state we wrote.
  await sql.unsafe(
    `INSERT INTO "${SCHEMA_NAME}"."${TABLE_NAME}" (hash, created_at) VALUES ($1, $2)`,
    [migration.hash, migration.folderMillis],
  );
}

async function main(): Promise<void> {
  const url = process.env.DATABASE_URL;
  if (!url) {
    console.error('DATABASE_URL must be set');
    process.exit(1);
  }

  const mode = process.argv[2] ?? 'apply';
  if (!['apply', 'status', 'baseline'].includes(mode)) {
    console.error(`Unknown mode: ${mode}. Use 'apply' (default), 'status', or 'baseline'.`);
    process.exit(1);
  }

  const sql = postgres(url, { max: 1, prepare: false });

  try {
    await ensureMigrationsTable(sql);
    const applied = await getAppliedHashes(sql);
    const migrations = readMigrations();
    const pending = migrations.filter((m) => !applied.has(m.hash));

    if (mode === 'status') {
      console.log(`Applied: ${applied.size}`);
      console.log(`Pending: ${pending.length}`);
      if (pending.length > 0) {
        console.log('');
        console.log('Pending migrations:');
        for (const m of pending) {
          const statements = splitStatements(m.sql);
          const conc = statements.filter((s) => s.needsAutocommit).length;
          const tx = statements.length - conc;
          console.log(`  ${m.tag} — ${tx} transactional + ${conc} concurrency-safe`);
        }
      }
      return;
    }

    if (mode === 'baseline') {
      if (pending.length === 0) {
        console.log('All migrations already tracked. Nothing to baseline.');
        return;
      }
      console.log(
        `Baselining ${pending.length} migration${
          pending.length === 1 ? '' : 's'
        } as applied without running them.`,
      );
      console.log(
        'This is correct ONLY when the schema is already in place (e.g. created via db:push).',
      );
      for (const m of pending) {
        await sql.unsafe(
          `INSERT INTO "${SCHEMA_NAME}"."${TABLE_NAME}" (hash, created_at) VALUES ($1, $2)`,
          [m.hash, m.folderMillis],
        );
        console.log(`  → ${m.tag} marked as applied`);
      }
      console.log(`Done. ${pending.length} baselined.`);
      return;
    }

    if (pending.length === 0) {
      console.log('No pending migrations.');
      return;
    }

    console.log(`Applying ${pending.length} migration${pending.length === 1 ? '' : 's'}...`);
    for (const m of pending) {
      const statements = splitStatements(m.sql);
      const conc = statements.filter((s) => s.needsAutocommit).length;
      console.log(`  → ${m.tag} (${statements.length} statements, ${conc} CONCURRENTLY)`);
      await applyMigration(sql, m);
    }
    console.log(`Done. ${pending.length} applied.`);
  } finally {
    await sql.end({ timeout: 5 });
  }
}

main().catch((err) => {
  console.error('Migration failed:', err);
  process.exit(1);
});