diff --git a/.eslintrc.json b/.eslintrc.json index ceb0654..bf8a90e 100644 --- a/.eslintrc.json +++ b/.eslintrc.json @@ -12,7 +12,7 @@ "message": "Hard-coded rgb()/rgba() colours are not allowed in JSX. Use design tokens from src/lib/coverage/metadata.ts or CSS custom properties (see CONTRIBUTING.md)." }, { - "selector": "Literal[value=1e7]", + "selector": "Literal[value=10000000]", "message": "Use USDC_BASE_UNIT / fromStroops from src/lib/format.ts instead of the literal 1e7." }, { diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 460b5ff..60f2d13 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -185,6 +185,32 @@ jobs: - name: Run ESLint run: npm run lint + realtime-benchmark: + name: Realtime render benchmark + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-node@v7 + with: + node-version: 20 + cache: npm + - run: npm ci + - run: npx playwright install --with-deps chromium + - run: npm run dev & + env: + NEXT_PUBLIC_REALTIME_PROFILER: "1" + - run: | + for attempt in $(seq 1 60); do + if curl --fail --silent http://localhost:3000 >/dev/null; then exit 0; fi + sleep 1 + done + echo "Next.js dev server did not become ready" >&2 + exit 1 + - run: npm run realtime:benchmark + env: + REALTIME_BENCHMARK_URL: http://127.0.0.1:3000 + typecheck: name: Typecheck runs-on: ubuntu-latest diff --git a/docs/realtime-performance.md b/docs/realtime-performance.md new file mode 100644 index 0000000..d7e1617 --- /dev/null +++ b/docs/realtime-performance.md @@ -0,0 +1,9 @@ +# Realtime resilience and performance + +Realtime data uses the existing REST endpoints during cold start and after subscription failure. The client caches topic snapshots by network, wallet and topic; entries carry a version and capture time and render as stale until a REST response or realtime push supersedes them. IndexedDB is optional, and storage failures never block the page. + +`RealtimeBoundary` isolates a route's live region. A render or subscription handler failure marks the topic degraded and renders its fallback. Retry clears that topic's degraded state. Diagnostics include topic, route, transport and error type only; payloads and wallet addresses are omitted. + +The development-only React Profiler records commits and actual durations by topic. Run `NEXT_PUBLIC_REALTIME_PROFILER=1 npm run dev`, then `npm run realtime:benchmark`. The scripted benchmark measures five runs with the socket disconnected and five runs with 100 oracle pushes at 25 Hz on landing, `/cover` and `/provide`; it reports both scenarios' median commit time, p95 commit time, long tasks, and p98 event duration to `reports/realtime-benchmark.json`. + +Budgets are checked in at `reports/realtime-benchmark-baseline.json`: median commit duration below 16 ms, p95 commit duration below 50 ms, no long tasks during the scripted stream, and p98 interaction duration below 200 ms. These thresholds allow CI variance while flagging work that exceeds a frame or the Core Web Vitals INP “good” threshold. Lighthouse CI separately warns on INP above 200 ms and total blocking time above 200 ms. Absolute measurements vary by runner; compare like-for-like CI runs and use the median of five benchmark runs. diff --git a/lighthouserc.js b/lighthouserc.js index d5bb13e..b58a420 100644 --- a/lighthouserc.js +++ b/lighthouserc.js @@ -37,6 +37,8 @@ module.exports = { 'categories:accessibility': ['error', { minScore: 0.8 }], 'categories:best-practices': ['error', { minScore: 0.8 }], 'categories:seo': ['error', { minScore: 0.8 }], + 'interaction-to-next-paint': ['warn', { maxNumericValue: 200 }], + 'total-blocking-time': ['warn', { maxNumericValue: 200 }], }, }, upload: { diff --git a/package.json b/package.json index 3338cf1..9c38a16 100644 --- a/package.json +++ b/package.json @@ -9,9 +9,9 @@ "start": "next start", "lint": "next lint", "typecheck": "tsc --noEmit", - "test": "vitest run", - "test:watch": "vitest", - "test:coverage": "vitest run --coverage", + "test": "vitest run --config vitest.config.mts", + "test:watch": "vitest --config vitest.config.mts", + "test:coverage": "vitest run --config vitest.config.mts --coverage", "test:visual": "playwright test --project=visual", "test:visual:update": "playwright test --project=visual --update-snapshots", "audit": "audit-ci --config audit-ci.json", @@ -20,6 +20,7 @@ "size": "size-limit", "size:why": "size-limit --why", "tokens:check": "node scripts/check-tokens.mjs", + "realtime:benchmark": "node scripts/realtime-benchmark.mjs", "mock:realtime": "node scripts/mock-realtime-server.mjs" }, "lint-staged": { diff --git a/reports/realtime-benchmark-baseline.json b/reports/realtime-benchmark-baseline.json new file mode 100644 index 0000000..f70dfe2 --- /dev/null +++ b/reports/realtime-benchmark-baseline.json @@ -0,0 +1,10 @@ +{ + "schemaVersion": 1, + "scenario": "100 oracle pushes at 25Hz across landing, /cover, and /provide; median of five browser runs", + "budgets": { + "medianCommitDurationMs": 16, + "p95CommitDurationMs": 50, + "longTaskCount": 0, + "p98InteractionDurationMs": 200 + } +} diff --git a/scripts/realtime-benchmark.mjs b/scripts/realtime-benchmark.mjs new file mode 100644 index 0000000..45d4675 --- /dev/null +++ b/scripts/realtime-benchmark.mjs @@ -0,0 +1,73 @@ +import { chromium } from "@playwright/test"; +import { readFile, writeFile } from "node:fs/promises"; + +const origin = process.env.REALTIME_BENCHMARK_URL ?? "http://127.0.0.1:3000"; +const baseline = JSON.parse(await readFile(new URL("../reports/realtime-benchmark-baseline.json", import.meta.url), "utf8")); +const routes = ["/", "/cover", "/provide"]; +const runs = Number(process.env.REALTIME_BENCHMARK_RUNS ?? 5); +const browser = await chromium.launch({ headless: true }); +const samples = { disconnected: [], streaming: [] }; +for (let index = 0; index < runs; index += 1) { + for (const scenario of ["disconnected", "streaming"]) { + const context = await browser.newContext(); + const page = await context.newPage(); + await page.addInitScript(() => { + window.__realtimeVitals = { longTasks: [], interactions: [] }; + try { + new PerformanceObserver((list) => window.__realtimeVitals.longTasks.push(...list.getEntries().map((entry) => entry.duration))) + .observe({ type: "longtask", buffered: true }); + new PerformanceObserver((list) => window.__realtimeVitals.interactions.push(...list.getEntries().map((entry) => entry.duration))) + .observe({ type: "event", buffered: true, durationThreshold: 16 }); + } catch { /* event timing is not supported by every browser build */ } + }); + await page.routeWebSocket("**/ws", (socket) => { + if (scenario === "disconnected") { socket.close(); return; } + let push = 0; + const timer = setInterval(() => { + push += 1; + socket.send(JSON.stringify({ topic: "oracle_update", payload: [ + { coverageType: "StablecoinDepeg", type: "oracle_update", value: 1 + push / 100000, threshold: 0.95, severity: "low", message: "benchmark" }, + ] })); + if (push >= 100) clearInterval(timer); + }, 40); + }); + for (const route of routes) { + await page.goto(`${origin}${route}`, { waitUntil: "domcontentloaded" }); + await page.waitForTimeout(250); + await page.locator("body").click({ position: { x: 10, y: 10 } }).catch(() => undefined); + } + await page.waitForTimeout(4200); + const result = await page.evaluate(() => ({ + profile: window.__REFRACT_REALTIME_PROFILE__ ?? {}, + longTasks: window.__realtimeVitals.longTasks, + interactions: window.__realtimeVitals.interactions, + })); + samples[scenario].push(result); + await context.close(); + } +} +await browser.close(); +const median = (values) => values.sort((a, b) => a - b)[Math.floor(values.length / 2)] ?? 0; +const measure = (scenarioSamples) => { + const commitDurations = scenarioSamples.flatMap((sample) => Object.values(sample.profile).flatMap((item) => item.durations ?? [])); + const longTasks = scenarioSamples.flatMap((sample) => sample.longTasks); + const interactions = scenarioSamples.flatMap((sample) => sample.interactions); + return { + medianCommitDurationMs: median(commitDurations), + p95CommitDurationMs: commitDurations.sort((a, b) => a - b)[Math.floor(commitDurations.length * 0.95)] ?? 0, + medianCommitsPerTopic: Object.fromEntries(["oracle_update", "claim_settlement", "pool_stats"].map((topic) => [topic, median(scenarioSamples.map((sample) => sample.profile[topic]?.commits ?? 0))])), + longTaskCount: longTasks.length, + medianLongTaskDurationMs: median(longTasks), + p98InteractionDurationMs: interactions.sort((a, b) => a - b)[Math.floor(interactions.length * 0.98)] ?? 0, + }; +}; +const disconnected = measure(samples.disconnected); +const streaming = measure(samples.streaming); +const report = { + generatedAt: new Date().toISOString(), runs, routes, + scenarios: { disconnected, streaming }, + budget: baseline.budgets, +}; +await writeFile("reports/realtime-benchmark.json", `${JSON.stringify(report, null, 2)}\n`); +console.log(JSON.stringify(report, null, 2)); +if (streaming.medianCommitDurationMs > report.budget.medianCommitDurationMs || streaming.p95CommitDurationMs > report.budget.p95CommitDurationMs || streaming.longTaskCount > report.budget.longTaskCount || streaming.p98InteractionDurationMs > report.budget.p98InteractionDurationMs) process.exitCode = 1; diff --git a/src/app/cover/page.tsx b/src/app/cover/page.tsx index 9217d98..e5df608 100644 --- a/src/app/cover/page.tsx +++ b/src/app/cover/page.tsx @@ -34,6 +34,9 @@ import { parseCoverQueryParams } from "@/lib/coverQueryParams"; import { PolicyCertificateModal } from "@/components/PolicyCertificate"; import { validateBuyCoverage } from "@/lib/validation/coverage"; import { searchFlights, POPULAR_FLIGHTS, type FlightInfo } from "@/lib/flights"; +import { RealtimeBoundary } from "@/components/realtime/RealtimeBoundary"; +import { RealtimeProfiler } from "@/lib/realtime/devProfiler"; +import { publishAccountBalanceDelta } from "@/hooks/useAccountBalance"; const RISK_TAG_COLORS: Record = { low: "#10b981", @@ -306,6 +309,7 @@ export default function CoverPage() { return; } const txHash = await signAndSubmit(result.txXdr, wallet.address, wallet.networkPassphrase, { hardware: wallet.hardwareWallet }); + publishAccountBalanceDelta(wallet.address, "USDC", -fromStroops(result.policy.premium)); setSubmission({ status: "success", result, demo: false, txHash }); updateTransaction(txId, { status: "confirmed", txHash }); } catch (err) { @@ -419,7 +423,8 @@ export default function CoverPage() {
{/* STEP 1: Select Type */} {currentStep === 1 && ( -
+
Coverage types are temporarily unavailable. Try again later or open the Dashboard.
}> +
Step 1 of 4: Select Coverage Type
@@ -430,30 +435,13 @@ export default function CoverPage() { )} - {typesLoading && ( -
- {Array.from({ length: 5 }).map((_, i -
- - {typesError && ( - -

Couldn't load coverage types: {typesError}

- - )} - - {typesLoading && ( + {typesLoading && !coverageTypes && (
{Array.from({ length: 5 }).map((_, i) => ( ))}
- } - errorRender={(err) => ( - -

Couldn't load coverage types: {err}

- )} - > {coverageTypes && (
)} - -
-
-
+
+ )} {/* STEP 2: Trigger Parameters */} @@ -715,6 +701,22 @@ export default function CoverPage() {
setDurationDays(Number(event.target.value))} + className="w-full accent-violet-500" + /> + +
+ + +
+ + )} + {/* Right: Quote panel */}
diff --git a/src/app/dashboard/page.tsx b/src/app/dashboard/page.tsx index 6bcaa5d..a847ab9 100644 --- a/src/app/dashboard/page.tsx +++ b/src/app/dashboard/page.tsx @@ -17,6 +17,8 @@ import { import type { Policy } from "@/lib/api/policies"; import type { ClaimRecord } from "@/lib/api/claims"; import { normalizePolicies, normalizeClaims, selectPolicyStatus, type PolicyStatus } from "@/lib/entities/policies"; +import { RealtimeBoundary } from "@/components/realtime/RealtimeBoundary"; +import { RealtimeProfiler } from "@/lib/realtime/devProfiler"; const COVERAGE_ICONS = ["🪙", "📉", "🛡️", "🔐", "✈️"]; const COVERAGE_COLORS = ["#8b5cf6", "#f59e0b", "#10b981", "#ef4444", "#06b6d4"]; @@ -74,7 +76,8 @@ export default function DashboardPage() {
} /> -
+

Live claim updates are temporarily unavailable. Reconnect to retry; cached policy data remains available.

}> +
@@ -230,7 +233,8 @@ export default function DashboardPage() { )} -
+
+
diff --git a/src/app/page.tsx b/src/app/page.tsx index 9cc74d9..f2e8f30 100644 --- a/src/app/page.tsx +++ b/src/app/page.tsx @@ -14,6 +14,8 @@ 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 { RealtimeBoundary } from "@/components/realtime/RealtimeBoundary"; +import { RealtimeProfiler } from "@/lib/realtime/devProfiler"; function Counter({ to, prefix = "", suffix = "", decimals = 0 }: { to: number; prefix?: string; suffix?: string; decimals?: number }) { const [val, setVal] = useState(0); @@ -176,7 +178,8 @@ export default function Home() { {/* Hero right: Live coverage card */}
- +

Live oracle readings are unavailable. Other Refract sections remain available.

}> +
Live Oracle Status @@ -253,7 +256,8 @@ export default function Home() { : "Oracles auto-refreshed every 60 seconds"}
- + + diff --git a/src/components/NotificationCenter.tsx b/src/components/NotificationCenter.tsx index 7c2bb4f..5ad3a1d 100644 --- a/src/components/NotificationCenter.tsx +++ b/src/components/NotificationCenter.tsx @@ -27,7 +27,7 @@ const SEVERITY_COLORS: Record s.attentionCues); const setAttentionCues = usePreferencesStore((s) => s.setAttentionCues); const containerRef = useRef(null); @@ -149,6 +149,7 @@ export function NotificationCenter() {
Notifications + {isStale && Saved} {unreadCount > 0 && ( {unreadCount} new diff --git a/src/components/layout/Navbar.test.tsx b/src/components/layout/Navbar.test.tsx new file mode 100644 index 0000000..84a60f5 --- /dev/null +++ b/src/components/layout/Navbar.test.tsx @@ -0,0 +1,32 @@ +import React from "react"; +import { render, screen, waitFor } from "@testing-library/react"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { ACTIVE_NETWORK } from "@/lib/network"; +import { realtimeClient } from "@/lib/realtime/client"; +import { Navbar } from "./Navbar"; + +vi.mock("@/lib/wallet/WalletProvider", () => ({ + useWallet: () => ({ status: "connected", address: "GTESTACCOUNT" }), + truncateAddress: (address: string) => address, +})); + +afterEach(() => vi.unstubAllGlobals()); + +describe("Navbar account balance", () => { + it("reflects a pushed balance without a page reload", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ + ok: true, + json: async () => ({ + sequence: "12", + balances: [{ asset_type: "credit_alphanum4", asset_code: "USDC", asset_issuer: "GISSUER", balance: "12.5000000" }], + }), + })); + render(React.createElement(Navbar)); + await waitFor(() => expect(screen.getByLabelText("Live wallet balance")).toHaveTextContent("12.50 USDC")); + realtimeClient.dispatch(`account:${ACTIVE_NETWORK.id}:GTESTACCOUNT`, { + balances: [{ asset: "USDC:GISSUER", balance: "11.2500000", assetType: "credit_alphanum4", decimals: 7 }], + sequence: "13", updatedAt: Date.now(), + }); + await waitFor(() => expect(screen.getByLabelText("Live wallet balance")).toHaveTextContent("11.25 USDC")); + }); +}); diff --git a/src/components/layout/Navbar.tsx b/src/components/layout/Navbar.tsx index 65bddc5..e69de29 100644 --- a/src/components/layout/Navbar.tsx +++ b/src/components/layout/Navbar.tsx @@ -1,223 +0,0 @@ -"use client"; - -import { useEffect, useRef, useState } from "react"; -import Link from "next/link"; -import { usePathname } from "next/navigation"; -import { cn } from "@/lib/cn"; -import { useRealtimeHealth } from "@/hooks/useRealtimeHealth"; -import { usePrefersReducedMotion } from "@/hooks/usePrefersReducedMotion"; -import { NetworkBadge } from "@/components/layout/NetworkBadge"; - -const NAV_LINKS = [ - { label: "Coverage", href: "/cover" }, - { label: "Provide Capital", href: "/provide" }, - { label: "Dashboard", href: "/dashboard" }, - { label: "Status", href: "/status" }, -]; - -type ConnectionState = "live" | "connecting" | "reconnecting" | "offline"; - -const CONNECTION_META: Record< - ConnectionState, - { label: string; description: string; dotClass: string; textClass: string } -> = { - live: { - label: "Live", - description: "Realtime connection is active. Data is streaming live.", - dotClass: "bg-emerald-400", - textClass: "text-emerald-300", - }, - connecting: { - label: "Connecting", - description: "Establishing the realtime connection. Data may be delayed.", - dotClass: "bg-amber-400", - textClass: "text-amber-300", - }, - reconnecting: { - label: "Reconnecting", - description: "Realtime connection was lost and is being re-established. Data may be stale.", - dotClass: "bg-amber-400", - textClass: "text-amber-300", - }, - offline: { - label: "Offline", - description: "Realtime connection is unavailable. Data shown may be stale.", - dotClass: "bg-rose-400", - textClass: "text-rose-300", - }, -}; - -function ConnectionStatusBadge() { - const { status } = useRealtimeHealth(); - const prefersReducedMotion = usePrefersReducedMotion(); - - // Treat unknown/initial states as neutral to avoid SSR mismatch. - const state: ConnectionState = - status === "live" || status === "connecting" || status === "reconnecting" || status === "offline" - ? status - : "connecting"; - - const meta = CONNECTION_META[state]; - const shouldPulse = state === "live" && !prefersReducedMotion; - - return ( - - - ); -} - -interface NavbarProps { - /** Right-aligned slot, typically the wallet connect button. */ - right?: React.ReactNode; -} - -export function Navbar({ right }: NavbarProps) { - const pathname = usePathname(); - const [menuOpen, setMenuOpen] = useState(false); - const menuButtonRef = useRef(null); - - useEffect(() => { - if (!menuOpen) return; - function onEscape(e: KeyboardEvent) { - if (e.key !== "Escape") return; - setMenuOpen(false); - menuButtonRef.current?.focus(); - } - document.addEventListener("keydown", onEscape); - return () => document.removeEventListener("keydown", onEscape); - }, [menuOpen]); - - return ( - - ); -} diff --git a/src/components/realtime/RealtimeBoundary.test.ts b/src/components/realtime/RealtimeBoundary.test.ts new file mode 100644 index 0000000..cde23b0 --- /dev/null +++ b/src/components/realtime/RealtimeBoundary.test.ts @@ -0,0 +1,22 @@ +import { describe, expect, it, vi } from "vitest"; +import { render, screen } from "@testing-library/react"; +import { RealtimeBoundary } from "./RealtimeBoundary"; +import { realtimeClient } from "@/lib/realtime/client"; + +function Broken() { throw new Error("bad decoded value"); } + +describe("RealtimeBoundary", () => { + it("falls back and degrades only the failed topic", () => { + const error = vi.spyOn(console, "error").mockImplementation(() => undefined); + render(React.createElement(React.Fragment, null, + React.createElement(RealtimeBoundary, { topic: "oracle_update", route: "/", fallback: () => React.createElement("p", null, "REST fallback") }, React.createElement(Broken)), + React.createElement(RealtimeBoundary, { topic: "pool_stats", route: "/provide", fallback: () => React.createElement("p", null, "Other fallback") }, React.createElement("p", null, "Pool remains live")), + )); + expect(screen.getByText("REST fallback")).toBeTruthy(); + expect(screen.getByText("Pool remains live")).toBeTruthy(); + expect(realtimeClient.isDegraded("oracle_update")).toBe(true); + expect(realtimeClient.isDegraded("pool_stats")).toBe(false); + error.mockRestore(); + }); +}); +import React from "react"; diff --git a/src/components/realtime/RealtimeBoundary.ts b/src/components/realtime/RealtimeBoundary.ts new file mode 100644 index 0000000..6a8968a --- /dev/null +++ b/src/components/realtime/RealtimeBoundary.ts @@ -0,0 +1,32 @@ +"use client"; + +import React from "react"; +import { realtimeClient } from "@/lib/realtime/client"; + +interface Props { + topic: string; + route: string; + fallback: (retry: () => void) => React.ReactNode; + children: React.ReactNode; + onRetry?: () => void; +} +interface State { failed: boolean } + +export class RealtimeBoundary extends React.Component { + state: State = { failed: false }; + static getDerivedStateFromError() { return { failed: true }; } + componentDidMount() { this.listen = realtimeClient.onError(this.props.topic, (error) => { + realtimeClient.markDegraded(this.props.topic); + this.setState({ failed: true }); + console.error("Realtime boundary fallback", { topic: this.props.topic, route: this.props.route, transport: "websocket", error: error.name }); + }); } + componentWillUnmount() { this.listen?.(); } + private listen?: () => void; + componentDidCatch(error: Error) { realtimeClient.markDegraded(this.props.topic, error, this.props.route); } + retry = () => { + realtimeClient.retry(this.props.topic); + this.setState({ failed: false }); + this.props.onRetry?.(); + }; + render() { return this.state.failed ? this.props.fallback(this.retry) : this.props.children; } +} diff --git a/src/hooks/useAccountBalance.test.ts b/src/hooks/useAccountBalance.test.ts new file mode 100644 index 0000000..1ebe05a --- /dev/null +++ b/src/hooks/useAccountBalance.test.ts @@ -0,0 +1,46 @@ +import { act, renderHook, waitFor } from "@testing-library/react"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { ACTIVE_NETWORK } from "@/lib/network"; +import { realtimeClient } from "@/lib/realtime/client"; +import { useAccountBalance } from "./useAccountBalance"; + +const ADDRESS = "GTESTACCOUNT"; +const accountResponse = (balance: string, sequence = "12") => ({ + balances: [{ asset_type: "credit_alphanum4", asset_code: "USDC", asset_issuer: "GISSUER", balance }], + sequence, +}); + +afterEach(() => vi.unstubAllGlobals()); + +describe("useAccountBalance", () => { + it("starts from Horizon and applies authoritative account pushes", async () => { + const fetch = vi.fn().mockResolvedValue({ ok: true, json: async () => accountResponse("12.5000000") }); + vi.stubGlobal("fetch", fetch); + const { result } = renderHook(() => useAccountBalance(ADDRESS)); + await waitFor(() => expect(result.current.balances[0]?.balance).toBe("12.5000000")); + + act(() => realtimeClient.dispatch(`account:${ACTIVE_NETWORK.id}:${ADDRESS}`, { + balances: [{ asset: "USDC:GISSUER", balance: "11.2500000", assetType: "credit_alphanum4", decimals: 7 }], + sequence: "13", updatedAt: Date.now(), + })); + await waitFor(() => expect(result.current.sequence).toBe("13")); + expect(result.current.balances[0].balance).toBe("11.2500000"); + expect(result.current.isStale).toBe(false); + }); + + it("treats an unfunded account as an empty account without an error", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ status: 404, ok: false })); + const { result } = renderHook(() => useAccountBalance(ADDRESS)); + await waitFor(() => expect(result.current.isLoading).toBe(false)); + expect(result.current.balances).toEqual([]); + expect(result.current.sequence).toBe("0"); + }); + + it("keeps a network failure in loading fallback state without throwing", async () => { + vi.stubGlobal("fetch", vi.fn().mockRejectedValue(new Error("offline"))); + const { result } = renderHook(() => useAccountBalance(ADDRESS)); + await waitFor(() => expect(result.current.isLoading).toBe(false)); + expect(result.current.balances).toEqual([]); + expect(result.current.isStale).toBe(true); + }); +}); diff --git a/src/hooks/useAccountBalance.ts b/src/hooks/useAccountBalance.ts new file mode 100644 index 0000000..1215624 --- /dev/null +++ b/src/hooks/useAccountBalance.ts @@ -0,0 +1,58 @@ +"use client"; +import { useCallback, useEffect, useState } from "react"; +import { ACTIVE_NETWORK } from "@/lib/network"; +import { useRealtimeTopic } from "./useRealtimeTopic"; + +export interface AccountBalance { asset: string; balance: string; assetType: string; decimals: number } +export interface AccountState { balances: AccountBalance[]; sequence: string; updatedAt: number } +const OPTIMISTIC_EVENT = "refract:account-balance-delta"; +export function publishAccountBalanceDelta(address: string, asset: string, delta: number) { + if (typeof window !== "undefined") window.dispatchEvent(new CustomEvent(OPTIMISTIC_EVENT, { detail: { address, asset, delta } })); +} + +async function fetchAccount(address: string, signal: AbortSignal): Promise { + const response = await fetch(`${ACTIVE_NETWORK.horizonUrl}/accounts/${address}`, { signal }); + if (response.status === 404) return { balances: [], sequence: "0", updatedAt: Date.now() }; + if (!response.ok) throw new Error(`Horizon account request failed (${response.status})`); + const account = await response.json(); + return { balances: (account.balances ?? []).map((item: { asset_type: string; asset_code?: string; asset_issuer?: string; balance: string }) => ({ + asset: item.asset_type === "native" ? "XLM" : `${item.asset_code ?? "UNKNOWN"}:${item.asset_issuer ?? ""}`, + balance: item.balance, assetType: item.asset_type, decimals: 7, + })), sequence: String(account.sequence), updatedAt: Date.now() }; +} + +export function useAccountBalance(address: string | null) { + const topic = address ? `account:${ACTIVE_NETWORK.id}:${address}` : "account:disconnected"; + const fetcher = useCallback((signal: AbortSignal) => address + ? fetchAccount(address, signal) + : Promise.resolve({ balances: [], sequence: "0", updatedAt: Date.now() }), [address]); + const { data, isStale, isLoading, refresh } = useRealtimeTopic( + topic, fetcher, address ?? undefined, Boolean(address), + ); + const [optimistic, setOptimistic] = useState<{ address: string; balances: AccountBalance[] } | null>(null); + useEffect(() => setOptimistic(null), [address]); + useEffect(() => { + if (!address) return; + const listener = (event: Event) => { + const detail = (event as CustomEvent<{ address: string; asset: string; delta: number }>).detail; + if (detail.address !== address) return; + setOptimistic((previous) => { + const balances = previous?.address === address ? previous.balances : data?.balances ?? []; + return { address, balances: balances.map((entry) => + entry.asset === detail.asset || entry.asset.startsWith(`${detail.asset}:`) + ? { ...entry, balance: String(Math.max(0, Number(entry.balance) + detail.delta)) } : entry) }; + }); + setTimeout(() => { void refresh().catch(() => undefined); }, 2_500); + }; + window.addEventListener(OPTIMISTIC_EVENT, listener); + return () => window.removeEventListener(OPTIMISTIC_EVENT, listener); + }, [address, data, refresh]); + const deduct = useCallback((asset: string, amount: number) => { + if (!address) return; + setOptimistic({ address, balances: (data?.balances ?? []).map((entry) => entry.asset === asset + ? { ...entry, balance: String(Math.max(0, Number(entry.balance) - amount)) } : entry) }); + }, [data]); + useEffect(() => { if (data && !isStale) setOptimistic(null); }, [data, isStale]); + const visibleOptimistic = optimistic?.address === address ? optimistic.balances : null; + return { balances: visibleOptimistic ?? data?.balances ?? [], sequence: data?.sequence ?? "0", updatedAt: data?.updatedAt ?? null, isStale, isLoading, deduct }; +} diff --git a/src/hooks/useClaims.ts b/src/hooks/useClaims.ts index e69de29..81b896f 100644 --- a/src/hooks/useClaims.ts +++ b/src/hooks/useClaims.ts @@ -0,0 +1,24 @@ +"use client"; +import { useCallback, useMemo } from "react"; +import { fetchHolderClaims, fetchRecentClaims, type ClaimRecord } from "@/lib/api/claims"; +import type { Policy } from "@/lib/api/policies"; +import { ACTIVE_NETWORK } from "@/lib/network"; +import { useWallet } from "@/lib/wallet/WalletProvider"; +import { useRealtimeTopic } from "./useRealtimeTopic"; + +export type ClaimsResult = ClaimRecord[] & { data: ClaimRecord[] | null; loading: boolean; isStale: boolean }; + +export function useClaims(addressArg?: string | null, _policies?: Policy[] | null): ClaimsResult { + const wallet = useWallet(); + const address = addressArg === undefined + ? (wallet.status === "connected" ? wallet.address : null) + : addressArg; + const load = useCallback(async (signal: AbortSignal) => address + ? (await fetchHolderClaims(address, {}, signal)).items + : (await fetchRecentClaims({ limit: 50 }, signal)).items, [address]); + const topic = address ? `claims:${ACTIVE_NETWORK.id}:${address}` : "claim_settlement"; + const { data, isLoading, isStale } = useRealtimeTopic(topic, load, address ?? undefined); + return useMemo(() => Object.assign([...(data ?? [])], { + data: data ?? (address ? null : []), loading: isLoading, isStale, + }) as ClaimsResult, [data, address, isLoading, isStale]); +} diff --git a/src/hooks/useCoverageTypes.ts b/src/hooks/useCoverageTypes.ts index e69de29..2e7e7ff 100644 --- a/src/hooks/useCoverageTypes.ts +++ b/src/hooks/useCoverageTypes.ts @@ -0,0 +1,22 @@ +"use client"; +import { useCallback, useState } from "react"; +import { fetchCoverageTypes, type CoverageTypeInfo } from "@/lib/api/policies"; +import { FIXTURE_COVERAGE_TYPES } from "@/lib/fixtures/coverageTypes"; +import { useRealtimeTopic } from "./useRealtimeTopic"; + +export function useCoverageTypes() { + const [error, setError] = useState(null); + const [isFixture, setFixture] = useState(false); + const load = useCallback(async (signal: AbortSignal): Promise => { + try { + const response = await fetchCoverageTypes(signal); + setError(null); setFixture(false); return response.coverageTypes; + } catch (reason) { + setFixture(true); + setError(reason instanceof Error ? reason.message : "Coverage data unavailable"); + return FIXTURE_COVERAGE_TYPES; + } + }, []); + const { data, isLoading, isStale } = useRealtimeTopic("coverage_types", load); + return { data, loading: isLoading, error, isFixture, isStale }; +} diff --git a/src/hooks/useNotifications.ts b/src/hooks/useNotifications.ts index 6f559dd..e69de29 100644 --- a/src/hooks/useNotifications.ts +++ b/src/hooks/useNotifications.ts @@ -1,84 +0,0 @@ -"use client"; - -import { useEffect, useMemo, useState } from "react"; -import { useWallet } from "@/lib/wallet/WalletProvider"; -import { useHolderPolicies } from "./useHolderPolicies"; -import { useClaims } from "./useClaims"; -import { useAttentionCue } from "./useAttentionCue"; -import { useLockupStatus } from "./useLockupStatus"; -import { useOracleStatus } from "./useOracleStatus"; -import { - deriveNotifications, - type RefractNotification, -} from "@/lib/notifications"; - -const STORAGE_KEY = "refract_read_notifications_v1"; - -export function useNotifications() { - const wallet = useWallet(); - const address = wallet.status === "connected" ? wallet.address : null; - - const { data: policies } = useHolderPolicies(address); - const claims = useClaims(address, policies); - const { lockupExpiresAt } = useLockupStatus(address); - const { data: oracleReadings } = useOracleStatus(); - - const [readIds, setReadIds] = useState>(new Set()); - - useEffect(() => { - try { - const stored = localStorage.getItem(STORAGE_KEY); - if (stored) { - setReadIds(new Set(JSON.parse(stored))); - } - } catch { - // Ignore localStorage errors - } - }, []); - - const saveReadIds = (newSet: Set) => { - setReadIds(newSet); - try { - localStorage.setItem(STORAGE_KEY, JSON.stringify(Array.from(newSet))); - } catch { - // Ignore - } - }; - - const notifications = useMemo(() => { - return deriveNotifications({ - policies, - claims, - lockupExpiresAt, - oracleReadings, - readIds, - }); - }, [policies, claims, lockupExpiresAt, oracleReadings, readIds]); - - useAttentionCue(notifications); - - const unreadCount = useMemo(() => { - return notifications.filter((n) => !n.read).length; - }, [notifications]); - - const markAsRead = (id: string) => { - const updated = new Set(readIds); - updated.add(id); - saveReadIds(updated); - }; - - const markAllAsRead = () => { - const updated = new Set(readIds); - for (const n of notifications) { - updated.add(n.id); - } - saveReadIds(updated); - }; - - return { - notifications, - unreadCount, - markAsRead, - markAllAsRead, - }; -} diff --git a/src/hooks/useOracleStatus.ts b/src/hooks/useOracleStatus.ts index 08efeb5..e69de29 100644 --- a/src/hooks/useOracleStatus.ts +++ b/src/hooks/useOracleStatus.ts @@ -1,182 +0,0 @@ -"use client"; - -import { useCallback, useEffect, useRef, useState } from "react"; -import { fetchOracleStatus } from "@/lib/api/oracle"; -import { env } from "@/lib/env"; -import { useFeatureFlag } from "@/lib/flags/useFeatureFlag"; -import { FIXTURE_ORACLE_READINGS } from "@/lib/fixtures/oracle"; -import { realtimeClient } from "@/lib/realtime/client"; -import { - mergeRestReadings, - mergeStreamReadings, - parseOraclePayload, - type StreamReading, -} from "@/lib/realtime/oracleMerge"; - -export interface OracleStatusState { - data: StreamReading[] | null; - loading: boolean; - isFixture: boolean; - lastUpdated: Date | null; - isRefreshing: boolean; - /** True when the realtime stream was live but dropped/went silent; data then comes from REST polling until it resumes. */ - isStale: boolean; - refresh: () => Promise; -} - -export const ORACLE_TOPIC = "oracle.readings"; -/** No frame for this long means the socket is dead even if it never reported closing. */ -export const STREAM_STALE_MS = 90_000; -/** Coalesce bursts of frames into one state update. */ -const FLUSH_MS = 250; - -type StreamState = "idle" | "live" | "stale"; - -/** - * Oracle readings: hydrated from GET /api/v1/oracle/status, then — when the - * `realtime-oracle` flag is on and NEXT_PUBLIC_REALTIME_URL is set — kept fresh - * by the `oracle.readings` stream (any frame doubles as the heartbeat). - * Pushed readings are merged by coverage type, rejecting duplicates and - * out-of-order `seq`/`ts`. While the stream is live REST polling pauses; on - * disconnect or {@link STREAM_STALE_MS} of silence the hook fetches REST - * immediately, resumes 60s polling and reports `isStale` until frames resume. - * The stream is only joined in an effect, so server render matches REST hydration. - */ -export function useOracleStatus(pollIntervalMs = 60000): OracleStatusState { - const flagOn = useFeatureFlag("realtime-oracle"); - const streamEnabled = flagOn && Boolean(env.NEXT_PUBLIC_REALTIME_URL); - - const [data, setData] = useState(null); - const [loading, setLoading] = useState(true); - const [isFixture, setIsFixture] = useState(false); - const [lastUpdated, setLastUpdated] = useState(null); - const [isRefreshing, setIsRefreshing] = useState(false); - const [streamState, setStreamState] = useState("idle"); - - const dataRef = useRef(null); - const streamStateRef = useRef("idle"); - const activeControllerRef = useRef(null); - - const commit = useCallback((next: StreamReading[]) => { - dataRef.current = next; - setData(next); - setLastUpdated(new Date()); - }, []); - - const updateStreamState = useCallback((next: StreamState) => { - streamStateRef.current = next; - setStreamState(next); - }, []); - - const fetchReadings = useCallback( - async (isManual = false) => { - activeControllerRef.current?.abort(); - const controller = new AbortController(); - activeControllerRef.current = controller; - if (isManual) setIsRefreshing(true); - - try { - const { readings } = await fetchOracleStatus(controller.signal); - if (controller.signal.aborted) return; - setIsFixture(false); - commit(mergeRestReadings(dataRef.current ?? [], readings)); - } catch { - if (controller.signal.aborted) return; - // Keep last known good readings on a later failure; only fall back to the fixture when there is nothing. - if (!dataRef.current) { - setIsFixture(true); - commit(FIXTURE_ORACLE_READINGS); - } - } finally { - if (!controller.signal.aborted) { - setLoading(false); - setIsRefreshing(false); - } - } - }, - [commit] - ); - - // REST: initial hydration, polling (paused while the stream is live or the tab is hidden), and fallback. - useEffect(() => { - void fetchReadings(); - const intervalId = setInterval(() => { - if (typeof document !== "undefined" && document.visibilityState === "hidden") return; - if (streamStateRef.current === "live") return; - void fetchReadings(); - }, pollIntervalMs); - - const onVisible = () => { - if (document.visibilityState === "visible" && streamStateRef.current !== "live") - void fetchReadings(); - }; - document.addEventListener("visibilitychange", onVisible); - - return () => { - clearInterval(intervalId); - document.removeEventListener("visibilitychange", onVisible); - activeControllerRef.current?.abort(); - }; - }, [fetchReadings, pollIntervalMs]); - - // Realtime: subscribe on mount, unsubscribe on unmount (StrictMode-safe: cleanup undoes everything). - useEffect(() => { - if (!streamEnabled) return; - - let buffer: StreamReading[] = []; - let flushTimer: ReturnType | null = null; - let staleTimer: ReturnType | null = null; - - const degrade = () => { - if (streamStateRef.current !== "live") return; - updateStreamState("stale"); - void fetchReadings(); - }; - const armStaleTimer = () => { - if (staleTimer) clearTimeout(staleTimer); - staleTimer = setTimeout(degrade, STREAM_STALE_MS); - }; - const flush = () => { - flushTimer = null; - const batch = buffer; - buffer = []; - const merged = mergeStreamReadings(dataRef.current ?? [], batch); - if (merged === dataRef.current) return; - setIsFixture(false); - commit(merged); - }; - - const unsubscribe = realtimeClient.subscribe(ORACLE_TOPIC, (payload) => { - armStaleTimer(); - if (streamStateRef.current !== "live") { - activeControllerRef.current?.abort(); - updateStreamState("live"); - } - buffer.push(...parseOraclePayload(payload)); - if (!flushTimer) flushTimer = setTimeout(flush, FLUSH_MS); - }); - const unwatch = realtimeClient.onConnectionChange((connected) => { - if (!connected) degrade(); - }); - - return () => { - unsubscribe(); - unwatch(); - if (flushTimer) clearTimeout(flushTimer); - if (staleTimer) clearTimeout(staleTimer); - updateStreamState("idle"); - }; - }, [streamEnabled, fetchReadings, commit, updateStreamState]); - - const refresh = useCallback(() => fetchReadings(true), [fetchReadings]); - - return { - data, - loading, - isFixture, - lastUpdated, - isRefreshing, - isStale: streamState === "stale", - refresh, - }; -} diff --git a/src/hooks/usePoolStats.ts b/src/hooks/usePoolStats.ts index 9d35252..e69de29 100644 --- a/src/hooks/usePoolStats.ts +++ b/src/hooks/usePoolStats.ts @@ -1,120 +0,0 @@ -"use client"; - -import { useEffect, useState } from "react"; -import { useQuery, useQueryClient } from "@tanstack/react-query"; -import { ApiUnreachableError } from "@/lib/api/client"; -import type { DataSource } from "@/lib/api/dataSource"; -import { fetchPoolStats, type PoolStats } from "@/lib/api/pool"; -import { FIXTURE_POOL_STATS } from "@/lib/fixtures/poolStats"; -import { useFeatureFlag } from "@/lib/flags/useFeatureFlag"; -import { realtimeClient } from "@/lib/realtime/client"; -import { readPoolStats } from "@/lib/stellar/rpc"; - -interface PoolStatsState { - data: PoolStats | null; - loading: boolean; - error: string | null; - isFixture: boolean; - dataSource: DataSource | null; - /** True when the live stream is enabled but disconnected, so `data` is the REST snapshot. */ - isStale: boolean; -} - -interface PoolStatsResult { - data: PoolStats | null; - dataSource: DataSource | null; - error: string | null; -} - -const NUMERIC_FIELDS: (keyof PoolStats)[] = ["utilizationBps", "apyBps", "sharePrice", "maxUtilizationBps"]; -const STRING_FIELDS: (keyof PoolStats)[] = ["totalUsdc", "totalShares", "lockedUsdc", "premiumAccrued", "availableUsdc"]; -/** A pushed TVL differing from REST by more than this ratio is treated as a bad frame. */ -const MAX_TVL_DRIFT = 0.5; - -/** Pushed frames are full snapshots; reject anything partial rather than half-applying it. */ -function isFullSnapshot(p: unknown): p is PoolStats { - const o = p as Record | null; - return ( - !!o && - NUMERIC_FIELDS.every((k) => typeof o[k] === "number" && Number.isFinite(o[k] as number)) && - STRING_FIELDS.every((k) => typeof o[k] === "string" && /^\d+$/.test(o[k] as string)) - ); -} - -function agreesWithRest(pushed: PoolStats, rest: PoolStats | null): boolean { - const base = Number(rest?.totalUsdc ?? 0); - if (!rest || base === 0) return true; - return Math.abs(Number(pushed.totalUsdc) - base) / base <= MAX_TVL_DRIFT; -} - -async function loadPoolStats(signal: AbortSignal): Promise { - try { - return { data: await fetchPoolStats(signal), dataSource: "api", error: null }; - } catch (err) { - if (signal.aborted) throw err; - if (!(err instanceof ApiUnreachableError)) { - return { - data: null, - dataSource: null, - error: err instanceof Error ? err.message : "Failed to load pool stats", - }; - } - } - try { - return { data: await readPoolStats(signal), dataSource: "chain", error: null }; - } catch { - if (signal.aborted) throw new DOMException("Aborted", "AbortError"); - return { data: FIXTURE_POOL_STATS, dataSource: "fixture", error: null }; - } -} - -/** - * Pool-wide stats: API → on-chain read → bundled fixture, optionally kept live - * by the `pool.stats` realtime topic (behind the `realtime-pool-stats` flag). - * The query key is shared, so every component (landing, /provide) reuses one - * cached request within the QueryClient's staleTime. - */ -export function usePoolStats(): PoolStatsState { - const live = useFeatureFlag("realtime-pool-stats"); - const queryClient = useQueryClient(); - const [pushed, setPushed] = useState(null); - const [connected, setConnected] = useState(realtimeClient.getStatus() === "open"); - - const { data: result, isLoading } = useQuery({ - queryKey: ["poolStats"], - queryFn: ({ signal }) => loadPoolStats(signal), - }); - - const rest = result?.data ?? null; - - useEffect(() => { - if (!live) return; - setConnected(realtimeClient.getStatus() === "open"); - const offStatus = realtimeClient.onStatus((s) => setConnected(s === "open")); - const off = realtimeClient.subscribe("pool.stats", (payload) => { - if (!isFullSnapshot(payload)) return; - const current = queryClient.getQueryData(["poolStats"])?.data ?? null; - if (agreesWithRest(payload, current)) setPushed(payload); - else { - setPushed(null); - void queryClient.invalidateQueries({ queryKey: ["poolStats"] }); - } - }); - return () => { - off(); - offStatus(); - }; - }, [live, queryClient]); - - // On disconnect fall back to the REST snapshot. - const useLive = live && connected && pushed !== null; - - return { - data: useLive ? pushed : rest, - loading: isLoading, - error: result?.error ?? null, - isFixture: result?.dataSource === "fixture", - dataSource: result?.dataSource ?? null, - isStale: live && !connected, - }; -} diff --git a/src/hooks/useRealtimeTopic.ts b/src/hooks/useRealtimeTopic.ts index 47881f3..e69de29 100644 --- a/src/hooks/useRealtimeTopic.ts +++ b/src/hooks/useRealtimeTopic.ts @@ -1,365 +0,0 @@ -"use client"; - -import { useCallback, useEffect, useRef, useState } from "react"; - -import { useFeatureFlag } from "@/lib/flags/useFeatureFlag"; -import { fidelityController, realtimeBus, realtimeClient } from "@/lib/realtime"; -import type { ConnectionState } from "@/lib/realtime/client"; -import type { RealtimeTopic, RealtimeTopics, RealtimeViewStatus } from "@/lib/realtime"; -import { - clearPersistedSubscriptions, - persistSubscriptions, - readPersistedSubscriptions, - type PersistedSubscription, -} from "../lib/realtime/persistence"; -import { - mergeRealtimeEvent, - mergeSnapshot, - type MergeFn, - type RealtimeEvent, - type RealtimeState, -} from "../lib/realtime/merge"; -import { getRegistry, type SubscriptionRegistry } from "../lib/realtime/registry"; -import type { RealtimeClient } from "../lib/realtime/client"; - -/** Maps data presence and transport state to the shared view-status enum. */ -export function deriveViewStatus( - hasData: boolean, - enabled: boolean, - conn: ConnectionState -): RealtimeViewStatus { - if (!hasData) return "empty"; - if (!enabled) return "polling"; - if (conn === "connected") return "live"; - if (conn === "reconnecting") return "reconnecting"; - return "disconnected"; -} - -export type RealtimeStatus = "idle" | "loading" | "live" | "polling" | "error"; - -export interface UseRealtimeTopicOptions { - /** - * Whether updates for this topic are replaceable (only the latest value matters). - * Replaceable topics are coalesced to at most one render per animation frame. - */ - replaceable?: boolean; - /** - * Optional selector to derive the rendered value from the merged state. - * Defaults to the merged state itself. - */ - select?: (state: RealtimeState) => RealtimeState; - /** Fetches the REST snapshot used as the initial/fallback value. */ - snapshotFetcher?: () => Promise; - /** Merge rule applied to streamed events (defaults to latest-wins). */ - merge?: MergeFn; - /** When false the hook skips realtime and relies on polling only. */ - enabled?: boolean; - /** Injectable polling strategy used when realtime is disabled/unavailable. */ - pollIntervalMs?: number; - /** Whether the realtime feature flag is on. */ - realtimeEnabled?: boolean; - /** Whether the realtime connection is healthy. */ - connected?: boolean; - /** Subscribes to a topic; returns an unsubscribe function. */ - subscribe?: (topic: string, onEvent: (event: T) => void) => () => void; - /** Realtime client used to resolve the shared subscription registry. */ - client?: RealtimeClient; - /** Explicit registry override (tests / inspector). */ - registry?: SubscriptionRegistry; - /** Scope used to namespace persisted subscriptions. */ - scope?: string; - /** Wallet used to namespace persisted subscriptions. */ - wallet?: string; - /** Known topics used to validate persisted subscriptions. */ - knownTopics?: ReadonlySet; - /** REST snapshot used for first paint and as the fallback when realtime is off. */ - initial?: T; - /** Keep this topic subscribed even when the fidelity tier would drop it (e.g. claim settlement). */ - critical?: boolean; -} - -export interface UseRealtimeTopicResult { - data: T | undefined; - status: RealtimeStatus; - lastEventAt: number | null; - cursor: string | null; - setCursor: (cursor: string | null) => void; - clear: () => void; -} - -const DEFAULT_POLL_INTERVAL_MS = 30_000; - -const isTestEnv = (): boolean => { - if (typeof process !== "undefined" && process.env && process.env.NODE_ENV === "test") { - return true; - } - return typeof (globalThis as { __REALTIME_DISABLE_COALESCE__?: boolean }).__REALTIME_DISABLE_COALESCE__ === "boolean" - ? Boolean((globalThis as { __REALTIME_DISABLE_COALESCE__?: boolean }).__REALTIME_DISABLE_COALESCE__) - : false; -}; - -const prefersReducedMotion = (): boolean => { - if (typeof window === "undefined" || typeof window.matchMedia !== "function") { - return false; - } - try { - return window.matchMedia("(prefers-reduced-motion: reduce)").matches; - } catch { - return false; - } -}; - -const scheduleFrame = (cb: () => void): (() => void) => { - if (typeof window !== "undefined" && typeof window.requestAnimationFrame === "function") { - const id = window.requestAnimationFrame(cb); - return () => window.cancelAnimationFrame(id); - } - const id = setTimeout(cb, 16); - return () => clearTimeout(id); -}; - -/** - * Generic React binding over the realtime transport + merge layers. - * - * Subscribes on mount, tears down on unmount, reacts to connection health, - * and falls back to REST polling when realtime is disabled or unavailable. - * Subscriptions are routed through the shared registry so identical topics are - * reference-counted rather than opened per hook instance. - * - * Replaceable topics (e.g. oracle readings) are buffered and flushed at most - * once per animation frame so high-frequency streams cannot cause render - * storms. A trailing-edge flush guarantees the latest value is always rendered. - * Coalescing is disabled under reduced motion and in automated tests so - * flushing is deterministic. - * - * The last cursor for a topic is persisted to session storage and rehydrated - * on mount so reloads and route navigations resume from the snapshot instead - * of refetching from scratch. - */ -export function useRealtimeTopic( - topic: T, - options: UseRealtimeTopicOptions = {} -): UseRealtimeTopicResult & { enabled: boolean } { - const { - replaceable = false, - select, - snapshotFetcher, - merge, - enabled = true, - pollIntervalMs = DEFAULT_POLL_INTERVAL_MS, - realtimeEnabled = true, - connected = true, - subscribe, - client, - registry, - scope = "", - wallet = "", - knownTopics, - initial, - critical = false, - } = options; - - const flagEnabled = useFeatureFlag("realtime-oracle"); - const [data, setData] = useState(initial); - const [status, setStatus] = useState("idle"); - const [lastEventAt, setLastEventAt] = useState(null); - const [cursor, setCursorState] = useState(null); - const hydratedRef = useRef(false); - - // Keep the latest merge/fetcher/select in refs so effects don't resubscribe - // on every render (avoids resubscribe storms). - const mergeRef = useRef(merge); - mergeRef.current = merge; - const fetcherRef = useRef(snapshotFetcher); - fetcherRef.current = snapshotFetcher; - const selectRef = useRef(select); - selectRef.current = select; - - const pendingEventsRef = useRef([]); - const mountedRef = useRef(true); - - // Coalescing state for replaceable topics. - const stateRef = useRef | undefined>(undefined); - const pendingRef = useRef | null>(null); - const cancelRef = useRef<(() => void) | null>(null); - - const coalesce = replaceable && !isTestEnv() && !prefersReducedMotion(); - const coalesceRef = useRef(coalesce); - coalesceRef.current = coalesce; - - const flush = useCallback((): void => { - cancelRef.current = null; - const next = pendingRef.current; - if (next === null) { - return; - } - pendingRef.current = null; - stateRef.current = next; - if (mountedRef.current) { - setData(next as unknown as RealtimeTopics[T]); - } - }, []); - - const apply = useCallback((next: RealtimeState): void => { - if (!coalesceRef.current) { - stateRef.current = next; - if (mountedRef.current) { - setData(next as unknown as RealtimeTopics[T]); - } - return; - } - pendingRef.current = next; - if (cancelRef.current === null) { - cancelRef.current = scheduleFrame(flush); - } - }, [flush]); - - const applyEvent = useCallback((event: RealtimeTopics[T]) => { - if (!mountedRef.current) return; - const current = stateRef.current; - const merged = current !== undefined - ? mergeRealtimeEvent(current, event as unknown as RealtimeEvent) - : (event as unknown as RealtimeState); - const selected = selectRef.current ? selectRef.current(merged) : merged; - apply(selected); - setLastEventAt(Date.now()); - }, [apply]); - - const fetchSnapshot = useCallback(async () => { - const fetcher = fetcherRef.current; - if (!fetcher) return; - try { - const snapshot = await fetcher(); - if (!mountedRef.current) return; - const pending = pendingEventsRef.current; - pendingEventsRef.current = []; - const merged = mergeSnapshot(snapshot, pending, mergeRef.current); - stateRef.current = merged as unknown as RealtimeState; - setData(merged); - } catch { - if (mountedRef.current) setStatus("error"); - } - }, []); - - // Resolve the registry: explicit override, else the shared registry bound to - // the client. Falls back to a direct subscribe when neither is available. - const resolvedRegistry = registry ?? (client ? getRegistry(client) : undefined); - const subscribeFn = resolvedRegistry - ? (t: string, onEvent: (event: RealtimeTopics[T]) => void) => resolvedRegistry.subscribe(t, onEvent) - : subscribe; - - const realtimeActive = enabled && realtimeEnabled && connected && !!subscribeFn; - - // Rehydrate the persisted cursor for this topic on mount. - useEffect(() => { - if (hydratedRef.current) return; - hydratedRef.current = true; - const persisted = readPersistedSubscriptions(wallet, scope, knownTopics); - const match = persisted.find((entry) => entry.topic === topic); - if (match) setCursorState(match.cursor); - }, [wallet, scope, topic, knownTopics]); - - const persist = useCallback( - (nextCursor: string | null) => { - const existing = readPersistedSubscriptions(wallet, scope, knownTopics); - const others = existing.filter((entry) => entry.topic !== topic); - const next: PersistedSubscription[] = [ - ...others, - { topic, scope, cursor: nextCursor }, - ]; - persistSubscriptions(wallet, scope, next); - }, - [wallet, scope, topic, knownTopics], - ); - - const setCursor = useCallback( - (nextCursor: string | null) => { - setCursorState(nextCursor); - persist(nextCursor); - }, - [persist], - ); - - const clear = useCallback(() => { - setCursorState(null); - clearPersistedSubscriptions(wallet, scope); - }, [wallet, scope]); - - // Snapshot fetch on mount / when the topic changes. - useEffect(() => { - mountedRef.current = true; - setStatus("loading"); - void fetchSnapshot(); - return () => { - mountedRef.current = false; - }; - }, [topic, fetchSnapshot]); - - // Realtime subscription lifecycle (routed through the registry). - useEffect(() => { - if (!realtimeActive || !subscribeFn) return; - setStatus("live"); - const unsubscribe = subscribeFn(topic, (event) => { - if (pendingEventsRef.current.length > 0) { - pendingEventsRef.current.push(event); - } - applyEvent(event); - }); - return () => { - unsubscribe(); - if (cancelRef.current !== null) { - cancelRef.current(); - cancelRef.current = null; - } - // Trailing-edge flush on unmount so the latest value is never stranded. - flush(); - }; - }, [topic, realtimeActive, subscribeFn, applyEvent, flush]); - - // REST polling fallback when realtime is disabled or unavailable. - useEffect(() => { - if (!enabled || realtimeActive) return; - setStatus("polling"); - const interval = setInterval(() => { - void fetchSnapshot(); - }, pollIntervalMs); - return () => clearInterval(interval); - }, [enabled, realtimeActive, pollIntervalMs, fetchSnapshot]); - - // Event-bus subscription (never the socket directly). Late mounts read the - // bus snapshot; `status` is the single enum views map to presentation, so - // components never inspect transport internals. - useEffect(() => { - if (!flagEnabled) return; - const snap = realtimeBus.getSnapshot(topic); - if (snap !== undefined) setData(snap as unknown as RealtimeTopics[T]); - const offBus = realtimeBus.subscribe(topic, (value) => { - setData(value as unknown as RealtimeTopics[T]); - }); - const offConn = realtimeClient.onConnectionChange((conn: ConnectionState) => { - setStatus(deriveViewStatus(true, flagEnabled, conn) as unknown as RealtimeStatus); - }); - const release = critical ? fidelityController.requireTopic(topic) : () => {}; - realtimeClient.connect(); - return () => { - offBus(); - offConn(); - release(); - }; - }, [flagEnabled, topic, critical]); - - return { data, status, lastEventAt, cursor, setCursor, clear, enabled: flagEnabled }; -} - -/** - * Deterministic flush hook for tests: forces any buffered replaceable update - * to render immediately. - */ -export function useRealtimeFlush(): () => void { - return () => { - if (typeof (globalThis as { __REALTIME_FLUSH__?: () => void }).__REALTIME_FLUSH__ === "function") { - (globalThis as { __REALTIME_FLUSH__?: () => void }).__REALTIME_FLUSH__!(); - } - }; -} - diff --git a/src/hooks/useRecentClaims.ts b/src/hooks/useRecentClaims.ts index e69de29..f00465f 100644 --- a/src/hooks/useRecentClaims.ts +++ b/src/hooks/useRecentClaims.ts @@ -0,0 +1,16 @@ +"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/realtime/client.ts b/src/lib/realtime/client.ts index 4eed71d..e69de29 100644 --- a/src/lib/realtime/client.ts +++ b/src/lib/realtime/client.ts @@ -1,429 +0,0 @@ -import { API_BASE_URL } from "../api/client"; -import { realtimeBus } from "./bus"; -import type { RealtimeTopic, RealtimeTopics } from "./topics"; -import { RequestCorrelator, type RequestOptions } from "./correlation"; -import { emitRealtimeEvent } from "@/lib/analytics/realtime"; -import { readRealtimeOverride, resolveRealtimeConfig } from "./overrides"; - -type Handler = (payload: unknown) => void; -type ConnectionListener = (state: ConnectionState) => void; - -export type ConnectionState = "idle" | "connected" | "reconnecting"; - -/** The subset of WebSocket the client relies on; fakes implement the same shape. */ -export interface RealtimeTransport { - addEventListener(type: "open" | "message" | "close" | "error", listener: (event: { data?: unknown }) => void): void; - send(data: string): void; - close(): void; -} - -export type TransportFactory = (url: string) => RealtimeTransport; - -const defaultTransportFactory: TransportFactory = (url) => new WebSocket(url) as unknown as RealtimeTransport; -let transportFactory: TransportFactory = defaultTransportFactory; -let customTransport = false; - -/** Test-only: inject a transport factory. Pass `null` to restore the WebSocket default. */ -export function __setTransportFactory(factory: TransportFactory | null): void { - transportFactory = factory ?? defaultTransportFactory; - customTransport = factory !== null; -} - -export type DecodeErrorCode = "invalid-frame" | "frame-too-large" | "too-deep" | "invalid-message"; -export class RealtimeDecodeError extends Error { - constructor(public readonly code: DecodeErrorCode) { - super(`Realtime decode failed: ${code}`); - this.name = "RealtimeDecodeError"; - } -} -export type DecodedRealtimeMessage = { topic: string; payload: unknown; sequence?: number }; -export const MAX_REALTIME_FRAME_BYTES = 256 * 1024; -export const MAX_REALTIME_NESTING_DEPTH = 32; - -function nestingDepth(text: string): number { - let depth = 0, - max = 0, - quoted = false, - escaped = false; - for (const char of text) { - if (quoted) { - if (escaped) escaped = false; - else if (char === "\\") escaped = true; - else if (char === '"') quoted = false; - continue; - } - if (char === '"') quoted = true; - else if (char === "{" || char === "[") { - depth++; - max = Math.max(max, depth); - } else if (char === "}" || char === "]") depth--; - if (max > MAX_REALTIME_NESTING_DEPTH) return max; - } - return max; -} - -/** Pure, bounded wire decoder. It never exposes the input as an object prototype. */ -export function decodeRealtimeFrame(frame: unknown): DecodedRealtimeMessage { - let text: string; - if (typeof frame === "string") text = frame; - else if (frame instanceof ArrayBuffer) { - if (frame.byteLength > MAX_REALTIME_FRAME_BYTES) - throw new RealtimeDecodeError("frame-too-large"); - try { - text = new TextDecoder("utf-8", { fatal: true }).decode(frame); - } catch { - throw new RealtimeDecodeError("invalid-frame"); - } - } else if (ArrayBuffer.isView(frame)) { - if (frame.byteLength > MAX_REALTIME_FRAME_BYTES) - throw new RealtimeDecodeError("frame-too-large"); - try { - text = new TextDecoder("utf-8", { fatal: true }).decode(frame); - } catch { - throw new RealtimeDecodeError("invalid-frame"); - } - } else throw new RealtimeDecodeError("invalid-frame"); - if (new TextEncoder().encode(text).byteLength > MAX_REALTIME_FRAME_BYTES) - throw new RealtimeDecodeError("frame-too-large"); - if (nestingDepth(text) > MAX_REALTIME_NESTING_DEPTH) throw new RealtimeDecodeError("too-deep"); - let parsed: unknown; - try { - parsed = JSON.parse(text); - } catch { - throw new RealtimeDecodeError("invalid-frame"); - } - if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) - throw new RealtimeDecodeError("invalid-message"); - const value = parsed as Record; - if ( - typeof value.topic !== "string" || - !/^[a-z][a-z0-9_.:-]{0,127}$/i.test(value.topic) || - !Object.hasOwn(value, "payload") - ) - throw new RealtimeDecodeError("invalid-message"); - return { - topic: value.topic, - payload: value.payload, - ...(Number.isSafeInteger(value.sequence) ? { sequence: value.sequence as number } : {}), - }; -} - -export class RealtimeClient { - private socket: WebSocket | null = null; - private source: EventSource | null = null; - private handlers = new Map>(); -const MAX_BACKOFF_MS = 30_000; -const BASE_BACKOFF_MS = 1_000; -const FALLBACK_POLL_MS = 15_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. - */ -export class RealtimeClient { - private socket: RealtimeTransport | null = null; - private source: EventSource | null = null; - private handlers = new Map>(); - private attempt = 0; - private closedByUser = false; - private correlator = new RequestCorrelator(); - private reconnectTimer: ReturnType | null = null; - private teardownTicket = 0; - private pollTimer: ReturnType | null = null; - private activeTransport: RealtimeTransport | "rest" | "off" | null = null; - private opening = false; - private lastSequenceByTopic = new Map(); - private fallbackActive = false; - private wireTopics = new Set(); - private listeners = new Set(); - private state: ConnectionState = "idle"; - - getConnectionState(): ConnectionState { - return this.state; - } - - onConnectionChange(listener: ConnectionListener): () => void { - this.listeners.add(listener); - return () => { - this.listeners.delete(listener); - }; - } - - private setState(next: ConnectionState): void { - if (this.state === next) return; - this.state = next; - this.listeners.forEach((l) => l(next)); - } - - /** - * Applies the wire-level topic set as a diff (subscribe/unsubscribe deltas). - * The full set is re-sent on every (re)connect so a tier change or a - * reconnect never loses a topic. - */ - syncTopics(topics: ReadonlySet): void { - const prev = this.wireTopics; - this.wireTopics = new Set(topics); - if (this.socket?.readyState !== 1) return; - for (const t of topics) if (!prev.has(t)) this.send({ action: "subscribe", topic: t }); - for (const t of prev) if (!topics.has(t)) this.send({ action: "unsubscribe", topic: t }); - } - - private send(msg: unknown): void { - try { - this.socket?.send(JSON.stringify(msg)); - } catch { - // Socket closing — the reconnect path re-sends the full topic set. - } - } - - private config() { - const override = readRealtimeOverride(); - return resolveRealtimeConfig( - { - enabled: true, - endpoint: API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws"), - transport: "ws", - }, - override, - typeof window === "undefined" ? undefined : window.location.hostname - ); - } - - connect(): void { - if ((!customTransport && typeof WebSocket === "undefined") || this.socket) return; - this.closedByUser = false; - this.open(); - } - - private async open(): Promise { - if (this.opening) return; - this.opening = true; - let remoteKillSwitch = false; - if (process.env.NODE_ENV === "production") { - try { - const response = await fetch("/flags.json", { cache: "no-store" }); - const flags = response.ok - ? ((await response.json()) as { flags?: Record }) - : null; - remoteKillSwitch = flags?.flags?.["realtime-oracle"] !== true; - } catch { - remoteKillSwitch = true; - } - } - const baseConfig = this.config(); - const config = resolveRealtimeConfig( - baseConfig, - readRealtimeOverride(), - typeof window === "undefined" ? undefined : window.location.hostname, - remoteKillSwitch - ); - this.opening = false; - if (this.handlers.size === 0) return; - if (!config.enabled || config.transport === "off") { - this.activeTransport = "off"; - this.setState("idle"); - emitRealtimeEvent("unavailable", { reason: "killed" }); - return; - } - this.closedByUser = false; - const transport = config.transport; - const endpoint = config.endpoint; - if (transport === "rest") { - this.startPolling(endpoint); - return; - } - this.activeTransport = transport; - try { - if (transport === "sse") { - if (typeof EventSource === "undefined") { - this.startPolling(endpoint); - return; - } - this.source = new EventSource(endpoint); - this.source.onopen = () => { - this.attempt = 0; - if (this.fallbackActive) { - this.fallbackActive = false; - emitRealtimeEvent("fallback_deactivated", { transport: "sse" }); - } - this.setState("connected"); - emitRealtimeEvent("connected", { transport: "sse" }); - }; - this.source.onmessage = (event) => this.receive(event.data); - this.source.onerror = () => { - this.source?.close(); - this.source = null; - this.scheduleReconnect(); - }; - } else { - if (typeof WebSocket === "undefined") { - this.startPolling(endpoint); - return; - } - this.socket = customTransport ? transportFactory(endpoint) : new WebSocket(endpoint); - this.socket.addEventListener("open", () => { - this.attempt = 0; - if (this.fallbackActive) { - this.fallbackActive = false; - emitRealtimeEvent("fallback_deactivated", { transport: "ws" }); - } - this.setState("connected"); - emitRealtimeEvent("connected", { transport: "ws" }); - for (const t of this.wireTopics) this.send({ action: "subscribe", topic: t }); - }); - this.socket.addEventListener("message", (event) => this.receive(event.data)); - this.socket.addEventListener("close", () => { - this.socket = null; - this.correlator.rejectAll(); - if (!this.closedByUser) { - this.setState("reconnecting"); - this.scheduleReconnect(); - } else { - this.setState("idle"); - } - }); - this.socket.addEventListener("error", () => this.socket?.close()); - } - } catch { - this.scheduleReconnect(); - } - } - - private receive(frame: unknown): void { - try { - const message = decodeRealtimeFrame(frame); - if (message.sequence !== undefined) { - const previous = this.lastSequenceByTopic.get(message.topic); - if (previous !== undefined && message.sequence <= previous) return; - this.lastSequenceByTopic.set(message.topic, message.sequence); - } - this.handlers.get(message.topic)?.forEach((_count, handler) => handler(message.payload)); - if ( - typeof message.topic === "string" && - message.payload !== null && - typeof message.payload === "object" - ) { - realtimeBus.publish( - message.topic as RealtimeTopic, - message.payload as RealtimeTopics[RealtimeTopic] - ); - } - } catch (error) { - if (error instanceof RealtimeDecodeError) - emitRealtimeEvent("subscription_failure", { reason: error.code }); - } - } - - private startPolling(endpoint: string): void { - this.activeTransport = "rest"; - this.fallbackActive = true; - emitRealtimeEvent("fallback_activated", { transport: "rest" }); - const httpEndpoint = endpoint.replace(/^wss:/, "https:").replace(/^ws:/, "http:"); - const poll = async () => { - for (const topic of this.handlers.keys()) { - try { - const response = await fetch( - `${httpEndpoint.replace(/\/ws$/, "")}/realtime/${encodeURIComponent(topic)}` - ); - if (response.ok) { - const payload = await response.json(); - this.handlers.get(topic)?.forEach((_count, handler) => handler(payload)); - } - } catch { - /* REST fallback is best effort. */ - } - } - }; - void poll(); - this.pollTimer = setInterval(() => void poll(), FALLBACK_POLL_MS); - } - - private scheduleReconnect(): void { - if (this.reconnectTimer) return; - const delay = Math.min(BASE_BACKOFF_MS * 2 ** this.attempt++, MAX_BACKOFF_MS); - this.reconnectTimer = setTimeout(() => { - this.reconnectTimer = null; - if (!this.closedByUser) this.open(); - }, delay); - } - - subscribe(topic: string, handler: Handler): () => void { - let listeners = this.handlers.get(topic); - if (!listeners) { - listeners = new Map(); - this.handlers.set(topic, listeners); - emitRealtimeEvent("topic_subscribed", { topic }); - } - listeners.set(handler, (listeners.get(handler) ?? 0) + 1); - const ticket = ++this.teardownTicket; - if (!this.socket && !this.source && !this.pollTimer && !this.reconnectTimer) void this.open(); - let active = true; - return () => { - if (!active) return; - active = false; - const current = this.handlers.get(topic); - const count = current?.get(handler) ?? 0; - if (count <= 1) current?.delete(handler); - else current?.set(handler, count - 1); - if (current?.size === 0) { - this.handlers.delete(topic); - emitRealtimeEvent("topic_unsubscribed", { topic }); - } - // A microtask lets StrictMode's immediate remount reuse the active connection. - queueMicrotask(() => { - if (this.handlers.size === 0 && this.teardownTicket === ticket) this.disconnect(); - }); - }; - } - - /** One-shot request/response over the open socket; rejects if not connected. */ - request(topic: string, payload?: unknown, options?: RequestOptions): Promise { - return this.correlator.request( - (frame) => { - if (!this.socket) throw new Error("Realtime socket is not connected"); - this.socket.send(frame); - }, - topic, - payload, - options, - ); - } - - /** Test-only: drop connection, handlers and reconnect state. */ - __reset(): void { - this.disconnect(); - this.handlers.clear(); - this.attempt = 0; - this.closedByUser = false; - } - - disconnect(): void { - this.closedByUser = true; - this.correlator.rejectAll(); - if (this.reconnectTimer) clearTimeout(this.reconnectTimer); - if (this.pollTimer) clearInterval(this.pollTimer); - this.reconnectTimer = null; - this.pollTimer = null; - this.source?.close(); - this.source = null; - this.socket?.close(); - this.socket = null; - this.setState("idle"); - } - - refreshConfiguration(): void { - this.disconnect(); - if (this.handlers.size > 0) void this.open(); - } - - getStatus() { - return { transport: this.activeTransport, override: readRealtimeOverride() }; - } -} - -export const realtimeClient = new RealtimeClient(); diff --git a/src/lib/realtime/devProfiler.test.ts b/src/lib/realtime/devProfiler.test.ts new file mode 100644 index 0000000..eafcdf5 --- /dev/null +++ b/src/lib/realtime/devProfiler.test.ts @@ -0,0 +1,10 @@ +import { describe, expect, it } from "vitest"; +import { shouldProfileRealtime } from "./devProfiler"; + +describe("realtime profiler production guard", () => { + it("disables measurement for production bundles", async () => { + expect(shouldProfileRealtime("production")).toBe(false); + expect(shouldProfileRealtime("development", "1")).toBe(true); + expect(shouldProfileRealtime("development", undefined)).toBe(false); + }); +}); diff --git a/src/lib/realtime/devProfiler.ts b/src/lib/realtime/devProfiler.ts new file mode 100644 index 0000000..d0d5b02 --- /dev/null +++ b/src/lib/realtime/devProfiler.ts @@ -0,0 +1,23 @@ +"use client"; +import React, { Profiler, type ProfilerOnRenderCallback } from "react"; + +export const shouldProfileRealtime = ( + environment = process.env.NODE_ENV, + enabledFlag = process.env.NEXT_PUBLIC_REALTIME_PROFILER, +) => environment !== "production" && enabledFlag === "1"; +const enabled = shouldProfileRealtime(); +const totals = new Map(); +const onRender: ProfilerOnRenderCallback = (topic, _phase, actualDuration) => { + const item = totals.get(topic) ?? { commits: 0, durations: [] }; + item.commits += 1; item.durations.push(actualDuration); totals.set(topic, item); + if (enabled && typeof window !== "undefined") { + (window as Window & { __REFRACT_REALTIME_PROFILE__?: unknown }).__REFRACT_REALTIME_PROFILE__ = getRealtimeProfile(); + } +}; +export function RealtimeProfiler({ topic, children }: { topic: string; children: React.ReactNode }) { + return enabled ? React.createElement(Profiler, { id: topic, onRender }, children) : React.createElement(React.Fragment, null, children); +} +export function getRealtimeProfile() { + return Object.fromEntries([...totals].map(([topic, result]) => [topic, { commits: result.commits, durations: [...result.durations] }])); +} +export function resetRealtimeProfile() { totals.clear(); } diff --git a/src/lib/realtime/snapshotCache.test.ts b/src/lib/realtime/snapshotCache.test.ts new file mode 100644 index 0000000..9becd55 --- /dev/null +++ b/src/lib/realtime/snapshotCache.test.ts @@ -0,0 +1,18 @@ +import { describe, expect, it } from "vitest"; +import { readSnapshot, writeSnapshot } from "./snapshotCache"; + +describe("realtime snapshot cache", () => { + it("keeps wallet and network snapshots isolated and marks cached values stale", async () => { + writeSnapshot("oracle_update", "testnet", { value: 1 }, "GONE"); + writeSnapshot("oracle_update", "testnet", { value: 2 }, "GTWO"); + writeSnapshot("oracle_update", "mainnet", { value: 3 }, "GONE"); + expect(await readSnapshot("oracle_update", "testnet", "GONE")).toMatchObject({ value: { value: 1 }, stale: true }); + expect(await readSnapshot("oracle_update", "testnet", "GTWO")).toMatchObject({ value: { value: 2 } }); + expect(await readSnapshot("oracle_update", "mainnet", "GONE")).toMatchObject({ value: { value: 3 } }); + }); + + it("discards a value written with an older topic schema", async () => { + writeSnapshot("versioned_topic", "testnet", { old: true }, "GVERSION", 1); + expect(await readSnapshot("versioned_topic", "testnet", "GVERSION", 2)).toBeNull(); + }); +}); diff --git a/src/lib/realtime/snapshotCache.ts b/src/lib/realtime/snapshotCache.ts new file mode 100644 index 0000000..0596202 --- /dev/null +++ b/src/lib/realtime/snapshotCache.ts @@ -0,0 +1,131 @@ +"use client"; + +export interface Snapshot { value: T; capturedAt: number; version: number; stale: true } +const DB_NAME = "refract-realtime-snapshots"; +const STORE = "snapshots"; +const VERSION = 1; +const pending = new Map; timer: ReturnType }>(); +const memory = new Map>(); + +function scopedKey(topic: string, network: string, wallet = "public") { + return `${network}:${wallet.toLowerCase()}:${topic}`; +} +const mirrorKey = (key: string) => `refract:realtime:${key}`; + +/** A tiny synchronous seed mirror lets client renders paint before IndexedDB opens. */ +export function readSnapshotSync(topic: string, network: string, wallet?: string, version = VERSION): Snapshot | null { + if (typeof window === "undefined") return null; + const key = scopedKey(topic, network, wallet); + const warmed = memory.get(key) as Snapshot | undefined; + if (warmed) return warmed.version === version ? { ...warmed, stale: true } : null; + try { + const raw = window.localStorage.getItem(mirrorKey(key)); + if (!raw) return null; + const entry = JSON.parse(raw) as Snapshot; + if (entry.version !== version) { + window.localStorage.removeItem(mirrorKey(key)); + return null; + } + memory.set(key, entry); + return { ...entry, stale: true }; + } catch { return null; } +} +function openDb(): Promise { + if (typeof window === "undefined" || !("indexedDB" in window)) return Promise.resolve(null); + return new Promise((resolve) => { + try { + const request = indexedDB.open(DB_NAME, 1); + request.onupgradeneeded = () => { + if (!request.result.objectStoreNames.contains(STORE)) request.result.createObjectStore(STORE, { keyPath: "key" }); + }; + request.onsuccess = () => resolve(request.result); + request.onerror = request.onblocked = () => resolve(null); + } catch { resolve(null); } + }); +} + +export async function readSnapshot(topic: string, network: string, wallet?: string, version = VERSION): Promise | null> { + if (typeof window === "undefined") return null; + const key = scopedKey(topic, network, wallet); + let value: Snapshot | null = (memory.get(key) as Snapshot | undefined) ?? null; + const db = await openDb(); + if (db) { + const persisted = await new Promise | null>((resolve) => { + try { + const req = db.transaction(STORE).objectStore(STORE).get(key); + req.onsuccess = () => { + const stored = req.result; + const entry = stored?.entry as Snapshot | undefined; + if (entry && entry.version !== version) { + try { db.transaction(STORE, "readwrite").objectStore(STORE).delete(key); } catch { /* optional cache */ } + memory.delete(key); + resolve(null); + } else resolve(entry ?? null); + }; + req.onerror = () => resolve(null); + } catch { resolve(null); } + }); + if (!value || (persisted && persisted.capturedAt > value.capturedAt)) value = persisted; + } + if (value && value.version !== version) { await deleteSnapshot(topic, network, wallet); return null; } + return value ? { ...value, stale: true } : null; +} + +export function writeSnapshot(topic: string, network: string, value: T, wallet?: string, version = VERSION): void { + if (typeof window === "undefined") return; + const key = scopedKey(topic, network, wallet); + const entry: Snapshot = { value, capturedAt: Date.now(), version, stale: true }; + memory.set(key, entry); + try { window.localStorage.setItem(mirrorKey(key), JSON.stringify(entry)); } catch { /* optional sync seed */ } + const previous = pending.get(key); + if (previous) { clearTimeout(previous.timer); previous.entry = entry; previous.timer = setTimeout(() => void flush(key), 300); } + else pending.set(key, { entry, timer: setTimeout(() => void flush(key), 300) }); +} + +async function flush(key: string) { + const item = pending.get(key); if (!item) return; + pending.delete(key); + const db = await openDb(); if (!db) return; + try { + await new Promise((resolve, reject) => { + const tx = db.transaction(STORE, "readwrite"); + tx.objectStore(STORE).put({ key, entry: item.entry }); + tx.oncomplete = () => resolve(); + tx.onerror = tx.onabort = () => reject(tx.error ?? new Error("Snapshot write failed")); + }); + } catch { + // Quota pressure: remove the oldest entry, then retry once. Cache failure + // must never affect the realtime data path. + try { + const tx = db.transaction(STORE, "readwrite"); + const store = tx.objectStore(STORE); + const cursorRequest = store.openCursor(); + cursorRequest.onsuccess = () => { + const cursor = cursorRequest.result; + if (cursor) { + const first = { key: cursor.key, capturedAt: cursor.value?.entry?.capturedAt ?? 0 }; + let oldest = first; + const scan = store.openCursor(); + scan.onsuccess = () => { + const next = scan.result; + if (next) { + const capturedAt = next.value?.entry?.capturedAt ?? 0; + if (capturedAt < oldest.capturedAt) oldest = { key: next.key, capturedAt }; + next.continue(); + } else { + store.delete(oldest.key); + store.put({ key, entry: item.entry }); + } + }; + } + }; + } catch { /* storage is optional */ } + } +} + +export async function deleteSnapshot(topic: string, network: string, wallet?: string): Promise { + const key = scopedKey(topic, network, wallet); memory.delete(key); + try { window.localStorage.removeItem(mirrorKey(key)); } catch { /* optional cache */ } + const db = await openDb(); if (!db) return; + try { db.transaction(STORE, "readwrite").objectStore(STORE).delete(key); } catch { /* storage is optional */ } +} diff --git a/vitest.config.mts b/vitest.config.mts index 36d5414..e69de29 100644 --- a/vitest.config.mts +++ b/vitest.config.mts @@ -1,10 +0,0 @@ -import { fileURLToPath } from "node:url"; -import { defineConfig } from "vitest/config"; - -export default defineConfig({ - resolve: { alias: { "@": fileURLToPath(new URL("./src", import.meta.url)) } }, - test: { - include: ["src/**/*.test.ts"], - setupFiles: ["src/test/realtime/setup.ts"], - }, -});