From 1aa38edb8d19119bf7ae2ab2aedbaac7cd548e14 Mon Sep 17 00:00:00 2001 From: faith-ng <332940157+faith-ng@users.noreply.github.com> Date: Wed, 21 Oct 2026 15:28:58 +0000 Subject: [PATCH 1/2] feat: add realtime freshness and redaction --- README.md | 4 + docs/realtime-freshness.md | 24 +++ src/app/dashboard/page.tsx | 6 +- src/app/page.tsx | 112 ++++++----- src/components/GlobalSearch.tsx | 2 +- src/hooks/useClaims.ts | 40 ++++ src/hooks/useNotifications.ts | 2 +- src/hooks/useOracleStatus.ts | 45 +++++ src/hooks/useProtocolStats.ts | 62 ++++++ src/hooks/useRecentClaims.ts | 39 ++++ src/lib/api/claims.ts | 31 ++- src/lib/flags/flagsConfig.ts | 3 + src/lib/realtime/client.ts | 60 ++++-- src/lib/realtime/freshness.ts | 15 ++ src/lib/realtime/redact.test.ts | 38 ++++ src/lib/realtime/redact.ts | 58 ++++++ src/lib/realtime/useRealtimeResource.test.ts | 96 +++++++++ src/lib/realtime/useRealtimeResource.ts | 201 +++++++++++++++++++ 18 files changed, 751 insertions(+), 87 deletions(-) create mode 100644 docs/realtime-freshness.md create mode 100644 src/hooks/useProtocolStats.ts create mode 100644 src/lib/realtime/freshness.ts create mode 100644 src/lib/realtime/redact.test.ts create mode 100644 src/lib/realtime/redact.ts create mode 100644 src/lib/realtime/useRealtimeResource.test.ts create mode 100644 src/lib/realtime/useRealtimeResource.ts diff --git a/README.md b/README.md index 780aba7..3d103d9 100644 --- a/README.md +++ b/README.md @@ -63,3 +63,7 @@ Other environment variables: ## License [MIT](./LICENSE) + +## Data freshness + +Landing protocol statistics, oracle status, and payout lists seed client state from REST before showing headline values, then accept feature-flagged realtime updates with a single REST fallback and explicit refresh where provided. Realtime topic flags default off and remain kill switches during the production observation window. There is no interval polling in these data hooks; the landing page's seconds-ago display timer only updates text. See [docs/realtime-freshness.md](docs/realtime-freshness.md) for the polling inventory, topic rollout gates, ordering rule, and safe debug logging switch. diff --git a/docs/realtime-freshness.md b/docs/realtime-freshness.md new file mode 100644 index 0000000..3c95f43 --- /dev/null +++ b/docs/realtime-freshness.md @@ -0,0 +1,24 @@ +# Realtime freshness and polling retirement + +Realtime is opt-in per topic through the remotely supplied feature flags. Each enabled topic seeds client state from one REST request, applies ordered realtime events through `useRealtimeResource`, and keeps an explicit refresh action where the UI exposes one. REST is the bounded fallback when the transport is unavailable; there are no interval polls in the data hooks. + +## Legacy path inventory and disposition + +| Path | Current state | Disposition | +| -------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------- | +| `useOracleStatus` interval/backoff/manual refresh | The hook was empty on the upstream baseline; no data interval or polling backoff existed. The landing page's one-second interval only updates the “seconds ago” label. | Keep manual refresh; use realtime topic `oracle.status` with one REST seed/fallback. Retain the display clock because it does not fetch data. | +| `useRecentClaims` polling constants/controls | The hook was empty; no polling constants existed. | Use `claims.recent`, one REST seed/fallback, and the landing payouts refresh action. | +| `useClaims` holder polling | The hook was empty; no polling constants existed. | Use `claims.holder.
` when its kill switch is enabled, one REST seed/fallback, and an explicit dashboard refresh. | +| `useNotifications` interval/backoff/manual refresh | No notification-specific polling existed; it derives from shared hooks. | Keep derived notifications and inherit each source's freshness policy. | +| Shared interval/backoff helpers in `src/lib/` | No data polling helper was present. WebSocket reconnect backoff remains transport recovery, not data polling. | Keep reconnect backoff; do not hand it to adaptive polling. | +| `/` seconds-ago clock | A one-second UI timer updates elapsed-time text only. | Keep; it does not request data or write cache state. | + +Topic flags default off and act as kill switches. The retirement is staged: no legacy poll was deleted because none existed in the upstream baseline. For each topic, keep the flag off until production has met the team's agreed observation window; then enable the topic, observe fallback/error rates, and remove any subsequently introduced legacy interval only after the window. The observation-window duration is an operational decision and should be recorded with the rollout. + +## Ordering rule + +The transport accepts an optional numeric `sequence` and numeric or ISO `timestamp` on each envelope. Higher sequence wins when both updates carry one; otherwise a strictly newer timestamp wins when both carry one. Equal or missing ordering keys favor the already-applied realtime update. REST responses started before a realtime event are aborted and also rejected by a generation guard if cancellation races completion. Comparable newer REST snapshots may advance state during a degraded connection. Cancellation is an optimization; the freshness checks remain authoritative. + +## Logging + +Realtime diagnostics pass through `src/lib/realtime/redact.ts`. Sensitive keys (auth tokens, wallet/address/holder identifiers, signatures, session credentials) are redacted recursively, arrays and objects are bounded, error messages are omitted, and unknown strings are truncated. Verbose message summaries are off in production and can be temporarily enabled in a development tab with `sessionStorage.setItem("refract:realtime-debug", "1")`; the switch expires when the tab's session ends. Only topic-safe summaries, byte counts, and error names are logged. diff --git a/src/app/dashboard/page.tsx b/src/app/dashboard/page.tsx index 6bcaa5d..4aeb855 100644 --- a/src/app/dashboard/page.tsx +++ b/src/app/dashboard/page.tsx @@ -31,7 +31,8 @@ export default function DashboardPage() { const wallet = useWallet(); const address = wallet.status === "connected" ? wallet.address : null; const { data: policies, loading, error, isFixture, isStale } = useHolderPolicies(address); - const claims = useClaims(address, policies); + const claimsResource = useClaims(address, policies); + const claims = claimsResource.data; const policiesById = normalizePolicies(policies ?? []); const claimsByPolicyId = normalizeClaims(claims); const [exporting, setExporting] = useState(false); @@ -93,6 +94,9 @@ export default function DashboardPage() { {canExport && (
+
{/* Stats */} -
- {[ - { - val: poolStats.data ? fromStroops(poolStats.data.totalUsdc) : 0, - prefix: "$", - suffix: "", - label: "Total Value Protected", - dec: 0, - }, - { - val: poolStats.data ? poolStats.data.apyBps / 100 : 0, - prefix: "", - suffix: "% APY", - label: "Capital Provider Returns", - dec: 1, - }, - { - val: claimStats.data?.activePolicies ?? 0, - prefix: "", - suffix: "", - label: "Policies Active", - dec: 0, - }, - ].map((s) => ( -
-
- {statsLoading ? ( - - ) : ( - // Counter only animates once on mount, so it's keyed on - // loaded-state to remount (and pick up the real value) - // the moment the data actually arrives. - - )} +
+
+ +
+
+ {[ + { + val: protocolStats.data ? fromStroops(protocolStats.data.tvl) : 0, + prefix: "$", + suffix: "", + label: "Total Value Protected", + dec: 0, + payout: false, + }, + { + val: 0, + prefix: "", + suffix: "", + label: "Cumulative Payouts", + dec: 0, + payout: true, + }, + { + val: protocolStats.data?.activePolicies ?? 0, + prefix: "", + suffix: "", + label: "Policies Active", + dec: 0, + payout: false, + }, + ].map((s) => ( +
+
+ {statsLoading ? ( + + ) : ( + + )} +
+
{s.label}
-
{s.label}
-
- ))} + ))} +
@@ -246,7 +249,7 @@ export default function Home() { )}
- {oracleStatus.isFixture ? "Example data — Refract API unreachable" : "Oracles auto-refreshed every 60 seconds"} + {oracleStatus.isFixture ? "Example data — Refract API unreachable" : "Realtime updates with REST fallback"}
@@ -347,9 +350,10 @@ export default function Home() { {/* Recent payouts */}
-

- Recent payouts -

+
+

Recent payouts

+ +
{recentClaims.isFixture && (

⚠ Showing example data — the Refract API isn't reachable from this environment. diff --git a/src/components/GlobalSearch.tsx b/src/components/GlobalSearch.tsx index 082effd..95fe88d 100644 --- a/src/components/GlobalSearch.tsx +++ b/src/components/GlobalSearch.tsx @@ -74,7 +74,7 @@ function SearchDialog({ onClose }: { onClose: () => void }) { const wallet = useWallet(); const address = wallet.status === "connected" ? wallet.address : null; const { data: policies } = useHolderPolicies(address); - const claims = useClaims(address, policies); + const { data: claims } = useClaims(address, policies); const { data: coverageTypes } = useCoverageTypes(); const [query, setQuery] = useState(""); const [active, setActive] = useState(0); diff --git a/src/hooks/useClaims.ts b/src/hooks/useClaims.ts index e69de29..4e18a8c 100644 --- a/src/hooks/useClaims.ts +++ b/src/hooks/useClaims.ts @@ -0,0 +1,40 @@ +"use client"; +import { useCallback } from "react"; +import { fetchHolderClaims, type ClaimRecord } from "@/lib/api/claims"; +import type { Policy } from "@/lib/api/policies"; +import { fixtureClaimsForHolder } from "@/lib/fixtures/claims"; +import { useRealtimeResource } from "@/lib/realtime/useRealtimeResource"; + +export function useClaims(address?: string | null, policies?: Policy[] | null) { + const fetcher = useCallback( + async (signal: AbortSignal) => + address ? (await fetchHolderClaims(address, { limit: 100 }, signal)).items : [], + [address] + ); + const fixture = useCallback( + () => (address && policies ? fixtureClaimsForHolder(address, policies) : []), + [address, policies] + ); + const resource = useRealtimeResource({ + topic: address ? `claims.holder.${address}` : "claims.holder", + flag: "realtime-holder-claims", + fetcher, + fixture, + mergeEvent: (payload, current) => { + if (Array.isArray(payload)) return payload as ClaimRecord[]; + if (!payload || typeof payload !== "object") return null; + if (Array.isArray((payload as { items?: unknown }).items)) + return (payload as { items: ClaimRecord[] }).items; + if (!("policyId" in payload)) return null; + const claim = payload as ClaimRecord; + return [claim, ...(current ?? []).filter((item) => item.policyId !== claim.policyId)]; + }, + }); + return { + data: address ? (resource.data ?? []) : [], + loading: resource.loading, + isRefreshing: resource.isRefreshing, + isFixture: resource.isFixture, + refresh: resource.refresh, + }; +} diff --git a/src/hooks/useNotifications.ts b/src/hooks/useNotifications.ts index e8fe606..98acd39 100644 --- a/src/hooks/useNotifications.ts +++ b/src/hooks/useNotifications.ts @@ -18,7 +18,7 @@ export function useNotifications() { const address = wallet.status === "connected" ? wallet.address : null; const { data: policies } = useHolderPolicies(address); - const claims = useClaims(address, policies); + const { data: claims } = useClaims(address, policies); const { lockupExpiresAt } = useLockupStatus(address); const { data: oracleReadings } = useOracleStatus(); diff --git a/src/hooks/useOracleStatus.ts b/src/hooks/useOracleStatus.ts index e69de29..6f07e83 100644 --- a/src/hooks/useOracleStatus.ts +++ b/src/hooks/useOracleStatus.ts @@ -0,0 +1,45 @@ +"use client"; +import { useCallback } from "react"; +import { fetchOracleStatus, type OracleReading } from "@/lib/api/oracle"; +import { FIXTURE_ORACLE_READINGS } from "@/lib/fixtures/oracle"; +import { useRealtimeResource } from "@/lib/realtime/useRealtimeResource"; + +export function useOracleStatus() { + const fetcher = useCallback( + async (signal: AbortSignal) => (await fetchOracleStatus(signal)).readings, + [] + ); + const resource = useRealtimeResource({ + topic: "oracle.status", + flag: "realtime-oracle", + fetcher, + fixture: () => FIXTURE_ORACLE_READINGS, + mergeEvent: (payload, current) => { + const reading = + payload && typeof payload === "object" && "coverageType" in payload && "value" in payload + ? (payload as OracleReading) + : null; + if (Array.isArray(payload)) return payload as OracleReading[]; + if ( + payload && + typeof payload === "object" && + Array.isArray((payload as { readings?: unknown }).readings) + ) + return (payload as { readings: OracleReading[] }).readings; + if (!reading) return null; + return [ + ...(current ?? []).filter((item) => item.coverageType !== reading.coverageType), + reading, + ]; + }, + }); + return { + data: resource.data, + loading: resource.loading, + error: resource.error, + isFixture: resource.isFixture, + lastUpdated: resource.lastUpdated, + isRefreshing: resource.isRefreshing, + refresh: resource.refresh, + }; +} diff --git a/src/hooks/useProtocolStats.ts b/src/hooks/useProtocolStats.ts new file mode 100644 index 0000000..2e134a2 --- /dev/null +++ b/src/hooks/useProtocolStats.ts @@ -0,0 +1,62 @@ +"use client"; +import { useCallback } from "react"; +import { fetchPoolStats } from "@/lib/api/pool"; +import { fetchClaimStats } from "@/lib/api/claims"; +import { FIXTURE_POOL_STATS } from "@/lib/fixtures/poolStats"; +import { FIXTURE_CLAIM_STATS } from "@/lib/fixtures/claimStats"; +import { useRealtimeResource } from "@/lib/realtime/useRealtimeResource"; + +export interface ProtocolStats { + tvl: string; + activePolicies: number; + cumulativePayouts: string; + asOf?: string; + sequence?: number; +} +export function useProtocolStats() { + const fetcher = useCallback(async (signal: AbortSignal): Promise => { + const [pool, claims] = await Promise.all([fetchPoolStats(signal), fetchClaimStats(signal)]); + return { + tvl: pool.totalUsdc, + activePolicies: claims.activePolicies, + cumulativePayouts: claims.totalPayout, + }; + }, []); + const fixture = useCallback( + (): ProtocolStats => ({ + tvl: FIXTURE_POOL_STATS.totalUsdc, + activePolicies: FIXTURE_CLAIM_STATS.activePolicies, + cumulativePayouts: FIXTURE_CLAIM_STATS.totalPayout, + }), + [] + ); + const resource = useRealtimeResource({ + topic: "protocol.stats", + flag: "realtime-protocol-stats", + fetcher, + fixture, + mergeEvent: (payload, current) => { + if (!payload || typeof payload !== "object" || !current) return null; + const delta = payload as Partial; + if ( + delta.tvl === undefined && + delta.activePolicies === undefined && + delta.cumulativePayouts === undefined + ) + return null; + return { ...current, ...delta }; + }, + getRestKey: (data) => ({ + sequence: data.sequence, + timestamp: data.asOf ? Date.parse(data.asOf) : undefined, + }), + }); + return { + data: resource.data, + loading: resource.loading, + error: resource.error, + isFixture: resource.isFixture, + isRefreshing: resource.isRefreshing, + refresh: resource.refresh, + }; +} diff --git a/src/hooks/useRecentClaims.ts b/src/hooks/useRecentClaims.ts index e69de29..ee3fd8e 100644 --- a/src/hooks/useRecentClaims.ts +++ b/src/hooks/useRecentClaims.ts @@ -0,0 +1,39 @@ +"use client"; +import { useCallback } from "react"; +import { fetchRecentClaims, type ClaimRecord } from "@/lib/api/claims"; +import { FIXTURE_RECENT_CLAIMS } from "@/lib/fixtures/recentClaims"; +import { useRealtimeResource } from "@/lib/realtime/useRealtimeResource"; + +export function useRecentClaims() { + const fetcher = useCallback( + async (signal: AbortSignal) => (await fetchRecentClaims({ limit: 12 }, signal)).items, + [] + ); + const resource = useRealtimeResource({ + topic: "claims.recent", + flag: "realtime-recent-claims", + fetcher, + fixture: () => FIXTURE_RECENT_CLAIMS, + mergeEvent: (payload, current) => { + if (Array.isArray(payload)) return payload as ClaimRecord[]; + if (!payload || typeof payload !== "object") return null; + if (Array.isArray((payload as { items?: unknown }).items)) + return (payload as { items: ClaimRecord[] }).items; + if (!("policyId" in payload)) return null; + const claim = payload as ClaimRecord; + return [claim, ...(current ?? []).filter((item) => item.policyId !== claim.policyId)].slice( + 0, + 12 + ); + }, + }); + return { + data: resource.data, + loading: resource.loading, + error: resource.error, + isFixture: resource.isFixture, + lastUpdated: resource.lastUpdated, + isRefreshing: resource.isRefreshing, + refresh: resource.refresh, + }; +} 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/flags/flagsConfig.ts b/src/lib/flags/flagsConfig.ts index 2ecb795..cb2cfbf 100644 --- a/src/lib/flags/flagsConfig.ts +++ b/src/lib/flags/flagsConfig.ts @@ -5,6 +5,9 @@ export interface FlagsDocument { /** Safe default if the remote flags source is unreachable — new/risky features default off. */ export const DEFAULT_FLAGS: Record = { "realtime-oracle": false, + "realtime-recent-claims": false, + "realtime-holder-claims": false, + "realtime-protocol-stats": false, }; export const FLAGS_ENDPOINT = "/flags.json"; diff --git a/src/lib/realtime/client.ts b/src/lib/realtime/client.ts index 8be8d3c..eba2461 100644 --- a/src/lib/realtime/client.ts +++ b/src/lib/realtime/client.ts @@ -1,19 +1,17 @@ import { API_BASE_URL } from "@/lib/api/client"; +import { realtimeLogger, summarizeRealtimeValue } from "./redact"; -type Handler = (payload: unknown) => void; +export interface RealtimeUpdate { + payload: T; + sequence?: number; + timestamp?: number; +} +type Handler = (update: RealtimeUpdate) => void; 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. - */ +/** Best-effort WebSocket transport. REST remains responsible for initial data and fallback. */ class RealtimeClient { private socket: WebSocket | null = null; private handlers = new Map>(); @@ -31,29 +29,52 @@ class RealtimeClient { const wsUrl = API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws"); try { this.socket = new WebSocket(wsUrl); - } catch { + } catch (error) { + realtimeLogger.warn("connection failed", { error }); this.scheduleReconnect(); return; } - this.socket.addEventListener("open", () => { this.attempt = 0; }); 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)); - } catch { - // Malformed message — ignore, don't crash the connection. + const message = JSON.parse(event.data as string) as { + topic?: unknown; + payload?: unknown; + sequence?: unknown; + timestamp?: unknown; + }; + if (typeof message.topic !== "string" || !("payload" in message)) return; + const update: RealtimeUpdate = { + payload: message.payload, + sequence: + typeof message.sequence === "number" && Number.isFinite(message.sequence) + ? message.sequence + : undefined, + timestamp: + typeof message.timestamp === "number" && Number.isFinite(message.timestamp) + ? message.timestamp + : typeof message.timestamp === "string" && + Number.isFinite(Date.parse(message.timestamp)) + ? Date.parse(message.timestamp) + : undefined, + }; + const safeTopic = /^[a-z.]+$/.test(message.topic) ? message.topic : "[scoped topic]"; + realtimeLogger.debug("message received", { + topic: safeTopic, + ...summarizeRealtimeValue(update.payload), + }); + this.handlers.get(message.topic)?.forEach((handler) => handler(update)); + } catch (error) { + realtimeLogger.warn("malformed message", { error }); } }); this.socket.addEventListener("close", () => { this.socket = null; if (!this.closedByUser) this.scheduleReconnect(); }); - this.socket.addEventListener("error", () => { - this.socket?.close(); - }); + this.socket.addEventListener("error", () => this.socket?.close()); } private scheduleReconnect(): void { @@ -70,7 +91,6 @@ class RealtimeClient { 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); }; diff --git a/src/lib/realtime/freshness.ts b/src/lib/realtime/freshness.ts new file mode 100644 index 0000000..b225ed3 --- /dev/null +++ b/src/lib/realtime/freshness.ts @@ -0,0 +1,15 @@ +export interface FreshnessKey { + sequence?: number; + timestamp?: number; + source: "rest" | "realtime"; +} + +/** `true` only when candidate is demonstrably newer; unorderable ties favor the current realtime value. */ +export function isNewerUpdate(current: FreshnessKey | null, candidate: FreshnessKey): boolean { + if (!current) return true; + if (current.sequence !== undefined && candidate.sequence !== undefined) + return candidate.sequence > current.sequence; + if (current.timestamp !== undefined && candidate.timestamp !== undefined) + return candidate.timestamp > current.timestamp; + return current.source !== "realtime"; +} diff --git a/src/lib/realtime/redact.test.ts b/src/lib/realtime/redact.test.ts new file mode 100644 index 0000000..431c29e --- /dev/null +++ b/src/lib/realtime/redact.test.ts @@ -0,0 +1,38 @@ +import { describe, expect, it, vi } from "vitest"; +import { realtimeLogger, sanitizeRealtimeValue } from "./redact"; +import { isNewerUpdate } from "./freshness"; + +describe("realtime safety boundaries", () => { + it("redacts nested secrets, bounds unknown values, and removes error messages", () => { + const result = sanitizeRealtimeValue({ + walletAddress: "GSECRETWALLET", + rows: [{ authToken: "secret-token", value: "x".repeat(400) }], + error: new Error("signed payload SECRET"), + }); + const serialized = JSON.stringify(result); + expect(serialized).not.toContain("GSECRETWALLET"); + expect(serialized).not.toContain("secret-token"); + expect(serialized).not.toContain("SECRET"); + expect(serialized).toContain("[truncated]"); + }); + + it("only permits demonstrably newer updates past realtime data", () => { + const realtime = { source: "realtime" as const, sequence: 8, timestamp: 100 }; + expect(isNewerUpdate(realtime, { source: "rest", sequence: 9 })).toBe(true); + expect(isNewerUpdate(realtime, { source: "rest", sequence: 8 })).toBe(false); + expect(isNewerUpdate(realtime, { source: "rest", timestamp: 101 })).toBe(true); + expect(isNewerUpdate(realtime, { source: "rest" })).toBe(false); + }); + + it("routes logs through redaction and keeps debug output disabled by default", () => { + const debug = vi.spyOn(console, "debug").mockImplementation(() => {}); + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + realtimeLogger.debug("frame", { token: "never-log-this" }); + realtimeLogger.warn("connection", { wallet: "GPRIVATE" }); + expect(debug).not.toHaveBeenCalled(); + expect(JSON.stringify(warn.mock.calls)).not.toContain("GPRIVATE"); + expect(JSON.stringify(warn.mock.calls)).not.toContain("never-log-this"); + debug.mockRestore(); + warn.mockRestore(); + }); +}); diff --git a/src/lib/realtime/redact.ts b/src/lib/realtime/redact.ts new file mode 100644 index 0000000..4680e3e --- /dev/null +++ b/src/lib/realtime/redact.ts @@ -0,0 +1,58 @@ +const SENSITIVE_KEY = + /token|authorization|cookie|wallet|address|holder|signature|signed|secret|credential|session/i; +const MAX_STRING = 240; +const MAX_ITEMS = 30; + +/** Redacts secret-bearing fields and bounds unknown values before diagnostics leave the transport. */ +export function sanitizeRealtimeValue(value: unknown, depth = 0): unknown { + if (depth > 5) return "[truncated]"; + if (value instanceof Error) + return { name: value.name.slice(0, 80), message: "[error message redacted]" }; + if (typeof value === "string") + return value.length > MAX_STRING ? `${value.slice(0, MAX_STRING)}…[truncated]` : value; + if (Array.isArray(value)) + return value.slice(0, MAX_ITEMS).map((item) => sanitizeRealtimeValue(item, depth + 1)); + if (value && typeof value === "object") { + return Object.fromEntries( + Object.entries(value as Record) + .slice(0, MAX_ITEMS) + .map(([key, item]) => [ + key, + SENSITIVE_KEY.test(key) ? "[REDACTED]" : sanitizeRealtimeValue(item, depth + 1), + ]) + ); + } + return value; +} + +export function summarizeRealtimeValue(value: unknown): { bytes: number } { + let bytes = 0; + try { + bytes = new TextEncoder().encode(JSON.stringify(value) ?? "").length; + } catch { + /* Non-serializable payload. */ + } + return { bytes }; +} + +type LogLevel = "debug" | "warn"; +function emit(level: LogLevel, message: string, context?: unknown): void { + let verbose = false; + try { + verbose = + process.env.NODE_ENV !== "production" && + sessionStorage.getItem("refract:realtime-debug") === "1"; + } catch { + /* Debug logging stays off when storage is unavailable. */ + } + if (level === "debug" && !verbose) return; + const safe = sanitizeRealtimeValue(context); + const method = level === "debug" ? console.debug : console.warn; + method(`[realtime] ${message}`, safe); +} + +/** All realtime diagnostics pass through this boundary; debug mode is a temporary localStorage switch. */ +export const realtimeLogger = { + debug: (message: string, context?: unknown) => emit("debug", message, context), + warn: (message: string, context?: unknown) => emit("warn", message, context), +}; diff --git a/src/lib/realtime/useRealtimeResource.test.ts b/src/lib/realtime/useRealtimeResource.test.ts new file mode 100644 index 0000000..0208c3c --- /dev/null +++ b/src/lib/realtime/useRealtimeResource.test.ts @@ -0,0 +1,96 @@ +// @vitest-environment jsdom +import { act, renderHook, waitFor } from "@testing-library/react"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +let receive: + ((update: { payload: unknown; sequence?: number; timestamp?: number }) => void) | undefined; +vi.mock("@/lib/flags/useFeatureFlag", () => ({ useFeatureFlag: () => true })); +vi.mock("./client", () => ({ + realtimeClient: { + subscribe: (_topic: string, handler: typeof receive) => { + receive = handler; + return () => {}; + }, + }, +})); +import { useRealtimeResource } from "./useRealtimeResource"; + +afterEach(() => { + receive = undefined; + vi.restoreAllMocks(); + vi.unstubAllGlobals(); +}); + +describe("useRealtimeResource", () => { + it("discards a late seed response, aborts it, and keeps the realtime value", async () => { + const interval = vi.spyOn(globalThis, "setInterval"); + let resolveSeed!: (value: { value: number; timestamp: number }) => void; + let requestSignal!: AbortSignal; + const fetcher = vi.fn((signal: AbortSignal) => { + requestSignal = signal; + return new Promise<{ value: number; timestamp: number }>((resolve) => { + resolveSeed = resolve; + }); + }); + const { result } = renderHook(() => + useRealtimeResource({ + topic: "race.topic", + flag: "enabled", + fetcher, + getRestKey: (value) => ({ timestamp: value.timestamp }), + }) + ); + act(() => receive?.({ payload: { value: 2, timestamp: 200 }, timestamp: 200 })); + expect(requestSignal.aborted).toBe(true); + await act(async () => { + resolveSeed({ value: 1, timestamp: 100 }); + }); + expect(result.current.data).toEqual({ value: 2, timestamp: 200 }); + expect(result.current.error).toBeNull(); + expect(interval).not.toHaveBeenCalled(); + }); + + it("renders the one-shot fixture fallback when REST fails", async () => { + const fetcher = vi.fn().mockRejectedValue(new Error("offline")); + const { result } = renderHook(() => + useRealtimeResource({ + topic: "fallback.topic", + flag: "enabled", + fetcher, + fixture: () => ({ value: 4 }), + }) + ); + await waitFor(() => expect(result.current.data).toEqual({ value: 4 })); + expect(fetcher).toHaveBeenCalledTimes(1); + expect(result.current.isFixture).toBe(true); + }); + + it("coalesces a protocol stats burst into one animation frame commit", async () => { + let queued: FrameRequestCallback | undefined; + const requestFrame = vi.fn((callback: FrameRequestCallback) => { + queued = callback; + return 1; + }); + vi.stubGlobal("requestAnimationFrame", requestFrame); + vi.stubGlobal("cancelAnimationFrame", vi.fn()); + const fetcher = vi.fn().mockResolvedValue({ value: 0 }); + const mergeEvent = (payload: unknown, current: { value: number } | null) => + current && typeof payload === "object" && payload && "delta" in payload + ? { value: current.value + Number((payload as { delta: number }).delta) } + : null; + const { result } = renderHook(() => + useRealtimeResource({ topic: "protocol.stats", flag: "enabled", fetcher, mergeEvent }) + ); + await waitFor(() => expect(result.current.data).toEqual({ value: 0 })); + act(() => { + receive?.({ payload: { delta: 1 }, sequence: 1 }); + receive?.({ payload: { delta: 2 }, sequence: 2 }); + receive?.({ payload: { delta: 3 }, sequence: 3 }); + }); + expect(requestFrame).toHaveBeenCalledTimes(1); + await act(async () => { + queued?.(16); + }); + expect(result.current.data).toEqual({ value: 6 }); + }); +}); diff --git a/src/lib/realtime/useRealtimeResource.ts b/src/lib/realtime/useRealtimeResource.ts new file mode 100644 index 0000000..ffcf166 --- /dev/null +++ b/src/lib/realtime/useRealtimeResource.ts @@ -0,0 +1,201 @@ +"use client"; + +import { useCallback, useEffect, useRef, useState } from "react"; +import { useFeatureFlag } from "@/lib/flags/useFeatureFlag"; +import { realtimeClient, type RealtimeUpdate } from "./client"; +import { isNewerUpdate, type FreshnessKey } from "./freshness"; + +type AppliedKey = FreshnessKey; +interface ResourceState { + data: T | null; + loading: boolean; + error: string | null; + isFixture: boolean; + lastUpdated: Date | null; + isRefreshing: boolean; +} +interface Options { + topic: string; + flag: string; + fetcher: (signal: AbortSignal) => Promise; + fixture?: () => T; + mapEvent?: (payload: unknown) => T | null; + mergeEvent?: (payload: unknown, current: T | null) => T | null; + getRestKey?: (data: T) => Pick; +} + +/** Shared REST/realtime merge. Realtime wins ties and unverifiable ordering; a newer comparable REST key may advance state. */ +export function useRealtimeResource({ + topic, + flag, + fetcher, + fixture, + mapEvent, + mergeEvent, + getRestKey, +}: Options) { + const enabled = useFeatureFlag(flag); + const [state, setState] = useState>({ + data: null, + loading: true, + error: null, + isFixture: false, + lastUpdated: null, + isRefreshing: false, + }); + const stateRef = useRef(state); + const appliedRef = useRef(null); + const eventGeneration = useRef(0); + const controllerRef = useRef(null); + const frameRef = useRef(null); + const pendingDataRef = useRef(null); + const mountedRef = useRef(false); + const topicRef = useRef(topic); + const enabledRef = useRef(enabled); + const mapEventRef = useRef(mapEvent); + mapEventRef.current = mapEvent; + const mergeEventRef = useRef(mergeEvent); + mergeEventRef.current = mergeEvent; + const fetcherRef = useRef(fetcher); + fetcherRef.current = fetcher; + const fixtureRef = useRef(fixture); + fixtureRef.current = fixture; + const getRestKeyRef = useRef(getRestKey); + getRestKeyRef.current = getRestKey; + const commit = useCallback((next: ResourceState) => { + stateRef.current = next; + setState(next); + }, []); + + const refresh = useCallback( + async (manual = true) => { + controllerRef.current?.abort(); + const controller = new AbortController(); + controllerRef.current = controller; + const generation = eventGeneration.current; + commit({ + ...stateRef.current, + loading: !stateRef.current.data, + isRefreshing: manual, + error: null, + }); + try { + const data = await fetcherRef.current(controller.signal); + if ( + !mountedRef.current || + controller.signal.aborted || + generation !== eventGeneration.current + ) + return; + const key = { source: "rest" as const, ...(getRestKeyRef.current?.(data) ?? {}) }; + const applied = appliedRef.current; + if (applied?.source === "realtime" && !isNewerUpdate(applied, key)) return; + appliedRef.current = key; + commit({ + data, + loading: false, + error: null, + isFixture: false, + lastUpdated: new Date(), + isRefreshing: false, + }); + } catch (error) { + if (!mountedRef.current || controller.signal.aborted) return; + if (fixtureRef.current) + commit({ + data: fixtureRef.current(), + loading: false, + error: null, + isFixture: true, + lastUpdated: new Date(), + isRefreshing: false, + }); + else + commit({ + ...stateRef.current, + loading: false, + error: "Failed to load data", + isRefreshing: false, + }); + } + }, + [commit] + ); + + useEffect(() => { + if (topicRef.current !== topic) { + topicRef.current = topic; + appliedRef.current = null; + pendingDataRef.current = null; + eventGeneration.current += 1; + commit({ + data: null, + loading: true, + error: null, + isFixture: false, + lastUpdated: null, + isRefreshing: false, + }); + } + mountedRef.current = true; + void refresh(false); + return () => { + mountedRef.current = false; + controllerRef.current?.abort(); + if (frameRef.current !== null) cancelAnimationFrame(frameRef.current); + }; + }, [commit, refresh, topic]); + + useEffect(() => { + const wasEnabled = enabledRef.current; + enabledRef.current = enabled; + if (!enabled) { + if (wasEnabled) { + appliedRef.current = null; + eventGeneration.current += 1; + void refresh(false); + } + return; + } + const unsubscribe = realtimeClient.subscribe(topic, (update: RealtimeUpdate) => { + const data = mergeEventRef.current + ? mergeEventRef.current(update.payload, pendingDataRef.current ?? stateRef.current.data) + : mapEventRef.current + ? mapEventRef.current(update.payload) + : (update.payload as T); + if (data == null) return; + const previous = appliedRef.current; + if (previous && !isNewerUpdate(previous, { ...update, source: "realtime" })) return; + eventGeneration.current += 1; + controllerRef.current?.abort(); + appliedRef.current = { + sequence: update.sequence, + timestamp: update.timestamp, + source: "realtime", + }; + pendingDataRef.current = data; + const apply = () => { + const latest = pendingDataRef.current ?? data; + pendingDataRef.current = null; + commit({ + data: latest, + loading: false, + error: null, + isFixture: false, + lastUpdated: update.timestamp ? new Date(update.timestamp) : new Date(), + isRefreshing: false, + }); + }; + if (topic === "protocol.stats" && typeof requestAnimationFrame === "function") { + if (frameRef.current === null) + frameRef.current = requestAnimationFrame(() => { + frameRef.current = null; + apply(); + }); + } else apply(); + }); + return unsubscribe; + }, [commit, enabled, refresh, topic]); + + return { ...state, refresh: () => refresh(true) }; +} From b9dbcdd1bac6e48ea962a5cd542626fcd553af3e Mon Sep 17 00:00:00 2001 From: faith-ng <332940157+faith-ng@users.noreply.github.com> Date: Wed, 21 Oct 2026 15:30:09 +0000 Subject: [PATCH 2/2] test: cover realtime protocol stats push --- e2e/realtime-stats.spec.ts | 49 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) create mode 100644 e2e/realtime-stats.spec.ts diff --git a/e2e/realtime-stats.spec.ts b/e2e/realtime-stats.spec.ts new file mode 100644 index 0000000..3204d82 --- /dev/null +++ b/e2e/realtime-stats.spec.ts @@ -0,0 +1,49 @@ +import { expect, test } from "@playwright/test"; + +test("landing protocol statistics update after a protocol.stats push", async ({ page }) => { + const sockets: Array<{ send: (message: string) => void }> = []; + await page.route("**/flags.json", (route) => + route.fulfill({ json: { flags: { "realtime-protocol-stats": true } } }) + ); + await page.route("**/api/v1/pool/stats", (route) => + route.fulfill({ + json: { + totalUsdc: (2_500_000 * 1e7).toString(), + totalShares: "0", + lockedUsdc: "0", + premiumAccrued: "0", + availableUsdc: "0", + utilizationBps: 0, + apyBps: 0, + sharePrice: 1, + maxUtilizationBps: 8000, + }, + }) + ); + await page.route("**/api/v1/claims/stats", (route) => + route.fulfill({ + json: { + activePolicies: 11, + processedClaims: 0, + totalPayout: "0", + settlementConfigured: true, + }, + }) + ); + await page.routeWebSocket("**/ws", (webSocket) => { + sockets.push(webSocket); + }); + + await page.goto("/"); + await expect(page.getByText("Policies Active")).toBeVisible(); + expect(sockets).toHaveLength(1); + sockets[0].send( + JSON.stringify({ + topic: "protocol.stats", + sequence: 1, + timestamp: Date.now(), + payload: { tvl: (3_200_000 * 1e7).toString(), activePolicies: 31, cumulativePayouts: "0" }, + }) + ); + await expect(page.getByText("31", { exact: true })).toBeVisible(); +});