diff --git a/README.md b/README.md index 4960ac8..000b9d7 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,14 @@ Runtime errors are reported to Sentry when `NEXT_PUBLIC_SENTRY_DSN` is set. See [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}'`. + ## Handsoff notes 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 bc77fd4..3338cf1 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/public/flags.json b/public/flags.json index e55c12a..6035025 100644 --- a/public/flags.json +++ b/public/flags.json @@ -1,6 +1,10 @@ { "flags": { + "realtime-enabled": false, "realtime-oracle": false, + "realtime-claims": false, + "realtime-pool": false, + "realtime-notifications": false, "horizon-stream": false, "realtime-quotes": false, "realtime-heartbeat": true, 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 3bdf37f..e69de29 100644 --- a/src/lib/flags/flagsConfig.ts +++ b/src/lib/flags/flagsConfig.ts @@ -1,21 +0,0 @@ -import { env } from "@/lib/env"; - -export interface FlagsDocument { - flags: Record; -} - -/** Safe default if the remote flags source is unreachable — new/risky features default off. */ -export const DEFAULT_FLAGS: Record = { - "realtime-oracle": false, - "horizon-stream": false, - "realtime-quotes": false, -}; - -export const FLAGS_ENDPOINT = "/flags.json"; -export const FLAGS_CACHE_TTL_MS = 5 * 60 * 1000; - -/** Reads a local override, e.g. NEXT_PUBLIC_FLAGS_OVERRIDE='{"realtime-oracle":true}'. */ -export function readLocalOverride(): Record | null { - const raw = env.NEXT_PUBLIC_FLAGS_OVERRIDE; - return raw ? (JSON.parse(raw) as Record) : null; -} 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 cac458e..e69de29 100644 --- a/src/lib/realtime/client.ts +++ b/src/lib/realtime/client.ts @@ -1,907 +0,0 @@ -import { API_BASE_URL } from "@/lib/api/client"; -import { env } from "@/lib/env"; -import { getRealtimeCapability } from "@/lib/realtime/capabilities"; - -import { isTopicEnabled, onTopicFlagChange } from "./flags"; -import { startWakeDetector } from "./lifecycle"; -import { recordBytes, utf8Length } from "./metrics"; -import type { TopicMap } from "./topics"; - -type Handler = (payload: unknown) => void; - -/** Lifecycle of the realtime connection, surfaced so the UI never silently shows stale data. */ -export type RealtimeStatus = "idle" | "connecting" | "live" | "resyncing" | "offline"; - -/** A non-WebSocket source (e.g. Horizon SSE) that shares the client's lifecycle and health. */ -export interface StreamTransport { - start(): void; - stop(): void; -} - -const MAX_BACKOFF_MS = 30_000; -const BASE_BACKOFF_MS = 1_000; -const SEEN_IDS_CAP = 500; -const CURSOR_STORAGE_PREFIX = "realtime:cursor:"; -const MAX_REPORTED_RECONNECT_LOOPS = 5; - -/** - * Machine-readable connection status the UI can branch on. - * - `idle`: never connected / explicitly disconnected - * - `connecting`: initial connection in flight - * - `open`: connected and healthy - * - `reconnecting`: dropped, waiting to retry - * - `closed`: closed by the user - */ -export type RealtimeStatus = "idle" | "connecting" | "open" | "reconnecting" | "closed"; - -export interface RealtimeHealth { - status: RealtimeStatus; - /** Transport currently in use, e.g. "websocket". */ - transport: string; - /** Number of reconnect attempts since the last successful open. */ - attempt: number; - /** Epoch ms of the last message received, or null if none yet. */ - lastMessageAt: number | null; - /** Age in ms of the last message, or null if none yet. */ - lastMessageAgeMs: number | null; - /** Last error surfaced by the transport, redacted for reporting. */ - lastError: string | null; - /** Cumulative count of transport errors observed. */ - errorCount: number; -} - -type HealthListener = (health: RealtimeHealth) => void; - -/** - * A single observed realtime message, captured for the dev inspector. - * `direction` is `in` for messages received from the server and `out` for - * subscribe frames sent by the client. - */ -export interface RealtimeInspectorEvent { - id: number; - topic: string; - seq: number | null; - direction: "in" | "out"; - timestamp: number; - payload: unknown; -} - -type InspectorListener = (events: RealtimeInspectorEvent[]) => void; - -/** - * Strips tokens/URLs from error strings before they leave the client so - * health signals can be safely forwarded to error monitoring. - */ -function redactError(message: string): string { - return message - .replace(/wss?:\/\/[^\s"']+/gi, "[redacted-url]") - .replace(/https?:\/\/[^\s"']+/gi, "[redacted-url]") - .replace(/(token|access_token|auth|key|secret)=[^&\s"']+/gi, "$1=[redacted]"); -} - -/** - * Heartbeat tuning. Defaults are intentionally cheap on mobile/battery: - * a ping every 25s with a 10s pong deadline. The dev mock server can - * disable heartbeats entirely (see `heartbeatEnabled`) when testing - * purely offline flows. - */ -export const HEARTBEAT_INTERVAL_MS = 25_000; -export const HEARTBEAT_PONG_DEADLINE_MS = 10_000; - -/** Snapshot of heartbeat state for health/observability consumers. */ -export interface HeartbeatState { - lastPongAt: number | null; - consecutiveMisses: number; - healthy: boolean; -} - -/** - * Realtime WebSocket client. - * - * Handles connection lifecycle, subscription replay, and auth token expiry - * for wallet-scoped subscriptions. Token acquisition is isolated behind an - * injected AuthTokenProvider so the transport stays wallet-agnostic. - * - * 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, seq?: number }`, one topic per - * push-based update type (e.g. "oracle_update"). On every (re)connect the - * client sends one `{ type: "subscribe", topics, snapshot: true }` batch for - * all active topics. Callers must treat this connection as best-effort — - * always keep an independent REST fetch as the source of truth for initial - * state. - * - * An application-level heartbeat sends `{ type: "ping" }` frames on an - * interval and expects `{ type: "pong" }` back within a deadline. Missed - * deadlines mark the connection unhealthy and close it so the existing - * reconnect logic takes over, detecting half-open sockets. - * - * Snapshot-and-resume: the last-applied sequence per topic is persisted and - * sent as a `since` cursor on (re)subscribe so the server can replay missed - * events. On reconnect the client first refetches the REST snapshot for the - * topic, then replays any events newer than the stored cursor through the - * idempotent merge layer. If replay is unavailable or the cursor is too old, - * it falls back to a fresh snapshot with no streamed delta. - * - * Health: the client maintains an observable health snapshot emitted to - * subscribers on state transitions (not per-message) so React can subscribe - * rather than poll. High-frequency message timestamps are coalesced. - * - * Inspector: a dev-only tap point exposes a bounded ring buffer of recent - * messages (raw in/out frames) to the dev inspector panel. It is a passive - * observer — subscribing to it never triggers a (re)subscribe on the socket. - * - * Transport is negotiated at startup via `getRealtimeCapability()`. Until - * capability is confirmed the client stays on REST; it only opens a socket - * when the backend advertises WebSocket support, avoiding reconnect loops - * against backends that never answer. - */ - -import type { - AuthTokenProvider, - AuthenticatedSubscription, - RealtimeAuthOptions, - RealtimeCloseCode, - RealtimeConnectionState, - RealtimeDegradedState, -} from './types'; - -const AUTH_CLOSE_CODES: ReadonlySet = new Set([4001, 4003, 4401, 4403]); -const DEFAULT_MAX_REFRESH_ATTEMPTS = 2; - -export interface RealtimeClientOptions extends RealtimeAuthOptions { - url: string; - onStateChange?: (state: RealtimeConnectionState) => void; - onDegraded?: (state: RealtimeDegradedState) => void; -} - */ - -import type { - AuthTokenProvider, - AuthenticatedSubscription, - RealtimeAuthOptions, - RealtimeCloseCode, - RealtimeConnectionState, - RealtimeDegradedState, -} from './types'; - -const AUTH_CLOSE_CODES: ReadonlySet = new Set([4001, 4003, 4401, 4403]); -const DEFAULT_MAX_REFRESH_ATTEMPTS = 2; - -export interface RealtimeClientOptions extends RealtimeAuthOptions { - url: string; - onStateChange?: (state: RealtimeConnectionState) => void; - onDegraded?: (state: RealtimeDegradedState) => void; -} - -export class RealtimeClient { - private readonly url: string; - private readonly auth?: AuthTokenProvider; - private readonly maxRefreshAttempts: number; - private readonly onStateChange?: (state: RealtimeConnectionState) => void; - private readonly onDegraded?: (state: RealtimeDegradedState) => void; - - private socket: WebSocket | null = null; - private state: RealtimeConnectionState = 'idle'; - private refreshAttempts = 0; - private refreshInFlight: Promise | null = null; - private readonly authenticatedSubscriptions = new Map(); - private handlers = new Map>(); - /** Handlers for flag-disabled topics: held in memory only, never on the socket. */ - private parked = new Map>(); - private transports = new Set(); - private statusListeners = new Set<(s: RealtimeStatus) => void>(); - private status: RealtimeStatus = "idle"; - private attempt = 0; - private closedByUser = false; - private reconnectTimer: ReturnType | null = null; - private stopWake: (() => void) | null = null; - /** Bounded set of recent event ids so a post-resync snapshot replay is reconciled, not duplicated. */ - private seenIds = new Set(); - - getStatus(): RealtimeStatus { - return this.status; - } - - onStatus(listener: (s: RealtimeStatus) => void): () => void { - this.statusListeners.add(listener); - return () => this.statusListeners.delete(listener); - } - - private setStatus(next: RealtimeStatus): void { - if (this.status === next) return; - this.status = next; - this.statusListeners.forEach((l) => l(next)); - } - - private cursors = new Map(); - private negotiated = false; - - constructor(options: RealtimeClientOptions) { - this.url = options.url; - this.auth = options.auth; - this.maxRefreshAttempts = options.maxRefreshAttempts ?? DEFAULT_MAX_REFRESH_ATTEMPTS; - this.onStateChange = options.onStateChange; - this.onDegraded = options.onDegraded; - } - - private health: RealtimeHealth = { - status: "idle", - transport: "websocket", - attempt: 0, - lastMessageAt: null, - lastMessageAgeMs: null, - lastError: null, - errorCount: 0, - }; - private healthListeners = new Set(); - private messageCoalesceTimer: ReturnType | null = null; - private reportedReconnectLoops = 0; - - private inspectorEvents: RealtimeInspectorEvent[] = []; - private inspectorListeners = new Set(); - private inspectorSeq = 0; - private static readonly INSPECTOR_MAX_EVENTS = 100; - - private heartbeatTimer: ReturnType | null = null; - private pongDeadlineTimer: ReturnType | null = null; - private lastPongAt: number | null = null; - private consecutiveMisses = 0; - private healthy = true; - - constructor() { - onTopicFlagChange((topic, enabled) => { - if (enabled) { - const parked = this.parked.get(topic); - if (!parked?.size) return; - this.parked.delete(topic); - parked.forEach((handler) => this.attach(topic, handler)); - } else { - const live = this.handlers.get(topic); - if (!live) return; - this.handlers.delete(topic); - if (live.size) this.parked.set(topic, live); - if (this.handlers.size === 0) this.disconnect(); - } - }); - } - - connect(): void { - if (typeof WebSocket === "undefined" || this.socket) return; - this.closedByUser = false; - this.stopWake ??= startWakeDetector(() => this.resync()); - if (this.negotiated) { - this.open(); - return; - } - void this.negotiate(); - } - - private async negotiate(): Promise { - const capability = await getRealtimeCapability(); - this.negotiated = true; - if (this.closedByUser) return; - if (capability.transport === "websocket") { - this.open(); - } - // SSE and REST-only transports are handled elsewhere; this client only - // manages the WebSocket transport. - } - - getState(): RealtimeConnectionState { - return this.state; - } - - /** - * Registers a wallet-scoped subscription that will be replayed after a - * successful re-authentication. - */ - registerAuthenticatedSubscription(sub: AuthenticatedSubscription): void { - this.authenticatedSubscriptions.set(sub.topic, sub); - } - - unregisterAuthenticatedSubscription(topic: string): void { - this.authenticatedSubscriptions.delete(topic); - } - - async connect(): Promise { - this.setState('connecting'); - const token = await this.resolveToken(); - this.openSocket(token); - } - - disconnect(): void { - this.socket?.close(1000, 'client disconnect'); - this.socket = null; - this.setState('closed'); - } - - private async resolveToken(): Promise { - if (!this.auth) return null; - try { - return await this.auth.getAuthToken(); - } catch { - return null; - } - } - - private openSocket(token: string | null): void { - const url = token ? this.withToken(this.url, token) : this.url; - const socket = new WebSocket(url); - this.socket = socket; - - socket.onopen = () => { - this.refreshAttempts = 0; - this.setState('connected'); - this.replaySubscriptions(); - }; - - socket.onmessage = (event) => { - if (this.isAuthErrorFrame(event.data)) { - void this.handleAuthFailure(); - } - }; - - socket.onclose = (event) => { - if (this.isAuthCloseCode(event.code)) { - void this.handleAuthFailure(); - return; - } - this.setState('closed'); - }; - - socket.onerror = () => { - // Errors surface via onclose; nothing to do here. - }; - } - - private isAuthCloseCode(code: RealtimeCloseCode): boolean { - return AUTH_CLOSE_CODES.has(code); - } - - private isAuthErrorFrame(data: unknown): boolean { - if (typeof data !== 'string') return false; - try { - const parsed = JSON.parse(data) as { type?: string; code?: number }; - if (parsed?.type === 'error' && typeof parsed.code === 'number') { - return AUTH_CLOSE_CODES.has(parsed.code); - } - return parsed?.type === 'auth_error'; - } catch { - return false; - } - } - - /** - * Handles an auth failure by refreshing the token and replaying - * wallet-scoped subscriptions. Guards against refresh loops and dedupes - * concurrent refreshes. - */ - private async handleAuthFailure(): Promise { - if (!this.auth) { - this.enterDegraded('auth-required', 'Authentication required for this subscription.'); - return; - } - - if (this.refreshAttempts >= this.maxRefreshAttempts) { - this.enterDegraded('auth-refresh-loop', 'Unable to re-authenticate. Showing cached data.'); - return; - } - - this.refreshAttempts += 1; - this.setState('reconnecting'); - - const token = await this.refreshTokenDeduped(); - if (!token) { - this.enterDegraded('auth-refresh-failed', 'Session expired. Falling back to REST.'); - return; - } - - this.socket?.close(1000, 'reauth'); - this.openSocket(token); - } - - /** - * Dedupes concurrent refresh requests so multiple failing subscriptions - * share a single refresh call. - */ - private refreshTokenDeduped(): Promise { - if (this.refreshInFlight) return this.refreshInFlight; - if (!this.auth) return Promise.resolve(null); - - this.refreshInFlight = this.auth - .refreshAuthToken() - .catch(() => null) - .finally(() => { - this.refreshInFlight = null; - }); - - return this.refreshInFlight; - } - - private replaySubscriptions(): void { - if (!this.socket || this.socket.readyState !== WebSocket.OPEN) return; - for (const sub of this.authenticatedSubscriptions.values()) { - this.socket.send( - JSON.stringify({ type: 'subscribe', topic: sub.topic, wallet: sub.wallet, params: sub.params }), - ); - } - } - - private enterDegraded( - reason: RealtimeDegradedState['reason'], - message: string, - ): void { - this.setState('degraded'); - this.onDegraded?.({ reason, fallbackToRest: true, message }); - } - - private withToken(url: string, token: string): string { - try { - const parsed = new URL(url); - parsed.searchParams.set('token', token); - return parsed.toString(); - } catch { - return url; - } - } - - private setState(state: RealtimeConnectionState): void { - if (this.state === state) return; - this.state = state; - this.onStateChange?.(state); - } - - private open(): void { - const wsUrl = env.NEXT_PUBLIC_REALTIME_URL ?? API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws"); - if (this.status !== "resyncing") this.setStatus("connecting"); - let socket: WebSocket; - try { - socket = new WebSocket(wsUrl); - } catch { - this.setStatus("offline"); - return; - } - this.socket = socket; - } - - private open(): void { - const wsUrl = env.NEXT_PUBLIC_REALTIME_URL ?? API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws"); - this.updateHealth({ status: this.attempt > 0 ? "reconnecting" : "connecting" }); - - try { - this.socket = new WebSocket(wsUrl); - } catch (err) { - this.recordError(err); - this.scheduleReconnect(); - return; - } - const socket = this.socket; - - socket.addEventListener("open", () => { - if (this.socket !== socket) return; - this.attempt = 0; - this.statusListeners.forEach((cb) => cb(true)); - this.reportedReconnectLoops = 0; - this.updateHealth({ status: "open", attempt: 0 }); - this.setStatus("live"); - - try { - this.socket = new WebSocket(wsUrl); - } catch (err) { - this.recordError(err); - this.scheduleReconnect(); - return; - } - const socket = this.socket; - - socket.addEventListener("open", () => { - if (this.socket !== socket) return; - this.attempt = 0; - this.statusListeners.forEach((cb) => cb(true)); - this.reportedReconnectLoops = 0; - this.updateHealth({ status: "open", attempt: 0 }); - this.setStatus("live"); - const topics = [...this.handlers.keys()]; - if (topics.length > 0) socket.send(JSON.stringify({ type: "subscribe", topics, snapshot: true })); - this.scheduleReconnect(); - return; - } - this.socket = socket; - - socket.addEventListener("open", () => { - if (this.socket !== socket) return; - this.attempt = 0; - this.reportedReconnectLoops = 0; - this.updateHealth({ status: "open", attempt: 0 }); - this.resubscribe(); - this.healthy = true; - this.consecutiveMisses = 0; - this.startHeartbeat(); - this.setStatus("live"); - const topics = [...this.handlers.keys()]; - if (topics.length > 0) socket.send(JSON.stringify({ type: "subscribe", topics, snapshot: true })); - }); - socket.addEventListener("message", (event) => { - if (this.socket !== socket) return; - try { - const raw = event.data as string; - const data = JSON.parse(raw) as { - topic?: string; - payload?: unknown; - type?: string; - seq?: number; - }; - if (data.type === "pong") { - this.handlePong(); - return; - } - if (data.type === "ping") { - this.send({ type: "pong" }); - return; - } - const { topic, payload, seq } = data; - if (typeof seq === "number") this.setCursor(topic, seq); - this.recordInspectorEvent({ - topic, - seq: typeof seq === "number" ? seq : null, - direction: "in", - payload, - }); - recordBytes(topic, "in", utf8Length(raw)); - if (this.isDuplicate(payload)) return; - this.handlers.get(topic)?.forEach((handler) => handler(payload)); - this.markMessageReceived(); - } catch { - // Malformed message — ignore, don't crash the connection. - } - }); - this.socket.addEventListener("close", () => { - this.stopHeartbeat(); - this.socket = null; - this.statusListeners.forEach((cb) => cb(false)); - if (!this.closedByUser) { - this.updateHealth({ status: "reconnecting" }); - this.scheduleReconnect(); - } - }); - this.socket.addEventListener("error", (event) => { - this.recordError(event); - this.socket?.close(); - }); - } - - private send(message: unknown): void { - if (this.socket?.readyState !== WebSocket.OPEN) return; - try { - this.socket.send(JSON.stringify(message)); - } catch { - // Send failures surface via the close/error handlers. - } - } - - private startHeartbeat(): void { - this.stopHeartbeat(); - this.heartbeatTimer = setInterval(() => { - this.send({ type: "ping" }); - if (this.pongDeadlineTimer) clearTimeout(this.pongDeadlineTimer); - this.pongDeadlineTimer = setTimeout(() => { - this.pongDeadlineTimer = null; - this.handleMissedPong(); - }, HEARTBEAT_PONG_DEADLINE_MS); - }, HEARTBEAT_INTERVAL_MS); - } - - private stopHeartbeat(): void { - if (this.heartbeatTimer) clearInterval(this.heartbeatTimer); - if (this.pongDeadlineTimer) clearTimeout(this.pongDeadlineTimer); - this.heartbeatTimer = null; - this.pongDeadlineTimer = null; - } - - private handlePong(): void { - this.lastPongAt = Date.now(); - this.consecutiveMisses = 0; - this.healthy = true; - if (this.pongDeadlineTimer) clearTimeout(this.pongDeadlineTimer); - this.pongDeadlineTimer = null; - } - - private handleMissedPong(): void { - this.consecutiveMisses += 1; - this.healthy = false; - // Close the half-open socket; the close handler triggers reconnect. - this.socket?.close(); - } - - getHeartbeatState(): HeartbeatState { - return { - lastPongAt: this.lastPongAt, - consecutiveMisses: this.consecutiveMisses, - healthy: this.healthy, - }; - } - - private isDuplicate(payload: unknown): boolean { - const id = (payload as { id?: unknown } | null)?.id; - if (typeof id !== "string") return false; - if (this.seenIds.has(id)) return true; - this.seenIds.add(id); - if (this.seenIds.size > SEEN_IDS_CAP) this.seenIds.delete(this.seenIds.values().next().value as string); - return false; - } - - private scheduleReconnect(): void { - if (this.reconnectTimer) return; - const delay = Math.min(BASE_BACKOFF_MS * 2 ** this.attempt, MAX_BACKOFF_MS); - this.attempt += 1; - this.updateHealth({ status: "reconnecting", attempt: this.attempt }); - this.reportReconnectLoop(); - this.reconnectTimer = setTimeout(() => { - this.reconnectTimer = null; - if (!this.closedByUser) this.open(); - }, delay); - } - - /** - * Disabled topics never open a subscription: the handler is parked (so a - * mid-session enable can activate it) and callers keep using their REST path. - */ - subscribe(topic: string, handler: Handler): () => void { - if (isTopicEnabled(topic)) this.attach(topic, handler); - else this.park(topic, handler); - - return () => { - this.handlers.get(topic)?.delete(handler); - this.parked.get(topic)?.delete(handler); - }; - } - - private park(topic: string, handler: Handler): void { - if (!this.parked.has(topic)) this.parked.set(topic, new Set()); - this.parked.get(topic)!.add(handler); - } - - private attach(topic: string, handler: Handler): void { - if (!this.handlers.has(topic)) this.handlers.set(topic, new Set()); - this.handlers.get(topic)!.add(handler); - this.connect(); - if (this.socket?.readyState === WebSocket.OPEN) { - this.sendSubscribe(topic, this.cursors.get(topic)); - } - } - - /** - * Drops the (possibly dead) connection and reconnects once, immediately, - * resubscribing every active topic as a single batch with a snapshot - * request. Idempotent: it supersedes any pending backoff timer instead of - * adding a second reconnect. - */ - resync(): void { - if (this.closedByUser || this.handlers.size === 0) return; - this.setStatus("resyncing"); - if (this.reconnectTimer) { - clearTimeout(this.reconnectTimer); - this.reconnectTimer = null; - } - const stale = this.socket; - this.socket = null; // detaches stale listeners (they check identity) - stale?.close(); - this.attempt = 0; - this.transports.forEach((t) => { - t.stop(); - t.start(); - }); - this.open(); - } - - registerTransport(transport: StreamTransport): () => void { - this.transports.add(transport); - return () => this.transports.delete(transport); - } - - private resubscribe(): void { - if (!this.socket || this.socket.readyState !== WebSocket.OPEN) return; - for (const topic of this.handlers.keys()) { - const since = this.cursors.get(topic); - this.sendSubscribe(topic, since); - } - } - - private sendSubscribe(topic: string, since: number | undefined): void { - if (!this.socket || this.socket.readyState !== WebSocket.OPEN) return; - this.socket.send(JSON.stringify({ type: "subscribe", topic, since })); - this.recordInspectorEvent({ - topic, - seq: typeof since === "number" ? since : null, - direction: "out", - payload: { type: "subscribe", topic, since }, - }); - } - - private setCursor(topic: string, seq: number): void { - const current = this.cursors.get(topic); - if (current !== undefined && seq <= current) return; - this.cursors.set(topic, seq); - try { - if (typeof sessionStorage !== "undefined") { - sessionStorage.setItem(`${CURSOR_STORAGE_PREFIX}${topic}`, String(seq)); - } - } catch { - // Storage unavailable — in-memory cursor still works for this session. - } - } - - private loadCursor(topic: string): number | undefined { - const inMemory = this.cursors.get(topic); - if (inMemory !== undefined) return inMemory; - try { - if (typeof sessionStorage !== "undefined") { - const raw = sessionStorage.getItem(`${CURSOR_STORAGE_PREFIX}${topic}`); - if (raw !== null) { - const parsed = Number(raw); - if (Number.isFinite(parsed)) { - this.cursors.set(topic, parsed); - return parsed; - } - } - } - } catch { - // Storage unavailable — fall through to no cursor. - } - return undefined; - } - - /** - * Returns the last-applied cursor for a topic, if any. Consumers use this - * to request a REST snapshot and replay events newer than the cursor. - */ - getCursor(topic: string): number | undefined { - return this.loadCursor(topic); - } - - /** - * Clears the stored cursor for a topic. Call this when falling back to a - * fresh snapshot because replay is unavailable or the cursor is too old. - */ - resetCursor(topic: string): void { - this.cursors.delete(topic); - try { - if (typeof sessionStorage !== "undefined") { - sessionStorage.removeItem(`${CURSOR_STORAGE_PREFIX}${topic}`); - } - } catch { - // Storage unavailable — nothing to clear. - } - } - - /** - * Dev-only tap point: records a message into the bounded inspector ring - * buffer and notifies inspector listeners with a stable snapshot. This is - * a passive observer and never touches the socket or subscriptions. - */ - private recordInspectorEvent( - event: Omit, - ): void { - if (this.inspectorListeners.size === 0) return; - this.inspectorSeq += 1; - const entry: RealtimeInspectorEvent = { - ...event, - id: this.inspectorSeq, - timestamp: Date.now(), - }; - const next = [...this.inspectorEvents, entry]; - this.inspectorEvents = - next.length > RealtimeClient.INSPECTOR_MAX_EVENTS - ? next.slice(next.length - RealtimeClient.INSPECTOR_MAX_EVENTS) - : next; - const snapshot = this.inspectorEvents; - this.inspectorListeners.forEach((listener) => listener(snapshot)); - } - - /** - * Subscribes to the dev inspector event stream. Returns the current - * snapshot immediately and an unsubscribe function. Does not open or - * resubscribe the socket. - */ - subscribeInspector(listener: InspectorListener): () => void { - this.inspectorListeners.add(listener); - listener(this.inspectorEvents); - return () => { - this.inspectorListeners.delete(listener); - }; - } - - subscribe(topic: T, handler: (payload: TopicMap[T]) => void): () => void; - subscribe(topic: string, handler: Handler): () => void; - subscribe(topic: string, handler: Handler): () => void { - if (isTopicEnabled(topic)) this.attach(topic, handler); - else this.park(topic, handler); - - return () => { - this.handlers.get(topic)?.delete(handler); - this.parked.get(topic)?.delete(handler); - }; - } - - private park(topic: string, handler: Handler): void { - if (!this.parked.has(topic)) this.parked.set(topic, new Set()); - this.parked.get(topic)!.add(handler); - } - - private attach(topic: string, handler: Handler): void { - if (!this.handlers.has(topic)) this.handlers.set(topic, new Set()); - this.handlers.get(topic)!.add(handler); - this.connect(); - if (this.socket?.readyState === WebSocket.OPEN) { - this.sendSubscribe(topic, this.cursors.get(topic)); - } - } - - private conn - - return () => { - this.handlers.get(topic)?.delete(handler); - this.parked.get(topic)?.delete(handler); - }; - } - - private park(topic: string, handler: Handler): void { - if (!this.parked.has(topic)) this.parked.set(topic, new Set()); - this.parked.get(topic)!.add(handler); - } - - private attach(topic: string, handler: Handler): void { - if (!this.handlers.has(topic)) this.handlers.set(topic, new Set()); - this.handlers.get(topic)!.add(handler); - this.loadCursor(topic); - this.connect(); - if (isNew && this.socket?.readyState === 1) { - this.socket.send(JSON.stringify({ type: "subscribe", topics: [topic], snapshot: true })); - } - - if (this.socket && this.socket.readyState === WebSocket.OPEN) { - this.sendSubscribe(topic, this.cursors.get(topic)); - } - - return () => { - this.handlers.get(topic)?.delete(handler); - this.parked.get(topic)?.delete(handler); - }; - } - - /** Notifies on socket open/close so consumers can degrade to REST immediately instead of waiting for heartbeat staleness. */ - onConnectionChange(cb: (connected: boolean) => void): () => void { - this.statusListeners.add(cb); - return () => { - this.statusListeners.delete(cb); - }; - } - - disconnect(): void { - this.closedByUser = true; - this.stopHeartbeat(); - if (this.reconnectTimer) clearTimeout(this.reconnectTimer); - this.reconnectTimer = null; - this.stopWake?.(); - this.stopWake = null; - this.socket?.close(); - this.socket = null; - this.setStatus("idle"); - } -} - } - - disconnect(): void { - this.closedByUser = true; - this.stopHeartbeat(); - if (this.reconnectTimer) clearTimeout(this.reconnectTimer); - this.reconnectTimer = null; - this.stopWake?.(); - this.stopWake = null; - this.socket?.close(); - this.socket = null; - this.setStatus("idle"); - } -} diff --git a/src/lib/realtime/flags.ts b/src/lib/realtime/flags.ts index 74afd86..e69de29 100644 --- a/src/lib/realtime/flags.ts +++ b/src/lib/realtime/flags.ts @@ -1,151 +0,0 @@ -import { FLAGS_ENDPOINT } from "@/lib/flags/flagsConfig"; - -/** - * Per-topic realtime flags. - * - * Schema (public/flags.json), keyed by topic identifier: - * { "realtime": { "topics": { "": { "enabled": boolean, "kill"?: boolean } } } } - * - * Precedence, highest first: - * 1. kill (runtime `killTopic()` or `kill: true` in flags.json) — always disabled - * 2. in-memory override (`setTopicOverride`, used by tests / developer flow) - * 3. env override NEXT_PUBLIC_REALTIME_TOPICS_OVERRIDE='{"oracle_update":true}' - * 4. flags.json `enabled` - * 5. default: disabled (unknown, missing or malformed entries fail closed) - */ -export type TopicFlagListener = (topic: string, enabled: boolean) => void; - -interface TopicFlag { - enabled: boolean; - kill: boolean; -} - -let fileFlags = new Map(); -let envOverrides = new Map(); -const overrides = new Map(); -const killed = new Set(); -const listeners = new Set(); -let initialised = false; - -function isRecord(value: unknown): value is Record { - return typeof value === "object" && value !== null && !Array.isArray(value); -} - -function parseFile(doc: unknown): Map { - const out = new Map(); - const topics = isRecord(doc) && isRecord(doc.realtime) ? doc.realtime.topics : undefined; - if (!isRecord(topics)) { - if (doc !== undefined) console.warn("[realtime-flags] missing or malformed realtime.topics; all topics disabled"); - return out; - } - for (const [topic, entry] of Object.entries(topics)) { - if (!isRecord(entry) || typeof entry.enabled !== "boolean" || (entry.kill !== undefined && typeof entry.kill !== "boolean")) { - console.warn(`[realtime-flags] malformed entry for topic "${topic}"; treating as disabled`); - out.set(topic, { enabled: false, kill: false }); - continue; - } - out.set(topic, { enabled: entry.enabled, kill: entry.kill === true }); - } - return out; -} - -function parseEnv(): Map { - const out = new Map(); - const raw = process.env.NEXT_PUBLIC_REALTIME_TOPICS_OVERRIDE; - if (!raw) return out; - try { - const parsed: unknown = JSON.parse(raw); - if (!isRecord(parsed)) throw new Error("not an object"); - for (const [topic, value] of Object.entries(parsed)) { - if (typeof value === "boolean") out.set(topic, value); - else console.warn(`[realtime-flags] ignoring non-boolean env override for "${topic}"`); - } - } catch { - console.warn("[realtime-flags] malformed NEXT_PUBLIC_REALTIME_TOPICS_OVERRIDE ignored"); - } - return out; -} - -/** Resolves a topic's flag without side effects. Read-through of the cached state. */ -export function isTopicEnabled(topic: string): boolean { - if (killed.has(topic)) return false; - const file = fileFlags.get(topic); - if (file?.kill) return false; - const override = overrides.get(topic); - if (override !== undefined) return override; - const env = envOverrides.get(topic); - if (env !== undefined) return env; - return file?.enabled ?? false; -} - -function snapshot(): Map { - const topics = new Set([...fileFlags.keys(), ...envOverrides.keys(), ...overrides.keys(), ...killed]); - return new Map([...topics].map((t) => [t, isTopicEnabled(t)])); -} - -function mutate(change: () => void): void { - const before = snapshot(); - change(); - const after = snapshot(); - for (const [topic, enabled] of after) { - if (before.get(topic) !== enabled && !(before.get(topic) === undefined && !enabled)) { - listeners.forEach((listener) => listener(topic, enabled)); - } - } -} - -/** Reads the flags document once and caches it. Never throws. */ -export function initRealtimeFlags(doc?: unknown): void { - mutate(() => { - fileFlags = parseFile(doc); - envOverrides = parseEnv(); - initialised = true; - }); -} - -/** Fetches /flags.json and initialises; any failure leaves every topic disabled. */ -export async function loadRealtimeFlags(fetchImpl: typeof fetch = fetch): Promise { - try { - const res = await fetchImpl(FLAGS_ENDPOINT, { cache: "no-store" }); - if (!res.ok) throw new Error(String(res.status)); - initRealtimeFlags(await res.json()); - } catch { - console.warn("[realtime-flags] could not load flags; all topics disabled"); - initRealtimeFlags(undefined); - } -} - -export function setTopicOverride(topic: string, value: boolean | null): void { - mutate(() => { - if (value === null) overrides.delete(topic); - else overrides.set(topic, value); - }); -} - -/** Kills a topic for already-connected sessions too (listeners tear subscriptions down). */ -export function killTopic(topic: string): void { - mutate(() => killed.add(topic)); -} - -export function reviveTopic(topic: string): void { - mutate(() => killed.delete(topic)); -} - -/** Observable for telemetry and for the registry to react to mid-session flips. */ -export function onTopicFlagChange(listener: TopicFlagListener): () => void { - listeners.add(listener); - return () => listeners.delete(listener); -} - -export function resetRealtimeFlagsForTests(): void { - fileFlags = new Map(); - envOverrides = new Map(); - overrides.clear(); - killed.clear(); - listeners.clear(); - initialised = false; -} - -export function realtimeFlagsInitialised(): boolean { - return initialised; -} diff --git a/src/lib/realtime/merge.ts b/src/lib/realtime/merge.ts index 54d844d..e69de29 100644 --- a/src/lib/realtime/merge.ts +++ b/src/lib/realtime/merge.ts @@ -1,359 +0,0 @@ -/** - * Realtime merge utilities. - * - * Event classes are either *replaceable* (only the newest value matters, e.g. - * oracle readings) or *append* (every event must be observed, e.g. logs). - * Replaceable classes are coalesced within a bounded window so a busy stream - * renders at most once per window while always flushing the latest value. - */ - -import type { RealtimeEvent } from './client'; -import { ShadowMode, type ShadowTopic } from './shadow'; - -export type MergeStrategy = 'replace' | 'append'; - -/** - * Event classes that are inherently replaceable. Only the newest value of a - * replaceable class is meaningful, so high-frequency bursts can be coalesced. - */ -export const REPLACEABLE_EVENT_CLASSES: ReadonlySet = new Set([ - 'oracle', - 'oracle-reading', - 'oracleReading', - 'price', - 'ticker', -]); - -export function isReplaceableEventClass(eventClass: string): boolean { - return REPLACEABLE_EVENT_CLASSES.has(eventClass); -} - -export function mergeStrategyFor(eventClass: string): MergeStrategy { - return isReplaceableEventClass(eventClass) ? 'replace' : 'append'; -} - -/** - * Merge a batch of events for a single topic. Replaceable classes keep only the - * latest event; append classes keep every event in arrival order. - */ -export function mergeEvents( - eventClass: string, - events: readonly T[], -): T[] { - if (events.length === 0) return []; - if (mergeStrategyFor(eventClass) === 'replace') { - return [events[events.length - 1]]; - } - return [...events]; -} - -/** - * Coalesces replaceable updates on a frame/timer budget. - * - * - `push` buffers the latest value per key. - * - `flush` drains the buffer and invokes the sink once with the latest value - * for each key (trailing-edge flush, so the final value is never dropped). - * - `schedule` requests a flush on the next animation frame (or a short timer - * fallback) unless coalescing is disabled. - * - * Coalescing is disabled under reduced motion and in automated tests, where - * flushing must be deterministic. - */ -export interface CoalescerOptions { - /** Called once per flush with the latest buffered value per key. */ - onFlush: (updates: Map) => void; - /** Override the coalescing window in ms. Defaults to one frame. */ - windowMs?: number; - /** Force-disable coalescing (e.g. reduced motion or tests). */ - disabled?: boolean; -} - -const DEFAULT_WINDOW_MS = 16; - -export function prefersReducedMotion(): boolean { - if (typeof window === 'undefined' || typeof window.matchMedia !== 'function') { - return false; - } - return window.matchMedia('(prefers-reduced-motion: reduce)').matches; -} - -export function isTestEnvironment(): boolean { - return ( - typeof process !== 'undefined' && - (process.env?.NODE_ENV === 'test' || process.env?.VITEST === 'true') - ); -} - -export function shouldCoalesce(disabled?: boolean): boolean { - if (disabled) return false; - if (isTestEnvironment()) return false; - if (prefersReducedMotion()) return false; - return true; -} - -export interface Coalescer { - push: (key: string, value: unknown) => void; - flush: () => void; - schedule: () => void; - dispose: () => void; -} - -export function createCoalescer(options: CoalescerOptions): Coalescer { - const { onFlush, windowMs = DEFAULT_WINDOW_MS } = options; - const buffer = new Map(); - let handle: ReturnType | null = null; - let frame: number | null = null; - let disposed = false; - - const clearScheduled = () => { - if (frame !== null && typeof cancelAnimationFrame === 'function') { - cancelAnimationFrame(frame); - } - frame = null; - if (handle !== null) { - clearTimeout(handle); - } - handle = null; - }; - - const flush = () => { - clearScheduled(); - if (buffer.size === 0) return; - const updates = new Map(buffer); - buffer.clear(); - onFlush(updates); - }; - - const schedule = () => { - if (disposed || handle !== null || frame !== null) return; - if (typeof requestAnimationFrame === 'function') { - frame = requestAnimationFrame(() => { - frame = null; - flush(); - }); - } else { - handle = setTimeout(() => { - handle = null; - flush(); - }, windowMs); - } - }; - - const push = (key: string, value: unknown) => { - if (disposed) return; - buffer.set(key, value); - if (shouldCoalesce(options.disabled)) { - schedule(); - } else { - // Deterministic path: flush immediately so tests and reduced-motion - // users observe every update without a frame delay. - flush(); - } - }; - - const dispose = () => { - disposed = true; - clearScheduled(); - buffer.clear(); - }; - - return { push, flush, schedule, dispose }; -} - -export type MergeFn = (current: T | undefined, incoming: T) => T; - -/** - * Default merge strategy: the latest value wins. - * Suitable for snapshot-style topics where each event carries full state. - */ -export const replaceMerge: MergeFn = (_current, incoming) => incoming; - -/** - * Merge a REST snapshot with any events that arrived while the snapshot was - * in flight. Events are applied in arrival order on top of the snapshot so a - * slow fetch never clobbers fresher streamed data. - */ -export function mergeSnapshot( - snapshot: T | undefined, - pendingEvents: readonly T[], - merge: MergeFn = replaceMerge as MergeFn, -): T | undefined { - let result = snapshot; - for (const event of pendingEvents) { - result = merge(result, event); - } - return result; -} - -/** - * Append-style merge for list topics: keeps the newest `limit` items and - * de-duplicates by `id` when present. - */ -export function appendMerge( - limit = 100, -): MergeFn { - return (current, incoming) => { - const base = current ?? []; - const next = incoming.id === undefined - ? [...base, incoming] - : [...base.filter((item) => item.id !== incoming.id), incoming]; - return next.length > limit ? next.slice(next.length - limit) : next; - }; -} - -export interface TopicState { - snapshot: T; - cursor: number; - appliedIds: Set; -} - -export interface MergeState { - polled: T | null; - realtime: T | null; -} - -export interface MergeOptions { - /** Maximum number of event ids retained for idempotency checks. */ - maxAppliedIds?: number; - /** Maximum cursor gap tolerated before falling back to a fresh snapshot. */ - maxReplayWindow?: number; - /** Shadow mode recorder; disabled by default. */ - shadow?: ShadowMode | null; - /** Topic used for shadow comparisons. */ - topic?: ShadowTopic; - /** Stable key for discrepancy reports. */ - key?: string; -} - -const DEFAULT_MAX_APPLIED_IDS = 512; -const DEFAULT_MAX_REPLAY_WINDOW = 1000; - -/** - * Create an empty topic state seeded from a REST snapshot. - * The snapshot's own cursor is used as the baseline so any event at or below - * it is considered already reflected and will not be re-applied. - */ -export function createTopicState(snapshot: T, cursor: number): TopicState { - return { snapshot, cursor, appliedIds: new Set() }; -} - -/** - * Determine whether a replayed event should be applied given the current - * cursor. Events at or below the cursor are already reflected in the snapshot - * (or were previously applied) and must be skipped to avoid double-apply. - */ -export function shouldApply(state: TopicState, event: RealtimeEvent): boolean { - if (typeof event.seq === 'number' && event.seq <= state.cursor) { - return false; - } - if (event.id && state.appliedIds.has(event.id)) { - return false; - } - return true; -} - -/** - * Idempotent merge reducer. Applies a single event to the topic state, - * advancing the cursor and recording the event id for dedupe. Returns the - * same state reference when the event is a no-op. - */ -export function applyEvent( - state: TopicState, - event: RealtimeEvent, - reducer: (snapshot: T, event: RealtimeEvent) => T, - options: MergeOptions = {}, -): TopicState { - if (!shouldApply(state, event)) { - return state; - } - - const maxAppliedIds = options.maxAppliedIds ?? DEFAULT_MAX_APPLIED_IDS; - const nextSnapshot = reducer(state.snapshot, event); - const nextCursor = - typeof event.seq === 'number' ? Math.max(state.cursor, event.seq) : state.cursor; - - const appliedIds = new Set(state.appliedIds); - if (event.id) { - appliedIds.add(event.id); - if (appliedIds.size > maxAppliedIds) { - const oldest = appliedIds.values().next().value; - if (oldest !== undefined) { - appliedIds.delete(oldest); - } - } - } - - return { snapshot: nextSnapshot, cursor: nextCursor, appliedIds }; -} - -/** - * Apply a batch of replayed events in order through the idempotent reducer. - */ -export function applyEvents( - state: TopicState, - events: RealtimeEvent[], - reducer: (snapshot: T, event: RealtimeEvent) => T, - options: MergeOptions = {}, -): TopicState { - return events.reduce((acc, event) => applyEvent(acc, event, reducer, options), state); -} - -/** - * Decide whether a stored cursor is still usable for replay against a freshly - * fetched snapshot cursor. If the gap is too large (or the snapshot cursor is - * behind the stored cursor, indicating an unknown/rolled-back state), replay - * is unavailable and the caller should fall back to a fresh snapshot with no - * streamed delta. - */ -export function canReplay( - storedCursor: number, - snapshotCursor: number, - options: MergeOptions = {}, -): boolean { - const maxReplayWindow = options.maxReplayWindow ?? DEFAULT_MAX_REPLAY_WINDOW; - if (storedCursor > snapshotCursor) { - return false; - } - return snapshotCursor - storedCursor <= maxReplayWindow; -} - -/** - * Snapshot-and-resume entry point. Given a freshly fetched REST snapshot and - * the stored cursor, produce the topic state to resume from. When replay is - * unavailable the state is reset to the snapshot with no streamed delta. - */ -export function resumeFromSnapshot( - snapshot: T, - snapshotCursor: number, - storedCursor: number | undefined, - options: MergeOptions = {}, -): TopicState { - if (storedCursor === undefined || !canReplay(storedCursor, snapshotCursor, options)) { - return createTopicState(snapshot, snapshotCursor); - } - return createTopicState(snapshot, snapshotCursor); -} - -/** - * Merge polled REST and realtime values, preferring REST. Shadow comparisons - * are recorded as a side effect but never influence the returned value. - */ -export function mergeValues>( - state: MergeState, - options: MergeOptions = {}, -): T | null { - const { shadow, topic, key } = options; - - if (shadow && topic && state.realtime) { - shadow.record( - topic, - key ?? topic, - state.realtime, - state.polled ?? null, - ); - } - - // REST is the known-good source of truth; realtime is only a fallback. - return state.polled ?? state.realtime ?? null; -} -} 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 }; +}