CaiWuXiTong/modules/EcommerceFinance/Repositories/StoreReceivingJdWalletBillRepository.php

<?php

namespace Modules\EcommerceFinance\Repositories;

use Carbon\CarbonImmutable;
use Illuminate\Database\DatabaseManager;
use Illuminate\Database\Query\Builder;
use Illuminate\Support\Facades\Cache;
use Modules\EcommerceFinance\Services\LedgerRemarkFilter;
use RuntimeException;

class StoreReceivingJdWalletBillRepository
{
    private const OPENING_BALANCE_CACHE_SECONDS = 300;

    public function __construct(private readonly DatabaseManager $databases) {}

    public function paginate(array $filters, string $sort, string $direction, int $page, int $perPage): array
    {
        $periods = $this->periods();
        $query = $this->query($filters, $periods);
        $summary = (clone $query)->selectRaw('COUNT(*) AS count, COALESCE(SUM(income), 0) AS income, COALESCE(SUM(expense), 0) AS expense')->first();
        $total = (int) $summary->count;
        $page = max(1, $page);
        $perPage = min(100, max(1, $perPage));
        // Skip deep pages using narrow indexed keys, then load JSON only for the visible rows.
        $keys = $this->ordered($query, $sort, $direction)->forPage($page, $perPage)->get(['generation', 'record_code']);
        $rows = [];
        foreach ($keys->groupBy('generation') as $generation => $group) {
            $records = $this->db()->table('jd_wallet_bill_rows')->where('generation', $generation)
                ->whereIn('record_code', $group->pluck('record_code')->all())->get(['record_code', 'row_data']);
            foreach ($records as $record) $rows[$generation . ':' . $record->record_code] = json_decode($record->row_data, true, 512, JSON_THROW_ON_ERROR);
        }
        $opening = $this->openingBalance($filters, $periods);
        $income = round((float) $summary->income, 2);
        $expense = round((float) $summary->expense, 2);
        return [
            'data' => $keys->map(fn ($key) => $rows[$key->generation . ':' . $key->record_code])->all(),
            'total' => $total, 'current_page' => $page, 'last_page' => max(1, (int) ceil($total / $perPage)), 'per_page' => $perPage,
            'summary' => [
                'count' => $total,
                'income' => $income,
                'expense' => $expense,
                'opening_balance' => $opening['balance'],
                'opening_balance_date' => $opening['date'],
                'opening_balance_source' => $opening['source'],
                'balance' => round($opening['balance'] + $income - $expense, 2),
            ],
            'sync' => ['status' => 'imported', 'source' => '胜算原始导出',
                'last_succeeded_at' => $periods->max('imported_at'),
                'periods' => $periods->map(fn ($p) => substr($p->period_start, 0, 7))->values()->all()],
        ];
    }

    public function exportRows(array $filters, string $sort, string $direction): iterable
    {
        $query = $this->ordered($this->query($filters, $this->periods()), $sort, $direction);
        return $this->stream($query);
    }

    private function stream(Builder $query): iterable
    {
        foreach ($query->select('row_data')->cursor() as $row) yield json_decode($row->row_data, true, 512, JSON_THROW_ON_ERROR);
    }

    public function filterOptions(): array
    {
        $q = $this->query([], $this->periods());
        return [
            'shops' => (clone $q)->distinct()->orderBy('store_name')->pluck('store_name')->all(),
            'bill_items' => (clone $q)->distinct()->orderBy('bill_item')->pluck('bill_item')->all(),
        ];
    }

    private function periods(): \Illuminate\Support\Collection
    {
        if (! $this->db()->getSchemaBuilder()->hasTable('jd_wallet_bill_periods')) throw new RuntimeException('jd_wallet_bill_mirror_not_ready');
        $periods = $this->db()->table('jd_wallet_bill_periods')->orderBy('period_start')->get();
        if ($periods->isEmpty()) throw new RuntimeException('jd_wallet_bill_mirror_not_ready');
        return $periods;
    }

    private function query(array $filters, \Illuminate\Support\Collection $periods): Builder
    {
        $q = $this->db()->table('jd_wallet_bill_rows')->whereIn('generation', $periods->pluck('active_generation')->all());
        foreach (['store_name', 'bill_item'] as $field) {
            if (($filters[$field] ?? '') !== '') $q->where($field, $filters[$field]);
        }
        LedgerRemarkFilter::itemsWhereIn($q, (array) ($filters['bill_items'] ?? []), 'bill_item');
        if (($filters['start'] ?? '') !== '') $q->where('occurred_at', '>=', $filters['start'] . ' 00:00:00');
        if (($filters['end'] ?? '') !== '') $q->where('occurred_at', '<', date('Y-m-d', strtotime($filters['end'] . ' +1 day')) . ' 00:00:00');
        if (($filters['keyword'] ?? '') !== '') {
            $like = '%' . str_replace(['\\', '%', '_'], ['\\\\', '\\%', '\\_'], $filters['keyword']) . '%';
            $q->where(fn (Builder $sub) => $sub->where('record_code', 'like', $like)->orWhere('order_no', 'like', $like)->orWhere('row_data', 'like', $like));
        }
        return $q;
    }

    private function ordered(Builder $q, string $sort, string $direction): Builder
    {
        $sort = in_array($sort, ['occurred_at', 'record_code', 'income', 'expense', 'balance'], true) ? $sort : 'occurred_at';
        $direction = $direction === 'asc' ? 'asc' : 'desc';
        return $q->orderBy($sort, $direction)->orderBy('record_code', $direction);
    }

    /**
     * 当前账期开始前，每个店铺账户最后一笔余额之和。
     *
     * 首个导入月份没有上月流水时，用当月首笔余额倒推期初，
     * 避免把没有上月数据误报成零余额。
     *
     * @return array{balance:float,date:?string,source:string}
     */
    private function openingBalance(array $filters, \Illuminate\Support\Collection $periods): array
    {
        $start = $this->date($filters['start'] ?? null) ?? $this->date($filters['end'] ?? null);
        if ($start === null) {
            return ['balance' => 0.0, 'date' => null, 'source' => 'none'];
        }

        $monthStart = $start->modify('first day of this month')->startOfDay();
        $nextMonthStart = $monthStart->modify('+1 month');
        $cutoff = $monthStart->format('Y-m-d H:i:s');
        $periodEnd = $nextMonthStart->format('Y-m-d H:i:s');
        $store = trim((string) ($filters['store_name'] ?? ''));
        $generations = $periods->pluck('active_generation')->filter()->sort()->values()->all();
        $cacheKey = 'jd-wallet-opening-balance:v1:' . hash('sha256', implode('|', [
            $cutoff,
            $store,
            implode(',', $generations),
        ]));

        $opening = Cache::remember(
            $cacheKey,
            self::OPENING_BALANCE_CACHE_SECONDS,
            function () use ($generations, $cutoff, $periodEnd, $store): array {
                $latestRows = $this->rankedBalances(
                    $generations,
                    $store,
                    '0000-01-01 00:00:00',
                    $cutoff,
                    'desc',
                );
                if ($latestRows->isNotEmpty()) {
                    return [
                        'balance' => round((float) $latestRows->sum('balance'), 2),
                        'source' => 'previous_period',
                    ];
                }

                $firstRows = $this->rankedBalances(
                    $generations,
                    $store,
                    $cutoff,
                    $periodEnd,
                    'asc',
                );
                if ($firstRows->isEmpty()) {
                    return ['balance' => 0.0, 'source' => 'none'];
                }

                return [
                    'balance' => round((float) $firstRows->sum(
                        static fn (object $row): float => (float) $row->balance
                            - (float) $row->income
                            + (float) $row->expense,
                    ), 2),
                    'source' => 'first_row_estimate',
                ];
            },
        );

        return [
            'balance' => (float) $opening['balance'],
            'date' => $monthStart->modify('-1 day')->format('Y-m-d'),
            'source' => (string) $opening['source'],
        ];
    }

    /**
     * @param list<string> $generations
     * @return \Illuminate\Support\Collection<int,object>
     */
    private function rankedBalances(
        array $generations,
        string $store,
        string $from,
        ?string $before,
        string $direction,
    ): \Illuminate\Support\Collection {
        $query = $this->db()->table('jd_wallet_bill_rows')
            ->whereIn('generation', $generations)
            ->whereNotNull('balance')
            ->where('occurred_at', '>=', $from)
            ->when($before !== null, fn (Builder $query) => $query->where('occurred_at', '<', $before))
            ->when($store !== '', fn (Builder $query) => $query->where('store_name', $store))
            ->select(['store_name', 'account_name', 'occurred_at', 'income', 'expense', 'balance', 'record_code']);

        $ranked = $query->selectRaw(
            'ROW_NUMBER() OVER (PARTITION BY store_name, account_name ORDER BY occurred_at '
            . $direction . ', LENGTH(record_code) ' . $direction . ', record_code ' . $direction
            . ') AS balance_rank',
        );

        return $this->db()->query()
            ->fromSub($ranked, 'ranked_balances')
            ->where('balance_rank', 1)
            ->get();
    }

    private function date(mixed $value): ?CarbonImmutable
    {
        $date = trim((string) $value);
        if (preg_match('/^\d{4}-\d{2}-\d{2}$/', $date) !== 1) {
            return null;
        }

        try {
            $parsed = CarbonImmutable::createFromFormat('!Y-m-d', $date);
        } catch (\Throwable) {
            return null;
        }

        return $parsed instanceof CarbonImmutable && $parsed->format('Y-m-d') === $date
            ? $parsed
            : null;
    }

    private function db(): \Illuminate\Database\Connection
    {
        return $this->databases->connection(config('upstream_mirrors.connection', 'business_mirror'));
    }
}