From 902ca5dc3c61a9b8b96d58376ac00aaa46bf65ae Mon Sep 17 00:00:00 2001 From: zhx Date: Fri, 26 Jun 2026 17:56:30 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0redis=E8=81=9A=E5=90=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docker-compose.yml | 2 + pkg/rocketmqx/rocketmqx.go | 5 + pkg/userleaderboard/store.go | 416 ++++++++++++++++++ pkg/userleaderboard/store_test.go | 52 +++ .../configs/config.docker.yaml | 7 + .../configs/config.tencent.example.yaml | 7 + services/activity-service/configs/config.yaml | 7 + services/activity-service/internal/app/app.go | 17 +- services/activity-service/internal/app/mq.go | 39 ++ .../activity-service/internal/app/services.go | 5 +- .../internal/app/user_leaderboard.go | 25 ++ .../internal/app/wallet_events.go | 88 ++++ .../wallet_events_user_leaderboard_test.go | 105 +++++ .../internal/config/config.go | 31 +- .../internal/config/config_test.go | 12 +- .../configs/config.docker.yaml | 7 +- .../configs/config.tencent.example.yaml | 7 +- services/gateway-service/configs/config.yaml | 7 +- services/gateway-service/internal/app/app.go | 78 ++-- .../gateway-service/internal/config/config.go | 25 +- .../internal/config/config_test.go | 6 + .../transport/http/activityapi/handler.go | 14 +- .../activityapi/user_leaderboard_handler.go | 268 ++--------- .../user_leaderboard_handler_test.go | 93 ++++ .../internal/transport/http/handler.go | 10 +- .../internal/transport/http/router.go | 2 +- .../mysql/initdb/001_wallet_service.sql | 12 + 27 files changed, 1055 insertions(+), 292 deletions(-) create mode 100644 pkg/userleaderboard/store.go create mode 100644 pkg/userleaderboard/store_test.go create mode 100644 services/activity-service/internal/app/user_leaderboard.go create mode 100644 services/activity-service/internal/app/wallet_events_user_leaderboard_test.go diff --git a/docker-compose.yml b/docker-compose.yml index bba19aeb..ed779c93 100644 --- a/docker-compose.yml +++ b/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: diff --git a/pkg/rocketmqx/rocketmqx.go b/pkg/rocketmqx/rocketmqx.go index 3dd58b59..483178f9 100644 --- a/pkg/rocketmqx/rocketmqx.go +++ b/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 diff --git a/pkg/userleaderboard/store.go b/pkg/userleaderboard/store.go new file mode 100644 index 00000000..0ec9adb1 --- /dev/null +++ b/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 +`) diff --git a/pkg/userleaderboard/store_test.go b/pkg/userleaderboard/store_test.go new file mode 100644 index 00000000..cef09ca2 --- /dev/null +++ b/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) + } +} diff --git a/services/activity-service/configs/config.docker.yaml b/services/activity-service/configs/config.docker.yaml index 629c51aa..4fb277dc 100644 --- a/services/activity-service/configs/config.docker.yaml +++ b/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 diff --git a/services/activity-service/configs/config.tencent.example.yaml b/services/activity-service/configs/config.tencent.example.yaml index 17476e63..8fe523d2 100644 --- a/services/activity-service/configs/config.tencent.example.yaml +++ b/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 diff --git a/services/activity-service/configs/config.yaml b/services/activity-service/configs/config.yaml index ccb3b10f..d17344ad 100644 --- a/services/activity-service/configs/config.yaml +++ b/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 diff --git a/services/activity-service/internal/app/app.go b/services/activity-service/internal/app/app.go index c856f1d9..a15f3383 100644 --- a/services/activity-service/internal/app/app.go +++ b/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() + } }) } diff --git a/services/activity-service/internal/app/mq.go b/services/activity-service/internal/app/mq.go index b1ed6d1e..044f36ac 100644 --- a/services/activity-service/internal/app/mq.go +++ b/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{ diff --git a/services/activity-service/internal/app/services.go b/services/activity-service/internal/app/services.go index 5ea125e5..4bdcffb8 100644 --- a/services/activity-service/internal/app/services.go +++ b/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 } diff --git a/services/activity-service/internal/app/user_leaderboard.go b/services/activity-service/internal/app/user_leaderboard.go new file mode 100644 index 00000000..a1c067a6 --- /dev/null +++ b/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 +} diff --git a/services/activity-service/internal/app/wallet_events.go b/services/activity-service/internal/app/wallet_events.go index 1a26b1c5..cc982131 100644 --- a/services/activity-service/internal/app/wallet_events.go +++ b/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 + } +} diff --git a/services/activity-service/internal/app/wallet_events_user_leaderboard_test.go b/services/activity-service/internal/app/wallet_events_user_leaderboard_test.go new file mode 100644 index 00000000..f5b03cbc --- /dev/null +++ b/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 +} diff --git a/services/activity-service/internal/config/config.go b/services/activity-service/internal/config/config.go index b9e3bd38..9a89404a 100644 --- a/services/activity-service/internal/config/config.go +++ b/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 } diff --git a/services/activity-service/internal/config/config_test.go b/services/activity-service/internal/config/config_test.go index 15a53d65..515ab57c 100644 --- a/services/activity-service/internal/config/config_test.go +++ b/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 { - 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) + 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 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) + } } diff --git a/services/gateway-service/configs/config.docker.yaml b/services/gateway-service/configs/config.docker.yaml index 9c851327..c92b350a 100644 --- a/services/gateway-service/configs/config.docker.yaml +++ b/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 计算。 - wallet_mysql_dsn: "hyapp:hyapp@tcp(mysql:3306)/hyapp_wallet?parseTime=true&charset=utf8mb4&loc=UTC" + # 活动用户榜单只读 activity-service 写入的 Redis 聚合结果；时间窗口统一按 UTC 计算。 + redis_addr: "redis:6379" + redis_password: "" + redis_db: 0 + key_prefix: "activity:user_leaderboard" auth_rate_limit: # public auth 固定窗口频控；Redis 不可用时 gateway 启动失败，避免失去入口保护。 enabled: true diff --git a/services/gateway-service/configs/config.tencent.example.yaml b/services/gateway-service/configs/config.tencent.example.yaml index 8e14bc58..55b239c8 100644 --- a/services/gateway-service/configs/config.tencent.example.yaml +++ b/services/gateway-service/configs/config.tencent.example.yaml @@ -62,8 +62,11 @@ app_config: key_prefix: "gateway:app_config:" cache_ttl: "5m" leaderboard: - # 活动用户榜单只读钱包送礼事实；线上建议配置只读账号。 - wallet_mysql_dsn: "TENCENT_MYSQL_USER:TENCENT_MYSQL_PASSWORD@tcp(TENCENT_MYSQL_HOST:3306)/hyapp_wallet?parseTime=true&charset=utf8mb4&loc=UTC" + # 活动用户榜单只读 activity-service 写入的 Redis 聚合结果；必须和 activity-service 使用同一 key_prefix。 + 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 diff --git a/services/gateway-service/configs/config.yaml b/services/gateway-service/configs/config.yaml index 18c140c7..2b9fcef1 100644 --- a/services/gateway-service/configs/config.yaml +++ b/services/gateway-service/configs/config.yaml @@ -58,8 +58,11 @@ app_config: key_prefix: "gateway:app_config:" cache_ttl: "5m" leaderboard: - # 活动用户榜单只读 wallet_transactions 的送礼事实；时间窗口统一按 UTC 计算。 - wallet_mysql_dsn: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_wallet?parseTime=true&charset=utf8mb4&loc=UTC" + # 活动用户榜单只读 activity-service 写入的 Redis 聚合结果；时间窗口统一按 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 diff --git a/services/gateway-service/internal/app/app.go b/services/gateway-service/internal/app/app.go index eb480372..96450704 100644 --- a/services/gateway-service/internal/app/app.go +++ b/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 { - _ = leaderboardWalletDB.Close() + if redisClose != nil { + _ = 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 { - _ = leaderboardWalletDB.Close() + if redisClose != nil { + _ = 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()) } diff --git a/services/gateway-service/internal/config/config.go b/services/gateway-service/internal/config/config.go index 7cd5fe88..8438011f 100644 --- a/services/gateway-service/internal/config/config.go +++ b/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 == "" { diff --git a/services/gateway-service/internal/config/config_test.go b/services/gateway-service/internal/config/config_test.go index 2a95b3af..2d7a725a 100644 --- a/services/gateway-service/internal/config/config_test.go +++ b/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) } diff --git a/services/gateway-service/internal/transport/http/activityapi/handler.go b/services/gateway-service/internal/transport/http/activityapi/handler.go index a711fc02..08a0f486 100644 --- a/services/gateway-service/internal/transport/http/activityapi/handler.go +++ b/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, diff --git a/services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler.go b/services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler.go index f1d1ad02..014ce87e 100644 --- a/services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler.go +++ b/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" - leaderboardTypeReceived = "received" + leaderboardTypeSent = userleaderboard.BoardSent + leaderboardTypeReceived = userleaderboard.BoardReceived leaderboardTypeRoom = "room" - leaderboardPeriodToday = "today" - leaderboardPeriodWeek = "week" - leaderboardPeriodMonth = "month" + leaderboardPeriodToday = userleaderboard.PeriodToday + leaderboardPeriodWeek = userleaderboard.PeriodWeek + 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) - if err != nil { - httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error") - return - } - items, err := listUserLeaderboardItems(request.Context(), h.walletDB, app, boardType, startMS, endMS, int(pageSize), int((page-1)*pageSize)) + query := userleaderboard.Query{ + AppCode: app, + BoardType: boardType, + Period: period, + Page: int(page), + 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, - Period: period, - StartAtMS: startMS, - EndAtMS: endMS, - ServerTimeMS: now.UnixMilli(), + BoardType: result.BoardType, + Period: result.Period, + StartAtMS: result.StartAtMS, + EndAtMS: result.EndAtMS, + 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)) { - case "", "today", "day", "daily": - return leaderboardPeriodToday - case "week", "weekly": - return leaderboardPeriodWeek - case "month", "monthly": - return leaderboardPeriodMonth - default: - return "" - } + return userleaderboard.NormalizePeriod(raw) } func userLeaderboardPeriodWindow(period string, now time.Time) (time.Time, time.Time) { - utcNow := now.UTC() - 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 - } + return userleaderboard.QueryWindow(period, now) } -func countUserLeaderboard(ctx context.Context, db *sql.DB, appCode string, boardType string, startMS int64, endMS int64) (int64, error) { - subject, predicate := userLeaderboardDimension(boardType) - // App 排行榜价值优先使用真实送礼贡献 heat_value；旧交易只兜底扣费金额，不再读取历史 GIFT_POINT 字段。 - 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 - 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 userLeaderboardItemsFromRedisEntries(entries []userleaderboard.Entry) []userLeaderboardItem { + out := make([]userLeaderboardItem, 0, len(entries)) + for _, entry := range entries { + out = append(out, userLeaderboardItemFromRedisEntry(entry)) + } + return out } -func listUserLeaderboardItems(ctx context.Context, db *sql.DB, appCode string, boardType string, startMS int64, endMS int64, limit int, offset int) ([]userLeaderboardItem, error) { - subject, predicate := userLeaderboardDimension(boardType) - // 列表排序和值统计统一走 heat_value，保证 App 和后台排行榜不被历史积分字段污染。 - 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 - 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" +func userLeaderboardItemFromRedisEntry(entry userleaderboard.Entry) userLeaderboardItem { + return userLeaderboardItem{ + Rank: entry.Rank, + UserID: entry.UserID, + GiftValue: entry.GiftValue, + GiftCount: entry.GiftCount, + TransactionCount: entry.TransactionCount, + LastGiftAtMS: entry.LastGiftAtMS, + User: &userLeaderboardUser{UserID: entry.UserID}, } } diff --git a/services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler_test.go b/services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler_test.go index 1fc4402b..557178ef 100644 --- a/services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler_test.go +++ b/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 diff --git a/services/gateway-service/internal/transport/http/handler.go b/services/gateway-service/internal/transport/http/handler.go index 6047ed27..2e1158d3 100644 --- a/services/gateway-service/internal/transport/http/handler.go +++ b/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 榜单只读钱包库连接。 -func (h *Handler) SetLeaderboardWalletDB(db *sql.DB) { - h.leaderboardWalletDB = db +// SetUserLeaderboardStore 注入 App 用户送礼榜 Redis 读模型；gateway 只读聚合结果。 +func (h *Handler) SetUserLeaderboardStore(store *userleaderboard.Store) { + h.userLeaderboard = store } // SetAppConfigReader 注入后台 App 配置读取依赖。 diff --git a/services/gateway-service/internal/transport/http/router.go b/services/gateway-service/internal/transport/http/router.go index 5c7343dd..84c81c62 100644 --- a/services/gateway-service/internal/transport/http/router.go +++ b/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, diff --git a/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql b/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql index b5913f6c..c0ed32ca 100644 --- a/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql +++ b/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',