/** * 업비트 WebSocket 단일 연결 브로커 (/upbit-ws). * 훅·차트마다 WebSocket 을 열면 모바일에서 429·재연결 폭주가 발생한다. */ import type { UpbitMessage } from './upbitApi'; import { isDesktop } from './platform'; import { getSecureApiOrigin } from './backendApi'; export type UpbitWsChannel = 'ticker' | 'orderbook' | 'trade'; type RawHandler = (msg: Record) => void; type ConnectionListener = (connected: boolean) => void; interface Listener { channel: UpbitWsChannel; codes: Set; handler: RawHandler; } const INITIAL_DELAY = 350; const RECONNECT_BASE = 3000; const RECONNECT_MAX = 45_000; const PING_INTERVAL = 20_000; const DISCONNECT_IDLE_MS = 200; const refCounts = new Map>(); const listeners = new Set(); const connectionListeners = new Set(); let ws: WebSocket | null = null; let connected = false; let connectTimer: ReturnType | null = null; let disconnectTimer: ReturnType | null = null; let pingTimer: ReturnType | null = null; let reconnectTimer: ReturnType | null = null; let retryCount = 0; let closingIntentionally = false; let wsFailureLogged = false; function getWsUrl(): string { if (isDesktop()) { const origin = getSecureApiOrigin().replace(/^http:\/\//i, 'https://'); const host = origin.replace(/^https:\/\//i, ''); return `wss://${host}/upbit-ws`; } const proto = window.location.protocol === 'https:' ? 'wss' : 'ws'; return `${proto}://${window.location.host}/upbit-ws`; } function notifyConnection(next: boolean): void { connected = next; connectionListeners.forEach(l => l(next)); } function bumpRef(channel: UpbitWsChannel, code: string, delta: number): void { const key = code.toUpperCase(); let m = refCounts.get(channel); if (!m) { m = new Map(); refCounts.set(channel, m); } const next = (m.get(key) ?? 0) + delta; if (next <= 0) m.delete(key); else m.set(key, next); } function codesForChannel(channel: UpbitWsChannel): string[] { return Array.from(refCounts.get(channel)?.keys() ?? []); } function listenerCount(): number { return listeners.size; } function mayOpenSocket(): boolean { if (listenerCount() === 0) return false; if (typeof document !== 'undefined' && document.visibilityState === 'hidden') return false; return true; } function cancelPendingDisconnect(): void { if (disconnectTimer) { clearTimeout(disconnectTimer); disconnectTimer = null; } } function cancelPendingConnect(): void { if (connectTimer) { clearTimeout(connectTimer); connectTimer = null; } } function clearPing(): void { if (pingTimer) { clearInterval(pingTimer); pingTimer = null; } } function sendSubscribe(): void { if (!ws || ws.readyState !== WebSocket.OPEN) return; const packet: unknown[] = [{ ticket: `gc-${Date.now()}` }]; for (const ch of ['ticker', 'orderbook', 'trade'] as UpbitWsChannel[]) { const codes = codesForChannel(ch); if (codes.length > 0) packet.push({ type: ch, codes }); } if (packet.length === 1) return; packet.push({ format: 'DEFAULT' }); ws.send(JSON.stringify(packet)); } function parseBinaryOrText(data: string | ArrayBuffer | Blob): Promise { if (typeof data === 'string') return Promise.resolve(data); if (data instanceof ArrayBuffer) { return Promise.resolve(new TextDecoder('utf-8').decode(data)); } if (data instanceof Blob) return data.text(); return Promise.resolve(null); } function dispatch(text: string): void { if (text === 'PING') { ws?.send('PONG'); return; } let msg: Record; try { msg = JSON.parse(text) as Record; } catch { return; } const type = msg.type as string | undefined; if (type !== 'ticker' && type !== 'orderbook' && type !== 'trade') return; const code = String(msg.code ?? msg.market ?? '').toUpperCase(); if (!code) return; for (const entry of listeners) { if (entry.channel !== type) continue; if (entry.codes.size > 0 && !entry.codes.has(code)) continue; try { entry.handler(msg); } catch (err) { console.error('[upbitWsBroker] handler error', err); } } } function scheduleReconnect(): void { if (!mayOpenSocket()) return; if (reconnectTimer) return; const delay = Math.min(RECONNECT_BASE * 2 ** retryCount, RECONNECT_MAX); retryCount += 1; reconnectTimer = setTimeout(() => { reconnectTimer = null; openSocket(); }, delay); } function teardownSocket(): void { clearPing(); if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } closingIntentionally = true; if (ws) { ws.onopen = null; ws.onmessage = null; ws.onerror = null; ws.onclose = null; if (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING) { ws.close(1000, 'broker idle'); } ws = null; } closingIntentionally = false; if (connected) notifyConnection(false); } function openSocket(): void { if (listenerCount() === 0) return; teardownSocket(); if (!mayOpenSocket()) return; try { const socket = new WebSocket(getWsUrl()); ws = socket; socket.binaryType = 'arraybuffer'; socket.onopen = () => { retryCount = 0; wsFailureLogged = false; notifyConnection(true); sendSubscribe(); clearPing(); pingTimer = setInterval(() => { if (socket.readyState === WebSocket.OPEN) socket.send('PING'); }, PING_INTERVAL); }; socket.onmessage = (event: MessageEvent) => { void parseBinaryOrText(event.data as string | ArrayBuffer | Blob).then(text => { if (text != null) dispatch(text); }); }; socket.onerror = () => { if (!wsFailureLogged) { wsFailureLogged = true; console.warn( '[upbitWsBroker] WebSocket 연결 실패 — 시세는 REST 폴링으로 대체됩니다. ' + '서버 nginx에 /upbit-ws 프록시가 있는지 확인하세요.', ); } }; socket.onclose = () => { clearPing(); notifyConnection(false); ws = null; if (!closingIntentionally && listenerCount() > 0) scheduleReconnect(); }; } catch (err) { if (!wsFailureLogged) { wsFailureLogged = true; console.warn('[upbitWsBroker] connect failed', err); } scheduleReconnect(); } } function scheduleConnect(): void { cancelPendingDisconnect(); if (ws?.readyState === WebSocket.OPEN) { sendSubscribe(); return; } if (connectTimer) return; connectTimer = setTimeout(() => { connectTimer = null; openSocket(); }, INITIAL_DELAY); } function scheduleDisconnect(): void { cancelPendingConnect(); cancelPendingDisconnect(); disconnectTimer = setTimeout(() => { disconnectTimer = null; if (listenerCount() > 0) return; teardownSocket(); }, DISCONNECT_IDLE_MS); } /** 연결 상태 구독 */ export function subscribeUpbitWsConnection(listener: ConnectionListener): () => void { connectionListeners.add(listener); listener(connected && ws?.readyState === WebSocket.OPEN); return () => connectionListeners.delete(listener); } if (typeof document !== 'undefined') { document.addEventListener('visibilitychange', () => { if (document.visibilityState === 'hidden') { cancelPendingConnect(); if (listenerCount() === 0) teardownSocket(); return; } if (mayOpenSocket()) scheduleConnect(); }); } /** * 업비트 WS 채널 구독 (참조 카운트). codes 가 비어 있으면 구독하지 않음. */ export function subscribeUpbitWs( channel: UpbitWsChannel, codes: string[], handler: RawHandler, ): () => void { const normalized = [...new Set(codes.map(c => c.toUpperCase()).filter(Boolean))]; if (normalized.length === 0) { return () => {}; } normalized.forEach(code => bumpRef(channel, code, 1)); const entry: Listener = { channel, codes: new Set(normalized), handler }; listeners.add(entry); scheduleConnect(); return () => { listeners.delete(entry); normalized.forEach(code => bumpRef(channel, code, -1)); if (listenerCount() === 0) { cancelPendingConnect(); scheduleDisconnect(); } else if (ws?.readyState === WebSocket.OPEN) { sendSubscribe(); } }; } /** 차트용 trade 구독 — UpbitMessage 타입 핸들러 */ export function subscribeUpbitTrade( market: string, handler: (msg: UpbitMessage) => void, ): () => void { return subscribeUpbitWs('trade', [market], raw => { if (raw.type !== 'trade') return; handler(raw as unknown as UpbitMessage); }); }