1// Effect-system integration against a real local Jazz server: two-client2// stage progression with single-device effect delivery, boot reconciliation3// over a copied journal, and the adjudicated stale-policy rejection path.4// Runs in its own task (test:effects) because the server needs FFI.5import { createDb, type Db, schema as s } from "jazz-tools";6import { deploy, startLocalJazzServer } from "jazz-tools/testing";7import type { EffectContext, EffectUnit } from "./schema/effects.ts";8import { createDiagnostics } from "./runtime/diagnostics.ts";9import { assert, assertCount } from "./runtime/test-assert.ts";10import { createMemoryJournalStorage } from "./runtime/write-journal.ts";11import { WriteLedger, type WriteLedgerEnvironment } from "./runtime/write-ledger.ts";1213const app = s.defineApp({14 notes: s.table({ text: s.string() }),15});16const openPermissions = s.definePermissions(app, ({ policy }) => {17 policy.notes.allowInsert.always();18 policy.notes.allowRead.always();19 policy.notes.allowUpdate.always();20 policy.notes.allowDelete.always();21});22// No write rule: a store deployed with this denies every insert.23const closedPermissions = s.definePermissions(app, ({ policy }) => {24 policy.notes.allowRead.always();25});2627function secret(fill: number): string {28 let binary = "";29 for (const byte of new Uint8Array(32).fill(fill)) binary += String.fromCharCode(byte);30 return btoa(binary).replaceAll("+", "-").replaceAll("/", "_").replace(/=+$/, "");31}3233async function within<T>(operation: Promise<T>, label: string, milliseconds = 20_000): Promise<T> {34 let timer: ReturnType<typeof setTimeout> | undefined;35 try {36 return await Promise.race([37 operation,38 new Promise<never>((_resolve, reject) => {39 timer = setTimeout(() => reject(new Error(`${label} timed out`)), milliseconds);40 }),41 ]);42 } finally {43 if (timer !== undefined) clearTimeout(timer);44 }45}4647type EffectCall = { handler: "onSynced" | "onRejected"; context: EffectContext };4849function recordingUnit(name: string, calls: EffectCall[]): EffectUnit<{ id: string }> {50 return {51 effectName: name,52 handlers: {53 onSynced: (_row, context) => void calls.push({ handler: "onSynced", context }),54 onRejected: (_row, context) => void calls.push({ handler: "onRejected", context }),55 },56 };57}5859function ledgerOver(db: Db, options: {60 storage?: ReturnType<typeof createMemoryJournalStorage>;61 units?: Map<string, EffectUnit<{ id: string }>>;62} = {}) {63 const diagnostics = createDiagnostics();64 const storage = options.storage ?? createMemoryJournalStorage();65 const units = options.units ?? new Map<string, EffectUnit<{ id: string }>>();66 const environment: WriteLedgerEnvironment = {67 getDb: () => Promise.resolve(db),68 syncConfigured: () => true,69 storage,70 resolveTable: (name) =>71 name === "notes"72 ? app.notes as unknown as { where(input: Record<string, unknown>): unknown }73 : null,74 resolveEffectUnit: (name) => units.get(name) ?? null,75 subscribeRuntimeRecreation: () => () => undefined,76 updateDiagnostics: (update) => update(diagnostics),77 now: Date.now,78 retryDelayMs: () => 250,79 probeTimeoutMs: 8000,80 };81 return { ledger: new WriteLedger(environment), storage, diagnostics };82}8384type Harness = {85 server: Awaited<ReturnType<typeof startLocalJazzServer>>;86 client(fill: number): Promise<Db>;87 stop(): Promise<void>;88};8990async function bootHarness(): Promise<Harness> {91 const server = await startLocalJazzServer({ allowLocalFirstAuth: true });92 await deploy({93 appId: server.appId,94 serverUrl: server.url,95 adminSecret: server.adminSecret,96 schema: app,97 permissions: openPermissions,98 });99 const clients: Db[] = [];100 return {101 server,102 async client(fill: number) {103 const db = await createDb({104 appId: server.appId,105 serverUrl: server.url,106 secret: secret(fill),107 userBranch: "main",108 driver: { type: "memory" },109 });110 clients.push(db);111 return db;112 },113 async stop() {114 await Promise.allSettled(115 clients.map((db, index) => within(db.logout(), `client ${index} cleanup`, 3_000)),116 );117 await within(server.stop(), "server cleanup", 3_000).catch(() => undefined);118 },119 };120}121122function waitFor(predicate: () => boolean, label: string, milliseconds = 20_000): Promise<void> {123 return within(124 (async () => {125 while (!predicate()) await new Promise((resolve) => setTimeout(resolve, 100));126 })(),127 label,128 milliseconds,129 );130}131132Deno.test("two clients: stages progress and onSynced fires only on the originating device", async () => {133 const harness = await bootHarness();134 try {135 const aliceDb = await harness.client(1);136 const bobDb = await harness.client(2);137 const aliceCalls: EffectCall[] = [];138 const bobCalls: EffectCall[] = [];139 const aliceUnit = recordingUnit("notifyTeam", aliceCalls);140 const alice = ledgerOver(aliceDb, { units: new Map([["notifyTeam", aliceUnit]]) });141 const bob = ledgerOver(bobDb, {142 units: new Map([["notifyTeam", recordingUnit("notifyTeam", bobCalls)]]),143 });144 await alice.ledger.arm();145 await bob.ledger.arm();146147 const handle = alice.ledger.perform<{ id: string; text: string }>(148 { kind: "insert", table: app.notes as never, values: { text: "from alice" } },149 { verb: "addNote", units: [aliceUnit] },150 );151 const stages: string[] = [handle.stage];152 handle.subscribe(() => {153 if (stages.at(-1) !== handle.stage) stages.push(handle.stage);154 });155 const row = await within(Promise.resolve(handle), "saved settlement");156 assert(row.text === "from alice", "await must resolve the created row at saved");157 await within(handle.synced, "synced settlement");158 assert(159 stages.join(",").endsWith("saved,synced") || stages.join(",") === "saved,synced" ||160 stages.join(",") === "saving,saved,synced",161 `stages must progress monotonically (saw ${stages.join(",")})`,162 );163164 const bobView = await within(165 bobDb.all(app.notes.where({ id: row.id }), { tier: "global" }),166 "bob read",167 );168 assertCount(bobView.length, 1, "the second client must sync the row");169 await waitFor(() => aliceCalls.length > 0, "originating-device effect delivery");170 assertCount(aliceCalls.length, 1, "onSynced must fire exactly once on the writer");171 assert(aliceCalls[0].handler === "onSynced", "confirmation runs the action handler");172 // Give any wrongly-armed effect a moment to surface before asserting.173 await new Promise((resolve) => setTimeout(resolve, 300));174 assertCount(bobCalls.length, 0, "effects run on the originating device only");175 assertCount(alice.ledger.getPendingSnapshot().count, 0, "nothing may stay pending");176 alice.ledger.dispose();177 bob.ledger.dispose();178 } finally {179 await harness.stop();180 }181});182183Deno.test("boot reconciliation: a journaled write settles and its effect fires after sync resolves", async () => {184 const harness = await bootHarness();185 try {186 const writerDb = await harness.client(1);187 const writer = ledgerOver(writerDb);188 await writer.ledger.arm();189 const handle = writer.ledger.perform<{ id: string; text: string }>(190 { kind: "insert", table: app.notes as never, values: { text: "journaled" } },191 { verb: "addNote", units: [recordingUnit("celebrate", [])] },192 );193 await within(Promise.resolve(handle), "saved settlement");194 await writer.ledger.flush();195 // Freeze the journal at the saved stage: this is the on-disk state a196 // crashed device would boot from.197 const frozen = writer.storage.text();198 assert(frozen !== null && frozen.includes(handle.writeId), "the write must be journaled");199 await within(handle.synced, "writer synced settlement");200 writer.ledger.dispose();201202 // A fresh runtime for the same account boots from the frozen journal.203 const rebootDb = await harness.client(1);204 const calls: EffectCall[] = [];205 const reboot = ledgerOver(rebootDb, {206 storage: createMemoryJournalStorage(frozen),207 units: new Map([["celebrate", recordingUnit("celebrate", calls)]]),208 });209 await reboot.ledger.arm();210 await waitFor(() => calls.length > 0, "re-armed effect delivery");211 assertCount(calls.length, 1, "the re-armed effect must fire exactly once");212 assert(calls[0].handler === "onSynced", "the confirmed write runs the action handler");213 assert(214 calls[0].context.journalId === `${handle.writeId}:celebrate`,215 "the idempotency key must survive the reboot",216 );217 assertCount(reboot.ledger.getPendingSnapshot().count, 0, "the entry must settle");218 reboot.ledger.dispose();219 } finally {220 await harness.stop();221 }222});223224Deno.test("stale-policy rejection: rejected stage, one compensation, engine rolls the row back", async () => {225 const harness = await bootHarness();226 try {227 const aliceDb = await harness.client(1);228 // Seed while the open policy is in force so the client holds it locally.229 const seeded = await within(230 aliceDb.insert(app.notes, { text: "seed" }).wait({ tier: "global" }),231 "seed insert",232 );233 const calls: EffectCall[] = [];234 const chargeCard = recordingUnit("chargeCard", calls);235 const alice = ledgerOver(aliceDb, { units: new Map([["chargeCard", chargeCard]]) });236 await alice.ledger.arm();237238 await within(aliceDb.disconnect(), "disconnect", 5_000);239 // The store's permissions tighten while the device is offline.240 await deploy({241 appId: harness.server.appId,242 serverUrl: harness.server.url,243 adminSecret: harness.server.adminSecret,244 schema: app,245 permissions: closedPermissions,246 });247248 // Accepted locally under the stale policy: no throw, a live handle.249 const handle = alice.ledger.perform<{ id: string; text: string }>(250 { kind: "insert", table: app.notes as never, values: { text: "stale write" } },251 { verb: "addNote", units: [chargeCard] },252 );253 await within(aliceDb.reconnect(), "reconnect", 10_000);254 await waitFor(() => handle.stage === "rejected", "adjudicated verdict");255 assert(handle.reason?.code === "permission_denied", "the verdict code must surface");256257 await waitFor(() => calls.length > 0, "compensation delivery");258 assertCount(calls.length, 1, "onRejected must fire exactly once");259 assert(calls[0].handler === "onRejected", "the denied write runs compensation");260261 // The engine itself rolls the rejected write back out of local reads: the262 // optimistic row must not linger as apparent success.263 const rows = await within(264 aliceDb.all(app.notes.where({}), { tier: "local" }),265 "post-rejection read",266 );267 assert(268 rows.every((row) => row.text !== "stale write"),269 "the rejected optimistic row must leave local query results",270 );271 assert(rows.some((row) => row.id === seeded.id), "unrelated rows must survive");272 assertCount(alice.ledger.getPendingSnapshot().count, 0, "the rejected write is not pending");273 alice.ledger.dispose();274 } finally {275 await harness.stop();276 }277});