spb/market-atlas

1"use client";23import { useEffect, useMemo, useRef, useSyncExternalStore } from "react";4import type { MarketEvent, StreamMessage, StreamQuote } from "./types";56/**7 * Market Atlas stream client (singleton per tab). One WebSocket, reference-counted channel8 * subscriptions, a quote store keyed by instrument id, and per-key listeners notified at most9 * every ~100 ms so tables never re-render per tick.10 */11type Listener = () => void;1213export type ConnectionState = "idle" | "connecting" | "open" | "reconnecting" | "closed";1415class MarketStreamClient {16  private ws: WebSocket | null = null;17  private refs = new Map<string, number>();18  private wanted = new Set<string>();19  private sent = new Set<string>();20  private quotes = new Map<string, StreamQuote>();21  private dirty = new Set<string>();22  private listeners = new Map<string, Set<Listener>>();23  private eventListeners = new Set<Listener>();24  private events: MarketEvent[] = [];25  private eventsVersion = 0;26  private tape: StreamQuote[] = [];27  private tapeVersion = 0;28  private flushTimer: ReturnType<typeof setTimeout> | null = null;29  private attempt = 0;30  private timer: ReturnType<typeof setTimeout> | null = null;31  state: ConnectionState = "idle";32  private stateListeners = new Set<Listener>();33  lastSeq = 0;34  frames = 0;3536  private url(): string {37    const env = process.env.NEXT_PUBLIC_WS_URL;38    if (env) return env;39    const proto = window.location.protocol === "https:" ? "wss" : "ws";40    return `${proto}://${window.location.host}/v1/stream`;41  }4243  private setState(s: ConnectionState) {44    if (this.state === s) return;45    this.state = s;46    for (const l of this.stateListeners) l();47  }4849  private ensure() {50    if (typeof window === "undefined") return;51    if (this.ws || this.timer) return;52    if (!this.wanted.size) return;53    this.setState(this.attempt ? "reconnecting" : "connecting");54    let ws: WebSocket;55    try {56      ws = new WebSocket(this.url());57    } catch {58      this.scheduleReconnect();59      return;60    }61    this.ws = ws;62    ws.onopen = () => {63      this.attempt = 0;64      this.sent.clear();65      this.setState("open");66      this.sync();67    };68    ws.onmessage = (ev) => this.onFrame(String(ev.data));69    ws.onclose = () => {70      this.ws = null;71      this.sent.clear();72      if (this.wanted.size) this.scheduleReconnect();73      else this.setState("closed");74    };75    ws.onerror = () => {76      /* onclose follows */77    };78  }7980  private scheduleReconnect() {81    if (this.timer) return;82    this.setState("reconnecting");83    const wait = Math.min(30_000, 1000 * 2 ** Math.min(6, this.attempt)) * (0.5 + Math.random());84    this.attempt++;85    this.timer = setTimeout(() => {86      this.timer = null;87      this.ensure();88    }, wait);89  }9091  private sync() {92    if (!this.ws || this.ws.readyState !== WebSocket.OPEN) return;93    const add = [...this.wanted].filter((c) => !this.sent.has(c));94    const del = [...this.sent].filter((c) => !this.wanted.has(c));95    if (add.length) this.ws.send(JSON.stringify({ action: "subscribe", channels: add }));96    if (del.length) this.ws.send(JSON.stringify({ action: "unsubscribe", channels: del }));97    for (const c of add) this.sent.add(c);98    for (const c of del) this.sent.delete(c);99    if (!this.wanted.size) {100      this.ws.close();101    }102  }103104  subscribe(channels: string[]): () => void {105    for (const c of channels) {106      this.refs.set(c, (this.refs.get(c) ?? 0) + 1);107      this.wanted.add(c);108    }109    this.ensure();110    this.sync();111    return () => {112      for (const c of channels) {113        const n = (this.refs.get(c) ?? 1) - 1;114        if (n <= 0) {115          this.refs.delete(c);116          this.wanted.delete(c);117        } else this.refs.set(c, n);118      }119      // Debounce unsubscribes so quick navigations reuse the socket.120      setTimeout(() => this.sync(), 500);121    };122  }123124  private onFrame(raw: string) {125    let msg: { type?: string; seq?: number; messages?: StreamMessage[] };126    try {127      msg = JSON.parse(raw);128    } catch {129      return;130    }131    if (msg.type !== "batch" || !Array.isArray(msg.messages)) return;132    this.frames++;133    if (typeof msg.seq === "number") this.lastSeq = msg.seq;134    for (const m of msg.messages) {135      if (m.type === "quote") {136        const prev = this.quotes.get(m.instrument_id);137        if (prev && prev.received > m.received) continue;138        this.quotes.set(m.instrument_id, m);139        this.dirty.add(m.instrument_id);140        this.dirty.add(`symbol:${m.symbol.toUpperCase()}`);141        if (!prev || prev.price !== m.price) {142          this.tape.push(m);143          if (this.tape.length > 200) this.tape.splice(0, this.tape.length - 200);144          this.tapeVersion++;145          this.dirty.add("__tape__");146        }147      } else if (m.type === "event") {148        this.events.unshift(m.event);149        if (this.events.length > 500) this.events.length = 500;150        this.eventsVersion++;151        this.dirty.add("__events__");152      } else if (m.type === "market_state") {153        this.dirty.add("__market__");154      }155    }156    this.scheduleFlush();157  }158159  private scheduleFlush() {160    if (this.flushTimer) return;161    this.flushTimer = setTimeout(() => {162      this.flushTimer = null;163      const keys = [...this.dirty];164      this.dirty.clear();165      for (const k of keys) {166        const ls = this.listeners.get(k);167        if (ls) for (const l of ls) l();168      }169      this.dirty.clear();170    }, 100);171  }172173  listen(key: string, l: Listener): () => void {174    let s = this.listeners.get(key);175    if (!s) {176      s = new Set();177      this.listeners.set(key, s);178    }179    s.add(l);180    return () => {181      s!.delete(l);182      if (!s!.size) this.listeners.delete(key);183    };184  }185186  listenState(l: Listener): () => void {187    this.stateListeners.add(l);188    return () => this.stateListeners.delete(l);189  }190191  quote(id: string): StreamQuote | undefined {192    return this.quotes.get(id);193  }194  quoteBySymbol(symbol: string): StreamQuote | undefined {195    const u = symbol.toUpperCase();196    for (const q of this.quotes.values()) if (q.symbol.toUpperCase() === u) return q;197    return undefined;198  }199  recentEvents(): MarketEvent[] {200    return this.events;201  }202  eventsSnapshot(): number {203    return this.eventsVersion;204  }205  tapeSnapshot(): number {206    return this.tapeVersion;207  }208  recentTape(): StreamQuote[] {209    return this.tape;210  }211}212213let client: MarketStreamClient | null = null;214export function marketStream(): MarketStreamClient {215  if (!client) client = new MarketStreamClient();216  return client;217}218219/** Keep the given channels subscribed while the component is mounted. */220export function useMarketStream(channels: string[]): ConnectionState {221  const key = channels.join("|");222  useEffect(() => {223    if (!channels.length) return;224    return marketStream().subscribe(channels);225    // eslint-disable-next-line react-hooks/exhaustive-deps226  }, [key]);227  return useSyncExternalStore(228    (l) => marketStream().listenState(l),229    () => marketStream().state,230    () => "idle" as ConnectionState,231  );232}233234/** Latest streamed quote for an instrument id (undefined until the first frame). Re-renders only that consumer. */235export function useLiveQuote(instrumentId: string | null | undefined): StreamQuote | undefined {236  const id = instrumentId ?? "";237  return useSyncExternalStore(238    (l) => (id ? marketStream().listen(id, l) : () => {}),239    () => (id ? marketStream().quote(id) : undefined),240    () => undefined,241  );242}243244/** Events received on the stream since mount (newest first), capped. */245export function useLiveEvents(max = 100): MarketEvent[] {246  const version = useSyncExternalStore(247    (l) => marketStream().listen("__events__", l),248    () => marketStream().eventsSnapshot(),249    () => 0,250  );251  const ref = useRef<MarketEvent[]>([]);252  return useMemo(() => {253    void version;254    ref.current = marketStream().recentEvents().slice(0, max);255    return ref.current;256  }, [version, max]);257}258259/** Recent price changes (tape). */260export function useTape(max = 60): StreamQuote[] {261  const version = useSyncExternalStore(262    (l) => marketStream().listen("__tape__", l),263    () => marketStream().tapeSnapshot(),264    () => 0,265  );266  return useMemo(() => {267    void version;268    const t = marketStream().recentTape();269    return t.slice(Math.max(0, t.length - max)).reverse();270  }, [version, max]);271}272273/** Ticks once per second — for relative "x s ago" labels. */274export function useNow(intervalMs = 1000): number {275  return useSyncExternalStore(276    (l) => {277      const t = setInterval(l, intervalMs);278      return () => clearInterval(t);279    },280    () => Math.floor(Date.now() / intervalMs) * intervalMs,281    () => 0,282  );283}284