diff --git a/README.md b/README.md index e2faf91..a210259 100644 --- a/README.md +++ b/README.md @@ -87,6 +87,10 @@ Runtime errors are reported to Sentry when `NEXT_PUBLIC_SENTRY_DSN` is set. See [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. + ## Realtime layer (dev) Realtime is off by default (`public/flags.json`; see `docs/realtime-protocol.md` and `docs/realtime-rollout.md`). 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/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(); +}); diff --git a/src/app/dashboard/page.tsx b/src/app/dashboard/page.tsx index a847ab9..b0151bb 100644 --- a/src/app/dashboard/page.tsx +++ b/src/app/dashboard/page.tsx @@ -33,7 +33,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); @@ -96,6 +97,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}
-
- ))} + ))} +
@@ -356,9 +359,10 @@ export default function Home() {

Recent payouts

Payout updates are temporarily unavailable.

}>
-

- 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/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 f00465f..e69de29 100644 --- a/src/hooks/useRecentClaims.ts +++ b/src/hooks/useRecentClaims.ts @@ -1,16 +0,0 @@ -"use client"; -import { useCallback, useEffect, useState } from "react"; -import { fetchRecentClaims, type ClaimRecord } from "@/lib/api/claims"; -import { FIXTURE_RECENT_CLAIMS } from "@/lib/fixtures/recentClaims"; -import { useRealtimeTopic } from "./useRealtimeTopic"; - -export function useRecentClaims() { - const load = useCallback(async (signal: AbortSignal) => { - try { const result = await fetchRecentClaims({ limit: 10 }, signal); setFixture(false); return result.items; } - catch { setFixture(true); return FIXTURE_RECENT_CLAIMS; } - }, []); - const { data, isStale, isLoading } = useRealtimeTopic("claim_settlement", load); - const [isFixture, setFixture] = useState(false); - useEffect(() => { if (data && data !== FIXTURE_RECENT_CLAIMS) setFixture(false); }, [data]); - return { data: data ?? FIXTURE_RECENT_CLAIMS, loading: isLoading, isFixture, isStale }; -} 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 f553be3..e69de29 100644 --- a/src/lib/flags/flagsConfig.ts +++ b/src/lib/flags/flagsConfig.ts @@ -1,23 +0,0 @@ -export interface FlagsDocument { - flags: Record; -} - -/** Safe default if the remote flags source is unreachable — new/risky features default off. */ -export const DEFAULT_FLAGS: Record = { - "realtime-oracle": false, - "realtime-pool-stats": false, -}; - -export const FLAGS_ENDPOINT = "/flags.json"; -export const FLAGS_CACHE_TTL_MS = 5 * 60 * 1000; - -/** Reads a local override, e.g. NEXT_PUBLIC_FLAGS_OVERRIDE='{"realtime-oracle":true}'. */ -export function readLocalOverride(): Record | null { - const raw = process.env.NEXT_PUBLIC_FLAGS_OVERRIDE; - if (!raw) return null; - try { - return JSON.parse(raw) as Record; - } catch { - return null; - } -} 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) }; +}