Add room turnover reward settlement

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`

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)

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

internal/config/config.go

@@ -13,9 +13,13 @@ type Config struct {
 	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
@@ -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,
+		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")),

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

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
+}

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" }

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