hy/yumi-cron
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

充值奖励结算

Browse Source
This commit is contained in:
zhx 2026-05-21 11:15:04 +08:00
parent 4777c6dabb
commit 9b760b562c
8 changed files with 870 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
cmd/week-star-job/main.go
View File

@ -29,15 +29,18 @@ func main() {
javaClient := integration.New(cfg) javaClient := integration.New(cfg)
settlementService := service.NewWeekStarSettlementService(cfg, repository, javaClient) settlementService := service.NewWeekStarSettlementService(cfg, repository, javaClient)
roomTurnoverRewardService := service.NewRoomTurnoverRewardSettlementService(cfg, repository, javaClient) roomTurnoverRewardService := service.NewRoomTurnoverRewardSettlementService(cfg, repository, javaClient)
rechargeRewardService := service.NewRechargeRewardSettlementService(cfg, repository, javaClient)
workerCtx, workerCancel := context.WithCancel(context.Background()) workerCtx, workerCancel := context.WithCancel(context.Background())
defer workerCancel() defer workerCancel()
settlementService.Start(workerCtx) settlementService.Start(workerCtx)
roomTurnoverRewardService.Start(workerCtx) roomTurnoverRewardService.Start(workerCtx)
rechargeRewardService.Start(workerCtx)
log.Printf( 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.WeekStarSettleInterval,
cfg.RoomTurnoverSettleInterval, cfg.RoomTurnoverSettleInterval,
cfg.RechargeRewardSettleInterval,
cfg.WeekStarTimezone, cfg.WeekStarTimezone,
cfg.JavaOtherBaseURL, cfg.JavaOtherBaseURL,
cfg.JavaWalletBaseURL, cfg.JavaWalletBaseURL,

6
cmd/week-star-settle-once/main.go
View File

@ -27,12 +27,16 @@ func main() {
javaClient := integration.New(cfg) javaClient := integration.New(cfg)
settlementService := service.NewWeekStarSettlementService(cfg, repository, javaClient) settlementService := service.NewWeekStarSettlementService(cfg, repository, javaClient)
roomTurnoverRewardService := service.NewRoomTurnoverRewardSettlementService(cfg, repository, javaClient) roomTurnoverRewardService := service.NewRoomTurnoverRewardSettlementService(cfg, repository, javaClient)
rechargeRewardService := service.NewRechargeRewardSettlementService(cfg, repository, javaClient)
if err := settlementService.SettleDueCycles(ctx); err != nil { if err := settlementService.SettleDueCycles(ctx); err != nil {
log.Fatalf("settle due cycles failed: %v", err) log.Fatalf("settle due cycles failed: %v", err)
} }
if err := roomTurnoverRewardService.SettleDueCycles(ctx); err != nil { if err := roomTurnoverRewardService.SettleDueCycles(ctx); err != nil {
log.Fatalf("settle room turnover reward due cycles failed: %v", err) 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")
} }

72
internal/config/config.go
View File

@ -8,27 +8,33 @@ import (
) )
type Config struct { type Config struct {
MySQLDSN string MySQLDSN string
RedisAddr string RedisAddr string
RedisPassword string RedisPassword string
RedisDB int RedisDB int
JavaOtherBaseURL string JavaOtherBaseURL string
JavaWalletBaseURL string JavaWalletBaseURL string
WeekStarDefaultSysOrigin string WeekStarDefaultSysOrigin string
WeekStarTimezone string WeekStarTimezone string
WeekStarSettleInterval time.Duration WeekStarSettleInterval time.Duration
RoomTurnoverDefaultSysOrigin string RoomTurnoverDefaultSysOrigin string
RoomTurnoverTimezone string RoomTurnoverTimezone string
RoomTurnoverSettleInterval time.Duration RoomTurnoverSettleInterval time.Duration
WeekStarRetryStreamKey string RechargeRewardDefaultSysOrigin string
WeekStarRetryConsumerGroup string RechargeRewardTimezone string
WeekStarRetryConsumerName string RechargeRewardStorageTimezone string
WeekStarRetryBatchSize int64 RechargeRewardStartDate string
WeekStarRetryBlock time.Duration RechargeRewardSettleInterval time.Duration
WeekStarRetryMinIdle time.Duration RechargeRewardCoinSellerGoldPerUSD int64
Timeout time.Duration WeekStarRetryStreamKey string
ShutdownTimeout time.Duration WeekStarRetryConsumerGroup string
AutoMigrate bool WeekStarRetryConsumerName string
WeekStarRetryBatchSize int64
WeekStarRetryBlock time.Duration
WeekStarRetryMinIdle time.Duration
Timeout time.Duration
ShutdownTimeout time.Duration
AutoMigrate bool
} }
func Load() Config { func Load() Config {
@ -74,6 +80,30 @@ func Load() Config {
[]string{"CHATAPP_ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS", "ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS"}, []string{"CHATAPP_ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS", "ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS"},
60, 60,
)) * time.Second, )) * 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"), 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"), 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")), WeekStarRetryConsumerName: getEnvAny([]string{"CHATAPP_WEEK_STAR_RETRY_CONSUMER_NAME", "WEEK_STAR_RETRY_CONSUMER_NAME"}, defaultConsumerName("week-star-retry")),

24
internal/config/config_test.go
View File

@ -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_DEFAULT_SYS_ORIGIN", "likei")
t.Setenv("CHATAPP_ROOM_TURNOVER_REWARD_TIMEZONE", "Asia/Riyadh") t.Setenv("CHATAPP_ROOM_TURNOVER_REWARD_TIMEZONE", "Asia/Riyadh")
t.Setenv("CHATAPP_ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS", "180") 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_STREAM_KEY", "chatapp-stream")
t.Setenv("CHATAPP_WEEK_STAR_RETRY_CONSUMER_GROUP", "chatapp-group") t.Setenv("CHATAPP_WEEK_STAR_RETRY_CONSUMER_GROUP", "chatapp-group")
t.Setenv("CHATAPP_WEEK_STAR_RETRY_CONSUMER_NAME", "chatapp-consumer") 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 { if cfg.RoomTurnoverDefaultSysOrigin != "LIKEI" || cfg.RoomTurnoverTimezone != "Asia/Riyadh" || cfg.RoomTurnoverSettleInterval != 180*time.Second {
t.Fatalf("unexpected room turnover config: %+v", cfg) 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" { if cfg.WeekStarRetryStreamKey != "chatapp-stream" || cfg.WeekStarRetryConsumerGroup != "chatapp-group" || cfg.WeekStarRetryConsumerName != "chatapp-consumer" {
t.Fatalf("unexpected retry consumer config: %+v", cfg) 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_DEFAULT_SYS_ORIGIN", "tarab")
t.Setenv("ROOM_TURNOVER_REWARD_TIMEZONE", "Asia/Shanghai") t.Setenv("ROOM_TURNOVER_REWARD_TIMEZONE", "Asia/Shanghai")
t.Setenv("ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS", "77") 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_STREAM_KEY", "legacy-stream")
t.Setenv("WEEK_STAR_RETRY_CONSUMER_GROUP", "legacy-group") t.Setenv("WEEK_STAR_RETRY_CONSUMER_GROUP", "legacy-group")
t.Setenv("WEEK_STAR_RETRY_CONSUMER_NAME", "legacy-consumer") 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 { if cfg.RoomTurnoverDefaultSysOrigin != "TARAB" || cfg.RoomTurnoverTimezone != "Asia/Shanghai" || cfg.RoomTurnoverSettleInterval != 77*time.Second {
t.Fatalf("unexpected legacy room turnover config: %+v", cfg) 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" { if cfg.WeekStarRetryStreamKey != "legacy-stream" || cfg.WeekStarRetryConsumerGroup != "legacy-group" || cfg.WeekStarRetryConsumerName != "legacy-consumer" {
t.Fatalf("unexpected retry consumer config: %+v", cfg) t.Fatalf("unexpected retry consumer config: %+v", cfg)
} }

6
internal/integration/java.go
View File

@ -138,7 +138,11 @@ func New(cfg config.Config) *Client {
} }
func (c *Client) SendActivityReward(ctx context.Context, req SendActivityRewardRequest) error { 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) { func (c *Client) ListLastWeekRoomContribution(ctx context.Context) ([]RoomContributionActivityCount, error) {

64
internal/model/models.go
View File

@ -152,3 +152,67 @@ type RoomTurnoverRewardRecord struct {
} }
func (RoomTurnoverRewardRecord) TableName() string { return "room_turnover_reward_record" } 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" }

663
internal/service/recharge_reward_settlement.go Normal file
View File

@ -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)
}

54
internal/service/recharge_reward_settlement_test.go Normal file
View File

@ -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)
}
}