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); } }