增加redis聚合

docker-compose.yml

@@ -151,6 +151,8 @@ services:
     depends_on:
       mysql:
         condition: service_healthy
+      redis:
+        condition: service_healthy
       rocketmq-broker:
         condition: service_started
       user-service:

pkg/rocketmqx/rocketmqx.go

@@ -54,6 +54,8 @@ type ConsumerConfig struct {
 	ConsumePullBatch   int32
 	ConsumePullTimeout time.Duration
 	ConsumeGoroutines  int
+	// ConsumeFromFirst 仅给可重建读模型使用；普通业务 consumer 保持客户端默认的 last offset，避免新组误重放历史副作用。
+	ConsumeFromFirst bool
 }
 
 // Message is the stable publish shape used by services.
@@ -209,6 +211,9 @@ func NewConsumer(cfg ConsumerConfig) (*Consumer, error) {
 	if cfg.ConsumeGoroutines > 0 {
 		options = append(options, consumer.WithConsumeGoroutineNums(cfg.ConsumeGoroutines))
 	}
+	if cfg.ConsumeFromFirst {
+		options = append(options, consumer.WithConsumeFromWhere(consumer.ConsumeFromFirstOffset))
+	}
 	client, err := rocketmq.NewPushConsumer(options...)
 	if err != nil {
 		return nil, err

pkg/userleaderboard/store.go

@@ -0,0 +1,416 @@
+// Package userleaderboard owns the Redis read model for App user gift rankings.
+package userleaderboard
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+	"hyapp/pkg/appcode"
+)
+
+const (
+	BoardSent     = "sent"
+	BoardReceived = "received"
+
+	PeriodToday = "today"
+	PeriodWeek  = "week"
+	PeriodMonth = "month"
+
+	defaultKeyPrefix       = "activity:user_leaderboard"
+	defaultPageSize        = 20
+	maxPageSize            = 100
+	defaultEventDedupeTTL  = 90 * 24 * time.Hour
+	dailyLeaderboardKeep   = 7 * 24 * time.Hour
+	weeklyLeaderboardKeep  = 21 * 24 * time.Hour
+	monthlyLeaderboardKeep = 35 * 24 * time.Hour
+)
+
+var ErrNotConfigured = errors.New("user leaderboard redis is not configured")
+
+// Store keeps sent/received gift leaderboard buckets in Redis zsets plus per-user hashes.
+// The write side consumes wallet committed facts; the read side is used by gateway only.
+type Store struct {
+	client    *redis.Client
+	keyPrefix string
+}
+
+// GiftEvent is the minimal WalletGiftDebited fact needed to update the user leaderboard.
+type GiftEvent struct {
+	AppCode      string
+	EventID      string
+	SenderUserID int64
+	TargetUserID int64
+	GiftValue    int64
+	GiftCount    int64
+	OccurredAtMS int64
+	DirectGift   bool
+}
+
+// Query describes one App leaderboard page. Now must be UTC-compatible; zero means current UTC time.
+type Query struct {
+	AppCode   string
+	BoardType string
+	Period    string
+	Page      int
+	PageSize  int
+	Now       time.Time
+}
+
+// Page is the Redis aggregate result returned to gateway before profile enrichment.
+type Page struct {
+	Items        []Entry
+	Total        int64
+	BoardType    string
+	Period       string
+	StartAtMS    int64
+	EndAtMS      int64
+	ServerTimeMS int64
+}
+
+// Entry contains only aggregate facts. gateway enriches user profile data separately.
+type Entry struct {
+	Rank             int64
+	UserID           string
+	GiftValue        int64
+	GiftCount        int64
+	TransactionCount int64
+	LastGiftAtMS     int64
+}
+
+// NewRedisClient creates and verifies the Redis dependency used by the leaderboard read model.
+func NewRedisClient(ctx context.Context, addr string, password string, db int) (*redis.Client, error) {
+	client := redis.NewClient(&redis.Options{
+		Addr:     strings.TrimSpace(addr),
+		Password: password,
+		DB:       db,
+	})
+	// 榜单 Redis 是 gateway 查询的唯一读模型；启动时探测失败比运行中慢查钱包库更容易定位。
+	if err := client.Ping(ctx).Err(); err != nil {
+		_ = client.Close()
+		return nil, err
+	}
+	return client, nil
+}
+
+// NewStore binds a verified Redis client to a stable key namespace.
+func NewStore(client *redis.Client, keyPrefix string) *Store {
+	keyPrefix = strings.Trim(strings.TrimSpace(keyPrefix), ":")
+	if keyPrefix == "" {
+		keyPrefix = defaultKeyPrefix
+	}
+	return &Store{client: client, keyPrefix: keyPrefix}
+}
+
+// ApplyGiftEvent increments sent and received buckets for day/week/month.
+// It returns false when the fact is outside the old SQL leaderboard scope, such as direct gifts.
+func (s *Store) ApplyGiftEvent(ctx context.Context, event GiftEvent) (bool, error) {
+	if s == nil || s.client == nil {
+		return false, ErrNotConfigured
+	}
+	event = normalizeGiftEvent(event)
+	if event.AppCode == "" || event.EventID == "" || event.SenderUserID <= 0 || event.TargetUserID <= 0 || event.GiftValue <= 0 || event.DirectGift {
+		return false, nil
+	}
+
+	occurredAt := time.UnixMilli(event.OccurredAtMS).UTC()
+	keys := []string{s.dedupeKey(event.AppCode, event.EventID)}
+	args := []any{
+		strconv.FormatInt(int64(defaultEventDedupeTTL/time.Millisecond), 10),
+		strconv.FormatInt(event.GiftValue, 10),
+		strconv.FormatInt(event.GiftCount, 10),
+		strconv.FormatInt(event.OccurredAtMS, 10),
+	}
+
+	updates := make([]leaderboardUpdate, 0, 6)
+	for _, period := range []string{PeriodToday, PeriodWeek, PeriodMonth} {
+		updates = append(updates,
+			s.updateFor(event.AppCode, BoardSent, period, event.SenderUserID, occurredAt),
+			s.updateFor(event.AppCode, BoardReceived, period, event.TargetUserID, occurredAt),
+		)
+	}
+	args = append(args, strconv.Itoa(len(updates)))
+	for _, update := range updates {
+		keys = append(keys, update.scoreKey, update.itemKey)
+		args = append(args, update.member, strconv.FormatInt(int64(update.ttl/time.Second), 10))
+	}
+
+	result, err := applyGiftEventScript.Run(ctx, s.client, keys, args...).Int64()
+	if err != nil {
+		return false, err
+	}
+	return result == 1, nil
+}
+
+// List reads a leaderboard page from the current Redis bucket. Empty keys are valid empty rankings.
+func (s *Store) List(ctx context.Context, query Query) (Page, error) {
+	if s == nil || s.client == nil {
+		return Page{}, ErrNotConfigured
+	}
+	query = normalizeQuery(query)
+	start, end := QueryWindow(query.Period, query.Now)
+	key := s.scoreKey(query.AppCode, query.BoardType, query.Period, start)
+	offset := int64((query.Page - 1) * query.PageSize)
+	stop := offset + int64(query.PageSize) - 1
+
+	total, err := s.client.ZCard(ctx, key).Result()
+	if err != nil {
+		return Page{}, err
+	}
+	rows, err := s.client.ZRevRangeWithScores(ctx, key, offset, stop).Result()
+	if err != nil {
+		return Page{}, err
+	}
+
+	items := make([]Entry, 0, len(rows))
+	for index, row := range rows {
+		userID, ok := row.Member.(string)
+		if !ok || strings.TrimSpace(userID) == "" {
+			continue
+		}
+		entry, err := s.entryFromHash(ctx, query.AppCode, query.BoardType, query.Period, start, userID, int64(math.Round(row.Score)))
+		if err != nil {
+			return Page{}, err
+		}
+		entry.Rank = offset + int64(index) + 1
+		items = append(items, entry)
+	}
+	return Page{
+		Items:        items,
+		Total:        total,
+		BoardType:    query.BoardType,
+		Period:       query.Period,
+		StartAtMS:    start.UnixMilli(),
+		EndAtMS:      end.UnixMilli(),
+		ServerTimeMS: query.Now.UTC().UnixMilli(),
+	}, nil
+}
+
+// RankForUser returns the caller's current aggregate rank for the same bucket used by List.
+func (s *Store) RankForUser(ctx context.Context, query Query, userID int64) (Entry, bool, error) {
+	if s == nil || s.client == nil {
+		return Entry{}, false, ErrNotConfigured
+	}
+	if userID <= 0 {
+		return Entry{}, false, nil
+	}
+	query = normalizeQuery(query)
+	start, _ := QueryWindow(query.Period, query.Now)
+	key := s.scoreKey(query.AppCode, query.BoardType, query.Period, start)
+	member := strconv.FormatInt(userID, 10)
+	rank, err := s.client.ZRevRank(ctx, key, member).Result()
+	if errors.Is(err, redis.Nil) {
+		return Entry{}, false, nil
+	}
+	if err != nil {
+		return Entry{}, false, err
+	}
+	entry, err := s.entryFromHash(ctx, query.AppCode, query.BoardType, query.Period, start, member, 0)
+	if err != nil {
+		return Entry{}, false, err
+	}
+	entry.Rank = rank + 1
+	return entry, true, nil
+}
+
+// NormalizeBoardType keeps gateway query aliases out of the Redis key layer.
+func NormalizeBoardType(raw string) string {
+	switch strings.ToLower(strings.TrimSpace(raw)) {
+	case "", BoardSent, "send", "sender", "gift_sent", "user_sent":
+		return BoardSent
+	case BoardReceived, "receive", "receiver", "gift_received", "user_received":
+		return BoardReceived
+	default:
+		return ""
+	}
+}
+
+// NormalizePeriod maps public aliases to the fixed UTC buckets used by activity and gateway.
+func NormalizePeriod(raw string) string {
+	switch strings.ToLower(strings.TrimSpace(raw)) {
+	case "", PeriodToday, "day", "daily":
+		return PeriodToday
+	case PeriodWeek, "weekly":
+		return PeriodWeek
+	case PeriodMonth, "monthly":
+		return PeriodMonth
+	default:
+		return ""
+	}
+}
+
+// QueryWindow preserves the existing gateway response contract: start is the UTC bucket boundary, end is request time.
+func QueryWindow(period string, now time.Time) (time.Time, time.Time) {
+	now = normalizedNow(now)
+	start, _ := periodBounds(period, now)
+	return start, now
+}
+
+type leaderboardUpdate struct {
+	scoreKey string
+	itemKey  string
+	member   string
+	ttl      time.Duration
+}
+
+func (s *Store) updateFor(app string, board string, period string, userID int64, occurredAt time.Time) leaderboardUpdate {
+	start, periodEnd := periodBounds(period, occurredAt)
+	member := strconv.FormatInt(userID, 10)
+	return leaderboardUpdate{
+		scoreKey: s.scoreKey(app, board, period, start),
+		itemKey:  s.itemKey(app, board, period, start, member),
+		member:   member,
+		ttl:      retentionTTL(period, start, periodEnd),
+	}
+}
+
+func (s *Store) entryFromHash(ctx context.Context, app string, board string, period string, start time.Time, userID string, scoreFallback int64) (Entry, error) {
+	values, err := s.client.HGetAll(ctx, s.itemKey(app, board, period, start, userID)).Result()
+	if err != nil {
+		return Entry{}, err
+	}
+	entry := Entry{
+		UserID:    userID,
+		GiftValue: scoreFallback,
+	}
+	if value := int64FromHash(values, "gift_value"); value > 0 {
+		entry.GiftValue = value
+	}
+	entry.GiftCount = int64FromHash(values, "gift_count")
+	entry.TransactionCount = int64FromHash(values, "transaction_count")
+	entry.LastGiftAtMS = int64FromHash(values, "last_gift_at_ms")
+	return entry, nil
+}
+
+func (s *Store) scoreKey(app string, board string, period string, start time.Time) string {
+	return fmt.Sprintf("%s:%s:%s:%s:%s:scores", s.keyPrefix, appcode.Normalize(app), board, period, bucketID(period, start))
+}
+
+func (s *Store) itemKey(app string, board string, period string, start time.Time, userID string) string {
+	return fmt.Sprintf("%s:%s:%s:%s:%s:users:%s", s.keyPrefix, appcode.Normalize(app), board, period, bucketID(period, start), userID)
+}
+
+func (s *Store) dedupeKey(app string, eventID string) string {
+	return fmt.Sprintf("%s:%s:dedupe:%s", s.keyPrefix, appcode.Normalize(app), strings.TrimSpace(eventID))
+}
+
+func normalizeGiftEvent(event GiftEvent) GiftEvent {
+	event.AppCode = appcode.Normalize(event.AppCode)
+	event.EventID = strings.TrimSpace(event.EventID)
+	if event.OccurredAtMS <= 0 {
+		event.OccurredAtMS = time.Now().UTC().UnixMilli()
+	}
+	return event
+}
+
+func normalizeQuery(query Query) Query {
+	query.AppCode = appcode.Normalize(query.AppCode)
+	query.BoardType = NormalizeBoardType(query.BoardType)
+	if query.BoardType == "" {
+		query.BoardType = BoardSent
+	}
+	query.Period = NormalizePeriod(query.Period)
+	if query.Period == "" {
+		query.Period = PeriodToday
+	}
+	if query.Page <= 0 {
+		query.Page = 1
+	}
+	if query.PageSize <= 0 {
+		query.PageSize = defaultPageSize
+	}
+	if query.PageSize > maxPageSize {
+		query.PageSize = maxPageSize
+	}
+	query.Now = normalizedNow(query.Now)
+	return query
+}
+
+func normalizedNow(now time.Time) time.Time {
+	if now.IsZero() {
+		return time.Now().UTC()
+	}
+	return now.UTC()
+}
+
+func periodBounds(period string, at time.Time) (time.Time, time.Time) {
+	at = normalizedNow(at)
+	dayStart := time.Date(at.Year(), at.Month(), at.Day(), 0, 0, 0, 0, time.UTC)
+	switch NormalizePeriod(period) {
+	case PeriodWeek:
+		weekday := int(dayStart.Weekday())
+		if weekday == 0 {
+			weekday = 7
+		}
+		start := dayStart.AddDate(0, 0, 1-weekday)
+		return start, start.AddDate(0, 0, 7)
+	case PeriodMonth:
+		start := time.Date(at.Year(), at.Month(), 1, 0, 0, 0, 0, time.UTC)
+		return start, start.AddDate(0, 1, 0)
+	default:
+		return dayStart, dayStart.AddDate(0, 0, 1)
+	}
+}
+
+func retentionTTL(period string, start time.Time, periodEnd time.Time) time.Duration {
+	switch NormalizePeriod(period) {
+	case PeriodMonth:
+		return periodEnd.Sub(start) + monthlyLeaderboardKeep
+	case PeriodWeek:
+		return periodEnd.Sub(start) + weeklyLeaderboardKeep
+	default:
+		return periodEnd.Sub(start) + dailyLeaderboardKeep
+	}
+}
+
+func bucketID(period string, start time.Time) string {
+	switch NormalizePeriod(period) {
+	case PeriodWeek:
+		year, week := start.ISOWeek()
+		return fmt.Sprintf("%04d%02d", year, week)
+	case PeriodMonth:
+		return fmt.Sprintf("%04d%02d", start.Year(), int(start.Month()))
+	default:
+		return start.Format("20060102")
+	}
+}
+
+func int64FromHash(values map[string]string, key string) int64 {
+	value, _ := strconv.ParseInt(values[key], 10, 64)
+	return value
+}
+
+var applyGiftEventScript = redis.NewScript(`
+if redis.call("EXISTS", KEYS[1]) == 1 then
+	return 0
+end
+redis.call("PSETEX", KEYS[1], ARGV[1], "1")
+local gift_value = tonumber(ARGV[2])
+local gift_count = tonumber(ARGV[3])
+local occurred_at_ms = tonumber(ARGV[4])
+local updates = tonumber(ARGV[5])
+for i = 0, updates - 1 do
+	local score_key = KEYS[2 + i * 2]
+	local item_key = KEYS[3 + i * 2]
+	local member = ARGV[6 + i * 2]
+	local ttl_seconds = tonumber(ARGV[7 + i * 2])
+	local current_value = redis.call("HINCRBY", item_key, "gift_value", gift_value)
+	redis.call("HINCRBY", item_key, "gift_count", gift_count)
+	redis.call("HINCRBY", item_key, "transaction_count", 1)
+	local previous_last = tonumber(redis.call("HGET", item_key, "last_gift_at_ms") or "0")
+	local last_gift_at_ms = previous_last
+	if occurred_at_ms > previous_last then
+		redis.call("HSET", item_key, "last_gift_at_ms", occurred_at_ms)
+		last_gift_at_ms = occurred_at_ms
+	end
+	redis.call("ZADD", score_key, tonumber(current_value) + last_gift_at_ms / 10000000000000000, member)
+	redis.call("EXPIRE", score_key, ttl_seconds)
+	redis.call("EXPIRE", item_key, ttl_seconds)
+end
+return 1
+`)

pkg/userleaderboard/store_test.go

@@ -0,0 +1,52 @@
+package userleaderboard
+
+import (
+	"testing"
+	"time"
+)
+
+func TestQueryWindowUsesExistingUTCBoundaries(t *testing.T) {
+	now := time.Date(2026, 5, 20, 15, 30, 0, 0, time.FixedZone("CST", 8*60*60))
+
+	todayStart, todayEnd := QueryWindow(PeriodToday, now)
+	if todayStart != time.Date(2026, 5, 20, 0, 0, 0, 0, time.UTC) {
+		t.Fatalf("today start = %s", todayStart)
+	}
+	if !todayEnd.Equal(now.UTC()) {
+		t.Fatalf("today end = %s", todayEnd)
+	}
+
+	weekStart, _ := QueryWindow(PeriodWeek, now)
+	if weekStart != time.Date(2026, 5, 18, 0, 0, 0, 0, time.UTC) {
+		t.Fatalf("week start = %s", weekStart)
+	}
+
+	monthStart, _ := QueryWindow(PeriodMonth, now)
+	if monthStart != time.Date(2026, 5, 1, 0, 0, 0, 0, time.UTC) {
+		t.Fatalf("month start = %s", monthStart)
+	}
+}
+
+func TestNormalizeAliases(t *testing.T) {
+	if NormalizeBoardType("receiver") != BoardReceived {
+		t.Fatal("receiver should normalize to received")
+	}
+	if NormalizeBoardType("gift_sent") != BoardSent {
+		t.Fatal("gift_sent should normalize to sent")
+	}
+	if NormalizePeriod("monthly") != PeriodMonth {
+		t.Fatal("monthly should normalize to month")
+	}
+}
+
+func TestStoreKeyPrefixAndBucketAreStable(t *testing.T) {
+	store := NewStore(nil, "activity:user_leaderboard:")
+	weekStart := time.Date(2026, 12, 28, 0, 0, 0, 0, time.UTC)
+
+	if got := store.scoreKey(" HyApp_Prod ", BoardSent, PeriodWeek, weekStart); got != "activity:user_leaderboard:hyapp_prod:sent:week:202653:scores" {
+		t.Fatalf("score key mismatch: %s", got)
+	}
+	if got := store.itemKey("hyapp_prod", BoardReceived, PeriodToday, time.Date(2026, 6, 26, 0, 0, 0, 0, time.UTC), "10001"); got != "activity:user_leaderboard:hyapp_prod:received:today:20260626:users:10001" {
+		t.Fatalf("item key mismatch: %s", got)
+	}
+}

services/activity-service/configs/config.docker.yaml

@@ -24,6 +24,12 @@ red_packet_broadcast_worker:
   enabled: true
 task_event_worker:
   enabled: true
+user_leaderboard_worker:
+  enabled: true
+  redis_addr: "redis:6379"
+  redis_password: ""
+  redis_db: 0
+  key_prefix: "activity:user_leaderboard"
 message_action_confirm_worker:
   enabled: true
   outbox_poll_interval: "1s"
@@ -82,6 +88,7 @@ rocketmq:
     invite_activity_consumer_group: "hyapp-activity-invite-wallet-outbox"
     red_packet_broadcast_consumer_group: "hyapp-activity-red-packet-wallet-outbox"
     task_consumer_group: "hyapp-activity-task-wallet-outbox"
+    user_leaderboard_consumer_group: "hyapp-activity-user-leaderboard-wallet-outbox"
     consumer_max_reconsume_times: 16
   user_outbox:
     enabled: true

services/activity-service/configs/config.tencent.example.yaml

@@ -24,6 +24,12 @@ red_packet_broadcast_worker:
   enabled: true
 task_event_worker:
   enabled: true
+user_leaderboard_worker:
+  enabled: true
+  redis_addr: "TENCENT_REDIS_HOST:6379"
+  redis_password: "TENCENT_REDIS_PASSWORD"
+  redis_db: 0
+  key_prefix: "activity:user_leaderboard"
 message_action_confirm_worker:
   enabled: true
   outbox_poll_interval: "1s"
@@ -82,6 +88,7 @@ rocketmq:
     invite_activity_consumer_group: "hyapp-activity-invite-wallet-outbox"
     red_packet_broadcast_consumer_group: "hyapp-activity-red-packet-wallet-outbox"
     task_consumer_group: "hyapp-activity-task-wallet-outbox"
+    user_leaderboard_consumer_group: "hyapp-activity-user-leaderboard-wallet-outbox"
     consumer_max_reconsume_times: 16
   user_outbox:
     enabled: true

services/activity-service/configs/config.yaml

@@ -24,6 +24,12 @@ red_packet_broadcast_worker:
   enabled: true
 task_event_worker:
   enabled: true
+user_leaderboard_worker:
+  enabled: true
+  redis_addr: "127.0.0.1:13379"
+  redis_password: ""
+  redis_db: 0
+  key_prefix: "activity:user_leaderboard"
 message_action_confirm_worker:
   enabled: true
   outbox_poll_interval: "1s"
@@ -82,6 +88,7 @@ rocketmq:
     invite_activity_consumer_group: "hyapp-activity-invite-wallet-outbox"
     red_packet_broadcast_consumer_group: "hyapp-activity-red-packet-wallet-outbox"
     task_consumer_group: "hyapp-activity-task-wallet-outbox"
+    user_leaderboard_consumer_group: "hyapp-activity-user-leaderboard-wallet-outbox"
     consumer_max_reconsume_times: 16
   user_outbox:
     enabled: true

services/activity-service/internal/app/app.go

@@ -46,6 +46,7 @@ type App struct {
 	luckyGiftWorkerOptions     luckygiftservice.WorkerOptions
 	messageActionWorkerOptions config.MessageActionConfirmWorkerConfig
 	messageActionOutboxTopic   string
+	userLeaderboardRedisClose  func() error
 	mqConsumers                []*rocketmqx.Consumer
 	messageActionProducer      *rocketmqx.Producer
 	userConn                   *grpc.ClientConn
@@ -68,9 +69,13 @@ func New(cfg config.Config) (*App, error) {
 	var clients externalClients
 	var mqConsumers []*rocketmqx.Consumer
 	var messageActionProducer *rocketmqx.Producer
+	var userLeaderboardRedisClose func() error
 	cleanup := func() {
 		shutdownConsumers(mqConsumers)
 		servicemq.ShutdownProducers([]*rocketmqx.Producer{messageActionProducer})
+		if userLeaderboardRedisClose != nil {
+			_ = userLeaderboardRedisClose()
+		}
 		clients.Close()
 		if listener != nil {
 			_ = listener.Close()
@@ -90,7 +95,13 @@ func New(cfg config.Config) (*App, error) {
 	}
 
 	server := grpcserver.New("activity-service")
-	services, err := buildServiceBundle(cfg, repository, clients)
+	userLeaderboardStore, closeLeaderboardRedis, err := openUserLeaderboardStore(startupCtx, cfg.UserLeaderboardWorker)
+	if err != nil {
+		cleanup()
+		return nil, err
+	}
+	userLeaderboardRedisClose = closeLeaderboardRedis
+	services, err := buildServiceBundle(cfg, repository, clients, userLeaderboardStore)
 	if err != nil {
 		cleanup()
 		return nil, err
@@ -134,6 +145,7 @@ func New(cfg config.Config) (*App, error) {
 		luckyGiftWorkerOptions:     luckyGiftWorkerOptions(cfg.NodeID, cfg.LuckyGiftWorker),
 		messageActionWorkerOptions: cfg.MessageActionConfirmWorker,
 		messageActionOutboxTopic:   cfg.RocketMQ.MessageActionOutbox.Topic,
+		userLeaderboardRedisClose:  userLeaderboardRedisClose,
 		mqConsumers:                mqConsumers,
 		messageActionProducer:      messageActionProducer,
 		userConn:                   clients.userConn,
@@ -202,6 +214,9 @@ func (a *App) Close() {
 			// MySQL 连接池最后关闭，保证 drain 中的查询或消费提交能完成。
 			_ = a.mysqlRepo.Close()
 		}
+		if a.userLeaderboardRedisClose != nil {
+			_ = a.userLeaderboardRedisClose()
+		}
 	})
 }
 

services/activity-service/internal/app/mq.go

@@ -96,6 +96,14 @@ func buildMQConsumers(cfg config.Config, services *serviceBundle) ([]*rocketmqx.
 		}
 		mqConsumers = append(mqConsumers, consumer)
 	}
+	if cfg.UserLeaderboardWorker.Enabled && cfg.RocketMQ.WalletOutbox.Enabled {
+		consumer, err := newUserLeaderboardWalletConsumer(cfg, services)
+		if err != nil {
+			shutdownConsumers(mqConsumers)
+			return nil, err
+		}
+		mqConsumers = append(mqConsumers, consumer)
+	}
 	if cfg.TaskEventWorker.Enabled && cfg.RocketMQ.WalletOutbox.Enabled {
 		consumer, err := newTaskWalletConsumer(cfg, services)
 		if err != nil {
@@ -262,6 +270,30 @@ func newRedPacketWalletConsumer(cfg config.Config, services *serviceBundle) (*ro
 	return consumer, nil
 }
 
+func newUserLeaderboardWalletConsumer(cfg config.Config, services *serviceBundle) (*rocketmqx.Consumer, error) {
+	if services == nil || services.userLeaderboard == nil {
+		return nil, errors.New("user leaderboard worker requires redis store")
+	}
+	// 用户送礼榜使用独立 consumer group；重放榜单读模型不能推进首充、任务、红包等业务位点。
+	consumer, err := rocketmqx.NewConsumer(walletOutboxUserLeaderboardConsumerConfig(cfg.RocketMQ))
+	if err != nil {
+		return nil, err
+	}
+	if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
+		event, ok, err := userLeaderboardGiftEventFromWalletMessage(message.Body)
+		if err != nil || !ok {
+			return err
+		}
+		// Redis 聚合自身用 wallet event_id 做幂等；MQ 重投只会重新确认同一事实，不会重复加分。
+		_, err = services.userLeaderboard.ApplyGiftEvent(appcode.WithContext(ctx, event.AppCode), event)
+		return err
+	}); err != nil {
+		_ = consumer.Shutdown()
+		return nil, err
+	}
+	return consumer, nil
+}
+
 func newTaskWalletConsumer(cfg config.Config, services *serviceBundle) (*rocketmqx.Consumer, error) {
 	consumer, err := rocketmqx.NewConsumer(walletOutboxConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.WalletOutbox.TaskConsumerGroup))
 	if err != nil {
@@ -333,6 +365,13 @@ func walletOutboxConsumerConfig(cfg config.RocketMQConfig, group string) rocketm
 	return rocketMQConsumerConfig(cfg, group, cfg.WalletOutbox.ConsumerMaxReconsumeTimes)
 }
 
+func walletOutboxUserLeaderboardConsumerConfig(cfg config.RocketMQConfig) rocketmqx.ConsumerConfig {
+	consumerConfig := walletOutboxConsumerConfig(cfg, cfg.WalletOutbox.UserLeaderboardConsumerGroup)
+	// 用户榜是 Redis 可重建读模型；新 consumer group 首次上线需要尽量从 broker 保留的最早消息补齐当前周期。
+	consumerConfig.ConsumeFromFirst = true
+	return consumerConfig
+}
+
 func messageActionProducerConfig(cfg config.RocketMQConfig) rocketmqx.ProducerConfig {
 	return rocketmqx.ProducerConfig{
 		EndpointConfig: rocketmqx.EndpointConfig{

services/activity-service/internal/app/services.go

@@ -6,6 +6,7 @@ import (
 	roomv1 "hyapp.local/api/proto/room/v1"
 	walletv1 "hyapp.local/api/proto/wallet/v1"
 	"hyapp/pkg/tencentim"
+	"hyapp/pkg/userleaderboard"
 	activityclient "hyapp/services/activity-service/internal/client"
 	"hyapp/services/activity-service/internal/config"
 	achievementservice "hyapp/services/activity-service/internal/service/achievement"
@@ -47,10 +48,11 @@ type serviceBundle struct {
 	firstRechargeReward         *firstrechargeservice.Service
 	cumulativeRecharge          *cumulativerechargeservice.Service
 	inviteActivityReward        *inviteactivityservice.Service
+	userLeaderboard             *userleaderboard.Store
 	broadcastPublisherAvailable bool
 }
 
-func buildServiceBundle(cfg config.Config, repository *mysqlstorage.Repository, clients externalClients) (*serviceBundle, error) {
+func buildServiceBundle(cfg config.Config, repository *mysqlstorage.Repository, clients externalClients, userLeaderboardStore *userleaderboard.Store) (*serviceBundle, error) {
 	walletClient := walletv1.NewWalletServiceClient(clients.walletConn)
 	roomClient := roomv1.NewRoomQueryServiceClient(clients.roomConn)
 
@@ -124,6 +126,7 @@ func buildServiceBundle(cfg config.Config, repository *mysqlstorage.Repository,
 		firstRechargeReward:         firstRechargeRewardSvc,
 		cumulativeRecharge:          cumulativeRechargeSvc,
 		inviteActivityReward:        inviteActivityRewardSvc,
+		userLeaderboard:             userLeaderboardStore,
 		broadcastPublisherAvailable: broadcastPublisher != nil,
 	}, nil
 }

services/activity-service/internal/app/user_leaderboard.go

@@ -0,0 +1,25 @@
+package app
+
+import (
+	"context"
+	"strings"
+
+	"hyapp/pkg/userleaderboard"
+	"hyapp/services/activity-service/internal/config"
+)
+
+func openUserLeaderboardStore(ctx context.Context, cfg config.UserLeaderboardWorkerConfig) (*userleaderboard.Store, func() error, error) {
+	if !cfg.Enabled {
+		return nil, nil, nil
+	}
+	redisAddr := strings.TrimSpace(cfg.RedisAddr)
+	if redisAddr == "" {
+		return nil, nil, userleaderboard.ErrNotConfigured
+	}
+	client, err := userleaderboard.NewRedisClient(ctx, redisAddr, cfg.RedisPassword, cfg.RedisDB)
+	if err != nil {
+		return nil, nil, err
+	}
+	// activity-service 只负责把 wallet_outbox 已提交事实投影成 Redis 读模型；key 命名由共享包和配置共同约束。
+	return userleaderboard.NewStore(client, cfg.KeyPrefix), client.Close, nil
+}

services/activity-service/internal/app/wallet_events.go

@@ -5,6 +5,7 @@ import (
 	"strings"
 
 	"hyapp/pkg/appcode"
+	"hyapp/pkg/userleaderboard"
 	"hyapp/pkg/walletmq"
 	broadcastdomain "hyapp/services/activity-service/internal/domain/broadcast"
 	cumulativerechargedomain "hyapp/services/activity-service/internal/domain/cumulativerecharge"
@@ -127,6 +128,39 @@ func redPacketEventFromWalletMessage(body []byte) (broadcastdomain.RedPacketWall
 	}, true, nil
 }
 
+func userLeaderboardGiftEventFromWalletMessage(body []byte) (userleaderboard.GiftEvent, bool, error) {
+	message, err := walletmq.DecodeWalletOutboxMessage(body)
+	if err != nil {
+		return userleaderboard.GiftEvent{}, false, err
+	}
+	if message.EventType != "WalletGiftDebited" {
+		return userleaderboard.GiftEvent{}, false, nil
+	}
+	var decoded map[string]any
+	if err := json.Unmarshal([]byte(message.PayloadJSON), &decoded); err != nil {
+		return userleaderboard.GiftEvent{}, false, err
+	}
+	if boolFromDecoded(decoded, "direct_gift") {
+		// 旧 gateway SQL 只统计 biz_type='gift_debit'，direct_gift_debit 不进榜；Redis 投影必须保持这个口径。
+		return userleaderboard.GiftEvent{}, false, nil
+	}
+	senderUserID := firstNonZeroInt64(int64FromDecoded(decoded, "sender_user_id"), message.UserID)
+	targetUserID := int64FromDecoded(decoded, "target_user_id")
+	giftValue, _ := firstPresentInt64(decoded, "heat_value", "coin_spent", "charge_amount")
+	if senderUserID <= 0 || targetUserID <= 0 || giftValue <= 0 {
+		return userleaderboard.GiftEvent{}, false, nil
+	}
+	return userleaderboard.GiftEvent{
+		AppCode:      appcode.Normalize(message.AppCode),
+		EventID:      message.EventID,
+		SenderUserID: senderUserID,
+		TargetUserID: targetUserID,
+		GiftValue:    giftValue,
+		GiftCount:    int64FromDecoded(decoded, "gift_count"),
+		OccurredAtMS: message.OccurredAtMS,
+	}, true, nil
+}
+
 type firstRechargePayloadFields struct {
 	sequence         int64
 	rechargeType     string
@@ -329,6 +363,9 @@ func int64FromDecoded(decoded map[string]any, key string) int64 {
 	case json.Number:
 		parsed, _ := value.Int64()
 		return parsed
+	case string:
+		parsed, _ := json.Number(strings.TrimSpace(value)).Int64()
+		return parsed
 	default:
 		return 0
 	}
@@ -337,3 +374,54 @@ func int64FromDecoded(decoded map[string]any, key string) int64 {
 func int32FromDecoded(decoded map[string]any, key string) int32 {
 	return int32(int64FromDecoded(decoded, key))
 }
+
+func firstPresentInt64(decoded map[string]any, keys ...string) (int64, bool) {
+	for _, key := range keys {
+		value, ok := int64FromDecodedIfPresent(decoded, key)
+		if ok {
+			return value, true
+		}
+	}
+	return 0, false
+}
+
+func int64FromDecodedIfPresent(decoded map[string]any, key string) (int64, bool) {
+	raw, exists := decoded[key]
+	if !exists {
+		return 0, false
+	}
+	switch value := raw.(type) {
+	case float64:
+		return int64(value), true
+	case int64:
+		return value, true
+	case json.Number:
+		parsed, err := value.Int64()
+		return parsed, err == nil
+	case string:
+		value = strings.TrimSpace(value)
+		if value == "" {
+			return 0, false
+		}
+		parsed, err := json.Number(value).Int64()
+		return parsed, err == nil
+	default:
+		return 0, false
+	}
+}
+
+func boolFromDecoded(decoded map[string]any, key string) bool {
+	switch value := decoded[key].(type) {
+	case bool:
+		return value
+	case string:
+		return strings.EqualFold(strings.TrimSpace(value), "true") || strings.TrimSpace(value) == "1"
+	case float64:
+		return value != 0
+	case json.Number:
+		parsed, _ := value.Int64()
+		return parsed != 0
+	default:
+		return false
+	}
+}

services/activity-service/internal/app/wallet_events_user_leaderboard_test.go

@@ -0,0 +1,105 @@
+package app
+
+import (
+	"encoding/json"
+	"testing"
+
+	"hyapp/pkg/walletmq"
+)
+
+func TestUserLeaderboardGiftEventFromWalletMessage(t *testing.T) {
+	body := mustWalletOutboxMessage(t, "WalletGiftDebited", map[string]any{
+		"sender_user_id": 10001,
+		"target_user_id": 20002,
+		"heat_value":     900,
+		"coin_spent":     1200,
+		"charge_amount":  1200,
+		"gift_count":     3,
+	})
+
+	event, ok, err := userLeaderboardGiftEventFromWalletMessage(body)
+	if err != nil || !ok {
+		t.Fatalf("gift event should parse: ok=%v err=%v", ok, err)
+	}
+	if event.AppCode != "hyapp_room" || event.EventID != "evt-1" || event.SenderUserID != 10001 || event.TargetUserID != 20002 {
+		t.Fatalf("event identity mismatch: %+v", event)
+	}
+	if event.GiftValue != 900 || event.GiftCount != 3 || event.OccurredAtMS != 1782461920000 {
+		t.Fatalf("event aggregate fields mismatch: %+v", event)
+	}
+}
+
+func TestUserLeaderboardGiftEventSkipsDirectGift(t *testing.T) {
+	body := mustWalletOutboxMessage(t, "WalletGiftDebited", map[string]any{
+		"sender_user_id": 10001,
+		"target_user_id": 20002,
+		"heat_value":     900,
+		"gift_count":     1,
+		"direct_gift":    true,
+	})
+
+	_, ok, err := userLeaderboardGiftEventFromWalletMessage(body)
+	if err != nil {
+		t.Fatalf("direct gift parse failed: %v", err)
+	}
+	if ok {
+		t.Fatal("direct gift must keep the old gift_debit SQL scope and skip user leaderboard")
+	}
+}
+
+func TestUserLeaderboardGiftEventKeepsSQLCoalesceSemantics(t *testing.T) {
+	fallbackBody := mustWalletOutboxMessage(t, "WalletGiftDebited", map[string]any{
+		"sender_user_id": 10001,
+		"target_user_id": 20002,
+		"coin_spent":     1200,
+		"charge_amount":  1500,
+		"gift_count":     1,
+	})
+	fallback, ok, err := userLeaderboardGiftEventFromWalletMessage(fallbackBody)
+	if err != nil || !ok {
+		t.Fatalf("fallback gift event should parse: ok=%v err=%v", ok, err)
+	}
+	if fallback.GiftValue != 1200 {
+		t.Fatalf("missing heat_value should fall back to coin_spent, got %+v", fallback)
+	}
+
+	zeroHeatBody := mustWalletOutboxMessage(t, "WalletGiftDebited", map[string]any{
+		"sender_user_id": 10001,
+		"target_user_id": 20002,
+		"heat_value":     0,
+		"coin_spent":     1200,
+		"charge_amount":  1500,
+		"gift_count":     1,
+	})
+	_, ok, err = userLeaderboardGiftEventFromWalletMessage(zeroHeatBody)
+	if err != nil {
+		t.Fatalf("zero heat gift event parse failed: %v", err)
+	}
+	if ok {
+		t.Fatal("heat_value=0 should not fall back to coin_spent because old SQL COALESCE treats zero as present")
+	}
+}
+
+func mustWalletOutboxMessage(t *testing.T, eventType string, payload map[string]any) []byte {
+	t.Helper()
+	payloadBytes, err := json.Marshal(payload)
+	if err != nil {
+		t.Fatalf("marshal payload failed: %v", err)
+	}
+	body, err := walletmq.EncodeWalletOutboxMessage(walletmq.WalletOutboxMessage{
+		AppCode:        "hyapp_room",
+		EventID:        "evt-1",
+		EventType:      eventType,
+		TransactionID:  "tx-1",
+		CommandID:      "cmd-1",
+		UserID:         10001,
+		AssetType:      "COIN",
+		AvailableDelta: -1200,
+		PayloadJSON:    string(payloadBytes),
+		OccurredAtMS:   1782461920000,
+	})
+	if err != nil {
+		t.Fatalf("encode wallet outbox failed: %v", err)
+	}
+	return body
+}

services/activity-service/internal/config/config.go

@@ -37,6 +37,8 @@ type Config struct {
 	RedPacketBroadcastWorker RedPacketBroadcastWorkerConfig `yaml:"red_packet_broadcast_worker"`
 	// TaskEventWorker 控制任务系统对 wallet/user outbox 事实的本地消费。
 	TaskEventWorker TaskEventWorkerConfig `yaml:"task_event_worker"`
+	// UserLeaderboardWorker 控制用户送礼榜单对 wallet_outbox 送礼事实的 Redis 投影。
+	UserLeaderboardWorker UserLeaderboardWorkerConfig `yaml:"user_leaderboard_worker"`
 	// LuckyGiftWorker 控制幸运礼物 draw outbox 的返奖和房间 IM 补偿。
 	LuckyGiftWorker LuckyGiftWorkerConfig `yaml:"lucky_gift_worker"`
 	// TencentIM 是 activity-service 发送全局/区域播报群消息的 REST 配置。
@@ -105,6 +107,20 @@ type TaskEventWorkerConfig struct {
 	Enabled bool `yaml:"enabled"`
 }
 
+// UserLeaderboardWorkerConfig 保存用户送礼榜单 Redis 投影策略。
+type UserLeaderboardWorkerConfig struct {
+	// Enabled 控制是否启动 wallet_outbox 用户榜单消费者；关闭时 gateway 只能读到已有 Redis 聚合。
+	Enabled bool `yaml:"enabled"`
+	// RedisAddr 是 activity 写入用户榜单聚合的 Redis 地址，必须与 gateway 读取地址一致。
+	RedisAddr string `yaml:"redis_addr"`
+	// RedisPassword 是 Redis 鉴权密码，本地默认留空。
+	RedisPassword string `yaml:"redis_password"`
+	// RedisDB 是 Redis logical DB，必须与 gateway 读取配置一致。
+	RedisDB int `yaml:"redis_db"`
+	// KeyPrefix 隔离用户榜单 key，必须与 gateway 读取配置一致。
+	KeyPrefix string `yaml:"key_prefix"`
+}
+
 // LuckyGiftWorkerConfig 保存幸运礼物抽奖副作用补偿策略。
 type LuckyGiftWorkerConfig struct {
 	Enabled              bool          `yaml:"enabled"`
@@ -193,6 +209,7 @@ type WalletOutboxMQConfig struct {
 	InviteActivityConsumerGroup     string `yaml:"invite_activity_consumer_group"`
 	RedPacketBroadcastConsumerGroup string `yaml:"red_packet_broadcast_consumer_group"`
 	TaskConsumerGroup               string `yaml:"task_consumer_group"`
+	UserLeaderboardConsumerGroup    string `yaml:"user_leaderboard_consumer_group"`
 	ConsumerMaxReconsumeTimes       int32  `yaml:"consumer_max_reconsume_times"`
 }
 
@@ -243,6 +260,9 @@ func Default() Config {
 		TaskEventWorker: TaskEventWorkerConfig{
 			Enabled: true,
 		},
+		UserLeaderboardWorker: UserLeaderboardWorkerConfig{
+			Enabled: false,
+		},
 		MessageActionConfirmWorker: MessageActionConfirmWorkerConfig{
 			Enabled:            true,
 			OutboxPollInterval: time.Second,
@@ -314,6 +334,7 @@ func defaultRocketMQConfig() RocketMQConfig {
 			InviteActivityConsumerGroup:     "hyapp-activity-invite-wallet-outbox",
 			RedPacketBroadcastConsumerGroup: "hyapp-activity-red-packet-wallet-outbox",
 			TaskConsumerGroup:               "hyapp-activity-task-wallet-outbox",
+			UserLeaderboardConsumerGroup:    "hyapp-activity-user-leaderboard-wallet-outbox",
 			ConsumerMaxReconsumeTimes:       16,
 		},
 		UserOutbox: UserOutboxMQConfig{
@@ -440,14 +461,19 @@ func Load(path string) (Config, error) {
 	if cfg.MessageActionConfirmWorker.PublishTimeout <= 0 {
 		cfg.MessageActionConfirmWorker.PublishTimeout = 5 * time.Second
 	}
+	cfg.UserLeaderboardWorker.RedisAddr = strings.TrimSpace(cfg.UserLeaderboardWorker.RedisAddr)
+	cfg.UserLeaderboardWorker.KeyPrefix = strings.TrimSpace(cfg.UserLeaderboardWorker.KeyPrefix)
 	rocketMQ, err := normalizeRocketMQConfig(cfg.RocketMQ)
 	if err != nil {
 		return Config{}, err
 	}
 	cfg.RocketMQ = rocketMQ
-	if (cfg.FirstRechargeRewardWorker.Enabled || cfg.CumulativeRechargeRewardWorker.Enabled || cfg.InviteActivityRewardWorker.Enabled || cfg.RedPacketBroadcastWorker.Enabled) && !cfg.RocketMQ.WalletOutbox.Enabled {
+	if (cfg.FirstRechargeRewardWorker.Enabled || cfg.CumulativeRechargeRewardWorker.Enabled || cfg.InviteActivityRewardWorker.Enabled || cfg.RedPacketBroadcastWorker.Enabled || cfg.UserLeaderboardWorker.Enabled) && !cfg.RocketMQ.WalletOutbox.Enabled {
 		return Config{}, errors.New("wallet outbox workers require rocketmq.wallet_outbox.enabled")
 	}
+	if cfg.UserLeaderboardWorker.Enabled && cfg.UserLeaderboardWorker.RedisAddr == "" {
+		return Config{}, errors.New("user leaderboard worker requires redis_addr")
+	}
 	if cfg.InviteActivityRewardWorker.Enabled && !cfg.RocketMQ.UserOutbox.Enabled {
 		return Config{}, errors.New("invite activity reward worker requires rocketmq.user_outbox.enabled")
 	}
@@ -497,6 +523,9 @@ func normalizeRocketMQConfig(cfg RocketMQConfig) (RocketMQConfig, error) {
 	if cfg.WalletOutbox.TaskConsumerGroup = strings.TrimSpace(cfg.WalletOutbox.TaskConsumerGroup); cfg.WalletOutbox.TaskConsumerGroup == "" {
 		cfg.WalletOutbox.TaskConsumerGroup = defaults.WalletOutbox.TaskConsumerGroup
 	}
+	if cfg.WalletOutbox.UserLeaderboardConsumerGroup = strings.TrimSpace(cfg.WalletOutbox.UserLeaderboardConsumerGroup); cfg.WalletOutbox.UserLeaderboardConsumerGroup == "" {
+		cfg.WalletOutbox.UserLeaderboardConsumerGroup = defaults.WalletOutbox.UserLeaderboardConsumerGroup
+	}
 	if cfg.WalletOutbox.ConsumerMaxReconsumeTimes <= 0 {
 		cfg.WalletOutbox.ConsumerMaxReconsumeTimes = defaults.WalletOutbox.ConsumerMaxReconsumeTimes
 	}

services/activity-service/internal/config/config_test.go

@@ -11,10 +11,13 @@ func TestLoadLocalEnablesOutboxMQConsumers(t *testing.T) {
 	if !cfg.RocketMQ.Enabled || !cfg.RocketMQ.RoomOutbox.Enabled || !cfg.RocketMQ.WalletOutbox.Enabled || !cfg.RocketMQ.UserOutbox.Enabled {
 		t.Fatalf("local config must enable all activity outbox MQ consumers: %+v", cfg.RocketMQ)
 	}
-	if !cfg.FirstRechargeRewardWorker.Enabled || !cfg.CumulativeRechargeRewardWorker.Enabled || !cfg.InviteActivityRewardWorker.Enabled || !cfg.RedPacketBroadcastWorker.Enabled {
+	if !cfg.FirstRechargeRewardWorker.Enabled || !cfg.CumulativeRechargeRewardWorker.Enabled || !cfg.InviteActivityRewardWorker.Enabled || !cfg.RedPacketBroadcastWorker.Enabled || !cfg.UserLeaderboardWorker.Enabled {
-		t.Fatalf("local config must enable wallet outbox workers: first=%+v cumulative=%+v invite=%+v red_packet=%+v", cfg.FirstRechargeRewardWorker, cfg.CumulativeRechargeRewardWorker, cfg.InviteActivityRewardWorker, cfg.RedPacketBroadcastWorker)
+		t.Fatalf("local config must enable wallet outbox workers: first=%+v cumulative=%+v invite=%+v red_packet=%+v user_leaderboard=%+v", cfg.FirstRechargeRewardWorker, cfg.CumulativeRechargeRewardWorker, cfg.InviteActivityRewardWorker, cfg.RedPacketBroadcastWorker, cfg.UserLeaderboardWorker)
 	}
-	if cfg.RocketMQ.WalletOutbox.InviteActivityConsumerGroup == "" || cfg.RocketMQ.UserOutbox.InviteActivityConsumerGroup == "" {
+	if cfg.UserLeaderboardWorker.RedisAddr != "127.0.0.1:13379" || cfg.UserLeaderboardWorker.KeyPrefix != "activity:user_leaderboard" {
+		t.Fatalf("local user leaderboard worker redis config mismatch: %+v", cfg.UserLeaderboardWorker)
+	}
+	if cfg.RocketMQ.WalletOutbox.InviteActivityConsumerGroup == "" || cfg.RocketMQ.WalletOutbox.UserLeaderboardConsumerGroup == "" || cfg.RocketMQ.UserOutbox.InviteActivityConsumerGroup == "" {
 		t.Fatalf("local config must configure invite activity consumer groups: wallet=%+v user=%+v", cfg.RocketMQ.WalletOutbox, cfg.RocketMQ.UserOutbox)
 	}
 }
@@ -36,4 +39,7 @@ func TestLoadTencentExampleEnablesRoomOutboxMQ(t *testing.T) {
 	if !cfg.InviteActivityRewardWorker.Enabled || cfg.RocketMQ.WalletOutbox.InviteActivityConsumerGroup == "" || cfg.RocketMQ.UserOutbox.InviteActivityConsumerGroup == "" {
 		t.Fatalf("tencent example must enable invite activity consumer groups: invite=%+v wallet=%+v user=%+v", cfg.InviteActivityRewardWorker, cfg.RocketMQ.WalletOutbox, cfg.RocketMQ.UserOutbox)
 	}
+	if !cfg.UserLeaderboardWorker.Enabled || cfg.UserLeaderboardWorker.RedisAddr == "" || cfg.UserLeaderboardWorker.RedisPassword == "" || cfg.RocketMQ.WalletOutbox.UserLeaderboardConsumerGroup == "" {
+		t.Fatalf("tencent example must configure user leaderboard worker: worker=%+v wallet=%+v", cfg.UserLeaderboardWorker, cfg.RocketMQ.WalletOutbox)
+	}
 }

services/gateway-service/configs/config.docker.yaml

@@ -58,8 +58,11 @@ app_config:
   key_prefix: "gateway:app_config:"
   cache_ttl: "5m"
 leaderboard:
-  # 活动用户榜单只读 wallet_transactions 的送礼事实；时间窗口统一按 UTC 计算。
+  # 活动用户榜单只读 activity-service 写入的 Redis 聚合结果；时间窗口统一按 UTC 计算。
-  wallet_mysql_dsn: "hyapp:hyapp@tcp(mysql:3306)/hyapp_wallet?parseTime=true&charset=utf8mb4&loc=UTC"
+  redis_addr: "redis:6379"
+  redis_password: ""
+  redis_db: 0
+  key_prefix: "activity:user_leaderboard"
 auth_rate_limit:
   # public auth 固定窗口频控；Redis 不可用时 gateway 启动失败，避免失去入口保护。
   enabled: true

services/gateway-service/configs/config.tencent.example.yaml

@@ -62,8 +62,11 @@ app_config:
   key_prefix: "gateway:app_config:"
   cache_ttl: "5m"
 leaderboard:
-  # 活动用户榜单只读钱包送礼事实；线上建议配置只读账号。
+  # 活动用户榜单只读 activity-service 写入的 Redis 聚合结果；必须和 activity-service 使用同一 key_prefix。
-  wallet_mysql_dsn: "TENCENT_MYSQL_USER:TENCENT_MYSQL_PASSWORD@tcp(TENCENT_MYSQL_HOST:3306)/hyapp_wallet?parseTime=true&charset=utf8mb4&loc=UTC"
+  redis_addr: "TENCENT_REDIS_HOST:6379"
+  redis_password: "TENCENT_REDIS_PASSWORD"
+  redis_db: 0
+  key_prefix: "activity:user_leaderboard"
 auth_rate_limit:
   # public auth 固定窗口频控；Redis 不可用时 gateway 启动失败，避免失去入口保护。
   enabled: true

services/gateway-service/configs/config.yaml

@@ -58,8 +58,11 @@ app_config:
   key_prefix: "gateway:app_config:"
   cache_ttl: "5m"
 leaderboard:
-  # 活动用户榜单只读 wallet_transactions 的送礼事实；时间窗口统一按 UTC 计算。
+  # 活动用户榜单只读 activity-service 写入的 Redis 聚合结果；时间窗口统一按 UTC 计算。
-  wallet_mysql_dsn: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_wallet?parseTime=true&charset=utf8mb4&loc=UTC"
+  redis_addr: "127.0.0.1:13379"
+  redis_password: ""
+  redis_db: 0
+  key_prefix: "activity:user_leaderboard"
 auth_rate_limit:
   # public auth 固定窗口频控；Redis 不可用时 gateway 启动失败，避免失去入口保护。
   enabled: true

services/gateway-service/internal/app/app.go

@@ -2,7 +2,6 @@ package app
 
 import (
 	"context"
-	"database/sql"
 	"errors"
 	"fmt"
 	"net"
@@ -12,12 +11,12 @@ import (
 	"sync"
 	"time"
 
-	_ "github.com/go-sql-driver/mysql"
 	"google.golang.org/grpc"
 	"hyapp/pkg/grpcclient"
 	serviceapp "hyapp/pkg/servicekit/app"
 	"hyapp/pkg/tencentcos"
 	"hyapp/pkg/tencentim"
+	"hyapp/pkg/userleaderboard"
 	"hyapp/services/gateway-service/internal/appconfig"
 	"hyapp/services/gateway-service/internal/auth"
 	"hyapp/services/gateway-service/internal/client"
@@ -36,7 +35,6 @@ type App struct {
 	activityConn *grpc.ClientConn
 	gameConn     *grpc.ClientConn
 	appConfig    *appconfig.MySQLReader
-	leaderboard  *sql.DB
 	redisClose   func() error
 	health       *healthcheck.State
 	closeOnce    sync.Once
@@ -125,18 +123,6 @@ func New(cfg config.Config) (*App, error) {
 		_ = gameConn.Close()
 		return nil, err
 	}
-	leaderboardWalletDB, err := openLeaderboardWalletDB(cfg.Leaderboard)
-	if err != nil {
-		if appConfigReader != nil {
-			_ = appConfigReader.Close()
-		}
-		_ = roomConn.Close()
-		_ = userConn.Close()
-		_ = walletConn.Close()
-		_ = activityConn.Close()
-		_ = gameConn.Close()
-		return nil, err
-	}
 	handler := httptransport.NewHandlerWithConfig(roomClient, roomGuardClient, userClient, userIdentityClient, httptransport.TencentIMConfig{
 		SDKAppID:          cfg.TencentIM.SDKAppID,
 		SecretKey:         cfg.TencentIM.SecretKey,
@@ -151,9 +137,6 @@ func New(cfg config.Config) (*App, error) {
 			if appConfigReader != nil {
 				_ = appConfigReader.Close()
 			}
-			if leaderboardWalletDB != nil {
-				_ = leaderboardWalletDB.Close()
-			}
 			_ = roomConn.Close()
 			_ = userConn.Close()
 			_ = walletConn.Close()
@@ -193,7 +176,19 @@ func New(cfg config.Config) (*App, error) {
 	handler.SetBroadcastClient(broadcastClient)
 	handler.SetGameClient(gameClient)
 	handler.SetStatisticsClient(statisticsClient)
-	handler.SetLeaderboardWalletDB(leaderboardWalletDB)
+	leaderboardRedisClose, err := configureUserLeaderboard(handler, cfg.Leaderboard)
+	if err != nil {
+		if appConfigReader != nil {
+			_ = appConfigReader.Close()
+		}
+		_ = roomConn.Close()
+		_ = userConn.Close()
+		_ = walletConn.Close()
+		_ = activityConn.Close()
+		_ = gameConn.Close()
+		return nil, err
+	}
+	redisClose := leaderboardRedisClose
 	if appConfigReader != nil {
 		handler.SetAppConfigReader(appConfigReader)
 	}
@@ -218,8 +213,8 @@ func New(cfg config.Config) (*App, error) {
 			if appConfigReader != nil {
 				_ = appConfigReader.Close()
 			}
-			if leaderboardWalletDB != nil {
+			if redisClose != nil {
-				_ = leaderboardWalletDB.Close()
+				_ = redisClose()
 			}
 			_ = roomConn.Close()
 			_ = userConn.Close()
@@ -252,13 +247,13 @@ func New(cfg config.Config) (*App, error) {
 		},
 	}
 	handler.SetLoginRisk(loginRiskConfig)
-	redisClose, err := configureAuthRateLimit(handler, cfg.AuthRateLimit, rateLimitConfig)
+	authRedisClose, err := configureAuthRateLimit(handler, cfg.AuthRateLimit, rateLimitConfig)
 	if err != nil {
 		if appConfigReader != nil {
 			_ = appConfigReader.Close()
 		}
-		if leaderboardWalletDB != nil {
+		if redisClose != nil {
-			_ = leaderboardWalletDB.Close()
+			_ = redisClose()
 		}
 		_ = roomConn.Close()
 		_ = userConn.Close()
@@ -267,6 +262,7 @@ func New(cfg config.Config) (*App, error) {
 		_ = gameConn.Close()
 		return nil, err
 	}
+	redisClose = joinClose(redisClose, authRedisClose)
 	loginRiskRedisClose, err := configureLoginRisk(handler, cfg.LoginRisk, loginRiskConfig)
 	if err != nil {
 		if redisClose != nil {
@@ -275,9 +271,6 @@ func New(cfg config.Config) (*App, error) {
 		if appConfigReader != nil {
 			_ = appConfigReader.Close()
 		}
-		if leaderboardWalletDB != nil {
-			_ = leaderboardWalletDB.Close()
-		}
 		_ = roomConn.Close()
 		_ = userConn.Close()
 		_ = walletConn.Close()
@@ -308,9 +301,6 @@ func New(cfg config.Config) (*App, error) {
 		if appConfigReader != nil {
 			_ = appConfigReader.Close()
 		}
-		if leaderboardWalletDB != nil {
-			_ = leaderboardWalletDB.Close()
-		}
 		_ = roomConn.Close()
 		_ = userConn.Close()
 		_ = walletConn.Close()
@@ -340,7 +330,6 @@ func New(cfg config.Config) (*App, error) {
 		activityConn: activityConn,
 		gameConn:     gameConn,
 		appConfig:    appConfigReader,
-		leaderboard:  leaderboardWalletDB,
 		redisClose:   redisClose,
 		health:       healthState,
 	}, nil
@@ -369,14 +358,6 @@ func openAppConfigReader(cfg config.AppConfigConfig) (*appconfig.MySQLReader, er
 	return reader, nil
 }
 
-func openLeaderboardWalletDB(cfg config.LeaderboardConfig) (*sql.DB, error) {
-	if strings.TrimSpace(cfg.WalletMySQLDSN) == "" {
-		return nil, nil
-	}
-
-	return sql.Open("mysql", strings.TrimSpace(cfg.WalletMySQLDSN))
-}
-
 func configureAuthRateLimit(handler *httptransport.Handler, cfg config.AuthRateLimitConfig, rateLimitConfig httptransport.AuthRateLimitConfig) (func() error, error) {
 	if !rateLimitConfig.Enabled {
 		// 显式关闭频控时仍记录配置，避免 handler 使用默认开启策略。
@@ -401,6 +382,22 @@ func configureAuthRateLimit(handler *httptransport.Handler, cfg config.AuthRateL
 	return redisClient.Close, nil
 }
 
+func configureUserLeaderboard(handler *httptransport.Handler, cfg config.LeaderboardConfig) (func() error, error) {
+	redisAddr := strings.TrimSpace(cfg.RedisAddr)
+	if redisAddr == "" {
+		return nil, nil
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	redisClient, err := userleaderboard.NewRedisClient(ctx, redisAddr, cfg.RedisPassword, cfg.RedisDB)
+	if err != nil {
+		return nil, fmt.Errorf("connect user leaderboard redis: %w", err)
+	}
+	// gateway 只读取 activity-service 维护的 Redis 聚合结果，不再在 HTTP 请求中扫描 wallet_transactions。
+	handler.SetUserLeaderboardStore(userleaderboard.NewStore(redisClient, cfg.KeyPrefix))
+	return redisClient.Close, nil
+}
+
 func configureLoginRisk(handler *httptransport.Handler, cfg config.LoginRiskConfig, riskConfig httptransport.LoginRiskConfig) (func() error, error) {
 	if !riskConfig.Enabled {
 		handler.SetLoginRisk(riskConfig)
@@ -480,9 +477,6 @@ func (a *App) Close() error {
 		if a.appConfig != nil {
 			err = errors.Join(err, a.appConfig.Close())
 		}
-		if a.leaderboard != nil {
-			err = errors.Join(err, a.leaderboard.Close())
-		}
 		if a.redisClose != nil {
 			err = errors.Join(err, a.redisClose())
 		}

services/gateway-service/internal/config/config.go

@@ -127,10 +127,18 @@ type AppConfigConfig struct {
 	CacheTTL time.Duration `yaml:"cache_ttl"`
 }
 
-// LeaderboardConfig 描述 App 活动榜单读取钱包送礼事实的只读数据源。
+// LeaderboardConfig 描述 App 活动榜单读取 Redis 聚合读模型的只读数据源。
 type LeaderboardConfig struct {
-	// WalletMySQLDSN 指向钱包库 hyapp_wallet；只读取 wallet_transactions 的 gift_debit 事实。
+	// WalletMySQLDSN 是旧 SQL 榜单路径的兼容字段；gateway 当前不再读取钱包库。
 	WalletMySQLDSN string `yaml:"wallet_mysql_dsn"`
+	// RedisAddr 是 activity-service 写入用户送礼榜单聚合的 Redis 地址。
+	RedisAddr string `yaml:"redis_addr"`
+	// RedisPassword 是 Redis 鉴权密码，本地默认留空。
+	RedisPassword string `yaml:"redis_password"`
+	// RedisDB 是 Redis logical DB，必须与 activity-service 写入配置一致。
+	RedisDB int `yaml:"redis_db"`
+	// KeyPrefix 隔离用户榜单 key，必须与 activity-service 写入配置一致。
+	KeyPrefix string `yaml:"key_prefix"`
 }
 
 // StatisticsConfig 描述 gateway 转发客户端 H5 埋点到 statistics-service 的内部 HTTP 入口。
@@ -312,7 +320,8 @@ func Default() Config {
 			CacheTTL:  5 * time.Minute,
 		},
 		Leaderboard: LeaderboardConfig{
-			WalletMySQLDSN: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_wallet?parseTime=true&charset=utf8mb4&loc=UTC",
+			RedisAddr: "127.0.0.1:13379",
+			KeyPrefix: "activity:user_leaderboard",
 		},
 		AuthRateLimit: AuthRateLimitConfig{
 			Enabled:              true,
@@ -451,6 +460,16 @@ func (cfg *Config) Normalize() error {
 		cfg.AppConfig.CacheTTL = 5 * time.Minute
 	}
 	cfg.Leaderboard.WalletMySQLDSN = strings.TrimSpace(cfg.Leaderboard.WalletMySQLDSN)
+	cfg.Leaderboard.RedisAddr = strings.TrimSpace(cfg.Leaderboard.RedisAddr)
+	cfg.Leaderboard.KeyPrefix = strings.TrimSpace(cfg.Leaderboard.KeyPrefix)
+	if cfg.Leaderboard.RedisAddr == "" {
+		cfg.Leaderboard.RedisAddr = strings.TrimSpace(cfg.AuthRateLimit.RedisAddr)
+		cfg.Leaderboard.RedisPassword = cfg.AuthRateLimit.RedisPassword
+		cfg.Leaderboard.RedisDB = cfg.AuthRateLimit.RedisDB
+	}
+	if cfg.Leaderboard.KeyPrefix == "" {
+		cfg.Leaderboard.KeyPrefix = "activity:user_leaderboard"
+	}
 	cfg.LoginRisk.RedisAddr = strings.TrimSpace(cfg.LoginRisk.RedisAddr)
 	cfg.LoginRisk.KeyPrefix = strings.TrimSpace(cfg.LoginRisk.KeyPrefix)
 	if cfg.LoginRisk.Enabled && cfg.LoginRisk.RedisAddr == "" {

services/gateway-service/internal/config/config_test.go

@@ -31,6 +31,9 @@ func TestLoad(t *testing.T) {
 	if cfg.AppConfig.MySQLDSN == "" || !strings.Contains(cfg.AppConfig.MySQLDSN, "hyapp_admin") {
 		t.Fatalf("gateway app config mysql dsn must point to admin config db: %+v", cfg.AppConfig)
 	}
+	if cfg.Leaderboard.RedisAddr != "127.0.0.1:13379" || cfg.Leaderboard.KeyPrefix != "activity:user_leaderboard" {
+		t.Fatalf("local leaderboard redis config mismatch: %+v", cfg.Leaderboard)
+	}
 	if !cfg.AuthRateLimit.Enabled || cfg.AuthRateLimit.RedisAddr != "127.0.0.1:13379" {
 		t.Fatalf("local auth rate redis config mismatch: %+v", cfg.AuthRateLimit)
 	}
@@ -73,6 +76,9 @@ func TestLoadTencentExample(t *testing.T) {
 	if cfg.AppConfig.MySQLDSN == "" || !strings.Contains(cfg.AppConfig.MySQLDSN, "hyapp_admin") {
 		t.Fatalf("tencent app config mysql dsn template must be explicit: %+v", cfg.AppConfig)
 	}
+	if cfg.Leaderboard.RedisAddr == "" || cfg.Leaderboard.RedisPassword == "" || cfg.Leaderboard.KeyPrefix != "activity:user_leaderboard" {
+		t.Fatalf("tencent leaderboard redis template must be explicit: %+v", cfg.Leaderboard)
+	}
 	if cfg.TencentIM.UserSigTTL != 24*time.Hour {
 		t.Fatalf("unexpected usersig ttl: %s", cfg.TencentIM.UserSigTTL)
 	}

services/gateway-service/internal/transport/http/activityapi/handler.go

@@ -1,8 +1,9 @@
 package activityapi
 
 import (
-	"database/sql"
+	"context"
 
+	"hyapp/pkg/userleaderboard"
 	"hyapp/services/gateway-service/internal/client"
 	"hyapp/services/gateway-service/internal/transport/http/httproutes"
 )
@@ -16,7 +17,7 @@ type Handler struct {
 	userProfileClient    client.UserProfileClient
 	roomQueryClient      client.RoomQueryClient
 	walletClient         client.WalletClient
-	walletDB             *sql.DB
+	userLeaderboard      UserLeaderboardStore
 	registrationReward   client.RegistrationRewardClient
 	firstRechargeReward  client.FirstRechargeRewardClient
 	cumulativeRecharge   client.CumulativeRechargeRewardClient
@@ -30,6 +31,11 @@ type Handler struct {
 	wheel                client.WheelClient
 }
 
+type UserLeaderboardStore interface {
+	List(ctx context.Context, query userleaderboard.Query) (userleaderboard.Page, error)
+	RankForUser(ctx context.Context, query userleaderboard.Query, userID int64) (userleaderboard.Entry, bool, error)
+}
+
 type Config struct {
 	TaskClient           client.TaskClient
 	GrowthLevelClient    client.GrowthLevelClient
@@ -37,7 +43,7 @@ type Config struct {
 	UserProfileClient    client.UserProfileClient
 	RoomQueryClient      client.RoomQueryClient
 	WalletClient         client.WalletClient
-	WalletDB             *sql.DB
+	UserLeaderboard      UserLeaderboardStore
 	RegistrationReward   client.RegistrationRewardClient
 	FirstRechargeReward  client.FirstRechargeRewardClient
 	CumulativeRecharge   client.CumulativeRechargeRewardClient
@@ -59,7 +65,7 @@ func New(config Config) *Handler {
 		userProfileClient:    config.UserProfileClient,
 		roomQueryClient:      config.RoomQueryClient,
 		walletClient:         config.WalletClient,
-		walletDB:             config.WalletDB,
+		userLeaderboard:      config.UserLeaderboard,
 		registrationReward:   config.RegistrationReward,
 		firstRechargeReward:  config.FirstRechargeReward,
 		cumulativeRecharge:   config.CumulativeRecharge,

services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler.go

@@ -1,16 +1,13 @@
 package activityapi
 
 import (
-	"context"
-	"database/sql"
-	"errors"
-	"fmt"
 	"net/http"
 	"strconv"
 	"strings"
 	"time"
 
 	"hyapp/pkg/appcode"
+	"hyapp/pkg/userleaderboard"
 	"hyapp/services/gateway-service/internal/auth"
 	"hyapp/services/gateway-service/internal/transport/http/httpkit"
 
@@ -19,13 +16,13 @@ import (
 )
 
 const (
-	leaderboardTypeSent     = "sent"
+	leaderboardTypeSent     = userleaderboard.BoardSent
-	leaderboardTypeReceived = "received"
+	leaderboardTypeReceived = userleaderboard.BoardReceived
 	leaderboardTypeRoom     = "room"
 
-	leaderboardPeriodToday = "today"
+	leaderboardPeriodToday = userleaderboard.PeriodToday
-	leaderboardPeriodWeek  = "week"
+	leaderboardPeriodWeek  = userleaderboard.PeriodWeek
-	leaderboardPeriodMonth = "month"
+	leaderboardPeriodMonth = userleaderboard.PeriodMonth
 
 	userLeaderboardMaxPageSize = 100
 )
@@ -98,36 +95,38 @@ func (h *Handler) listUserLeaderboards(writer http.ResponseWriter, request *http
 		h.listRoomGiftLeaderboards(writer, request, period, page, pageSize)
 		return
 	}
-	if h.walletDB == nil {
+	if h.userLeaderboard == nil {
 		httpkit.WriteError(writer, request, http.StatusServiceUnavailable, httpkit.CodeUpstreamError, "upstream service error")
 		return
 	}
 
 	now := time.Now().UTC()
-	start, end := userLeaderboardPeriodWindow(period, now)
-	startMS := start.UnixMilli()
-	endMS := end.UnixMilli()
 	app := appcode.FromContext(request.Context())
-	total, err := countUserLeaderboard(request.Context(), h.walletDB, app, boardType, startMS, endMS)
+	query := userleaderboard.Query{
-	if err != nil {
+		AppCode:   app,
-		httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error")
+		BoardType: boardType,
-		return
+		Period:    period,
-	}
+		Page:      int(page),
-	items, err := listUserLeaderboardItems(request.Context(), h.walletDB, app, boardType, startMS, endMS, int(pageSize), int((page-1)*pageSize))
+		PageSize:  int(pageSize),
+		Now:       now,
+	}
+	result, err := h.userLeaderboard.List(request.Context(), query)
 	if err != nil {
 		httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error")
 		return
 	}
+	items := userLeaderboardItemsFromRedisEntries(result.Items)
 
 	var myRank *userLeaderboardItem
 	viewerUserID := auth.UserIDFromContext(request.Context())
 	if viewerUserID > 0 && boardType != leaderboardTypeRoom {
-		item, found, err := userLeaderboardRankForUser(request.Context(), h.walletDB, app, boardType, startMS, endMS, viewerUserID)
+		entry, found, err := h.userLeaderboard.RankForUser(request.Context(), query, viewerUserID)
 		if err != nil {
 			httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error")
 			return
 		}
 		if found {
+			item := userLeaderboardItemFromRedisEntry(entry)
 			myRank = &item
 		}
 	}
@@ -138,14 +137,14 @@ func (h *Handler) listUserLeaderboards(writer http.ResponseWriter, request *http
 
 	httpkit.WriteOK(writer, request, userLeaderboardResponse{
 		Items:        items,
-		Total:        total,
+		Total:        result.Total,
 		Page:         page,
 		PageSize:     pageSize,
-		BoardType:    boardType,
+		BoardType:    result.BoardType,
-		Period:       period,
+		Period:       result.Period,
-		StartAtMS:    startMS,
+		StartAtMS:    result.StartAtMS,
-		EndAtMS:      endMS,
+		EndAtMS:      result.EndAtMS,
-		ServerTimeMS: now.UnixMilli(),
+		ServerTimeMS: result.ServerTimeMS,
 		MyRank:       myRank,
 	})
 }
@@ -180,219 +179,38 @@ func (h *Handler) listRoomGiftLeaderboards(writer http.ResponseWriter, request *
 
 func normalizeUserLeaderboardType(raw string) string {
 	switch strings.ToLower(strings.TrimSpace(raw)) {
-	case "", "sent", "send", "sender", "gift_sent", "user_sent":
-		return leaderboardTypeSent
-	case "received", "receive", "receiver", "gift_received", "user_received":
-		return leaderboardTypeReceived
 	case "room", "rooms", "room_gift", "room_gifts":
 		return leaderboardTypeRoom
 	default:
-		return ""
+		return userleaderboard.NormalizeBoardType(raw)
 	}
 }
 
 func normalizeUserLeaderboardPeriod(raw string) string {
-	switch strings.ToLower(strings.TrimSpace(raw)) {
+	return userleaderboard.NormalizePeriod(raw)
-	case "", "today", "day", "daily":
-		return leaderboardPeriodToday
-	case "week", "weekly":
-		return leaderboardPeriodWeek
-	case "month", "monthly":
-		return leaderboardPeriodMonth
-	default:
-		return ""
-	}
 }
 
 func userLeaderboardPeriodWindow(period string, now time.Time) (time.Time, time.Time) {
-	utcNow := now.UTC()
+	return userleaderboard.QueryWindow(period, now)
-	dayStart := time.Date(utcNow.Year(), utcNow.Month(), utcNow.Day(), 0, 0, 0, 0, time.UTC)
-	switch period {
-	case leaderboardPeriodWeek:
-		weekday := int(dayStart.Weekday())
-		if weekday == 0 {
-			weekday = 7
-		}
-		return dayStart.AddDate(0, 0, 1-weekday), utcNow
-	case leaderboardPeriodMonth:
-		return time.Date(utcNow.Year(), utcNow.Month(), 1, 0, 0, 0, 0, time.UTC), utcNow
-	default:
-		return dayStart, utcNow
-	}
 }
 
-func countUserLeaderboard(ctx context.Context, db *sql.DB, appCode string, boardType string, startMS int64, endMS int64) (int64, error) {
+func userLeaderboardItemsFromRedisEntries(entries []userleaderboard.Entry) []userLeaderboardItem {
-	subject, predicate := userLeaderboardDimension(boardType)
+	out := make([]userLeaderboardItem, 0, len(entries))
-	// App 排行榜价值优先使用真实送礼贡献 heat_value；旧交易只兜底扣费金额，不再读取历史 GIFT_POINT 字段。
+	for _, entry := range entries {
-	query := fmt.Sprintf(`
+		out = append(out, userLeaderboardItemFromRedisEntry(entry))
-		WITH gift_facts AS (
+	}
-			SELECT
+	return out
-				CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.sender_user_id')) AS SIGNED) AS sender_user_id,
-				CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.target_user_id')) AS SIGNED) AS target_user_id,
-				COALESCE(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.room_id')), '') AS room_id,
-				COALESCE(
-					CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.heat_value')) AS SIGNED),
-					CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.coin_spent')) AS SIGNED),
-					CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.charge_amount')) AS SIGNED),
-					0
-				) AS gift_value,
-				COALESCE(CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.gift_count')) AS SIGNED), 0) AS gift_count,
-				created_at_ms
-			FROM wallet_transactions
-			WHERE app_code = ? AND biz_type = 'gift_debit' AND status = 'succeeded'
-			  AND created_at_ms >= ? AND created_at_ms < ?
-		),
-		aggregated AS (
-			SELECT %s AS subject_id
-			FROM gift_facts
-			WHERE %s
-			GROUP BY %s
-		)
-		SELECT COUNT(*) FROM aggregated`, subject, predicate, subject)
-
-	var total int64
-	err := db.QueryRowContext(ctx, query, appCode, startMS, endMS).Scan(&total)
-	return total, err
 }
 
-func listUserLeaderboardItems(ctx context.Context, db *sql.DB, appCode string, boardType string, startMS int64, endMS int64, limit int, offset int) ([]userLeaderboardItem, error) {
+func userLeaderboardItemFromRedisEntry(entry userleaderboard.Entry) userLeaderboardItem {
-	subject, predicate := userLeaderboardDimension(boardType)
+	return userLeaderboardItem{
-	// 列表排序和值统计统一走 heat_value，保证 App 和后台排行榜不被历史积分字段污染。
+		Rank:             entry.Rank,
-	query := fmt.Sprintf(`
+		UserID:           entry.UserID,
-		WITH gift_facts AS (
+		GiftValue:        entry.GiftValue,
-			SELECT
+		GiftCount:        entry.GiftCount,
-				CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.sender_user_id')) AS SIGNED) AS sender_user_id,
+		TransactionCount: entry.TransactionCount,
-				CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.target_user_id')) AS SIGNED) AS target_user_id,
+		LastGiftAtMS:     entry.LastGiftAtMS,
-				COALESCE(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.room_id')), '') AS room_id,
+		User:             &userLeaderboardUser{UserID: entry.UserID},
-				COALESCE(
-					CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.heat_value')) AS SIGNED),
-					CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.coin_spent')) AS SIGNED),
-					CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.charge_amount')) AS SIGNED),
-					0
-				) AS gift_value,
-				COALESCE(CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.gift_count')) AS SIGNED), 0) AS gift_count,
-				created_at_ms
-			FROM wallet_transactions
-			WHERE app_code = ? AND biz_type = 'gift_debit' AND status = 'succeeded'
-			  AND created_at_ms >= ? AND created_at_ms < ?
-		),
-		aggregated AS (
-			SELECT %s AS subject_id,
-			       SUM(gift_value) AS gift_value,
-			       SUM(gift_count) AS gift_count,
-			       COUNT(*) AS transaction_count,
-			       MAX(created_at_ms) AS last_gift_at_ms
-			FROM gift_facts
-			WHERE %s
-			GROUP BY %s
-		),
-		ranked AS (
-			SELECT ROW_NUMBER() OVER (ORDER BY gift_value DESC, last_gift_at_ms DESC, subject_id ASC) AS rank_no,
-			       subject_id, gift_value, gift_count, transaction_count, last_gift_at_ms
-			FROM aggregated
-		)
-		SELECT rank_no, CAST(subject_id AS CHAR), gift_value, gift_count, transaction_count, last_gift_at_ms
-		FROM ranked
-		ORDER BY rank_no ASC
-		LIMIT ? OFFSET ?`, subject, predicate, subject)
-
-	rows, err := db.QueryContext(ctx, query, appCode, startMS, endMS, limit, offset)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-
-	items := make([]userLeaderboardItem, 0, limit)
-	for rows.Next() {
-		item, err := scanUserLeaderboardItem(rows, boardType)
-		if err != nil {
-			return nil, err
-		}
-		items = append(items, item)
-	}
-	return items, rows.Err()
-}
-
-func userLeaderboardRankForUser(ctx context.Context, db *sql.DB, appCode string, boardType string, startMS int64, endMS int64, userID int64) (userLeaderboardItem, bool, error) {
-	subject, predicate := userLeaderboardDimension(boardType)
-	// 当前用户名次查询只按真实贡献或扣费兜底统计，不再兼容 gift_point_added/gift_point_amount。
-	query := fmt.Sprintf(`
-		WITH gift_facts AS (
-			SELECT
-				CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.sender_user_id')) AS SIGNED) AS sender_user_id,
-				CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.target_user_id')) AS SIGNED) AS target_user_id,
-				COALESCE(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.room_id')), '') AS room_id,
-				COALESCE(
-					CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.heat_value')) AS SIGNED),
-					CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.coin_spent')) AS SIGNED),
-					CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.charge_amount')) AS SIGNED),
-					0
-				) AS gift_value,
-				COALESCE(CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.gift_count')) AS SIGNED), 0) AS gift_count,
-				created_at_ms
-			FROM wallet_transactions
-			WHERE app_code = ? AND biz_type = 'gift_debit' AND status = 'succeeded'
-			  AND created_at_ms >= ? AND created_at_ms < ?
-		),
-		aggregated AS (
-			SELECT %s AS subject_id,
-			       SUM(gift_value) AS gift_value,
-			       SUM(gift_count) AS gift_count,
-			       COUNT(*) AS transaction_count,
-			       MAX(created_at_ms) AS last_gift_at_ms
-			FROM gift_facts
-			WHERE %s
-			GROUP BY %s
-		),
-		ranked AS (
-			SELECT ROW_NUMBER() OVER (ORDER BY gift_value DESC, last_gift_at_ms DESC, subject_id ASC) AS rank_no,
-			       subject_id, gift_value, gift_count, transaction_count, last_gift_at_ms
-			FROM aggregated
-		)
-		SELECT rank_no, CAST(subject_id AS CHAR), gift_value, gift_count, transaction_count, last_gift_at_ms
-		FROM ranked
-		WHERE subject_id = ?
-		LIMIT 1`, subject, predicate, subject)
-
-	row := db.QueryRowContext(ctx, query, appCode, startMS, endMS, userID)
-	item, err := scanUserLeaderboardItem(row, boardType)
-	if err != nil {
-		if errors.Is(err, sql.ErrNoRows) {
-			return userLeaderboardItem{}, false, nil
-		}
-		return userLeaderboardItem{}, false, err
-	}
-	return item, true, nil
-}
-
-type userLeaderboardScanner interface {
-	Scan(dest ...any) error
-}
-
-func scanUserLeaderboardItem(scanner userLeaderboardScanner, boardType string) (userLeaderboardItem, error) {
-	var item userLeaderboardItem
-	var subject string
-	if err := scanner.Scan(&item.Rank, &subject, &item.GiftValue, &item.GiftCount, &item.TransactionCount, &item.LastGiftAtMS); err != nil {
-		return userLeaderboardItem{}, err
-	}
-	if boardType == leaderboardTypeRoom {
-		item.RoomID = subject
-		item.Room = &userLeaderboardRoom{RoomID: subject}
-		return item, nil
-	}
-	item.UserID = subject
-	item.User = &userLeaderboardUser{UserID: subject}
-	return item, nil
-}
-
-func userLeaderboardDimension(boardType string) (string, string) {
-	switch boardType {
-	case leaderboardTypeReceived:
-		return "target_user_id", "target_user_id > 0"
-	case leaderboardTypeRoom:
-		return "room_id", "room_id <> ''"
-	default:
-		return "sender_user_id", "sender_user_id > 0"
 	}
 }
 

services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler_test.go

@@ -2,9 +2,16 @@ package activityapi
 
 import (
 	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 	"time"
 
+	"hyapp/pkg/appcode"
+	"hyapp/pkg/userleaderboard"
+	"hyapp/services/gateway-service/internal/auth"
+
 	roomv1 "hyapp.local/api/proto/room/v1"
 )
 
@@ -81,6 +88,92 @@ func TestUserLeaderboardItemsFromRoomGiftLeaderboardUsesRoomCard(t *testing.T) {
 	}
 }
 
+func TestListUserLeaderboardsReadsRedisStore(t *testing.T) {
+	store := &fakeUserLeaderboardStore{
+		page: userleaderboard.Page{
+			Items: []userleaderboard.Entry{{
+				Rank:             1,
+				UserID:           "20002",
+				GiftValue:        900,
+				GiftCount:        3,
+				TransactionCount: 2,
+				LastGiftAtMS:     1782461920000,
+			}},
+			Total:        1,
+			BoardType:    leaderboardTypeSent,
+			Period:       leaderboardPeriodWeek,
+			StartAtMS:    1782172800000,
+			EndAtMS:      1782461920000,
+			ServerTimeMS: 1782461920000,
+		},
+		rank: userleaderboard.Entry{
+			Rank:             9,
+			UserID:           "10001",
+			GiftValue:        300,
+			GiftCount:        1,
+			TransactionCount: 1,
+			LastGiftAtMS:     1782460000000,
+		},
+		rankFound: true,
+	}
+	handler := &Handler{userLeaderboard: store}
+	request := httptest.NewRequest(http.MethodGet, "/api/v1/activities/user-leaderboards?board_type=sent&period=week&page=1&page_size=20", nil)
+	ctx := appcode.WithContext(request.Context(), "hyapp_room")
+	ctx = auth.WithUserID(ctx, 10001)
+	recorder := httptest.NewRecorder()
+
+	handler.listUserLeaderboards(recorder, request.WithContext(ctx))
+
+	if recorder.Code != http.StatusOK {
+		t.Fatalf("status = %d body=%s", recorder.Code, recorder.Body.String())
+	}
+	if store.listCalls != 1 || store.rankCalls != 1 {
+		t.Fatalf("store call count mismatch: list=%d rank=%d", store.listCalls, store.rankCalls)
+	}
+	if store.listQuery.AppCode != "hyapp_room" || store.listQuery.BoardType != leaderboardTypeSent || store.listQuery.Period != leaderboardPeriodWeek || store.listQuery.Page != 1 || store.listQuery.PageSize != 20 {
+		t.Fatalf("list query mismatch: %+v", store.listQuery)
+	}
+
+	var envelope struct {
+		Code string                  `json:"code"`
+		Data userLeaderboardResponse `json:"data"`
+	}
+	if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil {
+		t.Fatalf("decode response failed: %v", err)
+	}
+	if envelope.Code != "OK" || envelope.Data.Total != 1 || len(envelope.Data.Items) != 1 {
+		t.Fatalf("response mismatch: %+v", envelope)
+	}
+	if envelope.Data.Items[0].UserID != "20002" || envelope.Data.Items[0].GiftValue != 900 || envelope.Data.MyRank == nil || envelope.Data.MyRank.Rank != 9 {
+		t.Fatalf("leaderboard data mismatch: %+v my=%+v", envelope.Data.Items, envelope.Data.MyRank)
+	}
+}
+
+type fakeUserLeaderboardStore struct {
+	listCalls int
+	rankCalls int
+	listQuery userleaderboard.Query
+	rankQuery userleaderboard.Query
+	page      userleaderboard.Page
+	rank      userleaderboard.Entry
+	rankFound bool
+}
+
+func (f *fakeUserLeaderboardStore) List(_ context.Context, query userleaderboard.Query) (userleaderboard.Page, error) {
+	f.listCalls++
+	f.listQuery = query
+	return f.page, nil
+}
+
+func (f *fakeUserLeaderboardStore) RankForUser(_ context.Context, query userleaderboard.Query, userID int64) (userleaderboard.Entry, bool, error) {
+	f.rankCalls++
+	f.rankQuery = query
+	if userID != 10001 {
+		return userleaderboard.Entry{}, false, nil
+	}
+	return f.rank, f.rankFound, nil
+}
+
 type fakeUserLeaderboardRoomQueryClient struct {
 	snapshotResp *roomv1.GetRoomSnapshotResponse
 	snapshotErr  error

services/gateway-service/internal/transport/http/handler.go

@@ -2,7 +2,6 @@ package http
 
 import (
 	"context"
-	"database/sql"
 	"hyapp/services/gateway-service/internal/transport/http/httpkit"
 	"io"
 	"net/http"
@@ -12,6 +11,7 @@ import (
 	"hyapp/pkg/appcode"
 	"hyapp/pkg/imgroup"
 	"hyapp/pkg/tencentim"
+	"hyapp/pkg/userleaderboard"
 	"hyapp/services/gateway-service/internal/auth"
 	"hyapp/services/gateway-service/internal/client"
 	"hyapp/services/gateway-service/internal/transport/http/appapi"
@@ -57,7 +57,7 @@ type Handler struct {
 	gameClient           client.GameClient
 	statisticsClient     client.StatisticsClient
 	appConfigReader      appapi.ConfigReader
-	leaderboardWalletDB  *sql.DB
+	userLeaderboard      *userleaderboard.Store
 	tencentIM            TencentIMConfig
 	tencentIMImporter    TencentIMAccountImporter
 	tencentIMPublisher   TencentIMUserMessagePublisher
@@ -315,9 +315,9 @@ func (h *Handler) SetStatisticsClient(statisticsClient client.StatisticsClient)
 	h.statisticsClient = statisticsClient
 }
 
-// SetLeaderboardWalletDB 注入 App 榜单只读钱包库连接。
+// SetUserLeaderboardStore 注入 App 用户送礼榜 Redis 读模型；gateway 只读聚合结果。
-func (h *Handler) SetLeaderboardWalletDB(db *sql.DB) {
+func (h *Handler) SetUserLeaderboardStore(store *userleaderboard.Store) {
-	h.leaderboardWalletDB = db
+	h.userLeaderboard = store
 }
 
 // SetAppConfigReader 注入后台 App 配置读取依赖。

services/gateway-service/internal/transport/http/router.go

@@ -80,7 +80,7 @@ func (h *Handler) Routes(jwtVerifier *auth.Verifier) http.Handler {
 		UserProfileClient:    h.userProfileClient,
 		RoomQueryClient:      h.roomQueryClient,
 		WalletClient:         h.walletClient,
-		WalletDB:             h.leaderboardWalletDB,
+		UserLeaderboard:      h.userLeaderboard,
 		RegistrationReward:   h.registrationReward,
 		FirstRechargeReward:  h.firstRechargeReward,
 		CumulativeRecharge:   h.cumulativeRecharge,

services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql

@@ -32,9 +32,21 @@ CREATE TABLE IF NOT EXISTS wallet_transactions (
   PRIMARY KEY (app_code, transaction_id),
   UNIQUE KEY uk_wallet_tx_command (app_code, command_id),
   KEY idx_wallet_tx_biz_time (app_code, biz_type, created_at_ms),
+  KEY idx_wallet_tx_biz_status_time (app_code, biz_type, status, created_at_ms),
   KEY idx_wallet_tx_external_ref (app_code, external_ref)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='钱包交易主表';
 
+-- App 用户送礼榜历史应急查询会按 app_code/biz_type/status/created_at_ms 过滤；
+-- 老索引缺 status 时会在 10GB 级 wallet_transactions 上读取大量失败/非成功交易再做 JSON 聚合。
+SET @ddl := IF(
+  (SELECT COUNT(*) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'wallet_transactions' AND INDEX_NAME = 'idx_wallet_tx_biz_status_time') = 0,
+  'ALTER TABLE wallet_transactions ADD INDEX idx_wallet_tx_biz_status_time (app_code, biz_type, status, created_at_ms)',
+  'SELECT 1'
+);
+PREPARE stmt FROM @ddl;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;
+
 CREATE TABLE IF NOT EXISTS wallet_entries (
   app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码，用于多租户隔离',
   entry_id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY COMMENT '流水 ID',