1import { hasSharedColumnWrites, sealSharedColumnValues } from "./shared-field-write.ts";2import type { Db, MutationErrorEvent, QueryBuilder, TableProxy } from "jazz-tools";3// Package-owned reactive table store.4import type { RuntimeDiagnostics } from "./diagnostics.ts";5import { settleDurableWrite } from "./durability.ts";67/** The minimum shape every persisted row exposes to the framework. */8export type TableRow = { id: string };910/**11 * The row type of one declared schema table — `RowOf<typeof app.schema.tasks>`.12 * Lets author code derive row types from its schema without importing the13 * vendor module, keeping UI islands on public package seams.14 */15export type RowOf<Table> = Table extends { readonly _rowType: infer Row } ? Row : never;1617/**18 * A declared schema table (`schema.<name>`). It is both an insert/update/remove19 * target and a query source, so the store accepts the intersection Jazz expects.20 * `T` is the row type and `Init` the insert type (row minus server-owned fields).21 */22export type TableHandle<T extends TableRow, Init> = TableProxy<T, Init> & QueryBuilder<T>;2324/**25 * How far the latest write has travelled: `none` before any write settles,26 * `local` once saved on this device (the resting state when sync is not27 * configured), `global` once the store confirms it, `failed` when it was28 * denied.29 */30export type WriteDurability = "none" | "local" | "global" | "failed";3132/** Reactive table state including rows and the last observed durability tier. */33export type TableSnapshot<T extends TableRow> = {34 /** `ready` once the first result arrived; an empty `rows` is not `loading`. */35 status: "loading" | "ready" | "error";36 /** The current typed rows of the table. */37 rows: T[];38 /** The latest write's deepest tier; see {@link WriteDurability}. */39 durability: WriteDurability;40 /** The latest subscription or write error message, or `null`. */41 error: string | null;42};4344type MutationHandle = {45 /** The vendor batch id, used to attribute asynchronous rejections. */46 batchId?: unknown;47 wait(options: { tier: "local" | "global" }): Promise<unknown>;48};4950// Own-batch attribution keeps a bounded window of recent writes; late51// rejections older than this are diagnostics-only.52const MAX_TRACKED_BATCHES = 128;5354/** Runtime options controlling durability waits and diagnostics notifications. */55export type TableStoreOptions = {56 /**57 * Whether managed sync is configured; gates global-durability waits. Read58 * live on every write so stores follow sync elections and stops without59 * being recreated.60 */61 syncConfigured?: () => boolean;62 /** Called whenever a diagnostics counter changes. */63 onDiagnosticsChange?: () => void;64 /**65 * Runs before each mutation reaches the database: rejects to refuse the66 * write (the schema-compatibility gate), and may resolve with a release67 * function invoked once the write settles. Absent means unguarded writes.68 */69 guardWrite?: () => Promise<(() => void) | void>;70};7172/**73 * A generic, reactive store over a single declared table. It has no knowledge of74 * the application schema beyond {@link TableRow}: callers bind it to one of their75 * own `schema.<name>` tables and read/write typed rows through it.76 */77export class TableStore<T extends TableRow, Init> {78 readonly #db: Db;79 readonly #table: TableHandle<T, Init>;80 readonly #diagnostics: RuntimeDiagnostics;81 readonly #syncConfigured: () => boolean;82 readonly #diagnosticsChanged: () => void;83 readonly #guardWrite?: () => Promise<(() => void) | void>;84 readonly #listeners = new Set<() => void>();85 readonly #stopMutationErrors: () => void;86 readonly #ownBatches = new Set<unknown>();87 #batchTracking = false;88 #vendorUnsubscribe: (() => void) | null = null;89 #writeGeneration = 0;90 #snapshot: TableSnapshot<T> = {91 status: "loading",92 rows: [],93 durability: "none",94 error: null,95 };9697 /** Creates a store over one declared Jazz table and shared runtime diagnostics. */98 constructor(99 db: Db,100 table: TableHandle<T, Init>,101 diagnostics: RuntimeDiagnostics,102 options: TableStoreOptions = {},103 ) {104 this.#db = db;105 this.#table = table;106 this.#diagnostics = diagnostics;107 this.#syncConfigured = options.syncConfigured ?? (() => false);108 this.#diagnosticsChanged = options.onDiagnosticsChange ?? (() => undefined);109 this.#guardWrite = options.guardWrite;110 // Asynchronous write rejections that are not tied to an awaited mutation111 // still have to reach diagnostics and the UI, so the store owns this112 // listener. The vendor event is database-wide, so it is filtered to this113 // store's own batches — one table's rejection must not mark every store114 // failed or multiply diagnostics counts.115 this.#stopMutationErrors = db.onMutationError((event) => {116 if (this.#ownsBatch(event)) this.reportMutationError(event);117 });118 this.#diagnostics.activeMutationListeners += 1;119 this.#diagnostics.totalMutationListeners += 1;120 this.#diagnosticsChanged();121 }122123 /** Returns the current immutable snapshot. */124 getSnapshot = (): TableSnapshot<T> => this.#snapshot;125126 /** Subscribes to snapshot changes and opens the vendor subscription on first use. */127 subscribe = (listener: () => void): () => void => {128 const subscriber = () => listener();129 this.#listeners.add(subscriber);130 this.#diagnostics.activeConsumers += 1;131 this.#diagnosticsChanged();132 if (this.#listeners.size === 1) this.#openVendorSubscription();133 this.#emit();134 let active = true;135 return () => {136 if (!active) return;137 active = false;138 // After close() the listener set is already empty; a late unsubscribe139 // must not decrement consumers a second time.140 if (!this.#listeners.delete(subscriber)) return;141 this.#diagnostics.activeConsumers -= 1;142 this.#diagnosticsChanged();143 if (this.#listeners.size === 0) this.#closeVendorSubscription();144 else this.#emit();145 };146 };147148 /** Inserts a row and waits for local durability before resolving. */149 insert(values: Init): Promise<void> {150 // The no-seal path stays fully synchronous up to #perform so pending151 // writes register before the caller's next turn — the observability the152 // ledger and tests rely on.153 const tableName = (this.#table as unknown as { _table?: string })._table;154 if (!tableName || !hasSharedColumnWrites(tableName, values as Record<string, unknown>)) {155 return this.#perform(() => this.#db.insert(this.#table, values));156 }157 return (async () => {158 const sealed = await sealSharedColumnValues(tableName, values as Record<string, unknown>);159 await this.#perform(() => this.#db.insert(this.#table, sealed as Init));160 })();161 }162163 /** Updates a row and waits for local durability before resolving. */164 update(id: string, patch: Partial<Init>): Promise<void> {165 const tableName = (this.#table as unknown as { _table?: string })._table;166 if (!tableName || !hasSharedColumnWrites(tableName, patch as Record<string, unknown>)) {167 return this.#perform(() => this.#db.update(this.#table, id, patch));168 }169 return (async () => {170 // Shared columns seal here — before the engine — because only the171 // mutation layer can fetch the row whose group column names the scope.172 const sealed = await sealSharedColumnValues(173 tableName,174 patch as Record<string, unknown>,175 () =>176 this.#db.one(177 (this.#table as unknown as { where(input: unknown): never }).where({ id }),178 ) as Promise<Record<string, unknown> | null>,179 );180 await this.#perform(() => this.#db.update(this.#table, id, sealed as Partial<Init>));181 })();182 }183184 /** Removes a row and waits for local durability before resolving. */185 async remove(id: string): Promise<void> {186 await this.#perform(() => this.#db.delete(this.#table, id));187 }188189 /**190 * Removes a row and waits for local durability before resolving.191 *192 * @deprecated Use {@link TableStore.remove} — the removal verb every other193 * mutation surface uses. This alias remains for one release.194 */195 delete(id: string): Promise<void> {196 return this.remove(id);197 }198199 // The guard runs before the mutation reaches the vendor: a refusal (the200 // schema-compatibility gate) surfaces like any write failure, and the201 // release ends this write's hold on the cross-tab write lock once the202 // write settles — the quiescence bar an upgrade swap waits on.203 async #perform(create: () => MutationHandle): Promise<void> {204 let releaseGuard: (() => void) | undefined;205 try {206 releaseGuard = (await this.#guardWrite?.()) ?? undefined;207 } catch (error) {208 this.#fail(error);209 throw error;210 }211 try {212 await this.#settle(create());213 } finally {214 releaseGuard?.();215 }216 }217218 /** Projects an asynchronous Jazz mutation rejection into diagnostics and store state. */219 reportMutationError(event: MutationErrorEvent): void {220 this.#diagnostics.mutationErrors += 1;221 this.#diagnosticsChanged();222 this.#fail(new Error(`${event.code}: ${event.reason}`));223 }224225 /** Releases vendor subscriptions and mutation listeners owned by this store. */226 close(): void {227 this.#diagnostics.activeConsumers -= this.#listeners.size;228 this.#listeners.clear();229 this.#closeVendorSubscription();230 this.#stopMutationErrors();231 this.#diagnostics.activeMutationListeners -= 1;232 this.#diagnosticsChanged();233 }234235 #ownsBatch(event: MutationErrorEvent): boolean {236 const batchId = event.batch?.batchId;237 // Attribution is only possible when the vendor exposes batch ids on both238 // sides; without them, surface every event rather than silence real errors.239 if (batchId === undefined || !this.#batchTracking) return true;240 return this.#ownBatches.has(batchId);241 }242243 #trackBatch(mutation: MutationHandle): void {244 if (mutation.batchId === undefined) return;245 this.#batchTracking = true;246 this.#ownBatches.add(mutation.batchId);247 if (this.#ownBatches.size > MAX_TRACKED_BATCHES) {248 this.#ownBatches.delete(this.#ownBatches.values().next().value);249 }250 }251252 #updateDiagnostics = (update: (diagnostics: RuntimeDiagnostics) => void): void => {253 update(this.#diagnostics);254 this.#diagnosticsChanged();255 };256257 async #settle(mutation: MutationHandle): Promise<void> {258 const generation = ++this.#writeGeneration;259 this.#trackBatch(mutation);260 // A store that has not received its first query result stays loading —261 // a mutation must not present an empty table as ready.262 this.#set({263 status: this.#snapshot.status === "loading" ? "loading" : "ready",264 durability: "none",265 error: null,266 });267 try {268 // Generation guards keep a superseded write's late outcome — success or269 // rejection — diagnosable without overwriting the newer write's state.270 await settleDurableWrite(271 mutation,272 this.#updateDiagnostics,273 this.#syncConfigured() ? "background" : "none",274 {275 onLocal: () => {276 if (generation === this.#writeGeneration) {277 this.#set({ durability: "local", error: null });278 }279 },280 onGlobal: () => {281 if (generation === this.#writeGeneration) {282 this.#set({ durability: "global", error: null });283 }284 },285 onGlobalError: (error) => {286 if (generation === this.#writeGeneration) this.#fail(error);287 },288 },289 );290 } catch (error) {291 this.#fail(error);292 throw error;293 }294 }295296 #openVendorSubscription(): void {297 if (this.#vendorUnsubscribe) return;298 try {299 this.#vendorUnsubscribe = this.#db.subscribeAll(this.#table, (delta) => {300 this.#snapshot = {301 ...this.#snapshot,302 status: "ready",303 rows: delta.all,304 error: null,305 };306 this.#emit();307 });308 this.#diagnostics.activeVendorSubscriptions += 1;309 this.#diagnostics.totalVendorSubscriptions += 1;310 this.#diagnosticsChanged();311 } catch (error) {312 this.#fail(error);313 }314 }315316 #closeVendorSubscription(): void {317 if (!this.#vendorUnsubscribe) return;318 this.#vendorUnsubscribe();319 this.#diagnostics.unsubscribeCalls += 1;320 this.#vendorUnsubscribe = null;321 this.#diagnostics.activeVendorSubscriptions -= 1;322 this.#diagnosticsChanged();323 }324325 #set(change: Partial<TableSnapshot<T>>): void {326 this.#snapshot = { ...this.#snapshot, ...change };327 if (change.durability) {328 this.#diagnostics.lastWriteDurability = change.durability;329 this.#diagnosticsChanged();330 }331 this.#emit();332 }333334 #fail(error: unknown): void {335 this.#snapshot = {336 ...this.#snapshot,337 status: "error",338 durability: "failed",339 error: error instanceof Error ? error.message : String(error),340 };341 this.#diagnostics.lastWriteDurability = "failed";342 this.#diagnosticsChanged();343 this.#emit();344 }345346 #emit(): void {347 for (const listener of this.#listeners) listener();348 }349}350351export function createTableStore<T extends TableRow, Init>(352 db: Db,353 table: TableHandle<T, Init>,354 diagnostics: RuntimeDiagnostics,355 options: TableStoreOptions = {},356): TableStore<T, Init> {357 return new TableStore<T, Init>(db, table, diagnostics, options);358}