1.ws优化bet.win订阅

app/common/service/GameBetSettleService.php

@@ -32,6 +32,9 @@ final class GameBetSettleService
     /** 每期结算推送去重（避免事务重试 / recover 重复推 user.streak、wallet.changed） */
     private const SETTLE_NOTIFY_DEDUP_PREFIX = 'dfw:v1:settle:notify:';
 
+    /** 每期每用户 bet.win 去重（与 streak/wallet 分离，避免整期 dedup 吞掉中奖推送） */
+    private const BET_WIN_NOTIFY_DEDUP_PREFIX = 'dfw:v1:ws:betwin:';
+
     /**
      * 对指定期次按开奖号码结算所有「待开奖」注单；同一注单幂等（仅 status=1 会更新）。
      *
@@ -310,26 +313,88 @@ final class GameBetSettleService
      *
      * @param list<array<string, mixed>> $betWins
      */
-    public static function publishBetWinsAfterCommit(array $betWins): void
+    public static function publishBetWinsAfterCommit(array $betWins, int $periodId = 0): void
     {
         $now = time();
         foreach ($betWins as $payload) {
-            if (!is_array($payload) || empty($payload['user_id'])) {
+            if (!is_array($payload)) {
                 continue;
             }
-            $userId = filter_var($payload['user_id'], FILTER_VALIDATE_INT);
+            $userId = filter_var($payload['user_id'] ?? 0, FILTER_VALIDATE_INT);
             if ($userId === false || $userId <= 0) {
                 continue;
             }
+            if ($periodId > 0 && !self::markBetWinNotifyOnce($periodId, $userId)) {
+                continue;
+            }
             $isJackpot = !empty($payload['is_jackpot']);
-            $data = array_merge($payload, [
+            $data = GameWebSocketPayloadHelper::mergeUserStreakInto(array_merge($payload, [
                 'is_jackpot' => $isJackpot,
+                'is_win' => true,
                 'server_time' => $now,
-            ]);
+            ]), $userId);
             GameWebSocketEventBus::publish(self::TOPIC_BET_WIN, $data);
         }
     }
 
+    /**
+     * 从库内已结算中奖注单重建 bet.win 载荷（结算内存聚合缺失或推送被 dedup 拦截时的补偿）。
+     *
+     * @return list<array<string, mixed>>
+     */
+    public static function buildBetWinPayloadsFromSettledOrders(int $periodId, int $resultNumber): array
+    {
+        if ($periodId <= 0 || $resultNumber < 1) {
+            return [];
+        }
+        $rows = Db::name('bet_order')
+            ->where('period_id', $periodId)
+            ->whereIn('status', [self::PLAY_STATUS_SETTLED, self::PLAY_STATUS_PENDING_REVIEW])
+            ->order('id', 'asc')
+            ->select()
+            ->toArray();
+
+        /** @var array<int, array<string, mixed>> $winByUser */
+        $winByUser = [];
+        foreach ($rows as $bet) {
+            if (!is_array($bet)) {
+                continue;
+            }
+            $win = bcadd((string) ($bet['win_amount'] ?? '0'), '0', 2);
+            if (bccomp($win, '0', 2) <= 0) {
+                continue;
+            }
+            $userId = filter_var($bet['user_id'] ?? 0, FILTER_VALIDATE_INT);
+            if ($userId === false || $userId <= 0) {
+                continue;
+            }
+            if (!isset($winByUser[$userId])) {
+                $coin = Db::name('user')->where('id', $userId)->value('coin');
+                $winByUser[$userId] = [
+                    'user_id' => $userId,
+                    'period_id' => $periodId,
+                    'period_no' => (string) ($bet['period_no'] ?? ''),
+                    'result_number' => $resultNumber,
+                    'total_win' => '0.00',
+                    'balance_after' => (string) ($coin ?? '0'),
+                    'is_jackpot' => false,
+                    'bets' => [],
+                ];
+            }
+            if (StreakWinReward::isJackpotForStreakAtBet((int) ($bet['streak_at_bet'] ?? 0))) {
+                $winByUser[$userId]['is_jackpot'] = true;
+            }
+            $winByUser[$userId]['total_win'] = bcadd((string) $winByUser[$userId]['total_win'], $win, 2);
+            $betId = filter_var($bet['id'] ?? 0, FILTER_VALIDATE_INT);
+            $winByUser[$userId]['bets'][] = [
+                'bet_id' => $betId === false ? 0 : $betId,
+                'win_amount' => $win,
+            ];
+        }
+
+        return array_values($winByUser);
+    }
+
     /**
      * 大奖档命中时额外推送公共频道 jackpot.hit（与 bet.win 同一结算时刻，先后发出）。
      *
@@ -364,47 +429,73 @@ final class GameBetSettleService
     {
         $settledCount = filter_var($settleOut['settled_order_count'] ?? 0, FILTER_VALIDATE_INT);
         $betWins = is_array($settleOut['bet_wins'] ?? null) ? $settleOut['bet_wins'] : [];
-        $hasWins = $betWins !== [];
         $hasStreak = is_array($settleOut['user_streak_events'] ?? null) && $settleOut['user_streak_events'] !== [];
         $hasWallet = is_array($settleOut['wallet_events'] ?? null) && $settleOut['wallet_events'] !== [];
         if ($settledCount === false || $settledCount <= 0) {
-            if (!$hasWins && !$hasStreak && !$hasWallet) {
-                return;
+            if ($betWins === [] && !$hasStreak && !$hasWallet) {
+                if ($periodId <= 0 || $resultNumber < 1) {
+                    return;
+                }
+                $betWins = self::buildBetWinPayloadsFromSettledOrders($periodId, $resultNumber);
+                if ($betWins === []) {
+                    return;
+                }
             }
         }
-        if (!self::markSettlementNotifyOnce($periodId)) {
-            return;
-        }
 
-        $streakEvents = is_array($settleOut['user_streak_events'] ?? null) ? $settleOut['user_streak_events'] : [];
-        foreach ($streakEvents as $row) {
-            if (!is_array($row)) {
-                continue;
+        if (($settledCount !== false && $settledCount > 0) || $hasStreak || $hasWallet) {
+            if (self::markSettlementNotifyOnce($periodId)) {
+                $streakEvents = is_array($settleOut['user_streak_events'] ?? null) ? $settleOut['user_streak_events'] : [];
+                foreach ($streakEvents as $row) {
+                    if (!is_array($row)) {
+                        continue;
+                    }
+                    $userId = filter_var($row['user_id'] ?? 0, FILTER_VALIDATE_INT);
+                    if ($userId === false || $userId <= 0) {
+                        continue;
+                    }
+                    $streak = filter_var($row['current_streak'] ?? 0, FILTER_VALIDATE_INT);
+                    $extra = is_array($row['extra'] ?? null) ? $row['extra'] : [];
+                    GameWebSocketPayloadHelper::publishUserStreak($userId, $streak === false ? 0 : $streak, $extra);
+                }
+
+                $walletEvents = is_array($settleOut['wallet_events'] ?? null) ? $settleOut['wallet_events'] : [];
+                foreach ($walletEvents as $payload) {
+                    if (!is_array($payload)) {
+                        continue;
+                    }
+                    $userId = filter_var($payload['user_id'] ?? 0, FILTER_VALIDATE_INT);
+                    if ($userId === false || $userId <= 0) {
+                        continue;
+                    }
+                    GameWebSocketEventBus::publish('wallet.changed', GameWebSocketPayloadHelper::mergeUserStreakInto($payload, $userId));
+                }
+
+                $jackpotHits = is_array($settleOut['jackpot_hits'] ?? null) ? $settleOut['jackpot_hits'] : [];
+                self::publishJackpotHitsAfterCommit($jackpotHits, $periodId, $periodNo, $resultNumber);
             }
-            $userId = filter_var($row['user_id'] ?? 0, FILTER_VALIDATE_INT);
-            if ($userId === false || $userId <= 0) {
-                continue;
-            }
-            $streak = filter_var($row['current_streak'] ?? 0, FILTER_VALIDATE_INT);
-            $extra = is_array($row['extra'] ?? null) ? $row['extra'] : [];
-            GameWebSocketPayloadHelper::publishUserStreak($userId, $streak === false ? 0 : $streak, $extra);
         }
 
-        $walletEvents = is_array($settleOut['wallet_events'] ?? null) ? $settleOut['wallet_events'] : [];
-        foreach ($walletEvents as $payload) {
-            if (!is_array($payload) || empty($payload['user_id'])) {
-                continue;
-            }
-            $userId = filter_var($payload['user_id'], FILTER_VALIDATE_INT);
-            if ($userId === false || $userId <= 0) {
-                continue;
-            }
-            GameWebSocketEventBus::publish('wallet.changed', GameWebSocketPayloadHelper::mergeUserStreakInto($payload, $userId));
+        $effectiveBetWins = $betWins;
+        if ($effectiveBetWins === [] && $periodId > 0 && $resultNumber > 0) {
+            $effectiveBetWins = self::buildBetWinPayloadsFromSettledOrders($periodId, $resultNumber);
         }
+        self::publishBetWinsAfterCommit($effectiveBetWins, $periodId);
+    }
 
-        $jackpotHits = is_array($settleOut['jackpot_hits'] ?? null) ? $settleOut['jackpot_hits'] : [];
-        self::publishBetWinsAfterCommit($betWins);
-        self::publishJackpotHitsAfterCommit($jackpotHits, $periodId, $periodNo, $resultNumber);
+    private static function markBetWinNotifyOnce(int $periodId, int $userId): bool
+    {
+        if ($periodId <= 0 || $userId <= 0) {
+            return true;
+        }
+        $key = self::BET_WIN_NOTIFY_DEDUP_PREFIX . $periodId . ':' . $userId;
+        try {
+            $ok = Redis::set($key, '1', ['nx', 'ex' => 86400]);
+
+            return $ok === true || $ok === 'OK';
+        } catch (Throwable) {
+            return true;
+        }
     }
 
     private static function markSettlementNotifyOnce(int $periodId): bool
@@ -524,6 +615,25 @@ final class GameBetSettleService
         } catch (\Throwable) {
         }
 
+        $periodId = filter_var($row['period_id'] ?? 0, FILTER_VALIDATE_INT);
+        $periodNo = is_string($row['period_no'] ?? null) ? (string) $row['period_no'] : '';
+        $resultNumber = filter_var(
+            Db::name('game_record')->where('id', $periodId)->value('result_number'),
+            FILTER_VALIDATE_INT
+        );
+        if ($periodId !== false && $periodId > 0 && $resultNumber !== false && $resultNumber > 0 && bccomp($winAmount, '0', 2) > 0) {
+            self::publishBetWinsAfterCommit([[
+                'user_id' => $userId,
+                'period_id' => $periodId,
+                'period_no' => $periodNo,
+                'result_number' => $resultNumber,
+                'total_win' => $winAmount,
+                'balance_after' => is_string($balanceAfter ?? null) ? $balanceAfter : (string) (Db::name('user')->where('id', $userId)->value('coin') ?? '0'),
+                'is_jackpot' => StreakWinReward::isJackpotForStreakAtBet((int) ($row['streak_at_bet'] ?? 0)),
+                'bets' => [['bet_id' => $playRecordId, 'win_amount' => $winAmount]],
+            ]], $periodId);
+        }
+
         $out = ['ok' => true, 'msg' => __('Approved')];
         if (is_string($balanceAfter)) {
             $out['balance_after'] = $balanceAfter;