219 lines
7.6 KiB
Go
219 lines
7.6 KiB
Go
package wheel
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"log/slog"
|
||
"strings"
|
||
"time"
|
||
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/logx"
|
||
"hyapp/pkg/xerr"
|
||
domain "hyapp/services/activity-service/internal/domain/wheel"
|
||
)
|
||
|
||
// RewardWorkerOptions 定义补偿 worker 的租户、锁和重试边界。
|
||
type RewardWorkerOptions struct {
|
||
WorkerID string
|
||
AppCodes []string
|
||
PollInterval time.Duration
|
||
BatchSize int
|
||
LockTTL time.Duration
|
||
MaxRetry int
|
||
}
|
||
|
||
// RewardWorkerBatchResult 只统计本轮 claim 后的终态；失败项已经释放为 retryable/failed，不会静默丢失。
|
||
type RewardWorkerBatchResult struct {
|
||
Claimed int
|
||
Delivered int
|
||
Failed int
|
||
}
|
||
|
||
type wheelRewardOutboxPayload struct {
|
||
AppCode string `json:"app_code"`
|
||
DrawID string `json:"draw_id"`
|
||
DrawIDs []string `json:"draw_ids"`
|
||
CommandID string `json:"command_id"`
|
||
UserID int64 `json:"user_id"`
|
||
WheelID string `json:"wheel_id"`
|
||
SelectedTierID string `json:"selected_tier_id"`
|
||
RewardType string `json:"reward_type"`
|
||
RewardID string `json:"reward_id"`
|
||
RewardCount int64 `json:"reward_count"`
|
||
RewardCoins int64 `json:"reward_coins"`
|
||
Rewards []domain.DrawReward `json:"rewards"`
|
||
VisibleRegionID int64 `json:"visible_region_id"`
|
||
}
|
||
|
||
// RunRewardWorker 持续逐 App 分区补偿 WheelRewardSettlement。
|
||
// AppCode 必须写进 context 后再 claim 和调用钱包，避免后台根 context 被 appcode 默认值悄悄收敛成 lalu。
|
||
func (s *Service) RunRewardWorker(ctx context.Context, options RewardWorkerOptions) {
|
||
options = normalizeRewardWorkerOptions(options)
|
||
ticker := time.NewTicker(options.PollInterval)
|
||
defer ticker.Stop()
|
||
for {
|
||
if _, err := s.ProcessPendingRewards(ctx, options); err != nil && ctx.Err() == nil {
|
||
logx.Error(ctx, "wheel_reward_outbox_batch_failed", err, slog.String("worker_id", options.WorkerID))
|
||
}
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
}
|
||
}
|
||
}
|
||
|
||
// ProcessPendingRewards 执行一轮有界补偿。
|
||
// 钱包 RPC 使用 draw_id 派生的稳定 command_id；钱包 owner 会在同一事务内先查既有交易/权益及 request hash，命中则返回原收据，未命中才创建，因此 activity 只在收齐所有收据后标 delivered。
|
||
func (s *Service) ProcessPendingRewards(ctx context.Context, options RewardWorkerOptions) (RewardWorkerBatchResult, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return RewardWorkerBatchResult{}, err
|
||
}
|
||
options = normalizeRewardWorkerOptions(options)
|
||
var result RewardWorkerBatchResult
|
||
var batchErr error
|
||
for _, appCodeValue := range options.AppCodes {
|
||
if ctx.Err() != nil {
|
||
return result, errors.Join(batchErr, ctx.Err())
|
||
}
|
||
appCtx := appcode.WithContext(ctx, appCodeValue)
|
||
nowMS := s.now().UTC().UnixMilli()
|
||
records, err := s.repository.ClaimWheelRewardOutbox(appCtx, options.WorkerID, nowMS, nowMS+options.LockTTL.Milliseconds(), options.BatchSize, options.MaxRetry)
|
||
if err != nil {
|
||
batchErr = errors.Join(batchErr, fmt.Errorf("claim app=%s: %w", appCodeValue, err))
|
||
continue
|
||
}
|
||
result.Claimed += len(records)
|
||
for _, record := range records {
|
||
if err := s.processRewardOutboxRecord(appCtx, record, options); err != nil {
|
||
result.Failed++
|
||
batchErr = errors.Join(batchErr, fmt.Errorf("process app=%s outbox=%s: %w", appCodeValue, record.OutboxID, err))
|
||
continue
|
||
}
|
||
result.Delivered++
|
||
}
|
||
}
|
||
return result, batchErr
|
||
}
|
||
|
||
func (s *Service) processRewardOutboxRecord(ctx context.Context, record domain.RewardOutboxRecord, options RewardWorkerOptions) error {
|
||
payload, err := decodeWheelRewardOutboxPayload(record)
|
||
if err == nil {
|
||
var fulfilled domain.DrawResult
|
||
fulfilled, err = s.fulfillWheelReward(ctx, domain.DrawCommand{
|
||
CommandID: payload.CommandID,
|
||
WheelID: payload.WheelID,
|
||
UserID: payload.UserID,
|
||
VisibleRegionID: payload.VisibleRegionID,
|
||
}, domain.DrawResult{
|
||
DrawID: payload.DrawID,
|
||
DrawIDs: payload.DrawIDs,
|
||
CommandID: payload.CommandID,
|
||
WheelID: payload.WheelID,
|
||
SelectedTierID: payload.SelectedTierID,
|
||
RewardType: payload.RewardType,
|
||
RewardID: payload.RewardID,
|
||
RewardCount: payload.RewardCount,
|
||
RewardCoins: payload.RewardCoins,
|
||
RewardStatus: domain.StatusPending,
|
||
Rewards: payload.Rewards,
|
||
})
|
||
if err == nil {
|
||
drawIDs := payload.DrawIDs
|
||
if len(drawIDs) == 0 {
|
||
drawIDs = []string{payload.DrawID}
|
||
}
|
||
// draw、统计和 outbox 的最终提交带 worker fencing token；若 lease 已被接管，这里会回滚而不是覆盖新 worker。
|
||
return s.repository.MarkWheelRewardOutboxDelivered(ctx, record.OutboxID, options.WorkerID, drawIDs, fulfilled.WalletTransactionID, s.now().UTC().UnixMilli())
|
||
}
|
||
}
|
||
|
||
nowMS := s.now().UTC().UnixMilli()
|
||
nextRetryAtMS := nowMS + wheelRewardRetryDelay(record.RetryCount+1).Milliseconds()
|
||
markErr := s.repository.MarkWheelRewardOutboxFailed(ctx, record.OutboxID, options.WorkerID, nowMS, nextRetryAtMS, options.MaxRetry, err.Error())
|
||
if markErr != nil {
|
||
return errors.Join(err, fmt.Errorf("mark wheel reward retryable: %w", markErr))
|
||
}
|
||
return err
|
||
}
|
||
|
||
func decodeWheelRewardOutboxPayload(record domain.RewardOutboxRecord) (wheelRewardOutboxPayload, error) {
|
||
var payload wheelRewardOutboxPayload
|
||
if err := json.Unmarshal(record.Payload, &payload); err != nil {
|
||
return payload, fmt.Errorf("decode wheel reward payload: %w", err)
|
||
}
|
||
recordApp := appcode.Normalize(record.AppCode)
|
||
payloadApp := appcode.Normalize(payload.AppCode)
|
||
if payload.AppCode != "" && payloadApp != recordApp {
|
||
return payload, xerr.New(xerr.Conflict, "wheel reward payload app_code does not match outbox partition")
|
||
}
|
||
payload.AppCode = recordApp
|
||
payload.DrawID = strings.TrimSpace(payload.DrawID)
|
||
payload.CommandID = strings.TrimSpace(payload.CommandID)
|
||
payload.WheelID = strings.TrimSpace(payload.WheelID)
|
||
if payload.DrawID == "" || payload.CommandID == "" || payload.WheelID == "" || payload.UserID <= 0 {
|
||
return payload, xerr.New(xerr.InvalidArgument, "wheel reward payload identity is incomplete")
|
||
}
|
||
if len(payload.DrawIDs) == 0 {
|
||
payload.DrawIDs = []string{payload.DrawID}
|
||
}
|
||
for index := range payload.DrawIDs {
|
||
payload.DrawIDs[index] = strings.TrimSpace(payload.DrawIDs[index])
|
||
if payload.DrawIDs[index] == "" {
|
||
return payload, xerr.New(xerr.InvalidArgument, "wheel reward payload contains empty draw_id")
|
||
}
|
||
}
|
||
return payload, nil
|
||
}
|
||
|
||
func normalizeRewardWorkerOptions(options RewardWorkerOptions) RewardWorkerOptions {
|
||
options.WorkerID = strings.TrimSpace(options.WorkerID)
|
||
if options.WorkerID == "" {
|
||
options.WorkerID = "activity-wheel-reward"
|
||
}
|
||
seen := make(map[string]struct{}, len(options.AppCodes))
|
||
appCodes := make([]string, 0, len(options.AppCodes))
|
||
for _, value := range options.AppCodes {
|
||
value = appcode.Normalize(value)
|
||
if _, exists := seen[value]; exists {
|
||
continue
|
||
}
|
||
seen[value] = struct{}{}
|
||
appCodes = append(appCodes, value)
|
||
}
|
||
if len(appCodes) == 0 {
|
||
appCodes = []string{"lalu", "fami", "huwaa"}
|
||
}
|
||
options.AppCodes = appCodes
|
||
if options.PollInterval <= 0 {
|
||
options.PollInterval = time.Second
|
||
}
|
||
if options.BatchSize <= 0 {
|
||
options.BatchSize = 50
|
||
}
|
||
if options.LockTTL <= 0 {
|
||
options.LockTTL = 30 * time.Second
|
||
}
|
||
if options.MaxRetry <= 0 {
|
||
options.MaxRetry = 16
|
||
}
|
||
return options
|
||
}
|
||
|
||
func wheelRewardRetryDelay(attempt int) time.Duration {
|
||
if attempt < 1 {
|
||
attempt = 1
|
||
}
|
||
if attempt > 7 {
|
||
attempt = 7
|
||
}
|
||
delay := time.Second * time.Duration(1<<(attempt-1))
|
||
if delay > time.Minute {
|
||
return time.Minute
|
||
}
|
||
return delay
|
||
}
|