hejiu/common/websocket.js

/* 碰盏日记 - WebSocket 管理器
 * 单例模式，负责私信实时通信
 * 协议格式与 HaveADrink SDK ChatWebSocket 一致：{ mesgType, data }
 * 支持心跳、自动重连、消息队列
 */

const WS_HOST = 'wss://dev.wash-painting.cn/api/have_a_drink/v1/chat/chat'

// 客户端消息类型（与 SDK ClientMesgType 枚举值一致）
const ClientMesgType = {
  Chat: 'chat',
  Typing: 'typing',
  Read: 'read',
  Ping: 'ping'
}

// 服务端消息类型（与 SDK ServerMesgType 枚举值一致）
const ServerMesgType = {
  Chat: 'chat',
  Typing: 'typing',
  Read: 'read',
  Ack: 'ack',
  Pong: 'pong',
  Connected: 'connected',
  Kicked: 'kicked'
}

class WebSocketManager {
  constructor() {
    this.socketTask = null
    this.isConnected = false
    this.isConnecting = false
    this.listeners = {}       // 事件订阅 { event: [callbacks] }
    this.messageQueue = []    // 断线期间暂存消息
    this.heartbeatTimer = null
    this.heartbeatTimeout = null
    this.reconnectTimer = null
    this.reconnectAttempts = 0
    this.maxReconnectDelay = 30000
    this.heartbeatInterval = 30000   // 30s 发一次 ping
    this.heartbeatWait = 60000       // 60s 无 pong 则重连
    this.manualClose = false
    this.token = ''
    this.userId = ''          // 连接成功后服务端返回的用户ID
  }

  /**
   * 建立 WebSocket 连接
   * @param {string} token - 用户认证 token
   */
  connect(token) {
    if (this.isConnected || this.isConnecting) return
    this.token = token || uni.getStorageSync('auth_token') || ''
    if (!this.token) {
      console.warn('[WS] 无 token，跳过连接')
      return
    }
    this.manualClose = false
    this.isConnecting = true

    const url = WS_HOST
    console.log('[WS] 正在连接...')

    this.socketTask = uni.connectSocket({
      url,
      header: {
        'Authorization': `Bearer ${this.token}`
      },
      complete: () => {}
    })

    this.socketTask.onOpen(() => {
      console.log('[WS] 连接成功')
      this.isConnected = true
      this.isConnecting = false
      this.reconnectAttempts = 0
      this._startHeartbeat()
      this._flushQueue()
      this._emit('connect', {})
    })

    this.socketTask.onMessage((res) => {
      try {
        const msg = JSON.parse(res.data)
        this._handleMessage(msg)
      } catch (e) {
        console.warn('[WS] 消息解析失败', res.data)
      }
    })

    this.socketTask.onClose(() => {
      console.log('[WS] 连接关闭')
      this.isConnected = false
      this.isConnecting = false
      this._stopHeartbeat()
      this._emit('disconnect', {})
      if (!this.manualClose) {
        this._scheduleReconnect()
      }
    })

    this.socketTask.onError((err) => {
      console.warn('[WS] 连接错误', err)
      this.isConnected = false
      this.isConnecting = false
    })
  }

  /** 主动断开连接 */
  disconnect() {
    this.manualClose = true
    this._stopHeartbeat()
    this._clearReconnect()
    if (this.socketTask) {
      this.socketTask.close()
      this.socketTask = null
    }
    this.isConnected = false
  }

  /**
   * 发送消息（与 SDK writeClient* 方法格式一致）
   * @param {string} mesgType - 消息类型: chat | typing | read | ping
   * @param {object} data - 消息数据
   */
  send(mesgType, data = {}) {
    const payload = JSON.stringify({ mesgType, data })
    if (this.isConnected && this.socketTask) {
      this.socketTask.send({ data: payload })
    } else {
      // 断线暂存队列（ping 不暂存）
      if (mesgType !== ClientMesgType.Ping) {
        this.messageQueue.push(payload)
      }
      // 尝试重连
      if (!this.isConnecting && !this.manualClose) {
        this._scheduleReconnect()
      }
    }
  }

  /**
   * 订阅事件
   * @param {string} event - message | typing | read | ack | connected | kicked | connect | disconnect
   * @param {function} callback
   */
  on(event, callback) {
    if (!this.listeners[event]) {
      this.listeners[event] = []
    }
    this.listeners[event].push(callback)
  }

  /**
   * 取消订阅
   * @param {string} event
   * @param {function} callback
   */
  off(event, callback) {
    if (!this.listeners[event]) return
    if (callback) {
      this.listeners[event] = this.listeners[event].filter(cb => cb !== callback)
    } else {
      delete this.listeners[event]
    }
  }

  // ========== 内部方法 ==========

  /** 处理收到的消息（SDK ServerMesgType 协议） */
  _handleMessage(msg) {
    const { mesgType, data } = msg
    switch (mesgType) {
      case ServerMesgType.Chat:
        this._emit('message', data)
        break
      case ServerMesgType.Typing:
        this._emit('typing', data)
        break
      case ServerMesgType.Read:
        this._emit('read', data)
        break
      case ServerMesgType.Ack:
        this._emit('ack', data)
        break
      case ServerMesgType.Pong:
        this._onPong()
        break
      case ServerMesgType.Connected:
        // 服务端验证 token 成功后推送，携带 userId
        this.userId = data && data.userId ? data.userId : ''
        console.log('[WS] 服务端确认连接, userId:', this.userId)
        this._emit('connected', data)
        break
      case ServerMesgType.Kicked:
        console.warn('[WS] 被踢下线')
        this._emit('kicked', data)
        this.disconnect()
        break
      default:
        console.log('[WS] 未知消息类型', mesgType)
    }
  }

  /** 触发事件 */
  _emit(event, data) {
    const cbs = this.listeners[event]
    if (cbs) {
      cbs.forEach(cb => {
        try { cb(data) } catch (e) { console.warn('[WS] 事件回调异常', e) }
      })
    }
  }

  /** 启动心跳 */
  _startHeartbeat() {
    this._stopHeartbeat()
    this.heartbeatTimer = setInterval(() => {
      this.send(ClientMesgType.Ping, {})
      // 设置超时检测
      this.heartbeatTimeout = setTimeout(() => {
        console.warn('[WS] 心跳超时，触发重连')
        if (this.socketTask) {
          this.socketTask.close()
        }
      }, this.heartbeatWait)
    }, this.heartbeatInterval)
  }

  /** 收到 pong，清除超时 */
  _onPong() {
    if (this.heartbeatTimeout) {
      clearTimeout(this.heartbeatTimeout)
      this.heartbeatTimeout = null
    }
  }

  /** 停止心跳 */
  _stopHeartbeat() {
    if (this.heartbeatTimer) {
      clearInterval(this.heartbeatTimer)
      this.heartbeatTimer = null
    }
    if (this.heartbeatTimeout) {
      clearTimeout(this.heartbeatTimeout)
      this.heartbeatTimeout = null
    }
  }

  /** 指数退避重连 */
  _scheduleReconnect() {
    if (this.manualClose || this.reconnectTimer) return
    const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), this.maxReconnectDelay)
    this.reconnectAttempts++
    console.log(`[WS] ${delay}ms 后尝试第 ${this.reconnectAttempts} 次重连`)
    this.reconnectTimer = setTimeout(() => {
      this.reconnectTimer = null
      this.isConnecting = false
      this.connect(this.token)
    }, delay)
  }

  /** 清除重连计时器 */
  _clearReconnect() {
    if (this.reconnectTimer) {
      clearTimeout(this.reconnectTimer)
      this.reconnectTimer = null
    }
    this.reconnectAttempts = 0
  }

  /** 重连成功后重发队列消息 */
  _flushQueue() {
    if (!this.messageQueue.length) return
    console.log(`[WS] 重发 ${this.messageQueue.length} 条暂存消息`)
    const queue = [...this.messageQueue]
    this.messageQueue = []
    queue.forEach(payload => {
      if (this.socketTask && this.isConnected) {
        this.socketTask.send({ data: payload })
      }
    })
  }
}

// 导出全局单例
const wsManager = new WebSocketManager()
export default wsManager