diff --git a/cmd/week-star-job/main.go b/cmd/week-star-job/main.go index 35d635e..d6711e2 100644 --- a/cmd/week-star-job/main.go +++ b/cmd/week-star-job/main.go @@ -29,15 +29,18 @@ func main() { javaClient := integration.New(cfg) settlementService := service.NewWeekStarSettlementService(cfg, repository, javaClient) roomTurnoverRewardService := service.NewRoomTurnoverRewardSettlementService(cfg, repository, javaClient) + rechargeRewardService := service.NewRechargeRewardSettlementService(cfg, repository, javaClient) workerCtx, workerCancel := context.WithCancel(context.Background()) defer workerCancel() settlementService.Start(workerCtx) roomTurnoverRewardService.Start(workerCtx) + rechargeRewardService.Start(workerCtx) log.Printf( - "week-star-job started. weekStarInterval=%s roomTurnoverInterval=%s timezone=%s javaOther=%s javaWallet=%s", + "week-star-job started. weekStarInterval=%s roomTurnoverInterval=%s rechargeRewardInterval=%s timezone=%s javaOther=%s javaWallet=%s", cfg.WeekStarSettleInterval, cfg.RoomTurnoverSettleInterval, + cfg.RechargeRewardSettleInterval, cfg.WeekStarTimezone, cfg.JavaOtherBaseURL, cfg.JavaWalletBaseURL, diff --git a/cmd/week-star-settle-once/main.go b/cmd/week-star-settle-once/main.go index db71623..32490b3 100644 --- a/cmd/week-star-settle-once/main.go +++ b/cmd/week-star-settle-once/main.go @@ -27,12 +27,16 @@ func main() { javaClient := integration.New(cfg) settlementService := service.NewWeekStarSettlementService(cfg, repository, javaClient) roomTurnoverRewardService := service.NewRoomTurnoverRewardSettlementService(cfg, repository, javaClient) + rechargeRewardService := service.NewRechargeRewardSettlementService(cfg, repository, javaClient) if err := settlementService.SettleDueCycles(ctx); err != nil { log.Fatalf("settle due cycles failed: %v", err) } if err := roomTurnoverRewardService.SettleDueCycles(ctx); err != nil { log.Fatalf("settle room turnover reward due cycles failed: %v", err) } + if err := rechargeRewardService.SettleDueCycles(ctx); err != nil { + log.Fatalf("settle recharge reward due cycles failed: %v", err) + } - log.Printf("week-star and room-turnover-reward settle once finished") + log.Printf("week-star, room-turnover-reward and recharge-reward settle once finished") } diff --git a/internal/config/config.go b/internal/config/config.go index 6d114c6..e8528b6 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -8,27 +8,33 @@ import ( ) type Config struct { - MySQLDSN string - RedisAddr string - RedisPassword string - RedisDB int - JavaOtherBaseURL string - JavaWalletBaseURL string - WeekStarDefaultSysOrigin string - WeekStarTimezone string - WeekStarSettleInterval time.Duration - RoomTurnoverDefaultSysOrigin string - RoomTurnoverTimezone string - RoomTurnoverSettleInterval time.Duration - WeekStarRetryStreamKey string - WeekStarRetryConsumerGroup string - WeekStarRetryConsumerName string - WeekStarRetryBatchSize int64 - WeekStarRetryBlock time.Duration - WeekStarRetryMinIdle time.Duration - Timeout time.Duration - ShutdownTimeout time.Duration - AutoMigrate bool + MySQLDSN string + RedisAddr string + RedisPassword string + RedisDB int + JavaOtherBaseURL string + JavaWalletBaseURL string + WeekStarDefaultSysOrigin string + WeekStarTimezone string + WeekStarSettleInterval time.Duration + RoomTurnoverDefaultSysOrigin string + RoomTurnoverTimezone string + RoomTurnoverSettleInterval time.Duration + RechargeRewardDefaultSysOrigin string + RechargeRewardTimezone string + RechargeRewardStorageTimezone string + RechargeRewardStartDate string + RechargeRewardSettleInterval time.Duration + RechargeRewardCoinSellerGoldPerUSD int64 + WeekStarRetryStreamKey string + WeekStarRetryConsumerGroup string + WeekStarRetryConsumerName string + WeekStarRetryBatchSize int64 + WeekStarRetryBlock time.Duration + WeekStarRetryMinIdle time.Duration + Timeout time.Duration + ShutdownTimeout time.Duration + AutoMigrate bool } func Load() Config { @@ -74,6 +80,30 @@ func Load() Config { []string{"CHATAPP_ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS", "ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS"}, 60, )) * time.Second, + RechargeRewardDefaultSysOrigin: strings.ToUpper(getEnvAny( + []string{"CHATAPP_RECHARGE_REWARD_DEFAULT_SYS_ORIGIN", "RECHARGE_REWARD_DEFAULT_SYS_ORIGIN", "CHATAPP_WEEK_STAR_DEFAULT_SYS_ORIGIN", "WEEK_STAR_DEFAULT_SYS_ORIGIN"}, + "LIKEI", + )), + RechargeRewardTimezone: getEnvAny( + []string{"CHATAPP_RECHARGE_REWARD_TIMEZONE", "RECHARGE_REWARD_TIMEZONE", "CHATAPP_WEEK_STAR_TIMEZONE", "WEEK_STAR_TIMEZONE"}, + "Asia/Riyadh", + ), + RechargeRewardStorageTimezone: getEnvAny( + []string{"CHATAPP_RECHARGE_REWARD_STORAGE_TIMEZONE", "RECHARGE_REWARD_STORAGE_TIMEZONE"}, + "Asia/Shanghai", + ), + RechargeRewardStartDate: getEnvAny( + []string{"CHATAPP_RECHARGE_REWARD_START_DATE", "RECHARGE_REWARD_START_DATE"}, + "2026-05-21", + ), + RechargeRewardSettleInterval: time.Duration(getEnvIntAny( + []string{"CHATAPP_RECHARGE_REWARD_SETTLE_INTERVAL_SECONDS", "RECHARGE_REWARD_SETTLE_INTERVAL_SECONDS"}, + 60, + )) * time.Second, + RechargeRewardCoinSellerGoldPerUSD: int64(getEnvIntAny( + []string{"CHATAPP_RECHARGE_REWARD_COIN_SELLER_GOLD_PER_USD", "RECHARGE_REWARD_COIN_SELLER_GOLD_PER_USD"}, + 100000, + )), WeekStarRetryStreamKey: getEnvAny([]string{"CHATAPP_WEEK_STAR_RETRY_STREAM_KEY", "WEEK_STAR_RETRY_STREAM_KEY"}, "week-star:reward-retry"), WeekStarRetryConsumerGroup: getEnvAny([]string{"CHATAPP_WEEK_STAR_RETRY_CONSUMER_GROUP", "WEEK_STAR_RETRY_CONSUMER_GROUP"}, "week-star-retry"), WeekStarRetryConsumerName: getEnvAny([]string{"CHATAPP_WEEK_STAR_RETRY_CONSUMER_NAME", "WEEK_STAR_RETRY_CONSUMER_NAME"}, defaultConsumerName("week-star-retry")), diff --git a/internal/config/config_test.go b/internal/config/config_test.go index af08db3..ddb6c8f 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -19,6 +19,12 @@ func TestLoadPrefersChatappKeys(t *testing.T) { t.Setenv("CHATAPP_ROOM_TURNOVER_REWARD_DEFAULT_SYS_ORIGIN", "likei") t.Setenv("CHATAPP_ROOM_TURNOVER_REWARD_TIMEZONE", "Asia/Riyadh") t.Setenv("CHATAPP_ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS", "180") + t.Setenv("CHATAPP_RECHARGE_REWARD_DEFAULT_SYS_ORIGIN", "likei") + t.Setenv("CHATAPP_RECHARGE_REWARD_TIMEZONE", "Asia/Riyadh") + t.Setenv("CHATAPP_RECHARGE_REWARD_STORAGE_TIMEZONE", "Asia/Shanghai") + t.Setenv("CHATAPP_RECHARGE_REWARD_START_DATE", "2026-05-21") + t.Setenv("CHATAPP_RECHARGE_REWARD_SETTLE_INTERVAL_SECONDS", "90") + t.Setenv("CHATAPP_RECHARGE_REWARD_COIN_SELLER_GOLD_PER_USD", "100000") t.Setenv("CHATAPP_WEEK_STAR_RETRY_STREAM_KEY", "chatapp-stream") t.Setenv("CHATAPP_WEEK_STAR_RETRY_CONSUMER_GROUP", "chatapp-group") t.Setenv("CHATAPP_WEEK_STAR_RETRY_CONSUMER_NAME", "chatapp-consumer") @@ -55,6 +61,12 @@ func TestLoadPrefersChatappKeys(t *testing.T) { if cfg.RoomTurnoverDefaultSysOrigin != "LIKEI" || cfg.RoomTurnoverTimezone != "Asia/Riyadh" || cfg.RoomTurnoverSettleInterval != 180*time.Second { t.Fatalf("unexpected room turnover config: %+v", cfg) } + if cfg.RechargeRewardDefaultSysOrigin != "LIKEI" || cfg.RechargeRewardTimezone != "Asia/Riyadh" || cfg.RechargeRewardStorageTimezone != "Asia/Shanghai" { + t.Fatalf("unexpected recharge reward timezone config: %+v", cfg) + } + if cfg.RechargeRewardStartDate != "2026-05-21" || cfg.RechargeRewardSettleInterval != 90*time.Second || cfg.RechargeRewardCoinSellerGoldPerUSD != 100000 { + t.Fatalf("unexpected recharge reward settlement config: %+v", cfg) + } if cfg.WeekStarRetryStreamKey != "chatapp-stream" || cfg.WeekStarRetryConsumerGroup != "chatapp-group" || cfg.WeekStarRetryConsumerName != "chatapp-consumer" { t.Fatalf("unexpected retry consumer config: %+v", cfg) } @@ -82,6 +94,12 @@ func TestLoadFallsBackToLegacyKeys(t *testing.T) { t.Setenv("ROOM_TURNOVER_REWARD_DEFAULT_SYS_ORIGIN", "tarab") t.Setenv("ROOM_TURNOVER_REWARD_TIMEZONE", "Asia/Shanghai") t.Setenv("ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS", "77") + t.Setenv("RECHARGE_REWARD_DEFAULT_SYS_ORIGIN", "tarab") + t.Setenv("RECHARGE_REWARD_TIMEZONE", "Asia/Riyadh") + t.Setenv("RECHARGE_REWARD_STORAGE_TIMEZONE", "UTC") + t.Setenv("RECHARGE_REWARD_START_DATE", "2026-05-21") + t.Setenv("RECHARGE_REWARD_SETTLE_INTERVAL_SECONDS", "88") + t.Setenv("RECHARGE_REWARD_COIN_SELLER_GOLD_PER_USD", "100000") t.Setenv("WEEK_STAR_RETRY_STREAM_KEY", "legacy-stream") t.Setenv("WEEK_STAR_RETRY_CONSUMER_GROUP", "legacy-group") t.Setenv("WEEK_STAR_RETRY_CONSUMER_NAME", "legacy-consumer") @@ -115,6 +133,12 @@ func TestLoadFallsBackToLegacyKeys(t *testing.T) { if cfg.RoomTurnoverDefaultSysOrigin != "TARAB" || cfg.RoomTurnoverTimezone != "Asia/Shanghai" || cfg.RoomTurnoverSettleInterval != 77*time.Second { t.Fatalf("unexpected legacy room turnover config: %+v", cfg) } + if cfg.RechargeRewardDefaultSysOrigin != "TARAB" || cfg.RechargeRewardTimezone != "Asia/Riyadh" || cfg.RechargeRewardStorageTimezone != "UTC" { + t.Fatalf("unexpected legacy recharge reward timezone config: %+v", cfg) + } + if cfg.RechargeRewardStartDate != "2026-05-21" || cfg.RechargeRewardSettleInterval != 88*time.Second || cfg.RechargeRewardCoinSellerGoldPerUSD != 100000 { + t.Fatalf("unexpected legacy recharge reward settlement config: %+v", cfg) + } if cfg.WeekStarRetryStreamKey != "legacy-stream" || cfg.WeekStarRetryConsumerGroup != "legacy-group" || cfg.WeekStarRetryConsumerName != "legacy-consumer" { t.Fatalf("unexpected retry consumer config: %+v", cfg) } diff --git a/internal/integration/java.go b/internal/integration/java.go index 9efcaac..81d9ead 100644 --- a/internal/integration/java.go +++ b/internal/integration/java.go @@ -138,7 +138,11 @@ func New(cfg config.Config) *Client { } func (c *Client) SendActivityReward(ctx context.Context, req SendActivityRewardRequest) error { - return c.postJSON(ctx, c.cfg.JavaOtherBaseURL+"/props-activity-cnf/client/sendActivityReward", req, nil) + var resp resultResponse[json.RawMessage] + if err := c.postJSON(ctx, c.cfg.JavaOtherBaseURL+"/props-activity-cnf/client/sendActivityReward", req, &resp); err != nil { + return err + } + return ensureSuccessfulResult(resp.Code, resp.Message, resp.Success) } func (c *Client) ListLastWeekRoomContribution(ctx context.Context) ([]RoomContributionActivityCount, error) { diff --git a/internal/model/models.go b/internal/model/models.go index a43d6c3..123fa0f 100644 --- a/internal/model/models.go +++ b/internal/model/models.go @@ -152,3 +152,67 @@ type RoomTurnoverRewardRecord struct { } func (RoomTurnoverRewardRecord) TableName() string { return "room_turnover_reward_record" } + +// RechargeRewardConfig 保存充值奖励活动的系统级配置。 +type RechargeRewardConfig struct { + ID int64 `gorm:"column:id;primaryKey"` + SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_recharge_reward_sys_origin"` + Enabled bool `gorm:"column:enabled;index:idx_recharge_reward_enabled"` + Timezone string `gorm:"column:timezone;size:64"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` +} + +func (RechargeRewardConfig) TableName() string { return "recharge_reward_config" } + +// RechargeRewardLevel 保存达到某个充值门槛后发放的奖励配置。 +type RechargeRewardLevel struct { + ID int64 `gorm:"column:id;primaryKey"` + ConfigID int64 `gorm:"column:config_id;uniqueIndex:uk_recharge_reward_level,priority:1;uniqueIndex:uk_recharge_reward_amount,priority:1;index:idx_recharge_reward_level_config"` + Level int `gorm:"column:level;uniqueIndex:uk_recharge_reward_level,priority:2"` + RechargeAmountCents int64 `gorm:"column:recharge_amount_cents;uniqueIndex:uk_recharge_reward_amount,priority:2"` + RewardGold int64 `gorm:"column:reward_gold"` + RewardGroupID *int64 `gorm:"column:reward_group_id"` + RewardGroupName string `gorm:"column:reward_group_name;size:255"` + Enabled bool `gorm:"column:enabled"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` +} + +func (RechargeRewardLevel) TableName() string { return "recharge_reward_level" } + +// RechargeRewardGrantRecord 记录某个充值周期内用户各档位的实际发放状态。 +type RechargeRewardGrantRecord struct { + ID int64 `gorm:"column:id;primaryKey"` + ConfigID int64 `gorm:"column:config_id;index:idx_recharge_reward_grant_config_cycle,priority:1"` + LevelID int64 `gorm:"column:level_id"` + SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_recharge_reward_grant_level,priority:1;index:idx_recharge_reward_grant_date,priority:1"` + CycleKey string `gorm:"column:cycle_key;size:32;uniqueIndex:uk_recharge_reward_grant_level,priority:2;index:idx_recharge_reward_grant_config_cycle,priority:2"` + RechargeDate int `gorm:"column:recharge_date;index:idx_recharge_reward_grant_user,priority:2;index:idx_recharge_reward_grant_date,priority:2"` + PeriodStartAt time.Time `gorm:"column:period_start_at"` + PeriodEndAt time.Time `gorm:"column:period_end_at"` + UserID int64 `gorm:"column:user_id;uniqueIndex:uk_recharge_reward_grant_level,priority:3;index:idx_recharge_reward_grant_user,priority:1"` + Account string `gorm:"column:account;size:64"` + UserAvatar string `gorm:"column:user_avatar;size:1024"` + UserNickname string `gorm:"column:user_nickname;size:255"` + CountryCode string `gorm:"column:country_code;size:32"` + CountryName string `gorm:"column:country_name;size:128"` + RechargeAmountCents int64 `gorm:"column:recharge_amount_cents"` + Level int `gorm:"column:level;uniqueIndex:uk_recharge_reward_grant_level,priority:4"` + RechargeThresholdCents int64 `gorm:"column:recharge_threshold_cents"` + RewardGold int64 `gorm:"column:reward_gold"` + RewardGroupID *int64 `gorm:"column:reward_group_id"` + RewardGroupName string `gorm:"column:reward_group_name;size:255"` + RewardGroupTrackID int64 `gorm:"column:reward_group_track_id"` + GoldEventID string `gorm:"column:gold_event_id;size:128;index:idx_recharge_reward_gold_event"` + Status string `gorm:"column:status;size:32;index:idx_recharge_reward_grant_config_cycle,priority:3;index:idx_recharge_reward_grant_date,priority:3"` + RewardGroupStatus string `gorm:"column:reward_group_status;size:32"` + RewardGoldStatus string `gorm:"column:reward_gold_status;size:32"` + RetryCount int `gorm:"column:retry_count"` + LastError string `gorm:"column:last_error;size:1024"` + SentAt *time.Time `gorm:"column:sent_at"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` +} + +func (RechargeRewardGrantRecord) TableName() string { return "recharge_reward_grant_record" } diff --git a/internal/service/recharge_reward_settlement.go b/internal/service/recharge_reward_settlement.go new file mode 100644 index 0000000..7f57040 --- /dev/null +++ b/internal/service/recharge_reward_settlement.go @@ -0,0 +1,663 @@ +package service + +import ( + "context" + "fmt" + "log" + "sort" + "strconv" + "strings" + "time" + + "chatapp-cron/internal/config" + "chatapp-cron/internal/integration" + "chatapp-cron/internal/model" + "chatapp-cron/internal/repo" + "chatapp-cron/internal/util" + + "gorm.io/gorm/clause" +) + +const ( + rechargeRewardStatusPending = "PENDING" + rechargeRewardStatusSuccess = "SUCCESS" + rechargeRewardStatusFailed = "FAILED" + rechargeRewardStatusSkipped = "SKIPPED" + + rechargeRewardOrigin = "CUMULATIVE_RECHARGE_REWARDS" +) + +var rechargeRewardOrderPlatforms = []string{ + "GOOGLE", + "PAYER_MAX", + "AIRWALLEX", + "PAYNICORN", + "STRIPE", + "PAY_PAL", + "CLIPSPAY", +} + +type rechargeRewardLevel struct { + ID int64 + Level int + RechargeAmountCents int64 + RewardGold int64 + RewardGroupID *int64 + RewardGroupName string +} + +type rechargeRewardUserAmount struct { + UserID int64 `gorm:"column:user_id"` + RechargeAmountCents int64 `gorm:"column:recharge_amount_cents"` +} + +type rechargeRewardPeriod struct { + CycleKey string + RechargeDate int + StartAt time.Time + EndAt time.Time +} + +// RechargeRewardSettlementService 扫描充值奖励配置，按周期累计充值并自动发奖。 +type RechargeRewardSettlementService struct { + cfg config.Config + repo *repo.Repository + java *integration.Client + location *time.Location + storageLocation *time.Location +} + +func NewRechargeRewardSettlementService(cfg config.Config, repository *repo.Repository, javaClient *integration.Client) *RechargeRewardSettlementService { + location, err := time.LoadLocation(strings.TrimSpace(cfg.RechargeRewardTimezone)) + if err != nil { + location = time.UTC + } + storageLocation, err := time.LoadLocation(strings.TrimSpace(cfg.RechargeRewardStorageTimezone)) + if err != nil { + storageLocation = location + } + return &RechargeRewardSettlementService{ + cfg: cfg, + repo: repository, + java: javaClient, + location: location, + storageLocation: storageLocation, + } +} + +func (s *RechargeRewardSettlementService) Start(ctx context.Context) { + interval := s.cfg.RechargeRewardSettleInterval + if interval <= 0 { + interval = time.Minute + } + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + if ctx.Err() != nil { + return + } + if err := s.SettleDueCycles(ctx); err != nil && ctx.Err() == nil { + log.Printf("recharge reward settlement scan failed: %v", err) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } + }() +} + +func (s *RechargeRewardSettlementService) SettleDueCycles(ctx context.Context) error { + var configs []model.RechargeRewardConfig + if err := s.repo.DB.WithContext(ctx). + Where("enabled = ?", true). + Order("id desc"). + Find(&configs).Error; err != nil { + return err + } + if len(configs) == 0 { + return nil + } + + now := time.Now() + var lastErr error + for index := range configs { + if err := s.settleConfig(ctx, &configs[index], now); err != nil { + lastErr = err + log.Printf("recharge reward settle config failed. configId=%d err=%v", configs[index].ID, err) + } + } + return lastErr +} + +func (s *RechargeRewardSettlementService) settleConfig(ctx context.Context, configRow *model.RechargeRewardConfig, now time.Time) error { + if configRow == nil { + return nil + } + location := rechargeRewardLocation(configRow.Timezone, s.location) + startAt := rechargeRewardStartAt(s.cfg.RechargeRewardStartDate, location) + period, ok := rechargeRewardCurrentPeriod(now.In(location), startAt, location) + if !ok { + return nil + } + + lockKey := fmt.Sprintf("recharge-reward:settle-lock:%d:%s", configRow.ID, period.CycleKey) + acquired, err := s.repo.Redis.SetNX(ctx, lockKey, 1, 2*time.Minute).Result() + if err != nil { + return err + } + if !acquired { + return nil + } + + sysOrigin := normalizeRechargeRewardSysOrigin(configRow.SysOrigin, s.cfg.RechargeRewardDefaultSysOrigin) + var lastErr error + if err := s.settleCurrentPeriod(ctx, configRow, sysOrigin, period); err != nil { + lastErr = err + } + if err := s.dispatchUnfinishedRecords(ctx, configRow.ID); err != nil { + lastErr = err + } + return lastErr +} + +func (s *RechargeRewardSettlementService) settleCurrentPeriod(ctx context.Context, configRow *model.RechargeRewardConfig, sysOrigin string, period rechargeRewardPeriod) error { + levels, err := s.loadLevels(ctx, configRow.ID) + if err != nil { + return err + } + if len(levels) == 0 { + return nil + } + minThreshold := levels[0].RechargeAmountCents + amounts, err := s.loadRechargeAmounts(ctx, sysOrigin, period, minThreshold) + if err != nil { + return err + } + if len(amounts) == 0 { + return nil + } + + profiles, err := s.lookupUserProfiles(ctx, rechargeRewardAmountUserIDs(amounts)) + if err != nil { + return err + } + + now := time.Now() + var lastErr error + for _, amount := range amounts { + for _, level := range levels { + if amount.RechargeAmountCents < level.RechargeAmountCents { + continue + } + record, err := s.ensureGrantRecord(ctx, configRow.ID, sysOrigin, period, amount, level, profiles[amount.UserID], now) + if err != nil { + lastErr = err + continue + } + if record.Status == rechargeRewardStatusSuccess { + continue + } + if err := s.dispatchRewardRecord(ctx, &record); err != nil { + lastErr = err + } + } + } + return lastErr +} + +func (s *RechargeRewardSettlementService) loadLevels(ctx context.Context, configID int64) ([]rechargeRewardLevel, error) { + var rows []model.RechargeRewardLevel + if err := s.repo.DB.WithContext(ctx). + Where("config_id = ? AND enabled = ?", configID, true). + Order("recharge_amount_cents asc"). + Order("level asc"). + Find(&rows).Error; err != nil { + return nil, err + } + levels := make([]rechargeRewardLevel, 0, len(rows)) + for _, row := range rows { + if row.RechargeAmountCents <= 0 || (row.RewardGold <= 0 && (row.RewardGroupID == nil || *row.RewardGroupID <= 0)) { + continue + } + levels = append(levels, rechargeRewardLevel{ + ID: row.ID, + Level: row.Level, + RechargeAmountCents: row.RechargeAmountCents, + RewardGold: row.RewardGold, + RewardGroupID: normalizeRechargeRewardGroupID(row.RewardGroupID), + RewardGroupName: strings.TrimSpace(row.RewardGroupName), + }) + } + sort.Slice(levels, func(i, j int) bool { + if levels[i].RechargeAmountCents == levels[j].RechargeAmountCents { + return levels[i].Level < levels[j].Level + } + return levels[i].RechargeAmountCents < levels[j].RechargeAmountCents + }) + return levels, nil +} + +func (s *RechargeRewardSettlementService) loadRechargeAmounts(ctx context.Context, sysOrigin string, period rechargeRewardPeriod, minThresholdCents int64) ([]rechargeRewardUserAmount, error) { + startAt := rechargeRewardStorageTime(period.StartAt, s.storageLocation) + endAt := rechargeRewardStorageTime(period.EndAt, s.storageLocation) + coinSellerGoldPerUSD := s.cfg.RechargeRewardCoinSellerGoldPerUSD + if coinSellerGoldPerUSD <= 0 { + coinSellerGoldPerUSD = 100000 + } + + const query = ` +SELECT + source.user_id, + CAST(SUM(source.amount_cents) AS SIGNED) AS recharge_amount_cents +FROM ( + SELECT + t.user_id, + CAST(ROUND(SUM(IFNULL(t.unit_price, 0)) * 100, 0) AS SIGNED) AS amount_cents + FROM order_purchase_history AS t + INNER JOIN user_base_info AS u ON u.id = t.user_id + WHERE t.evn = 'PROD' + AND t.is_trial_period = 0 + AND t.status = 'COMPLETE' + AND t.pay_platform IN ? + AND (u.is_del = 0 OR u.is_del IS NULL) + AND UPPER(TRIM(u.origin_sys)) = ? + AND t.create_time >= ? + AND t.create_time < ? + GROUP BY t.user_id + UNION ALL + SELECT + t.user_id, + CAST(ROUND(SUM(IFNULL(t.compute_usd_amount, 0)) * 100, 0) AS SIGNED) AS amount_cents + FROM order_user_purchase_pay AS t + INNER JOIN user_base_info AS u ON u.id = t.user_id + WHERE t.evn = 'PROD' + AND t.factory_code = 'MIFA_PAY' + AND t.pay_status = 'SUCCESSFUL' + AND t.receipt_type = 'PAYMENT' + AND t.refund_status = 'NONE' + AND (u.is_del = 0 OR u.is_del IS NULL) + AND UPPER(TRIM(u.origin_sys)) = ? + AND IFNULL(t.update_time, t.create_time) >= ? + AND IFNULL(t.update_time, t.create_time) < ? + GROUP BY t.user_id + UNION ALL + SELECT + t.accept_user_id AS user_id, + CAST(ROUND(SUM(IFNULL(t.quantity, 0) * 100 / ?), 0) AS SIGNED) AS amount_cents + FROM likei_wallet.user_freight_balance_running_water AS t + INNER JOIN user_base_info AS u ON u.id = t.accept_user_id + WHERE t.origin = 'SHIPMENT' + AND t.type = 1 + AND t.accept_user_id IS NOT NULL + AND IFNULL(t.quantity, 0) > 0 + AND (u.is_del = 0 OR u.is_del IS NULL) + AND UPPER(TRIM(u.origin_sys)) = ? + AND t.create_time >= ? + AND t.create_time < ? + GROUP BY t.accept_user_id +) AS source +WHERE source.user_id > 0 +GROUP BY source.user_id +HAVING SUM(source.amount_cents) >= ? +ORDER BY SUM(source.amount_cents) DESC` + + var rows []rechargeRewardUserAmount + if err := s.repo.DB.WithContext(ctx).Raw( + query, + rechargeRewardOrderPlatforms, + sysOrigin, + startAt, + endAt, + sysOrigin, + startAt, + endAt, + coinSellerGoldPerUSD, + sysOrigin, + startAt, + endAt, + minThresholdCents, + ).Scan(&rows).Error; err != nil { + return nil, err + } + return rows, nil +} + +func (s *RechargeRewardSettlementService) ensureGrantRecord( + ctx context.Context, + configID int64, + sysOrigin string, + period rechargeRewardPeriod, + amount rechargeRewardUserAmount, + level rechargeRewardLevel, + profile rechargeRewardUserProfile, + now time.Time, +) (model.RechargeRewardGrantRecord, error) { + recordID, err := util.NextID() + if err != nil { + return model.RechargeRewardGrantRecord{}, err + } + rewardGroupStatus := rechargeRewardStatusSkipped + if level.RewardGroupID != nil && *level.RewardGroupID > 0 { + rewardGroupStatus = rechargeRewardStatusPending + } + rewardGoldStatus := rechargeRewardStatusSkipped + goldEventID := "" + if level.RewardGold > 0 { + rewardGoldStatus = rechargeRewardStatusPending + goldEventID = rechargeRewardGoldEventID(sysOrigin, period.CycleKey, amount.UserID, level.Level) + } + record := model.RechargeRewardGrantRecord{ + ID: recordID, + ConfigID: configID, + LevelID: level.ID, + SysOrigin: sysOrigin, + CycleKey: period.CycleKey, + RechargeDate: period.RechargeDate, + PeriodStartAt: period.StartAt, + PeriodEndAt: period.EndAt, + UserID: amount.UserID, + Account: profile.Account, + UserAvatar: profile.UserAvatar, + UserNickname: profile.UserNickname, + CountryCode: profile.CountryCode, + CountryName: profile.CountryName, + RechargeAmountCents: amount.RechargeAmountCents, + Level: level.Level, + RechargeThresholdCents: level.RechargeAmountCents, + RewardGold: level.RewardGold, + RewardGroupID: level.RewardGroupID, + RewardGroupName: level.RewardGroupName, + RewardGroupTrackID: recordID, + GoldEventID: goldEventID, + Status: rechargeRewardStatusPending, + RewardGroupStatus: rewardGroupStatus, + RewardGoldStatus: rewardGoldStatus, + CreateTime: now, + UpdateTime: now, + } + if err := s.repo.DB.WithContext(ctx). + Clauses(clause.OnConflict{DoNothing: true}). + Create(&record).Error; err != nil { + return model.RechargeRewardGrantRecord{}, err + } + + var saved model.RechargeRewardGrantRecord + if err := s.repo.DB.WithContext(ctx). + Where("sys_origin = ? AND cycle_key = ? AND user_id = ? AND level = ?", sysOrigin, period.CycleKey, amount.UserID, level.Level). + First(&saved).Error; err != nil { + return model.RechargeRewardGrantRecord{}, err + } + if saved.Status != rechargeRewardStatusSuccess && amount.RechargeAmountCents > saved.RechargeAmountCents { + if err := s.repo.DB.WithContext(ctx). + Model(&model.RechargeRewardGrantRecord{}). + Where("id = ?", saved.ID). + Updates(map[string]any{ + "recharge_amount_cents": amount.RechargeAmountCents, + "update_time": now, + }).Error; err != nil { + return model.RechargeRewardGrantRecord{}, err + } + saved.RechargeAmountCents = amount.RechargeAmountCents + } + return saved, nil +} + +func (s *RechargeRewardSettlementService) dispatchUnfinishedRecords(ctx context.Context, configID int64) error { + var rows []model.RechargeRewardGrantRecord + if err := s.repo.DB.WithContext(ctx). + Where("config_id = ? AND status IN ?", configID, []string{rechargeRewardStatusPending, rechargeRewardStatusFailed}). + Order("create_time asc"). + Limit(200). + Find(&rows).Error; err != nil { + return err + } + var lastErr error + for index := range rows { + if err := s.dispatchRewardRecord(ctx, &rows[index]); err != nil { + lastErr = err + } + } + return lastErr +} + +func (s *RechargeRewardSettlementService) dispatchRewardRecord(ctx context.Context, record *model.RechargeRewardGrantRecord) error { + if record == nil || record.Status == rechargeRewardStatusSuccess { + return nil + } + now := time.Now() + retryCount := record.RetryCount + if record.Status == rechargeRewardStatusFailed { + retryCount++ + } + + groupStatus := normalizeRechargeRewardPartStatus(record.RewardGroupStatus, record.RewardGroupID != nil && *record.RewardGroupID > 0) + if groupStatus != rechargeRewardStatusSuccess && groupStatus != rechargeRewardStatusSkipped { + if err := s.java.SendActivityReward(ctx, integration.SendActivityRewardRequest{ + TrackID: record.RewardGroupTrackID, + Origin: rechargeRewardOrigin, + SysOrigin: record.SysOrigin, + SourceGroupID: *record.RewardGroupID, + AcceptUserID: record.UserID, + }); err != nil { + return s.markRechargeRewardFailed(ctx, record.ID, retryCount, "reward_group_status", err, now) + } + groupStatus = rechargeRewardStatusSuccess + record.RewardGroupStatus = groupStatus + if err := s.repo.DB.WithContext(ctx). + Model(&model.RechargeRewardGrantRecord{}). + Where("id = ?", record.ID). + Updates(map[string]any{ + "reward_group_status": groupStatus, + "retry_count": retryCount, + "update_time": now, + }).Error; err != nil { + return err + } + } + + goldStatus := normalizeRechargeRewardPartStatus(record.RewardGoldStatus, record.RewardGold > 0) + if goldStatus != rechargeRewardStatusSuccess && goldStatus != rechargeRewardStatusSkipped { + if err := s.java.ChangeGoldBalance(ctx, integration.GoldReceiptCommand{ + ReceiptType: "INCOME", + UserID: record.UserID, + SysOrigin: record.SysOrigin, + EventID: record.GoldEventID, + Origin: rechargeRewardOrigin, + Remark: fmt.Sprintf("recharge reward %s level %d", record.CycleKey, record.Level), + Amount: integration.NewPennyAmountPayloadFromDollar(record.RewardGold), + CloseDelayAsset: false, + OpUserType: "BACK", + }); err != nil { + return s.markRechargeRewardFailed(ctx, record.ID, retryCount, "reward_gold_status", err, now) + } + goldStatus = rechargeRewardStatusSuccess + record.RewardGoldStatus = goldStatus + if err := s.repo.DB.WithContext(ctx). + Model(&model.RechargeRewardGrantRecord{}). + Where("id = ?", record.ID). + Updates(map[string]any{ + "reward_gold_status": goldStatus, + "retry_count": retryCount, + "update_time": now, + }).Error; err != nil { + return err + } + } + + if err := s.repo.DB.WithContext(ctx). + Model(&model.RechargeRewardGrantRecord{}). + Where("id = ?", record.ID). + Updates(map[string]any{ + "status": rechargeRewardStatusSuccess, + "reward_group_status": groupStatus, + "reward_gold_status": goldStatus, + "retry_count": retryCount, + "last_error": "", + "sent_at": now, + "update_time": now, + }).Error; err != nil { + return err + } + return nil +} + +func (s *RechargeRewardSettlementService) markRechargeRewardFailed(ctx context.Context, recordID int64, retryCount int, failedColumn string, err error, now time.Time) error { + updates := map[string]any{ + "status": rechargeRewardStatusFailed, + "retry_count": retryCount, + "last_error": truncateString(err.Error(), 1024), + "update_time": now, + } + if strings.TrimSpace(failedColumn) != "" { + updates[failedColumn] = rechargeRewardStatusFailed + } + if updateErr := s.repo.DB.WithContext(context.Background()). + Model(&model.RechargeRewardGrantRecord{}). + Where("id = ?", recordID). + Updates(updates).Error; updateErr != nil { + return updateErr + } + return err +} + +type rechargeRewardUserProfile struct { + UserID int64 + Account string + UserAvatar string + UserNickname string + CountryCode string + CountryName string +} + +func (s *RechargeRewardSettlementService) lookupUserProfiles(ctx context.Context, userIDs []int64) (map[int64]rechargeRewardUserProfile, error) { + result := make(map[int64]rechargeRewardUserProfile, len(userIDs)) + if len(userIDs) == 0 { + return result, nil + } + var rows []model.UserBaseInfo + if err := s.repo.DB.WithContext(ctx).Where("id IN ?", userIDs).Find(&rows).Error; err != nil { + return nil, err + } + for _, row := range rows { + result[row.ID] = rechargeRewardUserProfile{ + UserID: row.ID, + Account: strings.TrimSpace(row.Account), + UserAvatar: strings.TrimSpace(row.UserAvatar), + UserNickname: strings.TrimSpace(row.UserNickname), + CountryCode: strings.TrimSpace(row.CountryCode), + CountryName: strings.TrimSpace(row.CountryName), + } + } + for _, userID := range userIDs { + if _, ok := result[userID]; !ok { + result[userID] = rechargeRewardUserProfile{UserID: userID} + } + } + return result, nil +} + +func rechargeRewardCurrentPeriod(now time.Time, startAt time.Time, location *time.Location) (rechargeRewardPeriod, bool) { + local := now.In(location) + start := startAt.In(location) + if local.Before(start) { + return rechargeRewardPeriod{}, false + } + monthStart := time.Date(local.Year(), local.Month(), 1, 0, 0, 0, 0, location) + periodStart := monthStart + if periodStart.Before(start) { + periodStart = start + } + periodEnd := monthStart.AddDate(0, 1, 0) + if !local.Before(periodEnd) { + return rechargeRewardPeriod{}, false + } + return rechargeRewardPeriod{ + CycleKey: periodStart.Format("20060102"), + RechargeDate: periodStart.Year()*100 + int(periodStart.Month()), + StartAt: periodStart, + EndAt: periodEnd, + }, true +} + +func rechargeRewardStartAt(raw string, location *time.Location) time.Time { + value := strings.TrimSpace(raw) + if value == "" { + value = "2026-05-21" + } + parsed, err := time.ParseInLocation("2006-01-02", value, location) + if err != nil { + parsed, _ = time.ParseInLocation("2006-01-02", "2026-05-21", location) + } + return time.Date(parsed.Year(), parsed.Month(), parsed.Day(), 0, 0, 0, 0, location) +} + +func rechargeRewardLocation(timezone string, fallback *time.Location) *time.Location { + location, err := time.LoadLocation(strings.TrimSpace(timezone)) + if err == nil { + return location + } + if fallback != nil { + return fallback + } + return time.UTC +} + +func rechargeRewardStorageTime(t time.Time, storageLocation *time.Location) string { + if storageLocation == nil { + storageLocation = t.Location() + } + return t.In(storageLocation).Format("2006-01-02 15:04:05") +} + +func rechargeRewardAmountUserIDs(rows []rechargeRewardUserAmount) []int64 { + result := make([]int64, 0, len(rows)) + seen := make(map[int64]struct{}, len(rows)) + for _, row := range rows { + if row.UserID <= 0 { + continue + } + if _, exists := seen[row.UserID]; exists { + continue + } + seen[row.UserID] = struct{}{} + result = append(result, row.UserID) + } + return result +} + +func normalizeRechargeRewardSysOrigin(values ...string) string { + for _, value := range values { + normalized := strings.ToUpper(strings.TrimSpace(value)) + if normalized != "" { + return normalized + } + } + return "LIKEI" +} + +func normalizeRechargeRewardGroupID(id *int64) *int64 { + if id == nil || *id <= 0 { + return nil + } + return id +} + +func normalizeRechargeRewardPartStatus(status string, required bool) string { + normalized := strings.ToUpper(strings.TrimSpace(status)) + if !required { + return rechargeRewardStatusSkipped + } + switch normalized { + case rechargeRewardStatusSuccess, rechargeRewardStatusFailed, rechargeRewardStatusPending: + return normalized + default: + return rechargeRewardStatusPending + } +} + +func rechargeRewardGoldEventID(sysOrigin string, cycleKey string, userID int64, level int) string { + return "recharge-reward:" + strings.ToUpper(strings.TrimSpace(sysOrigin)) + ":" + cycleKey + ":" + strconv.FormatInt(userID, 10) + ":" + strconv.Itoa(level) +} diff --git a/internal/service/recharge_reward_settlement_test.go b/internal/service/recharge_reward_settlement_test.go new file mode 100644 index 0000000..aec9a64 --- /dev/null +++ b/internal/service/recharge_reward_settlement_test.go @@ -0,0 +1,54 @@ +package service + +import ( + "testing" + "time" +) + +func TestRechargeRewardCurrentPeriodStartsOnMay21ThenCalendarMonth(t *testing.T) { + location, err := time.LoadLocation("Asia/Riyadh") + if err != nil { + t.Fatalf("load location: %v", err) + } + startAt := rechargeRewardStartAt("2026-05-21", location) + + beforeStart := time.Date(2026, 5, 20, 23, 59, 59, 0, location) + if _, ok := rechargeRewardCurrentPeriod(beforeStart, startAt, location); ok { + t.Fatalf("period before start should be inactive") + } + + firstNow := time.Date(2026, 5, 21, 0, 0, 0, 0, location) + first, ok := rechargeRewardCurrentPeriod(firstNow, startAt, location) + if !ok { + t.Fatalf("first period should be active") + } + if first.CycleKey != "20260521" || first.RechargeDate != 202605 { + t.Fatalf("first period key/date = %s/%d", first.CycleKey, first.RechargeDate) + } + if !first.StartAt.Equal(startAt) { + t.Fatalf("first start = %s, want %s", first.StartAt, startAt) + } + if want := time.Date(2026, 6, 1, 0, 0, 0, 0, location); !first.EndAt.Equal(want) { + t.Fatalf("first end = %s, want %s", first.EndAt, want) + } + + secondNow := time.Date(2026, 6, 1, 0, 0, 0, 0, location) + second, ok := rechargeRewardCurrentPeriod(secondNow, startAt, location) + if !ok { + t.Fatalf("second period should be active") + } + if second.CycleKey != "20260601" || second.RechargeDate != 202606 { + t.Fatalf("second period key/date = %s/%d", second.CycleKey, second.RechargeDate) + } + if want := time.Date(2026, 6, 1, 0, 0, 0, 0, location); !second.StartAt.Equal(want) { + t.Fatalf("second start = %s, want %s", second.StartAt, want) + } +} + +func TestRechargeRewardGoldEventIDUsesCycleAndLevel(t *testing.T) { + got := rechargeRewardGoldEventID("likei", "20260601", 42, 3) + want := "recharge-reward:LIKEI:20260601:42:3" + if got != want { + t.Fatalf("event id = %q, want %q", got, want) + } +}