优化实时聊天：稳定 WS、按 seq 增量同步，并修复访客评价推送

- 客服/访客 WebSocket 不再因切换会话反复重连，断线自动恢复 - 新增 after_seq 增量拉取，重连与消息空洞时 catch-up - 结束会话同时推送给访客，预览页可弹出评价并支持兜底状态同步
2026-07-15 14:38:08 +08:00
parent ea3901d8b0
commit 1359bfe597
7 changed files with 758 additions and 107 deletions
@@ -10,11 +10,40 @@ import { ChatImage } from '@/components/common/ImagePreview'
 import { useAuth } from '@/stores/auth'
 import {
  addSessionNote, claimSession, endSession, getAvailableAgents, getCustomers, getKnowledgeEntries,
-  getSession, getSessions, markSessionRead, sendSessionMessage, transferSession, updateSessionPriority,
+  getSession, getSessionMessages, getSessions, markSessionRead, sendSessionMessage, transferSession, updateSessionPriority,
  uploadImage,
  type AvailableAgent, type Customer, type KnowledgeEntry, type Message, type Session, type SessionEvent,
 } from '@/services/api'

+/** 按 id 合并消息，再按 seq / id 排序 */
+function mergeMessagesBySeq(existing: Message[], incoming: Message[]): Message[] {
+  const map = new Map<number, Message>()
+  for (const item of existing) map.set(item.id, item)
+  for (const item of incoming) map.set(item.id, item)
+  return Array.from(map.values()).sort((a, b) => {
+    const seqDiff = (a.seq || 0) - (b.seq || 0)
+    return seqDiff !== 0 ? seqDiff : a.id - b.id
+  })
+}
+
+function maxMessageSeq(messages: Message[]): number {
+  return messages.reduce((acc, item) => Math.max(acc, item.seq || 0), 0)
+}
+
+function normalizeMessage(raw: Partial<Message> & { id?: number }, sessionId: number): Message | null {
+  if (!raw?.id) return null
+  return {
+    id: Number(raw.id),
+    session_id: Number(raw.session_id ?? sessionId),
+    sender_type: raw.sender_type === 'agent' ? 'agent' : 'visitor',
+    sender_id: raw.sender_id ?? null,
+    content: String(raw.content ?? ''),
+    type: raw.type === 'image' ? 'image' : 'text',
+    seq: Number(raw.seq ?? 0),
+    sent_at: raw.sent_at || new Date().toISOString(),
+  }
+}
+
 const endReasons = [
  { value: 'resolved', label: '已解决' },
  { value: 'no_response', label: '无人回复' },
@@ -128,6 +157,23 @@ const Dashboard = () => {
  const socketRef = useRef<WebSocket | null>(null)
  const lastTypingAt = useRef(0)
  const visitorTypingTimer = useRef<number | null>(null)
+  /** 当前选中会话，供稳定 WS 闭包读取，避免切会话时重连 */
+  const selectedIdRef = useRef<number | null>(null)
+  /** 各会话本地已同步到的最大 seq，用于重连增量 catch-up */
+  const lastSeqBySessionRef = useRef<Record<number, number>>({})
+  const loadAllRef = useRef<(opts?: { silent?: boolean }) => Promise<void>>(async () => {})
+  const loadDetailRef = useRef<(id: number, markRead?: boolean, opts?: { silent?: boolean }) => Promise<void>>(async () => {})
+  const syncAfterSeqRef = useRef<(id: number, opts?: { markRead?: boolean }) => Promise<void>>(async () => {})
+
+  const rememberSeq = useCallback((sessionId: number, seq: number) => {
+    if (!sessionId || !seq || seq <= 0) return
+    const prev = lastSeqBySessionRef.current[sessionId] || 0
+    if (seq > prev) lastSeqBySessionRef.current[sessionId] = seq
+  }, [])
+
+  const rememberMessagesSeq = useCallback((sessionId: number, messages: Message[]) => {
+    rememberSeq(sessionId, maxMessageSeq(messages))
+  }, [rememberSeq])

  const insertEmoji = (emoji: string) => {
    const { next, cursor } = insertAtCursor(messageInputRef.current, messageInput, emoji)
@@ -143,8 +189,8 @@ const Dashboard = () => {

  const isManager = user?.role === 'admin' || user?.role === 'supervisor'

-  const loadAll = useCallback(async () => {
-    setLoading(true)
+  const loadAll = useCallback(async (opts?: { silent?: boolean }) => {
+    if (!opts?.silent) setLoading(true)
    try {
      const [sessionRes, customerRes] = await Promise.all([
        getSessions({ page: 1, pageSize: 100 }),
@@ -162,20 +208,25 @@ const Dashboard = () => {
        initialLoad.current = false
      }
    } catch {
-      antMsg.error('加载会话失败')
-      setSessions([])
-      setCustomers({})
+      if (!opts?.silent) antMsg.error('加载会话失败')
+      if (!opts?.silent) {
+        setSessions([])
+        setCustomers({})
+      }
    } finally {
-      setLoading(false)
+      if (!opts?.silent) setLoading(false)
    }
  }, [])

-  const loadDetail = useCallback(async (id: number, markRead = true) => {
-    setDetailLoading(true)
+  const loadDetail = useCallback(async (id: number, markRead = true, opts?: { silent?: boolean }) => {
+    if (!opts?.silent) setDetailLoading(true)
    try {
      const response = await getSession(id)
      const data = response.data
-      setDetail({ messages: data.messages || [], events: data.events || [], pendingCount: data.pending_count || 0 })
+      const messages = data.messages || []
+      setDetail({ messages, events: data.events || [], pendingCount: data.pending_count || 0 })
+      rememberMessagesSeq(id, messages)
+      if (typeof data.max_seq === 'number') rememberSeq(id, data.max_seq)
      // 用详情接口的完整会话字段（含 IP/地区）回填列表，保证顶栏展示准确
      if (data.session) {
        setSessions(previous => previous.map(session =>
@@ -184,8 +235,8 @@ const Dashboard = () => {
                ...session,
                ...data.session,
                unread_count: markRead ? 0 : session.unread_count,
-                last_message: session.last_message,
-                last_message_at: session.last_message_at,
+                last_message: data.session.last_message ?? session.last_message,
+                last_message_at: data.session.last_message_at ?? session.last_message_at,
              }
            : session,
        ))
@@ -196,46 +247,217 @@ const Dashboard = () => {
        await markSessionRead(id)
      }
    } catch {
-      antMsg.error('加载消息失败')
+      if (!opts?.silent) antMsg.error('加载消息失败')
    } finally {
-      setDetailLoading(false)
+      if (!opts?.silent) setDetailLoading(false)
    }
    setTimeout(() => chatEndRef.current?.scrollIntoView({ behavior: 'smooth' }), 80)
-  }, [])
+  }, [rememberMessagesSeq, rememberSeq])

-  useEffect(() => { loadAll() }, [loadAll])
+  /** 重连 / 发现 seq 空洞时：只拉 after_seq 之后的消息并合并 */
+  const syncAfterSeq = useCallback(async (id: number, opts?: { markRead?: boolean }) => {
+    const afterSeq = lastSeqBySessionRef.current[id] || 0
+    // 本地尚无游标：退回全量详情
+    if (afterSeq <= 0) {
+      await loadDetail(id, opts?.markRead ?? true, { silent: true })
+      return
+    }
+    try {
+      let cursor = afterSeq
+      let guard = 0
+      let latest: Message | null = null
+      while (guard < 10) {
+        guard += 1
+        const response = await getSessionMessages(id, cursor, 200)
+        const batch = response.data.messages || []
+        const maxSeq = Number(response.data.max_seq || cursor)
+        if (batch.length > 0) {
+          latest = batch[batch.length - 1]
+          setDetail(previous => {
+            if (selectedIdRef.current !== id) return previous
+            if (!previous) return { messages: batch, events: [], pendingCount: 0 }
+            return { ...previous, messages: mergeMessagesBySeq(previous.messages, batch) }
+          })
+          rememberMessagesSeq(id, batch)
+        }
+        rememberSeq(id, maxSeq)
+        const lastBatchSeq = batch.length ? maxMessageSeq(batch) : cursor
+        if (!response.data.has_more || lastBatchSeq >= maxSeq || batch.length === 0) break
+        cursor = lastBatchSeq
+      }
+      if (latest && selectedIdRef.current === id) {
+        const preview = latest.type === 'image' ? '[图片]' : latest.content
+        setSessions(previous => previous.map(session =>
+          session.id === id
+            ? {
+                ...session,
+                last_message: preview,
+                last_message_at: latest!.sent_at,
+                unread_count: opts?.markRead === false ? session.unread_count : 0,
+              }
+            : session,
+        ))
+        setTimeout(() => chatEndRef.current?.scrollIntoView({ behavior: 'smooth' }), 40)
+      }
+      if (opts?.markRead !== false) {
+        await markSessionRead(id).catch(() => {})
+      }
+    } catch {
+      await loadDetail(id, opts?.markRead ?? true, { silent: true })
+    }
+  }, [loadDetail, rememberMessagesSeq, rememberSeq])
+
+  /** 将 WS 推送的消息直接追加；若 seq 有空洞则触发增量同步 */
+  const appendPushedMessage = useCallback((sessionId: number, raw: Partial<Message> & { id?: number }) => {
+    const msg = normalizeMessage(raw, sessionId)
+    if (!msg) return false
+
+    const known = lastSeqBySessionRef.current[sessionId] || 0
+    // 已知游标且出现空洞（漏消息）→ 走 after_seq 增量补齐
+    if (known > 0 && msg.seq > 0 && msg.seq > known + 1) {
+      void syncAfterSeqRef.current(sessionId, { markRead: true })
+      return true
+    }
+
+    const preview = msg.type === 'image' ? '[图片]' : msg.content
+    let appended = false
+    setDetail(previous => {
+      if (selectedIdRef.current !== sessionId) return previous
+      if (!previous) {
+        appended = true
+        return { messages: [msg], events: [], pendingCount: 0 }
+      }
+      if (previous.messages.some(item => item.id === msg.id)) {
+        appended = true
+        return previous
+      }
+      appended = true
+      return { ...previous, messages: mergeMessagesBySeq(previous.messages, [msg]) }
+    })
+    setSessions(previous => previous.map(session => {
+      if (session.id !== sessionId) return session
+      const viewing = selectedIdRef.current === sessionId
+      return {
+        ...session,
+        last_message: preview,
+        last_message_at: msg.sent_at,
+        unread_count: viewing ? 0 : (session.unread_count || 0) + (msg.sender_type === 'visitor' ? 1 : 0),
+      }
+    }))
+    if (msg.seq > 0) rememberSeq(sessionId, msg.seq)
+    if (selectedIdRef.current === sessionId) {
+      setTimeout(() => chatEndRef.current?.scrollIntoView({ behavior: 'smooth' }), 40)
+      void markSessionRead(sessionId).catch(() => {})
+    }
+    return appended
+  }, [rememberSeq])

+  useEffect(() => {
+    selectedIdRef.current = selectedId
+  }, [selectedId])
+
+  useEffect(() => {
+    loadAllRef.current = loadAll
+    loadDetailRef.current = loadDetail
+    syncAfterSeqRef.current = syncAfterSeq
+  }, [loadAll, loadDetail, syncAfterSeq])
+
+  useEffect(() => { void loadAll() }, [loadAll])
+
+  // 稳定 WebSocket：仅 token 变化时建连；切会话不重连；断线自动重连（ws / wss）
  useEffect(() => {
    if (!user?.token) return
-    const scheme = window.location.protocol === 'https:' ? 'wss:' : 'ws:'
-    const socket = new WebSocket(`${scheme}//${window.location.host}/api/ws`, ['kefu-v1', user.token])
-    socketRef.current = socket
-    socket.onmessage = event => {
-      try {
-        const payload = JSON.parse(event.data)
-        if (payload.type === 'typing' && payload.session_id === selectedId && payload.data?.from === 'visitor') {
-          setVisitorTyping(true)
-          if (visitorTypingTimer.current) clearTimeout(visitorTypingTimer.current)
-          visitorTypingTimer.current = window.setTimeout(() => setVisitorTyping(false), 1800)
-          return
-        }
-        if (payload.type === 'message' && payload.session_id === selectedId) {
-          setVisitorTyping(false)
-          loadDetail(payload.session_id)
-        }
-        if (payload.type === 'message' || payload.type === 'session_created' || payload.type === 'session_updated') {
-          loadAll()
-        }
-      } catch {
-        // ignore
+    let disposed = false
+    let reconnectTimer: number | null = null
+    let attempt = 0
+
+    const clearReconnect = () => {
+      if (reconnectTimer != null) {
+        window.clearTimeout(reconnectTimer)
+        reconnectTimer = null
      }
    }
+
+    const connect = () => {
+      if (disposed) return
+      const scheme = window.location.protocol === 'https:' ? 'wss:' : 'ws:'
+      const socket = new WebSocket(`${scheme}//${window.location.host}/api/ws`, ['kefu-v1', user.token])
+      socketRef.current = socket
+
+      socket.onopen = () => {
+        attempt = 0
+        // 重连后：列表静默刷新 + 当前会话按 seq 增量 catch-up
+        void loadAllRef.current({ silent: true })
+        const sid = selectedIdRef.current
+        if (sid != null) void syncAfterSeqRef.current(sid, { markRead: true })
+      }
+
+      socket.onmessage = event => {
+        try {
+          const payload = JSON.parse(event.data) as {
+            type?: string
+            session_id?: number | string
+            data?: Partial<Message> & { from?: string; status?: string; agent_name?: string }
+          }
+          const sid = Number(payload.session_id)
+          const current = selectedIdRef.current
+          const sameSession = Number.isFinite(sid) && current != null && sid === Number(current)
+
+          if (payload.type === 'typing' && sameSession && payload.data?.from === 'visitor') {
+            setVisitorTyping(true)
+            if (visitorTypingTimer.current) clearTimeout(visitorTypingTimer.current)
+            visitorTypingTimer.current = window.setTimeout(() => setVisitorTyping(false), 1800)
+            return
+          }
+
+          if (payload.type === 'message') {
+            if (sameSession) {
+              setVisitorTyping(false)
+              const ok = appendPushedMessage(sid, payload.data || {})
+              if (!ok) void syncAfterSeqRef.current(sid, { markRead: true })
+            }
+            void loadAllRef.current({ silent: true })
+            return
+          }
+
+          if (payload.type === 'session_created' || payload.type === 'session_updated') {
+            void loadAllRef.current({ silent: true })
+            if (sameSession) void syncAfterSeqRef.current(sid, { markRead: false })
+          }
+        } catch {
+          // ignore malformed frames
+        }
+      }
+
+      socket.onclose = () => {
+        if (socketRef.current === socket) socketRef.current = null
+        if (disposed) return
+        const delay = Math.min(1000 * (2 ** attempt), 15000)
+        attempt += 1
+        clearReconnect()
+        reconnectTimer = window.setTimeout(connect, delay)
+      }
+
+      socket.onerror = () => {
+        // onclose 会负责重连
+        try { socket.close() } catch { /* ignore */ }
+      }
+    }
+
+    connect()
+
    return () => {
-      socket.close()
+      disposed = true
+      clearReconnect()
+      const sock = socketRef.current
      socketRef.current = null
+      if (sock) {
+        sock.onclose = null
+        sock.close()
+      }
      if (visitorTypingTimer.current) clearTimeout(visitorTypingTimer.current)
    }
-  }, [user?.token, selectedId, loadAll, loadDetail])
+  }, [user?.token, appendPushedMessage])

  useEffect(() => {
    setVisitorTyping(false)
@@ -316,10 +538,16 @@ const Dashboard = () => {
    if (!selectedId || !canOperate || sending || (type === 'text' && !content.trim())) return
    setSending(true)
    try {
-      await sendSessionMessage(selectedId, content, type)
+      const response = await sendSessionMessage(selectedId, content, type)
      if (type === 'text') setMessageInput('')
-      await loadDetail(selectedId)
-      await loadAll()
+      // 发送成功后优先用返回体追加并推进 seq，失败再静默全量
+      if (response.data?.id) {
+        appendPushedMessage(selectedId, response.data)
+        void loadAll({ silent: true })
+      } else {
+        await syncAfterSeq(selectedId, { markRead: true })
+        await loadAll({ silent: true })
+      }
    } catch (error) {
      antMsg.error(error instanceof Error ? error.message : '发送失败')
    } finally {
@@ -347,11 +575,16 @@ const Dashboard = () => {
    setSending(true)
    try {
      const uploaded = await uploadImage(pendingImage.file)
-      await sendSessionMessage(selectedId, uploaded.data.url, 'image')
+      const response = await sendSessionMessage(selectedId, uploaded.data.url, 'image')
      URL.revokeObjectURL(pendingImage.preview)
      setPendingImage(null)
-      await loadDetail(selectedId)
-      await loadAll()
+      if (response.data?.id) {
+        appendPushedMessage(selectedId, response.data)
+        void loadAll({ silent: true })
+      } else {
+        await syncAfterSeq(selectedId, { markRead: true })
+        await loadAll({ silent: true })
+      }
    } catch (error) {
      antMsg.error(error instanceof Error ? error.message : '图片发送失败')
    } finally {