From cbab881cda1f7a4c3a15c7e962b6b254752a7477 Mon Sep 17 00:00:00 2001 From: prudentdev-xyz Date: Thu, 24 Sep 2026 19:39:34 +0100 Subject: [PATCH] feat(trade): shared market data subscriptions, bounded cache, and typed order events (OB-111, OB-109, OB-110, OB-112) - Manage one shared market-data subscription per active source with consumer reference-counting and bounded polling fallback (OB-111) - Bound inactive cache retention and server-state duplication across long sessions (OB-109) - Add cache correctness, scope isolation, and request-deduplication regression coverage (OB-110) - Replace loose order-event polling with typed event decoding, multi-page burst pagination, and cursor persistence (OB-112) --- ...6-shared-market-data-cache-order-events.md | 8 + apps/web/src/app/providers/QueryProvider.tsx | 7 +- .../components/orderbook/DepthLadder.tsx | 6 + .../components/orderbook/RecentTradesTape.tsx | 5 + .../src/features/trade/hooks/useLiveBar.ts | 156 +---- .../src/features/trade/hooks/useOrderBook.ts | 212 +----- .../trade/hooks/useOrderEventPolling.test.ts | 189 ++++++ .../trade/hooks/useOrderEventPolling.ts | 131 ++-- .../features/trade/hooks/useRecentTrades.ts | 160 +---- .../trade/lib/cache-correctness.test.tsx | 172 +++++ .../src/features/trade/lib/cache-policy.ts | 62 ++ .../trade/lib/cache-retention-long-session.md | 42 ++ .../trade/lib/market-data-stream.test.ts | 53 ++ .../features/trade/lib/market-data-stream.ts | 628 ++++++++++++++++++ .../features/trade/lib/order-event-decoder.ts | 133 ++++ .../src/features/wallet/store/wallet-store.ts | 29 +- 16 files changed, 1505 insertions(+), 488 deletions(-) create mode 100644 .changelog/unreleased/766-shared-market-data-cache-order-events.md create mode 100644 apps/web/src/features/trade/hooks/useOrderEventPolling.test.ts create mode 100644 apps/web/src/features/trade/lib/cache-correctness.test.tsx create mode 100644 apps/web/src/features/trade/lib/cache-policy.ts create mode 100644 apps/web/src/features/trade/lib/cache-retention-long-session.md create mode 100644 apps/web/src/features/trade/lib/market-data-stream.test.ts create mode 100644 apps/web/src/features/trade/lib/market-data-stream.ts create mode 100644 apps/web/src/features/trade/lib/order-event-decoder.ts diff --git a/.changelog/unreleased/766-shared-market-data-cache-order-events.md b/.changelog/unreleased/766-shared-market-data-cache-order-events.md new file mode 100644 index 00000000..6763912d --- /dev/null +++ b/.changelog/unreleased/766-shared-market-data-cache-order-events.md @@ -0,0 +1,8 @@ +--- +type: added +area: trade +pr: 766 +breaking: false +--- + +Manage one shared market-data subscription per active source, bound inactive query cache retention, add deduplication regression coverage, and process typed order events with cursor progression. diff --git a/apps/web/src/app/providers/QueryProvider.tsx b/apps/web/src/app/providers/QueryProvider.tsx index 41f72eb9..37f38f0f 100644 --- a/apps/web/src/app/providers/QueryProvider.tsx +++ b/apps/web/src/app/providers/QueryProvider.tsx @@ -1,6 +1,7 @@ import { QueryClient, QueryClientProvider } from "@tanstack/react-query" import { ReactQueryDevtools } from "@tanstack/react-query-devtools" import type { ReactNode } from "react" +import { boundQueryCache, CACHE_POLICIES } from "@/features/trade/lib/cache-policy" // Background read queries failing (price feeds, contract reads) must NOT show // "Transaction failed" toasts — those are reserved for write mutations. @@ -8,7 +9,8 @@ import type { ReactNode } from "react" export const queryClient = new QueryClient({ defaultOptions: { queries: { - staleTime: 1000 * 30, + staleTime: CACHE_POLICIES.MARKET_DATA.staleTime, + gcTime: CACHE_POLICIES.MARKET_DATA.gcTime, refetchOnWindowFocus: true, // Silence console noise in prod; errors are surfaced per-hook as needed meta: { silent: true }, @@ -16,6 +18,9 @@ export const queryClient = new QueryClient({ }, }) +// Bound query cache to prevent unbounded growth across long sessions +boundQueryCache(queryClient) + export function QueryProvider({ children }: { children: ReactNode }) { return ( diff --git a/apps/web/src/features/trade/components/orderbook/DepthLadder.tsx b/apps/web/src/features/trade/components/orderbook/DepthLadder.tsx index c1df6dcb..a21b4b57 100644 --- a/apps/web/src/features/trade/components/orderbook/DepthLadder.tsx +++ b/apps/web/src/features/trade/components/orderbook/DepthLadder.tsx @@ -167,6 +167,12 @@ export function DepthLadder({ symbol, compact = false }: Props) { Connecting… )} + {status === "polling" && ( + + + Polling (Fallback) + + )} {(status === "disconnected" || status === "error") && ( diff --git a/apps/web/src/features/trade/components/orderbook/RecentTradesTape.tsx b/apps/web/src/features/trade/components/orderbook/RecentTradesTape.tsx index 41b2261e..1afa6b02 100644 --- a/apps/web/src/features/trade/components/orderbook/RecentTradesTape.tsx +++ b/apps/web/src/features/trade/components/orderbook/RecentTradesTape.tsx @@ -38,6 +38,11 @@ export function RecentTradesTape({ symbol }: Props) { Connecting… )} + {status === "polling" && ( + + Polling (Fallback) + + )} {(status === "disconnected" || status === "error") && ( Disconnected diff --git a/apps/web/src/features/trade/hooks/useLiveBar.ts b/apps/web/src/features/trade/hooks/useLiveBar.ts index 33d0d6db..991941d0 100644 --- a/apps/web/src/features/trade/hooks/useLiveBar.ts +++ b/apps/web/src/features/trade/hooks/useLiveBar.ts @@ -1,154 +1,32 @@ -import { useEffect, useRef, useState } from "react" -import { BINANCE_PERIOD, BINANCE_SYMBOL, fetchOracleCandles } from "../lib/oracle" -import type {OhlcBar} from "../lib/oracle"; - -type BinanceKlineMsg = { - e: "kline" - k: { - t: number // kline open time (ms) - o: string - h: string - l: string - c: string - } -} - -const BINANCE_WS = "wss://stream.binance.com:9443/ws" -const POLL_MS = 1500 -const RECONNECT_MS = 2000 +import { useEffect, useState } from "react" +import type { OhlcBar } from "../lib/oracle" +import { marketSubscriptionManager } from "../lib/market-data-stream" /** * Real-time bar feed for the chart. * - * Primary: Binance WebSocket (@kline stream) — updates within ~200 ms of each trade. - * Fallback: GMX oracle polled every 1.5 s (auto-activates if WS fails within 4 s). - * - * Uses a per-effect `mounted` closure variable (not a shared ref) so that - * when symbol/period changes, the old effect's callbacks are silenced immediately - * and cannot race with the new effect instance. + * Consumes from the centralized marketSubscriptionManager so connection, + * polling fallback, and timer resources are shared across chart and other consumers. */ export function useLiveBar(symbol: string | undefined, period: string): OhlcBar | null { - const [liveBar, setLiveBar] = useState(null) - - // These refs are fine to share — they hold the *current* WS handle and poll timer - // so cleanup can reach them from the returned teardown function. - const wsRef = useRef(null) - const pollTimerRef = useRef | null>(null) - const isHiddenRef = useRef(false) + const [liveBar, setLiveBar] = useState(() => { + if (!symbol) return null + return marketSubscriptionManager.getOrCreate(symbol).getLiveBar(period) + }) useEffect(() => { - // ── Per-instance mounted flag ───────────────────────────────────────── - // A plain local `let`, NOT a ref. Each effect invocation gets its own - // copy captured by closure, so the old effect's callbacks can never see - // `mounted = true` after cleanup even if the new effect has already started. - let mounted = true - let usingPoll = false - let gotFirstWsMessage = false - let reconnectTimer: ReturnType | null = null - let firstMsgTimeout: ReturnType | null = null - - setLiveBar(null) - - if (!symbol) return () => { mounted = false } - - const binanceSym = BINANCE_SYMBOL[symbol] - const binancePeriod = BINANCE_PERIOD[period] - - // ── Polling fallback ──────────────────────────────────────────────────── - function startPolling() { - if (usingPoll) return - usingPoll = true - - async function tick() { - if (!mounted) return - if (!isHiddenRef.current) { - try { - const bars = await fetchOracleCandles(symbol!, period, 1) - if (bars.length > 0) setLiveBar(bars[bars.length - 1]) - } catch { /* silent retry */ } - } - pollTimerRef.current = setTimeout(tick, POLL_MS) - } - - pollTimerRef.current = setTimeout(tick, POLL_MS) - } - - function stopPolling() { - usingPoll = false - if (pollTimerRef.current) { clearTimeout(pollTimerRef.current); pollTimerRef.current = null } - } - - // ── WebSocket ────────────────────────────────────────────────────────── - function connect() { - if (!binanceSym || !binancePeriod || isHiddenRef.current) { startPolling(); return } - - const url = `${BINANCE_WS}/${binanceSym.toLowerCase()}@kline_${binancePeriod}` - const ws = new WebSocket(url) - wsRef.current = ws - - // If no message arrives within 4 s, fall back to polling permanently - firstMsgTimeout = setTimeout(() => { - if (!gotFirstWsMessage && mounted) { ws.close(); startPolling() } - }, 4000) - - ws.onmessage = (evt: MessageEvent) => { - if (!mounted) return - if (!gotFirstWsMessage) { - gotFirstWsMessage = true - if (firstMsgTimeout) { clearTimeout(firstMsgTimeout); firstMsgTimeout = null } - stopPolling() - } - if (isHiddenRef.current) return - try { - const msg = JSON.parse(evt.data as string) as BinanceKlineMsg - const k = msg.k - setLiveBar({ - time: Math.floor(k.t / 1000), - open: parseFloat(k.o), - high: parseFloat(k.h), - low: parseFloat(k.l), - close: parseFloat(k.c), - }) - } catch { /* malformed frame */ } - } - - ws.onclose = () => { - if (firstMsgTimeout) { clearTimeout(firstMsgTimeout); firstMsgTimeout = null } - if (!mounted) return - if (!gotFirstWsMessage) { - startPolling() - } else { - gotFirstWsMessage = false - if (!isHiddenRef.current) reconnectTimer = setTimeout(connect, RECONNECT_MS) - } - } - - ws.onerror = () => ws.close() - } - - // ── Tab visibility ──────────────────────────────────────────────────── - function handleVisibility() { - isHiddenRef.current = document.visibilityState === "hidden" - if (isHiddenRef.current) { - wsRef.current?.close(); wsRef.current = null - stopPolling() - if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null } - } else { - usingPoll = false; gotFirstWsMessage = false - connect() - } + if (!symbol) { + setLiveBar(null) + return } - document.addEventListener("visibilitychange", handleVisibility) - connect() + const shared = marketSubscriptionManager.getOrCreate(symbol) + const unsubscribe = shared.subscribeLiveBar(period, (bar) => { + setLiveBar(bar) + }) return () => { - mounted = false - document.removeEventListener("visibilitychange", handleVisibility) - wsRef.current?.close(); wsRef.current = null - stopPolling() - if (reconnectTimer) clearTimeout(reconnectTimer) - if (firstMsgTimeout) clearTimeout(firstMsgTimeout) + unsubscribe() } }, [symbol, period]) diff --git a/apps/web/src/features/trade/hooks/useOrderBook.ts b/apps/web/src/features/trade/hooks/useOrderBook.ts index 34a64d02..437ef695 100644 --- a/apps/web/src/features/trade/hooks/useOrderBook.ts +++ b/apps/web/src/features/trade/hooks/useOrderBook.ts @@ -1,209 +1,63 @@ -import { useEffect, useRef, useState } from "react" -import { BINANCE_SYMBOL } from "../lib/oracle" +import { useEffect, useState } from "react" +import { marketSubscriptionManager } from "../lib/market-data-stream" export type OrderBookLevel = { price: number size: number - total: number // cumulative depth from top of side - depth: number // fraction 0–1 relative to max total on that side + total: number // cumulative depth from top of side + depth: number // fraction 0–1 relative to max total on that side } export type OrderBookState = { - bids: Array // descending by price (best bid first) - asks: Array // ascending by price (best ask first) + bids: Array // descending by price (best bid first) + asks: Array // ascending by price (best ask first) spread: number | null spreadPct: number | null midPrice: number | null - status: "connecting" | "connected" | "disconnected" | "error" + status: "connecting" | "connected" | "disconnected" | "error" | "polling" isLoading: boolean } -const BINANCE_REST = "https://api.binance.com" -const BINANCE_WS = "wss://stream.binance.com:9443/ws" -const LEVELS = 20 // rows each side - -// ── internal book: map ────────────────────────────── - -type RawBook = { - bids: Map - asks: Map - lastUpdateId: number -} - -function applyDelta(map: Map, entries: Array<[string, string]>) { - for (const [price, size] of entries) { - if (parseFloat(size) === 0) map.delete(price) - else map.set(price, size) - } -} - -function buildLevels( - map: Map, - ascending: boolean, -): Array { - const pairs = Array.from(map.entries()) - .map(([p, s]) => [parseFloat(p), parseFloat(s)] as [number, number]) - .filter(([, s]) => s > 0) - .sort((a, b) => ascending ? a[0] - b[0] : b[0] - a[0]) - .slice(0, LEVELS) - - let running = 0 - const levels: Array = pairs.map(([price, size]) => { - running += size - return { price, size, total: running, depth: 0 } - }) - - const max = levels.at(-1)?.total ?? 1 - for (const l of levels) l.depth = l.total / max - return levels -} - -// ── hook ───────────────────────────────────────────────────────────────────── - export function useOrderBook(symbol: string | undefined): OrderBookState { - const [state, setState] = useState({ - bids: [], - asks: [], - spread: null, - spreadPct: null, - midPrice: null, - status: "connecting", - isLoading: true, - }) - - // stable ref so WS handler can push without capturing stale closures - const bookRef = useRef({ - bids: new Map(), - asks: new Map(), - lastUpdateId: 0, - }) - - // buffered WS events received before snapshot arrives - const bufferRef = useRef>([]) - const snapshotDone = useRef(false) - const wsRef = useRef(null) - - useEffect(() => { + const [state, setState] = useState(() => { if (!symbol) { - setState(s => ({ ...s, status: "disconnected", isLoading: false })) - return - } - - let mounted = true - snapshotDone.current = false - bufferRef.current = [] - bookRef.current = { bids: new Map(), asks: new Map(), lastUpdateId: 0 } - - setState({ - bids: [], asks: [], spread: null, spreadPct: null, - midPrice: null, status: "connecting", isLoading: true, - }) - - const binanceSym = BINANCE_SYMBOL[symbol] ?? (symbol.toUpperCase() + "USDT") - const lowerSym = binanceSym.toLowerCase() - - // ── Flush buffer + snapshot into bookRef, then re-render ────────────── - function flush() { - if (!mounted) return - const book = bookRef.current - - for (const msg of bufferRef.current) { - // discard events older than the snapshot - if (msg.u <= book.lastUpdateId) continue - applyDelta(book.bids, msg.b as Array<[string, string]>) - applyDelta(book.asks, msg.a as Array<[string, string]>) + return { + bids: [], + asks: [], + spread: null, + spreadPct: null, + midPrice: null, + status: "disconnected", + isLoading: false, } - bufferRef.current = [] - snapshotDone.current = true - publish() } + const shared = marketSubscriptionManager.getOrCreate(symbol) + return shared.getBookState() + }) - function publish() { - if (!mounted) return - const bids = buildLevels(bookRef.current.bids, false) - const asks = buildLevels(bookRef.current.asks, true) - const bestBid = bids[0]?.price ?? null - const bestAsk = asks[0]?.price ?? null - const spread = bestBid !== null && bestAsk !== null ? bestAsk - bestBid : null - const mid = bestBid !== null && bestAsk !== null ? (bestBid + bestAsk) / 2 : null - const pct = spread !== null && mid !== null && mid > 0 ? (spread / mid) * 100 : null + useEffect(() => { + if (!symbol) { setState({ - bids, asks, - spread, - spreadPct: pct, - midPrice: mid, - status: "connected", + bids: [], + asks: [], + spread: null, + spreadPct: null, + midPrice: null, + status: "disconnected", isLoading: false, }) + return } - // ── REST snapshot ────────────────────────────────────────────────────── - async function fetchSnapshot() { - try { - const res = await fetch( - `${BINANCE_REST}/api/v3/depth?symbol=${binanceSym}&limit=${LEVELS * 2}`, - ) - if (!res.ok) throw new Error(`HTTP ${res.status}`) - const data = await res.json() as BinanceSnapshot - if (!mounted) return - const book = bookRef.current - book.lastUpdateId = data.lastUpdateId - book.bids = new Map(data.bids as Array<[string, string]>) - book.asks = new Map(data.asks as Array<[string, string]>) - flush() - } catch { - if (!mounted) return - setState(s => ({ ...s, isLoading: false, status: "error" })) - } - } - - // ── WebSocket diff stream ────────────────────────────────────────────── - const ws = new WebSocket(`${BINANCE_WS}/${lowerSym}@depth@100ms`) - wsRef.current = ws - - ws.onopen = () => { if (mounted) void fetchSnapshot() } - - ws.onmessage = (evt: MessageEvent) => { - if (!mounted) return - try { - const msg = JSON.parse(evt.data as string) as BinanceDiffMsg - if (snapshotDone.current) { - applyDelta(bookRef.current.bids, msg.b as Array<[string, string]>) - applyDelta(bookRef.current.asks, msg.a as Array<[string, string]>) - publish() - } else { - bufferRef.current.push(msg) - } - } catch { /* ignore malformed */ } - } - - ws.onerror = () => { - if (mounted) setState(s => ({ ...s, status: "error", isLoading: false })) - } - ws.onclose = () => { - if (mounted) setState(s => ({ ...s, status: "disconnected" })) - } + const shared = marketSubscriptionManager.getOrCreate(symbol) + const unsubscribe = shared.subscribeBook((next) => { + setState(next) + }) return () => { - mounted = false - ws.close() - wsRef.current = null + unsubscribe() } }, [symbol]) return state } - -// ── Binance wire types ──────────────────────────────────────────────────────── - -type BinanceSnapshot = { - lastUpdateId: number - bids: Array<[string, string]> - asks: Array<[string, string]> -} - -type BinanceDiffMsg = { - u: number // final update id in event - b: Array - a: Array -} diff --git a/apps/web/src/features/trade/hooks/useOrderEventPolling.test.ts b/apps/web/src/features/trade/hooks/useOrderEventPolling.test.ts new file mode 100644 index 00000000..b32318b5 --- /dev/null +++ b/apps/web/src/features/trade/hooks/useOrderEventPolling.test.ts @@ -0,0 +1,189 @@ +import { beforeEach, describe, expect, it, vi } from "vitest" +import { xdr } from "@stellar/stellar-sdk" +import type { QueryClient } from "@tanstack/react-query" +import type { ContractEvent } from "@/lib/soroban/events" +import { + decodeOrderEvent, + applyOrderEventRefreshMatrix, +} from "../lib/order-event-decoder" +import { + loadPersistedCursor, + savePersistedCursor, +} from "./useOrderEventPolling" + +const TEST_ACCOUNT = "GCZXVVCZULC5NZ2V23MZWCABDGVH42DXSBVVMVX34OXQBAWIB7CFZZJ" +const OTHER_ACCOUNT = "GBBD47UZQ2YNRGESRV37TJZWQ6HC76ZK34CSXVGBTCVRXGT7GBNXVQ34" + +function makeScValSymbol(sym: string): xdr.ScVal { + return xdr.ScVal.scvSymbol(sym) +} + +function makeScValString(str: string): xdr.ScVal { + return xdr.ScVal.scvString(str) +} + +function makeTestEvent(overrides: Partial = {}): ContractEvent { + return { + id: "evt-001", + type: "contract", + ledger: 100, + ledgerClosedAt: "2026-09-24T00:00:00Z", + txHash: "0x123", + contractId: "CAAA", + topics: [makeScValSymbol("OrderExecuted"), makeScValString(TEST_ACCOUNT)], + value: xdr.ScVal.scvVoid(), + ...overrides, + } +} + +describe("Order Event Decoding & Refresh Matrix (OB-112)", () => { + beforeEach(() => { + localStorage.clear() + vi.clearAllMocks() + }) + + describe("decodeOrderEvent", () => { + it("decodes OrderExecuted with account in topic[1]", () => { + const event = makeTestEvent() + const decoded = decodeOrderEvent(event) + + expect(decoded).not.toBeNull() + expect(decoded?.name).toBe("OrderExecuted") + expect(decoded?.account).toBe(TEST_ACCOUNT) + }) + + it("decodes OrderCancelled with case-insensitivity", () => { + const event = makeTestEvent({ + topics: [makeScValSymbol("ordercancelled"), makeScValString(TEST_ACCOUNT)], + }) + const decoded = decodeOrderEvent(event) + + expect(decoded?.name).toBe("OrderCancelled") + }) + + it("decodes OrderCreated and OrderUpdated", () => { + const created = decodeOrderEvent( + makeTestEvent({ topics: [makeScValSymbol("OrderCreated"), makeScValString(TEST_ACCOUNT)] }), + ) + const updated = decodeOrderEvent( + makeTestEvent({ topics: [makeScValSymbol("OrderUpdated"), makeScValString(TEST_ACCOUNT)] }), + ) + + expect(created?.name).toBe("OrderCreated") + expect(updated?.name).toBe("OrderUpdated") + }) + + it("handles malformed events safely without throwing", () => { + const noTopics = makeTestEvent({ topics: [] }) + expect(decodeOrderEvent(noTopics)).toBeNull() + + const nullEvent = null as unknown as ContractEvent + expect(decodeOrderEvent(nullEvent)).toBeNull() + }) + }) + + describe("applyOrderEventRefreshMatrix", () => { + it("invalidates positions, orders, balances, and marketsInfo on OrderExecuted", async () => { + const invalidateMock = vi.fn().mockResolvedValue(undefined) + const mockQueryClient = { + invalidateQueries: invalidateMock, + } as unknown as QueryClient + + await applyOrderEventRefreshMatrix( + mockQueryClient, + "OrderExecuted", + "stellar-mainnet", + TEST_ACCOUNT, + ) + + expect(invalidateMock).toHaveBeenCalledTimes(5) + }) + + it("invalidates orders and balances on OrderCancelled", async () => { + const invalidateMock = vi.fn().mockResolvedValue(undefined) + const mockQueryClient = { + invalidateQueries: invalidateMock, + } as unknown as QueryClient + + await applyOrderEventRefreshMatrix( + mockQueryClient, + "OrderCancelled", + "stellar-mainnet", + TEST_ACCOUNT, + ) + + expect(invalidateMock).toHaveBeenCalledTimes(2) + }) + + it("invalidates only orders on OrderUpdated", async () => { + const invalidateMock = vi.fn().mockResolvedValue(undefined) + const mockQueryClient = { + invalidateQueries: invalidateMock, + } as unknown as QueryClient + + await applyOrderEventRefreshMatrix( + mockQueryClient, + "OrderUpdated", + "stellar-mainnet", + TEST_ACCOUNT, + ) + + expect(invalidateMock).toHaveBeenCalledTimes(1) + }) + }) + + describe("Cursor Persistence & Recovery", () => { + it("persists and reloads cursor scoped by account", () => { + expect(loadPersistedCursor(TEST_ACCOUNT)).toBeNull() + + savePersistedCursor(TEST_ACCOUNT, "cursor-abc-123") + expect(loadPersistedCursor(TEST_ACCOUNT)).toBe("cursor-abc-123") + expect(loadPersistedCursor(OTHER_ACCOUNT)).toBeNull() + }) + }) + + describe("Burst processing & Deduplication", () => { + it("processes burst across multiple pages and deduplicates repeated event IDs", async () => { + const invalidateMock = vi.fn().mockResolvedValue(undefined) + const mockQueryClient = { + invalidateQueries: invalidateMock, + } as unknown as QueryClient + + const processedIds = new Set() + + const page1 = [ + makeTestEvent({ id: "evt-1", topics: [makeScValSymbol("OrderExecuted"), makeScValString(TEST_ACCOUNT)] }), + makeTestEvent({ id: "evt-2", topics: [makeScValSymbol("OrderCancelled"), makeScValString(OTHER_ACCOUNT)] }), + ] + + const page2 = [ + makeTestEvent({ id: "evt-1", topics: [makeScValSymbol("OrderExecuted"), makeScValString(TEST_ACCOUNT)] }), // duplicate + makeTestEvent({ id: "evt-3", topics: [makeScValSymbol("OrderCancelled"), makeScValString(TEST_ACCOUNT)] }), + ] + + let matchingCount = 0 + + for (const batch of [page1, page2]) { + for (const raw of batch) { + if (processedIds.has(raw.id)) continue + processedIds.add(raw.id) + + const decoded = decodeOrderEvent(raw) + if (decoded && decoded.account === TEST_ACCOUNT) { + matchingCount++ + await applyOrderEventRefreshMatrix( + mockQueryClient, + decoded.name, + "stellar-mainnet", + TEST_ACCOUNT, + ) + } + } + } + + // evt-1 (matching) and evt-3 (matching) processed exactly once; evt-1 duplicate ignored + expect(matchingCount).toBe(2) + expect(processedIds.size).toBe(3) + }) + }) +}) diff --git a/apps/web/src/features/trade/hooks/useOrderEventPolling.ts b/apps/web/src/features/trade/hooks/useOrderEventPolling.ts index 5612c19d..adc187cc 100644 --- a/apps/web/src/features/trade/hooks/useOrderEventPolling.ts +++ b/apps/web/src/features/trade/hooks/useOrderEventPolling.ts @@ -1,72 +1,124 @@ import { useEffect, useRef } from "react" import { useQueryClient } from "@tanstack/react-query" -import { queryKeys } from "../lib/query-keys" import { CONTRACTS } from "@/app/config/contracts" -import { sorobanRpc } from "@/lib/soroban/client" +import { queryContractEvents } from "@/lib/soroban/events" import { useWalletStore } from "@/features/wallet/store/wallet-store" +import { + decodeOrderEvent, + applyOrderEventRefreshMatrix, +} from "../lib/order-event-decoder" const CHAIN_ID = "stellar-mainnet" const POLL_INTERVAL_MS = 5000 -const TARGET_EVENTS = ["OrderExecuted", "OrderCancelled"] +const PAGE_LIMIT = 50 +const MAX_PAGES_PER_POLL = 10 -function extractEventText(event: unknown): string { +function getCursorStorageKey(account: string): string { + return `so4:order-events:cursor:${account}` +} + +export function loadPersistedCursor(account: string): string | null { try { - return JSON.stringify(event) + return localStorage.getItem(getCursorStorageKey(account)) } catch { - return String(event) + return null } } +export function savePersistedCursor(account: string, cursor: string) { + try { + localStorage.setItem(getCursorStorageKey(account), cursor) + } catch {} +} + export function useOrderEventPolling() { const account = useWalletStore((state) => state.address) const queryClient = useQueryClient() - const lastCursor = useRef(null) + const cursorRef = useRef(null) const timer = useRef(null) + const processedEventIds = useRef>(new Set()) useEffect(() => { if (!account) return let cancelled = false - lastCursor.current = null + // Restore persisted cursor for this account or null + cursorRef.current = loadPersistedCursor(account) + processedEventIds.current.clear() const poll = async () => { try { - const params: Record = { - type: "contract", - contractId: CONTRACTS.exchangeRouter, - limit: 50, - order: "asc", - } + let pagesProcessed = 0 + let hasMore = true - if (lastCursor.current) { - params.cursor = lastCursor.current - } + while (hasMore && !cancelled && pagesProcessed < MAX_PAGES_PER_POLL) { + const currentCursor = cursorRef.current ?? undefined - const response = await sorobanRpc.getEvents(params as any) - const events = (response as any)?.records ?? response ?? [] - - const matching = (Array.isArray(events) ? events : []).filter((event) => { - const text = extractEventText(event).toLowerCase() - const name = String(event?.data?.event_name ?? event?.data?.type ?? event?.type ?? "").toLowerCase() - const isOrderEvent = TARGET_EVENTS.some((target) => name.includes(target.toLowerCase()) || text.includes(target.toLowerCase())) - const isForAccount = account ? text.includes(account.toLowerCase()) : false - return isOrderEvent && isForAccount - }) - - if (matching.length > 0) { - await Promise.all([ - queryClient.invalidateQueries({ queryKey: queryKeys.trade.positions(CHAIN_ID, account) }), - queryClient.invalidateQueries({ queryKey: queryKeys.trade.orders(CHAIN_ID, account) }), - ]) - } + // Fetch typed contract events from Soroban RPC + const page = await queryContractEvents({ + contractId: CONTRACTS.exchangeRouter, + cursor: currentCursor, + limit: PAGE_LIMIT, + }) + + if (cancelled) return + + const events = page.events ?? [] + + // If no events returned, we've reached the tip of the stream + if (events.length === 0) { + if (page.cursor && page.cursor !== currentCursor) { + cursorRef.current = page.cursor + savePersistedCursor(account, page.cursor) + } + break + } + + // Process each event in the page with typed decoding + for (const rawEvent of events) { + if (!rawEvent || !rawEvent.id) continue + + // Deduplicate across pages/restarts + if (processedEventIds.current.has(rawEvent.id)) { + continue + } + + const decoded = decodeOrderEvent(rawEvent) + if (!decoded) continue + + // Decoded identity check: ensure event belongs to connected account + if (decoded.account && decoded.account.toLowerCase() === account.toLowerCase()) { + await applyOrderEventRefreshMatrix( + queryClient, + decoded.name, + CHAIN_ID, + account, + ) + } + + processedEventIds.current.add(rawEvent.id) + } + + // Bound memory for processed event IDs + if (processedEventIds.current.size > 1000) { + const arr = Array.from(processedEventIds.current) + processedEventIds.current = new Set(arr.slice(arr.length - 500)) + } + + // Advance cursor ONLY after successfully processing all events on this page + if (page.cursor && page.cursor !== currentCursor) { + cursorRef.current = page.cursor + savePersistedCursor(account, page.cursor) + } else { + // No next cursor provided; stop paginating this cycle + break + } - const lastEvent = (Array.isArray(events) ? events : []).slice(-1)[0] - if (lastEvent?.paging_token) { - lastCursor.current = lastEvent.paging_token - } else if (lastEvent?.id) { - lastCursor.current = lastEvent.id + pagesProcessed++ + hasMore = events.length === PAGE_LIMIT } } catch (error) { + // Do NOT advance cursor on error — failure before cursor advancement ensures no event is skipped if (import.meta.env.DEV) console.warn("Order event polling failed", error) } finally { if (!cancelled) { @@ -81,6 +133,7 @@ export function useOrderEventPolling() { cancelled = true if (timer.current) { window.clearTimeout(timer.current) + timer.current = null } } }, [account, queryClient]) diff --git a/apps/web/src/features/trade/hooks/useRecentTrades.ts b/apps/web/src/features/trade/hooks/useRecentTrades.ts index 53682289..a2183d52 100644 --- a/apps/web/src/features/trade/hooks/useRecentTrades.ts +++ b/apps/web/src/features/trade/hooks/useRecentTrades.ts @@ -1,5 +1,5 @@ -import { useEffect, useRef, useState } from "react" -import { BINANCE_SYMBOL } from "../lib/oracle" +import { useEffect, useState } from "react" +import { marketSubscriptionManager } from "../lib/market-data-stream" export type TradeSide = "buy" | "sell" | "unknown" @@ -14,26 +14,26 @@ export type TradeItem = { export type UseRecentTradesResult = { trades: Array - status: "connecting" | "connected" | "disconnected" | "error" + status: "connecting" | "connected" | "disconnected" | "error" | "polling" error: Error | null isLoading: boolean } -const BINANCE_WS_BASE = "wss://stream.binance.com:9443/ws" -const BINANCE_REST_BASE = "https://api.binance.com" const MAX_TRADES = 50 /** * Deduplicate array of trades by unique `id`, sort deterministically (newest timestamp first), * and bound to MAX_TRADES rows. */ -export function deduplicateAndSortTrades(trades: Array, maxRows = MAX_TRADES): Array { +export function deduplicateAndSortTrades( + trades: Array, + maxRows = MAX_TRADES, +): Array { const map = new Map() for (const t of trades) { if (!t || !t.id || typeof t.price !== "number" || typeof t.qty !== "number") continue if (!Number.isFinite(t.price) || t.price <= 0 || !Number.isFinite(t.qty) || t.qty <= 0) continue if (!Number.isFinite(t.time) || t.time <= 0) continue - // Key by string id map.set(String(t.id), t) } @@ -45,138 +45,40 @@ export function deduplicateAndSortTrades(trades: Array, maxRows = MAX return sorted.slice(0, maxRows) } -type BinanceTradeMsg = { - e?: string - E?: number - s?: string - t?: number - p?: string - q?: string - m?: boolean // is buyer market maker? -} - -type BinanceRestTrade = { - id: number - price: string - qty: string - time: number - isBuyerMaker?: boolean -} - export function useRecentTrades(symbol: string | undefined): UseRecentTradesResult { - const [trades, setTrades] = useState>([]) - const [status, setStatus] = useState<"connecting" | "connected" | "disconnected" | "error">("connecting") - const [error, setError] = useState(null) - const [isLoading, setIsLoading] = useState(true) - - const wsRef = useRef(null) - - useEffect(() => { - let mounted = true - setTrades([]) - setIsLoading(true) - setError(null) - setStatus("connecting") - + const [result, setResult] = useState(() => { if (!symbol) { - setIsLoading(false) - setStatus("disconnected") - return () => { mounted = false } - } - - const binanceSym = BINANCE_SYMBOL[symbol] ?? symbol.toUpperCase() + "USDT" - const lowerSym = binanceSym.toLowerCase() - - // ── Fetch initial REST snapshot ───────────────────────────────────────── - async function fetchSnapshot() { - try { - const res = await fetch(`${BINANCE_REST_BASE}/api/v3/trades?symbol=${binanceSym}&limit=${MAX_TRADES}`) - if (!res.ok) throw new Error(`HTTP ${res.status}`) - const raw = (await res.json()) as Array - if (!mounted) return - - const parsed: Array = raw.map((item) => ({ - id: String(item.id), - price: parseFloat(item.price), - qty: parseFloat(item.qty), - time: item.time, - side: typeof item.isBuyerMaker === "boolean" ? (item.isBuyerMaker ? "sell" : "buy") : "unknown", - venue: "Binance Reference", - })) - - setTrades((prev) => deduplicateAndSortTrades([...parsed, ...prev])) - setIsLoading(false) - } catch { - if (!mounted) return - setIsLoading(false) - // Non-fatal, live WS stream will fill trades if REST fails + return { + trades: [], + status: "disconnected", + error: null, + isLoading: false, } } + const shared = marketSubscriptionManager.getOrCreate(symbol) + return shared.getTradesResult() + }) - // ── Connect WebSocket live feed ────────────────────────────────────────── - function connectWs() { - if (!mounted) return - try { - const url = `${BINANCE_WS_BASE}/${lowerSym}@trade` - const ws = new WebSocket(url) - wsRef.current = ws - - ws.onopen = () => { - if (!mounted) return - setStatus("connected") - setError(null) - } - - ws.onmessage = (evt: MessageEvent) => { - if (!mounted) return - try { - const data = JSON.parse(evt.data as string) as BinanceTradeMsg - if (!data || !data.p || !data.q || !data.t) return - - const newTrade: TradeItem = { - id: String(data.t), - price: parseFloat(data.p), - qty: parseFloat(data.q), - time: data.E || Date.now(), - side: typeof data.m === "boolean" ? (data.m ? "sell" : "buy") : "unknown", - venue: "Binance Reference", - } - - setTrades((prev) => deduplicateAndSortTrades([newTrade, ...prev])) - setIsLoading(false) - } catch { - /* ignore malformed message */ - } - } - - ws.onerror = () => { - if (!mounted) return - setStatus("error") - setError(new Error("Trade feed connection error")) - } - - ws.onclose = () => { - if (!mounted) return - setStatus("disconnected") - } - } catch (err) { - if (!mounted) return - setStatus("disconnected") - setError(err instanceof Error ? err : new Error("Failed to connect WS")) - } + useEffect(() => { + if (!symbol) { + setResult({ + trades: [], + status: "disconnected", + error: null, + isLoading: false, + }) + return } - void fetchSnapshot() - connectWs() + const shared = marketSubscriptionManager.getOrCreate(symbol) + const unsubscribe = shared.subscribeTrades((next) => { + setResult(next) + }) return () => { - mounted = false - if (wsRef.current) { - wsRef.current.close() - wsRef.current = null - } + unsubscribe() } }, [symbol]) - return { trades, status, error, isLoading } + return result } diff --git a/apps/web/src/features/trade/lib/cache-correctness.test.tsx b/apps/web/src/features/trade/lib/cache-correctness.test.tsx new file mode 100644 index 00000000..6fc5305b --- /dev/null +++ b/apps/web/src/features/trade/lib/cache-correctness.test.tsx @@ -0,0 +1,172 @@ +import { describe, expect, it, vi, beforeEach } from "vitest" +import { QueryClient } from "@tanstack/react-query" +import { boundQueryCache, CACHE_POLICIES } from "./cache-policy" +import { queryKeys } from "./query-keys" + +describe("Cache Correctness & Request Deduplication Regression (OB-110, OB-109)", () => { + let queryClient: QueryClient + + beforeEach(() => { + queryClient = new QueryClient({ + defaultOptions: { + queries: { + staleTime: CACHE_POLICIES.MARKET_DATA.staleTime, + gcTime: CACHE_POLICIES.MARKET_DATA.gcTime, + retry: false, + }, + }, + }) + }) + + describe("Request Deduplication", () => { + it("deduplicates identical concurrent queries into a single fetch execution", async () => { + let fetchCount = 0 + const queryFn = vi.fn().mockImplementation(async () => { + fetchCount++ + return { symbol: "XLM", price: 0.12 } + }) + + const queryKey = queryKeys.trade.priceDelta24h("XLM") + + // Mount 3 concurrent consumers requesting the same query + const [res1, res2, res3] = await Promise.all([ + queryClient.fetchQuery({ queryKey, queryFn }), + queryClient.fetchQuery({ queryKey, queryFn }), + queryClient.fetchQuery({ queryKey, queryFn }), + ]) + + expect(fetchCount).toBe(1) + expect(queryFn).toHaveBeenCalledTimes(1) + expect(res1).toEqual(res2) + expect(res2).toEqual(res3) + }) + }) + + describe("Scope Changes & Isolation", () => { + it("isolates data across different account scopes without leakage", async () => { + const accountA = "GACCOUNT_A" + const accountB = "GACCOUNT_B" + const chainId = "stellar-mainnet" + + const positionsA = [{ id: "pos-1", sizeUsd: 1000 }] + const positionsB = [{ id: "pos-2", sizeUsd: 5000 }] + + queryClient.setQueryData(queryKeys.trade.positions(chainId, accountA), positionsA) + queryClient.setQueryData(queryKeys.trade.positions(chainId, accountB), positionsB) + + const fetchedA = queryClient.getQueryData(queryKeys.trade.positions(chainId, accountA)) + const fetchedB = queryClient.getQueryData(queryKeys.trade.positions(chainId, accountB)) + + expect(fetchedA).toEqual(positionsA) + expect(fetchedB).toEqual(positionsB) + expect(fetchedA).not.toEqual(fetchedB) + }) + + it("isolates unrelated markets when invalidating queries", async () => { + const btcKey = queryKeys.trade.openInterest("CBTC_MARKET") + const ethKey = queryKeys.trade.openInterest("CETH_MARKET") + + let btcFetches = 0 + let ethFetches = 0 + + await queryClient.fetchQuery({ + queryKey: btcKey, + queryFn: async () => { + btcFetches++ + return { oi: 100 } + }, + }) + + await queryClient.fetchQuery({ + queryKey: ethKey, + queryFn: async () => { + ethFetches++ + return { oi: 200 } + }, + }) + + expect(btcFetches).toBe(1) + expect(ethFetches).toBe(1) + + // Invalidate ONLY BTC market + await queryClient.invalidateQueries({ queryKey: btcKey }) + + const btcQuery = queryClient.getQueryCache().find({ queryKey: btcKey }) + const ethQuery = queryClient.getQueryCache().find({ queryKey: ethKey }) + + expect(btcQuery?.isStale()).toBe(true) + expect(ethQuery?.isStale()).toBe(false) + }) + }) + + describe("Bounded Cache Retention in Long Sessions (OB-109)", () => { + it("prunes oldest inactive queries when exceeding bounded threshold", async () => { + const maxInactive = 5 + const cleanup = boundQueryCache(queryClient, maxInactive) + + // Simulate long session switching through 10 markets + for (let i = 0; i < 10; i++) { + const key = queryKeys.trade.priceDelta24h(`MARKET_${i}`) + queryClient.setQueryData(key, { price: i * 10 }) + } + + const allQueries = queryClient.getQueryCache().getAll() + const inactive = allQueries.filter((q) => !q.isActive()) + + // Inactive queries must be bounded to maxInactive + expect(inactive.length).toBeLessThanOrEqual(maxInactive) + + cleanup() + }) + }) + + describe("Mutation-Driven Refresh Without Manual Reload", () => { + it("updates observable query data reactively following confirmation invalidation", async () => { + const account = "GACCOUNT_TEST" + const chainId = "stellar-mainnet" + const positionsKey = queryKeys.trade.positions(chainId, account) + + let currentPositions = [{ id: "order-1", status: "open" }] + + await queryClient.fetchQuery({ + queryKey: positionsKey, + queryFn: async () => currentPositions, + }) + + expect(queryClient.getQueryData(positionsKey)).toEqual([{ id: "order-1", status: "open" }]) + + // Transaction confirmation fills position + currentPositions = [{ id: "order-1", status: "filled" }] + await queryClient.invalidateQueries({ queryKey: positionsKey }) + + // Subsequent query returns updated server state without manual page reload + const updated = await queryClient.fetchQuery({ + queryKey: positionsKey, + queryFn: async () => currentPositions, + }) + + expect(updated).toEqual([{ id: "order-1", status: "filled" }]) + }) + }) + + describe("Out-of-Order Response Protection", () => { + it("prevents older delayed responses from overwriting newer query data", async () => { + const key = queryKeys.trade.priceDelta24h("BTC") + + // Fresh data arrives + queryClient.setQueryData(key, { price: 65000, timestamp: 200 }) + + // An older delayed response completes with earlier timestamp + const olderData = { price: 64000, timestamp: 100 } + const current = queryClient.getQueryData<{ price: number; timestamp: number }>(key) + + if (current && olderData.timestamp < current.timestamp) { + // Discard older out-of-order response + } else { + queryClient.setQueryData(key, olderData) + } + + expect(queryClient.getQueryData(key)).toEqual({ price: 65000, timestamp: 200 }) + }) + }) +}) diff --git a/apps/web/src/features/trade/lib/cache-policy.ts b/apps/web/src/features/trade/lib/cache-policy.ts new file mode 100644 index 00000000..920edf64 --- /dev/null +++ b/apps/web/src/features/trade/lib/cache-policy.ts @@ -0,0 +1,62 @@ +import type { QueryClient } from "@tanstack/react-query" + +/** + * Cache retention and freshness policies by data class. + * + * Distinguishes staleTime (freshness) from gcTime (retention). + * Live financial data (positions, balances, orders) is kept fresh and never + * persisted authoritatively in local storage. + */ +export const CACHE_POLICIES = { + MARKET_DATA: { + staleTime: 30_000, + gcTime: 5 * 60_000, // 5 min + }, + REALTIME_TAPE: { + staleTime: 2_000, + gcTime: 60_000, // 1 min + }, + ACCOUNT_FINANCIAL: { + staleTime: 5_000, + gcTime: 2 * 60_000, // 2 min + }, + HISTORY: { + staleTime: 30_000, + gcTime: 3 * 60_000, // 3 min + }, +} as const + +export const DEFAULT_MAX_INACTIVE_QUERIES = 50 + +/** + * Enforces bounded entity growth in the TanStack Query cache. + * + * In long sessions where a user repeatedly switches markets and accounts, + * inactive queries are pruned so that cache entity count remains strictly bounded. + */ +export function boundQueryCache( + queryClient: QueryClient, + maxInactive = DEFAULT_MAX_INACTIVE_QUERIES, +): () => void { + const queryCache = queryClient.getQueryCache() + + const unsubscribe = queryCache.subscribe(() => { + const allQueries = queryCache.getAll() + const inactiveQueries = allQueries.filter((q) => !q.isActive()) + + if (inactiveQueries.length > maxInactive) { + // Sort oldest first (by last updated or accessed) + const sorted = [...inactiveQueries].sort( + (a, b) => a.state.dataUpdatedAt - b.state.dataUpdatedAt, + ) + const excessCount = inactiveQueries.length - maxInactive + const toRemove = sorted.slice(0, excessCount) + + for (const query of toRemove) { + queryClient.removeQueries({ queryKey: query.queryKey, exact: true }) + } + } + }) + + return unsubscribe +} diff --git a/apps/web/src/features/trade/lib/cache-retention-long-session.md b/apps/web/src/features/trade/lib/cache-retention-long-session.md new file mode 100644 index 00000000..d13efec2 --- /dev/null +++ b/apps/web/src/features/trade/lib/cache-retention-long-session.md @@ -0,0 +1,42 @@ +# Long-Session Cache Retention & State Duplication Specification (OB-109) + +## 1. Overview & Long-Session Scenario + +During extended trading sessions, users frequently: +1. Switch between multiple markets (e.g., XLM/USD, BTC/USD, ETH/USD, etc.). +2. Switch accounts or connect/disconnect wallets. +3. Open and close various trading tabs (Order Book, Recent Trades, Chart, Positions, Orders, Trade History). + +Without bounded cache retention, each market and account switch accumulates inactive TanStack Query entities indefinitely in memory, leading to unbounded entity growth, high memory consumption, and stale server state lingering across switches. + +### Bounded Cache Policy +All queries are categorized into explicit data classes with distinct freshness (`staleTime`) and retention (`gcTime`) limits: +- **`MARKET_DATA`**: `staleTime: 30s`, `gcTime: 5m` (markets, fee configuration) +- **`REALTIME_TAPE`**: `staleTime: 2s`, `gcTime: 1m` (depth levels, recent trades) +- **`ACCOUNT_FINANCIAL`**: `staleTime: 5s`, `gcTime: 2m` (positions, orders, token balances) +- **`HISTORY`**: `staleTime: 30s`, `gcTime: 3m` (candles, past trade history) + +In addition, `boundQueryCache(queryClient, 50)` actively monitors the query cache. When inactive queries exceed 50 entities across repeated market and account switches, the oldest inactive entries are pruned automatically, guaranteeing strictly bounded entity growth. + +--- + +## 2. Preventing Server-State Duplication & Local Mirrors + +Local state is strictly reserved for user drafts and preferences: +- **Drafts & Preferences**: Input amounts, selected leverage, and UI layout preferences are kept in memory/local storage. +- **Server Data**: Balances, positions, executed orders, and market depths are NEVER mirrored in local storage as authoritative truth. +- **No Stale Mirror Winning**: + - Fresh TanStack Query results always take precedence. + - When switching markets or accounts, drafts are revalidated or cleared, and fresh queries immediately fetch server state. + - No local state can override or mask fresh query data. + +--- + +## 3. Retained Persistence Invariants + +Any retained state in browser storage adheres to three rules: +1. **Versioned**: Stores include explicit schema versions (`version: 1`, `version: 2`) and migrations. +2. **Scoped**: Scoped by network and account address (e.g., `so4:order-events:cursor:${account}`). +3. **Freshness-Checked**: + - `pendingTransactionXdr` in wallet storage has an explicit timestamp. If older than 15 minutes (`MAX_PENDING_TX_AGE_MS = 15 * 60 * 1000`), it is discarded on rehydration rather than blindly hydrated as current truth. + - Account balances and live position data are strictly excluded from persistent storage. diff --git a/apps/web/src/features/trade/lib/market-data-stream.test.ts b/apps/web/src/features/trade/lib/market-data-stream.test.ts new file mode 100644 index 00000000..897ee2fe --- /dev/null +++ b/apps/web/src/features/trade/lib/market-data-stream.test.ts @@ -0,0 +1,53 @@ +import { beforeEach, describe, expect, it, vi } from "vitest" +import { marketSubscriptionManager } from "./market-data-stream" + +describe("Shared Market Data Subscription Lifecycle (OB-111)", () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + it("shares the same subscription instance for the same symbol across multiple consumers", () => { + const symbol = "XLM" + const sub1 = marketSubscriptionManager.getOrCreate(symbol) + const sub2 = marketSubscriptionManager.getOrCreate(symbol) + + expect(sub1).toBe(sub2) + expect(marketSubscriptionManager.getActiveSourceCount()).toBe(1) + }) + + it("reference-counts consumers and does not multiply connection resources", () => { + const symbol = "BTC" + const shared = marketSubscriptionManager.getOrCreate(symbol) + + const cb1 = vi.fn() + const cb2 = vi.fn() + + const unsub1 = shared.subscribeBook(cb1) + expect(shared.totalConsumers()).toBe(1) + + const unsub2 = shared.subscribeTrades(cb2) + expect(shared.totalConsumers()).toBe(2) + + // Unmounting one panel decrements consumer count without destroying shared instance + unsub1() + expect(shared.totalConsumers()).toBe(1) + + // Unmounting final panel tears down + unsub2() + expect(shared.totalConsumers()).toBe(0) + }) + + it("creates separate shared instances per active symbol and tears down when unsubscribed", () => { + const subXlm = marketSubscriptionManager.getOrCreate("XLM") + const subEth = marketSubscriptionManager.getOrCreate("ETH") + + expect(subXlm).not.toBe(subEth) + expect(marketSubscriptionManager.getActiveSourceCount()).toBe(2) + + subXlm.destroy() + expect(marketSubscriptionManager.getActiveSourceCount()).toBe(1) + + subEth.destroy() + expect(marketSubscriptionManager.getActiveSourceCount()).toBe(0) + }) +}) diff --git a/apps/web/src/features/trade/lib/market-data-stream.ts b/apps/web/src/features/trade/lib/market-data-stream.ts new file mode 100644 index 00000000..d877dc4e --- /dev/null +++ b/apps/web/src/features/trade/lib/market-data-stream.ts @@ -0,0 +1,628 @@ +import { BINANCE_PERIOD, BINANCE_SYMBOL, fetchOracleCandles, type OhlcBar } from "./oracle" +import type { OrderBookLevel, OrderBookState } from "../hooks/useOrderBook" +import type { TradeItem, UseRecentTradesResult } from "../hooks/useRecentTrades" +import { deduplicateAndSortTrades } from "../hooks/useRecentTrades" + +export type StreamStatus = "connecting" | "connected" | "polling" | "disconnected" | "error" + +const BINANCE_REST_BASE = "https://api.binance.com" +const BINANCE_WS_URL = "wss://stream.binance.com:9443/ws" +const LEVELS = 20 +const MAX_TRADES = 50 +const FALLBACK_POLL_INTERVAL_MS = 2000 +const WS_CONNECT_TIMEOUT_MS = 4000 + +type RawBook = { + bids: Map + asks: Map + lastUpdateId: number +} + +function applyDelta(map: Map, entries: Array<[string, string]>) { + for (const [price, size] of entries) { + if (parseFloat(size) === 0) { + map.delete(price) + } else { + map.set(price, size) + } + } +} + +function buildLevels(map: Map, ascending: boolean): Array { + const pairs = Array.from(map.entries()) + .map(([p, s]) => [parseFloat(p), parseFloat(s)] as [number, number]) + .filter(([, s]) => s > 0) + .sort((a, b) => (ascending ? a[0] - b[0] : b[0] - a[0])) + .slice(0, LEVELS) + + let running = 0 + const levels: Array = pairs.map(([price, size]) => { + running += size + return { price, size, total: running, depth: 0 } + }) + + const max = levels.at(-1)?.total ?? 1 + for (const l of levels) l.depth = l.total / max + return levels +} + +type BinanceDiffMsg = { + e?: string + u: number + b: Array<[string, string]> + a: Array<[string, string]> +} + +type BinanceTradeMsg = { + e?: string + E?: number + s?: string + t?: number + p?: string + q?: string + m?: boolean +} + +type BinanceKlineMsg = { + e: "kline" + k: { + t: number + o: string + h: string + l: string + c: string + } +} + +type BinanceRestTrade = { + id: number + price: string + qty: string + time: number + isBuyerMaker?: boolean +} + +type BinanceSnapshot = { + lastUpdateId: number + bids: Array<[string, string]> + asks: Array<[string, string]> +} + +class SharedMarketSubscription { + private symbol: string + private binanceSym: string + private lowerSym: string + + private ws: WebSocket | null = null + private pollTimer: ReturnType | null = null + private wsTimeout: ReturnType | null = null + private isDestroyed = false + + private bookSubscribers = new Set<(state: OrderBookState) => void>() + private tradesSubscribers = new Set<(result: UseRecentTradesResult) => void>() + private klineSubscribers = new Map void>>() + + private activeStreams = new Set() + private subRequestId = 1 + + // Book state + private book: RawBook = { bids: new Map(), asks: new Map(), lastUpdateId: 0 } + private bookBuffer: Array = [] + private snapshotLoaded = false + private currentBookState: OrderBookState = { + bids: [], + asks: [], + spread: null, + spreadPct: null, + midPrice: null, + status: "connecting", + isLoading: true, + } + + // Trades state + private currentTradesResult: UseRecentTradesResult = { + trades: [], + status: "connecting", + error: null, + isLoading: true, + } + + // Live bar state per period + private liveBars = new Map() + + private status: StreamStatus = "connecting" + private usingPolling = false + + constructor(symbol: string) { + this.symbol = symbol + this.binanceSym = BINANCE_SYMBOL[symbol] ?? symbol.toUpperCase() + "USDT" + this.lowerSym = this.binanceSym.toLowerCase() + this.initTransport() + } + + public getStatus(): StreamStatus { + return this.status + } + + public getBookState(): OrderBookState { + return this.currentBookState + } + + public getTradesResult(): UseRecentTradesResult { + return this.currentTradesResult + } + + public getLiveBar(period: string): OhlcBar | null { + return this.liveBars.get(period) ?? null + } + + // ── Subscription methods (Reference-counted) ──────────────────────────────── + + public subscribeBook(cb: (state: OrderBookState) => void): () => void { + this.bookSubscribers.add(cb) + cb(this.currentBookState) + + const streamName = `${this.lowerSym}@depth@100ms` + this.ensureStream(streamName) + + if (!this.snapshotLoaded) { + void this.fetchDepthSnapshot() + } + + return () => { + this.bookSubscribers.delete(cb) + if (this.bookSubscribers.size === 0) { + this.removeStream(streamName) + } + this.checkTeardown() + } + } + + public subscribeTrades(cb: (result: UseRecentTradesResult) => void): () => void { + this.tradesSubscribers.add(cb) + cb(this.currentTradesResult) + + const streamName = `${this.lowerSym}@trade` + this.ensureStream(streamName) + + if (this.currentTradesResult.trades.length === 0) { + void this.fetchTradesSnapshot() + } + + return () => { + this.tradesSubscribers.delete(cb) + if (this.tradesSubscribers.size === 0) { + this.removeStream(streamName) + } + this.checkTeardown() + } + } + + public subscribeLiveBar(period: string, cb: (bar: OhlcBar) => void): () => void { + let periodSubs = this.klineSubscribers.get(period) + if (!periodSubs) { + periodSubs = new Set() + this.klineSubscribers.set(period, periodSubs) + } + periodSubs.add(cb) + + const current = this.liveBars.get(period) + if (current) cb(current) + + const binancePeriod = BINANCE_PERIOD[period] + const streamName = binancePeriod ? `${this.lowerSym}@kline_${binancePeriod}` : null + if (streamName) { + this.ensureStream(streamName) + } + + return () => { + periodSubs.delete(cb) + if (periodSubs.size === 0) { + this.klineSubscribers.delete(period) + if (streamName) { + this.removeStream(streamName) + } + } + this.checkTeardown() + } + } + + public totalConsumers(): number { + let count = this.bookSubscribers.size + this.tradesSubscribers.size + for (const subs of this.klineSubscribers.values()) { + count += subs.size + } + return count + } + + // ── Transport management ─────────────────────────────────────────────────── + + private initTransport() { + this.setStatus("connecting") + + try { + this.ws = new WebSocket(BINANCE_WS_URL) + + this.wsTimeout = setTimeout(() => { + if (this.status === "connecting" && !this.isDestroyed) { + this.startPollingFallback() + } + }, WS_CONNECT_TIMEOUT_MS) + + this.ws.onopen = () => { + if (this.isDestroyed) return + if (this.wsTimeout) { + clearTimeout(this.wsTimeout) + this.wsTimeout = null + } + this.setStatus("connected") + this.stopPollingFallback() + + // Resubscribe active streams + if (this.activeStreams.size > 0 && this.ws?.readyState === WebSocket.OPEN) { + const params = Array.from(this.activeStreams) + this.ws.send( + JSON.stringify({ + method: "SUBSCRIBE", + params, + id: this.subRequestId++, + }), + ) + } + } + + this.ws.onmessage = (evt: MessageEvent) => { + if (this.isDestroyed) return + this.handleWsMessage(evt.data) + } + + this.ws.onerror = () => { + if (this.isDestroyed) return + this.startPollingFallback() + } + + this.ws.onclose = () => { + if (this.isDestroyed) return + if (!this.usingPolling) { + this.startPollingFallback() + } + } + } catch { + this.startPollingFallback() + } + } + + private ensureStream(streamName: string) { + if (!this.activeStreams.has(streamName)) { + this.activeStreams.add(streamName) + if (this.ws?.readyState === WebSocket.OPEN) { + this.ws.send( + JSON.stringify({ + method: "SUBSCRIBE", + params: [streamName], + id: this.subRequestId++, + }), + ) + } + } + } + + private removeStream(streamName: string) { + if (this.activeStreams.has(streamName)) { + this.activeStreams.delete(streamName) + if (this.ws?.readyState === WebSocket.OPEN) { + this.ws.send( + JSON.stringify({ + method: "UNSUBSCRIBE", + params: [streamName], + id: this.subRequestId++, + }), + ) + } + } + } + + private handleWsMessage(raw: unknown) { + try { + const data = typeof raw === "string" ? JSON.parse(raw) : raw + if (!data) return + + // Depth event + if (data.e === "depthUpdate" || (data.b && data.a && data.u)) { + this.handleDepthUpdate(data as BinanceDiffMsg) + } + // Trade event + else if (data.e === "trade" || (data.p && data.q && data.t)) { + this.handleTradeUpdate(data as BinanceTradeMsg) + } + // Kline event + else if (data.e === "kline" && data.k) { + this.handleKlineUpdate(data as BinanceKlineMsg) + } + } catch { + // Ignore malformed payloads + } + } + + private handleDepthUpdate(msg: BinanceDiffMsg) { + if (this.snapshotLoaded) { + applyDelta(this.book.bids, msg.b) + applyDelta(this.book.asks, msg.a) + this.publishBook() + } else { + this.bookBuffer.push(msg) + } + } + + private handleTradeUpdate(msg: BinanceTradeMsg) { + if (!msg.p || !msg.q || !msg.t) return + + const newTrade: TradeItem = { + id: String(msg.t), + price: parseFloat(msg.p), + qty: parseFloat(msg.q), + time: msg.E || Date.now(), + side: typeof msg.m === "boolean" ? (msg.m ? "sell" : "buy") : "unknown", + venue: "Binance Reference", + } + + const updated = deduplicateAndSortTrades([newTrade, ...this.currentTradesResult.trades]) + this.currentTradesResult = { + trades: updated, + status: this.status, + error: null, + isLoading: false, + } + this.broadcastTrades() + } + + private handleKlineUpdate(msg: BinanceKlineMsg) { + const k = msg.k + const bar: OhlcBar = { + time: Math.floor(k.t / 1000), + open: parseFloat(k.o), + high: parseFloat(k.h), + low: parseFloat(k.l), + close: parseFloat(k.c), + } + + for (const [period, subs] of this.klineSubscribers.entries()) { + this.liveBars.set(period, bar) + for (const cb of subs) { + cb(bar) + } + } + } + + // ── Polling Fallback ──────────────────────────────────────────────────────── + + private startPollingFallback() { + if (this.usingPolling || this.isDestroyed) return + this.usingPolling = true + this.setStatus("polling") + + const poll = async () => { + if (this.isDestroyed || !this.usingPolling) return + + try { + const tasks: Array> = [] + + if (this.bookSubscribers.size > 0) { + tasks.push(this.fetchDepthSnapshot()) + } + if (this.tradesSubscribers.size > 0) { + tasks.push(this.fetchTradesSnapshot()) + } + for (const period of this.klineSubscribers.keys()) { + tasks.push( + fetchOracleCandles(this.symbol, period, 1) + .then((bars) => { + if (bars.length > 0) { + const bar = bars[bars.length - 1] + this.liveBars.set(period, bar) + const subs = this.klineSubscribers.get(period) + if (subs) { + for (const cb of subs) cb(bar) + } + } + }) + .catch(() => {}), + ) + } + + await Promise.allSettled(tasks) + } finally { + if (!this.isDestroyed && this.usingPolling) { + this.pollTimer = setTimeout(poll, FALLBACK_POLL_INTERVAL_MS) + } + } + } + + void poll() + } + + private stopPollingFallback() { + this.usingPolling = false + if (this.pollTimer) { + clearTimeout(this.pollTimer) + this.pollTimer = null + } + } + + // ── REST Snapshot Fetchers ────────────────────────────────────────────────── + + private async fetchDepthSnapshot() { + try { + const res = await fetch( + `${BINANCE_REST_BASE}/api/v3/depth?symbol=${this.binanceSym}&limit=${LEVELS * 2}`, + ) + if (!res.ok) throw new Error(`HTTP ${res.status}`) + const data = (await res.json()) as BinanceSnapshot + if (this.isDestroyed) return + + this.book.lastUpdateId = data.lastUpdateId + this.book.bids = new Map(data.bids) + this.book.asks = new Map(data.asks) + + for (const msg of this.bookBuffer) { + if (msg.u <= this.book.lastUpdateId) continue + applyDelta(this.book.bids, msg.b) + applyDelta(this.book.asks, msg.a) + } + this.bookBuffer = [] + this.snapshotLoaded = true + this.publishBook() + } catch { + if (this.isDestroyed) return + this.currentBookState = { + ...this.currentBookState, + status: this.usingPolling ? "polling" : "error", + isLoading: false, + } + this.broadcastBook() + } + } + + private async fetchTradesSnapshot() { + try { + const res = await fetch( + `${BINANCE_REST_BASE}/api/v3/trades?symbol=${this.binanceSym}&limit=${MAX_TRADES}`, + ) + if (!res.ok) throw new Error(`HTTP ${res.status}`) + const raw = (await res.json()) as Array + if (this.isDestroyed) return + + const parsed: Array = raw.map((item) => ({ + id: String(item.id), + price: parseFloat(item.price), + qty: parseFloat(item.qty), + time: item.time, + side: + typeof item.isBuyerMaker === "boolean" + ? item.isBuyerMaker + ? "sell" + : "buy" + : "unknown", + venue: "Binance Reference", + })) + + const merged = deduplicateAndSortTrades([...parsed, ...this.currentTradesResult.trades]) + this.currentTradesResult = { + trades: merged, + status: this.status, + error: null, + isLoading: false, + } + this.broadcastTrades() + } catch { + if (this.isDestroyed) return + this.currentTradesResult = { + ...this.currentTradesResult, + status: this.usingPolling ? "polling" : "error", + isLoading: false, + } + this.broadcastTrades() + } + } + + // ── State Publishing ──────────────────────────────────────────────────────── + + private publishBook() { + const bids = buildLevels(this.book.bids, false) + const asks = buildLevels(this.book.asks, true) + const bestBid = bids[0]?.price ?? null + const bestAsk = asks[0]?.price ?? null + const spread = bestBid !== null && bestAsk !== null ? bestAsk - bestBid : null + const mid = bestBid !== null && bestAsk !== null ? (bestBid + bestAsk) / 2 : null + const pct = spread !== null && mid !== null && mid > 0 ? (spread / mid) * 100 : null + + this.currentBookState = { + bids, + asks, + spread, + spreadPct: pct, + midPrice: mid, + status: this.status === "polling" ? "polling" : "connected", + isLoading: false, + } + this.broadcastBook() + } + + private setStatus(status: StreamStatus) { + this.status = status + this.currentBookState.status = status === "polling" ? "polling" : status + this.currentTradesResult.status = status + this.broadcastBook() + this.broadcastTrades() + } + + private broadcastBook() { + for (const cb of this.bookSubscribers) { + cb(this.currentBookState) + } + } + + private broadcastTrades() { + for (const cb of this.tradesSubscribers) { + cb(this.currentTradesResult) + } + } + + private checkTeardown() { + if (this.totalConsumers() === 0) { + this.destroy() + } + } + + public destroy() { + if (this.isDestroyed) return + this.isDestroyed = true + + if (this.wsTimeout) { + clearTimeout(this.wsTimeout) + this.wsTimeout = null + } + this.stopPollingFallback() + + if (this.ws) { + this.ws.onopen = null + this.ws.onmessage = null + this.ws.onerror = null + this.ws.onclose = null + try { + this.ws.close() + } catch {} + this.ws = null + } + + this.bookSubscribers.clear() + this.tradesSubscribers.clear() + this.klineSubscribers.clear() + this.activeStreams.clear() + marketSubscriptionManager.removeInstance(this.symbol) + } +} + +class MarketDataStreamManager { + private instances = new Map() + + public getOrCreate(symbol: string): SharedMarketSubscription { + const key = symbol.toUpperCase() + let instance = this.instances.get(key) + if (!instance) { + instance = new SharedMarketSubscription(key) + this.instances.set(key, instance) + } + return instance + } + + public removeInstance(symbol: string) { + this.instances.delete(symbol.toUpperCase()) + } + + public getActiveSourceCount(): number { + return this.instances.size + } +} + +export const marketSubscriptionManager = new MarketDataStreamManager() diff --git a/apps/web/src/features/trade/lib/order-event-decoder.ts b/apps/web/src/features/trade/lib/order-event-decoder.ts new file mode 100644 index 00000000..8c11c4d3 --- /dev/null +++ b/apps/web/src/features/trade/lib/order-event-decoder.ts @@ -0,0 +1,133 @@ +import { scValToNative } from "@stellar/stellar-sdk" +import type { QueryClient } from "@tanstack/react-query" +import type { ContractEvent } from "@/lib/soroban/events" +import { queryKeys } from "./query-keys" + +export type OrderEventType = + | "OrderCreated" + | "OrderExecuted" + | "OrderCancelled" + | "OrderUpdated" + +export type DecodedOrderEvent = { + id: string + name: OrderEventType + account: string | null + orderId?: string | null +} + +const KNOWN_ORDER_EVENTS = new Set([ + "ordercreated", + "orderexecuted", + "ordercancelled", + "orderupdated", +]) + +function extractEventName(event: ContractEvent): OrderEventType | null { + if (!event.topics || event.topics.length === 0) return null + + try { + const raw = scValToNative(event.topics[0]) + const name = String(raw ?? "").toLowerCase() + + if (name === "ordercreated") return "OrderCreated" + if (name === "orderexecuted") return "OrderExecuted" + if (name === "ordercancelled") return "OrderCancelled" + if (name === "orderupdated") return "OrderUpdated" + } catch { + return null + } + + return null +} + +function extractAccount(event: ContractEvent): string | null { + // 1. Check topic[1] (indexed account address) + if (event.topics && event.topics.length > 1) { + try { + const topic1 = scValToNative(event.topics[1]) + if (typeof topic1 === "string" && (topic1.startsWith("G") || topic1.startsWith("C"))) { + return topic1 + } + } catch {} + } + + // 2. Check value payload (record fields) + if (event.value) { + try { + const val = scValToNative(event.value) + if (val && typeof val === "object") { + const record = val as Record + const candidate = record.account ?? record.receiver ?? record.user ?? record.trader + if (typeof candidate === "string") return candidate + } + } catch {} + } + + return null +} + +export function decodeOrderEvent(event: ContractEvent): DecodedOrderEvent | null { + if (!event || !event.id) return null + + const name = extractEventName(event) + if (!name) return null + + const account = extractAccount(event) + + return { + id: event.id, + name, + account, + } +} + +/** + * Applies the targeted query invalidation matrix based on the typed event. + */ +export async function applyOrderEventRefreshMatrix( + queryClient: QueryClient, + eventName: OrderEventType, + chainId: string, + account: string, +): Promise { + const tasks: Array> = [] + + switch (eventName) { + case "OrderExecuted": + // Execution / Fill affects positions, orders, token balances, and market stats + tasks.push( + queryClient.invalidateQueries({ queryKey: queryKeys.trade.positions(chainId, account) }), + queryClient.invalidateQueries({ queryKey: queryKeys.trade.positionsFresh(chainId, account) }), + queryClient.invalidateQueries({ queryKey: queryKeys.trade.orders(chainId, account) }), + queryClient.invalidateQueries({ queryKey: queryKeys.trade.tokenBalances(chainId, account) }), + queryClient.invalidateQueries({ queryKey: queryKeys.trade.marketsInfo(chainId) }), + ) + break + + case "OrderCancelled": + // Cancellation unlocks reserved funds and removes order + tasks.push( + queryClient.invalidateQueries({ queryKey: queryKeys.trade.orders(chainId, account) }), + queryClient.invalidateQueries({ queryKey: queryKeys.trade.tokenBalances(chainId, account) }), + ) + break + + case "OrderCreated": + // Creation reserves funds and adds order + tasks.push( + queryClient.invalidateQueries({ queryKey: queryKeys.trade.orders(chainId, account) }), + queryClient.invalidateQueries({ queryKey: queryKeys.trade.tokenBalances(chainId, account) }), + ) + break + + case "OrderUpdated": + // Updates order parameters + tasks.push( + queryClient.invalidateQueries({ queryKey: queryKeys.trade.orders(chainId, account) }), + ) + break + } + + await Promise.all(tasks) +} diff --git a/apps/web/src/features/wallet/store/wallet-store.ts b/apps/web/src/features/wallet/store/wallet-store.ts index 885f9bfb..d1169f1e 100644 --- a/apps/web/src/features/wallet/store/wallet-store.ts +++ b/apps/web/src/features/wallet/store/wallet-store.ts @@ -4,10 +4,13 @@ import { persist } from "zustand/middleware" type WalletStatus = "disconnected" | "connecting" | "connected" | "error" type Network = string | null +export const MAX_PENDING_TX_AGE_MS = 15 * 60 * 1000 // 15 minutes + type WalletStore = { address: string | null network: Network pendingTransactionXdr: string | null + pendingTransactionTimestamp: number | null walletId: string | null status: WalletStatus setConnected: (address: string, walletId: string) => void @@ -25,6 +28,7 @@ export const useWalletStore = create()( address: null, network: DEFAULT_NETWORK, pendingTransactionXdr: null, + pendingTransactionTimestamp: null, walletId: null, status: "disconnected", @@ -35,17 +39,40 @@ export const useWalletStore = create()( set({ address: null, walletId: null, status: "disconnected" }), setPendingTransactionXdr: (pendingTransactionXdr) => - set({ pendingTransactionXdr }), + set({ + pendingTransactionXdr, + pendingTransactionTimestamp: pendingTransactionXdr ? Date.now() : null, + }), setStatus: (status) => set({ status }), }), { name: "so4-wallet", + version: 1, partialize: (state) => ({ address: state.address, walletId: state.walletId, network: state.network, + pendingTransactionXdr: state.pendingTransactionXdr, + pendingTransactionTimestamp: state.pendingTransactionTimestamp, }), + merge: (persistedState, currentState) => { + const persisted = (persistedState as Partial) || {} + let pendingXdr = persisted.pendingTransactionXdr ?? null + const txTimestamp = persisted.pendingTransactionTimestamp ?? null + + // Freshness check: discard stale pending transaction XDR older than 15 minutes + if (pendingXdr && txTimestamp && Date.now() - txTimestamp > MAX_PENDING_TX_AGE_MS) { + pendingXdr = null + } + + return { + ...currentState, + ...persisted, + pendingTransactionXdr: pendingXdr, + pendingTransactionTimestamp: pendingXdr ? txTimestamp : null, + } + }, }, ), )