502 lines
17 KiB
Go
502 lines
17 KiB
Go
// Package userleaderboard owns the Redis read model for App user 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"
|
||
BoardGame = "game"
|
||
|
||
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")
|
||
|
||
var ErrRegionRequired = errors.New("user leaderboard region is required")
|
||
|
||
// Store keeps user leaderboard buckets in Redis zsets plus per-user hashes.
|
||
// Gift facts populate sent/received, while successful game debits populate game; gateway only reads this projection.
|
||
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
|
||
SenderRegionID int64
|
||
TargetUserID int64
|
||
TargetRegionID int64
|
||
GiftValue int64
|
||
GiftCount int64
|
||
OccurredAtMS int64
|
||
DirectGift bool
|
||
}
|
||
|
||
// GameEvent is the successful game debit fact used by the game-spend leaderboard.
|
||
// CoinSpent follows the existing game growth metric and deliberately excludes payout, refund and net-win values.
|
||
type GameEvent struct {
|
||
AppCode string
|
||
EventID string
|
||
UserID int64
|
||
RegionID int64
|
||
CoinSpent int64
|
||
OccurredAtMS int64
|
||
}
|
||
|
||
// Query describes one App leaderboard page. Now must be UTC-compatible; zero means current UTC time.
|
||
type Query struct {
|
||
AppCode string
|
||
RegionID int64
|
||
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()
|
||
regional := appcode.UsesRegionalLeaderboards(event.AppCode)
|
||
updates := make([]leaderboardUpdate, 0, 12)
|
||
for _, period := range []string{PeriodToday, PeriodWeek, PeriodMonth} {
|
||
// 全 App key 在兼容窗口继续写入，保证 gateway/activity 滚动发布期间旧实例仍能读到连续数据；
|
||
// Lalu/Fami 的新查询只读取 region key，因此 legacy 全局榜不会再造成跨区混排。
|
||
updates = append(updates,
|
||
s.updateFor(event.AppCode, 0, BoardSent, period, event.SenderUserID, occurredAt),
|
||
s.updateFor(event.AppCode, 0, BoardReceived, period, event.TargetUserID, occurredAt),
|
||
)
|
||
if regional && event.SenderRegionID > 0 {
|
||
updates = append(updates, s.updateFor(event.AppCode, event.SenderRegionID, BoardSent, period, event.SenderUserID, occurredAt))
|
||
}
|
||
if regional && event.TargetRegionID > 0 {
|
||
updates = append(updates, s.updateFor(event.AppCode, event.TargetRegionID, BoardReceived, period, event.TargetUserID, occurredAt))
|
||
}
|
||
}
|
||
return s.applyValueEvent(ctx, event.AppCode, event.EventID, event.GiftValue, event.GiftCount, event.OccurredAtMS, updates)
|
||
}
|
||
|
||
// ApplyGameEvent projects one successful game debit into UTC day/week/month game buckets.
|
||
// The Redis hash keeps the legacy gift_* field names because the public leaderboard JSON already exposes them;
|
||
// for board_type=game, gift_value means game_spend_coin, gift_count stays zero and transaction_count counts debit orders.
|
||
func (s *Store) ApplyGameEvent(ctx context.Context, event GameEvent) (bool, error) {
|
||
if s == nil || s.client == nil {
|
||
return false, ErrNotConfigured
|
||
}
|
||
event = normalizeGameEvent(event)
|
||
if event.AppCode == "" || event.EventID == "" || event.UserID <= 0 || event.CoinSpent <= 0 {
|
||
return false, nil
|
||
}
|
||
|
||
occurredAt := time.UnixMilli(event.OccurredAtMS).UTC()
|
||
regional := appcode.UsesRegionalLeaderboards(event.AppCode)
|
||
updates := make([]leaderboardUpdate, 0, 6)
|
||
for _, period := range []string{PeriodToday, PeriodWeek, PeriodMonth} {
|
||
updates = append(updates, s.updateFor(event.AppCode, 0, BoardGame, period, event.UserID, occurredAt))
|
||
if regional && event.RegionID > 0 {
|
||
updates = append(updates, s.updateFor(event.AppCode, event.RegionID, BoardGame, period, event.UserID, occurredAt))
|
||
}
|
||
}
|
||
return s.applyValueEvent(ctx, event.AppCode, event.EventID, event.CoinSpent, 0, event.OccurredAtMS, updates)
|
||
}
|
||
|
||
// 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)
|
||
if appcode.UsesRegionalLeaderboards(query.AppCode) && query.RegionID <= 0 {
|
||
return Page{}, ErrRegionRequired
|
||
}
|
||
start, end := QueryWindow(query.Period, query.Now)
|
||
key := s.scopedScoreKey(query.AppCode, query.RegionID, 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.RegionID, 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)
|
||
if appcode.UsesRegionalLeaderboards(query.AppCode) && query.RegionID <= 0 {
|
||
return Entry{}, false, ErrRegionRequired
|
||
}
|
||
start, _ := QueryWindow(query.Period, query.Now)
|
||
key := s.scopedScoreKey(query.AppCode, query.RegionID, 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.RegionID, 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
|
||
case BoardGame, "games", "gaming":
|
||
return BoardGame
|
||
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, regionID int64, board string, period string, userID int64, occurredAt time.Time) leaderboardUpdate {
|
||
start, periodEnd := periodBounds(period, occurredAt)
|
||
member := strconv.FormatInt(userID, 10)
|
||
return leaderboardUpdate{
|
||
scoreKey: s.scopedScoreKey(app, regionID, board, period, start),
|
||
itemKey: s.scopedItemKey(app, regionID, board, period, start, member),
|
||
member: member,
|
||
ttl: retentionTTL(period, start, periodEnd),
|
||
}
|
||
}
|
||
|
||
func (s *Store) entryFromHash(ctx context.Context, app string, regionID int64, board string, period string, start time.Time, userID string, scoreFallback int64) (Entry, error) {
|
||
values, err := s.client.HGetAll(ctx, s.scopedItemKey(app, regionID, 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) scopedScoreKey(app string, regionID int64, board string, period string, start time.Time) string {
|
||
if !appcode.UsesRegionalLeaderboards(app) || regionID <= 0 {
|
||
return s.scoreKey(app, board, period, start)
|
||
}
|
||
return fmt.Sprintf("%s:%s:region:%d:%s:%s:%s:scores", s.keyPrefix, appcode.Normalize(app), regionID, board, period, bucketID(period, start))
|
||
}
|
||
|
||
func (s *Store) scopedItemKey(app string, regionID int64, board string, period string, start time.Time, userID string) string {
|
||
if !appcode.UsesRegionalLeaderboards(app) || regionID <= 0 {
|
||
return s.itemKey(app, board, period, start, userID)
|
||
}
|
||
return fmt.Sprintf("%s:%s:region:%d:%s:%s:%s:users:%s", s.keyPrefix, appcode.Normalize(app), regionID, 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 normalizeGameEvent(event GameEvent) GameEvent {
|
||
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 (s *Store) applyValueEvent(ctx context.Context, app string, eventID string, valueDelta int64, countDelta int64, occurredAtMS int64, updates []leaderboardUpdate) (bool, error) {
|
||
keys := []string{s.dedupeKey(app, eventID)}
|
||
args := []any{
|
||
strconv.FormatInt(int64(defaultEventDedupeTTL/time.Millisecond), 10),
|
||
strconv.FormatInt(valueDelta, 10),
|
||
strconv.FormatInt(countDelta, 10),
|
||
strconv.FormatInt(occurredAtMS, 10),
|
||
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))
|
||
}
|
||
|
||
// Dedupe and all period updates execute without interleaving; after a successful script, relay retries cannot double-count.
|
||
result, err := applyValueEventScript.Run(ctx, s.client, keys, args...).Int64()
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
return result == 1, nil
|
||
}
|
||
|
||
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 applyValueEventScript = redis.NewScript(`
|
||
if redis.call("EXISTS", KEYS[1]) == 1 then
|
||
return 0
|
||
end
|
||
redis.call("PSETEX", KEYS[1], ARGV[1], "1")
|
||
local value_delta = tonumber(ARGV[2])
|
||
local count_delta = 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", value_delta)
|
||
redis.call("HINCRBY", item_key, "gift_count", count_delta)
|
||
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
|
||
`)
|