API index and search · Build metadata
Source snapshot
packages/wasm/src/index.ts
1// @rindle/wasm — the WASM backend: the in-process IVM engine (src/wasm/db.rs) behind the2// @rindle/client `Backend` seam (WASM-CLIENT-DESIGN.md §2.1). Also re-exports @rindle/client so a3// local app can import everything from here.4//5// The wasm is a `--target web` ESM artifact (packages/wasm/pkg, built by ./build.sh): it6// needs an explicit, one-time init — `await initWasm()` — before constructing a backend.7// In Node the bytes are read from the package; in a browser/bundler the wasm is fetched.89import init, { Db } from "../pkg/rindle.js";10import { localTableNames, Store, tableSpec } from "@rindle/client";11import type {12 Ast,13 Backend,14 ChangeEvent,15 ColsMap,16 Mutation,17 QueryId,18 Schema,19} from "@rindle/client";2021export * from "@rindle/client";2223// The wasm Db surface this backend uses (kept local so it doesn't depend on the generated .d.ts).24// One source per table (302-ROOM-STORE-SEPARATION-DESIGN.md): the engine holds ordinary tables25// only — a room-backed table is just another registered table whose deltas happen to arrive on a26// room channel. The multi-source merge surface that briefly lived here was removed with27// `rust/rindle/src/merge.rs`.28interface WasmDb {29 registerTable(table: string, spec: { columns: string[]; primaryKey: number[] }, local: boolean): void;30 unregisterTable(table: string): void;31 query(queryId: number, ast: Ast): { comparatorVersion: number; schema: unknown; snapshot: unknown[] };32 destroyQuery(queryId: number): void;33 write(): WasmWriteTxn;34 serverBatchBegin(deltas: ServerDeltaOp[]): void;35 serverBatchEnd(): Array<{ queryId: number; events: unknown[] }>;36}3738/** A raw staged write transaction over the wasm engine — the surface an optimistic client39 * mutator runs against (`get` reads the live state under this txn's staged overlay, so a40 * read-dependent mutator sees its own writes; OPTIMISTIC-WRITES-DESIGN.md §4.1). */41export interface WasmWriteTxn {42 add(table: string, row: unknown[]): void;43 remove(table: string, row: unknown[]): void;44 edit(table: string, oldRow: unknown[], newRow: unknown[]): void;45 get(table: string, pk: unknown[]): unknown[] | undefined;46 /** Run a one-shot query over the state this txn is mutating — the live base plus this47 * txn's own staged writes-so-far (read-your-writes; 203-MUTATOR-READS-DESIGN.md §5.2).48 * Synchronous (over a lazy read-cache fork of the staged buffer). Returns the query's rows49 * as keyed objects with their materialized `related` children nested by name (identical in50 * shape to a `view.data` row), in the query's order. */51 query(ast: Ast): unknown[];52 commit(): Array<{ queryId: number; events: unknown[] }>;53 rollback(): void;54}5556/** One base-table row op of a coherent server delta (the §1.3 `D`), bare cells. */57export type ServerDeltaOp =58 | { table: string; type: "add"; row: unknown[] }59 | { table: string; type: "remove"; row: unknown[] }60 | { table: string; type: "edit"; row: unknown[]; old: unknown[] };6162let initialized: Promise<void> | null = null;6364/** Initialize the wasm module (idempotent). Call once at startup before `new WasmBackend`65 * / `createWasmStore`. Browser/bundler: no args (the wasm is fetched). Node: the bytes are66 * read from the package. Pass `moduleOrPath` to override (a `WebAssembly.Module`, URL, or bytes). */67export function initWasm(moduleOrPath?: unknown): Promise<void> {68 if (!initialized) {69 initialized = (async () => {70 if (moduleOrPath !== undefined) {71 await init({ module_or_path: moduleOrPath });72 } else if ((globalThis as { process?: { versions?: { node?: string } } }).process?.versions?.node) {73 const { readFile } = await import("node:fs/promises");74 const bytes = await readFile(new URL("../pkg/rindle_bg.wasm", import.meta.url));75 await init({ module_or_path: bytes });76 } else {77 await init();78 }79 })();80 }81 return initialized;82}8384/** The in-process WASM backend. Requires {@link initWasm} to have resolved first. */85/** One engine commit's per-query batches (the shape `WriteTxn.commit` / `serverBatchEnd` return). */86type BatchSet = Array<{ queryId: number; events: unknown[] }>;8788export class WasmBackend<S extends ColsMap> implements Backend {89 private readonly db: WasmDb;90 private handler: (qid: QueryId, ev: ChangeEvent) => void = () => {};91 // The Store's commit-boundary handler ({@link Backend.onCommitBoundary}): `dispatch` brackets92 // each commit's per-query batch delivery with `begin`/`end` so the Store folds every affected93 // view before notifying any subscriber (cross-view-atomic notification). Defaults to a no-op so94 // a Store that never registers it (or none at all) just gets per-event notification.95 private boundaryHandler: (phase: "begin" | "end") => void = () => {};96 /** Local-only table names (`201-LOCAL-ONLY-TABLES-DESIGN.md` §4): registered UNTRACKED (so the97 * optimistic rewind never reverts them) and the only tables {@link writeLocal} accepts. */98 private readonly localTables: Set<string>;99100 // Non-reentrant delivery (#15): commits dispatch through a FIFO queue. A subscriber that101 // synchronously triggers another write enqueues that commit's batches; they drain only after102 // the in-flight commit's batches finish, preserving per-query commit order regardless of what103 // subscribers do.104 private readonly dispatchQueue: BatchSet[] = [];105 private draining = false;106107 constructor(schema: Schema<S>) {108 this.db = new Db() as unknown as WasmDb;109 this.localTables = localTableNames(schema);110 for (const name of Object.keys(schema.tables)) {111 // A local table registers as a source but skips optimistic tracking (C1) — the `local` bit112 // crosses to the engine here.113 this.db.registerTable(name, tableSpec(schema.tables[name]), this.localTables.has(name));114 }115 }116117 /** Register an additional base table after construction — for a SYNTHETIC aggregate table118 * (`AGGREGATE-SYNC-DESIGN.md` §3.3) that is not in the typed schema. The119 * `NormalizedBackend` registers each such `__agg_*` table once, before a query that reads120 * it, so the local engine can join to it (the relationship `count` it backs is shipped by121 * the server, never recomputed). Always synced (tracked) — synthetic tables are never local. */122 registerTable(name: string, spec: { columns: string[]; primaryKey: number[] }): void {123 this.db.registerTable(name, spec, false);124 }125126 /** Direct-commit a batch of LOCAL-only writes (`201-LOCAL-ONLY-TABLES-DESIGN.md` §6): push them127 * straight through the engine on the ordinary delivery path, OUTSIDE any optimistic cycle (a128 * local table is untracked, so it never rebases). Rejects a synced/tracked table (M2).129 * `onCommitted` fires between the engine commit and subscriber delivery ({@link writeWith}) —130 * the post-commit anchor the persistence tap needs (207 §5.1). */131 writeLocal(mutations: Mutation[], onCommitted?: () => void): void {132 for (const m of mutations) {133 if (!this.localTables.has(m.table)) {134 throw new Error(`writeLocal: table "${m.table}" is not local-only — a direct commit to a synced table is reverted on the next rewind (M2).`);135 }136 }137 this.writeWith((tx) => {138 for (const m of mutations) {139 if (m.op === "add") tx.add(m.table, m.row);140 else if (m.op === "remove") tx.remove(m.table, m.row);141 else tx.edit(m.table, m.old, m.new);142 }143 }, onCommitted);144 }145146 /** Remove a synthetic aggregate table registered by {@link registerTable}, once the last147 * query reading it has been unregistered (`AGGREGATE-SYNC-DESIGN.md` §4): frees the engine148 * source + its optimistic baseline so aggregate state is reclaimed, not permanent. Throws149 * if a query still reads it (the backend refcounts readers, so it calls this only at 0). */150 unregisterTable(name: string): void {151 this.db.unregisterTable(name);152 }153154 registerQuery(qid: QueryId, ast: Ast): void {155 const r = this.db.query(qid, ast);156 this.handler(qid, { type: "hello", schema: r.schema as never, comparatorVersion: r.comparatorVersion });157 this.handler(qid, { type: "snapshot", adds: r.snapshot as never, last: true });158 }159160 unregisterQuery(qid: QueryId): void {161 this.db.destroyQuery(qid);162 }163164 mutate(mutations: Mutation[]): Promise<void> {165 const tx = this.db.write();166 for (const m of mutations) {167 if (m.op === "add") tx.add(m.table, m.row);168 else if (m.op === "remove") tx.remove(m.table, m.row);169 else tx.edit(m.table, m.old, m.new);170 }171 this.dispatch(tx.commit() as BatchSet);172 return Promise.resolve();173 }174175 onEvent(handler: (qid: QueryId, ev: ChangeEvent) => void): void {176 this.handler = handler;177 }178179 /** Register the Store's commit-boundary handler ({@link Backend.onCommitBoundary}): `dispatch`180 * calls it around each commit's per-query batch delivery so the Store can fold every affected181 * view before notifying any subscriber. */182 onCommitBoundary(handler: (phase: "begin" | "end") => void): void {183 this.boundaryHandler = handler;184 }185186 /** Deliver one commit's per-query batches. Non-reentrant (#15): if a delivery is already in187 * progress (a subscriber re-entered via write()), enqueue and let the active drain pick it188 * up FIFO, so each query folds commits in order. Per-query isolated (#11): a view's fold or189 * subscriber throwing does NOT drop sibling queries' batches — the first error is re-raised190 * only after every query has been delivered.191 *192 * Cross-view-atomic notification: each commit is bracketed with `boundaryHandler("begin"/"end")`193 * so the Store folds ALL of this commit's views before notifying ANY subscriber (a subscriber194 * re-reading a sibling view then sees post-commit data). `begin`/`end` stay balanced even if a195 * fold throws (the `finally`), and a re-entrant write enqueued during the `end` flush drains as196 * its own bracketed commit in the next loop turn — preserving per-query commit order (#15). */197 private dispatch(batches: BatchSet): void {198 this.dispatchQueue.push(batches);199 if (this.draining) return; // an outer drain is running; it will deliver this set200 this.draining = true;201 let firstError: unknown;202 let hasError = false;203 const note = (err: unknown) => {204 if (!hasError) {205 hasError = true;206 firstError = err;207 }208 };209 try {210 while (this.dispatchQueue.length) {211 const next = this.dispatchQueue.shift()!;212 if (next.length === 0) continue; // an empty commit (no query changed) — no barrier needed213 this.boundaryHandler("begin");214 try {215 for (const b of next) {216 try {217 this.handler(b.queryId, { type: "batch", events: b.events as never });218 } catch (err) {219 note(err);220 }221 }222 } finally {223 try {224 this.boundaryHandler("end"); // flush: notify every folded view's subscribers225 } catch (err) {226 note(err);227 }228 }229 }230 } finally {231 this.draining = false;232 }233 if (hasError) throw firstError;234 }235236 // --- the optimistic-cycle surface (OPTIMISTIC-WRITES-DESIGN.md §1/§3/§9) ----------237 //238 // `@rindle/optimistic` drives the engine's fork/rebase loop through these three; the239 // plain local path never calls them.240241 /** Run `f` against a raw staged write txn (with the §4.1 `get` read path), commit, and242 * dispatch the resulting batches on the ordinary event stream. Inside an open server243 * batch the engine buffers the events into the cycle instead (commit returns `[]`),244 * so re-invocations dispatch nothing here — delivery is `serverBatchEnd`'s.245 *246 * `onCommitted` runs after `tx.commit()` returns but before `dispatch` delivers to247 * subscribers: a throw from `f` (pre-commit — engine untouched) skips it; a subscriber248 * throw re-raised by `dispatch` happens after it. It is the only point where "the commit249 * is applied" is knowable to a caller that must not confuse the two failure modes. */250 writeWith(f: (tx: WasmWriteTxn) => void, onCommitted?: () => void): void {251 const tx = this.db.write();252 f(tx);253 const batches = tx.commit() as BatchSet;254 onCommitted?.();255 this.dispatch(batches);256 }257258 /** Open a §1.3 reconcile cycle against the coherent server delta: the engine rewinds every259 * tracked table (optimistic layer un-applied, delta folded in, each table's baseline re-forked)260 * and starts buffering. Re-invoke the still-pending mutators via {@link writeWith}, then call261 * {@link serverBatchEnd}. On error the rebase state is poisoned — discard the backend and262 * re-hydrate. */263 serverBatchBegin(deltas: ServerDeltaOp[]): void {264 this.db.serverBatchBegin(deltas);265 }266267 /** Close the cycle: the whole buffered stream (rewind + re-invocations) coalesces to268 * the minimal net per query (§3 — a confirmed-correct prediction delivers nothing)269 * and dispatches as ONE batch per affected query on the ordinary event stream. */270 serverBatchEnd(): void {271 this.dispatch(this.db.serverBatchEnd() as BatchSet);272 }273}274275/** Convenience: init the wasm + return a ready local {@link Store}. */276export async function createWasmStore<S extends ColsMap>(schema: Schema<S>): Promise<Store<S>> {277 await initWasm();278 return new Store(schema, new WasmBackend(schema));279}280