CaiWuXiTong/modules/EcommerceFinance/Repositories/CourierInformationMirrorRepository.php

<?php

namespace Modules\EcommerceFinance\Repositories;

use Carbon\CarbonInterface;
use Illuminate\Database\ConnectionInterface;
use Illuminate\Database\Query\Builder;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\Schema;
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
use RuntimeException;
use Throwable;

/**
 * 工作平台 2.0 的快递资料只读镜像。
 *
 * 这里不直接连接 JianSun；所有查询都落在 caiwu_business_mirror 中，
 * 由 business-mirror:sync-tables 负责生成和刷新镜像表。
 */
class CourierInformationMirrorRepository
{
    public const DATASET_KEY = 'jiansun.dbo.wp_express';

    public const NETWORK_DATASET_KEY = 'jiansun.dbo.wp_express_expNet';

    public const ACCOUNT_DATASET_KEY = 'jiansun.dbo.wp_express_account';

    public const LOG_DATASET_KEY = 'jiansun.dbo.wp_express_extension';

    private readonly ConnectionInterface $connection;

    public function __construct(
        private readonly BusinessMirrorDatasetRegistry $datasets,
        private readonly GenericBusinessMirrorRepository $mirror,
    ) {
        $this->connection = $mirror->connection();
    }

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

        return $schema->hasTable('business_mirror_table_states')
            && $schema->hasTable($this->dataset()->targetTable());
    }

    public function ready(): bool
    {
        return $this->mirror->ready($this->dataset());
    }

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

    /**
     * @param array<string,mixed>|null $state
     * @return array{code:string,lag_minutes:?int}
     */
    public function syncStatus(?array $state = null, ?CarbonInterface $at = null): array
    {
        $state ??= $this->state();
        if (! is_array($state)) {
            return ['code' => 'unavailable', 'lag_minutes' => null];
        }

        $lastSucceeded = $this->timestamp($state['last_succeeded_at'] ?? null);
        $lagMinutes = $lastSucceeded === null
            ? null
            : (int) ceil(max(0, ($at ?? now())->getTimestamp() - $lastSucceeded) / 60);
        $status = (string) ($state['status'] ?? 'pending');

        if (($state['initial_completed_at'] ?? null) === null) {
            $code = $status === 'failed'
                ? 'initial_failed'
                : ($status === 'running' ? 'initializing' : 'pending');
        } elseif ($status === 'failed') {
            $code = 'failed';
        } elseif ($status === 'running') {
            $code = ($state['run_mode'] ?? null) === 'full' ? 'reconciling' : 'syncing';
        } elseif ($lagMinutes === null
            || $lagMinutes > max((int) config('upstream_mirrors.courier_information.stale_after_minutes', 60), 1)) {
            $code = 'delayed';
        } else {
            $code = 'healthy';
        }

        return ['code' => $code, 'lag_minutes' => $lagMinutes];
    }

    /**
     * @param array<string,string> $filters
     * @return array{data:list<object>,total:int}
     */
    public function paginate(array $filters, int $page, int $perPage): array
    {
        if (! $this->ready()) {
            throw new RuntimeException('business_mirror_dataset_not_ready');
        }

        $page = max($page, 1);
        $perPage = min(max($perPage, 1), 100);
        $query = $this->filteredExpressQuery($filters);
        $total = (clone $query)->count();
        $rows = $query->orderBy('Id')
            ->forPage($page, $perPage)
            ->get()
            ->all();

        return ['data' => $rows, 'total' => $total];
    }

    /**
     * 快递资料中的去重快递公司名称，按资料 Id 的首次出现顺序返回。
     *
     * @return list<string>
     */
    public function names(): array
    {
        if (! $this->ready()) {
            throw new RuntimeException('business_mirror_dataset_not_ready');
        }

        return $this->connection
            ->table($this->dataset()->targetTable())
            ->select(['Id', 'expname'])
            ->whereNotNull('expname')
            ->orderBy('Id')
            ->get()
            ->map(static fn ($row): string => trim((string) ($row->expname ?? '')))
            ->filter(static fn (string $name): bool => $name !== '')
            ->unique()
            ->values()
            ->all();
    }

    /**
     * @param array<string,string> $filters
     * @return \Generator<int,object>
     */
    public function exportRows(array $filters, int $batchSize = 500): \Generator
    {
        if (! $this->ready()) {
            throw new RuntimeException('business_mirror_dataset_not_ready');
        }

        $batchSize = min(max($batchSize, 1), 5000);
        $lastId = null;

        while (true) {
            $query = $this->filteredExpressQuery($filters);
            if ($lastId !== null) {
                $query->where('Id', '>', $lastId);
            }

            $rows = $query->orderBy('Id')->limit($batchSize)->get()->all();
            if ($rows === []) {
                return;
            }

            foreach ($rows as $row) {
                yield $row;
            }

            $lastId = (int) $rows[array_key_last($rows)]->Id;
            if (count($rows) < $batchSize) {
                return;
            }
        }
    }

    /**
     * 网点标签数据。快递编码为空时返回全部网点；传入编码时只返回对应网点。
     * @return array{data:list<object>,total:int}
     */
    public function paginateNetworks(?string $expcode, int $page, int $perPage): array
    {
        $dataset = $this->datasets->get(self::NETWORK_DATASET_KEY);
        if (! $this->mirror->ready($dataset)) {
            throw new RuntimeException('business_mirror_dataset_not_ready');
        }

        $query = $this->connection->table($dataset->targetTable())->select([
            'id', 'expid', 'expcode', 'netcode', 'netno', 'netname', 'isbill',
            'province', 'city', 'area', 'town', 'address', 'linker', 'linkermobile',
            'status', 'apitype', 'apipltmname', 'created', 'expnum',
        ]);
        if ($expcode !== null && $expcode !== '') {
            $query->where('expcode', $expcode);
        }

        $page = max($page, 1);
        $perPage = min(max($perPage, 1), 100);
        $total = (clone $query)->count();

        return [
            'data' => $query->orderByDesc('id')->forPage($page, $perPage)->get()->all(),
            'total' => $total,
        ];
    }

    /**
     * 打款账户标签数据。
     * @return array{data:list<object>,total:int}
     */
    public function paginateAccounts(?string $expcode, int $page, int $perPage): array
    {
        $dataset = $this->datasets->get(self::ACCOUNT_DATASET_KEY);
        if (! $this->mirror->ready($dataset)) {
            throw new RuntimeException('business_mirror_dataset_not_ready');
        }

        $query = $this->connection->table($dataset->targetTable())->select([
            'Id', 'expcode', 'expname', 'accountType', 'account', 'accountName',
            'accountkhh', 'createTime', 'invoice', 'erpId', 'erpAccountId',
        ]);
        if ($expcode !== null && $expcode !== '') {
            $query->where('expcode', $expcode);
        }

        $page = max($page, 1);
        $perPage = min(max($perPage, 1), 100);
        $total = (clone $query)->count();

        return [
            'data' => $query->orderByDesc('Id')->forPage($page, $perPage)->get()->all(),
            'total' => $total,
        ];
    }

    /**
     * 快递总览用：工作平台全部快递 + 按快递代码归组的打款账户。镜像未就绪返回 null。
     *
     * @return array{express:list<object>,accounts:array<string,list<object>>}|null
     */
    public function overviewSource(): ?array
    {
        if (! $this->available() || ! $this->ready()) {
            return null;
        }

        $express = $this->connection->table($this->dataset()->targetTable())
            ->select(['Id', 'expcode', 'expname', 'status', 'linker', 'linkmobile', 'paymethod', 'is_delete', 'modified'])
            ->orderBy('Id')
            ->get()
            ->filter(static fn ($row): bool => trim((string) ($row->expname ?? '')) !== '' && (string) ($row->is_delete ?? '0') !== '1')
            ->values()
            ->all();

        $accounts = [];
        $accountDataset = $this->datasets->get(self::ACCOUNT_DATASET_KEY);
        if ($this->mirror->ready($accountDataset)) {
            foreach ($this->connection->table($accountDataset->targetTable())
                ->select(['expcode', 'expname', 'accountType', 'account', 'accountName', 'accountkhh', 'erpAccountId'])
                ->orderBy('Id')
                ->get() as $row) {
                $accounts[(string) $row->expcode][] = $row;
            }
        }

        return ['express' => $express, 'accounts' => $accounts];
    }

    /**
     * 返回指定快递名称对应的电子签章经办人账号。
     *
     * 工作平台将该值保存在 wp_express.dzqzjbr 中，不能用联系人电话
     * 代替；两者在部分快递资料中可能不同。镜像未完成首次同步时返回空
     * 数组，避免影响协议签署页面的正常打开。
     *
     * @param list<string> $expnames
     * @return array<string,string>
     */
    public function signingAccountsByName(array $expnames): array
    {
        if (! $this->ready()) {
            return [];
        }

        $expnames = array_values(array_unique(array_filter(array_map(
            static fn ($name): string => trim((string) $name),
            $expnames,
        ), static fn (string $name): bool => $name !== '')));

        if ($expnames === []) {
            return [];
        }

        return $this->connection
            ->table($this->dataset()->targetTable())
            ->select(['expname', 'dzqzjbr', 'Id'])
            ->whereIn('expname', $expnames)
            ->orderBy('Id')
            ->get()
            ->groupBy(static fn ($row): string => (string) ($row->expname ?? ''))
            ->map(static function ($rows): string {
                foreach ($rows as $row) {
                    $account = trim((string) ($row->dzqzjbr ?? ''));
                    if ($account !== '') {
                        return $account;
                    }
                }

                return '';
            })
            ->filter(static fn (string $account): bool => $account !== '')
            ->all();
    }

    /**
     * 快递资料操作日志。
     * @return array{data:list<object>,total:int}
     */
    public function paginateLogs(?string $expcode, int $page, int $perPage): array
    {
        $dataset = $this->datasets->get(self::LOG_DATASET_KEY);
        if (! $this->mirror->ready($dataset)) {
            throw new RuntimeException('business_mirror_dataset_not_ready');
        }

        $query = $this->connection->table($dataset->targetTable())->select([
            'Id', 'expcode', 'doTime', 'createTime', 'modifyTime',
        ]);
        if ($expcode !== null && $expcode !== '') {
            $query->where('expcode', $expcode);
        }

        $page = max($page, 1);
        $perPage = min(max($perPage, 1), 100);
        $total = (clone $query)->count();

        return [
            'data' => $query->orderByDesc('Id')->forPage($page, $perPage)->get()->all(),
            'total' => $total,
        ];
    }

    /** @return array{main:bool,networks:bool,accounts:bool,logs:bool,network_ready:bool,account_ready:bool,log_ready:bool} */
    public function relatedAvailability(): array
    {
        try {
            $schema = Schema::connection($this->connection->getName());
            $networkDataset = $this->datasets->get(self::NETWORK_DATASET_KEY);
            $accountDataset = $this->datasets->get(self::ACCOUNT_DATASET_KEY);
            $logDataset = $this->datasets->get(self::LOG_DATASET_KEY);

            return [
                'main' => $schema->hasTable($this->dataset()->targetTable()),
                'networks' => $schema->hasTable($networkDataset->targetTable()),
                'accounts' => $schema->hasTable($accountDataset->targetTable()),
                'logs' => $schema->hasTable($logDataset->targetTable()),
                'network_ready' => $this->mirror->ready($networkDataset),
                'account_ready' => $this->mirror->ready($accountDataset),
                'log_ready' => $this->mirror->ready($logDataset),
            ];
        } catch (Throwable) {
            return ['main' => false, 'networks' => false, 'accounts' => false, 'logs' => false, 'network_ready' => false, 'account_ready' => false, 'log_ready' => false];
        }
    }

    /** @param array<string,string> $filters */
    private function filteredExpressQuery(array $filters): Builder
    {
        $query = $this->connection->table($this->dataset()->targetTable())->select([
            'Id', 'expcode', 'expname', 'outcode', 'scexpcode', 'status', 'paymethod',
            'addvalidsgn', 'kuaiyunsgn', 'autonosgn', 'packnosgn', 'skucountsplitsgn',
            'weight0sgn', 'stopdate', 'tradebtime', 'tradeetime', 'signcom', 'signmode',
            'linker', 'linkmobile', 'paytype', 'undefined1', 'undefined2', 'undefined3',
            'volume0sgn', 'handnosgn', 'decryptsgn', 'pj1sgn', 'undefinedtype',
            'undefined4', 'undefined5', 'dzqzjbr', 'guid', 'is_delete', 'exp_properties',
            'modified',
        ]);

        if (($filters['expcode'] ?? '') !== '') {
            $query->where('expcode', $filters['expcode']);
        }
        if (($filters['expname'] ?? '') !== '') {
            $query->whereRaw("expname LIKE ? ESCAPE '!'", [$this->contains($filters['expname'])]);
        }
        foreach (['undefined1', 'undefined2', 'undefined3'] as $column) {
            if (($filters[$column] ?? '') !== '') {
                $query->whereRaw($column . " LIKE ? ESCAPE '!'", [$this->contains($filters[$column])]);
            }
        }
        if (($filters['tradebtime_from'] ?? '') !== '') {
            $query->whereDate('tradebtime', '>=', $filters['tradebtime_from']);
        }
        if (($filters['tradebtime_to'] ?? '') !== '') {
            $query->whereDate('tradebtime', '<=', $filters['tradebtime_to']);
        }
        if (($filters['status'] ?? '') !== '') {
            $query->where('status', (int) $filters['status']);
        }

        return $query;
    }

    private function dataset(): \Modules\EcommerceFinance\Contracts\BusinessMirrorDataset
    {
        return $this->datasets->get(self::DATASET_KEY);
    }

    private function contains(string $value): string
    {
        return '%' . str_replace(['!', '%', '_'], ['!!', '!%', '!_'], trim($value)) . '%';
    }

    private function timestamp(mixed $value): ?int
    {
        if (! is_string($value) || $value === '') {
            return null;
        }

        try {
            return Carbon::parse($value)->getTimestamp();
        } catch (Throwable) {
            return null;
        }
    }
}