diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index b3a3c07..5713203 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -44,6 +44,10 @@ const wallet = useStoreSelector(walletStore, (s) => s); to avoid hydration mismatches. - Review checklist: reject any PR that subscribes to a whole store object. +## Testing realtime hooks + +Use the in-memory fake transport (`@/lib/realtime/testing`) instead of hand-rolled WebSocket mocks. See [docs/realtime.md](docs/realtime.md). + ## Realtime conventions - Subscribe only through `realtimeClient` from `@/lib/realtime`, inside a `useEffect`/`useLayoutEffect`; never `new WebSocket`/`new EventSource` or deep imports like `@/lib/realtime/client` (enforced by ESLint). Escape hatch: `// eslint-disable-next-line -- `. diff --git a/docs/attention-cues.md b/docs/attention-cues.md new file mode 100644 index 0000000..9b547e5 --- /dev/null +++ b/docs/attention-cues.md @@ -0,0 +1,20 @@ +# Attention cues (audio + haptics) + +Attention is governed centrally, like motion (`usePrefersReducedMotion`, +`prefers-reduced-motion` in `globals.css`). **No component plays sounds or +vibrates directly** — request a cue by severity through `AttentionCueService` +(`src/lib/attentionCue.ts`, wired by `src/hooks/useAttentionCue.ts`). + +Policy: + +- Only `critical` notifications (oracle trigger, claim settled/denied) cue. +- Off by default; persisted via `preferences.attentionCues`, toggled in the + Notification Center. +- Suppressed under `prefers-reduced-motion`, `Save-Data`, and quiet hours + (22:00–07:00 local). +- Audio is synthesized (Web Audio, no asset) and only starts after a user + gesture unlocks the `AudioContext`; earlier events are dropped, not queued. +- Bursts within 1s coalesce into one cue; the first emission (backfill) never + cues. `navigator.vibrate` is a silent no-op where unsupported (iOS). +- A cue never replaces the accessible (ARIA live) announcement. +- The `AudioContext` is closed on unmount. diff --git a/docs/realtime.md b/docs/realtime.md index 4f6c9bb..d81d641 100644 --- a/docs/realtime.md +++ b/docs/realtime.md @@ -19,6 +19,8 @@ The docs mirror the actual module layout so every path below is verifiable: | `src/lib/realtime/capabilities.ts` | Capability negotiation with the backend | | `src/lib/realtime/merge.ts` | Merging realtime payloads into cached REST data | | `src/lib/realtime/shadow.ts` | Shadow mode: observe realtime without applying it | +| `src/lib/realtime/contract.ts` | Topics, validators, fixtures, frames | +| `src/lib/realtime/testing.ts` | Fake transport for tests (never import from production) | | `public/flags.json` | Per-topic feature flag defaults | ## Topic / message contract @@ -43,6 +45,19 @@ A message has the shape: cached REST snapshot rather than replacing it wholesale. - Unknown `type` values are ignored (forward compatibility). +## Request/response + +`realtimeClient.request(topic, payload, { timeoutMs, signal })` multiplexes a +one-shot query over the open WebSocket. + +- Client frame: `{ id, topic, request }`; server replies `{ id, response }` or + `{ id, error: { code, message } }`. Frames with an `id` never reach topic + subscribers. +- Rejects with `RealtimeRequestError` (`timeout | aborted | disconnected | + overflow | server`). Pending requests reject on disconnect/reconnect and are + never replayed; at most 100 may be in flight. Payloads are never logged. +- SSE/long-poll fallbacks do not support requests; use REST. + ## Connection state machine `client.ts` drives a small state machine: @@ -100,6 +115,23 @@ The realtime contract above must be implemented in `refract-backend`. Until that lands, treat this document as the spec the backend is built against. Changes to the contract require a coordinated change in both repos. +## Testing + +Use the fake transport from `@/lib/realtime/testing` (tests only — never import +it from production code): + +```ts +const fake = installFakeRealtime(); +realtimeClient.subscribe("oracle_update", handler); +fake.current.open(); +fake.current.emit("oracle_update", { severity: "triggered", value: 9 }); // typed from the contract +fake.current.drop(); // force close; advance fake timers for reconnect +afterEach(() => fake.teardown()); +``` + +`src/lib/realtime/__tests__/topicCoverage.test.ts` fails when a contract topic +lacks a mock producer, a consumer, or valid fixtures (see its allowlist). + ## Operational runbook ### "Data stopped updating" diff --git a/src/components/NotificationCenter.tsx b/src/components/NotificationCenter.tsx index 8b47203..7c2bb4f 100644 --- a/src/components/NotificationCenter.tsx +++ b/src/components/NotificationCenter.tsx @@ -3,6 +3,7 @@ import { useEffect, useRef, useState } from "react"; import Link from "next/link"; import { useNotifications } from "@/hooks/useNotifications"; +import { usePreferencesStore } from "@/lib/store/preferencesStore"; import type { RefractNotification } from "@/lib/notifications"; import { VirtualList } from "@/components/ui/VirtualList"; @@ -27,6 +28,8 @@ const SEVERITY_COLORS: Record s.attentionCues); + const setAttentionCues = usePreferencesStore((s) => s.setAttentionCues); const containerRef = useRef(null); const buttonRef = useRef(null); @@ -163,6 +166,15 @@ export function NotificationCenter() { )} + +
{notifications.length === 0 ? (
diff --git a/src/hooks/useAttentionCue.ts b/src/hooks/useAttentionCue.ts new file mode 100644 index 0000000..3daf9dd --- /dev/null +++ b/src/hooks/useAttentionCue.ts @@ -0,0 +1,42 @@ +"use client"; + +import { useCallback, useEffect, useRef } from "react"; + +import { AttentionCueService } from "@/lib/attentionCue"; +import { usePreferencesStore } from "@/lib/store/preferencesStore"; +import type { RefractNotification } from "@/lib/notifications"; + +/** Requests audio/haptic cues for critical notifications that appear after first load. */ +export function useAttentionCue(notifications: RefractNotification[]) { + const service = useRef(null); + const seen = useRef | null>(null); + + useEffect(() => { + const svc = new AttentionCueService({ + enabled: () => usePreferencesStore.getState().attentionCues, + now: () => new Date(), + }); + service.current = svc; + const unlock = () => svc.unlock(); + window.addEventListener("pointerdown", unlock, { once: true }); + window.addEventListener("keydown", unlock, { once: true }); + return () => { + window.removeEventListener("pointerdown", unlock); + window.removeEventListener("keydown", unlock); + svc.dispose(); + service.current = null; + }; + }, []); + + useEffect(() => { + const ids = new Set(notifications.map((n) => n.id)); + // First emission is existing/backfilled state: seed, don't cue. + if (seen.current) { + const fresh = notifications.filter((n) => !seen.current!.has(n.id) && !n.read); + for (const n of fresh) service.current?.cue(n.severity); + } + seen.current = ids; + }, [notifications]); + + return { unlock: useCallback(() => service.current?.unlock(), []) }; +} diff --git a/src/hooks/useNotifications.ts b/src/hooks/useNotifications.ts index e8fe606..6f559dd 100644 --- a/src/hooks/useNotifications.ts +++ b/src/hooks/useNotifications.ts @@ -4,6 +4,7 @@ import { useEffect, useMemo, useState } from "react"; import { useWallet } from "@/lib/wallet/WalletProvider"; import { useHolderPolicies } from "./useHolderPolicies"; import { useClaims } from "./useClaims"; +import { useAttentionCue } from "./useAttentionCue"; import { useLockupStatus } from "./useLockupStatus"; import { useOracleStatus } from "./useOracleStatus"; import { @@ -54,6 +55,8 @@ export function useNotifications() { }); }, [policies, claims, lockupExpiresAt, oracleReadings, readIds]); + useAttentionCue(notifications); + const unreadCount = useMemo(() => { return notifications.filter((n) => !n.read).length; }, [notifications]); diff --git a/src/lib/__tests__/attentionCue.test.ts b/src/lib/__tests__/attentionCue.test.ts new file mode 100644 index 0000000..5fe30cf --- /dev/null +++ b/src/lib/__tests__/attentionCue.test.ts @@ -0,0 +1,87 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { AttentionCueService } from "../attentionCue"; + +const osc = () => ({ frequency: {}, connect: vi.fn(), start: vi.fn(), stop: vi.fn() }); +let created: number; +let closed: number; +let vibrate: ReturnType; +let enabled: boolean; +let now: Date; +let reduced: boolean; + +beforeEach(() => { + created = 0; + closed = 0; + enabled = true; + reduced = false; + now = new Date(2026, 0, 1, 12, 0, 0); + vibrate = vi.fn(); + vi.stubGlobal("AudioContext", function () { + created += 1; + return { + state: "running", + currentTime: 0, + destination: {}, + resume: vi.fn(), + close: () => { + closed += 1; + }, + createOscillator: osc, + createGain: () => ({ gain: {}, connect: vi.fn() }), + }; + }); + vi.stubGlobal("navigator", { vibrate }); + vi.stubGlobal("window", { matchMedia: () => ({ matches: reduced }) }); +}); +afterEach(() => vi.unstubAllGlobals()); + +const make = () => new AttentionCueService({ enabled: () => enabled, now: () => now }); + +describe("AttentionCueService", () => { + it("does nothing before user unlock", () => { + expect(make().cue("critical")).toBe(false); + expect(created).toBe(0); + }); + it("cues for critical after unlock and vibrates", () => { + const s = make(); + s.unlock(); + expect(s.cue("critical")).toBe(true); + expect(vibrate).toHaveBeenCalled(); + }); + it("is silent when the preference is off", () => { + const s = make(); + s.unlock(); + enabled = false; + expect(s.cue("critical")).toBe(false); + }); + it("ignores non-critical severities and replayed events", () => { + const s = make(); + s.unlock(); + expect(s.cue("warning")).toBe(false); + expect(s.cue("critical", { replayed: true })).toBe(false); + }); + it("coalesces bursts", () => { + const s = make(); + s.unlock(); + expect(s.cue("critical")).toBe(true); + expect(s.cue("critical")).toBe(false); + now = new Date(now.getTime() + 1500); + expect(s.cue("critical")).toBe(true); + }); + it("is silent during quiet hours and reduced motion", () => { + const s = make(); + s.unlock(); + now = new Date(2026, 0, 1, 23, 0, 0); + expect(s.cue("critical")).toBe(false); + now = new Date(2026, 0, 1, 12, 0, 0); + reduced = true; + expect(s.cue("critical")).toBe(false); + }); + it("releases the AudioContext on dispose", () => { + const s = make(); + s.unlock(); + s.dispose(); + expect(closed).toBe(1); + expect(s.cue("critical")).toBe(false); + }); +}); diff --git a/src/lib/attentionCue.ts b/src/lib/attentionCue.ts new file mode 100644 index 0000000..7695597 --- /dev/null +++ b/src/lib/attentionCue.ts @@ -0,0 +1,85 @@ +/** + * Attention-cue service: owns the AudioContext, unlock state and gating. + * Consumers request a cue by severity and never touch Web Audio directly. + * See docs/attention-cues.md for the policy. + */ +import type { RefractNotification } from "./notifications"; + +export const COALESCE_WINDOW_MS = 1_000; +export const QUIET_HOURS = { startHour: 22, endHour: 7 } as const; +const VIBRATION_PATTERN = [120, 60, 120]; + +export interface CueEnv { + enabled: () => boolean; + now: () => Date; +} + +type Severity = RefractNotification["severity"]; + +function inQuietHours(date: Date): boolean { + const h = date.getHours(); + return h >= QUIET_HOURS.startHour || h < QUIET_HOURS.endHour; +} + +export class AttentionCueService { + private ctx: AudioContext | null = null; + private unlocked = false; + private lastCueAt = -Infinity; + + constructor(private env: CueEnv) {} + + /** Call from a user gesture; creates/resumes the AudioContext. */ + unlock(): void { + if (this.unlocked || typeof AudioContext === "undefined") return; + this.ctx = new AudioContext(); + void this.ctx.resume?.(); + this.unlocked = true; + } + + /** + * Request a cue. Returns true when a cue fired. Replayed/backfilled events, + * non-critical severities, disabled preference, locked audio, reduced motion, + * save-data, quiet hours and bursts inside the coalescing window are dropped. + */ + cue(severity: Severity, opts: { replayed?: boolean } = {}): boolean { + if (severity !== "critical" || opts.replayed || !this.env.enabled()) return false; + if (!this.unlocked || !this.ctx) return false; + if (this.suppressed()) return false; + const now = this.env.now(); + if (now.getTime() - this.lastCueAt < COALESCE_WINDOW_MS) return false; + this.lastCueAt = now.getTime(); + this.beep(this.ctx); + try { + navigator.vibrate?.(VIBRATION_PATTERN); + } catch { + // Unsupported (iOS) — silent no-op. + } + return true; + } + + private suppressed(): boolean { + if (inQuietHours(this.env.now())) return true; + if (window.matchMedia?.("(prefers-reduced-motion: reduce)").matches) return true; + const conn = (navigator as Navigator & { connection?: { saveData?: boolean } }).connection; + return Boolean(conn?.saveData); + } + + private beep(ctx: AudioContext): void { + if (ctx.state === "suspended") void ctx.resume(); + const osc = ctx.createOscillator(); + const gain = ctx.createGain(); + osc.frequency.value = 880; + gain.gain.value = 0.08; + osc.connect(gain); + gain.connect(ctx.destination); + osc.start(); + osc.stop(ctx.currentTime + 0.15); + } + + dispose(): void { + void this.ctx?.close(); + this.ctx = null; + this.unlocked = false; + this.lastCueAt = -Infinity; + } +} diff --git a/src/lib/realtime/__tests__/correlation.test.ts b/src/lib/realtime/__tests__/correlation.test.ts new file mode 100644 index 0000000..f3b9d49 --- /dev/null +++ b/src/lib/realtime/__tests__/correlation.test.ts @@ -0,0 +1,81 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { realtimeClient } from "../client"; +import { MAX_PENDING_REQUESTS } from "../correlation"; +import { installFakeRealtime, type FakeRealtime } from "../testing"; + +let fake: FakeRealtime; +let t: ReturnType["current"]; +beforeEach(() => { + vi.useFakeTimers(); + fake = installFakeRealtime(); + realtimeClient.connect(); + t = fake.current; +}); +afterEach(() => { + fake.teardown(); + vi.useRealTimers(); +}); + +const idOf = (n: number) => JSON.parse(t.sent[n]).id as number; + +describe("request/response correlation", () => { + it("resolves, clears the timeout and never leaks responses to topic listeners", async () => { + const listener = vi.fn(); + realtimeClient.subscribe("snapshot", listener); + const p = realtimeClient.request("snapshot", {}, { timeoutMs: 500 }); + t.respond(idOf(0), "ok"); + await expect(p).resolves.toBe("ok"); + expect(vi.getTimerCount()).toBe(0); + expect(listener).not.toHaveBeenCalled(); + }); + + it("generates monotonic ids for concurrent requests", async () => { + const a = realtimeClient.request("q"); + const b = realtimeClient.request("q"); + expect(idOf(1)).toBeGreaterThan(idOf(0)); + t.respond(idOf(1), "b"); + t.respond(idOf(0), "a"); + await expect(Promise.all([a, b])).resolves.toEqual(["a", "b"]); + }); + + it("times out", async () => { + const p = realtimeClient.request("q", {}, { timeoutMs: 100 }); + const assertion = expect(p).rejects.toMatchObject({ reason: "timeout" }); + await vi.advanceTimersByTimeAsync(100); + await assertion; + }); + + it("aborts via AbortSignal", async () => { + const ac = new AbortController(); + const p = realtimeClient.request("q", {}, { signal: ac.signal }); + ac.abort(); + await expect(p).rejects.toMatchObject({ reason: "aborted" }); + await expect(realtimeClient.request("q", {}, { signal: ac.signal })).rejects.toMatchObject({ reason: "aborted" }); + }); + + it("rejects in-flight requests on disconnect and ignores stale/duplicate ids", async () => { + const p = realtimeClient.request("q"); + const id = idOf(0); + t.drop(); + await expect(p).rejects.toMatchObject({ reason: "disconnected" }); + expect(() => { + t.respond(id, "late"); + t.respond(id, "dup"); + }).not.toThrow(); + expect(vi.getTimerCount()).toBe(1); // only the reconnect timer remains + }); + + it("rejects pending requests and clears them on teardown", async () => { + const p = realtimeClient.request("q"); + realtimeClient.disconnect(); + await expect(p).rejects.toMatchObject({ reason: "disconnected" }); + expect(vi.getTimerCount()).toBe(0); + }); + + it("bounds the pending map", async () => { + const ps = Array.from({ length: MAX_PENDING_REQUESTS }, () => realtimeClient.request("q")); + await expect(realtimeClient.request("q")).rejects.toMatchObject({ reason: "overflow" }); + realtimeClient.disconnect(); + await Promise.allSettled(ps); + }); +}); diff --git a/src/lib/realtime/__tests__/testKit.test.ts b/src/lib/realtime/__tests__/testKit.test.ts new file mode 100644 index 0000000..bc03268 --- /dev/null +++ b/src/lib/realtime/__tests__/testKit.test.ts @@ -0,0 +1,67 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { realtimeClient } from "../client"; +import { installFakeRealtime, type FakeRealtime } from "../testing"; + +let fake: FakeRealtime; +beforeEach(() => { + vi.useFakeTimers(); + fake = installFakeRealtime(); +}); +afterEach(() => { + fake.teardown(); + expect(vi.getTimerCount()).toBe(0); + vi.useRealTimers(); +}); + +describe("fake transport kit", () => { + it("delivers typed events to subscribers and unsubscribes cleanly", () => { + const handler = vi.fn(); + const off = realtimeClient.subscribe("oracle_update", handler); + fake.current.open(); + fake.current.emit("oracle_update", { severity: "triggered", value: 9 }); + expect(handler).toHaveBeenCalledWith({ severity: "triggered", value: 9 }); + off(); + fake.current.emit("oracle_update", { severity: "normal", value: 1 }); + expect(handler).toHaveBeenCalledTimes(1); + }); + + it("rejects invalid payloads at the type level", () => { + const t = fake.transports[0] ?? (realtimeClient.subscribe("x", () => {}), fake.current); + // @ts-expect-error severity is not part of the contract + t.emit("oracle_update", { severity: "bogus", value: 1 }); + // @ts-expect-error unknown topic + t.emit("nope", {}); + }); + + it("reconnects with backoff after a forced drop and supports StrictMode double-mount", () => { + const a = realtimeClient.subscribe("oracle_update", () => {}); + a(); + const b = realtimeClient.subscribe("oracle_update", () => {}); + expect(fake.transports).toHaveLength(1); + fake.current.drop(); + vi.advanceTimersByTime(1000); + expect(fake.transports).toHaveLength(2); + b(); + }); + + it("leaves no timers after teardown mid-backoff", () => { + realtimeClient.subscribe("oracle_update", () => {}); + fake.current.error(); + expect(vi.getTimerCount()).toBe(1); + fake.teardown(); + expect(vi.getTimerCount()).toBe(0); + fake = installFakeRealtime(); + }); + + it("resolves requests via respond() and rejects on fail()", async () => { + realtimeClient.subscribe("oracle_update", () => {}); + const t = fake.current; + const ok = realtimeClient.request("snapshot", { a: 1 }); + const { id } = JSON.parse(t.sent[0]); + t.respond(id, { v: 1 }); + await expect(ok).resolves.toEqual({ v: 1 }); + const bad = realtimeClient.request("snapshot"); + t.fail(JSON.parse(t.sent[1]).id, "E", "boom"); + await expect(bad).rejects.toMatchObject({ reason: "server", code: "E" }); + }); +}); diff --git a/src/lib/realtime/__tests__/topicCoverage.test.ts b/src/lib/realtime/__tests__/topicCoverage.test.ts new file mode 100644 index 0000000..edadc61 --- /dev/null +++ b/src/lib/realtime/__tests__/topicCoverage.test.ts @@ -0,0 +1,62 @@ +import { readdirSync, readFileSync, statSync } from "node:fs"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; +import { REALTIME_TOPICS } from "../contract"; +import { MOCK_PRODUCERS } from "../testing/producers"; + +/** + * Topics intentionally lacking a real consumer. Every entry needs a reason and + * is itself validated: stale entries (unknown topic, or now consumed) fail. + */ +export const UNCONSUMED_ALLOWLIST: Record = { + oracle_update: "realtimeClient is wired behind the disabled `realtime-oracle` flag; no hook subscribes yet.", +}; + +const SRC = join(process.cwd(), "src"); +const SKIP = /(__tests__|\/testing|\.test\.tsx?$|\/realtime\/contract\.ts$)/; + +function walk(dir: string, out: string[] = []): string[] { + for (const name of readdirSync(dir)) { + const p = join(dir, name); + if (statSync(p).isDirectory()) walk(p, out); + else if (/\.tsx?$/.test(name) && !SKIP.test(p)) out.push(p); + } + return out; +} + +export function findConsumers(topic: string, files: Record): string[] { + const re = new RegExp(`subscribe\\(\\s*["'\`]${topic}["'\`]`); + return Object.keys(files).filter((f) => re.test(files[f])); +} + +const files = Object.fromEntries(walk(SRC).map((f) => [f, readFileSync(f, "utf8")])); +const topics = Object.keys(REALTIME_TOPICS); + +describe("realtime topic coverage", () => { + it.each(topics)("%s has a producer, a consumer (or allowlist entry) and valid fixtures", (topic) => { + const def = REALTIME_TOPICS[topic as keyof typeof REALTIME_TOPICS]; + expect(MOCK_PRODUCERS[topic], `${topic}: no mock producer`).toBeDefined(); + expect(def.fixtures.length, `${topic}: no fixtures`).toBeGreaterThan(0); + for (const fixture of def.fixtures) { + expect(() => def.parse(fixture), `${topic}: fixture fails validation`).not.toThrow(); + } + if (!(topic in UNCONSUMED_ALLOWLIST)) { + expect(findConsumers(topic, files), `${topic}: no consumer`).not.toHaveLength(0); + } + }); + + it("has no producers for unknown topics and a valid allowlist", () => { + for (const t of Object.keys(MOCK_PRODUCERS)) expect(topics, `producer for unknown topic ${t}`).toContain(t); + for (const [t, reason] of Object.entries(UNCONSUMED_ALLOWLIST)) { + expect(topics, `stale allowlist entry ${t}`).toContain(t); + expect(reason.length).toBeGreaterThan(0); + expect(findConsumers(t, files), `${t} is now consumed; remove from allowlist`).toHaveLength(0); + } + }); + + it("meta: detects a missing consumer and an invalid fixture", () => { + expect(findConsumers("x", { a: 'realtimeClient.subscribe("x", h)' })).toEqual(["a"]); + expect(findConsumers("x", { a: "nothing" })).toEqual([]); + expect(() => REALTIME_TOPICS.oracle_update.parse({ severity: "bad" })).toThrow(); + }); +}); diff --git a/src/lib/realtime/client.ts b/src/lib/realtime/client.ts index e69de29..5c444eb 100644 --- a/src/lib/realtime/client.ts +++ b/src/lib/realtime/client.ts @@ -0,0 +1,134 @@ +import { API_BASE_URL } from "../api/client"; + +import { RequestCorrelator, type RequestOptions } from "./correlation"; + +type Handler = (payload: unknown) => void; + +/** The subset of WebSocket the client relies on; fakes implement the same shape. */ +export interface RealtimeTransport { + addEventListener(type: "open" | "message" | "close" | "error", listener: (event: { data?: unknown }) => void): void; + send(data: string): void; + close(): void; +} + +export type TransportFactory = (url: string) => RealtimeTransport; + +const defaultTransportFactory: TransportFactory = (url) => new WebSocket(url) as unknown as RealtimeTransport; +let transportFactory: TransportFactory = defaultTransportFactory; +let customTransport = false; + +/** Test-only: inject a transport factory. Pass `null` to restore the WebSocket default. */ +export function __setTransportFactory(factory: TransportFactory | null): void { + transportFactory = factory ?? defaultTransportFactory; + customTransport = factory !== null; +} + +const MAX_BACKOFF_MS = 30_000; +const BASE_BACKOFF_MS = 1_000; + +/** + * 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. + */ +class RealtimeClient { + private socket: RealtimeTransport | null = null; + private handlers = new Map>(); + private attempt = 0; + private closedByUser = false; + private correlator = new RequestCorrelator(); + private reconnectTimer: ReturnType | null = null; + + connect(): void { + if ((!customTransport && typeof WebSocket === "undefined") || this.socket) return; + this.closedByUser = false; + this.open(); + } + + private open(): void { + const wsUrl = API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws"); + try { + this.socket = transportFactory(wsUrl); + } catch { + this.scheduleReconnect(); + return; + } + + this.socket.addEventListener("open", () => { + this.attempt = 0; + }); + this.socket.addEventListener("message", (event) => { + try { + const frame = JSON.parse(event.data as string) as { topic: string; payload: unknown }; + if (this.correlator.handleFrame(frame)) return; + const { topic, payload } = frame; + this.handlers.get(topic)?.forEach((handler) => handler(payload)); + } catch { + // Malformed message — ignore, don't crash the connection. + } + }); + this.socket.addEventListener("close", () => { + this.socket = null; + this.correlator.rejectAll(); + if (!this.closedByUser) this.scheduleReconnect(); + }); + this.socket.addEventListener("error", () => { + this.socket?.close(); + }); + } + + private scheduleReconnect(): void { + if (this.reconnectTimer) return; + const delay = Math.min(BASE_BACKOFF_MS * 2 ** this.attempt, MAX_BACKOFF_MS); + this.attempt += 1; + this.reconnectTimer = setTimeout(() => { + this.reconnectTimer = null; + if (!this.closedByUser) 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(); + + return () => { + this.handlers.get(topic)?.delete(handler); + }; + } + + /** One-shot request/response over the open socket; rejects if not connected. */ + request(topic: string, payload?: unknown, options?: RequestOptions): Promise { + return this.correlator.request( + (frame) => { + if (!this.socket) throw new Error("Realtime socket is not connected"); + this.socket.send(frame); + }, + topic, + payload, + options, + ); + } + + /** Test-only: drop connection, handlers and reconnect state. */ + __reset(): void { + this.disconnect(); + this.handlers.clear(); + this.attempt = 0; + this.closedByUser = false; + } + + disconnect(): void { + this.closedByUser = true; + this.correlator.rejectAll(); + if (this.reconnectTimer) clearTimeout(this.reconnectTimer); + this.socket?.close(); + this.socket = null; + } +} + +export const realtimeClient = new RealtimeClient(); diff --git a/src/lib/realtime/contract.ts b/src/lib/realtime/contract.ts new file mode 100644 index 0000000..68fdf1e --- /dev/null +++ b/src/lib/realtime/contract.ts @@ -0,0 +1,62 @@ +/** + * Realtime contract: the single source of truth for topics and frames. + * + * Push frame: `{ topic, payload }` + * Request frame: `{ id, topic, request: payload }` (client -> server) + * Response frame: `{ id, response: payload }` (server -> client) + * Error frame: `{ id, error: { code, message } }` (server -> client) + * + * Frames carrying an `id` are matched by the correlation layer and are never + * delivered to topic subscribers. Only the WebSocket transport supports + * requests; SSE/long-poll fallbacks must degrade to REST. + */ + +export interface TopicDef { + /** Runtime validator; returns the parsed payload or throws. */ + parse: (value: unknown) => T; + fixtures: readonly unknown[]; +} + +function isRecord(v: unknown): v is Record { + return typeof v === "object" && v !== null; +} + +export interface OracleUpdatePayload { + severity: "normal" | "high" | "triggered"; + value: number; +} + +export const REALTIME_TOPICS = { + oracle_update: { + parse(value: unknown): OracleUpdatePayload { + if ( + !isRecord(value) || + !["normal", "high", "triggered"].includes(value.severity as string) || + typeof value.value !== "number" + ) { + throw new Error("invalid oracle_update payload"); + } + return value as unknown as OracleUpdatePayload; + }, + fixtures: [ + { severity: "normal", value: 1 }, + { severity: "triggered", value: 9 }, + ], + } satisfies TopicDef, +} as const; + +export type RealtimeTopic = keyof typeof REALTIME_TOPICS; +export type TopicPayload = ReturnType<(typeof REALTIME_TOPICS)[K]["parse"]>; + +export interface RealtimeErrorFrame { + code: string; + message: string; +} + +export type RealtimeResponseFrame = + | { id: number; response: unknown } + | { id: number; error: RealtimeErrorFrame }; + +export function isResponseFrame(frame: unknown): frame is RealtimeResponseFrame { + return isRecord(frame) && typeof frame.id === "number" && ("response" in frame || "error" in frame); +} diff --git a/src/lib/realtime/correlation.ts b/src/lib/realtime/correlation.ts new file mode 100644 index 0000000..633bfb6 --- /dev/null +++ b/src/lib/realtime/correlation.ts @@ -0,0 +1,111 @@ +import { isResponseFrame } from "./contract"; + +export type RequestFailureReason = "timeout" | "aborted" | "disconnected" | "overflow" | "server"; + +export class RealtimeRequestError extends Error { + constructor( + readonly reason: RequestFailureReason, + message: string, + readonly code?: string, + ) { + super(message); + this.name = "RealtimeRequestError"; + } +} + +interface Pending { + resolve: (value: unknown) => void; + reject: (err: RealtimeRequestError) => void; + timer: ReturnType; + cleanup: () => void; +} + +export interface RequestOptions { + timeoutMs?: number; + signal?: AbortSignal; +} + +export const DEFAULT_REQUEST_TIMEOUT_MS = 10_000; +export const MAX_PENDING_REQUESTS = 100; + +/** + * Owns correlation IDs and the pending-request map. Transports only own + * framing: they call `send` to write a frame and `handleFrame` on dispatch. + */ +export class RequestCorrelator { + private nextId = 1; + private pending = new Map(); + + get size(): number { + return this.pending.size; + } + + request( + send: (frame: string) => void, + topic: string, + payload: unknown, + { timeoutMs = DEFAULT_REQUEST_TIMEOUT_MS, signal }: RequestOptions = {}, + ): Promise { + if (signal?.aborted) { + return Promise.reject(new RealtimeRequestError("aborted", "Request aborted")); + } + if (this.pending.size >= MAX_PENDING_REQUESTS) { + return Promise.reject(new RealtimeRequestError("overflow", "Too many pending requests")); + } + const id = this.nextId++; + return new Promise((resolve, reject) => { + const finish = () => { + const entry = this.pending.get(id); + if (!entry) return; + clearTimeout(entry.timer); + entry.cleanup(); + this.pending.delete(id); + }; + const onAbort = () => { + finish(); + reject(new RealtimeRequestError("aborted", "Request aborted")); + }; + signal?.addEventListener("abort", onAbort, { once: true }); + this.pending.set(id, { + resolve: (v) => { + finish(); + resolve(v as T); + }, + reject: (e) => { + finish(); + reject(e); + }, + timer: setTimeout(() => { + finish(); + reject(new RealtimeRequestError("timeout", `Request "${topic}" timed out`)); + }, timeoutMs), + cleanup: () => signal?.removeEventListener("abort", onAbort), + }); + try { + send(JSON.stringify({ id, topic, request: payload })); + } catch (err) { + this.pending.get(id)?.reject(new RealtimeRequestError("disconnected", String(err))); + } + }); + } + + /** Returns true when the frame was a response/error frame (consumed here). */ + handleFrame(frame: unknown): boolean { + if (!isResponseFrame(frame)) return false; + const entry = this.pending.get(frame.id); + // Stale or duplicate ids are consumed but ignored. + if (!entry) return true; + if ("error" in frame) { + entry.reject(new RealtimeRequestError("server", frame.error.message, frame.error.code)); + } else { + entry.resolve(frame.response); + } + return true; + } + + rejectAll(reason: RequestFailureReason = "disconnected"): void { + for (const entry of [...this.pending.values()]) { + entry.reject(new RealtimeRequestError(reason, "Realtime connection closed")); + } + } +} diff --git a/src/lib/realtime/testing/index.ts b/src/lib/realtime/testing/index.ts new file mode 100644 index 0000000..9b8924e --- /dev/null +++ b/src/lib/realtime/testing/index.ts @@ -0,0 +1,112 @@ +/** + * Test-only kit: an in-memory realtime transport. Import from + * `@/lib/realtime/testing` in tests only — nothing in `src/` outside tests may + * import it, so it never reaches the production bundle. + */ +import { + __setTransportFactory, + realtimeClient, + type RealtimeTransport, +} from "../client"; +import type { RealtimeTopic, TopicPayload } from "../contract"; + +type Listener = (event: { data?: unknown }) => void; + +export class FakeTransport implements RealtimeTransport { + readonly sent: string[] = []; + closed = false; + private listeners: Record = { open: [], message: [], close: [], error: [] }; + + addEventListener(type: "open" | "message" | "close" | "error", listener: Listener): void { + this.listeners[type].push(listener); + } + + send(data: string): void { + this.sent.push(data); + } + + close(): void { + if (this.closed) return; + this.closed = true; + this.fire("close"); + } + + private fire(type: string, event: { data?: unknown } = {}): void { + [...this.listeners[type]].forEach((l) => l(event)); + } + + get listenerCount(): number { + return Object.values(this.listeners).reduce((n, l) => n + l.length, 0); + } + + /** Force the connection open. */ + open(): void { + this.fire("open"); + } + + /** Emit a push event; the payload type is derived from the contract. */ + emit(topic: K, payload: TopicPayload): void { + this.fire("message", { data: JSON.stringify({ topic, payload }) }); + } + + /** Emit a raw server frame (responses, errors, malformed data). */ + emitRaw(frame: unknown): void { + this.fire("message", { data: typeof frame === "string" ? frame : JSON.stringify(frame) }); + } + + respond(id: number, response: unknown): void { + this.emitRaw({ id, response }); + } + + fail(id: number, code: string, message: string): void { + this.emitRaw({ id, error: { code, message } }); + } + + /** Force a transport error (the client closes and schedules a reconnect). */ + error(): void { + this.fire("error"); + } + + /** Simulate a server-side drop. */ + drop(): void { + this.closed = true; + this.fire("close"); + } +} + +export interface FakeRealtime { + /** Every transport the client opened, in order. */ + transports: FakeTransport[]; + /** The most recent transport. */ + readonly current: FakeTransport; + /** Resolves once the client has opened `n` transports (default 1). */ + waitForConnection(n?: number): Promise; + /** Tear down: disconnect, reset the client and restore the real transport. */ + teardown(): void; +} + +/** Install the fake transport into the realtime client. Call `teardown()` in afterEach. */ +export function installFakeRealtime(): FakeRealtime { + realtimeClient.__reset(); + const transports: FakeTransport[] = []; + __setTransportFactory(() => { + const t = new FakeTransport(); + transports.push(t); + return t; + }); + return { + transports, + get current() { + return transports[transports.length - 1]; + }, + async waitForConnection(n = 1) { + for (let i = 0; i < 100 && transports.length < n; i += 1) await Promise.resolve(); + if (transports.length < n) throw new Error(`Expected ${n} transport(s), got ${transports.length}`); + return transports[n - 1]; + }, + teardown() { + realtimeClient.__reset(); + __setTransportFactory(null); + }, + }; +} diff --git a/src/lib/realtime/testing/producers.ts b/src/lib/realtime/testing/producers.ts new file mode 100644 index 0000000..0db086a --- /dev/null +++ b/src/lib/realtime/testing/producers.ts @@ -0,0 +1,6 @@ +import { REALTIME_TOPICS, type RealtimeTopic } from "../contract"; + +/** Mock-server producer registry: one frame generator per contract topic. */ +export const MOCK_PRODUCERS: Record unknown> = { + oracle_update: () => REALTIME_TOPICS.oracle_update.fixtures[0], +} satisfies Partial unknown>>; diff --git a/src/lib/store/preferencesStore.ts b/src/lib/store/preferencesStore.ts index a3610a6..5012729 100644 --- a/src/lib/store/preferencesStore.ts +++ b/src/lib/store/preferencesStore.ts @@ -28,10 +28,12 @@ export interface PreferencesState { locale: string; compactNumbers: boolean; explorerNetwork: ExplorerNetworkPreference; + attentionCues: boolean; setMotion: (motion: MotionPreference) => void; setLocale: (locale: string) => void; setCompactNumbers: (compactNumbers: boolean) => void; setExplorerNetwork: (explorerNetwork: ExplorerNetworkPreference) => void; + setAttentionCues: (attentionCues: boolean) => void; reset: () => void; } @@ -43,6 +45,7 @@ export const DEFAULT_PREFERENCES = { locale: 'en-US', compactNumbers: false, explorerNetwork: 'auto' as ExplorerNetworkPreference, + attentionCues: false, }; /** @@ -54,7 +57,7 @@ interface PersistedEnvelope { version: number; state: Pick< PreferencesState, - 'motion' | 'locale' | 'compactNumbers' | 'explorerNetwork' + 'motion' | 'locale' | 'compactNumbers' | 'explorerNetwork' | 'attentionCues' >; } @@ -78,6 +81,7 @@ export const usePreferencesStore = create()( setLocale: (locale) => set({ locale }), setCompactNumbers: (compactNumbers) => set({ compactNumbers }), setExplorerNetwork: (explorerNetwork) => set({ explorerNetwork }), + setAttentionCues: (attentionCues) => set({ attentionCues }), reset: () => set({ ...DEFAULT_PREFERENCES }), }), { @@ -92,6 +96,7 @@ export const usePreferencesStore = create()( locale: state.locale, compactNumbers: state.compactNumbers, explorerNetwork: state.explorerNetwork, + attentionCues: state.attentionCues, }, }), // Rehydrate from the versioned envelope, falling back to defaults for @@ -108,6 +113,7 @@ export const usePreferencesStore = create()( locale: stored.locale ?? current.locale, compactNumbers: stored.compactNumbers ?? current.compactNumbers, explorerNetwork: stored.explorerNetwork ?? current.explorerNetwork, + attentionCues: stored.attentionCues ?? current.attentionCues, }; }, },