From 2fb3a1e5f5c8ecd0991d2f36b7b7c038fad9c504 Mon Sep 17 00:00:00 2001 From: hy Date: Sat, 18 Apr 2026 18:52:50 +0800 Subject: [PATCH] feat: add week star reward retry worker --- .gitignore | 3 + README.md | 27 ++- cmd/week-star-settle-once/main.go | 34 ++++ internal/config/config.go | 105 ++++++---- internal/config/config_test.go | 110 +++++++++++ internal/model/models.go | 18 +- internal/service/week_star_retry.go | 241 +++++++++++++++++++++++ internal/service/week_star_settlement.go | 44 +++-- 8 files changed, 517 insertions(+), 65 deletions(-) create mode 100644 cmd/week-star-settle-once/main.go create mode 100644 internal/config/config_test.go create mode 100644 internal/service/week_star_retry.go diff --git a/.gitignore b/.gitignore index 2f7896d..2eebe1b 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,4 @@ target/ +.DS_Store +/week-star-job +/week-star-settle-once diff --git a/README.md b/README.md index 7d24994..e1039fe 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ Standalone cron workers for chatapp activities. Current worker: -- `week-star-job`: specified-gift weekly rank settlement and reward dispatch +- `week-star-job`: specified-gift weekly rank settlement, reward dispatch, and failed reward retry Local run: @@ -22,14 +22,23 @@ cd /Users/hy/Documents/hy/chatapp3/chatapp3-deploy Required envs: -- `CRON_MYSQL_DSN` -- `CRON_REDIS_ADDR` -- `CRON_REDIS_PASSWORD` -- `CRON_REDIS_DB` -- `CRON_JAVA_OTHER_BASE_URL` -- `WEEK_STAR_DEFAULT_SYS_ORIGIN` -- `WEEK_STAR_TIMEZONE` -- `WEEK_STAR_SETTLE_INTERVAL_SECONDS` +- `CHATAPP_STORE_MYSQL_DSN` +- `CHATAPP_STORE_REDIS_ADDR` +- `CHATAPP_STORE_REDIS_PASSWORD` +- `CHATAPP_STORE_REDIS_DB` +- `CHATAPP_JAVA_OTHER_BASE_URL` +- `CHATAPP_WEEK_STAR_DEFAULT_SYS_ORIGIN` +- `CHATAPP_WEEK_STAR_TIMEZONE` +- `CHATAPP_WEEK_STAR_SETTLE_INTERVAL_SECONDS` +- `CHATAPP_WEEK_STAR_RETRY_STREAM_KEY` +- `CHATAPP_WEEK_STAR_RETRY_CONSUMER_GROUP` +- `CHATAPP_WEEK_STAR_RETRY_CONSUMER_NAME` +- `CHATAPP_WEEK_STAR_RETRY_BATCH_SIZE` +- `CHATAPP_WEEK_STAR_RETRY_BLOCK_MILLIS` +- `CHATAPP_WEEK_STAR_RETRY_MIN_IDLE_MILLIS` +- `CHATAPP_HTTP_TIMEOUT_SECONDS` + +Legacy `CRON_*` / `WEEK_STAR_*` keys are still supported as fallback, but deployment should use the `CHATAPP_*` namespace as the canonical source. Production topology: diff --git a/cmd/week-star-settle-once/main.go b/cmd/week-star-settle-once/main.go new file mode 100644 index 0000000..b4d17e9 --- /dev/null +++ b/cmd/week-star-settle-once/main.go @@ -0,0 +1,34 @@ +package main + +import ( + "chatapp-cron/internal/config" + "chatapp-cron/internal/integration" + "chatapp-cron/internal/repo" + "chatapp-cron/internal/service" + "context" + "log" + "time" +) + +func main() { + cfg := config.Load() + + repository, err := repo.New(cfg) + if err != nil { + log.Fatalf("init repository failed: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + if err := repository.Ping(ctx); err != nil { + log.Fatalf("ping dependencies failed: %v", err) + } + + javaClient := integration.New(cfg) + settlementService := service.NewWeekStarSettlementService(cfg, repository, javaClient) + if err := settlementService.SettleDueCycles(ctx); err != nil { + log.Fatalf("settle due cycles failed: %v", err) + } + + log.Printf("week-star settle once finished") +} diff --git a/internal/config/config.go b/internal/config/config.go index 7cda770..652806f 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -8,52 +8,79 @@ import ( ) type Config struct { - MySQLDSN string - RedisAddr string - RedisPassword string - RedisDB int - JavaOtherBaseURL string - WeekStarDefaultSysOrigin string - WeekStarTimezone string - WeekStarSettleInterval time.Duration - Timeout time.Duration - ShutdownTimeout time.Duration - AutoMigrate bool + 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 } func Load() Config { return Config{ - MySQLDSN: firstEnv("CRON_MYSQL_DSN", "INVITE_MYSQL_DSN", "root:root@tcp(127.0.0.1:3306)/chatapp3?charset=utf8mb4&parseTime=True&loc=Local"), - RedisAddr: firstEnv("CRON_REDIS_ADDR", "INVITE_REDIS_ADDR", "127.0.0.1:6379"), - RedisPassword: firstEnv("CRON_REDIS_PASSWORD", "INVITE_REDIS_PASSWORD", ""), - RedisDB: getEnvInt("CRON_REDIS_DB", getEnvInt("INVITE_REDIS_DB", 0)), - JavaOtherBaseURL: firstEnv("CRON_JAVA_OTHER_BASE_URL", "GAME_JAVA_OTHER_BASE_URL", "INVITE_JAVA_APP_BASE_URL", "http://127.0.0.1:2400"), - WeekStarDefaultSysOrigin: strings.ToUpper(firstEnv("WEEK_STAR_DEFAULT_SYS_ORIGIN", "", "LIKEI")), - WeekStarTimezone: firstEnv("WEEK_STAR_TIMEZONE", "", "Asia/Riyadh"), - WeekStarSettleInterval: time.Duration(getEnvInt("WEEK_STAR_SETTLE_INTERVAL_SECONDS", 60)) * time.Second, - Timeout: time.Duration(getEnvInt("CRON_HTTP_TIMEOUT_SECONDS", getEnvInt("INVITE_HTTP_TIMEOUT_SECONDS", 8))) * time.Second, - ShutdownTimeout: time.Duration(getEnvInt("CRON_SHUTDOWN_TIMEOUT_SECONDS", getEnvInt("INVITE_HTTP_SHUTDOWN_TIMEOUT_SECONDS", 30))) * time.Second, - AutoMigrate: getEnvBool("CRON_AUTO_MIGRATE", getEnvBool("INVITE_AUTO_MIGRATE", false)), + MySQLDSN: getEnvAny( + []string{"CHATAPP_STORE_MYSQL_DSN", "CRON_MYSQL_DSN", "INVITE_MYSQL_DSN"}, + "root:root@tcp(127.0.0.1:3306)/chatapp3?charset=utf8mb4&parseTime=True&loc=Local", + ), + RedisAddr: getEnvAny( + []string{"CHATAPP_STORE_REDIS_ADDR", "CRON_REDIS_ADDR", "INVITE_REDIS_ADDR"}, + "127.0.0.1:6379", + ), + RedisPassword: getEnvAny( + []string{"CHATAPP_STORE_REDIS_PASSWORD", "CRON_REDIS_PASSWORD", "INVITE_REDIS_PASSWORD"}, + "", + ), + RedisDB: getEnvIntAny( + []string{"CHATAPP_STORE_REDIS_DB", "CRON_REDIS_DB", "INVITE_REDIS_DB"}, + 0, + ), + JavaOtherBaseURL: getEnvAny( + []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", + ), + 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, + 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")), + WeekStarRetryBatchSize: int64(getEnvIntAny([]string{"CHATAPP_WEEK_STAR_RETRY_BATCH_SIZE", "WEEK_STAR_RETRY_BATCH_SIZE"}, 10)), + WeekStarRetryBlock: time.Duration(getEnvIntAny([]string{"CHATAPP_WEEK_STAR_RETRY_BLOCK_MILLIS", "WEEK_STAR_RETRY_BLOCK_MILLIS"}, 5000)) * time.Millisecond, + WeekStarRetryMinIdle: time.Duration(getEnvIntAny([]string{"CHATAPP_WEEK_STAR_RETRY_MIN_IDLE_MILLIS", "WEEK_STAR_RETRY_MIN_IDLE_MILLIS"}, 60000)) * time.Millisecond, + Timeout: time.Duration(getEnvIntAny([]string{"CHATAPP_HTTP_TIMEOUT_SECONDS", "CRON_HTTP_TIMEOUT_SECONDS", "INVITE_HTTP_TIMEOUT_SECONDS"}, 8)) * time.Second, + ShutdownTimeout: time.Duration(getEnvIntAny([]string{"CHATAPP_HTTP_SHUTDOWN_TIMEOUT_SECONDS", "CRON_SHUTDOWN_TIMEOUT_SECONDS", "INVITE_HTTP_SHUTDOWN_TIMEOUT_SECONDS"}, 30)) * time.Second, + AutoMigrate: getEnvBoolAny([]string{"CHATAPP_AUTO_MIGRATE", "CRON_AUTO_MIGRATE", "INVITE_AUTO_MIGRATE"}, false), } } -func firstEnv(keys ...string) string { - if len(keys) == 0 { - return "" - } - for _, key := range keys[:len(keys)-1] { - if strings.TrimSpace(key) == "" { +func getEnvAny(keys []string, fallback string) string { + for _, key := range keys { + key = strings.TrimSpace(key) + if key == "" { continue } if value := strings.TrimSpace(os.Getenv(key)); value != "" { return value } } - return keys[len(keys)-1] + return fallback } -func getEnvInt(key string, fallback int) int { - value := strings.TrimSpace(os.Getenv(key)) +func getEnvIntAny(keys []string, fallback int) int { + value := getEnvAny(keys, "") if value == "" { return fallback } @@ -64,8 +91,8 @@ func getEnvInt(key string, fallback int) int { return parsed } -func getEnvBool(key string, fallback bool) bool { - value := strings.TrimSpace(os.Getenv(key)) +func getEnvBoolAny(keys []string, fallback bool) bool { + value := getEnvAny(keys, "") if value == "" { return fallback } @@ -75,3 +102,15 @@ func getEnvBool(key string, fallback bool) bool { } return parsed } + +func defaultConsumerName(prefix string) string { + host, err := os.Hostname() + if err != nil { + return prefix + } + host = strings.TrimSpace(host) + if host == "" { + return prefix + } + return prefix + "-" + host +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..b053591 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,110 @@ +package config + +import ( + "testing" + "time" +) + +func TestLoadPrefersChatappKeys(t *testing.T) { + t.Setenv("CHATAPP_STORE_MYSQL_DSN", "chatapp-mysql") + t.Setenv("CRON_MYSQL_DSN", "legacy-mysql") + t.Setenv("CHATAPP_STORE_REDIS_ADDR", "chatapp-redis") + 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_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_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") + t.Setenv("CHATAPP_WEEK_STAR_RETRY_BATCH_SIZE", "11") + t.Setenv("CHATAPP_WEEK_STAR_RETRY_BLOCK_MILLIS", "2500") + t.Setenv("CHATAPP_WEEK_STAR_RETRY_MIN_IDLE_MILLIS", "4500") + t.Setenv("CHATAPP_HTTP_TIMEOUT_SECONDS", "9") + t.Setenv("CHATAPP_HTTP_SHUTDOWN_TIMEOUT_SECONDS", "19") + t.Setenv("CHATAPP_AUTO_MIGRATE", "true") + + cfg := Load() + + if cfg.MySQLDSN != "chatapp-mysql" { + t.Fatalf("expected chatapp mysql dsn, got %q", cfg.MySQLDSN) + } + if cfg.RedisAddr != "chatapp-redis" || cfg.RedisPassword != "chatapp-password" || cfg.RedisDB != 5 { + t.Fatalf("unexpected redis config: %+v", cfg) + } + if cfg.JavaOtherBaseURL != "http://chatapp-java" { + t.Fatalf("expected chatapp java base url, got %q", cfg.JavaOtherBaseURL) + } + if cfg.WeekStarDefaultSysOrigin != "LIKEI" { + t.Fatalf("expected normalized sys origin LIKEI, got %q", cfg.WeekStarDefaultSysOrigin) + } + if cfg.WeekStarTimezone != "Asia/Shanghai" { + t.Fatalf("expected chatapp timezone, got %q", cfg.WeekStarTimezone) + } + if cfg.WeekStarSettleInterval != 120*time.Second { + t.Fatalf("expected settle interval 120s, got %s", cfg.WeekStarSettleInterval) + } + if cfg.WeekStarRetryStreamKey != "chatapp-stream" || cfg.WeekStarRetryConsumerGroup != "chatapp-group" || cfg.WeekStarRetryConsumerName != "chatapp-consumer" { + t.Fatalf("unexpected retry consumer config: %+v", cfg) + } + if cfg.WeekStarRetryBatchSize != 11 || cfg.WeekStarRetryBlock != 2500*time.Millisecond || cfg.WeekStarRetryMinIdle != 4500*time.Millisecond { + t.Fatalf("unexpected retry timing config: %+v", cfg) + } + if cfg.Timeout != 9*time.Second || cfg.ShutdownTimeout != 19*time.Second { + t.Fatalf("unexpected http timeout config: %+v", cfg) + } + if !cfg.AutoMigrate { + t.Fatalf("expected auto migrate to be true") + } +} + +func TestLoadFallsBackToLegacyKeys(t *testing.T) { + t.Setenv("CRON_MYSQL_DSN", "legacy-mysql") + t.Setenv("CRON_REDIS_ADDR", "legacy-redis") + 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("WEEK_STAR_DEFAULT_SYS_ORIGIN", "tarab") + t.Setenv("WEEK_STAR_TIMEZONE", "Asia/Riyadh") + t.Setenv("WEEK_STAR_SETTLE_INTERVAL_SECONDS", "66") + 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") + t.Setenv("WEEK_STAR_RETRY_BATCH_SIZE", "7") + t.Setenv("WEEK_STAR_RETRY_BLOCK_MILLIS", "1700") + t.Setenv("WEEK_STAR_RETRY_MIN_IDLE_MILLIS", "2700") + t.Setenv("CRON_HTTP_TIMEOUT_SECONDS", "4") + t.Setenv("CRON_SHUTDOWN_TIMEOUT_SECONDS", "14") + t.Setenv("CRON_AUTO_MIGRATE", "true") + + cfg := Load() + + if cfg.MySQLDSN != "legacy-mysql" { + t.Fatalf("expected legacy mysql dsn, got %q", cfg.MySQLDSN) + } + if cfg.RedisAddr != "legacy-redis" || cfg.RedisPassword != "legacy-password" || cfg.RedisDB != 3 { + t.Fatalf("unexpected redis config: %+v", cfg) + } + if cfg.JavaOtherBaseURL != "http://legacy-java" { + t.Fatalf("expected legacy java base url, got %q", cfg.JavaOtherBaseURL) + } + 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.WeekStarRetryStreamKey != "legacy-stream" || cfg.WeekStarRetryConsumerGroup != "legacy-group" || cfg.WeekStarRetryConsumerName != "legacy-consumer" { + t.Fatalf("unexpected retry consumer config: %+v", cfg) + } + if cfg.WeekStarRetryBatchSize != 7 || cfg.WeekStarRetryBlock != 1700*time.Millisecond || cfg.WeekStarRetryMinIdle != 2700*time.Millisecond { + t.Fatalf("unexpected retry timing config: %+v", cfg) + } + if cfg.Timeout != 4*time.Second || cfg.ShutdownTimeout != 14*time.Second { + t.Fatalf("unexpected http timeout config: %+v", cfg) + } + if !cfg.AutoMigrate { + t.Fatalf("expected auto migrate to be true") + } +} diff --git a/internal/model/models.go b/internal/model/models.go index 40dd776..1c68d11 100644 --- a/internal/model/models.go +++ b/internal/model/models.go @@ -17,16 +17,16 @@ type UserBaseInfo struct { // TableName 返回用户基础资料表名。 func (UserBaseInfo) TableName() string { return "user_base_info" } -// WeekStarActivityConfig 对应周榜活动主配置，一条记录描述一个系统下的活动时间窗和开关状态。 +// WeekStarActivityConfig 对应周榜活动主配置，一条记录描述一个系统下的一个周榜活动时间窗。 type WeekStarActivityConfig struct { - ID int64 `gorm:"column:id;primaryKey"` // 配置主键 - SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_week_star_sys_origin"` // 系统标识，例如 LIKEI - Enabled bool `gorm:"column:enabled"` // 活动开关 - StartAt time.Time `gorm:"column:start_at"` // 活动开始时间 - EndAt time.Time `gorm:"column:end_at"` // 活动结束时间 - Timezone string `gorm:"column:timezone;size:64"` // 活动结算时区 - CreateTime time.Time `gorm:"column:create_time"` // 创建时间 - UpdateTime time.Time `gorm:"column:update_time"` // 更新时间 + ID int64 `gorm:"column:id;primaryKey"` // 配置主键 + SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_week_star_sys_origin_window,priority:1;index:idx_week_star_active_lookup,priority:1"` // 系统标识，例如 LIKEI + Enabled bool `gorm:"column:enabled;index:idx_week_star_active_lookup,priority:2"` // 活动开关 + StartAt time.Time `gorm:"column:start_at;uniqueIndex:uk_week_star_sys_origin_window,priority:2;index:idx_week_star_active_lookup,priority:3"` // 活动开始时间 + EndAt time.Time `gorm:"column:end_at;uniqueIndex:uk_week_star_sys_origin_window,priority:3;index:idx_week_star_active_lookup,priority:4"` // 活动结束时间 + Timezone string `gorm:"column:timezone;size:64"` // 活动结算时区 + CreateTime time.Time `gorm:"column:create_time"` // 创建时间 + UpdateTime time.Time `gorm:"column:update_time"` // 更新时间 } // TableName 返回周榜活动主配置表名。 diff --git a/internal/service/week_star_retry.go b/internal/service/week_star_retry.go new file mode 100644 index 0000000..ea94aea --- /dev/null +++ b/internal/service/week_star_retry.go @@ -0,0 +1,241 @@ +package service + +import ( + "context" + "errors" + "fmt" + "log" + "strconv" + "strings" + "time" + + "chatapp-cron/internal/model" + + "github.com/redis/go-redis/v9" + "gorm.io/gorm" +) + +const weekStarRewardRetryPendingTTL = 30 * time.Minute + +// startRetryWorker 启动失败补发任务消费协程，统一处理后台人工补发请求。 +func (s *WeekStarSettlementService) startRetryWorker(ctx context.Context) { + if strings.TrimSpace(s.cfg.WeekStarRetryStreamKey) == "" { + log.Printf("week star retry consumer disabled: stream key is blank") + return + } + if err := s.ensureRetryConsumerGroup(ctx); err != nil { + log.Printf("ensure week star retry consumer group failed before start: %v", err) + } + log.Printf( + "week star retry consumer started. stream=%s group=%s consumer=%s", + s.cfg.WeekStarRetryStreamKey, + s.cfg.WeekStarRetryConsumerGroup, + s.cfg.WeekStarRetryConsumerName, + ) + go s.consumeRetryLoop(ctx) +} + +// consumeRetryLoop 先认领超时 pending 任务，再持续读取新补发任务。 +func (s *WeekStarSettlementService) consumeRetryLoop(ctx context.Context) { + for { + if ctx.Err() != nil { + return + } + + reclaimed, err := s.reclaimPendingRetryTasks(ctx) + if err != nil { + s.handleRetryLoopError(ctx, "reclaim week star retry tasks", err) + continue + } + if reclaimed > 0 { + continue + } + + err = s.readNewRetryTasks(ctx) + if err == nil || errors.Is(err, redis.Nil) { + continue + } + s.handleRetryLoopError(ctx, "read week star retry tasks", err) + } +} + +// handleRetryLoopError 统一处理补发消费循环错误，并在必要时重建消费者组。 +func (s *WeekStarSettlementService) handleRetryLoopError(ctx context.Context, action string, err error) { + if ctx.Err() != nil { + return + } + if isWeekStarRetryNoGroupErr(err) { + if ensureErr := s.ensureRetryConsumerGroup(ctx); ensureErr != nil { + log.Printf("%s failed and recreate retry consumer group failed: %v", action, ensureErr) + sleepWeekStarRetry(ctx, time.Second) + return + } + return + } + log.Printf("%s failed: %v", action, err) + sleepWeekStarRetry(ctx, time.Second) +} + +// ensureRetryConsumerGroup 确保失败补发 Redis Stream 消费组存在。 +func (s *WeekStarSettlementService) ensureRetryConsumerGroup(ctx context.Context) error { + err := s.repo.Redis.XGroupCreateMkStream( + ctx, + s.cfg.WeekStarRetryStreamKey, + s.cfg.WeekStarRetryConsumerGroup, + "0", + ).Err() + if err != nil && !strings.Contains(strings.ToUpper(err.Error()), "BUSYGROUP") { + return err + } + return nil +} + +// reclaimPendingRetryTasks 认领长时间未确认的补发任务，避免 worker 中断后消息遗失。 +func (s *WeekStarSettlementService) reclaimPendingRetryTasks(ctx context.Context) (int, error) { + messages, _, err := s.repo.Redis.XAutoClaim(ctx, &redis.XAutoClaimArgs{ + Stream: s.cfg.WeekStarRetryStreamKey, + Group: s.cfg.WeekStarRetryConsumerGroup, + Consumer: s.cfg.WeekStarRetryConsumerName, + MinIdle: s.cfg.WeekStarRetryMinIdle, + Start: "0-0", + Count: s.cfg.WeekStarRetryBatchSize, + }).Result() + if err != nil { + return 0, err + } + if len(messages) == 0 { + return 0, nil + } + return len(messages), s.processRetryMessages(ctx, messages) +} + +// readNewRetryTasks 按消费者组协议读取新的补发任务。 +func (s *WeekStarSettlementService) readNewRetryTasks(ctx context.Context) error { + streams, err := s.repo.Redis.XReadGroup(ctx, &redis.XReadGroupArgs{ + Group: s.cfg.WeekStarRetryConsumerGroup, + Consumer: s.cfg.WeekStarRetryConsumerName, + Streams: []string{s.cfg.WeekStarRetryStreamKey, ">"}, + Count: s.cfg.WeekStarRetryBatchSize, + Block: s.cfg.WeekStarRetryBlock, + }).Result() + if err != nil { + return err + } + for _, stream := range streams { + if err := s.processRetryMessages(ctx, stream.Messages); err != nil { + return err + } + } + return nil +} + +// processRetryMessages 顺序处理补发任务，并在完成状态落库后 ACK + DEL。 +func (s *WeekStarSettlementService) processRetryMessages(ctx context.Context, messages []redis.XMessage) error { + for _, message := range messages { + ack, recordID, err := s.processRetryMessage(ctx, message) + if ack { + if ackErr := s.repo.Redis.XAck( + context.Background(), + s.cfg.WeekStarRetryStreamKey, + s.cfg.WeekStarRetryConsumerGroup, + message.ID, + ).Err(); ackErr != nil { + log.Printf("ack week star retry task failed. streamId=%s err=%v", message.ID, ackErr) + } + if delErr := s.repo.Redis.XDel(context.Background(), s.cfg.WeekStarRetryStreamKey, message.ID).Err(); delErr != nil { + log.Printf("delete week star retry task failed. streamId=%s err=%v", message.ID, delErr) + } + if recordID > 0 { + _, _ = s.repo.Redis.Del(context.Background(), s.rewardRetryPendingKey(recordID)).Result() + } + } + if err != nil { + return err + } + } + return nil +} + +// processRetryMessage 处理单条补发任务；只有在状态已经稳定落库后才确认消息。 +func (s *WeekStarSettlementService) processRetryMessage(ctx context.Context, message redis.XMessage) (bool, int64, error) { + recordID, err := parseWeekStarRetryRecordID(message) + if err != nil { + log.Printf("drop malformed week star retry task. streamId=%s err=%v values=%v", message.ID, err, message.Values) + return true, 0, nil + } + + var record model.WeekStarRewardRecord + err = s.repo.DB.WithContext(ctx).Where("id = ?", recordID).First(&record).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + log.Printf("skip missing week star retry record. recordId=%d", recordID) + return true, recordID, nil + } + if err != nil { + return false, recordID, err + } + if record.Status != weekStarRewardStatusFailed { + return true, recordID, nil + } + + handled, err := s.dispatchRewardRecord(ctx, &record, true) + if err != nil { + if handled { + log.Printf("week star manual retry finished with failed status. recordId=%d err=%v", recordID, err) + return true, recordID, nil + } + return false, recordID, err + } + return true, recordID, nil +} + +// parseWeekStarRetryRecordID 把 Redis Stream 消息解析成补发记录主键。 +func parseWeekStarRetryRecordID(message redis.XMessage) (int64, error) { + return toWeekStarRetryInt64(message.Values["recordId"]) +} + +// toWeekStarRetryInt64 把 Stream 字段安全转换成 int64。 +func toWeekStarRetryInt64(value any) (int64, error) { + switch typed := value.(type) { + case nil: + return 0, fmt.Errorf("value is nil") + case int64: + return typed, nil + case int32: + return int64(typed), nil + case int: + return int64(typed), nil + case uint64: + return int64(typed), nil + case float64: + return int64(typed), nil + case string: + return strconv.ParseInt(strings.TrimSpace(typed), 10, 64) + case []byte: + return strconv.ParseInt(strings.TrimSpace(string(typed)), 10, 64) + default: + return strconv.ParseInt(strings.TrimSpace(fmt.Sprint(typed)), 10, 64) + } +} + +// isWeekStarRetryNoGroupErr 判断是否是消费者组不存在错误。 +func isWeekStarRetryNoGroupErr(err error) bool { + if err == nil { + return false + } + return strings.Contains(strings.ToUpper(err.Error()), "NOGROUP") +} + +// rewardRetryPendingKey 返回补发任务去重 key，和 chatapp3-golang 保持一致。 +func (s *WeekStarSettlementService) rewardRetryPendingKey(recordID int64) string { + return strings.TrimSpace(s.cfg.WeekStarRetryStreamKey) + ":pending:" + strconv.FormatInt(recordID, 10) +} + +// sleepWeekStarRetry 支持被上下文中断的短暂退避休眠。 +func sleepWeekStarRetry(ctx context.Context, duration time.Duration) { + timer := time.NewTimer(duration) + defer timer.Stop() + select { + case <-ctx.Done(): + case <-timer.C: + } +} diff --git a/internal/service/week_star_settlement.go b/internal/service/week_star_settlement.go index 922e26f..9298503 100644 --- a/internal/service/week_star_settlement.go +++ b/internal/service/week_star_settlement.go @@ -82,8 +82,14 @@ func NewWeekStarSettlementService(cfg config.Config, repository *repo.Repository } } -// Start 启动定时扫描任务，按照配置周期持续检查是否存在待结算周榜。 +// Start 启动周榜结算扫描和失败补发消费任务。 func (s *WeekStarSettlementService) Start(ctx context.Context) { + s.startSettlementLoop(ctx) + s.startRetryWorker(ctx) +} + +// startSettlementLoop 按配置周期持续检查是否存在待结算周榜。 +func (s *WeekStarSettlementService) startSettlementLoop(ctx context.Context) { interval := s.cfg.WeekStarSettleInterval if interval <= 0 { interval = time.Minute @@ -195,7 +201,7 @@ func (s *WeekStarSettlementService) settleCycle(ctx context.Context, configRow * var rewardConfigs []model.WeekStarActivityRewardConfig if err := s.repo.DB.WithContext(ctx). Where("config_id = ?", configRow.ID). - Order("rank asc"). + Order("`rank` asc"). Find(&rewardConfigs).Error; err != nil { return err } @@ -282,14 +288,15 @@ func (s *WeekStarSettlementService) settleCycle(ctx context.Context, configRow * var pendingRecords []model.WeekStarRewardRecord if err := s.repo.DB.WithContext(ctx). Where("config_id = ? AND cycle_key = ? AND status = ?", configRow.ID, cycleKey, weekStarRewardStatusPending). - Order("rank asc"). + Order("`rank` asc"). Find(&pendingRecords).Error; err != nil { return err } var lastErr error for index := range pendingRecords { // 发奖逐条执行，单条失败不会阻塞同周期其他名次继续发放。 - if err := s.dispatchRewardRecord(ctx, &pendingRecords[index]); err != nil { + _, err := s.dispatchRewardRecord(ctx, &pendingRecords[index], false) + if err != nil { lastErr = err } } @@ -297,16 +304,22 @@ func (s *WeekStarSettlementService) settleCycle(ctx context.Context, configRow * } // dispatchRewardRecord 调用 Java 奖励接口发奖，并回写发奖成功或失败状态。 -func (s *WeekStarSettlementService) dispatchRewardRecord(ctx context.Context, record *model.WeekStarRewardRecord) error { +// handled=true 表示本次任务已经完成状态落库，调用方可以安全确认消息； +// handled=false 表示处理过程中断，调用方应保留消息供后续继续消费。 +func (s *WeekStarSettlementService) dispatchRewardRecord(ctx context.Context, record *model.WeekStarRewardRecord, isRetry bool) (bool, error) { if record == nil { - return nil + return true, nil } var configRow model.WeekStarActivityConfig if err := s.repo.DB.WithContext(ctx).Where("id = ?", record.ConfigID).First(&configRow).Error; err != nil { - return err + return false, err } now := time.Now() + retryCount := record.RetryCount + if isRetry { + retryCount++ + } err := s.java.SendActivityReward(ctx, integration.SendActivityRewardRequest{ TrackID: record.TrackID, Origin: weekStarRewardOrigin, @@ -318,7 +331,7 @@ func (s *WeekStarSettlementService) dispatchRewardRecord(ctx context.Context, re // 发奖失败时保留错误信息，方便后台做失败补发。 updates := map[string]any{ "status": weekStarRewardStatusFailed, - "retry_count": record.RetryCount, + "retry_count": retryCount, "last_error": truncateString(err.Error(), 1024), "update_time": now, } @@ -326,21 +339,24 @@ func (s *WeekStarSettlementService) dispatchRewardRecord(ctx context.Context, re Model(&model.WeekStarRewardRecord{}). Where("id = ?", record.ID). Updates(updates).Error; updateErr != nil { - return updateErr + return false, updateErr } - return err + return true, err } - return s.repo.DB.WithContext(ctx). + if err := s.repo.DB.WithContext(ctx). Model(&model.WeekStarRewardRecord{}). Where("id = ?", record.ID). Updates(map[string]any{ "status": weekStarRewardStatusSuccess, - "retry_count": record.RetryCount, + "retry_count": retryCount, "last_error": "", "sent_at": now, "update_time": now, - }).Error + }).Error; err != nil { + return false, err + } + return true, nil } // loadLiveRanking 从 Redis 实时榜读取前 N 名，并补齐用户资料供快照/发奖使用。 @@ -464,7 +480,7 @@ func (s *WeekStarSettlementService) toStorageWallClock(t time.Time) time.Time { local.Minute(), local.Second(), local.Nanosecond(), - time.Local, + s.location, ) }