live-hub-py/web/frontend/src/hooks/useCollabRealtime.ts

import { useEffect, useRef, useState } from 'react';

export interface CollabRealtimeEvent {
  type: 'collab.ready' | 'collab.ping' | 'collab.invalidate';
  revision: number;
  scope?: 'orders' | 'shops' | 'gifts' | 'all';
  action?: string;
  actor_id?: number | null;
  at?: string;
}

interface UseCollabRealtimeOptions {
  enabled: boolean;
  onInvalidate: (event: CollabRealtimeEvent) => void;
  /** 默认是虎牙协作台账；斗鱼页面传入自己的 WS 路径。 */
  path?: string;
}

const DEFAULT_WS_PATH = '/api/huya/external-orders/ws';
const HEARTBEAT_TIMEOUT_MS = 60_000;
const MAX_RECONNECT_DELAY_MS = 30_000;

function websocketUrl(path: string): string {
  const protocol = window.location.protocol === 'https:' ? 'wss' : 'ws';
  return `${protocol}://${window.location.host}${path}`;
}

/**
 * 协作台账实时失效通知。
 *
 * 服务端只推「数据变了」，前端收到后按自己的筛选条件静默重拉；
 * 断线自动重连，页面恢复可见/浏览器恢复在线时立即重连。
 */
export function useCollabRealtime({
  enabled,
  onInvalidate,
  path = DEFAULT_WS_PATH,
}: UseCollabRealtimeOptions) {
  const [connected, setConnected] = useState(false);
  const [reconnecting, setReconnecting] = useState(false);
  const onInvalidateRef = useRef(onInvalidate);
  const lastRevisionRef = useRef(0);

  useEffect(() => {
    onInvalidateRef.current = onInvalidate;
  }, [onInvalidate]);

  useEffect(() => {
    if (!enabled) {
      setConnected(false);
      setReconnecting(false);
      return undefined;
    }

    let stopped = false;
    let socket: WebSocket | null = null;
    let reconnectTimer: number | undefined;
    let heartbeatTimer: number | undefined;
    let attempts = 0;

    const clearReconnectTimer = () => {
      if (reconnectTimer !== undefined) {
        window.clearTimeout(reconnectTimer);
        reconnectTimer = undefined;
      }
    };

    const clearHeartbeatTimer = () => {
      if (heartbeatTimer !== undefined) {
        window.clearTimeout(heartbeatTimer);
        heartbeatTimer = undefined;
      }
    };

    const armHeartbeat = () => {
      clearHeartbeatTimer();
      heartbeatTimer = window.setTimeout(() => {
        // 服务端每 25 秒至少发一次 ping；超时说明连接已假死，主动重连。
        socket?.close();
      }, HEARTBEAT_TIMEOUT_MS);
    };

    const scheduleReconnect = () => {
      if (stopped || reconnectTimer !== undefined) return;
      setReconnecting(true);
      const delay = Math.min(
        MAX_RECONNECT_DELAY_MS,
        1_000 * 2 ** Math.min(attempts, 5),
      );
      attempts += 1;
      reconnectTimer = window.setTimeout(() => {
        reconnectTimer = undefined;
        connect();
      }, delay + Math.floor(Math.random() * 400));
    };

    const handleMessage = (event: MessageEvent) => {
      armHeartbeat();
      let message: CollabRealtimeEvent;
      try {
        message = JSON.parse(String(event.data)) as CollabRealtimeEvent;
      } catch {
        return;
      }
      if (!message || typeof message.revision !== 'number') return;
      const previousRevision = lastRevisionRef.current;
      lastRevisionRef.current = Math.max(previousRevision, message.revision);

      if (message.type === 'collab.invalidate') {
        onInvalidateRef.current(message);
        return;
      }

      if (message.type === 'collab.ready' && previousRevision > 0) {
        // 重连成功：断线期间可能漏过事件，补一次静默刷新。
        onInvalidateRef.current({
          type: 'collab.invalidate',
          revision: message.revision,
          scope: 'all',
          action: 'reconnect',
        });
      }
    };

    const connect = () => {
      if (stopped) return;
      if (
        socket &&
        (socket.readyState === WebSocket.CONNECTING ||
          socket.readyState === WebSocket.OPEN)
      ) {
        return;
      }

      const nextSocket = new WebSocket(websocketUrl(path));
      socket = nextSocket;

      nextSocket.onopen = () => {
        attempts = 0;
        setConnected(true);
        setReconnecting(false);
        armHeartbeat();
      };
      nextSocket.onmessage = handleMessage;
      nextSocket.onerror = () => nextSocket.close();
      nextSocket.onclose = () => {
        clearHeartbeatTimer();
        const wasActive = socket === nextSocket;
        if (wasActive) socket = null;
        setConnected(false);
        if (!stopped && wasActive) scheduleReconnect();
      };
    };

    const reconnectNow = () => {
      if (stopped) return;
      clearReconnectTimer();
      if (socket && socket.readyState === WebSocket.OPEN) return;
      attempts = 0;
      socket?.close();
      socket = null;
      connect();
    };

    const handleOnline = () => reconnectNow();
    const handleVisibility = () => {
      if (document.visibilityState === 'visible') reconnectNow();
    };

    connect();
    window.addEventListener('online', handleOnline);
    document.addEventListener('visibilitychange', handleVisibility);

    return () => {
      stopped = true;
      clearReconnectTimer();
      clearHeartbeatTimer();
      window.removeEventListener('online', handleOnline);
      document.removeEventListener('visibilitychange', handleVisibility);
      socket?.close();
      socket = null;
      setConnected(false);
      setReconnecting(false);
    };
  }, [enabled, path]);

  return { connected, reconnecting };
}