<?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;

/**
 * 跨进程消息桥接（基于 Redis Stream）
 *
 * 背景：Workerman 每个 Worker 是独立进程，内存不共享。要让 TCP 监听进程与
 * WebSocket 监听进程（以及 count>1 的多进程）之间互通消息，需要一个共享总线。
 *
 * 方案：
 *   - 发布：每条消息 XADD 到 Redis Stream（带来源 nodeId，用于去重）。
 *   - 消费：每个进程用非阻塞 XREAD 周期轮询该 Stream（不阻塞事件循环），
 *           收到非自身产生的消息后投递给本进程的本地订阅者。
 *   - Stream 通过 MAXLEN 近似裁剪，避免无限增长。
 *
 * 开关：cluster_enabled=1 且 Redis 可连时启用；否则 isEnabled()=false，
 *       Broker 退化为单进程本地路由，功能不受影响。
 *
 * 已知限制：跨进程的共享订阅($share/)会在每个进程各自 round-robin，
 *           若同一共享组成员分布在不同进程，可能重复投递（同进程内正确）。
 */
class Bridge
{
    /** @var array */
    protected $config;

    /** @var bool */
    protected $enabled = false;

    /** @var \Redis|null */
    protected $redis;

    /** @var string 本进程唯一标识 */
    protected $nodeId;

    /** @var string Stream key */
    protected $stream;

    /** @var string 读取游标 */
    protected $lastId = '0-0';

    /** @var int Stream 近似最大长度 */
    protected $maxlen;

    
    public function __construct(array $config)
    {
        $this->config = $config;
        $this->nodeId = getmypid() . '-' . substr(md5(uniqid('', true)), 0, 6);
        $prefix       = $config['redis_prefix'] ?? 'mqttbroker';
        $this->stream = $prefix . ':bus';
        $this->maxlen = (int) ($config['bridge_maxlen'] ?? 10000);
    }

    
    public function isEnabled(): bool
    {
        return $this->enabled;
    }

    
    public function nodeId(): string
    {
        return $this->nodeId;
    }

    /**
     * 连接 Redis 并初始化游标（须在 worker 进程内调用，即 onWorkerStart）
     */
    public function connect(): bool
    {
        if ((string) ($this->config['cluster_enabled'] ?? '0') !== '1') {
            $this->enabled = false;
            return false;
        }
        if (!class_exists(\Redis::class)) {
            $this->enabled = false;
            return false;
        }
        try {
            $r = new \Redis();
            $r->connect($this->config['redis_host'] ?? '127.0.0.1', (int) ($this->config['redis_port'] ?? 6379), 1.5);
            if (!empty($this->config['redis_password'])) {
                $r->auth((string) $this->config['redis_password']);
            }
            if (isset($this->config['redis_db']) && $this->config['redis_db'] !== '') {
                $r->select((int) $this->config['redis_db']);
            }
            $r->ping();
            $this->redis = $r;
            // 从当前末尾开始消费，忽略历史积压
            $last = $r->xRevRange($this->stream, '+', '-', 1);
            $this->lastId = $last ? (string) array_key_first($last) : '0-0';
            $this->enabled = true;
        } catch (\Throwable $e) {
            $this->enabled = false;
        }
        return $this->enabled;
    }

    /**
     * 将消息投递到总线（供其它进程消费）
     */
    public function publish(string $topic, string $msg, int $qos, int $retain, ?string $from): void
    {
        if (!$this->enabled) {
            return;
        }
        try {
            $payload = json_encode([
                'n' => $this->nodeId,
                't' => $topic,
                'p' => base64_encode($msg),
                'q' => $qos,
                'r' => $retain,
                'f' => (string) $from,
            ], JSON_UNESCAPED_UNICODE);
            // 近似裁剪，减少 XADD 开销
            $this->redis->xAdd($this->stream, '*', ['d' => $payload], $this->maxlen, true);
        } catch (\Throwable $e) {
            $this->reconnect();
        }
    }

    /**
     * 非阻塞轮询总线，对每条非本进程消息回调 $cb($topic,$msg,$qos,$retain,$from)
     */
    public function poll(callable $cb): void
    {
        if (!$this->enabled) {
            return;
        }
        try {
            $res = $this->redis->xRead([$this->stream => $this->lastId], 200); // 非阻塞，最多取 200 条
            if (!$res || empty($res[$this->stream])) {
                return;
            }
            foreach ($res[$this->stream] as $id => $fields) {
                $this->lastId = (string) $id;
                $d = json_decode($fields['d'] ?? '', true);
                if (!is_array($d)) {
                    continue;
                }
                if (($d['n'] ?? '') === $this->nodeId) {
                    continue; // 跳过自身产生的消息，避免回环
                }
                $cb(
                    (string) ($d['t'] ?? ''),
                    base64_decode($d['p'] ?? ''),
                    (int) ($d['q'] ?? 0),
                    (int) ($d['r'] ?? 0),
                    $d['f'] ?? null
                );
            }
        } catch (\Throwable $e) {
            $this->reconnect();
        }
    }

    
    protected function reconnect(): void
    {
        $this->enabled = false;
        try {
            $this->connect();
        } catch (\Throwable $e) {
        }
    }
}