Repository Public
1/**2 * Live market state of ONE instrument, outside React. Stream events only fill a pending buffer; a requestAnimationFrame loop3 * commits it at most every BOOK_MS (ticker, book and trades together; connection changes at once) and wakes only the4 * components subscribed to the slice that changed (`useSlice`). Twenty book updates a second therefore cost a handful of small5 * renders, never a render of the screen.6 *7 * Freshness is never taken on trust: `freshness()` is 'live' only while the stream says `connected` AND an event arrived less8 * than STALE_MS ago. Anything else is shown as reconnecting or stale, with the age of the last data.9 */10import { useEffect, useMemo, useState, useSyncExternalStore } from 'react';11import { dapi } from './api';12import { mergeBook, mergeTrades } from './reducers';13import type { Book, Instrument, Liquidity, PublicTrade, StreamStatus, Ticker } from './types';1415export const STALE_MS = 10_000;16export const BOOK_MS = 125;17export type Slice = 'ticker' | 'book' | 'trades' | 'meta';18export type Conn = 'connecting' | StreamStatus | 'error';19export type Fresh = 'connecting' | 'live' | 'reconnecting' | 'stale';20export interface Meta { instrument: Instrument | null; liquidity: Liquidity | null; conn: Conn; error: string | null }21export interface Perf { events: number; commits: number; bookCommits: number; since: number }2223export class MarketStore {24 readonly id: string;25 ticker: Ticker | null = null;26 book: Book | null = null;27 trades: PublicTrade[] = [];28 meta: Meta = { instrument: null, liquidity: null, conn: 'connecting', error: null };29 /** wall-clock time of the last stream event (not a slice: read by `freshness()` once a second) */30 lastEventAt = 0;31 perf: Perf = { events: 0, commits: 0, bookCommits: 0, since: Date.now() };3233 private listeners: Record<Slice, Set<() => void>> = { ticker: new Set(), book: new Set(), trades: new Set(), meta: new Set() };34 private pTicker: Ticker | null = null;35 private pBook: Book | null = null;36 private pTrades: PublicTrade[] = [];37 private pMeta: Partial<Meta> | null = null;38 private raf = 0;39 private lastBookCommit = 0;40 private es: EventSource | null = null;41 private poll: ReturnType<typeof setInterval> | null = null;42 private closed = false;4344 constructor(id: string) { this.id = id; }4546 subscribe = (slice: Slice, f: () => void): (() => void) => { this.listeners[slice].add(f); return () => { this.listeners[slice].delete(f); }; };47 private emit(slice: Slice) { for (const f of this.listeners[slice]) f(); }4849 start(): void {50 this.closed = false;51 if (import.meta.env.DEV) (window as unknown as { __spwStore?: MarketStore }).__spwStore = this; // development only: lets the e2e test read `perf`52 void this.load();53 this.open();54 // safety net when the stream is silent: one snapshot every 5 s keeps the age honest and recovers after an outage55 this.poll = setInterval(() => { if (Date.now() - this.lastEventAt > 5000 && document.visibilityState === 'visible') void this.load(); }, 5000);56 }57 stop(): void {58 this.closed = true;59 this.es?.close(); this.es = null;60 if (this.poll) clearInterval(this.poll);61 if (this.raf) cancelAnimationFrame(this.raf);62 this.raf = 0; this.poll = null;63 }6465 private async load(): Promise<void> {66 try {67 const m = await dapi.market(this.id);68 if (this.closed) return;69 // a snapshot proves the data path works, not that the stream is live: it never moves `lastEventAt` forward on its own70 this.pMeta = { ...(this.pMeta ?? {}), instrument: m.instrument, liquidity: m.liquidity, error: null };71 if (m.ticker && (!this.ticker || m.ticker.ts >= this.ticker.ts)) this.pTicker = m.ticker;72 if (m.book) this.pBook = m.book;73 if (m.trades.length) this.pTrades.push(...m.trades);74 this.schedule();75 } catch (e) {76 if (this.closed) return;77 this.pMeta = { ...(this.pMeta ?? {}), error: (e as { code?: string })?.code ?? 'error' };78 this.schedule();79 }80 }8182 private open(): void {83 let es: EventSource;84 try { es = new EventSource(dapi.streamUrl(this.id)); } catch { this.setConn('error'); return; }85 this.es = es;86 const parse = (e: Event): any => { try { return JSON.parse((e as MessageEvent).data); } catch { return null; } };87 const seen = () => { this.perf.events++; this.lastEventAt = Date.now(); };88 es.addEventListener('open', () => this.setConn('connected'));89 es.addEventListener('ticker', (e) => { const t = parse(e); if (t && t.instrument === this.id) { seen(); this.pTicker = t; this.schedule(); } });90 es.addEventListener('book', (e) => { const b = parse(e); if (b && b.instrument === this.id) { seen(); this.pBook = b; this.schedule(); } });91 es.addEventListener('trades', (e) => { const t = parse(e); const list = Array.isArray(t?.trades) ? t.trades : null; if (list && t.instrument === this.id) { seen(); this.pTrades.push(...list); this.schedule(); } });92 es.addEventListener('status', (e) => { const s = parse(e); const v = typeof s === 'string' ? s : s?.status ?? s?.state; if (v === 'connected' || v === 'reconnecting' || v === 'stale') this.setConn(v); });93 // the browser reconnects by itself; until it does, nothing on screen may pass for current94 // (the server also sends a named `error` event with a message: that one carries data and is not a connection failure)95 es.addEventListener('error', (e) => { if (typeof (e as MessageEvent).data === 'string') return; this.setConn(es.readyState === EventSource.CLOSED ? 'error' : 'reconnecting'); });96 }97 private setConn(conn: Conn) { this.pMeta = { ...(this.pMeta ?? {}), conn }; this.schedule(); }9899 private schedule(): void { if (!this.raf && !this.closed) this.raf = requestAnimationFrame(this.flush); }100 private flush = (): void => {101 this.raf = 0;102 const now = performance.now();103 this.perf.commits++;104 if (this.pMeta) { const m = this.pMeta; this.pMeta = null; this.meta = { ...this.meta, ...m }; this.emit('meta'); }105 if (this.pTicker || this.pBook || this.pTrades.length) {106 if (now - this.lastBookCommit >= BOOK_MS) {107 this.lastBookCommit = now;108 if (this.pTicker) { const t = this.pTicker; this.pTicker = null; if (!this.ticker || t.ts >= this.ticker.ts) { this.ticker = t; this.emit('ticker'); } }109 if (this.pBook) { const next = mergeBook(this.book, this.pBook, this.id); this.pBook = null; if (next !== this.book) { this.book = next; this.perf.bookCommits++; this.emit('book'); } }110 if (this.pTrades.length) { const next = mergeTrades(this.trades, this.pTrades); this.pTrades = []; if (next !== this.trades) { this.trades = next; this.emit('trades'); } }111 } else this.schedule();112 }113 };114115 freshness(now = Date.now()): { state: Fresh; ageMs: number | null } {116 const { conn } = this.meta, lastEventAt = this.lastEventAt;117 const dataTs = Math.max(lastEventAt, 0);118 const ageMs = dataTs ? Math.max(0, now - dataTs) : this.ticker ? Math.max(0, now - this.ticker.ts) : null;119 if (conn === 'stale' || conn === 'error') return { state: 'stale', ageMs };120 if (conn === 'reconnecting') return { state: ageMs !== null && ageMs > STALE_MS ? 'stale' : 'reconnecting', ageMs };121 if (!lastEventAt) return { state: conn === 'connected' && this.ticker && now - this.perf.since > STALE_MS ? 'stale' : 'connecting', ageMs };122 return { state: ageMs !== null && ageMs > STALE_MS ? 'stale' : 'live', ageMs };123 }124}125126const sliceOf = { ticker: (s: MarketStore) => s.ticker, book: (s: MarketStore) => s.book, trades: (s: MarketStore) => s.trades, meta: (s: MarketStore) => s.meta } as const;127type SliceValue = { ticker: Ticker | null; book: Book | null; trades: PublicTrade[]; meta: Meta };128/** re-renders the calling component only when this slice changes */129export function useSlice<K extends Slice>(store: MarketStore, slice: K): SliceValue[K] {130 return useSyncExternalStore((f) => store.subscribe(slice, f), () => sliceOf[slice](store) as SliceValue[K]);131}132/** freshness re-evaluated every second and on every connection change */133export function useFreshness(store: MarketStore): { state: Fresh; ageMs: number | null } {134 const [f, setF] = useState(() => store.freshness());135 useEffect(() => {136 const tick = () => setF((p) => { const n = store.freshness(); return n.state === p.state && (n.state === 'live' || Math.floor((n.ageMs ?? 0) / 1000) === Math.floor((p.ageMs ?? 0) / 1000)) ? p : n; });137 tick();138 const h = setInterval(tick, 1000), off = store.subscribe('meta', tick);139 return () => { clearInterval(h); off(); };140 }, [store]);141 return f;142}143/** one store per instrument, started on mount and closed on unmount */144export function useMarketStore(id: string): MarketStore {145 const store = useMemo(() => new MarketStore(id), [id]);146 useEffect(() => { store.start(); return () => store.stop(); }, [store]);147 return store;148}149