552 lines
17 KiB
Go
552 lines
17 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"log"
|
||
"math"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"chatapp-cron/internal/config"
|
||
"chatapp-cron/internal/integration"
|
||
"chatapp-cron/internal/model"
|
||
"chatapp-cron/internal/repo"
|
||
"chatapp-cron/internal/util"
|
||
|
||
"gorm.io/gorm"
|
||
"gorm.io/gorm/clause"
|
||
)
|
||
|
||
const (
|
||
// weekStarRewardStatusPending 表示已生成发奖记录，但尚未调用下游发奖接口。
|
||
weekStarRewardStatusPending = "PENDING"
|
||
// weekStarRewardStatusSuccess 表示奖励已经成功发放。
|
||
weekStarRewardStatusSuccess = "SUCCESS"
|
||
// weekStarRewardStatusFailed 表示调用发奖失败，等待后台重试。
|
||
weekStarRewardStatusFailed = "FAILED"
|
||
// weekStarRewardOrigin 是发奖来源标识，透传给 Java 奖励系统用于审计。
|
||
weekStarRewardOrigin = "ACTIVITY_REWARD"
|
||
)
|
||
|
||
// WeekStarRankingUserView 是结算过程和查询过程共用的榜单展示视图。
|
||
type WeekStarRankingUserView struct {
|
||
UserID int64
|
||
UserAvatar string
|
||
UserNickname string
|
||
Account string
|
||
CountryCode string
|
||
CountryName string
|
||
Rank int
|
||
ScoreGold int64
|
||
}
|
||
|
||
// weekStarUserProfile 是内部组装用的轻量用户资料对象。
|
||
type weekStarUserProfile struct {
|
||
UserID int64
|
||
Account string
|
||
UserAvatar string
|
||
UserNickname string
|
||
CountryCode string
|
||
CountryName string
|
||
}
|
||
|
||
// weekStarRankScore 保存 Redis 排行结果归一化后的分数和名次。
|
||
type weekStarRankScore struct {
|
||
UserID int64
|
||
ScoreGold int64
|
||
Rank int
|
||
}
|
||
|
||
// WeekStarSettlementService 负责扫描周期、冻结榜单快照并发放奖励。
|
||
type WeekStarSettlementService struct {
|
||
cfg config.Config
|
||
repo *repo.Repository
|
||
java *integration.Client
|
||
location *time.Location
|
||
}
|
||
|
||
// NewWeekStarSettlementService 创建周榜结算服务，并初始化活动使用的时区。
|
||
func NewWeekStarSettlementService(cfg config.Config, repository *repo.Repository, javaClient *integration.Client) *WeekStarSettlementService {
|
||
location, err := time.LoadLocation(strings.TrimSpace(cfg.WeekStarTimezone))
|
||
if err != nil {
|
||
// 配置异常时兜底为 UTC，避免服务启动失败。
|
||
location = time.UTC
|
||
}
|
||
return &WeekStarSettlementService{
|
||
cfg: cfg,
|
||
repo: repository,
|
||
java: javaClient,
|
||
location: location,
|
||
}
|
||
}
|
||
|
||
// Start 启动周榜结算扫描和失败补发消费任务。
|
||
func (s *WeekStarSettlementService) Start(ctx context.Context) {
|
||
s.startSettlementLoop(ctx)
|
||
s.startRetryWorker(ctx)
|
||
}
|
||
|
||
// startSettlementLoop 按配置周期持续检查是否存在待结算周榜。
|
||
func (s *WeekStarSettlementService) startSettlementLoop(ctx context.Context) {
|
||
interval := s.cfg.WeekStarSettleInterval
|
||
if interval <= 0 {
|
||
interval = time.Minute
|
||
}
|
||
go func() {
|
||
ticker := time.NewTicker(interval)
|
||
defer ticker.Stop()
|
||
for {
|
||
if ctx.Err() != nil {
|
||
return
|
||
}
|
||
if err := s.SettleDueCycles(ctx); err != nil && ctx.Err() == nil {
|
||
log.Printf("week star settlement scan failed: %v", err)
|
||
}
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
}
|
||
}
|
||
}()
|
||
}
|
||
|
||
// SettleDueCycles 扫描所有活动配置，并尝试结算已结束但尚未冻结的周期。
|
||
func (s *WeekStarSettlementService) SettleDueCycles(ctx context.Context) error {
|
||
var configs []model.WeekStarActivityConfig
|
||
if err := s.repo.DB.WithContext(ctx).Order("id desc").Find(&configs).Error; err != nil {
|
||
return err
|
||
}
|
||
now := time.Now().In(s.location)
|
||
var lastErr error
|
||
for index := range configs {
|
||
if err := s.settleConfigCycles(ctx, &configs[index], now); err != nil {
|
||
lastErr = err
|
||
log.Printf("week star settle config failed. configId=%d err=%v", configs[index].ID, err)
|
||
}
|
||
}
|
||
return lastErr
|
||
}
|
||
|
||
// settleConfigCycles 逐周遍历单个活动配置的时间窗，挑出所有已经结束的周期执行结算。
|
||
func (s *WeekStarSettlementService) settleConfigCycles(ctx context.Context, configRow *model.WeekStarActivityConfig, now time.Time) error {
|
||
if configRow == nil {
|
||
return nil
|
||
}
|
||
// 周榜按周一 00:00 切周，因此先把活动开始时间对齐到所在周的起点。
|
||
startCycle := s.cycleStart(s.fromStorageWallClock(configRow.StartAt))
|
||
endAt := s.fromStorageWallClock(configRow.EndAt)
|
||
var lastErr error
|
||
for cycleStart := startCycle; cycleStart.Before(endAt); cycleStart = cycleStart.AddDate(0, 0, 7) {
|
||
cycleEnd := cycleStart.AddDate(0, 0, 7)
|
||
periodStart := maxTime(cycleStart, s.fromStorageWallClock(configRow.StartAt))
|
||
periodEnd := minTime(cycleEnd, endAt)
|
||
if !periodEnd.After(periodStart) {
|
||
continue
|
||
}
|
||
if periodEnd.After(now) {
|
||
break
|
||
}
|
||
if err := s.settleCycle(ctx, configRow, cycleStart, periodStart, periodEnd); err != nil {
|
||
lastErr = err
|
||
}
|
||
}
|
||
return lastErr
|
||
}
|
||
|
||
// settleCycle 结算单个周期，流程包含加锁、读取实时榜、冻结快照、生成发奖记录并实际发奖。
|
||
func (s *WeekStarSettlementService) settleCycle(ctx context.Context, configRow *model.WeekStarActivityConfig, cycleStart, periodStart, periodEnd time.Time) error {
|
||
cycleKey := s.cycleKey(cycleStart)
|
||
lockKey := s.settleLockKey(configRow.ID, cycleKey)
|
||
|
||
// 使用 Redis 分布式锁保证同一周期只会被一个实例执行结算。
|
||
acquired, err := s.repo.Redis.SetNX(ctx, lockKey, 1, 5*time.Minute).Result()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !acquired {
|
||
return nil
|
||
}
|
||
|
||
// 只要快照或发奖记录已存在，就视为该周期已经执行过结算。
|
||
var snapshotCount int64
|
||
if err := s.repo.DB.WithContext(ctx).
|
||
Model(&model.WeekStarRankSnapshot{}).
|
||
Where("config_id = ? AND cycle_key = ?", configRow.ID, cycleKey).
|
||
Count(&snapshotCount).Error; err != nil {
|
||
return err
|
||
}
|
||
var rewardCount int64
|
||
if err := s.repo.DB.WithContext(ctx).
|
||
Model(&model.WeekStarRewardRecord{}).
|
||
Where("config_id = ? AND cycle_key = ?", configRow.ID, cycleKey).
|
||
Count(&rewardCount).Error; err != nil {
|
||
return err
|
||
}
|
||
if snapshotCount > 0 || rewardCount > 0 {
|
||
return nil
|
||
}
|
||
|
||
// 结算只冻结 Top3，用于历史榜单和后续奖励发放。
|
||
liveRanking, err := s.loadLiveRanking(ctx, configRow.ID, cycleKey, 3)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if len(liveRanking) == 0 {
|
||
return nil
|
||
}
|
||
|
||
var rewardConfigs []model.WeekStarActivityRewardConfig
|
||
if err := s.repo.DB.WithContext(ctx).
|
||
Where("config_id = ?", configRow.ID).
|
||
Order("`rank` asc").
|
||
Find(&rewardConfigs).Error; err != nil {
|
||
return err
|
||
}
|
||
rewardConfigMap := make(map[int]model.WeekStarActivityRewardConfig, len(rewardConfigs))
|
||
for _, row := range rewardConfigs {
|
||
rewardConfigMap[row.Rank] = row
|
||
}
|
||
|
||
// 先构造要落库的快照和发奖记录，避免边查边写导致状态不一致。
|
||
now := time.Now()
|
||
snapshots := make([]model.WeekStarRankSnapshot, 0, len(liveRanking))
|
||
rewardRecords := make([]model.WeekStarRewardRecord, 0, len(liveRanking))
|
||
for _, item := range liveRanking {
|
||
snapshotID, err := util.NextID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
snapshots = append(snapshots, model.WeekStarRankSnapshot{
|
||
ID: snapshotID,
|
||
ConfigID: configRow.ID,
|
||
CycleKey: cycleKey,
|
||
Rank: item.Rank,
|
||
PeriodStartAt: s.toStorageWallClock(periodStart),
|
||
PeriodEndAt: s.toStorageWallClock(periodEnd),
|
||
UserID: item.UserID,
|
||
Account: item.Account,
|
||
UserAvatar: item.UserAvatar,
|
||
UserNickname: item.UserNickname,
|
||
CountryCode: item.CountryCode,
|
||
CountryName: item.CountryName,
|
||
ScoreGold: item.ScoreGold,
|
||
CreateTime: now,
|
||
UpdateTime: now,
|
||
})
|
||
|
||
// 某个名次未配置奖励时，只保留快照，不生成发奖记录。
|
||
rewardConfig, ok := rewardConfigMap[item.Rank]
|
||
if !ok {
|
||
continue
|
||
}
|
||
recordID, err := util.NextID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
rewardRecords = append(rewardRecords, model.WeekStarRewardRecord{
|
||
ID: recordID,
|
||
ConfigID: configRow.ID,
|
||
CycleKey: cycleKey,
|
||
Rank: item.Rank,
|
||
UserID: item.UserID,
|
||
Account: item.Account,
|
||
UserAvatar: item.UserAvatar,
|
||
UserNickname: item.UserNickname,
|
||
CountryCode: item.CountryCode,
|
||
CountryName: item.CountryName,
|
||
RewardGroupID: rewardConfig.RewardGroupID,
|
||
RewardGroupName: rewardConfig.RewardGroupName,
|
||
ScoreGold: item.ScoreGold,
|
||
TrackID: recordID,
|
||
Status: weekStarRewardStatusPending,
|
||
RetryCount: 0,
|
||
CreateTime: now,
|
||
UpdateTime: now,
|
||
})
|
||
}
|
||
|
||
// 快照和发奖记录在同一个事务里落库，确保结算结果成对出现。
|
||
if err := s.repo.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
if len(snapshots) > 0 {
|
||
if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&snapshots).Error; err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if len(rewardRecords) > 0 {
|
||
if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&rewardRecords).Error; err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}); err != nil {
|
||
return err
|
||
}
|
||
|
||
var pendingRecords []model.WeekStarRewardRecord
|
||
if err := s.repo.DB.WithContext(ctx).
|
||
Where("config_id = ? AND cycle_key = ? AND status = ?", configRow.ID, cycleKey, weekStarRewardStatusPending).
|
||
Order("`rank` asc").
|
||
Find(&pendingRecords).Error; err != nil {
|
||
return err
|
||
}
|
||
var lastErr error
|
||
for index := range pendingRecords {
|
||
// 发奖逐条执行，单条失败不会阻塞同周期其他名次继续发放。
|
||
_, err := s.dispatchRewardRecord(ctx, &pendingRecords[index], false)
|
||
if err != nil {
|
||
lastErr = err
|
||
}
|
||
}
|
||
return lastErr
|
||
}
|
||
|
||
// dispatchRewardRecord 调用 Java 奖励接口发奖，并回写发奖成功或失败状态。
|
||
// handled=true 表示本次任务已经完成状态落库，调用方可以安全确认消息；
|
||
// handled=false 表示处理过程中断，调用方应保留消息供后续继续消费。
|
||
func (s *WeekStarSettlementService) dispatchRewardRecord(ctx context.Context, record *model.WeekStarRewardRecord, isRetry bool) (bool, error) {
|
||
if record == nil {
|
||
return true, nil
|
||
}
|
||
var configRow model.WeekStarActivityConfig
|
||
if err := s.repo.DB.WithContext(ctx).Where("id = ?", record.ConfigID).First(&configRow).Error; err != nil {
|
||
return false, err
|
||
}
|
||
|
||
now := time.Now()
|
||
retryCount := record.RetryCount
|
||
if isRetry {
|
||
retryCount++
|
||
}
|
||
err := s.java.SendActivityReward(ctx, integration.SendActivityRewardRequest{
|
||
TrackID: record.TrackID,
|
||
Origin: weekStarRewardOrigin,
|
||
SysOrigin: configRow.SysOrigin,
|
||
SourceGroupID: record.RewardGroupID,
|
||
AcceptUserID: record.UserID,
|
||
})
|
||
if err != nil {
|
||
// 发奖失败时保留错误信息，方便后台做失败补发。
|
||
updates := map[string]any{
|
||
"status": weekStarRewardStatusFailed,
|
||
"retry_count": retryCount,
|
||
"last_error": truncateString(err.Error(), 1024),
|
||
"update_time": now,
|
||
}
|
||
if updateErr := s.repo.DB.WithContext(context.Background()).
|
||
Model(&model.WeekStarRewardRecord{}).
|
||
Where("id = ?", record.ID).
|
||
Updates(updates).Error; updateErr != nil {
|
||
return false, updateErr
|
||
}
|
||
return true, err
|
||
}
|
||
|
||
if err := s.repo.DB.WithContext(ctx).
|
||
Model(&model.WeekStarRewardRecord{}).
|
||
Where("id = ?", record.ID).
|
||
Updates(map[string]any{
|
||
"status": weekStarRewardStatusSuccess,
|
||
"retry_count": retryCount,
|
||
"last_error": "",
|
||
"sent_at": now,
|
||
"update_time": now,
|
||
}).Error; err != nil {
|
||
return false, err
|
||
}
|
||
return true, nil
|
||
}
|
||
|
||
// loadLiveRanking 从 Redis 实时榜读取前 N 名，并补齐用户资料供快照/发奖使用。
|
||
func (s *WeekStarSettlementService) loadLiveRanking(ctx context.Context, configID int64, cycleKey string, limit int64) ([]WeekStarRankingUserView, error) {
|
||
if configID <= 0 || strings.TrimSpace(cycleKey) == "" || limit <= 0 {
|
||
return []WeekStarRankingUserView{}, nil
|
||
}
|
||
|
||
redisKey := s.rankRedisKey(configID, cycleKey)
|
||
rankScores, err := s.repo.Redis.ZRevRangeWithScores(ctx, redisKey, 0, limit-1).Result()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(rankScores) == 0 {
|
||
return []WeekStarRankingUserView{}, nil
|
||
}
|
||
|
||
userIDs := make([]int64, 0, len(rankScores))
|
||
normalizedScores := make([]weekStarRankScore, 0, len(rankScores))
|
||
for index, item := range rankScores {
|
||
userID, ok := parseRedisMemberInt64(item.Member)
|
||
if !ok || userID <= 0 {
|
||
// 跳过异常 member，避免脏数据影响正常结算。
|
||
continue
|
||
}
|
||
userIDs = append(userIDs, userID)
|
||
normalizedScores = append(normalizedScores, weekStarRankScore{
|
||
UserID: userID,
|
||
ScoreGold: int64(math.Round(item.Score)),
|
||
Rank: index + 1,
|
||
})
|
||
}
|
||
|
||
profileMap, err := s.lookupUserProfiles(ctx, userIDs)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
result := make([]WeekStarRankingUserView, 0, len(normalizedScores))
|
||
for _, item := range normalizedScores {
|
||
profile := profileMap[item.UserID]
|
||
result = append(result, WeekStarRankingUserView{
|
||
UserID: item.UserID,
|
||
UserAvatar: profile.UserAvatar,
|
||
UserNickname: profile.UserNickname,
|
||
Account: profile.Account,
|
||
CountryCode: profile.CountryCode,
|
||
CountryName: profile.CountryName,
|
||
Rank: item.Rank,
|
||
ScoreGold: item.ScoreGold,
|
||
})
|
||
}
|
||
return result, nil
|
||
}
|
||
|
||
// lookupUserProfiles 优先查本地用户表，查不到再回源 Java 用户资料接口兜底。
|
||
func (s *WeekStarSettlementService) lookupUserProfiles(ctx context.Context, userIDs []int64) (map[int64]weekStarUserProfile, error) {
|
||
result := make(map[int64]weekStarUserProfile, len(userIDs))
|
||
if len(userIDs) == 0 {
|
||
return result, nil
|
||
}
|
||
|
||
var rows []model.UserBaseInfo
|
||
if err := s.repo.DB.WithContext(ctx).Where("id IN ?", userIDs).Find(&rows).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
for _, row := range rows {
|
||
result[row.ID] = weekStarUserProfile{
|
||
UserID: row.ID,
|
||
Account: row.Account,
|
||
UserAvatar: row.UserAvatar,
|
||
UserNickname: row.UserNickname,
|
||
CountryCode: row.CountryCode,
|
||
CountryName: row.CountryName,
|
||
}
|
||
}
|
||
|
||
for _, userID := range userIDs {
|
||
if _, exists := result[userID]; exists {
|
||
continue
|
||
}
|
||
profile, err := s.java.GetUserProfile(ctx, userID)
|
||
if err != nil {
|
||
// 用户资料回源失败时仍保留空壳对象，避免整个结算流程中断。
|
||
result[userID] = weekStarUserProfile{UserID: userID}
|
||
continue
|
||
}
|
||
result[userID] = weekStarUserProfile{
|
||
UserID: userID,
|
||
UserAvatar: profile.UserAvatar,
|
||
UserNickname: profile.UserNickname,
|
||
}
|
||
}
|
||
return result, nil
|
||
}
|
||
|
||
// cycleStart 计算某个时间所在周的周一 00:00，作为周榜周期起点。
|
||
func (s *WeekStarSettlementService) cycleStart(t time.Time) time.Time {
|
||
local := t.In(s.location)
|
||
weekday := int(local.Weekday())
|
||
if weekday == 0 {
|
||
weekday = 7
|
||
}
|
||
startDay := local.AddDate(0, 0, -(weekday - 1))
|
||
return time.Date(startDay.Year(), startDay.Month(), startDay.Day(), 0, 0, 0, 0, s.location)
|
||
}
|
||
|
||
// cycleKey 把周期开始时间格式化为稳定的周期标识，例如 20260414。
|
||
func (s *WeekStarSettlementService) cycleKey(start time.Time) string {
|
||
return start.In(s.location).Format("20060102")
|
||
}
|
||
|
||
// toStorageWallClock 把活动时区时间转换为本地库里约定的墙上时间存储格式。
|
||
func (s *WeekStarSettlementService) toStorageWallClock(t time.Time) time.Time {
|
||
local := t.In(s.location)
|
||
return time.Date(
|
||
local.Year(),
|
||
local.Month(),
|
||
local.Day(),
|
||
local.Hour(),
|
||
local.Minute(),
|
||
local.Second(),
|
||
local.Nanosecond(),
|
||
s.location,
|
||
)
|
||
}
|
||
|
||
// fromStorageWallClock 把库里保存的墙上时间恢复为活动时区时间，便于业务计算。
|
||
func (s *WeekStarSettlementService) fromStorageWallClock(t time.Time) time.Time {
|
||
return time.Date(
|
||
t.Year(),
|
||
t.Month(),
|
||
t.Day(),
|
||
t.Hour(),
|
||
t.Minute(),
|
||
t.Second(),
|
||
t.Nanosecond(),
|
||
s.location,
|
||
)
|
||
}
|
||
|
||
// rankRedisKey 返回某个活动某个周期的实时排行 Redis Key。
|
||
func (s *WeekStarSettlementService) rankRedisKey(configID int64, cycleKey string) string {
|
||
return fmt.Sprintf("week-star:rank:%d:%s", configID, cycleKey)
|
||
}
|
||
|
||
// settleLockKey 返回某个活动某个周期的结算锁 Redis Key。
|
||
func (s *WeekStarSettlementService) settleLockKey(configID int64, cycleKey string) string {
|
||
return fmt.Sprintf("week-star:settle-lock:%d:%s", configID, cycleKey)
|
||
}
|
||
|
||
// parseRedisMemberInt64 兼容多种 Redis member 类型，并统一解析为用户 ID。
|
||
func parseRedisMemberInt64(member any) (int64, bool) {
|
||
switch value := member.(type) {
|
||
case string:
|
||
parsed, err := strconv.ParseInt(strings.TrimSpace(value), 10, 64)
|
||
return parsed, err == nil
|
||
case []byte:
|
||
parsed, err := strconv.ParseInt(strings.TrimSpace(string(value)), 10, 64)
|
||
return parsed, err == nil
|
||
case int64:
|
||
return value, true
|
||
case int:
|
||
return int64(value), true
|
||
default:
|
||
return 0, false
|
||
}
|
||
}
|
||
|
||
// truncateString 按指定长度截断错误信息，避免超出数据库字段限制。
|
||
func truncateString(value string, limit int) string {
|
||
if limit <= 0 || len(value) <= limit {
|
||
return value
|
||
}
|
||
return value[:limit]
|
||
}
|
||
|
||
// maxTime 返回两个时间里的较大值。
|
||
func maxTime(left, right time.Time) time.Time {
|
||
if left.After(right) {
|
||
return left
|
||
}
|
||
return right
|
||
}
|
||
|
||
// minTime 返回两个时间里的较小值。
|
||
func minTime(left, right time.Time) time.Time {
|
||
if left.Before(right) {
|
||
return left
|
||
}
|
||
return right
|
||
}
|