bot-28/app/Console/Commands/SyncWagering.php

<?php

namespace App\Console\Commands;

use App\Models\Wallet;
use App\Services\PromotionFacts;
use App\Services\WageringService;
use App\Support\SyncCommandLock;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Facades\Schema;

/** 刷新钱包缓存的“当前打码量”。只重算最近有投注变动的会员；--all 用于上线后首次全量计算。 */
class SyncWagering extends Command
{
    private const SINCE_KEY = 'wagering:sync:since';
    /** 上次刷新失败的会员，下次运行重试；时间窗口照常前进，避免一个坏钱包让扫描范围无限扩大 */
    private const FAILED_KEY = 'wagering:sync:failed';

    protected $signature = 'wagering:sync {--all : 重算全部钱包} {--member= : 只重算指定会员ID}';
    protected $description = '按有效投注刷新会员当前打码量/剩余打码量';

    public function handle(SyncCommandLock $lock): int
    {
        $exit = $lock->run('wagering:sync', fn () => $this->synchronize());
        if ($exit === null) $this->info('打码量同步已在运行，跳过本次执行');
        return $exit ?? self::SUCCESS;
    }

    private function synchronize(): int
    {
        if (!WageringService::enabled()) return self::SUCCESS;
        $startedAt = PromotionFacts::now();
        $errors = 0; $count = 0; $failed = [];
        $refresh = function ($wallets) use (&$errors, &$count, &$failed) {
            $facts = new PromotionFacts();
            foreach ($wallets as $wallet) {
                try { WageringService::refresh($wallet, $facts); ++$count; }
                catch (\Throwable $e) {
                    ++$errors; $failed[] = (string)$wallet->member_id;
                    Log::error('wagering_sync_failed', ['member_id' => $wallet->member_id, 'exception' => $e]);
                }
            }
        };

        if ($this->option('member')) {
            $refresh(Wallet::where('member_id', (string)$this->option('member'))->get());
        } elseif ($this->option('all')) {
            Wallet::query()->orderBy('id')->chunkById(200, $refresh);
        } else {
            // 与上次运行重叠两分钟，覆盖注单延迟结算和同步时钟误差。
            $since = Cache::get(self::SINCE_KEY) ?? date('Y-m-d H:i:s', strtotime($startedAt) - 600);
            $since = date('Y-m-d H:i:s', strtotime($since) - 120);
            $members = array_values(array_unique(array_merge($this->changedMembers($since), (array)Cache::get(self::FAILED_KEY, []))));
            foreach (array_chunk($members, 200) as $chunk) {
                $refresh(Wallet::whereIn('member_id', $chunk)->get());
            }
            Cache::forever(self::SINCE_KEY, $startedAt);
            Cache::forever(self::FAILED_KEY, array_values(array_unique($failed)));
        }
        $this->info('打码量同步完成，刷新：' . $count . '，失败：' . $errors);
        return $errors ? self::FAILURE : self::SUCCESS;
    }

    /** @return string[] */
    private function changedMembers(string $since): array
    {
        $members = [];
        foreach (['share_bet_facts', 'bets', 'jisu_game_order'] as $table) {
            if (!Schema::hasTable($table) || !Schema::hasColumn($table, 'updated_at')) continue;
            foreach (DB::table($table)->where('updated_at', '>=', $since)->whereNotNull('member_id')->distinct()->pluck('member_id') as $memberId) {
                $members[(string)$memberId] = true;
            }
        }
        return array_keys($members);
    }
}