goldenChart/frontend/src/utils/upbitWsBroker.ts

/**
 * 업비트 WebSocket 단일 연결 브로커 (/upbit-ws).
 * 훅·차트마다 WebSocket 을 열면 모바일에서 429·재연결 폭주가 발생한다.
 */
import type { UpbitMessage } from './upbitApi';

export type UpbitWsChannel = 'ticker' | 'orderbook' | 'trade';

type RawHandler = (msg: Record<string, unknown>) => void;
type ConnectionListener = (connected: boolean) => void;

interface Listener {
  channel: UpbitWsChannel;
  codes: Set<string>;
  handler: RawHandler;
}

const INITIAL_DELAY = 80;
const RECONNECT_BASE = 2000;
const RECONNECT_MAX = 30_000;
const PING_INTERVAL = 20_000;
const DISCONNECT_IDLE_MS = 5_000;

const refCounts = new Map<UpbitWsChannel, Map<string, number>>();
const listeners = new Set<Listener>();
const connectionListeners = new Set<ConnectionListener>();

let ws: WebSocket | null = null;
let connected = false;
let connectTimer: ReturnType<typeof setTimeout> | null = null;
let disconnectTimer: ReturnType<typeof setTimeout> | null = null;
let pingTimer: ReturnType<typeof setInterval> | null = null;
let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
let retryCount = 0;
let closingIntentionally = false;

function getWsUrl(): string {
  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<string | null> {
  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<string, unknown>;
  try {
    msg = JSON.parse(text) as Record<string, unknown>;
  } 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();

  try {
    const socket = new WebSocket(getWsUrl());
    ws = socket;
    socket.binaryType = 'arraybuffer';

    socket.onopen = () => {
      retryCount = 0;
      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 = () => {
      console.warn('[upbitWsBroker] socket error');
    };

    socket.onclose = () => {
      clearPing();
      notifyConnection(false);
      ws = null;
      if (!closingIntentionally && listenerCount() > 0) scheduleReconnect();
    };
  } catch (err) {
    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) 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);
  });
}