bot-28/app/Services/Reporting/ReportSync.php

<?php

namespace App\Services\Reporting;

use App\Services\Sharing\BetFactSynchronizer;
use App\Services\Sharing\EffectiveBet as Money;
use Carbon\Carbon;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;

/** Reporting projections only. Never settles orders, moves money or publishes sharing facts. */
class ReportSync
{
    public const ORDER_SOURCES = ['lhc_order', 'sport_game_order', 'sport_order', 'third_game_orders', 'bets', 'jisu_game_order'];
    public const MONEY_SOURCES = ['balance_logs', 'withdraws', 'payment_orders', 'agent_withdrawals', 'agent_balance_logs', 'share_ledger', 'agent_rebate_records'];
    public const RECHECK_BATCH = 200;

    public function sync(bool $full = false): array
    {
        $result = ['scanned' => 0, 'published' => 0, 'errors' => 0, 'sources' => []];
        $lotteryParents = [];
        foreach (array_merge(self::ORDER_SOURCES, self::MONEY_SOURCES, ['lhc_order_item']) as $source) {
            if (!Schema::hasTable($source)) { $result['sources'][$source] = 'absent'; continue; }
            $start = now('Asia/Shanghai')->format('Y-m-d H:i:s');
            $state = DB::table('reporting_sync_state')->where('source', $source)->first();
            $cursor = $state ? json_decode($state->state, true, 512, JSON_THROW_ON_ERROR) : null;
            $column = $source === 'sport_order' ? 'update_time' : 'updated_at';
            $hasUpdate = Schema::hasColumn($source, $column);
            $max = (int) DB::table($source)->max('id');
            $scan = function ($query) use ($source, &$result, &$lotteryParents) {
                $query->orderBy('id')->chunkById(200, function ($rows) use ($source, &$result, &$lotteryParents) {
                    $this->processRows($source, $rows, $result, $lotteryParents);
                });
            };
            $recheckId = 0;
            if ($full || !$cursor) {
                $scan(DB::table($source)->where('id', '<=', $max));
            } else {
                $lastId = min((int) $cursor['last_id'], $max);
                // Keep the ID and time ranges separate so either can use its own index.
                $scan(DB::table($source)->where('id', '>', $lastId)->where('id', '<=', $max));
                $cut = Carbon::parse($cursor['checked_at'], 'Asia/Shanghai')->subMinutes(5);
                $value = in_array($source, ['lhc_order', 'lhc_order_item', 'sport_order'], true) ? $cut->timestamp : $cut->format('Y-m-d H:i:s');
                if ($hasUpdate) $scan(DB::table($source)->where('id', '<=', $lastId)->where($column, '>=', $value));

                // A bad row must not pin the whole source's high-water mark. Retry errors independently.
                $retryIds = DB::table('reporting_errors')->where('source', $source)->orderBy('updated_at')->orderBy('id')
                    ->limit(self::RECHECK_BATCH)->pluck('source_id')->all();
                if ($retryIds) {
                    // Missing source rows remain visible as errors, but cannot starve the rest of the retry queue.
                    DB::table('reporting_errors')->where('source', $source)->whereIn('source_id', $retryIds)->update(['updated_at' => $start]);
                    $retry = DB::table($source)->whereIn('id', $retryIds)->where('id', '<=', $lastId);
                    if ($hasUpdate) $retry->where(fn ($q) => $q->whereNull($column)->orWhere($column, '<', $value));
                    $scan($retry);
                }

                // Missing timestamps still receive eventual correction checks, without a full scan each minute.
                $recheck = DB::table($source)->where('id', '>', (int) ($cursor['recheck_id'] ?? 0))->where('id', '<=', $lastId);
                if ($hasUpdate) $recheck->whereNull($column);
                if ($retryIds) $recheck->whereNotIn('id', $retryIds);
                $rows = $recheck->orderBy('id')->limit(self::RECHECK_BATCH)->get();
                $this->processRows($source, $rows, $result, $lotteryParents);
                $recheckId = $rows->count() === self::RECHECK_BATCH ? (int) $rows->last()->id : 0;
            }
            DB::table('reporting_sync_state')->updateOrInsert(['source' => $source], ['state' => json_encode([
                'checked_at' => $start, 'last_id' => $max, 'recheck_id' => $recheckId,
            ]), 'updated_at' => $start]);
            $errors = DB::table('reporting_errors')->where('source', $source)->count();
            $result['errors'] += $errors;
            $result['sources'][$source] = $errors ? 'errors' : 'ready';
        }
        DB::table('reporting_sync_state')->updateOrInsert(['source' => '__run__'], ['state' => json_encode(['complete' => $result['errors'] === 0, 'sources' => $result['sources']]), 'updated_at' => now()]);
        return $result;
    }

    private function processRows(string $source, $rows, array &$result, array &$lotteryParents): void
    {
        if ($rows->isEmpty()) return;
        $parents = $source === 'lhc_order_item'
            ? DB::table('lhc_order')->whereIn('ordernum', $rows->pluck('ordernum')->unique()->all())->get(['id', 'ordernum'])->groupBy('ordernum') : null;
        $successfulItems = [];
        foreach ($rows as $row) {
            ++$result['scanned'];
            try {
                $publishSource = $source; $id = (int) $row->id;
                if ($source === 'lhc_order_item') {
                    $matches = $parents->get($row->ordernum);
                    if (!$matches || $matches->count() !== 1) throw new \RuntimeException('六合彩子单对应的父单不明确');
                    $publishSource = 'lhc_order'; $id = (int) $matches->first()->id;
                }
                if ($publishSource === 'lhc_order' && array_key_exists($id, $lotteryParents)) {
                    if ($lotteryParents[$id] !== null) throw new \RuntimeException($lotteryParents[$id]);
                } else {
                    try { $this->publish($publishSource, $id); }
                    catch (\Throwable $e) {
                        if ($publishSource === 'lhc_order') $lotteryParents[$id] = $e->getMessage();
                        throw $e;
                    }
                    if ($publishSource === 'lhc_order') $lotteryParents[$id] = null;
                    ++$result['published'];
                }
                if ($source === 'lhc_order_item') $successfulItems[] = (string) $row->id;
            } catch (\Throwable $e) {
                DB::table('reporting_errors')->updateOrInsert(['source' => $source, 'source_id' => (string) $row->id],
                    ['candidate_at' => $this->time($row->created_at ?? $row->create_time ?? null), 'reason' => mb_substr($e->getMessage(), 0, 500), 'created_at' => now(), 'updated_at' => now()]);
            }
        }
        if ($successfulItems) DB::table('reporting_errors')->where('source', $source)->whereIn('source_id', $successfulItems)->delete();
    }

    public function publish(string $source, int $id): void
    {
        if (!in_array($source, array_merge(self::ORDER_SOURCES, self::MONEY_SOURCES), true)) throw new \InvalidArgumentException('不支持的数据来源');
        DB::transaction(function () use ($source, $id) {
            $record = DB::table($source)->where('id', $id)->lockForUpdate()->first();
            if (!$record) throw new \RuntimeException('原记录不存在');
            $row = (array) $record;
            $member = $row['member_id'] ?? null;
            $virtual = $member !== null && DB::table('users')->where('member_id', (string) $member)->where('from', 2)->exists();
            if ($virtual) {
                DB::table('reporting_orders')->where('parent_source', $source)->where('parent_id', (string) $id)->delete();
                DB::table('reporting_events')->where('source', $source)->where('source_id', (string) $id)->delete();
                DB::table('reporting_errors')->where('source', $source)->where('source_id', (string) $id)->delete(); return;
            }
            if (in_array($source, self::ORDER_SOURCES, true)) $this->orders($source, $row);
            else $this->money($source, $row);
            DB::table('reporting_errors')->where('source', $source)->where('source_id', (string) $id)->delete();
        });
    }

    private function orders(string $source, array $row): void
    {
        $at = $this->time($row['created_at'] ?? $row['create_time'] ?? null);
        if ((int) ($row['is_faker'] ?? 0) === 1) { DB::table('reporting_orders')->where('parent_source', $source)->where('parent_id', (string) $row['id'])->delete(); return; }
        $facts = [];
        if (in_array($source, ['bets', 'jisu_game_order'], true)) {
            $identity = ReportContext::identity($row, $at ?? now()->toDateTimeString(), $source);
            if ($identity['user_id'] === null || $at === null) throw new \RuntimeException('旧订单缺少会员或下单时间');
            $bet = Money::decimal($row['amount']);
            $settled = $source === 'bets' ? (int) $row['status'] === 2 : in_array((int) $row['status'], [1, 2], true);
            $void = (int) $row['status'] === 3;
            $payout = $settled ? Money::decimal($source === 'bets' ? ($row['profit'] ?? 0) : ($row['win_amount'] ?? 0)) : Money::zero();
            $facts[] = ['source' => $source, 'source_id' => (string) $row['id'], 'user_id' => $identity['user_id'],
                'category' => 'self:lottery', 'bet_at' => $at, 'bet_amount' => $bet, 'settled' => $settled, 'voided' => $void,
                'net_profit' => $settled ? bcsub($payout, $bet, 10) : Money::zero(), 'currency' => $this->currency()];
        } else $facts = (new BetFactSynchronizer())->projectForReporting($source, (object) $row);
        $context = DB::table('reporting_contexts')->where('source', $source)->where('source_id', (string) $row['id'])->first();
        $lotteryPayoutAt = $source === 'lhc_order' && empty($context->succeeded_at)
            ? $this->lotteryPayoutTime($row, $facts) : null;
        $keys = [];
        foreach ($facts as $fact) {
            $identity = ReportContext::identity($row + ['report_user_id' => $fact['user_id']], $fact['bet_at'], $source);
            $settledAt = $context->succeeded_at ?? null;
            if (!(bool) ($context->settlement_verified ?? false)) {
                if ($source === 'lhc_order') $settledAt = $settledAt ?? $lotteryPayoutAt;
                if ($source === 'third_game_orders') $settledAt = $this->time($row['last_update_time'] ?? null);
                if ($source === 'sport_order') $settledAt = $this->time($row['settlement_time'] ?? null) ?? $this->time($row['return_operation_time'] ?? null) ?? $settledAt;
            }
            $settledAt = $fact['settled'] || $fact['voided'] ? $settledAt : null;
            $payout = $fact['settled'] ? bcadd($fact['bet_amount'], $fact['net_profit'], 10) : Money::zero();
            $refund = $fact['voided'] ? $fact['bet_amount'] : Money::zero();
            if ($source === 'sport_order' && $fact['settled']) {
                $refund = Money::decimal($row['return_amount'] ?? 0);
                if (bccomp($refund, '0', 10) < 0 || bccomp($refund, $fact['bet_amount'], 10) > 0) throw new \RuntimeException('部分退款金额不正确');
                $payout = bcsub($payout, $refund, 10);
            }
            if (bccomp($payout, '0', 10) < 0) throw new \RuntimeException('派奖金额不能为负');
            $key = ['source' => $fact['source'], 'source_id' => $fact['source_id']]; $keys[] = $fact['source_id'];
            $values = $identity + ['parent_source' => $source, 'parent_id' => (string) $row['id'], 'category' => $fact['category'],
                'currency' => $source === 'third_game_orders' ? $fact['currency'] : $this->currency(), 'bet_at' => $fact['bet_at'], 'settled_at' => $settledAt,
                'settlement_candidate_at' => $this->time($row['updated_at'] ?? $row['update_time'] ?? null),
                'bet_amount' => $fact['bet_amount'], 'payout_amount' => $payout, 'net_profit' => $fact['net_profit'],
                'refund_amount' => $refund, 'settled' => (int) $fact['settled'], 'voided' => (int) $fact['voided'],
                'time_quality' => (!$fact['settled'] && !$fact['voided']) || $settledAt !== null ? 'confirmed' : 'missing_settlement_time'];
            $this->writeProjection('reporting_orders', $key, $values);
        }
        DB::table('reporting_orders')->where('parent_source', $source)->where('parent_id', (string) $row['id'])->when($keys, fn ($q) => $q->whereNotIn('source_id', $keys))->delete();
    }

    /** Restore missing win times only from matching, balanced wallet-credit receipts. */
    private function lotteryPayoutTime(array $order, array $facts): ?string
    {
        if (!$facts || !Schema::hasTable('balance_logs') || !isset($order['win_amount'])) return null;
        $expected = Money::zero();
        foreach ($facts as $fact) {
            if (!$fact['settled'] || $fact['voided']) return null;
            $expected = bcadd($expected, bcadd($fact['bet_amount'], $fact['net_profit'], 10), 10);
        }
        if (bccomp($expected, '0', 10) <= 0 || bccomp($expected, (string) $order['win_amount'], 10) !== 0) return null;
        $name = \App\Models\LhcOrder::getRemark((int) ($order['type'] ?? 0));
        if ($name === '') return null;
        $types = [$name . '中奖'];
        if (in_array((int) $order['type'], [5, 6], true)) $types[] = '中奖';
        $logs = DB::table('balance_logs')->where('member_id', (string) $order['member_id'])
            ->where('related_id', $order['id'])->where('type', 1)->whereIn('change_type', $types)->where('amount', '>', 0)->get();
        $credited = Money::zero(); $last = null;
        foreach ($logs as $log) {
            $at = $this->time($log->created_at);
            if ($at === null || $at < $facts[0]['bet_at'] || $at > now('Asia/Shanghai')->format('Y-m-d H:i:s')) return null;
            if ($log->before_balance === null || $log->after_balance === null
                || bccomp(bcsub((string) $log->after_balance, (string) $log->before_balance, 10), (string) $log->amount, 10) !== 0) return null;
            $credited = bcadd($credited, (string) $log->amount, 10);
            $last = $last === null ? $at : max($last, $at);
        }
        return bccomp($credited, $expected, 10) === 0 ? $last : null;
    }

    private function money(string $source, array $row): void
    {
        $candidate = $this->time($row['created_at'] ?? null);
        if ($candidate === null) throw new \RuntimeException('资金记录缺少创建时间');
        $context = DB::table('reporting_contexts')->where('source', $source)->where('source_id', (string) $row['id'])->first();
        $at = $context->succeeded_at ?? null; $kind = null; $amount = null; $customerFee = Money::zero(); $configuredFee = null;
        $category = null; $identityRow = $row;
        if ($source === 'balance_logs') {
            $amount = Money::decimal($row['amount']); $type = (string) $row['change_type'];
            if (bccomp($amount, '0', 10) > 0 && in_array($type, ['充值', '三方充值'], true)) $kind = 'deposit';
            elseif (bccomp($amount, '0', 10) > 0 && in_array($type, ['注册赠送', '优惠活动', '即充即送', '充值返现', '老用户回归', '比比返', '笔笔返', '返水', '回水', '余额宝利息', '流水解冻'], true) && (int) ($row['type'] ?? 1) === 1) $kind = 'bonus';
            elseif ($type === '人工充值') $kind = isset($row['related_id']) && (string) $row['related_id'] === '0' ? 'bonus' : 'manual_credit';
            elseif ($type === '订单补偿') $kind = 'manual_credit';
            elseif ($type === '人工扣款') { $kind = 'manual_debit'; $amount = bcsub('0', $amount, 10); }
            elseif ($type === '分享佣金领取') {
                $claim = DB::table('share_claims')->where('credited_log_id', $row['id'])->where('status', 'approved')->first();
                if (!$claim) throw new \RuntimeException('分享领取缺少审核通过的原记录');
                $kind = 'share_commission';
            }
            $at = $candidate;
            if ($kind === 'deposit' && ($row['related_id'] ?? null)) {
                $origin = $type === '三方充值' ? 'payment_orders' : 'recharges';
                $originContext = DB::table('reporting_contexts')->where('source', $origin)->where('source_id', (string) $row['related_id'])->first();
                if ($originContext && !$context) $context = $originContext;
                if ($origin === 'payment_orders' && Schema::hasTable($origin)) {
                    $original = DB::table($origin)->where('id', $row['related_id'])->where('member_id', $row['member_id'])->whereIn('type', [1, 3])->where('status', 2)->first();
                    if ($original) $configuredFee = $original->fee ?? null;
                }
            }
        } elseif ($source === 'withdraws' && (int) $row['status'] === 1) {
            $kind = 'withdraw'; $rate = Money::decimal($row['exchange_rate']);
            if (bccomp($rate, '0', 10) <= 0) throw new \RuntimeException('提现历史汇率无效');
            $amount = isset($row['to_account']) && is_numeric($row['to_account']) ? bcmul((string) $row['to_account'], $rate, 10) : null;
            $customerFee = bcmul(Money::decimal($row['service_charge'] ?? 0), $rate, 10);
        } elseif ($source === 'payment_orders' && (int) $row['status'] === 2 && in_array((int) $row['type'], [2, 4], true)) {
            $kind = 'withdraw'; $amount = (int) $row['type'] === 2 ? Money::decimal($row['amount']) : null; $configuredFee = $row['fee'] ?? null;
        } elseif ($source === 'agent_withdrawals' && $row['status'] === 'paid') {
            $kind = 'agent_withdraw'; $amount = Money::decimal($row['actual_amount']); $customerFee = Money::decimal($row['fee'] ?? 0); $at = $this->time($row['paid_at'] ?? null);
        } elseif ($source === 'agent_balance_logs' && $row['type'] === 'commission' && $row['reference_type'] === 'agent_rebate_record') {
            $kind = 'agent_commission'; $amount = Money::decimal($row['amount']); $at = $candidate;
            $record = DB::table('agent_rebate_records')->where('id', $row['reference_id'])->first();
            if ($record) $category = $this->agentCategory((string) $record->platform, (int) $record->game_type);
        } elseif ($source === 'share_ledger') {
            $kind = 'share_accrual'; $amount = Money::decimal($row['amount']); $at = $row['bet_date'] . ' 00:00:00'; $category = $row['category'];
            $identityRow['report_user_id'] = $row['user_id'];
        } elseif ($source === 'agent_rebate_records') {
            $kind = 'agent_accrual'; $amount = Money::zero(); $at = $row['settlement_date'] . ' 00:00:00';
            foreach (['flow', 'profit'] as $part) $amount = bcadd($amount, (string) $row[$part . '_commission'], 10);
            $category = $this->agentCategory((string) $row['platform'], (int) $row['game_type']);
        }
        if (!$kind) {
            $existing = DB::table('reporting_events')->where('source', $source)->where('source_id', (string) $row['id'])->first();
            $verified = $existing ? DB::table('reporting_reconciliations')->where('event_id', $existing->id)->first() : null;
            if ($verified) {
                if (bccomp($verified->actual_amount, '0', 10) !== 0) throw new \RuntimeException('来源状态与已核实的收付款冲突，请重新核对');
                return; // Confirmed zero payout may still have an actual external fee; preserve its audit and event identity.
            }
            DB::table('reporting_events')->where('source', $source)->where('source_id', (string) $row['id'])->delete(); return;
        }
        $identity = ReportContext::identity($identityRow, $at ?? $candidate, $source);
        if ($context) $identity = ['user_id' => $context->user_id, 'agent_id' => $context->agent_id, 'inviter_id' => $context->inviter_id,
            'client_platform' => $context->client_platform, 'attribution_quality' => $context->attribution_quality];
        if (in_array($source, ['agent_withdrawals', 'agent_balance_logs', 'agent_rebate_records'], true)) $identity['agent_id'] = (int) $row['agent_id'];
        if ($identity['user_id'] === null && $identity['agent_id'] === null) throw new \RuntimeException('资金记录没有可识别的归属');
        $key = ['source' => $source, 'source_id' => (string) $row['id'], 'kind' => $kind];
        $values = $identity + ['category' => $category, 'currency' => $this->currency(), 'amount' => $amount,
            'customer_fee' => $customerFee, 'configured_fee' => $configuredFee, 'occurred_at' => $at, 'candidate_at' => $candidate];
        $this->writeProjection('reporting_events', $key, $values);
    }

    private function writeProjection(string $table, array $key, array $values): void
    {
        $old = DB::table($table)->where($key)->first();
        if (!$old) { DB::table($table)->insert($key + $values + ['created_at' => now(), 'updated_at' => now()]); return; }
        $decimals = ['bet_amount', 'payout_amount', 'net_profit', 'refund_amount', 'amount', 'customer_fee', 'configured_fee'];
        foreach ($values as $column => $value) {
            $previous = $old->$column;
            $same = $value === null || $previous === null ? $value === $previous
                : (in_array($column, $decimals, true) ? bccomp((string) $value, (string) $previous, 10) === 0 : (string) $value === (string) $previous);
            if (!$same) { DB::table($table)->where($key)->update($values + ['updated_at' => now()]); return; }
        }
    }

    private function agentCategory(string $platform, int $type): ?string
    {
        if ($platform === 'sport') return 'self:sport';
        if (in_array($platform, ['lhc', 'jisu', 'pc28'], true)) return 'self:lottery';
        return $type >= 1 && $type <= 7 && $platform !== 'legacy' ? 'third:' . $type : null;
    }
    public function currency(): string { return strtoupper((string) config('reporting.currency', 'CNY')); }
    public function time($value): ?string
    {
        if ($value === null || $value === '' || $value === 0 || $value === '0') return null;
        try { return Money::time($value)->setTimezone('Asia/Shanghai')->format('Y-m-d H:i:s'); } catch (\Throwable $e) { return null; }
    }
}