feat：0820需求

2026-08-21 14:10:43 +08:00
parent 619c0cacc5
commit e8d4fd6cd4
31 changed files with 9112 additions and 3658 deletions
@@ -0,0 +1,97 @@
+import { getStorageFun, TOKEN_NAME, USER_DATA } from '@/utils/auth.js'
+import { getCurrentMember } from './api.js'
+import { KefuSocket } from './kefu-socket.js'
+
+export const KEFU_REALTIME_EVENT = 'kefu-realtime-event'
+export const KEFU_REALTIME_STATUS = 'kefu-realtime-status'
+
+let socket = null
+let connectionParams = null
+let startingPromise = null
+
+function storedMember() {
+  const value = getStorageFun(USER_DATA)
+  if (!value) return null
+  if (typeof value === 'object') return value
+  try {
+    return JSON.parse(value)
+  } catch (error) {
+    return null
+  }
+}
+
+async function resolveMemberId(explicitUserId) {
+  if (explicitUserId) return explicitUserId
+  const cached = storedMember()
+  if (cached && (cached.id || cached.userId)) return cached.id || cached.userId
+  const member = await getCurrentMember()
+  return member && (member.id || member.userId)
+}
+
+function createSocket() {
+  const instance = new KefuSocket({
+    onStatus(status) {
+      uni.$emit(KEFU_REALTIME_STATUS, status)
+    },
+    onEvent(event) {
+      uni.$emit(KEFU_REALTIME_EVENT, event)
+    },
+  })
+  socket = instance
+  instance.connect(connectionParams)
+  return instance
+}
+
+/**
+ * Start the foreground customer-service connection. Calls are idempotent so
+ * several pages can consume the same stream without opening duplicate sockets.
+ */
+export function startKefuRealtime(options = {}) {
+  if (!getStorageFun(TOKEN_NAME)) {
+    resetKefuRealtime()
+    return Promise.resolve(null)
+  }
+  if (startingPromise) return startingPromise
+
+  startingPromise = resolveMemberId(options.userId)
+    .then((userId) => {
+      if (!userId) return null
+      const nextParams = {
+        userId,
+        mchId: options.mchId || (connectionParams && connectionParams.mchId) || 1002,
+      }
+      const sameMember = connectionParams && String(connectionParams.userId) === String(userId)
+      connectionParams = nextParams
+      if (socket && sameMember) {
+        socket.reconnectNow()
+        return socket
+      }
+      if (socket) socket.close()
+      socket = null
+      return createSocket()
+    })
+    .catch(() => null)
+    .finally(() => {
+      startingPromise = null
+    })
+  return startingPromise
+}
+
+export function suspendKefuRealtime() {
+  if (socket) socket.close()
+  socket = null
+  uni.$emit(KEFU_REALTIME_STATUS, 'closed')
+}
+
+export function resumeKefuRealtime() {
+  if (!getStorageFun(TOKEN_NAME)) {
+    resetKefuRealtime()
+    return Promise.resolve(null)
+  }
+  return startKefuRealtime(connectionParams || {})
+}
+
+export function resetKefuRealtime() {
+  suspendKefuRealtime()
+  connectionParams = null
+}
@@ -1,6 +1,39 @@
 import { KEFU_BASE_URL } from '@/utils/config.js'

 const RECONNECT_DELAYS = [2000, 4000, 8000, 15000, 30000]
+const HEARTBEAT_INTERVAL = 20000
+const HEARTBEAT_TIMEOUT = 45000
+const CONNECT_TIMEOUT = 10000
+
+function decodeSocketPayload(raw) {
+  if (typeof raw === 'string') return raw
+
+  let bytes = null
+  if (typeof ArrayBuffer !== 'undefined' && raw instanceof ArrayBuffer) {
+    bytes = new Uint8Array(raw)
+  } else if (typeof ArrayBuffer !== 'undefined' && ArrayBuffer.isView && ArrayBuffer.isView(raw)) {
+    bytes = new Uint8Array(raw.buffer, raw.byteOffset, raw.byteLength)
+  }
+  if (!bytes) return raw
+
+  if (typeof TextDecoder !== 'undefined') {
+    try {
+      return new TextDecoder('utf-8').decode(bytes)
+    } catch (error) {
+      // Fall through to the decoder supported by older App WebViews.
+    }
+  }
+
+  let encoded = ''
+  for (let index = 0; index < bytes.length; index += 1) {
+    encoded += `%${bytes[index].toString(16).padStart(2, '0')}`
+  }
+  try {
+    return decodeURIComponent(encoded)
+  } catch (error) {
+    return Array.from(bytes).map(code => String.fromCharCode(code)).join('')
+  }
+}

 function getSocketBaseUrl() {
  const override = uni.getStorageSync('kefu-socket-base-url')
@@ -12,6 +45,14 @@ function getSocketBaseUrl() {
  return baseUrl.replace(/^http:/, 'ws:').replace(/^https:/, 'wss:')
 }

+function shouldUseNativeSocket() {
+  let enabled = false
+  // #ifdef H5
+  enabled = typeof window !== 'undefined' && typeof window.WebSocket === 'function'
+  // #endif
+  return enabled
+}
+
 export class KefuSocket {
  constructor(options = {}) {
    this.options = options
@@ -22,6 +63,8 @@ export class KefuSocket {
    this.reconnectCount = 0
    this.reconnectTimer = null
    this.heartbeatTimer = null
+    this.connectTimer = null
+    this.lastServerActivityAt = 0
    this.networkListener = status => this.handleNetworkChange(status)
    this.browserOnlineListener = () => this.handleNetworkChange({ isConnected: true })
    this.browserOfflineListener = () => this.handleNetworkChange({ isConnected: false })
@@ -41,30 +84,49 @@ export class KefuSocket {
      conversationId ? `conversationId=${encodeURIComponent(conversationId)}` : ''
    ].filter(Boolean).join('&')

-    const task = uni.connectSocket({
-      url: `${getSocketBaseUrl()}/ws/kefu?${query}`,
-      complete: () => {}
-    })
+    const url = `${getSocketBaseUrl()}/ws/kefu?${query}`
+    // H5 使用浏览器原生 WebSocket。部分 uni-app H5 版本的 SocketTask
+    // 不会稳定触发 onOpen，页面会误判为断线并永久停留在轮询状态。
+    const useNativeSocket = shouldUseNativeSocket()
+    const task = useNativeSocket
+      ? new window.WebSocket(url)
+      : uni.connectSocket({ url, complete: () => {} })
    this.socketTask = task

-    task.onOpen(() => {
+    const onOpen = () => {
      if (this.socketTask !== task) return
+      this.stopConnectTimeout()
      this.reconnectCount = 0
+      this.lastServerActivityAt = Date.now()
      this.options.onStatus && this.options.onStatus('online')
      this.startHeartbeat(task)
-    })
-    task.onMessage(event => {
+    }
+    const onMessage = event => {
      if (this.socketTask === task) this.handleMessage(event.data)
-    })
-    task.onError(() => this.handleDisconnect(task))
-    task.onClose(() => this.handleDisconnect(task))
+    }
+    const onDisconnect = () => this.handleDisconnect(task)
+
+    if (useNativeSocket) {
+      task.addEventListener('open', onOpen)
+      task.addEventListener('message', onMessage)
+      task.addEventListener('error', onDisconnect)
+      task.addEventListener('close', onDisconnect)
+    } else {
+      task.onOpen(onOpen)
+      task.onMessage(onMessage)
+      task.onError(onDisconnect)
+      task.onClose(onDisconnect)
+    }
+    this.startConnectTimeout(task)
  }

  handleMessage(raw) {
-    let event = raw
-    if (typeof raw === 'string') {
+    this.lastServerActivityAt = Date.now()
+    const payload = decodeSocketPayload(raw)
+    let event = payload
+    if (typeof payload === 'string') {
      try {
-        event = JSON.parse(raw)
+        event = JSON.parse(payload)
      } catch (error) {
        return
      }
@@ -77,12 +139,59 @@ export class KefuSocket {
    this.stopHeartbeat()
    this.heartbeatTimer = setInterval(() => {
      if (this.socketTask !== task) return
-      task.send({ data: 'ping', fail: () => this.handleDisconnect(task) })
-    }, 25000)
+      if (this.lastServerActivityAt && Date.now() - this.lastServerActivityAt > HEARTBEAT_TIMEOUT) {
+        this.forceReconnect('heartbeat timeout')
+        return
+      }
+      this.sendSocket(task, 'ping')
+    }, HEARTBEAT_INTERVAL)
+  }
+
+  sendSocket(task, data) {
+    try {
+      if (typeof window !== 'undefined' && task instanceof window.WebSocket) {
+        if (task.readyState === window.WebSocket.OPEN) task.send(data)
+        else this.handleDisconnect(task)
+        return
+      }
+      task.send({ data, fail: () => this.handleDisconnect(task) })
+    } catch (error) {
+      this.handleDisconnect(task)
+    }
+  }
+
+  closeSocket(task, reason) {
+    if (!task) return
+    try {
+      if (typeof window !== 'undefined' && task instanceof window.WebSocket) {
+        task.close(1000, reason)
+      } else {
+        task.close({ code: 1000, reason })
+      }
+    } catch (error) {
+      // App/native runtime may already have released the socket.
+    }
+  }
+
+  startConnectTimeout(task) {
+    this.stopConnectTimeout()
+    this.connectTimer = setTimeout(() => {
+      if (this.socketTask !== task) return
+      this.socketTask = null
+      this.closeSocket(task, 'connect timeout')
+      this.options.onStatus && this.options.onStatus('offline')
+      this.scheduleReconnect()
+    }, CONNECT_TIMEOUT)
+  }
+
+  stopConnectTimeout() {
+    if (this.connectTimer) clearTimeout(this.connectTimer)
+    this.connectTimer = null
  }

  handleDisconnect(task) {
    if (task && this.socketTask !== task) return
+    this.stopConnectTimeout()
    this.stopHeartbeat()
    this.socketTask = null
    if (this.manualClose) return
@@ -101,27 +210,41 @@ export class KefuSocket {
    }, delay)
  }

-  reconnectNow() {
+  reconnectNow(force = false) {
+    if (force) {
+      this.forceReconnect('app resumed')
+      return
+    }
    if (this.manualClose || this.socketTask || !this.connectionParams || !this.networkOnline) return
    if (this.reconnectTimer) clearTimeout(this.reconnectTimer)
    this.reconnectTimer = null
    this.connect(this.connectionParams)
  }

+  forceReconnect(reason = 'connection refresh') {
+    if (this.manualClose || !this.connectionParams || !this.networkOnline) return
+    if (this.reconnectTimer) clearTimeout(this.reconnectTimer)
+    this.reconnectTimer = null
+    this.stopConnectTimeout()
+    this.stopHeartbeat()
+
+    const task = this.socketTask
+    this.socketTask = null
+    this.closeSocket(task, reason)
+
+    this.options.onStatus && this.options.onStatus('connecting')
+    setTimeout(() => this.connect(this.connectionParams), 0)
+  }
+
  handleNetworkChange(status = {}) {
    this.networkOnline = status.isConnected !== false
    if (!this.networkOnline) {
      this.options.onStatus && this.options.onStatus('offline')
+      this.stopConnectTimeout()
      this.stopHeartbeat()
      const task = this.socketTask
      this.socketTask = null
-      if (task) {
-        try {
-          task.close({ code: 1000, reason: 'network offline' })
-        } catch (error) {
-          // The platform may have disposed the socket before the offline event.
-        }
-      }
+      this.closeSocket(task, 'network offline')
      return
    }
    this.reconnectNow()
@@ -155,6 +278,7 @@ export class KefuSocket {

  close() {
    this.manualClose = true
+    this.stopConnectTimeout()
    this.stopHeartbeat()
    if (this.reconnectTimer) clearTimeout(this.reconnectTimer)
    this.reconnectTimer = null
@@ -162,7 +286,7 @@ export class KefuSocket {
    this.unbindNetworkListeners()
    const task = this.socketTask
    this.socketTask = null
-    if (task) task.close({ code: 1000, reason: 'page hidden' })
+    this.closeSocket(task, 'page hidden')
    this.options.onStatus && this.options.onStatus('closed')
  }
 }