From 50de982847d5a551338a6eb22fabb532b6c096ae Mon Sep 17 00:00:00 2001 From: quartune <123062848+quartune@users.noreply.github.com> Date: Wed, 21 Oct 2026 09:17:46 +0000 Subject: [PATCH] feat(realtime): typed transport, mock server, per-topic flags, merge layer Closes #195 Closes #196 Closes #203 Closes #204 --- README.md | 8 + docs/realtime-protocol.md | 26 ++ docs/realtime-rollout.md | 22 + package.json | 3 +- playwright.config.ts | 27 +- public/flags.json | 6 +- scripts/mock-realtime-server.mjs | 255 +++++++++++ src/lib/flags/flagsConfig.ts | 4 + src/lib/flags/useFeatureFlag.ts | 3 +- src/lib/realtime/__tests__/client.test.ts | 199 +++++++++ src/lib/realtime/__tests__/flags.test.ts | 39 ++ src/lib/realtime/__tests__/merge.test.ts | 77 ++++ src/lib/realtime/__tests__/mockServer.test.ts | 96 ++++ src/lib/realtime/client.ts | 420 ++++++++++++++++-- src/lib/realtime/flags.ts | 29 ++ src/lib/realtime/merge.ts | 72 +++ src/lib/realtime/protocol.ts | 90 ++++ 17 files changed, 1316 insertions(+), 60 deletions(-) create mode 100644 docs/realtime-protocol.md create mode 100644 docs/realtime-rollout.md create mode 100644 scripts/mock-realtime-server.mjs create mode 100644 src/lib/realtime/__tests__/client.test.ts create mode 100644 src/lib/realtime/__tests__/flags.test.ts create mode 100644 src/lib/realtime/__tests__/merge.test.ts create mode 100644 src/lib/realtime/__tests__/mockServer.test.ts create mode 100644 src/lib/realtime/flags.ts create mode 100644 src/lib/realtime/merge.ts create mode 100644 src/lib/realtime/protocol.ts diff --git a/README.md b/README.md index 780aba7..bdd3476 100644 --- a/README.md +++ b/README.md @@ -63,3 +63,11 @@ Other environment variables: ## License [MIT](./LICENSE) + +## Realtime layer (dev) + +Realtime is off by default (`public/flags.json`; see `docs/realtime-protocol.md` and `docs/realtime-rollout.md`). +Run the deterministic mock server with `npm run mock:realtime` (prints `MOCK_REALTIME_URL=…`; add `--port=4010`). +Scenario flags: `--seed=N --count=N --dropAfter=N --latencyMs=N --duplicate --reorder --unknown --authFail`, or the same as query params on the WebSocket URL. +Playwright starts it on port 4010 automatically; point the app at it with `NEXT_PUBLIC_API_URL=http://127.0.0.1:4010/api/v1` +and `NEXT_PUBLIC_FLAGS_OVERRIDE='{"realtime-enabled":true,"realtime-oracle":true}'`. diff --git a/docs/realtime-protocol.md b/docs/realtime-protocol.md new file mode 100644 index 0000000..acc08a7 --- /dev/null +++ b/docs/realtime-protocol.md @@ -0,0 +1,26 @@ +# Realtime protocol (v1) + +WebSocket at `ws(s):///ws` (derived from `NEXT_PUBLIC_API_URL` by `deriveRealtimeUrl`). Types live in `src/lib/realtime/protocol.ts`; the transport is `src/lib/realtime/client.ts`; the dev/CI server is `scripts/mock-realtime-server.mjs`. + +## Server → client frame + +```json +{ "v": 1, "topic": "oracle", "seq": 12, "ts": 1700000012000, "payload": {} } +``` + +- `topic`: `oracle` | `claims` | `pool` | `notifications`. `seq` is per-topic and monotonic; `ts` is epoch ms and is the ordering/version key for the merge layer. +- Frames with an unknown `topic` or `v`, or that fail validation, are ignored and logged (reason only). +- Payloads: `oracle {id, price, updatedAt}`, `claims {id, policyId, status, amount, updatedAt}`, `pool {id, tvl, utilization, updatedAt}`, `notifications {id, kind, message, createdAt}`. +- `{"op":"pong"}` answers a client ping. + +## Client → server + +`{"op":"subscribe","topic":"oracle","cursor":12}` (cursor optional; the server replays frames with `seq > cursor`), `{"op":"unsubscribe","topic":"oracle"}`, `{"op":"ping"}`. + +## Close codes + +`4401` auth rejected — the client does not reconnect. Any other close triggers capped, jittered exponential reconnect (1s base, 30s cap) that re-subscribes with each topic's last `seq`. + +## Merge rules + +`src/lib/realtime/merge.ts` folds events into a REST snapshot: idempotent, version (`ts`) ordered, stale events dropped, unknown ids inserted, removals tombstoned. diff --git a/docs/realtime-rollout.md b/docs/realtime-rollout.md new file mode 100644 index 0000000..509b556 --- /dev/null +++ b/docs/realtime-rollout.md @@ -0,0 +1,22 @@ +# Realtime rollout and rollback runbook + +## Flags (`public/flags.json`) + +`realtime-enabled` (global kill switch) plus `realtime-oracle`, `realtime-claims`, `realtime-pool`, `realtime-notifications`. A topic is live only when **both** the kill switch and its own flag are `true` (`isRealtimeTopicEnabled`, `src/lib/realtime/flags.ts` — the single helper every consumer must use). All default to `false`. Missing, malformed or stale flag documents resolve to off. The file is public: booleans only, no secrets or internal hostnames. + +Flags are read from the same cached `/flags.json` in every tab, so tabs converge within `FLAGS_CACHE_TTL_MS` (5 min). Turning a topic off calls `realtimeClient.setFlags(...)`: the topic is unsubscribed on the wire and each subscriber's `onDisabled` fires so its hook falls back to REST — never a live subscription rendering nothing. + +## Stages + +| Stage | Audience | Advance when (over ≥ 24h) | +| --- | --- | --- | +| 1. mock/dev | `npm run mock:realtime` + `NEXT_PUBLIC_FLAGS_OVERRIDE` | smoke tests green | +| 2. internal | staging, one topic at a time (oracle → pool → claims → notifications) | connection success ≥ 99%, reconnects < 1 per session-hour, REST fallback < 1% of sessions | +| 3. allowlist / percentage | ship flag with wider audience | same thresholds hold at each step | +| 4. default-on | flip `realtime-enabled` and topic flags to `true` in `flags.json` | thresholds hold for a week | + +## Rollback + +1. Single topic: set that topic's flag to `false` in `public/flags.json` and deploy/publish the file. Everything: set `realtime-enabled` to `false`. +2. Verify: open the app, confirm no WebSocket to `/ws` for the topic in devtools → Network → WS, and that the hook's data still loads (REST). `realtimeClient.getState()` should be `idle`. +3. Notify: the on-call/maintainers channel and the backend owners of `refract-backend`; note the topic, time and symptom. diff --git a/package.json b/package.json index 51f59c9..3fb74fe 100644 --- a/package.json +++ b/package.json @@ -19,7 +19,8 @@ "analyze": "ANALYZE=true next build", "size": "size-limit", "size:why": "size-limit --why", - "tokens:check": "node scripts/check-tokens.mjs" + "tokens:check": "node scripts/check-tokens.mjs", + "mock:realtime": "node scripts/mock-realtime-server.mjs" }, "lint-staged": { "*.{ts,tsx}": [ diff --git a/playwright.config.ts b/playwright.config.ts index 2bfede7..0d6e75a 100644 --- a/playwright.config.ts +++ b/playwright.config.ts @@ -37,12 +37,23 @@ export default defineConfig({ use: { ...devices["Desktop Safari"] }, }, ], - webServer: process.env.PLAYWRIGHT_BASE_URL - ? undefined - : { - command: "npm run dev", - url: "http://127.0.0.1:3000", - reuseExistingServer: !process.env.CI, - timeout: 120_000, - }, + webServer: [ + // Deterministic mock realtime server on a fixed port for browser tests. + { + command: "node scripts/mock-realtime-server.mjs --port=4010", + port: 4010, + reuseExistingServer: !process.env.CI, + timeout: 30_000, + }, + ...(process.env.PLAYWRIGHT_BASE_URL + ? [] + : [ + { + command: "npm run dev", + url: "http://127.0.0.1:3000", + reuseExistingServer: !process.env.CI, + timeout: 120_000, + }, + ]), + ], }); diff --git a/public/flags.json b/public/flags.json index f8832b6..744425d 100644 --- a/public/flags.json +++ b/public/flags.json @@ -1,5 +1,9 @@ { "flags": { - "realtime-oracle": false + "realtime-enabled": false, + "realtime-oracle": false, + "realtime-claims": false, + "realtime-pool": false, + "realtime-notifications": false } } diff --git a/scripts/mock-realtime-server.mjs b/scripts/mock-realtime-server.mjs new file mode 100644 index 0000000..83ff731 --- /dev/null +++ b/scripts/mock-realtime-server.mjs @@ -0,0 +1,255 @@ +#!/usr/bin/env node +/** + * Dependency-free mock realtime server implementing docs/realtime-protocol.md. + * Dev/CI only — never imported by the app. Binds to an ephemeral port by default + * and prints `MOCK_REALTIME_URL=ws://127.0.0.1:/ws`. + * + * Scenarios are driven by one config object, from CLI flags (`--dropAfter=3`) or + * per-connection query params (`ws://…/ws?dropAfter=3&duplicate=1`): + * seed, count (frames/topic, default 5), dropAfter (close after N frames), + * latencyMs, duplicate, reorder, unknown (emit unknown topic/version frames), + * authFail (close 4401). Payloads are seeded; timestamps are synthetic + * (BASE_TS + seq * 1000), so output is identical run to run. + */ +import { createHash } from "node:crypto"; +import { createServer } from "node:http"; +import { pathToFileURL } from "node:url"; + +const GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; +const BASE_TS = 1_700_000_000_000; +const TOPICS = ["oracle", "claims", "pool", "notifications"]; +const DEFAULTS = { + seed: 1, + count: 5, + dropAfter: 0, + latencyMs: 0, + duplicate: false, + reorder: false, + unknown: false, + authFail: false, +}; + +function mulberry32(a) { + return () => { + a = (a + 0x6d2b79f5) | 0; + let t = Math.imul(a ^ (a >>> 15), 1 | a); + t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t; + return ((t ^ (t >>> 14)) >>> 0) / 4294967296; + }; +} + +const num = (v, d) => (v === undefined || v === "" || Number.isNaN(Number(v)) ? d : Number(v)); +const bool = (v, d) => (v === undefined ? d : v === true || v === "1" || v === "true"); + +/** Merges defaults ← base ← overrides (string or typed values) into a normalized scenario. */ +export function resolveScenario(...layers) { + const raw = Object.assign({}, ...layers); + return { + seed: num(raw.seed, DEFAULTS.seed), + count: num(raw.count, DEFAULTS.count), + dropAfter: num(raw.dropAfter, DEFAULTS.dropAfter), + latencyMs: num(raw.latencyMs, DEFAULTS.latencyMs), + duplicate: bool(raw.duplicate, DEFAULTS.duplicate), + reorder: bool(raw.reorder, DEFAULTS.reorder), + unknown: bool(raw.unknown, DEFAULTS.unknown), + authFail: bool(raw.authFail, DEFAULTS.authFail), + }; +} + +function payloadFor(topic, seq, rnd) { + const ts = BASE_TS + seq * 1000; + const n = Math.floor(rnd() * 1_000_000); + switch (topic) { + case "oracle": + return { id: `feed-${seq % 2}`, price: (n / 10000).toFixed(4), updatedAt: ts }; + case "claims": + return { + id: `claim-${seq}`, + policyId: `policy-${n % 3}`, + status: ["pending", "approved", "paid"][n % 3], + amount: String(n), + updatedAt: ts, + }; + case "pool": + return { id: "pool-0", tvl: String(n * 100), utilization: (n % 100) / 100, updatedAt: ts }; + default: + return { id: `note-${seq}`, kind: "info", message: `Notification ${seq}`, createdAt: ts }; + } +} + +/** The frames a subscriber to `topic` receives, in documented order. Pure and deterministic. */ +export function framesFor(topic, scenario, cursor = 0) { + const rnd = mulberry32(scenario.seed + TOPICS.indexOf(topic) * 7919); + let frames = []; + for (let seq = 1; seq <= scenario.count; seq++) { + const payload = payloadFor(topic, seq, rnd); + if (seq > cursor) frames.push({ v: 1, topic, seq, ts: BASE_TS + seq * 1000, payload }); + } + if (scenario.reorder) { + for (let i = 0; i + 1 < frames.length; i += 2) + [frames[i], frames[i + 1]] = [frames[i + 1], frames[i]]; + } + if (scenario.duplicate) frames = frames.flatMap((f) => [f, f]); + if (scenario.unknown && frames.length) { + frames.splice( + 1, + 0, + { v: 1, topic: "mystery", seq: 1, ts: BASE_TS, payload: {} }, + { v: 99, topic, seq: 1, ts: BASE_TS, payload: {} } + ); + } + return frames; +} + +function encode(text) { + const body = Buffer.from(text); + const len = body.length; + const head = len < 126 ? Buffer.from([0x81, len]) : Buffer.from([0x81, 126, len >> 8, len & 255]); + return Buffer.concat([head, body]); +} + +function closeFrame(code) { + return Buffer.from([0x88, 2, code >> 8, code & 255]); +} + +/** Decodes complete client frames from `buf`; returns [frames, remaining]. */ +function decode(buf) { + const out = []; + while (buf.length >= 2) { + const opcode = buf[0] & 15; + let len = buf[1] & 127; + let off = 2; + if (len === 126) { + if (buf.length < 4) break; + len = buf.readUInt16BE(2); + off = 4; + } + const masked = (buf[1] & 128) !== 0; + const end = off + (masked ? 4 : 0) + len; + if (buf.length < end) break; + const data = Buffer.from(buf.subarray(end - len, end)); + if (masked) for (let i = 0; i < len; i++) data[i] ^= buf[off + (i % 4)]; + out.push({ opcode, data }); + buf = buf.subarray(end); + } + return [out, buf]; +} + +/** + * Starts the server. `port: 0` (default) picks a free port. Resolves with + * `{ url, port, connections(), close() }`. + */ +export async function startMockRealtimeServer({ + port = 0, + host = "127.0.0.1", + scenario = {}, +} = {}) { + const base = resolveScenario(scenario); + let connections = 0; + const sockets = new Set(); + const server = createServer((_, res) => { + res.writeHead(426).end("websocket only"); + }); + + server.on("upgrade", (req, socket) => { + const key = req.headers["sec-websocket-key"]; + if (!key) return socket.destroy(); + const url = new URL(req.url, "http://x"); + const sc = resolveScenario(base, Object.fromEntries(url.searchParams)); + const accept = createHash("sha1") + .update(key + GUID) + .digest("base64"); + socket.write( + `HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n` + ); + connections++; + sockets.add(socket); + socket.on("close", () => sockets.delete(socket)); + socket.on("error", () => {}); + + if (sc.authFail) { + socket.end(closeFrame(4401)); + return; + } + + let pending = Buffer.alloc(0); + let sent = 0; + let dropped = false; + const timers = new Set(); + socket.on("close", () => timers.forEach(clearTimeout)); + + const emit = (frames) => { + frames.forEach((f, i) => { + const t = setTimeout( + () => { + timers.delete(t); + if (dropped || socket.destroyed) return; + socket.write(encode(JSON.stringify(f))); + sent++; + if (sc.dropAfter && sent >= sc.dropAfter) { + dropped = true; + socket.destroy(); + } + }, + sc.latencyMs * (i + 1) + ); + timers.add(t); + }); + }; + + socket.on("data", (chunk) => { + let frames; + [frames, pending] = decode(Buffer.concat([pending, chunk])); + for (const { opcode, data } of frames) { + if (opcode === 8) return socket.end(closeFrame(1000)); + if (opcode === 9) continue; + let msg; + try { + msg = JSON.parse(data.toString()); + } catch { + continue; + } + if (msg.op === "ping") socket.write(encode('{"op":"pong"}')); + else if (msg.op === "subscribe" && TOPICS.includes(msg.topic)) + emit(framesFor(msg.topic, sc, Number(msg.cursor) || 0)); + } + }); + }); + + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(port, host, resolve); + }); + const bound = server.address().port; + return { + port: bound, + url: `ws://${host}:${bound}/ws`, + connections: () => connections, + close: () => + new Promise((resolve) => { + sockets.forEach((s) => s.destroy()); + server.close(() => resolve()); + }), + }; +} + +if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { + const flags = Object.fromEntries( + process.argv + .slice(2) + .filter((a) => a.startsWith("--")) + .map((a) => { + const [k, v = "1"] = a.slice(2).split("="); + return [k, v]; + }) + ); + const { port, ...scenario } = flags; + const srv = await startMockRealtimeServer({ + port: Number(port ?? process.env.MOCK_REALTIME_PORT ?? 0), + scenario, + }); + console.log(`MOCK_REALTIME_URL=${srv.url}`); + const stop = () => srv.close().then(() => process.exit(0)); + process.on("SIGINT", stop); + process.on("SIGTERM", stop); +} diff --git a/src/lib/flags/flagsConfig.ts b/src/lib/flags/flagsConfig.ts index 2ecb795..d325f1d 100644 --- a/src/lib/flags/flagsConfig.ts +++ b/src/lib/flags/flagsConfig.ts @@ -4,7 +4,11 @@ export interface FlagsDocument { /** Safe default if the remote flags source is unreachable — new/risky features default off. */ export const DEFAULT_FLAGS: Record = { + "realtime-enabled": false, "realtime-oracle": false, + "realtime-claims": false, + "realtime-pool": false, + "realtime-notifications": false, }; export const FLAGS_ENDPOINT = "/flags.json"; diff --git a/src/lib/flags/useFeatureFlag.ts b/src/lib/flags/useFeatureFlag.ts index a6ec070..aff063f 100644 --- a/src/lib/flags/useFeatureFlag.ts +++ b/src/lib/flags/useFeatureFlag.ts @@ -1,6 +1,7 @@ "use client"; import { useEffect, useState } from "react"; +import { sanitizeFlags } from "../realtime/flags"; import { DEFAULT_FLAGS, FLAGS_CACHE_TTL_MS, FLAGS_ENDPOINT, readLocalOverride, type FlagsDocument } from "./flagsConfig"; let cache: { flags: Record; fetchedAt: number } | null = null; @@ -16,7 +17,7 @@ async function loadFlags(): Promise> { inflight = fetch(FLAGS_ENDPOINT) .then((res) => (res.ok ? (res.json() as Promise) : Promise.reject(new Error("flags fetch failed")))) .then((doc) => { - const flags = { ...DEFAULT_FLAGS, ...doc.flags }; + const flags = { ...DEFAULT_FLAGS, ...sanitizeFlags(doc?.flags) }; cache = { flags, fetchedAt: Date.now() }; return flags; }) diff --git a/src/lib/realtime/__tests__/client.test.ts b/src/lib/realtime/__tests__/client.test.ts new file mode 100644 index 0000000..45d6589 --- /dev/null +++ b/src/lib/realtime/__tests__/client.test.ts @@ -0,0 +1,199 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { RealtimeClient, type SocketLike } from "../client"; + +class FakeSocket implements SocketLike { + onopen: SocketLike["onopen"] = null; + onmessage: SocketLike["onmessage"] = null; + onclose: SocketLike["onclose"] = null; + onerror: SocketLike["onerror"] = null; + sent: Record[] = []; + closed = 0; + send(d: string) { + this.sent.push(JSON.parse(d)); + } + close() { + this.closed++; + } + open() { + this.onopen?.({}); + } + push(frame: unknown) { + this.onmessage?.({ data: typeof frame === "string" ? frame : JSON.stringify(frame) }); + } + drop(code = 1006) { + this.onclose?.({ code }); + } +} + +const ON = { "realtime-enabled": true, "realtime-oracle": true, "realtime-claims": true }; +const frame = (topic: string, seq: number) => ({ v: 1, topic, seq, ts: seq, payload: { id: "x" } }); + +let sockets: FakeSocket[]; +const logger = { warn: vi.fn() }; +const make = (extra = {}) => + new RealtimeClient({ + url: "ws://t/ws", + socketFactory: () => { + const s = new FakeSocket(); + sockets.push(s); + return s; + }, + random: () => 1, + flags: ON, + logger, + ...extra, + }); + +beforeEach(() => { + vi.useFakeTimers(); + sockets = []; + logger.warn.mockClear(); +}); +afterEach(() => vi.useRealTimers()); + +describe("RealtimeClient", () => { + it("does nothing while flags are off", () => { + const c = make({ flags: {} }); + c.subscribe("oracle", () => {}); + expect(sockets).toHaveLength(0); + expect(c.getState()).toBe("idle"); + }); + + it("walks the state machine and delivers typed messages", () => { + const c = make(); + const states: string[] = []; + c.observeState((s) => states.push(s)); + const got: number[] = []; + c.subscribe("oracle", (m) => got.push(m.seq)); + sockets[0].open(); + sockets[0].push(frame("oracle", 1)); + sockets[0].push(frame("claims", 9)); + expect(states).toEqual(["idle", "connecting", "open"]); + expect(got).toEqual([1]); + expect(sockets[0].sent).toEqual([{ op: "subscribe", topic: "oracle" }]); + }); + + it("follows the capped, jittered backoff schedule and resumes from the cursor", () => { + const c = make({ random: () => 0 }); // equal jitter floor = ceiling/2 + c.subscribe("oracle", () => {}); + sockets[0].open(); + sockets[0].push(frame("oracle", 7)); + const delays: number[] = []; + for (let i = 0; i < 8; i++) { + sockets[i].drop(); + expect(c.getState()).toBe("reconnecting"); + const before = sockets.length; + let waited = 0; + while (sockets.length === before) { + vi.advanceTimersByTime(100); + waited += 100; + } + delays.push(waited); + } + expect(delays).toEqual([500, 1000, 2000, 4000, 8000, 15000, 15000, 15000]); + sockets[sockets.length - 1].open(); + expect(sockets[sockets.length - 1].sent).toEqual([ + { op: "subscribe", topic: "oracle", cursor: 7 }, + ]); + }); + + it("ignores malformed, unknown-topic and unknown-version frames without throwing or leaking payloads", () => { + const c = make(); + const handler = vi.fn(); + c.subscribe("oracle", handler); + sockets[0].open(); + for (const bad of [ + "{nope", + "[]", + frame("mystery", 1), + { ...frame("oracle", 1), v: 99 }, + { v: 1, topic: "oracle", seq: "x", ts: 1, payload: {} }, + ]) { + sockets[0].push(bad); + } + expect(handler).not.toHaveBeenCalled(); + expect(logger.warn).toHaveBeenCalledTimes(5); + expect(JSON.stringify(logger.warn.mock.calls)).not.toContain('"id":"x"'); + }); + + it("reconnects when the heartbeat goes stale", () => { + const c = make({ heartbeatIntervalMs: 1000, staleAfterMs: 2500, random: () => 0 }); + c.subscribe("oracle", () => {}); + sockets[0].open(); + vi.advanceTimersByTime(2000); + expect(sockets[0].sent.filter((m) => m.op === "ping")).toHaveLength(2); + vi.advanceTimersByTime(1000); // 3000ms silent > 2500 + expect(c.getState()).toBe("reconnecting"); + expect(sockets[0].closed).toBe(1); + vi.advanceTimersByTime(600); + expect(sockets).toHaveLength(2); + }); + + it("any inbound frame keeps the connection alive", () => { + const c = make({ heartbeatIntervalMs: 1000, staleAfterMs: 2500 }); + c.subscribe("oracle", () => {}); + sockets[0].open(); + for (let i = 0; i < 5; i++) { + vi.advanceTimersByTime(1000); + sockets[0].push({ op: "pong" }); + } + expect(c.getState()).toBe("open"); + }); + + it("tears down cleanly: no timers or sockets left", () => { + const c = make(); + const h = c.subscribe("oracle", () => {}); + sockets[0].open(); + c.unsubscribe(h); + expect(c.getState()).toBe("idle"); + expect(sockets[0].closed).toBe(1); + expect(vi.getTimerCount()).toBe(0); + c.subscribe("claims", () => {}); + sockets[1].drop(); + c.close(); + expect(c.getState()).toBe("closed"); + expect(vi.getTimerCount()).toBe(0); + sockets[1].drop(); // double-close is harmless + expect(sockets).toHaveLength(2); + }); + + it("survives rapid subscribe/unsubscribe churn", () => { + const c = make(); + for (let i = 0; i < 50; i++) c.unsubscribe(c.subscribe("oracle", () => {})); + expect(vi.getTimerCount()).toBe(0); + expect(c.getState()).toBe("idle"); + const keep = c.subscribe("oracle", () => {}); + const extra = c.subscribe("oracle", () => {}); + sockets[sockets.length - 1].open(); + c.unsubscribe(extra); + expect(sockets[sockets.length - 1].sent.filter((m) => m.op === "unsubscribe")).toHaveLength(0); + c.unsubscribe(keep); + expect(c.getState()).toBe("idle"); + }); + + it("does not reconnect after an auth rejection", () => { + const c = make(); + c.subscribe("oracle", () => {}); + sockets[0].open(); + sockets[0].drop(4401); + vi.advanceTimersByTime(60_000); + expect(c.getState()).toBe("closed"); + expect(sockets).toHaveLength(1); + }); + + it("flipping a topic off suspends it, notifies, and unsubscribes on the wire", () => { + const c = make(); + const onDisabled = vi.fn(); + const handler = vi.fn(); + c.subscribe("oracle", handler, { onDisabled }); + c.subscribe("claims", () => {}); + sockets[0].open(); + c.setFlags({ ...ON, "realtime-oracle": false }); + expect(onDisabled).toHaveBeenCalledOnce(); + expect(sockets[0].sent).toContainEqual({ op: "unsubscribe", topic: "oracle" }); + sockets[0].push(frame("oracle", 1)); + expect(handler).not.toHaveBeenCalled(); + c.setFlags({ ...ON, "realtime-enabled": false }); + expect(c.getState()).toBe("idle"); + }); +}); diff --git a/src/lib/realtime/__tests__/flags.test.ts b/src/lib/realtime/__tests__/flags.test.ts new file mode 100644 index 0000000..8708668 --- /dev/null +++ b/src/lib/realtime/__tests__/flags.test.ts @@ -0,0 +1,39 @@ +import { readFileSync } from "node:fs"; +import { describe, expect, it } from "vitest"; +import { isRealtimeTopicEnabled, sanitizeFlags } from "../flags"; +import { REALTIME_TOPICS } from "../protocol"; + +describe("realtime flags", () => { + it("requires both the kill switch and the topic flag", () => { + expect(isRealtimeTopicEnabled({ "realtime-oracle": true }, "oracle")).toBe(false); + expect(isRealtimeTopicEnabled({ "realtime-enabled": true }, "oracle")).toBe(false); + expect( + isRealtimeTopicEnabled({ "realtime-enabled": true, "realtime-oracle": true }, "oracle") + ).toBe(true); + expect( + isRealtimeTopicEnabled({ "realtime-enabled": true, "realtime-oracle": true }, "claims") + ).toBe(false); + }); + + it("kill switch dominates per-topic flags", () => { + const all = Object.fromEntries(REALTIME_TOPICS.map((t) => [`realtime-${t}`, true])); + for (const t of REALTIME_TOPICS) + expect(isRealtimeTopicEnabled({ ...all, "realtime-enabled": false }, t)).toBe(false); + }); + + it("is safe for missing or malformed documents", () => { + for (const bad of [undefined, null, "x", 3, [], { "realtime-enabled": "true" }]) { + expect(isRealtimeTopicEnabled(sanitizeFlags(bad), "oracle")).toBe(false); + } + expect(sanitizeFlags({ a: true, b: "yes", c: 1 })).toEqual({ a: true }); + }); + + it("committed flags.json keeps every realtime flag off and carries only booleans", () => { + const { flags } = JSON.parse(readFileSync("public/flags.json", "utf8")) as { + flags: Record; + }; + expect(flags["realtime-enabled"]).toBe(false); + for (const t of REALTIME_TOPICS) expect(flags[`realtime-${t}`]).toBe(false); + for (const v of Object.values(flags)) expect(typeof v).toBe("boolean"); + }); +}); diff --git a/src/lib/realtime/__tests__/merge.test.ts b/src/lib/realtime/__tests__/merge.test.ts new file mode 100644 index 0000000..a06c517 --- /dev/null +++ b/src/lib/realtime/__tests__/merge.test.ts @@ -0,0 +1,77 @@ +import { describe, expect, it } from "vitest"; +import { createMerge, type MergeEvent } from "../merge"; + +interface Row { + id: string; + v: number; + n: string; +} +const m = createMerge({ getId: (r) => r.id, getVersion: (r) => r.v }); +const up = (id: string, v: number, n = `${id}@${v}`): MergeEvent => ({ + type: "upsert", + id, + version: v, + data: { id, v, n }, +}); +const rm = (id: string, v: number): MergeEvent => ({ type: "remove", id, version: v }); +const sorted = (rows: Row[]) => [...rows].sort((a, b) => a.id.localeCompare(b.id)); + +describe("merge", () => { + it("inserts entities missing from the snapshot", () => { + const s = m.reduce(m.init([{ id: "a", v: 1, n: "a" }]), up("b", 2)); + expect(m.view(s).map((r) => r.id)).toEqual(["a", "b"]); + }); + + it("ignores duplicates and returns the same state reference", () => { + const s1 = m.reduce(m.init(), up("a", 5)); + expect(m.reduce(s1, up("a", 5))).toBe(s1); + expect(m.view(s1)).toHaveLength(1); + }); + + it("drops stale and out-of-order events", () => { + const s = m.applyAll(m.init([{ id: "a", v: 10, n: "snap" }]), [ + up("a", 3), + up("a", 12), + up("a", 11), + ]); + expect(m.view(s)[0].n).toBe("a@12"); + }); + + it("removes and does not let a stale upsert resurrect", () => { + let s = m.applyAll(m.init([{ id: "a", v: 1, n: "a" }]), [rm("a", 5), up("a", 4)]); + expect(m.view(s)).toEqual([]); + s = m.reduce(s, up("a", 6)); + expect(m.view(s)).toHaveLength(1); + }); + + it("removal of an unknown entity is remembered as a tombstone", () => { + const s = m.applyAll(m.init(), [rm("x", 5), up("x", 2)]); + expect(m.view(s)).toEqual([]); + }); + + it("is idempotent and order-independent under shuffled duplicate streams", () => { + const events = [ + up("a", 1), + up("a", 4), + up("b", 2), + up("c", 3), + up("b", 7), + rm("c", 9), + up("d", 5), + up("a", 8), + ]; + const expected = sorted(m.view(m.applyAll(m.init(), events))); + let seed = 42; + const rnd = () => (seed = (seed * 1664525 + 1013904223) >>> 0) / 4294967296; + for (let run = 0; run < 50; run++) { + const stream = [...events, ...events.filter(() => rnd() > 0.5)]; + for (let i = stream.length - 1; i > 0; i--) { + const j = Math.floor(rnd() * (i + 1)); + [stream[i], stream[j]] = [stream[j], stream[i]]; + } + const once = m.applyAll(m.init(), stream); + expect(sorted(m.view(once))).toEqual(expected); + expect(m.applyAll(once, stream)).toBe(once); + } + }); +}); diff --git a/src/lib/realtime/__tests__/mockServer.test.ts b/src/lib/realtime/__tests__/mockServer.test.ts new file mode 100644 index 0000000..6ca0dc6 --- /dev/null +++ b/src/lib/realtime/__tests__/mockServer.test.ts @@ -0,0 +1,96 @@ +// @vitest-environment node +import { afterEach, describe, expect, it } from "vitest"; +// @ts-expect-error plain ESM script without types +import { + framesFor, + resolveScenario, + startMockRealtimeServer, +} from "../../../../scripts/mock-realtime-server.mjs"; +import { RealtimeClient } from "../client"; + +type Server = Awaited>; +const servers: Server[] = []; +const clients: RealtimeClient[] = []; +afterEach(async () => { + clients.splice(0).forEach((c) => c.close()); + await Promise.all(servers.splice(0).map((s) => s.close())); +}); + +async function setup(scenario = {}, flags = { "realtime-enabled": true, "realtime-oracle": true }) { + const server: Server = await startMockRealtimeServer({ scenario }); + servers.push(server); + const client = new RealtimeClient({ + url: server.url, + flags, + baseBackoffMs: 20, + maxBackoffMs: 40, + logger: { warn: () => {} }, + }); + clients.push(client); + return { server, client }; +} +const waitFor = async (cond: () => boolean) => { + for (let i = 0; i < 200 && !cond(); i++) await new Promise((r) => setTimeout(r, 10)); + expect(cond()).toBe(true); +}; + +describe("mock realtime server", () => { + it("is deterministic for a seed", () => { + const sc = resolveScenario({ seed: 3 }); + expect(framesFor("oracle", sc)).toEqual(framesFor("oracle", sc)); + expect(framesFor("oracle", sc)).not.toEqual(framesFor("oracle", resolveScenario({ seed: 4 }))); + }); + + it("documents scenario frames", () => { + const base = resolveScenario({ count: 4 }); + expect(framesFor("pool", base).map((f: { seq: number }) => f.seq)).toEqual([1, 2, 3, 4]); + expect( + framesFor("pool", resolveScenario({ count: 4, reorder: true })).map( + (f: { seq: number }) => f.seq + ) + ).toEqual([2, 1, 4, 3]); + expect( + framesFor("pool", resolveScenario({ count: 2, duplicate: true })).map( + (f: { seq: number }) => f.seq + ) + ).toEqual([1, 1, 2, 2]); + expect( + framesFor("pool", resolveScenario({ count: 3, unknown: true })).map( + (f: { topic: string }) => f.topic + ) + ).toEqual(["pool", "mystery", "pool", "pool", "pool"]); + expect(framesFor("pool", base, 2).map((f: { seq: number }) => f.seq)).toEqual([3, 4]); + }); + + it("streams frames to a real client end to end", async () => { + const { client } = await setup({ count: 3 }); + const seqs: number[] = []; + client.subscribe("oracle", (m) => seqs.push(m.seq)); + await waitFor(() => seqs.length === 3); + expect(seqs).toEqual([1, 2, 3]); + }); + + it("drops the connection and the client resumes from its cursor", async () => { + const { client, server } = await setup({ count: 4, dropAfter: 2 }); + const seqs: number[] = []; + client.subscribe("oracle", (m) => seqs.push(m.seq)); + await waitFor(() => seqs.length >= 4 && server.connections() >= 2); + expect(seqs.slice(0, 4)).toEqual([1, 2, 3, 4]); + }); + + it("rejects auth and the client stops", async () => { + const { client } = await setup({ authFail: true }); + client.subscribe("oracle", () => {}); + await waitFor(() => client.getState() === "closed"); + }); + + it("opens no connection when the topic flag is off", async () => { + const { client, server } = await setup( + {}, + { "realtime-enabled": true, "realtime-oracle": false } + ); + client.subscribe("oracle", () => {}); + await new Promise((r) => setTimeout(r, 100)); + expect(server.connections()).toBe(0); + }); +}); diff --git a/src/lib/realtime/client.ts b/src/lib/realtime/client.ts index 8be8d3c..3918ceb 100644 --- a/src/lib/realtime/client.ts +++ b/src/lib/realtime/client.ts @@ -1,87 +1,409 @@ -import { API_BASE_URL } from "@/lib/api/client"; - -type Handler = (payload: unknown) => void; - -const MAX_BACKOFF_MS = 30_000; -const BASE_BACKOFF_MS = 1_000; +import { API_BASE_URL } from "../api/client"; +import { DEFAULT_FLAGS } from "../flags/flagsConfig"; +import { isRealtimeTopicEnabled, sanitizeFlags } from "./flags"; +import { + CLOSE_AUTH_FAILED, + parseFrame, + type ClientOp, + type RealtimeTopic, + type TopicMessageMap, +} from "./protocol"; /** - * A small reconnecting WebSocket wrapper exposing a subscribe-by-topic API. - * The backend contract this assumes (documented, not yet implemented - * upstream): connect to `${wsUrl}`, receive JSON messages shaped - * `{ topic: string, payload: unknown }`, one topic per push-based update - * type (e.g. "oracle_update"). Callers must treat this connection as - * best-effort — always keep an independent REST fetch as the source of - * truth for initial state. + * Typed, reconnecting realtime transport (contract: docs/realtime-protocol.md). + * + * - `subscribe(topic, handler, opts)` / `unsubscribe(handle)`; handlers receive the + * topic's typed message. One server subscription per topic, ref-counted. + * - State machine: idle → connecting → open → reconnecting → closed, observable via + * `observeState`. The socket is opened lazily on the first enabled subscription and + * closed (back to idle) when the last one goes away, so nothing leaks. + * - Reconnect uses capped exponential backoff with equal jitter; a heartbeat ping runs + * while open and a connection silent for `staleAfterMs` is force-reconnected. + * Reconnects resume each topic from its last seen `seq` (cursor replay). + * - Inbound frames are validated; malformed/unknown topic/version frames are logged + * (reason only, never payloads) and dropped. Auth close (4401) is terminal. + * - Gated by flags (`realtime-enabled` + per-topic): a disabled topic is never sent to + * the server; flipping it off via `setFlags` suspends it and fires `onDisabled` so + * callers fall back to REST. No React imports, so it can run in a SharedWorker. + * - Out of scope here: hook wiring, authenticated subscriptions, multi-tab dedup. + * - Always keep an independent REST fetch as the source of truth for initial state. */ -class RealtimeClient { - private socket: WebSocket | null = null; - private handlers = new Map>(); +export type ConnectionState = "idle" | "connecting" | "open" | "reconnecting" | "closed"; + +export interface SocketLike { + send(data: string): void; + close(code?: number): void; + onopen: ((ev: unknown) => void) | null; + onmessage: ((ev: { data: unknown }) => void) | null; + onclose: ((ev: { code?: number }) => void) | null; + onerror: ((ev: unknown) => void) | null; +} + +export interface RealtimeClientOptions { + url?: string; + socketFactory?: (url: string) => SocketLike; + baseBackoffMs?: number; + maxBackoffMs?: number; + heartbeatIntervalMs?: number; + staleAfterMs?: number; + /** Returns [0,1); injectable for deterministic tests. */ + random?: () => number; + flags?: Record; + logger?: Pick; +} + +export interface SubscribeOptions { + /** Resume after this `seq` on first subscribe. */ + cursor?: number; + /** Called when the topic's flag turns off while subscribed; fall back to REST here. */ + onDisabled?: () => void; +} + +export interface SubscriptionHandle { + readonly id: number; + readonly topic: RealtimeTopic; +} + +type Handler = (message: TopicMessageMap[T]) => void; +interface Sub { + id: number; + topic: RealtimeTopic; + handler: Handler; + opts: SubscribeOptions; +} + +/** Derives the WebSocket URL from the API base the same way the REST client does. */ +export function deriveRealtimeUrl(apiBase: string = API_BASE_URL): string { + return apiBase.replace(/^http/, "ws").replace(/\/api\/v1\/?$/, "/ws"); +} + +export class RealtimeClient { + private opts: Required> & + Pick; + private flags: Record; + private state: ConnectionState = "idle"; + private stateListeners = new Set<(s: ConnectionState) => void>(); + private subs = new Map(); + private nextId = 1; + private cursors = new Map(); + private wire = new Set(); // topics subscribed on the current socket + private socket: SocketLike | null = null; private attempt = 0; - private closedByUser = false; private reconnectTimer: ReturnType | null = null; + private heartbeatTimer: ReturnType | null = null; + private lastSeenAt = 0; + private offline = false; + private envBound = false; + + constructor(options: RealtimeClientOptions = {}) { + this.flags = { ...DEFAULT_FLAGS, ...sanitizeFlags(options.flags) }; + this.opts = { + url: options.url, + socketFactory: options.socketFactory, + baseBackoffMs: options.baseBackoffMs ?? 1_000, + maxBackoffMs: options.maxBackoffMs ?? 30_000, + heartbeatIntervalMs: options.heartbeatIntervalMs ?? 15_000, + staleAfterMs: options.staleAfterMs ?? 45_000, + random: options.random ?? Math.random, + logger: options.logger ?? console, + }; + } + + getState(): ConnectionState { + return this.state; + } + + /** Observable state stream: emits the current state immediately, then every change. */ + observeState(listener: (s: ConnectionState) => void): () => void { + this.stateListeners.add(listener); + listener(this.state); + return () => this.stateListeners.delete(listener); + } + subscribe( + topic: T, + handler: Handler, + opts: SubscribeOptions = {} + ): SubscriptionHandle { + const id = this.nextId++; + this.subs.set(id, { id, topic, handler: handler as Handler, opts }); + if (opts.cursor !== undefined && !this.cursors.has(topic)) this.cursors.set(topic, opts.cursor); + this.sync(); + return { id, topic }; + } + + unsubscribe(handle: SubscriptionHandle): void { + if (!this.subs.delete(handle.id)) return; + this.sync(); + } + + /** Re-evaluates gating. Topics that turn off are suspended (and notified); ones that turn on resume. */ + setFlags(flags: unknown): void { + const before = new Map( + Array.from(this.subs.values()).map((s) => [s.id, this.enabled(s.topic)]) + ); + this.flags = { ...DEFAULT_FLAGS, ...sanitizeFlags(flags) }; + for (const s of Array.from(this.subs.values())) { + if (before.get(s.id) && !this.enabled(s.topic)) { + try { + s.opts.onDisabled?.(); + } catch { + this.opts.logger.warn("[realtime] onDisabled handler threw"); + } + } + } + this.sync(); + } + + /** Permanently stops the client (terminal until `connect()`); drops the socket and all timers. */ + close(): void { + this.teardownSocket(); + this.clearReconnect(); + this.unbindEnv(); + this.setState("closed"); + } + + /** Re-arms a closed client. */ connect(): void { - if (typeof WebSocket === "undefined" || this.socket) return; - this.closedByUser = false; - this.open(); + if (this.state === "closed") this.setState("idle"); + this.sync(); + } + + private enabled(topic: RealtimeTopic): boolean { + return isRealtimeTopicEnabled(this.flags, topic); + } + + private activeTopics(): Set { + const topics = new Set(); + this.subs.forEach((s) => this.enabled(s.topic) && topics.add(s.topic)); + return topics; + } + + /** Reconciles the socket and wire subscriptions with the current subscriptions + flags. */ + private sync(): void { + if (this.state === "closed") return; + const want = this.activeTopics(); + if (want.size === 0) { + this.teardownSocket(); + this.clearReconnect(); + this.unbindEnv(); + this.cursors.clear(); + this.setState("idle"); + return; + } + if (this.state === "idle") { + this.bindEnv(); + this.open(); + return; + } + if (this.state === "open") this.syncWire(want); + } + + private syncWire(want: Set): void { + this.wire.forEach((t) => { + if (!want.has(t)) { + this.send({ op: "unsubscribe", topic: t }); + this.wire.delete(t); + this.cursors.delete(t); + } + }); + want.forEach((t) => { + if (this.wire.has(t)) return; + const cursor = this.cursors.get(t); + this.send( + cursor === undefined ? { op: "subscribe", topic: t } : { op: "subscribe", topic: t, cursor } + ); + this.wire.add(t); + }); } private open(): void { - const wsUrl = API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws"); + if (this.socket || this.offline) { + if (this.offline) this.setState("reconnecting"); + return; + } + const url = this.opts.url ?? deriveRealtimeUrl(); + this.setState(this.attempt === 0 ? "connecting" : "reconnecting"); + let sock: SocketLike; try { - this.socket = new WebSocket(wsUrl); + sock = this.opts.socketFactory + ? this.opts.socketFactory(url) + : (new WebSocket(url) as unknown as SocketLike); } catch { this.scheduleReconnect(); return; } - - this.socket.addEventListener("open", () => { + this.socket = sock; + sock.onopen = () => { + if (this.socket !== sock) return; this.attempt = 0; - }); - this.socket.addEventListener("message", (event) => { + this.lastSeenAt = Date.now(); + this.setState("open"); + this.startHeartbeat(); + this.syncWire(this.activeTopics()); + }; + sock.onmessage = (ev) => { + if (this.socket !== sock) return; + this.lastSeenAt = Date.now(); + this.onFrame(ev.data); + }; + sock.onerror = () => { + /* close follows; handled there */ + }; + sock.onclose = (ev) => { + if (this.socket !== sock) return; // already replaced/torn down: guards double-close + this.socket = null; + this.stopHeartbeat(); + this.wire.clear(); + if (ev?.code === CLOSE_AUTH_FAILED) { + this.opts.logger.warn("[realtime] auth rejected; not reconnecting"); + this.setState("closed"); + return; + } + this.scheduleReconnect(); + }; + } + + private onFrame(raw: unknown): void { + const res = parseFrame(raw); + if (res.kind === "pong") return; + if (res.kind === "ignored") { + this.opts.logger.warn(`[realtime] ignored frame: ${res.reason}`); + return; + } + const { message } = res; + const prev = this.cursors.get(message.topic); + if (prev === undefined || message.seq > prev) this.cursors.set(message.topic, message.seq); + this.subs.forEach((s) => { + if (s.topic !== message.topic || !this.enabled(s.topic)) return; try { - const { topic, payload } = JSON.parse(event.data as string) as { topic: string; payload: unknown }; - this.handlers.get(topic)?.forEach((handler) => handler(payload)); + s.handler(message); } catch { - // Malformed message — ignore, don't crash the connection. + this.opts.logger.warn(`[realtime] handler for ${s.topic} threw`); } }); - this.socket.addEventListener("close", () => { - this.socket = null; - if (!this.closedByUser) this.scheduleReconnect(); - }); - this.socket.addEventListener("error", () => { - this.socket?.close(); - }); + } + + private send(op: ClientOp): void { + try { + this.socket?.send(JSON.stringify(op)); + } catch { + /* socket closing; reconnect path will resubscribe */ + } } private scheduleReconnect(): void { - if (this.reconnectTimer) return; - const delay = Math.min(BASE_BACKOFF_MS * 2 ** this.attempt, MAX_BACKOFF_MS); + if (this.reconnectTimer || this.state === "closed" || this.activeTopics().size === 0) return; + this.setState("reconnecting"); + if (this.offline) return; + const ceiling = Math.min(this.opts.maxBackoffMs, this.opts.baseBackoffMs * 2 ** this.attempt); + const delay = Math.min( + this.opts.maxBackoffMs, + ceiling / 2 + (this.opts.random() * ceiling) / 2 + ); this.attempt += 1; this.reconnectTimer = setTimeout(() => { this.reconnectTimer = null; - if (!this.closedByUser) this.open(); + this.open(); }, delay); } - subscribe(topic: string, handler: Handler): () => void { - if (!this.handlers.has(topic)) this.handlers.set(topic, new Set()); - this.handlers.get(topic)!.add(handler); - this.connect(); + private reconnectNow(): void { + if (this.state !== "reconnecting" && this.state !== "open") return; + this.clearReconnect(); + this.attempt = 0; + this.teardownSocket(); + this.open(); + } + + private startHeartbeat(): void { + this.stopHeartbeat(); + this.heartbeatTimer = setInterval(() => this.heartbeat(), this.opts.heartbeatIntervalMs); + } - return () => { - this.handlers.get(topic)?.delete(handler); - }; + private heartbeat(): void { + if (Date.now() - this.lastSeenAt > this.opts.staleAfterMs) { + this.opts.logger.warn("[realtime] connection stale; reconnecting"); + const sock = this.socket; + this.socket = null; + this.stopHeartbeat(); + this.wire.clear(); + this.detach(sock); + this.scheduleReconnect(); + return; + } + this.send({ op: "ping" }); + } + + private stopHeartbeat(): void { + if (this.heartbeatTimer) clearInterval(this.heartbeatTimer); + this.heartbeatTimer = null; } - disconnect(): void { - this.closedByUser = true; + private clearReconnect(): void { if (this.reconnectTimer) clearTimeout(this.reconnectTimer); - this.socket?.close(); + this.reconnectTimer = null; + } + + private detach(sock: SocketLike | null): void { + if (!sock) return; + sock.onopen = sock.onmessage = sock.onclose = sock.onerror = null; + try { + sock.close(1000); + } catch { + /* already closed */ + } + } + + private teardownSocket(): void { + const sock = this.socket; this.socket = null; + this.stopHeartbeat(); + this.wire.clear(); + this.detach(sock); + } + + private setState(next: ConnectionState): void { + if (this.state === next) return; + this.state = next; + this.stateListeners.forEach((l) => l(next)); + } + + private onOnline = () => { + this.offline = false; + if (this.state === "reconnecting") this.reconnectNow(); + }; + private onOffline = () => { + this.offline = true; + this.clearReconnect(); + if (this.state === "open" || this.state === "connecting") { + this.teardownSocket(); + this.setState("reconnecting"); + } + }; + private onVisibility = () => { + if (typeof document === "undefined" || document.visibilityState !== "visible") return; + if (this.state === "reconnecting" && !this.offline) this.reconnectNow(); + else if (this.state === "open") this.heartbeat(); + }; + + private bindEnv(): void { + if (this.envBound || typeof window === "undefined") return; + this.envBound = true; + this.offline = typeof navigator !== "undefined" && navigator.onLine === false; + window.addEventListener("online", this.onOnline); + window.addEventListener("offline", this.onOffline); + document.addEventListener("visibilitychange", this.onVisibility); + } + + private unbindEnv(): void { + if (!this.envBound) return; + this.envBound = false; + window.removeEventListener("online", this.onOnline); + window.removeEventListener("offline", this.onOffline); + document.removeEventListener("visibilitychange", this.onVisibility); } } +/** Shared instance. Flags default off, so nothing connects until `setFlags` enables a topic. */ export const realtimeClient = new RealtimeClient(); diff --git a/src/lib/realtime/flags.ts b/src/lib/realtime/flags.ts new file mode 100644 index 0000000..bfab998 --- /dev/null +++ b/src/lib/realtime/flags.ts @@ -0,0 +1,29 @@ +import type { RealtimeTopic } from "./protocol"; + +/** Global kill switch; dominates every per-topic flag. */ +export const REALTIME_GLOBAL_FLAG = "realtime-enabled"; + +export const REALTIME_TOPIC_FLAGS: Record = { + oracle: "realtime-oracle", + claims: "realtime-claims", + pool: "realtime-pool", + notifications: "realtime-notifications", +}; + +/** Keeps only boolean entries; a missing or malformed document yields no flags (everything off). */ +export function sanitizeFlags(raw: unknown): Record { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return {}; + const out: Record = {}; + for (const [key, value] of Object.entries(raw)) { + if (typeof value === "boolean") out[key] = value; + } + return out; +} + +/** The single gating helper every realtime consumer must use. A topic is on only if the kill switch AND its own flag are true. */ +export function isRealtimeTopicEnabled( + flags: Record | null | undefined, + topic: RealtimeTopic +): boolean { + return flags?.[REALTIME_GLOBAL_FLAG] === true && flags[REALTIME_TOPIC_FLAGS[topic]] === true; +} diff --git a/src/lib/realtime/merge.ts b/src/lib/realtime/merge.ts new file mode 100644 index 0000000..3817493 --- /dev/null +++ b/src/lib/realtime/merge.ts @@ -0,0 +1,72 @@ +/** + * Pure REST-snapshot + realtime-event reconciliation, independent of React and of + * the transport. Hooks map `RealtimeMessage`s to `MergeEvent`s (version = contract + * `ts`) and fold them with `reduce`. + * + * Rules: idempotent (an event whose version is not newer than what we hold for that + * entity is dropped, so redelivery changes nothing); out-of-order safe (older events + * are dropped); unknown ids are inserted; removals leave a tombstone so a stale + * upsert cannot resurrect a removed entity. Unchanged state keeps its reference. + */ +export type MergeEvent = + | { type: "upsert"; id: string; version: number; data: T } + | { type: "remove"; id: string; version: number }; + +export interface MergeState { + items: ReadonlyMap; + /** id → version at which the entity was removed. */ + tombstones: ReadonlyMap; +} + +export interface MergeConfig { + getId: (item: T) => string; + getVersion: (item: T) => number; +} + +export function createMerge({ getId, getVersion }: MergeConfig) { + const init = (snapshot: readonly T[] = []): MergeState => { + const items = new Map(); + for (const data of snapshot) { + const id = getId(data); + const version = getVersion(data); + const cur = items.get(id); + if (!cur || version > cur.version) items.set(id, { version, data }); + } + return { items, tombstones: new Map() }; + }; + + const reduce = (state: MergeState, event: MergeEvent): MergeState => { + const cur = state.items.get(event.id); + const dead = state.tombstones.get(event.id); + if (event.type === "upsert") { + if ((cur && event.version <= cur.version) || (dead !== undefined && event.version <= dead)) + return state; + const items = new Map(state.items); + items.set(event.id, { version: event.version, data: event.data }); + let tombstones = state.tombstones; + if (dead !== undefined) { + const t = new Map(tombstones); + t.delete(event.id); + tombstones = t; + } + return { items, tombstones }; + } + if ((cur && event.version < cur.version) || (dead !== undefined && event.version <= dead)) + return state; + const items = new Map(state.items); + items.delete(event.id); + const tombstones = new Map(state.tombstones); + tombstones.set(event.id, event.version); + return { items, tombstones }; + }; + + const applyAll = (state: MergeState, events: Iterable>): MergeState => { + let s = state; + for (const e of events) s = reduce(s, e); + return s; + }; + + const view = (state: MergeState): T[] => Array.from(state.items.values(), (e) => e.data); + + return { init, reduce, applyAll, view }; +} diff --git a/src/lib/realtime/protocol.ts b/src/lib/realtime/protocol.ts new file mode 100644 index 0000000..b2bc37e --- /dev/null +++ b/src/lib/realtime/protocol.ts @@ -0,0 +1,90 @@ +/** + * Realtime wire contract (see docs/realtime-protocol.md). Framework-agnostic: + * shared by the transport, the merge layer and the mock server's tests. + */ +export const PROTOCOL_VERSION = 1; + +export const REALTIME_TOPICS = ["oracle", "claims", "pool", "notifications"] as const; +export type RealtimeTopic = (typeof REALTIME_TOPICS)[number]; + +export interface OracleUpdate { + id: string; + price: string; + updatedAt: number; +} +export interface ClaimUpdate { + id: string; + policyId: string; + status: string; + amount: string; + updatedAt: number; +} +export interface PoolUpdate { + id: string; + tvl: string; + utilization: number; + updatedAt: number; +} +export interface NotificationEvent { + id: string; + kind: string; + message: string; + createdAt: number; +} + +/** Per-topic payload types. */ +export interface TopicPayloads { + oracle: OracleUpdate; + claims: ClaimUpdate; + pool: PoolUpdate; + notifications: NotificationEvent; +} + +/** Server → client frame for a topic. `seq` is per-topic and monotonic; `ts` is epoch ms. */ +export interface RealtimeMessage { + v: typeof PROTOCOL_VERSION; + topic: T; + seq: number; + ts: number; + payload: TopicPayloads[T]; +} + +/** Maps each topic to its handler's message type. */ +export type TopicMessageMap = { [T in RealtimeTopic]: RealtimeMessage }; + +export type ClientOp = + | { op: "subscribe"; topic: RealtimeTopic; cursor?: number } + | { op: "unsubscribe"; topic: RealtimeTopic } + | { op: "ping" }; + +/** Close code the server uses when it rejects the connection's credentials. */ +export const CLOSE_AUTH_FAILED = 4401; + +export type ParseResult = + | { kind: "message"; message: RealtimeMessage } + | { kind: "pong" } + | { kind: "ignored"; reason: string }; + +/** Parses an inbound frame. Never throws; anything unrecognised is reported as ignored. */ +export function parseFrame(raw: unknown): ParseResult { + if (typeof raw !== "string") return { kind: "ignored", reason: "non-text frame" }; + let data: unknown; + try { + data = JSON.parse(raw); + } catch { + return { kind: "ignored", reason: "malformed json" }; + } + if (!data || typeof data !== "object") return { kind: "ignored", reason: "not an object" }; + const f = data as Record; + if (f.op === "pong") return { kind: "pong" }; + if (f.v !== PROTOCOL_VERSION) + return { kind: "ignored", reason: `unsupported version ${String(f.v)}` }; + if (typeof f.topic !== "string" || !(REALTIME_TOPICS as readonly string[]).includes(f.topic)) { + return { kind: "ignored", reason: `unknown topic ${String(f.topic).slice(0, 32)}` }; + } + if (!Number.isFinite(f.seq) || !Number.isFinite(f.ts)) + return { kind: "ignored", reason: "bad seq/ts" }; + if (!f.payload || typeof f.payload !== "object") + return { kind: "ignored", reason: "bad payload" }; + return { kind: "message", message: f as unknown as RealtimeMessage }; +}