frost-mqtt-client/app/src/main/mqtt-manager.js

import { app } from 'electron';
import { readFileSync, writeFileSync, existsSync, mkdirSync } from 'fs';
import { join } from 'path';

import mqtt from 'mqtt';

// ==================== 类型定义 ====================

/**
 * 从公共类型定义文件引入 MQTT 相关类型
 * @typedef {import('@types/jsdoc').MqttQoS} MqttQoS
 * @typedef {import('@types/jsdoc').MqttServerStatus} MqttServerStatus
 * @typedef {import('@types/jsdoc').MqttTopic} MqttTopic
 * @typedef {import('@types/jsdoc').TopicConfig} TopicConfig
 * @typedef {import('@types/jsdoc').ServerConfig} ServerConfig
 * @typedef {import('@types/jsdoc').MqttServer} MqttServer
 * @typedef {import('@types/jsdoc').PublishOptions} PublishOptions
 * @typedef {import('@types/jsdoc').MqttMessage} MqttMessage
 * @typedef {import('@types/jsdoc').PublishRecord} PublishRecord
 */

// ==================== 类定义 ====================

/**
 * MQTT 管理器
 * 负责 MQTT 服务器配置持久化、连接管理、主题订阅/发布、消息存储等核心逻辑
 */
class MqttManager {

  constructor() {
    this.clients = new Map(); // serverId -> mqtt client
    this.servers = new Map(); // serverId -> server config
    this.messages = new Map(); // serverId -> message[]
    this.publishHistory = new Map(); // serverId -> publishHistory[]
    this.listeners = new Set(); // IPC callback functions
    this.msgIdCounter = 0; // 用于生成唯一消息 id
    this.dataPath = join(app.getPath('userData'), 'frost-mqtt-data.json');
    this.quickSendList = [];
    this.loadData();
  }

  /**
   * 生成唯一消息 id
   * @returns {string} 消息唯一标识
   */
  nextMsgId() {
    this.msgIdCounter = (this.msgIdCounter + 1) % Number.MAX_SAFE_INTEGER;
    return 'msg_' + Date.now() + '_' + this.msgIdCounter;
  }

  // ==================== 数据持久化 ====================

  /**
   * 获取用户数据目录，若不存在则创建
   * @returns {string} 用户数据目录路径
   */
  getDataDir() {

    let dir = app.getPath('userData');

    if (!existsSync(dir)) {
      mkdirSync(dir, { recursive: true });
    }

    return dir;

  }

  /**
   * 从本地文件加载服务器、消息和发布历史数据
   * 文件不存在或解析失败时初始化默认服务器
   */
  loadData() {
    try {
      if (existsSync(this.dataPath)) {

        let raw = readFileSync(this.dataPath, 'utf-8');
        let data = JSON.parse(raw);

        if (data.servers) {
          data.servers.forEach((s) => {
            s.status = 'disconnected';
            s.topics = (s.topics || []).map((t) => ({
              ...t,
              id: t.id || 'topic_' + Date.now() + '_' + Math.random().toString(36).slice(2, 6),
              comment: t.comment ?? '',
              subscribed: false,
            }));
            this.servers.set(s.id, s);
          });
        }

        if (data.messages) {
          Object.entries(data.messages).forEach(([id, msgs]) => {
            this.messages.set(id, msgs);
          });
        }

        if (data.publishHistory) {
          Object.entries(data.publishHistory).forEach(([id, history]) => {
            this.publishHistory.set(id, history);
          });
        }

        if (Array.isArray(data.quickSendList)) {
          this.quickSendList = data.quickSendList.filter((item) => {
            return item && typeof item.id === 'string' && typeof item.topic === 'string';
          });
        }

      } else {
        this.initDefaultServers();
      }
    } catch (e) {
      console.error('加载数据失败:', e.message);
      this.initDefaultServers();
    }
  }

  /**
   * 将当前服务器、消息和发布历史保存到本地文件
   */
  saveData() {
    try {

      let data = {
        servers: Array.from(this.servers.values()).map((s) => ({
          ...s,
          status: 'disconnected',
          topics: s.topics.map((t) => ({ ...t, subscribed: false })),
        })),
        messages: Object.fromEntries(this.messages),
        publishHistory: Object.fromEntries(this.publishHistory),
        quickSendList: this.quickSendList,
      };

      writeFileSync(this.dataPath, JSON.stringify(data, null, 2), 'utf-8');

    } catch (e) {
      console.error('保存数据失败:', e.message);
    }
  }

  /**
   * 初始化默认示例服务器
   */
  initDefaultServers() {

    let defaults = [
      {
        id: 'example_server',
        name: 'EMQX 公共测试',
        host: 'broker.emqx.io',
        port: 1883,
        protocol: 'mqtt',
        clientId: 'frost_client_' + Math.random().toString(36).slice(2, 8),
        username: '',
        password: '',
        keepAlive: 60,
        cleanSession: true,
        connectTimeout: 10000,
        reconnect: true,
        reconnectInterval: 5000,
        status: 'disconnected',
        topics: [
          { id: 'topic_default_001', topic: 'sensor/+/temperature', qos: 1, comment: '', subscribed: false },
          { id: 'topic_default_002', topic: 'device/status', qos: 2, comment: '', subscribed: false },
        ],
      },
    ];

    defaults.forEach((s) => {
      this.servers.set(s.id, s);
      this.messages.set(s.id, []);
      this.publishHistory.set(s.id, []);
    });

    this.saveData();

  }

  // ==================== 监听器 ====================

  /**
   * 添加 MQTT 事件监听器
   * @param {Function} callback - 回调函数，接收 (event, data) 参数
   */
  addListener(callback) {
    this.listeners.add(callback);
  }

  /**
   * 移除 MQTT 事件监听器
   * @param {Function} callback - 要移除的回调函数
   */
  removeListener(callback) {
    this.listeners.delete(callback);
  }

  /**
   * 通知所有监听器
   * @param {string} event - 事件名称
   * @param {*} data - 事件数据
   */
  notify(event, data) {
    this.listeners.forEach((cb) => {
      try {
        cb(event, data);
      } catch (e) {
        console.error('通知失败:', e);
      }
    });
  }

  // ==================== 服务器管理 ====================

  /**
   * 获取所有服务器列表
   * @returns {MqttServer[]} 服务器配置数组
   */
  getServers() {
    return Array.from(this.servers.values()).map((s) => ({
      ...s,
      password: s.password || '',
    }));
  }

  /**
   * 根据 id 获取服务器配置
   * @param {string} id - 服务器 id
   * @returns {MqttServer | null} 服务器配置或 null
   */
  getServer(id) {
    return this.servers.get(id) || null;
  }

  /**
   * 获取服务器配置字段 schema
   * @returns {Record<string, Function>} 字段名到处理函数的映射
   */
  getServerConfigSchema() {
    return {
      name: (c, e) => c.name ?? e?.name,
      host: (c, e) => c.host ?? e?.host,
      port: (c, e) => c.port || e?.port || 1883,
      protocol: (c, e) => c.protocol || e?.protocol || 'mqtt',
      clientId: (c, e) => c.clientId || e?.clientId || 'frost_' + Math.random().toString(36).slice(2, 10),
      username: (c, e) => c.username ?? e?.username ?? '',
      password: (c, e, opts) => c.password ?? (opts.keepExistingPassword ? e?.password : '') ?? '',
      keepAlive: (c, e) => c.keepAlive || e?.keepAlive || 60,
      cleanSession: (c) => c.cleanSession !== false,
      connectTimeout: (c, e) => c.connectTimeout ?? e?.connectTimeout ?? 10000,
      reconnect: (c) => c.reconnect !== false,
      reconnectInterval: (c, e) => c.reconnectInterval ?? e?.reconnectInterval ?? 5000,
    };
  }

  /**
   * 根据配置构建服务器字段
   * @param {ServerConfig} config - 新配置
   * @param {MqttServer | null} existing - 已有服务器配置
   * @param {{ keepExistingPassword?: boolean }} options - 可选项
   * @returns {ServerConfig} 处理后的字段对象
   */
  buildServerFields(config, existing = null, options = {}) {
    let schema = this.getServerConfigSchema();
    let fields = {};
    for (let key of Object.keys(schema)) {
      fields[key] = schema[key](config, existing, options);
    }
    return fields;
  }

  /**
   * 构建主题对象
   * @param {TopicConfig} t - 原始主题数据
   * @returns {MqttTopic} 标准化后的主题对象
   */
  buildTopic(t) {
    return {
      id: t.id || 'topic_' + Date.now() + '_' + Math.random().toString(36).slice(2, 6),
      topic: t.topic,
      qos: t.qos ?? 1,
      comment: t.comment ?? '',
      subscribed: false,
    };
  }

  /**
   * 添加服务器
   * @param {ServerConfig} config - 服务器配置
   * @returns {MqttServer} 创建后的服务器对象
   */
  addServer(config) {

    let id = 'srv_' + Date.now() + '_' + Math.random().toString(36).slice(2, 6);
    let server = {
      id,
      ...this.buildServerFields(config),
      status: 'disconnected',
      topics: [],
    };

    this.servers.set(id, server);
    this.messages.set(id, []);
    this.publishHistory.set(id, []);
    this.saveData();

    return server;

  }

  /**
   * 更新服务器配置
   * @param {string} id - 服务器 id
   * @param {ServerConfig} config - 新的服务器配置
   * @returns {MqttServer | null} 更新后的服务器对象或 null
   */
  updateServer(id, config) {

    let server = this.servers.get(id);

    if (!server) {
      return null;
    }

    Object.assign(server, this.buildServerFields(config, server, { keepExistingPassword: true }));

    this.saveData();

    return server;

  }

  /**
   * 删除服务器，并断开其连接
   * @param {string} id - 服务器 id
   */
  deleteServer(id) {
    // 先断开连接
    this.disconnect(id);
    this.servers.delete(id);
    this.messages.delete(id);
    this.publishHistory.delete(id);
    this.saveData();
  }

  /**
   * 导出所有服务器配置（不含运行时状态）
   * @returns {ServerConfig[]} 服务器配置数组
   */
  exportServers() {
    let schemaKeys = Object.keys(this.getServerConfigSchema());
    return Array.from(this.servers.values()).map((s) => {
      let exported = {};
      for (let key of schemaKeys) {
        exported[key] = s[key];
      }
      exported.topics = (s.topics || []).map((t) => this.buildTopicExport(t));
      return exported;
    });
  }

  /**
   * 构建导出用的主题对象
   * @param {MqttTopic} t - 主题对象
   * @returns {TopicConfig} 导出主题对象
   */
  buildTopicExport(t) {
    return {
      id: t.id,
      topic: t.topic,
      qos: t.qos,
      comment: t.comment ?? '',
    };
  }

  /**
   * 导入服务器配置列表，会覆盖现有配置
   * @param {ServerConfig[]} serverList - 服务器配置数组
   * @returns {MqttServer[]} 导入后的服务器列表
   */
  importServers(serverList) {
    if (!Array.isArray(serverList)) {
      throw new Error('导入格式错误：必须是服务器配置数组');
    }

    // 断开所有现有连接
    this.clients.forEach((client, id) => {
      this.disconnect(id);
    });
    this.clients.clear();

    this.servers.clear();
    this.messages.clear();
    this.publishHistory.clear();

    serverList.forEach((config) => {
      if (!config.name || !config.host) {
        return;
      }

      let id = 'srv_' + Date.now() + '_' + Math.random().toString(36).slice(2, 6);
      let server = {
        id,
        ...this.buildServerFields(config),
        status: 'disconnected',
        topics: (config.topics || []).map((t) => this.buildTopic(t)),
      };

      this.servers.set(id, server);
      this.messages.set(id, []);
      this.publishHistory.set(id, []);
    });

    this.saveData();
    return this.getServers();
  }

  // ==================== 快速发送 ====================

  /**
   * 获取快速发送列表
   * @returns {QuickSendItem[]} 快速发送项数组
   */
  getQuickSendList() {
    return this.quickSendList;
  }

  /**
   * 保存快速发送列表
   * @param {QuickSendItem[]} list - 快速发送项数组
   */
  saveQuickSendList(list) {
    if (!Array.isArray(list)) {
      return;
    }
    this.quickSendList = list.filter((item) => {
      return item && typeof item.id === 'string' && typeof item.topic === 'string';
    });
    this.saveData();
  }

  // ==================== 连接管理 ====================

  /**
   * 连接指定服务器
   * @param {string} id - 服务器 id
   * @returns {boolean} 是否成功发起连接
   */
  connect(id) {

    let server = this.servers.get(id);

    if (!server) {
      return false;
    }

    // 如果已连接，先断开
    if (this.clients.has(id)) {
      this.disconnect(id);
    }

    server.status = 'connecting';

    this.notify('status-change', { id, status: 'connecting' });

    let protocol = server.protocol || 'mqtt';
    let url = `${protocol}://${server.host}:${server.port}`;

    let options = {
      clientId: server.clientId,
      username: server.username || undefined,
      password: server.password || undefined,
      keepalive: server.keepAlive,
      clean: server.cleanSession,
      reconnectPeriod: 0,
      connectTimeout: server.connectTimeout,
    };

    try {

      let client = mqtt.connect(url, options);

      client.on('connect', () => {
        server.status = 'connected';
        this.notify('status-change', { id, status: 'connected' });
        // 只在首次连接成功后，再根据配置启用自动重连
        client.options.reconnectPeriod = server.reconnect ? server.reconnectInterval : 0;
        // 自动订阅已有主题
        server.topics.forEach((t) => {
          this._doSubscribe(id, t.topic, { qos: t.qos });
        });
      });

      client.on('error', (err) => {
        console.error(`[${server.name}] 连接错误:`, err.message);
        server.status = 'error';
        this.notify('status-change', { id, status: 'error', error: err.message });
      });

      client.on('close', () => {
        if (server.status === 'connected') {
          server.status = 'disconnected';
          server.topics.forEach((t) => {
            t.subscribed = false;
          });
          this.notify('status-change', { id, status: 'disconnected' });
        }
      });

      client.on('message', (topic, payload, packet) => {

        let msg = {
          time: new Date().toLocaleTimeString('zh-CN', { hour12: false }) + '.' + String(Date.now() % 1000).padStart(3, '0'),
          topic: topic,
          payload: payload.toString(),
          qos: packet.qos,
          retain: packet.retain,
          direction: 'sub',
          id: this.nextMsgId(),
        };

        let msgs = this.messages.get(id) || [];

        msgs.unshift(msg);

        if (msgs.length > 500) {
          msgs.pop();
        }

        this.messages.set(id, msgs);
        this.notify('message', { serverId: id, message: msg });

      });

      this.clients.set(id, client);

      return true;

    } catch (err) {
      server.status = 'error';
      this.notify('status-change', { id, status: 'error', error: err.message });
      return false;
    }
  }

  /**
   * 断开指定服务器的连接
   * @param {string} id - 服务器 id
   */
  disconnect(id) {

    let client = this.clients.get(id);

    if (client) {
      client.end(true);
      this.clients.delete(id);
    }

    let server = this.servers.get(id);

    if (server) {
      server.status = 'disconnected';
      server.topics.forEach((t) => {
        t.subscribed = false;
      });
      this.notify('status-change', { id, status: 'disconnected' });
    }

  }

  // ==================== 订阅管理 ====================

  /**
   * 执行订阅操作（内部方法）
   * @param {string} serverId - 服务器 id
   * @param {string} topic - 主题名称
   * @param {PublishOptions} opts - 订阅选项
   * @returns {boolean} 是否发起订阅
   */
  _doSubscribe(serverId, topic, opts) {

    let client = this.clients.get(serverId);
    let server = this.servers.get(serverId);

    if (!client || !server || server.status !== 'connected') {
      return false;
    }

    client.subscribe(topic, opts, (err) => {
      if (err) {
        console.error(`订阅失败 [${topic}]:`, err.message);
        this.notify('subscription-error', { serverId, topic, error: err.message });
      } else {
        let t = server.topics.find((t) => t.topic === topic);
        if (t) {
          t.subscribed = true;
        }
        this.notify('subscription-change', { serverId, topic, subscribed: true });
      }
    });

    return true;

  }

  /**
   * 为主题列表添加新主题
   * @param {string} serverId - 服务器 id
   * @param {string} topic - 主题名称
   * @param {MqttQoS} qos - QoS 等级
   * @param {string} comment - 主题备注
   * @returns {MqttTopic | null} 添加后的主题对象或 null
   */
  addTopic(serverId, topic, qos = 1, comment = '') {
    let server = this.servers.get(serverId);
    if (!server) {
      return null;
    }
    if (server.topics.find((t) => t.topic === topic)) {
      return null;
    }

    let topicId = 'topic_' + Date.now() + '_' + Math.random().toString(36).slice(2, 6);

    let newTopic = { id: topicId, topic, qos, comment, subscribed: false };
    server.topics.push(newTopic);

    this.saveData();

    // 如果已连接，自动订阅
    if (server.status === 'connected') {
      this._doSubscribe(serverId, topic, { qos });
    }

    return newTopic;

  }

  /**
   * 移除主题并取消订阅
   * @param {string} serverId - 服务器 id
   * @param {string} topic - 主题名称
   * @returns {boolean} 是否移除成功
   */
  removeTopic(serverId, topic) {

    let server = this.servers.get(serverId);

    if (!server) {
      return false;
    }

    // 先取消订阅
    this.unsubscribe(serverId, topic);

    server.topics = server.topics.filter((t) => t.topic !== topic);

    this.saveData();

    return true;

  }

  /**
   * 订阅指定主题
   * @param {string} serverId - 服务器 id
   * @param {string} topic - 主题名称
   * @returns {boolean} 是否发起订阅
   */
  subscribe(serverId, topic) {

    let server = this.servers.get(serverId);

    if (!server) {
      return false;
    }

    let t = server.topics.find((t) => t.topic === topic);

    if (!t) {
      return false;
    }

    return this._doSubscribe(serverId, topic, { qos: t.qos });

  }

  /**
   * 取消订阅指定主题
   * @param {string} serverId - 服务器 id
   * @param {string} topic - 主题名称
   * @returns {boolean} 是否发起取消订阅
   */
  unsubscribe(serverId, topic) {

    let client = this.clients.get(serverId);
    let server = this.servers.get(serverId);

    if (!client || !server || server.status !== 'connected') {
      return false;
    }

    client.unsubscribe(topic, {}, (err) => {
      if (!err) {
        let t = server.topics.find((t) => t.topic === topic);
        if (t) {
          t.subscribed = false;
        }
        this.notify('subscription-change', { serverId, topic, subscribed: false });
      }
    });

    return true;

  }

  /**
   * 更新主题信息，必要时取消旧主题并重新订阅新主题
   * @param {string} serverId - 服务器 id
   * @param {string} topicId - 主题 id
   * @param {string} newTopic - 新的主题名称
   * @param {number} newQos - 新的 QoS 等级
   * @param {string} comment - 新的主题备注
   * @returns {boolean} 是否更新成功
   */
  updateTopic(serverId, topicId, newTopic, newQos, comment = '') {

    let server = this.servers.get(serverId);

    if (!server) {
      return false;
    }

    let t = server.topics.find((item) => item.id === topicId);

    if (!t) {
      return false;
    }

    let topicChanged = t.topic !== newTopic;

    if (
      topicChanged &&
      server.topics.find((item) => item.topic === newTopic && item.id !== topicId)
    ) {
      return false;
    }

    let oldTopic = t.topic;
    let wasSubscribed = t.subscribed;

    // 如果主题名变更且已订阅，先取消旧主题订阅
    if (topicChanged && wasSubscribed) {
      this.unsubscribe(serverId, oldTopic);
    }

    t.topic = newTopic;
    t.qos = newQos;
    t.comment = comment;

    // 如果已连接且此前已订阅，重新订阅新主题
    if (server.status === 'connected' && wasSubscribed) {
      this._doSubscribe(serverId, newTopic, { qos: newQos });
    }

    this.saveData();
    this.notify('subscription-change', {
      serverId,
      topic: newTopic,
      subscribed: t.subscribed,
      topicId,
      comment: t.comment,
    });

    return true;

  }

  /**
   * 批量订阅服务器下所有未订阅的主题
   * @param {string} serverId - 服务器 id
   * @returns {boolean} 是否发起批量订阅
   */
  subscribeAll(serverId) {

    let client = this.clients.get(serverId);
    let server = this.servers.get(serverId);

    if (!client || !server || server.status !== 'connected') {
      return false;
    }

    let topicsToSubscribe = server.topics.filter((t) => !t.subscribed);

    if (topicsToSubscribe.length === 0) {
      return true;
    }

    let subMap = Object.fromEntries(topicsToSubscribe.map((t) => [t.topic, { qos: t.qos }]));

    client.subscribe(subMap, (err) => {
      if (err) {
        console.error('批量订阅失败:', err.message);
        topicsToSubscribe.forEach((t) => {
          this.notify('subscription-error', { serverId, topic: t.topic, error: err.message });
        });
      } else {
        topicsToSubscribe.forEach((t) => {
          t.subscribed = true;
          this.notify('subscription-change', {
            serverId,
            topic: t.topic,
            subscribed: true,
            topicId: t.id,
          });
        });
      }
    });

    return true;

  }

  // ==================== 消息发布 ====================

  /**
   * 向指定主题发布消息
   * @param {string} serverId - 服务器 id
   * @param {string} topic - 目标主题
   * @param {string} payload - 消息内容
   * @param {PublishOptions} opts - 发布选项
   * @returns {boolean} 是否发起发布
   */
  publish(serverId, topic, payload, opts = {}) {

    let client = this.clients.get(serverId);
    let server = this.servers.get(serverId);

    if (!client || !server || server.status !== 'connected') {
      return false;
    }

    let pubOpts = {
      qos: opts.qos ?? 1,
      retain: opts.retain ?? false,
    };

    client.publish(topic, payload, pubOpts, (err) => {
      if (err) {
        console.error(`发布失败 [${topic}]:`, err.message);
        this.notify('publish-error', { serverId, topic, error: err.message });
      }
    });

    // 记录发布历史
    let pubEntry = {
      time:
        new Date().toLocaleTimeString('zh-CN', { hour12: false }) +
        '.' +
        String(Date.now() % 1000).padStart(3, '0'),
      topic,
      payload,
      qos: pubOpts.qos,
      retain: pubOpts.retain,
      direction: 'pub',
    };

    let history = this.publishHistory.get(serverId) || [];

    history.unshift(pubEntry);

    if (history.length > 50) {
      history.pop();
    }

    this.publishHistory.set(serverId, history);

    // 同时添加到消息日志
    let pubMsgId = this.nextMsgId();
    let msgs = this.messages.get(serverId) || [];

    msgs.unshift({
      ...pubEntry,
      id: pubMsgId,
    });

    if (msgs.length > 500) {
      msgs.pop();
    }

    this.messages.set(serverId, msgs);

    this.notify('message', {
      serverId,
      message: {
        ...pubEntry,
        id: pubMsgId,
      },
    });

    this.saveData();

    return true;

  }

  // ==================== 消息管理 ====================

  /**
   * 获取指定服务器的消息列表
   * @param {string} serverId - 服务器 id
   * @returns {MqttMessage[]} 消息数组
   */
  getMessages(serverId) {
    return this.messages.get(serverId) || [];
  }

  /**
   * 获取指定服务器的发布历史
   * @param {string} serverId - 服务器 id
   * @returns {PublishRecord[]} 发布历史数组
   */
  getPublishHistory(serverId) {
    return this.publishHistory.get(serverId) || [];
  }

  /**
   * 清空指定服务器的消息列表
   * @param {string} serverId - 服务器 id
   */
  clearMessages(serverId) {
    this.messages.set(serverId, []);
    this.saveData();
  }

  /**
   * 清空指定服务器的发布历史
   * @param {string} serverId - 服务器 id
   */
  clearPublishHistory(serverId) {
    this.publishHistory.set(serverId, []);
    this.saveData();
  }

}

// 单例
const mqttManager = new MqttManager();
export default mqttManager;