401 lines
14 KiB
Go
401 lines
14 KiB
Go
package luckygift
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"log/slog"
|
||
"strings"
|
||
"time"
|
||
|
||
"google.golang.org/grpc"
|
||
walletv1 "hyapp.local/api/proto/wallet/v1"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/logx"
|
||
"hyapp/pkg/tencentim"
|
||
"hyapp/pkg/xerr"
|
||
domain "hyapp/services/activity-service/internal/domain/luckygift"
|
||
)
|
||
|
||
// Repository 是幸运礼物唯一持久化边界；RTP、奖池、风控计数必须在同一个 MySQL 事务内修改。
|
||
type Repository interface {
|
||
GetLuckyGiftConfig(ctx context.Context, poolID string) (domain.Config, bool, error)
|
||
UpsertLuckyGiftConfig(ctx context.Context, config domain.Config, nowMS int64) (domain.Config, error)
|
||
ListLuckyGiftConfigs(ctx context.Context) ([]domain.Config, error)
|
||
CheckLuckyGift(ctx context.Context, cmd domain.CheckCommand) (domain.CheckResult, error)
|
||
ExecuteLuckyGiftDraw(ctx context.Context, cmd domain.DrawCommand, nowMS int64) (domain.DrawResult, error)
|
||
ListLuckyGiftDraws(ctx context.Context, query domain.DrawQuery) ([]domain.DrawResult, int64, error)
|
||
GetLuckyGiftDrawSummary(ctx context.Context, query domain.DrawQuery) (domain.DrawSummary, error)
|
||
ClaimPendingLuckyGiftOutbox(ctx context.Context, workerID string, nowMS int64, lockTTL time.Duration, batchSize int) ([]domain.DrawOutbox, error)
|
||
MarkLuckyGiftOutboxDelivered(ctx context.Context, event domain.DrawOutbox, nowMS int64) error
|
||
MarkLuckyGiftOutboxRetryable(ctx context.Context, event domain.DrawOutbox, retryCount int, nextRetryAtMS int64, lastErr string, nowMS int64) error
|
||
MarkLuckyGiftOutboxFailed(ctx context.Context, event domain.DrawOutbox, retryCount int, lastErr string, nowMS int64) error
|
||
MarkLuckyGiftDrawGranted(ctx context.Context, appCode string, drawID string, walletTransactionID string, nowMS int64) error
|
||
MarkLuckyGiftDrawFailed(ctx context.Context, appCode string, drawID string, failureReason string, nowMS int64) error
|
||
}
|
||
|
||
// WalletClient 是幸运礼物返奖唯一账务依赖；实际余额变更通知继续由 wallet outbox -> notice-service 投递。
|
||
type WalletClient interface {
|
||
CreditLuckyGiftReward(ctx context.Context, req *walletv1.CreditLuckyGiftRewardRequest, opts ...grpc.CallOption) (*walletv1.CreditLuckyGiftRewardResponse, error)
|
||
}
|
||
|
||
// RoomPublisher 只负责投递房间群自定义消息，不拥有房间状态。
|
||
type RoomPublisher interface {
|
||
PublishGroupCustomMessage(ctx context.Context, message tencentim.CustomGroupMessage) error
|
||
}
|
||
|
||
// Service 承载幸运礼物查询、抽奖命令和后台规则配置。
|
||
type Service struct {
|
||
repository Repository
|
||
wallet WalletClient
|
||
publisher RoomPublisher
|
||
now func() time.Time
|
||
}
|
||
|
||
type Option func(*Service)
|
||
|
||
func WithWallet(client WalletClient) Option {
|
||
return func(s *Service) {
|
||
s.wallet = client
|
||
}
|
||
}
|
||
|
||
func WithRoomPublisher(publisher RoomPublisher) Option {
|
||
return func(s *Service) {
|
||
s.publisher = publisher
|
||
}
|
||
}
|
||
|
||
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) Check(ctx context.Context, cmd domain.CheckCommand) (domain.CheckResult, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.CheckResult{}, err
|
||
}
|
||
cmd.PoolID = normalizePoolID(cmd.PoolID)
|
||
cmd.GiftID = strings.TrimSpace(cmd.GiftID)
|
||
cmd.RoomID = strings.TrimSpace(cmd.RoomID)
|
||
if cmd.GiftID == "" || cmd.UserID <= 0 || cmd.RoomID == "" {
|
||
return domain.CheckResult{}, xerr.New(xerr.InvalidArgument, "lucky gift check command is incomplete")
|
||
}
|
||
return s.repository.CheckLuckyGift(ctx, cmd)
|
||
}
|
||
|
||
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.PoolID = normalizePoolID(cmd.PoolID)
|
||
cmd.GiftID = strings.TrimSpace(cmd.GiftID)
|
||
cmd.DeviceID = strings.TrimSpace(cmd.DeviceID)
|
||
cmd.RoomID = strings.TrimSpace(cmd.RoomID)
|
||
cmd.AnchorID = strings.TrimSpace(cmd.AnchorID)
|
||
if cmd.CommandID == "" || cmd.UserID <= 0 || cmd.DeviceID == "" || cmd.RoomID == "" || cmd.AnchorID == "" || cmd.GiftID == "" || cmd.CoinSpent <= 0 {
|
||
return domain.DrawResult{}, xerr.New(xerr.InvalidArgument, "lucky gift draw command is incomplete")
|
||
}
|
||
if cmd.TargetUserID < 0 || cmd.GiftCount < 0 {
|
||
return domain.DrawResult{}, xerr.New(xerr.InvalidArgument, "lucky gift draw metadata is invalid")
|
||
}
|
||
if cmd.PaidAtMS <= 0 {
|
||
cmd.PaidAtMS = s.now().UTC().UnixMilli()
|
||
}
|
||
return s.repository.ExecuteLuckyGiftDraw(ctx, cmd, s.now().UTC().UnixMilli())
|
||
}
|
||
|
||
type WorkerOptions struct {
|
||
WorkerID string
|
||
PollInterval time.Duration
|
||
BatchSize int
|
||
LockTTL time.Duration
|
||
MaxRetry int
|
||
PublishTimeout time.Duration
|
||
}
|
||
|
||
func (s *Service) RunWorker(ctx context.Context, options WorkerOptions) {
|
||
options = normalizeWorkerOptions(options)
|
||
ticker := time.NewTicker(options.PollInterval)
|
||
defer ticker.Stop()
|
||
for {
|
||
if _, err := s.ProcessPendingDrawOutbox(ctx, options); err != nil && ctx.Err() == nil {
|
||
// 批处理失败不删除 outbox；锁过期后下一轮或其他实例会继续补偿。
|
||
logx.Error(ctx, "lucky_gift_outbox_batch_failed", err, slog.String("worker_id", options.WorkerID))
|
||
}
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
}
|
||
}
|
||
}
|
||
|
||
func (s *Service) ProcessPendingDrawOutbox(ctx context.Context, options WorkerOptions) (int, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return 0, err
|
||
}
|
||
options = normalizeWorkerOptions(options)
|
||
events, err := s.repository.ClaimPendingLuckyGiftOutbox(ctx, options.WorkerID, s.now().UTC().UnixMilli(), options.LockTTL, options.BatchSize)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
processed := 0
|
||
for _, event := range events {
|
||
if err := s.processDrawOutbox(ctx, event, options); err != nil {
|
||
return processed, err
|
||
}
|
||
processed++
|
||
}
|
||
return processed, nil
|
||
}
|
||
|
||
func (s *Service) processDrawOutbox(ctx context.Context, event domain.DrawOutbox, options WorkerOptions) error {
|
||
payload, err := decodeDrawOutboxPayload(event.PayloadJSON)
|
||
if err != nil {
|
||
return s.markOutboxFailed(ctx, event, options, err, false)
|
||
}
|
||
eventCtx := appcode.WithContext(ctx, event.AppCode)
|
||
walletTransactionID := ""
|
||
credited := false
|
||
if payload.EffectiveRewardCoins > 0 {
|
||
if s.wallet == nil {
|
||
return s.markOutboxFailed(eventCtx, event, options, fmt.Errorf("lucky gift wallet client is not configured"), false)
|
||
}
|
||
resp, err := s.wallet.CreditLuckyGiftReward(eventCtx, &walletv1.CreditLuckyGiftRewardRequest{
|
||
CommandId: "lucky_reward:" + payload.DrawID,
|
||
AppCode: event.AppCode,
|
||
TargetUserId: payload.UserID,
|
||
Amount: payload.EffectiveRewardCoins,
|
||
DrawId: payload.DrawID,
|
||
RoomId: payload.RoomID,
|
||
GiftId: payload.GiftID,
|
||
PoolId: payload.PoolID,
|
||
Reason: "lucky_gift_reward",
|
||
})
|
||
if err != nil {
|
||
return s.markOutboxFailed(eventCtx, event, options, err, false)
|
||
}
|
||
walletTransactionID = resp.GetTransactionId()
|
||
credited = true
|
||
}
|
||
if s.publisher == nil {
|
||
return s.markOutboxFailed(eventCtx, event, options, fmt.Errorf("lucky gift room publisher is not configured"), credited)
|
||
}
|
||
if err := s.publishLuckyGiftDrawn(eventCtx, payload, options); err != nil {
|
||
return s.markOutboxFailed(eventCtx, event, options, err, credited)
|
||
}
|
||
nowMS := s.now().UTC().UnixMilli()
|
||
if err := s.repository.MarkLuckyGiftDrawGranted(eventCtx, event.AppCode, payload.DrawID, walletTransactionID, nowMS); err != nil {
|
||
return err
|
||
}
|
||
return s.repository.MarkLuckyGiftOutboxDelivered(eventCtx, event, nowMS)
|
||
}
|
||
|
||
func (s *Service) publishLuckyGiftDrawn(ctx context.Context, payload luckyGiftDrawnPayload, options WorkerOptions) error {
|
||
body, err := json.Marshal(payload)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
publishCtx, cancel := context.WithTimeout(ctx, options.PublishTimeout)
|
||
defer cancel()
|
||
return s.publisher.PublishGroupCustomMessage(publishCtx, tencentim.CustomGroupMessage{
|
||
GroupID: payload.RoomID,
|
||
EventID: payload.EventID,
|
||
Desc: "lucky_gift_drawn",
|
||
Ext: "room_system_message",
|
||
PayloadJSON: body,
|
||
})
|
||
}
|
||
|
||
func (s *Service) markOutboxFailed(ctx context.Context, event domain.DrawOutbox, options WorkerOptions, cause error, alreadyCredited bool) error {
|
||
nowMS := s.now().UTC().UnixMilli()
|
||
nextRetryCount := event.RetryCount + 1
|
||
if nextRetryCount >= options.MaxRetry {
|
||
if !alreadyCredited {
|
||
_ = s.repository.MarkLuckyGiftDrawFailed(ctx, event.AppCode, drawIDFromPayload(event.PayloadJSON), cause.Error(), nowMS)
|
||
}
|
||
return s.repository.MarkLuckyGiftOutboxFailed(ctx, event, nextRetryCount, cause.Error(), nowMS)
|
||
}
|
||
nextRetryAtMS := nowMS + luckyGiftBackoff(nextRetryCount).Milliseconds()
|
||
return s.repository.MarkLuckyGiftOutboxRetryable(ctx, event, nextRetryCount, nextRetryAtMS, cause.Error(), nowMS)
|
||
}
|
||
|
||
func (s *Service) GetConfig(ctx context.Context, poolID string) (domain.Config, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.Config{}, err
|
||
}
|
||
poolID = normalizePoolID(poolID)
|
||
config, exists, err := s.repository.GetLuckyGiftConfig(ctx, poolID)
|
||
if err != nil {
|
||
return domain.Config{}, err
|
||
}
|
||
if !exists {
|
||
return DefaultConfig(appcode.FromContext(ctx), poolID), nil
|
||
}
|
||
return config, nil
|
||
}
|
||
|
||
func (s *Service) UpsertConfig(ctx context.Context, config domain.Config) (domain.Config, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.Config{}, err
|
||
}
|
||
config.AppCode = appcode.FromContext(ctx)
|
||
config.PoolID = normalizePoolID(config.PoolID)
|
||
config.GiftID = config.PoolID
|
||
config.Tiers, config.MultiplierPPMs = normalizeTiers(config.Tiers, config.MultiplierPPMs, config.GiftPrice, config.HighMultiplier)
|
||
if err := validateConfig(config); err != nil {
|
||
return domain.Config{}, err
|
||
}
|
||
return s.repository.UpsertLuckyGiftConfig(ctx, config, s.now().UTC().UnixMilli())
|
||
}
|
||
|
||
func (s *Service) ListConfigs(ctx context.Context) ([]domain.Config, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return nil, err
|
||
}
|
||
return s.repository.ListLuckyGiftConfigs(ctx)
|
||
}
|
||
|
||
func (s *Service) ListDraws(ctx context.Context, query domain.DrawQuery) ([]domain.DrawResult, int64, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
query = normalizeDrawQuery(query)
|
||
if query.Page <= 0 {
|
||
query.Page = 1
|
||
}
|
||
if query.PageSize <= 0 {
|
||
query.PageSize = 20
|
||
}
|
||
if query.PageSize > 100 {
|
||
query.PageSize = 100
|
||
}
|
||
return s.repository.ListLuckyGiftDraws(ctx, 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.GetLuckyGiftDrawSummary(ctx, normalizeDrawQuery(query))
|
||
}
|
||
|
||
func (s *Service) requireRepository() error {
|
||
if s == nil || s.repository == nil {
|
||
return xerr.New(xerr.Unavailable, "lucky gift repository is not configured")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func normalizeDrawQuery(query domain.DrawQuery) domain.DrawQuery {
|
||
query.PoolID = normalizePoolID(query.PoolID)
|
||
query.GiftID = strings.TrimSpace(query.GiftID)
|
||
query.RoomID = strings.TrimSpace(query.RoomID)
|
||
query.Status = strings.TrimSpace(query.Status)
|
||
return query
|
||
}
|
||
|
||
type luckyGiftDrawnPayload struct {
|
||
EventType string `json:"event_type"`
|
||
EventID string `json:"event_id"`
|
||
AppCode string `json:"app_code"`
|
||
DrawID string `json:"draw_id"`
|
||
CommandID string `json:"command_id"`
|
||
PoolID string `json:"pool_id"`
|
||
RoomID string `json:"room_id"`
|
||
GiftID string `json:"gift_id"`
|
||
GiftCount int32 `json:"gift_count"`
|
||
UserID int64 `json:"user_id"`
|
||
SenderUserID int64 `json:"sender_user_id"`
|
||
TargetUserID int64 `json:"target_user_id"`
|
||
CoinSpent int64 `json:"coin_spent"`
|
||
RuleVersion int64 `json:"rule_version"`
|
||
ExperiencePool string `json:"experience_pool"`
|
||
SelectedTierID string `json:"selected_tier_id"`
|
||
MultiplierPPM int64 `json:"multiplier_ppm"`
|
||
BaseRewardCoins int64 `json:"base_reward_coins"`
|
||
RoomAtmosphereRewardCoins int64 `json:"room_atmosphere_reward_coins"`
|
||
ActivitySubsidyCoins int64 `json:"activity_subsidy_coins"`
|
||
EffectiveRewardCoins int64 `json:"effective_reward_coins"`
|
||
StageFeedback bool `json:"stage_feedback"`
|
||
HighMultiplier bool `json:"high_multiplier"`
|
||
CreatedAtMS int64 `json:"created_at_ms"`
|
||
}
|
||
|
||
func decodeDrawOutboxPayload(payloadJSON string) (luckyGiftDrawnPayload, error) {
|
||
var payload luckyGiftDrawnPayload
|
||
if err := json.Unmarshal([]byte(payloadJSON), &payload); err != nil {
|
||
return luckyGiftDrawnPayload{}, fmt.Errorf("invalid lucky gift outbox payload: %w", err)
|
||
}
|
||
if payload.EventType == "" {
|
||
payload.EventType = "lucky_gift_drawn"
|
||
}
|
||
if payload.EventID == "" && payload.DrawID != "" {
|
||
payload.EventID = "lucky_gift_drawn:" + payload.DrawID
|
||
}
|
||
if payload.SenderUserID == 0 {
|
||
payload.SenderUserID = payload.UserID
|
||
}
|
||
if payload.DrawID == "" || payload.EventID == "" || payload.RoomID == "" || payload.GiftID == "" || payload.UserID <= 0 {
|
||
return luckyGiftDrawnPayload{}, fmt.Errorf("lucky gift outbox payload is incomplete")
|
||
}
|
||
return payload, nil
|
||
}
|
||
|
||
func drawIDFromPayload(payloadJSON string) string {
|
||
payload, err := decodeDrawOutboxPayload(payloadJSON)
|
||
if err != nil {
|
||
return ""
|
||
}
|
||
return payload.DrawID
|
||
}
|
||
|
||
func normalizeWorkerOptions(options WorkerOptions) WorkerOptions {
|
||
options.WorkerID = strings.TrimSpace(options.WorkerID)
|
||
if options.WorkerID == "" {
|
||
options.WorkerID = "lucky-gift-worker"
|
||
}
|
||
if options.PollInterval <= 0 {
|
||
options.PollInterval = time.Second
|
||
}
|
||
if options.BatchSize <= 0 {
|
||
options.BatchSize = 100
|
||
}
|
||
if options.BatchSize > 500 {
|
||
options.BatchSize = 500
|
||
}
|
||
if options.LockTTL <= 0 {
|
||
options.LockTTL = 30 * time.Second
|
||
}
|
||
if options.MaxRetry <= 0 {
|
||
options.MaxRetry = 8
|
||
}
|
||
if options.PublishTimeout <= 0 {
|
||
options.PublishTimeout = 5 * time.Second
|
||
}
|
||
return options
|
||
}
|
||
|
||
func luckyGiftBackoff(retryCount int) time.Duration {
|
||
if retryCount < 1 {
|
||
retryCount = 1
|
||
}
|
||
if retryCount > 6 {
|
||
retryCount = 6
|
||
}
|
||
return time.Duration(1<<retryCount) * time.Second
|
||
}
|