<?php
// +----------------------------------------------------------------------
// | YwxApp [ WE CAN DO IT JUST THINK ]
// +----------------------------------------------------------------------
// | Copyright (c) 2026-2036 http://ywxapp.cn All rights reserved.
// +----------------------------------------------------------------------
// | Author: ywxapp<admin@ywxapp.cn>
// +----------------------------------------------------------------------
namespace addon\iot\service;

use think\facade\Db;
use think\facade\Log;
use Workerman\Worker;
use Workerman\Timer;
use Workerman\Connection\TcpConnection;
use Workerman\Protocols\Websocket;
use ywxapp\service\AddonService;
use addon\iot\protocol\Mqtt;
use addon\iot\service\Bridge;
use addon\iot\service\DeviceService;
use addon\iot\service\Forwarder;
use addon\iot\service\RuleEngine;

/*
 * MQTT Broker 服务（兼容 MQTT 3.1.1 与 5.0）—— 类 EMQX 轻量单机版
 *
 * 基础能力（原有）：
 *  - CONNECT/CONNACK、PUBLISH + QoS0/1/2、SUBSCRIBE/UNSUBSCRIBE、PING、DISCONNECT、AUTH
 *  - 通配符 + / #、保留消息、遗嘱、会话保持、TCP 分包（交由 Mqtt 协议类）
 *
 * 增强能力（本次）：
 *  - 认证：账号密码校验（Auth 服务，兼容匿名开关）
 *  - ACL ：发布/订阅主题级权限（Auth 服务，$SYS 只读）
 *  - 传输：WebSocket 接入（ws://，MQTT over WebSocket）+ 可选 TLS（mqtts）
 *  - 持久化：保留消息重启恢复 + clean=false 会话离线消息补投（Store 服务）
 *  - 共享订阅：$share/{group}/{filter} 组内负载均衡（round-robin）
 *  - 可观测：$SYS/broker/# 系统主题周期发布运行指标
 *
 * 仍为单进程内存态（worker->count=1），会话/订阅存内存，保留/离线落 DB。
 */
/**
 * Broker 类
 *
 * @author ywxapp <admin@ywxapp.cn>
 */
class Broker
{
    /* ---------- MQTT 控制报文类型 ---------- */
    const CMD_CONNECT     = 1;
    const CMD_CONNACK     = 2;
    const CMD_PUBLISH     = 3;
    const CMD_PUBACK      = 4;
    const CMD_PUBREC      = 5;
    const CMD_PUBREL      = 6;
    const CMD_PUBCOMP     = 7;
    const CMD_SUBSCRIBE   = 8;
    const CMD_SUBACK      = 9;
    const CMD_UNSUBSCRIBE = 10;
    const CMD_UNSUBACK    = 11;
    const CMD_PINGREQ     = 12;
    const CMD_PINGRESP    = 13;
    const CMD_DISCONNECT  = 14;
    const CMD_AUTH        = 15;

    /* ---------- CONNACK 返回/原因码（3.1.1 / 5.0） ---------- */
    const RC_ACCEPTED           = 0;
    const RC_UNACCEPTABLE_PROTO = [4 => 1, 5 => 0x84];   // 3.1.1:1  5.0:0x84
    const RC_NOT_AUTHORIZED     = [4 => 5, 5 => 0x87];   // 3.1.1:5  5.0:0x87
    const RC_BAD_CREDENTIALS    = [4 => 4, 5 => 0x86];   // 3.1.1:4  5.0:0x86

    /** @var Worker TCP 监听 */
    protected $worker;

    /** @var Worker|null WebSocket 监听 */
    protected $wsWorker;

    /** @var array 运行配置 */
    protected $config = [];

    /** @var Auth */
    protected $auth;

    /** @var Store */
    protected $store;

    /** @var Bridge 跨进程消息桥接 */
    protected $bridge;

    /** @var Forwarder 外部 Broker 转发（规则引擎使用） */
    protected $forwarder;

    /** @var RuleEngine 规则引擎 */
    protected $ruleEngine;

    /** @var bool 是否主进程（仅主进程发布 $SYS） */
    protected $isPrimary = true;

    /**
     * @var array 会话表
     * client_id => [
     *   'conn'=>TcpConnection|null, 'clean'=>bool, 'username'=>?string,
     *   'subscriptions'=>[key => ['filter'=>string,'share'=>?string,'qos'=>int,'noLocal'=>bool,'rap'=>bool,'rh'=>int]]
     * ]
     */
    protected $sessions = [];

    /** @var array 保留消息 topic => ['payload'=>string,'qos'=>int] */
    protected $retained = [];

    /** @var array 共享订阅 round-robin 游标 groupKey => int */
    protected $rrIndex = [];

    /** @var array 运行统计 */
    protected $stats = ['recv' => 0, 'sent' => 0, 'startTime' => 0];

    
    public function start(string $host = '0.0.0.0', int $port = 1883, string $mode = 'both'): void
    {
        // Workerman 运行时文件：pidFile 供 stop/restart 命令定位 master 进程；daemon 模式日志落点
        $runtimeDir = rtrim((string) runtime_path(), DIRECTORY_SEPARATOR);
        Worker::$pidFile     = $runtimeDir . DIRECTORY_SEPARATOR . 'iot.pid';
        Worker::$stdoutFile  = $runtimeDir . DIRECTORY_SEPARATOR . 'iot.log';
        // ⚠️ Workerman::parseCommand() 直接扫描 argv，取「第一个命中的白名单词」
        //    （start/stop/restart/reload/status/connections）当作自己的命令；一个都命中不了就 exit(usage)。
        //    · 合并写法 `php think iot start`：argv 里本就有 start，无需注入；
        //    · `php think iot restart`：会命中 restart —— Workerman 的 restart 分支在 pid 文件已被
        //      Iot::doRestart() 删除的前提下等价于 start（只是多等一次旧 master 退出），故可共用；
        //    · 旧写法 `php think iot:start` / `php think iot:restart`：命令名带冒号，命中不了白名单，
        //      必须靠官方 library 模式入口 Worker::$command 注入兜底，否则 Workerman 会打印 usage 退出
        //      （daemon 模式下这行 usage 只会出现在 runtime/iot.log 里，终端看不到）。
        if (DIRECTORY_SEPARATOR === '/' && Worker::$command === '') {
            Worker::$command = 'start';
        }

        $this->config = $this->loadConfig();
        $this->auth   = new Auth($this->config);
        $this->store  = new Store($this->config);
        $this->bridge = new Bridge($this->config);
        $this->forwarder  = new Forwarder($this->config);
        $this->ruleEngine = new RuleEngine($this->config, $this->forwarder);

        if ($host === '0.0.0.0' && !empty($this->config['host'])) {
            $host = $this->config['host'];
        }
        if ($port === 1883 && !empty($this->config['port'])) {
            $port = (int) $this->config['port'];
        }

        // 启动模式：both=TCP+WS 同进程（仅 Linux）；tcp=仅 TCP；ws=仅 WebSocket（Windows 双进程）
        $isWindows = DIRECTORY_SEPARATOR === '\\';
        $wsPort    = (int) ($this->config['ws_port'] ?? 0);
        $wsForce   = (string) ($this->config['ws_force'] ?? '0') === '1';
        $wsEnabled = $wsPort > 0 && (!$isWindows || $wsForce);
        $wantTcp   = $mode !== 'ws';
        $wantWs    = $wsEnabled && $mode !== 'tcp';

        // Workerman 限制：Windows 下一个 PHP 进程只能创建一个 worker（监听一个端口），
        // TCP 与 WebSocket 必须拆成两个独立进程，经 Redis Stream 桥接（cluster_enabled）互通。
        if ($isWindows && $mode === 'both' && $wsEnabled) {
            $this->log(1, '[Windows] Workerman 一个进程只能监听一个端口，无法同时启动 TCP + WebSocket。');
            $this->log(1, '[Windows] 请开两个 cmd 窗口分别运行：');
            $this->log(1, '[Windows]   窗口1: php think iot start --tcp-only');
            $this->log(1, '[Windows]   窗口2: php think iot start --ws-only');
            $this->log(1, '[Windows] 双进程需后台「服务设置」开启『启用集群桥接』(cluster_enabled=1) 且 Redis 可连，消息才会互通。');
            $this->log(1, '[Windows] 本进程仅启动 TCP 监听，WebSocket 请另开窗口运行 --ws-only。');
            $wantWs = false;
        }

        // 主 TCP 监听（可选 TLS）
        $context = [];
        $scheme  = 'tcp';
        if ((string) ($this->config['ssl_enabled'] ?? '0') === '1'
            && !empty($this->config['ssl_cert']) && !empty($this->config['ssl_key'])) {
            $context['ssl'] = [
                'local_cert'  => $this->config['ssl_cert'],
                'local_pk'    => $this->config['ssl_key'],
                'verify_peer' => false,
            ];
            $scheme = 'tls'; // Workerman transport 见下方 setter
        }
        if ($wantTcp) {
            $this->worker = new Worker("tcp://{$host}:{$port}", $context);
            $this->worker->name     = 'MQTTBroker';
            $this->worker->count    = 1; // 内存态，单进程保证状态一致
            $this->worker->protocol = Mqtt::class;
            if ($scheme === 'tls') {
                $this->worker->transport = 'ssl';
            }
            $this->worker->onConnect     = [$this, 'onTcpConnect'];
            $this->worker->onMessage     = [$this, 'onTcpMessage'];
            $this->worker->onClose       = [$this, 'onTcpClose'];
            $this->worker->onWorkerStart = [$this, 'onWorkerStart'];
        }

        // WebSocket 监听（MQTT over WebSocket）
        if ($wantWs) {
            $this->wsWorker = new Worker("websocket://{$host}:{$wsPort}");
            $this->wsWorker->name    = 'MQTTBroker-WS';
            $this->wsWorker->count   = 1;
            $this->wsWorker->onConnect          = [$this, 'onWsConnect'];
            $this->wsWorker->onWebSocketConnect = [$this, 'onWsHandshake'];
            $this->wsWorker->onMessage          = [$this, 'onWsMessage'];
            $this->wsWorker->onClose            = [$this, 'onTcpClose'];
            // 关键：WS 进程也须触发 onWorkerStart，否则 Bridge 不连接 Redis、不轮询总线，
            // 保留消息不恢复 → TCP 与 WS 双进程消息无法互通
            $this->wsWorker->onWorkerStart      = [$this, 'onWorkerStart'];
        }

        $parts = [];
        if ($wantTcp) {
            $parts[] = "MQTT(TCP) {$host}:{$port} ({$scheme})";
        }
        if ($wantWs) {
            $parts[] = "WebSocket {$host}:{$wsPort}";
        }
        $this->log(2, 'MQTT Broker 启动: ' . implode(' / ', $parts) . '（MQTT 3.1.1/5.0）');
        Worker::runAll();
    }

    /**
     * worker 启动：建表自愈、恢复保留消息、保活检测、$SYS 周期发布
     */
    public function onWorkerStart(Worker $worker): void
    {
        $this->stats['startTime'] = time();
        $this->isPrimary = ($worker->name === 'MQTTBroker'); // TCP 为主进程

        // 建表自愈 + 恢复保留消息
        $this->auth->ensureTables();
        $this->store->ensureTables();
        $this->ruleEngine->ensureTables();
        $this->ensureStatsTable();
        $this->retained = $this->store->loadRetained();

        // 跨进程桥接（须在 fork 后的 worker 进程内建立连接）
        if ($this->bridge->connect()) {
            Timer::add(0.05, function () {
                $this->bridge->poll([$this, 'onBridgeMessage']);
            });
            $this->log(2, "跨进程桥接已启用 node={$this->bridge->nodeId()}");
        }

        // 保活检测：超过 1.5*keepalive 无报文则断开（TCP/WS 通用）
        Timer::add(5, function () use ($worker) {
            $now = time();
            foreach ($worker->connections as $conn) {
                if (!empty($conn->mqttConnected) && !empty($conn->mqttKeepAlive)) {
                    $idle = $now - ($conn->mqttLastTime ?? $now);
                    if ($idle > $conn->mqttKeepAlive * 1.5) {
                        $this->log(1, '[保活] ' . ($conn->clientId ?? '?') . ' (ws=' . (!empty($conn->isWs) ? 'Y' : 'N') . ') 空闲 ' . $idle . 's > keepalive*1.5=' . ($conn->mqttKeepAlive * 1.5) . 's，主动断开');
                        $conn->close();
                    }
                }
            }
        });

        // $SYS 系统主题周期发布（仅主进程，避免多进程重复）
        if ($this->isPrimary && (string) ($this->config['sys_enabled'] ?? '1') === '1') {
            $interval = max(2, (int) ($this->config['sys_interval'] ?? 10));
            Timer::add($interval, [$this, 'publishSysTopics']);
        }

        // 实时指标：每个进程把本进程指标写入 Redis 共享区（TTL 10s 自动清理，进程退出即失效）
        $this->persistLocalStats();
        Timer::add(2, [$this, 'persistLocalStats']);

        // 仅主进程：聚合全部进程指标落库（DB 单行 id=1，后台仪表盘跨进程读取）
        if ($this->isPrimary) {
            $this->persistStats();
            Timer::add(2, [$this, 'persistStats']);
            // 手动发布出站队列消费（跨进程内部路由，零外部依赖）
            Timer::add(0.5, [$this, 'consumeOutbox']);
            // 场景引擎：定时触发器每分钟扫描 cron
            Timer::add(60, function () {
                SceneService::onTick();
            });
            // 传感器时序保留期清理（每天一次；按 create_at 索引删，默认保留 90 天）
            Timer::add(86400, function () {
                SensorPointService::cleanup();
            });
        }
    }

    /* ============================================================
     * TCP / WebSocket 连接层回调
     * ========================================================== */
    
    public function onTcpConnect(TcpConnection $connection): void
    {
        $connection->mqttConnected        = false;
        $connection->clientId             = null;
        $connection->mqttProperDisconnect = false;
        $connection->isWs                 = false;
        $connection->mqttPubTokens        = (int) ($this->config['publish_rate_limit'] ?? 0);
        $connection->mqttPubWindow        = time();

        // 连接数硬限流：超过上限直接拒绝，避免内存雪崩
        $max = (int) ($this->config['max_connections'] ?? 0);
        if ($max > 0 && count($this->sessions) >= $max) {
            $this->log(1, "连接数已达上限({$max})，拒绝新连接 " . ($connection->getRemoteIp() ?? ''));
            $connection->close();
        }
    }

    
    public function onWsConnect(TcpConnection $connection): void
    {
        $this->onTcpConnect($connection);
        $connection->isWs          = true;
        $connection->wsBuf         = '';
        $connection->websocketType = Websocket::BINARY_TYPE_ARRAYBUFFER; // MQTT over WS 用二进制帧
    }

    /**
     * WebSocket 握手：回应子协议 mqtt（部分客户端要求）
     */
    public function onWsHandshake(TcpConnection $connection, $httpBuffer): void
    {
        try {
            if (stripos($httpBuffer, 'sec-websocket-protocol') !== false) {
                $connection->headers = $connection->headers ?? [];
                $connection->headers[] = 'Sec-WebSocket-Protocol: mqtt';
            }
        } catch (\Throwable $e) {
        }
    }

    /**
     * TCP：$data 已由 Mqtt::decode() 解析为 ['cmd','flags','body']
     */
    public function onTcpMessage(TcpConnection $connection, $data): void
    {
        $connection->mqttLastTime = time();
        $this->handlePacket(
            $connection,
            (int) ($data['cmd'] ?? 0),
            (int) ($data['flags'] ?? 0),
            $data['body'] ?? ''
        );
    }

    /**
     * WebSocket：payload 是原始 MQTT 字节，需自行分包（ws 帧≠mqtt 报文边界）
     */
    public function onWsMessage(TcpConnection $connection, $data): void
    {
        $connection->mqttLastTime = time();
        $connection->wsBuf = ($connection->wsBuf ?? '') . $data;
        while (($len = Mqtt::input($connection->wsBuf)) !== 0) {
            if ($len < 0) {
                $connection->close();
                return;
            }
            $packet = substr($connection->wsBuf, 0, $len);
            $connection->wsBuf = substr($connection->wsBuf, $len);
            $decoded = Mqtt::decode($packet);
            $this->handlePacket(
                $connection,
                (int) ($decoded['cmd'] ?? 0),
                (int) ($decoded['flags'] ?? 0),
                $decoded['body'] ?? ''
            );
            if ($connection->wsBuf === '') {
                break;
            }
        }
    }

    
    public function onTcpClose(TcpConnection $connection): void
    {
        $cid = $connection->clientId ?? null;
        if (!$cid || !isset($this->sessions[$cid])) {
            return;
        }
        // 异常断开且有遗嘱 → 发布遗嘱
        if (empty($connection->mqttProperDisconnect) && !empty($connection->mqttWill)) {
            $w = $connection->mqttWill;
            $this->distribute($w['topic'], $w['payload'], $w['qos'], $w['retain'], $cid);
        }
        if (!empty($connection->mqttClean)) {
            unset($this->sessions[$cid]);
        } else {
            $this->sessions[$cid]['conn'] = null; // 保留订阅，标记离线
        }
        $this->logClose($cid);
        // 设备离线告警（绑定设备断开时触发，带抑制窗口）
        if (str_starts_with($cid, 'dev_')) {
            AlertService::onDeviceOffline($cid);
        }
    }

    /* ============================================================
     * 统一发送：TCP 走协议类 encode（数组）；WS 需手动 encode 成字节
     * ========================================================== */
    
    protected function rawSend(TcpConnection $connection, array $packet): void
    {
        if (!empty($connection->isWs)) {
            $connection->send(Mqtt::encode($packet));
        } else {
            $connection->send($packet);
        }
        $this->stats['sent']++;
    }

    
    protected function sendPacket(TcpConnection $connection, int $cmd, int $flags, string $body): void
    {
        $this->rawSend($connection, ['cmd' => $cmd, 'flags' => $flags, 'body' => $body]);
    }

    /* ============================================================
     * 报文分发（分包/解码已由 Mqtt 协议类完成）
     * ========================================================== */
    
    protected function handlePacket(TcpConnection $connection, int $cmd, int $flags, string $payload): void
    {
        if (!$connection->mqttConnected && $cmd !== self::CMD_CONNECT) {
            $connection->close();
            return;
        }
        switch ($cmd) {
            case self::CMD_CONNECT:      $this->handleConnect($connection, $flags, $payload); break;
            case self::CMD_PUBLISH:      $this->handlePublish($connection, $flags, $payload); break;
            case self::CMD_PUBACK:       $this->handlePubAck($connection, $payload); break;
            case self::CMD_PUBREC:       $this->handlePubRec($connection, $payload); break;
            case self::CMD_PUBREL:       $this->handlePubRel($connection, $payload); break;
            case self::CMD_PUBCOMP:      $this->handlePubComp($connection, $payload); break;
            case self::CMD_SUBSCRIBE:    $this->handleSubscribe($connection, $payload); break;
            case self::CMD_UNSUBSCRIBE:  $this->handleUnsubscribe($connection, $payload); break;
            case self::CMD_PINGREQ:      $this->sendPacket($connection, self::CMD_PINGRESP, 0, ''); break;
            case self::CMD_DISCONNECT:
                $connection->mqttProperDisconnect = true;
                $connection->close();
                break;
            case self::CMD_AUTH:
                if (($connection->mqttProto ?? 0) == 5) {
                    $this->sendPacket($connection, self::CMD_AUTH, 0, "\x00\x00");
                }
                break;
            default:
                $connection->close();
        }
    }

    /* ============================================================
     * CONNECT / CONNACK
     * ========================================================== */
    
    protected function handleConnect(TcpConnection $connection, int $flags, string $payload): void
    {
        $p = 0;
        try {
            list($protoName, $p) = $this->readStr($payload, $p);
            $level  = ord($payload[$p]); $p++;
            $cflags = ord($payload[$p]); $p++;
            $keepAlive = unpack('n', substr($payload, $p, 2))[1]; $p += 2;

            if ($level === 5) {
                if ($protoName !== 'MQTT') { $this->connAckError($connection, 5, self::RC_UNACCEPTABLE_PROTO[5]); return; }
                $version = 5;
            } elseif ($level === 4) {
                if ($protoName !== 'MQTT') { $this->connAckError($connection, 4, self::RC_UNACCEPTABLE_PROTO[4]); return; }
                $version = 4;
            } elseif ($level === 3) {
                if ($protoName !== 'MQIsdp') { $this->connAckError($connection, 4, self::RC_UNACCEPTABLE_PROTO[4]); return; }
                $version = 4;
            } else {
                $v = ($level === 5) ? 5 : 4;
                $this->connAckError($connection, $v, self::RC_UNACCEPTABLE_PROTO[$v]);
                return;
            }

            if ($version === 5) {
                list($pl, $p) = $this->readVarInt($payload, $p);
                $p += $pl; // 跳过连接属性
            }

            list($clientId, $p) = $this->readStr($payload, $p);
            if ($clientId === '') {
                $this->connAckError($connection, $version, $version === 5 ? 0x85 : 2);
                return;
            }

            $will = null;
            if ($cflags & 0x04) {
                if ($version === 5) {
                    list($wpl, $p) = $this->readVarInt($payload, $p);
                    $p += $wpl; // 跳过遗嘱属性
                }
                list($willTopic, $p)   = $this->readStr($payload, $p);
                list($willPayload, $p) = $this->readStr($payload, $p);
                $will = [
                    'topic'   => $willTopic,
                    'payload' => $willPayload,
                    'qos'     => ($cflags >> 3) & 0x03,
                    'retain'  => ($cflags & 0x20) ? 1 : 0,
                ];
            }
            $username = null;
            if ($cflags & 0x80) {
                list($username, $p) = $this->readStr($payload, $p);
            }
            $password = null;
            if ($cflags & 0x40) {
                list($password, $p) = $this->readStr($payload, $p);
            }

            // ---- 认证 ----
            $authRes = $this->auth->authenticate($username, $password, $clientId);
            if (!$authRes['ok']) {
                $rc = (stripos($authRes['reason'], 'password') !== false || stripos($authRes['reason'], 'credential') !== false)
                    ? self::RC_BAD_CREDENTIALS[$version]
                    : self::RC_NOT_AUTHORIZED[$version];
                $this->connAckError($connection, $version, $rc);
                $this->log(1, "认证失败 client={$clientId} user={$username} reason={$authRes['reason']}");
                return;
            }

            $clean = ($cflags & 0x02) ? true : false;
            $sessionPresent = 0;
            if (isset($this->sessions[$clientId])) {
                try { $this->sessions[$clientId]['conn'] && $this->sessions[$clientId]['conn']->close(); } catch (\Throwable $e) {}
                if (!$clean) {
                    $sessionPresent = 1;
                    $subs = $this->sessions[$clientId]['subscriptions'];
                } else {
                    $subs = [];
                }
            } else {
                $subs = [];
            }

            $this->sessions[$clientId] = [
                'conn'          => $connection,
                'clean'         => $clean,
                'username'      => $username,
                'subscriptions' => $subs,
                // 身份快照：投递复核（subscriberMayReceive）在离线会话上没有 conn 时要用
                'superuser'     => !empty($authRes['superuser']),
                'authReason'    => (string) ($authRes['reason'] ?? ''),
                'memberId'      => (int) ($authRes['member_id'] ?? 0),
            ];

            $connection->mqttConnected        = true;
            $connection->clientId             = $clientId;
            $connection->mqttUsername         = $username;
            $connection->mqttSuperuser        = !empty($authRes['superuser']);
            // 认证来源与会员ID：ACL 私有主题归属校验要用（device本体看 reason=ok，会员看 reason=member/ticket）
            $connection->mqttAuthReason       = (string) ($authRes['reason'] ?? '');
            $connection->mqttMemberId         = (int) ($authRes['member_id'] ?? 0);
            $connection->mqttProto            = $version;
            $connection->mqttKeepAlive        = $keepAlive > 0 ? $keepAlive : (int) ($this->config['max_keepalive'] ?? 60);
            $connection->mqttLastTime         = time();
            $connection->mqttOutId            = 1;
            $connection->mqttOutFlight        = [];
            $connection->mqttInbound          = [];
            $connection->mqttWill             = $will;
            $connection->mqttClean            = $clean;
            $connection->mqttProperDisconnect = false;

            $this->sendConnAck($connection, $sessionPresent, self::RC_ACCEPTED);
            $this->logConnect($clientId, $username, $connection->getRemoteIp());
            // 认证身份留痕：排查「连上了却订阅不了」时，第一眼要看的就是这个身份是不是自己以为的那个
            // （member/ticket=会员身份；ok=iot_auth 认证账号；anonymous-named=填了用户名但密码错/开关未开）
            $this->log(1, sprintf(
                '认证通过 client=%s user=%s 身份=%s member=%d super=%d ip=%s',
                $clientId,
                (string) $username,
                (string) ($authRes['reason'] ?? ''),
                (int) ($authRes['member_id'] ?? 0),
                !empty($authRes['superuser']) ? 1 : 0,
                $connection->getRemoteIp()
            ));

            // 会话恢复：补投离线消息 + 已存订阅的保留消息
            if (!$clean && $sessionPresent) {
                $this->flushOffline($connection, $clientId);
            }
        } catch (\Throwable $e) {
            $this->log(1, "CONNECT 解析失败：{$e->getMessage()}");
            $connection->close();
        }
    }

    
    protected function sendConnAck(TcpConnection $connection, int $sessionPresent, int $reasonCode): void
    {
        $version = $connection->mqttProto ?? 4;
        if ($version === 5) {
            $body = chr($sessionPresent ? 1 : 0) . chr($reasonCode) . "\x00";
        } else {
            $body = chr($sessionPresent ? 1 : 0) . chr($reasonCode);
        }
        $this->sendPacket($connection, self::CMD_CONNACK, 0, $body);
    }

    
    protected function connAckError(TcpConnection $connection, int $version, int $reasonCode): void
    {
        $connection->mqttProto = $version;
        $this->sendConnAck($connection, 0, $reasonCode);
        $connection->close();
    }

    /* ============================================================
     * PUBLISH 与 QoS 流
     * ========================================================== */
    
    protected function handlePublish(TcpConnection $connection, int $flags, string $payload): void
    {
        // 单连接发布速率限制（令牌桶，滑动窗口 1s）
        $rate = $connection->mqttPubTokens ?? 0;
        if ($rate > 0) {
            $now = time();
            if (($connection->mqttPubWindow ?? 0) !== $now) {
                $connection->mqttPubWindow = $now;
                $connection->mqttPubCount  = 0;
            }
            $connection->mqttPubCount = ($connection->mqttPubCount ?? 0) + 1;
            if ($connection->mqttPubCount > $rate) {
                $this->log(2, "连接 {$connection->clientId} 发布超限({$rate}/s)，丢弃消息");
                return; // 静默丢弃，避免放大风暴
            }
        }

        $qos    = ($flags >> 1) & 3;
        $retain = $flags & 1;
        $p = 0;
        list($topic, $p) = $this->readStr($payload, $p);
        $packetId = 0;
        if ($qos > 0) {
            $packetId = unpack('n', substr($payload, $p, 2))[1];
            $p += 2;
        }
        if (($connection->mqttProto ?? 4) === 5) {
            list($pl, $p) = $this->readVarInt($payload, $p);
            $p += $pl; // 跳过发布属性
        }
        $msg = substr($payload, $p);

        // ---- ACL 发布校验 ----
        if (!$this->auth->checkAcl($connection->mqttUsername ?? null, $connection->clientId, $topic, Auth::ACT_PUB, !empty($connection->mqttSuperuser), (string) ($connection->mqttAuthReason ?? ''), (int) ($connection->mqttMemberId ?? 0))) {
            // 拒绝：5.0 回带原因码的 ACK，3.1.1 静默丢弃
            if ($qos === 1) {
                $this->sendReasonPacket(self::CMD_PUBACK, $packetId, $connection->mqttProto, 0x87, $connection);
            } elseif ($qos === 2) {
                $this->sendReasonPacket(self::CMD_PUBREC, $packetId, $connection->mqttProto, 0x87, $connection);
            }
            $this->log(1, sprintf(
                '发布被 ACL 拒绝 client=%s topic=%s user=%s 身份=%s member=%d 原因=%s',
                (string) $connection->clientId,
                $topic,
                (string) ($connection->mqttUsername ?? ''),
                (string) ($connection->mqttAuthReason ?? ''),
                (int) ($connection->mqttMemberId ?? 0),
                $this->auth->lastDenyReason()
            ));
            return;
        }

        if ($qos === 1) {
            $this->sendReasonPacket(self::CMD_PUBACK, $packetId, $connection->mqttProto, 0, $connection);
            $this->distribute($topic, $msg, $qos, $retain, $connection->clientId);
        } elseif ($qos === 2) {
            $this->sendReasonPacket(self::CMD_PUBREC, $packetId, $connection->mqttProto, 0, $connection);
            $connection->mqttInbound[$packetId] = ['topic' => $topic, 'msg' => $msg, 'qos' => $qos, 'retain' => $retain];
        } else {
            $this->distribute($topic, $msg, $qos, $retain, $connection->clientId);
        }
        // 钩子：消息被 Broker 接受并分发（跳过 $SYS 内部指标主题以降低噪声）
        // qos2 的实际分发在 PUBREL 阶段，钩子改在 handlePubRel 中触发，避免重复
        if (($qos === 0 || $qos === 1) && !str_starts_with($topic, '$SYS')) {
            event('mqtt_message_published', [
                'topic'    => $topic,
                'payload'  => $msg,
                'qos'      => $qos,
                'retain'   => (bool)$retain,
                'clientId' => $connection->clientId,
            ]);
            // 设备维度日志（状态/事件上报留痕）
            DeviceLogService::log($topic, $msg, 'in');
            // 传感器结构化时序落库 + 云端阈值复核（曲线/统计/漏报兜底）
            SensorPointService::capture($topic, $msg);
            // 上线引脚表缓存（/status 里的 pins → iot_device.pins；state 报文因此不必带 pin）
            DeviceService::onStatusMessage($topic, $msg);
            // 场景引擎：设备状态/事件触发评估（动作异步入 outbox）
            SceneService::onMessagePublished($topic, $msg);
            // OTA 进度上报处理（升级完成/失败 → 告警通知）
            if (preg_match('#^/devices/([^/]+)/ota/report$#', $topic, $m)) {
                OtaService::onReport($m[1], $msg);
            }
            // 设备主动事件上报 → 告警 + 实时通知
            if (preg_match('#^/devices/([^/]+)/event$#', $topic, $m)) {
                AlertService::onDeviceEvent($m[1], $msg);
            }
            // 传感器配置回执（设备写入 NVS 后回 config_ack → 标记配置已生效）
            if (preg_match('#^/devices/([^/]+)/ack$#', $topic, $m)) {
                $ack = json_decode($msg, true);
                if (is_array($ack)) {
                    SensorConfigService::onAck($m[1], $ack);
                }
            }
            // 设备注册主题 system/device/register/{sn}：激活入网并回执
            if (preg_match('#^system/device/register/([^/]+)$#', $topic, $m)) {
                $this->handleDeviceRegister($m[1], $msg);
            }
        }
    }

    /**
     * 设备注册主题处理：system/device/register/{mac}
     *
     * 新协议（即插即用）：设备上电 publish retained 注册消息
     *   {"mac":"a0b1c2d3e4f5","type":"esp32c3_temp","model":"KF-C3-3KANG","ver":"1.0"}
     * 服务端自动创建设备入网（mac 唯一身份），无需后台预登记；
     * 旧协议兼容：payload 为裸 token 或 {"token":"..."} 时走 activate_token 激活。
     * 处理结果发布到 system/device/register/{mac}/resp（设备需预先订阅该主题，回执不 retained）。
     * 成功回执：{"code":0,"message":"ok","data":{uuid,secret(仅首次),topics,dev_type,...}}
     * 失败回执：{"code":1,"message":"错误原因","data":null}
     */
    protected function handleDeviceRegister(string $sn, string $msg): void
    {
        $decoded = json_decode($msg, true);
        $info    = is_array($decoded) ? $decoded : [];
        $token   = trim((string) ($info['token'] ?? ''));
        if ($token === '' && !is_array($decoded)) {
            $token = trim($msg); // 裸 token 字符串兼容
        }
        // 新协议：payload 携带 mac → mac 即插即用注册
        $payloadMac = str_replace([':', '-', ' '], '', strtolower((string) ($info['mac'] ?? '')));
        if ($payloadMac !== '') {
            if ($payloadMac !== $sn) {
                $out = ['code' => 1, 'message' => 'payload mac 与主题不一致', 'data' => null];
            } else {
                try {
                    $res = DeviceService::registerByMac($sn, $info);
                    $out = ['code' => 0, 'message' => 'ok', 'data' => $res];
                } catch (\Throwable $e) {
                    $out = ['code' => 1, 'message' => $e->getMessage(), 'data' => null];
                }
            }
        } else {
            // 旧协议兼容：activate_token 激活入网
            try {
                $res = DeviceService::registerViaMqtt($sn, $token);
                $out = ['code' => 0, 'message' => 'ok', 'data' => $res];
            } catch (\Throwable $e) {
                $out = ['code' => 1, 'message' => $e->getMessage(), 'data' => null];
            }
        }
        $this->log(1, "设备注册: sn={$sn} code={$out['code']} {$out['message']}");
        $this->distribute(DeviceService::registerRespTopic($sn), json_encode($out, JSON_UNESCAPED_UNICODE), 1, 0, null);
    }

    
    protected function handlePubAck(TcpConnection $connection, string $payload): void
    {
        $id = unpack('n', substr($payload, 0, 2))[1];
        unset($connection->mqttOutFlight[$id]);
    }

    
    protected function handlePubRec(TcpConnection $connection, string $payload): void
    {
        $id = unpack('n', substr($payload, 0, 2))[1];
        if (isset($connection->mqttOutFlight[$id])) {
            $this->sendReasonPacket(self::CMD_PUBREL, $id, $connection->mqttProto, 0, $connection);
            $connection->mqttOutFlight[$id]['state'] = 'pubrel';
        }
    }

    
    protected function handlePubRel(TcpConnection $connection, string $payload): void
    {
        $id = unpack('n', substr($payload, 0, 2))[1];
        if (isset($connection->mqttInbound[$id])) {
            $m = $connection->mqttInbound[$id];
            unset($connection->mqttInbound[$id]);
            $this->distribute($m['topic'], $m['msg'], $m['qos'], $m['retain'], $connection->clientId);
            // 钩子：qos2 消息在 PUBREL 确认后才真正分发，此处触发（跳过 $SYS 内部主题）
            if (!str_starts_with($m['topic'], '$SYS')) {
                event('mqtt_message_published', [
                    'topic'    => $m['topic'],
                    'payload'  => $m['msg'],
                    'qos'      => $m['qos'],
                    'retain'   => (bool)$m['retain'],
                    'clientId' => $connection->clientId,
                ]);
                DeviceLogService::log($m['topic'], $m['msg'], 'in');
                SensorPointService::capture($m['topic'], $m['msg']);
                DeviceService::onStatusMessage($m['topic'], $m['msg']);   // 同上（QoS1 重传路径）
                SceneService::onMessagePublished($m['topic'], $m['msg']);
                if (preg_match('#^/devices/([^/]+)/ota/report$#', $m['topic'], $mm)) {
                    OtaService::onReport($mm[1], $m['msg']);
                }
                if (preg_match('#^/devices/([^/]+)/event$#', $m['topic'], $mm)) {
                    AlertService::onDeviceEvent($mm[1], $m['msg']);
                }
                // 传感器配置回执（同上，QoS1 重传路径）
                if (preg_match('#^/devices/([^/]+)/ack$#', $m['topic'], $mm)) {
                    $ack = json_decode($m['msg'], true);
                    if (is_array($ack)) {
                        SensorConfigService::onAck($mm[1], $ack);
                    }
                }
                if (preg_match('#^system/device/register/([^/]+)$#', $m['topic'], $mm)) {
                    $this->handleDeviceRegister($mm[1], $m['msg']);
                }
            }
        }
        $this->sendReasonPacket(self::CMD_PUBCOMP, $id, $connection->mqttProto, 0, $connection);
    }

    
    protected function handlePubComp(TcpConnection $connection, string $payload): void
    {
        $id = unpack('n', substr($payload, 0, 2))[1];
        unset($connection->mqttOutFlight[$id]);
    }

    
    protected function sendReasonPacket(int $cmd, int $id, int $version, int $reasonCode, TcpConnection $connection): void
    {
        $body = pack('n', $id);
        if ($version === 5) {
            $body .= chr($reasonCode) . "\x00";
        }
        $this->sendPacket($connection, $cmd, ($cmd === self::CMD_PUBREL ? 2 : 0), $body);
    }

    /* ============================================================
     * SUBSCRIBE / UNSUBSCRIBE
     * ========================================================== */
    
    protected function handleSubscribe(TcpConnection $connection, string $payload): void
    {
        $p = 0;
        $packetId = unpack('n', substr($payload, $p, 2))[1];
        $p += 2;
        if (($connection->mqttProto ?? 4) === 5) {
            list($pl, $p) = $this->readVarInt($payload, $p);
            $p += $pl; // 跳过订阅属性
        }
        $granted = [];
        $cid = $connection->clientId;
        while ($p < strlen($payload)) {
            list($rawFilter, $p) = $this->readStr($payload, $p);
            $opt     = ord($payload[$p]); $p++;
            $reqQos  = $opt & 0x03;
            $noLocal = ($opt & 0x04) ? true : false;
            $rap     = ($opt & 0x08) ? true : false;
            $rh      = ($opt >> 4) & 0x03;

            // 解析共享订阅 $share/{group}/{filter}
            $share  = null;
            $filter = $rawFilter;
            if (strncmp($rawFilter, '$share/', 7) === 0) {
                $rest = substr($rawFilter, 7);
                $slash = strpos($rest, '/');
                if ($slash !== false) {
                    $share  = substr($rest, 0, $slash);
                    $filter = substr($rest, $slash + 1);
                }
            }

            // ---- ACL 订阅校验 ----
            if (!$this->auth->checkAcl($connection->mqttUsername ?? null, $cid, $filter, Auth::ACT_SUB, !empty($connection->mqttSuperuser), (string) ($connection->mqttAuthReason ?? ''), (int) ($connection->mqttMemberId ?? 0))) {
                $granted[] = ($connection->mqttProto ?? 4) === 5 ? 0x87 : 0x80; // 拒绝
                // 日志必须带「身份 + 原因」：SUBACK 128 的四种成因（匿名降级/非主人/设备不存在/ACL 兜底）
                // 在客户端看起来完全一样，只有这里能区分。
                $this->log(1, sprintf(
                    '订阅被 ACL 拒绝 client=%s topic=%s user=%s 身份=%s member=%d 原因=%s',
                    $cid,
                    $rawFilter,
                    (string) ($connection->mqttUsername ?? ''),
                    (string) ($connection->mqttAuthReason ?? ''),
                    (int) ($connection->mqttMemberId ?? 0),
                    $this->auth->lastDenyReason()
                ));
                continue;
            }

            $this->sessions[$cid]['subscriptions'][$rawFilter] = [
                'filter' => $filter, 'share' => $share,
                'qos' => $reqQos, 'noLocal' => $noLocal, 'rap' => $rap, 'rh' => $rh,
            ];
            $this->logSubscribe($cid, $rawFilter, $reqQos);
            $granted[] = $reqQos;

            // 共享订阅不投递保留消息（MQTT 规范）
            if ($share === null) {
                $this->deliverRetained($connection, $filter, $reqQos, $rh);
            }
        }
        $this->sendSubAck($connection, $packetId, $granted);
    }

    
    protected function sendSubAck(TcpConnection $connection, int $packetId, array $granted): void
    {
        $body = pack('n', $packetId);
        if (($connection->mqttProto ?? 4) === 5) {
            $body .= "\x00";
        }
        foreach ($granted as $g) {
            $body .= chr($g);
        }
        $this->sendPacket($connection, self::CMD_SUBACK, 0, $body);
    }

    
    protected function handleUnsubscribe(TcpConnection $connection, string $payload): void
    {
        $p = 0;
        $packetId = unpack('n', substr($payload, $p, 2))[1];
        $p += 2;
        if (($connection->mqttProto ?? 4) === 5) {
            list($pl, $p) = $this->readVarInt($payload, $p);
            $p += $pl;
        }
        $cid = $connection->clientId;
        while ($p < strlen($payload)) {
            list($filter, $p) = $this->readStr($payload, $p);
            unset($this->sessions[$cid]['subscriptions'][$filter]);
            $this->logUnsubscribe($cid, $filter);
        }
        $body = pack('n', $packetId);
        if (($connection->mqttProto ?? 4) === 5) {
            $body .= "\x00";
        }
        $this->sendPacket($connection, self::CMD_UNSUBACK, 0, $body);
    }

    /* ============================================================
     * 消息分发 / 投递 / 保留 / 离线
     * ========================================================== */
    /**
     * 消息分发入口：本地路由 + 跨进程桥接
     */
    protected function distribute(string $topic, string $msg, int $qos, int $retain, ?string $fromClientId): void
    {
        $this->routeMessage($topic, $msg, $qos, $retain, $fromClientId, false);
        // 桥接给其它进程（TCP/WS/多进程），由对端投递给其本地订阅者
        if ($this->bridge->isEnabled()) {
            $this->bridge->publish($topic, $msg, $qos, $retain, $fromClientId);
        }
        // 规则引擎：匹配源主题后转发到外部 Broker（如 EMQX）
        $this->ruleEngine->matchAndForward($topic, $msg, $qos);
    }

    /**
     * 桥接消息回调：仅做本地投递，不再持久化/记录/回环桥接
     */
    public function onBridgeMessage(string $topic, string $msg, int $qos, int $retain, ?string $fromClientId): void
    {
        $this->routeMessage($topic, $msg, $qos, $retain, $fromClientId, true);
    }

    /**
     * 本地路由：保留消息同步 + 本进程订阅者投递
     * @param bool $bridged 是否来自其它进程（true 时跳过 DB 持久化/日志，仅内存同步 retained）
     */
    protected function routeMessage(string $topic, string $msg, int $qos, int $retain, ?string $fromClientId, bool $bridged): void
    {
        $isSys = strncmp($topic, '$SYS', 4) === 0;

        // 保留消息处理（$SYS 不落保留）；内存双进程都同步，DB 仅源进程写
        if ($retain && !$isSys) {
            if ($msg === '') {
                unset($this->retained[$topic]);
                if (!$bridged) {
                    $this->store->deleteRetained($topic);
                }
            } else {
                $this->retained[$topic] = ['payload' => $msg, 'qos' => $qos];
                if (!$bridged) {
                    $this->store->saveRetained($topic, $msg, $qos);
                }
            }
        }

        if (!$isSys && !$bridged) {
            $this->stats['recv']++;
            $source = ((string) $fromClientId === 'manual') ? 'manual' : 'client';
            $this->logMessage((string) $fromClientId, $topic, $msg, $qos, $retain, $source);
        }

        // 共享订阅分组收集： groupKey => [ [cid,conn,sub], ... ]
        $sharedGroups = [];

        // 私有主题（/devices/、/notify/）投递复核开关：订阅过滤器可能含通配符（# / +），
        // 订阅时刻无法表达归属，必须在投递时按「真实主题」再校验一次
        $recheckPrivate = $this->isPrivateTopic($topic);

        foreach ($this->sessions as $cid => $sess) {
            $conn = $sess['conn'] ?? null;
            foreach ($sess['subscriptions'] as $sub) {
                if (!$this->topicMatch($sub['filter'], $topic)) {
                    continue;
                }
                // ★ 防 `#` 通配符订阅绕过：订阅了 # 也不代表有权收到 /devices/{别人的}/state
                if ($recheckPrivate && !$this->subscriberMayReceive($conn, $sess, (string) $cid, $topic)) {
                    continue;
                }
                $effQos = min($qos, $sub['qos']);

                // 共享订阅：仅在线成员参与，稍后 round-robin 选一个
                if ($sub['share'] !== null) {
                    if ($conn) {
                        $gk = $sub['share'] . '|' . $sub['filter'];
                        $sharedGroups[$gk][] = ['cid' => $cid, 'conn' => $conn, 'qos' => $effQos];
                    }
                    continue;
                }

                // noLocal：不投递给发布者自身
                if (!empty($sub['noLocal']) && $cid === $fromClientId) {
                    continue;
                }

                if (!$conn) {
                    // 离线会话：clean=false 且 QoS>0 入离线队列
                    if (empty($sess['clean']) && $effQos > 0 && !$isSys) {
                        $this->store->queueOffline($cid, $topic, $msg, $effQos);
                    }
                    continue;
                }
                $this->deliver($conn, $topic, $msg, $effQos, $retain, $cid);
            }
        }

        // 共享订阅投递：每组按 round-robin 选一个成员
        foreach ($sharedGroups as $gk => $members) {
            $n = count($members);
            if ($n === 0) {
                continue;
            }
            $idx = ($this->rrIndex[$gk] ?? 0) % $n;
            $this->rrIndex[$gk] = $idx + 1;
            $m = $members[$idx];
            $this->deliver($m['conn'], $topic, $msg, $m['qos'], $retain, $m['cid']);
        }
    }

    
    /** 私有命名空间判断（带归属语义的主题前缀） */
    protected function isPrivateTopic(string $topic): bool
    {
        return strncmp($topic, '/devices/', 9) === 0 || strncmp($topic, '/notify/', 8) === 0;
    }

    /**
     * 投递复核：订阅者是否有权接收该「真实主题」（不含通配符）
     *
     * 存在的原因：SUBSCRIBE 校验看到的是「过滤器」，而 `#`、`+/...` 这类通配符不以
     * /devices/ 开头 → 绕过 Auth::checkPrivateTopic() 的硬归属分支，落到 acl_default 兜底。
     * 一旦兜底为「允许」，订阅 `#` 的匿名客户端就能收到全站设备消息 → 必须在投递时复核。
     *
     * @param TcpConnection|null $conn 在线连接（离线会话为 null，改用会话身份快照）
     * @param array              $sess 会话条目（含 username/superuser/authReason/memberId）
     */
    protected function subscriberMayReceive(?TcpConnection $conn, array $sess, string $cid, string $topic): bool
    {
        if ($conn) {
            $username   = $conn->mqttUsername ?? null;
            $clientId   = (string) ($conn->clientId ?? $cid);
            $superuser  = !empty($conn->mqttSuperuser);
            $authReason = (string) ($conn->mqttAuthReason ?? '');
            $memberId   = (int) ($conn->mqttMemberId ?? 0);
        } else {
            $username   = $sess['username'] ?? null;
            $clientId   = $cid;
            $superuser  = !empty($sess['superuser']);
            $authReason = (string) ($sess['authReason'] ?? '');
            $memberId   = (int) ($sess['memberId'] ?? 0);
        }
        return $this->auth->checkAcl($username, $clientId, $topic, Auth::ACT_SUB, $superuser, $authReason, $memberId);
    }

    protected function deliver(TcpConnection $conn, string $topic, string $msg, int $qos, int $retain, string $subClientId): void
    {
        if ($qos === 0) {
            $this->rawSend($conn, $this->buildPublish($conn->mqttProto, $topic, $msg, 0, 0, $retain, false));
            return;
        }
        $id = $this->nextId($conn);
        $conn->mqttOutFlight[$id] = ['state' => 'publish'];
        $this->rawSend($conn, $this->buildPublish($conn->mqttProto, $topic, $msg, $qos, $id, $retain, false));
    }

    
    protected function deliverRetained(TcpConnection $conn, string $filter, int $reqQos, int $rh): void
    {
        if ($rh === 2) {
            return; // RetainHandling=2：不发送保留消息
        }
        foreach ($this->retained as $topic => $r) {
            if (!$this->topicMatch($filter, $topic)) {
                continue;
            }
            // 保留消息同样按真实主题复核：订阅 # 时不能靠 retained 兜出私有主题的历史数据
            if ($this->isPrivateTopic($topic)
                && !$this->subscriberMayReceive($conn, [], (string) ($conn->clientId ?? ''), $topic)) {
                continue;
            }
            $effQos = min($r['qos'], $reqQos);
            $this->deliver($conn, $topic, $r['payload'], $effQos, 1, $conn->clientId);
        }
    }

    /**
     * 会话重连：补投离线消息
     */
    protected function flushOffline(TcpConnection $conn, string $clientId): void
    {
        $msgs = $this->store->popOffline($clientId);
        foreach ($msgs as $m) {
            // 兜底：离线队列里可能残留越权的私有主题消息（历史入队），补投前再复核一次
            if ($this->isPrivateTopic((string) $m['topic'])
                && !$this->subscriberMayReceive($conn, [], $clientId, (string) $m['topic'])) {
                continue;
            }
            $this->deliver($conn, $m['topic'], $m['payload'], (int) $m['qos'], 0, $clientId);
        }
        if (!empty($msgs)) {
            $this->log(2, "补投离线消息 client={$clientId} count=" . count($msgs));
        }
    }

    
    protected function buildPublish(int $version, string $topic, string $msg, int $qos, int $packetId, int $retain, bool $dup): array
    {
        $flags = ($dup ? 8 : 0) | ($qos << 1) | ($retain ? 1 : 0);
        $body  = $this->writeStr($topic);
        if ($qos > 0) {
            $body .= pack('n', $packetId);
        }
        if ($version === 5) {
            $body .= "\x00"; // 属性长度 0
        }
        $body .= $msg;
        return ['cmd' => self::CMD_PUBLISH, 'flags' => $flags, 'body' => $body];
    }

    /* ============================================================
     * $SYS 系统主题
     * ========================================================== */
    
    public function publishSysTopics(): void
    {
        $agg = $this->aggregatedStats(); // 全局聚合数（多进程模式下所有进程之和）
        $uptime = time() - ($this->stats['startTime'] ?: time());
        $metrics = [
            '$SYS/broker/uptime'            => (string) $uptime,
            '$SYS/broker/clients/connected' => (string) $agg['clients_online'],
            '$SYS/broker/clients/total'     => (string) $agg['clients_total'],
            '$SYS/broker/subscriptions/count' => (string) $agg['subscriptions'],
            '$SYS/broker/messages/received' => (string) $agg['messages_received'],
            '$SYS/broker/messages/sent'     => (string) $agg['messages_sent'],
            '$SYS/broker/retained/count'    => (string) $agg['retained'],
            '$SYS/broker/timestamp'         => (string) time(),
        ];
        foreach ($metrics as $t => $v) {
            $this->distribute($t, $v, 0, 0, null);
        }
    }

    
    protected function countSubscriptions(): int
    {
        $c = 0;
        foreach ($this->sessions as $s) {
            $c += count($s['subscriptions']);
        }
        return $c;
    }

    /* ============================================================
     * 实时指标快照（后台仪表盘跨进程读取，DB 单行 id=1）
     * ========================================================== */
    
    protected function ensureStatsTable(): void
    {
        try {
            Db::execute("CREATE TABLE IF NOT EXISTS `wxapp_iot_stats` (
                `id` int unsigned NOT NULL,
                `uptime` int DEFAULT '0',
                `clients_online` int DEFAULT '0',
                `clients_total` int DEFAULT '0',
                `subscriptions` int DEFAULT '0',
                `messages_received` int DEFAULT '0',
                `messages_sent` int DEFAULT '0',
                `retained` int DEFAULT '0',
                `timestamp` int DEFAULT '0',
                PRIMARY KEY (`id`)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4");
        } catch (\Throwable $e) {
        }
    }

    /**
     * 本进程运行指标（连接数/订阅数/收发数/保留数）
     */
    protected function localStats(): array
    {
        $online = 0;
        foreach ($this->sessions as $sess) {
            if (!empty($sess['conn'])) {
                $online++;
            }
        }
        return [
            'clients_online'    => $online,
            'clients_total'     => count($this->sessions),
            'subscriptions'     => $this->countSubscriptions(),
            'messages_received' => $this->stats['recv'],
            'messages_sent'     => $this->stats['sent'],
            'retained'          => count($this->retained),
        ];
    }

    /**
     * 把本进程指标写入 Redis 共享区（TTL 10s）。所有进程周期调用。
     */
    public function persistLocalStats(): void
    {
        $this->bridge->writeStats($this->localStats());
    }

    /**
     * 聚合后的全局指标：启用桥接时汇总所有进程，否则退化为本进程
     */
    protected function aggregatedStats(): array
    {
        if ($this->bridge->isEnabled()) {
            return $this->bridge->aggregateStats();
        }
        return $this->localStats();
    }

    /**
     * 将聚合后的全局指标写入 DB（单行 id=1）。供后台实时监控仪表盘轮询读取。
     */
    public function persistStats(): void
    {
        $this->persistLocalStats(); // 先刷新本进程指标，再聚合
        $agg = $this->aggregatedStats();
        $data = [
            'uptime'            => time() - ($this->stats['startTime'] ?: time()),
            'clients_online'    => $agg['clients_online'],
            'clients_total'     => $agg['clients_total'],
            'subscriptions'     => $agg['subscriptions'],
            'messages_received' => $agg['messages_received'],
            'messages_sent'     => $agg['messages_sent'],
            'retained'          => $agg['retained'],
            'timestamp'         => time(),
        ];
        try {
            $row = Db::name('iot_stats')->where('id', 1)->find();
            if ($row) {
                Db::name('iot_stats')->where('id', 1)->update($data);
            } else {
                Db::name('iot_stats')->insert(array_merge(['id' => 1], $data));
            }
        } catch (\Throwable $e) {
        }
    }

    /**
     * 消费手动发布出站队列：取出待投递消息并经内部路由分发给本地/桥接订阅者。
     * 后台"发布消息"与对外 REST API 均通过此通道投递，零外部依赖（不依赖 workerman/mqtt）。
     */
    public function consumeOutbox(): void
    {
        $rows = $this->store->popOutbox(100);
        if (empty($rows)) {
            return;
        }
        $ids = [];
        foreach ($rows as $r) {
            $ids[] = $r['id'];
            $this->distribute((string) $r['topic'], (string) $r['payload'], (int) $r['qos'], (int) $r['retain'], 'manual');
        }
        $this->store->markOutboxSent($ids);
    }

    /* ============================================================
     * 工具：字符串 / 变长整数 / 主题匹配
     * ========================================================== */
    
    protected function readStr(string $buf, int $pos): array
    {
        $len = unpack('n', substr($buf, $pos, 2))[1];
        $pos += 2;
        $str = substr($buf, $pos, $len);
        $pos += $len;
        return [$str, $pos];
    }

    
    protected function writeStr(string $str): string
    {
        return pack('n', strlen($str)) . $str;
    }

    
    protected function readVarInt(string $buf, int $pos): array
    {
        $multiplier = 1;
        $value      = 0;
        $i          = 0;
        do {
            $byte = ord($buf[$pos + $i]);
            $value += ($byte & 0x7F) * $multiplier;
            $multiplier *= 128;
            $i++;
            if ($i > 4) {
                break;
            }
        } while (($byte & 0x80) !== 0);
        return [$value, $pos + $i];
    }

    
    protected function writeVarInt(int $value): string
    {
        $bytes = '';
        do {
            $byte = $value % 128;
            $value = intdiv($value, 128);
            if ($value > 0) {
                $byte |= 0x80;
            }
            $bytes .= chr($byte);
        } while ($value > 0);
        return $bytes;
    }

    
    protected function topicMatch(string $filter, string $topic): bool
    {
        if ($filter === $topic) {
            return true;
        }
        $f = explode('/', $filter);
        $t = explode('/', $topic);
        foreach ($f as $i => $seg) {
            if ($seg === '#') {
                return true;
            }
            if ($seg === '+') {
                if (!isset($t[$i])) {
                    return false;
                }
                continue;
            }
            if (!isset($t[$i]) || $seg !== $t[$i]) {
                return false;
            }
        }
        return count($f) === count($t);
    }

    
    protected function nextId(TcpConnection $conn): int
    {
        $id = $conn->mqttOutId;
        $conn->mqttOutId = ($id >= 65535) ? 1 : $id + 1;
        return $id;
    }

    /* ============================================================
     * 配置与日志
     * ========================================================== */
    
    protected function loadConfig(): array
    {
        try {
            $saved = AddonService::config('iot');
        } catch (\Throwable $e) {
            $saved = [];
        }
        $defaults = [
            'port'            => 1883,
            'host'            => '0.0.0.0',
            'ws_port'         => 8083,
            'ws_force'        => '0',   // Windows 开发机强制启用 WebSocket（1=开启，生产 Linux 恒启用）
            'ssl_enabled'     => '0',
            'ssl_cert'        => '',
            'ssl_key'         => '',
            'allow_anonymous' => '1',
            'acl_enabled'     => '0',
            'acl_default'     => 'allow',
            'max_keepalive'   => 60,
            'sys_enabled'     => '1',
            'sys_interval'    => 10,
            'persist_retain'  => '1',
            'persist_offline' => '1',
            'offline_limit'   => 1000,
            'max_connections' => 5000,   // 单进程连接数硬上限，超出拒绝新连接防止雪崩
            'publish_rate_limit' => 50,  // 单连接每秒最大发布消息数（令牌桶，0=不限）
            'cluster_enabled' => '0',
            'rule_enabled'     => '1',
            'redis_host'      => '127.0.0.1',
            'redis_port'      => 6379,
            'redis_db'        => 0,
            'redis_password'  => '',
            'redis_prefix'    => 'iot',
            'log_level'       => '1',
        ];
        return array_merge($defaults, is_array($saved) ? $saved : []);
    }

    
    protected function log(int $level, string $msg): void
    {
        $cfgLevel = (int) ($this->config['log_level'] ?? '1');
        if ($cfgLevel === 0 || $level > $cfgLevel) {
            return;
        }
        Log::info("[iot] {$msg}");
    }

    /* ============================================================
     * 进程管理（供 iot stop / iot restart 使用；纯 PHP，不依赖 shell_exec/pgrep）
     * ========================================================== */

    /**
     * Broker pid 文件绝对路径（start / stop / restart 必须同一处，避免路径不一致）
     */
    public static function pidFilePath(): string
    {
        return rtrim((string) runtime_path(), DIRECTORY_SEPARATOR) . DIRECTORY_SEPARATOR . 'iot.pid';
    }

    /**
     * 当前插件配置（供 iot status 展示；读失败返回 null，以便区分「读取失败」与「键缺失」）
     */
    public static function currentConfig(): ?array
    {
        try {
            return AddonService::config('iot') ?: [];
        } catch (\Throwable $e) {
            return null;
        }
    }

    /**
     * 本 Broker 实际监听的端口（停止/重启后复核端口是否真的释放）
     *
     * @return int[]
     */
    public static function configuredPorts(): array
    {
        $saved = self::currentConfig() ?? [];
        $ports = [
            (int) ($saved['port'] ?? 1883) ?: 1883,
            (int) ($saved['ws_port'] ?? 8083),
        ];
        return array_values(array_unique(array_filter($ports, static fn (int $p): bool => $p > 0)));
    }

    /**
     * 端口是否仍被占用（true=占用）
     */
    public static function portBusy(int $port): bool
    {
        set_error_handler(static fn (): bool => true);
        $server = stream_socket_server("tcp://0.0.0.0:{$port}", $errno, $error);
        restore_error_handler();
        if ($server === false) {
            return true;
        }
        fclose($server);
        return false;
    }

    /**
     * 端口复核结果（供命令输出）
     *
     * @return array{lines:string[],busy:int[]}
     */
    public static function portReport(): array
    {
        $lines = [];
        $busy  = [];
        foreach (self::configuredPorts() as $port) {
            if (self::portBusy($port)) {
                $busy[]  = $port;
                $lines[] = "端口 {$port} 仍被占用";
            } else {
                $lines[] = "端口 {$port} 已释放";
            }
        }
        return ['lines' => $lines, 'busy' => $busy];
    }

    /** /proc 扫描：pid => 进程标题 */
    protected static function scanProcTitle(): array
    {
        $list = [];
        if (!is_dir('/proc')) {
            return $list;
        }
        foreach ((array) glob('/proc/[0-9]*/cmdline') as $file) {
            $pid = (int) basename(dirname($file));
            if ($pid === getmypid()) {
                continue;
            }
            $cmd = (string) @file_get_contents($file);
            if ($cmd === '') {
                continue;
            }
            $list[$pid] = trim(str_replace("\0", ' ', $cmd));
        }
        return $list;
    }

    /** 读父进程 pid（/proc/{pid}/stat 的 comm 字段可能含空格/右括号，须从最后一个 ')' 切） */
    protected static function ppidOf(int $pid): int
    {
        return (int) (self::statFields($pid)[1] ?? 0);
    }

    /**
     * 读 /proc/{pid}/stat 中 comm 之后的字段（下标 0 = state，即 stat 的第 3 个字段）
     *
     * @return string[]
     */
    protected static function statFields(int $pid): array
    {
        $stat = (string) @file_get_contents('/proc/' . $pid . '/stat');
        $pos  = strrpos($stat, ')');
        if ($pos === false) {
            return [];
        }
        return explode(' ', trim(substr($stat, $pos + 2)));
    }

    /**
     * 进程启动时刻（unix 时间戳，取不到返回 0）。
     * starttime = stat 第 22 个字段（= comm 之后的下标 19），单位是时钟滴答（Linux 默认 HZ=100）。
     */
    public static function processStartTime(int $pid): int
    {
        $fields = self::statFields($pid);
        $ticks  = (int) ($fields[19] ?? 0);
        if ($ticks <= 0) {
            return 0;
        }
        $uptime = (float) @file_get_contents('/proc/uptime');
        if ($uptime <= 0) {
            return 0;
        }
        return (int) (time() - ($uptime - $ticks / 100));
    }

    /**
     * 采集 Broker 相关进程。
     *
     * ⚠️ Workerman 启动后会用 cli_set_process_title() 改写进程标题：
     *    master → 「WorkerMan: master process  start_file=/站点/think」
     *    worker → 「WorkerMan: worker process  MQTTBroker tcp://0.0.0.0:1883」
     * 因此 `pgrep -f 'iot start'` / `pkill -f iot:start` 找不到任何进程，
     * 必须按标题特征匹配（这也是 pid 文件丢失后唯一的救命通道）。
     *
     * @return array<int,array{role:string,cmd:string}> pid => ['role' => master|worker|orphan, 'cmd' => 标题]
     */
    public static function findBrokerProcesses(): array
    {
        $titles = self::scanProcTitle();
        if (empty($titles)) {
            return [];
        }
        $root    = rtrim(str_replace('\\', '/', (string) app()->getRootPath()), '/');
        $filePid = (int) trim((string) @file_get_contents(self::pidFilePath()));

        // ① worker：标题前缀固定为 Workerman: worker process + 我们设置的 name（MQTTBroker / MQTTBroker-WS）
        $workers  = [];
        $ppidList = [];
        foreach ($titles as $pid => $cmd) {
            if (str_contains($cmd, 'MQTTBroker')) {
                $ppid = self::ppidOf($pid);
                $workers[$pid] = ['ppid' => $ppid, 'cmd' => $cmd];
                $ppidList[]    = $ppid;
            }
        }

        // ② master：pidFile 优先；pidFile 丢失时按「标题特征 + 存在我们的 worker 子进程」双重确认，
        //    避免误伤同一站点根下的其它 Workerman 服务（它们的 start_file 也是 .../think）
        $found = [];
        foreach ($titles as $pid => $cmd) {
            $norm      = str_replace('\\', '/', $cmd);
            $fromFile  = ($filePid > 0 && $pid === $filePid);
            // 标题尚未被改写的短暂窗口：原始命令行含插件命令（iot:start / iot start / iot restart…）
            $rawStart  = (bool) preg_match('#iot[: ](start|stop|restart|status)#', $norm);
            $titleOurs = str_contains($norm, 'WorkerMan') || str_contains($norm, 'start_file=');
            $rootMatch = $root !== '' && str_contains($norm, $root) && str_contains($norm, 'start_file=');
            $hasWorker = in_array($pid, $ppidList, true);

            if (!$fromFile && !$rawStart && !$rootMatch) {
                continue;   // 与本插件无关
            }
            if (!$titleOurs && !$hasWorker) {
                continue;   // pid 已被系统回收给别的程序 → 绝不碰
            }
            if (!$fromFile && !$hasWorker) {
                continue;   // 同站点根下其它 Workerman 服务 → 不误伤
            }
            $found[$pid] = ['role' => 'master', 'cmd' => $cmd];
        }

        // ③ worker 归类：master 已死的 worker 会成为孤儿并继续持有监听端口，必须能杀到
        foreach ($workers as $pid => $w) {
            if (isset($found[$pid])) {
                continue;
            }
            $found[$pid] = [
                'role' => isset($found[$w['ppid']]) ? 'worker' : 'orphan',
                'cmd'  => $w['cmd'],
            ];
        }
        return $found;
    }

    /**
     * 停止进程：SIGINT 优雅退出 → 超时 SIGKILL 兜底 → 复核存活。
     * （Workerman 自带的 stop 在失败时只写一行日志就退出，看不到、也没兜底）
     *
     * @param int[] $pids
     * @return array{ok:bool,supported:bool,alive:int[],steps:string[]}
     */
    public static function stopProcesses(array $pids, int $timeout = 8, bool $force = false): array
    {
        $result = ['ok' => false, 'supported' => true, 'alive' => [], 'steps' => []];
        if (!function_exists('posix_kill')) {
            $result['supported'] = false;
            return $result;
        }
        $pids = array_values(array_unique(array_filter(array_map('intval', $pids))));
        $pids = array_values(array_filter($pids, static fn (int $p): bool => $p > 0 && $p !== getmypid()));
        if (empty($pids)) {
            $result['ok'] = true;
            return $result;
        }

        // ① 发停止信号（2=SIGINT 优雅；9=SIGKILL 强杀）
        foreach ($pids as $pid) {
            if (@posix_kill($pid, $force ? 9 : 2)) {
                $result['steps'][] = $force
                    ? "PID {$pid} 已发送 SIGKILL（强杀）"
                    : "PID {$pid} 已发送 SIGINT（优雅停止）";
            } else {
                $result['steps'][] = "PID {$pid} 发信号失败（可能已消失或无权限）";
            }
        }

        // ② 等优雅退出
        $deadline = time() + max(1, $timeout);
        while (time() < $deadline) {
            if (empty(array_filter($pids, static fn (int $p): bool => @posix_kill($p, 0)))) {
                break;
            }
            usleep(200000);
        }

        // ③ 超时强杀兜底
        foreach ($pids as $pid) {
            if (@posix_kill($pid, 0)) {
                @posix_kill($pid, 9);
                $result['steps'][] = "PID {$pid} 等待超时，已强杀（SIGKILL）";
            }
        }
        usleep(600000);

        // ④ 复核
        foreach ($pids as $pid) {
            if (@posix_kill($pid, 0)) {
                $result['alive'][] = $pid;
            }
        }
        $result['ok'] = empty($result['alive']);
        return $result;
    }

    /* ---------- DB 落库（全部防御式） ---------- */
    
    protected function logConnect(string $clientId, $username, string $ip): void
    {
        try {
            Db::name('iot_connection')->insert([
                'client_id'   => $clientId,
                'username'    => $username,
                'ip'          => $ip,
                'status'      => 1,
                'create_at'   => time(),
                'update_at'   => time(),
            ]);
        } catch (\Throwable $e) {
            $this->log(1, "连接日志写入失败：{$e->getMessage()}");
        }
    }

    
    protected function logClose(string $clientId): void
    {
        try {
            Db::name('iot_connection')
                ->where('client_id', $clientId)
                ->order('id', 'desc')
                ->limit(1)
                ->update(['status' => 0, 'update_at' => time()]);
        } catch (\Throwable $e) {
        }
    }

    
    protected function logSubscribe(string $clientId, string $topic, int $qos): void
    {
        try {
            $exists = Db::name('iot_topic')->where('client_id', $clientId)->where('topic', $topic)->find();
            if ($exists) {
                Db::name('iot_topic')->where('id', $exists['id'])->update(['qos' => $qos, 'create_at' => time()]);
            } else {
                Db::name('iot_topic')->insert([
                    'client_id'   => $clientId,
                    'topic'       => $topic,
                    'qos'         => $qos,
                    'create_at'   => time(),
                ]);
            }
        } catch (\Throwable $e) {
        }
    }

    
    protected function logUnsubscribe(string $clientId, string $topic): void
    {
        try {
            Db::name('iot_topic')->where('client_id', $clientId)->where('topic', $topic)->delete();
        } catch (\Throwable $e) {
        }
    }

    
    protected function logMessage(string $clientId, string $topic, string $msg, int $qos, int $retain, string $source = 'client'): void
    {
        $lvl = (int) ($this->config['log_level'] ?? '1');
        if ($lvl < 2) {
            return;
        }
        try {
            Db::name('iot_message')->insert([
                'client_id'   => $clientId,
                'topic'       => $topic,
                'payload'     => $msg,
                'qos'         => $qos,
                'retain'      => $retain ? 1 : 0,
                'source'      => $source,
                'create_at'   => time(),
            ]);
        } catch (\Throwable $e) {
        }
    }
}