1import { sealSharedColumnValuesSync } from "./shared-field-write.ts";2import type { Db, MutationErrorEvent, TableProxy } from "jazz-tools";3import { syncing } from "./config.ts";4import type { RuntimeDiagnostics } from "./diagnostics.ts";5import { getRuntime, runtimeRecreatedEvent, updateRuntimeDiagnostics } from "./runtime.ts";6import type { TableRow, WriteDurability } from "./table-store.ts";7import type { WriteHandle } from "./write-handle.ts";8import { getWriteLedger, type WriteLedger } from "./write-ledger.ts";910/** Observable state shared by every mutation consumer for one table. */11export type TableMutationSnapshot = {12 /** Writes on this table still awaiting their sync fate. */13 pending: number;14 /** The latest write's deepest tier; see {@link WriteDurability}. */15 durability: WriteDurability;16 /** The latest table-scoped rejection message, or `null`. */17 error: string | null;18};1920// Own-batch attribution keeps a bounded window of recent writes; late21// rejections older than this are diagnostics-only.22const MAX_TRACKED_BATCHES = 128;2324/** Runtime seams used by table-mutation tests and the package-wide registry. */25export type TableMutationEnvironment = {26 getDb(): Promise<Db>;27 syncConfigured(): boolean;28 /** The per-write ledger table mutations route through. */29 getLedger(): WriteLedger;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<() => void>();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: () => void): () => 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 /**88 * Inserts a row. Awaiting the returned handle resolves with the typed89 * created row at `saved`; the handle's stage promises track the sync fate.90 */91 insert(values: Init): WriteHandle<T> {92 return this.#track(() =>93 this.#environment.getLedger().perform<T>({94 kind: "insert",95 table: this.#table as TableProxy<unknown, unknown>,96 values: this.#sealSharedColumns(values as Record<string, unknown>) as Init,97 })98 );99 }100101 /** Updates a row. Awaiting the returned handle resolves at `saved`.102 * A patch touching a shared encrypted column must include the row's group103 * column — the verb path seals synchronously before journaling and cannot104 * fetch the row. */105 update(id: string, patch: Partial<Init>): WriteHandle<void> {106 return this.#track(() =>107 this.#environment.getLedger().perform<void>({108 kind: "update",109 table: this.#table as TableProxy<unknown, unknown>,110 id,111 patch: this.#sealSharedColumns(patch as Record<string, unknown>),112 })113 );114 }115116 // Sealing precedes journaling so the durable journal holds ciphertext and117 // replayed writes stay sealed.118 #sealSharedColumns(values: Record<string, unknown>): Record<string, unknown> {119 const tableName = (this.#table as unknown as { _table?: string })._table;120 if (!tableName) return values;121 return sealSharedColumnValuesSync(tableName, values);122 }123124 /** Deletes a row. Awaiting the returned handle resolves at `saved`. */125 remove(id: string): WriteHandle<void> {126 return this.#track(() =>127 this.#environment.getLedger().perform<void>({128 kind: "remove",129 table: this.#table as TableProxy<unknown, unknown>,130 id,131 })132 );133 }134135 /** Releases listeners owned by an evicted or hot-reloaded store. */136 dispose(): void {137 if (this.#disposed) return;138 this.#disposed = true;139 this.#listeners.clear();140 this.#stop();141 }142143 #track<Result>(start: () => WriteHandle<Result>): WriteHandle<Result> {144 if (this.#disposed) throw new Error("table mutation store has been released");145 const generation = ++this.#writeGeneration;146 let handle: WriteHandle<Result>;147 try {148 // A refusal — the ledger's write guard (schema-compatibility gate) or a149 // current-policy violation — leaves through this call with no journal150 // entry; the ledger has already counted it.151 handle = start();152 } catch (error) {153 this.#fail(error);154 throw error;155 }156 this.#trackBatch(handle.batchId);157 this.#set({ pending: this.#snapshot.pending + 1, durability: "none", error: null });158 // Generation guards keep a superseded write's late outcome — success or159 // rejection — diagnosable without overwriting the newer write's state.160 handle.saved.then(161 () => {162 this.#trackBatch(handle.batchId);163 this.#set({164 pending: Math.max(0, this.#snapshot.pending - 1),165 ...(generation === this.#writeGeneration ? { durability: "local" as const } : {}),166 });167 },168 (error) => {169 this.#set({ pending: Math.max(0, this.#snapshot.pending - 1) });170 if (generation === this.#writeGeneration) this.#fail(error);171 },172 );173 handle.synced.then(174 () => {175 // Without configured sync the ledger settles at local durability; the176 // table-scoped durability label stays `local` in that mode.177 if (generation === this.#writeGeneration && this.#environment.syncConfigured()) {178 this.#set({ durability: "global", error: null });179 }180 },181 (error) => {182 if (generation === this.#writeGeneration) this.#fail(error);183 },184 );185 return handle;186 }187188 #ownsBatch(event: MutationErrorEvent): boolean {189 const batchId = event.batch?.batchId;190 // Attribution is only possible when the vendor exposes batch ids on both191 // sides; without them, surface every event rather than silence real errors.192 if (batchId === undefined || !this.#batchTracking) return true;193 return this.#ownBatches.has(String(batchId));194 }195196 #trackBatch(batchId: string | null): void {197 if (batchId === null) return;198 this.#batchTracking = true;199 this.#ownBatches.add(batchId);200 if (this.#ownBatches.size > MAX_TRACKED_BATCHES) {201 this.#ownBatches.delete(this.#ownBatches.values().next().value);202 }203 }204205 #start(): void {206 this.#stopRuntimeRecreation = this.#environment.subscribeRuntimeRecreation(() =>207 this.#attach()208 );209 this.#attach();210 }211212 #attach(): void {213 const generation = ++this.#generation;214 this.#detachMutationErrors();215 void this.#environment.getDb().then((db) => {216 if (this.#disposed || generation !== this.#generation || this.#listeners.size === 0) return;217 let stop: () => void;218 try {219 stop = db.onMutationError((event) => this.#reportMutationError(event, generation));220 } catch (error) {221 this.#fail(error, generation);222 return;223 }224 if (this.#disposed || generation !== this.#generation || this.#listeners.size === 0) {225 stop();226 return;227 }228 this.#stopMutationErrors = stop;229 this.#updateDiagnostics((diagnostics) => {230 diagnostics.activeMutationListeners += 1;231 diagnostics.totalMutationListeners += 1;232 });233 }, (error) => this.#fail(error, generation));234 }235236 #reportMutationError(event: MutationErrorEvent, generation: number): void {237 if (generation !== this.#generation) return;238 // The vendor event is database-wide; only this store's own batches may239 // mark it failed or count against its diagnostics.240 if (!this.#ownsBatch(event)) return;241 this.#updateDiagnostics((diagnostics) => diagnostics.mutationErrors += 1);242 this.#fail(new Error(`${event.code}: ${event.reason}`));243 }244245 #stop(): void {246 this.#generation += 1;247 this.#detachMutationErrors();248 this.#stopRuntimeRecreation?.();249 this.#stopRuntimeRecreation = null;250 }251252 #detachMutationErrors(): void {253 if (!this.#stopMutationErrors) return;254 this.#stopMutationErrors();255 this.#stopMutationErrors = null;256 this.#updateDiagnostics((diagnostics) => diagnostics.activeMutationListeners -= 1);257 }258259 #fail(error: unknown, generation = this.#generation): void {260 if (this.#disposed || generation !== this.#generation) return;261 this.#set({262 durability: "failed",263 error: error instanceof Error ? error.message : String(error),264 });265 }266267 #set(change: Partial<TableMutationSnapshot>): void {268 this.#snapshot = { ...this.#snapshot, ...change };269 if (change.durability) {270 this.#updateDiagnostics((diagnostics) =>271 diagnostics.lastWriteDurability = change.durability!272 );273 }274 for (const listener of this.#listeners) listener();275 }276277 #updateDiagnostics(update: (diagnostics: RuntimeDiagnostics) => void): void {278 this.#environment.updateDiagnostics(update);279 }280}281282/** Retained ownership of one shared table-mutation store. */283export type TableMutationLease<T extends TableRow, Init> = {284 store: TableMutationStore<T, Init>;285 release(): void;286};287288type Entry = { store: TableMutationStore<TableRow, unknown>; leases: number };289290/** Shares mutation state and one vendor error listener per schema table. */291export class TableMutationRegistry {292 readonly #environment: TableMutationEnvironment;293 readonly #schemas = new WeakMap<object, Map<string, Entry>>();294 readonly #stores = new Set<TableMutationStore<TableRow, unknown>>();295296 /** Creates an isolated registry over one runtime environment. */297 constructor(environment: TableMutationEnvironment) {298 this.#environment = environment;299 }300301 /** Acquires the shared mutation surface for a typed table. */302 acquire<T extends TableRow, Init>(table: TableProxy<T, Init>): TableMutationLease<T, Init> {303 const schema = table._schema as object;304 let entries = this.#schemas.get(schema);305 if (!entries) {306 entries = new Map();307 this.#schemas.set(schema, entries);308 }309 let entry = entries.get(table._table);310 if (!entry) {311 const store = new TableMutationStore(312 table,313 this.#environment,314 () => this.#evictIfUnused(entries!, table._table),315 );316 entry = { store: store as unknown as TableMutationStore<TableRow, unknown>, leases: 0 };317 entries.set(table._table, entry);318 this.#stores.add(entry.store);319 }320 entry.leases += 1;321 let retained = true;322 return {323 store: entry.store as unknown as TableMutationStore<T, Init>,324 release: () => {325 if (!retained) return;326 retained = false;327 entry!.leases -= 1;328 this.#evictIfUnused(entries!, table._table);329 },330 };331 }332333 /** Disposes all active stores during hot replacement. */334 dispose(): void {335 for (const store of this.#stores) store.dispose();336 this.#stores.clear();337 }338339 #evictIfUnused(entries: Map<string, Entry>, key: string): void {340 const entry = entries.get(key);341 if (!entry || entry.leases > 0 || entry.store.consumerCount > 0) return;342 entries.delete(key);343 this.#stores.delete(entry.store);344 entry.store.dispose();345 }346}347348const defaultRegistry = new TableMutationRegistry({349 getDb: async () => (await getRuntime()).db,350 syncConfigured: syncing,351 getLedger: getWriteLedger,352 subscribeRuntimeRecreation(listener) {353 globalThis.addEventListener(runtimeRecreatedEvent, listener);354 return () => globalThis.removeEventListener(runtimeRecreatedEvent, listener);355 },356 updateDiagnostics: updateRuntimeDiagnostics,357});358359/** Acquires the package-wide typed mutation surface for one table. */360export function acquireTableMutations<T extends TableRow, Init>(361 table: TableProxy<T, Init>,362): TableMutationLease<T, Init> {363 return defaultRegistry.acquire(table);364}365366import.meta.hot?.dispose(() => defaultRegistry.dispose());