Harden realtime controls, endpoints, and lifecycle

.eslintrc.json

@@ -1,6 +1,17 @@
 {
   "extends": "next/core-web-vitals",
   "rules": {
+    "no-restricted-imports": [
+      "error",
+      {
+        "patterns": [
+          {
+            "group": ["@/lib/realtime/*", "**/realtime/**"],
+            "message": "Import realtime APIs from @/lib/realtime."
+          }
+        ]
+      }
+    ],
     "no-restricted-syntax": [
       "error",
       {
@@ -12,7 +23,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."
       },
       {
@@ -23,7 +34,16 @@
   },
   "overrides": [
     {
-      "files": ["src/lib/coverage/metadata.ts", "src/lib/coverage/colors.ts", "src/app/global-error.tsx", "src/lib/format.ts"],
+      "files": ["src/lib/realtime/**"],
+      "rules": { "no-restricted-imports": "off" }
+    },
+    {
+      "files": [
+        "src/lib/coverage/metadata.ts",
+        "src/lib/coverage/colors.ts",
+        "src/app/global-error.tsx",
+        "src/lib/format.ts"
+      ],
       "rules": {
         "no-restricted-syntax": "off"
       }

docs/realtime-operations.md

@@ -0,0 +1,19 @@
+# Realtime operations and boundaries
+
+## Emergency disable
+
+Serve `/realtime-control.json` from the frontend origin or its CDN with a cache lifetime no longer than five seconds and purge it when tripping the switch. Set `{ "disabled": true }` to stop new connections and close active sockets. Clients check at startup, before reconnects, every five seconds, and when a tab becomes visible. Fetches have a three second timeout and never block app startup. A failed fetch keeps the last known state. The switch is latched in a running page; restoring `{ "disabled": false }` does not re-enable a tripped client. Reload clients after the incident is resolved. `tripKillSwitch("manual" | "auto")` is available to operator/developer code and logs the source. The current implementation does not automatically trip on connection failures.
+
+The checked-in JSON is a safe serving default. Production operations must publish and purge the same JSON through the configured CDN/backend path; frontend-only deployment is not required. Coordinate the control path and cache TTL with refract-backend/on-call before enabling realtime.
+
+## Endpoint policy
+
+`NEXT_PUBLIC_API_URL` selects the API and its corresponding `ws:`/`wss:` realtime origin. Keep it set in production, preview, and CI. Next emits `Content-Security-Policy-Report-Only`; review browser reports for one release before changing the header to enforced `Content-Security-Policy`. Add any intentional connection origin through the canonical API environment configuration, then validate the generated header. Development defaults to the local API at port 4001.
+
+## Public module boundary
+
+Consumers import supported APIs from `@/lib/realtime`. Files inside `src/lib/realtime/` are implementation details and may change. To make a new API public, add an explicit named export and its public type to `index.ts`; do not export internal helpers wholesale. The lint rule rejects deep imports outside the module, and is disabled only within the module itself.
+
+## Page lifecycle
+
+Hiding the document, `pagehide`, and `freeze` suspend the socket and cancel reconnect timers. Becoming visible, `pageshow`, and `resume` reconnect and consumers clear their live payload while suspended; their REST query remains authoritative. This repository currently has no outbound realtime mutation queue or cross-tab leader-election lease, so there is no pending realtime write or leader role to flush/relinquish.

next.config.js

@@ -2,16 +2,27 @@ const isPreview = process.env.VERCEL_ENV === "preview";
 const withBundleAnalyzer = require("@next/bundle-analyzer")({
   enabled: process.env.ANALYZE === "true",
 });
+const { endpointFromApiBase } = require("./src/lib/realtime/endpointConfig");
+
+function realtimeConnectSources() {
+  let endpoint;
+  try {
+    endpoint = endpointFromApiBase(
+      process.env.NEXT_PUBLIC_API_URL || "http://localhost:4001/api/v1"
+    );
+  } catch {
+    throw new Error("NEXT_PUBLIC_API_URL must be an absolute URL to build the realtime CSP");
+  }
+  return `'self' ${endpoint.websocketOrigin} ${endpoint.apiOrigin}`;
+}
 
 /** @type {import('next').NextConfig} */
 const nextConfig = {
   reactStrictMode: true,
   env: {
     // Preview deployments are ephemeral and have no reachable backend, so force
     // the app's fixture-fallback demo mode instead of leaking a real API URL.
-    NEXT_PUBLIC_API_URL: isPreview
-      ? ""
-      : process.env.NEXT_PUBLIC_API_URL || "",
+    NEXT_PUBLIC_API_URL: isPreview ? "" : process.env.NEXT_PUBLIC_API_URL || "",
   },
   async headers() {
     return [
@@ -22,6 +33,15 @@ const nextConfig = {
           { key: "X-Content-Type-Options", value: "nosniff" },
         ],
       },
+      {
+        source: "/(.*)",
+        headers: [
+          {
+            key: "Content-Security-Policy-Report-Only",
+            value: `default-src 'self'; connect-src ${realtimeConnectSources()}`,
+          },
+        ],
+      },
     ];
   },
 };

public/realtime-control.json

@@ -0,0 +1 @@
+{ "disabled": false }

src/app/providers.tsx

@@ -1,8 +1,9 @@
 "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 { bindRealtimeLifecycle, realtimeClient } from "@/lib/realtime";
 
 export function Providers({ children }: { children: React.ReactNode }) {
   // One client per browser session. Query functions already fold failures
@@ -11,6 +12,8 @@ export function Providers({ children }: { children: React.ReactNode }) {
     () => new QueryClient({ defaultOptions: { queries: { retry: false, staleTime: 30_000 } } })
   );
 
+  useEffect(() => bindRealtimeLifecycle(realtimeClient), []);
+
   return (
     <QueryClientProvider client={queryClient}>
       <WalletProvider>{children}</WalletProvider>

src/lib/realtime/allowlist.test.ts

@@ -0,0 +1,16 @@
+import { describe, expect, it } from "vitest";
+import { buildRealtimeCsp, isAllowedRealtimeUrl, resolveRealtimeEndpoint } from "./allowlist";
+
+describe("realtime endpoint allowlist", () => {
+  it("derives a single websocket origin and CSP connect policy", () => {
+    const endpoint = resolveRealtimeEndpoint("https://api.example.com/api/v1");
+    expect(endpoint.url).toBe("wss://api.example.com/ws");
+    expect(isAllowedRealtimeUrl(endpoint.url, endpoint)).toBe(true);
+    expect(isAllowedRealtimeUrl("wss://other.example/ws", endpoint)).toBe(false);
+    expect(buildRealtimeCsp(endpoint)).toContain("wss://api.example.com https://api.example.com");
+  });
+  it("rejects malformed and untrusted origins", () => {
+    expect(isAllowedRealtimeUrl("javascript:alert(1)")).toBe(false);
+    expect(isAllowedRealtimeUrl("wss://attacker.example/socket")).toBe(false);
+  });
+});

src/lib/realtime/allowlist.ts

@@ -0,0 +1,40 @@
+import { API_BASE_URL } from "../api/client";
+import { endpointFromApiBase } from "./endpointConfig";
+
+export type RealtimeEndpoint = {
+  url: string;
+  origin: string;
+  websocketOrigin: string;
+  sseOrigin: string;
+};
+
+export function resolveRealtimeEndpoint(apiBaseUrl: string): RealtimeEndpoint {
+  const config = endpointFromApiBase(apiBaseUrl);
+  return {
+    url: config.url,
+    origin: config.apiOrigin,
+    websocketOrigin: config.websocketOrigin,
+    sseOrigin: config.apiOrigin,
+  };
+}
+
+export const REALTIME_ENDPOINT = resolveRealtimeEndpoint(API_BASE_URL);
+
+export function isAllowedRealtimeUrl(
+  url: string,
+  endpoint: RealtimeEndpoint = REALTIME_ENDPOINT
+): boolean {
+  try {
+    const candidate = new URL(url);
+    return (
+      candidate.origin === endpoint.websocketOrigin &&
+      candidate.protocol === (endpoint.websocketOrigin.startsWith("wss:") ? "wss:" : "ws:")
+    );
+  } catch {
+    return false;
+  }
+}
+
+export function buildRealtimeCsp(endpoint: RealtimeEndpoint): string {
+  return `default-src 'self'; connect-src 'self' ${endpoint.websocketOrigin} ${endpoint.sseOrigin}`;
+}

src/lib/realtime/client.ts

@@ -1,6 +1,20 @@
-import { API_BASE_URL } from "@/lib/api/client";
+import { API_BASE_URL } from "../api/client";
+import { isAllowedRealtimeUrl, resolveRealtimeEndpoint } from "./allowlist";
+import {
+  checkRemoteKillSwitch,
+  isRealtimeDisabled,
+  startKillSwitchPolling,
+  subscribeKillSwitch,
+} from "./killSwitch";
 
 type Handler = (payload: unknown) => void;
+export class RealtimeEndpointError extends Error {
+  readonly code = "REALTIME_ENDPOINT_NOT_ALLOWED";
+  constructor(url: string) {
+    super(`Realtime endpoint is not allowlisted: ${url}`);
+    this.name = "RealtimeEndpointError";
+  }
+}
 
 const MAX_BACKOFF_MS = 30_000;
 const BASE_BACKOFF_MS = 1_000;
@@ -14,21 +28,45 @@ const BASE_BACKOFF_MS = 1_000;
  * best-effort — always keep an independent REST fetch as the source of
  * truth for initial state.
  */
-class RealtimeClient {
+export class RealtimeClient {
   private socket: WebSocket | null = null;
   private handlers = new Map<string, Set<Handler>>();
   private attempt = 0;
   private closedByUser = false;
   private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
+  private removeKillSwitchListener: (() => void) | null = null;
+  private suspended = false;
+  private opening = false;
 
   connect(): void {
-    if (typeof WebSocket === "undefined" || this.socket) return;
+    if (
+      typeof WebSocket === "undefined" ||
+      this.socket ||
+      this.opening ||
+      isRealtimeDisabled() ||
+      this.suspended
+    )
+      return;
+    startKillSwitchPolling();
+    if (!this.removeKillSwitchListener)
+      this.removeKillSwitchListener = subscribeKillSwitch(({ disabled }) => {
+        if (disabled) {
+          this.handlers.clear();
+          this.stopSocket();
+        }
+      });
     this.closedByUser = false;
     this.open();
   }
 
-  private open(): void {
-    const wsUrl = API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws");
+  private async open(): Promise<void> {
+    if (this.opening || isRealtimeDisabled() || this.suspended) return;
+    this.opening = true;
+    await checkRemoteKillSwitch();
+    this.opening = false;
+    if (isRealtimeDisabled() || this.suspended || this.closedByUser || this.socket) return;
+    const wsUrl = resolveRealtimeEndpoint(API_BASE_URL).url;
+    if (!isAllowedRealtimeUrl(wsUrl)) throw new RealtimeEndpointError(wsUrl);
     try {
       this.socket = new WebSocket(wsUrl);
     } catch {
@@ -41,15 +79,18 @@ class RealtimeClient {
     });
     this.socket.addEventListener("message", (event) => {
       try {
-        const { topic, payload } = JSON.parse(event.data as string) as { topic: string; payload: unknown };
+        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.
       }
     });
     this.socket.addEventListener("close", () => {
       this.socket = null;
-      if (!this.closedByUser) this.scheduleReconnect();
+      if (!this.closedByUser && !isRealtimeDisabled() && !this.suspended) this.scheduleReconnect();
     });
     this.socket.addEventListener("error", () => {
       this.socket?.close();
@@ -62,11 +103,12 @@ class RealtimeClient {
     this.attempt += 1;
     this.reconnectTimer = setTimeout(() => {
       this.reconnectTimer = null;
-      if (!this.closedByUser) this.open();
+      if (!this.closedByUser && !isRealtimeDisabled() && !this.suspended) this.open();
     }, delay);
   }
 
   subscribe(topic: string, handler: Handler): () => void {
+    if (isRealtimeDisabled()) return () => undefined;
     if (!this.handlers.has(topic)) this.handlers.set(topic, new Set());
     this.handlers.get(topic)!.add(handler);
     this.connect();
@@ -78,7 +120,23 @@ class RealtimeClient {
 
   disconnect(): void {
     this.closedByUser = true;
+    this.stopSocket();
+  }
+
+  /** Page lifecycle adapter commands; handlers remain registered for resync. */
+  suspend(): void {
+    this.suspended = true;
+    this.stopSocket();
+  }
+  resume(): void {
+    this.suspended = false;
+    this.closedByUser = false;
+    this.connect();
+  }
+
+  private stopSocket(): void {
     if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
+    this.reconnectTimer = null;
     this.socket?.close();
     this.socket = null;
   }

src/lib/realtime/endpointConfig.d.ts

@@ -0,0 +1,5 @@
+export function endpointFromApiBase(apiBaseUrl: string): {
+  url: string;
+  apiOrigin: string;
+  websocketOrigin: string;
+};

src/lib/realtime/endpointConfig.js

@@ -0,0 +1,14 @@
+/** Canonical API-to-realtime endpoint derivation shared by Next config and browser code. */
+function endpointFromApiBase(apiBaseUrl) {
+  const api = new URL(apiBaseUrl || "http://localhost:4001/api/v1");
+  const websocketProtocol = api.protocol === "https:" ? "wss:" : "ws:";
+  const websocket = new URL(api.pathname.replace(/\/api\/v1\/?$/, "/ws"), api);
+  websocket.protocol = websocketProtocol;
+  return {
+    url: websocket.toString(),
+    apiOrigin: api.origin,
+    websocketOrigin: `${websocket.protocol}//${websocket.host}`,
+  };
+}
+
+module.exports = { endpointFromApiBase };

src/lib/realtime/index.ts

@@ -0,0 +1,9 @@
+export { realtimeClient, RealtimeEndpointError } from "./client";
+export type { RealtimeClient } from "./client";
+export { useRealtimeTopic } from "./useRealtimeTopic";
+export { bindRealtimeLifecycle } from "./lifecycle";
+export { REALTIME_ENDPOINT, isAllowedRealtimeUrl } from "./allowlist";
+export type { RealtimeEndpoint } from "./allowlist";
+export const REALTIME_TOPICS = { ORACLE_UPDATE: "oracle_update" } as const;
+export { isRealtimeDisabled, getKillSwitchState, tripKillSwitch } from "./killSwitch";
+export type { KillSwitchState, KillSwitchSource } from "./killSwitch";