Refactor ChatConnect and related classes to use dynamic typing for properties and method parameters. Update event handling to use a consistent naming convention for emitted events. Adjust WebSocket message handling methods to accept event arrays instead of specific data structures. Update configuration for Swoole to increase ping intervals and timeouts.

2026-01-29 03:04:40 +08:00
parent b8cd0c9418
commit ca996c227c
27 changed files with 441 additions and 148 deletions
@@ -31,7 +31,7 @@ class AssignService
     * @param int $sessionId 会话ID
     * @return int|null 分配的客服ID，null表示无可用客服
     */
-    public function assignSession(int $userId, int $sessionId): ?int
+    public function assignSession($userId, $sessionId): ?int
    {
        $redis = $this->getRedis();
        $lockKey = self::LOCK_PREFIX . $userId;
@@ -85,7 +85,7 @@ class AssignService
    /**
     * 获取在线客服列表
     */
-    public function getOnlineAgents(): array
+    public function getOnlineAgents()
    {
        $redis = $this->getRedis();
        $enabledAdmins = ChatAdminStatus::getEnabledAdminIds();
@@ -103,7 +103,7 @@ class AssignService
    /**
     * 选择会话数最少的客服
     */
-    public function selectLeastLoadAgent(array $onlineAgents): ?int
+    public function selectLeastLoadAgent($onlineAgents): ?int
    {
        $redis = $this->getRedis();
        $loads = [];
@@ -129,7 +129,7 @@ class AssignService
    /**
     * 添加到离线队列 (按余额降序)
     */
-    public function addToOfflineQueue(int $userId, int $sessionId): void
+    public function addToOfflineQueue($userId, $sessionId)
    {
        $redis = $this->getRedis();

@@ -151,7 +151,7 @@ class AssignService
    /**
     * 客服上线时处理队列
     */
-    public function processOfflineQueue(int $adminId): array
+    public function processOfflineQueue($adminId)
    {
        $redis = $this->getRedis();
        $processed = [];
@@ -190,7 +190,7 @@ class AssignService
    /**
     * 释放会话 (会话结束时调用)
     */
-    public function releaseSession(int $sessionId, int $adminId): void
+    public function releaseSession($sessionId, $adminId)
    {
        $redis = $this->getRedis();

@@ -207,7 +207,7 @@ class AssignService
    /**
     * 设置客服在线状态
     */
-    public function setAgentOnline(int $adminId): void
+    public function setAgentOnline($adminId)
    {
        $redis = $this->getRedis();
        $redis->setex(self::ONLINE_PREFIX . $adminId, 60, '1');
@@ -217,7 +217,7 @@ class AssignService
    /**
     * 刷新客服在线状态 (心跳续期)
     */
-    public function refreshAgentOnline(int $adminId): void
+    public function refreshAgentOnline($adminId)
    {
        $redis = $this->getRedis();
        $redis->expire(self::ONLINE_PREFIX . $adminId, 60);
@@ -226,7 +226,7 @@ class AssignService
    /**
     * 设置客服离线
     */
-    public function setAgentOffline(int $adminId): void
+    public function setAgentOffline($adminId)
    {
        $redis = $this->getRedis();
        $redis->del(self::ONLINE_PREFIX . $adminId);
@@ -236,7 +236,7 @@ class AssignService
    /**
     * 获取客服最大会话数
     */
-    private function getAgentMaxSessions(int $adminId): int
+    private function getAgentMaxSessions($adminId)
    {
        $status = ChatAdminStatus::getByAdminId($adminId);
        return $status['max_sessions'] ?? self::MAX_SESSIONS;
@@ -245,7 +245,7 @@ class AssignService
    /**
     * 获取用户已有会话的客服ID
     */
-    private function getExistingSessionAgent(int $userId): ?int
+    private function getExistingSessionAgent($userId): ?int
    {
        $session = ChatSession::getActiveByUserId($userId);
        return $session['admin_id'] ?? null;
@@ -254,7 +254,7 @@ class AssignService
    /**
     * 绑定会话到客服
     */
-    private function bindSessionToAgent(int $sessionId, int $adminId): void
+    private function bindSessionToAgent($sessionId, $adminId)
    {
        $redis = $this->getRedis();

@@ -272,7 +272,7 @@ class AssignService
    /**
     * 释放分配锁
     */
-    private function releaseLock(string $key, string $value): void
+    private function releaseLock($key, $value)
    {
        $redis = $this->getRedis();

@@ -290,7 +290,7 @@ class AssignService
    /**
     * 获取Redis实例
     */
-    private function getRedis(): \Redis
+    private function getRedis()
    {
        return Cache::store('redis')->handler();
    }
@@ -6,6 +6,7 @@ use app\models\chat\ChatMessage;
 use app\models\chat\ChatSession;
 use app\utils\Snowflake;
 use think\facade\Cache;
+use think\swoole\WebSocket;

 /**
 * 消息服务
@@ -20,12 +21,173 @@ class MessageService
    private const MAX_CONTENT_LENGTH = 500;
    private const RECONNECT_FETCH_LIMIT = 50;

+    // Redis Key 前缀
+    private const CONN_USER_PREFIX = 'cs:conn:user:';
+    private const CONN_AGENT_PREFIX = 'cs:conn:agent:';
+
+    /**
+     * 发送消息（WebSocket事件处理入口）
+     * @param WebSocket $ws
+     * @param array $data [sessionId, msgType, content, clientMsgId]
+     */
+    public function sendMessage($ws, $data)
+    {
+        echo "[MessageService] ========== 收到发送消息请求 ==========\n";
+        echo "[MessageService] 数据: " . json_encode($data, JSON_UNESCAPED_UNICODE) . "\n";
+
+        $sessionId = $data['sessionId'] ?? 0;
+        $msgType = $data['msgType'] ?? ChatMessage::MSG_TYPE_TEXT;
+        $content = $data['content'] ?? '';
+        $clientMsgId = $data['clientMsgId'] ?? null;
+
+        // 验证会话
+        $session = ChatSession::find($sessionId);
+        if (!$session) {
+            echo "[MessageService] 会话不存在: {$sessionId}\n";
+            $ws->emit('chat_error', ['error' => 'session_not_found']);
+            return;
+        }
+        $session = $session->toArray();
+        echo "[MessageService] 会话信息: user_id={$session['user_id']}, admin_id=" . ($session['admin_id'] ?? 'null') . "\n";
+
+        // 验证内容
+        $validation = $this->validateContent($content, $msgType);
+        if (!$validation['valid']) {
+            echo "[MessageService] 内容验证失败: {$validation['error']}\n";
+            $ws->emit('chat_error', ['error' => $validation['error']]);
+            return;
+        }
+
+        // 判断发送者类型（通过fd查找）
+        $fd = $ws->getSender();
+        $redis = $this->getRedis();
+
+        echo "[MessageService] 发送者FD: {$fd}\n";
+
+        // 检查是用户还是客服
+        $senderType = null;
+        $senderId = null;
+        $targetFd = null;
+
+        // 检查是否是客服发送
+        $agentId = $session['admin_id'];
+        if ($agentId) {
+            $agentFd = $redis->get(self::CONN_AGENT_PREFIX . $agentId);
+            echo "[MessageService] 客服FD (Redis): " . ($agentFd ?: 'null') . "\n";
+            if ($agentFd && (int)$agentFd === $fd) {
+                $senderType = ChatMessage::SENDER_ADMIN;
+                $senderId = $agentId;
+                // 目标是用户
+                $targetFd = $redis->get(self::CONN_USER_PREFIX . $session['user_id']);
+                echo "[MessageService] 识别为客服发送, 目标用户FD: " . ($targetFd ?: 'null') . "\n";
+            }
+        }
+
+        // 检查是否是用户发送
+        if ($senderType === null) {
+            $userId = $session['user_id'];
+            $userFd = $redis->get(self::CONN_USER_PREFIX . $userId);
+            echo "[MessageService] 用户FD (Redis): " . ($userFd ?: 'null') . "\n";
+            if ($userFd && (int)$userFd === $fd) {
+                $senderType = ChatMessage::SENDER_USER;
+                $senderId = $userId;
+                // 目标是客服
+                if ($agentId) {
+                    $targetFd = $redis->get(self::CONN_AGENT_PREFIX . $agentId);
+                    echo "[MessageService] 识别为用户发送, 目标客服FD: " . ($targetFd ?: 'null') . "\n";
+                }
+            }
+        }
+
+        if ($senderType === null) {
+            echo "[MessageService] 无法识别发送者身份, FD: {$fd}\n";
+            $ws->emit('chat_error', ['error' => 'unauthorized']);
+            return;
+        }
+
+        echo "[MessageService] 发送者类型: {$senderType}, 发送者ID: {$senderId}\n";
+
+        // 创建消息
+        $message = $this->createMessage([
+            'session_id' => $sessionId,
+            'sender_type' => $senderType,
+            'sender_id' => $senderId,
+            'msg_type' => $msgType,
+            'content' => $content,
+            'client_msg_id' => $clientMsgId,
+        ]);
+
+        // 发送确认给发送者
+        $ws->emit('chat_message_sent', [
+            'success' => true,
+            'msgId' => $message['msg_id'],
+            'clientMsgId' => $clientMsgId,
+            'sessionId' => $sessionId,
+        ]);
+
+        // 推送给目标
+        if ($targetFd) {
+            $targetFd = (int)$targetFd;
+            $server = app('swoole.server');
+
+            echo "[MessageService] 准备推送消息到FD: {$targetFd}\n";
+
+            if ($server->isEstablished($targetFd)) {
+                $payload = [
+                    'event' => 'chat_message_new',
+                    'data' => [
+                        'msgId' => $message['msg_id'],
+                        'sessionId' => $sessionId,
+                        'senderType' => $senderType,
+                        'msgType' => $msgType,
+                        'content' => $content,
+                        'createTime' => $message['create_time'],
+                    ]
+                ];
+
+                // 使用 Socket.IO 协议格式发送
+                $result = $this->emitToFd($targetFd, 'chat_message_new', $payload['data']);
+                echo "[MessageService] 推送结果: " . ($result ? '成功' : '失败') . "\n";
+
+                // 更新消息状态为已发送
+                $this->updateMessageStatus($message['msg_id'], ChatMessage::STATUS_SENT);
+            } else {
+                echo "[MessageService] 目标FD {$targetFd} 连接已断开\n";
+            }
+        } else {
+            echo "[MessageService] 目标FD为空，无法推送\n";
+        }
+    }
+
+    /**
+     * 向指定fd发送Socket.IO事件
+     */
+    private function emitToFd($fd, $event, $data)
+    {
+        $server = app('swoole.server');
+        if (!$server->isEstablished($fd)) {
+            return false;
+        }
+
+        // Socket.IO v2 协议格式: 42["event",{data}]
+        $packet = '42' . json_encode([$event, $data]);
+        return $server->push($fd, $packet);
+    }
+
+    /**
+     * 获取Redis实例
+     */
+    private function getRedis()
+    {
+        return Cache::store('redis')->handler();
+    }
+
    /**
     * 创建消息
     * @param array $data [session_id, sender_type, sender_id, msg_type, content]
     * @return array 消息数据
     */
-    public function createMessage(array $data): array
+    public function createMessage($data)
    {
        // 生成雪花ID
        $msgId = Snowflake::getInstance()->nextId();
@@ -74,7 +236,7 @@ class MessageService
     * @param array $payload 消息内容
     * @return bool
     */
-    public function pushMessage(int $msgId, int $targetFd, array $payload): bool
+    public function pushMessage($msgId, $targetFd, $payload)
    {
        $server = app('swoole.server');

@@ -110,7 +272,7 @@ class MessageService
    /**
     * 更新消息状态
     */
-    public function updateMessageStatus(int $msgId, int $status): void
+    public function updateMessageStatus($msgId, $status)
    {
        $update = ['status' => $status];

@@ -126,7 +288,7 @@ class MessageService
    /**
     * 获取会话未读消息 (重连后拉取)
     */
-    public function getUnreadMessages(int $sessionId, int $receiverType, int $limit = null): array
+    public function getUnreadMessages($sessionId, $receiverType, $limit = null)
    {
        $limit = $limit ?? self::RECONNECT_FETCH_LIMIT;
        return ChatMessage::getUnreadBySessionId($sessionId, $receiverType, $limit);
@@ -135,7 +297,7 @@ class MessageService
    /**
     * 批量标记消息为已读
     */
-    public function markMessagesAsRead(array $msgIds): void
+    public function markMessagesAsRead($msgIds)
    {
        if (empty($msgIds)) {
            return;
@@ -152,7 +314,7 @@ class MessageService
    /**
     * 标记消息为已送达
     */
-    public function markMessageDelivered(int $msgId): void
+    public function markMessageDelivered($msgId)
    {
        ChatMessage::where('msg_id', $msgId)
            ->where('status', '<', ChatMessage::STATUS_DELIVERED)
@@ -165,7 +327,7 @@ class MessageService
    /**
     * 获取会话消息历史
     */
-    public function getMessageHistory(int $sessionId, int $limit = 50, int $lastId = 0): array
+    public function getMessageHistory($sessionId, $limit = 50, $lastId = 0)
    {
        return ChatMessage::getBySessionId($sessionId, $limit, $lastId);
    }
@@ -173,7 +335,7 @@ class MessageService
    /**
     * 增加重试次数
     */
-    private function incrementRetryCount(int $msgId): void
+    private function incrementRetryCount($msgId)
    {
        ChatMessage::where('msg_id', $msgId)->inc('retry_count')->update();
    }
@@ -181,7 +343,7 @@ class MessageService
    /**
     * 验证消息内容
     */
-    public function validateContent(string $content, int $msgType): array
+    public function validateContent($content, $msgType)
    {
        if ($msgType === ChatMessage::MSG_TYPE_TEXT) {
            if (empty(trim($content))) {
@@ -19,7 +19,7 @@ class SessionService
    private const CONN_USER_PREFIX = 'cs:conn:user:';
    private const CONN_AGENT_PREFIX = 'cs:conn:agent:';

-    private AssignService $assignService;
+    private $assignService;

    public function __construct()
    {
@@ -32,7 +32,7 @@ class SessionService
     * @param int $source 来源 1=PC 2=Game 3=Portal
     * @return array 会话数据
     */
-    public function createSession(int $userId, int $source): array
+    public function createSession($userId, $source)
    {
        $redis = $this->getRedis();

@@ -70,7 +70,7 @@ class SessionService
    /**
     * 获取用户活跃会话
     */
-    public function getActiveSession(int $userId): ?array
+    public function getActiveSession($userId)
    {
        return ChatSession::getActiveByUserId($userId);
    }
@@ -78,7 +78,7 @@ class SessionService
    /**
     * 获取会话详情(含用户信息)
     */
-    public function getSessionDetail(int $sessionId): ?array
+    public function getSessionDetail($sessionId)
    {
        $session = ChatSession::find($sessionId);
        if (!$session) {
@@ -105,7 +105,7 @@ class SessionService
    /**
     * 结束会话
     */
-    public function endSession(int $sessionId, int $operatorId): bool
+    public function endSession($sessionId, $operatorId)
    {
        $session = ChatSession::find($sessionId);
        if (!$session) {
@@ -141,7 +141,7 @@ class SessionService
    /**
     * 转接会话
     */
-    public function transferSession(int $sessionId, int $newAdminId): bool
+    public function transferSession($sessionId, $newAdminId)
    {
        $session = ChatSession::find($sessionId);
        if (!$session || $session['status'] !== ChatSession::STATUS_ACTIVE) {
@@ -175,7 +175,7 @@ class SessionService
    /**
     * 会话评价
     */
-    public function rateSession(int $sessionId, int $rating, ?string $content = null): bool
+    public function rateSession($sessionId, $rating, $content = null)
    {
        $session = ChatSession::find($sessionId);
        if (!$session) {
@@ -197,7 +197,7 @@ class SessionService
    /**
     * 注册用户连接
     */
-    public function registerUserConnection(int $userId, int $fd): void
+    public function registerUserConnection($userId, $fd)
    {
        $redis = $this->getRedis();
        $redis->setex(self::CONN_USER_PREFIX . $userId, 60, $fd);
@@ -206,7 +206,7 @@ class SessionService
    /**
     * 注册客服连接
     */
-    public function registerAgentConnection(int $adminId, int $fd): void
+    public function registerAgentConnection($adminId, $fd)
    {
        $redis = $this->getRedis();
        $redis->setex(self::CONN_AGENT_PREFIX . $adminId, 60, $fd);
@@ -215,7 +215,7 @@ class SessionService
    /**
     * 刷新用户连接TTL
     */
-    public function refreshUserConnection(int $userId): void
+    public function refreshUserConnection($userId)
    {
        $redis = $this->getRedis();
        $redis->expire(self::CONN_USER_PREFIX . $userId, 60);
@@ -224,7 +224,7 @@ class SessionService
    /**
     * 刷新客服连接TTL
     */
-    public function refreshAgentConnection(int $adminId): void
+    public function refreshAgentConnection($adminId)
    {
        $redis = $this->getRedis();
        $redis->expire(self::CONN_AGENT_PREFIX . $adminId, 60);
@@ -233,7 +233,7 @@ class SessionService
    /**
     * 获取用户连接FD
     */
-    public function getUserFd(int $userId): ?int
+    public function getUserFd($userId): ?int
    {
        $redis = $this->getRedis();
        $fd = $redis->get(self::CONN_USER_PREFIX . $userId);
@@ -243,7 +243,7 @@ class SessionService
    /**
     * 获取客服连接FD
     */
-    public function getAgentFd(int $adminId): ?int
+    public function getAgentFd($adminId): ?int
    {
        $redis = $this->getRedis();
        $fd = $redis->get(self::CONN_AGENT_PREFIX . $adminId);
@@ -253,7 +253,7 @@ class SessionService
    /**
     * 清理用户连接
     */
-    public function clearUserConnection(int $userId): void
+    public function clearUserConnection($userId)
    {
        $redis = $this->getRedis();
        $redis->del(self::CONN_USER_PREFIX . $userId);
@@ -262,7 +262,7 @@ class SessionService
    /**
     * 清理客服连接
     */
-    public function clearAgentConnection(int $adminId): void
+    public function clearAgentConnection($adminId)
    {
        $redis = $this->getRedis();
        $redis->del(self::CONN_AGENT_PREFIX . $adminId);
@@ -271,7 +271,7 @@ class SessionService
    /**
     * 获取客服会话列表
     */
-    public function getAgentSessions(int $adminId): array
+    public function getAgentSessions($adminId)
    {
        $sessions = ChatSession::getActiveByAdminId($adminId);

@@ -297,7 +297,7 @@ class SessionService
    /**
     * 获取待分配会话列表
     */
-    public function getPendingSessions(): array
+    public function getPendingSessions()
    {
        $sessions = ChatSession::where('status', ChatSession::STATUS_PENDING)
            ->order('create_time', 'asc')
@@ -319,7 +319,7 @@ class SessionService
    /**
     * 获取Redis实例
     */
-    private function getRedis(): \Redis
+    private function getRedis()
    {
        return Cache::store('redis')->handler();
    }