diff --git a/src/common/hooks/useSse.ts b/src/common/hooks/useSse.ts index 3a203e3..a8fd6e6 100644 --- a/src/common/hooks/useSse.ts +++ b/src/common/hooks/useSse.ts @@ -1,186 +1,179 @@ import { ref } from 'vue' import { fetchEventSource } from '@microsoft/fetch-event-source' import { sseBus } from '@/stores/sseBus' +import type { SseEvents } from '@/stores/sseBus' import { useAccountStore } from '@/stores/modules/account' +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 +let isRefreshing = false + +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 +} + +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 +} + /** - * SSE 连接管理 composable(基于 @microsoft/fetch-event-source) + * Global SSE connection manager. * - * 设计原则: - * - 全局唯一:同一时间只有一条活跃的 SSE 连接 - * - token 驱动:仅 accessToken 变更时才断开重连 - * - 防重复:已连接(相同 token)时 connect() 直接跳过 - * - 所有业务变更通过 sseBus 事件总线推送,各组件自行订阅 - * - * 使用示例: - * ```ts - * const { isConnected, connect, disconnect } = useSse() - * watch(() => accountStore.accessToken, (v) => v ? connect() : disconnect()) - * ``` + * Module-level state guarantees a single active connection across all callers. */ export function useSse() { - /** SSE 连接是否已建立 */ - const isConnected = ref(false) - /** 用于取消 inflight 连接的控制器 */ - let abortController: AbortController | null = null - /** 当前连接使用的 token(用于判断是否需要重连) */ - let currentToken: string | null = null - /** 是否正在建立连接(防并发) */ - let connecting = false - /** 当前连接的重试计数 */ - let retryCount = 0 - /** 是否正在刷新 token(防并发) */ - let isRefreshing = false - - /** - * 建立 SSE 连接 - * - * 通过 fetch-event-source 发起,设置 saToken 请求头鉴权。 - * 连接期间持续接收服务端推送事件,直到 abort 或服务端关闭。 - * - * 防重复:如果已用相同 token 连接或正在连接中,直接跳过。 - */ async function connect(): Promise { const accountStore = useAccountStore() const token = accountStore.accessToken + if (!token) { - console.warn('[SSE] 无 accessToken,跳过连接') + console.warn('[SSE] Missing accessToken, skip connection') return } - // 相同 token 已连接或正在连接中,跳过 if (currentToken === token && (isConnected.value || connecting)) { return } - // 取消旧连接 abort() - connecting = true currentToken = token abortController = new AbortController() - // Vite 代理: /api/sse/subscribe → 后端 /sse/subscribe - fetchEventSource('/api/sse/subscribe', { + 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) + } - /** 连接建立 */ - async onopen() { connecting = false isConnected.value = true retryCount = 0 - console.log('[SSE] 连接已建立') + sseBus.emit('connected', undefined) + console.log('[SSE] Connected') }, - /** 收到服务端事件 */ - onmessage(msg) { - // 心跳和 connected 确认消息可能 data 为空 - if (!msg.data) return + onmessage(message) { + if (!message.event || !message.data) { + return + } try { - const data = JSON.parse(msg.data) - // eslint-disable-next-line @typescript-eslint/no-explicit-any - sseBus.emit(msg.event as any, data) + const data = JSON.parse(message.data) + sseBus.emit(message.event as keyof SseEvents, data) } catch { - console.warn('[SSE] 事件数据解析失败:', msg.event, msg.data) + console.warn('[SSE] Failed to parse event payload:', message.event, message.data) } }, - /** 连接关闭 */ onclose() { - connecting = false - isConnected.value = false - console.log('[SSE] 连接已关闭') + resetConnectionState() + console.log('[SSE] Closed') }, - /** - * 连接错误处理(必须同步,签名: (err) => number | void) - * - * 401 → fire-and-forget 刷新 token,阻止当前连接重连, - * 刷新成功后由 Vue watcher 触发 connect() 自动重连 - * 其他错误 → 返回重试延迟(ms),库自动排队重连 - */ - onerror(err) { - connecting = false - isConnected.value = false + onerror(error) { + resetConnectionState() - // 鉴权失败 → 刷新 token(异步 fire-and-forget,onerror 本身保持同步) - if (err.response?.status === 401) { - if (isRefreshing) { - // 已有刷新在进行中,直接放弃本次连接 - throw err - } - - isRefreshing = true - const store = useAccountStore() - - if (!store.refreshToken) { - sseBus.emit('error', { message: '登录已过期,请重新登录', retryCount: 0 }) - throw err - } - - // 异步刷新,不阻塞 onerror 返回 - store - .refreshAccessToken() - .then((newToken) => { - if (newToken) { - console.log('[SSE] token 刷新成功,重新建立连接') - retryCount = 0 - // connect() 由 App.vue 的 watch 触发(accessToken 变更) - } - }) - .catch(() => { - console.warn('[SSE] token 刷新失败') - sseBus.emit('error', { message: '实时连接异常,请刷新页面', retryCount: 0 }) - }) - .finally(() => { - isRefreshing = false - }) - - throw err // 不自动重连 + if (isSseHttpError(error) && error.response.status === 401) { + void refreshTokenAndReconnect() + throw error } - // 其他错误:指数退避重连 - retryCount++ - if (retryCount >= 5) { + retryCount += 1 + if (retryCount >= MAX_RETRY_COUNT) { sseBus.emit('error', { - message: '实时连接异常,请刷新页面', + message: 'Realtime connection failed. Please refresh the page.', retryCount }) - throw err // 超过阈值,停止重连 + throw error } - // 返回重试间隔(ms),指数退避:3s, 6s, 12s, 24s,最大 30s - const delay = Math.min(3000 * Math.pow(2, retryCount - 1), 30000) - console.warn(`[SSE] 连接错误,${delay / 1000}s 后重试 (${retryCount}/5)`) + const delay = getRetryDelay() + console.warn(`[SSE] Connection error, retry in ${delay / 1000}s (${retryCount}/${MAX_RETRY_COUNT})`) return delay } }) } - /** - * 取消当前 SSE 连接 - */ function abort(): void { if (abortController) { abortController.abort() abortController = null } - connecting = false - isConnected.value = false + currentToken = null retryCount = 0 + resetConnectionState() } - /** - * 断开 SSE 连接 - * - * 在用户登出时调用。 - */ function disconnect(): void { abort() - console.log('[SSE] 连接已断开') + console.log('[SSE] Disconnected') + } + + async function refreshTokenAndReconnect(): Promise { + 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 { diff --git a/src/stores/sseBus.ts b/src/stores/sseBus.ts index cab447b..8185c28 100644 --- a/src/stores/sseBus.ts +++ b/src/stores/sseBus.ts @@ -1,40 +1,19 @@ -// webgame-frontend/src/stores/sseBus.ts import mitt from 'mitt' import type { RoomItem } from '@/api/modules/game' -/** - * SSE 推送事件类型映射 - * - * 每个事件类型对应一个数据类型,类型安全。 - * connected / error 由 useSse 连接管理发出,其余由服务端 SSE 事件映射而来。 - */ -type SseEvents = { - /** SSE 连接成功 */ - connected: void - /** SSE 连接异常 */ +export type SseEvents = { + connected: undefined error: { message: string; retryCount: number } - /** 新房间创建 */ room_created: RoomItem - /** 房间状态更新 */ room_updated: RoomItem - /** 房间移除 */ room_removed: { roomId: string } - /** 被踢出房间 */ kicked: { roomId: string } - /** 所在房间游戏开始 */ game_started: { roomId: string } - /** 余额变动 */ balance_chg: { balance: number; delta: number } } /** - * 全局 SSE 事件总线(发布-订阅模式) - * - * 示例: - * ```ts - * import { sseBus } from '@/stores/sseBus' - * const off = sseBus.on('room_updated', (room) => { ... }) - * onUnmounted(off) - * ``` + * Global SSE event bus. The connection layer only dispatches events; feature + * modules decide how to handle them. */ export const sseBus = mitt()