mirror->available()) { throw new RuntimeException('jushuitan_cost_price_history_mirror_schema_missing'); } $state = $this->mirror->state(); if (! $full && ($state['initial_completed_at'] ?? null) === null) { return ['status' => 'requires_full_sync', 'mode' => 'incremental', 'pages' => 0, 'source_rows' => 0, 'rows' => 0]; } $mode = $full ? 'full' : 'incremental'; $resume = is_array($state) && ($state['run_mode'] ?? null) === $mode && in_array($state['status'] ?? null, ['running', 'failed'], true) && ($state['window_started_at'] ?? null) !== null && ($state['window_ends_at'] ?? null) !== null; $generation = $resume && is_string($state['generation'] ?? null) ? (string) $state['generation'] : (string) Str::uuid(); if ($resume) { $since = Carbon::parse((string) $state['window_started_at']); $until = Carbon::parse((string) $state['window_ends_at']); } elseif ($full) { $since = $this->historyStart(); $until = now(); } else { $overlap = max((int) config('services.jushuitan.cost_price_history_overlap_hours', 48), 1); $since = Carbon::parse((string) $state['last_succeeded_at'])->subHours($overlap); $until = now(); } $state = $this->mirror->startRun($mode, $generation, $since, $until, $resume); $windowStartedAt = ($state['cursor_window_started_at'] ?? null) === null ? $since->copy() : Carbon::parse((string) $state['cursor_window_started_at']); $page = max((int) ($state['cursor_page'] ?? 1), 1); $pageSize = min(max( $pageSize ?? (int) config('services.jushuitan.cost_price_history_page_size', 50), 1, ), 50); $maxPages = max( $maxPages ?? (int) config('services.jushuitan.cost_price_history_max_pages', 30), 0, ); $pages = 0; $processedSourceRows = (int) ($state['last_processed_source_rows'] ?? 0); $upsertedRows = (int) ($state['last_upserted_rows'] ?? 0); $windowDays = min(max( (int) config('services.jushuitan.cost_price_history_window_days', 7), 1, ), 7); try { while ($windowStartedAt->lt($until)) { $windowEndsAt = $windowStartedAt->copy()->addDays($windowDays); if ($windowEndsAt->gte($until)) { $windowEndsAt = $until->copy(); } $result = $this->upstream->getCostPriceOperationHistory( $windowStartedAt, $windowEndsAt, $page, $pageSize, ); $records = $result['records']; $processedSourceRows += $result['source_rows']; $upsertedRows += count($records); $pages++; $hasNextPage = $result['has_next']; // 相邻窗口在边界秒保留一次重叠,避免接口时间精度高于秒时漏掉记录; // operation_id 的幂等 upsert 会消除边界重复。 $nextWindowStartedAt = $hasNextPage ? $windowStartedAt->copy() : $windowEndsAt->copy(); $nextPage = $hasNextPage ? $page + 1 : 1; $this->mirror->storePage( $records, $nextWindowStartedAt, $nextPage, $generation, $processedSourceRows, $upsertedRows, ); $windowStartedAt = $nextWindowStartedAt; $page = $nextPage; if ($maxPages > 0 && $pages >= $maxPages && $windowStartedAt->lt($until)) { return [ 'status' => 'running', 'mode' => $mode, 'pages' => $pages, 'source_rows' => $processedSourceRows, 'rows' => $upsertedRows, ]; } } $this->mirror->complete( $mode, $generation, $until, $processedSourceRows, $upsertedRows, ); return [ 'status' => 'succeeded', 'mode' => $mode, 'pages' => $pages, 'source_rows' => $processedSourceRows, 'rows' => $upsertedRows, ]; } catch (Throwable $exception) { $message = trim($exception->getMessage()); $errorCode = preg_match('/^[a-z0-9_]+$/', $message) === 1 ? $message : $exception::class; $this->mirror->fail($errorCode); throw $exception; } } private function historyStart(): Carbon { $configured = trim((string) config( 'services.jushuitan.cost_price_history_start', '2000-01-01 00:00:00', )); return Carbon::parse($configured === '' ? '2000-01-01 00:00:00' : $configured); } }