From 380bfaaea69a5cb032bebca30e5dc4a57dd743df Mon Sep 17 00:00:00 2001 From: kuaifan Date: Sun, 9 Aug 2026 13:00:23 +0000 Subject: [PATCH] feat(file): index chat message attachments --- .../Commands/BackfillMessageAttachments.php | 131 ++++++++++ app/Http/Controllers/IndexController.php | 3 + app/Models/WebSocketDialogMsg.php | 3 + app/Models/WebSocketDialogMsgAttachment.php | 49 ++++ .../WebSocketDialogMsgAttachmentBackfill.php | 52 ++++ .../MessageAttachmentBackfillService.php | 193 ++++++++++++++ app/Services/MessageAttachmentService.php | 236 ++++++++++++++++++ app/Tasks/MessageAttachmentBackfillTask.php | 22 ++ ...eb_socket_dialog_msg_attachments_table.php | 37 +++ ..._dialog_msg_attachment_backfills_table.php | 47 ++++ .../MessageAttachmentBackfillServiceTest.php | 87 +++++++ tests/Unit/MessageAttachmentServiceTest.php | 85 +++++++ 12 files changed, 945 insertions(+) create mode 100644 app/Console/Commands/BackfillMessageAttachments.php create mode 100644 app/Models/WebSocketDialogMsgAttachment.php create mode 100644 app/Models/WebSocketDialogMsgAttachmentBackfill.php create mode 100644 app/Services/MessageAttachmentBackfillService.php create mode 100644 app/Services/MessageAttachmentService.php create mode 100644 app/Tasks/MessageAttachmentBackfillTask.php create mode 100644 database/migrations/2026_08_04_000001_create_web_socket_dialog_msg_attachments_table.php create mode 100644 database/migrations/2026_08_04_000002_create_web_socket_dialog_msg_attachment_backfills_table.php create mode 100644 tests/Feature/MessageAttachmentBackfillServiceTest.php create mode 100644 tests/Unit/MessageAttachmentServiceTest.php diff --git a/app/Console/Commands/BackfillMessageAttachments.php b/app/Console/Commands/BackfillMessageAttachments.php new file mode 100644 index 000000000..4ec4eb4ca --- /dev/null +++ b/app/Console/Commands/BackfillMessageAttachments.php @@ -0,0 +1,131 @@ +option('status')) { + return $this->showStatus(); + } + + $after = max(0, intval($this->option('after'))); + $before = max(0, intval($this->option('before'))); + $batch = min(2000, max(1, intval($this->option('batch')))); + $limit = max(0, intval($this->option('limit'))); + $sleep = min(60000, max(0, intval($this->option('sleep')))); + $dryRun = boolval($this->option('dry-run')); + + if ($before === 0) { + $before = intval(WebSocketDialogMsg::withTrashed()->max('id')); + } + if ($before <= $after) { + $this->info("没有需要处理的消息（after={$after}, before={$before}）"); + return self::SUCCESS; + } + + $this->info(sprintf( + '开始回填消息附件：范围 (%d, %d]，批量 %d%s', + $after, + $before, + $batch, + $dryRun ? '，仅预览' : '' + )); + + $lastId = $after; + $processed = 0; + $attachmentCount = 0; + $emptyCount = 0; + $failedIds = []; + + while ($lastId < $before && ($limit === 0 || $processed < $limit)) { + $take = $limit > 0 ? min($batch, $limit - $processed) : $batch; + $messages = MessageAttachmentBackfillService::candidateQuery($lastId, $before) + ->take($take) + ->get(); + + if ($messages->isEmpty()) { + break; + } + + foreach ($messages as $message) { + $lastId = intval($message->id); + $processed++; + try { + if ($dryRun) { + $data = Base::json2array($message->getRawOriginal('msg')); + $count = count(MessageAttachmentService::extract((string)$message->type, $data)); + } else { + $count = MessageAttachmentService::sync($message); + } + $attachmentCount += $count; + if ($count === 0) { + $emptyCount++; + } + } catch (\Throwable $e) { + $failedIds[] = $lastId; + $this->warn("消息 {$lastId} 处理失败：{$e->getMessage()}"); + } + } + + $this->line("已处理 {$processed} 条消息，附件 {$attachmentCount} 条，当前消息ID {$lastId}"); + if ($sleep > 0 && $lastId < $before) { + usleep($sleep * 1000); + } + } + + $this->newLine(); + $this->info("回填结束：消息 {$processed} 条，附件 {$attachmentCount} 条，无有效附件 {$emptyCount} 条"); + $this->info("续跑参数：--after={$lastId} --before={$before}"); + if (!empty($failedIds)) { + $this->error('失败消息ID：' . implode(',', $failedIds)); + return self::FAILURE; + } + + return self::SUCCESS; + } + + private function showStatus(): int + { + $state = WebSocketDialogMsgAttachmentBackfill::query()->orderBy('id')->first(); + if (!$state) { + $this->warn('自动回填任务尚未登记，请先执行数据库迁移'); + return self::FAILURE; + } + + $this->table(['项目', '值'], [ + ['状态', $state->status], + ['消息快照上界', $state->before_msg_id], + ['当前消息ID', $state->last_msg_id], + ['成功处理消息', $state->processed_count], + ['已索引附件', $state->attachment_count], + ['无有效附件消息', $state->empty_count], + ['失败尝试', $state->failure_count], + ['当前连续重试', $state->retry_count], + ['下次重试时间', $state->next_retry_at?->toDateTimeString() ?: '-'], + ['最近错误', $state->last_error ?: '-'], + ['首次开始时间', $state->started_at?->toDateTimeString() ?: '-'], + ['完成时间', $state->completed_at?->toDateTimeString() ?: '-'], + ]); + return self::SUCCESS; + } +} diff --git a/app/Http/Controllers/IndexController.php b/app/Http/Controllers/IndexController.php index d08bd1671..0d015095c 100755 --- a/app/Http/Controllers/IndexController.php +++ b/app/Http/Controllers/IndexController.php @@ -25,6 +25,7 @@ use App\Tasks\ManticoreSyncTask; use App\Tasks\UnclaimedTaskRemindTask; use App\Tasks\TodoRemindTask; use App\Tasks\AiTaskLoopTask; +use App\Tasks\MessageAttachmentBackfillTask; use Hhxsv5\LaravelS\Swoole\Task\Task; use App\Module\PatchedAvatar as Avatar; @@ -281,6 +282,8 @@ class IndexController extends InvokeController Task::deliver(new ManticoreSyncTask()); // AI 任务建议 Task::deliver(new AiTaskLoopTask()); + // 聊天附件历史索引回填 + Task::deliver(new MessageAttachmentBackfillTask()); return "success"; } diff --git a/app/Models/WebSocketDialogMsg.php b/app/Models/WebSocketDialogMsg.php index 909c2c224..99950a821 100644 --- a/app/Models/WebSocketDialogMsg.php +++ b/app/Models/WebSocketDialogMsg.php @@ -11,6 +11,7 @@ use App\Tasks\PushTask; use App\Models\ProjectTaskRelation; use App\Exceptions\ApiException; use App\Tasks\WebSocketDialogMsgTask; +use App\Services\MessageAttachmentService; use Hhxsv5\LaravelS\Swoole\Task\Task; use Illuminate\Database\Eloquent\SoftDeletes; @@ -1477,6 +1478,7 @@ class WebSocketDialogMsg extends AbstractModel ]; $dialogMsg->updateInstance($updateData); $dialogMsg->generateKeyAndSave($search_key); + MessageAttachmentService::syncSafely($dialogMsg, true); ProjectTaskRelation::recordMentionsFromMessage($dialogMsg); // WebSocketDialogUser::whereDialogId($dialog->id)->whereUserid($sender)->whereHide(1)->change([ @@ -1544,6 +1546,7 @@ class WebSocketDialogMsg extends AbstractModel 'updated_at' => Carbon::now()->toDateTimeString('millisecond'), ]); }); + MessageAttachmentService::syncSafely($dialogMsg); ProjectTaskRelation::recordMentionsFromMessage($dialogMsg); // $task = new WebSocketDialogMsgTask($dialogMsg->id); diff --git a/app/Models/WebSocketDialogMsgAttachment.php b/app/Models/WebSocketDialogMsgAttachment.php new file mode 100644 index 000000000..9599f362b --- /dev/null +++ b/app/Models/WebSocketDialogMsgAttachment.php @@ -0,0 +1,49 @@ + 'datetime', + 'completed_at' => 'datetime', + 'next_retry_at' => 'datetime', + ]; +} diff --git a/app/Services/MessageAttachmentBackfillService.php b/app/Services/MessageAttachmentBackfillService.php new file mode 100644 index 000000000..6a8c00615 --- /dev/null +++ b/app/Services/MessageAttachmentBackfillService.php @@ -0,0 +1,193 @@ +where('id', '>', $after) + ->where('id', '<=', $before) + ->where(function ($query) { + $query->where('type', 'file') + ->orWhere(function ($text) { + $text->where('type', 'text') + ->where(function ($images) { + $images->where('msg', 'like', '%browse%') + ->orWhere('msg', 'like', '%[:IMAGE:browse:%'); + }); + }); + }) + ->orderBy('id'); + } + + /** + * 在时间预算内推进自动回填；已有任务运行时直接跳过。 + */ + public static function processAutomatic( + int $batchSize = self::DEFAULT_BATCH_SIZE, + int $timeLimit = self::DEFAULT_TIME_LIMIT + ): array { + $batchSize = min(2000, max(1, $batchSize)); + $timeLimit = min(50, max(1, $timeLimit)); + $lockKey = config('app.env') . ':' . self::LOCK_KEY; + $lock = Cache::store('redis')->lock($lockKey, self::LOCK_SECONDS); + if (!$lock->get()) { + return ['status' => 'locked']; + } + + try { + return self::processState($batchSize, $timeLimit); + } finally { + $lock->release(); + } + } + + private static function processState(int $batchSize, int $timeLimit): array + { + $state = WebSocketDialogMsgAttachmentBackfill::query()->orderBy('id')->first(); + if (!$state) { + return ['status' => 'missing']; + } + if ($state->status === WebSocketDialogMsgAttachmentBackfill::STATUS_COMPLETED) { + return ['status' => WebSocketDialogMsgAttachmentBackfill::STATUS_COMPLETED]; + } + if ($state->next_retry_at && $state->next_retry_at->isFuture()) { + return ['status' => 'waiting']; + } + + $before = intval($state->before_msg_id); + if (!$state->started_at) { + // 迁移到新服务启动之间仍可能由旧进程写入消息，首次运行时补齐这段升级窗口。 + $before = max($before, intval(WebSocketDialogMsg::withTrashed()->max('id'))); + } + + $state->update([ + 'status' => WebSocketDialogMsgAttachmentBackfill::STATUS_RUNNING, + 'before_msg_id' => $before, + 'started_at' => $state->started_at ?: now(), + 'next_retry_at' => null, + ]); + + $startedAt = microtime(true); + $lastId = intval($state->last_msg_id); + $processed = intval($state->processed_count); + $attachmentCount = intval($state->attachment_count); + $emptyCount = intval($state->empty_count); + + while ($lastId < $before && microtime(true) - $startedAt < $timeLimit) { + $messages = self::candidateQuery($lastId, $before) + ->take($batchSize) + ->get(); + + if ($messages->isEmpty()) { + return self::complete($state, $before, $processed, $attachmentCount, $emptyCount); + } + + foreach ($messages as $message) { + try { + $count = MessageAttachmentService::sync($message); + } catch (\Throwable $e) { + self::fail($state, $lastId, $processed, $attachmentCount, $emptyCount, $e); + return [ + 'status' => WebSocketDialogMsgAttachmentBackfill::STATUS_FAILED, + 'last_msg_id' => $lastId, + ]; + } + + $lastId = intval($message->id); + $processed++; + $attachmentCount += $count; + if ($count === 0) { + $emptyCount++; + } + } + + $state->update([ + 'last_msg_id' => $lastId, + 'processed_count' => $processed, + 'attachment_count' => $attachmentCount, + 'empty_count' => $emptyCount, + 'retry_count' => 0, + 'last_error' => null, + ]); + } + + if ($lastId >= $before) { + return self::complete($state, $before, $processed, $attachmentCount, $emptyCount); + } + + $state->update(['status' => WebSocketDialogMsgAttachmentBackfill::STATUS_PENDING]); + return [ + 'status' => WebSocketDialogMsgAttachmentBackfill::STATUS_PENDING, + 'last_msg_id' => $lastId, + 'processed_count' => $processed, + 'attachment_count' => $attachmentCount, + ]; + } + + private static function complete( + WebSocketDialogMsgAttachmentBackfill $state, + int $before, + int $processed, + int $attachmentCount, + int $emptyCount + ): array { + $state->update([ + 'status' => WebSocketDialogMsgAttachmentBackfill::STATUS_COMPLETED, + 'last_msg_id' => $before, + 'processed_count' => $processed, + 'attachment_count' => $attachmentCount, + 'empty_count' => $emptyCount, + 'retry_count' => 0, + 'last_error' => null, + 'next_retry_at' => null, + 'completed_at' => now(), + ]); + + return [ + 'status' => WebSocketDialogMsgAttachmentBackfill::STATUS_COMPLETED, + 'last_msg_id' => $before, + 'processed_count' => $processed, + 'attachment_count' => $attachmentCount, + ]; + } + + private static function fail( + WebSocketDialogMsgAttachmentBackfill $state, + int $lastId, + int $processed, + int $attachmentCount, + int $emptyCount, + \Throwable $error + ): void { + $retryCount = intval($state->retry_count) + 1; + $retryMinutes = min(30, 2 ** min(4, $retryCount - 1)); + $state->update([ + 'status' => WebSocketDialogMsgAttachmentBackfill::STATUS_FAILED, + 'last_msg_id' => $lastId, + 'processed_count' => $processed, + 'attachment_count' => $attachmentCount, + 'empty_count' => $emptyCount, + 'failure_count' => intval($state->failure_count) + 1, + 'retry_count' => $retryCount, + 'last_error' => mb_substr($error->getMessage(), 0, 2000), + 'next_retry_at' => now()->addMinutes($retryMinutes), + ]); + } +} diff --git a/app/Services/MessageAttachmentService.php b/app/Services/MessageAttachmentService.php new file mode 100644 index 000000000..67e678ac2 --- /dev/null +++ b/app/Services/MessageAttachmentService.php @@ -0,0 +1,236 @@ +getRawOriginal('msg')); + $attachments = self::extract((string)$message->type, $messageData); + if (empty($attachments) && !$removeMissing) { + return 0; + } + + DB::transaction(function () use ($message, $attachments) { + if (empty($attachments)) { + WebSocketDialogMsgAttachment::whereMsgId($message->id)->delete(); + return; + } + + $now = now(); + $sourceType = $attachments[0]['source_type']; + $positions = []; + $rows = []; + foreach ($attachments as $attachment) { + $positions[] = $attachment['position']; + $rows[] = array_merge($attachment, [ + 'msg_id' => intval($message->id), + 'dialog_id' => intval($message->dialog_id), + 'created_at' => $message->created_at ?: $now, + 'updated_at' => $now, + ]); + } + + DB::table('web_socket_dialog_msg_attachments')->upsert( + $rows, + ['msg_id', 'source_type', 'position'], + ['dialog_id', 'kind', 'name', 'ext', 'path', 'thumb', 'size', 'width', 'height', 'updated_at'] + ); + + WebSocketDialogMsgAttachment::whereMsgId($message->id) + ->where(function ($query) use ($sourceType, $positions) { + $query->where('source_type', '!=', $sourceType) + ->orWhereNotIn('position', $positions); + }) + ->delete(); + }); + + return count($attachments); + } + + /** + * 双写失败不影响聊天主链路，历史回填命令可再次同步失败消息。 + */ + public static function syncSafely(WebSocketDialogMsg $message, bool $removeMissing = false): bool + { + if (!in_array($message->type, ['file', 'text'], true)) { + return true; + } + + try { + self::sync($message, $removeMissing); + return true; + } catch (\Throwable $e) { + Log::warning('Message attachment sync failed', [ + 'msg_id' => intval($message->id), + 'dialog_id' => intval($message->dialog_id), + 'message' => $e->getMessage(), + ]); + return false; + } + } + + private static function extractFileMessage(array $messageData): ?array + { + $path = self::normalizeStoredPath((string)($messageData['path'] ?? '')); + if ($path === '') { + return null; + } + + $ext = strtolower((string)($messageData['ext'] ?? self::pathExtension($path))); + $thumb = self::normalizeStoredPath((string)($messageData['thumb'] ?? '')); + + return [ + 'source_type' => WebSocketDialogMsgAttachment::SOURCE_FILE_MESSAGE, + 'kind' => in_array($ext, File::imageExt, true) + ? WebSocketDialogMsgAttachment::KIND_IMAGE + : WebSocketDialogMsgAttachment::KIND_FILE, + 'position' => 0, + 'name' => (string)($messageData['name'] ?? ''), + 'ext' => $ext, + 'path' => $path, + 'thumb' => $thumb, + 'size' => max(0, intval($messageData['size'] ?? 0)), + 'width' => max(0, intval($messageData['width'] ?? 0)), + 'height' => max(0, intval($messageData['height'] ?? 0)), + ]; + } + + private static function extractInlineImages(string $html): array + { + if ($html === '') { + return []; + } + + $attachments = []; + if (str_contains($html, 'loadHTML( + '
' . $html . '
', + LIBXML_HTML_NOIMPLIED | LIBXML_HTML_NODEFDTD | LIBXML_NOERROR | LIBXML_NOWARNING + ); + libxml_clear_errors(); + libxml_use_internal_errors($previous); + + if ($loaded) { + $nodes = (new DOMXPath($document))->query( + "//img[contains(concat(' ', normalize-space(@class), ' '), ' browse ')]" + ); + foreach ($nodes ?: [] as $node) { + if (!$node instanceof DOMElement) { + continue; + } + self::appendInlineImage($attachments, [ + 'src' => $node->getAttribute('src'), + 'width' => $node->getAttribute('width'), + 'height' => $node->getAttribute('height'), + 'name' => $node->getAttribute('alt'), + ]); + } + } + } + + if (empty($attachments)) { + preg_match_all('/\[:IMAGE:browse:([^:]*):([^:]*):(.*?):(.*?):\]/i', $html, $matches, PREG_SET_ORDER); + foreach ($matches as $match) { + self::appendInlineImage($attachments, [ + 'src' => $match[3] ?? '', + 'width' => $match[1] ?? 0, + 'height' => $match[2] ?? 0, + 'name' => $match[4] ?? '', + ]); + } + } + + foreach ($attachments as $position => &$attachment) { + $attachment['position'] = $position; + } + unset($attachment); + + return $attachments; + } + + private static function appendInlineImage(array &$attachments, array $image): void + { + $thumb = self::normalizeStoredPath(html_entity_decode(trim((string)$image['src']), ENT_QUOTES | ENT_HTML5)); + if ($thumb === '' || str_contains($thumb, 'images/other/imgerr.jpg')) { + return; + } + + $path = Base::thumbRestore($thumb); + $attachments[] = [ + 'source_type' => WebSocketDialogMsgAttachment::SOURCE_INLINE_IMAGE, + 'kind' => WebSocketDialogMsgAttachment::KIND_IMAGE, + 'position' => 0, + 'name' => trim(strip_tags((string)$image['name'])), + 'ext' => self::pathExtension($path), + 'path' => $path, + 'thumb' => $thumb, + 'size' => self::localFileSize($path), + 'width' => max(0, intval($image['width'])), + 'height' => max(0, intval($image['height'])), + ]; + } + + private static function normalizeStoredPath(string $path): string + { + $path = trim(str_replace('{{RemoteURL}}', '', $path)); + if ($path === '' || str_starts_with($path, 'data:')) { + return ''; + } + + $path = Base::unFillUrl($path); + if (str_starts_with($path, '/') && !str_starts_with($path, '//')) { + $path = ltrim($path, '/'); + } + return $path; + } + + private static function pathExtension(string $path): string + { + $urlPath = parse_url($path, PHP_URL_PATH); + return strtolower(pathinfo(urldecode((string)($urlPath ?: $path)), PATHINFO_EXTENSION)); + } + + private static function localFileSize(string $path): int + { + if ($path === '' || preg_match('/^[a-z][a-z0-9+.-]*:\/\//i', $path)) { + return 0; + } + + $file = public_path(ltrim($path, '/')); + return is_file($file) ? max(0, intval(filesize($file))) : 0; + } +} diff --git a/app/Tasks/MessageAttachmentBackfillTask.php b/app/Tasks/MessageAttachmentBackfillTask.php new file mode 100644 index 000000000..767a00222 --- /dev/null +++ b/app/Tasks/MessageAttachmentBackfillTask.php @@ -0,0 +1,22 @@ +bigIncrements('id'); + $table->unsignedBigInteger('msg_id')->comment('消息ID'); + $table->unsignedBigInteger('dialog_id')->comment('会话ID'); + $table->string('source_type', 20)->comment('来源类型：file_message、inline_image'); + $table->string('kind', 20)->comment('附件类型：file、image'); + $table->unsignedSmallInteger('position')->default(0)->comment('消息内附件顺序'); + $table->string('name')->default('')->comment('附件名称'); + $table->string('ext', 20)->default('')->comment('文件扩展名'); + $table->text('path')->nullable()->comment('原文件地址'); + $table->text('thumb')->nullable()->comment('缩略图地址'); + $table->unsignedBigInteger('size')->default(0)->comment('文件大小(B)'); + $table->unsignedInteger('width')->default(0)->comment('图片宽度'); + $table->unsignedInteger('height')->default(0)->comment('图片高度'); + $table->timestamps(); + + $table->unique(['msg_id', 'source_type', 'position'], 'uniq_ws_msg_attachment_position'); + $table->index(['dialog_id', 'id'], 'idx_ws_msg_attachment_dialog'); + $table->index(['kind', 'id'], 'idx_ws_msg_attachment_kind'); + }); + } + + public function down(): void + { + Schema::dropIfExists('web_socket_dialog_msg_attachments'); + } +}; diff --git a/database/migrations/2026_08_04_000002_create_web_socket_dialog_msg_attachment_backfills_table.php b/database/migrations/2026_08_04_000002_create_web_socket_dialog_msg_attachment_backfills_table.php new file mode 100644 index 000000000..b81567abf --- /dev/null +++ b/database/migrations/2026_08_04_000002_create_web_socket_dialog_msg_attachment_backfills_table.php @@ -0,0 +1,47 @@ +bigIncrements('id'); + $table->string('status', 20)->default('pending')->comment('状态：pending、running、failed、completed'); + $table->unsignedBigInteger('before_msg_id')->default(0)->comment('启动时消息快照上界'); + $table->unsignedBigInteger('last_msg_id')->default(0)->comment('已处理到的消息ID'); + $table->unsignedBigInteger('processed_count')->default(0)->comment('成功处理消息数'); + $table->unsignedBigInteger('attachment_count')->default(0)->comment('已索引附件数'); + $table->unsignedBigInteger('empty_count')->default(0)->comment('无有效附件消息数'); + $table->unsignedBigInteger('failure_count')->default(0)->comment('失败尝试次数'); + $table->unsignedInteger('retry_count')->default(0)->comment('当前消息连续重试次数'); + $table->text('last_error')->nullable()->comment('最近一次错误'); + $table->timestamp('started_at')->nullable()->comment('首次开始时间'); + $table->timestamp('completed_at')->nullable()->comment('完成时间'); + $table->timestamp('next_retry_at')->nullable()->comment('下次重试时间'); + $table->timestamps(); + + $table->index(['status', 'next_retry_at'], 'idx_ws_msg_attachment_backfill_pending'); + }); + + $beforeMsgId = intval(DB::table('web_socket_dialog_msgs')->max('id')); + $now = now(); + DB::table('web_socket_dialog_msg_attachment_backfills')->insert([ + 'status' => $beforeMsgId > 0 ? 'pending' : 'completed', + 'before_msg_id' => $beforeMsgId, + 'last_msg_id' => 0, + 'completed_at' => $beforeMsgId > 0 ? null : $now, + 'created_at' => $now, + 'updated_at' => $now, + ]); + } + + public function down(): void + { + Schema::dropIfExists('web_socket_dialog_msg_attachment_backfills'); + } +}; diff --git a/tests/Feature/MessageAttachmentBackfillServiceTest.php b/tests/Feature/MessageAttachmentBackfillServiceTest.php new file mode 100644 index 000000000..7b7fae33b --- /dev/null +++ b/tests/Feature/MessageAttachmentBackfillServiceTest.php @@ -0,0 +1,87 @@ +makeMessage('file', [ + 'name' => 'backfill.pdf', + 'ext' => 'pdf', + 'path' => 'uploads/test/backfill.pdf', + 'size' => 1024, + ]); + $imageMessage = $this->makeMessage('text', [ + 'text' => '

', + ]); + + $state = WebSocketDialogMsgAttachmentBackfill::query()->firstOrFail(); + $state->update([ + 'status' => WebSocketDialogMsgAttachmentBackfill::STATUS_PENDING, + // 模拟迁移后、服务重启前又产生了一条图片消息。 + 'before_msg_id' => $fileMessage->id, + 'last_msg_id' => $fileMessage->id - 1, + 'processed_count' => 0, + 'attachment_count' => 0, + 'empty_count' => 0, + 'failure_count' => 0, + 'retry_count' => 0, + 'last_error' => null, + 'started_at' => null, + 'completed_at' => null, + 'next_retry_at' => null, + ]); + + $result = MessageAttachmentBackfillService::processAutomatic(1, 5); + + $state->refresh(); + $this->assertSame(WebSocketDialogMsgAttachmentBackfill::STATUS_COMPLETED, $result['status']); + $this->assertSame(WebSocketDialogMsgAttachmentBackfill::STATUS_COMPLETED, $state->status); + $this->assertSame($imageMessage->id, $state->before_msg_id); + $this->assertSame($imageMessage->id, $state->last_msg_id); + $this->assertSame(2, $state->processed_count); + $this->assertSame(2, $state->attachment_count); + $this->assertNotNull($state->completed_at); + $this->assertSame(2, WebSocketDialogMsgAttachment::whereIn('msg_id', [ + $fileMessage->id, + $imageMessage->id, + ])->count()); + } + + public function test_automatic_backfill_waits_until_retry_time(): void + { + $state = WebSocketDialogMsgAttachmentBackfill::query()->firstOrFail(); + $state->update([ + 'status' => WebSocketDialogMsgAttachmentBackfill::STATUS_FAILED, + 'next_retry_at' => now()->addMinutes(5), + ]); + + $result = MessageAttachmentBackfillService::processAutomatic(1, 1); + + $this->assertSame('waiting', $result['status']); + $this->assertSame(WebSocketDialogMsgAttachmentBackfill::STATUS_FAILED, $state->fresh()->status); + } + + private function makeMessage(string $type, array $data): WebSocketDialogMsg + { + $message = WebSocketDialogMsg::createInstance([ + 'dialog_id' => 0, + 'userid' => 0, + 'type' => $type, + 'key' => '', + 'msg' => json_encode($data, JSON_UNESCAPED_UNICODE), + ]); + $message->save(); + return $message; + } +} diff --git a/tests/Unit/MessageAttachmentServiceTest.php b/tests/Unit/MessageAttachmentServiceTest.php new file mode 100644 index 000000000..f246fad57 --- /dev/null +++ b/tests/Unit/MessageAttachmentServiceTest.php @@ -0,0 +1,85 @@ + '方案.pdf', + 'ext' => 'PDF', + 'path' => 'uploads/chat/202608/1/plan.pdf', + 'thumb' => 'images/ext/pdf.png', + 'size' => 2048, + ]); + + $this->assertCount(1, $attachments); + $this->assertSame(WebSocketDialogMsgAttachment::SOURCE_FILE_MESSAGE, $attachments[0]['source_type']); + $this->assertSame(WebSocketDialogMsgAttachment::KIND_FILE, $attachments[0]['kind']); + $this->assertSame('pdf', $attachments[0]['ext']); + $this->assertSame(2048, $attachments[0]['size']); + } + + public function test_classifies_uploaded_image_as_image_attachment(): void + { + $attachments = MessageAttachmentService::extract('file', [ + 'name' => 'design.png', + 'ext' => 'png', + 'path' => 'uploads/chat/202608/1/design.png', + 'thumb' => 'uploads/chat/202608/1/design.png_thumb.jpg', + 'size' => 4096, + 'width' => 1200, + 'height' => 800, + ]); + + $this->assertSame(WebSocketDialogMsgAttachment::KIND_IMAGE, $attachments[0]['kind']); + $this->assertSame(1200, $attachments[0]['width']); + $this->assertSame(800, $attachments[0]['height']); + } + + public function test_extracts_each_browse_image_and_ignores_emoticons(): void + { + $html = <<<'HTML' +

+ 设计图 + 笑脸 + 重复图片 +

+HTML; + + $attachments = MessageAttachmentService::extract('text', ['text' => $html]); + + $this->assertCount(2, $attachments); + $this->assertSame([0, 1], array_column($attachments, 'position')); + $this->assertSame('uploads/chat/202608/1/design.png', $attachments[0]['path']); + $this->assertSame('uploads/chat/202608/1/design.png_thumb.jpg', $attachments[0]['thumb']); + $this->assertSame('设计图', $attachments[0]['name']); + $this->assertSame('重复图片', $attachments[1]['name']); + } + + public function test_extracts_legacy_image_token(): void + { + $attachments = MessageAttachmentService::extract('text', [ + 'text' => '

旧消息

[:IMAGE:browse:120:80:uploads/chat/202001/1/legacy.webp:旧截图:]', + ]); + + $this->assertCount(1, $attachments); + $this->assertSame('uploads/chat/202001/1/legacy.webp', $attachments[0]['path']); + $this->assertSame('webp', $attachments[0]['ext']); + $this->assertSame(120, $attachments[0]['width']); + $this->assertSame(80, $attachments[0]['height']); + } + + public function test_ignores_broken_placeholder_and_non_attachment_messages(): void + { + $this->assertSame([], MessageAttachmentService::extract('text', [ + 'text' => '', + ])); + $this->assertSame([], MessageAttachmentService::extract('meeting', ['name' => '周会'])); + } +}