From 810dc0345bdd3d25a86cfb3049aac22f2abaaa33 Mon Sep 17 00:00:00 2001 From: hy001 Date: Thu, 21 May 2026 10:50:18 +0800 Subject: [PATCH] fix: trim rtc reward streams --- internal/config/config.go | 23 +++--- internal/service/luckygift/reward_event.go | 2 + .../service/luckygift/reward_event_test.go | 52 +++++++++++++- .../service/voiceroomrocket/event_test.go | 25 +++++++ internal/service/voiceroomrocket/notify.go | 71 +++++++++---------- 5 files changed, 125 insertions(+), 48 deletions(-) diff --git a/internal/config/config.go b/internal/config/config.go index f64bab7..01e3ec6 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -167,6 +167,7 @@ type VoiceRoomRocketConfig struct { InRoomRewardDueZSetKey string RoomLockTTLSeconds int BroadcastStreamKey string + BroadcastStreamMaxLen int64 InRoomRewardScanIntervalSeconds int InRoomRewardBatchSize int InRoomRewardUserLimit int @@ -229,10 +230,11 @@ type GameOpenConfig struct { // LuckyGiftConfig 保存幸运礼物模块配置。 type LuckyGiftConfig struct { - Enabled bool - Timeout time.Duration - RewardStreamKey string - Providers []LuckyGiftProviderConfig + Enabled bool + Timeout time.Duration + RewardStreamKey string + RewardStreamMaxLen int64 + Providers []LuckyGiftProviderConfig } // BinanceRechargeConfig 保存币安自动充值验证的运行配置。 @@ -433,6 +435,10 @@ func Load() Config { []string{"CHATAPP_VOICE_ROOM_ROCKET_BROADCAST_STREAM_KEY", "VOICE_ROOM_ROCKET_BROADCAST_STREAM_KEY"}, "voice_room:rocket:broadcast", ), + BroadcastStreamMaxLen: int64(getEnvIntAny( + []string{"CHATAPP_VOICE_ROOM_ROCKET_BROADCAST_STREAM_MAX_LEN", "VOICE_ROOM_ROCKET_BROADCAST_STREAM_MAX_LEN"}, + 10000, + )), InRoomRewardScanIntervalSeconds: getEnvIntAny( []string{"CHATAPP_VOICE_ROOM_ROCKET_IN_ROOM_SCAN_INTERVAL_SECONDS", "VOICE_ROOM_ROCKET_IN_ROOM_SCAN_INTERVAL_SECONDS"}, 1, @@ -515,10 +521,11 @@ func Load() Config { TestPassword: getEnvAny([]string{"CHATAPP_GAME_OPEN_TEST_PASSWORD", "GAME_OPEN_TEST_PASSWORD"}, "1234567"), }, LuckyGift: LuckyGiftConfig{ - Enabled: getEnvBoolAny([]string{"CHATAPP_LUCKY_GIFT_ENABLED", "LUCKY_GIFT_ENABLED", "LUCKY_GIFT_GO_ENABLED"}, false), - Timeout: time.Duration(getEnvIntAny([]string{"CHATAPP_LUCKY_GIFT_TIMEOUT_MILLIS", "LUCKY_GIFT_TIMEOUT_MILLIS", "LUCKY_GIFT_GO_TIMEOUT_MILLIS"}, 8000)) * time.Millisecond, - RewardStreamKey: getEnvAny([]string{"CHATAPP_LUCKY_GIFT_REWARD_STREAM_KEY", "LUCKY_GIFT_REWARD_STREAM_KEY"}, "lucky-gift:reward"), - Providers: loadLuckyGiftConfigs(), + Enabled: getEnvBoolAny([]string{"CHATAPP_LUCKY_GIFT_ENABLED", "LUCKY_GIFT_ENABLED", "LUCKY_GIFT_GO_ENABLED"}, false), + Timeout: time.Duration(getEnvIntAny([]string{"CHATAPP_LUCKY_GIFT_TIMEOUT_MILLIS", "LUCKY_GIFT_TIMEOUT_MILLIS", "LUCKY_GIFT_GO_TIMEOUT_MILLIS"}, 8000)) * time.Millisecond, + RewardStreamKey: getEnvAny([]string{"CHATAPP_LUCKY_GIFT_REWARD_STREAM_KEY", "LUCKY_GIFT_REWARD_STREAM_KEY"}, "lucky-gift:reward"), + RewardStreamMaxLen: int64(getEnvIntAny([]string{"CHATAPP_LUCKY_GIFT_REWARD_STREAM_MAX_LEN", "LUCKY_GIFT_REWARD_STREAM_MAX_LEN"}, 100000)), + Providers: loadLuckyGiftConfigs(), }, BinanceRecharge: BinanceRechargeConfig{ BaseURL: getEnvAny([]string{"CHATAPP_BINANCE_RECHARGE_BASE_URL", "BINANCE_RECHARGE_BASE_URL"}, "https://api.binance.com"), diff --git a/internal/service/luckygift/reward_event.go b/internal/service/luckygift/reward_event.go index 6440dbf..e855f2a 100644 --- a/internal/service/luckygift/reward_event.go +++ b/internal/service/luckygift/reward_event.go @@ -78,6 +78,8 @@ func (s *LuckyGiftService) publishRewardEvent( _, err := s.repo.Redis.XAdd(ctx, &redis.XAddArgs{ Stream: streamKey, + MaxLen: s.cfg.LuckyGift.RewardStreamMaxLen, + Approx: s.cfg.LuckyGift.RewardStreamMaxLen > 0, Values: map[string]any{ "eventId": event.EventID, "businessId": event.BusinessID, diff --git a/internal/service/luckygift/reward_event_test.go b/internal/service/luckygift/reward_event_test.go index cf011a0..e03a2e7 100644 --- a/internal/service/luckygift/reward_event_test.go +++ b/internal/service/luckygift/reward_event_test.go @@ -4,6 +4,7 @@ import ( "chatapp3-golang/internal/config" "context" "encoding/json" + "strconv" "testing" "github.com/alicebob/miniredis/v2" @@ -23,7 +24,8 @@ func TestPublishRewardEventWritesWinningResultsToStream(t *testing.T) { service := &LuckyGiftService{ cfg: config.Config{ LuckyGift: config.LuckyGiftConfig{ - RewardStreamKey: "lucky-gift:reward", + RewardStreamKey: "lucky-gift:reward", + RewardStreamMaxLen: 2, }, }, repo: luckyGiftPorts{Redis: client}, @@ -89,7 +91,8 @@ func TestPublishRewardEventSkipsZeroRewardTotal(t *testing.T) { service := &LuckyGiftService{ cfg: config.Config{ LuckyGift: config.LuckyGiftConfig{ - RewardStreamKey: "lucky-gift:reward", + RewardStreamKey: "lucky-gift:reward", + RewardStreamMaxLen: 2, }, }, repo: luckyGiftPorts{Redis: client}, @@ -107,3 +110,48 @@ func TestPublishRewardEventSkipsZeroRewardTotal(t *testing.T) { t.Fatalf("expected no stream records for zero reward total, got %d", len(streams)) } } + +func TestPublishRewardEventTrimsStreamByConfiguredMaxLen(t *testing.T) { + mr, err := miniredis.Run() + if err != nil { + t.Fatalf("miniredis run failed: %v", err) + } + defer mr.Close() + + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + defer client.Close() + + service := &LuckyGiftService{ + cfg: config.Config{ + LuckyGift: config.LuckyGiftConfig{ + RewardStreamKey: "lucky-gift:reward", + RewardStreamMaxLen: 2, + }, + }, + repo: luckyGiftPorts{Redis: client}, + } + + for idx := 1; idx <= 3; idx++ { + req := LuckyGiftDrawRequest{ + BusinessID: "biz-" + strconv.Itoa(idx), + SysOrigin: "LIKEI", + RoomID: 2001, + SendUserID: 3001, + GiftID: 4001, + GiftPrice: 500, + GiftNum: 2, + } + results := []LuckyGiftDrawResult{{ID: "order-a", AcceptUserID: 9001, RewardNum: 88}} + if err := service.publishRewardEvent(context.Background(), req, results, 88); err != nil { + t.Fatalf("publishRewardEvent returned error: %v", err) + } + } + + count, err := client.XLen(context.Background(), "lucky-gift:reward").Result() + if err != nil { + t.Fatalf("xlen returned error: %v", err) + } + if count != 2 { + t.Fatalf("expected trimmed stream length 2, got %d", count) + } +} diff --git a/internal/service/voiceroomrocket/event_test.go b/internal/service/voiceroomrocket/event_test.go index 471f68e..403c3da 100644 --- a/internal/service/voiceroomrocket/event_test.go +++ b/internal/service/voiceroomrocket/event_test.go @@ -816,6 +816,29 @@ func TestNotifyRewardRecordPublishesIMAndRedisStream(t *testing.T) { } } +func TestPublishRewardMessageTrimsBroadcastStreamByConfiguredMaxLen(t *testing.T) { + service, _, rdb, mr := newTestServiceWithRedis(t, newFakeRocketGateway()) + defer mr.Close() + service.cfg.VoiceRoomRocket.BroadcastStreamMaxLen = 2 + + for idx := 1; idx <= 3; idx++ { + service.publishRewardMessage(context.Background(), 9001, "launch-"+strconv.Itoa(idx), map[string]any{ + "type": imTypeRewardPopup, + "data": map[string]any{ + "roomId": "9001", + }, + }) + } + + count, err := rdb.XLen(context.Background(), "voice_room:rocket:broadcast").Result() + if err != nil { + t.Fatalf("xlen broadcast: %v", err) + } + if count != 2 { + t.Fatalf("broadcast stream count = %d, want 2", count) + } +} + func TestStatusUpdatePublishesRegionIM(t *testing.T) { gateway := newFakeRocketGateway() service, db, _, mr := newTestServiceWithRedis(t, gateway) @@ -1010,6 +1033,8 @@ func newTestServiceWithRedis(t *testing.T, gateway *fakeRocketGateway) (*Service service.userProfiles = gateway service.wallet = gateway service.rewardGateway = gateway + service.cfg.VoiceRoomRocket.BroadcastStreamKey = "voice_room:rocket:broadcast" + service.cfg.VoiceRoomRocket.BroadcastStreamMaxLen = 100 service.cfg.VoiceRoomRocket.InRoomRewardUserLimit = 3 service.cfg.VoiceRoomRocket.InRoomRewardDueZSetKey = "voice_room:rocket:in_room_due" return service, db, rdb, mr diff --git a/internal/service/voiceroomrocket/notify.go b/internal/service/voiceroomrocket/notify.go index 51d7cdd..d4e2c72 100644 --- a/internal/service/voiceroomrocket/notify.go +++ b/internal/service/voiceroomrocket/notify.go @@ -13,6 +13,22 @@ import ( "gorm.io/gorm" ) +func (s *Service) broadcastStreamArgs(values map[string]any) *redis.XAddArgs { + streamKey := strings.TrimSpace(s.cfg.VoiceRoomRocket.BroadcastStreamKey) + if streamKey == "" { + streamKey = "voice_room:rocket:broadcast" + } + args := &redis.XAddArgs{ + Stream: streamKey, + Values: values, + } + if s.cfg.VoiceRoomRocket.BroadcastStreamMaxLen > 0 { + args.MaxLen = s.cfg.VoiceRoomRocket.BroadcastStreamMaxLen + args.Approx = true + } + return args +} + func (s *Service) notifyStatusUpdate(ctx context.Context, sysOrigin string, roomID int64, senderUserIDs ...int64) error { data := map[string]any{ "roomId": strconv.FormatInt(roomID, 10), @@ -35,18 +51,11 @@ func (s *Service) notifyStatusUpdate(ctx context.Context, sysOrigin string, room "roomId": strconv.FormatInt(roomID, 10), "data": data, }) - streamKey := strings.TrimSpace(s.cfg.VoiceRoomRocket.BroadcastStreamKey) - if streamKey == "" { - streamKey = "voice_room:rocket:broadcast" - } - _ = s.repo.Redis.XAdd(ctx, &redis.XAddArgs{ - Stream: streamKey, - Values: map[string]any{ - "type": imTypeStatusUpdate, - "roomId": strconv.FormatInt(roomID, 10), - "body": string(body), - }, - }).Err() + _ = s.repo.Redis.XAdd(ctx, s.broadcastStreamArgs(map[string]any{ + "type": imTypeStatusUpdate, + "roomId": strconv.FormatInt(roomID, 10), + "body": string(body), + })).Err() } _ = s.notifyRegionStatusUpdate(ctx, sysOrigin, senderUserID, data) if s.im == nil { @@ -147,19 +156,12 @@ func (s *Service) notifyLaunch(ctx context.Context, launch model.VoiceRoomRocket } if s.repo.Redis != nil { payload, _ := json.Marshal(body) - streamKey := strings.TrimSpace(s.cfg.VoiceRoomRocket.BroadcastStreamKey) - if streamKey == "" { - streamKey = "voice_room:rocket:broadcast" - } - _ = s.repo.Redis.XAdd(ctx, &redis.XAddArgs{ - Stream: streamKey, - Values: map[string]any{ - "type": imTypeLaunchBroadcast, - "roomId": strconv.FormatInt(launch.RoomID, 10), - "launchNo": launch.LaunchNo, - "body": string(payload), - }, - }).Err() + _ = s.repo.Redis.XAdd(ctx, s.broadcastStreamArgs(map[string]any{ + "type": imTypeLaunchBroadcast, + "roomId": strconv.FormatInt(launch.RoomID, 10), + "launchNo": launch.LaunchNo, + "body": string(payload), + })).Err() } if s.im == nil { return nil @@ -241,19 +243,12 @@ func (s *Service) publishRewardMessage(ctx context.Context, roomID int64, launch return } payload, _ := json.Marshal(body) - streamKey := strings.TrimSpace(s.cfg.VoiceRoomRocket.BroadcastStreamKey) - if streamKey == "" { - streamKey = "voice_room:rocket:broadcast" - } - _ = s.repo.Redis.XAdd(ctx, &redis.XAddArgs{ - Stream: streamKey, - Values: map[string]any{ - "type": body["type"], - "roomId": strconv.FormatInt(roomID, 10), - "launchNo": launchNo, - "body": string(payload), - }, - }).Err() + _ = s.repo.Redis.XAdd(ctx, s.broadcastStreamArgs(map[string]any{ + "type": body["type"], + "roomId": strconv.FormatInt(roomID, 10), + "launchNo": launchNo, + "body": string(payload), + })).Err() } func (s *Service) resolveRoomAccount(ctx context.Context, roomID int64) string {