/**
 * Postgres server storage: the production database path.
 *
 * Semantics mirror `SqliteServerStorage` exactly (both run the shared
 * storage contract in `test/storage-contract.ts`); the difference is that
 * scope fanout MUST survive contact with Postgres — no scan-before-LIMIT,
 * ever (performance-by-construction):
 *
 *   - both the commit log and the current-row table carry a (table, var,
 *     value) inverted scope index;
 *   - `readCommitWindow`/`scanRows` select candidates through that index
 *     (`sync_change_scopes` / `sync_row_scopes`), ordered + LIMITed at the
 *     index, then verify the full multi-variable match against the stored
 *     scope map — never a log or table scan;
 *   - the candidate indexes are *covering* for their selection: the ordered
 *     column (`commit_seq` / `row_id`) sits in the index key after
 *     (tbl, var, value), so the planner does an index range scan and returns
 *     already-ordered candidates without a heap sort. `test/postgres-explain
 *     .test.ts` asserts an `Index` node (never `Seq Scan`) so the regression
 *     cannot silently return.
 *
 * Storage is written against the `PgExecutor` seam (zero runtime deps); the
 * production driver (Bun.sql / node-postgres) and the test driver (pglite)
 * are wired by the host. See `pg-executor.ts` and the server README.
 *
 * ## commitSeq allocation under concurrency
 *
 * Per-partition `commitSeq` is a dense, gap-free counter (§2.1), allocated
 * inside the push transaction by `UPDATE sync_partitions SET
 * max_commit_seq = max_commit_seq + 1 … RETURNING`. The `UPDATE` takes a
 * row-level write lock on the partition row for the duration of the
 * transaction, so two concurrent pushes to the same partition serialize on
 * that row: the second blocks until the first commits, then reads the
 * advanced counter. This keeps the sequence dense (a Postgres `SEQUENCE`
 * would leave gaps on rollback, which the pull-window arithmetic in §4.5
 * does not tolerate). Cross-partition pushes never contend.
 */
import type { PushOperationResult } from '@syncular/core';
import {
  bindAuthoritativePartition,
  postgresPlaceholders,
  prepareAuthoritativeQuery,
} from './authoritative-query';
import { syncError } from './errors';
import {
  asBytes,
  asNumber,
  type PgExecutor,
  type PgQueryable,
} from './pg-executor';
import {
  commitWindowPageSql,
  deleteRowSql,
  dropTableDdl,
  indexRowPageStatement,
  layoutsOf,
  migratePayload,
  parseLayouts,
  retiredTableNames,
  rewritePlan,
  rewriteRowSql,
  rewriteValues,
  SCHEMA_META_DDL_POSTGRES,
  type StoredColumnLayout,
  scanRowPageSql,
  schemaDdl,
  selectRowScopesSql,
  selectRowSql,
  selectRowsForRewriteSql,
  upsertSql,
  upsertValues,
} from './relational-rows';
import type { CompiledSchema, CompiledTable } from './schema';
import { matchesEffective } from './scopes';
import type {
  AuthoritativeQueryRequest,
  AuthoritativeQueryResult,
  ClientCursorInfo,
  ClientRecord,
  ClientSubscription,
  CommitMetadata,
  CommitMetadataQuery,
  CommitWindowQuery,
  DurableJsonValue,
  IndexRowScanQuery,
  NewCommit,
  NewReaction,
  PartitionRegistryEntry,
  PrunedReactionCounts,
  ReactionClaimQuery,
  ReactionFailure,
  ReactionFailureUpdate,
  ReactionListQuery,
  ReactionPruneQuery,
  RowScanQuery,
  ScopeActivityQuery,
  ScopeCommitActivity,
  ServerStorage,
  StorageTransaction,
  StoredChange,
  StoredCommit,
  StoredPushResult,
  StoredReaction,
  StoredRow,
} from './storage';
import {
  isPostgresConstraintError,
  StorageConstraintError,
} from './storage-errors';
import { assertScopeIndexedScan, resolveIndexRowScan } from './storage-query';

/**
 * Schema DDL. Applied by `PostgresServerStorage.migrate()` (idempotent).
 *
 * Covering index design (the reason this file exists):
 *   - `sync_change_scopes_pk (partition, tbl, var, value, commit_seq)` — the
 *     candidate scan for `readCommitWindow` ranges on the (partition, tbl,
 *     var, value IN …) prefix and returns `commit_seq` already ascending;
 *   - `sync_row_scopes_pk (partition, tbl, var, value, row_id)` — same shape
 *     for `scanRows`, ordered by `row_id`.
 * Both are the PRIMARY KEY, so they are the clustering/covering index for
 * their table. No secondary index is needed for the hot path.
 *
 * Blob reference index (§5.9.4) — parity with the SQLite dialect's
 * `sync_blob_refs`:
 *   - PRIMARY KEY `(partition, tbl, row_id, blob_id)` — the by-row prefix
 *     lets `setBlobRefs` replace a row's set with a single ranged DELETE and
 *     the delete path clear it by (partition, tbl, row_id);
 *   - the secondary `sync_blob_refs_by_blob (partition, blob_id)` index
 *     drives `listRowsReferencingBlob` (the download-authorization candidate
 *     set, §5.9.5) as an index range, never a scan. `postgres-explain
 *     .test.ts` asserts an `Index` node here too.
 */
export const POSTGRES_DDL = `
CREATE TABLE IF NOT EXISTS sync_partitions(
  partition TEXT PRIMARY KEY,
  max_commit_seq BIGINT NOT NULL DEFAULT 0,
  horizon_seq BIGINT NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS sync_partition_registry(
  partition TEXT PRIMARY KEY,
  log_epoch TEXT NOT NULL,
  epoch_required BOOLEAN NOT NULL DEFAULT FALSE,
  last_authenticated_at_ms BIGINT NOT NULL
);
CREATE TABLE IF NOT EXISTS sync_row_scopes(
  partition TEXT NOT NULL, tbl TEXT NOT NULL,
  var TEXT NOT NULL, value TEXT NOT NULL, row_id TEXT NOT NULL,
  PRIMARY KEY(partition, tbl, var, value, row_id)
);
CREATE TABLE IF NOT EXISTS sync_commits(
  partition TEXT NOT NULL, commit_seq BIGINT NOT NULL,
  client_id TEXT NOT NULL, client_commit_id TEXT NOT NULL,
  actor_id TEXT NOT NULL, created_at_ms BIGINT NOT NULL,
  PRIMARY KEY(partition, commit_seq)
);
CREATE INDEX IF NOT EXISTS sync_commits_by_time
  ON sync_commits(partition, created_at_ms);
CREATE TABLE IF NOT EXISTS sync_changes(
  partition TEXT NOT NULL, commit_seq BIGINT NOT NULL, idx INTEGER NOT NULL,
  tbl TEXT NOT NULL, row_id TEXT NOT NULL, op SMALLINT NOT NULL,
  row_version BIGINT, scopes JSONB NOT NULL, payload BYTEA,
  PRIMARY KEY(partition, commit_seq, idx)
);
CREATE INDEX IF NOT EXISTS sync_changes_by_table
  ON sync_changes(partition, commit_seq, tbl, idx);
CREATE TABLE IF NOT EXISTS sync_change_scopes(
  partition TEXT NOT NULL, tbl TEXT NOT NULL,
  var TEXT NOT NULL, value TEXT NOT NULL, commit_seq BIGINT NOT NULL,
  PRIMARY KEY(partition, tbl, var, value, commit_seq)
);
CREATE TABLE IF NOT EXISTS sync_push_results(
  partition TEXT NOT NULL, client_id TEXT NOT NULL,
  client_commit_id TEXT NOT NULL, result JSONB NOT NULL,
  PRIMARY KEY(partition, client_id, client_commit_id)
);
CREATE TABLE IF NOT EXISTS sync_reactions(
  partition TEXT NOT NULL, idempotency_key TEXT NOT NULL,
  type TEXT NOT NULL, version INTEGER NOT NULL, payload JSONB NOT NULL,
  source_client_id TEXT NOT NULL, source_client_commit_id TEXT NOT NULL,
  source_commit_seq BIGINT NOT NULL, created_at_ms BIGINT NOT NULL,
  available_at_ms BIGINT NOT NULL, status TEXT NOT NULL,
  attempts INTEGER NOT NULL, max_attempts INTEGER NOT NULL,
  lease_owner TEXT, lease_expires_at_ms BIGINT, completed_at_ms BIGINT,
  last_failure JSONB,
  PRIMARY KEY(partition, idempotency_key),
  CHECK(status IN ('pending', 'leased', 'completed', 'dead-letter'))
);
CREATE INDEX IF NOT EXISTS sync_reactions_due
  ON sync_reactions(partition, status, available_at_ms, created_at_ms, idempotency_key);
CREATE INDEX IF NOT EXISTS sync_reactions_lease
  ON sync_reactions(partition, status, lease_expires_at_ms);
CREATE INDEX IF NOT EXISTS sync_reactions_completed
  ON sync_reactions(partition, status, completed_at_ms, idempotency_key);
CREATE INDEX IF NOT EXISTS sync_reactions_dead_letter
  ON sync_reactions(partition, status, available_at_ms, idempotency_key);
CREATE TABLE IF NOT EXISTS sync_clients(
  partition TEXT NOT NULL, client_id TEXT NOT NULL, actor_id TEXT NOT NULL,
  wire_version INTEGER NOT NULL DEFAULT 1,
  cursor BIGINT NOT NULL, subscriptions JSONB NOT NULL,
  updated_at_ms BIGINT NOT NULL,
  PRIMARY KEY(partition, client_id)
);
ALTER TABLE sync_clients
  ADD COLUMN IF NOT EXISTS wire_version INTEGER NOT NULL DEFAULT 1;
CREATE TABLE IF NOT EXISTS sync_blob_refs(
  partition TEXT NOT NULL, tbl TEXT NOT NULL, row_id TEXT NOT NULL,
  blob_id TEXT NOT NULL,
  PRIMARY KEY(partition, tbl, row_id, blob_id)
);
CREATE INDEX IF NOT EXISTS sync_blob_refs_by_blob
  ON sync_blob_refs(partition, blob_id);
`;

interface SerializedResult {
  opIndex: number;
  status: string;
  code?: string;
  message?: string;
  serverVersion?: number;
  serverRow?: string;
  retryable?: boolean;
  details?: import('@syncular/core').RejectionDetails;
}

function toBase64(bytes: Uint8Array): string {
  return Buffer.from(bytes).toString('base64');
}

function fromBase64(text: string): Uint8Array {
  return new Uint8Array(Buffer.from(text, 'base64'));
}

interface PostgresReactionRecord {
  idempotency_key: string;
  type: string;
  version: unknown;
  payload: unknown;
  source_client_id: string;
  source_client_commit_id: string;
  source_commit_seq: unknown;
  created_at_ms: unknown;
  available_at_ms: unknown;
  status: StoredReaction['status'];
  attempts: unknown;
  max_attempts: unknown;
  lease_owner: string | null;
  lease_expires_at_ms: unknown | null;
  completed_at_ms: unknown | null;
  last_failure: unknown | null;
}

function toStoredReaction(record: PostgresReactionRecord): StoredReaction {
  return {
    idempotencyKey: record.idempotency_key,
    type: record.type,
    version: asNumber(record.version),
    payload: asJson(record.payload) as DurableJsonValue,
    sourceClientId: record.source_client_id,
    sourceClientCommitId: record.source_client_commit_id,
    sourceCommitSeq: asNumber(record.source_commit_seq),
    createdAtMs: asNumber(record.created_at_ms),
    maxAttempts: asNumber(record.max_attempts),
    status: record.status,
    attempts: asNumber(record.attempts),
    availableAtMs: asNumber(record.available_at_ms),
    ...(record.lease_owner !== null ? { leaseOwner: record.lease_owner } : {}),
    ...(record.lease_expires_at_ms !== null
      ? { leaseExpiresAtMs: asNumber(record.lease_expires_at_ms) }
      : {}),
    ...(record.completed_at_ms !== null
      ? { completedAtMs: asNumber(record.completed_at_ms) }
      : {}),
    ...(record.last_failure !== null
      ? { lastFailure: asJson(record.last_failure) as ReactionFailure }
      : {}),
  };
}

/**
 * Serialize a push result to a JSON-able object (stored in a JSONB column).
 * `serverRow` bytes are base64-encoded — JSONB cannot hold raw bytes.
 */
function serializePushResult(result: StoredPushResult): unknown {
  return {
    status: result.status,
    ...(result.commitSeq !== undefined ? { commitSeq: result.commitSeq } : {}),
    ...(result.recordedAtMs !== undefined
      ? { recordedAtMs: result.recordedAtMs }
      : {}),
    ...(result.cacheIdentity !== undefined
      ? { cacheIdentity: result.cacheIdentity }
      : {}),
    results: result.results.map((record) => {
      if (record.status === 'conflict') {
        return {
          opIndex: record.opIndex,
          status: record.status,
          code: record.code,
          message: record.message,
          serverVersion: record.serverVersion,
          serverRow: toBase64(record.serverRow),
        };
      }
      if (record.status === 'error') {
        return {
          opIndex: record.opIndex,
          status: record.status,
          code: record.code,
          message: record.message,
          retryable: record.retryable,
          ...(record.details !== undefined ? { details: record.details } : {}),
        };
      }
      return { opIndex: record.opIndex, status: record.status };
    }),
  };
}

function deserializePushResult(value: unknown): StoredPushResult {
  const parsed = value as {
    status: 'applied' | 'rejected';
    commitSeq?: number;
    recordedAtMs?: number;
    cacheIdentity?: string;
    results: SerializedResult[];
  };
  const results: PushOperationResult[] = parsed.results.map((record) => {
    if (record.status === 'conflict') {
      return {
        opIndex: record.opIndex,
        status: 'conflict',
        code: record.code ?? '',
        message: record.message ?? '',
        serverVersion: record.serverVersion ?? 0,
        serverRow: fromBase64(record.serverRow ?? ''),
      };
    }
    if (record.status === 'error') {
      return {
        opIndex: record.opIndex,
        status: 'error',
        code: record.code ?? '',
        message: record.message ?? '',
        retryable: record.retryable ?? false,
        ...(record.details !== undefined ? { details: record.details } : {}),
      };
    }
    return { opIndex: record.opIndex, status: 'applied' };
  });
  return {
    status: parsed.status,
    ...(parsed.commitSeq !== undefined ? { commitSeq: parsed.commitSeq } : {}),
    ...(parsed.recordedAtMs !== undefined
      ? { recordedAtMs: parsed.recordedAtMs }
      : {}),
    ...(parsed.cacheIdentity !== undefined
      ? { cacheIdentity: parsed.cacheIdentity }
      : {}),
    results,
  };
}

/**
 * Some drivers return a JSONB column already parsed (pglite, node-postgres);
 * accept a string too for defensiveness.
 */
function asJson<T>(value: unknown): T {
  if (typeof value === 'string') return JSON.parse(value) as T;
  return value as T;
}

interface RowRecord {
  row_id: string;
  server_version: unknown;
  scopes: unknown;
  payload: unknown;
}

interface ChangeRecord {
  tbl: string;
  row_id: string;
  op: unknown;
  row_version: unknown;
  scopes: unknown;
  payload: unknown;
}

/**
 * One result row of `commitWindowPageSql` (candidate LEFT JOIN commit meta
 * LEFT JOIN changes): meta/change columns are NULL when the joined row
 * vanished (LEFT JOIN contract).
 */
interface CommitWindowRecord {
  commit_seq: unknown;
  actor_id: string | null;
  created_at_ms: unknown;
  tbl: string | null;
  row_id: string | null;
  op: unknown;
  row_version: unknown;
  scopes: unknown;
  payload: unknown;
}

function toStoredRow(record: RowRecord): StoredRow {
  return {
    rowId: record.row_id,
    serverVersion: asNumber(record.server_version),
    scopes: asJson<Record<string, string>>(record.scopes),
    payload: asBytes(record.payload),
  };
}

function toStoredChange(record: ChangeRecord): StoredChange {
  return {
    table: record.tbl,
    rowId: record.row_id,
    op: asNumber(record.op) === 1 ? 'upsert' : 'delete',
    ...(record.row_version !== null && record.row_version !== undefined
      ? { rowVersion: asNumber(record.row_version) }
      : {}),
    scopes: asJson<Record<string, string>>(record.scopes),
    ...(record.payload !== null && record.payload !== undefined
      ? { payload: asBytes(record.payload) }
      : {}),
  };
}

/**
 * Migration rewrite: keyset-paged walk of a row table inside the migration
 * transaction. When `oldLayout` is
 * given every payload re-encodes under the current columns, and the
 * projection (when materialized) refreshes from the payload either way.
 */
async function rewriteRowsOn(
  q: PgQueryable,
  table: CompiledTable,
  oldLayout: readonly StoredColumnLayout[] | undefined,
): Promise<void> {
  const select = selectRowsForRewriteSql(table, 'postgres');
  const update = rewriteRowSql(table, 'postgres');
  const BATCH = 500;
  let afterPartition = '';
  let afterRowId = '';
  for (;;) {
    const { rows } = await q.query<{
      partition: string;
      row_id: string;
      payload: unknown;
    }>(select, [afterPartition, afterRowId, BATCH]);
    if (rows.length === 0) break;
    for (const row of rows) {
      const bytes = asBytes(row.payload);
      const payload =
        oldLayout !== undefined
          ? migratePayload(oldLayout, table, bytes)
          : bytes;
      await q.query(
        update,
        rewriteValues(table, row.partition, row.row_id, payload, 'postgres'),
      );
    }
    const last = rows[rows.length - 1];
    if (last === undefined || rows.length < BATCH) break;
    afterPartition = last.partition;
    afterRowId = last.row_id;
  }
}

/** Shared read/write query logic, parameterized by the queryable in scope. */
async function getRowOn(
  q: PgQueryable,
  compiled: CompiledTable,
  partition: string,
  rowId: string,
): Promise<StoredRow | undefined> {
  const { rows } = await q.query<RowRecord>(
    selectRowSql(compiled, 'postgres'),
    [partition, rowId],
  );
  const record = rows[0];
  return record === undefined ? undefined : toStoredRow(record);
}

async function scanRowsByIndexOn(
  q: PgQueryable,
  compiled: CompiledTable,
  partition: string,
  query: IndexRowScanQuery,
): Promise<StoredRow[]> {
  const index = resolveIndexRowScan(compiled, query);
  const statement = indexRowPageStatement(
    compiled,
    index,
    query.values,
    partition,
    query.afterRowId,
    query.limit,
    'postgres',
  );
  const { rows } = await q.query<RowRecord>(statement.sql, statement.params);
  return rows.map(toStoredRow);
}

async function writeRowOn(
  q: PgQueryable,
  compiled: CompiledTable,
  partition: string,
  row: StoredRow,
): Promise<void> {
  await q.query(
    upsertSql(compiled, 'postgres'),
    upsertValues(compiled, partition, row, 'postgres'),
  );
  await q.query(
    'DELETE FROM sync_row_scopes WHERE partition=$1 AND tbl=$2 AND row_id=$3',
    [partition, compiled.name, row.rowId],
  );
  for (const [variable, value] of Object.entries(row.scopes)) {
    await q.query(
      `INSERT INTO sync_row_scopes(partition, tbl, var, value, row_id)
       VALUES ($1,$2,$3,$4,$5) ON CONFLICT DO NOTHING`,
      [partition, compiled.name, variable, value, row.rowId],
    );
  }
}

async function getPushResultOn(
  q: PgQueryable,
  partition: string,
  clientId: string,
  clientCommitId: string,
): Promise<StoredPushResult | undefined> {
  const { rows } = await q.query<{ result: unknown }>(
    'SELECT result FROM sync_push_results WHERE partition=$1 AND client_id=$2 AND client_commit_id=$3',
    [partition, clientId, clientCommitId],
  );
  if (rows[0] === undefined) return undefined;
  try {
    return deserializePushResult(asJson(rows[0].result));
  } catch {
    throw syncError(
      'sync.idempotency_cache_miss',
      'persisted push result unreadable (§6.3)',
    );
  }
}

class PostgresTransaction implements StorageTransaction {
  #client: PgQueryable;
  #partition: string;
  #resolveTable: (name: string) => CompiledTable;
  #open = true;
  #pushApplySavepoint = false;
  /** Resolves/rejects the `transaction(fn)` wrapper (see `begin`). */
  #resolve: () => void;
  #reject: (error: unknown) => void;

  constructor(
    client: PgQueryable,
    partition: string,
    resolveTable: (name: string) => CompiledTable,
    resolve: () => void,
    reject: (error: unknown) => void,
  ) {
    this.#client = client;
    this.#partition = partition;
    this.#resolveTable = resolveTable;
    this.#resolve = resolve;
    this.#reject = reject;
  }

  #assertOpen(): void {
    if (!this.#open) throw new Error('transaction already finished');
  }

  getRow(table: string, rowId: string): Promise<StoredRow | undefined> {
    this.#assertOpen();
    return getRowOn(
      this.#client,
      this.#resolveTable(table),
      this.#partition,
      rowId,
    );
  }

  getPushResult(
    clientId: string,
    clientCommitId: string,
  ): Promise<StoredPushResult | undefined> {
    this.#assertOpen();
    // Runs on this transaction's pinned client: the push layer's duplicate
    // re-check happens while the partition lock is held, and a pool-level
    // read there would wait for a second connection.
    return getPushResultOn(
      this.#client,
      this.#partition,
      clientId,
      clientCommitId,
    );
  }

  async scanRows(query: RowScanQuery): Promise<StoredRow[]> {
    this.#assertOpen();
    const firstVariable = assertScopeIndexedScan(query);
    const firstValues = query.scopeFilter[firstVariable] ?? [];
    if (firstValues.length === 0) return [];
    const sql = scanRowPageSql(
      this.#resolveTable(query.table),
      firstValues.length,
      'postgres',
    );
    const rows: StoredRow[] = [];
    let afterRowId = query.afterRowId ?? '';
    const batchSize = Math.max(64, query.limit);
    while (rows.length < query.limit) {
      const { rows: records } = await this.#client.query<RowRecord>(sql, [
        this.#partition,
        query.table,
        firstVariable,
        ...firstValues,
        afterRowId,
        batchSize,
      ]);
      if (records.length === 0) break;
      for (const record of records) {
        afterRowId = record.row_id;
        if (record.payload === null || record.payload === undefined) continue;
        const stored = toStoredRow(record);
        if (!matchesEffective(stored.scopes, query.scopeFilter)) continue;
        rows.push(stored);
        if (rows.length >= query.limit) break;
      }
      if (records.length < batchSize) break;
    }
    return rows;
  }

  scanRowsByIndex(query: IndexRowScanQuery): Promise<StoredRow[]> {
    this.#assertOpen();
    return scanRowsByIndexOn(
      this.#client,
      this.#resolveTable(query.table),
      this.#partition,
      query,
    );
  }

  async lockPartitionForPush(): Promise<void> {
    this.#assertOpen();
    await this.#client.query(
      `INSERT INTO sync_partitions(partition, max_commit_seq) VALUES ($1, 0)
       ON CONFLICT (partition) DO NOTHING`,
      [this.#partition],
    );
    await this.#client.query(
      'SELECT max_commit_seq FROM sync_partitions WHERE partition=$1 FOR UPDATE',
      [this.#partition],
    );
    await this.#client.query('SAVEPOINT syncular_push_candidate');
    this.#pushApplySavepoint = true;
  }

  async commitRejectedPushResult(
    clientId: string,
    clientCommitId: string,
    result: StoredPushResult,
  ): Promise<void> {
    this.#assertOpen();
    if (!this.#pushApplySavepoint) {
      throw new Error('push rejection requires its apply savepoint');
    }
    await this.#client.query('ROLLBACK TO SAVEPOINT syncular_push_candidate');
    await this.#client.query('RELEASE SAVEPOINT syncular_push_candidate');
    this.#pushApplySavepoint = false;
    await this.putPushResult(clientId, clientCommitId, result);
    await this.commit();
  }

  async upsertRow(
    table: string,
    row: StoredRow,
    context?: { readonly opIndex: number },
  ): Promise<void> {
    this.#assertOpen();
    try {
      await writeRowOn(
        this.#client,
        this.#resolveTable(table),
        this.#partition,
        row,
      );
    } catch (error) {
      if (isPostgresConstraintError(error)) {
        throw new StorageConstraintError(error, context?.opIndex);
      }
      throw error;
    }
  }

  async deleteRow(table: string, rowId: string): Promise<void> {
    this.#assertOpen();
    await this.#client.query(
      deleteRowSql(this.#resolveTable(table), 'postgres'),
      [this.#partition, rowId],
    );
    await this.#client.query(
      'DELETE FROM sync_row_scopes WHERE partition=$1 AND tbl=$2 AND row_id=$3',
      [this.#partition, table, rowId],
    );
    // §5.9.4: a deleted row references no blobs.
    await this.#client.query(
      'DELETE FROM sync_blob_refs WHERE partition=$1 AND tbl=$2 AND row_id=$3',
      [this.#partition, table, rowId],
    );
  }

  async setBlobRefs(
    table: string,
    rowId: string,
    blobIds: readonly string[],
  ): Promise<void> {
    this.#assertOpen();
    // Replace the row's reference set atomically inside the commit tx (§5.9.4).
    await this.#client.query(
      'DELETE FROM sync_blob_refs WHERE partition=$1 AND tbl=$2 AND row_id=$3',
      [this.#partition, table, rowId],
    );
    for (const blobId of blobIds) {
      await this.#client.query(
        `INSERT INTO sync_blob_refs(partition, tbl, row_id, blob_id)
         VALUES ($1,$2,$3,$4) ON CONFLICT DO NOTHING`,
        [this.#partition, table, rowId, blobId],
      );
    }
  }

  async appendCommit(commit: NewCommit): Promise<number> {
    this.#assertOpen();
    const q = this.#client;
    const p = this.#partition;
    // Allocate the next dense commitSeq under a per-partition row lock: the
    // UPDATE … RETURNING serializes concurrent pushes to this partition and
    // never leaves a gap on rollback (see the file header).
    const { rows } = await q.query<{ max_commit_seq: unknown }>(
      `INSERT INTO sync_partitions(partition, max_commit_seq) VALUES ($1, 1)
       ON CONFLICT (partition) DO UPDATE
         SET max_commit_seq = sync_partitions.max_commit_seq + 1
       RETURNING max_commit_seq`,
      [p],
    );
    const commitSeq = asNumber(rows[0]?.max_commit_seq);
    await q.query(
      `INSERT INTO sync_commits(partition, commit_seq, client_id, client_commit_id, actor_id, created_at_ms)
       VALUES ($1,$2,$3,$4,$5,$6)`,
      [
        p,
        commitSeq,
        commit.clientId,
        commit.clientCommitId,
        commit.actorId,
        commit.createdAtMs,
      ],
    );
    for (let idx = 0; idx < commit.changes.length; idx++) {
      const change = commit.changes[idx];
      if (change === undefined) continue;
      await q.query(
        `INSERT INTO sync_changes(partition, commit_seq, idx, tbl, row_id, op, row_version, scopes, payload)
         VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)`,
        [
          p,
          commitSeq,
          idx,
          change.table,
          change.rowId,
          change.op === 'upsert' ? 1 : 2,
          change.rowVersion ?? null,
          JSON.stringify(change.scopes),
          change.payload ?? null,
        ],
      );
      for (const [variable, value] of Object.entries(change.scopes)) {
        await q.query(
          `INSERT INTO sync_change_scopes(partition, tbl, var, value, commit_seq)
           VALUES ($1,$2,$3,$4,$5) ON CONFLICT DO NOTHING`,
          [p, change.table, variable, value, commitSeq],
        );
      }
    }
    return commitSeq;
  }

  async putPushResult(
    clientId: string,
    clientCommitId: string,
    result: StoredPushResult,
  ): Promise<void> {
    this.#assertOpen();
    await this.#client.query(
      `INSERT INTO sync_push_results(partition, client_id, client_commit_id, result)
       VALUES ($1,$2,$3,$4)
       ON CONFLICT (partition, client_id, client_commit_id) DO NOTHING`,
      [
        this.#partition,
        clientId,
        clientCommitId,
        JSON.stringify(serializePushResult(result)),
      ],
    );
  }

  async enqueueReactions(reactions: readonly NewReaction[]): Promise<void> {
    this.#assertOpen();
    for (const reaction of reactions) {
      await this.#client.query(
        `INSERT INTO sync_reactions(
           partition, idempotency_key, type, version, payload,
           source_client_id, source_client_commit_id, source_commit_seq,
           created_at_ms, available_at_ms, status, attempts, max_attempts
         ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$9,'pending',0,$10)`,
        [
          this.#partition,
          reaction.idempotencyKey,
          reaction.type,
          reaction.version,
          JSON.stringify(reaction.payload),
          reaction.sourceClientId,
          reaction.sourceClientCommitId,
          reaction.sourceCommitSeq,
          reaction.createdAtMs,
          reaction.maxAttempts,
        ],
      );
    }
  }

  async commit(): Promise<void> {
    this.#assertOpen();
    this.#open = false;
    // Signal the transaction wrapper to COMMIT; wait for it to actually land.
    this.#resolve();
    await this.#done;
  }

  async rollback(): Promise<void> {
    if (!this.#open) return;
    this.#open = false;
    this.#reject(new RollbackSignal());
    // Swallow — a rollback is a normal outcome, not an error to the caller.
    await this.#done.catch(() => {});
  }

  /** Set by `begin`: resolves when BEGIN…COMMIT/ROLLBACK has fully landed. */
  #done!: Promise<void>;
  _attachDone(done: Promise<void>): void {
    this.#done = done;
  }
}

/** Internal marker: a caller-requested rollback, not a real failure. */
class RollbackSignal extends Error {
  constructor() {
    super('rollback');
    this.name = 'RollbackSignal';
  }
}

export class PostgresServerStorage implements ServerStorage {
  readonly #exec: PgExecutor;
  /** Set by `ensureSchema`: app-table lookup for the relational row store. */
  #tables: ReadonlyMap<string, CompiledTable> | undefined;
  #schemaVersion: number | undefined;

  constructor(exec: PgExecutor) {
    this.#exec = exec;
  }

  /** Apply the schema DDL (idempotent). Call once before use. */
  async migrate(): Promise<void> {
    // Split on the statement boundary so drivers that reject multi-statement
    // query strings (node-postgres extended protocol) still apply each DDL.
    const statements = POSTGRES_DDL.split(';')
      .map((s) => s.trim())
      .filter((s) => s.length > 0);
    for (const statement of statements) {
      await this.#exec.query(statement);
    }
  }

  /** Resolve a table's compiled schema; row operations require `ensureSchema`. */
  table(name: string): CompiledTable {
    const table = this.#tables?.get(name);
    if (table === undefined) {
      throw new Error(
        `unknown table ${JSON.stringify(name)} — ensureSchema(schema) must run before row operations`,
      );
    }
    return table;
  }

  async ensureSchema(schema: CompiledSchema): Promise<void> {
    // Memoized fast path: same instance, same schema version.
    if (this.#schemaVersion === schema.version) return;
    await this.migrate();
    await this.#exec.query(SCHEMA_META_DDL_POSTGRES);
    const marker = await this.#exec.query<{
      schema_version: unknown;
      layouts: unknown;
    }>('SELECT schema_version, layouts FROM sync_schema_meta WHERE id=1');
    const stored =
      marker.rows[0] === undefined
        ? undefined
        : asNumber(marker.rows[0].schema_version);
    if (stored !== undefined && stored > schema.version) {
      throw new Error(
        `stored schema version ${stored} is newer than the configured schema (${schema.version}) — refusing to run an older server against a migrated database`,
      );
    }
    if (stored === undefined || stored < schema.version) {
      // Introspect existing app tables, apply the migration subset
      // (CREATE TABLE / ADD COLUMN / rebuild indexes), then rewrite stored
      // rows (payload re-encode for layout changes and/or projection
      // backfill for flipped-on materialization) — all inside one
      // transaction (Postgres DDL is transactional — a failed bump leaves
      // no half-state).
      const layouts = parseLayouts(
        typeof marker.rows[0]?.layouts === 'string'
          ? marker.rows[0].layouts
          : undefined,
      );
      const retiredTables = retiredTableNames(schema, layouts);
      await this.#exec.transaction(async (client) => {
        const existing = new Map<string, ReadonlySet<string>>();
        const existingIndexes = new Map<string, ReadonlySet<string>>();
        for (const table of schema.tables.values()) {
          const { rows } = await client.query<{ column_name: string }>(
            `SELECT column_name FROM information_schema.columns
             WHERE table_schema = current_schema() AND table_name = $1`,
            [table.name],
          );
          if (rows.length > 0) {
            existing.set(table.name, new Set(rows.map((r) => r.column_name)));
            // Parity with the SQLite/D1 `origin === 'c'` filter: only
            // free-standing indexes enter the rebuild set. Constraint-owned
            // indexes (PRIMARY KEY, UNIQUE constraints) can only be removed
            // through their constraint, so DROP INDEX on them would abort
            // the migration transaction.
            const indexes = await client.query<{ index_name: string }>(
              `SELECT index_class.relname AS index_name
                 FROM pg_catalog.pg_class AS table_class
                 JOIN pg_catalog.pg_namespace AS namespace
                   ON namespace.oid = table_class.relnamespace
                 JOIN pg_catalog.pg_index AS index_meta
                   ON index_meta.indrelid = table_class.oid
                 JOIN pg_catalog.pg_class AS index_class
                   ON index_class.oid = index_meta.indexrelid
                WHERE namespace.nspname = current_schema()
                  AND table_class.relname = $1
                  AND NOT index_meta.indisprimary
                  AND NOT EXISTS (
                    SELECT 1
                      FROM pg_catalog.pg_constraint AS owning_constraint
                     WHERE owning_constraint.conindid = index_meta.indexrelid
                  )`,
              [table.name],
            );
            existingIndexes.set(
              table.name,
              new Set(indexes.rows.map((index) => index.index_name)),
            );
          }
        }
        for (const tableName of retiredTables) {
          await client.query('DELETE FROM sync_row_scopes WHERE tbl=$1', [
            tableName,
          ]);
          await client.query(dropTableDdl(tableName));
        }
        for (const statement of schemaDdl(
          schema,
          existing,
          'postgres',
          existingIndexes,
        )) {
          await client.query(statement);
        }
        for (const table of schema.tables.values()) {
          const oldLayout = layouts[table.name];
          const plan = rewritePlan(table, oldLayout, existing.get(table.name));
          if (!plan.migrate && !plan.backfill) continue;
          await rewriteRowsOn(
            client,
            table,
            plan.migrate ? oldLayout : undefined,
          );
        }
        await client.query(
          `INSERT INTO sync_schema_meta(id, schema_version, layouts) VALUES (1, $1, $2)
           ON CONFLICT (id) DO UPDATE
             SET schema_version=EXCLUDED.schema_version, layouts=EXCLUDED.layouts`,
          [schema.version, layoutsOf(schema)],
        );
      });
    }
    this.#tables = schema.tables;
    this.#schemaVersion = schema.version;
  }

  async touchPartition(
    partition: string,
    authenticatedAtMs: number,
    initialLogEpoch: string,
  ): Promise<PartitionRegistryEntry> {
    if (initialLogEpoch.length === 0) {
      throw new Error('initial log epoch must be non-empty');
    }
    const { rows } = await this.#exec.query<{
      log_epoch: string;
      epoch_required: boolean;
      last_authenticated_at_ms: unknown;
    }>(
      `INSERT INTO sync_partition_registry(
         partition, log_epoch, last_authenticated_at_ms
       ) VALUES ($1,$2,$3)
       ON CONFLICT(partition) DO UPDATE SET
         last_authenticated_at_ms=EXCLUDED.last_authenticated_at_ms
       RETURNING log_epoch, epoch_required, last_authenticated_at_ms`,
      [partition, initialLogEpoch, authenticatedAtMs],
    );
    const row = rows[0];
    if (row === undefined)
      throw new Error('partition registry write did not persist');
    return {
      partition,
      logEpoch: row.log_epoch,
      epochRequired: row.epoch_required,
      lastAuthenticatedAtMs: asNumber(row.last_authenticated_at_ms),
    };
  }

  async rotatePartitionLogEpoch(
    partition: string,
    logEpoch: string,
    authenticatedAtMs: number,
  ): Promise<PartitionRegistryEntry> {
    if (logEpoch.length === 0) throw new Error('log epoch must be non-empty');
    await this.#exec.transaction(async (client) => {
      await client.query(
        `INSERT INTO sync_partition_registry(
           partition, log_epoch, epoch_required, last_authenticated_at_ms
         ) VALUES ($1,$2,TRUE,$3)
         ON CONFLICT(partition) DO UPDATE SET
           log_epoch=EXCLUDED.log_epoch,
           epoch_required=TRUE,
           last_authenticated_at_ms=EXCLUDED.last_authenticated_at_ms`,
        [partition, logEpoch, authenticatedAtMs],
      );
      await client.query('DELETE FROM sync_clients WHERE partition=$1', [
        partition,
      ]);
    });
    return {
      partition,
      logEpoch,
      epochRequired: true,
      lastAuthenticatedAtMs: authenticatedAtMs,
    };
  }

  async listPartitionRegistry(): Promise<PartitionRegistryEntry[]> {
    const { rows } = await this.#exec.query<{
      partition: string;
      log_epoch: string;
      epoch_required: boolean;
      last_authenticated_at_ms: unknown;
    }>(
      `SELECT partition, log_epoch, epoch_required, last_authenticated_at_ms
         FROM sync_partition_registry ORDER BY partition`,
      [],
    );
    return rows.map((row) => ({
      partition: row.partition,
      logEpoch: row.log_epoch,
      epochRequired: row.epoch_required,
      lastAuthenticatedAtMs: asNumber(row.last_authenticated_at_ms),
    }));
  }

  /**
   * Open a real Postgres transaction. The push handler drives the returned
   * `StorageTransaction` imperatively (getRow/upsert/…/commit), but the
   * driver seam models a transaction as a callback scope. We bridge the two:
   * `transaction(fn)` blocks inside `fn` on a promise that the imperative
   * `commit()`/`rollback()` resolves/rejects, so BEGIN…COMMIT wraps exactly
   * the handler's writes on one pinned connection.
   */
  async begin(partition: string): Promise<StorageTransaction> {
    let resolveReady!: (tx: PostgresTransaction) => void;
    const ready = new Promise<PostgresTransaction>((r) => {
      resolveReady = r;
    });
    let resolveScope!: () => void;
    let rejectScope!: (error: unknown) => void;
    const scope = new Promise<void>((res, rej) => {
      resolveScope = res;
      rejectScope = rej;
    });
    const done = this.#exec
      .transaction(async (client) => {
        const tx = new PostgresTransaction(
          client,
          partition,
          (name) => this.table(name),
          resolveScope,
          rejectScope,
        );
        resolveReady(tx);
        // Hold the transaction open until commit()/rollback() settles `scope`.
        await scope;
      })
      .catch((error: unknown) => {
        if (error instanceof RollbackSignal) return;
        throw error;
      });
    const tx = await ready;
    tx._attachDone(done);
    return tx;
  }

  async getMaxCommitSeq(partition: string): Promise<number> {
    const { rows } = await this.#exec.query<{ max_commit_seq: unknown }>(
      'SELECT max_commit_seq FROM sync_partitions WHERE partition=$1',
      [partition],
    );
    return rows[0] === undefined ? 0 : asNumber(rows[0].max_commit_seq);
  }

  async queryAuthoritative(
    partition: string,
    query: AuthoritativeQueryRequest,
  ): Promise<AuthoritativeQueryResult> {
    if (this.#tables === undefined) {
      throw new Error(
        'ensureSchema(schema) must run before registered queries',
      );
    }
    const prepared = bindAuthoritativePartition(
      prepareAuthoritativeQuery(
        query.sql,
        query.params,
        query.tables,
        this.#tables,
      ),
      partition,
    );
    return this.#exec.transaction(async (client) => {
      await client.query(
        'SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY',
      );
      const result = await client.query<Readonly<Record<string, unknown>>>(
        postgresPlaceholders(prepared.sql),
        prepared.params,
      );
      const cursor = await client.query<{ max_commit_seq: unknown }>(
        'SELECT max_commit_seq FROM sync_partitions WHERE partition=$1',
        [partition],
      );
      return {
        rows: result.rows,
        maxCommitSeq:
          cursor.rows[0] === undefined
            ? 0
            : asNumber(cursor.rows[0].max_commit_seq),
      };
    });
  }

  async getHorizonSeq(partition: string): Promise<number> {
    const { rows } = await this.#exec.query<{ horizon_seq: unknown }>(
      'SELECT horizon_seq FROM sync_partitions WHERE partition=$1',
      [partition],
    );
    return rows[0] === undefined ? 0 : asNumber(rows[0].horizon_seq);
  }

  async setHorizonSeq(partition: string, seq: number): Promise<void> {
    await this.#exec.query(
      `INSERT INTO sync_partitions(partition, horizon_seq) VALUES ($1,$2)
       ON CONFLICT (partition) DO UPDATE SET horizon_seq=EXCLUDED.horizon_seq`,
      [partition, seq],
    );
  }

  async pruneCommitsThrough(partition: string, seq: number): Promise<number> {
    const removed = await this.#exec.query(
      'DELETE FROM sync_commits WHERE partition=$1 AND commit_seq<=$2',
      [partition, seq],
    );
    await this.#exec.query(
      'DELETE FROM sync_changes WHERE partition=$1 AND commit_seq<=$2',
      [partition, seq],
    );
    await this.#exec.query(
      'DELETE FROM sync_change_scopes WHERE partition=$1 AND commit_seq<=$2',
      [partition, seq],
    );
    return removed.rowCount;
  }

  async getCommitSeqBefore(
    partition: string,
    createdBeforeMs: number,
  ): Promise<number> {
    const { rows } = await this.#exec.query<{ seq: unknown }>(
      'SELECT max(commit_seq) AS seq FROM sync_commits WHERE partition=$1 AND created_at_ms<$2',
      [partition, createdBeforeMs],
    );
    const seq = rows[0]?.seq;
    return seq === null || seq === undefined ? 0 : asNumber(seq);
  }

  getRow(
    partition: string,
    table: string,
    rowId: string,
  ): Promise<StoredRow | undefined> {
    return getRowOn(this.#exec, this.table(table), partition, rowId);
  }

  getPushResult(
    partition: string,
    clientId: string,
    clientCommitId: string,
  ): Promise<StoredPushResult | undefined> {
    return getPushResultOn(this.#exec, partition, clientId, clientCommitId);
  }

  async claimReactions(
    partition: string,
    query: ReactionClaimQuery,
  ): Promise<StoredReaction[]> {
    if (query.types.length === 0 || query.limit <= 0) return [];
    const typeParams = query.types.map((_, index) => `$${index + 2}`).join(',');
    const nowParam = query.types.length + 2;
    const limitParam = nowParam + 1;
    const workerParam = limitParam + 1;
    const expiresParam = workerParam + 1;
    const { rows } = await this.#exec.query<PostgresReactionRecord>(
      `WITH due AS (
         SELECT partition, idempotency_key
           FROM sync_reactions
          WHERE partition=$1 AND type IN (${typeParams})
            AND ((status='pending' AND available_at_ms<=$${nowParam})
              OR (status='leased' AND lease_expires_at_ms<=$${nowParam}))
          ORDER BY CASE WHEN status='leased' THEN lease_expires_at_ms
                        ELSE available_at_ms END,
                   created_at_ms, idempotency_key
          FOR UPDATE SKIP LOCKED
          LIMIT $${limitParam}
       )
       UPDATE sync_reactions AS reaction
          SET status='leased', attempts=reaction.attempts+1,
              lease_owner=$${workerParam}, lease_expires_at_ms=$${expiresParam},
              completed_at_ms=NULL
         FROM due
        WHERE reaction.partition=due.partition
          AND reaction.idempotency_key=due.idempotency_key
      RETURNING reaction.*`,
      [
        partition,
        ...query.types,
        query.nowMs,
        query.limit,
        query.leaseOwner,
        Math.min(Number.MAX_SAFE_INTEGER, query.nowMs + query.leaseDurationMs),
      ],
    );
    return rows
      .map(toStoredReaction)
      .sort(
        (a, b) =>
          a.createdAtMs - b.createdAtMs ||
          a.idempotencyKey.localeCompare(b.idempotencyKey),
      );
  }

  async completeReaction(
    partition: string,
    idempotencyKey: string,
    leaseOwner: string,
    completedAtMs: number,
  ): Promise<boolean> {
    const result = await this.#exec.query(
      `UPDATE sync_reactions
          SET status='completed', completed_at_ms=$4,
              lease_owner=NULL, lease_expires_at_ms=NULL
        WHERE partition=$1 AND idempotency_key=$2
          AND status='leased' AND lease_owner=$3`,
      [partition, idempotencyKey, leaseOwner, completedAtMs],
    );
    return result.rowCount === 1;
  }

  async extendReactionLease(
    partition: string,
    idempotencyKey: string,
    leaseOwner: string,
    leaseExpiresAtMs: number,
  ): Promise<boolean> {
    const result = await this.#exec.query(
      `UPDATE sync_reactions SET lease_expires_at_ms=$4
        WHERE partition=$1 AND idempotency_key=$2
          AND status='leased' AND lease_owner=$3`,
      [partition, idempotencyKey, leaseOwner, leaseExpiresAtMs],
    );
    return result.rowCount === 1;
  }

  async failReaction(
    partition: string,
    idempotencyKey: string,
    update: ReactionFailureUpdate,
  ): Promise<boolean> {
    const result = await this.#exec.query(
      `UPDATE sync_reactions
          SET status=$4, available_at_ms=$5, last_failure=$6,
              lease_owner=NULL, lease_expires_at_ms=NULL
        WHERE partition=$1 AND idempotency_key=$2
          AND status='leased' AND lease_owner=$3`,
      [
        partition,
        idempotencyKey,
        update.leaseOwner,
        update.retryAtMs === undefined ? 'dead-letter' : 'pending',
        update.retryAtMs ?? update.failure.atMs,
        JSON.stringify(update.failure),
      ],
    );
    return result.rowCount === 1;
  }

  async retryReaction(
    partition: string,
    idempotencyKey: string,
    nowMs: number,
  ): Promise<boolean> {
    const result = await this.#exec.query(
      `UPDATE sync_reactions
          SET status='pending', attempts=0, available_at_ms=$3,
              last_failure=NULL, lease_owner=NULL, lease_expires_at_ms=NULL,
              completed_at_ms=NULL
        WHERE partition=$1 AND idempotency_key=$2 AND status='dead-letter'`,
      [partition, idempotencyKey, nowMs],
    );
    return result.rowCount === 1;
  }

  async getReaction(
    partition: string,
    idempotencyKey: string,
  ): Promise<StoredReaction | undefined> {
    const { rows } = await this.#exec.query<PostgresReactionRecord>(
      'SELECT * FROM sync_reactions WHERE partition=$1 AND idempotency_key=$2',
      [partition, idempotencyKey],
    );
    return rows[0] === undefined ? undefined : toStoredReaction(rows[0]);
  }

  async listReactions(
    partition: string,
    query: ReactionListQuery,
  ): Promise<StoredReaction[]> {
    const where = ['partition=$1'];
    const params: unknown[] = [partition];
    if (query.statuses !== undefined && query.statuses.length > 0) {
      const placeholders = query.statuses.map(
        (_, index) => `$${params.length + index + 1}`,
      );
      where.push(`status IN (${placeholders.join(',')})`);
      params.push(...query.statuses);
    }
    if (query.types !== undefined && query.types.length > 0) {
      const placeholders = query.types.map(
        (_, index) => `$${params.length + index + 1}`,
      );
      where.push(`type IN (${placeholders.join(',')})`);
      params.push(...query.types);
    }
    params.push(query.limit);
    const { rows } = await this.#exec.query<PostgresReactionRecord>(
      `SELECT * FROM sync_reactions WHERE ${where.join(' AND ')}
        ORDER BY created_at_ms DESC, idempotency_key DESC LIMIT $${params.length}`,
      params,
    );
    return rows.map(toStoredReaction);
  }

  async pruneReactions(
    partition: string,
    query: ReactionPruneQuery,
  ): Promise<PrunedReactionCounts> {
    if (query.limit <= 0) return { completed: 0, deadLetter: 0 };
    const { rows } = await this.#exec.query<{
      status: 'completed' | 'dead-letter';
    }>(
      `WITH targets AS (
         SELECT partition, idempotency_key
           FROM sync_reactions
          WHERE partition=$1
            AND ((status='completed' AND completed_at_ms IS NOT NULL
                  AND completed_at_ms<$2)
              OR (status='dead-letter' AND available_at_ms<$3))
          ORDER BY CASE WHEN status='completed' THEN completed_at_ms
                        ELSE available_at_ms END,
                   idempotency_key
          FOR UPDATE SKIP LOCKED
          LIMIT $4
       )
       DELETE FROM sync_reactions AS reaction
        USING targets
        WHERE reaction.partition=targets.partition
          AND reaction.idempotency_key=targets.idempotency_key
          AND ((reaction.status='completed'
                AND reaction.completed_at_ms IS NOT NULL
                AND reaction.completed_at_ms<$2)
            OR (reaction.status='dead-letter'
                AND reaction.available_at_ms<$3))
      RETURNING reaction.status`,
      [
        partition,
        query.completedBeforeMs,
        query.deadLetterBeforeMs,
        query.limit,
      ],
    );
    return {
      completed: rows.filter((row) => row.status === 'completed').length,
      deadLetter: rows.filter((row) => row.status === 'dead-letter').length,
    };
  }

  async readCommitWindow(
    partition: string,
    query: CommitWindowQuery,
  ): Promise<StoredCommit[]> {
    const variables = Object.keys(query.scopeFilter).sort();
    const firstVariable = variables[0];
    if (firstVariable === undefined) return [];
    const firstValues = query.scopeFilter[firstVariable] ?? [];
    if (firstValues.length === 0) return [];
    // Candidates via the inverted index (one variable): the (partition, tbl,
    // var, value, commit_seq) PK makes the candidate subquery an index range
    // scan returning commit_seq already ascending (see postgres-explain
    // .test.ts), LEFT JOINed to the commit meta + the table's changes — one
    // round trip per page, never two per candidate (see
    // `commitWindowPageSql`). Exact multi-variable verification against the
    // stored scope map happens below.
    const sql = commitWindowPageSql(firstValues.length, 'postgres');
    const commits: StoredCommit[] = [];
    let deliveredChanges = 0;
    let afterSeq = query.afterSeq;
    const batchSize = Math.max(64, query.limitChanges);
    while (deliveredChanges < query.limitChanges) {
      const { rows: records } = await this.#exec.query<CommitWindowRecord>(
        sql,
        [
          partition,
          query.table,
          firstVariable,
          ...firstValues,
          afterSeq,
          query.throughSeq,
          batchSize,
        ],
      );
      if (records.length === 0) break;
      // Fold the page: rows arrive ordered (commit_seq, idx); consecutive
      // rows with the same commit_seq are one candidate. Every candidate
      // advances the cursor — including vanished commits (NULL meta, LEFT
      // JOIN contract) and commits whose changes all fail the exact
      // multi-variable match.
      let candidateCount = 0;
      let i = 0;
      while (i < records.length) {
        const head = records[i];
        if (head === undefined) break;
        const commitSeq = asNumber(head.commit_seq);
        candidateCount += 1;
        afterSeq = commitSeq;
        const changes: StoredChange[] = [];
        for (; i < records.length; i++) {
          const record = records[i];
          if (record === undefined || asNumber(record.commit_seq) !== commitSeq)
            break;
          // NULL tbl: a candidate with no change rows for the table.
          if (record.tbl === null || record.row_id === null) continue;
          const change = toStoredChange({
            tbl: record.tbl,
            row_id: record.row_id,
            op: record.op,
            row_version: record.row_version,
            scopes: record.scopes,
            payload: record.payload,
          });
          if (matchesEffective(change.scopes, query.scopeFilter)) {
            changes.push(change);
          }
        }
        // NULL meta: the commit vanished — cursor advanced, nothing emitted.
        if (head.actor_id === null) continue;
        if (changes.length === 0) continue;
        commits.push({
          commitSeq,
          createdAtMs: asNumber(head.created_at_ms),
          actorId: head.actor_id,
          changes,
        });
        deliveredChanges += changes.length;
        if (deliveredChanges >= query.limitChanges) break;
      }
      if (deliveredChanges >= query.limitChanges) break;
      if (candidateCount < batchSize) break;
    }
    return commits;
  }

  async scanRows(partition: string, query: RowScanQuery): Promise<StoredRow[]> {
    const firstVariable = assertScopeIndexedScan(query);
    const firstValues = query.scopeFilter[firstVariable] ?? [];
    if (firstValues.length === 0) return [];
    // Candidates via the inverted index (ordered + LIMITed at the covering
    // PK) LEFT JOINed to the row table — one round trip per page, never one
    // per row (see `scanRowPageSql`). Exact multi-variable verification
    // against the stored scope map happens below.
    const sql = scanRowPageSql(
      this.table(query.table),
      firstValues.length,
      'postgres',
    );
    const rows: StoredRow[] = [];
    let afterRowId = query.afterRowId ?? '';
    const batchSize = Math.max(64, query.limit);
    while (rows.length < query.limit) {
      const { rows: records } = await this.#exec.query<RowRecord>(sql, [
        partition,
        query.table,
        firstVariable,
        ...firstValues,
        afterRowId,
        batchSize,
      ]);
      if (records.length === 0) break;
      for (const record of records) {
        afterRowId = record.row_id;
        // NULL payload: an index candidate whose row vanished — it still
        // advances the keyset cursor (LEFT JOIN contract) but yields no row.
        if (record.payload === null || record.payload === undefined) continue;
        const stored = toStoredRow(record);
        if (!matchesEffective(stored.scopes, query.scopeFilter)) continue;
        rows.push(stored);
        if (rows.length >= query.limit) break;
      }
      if (records.length < batchSize) break;
    }
    return rows;
  }

  scanRowsByIndex(
    partition: string,
    query: IndexRowScanQuery,
  ): Promise<StoredRow[]> {
    return scanRowsByIndexOn(
      this.#exec,
      this.table(query.table),
      partition,
      query,
    );
  }

  async getClientRecord(
    partition: string,
    clientId: string,
  ): Promise<ClientRecord | undefined> {
    const { rows } = await this.#exec.query<{
      client_id: string;
      actor_id: string;
      wire_version: unknown;
      cursor: unknown;
      subscriptions: unknown;
      updated_at_ms: unknown;
    }>(
      'SELECT client_id, actor_id, wire_version, cursor, subscriptions, updated_at_ms FROM sync_clients WHERE partition=$1 AND client_id=$2',
      [partition, clientId],
    );
    const record = rows[0];
    if (record === undefined) return undefined;
    return {
      clientId: record.client_id,
      actorId: record.actor_id,
      wireVersion: asNumber(record.wire_version),
      cursor: asNumber(record.cursor),
      updatedAtMs: asNumber(record.updated_at_ms),
      subscriptions: asJson<ClientSubscription[]>(record.subscriptions),
    };
  }

  async putClientRecord(
    partition: string,
    record: ClientRecord,
  ): Promise<void> {
    await this.#exec.query(
      `INSERT INTO sync_clients(partition, client_id, actor_id, wire_version, cursor, subscriptions, updated_at_ms)
       VALUES ($1,$2,$3,$4,$5,$6,$7)
       ON CONFLICT (partition, client_id) DO UPDATE
         SET actor_id=EXCLUDED.actor_id, wire_version=EXCLUDED.wire_version,
             cursor=EXCLUDED.cursor,
             subscriptions=EXCLUDED.subscriptions,
             updated_at_ms=EXCLUDED.updated_at_ms`,
      [
        partition,
        record.clientId,
        record.actorId,
        record.wireVersion,
        record.cursor,
        JSON.stringify(record.subscriptions),
        record.updatedAtMs,
      ],
    );
  }

  async listClientCursors(partition: string): Promise<ClientCursorInfo[]> {
    const { rows } = await this.#exec.query<{
      client_id: string;
      cursor: unknown;
      updated_at_ms: unknown;
    }>(
      'SELECT client_id, cursor, updated_at_ms FROM sync_clients WHERE partition=$1',
      [partition],
    );
    return rows.map((r) => ({
      clientId: r.client_id,
      cursor: asNumber(r.cursor),
      updatedAtMs: asNumber(r.updated_at_ms),
    }));
  }

  async listRowsReferencingBlob(
    partition: string,
    blobId: string,
  ): Promise<
    {
      readonly table: string;
      readonly rowId: string;
      readonly scopes: Record<string, string>;
    }[]
  > {
    // Candidate rows via the by-blob index (§5.9.4/§5.9.5); each row's
    // stored scopes come from sync_rows for the §3.4 authorization test.
    const { rows: refs } = await this.#exec.query<{
      tbl: string;
      row_id: string;
    }>(
      'SELECT tbl, row_id FROM sync_blob_refs WHERE partition=$1 AND blob_id=$2',
      [partition, blobId],
    );
    const out: {
      table: string;
      rowId: string;
      scopes: Record<string, string>;
    }[] = [];
    for (const ref of refs) {
      const compiled = this.#tables?.get(ref.tbl);
      if (compiled === undefined) continue; // table no longer in the schema
      const { rows } = await this.#exec.query<{ scopes: unknown }>(
        selectRowScopesSql(compiled, 'postgres'),
        [partition, ref.row_id],
      );
      const row = rows[0];
      if (row === undefined) continue;
      out.push({
        table: ref.tbl,
        rowId: ref.row_id,
        scopes: asJson<Record<string, string>>(row.scopes),
      });
    }
    return out;
  }

  async listReferencedBlobIds(partition: string): Promise<string[]> {
    const { rows } = await this.#exec.query<{ blob_id: string }>(
      'SELECT DISTINCT blob_id FROM sync_blob_refs WHERE partition=$1',
      [partition],
    );
    return rows.map((r) => r.blob_id);
  }

  // -- admin/console read surface --------------------------------------------

  async listClientRecords(partition: string): Promise<ClientRecord[]> {
    const { rows } = await this.#exec.query<{
      client_id: string;
      actor_id: string;
      wire_version: unknown;
      cursor: unknown;
      subscriptions: unknown;
      updated_at_ms: unknown;
    }>(
      'SELECT client_id, actor_id, wire_version, cursor, subscriptions, updated_at_ms FROM sync_clients WHERE partition=$1 ORDER BY updated_at_ms DESC',
      [partition],
    );
    return rows.map((r) => ({
      clientId: r.client_id,
      actorId: r.actor_id,
      wireVersion: asNumber(r.wire_version),
      cursor: asNumber(r.cursor),
      updatedAtMs: asNumber(r.updated_at_ms),
      subscriptions: (typeof r.subscriptions === 'string'
        ? JSON.parse(r.subscriptions)
        : r.subscriptions) as ClientSubscription[],
    }));
  }

  async listCommitMetadata(
    partition: string,
    query: CommitMetadataQuery,
  ): Promise<CommitMetadata[]> {
    const { rows } = query.table
      ? await this.#exec.query<{
          commit_seq: unknown;
          client_id: string;
          client_commit_id: string;
          actor_id: string;
          created_at_ms: unknown;
        }>(
          `SELECT c.commit_seq, c.client_id, c.client_commit_id, c.actor_id, c.created_at_ms
           FROM sync_commits c
           WHERE c.partition=$1 AND c.commit_seq>$2
             AND EXISTS (SELECT 1 FROM sync_changes ch
               WHERE ch.partition=c.partition AND ch.commit_seq=c.commit_seq AND ch.tbl=$3)
           ORDER BY c.commit_seq DESC LIMIT $4`,
          [partition, query.afterSeq, query.table, query.limit],
        )
      : await this.#exec.query<{
          commit_seq: unknown;
          client_id: string;
          client_commit_id: string;
          actor_id: string;
          created_at_ms: unknown;
        }>(
          `SELECT commit_seq, client_id, client_commit_id, actor_id, created_at_ms
           FROM sync_commits
           WHERE partition=$1 AND commit_seq>$2
           ORDER BY commit_seq DESC LIMIT $3`,
          [partition, query.afterSeq, query.limit],
        );
    const out: CommitMetadata[] = [];
    for (const row of rows) {
      const commitSeq = asNumber(row.commit_seq);
      const changes = await this.#exec.query<{ tbl: string; n: unknown }>(
        'SELECT tbl, count(*) AS n FROM sync_changes WHERE partition=$1 AND commit_seq=$2 GROUP BY tbl',
        [partition, commitSeq],
      );
      out.push({
        commitSeq,
        clientId: row.client_id,
        clientCommitId: row.client_commit_id,
        actorId: row.actor_id,
        createdAtMs: asNumber(row.created_at_ms),
        changeCount: changes.rows.reduce((sum, c) => sum + asNumber(c.n), 0),
        tables: changes.rows.map((c) => c.tbl),
      });
    }
    return out;
  }

  async scopeActivity(
    partition: string,
    query: ScopeActivityQuery,
  ): Promise<ScopeCommitActivity[]> {
    const { rows } = await this.#exec.query<{
      commit_seq: unknown;
      tbl: string;
    }>(
      `SELECT DISTINCT commit_seq, tbl FROM sync_change_scopes
       WHERE partition=$1 AND var=$2 AND value=$3
       ORDER BY commit_seq DESC LIMIT $4`,
      [partition, query.variable, query.value, query.limit],
    );
    const out: ScopeCommitActivity[] = [];
    for (const row of rows) {
      const commitSeq = asNumber(row.commit_seq);
      const meta = await this.#exec.query<{
        actor_id: string;
        created_at_ms: unknown;
      }>(
        'SELECT actor_id, created_at_ms FROM sync_commits WHERE partition=$1 AND commit_seq=$2',
        [partition, commitSeq],
      );
      const metaRow = meta.rows[0];
      if (metaRow === undefined) continue;
      const count = await this.#exec.query<{ n: unknown }>(
        'SELECT count(*) AS n FROM sync_changes WHERE partition=$1 AND commit_seq=$2 AND tbl=$3',
        [partition, commitSeq, row.tbl],
      );
      out.push({
        commitSeq,
        table: row.tbl,
        createdAtMs: asNumber(metaRow.created_at_ms),
        actorId: metaRow.actor_id,
        changeCount:
          count.rows[0] === undefined ? 0 : asNumber(count.rows[0].n),
      });
    }
    return out;
  }

  async getRowScopes(
    partition: string,
    table: string,
    rowId: string,
  ): Promise<
    { serverVersion: number; scopes: Record<string, string> } | undefined
  > {
    const { rows } = await this.#exec.query<{
      server_version: unknown;
      scopes: unknown;
    }>(selectRowScopesSql(this.table(table), 'postgres'), [partition, rowId]);
    const row = rows[0];
    if (row === undefined) return undefined;
    return {
      serverVersion: asNumber(row.server_version),
      scopes: (typeof row.scopes === 'string'
        ? JSON.parse(row.scopes)
        : row.scopes) as Record<string, string>,
    };
  }

  async listPartitions(): Promise<string[]> {
    return (await this.listPartitionRegistry()).map((entry) => entry.partition);
  }
}