1import type { Db, MutationErrorEvent, QueryBuilder, TableProxy } from "jazz-tools";2// Package-owned reactive table store.3import type { RuntimeDiagnostics } from "./diagnostics.ts";4import { settleDurableWrite } from "./durability.ts";56/** The minimum shape every persisted row exposes to the framework. */7export type TableRow = { id: string };89/**10 * The row type of one declared schema table — `RowOf<typeof app.schema.tasks>`.11 * Lets author code derive row types from its schema without importing the12 * vendor module, keeping UI islands on public package seams.13 */14export type RowOf<Table> = Table extends { readonly _rowType: infer Row } ? Row : never;1516/**17 * A declared schema table (`schema.<name>`). It is both an insert/update/delete18 * target and a query source, so the store accepts the intersection Jazz expects.19 * `T` is the row type and `Init` the insert type (row minus server-owned fields).20 */21export type TableHandle<T extends TableRow, Init> = TableProxy<T, Init> & QueryBuilder<T>;2223/** Reactive table state including rows and the last observed durability tier. */24export type TableSnapshot<T extends TableRow> = {25 status: "loading" | "ready" | "error";26 rows: T[];27 durability: "none" | "local" | "global" | "failed";28 error: string | null;29};3031type Listener = () => void;32type MutationHandle = {33 /** The vendor batch id, used to attribute asynchronous rejections. */34 batchId?: unknown;35 wait(options: { tier: "local" | "global" }): Promise<unknown>;36};3738// Own-batch attribution keeps a bounded window of recent writes; late39// rejections older than this are diagnostics-only.40const MAX_TRACKED_BATCHES = 128;4142/** Runtime options controlling durability waits and diagnostics notifications. */43export type TableStoreOptions = {44 /**45 * Whether managed sync is configured; gates global-durability waits. Read46 * live on every write so stores follow sync elections and stops without47 * being recreated.48 */49 syncConfigured?: () => boolean;50 /** Called whenever a diagnostics counter changes. */51 onDiagnosticsChange?: () => void;52};5354/**55 * A generic, reactive store over a single declared table. It has no knowledge of56 * the application schema beyond {@link TableRow}: callers bind it to one of their57 * own `schema.<name>` tables and read/write typed rows through it.58 */59export class TableStore<T extends TableRow, Init> {60 readonly #db: Db;61 readonly #table: TableHandle<T, Init>;62 readonly #diagnostics: RuntimeDiagnostics;63 readonly #syncConfigured: () => boolean;64 readonly #diagnosticsChanged: () => void;65 readonly #listeners = new Set<Listener>();66 readonly #stopMutationErrors: () => void;67 readonly #ownBatches = new Set<unknown>();68 #batchTracking = false;69 #vendorUnsubscribe: (() => void) | null = null;70 #writeGeneration = 0;71 #snapshot: TableSnapshot<T> = {72 status: "loading",73 rows: [],74 durability: "none",75 error: null,76 };7778 /** Creates a store over one declared Jazz table and shared runtime diagnostics. */79 constructor(80 db: Db,81 table: TableHandle<T, Init>,82 diagnostics: RuntimeDiagnostics,83 options: TableStoreOptions = {},84 ) {85 this.#db = db;86 this.#table = table;87 this.#diagnostics = diagnostics;88 this.#syncConfigured = options.syncConfigured ?? (() => false);89 this.#diagnosticsChanged = options.onDiagnosticsChange ?? (() => undefined);90 // Asynchronous write rejections that are not tied to an awaited mutation91 // still have to reach diagnostics and the UI, so the store owns this92 // listener. The vendor event is database-wide, so it is filtered to this93 // store's own batches — one table's rejection must not mark every store94 // failed or multiply diagnostics counts.95 this.#stopMutationErrors = db.onMutationError((event) => {96 if (this.#ownsBatch(event)) this.reportMutationError(event);97 });98 this.#diagnostics.activeMutationListeners += 1;99 this.#diagnostics.totalMutationListeners += 1;100 this.#diagnosticsChanged();101 }102103 /** Returns the current immutable snapshot. */104 getSnapshot = (): TableSnapshot<T> => this.#snapshot;105106 /** Subscribes to snapshot changes and opens the vendor subscription on first use. */107 subscribe = (listener: Listener): () => void => {108 const subscriber = () => listener();109 this.#listeners.add(subscriber);110 this.#diagnostics.activeConsumers += 1;111 this.#diagnosticsChanged();112 if (this.#listeners.size === 1) this.#openVendorSubscription();113 this.#emit();114 let active = true;115 return () => {116 if (!active) return;117 active = false;118 // After close() the listener set is already empty; a late unsubscribe119 // must not decrement consumers a second time.120 if (!this.#listeners.delete(subscriber)) return;121 this.#diagnostics.activeConsumers -= 1;122 this.#diagnosticsChanged();123 if (this.#listeners.size === 0) this.#closeVendorSubscription();124 else this.#emit();125 };126 };127128 /** Inserts a row and waits for local durability before resolving. */129 async insert(values: Init): Promise<void> {130 await this.#settle(this.#db.insert(this.#table, values));131 }132133 /** Updates a row and waits for local durability before resolving. */134 async update(id: string, patch: Partial<Init>): Promise<void> {135 await this.#settle(this.#db.update(this.#table, id, patch));136 }137138 /** Deletes a row and waits for local durability before resolving. */139 async delete(id: string): Promise<void> {140 await this.#settle(this.#db.delete(this.#table, id));141 }142143 /** Projects an asynchronous Jazz mutation rejection into diagnostics and store state. */144 reportMutationError(event: MutationErrorEvent): void {145 this.#diagnostics.mutationErrors += 1;146 this.#diagnosticsChanged();147 this.#fail(new Error(`${event.code}: ${event.reason}`));148 }149150 /** Releases vendor subscriptions and mutation listeners owned by this store. */151 close(): void {152 this.#diagnostics.activeConsumers -= this.#listeners.size;153 this.#listeners.clear();154 this.#closeVendorSubscription();155 this.#stopMutationErrors();156 this.#diagnostics.activeMutationListeners -= 1;157 this.#diagnosticsChanged();158 }159160 #ownsBatch(event: MutationErrorEvent): boolean {161 const batchId = event.batch?.batchId;162 // Attribution is only possible when the vendor exposes batch ids on both163 // sides; without them, surface every event rather than silence real errors.164 if (batchId === undefined || !this.#batchTracking) return true;165 return this.#ownBatches.has(batchId);166 }167168 #trackBatch(mutation: MutationHandle): void {169 if (mutation.batchId === undefined) return;170 this.#batchTracking = true;171 this.#ownBatches.add(mutation.batchId);172 if (this.#ownBatches.size > MAX_TRACKED_BATCHES) {173 this.#ownBatches.delete(this.#ownBatches.values().next().value);174 }175 }176177 #updateDiagnostics = (update: (diagnostics: RuntimeDiagnostics) => void): void => {178 update(this.#diagnostics);179 this.#diagnosticsChanged();180 };181182 async #settle(mutation: MutationHandle): Promise<void> {183 const generation = ++this.#writeGeneration;184 this.#trackBatch(mutation);185 // A store that has not received its first query result stays loading —186 // a mutation must not present an empty table as ready.187 this.#set({188 status: this.#snapshot.status === "loading" ? "loading" : "ready",189 durability: "none",190 error: null,191 });192 try {193 // Generation guards keep a superseded write's late outcome — success or194 // rejection — diagnosable without overwriting the newer write's state.195 await settleDurableWrite(196 mutation,197 this.#updateDiagnostics,198 this.#syncConfigured() ? "background" : "none",199 {200 onLocal: () => {201 if (generation === this.#writeGeneration) {202 this.#set({ durability: "local", error: null });203 }204 },205 onGlobal: () => {206 if (generation === this.#writeGeneration) {207 this.#set({ durability: "global", error: null });208 }209 },210 onGlobalError: (error) => {211 if (generation === this.#writeGeneration) this.#fail(error);212 },213 },214 );215 } catch (error) {216 this.#fail(error);217 throw error;218 }219 }220221 #openVendorSubscription(): void {222 if (this.#vendorUnsubscribe) return;223 try {224 this.#vendorUnsubscribe = this.#db.subscribeAll(this.#table, (delta) => {225 this.#snapshot = {226 ...this.#snapshot,227 status: "ready",228 rows: delta.all,229 error: null,230 };231 this.#emit();232 });233 this.#diagnostics.activeVendorSubscriptions += 1;234 this.#diagnostics.totalVendorSubscriptions += 1;235 this.#diagnosticsChanged();236 } catch (error) {237 this.#fail(error);238 }239 }240241 #closeVendorSubscription(): void {242 if (!this.#vendorUnsubscribe) return;243 this.#vendorUnsubscribe();244 this.#diagnostics.unsubscribeCalls += 1;245 this.#vendorUnsubscribe = null;246 this.#diagnostics.activeVendorSubscriptions -= 1;247 this.#diagnosticsChanged();248 }249250 #set(change: Partial<TableSnapshot<T>>): void {251 this.#snapshot = { ...this.#snapshot, ...change };252 if (change.durability) {253 this.#diagnostics.lastWriteDurability = change.durability;254 this.#diagnosticsChanged();255 }256 this.#emit();257 }258259 #fail(error: unknown): void {260 this.#snapshot = {261 ...this.#snapshot,262 status: "error",263 durability: "failed",264 error: error instanceof Error ? error.message : String(error),265 };266 this.#diagnostics.lastWriteDurability = "failed";267 this.#diagnosticsChanged();268 this.#emit();269 }270271 #emit(): void {272 for (const listener of this.#listeners) listener();273 }274}275276export function createTableStore<T extends TableRow, Init>(277 db: Db,278 table: TableHandle<T, Init>,279 diagnostics: RuntimeDiagnostics,280 options: TableStoreOptions = {},281): TableStore<T, Init> {282 return new TableStore<T, Init>(db, table, diagnostics, options);283}