373 lines
15 KiB
Go
373 lines
15 KiB
Go
package gameking
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"net/http"
|
||
"strconv"
|
||
"time"
|
||
|
||
"chatapp3-golang/internal/model"
|
||
"chatapp3-golang/internal/utils"
|
||
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
const (
|
||
settlementDispatchBatch = 4
|
||
settlementDueBatch = 10
|
||
dailySettlementDueBatch = 10
|
||
staleDeliveryBatch = 20
|
||
)
|
||
|
||
// Settle 在结算截点后分别冻结 TYCOON/VICTORIOUS 的 OVERALL Top30，并发送 PENDING 奖励。
|
||
// 重复调用不会重建快照，也不会自动重试 FAILED/UNKNOWN。
|
||
func (s *Service) Settle(ctx context.Context, activityID int64) error {
|
||
if activityID <= 0 {
|
||
return NewAppError(http.StatusBadRequest, "invalid_activity_id", "activityId is required")
|
||
}
|
||
if err := s.freezeSettlement(ctx, activityID); err != nil {
|
||
return err
|
||
}
|
||
var items []model.YumiGameKingDeliveryItem
|
||
if err := s.db.WithContext(ctx).
|
||
Where("activity_id = ? AND owner_type = ? AND delivery_status = ?", activityID, OwnerSettlement, DeliveryPending).
|
||
Order("id ASC").Limit(settlementDispatchBatch).Find(&items).Error; err != nil {
|
||
return err
|
||
}
|
||
// 人工结算同样限制每次最多发送 4 组，最坏外部调用约 80 秒并为 DB 留出余量；
|
||
// 剩余 PENDING 由后续 settle-due 批次继续恢复，FAILED/UNKNOWN 不会自动发送。
|
||
var lastErr error
|
||
for _, item := range items {
|
||
if err := s.dispatchDeliveryItem(ctx, item.ID, false); err != nil {
|
||
lastErr = err
|
||
}
|
||
}
|
||
return lastErr
|
||
}
|
||
|
||
func (s *Service) freezeSettlement(ctx context.Context, activityID int64) error {
|
||
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
var activity model.YumiGameKingActivity
|
||
// 排他活动锁与事件/抽奖共享锁构成冻结闸门：先进入的业务事务完成后，状态在同一
|
||
// 事务切为 PROCESSING；此后新事件会在共享锁放行后看到非 NOT_STARTED 并拒绝入账。
|
||
if err := withWriteLock(tx).Where("id = ?", activityID).First(&activity).Error; err != nil {
|
||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||
return NewAppError(http.StatusNotFound, "activity_not_found", "game king activity was not found")
|
||
}
|
||
return err
|
||
}
|
||
now := time.Now()
|
||
if !activity.Enabled {
|
||
return NewAppError(http.StatusConflict, "activity_disabled", "disabled activity cannot be settled")
|
||
}
|
||
if now.Before(activity.SettlementTime) {
|
||
return NewAppError(http.StatusConflict, "settlement_not_due", "activity settlement time has not arrived")
|
||
}
|
||
if activity.SettlementStatus != SettlementNotStarted {
|
||
return nil
|
||
}
|
||
if err := tx.Model(&model.YumiGameKingActivity{}).Where("id = ?", activity.ID).
|
||
Updates(map[string]any{"settlement_status": SettlementProcessing, "update_time": now}).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
var rewards []model.YumiGameKingRankReward
|
||
if err := tx.Where("activity_id = ? AND ranking_period = ?", activity.ID, rankingPeriodOverall).Order("ranking_type ASC, start_rank ASC").Find(&rewards).Error; err != nil {
|
||
return err
|
||
}
|
||
if len(rewards) != len(fixedOverallRankRanges)*2 {
|
||
return NewAppError(http.StatusConflict, "rank_reward_config_invalid", "six reward tiers are required for each ranking")
|
||
}
|
||
created := 0
|
||
for _, rankingType := range []string{rankingTypeTycoon, rankingTypeVictorious} {
|
||
scoreColumn, reachedColumn := "total_consumed", "consume_reached_time"
|
||
if rankingType == rankingTypeVictorious {
|
||
scoreColumn, reachedColumn = "total_won", "win_reached_time"
|
||
}
|
||
var users []model.YumiGameKingUser
|
||
if err := tx.Where("activity_id = ? AND "+scoreColumn+" > 0", activity.ID).
|
||
Order(scoreColumn + " DESC, " + reachedColumn + " ASC, user_id ASC").Limit(30).Find(&users).Error; err != nil {
|
||
return err
|
||
}
|
||
boardRewards := make([]model.YumiGameKingRankReward, 0, len(fixedOverallRankRanges))
|
||
for _, reward := range rewards {
|
||
if reward.RankingType == rankingType {
|
||
boardRewards = append(boardRewards, reward)
|
||
}
|
||
}
|
||
for index, user := range users {
|
||
rank := index + 1
|
||
reward, ok := rankRewardFor(boardRewards, rank)
|
||
if !ok {
|
||
return NewAppError(http.StatusConflict, "rank_reward_config_invalid", "rank reward tier does not cover Top30")
|
||
}
|
||
recordID, err := utils.NextID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
itemID, err := utils.NextID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
businessNo := fmt.Sprintf("YUMI_GAME_KING_SETTLEMENT:%d:%s:%s:%d", activity.ID, rankingPeriodOverall, rankingType, rank)
|
||
score := user.TotalConsumed
|
||
if rankingType == rankingTypeVictorious {
|
||
score = user.TotalWon
|
||
}
|
||
record := model.YumiGameKingSettlementRecord{
|
||
ID: recordID, ActivityID: activity.ID, RankingType: rankingType, RankingPeriod: rankingPeriodOverall, PeriodKey: rankingPeriodOverall, UserID: user.UserID, RankNo: rank,
|
||
TotalConsumed: user.TotalConsumed, Score: score, RankRewardID: reward.ID, ResourceGroupID: reward.ResourceGroupID,
|
||
BusinessNo: businessNo, DeliveryStatus: DeliveryPending, CreateTime: now, UpdateTime: now,
|
||
}
|
||
if err := tx.Create(&record).Error; err != nil {
|
||
return err
|
||
}
|
||
if err := tx.Create(&model.YumiGameKingDeliveryItem{
|
||
ID: itemID, OwnerType: OwnerSettlement, OwnerID: record.ID, ActivityID: activity.ID,
|
||
UserID: user.UserID, ResourceGroupID: reward.ResourceGroupID,
|
||
DeliveryStatus: DeliveryPending, CreateTime: now, UpdateTime: now,
|
||
}).Error; err != nil {
|
||
return err
|
||
}
|
||
created++
|
||
}
|
||
}
|
||
if created == 0 {
|
||
return tx.Model(&model.YumiGameKingActivity{}).Where("id = ?", activity.ID).
|
||
Updates(map[string]any{"settlement_status": SettlementCompleted, "update_time": now}).Error
|
||
}
|
||
return nil
|
||
})
|
||
}
|
||
|
||
func rankRewardFor(rows []model.YumiGameKingRankReward, rank int) (model.YumiGameKingRankReward, bool) {
|
||
for _, row := range rows {
|
||
if rank >= row.StartRank && rank <= row.EndRank {
|
||
return row, true
|
||
}
|
||
}
|
||
return model.YumiGameKingRankReward{}, false
|
||
}
|
||
|
||
// nextDailySettlement 返回活动时区内最早尚未建立门闩且已经到达结算等待时间的自然日。
|
||
// 每次只推进一个连续日期，空榜也会写门闩，因此 cron 中断数日后仍能顺序补齐且不会
|
||
// 给结算截点之后才迟到的历史流水重新制造获奖机会。
|
||
func (s *Service) nextDailySettlement(ctx context.Context, activity model.YumiGameKingActivity, now time.Time) (time.Time, time.Time, bool, error) {
|
||
_, location, err := normalizeTimezone(activity.Timezone)
|
||
if err != nil {
|
||
return time.Time{}, time.Time{}, false, err
|
||
}
|
||
statDate := dateKey(activity.StartTime, location)
|
||
var latest model.YumiGameKingPeriodSettlement
|
||
err = s.db.WithContext(ctx).
|
||
Where("activity_id = ? AND period_type = ?", activity.ID, rankingPeriodDaily).
|
||
// YYYY-MM-DD 的 period_key 与日期顺序一致，并可直接利用唯一索引前缀，避免
|
||
// 每次 cron 为寻找最新门闩额外排序该活动的全部历史日期。
|
||
Order("period_key DESC").Limit(1).First(&latest).Error
|
||
if err == nil {
|
||
statDate = latest.StatDate.AddDate(0, 0, 1)
|
||
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
|
||
return time.Time{}, time.Time{}, false, err
|
||
}
|
||
lastDate := dateKey(activity.EndTime.Add(-time.Nanosecond), location)
|
||
if statDate.After(lastDate) {
|
||
return time.Time{}, time.Time{}, false, nil
|
||
}
|
||
dueTime := dailySettlementDueTime(activity, statDate, location)
|
||
return statDate, dueTime, !now.Before(dueTime), nil
|
||
}
|
||
|
||
func dailySettlementDueTime(activity model.YumiGameKingActivity, statDate time.Time, location *time.Location) time.Time {
|
||
periodEndLocal := time.Date(statDate.Year(), statDate.Month(), statDate.Day()+1, 0, 0, 0, 0, location)
|
||
dueBase := periodEndLocal
|
||
if activity.EndTime.Before(dueBase) {
|
||
dueBase = activity.EndTime
|
||
}
|
||
return dueBase.Add(time.Duration(activity.SettlementDelayMinutes) * time.Minute)
|
||
}
|
||
|
||
// freezeDailySettlement 在活动排他锁内先建立 DAILY 门闩，再读取日聚合 Top3。
|
||
// 事件事务持有同一活动的共享锁并检查门闩，所以锁释放后的迟到事件只补总榜。
|
||
func (s *Service) freezeDailySettlement(ctx context.Context, activityID int64, statDate, dueTime time.Time) (bool, error) {
|
||
createdGate := false
|
||
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
var activity model.YumiGameKingActivity
|
||
if err := withWriteLock(tx).Where("id = ?", activityID).First(&activity).Error; err != nil {
|
||
return err
|
||
}
|
||
if !activity.Enabled {
|
||
return NewAppError(http.StatusConflict, "activity_disabled", "disabled activity cannot be settled")
|
||
}
|
||
if time.Now().Before(dueTime) {
|
||
return NewAppError(http.StatusConflict, "daily_settlement_not_due", "daily settlement time has not arrived")
|
||
}
|
||
periodKey := statDate.Format("2006-01-02")
|
||
var existing model.YumiGameKingPeriodSettlement
|
||
findErr := tx.Where("activity_id = ? AND period_type = ? AND period_key = ?", activity.ID, rankingPeriodDaily, periodKey).First(&existing).Error
|
||
if findErr == nil {
|
||
return nil
|
||
}
|
||
if !errors.Is(findErr, gorm.ErrRecordNotFound) {
|
||
return findErr
|
||
}
|
||
gateID, err := utils.NextID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
now := time.Now()
|
||
gate := model.YumiGameKingPeriodSettlement{
|
||
ID: gateID, ActivityID: activity.ID, PeriodType: rankingPeriodDaily,
|
||
PeriodKey: periodKey, StatDate: statDate, SnapshotDueTime: dueTime,
|
||
Status: SettlementProcessing, CreateTime: now, UpdateTime: now,
|
||
}
|
||
if err := tx.Create(&gate).Error; err != nil {
|
||
return err
|
||
}
|
||
createdGate = true
|
||
|
||
var rewards []model.YumiGameKingRankReward
|
||
if err := tx.Where("activity_id = ? AND ranking_period = ?", activity.ID, rankingPeriodDaily).
|
||
Order("ranking_type ASC, start_rank ASC").Find(&rewards).Error; err != nil {
|
||
return err
|
||
}
|
||
// 迁移前已经开始的旧活动没有日榜奖励配置；空配置只建立完成门闩，不擅自
|
||
// 复制总榜奖励。新活动启用前则由配置校验强制两个榜单各三档。
|
||
if len(rewards) == 0 {
|
||
return tx.Model(&model.YumiGameKingPeriodSettlement{}).Where("id = ?", gate.ID).
|
||
Updates(map[string]any{"status": SettlementCompleted, "update_time": now}).Error
|
||
}
|
||
if len(rewards) != len(fixedDailyRankRanges)*2 {
|
||
return NewAppError(http.StatusConflict, "daily_rank_reward_config_invalid", "DAILY Top1, Top2 and Top3 rewards are required for each ranking")
|
||
}
|
||
|
||
createdRecords := 0
|
||
for _, rankingType := range []string{rankingTypeTycoon, rankingTypeVictorious} {
|
||
scoreColumn, reachedColumn := "total_consumed", "consume_reached_time"
|
||
if rankingType == rankingTypeVictorious {
|
||
scoreColumn, reachedColumn = "total_won", "win_reached_time"
|
||
}
|
||
var users []model.YumiGameKingUserDaily
|
||
if err := tx.Where("activity_id = ? AND stat_date = ? AND "+scoreColumn+" > 0", activity.ID, statDate).
|
||
Order(scoreColumn + " DESC, " + reachedColumn + " ASC, user_id ASC").Limit(3).Find(&users).Error; err != nil {
|
||
return err
|
||
}
|
||
boardRewards := make([]model.YumiGameKingRankReward, 0, len(fixedDailyRankRanges))
|
||
for _, reward := range rewards {
|
||
if reward.RankingType == rankingType {
|
||
boardRewards = append(boardRewards, reward)
|
||
}
|
||
}
|
||
for index, user := range users {
|
||
rank := index + 1
|
||
reward, ok := rankRewardFor(boardRewards, rank)
|
||
if !ok {
|
||
return NewAppError(http.StatusConflict, "daily_rank_reward_config_invalid", "DAILY reward tier does not cover Top3")
|
||
}
|
||
recordID, idErr := utils.NextID()
|
||
if idErr != nil {
|
||
return idErr
|
||
}
|
||
itemID, idErr := utils.NextID()
|
||
if idErr != nil {
|
||
return idErr
|
||
}
|
||
score := user.TotalConsumed
|
||
if rankingType == rankingTypeVictorious {
|
||
score = user.TotalWon
|
||
}
|
||
storedDate := statDate
|
||
record := model.YumiGameKingSettlementRecord{
|
||
ID: recordID, ActivityID: activity.ID, RankingType: rankingType,
|
||
RankingPeriod: rankingPeriodDaily, PeriodKey: periodKey, StatDate: &storedDate,
|
||
UserID: user.UserID, RankNo: rank, TotalConsumed: user.TotalConsumed, Score: score,
|
||
RankRewardID: reward.ID, ResourceGroupID: reward.ResourceGroupID,
|
||
BusinessNo: fmt.Sprintf("YUMI_GAME_KING_SETTLEMENT:%d:%s:%s:%s:%d", activity.ID, rankingPeriodDaily, periodKey, rankingType, rank),
|
||
DeliveryStatus: DeliveryPending, CreateTime: now, UpdateTime: now,
|
||
}
|
||
if err := tx.Create(&record).Error; err != nil {
|
||
return err
|
||
}
|
||
if err := tx.Create(&model.YumiGameKingDeliveryItem{
|
||
ID: itemID, OwnerType: OwnerSettlement, OwnerID: record.ID, ActivityID: activity.ID,
|
||
UserID: user.UserID, ResourceGroupID: reward.ResourceGroupID,
|
||
DeliveryStatus: DeliveryPending, CreateTime: now, UpdateTime: now,
|
||
}).Error; err != nil {
|
||
return err
|
||
}
|
||
createdRecords++
|
||
}
|
||
}
|
||
if createdRecords == 0 {
|
||
return tx.Model(&model.YumiGameKingPeriodSettlement{}).Where("id = ?", gate.ID).
|
||
Updates(map[string]any{"status": SettlementCompleted, "update_time": now}).Error
|
||
}
|
||
return nil
|
||
})
|
||
return createdGate && err == nil, err
|
||
}
|
||
|
||
// SettleDue 是外部 cron 的唯一批量触发业务方法；本仓库不启动 ticker。
|
||
func (s *Service) SettleDue(ctx context.Context) (*SettleDueResult, error) {
|
||
// 先把过期 PROCESSING 保守转 UNKNOWN；只做状态修复，不盲发。
|
||
if err := s.markStaleProcessingUnknown(ctx, staleDeliveryBatch); err != nil {
|
||
return nil, err
|
||
}
|
||
result := &SettleDueResult{FailedIDs: []string{}}
|
||
now := time.Now()
|
||
var dailyActivities []model.YumiGameKingActivity
|
||
if err := s.db.WithContext(ctx).
|
||
Where("enabled = ? AND start_time < ?", true, now).
|
||
Order("start_time ASC, id ASC").Find(&dailyActivities).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
for _, activity := range dailyActivities {
|
||
if result.DailyScanned >= dailySettlementDueBatch {
|
||
break
|
||
}
|
||
statDate, dueTime, due, err := s.nextDailySettlement(ctx, activity, now)
|
||
if err != nil {
|
||
result.FailedIDs = append(result.FailedIDs, fmt.Sprintf("%d:%s", activity.ID, rankingPeriodDaily))
|
||
continue
|
||
}
|
||
if !due {
|
||
continue
|
||
}
|
||
result.Scanned++
|
||
result.DailyScanned++
|
||
created, err := s.freezeDailySettlement(ctx, activity.ID, statDate, dueTime)
|
||
if err != nil {
|
||
result.FailedIDs = append(result.FailedIDs, fmt.Sprintf("%d:%s:%s", activity.ID, rankingPeriodDaily, statDate.Format("2006-01-02")))
|
||
continue
|
||
}
|
||
if created {
|
||
result.Settled++
|
||
result.DailySettled++
|
||
}
|
||
}
|
||
var activities []model.YumiGameKingActivity
|
||
if err := s.db.WithContext(ctx).
|
||
Where("enabled = ? AND settlement_time <= ? AND settlement_status = ?", true, now, SettlementNotStarted).
|
||
Order("settlement_time ASC, id ASC").Limit(settlementDueBatch).Find(&activities).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
for _, activity := range activities {
|
||
result.Scanned++
|
||
result.OverallScanned++
|
||
// 批量端点只快速冻结排名并生成 PENDING，不在单个活动内串行发送 Top30。
|
||
if err := s.freezeSettlement(ctx, activity.ID); err != nil {
|
||
result.FailedIDs = append(result.FailedIDs, strconv.FormatInt(activity.ID, 10)+":"+rankingPeriodOverall)
|
||
continue
|
||
}
|
||
result.Settled++
|
||
result.OverallSettled++
|
||
}
|
||
// 每次最多领取 4 个全局 PENDING；单项 dispatch 硬超时 20 秒，外部调用上界约
|
||
// 80 秒，为冻结/状态修复留约 40 秒，chatapp-cron 应配置 >=120 秒超时。
|
||
if _, err := s.RecoverPendingDeliveries(ctx, settlementDispatchBatch); err != nil {
|
||
return nil, err
|
||
}
|
||
return result, nil
|
||
}
|