+
+
+
+ ⬡
+
+
Refract
+
+
+
+ {NAV_LINKS.map((item) => {
+ const active = pathname === item.href;
+ return (
- {link.label}
+ {item.label}
-
- );
- })}
-
+ );
+ })}
+
+
-
+
+
+ {right}
+
+
+
+
+
+ {menuOpen && (
+
+
+ {NAV_LINKS.map((item) => {
+ const active = pathname === item.href;
+ return (
+ setMenuOpen(false)}
+ className={cn(
+ "rounded-md px-3 py-2.5 text-sm font-medium",
+ active ? "bg-pm-violet/10 text-pm-violet" : "text-pm-text/70"
+ )}
+ >
+ {item.label}
+
+ );
+ })}
+
+ {right &&
{right}
}
-
-
+ )}
+
);
}
diff --git a/src/components/status/ProtocolStatusView.tsx b/src/components/status/ProtocolStatusView.tsx
new file mode 100644
index 0000000..600f7c3
--- /dev/null
+++ b/src/components/status/ProtocolStatusView.tsx
@@ -0,0 +1,106 @@
+"use client";
+
+import { useEffect, useState } from "react";
+
+import { Badge, Card, RealtimeView, Skeleton, TextBlockSkeleton } from "@/components/ui";
+import { useRealtimeTopic } from "@/hooks/useRealtimeTopic";
+import { ACTIVE_NETWORK } from "@/lib/network";
+import type { ProtocolStatus, ProtocolStatusLevel } from "@/lib/realtime";
+import { fetchProtocolStatus } from "@/lib/realtime/status";
+
+const LEVEL_COPY: Record<
+ ProtocolStatusLevel,
+ { label: string; icon: string; description: string }
+> = {
+ healthy: { label: "Healthy", icon: "✓", description: "All systems are operating normally." },
+ degraded: {
+ label: "Degraded",
+ icon: "!",
+ description: "Some data may be delayed or temporarily unavailable.",
+ },
+ maintenance: {
+ label: "Maintenance",
+ icon: "⚙",
+ description: "Scheduled maintenance is in progress.",
+ },
+ unknown: {
+ label: "Unknown",
+ icon: "?",
+ description: "We can't determine protocol status right now.",
+ },
+};
+
+const formatTime = (iso: string) =>
+ Number.isNaN(Date.parse(iso))
+ ? "—"
+ : new Date(iso).toLocaleString(undefined, { dateStyle: "medium", timeStyle: "short" });
+
+export function ProtocolStatusView({ initial }: { initial?: ProtocolStatus }) {
+ const [snapshot, setSnapshot] = useState
(initial);
+ const { data: live, status } = useRealtimeTopic("protocol_status", { initial });
+
+ // REST snapshot: first paint when nothing was server-rendered, and refetch on network switch.
+ useEffect(() => {
+ let cancelled = false;
+ void fetchProtocolStatus().then((s) => !cancelled && setSnapshot(s));
+ return () => {
+ cancelled = true;
+ };
+ }, [ACTIVE_NETWORK.id]);
+
+ const current = live ?? snapshot;
+ // Never show a banner that belongs to another environment.
+ const data =
+ current && (!current.network || current.network === ACTIVE_NETWORK.id) ? current : undefined;
+ const copy = LEVEL_COPY[data?.level ?? "unknown"];
+
+ return (
+
+
+
+
+ }
+ >
+
+
+
+ {copy.icon}
+ {copy.label}
+
+ {ACTIVE_NETWORK.label}
+
+ {copy.description}
+ {data?.maintenance && (
+
+
Maintenance window: {formatTime(data.maintenance.startsAt)} –{" "}
+ {formatTime(data.maintenance.endsAt)}
+ {data.maintenance.message && (
+
{data.maintenance.message}
+ )}
+
+ Last updated {data ? formatTime(data.updatedAt) : "—"}
+
+ {data?.incidents && data.incidents.length > 0 && (
+
+ {data.incidents.map((i) => (
+ -
+ {i.title} — {formatTime(i.at)} ({i.resolved ? "resolved" : "ongoing"})
+
+ ))}
+
+ )}
+
+
+ );
+}
diff --git a/src/components/ui/RealtimeSkeletons.tsx b/src/components/ui/RealtimeSkeletons.tsx
new file mode 100644
index 0000000..a66fca1
--- /dev/null
+++ b/src/components/ui/RealtimeSkeletons.tsx
@@ -0,0 +1,80 @@
+"use client";
+
+import type { ReactNode } from "react";
+
+import { useReducedMotion } from "@/hooks/usePrefersReducedMotion";
+import type { RealtimeViewStatus } from "@/lib/realtime";
+import { cn } from "@/lib/cn";
+import { Skeleton } from "./Skeleton";
+
+/**
+ * State matrix for realtime-backed views (status -> presentation):
+ *
+ * | status | content | indicator |
+ * |--------------|--------------------|---------------------------------|
+ * | empty | skeleton (no data) | none |
+ * | live | data | "Live" |
+ * | disconnected | data (kept) | "Offline — showing last values" |
+ * | reconnecting | data (kept) | "Reconnecting…" |
+ * | polling | data | "Updating periodically" |
+ */
+const STATUS_LABEL: Record, string> = {
+ live: "Live",
+ disconnected: "Offline — showing last values",
+ reconnecting: "Reconnecting…",
+ polling: "Updating periodically",
+};
+
+interface RealtimeViewProps {
+ status: RealtimeViewStatus;
+ /** Skeleton sized to match the final content so layout does not shift. */
+ skeleton: ReactNode;
+ children: ReactNode;
+ className?: string;
+}
+
+/**
+ * Renders the skeleton only for a genuinely empty state; once data exists it
+ * stays on screen through disconnects/reconnects (no flicker on refresh).
+ */
+export function RealtimeView({ status, skeleton, children, className }: RealtimeViewProps) {
+ const reduced = useReducedMotion();
+ if (status === "empty") {
+ return (
+
+ {skeleton}
+
+ );
+ }
+ return (
+
+
+ {STATUS_LABEL[status]}
+
+ {children}
+
+ {Array.from({ length: lines }).map((_, i) => (
+
+ ))}
+
+ );
+}
diff --git a/src/components/ui/index.ts b/src/components/ui/index.ts
index 663cd8e..3b28ad7 100644
--- a/src/components/ui/index.ts
+++ b/src/components/ui/index.ts
@@ -9,3 +9,4 @@ export { Container } from "./Container";
export { AddressInput } from "./AddressInput";
export { LiveCountdown } from "./LiveCountdown";
export { RelativeDate } from "./RelativeDate";
+export { RealtimeView, TextBlockSkeleton } from "./RealtimeSkeletons";
diff --git a/src/hooks/useRealtimeTopic.ts b/src/hooks/useRealtimeTopic.ts
index a01c1b5..47881f3 100644
--- a/src/hooks/useRealtimeTopic.ts
+++ b/src/hooks/useRealtimeTopic.ts
@@ -1,16 +1,41 @@
-import { useCallback, useEffect, useRef, useState } from 'react';
-import { useCallback, useEffect, useRef, useState } from 'react';
+"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';
+} 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";
-export type RealtimeStatus = 'idle' | 'loading' | 'live' | 'polling' | 'error';
+/** 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 {
/**
@@ -47,6 +72,10 @@ export interface UseRealtimeTopicOptions {
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 {
@@ -61,27 +90,27 @@ export interface UseRealtimeTopicResult {
const DEFAULT_POLL_INTERVAL_MS = 30_000;
const isTestEnv = (): boolean => {
- if (typeof process !== 'undefined' && process.env && process.env.NODE_ENV === 'test') {
+ 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'
+ 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') {
+ if (typeof window === "undefined" || typeof window.matchMedia !== "function") {
return false;
}
try {
- return window.matchMedia('(prefers-reduced-motion: reduce)').matches;
+ return window.matchMedia("(prefers-reduced-motion: reduce)").matches;
} catch {
return false;
}
};
const scheduleFrame = (cb: () => void): (() => void) => {
- if (typeof window !== 'undefined' && typeof window.requestAnimationFrame === 'function') {
+ if (typeof window !== "undefined" && typeof window.requestAnimationFrame === "function") {
const id = window.requestAnimationFrame(cb);
return () => window.cancelAnimationFrame(id);
}
@@ -107,10 +136,10 @@ const scheduleFrame = (cb: () => void): (() => void) => {
* on mount so reloads and route navigations resume from the snapshot instead
* of refetching from scratch.
*/
-export function useRealtimeTopic(
- topic: string,
- options: UseRealtimeTopicOptions = {},
-): UseRealtimeTopicResult {
+export function useRealtimeTopic(
+ topic: T,
+ options: UseRealtimeTopicOptions = {}
+): UseRealtimeTopicResult & { enabled: boolean } {
const {
replaceable = false,
select,
@@ -123,13 +152,16 @@ export function useRealtimeTopic(
subscribe,
client,
registry,
- scope = '',
- wallet = '',
+ scope = "",
+ wallet = "",
knownTopics,
+ initial,
+ critical = false,
} = options;
- const [data, setData] = useState(undefined);
- const [status, setStatus] = useState('idle');
+ 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);
@@ -143,12 +175,12 @@ export function useRealtimeTopic(
const selectRef = useRef(select);
selectRef.current = select;
- const pendingEventsRef = useRef([]);
+ const pendingEventsRef = useRef([]);
const mountedRef = useRef(true);
// Coalescing state for replaceable topics.
- const stateRef = useRef | undefined>(undefined);
- const pendingRef = useRef | null>(null);
+ const stateRef = useRef | undefined>(undefined);
+ const pendingRef = useRef | null>(null);
const cancelRef = useRef<(() => void) | null>(null);
const coalesce = replaceable && !isTestEnv() && !prefersReducedMotion();
@@ -164,15 +196,15 @@ export function useRealtimeTopic(
pendingRef.current = null;
stateRef.current = next;
if (mountedRef.current) {
- setData(next as unknown as T);
+ setData(next as unknown as RealtimeTopics[T]);
}
}, []);
- const apply = useCallback((next: RealtimeState): void => {
+ const apply = useCallback((next: RealtimeState): void => {
if (!coalesceRef.current) {
stateRef.current = next;
if (mountedRef.current) {
- setData(next as unknown as T);
+ setData(next as unknown as RealtimeTopics[T]);
}
return;
}
@@ -182,12 +214,12 @@ export function useRealtimeTopic(
}
}, [flush]);
- const applyEvent = useCallback((event: T) => {
+ 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);
+ ? 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());
@@ -202,10 +234,10 @@ export function useRealtimeTopic(
const pending = pendingEventsRef.current;
pendingEventsRef.current = [];
const merged = mergeSnapshot(snapshot, pending, mergeRef.current);
- stateRef.current = merged as unknown as RealtimeState;
+ stateRef.current = merged as unknown as RealtimeState;
setData(merged);
} catch {
- if (mountedRef.current) setStatus('error');
+ if (mountedRef.current) setStatus("error");
}
}, []);
@@ -213,7 +245,7 @@ export function useRealtimeTopic(
// 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: T) => void) => resolvedRegistry.subscribe(t, onEvent)
+ ? (t: string, onEvent: (event: RealtimeTopics[T]) => void) => resolvedRegistry.subscribe(t, onEvent)
: subscribe;
const realtimeActive = enabled && realtimeEnabled && connected && !!subscribeFn;
@@ -256,7 +288,7 @@ export function useRealtimeTopic(
// Snapshot fetch on mount / when the topic changes.
useEffect(() => {
mountedRef.current = true;
- setStatus('loading');
+ setStatus("loading");
void fetchSnapshot();
return () => {
mountedRef.current = false;
@@ -266,7 +298,7 @@ export function useRealtimeTopic(
// Realtime subscription lifecycle (routed through the registry).
useEffect(() => {
if (!realtimeActive || !subscribeFn) return;
- setStatus('live');
+ setStatus("live");
const unsubscribe = subscribeFn(topic, (event) => {
if (pendingEventsRef.current.length > 0) {
pendingEventsRef.current.push(event);
@@ -287,14 +319,36 @@ export function useRealtimeTopic(
// REST polling fallback when realtime is disabled or unavailable.
useEffect(() => {
if (!enabled || realtimeActive) return;
- setStatus('polling');
+ setStatus("polling");
const interval = setInterval(() => {
void fetchSnapshot();
}, pollIntervalMs);
return () => clearInterval(interval);
}, [enabled, realtimeActive, pollIntervalMs, fetchSnapshot]);
- return { data, status, lastEventAt, cursor, setCursor, clear };
+ // 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 };
}
/**
@@ -303,7 +357,7 @@ export function useRealtimeTopic(
*/
export function useRealtimeFlush(): () => void {
return () => {
- if (typeof (globalThis as { __REALTIME_FLUSH__?: () => void }).__REALTIME_FLUSH__ === 'function') {
+ if (typeof (globalThis as { __REALTIME_FLUSH__?: () => void }).__REALTIME_FLUSH__ === "function") {
(globalThis as { __REALTIME_FLUSH__?: () => void }).__REALTIME_FLUSH__!();
}
};
diff --git a/src/lib/realtime/bus.ts b/src/lib/realtime/bus.ts
new file mode 100644
index 0000000..877d02d
--- /dev/null
+++ b/src/lib/realtime/bus.ts
@@ -0,0 +1,74 @@
+import { SNAPSHOT_TOPICS, type RealtimeTopic, type RealtimeTopics } from "./topics";
+
+type BusHandler = (payload: RealtimeTopics[T]) => void;
+
+const MAX_HANDLERS_PER_TOPIC = 100;
+
+/**
+ * Typed pub/sub between the realtime client (which decodes and validates) and
+ * in-app consumers. Per-topic arrival order is preserved, one throwing handler
+ * never blocks the others, and snapshot topics replay their latest value to
+ * late subscribers.
+ */
+export class RealtimeBus {
+ private handlers = new Map>>();
+ private latest = new Map();
+ private queues = new Map();
+ private draining = new Set();
+
+ publish(topic: T, payload: RealtimeTopics[T]): void {
+ if (SNAPSHOT_TOPICS.has(topic)) this.latest.set(topic, payload);
+ const queue = this.queues.get(topic) ?? [];
+ queue.push(payload);
+ this.queues.set(topic, queue);
+ // Re-entrant publishes are queued and delivered after the current event.
+ if (this.draining.has(topic)) return;
+ this.draining.add(topic);
+ try {
+ while (queue.length) {
+ const next = queue.shift() as RealtimeTopics[T];
+ for (const handler of [...(this.handlers.get(topic) ?? [])]) {
+ try {
+ (handler as BusHandler)(next);
+ } catch {
+ console.error(`[realtime] handler for "${topic}" threw`);
+ }
+ }
+ }
+ } finally {
+ this.draining.delete(topic);
+ }
+ }
+
+ subscribe(topic: T, handler: BusHandler): () => void {
+ let set = this.handlers.get(topic);
+ if (!set) this.handlers.set(topic, (set = new Set()));
+ if (set.size >= MAX_HANDLERS_PER_TOPIC && !set.has(handler as BusHandler)) {
+ console.error(`[realtime] handler limit reached for "${topic}"`);
+ return () => {};
+ }
+ // Set semantics make a StrictMode double-registration of the same handler a no-op.
+ set.add(handler as BusHandler);
+ return () => {
+ this.handlers.get(topic)?.delete(handler as BusHandler);
+ };
+ }
+
+ /** Latest value of a snapshot topic, or undefined if none has arrived. */
+ getSnapshot(topic: T): RealtimeTopics[T] | undefined {
+ return this.latest.get(topic) as RealtimeTopics[T] | undefined;
+ }
+
+ handlerCount(topic: RealtimeTopic): number {
+ return this.handlers.get(topic)?.size ?? 0;
+ }
+
+ reset(): void {
+ this.handlers.clear();
+ this.latest.clear();
+ this.queues.clear();
+ this.draining.clear();
+ }
+}
+
+export const realtimeBus = new RealtimeBus();
diff --git a/src/lib/realtime/client.ts b/src/lib/realtime/client.ts
index 5c444eb..f80e126 100644
--- a/src/lib/realtime/client.ts
+++ b/src/lib/realtime/client.ts
@@ -1,8 +1,12 @@
import { API_BASE_URL } from "../api/client";
-
+import { realtimeBus } from "./bus";
+import type { RealtimeTopic, RealtimeTopics } from "./topics";
import { RequestCorrelator, type RequestOptions } from "./correlation";
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 {
@@ -42,6 +46,47 @@ class RealtimeClient {
private closedByUser = false;
private correlator = new RequestCorrelator();
private reconnectTimer: ReturnType | null = null;
+ 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.
+ }
+ }
connect(): void {
if ((!customTransport && typeof WebSocket === "undefined") || this.socket) return;
@@ -60,6 +105,8 @@ class RealtimeClient {
this.socket.addEventListener("open", () => {
this.attempt = 0;
+ this.setState("connected");
+ for (const t of this.wireTopics) this.send({ action: "subscribe", topic: t });
});
this.socket.addEventListener("message", (event) => {
try {
@@ -67,6 +114,10 @@ class RealtimeClient {
if (this.correlator.handleFrame(frame)) return;
const { topic, payload } = frame;
this.handlers.get(topic)?.forEach((handler) => handler(payload));
+ // Decoded events fan out through the bus so consumers never own a socket subscription.
+ if (typeof topic === "string" && payload !== null && typeof payload === "object") {
+ realtimeBus.publish(topic as RealtimeTopic, payload as RealtimeTopics[RealtimeTopic]);
+ }
} catch {
// Malformed message — ignore, don't crash the connection.
}
@@ -74,7 +125,12 @@ class RealtimeClient {
this.socket.addEventListener("close", () => {
this.socket = null;
this.correlator.rejectAll();
- if (!this.closedByUser) this.scheduleReconnect();
+ if (!this.closedByUser) {
+ this.setState("reconnecting");
+ this.scheduleReconnect();
+ } else {
+ this.setState("idle");
+ }
});
this.socket.addEventListener("error", () => {
this.socket?.close();
@@ -128,6 +184,7 @@ class RealtimeClient {
if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
this.socket?.close();
this.socket = null;
+ this.setState("idle");
}
}
diff --git a/src/lib/realtime/fidelity.ts b/src/lib/realtime/fidelity.ts
new file mode 100644
index 0000000..2c5a5ef
--- /dev/null
+++ b/src/lib/realtime/fidelity.ts
@@ -0,0 +1,237 @@
+import { realtimeClient } from "./client";
+import type { RealtimeTopic } from "./topics";
+
+export type FidelityTier = "full" | "lite" | "minimal";
+
+export interface FidelityProfile {
+ /** Topics held on the wire in this tier. */
+ topics: readonly RealtimeTopic[];
+ /** Minimum ms between delivered events per volatile topic (0 = deliver all). */
+ coalesceMs: Partial>;
+ transport: "websocket" | "polling";
+}
+
+/** Declarative tier definitions — mirrored in docs/realtime-fidelity.md. */
+export const FIDELITY_PROFILES: Record = {
+ full: {
+ topics: ["oracle_update", "protocol_status", "network_health"],
+ coalesceMs: {},
+ transport: "websocket",
+ },
+ lite: {
+ topics: ["oracle_update", "protocol_status"],
+ coalesceMs: { oracle_update: 5_000 },
+ transport: "websocket",
+ },
+ minimal: { topics: ["protocol_status"], coalesceMs: {}, transport: "polling" },
+};
+
+export const TIER_ORDER: readonly FidelityTier[] = ["full", "lite", "minimal"];
+
+/** Hold times before a tier change is applied, so a flapping connection does not thrash. */
+export const DOWNGRADE_HOLD_MS = 3_000;
+export const UPGRADE_HOLD_MS = 30_000;
+
+export interface FidelitySignals {
+ effectiveType?: string;
+ saveData?: boolean;
+ batteryLevel?: number;
+ charging?: boolean;
+ deviceMemory?: number;
+ hardwareConcurrency?: number;
+}
+
+/** Pure tier selection from signals; missing signals assume an unconstrained client. */
+export function selectTier(s: FidelitySignals): FidelityTier {
+ if (s.saveData || s.effectiveType === "slow-2g" || s.effectiveType === "2g") return "minimal";
+ const lowBattery = s.batteryLevel !== undefined && s.batteryLevel < 0.15 && !s.charging;
+ const weakDevice =
+ (s.deviceMemory !== undefined && s.deviceMemory <= 2) ||
+ (s.hardwareConcurrency !== undefined && s.hardwareConcurrency <= 2);
+ if (lowBattery) return "minimal";
+ if (
+ s.effectiveType === "3g" ||
+ weakDevice ||
+ (s.batteryLevel !== undefined && s.batteryLevel < 0.3 && !s.charging)
+ ) {
+ return "lite";
+ }
+ return "full";
+}
+
+type NavigatorSignals = Navigator & {
+ connection?: {
+ effectiveType?: string;
+ saveData?: boolean;
+ addEventListener?: Navigator["addEventListener"];
+ removeEventListener?: Navigator["removeEventListener"];
+ };
+ deviceMemory?: number;
+ getBattery?: () => Promise<{
+ level: number;
+ charging: boolean;
+ addEventListener: EventTarget["addEventListener"];
+ removeEventListener: EventTarget["removeEventListener"];
+ }>;
+};
+
+/** SSR-safe, never throws. */
+export function readSignals(
+ battery?: { level: number; charging: boolean } | null
+): FidelitySignals {
+ if (typeof navigator === "undefined") return {};
+ const nav = navigator as NavigatorSignals;
+ try {
+ return {
+ effectiveType: nav.connection?.effectiveType,
+ saveData: nav.connection?.saveData,
+ deviceMemory: nav.deviceMemory,
+ hardwareConcurrency: nav.hardwareConcurrency,
+ batteryLevel: battery?.level,
+ charging: battery?.charging,
+ };
+ } catch {
+ return {};
+ }
+}
+
+type TierListener = (tier: FidelityTier) => void;
+
+/**
+ * Owns the active tier. Candidate tiers from signals are applied only after a
+ * hold period (hysteresis); manual override applies immediately. Topic changes
+ * are pushed to the client as a diff and are applied synchronously, so two
+ * rapid changes cannot interleave.
+ */
+export class FidelityController {
+ private tier: FidelityTier = "full";
+ private override: FidelityTier | null = null;
+ private pending: { tier: FidelityTier; timer: ReturnType } | null = null;
+ private required = new Map();
+ private listeners = new Set();
+ private battery: {
+ level: number;
+ charging: boolean;
+ addEventListener: EventTarget["addEventListener"];
+ removeEventListener: EventTarget["removeEventListener"];
+ } | null = null;
+ private teardown: (() => void)[] = [];
+ private started = false;
+
+ getTier(): FidelityTier {
+ return this.override ?? this.tier;
+ }
+
+ getOverride(): FidelityTier | null {
+ return this.override;
+ }
+
+ onChange(listener: TierListener): () => void {
+ this.listeners.add(listener);
+ return () => {
+ this.listeners.delete(listener);
+ };
+ }
+
+ /** Topics a currently-visible critical action needs; never dropped by a downgrade. */
+ requireTopic(topic: RealtimeTopic): () => void {
+ this.required.set(topic, (this.required.get(topic) ?? 0) + 1);
+ this.applyTopics();
+ let released = false;
+ return () => {
+ if (released) return;
+ released = true;
+ const n = (this.required.get(topic) ?? 1) - 1;
+ if (n <= 0) this.required.delete(topic);
+ else this.required.set(topic, n);
+ this.applyTopics();
+ };
+ }
+
+ setOverride(tier: FidelityTier | null): void {
+ this.override = tier;
+ this.clearPending();
+ this.applyTopics();
+ this.emit();
+ }
+
+ /** Feed new signals; applies the resulting tier after the hysteresis hold. */
+ evaluate(signals: FidelitySignals): void {
+ const candidate = selectTier(signals);
+ if (candidate === this.tier) return this.clearPending();
+ if (this.pending?.tier === candidate) return;
+ this.clearPending();
+ const downgrade = TIER_ORDER.indexOf(candidate) > TIER_ORDER.indexOf(this.tier);
+ this.pending = {
+ tier: candidate,
+ timer: setTimeout(
+ () => {
+ this.pending = null;
+ this.tier = candidate;
+ this.applyTopics();
+ this.emit();
+ },
+ downgrade ? DOWNGRADE_HOLD_MS : UPGRADE_HOLD_MS
+ ),
+ };
+ }
+
+ start(): () => void {
+ if (this.started || typeof navigator === "undefined") return () => {};
+ this.started = true;
+ const nav = navigator as NavigatorSignals;
+ const reevaluate = () => this.evaluate(readSignals(this.battery));
+ const conn = nav.connection;
+ conn?.addEventListener?.("change", reevaluate);
+ this.teardown.push(() => conn?.removeEventListener?.("change", reevaluate));
+ // Re-assert the topic set on every (re)connect; the client also re-sends it.
+ this.teardown.push(realtimeClient.onConnectionChange(() => this.applyTopics()));
+ try {
+ void nav
+ .getBattery?.()
+ .then((b) => {
+ if (!this.started) return;
+ this.battery = b;
+ b.addEventListener("levelchange", reevaluate);
+ b.addEventListener("chargingchange", reevaluate);
+ this.teardown.push(() => {
+ b.removeEventListener("levelchange", reevaluate);
+ b.removeEventListener("chargingchange", reevaluate);
+ });
+ reevaluate();
+ })
+ .catch(() => {});
+ } catch {
+ // Battery API unavailable — assume unconstrained.
+ }
+ reevaluate();
+ return () => this.stop();
+ }
+
+ stop(): void {
+ this.started = false;
+ this.clearPending();
+ this.teardown.forEach((fn) => fn());
+ this.teardown = [];
+ this.battery = null;
+ }
+
+ private clearPending(): void {
+ if (this.pending) clearTimeout(this.pending.timer);
+ this.pending = null;
+ }
+
+ private applyTopics(): void {
+ const profile = FIDELITY_PROFILES[this.getTier()];
+ const topics = new Set(profile.transport === "websocket" ? profile.topics : []);
+ this.required.forEach((_, t) => topics.add(t));
+ realtimeClient.syncTopics(topics);
+ }
+
+ private emit(): void {
+ const tier = this.getTier();
+ this.listeners.forEach((l) => l(tier));
+ }
+}
+
+export const fidelityController = new FidelityController();
diff --git a/src/lib/realtime/index.ts b/src/lib/realtime/index.ts
index 525ff5a..e69de29 100644
--- a/src/lib/realtime/index.ts
+++ b/src/lib/realtime/index.ts
@@ -1,11 +0,0 @@
-/** Public entry point of the realtime module — import from "@/lib/realtime", never from internals. */
-export { realtimeClient } from "./client";
-export {
- initRealtimeFlags,
- loadRealtimeFlags,
- isTopicEnabled,
- setTopicOverride,
- killTopic,
- reviveTopic,
- onTopicFlagChange,
-} from "./flags";
diff --git a/src/lib/realtime/realtime.test.ts b/src/lib/realtime/realtime.test.ts
index e5f9ffd..e69de29 100644
--- a/src/lib/realtime/realtime.test.ts
+++ b/src/lib/realtime/realtime.test.ts
@@ -1,79 +0,0 @@
-import { describe, it, expect, vi, afterEach } from "vitest";
-import { Coalescer, TopicSequencer, type Envelope } from "./merge";
-import { SseTransport } from "./sse";
-
-const env = (seq: number, kind: Envelope["kind"] = "replace"): Envelope => ({ topic: "t", seq, kind, payload: seq });
-
-describe("TopicSequencer", () => {
- it("reorders within the window and drops duplicates", () => {
- const gap = vi.fn();
- const s = new TopicSequencer(gap, 8);
- expect(s.push(env(1)).map((e) => e.seq)).toEqual([1]);
- expect(s.push(env(3))).toEqual([]);
- expect(s.push(env(2)).map((e) => e.seq)).toEqual([2, 3]);
- expect(s.push(env(2))).toEqual([]);
- expect(gap).not.toHaveBeenCalled();
- });
-
- it("flags a gap when the window overflows", () => {
- const gap = vi.fn();
- const s = new TopicSequencer(gap, 2);
- s.push(env(1));
- s.push(env(3));
- s.push(env(4));
- const out = s.push(env(5));
- expect(gap).toHaveBeenCalledWith(2, 2);
- expect(out.map((e) => e.seq)).toEqual([3, 4, 5]);
- });
-
- it("flush() resolves an outstanding hole", () => {
- const gap = vi.fn();
- const s = new TopicSequencer(gap);
- s.push(env(1));
- s.push(env(4));
- expect(s.flush().map((e) => e.seq)).toEqual([4]);
- expect(gap).toHaveBeenCalledWith(2, 3);
- });
-});
-
-describe("Coalescer", () => {
- afterEach(() => vi.useRealTimers());
-
- it("keeps latest replace event but never drops append events", () => {
- vi.useFakeTimers();
- const deliver = vi.fn();
- const c = new Coalescer(deliver, 50);
- for (let i = 0; i < 5_000; i++) c.add({ topic: "oracle", kind: "replace", payload: i });
- for (let i = 0; i < 1_000; i++) c.add({ topic: "claims", kind: "append", payload: i });
- vi.advanceTimersByTime(60);
- const oracle = deliver.mock.calls.filter(([t]) => t === "oracle");
- const claims = deliver.mock.calls.filter(([t]) => t === "claims");
- expect(oracle).toEqual([["oracle", 4_999]]);
- expect(claims).toHaveLength(1_000);
- });
-});
-
-describe("SseTransport", () => {
- afterEach(() => vi.unstubAllGlobals());
-
- it("passes the token in the query and closes on error so the client owns retry", () => {
- const close = vi.fn();
- const instances: { url: string; onerror?: () => void }[] = [];
- vi.stubGlobal(
- "EventSource",
- class {
- onerror?: () => void;
- close = close;
- constructor(public url: string) {
- instances.push(this);
- }
- }
- );
- const onClose = vi.fn();
- new SseTransport("http://x/events", "http://x/sub", () => "tok").connect({ onOpen: vi.fn(), onMessage: vi.fn(), onClose });
- expect(instances[0].url).toBe("http://x/events?token=tok");
- instances[0].onerror?.();
- expect(close).toHaveBeenCalled();
- expect(onClose).toHaveBeenCalled();
- });
-});
diff --git a/src/lib/realtime/status.ts b/src/lib/realtime/status.ts
new file mode 100644
index 0000000..ea11a0c
--- /dev/null
+++ b/src/lib/realtime/status.ts
@@ -0,0 +1,13 @@
+import { apiRequest } from "@/lib/api/client";
+import { ACTIVE_NETWORK } from "@/lib/network";
+import type { ProtocolStatus } from "./topics";
+
+/** REST snapshot; falls back to "unknown" so the page is useful with the backend (or realtime) down. */
+export async function fetchProtocolStatus(): Promise {
+ try {
+ const status = await apiRequest("/status");
+ return { ...status, network: status.network ?? ACTIVE_NETWORK.id };
+ } catch {
+ return { level: "unknown", updatedAt: new Date().toISOString(), network: ACTIVE_NETWORK.id };
+ }
+}
diff --git a/src/lib/realtime/topics.ts b/src/lib/realtime/topics.ts
index 7260f9c..e69de29 100644
--- a/src/lib/realtime/topics.ts
+++ b/src/lib/realtime/topics.ts
@@ -1,24 +0,0 @@
-/** Typed topic map for the realtime client. */
-
-/** Pseudonymous, pool-level activity only — never a wallet address or position size. */
-export interface PoolActivityEvent {
- id: string;
- kind: "deposit" | "withdraw" | "coverage_written";
- /** Unix ms. */
- at: number;
- /** Aggregate USDC bucket label (e.g. "$1k–10k"), never an exact per-LP amount. */
- sizeBucket?: string;
-}
-
-export interface PoolPresence {
- /** Aggregate count of providers currently active. */
- active: number;
-}
-
-export interface TopicMap {
- oracle_update: unknown;
- pool_activity: PoolActivityEvent;
- pool_presence: PoolPresence;
-}
-
-export type TopicName = keyof TopicMap;