Ekklesia/core/src/database/pool.ts

import { Pool, type PoolConfig } from "pg";
import { type DatabaseTransportPolicy, databaseTlsConnectionOptions } from "./transport.js";

export type PostgresPoolErrorCode = "POSTGRES_POOL_CONFIGURATION_INVALID";

/** Never carries the offending connection string, host, or underlying driver error. */
export class PostgresPoolError extends Error {
  readonly code: PostgresPoolErrorCode;

  constructor(code: PostgresPoolErrorCode) {
    super("PostgreSQL pool configuration is invalid.");
    this.name = "PostgresPoolError";
    this.code = code;
    this.stack = `${this.name}: ${this.message}`;
    Object.freeze(this);
  }
}

const maximumPostgresTimeoutMs = 2_147_483_647;

export interface PostgresPoolOptions {
  readonly connectionTimeoutMs?: number;
  /** Bounds both `query_timeout` and `statement_timeout`: `connectionTimeoutMs` alone only bounds
   * establishing a connection, not a hung query already running on one. */
  readonly operationTimeoutMs?: number;
  /** The validated transport policy (`./transport.js`). Omitted only by isolated unit-test and
   * single-domain factories that connect to a disposable loopback database; every production
   * executable passes one explicitly. When omitted, no `ssl` option is set and the connection is
   * plaintext — exactly the pre-existing development behavior. */
  readonly transport?: DatabaseTransportPolicy;
}

export interface PostgresPoolHandle {
  readonly close: () => Promise<void>;
  readonly pool: Pool;
}

function validateConnectionString(connectionString: string): string {
  if (connectionString.trim().length === 0) {
    throw new PostgresPoolError("POSTGRES_POOL_CONFIGURATION_INVALID");
  }
  let protocol: string;
  try {
    protocol = new URL(connectionString).protocol;
  } catch {
    throw new PostgresPoolError("POSTGRES_POOL_CONFIGURATION_INVALID");
  }
  if (protocol !== "postgres:" && protocol !== "postgresql:") {
    throw new PostgresPoolError("POSTGRES_POOL_CONFIGURATION_INVALID");
  }
  return connectionString;
}

/** Mirrors the validation already used by `core/src/identity/database.ts`,
 * `core/src/authorization/database.ts`, and `core/src/migrations/postgres-database.ts`: an
 * explicit timeout must be a positive safe integer no larger than PostgreSQL's own maximum, or it
 * is rejected as invalid configuration rather than silently passed through to `pg`. */
function validateTimeout(value: unknown): number | undefined {
  if (value === undefined) {
    return undefined;
  }
  if (
    typeof value !== "number" ||
    !Number.isSafeInteger(value) ||
    value <= 0 ||
    value > maximumPostgresTimeoutMs
  ) {
    throw new PostgresPoolError("POSTGRES_POOL_CONFIGURATION_INVALID");
  }
  return value;
}

/**
 * Ekklesia's one shared, validated `pg.Pool` constructor for a single physical database
 * connection meant to back more than one typed Drizzle view (for example Core's identity and
 * authorization schemas sharing one pool in `apps/api`) — one pool, one mandatory error listener,
 * one `close()`. Building the pool performs no network I/O until a query is issued, and this
 * function is only ever invoked by explicit caller code — never during module import.
 *
 * This does not replace `createIdentityDrizzleDatabase` or `createAuthorizationDrizzleDatabase`:
 * those remain independent, self-contained factories (each owns a private pool and its own
 * `close()`) for callers — tests included — that only need one domain. Use this function, plus
 * `createIdentityDrizzleView`/`createAuthorizationDrizzleView`, specifically when more than one
 * domain must share one physical connection pool.
 */
export function createValidatedPostgresPool(
  connectionString: string,
  options: PostgresPoolOptions = {},
): PostgresPoolHandle {
  const validatedConnectionString = validateConnectionString(connectionString);
  const validatedConnectionTimeoutMs = validateTimeout(options.connectionTimeoutMs);
  const validatedOperationTimeoutMs = validateTimeout(options.operationTimeoutMs);
  // `databaseTlsConnectionOptions` also rejects a URL carrying its own TLS parameters, so the
  // pinned driver's `Object.assign({}, config, parse(connectionString))` precedence can never
  // silently replace the policy's `ssl` object with a weaker one.
  const ssl =
    options.transport === undefined
      ? undefined
      : databaseTlsConnectionOptions(validatedConnectionString, options.transport);
  const poolConfig: PoolConfig = {
    connectionString: validatedConnectionString,
    ...(ssl === undefined ? {} : { ssl }),
    ...(validatedConnectionTimeoutMs === undefined
      ? {}
      : { connectionTimeoutMillis: validatedConnectionTimeoutMs }),
    ...(validatedOperationTimeoutMs === undefined
      ? {}
      : {
          query_timeout: validatedOperationTimeoutMs,
          statement_timeout: validatedOperationTimeoutMs,
        }),
  };

  const pool = new Pool(poolConfig);
  // Same rationale as core/src/identity/database.ts: pg's own documentation requires an error
  // listener on the pool, or a server-initiated termination of an idle client becomes an
  // unhandled Node.js error and can crash the process.
  pool.on("error", () => undefined);

  return Object.freeze({
    close: () => pool.end(),
    pool,
  });
}