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; } }