import http from 'node:http';
import { WebSocketServer, type WebSocket } from 'ws';
import {
  auditTableDef,
  auditTableName,
  checkNotDeclared,
  isAudited,
  withAuditTables,
} from './db/audit.ts';
import { recordKey } from './db/identity.ts';
import { compareRecordIds } from './db/ordering.ts';
import { Store } from './db/store.ts';
import {
  entriesReading,
  planView,
  type PlannedAntiJoin,
  type PlannedJoin,
  type ViewEntry,
} from './db/views.ts';
import { createAuthState } from './handlers/auth.ts';
import { handleCommitEvent } from './handlers/commitEvent.ts';
import { crudEventHandler } from './handlers/crudEvents.ts';
import { pendingKey, viewRow, waitForPage } from './handlers/dataLogon.ts';
import {
  cookieAuthentication,
  createHttpRequestListener,
  parseCookies,
  SESSION_AUTH_TOKEN_COOKIE,
} from './handlers/http.ts';
import { permittedQueryRow, requestingUser } from './handlers/resourceAuth.ts';
import { Connection, type Subscription, type ViewRow } from './protocol/connection.ts';
import { isLegacyFidelity, liveModifyMode } from './protocol/fidelity.ts';
import { replyNotLoggedOn, routeMessage } from './protocol/router.ts';
import { dataserverRow, stampRow, type RowOperation } from './protocol/rowUpdate.ts';
import type {
  BroadcastFn,
  BroadcastTableChangeFn,
  EventHandler,
  GenesisMessage,
  MockServerConfig,
  QueryDef,
  RequestReplyDef,
  Row,
  TableDef,
  ViewDef,
} from './types.ts';

// WebSocket.OPEN: what a socket of the engine's own reports.
const WS_OPEN = 1;

// One view row a live change pushes, and how.
type ViewChange = [RowOperation, Row];

// A broadcast change to one table (see tableChange in createMockServer).
interface TableChange {
  before: () => Row[];
  after: () => Row[];
  records: Set<string>;
}

// Whether two values are the same: the same primitive, or the same JSON.
function sameValue(a: unknown, b: unknown): boolean {
  return Object.is(a, b) || JSON.stringify(a) === JSON.stringify(b);
}

// Whether two rows carry the same columns and values.
function sameValues(a: Row, b: Row): boolean {
  const keys = new Set([...Object.keys(a), ...Object.keys(b)]);
  for (const key of keys) {
    if (!sameValue(a[key], b[key])) return false;
  }
  return true;
}

// What a live MODIFY sends a client that holds `held`: each field whose value
// differs in `next`, a field that became null (or that `next` no longer
// carries, as a column now hidden from the user) as null — as GSF's
// diffToGenesisSet and the hidden-field handling of
// QueryUpdate.Modified.setModifyFromQueryRow write it (v8.15.29). Null and
// absent are the same value. In column order.
function changedFields(held: Row, next: Row): Row {
  const changed: Row = {};
  for (const field of new Set([...Object.keys(held), ...Object.keys(next)])) {
    const value = next[field] ?? null;
    if (!sameValue(held[field] ?? null, value)) changed[field] = value;
  }
  return changed;
}

// The held row a moving view evicts: the oldest version the client holds, by
// TIMESTAMP and then RECORD_ID, as MovingViewClientIndex's timestampIndex
// orders them (v8.15.29) — not the row last in the snapshot order. Rows with
// no ROW_REF can't be named in a DELETE, so are never chosen.
function oldestHeld(held: Map<string, ViewRow>): [string, ViewRow] | undefined {
  let oldest: [string, ViewRow] | undefined;
  for (const candidate of held) {
    const [, entry] = candidate;
    if (entry.rowRef === undefined || entry.rowRef === null) continue;
    if (
      !oldest ||
      (compareRecordIds(entry.timestamp, oldest[1].timestamp) ||
        compareRecordIds(entry.rowRef, oldest[1].rowRef)) < 0
    ) {
      oldest = candidate;
    }
  }
  return oldest;
}

export interface MockServer {
  start(port: number): WebSocketServer;
  // Like start(), but resolves with the ACTUAL bound port once listening —
  // pass 0 (the default) to let the OS pick a free one. This is the way to
  // run several servers in parallel (per test worker/shard) without
  // hand-assigning ports.
  listen(port?: number): Promise<number>;
  stop(): Promise<void>;
  store: Store;
  broadcast: BroadcastFn;
  broadcastTableChange: BroadcastTableChangeFn;
  setTable(tableName: string, tableDef: TableDef): void;
  setView(viewName: string, viewDef: ViewDef): void;
  setQuery(resourceName: string, queryDef: QueryDef): void;
  setRequestReply(resourceName: string, requestReplyDef: RequestReplyDef): void;
  setEventHandler(eventName: string, handler: EventHandler): void;
  getQuery(resourceName: string): QueryDef | undefined;
  getRequestReply(resourceName: string): RequestReplyDef | undefined;
  getEventHandler(eventName: string): EventHandler | undefined;
  extendQuery(resourceName: string, augment: (current: QueryDef | undefined) => QueryDef): void;
  extendRequestReply(
    resourceName: string,
    augment: (current: RequestReplyDef | undefined) => RequestReplyDef,
  ): void;
  extendEventHandler(
    eventName: string,
    augment: (current: EventHandler | undefined) => EventHandler,
  ): void;
}

// The status a cookie-authenticated router refuses an upgrade with.
const HTTP_UNAUTHORIZED = 401;

export function createMockServer(config: MockServerConfig): MockServer {
  // Declared columns are held by type (db/values.ts), except under
  // fidelity: 'legacy', which keeps rows as given.
  const store = new Store({
    normalizeValues: () => !isLegacyFidelity(config),
    holdSharedSequences: () => !isLegacyFidelity(config),
  });
  // An audited table's audit table is a table like any other from here on.
  if (config.tables) config.tables = withAuditTables(config.tables);
  for (const [tableName, tableDef] of Object.entries(config.tables ?? {})) {
    store.registerTable(tableName, tableDef);
  }
  for (const [viewName, viewDef] of Object.entries(config.views ?? {})) {
    store.registerView(viewName, viewDef);
  }

  const connections = new Set<Connection>();

  // Pushes an unsolicited row update to every currently-subscribed connection
  // whose DATA_LOGON resource matches, filtered by that subscription's own
  // CRITERIA_MATCH. Event handlers call this after mutating the store so open
  // grids reflect inserts/amends/deletes without the client re-querying.
  // DELETE rows are PK-only stubs (see broadcastTableChange), so a criteria
  // predicate would evaluate against missing fields and drop the push —
  // leaving the client holding a ghost row. A DELETE therefore skips the
  // predicate; which subscriptions it reaches is decided by what each client
  // holds (viewUpdate), or, in legacy mode, it reaches all of them.
  // 'error' is kept distinct from 'no-match' because only a clean no-match may
  // be turned into a DELETE below; a broken predicate must not evict rows.
  function evaluateSubscription(
    subscription: Subscription,
    sourceRef: string,
    row: Row,
  ): 'match' | 'no-match' | 'error' {
    try {
      return subscription.predicate(row) ? 'match' : 'no-match';
    } catch (error) {
      console.error(
        `[mock-server] Error evaluating predicate for subscription ${sourceRef}:`,
        error,
      );
      return 'error';
    }
  }

  // A row as one subscription sends it: its columns (dataserverRow), each
  // BigDecimal a scaled string (Subscription.encode).
  function wireRow(subscription: Subscription, permitted: Row, row: Row): Row {
    const sent = dataserverRow(permitted, row, config.queries?.[subscription.resourceName]?.fields);
    return subscription.encode ? subscription.encode(sent) : sent;
  }

  // The operation to push to one subscription, or null to skip it entirely.
  // A MODIFY that moved the row out of this subscription's CRITERIA_MATCH
  // becomes a DELETE so the client drops it rather than keeping a stale ghost
  // row; a predicate that threw skips instead, since a broken filter must
  // never evict rows.
  function resolveSubscriptionOperation(
    subscription: Subscription,
    sourceRef: string,
    operation: RowOperation,
    row: Row,
  ): RowOperation | null {
    if (operation === 'DELETE') return 'DELETE';
    const verdict = evaluateSubscription(subscription, sourceRef, row);
    if (verdict === 'match') return operation;
    if (verdict === 'error') return null;
    return operation === 'MODIFY' ? 'DELETE' : null;
  }

  // The rows of the QUERY_UPDATE one 'gsf' subscription's client gets for a
  // live change, decided from what that client holds as the real
  // dataserver's client index decides (ClientIndex and MovingViewClientIndex,
  // v8.15.29); none to push nothing. Keeps the subscription's pending and
  // held rows up to date. `operation` is the change after the query filter
  // and CRITERIA_MATCH (a MODIFY that left them arrives as a DELETE); the
  // user's rowFilter is applied here too, so leaving it counts as leaving the
  // subscription.
  // - A held row gets a MODIFY of what changed since the client last got it
  //   (liveModify 'full': the whole row), and nothing when nothing it can
  //   see changed; a DELETE when it is gone or out of scope. An INSERT of a
  //   row it holds is ignored.
  // - A row the client doesn't hold that is (now) in scope is pushed as an
  //   INSERT and becomes held: a new row, one waiting for MORE_ROWS, one a
  //   MODIFY brought back into the criteria (processInsertUpdate, and
  //   putNewRowEntry for a MODIFY). The snapshot's order doesn't matter, and
  //   no later page repeats it. A full view takes it only when it moves:
  //   the held row with the oldest TIMESTAMP is evicted, as a DELETE after
  //   the INSERT in the same frame (nothing at all when that is the new row
  //   itself). Without a moving view a full view takes nothing
  //   (MaxViewReached), and the row waits for a page if it sorts after the
  //   read position (waitForPage) — as an evicted row does too.
  // - Out of scope or deleted, a waiting row is dropped and a row the client
  //   never held gets nothing.
  // A row without a ROW_REF can't be matched to anything the client holds,
  // so changes to it aren't pushed.
  function viewUpdate(
    subscription: Subscription,
    connection: Connection,
    query: QueryDef | undefined,
    operation: RowOperation,
    row: Row,
    rowRef: unknown,
  ): Row[] {
    const key = pendingKey(rowRef);
    const { pending, held } = subscription;
    if (key === undefined || !pending || !held) return [];
    const permitted =
      operation === 'DELETE'
        ? undefined
        : permittedQueryRow(row, query?.auth, requestingUser(connection), subscription.fieldNames);
    const current = held.get(key);
    if (!permitted) {
      pending.delete(key);
      if (!current) return [];
      held.delete(key);
      return [stampRow({}, rowRef, 'DELETE')];
    }
    const entry = viewRow(wireRow(subscription, permitted, row), rowRef, row);
    if (current) {
      if (operation === 'INSERT') return [];
      held.set(key, entry);
      const changed = changedFields(current.row, entry.row);
      if (Object.keys(changed).length === 0) return [];
      return [stampRow(liveModifyMode(config) === 'full' ? entry.row : changed, rowRef, 'MODIFY')];
    }
    const waiting = (view: ViewRow): ViewRow => ({
      ...view,
      row: stampRow(view.row, view.rowRef, 'INSERT'),
    });
    const maxView = subscription.maxView ?? Infinity;
    if (!subscription.movingView && held.size >= maxView) {
      waitForPage(subscription, key, waiting(entry));
      return [];
    }
    pending.delete(key);
    held.set(key, entry);
    const rows = [stampRow(entry.row, rowRef, 'INSERT')];
    const evicted = held.size > maxView ? oldestHeld(held) : undefined;
    if (evicted) {
      const [evictedKey, evictedRow] = evicted;
      held.delete(evictedKey);
      waitForPage(subscription, evictedKey, waiting(evictedRow));
      if (evictedKey === key) return [];
      rows.push(stampRow({}, evictedRow.rowRef, 'DELETE'));
    }
    return rows;
  }

  function pushChange(resourceName: string, operation: RowOperation, row: Row): void {
    const query = config.queries?.[resourceName];
    // The query's own filter (where-clause equivalent) applies to live pushes
    // too — a row the snapshot would exclude must not leak in via broadcast.
    // A MODIFY that moves a row OUT of the filter is converted to a DELETE
    // rather than dropped: a swallowed removal leaves the client rendering the
    // row with its stale pre-modify values until a full reload. An INSERT
    // outside the filter is simply dropped (the client never held it).
    let effectiveOperation = operation;
    if (operation !== 'DELETE' && query?.filter && !query.filter(row)) {
      if (operation !== 'MODIFY') return;
      effectiveOperation = 'DELETE';
    }
    for (const connection of connections) {
      for (const [sourceRef, subscription] of connection.subscriptions) {
        if (subscription.resourceName !== resourceName) continue;
        const resolved = resolveSubscriptionOperation(
          subscription,
          sourceRef,
          effectiveOperation,
          row,
        );
        if (resolved === null) continue;
        // Per-user permissioning (auth maps / hideFields equivalent): the
        // ROW_REF is computed from the unstripped row, since pk fields may
        // themselves be hidden from this user.
        const rowRef = subscription.getRowRef(row);
        const legacy = isLegacyFidelity(config);
        let rows: Row[];
        if (subscription.held) {
          rows = viewUpdate(subscription, connection, query, resolved, row, rowRef);
        } else if (resolved === 'DELETE') {
          // GSF's DELETE row is DETAILS {OPERATION, ROW_REF} and nothing else.
          // Legacy mode sent the pk fields: the stub it was given for a real
          // DELETE, or one rebuilt from the key for a change that left the
          // subscription (the full row may carry columns hideFields strips).
          let outRow: Row = {};
          if (legacy) {
            outRow =
              operation !== 'DELETE' && query
                ? store.rowStubFromKey(query.source, rowRef as string)
                : row;
          }
          rows = [stampRow(outRow, rowRef, resolved, legacy)];
        } else {
          // Only the subscription's columns, minus the ones this user can't
          // see; always the whole row.
          const permitted = permittedQueryRow(
            row,
            query?.auth,
            requestingUser(connection),
            subscription.fieldNames,
          );
          rows = permitted
            ? [stampRow(wireRow(subscription, permitted, row), rowRef, resolved, legacy)]
            : [];
        }
        if (rows.length === 0) continue;
        subscription.sequenceId += 1;
        // A live push carries neither MORE_ROWS nor ROWS_COUNT on the real
        // server — those describe the snapshot. Legacy mode sent ROWS_COUNT: 1.
        connection.send({
          MESSAGE_TYPE: 'QUERY_UPDATE',
          SOURCE_REF: sourceRef,
          ROW: rows,
          ...(legacy ? { ROWS_COUNT: 1 } : {}),
          SEQUENCE_ID: subscription.sequenceId,
        });
      }
    }
  }

  // Pushes an unsolicited row update to every subscription on the resource
  // (see pushChange). A DELETE given as a pk-only row is resolved to the
  // record(s) removed from that key that no broadcast has announced yet —
  // their RECORD_IDs are the ROW_REFs a client may hold — and those are
  // consumed, so a later delete of a re-inserted key names its own record.
  const broadcast: BroadcastFn = (resourceName, operation, row) => {
    const source = config.queries?.[resourceName]?.source;
    if (operation !== 'DELETE' || !source) {
      pushChange(resourceName, operation, row);
      return;
    }
    const removed = store.takeRemovedRecordsFor(source, row);
    if (isLegacyFidelity(config)) {
      pushChange(resourceName, operation, row);
      return;
    }
    const records = row.RECORD_ID === undefined || row.RECORD_ID === null ? removed : [row];
    if (records.length === 0) records.push(row);
    for (const record of records) pushChange(resourceName, 'DELETE', record);
  };

  // A change being broadcast to a table, as the views that join it see it:
  // the table's rows before and after it, and the RECORD_IDs it touched (the
  // changed record and every record its key lost). The store's
  // previous-record entries are consumed straight away, so the next change
  // starts from this one; the rows are only gathered if a view needs them.
  function tableChange(
    tableName: string,
    operation: RowOperation,
    current: Row | undefined,
    removed: Row[],
  ): TableChange {
    const currentId = recordKey(current);
    const previous = current ? store.takePreviousRecord(tableName, current) : undefined;
    const lost = new Map(
      removed.map((record) => [
        recordKey(record),
        store.takePreviousRecord(tableName, record) ?? record,
      ]),
    );
    // The changed record was in the table before as it was before its first
    // modify — or as it is, when nothing modified it. An INSERT's wasn't, nor
    // was one that replaced another record under its key.
    const currentWasThere = operation !== 'INSERT' && (previous !== undefined || lost.size === 0);
    const records = new Set(
      [currentId, ...lost.keys()].filter((id): id is string => id !== undefined),
    );
    let before: Row[] | undefined;
    return {
      records,
      before: () => {
        if (before) return before;
        const restored = new Set<string | undefined>();
        before = store.getAllRows(tableName).flatMap((row) => {
          const id = recordKey(row);
          if (id !== undefined && id === currentId) {
            return currentWasThere ? [previous ?? row] : [];
          }
          // A record a DELETE names before it is removed.
          if (lost.has(id)) {
            restored.add(id);
            return [lost.get(id)!];
          }
          return [row];
        });
        for (const [id, record] of lost) if (!restored.has(id)) before.push(record);
        return before;
      },
      // Without the records the key lost, even one still stored when its
      // DELETE is broadcast.
      after: () => store.getAllRows(tableName).filter((row) => !lost.has(recordKey(row))),
    };
  }

  // A live change to a table a view joins (at any depth) or anti-joins, under
  // fidelity 'gsf'. Only the joins that read the table with `backwardsJoin`
  // listen to it — GSF listens to no other joined table — and every
  // anti-join does. The view is rebuilt through them as it was before the
  // change and as it is after it, and compared, row by base record (the
  // ROW_REF): a row the change brought in (an INNER join that now matches) is
  // pushed as an INSERT, one it took out as a DELETE, and one joined to a
  // changed record before or after, or whose values changed, as a MODIFY.
  // A row joined to the changed record is pushed even when its values look
  // the same: the before-state is the store's (a record as it was before its
  // first unannounced modify), and a client may hold newer values than that.
  // A view that joins its own base table gets the changed base rows
  // themselves from broadcastTableChange's base path, not from here.
  function joinedViewChanges(
    viewName: string,
    view: ViewDef,
    tableName: string,
    change: TableChange,
  ): ViewChange[] {
    const plan = planView(viewName, view);
    const listening = entriesReading(plan, tableName).filter(
      (entry) => !('backwardsJoin' in entry) || entry.backwardsJoin,
    );
    if (listening.length === 0) return [];
    // By definition: every materialization plans the view afresh.
    const listeningDefs = new Set<object>(listening.map((entry) => entry.def));
    const listeningJoins = plan.joins.flatMap((join, index) =>
      listeningDefs.has(join.def) ? [index] : [],
    );
    const touched = ({ matches }: ViewEntry) =>
      listeningJoins.some((index) => {
        const id = recordKey(matches[index]);
        return id !== undefined && change.records.has(id);
      });
    const rowsOf = (rows: () => Row[]) => (entry: PlannedJoin | PlannedAntiJoin) =>
      listeningDefs.has(entry.def) ? rows() : undefined;
    const was = store.getViewEntriesFrom(viewName, { rowsOf: rowsOf(change.before) });
    const now = store.getViewEntriesFrom(viewName, { rowsOf: rowsOf(change.after) });
    const ownRow = (entry: ViewEntry) =>
      view.base === tableName && change.records.has(recordKey(entry.row) ?? '');
    const wasByRecord = new Map(was.map((entry) => [recordKey(entry.row), entry]));
    const nowRecords = new Set(now.map((entry) => recordKey(entry.row)));
    const changes: ViewChange[] = [];
    for (const entry of now) {
      if (ownRow(entry)) continue;
      const old = wasByRecord.get(recordKey(entry.row));
      if (!old) changes.push(['INSERT', entry.row]);
      else if (touched(old) || touched(entry) || !sameValues(old.row, entry.row)) {
        changes.push(['MODIFY', entry.row]);
      }
    }
    for (const entry of was) {
      if (!ownRow(entry) && !nowRecords.has(recordKey(entry.row))) {
        changes.push(['DELETE', entry.row]);
      }
    }
    return changes;
  }

  // Event handlers mutate one table (e.g. TRADE) but grids may be subscribed to
  // a view built on top of it (e.g. ALL_TRADES -> TRADE_VIEW). This finds every
  // query resource backed by that table — directly or via a view — and pushes
  // the (re-joined, for views) row to each one's subscribers. `pk` is the
  // store key: the single pk field's value, or store.rowKey(...) for
  // composite-keyed tables. ROW_REF is the record's RECORD_ID, so a DELETE is
  // broadcast with every record the key lost that no broadcast announced yet
  // (Store.takeRemovedRows; held-rows tracking drops the ones a client
  // doesn't hold), and a record replaced under the same key (deleted, then
  // re-inserted, before one broadcast) gets a DELETE of its old ROW_REF before
  // the new one goes out. A view row is built from the changed base row
  // alone, so it is found whatever the view's own pkField; a MODIFY that takes
  // it out of the view (an INNER join or an anti-join that no longer lets it
  // through) is pushed as a DELETE, and an INSERT that doesn't make it into
  // the view isn't pushed at all. A table a view joins is handled by
  // joinedViewChanges. Legacy ROW_REFs are the key itself, so there a DELETE
  // is the pk stub and a replacement is the one INSERT/MODIFY, as before; and
  // a change to a joined table pushes every view row as a MODIFY.
  const broadcastTableChange: BroadcastTableChangeFn = (tableName, operation, pk) => {
    const legacy = isLegacyFidelity(config);
    const current = operation === 'DELETE' ? undefined : store.getRow(tableName, pk);
    let removed = store
      .takeRemovedRows(tableName, pk)
      .filter((record) => record.RECORD_ID !== current?.RECORD_ID);
    // A DELETE broadcast before the row is removed still names it.
    const live = store.getRow(tableName, pk);
    if (operation === 'DELETE' && removed.length === 0 && live) removed = [live];
    const change = tableChange(tableName, operation, current, removed);
    if (legacy) removed = [];
    const viewChanges = new Map<string, ViewChange[]>();
    for (const [queryName, query] of Object.entries(config.queries ?? {})) {
      const isDirect = query.source === tableName;
      const view = store.views.get(query.source);
      const isViewOnTable = view?.base === tableName;

      if (view && !legacy) {
        let changes = viewChanges.get(query.source);
        if (!changes) {
          changes = joinedViewChanges(query.source, view, tableName, change);
          viewChanges.set(query.source, changes);
        }
        for (const [rowOperation, row] of changes) pushChange(queryName, rowOperation, row);
      }

      // Legacy: a view that JOINS (or anti-joins) the mutated table
      // denormalizes its columns into any number of view rows, so the view is
      // re-materialized and every row pushed as a MODIFY.
      if (
        legacy &&
        !isDirect &&
        !isViewOnTable &&
        ((view?.joins ?? []).some((join) => join.table === tableName) ||
          (view?.antiJoins ?? []).some((antiJoin) => antiJoin.table === tableName))
      ) {
        for (const viewRow of store.getViewRows(query.source)) {
          pushChange(queryName, 'MODIFY', viewRow);
        }
        continue;
      }
      if (!isDirect && !isViewOnTable) continue;

      // The removed base records carry the RECORD_ID a view row's ROW_REF
      // comes from too; the DELETE body is DETAILS only either way.
      for (const record of removed) pushChange(queryName, 'DELETE', record);
      if (operation === 'DELETE') {
        if (legacy) {
          const stub = store.rowStubFromKey(query.source, pk);
          if (Object.keys(stub).length > 0) pushChange(queryName, 'DELETE', stub);
        }
        continue;
      }
      if (!current) continue;
      if (isDirect) {
        pushChange(queryName, operation, current);
        continue;
      }
      // Only the base read is narrowed: a view that also joins its base table
      // still joins every row of it.
      const viewRow = store.getViewRowsFrom(query.source, { base: [current] })[0];
      if (viewRow) {
        pushChange(queryName, operation, viewRow);
      } else if (operation === 'MODIFY') {
        pushChange(
          queryName,
          'DELETE',
          legacy
            ? store.rowStubFromKey(query.source, store.getRowRef(query.source, current) as string)
            : current,
        );
      }
    }
  };

  // fidelity: 'legacy' HTTP sessions, MOCK_SESSION_ID -> username (handlers/http.ts).
  const legacyHttpSessions = new Map<string, string>();
  // Sessions and tokens issued over either transport (handlers/auth.ts), so a
  // WS connection can authenticate with a token from an earlier HTTP login
  // (foundation-comms sends the stored SESSION_AUTH_TOKEN on every message, and
  // its token refresh sends EVENT_LOGIN_AUTH with only
  // DETAILS.REFRESH_AUTH_TOKEN, no credentials).
  const auth = createAuthState();

  // Runtime override/extend API. Every handler reads off this same `config`
  // object at call time (both the WS path in routeMessage below and the HTTP
  // login transport, whose handler was created once in start() and holds a
  // reference to this exact object) — so these mutate config's own
  // properties in place rather than rebinding the `config` variable, which
  // would only be visible to callers still reading the original reference.
  // Useful both for composing a project's config from a shared base
  // (`setEventHandler`/`setQuery` after createMockServer) and for tests that
  // want to inject a one-off failure/edge case (`extendEventHandler`) without
  // spinning up a second server.
  function setTable(tableName: string, tableDef: TableDef): void {
    checkNotDeclared(config.tables ?? {}, tableName, tableDef);
    const tables = { ...config.tables, [tableName]: tableDef };
    // Nor may it take over a table declared under its audit table's name.
    if (isAudited(tableDef)) {
      const auditName = auditTableName(tableName);
      checkNotDeclared(tables, auditName, config.tables?.[auditName]);
    }
    config.tables = tables;
    store.registerTable(tableName, tableDef);
    // A new definition makes a new audit table too, empty as the table's
    // own rows are its new ones.
    if (isAudited(tableDef)) {
      setTable(auditTableName(tableName), auditTableDef(tableName, tableDef));
    }
  }

  function setView(viewName: string, viewDef: ViewDef): void {
    config.views = { ...config.views, [viewName]: viewDef };
    store.registerView(viewName, viewDef);
  }

  function setQuery(resourceName: string, queryDef: QueryDef): void {
    config.queries = { ...config.queries, [resourceName]: queryDef };
  }

  function setRequestReply(resourceName: string, requestReplyDef: RequestReplyDef): void {
    config.requestReplies = { ...config.requestReplies, [resourceName]: requestReplyDef };
  }

  function setEventHandler(eventName: string, handler: EventHandler): void {
    config.eventHandlers = { ...config.eventHandlers, [eventName]: handler };
  }

  function getQuery(resourceName: string): QueryDef | undefined {
    return config.queries?.[resourceName];
  }

  function getRequestReply(resourceName: string): RequestReplyDef | undefined {
    return config.requestReplies?.[resourceName];
  }

  // An eventHandlers entry, else the generic CRUD handler a table registers
  // (TableDef.events) — so extendEventHandler can wrap a generic event; the
  // wrapper is then an explicit handler, which wins.
  function getEventHandler(eventName: string): EventHandler | undefined {
    return config.eventHandlers?.[eventName] ?? crudEventHandler(eventName, config);
  }

  // augment(current) => replacement. `current` is undefined if nothing was
  // registered under that name yet, so augment can also be used to define a
  // brand new query/handler in terms of "there was nothing before".
  function extendQuery(
    resourceName: string,
    augment: (current: QueryDef | undefined) => QueryDef,
  ): void {
    setQuery(resourceName, augment(getQuery(resourceName)));
  }

  function extendRequestReply(
    resourceName: string,
    augment: (current: RequestReplyDef | undefined) => RequestReplyDef,
  ): void {
    setRequestReply(resourceName, augment(getRequestReply(resourceName)));
  }

  function extendEventHandler(
    eventName: string,
    augment: (current: EventHandler | undefined) => EventHandler,
  ): void {
    setEventHandler(eventName, augment(getEventHandler(eventName)));
  }

  // An event POSTed over HTTP (HttpRouteCtx.commitEvent): run through the
  // socket's commit path on a connection of its own, whose one reply is the
  // answer.
  function commitEventOverHttp(message: GenesisMessage, userName: string | undefined): Row {
    let reply: Row = {};
    const connection = new Connection({
      readyState: WS_OPEN,
      send: (data) => {
        reply = JSON.parse(data);
      },
    });
    connection.authenticated = userName !== undefined;
    connection.userName = userName;
    if (config.auth?.requireAuth && !connection.authenticated) {
      replyNotLoggedOn(message, connection);
    } else {
      handleCommitEvent(message, connection, config, store, { broadcast, broadcastTableChange });
    }
    return reply;
  }

  let wss: WebSocketServer | undefined;
  let httpServer: http.Server | undefined;

  function start(port: number): WebSocketServer {
    // WS (datasource/commit protocol) and HTTP (foundation-auth's login
    // transport) share one http.Server, matching how the real Genesis gateway
    // and the vite dev proxy present a single origin to the client. Built on
    // node:http + the `ws` package rather than a runtime-specific API (e.g.
    // Bun.serve) so this runs unmodified under both Node (>=22.6, via
    // `node --experimental-strip-types`) and Bun.
    httpServer = http.createServer(
      createHttpRequestListener(config, legacyHttpSessions, auth, {
        store,
        broadcast,
        broadcastTableChange,
        commitEvent: commitEventOverHttp,
      }),
    );
    // A cookie-authenticated router answers an upgrade without a live
    // sessionAuthToken cookie with HTTP 401 (PreHttpHandlerProvider.kt:61-95
    // @v8.15.29). Read per upgrade, like every config field.
    const cookieToken = (req: http.IncomingMessage) =>
      parseCookies(req.headers.cookie)[SESSION_AUTH_TOKEN_COOKIE];
    wss = new WebSocketServer({
      server: httpServer,
      verifyClient: ({ req }, done) => {
        if (!cookieAuthentication(config)) return done(true);
        const token = cookieToken(req);
        done(token !== undefined && auth.sessions.has(token), HTTP_UNAUTHORIZED);
      },
    });

    wss.on('connection', (ws: WebSocket, req: http.IncomingMessage) => {
      const connection = new Connection(ws);
      if (cookieAuthentication(config)) connection.cookieSessionAuthToken = cookieToken(req);
      const hostname = req.headers.hostname;
      connection.hostnameHeader = typeof hostname === 'string' ? hostname : undefined;
      connection.remoteAddress = req.socket.remoteAddress;
      connections.add(connection);

      ws.on('message', (data) => {
        let message;
        try {
          message = JSON.parse(data.toString());
        } catch {
          console.warn('[mock-server] Ignoring non-JSON message');
          return;
        }
        try {
          // Inbound hook: lets a project rewrite messages before routing
          // (e.g. normalize a CRITERIA_MATCH dialect) or handle them fully
          // itself (return false) — instead of monkey-patching ws.on.
          if (config.onMessage) {
            const hooked = config.onMessage(message, {
              connection,
              store,
              broadcast,
              broadcastTableChange,
            });
            if (hooked === false) return;
            if (hooked) message = hooked;
          }
          routeMessage(
            message,
            connection,
            config,
            store,
            { broadcast, broadcastTableChange },
            auth,
          );
        } catch (error) {
          console.error(`[mock-server] Error handling ${message.MESSAGE_TYPE}:`, error);
        }
      });

      ws.on('close', () => {
        connections.delete(connection);
      });
    });

    httpServer.listen(port);
    // Log the ACTUAL bound port once listening — with port 0 (ephemeral, the
    // parallel-servers pattern) the requested port would read "0".
    httpServer.once('listening', () => {
      console.log(
        `[mock-server] Listening on ws://localhost:${boundPort()} (HTTP login endpoints on the same port)`,
      );
    });
    return wss;
  }

  function boundPort(): number | undefined {
    const address = httpServer?.address();
    return typeof address === 'object' && address ? address.port : undefined;
  }

  // start() + resolve with the OS-assigned port once actually listening.
  // Pass no argument (port 0) to run any number of servers side by side —
  // each instance's store/connections/sessions are fully isolated already,
  // so parallel test workers just call listen() and hand the returned port
  // to their own client (API_HOST=ws://localhost:<port>).
  function listen(port = 0): Promise<number> {
    return new Promise((resolve, reject) => {
      if (httpServer?.listening) {
        resolve(boundPort()!);
        return;
      }
      if (!httpServer) start(port);
      httpServer!.once('error', reject);
      httpServer!.once('listening', () => resolve(boundPort()!));
    });
  }

  function stop(): Promise<void> {
    return new Promise((resolve) => {
      // Release the handles up front: a closed http.Server can never emit
      // 'listening' again, so leaving them set makes a later listen() wait
      // forever instead of starting a fresh server (restart-between-tests
      // harnesses deadlock with no error).
      const socketServer = wss;
      const server = httpServer;
      wss = undefined;
      httpServer = undefined;
      if (!socketServer || !server) {
        resolve();
        return;
      }
      // wss.close() on a {server}-attached WebSocketServer does NOT close live
      // client sockets, and httpServer.close() waits for every open connection
      // — so with a browser still attached, a plain close chain never resolves
      // (and the showcase's SIGINT handler would hang forever). Terminate WS
      // clients and drop idle keep-alive HTTP sockets explicitly first.
      for (const client of socketServer.clients) client.terminate();
      socketServer.close(() => {
        server.close(() => resolve());
        // Node >=18.2; optional-called because Bun's node:http may lack it.
        server.closeAllConnections?.();
      });
    });
  }

  return {
    start,
    listen,
    stop,
    store,
    broadcast,
    broadcastTableChange,
    setTable,
    setView,
    setQuery,
    setRequestReply,
    setEventHandler,
    getQuery,
    getRequestReply,
    getEventHandler,
    extendQuery,
    extendRequestReply,
    extendEventHandler,
  };
}