From 49d788be18935ba95351e038ee4558b29e92d892 Mon Sep 17 00:00:00 2001 From: z Date: Tue, 5 May 2026 01:42:05 +0800 Subject: [PATCH] Add room turnover reward settlement --- README.md | 5 + cmd/week-star-job/main.go | 6 +- cmd/week-star-settle-once/main.go | 6 +- internal/config/config.go | 58 ++- internal/config/config_test.go | 20 + internal/integration/java.go | 153 +++++++ internal/model/models.go | 56 +++ .../room_turnover_reward_settlement.go | 423 ++++++++++++++++++ 8 files changed, 706 insertions(+), 21 deletions(-) create mode 100644 internal/service/room_turnover_reward_settlement.go diff --git a/README.md b/README.md index e1039fe..c361016 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,7 @@ Standalone cron workers for chatapp activities. Current worker: - `week-star-job`: specified-gift weekly rank settlement, reward dispatch, and failed reward retry +- `week-star-job`: room turnover weekly reward settlement and backfill dispatch Local run: @@ -27,9 +28,13 @@ Required envs: - `CHATAPP_STORE_REDIS_PASSWORD` - `CHATAPP_STORE_REDIS_DB` - `CHATAPP_JAVA_OTHER_BASE_URL` +- `CHATAPP_JAVA_WALLET_BASE_URL` - `CHATAPP_WEEK_STAR_DEFAULT_SYS_ORIGIN` - `CHATAPP_WEEK_STAR_TIMEZONE` - `CHATAPP_WEEK_STAR_SETTLE_INTERVAL_SECONDS` +- `CHATAPP_ROOM_TURNOVER_REWARD_DEFAULT_SYS_ORIGIN` +- `CHATAPP_ROOM_TURNOVER_REWARD_TIMEZONE` +- `CHATAPP_ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS` - `CHATAPP_WEEK_STAR_RETRY_STREAM_KEY` - `CHATAPP_WEEK_STAR_RETRY_CONSUMER_GROUP` - `CHATAPP_WEEK_STAR_RETRY_CONSUMER_NAME` diff --git a/cmd/week-star-job/main.go b/cmd/week-star-job/main.go index c033a26..35d635e 100644 --- a/cmd/week-star-job/main.go +++ b/cmd/week-star-job/main.go @@ -28,15 +28,19 @@ func main() { javaClient := integration.New(cfg) settlementService := service.NewWeekStarSettlementService(cfg, repository, javaClient) + roomTurnoverRewardService := service.NewRoomTurnoverRewardSettlementService(cfg, repository, javaClient) workerCtx, workerCancel := context.WithCancel(context.Background()) defer workerCancel() settlementService.Start(workerCtx) + roomTurnoverRewardService.Start(workerCtx) log.Printf( - "week-star-job started. interval=%s timezone=%s javaOther=%s", + "week-star-job started. weekStarInterval=%s roomTurnoverInterval=%s timezone=%s javaOther=%s javaWallet=%s", cfg.WeekStarSettleInterval, + cfg.RoomTurnoverSettleInterval, cfg.WeekStarTimezone, cfg.JavaOtherBaseURL, + cfg.JavaWalletBaseURL, ) shutdownSignal, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) diff --git a/cmd/week-star-settle-once/main.go b/cmd/week-star-settle-once/main.go index b4d17e9..db71623 100644 --- a/cmd/week-star-settle-once/main.go +++ b/cmd/week-star-settle-once/main.go @@ -26,9 +26,13 @@ func main() { javaClient := integration.New(cfg) settlementService := service.NewWeekStarSettlementService(cfg, repository, javaClient) + roomTurnoverRewardService := service.NewRoomTurnoverRewardSettlementService(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) + } - log.Printf("week-star settle once finished") + log.Printf("week-star and room-turnover-reward settle once finished") } diff --git a/internal/config/config.go b/internal/config/config.go index 652806f..6d114c6 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -8,23 +8,27 @@ import ( ) type Config struct { - MySQLDSN string - RedisAddr string - RedisPassword string - RedisDB int - JavaOtherBaseURL string - WeekStarDefaultSysOrigin string - WeekStarTimezone string - WeekStarSettleInterval 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 + 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 { @@ -49,11 +53,27 @@ func Load() Config { []string{"CHATAPP_JAVA_OTHER_BASE_URL", "CRON_JAVA_OTHER_BASE_URL", "GAME_JAVA_OTHER_BASE_URL", "INVITE_JAVA_APP_BASE_URL"}, "http://127.0.0.1:2400", ), + JavaWalletBaseURL: getEnvAny( + []string{"CHATAPP_JAVA_WALLET_BASE_URL", "CRON_JAVA_WALLET_BASE_URL", "GAME_JAVA_WALLET_BASE_URL"}, + "http://127.0.0.1:2000", + ), WeekStarDefaultSysOrigin: strings.ToUpper( getEnvAny([]string{"CHATAPP_WEEK_STAR_DEFAULT_SYS_ORIGIN", "WEEK_STAR_DEFAULT_SYS_ORIGIN"}, "LIKEI"), ), - WeekStarTimezone: getEnvAny([]string{"CHATAPP_WEEK_STAR_TIMEZONE", "WEEK_STAR_TIMEZONE"}, "Asia/Riyadh"), - WeekStarSettleInterval: time.Duration(getEnvIntAny([]string{"CHATAPP_WEEK_STAR_SETTLE_INTERVAL_SECONDS", "WEEK_STAR_SETTLE_INTERVAL_SECONDS"}, 60)) * time.Second, + WeekStarTimezone: getEnvAny([]string{"CHATAPP_WEEK_STAR_TIMEZONE", "WEEK_STAR_TIMEZONE"}, "Asia/Riyadh"), + WeekStarSettleInterval: time.Duration(getEnvIntAny([]string{"CHATAPP_WEEK_STAR_SETTLE_INTERVAL_SECONDS", "WEEK_STAR_SETTLE_INTERVAL_SECONDS"}, 60)) * time.Second, + RoomTurnoverDefaultSysOrigin: strings.ToUpper(getEnvAny( + []string{"CHATAPP_ROOM_TURNOVER_REWARD_DEFAULT_SYS_ORIGIN", "ROOM_TURNOVER_REWARD_DEFAULT_SYS_ORIGIN", "CHATAPP_WEEK_STAR_DEFAULT_SYS_ORIGIN", "WEEK_STAR_DEFAULT_SYS_ORIGIN"}, + "LIKEI", + )), + RoomTurnoverTimezone: getEnvAny( + []string{"CHATAPP_ROOM_TURNOVER_REWARD_TIMEZONE", "ROOM_TURNOVER_REWARD_TIMEZONE", "CHATAPP_WEEK_STAR_TIMEZONE", "WEEK_STAR_TIMEZONE"}, + "Asia/Riyadh", + ), + RoomTurnoverSettleInterval: time.Duration(getEnvIntAny( + []string{"CHATAPP_ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS", "ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS"}, + 60, + )) * time.Second, 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 b053591..af08db3 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -12,9 +12,13 @@ func TestLoadPrefersChatappKeys(t *testing.T) { t.Setenv("CHATAPP_STORE_REDIS_PASSWORD", "chatapp-password") t.Setenv("CHATAPP_STORE_REDIS_DB", "5") t.Setenv("CHATAPP_JAVA_OTHER_BASE_URL", "http://chatapp-java") + t.Setenv("CHATAPP_JAVA_WALLET_BASE_URL", "http://chatapp-wallet") t.Setenv("CHATAPP_WEEK_STAR_DEFAULT_SYS_ORIGIN", "likei") t.Setenv("CHATAPP_WEEK_STAR_TIMEZONE", "Asia/Shanghai") t.Setenv("CHATAPP_WEEK_STAR_SETTLE_INTERVAL_SECONDS", "120") + 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_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") @@ -36,6 +40,9 @@ func TestLoadPrefersChatappKeys(t *testing.T) { if cfg.JavaOtherBaseURL != "http://chatapp-java" { t.Fatalf("expected chatapp java base url, got %q", cfg.JavaOtherBaseURL) } + if cfg.JavaWalletBaseURL != "http://chatapp-wallet" { + t.Fatalf("expected chatapp java wallet base url, got %q", cfg.JavaWalletBaseURL) + } if cfg.WeekStarDefaultSysOrigin != "LIKEI" { t.Fatalf("expected normalized sys origin LIKEI, got %q", cfg.WeekStarDefaultSysOrigin) } @@ -45,6 +52,9 @@ func TestLoadPrefersChatappKeys(t *testing.T) { if cfg.WeekStarSettleInterval != 120*time.Second { t.Fatalf("expected settle interval 120s, got %s", cfg.WeekStarSettleInterval) } + if cfg.RoomTurnoverDefaultSysOrigin != "LIKEI" || cfg.RoomTurnoverTimezone != "Asia/Riyadh" || cfg.RoomTurnoverSettleInterval != 180*time.Second { + t.Fatalf("unexpected room turnover config: %+v", cfg) + } if cfg.WeekStarRetryStreamKey != "chatapp-stream" || cfg.WeekStarRetryConsumerGroup != "chatapp-group" || cfg.WeekStarRetryConsumerName != "chatapp-consumer" { t.Fatalf("unexpected retry consumer config: %+v", cfg) } @@ -65,9 +75,13 @@ func TestLoadFallsBackToLegacyKeys(t *testing.T) { t.Setenv("CRON_REDIS_PASSWORD", "legacy-password") t.Setenv("CRON_REDIS_DB", "3") t.Setenv("CRON_JAVA_OTHER_BASE_URL", "http://legacy-java") + t.Setenv("CRON_JAVA_WALLET_BASE_URL", "http://legacy-wallet") t.Setenv("WEEK_STAR_DEFAULT_SYS_ORIGIN", "tarab") t.Setenv("WEEK_STAR_TIMEZONE", "Asia/Riyadh") t.Setenv("WEEK_STAR_SETTLE_INTERVAL_SECONDS", "66") + 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("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") @@ -89,12 +103,18 @@ func TestLoadFallsBackToLegacyKeys(t *testing.T) { if cfg.JavaOtherBaseURL != "http://legacy-java" { t.Fatalf("expected legacy java base url, got %q", cfg.JavaOtherBaseURL) } + if cfg.JavaWalletBaseURL != "http://legacy-wallet" { + t.Fatalf("expected legacy java wallet base url, got %q", cfg.JavaWalletBaseURL) + } if cfg.WeekStarDefaultSysOrigin != "TARAB" { t.Fatalf("expected normalized sys origin TARAB, got %q", cfg.WeekStarDefaultSysOrigin) } if cfg.WeekStarSettleInterval != 66*time.Second { t.Fatalf("expected settle interval 66s, got %s", cfg.WeekStarSettleInterval) } + if cfg.RoomTurnoverDefaultSysOrigin != "TARAB" || cfg.RoomTurnoverTimezone != "Asia/Shanghai" || cfg.RoomTurnoverSettleInterval != 77*time.Second { + t.Fatalf("unexpected legacy room turnover 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 25bd516..9efcaac 100644 --- a/internal/integration/java.go +++ b/internal/integration/java.go @@ -32,6 +32,45 @@ type UserProfile struct { UserNickname string `json:"userNickname"` } +type RoomContributionActivityCount struct { + ID string `json:"id"` + RoomID Int64Value `json:"roomId"` + ContributionValue AmountValue `json:"contributionValue"` + DateNumber int `json:"dateNumber"` + Received bool `json:"received"` + RewardCoinsSent bool `json:"rewardCoinsSent"` +} + +type RoomProfile struct { + ID Int64Value `json:"id"` + RoomAccount string `json:"roomAccount"` + UserID Int64Value `json:"userId"` + RoomCover string `json:"roomCover"` + RoomName string `json:"roomName"` + SysOrigin string `json:"sysOrigin"` + CountryCode string `json:"countryCode"` + CountryName string `json:"countryName"` +} + +type GoldReceiptCommand struct { + ReceiptType string `json:"receiptType"` + UserID int64 `json:"userId"` + SysOrigin string `json:"sysOrigin"` + EventID string `json:"eventId"` + Origin string `json:"origin,omitempty"` + Remark string `json:"remark,omitempty"` + Amount PennyAmountPayload `json:"amount"` + CloseDelayAsset bool `json:"closeDelayAsset"` + OpUserType string `json:"opUserType"` + CustomizeOrigin string `json:"customizeOrigin,omitempty"` + CustomizeOriginDesc string `json:"customizeOriginDesc,omitempty"` +} + +type PennyAmountPayload struct { + PennyAmount int64 `json:"pennyAmount,string"` + DollarAmount int64 `json:"dollarAmount"` +} + type resultResponse[T any] struct { Code int `json:"code"` Message string `json:"message"` @@ -58,6 +97,37 @@ func (v *Int64Value) UnmarshalJSON(data []byte) error { return nil } +type AmountValue int64 + +func (v *AmountValue) UnmarshalJSON(data []byte) error { + raw := strings.TrimSpace(string(data)) + if raw == "" || raw == "null" { + *v = 0 + return nil + } + if strings.HasPrefix(raw, "\"") && strings.HasSuffix(raw, "\"") { + raw = strings.Trim(raw, "\"") + } + if raw == "" { + *v = 0 + return nil + } + floatRaw := raw + if dot := strings.Index(raw, "."); dot >= 0 { + raw = raw[:dot] + } + if parsed, err := strconv.ParseInt(raw, 10, 64); err == nil { + *v = AmountValue(parsed) + return nil + } + parsed, err := strconv.ParseFloat(floatRaw, 64) + if err != nil { + return err + } + *v = AmountValue(int64(parsed)) + return nil +} + func New(cfg config.Config) *Client { return &Client{ cfg: cfg, @@ -71,6 +141,66 @@ func (c *Client) SendActivityReward(ctx context.Context, req SendActivityRewardR return c.postJSON(ctx, c.cfg.JavaOtherBaseURL+"/props-activity-cnf/client/sendActivityReward", req, nil) } +func (c *Client) ListLastWeekRoomContribution(ctx context.Context) ([]RoomContributionActivityCount, error) { + endpoint := c.cfg.JavaOtherBaseURL + "/room-contribution-activity-count/client/listLastWeek" + var resp resultResponse[[]RoomContributionActivityCount] + if err := c.getJSON(ctx, endpoint, &resp); err != nil { + return nil, err + } + return resp.Body, nil +} + +func (c *Client) MarkRoomTurnoverRewardCoinsSent(ctx context.Context, id string) (bool, error) { + id = strings.TrimSpace(id) + if id == "" { + return false, fmt.Errorf("empty room contribution id") + } + endpoint := c.cfg.JavaOtherBaseURL + "/room-contribution-activity-count/client/rewardCoinsSent?id=" + url.QueryEscape(id) + var resp resultResponse[bool] + if err := c.postEmpty(ctx, endpoint, &resp); err != nil { + return false, err + } + return resp.Body, nil +} + +func (c *Client) MapRoomProfiles(ctx context.Context, roomIDs []int64) (map[int64]RoomProfile, error) { + if len(roomIDs) == 0 { + return map[int64]RoomProfile{}, nil + } + endpoint := c.cfg.JavaOtherBaseURL + "/room-manager/client/mapProfileByRoomIds" + var resp resultResponse[map[string]RoomProfile] + if err := c.postJSON(ctx, endpoint, roomIDs, &resp); err != nil { + return nil, err + } + result := make(map[int64]RoomProfile, len(resp.Body)) + for key, value := range resp.Body { + roomID, err := strconv.ParseInt(strings.TrimSpace(key), 10, 64) + if err != nil { + roomID = int64(value.ID) + } + if roomID <= 0 { + continue + } + result[roomID] = value + } + return result, nil +} + +func (c *Client) ChangeGoldBalance(ctx context.Context, cmd GoldReceiptCommand) error { + var resp resultResponse[json.RawMessage] + if err := c.postJSON(ctx, c.cfg.JavaWalletBaseURL+"/wallet/gold/client/balance/change", cmd, &resp); err != nil { + return err + } + return ensureSuccessfulResult(resp.Code, resp.Message, resp.Success) +} + +func NewPennyAmountPayloadFromDollar(amount int64) PennyAmountPayload { + return PennyAmountPayload{ + PennyAmount: amount * 100, + DollarAmount: amount, + } +} + func (c *Client) GetUserProfile(ctx context.Context, userID int64) (UserProfile, error) { endpoint := c.cfg.JavaOtherBaseURL + "/user-profile/client/getByUserId?userId=" + url.QueryEscape(strconv.FormatInt(userID, 10)) var resp resultResponse[UserProfile] @@ -114,6 +244,19 @@ func (c *Client) postJSON(ctx context.Context, endpoint string, payload any, out return decodeResponse(resp, out) } +func (c *Client) postEmpty(ctx context.Context, endpoint string, out any) error { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, nil) + if err != nil { + return err + } + resp, err := c.httpClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + return decodeResponse(resp, out) +} + func decodeResponse(resp *http.Response, out any) error { body, err := io.ReadAll(resp.Body) if err != nil { @@ -130,3 +273,13 @@ func decodeResponse(resp *http.Response, out any) error { } return nil } + +func ensureSuccessfulResult(code int, message string, success *bool) error { + if success != nil && !*success { + return fmt.Errorf("java api returned unsuccessful result: code=%d message=%s", code, strings.TrimSpace(message)) + } + if code != 0 && code != 200 { + return fmt.Errorf("java api returned error result: code=%d message=%s", code, strings.TrimSpace(message)) + } + return nil +} diff --git a/internal/model/models.go b/internal/model/models.go index 1c68d11..90a62f9 100644 --- a/internal/model/models.go +++ b/internal/model/models.go @@ -94,3 +94,59 @@ type WeekStarRewardRecord struct { // TableName 返回周榜发奖记录表名。 func (WeekStarRewardRecord) TableName() string { return "week_star_reward_record" } + +// RoomTurnoverRewardConfig 保存房间流水奖励的系统级配置。 +type RoomTurnoverRewardConfig struct { + ID int64 `gorm:"column:id;primaryKey"` + SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_room_turnover_reward_sys_origin"` + Enabled bool `gorm:"column:enabled;index:idx_room_turnover_reward_enabled"` + Timezone string `gorm:"column:timezone;size:64"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` +} + +func (RoomTurnoverRewardConfig) TableName() string { return "room_turnover_reward_config" } + +// RoomTurnoverRewardLevel 保存达到某个流水门槛后发放的金币奖励。 +type RoomTurnoverRewardLevel struct { + ID int64 `gorm:"column:id;primaryKey"` + ConfigID int64 `gorm:"column:config_id;uniqueIndex:uk_room_turnover_reward_level,priority:1;uniqueIndex:uk_room_turnover_reward_threshold,priority:1;index:idx_room_turnover_reward_level_config"` + Level int `gorm:"column:level;uniqueIndex:uk_room_turnover_reward_level,priority:2"` + TurnoverThreshold int64 `gorm:"column:turnover_threshold;uniqueIndex:uk_room_turnover_reward_threshold,priority:2"` + RewardGold int64 `gorm:"column:reward_gold"` + Enabled bool `gorm:"column:enabled"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` +} + +func (RoomTurnoverRewardLevel) TableName() string { return "room_turnover_reward_level" } + +// RoomTurnoverRewardRecord 记录每周给房主发放的最高档房间流水奖励。 +type RoomTurnoverRewardRecord struct { + ID int64 `gorm:"column:id;primaryKey"` + ConfigID int64 `gorm:"column:config_id;index:idx_room_turnover_reward_record_config"` + SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_room_turnover_reward_record_room,priority:1;index:idx_room_turnover_reward_record_cycle,priority:1"` + CycleKey string `gorm:"column:cycle_key;size:32;uniqueIndex:uk_room_turnover_reward_record_room,priority:2;index:idx_room_turnover_reward_record_cycle,priority:2"` + PeriodStartAt time.Time `gorm:"column:period_start_at"` + PeriodEndAt time.Time `gorm:"column:period_end_at"` + RoomID int64 `gorm:"column:room_id;uniqueIndex:uk_room_turnover_reward_record_room,priority:3;index:idx_room_turnover_reward_record_room"` + RoomAccount string `gorm:"column:room_account;size:64"` + RoomName string `gorm:"column:room_name;size:255"` + RoomCover string `gorm:"column:room_cover;size:1024"` + OwnerUserID int64 `gorm:"column:owner_user_id;index:idx_room_turnover_reward_record_owner"` + CountryCode string `gorm:"column:country_code;size:32"` + CountryName string `gorm:"column:country_name;size:128"` + TurnoverAmount int64 `gorm:"column:turnover_amount"` + Level int `gorm:"column:level"` + TurnoverThreshold int64 `gorm:"column:turnover_threshold"` + RewardGold int64 `gorm:"column:reward_gold"` + TrackID string `gorm:"column:track_id;size:128;uniqueIndex:uk_room_turnover_reward_track"` + Status string `gorm:"column:status;size:32;index:idx_room_turnover_reward_record_status"` + 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 (RoomTurnoverRewardRecord) TableName() string { return "room_turnover_reward_record" } diff --git a/internal/service/room_turnover_reward_settlement.go b/internal/service/room_turnover_reward_settlement.go new file mode 100644 index 0000000..74853b2 --- /dev/null +++ b/internal/service/room_turnover_reward_settlement.go @@ -0,0 +1,423 @@ +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 ( + roomTurnoverRewardStatusPending = "PENDING" + roomTurnoverRewardStatusSuccess = "SUCCESS" + roomTurnoverRewardStatusFailed = "FAILED" + roomTurnoverRewardGoldOrigin = "ROOM_CONTRIBUTION_REWARD" +) + +type roomTurnoverRewardLevel struct { + Level int + TurnoverThreshold int64 + RewardGold int64 +} + +// RoomTurnoverRewardSettlementService 负责房间流水奖励的周结算和补发。 +type RoomTurnoverRewardSettlementService struct { + cfg config.Config + repo *repo.Repository + java *integration.Client + location *time.Location +} + +func NewRoomTurnoverRewardSettlementService(cfg config.Config, repository *repo.Repository, javaClient *integration.Client) *RoomTurnoverRewardSettlementService { + location, err := time.LoadLocation(strings.TrimSpace(cfg.RoomTurnoverTimezone)) + if err != nil { + location = time.UTC + } + return &RoomTurnoverRewardSettlementService{ + cfg: cfg, + repo: repository, + java: javaClient, + location: location, + } +} + +func (s *RoomTurnoverRewardSettlementService) Start(ctx context.Context) { + interval := s.cfg.RoomTurnoverSettleInterval + 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("room turnover reward settlement scan failed: %v", err) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } + }() +} + +func (s *RoomTurnoverRewardSettlementService) SettleDueCycles(ctx context.Context) error { + var configs []model.RoomTurnoverRewardConfig + 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 + } + + contributions, err := s.java.ListLastWeekRoomContribution(ctx) + if err != nil { + return err + } + if len(contributions) == 0 { + return nil + } + + roomProfiles, err := s.java.MapRoomProfiles(ctx, uniqueRoomTurnoverRewardRoomIDs(contributions)) + if err != nil { + return err + } + + var lastErr error + for index := range configs { + if err := s.settleConfig(ctx, &configs[index], contributions, roomProfiles); err != nil { + lastErr = err + log.Printf("room turnover reward settle config failed. configId=%d err=%v", configs[index].ID, err) + } + } + return lastErr +} + +func (s *RoomTurnoverRewardSettlementService) settleConfig( + ctx context.Context, + configRow *model.RoomTurnoverRewardConfig, + contributions []integration.RoomContributionActivityCount, + roomProfiles map[int64]integration.RoomProfile, +) error { + if configRow == nil { + return nil + } + sysOrigin := normalizeRoomTurnoverRewardSysOrigin(configRow.SysOrigin, s.cfg.RoomTurnoverDefaultSysOrigin) + levels, err := s.loadLevels(ctx, configRow.ID) + if err != nil { + return err + } + if len(levels) == 0 { + return nil + } + + byCycle := make(map[string][]integration.RoomContributionActivityCount) + for _, contribution := range contributions { + roomID := int64(contribution.RoomID) + if roomID <= 0 { + continue + } + profile, ok := roomProfiles[roomID] + if !ok { + continue + } + profileSysOrigin := strings.ToUpper(strings.TrimSpace(profile.SysOrigin)) + if profileSysOrigin != "" && profileSysOrigin != sysOrigin { + continue + } + cycleKey := roomTurnoverRewardCycleKey(contribution.DateNumber, time.Now().In(s.location)) + if cycleKey == "" { + continue + } + byCycle[cycleKey] = append(byCycle[cycleKey], contribution) + } + + var lastErr error + for cycleKey, cycleContributions := range byCycle { + if err := s.settleCycle(ctx, configRow, sysOrigin, cycleKey, levels, cycleContributions, roomProfiles); err != nil { + lastErr = err + } + } + return lastErr +} + +func (s *RoomTurnoverRewardSettlementService) settleCycle( + ctx context.Context, + configRow *model.RoomTurnoverRewardConfig, + sysOrigin string, + cycleKey string, + levels []roomTurnoverRewardLevel, + contributions []integration.RoomContributionActivityCount, + roomProfiles map[int64]integration.RoomProfile, +) error { + lockKey := fmt.Sprintf("room-turnover-reward:settle-lock:%d:%s", configRow.ID, cycleKey) + acquired, err := s.repo.Redis.SetNX(ctx, lockKey, 1, 5*time.Minute).Result() + if err != nil { + return err + } + if !acquired { + return nil + } + + periodStart, periodEnd, err := s.periodBounds(cycleKey) + if err != nil { + return err + } + + now := time.Now() + var lastErr error + for _, contribution := range contributions { + roomID := int64(contribution.RoomID) + profile := roomProfiles[roomID] + level, ok := pickRoomTurnoverRewardLevel(levels, int64(contribution.ContributionValue)) + if !ok { + continue + } + record, err := s.ensureRewardRecord(ctx, configRow.ID, sysOrigin, cycleKey, periodStart, periodEnd, contribution, profile, level, now) + if err != nil { + lastErr = err + continue + } + if record.Status == roomTurnoverRewardStatusSuccess { + if contribution.ID != "" { + if _, markErr := s.java.MarkRoomTurnoverRewardCoinsSent(ctx, contribution.ID); markErr != nil { + lastErr = markErr + log.Printf("mark room turnover reward source failed. recordId=%d sourceId=%s err=%v", record.ID, contribution.ID, markErr) + } + } + continue + } + if record.Status != roomTurnoverRewardStatusPending && record.Status != roomTurnoverRewardStatusFailed { + continue + } + if err := s.dispatchRewardRecord(ctx, &record, contribution.ID); err != nil { + lastErr = err + } + } + return lastErr +} + +func (s *RoomTurnoverRewardSettlementService) loadLevels(ctx context.Context, configID int64) ([]roomTurnoverRewardLevel, error) { + var rows []model.RoomTurnoverRewardLevel + if err := s.repo.DB.WithContext(ctx). + Where("config_id = ? AND enabled = ?", configID, true). + Order("turnover_threshold desc"). + Find(&rows).Error; err != nil { + return nil, err + } + levels := make([]roomTurnoverRewardLevel, 0, len(rows)) + for _, row := range rows { + if row.TurnoverThreshold <= 0 || row.RewardGold <= 0 { + continue + } + levels = append(levels, roomTurnoverRewardLevel{ + Level: row.Level, + TurnoverThreshold: row.TurnoverThreshold, + RewardGold: row.RewardGold, + }) + } + sort.Slice(levels, func(i, j int) bool { + return levels[i].TurnoverThreshold > levels[j].TurnoverThreshold + }) + return levels, nil +} + +func (s *RoomTurnoverRewardSettlementService) ensureRewardRecord( + ctx context.Context, + configID int64, + sysOrigin string, + cycleKey string, + periodStart time.Time, + periodEnd time.Time, + contribution integration.RoomContributionActivityCount, + profile integration.RoomProfile, + level roomTurnoverRewardLevel, + now time.Time, +) (model.RoomTurnoverRewardRecord, error) { + roomID := int64(contribution.RoomID) + recordID, err := util.NextID() + if err != nil { + return model.RoomTurnoverRewardRecord{}, err + } + record := model.RoomTurnoverRewardRecord{ + ID: recordID, + ConfigID: configID, + SysOrigin: sysOrigin, + CycleKey: cycleKey, + PeriodStartAt: periodStart, + PeriodEndAt: periodEnd, + RoomID: roomID, + RoomAccount: profile.RoomAccount, + RoomName: profile.RoomName, + RoomCover: profile.RoomCover, + OwnerUserID: int64(profile.UserID), + CountryCode: profile.CountryCode, + CountryName: profile.CountryName, + TurnoverAmount: int64(contribution.ContributionValue), + Level: level.Level, + TurnoverThreshold: level.TurnoverThreshold, + RewardGold: level.RewardGold, + TrackID: roomTurnoverRewardTrackID(sysOrigin, cycleKey, roomID), + Status: roomTurnoverRewardStatusPending, + CreateTime: now, + UpdateTime: now, + } + if record.OwnerUserID <= 0 { + return model.RoomTurnoverRewardRecord{}, fmt.Errorf("room owner missing. roomId=%d", roomID) + } + if err := s.repo.DB.WithContext(ctx). + Clauses(clause.OnConflict{DoNothing: true}). + Create(&record).Error; err != nil { + return model.RoomTurnoverRewardRecord{}, err + } + + var saved model.RoomTurnoverRewardRecord + if err := s.repo.DB.WithContext(ctx). + Where("sys_origin = ? AND cycle_key = ? AND room_id = ?", sysOrigin, cycleKey, roomID). + First(&saved).Error; err != nil { + return model.RoomTurnoverRewardRecord{}, err + } + return saved, nil +} + +func (s *RoomTurnoverRewardSettlementService) dispatchRewardRecord(ctx context.Context, record *model.RoomTurnoverRewardRecord, sourceID string) error { + if record == nil { + return nil + } + now := time.Now() + retryCount := record.RetryCount + if record.Status == roomTurnoverRewardStatusFailed { + retryCount++ + } + + err := s.java.ChangeGoldBalance(ctx, integration.GoldReceiptCommand{ + ReceiptType: "INCOME", + UserID: record.OwnerUserID, + SysOrigin: record.SysOrigin, + EventID: record.TrackID, + Origin: roomTurnoverRewardGoldOrigin, + Remark: fmt.Sprintf("room turnover reward %s room %d", record.CycleKey, record.RoomID), + Amount: integration.NewPennyAmountPayloadFromDollar(record.RewardGold), + CloseDelayAsset: false, + OpUserType: "BACK", + }) + if err != nil { + updates := map[string]any{ + "status": roomTurnoverRewardStatusFailed, + "retry_count": retryCount, + "last_error": truncateString(err.Error(), 1024), + "update_time": now, + } + if updateErr := s.repo.DB.WithContext(context.Background()). + Model(&model.RoomTurnoverRewardRecord{}). + Where("id = ?", record.ID). + Updates(updates).Error; updateErr != nil { + return updateErr + } + return err + } + + if err := s.repo.DB.WithContext(ctx). + Model(&model.RoomTurnoverRewardRecord{}). + Where("id = ?", record.ID). + Updates(map[string]any{ + "status": roomTurnoverRewardStatusSuccess, + "retry_count": retryCount, + "last_error": "", + "sent_at": now, + "update_time": now, + }).Error; err != nil { + return err + } + + if strings.TrimSpace(sourceID) == "" { + return nil + } + if marked, err := s.java.MarkRoomTurnoverRewardCoinsSent(ctx, sourceID); err != nil { + return err + } else if !marked { + log.Printf("room turnover reward source was not modified. recordId=%d sourceId=%s", record.ID, sourceID) + } + return nil +} + +func (s *RoomTurnoverRewardSettlementService) periodBounds(cycleKey string) (time.Time, time.Time, error) { + start, err := time.ParseInLocation("20060102", cycleKey, s.location) + if err != nil { + return time.Time{}, time.Time{}, err + } + return start, start.AddDate(0, 0, 7), nil +} + +func pickRoomTurnoverRewardLevel(levels []roomTurnoverRewardLevel, turnoverAmount int64) (roomTurnoverRewardLevel, bool) { + if turnoverAmount <= 0 { + return roomTurnoverRewardLevel{}, false + } + for _, level := range levels { + if turnoverAmount >= level.TurnoverThreshold { + return level, true + } + } + return roomTurnoverRewardLevel{}, false +} + +func uniqueRoomTurnoverRewardRoomIDs(contributions []integration.RoomContributionActivityCount) []int64 { + result := make([]int64, 0, len(contributions)) + seen := make(map[int64]struct{}, len(contributions)) + for _, item := range contributions { + roomID := int64(item.RoomID) + if roomID <= 0 { + continue + } + if _, exists := seen[roomID]; exists { + continue + } + seen[roomID] = struct{}{} + result = append(result, roomID) + } + return result +} + +func normalizeRoomTurnoverRewardSysOrigin(values ...string) string { + for _, value := range values { + normalized := strings.ToUpper(strings.TrimSpace(value)) + if normalized != "" { + return normalized + } + } + return "LIKEI" +} + +func roomTurnoverRewardCycleKey(dateNumber int, now time.Time) string { + if dateNumber > 0 { + return fmt.Sprintf("%08d", dateNumber) + } + weekday := int(now.Weekday()) + if weekday == 0 { + weekday = 7 + } + monday := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()).AddDate(0, 0, 1-weekday-7) + return monday.Format("20060102") +} + +func roomTurnoverRewardTrackID(sysOrigin string, cycleKey string, roomID int64) string { + return "room-turnover:" + strings.ToUpper(strings.TrimSpace(sysOrigin)) + ":" + cycleKey + ":" + strconv.FormatInt(roomID, 10) +}