connection = $database->connection( (string) config('upstream_mirrors.connection', 'business_mirror') ); } public function ready(): bool { $schema = Schema::connection($this->connection->getName()); return $schema->hasTable('business_mirror_sync_states') && $schema->hasTable('product_price_mirror_rows'); } public function initialized(): bool { $state = $this->state(); return is_array($state) && ($state['initial_completed_at'] ?? null) !== null; } /** @return array|null */ public function state(): ?array { $row = $this->connection->table('business_mirror_sync_states') ->where('dataset_key', self::DATASET_KEY) ->first(); return $row === null ? null : (array) $row; } public function paginate(array $filters, int $page, int $limit): LengthAwarePaginator { $page = max($page, 1); $limit = min(max($limit, 1), 100); $query = $this->connection->table('product_price_mirror_rows'); if (($filters['barcode'] ?? '') !== '') { $query->whereRaw("barcode LIKE ? ESCAPE '!'", [$this->contains((string) $filters['barcode'])]); } if (($filters['completed_from'] ?? '') !== '') { $query->where('completed_at', '>=', (string) $filters['completed_from'] . ' 00:00:00'); } if (($filters['completed_to'] ?? '') !== '') { $query->where('completed_at', '<', Carbon::parse((string) $filters['completed_to'])->addDay()->startOfDay()); } return $query ->orderByDesc('completed_at') ->orderByDesc('business_id') ->orderBy('barcode') ->paginate($limit, [ 'barcode', 'spu', 'product_name', 'category', 'production_status', 'minimum_price', 'price_limit', 'purchase_price', 'shipping_fee', 'jingxi_limit_price', 'approval_status', 'approval_result', 'workflow_title', 'completed_at', ], 'page', $page); } public function startRun(string $mode, Carbon $since, Carbon $until, bool $resume): array { $now = now(); $values = [ 'source_key' => (string) config('upstream_mirrors.product_prices.source_key', 'operations_replica'), 'source_object' => (string) config('upstream_mirrors.product_prices.source_object', 'rpt.Bulk_Price_Limit'), 'status' => 'running', 'run_mode' => $mode, 'last_attempted_at' => $now, 'last_error_code' => null, 'updated_at' => $now, ]; if (! $resume) { $values += [ 'window_started_at' => $since, 'window_ends_at' => $until, 'cursor_completed_at' => null, 'cursor_business_id' => null, 'cursor_product_code' => null, 'last_upserted_rows' => 0, 'created_at' => $now, ]; } $query = $this->connection->table('business_mirror_sync_states') ->where('dataset_key', self::DATASET_KEY); if ($query->exists()) { $query->update($values); } else { $this->connection->table('business_mirror_sync_states')->insert( ['dataset_key' => self::DATASET_KEY, 'created_at' => $now] + $values, ); } return $this->state() ?? []; } public function storeBatch(array $rows, array $cursor, int $cumulativeRows): void { $now = now(); $mapped = array_map(function (array $row) use ($now): array { $businessId = (string) ($row['business_id'] ?? ''); $barcode = (string) ($row['barcode'] ?? ''); $cursorCompletedAt = (string) ($row['cursor_completed_at'] ?? $row['completed_at'] ?? ''); $values = [ 'business_id' => $businessId, 'barcode' => $barcode, 'spu' => $row['spu'] ?? null, 'product_name' => $row['product_name'] ?? null, 'category' => $row['category'] ?? null, 'production_status' => $row['production_status'] ?? null, 'minimum_price' => $row['minimum_price'] ?? null, 'price_limit' => $row['price_limit'] ?? null, 'purchase_price' => $row['purchase_price'] ?? null, 'shipping_fee' => $row['shipping_fee'] ?? null, 'jingxi_limit_price' => $row['jingxi_limit_price'] ?? null, 'approval_status' => $row['approval_status'] ?? null, 'approval_result' => $row['approval_result'] ?? null, 'workflow_title' => $row['workflow_title'] ?? null, 'completed_at' => $row['completed_at'], ]; return $values + [ 'upstream_key' => hash('sha256', $cursorCompletedAt . '|' . $businessId . '|' . $barcode), 'row_hash' => hash('sha256', json_encode($values, JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION)), 'synced_at' => $now, 'created_at' => $now, 'updated_at' => $now, ]; }, $rows); $this->connection->transaction(function () use ($mapped, $cursor, $cumulativeRows, $now): void { if ($mapped !== []) { $this->connection->table('product_price_mirror_rows')->upsert( $mapped, ['upstream_key'], [ 'business_id', 'barcode', 'spu', 'product_name', 'category', 'production_status', 'minimum_price', 'price_limit', 'purchase_price', 'shipping_fee', 'jingxi_limit_price', 'approval_status', 'approval_result', 'workflow_title', 'completed_at', 'row_hash', 'synced_at', 'updated_at', ], ); } $this->connection->table('business_mirror_sync_states') ->where('dataset_key', self::DATASET_KEY) ->update([ 'cursor_completed_at' => $cursor['completed_at'], 'cursor_business_id' => $cursor['business_id'], 'cursor_product_code' => $cursor['barcode'], 'last_upserted_rows' => $cumulativeRows, 'updated_at' => $now, ]); }); } public function complete(string $mode, Carbon $windowEnd, int $rows): void { $now = now(); $values = [ 'status' => 'succeeded', 'last_succeeded_at' => $windowEnd, 'last_upserted_rows' => $rows, 'cursor_completed_at' => null, 'cursor_business_id' => null, 'cursor_product_code' => null, 'last_error_code' => null, 'updated_at' => $now, ]; if ($mode === 'full') { $values['initial_completed_at'] = $now; } $this->connection->table('business_mirror_sync_states') ->where('dataset_key', self::DATASET_KEY) ->update($values); } public function fail(string $errorCode): void { $this->connection->table('business_mirror_sync_states') ->where('dataset_key', self::DATASET_KEY) ->update([ 'status' => 'failed', 'last_error_code' => mb_substr($errorCode, 0, 80), 'updated_at' => now(), ]); } private function contains(string $value): string { return '%' . str_replace(['!', '%', '_'], ['!!', '!%', '!_'], trim($value)) . '%'; } }