nxxmdata/backend/utils/websocket.js

/**
 * WebSocket实时通信系统
 * @file websocket.js
 * @description 实现实时数据推送，替代轮询机制
 */
const socketIO = require('socket.io');
const jwt = require('jsonwebtoken');
const { User, Role } = require('../models');
const logger = require('./logger');

class WebSocketManager {
  constructor() {
    this.io = null;
    this.connectedClients = new Map(); // 存储连接的客户端信息
    this.rooms = {
      admins: 'admin_room',
      users: 'user_room',
      farms: 'farm_', // farm_1, farm_2 等
      devices: 'device_', // device_1, device_2 等
    };
  }

  /**
   * 初始化WebSocket服务器
   * @param {Object} server HTTP服务器实例
   */
  init(server) {
    this.io = socketIO(server, {
      cors: {
        origin: ["http://localhost:5300", "http://localhost:3000"],
        methods: ["GET", "POST"],
        credentials: true
      },
      pingTimeout: 60000,
      pingInterval: 25000
    });

    // 中间件：认证
    this.io.use(async (socket, next) => {
      try {
        const token = socket.handshake.auth.token || socket.handshake.headers.authorization?.split(' ')[1];
        
        if (!token) {
          return next(new Error('未提供认证令牌'));
        }

        // 验证JWT令牌
        const decoded = jwt.verify(token, process.env.JWT_SECRET || 'your_jwt_secret_key');
        
        // 获取用户信息和角色
        const user = await User.findByPk(decoded.id, {
          include: [{
            model: Role,
            as: 'role',
            attributes: ['name']
          }]
        });

        if (!user) {
          return next(new Error('用户不存在'));
        }

        // 将用户信息附加到socket
        socket.user = {
          id: user.id,
          username: user.username,
          email: user.email,
          roles: user.role ? [user.role.name] : []
        };

        next();
      } catch (error) {
        logger.error('WebSocket认证失败:', error);
        next(new Error('认证失败'));
      }
    });

    // 连接事件处理
    this.io.on('connection', (socket) => {
      this.handleConnection(socket);
    });

    logger.info('WebSocket服务器初始化完成');
    return this.io;
  }

  /**
   * 处理客户端连接
   * @param {Object} socket Socket实例
   */
  handleConnection(socket) {
    const user = socket.user;
    
    // 存储客户端信息
    this.connectedClients.set(socket.id, {
      userId: user.id,
      username: user.username,
      roles: user.roles,
      connectedAt: new Date()
    });

    // 加入相应的房间
    this.joinRooms(socket, user);

    logger.info(`用户 ${user.username} 已连接 WebSocket，连接ID: ${socket.id}`);

    // 发送连接成功消息
    socket.emit('connected', {
      message: '实时连接已建立',
      timestamp: new Date(),
      user: {
        id: user.id,
        username: user.username
      }
    });

    // 处理客户端事件
    this.setupSocketEvents(socket);

    // 断开连接处理
    socket.on('disconnect', () => {
      this.handleDisconnection(socket);
    });
  }

  /**
   * 让用户加入相应的房间
   * @param {Object} socket Socket实例
   * @param {Object} user 用户信息
   */
  joinRooms(socket, user) {
    // 所有用户加入用户房间
    socket.join(this.rooms.users);

    // 管理员加入管理员房间
    if (user.roles && user.roles.includes('admin')) {
      socket.join(this.rooms.admins);
    }

    // 可以根据业务需求加入特定的农场或设备房间
    // 这里暂时加入全局房间，后续可以根据用户权限细化
  }

  /**
   * 设置Socket事件监听
   * @param {Object} socket Socket实例
   */
  setupSocketEvents(socket) {
    // 订阅特定农场的数据
    socket.on('subscribe_farm', (farmId) => {
      socket.join(`${this.rooms.farms}${farmId}`);
      logger.info(`用户 ${socket.user.username} 订阅农场 ${farmId} 的实时数据`);
    });

    // 取消订阅农场数据
    socket.on('unsubscribe_farm', (farmId) => {
      socket.leave(`${this.rooms.farms}${farmId}`);
      logger.info(`用户 ${socket.user.username} 取消订阅农场 ${farmId} 的实时数据`);
    });

    // 订阅特定设备的数据
    socket.on('subscribe_device', (deviceId) => {
      socket.join(`${this.rooms.devices}${deviceId}`);
      logger.info(`用户 ${socket.user.username} 订阅设备 ${deviceId} 的实时数据`);
    });

    // 心跳检测
    socket.on('ping', () => {
      socket.emit('pong', { timestamp: new Date() });
    });
  }

  /**
   * 处理客户端断开连接
   * @param {Object} socket Socket实例
   */
  handleDisconnection(socket) {
    const clientInfo = this.connectedClients.get(socket.id);
    
    if (clientInfo) {
      logger.info(`用户 ${clientInfo.username} 断开 WebSocket 连接，连接ID: ${socket.id}`);
      this.connectedClients.delete(socket.id);
    }
  }

  /**
   * 广播设备状态更新
   * @param {Object} deviceData 设备数据
   */
  broadcastDeviceUpdate(deviceData) {
    if (!this.io) return;

    // 向所有用户广播设备更新
    this.io.to(this.rooms.users).emit('device_update', {
      type: 'device_status',
      data: deviceData,
      timestamp: new Date()
    });

    // 向特定设备的订阅者发送更新
    this.io.to(`${this.rooms.devices}${deviceData.id}`).emit('device_detail_update', {
      type: 'device_detail',
      data: deviceData,
      timestamp: new Date()
    });

    logger.info(`设备状态更新已广播: 设备ID ${deviceData.id}`);
  }

  /**
   * 广播预警信息
   * @param {Object} alertData 预警数据
   */
  broadcastAlert(alertData) {
    if (!this.io) return;

    // 向管理员发送紧急预警
    if (alertData.level === 'critical' || alertData.level === 'high') {
      this.io.to(this.rooms.admins).emit('urgent_alert', {
        type: 'urgent_alert',
        data: alertData,
        timestamp: new Date()
      });
    }

    // 向所有用户广播预警
    this.io.to(this.rooms.users).emit('alert_update', {
      type: 'new_alert',
      data: alertData,
      timestamp: new Date()
    });

    // 向特定农场的订阅者发送预警
    if (alertData.farm_id) {
      this.io.to(`${this.rooms.farms}${alertData.farm_id}`).emit('farm_alert', {
        type: 'farm_alert',
        data: alertData,
        timestamp: new Date()
      });
    }

    logger.info(`预警信息已广播: 预警ID ${alertData.id}, 级别: ${alertData.level}`);
  }

  /**
   * 广播动物健康数据更新
   * @param {Object} animalData 动物数据
   */
  broadcastAnimalUpdate(animalData) {
    if (!this.io) return;

    this.io.to(this.rooms.users).emit('animal_update', {
      type: 'animal_health',
      data: animalData,
      timestamp: new Date()
    });

    // 向特定农场的订阅者发送动物更新
    if (animalData.farm_id) {
      this.io.to(`${this.rooms.farms}${animalData.farm_id}`).emit('farm_animal_update', {
        type: 'farm_animal',
        data: animalData,
        timestamp: new Date()
      });
    }

    logger.info(`动物健康数据更新已广播: 动物ID ${animalData.id}`);
  }

  /**
   * 广播系统统计数据更新
   * @param {Object} statsData 统计数据
   */
  broadcastStatsUpdate(statsData) {
    if (!this.io) return;

    this.io.to(this.rooms.users).emit('stats_update', {
      type: 'system_stats',
      data: statsData,
      timestamp: new Date()
    });

    logger.info('系统统计数据更新已广播');
  }

  /**
   * 发送性能监控数据
   * @param {Object} performanceData 性能数据
   */
  broadcastPerformanceUpdate(performanceData) {
    if (!this.io) return;

    // 只向管理员发送性能数据
    this.io.to(this.rooms.admins).emit('performance_update', {
      type: 'system_performance',
      data: performanceData,
      timestamp: new Date()
    });

    logger.info('性能监控数据已发送给管理员');
  }

  /**
   * 获取连接统计信息
   * @returns {Object} 连接统计
   */
  getConnectionStats() {
    return {
      totalConnections: this.connectedClients.size,
      connectedUsers: Array.from(this.connectedClients.values()).map(client => ({
        userId: client.userId,
        username: client.username,
        connectedAt: client.connectedAt
      }))
    };
  }

  /**
   * 向特定用户发送消息
   * @param {number} userId 用户ID
   * @param {string} event 事件名称
   * @param {Object} data 数据
   */
  sendToUser(userId, event, data) {
    if (!this.io) return;

    for (const [socketId, clientInfo] of this.connectedClients.entries()) {
      if (clientInfo.userId === userId) {
        this.io.to(socketId).emit(event, data);
        logger.info(`消息已发送给用户 ${clientInfo.username}: ${event}`);
      }
    }
  }
}

// 创建单例实例
const webSocketManager = new WebSocketManager();

module.exports = webSocketManager;