From 6fdaa63b4e411ef3b8488b3e5903c4fc8fe3e3ce Mon Sep 17 00:00:00 2001 From: Noah Akerityo Date: Wed, 21 Oct 2026 18:25:57 +0000 Subject: [PATCH] feat: improve realtime batching and data freshness --- .github/workflows/ci.yml | 3 + openapi/refract-api.yaml | 12 + package.json | 4 +- scripts/fixtures/malformed-openapi.yaml | 5 + scripts/gen-realtime-types.mjs | 128 +++++++ scripts/gen-realtime-types.test.ts | 16 + src/app/globals.css | 9 + src/app/page.tsx | 3 + src/app/provide/page.tsx | 4 +- src/components/DataFreshness.tsx | 86 +++++ src/hooks/useOracleStatus.ts | 70 ++++ src/hooks/usePoolStats.ts | 72 ++++ src/hooks/useRecentClaims.ts | 53 +++ src/lib/api/claims.ts | 31 +- src/lib/api/oracle.ts | 4 +- src/lib/realtime/client.test.ts | 32 ++ src/lib/realtime/client.ts | 393 +++++++++++++++++++-- src/lib/realtime/freshness.test.ts | 21 ++ src/lib/realtime/freshness.ts | 16 + src/lib/realtime/generated/index.ts | 3 + src/lib/realtime/generated/schemas.test.ts | 44 +++ src/lib/realtime/generated/schemas.ts | 74 ++++ src/lib/realtime/types.ts | 9 + 23 files changed, 1051 insertions(+), 41 deletions(-) create mode 100644 scripts/fixtures/malformed-openapi.yaml create mode 100644 scripts/gen-realtime-types.mjs create mode 100644 scripts/gen-realtime-types.test.ts create mode 100644 src/components/DataFreshness.tsx create mode 100644 src/lib/realtime/client.test.ts create mode 100644 src/lib/realtime/freshness.test.ts create mode 100644 src/lib/realtime/freshness.ts create mode 100644 src/lib/realtime/generated/index.ts create mode 100644 src/lib/realtime/generated/schemas.test.ts create mode 100644 src/lib/realtime/generated/schemas.ts create mode 100644 src/lib/realtime/types.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f273c23..c662b1e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -23,6 +23,9 @@ jobs: - name: Install dependencies run: npm ci + - name: Check realtime generated types + run: npm run check:realtime-types + - name: Run ESLint run: npm run lint diff --git a/openapi/refract-api.yaml b/openapi/refract-api.yaml index 1740062..b9713ae 100644 --- a/openapi/refract-api.yaml +++ b/openapi/refract-api.yaml @@ -131,6 +131,18 @@ components: type: string enum: [low, medium, high, triggered] message: { type: string } + RealtimeEnvelope: + type: object + required: [topic, payload] + properties: + topic: { type: string } + payload: + type: object + additionalProperties: true + metadata: + type: object + nullable: true + additionalProperties: true ClaimRecord: type: object required: [policyId, holder, coverageType, triggered, payout, reason, processedAt] diff --git a/package.json b/package.json index 51f59c9..09ad3b7 100644 --- a/package.json +++ b/package.json @@ -19,7 +19,9 @@ "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", + "gen:realtime-types": "node scripts/gen-realtime-types.mjs", + "check:realtime-types": "node scripts/gen-realtime-types.mjs --check" }, "lint-staged": { "*.{ts,tsx}": [ diff --git a/scripts/fixtures/malformed-openapi.yaml b/scripts/fixtures/malformed-openapi.yaml new file mode 100644 index 0000000..1ffb19b --- /dev/null +++ b/scripts/fixtures/malformed-openapi.yaml @@ -0,0 +1,5 @@ +openapi: 3.0.3 +components: + schemas: + Unrelated: + type: object diff --git a/scripts/gen-realtime-types.mjs b/scripts/gen-realtime-types.mjs new file mode 100644 index 0000000..4ea62fa --- /dev/null +++ b/scripts/gen-realtime-types.mjs @@ -0,0 +1,128 @@ +import { readFileSync, writeFileSync } from "node:fs"; +import { resolve } from "node:path"; + +const root = process.cwd(); +const specPath = resolve(root, process.env.REALTIME_OPENAPI_SPEC ?? "openapi/refract-api.yaml"); +const outPath = resolve(root, "src/lib/realtime/generated/schemas.ts"); +const indexPath = resolve(root, "src/lib/realtime/generated/index.ts"); +const check = process.argv.includes("--check"); +const spec = readFileSync(specPath, "utf8"); +if ( + !/^openapi:\s*3\./m.test(spec) || + !/^components:\s*$/m.test(spec) || + !/^ schemas:\s*$/m.test(spec) +) + throw new Error("Invalid OpenAPI document: expected OpenAPI 3 and components.schemas"); +const wanted = ["OracleReading", "ClaimRecord", "PoolStats", "RealtimeEnvelope"]; +const schemas = {}; +for (const name of wanted) { + const start = spec.indexOf(` ${name}:`); + if (start < 0) throw new Error(`Required realtime schema '${name}' is missing from ${specPath}`); + const boundary = spec.slice(start + 5).search(/\n [A-Z][A-Za-z0-9]*:/); + const next = boundary < 0 ? -1 : start + 5 + boundary; + const block = spec.slice(start, next < 0 ? undefined : next); + const requiredLine = block.match(/^ required:\s*\[([^\]]*)\]/m)?.[1]; + if (!requiredLine) throw new Error(`Schema '${name}' has no required field declaration`); + const required = new Set( + requiredLine + .split(",") + .map((v) => v.trim()) + .filter(Boolean) + ); + const propsStart = block.indexOf(" properties:"); + if (propsStart < 0) throw new Error(`Schema '${name}' has no properties`); + const propsText = block.slice(propsStart + 7); + const props = []; + const lines = propsText.split("\n"); + for (let i = 1; i < lines.length; i++) { + const line = lines[i]; + const field = line.match(/^ ([A-Za-z][\w]*):(?:\s*\{\s*(.*?)\s*\})?\s*$/); + if (!field) continue; + let descriptor = field[2] ?? ""; + if (!descriptor) { + const nested = lines.slice(i + 1, i + 5).join(" "); + descriptor = nested; + } + const type = descriptor.match( + /(?:^|[,\s])type:\s*(string|number|integer|boolean|object|array)/ + )?.[1]; + const enumBody = descriptor.match(/enum:\s*\[([^\]]*)\]/)?.[1]; + const nullable = /nullable:\s*true/.test(descriptor); + if (!type && !enumBody) throw new Error(`Unsupported or malformed field '${name}.${field[1]}'`); + let tsType = + type === "integer" || type === "number" + ? "number" + : type === "boolean" + ? "boolean" + : type === "object" + ? "Record" + : type === "array" + ? "unknown[]" + : "string"; + let zod = + type === "number" || type === "integer" + ? "z.number()" + : type === "boolean" + ? "z.boolean()" + : type === "object" + ? "z.record(z.string(), z.unknown())" + : "z.string()"; + if (enumBody) { + const vals = enumBody + .split(",") + .map((v) => v.trim()) + .filter(Boolean); + tsType = vals + .map((v) => v.replace(/^['"]|['"]$/g, "")) + .map((v) => JSON.stringify(v)) + .join(" | "); + zod = `z.enum([${vals.map((v) => JSON.stringify(v.replace(/^['"]|['"]$/g, ""))).join(", ")}])`; + } + if (nullable) { + tsType += " | null"; + zod += ".nullable()"; + } + const optional = !required.has(field[1]); + props.push({ key: field[1], tsType, zod: optional ? `${zod}.optional()` : zod, optional }); + } + props.sort((a, b) => (a.key < b.key ? -1 : a.key > b.key ? 1 : 0)); + for (const field of required) + if (!props.some((p) => p.key === field)) + throw new Error(`Required field '${name}.${field}' is missing from properties`); + schemas[name] = props; +} +const lines = [ + "// Generated by scripts/gen-realtime-types.mjs. Do not edit.", + 'import { z } from "zod";', + "", +]; +for (const [name, props] of Object.entries(schemas)) { + lines.push(`export interface ${name} {`); + for (const p of props) lines.push(` ${p.key}${p.optional ? "?" : ""}: ${p.tsType};`); + lines.push("}", `export const ${name}Schema = z.object({`); + for (const p of props) lines.push(` ${p.key}: ${p.zod},`); + lines.push("});", ""); +} +const output = `${lines.join("\n").trimEnd()}\n`; +const indexOutput = `// Generated by scripts/gen-realtime-types.mjs. Do not edit.\nexport { ${wanted.map((n) => `${n}Schema`).join(", ")} } from "./schemas";\nexport type { ${wanted.join(", ")} } from "./schemas";\n`; +if (check) { + let current = ""; + try { + current = readFileSync(outPath, "utf8"); + } catch { + /* drift */ + } + let currentIndex = ""; + try { + currentIndex = readFileSync(indexPath, "utf8"); + } catch { + /* drift */ + } + if (current !== output || currentIndex !== indexOutput) { + console.error("Realtime generated files are stale. Run npm run gen:realtime-types."); + process.exitCode = 1; + } +} else { + writeFileSync(outPath, output); + writeFileSync(indexPath, indexOutput); +} diff --git a/scripts/gen-realtime-types.test.ts b/scripts/gen-realtime-types.test.ts new file mode 100644 index 0000000..e8ed41c --- /dev/null +++ b/scripts/gen-realtime-types.test.ts @@ -0,0 +1,16 @@ +import { spawnSync } from "node:child_process"; +import { resolve } from "node:path"; +import { describe, expect, it } from "vitest"; + +describe("realtime type generator", () => { + it("fails loudly when the OpenAPI document omits a realtime schema", () => { + const script = resolve(process.cwd(), "scripts/gen-realtime-types.mjs"); + const fixture = resolve(process.cwd(), "scripts/fixtures/malformed-openapi.yaml"); + const result = spawnSync(process.execPath, [script], { + env: { ...process.env, REALTIME_OPENAPI_SPEC: fixture }, + encoding: "utf8", + }); + expect(result.status).not.toBe(0); + expect(result.stderr).toMatch(/Required realtime schema/); + }); +}); diff --git a/src/app/globals.css b/src/app/globals.css index e69de29..e935d7f 100644 --- a/src/app/globals.css +++ b/src/app/globals.css @@ -0,0 +1,9 @@ +/* Realtime freshness status stays static when reduced motion is enabled. */ +.data-freshness { display: inline-flex; align-items: center; gap: .4rem; min-width: 8rem; font-size: .75rem; line-height: 1.25rem; color: var(--pm-text-muted, #a1a1aa); } +.data-freshness__dot { width: .4rem; height: .4rem; flex: 0 0 .4rem; border-radius: 50%; background: #a1a1aa; } +.data-freshness--live { color: #34d399; } +.data-freshness--live .data-freshness__dot { background: #34d399; } +.data-freshness--muted .data-freshness__dot { background: #fbbf24; } +.data-freshness__dot--pulse { animation: freshness-pulse 1.8s ease-in-out infinite; } +@keyframes freshness-pulse { 50% { opacity: .4; transform: scale(.8); } } +@media (prefers-reduced-motion: reduce) { .data-freshness__dot--pulse { animation: none; } } diff --git a/src/app/page.tsx b/src/app/page.tsx index 38753ab..2a61b46 100644 --- a/src/app/page.tsx +++ b/src/app/page.tsx @@ -14,6 +14,7 @@ import { formatUsd, formatRelativeTime, fromStroops } from "@/lib/format"; import { truncateAddress } from "@/lib/wallet/WalletProvider"; import { ACTIVE_NETWORK } from "@/lib/network"; import { HowItWorks } from "@/components/HowItWorks"; +import { DataFreshness } from "@/components/DataFreshness"; function Counter({ to, prefix = "", suffix = "", decimals = 0 }: { to: number; prefix?: string; suffix?: string; decimals?: number }) { const [val, setVal] = useState(0); @@ -180,6 +181,7 @@ export default function Home() {
Live Oracle Status +
diff --git a/src/components/DataFreshness.tsx b/src/components/DataFreshness.tsx new file mode 100644 index 0000000..863956a --- /dev/null +++ b/src/components/DataFreshness.tsx @@ -0,0 +1,86 @@ +"use client"; + +import { useEffect, useRef, useState } from "react"; +import { usePrefersReducedMotion } from "@/hooks/usePrefersReducedMotion"; +import { getFreshness, type DataSource, type Freshness } from "@/lib/realtime/freshness"; +import { realtimeClient } from "@/lib/realtime/client"; + +export function DataFreshness({ + updatedAt, + source = "api", + reconnecting = false, + staleAfterMs = 30_000, +}: { + updatedAt?: number | null; + source?: DataSource; + reconnecting?: boolean; + staleAfterMs?: number; +}) { + const reduced = usePrefersReducedMotion(); + const [now, setNow] = useState(null); + const [connection, setConnection] = useState(realtimeClient.getConnectionState()); + const [everConnected, setEverConnected] = useState(false); + const reconnectStarted = useRef(null); + useEffect(() => { + const unsubscribe = realtimeClient.onConnectionState((state) => { + setConnection(state); + if (state === "connected") { + setEverConnected(true); + reconnectStarted.current = null; + } else if (reconnectStarted.current === null) reconnectStarted.current = Date.now(); + }); + setNow(Date.now()); + const timer = window.setInterval(() => setNow(Date.now()), 2_000); + return () => { + unsubscribe(); + window.clearInterval(timer); + }; + }, []); + const graceStarted = + everConnected && connection !== "connected" ? reconnectStarted.current : null; + const freshness: Freshness = + now === null ? "unknown" : getFreshness(updatedAt ?? null, now, staleAfterMs, graceStarted); + const withinReconnectGrace = + everConnected && + connection !== "connected" && + reconnectStarted.current !== null && + now !== null && + now - reconnectStarted.current < 5_000; + const displayedFreshness = freshness; + const age = updatedAt && now !== null ? Math.max(0, Math.floor((now - updatedAt) / 1000)) : null; + const label = reconnecting + ? "Reconnecting" + : source === "fixture" + ? "Offline data" + : source !== "realtime" + ? source === "chain" + ? "On-chain data" + : "REST data" + : displayedFreshness === "fresh" + ? "Live" + : displayedFreshness === "aging" + ? "Updating" + : displayedFreshness === "stale" + ? `Updated ${age}s ago` + : "Waiting for data"; + const tone = + reconnecting || + (everConnected && connection !== "connected" && !withinReconnectGrace) || + displayedFreshness === "stale" + ? "muted" + : displayedFreshness === "fresh" && source === "realtime" + ? "live" + : "quiet"; + return ( + + + ); +} diff --git a/src/hooks/useOracleStatus.ts b/src/hooks/useOracleStatus.ts index e69de29..b54837b 100644 --- a/src/hooks/useOracleStatus.ts +++ b/src/hooks/useOracleStatus.ts @@ -0,0 +1,70 @@ +"use client"; + +import { useEffect } from "react"; +import { useQuery, useQueryClient } from "@tanstack/react-query"; +import { fetchOracleStatus, type OracleReading } from "@/lib/api/oracle"; +import { FIXTURE_ORACLE_READINGS } from "@/lib/fixtures/oracle"; +import { queryFnWithFixtureFallback, type FixtureResult } from "@/lib/query/fixtureFallback"; +import { realtimeClient } from "@/lib/realtime/client"; + +export function useOracleStatus() { + const client = useQueryClient(); + const query = useQuery({ + queryKey: ["oracleStatus"], + queryFn: async ({ + signal, + }): Promise< + FixtureResult<{ readings: OracleReading[] }> & { source?: "api" | "fixture" | "realtime" } + > => { + const result = await queryFnWithFixtureFallback( + () => fetchOracleStatus(signal), + () => ({ readings: FIXTURE_ORACLE_READINGS }), + { signal, fallbackOnAnyError: true } + ); + return { ...result, source: result.isFixture ? "fixture" : "api" }; + }, + refetchInterval: 60_000, + }); + useEffect( + () => + realtimeClient.subscribe("oracle_update", (payload) => { + const reading = payload as OracleReading; + if (!reading || typeof reading !== "object" || typeof reading.coverageType !== "string") + return; + client.setQueryData( + ["oracleStatus"], + ( + current: + | { data: { readings: OracleReading[] }; isFixture: boolean; source?: string } + | undefined + ) => ({ + data: { + readings: [ + ...(current?.data.readings ?? []).filter( + (item) => item.coverageType !== reading.coverageType + ), + reading, + ], + }, + isFixture: false, + source: "realtime", + }) + ); + }), + [client] + ); + const result = query.data?.data; + return { + data: result?.readings ?? null, + loading: query.isLoading, + isFixture: query.data?.isFixture ?? false, + lastUpdated: query.dataUpdatedAt ? new Date(query.dataUpdatedAt) : undefined, + dataSource: query.data?.isFixture + ? ("fixture" as const) + : query.data?.source === "realtime" + ? ("realtime" as const) + : ("api" as const), + isRefreshing: query.isFetching, + refresh: query.refetch, + }; +} diff --git a/src/hooks/usePoolStats.ts b/src/hooks/usePoolStats.ts index e69de29..b98bc21 100644 --- a/src/hooks/usePoolStats.ts +++ b/src/hooks/usePoolStats.ts @@ -0,0 +1,72 @@ +"use client"; + +import { useEffect } from "react"; +import { useQuery, useQueryClient } from "@tanstack/react-query"; +import { fetchPoolStats, type PoolStats } from "@/lib/api/pool"; +import { FIXTURE_POOL_STATS } from "@/lib/fixtures/poolStats"; +import { ApiUnreachableError } from "@/lib/api/client"; +import { readPoolStats } from "@/lib/stellar/rpc"; +import { realtimeClient } from "@/lib/realtime/client"; + +export function usePoolStats() { + const client = useQueryClient(); + const query = useQuery({ + queryKey: ["poolStats"], + queryFn: async ({ signal }) => { + try { + return { + data: await fetchPoolStats(signal), + dataSource: "api" as const, + isFixture: false, + error: undefined, + }; + } catch (apiError) { + if (!(apiError instanceof ApiUnreachableError)) + return { + data: null, + dataSource: null, + isFixture: false, + error: apiError instanceof Error ? apiError.message : "Failed to load pool stats", + }; + try { + return { + data: await readPoolStats(), + dataSource: "chain" as const, + isFixture: false, + error: undefined, + }; + } catch { + return { + data: FIXTURE_POOL_STATS, + dataSource: "fixture" as const, + isFixture: true, + error: undefined, + }; + } + } + }, + refetchInterval: 60_000, + }); + useEffect( + () => + realtimeClient.subscribe("pool_update", (payload) => { + if (!payload || typeof payload !== "object") return; + client.setQueryData(["poolStats"], { + data: payload as PoolStats, + dataSource: "realtime", + isFixture: false, + }); + }), + [client] + ); + const fixture = query.data?.isFixture ?? false; + return { + data: query.data?.data ?? null, + loading: query.isLoading, + isFixture: fixture, + dataSource: query.data?.dataSource ?? null, + lastUpdated: query.dataUpdatedAt ? new Date(query.dataUpdatedAt) : undefined, + error: query.error?.message, + refreshFromChain: readPoolStats, + }; +} diff --git a/src/hooks/useRecentClaims.ts b/src/hooks/useRecentClaims.ts index e69de29..f6fdd6d 100644 --- a/src/hooks/useRecentClaims.ts +++ b/src/hooks/useRecentClaims.ts @@ -0,0 +1,53 @@ +"use client"; + +import { useEffect } from "react"; +import { useQuery, useQueryClient } from "@tanstack/react-query"; +import { fetchRecentClaims, type ClaimRecord } from "@/lib/api/claims"; +import { FIXTURE_RECENT_CLAIMS } from "@/lib/fixtures/recentClaims"; +import { queryFnWithFixtureFallback } from "@/lib/query/fixtureFallback"; +import { realtimeClient } from "@/lib/realtime/client"; + +export function useRecentClaims() { + const client = useQueryClient(); + const query = useQuery({ + queryKey: ["recentClaims"], + queryFn: ({ signal }) => + queryFnWithFixtureFallback( + () => fetchRecentClaims({ limit: 10 }, signal), + () => ({ items: FIXTURE_RECENT_CLAIMS, hasMore: false }), + { signal, fallbackOnAnyError: true } + ), + refetchInterval: 60_000, + }); + useEffect( + () => + realtimeClient.subscribe("claim_update", (payload) => { + const claim = payload as ClaimRecord; + if (!claim || typeof claim !== "object" || typeof claim.policyId !== "string") return; + client.setQueryData( + ["recentClaims"], + ( + current: + { data: { items: ClaimRecord[]; hasMore: boolean }; isFixture: boolean } | undefined + ) => ({ + data: { + items: [ + claim, + ...(current?.data.items ?? []).filter((item) => item.policyId !== claim.policyId), + ].slice(0, 10), + hasMore: current?.data.hasMore ?? false, + }, + isFixture: false, + }) + ); + }), + [client] + ); + return { + data: query.data?.data?.items ?? [], + loading: query.isLoading, + isFixture: query.data?.isFixture ?? false, + lastUpdated: query.dataUpdatedAt ? new Date(query.dataUpdatedAt) : undefined, + dataSource: query.data?.isFixture ? ("fixture" as const) : ("api" as const), + }; +} diff --git a/src/lib/api/claims.ts b/src/lib/api/claims.ts index ddbdd2f..5b63dcb 100644 --- a/src/lib/api/claims.ts +++ b/src/lib/api/claims.ts @@ -34,9 +34,12 @@ function buildPaginationQuery({ cursor, limit }: PaginationParams = {}): string * Normalizes a list response that may or may not include pagination metadata. * Falls back to the legacy `{ claims }` shape when metadata is absent. */ -function normalizeClaimsResponse( - response: { claims?: ClaimRecord[]; items?: ClaimRecord[]; nextCursor?: string; hasMore?: boolean }, -): PaginatedResponse { +function normalizeClaimsResponse(response: { + claims?: ClaimRecord[]; + items?: ClaimRecord[]; + nextCursor?: string; + hasMore?: boolean; +}): PaginatedResponse { const items = response.items ?? response.claims ?? []; const hasMore = response.hasMore ?? false; return { items, nextCursor: response.nextCursor, hasMore }; @@ -45,20 +48,28 @@ function normalizeClaimsResponse( export function fetchHolderClaims( address: string, params: PaginationParams = {}, - signal?: AbortSignal, + signal?: AbortSignal ): Promise> { - return apiRequest(`/claims/holder/${address}${buildPaginationQuery(params)}`, { signal }).then( - normalizeClaimsResponse, + return apiRequest<{ + claims?: ClaimRecord[]; + items?: ClaimRecord[]; + nextCursor?: string; + hasMore?: boolean; + }>(`/claims/holder/${address}${buildPaginationQuery(params)}`, { signal }).then( + normalizeClaimsResponse ); } export function fetchRecentClaims( params: PaginationParams = {}, - signal?: AbortSignal, + signal?: AbortSignal ): Promise> { - return apiRequest(`/claims/recent${buildPaginationQuery(params)}`, { signal }).then( - normalizeClaimsResponse, - ); + return apiRequest<{ + claims?: ClaimRecord[]; + items?: ClaimRecord[]; + nextCursor?: string; + hasMore?: boolean; + }>(`/claims/recent${buildPaginationQuery(params)}`, { signal }).then(normalizeClaimsResponse); } /** Types are generated from openapi/refract-api.yaml, which mirrors ClaimService.getStats()'s response in refract-backend. */ diff --git a/src/lib/api/oracle.ts b/src/lib/api/oracle.ts index c1b0eaa..e64d55d 100644 --- a/src/lib/api/oracle.ts +++ b/src/lib/api/oracle.ts @@ -1,8 +1,8 @@ import { apiRequest } from "./client"; -import type { ApiSchemas } from "./generated"; +import type { OracleReading } from "@/lib/realtime/generated/schemas"; /** Types are generated from openapi/refract-api.yaml, which mirrors OracleReading from refract-backend/src/oracle/oracle-reading.ts. */ -export type OracleReading = ApiSchemas["OracleReading"]; +export type { OracleReading }; export function fetchOracleStatus(signal?: AbortSignal): Promise<{ readings: OracleReading[] }> { return apiRequest("/oracle/status", { signal }); diff --git a/src/lib/realtime/client.test.ts b/src/lib/realtime/client.test.ts new file mode 100644 index 0000000..c776105 --- /dev/null +++ b/src/lib/realtime/client.test.ts @@ -0,0 +1,32 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { RealtimeClient } from "./client"; + +describe("RealtimeClient outbound batching", () => { + afterEach(() => vi.unstubAllGlobals()); + + it("writes twenty topic operations as one frame", () => { + const socket = { readyState: 1, send: vi.fn() }; + vi.stubGlobal("WebSocket", { OPEN: 1 }); + const client = new RealtimeClient() as any; + client.socket = socket; + for (let i = 0; i < 20; i++) client.enqueue(`topic-${i}`, "subscribe"); + client.flush(); + expect(socket.send).toHaveBeenCalledTimes(1); + expect(JSON.parse(socket.send.mock.calls[0]![0]).operations).toHaveLength(20); + }); + + it("elides subscribe then unsubscribe before the queued flush", () => { + const client = new RealtimeClient() as any; + client.enqueue("oracle_update", "subscribe"); + client.enqueue("oracle_update", "unsubscribe"); + expect(client.pending.size).toBe(0); + }); + + it("bounds queued topic operations and warns on overflow", () => { + const client = new RealtimeClient() as any; + const warning = vi.spyOn(console, "warn").mockImplementation(() => {}); + for (let i = 0; i < 501; i++) client.enqueue(`topic-${i}`, "subscribe"); + expect(client.pending.size).toBe(500); + expect(warning).toHaveBeenCalled(); + }); +}); diff --git a/src/lib/realtime/client.ts b/src/lib/realtime/client.ts index 8be8d3c..0accb86 100644 --- a/src/lib/realtime/client.ts +++ b/src/lib/realtime/client.ts @@ -1,33 +1,254 @@ -import { API_BASE_URL } from "@/lib/api/client"; +import { API_BASE_URL } from "../api/client"; +import { ClaimRecordSchema, OracleReadingSchema, PoolStatsSchema } from "./generated"; +import type { RealtimePayloads, RealtimeTopic } from "./types"; type Handler = (payload: unknown) => void; - +type ConnectionState = "connecting" | "connected" | "reconnecting" | "disconnected"; const MAX_BACKOFF_MS = 30_000; const BASE_BACKOFF_MS = 1_000; +const MAX_PENDING_TOPICS = 500; +const CHANNEL = `${process.env.NEXT_PUBLIC_APP_INSTANCE ?? "refract"}-realtime-v1`; +const LEASE_KEY = `${CHANNEL}-leader`; +const LEASE_MS = 4_000; +const HEARTBEAT_MS = 1_000; +type WireMessage = { topic: string; payload: unknown }; +type Operation = { topic: string; op: "subscribe" | "unsubscribe" }; +type TabEnvelope = { + version: 1; + instance: string; + sender: string; + kind: "event" | "state" | "control"; + seq: number; + topic?: string; + payload?: unknown; + state?: ConnectionState; + operations?: Operation[]; +}; -/** - * 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 { +/** Reconnecting topic client with a bounded outbound microtask queue and cross-tab fan-out. */ +export class RealtimeClient { private socket: WebSocket | null = null; private handlers = new Map>(); + private pending = new Map(); private attempt = 0; private closedByUser = false; private reconnectTimer: ReturnType | null = null; + private flushScheduled = false; + private channel: BroadcastChannel | null = null; + private tabId = `${Date.now()}-${Math.random().toString(36).slice(2)}`; + private leader = true; + private usesWebLocks = false; + private releaseLeaderLock: (() => void) | null = null; + private electionTimer: ReturnType | null = null; + private heartbeatTimer: ReturnType | null = null; + private sequence = 0; + private lastSequence = new Map(); + private followerTopics = new Map>(); + private state: ConnectionState = "disconnected"; + private stateHandlers = new Set<(state: ConnectionState) => void>(); connect(): void { - if (typeof WebSocket === "undefined" || this.socket) return; - this.closedByUser = false; - this.open(); + if (typeof WebSocket === "undefined" || this.closedByUser) return; + this.startTabCoordination(); + if (this.channel && !this.leader) return; + if (!this.socket) this.open(); + } + + private startTabCoordination(): void { + if ( + this.channel || + typeof BroadcastChannel === "undefined" || + typeof localStorage === "undefined" + ) + return; + try { + this.channel = new BroadcastChannel(CHANNEL); + this.channel.onmessage = (event: MessageEvent) => this.receiveTabMessage(event.data); + this.usesWebLocks = typeof navigator !== "undefined" && !!navigator.locks; + this.leader = this.usesWebLocks ? false : this.claimLeadership(); + if (this.usesWebLocks) void this.acquireLeaderLock(); + this.electionTimer = setInterval(() => { + const wasLeader = this.leader; + if (wasLeader) this.writeLease(); + else if (this.usesWebLocks) void this.acquireLeaderLock(); + else this.leader = this.claimLeadership(); + if (!wasLeader && this.leader) { + this.flush(); + this.open(); + this.resubscribeAll(); + } + if (wasLeader && !this.leader) this.socket?.close(); + }, HEARTBEAT_MS); + if (this.leader) this.writeLease(); + } catch { + this.channel?.close(); + this.channel = null; + this.leader = true; + } + } + + private acquireLeaderLock(): void { + if (!this.usesWebLocks || this.leader || this.closedByUser) return; + let acquired = false; + let release!: () => void; + const hold = new Promise((resolve) => { + release = resolve; + }); + void navigator.locks + .request(CHANNEL, { ifAvailable: true }, async (lock) => { + if (!lock) return; + acquired = true; + this.releaseLeaderLock = release; + this.leader = true; + this.writeLease(); + this.open(); + this.resubscribeAll(); + await hold; + }) + .then(() => { + if (!acquired) return; + this.releaseLeaderLock = null; + this.leader = false; + }) + .catch(() => { + this.usesWebLocks = false; + this.leader = this.claimLeadership(); + if (this.leader) this.open(); + }); + } + + private claimLeadership(): boolean { + try { + const lease = JSON.parse(localStorage.getItem(LEASE_KEY) || "null") as { + id?: string; + expires?: number; + } | null; + if (!lease || !lease.expires || lease.expires < Date.now() || lease.id === this.tabId) { + this.writeLease(); + return true; + } + return false; + } catch { + return true; + } + } + + private writeLease(): void { + try { + localStorage.setItem( + LEASE_KEY, + JSON.stringify({ id: this.tabId, expires: Date.now() + LEASE_MS }) + ); + } catch { + /* storage may be disabled */ + } + } + + private receiveTabMessage(raw: unknown): void { + if (!raw || typeof raw !== "object") { + console.warn("Ignored malformed realtime tab message"); + return; + } + const msg = raw as Partial; + if ( + msg.version !== 1 || + msg.instance !== CHANNEL || + typeof msg.sender !== "string" || + !Number.isSafeInteger(msg.seq) || + (msg.kind !== "event" && msg.kind !== "state" && msg.kind !== "control") + ) { + console.warn("Ignored malformed realtime tab message"); + return; + } + if (msg.sender === this.tabId || (this.lastSequence.get(msg.sender!) ?? 0) >= msg.seq!) return; + try { + if (JSON.stringify(raw).length > 65_536) { + console.warn("Ignored oversized realtime tab message"); + return; + } + } catch { + return; + } + if (msg.kind === "event" && (typeof msg.topic !== "string" || !Object.hasOwn(msg, "payload"))) { + console.warn("Ignored malformed realtime tab message"); + return; + } + if ( + msg.kind === "state" && + !["connecting", "connected", "reconnecting", "disconnected"].includes(String(msg.state)) + ) { + console.warn("Ignored malformed realtime tab message"); + return; + } + this.lastSequence.set(msg.sender!, msg.seq!); + if ( + msg.kind === "control" && + (!Array.isArray(msg.operations) || + msg.operations.length > MAX_PENDING_TOPICS || + !msg.operations.every( + (op) => + op && typeof op.topic === "string" && (op.op === "subscribe" || op.op === "unsubscribe") + )) + ) { + console.warn("Ignored malformed realtime tab message"); + return; + } + if (msg.kind === "event" && typeof msg.topic === "string") + this.dispatch(msg.topic, msg.payload); + if ( + msg.kind === "state" && + ["connecting", "connected", "reconnecting", "disconnected"].includes(String(msg.state)) + ) { + this.setState(msg.state as ConnectionState, false); + if (msg.state === "connecting" && !this.leader) this.resubscribeAll(); + } + if ( + msg.kind === "control" && + this.leader && + Array.isArray(msg.operations) && + msg.operations.length <= MAX_PENDING_TOPICS && + msg.operations.every( + (op) => + op && typeof op.topic === "string" && (op.op === "subscribe" || op.op === "unsubscribe") + ) + ) { + const topics = this.followerTopics.get(msg.sender!) ?? new Set(); + for (const { topic, op } of msg.operations) { + const previouslyNeeded = this.handlers.get(topic)?.size + ? true + : [...this.followerTopics.values()].some((set) => set.has(topic)); + if (op === "subscribe") topics.add(topic); + else topics.delete(topic); + const nowNeeded = this.handlers.get(topic)?.size + ? true + : [...this.followerTopics.values()].some((set) => set.has(topic)) || topics.has(topic); + if (previouslyNeeded !== nowNeeded) + this.enqueue(topic, nowNeeded ? "subscribe" : "unsubscribe"); + } + if (topics.size) this.followerTopics.set(msg.sender!, topics); + else this.followerTopics.delete(msg.sender!); + this.flush(); + } + } + + private broadcast(message: Omit): void { + if (!this.channel) return; + try { + this.channel.postMessage({ + ...message, + version: 1, + instance: CHANNEL, + sender: this.tabId, + seq: ++this.sequence, + } satisfies TabEnvelope); + } catch { + /* oversized/uncloneable payload */ + } } private open(): void { + if (this.socket || (this.channel && !this.leader)) return; + this.setState(this.attempt ? "reconnecting" : "connecting"); const wsUrl = API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws"); try { this.socket = new WebSocket(wsUrl); @@ -35,25 +256,58 @@ class RealtimeClient { this.scheduleReconnect(); return; } - this.socket.addEventListener("open", () => { this.attempt = 0; + this.setState("connected"); + this.flush(); + this.resubscribeAll(); }); this.socket.addEventListener("message", (event) => { try { - const { topic, payload } = JSON.parse(event.data as string) as { topic: string; payload: unknown }; - this.handlers.get(topic)?.forEach((handler) => handler(payload)); + const msg = JSON.parse(event.data as string) as WireMessage; + if (typeof msg.topic !== "string" || !Object.hasOwn(msg, "payload")) return; + this.dispatch(msg.topic, msg.payload); + this.broadcast({ kind: "event", topic: msg.topic, payload: msg.payload }); } catch { - // Malformed message — ignore, don't crash the connection. + /* Ignore malformed server messages. */ } }); this.socket.addEventListener("close", () => { this.socket = null; - if (!this.closedByUser) this.scheduleReconnect(); - }); - this.socket.addEventListener("error", () => { - this.socket?.close(); + if (!this.closedByUser) { + this.setState("reconnecting"); + this.scheduleReconnect(); + } else this.setState("disconnected"); }); + this.socket.addEventListener("error", () => this.socket?.close()); + } + + private dispatch(topic: string, payload: unknown): void { + const schemas = { + oracle_update: OracleReadingSchema, + claim_update: ClaimRecordSchema, + pool_update: PoolStatsSchema, + }; + if (!Object.hasOwn(schemas, topic)) return; + const result = schemas[topic as RealtimeTopic].safeParse(payload); + if (!result.success) { + console.warn(`Ignored malformed realtime payload for ${topic}`); + return; + } + this.handlers.get(topic)?.forEach((handler) => handler(result.data)); + } + private setState(state: ConnectionState, fanout = true): void { + this.state = state; + this.stateHandlers.forEach((handler) => handler(state)); + if (fanout) this.broadcast({ kind: "state", state }); + } + getConnectionState(): ConnectionState { + return this.state; + } + onConnectionState(handler: (state: ConnectionState) => void): () => void { + this.stateHandlers.add(handler); + handler(this.state); + return () => this.stateHandlers.delete(handler); } private scheduleReconnect(): void { @@ -66,21 +320,106 @@ class RealtimeClient { }, delay); } - subscribe(topic: string, handler: Handler): () => void { + private enqueue(topic: string, operation: "subscribe" | "unsubscribe"): void { + const previous = this.pending.get(topic); + if (previous === "subscribe" && operation === "unsubscribe") { + this.pending.delete(topic); + return; + } + if (!this.pending.has(topic) && this.pending.size >= MAX_PENDING_TOPICS) { + const oldest = this.pending.keys().next().value as string | undefined; + if (oldest) this.pending.delete(oldest); + console.warn("Realtime outbound queue full; dropped oldest topic operation"); + } + this.pending.delete(topic); + this.pending.set(topic, operation); + if (!this.flushScheduled) { + this.flushScheduled = true; + queueMicrotask(() => { + this.flushScheduled = false; + this.flush(); + }); + } + } + + /** Flush queued subscription operations in a single frame; safe to call before disconnect. */ + flush(): void { + if (!this.pending.size) return; + const operations = [...this.pending].map(([topic, op]) => ({ topic, op })); + if (this.channel && !this.leader) { + this.pending.clear(); + this.broadcast({ kind: "control", operations }); + return; + } + if (!this.socket || this.socket.readyState !== WebSocket.OPEN) return; + this.pending.clear(); + try { + this.socket.send(JSON.stringify({ type: "batch", operations })); + } catch { + operations.forEach(({ topic, op }) => { + if (!this.pending.has(topic)) this.pending.set(topic, op); + }); + } + } + + /** Send boundary-sensitive frames (for example authentication) immediately. */ + sendImmediately(frame: unknown): void { + this.flush(); + if (this.socket?.readyState === WebSocket.OPEN) this.socket.send(JSON.stringify(frame)); + } + + private resubscribeAll(): void { + this.handlers.forEach((handlers, topic) => { + if (handlers.size) this.enqueue(topic, "subscribe"); + }); + if (this.leader) + this.followerTopics.forEach((topics) => + topics.forEach((topic) => this.enqueue(topic, "subscribe")) + ); + this.flush(); + } + + subscribe( + topic: Topic, + handler: (payload: RealtimePayloads[Topic]) => void + ): () => void { if (!this.handlers.has(topic)) this.handlers.set(topic, new Set()); - this.handlers.get(topic)!.add(handler); + const handlers = this.handlers.get(topic)!; + if (handlers.size === 0 && ![...this.followerTopics.values()].some((set) => set.has(topic))) + this.enqueue(topic, "subscribe"); + handlers.add(handler as Handler); this.connect(); - return () => { - this.handlers.get(topic)?.delete(handler); + const current = this.handlers.get(topic); + current?.delete(handler as Handler); + if (current?.size === 0) { + this.handlers.delete(topic); + if (![...this.followerTopics.values()].some((set) => set.has(topic))) + this.enqueue(topic, "unsubscribe"); + } }; } disconnect(): void { + this.flush(); this.closedByUser = true; if (this.reconnectTimer) clearTimeout(this.reconnectTimer); + if (this.electionTimer) clearInterval(this.electionTimer); + if (this.heartbeatTimer) clearInterval(this.heartbeatTimer); + if (this.leader) { + try { + localStorage.removeItem(LEASE_KEY); + } catch { + /* ignore */ + } + } + this.releaseLeaderLock?.(); + this.releaseLeaderLock = null; + this.channel?.close(); + this.channel = null; this.socket?.close(); this.socket = null; + this.setState("disconnected"); } } diff --git a/src/lib/realtime/freshness.test.ts b/src/lib/realtime/freshness.test.ts new file mode 100644 index 0000000..12ce94b --- /dev/null +++ b/src/lib/realtime/freshness.test.ts @@ -0,0 +1,21 @@ +import { describe, expect, it } from "vitest"; +import { getFreshness } from "./freshness"; + +describe("realtime freshness thresholds", () => { + it("uses fresh, aging, and stale boundaries consistently", () => { + expect(getFreshness(70_000, 100_000, 30_000)).toBe("fresh"); + expect(getFreshness(69_999, 100_000, 30_000)).toBe("aging"); + expect(getFreshness(40_000, 100_000, 30_000)).toBe("aging"); + expect(getFreshness(39_999, 100_000, 30_000)).toBe("stale"); + }); + + it("handles absent timestamps and future clock skew", () => { + expect(getFreshness(null, 100_000)).toBe("unknown"); + expect(getFreshness(101_000, 100_000)).toBe("fresh"); + }); + + it("holds the previous non-stale state through brief reconnect churn", () => { + expect(getFreshness(0, 65_000, 30_000, 62_000)).toBe("aging"); + expect(getFreshness(0, 67_000, 30_000, 62_000)).toBe("stale"); + }); +}); diff --git a/src/lib/realtime/freshness.ts b/src/lib/realtime/freshness.ts new file mode 100644 index 0000000..4f6263e --- /dev/null +++ b/src/lib/realtime/freshness.ts @@ -0,0 +1,16 @@ +export type DataSource = "realtime" | "api" | "chain" | "fixture"; +export type Freshness = "fresh" | "aging" | "stale" | "unknown"; + +export function getFreshness( + timestamp: number | null, + now: number, + windowMs = 30_000, + reconnectStartedAt: number | null = null +): Freshness { + if (timestamp === null || !Number.isFinite(timestamp)) return "unknown"; + const age = Math.max(0, now - timestamp); + if (age <= windowMs) return "fresh"; + if (age <= windowMs * 2 || (reconnectStartedAt !== null && now - reconnectStartedAt < 5_000)) + return "aging"; + return "stale"; +} diff --git a/src/lib/realtime/generated/index.ts b/src/lib/realtime/generated/index.ts new file mode 100644 index 0000000..7024eb2 --- /dev/null +++ b/src/lib/realtime/generated/index.ts @@ -0,0 +1,3 @@ +// Generated by scripts/gen-realtime-types.mjs. Do not edit. +export { OracleReadingSchema, ClaimRecordSchema, PoolStatsSchema, RealtimeEnvelopeSchema } from "./schemas"; +export type { OracleReading, ClaimRecord, PoolStats, RealtimeEnvelope } from "./schemas"; diff --git a/src/lib/realtime/generated/schemas.test.ts b/src/lib/realtime/generated/schemas.test.ts new file mode 100644 index 0000000..fa9df21 --- /dev/null +++ b/src/lib/realtime/generated/schemas.test.ts @@ -0,0 +1,44 @@ +import { describe, expect, it } from "vitest"; +import { ClaimRecordSchema, OracleReadingSchema, RealtimeEnvelopeSchema } from "./schemas"; + +describe("generated realtime validators", () => { + it("accepts the optional settlement hash and rejects malformed readings", () => { + expect( + ClaimRecordSchema.safeParse({ + policyId: "p1", + holder: "G...", + coverageType: 1, + triggered: true, + payout: "100", + reason: "triggered", + processedAt: 1, + }).success + ).toBe(true); + expect( + OracleReadingSchema.safeParse({ + coverageType: "StablecoinDepeg", + type: "oracle_update", + value: 1, + threshold: 0.95, + severity: "invalid", + message: "bad", + }).success + ).toBe(false); + }); + + it("preserves optional, nullable, and nested envelope fields", () => { + expect( + RealtimeEnvelopeSchema.safeParse({ + topic: "oracle_update", + payload: { value: 1 }, + metadata: null, + }).success + ).toBe(true); + expect( + RealtimeEnvelopeSchema.safeParse({ + topic: "oracle_update", + payload: { nested: { value: 1 } }, + }).success + ).toBe(true); + }); +}); diff --git a/src/lib/realtime/generated/schemas.ts b/src/lib/realtime/generated/schemas.ts new file mode 100644 index 0000000..4a765fc --- /dev/null +++ b/src/lib/realtime/generated/schemas.ts @@ -0,0 +1,74 @@ +// Generated by scripts/gen-realtime-types.mjs. Do not edit. +import { z } from "zod"; + +export interface OracleReading { + coverageType: string; + message: string; + severity: "low" | "medium" | "high" | "triggered"; + threshold: number; + type: "oracle_update"; + value: number; +} +export const OracleReadingSchema = z.object({ + coverageType: z.string(), + message: z.string(), + severity: z.enum(["low", "medium", "high", "triggered"]), + threshold: z.number(), + type: z.enum(["oracle_update"]), + value: z.number(), +}); + +export interface ClaimRecord { + coverageType: number; + holder: string; + payout: string; + policyId: string; + processedAt: number; + reason: string; + settlementTxHash?: string; + triggered: boolean; +} +export const ClaimRecordSchema = z.object({ + coverageType: z.number(), + holder: z.string(), + payout: z.string(), + policyId: z.string(), + processedAt: z.number(), + reason: z.string(), + settlementTxHash: z.string().optional(), + triggered: z.boolean(), +}); + +export interface PoolStats { + apyBps: number; + availableUsdc: string; + lockedUsdc: string; + maxUtilizationBps: number; + premiumAccrued: string; + sharePrice: number; + totalShares: string; + totalUsdc: string; + utilizationBps: number; +} +export const PoolStatsSchema = z.object({ + apyBps: z.number(), + availableUsdc: z.string(), + lockedUsdc: z.string(), + maxUtilizationBps: z.number(), + premiumAccrued: z.string(), + sharePrice: z.number(), + totalShares: z.string(), + totalUsdc: z.string(), + utilizationBps: z.number(), +}); + +export interface RealtimeEnvelope { + metadata?: Record | null; + payload: Record; + topic: string; +} +export const RealtimeEnvelopeSchema = z.object({ + metadata: z.record(z.string(), z.unknown()).nullable().optional(), + payload: z.record(z.string(), z.unknown()), + topic: z.string(), +}); diff --git a/src/lib/realtime/types.ts b/src/lib/realtime/types.ts new file mode 100644 index 0000000..9ac54b5 --- /dev/null +++ b/src/lib/realtime/types.ts @@ -0,0 +1,9 @@ +import type { ClaimRecord, OracleReading, PoolStats } from "./generated"; + +export interface RealtimePayloads { + oracle_update: OracleReading; + claim_update: ClaimRecord; + pool_update: PoolStats; +} + +export type RealtimeTopic = keyof RealtimePayloads;