|
|
@@ -13,49 +13,101 @@ 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;
|
|
|
- $query = DB::table($source); $hasUpdate = Schema::hasColumn($source, 'updated_at');
|
|
|
$column = $source === 'sport_order' ? 'update_time' : 'updated_at';
|
|
|
- if (!$full && $cursor && ($hasUpdate || $source === 'sport_order')) {
|
|
|
+ $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');
|
|
|
- $query->where(function ($q) use ($column, $value, $cursor) { $q->where($column, '>=', $value)->orWhereNull($column)->orWhere('id', '>', $cursor['last_id']); });
|
|
|
- } elseif (!$full && $cursor && in_array($source, ['share_ledger', 'agent_balance_logs'], true)) $query->where('id', '>', $cursor['last_id']);
|
|
|
- $max = (int) DB::table($source)->max('id'); $errors = 0;
|
|
|
- $query->where('id', '<=', $max)->orderBy('id')->chunkById(200, function ($rows) use ($source, &$result, &$errors) {
|
|
|
- foreach ($rows as $row) {
|
|
|
- ++$result['scanned'];
|
|
|
- try {
|
|
|
- if ($source === 'lhc_order_item') {
|
|
|
- $parents = DB::table('lhc_order')->where('ordernum', $row->ordernum)->limit(2)->get();
|
|
|
- if ($parents->count() !== 1) throw new \RuntimeException('六合彩子单对应的父单不明确');
|
|
|
- $this->publish('lhc_order', (int) $parents[0]->id);
|
|
|
- DB::table('reporting_errors')->where('source', $source)->where('source_id', (string) $row->id)->delete();
|
|
|
- } else $this->publish($source, (int) $row->id);
|
|
|
- ++$result['published'];
|
|
|
- }
|
|
|
- catch (\Throwable $e) {
|
|
|
- ++$errors; ++$result['errors'];
|
|
|
- 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 ($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);
|
|
|
}
|
|
|
- });
|
|
|
- if (!$errors) DB::table('reporting_sync_state')->updateOrInsert(['source' => $source], ['state' => json_encode(['checked_at' => $start, 'last_id' => $max]), 'updated_at' => $start]);
|
|
|
+
|
|
|
+ // 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('不支持的数据来源');
|
|
|
@@ -120,9 +172,7 @@ class ReportSync
|
|
|
'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'];
|
|
|
- $old = DB::table('reporting_orders')->where($key)->first();
|
|
|
- if (!$old) $values['created_at'] = now(); $values['updated_at'] = now();
|
|
|
- DB::table('reporting_orders')->updateOrInsert($key, $values);
|
|
|
+ $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();
|
|
|
}
|
|
|
@@ -219,11 +269,22 @@ class ReportSync
|
|
|
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];
|
|
|
- $old = DB::table('reporting_events')->where($key)->first();
|
|
|
$values = $identity + ['category' => $category, 'currency' => $this->currency(), 'amount' => $amount,
|
|
|
- 'customer_fee' => $customerFee, 'configured_fee' => $configuredFee, 'occurred_at' => $at, 'candidate_at' => $candidate, 'updated_at' => now()];
|
|
|
- if (!$old) $values['created_at'] = now();
|
|
|
- DB::table('reporting_events')->updateOrInsert($key, $values);
|
|
|
+ '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
|