API index and search · Build metadata
Source snapshot
rust/rindle-replica-node/src/index.ts
1// @rindle/replica — the native (napi-rs) backend for @rindle/client: the SQLite-backed2// `rindle-replica` live-query engine, in-process. It speaks the same `Backend` seam as3// @rindle/wasm, so the SAME Store/ArrayView drives it — only the engine differs (native4// SQLite with read-only batch-overlay derivation instead of the wasm memory engine). Re-exports @rindle/client5// so a local app can import everything from here.6//7// Unlike @rindle/wasm there is no async init: the native addon loads synchronously when this8// module is imported (`../index.js` is the napi loader that `require`s the `.node`).910import { Db } from "../index.js";11import { Store, tableSpec } from "@rindle/client";12import type { Ast, Backend, ChangeEvent, ColsMap, Mutation, QueryId, Schema } from "@rindle/client";1314export * from "@rindle/client";1516// The raw native `Db` (register/query/commit + the normalized `queryNormalized`/17// `commitNormalized`) — the low-level handle a normalized server source wraps.18export { Db as NativeReplicaDb } from "../index.js";1920// One server-side mutation's open transaction (`Db.beginMutation`) — what a server21// mutator registry runs against (exec/queryRows; commitWithLmid lands effects + lmid).22export type { NativeMutationTxn } from "../index.js";2324// The CLUSTER-backed, async-push engine (CLUSTER-FOLD-IN-DESIGN.md WS6): `commit*` returns25// `{cv}` immediately; per-query batches / per-connection progress frames / faults arrive on26// `onEvent`. Used by the optimistic server to derive queries in parallel across worker threads.27export { ClusterDb as NativeClusterDb } from "../index.js";2829// One cluster-backed mutation's open transaction (`ClusterDb.beginMutation`).30export type { ClusterMutationTxn } from "../index.js";3132// The native `Db` surface this backend uses (kept local so it doesn't depend on the33// generated `.d.ts` shape, which types the JSON returns as `any`).34interface NativeDb {35 registerTable(name: string, columns: string[], primaryKey: number[], columnTypes: string[]): void;36 query(queryId: number, astJson: string): { comparatorVersion: number; schema: unknown; snapshot: unknown[] };37 destroyQuery(queryId: number): void;38 commit(mutations: unknown): Array<{ queryId: number; events: unknown[] }>;39}4041/** One engine commit's per-query batches (the shape `NativeDb.commit` returns) — identical to the42 * wasm engine's, so the same `dispatch` bracket gives the same cross-view-atomic guarantee. */43type BatchSet = Array<{ queryId: number; events: unknown[] }>;4445/** The in-process, SQLite-backed native backend. Each instance owns its own replica (a46 * temporary WAL database). Registers every schema table on construction. This wrapper has no47 * database-path option and does not persist application data across process restarts. */48export class ReplicaBackend<S extends ColsMap> implements Backend {49 private readonly db: NativeDb;50 private handler: (qid: QueryId, ev: ChangeEvent) => void = () => {};51 // The Store's commit-boundary handler ({@link Backend.onCommitBoundary}): `dispatch` brackets each52 // commit's per-query batch delivery with `begin`/`end` so the Store folds every affected view53 // before notifying any subscriber (cross-view-atomic notification). `ReplicaBackend` has the SAME54 // `commit() → per-query BatchSet` fan-out as the wasm engine, so it needs the SAME barrier — a55 // subscriber re-reading a sibling view would otherwise observe that sibling's pre-commit state.56 // Defaults to a no-op so a Store that never registers it just gets per-event notification.57 private boundaryHandler: (phase: "begin" | "end") => void = () => {};5859 // Non-reentrant delivery (mirrors `WasmBackend`): a subscriber that synchronously triggers another60 // write enqueues that commit's batches; they drain only after the in-flight commit's batches61 // finish, preserving per-query commit order regardless of what subscribers do.62 private readonly dispatchQueue: BatchSet[] = [];63 private draining = false;6465 constructor(schema: Schema<S>) {66 this.db = new Db() as unknown as NativeDb;67 for (const name of Object.keys(schema.tables)) {68 const meta = schema.tables[name];69 const { columns, primaryKey } = tableSpec(meta);70 const cols = meta.columns as unknown as Record<string, { type: string }>;71 const columnTypes = columns.map((c) => cols[c].type);72 this.db.registerTable(name, columns, primaryKey, columnTypes);73 }74 }7576 registerQuery(qid: QueryId, ast: Ast): void {77 const r = this.db.query(qid, JSON.stringify(ast));78 this.handler(qid, { type: "hello", schema: r.schema as never, comparatorVersion: r.comparatorVersion });79 this.handler(qid, { type: "snapshot", adds: r.snapshot as never, last: true });80 }8182 unregisterQuery(qid: QueryId): void {83 this.db.destroyQuery(qid);84 }8586 mutate(mutations: Mutation[]): Promise<void> {87 this.dispatch(this.db.commit(mutations) as BatchSet);88 return Promise.resolve();89 }9091 onEvent(handler: (qid: QueryId, ev: ChangeEvent) => void): void {92 this.handler = handler;93 }9495 /** Register the Store's commit-boundary handler ({@link Backend.onCommitBoundary}): `dispatch`96 * calls it around each commit's per-query batch delivery so the Store can fold every affected97 * view before notifying any subscriber. */98 onCommitBoundary(handler: (phase: "begin" | "end") => void): void {99 this.boundaryHandler = handler;100 }101102 /** Deliver one commit's per-query batches — a faithful mirror of `WasmBackend.dispatch`.103 * Non-reentrant: if a delivery is already in progress (a subscriber re-entered via `write()`),104 * enqueue and let the active drain pick it up FIFO, so each query folds commits in order.105 * Per-query isolated: a view's fold or subscriber throwing does NOT drop sibling queries'106 * batches — the first error is re-raised only after every query has been delivered.107 *108 * Cross-view-atomic notification: each commit is bracketed with `boundaryHandler("begin"/"end")`109 * so the Store folds ALL of this commit's views before notifying ANY subscriber (a subscriber110 * re-reading a sibling view then sees post-commit data). `begin`/`end` stay balanced even if a111 * fold throws (the `finally`), and a re-entrant write enqueued during the `end` flush drains as112 * its own bracketed commit in the next loop turn. */113 private dispatch(batches: BatchSet): void {114 this.dispatchQueue.push(batches);115 if (this.draining) return; // an outer drain is running; it will deliver this set116 this.draining = true;117 let firstError: unknown;118 let hasError = false;119 const note = (err: unknown) => {120 if (!hasError) {121 hasError = true;122 firstError = err;123 }124 };125 try {126 while (this.dispatchQueue.length) {127 const next = this.dispatchQueue.shift()!;128 if (next.length === 0) continue; // an empty commit (no query changed) — no barrier needed129 this.boundaryHandler("begin");130 try {131 for (const b of next) {132 try {133 this.handler(b.queryId, { type: "batch", events: b.events as never });134 } catch (err) {135 note(err);136 }137 }138 } finally {139 try {140 this.boundaryHandler("end"); // flush: notify every folded view's subscribers141 } catch (err) {142 note(err);143 }144 }145 }146 } finally {147 this.draining = false;148 }149 if (hasError) throw firstError;150 }151}152153/** Create a local {@link Store} over a fresh temporary SQLite database. The schema creates the154 * tables; there is no file-path or restart-persistence option. Destroy materialized views when155 * finished. The native replica's temporary files are removed when its handle is finalized. */156export function createReplicaStore<S extends ColsMap>(schema: Schema<S>): Store<S> {157 return new Store(schema, new ReplicaBackend(schema));158}159