Repository Public
1/**2 * The selected VENUE account, kept current: `GET /live/accounts/:id` (venue state, orders, fills, funding, risk events) and the3 * Server-Sent Events of `…/stream` merged into the same cache entry — `account` (the venue's state after each reconciliation),4 * `order`, `action`, `risk`, `status`. Everything shown comes from the server; nothing is predicted locally.5 * While the stream is up the account is polled slowly (the stream carries the changes); without it, every 8 s.6 */7import { useQuery, useQueryClient, type QueryClient } from '@tanstack/react-query';8import { useEffect, useRef, useState } from 'react';9import { lapi, type LiveAction, type LiveDetail, type LiveOrder } from './live';1011export const liveKey = (id: string | null | undefined) => ['deriv', 'live', 'account', id] as const;12export const refreshLive = (qc: QueryClient, id?: string) => Promise.all([qc.invalidateQueries({ queryKey: ['deriv', 'live', 'accounts'] }), qc.invalidateQueries({ queryKey: id ? liveKey(id) : ['deriv', 'live', 'account'] }),13 qc.invalidateQueries({ queryKey: ['deriv', 'live', 'collateral'] })]);1415/** an `order` event replaces the row with the same client id, or is put first */16export function mergeOrder(list: LiveOrder[], o: LiveOrder): LiveOrder[] {17 const i = list.findIndex((x) => x.clientId === o.clientId);18 if (i < 0) return [o, ...list].slice(0, 100);19 if (list[i]!.updatedAt > o.updatedAt) return list;20 const next = [...list]; next[i] = o; return next;21}22/** one stream event applied to the cached account; unknown events change nothing */23export function applyLiveEvent(d: LiveDetail, event: string, data: any): LiveDetail {24 if (!data || typeof data !== 'object') return d;25 if (event === 'account') {26 const readable = data.state !== undefined && data.state !== null;27 return { ...d, venueState: { state: readable ? data.state : d.venueState.state, positions: Array.isArray(data.positions) ? data.positions : d.venueState.positions, openOrders: Array.isArray(data.openOrders) ? data.openOrders : d.venueState.openOrders,28 error: typeof data.error === 'string' ? data.error : readable ? null : d.venueState.error, ts: Number(data.ts) || d.venueState.ts, source: 'venue' },29 divergence: data.divergence === undefined ? d.divergence : data.divergence, halt: data.halt === undefined ? d.halt : data.halt, mode: data.mode === 'MANUAL_LIVE' || data.mode === 'READ_ONLY' ? data.mode : d.mode };30 }31 if (event === 'order' && typeof data.clientId === 'string') return { ...d, orders: mergeOrder(d.orders, data as LiveOrder) };32 if (event === 'action' && typeof data.id === 'string') { const a = data as LiveAction, i = d.actions.findIndex((x) => x.id === a.id); return { ...d, actions: i < 0 ? [a, ...d.actions].slice(0, 50) : d.actions.map((x) => (x.id === a.id ? a : x)) }; }33 if (event === 'risk') return { ...d, riskEvents: [{ id: `s${data.ts}${data.kind}${d.riskEvents.length}`, ts: Number(data.ts) || Date.now(), kind: String(data.kind ?? 'event'), code: typeof data.code === 'string' ? data.code : null, message: String(data.message ?? ''), data: data.data ?? {} }, ...d.riskEvents].slice(0, 100) };34 return d;35}3637export type LiveConn = 'connecting' | 'stream' | 'polling';38export function useLiveAccount(id: string | null): { detail: LiveDetail | undefined; error: unknown; conn: LiveConn; /** the reconciliation worker could not read the venue */ workerError: string | null; refresh: () => void } {39 const qc = useQueryClient();40 const [conn, setConn] = useState<LiveConn>('connecting'), [workerError, setWorkerError] = useState<string | null>(null);41 const q = useQuery({ queryKey: liveKey(id), enabled: !!id, queryFn: () => lapi.account(id!), refetchInterval: conn === 'stream' ? 30_000 : 8000, retry: 0 });42 const pending = useRef<ReturnType<typeof setTimeout> | null>(null);43 useEffect(() => {44 if (!id) return;45 setConn('connecting'); setWorkerError(null);46 let es: EventSource | null = null;47 try { es = new EventSource(lapi.streamUrl(id)); } catch { setConn('polling'); return; }48 const refetchSoon = () => { if (pending.current) return; pending.current = setTimeout(() => { pending.current = null; void qc.invalidateQueries({ queryKey: liveKey(id) }); void qc.invalidateQueries({ queryKey: ['deriv', 'live', 'accounts'] }); }, 400); };49 const on = (event: string) => (e: Event) => {50 let data: any; try { data = JSON.parse((e as MessageEvent).data); } catch { return; }51 setConn('stream');52 if (event === 'status') { setWorkerError(typeof data?.error === 'string' ? data.error : null); refetchSoon(); return; } // mode / halt / divergence changed: read the account again53 if (event === 'account') setWorkerError(null);54 qc.setQueryData<LiveDetail>(liveKey(id), (d) => (d ? applyLiveEvent(d, event, data) : d));55 if (event === 'order' || event === 'action' || event === 'risk') refetchSoon(); // fills, funding and events are read from the account route56 };57 for (const ev of ['status', 'account', 'order', 'action', 'risk']) es.addEventListener(ev, on(ev));58 es.addEventListener('open', () => setConn('stream'));59 // the browser reconnects by itself; meanwhile (or when the server refuses a ninth stream) the account is polled60 es.addEventListener('error', (e) => { if (typeof (e as MessageEvent).data === 'string') return; setConn('polling'); });61 return () => { es?.close(); if (pending.current) { clearTimeout(pending.current); pending.current = null; } };62 }, [id, qc]);63 return { detail: id && q.data?.id === id ? q.data : undefined, error: q.error, conn, workerError, refresh: () => { void qc.invalidateQueries({ queryKey: liveKey(id) }); } };64}65