1// The convergence fixture (lofi's exit criterion for the sync-node spike):2// two devices of ONE account (same secret, separate runtimes — lofi's3// per-device sync model) make concurrent OFFLINE edits, reconnect, and must4// both observe the union. Stages mirror lofi's convergence runner: prepare →5// concurrent offline edits → reconnect → converge.6//7// Test 1 runs against a hosted daemon when LOFI_NODE_URL / LOFI_NODE_APP_ID /8// LOFI_NODE_ADMIN_SECRET are set (else an in-memory node it spins itself).9// Test 2 is the full topology: each device on its own node, nodes paired over10// the iroh tunnel.1112import { assert, assertEquals } from "@std/assert";13import { createDb, schema as s } from "jazz-tools";14import { deploy } from "jazz-tools/testing";15import { definePermissions } from "jazz-tools/permissions";16import { createSyncNode, type SyncNode } from "../src/node.ts";17import { decodeAppTicket } from "../src/appticket.ts";18import { resolveIrohLib } from "../src/native/loader.ts";1920const app = s.defineApp({21 notes: s.table({ text: s.string() }),22});2324// Per-account data: both devices share the account, so $createdBy policies25// cover the whole multi-device fixture.26const permissions = definePermissions(app, ({ policy, session }) => [27 policy.notes.allowRead.where({ $createdBy: session.userId }),28 policy.notes.allowInsert.where({ $createdBy: session.userId }),29 policy.notes.allowUpdate30 .whereOld({ $createdBy: session.userId })31 .whereNew({ $createdBy: session.userId }),32 policy.notes.allowDelete.where({ $createdBy: session.userId }),33]);3435function accountSecret(fill: number): string {36 let binary = "";37 for (const byte of new Uint8Array(32).fill(fill)) binary += String.fromCharCode(byte);38 return btoa(binary).replaceAll("+", "-").replaceAll("/", "_").replace(/=+$/, "");39}4041async function within<T>(operation: Promise<T>, label: string, milliseconds = 15_000): Promise<T> {42 let timer: ReturnType<typeof setTimeout> | undefined;43 try {44 return await Promise.race([45 operation,46 new Promise<never>((_resolve, reject) => {47 timer = setTimeout(() => reject(new Error(`${label} timed out`)), milliseconds);48 }),49 ]);50 } finally {51 if (timer !== undefined) clearTimeout(timer);52 }53}5455type Db = Awaited<ReturnType<typeof createDb>>;5657async function pollUntil(58 label: string,59 predicate: () => Promise<boolean>,60 milliseconds = 20_000,61): Promise<void> {62 const deadline = Date.now() + milliseconds;63 while (Date.now() < deadline) {64 if (await predicate()) return;65 await new Promise((r) => setTimeout(r, 500));66 }67 throw new Error(`${label} did not become true within ${milliseconds}ms`);68}6970/** The fixture proper: devices A and B already connected to their servers. */71async function runConvergenceFixture(deviceA: Db, deviceB: Db): Promise<void> {72 // Stage: prepare — a seeded row must reach both devices online.73 const seed = await within(74 deviceA.insert(app.notes, { text: "seed" }).wait({ tier: "global" }),75 "seed global durability",76 );77 await pollUntil(78 "device B observes the seed",79 async () =>80 (await deviceB.all(app.notes.where({ id: seed.id }), { tier: "global" })).length === 1,81 );8283 // Stage: concurrent offline edits.84 await deviceA.disconnect();85 await deviceB.disconnect();86 const fromA = deviceA.insert(app.notes, { text: "offline edit from device A" });87 const fromB = deviceB.insert(app.notes, { text: "offline edit from device B" });88 assertEquals(fromA.value.text, "offline edit from device A", "A's edit applied locally offline");89 assertEquals(fromB.value.text, "offline edit from device B", "B's edit applied locally offline");9091 // Stage: reconnect.92 await deviceA.reconnect();93 await deviceB.reconnect();94 await within(fromA.wait({ tier: "global" }), "A's edit global durability");95 await within(fromB.wait({ tier: "global" }), "B's edit global durability");9697 // Stage: converge — both devices observe the union.98 for (const [name, device] of [["A", deviceA], ["B", deviceB]] as const) {99 await pollUntil(`device ${name} observes all three notes`, async () => {100 const rows = await device.all(app.notes, { tier: "global" });101 const texts = new Set(rows.map((r) => r.text));102 return texts.has("seed") && texts.has("offline edit from device A") &&103 texts.has("offline edit from device B");104 });105 }106}107108interface Target {109 appId: string;110 serverUrl: string;111 adminSecret: string;112}113114async function connectDevices(a: Target, b: Target): Promise<[Db, Db]> {115 const secret = accountSecret(7);116 const deviceA = await createDb({117 appId: a.appId,118 serverUrl: a.serverUrl,119 secret,120 userBranch: "main",121 driver: { type: "memory" },122 });123 const deviceB = await createDb({124 appId: b.appId,125 serverUrl: b.serverUrl,126 secret,127 userBranch: "main",128 driver: { type: "memory" },129 });130 return [deviceA, deviceB];131}132133Deno.test({134 name: "convergence fixture: two devices through one lofi-node",135 sanitizeOps: false,136 sanitizeResources: false,137 fn: async () => {138 const hostedUrl = Deno.env.get("LOFI_NODE_URL");139 let node: SyncNode | null = null;140 let target: Target;141 if (hostedUrl) {142 target = {143 serverUrl: hostedUrl,144 appId: Deno.env.get("LOFI_NODE_APP_ID")!,145 adminSecret: Deno.env.get("LOFI_NODE_ADMIN_SECRET")!,146 };147 console.log(` → running against hosted daemon at ${hostedUrl}`);148 } else {149 node = await createSyncNode({150 appId: crypto.randomUUID(),151 backendSecret: "lofi_backend_conv",152 adminSecret: "lofi_admin_conv",153 inMemory: true,154 mesh: "off",155 });156 target = {157 serverUrl: node.url.replace(/^ws/, "http"),158 appId: node.appId,159 adminSecret: "lofi_admin_conv",160 };161 }162 await deploy({163 appId: target.appId,164 serverUrl: target.serverUrl,165 adminSecret: target.adminSecret,166 schema: app,167 permissions,168 });169 const [deviceA, deviceB] = await connectDevices(target, target);170 try {171 await runConvergenceFixture(deviceA, deviceB);172 } finally {173 await Promise.allSettled([174 within(deviceA.logout(), "device A cleanup", 3000),175 within(deviceB.logout(), "device B cleanup", 3000),176 ]);177 await node?.stop();178 }179 },180});181182Deno.test({183 name: "convergence fixture: two devices on two nodes paired over iroh",184 ignore: resolveIrohLib().status !== "ok",185 sanitizeOps: false,186 sanitizeResources: false,187 fn: async () => {188 const shared = {189 appId: crypto.randomUUID(),190 backendSecret: "lofi_backend_conv2",191 adminSecret: "lofi_admin_conv2",192 };193 const root = await createSyncNode({ ...shared, inMemory: true });194 const leaf = await createSyncNode({195 ...shared,196 inMemory: true,197 upstream: { peer: root.ticket() },198 });199 const rootTarget: Target = {200 serverUrl: root.url.replace(/^ws/, "http"),201 appId: shared.appId,202 adminSecret: shared.adminSecret,203 };204 const leafTarget: Target = {205 serverUrl: leaf.url.replace(/^ws/, "http"),206 appId: shared.appId,207 adminSecret: shared.adminSecret,208 };209 await deploy({ ...rootTarget, schema: app, permissions });210 await deploy({ ...leafTarget, schema: app, permissions });211212 // Device A on the root, device B on the leaf: convergence must cross the213 // iroh tunnel.214 const [deviceA, deviceB] = await connectDevices(rootTarget, leafTarget);215 try {216 await runConvergenceFixture(deviceA, deviceB);217 } finally {218 await Promise.allSettled([219 within(deviceA.logout(), "device A cleanup", 3000),220 within(deviceB.logout(), "device B cleanup", 3000),221 ]);222 await leaf.stop();223 await root.stop();224 }225 },226});227228Deno.test({229 name: "convergence fixture: ticket-gated node — jazz clients sync through /t/<secret> unchanged",230 sanitizeOps: false,231 sanitizeResources: false,232 fn: async () => {233 const node = await createSyncNode({234 appId: crypto.randomUUID(),235 backendSecret: "lofi_backend_gate",236 adminSecret: "lofi_admin_gate",237 inMemory: true,238 mesh: "off",239 access: "ticket",240 });241 try {242 // Provision ticket for the deploy (the gate injects the node's admin243 // secret — the client passes a placeholder), sync ticket for devices.244 const provisionTicket = decodeAppTicket(245 (await node.issueTicket({ label: "setup", scope: "provision" })).ticket,246 );247 const issued = await node.issueTicket({ label: "fixture" });248 const parsed = decodeAppTicket(issued.ticket);249 assert(parsed !== null && provisionTicket !== null, "issued tickets decode");250 const gatedUrl = parsed.url; // http://127.0.0.1:<gatePort>/t/<secret>251 assertEquals(parsed.appId, node.appId);252253 // ZERO client changes: ticket URLs are the serverUrl for BOTH the254 // deploy (provision scope, placeholder secret) and the devices (sync).255 await deploy({256 serverUrl: provisionTicket.url,257 appId: node.appId,258 adminSecret: "placeholder-the-gate-injects-the-real-one",259 schema: app,260 permissions,261 });262 const target = {263 serverUrl: gatedUrl,264 appId: node.appId,265 adminSecret: "lofi_admin_gate",266 };267 const [deviceA, deviceB] = await connectDevices(target, target);268 try {269 await runConvergenceFixture(deviceA, deviceB);270 } finally {271 await Promise.allSettled([272 within(deviceA.logout(), "device A cleanup", 3000),273 within(deviceB.logout(), "device B cleanup", 3000),274 ]);275 }276277 // Negative: a revoked ticket is rejected at the gate (401 → jazz client278 // fetch fails), and a bogus secret never reaches Jazz.279 await node.revokeTicket(issued.id);280 const res = await fetch(`${gatedUrl}/apps/${node.appId}/schemas`, {281 headers: { "X-Jazz-Admin-Secret": "lofi_admin_gate" },282 });283 await res.body?.cancel();284 assertEquals(res.status, 401, "revoked ticket → 401 at the gate");285 const tickets = await node.listTickets();286 assertEquals(tickets.length, 2);287 assert(288 tickets.find((t) => t.id === issued.id)?.revoked,289 "status reflects revocation",290 );291 } finally {292 await node.stop();293 }294 },295});