417 lines
13 KiB
Go
417 lines
13 KiB
Go
// 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
|
|
`)
|