hejiu/common/websocket.js

/* 碰盏日记 - WebSocket 管理器
 * 单例模式，负责私信实时通信
 * 支持心跳、自动重连、消息队列
 */

const WS_HOST = 'wss://dev.wash-painting.cn/ws'

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 = ''
  }

  /**
   * 建立 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}?token=${this.token}`
    console.log('[WS] 正在连接...', url)

    this.socketTask = uni.connectSocket({
      url,
      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
  }

  /**
   * 发送消息
   * @param {string} type - 消息类型: chat | typing | read | ping
   * @param {object} data - 消息数据
   */
  send(type, data = {}) {
    const payload = JSON.stringify({ type, data, ts: Date.now() })
    if (this.isConnected && this.socketTask) {
      this.socketTask.send({ data: payload })
    } else {
      // 断线暂存队列
      if (type !== 'ping') {
        this.messageQueue.push(payload)
      }
      // 尝试重连
      if (!this.isConnecting && !this.manualClose) {
        this._scheduleReconnect()
      }
    }
  }

  /**
   * 订阅事件
   * @param {string} event - message | typing | read | ack | 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]
    }
  }

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

  /** 处理收到的消息 */
  _handleMessage(msg) {
    const { type, data } = msg
    switch (type) {
      case 'pong':
        this._onPong()
        break
      case 'chat':
        this._emit('message', data)
        break
      case 'typing':
        this._emit('typing', data)
        break
      case 'read':
        this._emit('read', data)
        break
      case 'ack':
        this._emit('ack', data)
        break
      default:
        console.log('[WS] 未知消息类型', type)
    }
  }

  /** 触发事件 */
  _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('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