From 16aad3603a1e19cb0c59a85d57a388a2fabe3a20 Mon Sep 17 00:00:00 2001 From: driftsorbit Date: Wed, 21 Oct 2026 18:22:23 +0000 Subject: [PATCH 1/2] feat(realtime): add warm lifecycle and control handling Closes #263 Closes #262 Closes #261 Closes #260 --- .eslintrc.json | 2 +- src/app/dashboard/page.tsx | 2 +- src/app/layout.tsx | 19 ++ src/app/page.tsx | 2 +- src/app/providers.tsx | 74 ++++++- src/components/layout/Navbar.tsx | 28 +++ src/hooks/useOracleStatus.ts | 76 +++++++ src/hooks/useRealtimeTopic.ts | 28 +++ src/lib/api/claims.ts | 36 ++-- src/lib/api/oracle.ts | 4 +- src/lib/domain/realtime.ts | 29 +++ src/lib/entities/policies.ts | 2 +- src/lib/export/exportPolicyHistory.ts | 2 +- src/lib/fixtures/claims.ts | 2 +- src/lib/fixtures/oracle.ts | 2 +- src/lib/fixtures/recentClaims.ts | 9 +- src/lib/format.ts | 5 +- src/lib/notifications.test.ts | 8 +- src/lib/notifications.ts | 20 +- src/lib/portfolio/selectors.ts | 2 +- src/lib/realtime/README.md | 15 ++ src/lib/realtime/client.test.ts | 158 ++++++++++++++ src/lib/realtime/client.ts | 277 +++++++++++++++++++++---- src/lib/realtime/clock.ts | 18 ++ src/lib/realtime/contract.ts | 25 +++ src/lib/realtime/mappers/index.test.ts | 34 +++ src/lib/realtime/mappers/index.ts | 123 +++++++++++ src/lib/realtime/preferences.ts | 30 +++ src/lib/realtime/testClock.ts | 10 + src/mocks/realtimeServer.ts | 57 +++++ vitest.config.mts | 5 +- vitest.config.ts | 0 32 files changed, 1019 insertions(+), 85 deletions(-) create mode 100644 src/hooks/useRealtimeTopic.ts create mode 100644 src/lib/domain/realtime.ts create mode 100644 src/lib/realtime/README.md create mode 100644 src/lib/realtime/client.test.ts create mode 100644 src/lib/realtime/clock.ts create mode 100644 src/lib/realtime/contract.ts create mode 100644 src/lib/realtime/mappers/index.test.ts create mode 100644 src/lib/realtime/mappers/index.ts create mode 100644 src/lib/realtime/preferences.ts create mode 100644 src/lib/realtime/testClock.ts create mode 100644 src/mocks/realtimeServer.ts delete mode 100644 vitest.config.ts diff --git a/.eslintrc.json b/.eslintrc.json index 3cbf99e..9f8fe95 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/src/app/dashboard/page.tsx b/src/app/dashboard/page.tsx index 6bcaa5d..9b16884 100644 --- a/src/app/dashboard/page.tsx +++ b/src/app/dashboard/page.tsx @@ -15,7 +15,7 @@ import { downloadExport, } from "@/lib/export/exportPolicyHistory"; import type { Policy } from "@/lib/api/policies"; -import type { ClaimRecord } from "@/lib/api/claims"; +import type { ClaimRecord } from "@/lib/domain/realtime"; import { normalizePolicies, normalizeClaims, selectPolicyStatus, type PolicyStatus } from "@/lib/entities/policies"; const COVERAGE_ICONS = ["🪙", "📉", "🛡️", "🔐", "✈️"]; diff --git a/src/app/layout.tsx b/src/app/layout.tsx index e69de29..4f773be 100644 --- a/src/app/layout.tsx +++ b/src/app/layout.tsx @@ -0,0 +1,19 @@ +import type { Metadata } from "next"; +import "./globals.css"; +import { Providers } from "./providers"; + +export const metadata: Metadata = { + title: "Refract Protocol", + description: "Parametric coverage powered by live on-chain oracles.", +}; + +/** App shell. Realtime startup is owned by the hydrated Providers tree. */ +export default function RootLayout({ children }: Readonly<{ children: React.ReactNode }>) { + return ( + + + {children} + + + ); +} diff --git a/src/app/page.tsx b/src/app/page.tsx index 38753ab..d329ea8 100644 --- a/src/app/page.tsx +++ b/src/app/page.tsx @@ -8,7 +8,7 @@ import { useOracleStatus } from "@/hooks/useOracleStatus"; import { useRecentClaims } from "@/hooks/useRecentClaims"; import { usePoolStats } from "@/hooks/usePoolStats"; import { useClaimStats } from "@/hooks/useClaimStats"; -import type { OracleReading } from "@/lib/api/oracle"; +import type { OracleReading } from "@/lib/domain/realtime"; import type { BadgeTone } from "@/components/ui/Badge"; import { formatUsd, formatRelativeTime, fromStroops } from "@/lib/format"; import { truncateAddress } from "@/lib/wallet/WalletProvider"; diff --git a/src/app/providers.tsx b/src/app/providers.tsx index 653e41b..58caf25 100644 --- a/src/app/providers.tsx +++ b/src/app/providers.tsx @@ -1,8 +1,12 @@ "use client"; -import { useState } from "react"; +import { useEffect, useState } from "react"; import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; import { WalletProvider } from "@/lib/wallet/WalletProvider"; +import { ToastProvider, useToast } from "@/components/ui/ToastProvider"; +import { useFeatureFlag } from "@/lib/flags/useFeatureFlag"; +import { realtimeClient } from "@/lib/realtime/client"; +import { isRealtimeAllowed } from "@/lib/realtime/preferences"; export function Providers({ children }: { children: React.ReactNode }) { // One client per browser session. Query functions already fold failures @@ -13,7 +17,73 @@ export function Providers({ children }: { children: React.ReactNode }) { return ( - {children} + + + {children} + ); } + +/** Owns the one app-shell connection. All route hooks attach to this client. */ +function RealtimeLifecycle() { + const enabled = useFeatureFlag("realtime-oracle"); + const { toast } = useToast(); + const [preferenceRevision, setPreferenceRevision] = useState(0); + + useEffect(() => { + const onPreferenceChange = () => setPreferenceRevision((revision) => revision + 1); + window.addEventListener("refract-realtime-preference-change", onPreferenceChange); + return () => + window.removeEventListener("refract-realtime-preference-change", onPreferenceChange); + }, []); + + useEffect(() => { + if (!enabled || !isRealtimeAllowed()) return; + + // This runs after hydration, so it cannot delay server rendering or first paint. + realtimeClient.warm(); + const onControl = realtimeClient.subscribeControl((control) => { + if (control.type === "revocation") { + toast({ + title: "Live updates stopped", + description: control.reason, + tone: "danger", + duration: 0, + }); + } else if (control.type === "maintenance" || control.type === "drain") { + toast({ + title: "Live updates paused", + description: + control.reason ?? "The service is restarting. Updates will resume automatically.", + duration: 6000, + }); + } else if (control.type === "drain-complete") { + toast({ title: "Live updates resumed", duration: 3000 }); + } + }); + + // Warming is delegated so navigation links in the shell and route content + // both trigger it without each link taking ownership of the socket. + const warmOnIntent = (event: Event) => { + const target = event.target; + if (!(target instanceof Element)) return; + const link = target.closest("a[href]"); + if ( + link && + ["/cover", "/provide"].includes(new URL((link as HTMLAnchorElement).href).pathname) + ) + realtimeClient.warm(); + }; + document.addEventListener("pointerover", warmOnIntent, { passive: true }); + document.addEventListener("focusin", warmOnIntent); + return () => { + onControl(); + document.removeEventListener("pointerover", warmOnIntent); + document.removeEventListener("focusin", warmOnIntent); + realtimeClient.disconnect(); + }; + }, [enabled, preferenceRevision, toast]); + + return null; +} diff --git a/src/components/layout/Navbar.tsx b/src/components/layout/Navbar.tsx index e69de29..5410296 100644 --- a/src/components/layout/Navbar.tsx +++ b/src/components/layout/Navbar.tsx @@ -0,0 +1,28 @@ +"use client"; + +import Link from "next/link"; +import type { ReactNode } from "react"; + +export function Navbar({ right }: { right?: ReactNode }) { + return ( +
+ +
+ ); +} diff --git a/src/hooks/useOracleStatus.ts b/src/hooks/useOracleStatus.ts index e69de29..f3f49af 100644 --- a/src/hooks/useOracleStatus.ts +++ b/src/hooks/useOracleStatus.ts @@ -0,0 +1,76 @@ +"use client"; + +import { useCallback, useEffect, useState } from "react"; +import { fetchOracleStatus } from "@/lib/api/oracle"; +import type { OracleReading } from "@/lib/domain/realtime"; +import { FIXTURE_ORACLE_READINGS } from "@/lib/fixtures/oracle"; +import { useFeatureFlag } from "@/lib/flags/useFeatureFlag"; +import { mapOracleUpdate } from "@/lib/realtime/mappers"; +import { REALTIME_TOPICS } from "@/lib/realtime/contract"; +import { useRealtimeTopic } from "./useRealtimeTopic"; +import { realRealtimeClock } from "@/lib/realtime/clock"; +import { isRealtimeAllowed } from "@/lib/realtime/preferences"; + +export interface OracleStatusState { + data: OracleReading[]; + lastUpdated: Date | null; + isRefreshing: boolean; + isFixture: boolean; + refresh: () => Promise; +} + +/** REST remains the initial source of truth; mapped stream updates replace one reading by coverage type. */ +export function useOracleStatus(): OracleStatusState { + const realtimeEnabled = useFeatureFlag("realtime-oracle"); + const [preferenceRevision, setPreferenceRevision] = useState(0); + const [data, setData] = useState(FIXTURE_ORACLE_READINGS); + const [lastUpdated, setLastUpdated] = useState(null); + const [isRefreshing, setIsRefreshing] = useState(false); + const [isFixture, setIsFixture] = useState(true); + + useEffect(() => { + const onPreferenceChange = () => setPreferenceRevision((revision) => revision + 1); + window.addEventListener("refract-realtime-preference-change", onPreferenceChange); + return () => + window.removeEventListener("refract-realtime-preference-change", onPreferenceChange); + }, []); + + const refresh = useCallback(async () => { + setIsRefreshing(true); + try { + const result = await fetchOracleStatus(); + setData(result.readings); + setLastUpdated(new Date(realRealtimeClock.now())); + setIsFixture(false); + } catch { + setData(FIXTURE_ORACLE_READINGS); + setIsFixture(true); + } finally { + setIsRefreshing(false); + } + }, []); + + useEffect(() => { + void refresh(); + const timer = realRealtimeClock.setInterval(() => void refresh(), 60_000); + return () => realRealtimeClock.clearInterval(timer); + }, [refresh]); + + const streamed = useRealtimeTopic( + REALTIME_TOPICS.oracle, + mapOracleUpdate, + undefined, + realtimeEnabled && isRealtimeAllowed() + ); + useEffect(() => { + if (!realtimeEnabled || !isRealtimeAllowed() || !streamed) return; + setData((current) => [ + ...current.filter((item) => item.coverageType !== streamed.coverageType), + streamed, + ]); + setLastUpdated(new Date(realRealtimeClock.now())); + setIsFixture(false); + }, [preferenceRevision, realtimeEnabled, streamed]); + + return { data, lastUpdated, isRefreshing, isFixture, refresh }; +} diff --git a/src/hooks/useRealtimeTopic.ts b/src/hooks/useRealtimeTopic.ts new file mode 100644 index 0000000..fce10d6 --- /dev/null +++ b/src/hooks/useRealtimeTopic.ts @@ -0,0 +1,28 @@ +"use client"; + +import { useEffect, useState } from "react"; +import { realtimeClient } from "@/lib/realtime/client"; +import type { MappingResult } from "@/lib/realtime/mappers"; + +/** Subscribe to a wire topic while exposing only its mapped domain value. */ +export function useRealtimeTopic( + topic: string, + map: (payload: unknown) => MappingResult, + onValue?: (value: T) => void, + enabled = true +): T | null { + const [value, setValue] = useState(null); + useEffect(() => { + if (!enabled) return; + return realtimeClient.subscribe(topic, (payload) => { + const result = map(payload); + if (!result.ok) { + console.warn(`[realtime] Dropped ${topic} payload: ${result.reason}`, payload); + return; + } + setValue(result.value); + onValue?.(result.value); + }); + }, [topic, map, onValue, enabled]); + return value; +} diff --git a/src/lib/api/claims.ts b/src/lib/api/claims.ts index ddbdd2f..1234eb2 100644 --- a/src/lib/api/claims.ts +++ b/src/lib/api/claims.ts @@ -1,8 +1,9 @@ import { apiRequest } from "./client"; import type { ApiSchemas } from "./generated"; +import type { ClaimRecord } from "@/lib/domain/realtime"; /** Types are generated from openapi/refract-api.yaml, which mirrors ClaimResult from refract-backend/src/claim/claim-result.ts. */ -export type ClaimRecord = ApiSchemas["ClaimRecord"]; +export type { ClaimRecord } from "@/lib/domain/realtime"; /** * Assumed pagination contract (to be coordinated with refract-backend): @@ -22,6 +23,13 @@ export interface PaginatedResponse { hasMore: boolean; } +interface ClaimsEndpointResponse { + claims?: ApiSchemas["ClaimRecord"][]; + items?: ApiSchemas["ClaimRecord"][]; + nextCursor?: string; + hasMore?: boolean; +} + function buildPaginationQuery({ cursor, limit }: PaginationParams = {}): string { const params = new URLSearchParams(); if (cursor) params.set("cursor", cursor); @@ -34,10 +42,11 @@ 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 { - const items = response.items ?? response.claims ?? []; +function normalizeClaimsResponse(response: ClaimsEndpointResponse): PaginatedResponse { + const items = (response.items ?? response.claims ?? []).map((claim) => ({ + ...claim, + processedAt: claim.processedAt * 1000, + })); const hasMore = response.hasMore ?? false; return { items, nextCursor: response.nextCursor, hasMore }; } @@ -45,20 +54,21 @@ 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/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/recent${buildPaginationQuery(params)}`, { + signal, + }).then(normalizeClaimsResponse); } /** Types are generated from openapi/refract-api.yaml, which mirrors ClaimService.getStats()'s response in refract-backend. */ diff --git a/src/lib/api/oracle.ts b/src/lib/api/oracle.ts index c1b0eaa..ba680c3 100644 --- a/src/lib/api/oracle.ts +++ b/src/lib/api/oracle.ts @@ -1,8 +1,8 @@ import { apiRequest } from "./client"; -import type { ApiSchemas } from "./generated"; +import type { OracleReading } from "@/lib/domain/realtime"; /** Types are generated from openapi/refract-api.yaml, which mirrors OracleReading from refract-backend/src/oracle/oracle-reading.ts. */ -export type OracleReading = ApiSchemas["OracleReading"]; +export type { OracleReading } from "@/lib/domain/realtime"; export function fetchOracleStatus(signal?: AbortSignal): Promise<{ readings: OracleReading[] }> { return apiRequest("/oracle/status", { signal }); diff --git a/src/lib/domain/realtime.ts b/src/lib/domain/realtime.ts new file mode 100644 index 0000000..8ecd784 --- /dev/null +++ b/src/lib/domain/realtime.ts @@ -0,0 +1,29 @@ +/** Stable frontend shapes shared by REST responses and realtime events. */ +export interface OracleReading { + coverageType: string; + type: "oracle_update"; + value: number; + threshold: number; + severity: "low" | "medium" | "high" | "triggered"; + message: string; +} + +export interface ClaimRecord { + policyId: string; + holder: string; + coverageType: number; + triggered: boolean; + /** Base-unit amount, kept as a string to preserve integer precision. */ + payout: string; + reason: string; + /** Unix timestamp in milliseconds, the shared frontend representation. */ + processedAt: number; + settlementTxHash?: string; +} + +export interface NotificationEvent { + id: string; + message: string; + /** Unix timestamp in milliseconds. */ + timestamp: number; +} diff --git a/src/lib/entities/policies.ts b/src/lib/entities/policies.ts index 8f5bf9f..0258b2b 100644 --- a/src/lib/entities/policies.ts +++ b/src/lib/entities/policies.ts @@ -1,5 +1,5 @@ import type { Policy } from "@/lib/api/policies"; -import type { ClaimRecord } from "@/lib/api/claims"; +import type { ClaimRecord } from "@/lib/domain/realtime"; export type PolicyStatus = "active" | "paid" | "expired"; diff --git a/src/lib/export/exportPolicyHistory.ts b/src/lib/export/exportPolicyHistory.ts index f67a87c..0feac3b 100644 --- a/src/lib/export/exportPolicyHistory.ts +++ b/src/lib/export/exportPolicyHistory.ts @@ -1,6 +1,6 @@ import { fromStroops, formatUsd } from "@/lib/format"; import type { Policy } from "@/lib/api/policies"; -import type { ClaimRecord } from "@/lib/api/claims"; +import type { ClaimRecord } from "@/lib/domain/realtime"; export interface ExportDataset { address: string; diff --git a/src/lib/fixtures/claims.ts b/src/lib/fixtures/claims.ts index 1e10ffc..f1cddbf 100644 --- a/src/lib/fixtures/claims.ts +++ b/src/lib/fixtures/claims.ts @@ -1,4 +1,4 @@ -import type { ClaimRecord } from "@/lib/api/claims"; +import type { ClaimRecord } from "@/lib/domain/realtime"; import type { Policy } from "@/lib/api/policies"; /** diff --git a/src/lib/fixtures/oracle.ts b/src/lib/fixtures/oracle.ts index ef24655..c062b2e 100644 --- a/src/lib/fixtures/oracle.ts +++ b/src/lib/fixtures/oracle.ts @@ -1,4 +1,4 @@ -import type { OracleReading } from "@/lib/api/oracle"; +import type { OracleReading } from "@/lib/domain/realtime"; /** * FIXTURE — not live data. diff --git a/src/lib/fixtures/recentClaims.ts b/src/lib/fixtures/recentClaims.ts index 96be662..c5930fe 100644 --- a/src/lib/fixtures/recentClaims.ts +++ b/src/lib/fixtures/recentClaims.ts @@ -1,4 +1,5 @@ -import type { ClaimRecord } from "@/lib/api/claims"; +import type { ClaimRecord } from "@/lib/domain/realtime"; +import { USDC_BASE_UNIT } from "@/lib/format"; /** * FIXTURE — not live data. @@ -18,7 +19,7 @@ export const FIXTURE_RECENT_CLAIMS: ClaimRecord[] = [ holder: "GDEMOHOLDERADDRESSAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", coverageType: 0, triggered: true, - payout: (48_200 * 1e7).toString(), + payout: (48_200 * USDC_BASE_UNIT).toString(), reason: "USDC price: $0.9412 (-5.88% from peg)", processedAt: Date.now() - 3 * 86_400_000, }, @@ -27,7 +28,7 @@ export const FIXTURE_RECENT_CLAIMS: ClaimRecord[] = [ holder: "GDEMOHOLDERADDRESSBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB", coverageType: 1, triggered: true, - payout: (120_000 * 1e7).toString(), + payout: (120_000 * USDC_BASE_UNIT).toString(), reason: "XLM 24h change: -34.20% (trigger at -30%)", processedAt: Date.now() - 11 * 86_400_000, }, @@ -36,7 +37,7 @@ export const FIXTURE_RECENT_CLAIMS: ClaimRecord[] = [ holder: "GDEMOHOLDERADDRESSCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC", coverageType: 4, triggered: true, - payout: (1_400 * 1e7).toString(), + payout: (1_400 * USDC_BASE_UNIT).toString(), reason: "Flight BA249: 145m delay (trigger at 120m)", processedAt: Date.now() - 14 * 86_400_000, }, diff --git a/src/lib/format.ts b/src/lib/format.ts index 6c0de25..042acde 100644 --- a/src/lib/format.ts +++ b/src/lib/format.ts @@ -6,6 +6,7 @@ import { usePreferencesStore } from "@/lib/store/preferencesStore"; * Number of decimal places USDC supports (7-decimal base units / stroops). */ export const USDC_DECIMALS = 7; +export const USDC_BASE_UNIT = 10 ** USDC_DECIMALS; /** * Parse a user-entered amount string into a finite number. @@ -32,7 +33,7 @@ export function toStroops(amount: number): bigint { if (!Number.isFinite(amount)) { throw new TypeError(`toStroops: expected a finite number, received ${amount}`); } - return BigInt(Math.round(amount * 10 ** USDC_DECIMALS)); + return BigInt(Math.round(amount * USDC_BASE_UNIT)); } /** @@ -45,7 +46,7 @@ export const MIN_DEPOSIT_USDC = 1; /** Converts a base-unit string (1e7 per USDC) back to a human float. */ export function fromStroops(value: string | number): number { - return Number(value) / 10 ** USDC_DECIMALS; + return Number(value) / USDC_BASE_UNIT; } /** diff --git a/src/lib/notifications.test.ts b/src/lib/notifications.test.ts index 0f5146a..85db87f 100644 --- a/src/lib/notifications.test.ts +++ b/src/lib/notifications.test.ts @@ -1,7 +1,7 @@ import { describe, it, expect } from "vitest"; import { deriveNotifications, ONE_DAY_MS, SEVEN_DAYS_MS } from "./notifications"; import type { Policy } from "./api/policies"; -import type { ClaimRecord } from "./api/claims"; +import type { ClaimRecord } from "./domain/realtime"; import type { OracleReading } from "./api/oracle"; describe("deriveNotifications", () => { @@ -51,7 +51,7 @@ describe("deriveNotifications", () => { triggered: true, payout: "50000000000", reason: "USDC depegged below $0.98", - processedAt: Math.floor((now - 3600000) / 1000), + processedAt: now - 3600000, }, ]; @@ -89,11 +89,11 @@ describe("deriveNotifications", () => { triggered: true, payout: "50000000000", reason: "USDC depegged", - processedAt: Math.floor(now / 1000), + processedAt: now, }, ]; - const readIds = new Set([`claim-pol-1-${Math.floor(now / 1000)}`]); + const readIds = new Set([`claim-pol-1-${now}`]); const notifs = deriveNotifications({ claims: mockClaims, now, readIds }); expect(notifs[0].read).toBe(true); }); diff --git a/src/lib/notifications.ts b/src/lib/notifications.ts index 8ccaf83..7b50614 100644 --- a/src/lib/notifications.ts +++ b/src/lib/notifications.ts @@ -1,14 +1,10 @@ import type { Policy } from "./api/policies"; -import type { ClaimRecord } from "./api/claims"; -import type { OracleReading } from "./api/oracle"; +import type { ClaimRecord } from "./domain/realtime"; +import type { OracleReading } from "./domain/realtime"; import { formatUsd, fromStroops } from "./format"; export type NotificationType = - | "policy_expiring" - | "claim_payout" - | "lockup_expired" - | "lockup_soon" - | "oracle_trigger"; + "policy_expiring" | "claim_payout" | "lockup_expired" | "lockup_soon" | "oracle_trigger"; export interface RefractNotification { id: string; @@ -80,7 +76,7 @@ export function deriveNotifications({ type: "claim_payout", title: `Claim Payout Disbursed`, message: `${payoutFormatted} was automatically disbursed for policy ${claim.policyId.slice(0, 8)}… (${claim.reason}).`, - timestamp: claim.processedAt * 1000, + timestamp: claim.processedAt, read: readIds.has(id), href: "/dashboard", severity: "success", @@ -98,7 +94,8 @@ export function deriveNotifications({ id, type: "lockup_expired", title: "Capital Lockup Ended", - message: "Your liquidity pool deposit lockup has ended. Funds are now available for withdrawal.", + message: + "Your liquidity pool deposit lockup has ended. Funds are now available for withdrawal.", timestamp: lockupMs, read: readIds.has(id), href: "/provide", @@ -127,7 +124,10 @@ export function deriveNotifications({ notifications.push({ id, type: "oracle_trigger", - title: reading.severity === "triggered" ? `Oracle Trigger Activated` : `Oracle Alert: Elevated Risk`, + title: + reading.severity === "triggered" + ? `Oracle Trigger Activated` + : `Oracle Alert: Elevated Risk`, message: `${reading.coverageType}: ${reading.message}`, timestamp: now, read: readIds.has(id), diff --git a/src/lib/portfolio/selectors.ts b/src/lib/portfolio/selectors.ts index a6d9f66..c8fd0a6 100644 --- a/src/lib/portfolio/selectors.ts +++ b/src/lib/portfolio/selectors.ts @@ -7,7 +7,7 @@ */ import type { Policy } from "@/lib/api/policies"; -import type { ClaimRecord } from "@/lib/api/claims"; +import type { ClaimRecord } from "@/lib/domain/realtime"; import { fromStroops } from "@/lib/format"; export type PolicyStatus = "active" | "expired" | "paid"; diff --git a/src/lib/realtime/README.md b/src/lib/realtime/README.md new file mode 100644 index 0000000..d873e01 --- /dev/null +++ b/src/lib/realtime/README.md @@ -0,0 +1,15 @@ +# Realtime contract + +The browser connects to `/ws` after the app has hydrated when `realtime-oracle` is enabled. The shell checks `navigator.connection.saveData`, `(prefers-reduced-data: reduce)`, and the `setRealtimeOptOut` preference before warming. Hovering or focusing a link to `/cover` or `/provide` also warms the shared client. A warm connection with no topic subscribers closes after 15 seconds. Hooks only subscribe to the shared client and map each wire payload to a REST-shaped domain value. + +Domain frames have the shape `{ "topic": "oracle_update", "payload": ... }`, `{ "topic": "claim_update", "payload": ... }`, or `{ "topic": "notification_update", "payload": ... }`. The claim mapper converts the backend Unix seconds to the shared frontend Unix milliseconds used by REST-derived data. Claim amounts remain base-unit strings. Unknown, malformed, or unexpected payloads are dropped and reported to the console. + +The reserved `$control` topic never reaches domain handlers. Its payload types are: + +- `{ "type": "ping" }` / `{ "type": "pong" }`: the client sends a ping every 30 seconds while a topic is subscribed; the server must answer with pong within 10 seconds or the client reconnects. +- `{ "type": "revocation", "topic": "oracle_update", "reason": "..." }`: remove the topic handlers and show the reason; that topic is not retried. +- `{ "type": "maintenance", "until": 1792600000000, "reason": "..." }`: postpone a reconnect until the later of the maintenance deadline and current backoff. +- `{ "type": "drain", "until": 1792600000000, "reason": "..." }`: same reconnect behavior during graceful server drain. +- `{ "type": "drain-complete" }`: clear the maintenance deadline and reconnect immediately if topics remain subscribed. + +`RealtimeClient` accepts an injected clock and socket factory. Realtime unit tests use Vitest fake timers through `fakeRealtimeClock`; browser tests should wait on observable UI state rather than sleeping for an assumed duration. diff --git a/src/lib/realtime/client.test.ts b/src/lib/realtime/client.test.ts new file mode 100644 index 0000000..57f1f8d --- /dev/null +++ b/src/lib/realtime/client.test.ts @@ -0,0 +1,158 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { RealtimeClient } from "./client"; +import { fakeRealtimeClock } from "./testClock"; +import { createMockRealtimeServer } from "@/mocks/realtimeServer"; + +describe("RealtimeClient lifecycle", () => { + beforeEach(() => { + vi.stubGlobal("window", {}); + vi.useFakeTimers(); + vi.setSystemTime(1_000_000); + }); + afterEach(() => { + vi.useRealTimers(); + vi.unstubAllGlobals(); + }); + + it("warms once, then an unused connection closes at the idle deadline", async () => { + const server = createMockRealtimeServer(); + const client = new RealtimeClient({ + clock: fakeRealtimeClock, + socketFactory: server.socketFactory, + idleTimeoutMs: 2_000, + }); + client.warm(); + expect(server.sockets).toHaveLength(1); + await vi.advanceTimersByTimeAsync(1_999); + expect(server.sockets[0].closed).toBe(false); + await vi.advanceTimersByTimeAsync(1); + expect(server.sockets[0].closed).toBe(true); + }); + + it("delivers the first event immediately after a warm subscribe and pays the handshake when cold", async () => { + const warmServer = createMockRealtimeServer(100); + const warmClient = new RealtimeClient({ + clock: fakeRealtimeClock, + socketFactory: warmServer.socketFactory, + }); + warmClient.warm(); + await vi.advanceTimersByTimeAsync(100); + const warmHandler = vi.fn(); + const warmSubscribeAt = Date.now(); + warmClient.subscribe("oracle_update", warmHandler); + warmServer.message(warmServer.sockets[0], "oracle_update", { ready: true }); + expect(warmHandler).toHaveBeenCalledTimes(1); + expect(Date.now() - warmSubscribeAt).toBe(0); + warmClient.disconnect(); + + const coldServer = createMockRealtimeServer(100); + const coldClient = new RealtimeClient({ + clock: fakeRealtimeClock, + socketFactory: coldServer.socketFactory, + }); + const coldHandler = vi.fn(); + const coldSubscribeAt = Date.now(); + coldClient.subscribe("oracle_update", coldHandler); + await vi.advanceTimersByTimeAsync(100); + coldServer.message(coldServer.sockets[0], "oracle_update", { ready: true }); + expect(coldHandler).toHaveBeenCalledTimes(1); + expect(Date.now() - coldSubscribeAt).toBe(100); + coldClient.disconnect(); + }); + + it("does no socket work during server rendering", () => { + vi.unstubAllGlobals(); + const socketFactory = vi.fn(); + const client = new RealtimeClient({ socketFactory }); + client.warm(); + expect(socketFactory).not.toHaveBeenCalled(); + }); + + it("uses deterministic exponential backoff and honors a later maintenance deadline", async () => { + const server = createMockRealtimeServer(); + const client = new RealtimeClient({ + clock: fakeRealtimeClock, + socketFactory: server.socketFactory, + }); + client.subscribe("oracle_update", () => {}); + await vi.advanceTimersByTimeAsync(0); + const first = server.sockets[0]; + server.control(first, { type: "maintenance", until: Date.now() + 1_500 }); + first.close(); + await vi.advanceTimersByTimeAsync(1_499); + expect(server.sockets).toHaveLength(1); + await vi.advanceTimersByTimeAsync(1); + expect(server.sockets).toHaveLength(2); + await vi.advanceTimersByTimeAsync(30_000); + expect(server.sockets).toHaveLength(2); + client.disconnect(); + }); + + it("sends clock-controlled heartbeats and clears the timeout on pong", async () => { + const server = createMockRealtimeServer(); + const client = new RealtimeClient({ + clock: fakeRealtimeClock, + socketFactory: server.socketFactory, + heartbeatIntervalMs: 1_000, + heartbeatTimeoutMs: 500, + }); + client.subscribe("oracle_update", () => {}); + await vi.advanceTimersByTimeAsync(0); + await vi.advanceTimersByTimeAsync(1_000); + expect(JSON.parse(server.sockets[0].sent[0])).toEqual({ + topic: "$control", + payload: { type: "ping" }, + }); + await vi.advanceTimersByTimeAsync(500); + expect(server.sockets[0].closed).toBe(false); + client.disconnect(); + }); + + it("removes revoked subscriptions, reports the reason, and never forwards controls", async () => { + const server = createMockRealtimeServer(); + const client = new RealtimeClient({ + clock: fakeRealtimeClock, + socketFactory: server.socketFactory, + }); + const domain = vi.fn(); + const control = vi.fn(); + const unsubscribe = client.subscribe("oracle_update", domain); + client.subscribeControl(control); + await vi.advanceTimersByTimeAsync(0); + const socket = server.sockets[0]; + server.control(socket, { + type: "revocation", + topic: "oracle_update", + reason: "Access expired", + }); + server.control(socket, { + type: "revocation", + topic: "oracle_update", + reason: "Access expired", + }); + server.message(socket, "oracle_update", { value: 5 }); + expect(domain).not.toHaveBeenCalled(); + expect(control).toHaveBeenCalledWith({ + type: "revocation", + topic: "oracle_update", + reason: "Access expired", + }); + expect(control).toHaveBeenCalledTimes(1); + unsubscribe(); + client.disconnect(); + }); + + it("reconnects immediately when the server signals drain complete", async () => { + const server = createMockRealtimeServer(); + const client = new RealtimeClient({ + clock: fakeRealtimeClock, + socketFactory: server.socketFactory, + }); + client.subscribe("oracle_update", () => {}); + await vi.advanceTimersByTimeAsync(0); + server.control(server.sockets[0], { type: "drain", until: Date.now() + 60_000 }); + server.control(server.sockets[0], { type: "drain-complete" }); + expect(server.sockets).toHaveLength(2); + client.disconnect(); + }); +}); diff --git a/src/lib/realtime/client.ts b/src/lib/realtime/client.ts index 8be8d3c..c61b6c0 100644 --- a/src/lib/realtime/client.ts +++ b/src/lib/realtime/client.ts @@ -1,86 +1,285 @@ import { API_BASE_URL } from "@/lib/api/client"; +import { REALTIME_CONTROL_TOPIC, type RealtimeControl, type RealtimeEnvelope } from "./contract"; +import { realRealtimeClock, type RealtimeClock } from "./clock"; type Handler = (payload: unknown) => void; +export interface RealtimeClientOptions { + clock?: RealtimeClock; + socketFactory?: (url: string) => WebSocket; + idleTimeoutMs?: number; + heartbeatIntervalMs?: number; + heartbeatTimeoutMs?: number; + onSubscriberError?: (topic: string, error: unknown) => void; +} const MAX_BACKOFF_MS = 30_000; const BASE_BACKOFF_MS = 1_000; +const DEFAULT_IDLE_TIMEOUT_MS = 15_000; +const DEFAULT_HEARTBEAT_INTERVAL_MS = 30_000; +const DEFAULT_HEARTBEAT_TIMEOUT_MS = 10_000; -/** - * A small reconnecting WebSocket wrapper exposing a subscribe-by-topic API. - * The backend contract this assumes (documented, not yet implemented - * upstream): connect to `${wsUrl}`, receive JSON messages shaped - * `{ topic: string, payload: unknown }`, one topic per push-based update - * type (e.g. "oracle_update"). Callers must treat this connection as - * best-effort — always keep an independent REST fetch as the source of - * truth for initial state. - */ -class RealtimeClient { +/** Browser WebSocket client. A shell owner may warm it before hooks subscribe; + * unused warm connections close after an idle window. */ +export class RealtimeClient { private socket: WebSocket | null = null; private handlers = new Map>(); + private controlHandlers = new Set<(control: RealtimeControl) => void>(); private attempt = 0; private closedByUser = false; private reconnectTimer: ReturnType | null = null; + private reconnectBackoffMs = 0; + private idleTimer: ReturnType | null = null; + private heartbeatTimer: ReturnType | null = null; + private heartbeatTimeoutTimer: ReturnType | null = null; + private heartbeatSocket: WebSocket | null = null; + private maintenanceUntil = 0; + private readonly intentionallyClosing = new WeakSet(); + private revokedTopics = new Set(); + private readonly clock: RealtimeClock; + private readonly socketFactory: (url: string) => WebSocket; + private readonly idleTimeoutMs: number; + private readonly heartbeatIntervalMs: number; + private readonly heartbeatTimeoutMs: number; + private readonly onSubscriberError?: RealtimeClientOptions["onSubscriberError"]; + + constructor(options: RealtimeClientOptions = {}) { + this.clock = options.clock ?? realRealtimeClock; + this.socketFactory = options.socketFactory ?? ((url) => new WebSocket(url)); + this.idleTimeoutMs = options.idleTimeoutMs ?? DEFAULT_IDLE_TIMEOUT_MS; + this.heartbeatIntervalMs = options.heartbeatIntervalMs ?? DEFAULT_HEARTBEAT_INTERVAL_MS; + this.heartbeatTimeoutMs = options.heartbeatTimeoutMs ?? DEFAULT_HEARTBEAT_TIMEOUT_MS; + this.onSubscriberError = options.onSubscriberError; + } connect(): void { - if (typeof WebSocket === "undefined" || this.socket) return; + if (typeof window === "undefined" || this.socket) return; this.closedByUser = false; this.open(); + this.scheduleIdleClose(); + } + + warm(): void { + this.connect(); } private open(): void { + if (typeof window === "undefined" || this.socket || this.closedByUser) return; const wsUrl = API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws"); try { - this.socket = new WebSocket(wsUrl); + this.socket = this.socketFactory(wsUrl); } catch { this.scheduleReconnect(); return; } - - this.socket.addEventListener("open", () => { + const socket = this.socket; + socket.addEventListener("open", () => { + if (this.socket !== socket) return; this.attempt = 0; + if (this.hasSubscribers()) this.startHeartbeat(socket); }); - this.socket.addEventListener("message", (event) => { - try { - const { topic, payload } = JSON.parse(event.data as string) as { topic: string; payload: unknown }; - this.handlers.get(topic)?.forEach((handler) => handler(payload)); - } catch { - // Malformed message — ignore, don't crash the connection. - } + socket.addEventListener("message", (event) => { + if (this.socket === socket) this.dispatch(event.data); }); - this.socket.addEventListener("close", () => { - this.socket = null; - if (!this.closedByUser) this.scheduleReconnect(); + socket.addEventListener("close", () => { + this.stopHeartbeat(socket); + if (this.socket === socket) this.socket = null; + if (this.intentionallyClosing.has(socket)) { + this.intentionallyClosing.delete(socket); + return; + } + if (!this.closedByUser && this.hasSubscribers()) this.scheduleReconnect(); + else if (!this.closedByUser) this.scheduleIdleClose(); }); - this.socket.addEventListener("error", () => { - this.socket?.close(); + socket.addEventListener("error", () => socket.close()); + } + + private dispatch(data: unknown): void { + let envelope: RealtimeEnvelope; + try { + const parsed: unknown = JSON.parse(String(data)); + if ( + !parsed || + typeof parsed !== "object" || + typeof (parsed as RealtimeEnvelope).topic !== "string" + ) + return; + envelope = parsed as RealtimeEnvelope; + } catch { + return; + } + if (envelope.topic === REALTIME_CONTROL_TOPIC) { + this.dispatchControl(envelope.payload); + return; + } + this.handlers.get(envelope.topic)?.forEach((handler) => { + try { + handler(envelope.payload); + } catch (error) { + this.onSubscriberError?.(envelope.topic, error); + } }); } - private scheduleReconnect(): void { - if (this.reconnectTimer) return; - const delay = Math.min(BASE_BACKOFF_MS * 2 ** this.attempt, MAX_BACKOFF_MS); - this.attempt += 1; - this.reconnectTimer = setTimeout(() => { + private dispatchControl(payload: unknown): void { + if ( + !payload || + typeof payload !== "object" || + typeof (payload as RealtimeControl).type !== "string" + ) + return; + const control = payload as RealtimeControl; + if (control.type === "ping") { + this.sendControl({ type: "pong" }); + return; + } + if (control.type === "pong") { + if (this.heartbeatTimeoutTimer) this.clock.clearTimeout(this.heartbeatTimeoutTimer); + this.heartbeatTimeoutTimer = null; + return; + } + if (control.type === "revocation") { + if (typeof control.topic !== "string" || typeof control.reason !== "string") return; + if (!this.handlers.has(control.topic)) return; + this.revokedTopics.add(control.topic); + this.handlers.delete(control.topic); + this.controlHandlers.forEach((handler) => handler(control)); + if (!this.hasSubscribers()) { + this.stopHeartbeat(); + this.scheduleIdleClose(); + } + return; + } + if (control.type === "maintenance" || control.type === "drain") { + if (typeof control.until !== "number" || !Number.isFinite(control.until)) return; + this.maintenanceUntil = Math.max(this.maintenanceUntil, control.until); + this.controlHandlers.forEach((handler) => handler(control)); + if (this.reconnectTimer) { + this.clock.clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + this.scheduleReconnect(false); + } + return; + } + if (control.type === "drain-complete") { + this.maintenanceUntil = 0; + this.controlHandlers.forEach((handler) => handler(control)); + if (this.reconnectTimer) this.clock.clearTimeout(this.reconnectTimer); this.reconnectTimer = null; - if (!this.closedByUser) this.open(); - }, delay); + this.reconnectBackoffMs = 0; + this.attempt = 0; + if (this.socket) { + const previousSocket = this.socket; + this.intentionallyClosing.add(previousSocket); + this.socket = null; + previousSocket.close(); + } + if (this.hasSubscribers()) this.open(); + return; + } + } + + private scheduleReconnect(advanceBackoff = true): void { + if (this.reconnectTimer || this.closedByUser || !this.hasSubscribers()) return; + if (advanceBackoff || this.reconnectBackoffMs === 0) { + this.reconnectBackoffMs = Math.min(BASE_BACKOFF_MS * 2 ** this.attempt, MAX_BACKOFF_MS); + this.attempt += 1; + } + const delay = Math.max(this.reconnectBackoffMs, this.maintenanceUntil - this.clock.now()); + this.reconnectTimer = this.clock.setTimeout( + () => { + this.reconnectTimer = null; + this.reconnectBackoffMs = 0; + if (!this.closedByUser && this.hasSubscribers()) this.open(); + }, + Math.max(0, delay) + ); + } + + private hasSubscribers(): boolean { + return [...this.handlers.values()].some((handlers) => handlers.size > 0); + } + + private startHeartbeat(socket: WebSocket): void { + if (this.heartbeatSocket === socket && this.heartbeatTimer) return; + this.stopHeartbeat(); + this.heartbeatSocket = socket; + this.heartbeatTimer = this.clock.setInterval(() => { + if (this.socket !== socket || socket.readyState !== 1) return; + if (this.heartbeatTimeoutTimer) this.clock.clearTimeout(this.heartbeatTimeoutTimer); + this.heartbeatTimeoutTimer = this.clock.setTimeout(() => { + this.heartbeatTimeoutTimer = null; + if (this.socket === socket) socket.close(); + }, this.heartbeatTimeoutMs); + this.sendControl({ type: "ping" }); + }, this.heartbeatIntervalMs); + } + + private stopHeartbeat(socket?: WebSocket): void { + if (socket && this.heartbeatSocket !== socket) return; + if (this.heartbeatTimer) this.clock.clearInterval(this.heartbeatTimer); + if (this.heartbeatTimeoutTimer) this.clock.clearTimeout(this.heartbeatTimeoutTimer); + this.heartbeatTimer = null; + this.heartbeatTimeoutTimer = null; + this.heartbeatSocket = null; + } + + private sendControl(payload: RealtimeControl): void { + if (this.socket?.readyState !== 1) return; + this.socket.send(JSON.stringify({ topic: REALTIME_CONTROL_TOPIC, payload })); + } + + private scheduleIdleClose(): void { + if (this.hasSubscribers() || this.idleTimer) return; + this.idleTimer = this.clock.setTimeout(() => { + this.idleTimer = null; + if (!this.hasSubscribers()) this.closeSocket(); + }, this.idleTimeoutMs); + } + + private cancelIdleClose(): void { + if (this.idleTimer) this.clock.clearTimeout(this.idleTimer); + this.idleTimer = null; + } + + private closeSocket(): void { + this.closedByUser = true; + if (this.reconnectTimer) this.clock.clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + this.reconnectBackoffMs = 0; + this.stopHeartbeat(); + this.socket?.close(); + this.socket = null; } subscribe(topic: string, handler: Handler): () => void { + if (this.revokedTopics.has(topic)) return () => {}; + this.cancelIdleClose(); if (!this.handlers.has(topic)) this.handlers.set(topic, new Set()); this.handlers.get(topic)!.add(handler); this.connect(); - + if (this.socket?.readyState === 1) this.startHeartbeat(this.socket); + let active = true; return () => { - this.handlers.get(topic)?.delete(handler); + if (!active) return; + active = false; + const handlers = this.handlers.get(topic); + handlers?.delete(handler); + if (handlers?.size === 0) this.handlers.delete(topic); + if (!this.hasSubscribers()) { + this.stopHeartbeat(); + this.scheduleIdleClose(); + } }; } + subscribeControl(handler: (control: RealtimeControl) => void): () => void { + this.controlHandlers.add(handler); + return () => this.controlHandlers.delete(handler); + } + disconnect(): void { - this.closedByUser = true; - if (this.reconnectTimer) clearTimeout(this.reconnectTimer); - this.socket?.close(); - this.socket = null; + this.cancelIdleClose(); + this.closeSocket(); } } diff --git a/src/lib/realtime/clock.ts b/src/lib/realtime/clock.ts new file mode 100644 index 0000000..0303188 --- /dev/null +++ b/src/lib/realtime/clock.ts @@ -0,0 +1,18 @@ +export type TimerHandle = ReturnType; + +/** Time source used by realtime lifecycle code. Tests can advance its timers deterministically. */ +export interface RealtimeClock { + now(): number; + setTimeout(callback: () => void, delayMs: number): TimerHandle; + clearTimeout(handle: TimerHandle): void; + setInterval(callback: () => void, intervalMs: number): TimerHandle; + clearInterval(handle: TimerHandle): void; +} + +export const realRealtimeClock: RealtimeClock = { + now: () => Date.now(), + setTimeout: (callback, delay) => setTimeout(callback, delay), + clearTimeout: (handle) => clearTimeout(handle), + setInterval: (callback, interval) => setInterval(callback, interval), + clearInterval: (handle) => clearInterval(handle), +}; diff --git a/src/lib/realtime/contract.ts b/src/lib/realtime/contract.ts new file mode 100644 index 0000000..2095aca --- /dev/null +++ b/src/lib/realtime/contract.ts @@ -0,0 +1,25 @@ +/** Wire topics and control frames for the frontend realtime connection. */ +export const REALTIME_CONTROL_TOPIC = "$control"; +export const REALTIME_TOPICS = { + oracle: "oracle_update", + claim: "claim_update", + notification: "notification_update", +} as const; + +/** Control frames use the reserved `$control` topic and never reach domain handlers. + * `revocation` removes one topic subscription permanently; `maintenance` defers + * reconnects until `until` (Unix milliseconds); `drain` enters the same deferred + * state; `drain-complete` clears it and reconnects immediately when needed. + */ +export type RealtimeControl = + | { type: "ping" } + | { type: "pong" } + | { type: "revocation"; topic: string; reason: string } + | { type: "maintenance"; until: number; reason?: string } + | { type: "drain"; until: number; reason?: string } + | { type: "drain-complete" }; + +export interface RealtimeEnvelope { + topic: string; + payload: unknown; +} diff --git a/src/lib/realtime/mappers/index.test.ts b/src/lib/realtime/mappers/index.test.ts new file mode 100644 index 0000000..57add14 --- /dev/null +++ b/src/lib/realtime/mappers/index.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, it } from "vitest"; +import { FIXTURE_ORACLE_READINGS } from "@/lib/fixtures/oracle"; +import { FIXTURE_RECENT_CLAIMS } from "@/lib/fixtures/recentClaims"; +import { mapClaimUpdate, mapOracleUpdate } from "./index"; + +describe("realtime domain mappers", () => { + it("preserves the REST oracle shape and values", () => { + const restReading = FIXTURE_ORACLE_READINGS[0]; + expect(mapOracleUpdate(restReading)).toEqual({ ok: true, value: restReading }); + }); + + it("normalizes claim wire seconds into the REST-derived domain timestamp in milliseconds", () => { + const restClaim = FIXTURE_RECENT_CLAIMS[0]; + expect(mapClaimUpdate({ ...restClaim, processedAt: restClaim.processedAt / 1000 })).toEqual({ + ok: true, + value: restClaim, + }); + }); + + it("drops malformed and precision-unsafe claim payloads", () => { + expect(mapClaimUpdate({ policyId: "1", payout: 1.25 })).toMatchObject({ ok: false }); + expect( + mapClaimUpdate({ + policyId: "1", + holder: "G", + coverageType: 1, + triggered: true, + payout: 1250000, + reason: "trigger", + processedAt: "2024-01-01", + }) + ).toMatchObject({ ok: false }); + }); +}); diff --git a/src/lib/realtime/mappers/index.ts b/src/lib/realtime/mappers/index.ts new file mode 100644 index 0000000..c8834b3 --- /dev/null +++ b/src/lib/realtime/mappers/index.ts @@ -0,0 +1,123 @@ +import type { ClaimRecord, NotificationEvent, OracleReading } from "@/lib/domain/realtime"; +import { REALTIME_TOPICS } from "../contract"; + +export type MappingResult = { ok: true; value: T } | { ok: false; reason: string }; +export type RealtimeDomainEvent = + | { topic: typeof REALTIME_TOPICS.oracle; value: OracleReading } + | { topic: typeof REALTIME_TOPICS.claim; value: ClaimRecord } + | { + topic: typeof REALTIME_TOPICS.notification; + value: NotificationEvent; + }; + +const failed = (reason: string): MappingResult => ({ ok: false, reason }); +const record = (value: unknown): value is Record => + value !== null && typeof value === "object" && !Array.isArray(value); + +export function mapOracleUpdate(payload: unknown): MappingResult { + if (!record(payload)) return failed("expected an oracle reading object"); + const { coverageType, type, value, threshold, severity, message } = payload; + if ( + typeof coverageType !== "string" || + type !== "oracle_update" || + typeof value !== "number" || + !Number.isFinite(value) || + typeof threshold !== "number" || + !Number.isFinite(threshold) || + !["low", "medium", "high", "triggered"].includes(String(severity)) || + typeof message !== "string" + ) + return failed("oracle reading does not match the REST domain shape"); + return { + ok: true, + value: { + coverageType, + type, + value, + threshold, + severity: severity as OracleReading["severity"], + message, + }, + }; +} + +export function mapClaimUpdate(payload: unknown): MappingResult { + if (!record(payload)) return failed("expected a claim object"); + const { + policyId, + holder, + coverageType, + triggered, + payout, + reason, + processedAt, + settlementTxHash, + } = payload; + if ( + typeof policyId !== "string" || + typeof holder !== "string" || + typeof coverageType !== "number" || + !Number.isSafeInteger(coverageType) || + typeof triggered !== "boolean" || + typeof payout !== "string" || + !/^\d+$/.test(payout) || + typeof reason !== "string" || + typeof processedAt !== "number" || + !Number.isFinite(processedAt) || + (settlementTxHash !== undefined && typeof settlementTxHash !== "string") + ) { + return failed("claim does not match the REST domain shape"); + } + const processedAtMs = Math.round(processedAt * 1000); + if (!Number.isSafeInteger(processedAtMs)) { + return failed("claim timestamp is outside the supported range"); + } + return { + ok: true, + value: { + policyId, + holder, + coverageType, + triggered, + payout, + reason, + processedAt: processedAtMs, + ...(settlementTxHash === undefined ? {} : { settlementTxHash }), + }, + }; +} + +export function mapNotificationUpdate(payload: unknown): MappingResult { + if ( + !record(payload) || + typeof payload.id !== "string" || + typeof payload.message !== "string" || + typeof payload.timestamp !== "number" || + !Number.isFinite(payload.timestamp) + ) { + return failed("notification does not match the domain shape"); + } + return { + ok: true, + value: { id: payload.id, message: payload.message, timestamp: payload.timestamp }, + }; +} + +export function mapRealtimeEvent( + topic: string, + payload: unknown +): MappingResult { + if (topic === REALTIME_TOPICS.oracle) { + const result = mapOracleUpdate(payload); + return result.ok ? { ok: true, value: { topic, value: result.value } } : result; + } + if (topic === REALTIME_TOPICS.claim) { + const result = mapClaimUpdate(payload); + return result.ok ? { ok: true, value: { topic, value: result.value } } : result; + } + if (topic === REALTIME_TOPICS.notification) { + const result = mapNotificationUpdate(payload); + return result.ok ? { ok: true, value: { topic, value: result.value } } : result; + } + return failed(`unsupported realtime topic: ${topic}`); +} diff --git a/src/lib/realtime/preferences.ts b/src/lib/realtime/preferences.ts new file mode 100644 index 0000000..dbe4fda --- /dev/null +++ b/src/lib/realtime/preferences.ts @@ -0,0 +1,30 @@ +export const REALTIME_OPT_OUT_KEY = "refract_realtime_opt_out"; + +export function isRealtimeOptedOut(): boolean { + if (typeof window === "undefined") return false; + try { + return window.localStorage.getItem(REALTIME_OPT_OUT_KEY) === "true"; + } catch { + return false; + } +} + +export function isRealtimeAllowed(): boolean { + if (typeof window === "undefined" || isRealtimeOptedOut()) return false; + const connection = (navigator as Navigator & { connection?: { saveData?: boolean } }).connection; + return ( + connection?.saveData !== true && !window.matchMedia?.("(prefers-reduced-data: reduce)").matches + ); +} + +/** Persist the user's choice for the next app-shell lifecycle. */ +export function setRealtimeOptOut(optOut: boolean): void { + if (typeof window === "undefined") return; + try { + if (optOut) window.localStorage.setItem(REALTIME_OPT_OUT_KEY, "true"); + else window.localStorage.removeItem(REALTIME_OPT_OUT_KEY); + } catch { + // Storage can be unavailable in private browsing; the current session remains usable. + } + window.dispatchEvent(new Event("refract-realtime-preference-change")); +} diff --git a/src/lib/realtime/testClock.ts b/src/lib/realtime/testClock.ts new file mode 100644 index 0000000..395a5ba --- /dev/null +++ b/src/lib/realtime/testClock.ts @@ -0,0 +1,10 @@ +import type { RealtimeClock } from "./clock"; + +/** Use after `vi.useFakeTimers()`; every client timer then advances with Vitest. */ +export const fakeRealtimeClock: RealtimeClock = { + now: () => Date.now(), + setTimeout: (callback, delay) => setTimeout(callback, delay), + clearTimeout: (handle) => clearTimeout(handle), + setInterval: (callback, interval) => setInterval(callback, interval), + clearInterval: (handle) => clearInterval(handle), +}; diff --git a/src/mocks/realtimeServer.ts b/src/mocks/realtimeServer.ts new file mode 100644 index 0000000..8d6aa1c --- /dev/null +++ b/src/mocks/realtimeServer.ts @@ -0,0 +1,57 @@ +/** In-memory WebSocket stand-in for realtime client tests and local scenarios. */ +export class MockRealtimeSocket extends EventTarget { + static readonly OPEN = 1; + readyState = 0; + sent: string[] = []; + closed = false; + constructor( + readonly url: string, + openDelayMs = 0, + private readonly autoPong = true + ) { + super(); + setTimeout(() => { + if (this.closed) return; + this.readyState = MockRealtimeSocket.OPEN; + this.dispatchEvent(new Event("open")); + }, openDelayMs); + } + send(data: string): void { + this.sent.push(data); + if (!this.autoPong) return; + try { + const envelope = JSON.parse(data) as { topic?: string; payload?: { type?: string } }; + if (envelope.topic === "$control" && envelope.payload?.type === "ping") { + this.emit("$control", { type: "pong" }); + } + } catch { + // Ignore invalid test client frames. + } + } + close(): void { + if (this.closed) return; + this.closed = true; + this.readyState = 3; + this.dispatchEvent(new Event("close")); + } + emit(topic: string, payload: unknown): void { + const event = new Event("message"); + Object.defineProperty(event, "data", { value: JSON.stringify({ topic, payload }) }); + this.dispatchEvent(event); + } +} + +export function createMockRealtimeServer(openDelayMs = 0, autoPong = true) { + const sockets: MockRealtimeSocket[] = []; + return { + sockets, + socketFactory: (url: string) => { + const socket = new MockRealtimeSocket(url, openDelayMs, autoPong); + sockets.push(socket); + return socket as unknown as WebSocket; + }, + control: (socket: MockRealtimeSocket, payload: unknown) => socket.emit("$control", payload), + message: (socket: MockRealtimeSocket, topic: string, payload: unknown) => + socket.emit(topic, payload), + }; +} diff --git a/vitest.config.mts b/vitest.config.mts index 3505082..c1830de 100644 --- a/vitest.config.mts +++ b/vitest.config.mts @@ -3,5 +3,8 @@ import { defineConfig } from "vitest/config"; export default defineConfig({ resolve: { alias: { "@": fileURLToPath(new URL("./src", import.meta.url)) } }, - test: { include: ["src/**/*.test.ts"] }, + test: { + include: ["src/**/*.test.ts"], + fakeTimers: { toFake: ["setTimeout", "clearTimeout", "setInterval", "clearInterval", "Date"] }, + }, }); diff --git a/vitest.config.ts b/vitest.config.ts deleted file mode 100644 index e69de29..0000000 From 14c85dafc46d8e1e31677dcbf2b79a475f7f4698 Mon Sep 17 00:00:00 2001 From: driftsorbit Date: Wed, 21 Oct 2026 18:23:37 +0000 Subject: [PATCH 2/2] test(realtime): verify shell renders without socket work --- src/lib/realtime/client.test.ts | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/lib/realtime/client.test.ts b/src/lib/realtime/client.test.ts index 57f1f8d..1d2bb71 100644 --- a/src/lib/realtime/client.test.ts +++ b/src/lib/realtime/client.test.ts @@ -1,4 +1,6 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { createElement } from "react"; +import { renderToString } from "react-dom/server"; import { RealtimeClient } from "./client"; import { fakeRealtimeClock } from "./testClock"; import { createMockRealtimeServer } from "@/mocks/realtimeServer"; @@ -60,11 +62,15 @@ describe("RealtimeClient lifecycle", () => { coldClient.disconnect(); }); - it("does no socket work during server rendering", () => { + it("does no socket work while the app shell renders on the server", () => { vi.unstubAllGlobals(); const socketFactory = vi.fn(); const client = new RealtimeClient({ socketFactory }); - client.warm(); + function ServerShell() { + client.warm(); + return createElement("main", null, "server content"); + } + expect(renderToString(createElement(ServerShell))).toContain("server content"); expect(socketFactory).not.toHaveBeenCalled(); });