1import type { Db, MutationErrorEvent, TableProxy } from "jazz-tools";2import { syncing } from "./config.ts";3import type { RuntimeDiagnostics } from "./diagnostics.ts";4import { settleDurableWrite } from "./durability.ts";5import { getRuntime, runtimeRecreatedEvent, updateRuntimeDiagnostics } from "./runtime.ts";6import type { TableRow } from "./table-store.ts";78/** Observable state shared by every mutation consumer for one table. */9export type TableMutationSnapshot = {10 pending: number;11 durability: "none" | "local" | "global" | "failed";12 error: string | null;13};1415type Listener = () => void;16type MutationHandle<T> = {17 /** The vendor batch id, used to attribute asynchronous rejections. */18 batchId?: unknown;19 wait(options: { tier: "local" | "global" }): Promise<T>;20};2122// Own-batch attribution keeps a bounded window of recent writes; late23// rejections older than this are diagnostics-only.24const MAX_TRACKED_BATCHES = 128;2526/** Runtime seams used by table-mutation tests and the package-wide registry. */27export type TableMutationEnvironment = {28 getDb(): Promise<Db>;29 syncConfigured(): boolean;30 subscribeRuntimeRecreation(listener: () => void): () => void;31 updateDiagnostics(update: (diagnostics: RuntimeDiagnostics) => void): void;32};3334/** Framework-neutral typed mutations and observable durability for one table. */35export class TableMutationStore<T extends TableRow, Init> {36 readonly #table: TableProxy<T, Init>;37 readonly #environment: TableMutationEnvironment;38 readonly #onIdle: () => void;39 readonly #listeners = new Set<Listener>();40 readonly #ownBatches = new Set<unknown>();41 #batchTracking = false;42 #snapshot: TableMutationSnapshot = { pending: 0, durability: "none", error: null };43 #stopMutationErrors: (() => void) | null = null;44 #stopRuntimeRecreation: (() => void) | null = null;45 #generation = 0;46 #writeGeneration = 0;47 #disposed = false;4849 /** Creates one registry-owned mutation store. */50 constructor(51 table: TableProxy<T, Init>,52 environment: TableMutationEnvironment,53 onIdle: () => void,54 ) {55 this.#table = table;56 this.#environment = environment;57 this.#onIdle = onIdle;58 }5960 /** Returns current pending, durability, and rejection state. */61 getSnapshot = (): TableMutationSnapshot => this.#snapshot;6263 /** Retains the table-scoped error listener for one consumer. */64 subscribe = (listener: Listener): () => void => {65 if (this.#disposed) throw new Error("table mutation store has been released");66 const subscriber = () => listener();67 this.#listeners.add(subscriber);68 if (this.#listeners.size === 1) this.#start();69 listener();70 let active = true;71 return () => {72 if (!active) return;73 active = false;74 this.#listeners.delete(subscriber);75 if (this.#listeners.size === 0) {76 this.#stop();77 this.#onIdle();78 }79 };80 };8182 /** Number of mounted consumers sharing this table-scoped mutation state. */83 get consumerCount(): number {84 return this.#listeners.size;85 }8687 /** Inserts and returns the typed created row after local durability. */88 async insert(values: Init): Promise<T> {89 return await this.#perform((db) => db.insert(this.#table, values));90 }9192 /** Updates a row and resolves after local durability. */93 async update(id: string, patch: Partial<Init>): Promise<void> {94 await this.#perform((db) => db.update(this.#table, id, patch));95 }9697 /** Deletes a row and resolves after local durability. */98 async remove(id: string): Promise<void> {99 await this.#perform((db) => db.delete(this.#table, id));100 }101102 /** Releases listeners owned by an evicted or hot-reloaded store. */103 dispose(): void {104 if (this.#disposed) return;105 this.#disposed = true;106 this.#listeners.clear();107 this.#stop();108 }109110 async #perform<Result>(create: (db: Db) => MutationHandle<Result>): Promise<Result> {111 if (this.#disposed) throw new Error("table mutation store has been released");112 const db = await this.#environment.getDb();113 let mutation: MutationHandle<Result>;114 try {115 mutation = create(db);116 } catch (error) {117 this.#updateDiagnostics((diagnostics) => diagnostics.mutationErrors += 1);118 this.#fail(error);119 throw error;120 }121 return await this.#settle(mutation);122 }123124 #ownsBatch(event: MutationErrorEvent): boolean {125 const batchId = event.batch?.batchId;126 // Attribution is only possible when the vendor exposes batch ids on both127 // sides; without them, surface every event rather than silence real errors.128 if (batchId === undefined || !this.#batchTracking) return true;129 return this.#ownBatches.has(batchId);130 }131132 #trackBatch<Result>(mutation: MutationHandle<Result>): void {133 if (mutation.batchId === undefined) return;134 this.#batchTracking = true;135 this.#ownBatches.add(mutation.batchId);136 if (this.#ownBatches.size > MAX_TRACKED_BATCHES) {137 this.#ownBatches.delete(this.#ownBatches.values().next().value);138 }139 }140141 async #settle<Result>(mutation: MutationHandle<Result>): Promise<Result> {142 const generation = ++this.#writeGeneration;143 this.#trackBatch(mutation);144 this.#set({ pending: this.#snapshot.pending + 1, durability: "none", error: null });145 try {146 // Generation guards keep a superseded write's late outcome — success or147 // rejection — diagnosable without overwriting the newer write's state.148 return await settleDurableWrite(149 mutation,150 (update) => this.#updateDiagnostics(update),151 this.#environment.syncConfigured() ? "background" : "none",152 {153 onLocal: () =>154 this.#set({155 pending: Math.max(0, this.#snapshot.pending - 1),156 ...(generation === this.#writeGeneration ? { durability: "local" as const } : {}),157 }),158 onGlobal: () => {159 if (generation === this.#writeGeneration) {160 this.#set({ durability: "global", error: null });161 }162 },163 onGlobalError: (error) => {164 if (generation === this.#writeGeneration) this.#fail(error);165 },166 },167 );168 } catch (error) {169 this.#set({ pending: Math.max(0, this.#snapshot.pending - 1) });170 this.#fail(error);171 throw error;172 }173 }174175 #start(): void {176 this.#stopRuntimeRecreation = this.#environment.subscribeRuntimeRecreation(() =>177 this.#attach()178 );179 this.#attach();180 }181182 #attach(): void {183 const generation = ++this.#generation;184 this.#detachMutationErrors();185 void this.#environment.getDb().then((db) => {186 if (this.#disposed || generation !== this.#generation || this.#listeners.size === 0) return;187 let stop: () => void;188 try {189 stop = db.onMutationError((event) => this.#reportMutationError(event, generation));190 } catch (error) {191 this.#fail(error, generation);192 return;193 }194 if (this.#disposed || generation !== this.#generation || this.#listeners.size === 0) {195 stop();196 return;197 }198 this.#stopMutationErrors = stop;199 this.#updateDiagnostics((diagnostics) => {200 diagnostics.activeMutationListeners += 1;201 diagnostics.totalMutationListeners += 1;202 });203 }, (error) => this.#fail(error, generation));204 }205206 #reportMutationError(event: MutationErrorEvent, generation: number): void {207 if (generation !== this.#generation) return;208 // The vendor event is database-wide; only this store's own batches may209 // mark it failed or count against its diagnostics.210 if (!this.#ownsBatch(event)) return;211 this.#updateDiagnostics((diagnostics) => diagnostics.mutationErrors += 1);212 this.#fail(new Error(`${event.code}: ${event.reason}`));213 }214215 #stop(): void {216 this.#generation += 1;217 this.#detachMutationErrors();218 this.#stopRuntimeRecreation?.();219 this.#stopRuntimeRecreation = null;220 }221222 #detachMutationErrors(): void {223 if (!this.#stopMutationErrors) return;224 this.#stopMutationErrors();225 this.#stopMutationErrors = null;226 this.#updateDiagnostics((diagnostics) => diagnostics.activeMutationListeners -= 1);227 }228229 #fail(error: unknown, generation = this.#generation): void {230 if (this.#disposed || generation !== this.#generation) return;231 this.#set({232 durability: "failed",233 error: error instanceof Error ? error.message : String(error),234 });235 }236237 #set(change: Partial<TableMutationSnapshot>): void {238 this.#snapshot = { ...this.#snapshot, ...change };239 if (change.durability) {240 this.#updateDiagnostics((diagnostics) =>241 diagnostics.lastWriteDurability = change.durability!242 );243 }244 for (const listener of this.#listeners) listener();245 }246247 #updateDiagnostics(update: (diagnostics: RuntimeDiagnostics) => void): void {248 this.#environment.updateDiagnostics(update);249 }250}251252/** Retained ownership of one shared table-mutation store. */253export type TableMutationLease<T extends TableRow, Init> = {254 store: TableMutationStore<T, Init>;255 release(): void;256};257258type Entry = { store: TableMutationStore<TableRow, unknown>; leases: number };259260/** Shares mutation state and one vendor error listener per schema table. */261export class TableMutationRegistry {262 readonly #environment: TableMutationEnvironment;263 readonly #schemas = new WeakMap<object, Map<string, Entry>>();264 readonly #stores = new Set<TableMutationStore<TableRow, unknown>>();265266 /** Creates an isolated registry over one runtime environment. */267 constructor(environment: TableMutationEnvironment) {268 this.#environment = environment;269 }270271 /** Acquires the shared mutation surface for a typed table. */272 acquire<T extends TableRow, Init>(table: TableProxy<T, Init>): TableMutationLease<T, Init> {273 const schema = table._schema as object;274 let entries = this.#schemas.get(schema);275 if (!entries) {276 entries = new Map();277 this.#schemas.set(schema, entries);278 }279 let entry = entries.get(table._table);280 if (!entry) {281 const store = new TableMutationStore(282 table,283 this.#environment,284 () => this.#evictIfUnused(entries!, table._table),285 );286 entry = { store: store as TableMutationStore<TableRow, unknown>, leases: 0 };287 entries.set(table._table, entry);288 this.#stores.add(entry.store);289 }290 entry.leases += 1;291 let retained = true;292 return {293 store: entry.store as TableMutationStore<T, Init>,294 release: () => {295 if (!retained) return;296 retained = false;297 entry!.leases -= 1;298 this.#evictIfUnused(entries!, table._table);299 },300 };301 }302303 /** Disposes all active stores during hot replacement. */304 dispose(): void {305 for (const store of this.#stores) store.dispose();306 this.#stores.clear();307 }308309 #evictIfUnused(entries: Map<string, Entry>, key: string): void {310 const entry = entries.get(key);311 if (!entry || entry.leases > 0 || entry.store.consumerCount > 0) return;312 entries.delete(key);313 this.#stores.delete(entry.store);314 entry.store.dispose();315 }316}317318const defaultRegistry = new TableMutationRegistry({319 getDb: async () => (await getRuntime()).db,320 syncConfigured: syncing,321 subscribeRuntimeRecreation(listener) {322 globalThis.addEventListener(runtimeRecreatedEvent, listener);323 return () => globalThis.removeEventListener(runtimeRecreatedEvent, listener);324 },325 updateDiagnostics: updateRuntimeDiagnostics,326});327328/** Acquires the package-wide typed mutation surface for one table. */329export function acquireTableMutations<T extends TableRow, Init>(330 table: TableProxy<T, Init>,331): TableMutationLease<T, Init> {332 return defaultRegistry.acquire(table);333}334335import.meta.hot?.dispose(() => defaultRegistry.dispose());