1import { ScenarioError } from "./errors.ts";2import {3 type ConvergeSource,4 convergeSourceOf,5 type ScenarioPeer,6 type ScenarioPeerControls,7 type ScenarioTable,8 type ScenarioTableKey,9} from "./peer.ts";1011/** Options accepted by {@link converge} alongside the peers. */12export interface ConvergeOptions<A> {13 /** Tables to compare. Defaults to every table of the app. */14 tables?: readonly ScenarioTableKey<A>[];15 /** Deadline for the peers to reach identical views. Defaults to 20 seconds. */16 timeoutMs?: number;17 /** Interval between comparison rounds. Defaults to 250 milliseconds. */18 pollMs?: number;19 /**20 * How to compare counter-merged columns. `"presence"` (the default) checks21 * only that the row and column exist: a replica that watched a counter's22 * history legitimately reads a different total than a fresh boot of the same23 * account, so value equality across live replicas is not part of the24 * convergence contract. `"value"` opts into strict equality anyway.25 */26 counterColumns?: "presence" | "value";27}2829type NormalizedTable = Map<string, string>;3031function stableSerialize(value: unknown): string {32 return JSON.stringify(value, function (key, cooked: unknown) {33 const raw: unknown = (this as Record<string, unknown>)[key];34 if (raw instanceof Date) return `date:${raw.getTime()}`;35 if (Array.isArray(cooked)) return cooked;36 if (typeof cooked === "object" && cooked !== null) {37 const record = cooked as Record<string, unknown>;38 return Object.fromEntries(Object.keys(record).sort().map((k) => [k, record[k]]));39 }40 return cooked;41 });42}4344function normalizeRows(45 rows: readonly Record<string, unknown>[],46 skipColumns: ReadonlySet<string>,47 setColumns: ReadonlySet<string>,48): NormalizedTable {49 const normalized: NormalizedTable = new Map();50 for (const row of rows) {51 const id = String(row.id);52 const shaped: Record<string, unknown> = {};53 for (const [column, value] of Object.entries(row)) {54 if (skipColumns.has(column)) {55 shaped[column] = value === undefined ? "absent" : "present";56 } else if (setColumns.has(column) && Array.isArray(value)) {57 shaped[column] = value.map((element) => stableSerialize(element)).sort();58 } else {59 shaped[column] = value;60 }61 }62 normalized.set(id, stableSerialize(shaped));63 }64 return normalized;65}6667function diffTables(68 table: string,69 views: readonly { peer: string; rows: NormalizedTable }[],70): string[] {71 const lines: string[] = [];72 const reference = views[0];73 for (const view of views.slice(1)) {74 for (const [id, serialized] of reference.rows) {75 const other = view.rows.get(id);76 if (other === undefined) {77 lines.push(`${table}: row ${id} present on ${reference.peer}, missing on ${view.peer}`);78 } else if (other !== serialized) {79 lines.push(80 `${table}: row ${id} differs\n ${reference.peer}: ${serialized}\n ${view.peer}: ${other}`,81 );82 }83 }84 for (const id of view.rows.keys()) {85 if (!reference.rows.has(id)) {86 lines.push(`${table}: row ${id} present on ${view.peer}, missing on ${reference.peer}`);87 }88 }89 }90 return lines;91}9293function isPeer(candidate: unknown): candidate is ScenarioPeerControls {94 return typeof candidate === "object" && candidate !== null &&95 typeof (candidate as ScenarioPeerControls).settle === "function";96}9798async function tableRound(99 sources: readonly { name: string; source: Extract<ConvergeSource, { kind: "tables" }> }[],100 tables: readonly string[],101 counterColumns: "presence" | "value" | undefined,102): Promise<string[]> {103 const diff: string[] = [];104 for (const table of tables) {105 const skip = counterColumns === "value"106 ? new Set<string>()107 : sources[0].source.counterColumns(table);108 const sets = sources[0].source.setColumns(table);109 const views = await Promise.all(sources.map(async ({ name, source }) => ({110 peer: name,111 rows: normalizeRows(await source.readTable(table, "global"), skip, sets),112 })));113 diff.push(...diffTables(table, views));114 }115 return diff;116}117118async function snapshotRound(119 sources: readonly { name: string; source: Extract<ConvergeSource, { kind: "snapshot" }> }[],120): Promise<string[]> {121 const views = await Promise.all(sources.map(async ({ name, source }) => ({122 peer: name,123 view: stableSerialize(await source.snapshot()),124 })));125 const reference = views[0];126 return views.slice(1)127 .filter((view) => view.view !== reference.view)128 .map((view) =>129 `snapshot differs\n ${reference.peer}: ${reference.view}\n ${view.peer}: ${view.view}`130 );131}132133/**134 * Assert that all peers converge to identical views. Settles every peer's135 * issued writes, then polls until every peer reads the same state, or fails136 * at the deadline with a per-peer diff. All peers must be online. Headless137 * peers compare their tables row by row; browser peers compare their138 * app-supplied value-free snapshots. Accepts an options object after the139 * peers: `converge(alice, bob, { timeoutMs: 30_000 })`.140 */141export async function converge<A>(142 ...args: readonly (ScenarioPeer<A> | ScenarioPeerControls | ConvergeOptions<A>)[]143): Promise<void> {144 const peers = args.filter(isPeer);145 const options = (args.find((arg) => !isPeer(arg)) ?? {}) as ConvergeOptions<A>;146 if (peers.length < 2) {147 throw new ScenarioError("convergence", "converge needs at least two peers");148 }149 for (const peer of peers) {150 if (peer.isOffline) {151 throw new ScenarioError(152 "convergence",153 `${peer.name} is offline; call ${peer.name}.online() before asserting convergence`,154 { peer: peer.name },155 );156 }157 }158 const resolved = peers.map((peer) => {159 const source = convergeSourceOf(peer);160 if (source === undefined) {161 throw new ScenarioError(162 "convergence",163 `${peer.name} does not expose a converge source`,164 { peer: peer.name },165 );166 }167 return { name: peer.name, source };168 });169 const tableSources = resolved.filter(170 (entry): entry is { name: string; source: Extract<ConvergeSource, { kind: "tables" }> } =>171 entry.source.kind === "tables",172 );173 const snapshotSources = resolved.filter(174 (entry): entry is { name: string; source: Extract<ConvergeSource, { kind: "snapshot" }> } =>175 entry.source.kind === "snapshot",176 );177 if (tableSources.length > 0 && snapshotSources.length > 0) {178 throw new ScenarioError(179 "convergence",180 "cannot converge headless and browser peers against each other",181 );182 }183184 const timeoutMs = options.timeoutMs ?? 20_000;185 const pollMs = options.pollMs ?? 250;186 const deadline = Date.now() + timeoutMs;187 await Promise.all(peers.map((peer) => peer.settle({ timeoutMs })));188189 const tables = (options.tables as readonly string[] | undefined) ??190 tableSources[0]?.source.tableKeys ?? [];191 let lastDiff: string[] = [];192 while (true) {193 lastDiff = tableSources.length > 0194 ? await tableRound(tableSources, tables, options.counterColumns)195 : await snapshotRound(snapshotSources);196 if (lastDiff.length === 0) return;197 if (Date.now() >= deadline) break;198 await new Promise((resolve) => setTimeout(resolve, pollMs));199 }200 throw new ScenarioError(201 "convergence",202 `peers did not converge within ${timeoutMs}ms`,203 { details: lastDiff.join("\n") },204 );205}206207/**208 * A reference to one of the app's tables, as the row assertions accept it:209 * any of the app object's own table handles (e.g. `app.documents`) matches.210 */211export type ScenarioTableRef<Row> = { readonly _table: string; readonly _rowType: Row };212213function tableFacade<Row>(214 peer: ScenarioPeer<unknown>,215 table: ScenarioTableRef<Row>,216): ScenarioTable<Row, unknown, unknown> {217 const facade = (peer.db as Record<string, ScenarioTable<Row, unknown, unknown>>)[table._table];218 if (facade === undefined) {219 throw new ScenarioError(220 "assertion",221 `table ${table._table} is not part of ${peer.name}'s app`,222 { peer: peer.name },223 );224 }225 return facade;226}227228/**229 * Assert that a peer's local view contains the row, optionally matching a230 * subset of its columns.231 */232export async function assertRow<Row>(233 peer: ScenarioPeer<unknown>,234 table: ScenarioTableRef<Row>,235 id: string,236 partial?: Partial<Row>,237): Promise<void> {238 const row = await tableFacade(peer, table).get(id);239 if (row === undefined) {240 throw new ScenarioError(241 "assertion",242 `expected ${peer.name} to see row ${id} in ${table._table}, but it is absent`,243 { peer: peer.name },244 );245 }246 for (const [column, expected] of Object.entries(partial ?? {})) {247 const actual = (row as Record<string, unknown>)[column];248 if (stableSerialize(actual) !== stableSerialize(expected)) {249 throw new ScenarioError(250 "assertion",251 `expected ${table._table} row ${id} on ${peer.name} to have ` +252 `${column} = ${stableSerialize(expected)}, found ${stableSerialize(actual)}`,253 { peer: peer.name },254 );255 }256 }257}258259/** Assert that a peer's local view does not contain the row. */260export async function assertNoRow(261 peer: ScenarioPeer<unknown>,262 table: ScenarioTableRef<unknown>,263 id: string,264): Promise<void> {265 const row = await tableFacade(peer, table).get(id);266 if (row !== undefined) {267 throw new ScenarioError(268 "assertion",269 `expected ${peer.name} to no longer see row ${id} in ${table._table}, but it is present`,270 { peer: peer.name },271 );272 }273}274275/** Assert the number of rows a peer's local view of a table contains. */276export async function assertRowCount(277 peer: ScenarioPeer<unknown>,278 table: ScenarioTableRef<unknown>,279 expected: number,280): Promise<void> {281 const rows = await tableFacade(peer, table).all();282 if (rows.length !== expected) {283 throw new ScenarioError(284 "assertion",285 `expected ${table._table} on ${peer.name} to hold ${expected} row(s), found ${rows.length}`,286 { peer: peer.name },287 );288 }289}