fix: trim rtc reward streams

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
+	Enabled            bool
-	Timeout         time.Duration
+	Timeout            time.Duration
-	RewardStreamKey string
+	RewardStreamKey    string
-	Providers       []LuckyGiftProviderConfig
+	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),
+			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,
+			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"),
+			RewardStreamKey:    getEnvAny([]string{"CHATAPP_LUCKY_GIFT_REWARD_STREAM_KEY", "LUCKY_GIFT_REWARD_STREAM_KEY"}, "lucky-gift:reward"),
-			Providers:       loadLuckyGiftConfigs(),
+			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"),

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,

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)
+	}
+}

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

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)
+		_ = s.repo.Redis.XAdd(ctx, s.broadcastStreamArgs(map[string]any{
-		if streamKey == "" {
+			"type":   imTypeStatusUpdate,
-			streamKey = "voice_room:rocket:broadcast"
+			"roomId": strconv.FormatInt(roomID, 10),
-		}
+			"body":   string(body),
-		_ = s.repo.Redis.XAdd(ctx, &redis.XAddArgs{
+		})).Err()
-			Stream: streamKey,
-			Values: 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)
+		_ = s.repo.Redis.XAdd(ctx, s.broadcastStreamArgs(map[string]any{
-		if streamKey == "" {
+			"type":     imTypeLaunchBroadcast,
-			streamKey = "voice_room:rocket:broadcast"
+			"roomId":   strconv.FormatInt(launch.RoomID, 10),
-		}
+			"launchNo": launch.LaunchNo,
-		_ = s.repo.Redis.XAdd(ctx, &redis.XAddArgs{
+			"body":     string(payload),
-			Stream: streamKey,
+		})).Err()
-			Values: 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)
+	_ = s.repo.Redis.XAdd(ctx, s.broadcastStreamArgs(map[string]any{
-	if streamKey == "" {
+		"type":     body["type"],
-		streamKey = "voice_room:rocket:broadcast"
+		"roomId":   strconv.FormatInt(roomID, 10),
-	}
+		"launchNo": launchNo,
-	_ = s.repo.Redis.XAdd(ctx, &redis.XAddArgs{
+		"body":     string(payload),
-		Stream: streamKey,
+	})).Err()
-		Values: 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 {