499 lines
20 KiB
Go
499 lines
20 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"
|
||
broadcastdomain "hyapp/services/activity-service/internal/domain/broadcast"
|
||
domain "hyapp/services/activity-service/internal/domain/luckygift"
|
||
broadcastservice "hyapp/services/activity-service/internal/service/broadcast"
|
||
)
|
||
|
||
// Repository 是幸运礼物唯一持久化边界；RTP、奖池、风控计数必须在同一个 MySQL 事务内修改。
|
||
type Repository interface {
|
||
GetLuckyGiftRuleConfig(ctx context.Context, poolID string) (domain.RuleConfig, bool, error)
|
||
PublishLuckyGiftRuleConfig(ctx context.Context, config domain.RuleConfig, nowMS int64) (domain.RuleConfig, error)
|
||
ListLuckyGiftRuleConfigs(ctx context.Context) ([]domain.RuleConfig, 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
|
||
}
|
||
|
||
// RegionBroadcaster 只写 activity-service 自有的区域播报 outbox；真正发腾讯 IM 仍由 broadcast worker 补偿。
|
||
type RegionBroadcaster interface {
|
||
PublishRegionBroadcast(ctx context.Context, input broadcastservice.PublishInput) (broadcastdomain.PublishResult, error)
|
||
}
|
||
|
||
// Service 承载幸运礼物查询、抽奖命令和后台规则配置。
|
||
type Service struct {
|
||
repository Repository
|
||
wallet WalletClient
|
||
publisher RoomPublisher
|
||
broadcaster RegionBroadcaster
|
||
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 WithRegionBroadcaster(broadcaster RegionBroadcaster) Option {
|
||
return func(s *Service) {
|
||
s.broadcaster = broadcaster
|
||
}
|
||
}
|
||
|
||
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
|
||
}
|
||
// Check 是送礼前拦截，不产生抽奖事实；这里只归一化入参，最终启停仍以仓储中的当前规则为准。
|
||
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
|
||
}
|
||
// Draw 只接受扣费成功后的服务端事实；客户端不能提交价格、倍率、奖励或风控绕过字段。
|
||
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 {
|
||
// paid_at_ms 是扣费事实时间，缺失时用 UTC 服务端时间补齐，避免使用本地时区切风控窗口。
|
||
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)
|
||
// outbox 先抢占再处理，避免多实例同时给同一 draw 返奖或重复投递房间 IM。
|
||
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 {
|
||
// 单条失败立即返回，让外层记录批次错误；已处理成功的事件已经各自落 delivered，不会被回滚。
|
||
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 {
|
||
// payload 不可解析说明抽奖事实旁路已经损坏，重试也无法恢复，直接走失败收敛。
|
||
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)
|
||
}
|
||
// 返奖 command_id 派生自 draw_id，钱包侧可用它做幂等；worker 重试不会重复加币。
|
||
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 {
|
||
// 已返奖但缺少房间发布依赖时不能把 draw 标记 failed，否则会误导账务；只让 outbox 保持可重试。
|
||
return s.markOutboxFailed(eventCtx, event, options, fmt.Errorf("lucky gift room publisher is not configured"), credited)
|
||
}
|
||
if shouldPublishLuckyGiftRegionBroadcast(payload) {
|
||
if s.broadcaster == nil {
|
||
return s.markOutboxFailed(eventCtx, event, options, fmt.Errorf("lucky gift region broadcaster is not configured"), credited)
|
||
}
|
||
// 区域播报先写 broadcast outbox，不直接发 IM；这里失败意味着大额表现未落地，需要保留重试。
|
||
if err := s.publishLuckyGiftRegionBroadcast(eventCtx, payload); err != nil {
|
||
return s.markOutboxFailed(eventCtx, event, options, err, credited)
|
||
}
|
||
}
|
||
if err := s.publishLuckyGiftDrawn(eventCtx, payload, options); err != nil {
|
||
return s.markOutboxFailed(eventCtx, event, options, err, credited)
|
||
}
|
||
nowMS := s.now().UTC().UnixMilli()
|
||
// 所有外部副作用都成功后再把 draw 从 pending 收敛到 granted；客户端据此区分“处理中”和“已完成”。
|
||
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
|
||
}
|
||
// 房间 IM 只是表现层事实同步，不修改 Room Cell；超时由 outbox 重试兜底。
|
||
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) publishLuckyGiftRegionBroadcast(ctx context.Context, payload luckyGiftDrawnPayload) error {
|
||
if payload.VisibleRegionID <= 0 {
|
||
return nil
|
||
}
|
||
body, err := json.Marshal(luckyGiftRegionBroadcastPayload(payload, s.now().UTC().UnixMilli()))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = s.broadcaster.PublishRegionBroadcast(ctx, broadcastservice.PublishInput{
|
||
EventID: "lucky_gift_big_win:" + payload.DrawID,
|
||
BroadcastType: broadcastdomain.TypeLuckyGiftBigWin,
|
||
RegionID: payload.VisibleRegionID,
|
||
PayloadJSON: string(body),
|
||
})
|
||
return err
|
||
}
|
||
|
||
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 {
|
||
// 未返奖的 draw 可以标记 failed；已返奖但 IM 失败时不能覆盖账务事实，只停 outbox 并告警。
|
||
_ = s.repository.MarkLuckyGiftDrawFailed(ctx, event.AppCode, drawIDFromPayload(event.PayloadJSON), cause.Error(), nowMS)
|
||
}
|
||
return s.repository.MarkLuckyGiftOutboxFailed(ctx, event, nextRetryCount, cause.Error(), nowMS)
|
||
}
|
||
// 指数退避降低故障期间对钱包和 IM 的压力，同时保留 pending 状态给客户端展示“处理中”。
|
||
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.RuleConfig, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.RuleConfig{}, err
|
||
}
|
||
poolID = normalizePoolID(poolID)
|
||
config, exists, err := s.repository.GetLuckyGiftRuleConfig(ctx, poolID)
|
||
if err != nil {
|
||
return domain.RuleConfig{}, err
|
||
}
|
||
if !exists {
|
||
// 未配置时返回 disabled v2 草稿，后台可以直接编辑发布；运行抽奖不会读取这个草稿。
|
||
return DefaultRuleConfig(appcode.FromContext(ctx), poolID), nil
|
||
}
|
||
return config, 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.AppCode = appcode.FromContext(ctx)
|
||
config.PoolID = normalizePoolID(config.PoolID)
|
||
config = normalizeRuleConfig(config)
|
||
if err := validateRuleConfig(config); err != nil {
|
||
return domain.RuleConfig{}, err
|
||
}
|
||
return s.repository.PublishLuckyGiftRuleConfig(ctx, config, s.now().UTC().UnixMilli())
|
||
}
|
||
|
||
func (s *Service) ListConfigs(ctx context.Context) ([]domain.RuleConfig, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return nil, err
|
||
}
|
||
return s.repository.ListLuckyGiftRuleConfigs(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"`
|
||
VisibleRegionID int64 `json:"visible_region_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 shouldPublishLuckyGiftRegionBroadcast(payload luckyGiftDrawnPayload) bool {
|
||
// 当前产品策略先把 1 倍作为默认区域飘屏阈值；后续接后台配置时只替换这里的阈值来源。
|
||
return payload.VisibleRegionID > 0 && payload.EffectiveRewardCoins > 0 && payload.MultiplierPPM >= 1_000_000
|
||
}
|
||
|
||
func luckyGiftRegionBroadcastPayload(payload luckyGiftDrawnPayload, sentAtMS int64) map[string]any {
|
||
return map[string]any{
|
||
"event_type": "lucky_gift_drawn",
|
||
"event_id": "lucky_gift_big_win:" + payload.DrawID,
|
||
"broadcast_type": broadcastdomain.TypeLuckyGiftBigWin,
|
||
"scope": broadcastdomain.ScopeRegion,
|
||
"app_code": payload.AppCode,
|
||
"region_id": payload.VisibleRegionID,
|
||
"visible_region_id": payload.VisibleRegionID,
|
||
"draw_id": payload.DrawID,
|
||
"command_id": payload.CommandID,
|
||
"pool_id": payload.PoolID,
|
||
"room_id": payload.RoomID,
|
||
"gift_id": payload.GiftID,
|
||
"gift_count": payload.GiftCount,
|
||
"user_id": payload.UserID,
|
||
"sender_user_id": payload.SenderUserID,
|
||
"target_user_id": payload.TargetUserID,
|
||
"coin_spent": payload.CoinSpent,
|
||
"rule_version": payload.RuleVersion,
|
||
"experience_pool": payload.ExperiencePool,
|
||
"selected_tier_id": payload.SelectedTierID,
|
||
"multiplier_ppm": payload.MultiplierPPM,
|
||
"base_reward_coins": payload.BaseRewardCoins,
|
||
"room_atmosphere_reward_coins": payload.RoomAtmosphereRewardCoins,
|
||
"activity_subsidy_coins": payload.ActivitySubsidyCoins,
|
||
"effective_reward_coins": payload.EffectiveRewardCoins,
|
||
"stage_feedback": payload.StageFeedback,
|
||
"high_multiplier": payload.HighMultiplier,
|
||
"created_at_ms": payload.CreatedAtMS,
|
||
"sent_at_ms": sentAtMS,
|
||
"action": map[string]any{
|
||
"type": "enter_room",
|
||
"room_id": payload.RoomID,
|
||
},
|
||
}
|
||
}
|
||
|
||
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)
|
||
}
|
||
// 兼容早期 payload 缺省字段，避免历史 pending outbox 因新增字段永久失败。
|
||
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
|
||
}
|
||
// 这些字段是返奖、房间 IM 和幂等去重的最小闭环，缺任何一个都不能安全补偿。
|
||
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
|
||
}
|