348 lines
13 KiB
Go
348 lines
13 KiB
Go
package wheel
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/json"
|
||
"fmt"
|
||
"log/slog"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"google.golang.org/grpc"
|
||
walletv1 "hyapp.local/api/proto/wallet/v1"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/logx"
|
||
"hyapp/pkg/xerr"
|
||
domain "hyapp/services/activity-service/internal/domain/wheel"
|
||
)
|
||
|
||
type Repository interface {
|
||
PublishWheelRuleConfig(ctx context.Context, config domain.RuleConfig, nowMS int64) (domain.RuleConfig, error)
|
||
GetWheelRuleConfig(ctx context.Context, wheelID string) (domain.RuleConfig, bool, error)
|
||
ExecuteWheelDraw(ctx context.Context, cmd domain.DrawCommand, nowMS int64) (domain.DrawResult, error)
|
||
ListWheelDraws(ctx context.Context, query domain.DrawQuery) ([]domain.DrawResult, int64, error)
|
||
GetWheelDrawSummary(ctx context.Context, query domain.DrawQuery) (domain.DrawSummary, error)
|
||
MarkWheelDrawsGranted(ctx context.Context, appCode string, drawIDs []string, transactionID string, nowMS int64) error
|
||
ClaimWheelRewardOutbox(ctx context.Context, workerID string, nowMS int64, lockUntilMS int64, limit int, maxRetry int) ([]domain.RewardOutboxRecord, error)
|
||
MarkWheelRewardOutboxDelivered(ctx context.Context, outboxID string, workerID string, drawIDs []string, transactionID string, nowMS int64) error
|
||
MarkWheelRewardOutboxFailed(ctx context.Context, outboxID string, workerID string, nowMS int64, nextRetryAtMS int64, maxRetry int, lastError string) error
|
||
}
|
||
|
||
type WalletClient interface {
|
||
CreditWheelReward(ctx context.Context, req *walletv1.CreditWheelRewardRequest, opts ...grpc.CallOption) (*walletv1.CreditWheelRewardResponse, error)
|
||
GrantResource(ctx context.Context, req *walletv1.GrantResourceRequest, opts ...grpc.CallOption) (*walletv1.ResourceGrantResponse, error)
|
||
}
|
||
|
||
type Service struct {
|
||
repository Repository
|
||
wallet WalletClient
|
||
now func() time.Time
|
||
}
|
||
|
||
type Option func(*Service)
|
||
|
||
func WithWallet(client WalletClient) Option {
|
||
return func(s *Service) {
|
||
s.wallet = client
|
||
}
|
||
}
|
||
|
||
func New(repository Repository, options ...Option) *Service {
|
||
service := &Service{repository: repository, now: time.Now}
|
||
for _, option := range options {
|
||
if option != nil {
|
||
option(service)
|
||
}
|
||
}
|
||
return service
|
||
}
|
||
|
||
func (s *Service) SetClock(now func() time.Time) {
|
||
if now != nil {
|
||
s.now = now
|
||
}
|
||
}
|
||
|
||
func (s *Service) Draw(ctx context.Context, cmd domain.DrawCommand) (domain.DrawResult, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.DrawResult{}, err
|
||
}
|
||
cmd.CommandID = strings.TrimSpace(cmd.CommandID)
|
||
cmd.WheelID = NormalizeWheelID(cmd.WheelID)
|
||
cmd.DeviceID = strings.TrimSpace(cmd.DeviceID)
|
||
if cmd.DrawCount <= 0 {
|
||
cmd.DrawCount = 1
|
||
}
|
||
if cmd.CommandID == "" || cmd.WheelID == "" || cmd.UserID <= 0 || cmd.DeviceID == "" || cmd.CoinSpent <= 0 {
|
||
return domain.DrawResult{}, xerr.New(xerr.InvalidArgument, "wheel draw command is incomplete")
|
||
}
|
||
if cmd.PaidAtMS <= 0 {
|
||
// paid_at_ms 来自钱包扣费事实；缺失时使用 UTC 服务端时间，保证 RTP 窗口和审计时间不受本地时区影响。
|
||
cmd.PaidAtMS = s.now().UTC().UnixMilli()
|
||
}
|
||
result, err := s.repository.ExecuteWheelDraw(ctx, cmd, s.now().UTC().UnixMilli())
|
||
if err != nil {
|
||
return domain.DrawResult{}, err
|
||
}
|
||
return s.fulfillWheelRewardFastPath(ctx, cmd, result), nil
|
||
}
|
||
|
||
func (s *Service) UpsertConfig(ctx context.Context, config domain.RuleConfig) (domain.RuleConfig, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.RuleConfig{}, err
|
||
}
|
||
config = NormalizeRuleConfig(config)
|
||
if err := ValidateRuleConfig(config); err != nil {
|
||
return domain.RuleConfig{}, err
|
||
}
|
||
return s.repository.PublishWheelRuleConfig(ctx, config, s.now().UTC().UnixMilli())
|
||
}
|
||
|
||
func (s *Service) GetConfig(ctx context.Context, wheelID string) (domain.RuleConfig, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.RuleConfig{}, err
|
||
}
|
||
config, exists, err := s.repository.GetWheelRuleConfig(ctx, NormalizeWheelID(wheelID))
|
||
if err != nil {
|
||
return domain.RuleConfig{}, err
|
||
}
|
||
if !exists {
|
||
return domain.RuleConfig{}, xerr.New(xerr.NotFound, "wheel config not found")
|
||
}
|
||
return config, nil
|
||
}
|
||
|
||
func (s *Service) ListDraws(ctx context.Context, query domain.DrawQuery) ([]domain.DrawResult, int64, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
return s.repository.ListWheelDraws(ctx, normalizeDrawQuery(query))
|
||
}
|
||
|
||
func (s *Service) GetDrawSummary(ctx context.Context, query domain.DrawQuery) (domain.DrawSummary, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.DrawSummary{}, err
|
||
}
|
||
return s.repository.GetWheelDrawSummary(ctx, normalizeDrawQuery(query))
|
||
}
|
||
|
||
func (s *Service) fulfillWheelRewardFastPath(ctx context.Context, cmd domain.DrawCommand, result domain.DrawResult) domain.DrawResult {
|
||
if result.RewardStatus == domain.StatusGranted {
|
||
return result
|
||
}
|
||
fulfilled, err := s.fulfillWheelReward(ctx, cmd, result)
|
||
if err != nil {
|
||
// 抽奖结果已经在事务内确认，发放失败时不能回滚抽奖事实；保留 pending/outbox，补偿 worker 会使用相同 command_id 继续核账和补发。
|
||
logx.Error(ctx, "wheel_reward_fast_path_failed", err,
|
||
slog.String("draw_id", result.DrawID),
|
||
slog.String("command_id", result.CommandID),
|
||
)
|
||
return result
|
||
}
|
||
result = fulfilled
|
||
drawIDs := []string{result.DrawID}
|
||
if result.DrawID == "" {
|
||
drawIDs = append([]string(nil), result.DrawIDs...)
|
||
}
|
||
if err := s.repository.MarkWheelDrawsGranted(appcode.WithContext(context.WithoutCancel(ctx), appcode.FromContext(ctx)), appcode.FromContext(ctx), drawIDs, result.WalletTransactionID, s.now().UTC().UnixMilli()); err != nil {
|
||
logx.Error(ctx, "wheel_reward_fast_path_mark_granted_failed", err, slog.String("draw_id", result.DrawID), slog.String("wallet_transaction_id", result.WalletTransactionID))
|
||
}
|
||
return result
|
||
}
|
||
|
||
// fulfillWheelReward 调用 wallet owner 的幂等账务命令完成一组奖品发放。
|
||
// CreditWheelReward/GrantResource 在钱包事务内都会先按稳定 command_id 和 request hash 查询既有交易或权益；因此 worker 在超时、进程崩溃或标记 delivered 失败后重放时，先得到原收据而不是再次入账。
|
||
func (s *Service) fulfillWheelReward(ctx context.Context, cmd domain.DrawCommand, result domain.DrawResult) (domain.DrawResult, error) {
|
||
if s.wallet == nil {
|
||
return result, xerr.New(xerr.Unavailable, "wheel reward wallet client is not configured")
|
||
}
|
||
rewards := wheelResultRewards(result)
|
||
if len(rewards) == 0 {
|
||
return result, xerr.New(xerr.InvalidArgument, "wheel reward snapshot is empty")
|
||
}
|
||
receipts := make([]string, 0, len(rewards))
|
||
for index, reward := range rewards {
|
||
receipt, coinBalance, err := s.fulfillWheelRewardItem(ctx, cmd, result, reward, index, len(rewards))
|
||
if err != nil {
|
||
return result, fmt.Errorf("fulfill reward item %d type=%s tier=%s: %w", index+1, reward.RewardType, reward.SelectedTierID, err)
|
||
}
|
||
if receipt != "" {
|
||
receipts = append(receipts, receipt)
|
||
}
|
||
if coinBalance > 0 {
|
||
result.CoinBalanceAfter = coinBalance
|
||
}
|
||
}
|
||
result.RewardStatus = domain.StatusGranted
|
||
result.WalletTransactionID = wheelRewardReceiptReference(receipts)
|
||
return result, nil
|
||
}
|
||
|
||
func wheelRewardReceiptReference(receipts []string) string {
|
||
joined := strings.Join(receipts, ",")
|
||
const rewardTransactionIDMaxLength = 128
|
||
if len(joined) <= rewardTransactionIDMaxLength {
|
||
return joined
|
||
}
|
||
// wheel_draw_records 的历史字段是 VARCHAR(128)，而钱包 transaction/grant ID 各自约 68 字符；
|
||
// 多奖品直接逗号拼接会让钱包已经全部发放后仍无法标记 granted。超长集合保存稳定摘要，逐项原收据仍由钱包 owner 按派生 command_id 可审计查询。
|
||
sum := sha256.Sum256([]byte(joined))
|
||
return fmt.Sprintf("batch:%d:%x", len(receipts), sum)
|
||
}
|
||
|
||
func (s *Service) fulfillWheelRewardItem(ctx context.Context, cmd domain.DrawCommand, result domain.DrawResult, reward domain.DrawReward, index, total int) (string, int64, error) {
|
||
reward.RewardType = strings.TrimSpace(strings.ToLower(reward.RewardType))
|
||
switch reward.RewardType {
|
||
case domain.RewardTypeCoin:
|
||
return s.creditWheelCoinReward(ctx, cmd, result, reward, index, total)
|
||
case domain.RewardTypeGift, domain.RewardTypeProp, domain.RewardTypeDress:
|
||
return s.grantWheelResourceReward(ctx, cmd, result, reward, index, total)
|
||
default:
|
||
return "", 0, fmt.Errorf("unsupported wheel reward type %q", reward.RewardType)
|
||
}
|
||
}
|
||
|
||
func (s *Service) creditWheelCoinReward(ctx context.Context, cmd domain.DrawCommand, result domain.DrawResult, reward domain.DrawReward, index, total int) (string, int64, error) {
|
||
if reward.RewardCoins <= 0 {
|
||
return "", 0, fmt.Errorf("wheel coin reward amount must be positive")
|
||
}
|
||
resp, err := s.wallet.CreditWheelReward(appcode.WithContext(ctx, appcode.FromContext(ctx)), &walletv1.CreditWheelRewardRequest{
|
||
CommandId: wheelRewardCommandID("wheel_reward", result.DrawID, index, total),
|
||
AppCode: appcode.FromContext(ctx),
|
||
TargetUserId: cmd.UserID,
|
||
Amount: reward.RewardCoins,
|
||
DrawId: result.DrawID,
|
||
WheelId: result.WheelID,
|
||
SelectedTierId: reward.SelectedTierID,
|
||
Reason: "wheel_reward",
|
||
VisibleRegionId: cmd.VisibleRegionID,
|
||
})
|
||
if err != nil {
|
||
return "", 0, err
|
||
}
|
||
return resp.GetTransactionId(), resp.GetBalance().GetAvailableAmount(), nil
|
||
}
|
||
|
||
func (s *Service) grantWheelResourceReward(ctx context.Context, cmd domain.DrawCommand, result domain.DrawResult, reward domain.DrawReward, index, total int) (string, int64, error) {
|
||
resourceID := wheelRewardResourceID(reward)
|
||
if resourceID <= 0 {
|
||
// gift/prop/dress 的真实发放 owner 是 wallet 资源权益；缺 resource_id 时不能用展示 ID 猜测，必须让记录保持 pending 以便后台修配置后补偿。
|
||
return "", 0, fmt.Errorf("wheel resource reward missing resource_id")
|
||
}
|
||
quantity := reward.RewardCount
|
||
if quantity <= 0 {
|
||
quantity = 1
|
||
}
|
||
resp, err := s.wallet.GrantResource(appcode.WithContext(ctx, appcode.FromContext(ctx)), &walletv1.GrantResourceRequest{
|
||
CommandId: wheelRewardCommandID("wheel_resource", result.DrawID, index, total),
|
||
AppCode: appcode.FromContext(ctx),
|
||
TargetUserId: cmd.UserID,
|
||
ResourceId: resourceID,
|
||
Quantity: quantity,
|
||
DurationMs: wheelRewardDurationMS(reward),
|
||
Reason: "wheel_reward",
|
||
OperatorUserId: cmd.UserID,
|
||
GrantSource: "wheel_reward",
|
||
})
|
||
if err != nil {
|
||
return "", 0, err
|
||
}
|
||
return resp.GetGrant().GetGrantId(), 0, nil
|
||
}
|
||
|
||
func wheelResultRewards(result domain.DrawResult) []domain.DrawReward {
|
||
if len(result.Rewards) > 0 {
|
||
return result.Rewards
|
||
}
|
||
if result.RewardType == "" {
|
||
return nil
|
||
}
|
||
return []domain.DrawReward{{
|
||
SelectedTierID: result.SelectedTierID,
|
||
RewardType: result.RewardType,
|
||
RewardID: result.RewardID,
|
||
RewardCount: result.RewardCount,
|
||
RewardCoins: result.RewardCoins,
|
||
RTPValueCoins: result.RTPValueCoins,
|
||
RewardStatus: result.RewardStatus,
|
||
MetadataJSON: result.MetadataJSON,
|
||
}}
|
||
}
|
||
|
||
func wheelRewardCommandID(prefix, drawID string, index, total int) string {
|
||
if total <= 1 {
|
||
return prefix + ":" + drawID
|
||
}
|
||
return fmt.Sprintf("%s:%s:%d", prefix, drawID, index+1)
|
||
}
|
||
|
||
func wheelRewardResourceID(reward domain.DrawReward) int64 {
|
||
if value := wheelRewardMetadataInt64(reward.MetadataJSON, "resource_id", "resourceId"); value > 0 {
|
||
return value
|
||
}
|
||
value, _ := strconv.ParseInt(strings.TrimSpace(reward.RewardID), 10, 64)
|
||
return value
|
||
}
|
||
|
||
func wheelRewardDurationMS(reward domain.DrawReward) int64 {
|
||
return wheelRewardMetadataInt64(reward.MetadataJSON, "duration_ms", "durationMs", "validity_ms", "validityMs")
|
||
}
|
||
|
||
func wheelRewardMetadataInt64(raw string, keys ...string) int64 {
|
||
metadata := map[string]any{}
|
||
if err := json.Unmarshal([]byte(strings.TrimSpace(raw)), &metadata); err != nil {
|
||
return 0
|
||
}
|
||
for _, key := range keys {
|
||
if value, ok := metadata[key]; ok {
|
||
if parsed := anyToInt64(value); parsed > 0 {
|
||
return parsed
|
||
}
|
||
}
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func anyToInt64(value any) int64 {
|
||
switch typed := value.(type) {
|
||
case float64:
|
||
return int64(typed)
|
||
case int64:
|
||
return typed
|
||
case int:
|
||
return int64(typed)
|
||
case json.Number:
|
||
parsed, _ := typed.Int64()
|
||
return parsed
|
||
case string:
|
||
parsed, _ := strconv.ParseInt(strings.TrimSpace(typed), 10, 64)
|
||
return parsed
|
||
default:
|
||
return 0
|
||
}
|
||
}
|
||
|
||
func normalizeDrawQuery(query domain.DrawQuery) domain.DrawQuery {
|
||
query.WheelID = NormalizeWheelID(query.WheelID)
|
||
query.Status = strings.TrimSpace(query.Status)
|
||
if query.Page <= 0 {
|
||
query.Page = 1
|
||
}
|
||
if query.PageSize <= 0 {
|
||
query.PageSize = 20
|
||
}
|
||
if query.PageSize > 100 {
|
||
query.PageSize = 100
|
||
}
|
||
return query
|
||
}
|
||
|
||
func (s *Service) requireRepository() error {
|
||
if s == nil || s.repository == nil {
|
||
return xerr.New(xerr.Unavailable, "wheel repository is not configured")
|
||
}
|
||
return nil
|
||
}
|