CaiWuXiTong/modules/EcommerceFinance/Repositories/ProductPriceMirrorRepository.php

<?php

namespace Modules\EcommerceFinance\Repositories;

use Illuminate\Database\ConnectionInterface;
use Illuminate\Database\DatabaseManager;
use Illuminate\Pagination\LengthAwarePaginator;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\Schema;

class ProductPriceMirrorRepository
{
    public const DATASET_KEY = 'supplier-payables.product-prices';

    private readonly ConnectionInterface $connection;

    public function __construct(DatabaseManager $database)
    {
        $this->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<string,mixed>|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)) . '%';
    }
}