MQTTvue/mqtt-vue-dashboard/server/dist/services/mqttBrokerService.js

"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
    return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.MqttBrokerService = void 0;
const aedes_1 = __importDefault(require("aedes"));
const net_1 = require("net");
const device_1 = require("../models/device");
const mqttMessage_1 = require("../models/mqttMessage");
const clientConnection_1 = require("../models/clientConnection");
const clientAuth_1 = require("../models/clientAuth");
const clientAcl_1 = require("../models/clientAcl");
const authLog_1 = require("../models/authLog");
const loggerService_1 = require("./loggerService");
const websocketService_1 = require("./websocketService");
const ota_1 = require("../models/ota");
const firmware_1 = require("../models/firmware");
class MqttBrokerService {
    constructor() {
        this.connectedClients = new Map();
        this.syncInterval = null;
        this.port = parseInt(process.env.MQTT_BROKER_PORT || '1883', 10);
        this.broker = new aedes_1.default({
            id: 'mqtt-vue-dashboard-broker',
            concurrency: 100,
            heartbeatInterval: 60000,
            connectTimeout: 30000,
        });
        this.server = (0, net_1.createServer)(this.broker.handle);
        this.setupBrokerEvents();
    }
    static getInstance() {
        if (!MqttBrokerService.instance) {
            MqttBrokerService.instance = new MqttBrokerService();
        }
        return MqttBrokerService.instance;
    }
    start() {
        return new Promise((resolve, reject) => {
            this.server.listen(this.port, () => {
                console.log(`MQTT Broker 已启动，监听端口: ${this.port}`);
                loggerService_1.LoggerService.info('MQTT Broker 已启动', {
                    source: 'mqtt_broker',
                    module: 'startup',
                    details: JSON.stringify({ port: this.port })
                }).catch(() => { });
                this.startDeviceStatusSync();
                resolve();
            });
            this.server.on('error', (err) => {
                console.error('MQTT Broker 启动失败:', err.message);
                loggerService_1.LoggerService.error('MQTT Broker 启动失败', {
                    source: 'mqtt_broker',
                    module: 'startup',
                    details: JSON.stringify({ error: err.message })
                }).catch(() => { });
                reject(err);
            });
        });
    }
    stop() {
        return new Promise((resolve) => {
            if (this.syncInterval) {
                clearInterval(this.syncInterval);
                this.syncInterval = null;
            }
            this.server.close(() => {
                this.broker.close(() => {
                    console.log('MQTT Broker 已停止');
                    resolve();
                });
            });
        });
    }
    setupBrokerEvents() {
        this.broker.authenticate = ((client, username, password, done) => {
            this.authenticateClient(client, username, password, done);
        });
        this.broker.authorizePublish = this.authorizePublish.bind(this);
        this.broker.authorizeSubscribe = this.authorizeSubscribe.bind(this);
        this.broker.on('client', (client) => {
            this.handleClientConnect(client);
        });
        this.broker.on('clientDisconnect', (client) => {
            this.handleClientDisconnect(client);
        });
        this.broker.on('publish', (packet, client) => {
            if (client) {
                this.handleMessagePublish(packet, client);
            }
        });
        this.broker.on('subscribe', (subscriptions, client) => {
            if (client) {
                this.handleClientSubscribe(subscriptions, client);
            }
        });
        this.broker.on('unsubscribe', (subscriptions, client) => {
            if (client) {
                this.handleClientUnsubscribe(subscriptions, client);
            }
        });
        this.broker.on('clientError', (client, err) => {
            console.error(`客户端错误 [${client.id}]:`, err.message);
        });
    }
    async authenticateClient(client, username, password, callback) {
        const clientId = client.id;
        const ip = this.getClientIp(client);
        if (!username || !password) {
            const allowAnonymous = process.env.MQTT_ALLOW_ANONYMOUS === 'true';
            if (allowAnonymous) {
                this.logAuth(clientId, username || '', ip, 'connect', 'success', '');
                callback(null, true);
                return;
            }
            this.logAuth(clientId, username || '', ip, 'connect', 'failure', '缺少用户名或密码');
            callback(null, false);
            return;
        }
        try {
            let authRecord = await clientAuth_1.ClientAuthModel.findByUsernameAndClientId(username, clientId);
            if (!authRecord) {
                authRecord = await clientAuth_1.ClientAuthModel.findByUsername(username);
            }
            if (!authRecord) {
                this.logAuth(clientId, username, ip, 'connect', 'failure', '认证记录不存在');
                callback(null, false);
                return;
            }
            if (authRecord.status === 'disabled') {
                this.logAuth(clientId, username, ip, 'connect', 'failure', '客户端已禁用');
                callback(null, false);
                return;
            }
            const passwordStr = password.toString();
            const isMatch = clientAuth_1.ClientAuthModel.verifyPassword(passwordStr, authRecord.salt || '', authRecord.password_hash, authRecord.use_salt);
            if (!isMatch) {
                this.logAuth(clientId, username, ip, 'connect', 'failure', '密码错误');
                callback(null, false);
                return;
            }
            await clientAuth_1.ClientAuthModel.updateLastLogin(username, clientId);
            this.logAuth(clientId, username, ip, 'connect', 'success', '');
            callback(null, true);
        }
        catch (error) {
            console.error('认证过程出错:', error);
            this.logAuth(clientId, username || '', ip, 'connect', 'failure', '认证服务错误');
            callback(null, false);
        }
    }
    async authorizePublish(client, packet, callback) {
        const username = client.username;
        const topic = packet.topic;
        try {
            if (topic && topic.startsWith('$SYS/')) {
                callback(null);
                return;
            }
            const isSuperuser = await this.isSuperuser(username);
            if (isSuperuser) {
                callback(null);
                return;
            }
            const hasPermission = await this.checkAclPermission(username, topic, 'publish');
            if (hasPermission) {
                callback(null);
            }
            else {
                this.logAuth(client.id, username || '', this.getClientIp(client), 'publish', 'failure', `无权发布到主题: ${topic}`);
                callback(new Error(`无权发布到主题: ${topic}`));
            }
        }
        catch (error) {
            console.error('发布授权检查出错:', error);
            callback(null);
        }
    }
    async authorizeSubscribe(client, subscription, callback) {
        const username = client.username;
        const topic = subscription.topic;
        try {
            const isSuperuser = await this.isSuperuser(username);
            if (isSuperuser) {
                callback(null, subscription);
                return;
            }
            if (!topic) {
                callback(null, subscription);
                return;
            }
            const hasPermission = await this.checkAclPermission(username, topic, 'subscribe');
            if (!hasPermission) {
                this.logAuth(client.id, username || '', this.getClientIp(client), 'subscribe', 'failure', `无权订阅主题: ${topic}`);
                callback(new Error(`无权订阅主题: ${topic}`));
                return;
            }
            callback(null, subscription);
        }
        catch (error) {
            console.error('订阅授权检查出错:', error);
            callback(null, subscription);
        }
    }
    async isSuperuser(username) {
        if (!username)
            return false;
        try {
            const authRecord = await clientAuth_1.ClientAuthModel.findByUsername(username);
            return authRecord ? (authRecord.is_superuser === true || authRecord.is_superuser === 1) : false;
        }
        catch {
            return false;
        }
    }
    async checkAclPermission(username, topic, action) {
        if (!username)
            return true;
        try {
            const aclRules = await clientAcl_1.ClientAclModel.getByUsername(username);
            if (!aclRules || aclRules.length === 0)
                return true;
            for (const rule of aclRules) {
                if (this.topicMatches(topic, rule.topic)) {
                    if (rule.action === 'pubsub' || rule.action === action) {
                        return rule.permission === 'allow';
                    }
                }
            }
            return true;
        }
        catch {
            return true;
        }
    }
    topicMatches(topic, pattern) {
        if (pattern === '#')
            return true;
        if (pattern === topic)
            return true;
        const patternParts = pattern.split('/');
        const topicParts = topic.split('/');
        for (let i = 0; i < patternParts.length; i++) {
            if (patternParts[i] === '#')
                return true;
            if (i >= topicParts.length)
                return false;
            if (patternParts[i] !== '+' && patternParts[i] !== topicParts[i])
                return false;
        }
        return patternParts.length === topicParts.length;
    }
    getClientIp(client) {
        if (client.conn && client.conn.remoteAddress) {
            const remoteAddress = client.conn.remoteAddress;
            if (remoteAddress === '::1') {
                return '127.0.0.1';
            }
            if (remoteAddress.startsWith('::ffff:')) {
                return remoteAddress.substring(7);
            }
            return remoteAddress;
        }
        return 'unknown';
    }
    async handleClientConnect(client) {
        const clientId = client.id;
        const username = client.username || '';
        const ip = this.getClientIp(client);
        this.connectedClients.set(clientId, {
            id: clientId,
            username,
            ip,
            connectedAt: new Date()
        });
        console.log(`客户端已连接: ${clientId} (${username}) from ${ip}`);
        try {
            let device = await device_1.DeviceModel.getByClientId(clientId);
            if (device) {
                await device_1.DeviceModel.update(clientId, {
                    status: 'online',
                    last_event_time: new Date(),
                    last_online_time: new Date(),
                    device_ip_port: ip,
                    last_ip_port: ip,
                    connect_count: (device.connect_count || 0) + 1
                });
            }
            else {
                await device_1.DeviceModel.create({
                    clientid: clientId,
                    device_name: clientId,
                    username,
                    status: 'online',
                    last_event_time: new Date(),
                    last_online_time: new Date(),
                    device_ip_port: ip,
                    last_ip_port: ip,
                    connect_count: 1
                });
            }
            await clientConnection_1.ClientConnectionModel.create({
                clientid: clientId,
                username,
                event: 'client.connected',
                timestamp: new Date(),
                connected_at: new Date(),
                node: 'mqtt-vue-dashboard',
                peername: ip,
                sockname: `0.0.0.0:${this.port}`,
                proto_name: 'MQTT',
                proto_ver: 4,
                keepalive: 60,
                clean_start: 1
            });
            this.broadcastToWebSocket('device_connected', {
                clientid: clientId,
                username,
                ip,
                timestamp: new Date().toISOString()
            });
            await this.executePendingOTATasks(clientId);
        }
        catch (error) {
            console.error('处理客户端连接事件出错:', error);
        }
    }
    async handleClientDisconnect(client) {
        const clientId = client.id;
        const clientInfo = this.connectedClients.get(clientId);
        const username = clientInfo?.username || '';
        const ip = clientInfo?.ip || 'unknown';
        const connectedAt = clientInfo?.connectedAt;
        this.connectedClients.delete(clientId);
        console.log(`客户端已断开: ${clientId}`);
        try {
            await device_1.DeviceModel.update(clientId, {
                status: 'offline',
                last_event_time: new Date(),
                last_offline_time: new Date()
            });
            let connectionDuration;
            if (connectedAt) {
                connectionDuration = Math.floor((Date.now() - connectedAt.getTime()) / 1000);
                await device_1.DeviceModel.updateOnlineDuration(clientId, connectionDuration);
            }
            await clientConnection_1.ClientConnectionModel.create({
                clientid: clientId,
                username,
                event: 'client.disconnected',
                timestamp: new Date(),
                connected_at: connectedAt || undefined,
                node: 'mqtt-vue-dashboard',
                peername: ip,
                sockname: `0.0.0.0:${this.port}`,
                proto_name: 'MQTT',
                proto_ver: 4,
                keepalive: 60,
                clean_start: 1,
                reason: 'normal',
                connection_duration: connectionDuration
            });
            this.broadcastToWebSocket('device_disconnected', {
                clientid: clientId,
                username,
                timestamp: new Date().toISOString()
            });
        }
        catch (error) {
            console.error('处理客户端断开事件出错:', error);
        }
    }
    async handleMessagePublish(packet, client) {
        const clientId = client.id;
        const username = client.username || '';
        const topic = packet.topic;
        const payload = packet.payload?.toString() || '';
        const qosValue = typeof packet.qos === 'number' ? Math.min(packet.qos, 2) : 0;
        try {
            if (topic.startsWith('$SYS/'))
                return;
            await mqttMessage_1.MqttMessageModel.create({
                clientid: clientId,
                topic,
                payload,
                qos: qosValue,
                retain: packet.retain ? 1 : 0,
                message_type: 'publish',
                timestamp: Date.now(),
                node: 'mqtt-vue-dashboard',
                username,
                proto_ver: 4,
                payload_format: this.detectPayloadFormat(payload),
                message_time: new Date()
            });
            this.broadcastToWebSocket('mqtt_message', {
                clientid: clientId,
                topic,
                payload,
                qos: qosValue,
                retain: packet.retain,
                timestamp: Date.now()
            });
            await this.handleDeviceMessage(clientId, topic, payload);
        }
        catch (error) {
            console.error('处理消息发布事件出错:', error);
        }
    }
    async handleClientSubscribe(subscriptions, client) {
        const clientId = client.id;
        const username = client.username || '';
        const subsList = Array.isArray(subscriptions) ? subscriptions : [subscriptions];
        for (const sub of subsList) {
            try {
                const qosValue = typeof sub.qos === 'number' ? Math.min(sub.qos, 2) : 0;
                await mqttMessage_1.MqttMessageModel.create({
                    clientid: clientId,
                    topic: sub.topic,
                    payload: '',
                    qos: qosValue,
                    retain: 0,
                    message_type: 'subscribe',
                    timestamp: Date.now(),
                    node: 'mqtt-vue-dashboard',
                    username,
                    proto_ver: 4,
                    payload_format: 'text',
                    message_time: new Date()
                });
                this.broadcastToWebSocket('mqtt_subscribe', {
                    clientid: clientId,
                    topic: sub.topic,
                    qos: qosValue,
                    timestamp: Date.now()
                });
            }
            catch (error) {
                console.error('处理订阅事件出错:', error);
            }
        }
    }
    async handleClientUnsubscribe(subscriptions, client) {
        const clientId = client.id;
        const username = client.username || '';
        const subsList = Array.isArray(subscriptions) ? subscriptions : [subscriptions];
        for (const topic of subsList) {
            try {
                await mqttMessage_1.MqttMessageModel.create({
                    clientid: clientId,
                    topic: typeof topic === 'string' ? topic : topic.topic,
                    payload: '',
                    qos: 0,
                    retain: 0,
                    message_type: 'unsubscribe',
                    timestamp: Date.now(),
                    node: 'mqtt-vue-dashboard',
                    username,
                    proto_ver: 4,
                    payload_format: 'text',
                    message_time: new Date()
                });
            }
            catch (error) {
                console.error('处理取消订阅事件出错:', error);
            }
        }
    }
    async handleDeviceMessage(clientId, topic, payload) {
        try {
            const topicParts = topic.split('/');
            if (topicParts.length < 2)
                return;
            const deviceId = topicParts[1];
            const messageType = topicParts.slice(2).join('/');
            if (messageType === 'ota/status' || messageType === 'ota/progress') {
                await this.handleOtaMessage(deviceId, payload, messageType);
            }
            else if (messageType === 'relay/state') {
                await this.handleRelayStateMessage(deviceId, payload);
            }
            else if (messageType === 'wifi/rssi') {
                await this.handleRssiMessage(deviceId, payload);
            }
            else if (messageType === 'wifi/info') {
                await this.handleWifiInfoMessage(deviceId, payload);
            }
            else if (messageType === 'wifi/status') {
                await this.handleWifiStatusMessage(deviceId, payload);
            }
            else if (messageType === 'sensor/data' || messageType === 'data') {
                await this.handleSensorData(deviceId, topic, payload);
            }
        }
        catch (error) {
            console.error('处理设备消息出错:', error);
        }
    }
    async handleOtaMessage(deviceId, payload, messageType) {
        try {
            const data = JSON.parse(payload);
            let taskId = data.task_id || data.tid;
            if (taskId === undefined || taskId === null)
                return;
            let task = await ota_1.OTATaskModel.getById(taskId);
            if (!task)
                return;
            if (messageType === 'ota/progress') {
                await ota_1.OTATaskModel.updateStatusAndProgress(taskId, task.status, data.progress || 0);
            }
            else if (messageType === 'ota/status') {
                const validStatuses = ['pending', 'downloading', 'installing', 'success', 'failed', 'ready'];
                const status = data.status;
                if (!validStatuses.includes(status))
                    return;
                const progress = data.progress !== undefined ? data.progress : (status === 'success' || status === 'ready' ? 100 : task.progress);
                if (status === 'ready') {
                    const firmware = await firmware_1.FirmwareFileModel.getById(task.firmware_id);
                    if (firmware && data.firmware_version) {
                        if (data.firmware_version === firmware.version) {
                            await ota_1.OTATaskModel.updateStatusAndProgress(taskId, 'success', 100);
                        }
                        else {
                            await device_1.DeviceModel.update(deviceId, { firmware_version: data.firmware_version });
                            await this.publishOtaCommand(taskId);
                            return;
                        }
                    }
                    else {
                        await ota_1.OTATaskModel.updateStatusAndProgress(taskId, 'success', 100);
                    }
                }
                else {
                    await ota_1.OTATaskModel.updateStatusAndProgress(taskId, status, progress);
                }
                if (status === 'success' || status === 'ready') {
                    if (data.firmware_version) {
                        await device_1.DeviceModel.update(deviceId, { firmware_version: data.firmware_version });
                    }
                }
                if (status === 'success' || status === 'failed') {
                    await ota_1.OTATaskModel.updateResult(taskId, status, data.error_message);
                }
            }
        }
        catch (error) {
            console.error('处理OTA消息出错:', error);
        }
    }
    async handleRelayStateMessage(deviceId, payload) {
        try {
            const data = JSON.parse(payload);
            this.broadcastToWebSocket('device_relay_state', {
                deviceId,
                state: data.state || data,
                timestamp: new Date().toISOString()
            });
        }
        catch (error) {
            console.error('处理继电器状态消息出错:', error);
        }
    }
    async handleRssiMessage(deviceId, payload) {
        try {
            const rssi = parseInt(payload, 10);
            if (!isNaN(rssi)) {
                await device_1.DeviceModel.update(deviceId, { rssi });
                this.broadcastToWebSocket('device_rssi', {
                    deviceId,
                    rssi,
                    timestamp: new Date().toISOString()
                });
            }
        }
        catch (error) {
            console.error('处理RSSI消息出错:', error);
        }
    }
    async handleWifiInfoMessage(deviceId, payload) {
        try {
            const data = JSON.parse(payload);
            this.broadcastToWebSocket('device_wifi_info', {
                deviceId,
                ...data,
                timestamp: new Date().toISOString()
            });
        }
        catch (error) {
            console.error('处理WiFi信息消息出错:', error);
        }
    }
    async handleWifiStatusMessage(deviceId, payload) {
        try {
            const data = JSON.parse(payload);
            this.broadcastToWebSocket('device_wifi_status', {
                deviceId,
                ...data,
                timestamp: new Date().toISOString()
            });
        }
        catch (error) {
            console.error('处理WiFi状态消息出错:', error);
        }
    }
    async handleSensorData(deviceId, topic, payload) {
        try {
            const { SensorDataModel } = require('../models/sensorData');
            const data = JSON.parse(payload);
            await SensorDataModel.create({
                device_id: deviceId,
                topic,
                data_type: data.type || 'unknown',
                value: payload,
                timestamp: new Date()
            });
            this.broadcastToWebSocket('sensor_data', {
                deviceId,
                topic,
                data,
                timestamp: new Date().toISOString()
            });
        }
        catch (error) {
            console.error('处理传感器数据出错:', error);
        }
    }
    async publishOtaCommand(taskId) {
        try {
            const task = await ota_1.OTATaskModel.getById(taskId);
            if (!task)
                return;
            const firmware = await firmware_1.FirmwareFileModel.getById(task.firmware_id);
            if (!firmware) {
                await ota_1.OTATaskModel.updateResult(taskId, 'failed', '固件不存在');
                return;
            }
            let otaServerUrl = process.env.OTA_SERVER_URL || process.env.BACKEND_URL || `http://localhost:${process.env.PORT || 3002}`;
            otaServerUrl = otaServerUrl.replace(/\/$/, '');
            const otaCommand = {
                act: 'upgrade',
                ver: firmware.version,
                url: `${otaServerUrl}/api/ota/firmware/${firmware.id}`,
                md5: firmware.md5sum,
                tid: taskId,
                rc: 3,
                ri: 10000,
                to: 30000
            };
            this.broker.publish({
                topic: `device/${task.device_id}/ota`,
                payload: Buffer.from(JSON.stringify(otaCommand)),
                qos: 1,
                retain: false
            }, (err) => {
                if (err) {
                    console.error(`OTA指令发布失败，任务ID: ${taskId}`, err);
                    ota_1.OTATaskModel.updateResult(taskId, 'failed', `OTA指令发送失败: ${err.message}`);
                }
                else {
                    console.log(`OTA指令已发布，任务ID: ${taskId}, 设备: ${task.device_id}`);
                }
            });
        }
        catch (error) {
            console.error('发布OTA指令出错:', error);
        }
    }
    async executePendingOTATasks(deviceId) {
        try {
            const incompleteTasks = await ota_1.OTATaskModel.getIncompleteTasksByDeviceId(deviceId);
            if (incompleteTasks.length === 0)
                return;
            console.log(`设备 ${deviceId} 上线，发现 ${incompleteTasks.length} 个未完成OTA任务`);
            for (const task of incompleteTasks) {
                if (task.id) {
                    await this.publishOtaCommand(task.id);
                }
            }
        }
        catch (error) {
            console.error(`执行设备 ${deviceId} 的待处理OTA任务出错:`, error);
        }
    }
    startDeviceStatusSync() {
        this.syncInterval = setInterval(async () => {
            try {
                const allDevices = await device_1.DeviceModel.getAll();
                const onlineClientIds = new Set(this.connectedClients.keys());
                for (const device of allDevices) {
                    const isOnline = onlineClientIds.has(device.clientid);
                    const newStatus = isOnline ? 'online' : 'offline';
                    if (device.status !== newStatus) {
                        await device_1.DeviceModel.update(device.clientid, {
                            status: newStatus,
                            last_event_time: new Date(),
                            last_online_time: isOnline ? new Date() : device.last_online_time,
                            last_offline_time: !isOnline ? new Date() : device.last_offline_time
                        });
                    }
                }
            }
            catch (error) {
                console.error('设备状态同步出错:', error);
            }
        }, 30000);
    }
    getConnectedClients() {
        return Array.from(this.connectedClients.values());
    }
    getConnectedClientCount() {
        return this.connectedClients.size;
    }
    getBroker() {
        return this.broker;
    }
    disconnectClient(clientId) {
        const clients = this.broker.clients;
        const client = clients ? clients[clientId] : null;
        if (client) {
            client.close();
            return true;
        }
        return false;
    }
    publish(topic, payload, options) {
        return new Promise((resolve, reject) => {
            this.broker.publish({
                topic,
                payload: Buffer.isBuffer(payload) ? payload : Buffer.from(payload),
                qos: options?.qos || 0,
                retain: options?.retain || false
            }, (err) => {
                if (err)
                    reject(err);
                else
                    resolve();
            });
        });
    }
    detectPayloadFormat(payload) {
        try {
            JSON.parse(payload);
            return 'json';
        }
        catch {
            if (/^[0-9a-fA-F]+$/.test(payload))
                return 'bin';
            return 'text';
        }
    }
    logAuth(clientId, username, ip, operationType, result, reason) {
        authLog_1.AuthLogModel.create({
            clientid: clientId,
            username,
            ip_address: ip,
            operation_type: operationType,
            result,
            reason
        }).catch((err) => {
            console.error('写入认证日志失败:', err);
        });
    }
    broadcastToWebSocket(event, data) {
        try {
            const wsService = (0, websocketService_1.getWebSocketService)();
            if (wsService) {
                const io = wsService.io;
                if (io) {
                    io.emit(event, data);
                }
            }
        }
        catch (error) {
        }
    }
}
exports.MqttBrokerService = MqttBrokerService;
//# sourceMappingURL=mqttBrokerService.js.map