API index and search · Build metadata

Source snapshot

packages/normalized/src/backend.ts

1// NormalizedBackend — the composition that turns a NORMALIZED server stream into the flat2// `Backend` seam the existing `Store` already drives (NORMALIZED-CHANGES-DESIGN.md §5/§7).3//4// A normalized client runs its OWN local engine and queries over its base tables. This5// backend wires that up by composing three pieces behind one `Backend`:6//   - a local `@rindle/wasm` engine (the base tables + local IVM that materializes results),7//   - `NormalizedSync` (cross-query refcount → net base-table mutations), and8//   - a `NormalizedSource` (the server's per-query normalized stream).9//10// The flow per query: `registerQuery` registers it BOTH on the local engine (→ an empty11// `FlatArrayView`) and on the server (→ its normalized footprint stream). Each normalized12// batch folds through `NormalizedSync` into net base mutations, which are applied to the13// local engine — whose own flat change stream then updates every affected view. Because the14// local engine fans a base write to ALL local queries, a row synced for one query updates15// any other query that reads it (the "local query resolution" the design is built for).16//17// Since this implements `Backend`, `new Store(schema, new NormalizedBackend(...))` reuses the18// whole Store/ArrayView machinery unchanged — the normalized path adds composition, not a19// second materialization layer.2021import { normalizedTableSchemas, Store, tableSpec } from "@rindle/client";22import type {23  Ast,24  Backend,25  BackendDevObserver,26  ChangeEvent,27  ColsMap,28  Mutation,29  NormalizedEvent,30  NormalizedSource,31  NormalizedTableSchema,32  QueryId,33  RemoteQuery,34  ResultType,35  Schema,36} from "@rindle/client";37import { WasmBackend } from "@rindle/wasm";3839import { aggTableSchemas, rewriteAggregates } from "./agg-table.ts";40import { NormalizedSync, type ColCounts, type PkCols } from "./sync.ts";4142// The normalized seam + event/table-schema types live in `@rindle/client` (sibling of `Backend`),43// so a ws (`@rindle/remote`) source and the in-process native source emit the same shape and44// `@rindle/normalized` needs no dependency on `@rindle/remote`. Re-exported for callers.45export type { NormalizedEvent, NormalizedSource, NormalizedTableSchema } from "@rindle/client";4647/** Each table's primary-key column indices, from the typed schema. */48function pkColsFromSchema<S extends ColsMap>(schema: Schema<S>): PkCols {49  const out: PkCols = {};50  for (const name of Object.keys(schema.tables)) out[name] = tableSpec(schema.tables[name]).primaryKey;51  return out;52}5354/** Each table's FULL column count (the union-row width), from the typed schema. */55function colCountsFromSchema<S extends ColsMap>(schema: Schema<S>): ColCounts {56  const out: ColCounts = {};57  for (const name of Object.keys(schema.tables)) out[name] = tableSpec(schema.tables[name]).columns.length;58  return out;59}6061/** Each table's column name → base ColId, from the typed schema (for mapping a projected62 *  hello's columns back to base positions, PROJECTION-SUPPORT-DESIGN.md §5.2). */63function colIndexFromSchema<S extends ColsMap>(schema: Schema<S>): Record<string, Map<string, number>> {64  const out: Record<string, Map<string, number>> = {};65  for (const name of Object.keys(schema.tables)) {66    const cols = tableSpec(schema.tables[name]).columns;67    out[name] = new Map(cols.map((c, i) => [c, i]));68  }69  return out;70}7172export class NormalizedBackend<S extends ColsMap> implements Backend {73  private readonly local: WasmBackend<S>;74  private readonly sync: NormalizedSync;75  private readonly source: NormalizedSource;76  private handler: (qid: QueryId, ev: ChangeEvent) => void = () => {};77  /** The Store's commit-boundary handler ({@link Backend.onCommitBoundary}), forwarded from the78   *  local engine's `dispatch` brackets so the Store folds every affected view before notifying any79   *  subscriber (cross-view-atomic notification). A normalized batch becomes net base mutations80   *  applied to the local engine in one commit, which fans to every view reading those rows — this81   *  carries that commit's boundary up so all those views notify together. */82  private boundaryHandler: (phase: "begin" | "end") => void = () => {};83  private readonly devObservers = new Set<BackendDevObserver>();84  private readonly remoteSubs = new Map<string, RemoteSub>();85  private readonly localToRemote = new Map<QueryId, string>();86  private readonly remoteRetainToLocal = new Map<QueryId, QueryId | undefined>();87  private readonly sourceToRemote = new Map<QueryId, string>();88  private readonly resultTypes = new Map<QueryId, ResultType>();89  private readonly hydrated = new Set<QueryId>();90  /** Local qids that are JUST hydrating on the in-flight `onNormalized` snapshot — set only for the91   *  duration of its `local.mutate`, so the local-event forwarder stamps their fold `catchUp`: the92   *  whole first result set arrives as a `batch` (we hydrate the view by mutating the embedded engine,93   *  which speaks in batches — there is no second `snapshot`), and `catchUp` is the flag that tells the94   *  Store this batch IS hydration (retire the SSR seed, phase it as a snapshot for narration). The95   *  optimistic backend does the identical thing across its reconcile cycle. `null` outside a hydrate. */96  private catchUpQids: Set<QueryId> | null = null;97  private resultTypeHandler: (qid: QueryId, rt: ResultType) => void = () => {};98  /** The client's OWN typed per-table schemas (for CRIT#4 validation), fixed at construction. */99  private readonly clientTables: NormalizedTableSchema[];100  /** Synthetic aggregate tables (`__agg_*`) registered so far, by name (§3.3). Per aggregate101   *  DEFINITION (not per query), so two queries over the same count share one table. */102  private readonly synthetic = new Map<string, NormalizedTableSchema>();103  /** table → (column name → base ColId), for mapping a projected hello to base positions. */104  private readonly colIndex: Record<string, Map<string, number>>;105  /** table → full column count, to detect whether a hello's table is projected. */106  private readonly colCounts: ColCounts;107  /** Synthetic table name → how many registered queries reference it. Materialized on `0→1`,108   *  reclaimed (engine source + refcount layer) on `1→0` — so aggregate state is not permanent (§4). */109  private readonly syntheticRefs = new Map<string, number>();110  /** queryId → the synthetic tables it referenced at registration, to decrement on teardown. */111  private readonly queryAggTables = new Map<QueryId, string[]>();112113  constructor(schema: Schema<S>, source: NormalizedSource) {114    this.local = new WasmBackend(schema);115    // The local engine's flat stream IS this backend's stream upward (→ the Store's views). A116    // just-hydrating query's first result set arrives here as a `batch` (we fold it by mutating the117    // engine); stamp it `catchUp` so the Store treats it as hydration (retire the SSR seed, phase it118    // as a snapshot) rather than an incremental change — see {@link catchUpQids}.119    this.local.onEvent((qid, ev) => {120      const stamped = ev.type === "batch" && this.catchUpQids?.has(qid) ? { ...ev, catchUp: true } : ev;121      this.handler(qid, stamped);122    });123    // Forward the local engine's commit brackets up to the Store (cross-view-atomic notification):124    // a normalized batch's net base mutations commit to `this.local` in one transaction, so its125    // commit boundary is ours — every view those rows touch then notifies together.126    this.local.onCommitBoundary((phase) => this.boundaryHandler(phase));127    this.colIndex = colIndexFromSchema(schema);128    this.colCounts = colCountsFromSchema(schema);129    this.sync = new NormalizedSync(pkColsFromSchema(schema), this.colCounts);130    this.source = source;131    // Hand the source our OWN typed schema so it validates each server hello against it and132    // rejects a column-order / PK skew instead of silently transposing positional cells (CRIT#4).133    this.clientTables = normalizedTableSchemas(schema);134    this.source.expectClientSchema?.(this.clientTables);135    this.source.onNormalized((qid, ev) => this.onNormalized(qid, ev));136  }137138  registerQuery(qid: QueryId, ast: Ast, remote?: RemoteQuery): void {139    // A relationship aggregate (`count(child)`) is DISPLAYED from a server-authoritative140    // synthetic base table, not recomputed locally (AGGREGATE-SYNC-DESIGN.md §3.3): register141    // that table (engine + refcount layer + hello validation), then drive the local engine142    // off a rewritten AST whose `count` relationships read it with a plain projected join.143    this.ensureSyntheticTables(qid, ast);144    // Local first: builds the (empty) view synchronously. Then the server stream hydrates it.145    this.local.registerQuery(qid, rewriteAggregates(ast));146    if (remote) {147      this.retainRemote(qid, remote, qid);148    } else {149      this.hydrated.add(qid);150      this.setResultType(qid, "complete");151    }152  }153154  /** Register every synthetic aggregate table `ast` needs that we haven't seen yet: on the155   *  local engine (so it can join to it), on `NormalizedSync` (so its rows refcount/GC by156   *  group key), and into the source's expected-schema set (so the server's `hello` — which157   *  advertises the same table — passes CRIT#4 validation, the client deriving the schema158   *  identically). Idempotent across queries that share an aggregate definition. */159  private ensureSyntheticTables(qid: QueryId, ast: Ast): void {160    if (this.queryAggTables.has(qid)) return; // idempotent per qid161    let added = false;162    const names: string[] = [];163    for (const t of aggTableSchemas(ast)) {164      names.push(t.name);165      const prev = this.syntheticRefs.get(t.name) ?? 0;166      this.syntheticRefs.set(t.name, prev + 1);167      if (prev > 0) continue; // another query already materialized it — just refcount168      this.synthetic.set(t.name, t);169      this.local.registerTable(t.name, { columns: t.columns, primaryKey: t.primaryKey });170      this.sync.registerTable(t.name, t.primaryKey);171      added = true;172    }173    if (names.length) this.queryAggTables.set(qid, names);174    if (added) this.source.expectClientSchema?.([...this.clientTables, ...this.synthetic.values()]);175  }176177  /** Decrement each synthetic table query `qid` referenced; remove the ones that reach 0 (no178   *  reader left) from the engine + refcount layer — aggregate state reclaimed, not permanent179   *  (§4). Runs AFTER `local.unregisterQuery(qid)` so the source has no live connection when180   *  `unregisterTable` frees it. */181  private releaseSyntheticTables(qid: QueryId): void {182    const names = this.queryAggTables.get(qid);183    if (!names) return;184    this.queryAggTables.delete(qid);185    let removed = false;186    for (const name of names) {187      const next = (this.syntheticRefs.get(name) ?? 1) - 1;188      if (next > 0) {189        this.syntheticRefs.set(name, next);190        continue;191      }192      this.syntheticRefs.delete(name);193      this.local.unregisterTable(name);194      this.sync.unregisterTable(name);195      this.synthetic.delete(name);196      removed = true;197    }198    if (removed) this.source.expectClientSchema?.([...this.clientTables, ...this.synthetic.values()]);199  }200201  unregisterQuery(qid: QueryId): void {202    const remoteQid = this.releaseRemote(qid);203    if (remoteQid !== undefined) {204      // Drop this remote footprint's refcounts; rows referenced by no other named query are205      // GC'd from the local base tables (→ other views shrink if they shared them).206      const muts = this.sync.dropQuery(remoteQid);207      if (muts.length) void this.local.mutate(muts);208    }209    this.local.unregisterQuery(qid);210    // Pipeline gone (no live conn) + `__agg` rows GC'd above → free any now-unread synthetic table.211    this.releaseSyntheticTables(qid);212    this.resultTypes.delete(qid);213    this.hydrated.delete(qid);214  }215216  retainRemoteQuery(qid: QueryId, remote: RemoteQuery, localQueryId?: QueryId, ast?: Ast): void {217    if (ast) this.ensureSyntheticTables(qid, ast);218    this.retainRemote(qid, remote, localQueryId);219  }220221  releaseRemoteQuery(qid: QueryId): void {222    const remoteQid = this.releaseRemote(qid);223    this.resultTypes.delete(qid);224    this.hydrated.delete(qid);225    this.releaseSyntheticTables(qid);226    if (remoteQid === undefined) return;227    const muts = this.sync.dropQuery(remoteQid);228    if (muts.length) void this.local.mutate(muts);229  }230231  /** Writes are authoritative-only here: send to the server, the stream reconciles locally.232   *  (Optimistic local apply + rebase is Slice 6.) */233  mutate(mutations: Mutation[]): Promise<void> {234    return this.source.mutate(mutations);235  }236237  onEvent(handler: (qid: QueryId, ev: ChangeEvent) => void): void {238    this.handler = handler;239  }240241  onCommitBoundary(handler: (phase: "begin" | "end") => void): void {242    this.boundaryHandler = handler;243  }244245  onResultType(handler: (qid: QueryId, rt: ResultType) => void): void {246    this.resultTypeHandler = handler;247  }248249  __attachDevtoolsServerDeltas(observer: BackendDevObserver): () => void {250    this.devObservers.add(observer);251    return () => {252      this.devObservers.delete(observer);253    };254  }255256  private onNormalized(qid: QueryId, ev: NormalizedEvent): void {257    this.emitServerDelta(qid, ev);258    // The slim hello carries table schemas + fingerprint; envelope validation (epoch/seq/gap)259    // is the source's job (the ws Subscriber, §5.3) — it precedes every (re)hydrate snapshot.260    if (ev.type === "hello") {261      // Learn this query's per-table column map (PROJECTION-SUPPORT-DESIGN.md §5.2): map each262      // advertised column to its base ColId BY NAME. The hello may carry FEWER columns than the263      // client's schema (a projection) or MORE (an EXPANDED server table mid an264      // `expand-then-contract` migration) — a column the client lacks maps to `-1`, a DROP265      // sentinel the sync layer discards. Register the map so the sync layer scatters the rows266      // into the shared full-width union; a table whose map is an in-order, drop-free full width267      // IS '*' (a verbatim full row) and is left unregistered. Idempotent across re-hydrate epochs.268      for (const t of ev.tables) {269        const full = this.colCounts[t.name];270        if (full === undefined) continue; // unknown/synthetic table → '*' (full presence)271        const index = this.colIndex[t.name];272        const cols = t.columns.map((name) => index?.get(name) ?? -1);273        // Register a non-trivial map; otherwise revert to '*' — and CLEAR any stale map a prior274        // epoch left (a server that expanded then contracted back), so the now-exact rows don't275        // scatter through a `-1`-bearing layout (silent cell corruption).276        if (cols.length !== full || cols.some((c, i) => c !== i)) this.sync.registerProjection(qid, t.name, cols);277        else this.sync.unregisterProjection(qid, t.name);278      }279      return;280    }281    // A `snapshot` is the seq-0 baseline — initial OR a re-hydrate under a new epoch (§5.3). Both282    // go through `rehydrate` (a footprint DIFF): for the first one the footprint is empty so it283    // degenerates to all-adds; for a re-hydrate it removes rows that left during the gap. A284    // `batch` is an incremental delta → `applyBatch`.285    const muts = ev.type === "snapshot" ? this.sync.rehydrate(qid, ev.ops) : this.sync.applyBatch(qid, ev.ops);286    if (ev.type !== "snapshot") {287      if (muts.length) void this.local.mutate(muts); // incremental delta → local engine → views288      return;289    }290    // A snapshot is the query's hydration point. Flip its views to `complete` BEFORE folding the rows291    // (the Store retires the SSR seed only once the query is authoritative), and capture the local292    // qids that transition into hydrated on THIS snapshot so the forwarder stamps their fold `catchUp`293    // — the first result set is delivered by mutating the engine, i.e. as a `batch`, not a `snapshot`.294    // (An already-live view reading the same rows sees them as a genuine incremental add, so it is NOT295    // in this set and folds a normal batch.) Order mirrors the optimistic backend (hydrate → complete,296    // THEN the catch-up delta).297    const wasHydrated = new Set(this.hydrated);298    this.markSubHydrated(qid);299    let newly: Set<QueryId> | null = null;300    for (const localQid of this.hydrated) {301      if (!wasHydrated.has(localQid)) (newly ??= new Set()).add(localQid);302    }303    this.catchUpQids = newly;304    try {305      if (muts.length) void this.local.mutate(muts); // → local engine → flat stream → views306    } finally {307      this.catchUpQids = null;308    }309    // A hydration can fold NOTHING: a 0-row result, or one whose rows are already present in the engine310    // (fully covered by an already-hydrated sibling → the shared rows dedup to 0 net muts). Then no311    // batch was emitted above, so the Store never saw the hydration point and the SSR seed would stay312    // stuck. Signal it explicitly with an EMPTY catch-up per just-hydrated qid — the Store retires the313    // seed and reveals whatever is already in the view's tree.314    if (!muts.length && newly) {315      for (const localQid of newly) this.handler(localQid, { type: "batch", events: [], catchUp: true });316    }317  }318319  private emitServerDelta(sourceQid: QueryId, ev: NormalizedEvent): void {320    if (!this.devObservers.size) return;321    for (const qid of this.localQidsForSource(sourceQid)) {322      for (const o of this.devObservers) o.onServerDelta?.(qid, { format: "normalized", event: ev });323    }324  }325326  private localQidsForSource(sourceQid: QueryId): QueryId[] {327    const key = this.sourceToRemote.get(sourceQid);328    const sub = key ? this.remoteSubs.get(key) : undefined;329    if (!sub) return [sourceQid];330    const localQids = [...sub.localQids.keys()];331    return localQids.length ? localQids : [sourceQid];332  }333334  private retainRemote(retainQid: QueryId, remote: RemoteQuery, localQueryId: QueryId | undefined = retainQid): void {335    const key = remoteKey(remote);336    let sub = this.remoteSubs.get(key);337    let isNew = false;338    if (!sub) {339      sub = { sourceQid: retainQid, remote, refCount: 0, localQids: new Map(), hydrated: false };340      this.remoteSubs.set(key, sub);341      this.sourceToRemote.set(sub.sourceQid, key);342      isNew = true;343    }344    sub.refCount++;345    if (localQueryId !== undefined) {346      sub.localQids.set(localQueryId, (sub.localQids.get(localQueryId) ?? 0) + 1);347      if (sub.hydrated) this.hydrated.add(localQueryId);348      else this.hydrated.delete(localQueryId);349      this.recomputeResultType(localQueryId);350    }351    this.localToRemote.set(retainQid, key);352    this.remoteRetainToLocal.set(retainQid, localQueryId);353    if (isNew) this.source.registerQuery(sub.sourceQid, remote);354  }355356  private releaseRemote(retainQid: QueryId): QueryId | undefined {357    const key = this.localToRemote.get(retainQid);358    if (!key) return undefined;359    this.localToRemote.delete(retainQid);360    const localQueryId = this.remoteRetainToLocal.get(retainQid);361    this.remoteRetainToLocal.delete(retainQid);362    const sub = this.remoteSubs.get(key);363    if (!sub) return undefined;364    sub.refCount--;365    if (localQueryId !== undefined) {366      const refs = (sub.localQids.get(localQueryId) ?? 0) - 1;367      if (refs > 0) {368        sub.localQids.set(localQueryId, refs);369      } else {370        sub.localQids.delete(localQueryId);371        if (!this.hasRemoteDependency(localQueryId)) {372          this.hydrated.add(localQueryId);373          this.recomputeResultType(localQueryId);374        }375      }376    }377    if (sub.refCount > 0) return undefined;378    this.source.unregisterQuery(sub.sourceQid);379    this.sourceToRemote.delete(sub.sourceQid);380    this.remoteSubs.delete(key);381    return sub.sourceQid;382  }383384  private markSubHydrated(sourceQid: QueryId): void {385    const key = this.sourceToRemote.get(sourceQid);386    if (!key) return;387    const sub = this.remoteSubs.get(key);388    if (!sub || sub.hydrated) return;389    sub.hydrated = true;390    for (const localQid of sub.localQids.keys()) {391      this.hydrated.add(localQid);392      this.recomputeResultType(localQid);393    }394  }395396  private hasRemoteDependency(localQid: QueryId): boolean {397    for (const sub of this.remoteSubs.values()) {398      if (sub.localQids.has(localQid)) return true;399    }400    return false;401  }402403  private setResultType(qid: QueryId, rt: ResultType): void {404    if (this.resultTypes.get(qid) === rt) return;405    this.resultTypes.set(qid, rt);406    this.resultTypeHandler(qid, rt);407  }408409  private recomputeResultType(qid: QueryId): void {410    this.setResultType(qid, this.hydrated.has(qid) ? "complete" : "unknown");411  }412}413414interface RemoteSub {415  sourceQid: QueryId;416  remote: RemoteQuery;417  refCount: number;418  localQids: Map<QueryId, number>;419  hydrated: boolean;420}421422function remoteKey(remote: RemoteQuery): string {423  return stableJson([remote.name, remote.args]);424}425426function stableJson(value: unknown): string {427  if (value === null || typeof value !== "object") return JSON.stringify(value);428  if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`;429  const obj = value as Record<string, unknown>;430  return `{${Object.keys(obj)431    .sort()432    .map((k) => `${JSON.stringify(k)}:${stableJson(obj[k])}`)433    .join(",")}}`;434}435436/** A local-first {@link Store} whose base tables are fed by a server's normalized stream.437 *  The returned Store is the ordinary `@rindle/client` Store — `store.query…materialize()` and438 *  `store.write(…)` work as always; only the backend composition differs. */439export function createNormalizedStore<S extends ColsMap>(440  schema: Schema<S>,441  source: NormalizedSource,442): Store<S> {443  return new Store(schema, new NormalizedBackend(schema, source));444}445