API index and search · Build metadata

Source snapshot

packages/client/src/view.ts

1// The JS ArrayView — the materialized, typed result tree a backend's ChangeEvent stream2// folds into (WASM-CLIENT-DESIGN.md §7, §8). A faithful port of the View's `apply_change`3// (src/view.rs / src/flat_receiver.rs), the §6 receiver contract of FLAT-CHANGES-DESIGN.md:4// descend the path with the in-view gate, then Add / Remove / Edit (rc + the edit-move5// ghost/adjusted-position/merge protocol) located by value-driven binary search.6//7// Reads are cheap and reference-stable: each node memoizes its projected output (`out`),8// invalidated only along the root→mutation spine, so `.data` re-projects only what changed9// and untouched subtrees keep their object identity (React/Solid memoize on it).1011import { compareRows } from "./compare.ts";12import type { ColType, FlatChange, FlatOp, QueryId, ResultType, WireNode, WireSchema, WireValue } from "./types.ts";1314/** Per-level column types (parallel to the WireSchema), used to JSON.parse json columns on15 *  projection. Built by the Store from the typed schema; absent ⇒ no parsing (bare values). */16export interface ViewTypes {17  columnTypes: ColType[];18  rels: Record<number, ViewTypes>;19}2021/** The phase of an {@link ArrayView.onChanges} delivery: the initial hydrate `snapshot` vs a later22 *  incremental `batch` — the same distinction a narrator draws (`ChangeEvent` `snapshot`/`batch`). */23export type ChangePhase = "snapshot" | "batch";2425/** A per-view change listener ({@link ArrayView.onChanges}): the net `FlatChange[]` this view folded,26 *  the {@link ChangePhase} it arrived on, and the view's `WireSchema` (the position→name source for27 *  `resolveChange`). This is the DIFF the data channel ({@link ArrayView.subscribe}) discards — the28 *  seam a narrator drives off. The `schema` is passed (not closed over) because the first `snapshot`29 *  fires synchronously inside `materialize`, before the caller holds the view handle. */30export type ViewChangeListener = (changes: FlatChange[], phase: ChangePhase, schema: WireSchema) => void;3132/** The public ArrayView contract `materialize()` returns. */33export interface ArrayView<R> {34  /** The current materialized result (reference-stable where data is unchanged). */35  readonly data: readonly R[];36  /** The engine query id the Store assigned this view (1:1 with the view). The same `qid` the raw37   *  change stream ({@link Store.subscribeChanges}) tags each frame with, so a consumer can38   *  correlate this query with its `ChangeEvent`s (e.g. to bind a narrator) straight off39   *  `materialize(query).qid` — no separate handle needed. */40  readonly qid: QueryId;41  /** The query's view `WireSchema` (the position→name source for {@link resolveChange}), captured42   *  from its `hello` frame. `null` while PENDING — a remote backend's `hello` arrives async; an43   *  in-process backend (wasm/replica) populates it synchronously during `materialize`. */44  readonly schema: WireSchema | null;45  /** The query's SERVER-CHANNEL state (`unknown` while loading, `complete` once server-authoritative;46   *  the `error` variant is reserved and currently unproduced). A pending optimistic mutation no47   *  longer moves this — it is a separate axis (FOLDED-MUTATIONS-DESIGN §7). `complete` for backends48   *  with no server lifecycle. Changes notify subscribers. */49  readonly resultType: ResultType;50  /** Subscribe; fires immediately with the current data, then after each applied batch (and after51   *  a {@link resultType} change — re-read `resultType` in the listener). */52  subscribe(listener: (data: readonly R[]) => void): () => void;53  /** Subscribe to this view's folded CHANGE stream — the `FlatChange[]` it applies, NET of no-op54   *  cycles (a rebase's balanced `remove`+`add` / edit round-trip cancels, so a correctly predicted55   *  optimistic write delivers nothing here). Carries the diff {@link subscribe} throws away; the56   *  per-view seam a narrator rides. Does NOT replay on subscribe — attach via57   *  `store.materialize(query, { onChanges })` to catch a synchronous backend's first snapshot.58   *  A view with any change listener also enriches its own `remove` ops with the evicted subtree59   *  (per-view, no global opt-in). Returns a detach function. */60  onChanges(listener: ViewChangeListener): () => void;61  /** Tear down + stop receiving updates. */62  destroy(): void;63}6465/** What a top-level `.one()` query materializes to: the single row (or `null`), not an array.66 *  A thin adapter over a {@link FlatArrayView} — all the folding is shared; only the result67 *  boundary unwraps (`data[0] ?? null`). Reference identity of the row is preserved. */68export interface SingularArrayView<R> {69  /** The single current row, or `null` when the query matches nothing. */70  readonly data: R | null;71  /** The engine query id — see {@link ArrayView.qid}. */72  readonly qid: QueryId;73  /** The query's view `WireSchema` — see {@link ArrayView.schema}. */74  readonly schema: WireSchema | null;75  /** The query's lifecycle state — see {@link ArrayView.resultType}. */76  readonly resultType: ResultType;77  /** Subscribe; fires immediately with the current row, then after each applied batch (and after78   *  a {@link resultType} change). */79  subscribe(listener: (data: R | null) => void): () => void;80  /** Subscribe to the view's folded CHANGE stream — see {@link ArrayView.onChanges}. The changes81   *  are the same positional `FlatChange`s as the plural view (a `.one()` is just the list capped to82   *  one), so a narrator resolves them identically. */83  onChanges(listener: ViewChangeListener): () => void;84  /** Tear down + stop receiving updates. */85  destroy(): void;86}8788/** Wrap a plural {@link FlatArrayView} as a {@link SingularArrayView} for a `.one()` query89 *  (the engine caps it to `limit = 1`, so the top list holds at most one node). */90export class SingularView<R> implements SingularArrayView<R> {91  private readonly inner: ArrayView<R>;9293  constructor(inner: ArrayView<R>) {94    this.inner = inner;95  }9697  get data(): R | null {98    return this.inner.data[0] ?? null;99  }100101  get qid(): QueryId {102    return this.inner.qid;103  }104105  get schema(): WireSchema | null {106    return this.inner.schema;107  }108109  get resultType(): ResultType {110    return this.inner.resultType;111  }112113  subscribe(listener: (data: R | null) => void): () => void {114    return this.inner.subscribe((d) => listener(d[0] ?? null));115  }116117  onChanges(listener: ViewChangeListener): () => void {118    return this.inner.onChanges(listener);119  }120121  destroy(): void {122    this.inner.destroy();123  }124}125126/** An internal reconstruction node: the row, multi-path refcount, per-slot child lists,127 *  and a memoized projected output (`null` ⇒ stale, rebuilt on next read). */128interface Node {129  row: WireValue[];130  rc: number;131  rels: Node[][];132  out: unknown;133}134135function binarySearch(136  list: Node[],137  row: WireValue[],138  sort: [number, boolean][],139): { found: boolean; index: number } {140  let lo = 0;141  let hi = list.length;142  while (lo < hi) {143    const mid = (lo + hi) >>> 1;144    const c = compareRows(list[mid].row, row, sort);145    if (c < 0) lo = mid + 1;146    else if (c > 0) hi = mid;147    else return { found: true, index: mid };148  }149  return { found: false, index: lo };150}151152/** Build a fresh node (rc = 1) with its inline subtree, mirroring `init_rels_for_new_entry`:153 *  one list per declared slot; fill in-view slots from the payload (children inserted in the154 *  order shipped — NEVER re-sorted — folding same-sort-key dups to rc += 1). */155function buildNode(wnode: WireNode, schema: WireSchema): Node {156  const rels: Node[][] = schema.relationships.map(() => []);157  for (const { rel, children } of wnode.rels) {158    const child = schema.relationships[rel]?.child;159    if (!child) continue; // join-only / gating slot — not materialized160    const built: Node[] = [];161    for (const c of children) {162      const at = binarySearch(built, c.row, child.sort);163      if (at.found) built[at.index].rc += 1;164      else built.splice(at.index, 0, buildNode(c, child));165    }166    rels[rel] = built;167  }168  return { row: wnode.row, rc: 1, rels, out: null };169}170171function cloneNode(n: Node): Node {172  return { row: n.row.slice(), rc: n.rc, rels: n.rels.map((r) => r.map(cloneNode)), out: null };173}174175/** Reconstruct a positional {@link WireNode} (row + its populated child slots) from a maintained176 *  {@link Node} — the inverse of {@link buildNode}. Used to attach a removed subtree to a `remove`177 *  op so a change consumer (a narrator) can resolve its nested subs, which a bare positional remove178 *  (just the leaving row) cannot carry. Only in-view (`child !== null`), non-empty slots are emitted,179 *  matching how an `add` node ships only populated relationships. */180function toWireNode(node: Node, schema: WireSchema): WireNode {181  const rels: { rel: number; children: WireNode[] }[] = [];182  for (const rel of schema.relationships) {183    if (rel.child === null) continue;184    const childList = node.rels[rel.slot] ?? [];185    if (childList.length === 0) continue;186    rels.push({ rel: rel.slot, children: childList.map((c) => toWireNode(c, rel.child as WireSchema)) });187  }188  return { row: node.row, rels };189}190191/** Elementwise equality over positional bare-cell rows — THE row comparator for every JS-side192 *  diff (views, aggregate heads, the persistence mirror), exported so no caller grows a drifted193 *  private copy. Per cell: `===` keeps `-0 === 0` (the engine's key semantics treat them equal);194 *  the `Object.is` arm makes a NaN cell equal itself (under `!==` alone, a NaN-bearing row never195 *  matches any copy of itself, so every diff re-emits it as a spurious edit forever). */196export function rowsEqual(a: WireValue[], b: WireValue[]): boolean {197  if (a.length !== b.length) return false;198  for (let i = 0; i < a.length; i++) {199    if (a[i] !== b[i] && !Object.is(a[i], b[i])) return false;200  }201  return true;202}203204/** JSON.stringify for keying a row, except the numbers JSON folds together — NaN/±Infinity (→ null)205 *  and -0 (→ 0) — get a sentinel encoding, so two rows that `rowsEqual` (Object.is) distinguishes206 *  never share a net key (a `remove [1, Infinity]` must not cancel an `add [1, null]`). */207function keyRow(row: WireValue[]): string {208  return JSON.stringify(row, (_k, v: unknown) =>209    typeof v === "number" && (!Number.isFinite(v) || Object.is(v, -0)) ? [" num", Object.is(v, -0) ? "-0" : String(v)] : v,210  );211}212213/** The key a subtractive net matches an op on: the path (parent locators, one hop per level) plus the214 *  row bytes. Two ops with the same key touch the same row at the same tree position. */215function opRowKey(path: FlatChange["path"], row: WireValue[]): string {216  let k = "";217  for (const seg of path) k += seg.rel + ":" + keyRow(seg.parentRow) + "/";218  return k + "|" + keyRow(row);219}220221/** Walk `path` down `schema` to the changed level's `WireSchema` — `null` when a hop crosses a222 *  gating (join-only) slot; such ops never fold, so they never reach the net. */223function levelSchemaAt(schema: WireSchema, path: FlatChange["path"]): WireSchema | null {224  let level: WireSchema = schema;225  for (const seg of path) {226    const child = level.relationships[seg.rel]?.child;227    if (!child) return null;228    level = child;229  }230  return level;231}232233/** Deep structural equality of two canonical (maintained-order) subtrees: rows via Object.is,234 *  per-slot children pairwise, rc included — a dup-path child only equals an equally-duplicated one. */235function nodesEqual(a: Node, b: Node): boolean {236  if (a.rc !== b.rc || !rowsEqual(a.row, b.row) || a.rels.length !== b.rels.length) return false;237  for (let s = 0; s < a.rels.length; s++) {238    const ra = a.rels[s];239    const rb = b.rels[s];240    if (ra.length !== rb.length) return false;241    for (let i = 0; i < ra.length; i++) if (!nodesEqual(ra[i], rb[i])) return false;242  }243  return true;244}245246/** Cancel no-op cycles in a folded batch SUBTRACTIVELY — the safe half of the removed engine247 *  coalescer (OPTIMISTIC-WRITES-DESIGN §3, and `wasm/db.rs` "Step 2 of the removal"). A rebase248 *  re-invokes a still-pending prediction, emitting a balanced `remove R`+`add R` (or `edit b→a`+249 *  `edit a→b`) that nets to nothing; so a correctly predicted optimistic write delivers ZERO changes250 *  to a narrator, and only genuine (mispredicted / other-client) changes survive. We ONLY drop exact251 *  inverse pairs — byte-identical row at the same path AND a structurally identical subtree: a row252 *  that leaves and re-enters within one batch carrying a CHANGED child set is NOT an inverse (the253 *  child never got its own event while the row was out — the re-add is its only carrier), so that254 *  pair survives. Survivors pass through UNMODIFIED: we never rewrite a locator, reorder, or255 *  reconstruct a gated subtree (the reconstructive netting that shipped four correctness bugs and256 *  was pulled from the engine). Because this feeds prose, not a locate-by-value receiver, even an257 *  imperfect net is at worst a spurious line, never a corrupt view — and every conservative bail258 *  here (unknown subtree, gated level) errs on the spurious-line side, never the missed-line side.259 *  Order-preserving for survivors; returns the input array unchanged when nothing cancels. */260function netChanges(changes: FlatChange[], schema: WireSchema): FlatChange[] {261  if (changes.length < 2) return changes;262  const alive = new Array<boolean>(changes.length).fill(true);263  const removes = new Map<string, number[]>(); // key → indices of not-yet-cancelled removes264  const adds = new Map<string, number[]>(); // key → indices of not-yet-cancelled adds265  const edits = new Map<string, number[]>(); // pathKey+old+new → indices of not-yet-cancelled edits266  // The subtree an add/remove carries, canonicalized to maintained (sorted, dup-folded) form so a267  // shipped-order `add` node and a maintained-order enriched `remove` node compare structurally.268  // Built ONLY when two ops collide on the row key — the common no-collision delivery never builds269  // one — and memoized per op index. `null` ⇒ subtree unknown (a gated level, or a remove that was270  // not enriched); conservative: an unknown subtree never cancels, so we err toward a spurious line.271  const subCache = new Map<number, Node | null>();272  const subtreeOf = (i: number): Node | null => {273    let s = subCache.get(i);274    if (s !== undefined) return s;275    const { path, op } = changes[i];276    const level = levelSchemaAt(schema, path);277    // A remove reaching the net actually evicted its node (only applied ops are delivered) and so was278    // enriched with its subtree on `op.node`; an add always carries `op.node`. Either is canonicalized.279    const wnode = op.tag === "add" ? op.node : op.tag === "remove" ? op.node : undefined;280    s = level && wnode ? buildNode(wnode, level) : null;281    subCache.set(i, s);282    return s;283  };284  // Take the LATEST pending opposite at the same key whose subtree also matches (LIFO, like the285  // plain `take` below, so nested cycles unwind innermost-first).286  const takeInverse = (m: Map<string, number[]>, k: string, i: number): number | undefined => {287    const arr = m.get(k);288    if (!arr || arr.length === 0) return undefined;289    const mine = subtreeOf(i);290    if (mine === null) return undefined;291    for (let x = arr.length - 1; x >= 0; x--) {292      const theirs = subtreeOf(arr[x]);293      if (theirs !== null && nodesEqual(mine, theirs)) {294        const j = arr[x];295        arr.splice(x, 1);296        return j;297      }298    }299    return undefined;300  };301  const take = (m: Map<string, number[]>, k: string): number | undefined => {302    const arr = m.get(k);303    return arr && arr.length ? arr.pop() : undefined;304  };305  const put = (m: Map<string, number[]>, k: string, i: number): void => {306    const arr = m.get(k);307    if (arr) arr.push(i);308    else m.set(k, [i]);309  };310  let cancelled = false;311  for (let i = 0; i < changes.length; i++) {312    const { path, op } = changes[i];313    if (op.tag === "remove") {314      const k = opRowKey(path, op.row);315      const j = takeInverse(adds, k, i);316      if (j !== undefined) {317        alive[i] = alive[j] = false;318        cancelled = true;319      } else put(removes, k, i);320    } else if (op.tag === "add") {321      const k = opRowKey(path, op.node.row);322      const j = takeInverse(removes, k, i);323      if (j !== undefined) {324        alive[i] = alive[j] = false;325        cancelled = true;326      } else put(adds, k, i);327    } else {328      // edit: cancel against a prior INVERSE edit (a→b then b→a) at the same path. The old/new329      // bytes are part of the map key, so the inverse lookup is O(1) instead of a rescan.330      const pk = opRowKey(path, []);331      const oldS = keyRow(op.old);332      const newS = keyRow(op.new);333      const j = take(edits, pk + "" + newS + "" + oldS);334      if (j !== undefined) {335        alive[i] = alive[j] = false;336        cancelled = true;337      } else put(edits, pk + "" + oldS + "" + newS, i);338    }339  }340  if (!cancelled) return changes;341  const out: FlatChange[] = [];342  for (let i = 0; i < changes.length; i++) if (alive[i]) out.push(changes[i]);343  return out;344}345346const EMPTY: readonly never[] = Object.freeze([]);347348export class FlatArrayView<R = unknown> implements ArrayView<R> {349  // The engine query id (Store-assigned, 1:1 with this view), exposed read-only via {@link qid}.350  // `0` ⇒ unbound (a bare view never registered with a Store); the Store passes it at construction.351  private readonly _qid: QueryId;352  // `null` ⇒ PENDING (no schema yet): a remote backend's `hello` arrives async, so the view353  // exists (and reads as `[]`) before its schema lands. Set by `reset` (first hello / re-hydrate).354  // Exposed read-only via the {@link schema} getter (the position→name source for `resolveChange`).355  private _schema: WireSchema | null;356  private types?: ViewTypes;357  // SSR first-paint seed (SSR-DESIGN.md §6): pre-projected rows installed by `seed`, returned by358  // `data` until the maintained tree's first live SNAPSHOT lands (the Store calls `retireSeed` then).359  // It deliberately SURVIVES `reset` (the `hello`), so a seeded view bridges the `hello`→snapshot gap360  // instead of flashing empty in between (a `hello` sets the schema but its data arrives one round-trip361  // later, on the snapshot). `null` ⇒ no seed.362  private seeded: readonly R[] | null = null;363  private top: Node[] = [];364  private dirty = true;365  private cached: readonly R[] | null = null;366  // `complete` by default: a backend with no server lifecycle (the in-process engine) never pushes367  // a resultType, so its views read as authoritative. The Store overrides this for backends that do368  // (the optimistic backend: `unknown` until hydrated, etc.).369  private rt: ResultType = "complete";370  private readonly listeners = new Set<(data: readonly R[]) => void>();371  // The per-view CHANGE stream ({@link onChanges}) — the diff a narrator rides, distinct from the372  // data channel above. Empty until a narrator attaches: an idle view pays one `.size` check per fold.373  private readonly changeListeners = new Set<ViewChangeListener>();374  // Changes folded but not yet delivered to `changeListeners` — buffered while the Store defers a375  // commit (mirrors the deferred `notify`), then netted + flushed at the commit boundary. Segmented376  // BY PHASE (not one last-writer-wins phase scalar): a bracket that folds both a `snapshot` and a377  // `batch` into this view keeps them apart, each netted and delivered under its OWN phase, so a378  // snapshot is never mislabeled `batch` (nor a batch dropped as `snapshot`). Consecutive same-phase379  // folds coalesce into one segment so a whole commit's batch nets together.380  private pendingSegments: { phase: ChangePhase; changes: FlatChange[] }[] = [];381382  constructor(schema?: WireSchema, types?: ViewTypes, qid: QueryId = 0) {383    this._schema = schema ?? null;384    this.types = types;385    this._qid = qid;386  }387388  get qid(): QueryId {389    return this._qid;390  }391392  get schema(): WireSchema | null {393    return this._schema;394  }395396  get resultType(): ResultType {397    return this.rt;398  }399400  /** Set the query's lifecycle state (the Store routes the backend's per-query signal here).401   *  Notifies subscribers on a change so a status-bound listener (React `useQueryStatus`) re-reads,402   *  WITHOUT re-projecting data (it is unchanged). */403  setResultType(rt: ResultType): void {404    if (this.rt === rt) return;405    this.rt = rt;406    for (const l of this.listeners) l(this.data);407  }408409  /** (Re)bind to a schema and clear the tree IN PLACE. The first `hello` (pending → ready)410   *  and a re-hydrate (gap → new epoch — FLAT-CHANGES-DESIGN.md §2.3) both go through here, so411   *  the caller's view reference and its subscribers survive a re-subscribe. Does NOT notify —412   *  the snapshot that follows (`applyChanges`) does, avoiding an empty-then-filled flicker.413   *  KEEPS any SSR `seeded` rows: they are retired only when the first live snapshot lands414   *  ({@link retireSeed}, driven by the Store), so `data` shows the seed — not an empty tree —415   *  across the whole `hello`→first-`snapshot` gap. */416  reset(schema: WireSchema, types?: ViewTypes): void {417    this._schema = schema;418    this.types = types;419    this.top = [];420    this.cached = null;421    this.dirty = true;422    // Drop any changes buffered under the OLD epoch: a re-hydrate cuts a new schema, and423    // `deliverChanges` resolves against the current `_schema`, so pre-epoch changes must never424    // survive to be delivered against the new one (mirrors `destroy`). The snapshot that follows425    // this reset re-establishes the buffer for the new epoch.426    this.pendingSegments = [];427  }428429  /** Install a pre-projected SSR first-paint snapshot (SSR-DESIGN.md §6). The rows are already430   *  in result shape (json columns parsed, relationships nested), so a view with no live backend431   *  (the server one-shot Store) reads them directly, and a browser view shows them until its432   *  first live snapshot lands ({@link retireSeed}). Does NOT notify — it is set at materialize433   *  time, before any subscriber, and the live snapshot that follows notifies. */434  seed(rows: readonly R[]): void {435    this.seeded = rows;436  }437438  /** Retire the SSR first-paint seed — the Store calls this as it folds the maintained tree's first439   *  live snapshot, so `data` switches from the seed to the live tree with no empty gap between them440   *  (the seed deliberately survived the earlier `reset`/`hello`). Idempotent. Does NOT notify — the441   *  snapshot fold it accompanies does; BUT when that fold is empty (a 0-row result, or rows already442   *  in `top`) it notifies nothing, so the Store forces a {@link notify} on the strength of the `true`443   *  return here — else the view reads the live tree yet never re-renders (a frozen seed). Returns444   *  whether a live seed was actually cleared (so the Store knows a forced notify is owed). */445  retireSeed(): boolean {446    if (this.seeded === null) return false;447    this.seeded = null;448    return true;449  }450451  /** Apply a batch (the hydrate snapshot or one transaction's events) in order, then452   *  notify subscribers once. Order is significant (FLAT-CHANGES-DESIGN.md §5.4). A no-op453   *  while pending (changes never precede the `hello` that resets the schema).454   *455   *  `enrichRemoves` ⇒ before a removed node is dropped, reconstruct its full subtree and attach it456   *  to the `remove` op's `node` (in place, so the same event object the Store fans out to its457   *  change subscribers carries it). Off by default — paid only when a consumer asked for it, and458   *  only on a real eviction (an rc-decrement that keeps the row leaves `node` absent). A view with459   *  an attached {@link onChanges} listener ALSO enriches (per-view, no global opt-in), so a460   *  narrator can resolve a removed row's subs whether or not the store-global counter is set.461   *462   *  `deferNotify` ⇒ fold but do NOT notify; the caller is responsible for calling {@link flush}463   *  later. The Store uses this to fold every view in one commit before notifying any subscriber464   *  (cross-view-atomic notification — `Store.onCommitBoundary`); standalone use leaves it off, so465   *  a bare view still fires its subscribers after each applied batch.466   *467   *  `phase` tags the {@link onChanges} delivery (`snapshot` for the hydrate, `batch` otherwise); it468   *  does not affect the fold. Returns whether the batch changed the view (so a deferring caller469   *  knows it must be flushed). */470  applyChanges(events: FlatChange[], enrichRemoves = false, deferNotify = false, phase: ChangePhase = "batch"): boolean {471    if (this._schema === null) return false;472    // Enrich this view's removes when a narrator is attached (see the `enrichRemoves` doc above) —473    // the per-view twin of the store's global `removedSubtree` counter.474    const enrich = enrichRemoves || this.changeListeners.size > 0;475    // Deliver ONLY the ops that actually changed the tree: a duplicate-path add (rc bump), an476    // rc-decrement remove that left the row, a no-op edit, and a gated op all return false — none is477    // a real change, so none should reach a narrator. (Enrichment already attached the subtree to a478    // real remove during the fold, so a surviving remove carries its `node`.)479    const applied: FlatChange[] = [];480    for (const e of events) {481      if (this.applyAt(this.top, this._schema, e.path, e.op, 0, enrich)) applied.push(e);482    }483    if (applied.length === 0) return false;484    this.dirty = true;485    if (deferNotify) {486      if (this.changeListeners.size > 0) {487        const segs = this.pendingSegments;488        const last = segs.length > 0 ? segs[segs.length - 1] : undefined;489        if (last && last.phase === phase) for (const e of applied) last.changes.push(e);490        else segs.push({ phase, changes: applied.slice() });491      }492      return true;493    }494    // Inline (standalone / no commit bracket): notify data subscribers, then deliver the change495    // stream — each isolated, first error re-raised once both have run (mirroring the Store's496    // per-view flush isolation) so a throwing narration listener never starves the data channel,497    // aborts the fold's return, or propagates raw into the engine's push path.498    let firstError: unknown;499    let hasError = false;500    const note = (e: unknown): void => {501      if (!hasError) {502        hasError = true;503        firstError = e;504      }505    };506    try {507      this.notify();508    } catch (e) {509      note(e);510    }511    this.deliverChanges(applied, phase, note);512    if (hasError) throw firstError;513    return true;514  }515516  /** Notify subscribers with the current data. The deferred half of {@link applyChanges} (when517   *  `deferNotify` was set): the Store calls this at the commit-notify barrier — after every view518   *  touched by the same commit has folded — so a subscriber that re-reads a sibling view inside519   *  its callback observes post-commit data, never a torn mid-commit state. */520  flush(): void {521    // Take the buffer BEFORE delivery so a throwing listener can never leave changes buffered to be522    // re-delivered — merged and cross-phase-netted — into a LATER commit's flush. Notify and every523    // segment's delivery are isolated; the first error is re-raised after all have run, so one bad524    // subscriber starves neither the data channel nor a sibling segment (matching `flushCommit`).525    const segs = this.pendingSegments;526    this.pendingSegments = [];527    let firstError: unknown;528    let hasError = false;529    const note = (e: unknown): void => {530      if (!hasError) {531        hasError = true;532        firstError = e;533      }534    };535    try {536      this.notify();537    } catch (e) {538      note(e);539    }540    for (const seg of segs) this.deliverChanges(seg.changes, seg.phase, note);541    if (hasError) throw firstError;542  }543544  get data(): readonly R[] {545    // A live SSR seed owns `data` until it is retired on the first live snapshot ({@link retireSeed}) —546    // this spans pending (no schema) AND the post-`hello`, pre-snapshot window (schema set, tree still547    // empty), so a seeded query never flashes empty during the handoff. `null` seed ⇒ normal behavior.548    if (this.seeded !== null) return this.seeded;549    if (this._schema === null) return EMPTY as readonly R[];550    if (!this.dirty && this.cached !== null) return this.cached;551    this.cached = this.top.map((n) => this.project(n, this._schema as WireSchema, this.types)) as R[];552    this.dirty = false;553    return this.cached;554  }555556  subscribe(listener: (data: readonly R[]) => void): () => void {557    this.listeners.add(listener);558    listener(this.data);559    return () => {560      this.listeners.delete(listener);561    };562  }563564  onChanges(listener: ViewChangeListener): () => void {565    this.changeListeners.add(listener);566    return () => {567      this.changeListeners.delete(listener);568    };569  }570571  /** Net the folded batch and hand the survivors to the change listeners, AFTER the data `notify`572   *  (the order `Store.subscribeChanges` consumers already observe). A batch that nets to nothing —573   *  a correctly predicted rebase — makes no call, so a narrator sees only real change. Each listener574   *  is isolated: a throwing one reports to `note` (the caller re-raises the first) and the rest still575   *  run, so one bad narration template never starves a sibling listener nor corrupts the view. */576  private deliverChanges(events: FlatChange[], phase: ChangePhase, note: (e: unknown) => void): void {577    if (this.changeListeners.size === 0) return;578    // `_schema` is non-null here: `applyChanges` early-returns while pending, and `flush` runs only579    // after a fold — so a listener always has the position→name source in hand.580    const schema = this._schema as WireSchema;581    const net = netChanges(events, schema);582    if (net.length === 0) return;583    for (const l of this.changeListeners) {584      try {585        l(net, phase, schema);586      } catch (e) {587        note(e);588      }589    }590  }591592  destroy(): void {593    this.listeners.clear();594    this.changeListeners.clear();595    this.pendingSegments = [];596    this.top = [];597  }598599  // --- apply -------------------------------------------------------------------600601  private applyAt(list: Node[], schema: WireSchema, path: FlatChange["path"], op: FlatOp, depth: number, enrichRemoves: boolean): boolean {602    if (depth === path.length) {603      return this.applyOp(list, schema, op, enrichRemoves);604    }605    const seg = path[depth];606    const child = schema.relationships[seg.rel]?.child;607    if (!child) return false; // in-view gate: a gating slot drops the whole change608    const { found, index } = binarySearch(list, seg.parentRow, schema.sort);609    if (!found) throw new Error("flat ArrayView: parent not found at path hop (inconsistent stream)");610    const node = list[index];611    const changed = this.applyAt(node.rels[seg.rel], child, path, op, depth + 1, enrichRemoves);612    if (changed) node.out = null; // a descendant changed → this node must re-project its rel array613    return changed;614  }615616  private applyOp(list: Node[], schema: WireSchema, op: FlatOp, enrichRemoves: boolean): boolean {617    if (op.tag === "add") return this.applyAdd(list, schema, op.node);618    if (op.tag === "remove") {619      const removed = this.applyRemove(list, schema, op.row);620      // Attach the full removed subtree (in place) for an opted-in consumer; absent when the row621      // only lost a reference (rc > 1) and so did not actually leave the result.622      if (removed && enrichRemoves) op.node = toWireNode(removed, schema);623      return removed !== null;624    }625    return this.applyEdit(list, schema, op.old, op.new);626  }627628  private applyAdd(list: Node[], schema: WireSchema, wnode: WireNode): boolean {629    const at = binarySearch(list, wnode.row, schema.sort);630    if (at.found) {631      list[at.index].rc += 1; // duplicate path to an existing row; ignore subtree632      return false;633    }634    list.splice(at.index, 0, buildNode(wnode, schema));635    return true;636  }637638  /** Drop one path to `row`. Returns the evicted {@link Node} (its full subtree intact) when the639   *  last path went away — `null` when another path still holds it (rc decremented, nothing left640   *  the result). */641  private applyRemove(list: Node[], schema: WireSchema, row: WireValue[]): Node | null {642    const at = binarySearch(list, row, schema.sort);643    if (!at.found) throw new Error("flat ArrayView: remove of a non-existent node");644    const node = list[at.index];645    if (node.rc === 1) {646      list.splice(at.index, 1);647      return node;648    }649    node.rc -= 1;650    return null;651  }652653  private applyEdit(list: Node[], schema: WireSchema, oldRow: WireValue[], newRow: WireValue[]): boolean {654    if (rowsEqual(oldRow, newRow)) return false;655    const sort = schema.sort;656    if (compareRows(oldRow, newRow, sort) === 0) {657      // Sort key unchanged → edit in place (keeps position + children + rc).658      const at = binarySearch(list, oldRow, sort);659      if (!at.found) throw new Error("flat ArrayView: edit of a non-existent node");660      list[at.index].row = newRow;661      list[at.index].out = null;662      return true;663    }664665    // Sort key changed → the row may move; rc may be > 1.666    const oldAt = binarySearch(list, oldRow, sort);667    if (!oldAt.found) throw new Error("flat ArrayView: edit old node does not exist");668    const oldPos = oldAt.index;669    const oldRc = list[oldPos].rc;670    const raw = binarySearch(list, newRow, sort);671    const found = raw.found;672    const pos = raw.index;673    const oldEntry = cloneNode(list[oldPos]); // capture (with children) BEFORE mutating674675    // Fast path: rc==1 and the row lands in the same slot after removal → edit in place.676    if (oldRc === 1 && (pos === oldPos || pos - 1 === oldPos)) {677      list[oldPos].row = newRow;678      list[oldPos].out = null;679      return true;680    }681682    // General move.683    const newRc = oldRc - 1;684    let adjusted: number;685    if (newRc === 0) {686      list.splice(oldPos, 1);687      adjusted = oldPos < pos ? pos - 1 : pos;688    } else {689      list[oldPos].rc = newRc; // ghost survives for the other path(s); its row is unchanged690      adjusted = pos;691    }692    if (found) {693      // Merge into the existing destination entry (keep its children); bump its rc.694      const existingRc = list[adjusted].rc;695      list[adjusted].row = newRow;696      list[adjusted].rc = existingRc + 1;697      list[adjusted].out = null;698    } else {699      // Move the (edited) old entry — keeping its children — to the new position.700      oldEntry.row = newRow;701      oldEntry.rc = 1;702      list.splice(adjusted, 0, oldEntry);703    }704    return true;705  }706707  // --- projection (memoized, structurally shared) -----------------------------708709  private project(node: Node, schema: WireSchema, types?: ViewTypes): unknown {710    if (node.out !== null) return node.out;711    const obj: Record<string, unknown> = {};712    for (let i = 0; i < schema.columns.length; i++) {713      const v = node.row[i];714      // A projected query carries an ABSENT cell (`undefined`, PROJECTION-SUPPORT-DESIGN.md715      // §5.3) for a column it did not sync; omit it from the result object so a query never716      // resolves a row from columns it did not select (§6). A present-and-null cell is `null`717      // and is kept. For a `'*'` query no cell is ever `undefined`, so this is inert (§7).718      if (v === undefined) continue;719      // convert-once: a json column's raw string is parsed to an object on store (cached in `out`).720      obj[schema.columns[i]] = types?.columnTypes[i] === "json" && typeof v === "string" ? JSON.parse(v) : v;721    }722    for (const rel of schema.relationships) {723      if (rel.child === null) continue; // gating slot — not part of .data724      const childList = node.rels[rel.slot] ?? [];725      const ct = types?.rels[rel.slot];726      if (rel.project) {727        // A scalar-projected relationship aggregate (REDUCE-DESIGN.md §9): unwrap the one-row728        // child to a bare scalar (its project.col cell), substituting the aggregate identity729        // when empty (a childless parent). A json-typed projected column is parsed like a column.730        const cell = childList.length > 0 ? childList[0].row[rel.project.col] : rel.project.identity;731        const projType = ct?.columnTypes[rel.project.col];732        obj[rel.name] = projType === "json" && typeof cell === "string" ? JSON.parse(cell) : cell;733        continue;734      }735      obj[rel.name] = rel.child.singular736        ? childList.length > 0737          ? this.project(childList[0], rel.child, ct)738          : null739        : childList.map((c) => this.project(c, rel.child as WireSchema, ct));740    }741    node.out = obj;742    return obj;743  }744745  /** Notify data subscribers with the current {@link data}. Normally driven by a fold ({@link746   *  applyChanges}/{@link flush}); the Store also calls it directly to land a seed retirement whose747   *  accompanying fold was empty (see {@link retireSeed}). */748  notify(): void {749    const d = this.data;750    for (const l of this.listeners) l(d);751  }752}753