// In-memory tables + views + sequence generation. No persistence, no external deps.
// This is the "database" — disposable and rebuilt from seed data on every server start.

import type { Row, SequenceFormat, TableDef, ViewDef } from '../types.ts';
import { type AuditContext, auditRecord, auditTableName, isAudited } from './audit.ts';
import { RECORD_ID, RecordClock, recordKey, TIMESTAMP, withoutRecordFields } from './identity.ts';
import { columnScale, storedValue, unlistedEnumValue, type ValueField } from './values.ts';
import {
  materializeEntries,
  planView,
  type PlannedAntiJoin,
  type PlannedJoin,
  type ViewEntry,
} from './views.ts';

// What a view is materialized from: the base rows, and a join's or
// anti-join's rows by its planned entry (undefined: the table as it is).
// Either left out reads the tables as they are.
export interface ViewRowsSource {
  base?: Row[];
  rowsOf?: (entry: PlannedJoin | PlannedAntiJoin) => Row[] | undefined;
}

type Pk = string;

// A field the store fills on insert when the record leaves it out: the
// sequencePrefix key, and every TableDef.generated entry it can mint.
type Generator =
  | { field: string; kind: 'SEQUENCE'; prefix: string }
  | { field: string; kind: 'AUTO_INCREMENT' };

interface TableConfig {
  pkFields: string[];
  sequencePrefix?: string;
  sequenceWidth: number;
  sequenceFormat: SequenceFormat;
  generators: Generator[];
  // The declared columns (TableDef.fields), whose values are held by type.
  columns?: ValueField[];
}

export interface StoreOptions {
  // Whether values written to a declared column are held by its type (see
  // db/values.ts). Read on every write; the mock server turns it off under
  // fidelity: 'legacy', where rows are kept as given. Default: on.
  normalizeValues?: () => boolean;
  // Whether a sequence another table shares is kept past the values that
  // table has used when one of them is registered (again). The mock server
  // turns it off under fidelity: 'legacy', where registering a table set the
  // counter from its own seed rows alone. Default: on.
  holdSharedSequences?: () => boolean;
}

// The generators of a table, the key's own sequence first. A SEQUENCE with
// neither its own prefix nor the table's can't mint anything and is left to
// the caller.
function generatorsOf(
  pkFields: string[],
  { sequencePrefix, generated = [] }: TableDef,
): Generator[] {
  const generators: Generator[] = [];
  if (sequencePrefix && pkFields.length === 1) {
    generators.push({ field: pkFields[0], kind: 'SEQUENCE', prefix: sequencePrefix });
  }
  for (const entry of generated) {
    if (generators.some((generator) => generator.field === entry.field)) continue;
    if (entry.kind === 'AUTO_INCREMENT') {
      generators.push({ field: entry.field, kind: 'AUTO_INCREMENT' });
      continue;
    }
    const prefix = entry.prefix ?? sequencePrefix;
    if (prefix) generators.push({ field: entry.field, kind: 'SEQUENCE', prefix });
  }
  return generators;
}

const isMissing = (value: unknown) => value === undefined || value === null;

// Composite keys join the pk fields' values with '|' (so values must not
// contain '|'). Single-field keys stay the raw value, preserving
// getRow('TABLE', 'TR001')-style lookups.
const COMPOSITE_KEY_SEPARATOR = '|';

const MAX_REMOVED_PER_KEY = 16;
const MAX_REMOVED_KEYS = 10_000;

// IdStrategy.java: a 15-digit counter, the sequence id, the location
// (DbConfig's default 'LO') and the system id (always 1).
const GENESIS_SEQUENCE_DIGITS = 15;
const GENESIS_LOCATION = 'LO';
const GENESIS_SYSTEM = '1';

function normalizePkFields(pkField: string | string[]): string[] {
  return Array.isArray(pkField) ? pkField : [pkField];
}

function keyFor(pkFields: string[], row: Row): Pk {
  if (pkFields.length === 1) return row[pkFields[0]];
  return pkFields.map((field) => String(row[field])).join(COMPOSITE_KEY_SEPARATOR);
}

function escapeRegExp(text: string): string {
  return text.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
}

// The counter inside a generated value, or undefined when the value isn't
// one a sequence with this prefix and format could have produced.
function sequenceNumber(
  prefix: string,
  format: SequenceFormat,
  value: unknown,
): number | undefined {
  if (typeof value !== 'string') return undefined;
  const digits =
    format === 'genesis'
      ? new RegExp(`^(\\d+)${escapeRegExp(prefix)}${GENESIS_LOCATION}${GENESIS_SYSTEM}$`).exec(
          value,
        )?.[1]
      : value.startsWith(prefix)
        ? value.slice(prefix.length)
        : undefined;
  if (digits === undefined) return undefined;
  const n = parseInt(digits, 10);
  return Number.isFinite(n) ? n : undefined;
}

// Counters are shared by tables using the same prefix in the same format.
function counterKey(prefix: string, format: SequenceFormat): string {
  return format === 'genesis' ? `${prefix}\u0000genesis` : prefix;
}

// An AUTO_INCREMENT counter belongs to its table's field.
const autoIncrementKey = (tableName: string, field: string) => `${tableName}\u0000${field}`;

// What a transaction keeps of a table from before its first write in it: the
// rows, and the removed and previous records the write's bookkeeping changes.
// Row objects are never changed in place once stored, so copying the maps
// is enough.
interface TableSavepoint {
  rows: Map<Pk, Row>;
  removed: Map<Pk, Row[]> | undefined;
  previous: Map<string, Row> | undefined;
}

function restoreEntry<V>(maps: Map<string, V>, tableName: string, saved: V | undefined): void {
  if (saved === undefined) maps.delete(tableName);
  else maps.set(tableName, saved);
}

export class Store {
  private tables = new Map<string, Map<Pk, Row>>();
  private tableConfig = new Map<string, TableConfig>();
  private sequenceCounters = new Map<string, number>();
  // Per sequence counter and table, the highest value the table has taken
  // from it: its seed rows' and the ones its writes used.
  private sequenceUse = new Map<string, Map<string, number>>();
  private autoIncrementCounters = new Map<string, number>();
  private clock = new RecordClock();
  // Immutable snapshot of each table's rows as registered, used only for field
  // metadata derivation (see getMetadataRows) so deleting every live row can't
  // collapse a resource's schema to zero fields.
  private seedRows = new Map<string, Row[]>();
  // Per table and key, every record the key lost (deleted, or displaced by
  // an insert or a key change landing on it) and no broadcast has announced
  // yet, oldest first — see takeRemovedRows. Bounded: a key keeps its last
  // MAX_REMOVED_PER_KEY records and a table its last MAX_REMOVED_KEYS keys,
  // for handlers that delete without ever broadcasting.
  private removedRows = new Map<string, Map<Pk, Row[]>>();
  // Per table, each record a modify changed since a broadcast last announced
  // it, as it was before the first such modify (by RECORD_ID) — see
  // takePreviousRecord. Bounded like removedRows.
  private previousRows = new Map<string, Map<string, Row>>();
  // While a transaction is open (see begin), each table it has written to,
  // as it was before.
  private savepoint: Map<string, TableSavepoint> | undefined;
  // Each audited table's audit table (db/audit.ts), and while an event's
  // transaction is open, the event its writes are audited for and the audit
  // rows they added.
  private auditTables = new Map<string, string>();
  private auditContext: AuditContext | undefined;
  private auditWrites: Array<{ tableName: string; key: Pk }> = [];
  readonly views = new Map<string, ViewDef>();
  private readonly normalizeValues: () => boolean;
  private readonly holdSharedSequences: () => boolean;
  // ENUM values already warned about, as 'TABLE.FIELD=value'.
  private readonly unlistedWarned = new Set<string>();

  constructor({
    normalizeValues = () => true,
    holdSharedSequences = () => true,
  }: StoreOptions = {}) {
    this.normalizeValues = normalizeValues;
    this.holdSharedSequences = holdSharedSequences;
  }

  // Opens a transaction: the commit-event path wraps every handler in one
  // (handlers/commitEvent.ts), so an event that NACKs leaves the tables as
  // they were, as a GSF event whose onValidate NACKs never reaches its
  // onCommit. Writes go through as usual; the first write to a table saves
  // it, and rollback() puts back every table the transaction wrote to.
  // Sequence and auto-increment counters and the RECORD_ID clock are not put
  // back: a value used stays used, as a database sequence's does. Not nested,
  // and the engine's own: an event that finds one open answers INTERNAL_ERROR.
  // Saving copies the table's row map, O(rows) per event and table written.
  // With an audit context, each write to an audited table also adds its
  // audit row (see audit.ts), undone with the rest by rollback().
  begin(audit?: AuditContext): void {
    if (this.savepoint) throw new Error('A store transaction is already open');
    this.savepoint = new Map();
    this.auditContext = audit;
    this.auditWrites = [];
  }

  // Keeps every write since begin().
  commit(): void {
    this.savepoint = undefined;
    this.auditContext = undefined;
  }

  // The audit rows the open transaction added (table and key), for the live
  // pushes that announce them once it commits.
  takeAuditWrites(): Array<{ tableName: string; key: Pk }> {
    return this.auditWrites.splice(0);
  }

  // Undoes every write since begin().
  rollback(): void {
    const savepoint = this.savepoint;
    this.savepoint = undefined;
    this.auditContext = undefined;
    this.auditWrites = [];
    for (const [tableName, saved] of savepoint ?? []) {
      this.tables.set(tableName, saved.rows);
      restoreEntry(this.removedRows, tableName, saved.removed);
      restoreEntry(this.previousRows, tableName, saved.previous);
    }
  }

  get inTransaction(): boolean {
    return this.savepoint !== undefined;
  }

  // Saves a table before the open transaction's first change to it.
  private save(tableName: string): void {
    if (!this.savepoint || this.savepoint.has(tableName)) return;
    const rows = this.tables.get(tableName);
    if (!rows) return;
    const removed = this.removedRows.get(tableName);
    const previous = this.previousRows.get(tableName);
    this.savepoint.set(tableName, {
      rows: new Map(rows),
      removed: removed && new Map(removed),
      previous: previous && new Map(previous),
    });
  }

  registerTable(tableName: string, definition: TableDef): this {
    const {
      pkField,
      sequencePrefix,
      sequenceWidth = 3,
      sequenceFormat = 'prefix',
      rows = [],
    } = definition;
    const pkFields = normalizePkFields(pkField);
    const config: TableConfig = {
      pkFields,
      sequencePrefix,
      sequenceWidth,
      sequenceFormat,
      generators: generatorsOf(pkFields, definition),
      ...(definition.fields
        ? {
            // Key columns stay as given: every lookup (getRow, updateRow,
            // broadcastTableChange) names a row by the key the caller has.
            columns: definition.fields
              .filter((field) => !pkFields.includes(field.name))
              .map((field) => ({
                name: field.name,
                type: field.type,
                scale: columnScale(field),
                values: field.values,
              })),
          }
        : {}),
    };
    const seeded = rows.map((row) => this.withStoredValues(tableName, config, { ...row }));
    const table = new Map<Pk, Row>();
    const seenIds = new Set<string>();
    let restamped = 0;
    for (const row of seeded) {
      const stamped = this.stampSeedRow({ ...row }, seenIds);
      if (stamped.restamped) restamped += 1;
      table.set(keyFor(pkFields, row), stamped.row);
    }
    if (restamped > 0) {
      console.warn(
        `[mock-server] ${tableName}: re-stamped ${restamped} seed row(s) whose RECORD_ID was ` +
          'not a safe integer or a string, or repeated another row, so every ROW_REF stays unique',
      );
    }
    this.tables.set(tableName, table);
    this.tableConfig.set(tableName, config);
    if (isAudited(definition)) this.auditTables.set(tableName, auditTableName(tableName));
    else this.auditTables.delete(tableName);
    // Registering a table again replaces it whole: a rollback leaves it be.
    this.savepoint?.delete(tableName);
    this.removedRows.delete(tableName);
    this.previousRows.delete(tableName);
    this.seedRows.set(
      tableName,
      seeded.map((row) => ({ ...row })),
    );

    // Each counter starts after the highest value the seed rows carry (two
    // fields of one table sharing a prefix share the counter). The key's
    // sequence only exists for a single-field pk. A sequence is shared by
    // every table with its prefix: it also starts after the values the other
    // tables on it have taken, so no id repeats across them (an audit
    // table's default sequence AU, an app table on the same prefix). This
    // table's own earlier values don't count: registering it again resets it.
    const starts = new Map<string, number>();
    for (const generator of config.generators) {
      let maxN = 0;
      for (const row of seeded) {
        const n = this.generatedNumber(config, generator, row[generator.field]);
        if (n !== undefined) maxN = Math.max(maxN, n);
      }
      const counter = this.counterOf(tableName, config, generator);
      starts.set(counter, Math.max(starts.get(counter) ?? 1, maxN + 1));
    }
    for (const generator of config.generators) {
      const counter = this.counterOf(tableName, config, generator);
      const seeded = starts.get(counter)!;
      let start = seeded;
      if (generator.kind === 'SEQUENCE') {
        const use = this.sequenceUse.get(counter) ?? new Map<string, number>();
        this.sequenceUse.set(counter, use.set(tableName, seeded - 1));
        if (this.holdSharedSequences()) {
          const others = [...use].filter(([table]) => table !== tableName).map(([, n]) => n);
          start = Math.max(seeded, ...others.map((n) => n + 1));
        }
      }
      this.countersFor(generator).set(counter, start);
    }
    return this;
  }

  private counterOf(tableName: string, config: TableConfig, generator: Generator): string {
    return generator.kind === 'SEQUENCE'
      ? counterKey(generator.prefix, config.sequenceFormat)
      : autoIncrementKey(tableName, generator.field);
  }

  private countersFor(generator: Generator): Map<string, number> {
    return generator.kind === 'SEQUENCE' ? this.sequenceCounters : this.autoIncrementCounters;
  }

  // The counter inside a value a generator could have produced.
  private generatedNumber(
    config: TableConfig,
    generator: Generator,
    value: unknown,
  ): number | undefined {
    if (generator.kind === 'SEQUENCE') {
      return sequenceNumber(generator.prefix, config.sequenceFormat, value);
    }
    // A key column keeps a seeded '2' as given, but the counter still has to
    // start after it.
    const number = typeof value === 'string' && /^\d+$/.test(value) ? Number(value) : value;
    return typeof number === 'number' && Number.isSafeInteger(number) ? number : undefined;
  }

  // Throws on a join model GSF would refuse to compile (see planView).
  registerView(viewName: string, definition: ViewDef): this {
    planView(viewName, definition);
    this.views.set(viewName, definition);
    return this;
  }

  // Seed rows keep a RECORD_ID they bring (e.g. rows copied from a real
  // server) when it is usable: a safe integer, or a string, not used by an
  // earlier seed row. Anything else — a missing id, a number above 2^53 that
  // JSON.parse already rounded onto its neighbour's, a repeat — gets a fresh
  // stamp, in seed order, so no two rows share a ROW_REF. A missing or
  // unusable TIMESTAMP becomes the RECORD_ID.
  private stampSeedRow(row: Row, seenIds: Set<string>): { row: Row; restamped: boolean } {
    const given = row[RECORD_ID];
    const usable = (value: unknown) =>
      (typeof value === 'number' && Number.isSafeInteger(value)) ||
      (typeof value === 'string' && value !== '');
    const missing = given === undefined || given === null;
    const keep = !missing && usable(given) && !seenIds.has(String(given));
    if (keep) {
      this.clock.observe(given);
    } else {
      row[RECORD_ID] = this.clock.next();
    }
    seenIds.add(String(row[RECORD_ID]));
    if (!usable(row[TIMESTAMP])) row[TIMESTAMP] = row[RECORD_ID];
    this.clock.observe(row[TIMESTAMP]);
    return { row, restamped: !missing && !keep };
  }

  // Each declared column's value as the store holds it (db/values.ts
  // storedValue): dates as epoch millis, BigDecimals rounded to their scale,
  // numbers and booleans given as strings read as such. An ENUM value its
  // field doesn't list is kept, with a warning (once per value): the real
  // table could never hold it. Changes `row` in place and returns it.
  private withStoredValues(tableName: string, config: TableConfig, row: Row): Row {
    if (!config.columns || !this.normalizeValues()) return row;
    for (const column of config.columns) {
      if (!Object.hasOwn(row, column.name)) continue;
      row[column.name] = storedValue(column, row[column.name]);
      const unlisted = unlistedEnumValue(column, row[column.name]);
      const warning = `${tableName}.${column.name}=${String(unlisted)}`;
      if (unlisted === undefined || this.unlistedWarned.has(warning)) continue;
      this.unlistedWarned.add(warning);
      console.warn(
        `[mock-server] ${tableName}.${column.name}: ${JSON.stringify(unlisted)} is not one of ` +
          `its ENUM values (${column.values!.join(', ')})`,
      );
    }
    return row;
  }

  // A newly inserted row gets RECORD_ID = TIMESTAMP = a fresh stamp, whatever
  // it was given — as on the real server — appended after its own fields.
  private stampNewRow(row: Row): Row {
    delete row[RECORD_ID];
    delete row[TIMESTAMP];
    const stamp = this.clock.next();
    row[RECORD_ID] = stamp;
    row[TIMESTAMP] = stamp;
    return row;
  }

  private noteRemoved(tableName: string, key: Pk, row: Row): void {
    let removed = this.removedRows.get(tableName);
    if (!removed) {
      removed = new Map();
      this.removedRows.set(tableName, removed);
    }
    const records = removed.get(key) ?? [];
    // Re-set so the key moves to the end: the oldest keys are dropped first.
    removed.delete(key);
    removed.set(key, [...records, row].slice(-MAX_REMOVED_PER_KEY));
    if (removed.size > MAX_REMOVED_KEYS) removed.delete(removed.keys().next().value!);
  }

  private notePrevious(tableName: string, row: Row): void {
    const id = recordKey(row);
    if (id === undefined) return;
    let previous = this.previousRows.get(tableName);
    if (!previous) {
      previous = new Map();
      this.previousRows.set(tableName, previous);
    }
    if (previous.has(id)) return;
    previous.set(id, row);
    if (previous.size > MAX_REMOVED_KEYS) previous.delete(previous.keys().next().value!);
  }

  // A supplied value inside a generator's own range must advance its counter
  // too. Without this, inserting e.g. 'TR002' by hand leaves the counter at
  // 2, and the next generated insert mints 'TR002' again and silently
  // overwrites the hand-keyed row (Map.set, no error, event still acks
  // success).
  private observeGeneratedValues(tableName: string, config: TableConfig, row: Row): void {
    for (const generator of config.generators) {
      const n = this.generatedNumber(config, generator, row[generator.field]);
      if (n === undefined) continue;
      const counters = this.countersFor(generator);
      const counter = this.counterOf(tableName, config, generator);
      if (n + 1 > (counters.get(counter) ?? 1)) counters.set(counter, n + 1);
      if (generator.kind === 'SEQUENCE') {
        const use = this.sequenceUse.get(counter) ?? new Map<string, number>();
        if (n > (use.get(tableName) ?? 0)) this.sequenceUse.set(counter, use.set(tableName, n));
      }
    }
  }

  nextSequenceValue(sequencePrefix: string, width: number): string {
    const n = this.sequenceCounters.get(sequencePrefix) ?? 1;
    this.sequenceCounters.set(sequencePrefix, n + 1);
    return `${sequencePrefix}${String(n).padStart(width, '0')}`;
  }

  private generateSequenceValue(config: TableConfig, prefix: string): string {
    if (config.sequenceFormat !== 'genesis') {
      return this.nextSequenceValue(prefix, config.sequenceWidth);
    }
    const key = counterKey(prefix, 'genesis');
    const n = this.sequenceCounters.get(key) ?? 1;
    this.sequenceCounters.set(key, n + 1);
    return `${String(n).padStart(GENESIS_SEQUENCE_DIGITS, '0')}${prefix}${GENESIS_LOCATION}${GENESIS_SYSTEM}`;
  }

  // Fills every generated field the record leaves out (undefined or null).
  private generateValues(tableName: string, config: TableConfig, row: Row): void {
    for (const generator of config.generators) {
      if (!isMissing(row[generator.field])) continue;
      if (generator.kind === 'SEQUENCE') {
        row[generator.field] = this.generateSequenceValue(config, generator.prefix);
        continue;
      }
      const counter = autoIncrementKey(tableName, generator.field);
      const n = this.autoIncrementCounters.get(counter) ?? 1;
      this.autoIncrementCounters.set(counter, n + 1);
      row[generator.field] = n;
    }
  }

  getAllRows(tableName: string): Row[] {
    const table = this.tables.get(tableName);
    if (!table) return [];
    return Array.from(table.values());
  }

  getRow(tableName: string, pk: Pk): Row | undefined {
    return this.tables.get(tableName)?.get(pk);
  }

  // The storage/broadcast key for a row: the pk field's value, or for
  // composite-keyed tables the pk fields' values joined with '|'. Event
  // handlers on composite tables pass this to broadcastTableChange.
  rowKey(tableName: string, row: Row): Pk {
    const config = this.tableConfig.get(tableName);
    if (!config) throw new Error(`Unknown table: ${tableName}`);
    return keyFor(config.pkFields, row);
  }

  // Fallback matching for tables where the caller doesn't have the full key:
  // try the pk field(s) first, then any single field that uniquely identifies
  // one row, then the full combination of provided fields.
  findRow(tableName: string, criteria: Row): Row | undefined {
    const table = this.tables.get(tableName);
    if (!table) return undefined;
    const { pkFields } = this.tableConfig.get(tableName) ?? { pkFields: [] };

    if (pkFields.length > 0 && pkFields.every((field) => criteria[field] !== undefined)) {
      const byKey = table.get(keyFor(pkFields, criteria));
      if (byKey) return byKey;
    }

    const rows = Array.from(table.values());
    const matchesAllCriteria = (row: Row) =>
      Object.entries(criteria).every(([field, value]) => row[field] === value);
    for (const [field, value] of Object.entries(criteria)) {
      const matches = rows.filter((row) => row[field] === value);
      if (matches.length === 1 && matchesAllCriteria(matches[0])) return matches[0];
    }

    return rows.find(matchesAllCriteria);
  }

  // A copy of the record with its missing generated fields filled in, as the
  // database does before checking its unique indexes. Each value used is
  // gone even if the record is then rejected, as a real sequence's is.
  withGeneratedValues(tableName: string, record: Row): Row {
    const config = this.tableConfig.get(tableName);
    if (!config) throw new Error(`Unknown table: ${tableName}`);
    const row = { ...record };
    this.generateValues(tableName, config, row);
    return row;
  }

  // Generates every missing (undefined or null) sequence and auto-increment
  // field — the sequencePrefix key and TableDef.generated — and keeps a
  // supplied one, as GSF does; stamps RECORD_ID/TIMESTAMP. Declared columns
  // are held by type (withStoredValues), as updateRow's patch is. A row already
  // stored under the same key is replaced (the real database would reject the
  // insert with DUPLICATE_KEY — check first and throw NackError.duplicateKey
  // for that, as the generic CRUD events do).
  insertRow(tableName: string, record: Row): Row {
    const table = this.tables.get(tableName);
    const config = this.tableConfig.get(tableName);
    if (!table || !config) throw new Error(`Unknown table: ${tableName}`);

    this.save(tableName);
    const row = this.withStoredValues(tableName, config, { ...record });
    this.generateValues(tableName, config, row);
    const missing = config.pkFields.filter((field) => row[field] === undefined);
    if (missing.length > 0) {
      throw new Error(
        `Primary key field(s) '${missing.join("', '")}' are required for inserting into table '${tableName}'`,
      );
    }
    this.observeGeneratedValues(tableName, config, row);
    this.stampNewRow(row);
    const key = keyFor(config.pkFields, row);
    const displaced = table.get(key);
    if (displaced) this.noteRemoved(tableName, key, displaced);
    table.set(key, row);
    this.audit(tableName, row);
    return row;
  }

  // Adds a write's audit row, when the table is audited and an event's
  // transaction is open.
  private audit(tableName: string, record: Row): void {
    const auditTable = this.auditTables.get(tableName);
    const context = this.auditContext;
    if (!auditTable || !context || !this.tables.has(auditTable)) return;
    const row = this.insertRow(auditTable, auditRecord(record, context, Date.now()));
    this.auditWrites.push({ tableName: auditTable, key: this.rowKey(auditTable, row) });
  }

  // Merges the patch, moves TIMESTAMP on and keeps RECORD_ID (a patch can't
  // change a row's identity).
  updateRow(tableName: string, pk: Pk, patch: Row): Row | undefined {
    const table = this.tables.get(tableName);
    const config = this.tableConfig.get(tableName);
    if (!table || !config || !table.has(pk)) return undefined;
    this.save(tableName);
    const existing = table.get(pk)!;
    this.notePrevious(tableName, existing);
    const updated = { ...existing, ...this.withStoredValues(tableName, config, { ...patch }) };
    updated[RECORD_ID] = existing[RECORD_ID];
    updated[TIMESTAMP] = this.clock.next();
    // A patch that changes a pk field (e.g. an entity-management form posting
    // its full DETAILS after the user edited the key column) must move the row
    // to its new key. Leaving it under the old one makes the index disagree
    // with the row itself: getRow(newKey) misses, while rowKey()/getRowRef()
    // report the new key, so the matching broadcast finds nothing to push.
    const nextKey = keyFor(config.pkFields, updated);
    if (nextKey !== pk) {
      table.delete(pk);
      const displaced = table.get(nextKey);
      if (displaced) this.noteRemoved(tableName, nextKey, displaced);
    }
    table.set(nextKey, updated);
    this.audit(tableName, updated);
    return updated;
  }

  deleteRow(tableName: string, pk: Pk): boolean {
    const table = this.tables.get(tableName);
    const existing = table?.get(pk);
    if (!table || !existing) return false;
    this.save(tableName);
    table.delete(pk);
    this.noteRemoved(tableName, pk, existing);
    this.audit(tableName, existing);
    return true;
  }

  // Reads and clears every record a key lost that no broadcast announced
  // yet, oldest first: what a DELETE broadcast names (their RECORD_IDs are
  // the ROW_REFs a client may hold — held-rows tracking drops the ones it
  // doesn't), including the old record when a key's row was replaced before
  // one broadcast.
  takeRemovedRows(tableName: string, key: Pk): Row[] {
    this.save(tableName);
    const removed = this.removedRows.get(tableName);
    const records = removed?.get(key) ?? [];
    removed?.delete(key);
    return records;
  }

  // Reads and clears how a record was before the modifies no broadcast has
  // announced yet changed it (undefined when none did): with the removed
  // records, what a live change to a joined table is diffed against.
  takePreviousRecord(tableName: string, row: Row): Row | undefined {
    const id = recordKey(row);
    const previous = this.previousRows.get(tableName);
    if (id === undefined || !previous) return undefined;
    this.save(tableName);
    const record = previous.get(id);
    previous.delete(id);
    return record;
  }

  // The same, for a row of a table or view given by its (base) pk fields —
  // what a public broadcast(resource, 'DELETE', pkStub) names.
  takeRemovedRecordsFor(resourceName: string, row: Row): Row[] {
    const located = this.locateRecord(resourceName, row);
    return located ? this.takeRemovedRows(located.tableName, located.key) : [];
  }

  // Materializes a view (see the README's "Views: the join model"): base
  // rows minus anti-joined ones, joined in order — OUTER keeps a row with no
  // match, INNER drops it — plus the select aliases and derived fields.
  // `tables` substitutes the rows of any table the view reads (its base or a
  // joined one) — e.g. a table's earlier state, to rebuild the view as it
  // was.
  getViewRows(viewName: string, tables: Record<string, Row[]> = {}): Row[] {
    const rowsOf = (tableName: string) =>
      Object.hasOwn(tables, tableName) ? tables[tableName] : this.getAllRows(tableName);
    return this.getViewRowsFrom(viewName, {
      base: rowsOf(this.viewOf(viewName).base),
      rowsOf: (entry) => rowsOf(entry.table),
    });
  }

  // The same, with the base rows given apart from the joined ones — e.g. one
  // base row, to build that row's view row when the base table is also joined
  // — and a join's or anti-join's rows given by its planned entry (undefined:
  // the table as it is).
  getViewRowsFrom(viewName: string, source: ViewRowsSource): Row[] {
    return this.getViewEntriesFrom(viewName, source).map((entry) => entry.row);
  }

  // The same, each row with the row every join matched (in join order).
  getViewEntriesFrom(viewName: string, source: ViewRowsSource): ViewEntry[] {
    const view = this.viewOf(viewName);
    return materializeEntries(viewName, view, {
      base: source.base ?? this.getAllRows(view.base),
      rowsOf: (entry) => source.rowsOf?.(entry) ?? this.getAllRows(entry.table),
    });
  }

  private viewOf(viewName: string): ViewDef {
    const view = this.views.get(viewName);
    if (!view) throw new Error(`Unknown view: ${viewName}`);
    return view;
  }

  // Resolves a queryable resource (table or view) by name to its current rows.
  getResourceRows(resourceName: string): Row[] {
    if (this.views.has(resourceName)) return this.getViewRows(resourceName);
    return this.getAllRows(resourceName);
  }

  // Rows to derive META_REQUEST/JSON_SCHEMA_REQUEST field metadata from:
  // current rows when there are any, else the registration-time seed rows
  // (re-joined through the view definition for views). Without this fallback,
  // deleting a table's last row would make the next metadata reply say the
  // resource has zero fields — which entity-management renders as a form with
  // no inputs and a delete button that can't build a payload, silently.
  // RECORD_ID and TIMESTAMP are left out: they are not fields a dataserver
  // row or an event's DETAILS carry.
  getMetadataRows(resourceName: string): Row[] {
    const live = this.getResourceRows(resourceName);
    if (live.length > 0) return live.map(withoutRecordFields);
    const view = this.views.get(resourceName);
    const seedOf = (tableName: string) => this.seedRows.get(tableName) ?? [];
    const seed = view
      ? this.getViewRowsFrom(resourceName, {
          base: seedOf(view.base),
          rowsOf: (entry) => seedOf(entry.table),
        })
      : seedOf(resourceName);
    return seed.map(withoutRecordFields);
  }

  // Whether tableName is a table (not a view).
  hasTable(tableName: string): boolean {
    return this.tableConfig.has(tableName);
  }

  // Row-identity fields for a table or view; empty when unknown.
  getPkFields(resourceName: string): string[] {
    const view = this.views.get(resourceName);
    if (view) {
      if (view.pkField) return normalizePkFields(view.pkField);
      return this.tableConfig.get(view.base)?.pkFields ?? [];
    }
    return this.tableConfig.get(resourceName)?.pkFields ?? [];
  }

  // The pk-based row reference: the pk field's value, or the composite key.
  // This was DETAILS.ROW_REF up to 15.47 and still is under
  // fidelity: 'legacy'; by default ROW_REF is getRecordRef's.
  getRowRef(resourceName: string, row: Row): unknown {
    const pkFields = this.getPkFields(resourceName);
    if (pkFields.length === 0) return undefined;
    return keyFor(pkFields, row);
  }

  // The DETAILS.ROW_REF GSF stamps on dataserver rows: the row's RECORD_ID as
  // a string (GenesisSetMaskingJsonSerializer: value.toString()). A view row
  // carries its base row's RECORD_ID. A row passed without one (a hand-built
  // row, or a pk-only stub) is matched to the stored record by its pk — a
  // just-deleted record included — and only when that fails too does this
  // fall back to the pk-based reference.
  getRecordRef(resourceName: string, row: Row): unknown {
    const id = recordKey(row) ?? recordKey(this.findStoredRecord(resourceName, row));
    return id ?? this.getRowRef(resourceName, row);
  }

  private locateRecord(resourceName: string, row: Row): { tableName: string; key: Pk } | undefined {
    const view = this.views.get(resourceName);
    const tableName = view ? view.base : resourceName;
    const pkFields = this.tableConfig.get(tableName)?.pkFields ?? [];
    if (pkFields.length === 0 || pkFields.some((field) => row[field] === undefined)) {
      return undefined;
    }
    return { tableName, key: keyFor(pkFields, row) };
  }

  // The stored record a pk-only row names: the live one, else the newest
  // record removed from that key and not yet broadcast.
  private findStoredRecord(resourceName: string, row: Row): Row | undefined {
    const located = this.locateRecord(resourceName, row);
    if (!located) return undefined;
    const { tableName, key } = located;
    return (
      this.tables.get(tableName)?.get(key) ?? this.removedRows.get(tableName)?.get(key)?.at(-1)
    );
  }

  // Rebuilds a DELETE stub row carrying the pk field(s) from a stored key, so
  // subscribers can be told which row to drop after it's already gone.
  // (The DELETE body under fidelity: 'legacy'; GSF sends DETAILS only.)
  rowStubFromKey(resourceName: string, key: Pk): Row {
    const pkFields = this.getPkFields(resourceName);
    if (pkFields.length <= 1) return pkFields.length ? { [pkFields[0]]: key } : {};
    const parts = String(key).split(COMPOSITE_KEY_SEPARATOR);
    return Object.fromEntries(pkFields.map((field, i) => [field, parts[i]]));
  }
}