163 lines
4.6 KiB
PHP
163 lines
4.6 KiB
PHP
<?php
|
||
|
||
namespace App\Services;
|
||
|
||
use App\Clients\MonoClient;
|
||
use Illuminate\Support\Facades\DB;
|
||
use Illuminate\Support\Collection;
|
||
use Carbon\Carbon;
|
||
|
||
class MessageSyncService
|
||
{
|
||
private MonoClient $monoClient;
|
||
|
||
public function __construct(MonoClient $monoClient)
|
||
{
|
||
$this->monoClient = $monoClient;
|
||
}
|
||
|
||
/**
|
||
* 根据消息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());
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 批量同步消息（通过mono消费）
|
||
*/
|
||
public function syncMessages(array $messageIds): array
|
||
{
|
||
$results = [];
|
||
|
||
foreach ($messageIds as $msgId) {
|
||
$results[] = $this->syncSingleMessage($msgId);
|
||
}
|
||
|
||
return $results;
|
||
}
|
||
|
||
/**
|
||
* 通过mono消费单个消息
|
||
*/
|
||
private function syncSingleMessage(string $msgId): array
|
||
{
|
||
try {
|
||
$response = $this->monoClient->consumeMessage($msgId);
|
||
$body = $response->json();
|
||
|
||
if ($response->successful() && ($body['code'] ?? -1) === 0) {
|
||
return [
|
||
'msg_id' => $msgId,
|
||
'success' => true,
|
||
'response' => $body,
|
||
];
|
||
} else {
|
||
return [
|
||
'msg_id' => $msgId,
|
||
'success' => false,
|
||
'error' => $body['message'] ?? ('HTTP ' . $response->status() . ': ' . $response->body()),
|
||
'response' => $body,
|
||
];
|
||
}
|
||
} catch (\Exception $e) {
|
||
return [
|
||
'msg_id' => $msgId,
|
||
'success' => false,
|
||
'error' => '请求失败: ' . $e->getMessage(),
|
||
];
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 解析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;
|
||
}
|
||
}
|