feat(messages): 接入用户聊天 SSE 实时刷新

docs/Phase2-聊天实时推送验收.md

@@ -0,0 +1,33 @@
+# Phase 2 聊天实时推送验收
+
+> 日期：2026-10-30
+> 范围：用户端聊天 SSE 实时刷新、轮询降频兜底。
+
+## 1. 交付内容
+
+- 新增 `frontend/src/features/messages/ChatEventsBridge.tsx`
+  - 使用 `fetch` 流读取 `GET /api/chats/events`，解决浏览器 `EventSource` 无法附加 `Authorization` 的问题。
+  - 自动携带 Bearer token、Cookie 与分销站点头 `X-Distribution-Site-Code`。
+  - 解析 `event:` / `data:` 块，兼容 CRLF 与多行 data。
+  - 收到事件后刷新：
+    - `chat-conversations`
+    - `chat-unread-count`
+    - `notifications`
+    - 对应会话的 `chat-messages` 与 `chat-conversation`
+  - 断线后 3 秒自动重连，组件卸载时中止连接并清理定时器。
+- `frontend/src/App.tsx` 在 `QueryClientProvider` 与 `BrowserRouter` 内挂载桥接组件。
+- `useChatMessages` 轮询从 6 秒调整为 30 秒，仅作为 SSE 不可用时的兜底，减少无效请求。
+
+## 2. 验证记录
+
+| 检查 | 结果 |
+|---|---|
+| `curl --max-time 3 -H 'Accept: text/event-stream' /api/chats/events` | 返回 `event: connected` 与 `data: {"ok":true}` |
+| SSE 鉴权 | 复用用户 Bearer token，后端返回 200 |
+| 前端类型检查 / 构建 | 见本次提交时的 `npx tsc --noEmit`、`npm run build` 结果 |
+
+## 3. 说明
+
+- 当前仍是「SSE 实时 + 30 秒轮询兜底」；后续可在连接状态稳定后进一步关闭轮询。
+- 事件中的消息体暂不直接写入 React Query 缓存，而是触发对应会话刷新，避免分页合并与去重逻辑分叉。
+- 多实例部署时，SSE 只推送当前进程内连接；若后端横向扩容，需要 Redis Pub/Sub 或粘性会话。

frontend/src/App.tsx

@@ -5,6 +5,7 @@ import { QueryClient, QueryClientProvider } from '@tanstack/react-query'
 import { BrowserRouter } from 'react-router-dom'
 
 import { AppRouter } from '@/app/AppRouter'
+import { ChatEventsBridge } from '@/features/messages/ChatEventsBridge'
 import { useSessionStore } from '@/stores/session'
 import { jieyouTheme } from '@/theme'
 
@@ -41,6 +42,7 @@ export default function App() {
         <QueryClientProvider client={queryClient}>
           <BrowserRouter>
             <SessionBootstrap />
+            <ChatEventsBridge />
             <AppRouter />
           </BrowserRouter>
         </QueryClientProvider>

frontend/src/features/messages/ChatEventsBridge.tsx

@@ -0,0 +1,118 @@
+import { useEffect } from 'react'
+import { useQueryClient } from '@tanstack/react-query'
+
+import { useSessionStore } from '@/stores/session'
+
+interface ChatEvent {
+  type: string
+  conversation_id: number
+}
+
+function getSiteCode() {
+  try {
+    return sessionStorage.getItem('jieyou_site_code') || ''
+  } catch {
+    return ''
+  }
+}
+
+function parseEventBlock(block: string): ChatEvent | null {
+  let data = ''
+  for (const rawLine of block.split('\n')) {
+    const line = rawLine.replace(/\r$/, '')
+    if (!line.startsWith('data:')) continue
+    data += line.slice(5).trimStart() + '\n'
+  }
+  const payload = data.trim()
+  if (!payload) return null
+  try {
+    const parsed = JSON.parse(payload) as Partial<ChatEvent>
+    if (!parsed.type || !parsed.conversation_id) return null
+    return { type: parsed.type, conversation_id: Number(parsed.conversation_id) }
+  } catch {
+    return null
+  }
+}
+
+/**
+ * 用户端聊天 SSE 桥接：EventSource 无法附加 Authorization，这里用 fetch 流读取。
+ * SSE 只作为实时加速，页面仍保留低频轮询兜底。
+ */
+export function ChatEventsBridge() {
+  const queryClient = useQueryClient()
+  const token = useSessionStore(state => state.token)
+
+  useEffect(() => {
+    if (!token) return
+
+    const controller = new AbortController()
+    let stopped = false
+    let retryTimer: number | undefined
+
+    const invalidate = (conversationID: number) => {
+      void queryClient.invalidateQueries({ queryKey: ['chat-conversations'] })
+      void queryClient.invalidateQueries({ queryKey: ['chat-unread-count'] })
+      void queryClient.invalidateQueries({ queryKey: ['notifications'] })
+      void queryClient.invalidateQueries({
+        queryKey: ['chat-messages', String(conversationID)],
+      })
+      void queryClient.invalidateQueries({
+        queryKey: ['chat-conversation', String(conversationID)],
+      })
+    }
+
+    const connect = async () => {
+      try {
+        const headers: Record<string, string> = {
+          Accept: 'text/event-stream',
+          Authorization: `Bearer ${token}`,
+        }
+        const siteCode = getSiteCode()
+        if (siteCode) headers['X-Distribution-Site-Code'] = siteCode
+
+        const response = await fetch('/api/chats/events', {
+          headers,
+          credentials: 'include',
+          signal: controller.signal,
+        })
+        if (!response.ok || !response.body) {
+          throw new Error(`chat events unavailable: ${response.status}`)
+        }
+
+        const reader = response.body.getReader()
+        const decoder = new TextDecoder()
+        let buffer = ''
+
+        while (!stopped) {
+          const { value, done } = await reader.read()
+          if (done) break
+          buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, '\n')
+          let boundary = buffer.indexOf('\n\n')
+          while (boundary >= 0) {
+            const block = buffer.slice(0, boundary)
+            buffer = buffer.slice(boundary + 2)
+            const event = parseEventBlock(block)
+            if (event) invalidate(event.conversation_id)
+            boundary = buffer.indexOf('\n\n')
+          }
+        }
+      } catch {
+        // 断线后延迟重连；页面轮询仍可兜底。
+      }
+
+      if (!stopped) {
+        retryTimer = window.setTimeout(() => void connect(), 3000)
+      }
+    }
+
+    void connect()
+
+    return () => {
+      stopped = true
+      controller.abort()
+      if (retryTimer) window.clearTimeout(retryTimer)
+    }
+  }, [queryClient, token])
+
+  return null
+}

frontend/src/features/messages/api.ts

@@ -84,7 +84,7 @@ export function useChatMessages(id?: string, enabled = true) {
     getNextPageParam: last =>
       last.page * last.page_size < last.total ? last.page + 1 : undefined,
     enabled: Boolean(id) && enabled,
-    refetchInterval: 6000,
+    refetchInterval: 30_000,
   })
 }