1// Headless scenario adapter: real `createDb` clients on a memory driver2// syncing through a local Jazz server. `jazz-tools` and `jazz-tools/testing`3// are imported dynamically inside the boot path only — this module is loaded4// by the `./testing` entry under the FFI-free unit-test task, so nothing5// FFI-adjacent may load at import time.6import type { CompiledPermissions } from "jazz-tools";7import { ScenarioError } from "./errors.ts";8import {9 registerConvergeSource,10 type ScenarioDb,11 type ScenarioPeer,12 type ScenarioReadOptions,13 type ScenarioSettleOptions,14} from "./peer.ts";1516type JazzTesting = typeof import("jazz-tools/testing");17type JazzRoot = typeof import("jazz-tools");18type Db = Awaited<ReturnType<JazzRoot["createDb"]>>;1920/** The app-object shape the headless adapter needs: table handles plus the compiled schema. */21export type ScenarioApp = {22 readonly wasmSchema: {23 readonly [table: string]: {24 readonly columns: readonly {25 readonly name: string;26 readonly nullable: boolean;27 readonly default?: unknown;28 readonly references?: string;29 readonly column_type: { readonly type: string };30 readonly merge_strategy?: string;31 }[];32 };33 };34};3536type TableHandle = {37 readonly _table: string;38 where(conditions: Record<string, unknown>): unknown;39};4041type WriteWaiter = { wait(options: { tier: "local" | "global" }): Promise<unknown> };4243function mergeColumns(44 schema: ScenarioApp["wasmSchema"],45 table: string,46 strategy: string,47): ReadonlySet<string> {48 const names = new Set<string>();49 for (const column of schema[table]?.columns ?? []) {50 if (column.merge_strategy === strategy) names.add(column.name);51 }52 return names;53}5455async function within<T>(operation: Promise<T>, label: string, milliseconds: number): Promise<T> {56 let timer: ReturnType<typeof setTimeout> | undefined;57 try {58 return await Promise.race([59 operation,60 new Promise<never>((_resolve, reject) => {61 timer = setTimeout(() => reject(new Error(`${label} timed out`)), milliseconds);62 }),63 ]);64 } finally {65 if (timer !== undefined) clearTimeout(timer);66 }67}6869function fillSecret(fill: number): string {70 let binary = "";71 for (const byte of new Uint8Array(32).fill(fill)) binary += String.fromCharCode(byte);72 return btoa(binary).replaceAll("+", "-").replaceAll("/", "_").replace(/=+$/, "");73}7475function tableKeysOf(app: object): readonly string[] {76 return Object.entries(app)77 .filter(([, value]) =>78 typeof value === "object" && value !== null &&79 typeof (value as { _table?: unknown })._table === "string"80 )81 .map(([key]) => key);82}8384/** A booted headless scenario: a local Jazz server with a deployed app and peer factory. */85export interface HeadlessScenarioHarness<A> {86 /** Boot a new named peer synced to the harness's server. */87 peer(name: string): Promise<ScenarioPeer<A>>;88 /** Tear down every peer and the server, with bounded waits. */89 stop(): Promise<void>;90}9192/**93 * Boot a local Jazz server, deploy the app, and return a peer factory.94 * Requires FFI (the local server is native), so callers live in the95 * scenario test task, never in the unit-test task.96 */97export async function createHeadlessScenarioHarness<A extends ScenarioApp>(98 app: A,99 permissions: CompiledPermissions,100): Promise<HeadlessScenarioHarness<A>> {101 const testing: JazzTesting = await import("jazz-tools/testing");102 const root: JazzRoot = await import("jazz-tools");103 const server = await testing.startLocalJazzServer({ allowLocalFirstAuth: true });104 try {105 await testing.deploy({106 appId: server.appId,107 serverUrl: server.url,108 adminSecret: server.adminSecret,109 schema: app as never,110 permissions,111 });112 } catch (error) {113 await within(server.stop(), "server cleanup", 3_000).catch(() => undefined);114 throw new ScenarioError("deploy", "deploying the app schema failed", { cause: error });115 }116117 const clients: Db[] = [];118 let nextFill = 1;119120 const bootClient = (fill: number): Promise<Db> =>121 root.createDb({122 appId: server.appId,123 serverUrl: server.url,124 secret: fillSecret(fill),125 userBranch: "main",126 driver: { type: "memory" },127 });128129 const makePeer = async (name: string): Promise<ScenarioPeer<A>> => {130 const fill = nextFill++;131 let db = await bootClient(fill);132 clients.push(db);133 let offline = false;134 let outstanding: WriteWaiter[] = [];135136 const track = (handle: WriteWaiter): void => {137 outstanding.push(handle);138 };139140 const readTable = async (141 key: string,142 tier: "local" | "global",143 ): Promise<readonly Record<string, unknown>[]> => {144 const handle = (app as Record<string, unknown>)[key] as TableHandle;145 const rows = await db.all(handle as never, { tier });146 return rows as readonly Record<string, unknown>[];147 };148149 const facade: Record<string, unknown> = {};150 for (const key of tableKeysOf(app)) {151 const handle = (app as Record<string, unknown>)[key] as TableHandle;152 // Writes apply to the local view immediately; durability waits are left153 // to settle()/converge because local-tier waits do not resolve while154 // the peer is disconnected, and offline writes must not block.155 // The async wrappers turn the vendor's synchronous write errors (e.g.156 // updating a deleted row) into rejections, so callers can uniformly157 // `.catch` them.158 facade[key] = {159 insert: async (values: Record<string, unknown>) => {160 const result = db.insert(handle as never, values as never);161 track(result);162 return await Promise.resolve(result.value);163 },164 update: async (id: string, patch: Record<string, unknown>) => {165 const result = db.update(handle as never, id, patch as never);166 track(result);167 await Promise.resolve();168 },169 remove: async (id: string) => {170 const result = db.delete(handle as never, id);171 track(result);172 await Promise.resolve();173 },174 all: async (where?: Record<string, unknown>, options?: ScenarioReadOptions) => {175 const query = where === undefined ? handle : handle.where(where);176 return await db.all(query as never, { tier: options?.tier ?? "local" });177 },178 get: async (id: string, options?: ScenarioReadOptions) => {179 const row = await db.one(180 handle.where({ id }) as never,181 { tier: options?.tier ?? "local" },182 );183 return row ?? undefined;184 },185 };186 }187188 const peer: ScenarioPeer<A> = {189 name,190 get isOffline() {191 return offline;192 },193 db: facade as ScenarioDb<A>,194 async offline() {195 if (offline) return;196 offline = true;197 await db.disconnect();198 },199 async online() {200 if (!offline) return;201 offline = false;202 await db.reconnect();203 },204 async settle(options?: ScenarioSettleOptions) {205 const timeoutMs = options?.timeoutMs ?? 20_000;206 const pending = outstanding;207 outstanding = [];208 const hint = offline ? ` — ${name} is offline; call ${name}.online() before settling` : "";209 // A rejected write is settled: the sync node adjudicated it. Only the210 // durability wait itself failing (e.g. an offline peer) is an error.211 await within(212 Promise.all(213 pending.map((handle) => handle.wait({ tier: "global" }).catch(() => undefined)),214 ),215 `settle of ${pending.length} write(s) on ${name}${hint}`,216 timeoutMs,217 );218 },219 async restart() {220 outstanding = [];221 await within(db.logout(), `${name} logout`, 5_000).catch(() => undefined);222 const index = clients.indexOf(db);223 if (index !== -1) clients.splice(index, 1);224 db = await bootClient(fill);225 clients.push(db);226 offline = false;227 },228 };229230 registerConvergeSource(peer, {231 kind: "tables",232 tableKeys: tableKeysOf(app),233 counterColumns: (table) => mergeColumns(app.wasmSchema, table, "Counter"),234 setColumns: (table) => mergeColumns(app.wasmSchema, table, "GSet"),235 readTable,236 });237 return peer;238 };239240 return {241 peer: makePeer,242 async stop() {243 await Promise.allSettled(244 clients.map((client, index) => within(client.logout(), `client ${index} cleanup`, 3_000)),245 );246 await within(server.stop(), "server cleanup", 3_000).catch(() => undefined);247 },248 };249}