import { randomBytes } from "node:crypto";
import {
  isModuleDatabaseError,
  type ModuleDatabaseErrorCode,
  type ModuleDatabaseExecutor,
  type ModuleDatabaseIsolationLevel,
  type ModuleDatabaseTransaction,
} from "@ekklesia/contracts";
import { Client, Pool, type PoolClient } from "pg";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { createValidatedPostgresPool } from "../src/database/index.js";
import { createModuleDatabaseExecutor } from "../src/modules/index.js";

/**
 * Module Database ABI v1 (ADR 0050) against real PostgreSQL.
 *
 * Everything the ABI promises that depends on PostgreSQL itself is proven here, not mocked: type
 * OIDs and decoding, the one-statement rule, SQLSTATEs, transactions, isolation, savepoints,
 * session pinning and reset, and that every failure leaves the pool's next borrower a clean
 * connection.
 *
 * The disposable database is created with deliberately hostile defaults (a non-UTC `TimeZone`, a
 * non-ISO `DateStyle`, `sql_standard` intervals, `escape` bytea output, and reduced float digits),
 * so every decoding assertion also proves the executor pins its own session settings, and every
 * pool-safety assertion proves it gives the database's defaults back afterwards. The fixture table
 * is created directly in that disposable database; no migration runs.
 */
const integrationDatabaseUrlEnvironment = "EKKLESIA_MIGRATION_TEST_DATABASE_URL";
const trackedDatabases = new Set<string>();

class IntegrationHarnessError extends Error {
  constructor(message: string) {
    super(message);
    this.name = "IntegrationHarnessError";
    this.stack = `${this.name}: ${this.message}`;
  }
}

function baseDatabaseUrl(): string {
  const value = process.env[integrationDatabaseUrlEnvironment];
  if (typeof value !== "string" || value.trim() === "") {
    throw new IntegrationHarnessError(
      "The module database ABI integration database is not configured.",
    );
  }
  try {
    const parsed = new URL(value);
    if (parsed.protocol !== "postgres:" && parsed.protocol !== "postgresql:") {
      throw new TypeError("Unsupported protocol.");
    }
  } catch {
    throw new IntegrationHarnessError(
      "The module database ABI integration database configuration is invalid.",
    );
  }
  return value;
}

function quoteDatabaseName(databaseName: string): string {
  if (!/^[a-z][a-z0-9_]{0,62}$/u.test(databaseName)) {
    throw new IntegrationHarnessError("The generated integration database name is invalid.");
  }
  return `"${databaseName}"`;
}

async function withAdministrativeClient<T>(
  connectionString: string,
  run: (client: Client) => Promise<T>,
): Promise<T> {
  const client = new Client({ connectionString });
  client.on("error", () => undefined);
  try {
    await client.connect();
  } catch {
    await client.end().catch(() => undefined);
    throw new IntegrationHarnessError("The integration database connection failed.");
  }
  try {
    return await run(client);
  } finally {
    await client.end().catch(() => undefined);
  }
}

async function createDisposableDatabase(): Promise<string> {
  const name = `ekk_db_abi_${process.pid}_${randomBytes(4).toString("hex")}`;
  const quoted = quoteDatabaseName(name);
  await withAdministrativeClient(baseDatabaseUrl(), async (admin) => {
    await admin.query(`CREATE DATABASE ${quoted} TEMPLATE template0`);
    trackedDatabases.add(name);
    // Hostile defaults: everything the ABI pins is set to something else here.
    await admin.query(`ALTER DATABASE ${quoted} SET TimeZone TO 'America/Chicago'`);
    await admin.query(`ALTER DATABASE ${quoted} SET DateStyle TO 'SQL, DMY'`);
    await admin.query(`ALTER DATABASE ${quoted} SET IntervalStyle TO 'sql_standard'`);
    await admin.query(`ALTER DATABASE ${quoted} SET bytea_output TO 'escape'`);
    await admin.query(`ALTER DATABASE ${quoted} SET extra_float_digits TO -3`);
  });
  const parsed = new URL(baseDatabaseUrl());
  parsed.pathname = `/${name}`;
  parsed.searchParams.delete("application_name");
  return parsed.toString();
}

async function dropDatabase(name: string): Promise<void> {
  await withAdministrativeClient(baseDatabaseUrl(), async (admin) => {
    await admin.query(`DROP DATABASE IF EXISTS ${quoteDatabaseName(name)} WITH (FORCE)`);
  });
  trackedDatabases.delete(name);
}

let databaseUrl: string;
let pool: Pool;
let closePool: () => Promise<void>;
let database: ModuleDatabaseExecutor;

beforeAll(async () => {
  databaseUrl = await createDisposableDatabase();
  await withAdministrativeClient(databaseUrl, async (admin) => {
    await admin.query(
      "CREATE TABLE example_records (id integer PRIMARY KEY, value text NOT NULL, secret text UNIQUE)",
    );
    await admin.query(
      "CREATE TABLE example_deferred (id integer, CONSTRAINT example_deferred_id_key UNIQUE (id) DEFERRABLE INITIALLY DEFERRED)",
    );
  });
  // The production pool: validated, with Core's own per-statement bounds.
  ({ close: closePool, pool } = createValidatedPostgresPool(databaseUrl, {
    connectionTimeoutMs: 10_000,
    operationTimeoutMs: 20_000,
  }));
  database = createModuleDatabaseExecutor(pool, { moduleId: "example" });
});

afterAll(async () => {
  await closePool?.();
  for (const name of [...trackedDatabases]) {
    await dropDatabase(name).catch(() => undefined);
  }
});

async function rejection(promise: Promise<unknown>): Promise<unknown> {
  try {
    await promise;
  } catch (error) {
    return error;
  }
  throw new Error("Expected the operation to reject.");
}

/** Asserts a sanitized ABI error: the code, PostgreSQL's SQLSTATE or none, and nothing else. */
async function expectAbiError(
  promise: Promise<unknown>,
  code: ModuleDatabaseErrorCode,
  sqlState?: string,
  secrets: readonly string[] = [],
): Promise<void> {
  const error = await rejection(promise);
  expect(isModuleDatabaseError(error)).toBe(true);
  expect((error as { code: string }).code).toBe(code);
  expect((error as { sqlState?: string }).sqlState).toBe(sqlState);
  expect(Object.keys(error as object).sort()).toEqual(
    sqlState === undefined ? ["code", "name"] : ["code", "name", "sqlState"],
  );
  expect((error as Error).cause).toBeUndefined();
  expect((error as Error).stack).toBe(`ModuleDatabaseError: ${(error as Error).message}`);
  const serialized = `${JSON.stringify(error)} ${String(error)}`;
  for (const secret of [...secrets, databaseUrl, "example_records", "postgres://"]) {
    expect(serialized).not.toContain(secret);
  }
}

async function reset(): Promise<void> {
  await withAdministrativeClient(databaseUrl, async (admin) => {
    await admin.query("TRUNCATE example_records, example_deferred");
  });
}

async function storedIds(): Promise<number[]> {
  return withAdministrativeClient(databaseUrl, async (admin) => {
    const result = await admin.query<{ id: number }>("SELECT id FROM example_records ORDER BY id");
    return result.rows.map((row) => row.id);
  });
}

// --- Queries ------------------------------------------------------------------------------------

describe("query", () => {
  it("runs a simple select in object mode by default", async () => {
    const result = await database.query({ text: "SELECT 1 AS one, 'two' AS two" });
    expect(result).toEqual({
      command: "SELECT",
      fields: [
        { name: "one", typeOid: 23 },
        { name: "two", typeOid: 25 },
      ],
      rowCount: 1,
      rows: [{ one: 1, two: "two" }],
    });
  });

  it("binds parameters and never interpolates them", async () => {
    const hostile = "x'); DROP TABLE example_records; --";
    const result = await database.query({ text: "SELECT $1::text AS v", values: [hostile] });
    expect(result.rows).toEqual([{ v: hostile }]);
    expect(await storedIds()).toEqual([]);
  });

  it("keeps array-mode columns positional, and returns metadata even with no rows", async () => {
    const result = await database.query({
      rowMode: "array",
      text: "SELECT 1 AS a, 'b' AS b WHERE false",
    });
    expect(result.rows).toEqual([]);
    expect(result.rowCount).toBe(0);
    expect(result.fields).toEqual([
      { name: "a", typeOid: 23 },
      { name: "b", typeOid: 25 },
    ]);
  });

  it("returns duplicate column names positionally in array mode, and refuses them in object mode", async () => {
    const arrayResult = await database.query({ rowMode: "array", text: "SELECT 1 AS a, 2 AS a" });
    expect(arrayResult.rows).toEqual([[1, 2]]);
    expect(arrayResult.fields.map((field) => field.name)).toEqual(["a", "a"]);
    await expectAbiError(
      database.query({ text: "SELECT 1 AS a, 2 AS a" }),
      "MODULE_DATABASE_REQUEST_INVALID",
    );
  });

  it("finds duplicate object-mode columns only after the statement ran, so atomicity needs a transaction", async () => {
    await reset();
    // At the top level, the mutation has happened by the time the result's shape is known.
    await expectAbiError(
      database.query({
        text: "INSERT INTO example_records (id, value) VALUES (1, 'a') RETURNING id, id",
      }),
      "MODULE_DATABASE_REQUEST_INVALID",
    );
    expect(await storedIds()).toEqual([1]);
    // Inside a transaction, the same error rolls the mutation back with everything else.
    await expectAbiError(
      database.transaction((tx) =>
        tx.query({
          text: "INSERT INTO example_records (id, value) VALUES (2, 'b') RETURNING id, id",
        }),
      ),
      "MODULE_DATABASE_REQUEST_INVALID",
    );
    expect(await storedIds()).toEqual([1]);
  });

  it("keeps a column named __proto__ a column", async () => {
    const result = await database.query({ text: 'SELECT 1 AS "__proto__"' });
    const row = result.rows[0] as Record<string, unknown>;
    expect(Object.hasOwn(row, "__proto__")).toBe(true);
    expect(Object.getPrototypeOf(row)).toBe(Object.prototype);
  });

  it("reports command tags and row counts for writes, and RETURNING rows as ordinary rows", async () => {
    await reset();
    const inserted = await database.query({
      text: "INSERT INTO example_records (id, value) VALUES ($1, $2), ($3, $4) RETURNING id, value",
      values: [1, "a", 2, "b"],
    });
    expect(inserted.command).toBe("INSERT");
    expect(inserted.rowCount).toBe(2);
    expect(inserted.rows).toEqual([
      { id: 1, value: "a" },
      { id: 2, value: "b" },
    ]);
    const updated = await database.query({
      text: "UPDATE example_records SET value = 'c' WHERE id > $1",
      values: [0],
    });
    expect(updated).toMatchObject({ command: "UPDATE", rowCount: 2, rows: [] });
    const deleted = await database.query({ text: "DELETE FROM example_records" });
    expect(deleted).toMatchObject({ command: "DELETE", rowCount: 2 });
    const created = await database.query({ text: "CREATE TEMPORARY TABLE scratch (x int)" });
    expect(created).toMatchObject({ command: "CREATE", rowCount: null });
  });
});

describe("one statement per query", () => {
  it("runs exactly one statement, a trailing semicolon included", async () => {
    expect((await database.query({ text: "SELECT 1 AS n" })).rows).toEqual([{ n: 1 }]);
    expect((await database.query({ text: "SELECT 1 AS n;" })).rows).toEqual([{ n: 1 }]);
  });

  it("refuses two statements, with PostgreSQL's own SQLSTATE, and runs neither", async () => {
    await reset();
    await expectAbiError(
      database.query({ text: "SELECT 1; SELECT 2" }),
      "MODULE_DATABASE_QUERY_FAILED",
      "42601",
    );
    await expectAbiError(
      database.query({
        text: "INSERT INTO example_records (id, value) VALUES (1, 'a'); INSERT INTO example_records (id, value) VALUES (2, 'b')",
      }),
      "MODULE_DATABASE_QUERY_FAILED",
      "42601",
    );
    await expectAbiError(
      database.query({ text: "SELECT $1::int; SELECT 2", values: [1] }),
      "MODULE_DATABASE_QUERY_FAILED",
      "42601",
    );
    expect(await storedIds()).toEqual([]);
  });

  it("never mistakes semicolons in literals, identifiers, or comments for a second statement", async () => {
    const result = await database.query({
      rowMode: "array",
      text: [
        "SELECT ';' AS a,",
        "  $$;SELECT 2;$$ AS b,",
        "  $tag$;$tag$ AS c,",
        "  E'\\x3B' AS d,",
        '  1 AS "semi;colon" -- ; SELECT 3',
        "/* ; SELECT 4; */",
      ].join("\n"),
    });
    expect(result.rows).toEqual([[";", ";SELECT 2;", ";", ";", 1]]);
    expect(result.fields.map((field) => field.name)).toEqual(["a", "b", "c", "d", "semi;colon"]);
  });

  it("refuses statement text that holds no statement", async () => {
    for (const text of [" ", "-- only a comment", "/* nothing */", ";"]) {
      await expectAbiError(database.query({ text }), "MODULE_DATABASE_REQUEST_INVALID");
    }
  });
});

// --- Parameters ---------------------------------------------------------------------------------

describe("parameters", () => {
  it("binds every allowed category with its exact value", async () => {
    const result = await database.query({
      rowMode: "array",
      text: [
        "SELECT $1::text IS NULL, $2::bool, $3::float8, $4::float8, $5::float8, $6::float8,",
        "  $7::int8, $8::numeric, $9::text, $10::bytea, $11::bytea, $12::int4",
      ].join(" "),
      values: [
        null,
        true,
        0.1,
        Number.NaN,
        Number.POSITIVE_INFINITY,
        Number.NEGATIVE_INFINITY,
        9_223_372_036_854_775_807n,
        12_345_678_901_234_567_890n,
        'it\'s "quoted" \\ and ünïcödé 🙂',
        new Uint8Array([0, 1, 254, 255]),
        Buffer.from([7]),
        -2_147_483_648,
      ],
    });
    const row = result.rows[0] as readonly unknown[];
    expect(row.slice(0, 3)).toEqual([true, true, 0.1]);
    expect(row[3]).toBeNaN();
    expect(row.slice(4, 10)).toEqual([
      Number.POSITIVE_INFINITY,
      Number.NEGATIVE_INFINITY,
      "9223372036854775807",
      "12345678901234567890",
      'it\'s "quoted" \\ and ünïcödé 🙂',
      new Uint8Array([0, 1, 254, 255]),
    ]);
    expect(row.slice(10)).toEqual([new Uint8Array([7]), -2_147_483_648]);
  });

  it("binds arrays, nulls inside them, and more dimensions", async () => {
    const result = await database.query({
      rowMode: "array",
      text: "SELECT $1::int[], $2::text[], $3::bytea[], $4::int[], $5::int8[], $6::text[], $7::bool[]",
      values: [
        [1, null, 3],
        ['a"b', "c,d", "e\\f", "", "NULL", null, "{x}"],
        [new Uint8Array([0, 255]), null],
        [
          [1, 2],
          [3, 4],
        ],
        [9_223_372_036_854_775_807n],
        [],
        [true, false],
      ],
    });
    expect(result.rows[0]).toEqual([
      [1, null, 3],
      ['a"b', "c,d", "e\\f", "", "NULL", null, "{x}"],
      [new Uint8Array([0, 255]), null],
      [
        [1, 2],
        [3, 4],
      ],
      ["9223372036854775807"],
      [],
      [true, false],
    ]);
  });

  it("finds rows with an array parameter", async () => {
    await reset();
    await database.query({
      text: "INSERT INTO example_records (id, value) VALUES (1, 'a'), (2, 'b'), (3, 'c')",
    });
    const result = await database.query({
      text: "SELECT id FROM example_records WHERE id = ANY($1) ORDER BY id",
      values: [[1, 3]],
    });
    expect(result.rows).toEqual([{ id: 1 }, { id: 3 }]);
  });

  const refused: readonly [string, unknown][] = [
    ["a Date", new Date("2026-10-07T00:00:00Z")],
    ["a plain object", { secret: "hunter2" }],
    ["undefined", undefined],
    ["a function", () => "hunter2"],
    ["a symbol", Symbol("hunter2")],
    ["a class instance", new URL("https://hunter2.invalid")],
    ["an array holding a Date", [new Date("2026-10-07T00:00:00Z")]],
    ["an array holding an object", [{ secret: "hunter2" }]],
    ["an array holding undefined", ["hunter2", undefined]],
    ["ragged sub-arrays", [["hunter2"], ["a", "b"]]],
  ];

  for (const [label, value] of refused) {
    it(`refuses ${label} before PostgreSQL sees anything`, async () => {
      await reset();
      await expectAbiError(
        database.query({
          text: "INSERT INTO example_records (id, value) VALUES (1, $1::text)",
          values: [value as never],
        }),
        "MODULE_DATABASE_PARAMETER_INVALID",
        undefined,
        ["hunter2", "2026-10-07", "INSERT"],
      );
      expect(await storedIds()).toEqual([]);
    });
  }
});

// --- Results ------------------------------------------------------------------------------------

describe("result decoding", () => {
  it("decodes every type in ADR 0050's table deterministically, despite hostile database defaults", async () => {
    const result = await database.query({
      text: [
        "SELECT",
        "  NULL::text AS null_value,",
        "  true AS bool, false AS bool_false,",
        "  (-32768)::int2 AS int2, 2147483647::int4 AS int4,",
        "  9223372036854775807::int8 AS int8,",
        "  1.5::float4 AS float4, 0.1::float8 AS float8, 'NaN'::float8 AS nan,",
        "  'Infinity'::float8 AS inf, '-Infinity'::float4 AS neg_inf,",
        "  12345678901234567890.123456789::numeric AS numeric, 'NaN'::numeric AS numeric_nan,",
        "  1.10::decimal(5,2) AS decimal,",
        "  'text'::text AS text, 'var'::varchar(10) AS varchar, 'ch'::char(3) AS bpchar,",
        "  'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'::uuid AS uuid,",
        "  '2026-10-07 12:34:56.789'::timestamp AS timestamp,",
        "  '2026-10-07 12:34:56.789+02'::timestamptz AS timestamptz,",
        "  '2026-10-07'::date AS date,",
        "  '12:34:56.5'::time AS time, '12:34:56+05:30'::timetz AS timetz,",
        "  '1 year 2 mons 3 days 04:05:06.7'::interval AS interval,",
        "  '-1 day'::interval AS negative_interval,",
        `  '{"big": 12345678901234567890, "f": 0.10000000000000000001}'::json AS json,`,
        `  '{"big": 12345678901234567890}'::jsonb AS jsonb,`,
        "  '\\x00ff10'::bytea AS bytea, ''::bytea AS empty_bytea",
      ].join("\n"),
    });
    const row = result.rows[0] as Record<string, unknown>;

    expect(row).toMatchObject({
      bool: true,
      bool_false: false,
      bpchar: "ch ",
      date: "2026-10-07",
      decimal: "1.10",
      float4: 1.5,
      float8: 0.1,
      inf: Number.POSITIVE_INFINITY,
      int2: -32_768,
      int4: 2_147_483_647,
      int8: "9223372036854775807",
      interval: "1 year 2 mons 3 days 04:05:06.7",
      json: '{"big": 12345678901234567890, "f": 0.10000000000000000001}',
      jsonb: '{"big": 12345678901234567890}',
      neg_inf: Number.NEGATIVE_INFINITY,
      negative_interval: "-1 days",
      null_value: null,
      numeric: "12345678901234567890.123456789",
      numeric_nan: "NaN",
      text: "text",
      time: "12:34:56.5",
      timestamp: "2026-10-07 12:34:56.789",
      timestamptz: "2026-10-07 10:34:56.789+00",
      timetz: "12:34:56+05:30",
      uuid: "a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11",
      varchar: "var",
    });
    expect(row.nan).toBeNaN();
    expect(row.bytea).toEqual(new Uint8Array([0, 255, 16]));
    expect(Object.getPrototypeOf(row.bytea)).toBe(Uint8Array.prototype);
    expect(row.empty_bytea).toEqual(new Uint8Array([]));
    expect(result.fields.map((field) => [field.name, field.typeOid])).toEqual([
      ["null_value", 25],
      ["bool", 16],
      ["bool_false", 16],
      ["int2", 21],
      ["int4", 23],
      ["int8", 20],
      ["float4", 700],
      ["float8", 701],
      ["nan", 701],
      ["inf", 701],
      ["neg_inf", 700],
      ["numeric", 1700],
      ["numeric_nan", 1700],
      ["decimal", 1700],
      ["text", 25],
      ["varchar", 1043],
      ["bpchar", 1042],
      ["uuid", 2950],
      ["timestamp", 1114],
      ["timestamptz", 1184],
      ["date", 1082],
      ["time", 1083],
      ["timetz", 1266],
      ["interval", 1186],
      ["negative_interval", 1186],
      ["json", 114],
      ["jsonb", 3802],
      ["bytea", 17],
      ["empty_bytea", 17],
    ]);
  });

  it("keeps float precision exact: the shortest text that round-trips", async () => {
    const values = [0.1, 1 / 3, 2 ** -1074, Number.MAX_VALUE, 123_456_789.123_456_78];
    const result = await database.query({
      rowMode: "array",
      text: "SELECT unnest($1::float8[])",
      values: [values],
    });
    expect(result.rows.map((row) => row[0])).toEqual(values);
  });

  it("decodes representative arrays element by element", async () => {
    const result = await database.query({
      text: [
        "SELECT",
        "  ARRAY[1, NULL, 9223372036854775807]::int8[] AS int8s,",
        "  ARRAY[1.10, 2]::numeric[] AS numerics,",
        "  ARRAY[true, false, NULL] AS bools,",
        "  ARRAY[[1, 2], [3, 4]]::int4[] AS matrix,",
        "  ARRAY[0.5, 'NaN']::float8[] AS floats,",
        `  ARRAY['a"b', 'c,d', '', 'NULL', NULL, '{}', 'back\\slash']::text[] AS texts,`,
        "  ARRAY['\\x00ff'::bytea, NULL] AS byteas,",
        "  ARRAY['2026-10-07 12:00:00+02'::timestamptz] AS instants,",
        "  ARRAY['2026-10-07'::date] AS dates,",
        "  ARRAY['1 day'::interval] AS intervals,",
        `  ARRAY['{"n": 12345678901234567890}'::jsonb] AS documents,`,
        "  ARRAY['a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'::uuid] AS uuids,",
        "  '{}'::int[] AS empty,",
        "  '[0:1]={5,6}'::int[] AS lower_bound",
      ].join("\n"),
    });
    const row = result.rows[0] as Record<string, unknown>;
    expect(row).toEqual({
      bools: [true, false, null],
      byteas: [new Uint8Array([0, 255]), null],
      dates: ["2026-10-07"],
      documents: ['{"n": 12345678901234567890}'],
      empty: [],
      floats: [0.5, Number.NaN],
      instants: ["2026-10-07 10:00:00+00"],
      int8s: ["1", null, "9223372036854775807"],
      intervals: ["1 day"],
      lower_bound: [5, 6],
      matrix: [
        [1, 2],
        [3, 4],
      ],
      numerics: ["1.10", "2"],
      texts: ['a"b', "c,d", "", "NULL", null, "{}", "back\\slash"],
      uuids: ["a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11"],
    });
  });

  it("decodes the same instant to the same text whatever the session started as", async () => {
    const instant = "2026-03-08 08:30:00+00";
    const first = await database.query({ text: "SELECT $1::timestamptz AS t", values: [instant] });
    const second = await database.transaction((tx) =>
      tx.query({ text: "SELECT $1::timestamptz AS t", values: [instant] }),
    );
    expect(first.rows).toEqual([{ t: instant }]);
    expect(second.rows).toEqual([{ t: instant }]);
  });
});

// --- Errors -------------------------------------------------------------------------------------

describe("errors", () => {
  it("reports a syntax error as a failed query with PostgreSQL's SQLSTATE and nothing more", async () => {
    await expectAbiError(
      database.query({ text: "SELEC secret_column FROM example_records" }),
      "MODULE_DATABASE_QUERY_FAILED",
      "42601",
      ["SELEC", "secret_column", "syntax"],
    );
    await expectAbiError(
      database.query({ text: "SELECT missing_column FROM example_records" }),
      "MODULE_DATABASE_QUERY_FAILED",
      "42703",
      ["missing_column"],
    );
  });

  it("keeps a unique violation's SQLSTATE and hides its message, detail, constraint, and value", async () => {
    await reset();
    await database.query({
      text: "INSERT INTO example_records (id, value, secret) VALUES (1, 'a', $1)",
      values: ["hunter2"],
    });
    await expectAbiError(
      database.query({
        text: "INSERT INTO example_records (id, value, secret) VALUES (2, 'b', $1)",
        values: ["hunter2"],
      }),
      "MODULE_DATABASE_QUERY_FAILED",
      "23505",
      ["hunter2", "example_records_secret_key", "duplicate", "Key ("],
    );
  });

  it("reports a primary-key conflict inside a transaction the same way", async () => {
    await reset();
    await database.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
    await expectAbiError(
      database.transaction(async (tx) => {
        await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'b')" });
      }),
      "MODULE_DATABASE_QUERY_FAILED",
      "23505",
      ["example_records_pkey"],
    );
  });
});

// --- Transactions -------------------------------------------------------------------------------

describe("transactions", () => {
  it("commits a resolved callback and returns its value", async () => {
    await reset();
    const value = await database.transaction(async (tx) => {
      await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
      await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" });
      return "committed";
    });
    expect(value).toBe("committed");
    expect(await storedIds()).toEqual([1, 2]);
  });

  it("rolls back a callback that throws, and rejects with the callback's own error", async () => {
    await reset();
    const own = new Error("module failure");
    const error = await rejection(
      database.transaction(async (tx) => {
        await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
        throw own;
      }),
    );
    expect(error).toBe(own);
    expect(await storedIds()).toEqual([]);
  });

  it("sees its own writes, and other connections do not until it commits", async () => {
    await reset();
    await database.transaction(async (tx) => {
      await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
      expect(
        (await tx.query({ text: "SELECT count(*)::int AS n FROM example_records" })).rows,
      ).toEqual([{ n: 1 }]);
      expect(await storedIds()).toEqual([]);
    });
    expect(await storedIds()).toEqual([1]);
  });

  it("never retries: a serialization failure reaches the callback once and rejects the transaction", async () => {
    await reset();
    await database.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
    let attempts = 0;
    await expectAbiError(
      database.transaction(
        async (tx) => {
          attempts += 1;
          await tx.query({ text: "SELECT value FROM example_records WHERE id = 1" });
          // Another connection commits a change to the same row after this snapshot was taken.
          await database.query({ text: "UPDATE example_records SET value = 'b' WHERE id = 1" });
          await tx.query({ text: "UPDATE example_records SET value = 'c' WHERE id = 1" });
        },
        { isolation: "repeatable read" },
      ),
      "MODULE_DATABASE_QUERY_FAILED",
      "40001",
    );
    expect(attempts).toBe(1);
    const rows = await database.query({ text: "SELECT value FROM example_records WHERE id = 1" });
    expect(rows.rows).toEqual([{ value: "b" }]);
  });

  it("reports a failure at COMMIT, such as a deferred constraint, with its SQLSTATE", async () => {
    await reset();
    let attempts = 0;
    await expectAbiError(
      database.transaction(async (tx) => {
        attempts += 1;
        await tx.query({ text: "INSERT INTO example_deferred (id) VALUES (1), (1)" });
      }),
      "MODULE_DATABASE_TRANSACTION_FAILED",
      "23505",
      ["example_deferred_id_key"],
    );
    expect(attempts).toBe(1);
    const count = await database.query({ text: "SELECT count(*)::int AS n FROM example_deferred" });
    expect(count.rows).toEqual([{ n: 0 }]);
  });

  it("never commits a transaction a failed statement aborted, even if the callback carries on", async () => {
    await reset();
    await expectAbiError(
      database.transaction(async (tx) => {
        await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
        await expectAbiError(
          tx.query({ text: "SELECT 1/0" }),
          "MODULE_DATABASE_QUERY_FAILED",
          "22012",
        );
        await expectAbiError(
          tx.query({ text: "SELECT 1" }),
          "MODULE_DATABASE_QUERY_FAILED",
          "25P02",
        );
        return "carried on";
      }),
      "MODULE_DATABASE_TRANSACTION_FAILED",
    );
    expect(await storedIds()).toEqual([]);
  });

  it("revokes an escaped executor, which then reaches nothing", async () => {
    await reset();
    let escaped: ModuleDatabaseTransaction | undefined;
    await database.transaction((tx) => {
      escaped = tx;
    });
    await expectAbiError(
      (escaped as ModuleDatabaseTransaction).query({
        text: "INSERT INTO example_records (id, value) VALUES (1, 'a')",
      }),
      "MODULE_DATABASE_SCOPE_CLOSED",
    );
    await expectAbiError(
      (escaped as ModuleDatabaseTransaction).savepoint(async () => undefined),
      "MODULE_DATABASE_SCOPE_CLOSED",
    );
    expect(await storedIds()).toEqual([]);
  });

  it("refuses writes in a read-only transaction", async () => {
    await reset();
    await expectAbiError(
      database.transaction(
        async (tx) => {
          expect((await tx.query({ text: "SHOW transaction_read_only" })).rows).toEqual([
            { transaction_read_only: "on" },
          ]);
          await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
        },
        { readOnly: true },
      ),
      "MODULE_DATABASE_QUERY_FAILED",
      "25006",
    );
    expect(await storedIds()).toEqual([]);
  });

  const levels: readonly [ModuleDatabaseIsolationLevel | undefined, string][] = [
    [undefined, "read committed"],
    ["read committed", "read committed"],
    ["repeatable read", "repeatable read"],
    ["serializable", "serializable"],
  ];
  for (const [isolation, expected] of levels) {
    it(`runs ${isolation ?? "the default"} isolation as ${expected}, read-write`, async () => {
      const rows = await database.transaction(
        async (tx) =>
          (
            await tx.query({
              text: "SELECT current_setting('transaction_isolation') AS level, current_setting('transaction_read_only') AS ro",
            })
          ).rows,
        isolation === undefined ? undefined : { isolation },
      );
      expect(rows).toEqual([{ level: expected, ro: "off" }]);
    });
  }

  it("overrides a database whose default isolation is different", async () => {
    const name = new URL(databaseUrl).pathname.slice(1);
    await withAdministrativeClient(databaseUrl, async (admin) => {
      await admin.query(
        `ALTER DATABASE ${quoteDatabaseName(name)} SET default_transaction_isolation TO 'serializable'`,
      );
    });
    // A new pool, so its connections start with the new database default.
    const freshPool = new Pool({ connectionString: databaseUrl, max: 1 });
    freshPool.on("error", () => undefined);
    try {
      const raw = await freshPool.query<{ level: string }>(
        "SELECT current_setting('default_transaction_isolation') AS level",
      );
      expect(raw.rows).toEqual([{ level: "serializable" }]);
      const fresh = createModuleDatabaseExecutor(freshPool, { moduleId: "example" });
      const rows = await fresh.transaction(
        async (tx) =>
          (await tx.query({ text: "SELECT current_setting('transaction_isolation') AS level" }))
            .rows,
      );
      expect(rows).toEqual([{ level: "read committed" }]);
    } finally {
      await freshPool.end();
      await withAdministrativeClient(databaseUrl, async (admin) => {
        await admin.query(
          `ALTER DATABASE ${quoteDatabaseName(name)} RESET default_transaction_isolation`,
        );
      });
    }
  });

  it("refuses invalid options before it starts anything", async () => {
    let called = false;
    for (const options of [
      { isolation: "read uncommitted" },
      { deferrable: true },
      { readOnly: "yes" },
      { isolation: "serializable", timeout: 5 },
    ]) {
      await expectAbiError(
        database.transaction(async () => {
          called = true;
        }, options as never),
        "MODULE_DATABASE_TRANSACTION_OPTION_INVALID",
      );
    }
    expect(called).toBe(false);
  });

  it("detects transaction control sent through query once PostgreSQL has run it, then sends nothing more and commits nothing", async () => {
    // Forbidden by ADR 0050, and not refusable beforehand without parsing SQL. Core detects each
    // from PostgreSQL's command tag or the protocol's transaction status.
    for (const statement of ["SAVEPOINT mine", "BEGIN", "START TRANSACTION", "ROLLBACK", "ABORT"]) {
      await reset();
      await expectAbiError(
        database.transaction(async (tx) => {
          await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
          await expectAbiError(tx.query({ text: statement }), "MODULE_DATABASE_REQUEST_INVALID");
          // Anything after the violation is refused without being sent: it would otherwise run
          // outside any transaction.
          await expectAbiError(
            tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }),
            "MODULE_DATABASE_REQUEST_INVALID",
          );
          await expectAbiError(
            tx.savepoint(async () => undefined),
            "MODULE_DATABASE_REQUEST_INVALID",
          );
          return "carried on";
        }),
        "MODULE_DATABASE_REQUEST_INVALID",
      );
      expect(await storedIds()).toEqual([]);
    }
  });

  it("cannot undo a COMMIT sent through query: PostgreSQL has committed, and Core reports the violation and commits nothing more", async () => {
    for (const statement of ["COMMIT", "END"]) {
      await reset();
      await expectAbiError(
        database.transaction(async (tx) => {
          await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
          await expectAbiError(tx.query({ text: statement }), "MODULE_DATABASE_REQUEST_INVALID");
          await expectAbiError(
            tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }),
            "MODULE_DATABASE_REQUEST_INVALID",
          );
          return "carried on";
        }),
        "MODULE_DATABASE_REQUEST_INVALID",
      );
      // The documented limit: the work before the module's own COMMIT was committed by PostgreSQL
      // when it ran that statement. Nothing after it ran.
      expect(await storedIds()).toEqual([1]);
    }
  });

  it("applies Core's statement bound inside a transaction, and transaction-local settings work", async () => {
    await expectAbiError(
      database.transaction(async (tx) => {
        await tx.query({ text: "SET LOCAL statement_timeout = 50" });
        await tx.query({ text: "SELECT pg_sleep(2)" });
      }),
      "MODULE_DATABASE_QUERY_FAILED",
      "57014",
    );
  });
});

// --- Forgotten await ----------------------------------------------------------------------------

describe("outstanding work", () => {
  it("never commits a transaction whose callback forgot to await a write", async () => {
    await reset();
    let forgotten: Promise<unknown> | undefined;
    await expectAbiError(
      database.transaction((tx) => {
        forgotten = tx.query({
          text: "INSERT INTO example_records (id, value) SELECT 1, 'a' FROM pg_sleep(0.2)",
        });
      }),
      "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING",
    );
    // The forgotten statement itself ran (Core let it settle before rolling back) but its write
    // was rolled back with the transaction.
    await expect(forgotten).resolves.toMatchObject({ command: "INSERT", rowCount: 1 });
    expect(await storedIds()).toEqual([]);
  });

  it("never commits when the callback awaited some writes and forgot one", async () => {
    await reset();
    await expectAbiError(
      database.transaction(async (tx) => {
        await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
        void tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" });
      }),
      "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING",
    );
    expect(await storedIds()).toEqual([]);
  });

  it("keeps the failing query's own error when one half of a Promise.all fails while the other is still running", async () => {
    await reset();
    await database.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
    let sibling: Promise<unknown> | undefined;
    await expectAbiError(
      database.transaction(async (tx) => {
        const conflicting = tx.query({
          text: "INSERT INTO example_records (id, value) VALUES (1, 'dup')",
        });
        sibling = tx.query({
          text: "INSERT INTO example_records (id, value) SELECT 2, 'b' FROM pg_sleep(0.1)",
        });
        sibling.catch(() => undefined);
        await Promise.all([conflicting, sibling]);
      }),
      // The PostgreSQL failure, not MODULE_DATABASE_SCOPE_WORK_OUTSTANDING.
      "MODULE_DATABASE_QUERY_FAILED",
      "23505",
    );
    // The sibling was still outstanding when the callback rejected: it settled afterwards (failing,
    // because the transaction was already aborted), before Core rolled back.
    await expect(sibling).rejects.toMatchObject({ sqlState: "25P02" });
    expect(await storedIds()).toEqual([1]);
  });

  it("keeps a savepoint's failing query error from a Promise.all, and the transaction carries on", async () => {
    await reset();
    await database.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
    await database.transaction(async (tx) => {
      await expectAbiError(
        tx.savepoint(async (sp) => {
          const sibling = sp.query({
            text: "INSERT INTO example_records (id, value) SELECT 2, 'b' FROM pg_sleep(0.1)",
          });
          sibling.catch(() => undefined);
          await Promise.all([
            sp.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'dup')" }),
            sibling,
          ]);
        }),
        "MODULE_DATABASE_QUERY_FAILED",
        "23505",
      );
      await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (3, 'c')" });
    });
    expect(await storedIds()).toEqual([1, 3]);
  });

  it("commits a Promise.all whose queries all succeed", async () => {
    await reset();
    await database.transaction(async (tx) => {
      await Promise.all([
        tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }),
        tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }),
      ]);
    });
    expect(await storedIds()).toEqual([1, 2]);
  });

  it("treats an unawaited savepoint as outstanding work", async () => {
    await reset();
    let forgotten: Promise<unknown> | undefined;
    await expectAbiError(
      database.transaction((tx) => {
        forgotten = tx.savepoint(async (sp) => {
          await sp.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
        });
        // Observed below; handled now so its rejection is never briefly unhandled.
        forgotten.catch(() => undefined);
      }),
      "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING",
    );
    await expect(forgotten).rejects.toMatchObject({ code: "MODULE_DATABASE_SCOPE_CLOSED" });
    expect(await storedIds()).toEqual([]);
  });
});

// --- Savepoints ---------------------------------------------------------------------------------

describe("savepoints", () => {
  it("releases a resolved savepoint, keeping its work", async () => {
    await reset();
    const value = await database.transaction(async (tx) =>
      tx.savepoint(async (sp) => {
        await sp.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
        return "kept";
      }),
    );
    expect(value).toBe("kept");
    expect(await storedIds()).toEqual([1]);
  });

  it("rolls back a rejected savepoint and lets the transaction carry on and commit", async () => {
    await reset();
    const own = new Error("savepoint failure");
    await database.transaction(async (tx) => {
      await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
      const error = await rejection(
        tx.savepoint(async (sp) => {
          await sp.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" });
          throw own;
        }),
      );
      expect(error).toBe(own);
      await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (3, 'c')" });
    });
    expect(await storedIds()).toEqual([1, 3]);
  });

  it("recovers from a failed statement inside a savepoint, the usual way to continue after one", async () => {
    await reset();
    await database.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
    await database.transaction(async (tx) => {
      await expectAbiError(
        tx.savepoint(async (sp) => {
          await sp.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'dup')" });
        }),
        "MODULE_DATABASE_QUERY_FAILED",
        "23505",
      );
      await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" });
    });
    expect(await storedIds()).toEqual([1, 2]);
  });

  it("reports a savepoint whose callback swallowed a failed statement, and keeps the transaction usable", async () => {
    await reset();
    await database.transaction(async (tx) => {
      await expectAbiError(
        tx.savepoint(async (sp) => {
          await sp.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
          await rejection(sp.query({ text: "SELECT 1/0" }));
          return "swallowed";
        }),
        "MODULE_DATABASE_TRANSACTION_FAILED",
        "25P02",
      );
      await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" });
    });
    expect(await storedIds()).toEqual([2]);
  });

  it("nests and sequences savepoints, keeping exactly the work that was released", async () => {
    await reset();
    await database.transaction(async (tx) => {
      await tx.savepoint(async (outer) => {
        await outer.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
        await rejection(
          outer.savepoint(async (inner) => {
            await inner.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" });
            throw new Error("inner failure");
          }),
        );
        await outer.savepoint(async (inner) => {
          await inner.query({ text: "INSERT INTO example_records (id, value) VALUES (3, 'c')" });
        });
      });
      await tx.savepoint(async (second) => {
        await second.query({ text: "INSERT INTO example_records (id, value) VALUES (4, 'd')" });
      });
    });
    expect(await storedIds()).toEqual([1, 3, 4]);
  });

  it("revokes an escaped savepoint executor", async () => {
    await reset();
    await database.transaction(async (tx) => {
      let escaped: ModuleDatabaseTransaction | undefined;
      await tx.savepoint((sp) => {
        escaped = sp;
      });
      await expectAbiError(
        (escaped as ModuleDatabaseTransaction).query({
          text: "INSERT INTO example_records (id, value) VALUES (1, 'a')",
        }),
        "MODULE_DATABASE_SCOPE_CLOSED",
      );
    });
    expect(await storedIds()).toEqual([]);
  });

  it("refuses overlapping scopes: the parent is inactive while a savepoint is open", async () => {
    await reset();
    await database.transaction(async (tx) => {
      await tx.savepoint(async (sp) => {
        await expectAbiError(
          tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }),
          "MODULE_DATABASE_SCOPE_INACTIVE",
        );
        await expectAbiError(
          tx.savepoint(async () => undefined),
          "MODULE_DATABASE_SCOPE_INACTIVE",
        );
        await sp.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" });
      });
      await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (3, 'c')" });
    });
    expect(await storedIds()).toEqual([2, 3]);
  });

  it("rolls back, never releases, a savepoint whose callback forgot to await", async () => {
    await reset();
    await database.transaction(async (tx) => {
      await expectAbiError(
        tx.savepoint((sp) => {
          void sp.query({
            text: "INSERT INTO example_records (id, value) SELECT 1, 'a' FROM pg_sleep(0.2)",
          });
        }),
        "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING",
      );
      await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" });
    });
    expect(await storedIds()).toEqual([2]);
  });
});

// --- Pool safety and session state --------------------------------------------------------------

interface BorrowerState {
  readonly bytea: string;
  readonly datestyle: string;
  readonly listening: number;
  readonly locks: number;
  readonly status: string | null;
  readonly temporaryTables: number;
  readonly timezone: string;
  readonly workMem: string;
}

/** What the next borrower of a one-connection pool finds, through the raw driver, as Core's own
 * stores would. */
async function nextBorrower(onePool: Pool): Promise<BorrowerState> {
  const client = (await onePool.connect()) as PoolClient & {
    getTransactionStatus(): string | null;
  };
  try {
    const status = client.getTransactionStatus();
    const result = await client.query<{
      bytea: string;
      datestyle: string;
      listening: number;
      locks: number;
      temporary_tables: number;
      timezone: string;
      work_mem: string;
    }>(
      [
        "SELECT current_setting('TimeZone') AS timezone, current_setting('DateStyle') AS datestyle,",
        "  current_setting('bytea_output') AS bytea, current_setting('work_mem') AS work_mem,",
        "  (SELECT count(*)::int FROM pg_listening_channels()) AS listening,",
        "  (SELECT count(*)::int FROM pg_locks WHERE locktype = 'advisory' AND pid = pg_backend_pid()) AS locks,",
        "  (SELECT count(*)::int FROM pg_class WHERE relnamespace = pg_my_temp_schema()) AS temporary_tables",
      ].join(" "),
    );
    const row = result.rows[0];
    if (row === undefined) {
      throw new IntegrationHarnessError("The borrower probe returned no row.");
    }
    return {
      bytea: row.bytea,
      datestyle: row.datestyle,
      listening: row.listening,
      locks: row.locks,
      status,
      temporaryTables: row.temporary_tables,
      timezone: row.timezone,
      workMem: row.work_mem,
    };
  } finally {
    client.release();
  }
}

/** Exactly what a fresh connection to the hostile database has. */
const clean: BorrowerState = {
  bytea: "escape",
  datestyle: "SQL, DMY",
  listening: 0,
  locks: 0,
  status: "I",
  temporaryTables: 0,
  timezone: "America/Chicago",
  workMem: "4MB",
};

describe("pool safety", () => {
  let onePool: Pool;
  let single: ModuleDatabaseExecutor;

  beforeAll(() => {
    onePool = new Pool({ connectionString: databaseUrl, max: 1 });
    onePool.on("error", () => undefined);
    single = createModuleDatabaseExecutor(onePool, { moduleId: "example" });
  });

  afterAll(async () => {
    await onePool?.end();
  });

  const failures: readonly [string, (executor: ModuleDatabaseExecutor) => Promise<unknown>][] = [
    ["a successful query", (executor) => executor.query({ text: "SELECT 1" })],
    [
      "a successful transaction",
      (executor) => executor.transaction(async (tx) => tx.query({ text: "SELECT 1" })),
    ],
    [
      "a callback throw",
      (executor) => executor.transaction(async () => Promise.reject(new Error("x"))),
    ],
    ["a SQL error", (executor) => executor.query({ text: "SELECT 1/0" })],
    [
      "a SQL error inside a transaction",
      (executor) => executor.transaction((tx) => tx.query({ text: "SELECT 1/0" })),
    ],
    [
      "an aborted transaction the callback ignored",
      (executor) =>
        executor.transaction(async (tx) => {
          await rejection(tx.query({ text: "SELECT 1/0" }));
        }),
    ],
    [
      "escaped executor use",
      async (executor) => {
        let escaped: ModuleDatabaseTransaction | undefined;
        await executor.transaction((tx) => {
          escaped = tx;
        });
        return (escaped as ModuleDatabaseTransaction).query({ text: "SELECT 1" });
      },
    ],
    [
      "outstanding work",
      (executor) =>
        executor.transaction((tx) => {
          void tx.query({ text: "SELECT pg_sleep(0.1)" });
        }),
    ],
    [
      "a savepoint rollback",
      (executor) =>
        executor.transaction(async (tx) => {
          await rejection(tx.savepoint(async () => Promise.reject(new Error("x"))));
        }),
    ],
    [
      "an outstanding savepoint",
      (executor) =>
        executor.transaction((tx) => {
          // The module ignores the savepoint's own rejection; it is handled here only so the test
          // run has no unhandled rejection.
          tx.savepoint((sp) => sp.query({ text: "SELECT pg_sleep(0.1)" })).catch(() => undefined);
        }),
    ],
    ["a refused top-level BEGIN", (executor) => executor.query({ text: "BEGIN" })],
    [
      "a refused COMMIT inside a transaction",
      (executor) => executor.transaction((tx) => tx.query({ text: "COMMIT" })),
    ],
    ["session-level settings", (executor) => executor.query({ text: "SET work_mem = '64MB'" })],
    [
      "session-level settings inside a transaction",
      (executor) =>
        executor.transaction(async (tx) => {
          await tx.query({ text: "SET work_mem = '64MB'" });
          await tx.query({ text: "SET TimeZone = 'Asia/Tokyo'" });
        }),
    ],
    [
      "a temporary table",
      (executor) => executor.query({ text: "CREATE TEMPORARY TABLE scratch (x int)" }),
    ],
    ["LISTEN", (executor) => executor.query({ text: "LISTEN example_channel" })],
    [
      "LISTEN inside a transaction",
      (executor) => executor.transaction((tx) => tx.query({ text: "LISTEN example_channel" })),
    ],
    [
      "a session advisory lock",
      (executor) => executor.query({ text: "SELECT pg_advisory_lock(42)" }),
    ],
    ["a prepared statement", (executor) => executor.query({ text: "PREPARE mine AS SELECT 1" })],
    [
      "a terminated connection",
      (executor) =>
        executor.transaction(async (tx) => {
          await tx.query({ text: "SELECT pg_terminate_backend(pg_backend_pid())" });
        }),
    ],
  ];

  for (const [label, run] of failures) {
    it(`hands the next borrower a clean connection after ${label}`, async () => {
      await run(single).catch(() => undefined);
      expect(await nextBorrower(onePool)).toEqual(clean);
      // And the ABI keeps working on it.
      expect((await single.query({ text: "SELECT 1 AS n" })).rows).toEqual([{ n: 1 }]);
    });
  }

  it("destroys, never reuses, a connection after transaction-control misuse", async () => {
    const pid = async (executor: ModuleDatabaseExecutor | ModuleDatabaseTransaction) =>
      (
        (await executor.query({ text: "SELECT pg_backend_pid() AS pid" })).rows[0] as {
          pid: number;
        }
      ).pid;
    for (const statement of ["COMMIT", "ROLLBACK", "SAVEPOINT mine", "BEGIN"]) {
      let used: number | undefined;
      await rejection(
        single.transaction(async (tx) => {
          used = await pid(tx);
          await rejection(tx.query({ text: statement }));
        }),
      );
      expect(used).toBeTypeOf("number");
      expect(await pid(single)).not.toBe(used);
      expect(await nextBorrower(onePool)).toEqual(clean);
    }
    for (const statement of ["BEGIN", "COMMIT"]) {
      const before = await pid(single);
      // With one connection, the next top-level query would reuse it unless it was destroyed.
      await rejection(single.query({ text: statement }));
      expect(await pid(single)).not.toBe(before);
      expect(await nextBorrower(onePool)).toEqual(clean);
    }
  });

  it("reports a terminated connection without a SQLSTATE of its own making", async () => {
    const error = await rejection(
      single.transaction(async (tx) => {
        await tx.query({ text: "SELECT pg_terminate_backend(pg_backend_pid())" });
      }),
    );
    expect(isModuleDatabaseError(error)).toBe(true);
    // PostgreSQL may report the termination itself (57P01) or the driver may only see the socket
    // close; either way the code is Core's, and any SQLSTATE is PostgreSQL's.
    expect((error as { code: string }).code).toMatch(
      /^MODULE_DATABASE_(QUERY_FAILED|TRANSACTION_FAILED)$/u,
    );
    expect([undefined, "57P01"]).toContain((error as { sqlState?: string }).sqlState);
    expect(await nextBorrower(onePool)).toEqual(clean);
  });
});

describe("session state", () => {
  it("lets no session setting, temporary table, or listener carry from one query to the next", async () => {
    await database.query({ text: "SET work_mem = '64MB'" });
    await database.query({ text: "CREATE TEMPORARY TABLE scratch (x int)" });
    await database.query({ text: "LISTEN example_channel" });
    const result = await database.query({
      text: [
        "SELECT current_setting('work_mem') AS work_mem,",
        "  to_regclass('pg_temp.scratch') IS NULL AS no_scratch,",
        "  (SELECT count(*)::int FROM pg_listening_channels()) AS listening",
      ].join(" "),
    });
    expect(result.rows).toEqual([{ listening: 0, no_scratch: true, work_mem: "4MB" }]);
  });

  it("keeps transaction-local settings for the rest of the transaction, and no longer", async () => {
    const inside = await database.transaction(async (tx) => {
      await tx.query({ text: "SET LOCAL work_mem = '64MB'" });
      return (await tx.query({ text: "SELECT current_setting('work_mem') AS work_mem" })).rows;
    });
    expect(inside).toEqual([{ work_mem: "64MB" }]);
    const after = await database.query({ text: "SELECT current_setting('work_mem') AS work_mem" });
    expect(after.rows).toEqual([{ work_mem: "4MB" }]);
  });

  it("pins the ABI's settings for every operation, and never leaves them for Core's own borrowers", async () => {
    const pinned = await database.query({
      text: [
        "SELECT current_setting('TimeZone') AS tz, current_setting('DateStyle') AS ds,",
        "  current_setting('IntervalStyle') AS iv, current_setting('bytea_output') AS bytea,",
        "  current_setting('extra_float_digits') AS digits",
      ].join(" "),
    });
    expect(pinned.rows).toEqual([
      { bytea: "hex", digits: "1", ds: "ISO, MDY", iv: "postgres", tz: "UTC" },
    ]);
    const core = await pool.query<{ tz: string }>("SELECT current_setting('TimeZone') AS tz");
    expect(core.rows).toEqual([{ tz: "America/Chicago" }]);
  });

  it("keeps Core's own connection-level statement bound after resetting a session", async () => {
    await database.query({ text: "SET statement_timeout = 0" });
    const bound = await pool.query<{ timeout: string }>(
      "SELECT current_setting('statement_timeout') AS timeout",
    );
    expect(bound.rows).toEqual([{ timeout: "20s" }]);
  });
});