From 9e36c91e132ee606ab15909bef1b5a8509951385 Mon Sep 17 00:00:00 2001
From: Azure <983547216@qq.com>
Date: Wed, 24 Jun 2026 09:14:51 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E9=87=8D=E6=96=B0=E5=AE=9E=E7=8E=B0Sse?=
 =?UTF-8?q?Client?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/common/hooks/useSse.ts           | 227 ++++++++-------------------
 src/common/sse/SseClient.ts          | 168 ++++++++++++++++++++
 src/components/baseLayouts/Lobby.vue |  26 ++-
 src/stores/sseBus.ts                 |  10 +-
 4 files changed, 264 insertions(+), 167 deletions(-)
 create mode 100644 src/common/sse/SseClient.ts

diff --git a/src/common/hooks/useSse.ts b/src/common/hooks/useSse.ts
index a8fd6e6..f4bb0cf 100644
--- a/src/common/hooks/useSse.ts
+++ b/src/common/hooks/useSse.ts
@@ -1,184 +1,83 @@
 import { ref } from 'vue'
-import { fetchEventSource } from '@microsoft/fetch-event-source'
-import { sseBus } from '@/stores/sseBus'
+import { SseClient } from '@/common/sse/SseClient'
 import type { SseEvents } from '@/stores/sseBus'
+import { sseBus } from '@/stores/sseBus'
 import { useAccountStore } from '@/stores/modules/account'
 
+/** SSE 连接 URL */
 const SSE_URL = '/api/sse/subscribe'
-const MAX_RETRY_COUNT = 5
-const BASE_RETRY_DELAY_MS = 3000
-const MAX_RETRY_DELAY_MS = 30000
 
-const isConnected = ref(false)
-let abortController: AbortController | null = null
-let currentToken: string | null = null
-let connecting = false
-let retryCount = 0
+/** 是否正在刷新 token（防并发） */
 let isRefreshing = false
 
-class SseHttpError extends Error {
-  constructor(
-    message: string,
-    readonly response: Response
-  ) {
-    super(message)
-    this.name = 'SseHttpError'
+// ==================== 模块级单例 ====================
+
+const isConnected = ref(false)
+
+const client = new SseClient(
+  SSE_URL,
+  () => useAccountStore().accessToken,
+  {
+    onConnected() {
+      isConnected.value = true
+      sseBus.emit('connected', undefined)
+    },
+    onMessage(event: string, data: string) {
+      try {
+        const payload = JSON.parse(data)
+        sseBus.emit(event as keyof SseEvents, payload)
+      } catch {
+        console.warn('[SSE] 事件数据解析失败:', event, data)
+      }
+    },
+    onError(message: string) {
+      sseBus.emit('error', { message, retryCount: 0 })
+    },
+    onUnauthorized() {
+      void refreshTokenAndReconnect()
+    }
+  }
+)
+
+// ==================== Token 刷新 ====================
+
+async function refreshTokenAndReconnect(): Promise<void> {
+  if (isRefreshing) return
+
+  const accountStore = useAccountStore()
+  if (!accountStore.refreshToken) {
+    sseBus.emit('error', { message: '登录已过期，请重新登录', retryCount: 0 })
+    return
+  }
+
+  isRefreshing = true
+  try {
+    const newToken = await accountStore.refreshAccessToken()
+    if (newToken) {
+      await client.connect()
+    }
+  } catch {
+    sseBus.emit('error', { message: '实时连接异常，请刷新页面', retryCount: 0 })
+  } finally {
+    isRefreshing = false
   }
 }
 
-function isSseHttpError(error: unknown): error is SseHttpError {
-  return error instanceof SseHttpError
-}
-
-function getRetryDelay(): number {
-  return Math.min(BASE_RETRY_DELAY_MS * Math.pow(2, retryCount - 1), MAX_RETRY_DELAY_MS)
-}
-
-function resetConnectionState(): void {
-  connecting = false
-  isConnected.value = false
-}
+// ==================== Composable ====================
 
 /**
- * Global SSE connection manager.
+ * 全局 SSE 连接管理
  *
- * Module-level state guarantees a single active connection across all callers.
+ * SseClient 只管连接生命周期，业务消息通过 sseBus 分发。
+ * 组件只需 `useSse().connect()` / `useSse().disconnect()`。
  */
 export function useSse() {
-  async function connect(): Promise<void> {
-    const accountStore = useAccountStore()
-    const token = accountStore.accessToken
-
-    if (!token) {
-      console.warn('[SSE] Missing accessToken, skip connection')
-      return
-    }
-
-    if (currentToken === token && (isConnected.value || connecting)) {
-      return
-    }
-
-    abort()
-    connecting = true
-    currentToken = token
-    abortController = new AbortController()
-
-    await fetchEventSource(SSE_URL, {
-      headers: { saToken: token },
-      signal: abortController.signal,
-      openWhenHidden: true,
-
-      async onopen(response) {
-        const contentType = response.headers.get('content-type') || ''
-
-        if (response.status === 401) {
-          throw new SseHttpError('SSE unauthorized', response)
-        }
-
-        if (!response.ok) {
-          throw new SseHttpError(`SSE connection failed: ${response.status}`, response)
-        }
-
-        if (!contentType.includes('text/event-stream')) {
-          throw new SseHttpError(`Invalid SSE content-type: ${contentType || 'empty'}`, response)
-        }
-
-        connecting = false
-        isConnected.value = true
-        retryCount = 0
-        sseBus.emit('connected', undefined)
-        console.log('[SSE] Connected')
-      },
-
-      onmessage(message) {
-        if (!message.event || !message.data) {
-          return
-        }
-
-        try {
-          const data = JSON.parse(message.data)
-          sseBus.emit(message.event as keyof SseEvents, data)
-        } catch {
-          console.warn('[SSE] Failed to parse event payload:', message.event, message.data)
-        }
-      },
-
-      onclose() {
-        resetConnectionState()
-        console.log('[SSE] Closed')
-      },
-
-      onerror(error) {
-        resetConnectionState()
-
-        if (isSseHttpError(error) && error.response.status === 401) {
-          void refreshTokenAndReconnect()
-          throw error
-        }
-
-        retryCount += 1
-        if (retryCount >= MAX_RETRY_COUNT) {
-          sseBus.emit('error', {
-            message: 'Realtime connection failed. Please refresh the page.',
-            retryCount
-          })
-          throw error
-        }
-
-        const delay = getRetryDelay()
-        console.warn(`[SSE] Connection error, retry in ${delay / 1000}s (${retryCount}/${MAX_RETRY_COUNT})`)
-        return delay
-      }
-    })
-  }
-
-  function abort(): void {
-    if (abortController) {
-      abortController.abort()
-      abortController = null
-    }
-
-    currentToken = null
-    retryCount = 0
-    resetConnectionState()
-  }
-
-  function disconnect(): void {
-    abort()
-    console.log('[SSE] Disconnected')
-  }
-
-  async function refreshTokenAndReconnect(): Promise<void> {
-    if (isRefreshing) {
-      return
-    }
-
-    const accountStore = useAccountStore()
-    if (!accountStore.refreshToken) {
-      sseBus.emit('error', { message: 'Login expired. Please sign in again.', retryCount: 0 })
-      return
-    }
-
-    isRefreshing = true
-    try {
-      const newToken = await accountStore.refreshAccessToken()
-      if (newToken) {
-        retryCount = 0
-        await connect()
-      }
-    } catch {
-      sseBus.emit('error', {
-        message: 'Realtime connection failed. Please refresh the page.',
-        retryCount: 0
-      })
-    } finally {
-      isRefreshing = false
-    }
-  }
-
   return {
     isConnected,
-    connect,
-    disconnect
+    connect: () => client.connect(),
+    disconnect: () => {
+      client.disconnect()
+      isConnected.value = false
+    }
   }
 }
diff --git a/src/common/sse/SseClient.ts b/src/common/sse/SseClient.ts
new file mode 100644
index 0000000..de0d811
--- /dev/null
+++ b/src/common/sse/SseClient.ts
@@ -0,0 +1,168 @@
+import { fetchEventSource } from '@microsoft/fetch-event-source'
+
+/**
+ * SSE 连接客户端
+ *
+ * 只负责连接生命周期（建连、断开、重连、心跳），不包含任何业务逻辑。
+ * 收到消息后通过 onMessage 回调抛出，由业务层自行分发。
+ */
+export class SseClient {
+  private abortController: AbortController | null = null
+  private retryCount = 0
+  private _connected = false
+  private _connecting = false
+
+  /** 重连配置 */
+  private readonly maxRetries: number
+  private readonly baseRetryDelayMs: number
+  private readonly maxRetryDelayMs: number
+
+  constructor(
+    private readonly url: string,
+    private readonly getToken: () => string | null,
+    private readonly callbacks: SseClientCallbacks,
+    opts?: SseClientOptions
+  ) {
+    this.maxRetries = opts?.maxRetries ?? 5
+    this.baseRetryDelayMs = opts?.baseRetryDelayMs ?? 3000
+    this.maxRetryDelayMs = opts?.maxRetryDelayMs ?? 30000
+  }
+
+  get connected(): boolean {
+    return this._connected
+  }
+
+  get connecting(): boolean {
+    return this._connecting
+  }
+
+  /**
+   * 建立 SSE 连接
+   *
+   * 幂等调用：已连接或正在连接则跳过。
+   * 连接断开时自动指数退避重连，最多 maxRetries 次。
+   */
+  async connect(): Promise<void> {
+    const token = this.getToken()
+    if (!token) {
+      this.callbacks.onError?.('缺少 accessToken')
+      return
+    }
+
+    // 幂等：相同 token 已连接或正在连接则跳过
+    if (this._connected || this._connecting) {
+      return
+    }
+
+    this.disconnect()
+
+    this._connecting = true
+    this.abortController = new AbortController()
+
+    await fetchEventSource(this.url, {
+      headers: { saToken: token },
+      signal: this.abortController.signal,
+      openWhenHidden: true,
+
+      onopen: async (response) => {
+        const contentType = response.headers.get('content-type') || ''
+
+        if (response.status === 401) {
+          throw new SseHttpError('SSE 鉴权失败', response)
+        }
+        if (!response.ok) {
+          throw new SseHttpError(`SSE 连接失败: ${response.status}`, response)
+        }
+        if (!contentType.includes('text/event-stream')) {
+          throw new SseHttpError(`非法的响应类型: ${contentType || '空'}`, response)
+        }
+
+        this._connecting = false
+        this._connected = true
+        this.retryCount = 0
+        this.callbacks.onConnected?.()
+      },
+
+      onmessage: (message) => {
+        if (!message.event || !message.data) {
+          return
+        }
+        this.callbacks.onMessage(message.event, message.data)
+      },
+
+      onclose: () => {
+        this._connecting = false
+        this._connected = false
+      },
+
+      onerror: (error) => {
+        this._connecting = false
+        this._connected = false
+
+        // 401 → 通知外部刷新 token
+        if (isSseHttpError(error) && error.response.status === 401) {
+          this.callbacks.onUnauthorized?.()
+          throw error
+        }
+
+        this.retryCount += 1
+        if (this.retryCount >= this.maxRetries) {
+          this.callbacks.onError?.('实时连接异常，请刷新页面')
+          throw error
+        }
+
+        const delay = Math.min(
+          this.baseRetryDelayMs * Math.pow(2, this.retryCount - 1),
+          this.maxRetryDelayMs
+        )
+        return delay
+      }
+    })
+  }
+
+  /** 断开连接（主动关闭，不重连） */
+  disconnect(): void {
+    if (this.abortController) {
+      this.abortController.abort()
+      this.abortController = null
+    }
+    this.retryCount = 0
+    this._connected = false
+    this._connecting = false
+  }
+}
+
+// ==================== 类型定义 ====================
+
+export interface SseClientCallbacks {
+  /** 连接成功 */
+  onConnected?: () => void
+  /** 收到消息 (eventName, rawData) */
+  onMessage: (event: string, data: string) => void
+  /** 连接异常（重连耗尽或 token 缺失） */
+  onError?: (message: string) => void
+  /** 收到 401，需刷新 token */
+  onUnauthorized?: () => void
+}
+
+export interface SseClientOptions {
+  maxRetries?: number
+  baseRetryDelayMs?: number
+  maxRetryDelayMs?: number
+}
+
+// ==================== 内部工具 ====================
+
+class SseHttpError extends Error {
+  constructor(
+    message: string,
+    readonly response: Response
+  ) {
+    super(message)
+    this.name = 'SseHttpError'
+  }
+}
+
+function isSseHttpError(error: unknown): error is SseHttpError {
+  return error instanceof SseHttpError
+}
diff --git a/src/components/baseLayouts/Lobby.vue b/src/components/baseLayouts/Lobby.vue
index 6b723a3..cf52b05 100644
--- a/src/components/baseLayouts/Lobby.vue
+++ b/src/components/baseLayouts/Lobby.vue
@@ -89,9 +89,17 @@ watch(activeGameId, () => {
   fetchRoomList()
 })
 
-/** 首次加载 */
+/**
+ * 首次加载：HTTP 全量拉取（SSE 可能尚未连接，HTTP 是最可靠的基线）
+ *
+ * 后续增量由 SSE 事件（room_created/room_updated/room_removed）驱动，
+ * SSE 重连时通过 connected 事件触发全量覆盖作为数据补偿。
+ */
 fetchRoomList()
 
+/** 是否已完成首次加载（用于区分 SSE 首次 connected 和重连） */
+let initialLoadDone = false
+
 // ===== SSE 实时事件订阅处理函数 =====
 /** 新房间创建 → 添加到列表头部 */
 function onRoomCreated(room: RoomItem) {
@@ -108,17 +116,33 @@ function onRoomUpdated(room: RoomItem) {
 function onRoomRemoved({ roomId }: { roomId: string }) {
   rooms.value = rooms.value.filter((r) => r.id !== roomId)
 }
+/**
+ * SSE 重连 → 全量拉取房间列表覆盖增量丢失
+ *
+ * 首次 connected 跳过（已由上面的 fetchRoomList() 处理），
+ * 仅重连时触发全量覆盖，避免初始重复请求。
+ * 重连期间的增量事件可能丢失，HTTP 全量拉取作为补偿。
+ */
+function onSseConnected() {
+  if (!initialLoadDone) {
+    initialLoadDone = true
+    return
+  }
+  fetchRoomList()
+}
 
 // ===== SSE 实时事件订阅 =====
 onMounted(() => {
   sseBus.on('room_created', onRoomCreated)
   sseBus.on('room_updated', onRoomUpdated)
   sseBus.on('room_removed', onRoomRemoved)
+  sseBus.on('connected', onSseConnected)
 })
 onUnmounted(() => {
   sseBus.off('room_created', onRoomCreated)
   sseBus.off('room_updated', onRoomUpdated)
   sseBus.off('room_removed', onRoomRemoved)
+  sseBus.off('connected', onSseConnected)
 })
 
 /** 筛选后的房间列表 */
diff --git a/src/stores/sseBus.ts b/src/stores/sseBus.ts
index 8185c28..c997662 100644
--- a/src/stores/sseBus.ts
+++ b/src/stores/sseBus.ts
@@ -13,7 +13,13 @@ export type SseEvents = {
 }
 
 /**
- * Global SSE event bus. The connection layer only dispatches events; feature
- * modules decide how to handle them.
+ * 全局 SSE 事件总线（发布-订阅模式）
+ *
+ * 连接层只负责解析 SSE 消息并 emit 到总线，业务模块自行订阅处理。
+ * 示例:
+ * ```ts
+ * import { sseBus } from '@/stores/sseBus'
+ * sseBus.on('room_updated', (room) => { ... })
+ * ```
  */
 export const sseBus = mitt<SseEvents>()