orca/cloud/apps/relay/src/database.ts

import { mkdirSync } from 'node:fs'
import { performance } from 'node:perf_hooks'
import { join } from 'node:path'
import { DatabaseSync } from 'node:sqlite'
import pg from 'pg'
import { RELAY_REGIONS } from '@orca-cloud/relay-contract'
import {
  emptyPostgresPoolPressureCounts,
  isPostgresPoolConnectFailure,
  PostgresPoolPressure,
  type PostgresPoolPressureCounts
} from './postgres-pool-pressure.js'
import { applyPostgresSchema } from './postgres-schema-startup.js'
import { POSTGRES_STATEMENT_STATS_MIGRATION } from './postgres-statement-stats.js'
import { reportPostgresQueryFailure } from './postgres-query-failure.js'
import {
  CellInventoryHoldSamples,
  emptyCellInventoryHoldCounts,
  type CellInventoryHoldCounts,
  type CellLockHoldSite
} from './cell-inventory-hold-samples.js'

export const POSTGRES_LOCK_TIMEOUT_MS = 1_000

function setLocalLockTimeout(milliseconds: number): string {
  if (!Number.isInteger(milliseconds) || milliseconds < 1) {
    throw new Error('invalid_lock_timeout')
  }
  return `SET LOCAL lock_timeout = '${milliseconds}ms'`
}

// Region CHECK lists come from the contract so a new region cannot leave a
// column rejecting values the rest of the relay already accepts.
const REGION_LIST = RELAY_REGIONS.map((region) => `'${region}'`).join(', ')

// A host that was just moved is not a candidate again for this long, so a
// desktop whose region probe flips cannot walk itself back and forth.
export const REGIONAL_REHOME_DEFAULT_HOST_COOLDOWN_MS = 7 * 24 * 60 * 60_000

export type SqlRow = Record<string, unknown>
export type RelayLockOptions = {
  failIfUnavailable?: boolean
  // Only honoured inside a transaction: SET LOCAL is a no-op in autocommit.
  lockTimeoutMs?: number
  // Report how long this lock is held to COMMIT. The hold, not the wait, is what
  // forms the queue, and nothing measured it before.
  measureHoldMs?: boolean
  // Labels the measured hold; unset reads as the full-inventory lock.
  holdSite?: CellLockHoldSite
  // The statement carries its own row-lock clause (a CTE locks inside the
  // statement), so none is appended. failIfUnavailable must match that clause.
  lockClauseInStatement?: boolean
}

type MeasuredHold = { holdMs: number; site: CellLockHoldSite }

// A transaction that can report how long it held a measured lock before COMMIT.
type HoldMeasuringTransaction = { consumeHold(): MeasuredHold | undefined }

function recordMeasuredHold(holds: CellInventoryHoldSamples, transaction: unknown): void {
  const hold = (transaction as HoldMeasuringTransaction).consumeHold?.()
  if (hold) holds.record(hold.holdMs, hold.site)
}

function lockedStatement(sql: string, options: RelayLockOptions): string {
  if (options.lockClauseInStatement) return sql
  return `${sql} FOR UPDATE${options.failIfUnavailable ? ' NOWAIT' : ''}`
}
export type RelayTransactionOptions = { reportRetries?: boolean }

export interface RelayDatabase {
  readonly dialect?: 'sqlite' | 'postgres'
  query(sql: string, params?: unknown[]): Promise<SqlRow[]>
  queryLocked(
    sql: string,
    params?: unknown[],
    options?: RelayLockOptions
  ): Promise<SqlRow[]>
  transaction<T>(
    operation: (transaction: RelayDatabase) => Promise<T>,
    options?: RelayTransactionOptions
  ): Promise<T>
  // Only on a PostgreSQL transaction handle; see commitWithFinalWrite.
  commitWithFinal?(sql: string, params?: unknown[]): Promise<boolean>
  close(): Promise<void>
}

// Runs a single-row write with RETURNING as the transaction's last statement and reports
// whether it changed a row. On PostgreSQL the write and COMMIT go as one message, so the
// row lock is held for no round trip; a false result has already rolled the transaction back.
export async function commitWithFinalWrite(
  database: RelayDatabase,
  sql: string,
  params: unknown[] = []
): Promise<boolean> {
  if (database.commitWithFinal) return await database.commitWithFinal(sql, params)
  return (await database.query(sql, params)).length > 0
}

// RULE - no new index and no new column on `relay_control_connection_reservations`,
// `relay_confirm_results`, `relay_audit_events`, `relay_connection_bases`, or any other large
// table may be added to SCHEMA or to POSTGRES_SCHEMA_MIGRATIONS. The catalog pre-check skips a
// lock-taking statement only once the object exists, so a brand-new one reports missing on every
// director at once and each runs a non-concurrent build over the whole table. POSTGRES_LOCK_TIMEOUT_MS
// bounds how long that build waits for its lock, not how long it holds it. Build the index out of
// band with CREATE INDEX CONCURRENTLY first, then add it here, where the pre-check skips it forever
// after. relay-schema-lock-targets.test.ts pins the current list, so an addition fails CI.
// Constraint swaps are matched by NAME in pg_constraint, never by body, because the CHECK list is
// generated from REGION_LIST. Changing a constraint's definition under the same name therefore does
// nothing on boot: an operator drops it, and the next boot adds the current definition back.
// relay_confirmable_splices, relay_cell_drain_attempts and relay_migration_leases are no longer
// created; nothing ever wrote them. Databases that have them keep them empty until a drop is safe:
// an older image still creates them at boot, and a drop racing that CREATE can fail its schema step.
// Account erasure in orca-cloud must be deployed with retired-table support (orca-cloud#493) first.
const SCHEMA = `
CREATE TABLE IF NOT EXISTS relay_invites (
  user_id TEXT NOT NULL,
  relay_host_id TEXT NOT NULL,
  relay_device_id TEXT NOT NULL,
  token_hash TEXT NOT NULL UNIQUE,
  state TEXT NOT NULL,
  attempt_count BIGINT NOT NULL,
  max_attempts BIGINT NOT NULL,
  expires_at BIGINT NOT NULL,
  reservation_id TEXT,
  reservation_expires_at BIGINT,
  cooldown_until BIGINT,
  created_at BIGINT NOT NULL,
  updated_at BIGINT NOT NULL,
  PRIMARY KEY (user_id, relay_host_id, relay_device_id, token_hash)
);
CREATE INDEX IF NOT EXISTS relay_invites_device
  ON relay_invites(user_id, relay_host_id, relay_device_id);

-- schema-deferrable: created out of band, so a boot that cannot take the lock must retry
-- Why: the credential sweep matches (state, expires_at) every cycle while invites in a terminal
-- state accumulate for the life of the database. Unindexed it seq-scans the whole table inside the
-- maintenance transaction. Partial, so the index holds only the states the sweep can act on.
CREATE INDEX IF NOT EXISTS relay_invites_sweep_expiry
  ON relay_invites(expires_at) WHERE state IN ('available', 'reserved', 'cooldown');

-- schema-deferrable: created out of band, so a boot that cannot take the lock must retry
-- Why: the second sweep pass matches (state, reservation_expires_at) over the same table.
CREATE INDEX IF NOT EXISTS relay_invites_sweep_reservation
  ON relay_invites(reservation_expires_at) WHERE state = 'reserved';

CREATE TABLE IF NOT EXISTS relay_devices (
  user_id TEXT NOT NULL,
  relay_host_id TEXT NOT NULL,
  relay_device_id TEXT NOT NULL,
  current_hash TEXT NOT NULL,
  current_version BIGINT NOT NULL,
  current_expires_at BIGINT NOT NULL,
  grace_hash TEXT,
  grace_version BIGINT,
  grace_expires_at BIGINT,
  revoked_at BIGINT,
  updated_at BIGINT NOT NULL,
  PRIMARY KEY (user_id, relay_host_id, relay_device_id)
);
CREATE INDEX IF NOT EXISTS relay_devices_current_hash ON relay_devices(relay_host_id, current_hash);
CREATE INDEX IF NOT EXISTS relay_devices_grace_hash ON relay_devices(relay_host_id, grace_hash);

CREATE TABLE IF NOT EXISTS relay_install_results (
  user_id TEXT NOT NULL,
  relay_host_id TEXT NOT NULL,
  relay_device_id TEXT NOT NULL,
  req_id TEXT NOT NULL,
  authorization_mode TEXT NOT NULL,
  result_json TEXT NOT NULL,
  committed_at BIGINT NOT NULL,
  PRIMARY KEY (user_id, relay_host_id, relay_device_id, req_id)
);

CREATE TABLE IF NOT EXISTS relay_connection_bases (
  basis_conn_id TEXT PRIMARY KEY,
  user_id TEXT NOT NULL,
  relay_host_id TEXT NOT NULL,
  relay_device_id TEXT NOT NULL,
  owning_control_generation BIGINT NOT NULL,
  credential_kind TEXT NOT NULL,
  invite_token_hash TEXT,
  accepted_credential_version BIGINT,
  accepted_as TEXT,
  deadline BIGINT NOT NULL,
  active BIGINT NOT NULL,
  created_at BIGINT NOT NULL
);

-- Why: the maintenance sweep matches (active, deadline) while inactive bases
-- accumulate unboundedly. Unindexed it seq-scans millions of rows every cycle
-- and holds the maintenance transaction open long enough to time out
-- assignment lock waits.
-- Why not a partial index on active = 1: a basis is inserted active and flipped to 0, so each
-- deactivation leaves a dead entry in that index too. Measured on production-shaped history it
-- carries the same dead entries as this one, the planner picks this one in every state, and it
-- costs ~65 bytes of WAL per insert. Bloat here is cured by reaping and vacuum, not by a narrower
-- index.
CREATE INDEX IF NOT EXISTS relay_connection_bases_active_deadline
  ON relay_connection_bases(active, deadline);

CREATE TABLE IF NOT EXISTS relay_direct_authorizations (
  direct_auth_id TEXT PRIMARY KEY,
  user_id TEXT NOT NULL,
  relay_host_id TEXT NOT NULL,
  relay_device_id TEXT NOT NULL,
  owning_control_generation BIGINT NOT NULL,
  deadline BIGINT NOT NULL,
  consumed_at BIGINT
);

-- schema-deferrable: created out of band, so a boot that cannot take the lock must retry
-- Why: the sweep expires pending authorizations by (consumed_at IS NULL, deadline), and consumed
-- rows are never deleted. Partial, so the index stays the size of the pending set.
CREATE INDEX IF NOT EXISTS relay_direct_authorizations_pending_deadline
  ON relay_direct_authorizations(deadline) WHERE consumed_at IS NULL;

CREATE TABLE IF NOT EXISTS relay_confirm_results (
  user_id TEXT NOT NULL,
  relay_host_id TEXT NOT NULL,
  req_id TEXT NOT NULL,
  basis_conn_id TEXT NOT NULL,
  tuple_json TEXT NOT NULL,
  result_json TEXT NOT NULL,
  committed_at BIGINT NOT NULL,
  PRIMARY KEY (user_id, relay_host_id, req_id)
);

CREATE TABLE IF NOT EXISTS relay_assignments (
  user_id TEXT NOT NULL,
  relay_host_id TEXT NOT NULL,
  cell_id TEXT NOT NULL,
  assignment_epoch BIGINT NOT NULL,
  lease_expires_at BIGINT NOT NULL,
  last_activity_at BIGINT NOT NULL,
  reserved_controls BIGINT NOT NULL,
  reserved_splices BIGINT NOT NULL,
  reserved_invites BIGINT NOT NULL,
  pending_installs BIGINT NOT NULL,
  pending_confirmations BIGINT NOT NULL,
  migration_leases BIGINT NOT NULL,
  PRIMARY KEY (user_id, relay_host_id)
);

CREATE TABLE IF NOT EXISTS relay_assignment_region_preferences (
  user_id TEXT NOT NULL,
  relay_host_id TEXT NOT NULL,
  preferred_region TEXT NOT NULL
    CHECK (preferred_region IN (${REGION_LIST})),
  observed_at BIGINT NOT NULL,
  PRIMARY KEY (user_id, relay_host_id)
);
CREATE INDEX IF NOT EXISTS relay_assignment_region_preferences_observed
  ON relay_assignment_region_preferences(observed_at);

CREATE TABLE IF NOT EXISTS relay_region_decisions (
  user_id TEXT NOT NULL, relay_host_id TEXT NOT NULL,
  generation BIGINT NOT NULL, expires_at BIGINT NOT NULL,
  assignment_epoch BIGINT NOT NULL, incumbent_region TEXT NOT NULL,
  policy_version BIGINT NOT NULL, outcome TEXT NOT NULL,
  cohort_bucket BIGINT NOT NULL DEFAULT 0,
  last_considered_at BIGINT NOT NULL DEFAULT 0,
  preferred_region TEXT, observed_at BIGINT NOT NULL, report_json TEXT,
  PRIMARY KEY (user_id, relay_host_id)
);
CREATE TABLE IF NOT EXISTS relay_control_capabilities (
  user_id TEXT NOT NULL, relay_host_id TEXT NOT NULL, activity_id TEXT NOT NULL,
  cell_id TEXT NOT NULL, cell_incarnation TEXT NOT NULL,
  assignment_epoch BIGINT NOT NULL, generation BIGINT NOT NULL,
  finish_existing BIGINT NOT NULL,
  idle_regional_rehome BIGINT NOT NULL DEFAULT 0,
  PRIMARY KEY (user_id, relay_host_id, activity_id)
);
CREATE TABLE IF NOT EXISTS relay_region_rehome_worker_state (
  worker_id TEXT PRIMARY KEY,
  next_dispatch_at BIGINT NOT NULL,
  paused_until BIGINT NOT NULL,
  consecutive_failures BIGINT NOT NULL,
  updated_at BIGINT NOT NULL
);

CREATE TABLE IF NOT EXISTS relay_region_rehome_control (
  control_id TEXT PRIMARY KEY,
  generation BIGINT NOT NULL,
  enabled BIGINT NOT NULL,
  observation_started_at BIGINT NOT NULL,
  not_before BIGINT NOT NULL,
  rate_per_minute BIGINT NOT NULL,
  preference_max_age_ms BIGINT NOT NULL,
  host_cooldown_ms BIGINT NOT NULL
    DEFAULT ${REGIONAL_REHOME_DEFAULT_HOST_COOLDOWN_MS},
  drain_grace_ms BIGINT NOT NULL,
  updated_at BIGINT NOT NULL
);

CREATE TABLE IF NOT EXISTS relay_region_rehome_attempts (
  attempt_id TEXT PRIMARY KEY,
  user_id TEXT NOT NULL,
  relay_host_id TEXT NOT NULL,
  preferred_region TEXT NOT NULL
    CONSTRAINT relay_region_rehome_attempts_preferred_region_valid
    CHECK (preferred_region IN (${REGION_LIST})),
  source_cell_id TEXT NOT NULL,
  source_cell_incarnation TEXT NOT NULL,
  source_generation BIGINT NOT NULL DEFAULT 0,
  target_cell_id TEXT NOT NULL,
  target_cell_incarnation TEXT NOT NULL,
  previous_epoch BIGINT NOT NULL,
  assignment_epoch BIGINT NOT NULL,
  drain_grace_ms BIGINT NOT NULL,
  send_attempts BIGINT NOT NULL,
  last_send_attempt_at BIGINT,
  drain_receipt_at BIGINT,
  drain_outcome TEXT CHECK (
    drain_outcome IN ('accepted', 'already-accepted', 'host-not-connected')
  ),
  completed_at BIGINT,
  aborted_at BIGINT,
  abort_reason TEXT,
  created_at BIGINT NOT NULL,
  updated_at BIGINT NOT NULL,
  UNIQUE (user_id, relay_host_id, assignment_epoch)
);
CREATE INDEX IF NOT EXISTS relay_region_rehome_attempts_pending
  ON relay_region_rehome_attempts(drain_receipt_at, last_send_attempt_at, completed_at, aborted_at);
CREATE INDEX IF NOT EXISTS relay_region_rehome_attempts_host_recency
  ON relay_region_rehome_attempts(user_id, relay_host_id, created_at);

CREATE TABLE IF NOT EXISTS relay_cells (
  cell_id TEXT PRIMARY KEY,
  cell_url TEXT NOT NULL UNIQUE,
  enabled BIGINT NOT NULL,
  capacity_requests BIGINT NOT NULL,
  reserved_requests BIGINT NOT NULL,
  observed_requests BIGINT NOT NULL,
  last_heartbeat_at BIGINT NOT NULL,
  updated_at BIGINT NOT NULL
);

CREATE TABLE IF NOT EXISTS relay_cell_regions (
  cell_id TEXT PRIMARY KEY,
  region TEXT NOT NULL CHECK (region IN (${REGION_LIST}))
);

CREATE TABLE IF NOT EXISTS relay_cell_admission (
  cell_id TEXT PRIMARY KEY,
  admission_state TEXT NOT NULL
    CHECK (admission_state IN ('existing-only', 'migration-only', 'general')),
  updated_at BIGINT NOT NULL,
  roll_isolated_at BIGINT
);

CREATE TABLE IF NOT EXISTS relay_admission_selectors (
  selector_id TEXT PRIMARY KEY,
  generation BIGINT NOT NULL,
  attempt_id TEXT,
  membership_json TEXT NOT NULL,
  updated_at BIGINT NOT NULL
);

CREATE TABLE IF NOT EXISTS relay_admission_selector_intents (
  attempt_id TEXT PRIMARY KEY,
  expected_generation BIGINT NOT NULL,
  intended_generation BIGINT NOT NULL,
  previous_membership_json TEXT NOT NULL,
  membership_json TEXT NOT NULL,
  created_at BIGINT NOT NULL,
  committed_at BIGINT
);

CREATE TABLE IF NOT EXISTS relay_admission_selector_cell_additions (
  attempt_id TEXT PRIMARY KEY,
  cells_json TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS relay_cell_runtime (
  cell_id TEXT PRIMARY KEY,
  cell_url TEXT NOT NULL,
  cell_incarnation TEXT NOT NULL,
  started_at BIGINT NOT NULL,
  ready BIGINT NOT NULL,
  observed_requests BIGINT NOT NULL,
  last_heartbeat_at BIGINT NOT NULL,
  updated_at BIGINT NOT NULL
);
CREATE INDEX IF NOT EXISTS relay_cell_runtime_heartbeat
  ON relay_cell_runtime(ready, last_heartbeat_at);

CREATE TABLE IF NOT EXISTS relay_cell_capabilities (
  cell_id TEXT PRIMARY KEY,
  cell_incarnation TEXT NOT NULL,
  regional_rehome_protocol BIGINT NOT NULL,
  last_heartbeat_at BIGINT NOT NULL
);

CREATE TABLE IF NOT EXISTS relay_cell_rehome_safety (
  cell_id TEXT PRIMARY KEY,
  cell_incarnation TEXT NOT NULL,
  observed_at BIGINT NOT NULL,
  sql_failures BIGINT NOT NULL,
  reconnects BIGINT NOT NULL,
  control_activity_recovery_failures BIGINT NOT NULL,
  database_pool_waiting BIGINT NOT NULL,
  database_pool_waiters_max BIGINT NOT NULL,
  database_pool_wait_ms_max BIGINT NOT NULL
);

CREATE TABLE IF NOT EXISTS relay_cell_connection_limits (
  cell_id TEXT PRIMARY KEY,
  hard_cap BIGINT NOT NULL,
  unobserved_bound BIGINT NOT NULL,
  updated_at BIGINT NOT NULL
);

CREATE TABLE IF NOT EXISTS relay_cell_connection_runtime (
  cell_id TEXT PRIMARY KEY,
  cell_incarnation TEXT NOT NULL,
  total_connections BIGINT NOT NULL,
  in_flight_connections BIGINT NOT NULL,
  reserved_connection_units BIGINT NOT NULL,
  enforced_connection_units BIGINT NOT NULL,
  last_heartbeat_at BIGINT NOT NULL,
  updated_at BIGINT NOT NULL
);
CREATE INDEX IF NOT EXISTS relay_cell_connection_runtime_heartbeat
  ON relay_cell_connection_runtime(last_heartbeat_at);

CREATE TABLE IF NOT EXISTS relay_cell_connection_snapshots (
  cell_id TEXT PRIMARY KEY,
  cell_incarnation TEXT NOT NULL,
  inclusion_watermark BIGINT NOT NULL,
  total_connections BIGINT NOT NULL,
  in_flight_connections BIGINT NOT NULL,
  reserved_connection_units BIGINT NOT NULL,
  enforced_connection_units BIGINT NOT NULL,
  snapshot_at BIGINT NOT NULL
);
CREATE INDEX IF NOT EXISTS relay_cell_connection_snapshot_freshness
  ON relay_cell_connection_snapshots(snapshot_at);

CREATE TABLE IF NOT EXISTS relay_cell_fences (
  cell_id TEXT PRIMARY KEY,
  cell_incarnation TEXT NOT NULL,
  attested_at BIGINT NOT NULL,
  expires_at BIGINT NOT NULL
);
CREATE INDEX IF NOT EXISTS relay_cell_fences_expiry
  ON relay_cell_fences(expires_at);

CREATE TABLE IF NOT EXISTS relay_cell_committed_fences (
  cell_id TEXT PRIMARY KEY,
  attempt_id TEXT NOT NULL UNIQUE,
  cell_incarnation TEXT NOT NULL,
  attested_at BIGINT NOT NULL,
  expires_at BIGINT NOT NULL
);
CREATE INDEX IF NOT EXISTS relay_cell_committed_fences_expiry
  ON relay_cell_committed_fences(expires_at);

CREATE TABLE IF NOT EXISTS relay_cell_legacy_fence_adoptions (
  cell_id TEXT PRIMARY KEY,
  cell_incarnation TEXT NOT NULL,
  attested_at BIGINT NOT NULL,
  expires_at BIGINT NOT NULL
);
CREATE INDEX IF NOT EXISTS relay_cell_legacy_fence_adoptions_expiry
  ON relay_cell_legacy_fence_adoptions(expires_at);

CREATE TABLE IF NOT EXISTS relay_cell_fence_attempts (
  attempt_id TEXT PRIMARY KEY,
  environment TEXT NOT NULL,
  cell_id TEXT NOT NULL,
  cell_incarnation TEXT NOT NULL,
  mig_name TEXT NOT NULL,
  instance_group TEXT NOT NULL,
  generation_identity TEXT NOT NULL,
  fence_commit TEXT NOT NULL,
  plan_sha256 TEXT NOT NULL,
  gce_operation TEXT,
  created_at BIGINT NOT NULL,
  expires_at BIGINT NOT NULL,
  apply_started_at BIGINT,
  completed_at BIGINT,
  aborted_at BIGINT
);
CREATE INDEX IF NOT EXISTS relay_cell_fence_attempts_expiry
  ON relay_cell_fence_attempts(expires_at);
CREATE INDEX IF NOT EXISTS relay_cell_fence_attempts_cell
  ON relay_cell_fence_attempts(cell_id, created_at);

CREATE TABLE IF NOT EXISTS relay_cell_fence_plan_bindings (
  attempt_id TEXT PRIMARY KEY,
  plan_object_name TEXT NOT NULL,
  plan_object_generation TEXT,
  var_file_sha256 TEXT NOT NULL,
  terraform_state_lineage TEXT NOT NULL,
  terraform_state_serial BIGINT NOT NULL,
  terraform_state_object_generation TEXT NOT NULL,
  terraform_state_object_sha256 TEXT NOT NULL,
  request_reason TEXT NOT NULL,
  FOREIGN KEY (attempt_id) REFERENCES relay_cell_fence_attempts(attempt_id)
);

CREATE TABLE IF NOT EXISTS relay_cell_fence_apply_invocations (
  invocation_id TEXT PRIMARY KEY,
  attempt_id TEXT NOT NULL,
  request_reason TEXT NOT NULL UNIQUE,
  started_at BIGINT NOT NULL,
  gce_operation TEXT,
  FOREIGN KEY (attempt_id) REFERENCES relay_cell_fence_attempts(attempt_id)
);
CREATE INDEX IF NOT EXISTS relay_cell_fence_apply_invocations_attempt
  ON relay_cell_fence_apply_invocations(attempt_id, started_at);

CREATE TABLE IF NOT EXISTS relay_cell_drain_attempt_states (
  attempt_id TEXT PRIMARY KEY,
  cell_id TEXT NOT NULL,
  cell_incarnation TEXT NOT NULL,
  trace_value TEXT NOT NULL UNIQUE,
  planned_grace_ms BIGINT NOT NULL,
  state TEXT NOT NULL CHECK (
    state IN (
      'prepared',
      'send-may-have-started',
      'application-receipt',
      'proven-not-delivered'
    )
  ),
  prepared_at BIGINT NOT NULL,
  send_may_have_started_at BIGINT,
  send_permit_expires_at BIGINT,
  application_receipt_at BIGINT,
  backend_success_status BIGINT,
  backend_instance TEXT,
  receipt_cell_incarnation TEXT,
  retry_after BIGINT,
  recover_forward_attempted_at BIGINT,
  proven_not_delivered_at BIGINT
);
CREATE INDEX IF NOT EXISTS relay_cell_drain_attempt_states_cell
  ON relay_cell_drain_attempt_states(cell_id, prepared_at);

CREATE TABLE IF NOT EXISTS relay_cell_drain_recovery_attempts (
  drain_attempt_id TEXT NOT NULL,
  cell_incarnation TEXT NOT NULL,
  attempted_at BIGINT NOT NULL,
  PRIMARY KEY (drain_attempt_id, cell_incarnation),
  FOREIGN KEY (drain_attempt_id) REFERENCES relay_cell_drain_attempt_states(attempt_id)
);

CREATE TABLE IF NOT EXISTS relay_assignment_activity_leases (
  user_id TEXT NOT NULL,
  relay_host_id TEXT NOT NULL,
  activity_id TEXT NOT NULL,
  activity_kind TEXT NOT NULL,
  cell_id TEXT NOT NULL,
  request_units BIGINT NOT NULL,
  expires_at BIGINT NOT NULL,
  updated_at BIGINT NOT NULL,
  PRIMARY KEY (user_id, relay_host_id, activity_id)
);
-- expires_at is deliberately unindexed: every control renewal writes it (~471/s), so an index on
-- it makes each renewal a non-HOT update that rewrites index entries. Its only reader is the 30s
-- expiry sweep, which seq-scans 14.8k rows / 7MB in a few milliseconds.

CREATE TABLE IF NOT EXISTS relay_control_connection_reservations (
  reservation_id TEXT PRIMARY KEY,
  idempotency_key TEXT NOT NULL UNIQUE,
  user_id TEXT NOT NULL,
  relay_host_id TEXT NOT NULL,
  assignment_epoch BIGINT NOT NULL,
  cell_id TEXT NOT NULL,
  state TEXT NOT NULL
    CHECK (state IN ('reserved', 'late-arrival-debt', 'claimed', 'released')),
  inclusion_watermark BIGINT,
  claim_activity_id TEXT,
  created_at BIGINT NOT NULL,
  timeout_at BIGINT NOT NULL,
  claimed_at BIGINT,
  released_at BIGINT,
  updated_at BIGINT NOT NULL
);
CREATE INDEX IF NOT EXISTS relay_control_connection_reservation_headroom
  ON relay_control_connection_reservations(cell_id, state);
CREATE INDEX IF NOT EXISTS relay_control_connection_reservation_assignment
  ON relay_control_connection_reservations(
    user_id, relay_host_id, assignment_epoch, cell_id, created_at
  );

CREATE TABLE IF NOT EXISTS relay_rate_windows (
  scope_key TEXT NOT NULL,
  window_kind TEXT NOT NULL,
  window_started_at BIGINT NOT NULL,
  count BIGINT NOT NULL,
  PRIMARY KEY (scope_key, window_kind, window_started_at)
);

-- schema-deferrable: created out of band, so a boot that cannot take the lock must retry
-- Why: window_started_at is the PRIMARY KEY's last column, so the sweep's 24h retention delete
-- cannot use it and seq-scans instead.
CREATE INDEX IF NOT EXISTS relay_rate_windows_started
  ON relay_rate_windows(window_started_at);

CREATE TABLE IF NOT EXISTS relay_assignment_migrations (
  user_id TEXT NOT NULL,
  relay_host_id TEXT NOT NULL,
  source_cell_id TEXT NOT NULL,
  target_cell_id TEXT NOT NULL,
  previous_epoch BIGINT NOT NULL,
  assignment_epoch BIGINT NOT NULL,
  source_request_units BIGINT NOT NULL,
  target_reserved_units BIGINT NOT NULL,
  expires_at BIGINT NOT NULL,
  target_registered_at BIGINT,
  completed_at BIGINT,
  aborted_at BIGINT,
  created_at BIGINT NOT NULL,
  updated_at BIGINT NOT NULL,
  PRIMARY KEY (user_id, relay_host_id, assignment_epoch)
);
CREATE INDEX IF NOT EXISTS relay_assignment_migrations_active
  ON relay_assignment_migrations(expires_at, completed_at, aborted_at);

CREATE TABLE IF NOT EXISTS relay_assignment_migration_incarnations (
  user_id TEXT NOT NULL,
  relay_host_id TEXT NOT NULL,
  assignment_epoch BIGINT NOT NULL,
  source_cell_incarnation TEXT NOT NULL,
  target_cell_incarnation TEXT NOT NULL,
  PRIMARY KEY (user_id, relay_host_id, assignment_epoch)
);

CREATE TABLE IF NOT EXISTS relay_post_drain_migration_pins (
  user_id TEXT NOT NULL,
  relay_host_id TEXT NOT NULL,
  assignment_epoch BIGINT NOT NULL,
  drain_attempt_id TEXT NOT NULL,
  source_cell_id TEXT NOT NULL,
  source_cell_incarnation TEXT NOT NULL,
  target_cell_id TEXT NOT NULL,
  target_cell_incarnation TEXT NOT NULL,
  source_request_units BIGINT NOT NULL,
  target_reserved_units BIGINT NOT NULL,
  pinned_at BIGINT NOT NULL,
  PRIMARY KEY (user_id, relay_host_id, assignment_epoch)
);
CREATE INDEX IF NOT EXISTS relay_post_drain_migration_pins_attempt
  ON relay_post_drain_migration_pins(drain_attempt_id);

CREATE TABLE IF NOT EXISTS relay_audit_events (
  id TEXT PRIMARY KEY,
  at BIGINT NOT NULL,
  type TEXT NOT NULL,
  user_id TEXT,
  relay_host_id TEXT,
  relay_device_id TEXT,
  detail_json TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS relay_audit_events_at ON relay_audit_events(at);
`

// Rehoming is bidirectional, but tables created before that carry the
// original single-region column check. The old constraint is the one Postgres
// auto-named; the replacement is named, so both statements are no-ops on a
// database the current schema created and neither can drop the other.
export const POSTGRES_SCHEMA_MIGRATIONS = [
  POSTGRES_STATEMENT_STATS_MIGRATION,
  `ALTER TABLE relay_region_decisions ADD COLUMN IF NOT EXISTS last_considered_at BIGINT NOT NULL DEFAULT 0`,
  `ALTER TABLE relay_region_decisions ADD COLUMN IF NOT EXISTS cohort_bucket BIGINT NOT NULL DEFAULT 0`,
  `ALTER TABLE relay_region_rehome_attempts
     DROP CONSTRAINT IF EXISTS relay_region_rehome_attempts_preferred_region_check`,
  `ALTER TABLE relay_region_rehome_attempts
     ADD CONSTRAINT relay_region_rehome_attempts_preferred_region_valid
     CHECK (preferred_region IN (${REGION_LIST}))`,
  `ALTER TABLE relay_region_rehome_control
     ADD COLUMN IF NOT EXISTS host_cooldown_ms BIGINT NOT NULL
     DEFAULT ${REGIONAL_REHOME_DEFAULT_HOST_COOLDOWN_MS}`,
  `ALTER TABLE relay_control_capabilities ADD COLUMN IF NOT EXISTS idle_regional_rehome BIGINT NOT NULL DEFAULT 0`,
  `ALTER TABLE relay_region_rehome_attempts ADD COLUMN IF NOT EXISTS source_generation BIGINT NOT NULL DEFAULT 0`,
  // Nullable with no default, so the rewrite is catalog-only; every row
  // aborted before this column existed reads as an unattributed abort.
  `ALTER TABLE relay_region_rehome_attempts ADD COLUMN IF NOT EXISTS abort_reason TEXT`,
  // migration-only alone cannot say why a cell is parked there: five flows park loaded cells in
  // that state durably. This stamps only the same-cap roll's isolate step, so placement can tell a
  // cell being rolled from an evacuation target or an Asia rollback. Nullable with no default, so
  // the rewrite is catalog-only; a cell isolated before this column existed reads as unmarked and
  // keeps its hosts pinned, which is the pre-existing behaviour.
  `ALTER TABLE relay_cell_admission ADD COLUMN IF NOT EXISTS roll_isolated_at BIGINT`,
  // Dropped, not created: see the comment on relay_assignment_activity_leases. Deferrable because
  // this is the one boot where it has to take ACCESS EXCLUSIVE on a table under continuous write,
  // and all 28 directors reach it at once; a lock timeout here must not restart the instance, which
  // would only re-queue the same DDL behind the same writers. Once it wins, the pre-check answers
  // absent and no later boot sends it at all.
  `-- schema-deferrable: one boot has to win ACCESS EXCLUSIVE on a table written ~475/s
   DROP INDEX IF EXISTS relay_assignment_activity_expiry`,
  // The drop is what makes HOT legal; this is what makes it possible. A renewal can only reuse the
  // row's own page when that page has room for a second version, and at the default fillfactor of
  // 100 a freshly filled page has none - measured at 0.5% HOT with the index gone and the default,
  // against 100% at 70. Takes SHARE UPDATE EXCLUSIVE, which blocks vacuum and DDL but no reader or
  // writer, and only for the catalog write. Applies to pages as they refill, so the table converges
  // over its own renewal cycle rather than at boot.
  // Deferrable for the same reason, though SHARE UPDATE EXCLUSIVE blocks only vacuum and DDL: it
  // buys nothing until the drop lands, so a boot that deferred the drop should defer this too.
  `-- schema-deferrable: buys nothing until the drop above lands
   ALTER TABLE relay_assignment_activity_leases SET (fillfactor = 70)`
]

// The exact statement list a Postgres boot applies, in order, so the lock-target census can read
// what production runs rather than a copy of it. The SQLite path keeps SCHEMA on its own.
export function relayPostgresSchemaStatements(): string[] {
  return [...SCHEMA.split(';'), ...POSTGRES_SCHEMA_MIGRATIONS]
    .map((statement) => statement.trim())
    .filter((statement) => statement.length > 0)
}

function postgresSql(sql: string): string {
  let index = 0
  return sql.replace(/\?/g, () => `$${++index}`)
}

function returnsRows(sql: string): boolean {
  return /^\s*(select|with)/i.test(sql) || /returning/i.test(sql)
}

const POSTGRES_TRANSACTION_PHASES = [
  ['relay_region_rehome_', 'regional-rehome'],
  ['relay_assignment_activity_leases', 'activity-lease'],
  ['relay_assignment_migration', 'migration'],
  ['relay_post_drain_migration_pins', 'migration'],
  ['relay_cell_connection_runtime', 'cell-runtime'],
  ['relay_cell_connection_snapshots', 'cell-runtime'],
  ['relay_cell_runtime', 'cell-runtime'],
  ['relay_cell_drain_', 'cell-operation'],
  ['relay_cell_fence', 'cell-operation'],
  ['relay_cell_committed_fences', 'cell-operation'],
  ['relay_cell_legacy_fence_adoptions', 'cell-operation'],
  ['relay_admission_selector', 'admission'],
  ['relay_cell_admission', 'admission'],
  ['relay_control_connection_reservations', 'connection'],
  ['relay_connection_bases', 'connection'],
  ['relay_direct_authorizations', 'connection'],
  ['relay_confirm_results', 'connection'],
  ['relay_assignment_region_preferences', 'assignment'],
  ['relay_assignments', 'assignment'],
  ['relay_cell_', 'cell-inventory'],
  ['relay_cells', 'cell-inventory'],
  ['relay_invites', 'credential'],
  ['relay_devices', 'credential'],
  ['relay_install_results', 'credential'],
  ['relay_rate_windows', 'rate-limit'],
  ['relay_audit_events', 'audit']
] as const

const postgresTransactionPhaseByError = new WeakMap<object, string>()

function postgresTransactionPhase(sql: string): string {
  const normalized = sql.toLowerCase()
  return POSTGRES_TRANSACTION_PHASES.find(([table]) => normalized.includes(table))?.[1] ?? 'other'
}

function rememberPostgresTransactionPhase(error: unknown, sql: string): void {
  if (typeof error === 'object' && error !== null) {
    postgresTransactionPhaseByError.set(error, postgresTransactionPhase(sql))
  }
}

function postgresTransactionErrorPhase(error: unknown): string {
  return typeof error === 'object' && error !== null
    ? (postgresTransactionPhaseByError.get(error) ?? 'transaction')
    : 'transaction'
}

class SqliteTransaction implements RelayDatabase {
  readonly dialect = 'sqlite' as const
  private held: { fromMs: number; site: CellLockHoldSite } | undefined

  constructor(protected readonly database: DatabaseSync) {}

  consumeHold(): MeasuredHold | undefined {
    if (this.held === undefined) return undefined
    const hold = { holdMs: performance.now() - this.held.fromMs, site: this.held.site }
    this.held = undefined
    return hold
  }

  protected noteHeld(options: RelayLockOptions): void {
    if (options.measureHoldMs && this.held === undefined) {
      this.held = { fromMs: performance.now(), site: options.holdSite ?? 'inventory' }
    }
  }

  async query(sql: string, params: unknown[] = []): Promise<SqlRow[]> {
    const statement = this.database.prepare(sql)
    const bound = params.map((value) => (value === undefined ? null : value)) as never[]
    if (returnsRows(sql)) return statement.all(...bound) as SqlRow[]
    const result = statement.run(...bound)
    return [{ changes: Number(result.changes) }]
  }

  async queryLocked(
    sql: string,
    params: unknown[] = [],
    options: RelayLockOptions = {}
  ): Promise<SqlRow[]> {
    const rows = await this.query(sql, params)
    this.noteHeld(options)
    return rows
  }

  async transaction<T>(
    operation: (transaction: RelayDatabase) => Promise<T>,
    _options: RelayTransactionOptions = {}
  ): Promise<T> {
    return await operation(this)
  }

  async close(): Promise<void> {}
}

class SqliteDatabase extends SqliteTransaction {
  private tail: Promise<void> = Promise.resolve()
  private readonly holds = new CellInventoryHoldSamples()

  consumeHoldCounts(): CellInventoryHoldCounts {
    return this.holds.consumeCounts()
  }

  override async query(sql: string, params: unknown[] = []): Promise<SqlRow[]> {
    await this.tail
    return await super.query(sql, params)
  }

  override async transaction<T>(operation: (transaction: RelayDatabase) => Promise<T>): Promise<T> {
    const previous = this.tail
    let release!: () => void
    this.tail = new Promise((resolve) => (release = resolve))
    await previous
    try {
      this.database.exec('BEGIN IMMEDIATE')
      const transaction = new SqliteTransaction(this.database)
      try {
        const result = await operation(transaction)
        this.database.exec('COMMIT')
        recordMeasuredHold(this.holds, transaction)
        return result
      } catch (error) {
        this.database.exec('ROLLBACK')
        throw error
      }
    } finally {
      release()
    }
  }

  override async close(): Promise<void> {
    await this.tail
    this.database.close()
  }
}

// Literals for a simple-query message, which carries no bind parameters. Only safe
// integers and strings: anything else is a caller bug, not something to stringify.
function inlinePostgresParameters(sql: string, params: unknown[], client: pg.PoolClient): string {
  let index = 0
  const inlined = sql.replace(/\?/g, () => {
    const value = params[index++]
    if (typeof value === 'number' && Number.isSafeInteger(value)) return String(value)
    if (typeof value === 'string') return client.escapeLiteral(value)
    throw new Error('unsupported_inline_parameter')
  })
  if (index !== params.length) throw new Error('inline_parameter_count_mismatch')
  return inlined
}

class PostgresTransaction implements RelayDatabase {
  readonly dialect = 'postgres' as const
  private held: { fromMs: number; site: CellLockHoldSite } | undefined
  private lockUnavailable = 0
  private lockTimeouts = 0
  private state: 'open' | 'committed' | 'rolled-back' = 'open'

  constructor(protected readonly client: pg.PoolClient) {}

  get open(): boolean {
    return this.state === 'open'
  }

  async commitWithFinal(sql: string, params: unknown[] = []): Promise<boolean> {
    this.assertNotCommitted()
    // Zero rows divides by zero, so the message stops before COMMIT exactly when the write missed.
    const message =
      `WITH final_write AS (${inlinePostgresParameters(sql, params, this.client)}) ` +
      'SELECT 1 / (SELECT count(*)::int FROM final_write); COMMIT'
    try {
      await this.client.query(message)
    } catch (error) {
      // Simple query stops at the first error, so any server error means COMMIT never ran:
      // retryable codes take the caller's normal rollback-and-retry path. A lost connection
      // leaves the outcome unknown, and nothing retries it.
      if (String((error as { code?: unknown }).code) !== '22012') {
        rememberPostgresTransactionPhase(error, sql)
        throw error
      }
      await this.client.query('ROLLBACK')
      this.state = 'rolled-back'
      return false
    }
    this.state = 'committed'
    return true
  }

  private assertNotCommitted(): void {
    // A later statement would run in autocommit, outside the work it belongs to.
    if (this.state === 'committed') throw new Error('postgres_transaction_already_committed')
  }

  consumeHold(): MeasuredHold | undefined {
    if (this.held === undefined) return undefined
    const hold = { holdMs: performance.now() - this.held.fromMs, site: this.held.site }
    this.held = undefined
    return hold
  }

  // Drained by the owning database on both the commit and the rollback path: a
  // 55P03 rolls the transaction back, so counting only on success would drop it.
  consumeLockUnavailable(): number {
    const count = this.lockUnavailable
    this.lockUnavailable = 0
    return count
  }

  consumeLockTimeouts(): number {
    const count = this.lockTimeouts
    this.lockTimeouts = 0
    return count
  }

  async query(sql: string, params: unknown[] = []): Promise<SqlRow[]> {
    this.assertNotCommitted()
    try {
      const result = await this.client.query(postgresSql(sql), params)
      return returnsRows(sql) ? (result.rows as SqlRow[]) : [{ changes: result.rowCount ?? 0 }]
    } catch (error) {
      rememberPostgresTransactionPhase(error, sql)
      throw error
    }
  }

  async queryLocked(
    sql: string,
    params: unknown[] = [],
    options: RelayLockOptions = {}
  ): Promise<SqlRow[]> {
    // SET LOCAL lasts to COMMIT, so a bound left in place would silently govern
    // every later locked statement in the transaction and misattribute its 55P03s.
    const bounded = options.lockTimeoutMs !== undefined && !options.failIfUnavailable
    try {
      // A blocked waiter holds its pooled client for the whole lock_timeout, so
      // hot tiny-table locks bound their own wait well under the pool default.
      if (bounded) await this.query(setLocalLockTimeout(options.lockTimeoutMs!))
      const rows = await this.query(lockedStatement(sql, options), params)
      if (options.measureHoldMs && this.held === undefined) {
        this.held = { fromMs: performance.now(), site: options.holdSite ?? 'inventory' }
      }
      return rows
    } catch (error) {
      if (
        options.failIfUnavailable &&
        String((error as { code?: unknown }).code) === '55P03'
      ) {
        if (options.measureHoldMs) this.lockUnavailable += 1
        throw new Error('database_lock_unavailable')
      }
      // A bounded wait that expires raises the same 55P03 without NOWAIT. This is
      // the request path, so it is counted apart from by-design sweep deferrals.
      if (bounded && options.measureHoldMs && String((error as { code?: unknown }).code) === '55P03') {
        this.lockTimeouts += 1
      }
      throw error
    } finally {
      // Restore on the error path too: the transaction may still be retried or
      // continue with unrelated locks after a caught lock failure.
      if (bounded) await this.query(setLocalLockTimeout(POSTGRES_LOCK_TIMEOUT_MS)).catch(() => undefined)
    }
  }

  async transaction<T>(
    operation: (transaction: RelayDatabase) => Promise<T>,
    _options: RelayTransactionOptions = {}
  ): Promise<T> {
    return await operation(this)
  }

  async close(): Promise<void> {}
}

const POSTGRES_TRANSACTION_ATTEMPTS = 3
const POSTGRES_RETRY_MAX_DELAY_MS = 25
const POSTGRES_CONNECTION_TIMEOUT_MS = 2_000
// Derivation: a control renewal must land inside its own 30s tick
// (RELAY_PROTOCOL_LIMITS.controlPingIntervalMs * 2), and a transaction gets
// POSTGRES_TRANSACTION_ATTEMPTS tries, so the worst case a renewal can spend in
// Postgres is attempts * timeout. 5s keeps that at 15s, half the tick, and still
// leaves room for the connect timeout above.
export const POSTGRES_STATEMENT_TIMEOUT_MS = 5_000
const POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS = 5_000

export function relayPostgresStatementTimeoutMs(
  env: NodeJS.ProcessEnv = process.env
): number {
  const configured = env.ORCA_RELAY_POSTGRES_STATEMENT_TIMEOUT_MS
  if (configured === undefined || configured === '') return POSTGRES_STATEMENT_TIMEOUT_MS
  const milliseconds = Number(configured)
  // 0 is PostgreSQL's "no timeout"; refusing it keeps the deadline this exists
  // to enforce from being disabled by a typo in an environment variable.
  if (!Number.isInteger(milliseconds) || milliseconds < 1) {
    throw new Error('invalid_statement_timeout')
  }
  return milliseconds
}

function retryablePostgresTransactionError(error: unknown): boolean {
  const code = String((error as { code?: unknown }).code)
  // 57014 is the pool statement_timeout firing. It aborts the transaction the
  // same way a lock timeout does, so it belongs on the bounded retry path
  // rather than surfacing as a terminal failure to the caller.
  return code === '40P01' || code === '40001' || code === '55P03' || code === '57014'
}

export function isRelayDatabaseTransientError(error: unknown): boolean {
  // Runs inside the query catch, where a thrown null or undefined would turn a
  // database failure into a TypeError that buries it.
  const code = String((error as { code?: unknown } | null)?.code)
  if (['40P01', '40001', '55P03', '57014', '53300', '57P03', '08001', '08006'].includes(code)) {
    return true
  }
  // A pool that cannot hand out a client reports no SQLSTATE at all, so the
  // acquire boundary owns that vocabulary.
  return isPostgresPoolConnectFailure(error)
}

async function waitForPostgresRetry(random: () => number = Math.random): Promise<void> {
  const delayMs = Math.floor(random() * (POSTGRES_RETRY_MAX_DELAY_MS + 1))
  await new Promise((resolve) => setTimeout(resolve, delayMs))
}

export class PostgresDatabase implements RelayDatabase {
  readonly dialect = 'postgres' as const
  private readonly pressure: PostgresPoolPressure
  private readonly holds = new CellInventoryHoldSamples()

  consumeHoldCounts(): CellInventoryHoldCounts {
    return this.holds.consumeCounts()
  }

  constructor(private readonly pool: pg.Pool) {
    this.pressure = new PostgresPoolPressure(pool)
  }

  async query(sql: string, params: unknown[] = []): Promise<SqlRow[]> {
    const startedAt = performance.now()
    let phase: 'acquire' | 'execute' = 'acquire'
    let client: pg.PoolClient | undefined
    try {
      client = await this.pressure.connect()
      phase = 'execute'
      const result = await client.query(postgresSql(sql), params)
      return returnsRows(sql) ? (result.rows as SqlRow[]) : [{ changes: result.rowCount ?? 0 }]
    } catch (error) {
      reportPostgresQueryFailure({
        error,
        phase,
        sql,
        // Passed in rather than re-derived: the log has to say what the routes
        // actually did, and one classifier cannot drift from itself.
        transient: isRelayDatabaseTransientError(error),
        elapsedMs: performance.now() - startedAt,
        pool: this.pool
      })
      throw error
    } finally {
      client?.release()
    }
  }

  async queryLocked(
    sql: string,
    params: unknown[] = [],
    options: RelayLockOptions = {}
  ): Promise<SqlRow[]> {
    try {
      // No transaction here, so options.lockTimeoutMs cannot apply: SET LOCAL
      // would be discarded at the autocommit boundary before the lock is taken.
      return await this.query(lockedStatement(sql, options), params)
    } catch (error) {
      if (
        options.failIfUnavailable &&
        String((error as { code?: unknown }).code) === '55P03'
      ) {
        if (options.measureHoldMs) this.holds.recordUnavailable()
        throw new Error('database_lock_unavailable')
      }
      throw error
    }
  }

  async transaction<T>(
    operation: (transaction: RelayDatabase) => Promise<T>,
    options: RelayTransactionOptions = {}
  ): Promise<T> {
    for (let attempt = 1; attempt <= POSTGRES_TRANSACTION_ATTEMPTS; attempt++) {
      const client = await this.pressure.connect()
      const transaction = new PostgresTransaction(client)
      try {
        await client.query('BEGIN')
        const result = await operation(transaction)
        if (transaction.open) await client.query('COMMIT')
        recordMeasuredHold(this.holds, transaction)
        this.holds.recordUnavailable(transaction.consumeLockUnavailable())
        this.holds.recordLockTimeout(transaction.consumeLockTimeouts())
        return result
      } catch (error) {
        const open = transaction.open
        if (open) await client.query('ROLLBACK').catch(() => undefined)
        this.holds.recordUnavailable(transaction.consumeLockUnavailable())
        this.holds.recordLockTimeout(transaction.consumeLockTimeouts())
        // Once the fused commit has ended the transaction, a retry would apply the work twice.
        if (
          !open ||
          !retryablePostgresTransactionError(error) ||
          attempt === POSTGRES_TRANSACTION_ATTEMPTS
        ) {
          if (retryablePostgresTransactionError(error) && options.reportRetries !== false) {
            console.warn(
              JSON.stringify({
                event: 'orca_relay_postgres_transaction_exhausted',
                code: String((error as { code?: unknown }).code),
                attempts: attempt,
                phase: postgresTransactionErrorPhase(error)
              })
            )
          }
          throw error
        }
        if (options.reportRetries !== false) {
          console.warn(
            JSON.stringify({
              event: 'orca_relay_postgres_transaction_retry',
              code: String((error as { code?: unknown }).code),
              attempt,
              phase: postgresTransactionErrorPhase(error)
            })
          )
        }
      } finally {
        client.release()
      }
      // A PostgreSQL transaction is unusable after an abort, so retry all work
      // on a fresh pooled client with a small full-jitter delay.
      await waitForPostgresRetry()
    }
    throw new Error('postgres_transaction_retry_exhausted')
  }

  async close(): Promise<void> {
    await this.pool.end()
  }

  consumePoolPressure(): PostgresPoolPressureCounts {
    return this.pressure.consumeCounts()
  }

  peekPoolPressure(): PostgresPoolPressureCounts {
    return this.pressure.peekCounts()
  }
}

export function consumeRelayDatabasePoolPressure(
  database: RelayDatabase
): PostgresPoolPressureCounts {
  return database instanceof PostgresDatabase
    ? database.consumePoolPressure()
    : emptyPostgresPoolPressureCounts()
}

export function consumeRelayCellInventoryHold(
  database: RelayDatabase
): CellInventoryHoldCounts {
  const holder = database as { consumeHoldCounts?: () => CellInventoryHoldCounts }
  return holder.consumeHoldCounts?.() ?? emptyCellInventoryHoldCounts()
}

export function readRelayDatabasePoolPressure(
  database: RelayDatabase
): PostgresPoolPressureCounts {
  return database instanceof PostgresDatabase
    ? database.peekPoolPressure()
    : emptyPostgresPoolPressureCounts()
}

export function absorbPostgresIdleClientErrors(pool: Pick<pg.Pool, 'on'>): void {
  pool.on('error', () => {
    // Why: node-postgres removes failed idle clients itself; leaving `error`
    // unhandled would crash the cell and turn a SQL outage into autoheal churn.
    console.warn('[orca-relay] idle PostgreSQL client failed')
  })
}

async function applySchema(database: RelayDatabase): Promise<void> {
  for (const statement of SCHEMA.split(';')) {
    if (statement.trim()) await database.query(statement)
  }
  for (const [table, column] of [
    ['relay_control_capabilities', 'idle_regional_rehome'],
    ['relay_region_rehome_attempts', 'source_generation']
  ]) {
    const columns = await database.query('SELECT name FROM pragma_table_info(?)', [table])
    if (!columns.some((existing) => existing.name === column)) {
      await database.query(`ALTER TABLE ${table} ADD COLUMN ${column} BIGINT NOT NULL DEFAULT 0`)
    }
  }
}

// Why: DDL is not a request. A CREATE INDEX on a grown table legitimately runs
// longer than the request statement_timeout, and inheriting that timeout would
// make every startup fail at the same statement instead of finishing once. One
// short-lived connection of its own, ended before the serving pool opens, keeps
// the untimed session off the request path entirely.
async function applySchemaOnUntimedPool(
  databaseUrl: string,
  applicationName: string | undefined
): Promise<void> {
  const pool = new pg.Pool({
    connectionString: databaseUrl,
    max: 1,
    application_name: applicationName ? `${applicationName}/schema` : undefined,
    connectionTimeoutMillis: POSTGRES_CONNECTION_TIMEOUT_MS,
    statement_timeout: 0,
    // Kept: a DDL blocked behind another director's ACCESS EXCLUSIVE lock must
    // yield to the bounded schema retry instead of holding the connection.
    lock_timeout: POSTGRES_LOCK_TIMEOUT_MS,
    idle_in_transaction_session_timeout: POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS
  })
  absorbPostgresIdleClientErrors(pool)
  const database = new PostgresDatabase(pool)
  try {
    await applyPostgresSchema(
      relayPostgresSchemaStatements(),
      async (statement) => await database.query(statement),
      // Asks the catalog whether each index or column is already there. CREATE INDEX IF NOT EXISTS
      // and ALTER TABLE ADD COLUMN IF NOT EXISTS take their relation lock before the server
      // evaluates the existence test, so on an already-migrated database the boot still joins the
      // lock queue - and relation locks are granted in queue order, so every writer queues behind
      // it. The catalog read takes no lock on the table.
      { catalogQuery: async (sql, params) => await database.query(sql, params) }
    )
  } finally {
    await database.close().catch(() => undefined)
  }
}

async function backfillRelayCellRegions(database: RelayDatabase): Promise<void> {
  await database.query(
    `INSERT INTO relay_cell_regions (cell_id, region)
     SELECT cell_id, 'us-central1' FROM relay_cells WHERE true
     ON CONFLICT (cell_id) DO NOTHING`
  )
}

export type RelayDatabaseOpenInput = {
  databaseUrl?: string
  dataDir: string
  poolMax?: number
  applicationName?: string
  statementTimeoutMs?: number
  // Directors own the PostgreSQL schema. A cell skips it and never touches the database
  // at boot, so it starts listening while the database is down and stays unready until
  // its first successful query.
  appliesPostgresSchema?: boolean
}

export async function openRelayDatabase(input: RelayDatabaseOpenInput): Promise<RelayDatabase> {
  let database: RelayDatabase
  const appliesPostgresSchema = input.appliesPostgresSchema !== false
  if (input.databaseUrl) {
    if (appliesPostgresSchema) {
      await applySchemaOnUntimedPool(input.databaseUrl, input.applicationName)
    }
    const pool = new pg.Pool({
      connectionString: input.databaseUrl,
      max: input.poolMax ?? 10,
      application_name: input.applicationName,
      connectionTimeoutMillis: POSTGRES_CONNECTION_TIMEOUT_MS,
      statement_timeout: input.statementTimeoutMs ?? relayPostgresStatementTimeoutMs(),
      lock_timeout: POSTGRES_LOCK_TIMEOUT_MS,
      idle_in_transaction_session_timeout: POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS
    })
    absorbPostgresIdleClientErrors(pool)
    database = new PostgresDatabase(pool)
  } else {
    mkdirSync(input.dataDir, { recursive: true })
    const sqlite = new DatabaseSync(join(input.dataDir, 'orca-relay.sqlite'))
    sqlite.exec('PRAGMA journal_mode = WAL; PRAGMA foreign_keys = ON;')
    database = new SqliteDatabase(sqlite)
  }
  try {
    if (!input.databaseUrl) await applySchema(database)
    if (!input.databaseUrl || appliesPostgresSchema) await backfillRelayCellRegions(database)
    return database
  } catch (error) {
    await database.close().catch(() => undefined)
    throw error
  }
}

export async function openInMemoryRelayDatabase(): Promise<RelayDatabase> {
  const sqlite = new DatabaseSync(':memory:')
  sqlite.exec('PRAGMA foreign_keys = ON;')
  const database = new SqliteDatabase(sqlite)
  await applySchema(database)
  await backfillRelayCellRegions(database)
  return database
}