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

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

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.loadData()
  }

  nextMsgId() {
    this.msgIdCounter = (this.msgIdCounter + 1) % Number.MAX_SAFE_INTEGER
    return 'msg_' + Date.now() + '_' + this.msgIdCounter
  }

  // ==================== 数据持久化 ====================
  getDataDir() {
    const dir = app.getPath('userData')
    if (!existsSync(dir)) {
      mkdirSync(dir, { recursive: true })
    }
    return dir
  }

  loadData() {
    try {
      if (existsSync(this.dataPath)) {
        const raw = readFileSync(this.dataPath, 'utf-8')
        const 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),
              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)
          })
        }
      } else {
        this.initDefaultServers()
      }
    } catch (e) {
      console.error('加载数据失败:', e.message)
      this.initDefaultServers()
    }
  }

  saveData() {
    try {
      const 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)
      }
      writeFileSync(this.dataPath, JSON.stringify(data, null, 2), 'utf-8')
    } catch (e) {
      console.error('保存数据失败:', e.message)
    }
  }

  initDefaultServers() {
    const 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,
        status: 'disconnected',
        topics: [
          { id: 'topic_default_001', topic: 'sensor/+/temperature', qos: 1, subscribed: false },
          { id: 'topic_default_002', topic: 'device/status', qos: 2, subscribed: false }
        ]
      }
    ]
    defaults.forEach((s) => {
      this.servers.set(s.id, s)
      this.messages.set(s.id, [])
      this.publishHistory.set(s.id, [])
    })
    this.saveData()
  }

  // ==================== 监听器 ====================
  addListener(callback) {
    this.listeners.add(callback)
  }

  removeListener(callback) {
    this.listeners.delete(callback)
  }

  notify(event, data) {
    this.listeners.forEach((cb) => {
      try {
        cb(event, data)
      } catch (e) {
        console.error('通知失败:', e)
      }
    })
  }

  // ==================== 服务器管理 ====================
  getServers() {
    return Array.from(this.servers.values()).map((s) => ({
      ...s,
      password: s.password ? '******' : ''
    }))
  }

  getServer(id) {
    return this.servers.get(id) || null
  }

  addServer(config) {
    const id = 'srv_' + Date.now() + '_' + Math.random().toString(36).slice(2, 6)
    const server = {
      id,
      name: config.name,
      host: config.host,
      port: config.port || 1883,
      protocol: config.protocol || 'mqtt',
      clientId: config.clientId || 'frost_' + Math.random().toString(36).slice(2, 10),
      username: config.username || '',
      password: config.password || '',
      keepAlive: config.keepAlive || 60,
      cleanSession: config.cleanSession !== false,
      status: 'disconnected',
      topics: []
    }
    this.servers.set(id, server)
    this.messages.set(id, [])
    this.publishHistory.set(id, [])
    this.saveData()
    return server
  }

  updateServer(id, config) {
    const server = this.servers.get(id)
    if (!server) return null
    Object.assign(server, {
      name: config.name,
      host: config.host,
      port: config.port || 1883,
      protocol: config.protocol || 'mqtt',
      clientId: config.clientId || server.clientId,
      username: config.username || '',
      keepAlive: config.keepAlive || 60,
      cleanSession: config.cleanSession !== false
    })
    if (config.password) {
      server.password = config.password
    }
    this.saveData()
    return server
  }

  deleteServer(id) {
    // 先断开连接
    this.disconnect(id)
    this.servers.delete(id)
    this.messages.delete(id)
    this.publishHistory.delete(id)
    this.saveData()
  }

  // ==================== 连接管理 ====================
  connect(id) {
    const 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' })

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

    const options = {
      clientId: server.clientId,
      username: server.username || undefined,
      password: server.password || undefined,
      keepalive: server.keepAlive,
      clean: server.cleanSession,
      reconnectPeriod: 5000,
      connectTimeout: 10000
    }

    try {
      const client = mqtt.connect(url, options)

      client.on('connect', () => {
        server.status = 'connected'
        this.notify('status-change', { id, status: 'connected' })
        // 自动订阅已有主题
        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) => {
        const msg = {
          time:
            new Date().toLocaleTimeString('zh-CN', { hour12: false }) +
            '.' +
            String(Date.now() % 1000).padStart(3, '0'),
          topic,
          payload: payload.toString(),
          qos: packet.qos,
          retain: packet.retain,
          direction: 'sub',
        id: this.nextMsgId()
      }
      const 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
    }
  }

  disconnect(id) {
    const client = this.clients.get(id)
    if (client) {
      client.end(true)
      this.clients.delete(id)
    }
    const server = this.servers.get(id)
    if (server) {
      server.status = 'disconnected'
      server.topics.forEach((t) => {
        t.subscribed = false
      })
      this.notify('status-change', { id, status: 'disconnected' })
    }
  }

  // ==================== 订阅管理 ====================
  _doSubscribe(serverId, topic, opts) {
    const client = this.clients.get(serverId)
    const 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 {
        const t = server.topics.find((t) => t.topic === topic)
        if (t) t.subscribed = true
        this.notify('subscription-change', { serverId, topic, subscribed: true })
      }
    })
    return true
  }

  addTopic(serverId, topic, qos = 1) {
    const server = this.servers.get(serverId)
    if (!server) return false
    if (server.topics.find((t) => t.topic === topic)) return false

    const topicId = 'topic_' + Date.now() + '_' + Math.random().toString(36).slice(2, 6)
    server.topics.push({ id: topicId, topic, qos, subscribed: false })
    this.saveData()

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

  removeTopic(serverId, topic) {
    const 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
  }

  subscribe(serverId, topic) {
    const server = this.servers.get(serverId)
    if (!server) return false
    const t = server.topics.find((t) => t.topic === topic)
    if (!t) return false
    return this._doSubscribe(serverId, topic, { qos: t.qos })
  }

  unsubscribe(serverId, topic) {
    const client = this.clients.get(serverId)
    const server = this.servers.get(serverId)
    if (!client || !server || server.status !== 'connected') return false

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

  updateTopic(serverId, topicId, newTopic, newQos) {
    const server = this.servers.get(serverId)
    if (!server) return false
    const t = server.topics.find((item) => item.id === topicId)
    if (!t) return false

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

    const oldTopic = t.topic
    const wasSubscribed = t.subscribed

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

    t.topic = newTopic
    t.qos = newQos

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

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

  subscribeAll(serverId) {
    const client = this.clients.get(serverId)
    const server = this.servers.get(serverId)
    if (!client || !server || server.status !== 'connected') return false

    const topicsToSubscribe = server.topics.filter((t) => !t.subscribed)
    if (topicsToSubscribe.length === 0) return true

    const 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
  }

  // ==================== 消息发布 ====================
  publish(serverId, topic, payload, opts = {}) {
    const client = this.clients.get(serverId)
    const server = this.servers.get(serverId)
    if (!client || !server || server.status !== 'connected') return false

    const 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 })
      }
    })

    // 记录发布历史
    const 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'
    }
    const history = this.publishHistory.get(serverId) || []
    history.unshift(pubEntry)
    if (history.length > 50) history.pop()
    this.publishHistory.set(serverId, history)

    // 同时添加到消息日志
    const pubMsgId = this.nextMsgId()
    const 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
  }

  // ==================== 消息管理 ====================
  getMessages(serverId) {
    return this.messages.get(serverId) || []
  }

  getPublishHistory(serverId) {
    return this.publishHistory.get(serverId) || []
  }

  clearMessages(serverId) {
    this.messages.set(serverId, [])
    this.saveData()
  }

  clearPublishHistory(serverId) {
    this.publishHistory.set(serverId, [])
    this.saveData()
  }
}

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