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(
+ '
![]()
+
+
+
+
旧消息
[: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' => '周会']));
+ }
+}