ApiCore/ApiCore-Backend/test/utils/migration-harness.ts

import { execFile } from 'node:child_process';
import { randomBytes } from 'node:crypto';
import * as fs from 'node:fs/promises';
import * as path from 'node:path';
import { promisify } from 'node:util';
import { Client } from 'pg';

const execFileAsync = promisify(execFile);

const BACKEND_ROOT = path.resolve(__dirname, '../..');
const MIGRATIONS_DIR = path.join(BACKEND_ROOT, 'prisma', 'migrations');
const SCHEMA_FILE = path.join(BACKEND_ROOT, 'prisma', 'schema.prisma');
const SCRATCH_ROOT = path.join(BACKEND_ROOT, '.tmp-migrations');
const PRISMA_CLI = path.join(
  BACKEND_ROOT,
  'node_modules',
  'prisma',
  'build',
  'index.js',
);
const DEPLOY_TIMEOUT_MS = 100_000;
/**
 * `prisma migrate deploy` takes pg_advisory_lock(72707369) and gives up after
 * 10 s. Advisory locks are per DATABASE, not per schema, so scratch schemas
 * of parallel migration suites (honey-ledger, t4, t6) collide on it (P1002 in
 * CI). The harness therefore runs one deploy at a time across jest workers
 * with its own, blocking advisory lock; the session lock disappears with the
 * connection should a worker die.
 */
const HARNESS_DEPLOY_LOCK_KEY = 72707370;
const SCHEMA_NAME_PATTERN = /^mig_[0-9a-f]{12}$/;

export interface ScratchDatabase {
  readonly schema: string;
  query<T = Record<string, unknown>>(
    sql: string,
    params?: unknown[],
  ): Promise<T[]>;
  /** Applies every migration up to and including the named one. */
  migrateThrough(migrationName: string): Promise<void>;
  drop(): Promise<void>;
}

function baseConnectionString(): string {
  const url = process.env.DATABASE_URL;
  if (!url) {
    throw new Error('DATABASE_URL is not set');
  }
  return url;
}

function withSchema(connectionString: string, schema: string): string {
  const url = new URL(connectionString);
  url.searchParams.set('schema', schema);
  return url.toString();
}

function withoutSchemaParam(connectionString: string): string {
  const url = new URL(connectionString);
  url.searchParams.delete('schema');
  return url.toString();
}

async function listMigrationNames(): Promise<string[]> {
  const entries = await fs.readdir(MIGRATIONS_DIR, { withFileTypes: true });
  return entries
    .filter((entry) => entry.isDirectory())
    .map((entry) => entry.name)
    .sort();
}

/**
 * Creates a fresh Postgres schema and migrates it up to, but not including,
 * `beforeMigration`. The migration folders are copied below
 * `.tmp-migrations/` inside the backend so that the prisma config resolves
 * the installed `prisma` package.
 */
export async function createScratchDatabase(
  beforeMigration: string,
): Promise<ScratchDatabase> {
  const names = await listMigrationNames();
  const target = names.indexOf(beforeMigration);
  if (target < 0) {
    throw new Error(`Unknown migration: ${beforeMigration}`);
  }

  const id = randomBytes(6).toString('hex');
  const schema = `mig_${id}`;
  if (!SCHEMA_NAME_PATTERN.test(schema)) {
    throw new Error('Unexpected scratch schema name');
  }
  const workDir = path.join(SCRATCH_ROOT, id);
  const scratchUrl = withSchema(baseConnectionString(), schema);

  const admin = new Client({
    connectionString: withoutSchemaParam(baseConnectionString()),
  });
  await admin.connect();
  await admin.query(`CREATE SCHEMA "${schema}"`);

  await fs.mkdir(workDir, { recursive: true });
  await fs.copyFile(
    path.join(MIGRATIONS_DIR, 'migration_lock.toml'),
    path.join(workDir, 'migration_lock.toml'),
  );
  // The URL reaches the config through the environment, never through disk.
  await fs.writeFile(
    path.join(workDir, 'prisma.config.ts'),
    [
      "import { defineConfig } from 'prisma/config';",
      '',
      'export default defineConfig({',
      `  schema: ${JSON.stringify(SCHEMA_FILE)},`,
      `  migrations: { path: ${JSON.stringify(workDir)} },`,
      "  datasource: { url: process.env['MIGRATION_HARNESS_URL'] },",
      '});',
      '',
    ].join('\n'),
  );

  const copied = new Set<string>();
  const deploy = async (upToExclusive: number): Promise<void> => {
    for (const name of names.slice(0, upToExclusive)) {
      if (!copied.has(name)) {
        await fs.cp(path.join(MIGRATIONS_DIR, name), path.join(workDir, name), {
          recursive: true,
        });
        copied.add(name);
      }
    }
    await admin.query('SELECT pg_advisory_lock($1)', [HARNESS_DEPLOY_LOCK_KEY]);
    try {
      await execFileAsync(
        process.execPath,
        [
          PRISMA_CLI,
          'migrate',
          'deploy',
          '--config',
          path.join(workDir, 'prisma.config.ts'),
        ],
        {
          cwd: BACKEND_ROOT,
          env: { ...process.env, MIGRATION_HARNESS_URL: scratchUrl },
          timeout: DEPLOY_TIMEOUT_MS,
        },
      );
    } finally {
      await admin.query('SELECT pg_advisory_unlock($1)', [
        HARNESS_DEPLOY_LOCK_KEY,
      ]);
    }
  };

  let client: Client | null = null;
  const connected = async (): Promise<Client> => {
    if (!client) {
      client = new Client({
        connectionString: withoutSchemaParam(baseConnectionString()),
      });
      await client.connect();
      await client.query(`SET search_path TO "${schema}"`);
    }
    return client;
  };

  try {
    await deploy(target);
  } catch (error) {
    await fs.rm(workDir, { recursive: true, force: true });
    await admin.query(`DROP SCHEMA IF EXISTS "${schema}" CASCADE`);
    await admin.end();
    throw error;
  }

  return {
    schema,
    async query<T = Record<string, unknown>>(
      sql: string,
      params: unknown[] = [],
    ): Promise<T[]> {
      const active = await connected();
      const result = await active.query(sql, params);
      return result.rows as T[];
    },
    async migrateThrough(migrationName: string): Promise<void> {
      const index = names.indexOf(migrationName);
      if (index < 0) {
        throw new Error(`Unknown migration: ${migrationName}`);
      }
      await deploy(index + 1);
    },
    async drop(): Promise<void> {
      if (client) {
        await client.end();
        client = null;
      }
      await admin.query(`DROP SCHEMA IF EXISTS "${schema}" CASCADE`);
      await admin.end();
      await fs.rm(workDir, { recursive: true, force: true });
    },
  };
}