POSTGRESQL REFERENCE · WORKING SOURCE

src/db.ts

This file ships in the runnable source archive. Synthetic fixtures and demonstration policies are identified in the README.

import pg from "pg";
import type { Pool, PoolClient } from "pg";
import type { Actor } from "./contracts.ts";
import { AppError, canonical, fail, hash, uid } from "./errors.ts";
pg.types.setTypeParser(1184, (value) => new Date(value).toISOString());
pg.types.setTypeParser(1082, (value) => value);
export const pool = new pg.Pool({
  connectionString: Deno.env.get("DATABASE_URL"),
  max: 16,
  connectionTimeoutMillis: 5000,
  idleTimeoutMillis: 5000,
});
export type DB = PoolClient;
export type DBPool = Pool;
export async function now(db: DB): Promise<string> {
  if (Deno.env.get("DEMO_MODE") === "true") {
    return (await db.query("SELECT now_at AS now FROM demo_clock WHERE id=1")).rows[0].now;
  }
  return (await db.query("SELECT clock_timestamp() AS now")).rows[0].now;
}
export async function warehouseAccess(db: DB, actor: Actor, warehouseId: string, lock = false) {
  const rows =
    (await db.query(`SELECT * FROM warehouses WHERE id=$1 ${lock ? "FOR UPDATE" : ""}`, [warehouseId])).rows;
  if (!rows.length) fail(404, "NOT_FOUND", "Warehouse not found.");
  const w = rows[0];
  if (actor.role === "operator") {
    if (w.operator_org_id !== actor.orgId) fail(404, "NOT_FOUND", "Warehouse not found.");
  } else if (
    !(await db.query("SELECT 1 FROM operator_customers WHERE operator_org_id=$1 AND customer_org_id=$2", [
      w.operator_org_id,
      actor.orgId,
    ])).rowCount
  ) fail(404, "NOT_FOUND", "Warehouse not found.");
  return w;
}
export async function transaction<T>(fn: (db: DB) => Promise<T>, source: DBPool = pool): Promise<T> {
  for (let attempt = 0; attempt < 3; attempt++) {
    const db = await source.connect();
    try {
      await db.query("BEGIN ISOLATION LEVEL READ COMMITTED");
      await db.query("SET LOCAL lock_timeout='5s'");
      const value = await fn(db);
      await db.query("COMMIT");
      return value;
    } catch (error) {
      await db.query("ROLLBACK").catch(() => {});
      const code = (error as { code?: string }).code;
      if (["40001", "40P01", "55P03"].includes(code ?? "")) {
        if (attempt === 2) {
          throw new AppError(
            503,
            "RETRY_TRANSACTION",
            "Another operation is using this warehouse. Retry with the same request key.",
            undefined,
            true,
          );
        }
        await new Promise((r) => setTimeout(r, 20 + Math.random() * 60 * (attempt + 1)));
        continue;
      }
      throw error;
    } finally {
      db.release();
    }
  }
  throw new Error("Unreachable transaction result");
}
export async function warehouseTx<T>(
  actor: Actor,
  warehouseId: string,
  fn: (db: DB, clock: string) => Promise<T>,
  idem?: { operation: string; key: string; payload: unknown },
): Promise<T> {
  return await transaction(async (db) => {
    await warehouseAccess(db, actor, warehouseId, true);
    // Separate statement after the lock: READ COMMITTED now sees the winner's writes.
    const clock = await now(db);
    let payloadHash = "";
    if (idem) {
      if (!idem.key || idem.key.length > 160) {
        fail(400, "IDEMPOTENCY_REQUIRED", "Provide an Idempotency-Key of 1–160 characters.");
      }
      payloadHash = await hash(canonical(idem.payload));
      const prior =
        (await db.query("SELECT * FROM idempotency WHERE actor_id=$1 AND operation=$2 AND key=$3", [
          actor.id,
          idem.operation,
          idem.key,
        ])).rows[0];
      if (prior) {
        if (prior.payload_hash !== payloadHash) {
          fail(409, "IDEMPOTENCY_CONFLICT", "This request key was already used with a different payload.");
        }
        return prior.result as T;
      }
    }
    const result = await fn(db, clock);
    if (idem) {
      await db.query(
        "INSERT INTO idempotency(actor_id,operation,key,payload_hash,result) VALUES($1,$2,$3,$4,$5)",
        [actor.id, idem.operation, idem.key, payloadHash, JSON.stringify(result)],
      );
    }
    return result;
  });
}
export async function audit(
  db: DB,
  actor: Actor,
  clock: string,
  action: string,
  options: {
    warehouseId?: string;
    customerOrgId?: string;
    reservationId?: string;
    reason?: string;
    data?: unknown;
  } = {},
) {
  await db.query(
    "INSERT INTO audit_events(id,actor_id,customer_org_id,warehouse_id,reservation_id,action,occurred_at,reason,data) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9)",
    [
      uid("aud"),
      actor.id,
      options.customerOrgId ?? null,
      options.warehouseId ?? null,
      options.reservationId ?? null,
      action,
      clock,
      options.reason ?? action,
      JSON.stringify(options.data ?? {}),
    ],
  );
}