feat(realtime): warm shared connections and handle control frames

.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."
       },
       {

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 = ["🪙", "📉", "🛡️", "🔐", "✈️"];

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 (
+    <html lang="en">
+      <body>
+        <Providers>{children}</Providers>
+      </body>
+    </html>
+  );
+}

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";

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 (
     <QueryClientProvider client={queryClient}>
-      <WalletProvider>{children}</WalletProvider>
+      <ToastProvider>
+        <RealtimeLifecycle />
+        <WalletProvider>{children}</WalletProvider>
+      </ToastProvider>
     </QueryClientProvider>
   );
 }
+
+/** 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;
+}

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 (
+    <header className="border-b border-white/10 bg-pm-bg/90">
+      <nav
+        aria-label="Main navigation"
+        className="mx-auto flex max-w-7xl items-center justify-between px-4 py-4 sm:px-6"
+      >
+        <Link href="/" className="font-semibold tracking-tight text-white">
+          Refract
+        </Link>
+        <div className="flex items-center gap-5 text-sm text-white/80">
+          <Link href="/cover" className="hover:text-white">
+            Get Coverage
+          </Link>
+          <Link href="/provide" className="hover:text-white">
+            Provide Capital
+          </Link>
+          {right}
+        </div>
+      </nav>
+    </header>
+  );
+}

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<void>;
+}
+
+/** 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<OracleReading[]>(FIXTURE_ORACLE_READINGS);
+  const [lastUpdated, setLastUpdated] = useState<Date | null>(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 };
+}

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<T>(
+  topic: string,
+  map: (payload: unknown) => MappingResult<T>,
+  onValue?: (value: T) => void,
+  enabled = true
+): T | null {
+  const [value, setValue] = useState<T | null>(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;
+}

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<T> {
   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,31 +42,33 @@ 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<ClaimRecord> {
-  const items = response.items ?? response.claims ?? [];
+function normalizeClaimsResponse(response: ClaimsEndpointResponse): PaginatedResponse<ClaimRecord> {
+  const items = (response.items ?? response.claims ?? []).map((claim) => ({
+    ...claim,
+    processedAt: claim.processedAt * 1000,
+  }));
   const hasMore = response.hasMore ?? false;
   return { items, nextCursor: response.nextCursor, hasMore };
 }
 
 export function fetchHolderClaims(
   address: string,
   params: PaginationParams = {},
-  signal?: AbortSignal,
+  signal?: AbortSignal
 ): Promise<PaginatedResponse<ClaimRecord>> {
-  return apiRequest(`/claims/holder/${address}${buildPaginationQuery(params)}`, { signal }).then(
-    normalizeClaimsResponse,
-  );
+  return apiRequest<ClaimsEndpointResponse>(
+    `/claims/holder/${address}${buildPaginationQuery(params)}`,
+    { signal }
+  ).then(normalizeClaimsResponse);
 }
 
 export function fetchRecentClaims(
   params: PaginationParams = {},
-  signal?: AbortSignal,
+  signal?: AbortSignal
 ): Promise<PaginatedResponse<ClaimRecord>> {
-  return apiRequest(`/claims/recent${buildPaginationQuery(params)}`, { signal }).then(
-    normalizeClaimsResponse,
-  );
+  return apiRequest<ClaimsEndpointResponse>(`/claims/recent${buildPaginationQuery(params)}`, {
+    signal,
+  }).then(normalizeClaimsResponse);
 }
 
 /** Types are generated from openapi/refract-api.yaml, which mirrors ClaimService.getStats()'s response in refract-backend. */

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 });

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;
+}

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";
 

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;