CaiWuXiTong/modules/EcommerceFinance/Services/JushuitanCostPriceSnapshotDiffer.php

<?php

namespace Modules\EcommerceFinance\Services;

use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\Schema;
use Illuminate\Support\Carbon;
use Illuminate\Support\Str;
use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
use Modules\EcommerceFinance\Repositories\ProductCatalogMirrorRepository;
use Throwable;

/** 当前镜像的观测记录；不写正式操作日志，也不推测真实改价时间或历史价格。 */
class JushuitanCostPriceSnapshotDiffer
{
    public const DATASET_KEY = 'jushuitan.product-cost-price-observations';
    public const SNAPSHOTS = 'jushuitan_cost_price_observation_baselines';
    public const OBSERVATIONS = 'jushuitan_cost_price_observations';
    public const STATES = 'business_mirror_table_states';

    public function __construct(
        private readonly GenericBusinessMirrorRepository $mirror,
        private readonly BusinessMirrorDatasetRegistry $datasets,
    ) {}

    public function sync(int $batchSize = 500, int $maxBatches = 20): array
    {
        $lock = Cache::lock('jushuitan-cost-price-observations', 1800);
        if (! $lock->get()) {
            return ['status' => 'locked', 'rows' => 0, 'changes' => 0];
        }
        try {
            return $this->run(max(1, min(1000, $batchSize)), max(1, min(2000, $maxBatches)));
        } finally {
            $lock->release();
        }
    }

    private function run(int $batchSize, int $maxBatches): array
    {
        $db = $this->mirror->connection();
        $schema = Schema::connection($db->getName());
        $dataset = $this->datasets->get(ProductCatalogMirrorRepository::DATASET_KEY);
        if (! $this->mirror->ready($dataset)
            || ! $schema->hasTable(self::SNAPSHOTS) || ! $schema->hasTable(self::OBSERVATIONS)) {
            return ['status' => 'source_not_ready', 'rows' => 0, 'changes' => 0];
        }
        $source = $this->mirror->state($dataset);
        if (($source['status'] ?? '') !== 'succeeded') {
            return ['status' => 'source_not_ready', 'rows' => 0, 'changes' => 0];
        }
        $state = $db->table(self::STATES)->where('dataset_key', self::DATASET_KEY)->first();
        $resume = $state && in_array($state->status, ['running', 'failed'], true);
        $cursor = $resume ? json_decode($state->cursor ?? 'null', true) : null;
        // 固定窗口、两分钟重叠，兼容源镜像批次时间精度；重复读取不重复记变价。
        $from = $resume ? $state->window_started_at
            : ($state?->window_ends_at ? Carbon::parse($state->window_ends_at)->subMinutes(2)->format('Y-m-d H:i:s.u') : null);
        $until = $resume ? $state->window_ends_at : now()->subSeconds(2)->format('Y-m-d H:i:s.u');
        $db->table(self::STATES)->updateOrInsert(['dataset_key' => self::DATASET_KEY], [
            'source_key' => 'operations_replica', 'source_object' => 'mirror.' . $dataset->targetTable(),
            'target_table' => self::OBSERVATIONS, 'status' => 'running',
            'run_mode' => $state?->initial_completed_at ? 'incremental' : 'baseline',
            'generation' => $resume ? $state->generation : (string) Str::uuid(),
            'window_started_at' => $from, 'window_ends_at' => $until,
            'cursor' => $cursor ? json_encode($cursor) : null,
            'last_attempted_at' => now(), 'last_error_code' => null, 'updated_at' => now(),
        ]);
        $scanned = $changes = 0;
        try {
            for ($batch = 0; $batch < $maxBatches; $batch++) {
                $query = $db->table($dataset->targetTable())
                    ->select(['_mirror_key', '_mirror_synced_at', 'sku_id', 'cost_price', 'modified'])
                    ->where('sku_type', 'normal')->whereNotNull('cost_price')
                    ->where('_mirror_synced_at', '<=', $until);
                if ($from !== null) {
                    $query->where('_mirror_synced_at', '>=', $from);
                }
                if ($cursor) {
                    $query->where(static fn ($q) => $q->where('_mirror_synced_at', '>', $cursor['at'])
                        ->orWhere(static fn ($q) => $q->where('_mirror_synced_at', $cursor['at'])->where('_mirror_key', '>', $cursor['key'])));
                }
                $rows = $query->orderBy('_mirror_synced_at')->orderBy('_mirror_key')->limit($batchSize)->get();
                if ($rows->isEmpty()) {
                    $db->table(self::STATES)->where('dataset_key', self::DATASET_KEY)->update([
                        'status' => 'succeeded', 'cursor' => null, 'high_watermark' => json_encode(['at' => $until]),
                        'initial_completed_at' => $state?->initial_completed_at ?? now(),
                        'last_succeeded_at' => now(), 'updated_at' => now(),
                    ]);
                    return ['status' => 'succeeded', 'rows' => $scanned, 'changes' => $changes];
                }
                $previous = $db->table(self::SNAPSHOTS)->whereIn('source_key', $rows->pluck('_mirror_key'))->get()->keyBy('source_key');
                $snapshots = $events = [];
                $observedAt = now()->format('Y-m-d H:i:s.u');
                foreach ($rows as $row) {
                    $sku = trim((string) $row->sku_id);
                    if ($sku === '' || ! is_numeric($row->cost_price)) {
                        continue;
                    }
                    $price = number_format((float) $row->cost_price, 4, '.', '');
                    $old = $previous->get($row->_mirror_key);
                    if ($old && $old->sku_id === $sku && number_format((float) $old->cost_price, 4, '.', '') !== $price) {
                        $events[] = [
                            'id' => hash('sha256', $row->_mirror_key . '|' . $old->captured_at . '|' . $observedAt . '|' . $price),
                            'source_key' => $row->_mirror_key, 'sku_id' => $sku,
                            'before_price' => $old->cost_price, 'after_price' => $price,
                            'previous_observed_at' => $old->captured_at, 'observed_at' => $observedAt,
                            'source_modified_at' => $row->modified,
                        ];
                    }
                    $snapshots[] = ['source_key' => $row->_mirror_key, 'sku_id' => $sku, 'cost_price' => $price, 'captured_at' => $observedAt];
                }
                $last = $rows->last();
                $cursor = ['at' => $last->_mirror_synced_at, 'key' => $last->_mirror_key];
                // 价格基线、观测事件和游标同一事务；中断续传不会漏写或重复记账。
                $db->transaction(function () use ($db, $snapshots, $events, $cursor): void {
                    if ($snapshots) {
                        $db->table(self::SNAPSHOTS)->upsert($snapshots, ['source_key'], ['sku_id', 'cost_price', 'captured_at']);
                    }
                    if ($events) {
                        $db->table(self::OBSERVATIONS)->insert($events);
                    }
                    $db->table(self::STATES)->where('dataset_key', self::DATASET_KEY)->update([
                        'cursor' => json_encode($cursor), 'updated_at' => now(),
                    ]);
                });
                $scanned += $rows->count();
                $changes += count($events);
            }
            return ['status' => 'running', 'rows' => $scanned, 'changes' => $changes];
        } catch (Throwable $exception) {
            $db->table(self::STATES)->where('dataset_key', self::DATASET_KEY)->update([
                'status' => 'failed', 'last_error_code' => 'snapshot_comparison_failed', 'updated_at' => now(),
            ]);
            throw $exception;
        }
    }
}