agentClient = $agentClient; } /** * 根据消息ID列表从crmslave数据库获取消息数据 */ public function getMessagesByIds(array $messageIds): Collection { if (empty($messageIds)) { return collect(); } try { $messages = DB::connection('crmslave') ->table('system_publish_event') ->whereIn('msg_id', $messageIds) ->select([ 'msg_id', 'event_type', 'trace_id', 'event_param', 'event_property', 'timestamp' ]) ->get(); return $messages->map(function ($message) { return [ 'msg_id' => $message->msg_id, 'event_type' => $message->event_type, 'trace_id' => $message->trace_id, 'event_param' => $message->event_param, 'event_property' => $message->event_property, 'timestamp' => $message->timestamp, 'parsed_param' => $this->parseJsonField($message->event_param), 'parsed_property' => $this->parseJsonField($message->event_property), ]; }); } catch (\Exception $e) { throw new \RuntimeException('从crmslave数据库获取消息失败: ' . $e->getMessage()); } } /** * 批量同步消息到agent */ public function syncMessages(array $messageIds): array { $messages = $this->getMessagesByIds($messageIds); $results = []; foreach ($messages as $message) { $result = $this->syncSingleMessage($message); $results[] = [ 'msg_id' => $message['msg_id'], 'success' => $result['success'], 'response' => $result['response'] ?? null, 'error' => $result['error'] ?? null, 'request_data' => $result['request_data'] ?? null, ]; } return $results; } /** * 同步单个消息到agent */ private function syncSingleMessage(array $message): array { try { $requestData = $this->buildAgentRequest($message); $response = $this->agentClient->dispatchMessage($requestData); if ($response->successful()) { return [ 'success' => true, 'response' => $response->json(), 'request_data' => $requestData, ]; } else { return [ 'success' => false, 'error' => 'HTTP ' . $response->status() . ': ' . $response->body(), 'request_data' => $requestData, ]; } } catch (\Exception $e) { return [ 'success' => false, 'error' => '请求失败: ' . $e->getMessage(), 'request_data' => $requestData ?? null, ]; } } /** * 构建agent接口请求数据 */ private function buildAgentRequest(array $message): array { $parsedParam = $message['parsed_param']; $parsedProperty = $message['parsed_property']; return [ 'topic_name' => $message['event_type'], 'msg_body' => [ 'id' => $message['msg_id'], 'data' => $parsedParam, 'timestamp' => $message['timestamp'], 'property' => $parsedProperty, ], 'target_service' => [1], // 默认目标服务 'trace_id' => $message['trace_id'], ]; } /** * 解析JSON字段 */ private function parseJsonField(?string $jsonString): mixed { if (empty($jsonString)) { return null; } try { return json_decode($jsonString, true); } catch (\Exception $e) { return $jsonString; // 如果解析失败,返回原始字符串 } } /** * 验证消息ID格式 */ public function validateMessageIds(array $messageIds): array { $errors = []; foreach ($messageIds as $index => $messageId) { if (empty($messageId)) { $errors[] = "第 " . ($index + 1) . " 行: 消息ID不能为空"; continue; } if (!is_string($messageId) && !is_numeric($messageId)) { $errors[] = "第 " . ($index + 1) . " 行: 消息ID格式无效"; continue; } // 可以添加更多的格式验证规则 } return $errors; } /** * 获取消息统计信息 */ public function getMessageStats(array $messageIds): array { $messages = $this->getMessagesByIds($messageIds); $stats = [ 'total_requested' => count($messageIds), 'total_found' => $messages->count(), 'total_missing' => count($messageIds) - $messages->count(), 'event_types' => $messages->groupBy('event_type')->map->count(), 'missing_ids' => array_diff($messageIds, $messages->pluck('msg_id')->toArray()), ]; return $stats; } }