<?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\mqttbroker\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\mqttbroker\protocol\Mqtt;
use addon\mqttbroker\service\Bridge;
use addon\mqttbroker\service\Forwarder;
use addon\mqttbroker\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): void
    {
        $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'];
        }

        // 主 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
        }
        $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）
        // 注意：Workerman 在 Windows 下不支持“单文件多监听”，故仅在非 Windows(生产 Linux) 启用。
        $wsPort    = (int) ($this->config['ws_port'] ?? 0);
        $isWindows = DIRECTORY_SEPARATOR === '\\';
        if ($wsPort > 0 && !$isWindows) {
            $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'];
        }

        $tip = "MQTT Broker 启动于 {$host}:{$port}（{$scheme}, MQTT 3.1.1/5.0）";
        if ($wsPort > 0 && !$isWindows) {
            $tip .= "，WebSocket 于 {$host}:{$wsPort}";
        } elseif ($wsPort > 0 && $isWindows) {
            $tip .= "（Windows 单进程调试：WebSocket 未启用，生产 Linux 下自动启用）";
        }
        $this->log(2, $tip);
        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 无报文则断开
        Timer::add(5, function () use ($worker) {
            $now = time();
            foreach ($worker->connections as $conn) {
                if (!empty($conn->mqttConnected) && !empty($conn->mqttKeepAlive)) {
                    if ($now - ($conn->mqttLastTime ?? $now) > $conn->mqttKeepAlive * 1.5) {
                        $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']);
        }

        // 实时指标快照：周期写入 DB（后台仪表盘跨进程读取），仅主进程
        if ($this->isPrimary) {
            $this->persistStats();
            Timer::add(2, [$this, 'persistStats']);
            // 手动发布出站队列消费（跨进程内部路由，零外部依赖）
            Timer::add(0.5, [$this, 'consumeOutbox']);
        }
    }

    /* ============================================================
     * 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);
    }

    /* ============================================================
     * 统一发送：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,
            ];

            $connection->mqttConnected        = true;
            $connection->clientId             = $clientId;
            $connection->mqttUsername         = $username;
            $connection->mqttSuperuser        = !empty($authRes['superuser']);
            $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());

            // 会话恢复：补投离线消息 + 已存订阅的保留消息
            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))) {
            // 拒绝：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, "发布被 ACL 拒绝 client={$connection->clientId} topic={$topic}");
            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,
            ]);
        }
    }

    
    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,
                ]);
            }
        }
        $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))) {
                $granted[] = ($connection->mqttProto ?? 4) === 5 ? 0x87 : 0x80; // 拒绝
                $this->log(1, "订阅被 ACL 拒绝 client={$cid} topic={$rawFilter}");
                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 = [];

        foreach ($this->sessions as $cid => $sess) {
            $conn = $sess['conn'] ?? null;
            foreach ($sess['subscriptions'] as $sub) {
                if (!$this->topicMatch($sub['filter'], $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 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)) {
                $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) {
            $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
    {
        $online = 0;
        foreach ($this->sessions as $sess) {
            if (!empty($sess['conn'])) {
                $online++;
            }
        }
        $uptime = time() - ($this->stats['startTime'] ?: time());
        $metrics = [
            '$SYS/broker/uptime'            => (string) $uptime,
            '$SYS/broker/clients/connected' => (string) $online,
            '$SYS/broker/clients/total'     => (string) count($this->sessions),
            '$SYS/broker/subscriptions/count' => (string) $this->countSubscriptions(),
            '$SYS/broker/messages/received' => (string) $this->stats['recv'],
            '$SYS/broker/messages/sent'     => (string) $this->stats['sent'],
            '$SYS/broker/retained/count'    => (string) count($this->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_mqttbroker_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) {
        }
    }

    /**
     * 将当前运行指标写入 DB（单行）。供后台实时监控仪表盘轮询读取。
     */
    public function persistStats(): void
    {
        $online = 0;
        foreach ($this->sessions as $sess) {
            if (!empty($sess['conn'])) {
                $online++;
            }
        }
        $data = [
            'uptime'            => time() - ($this->stats['startTime'] ?: time()),
            '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),
            'timestamp'         => time(),
        ];
        try {
            $row = Db::name('mqttbroker_stats')->where('id', 1)->find();
            if ($row) {
                Db::name('mqttbroker_stats')->where('id', 1)->update($data);
            } else {
                Db::name('mqttbroker_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('mqttbroker');
        } catch (\Throwable $e) {
            $saved = [];
        }
        $defaults = [
            'port'            => 1883,
            'host'            => '0.0.0.0',
            'ws_port'         => 8083,
            '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'    => 'mqttbroker',
            '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("[mqttbroker] {$msg}");
    }

    /* ---------- DB 落库（全部防御式） ---------- */
    
    protected function logConnect(string $clientId, $username, string $ip): void
    {
        try {
            Db::name('mqttbroker_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('mqttbroker_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('mqttbroker_topic')->where('client_id', $clientId)->where('topic', $topic)->find();
            if ($exists) {
                Db::name('mqttbroker_topic')->where('id', $exists['id'])->update(['qos' => $qos, 'create_at' => time()]);
            } else {
                Db::name('mqttbroker_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('mqttbroker_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('mqttbroker_message')->insert([
                'client_id'   => $clientId,
                'topic'       => $topic,
                'payload'     => $msg,
                'qos'         => $qos,
                'retain'      => $retain ? 1 : 0,
                'source'      => $source,
                'create_at'   => time(),
            ]);
        } catch (\Throwable $e) {
        }
    }
}