2026-05-25 21:19:09 +08:00

679 lines
20 KiB
Go

package highwin
import (
"context"
"fmt"
"math/rand"
"net/http"
"strconv"
"strings"
"sync"
"time"
"chatapp3-golang/internal/common"
"chatapp3-golang/internal/service/regionimgroup"
"github.com/redis/go-redis/v9"
"gorm.io/gorm"
)
const (
SimulationKindLuckyGift = "LUCKY_GIFT"
SimulationKindGameWin = "GAME_WIN"
adminTimeLayout = "2006-01-02 15:04:05"
)
type BroadcastSimulationRequest struct {
SysOrigin string `json:"sysOrigin"`
RegionCode string `json:"regionCode"`
Kind string `json:"kind"`
SenderUserID string `json:"senderUserId"`
RoomID string `json:"roomId"`
Multiple string `json:"multiple"`
WinAmount string `json:"winAmount"`
AcceptUserID string `json:"acceptUserId"`
GiftID string `json:"giftId"`
GiftCandy string `json:"giftCandy"`
GiftQuantity string `json:"giftQuantity"`
GameID string `json:"gameId"`
GameRoundID string `json:"gameRoundId"`
GameURL string `json:"gameUrl"`
BetAmount string `json:"betAmount"`
ProviderOrder string `json:"providerOrderId"`
}
type BroadcastSimulationResponse struct {
SysOrigin string `json:"sysOrigin"`
RegionCode string `json:"regionCode"`
GroupID string `json:"groupId"`
Type string `json:"type"`
Data map[string]any `json:"data"`
}
type BroadcastSimulationWorker struct {
db *gorm.DB
redis *redis.Client
broadcaster RegionBroadcaster
mu sync.Mutex
cancel context.CancelFunc
config ContinuousSimulationRequest
status ContinuousSimulationStatus
}
type ContinuousSimulationRequest struct {
SysOrigin string `json:"sysOrigin"`
RegionCodes []string `json:"regionCodes"`
EnableLuckyGift bool `json:"enableLuckyGift"`
EnableGameWin bool `json:"enableGameWin"`
MinDelaySeconds int `json:"minDelaySeconds"`
MaxDelaySeconds int `json:"maxDelaySeconds"`
BetAmounts []int64 `json:"betAmounts"`
Multiples []float64 `json:"multiples"`
GameWinAmounts []int64 `json:"gameWinAmounts"`
LuckyGiftWinAmounts []int64 `json:"luckyGiftWinAmounts"`
GiftCandyOptions []int64 `json:"giftCandyOptions"`
GiftQuantityOptions []int64 `json:"giftQuantityOptions"`
}
type ContinuousSimulationStatus struct {
Running bool `json:"running"`
SysOrigin string `json:"sysOrigin"`
RegionCodes []string `json:"regionCodes"`
StartedAt string `json:"startedAt,omitempty"`
LastRunAt string `json:"lastRunAt,omitempty"`
LastError string `json:"lastError,omitempty"`
NextDelaySec int `json:"nextDelaySec"`
LuckyGiftCount int64 `json:"luckyGiftCount"`
GameWinCount int64 `json:"gameWinCount"`
TotalCount int64 `json:"totalCount"`
Config any `json:"config,omitempty"`
UpdateTime time.Time `json:"-"`
}
type continuousCandidateUser struct {
ID int64
}
type continuousCandidateGame struct {
GameID string
Cover string
}
type continuousCandidateRoom struct {
RoomID int64
}
func NewBroadcastSimulationWorker(db *gorm.DB, redisClient *redis.Client, broadcaster RegionBroadcaster) *BroadcastSimulationWorker {
return &BroadcastSimulationWorker{
db: db,
redis: redisClient,
broadcaster: broadcaster,
}
}
func (w *BroadcastSimulationWorker) Start(ctx context.Context, req ContinuousSimulationRequest) (*ContinuousSimulationStatus, error) {
req = normalizeContinuousRequest(req)
if err := validateContinuousRequest(req); err != nil {
return nil, err
}
if w == nil || w.db == nil || w.redis == nil || w.broadcaster == nil {
return nil, common.NewAppError(http.StatusInternalServerError, "simulation_worker_not_configured", "simulation worker is not configured")
}
w.mu.Lock()
if w.cancel != nil {
w.cancel()
}
workerCtx, cancel := context.WithCancel(context.Background())
w.cancel = cancel
w.config = req
w.status = ContinuousSimulationStatus{
Running: true,
SysOrigin: req.SysOrigin,
RegionCodes: append([]string(nil), req.RegionCodes...),
StartedAt: time.Now().Format(adminTimeLayout),
Config: req,
}
status := w.status
w.mu.Unlock()
go w.run(workerCtx, req)
return &status, nil
}
func (w *BroadcastSimulationWorker) Stop() *ContinuousSimulationStatus {
if w == nil {
return &ContinuousSimulationStatus{}
}
w.mu.Lock()
defer w.mu.Unlock()
if w.cancel != nil {
w.cancel()
w.cancel = nil
}
w.status.Running = false
w.status.NextDelaySec = 0
w.status.UpdateTime = time.Now()
status := w.status
status.Config = w.config
return &status
}
func (w *BroadcastSimulationWorker) Status() *ContinuousSimulationStatus {
if w == nil {
return &ContinuousSimulationStatus{}
}
w.mu.Lock()
defer w.mu.Unlock()
status := w.status
status.Config = w.config
return &status
}
func (w *BroadcastSimulationWorker) run(ctx context.Context, req ContinuousSimulationRequest) {
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
for {
delay := randomDelaySeconds(rng, req.MinDelaySeconds, req.MaxDelaySeconds)
w.updateStatus(func(status *ContinuousSimulationStatus) {
status.NextDelaySec = delay
})
timer := time.NewTimer(time.Duration(delay) * time.Second)
select {
case <-ctx.Done():
timer.Stop()
w.updateStatus(func(status *ContinuousSimulationStatus) {
status.Running = false
status.NextDelaySec = 0
})
return
case <-timer.C:
}
w.runOnce(ctx, req, rng)
}
}
func (w *BroadcastSimulationWorker) runOnce(ctx context.Context, req ContinuousSimulationRequest, rng *rand.Rand) {
regionCode := req.RegionCodes[rng.Intn(len(req.RegionCodes))]
userID, err := w.randomOfflineUser(ctx, req.SysOrigin, regionCode, rng)
if err != nil {
w.setLastError(err)
return
}
roomID, err := w.randomActiveRoom(ctx, req.SysOrigin, rng)
if err != nil {
w.setLastError(err)
return
}
if req.EnableLuckyGift {
winAmount := pickInt64(rng, req.LuckyGiftWinAmounts)
if winAmount <= 0 {
winAmount = pickInt64(rng, req.GameWinAmounts)
}
err := w.sendContinuous(ctx, BroadcastSimulationRequest{
SysOrigin: req.SysOrigin,
RegionCode: regionCode,
Kind: SimulationKindLuckyGift,
SenderUserID: strconv.FormatInt(userID, 10),
RoomID: strconv.FormatInt(roomID, 10),
Multiple: formatFloat(pickFloat64(rng, req.Multiples)),
WinAmount: strconv.FormatInt(winAmount, 10),
GiftCandy: strconv.FormatInt(pickInt64(rng, req.GiftCandyOptions), 10),
GiftQuantity: strconv.FormatInt(pickInt64(rng, req.GiftQuantityOptions), 10),
ProviderOrder: fmt.Sprintf("worker-lucky-%d", time.Now().UnixNano()),
})
if err != nil {
w.setLastError(err)
} else {
w.incrementCount(true)
}
}
if req.EnableGameWin {
game, err := w.randomGame(ctx, req.SysOrigin)
if err != nil {
w.setLastError(err)
return
}
err = w.sendContinuous(ctx, BroadcastSimulationRequest{
SysOrigin: req.SysOrigin,
RegionCode: regionCode,
Kind: SimulationKindGameWin,
SenderUserID: strconv.FormatInt(userID, 10),
RoomID: strconv.FormatInt(roomID, 10),
Multiple: formatFloat(pickFloat64(rng, req.Multiples)),
WinAmount: strconv.FormatInt(pickInt64(rng, req.GameWinAmounts), 10),
GameID: game.GameID,
GameURL: game.Cover,
GameRoundID: fmt.Sprintf("worker-game-%d", time.Now().UnixNano()),
BetAmount: strconv.FormatInt(pickInt64(rng, req.BetAmounts), 10),
})
if err != nil {
w.setLastError(err)
} else {
w.incrementCount(false)
}
}
}
func SendSimulationBroadcast(
ctx context.Context,
broadcaster RegionBroadcaster,
req BroadcastSimulationRequest,
) (*BroadcastSimulationResponse, error) {
if broadcaster == nil {
return nil, common.NewAppError(http.StatusInternalServerError, "region_broadcaster_missing", "region broadcaster is missing")
}
messageType, err := simulationMessageType(req.Kind)
if err != nil {
return nil, err
}
userID, err := parseRequiredInt64(req.SenderUserID, "senderUserId")
if err != nil {
return nil, err
}
roomID := strings.TrimSpace(req.RoomID)
if roomID == "" {
return nil, common.NewAppError(http.StatusBadRequest, "missing_room_id", "roomId is required")
}
amount, err := parseRequiredInt64(req.WinAmount, "winAmount")
if err != nil {
return nil, err
}
multiple, err := parseRequiredFloat64(req.Multiple, "multiple")
if err != nil {
return nil, err
}
if !ShouldBroadcast(multiple, amount) {
return nil, common.NewAppError(http.StatusBadRequest, "broadcast_threshold_not_met", "multiple must be >= 10 and winAmount must be > 0")
}
data := simulationData(messageType, req, amount)
event := BroadcastEvent{
SysOrigin: req.SysOrigin,
RegionCode: req.RegionCode,
Type: messageType,
UserID: userID,
RoomID: roomID,
Multiple: multiple,
Amount: amount,
Data: data,
}
if err := SendRegionBroadcast(ctx, broadcaster, event); err != nil {
return nil, err
}
previewData := cloneData(data)
setRequired(previewData, "userId", strconv.FormatInt(userID, 10))
setRequired(previewData, "sendUserId", strconv.FormatInt(userID, 10))
setRequired(previewData, "senderUserId", strconv.FormatInt(userID, 10))
setRequired(previewData, "roomId", roomID)
setRequired(previewData, "multiple", normalizeMultiple(multiple))
setRequired(previewData, "winMultiple", normalizeMultiple(multiple))
setRequired(previewData, "winAmount", amount)
setRequired(previewData, "awardAmount", amount)
setRequired(previewData, "msg", highWinMessage(previewData, messageType, strconv.FormatInt(userID, 10), multiple, amount))
return &BroadcastSimulationResponse{
SysOrigin: strings.ToUpper(strings.TrimSpace(req.SysOrigin)),
RegionCode: strings.TrimSpace(req.RegionCode),
Type: messageType,
Data: previewData,
}, nil
}
func (w *BroadcastSimulationWorker) sendContinuous(ctx context.Context, req BroadcastSimulationRequest) error {
_, err := SendSimulationBroadcast(ctx, w.broadcaster, req)
return err
}
func (w *BroadcastSimulationWorker) randomOfflineUser(ctx context.Context, sysOrigin string, regionCode string, rng *rand.Rand) (int64, error) {
onlineIDs := w.onlineUserIDs(ctx, sysOrigin)
query := w.db.WithContext(ctx).
Table("user_base_info AS user").
Select("user.id AS id").
Joins("LEFT JOIN voice_room_region_country AS region ON region.sys_origin = ? AND region.country_code = user.country_code AND region.enabled = ?", sysOrigin, true).
Where("user.origin_sys = ?", sysOrigin).
Where("(region.region_code = ? OR user.country_code = ?)", regionCode, regionCode).
Where("user.id > 0")
if len(onlineIDs) > 0 && len(onlineIDs) <= 1000 {
query = query.Where("user.id NOT IN ?", onlineIDs)
}
var users []continuousCandidateUser
if err := query.Order("RAND()").Limit(50).Scan(&users).Error; err != nil {
return 0, err
}
filtered := make([]continuousCandidateUser, 0, len(users))
onlineSet := make(map[int64]struct{}, len(onlineIDs))
for _, id := range onlineIDs {
onlineSet[id] = struct{}{}
}
for _, user := range users {
if user.ID <= 0 {
continue
}
if _, online := onlineSet[user.ID]; online {
continue
}
filtered = append(filtered, user)
}
if len(filtered) == 0 {
return 0, common.NewAppError(http.StatusNotFound, "simulation_offline_user_not_found", "offline user not found for selected region")
}
return filtered[rng.Intn(len(filtered))].ID, nil
}
func (w *BroadcastSimulationWorker) randomActiveRoom(ctx context.Context, sysOrigin string, rng *rand.Rand) (int64, error) {
rooms, err := w.activeRooms(ctx, sysOrigin, 10)
if err != nil {
return 0, err
}
if len(rooms) == 0 {
return 0, nil
}
return rooms[rng.Intn(len(rooms))].RoomID, nil
}
func (w *BroadcastSimulationWorker) randomGame(ctx context.Context, sysOrigin string) (continuousCandidateGame, error) {
var game continuousCandidateGame
err := w.db.WithContext(ctx).
Table("sys_game_list_config").
Select("game_id AS game_id, cover AS cover").
Where("sys_origin = ? AND cover <> ''", sysOrigin).
Order("RAND()").
Limit(1).
Scan(&game).Error
if err != nil {
return continuousCandidateGame{}, err
}
if strings.TrimSpace(game.GameID) == "" || strings.TrimSpace(game.Cover) == "" {
return continuousCandidateGame{}, common.NewAppError(http.StatusNotFound, "simulation_game_not_found", "game cover not found")
}
return game, nil
}
func (w *BroadcastSimulationWorker) activeRooms(ctx context.Context, sysOrigin string, minUsers int64) ([]continuousCandidateRoom, error) {
var cursor uint64
var rooms []continuousCandidateRoom
prefix := fmt.Sprintf("voice_room:online:%s:", strings.ToUpper(strings.TrimSpace(sysOrigin)))
for {
keys, next, err := w.redis.Scan(ctx, cursor, prefix+"*", 100).Result()
if err != nil {
return nil, err
}
for _, key := range keys {
count, err := w.redis.SCard(ctx, key).Result()
if err != nil || count <= minUsers {
continue
}
roomID, err := strconv.ParseInt(strings.TrimPrefix(key, prefix), 10, 64)
if err != nil || roomID <= 0 {
continue
}
rooms = append(rooms, continuousCandidateRoom{RoomID: roomID})
}
cursor = next
if cursor == 0 {
break
}
}
return rooms, nil
}
func (w *BroadcastSimulationWorker) onlineUserIDs(ctx context.Context, sysOrigin string) []int64 {
var cursor uint64
prefix := fmt.Sprintf("voice_room:online:%s:", strings.ToUpper(strings.TrimSpace(sysOrigin)))
seen := map[int64]struct{}{}
for {
keys, next, err := w.redis.Scan(ctx, cursor, prefix+"*", 100).Result()
if err != nil {
return nil
}
for _, key := range keys {
members, err := w.redis.SMembers(ctx, key).Result()
if err != nil {
continue
}
for _, raw := range members {
id, err := strconv.ParseInt(strings.TrimSpace(raw), 10, 64)
if err == nil && id > 0 {
seen[id] = struct{}{}
}
}
}
cursor = next
if cursor == 0 {
break
}
}
ids := make([]int64, 0, len(seen))
for id := range seen {
ids = append(ids, id)
}
return ids
}
func (w *BroadcastSimulationWorker) updateStatus(fn func(*ContinuousSimulationStatus)) {
w.mu.Lock()
defer w.mu.Unlock()
fn(&w.status)
w.status.UpdateTime = time.Now()
}
func (w *BroadcastSimulationWorker) setLastError(err error) {
if err == nil {
return
}
w.updateStatus(func(status *ContinuousSimulationStatus) {
status.LastError = err.Error()
status.LastRunAt = time.Now().Format(adminTimeLayout)
})
}
func (w *BroadcastSimulationWorker) incrementCount(luckyGift bool) {
w.updateStatus(func(status *ContinuousSimulationStatus) {
status.LastError = ""
status.LastRunAt = time.Now().Format(adminTimeLayout)
status.TotalCount++
if luckyGift {
status.LuckyGiftCount++
} else {
status.GameWinCount++
}
})
}
func normalizeContinuousRequest(req ContinuousSimulationRequest) ContinuousSimulationRequest {
req.SysOrigin = strings.ToUpper(strings.TrimSpace(req.SysOrigin))
req.RegionCodes = normalizeStringList(req.RegionCodes)
if req.MinDelaySeconds < 0 {
req.MinDelaySeconds = 0
}
if req.MaxDelaySeconds < req.MinDelaySeconds {
req.MaxDelaySeconds = req.MinDelaySeconds
}
req.BetAmounts = positiveInt64List(req.BetAmounts, []int64{100, 200, 500, 1000})
req.Multiples = positiveFloat64List(req.Multiples, []float64{10, 20, 50})
req.GameWinAmounts = positiveInt64List(req.GameWinAmounts, []int64{1000, 2000, 5000})
req.LuckyGiftWinAmounts = positiveInt64List(req.LuckyGiftWinAmounts, []int64{1000, 2000, 5000})
req.GiftCandyOptions = positiveInt64List(req.GiftCandyOptions, []int64{100, 200, 500})
req.GiftQuantityOptions = positiveInt64List(req.GiftQuantityOptions, []int64{1, 5, 10})
return req
}
func validateContinuousRequest(req ContinuousSimulationRequest) error {
if req.SysOrigin == "" {
return common.NewAppError(http.StatusBadRequest, "missing_sys_origin", "sysOrigin is required")
}
if len(req.RegionCodes) == 0 {
return common.NewAppError(http.StatusBadRequest, "missing_region_codes", "regionCodes is required")
}
if !req.EnableLuckyGift && !req.EnableGameWin {
return common.NewAppError(http.StatusBadRequest, "missing_broadcast_type", "enable at least one broadcast type")
}
return nil
}
func normalizeStringList(values []string) []string {
seen := map[string]struct{}{}
result := make([]string, 0, len(values))
for _, value := range values {
value = strings.ToUpper(strings.TrimSpace(value))
if value == "" {
continue
}
if _, ok := seen[value]; ok {
continue
}
seen[value] = struct{}{}
result = append(result, value)
}
return result
}
func positiveInt64List(values []int64, fallback []int64) []int64 {
result := make([]int64, 0, len(values))
for _, value := range values {
if value > 0 {
result = append(result, value)
}
}
if len(result) == 0 {
return append([]int64(nil), fallback...)
}
return result
}
func positiveFloat64List(values []float64, fallback []float64) []float64 {
result := make([]float64, 0, len(values))
for _, value := range values {
if value >= ThresholdMultiple {
result = append(result, value)
}
}
if len(result) == 0 {
return append([]float64(nil), fallback...)
}
return result
}
func randomDelaySeconds(rng *rand.Rand, min, max int) int {
if max <= min {
return min
}
return min + rng.Intn(max-min+1)
}
func pickInt64(rng *rand.Rand, values []int64) int64 {
if len(values) == 0 {
return 0
}
return values[rng.Intn(len(values))]
}
func pickFloat64(rng *rand.Rand, values []float64) float64 {
if len(values) == 0 {
return 0
}
return values[rng.Intn(len(values))]
}
func formatFloat(value float64) string {
return strconv.FormatFloat(value, 'f', -1, 64)
}
func simulationMessageType(kind string) (string, error) {
switch strings.ToUpper(strings.TrimSpace(kind)) {
case SimulationKindLuckyGift:
return MessageTypeLuckyGift, nil
case SimulationKindGameWin:
return MessageTypeBaishunWin, nil
default:
return "", common.NewAppError(http.StatusBadRequest, "invalid_broadcast_kind", "kind must be LUCKY_GIFT or GAME_WIN")
}
}
func simulationData(messageType string, req BroadcastSimulationRequest, amount int64) map[string]any {
now := strconv.FormatInt(time.Now().UnixMilli(), 10)
switch messageType {
case MessageTypeLuckyGift:
data := map[string]any{
"businessId": firstNonBlank(req.ProviderOrder, "admin-sim-lucky-gift-"+now),
"orderId": firstNonBlank(req.ProviderOrder, "admin-sim-order-"+now),
"globalNews": true,
"multipleType": "WIN",
}
setOptional(data, "acceptUserId", req.AcceptUserID)
setOptional(data, "giftId", req.GiftID)
setOptionalInt64(data, "giftCandy", req.GiftCandy)
setOptionalInt64(data, "giftQuantity", req.GiftQuantity)
return data
default:
data := map[string]any{
"gameRoundId": firstNonBlank(req.GameRoundID, "admin-sim-game-round-"+now),
"currencyDiff": amount,
}
setOptional(data, "gameId", req.GameID)
setOptional(data, "gameUrl", req.GameURL)
setOptionalInt64(data, "betAmount", req.BetAmount)
return data
}
}
func setOptional(data map[string]any, key string, value string) {
if value = strings.TrimSpace(value); value != "" {
data[key] = value
}
}
func setOptionalInt64(data map[string]any, key string, value string) {
parsed, err := parseOptionalInt64(value)
if err == nil && parsed > 0 {
data[key] = parsed
}
}
func firstNonBlank(values ...string) string {
for _, value := range values {
if trimmed := strings.TrimSpace(value); trimmed != "" {
return trimmed
}
}
return ""
}
func parseRequiredInt64(value string, field string) (int64, error) {
parsed, err := parseOptionalInt64(value)
if err != nil || parsed <= 0 {
return 0, common.NewAppError(http.StatusBadRequest, "invalid_"+field, fmt.Sprintf("%s must be a positive integer", field))
}
return parsed, nil
}
func parseOptionalInt64(value string) (int64, error) {
raw := strings.TrimSpace(value)
if raw == "" {
return 0, nil
}
return strconv.ParseInt(raw, 10, 64)
}
func parseRequiredFloat64(value string, field string) (float64, error) {
raw := strings.TrimSpace(value)
if raw == "" {
return 0, common.NewAppError(http.StatusBadRequest, "invalid_"+field, fmt.Sprintf("%s must be a positive number", field))
}
parsed, err := strconv.ParseFloat(raw, 64)
if err != nil || parsed <= 0 {
return 0, common.NewAppError(http.StatusBadRequest, "invalid_"+field, fmt.Sprintf("%s must be a positive number", field))
}
return parsed, nil
}
var _ RegionBroadcaster = (*regionimgroup.Service)(nil)