feat(realtime): SSE fallback, ordering/backpressure, tab dedup, live pool stats

public/flags.json

@@ -1,5 +1,6 @@
 {
   "flags": {
-    "realtime-oracle": false
+    "realtime-oracle": false,
+    "realtime-pool-stats": false
   }
 }

src/hooks/usePoolStats.ts

@@ -0,0 +1,111 @@
+"use client";
+
+import { useEffect, useState } from "react";
+import { useQuery, useQueryClient } from "@tanstack/react-query";
+import { ApiUnreachableError } from "@/lib/api/client";
+import type { DataSource } from "@/lib/api/dataSource";
+import { fetchPoolStats, type PoolStats } from "@/lib/api/pool";
+import { FIXTURE_POOL_STATS } from "@/lib/fixtures/poolStats";
+import { useFeatureFlag } from "@/lib/flags/useFeatureFlag";
+import { realtimeClient } from "@/lib/realtime/client";
+import { readPoolStats } from "@/lib/stellar/rpc";
+
+interface PoolStatsState {
+  data: PoolStats | null;
+  loading: boolean;
+  error: string | null;
+  isFixture: boolean;
+  dataSource: DataSource | null;
+  /** True when the live stream is enabled but disconnected, so `data` is the REST snapshot. */
+  isStale: boolean;
+}
+
+interface PoolStatsResult {
+  data: PoolStats | null;
+  dataSource: DataSource | null;
+  error: string | null;
+}
+
+const NUMERIC_FIELDS: (keyof PoolStats)[] = ["utilizationBps", "apyBps", "sharePrice", "maxUtilizationBps"];
+const STRING_FIELDS: (keyof PoolStats)[] = ["totalUsdc", "totalShares", "lockedUsdc", "premiumAccrued", "availableUsdc"];
+/** A pushed TVL differing from REST by more than this ratio is treated as a bad frame. */
+const MAX_TVL_DRIFT = 0.5;
+
+/** Pushed frames are full snapshots; reject anything partial rather than half-applying it. */
+function isFullSnapshot(p: unknown): p is PoolStats {
+  const o = p as Record<string, unknown> | null;
+  return (
+    !!o &&
+    NUMERIC_FIELDS.every((k) => typeof o[k] === "number" && Number.isFinite(o[k] as number)) &&
+    STRING_FIELDS.every((k) => typeof o[k] === "string" && /^\d+$/.test(o[k] as string))
+  );
+}
+
+function agreesWithRest(pushed: PoolStats, rest: PoolStats | null): boolean {
+  const base = Number(rest?.totalUsdc ?? 0);
+  if (!rest || base === 0) return true;
+  return Math.abs(Number(pushed.totalUsdc) - base) / base <= MAX_TVL_DRIFT;
+}
+
+/**
+ * Pool-wide stats: API → on-chain read → bundled fixture, optionally kept live
+ * by the `pool.stats` realtime topic (behind the `realtime-pool-stats` flag).
+ */
+export function usePoolStats(): PoolStatsState {
+  const live = useFeatureFlag("realtime-pool-stats");
+  const queryClient = useQueryClient();
+  const [pushed, setPushed] = useState<PoolStats | null>(null);
+  const [connected, setConnected] = useState(realtimeClient.getStatus() === "open");
+
+  const { data: result, isLoading } = useQuery<PoolStatsResult>({
+    queryKey: ["poolStats"],
+    queryFn: async ({ signal }) => {
+      try {
+        return { data: await fetchPoolStats(signal), dataSource: "api", error: null };
+      } catch (err) {
+        if (signal?.aborted) throw err;
+        if (!(err instanceof ApiUnreachableError)) {
+          return { data: null, dataSource: null, error: err instanceof Error ? err.message : "Failed to load pool stats" };
+        }
+        try {
+          return { data: await readPoolStats(signal), dataSource: "chain", error: null };
+        } catch {
+          return { data: FIXTURE_POOL_STATS, dataSource: "fixture", error: null };
+        }
+      }
+    },
+  });
+
+  const rest = result?.data ?? null;
+
+  useEffect(() => {
+    if (!live) return;
+    setConnected(realtimeClient.getStatus() === "open");
+    const offStatus = realtimeClient.onStatus((s) => setConnected(s === "open"));
+    const off = realtimeClient.subscribe("pool.stats", (payload) => {
+      if (!isFullSnapshot(payload)) return;
+      const current = queryClient.getQueryData<PoolStatsResult>(["poolStats"])?.data ?? null;
+      if (agreesWithRest(payload, current)) setPushed(payload);
+      else {
+        setPushed(null);
+        void queryClient.invalidateQueries({ queryKey: ["poolStats"] });
+      }
+    });
+    return () => {
+      off();
+      offStatus();
+    };
+  }, [live, queryClient]);
+
+  // On disconnect fall back to the REST snapshot.
+  const useLive = live && connected && pushed !== null;
+
+  return {
+    data: useLive ? pushed : rest,
+    loading: isLoading,
+    error: result?.error ?? null,
+    isFixture: result?.dataSource === "fixture",
+    dataSource: result?.dataSource ?? null,
+    isStale: live && !connected,
+  };
+}

src/lib/flags/flagsConfig.ts

@@ -5,6 +5,7 @@ export interface FlagsDocument {
 /** Safe default if the remote flags source is unreachable — new/risky features default off. */
 export const DEFAULT_FLAGS: Record<string, boolean> = {
   "realtime-oracle": false,
+  "realtime-pool-stats": false,
 };
 
 export const FLAGS_ENDPOINT = "/flags.json";

src/lib/realtime/client.ts

@@ -1,59 +1,155 @@
 import { API_BASE_URL } from "@/lib/api/client";
+import { Coalescer, TopicSequencer, type Envelope } from "./merge";
+import { SseTransport } from "./sse";
+import { TabLeader } from "./tabs";
+import { WebSocketTransport, type Transport, type TransportKind } from "./transport";
 
 type Handler = (payload: unknown) => void;
+type Status = "idle" | "connecting" | "open" | "closed";
 
 const MAX_BACKOFF_MS = 30_000;
 const BASE_BACKOFF_MS = 1_000;
+const GAP_TIMEOUT_MS = 2_000;
 
 /**
- * A small reconnecting WebSocket wrapper exposing a subscribe-by-topic API.
+ * A small reconnecting realtime 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.
+ * upstream): JSON messages shaped
+ * `{ topic: string, seq?: number, kind?: "replace" | "append", payload: unknown }`.
+ * WebSocket is preferred; if it never opens, the client falls back to SSE
+ * (outbound `subscribe`/`resync` messages are POSTed since SSE is read-only).
+ * Only one browser tab holds the connection; others receive its messages over
+ * BroadcastChannel. Callers must treat this connection as best-effort — always
+ * keep an independent REST fetch as the source of truth for initial state.
  */
 class RealtimeClient {
-  private socket: WebSocket | null = null;
+  private transport: Transport | null = null;
   private handlers = new Map<string, Set<Handler>>();
+  private sequencers = new Map<string, TopicSequencer>();
+  private gapTimers = new Map<string, ReturnType<typeof setTimeout>>();
+  private coalescer = new Coalescer((topic, payload) => this.handlers.get(topic)?.forEach((h) => h(payload)));
+  private statusListeners = new Set<(status: Status) => void>();
+  private status: Status = "idle";
   private attempt = 0;
+  private everOpened = false;
+  private useSse = false;
   private closedByUser = false;
+  private started = false;
   private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
+  private leader: TabLeader | null = null;
+  private subscribed = new Set<string>();
+
+  get activeTransport(): TransportKind | null {
+    return this.transport?.kind ?? null;
+  }
+
+  getStatus(): Status {
+    return this.status;
+  }
+
+  onStatus(listener: (status: Status) => void): () => void {
+    this.statusListeners.add(listener);
+    return () => this.statusListeners.delete(listener);
+  }
+
+  private setStatus(status: Status): void {
+    if (this.status === status) return;
+    this.status = status;
+    this.statusListeners.forEach((l) => l(status));
+  }
 
   connect(): void {
-    if (typeof WebSocket === "undefined" || this.socket) return;
+    if (typeof WebSocket === "undefined" || this.started) return;
+    this.started = true;
     this.closedByUser = false;
-    this.open();
+    this.leader = new TabLeader(
+      (isLeader) => (isLeader ? this.open() : this.closeTransport()),
+      (raw) => this.ingest(raw),
+      () => this.syncTopics()
+    );
+    this.leader.start();
+  }
+
+  private baseUrl(): string {
+    return API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws");
   }
 
   private open(): void {
-    const wsUrl = API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws");
+    if (this.transport || !this.leader?.isLeader) return;
+    const wsUrl = this.baseUrl();
+    const httpUrl = API_BASE_URL.replace(/\/api\/v1$/, "");
+    this.transport = this.useSse
+      ? new SseTransport(`${httpUrl}/events`, `${httpUrl}/events/subscribe`)
+      : new WebSocketTransport(wsUrl);
+    this.setStatus("connecting");
+    const transport = this.transport;
+    transport.connect({
+      onOpen: () => {
+        this.attempt = 0;
+        this.subscribed.clear();
+        this.everOpened = true;
+        this.setStatus("open");
+        this.syncTopics();
+      },
+      onMessage: (raw) => {
+        this.leader?.broadcast(raw);
+        this.ingest(raw);
+      },
+      onClose: () => {
+        if (this.transport !== transport) return;
+        this.transport = null;
+        this.setStatus("closed");
+        // WebSocket never established (blocked proxy): switch to SSE.
+        if (!this.everOpened && !this.useSse && typeof EventSource !== "undefined") this.useSse = true;
+        if (!this.closedByUser) this.scheduleReconnect();
+      },
+    });
+  }
+
+  private closeTransport(): void {
+    const t = this.transport;
+    this.transport = null;
+    t?.close();
+  }
+
+  private ingest(raw: string): void {
+    let env: Envelope;
     try {
-      this.socket = new WebSocket(wsUrl);
+      env = JSON.parse(raw) as Envelope;
     } catch {
-      this.scheduleReconnect();
-      return;
+      return; // Malformed message — ignore, don't crash the connection.
     }
+    if (typeof env?.topic !== "string") return;
 
-    this.socket.addEventListener("open", () => {
-      this.attempt = 0;
-    });
-    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.
-      }
-    });
-    this.socket.addEventListener("close", () => {
-      this.socket = null;
-      if (!this.closedByUser) this.scheduleReconnect();
-    });
-    this.socket.addEventListener("error", () => {
-      this.socket?.close();
-    });
+    let seqr = this.sequencers.get(env.topic);
+    if (!seqr) {
+      seqr = new TopicSequencer((fromSeq, toSeq) => this.transport?.send({ type: "resync", topic: env.topic, fromSeq, toSeq }));
+      this.sequencers.set(env.topic, seqr);
+    }
+    const ready = seqr.push(env);
+    ready.forEach((e) => this.coalescer.add(e));
+
+    // A hole is outstanding if nothing was released for an out-of-order arrival: resolve it after a timeout.
+    clearTimeout(this.gapTimers.get(env.topic));
+    if (env.seq !== undefined && !ready.length) {
+      this.gapTimers.set(
+        env.topic,
+        setTimeout(() => {
+          seqr!.flush().forEach((e) => this.coalescer.add(e));
+        }, GAP_TIMEOUT_MS)
+      );
+    }
+  }
+
+  /** Topics needed by this tab or any other tab; the leader subscribes to the union. */
+  private syncTopics(): void {
+    const own = [...this.handlers.keys()].filter((t) => this.handlers.get(t)!.size > 0);
+    this.leader?.announceTopics(own);
+    if (!this.leader?.isLeader) return;
+    const all = new Set([...own, ...this.leader.remoteTopics()]);
+    all.forEach((topic) => !this.subscribed.has(topic) && this.transport?.send({ type: "subscribe", topic }));
+    this.subscribed.forEach((topic) => !all.has(topic) && this.transport?.send({ type: "unsubscribe", topic }));
+    this.subscribed = all;
   }
 
   private scheduleReconnect(): void {
@@ -67,20 +163,29 @@ class RealtimeClient {
   }
 
   subscribe(topic: string, handler: Handler): () => void {
+    const isNew = !this.handlers.get(topic)?.size;
     if (!this.handlers.has(topic)) this.handlers.set(topic, new Set());
     this.handlers.get(topic)!.add(handler);
     this.connect();
+    if (isNew) this.syncTopics();
 
     return () => {
       this.handlers.get(topic)?.delete(handler);
+      if (!this.handlers.get(topic)?.size) this.syncTopics();
     };
   }
 
   disconnect(): void {
     this.closedByUser = true;
+    this.started = false;
     if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
-    this.socket?.close();
-    this.socket = null;
+    this.reconnectTimer = null;
+    this.gapTimers.forEach(clearTimeout);
+    this.gapTimers.clear();
+    this.leader?.stop();
+    this.leader = null;
+    this.closeTransport();
+    this.setStatus("closed");
   }
 }