bot-28/app/Services/Sharing/BetFactSynchronizer.php

<?php

namespace App\Services\Sharing;

use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
use InvalidArgumentException;
use RuntimeException;

/** The only importer of current real-money orders into the new sharing facts. */
class BetFactSynchronizer
{
    private array $users = [];
    private const LOTTERY_NAMES = [1 => '新澳门六合彩', 2 => '香港六合彩', 3 => '澳门六合彩', 4 => '极速六合彩',
        5 => '加拿大28', 6 => '极速28', 7 => '极速赛车', 8 => '极速时时彩', 9 => '极速飞艇',
        10 => 'SG时时彩', 11 => 'SG飞艇', 12 => '极速快3', 13 => 'SG快3'];

    /** Read-only projection shared with reports; does not publish or alter sharing facts. */
    public function projectForReporting(string $source, object $order): array
    {
        if (!in_array($source, ['lhc_order', 'sport_order', 'sport_game_order', 'third_game_orders'], true)) throw new InvalidArgumentException('Unsupported reporting source');
        if ((int) ($order->is_faker ?? 0) === 1) return [];
        $this->users = [];
        return $source === 'lhc_order' ? $this->lotteryOrder($order, false)['facts'] : array_values(array_filter([$this->singleOrder($source, $order)]));
    }

    public function sync(?string $from = null, ?string $to = null, bool $write = false): array
    {
        foreach ([$from, $to] as $date) {
            if ($date !== null && (!preg_match('/^\d{4}-\d{2}-\d{2}$/D', $date)
                || \Carbon\Carbon::parse($date)->toDateString() !== $date)) {
                throw new InvalidArgumentException('Dates must use Y-m-d');
            }
        }
        if ($from !== null && $to !== null && $from > $to) throw new InvalidArgumentException('Invalid date range');
        $result = ['scanned' => 0, 'created' => 0, 'updated' => 0, 'unchanged' => 0, 'skipped' => 0,
            'errors' => [], 'changed_dates' => [], 'write' => $write];
        $this->users = [];
        foreach (['lhc_order', 'sport_game_order', 'sport_order', 'third_game_orders'] as $table) {
            if (!Schema::hasTable($table)) continue;
            $cursorColumn = $table === 'sport_order' ? 'update_time' : 'updated_at';
            $cursorEnabled = $write && $from === null && $to === null && Schema::hasColumn($table, $cursorColumn);
            $startedAt = now()->setTimezone('Asia/Shanghai')->format('Y-m-d H:i:s');
            $highestIdAtStart = $cursorEnabled ? (int) DB::table($table)->max('id') : 0;
            $errorCount = count($result['errors']);
            $query = DB::table($table);
            if ($cursorEnabled) {
                $lastRun = DB::table('share_system_state')->where('key', 'facts_cursor:' . $table)->value('value');
                if ($lastRun) {
                    $cursor = json_decode($lastRun, true, 512, JSON_THROW_ON_ERROR);
                    $cutoff = EffectiveBet::time($cursor['updated_at'])->subMinutes(5);
                    $cutoffValue = in_array($table, ['lhc_order', 'sport_order'], true)
                        ? $cutoff->timestamp : $cutoff->setTimezone('Asia/Shanghai')->format('Y-m-d H:i:s');
                    $query->where(function ($query) use ($cursorColumn, $cutoffValue, $cursor) {
                        $query->where($cursorColumn, '>=', $cutoffValue)->orWhereNull($cursorColumn)
                            ->orWhere('id', '>', (int) $cursor['last_id']);
                    });
                }
            }
            $query->orderBy('id')->chunkById(200, function ($orders) use ($table, $from, $to, $write, &$result) {
                foreach ($orders as $order) {
                    $result['scanned']++;
                    try {
                        $changes = $this->synchronizeOrder($table, (int) $order->id, $from, $to, $write);
                        foreach (['created', 'updated', 'unchanged', 'skipped'] as $key) $result[$key] += $changes[$key];
                        foreach ($changes['changed_dates'] as $date) $result['changed_dates'][$date] = $date;
                        $result['errors'] = array_merge($result['errors'], $changes['errors']);
                    } catch (\Throwable $e) {
                        $result['skipped']++;
                        $result['errors'][] = ['source' => $table, 'source_id' => (string) $order->id, 'reason' => $e->getMessage()];
                    }
                }
            });
            if ($cursorEnabled && count($result['errors']) === $errorCount) {
                DB::table('share_system_state')->updateOrInsert(['key' => 'facts_cursor:' . $table],
                    ['value' => json_encode(['updated_at' => $startedAt, 'last_id' => $highestIdAtStart]),
                        'updated_at' => $startedAt, 'created_at' => $startedAt]);
            }
        }
        $result['changed_dates'] = array_values($result['changed_dates']);
        sort($result['changed_dates']);
        return $result;
    }

    /** A source order and all of its child facts are one publication unit. */
    private function synchronizeOrder(string $table, int $id, ?string $from, ?string $to, bool $write): array
    {
        $operation = function () use ($table, $id, $from, $to, $write) {
            if ($write) {
                // Use the same first lock as settlement, relation binding and policy changes.
                $guard = DB::table('share_system_state')->where('key', 'relations_lock')->lockForUpdate()->first();
                if (!$guard) throw new RuntimeException('Sharing migration/relations lock is missing');
                $this->users = [];
            }
            // The chunk row may predate a lock wait or a concurrent source correction.
            $query = DB::table($table)->where('id', $id);
            $order = ($write ? $query->lockForUpdate() : $query)->first();
            if (!$order) throw new RuntimeException('Source order disappeared before synchronization');
            $result = ['created' => 0, 'updated' => 0, 'unchanged' => 0, 'skipped' => 0, 'changed_dates' => [], 'errors' => []];
            if ((int) ($order->is_faker ?? 0) === 1) { $result['skipped']++; return $result; }
            try {
                $facts = $table === 'lhc_order' ? $this->lotteryOrder($order, $write)['facts'] : [$this->singleOrder($table, $order)];
            } catch (\Throwable $e) {
                $isolated = $this->quarantine($table, $order, $e->getMessage(), $write);
                $result['updated'] = $isolated['updated'];
                $result['changed_dates'] = $isolated['changed_dates'];
                $result['skipped']++;
                $result['errors'][] = ['source' => $table, 'source_id' => (string) $id, 'reason' => $e->getMessage(),
                    'isolated_facts' => $isolated['updated']];
                return $result;
            }
            if ($facts === []) { $result['skipped']++; return $result; }
            foreach ($facts as $fact) {
                if ($fact === null || ($from !== null && $fact['bet_date'] < $from)
                    || ($to !== null && $fact['bet_date'] > $to)) { $result['skipped']++; continue; }
                $change = $this->persist($fact, $write);
                $result[$change['status']]++;
                $result['changed_dates'] = array_merge($result['changed_dates'], $change['dates']);
            }
            return $result;
        };
        return $write ? DB::transaction($operation, 3) : $operation();
    }

    /** Also used by the guarded historical PC28 rebate recalculation. */
    public function lotteryOrder(object $order, bool $lock = false): array
    {
        $type = (int) ($order->type ?? 0);
        if (!isset(self::LOTTERY_NAMES[$type])) throw new RuntimeException('Unknown lottery type');
        $betAt = $this->time($order->created_at ?? null);
        $base = ['member_id' => (string) $order->member_id, 'bet_at' => $betAt, 'date' => EffectiveBet::newYorkDate($order->created_at ?? null),
            'total_bet' => EffectiveBet::zero(), 'facts' => [], 'unsettled' => false];
        if (trim((string) ($order->ordernum ?? '')) === '') throw new RuntimeException('Missing lottery order number');
        if (DB::table('lhc_order')->where('ordernum', $order->ordernum)->count() !== 1) {
            throw new RuntimeException('Ambiguous lottery parent order number');
        }
        $itemQuery = DB::table('lhc_order_item')->where('ordernum', $order->ordernum)->orderBy('id');
        $items = ($lock ? $itemQuery->lockForUpdate() : $itemQuery)->get();
        if ($items->isEmpty()) throw new RuntimeException('Lottery order has no identifiable child bets');
        $bet = EffectiveBet::decimal($order->amount);
        if (bccomp($bet, '0', EffectiveBet::SCALE) <= 0) throw new RuntimeException('Lottery stake is not positive');
        $total = bcmul($bet, (string) $items->count(), EffectiveBet::SCALE);
        if (isset($order->total_amount) && bccomp((string) $order->total_amount, $total, EffectiveBet::SCALE) !== 0) {
            throw new RuntimeException('Lottery total does not match child stakes');
        }
        $base['total_bet'] = $total;
        $refund = $this->refundAmount($order, [self::LOTTERY_NAMES[$type] . '退款', '投注退分']);
        if (bccomp($refund, '0', EffectiveBet::SCALE) > 0 && bccomp($refund, $total, EffectiveBet::SCALE) !== 0
            && !in_array((int) $order->lottery_status, [3, 4], true)) {
            throw new RuntimeException('Partial or inconsistent lottery refund requires review');
        }
        $parentVoid = in_array((int) $order->lottery_status, [3, 4], true) || bccomp($refund, '0', EffectiveBet::SCALE) > 0;
        $user = $this->user((string) $order->member_id);
        if ($user === null) return $base;
        foreach ($items as $item) {
            if ((string) ($item->member_id ?? '') !== (string) $order->member_id) throw new RuntimeException('Lottery child member mismatch');
            $void = $parentVoid || in_array((int) $item->lottery_status, [3, 4], true);
            $settled = !$void && in_array((int) $order->lottery_status, [1, 2], true)
                && in_array((int) $item->lottery_status, [1, 2], true);
            $base['unsettled'] = $base['unsettled'] || (!$settled && !$void);
            $profit = $settled ? ((int) $item->lottery_status === 1 ? bcsub('0', $bet, EffectiveBet::SCALE)
                : bcsub(EffectiveBet::decimal($item->win_amount ?? null), $bet, EffectiveBet::SCALE)) : EffectiveBet::zero();
            $base['facts'][] = $this->fact('lhc_order_item', (string) $item->id, $user, 'self:lottery', $betAt,
                $settled || $void ? $this->time($item->updated_at ?? $order->updated_at ?? null) : null,
                $bet, $profit, $settled, $void, $this->nativeCurrency());
        }
        return $base;
    }

    private function singleOrder(string $table, object $order): ?array
    {
        $previous = null;
        if ($table === 'third_game_orders' && !empty($order->order_key) && Schema::hasTable('share_bet_facts')) {
            $previous = DB::table('share_bet_facts')->where('source', $table)->where('source_id', $order->order_key)->first();
        }
        $member = (string) ($table === 'sport_order' ? $order->user_id : ($order->member_id ?? ''));
        $user = $this->user($member, $table === 'third_game_orders' ? ($order->user_id ?? $previous->user_id ?? null) : null);
        if ($user === null) return null;
        $sourceId = (string) $order->id;
        $void = false;
        if ($table === 'third_game_orders') {
            $sourceId = (string) ($order->order_key ?? '');
            if (!preg_match('/^[a-f0-9]{64}$/D', $sourceId)) throw new RuntimeException('Third-party order key is invalid');
            $type = (int) $order->game_type;
            $settled = (int) $order->status === 1;
            $void = in_array((int) $order->status, [2, 3], true);
            if (($type < 1 || $type > 7) && !$previous) throw new RuntimeException('Unknown third-party category');
            $category = $type >= 1 && $type <= 7 ? 'third:' . $type : $previous->category;
            $bet = EffectiveBet::decimal($order->bet_amount ?? (!$settled ? ($previous->bet_amount ?? '0') : null));
            $profit = $settled ? EffectiveBet::decimal($order->settled_amount) : EffectiveBet::zero();
            $betAt = $this->time($order->bet_time ?? $previous->bet_at ?? null);
            $settledAt = $settled || $void ? $this->time($order->last_update_time ?? $order->updated_at ?? null) : null;
            $currency = strtoupper(trim((string) ($order->currency ?? $previous->currency ?? '')));
            if ($currency === '') throw new RuntimeException('Third-party currency is missing');
        } else {
            $category = 'self:sport';
            $bet = EffectiveBet::decimal($order->amount);
            $betAt = $this->time($table === 'sport_order' ? ($order->create_time ?? null) : ($order->created_at ?? null));
            if ($table === 'sport_order') {
                $void = (int) $order->status === 2 || (int) ($order->return_status ?? 0) === 2
                    || bccomp($this->refundAmount($order, ['体彩退款'], $member), '0', EffectiveBet::SCALE) > 0;
                $settled = !$void && (int) $order->status === 1 && (int) ($order->pay_status ?? 0) === 1
                    && (int) ($order->settlement_status ?? 0) === 1;
                $profit = $settled ? bcsub(bcadd(EffectiveBet::decimal($order->win_amount ?? '0'),
                    EffectiveBet::decimal($order->return_amount ?? '0'), EffectiveBet::SCALE), $bet, EffectiveBet::SCALE) : EffectiveBet::zero();
                $settledAt = $settled || $void ? $this->time($void
                    ? (($order->return_operation_time ?? null) ?: ($order->update_time ?? null)) : ($order->settlement_time ?? null)) : null;
            } else {
                $void = (int) $order->status === 3 || bccomp($this->refundAmount($order, ['足球游戏退款']), '0', EffectiveBet::SCALE) > 0;
                $settled = !$void && in_array((int) $order->status, [1, 2], true);
                $profit = !$settled ? EffectiveBet::zero() : ((int) $order->status === 1 ? bcsub('0', $bet, EffectiveBet::SCALE)
                    : bcsub(EffectiveBet::decimal($order->win_amount), $bet, EffectiveBet::SCALE));
                $settledAt = $settled || $void ? $this->time($order->updated_at ?? null) : null;
            }
            $currency = $this->nativeCurrency();
        }
        return $this->fact($table, $sourceId, $user, $category, $betAt, $settledAt, $bet, $profit, $settled, $void, $currency);
    }

    private function fact(string $source, string $id, object $user, string $category, string $betAt, ?string $settledAt,
        string $bet, string $profit, bool $settled, bool $void, string $currency): array
    {
        if (bccomp($bet, '0', EffectiveBet::SCALE) < 0) throw new RuntimeException('Negative stake');
        $fact = ['source' => $source, 'source_id' => $id, 'user_id' => (int) $user->id,
            'member_id' => (string) $user->member_id, 'category' => $category, 'bet_at' => $betAt,
            'bet_date' => substr($betAt, 0, 10), 'settled_at' => $settledAt, 'bet_amount' => $bet,
            'net_profit' => $profit, 'effective_amount' => EffectiveBet::amount($bet, $profit, $settled, $void),
            'settled' => $settled, 'voided' => $void, 'currency' => $currency];
        $fact['source_version'] = hash('sha256', json_encode($fact, JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR));
        return $fact;
    }

    private function user(string $member, $userId = null): ?object
    {
        $key = $userId !== null ? 'id:' . $userId : 'member:' . $member;
        if (array_key_exists($key, $this->users)) return $this->users[$key];
        $query = DB::table('users');
        $rows = ($userId !== null ? $query->where('id', $userId) : $query->where('member_id', $member))->limit(2)->get();
        if ($rows->count() !== 1) throw new RuntimeException('User mapping is missing or ambiguous');
        $user = $rows->first();
        if ($member !== '' && (string) $user->member_id !== $member) throw new RuntimeException('User identity mismatch');
        return $this->users[$key] = ((int) ($user->from ?? 0) === 2 ? null : $user);
    }

    private function refundAmount(object $order, array $types, ?string $member = null): string
    {
        if (!Schema::hasTable('balance_logs')) return EffectiveBet::zero();
        return EffectiveBet::decimal(DB::table('balance_logs')->where('member_id', $member ?? $order->member_id)
            ->where('related_id', $order->id)->whereIn('change_type', $types)->where('amount', '>', 0)->sum('amount'));
    }

    private function time($value): string
    {
        return EffectiveBet::time($value)->setTimezone('Asia/Shanghai')->format('Y-m-d H:i:s');
    }

    private function nativeCurrency(): string
    {
        return strtoupper(trim((string) config('third_game.currency', 'CNY')));
    }

    private function persist(array $fact, bool $write): array
    {
        if ($write && DB::transactionLevel() === 0) {
            throw new RuntimeException('Fact publication requires its source order transaction');
        }
        $query = DB::table('share_bet_facts')->where('source', $fact['source'])->where('source_id', $fact['source_id']);
        $existing = ($write ? $query->lockForUpdate() : $query)->first();
        if ($existing && hash_equals((string) $existing->source_version, $fact['source_version'])) {
            return ['status' => 'unchanged', 'dates' => []];
        }
        $dates = array_values(array_unique(array_filter([$existing->bet_date ?? null, $fact['bet_date']])));
        if ($write) {
            $now = now()->setTimezone('Asia/Shanghai')->format('Y-m-d H:i:s');
            if ($existing) DB::table('share_bet_facts')->where('id', $existing->id)->update($fact + ['updated_at' => $now]);
            else DB::table('share_bet_facts')->insert($fact + ['created_at' => $now, 'updated_at' => $now]);
            $this->markDirty($dates, $now);
        }
        return ['status' => $existing ? 'updated' : 'created', 'dates' => $dates];
    }

    /** Only quarantine previously published facts identifiable from the current source row. */
    private function quarantine(string $table, object $order, string $reason, bool $write): array
    {
        if ($table === 'lhc_order') {
            if (trim((string) ($order->ordernum ?? '')) === '' || !Schema::hasTable('lhc_order_item')) {
                return ['updated' => 0, 'changed_dates' => []];
            }
            $items = DB::table('lhc_order_item')->where('ordernum', $order->ordernum)->orderBy('id');
            $ids = ($write ? $items->lockForUpdate() : $items)->pluck('id')->map(fn ($id) => (string) $id)->all();
            $source = 'lhc_order_item';
        } else {
            $ids = [(string) ($table === 'third_game_orders' ? ($order->order_key ?? '') : $order->id)];
            $source = $table;
        }
        $ids = array_values(array_filter($ids, fn ($id) => $id !== ''));
        if ($ids === []) return ['updated' => 0, 'changed_dates' => []];
        $query = DB::table('share_bet_facts')->where('source', $source)->whereIn('source_id', $ids)->orderBy('id');
        $facts = ($write ? $query->lockForUpdate() : $query)->get();
        $now = now()->setTimezone('Asia/Shanghai')->format('Y-m-d H:i:s');
        $updated = 0; $dates = [];
        foreach ($facts as $fact) {
            $version = hash('sha256', json_encode(['invalid_source' => $source, 'source_id' => $fact->source_id,
                'reason' => $reason, 'order' => $order], JSON_UNESCAPED_UNICODE | JSON_INVALID_UTF8_SUBSTITUTE | JSON_THROW_ON_ERROR));
            if ((string) $fact->source_version === $version && !(bool) $fact->settled
                && bccomp((string) $fact->effective_amount, '0', EffectiveBet::SCALE) === 0) continue;
            $updated++; $dates[] = $fact->bet_date;
            if ($write) DB::table('share_bet_facts')->where('id', $fact->id)->update([
                'settled' => false, 'effective_amount' => EffectiveBet::zero(), 'source_version' => $version, 'updated_at' => $now]);
        }
        $dates = array_values(array_unique($dates));
        if ($write && $dates !== []) $this->markDirty($dates, $now);
        return ['updated' => $updated, 'changed_dates' => $dates];
    }

    private function markDirty(array $dates, string $now): void
    {
        DB::table('share_system_state')->insertOrIgnore(['key' => 'facts_dirty_dates', 'value' => '[]', 'created_at' => $now, 'updated_at' => $now]);
        $state = DB::table('share_system_state')->where('key', 'facts_dirty_dates')->lockForUpdate()->first();
        $dirty = json_decode($state->value, true, 512, JSON_THROW_ON_ERROR);
        $dirty = array_values(array_unique(array_merge($dirty, $dates)));
        sort($dirty);
        DB::table('share_system_state')->where('key', 'facts_dirty_dates')->update(['value' => json_encode($dirty), 'updated_at' => $now]);
    }
}