1.重构websocket连接

app/process/GameWebSocketServer.php

@@ -4,26 +4,228 @@ declare(strict_types=1);
 
 namespace app\process;
 
-use app\common\service\GameWebSocketEventBus;
 use app\common\service\GameLiveService;
+use app\common\service\GameWebSocketAuthHelper;
+use app\common\service\GameWebSocketDispatcher;
+use app\common\service\GameWebSocketEventBus;
 use app\common\service\GameWebSocketPayloadHelper;
+use app\common\service\GameWebSocketSubscriptionRegistry;
+use support\Log;
+use Throwable;
 use Workerman\Connection\TcpConnection;
 use Workerman\Timer;
 
 /**
- * 后台测试页 WebSocket 服务（仅用于连接联调）
+ * H5/后台 WebSocket 服务（重构版，2026-05）
+ *
+ * 设计与 docs/36字花-移动端接口设计草案.md §7 对齐：
+ *
+ * - 握手鉴权（GameWebSocketAuthHelper）
+ *   - mobile：URL Query `auth_token` + `user_token`，绑定 user_id，user 级主题按 user_id 过滤
+ *   - admin： URL Query `admin_ws_token`（后台 wsConfig 签发，写 Redis 短时签名），user_id=0，
+ *            user 级主题不过滤（运维/联调可观测全量）
+ * - 客户端 -> 服务端：`{"action":"ping"}` / `{"action":"subscribe","topics":[...]}`
+ * - 服务端 -> 客户端：`ws.connected` / `ws.subscribed` / `pong` / `ws.error` / 业务事件帧
+ * - 主动心跳超时：连接 60s 内无任何上行报文（含 ping）即被 server 主动 close，
+ *   触发客户端重连，避免半关闭僵尸连接堵在分发表里
+ * - 事件分发由 GameWebSocketDispatcher 负责；每次 publish/consume/dispatch/send/订阅/关闭
+ *   都写入独立日志通道 runtime/logs/ws.log
+ *
+ * 高可用：
+ * - 消费 Timer 的 popBatch 异常已被 EventBus 内部捕获并记录，下次 tick 自动恢复
+ * - 任一连接 send 异常不会影响其它连接的下发（Dispatcher 内单连接 try/catch）
+ * - admin.live.snapshot 兜底直推 Timer 与队列消费解耦，Redis 异常时后台仍能看到对局状态
  */
 class GameWebSocketServer
 {
-    /** @var array<int, TcpConnection> */
+    /** @var array<int, TcpConnection> connection_id => TcpConnection（仅本进程内） */
     private static array $connections = [];
 
+    /** @var array<int, array<string, mixed>> connection_id => 鉴权元数据（mode/admin_id 等，便于日志/排查） */
+    private static array $connectionAuth = [];
+
     private static bool $eventBusConsumerStarted = false;
     private static bool $adminSnapshotTickerStarted = false;
+    private static bool $heartbeatCheckerStarted = false;
+
+    /** 客户端 60s 内无任何上行报文（含 ping）即被主动断开 */
+    private const HEARTBEAT_IDLE_SECONDS = 60;
+
+    /** 事件队列每 N ms 拉一次（默认 1s；与文档 §7.1.3 一致） */
+    private const QUEUE_TICK_INTERVAL = 1;
+
+    /** 心跳超时检查间隔（秒） */
+    private const HEARTBEAT_CHECK_INTERVAL = 10;
+
+    public function onWorkerStart(): void
+    {
+        Log::channel('ws')->info('ws worker start', [
+            'pid' => function_exists('posix_getpid') ? posix_getpid() : getmypid(),
+        ]);
+        self::ensureEventBusConsumer();
+        self::ensureAdminLiveSnapshotTicker();
+        self::ensureHeartbeatChecker();
+    }
+
+    public function onConnect(TcpConnection $connection): void
+    {
+        $connection->topics = [];
+        $connection->wsAuth = null;
+        self::$connections[$connection->id] = $connection;
+        Log::channel('ws')->debug('tcp onConnect', [
+            'connection_id' => $connection->id,
+            'remote' => method_exists($connection, 'getRemoteIp') ? $connection->getRemoteIp() : '',
+        ]);
+    }
 
     /**
-     * 从 Redis 队列拉取事件并推送给已订阅连接。
-     * 部分环境下 WebSocket 进程的 onWorkerStart 可能未触发，因此在首帧握手处也会兜底启动一次（全局仅注册一个 Timer）。
+     * Workerman 在 WebSocket 握手时回调；$request 可能是 string（HTTP raw）或 Webman\Http\Request 对象，
+     * 这里仅尝试从 connection 内置变量取 QueryString 做最大兼容。
+     */
+    public function onWebSocketConnect(TcpConnection $connection, mixed $request = null): void
+    {
+        self::ensureEventBusConsumer();
+        self::ensureAdminLiveSnapshotTicker();
+        self::ensureHeartbeatChecker();
+
+        $queryString = self::extractQueryString($connection, $request);
+        $query = GameWebSocketAuthHelper::parseQueryString($queryString);
+
+        $auth = GameWebSocketAuthHelper::authorize($query);
+        $remoteIp = self::remoteIp($connection);
+
+        if (!$auth['ok']) {
+            Log::channel('ws')->warning('handshake denied', [
+                'connection_id' => $connection->id,
+                'remote_ip' => $remoteIp,
+                'reason' => $auth['reason'],
+                'query_keys' => array_keys($query),
+            ]);
+            GameWebSocketDispatcher::sendDirect($connection, 'ws.error', [
+                'code' => 1101,
+                'message' => 'Authentication failed: ' . ($auth['reason'] ?: 'unauthorized'),
+            ], 'handshake_denied');
+            try {
+                $connection->close();
+            } catch (Throwable) {
+            }
+            return;
+        }
+
+        $connection->wsAuth = $auth;
+        self::$connectionAuth[$connection->id] = $auth;
+        GameWebSocketSubscriptionRegistry::registerConnection($connection->id, (int) $auth['user_id'], $remoteIp);
+
+        Log::channel('ws')->info('handshake ok', [
+            'connection_id' => $connection->id,
+            'remote_ip' => $remoteIp,
+            'mode' => $auth['mode'],
+            'user_id' => $auth['user_id'],
+            'admin_id' => $auth['admin_id'],
+        ]);
+
+        GameWebSocketDispatcher::sendDirect($connection, 'ws.connected', [
+            'message' => 'WebSocket connected',
+            'connection_id' => $connection->id,
+            'mode' => $auth['mode'],
+            'user_id' => $auth['user_id'],
+            'server_time' => time(),
+            'heartbeat_interval' => 30,
+            'idle_timeout' => self::HEARTBEAT_IDLE_SECONDS,
+        ], 'handshake_ok');
+    }
+
+    public function onMessage(TcpConnection $connection, string $payload): void
+    {
+        GameWebSocketSubscriptionRegistry::touch($connection->id);
+
+        $decoded = json_decode($payload, true);
+        if (!is_array($decoded)) {
+            Log::channel('ws')->debug('invalid json payload', [
+                'connection_id' => $connection->id,
+                'payload_size' => strlen($payload),
+            ]);
+            GameWebSocketDispatcher::sendDirect($connection, 'ws.error', [
+                'message' => 'Invalid JSON payload',
+            ], 'invalid_json');
+            return;
+        }
+
+        $action = isset($decoded['action']) && is_string($decoded['action']) ? trim($decoded['action']) : '';
+
+        if ($action === 'ping') {
+            GameWebSocketDispatcher::sendDirect($connection, 'pong', [
+                'server_time' => date('Y-m-d H:i:s'),
+                'server_time_ts' => time(),
+            ], 'pong');
+            return;
+        }
+
+        if ($action === 'subscribe') {
+            $rawTopics = $decoded['topics'] ?? [];
+            $rawList = is_array($rawTopics) ? $rawTopics : [];
+            $finalTopics = GameWebSocketSubscriptionRegistry::replaceSubscriptions($connection->id, $rawList);
+            Log::channel('ws')->info('subscribe', [
+                'connection_id' => $connection->id,
+                'user_id' => GameWebSocketSubscriptionRegistry::userIdOf($connection->id),
+                'topics' => $finalTopics,
+                'requested_count' => is_array($rawTopics) ? count($rawTopics) : 0,
+            ]);
+            GameWebSocketDispatcher::sendDirect($connection, 'ws.subscribed', [
+                'topics' => $finalTopics,
+            ], 'subscribed');
+            self::pushAdminTestOddsPreview($connection, $finalTopics);
+            return;
+        }
+
+        Log::channel('ws')->debug('unsupported action', [
+            'connection_id' => $connection->id,
+            'action' => $action,
+        ]);
+        GameWebSocketDispatcher::sendDirect($connection, 'ws.error', [
+            'message' => 'Unsupported action',
+            'received_action' => $action,
+        ], 'unsupported_action');
+    }
+
+    public function onError(TcpConnection $connection, int $code, string $msg): void
+    {
+        Log::channel('ws')->warning('connection error', [
+            'connection_id' => $connection->id,
+            'code' => $code,
+            'detail' => $msg,
+        ]);
+        try {
+            $connection->send(json_encode([
+                'event' => 'ws.error',
+                'message' => 'Server internal error',
+                'code' => $code,
+                'detail' => $msg,
+            ], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
+        } catch (Throwable) {
+        }
+    }
+
+    public function onClose(TcpConnection $connection): void
+    {
+        $auth = self::$connectionAuth[$connection->id] ?? [];
+        Log::channel('ws')->info('connection close', [
+            'connection_id' => $connection->id,
+            'user_id' => GameWebSocketSubscriptionRegistry::userIdOf($connection->id),
+            'mode' => $auth['mode'] ?? '',
+            'remaining_connections' => max(0, GameWebSocketSubscriptionRegistry::connectionCount() - 1),
+        ]);
+        GameWebSocketSubscriptionRegistry::unregisterConnection($connection->id);
+        unset(self::$connections[$connection->id], self::$connectionAuth[$connection->id]);
+    }
+
+    // ============================================================
+    // 内部：Timer / 兜底
+    // ============================================================
+
+    /**
+     * 每秒拉取 Redis 队列事件并分发到对应订阅连接。
+     * popBatch 内部已捕获 Redis 异常并写 ws 日志，Timer 不会因此停跑。
      */
     private static function ensureEventBusConsumer(): void
     {
@@ -31,41 +233,36 @@ class GameWebSocketServer
             return;
         }
         self::$eventBusConsumerStarted = true;
-        Timer::add(1, static function (): void {
-            $events = GameWebSocketEventBus::popBatch();
-            if ($events === []) {
-                return;
-            }
-            foreach ($events as $event) {
-                $topic = $event['topic'] ?? '';
-                if (!is_string($topic) || $topic === '') {
-                    continue;
+        Timer::add(self::QUEUE_TICK_INTERVAL, static function (): void {
+            try {
+                $events = GameWebSocketEventBus::popBatch();
+                if ($events === []) {
+                    return;
                 }
-                $eventName = $event['event'] ?? $topic;
-                $data = $event['data'] ?? [];
-                if (!is_array($data)) {
-                    $data = [];
-                }
-                $serverTime = $event['server_time'] ?? time();
-                foreach (self::$connections as $connection) {
-                    $topics = $connection->topics ?? [];
-                    if (!is_array($topics) || !in_array($topic, $topics, true)) {
-                        continue;
+                foreach ($events as $event) {
+                    try {
+                        GameWebSocketDispatcher::dispatch($event, self::$connections);
+                    } catch (Throwable $e) {
+                        Log::channel('ws')->error('dispatch loop exception', [
+                            'topic' => $event['topic'] ?? '',
+                            'error' => $e->getMessage(),
+                        ]);
                     }
-                    $connection->send(json_encode([
-                        'event' => $eventName,
-                        'topic' => $topic,
-                        'data' => $data,
-                        'server_time' => $serverTime,
-                    ], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
                 }
+            } catch (Throwable $e) {
+                Log::channel('ws')->error('event bus consumer tick exception', [
+                    'error' => $e->getMessage(),
+                ]);
             }
         });
+        Log::channel('ws')->info('event bus consumer started', [
+            'tick_interval_seconds' => self::QUEUE_TICK_INTERVAL,
+        ]);
     }
 
     /**
-     * 兜底直推：admin.live.snapshot 每秒主动构建并广播。
-     * 目的：即使 Redis 队列不可用，也能保证 /admin/game/live 实时看到对局变化。
+     * admin.live.snapshot 兜底：每秒构建一次快照直接推送（不依赖 Redis 队列）。
+     * 用于 /admin/game/live 实时对局页，确保 Redis 异常时后台仍能看到对局状态。
      */
     private static function ensureAdminLiveSnapshotTicker(): void
     {
@@ -74,18 +271,18 @@ class GameWebSocketServer
         }
         self::$adminSnapshotTickerStarted = true;
         Timer::add(1, static function (): void {
-            $hasAdminSubscriber = false;
-            foreach (self::$connections as $connection) {
-                $topics = $connection->topics ?? [];
-                if (is_array($topics) && in_array('admin.live.snapshot', $topics, true)) {
-                    $hasAdminSubscriber = true;
-                    break;
-                }
-            }
-            if (!$hasAdminSubscriber) {
+            $cids = GameWebSocketSubscriptionRegistry::connectionsForTopic('admin.live.snapshot');
+            if ($cids === []) {
+                return;
+            }
+            try {
+                $snapshot = GameLiveService::buildSnapshot(null);
+            } catch (Throwable $e) {
+                Log::channel('ws')->error('admin snapshot build failed', [
+                    'error' => $e->getMessage(),
+                ]);
                 return;
             }
-            $snapshot = GameLiveService::buildSnapshot(null);
             $payload = json_encode([
                 'event' => 'admin.live.snapshot',
                 'topic' => 'admin.live.snapshot',
@@ -95,106 +292,60 @@ class GameWebSocketServer
             if (!is_string($payload) || $payload === '') {
                 return;
             }
-            foreach (self::$connections as $connection) {
-                $topics = $connection->topics ?? [];
-                if (!is_array($topics) || !in_array('admin.live.snapshot', $topics, true)) {
+            $sent = 0;
+            $failed = 0;
+            foreach ($cids as $cid) {
+                if (!isset(self::$connections[$cid])) {
                     continue;
                 }
-                $connection->send($payload);
+                try {
+                    self::$connections[$cid]->send($payload);
+                    $sent++;
+                } catch (Throwable $e) {
+                    $failed++;
+                    Log::channel('ws')->warning('admin snapshot send failed', [
+                        'connection_id' => $cid,
+                        'error' => $e->getMessage(),
+                    ]);
+                }
             }
         });
     }
 
-    public function onWorkerStart(): void
+    /**
+     * 心跳超时检查器：每 10s 扫描一次，关闭 60s 无上行报文的连接。
+     * 触发客户端重连，避免半关闭僵尸连接持续接收推送但不实际送达。
+     */
+    private static function ensureHeartbeatChecker(): void
     {
-        self::ensureEventBusConsumer();
-        self::ensureAdminLiveSnapshotTicker();
-    }
-
-    public function onConnect(TcpConnection $connection): void
-    {
-        $connection->topics = [];
-        self::$connections[$connection->id] = $connection;
-    }
-
-    public function onWebSocketConnect(TcpConnection $connection): void
-    {
-        self::ensureEventBusConsumer();
-        self::ensureAdminLiveSnapshotTicker();
-        $connection->send(json_encode([
-            'event' => 'ws.connected',
-            'message' => 'WebSocket connected',
-            'connection_id' => $connection->id,
-            'server_time' => time(),
-            'heartbeat_interval' => 30,
-        ], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
-    }
-
-    public function onMessage(TcpConnection $connection, string $payload): void
-    {
-        $decoded = json_decode($payload, true);
-        if (!is_array($decoded)) {
-            $connection->send(json_encode([
-                'event' => 'ws.error',
-                'message' => 'Invalid JSON payload',
-            ], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
+        if (self::$heartbeatCheckerStarted) {
             return;
         }
-
-        $action = $decoded['action'] ?? '';
-        if ($action === 'ping') {
-            $connection->send(json_encode([
-                'event' => 'pong',
-                'server_time' => date('Y-m-d H:i:s'),
-            ], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
-            return;
-        }
-
-        if ($action === 'subscribe') {
-            $topics = $decoded['topics'] ?? [];
-            $sanitized = [];
-            if (is_array($topics)) {
-                foreach ($topics as $topic) {
-                    if (!is_string($topic)) {
-                        continue;
+        self::$heartbeatCheckerStarted = true;
+        Timer::add(self::HEARTBEAT_CHECK_INTERVAL, static function (): void {
+            $cutoff = time() - self::HEARTBEAT_IDLE_SECONDS;
+            $stale = GameWebSocketSubscriptionRegistry::staleConnections($cutoff);
+            if ($stale === []) {
+                return;
+            }
+            foreach ($stale as $cid) {
+                Log::channel('ws')->info('close idle connection', [
+                    'connection_id' => $cid,
+                    'user_id' => GameWebSocketSubscriptionRegistry::userIdOf($cid),
+                    'idle_seconds' => self::HEARTBEAT_IDLE_SECONDS,
+                ]);
+                if (isset(self::$connections[$cid])) {
+                    try {
+                        self::$connections[$cid]->close();
+                    } catch (Throwable) {
                     }
-                    $value = trim($topic);
-                    if ($value === '') {
-                        continue;
-                    }
-                    $sanitized[] = $value;
                 }
             }
-            $connection->topics = array_values(array_unique($sanitized));
-            $connection->send(json_encode([
-                'event' => 'ws.subscribed',
-                'topics' => $connection->topics,
-            ], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
-            self::pushAdminTestOddsPreview($connection, $connection->topics);
-            return;
-        }
-
-        $connection->send(json_encode([
-            'event' => 'ws.error',
-            'message' => 'Unsupported action',
-            'received_action' => $action,
-        ], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
-    }
-
-    public function onError(TcpConnection $connection, int $code, string $msg): void
-    {
-        $connection->send(json_encode([
-            'event' => 'ws.error',
-            'message' => 'Server internal error',
-            'code' => $code,
-            'detail' => $msg,
-        ], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
-    }
-
-    public function onClose(TcpConnection $connection): void
-    {
-        $connection->topics = [];
-        unset(self::$connections[$connection->id]);
+        });
+        Log::channel('ws')->info('heartbeat checker started', [
+            'idle_seconds' => self::HEARTBEAT_IDLE_SECONDS,
+            'check_interval' => self::HEARTBEAT_CHECK_INTERVAL,
+        ]);
     }
 
     /**
@@ -210,12 +361,59 @@ class GameWebSocketServer
         }
         $serverTime = time();
         foreach ($frames as $frame) {
-            $connection->send(json_encode([
-                'event' => $frame['event'],
+            GameWebSocketDispatcher::sendDirect($connection, $frame['event'], [
                 'topic' => $frame['topic'],
                 'data' => $frame['data'],
                 'server_time' => $serverTime,
-            ], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
+            ], 'admin_test_preview');
         }
     }
+
+    /**
+     * 兼容多种 Workerman 版本：尝试取握手 URI 的 query 部分。
+     * Workerman v5 中 onWebSocketConnect 第二参数已是 array $headers/$request；这里尽量兼容字符串与对象。
+     */
+    private static function extractQueryString(TcpConnection $connection, mixed $request): string
+    {
+        if (isset($connection->wsHandshakeQuery) && is_string($connection->wsHandshakeQuery)) {
+            return (string) $connection->wsHandshakeQuery;
+        }
+        if (isset($connection->onWebSocketConnect)) {
+            // noop
+        }
+        $serverUri = '';
+        if (isset($_SERVER['REQUEST_URI']) && is_string($_SERVER['REQUEST_URI'])) {
+            $serverUri = $_SERVER['REQUEST_URI'];
+        }
+        if ($serverUri === '' && isset($connection->headers) && is_array($connection->headers)) {
+            $line = $connection->headers['get'] ?? $connection->headers['GET'] ?? '';
+            if (is_string($line)) {
+                $serverUri = $line;
+            }
+        }
+        if ($serverUri === '' && is_string($request)) {
+            if (preg_match('#^GET\s+([^\s]+)#i', $request, $m) === 1) {
+                $serverUri = $m[1];
+            }
+        }
+        if ($serverUri === '' && is_object($request) && method_exists($request, 'queryString')) {
+            $serverUri = '?' . (string) $request->queryString();
+        }
+        $qPos = strpos($serverUri, '?');
+        if ($qPos === false) {
+            return '';
+        }
+        return substr($serverUri, $qPos + 1);
+    }
+
+    private static function remoteIp(TcpConnection $connection): string
+    {
+        try {
+            if (method_exists($connection, 'getRemoteIp')) {
+                return (string) $connection->getRemoteIp();
+            }
+        } catch (Throwable) {
+        }
+        return '';
+    }
 }