diff --git a/server/internal/handler/router.go b/server/internal/handler/router.go index e6207e5..43c2b88 100644 --- a/server/internal/handler/router.go +++ b/server/internal/handler/router.go @@ -61,6 +61,7 @@ func SetupRoutes(r *gin.Engine, store storage.ObjectStorage, storageCfg config.S sessions.POST("/:id/end", session.End) sessions.POST("/:id/archive", session.Archive) sessions.PUT("/:id/priority", session.UpdatePriority) + sessions.GET("/:id/messages", session.ListMessages) sessions.POST("/:id/messages", session.SendMessage) sessions.POST("/:id/read", session.MarkRead) sessions.POST("/:id/notes", session.AddNote) diff --git a/server/internal/handler/session.go b/server/internal/handler/session.go index 6c2c602..d9b8562 100644 --- a/server/internal/handler/session.go +++ b/server/internal/handler/session.go @@ -2,6 +2,7 @@ package handler import ( "net/http" + "strconv" "strings" "time" "unicode/utf8" @@ -145,9 +146,12 @@ func broadcastSessionMessage(session *model.Session, message model.Message) { func broadcastSessionUpdate(session *model.Session) { payload, err := ws.NewEvent("session_updated", session.ID, gin.H{"status": session.Status}) - if err == nil { - ws.DefaultHub.BroadcastToTenantStaff(session.TenantID, payload) + if err != nil { + return } + // 坐席列表需要刷新；访客也必须收到 ended/active，才能弹出评价或更新接待状态 + ws.DefaultHub.BroadcastToTenantStaff(session.TenantID, payload) + ws.DefaultHub.BroadcastToVisitor(session.TenantID, session.ID, payload) } func loadAssignableAgent(tenantID, agentID uint) error { @@ -351,7 +355,79 @@ func (h *SessionHandler) Get(c *gin.Context) { var pendingCount int64 model.DB.Model(&model.Session{}).Where("customer_id = ? AND tenant_id = ? AND status = ?", session.CustomerID, session.TenantID, "waiting").Count(&pendingCount) - middleware.JSON(c, gin.H{"session": session, "messages": messages, "events": events, "pending_count": pendingCount}) + maxSeq := 0 + for _, m := range messages { + if m.Seq > maxSeq { + maxSeq = m.Seq + } + } + middleware.JSON(c, gin.H{ + "session": session, + "messages": messages, + "events": events, + "pending_count": pendingCount, + "max_seq": maxSeq, + }) +} + +// ListMessages 按 seq 增量拉取消息：after_seq 之后的新消息（重连 catch-up）。 +// GET /sessions/:id/messages?after_seq=12 +func (h *SessionHandler) ListMessages(c *gin.Context) { + session, ok := loadTenantSession(c, c.Param("id")) + if !ok { + return + } + if !canReadSession(c, session) { + c.JSON(http.StatusForbidden, gin.H{"code": 403, "message": "无权查看该会话消息"}) + return + } + + afterSeq := 0 + if raw := strings.TrimSpace(c.Query("after_seq")); raw != "" { + parsed, err := strconv.Atoi(raw) + if err != nil || parsed < 0 { + c.JSON(http.StatusBadRequest, gin.H{"code": 400, "message": "after_seq 无效"}) + return + } + afterSeq = parsed + } + + limit := 200 + if raw := strings.TrimSpace(c.Query("limit")); raw != "" { + if parsed, err := strconv.Atoi(raw); err == nil && parsed > 0 { + if parsed > 500 { + parsed = 500 + } + limit = parsed + } + } + + query := model.DB.Where("session_id = ?", session.ID) + if afterSeq > 0 { + query = query.Where("seq > ?", afterSeq) + } + + var messages []model.Message + if err := query.Order("seq asc").Limit(limit).Find(&messages).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"code": 500, "message": "查询消息失败"}) + return + } + + var maxSeq int + if err := model.DB.Model(&model.Message{}). + Where("session_id = ?", session.ID). + Select("COALESCE(MAX(seq), 0)"). + Scan(&maxSeq).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"code": 500, "message": "查询消息失败"}) + return + } + + middleware.JSON(c, gin.H{ + "messages": messages, + "after_seq": afterSeq, + "max_seq": maxSeq, + "has_more": len(messages) >= limit && (len(messages) == 0 || messages[len(messages)-1].Seq < maxSeq), + }) } func (h *SessionHandler) MarkRead(c *gin.Context) { diff --git a/server/internal/handler/widget.go b/server/internal/handler/widget.go index bb1e338..e97b23e 100644 --- a/server/internal/handler/widget.go +++ b/server/internal/handler/widget.go @@ -479,15 +479,61 @@ func (h *WidgetHandler) GetMessages(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"code": 400, "message": "会话参数错误"}) return } - if _, ok := loadVisitorSession(c, uint(sessionID), visitorTokenFromRequest(c, "")); !ok { + session, ok := loadVisitorSession(c, uint(sessionID), visitorTokenFromRequest(c, "")) + if !ok { return } + afterSeq := 0 + if raw := strings.TrimSpace(c.Query("after_seq")); raw != "" { + parsed, err := strconv.Atoi(raw) + if err != nil || parsed < 0 { + c.JSON(http.StatusBadRequest, gin.H{"code": 400, "message": "after_seq 无效"}) + return + } + afterSeq = parsed + } + + limit := 200 + if raw := strings.TrimSpace(c.Query("limit")); raw != "" { + if parsed, err := strconv.Atoi(raw); err == nil && parsed > 0 { + if parsed > 500 { + parsed = 500 + } + limit = parsed + } + } + + query := model.DB.Where("session_id = ?", sessionID) + if afterSeq > 0 { + query = query.Where("seq > ?", afterSeq) + } + var messages []model.Message - if err := model.DB.Where("session_id = ?", sessionID).Order("seq asc").Find(&messages).Error; err != nil { + if err := query.Order("seq asc").Limit(limit).Find(&messages).Error; err != nil { c.JSON(http.StatusInternalServerError, gin.H{"code": 500, "message": "查询消息失败"}) return } - c.JSON(http.StatusOK, gin.H{"code": 0, "data": messages}) + var maxSeq int + if err := model.DB.Model(&model.Message{}). + Where("session_id = ?", sessionID). + Select("COALESCE(MAX(seq), 0)"). + Scan(&maxSeq).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"code": 500, "message": "查询消息失败"}) + return + } + + hasMore := len(messages) >= limit && (len(messages) == 0 || messages[len(messages)-1].Seq < maxSeq) + c.JSON(http.StatusOK, gin.H{ + "code": 0, + "data": gin.H{ + "messages": messages, + "after_seq": afterSeq, + "max_seq": maxSeq, + "has_more": hasMore, + "session_status": session.Status, + "satisfaction_score": session.SatisfactionScore, + }, + }) } diff --git a/web/src/pages/agent/Dashboard.tsx b/web/src/pages/agent/Dashboard.tsx index 7b07625..f520d1d 100644 --- a/web/src/pages/agent/Dashboard.tsx +++ b/web/src/pages/agent/Dashboard.tsx @@ -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() + 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 & { 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(null) const lastTypingAt = useRef(0) const visitorTypingTimer = useRef(null) + /** 当前选中会话，供稳定 WS 闭包读取，避免切会话时重连 */ + const selectedIdRef = useRef(null) + /** 各会话本地已同步到的最大 seq，用于重连增量 catch-up */ + const lastSeqBySessionRef = useRef>({}) + const loadAllRef = useRef<(opts?: { silent?: boolean }) => Promise>(async () => {}) + const loadDetailRef = useRef<(id: number, markRead?: boolean, opts?: { silent?: boolean }) => Promise>(async () => {}) + const syncAfterSeqRef = useRef<(id: number, opts?: { markRead?: boolean }) => Promise>(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 & { 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 & { 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 { diff --git a/web/src/services/api.ts b/web/src/services/api.ts index 860c412..c5a35b2 100644 --- a/web/src/services/api.ts +++ b/web/src/services/api.ts @@ -125,7 +125,29 @@ export const getSessions = (params?: { if (params?.pageSize) search.set('pageSize', String(params.pageSize)) return getList(`/sessions?${search}`) } -export const getSession = (id: number) => get<{ session: Session; messages: Message[]; events: SessionEvent[]; pending_count: number }>(`/sessions/${id}`) +export const getSession = (id: number) => get<{ + session: Session + messages: Message[] + events: SessionEvent[] + pending_count: number + max_seq?: number +}>(`/sessions/${id}`) + +/** 按 seq 增量拉取消息（重连 catch-up）：seq > after_seq */ +export interface SessionMessagesSync { + messages: Message[] + after_seq: number + max_seq: number + has_more: boolean +} +export const getSessionMessages = (id: number, afterSeq = 0, limit = 200) => { + const search = new URLSearchParams() + if (afterSeq > 0) search.set('after_seq', String(afterSeq)) + if (limit > 0) search.set('limit', String(limit)) + const qs = search.toString() + return get(`/sessions/${id}/messages${qs ? `?${qs}` : ''}`) +} + export const assignSession = (id: number, agentId: number) => post(`/sessions/${id}/assign`, { agent_id: agentId }) export const claimSession = (id: number) => post(`/sessions/${id}/assign`, {}) export const transferSession = (id: number, agentId: number) => post(`/sessions/${id}/transfer`, { agent_id: agentId }) diff --git a/web/src/widgets/VisitorChat.tsx b/web/src/widgets/VisitorChat.tsx index 5ec5ae9..5587fbc 100644 --- a/web/src/widgets/VisitorChat.tsx +++ b/web/src/widgets/VisitorChat.tsx @@ -12,6 +12,8 @@ interface Message { content: string time: string type?: 'text' | 'image' + /** 服务端单调序号，用于增量同步 */ + seq?: number } const defaultQuickQuestions = ['查询订单状态', '退换货政策', '配送时效说明'] @@ -73,6 +75,22 @@ const VisitorChat = ({ const textInputRef = useRef(null) const socketRef = useRef(null) const lastTypingAt = useRef(0) + const sessionIdRef = useRef(sessionId) + const visitorTokenRef = useRef(visitorToken) + const agentNameRef = useRef(agentName) + const loadMessagesRef = useRef<(sid?: number, token?: string, opts?: { afterSeq?: number; full?: boolean }) => Promise>(async () => {}) + const msgsKeyRef = useRef(msgsKey) + /** 本地已同步到的最大 seq（从缓存消息初始化） */ + const lastSeqRef = useRef((() => { + try { + const saved = localStorage.getItem(msgsKey) + if (!saved) return 0 + const cached = JSON.parse(saved) as Message[] + return cached.reduce((acc, m) => Math.max(acc, m.seq || 0), 0) + } catch { + return 0 + } + })()) const insertEmoji = (emoji: string) => { const { next, cursor } = insertAtCursor(textInputRef.current, input, emoji) @@ -86,32 +104,163 @@ const VisitorChat = ({ }) } - const loadMessages = useCallback(async (sid?: number, token?: string) => { - const s = sid || sessionId - const visitorCredential = token || visitorToken || localStorage.getItem(tokenKey) || '' + const mapServerMessage = (m: { + id: number + sender_type?: string + content?: string + type?: string + sent_at?: string + seq?: number + }): Message => ({ + id: Number(m.id), + sender: m.sender_type === 'agent' ? 'agent' : 'visitor', + content: String(m.content ?? ''), + type: m.type === 'image' ? 'image' : 'text', + seq: typeof m.seq === 'number' ? m.seq : Number(m.seq || 0) || undefined, + time: m.sent_at + ? new Date(m.sent_at).toLocaleTimeString('zh-CN', { hour: '2-digit', minute: '2-digit' }) + : new Date().toLocaleTimeString('zh-CN', { hour: '2-digit', minute: '2-digit' }), + }) + + const rememberSeq = (seq?: number) => { + if (seq && seq > lastSeqRef.current) lastSeqRef.current = seq + } + + const mergeVisitorMessages = (existing: Message[], incoming: Message[]): Message[] => { + const map = new Map() + for (const item of existing) { + if (item.id > 0) map.set(item.id, item) + } + // 保留未确认的乐观气泡 + const temps = existing.filter(item => item.id < 0) + for (const item of incoming) map.set(item.id, item) + const merged = Array.from(map.values()) + // 去掉已被服务端确认的乐观气泡 + const remainingTemps = temps.filter(temp => + !incoming.some(srv => + srv.sender === 'visitor' + && srv.content === temp.content + && (srv.type || 'text') === (temp.type || 'text'), + ), + ) + return [...merged, ...remainingTemps].sort((a, b) => { + const as = a.seq || 0 + const bs = b.seq || 0 + if (as && bs && as !== bs) return as - bs + if (a.id > 0 && b.id > 0) return a.id - b.id + return a.id - b.id + }) + } + + const applySessionStatus = useCallback((status?: string, satisfactionScore?: number | null) => { + if (!status) return + if (status === 'ended' || status === 'archived') { + setSessionEnded(true) + const alreadyRated = satisfactionScore != null && Number(satisfactionScore) > 0 + if (alreadyRated) { + setRated(true) + setShowRating(false) + } else { + setRated(false) + setShowRating(true) + } + } else if (status === 'active') { + setAgentsOnline(true) + } + }, []) + + const loadMessages = useCallback(async ( + sid?: number, + token?: string, + opts?: { afterSeq?: number; full?: boolean }, + ) => { + const s = sid || sessionIdRef.current || sessionId + const visitorCredential = token || visitorTokenRef.current || visitorToken || localStorage.getItem(tokenKey) || '' if (!s || !visitorCredential) return + const full = opts?.full === true || (opts?.afterSeq == null && lastSeqRef.current <= 0) + const afterSeq = full ? 0 : (opts?.afterSeq ?? lastSeqRef.current) try { - const res = await fetch(`/api/widget/messages?session_id=${s}`, { + const qs = new URLSearchParams({ session_id: String(s) }) + if (afterSeq > 0) qs.set('after_seq', String(afterSeq)) + const res = await fetch(`/api/widget/messages?${qs}`, { headers: { 'X-Visitor-Token': visitorCredential }, }) const json = await res.json() - if (json.code === 0 && json.data && json.data.length > 0) { - const msgs: Message[] = json.data.map((m: any) => ({ - id: m.id, - sender: m.sender_type === 'agent' ? 'agent' : 'visitor', - content: m.content, - type: m.type === 'image' ? 'image' : 'text', - time: new Date(m.sent_at).toLocaleTimeString('zh-CN', { hour: '2-digit', minute: '2-digit' }), - })) - setMessages(msgs) - localStorage.setItem(msgsKey, JSON.stringify(msgs)) + if (json.code !== 0 || !json.data) return + + // 兼容旧版 data 为数组的响应 + const payload = json.data + const list: any[] = Array.isArray(payload) + ? payload + : (Array.isArray(payload.messages) ? payload.messages : []) + const maxSeq = Array.isArray(payload) + ? list.reduce((acc, m) => Math.max(acc, Number(m.seq || 0)), 0) + : Number(payload.max_seq || 0) + const hasMore = !Array.isArray(payload) && Boolean(payload.has_more) + + if (!Array.isArray(payload)) { + applySessionStatus(payload.session_status, payload.satisfaction_score) + } + + if (list.length === 0) { + if (maxSeq > 0) rememberSeq(maxSeq) + return + } + + const mapped = list.map((m: any) => mapServerMessage(m)) + setMessages(prev => { + const next = afterSeq > 0 ? mergeVisitorMessages(prev, mapped) : mapped + localStorage.setItem(msgsKeyRef.current, JSON.stringify(next.filter(m => m.id > 0))) + return next + }) + const localMax = mapped.reduce((acc, m) => Math.max(acc, m.seq || 0), 0) + rememberSeq(Math.max(maxSeq, localMax)) + + // 一次未拉完则继续增量 + if (hasMore && localMax > afterSeq) { + await loadMessages(s, visitorCredential, { afterSeq: localMax }) } } catch { /* ignore */ } - }, [sessionId, visitorToken, tokenKey, msgsKey]) + }, [sessionId, visitorToken, tokenKey, applySessionStatus]) + + const appendPushedMessage = useCallback((raw: { + id?: number + sender_type?: string + content?: string + type?: string + sent_at?: string + seq?: number + }) => { + if (!raw?.id) return false + const msg = mapServerMessage(raw as { + id: number + sender_type?: string + content?: string + type?: string + sent_at?: string + seq?: number + }) + const known = lastSeqRef.current + // seq 空洞 → 走增量拉取补齐 + if (known > 0 && msg.seq && msg.seq > known + 1) { + void loadMessagesRef.current(undefined, undefined, { afterSeq: known }) + return true + } + setMessages(prev => { + if (prev.some(item => item.id === msg.id)) return prev + const next = mergeVisitorMessages(prev, [msg]) + localStorage.setItem(msgsKeyRef.current, JSON.stringify(next.filter(m => m.id > 0))) + return next + }) + if (msg.seq) rememberSeq(msg.seq) + return true + }, []) const initSession = useCallback(async () => { if (sessionId && visitorToken) { - await loadMessages(sessionId, visitorToken) + // 本地有 seq 游标时增量补齐，否则全量 + const after = lastSeqRef.current + await loadMessages(sessionId, visitorToken, after > 0 ? { afterSeq: after } : { full: true }) return } if (initRef.current) return @@ -133,10 +282,14 @@ const VisitorChat = ({ if (data.display_name) setDisplayName(data.display_name) if (data.agent_name) setAgentName(data.agent_name) else if (data.agent_nickname) setAgentName(data.agent_nickname) - if (data.session_status === 'ended') setSessionEnded(true) + if (data.session_status === 'ended' || data.session_status === 'archived') { + setSessionEnded(true) + setShowRating(true) + } localStorage.setItem(storageKey, String(sid)) localStorage.setItem(tokenKey, token) - await loadMessages(sid, token) + lastSeqRef.current = 0 + await loadMessages(sid, token, { full: true }) } else { setSendError(json.message || '初始化会话失败') } @@ -146,59 +299,158 @@ const VisitorChat = ({ } }, [sessionId, visitorToken, loadMessages, channelKey, storageKey, tokenKey]) + useEffect(() => { + sessionIdRef.current = sessionId + visitorTokenRef.current = visitorToken + agentNameRef.current = agentName + loadMessagesRef.current = loadMessages + msgsKeyRef.current = msgsKey + }, [sessionId, visitorToken, agentName, loadMessages, msgsKey]) + useEffect(() => { if (open) initSession() }, [open, initSession]) + // 兜底轮询：按 after_seq 增量对齐 useEffect(() => { if (sessionId && visitorToken && open) { - pollRef.current = window.setInterval(() => loadMessages(), 3000) + pollRef.current = window.setInterval(() => { + const after = lastSeqRef.current + void loadMessages(undefined, undefined, after > 0 ? { afterSeq: after } : { full: true }) + }, 8000) return () => { if (pollRef.current) clearInterval(pollRef.current) } } }, [sessionId, visitorToken, open, loadMessages]) + // 访客 WS：session / token 就绪后建连；断线指数退避重连；消息直接追加 useEffect(() => { if (!sessionId || !visitorToken || !open) return - const scheme = window.location.protocol === 'https:' ? 'wss:' : 'ws:' - const socket = new WebSocket( - `${scheme}//${window.location.host}/api/widget/ws?session_id=${sessionId}`, - ['kefu-visitor-v1', visitorToken], - ) - socketRef.current = socket - socket.onmessage = (event) => { - try { - const payload = JSON.parse(event.data) - if (payload.session_id === sessionId && payload.type === 'typing') { - if (payload.data?.from && payload.data.from !== 'agent') return - setAgentTyping(true) - if (typingTimerRef.current) clearTimeout(typingTimerRef.current) - typingTimerRef.current = window.setTimeout(() => setAgentTyping(false), 1800) - return - } - if (payload.session_id === sessionId && (payload.type === 'message' || payload.type === 'session_updated')) { - if (payload.type === 'session_updated') { - if (payload.data?.status === 'ended') { - setSessionEnded(true) - setShowRating(true) - } - if (payload.data?.status === 'active') { - setAgentsOnline(true) - setAgentName(payload.data?.agent_name || agentName) - } - } - setAgentTyping(false) - loadMessages(sessionId, visitorToken) - } - } 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 sid = sessionIdRef.current + const token = visitorTokenRef.current + if (!sid || !token) return + + const scheme = window.location.protocol === 'https:' ? 'wss:' : 'ws:' + const socket = new WebSocket( + `${scheme}//${window.location.host}/api/widget/ws?session_id=${sid}`, + ['kefu-visitor-v1', token], + ) + socketRef.current = socket + + socket.onopen = () => { + attempt = 0 + // 重连后按 seq 增量 catch-up + const after = lastSeqRef.current + void loadMessagesRef.current(sid, token, after > 0 ? { afterSeq: after } : { full: true }) + } + + socket.onmessage = (event) => { + try { + const payload = JSON.parse(event.data) as { + type?: string + session_id?: number | string + data?: { + id?: number + sender_type?: string + content?: string + type?: string + sent_at?: string + from?: string + status?: string + agent_name?: string + } + } + const eventSid = Number(payload.session_id) + const currentSid = Number(sessionIdRef.current) + if (!Number.isFinite(eventSid) || eventSid !== currentSid) return + + if (payload.type === 'typing') { + if (payload.data?.from && payload.data.from !== 'agent') return + setAgentTyping(true) + if (typingTimerRef.current) clearTimeout(typingTimerRef.current) + typingTimerRef.current = window.setTimeout(() => setAgentTyping(false), 1800) + return + } + + if (payload.type === 'session_updated') { + const status = payload.data?.status + if (status === 'ended' || status === 'archived') { + setSessionEnded(true) + setShowRating(true) + setAgentTyping(false) + return + } + if (status === 'active') { + setAgentsOnline(true) + setAgentName(payload.data?.agent_name || agentNameRef.current) + } + setAgentTyping(false) + const after = lastSeqRef.current + void loadMessagesRef.current( + currentSid, + visitorTokenRef.current, + after > 0 ? { afterSeq: after } : { full: true }, + ) + return + } + + if (payload.type === 'message') { + setAgentTyping(false) + const ok = appendPushedMessage(payload.data || {}) + if (!ok) { + const after = lastSeqRef.current + void loadMessagesRef.current( + currentSid, + visitorTokenRef.current, + after > 0 ? { afterSeq: after } : { full: true }, + ) + } + } + } catch { + // ignore + } + } + + 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 = () => { + 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 (typingTimerRef.current) clearTimeout(typingTimerRef.current) } - }, [sessionId, visitorToken, open, loadMessages, agentName]) + }, [sessionId, visitorToken, open, appendPushedMessage]) useEffect(() => { chatEndRef.current?.scrollIntoView({ behavior: 'smooth' }) @@ -382,9 +634,12 @@ const VisitorChat = ({ if (json.code === 0) { setRated(true) setShowRating(false) + setSendError('') + } else { + setSendError(json.message || '评价提交失败') } } catch { - // keep modal + setSendError('评价提交失败，请稍后重试') } } @@ -520,6 +775,19 @@ const VisitorChat = ({