CaiWuXiTong/modules/EcommerceFinance/Repositories/ProductPurchasePriceMirrorRepository.php

<?php

namespace Modules\EcommerceFinance\Repositories;

use Illuminate\Database\ConnectionInterface;
use Illuminate\Database\DatabaseManager;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\Schema;
use RuntimeException;

class ProductPurchasePriceMirrorRepository
{
    public const DATASET_KEY = 'product-inventory.purchase-order-prices';

    private readonly ConnectionInterface $connection;

    public function __construct(DatabaseManager $database)
    {
        $this->connection = $database->connection(
            (string) config('upstream_mirrors.connection', 'business_mirror')
        );
    }

    public function available(): bool
    {
        $schema = Schema::connection($this->connection->getName());

        return $schema->hasTable('product_purchase_price_sync_states')
            && $schema->hasTable('product_purchase_price_mirror_rows');
    }

    public function initialized(): bool
    {
        $state = $this->state();

        return is_array($state) && ($state['initial_completed_at'] ?? null) !== null;
    }

    /** @return array<string,mixed>|null */
    public function state(): ?array
    {
        if (! $this->available()) {
            return null;
        }

        $row = $this->connection->table('product_purchase_price_sync_states')
            ->where('dataset_key', self::DATASET_KEY)
            ->first();

        return $row === null ? null : (array) $row;
    }

    public function startRun(
        string $mode,
        string $generation,
        Carbon $since,
        Carbon $until,
        bool $resume,
    ): array {
        $now = now();
        $values = [
            'source_key' => (string) config('upstream_mirrors.product_purchase_prices.source_key', 'jiansun'),
            'source_object' => (string) config(
                'upstream_mirrors.product_purchase_prices.source_object',
                'dbo.jst_purchase + dbo.jst_purchaseItem',
            ),
            'status' => 'running',
            'run_mode' => $mode,
            'generation' => $generation,
            'last_attempted_at' => $now,
            'last_error_code' => null,
            'updated_at' => $now,
        ];

        if (! $resume) {
            $values += [
                'window_started_at' => $since,
                'window_ends_at' => $until,
                'cursor_updated_at' => null,
                'cursor_item_id' => null,
                'last_upserted_rows' => 0,
            ];
        }

        $query = $this->connection->table('product_purchase_price_sync_states')
            ->where('dataset_key', self::DATASET_KEY);

        if ($query->exists()) {
            $query->update($values);
        } else {
            $this->connection->table('product_purchase_price_sync_states')->insert(
                ['dataset_key' => self::DATASET_KEY, 'created_at' => $now] + $values,
            );
        }

        return $this->state() ?? [];
    }

    /** @param list<array<string,mixed>> $rows */
    public function storeBatch(array $rows, array $cursor, string $generation, int $cumulativeRows): void
    {
        $now = now();
        $mapped = array_map(function (array $row) use ($generation, $now): array {
            $values = [
                'item_id' => (int) $row['item_id'],
                'purchase_order_id' => (int) $row['purchase_order_id'],
                'sku_id' => $this->nullableString($row['sku_id'] ?? null),
                'style_id' => $this->nullableString($row['style_id'] ?? null),
                'unit_price' => $row['unit_price'] ?? null,
                'document_status' => $this->nullableString($row['document_status'] ?? null),
                'labels' => $this->nullableString($row['labels'] ?? null),
                'finished_at' => $row['finished_at'] ?? null,
                'purchase_updated_at' => $row['purchase_updated_at'],
            ];

            return $values + [
                '_mirror_generation' => $generation,
                'row_hash' => hash('sha256', json_encode(
                    $values,
                    JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION,
                )),
                '_mirror_synced_at' => $now,
                'created_at' => $now,
                'updated_at' => $now,
            ];
        }, $rows);

        $this->connection->transaction(function () use ($mapped, $cursor, $cumulativeRows, $now): void {
            $this->connection->table('product_purchase_price_mirror_rows')->upsert(
                $mapped,
                ['item_id'],
                [
                    'purchase_order_id', 'sku_id', 'style_id', 'unit_price',
                    'document_status', 'labels', 'finished_at', 'purchase_updated_at',
                    '_mirror_generation', 'row_hash', '_mirror_synced_at', 'updated_at',
                ],
            );

            $this->connection->table('product_purchase_price_sync_states')
                ->where('dataset_key', self::DATASET_KEY)
                ->update([
                    'cursor_updated_at' => $cursor['updated_at'],
                    'cursor_item_id' => $cursor['item_id'],
                    'last_upserted_rows' => $cumulativeRows,
                    'updated_at' => $now,
                ]);
        });
    }

    public function complete(string $mode, string $generation, Carbon $until, int $rows): void
    {
        $now = now();

        $this->connection->transaction(function () use ($mode, $generation, $until, $rows, $now): void {
            if ($mode === 'full') {
                $existingRows = $this->connection->table('product_purchase_price_mirror_rows')->count();
                if ($rows === 0 && $existingRows > 0) {
                    throw new RuntimeException('product_purchase_price_unexpected_empty_source');
                }

                $this->connection->table('product_purchase_price_mirror_rows')
                    ->where(function ($query) use ($generation): void {
                        $query->whereNull('_mirror_generation')
                            ->orWhere('_mirror_generation', '!=', $generation);
                    })
                    ->delete();
            }

            $values = [
                'status' => 'succeeded',
                'cursor_updated_at' => null,
                'cursor_item_id' => null,
                'last_succeeded_at' => $until,
                'last_upserted_rows' => $rows,
                'mirror_row_count' => $this->connection
                    ->table('product_purchase_price_mirror_rows')
                    ->count(),
                'last_error_code' => null,
                'updated_at' => $now,
            ];

            if ($mode === 'full') {
                $values['initial_completed_at'] = $now;
                $values['last_full_succeeded_at'] = $now;
            }

            $this->connection->table('product_purchase_price_sync_states')
                ->where('dataset_key', self::DATASET_KEY)
                ->update($values);
        });
    }

    public function fail(string $errorCode): void
    {
        $this->connection->table('product_purchase_price_sync_states')
            ->where('dataset_key', self::DATASET_KEY)
            ->update([
                'status' => 'failed',
                'last_error_code' => mb_substr($errorCode, 0, 120),
                'updated_at' => now(),
            ]);
    }

    private function nullableString(mixed $value): ?string
    {
        if ($value === null) {
            return null;
        }

        $value = trim((string) $value);

        return $value === '' ? null : $value;
    }
}