diff --git a/internal/router/region_im_group_routes.go b/internal/router/region_im_group_routes.go index 8eb8fd5..2615275 100644 --- a/internal/router/region_im_group_routes.go +++ b/internal/router/region_im_group_routes.go @@ -4,6 +4,7 @@ import ( "net/http" "chatapp3-golang/internal/config" + "chatapp3-golang/internal/repo" "chatapp3-golang/internal/service/highwin" "chatapp3-golang/internal/service/regionimgroup" @@ -14,6 +15,7 @@ import ( func registerRegionIMGroupRoutes( engine *gin.Engine, cfg config.Config, + repository *repo.Repository, javaClient authGateway, service *regionimgroup.Service, ) { @@ -23,7 +25,7 @@ func registerRegionIMGroupRoutes( registerRegionIMGroupAppRoutes(engine, javaClient, service) registerRegionIMGroupInternalRoutes(engine, cfg, service) - registerRegionIMGroupAdminRoutes(engine, javaClient, service) + registerRegionIMGroupAdminRoutes(engine, repository, javaClient, service) } func registerRegionIMGroupAppRoutes(engine *gin.Engine, javaClient authGateway, service *regionimgroup.Service) { @@ -63,10 +65,14 @@ func registerRegionIMGroupInternalRoutes(engine *gin.Engine, cfg config.Config, }) } -func registerRegionIMGroupAdminRoutes(engine *gin.Engine, javaClient authGateway, service *regionimgroup.Service) { +func registerRegionIMGroupAdminRoutes(engine *gin.Engine, repository *repo.Repository, javaClient authGateway, service *regionimgroup.Service) { registerRegionIMGroupAdminGroup(engine.Group("/region/config"), javaClient, service) registerRegionIMGroupAdminGroup(engine.Group("/resident-activity/voice-room-red-packet"), javaClient, service) - registerRegionBroadcastSimulationAdminGroup(engine.Group("/operate/region-broadcast"), javaClient, service) + var worker *highwin.BroadcastSimulationWorker + if repository != nil { + worker = highwin.NewBroadcastSimulationWorker(repository.DB, repository.Redis, service) + } + registerRegionBroadcastSimulationAdminGroup(engine.Group("/operate/region-broadcast"), javaClient, service, worker) } func registerRegionIMGroupAdminGroup(group *gin.RouterGroup, javaClient authGateway, service *regionimgroup.Service) { @@ -107,7 +113,7 @@ func registerRegionIMGroupAdminGroup(group *gin.RouterGroup, javaClient authGate }) } -func registerRegionBroadcastSimulationAdminGroup(group *gin.RouterGroup, javaClient authGateway, service *regionimgroup.Service) { +func registerRegionBroadcastSimulationAdminGroup(group *gin.RouterGroup, javaClient authGateway, service *regionimgroup.Service, worker *highwin.BroadcastSimulationWorker) { group.Use(consoleAuthMiddleware(javaClient)) group.POST("/simulate", func(c *gin.Context) { var req highwin.BroadcastSimulationRequest @@ -122,4 +128,23 @@ func registerRegionBroadcastSimulationAdminGroup(group *gin.RouterGroup, javaCli } writeOK(c, resp) }) + group.GET("/worker/status", func(c *gin.Context) { + writeOK(c, worker.Status()) + }) + group.POST("/worker/start", func(c *gin.Context) { + var req highwin.ContinuousSimulationRequest + if err := c.ShouldBindJSON(&req); err != nil { + writeError(c, regionimgroup.NewAppError(http.StatusBadRequest, "bad_request", err.Error())) + return + } + resp, err := worker.Start(c.Request.Context(), req) + if err != nil { + writeError(c, err) + return + } + writeOK(c, resp) + }) + group.POST("/worker/stop", func(c *gin.Context) { + writeOK(c, worker.Stop()) + }) } diff --git a/internal/router/router.go b/internal/router/router.go index d5a1b95..243fe52 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -93,7 +93,7 @@ func NewRouter( registerVipRoutes(engine, javaClient, services.VIP) registerWheelRoutes(engine, cfg, javaClient, services.Wheel) registerSmashEggRoutes(engine, javaClient, services.SmashEgg) - registerRegionIMGroupRoutes(engine, cfg, javaClient, services.RegionIMGroup) + registerRegionIMGroupRoutes(engine, cfg, repository, javaClient, services.RegionIMGroup) registerVoiceRoomRedPacketRoutes(engine, javaClient, services.VoiceRoomRedPacket) registerVoiceRoomRocketRoutes(engine, cfg, javaClient, services.VoiceRoomRocket) registerWeekStarRoutes(engine, javaClient, services.WeekStar) diff --git a/internal/service/highwin/admin.go b/internal/service/highwin/admin.go index af0cda4..faf9f19 100644 --- a/internal/service/highwin/admin.go +++ b/internal/service/highwin/admin.go @@ -3,18 +3,24 @@ 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 { @@ -45,6 +51,216 @@ type BroadcastSimulationResponse struct { 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, @@ -111,6 +327,266 @@ func SendSimulationBroadcast( }, 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: