1import type { Db, MutationErrorEvent, TableProxy } from "jazz-tools";2import { schema as s } from "jazz-tools";3import { createDiagnostics } from "./diagnostics.ts";4import { type TableMutationEnvironment, TableMutationRegistry } from "./table-mutations.ts";5import { assert, assertCount } from "./test-assert.ts";6import { WriteHandle } from "./write-handle.ts";7import { WriteLedger } from "./write-ledger.ts";8import { createMemoryJournalStorage } from "./write-journal.ts";910type Row = { id: string; title: string; archived: boolean };11type Init = Omit<Row, "id">;12const table = { _table: "records", _schema: {} } as TableProxy<Row, Init>;1314class FakeDb {15 mutationListeners = new Set<(event: MutationErrorEvent) => void>();16 localWaits = 0;17 globalWaits = 0;18 failLocal: Error | null = null;19 nextId = 0;20 batchIds = false;21 nextBatch = 0;22 operations: string[] = [];2324 value(): Db {25 return this as unknown as Db;26 }2728 onMutationError(listener: (event: MutationErrorEvent) => void): () => void {29 this.mutationListeners.add(listener);30 return () => this.mutationListeners.delete(listener);31 }3233 insert(_table: TableProxy<Row, Init>, values: Init) {34 this.operations.push("insert");35 const row = { id: `row-${++this.nextId}`, ...values };36 return { value: row, ...this.handle(row) };37 }3839 update(_table: TableProxy<Row, Init>, _id: string, _patch: Partial<Init>) {40 this.operations.push("update");41 return this.handle(undefined);42 }4344 delete(_table: TableProxy<Row, Init>, _id: string) {45 this.operations.push("delete");46 return this.handle(undefined);47 }4849 all(): Promise<unknown[]> {50 return Promise.resolve([]);51 }5253 reject(event: MutationErrorEvent): void {54 for (const listener of this.mutationListeners) listener(event);55 }5657 private handle<T>(value: T) {58 return {59 ...(this.batchIds ? { batchId: `batch-${++this.nextBatch}` } : {}),60 wait: ({ tier }: { tier: "local" | "global" }) => {61 if (tier === "local") {62 this.localWaits += 1;63 if (this.failLocal) {64 const error = this.failLocal;65 this.failLocal = null;66 return Promise.reject(error);67 }68 } else {69 this.globalWaits += 1;70 }71 return Promise.resolve(value);72 },73 };74 }75}7677function environment(db: FakeDb, syncConfigured = false) {78 const diagnostics = createDiagnostics();79 const recreation = new Set<() => void>();80 const ledger = new WriteLedger({81 getDb: () => Promise.resolve(db.value()),82 syncConfigured: () => syncConfigured,83 storage: createMemoryJournalStorage(),84 resolveTable: () => null,85 resolveEffectUnit: () => null,86 subscribeRuntimeRecreation(listener) {87 recreation.add(listener);88 return () => recreation.delete(listener);89 },90 updateDiagnostics(update) {91 update(diagnostics);92 },93 now: () => 0,94 });95 const value: TableMutationEnvironment = {96 getDb: () => Promise.resolve(db.value()),97 syncConfigured: () => syncConfigured,98 getLedger: () => ledger,99 subscribeRuntimeRecreation(listener) {100 recreation.add(listener);101 return () => recreation.delete(listener);102 },103 updateDiagnostics(update) {104 update(diagnostics);105 },106 };107 return { value, diagnostics, recreation, ledger };108}109110async function flush(): Promise<void> {111 await Promise.resolve();112 await Promise.resolve();113 await Promise.resolve();114}115116Deno.test("table mutation consumers share one listener and release it after final teardown", async () => {117 const db = new FakeDb();118 const runtime = environment(db);119 const registry = new TableMutationRegistry(runtime.value);120 const first = registry.acquire(table);121 const second = registry.acquire(table);122 assert(first.store === second.store, "one table created duplicate mutation stores");123 const stopFirst = first.store.subscribe(() => undefined);124 const stopSecond = second.store.subscribe(() => undefined);125 await flush();126 assertCount(db.mutationListeners.size, 1, "one table opened duplicate mutation listeners");127 assertCount(runtime.diagnostics.activeMutationListeners, 1, "listener diagnostics drifted");128 stopFirst();129 first.release();130 assertCount(db.mutationListeners.size, 1, "one remaining consumer lost mutation errors");131 stopSecond();132 second.release();133 assertCount(db.mutationListeners.size, 0, "final teardown retained the mutation listener");134 assertCount(runtime.diagnostics.activeMutationListeners, 0, "listener diagnostics leaked");135 runtime.ledger.dispose();136});137138Deno.test("typed mutations return inserted rows and wait for local durability", async () => {139 const db = new FakeDb();140 const runtime = environment(db);141 const store = new TableMutationRegistry(runtime.value).acquire(table).store;142 const created = await store.insert({ title: "created", archived: false });143 await store.update(created.id, { archived: true });144 await store.remove(created.id);145 await flush();146 assert(created.id === "row-1" && created.title === "created", "insert omitted the created row");147 assert(db.operations.join(",") === "insert,update,delete", "typed operations missed the DB");148 assertCount(db.localWaits, 3, "mutations resolved before local durability");149 assertCount(runtime.diagnostics.localWaitCalls, 3, "local waits were not diagnosable");150 assert(store.getSnapshot().durability === "local", "local durability was not observable");151 runtime.ledger.dispose();152});153154Deno.test("managed mutations continue global durability tracking in the background", async () => {155 const db = new FakeDb();156 const runtime = environment(db, true);157 const store = new TableMutationRegistry(runtime.value).acquire(table).store;158 await store.insert({ title: "managed", archived: false });159 await flush();160 assertCount(db.globalWaits, 1, "managed mutation omitted its global wait");161 assert(store.getSnapshot().durability === "global", "global confirmation was not observable");162 assertCount(runtime.diagnostics.pendingGlobalWrites, 0, "global wait remained pending");163 runtime.ledger.dispose();164});165166Deno.test("local and asynchronous permission rejections reach public mutation state", async () => {167 const db = new FakeDb();168 const runtime = environment(db);169 const lease = new TableMutationRegistry(runtime.value).acquire(table);170 const stop = lease.store.subscribe(() => undefined);171 await flush();172 db.failLocal = new Error("local permission denied");173 await lease.store.insert({ title: "blocked", archived: false }).then(174 () => undefined,175 () => undefined,176 );177 await flush();178 assert(lease.store.getSnapshot().error === "local permission denied", "local rejection hidden");179 db.reject({ code: "WriteRejected", reason: "revoked" } as MutationErrorEvent);180 assert(lease.store.getSnapshot().error?.includes("revoked"), "async rejection hidden");181 assertCount(runtime.diagnostics.mutationErrors, 2, "rejections were not diagnosed");182 stop();183 lease.release();184 runtime.ledger.dispose();185});186187Deno.test("asynchronous rejections reach only the store owning the batch", async () => {188 const db = new FakeDb();189 db.batchIds = true;190 const runtime = environment(db);191 const lease = new TableMutationRegistry(runtime.value).acquire(table);192 const stop = lease.store.subscribe(() => undefined);193 await flush();194 await lease.store.insert({ title: "mine", archived: false });195 await flush();196 db.reject(197 {198 code: "WriteRejected",199 reason: "another table's write",200 batch: { batchId: "batch-999" },201 } as unknown as MutationErrorEvent,202 );203 assert(204 lease.store.getSnapshot().durability !== "failed",205 "a foreign batch rejection must not fail this store",206 );207 assertCount(runtime.diagnostics.mutationErrors, 0, "a foreign batch must not be counted here");208 db.reject(209 {210 code: "WriteRejected",211 reason: "revoked",212 batch: { batchId: "batch-1" },213 } as unknown as MutationErrorEvent,214 );215 assert(lease.store.getSnapshot().error?.includes("revoked"), "own rejection was hidden");216 assertCount(runtime.diagnostics.mutationErrors, 1, "the owned rejection must count once");217 stop();218 lease.release();219 runtime.ledger.dispose();220});221222Deno.test("mutation types follow the exact Jazz table insert and row shapes", () => {223 const app = s.defineApp({ records: s.table({ title: s.string(), archived: s.boolean() }) });224 const runtime = environment(new FakeDb());225 const registry = new TableMutationRegistry(runtime.value);226 const lease = registry.acquire(app.records);227 const result: WriteHandle<s.RowOf<typeof app.records>> = lease.store.insert({228 title: "typed",229 archived: false,230 });231 assert(result instanceof WriteHandle, "typed insert did not return a write handle");232 assert(typeof result.then === "function", "the write handle must stay thenable");233 lease.release();234 runtime.ledger.dispose();235});