diff --git a/cmd/api/main.go b/cmd/api/main.go index bee22da..1df1105 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -61,6 +61,7 @@ func main() { voiceRoomRedPacketService := voiceroomredpacket.NewService(app.Config, app.Repository.DB, app.Repository.Redis, app.Gateways.Wallet, regionIMGroupService, app.Gateways.Room) voiceRoomRedPacketService.SetRegionResolver(&app.Gateways) voiceRoomRocketService := voiceroomrocket.NewService(app.Config, app.Repository.DB, app.Repository.Redis, &app.Gateways) + voiceRoomRocketService.SetRegionBroadcaster(regionIMGroupService) weekStarService := weekstar.NewWeekStarService(app.Config, app.Repository.DB, app.Repository.Redis, &app.Gateways) roomTurnoverRewardService := roomturnoverreward.NewService(app.Config, app.Repository.DB, &app.Gateways) hostCenterService := hostcenter.NewService(app.Config, &app.Gateways) diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go index aa6e57f..bbe3604 100644 --- a/cmd/consumer/main.go +++ b/cmd/consumer/main.go @@ -31,6 +31,7 @@ func main() { voiceRoomRedPacketService := voiceroomredpacket.NewService(app.Config, app.Repository.DB, app.Repository.Redis, app.Gateways.Wallet, regionIMGroupService, app.Gateways.Room) voiceRoomRedPacketService.SetRegionResolver(&app.Gateways) voiceRoomRocketService := voiceroomrocket.NewService(app.Config, app.Repository.DB, app.Repository.Redis, &app.Gateways) + voiceRoomRocketService.SetRegionBroadcaster(regionIMGroupService) workerCtx, workerCancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer workerCancel() diff --git a/internal/service/voiceroomrocket/config.go b/internal/service/voiceroomrocket/config.go index 10883ad..d56d385 100644 --- a/internal/service/voiceroomrocket/config.go +++ b/internal/service/voiceroomrocket/config.go @@ -39,13 +39,12 @@ func (s *Service) GetAdminConfig(ctx context.Context, sysOrigin string) (*Config // SaveAdminConfig 保存后台系统配置。 func (s *Service) SaveAdminConfig(ctx context.Context, req SaveConfigRequest) (*ConfigResponse, error) { sysOrigin := s.normalizeSysOrigin(req.SysOrigin) - timezone := normalizeTimezone(req.Timezone) - if _, err := time.LoadLocation(timezone); err != nil { - return nil, NewAppError(http.StatusBadRequest, "invalid_timezone", err.Error()) - } + timezone := defaultTimezone maxLevel := req.MaxLevel if maxLevel <= 0 { maxLevel = defaultMaxLevel + } else if maxLevel > defaultMaxLevel { + maxLevel = defaultMaxLevel } rankingLimit := req.RankingTopLimit if rankingLimit <= 0 || rankingLimit > 100 { @@ -206,11 +205,28 @@ func (s *Service) ListRewardConfigs(ctx context.Context, sysOrigin string, level // SaveRewardConfigs 批量保存后台奖励配置。 func (s *Service) SaveRewardConfigs(ctx context.Context, req SaveRewardConfigsRequest) ([]RewardConfigPayload, error) { sysOrigin := s.normalizeSysOrigin(req.SysOrigin) - if len(req.Rewards) == 0 { - return nil, NewAppError(http.StatusBadRequest, "missing_reward_configs", "rewards is required") + targetLevels := map[int]struct{}{} + if req.Level > 0 { + if req.Level > defaultMaxLevel { + return nil, NewAppError(http.StatusBadRequest, "invalid_level", "level must be between 1 and 6") + } + targetLevels[req.Level] = struct{}{} } if err := s.repo.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error { now := time.Now() + savedIDsByLevel := map[int][]int64{} + if len(req.Rewards) == 0 { + if len(targetLevels) == 0 { + return NewAppError(http.StatusBadRequest, "missing_reward_configs", "rewards is required") + } + for level := range targetLevels { + if err := tx.Where("sys_origin = ? AND level = ?", sysOrigin, level). + Delete(&model.VoiceRoomRocketRewardConfig{}).Error; err != nil { + return err + } + } + return nil + } for _, item := range req.Rewards { scene := strings.ToUpper(strings.TrimSpace(item.RewardScene)) if scene != rewardSceneTop1 && scene != rewardSceneIgnite && scene != rewardSceneInRoom { @@ -219,6 +235,10 @@ func (s *Service) SaveRewardConfigs(ctx context.Context, req SaveRewardConfigsRe if item.Level <= 0 || item.Level > defaultMaxLevel { return NewAppError(http.StatusBadRequest, "invalid_level", "level must be between 1 and 6") } + if req.Level > 0 && item.Level != req.Level { + return NewAppError(http.StatusBadRequest, "invalid_level", "reward level must match request level") + } + targetLevels[item.Level] = struct{}{} rewardType := strings.ToUpper(strings.TrimSpace(item.RewardType)) if rewardType == "" { return NewAppError(http.StatusBadRequest, "missing_reward_type", "rewardType is required") @@ -226,12 +246,22 @@ func (s *Service) SaveRewardConfigs(ctx context.Context, req SaveRewardConfigsRe if !isSupportedRewardType(rewardType) { return NewAppError(http.StatusBadRequest, "invalid_reward_type", "rewardType is invalid") } + if rewardType == rewardTypeNone && scene != rewardSceneInRoom { + return NewAppError(http.StatusBadRequest, "invalid_reward_type", "NONE reward is only allowed for IN_ROOM") + } rewardName := strings.TrimSpace(item.RewardName) + if rewardType == rewardTypeNone && rewardName == "" { + rewardName = "轮空" + } if rewardName == "" { return NewAppError(http.StatusBadRequest, "missing_reward_name", "rewardName is required") } amount := item.RewardAmount - if amount <= 0 { + if rewardType == rewardTypeNone { + amount = 0 + item.RewardItemID = nil + item.ExpireDays = nil + } else if amount <= 0 { amount = 1 } var row model.VoiceRoomRocketRewardConfig @@ -268,12 +298,22 @@ func (s *Service) SaveRewardConfigs(ctx context.Context, req SaveRewardConfigsRe if err := tx.Save(&row).Error; err != nil { return err } + savedIDsByLevel[item.Level] = append(savedIDsByLevel[item.Level], row.ID) + } + for level := range targetLevels { + query := tx.Where("sys_origin = ? AND level = ?", sysOrigin, level) + if ids := savedIDsByLevel[level]; len(ids) > 0 { + query = query.Where("id NOT IN ?", ids) + } + if err := query.Delete(&model.VoiceRoomRocketRewardConfig{}).Error; err != nil { + return err + } } return nil }); err != nil { return nil, err } - return s.ListRewardConfigs(ctx, sysOrigin, 0) + return s.ListRewardConfigs(ctx, sysOrigin, req.Level) } func (s *Service) getConfigResponse(ctx context.Context, sysOrigin string, currentLevel int) (*ConfigResponse, error) { @@ -304,7 +344,7 @@ func (s *Service) loadConfig(ctx context.Context, sysOrigin string) (configSnaps return configSnapshot{ SysOrigin: sysOrigin, Enabled: false, - Timezone: normalizeTimezone(s.cfg.VoiceRoomRocket.Timezone), + Timezone: defaultTimezone, MaxLevel: defaultMaxLevel, RankingTopLimit: defaultRankingTopLimit, RewardRecordKeepDays: defaultRewardRecordKeepDays, @@ -320,7 +360,7 @@ func (s *Service) loadConfig(ctx context.Context, sysOrigin string) (configSnaps ID: row.ID, SysOrigin: row.SysOrigin, Enabled: row.Enabled, - Timezone: normalizeTimezone(row.Timezone), + Timezone: defaultTimezone, MaxLevel: positiveOrDefault(row.MaxLevel, defaultMaxLevel), RankingTopLimit: clampPositive(row.RankingTopLimit, defaultRankingTopLimit, 100), RewardRecordKeepDays: positiveOrDefault(row.RewardRecordKeepDays, defaultRewardRecordKeepDays), diff --git a/internal/service/voiceroomrocket/event.go b/internal/service/voiceroomrocket/event.go index 3b3c972..856d909 100644 --- a/internal/service/voiceroomrocket/event.go +++ b/internal/service/voiceroomrocket/event.go @@ -193,7 +193,10 @@ func (s *Service) ProcessGiftEvent(ctx context.Context, event giftEvent, rawPayl return err } if newEnergy >= statusRow.NeedEnergy { - launch, err := s.launchLocked(ctx, tx, statusRow, sendUserID, now) + if err := s.markLevelReachedLocked(ctx, tx, statusRow, sendUserID, now); err != nil { + return err + } + launch, err := s.scheduleLaunchLocked(ctx, tx, statusRow, sendUserID, now) if err != nil { return err } @@ -209,7 +212,7 @@ func (s *Service) ProcessGiftEvent(ctx context.Context, event giftEvent, rawPayl if result.Launched { _ = s.notifyLaunch(context.Background(), result.Launch) } else if resp.Processed { - _ = s.notifyStatusUpdate(context.Background(), sysOrigin, roomID) + _ = s.notifyStatusUpdate(context.Background(), sysOrigin, roomID, sendUserID) } return resp, nil } @@ -292,6 +295,35 @@ func (s *Service) addContribution(ctx context.Context, tx *gorm.DB, statusRow mo }).Error } +func (s *Service) markLevelReachedLocked(ctx context.Context, tx *gorm.DB, statusRow model.VoiceRoomRocketStatus, igniteUserID int64, now time.Time) error { + return tx.WithContext(ctx).Model(&model.VoiceRoomRocketLevelRound{}). + Where( + "sys_origin = ? AND room_id = ? AND day_key = ? AND round_no = ? AND level = ?", + statusRow.SysOrigin, statusRow.RoomID, statusRow.DayKey, statusRow.RoundNo, statusRow.CurrentLevel, + ). + Updates(map[string]any{ + "final_energy": statusRow.NeedEnergy, + "status": levelRoundStatusReached, + "ignite_user_id": igniteUserID, + "reach_time": now, + "update_time": now, + }).Error +} + +func (s *Service) scheduleLaunchLocked(ctx context.Context, tx *gorm.DB, statusRow model.VoiceRoomRocketStatus, igniteUserID int64, now time.Time) (launchResult, error) { + if s.repo.Redis == nil { + return s.launchLocked(ctx, tx, statusRow, igniteUserID, now) + } + due := now.Add(time.Duration(defaultLaunchDelaySeconds) * time.Second) + if err := s.repo.Redis.ZAdd(ctx, s.launchDueZSetKey(), redis.Z{ + Score: float64(due.Unix()), + Member: strconv.FormatInt(statusRow.ID, 10), + }).Err(); err != nil { + return launchResult{}, err + } + return launchResult{Launched: false}, nil +} + func (s *Service) launchLocked(ctx context.Context, tx *gorm.DB, statusRow model.VoiceRoomRocketStatus, igniteUserID int64, now time.Time) (launchResult, error) { top1UserID, err := s.resolveTop1UserID(ctx, tx, statusRow) if err != nil { @@ -326,7 +358,6 @@ func (s *Service) launchLocked(ctx context.Context, tx *gorm.DB, statusRow model } return launchResult{}, err } - reachTime := now if err := tx.Model(&model.VoiceRoomRocketLevelRound{}). Where( "sys_origin = ? AND room_id = ? AND day_key = ? AND round_no = ? AND level = ?", @@ -337,7 +368,7 @@ func (s *Service) launchLocked(ctx context.Context, tx *gorm.DB, statusRow model "status": levelRoundStatusLaunched, "top1_user_id": top1UserID, "ignite_user_id": igniteUserID, - "reach_time": reachTime, + "reach_time": gorm.Expr("COALESCE(reach_time, ?)", now), "launch_time": now, "update_time": now, }).Error; err != nil { diff --git a/internal/service/voiceroomrocket/event_test.go b/internal/service/voiceroomrocket/event_test.go index 237edc1..292e00a 100644 --- a/internal/service/voiceroomrocket/event_test.go +++ b/internal/service/voiceroomrocket/event_test.go @@ -2,7 +2,9 @@ package voiceroomrocket import ( "context" + "encoding/json" "strconv" + "strings" "testing" "time" @@ -199,11 +201,19 @@ func TestInRoomRewardWorkerSnapshotsSelectsAndGrants(t *testing.T) { if err != nil { t.Fatalf("ProcessGiftPayload() error = %v", err) } - if !resp.Launched || resp.LaunchNo == "" { - t.Fatalf("launch response = %+v", resp) + if !resp.Processed || resp.Launched || resp.LaunchNo != "" { + t.Fatalf("launch should be delayed, response = %+v", resp) } - processed, err := service.ProcessDueInRoomRewards(ctx, time.Now().Add(20*time.Second), 10) + launched, err := service.ProcessDueLaunches(ctx, time.Now().Add(11*time.Second), 10) + if err != nil { + t.Fatalf("ProcessDueLaunches() error = %v", err) + } + if launched != 1 { + t.Fatalf("due launches = %d, want 1", launched) + } + + processed, err := service.ProcessDueInRoomRewards(ctx, time.Now().Add(30*time.Second), 10) if err != nil { t.Fatalf("ProcessDueInRoomRewards() error = %v", err) } @@ -245,6 +255,87 @@ func TestInRoomRewardWorkerSnapshotsSelectsAndGrants(t *testing.T) { } } +func TestProcessGiftPayloadDelaysLaunchWithRedis(t *testing.T) { + gateway := newFakeRocketGateway() + service, db, _, mr := newTestServiceWithRedis(t, gateway) + defer mr.Close() + + ctx := context.Background() + seedRocketConfig(t, db) + seedLevelConfigs(t, db, []int64{100, 200, 300, 400, 500, 600}) + seedRewardConfig(t, db, 1, rewardSceneTop1, rewardTypeGold, 10) + + payload := `{ + "trackId":"track-launch-delay", + "sysOrigin":"LIKEI", + "sendUserId":"1001", + "roomId":"9001", + "quantity":1, + "giftConfig":{"id":"11","giftCandy":100,"type":"GOLD","giftTab":"NORMAL"}, + "createTime":` + strconvFormatMillis(time.Now()) + ` + }` + resp, err := service.ProcessGiftPayload(ctx, payload) + if err != nil { + t.Fatalf("ProcessGiftPayload() error = %v", err) + } + if !resp.Processed || resp.Launched { + t.Fatalf("response = %+v, want processed delayed launch", resp) + } + + var status model.VoiceRoomRocketStatus + if err := db.Where("sys_origin = ? AND room_id = ?", "LIKEI", int64(9001)).First(&status).Error; err != nil { + t.Fatalf("load status: %v", err) + } + if status.Status != statusLaunching || status.CurrentEnergy != 100 { + t.Fatalf("status = %+v, want launching full energy", status) + } + var launchCount int64 + if err := db.Model(&model.VoiceRoomRocketLaunch{}).Count(&launchCount).Error; err != nil { + t.Fatalf("count launch before due: %v", err) + } + if launchCount != 0 { + t.Fatalf("launch count before due = %d, want 0", launchCount) + } + + processed, err := service.ProcessDueLaunches(ctx, time.Now().Add(9*time.Second), 10) + if err != nil { + t.Fatalf("ProcessDueLaunches(before due) error = %v", err) + } + if processed != 0 { + t.Fatalf("launches before due = %d, want 0", processed) + } + processed, err = service.ProcessDueLaunches(ctx, time.Now().Add(11*time.Second), 10) + if err != nil { + t.Fatalf("ProcessDueLaunches(after due) error = %v", err) + } + if processed != 1 { + t.Fatalf("launches after due = %d, want 1", processed) + } + if err := db.Model(&model.VoiceRoomRocketLaunch{}).Count(&launchCount).Error; err != nil { + t.Fatalf("count launch after due: %v", err) + } + if launchCount != 1 { + t.Fatalf("launch count after due = %d, want 1", launchCount) + } +} + +func TestPickInRoomRewardCanBeEmpty(t *testing.T) { + service, _ := newTestService(t) + launch := model.VoiceRoomRocketLaunch{LaunchNo: "VRR-empty"} + rewards := []model.VoiceRoomRocketRewardConfig{ + { + RewardScene: rewardSceneInRoom, + RewardType: rewardTypeNone, + RewardName: "轮空", + Weight: 1, + }, + } + got := service.pickInRoomRewardsForUser(launch, rewards, 1001) + if len(got) != 0 { + t.Fatalf("picked rewards = %+v, want empty", got) + } +} + func TestNotifyLaunchPublishesIMAndRedisStream(t *testing.T) { gateway := newFakeRocketGateway() service, db, rdb, mr := newTestServiceWithRedis(t, gateway) @@ -278,6 +369,42 @@ func TestNotifyLaunchPublishesIMAndRedisStream(t *testing.T) { } } +func TestStatusUpdatePublishesRegionIM(t *testing.T) { + gateway := newFakeRocketGateway() + service, db, _, mr := newTestServiceWithRedis(t, gateway) + defer mr.Close() + service.regionResolver = fakeRocketRegionResolver{regions: map[string]string{"LIKEI|SA": "AR"}} + regionIM := &fakeRocketRegionBroadcaster{} + service.SetRegionBroadcaster(regionIM) + + ctx := context.Background() + seedRocketConfig(t, db) + seedLevelConfigs(t, db, []int64{100, 200, 300, 400, 500, 600}) + + payload := `{ + "trackId":"track-region-status", + "sysOrigin":"LIKEI", + "sendUserId":"1001", + "roomId":"9001", + "quantity":1, + "giftConfig":{"id":"11","giftCandy":10,"type":"GOLD","giftTab":"NORMAL"}, + "createTime":` + strconvFormatMillis(time.Now()) + ` + }` + resp, err := service.ProcessGiftPayload(ctx, payload) + if err != nil { + t.Fatalf("ProcessGiftPayload() error = %v", err) + } + if !resp.Processed { + t.Fatalf("response = %+v, want processed", resp) + } + if len(regionIM.messages) != 1 { + t.Fatalf("region messages = %+v, want one", regionIM.messages) + } + if regionIM.messages[0].regionCode != "AR" || regionIM.messages[0].messageType != imTypeStatusUpdate { + t.Fatalf("region message = %+v", regionIM.messages[0]) + } +} + func TestDayKeyUsesUTCPlus3BusinessDay(t *testing.T) { location := resolveLocation("Asia/Riyadh") before := time.Date(2026, 5, 13, 20, 59, 59, 0, time.UTC) @@ -290,6 +417,110 @@ func TestDayKeyUsesUTCPlus3BusinessDay(t *testing.T) { } } +func TestSaveAdminConfigForcesRiyadhTimezoneAndReturnsEnergyRule(t *testing.T) { + service, _ := newTestService(t) + ctx := context.Background() + + resp, err := service.SaveAdminConfig(ctx, SaveConfigRequest{ + SysOrigin: "LIKEI", + Enabled: true, + Timezone: "UTC", + MaxLevel: 6, + RankingTopLimit: 50, + RewardRecordKeepDays: 35, + BroadcastDurationSeconds: 7, + InRoomRewardDelaySeconds: 10, + EnergyRuleJSON: json.RawMessage(`{"allowGiftIds":[11]}`), + RuleText: json.RawMessage(`{"title":"rocket"}`), + }) + if err != nil { + t.Fatalf("SaveAdminConfig() error = %v", err) + } + if resp.Timezone != defaultTimezone { + t.Fatalf("timezone = %s, want %s", resp.Timezone, defaultTimezone) + } + if string(resp.EnergyRuleJSON) != `{"allowGiftIds":[11]}` { + t.Fatalf("energyRuleJson = %s", string(resp.EnergyRuleJSON)) + } +} + +func TestSaveRewardConfigsReplacesAndClearsSelectedLevel(t *testing.T) { + service, db := newTestService(t) + ctx := context.Background() + seedRewardConfig(t, db, 1, rewardSceneTop1, rewardTypeGold, 10) + seedRewardConfig(t, db, 1, rewardSceneIgnite, rewardTypeGold, 5) + seedRewardConfig(t, db, 2, rewardSceneTop1, rewardTypeGold, 20) + + saved, err := service.SaveRewardConfigs(ctx, SaveRewardConfigsRequest{ + SysOrigin: "LIKEI", + Level: 1, + Rewards: []RewardConfigPayload{ + { + Level: 1, + RewardScene: rewardSceneTop1, + RewardType: rewardTypeGold, + RewardName: "new gold", + RewardAmount: 99, + Enabled: true, + Sort: 1, + }, + }, + }) + if err != nil { + t.Fatalf("SaveRewardConfigs(replace) error = %v", err) + } + if len(saved) != 1 || saved[0].RewardAmount != 99 { + t.Fatalf("saved rewards = %+v, want one amount 99", saved) + } + + var level1Count int64 + if err := db.Model(&model.VoiceRoomRocketRewardConfig{}). + Where("sys_origin = ? AND level = ?", "LIKEI", 1). + Count(&level1Count).Error; err != nil { + t.Fatalf("count level 1 rewards: %v", err) + } + if level1Count != 1 { + t.Fatalf("level 1 reward count = %d, want 1", level1Count) + } + var level2Count int64 + if err := db.Model(&model.VoiceRoomRocketRewardConfig{}). + Where("sys_origin = ? AND level = ?", "LIKEI", 2). + Count(&level2Count).Error; err != nil { + t.Fatalf("count level 2 rewards: %v", err) + } + if level2Count != 1 { + t.Fatalf("level 2 reward count = %d, want 1", level2Count) + } + + saved, err = service.SaveRewardConfigs(ctx, SaveRewardConfigsRequest{ + SysOrigin: "LIKEI", + Level: 1, + Rewards: []RewardConfigPayload{}, + }) + if err != nil { + t.Fatalf("SaveRewardConfigs(clear) error = %v", err) + } + if len(saved) != 0 { + t.Fatalf("saved rewards after clear = %+v, want empty", saved) + } + if err := db.Model(&model.VoiceRoomRocketRewardConfig{}). + Where("sys_origin = ? AND level = ?", "LIKEI", 1). + Count(&level1Count).Error; err != nil { + t.Fatalf("count cleared level 1 rewards: %v", err) + } + if level1Count != 0 { + t.Fatalf("level 1 reward count after clear = %d, want 0", level1Count) + } + if err := db.Model(&model.VoiceRoomRocketRewardConfig{}). + Where("sys_origin = ? AND level = ?", "LIKEI", 2). + Count(&level2Count).Error; err != nil { + t.Fatalf("count retained level 2 rewards: %v", err) + } + if level2Count != 1 { + t.Fatalf("level 2 reward count after clear = %d, want 1", level2Count) + } +} + func newTestService(t *testing.T) (*Service, *gorm.DB) { t.Helper() db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{}) @@ -351,6 +582,21 @@ type fakeRocketIMMessage struct { body any } +type fakeRocketRegionResolver struct { + regions map[string]string +} + +type fakeRocketRegionBroadcaster struct { + messages []fakeRocketRegionMessage +} + +type fakeRocketRegionMessage struct { + sysOrigin string + regionCode string + messageType string + body any +} + func newFakeRocketGateway() *fakeRocketGateway { return &fakeRocketGateway{goldEvents: map[string]integration.GoldReceiptCommand{}} } @@ -364,7 +610,7 @@ func (g *fakeRocketGateway) MapRoomProfiles(ctx context.Context, roomIDs []int64 } func (g *fakeRocketGateway) GetUserProfile(ctx context.Context, userID int64) (integration.UserProfile, error) { - return integration.UserProfile{ID: integration.Int64Value(userID), Account: strconv.FormatInt(userID, 10), UserNickname: "user"}, nil + return integration.UserProfile{ID: integration.Int64Value(userID), Account: strconv.FormatInt(userID, 10), UserNickname: "user", CountryCode: "SA"}, nil } func (g *fakeRocketGateway) ChangeGoldBalance(ctx context.Context, cmd integration.GoldReceiptCommand) error { @@ -404,6 +650,24 @@ func (g *fakeRocketIM) SendCustomGroupMessage(ctx context.Context, groupID strin return nil } +func (r fakeRocketRegionResolver) ResolveRegionCodeByCountryCode(ctx context.Context, sysOrigin string, countryCode string) (string, error) { + return r.regions[sysOrigin+"|"+countryCode], nil +} + +func (b *fakeRocketRegionBroadcaster) SendRegionCustomMessage(ctx context.Context, sysOrigin string, regionCode string, body any) (string, error) { + messageType := "" + if payload, ok := body.(map[string]any); ok { + messageType, _ = payload["type"].(string) + } + b.messages = append(b.messages, fakeRocketRegionMessage{ + sysOrigin: sysOrigin, + regionCode: regionCode, + messageType: messageType, + body: body, + }) + return "@region-" + strings.ToLower(regionCode), nil +} + func seedRocketConfig(t *testing.T, db *gorm.DB) { t.Helper() now := time.Now() diff --git a/internal/service/voiceroomrocket/in_room_reward.go b/internal/service/voiceroomrocket/in_room_reward.go index 8f79b5b..ce23d87 100644 --- a/internal/service/voiceroomrocket/in_room_reward.go +++ b/internal/service/voiceroomrocket/in_room_reward.go @@ -232,20 +232,19 @@ func (s *Service) pickInRoomRewardsForUser(launch model.VoiceRoomRocketLaunch, r if len(rewards) == 0 { return nil } - fixed := make([]model.VoiceRoomRocketRewardConfig, 0, len(rewards)) weighted := make([]model.VoiceRoomRocketRewardConfig, 0, len(rewards)) totalWeight := 0 for _, reward := range rewards { - if reward.Weight <= 0 { - fixed = append(fixed, reward) - continue + weight := reward.Weight + if weight <= 0 { + weight = 1 } + reward.Weight = weight weighted = append(weighted, reward) - totalWeight += reward.Weight + totalWeight += weight } - result := append([]model.VoiceRoomRocketRewardConfig(nil), fixed...) if totalWeight <= 0 { - return result + return nil } rng := rand.New(rand.NewSource(seedFromStrings(launch.LaunchNo, strconv.FormatInt(userID, 10), "reward"))) point := rng.Intn(totalWeight) + 1 @@ -253,11 +252,13 @@ func (s *Service) pickInRoomRewardsForUser(launch model.VoiceRoomRocketLaunch, r for _, reward := range weighted { acc += reward.Weight if point <= acc { - result = append(result, reward) - break + if strings.EqualFold(strings.TrimSpace(reward.RewardType), rewardTypeNone) { + return nil + } + return []model.VoiceRoomRocketRewardConfig{reward} } } - return result + return nil } func seedFromStrings(parts ...string) int64 { diff --git a/internal/service/voiceroomrocket/launch_worker.go b/internal/service/voiceroomrocket/launch_worker.go new file mode 100644 index 0000000..19ef6a7 --- /dev/null +++ b/internal/service/voiceroomrocket/launch_worker.go @@ -0,0 +1,142 @@ +package voiceroomrocket + +import ( + "context" + "errors" + "strconv" + "strings" + "time" + + "chatapp3-golang/internal/model" + + "github.com/redis/go-redis/v9" + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +// ProcessDueLaunches 扫描进度满 10 秒后的火箭发射任务。 +func (s *Service) ProcessDueLaunches(ctx context.Context, now time.Time, limit int64) (int, error) { + if limit <= 0 { + limit = int64(defaultWorkerBatchSize) + } + processed := 0 + if s.repo.Redis != nil { + statusIDs, err := s.repo.Redis.ZRangeByScore(ctx, s.launchDueZSetKey(), &redis.ZRangeBy{ + Min: "-inf", + Max: strconv.FormatInt(now.Unix(), 10), + Count: limit, + }).Result() + if err != nil { + return 0, err + } + for _, rawID := range statusIDs { + statusID, parseErr := strconv.ParseInt(strings.TrimSpace(rawID), 10, 64) + if parseErr != nil || statusID <= 0 { + _ = s.repo.Redis.ZRem(context.Background(), s.launchDueZSetKey(), rawID).Err() + continue + } + done, err := s.processDueLaunchStatus(ctx, statusID, now) + if err != nil { + return processed, err + } + if done { + if err := s.repo.Redis.ZRem(context.Background(), s.launchDueZSetKey(), rawID).Err(); err != nil { + return processed, err + } + processed++ + } + } + } + backfilled, err := s.processStaleLaunchingStatuses(ctx, now, limit) + if err != nil { + return processed, err + } + return processed + backfilled, nil +} + +func (s *Service) processStaleLaunchingStatuses(ctx context.Context, now time.Time, limit int64) (int, error) { + if limit <= 0 { + limit = int64(defaultWorkerBatchSize) + } + var rows []model.VoiceRoomRocketStatus + if err := s.repo.DB.WithContext(ctx). + Where("status = ?", statusLaunching). + Order("update_time ASC"). + Limit(int(limit)). + Find(&rows).Error; err != nil { + return 0, err + } + processed := 0 + for _, row := range rows { + done, err := s.processDueLaunchStatus(ctx, row.ID, now) + if err != nil { + return processed, err + } + if done { + processed++ + } + } + return processed, nil +} + +func (s *Service) processDueLaunchStatus(ctx context.Context, statusID int64, now time.Time) (bool, error) { + var launch model.VoiceRoomRocketLaunch + done := false + var launched bool + err := s.repo.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var statusRow model.VoiceRoomRocketStatus + err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). + Where("id = ?", statusID). + First(&statusRow).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + done = true + return nil + } + if err != nil { + return err + } + if statusRow.Status != statusLaunching || statusRow.CurrentEnergy < statusRow.NeedEnergy { + done = true + return nil + } + var round model.VoiceRoomRocketLevelRound + err = tx.Clauses(clause.Locking{Strength: "UPDATE"}). + Where( + "sys_origin = ? AND room_id = ? AND day_key = ? AND round_no = ? AND level = ?", + statusRow.SysOrigin, statusRow.RoomID, statusRow.DayKey, statusRow.RoundNo, statusRow.CurrentLevel, + ). + First(&round).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil + } + if err != nil { + return err + } + if round.ReachTime == nil || round.IgniteUserID == nil || *round.IgniteUserID <= 0 { + return nil + } + due := round.ReachTime.Add(time.Duration(defaultLaunchDelaySeconds) * time.Second) + if now.Before(due) { + return nil + } + result, err := s.launchLocked(ctx, tx, statusRow, *round.IgniteUserID, now) + if err != nil { + return err + } + launch = result.Launch + launched = result.Launched + done = true + return nil + }) + if err != nil { + return false, err + } + if launched { + _ = s.notifyLaunch(context.Background(), launch) + } + return done, nil +} + +func (s *Service) launchDueZSetKey() string { + return "voice_room:rocket:launch_due" +} diff --git a/internal/service/voiceroomrocket/lifecycle.go b/internal/service/voiceroomrocket/lifecycle.go index d29897e..738fa7b 100644 --- a/internal/service/voiceroomrocket/lifecycle.go +++ b/internal/service/voiceroomrocket/lifecycle.go @@ -17,6 +17,7 @@ func (s *Service) Start(ctx context.Context) error { if err := s.StartMessageConsumer(ctx); err != nil { return err } + s.StartLaunchWorker(ctx) s.StartInRoomRewardWorker(ctx) s.StartRewardGrantWorker(ctx) return nil @@ -38,6 +39,33 @@ func (s *Service) StartMessageConsumer(ctx context.Context) error { return nil } +// StartLaunchWorker 启动满进度后延迟发射扫描。 +func (s *Service) StartLaunchWorker(ctx context.Context) { + interval := time.Duration(s.cfg.VoiceRoomRocket.InRoomRewardScanIntervalSeconds) * time.Second + if interval <= 0 { + interval = defaultWorkerInterval + } + batchSize := int64(s.cfg.VoiceRoomRocket.InRoomRewardBatchSize) + if batchSize <= 0 { + batchSize = defaultWorkerBatchSize + } + log.Printf("voice room rocket launch worker started. interval=%s batch=%d delay=%ds", interval, batchSize, defaultLaunchDelaySeconds) + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + if _, err := s.ProcessDueLaunches(ctx, time.Now(), batchSize); err != nil && ctx.Err() == nil { + log.Printf("voice room rocket launch scan failed: %v", err) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } + }() +} + // StartInRoomRewardWorker 启动在房奖励到期扫描。 func (s *Service) StartInRoomRewardWorker(ctx context.Context) { if s.repo.Redis == nil { diff --git a/internal/service/voiceroomrocket/mapper.go b/internal/service/voiceroomrocket/mapper.go index 681a17c..38c4939 100644 --- a/internal/service/voiceroomrocket/mapper.go +++ b/internal/service/voiceroomrocket/mapper.go @@ -177,6 +177,9 @@ func rewardPreviewPayload(rows []model.VoiceRoomRocketRewardConfig) RewardPrevie }) resp := RewardPreviewResponse{} for _, row := range rows { + if strings.EqualFold(strings.TrimSpace(row.RewardType), rewardTypeNone) { + continue + } payload := rewardPayload(row) switch strings.ToUpper(strings.TrimSpace(row.RewardScene)) { case rewardSceneTop1: @@ -211,7 +214,8 @@ func isSupportedRewardType(rewardType string) bool { rewardTypeChatBubble, rewardTypeVIPCard, rewardTypeGift, - rewardTypeBadge: + rewardTypeBadge, + rewardTypeNone: return true default: return false @@ -234,6 +238,10 @@ func configResponse(snapshot configSnapshot, levels []model.VoiceRoomRocketLevel if strings.TrimSpace(snapshot.RuleText) != "" { ruleText = json.RawMessage(snapshot.RuleText) } + var energyRuleJSON json.RawMessage + if strings.TrimSpace(snapshot.EnergyRuleJSON) != "" { + energyRuleJSON = json.RawMessage(snapshot.EnergyRuleJSON) + } return &ConfigResponse{ Configured: snapshot.Configured, ID: snapshot.ID, @@ -245,6 +253,7 @@ func configResponse(snapshot configSnapshot, levels []model.VoiceRoomRocketLevel RewardRecordKeepDays: snapshot.RewardRecordKeepDays, BroadcastDurationSeconds: snapshot.BroadcastDurationSeconds, InRoomRewardDelaySeconds: snapshot.InRoomRewardDelaySeconds, + EnergyRuleJSON: energyRuleJSON, RuleText: ruleText, Levels: levelPayloads, RewardPreview: rewardPreviewPayload(rewards), diff --git a/internal/service/voiceroomrocket/notify.go b/internal/service/voiceroomrocket/notify.go index 48966fa..71ef797 100644 --- a/internal/service/voiceroomrocket/notify.go +++ b/internal/service/voiceroomrocket/notify.go @@ -13,12 +13,27 @@ import ( "gorm.io/gorm" ) -func (s *Service) notifyStatusUpdate(ctx context.Context, sysOrigin string, roomID int64) error { +func (s *Service) notifyStatusUpdate(ctx context.Context, sysOrigin string, roomID int64, senderUserIDs ...int64) error { + data := map[string]any{ + "roomId": strconv.FormatInt(roomID, 10), + } + if status, err := s.GetStatus(ctx, AuthUser{SysOrigin: sysOrigin}, roomID); err == nil && status != nil { + payload, _ := json.Marshal(status) + _ = json.Unmarshal(payload, &data) + } + senderUserID := int64(0) + if len(senderUserIDs) > 0 { + senderUserID = senderUserIDs[0] + } + if senderUserID > 0 { + data["senderUserId"] = strconv.FormatInt(senderUserID, 10) + } if s.repo.Redis != nil { body, _ := json.Marshal(map[string]any{ "type": imTypeStatusUpdate, "sysOrigin": sysOrigin, "roomId": strconv.FormatInt(roomID, 10), + "data": data, }) streamKey := strings.TrimSpace(s.cfg.VoiceRoomRocket.BroadcastStreamKey) if streamKey == "" { @@ -33,18 +48,45 @@ func (s *Service) notifyStatusUpdate(ctx context.Context, sysOrigin string, room }, }).Err() } + _ = s.notifyRegionStatusUpdate(ctx, sysOrigin, senderUserID, data) if s.im == nil { return nil } groupID := s.resolveRoomAccount(ctx, roomID) return s.im.SendCustomGroupMessage(ctx, groupID, map[string]any{ "type": imTypeStatusUpdate, - "data": map[string]any{ - "roomId": strconv.FormatInt(roomID, 10), - }, + "data": data, }) } +func (s *Service) notifyRegionStatusUpdate(ctx context.Context, sysOrigin string, senderUserID int64, data map[string]any) error { + if s.regionBroadcast == nil || s.regionResolver == nil || s.userProfiles == nil || senderUserID <= 0 { + return nil + } + profile, err := s.userProfiles.GetUserProfile(ctx, senderUserID) + if err != nil { + return err + } + countryCode := strings.TrimSpace(profile.CountryCode) + if countryCode == "" { + return nil + } + regionCode, err := s.regionResolver.ResolveRegionCodeByCountryCode(ctx, sysOrigin, countryCode) + if err != nil { + return err + } + regionCode = strings.TrimSpace(regionCode) + if regionCode == "" { + return nil + } + body := map[string]any{ + "type": imTypeStatusUpdate, + "data": data, + } + _, err = s.regionBroadcast.SendRegionCustomMessage(ctx, sysOrigin, regionCode, body) + return err +} + func (s *Service) notifyLaunch(ctx context.Context, launch model.VoiceRoomRocketLaunch) error { triggerUser := map[string]any{} if launch.IgniteUserID != nil && s.userProfiles != nil { diff --git a/internal/service/voiceroomrocket/reward.go b/internal/service/voiceroomrocket/reward.go index c9f88ae..6fcd1f8 100644 --- a/internal/service/voiceroomrocket/reward.go +++ b/internal/service/voiceroomrocket/reward.go @@ -46,6 +46,9 @@ func (s *Service) createImmediateRewardRecords(ctx context.Context, tx *gorm.DB, } func (s *Service) createRewardRecord(ctx context.Context, tx *gorm.DB, launch model.VoiceRoomRocketLaunch, reward model.VoiceRoomRocketRewardConfig, userID int64, now time.Time) (model.VoiceRoomRocketRewardRecord, error) { + if strings.EqualFold(strings.TrimSpace(reward.RewardType), rewardTypeNone) { + return model.VoiceRoomRocketRewardRecord{}, nil + } id, err := utils.NextID() if err != nil { return model.VoiceRoomRocketRewardRecord{}, err diff --git a/internal/service/voiceroomrocket/status.go b/internal/service/voiceroomrocket/status.go index 7cc0c42..9b130cf 100644 --- a/internal/service/voiceroomrocket/status.go +++ b/internal/service/voiceroomrocket/status.go @@ -3,6 +3,7 @@ package voiceroomrocket import ( "context" "errors" + "math" "net/http" "time" @@ -43,20 +44,23 @@ func (s *Service) GetStatus(ctx context.Context, user AuthUser, roomID int64) (* shake := status.NeedEnergy > 0 && float64(status.CurrentEnergy)*100/float64(status.NeedEnergy) >= parsePercent(currentLevelConfig.ShakeThresholdPercent) kings, _ := s.topKingRecords(ctx, snapshot.SysOrigin, roomID, status.DayKey, status.RoundNo, status.CurrentLevel, 3) + countdownSeconds := s.launchCountdownSeconds(ctx, status, time.Now()) return &StatusResponse{ - RoomID: roomID, - DayKey: status.DayKey, - RoundNo: status.RoundNo, - DisplayRound: status.RoundNo > 1, - CurrentLevel: status.CurrentLevel, - CurrentEnergy: status.CurrentEnergy, - NeedEnergy: status.NeedEnergy, - EnergyPercent: percent, - DisplayPercent: displayPercent, - Shake: shake, - Levels: buildLevelPayloads(levels), - RewardPreview: rewardPreviewPayload(rewards), - RocketKings: kings, + RoomID: roomID, + DayKey: status.DayKey, + RoundNo: status.RoundNo, + DisplayRound: status.RoundNo > 1, + Status: status.Status, + CurrentLevel: status.CurrentLevel, + CurrentEnergy: status.CurrentEnergy, + NeedEnergy: status.NeedEnergy, + EnergyPercent: percent, + DisplayPercent: displayPercent, + LaunchCountdownSeconds: countdownSeconds, + Shake: shake, + Levels: buildLevelPayloads(levels), + RewardPreview: rewardPreviewPayload(rewards), + RocketKings: kings, }, nil } @@ -111,6 +115,28 @@ func (s *Service) loadOrInitStatus(ctx context.Context, tx *gorm.DB, sysOrigin s return row, nil } +func (s *Service) launchCountdownSeconds(ctx context.Context, status model.VoiceRoomRocketStatus, now time.Time) int { + if status.Status != statusLaunching { + return 0 + } + var round model.VoiceRoomRocketLevelRound + err := s.repo.DB.WithContext(ctx). + Where( + "sys_origin = ? AND room_id = ? AND day_key = ? AND round_no = ? AND level = ?", + status.SysOrigin, status.RoomID, status.DayKey, status.RoundNo, status.CurrentLevel, + ). + First(&round).Error + if err != nil || round.ReachTime == nil || round.ReachTime.IsZero() { + return defaultLaunchDelaySeconds + } + due := round.ReachTime.Add(time.Duration(defaultLaunchDelaySeconds) * time.Second) + remaining := due.Sub(now).Seconds() + if remaining <= 0 { + return 0 + } + return int(math.Ceil(remaining)) +} + func (s *Service) ensureLevelRound(ctx context.Context, tx *gorm.DB, sysOrigin string, roomID int64, day string, roundNo int, level int, needEnergy int64, now time.Time) error { db := s.repo.DB.WithContext(ctx) if tx != nil { diff --git a/internal/service/voiceroomrocket/types.go b/internal/service/voiceroomrocket/types.go index a4654ee..2efdf38 100644 --- a/internal/service/voiceroomrocket/types.go +++ b/internal/service/voiceroomrocket/types.go @@ -28,6 +28,7 @@ const ( defaultBroadcastDurationSeconds = 7 defaultInRoomRewardDelaySeconds = 10 defaultInRoomRewardUserLimit = 3 + defaultLaunchDelaySeconds = 10 defaultWorkerBatchSize = 100 defaultWorkerInterval = time.Second defaultRoomLockTTL = 10 * time.Second @@ -36,6 +37,7 @@ const ( statusLaunching = "LAUNCHING" levelRoundStatusCharging = "CHARGING" + levelRoundStatusReached = "REACHED" levelRoundStatusLaunched = "LAUNCHED" eventStatusIgnored = "IGNORED" @@ -66,6 +68,7 @@ const ( rewardTypeVIPCard = "VIP_CARD" rewardTypeGift = "GIFT" rewardTypeBadge = "BADGE" + rewardTypeNone = "NONE" propsTypeNobleVIP = "NOBLE_VIP" propsTypeRide = "RIDE" @@ -101,6 +104,14 @@ type userProfileGateway interface { GetUserProfile(ctx context.Context, userID int64) (integration.UserProfile, error) } +type regionResolver interface { + ResolveRegionCodeByCountryCode(ctx context.Context, sysOrigin string, countryCode string) (string, error) +} + +type regionBroadcaster interface { + SendRegionCustomMessage(ctx context.Context, sysOrigin string, regionCode string, body any) (string, error) +} + type walletGateway interface { ChangeGoldBalance(ctx context.Context, cmd integration.GoldReceiptCommand) error ExistsGoldEvent(ctx context.Context, eventID string) (bool, error) @@ -128,16 +139,18 @@ type ports struct { // Service 承接语音房火箭配置、状态、送礼加能量、发射、榜单和奖励记录。 type Service struct { - cfg config.Config - repo ports - gateways gatewayBundle - im imGateway - roomProfiles roomProfileGateway - userProfiles userProfileGateway - wallet walletGateway - rewardGateway rewardGateway - location *time.Location - rocketConsumer rmq.PushConsumer + cfg config.Config + repo ports + gateways gatewayBundle + im imGateway + roomProfiles roomProfileGateway + userProfiles userProfileGateway + wallet walletGateway + rewardGateway rewardGateway + regionResolver regionResolver + regionBroadcast regionBroadcaster + location *time.Location + rocketConsumer rmq.PushConsumer } // NewService 创建语音房火箭服务。 @@ -162,6 +175,9 @@ func NewService(cfg config.Config, db voiceRoomRocketDB, cache redis.Cmdable, ga service.userProfiles = gateways service.wallet = gateways service.rewardGateway = gateways + if resolver, ok := any(gateways).(regionResolver); ok { + service.regionResolver = resolver + } } return service } @@ -173,6 +189,13 @@ func (s *Service) SetIMGateway(gateway imGateway) { } } +// SetRegionBroadcaster 允许装配区域 IM 广播服务。 +func (s *Service) SetRegionBroadcaster(broadcaster regionBroadcaster) { + if s != nil { + s.regionBroadcast = broadcaster + } +} + type flexibleInt64 int64 // FlexibleInt64 暴露给本地验证和路由测试，兼容 JSON 字符串形式的 Snowflake ID。 @@ -331,6 +354,7 @@ type ConfigResponse struct { RewardRecordKeepDays int `json:"rewardRecordKeepDays"` BroadcastDurationSeconds int `json:"broadcastDurationSeconds"` InRoomRewardDelaySeconds int `json:"inRoomRewardDelaySeconds"` + EnergyRuleJSON json.RawMessage `json:"energyRuleJson,omitempty"` RuleText json.RawMessage `json:"ruleText,omitempty"` Levels []LevelConfigPayload `json:"levels"` RewardPreview RewardPreviewResponse `json:"rewardPreview"` @@ -357,23 +381,26 @@ type SaveLevelConfigsRequest struct { type SaveRewardConfigsRequest struct { SysOrigin string `json:"sysOrigin"` + Level int `json:"level"` Rewards []RewardConfigPayload `json:"rewards"` } type StatusResponse struct { - RoomID int64 `json:"roomId,string"` - DayKey string `json:"dayKey"` - RoundNo int `json:"roundNo"` - DisplayRound bool `json:"displayRound"` - CurrentLevel int `json:"currentLevel"` - CurrentEnergy int64 `json:"currentEnergy"` - NeedEnergy int64 `json:"needEnergy"` - EnergyPercent string `json:"energyPercent"` - DisplayPercent int `json:"displayPercent"` - Shake bool `json:"shake"` - Levels []LevelConfigPayload `json:"levels"` - RewardPreview RewardPreviewResponse `json:"rewardPreview"` - RocketKings []KingRecord `json:"rocketKings"` + RoomID int64 `json:"roomId,string"` + DayKey string `json:"dayKey"` + RoundNo int `json:"roundNo"` + DisplayRound bool `json:"displayRound"` + Status string `json:"status"` + CurrentLevel int `json:"currentLevel"` + CurrentEnergy int64 `json:"currentEnergy"` + NeedEnergy int64 `json:"needEnergy"` + EnergyPercent string `json:"energyPercent"` + DisplayPercent int `json:"displayPercent"` + LaunchCountdownSeconds int `json:"launchCountdownSeconds,omitempty"` + Shake bool `json:"shake"` + Levels []LevelConfigPayload `json:"levels"` + RewardPreview RewardPreviewResponse `json:"rewardPreview"` + RocketKings []KingRecord `json:"rocketKings"` } type KingResponse struct {