hy/yumi-golang
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: complete voice room rocket launch rewards

Browse Source
This commit is contained in:
hy001 2026-05-14 20:02:43 +08:00
parent dd9020a542
commit 51b8b52c57
13 changed files with 684 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
cmd/api/main.go
View File

@ -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)

1
cmd/consumer/main.go
View File

@ -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()

60
internal/service/voiceroomrocket/config.go
View File

@ -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),

39
internal/service/voiceroomrocket/event.go
View File

@ -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 {

272
internal/service/voiceroomrocket/event_test.go
View File

@ -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()

21
internal/service/voiceroomrocket/in_room_reward.go
View File

@ -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 {

142
internal/service/voiceroomrocket/launch_worker.go Normal file
View File

@ -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"
}

28
internal/service/voiceroomrocket/lifecycle.go
View File

@ -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 {

11
internal/service/voiceroomrocket/mapper.go
View File

@ -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),

50
internal/service/voiceroomrocket/notify.go
View File

@ -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 {

3
internal/service/voiceroomrocket/reward.go
View File

@ -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

52
internal/service/voiceroomrocket/status.go
View File

@ -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 {

73
internal/service/voiceroomrocket/types.go
View File

@ -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 {