diff --git a/README.md b/README.md index d39e3ba..b23180c 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ internal/utils ### 1. `cmd/api` - 只做启动和依赖装配。 - 不承担业务逻辑。 -- 当前线上只部署 `golang` API 进程,所以这里会同时启动 `registerreward` 的 Redis Stream consumer,确保注册成功事件能被消费并发奖。 +- 当前线上只部署 `golang` API 进程,所以这里会同时启动 `registerreward` 的 Redis Stream consumer 和 `weekstar` 的 RocketMQ consumer,确保注册成功与周星送礼事件能被消费。 - 不启动任何 `weekstar` 结算 goroutine。 ### 2. `cmd/consumer` diff --git a/cmd/api/main.go b/cmd/api/main.go index 8adf7cb..22e52ad 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -56,6 +56,9 @@ func main() { if err := registerRewardService.Start(workerCtx); err != nil { log.Fatalf("start register reward consumer failed: %v", err) } + if err := weekStarService.StartMessageConsumer(workerCtx); err != nil { + log.Fatalf("start week star message consumer failed: %v", err) + } gameProviders := gameprovider.NewRegistry( baishun.NewAppProvider(baishunService), diff --git a/internal/service/weekstar/event.go b/internal/service/weekstar/event.go index 67c3682..0fd4cb3 100644 --- a/internal/service/weekstar/event.go +++ b/internal/service/weekstar/event.go @@ -14,9 +14,6 @@ import ( // ProcessGiftEvent 按送礼事件发生时间归属配置,并把积分累加到对应周期实时榜。 func (s *WeekStarService) ProcessGiftEvent(ctx context.Context, event weekStarGiftEvent, rawPayload string) error { sysOrigin := s.normalizeSysOrigin(event.SysOrigin) - if sysOrigin != s.normalizeSysOrigin(s.cfg.WeekStar.DefaultSysOrigin) { - return nil - } trackID := strings.TrimSpace(event.TrackID) if trackID == "" || int64(event.SendUserID) <= 0 { return nil diff --git a/internal/service/weekstar/page.go b/internal/service/weekstar/page.go index 7339e96..ab5b573 100644 --- a/internal/service/weekstar/page.go +++ b/internal/service/weekstar/page.go @@ -303,7 +303,7 @@ func (s *WeekStarService) listSnapshotRanking(ctx context.Context, configID int6 return result, nil } -// loadLiveRanking 从 Redis 实时榜读取前 N 名,并补齐用户资料用于 H5 和结算展示。 +// loadLiveRanking 读取实时榜前 N 名,并补齐用户资料用于 H5 和结算展示。 func (s *WeekStarService) loadLiveRanking(ctx context.Context, configID int64, cycleKey string, limit int64) ([]WeekStarRankingUserView, error) { if limit <= 0 { return []WeekStarRankingUserView{}, nil @@ -312,9 +312,6 @@ func (s *WeekStarService) loadLiveRanking(ctx context.Context, configID int64, c if err != nil && !errors.Is(err, redis.Nil) { return nil, err } - if len(zs) == 0 { - return []WeekStarRankingUserView{}, nil - } userIDs := make([]int64, 0, len(zs)) rankScores := make([]weekStarRankScore, 0, len(zs)) @@ -330,6 +327,16 @@ func (s *WeekStarService) loadLiveRanking(ctx context.Context, configID int64, c Rank: index + 1, }) } + if len(rankScores) == 0 { + rankScores, err = s.loadLiveRankingFromGiftLogs(ctx, configID, cycleKey, limit) + if err != nil { + return nil, err + } + userIDs = make([]int64, 0, len(rankScores)) + for _, item := range rankScores { + userIDs = append(userIDs, item.UserID) + } + } profileMap, err := s.lookupUserProfiles(ctx, userIDs) if err != nil { return nil, err @@ -352,6 +359,30 @@ func (s *WeekStarService) loadLiveRanking(ctx context.Context, configID int64, c return result, nil } +func (s *WeekStarService) loadLiveRankingFromGiftLogs(ctx context.Context, configID int64, cycleKey string, limit int64) ([]weekStarRankScore, error) { + if configID <= 0 || strings.TrimSpace(cycleKey) == "" || limit <= 0 { + return []weekStarRankScore{}, nil + } + + var rows []weekStarRankScore + if err := s.repo.DB.WithContext(ctx). + Model(&model.WeekStarGiftLog{}). + Select("send_user_id AS user_id, SUM(score_gold) AS score_gold"). + Where("config_id = ? AND cycle_key = ? AND status = ?", configID, cycleKey, weekStarGiftLogStatusSuccess). + Group("send_user_id"). + Having("SUM(score_gold) > 0"). + Order("SUM(score_gold) DESC"). + Order("send_user_id ASC"). + Limit(int(limit)). + Scan(&rows).Error; err != nil { + return nil, err + } + for index := range rows { + rows[index].Rank = index + 1 + } + return rows, nil +} + // lookupUserProfiles 优先查本地资料,查不到时回源 Java 用户资料接口兜底。 func (s *WeekStarService) lookupUserProfiles(ctx context.Context, userIDs []int64) (map[int64]weekStarUserProfile, error) { result := make(map[int64]weekStarUserProfile, len(userIDs)) diff --git a/internal/service/weekstar/page_test.go b/internal/service/weekstar/page_test.go index 6e5aa28..67a1b87 100644 --- a/internal/service/weekstar/page_test.go +++ b/internal/service/weekstar/page_test.go @@ -86,6 +86,145 @@ func TestCurrentRankingReturnsLiveRank(t *testing.T) { } } +func TestCurrentRankingFallsBackToGiftLogsWhenRedisEmpty(t *testing.T) { + ctx := context.Background() + service, db := newTestWeekStarService(t) + if err := db.AutoMigrate(&model.UserBaseInfo{}, &model.WeekStarGiftLog{}); err != nil { + t.Fatalf("auto migrate ranking dependencies: %v", err) + } + redisServer := miniredis.RunT(t) + redisClient := redis.NewClient(&redis.Options{Addr: redisServer.Addr()}) + defer redisClient.Close() + service.repo.Redis = redisClient + + now := time.Now().In(service.location) + cycleStart, cycleEnd := service.cycleBounds(now) + cycleKey := service.cycleKey(cycleStart) + configRow := model.WeekStarActivityConfig{ + ID: 102, + SysOrigin: "LIKEI", + Enabled: true, + StartAt: service.toStorageWallClock(cycleStart.Add(-time.Hour)), + EndAt: service.toStorageWallClock(cycleEnd.Add(time.Hour)), + Timezone: "Asia/Riyadh", + CreateTime: now, + UpdateTime: now, + } + if err := db.Create(&configRow).Error; err != nil { + t.Fatalf("create config: %v", err) + } + if err := db.Create(&model.WeekStarActivityGiftConfig{ + ConfigID: configRow.ID, + GiftID: 2001, + GiftName: "Gift A", + GiftCandy: 100, + Sort: 1, + }).Error; err != nil { + t.Fatalf("create gift config: %v", err) + } + if err := db.Create([]model.UserBaseInfo{ + {ID: 3001, Account: "3001", UserNickname: "Alice", CountryCode: "SA", CountryName: "Saudi Arabia"}, + {ID: 3002, Account: "3002", UserNickname: "Bob", CountryCode: "AE", CountryName: "UAE"}, + }).Error; err != nil { + t.Fatalf("create users: %v", err) + } + if err := db.Create([]model.WeekStarGiftLog{ + {ID: 1, ConfigID: configRow.ID, CycleKey: cycleKey, TrackID: "track-1", SendUserID: 3001, GiftID: 2001, Quantity: 1, AcceptUserSize: 1, ScoreGold: 250, EventTime: now, Status: weekStarGiftLogStatusSuccess, CreateTime: now, UpdateTime: now}, + {ID: 2, ConfigID: configRow.ID, CycleKey: cycleKey, TrackID: "track-2", SendUserID: 3002, GiftID: 2001, Quantity: 1, AcceptUserSize: 1, ScoreGold: 400, EventTime: now, Status: weekStarGiftLogStatusSuccess, CreateTime: now, UpdateTime: now}, + {ID: 3, ConfigID: configRow.ID, CycleKey: cycleKey, TrackID: "track-3", SendUserID: 3002, GiftID: 2001, Quantity: 1, AcceptUserSize: 1, ScoreGold: 100, EventTime: now, Status: weekStarGiftLogStatusSuccess, CreateTime: now, UpdateTime: now}, + {ID: 4, ConfigID: configRow.ID, CycleKey: cycleKey, TrackID: "track-4", SendUserID: 3001, GiftID: 2001, Quantity: 1, AcceptUserSize: 1, ScoreGold: 999, EventTime: now, Status: weekStarGiftLogStatusFailed, CreateTime: now, UpdateTime: now}, + }).Error; err != nil { + t.Fatalf("create gift logs: %v", err) + } + + resp, err := service.CurrentRanking(ctx, "LIKEI", 20) + if err != nil { + t.Fatalf("CurrentRanking() error = %v", err) + } + if len(resp.Records) != 2 { + t.Fatalf("Records length = %d, want 2", len(resp.Records)) + } + if resp.Records[0].UserID != 3002 || resp.Records[0].Rank != 1 || resp.Records[0].ScoreGold != 500 { + t.Fatalf("top record = %#v", resp.Records[0]) + } + if resp.Records[1].UserID != 3001 || resp.Records[1].Rank != 2 || resp.Records[1].ScoreGold != 250 { + t.Fatalf("second record = %#v", resp.Records[1]) + } +} + +func TestProcessGiftEventCountsConfiguredNonDefaultSysOrigin(t *testing.T) { + ctx := context.Background() + service, db := newTestWeekStarService(t) + if err := db.AutoMigrate(&model.WeekStarGiftLog{}); err != nil { + t.Fatalf("auto migrate gift log: %v", err) + } + redisServer := miniredis.RunT(t) + redisClient := redis.NewClient(&redis.Options{Addr: redisServer.Addr()}) + defer redisClient.Close() + service.repo.Redis = redisClient + + now := time.Now().In(service.location) + cycleStart, cycleEnd := service.cycleBounds(now) + configRow := model.WeekStarActivityConfig{ + ID: 103, + SysOrigin: "TARAB", + Enabled: true, + StartAt: service.toStorageWallClock(cycleStart.Add(-time.Hour)), + EndAt: service.toStorageWallClock(cycleEnd.Add(time.Hour)), + Timezone: "Asia/Riyadh", + CreateTime: now, + UpdateTime: now, + } + if err := db.Create(&configRow).Error; err != nil { + t.Fatalf("create config: %v", err) + } + if err := db.Create(&model.WeekStarActivityGiftConfig{ + ConfigID: configRow.ID, + GiftID: 2001, + GiftName: "Gift A", + GiftCandy: 100, + Sort: 1, + }).Error; err != nil { + t.Fatalf("create gift config: %v", err) + } + + err := service.ProcessGiftEvent(ctx, weekStarGiftEvent{ + TrackID: "tarab-track-1", + SysOrigin: "TARAB", + SendUserID: flexibleInt64(3001), + Quantity: flexibleInt64(2), + CreateTime: flexibleInt64(now.UnixMilli()), + Accepts: []weekStarGiftEventUser{ + {AcceptUserID: flexibleInt64(4001)}, + }, + GiftConfig: weekStarGiftEventGift{ + ID: flexibleInt64(2001), + GiftCandy: flexibleInt64(100), + GiftName: "Gift A", + }, + }, `{"trackId":"tarab-track-1"}`) + if err != nil { + t.Fatalf("ProcessGiftEvent() error = %v", err) + } + + score, err := redisClient.ZScore(ctx, service.rankRedisKey(configRow.ID, service.cycleKey(cycleStart)), "3001").Result() + if err != nil { + t.Fatalf("ZScore() error = %v", err) + } + if score != 200 { + t.Fatalf("score = %v, want 200", score) + } + var count int64 + if err := db.Model(&model.WeekStarGiftLog{}). + Where("config_id = ? AND cycle_key = ? AND send_user_id = ? AND status = ?", configRow.ID, service.cycleKey(cycleStart), 3001, weekStarGiftLogStatusSuccess). + Count(&count).Error; err != nil { + t.Fatalf("count gift log: %v", err) + } + if count != 1 { + t.Fatalf("success gift log count = %d, want 1", count) + } +} + func TestCurrentRankingReturnsDisabledWhenNoActiveConfig(t *testing.T) { service, _ := newTestWeekStarService(t) diff --git a/internal/service/weekstar/types.go b/internal/service/weekstar/types.go index 5696017..ace9e49 100644 --- a/internal/service/weekstar/types.go +++ b/internal/service/weekstar/types.go @@ -264,7 +264,7 @@ type weekStarUserProfile struct { CountryName string } -// weekStarRankScore 保存 Redis 榜单归一化后的名次和分值。 +// weekStarRankScore 保存榜单归一化后的名次和分值。 type weekStarRankScore struct { UserID int64 ScoreGold int64