Jyotisha/frontend/tests/helpers/postgres-fixture.ts

import { execFileSync, spawnSync } from "node:child_process";
import {
  chmodSync,
  mkdirSync,
  mkdtempSync,
  readFileSync,
  rmSync,
  writeFileSync,
} from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";

export type PostgresFixture = {
  projectName: string;
  databaseEnvFile: string;
  hostPort: number;
  connectionUrl(role: string, password: string): string;
  psql(sql: string): string;
  psqlAs(role: string, password: string, sql: string): string;
  psqlScriptAs(role: string, password: string, sql: string): string;
  stop(): void;
};

const databaseEnvironment = `POSTGRES_DB=jyotisha
POSTGRES_USER=postgres
POSTGRES_PASSWORD=postgres-test-password
SCHEMA_OWNER_PASSWORD=schema-owner-test-password
IDENTITY_RUNTIME_PASSWORD=identity-runtime-test-password
APP_RUNTIME_PASSWORD=app-runtime-test-password
SERVICE_RUNTIME_PASSWORD=service-runtime-test-password
ADMIN_RUNTIME_PASSWORD=admin-runtime-test-password
MIGRATION_RUNNER_PASSWORD=migration-runner-test-password
BACKUP_READER_PASSWORD=backup-reader-test-password
STAGING_BACKUP_ENCRYPTION_KEY=staging-backup-test-password
SCHEMA_DATABASE_URL=postgresql://schema_owner:schema-owner-test-password@postgres:5432/jyotisha
`;

function isPortAvailable(port: number): boolean {
  return (
    spawnSync(
      process.execPath,
      [
        "-e",
        `const server = require("node:net").createServer();
server.once("error", () => process.exit(1));
server.listen({ host: "127.0.0.1", port: Number(process.argv[1]) }, () =>
  server.close(() => process.exit(0)),
);`,
        String(port),
      ],
      { stdio: "ignore" },
    ).status === 0
  );
}

type PortReservation = { port: number; directory: string };

function reserveAvailablePort(): PortReservation {
  for (let port = 55432; port <= 55531; port += 1) {
    const directory = join(tmpdir(), `jyotisha-postgres-port-${port}.lock`);
    try {
      mkdirSync(directory);
    } catch (error) {
      if ((error as NodeJS.ErrnoException).code === "EEXIST") continue;
      throw error;
    }

    if (isPortAvailable(port)) {
      return { port, directory };
    }
    rmSync(directory, { force: true, recursive: true });
  }

  throw new Error("no available PostgreSQL test port in 55432..55531");
}

function releasePort(reservation: PortReservation): void {
  rmSync(reservation.directory, { force: true, recursive: true });
}

const DEFAULT_MAX_CONCURRENT_POSTGRES_FIXTURES = 2;
/**
 * How long the slot queue may stand still before a waiter gives up (BUG-1254).
 * It used to be a fixed 5-minute wait from the moment a file started queueing;
 * with 46 fixture files sharing 2 slots the queue itself outlasted that, and
 * whichever file happened to be last failed the gate at random. The window now
 * restarts whenever a slot changes hands, so a moving queue is waited out and
 * only a queue where no slot changes for this long (a hung holder) fails.
 */
export const FIXTURE_SLOT_STALL_MS = 10 * 60 * 1000;

type FixtureSlot = { id: number; release(): void };

function maxConcurrentPostgresFixtures(): number {
  const raw = process.env.JYOTISHA_POSTGRES_MAX_FIXTURES;
  if (raw === undefined || raw === "") {
    return DEFAULT_MAX_CONCURRENT_POSTGRES_FIXTURES;
  }
  const parsed = Number(raw);
  if (!Number.isInteger(parsed) || parsed < 1) {
    throw new Error("JYOTISHA_POSTGRES_MAX_FIXTURES must be a positive integer");
  }
  return parsed;
}

function fixtureSlotsRoot(): string {
  return join(tmpdir(), "jyotisha-postgres-slots");
}

function fixtureSlotDirectory(id: number): string {
  return join(fixtureSlotsRoot(), `slot-${id}`);
}

function pidIsAlive(pid: number): boolean {
  if (!Number.isInteger(pid) || pid <= 0) {
    return false;
  }
  try {
    process.kill(pid, 0);
    return true;
  } catch {
    return false;
  }
}

function tryClaimSlot(id: number): boolean {
  const directory = fixtureSlotDirectory(id);
  mkdirSync(fixtureSlotsRoot(), { recursive: true });
  try {
    mkdirSync(directory);
  } catch (error) {
    if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error;
    try {
      const pid = Number(readFileSync(join(directory, "pid"), "utf8"));
      if (pidIsAlive(pid)) return false;
      rmSync(directory, { force: true, recursive: true });
      mkdirSync(directory);
    } catch {
      return false;
    }
  }
  writeFileSync(join(directory, "pid"), String(process.pid));
  return true;
}

/** Who holds each slot right now: `id:pid` per held slot, empty when free. */
function slotOwnersSnapshot(limit: number): string {
  const owners: string[] = [];
  for (let id = 0; id < limit; id += 1) {
    try {
      owners.push(`${id}:${readFileSync(join(fixtureSlotDirectory(id), "pid"), "utf8")}`);
    } catch {
      owners.push(`${id}:`);
    }
  }
  return owners.join("|");
}

/**
 * The waiter's deadline after one poll: pushed out by a full stall window when
 * the slot owners changed since the last poll, unchanged otherwise.
 */
export function nextFixtureSlotDeadline(
  previousOwners: string,
  currentOwners: string,
  deadline: number,
  now: number,
  stallMs: number = FIXTURE_SLOT_STALL_MS,
): number {
  return currentOwners === previousOwners ? deadline : now + stallMs;
}

function acquireFixtureSlot(): FixtureSlot {
  const limit = maxConcurrentPostgresFixtures();
  let owners = slotOwnersSnapshot(limit);
  let deadline = Date.now() + FIXTURE_SLOT_STALL_MS;
  while (Date.now() < deadline) {
    for (let id = 0; id < limit; id += 1) {
      if (tryClaimSlot(id)) {
        return {
          id,
          release() {
            rmSync(fixtureSlotDirectory(id), { force: true, recursive: true });
          },
        };
      }
    }
    Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 50);
    const current = slotOwnersSnapshot(limit);
    deadline = nextFixtureSlotDeadline(owners, current, deadline, Date.now());
    owners = current;
  }
  throw new Error(
    `timed out waiting for a PostgreSQL fixture slot (${limit} concurrent compose networks): no slot changed hands for ${FIXTURE_SLOT_STALL_MS / 60000} minutes, so a holder is likely hung. Docker address pools cannot host one network per parallel test file.`,
  );
}

export function startPostgresFixture(): PostgresFixture {
  const slot = acquireFixtureSlot();
  const projectName = `jyotisha-postgres-${process.pid}-${Date.now()}`;
  let temporaryDirectory: string;
  try {
    temporaryDirectory = mkdtempSync(join(tmpdir(), "jyotisha-postgres-"));
  } catch (error) {
    slot.release();
    throw error;
  }
  const databaseEnvFile = join(temporaryDirectory, "database.env");
  let portReservation: PortReservation;
  try {
    portReservation = reserveAvailablePort();
  } catch (error) {
    rmSync(temporaryDirectory, { force: true, recursive: true });
    slot.release();
    throw error;
  }
  const hostPort = portReservation.port;
  const composeArguments = [
    "compose",
    "--project-name",
    projectName,
    "--env-file",
    databaseEnvFile,
    "-f",
    "../deploy/docker-compose.postgres.yml",
    "-f",
    "../deploy/docker-compose.postgres-ci.yml",
  ];
  const environment = {
    ...process.env,
    DATABASE_ENV_FILE: databaseEnvFile,
    POSTGRES_HOST_PORT: String(hostPort),
  };

  try {
    writeFileSync(databaseEnvFile, databaseEnvironment, { mode: 0o600 });
    chmodSync(databaseEnvFile, 0o600);
    execFileSync(
      "docker",
      [...composeArguments, "up", "-d", "--wait", "postgres"],
      { env: environment, stdio: "inherit" },
    );
  } catch (error) {
    try {
      try {
        // The teardown below deletes the only account of why the server refused to start.
        execFileSync(
          "docker",
          [...composeArguments, "logs", "--no-color", "--tail", "80", "postgres"],
          { env: environment, stdio: "inherit" },
        );
      } catch {
        // Preserve the original startup failure.
      }
      try {
        execFileSync(
          "docker",
          [...composeArguments, "down", "-v", "--remove-orphans"],
          { env: environment, stdio: "inherit" },
        );
      } catch {
        // Preserve the original startup failure.
      }
    } finally {
      releasePort(portReservation);
      rmSync(temporaryDirectory, { force: true, recursive: true });
      slot.release();
    }
    throw error;
  }

  return {
    projectName,
    databaseEnvFile,
    hostPort,
    connectionUrl(role, password) {
      return `postgresql://${role}:${password}@127.0.0.1:${hostPort}/jyotisha`;
    },
    psql(sql) {
      return execFileSync(
        "docker",
        [...composeArguments, "exec", "-T", "postgres", "psql", "-U", "postgres", "-d", "jyotisha", "-Atc", sql],
        { encoding: "utf8", env: environment },
      )
        .trim()
        // boolean::text is "false"; psql -A prints "f". Keep colon-joined
        // results on the t/f alphabet so tests can assert "f:" stably.
        .replace(/(^|:)false(?=:|$)/gm, "$1f");
    },
    psqlAs(role, password, sql) {
      return execFileSync(
        "docker",
        [
          ...composeArguments,
          "exec",
          "-T",
          "-e",
          `PGPASSWORD=${password}`,
          "postgres",
          "psql",
          "-v",
          "ON_ERROR_STOP=1",
          "-U",
          role,
          "-d",
          "jyotisha",
          "-Atc",
          sql,
        ],
        { encoding: "utf8", env: environment },
      ).trim();
    },
    psqlScriptAs(role, password, sql) {
      return execFileSync(
        "docker",
        [
          ...composeArguments,
          "exec",
          "-T",
          "-e",
          `PGPASSWORD=${password}`,
          "postgres",
          "psql",
          "-X",
          "-At",
          "-v",
          "ON_ERROR_STOP=1",
          "--single-transaction",
          "-f",
          "-",
          "-U",
          role,
          "-d",
          "jyotisha",
        ],
        {
          encoding: "utf8",
          env: environment,
          input: sql,
          maxBuffer: 16 * 1024 * 1024,
        },
      ).trim();
    },
    stop() {
      try {
        execFileSync(
          "docker",
          [...composeArguments, "down", "-v", "--remove-orphans"],
          { env: environment, stdio: "inherit" },
        );
      } finally {
        releasePort(portReservation);
        rmSync(temporaryDirectory, { force: true, recursive: true });
        slot.release();
      }
    },
  };
}