From 2e48d6dc6794efaaec57a57326eab4341aeaf69b Mon Sep 17 00:00:00 2001 From: hy001 Date: Thu, 16 Apr 2026 20:07:14 +0800 Subject: [PATCH] feat: add register reward service support --- cmd/api/main.go | 8 +- internal/config/config.go | 25 ++ internal/model/models.go | 86 ++++-- internal/repo/repo.go | 2 + internal/service/register_reward.go | 458 ++++++++++++++++++++++++++++ 5 files changed, 549 insertions(+), 30 deletions(-) create mode 100644 internal/service/register_reward.go diff --git a/cmd/api/main.go b/cmd/api/main.go index 846e448..562b508 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -33,9 +33,15 @@ func main() { javaClient := integration.New(cfg) inviteService := service.NewInviteService(cfg, repository, javaClient) baishunService := service.NewBaishunService(cfg, repository, javaClient) + registerRewardService := service.NewRegisterRewardService(cfg, repository, javaClient) + workerCtx, workerCancel := context.WithCancel(context.Background()) + defer workerCancel() + if err := registerRewardService.Start(workerCtx); err != nil { + log.Fatalf("start register reward consumer failed: %v", err) + } router := invitehttp.NewRouter(cfg, repository, javaClient, inviteService, baishunService) - log.Printf("chatapp3-golang invite api listening on %s", cfg.ListenAddr) + log.Printf("chatapp3-golang api listening on %s", cfg.ListenAddr) if err := router.Run(cfg.ListenAddr); err != nil { log.Fatalf("run server failed: %v", err) } diff --git a/internal/config/config.go b/internal/config/config.go index 4a1cfe8..6fa0e36 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -3,6 +3,7 @@ package config import ( "os" "strconv" + "strings" "time" ) @@ -29,6 +30,12 @@ type Config struct { BaishunLaunchCodeTTLSeconds int BaishunSSTokenTTLSeconds int InternalCallbackSecret string + RegisterRewardStreamKey string + RegisterRewardConsumerGroup string + RegisterRewardConsumerName string + RegisterRewardBatchSize int64 + RegisterRewardBlock time.Duration + RegisterRewardMinIdle time.Duration Timeout time.Duration } @@ -56,6 +63,12 @@ func Load() Config { BaishunLaunchCodeTTLSeconds: getEnvInt("BAISHUN_LAUNCH_CODE_TTL_SECONDS", 300), BaishunSSTokenTTLSeconds: getEnvInt("BAISHUN_SS_TOKEN_TTL_SECONDS", 86400), InternalCallbackSecret: getEnv("GAME_INTERNAL_CALLBACK_SECRET", ""), + RegisterRewardStreamKey: getEnv("REGISTER_REWARD_STREAM_KEY", "activity:register:reward"), + RegisterRewardConsumerGroup: getEnv("REGISTER_REWARD_CONSUMER_GROUP", "register-reward"), + RegisterRewardConsumerName: getEnv("REGISTER_REWARD_CONSUMER_NAME", defaultConsumerName("register-reward")), + RegisterRewardBatchSize: int64(getEnvInt("REGISTER_REWARD_BATCH_SIZE", 10)), + RegisterRewardBlock: time.Duration(getEnvInt("REGISTER_REWARD_BLOCK_MILLIS", 5000)) * time.Millisecond, + RegisterRewardMinIdle: time.Duration(getEnvInt("REGISTER_REWARD_MIN_IDLE_MILLIS", 60000)) * time.Millisecond, Timeout: time.Duration(getEnvInt("INVITE_HTTP_TIMEOUT_SECONDS", 8)) * time.Second, } } @@ -102,3 +115,15 @@ func getEnvBool(key string, fallback bool) bool { } return parsed } + +func defaultConsumerName(prefix string) string { + host, err := os.Hostname() + if err != nil { + return prefix + } + host = strings.TrimSpace(host) + if host == "" { + return prefix + } + return prefix + "-" + host +} diff --git a/internal/model/models.go b/internal/model/models.go index bcd8c99..a9685ed 100644 --- a/internal/model/models.go +++ b/internal/model/models.go @@ -21,6 +21,34 @@ type UserBaseInfo struct { func (UserBaseInfo) TableName() string { return "user_base_info" } +type ResidentRegisterRewardConfig struct { + ID int64 `gorm:"column:id;primaryKey"` + SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_resident_register_reward_sys_origin"` + Enabled bool `gorm:"column:enabled"` + GoldAmount int64 `gorm:"column:gold_amount"` + RewardGroupID *int64 `gorm:"column:reward_group_id"` + CreateTime time.Time `gorm:"column:create_timestamp"` + UpdateTime time.Time `gorm:"column:update_timestamp"` +} + +func (ResidentRegisterRewardConfig) TableName() string { return "resident_register_reward_config" } + +type ResidentRegisterRewardLog struct { + ID int64 `gorm:"column:id;primaryKey"` + EventID string `gorm:"column:event_id;size:128;uniqueIndex:uk_register_reward_event"` + StreamMessageID string `gorm:"column:stream_message_id;size:64"` + SysOrigin string `gorm:"column:sys_origin;size:32;index:idx_register_reward_sys_user,priority:1"` + UserID int64 `gorm:"column:user_id;index:idx_register_reward_sys_user,priority:2"` + GoldAmount int64 `gorm:"column:gold_amount"` + RewardGroupID *int64 `gorm:"column:reward_group_id"` + Status string `gorm:"column:status;size:32"` + ErrorMessage string `gorm:"column:error_message;size:255"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` +} + +func (ResidentRegisterRewardLog) TableName() string { return "resident_register_reward_log" } + type InviteCampaignConfig struct { ID int64 `gorm:"column:id;primaryKey"` SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_invite_campaign_sys_origin"` @@ -71,45 +99,45 @@ type InviteCampaignRelation struct { func (InviteCampaignRelation) TableName() string { return "invite_campaign_relation" } type InviteCampaignInviteeProgress struct { - ID int64 `gorm:"column:id;primaryKey"` - RelationID int64 `gorm:"column:relation_id;uniqueIndex:uk_invitee_progress_relation"` - SysOrigin string `gorm:"column:sys_origin;size:32"` - InviterUserID int64 `gorm:"column:inviter_user_id;index:idx_invitee_progress_inviter"` - InviteeUserID int64 `gorm:"column:invitee_user_id;uniqueIndex:uk_invitee_progress_invitee"` - TotalRecharge int64 `gorm:"column:total_recharge_coins"` - ValidUser bool `gorm:"column:valid_user"` - ValidTime *time.Time `gorm:"column:valid_time"` - RewardedRuleIDs string `gorm:"column:rewarded_rule_ids;size:255"` - CreateTime time.Time `gorm:"column:create_time"` - UpdateTime time.Time `gorm:"column:update_time"` + ID int64 `gorm:"column:id;primaryKey"` + RelationID int64 `gorm:"column:relation_id;uniqueIndex:uk_invitee_progress_relation"` + SysOrigin string `gorm:"column:sys_origin;size:32"` + InviterUserID int64 `gorm:"column:inviter_user_id;index:idx_invitee_progress_inviter"` + InviteeUserID int64 `gorm:"column:invitee_user_id;uniqueIndex:uk_invitee_progress_invitee"` + TotalRecharge int64 `gorm:"column:total_recharge_coins"` + ValidUser bool `gorm:"column:valid_user"` + ValidTime *time.Time `gorm:"column:valid_time"` + RewardedRuleIDs string `gorm:"column:rewarded_rule_ids;size:255"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` } func (InviteCampaignInviteeProgress) TableName() string { return "invite_campaign_invitee_progress" } type InviteCampaignMonthlyProgress struct { - ID int64 `gorm:"column:id;primaryKey"` - SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_invite_monthly_progress,priority:1"` - InviterUserID int64 `gorm:"column:inviter_user_id;uniqueIndex:uk_invite_monthly_progress,priority:2"` - MonthKey string `gorm:"column:month_key;size:6;uniqueIndex:uk_invite_monthly_progress,priority:3"` - InviteCount int64 `gorm:"column:invite_count"` - ValidUserCount int64 `gorm:"column:valid_user_count"` - TotalRecharge int64 `gorm:"column:total_recharge_coins"` - RewardGoldCoins int64 `gorm:"column:reward_gold_coins"` - CreateTime time.Time `gorm:"column:create_time"` - UpdateTime time.Time `gorm:"column:update_time"` + ID int64 `gorm:"column:id;primaryKey"` + SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_invite_monthly_progress,priority:1"` + InviterUserID int64 `gorm:"column:inviter_user_id;uniqueIndex:uk_invite_monthly_progress,priority:2"` + MonthKey string `gorm:"column:month_key;size:6;uniqueIndex:uk_invite_monthly_progress,priority:3"` + InviteCount int64 `gorm:"column:invite_count"` + ValidUserCount int64 `gorm:"column:valid_user_count"` + TotalRecharge int64 `gorm:"column:total_recharge_coins"` + RewardGoldCoins int64 `gorm:"column:reward_gold_coins"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` } func (InviteCampaignMonthlyProgress) TableName() string { return "invite_campaign_monthly_progress" } type InviteCampaignTaskClaim struct { - ID int64 `gorm:"column:id;primaryKey"` - SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_invite_task_claim,priority:1"` - InviterUserID int64 `gorm:"column:inviter_user_id;uniqueIndex:uk_invite_task_claim,priority:2"` - MonthKey string `gorm:"column:month_key;size:6;uniqueIndex:uk_invite_task_claim,priority:3"` - RuleID int64 `gorm:"column:rule_id;uniqueIndex:uk_invite_task_claim,priority:4"` - ClaimTime time.Time `gorm:"column:claim_time"` - CreateTime time.Time `gorm:"column:create_time"` - UpdateTime time.Time `gorm:"column:update_time"` + ID int64 `gorm:"column:id;primaryKey"` + SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_invite_task_claim,priority:1"` + InviterUserID int64 `gorm:"column:inviter_user_id;uniqueIndex:uk_invite_task_claim,priority:2"` + MonthKey string `gorm:"column:month_key;size:6;uniqueIndex:uk_invite_task_claim,priority:3"` + RuleID int64 `gorm:"column:rule_id;uniqueIndex:uk_invite_task_claim,priority:4"` + ClaimTime time.Time `gorm:"column:claim_time"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` } func (InviteCampaignTaskClaim) TableName() string { return "invite_campaign_task_claim" } diff --git a/internal/repo/repo.go b/internal/repo/repo.go index e30950c..4072bf5 100644 --- a/internal/repo/repo.go +++ b/internal/repo/repo.go @@ -44,6 +44,8 @@ func (r *Repository) Ping(ctx context.Context) error { func (r *Repository) AutoMigrate() error { if err := r.DB.AutoMigrate( + &model.ResidentRegisterRewardConfig{}, + &model.ResidentRegisterRewardLog{}, &model.InviteCampaignConfig{}, &model.InviteCampaignRewardRule{}, &model.InviteCampaignRelation{}, diff --git a/internal/service/register_reward.go b/internal/service/register_reward.go new file mode 100644 index 0000000..cce740f --- /dev/null +++ b/internal/service/register_reward.go @@ -0,0 +1,458 @@ +package service + +import ( + "chatapp3-golang/internal/config" + "chatapp3-golang/internal/integration" + "chatapp3-golang/internal/model" + "chatapp3-golang/internal/repo" + "chatapp3-golang/internal/util" + "context" + "errors" + "fmt" + "log" + "strconv" + "strings" + "time" + + "github.com/redis/go-redis/v9" + "gorm.io/gorm" +) + +const ( + registerRewardStatusPending = "PENDING" + registerRewardStatusSuccess = "SUCCESS" + registerRewardStatusSkipped = "SKIPPED" + registerRewardStatusFailed = "FAILED" + registerRewardGoldOrigin = "REGISTER_REWARDS" + registerRewardGoldDesc = "Register reward" + registerRewardPropsOrigin = "ACTIVITY_REWARD" +) + +type registerRewardEvent struct { + EventID string + SysOrigin string + UserID int64 + CountryID int64 +} + +type registerRewardResolvedConfig struct { + GoldAmount int64 + RewardGroupID *int64 +} + +type RegisterRewardService struct { + cfg config.Config + repo *repo.Repository + java *integration.Client +} + +func NewRegisterRewardService(cfg config.Config, repository *repo.Repository, javaClient *integration.Client) *RegisterRewardService { + return &RegisterRewardService{ + cfg: cfg, + repo: repository, + java: javaClient, + } +} + +func (s *RegisterRewardService) Start(ctx context.Context) error { + if strings.TrimSpace(s.cfg.RegisterRewardStreamKey) == "" { + log.Printf("register reward consumer disabled: stream key is blank") + return nil + } + if err := s.ensureConsumerGroup(ctx); err != nil { + return err + } + log.Printf( + "register reward consumer started. stream=%s group=%s consumer=%s", + s.cfg.RegisterRewardStreamKey, + s.cfg.RegisterRewardConsumerGroup, + s.cfg.RegisterRewardConsumerName, + ) + go s.consumeLoop(ctx) + return nil +} + +func (s *RegisterRewardService) consumeLoop(ctx context.Context) { + for { + if ctx.Err() != nil { + return + } + + reclaimed, err := s.reclaimPending(ctx) + if err != nil { + s.handleLoopError(ctx, "reclaim register reward messages", err) + continue + } + if reclaimed > 0 { + continue + } + + err = s.readNewMessages(ctx) + if err == nil || errors.Is(err, redis.Nil) { + continue + } + s.handleLoopError(ctx, "read register reward messages", err) + } +} + +func (s *RegisterRewardService) handleLoopError(ctx context.Context, action string, err error) { + if ctx.Err() != nil { + return + } + if isRegisterRewardNoGroupErr(err) { + if ensureErr := s.ensureConsumerGroup(ctx); ensureErr != nil { + log.Printf("%s failed and recreate consumer group failed: %v", action, ensureErr) + sleepWithContext(ctx, time.Second) + return + } + return + } + log.Printf("%s failed: %v", action, err) + sleepWithContext(ctx, time.Second) +} + +func (s *RegisterRewardService) ensureConsumerGroup(ctx context.Context) error { + err := s.repo.Redis.XGroupCreateMkStream( + ctx, + s.cfg.RegisterRewardStreamKey, + s.cfg.RegisterRewardConsumerGroup, + "0", + ).Err() + if err != nil && !strings.Contains(strings.ToUpper(err.Error()), "BUSYGROUP") { + return err + } + return nil +} + +func (s *RegisterRewardService) reclaimPending(ctx context.Context) (int, error) { + messages, _, err := s.repo.Redis.XAutoClaim(ctx, &redis.XAutoClaimArgs{ + Stream: s.cfg.RegisterRewardStreamKey, + Group: s.cfg.RegisterRewardConsumerGroup, + Consumer: s.cfg.RegisterRewardConsumerName, + MinIdle: s.cfg.RegisterRewardMinIdle, + Start: "0-0", + Count: s.cfg.RegisterRewardBatchSize, + }).Result() + if err != nil { + return 0, err + } + if len(messages) == 0 { + return 0, nil + } + return len(messages), s.processMessages(ctx, messages) +} + +func (s *RegisterRewardService) readNewMessages(ctx context.Context) error { + streams, err := s.repo.Redis.XReadGroup(ctx, &redis.XReadGroupArgs{ + Group: s.cfg.RegisterRewardConsumerGroup, + Consumer: s.cfg.RegisterRewardConsumerName, + Streams: []string{s.cfg.RegisterRewardStreamKey, ">"}, + Count: s.cfg.RegisterRewardBatchSize, + Block: s.cfg.RegisterRewardBlock, + }).Result() + if err != nil { + return err + } + for _, stream := range streams { + if err := s.processMessages(ctx, stream.Messages); err != nil { + return err + } + } + return nil +} + +func (s *RegisterRewardService) processMessages(ctx context.Context, messages []redis.XMessage) error { + for _, message := range messages { + ack, err := s.processMessage(ctx, message) + if ack { + if ackErr := s.repo.Redis.XAck( + context.Background(), + s.cfg.RegisterRewardStreamKey, + s.cfg.RegisterRewardConsumerGroup, + message.ID, + ).Err(); ackErr != nil { + log.Printf("ack register reward message failed. streamId=%s err=%v", message.ID, ackErr) + } + } + if err != nil { + return err + } + } + return nil +} + +func (s *RegisterRewardService) processMessage(ctx context.Context, message redis.XMessage) (bool, error) { + event, err := parseRegisterRewardEvent(message) + if err != nil { + log.Printf("drop malformed register reward message. streamId=%s err=%v values=%v", message.ID, err, message.Values) + return true, nil + } + + logRecord, terminal, err := s.getOrCreateLog(ctx, event, message.ID) + if err != nil { + return false, err + } + if terminal { + return true, nil + } + + resolvedConfig, ack, err := s.resolveRewardConfig(ctx, logRecord, event) + if err != nil || ack { + return ack, err + } + + if err := s.dispatchGold(ctx, event, resolvedConfig); err != nil { + _ = s.updateLogStatus(ctx, logRecord.ID, registerRewardStatusFailed, err.Error(), resolvedConfig) + return false, err + } + if err := s.dispatchProps(ctx, logRecord.ID, event, resolvedConfig); err != nil { + _ = s.updateLogStatus(ctx, logRecord.ID, registerRewardStatusFailed, err.Error(), resolvedConfig) + return false, err + } + if err := s.updateLogStatus(ctx, logRecord.ID, registerRewardStatusSuccess, "", resolvedConfig); err != nil { + return false, err + } + return true, nil +} + +func (s *RegisterRewardService) resolveRewardConfig(ctx context.Context, logRecord *model.ResidentRegisterRewardLog, event registerRewardEvent) (registerRewardResolvedConfig, bool, error) { + if logRecord.Status == registerRewardStatusFailed && (logRecord.GoldAmount > 0 || logRecord.RewardGroupID != nil) { + return registerRewardResolvedConfig{ + GoldAmount: logRecord.GoldAmount, + RewardGroupID: logRecord.RewardGroupID, + }, false, nil + } + + cfg, err := s.loadRegisterRewardConfig(ctx, event.SysOrigin) + if err != nil { + return registerRewardResolvedConfig{}, false, err + } + if cfg == nil || !cfg.Enabled { + err = s.updateLogStatus( + ctx, + logRecord.ID, + registerRewardStatusSkipped, + "register reward config missing or disabled", + registerRewardResolvedConfig{}, + ) + return registerRewardResolvedConfig{}, err == nil, err + } + + resolved := registerRewardResolvedConfig{ + GoldAmount: cfg.GoldAmount, + RewardGroupID: cfg.RewardGroupID, + } + if resolved.GoldAmount <= 0 && resolved.RewardGroupID == nil { + err = s.updateLogStatus( + ctx, + logRecord.ID, + registerRewardStatusSkipped, + "register reward config has no reward payload", + resolved, + ) + return registerRewardResolvedConfig{}, err == nil, err + } + if err := s.updateLogStatus(ctx, logRecord.ID, registerRewardStatusPending, "", resolved); err != nil { + return registerRewardResolvedConfig{}, false, err + } + return resolved, false, nil +} + +func (s *RegisterRewardService) loadRegisterRewardConfig(ctx context.Context, sysOrigin string) (*model.ResidentRegisterRewardConfig, error) { + var cfg model.ResidentRegisterRewardConfig + err := s.repo.DB.WithContext(ctx). + Where("sys_origin = ?", strings.ToUpper(strings.TrimSpace(sysOrigin))). + First(&cfg).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return &cfg, nil +} + +func (s *RegisterRewardService) dispatchGold(ctx context.Context, event registerRewardEvent, resolved registerRewardResolvedConfig) error { + if resolved.GoldAmount <= 0 { + return nil + } + return s.java.ChangeGoldBalance(ctx, integration.GoldReceiptCommand{ + ReceiptType: "INCOME", + UserID: event.UserID, + SysOrigin: event.SysOrigin, + EventID: event.EventID + ":gold", + Remark: registerRewardGoldDesc, + Amount: integration.NewPennyAmountPayloadFromDollar(resolved.GoldAmount), + CloseDelayAsset: false, + OpUserType: "APP", + CustomizeOrigin: registerRewardGoldOrigin, + CustomizeOriginDesc: registerRewardGoldDesc, + }) +} + +func (s *RegisterRewardService) dispatchProps(ctx context.Context, logID int64, event registerRewardEvent, resolved registerRewardResolvedConfig) error { + if resolved.RewardGroupID == nil { + return nil + } + return s.java.GrantProps(ctx, integration.GrantPropsRequest{ + TrackID: logID, + AcceptUserID: event.UserID, + SysOrigin: event.SysOrigin, + Origin: registerRewardPropsOrigin, + SourceGroupID: *resolved.RewardGroupID, + }) +} + +func (s *RegisterRewardService) getOrCreateLog(ctx context.Context, event registerRewardEvent, streamMessageID string) (*model.ResidentRegisterRewardLog, bool, error) { + var record model.ResidentRegisterRewardLog + err := s.repo.DB.WithContext(ctx). + Where("event_id = ?", event.EventID). + First(&record).Error + if err == nil { + if record.StreamMessageID != streamMessageID { + if updateErr := s.repo.DB.WithContext(ctx). + Model(&model.ResidentRegisterRewardLog{}). + Where("id = ?", record.ID). + Updates(map[string]any{ + "stream_message_id": streamMessageID, + "update_time": time.Now(), + }).Error; updateErr == nil { + record.StreamMessageID = streamMessageID + } + } + return &record, isRegisterRewardTerminalStatus(record.Status), nil + } + if !errors.Is(err, gorm.ErrRecordNotFound) { + return nil, false, err + } + + id, idErr := util.NextID() + if idErr != nil { + return nil, false, idErr + } + now := time.Now() + record = model.ResidentRegisterRewardLog{ + ID: id, + EventID: event.EventID, + StreamMessageID: streamMessageID, + SysOrigin: event.SysOrigin, + UserID: event.UserID, + Status: registerRewardStatusPending, + CreateTime: now, + UpdateTime: now, + } + if err := s.repo.DB.WithContext(ctx).Create(&record).Error; err != nil { + if strings.Contains(strings.ToLower(err.Error()), "duplicate") { + return s.getOrCreateLog(ctx, event, streamMessageID) + } + return nil, false, err + } + return &record, false, nil +} + +func (s *RegisterRewardService) updateLogStatus(ctx context.Context, logID int64, status, errorMessage string, resolved registerRewardResolvedConfig) error { + return s.repo.DB.WithContext(ctx). + Model(&model.ResidentRegisterRewardLog{}). + Where("id = ?", logID). + Updates(map[string]any{ + "status": status, + "error_message": truncateRegisterRewardMessage(errorMessage), + "gold_amount": resolved.GoldAmount, + "reward_group_id": resolved.RewardGroupID, + "update_time": time.Now(), + }).Error +} + +func parseRegisterRewardEvent(message redis.XMessage) (registerRewardEvent, error) { + eventID := strings.TrimSpace(toRegisterRewardString(message.Values["eventId"])) + if eventID == "" { + return registerRewardEvent{}, fmt.Errorf("eventId is required") + } + sysOrigin := strings.ToUpper(strings.TrimSpace(toRegisterRewardString(message.Values["sysOrigin"]))) + if sysOrigin == "" { + return registerRewardEvent{}, fmt.Errorf("sysOrigin is required") + } + userID, err := toRegisterRewardInt64(message.Values["userId"]) + if err != nil { + return registerRewardEvent{}, fmt.Errorf("invalid userId: %w", err) + } + if userID <= 0 { + return registerRewardEvent{}, fmt.Errorf("invalid userId") + } + countryID, _ := toRegisterRewardInt64(message.Values["countryId"]) + return registerRewardEvent{ + EventID: eventID, + SysOrigin: sysOrigin, + UserID: userID, + CountryID: countryID, + }, nil +} + +func toRegisterRewardString(value any) string { + switch typed := value.(type) { + case nil: + return "" + case string: + return typed + case []byte: + return string(typed) + default: + return fmt.Sprint(typed) + } +} + +func toRegisterRewardInt64(value any) (int64, error) { + switch typed := value.(type) { + case nil: + return 0, fmt.Errorf("value is nil") + case int64: + return typed, nil + case int32: + return int64(typed), nil + case int: + return int64(typed), nil + case uint64: + return int64(typed), nil + case float64: + return int64(typed), nil + case string: + return strconv.ParseInt(strings.TrimSpace(typed), 10, 64) + case []byte: + return strconv.ParseInt(strings.TrimSpace(string(typed)), 10, 64) + default: + return strconv.ParseInt(strings.TrimSpace(fmt.Sprint(typed)), 10, 64) + } +} + +func isRegisterRewardTerminalStatus(status string) bool { + switch strings.ToUpper(strings.TrimSpace(status)) { + case registerRewardStatusSuccess, registerRewardStatusSkipped: + return true + default: + return false + } +} + +func isRegisterRewardNoGroupErr(err error) bool { + if err == nil { + return false + } + return strings.Contains(strings.ToUpper(err.Error()), "NOGROUP") +} + +func truncateRegisterRewardMessage(message string) string { + message = strings.TrimSpace(message) + if len(message) <= 255 { + return message + } + return message[:255] +} + +func sleepWithContext(ctx context.Context, duration time.Duration) { + timer := time.NewTimer(duration) + defer timer.Stop() + select { + case <-ctx.Done(): + case <-timer.C: + } +}