修复周星

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`

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

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

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

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)
 

internal/service/weekstar/types.go

@@ -264,7 +264,7 @@ type weekStarUserProfile struct {
 	CountryName  string
 }
 
-// weekStarRankScore 保存 Redis 榜单归一化后的名次和分值。
+// weekStarRankScore 保存榜单归一化后的名次和分值。
 type weekStarRankScore struct {
 	UserID    int64
 	ScoreGold int64