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

feat: add week star reward retry worker

Browse Source
This commit is contained in:
hy 2026-04-18 18:52:50 +08:00
parent a311310180
commit 2fb3a1e5f5
8 changed files with 517 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@ -1 +1,4 @@
target/ target/
.DS_Store
/week-star-job
/week-star-settle-once

27
README.md
View File

@ -3,7 +3,7 @@
Standalone cron workers for chatapp activities. Standalone cron workers for chatapp activities.
Current worker: 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: Local run:
@ -22,14 +22,23 @@ cd /Users/hy/Documents/hy/chatapp3/chatapp3-deploy
Required envs: Required envs:
- `CRON_MYSQL_DSN` - `CHATAPP_STORE_MYSQL_DSN`
- `CRON_REDIS_ADDR` - `CHATAPP_STORE_REDIS_ADDR`
- `CRON_REDIS_PASSWORD` - `CHATAPP_STORE_REDIS_PASSWORD`
- `CRON_REDIS_DB` - `CHATAPP_STORE_REDIS_DB`
- `CRON_JAVA_OTHER_BASE_URL` - `CHATAPP_JAVA_OTHER_BASE_URL`
- `WEEK_STAR_DEFAULT_SYS_ORIGIN` - `CHATAPP_WEEK_STAR_DEFAULT_SYS_ORIGIN`
- `WEEK_STAR_TIMEZONE` - `CHATAPP_WEEK_STAR_TIMEZONE`
- `WEEK_STAR_SETTLE_INTERVAL_SECONDS` - `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: Production topology:

34
cmd/week-star-settle-once/main.go Normal file
View File

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

105
internal/config/config.go
View File

@ -8,52 +8,79 @@ 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
WeekStarDefaultSysOrigin string WeekStarDefaultSysOrigin string
WeekStarTimezone string WeekStarTimezone string
WeekStarSettleInterval time.Duration WeekStarSettleInterval time.Duration
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 {
return 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"), MySQLDSN: getEnvAny(
RedisAddr: firstEnv("CRON_REDIS_ADDR", "INVITE_REDIS_ADDR", "127.0.0.1:6379"), []string{"CHATAPP_STORE_MYSQL_DSN", "CRON_MYSQL_DSN", "INVITE_MYSQL_DSN"},
RedisPassword: firstEnv("CRON_REDIS_PASSWORD", "INVITE_REDIS_PASSWORD", ""), "root:root@tcp(127.0.0.1:3306)/chatapp3?charset=utf8mb4&parseTime=True&loc=Local",
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"), RedisAddr: getEnvAny(
WeekStarDefaultSysOrigin: strings.ToUpper(firstEnv("WEEK_STAR_DEFAULT_SYS_ORIGIN", "", "LIKEI")), []string{"CHATAPP_STORE_REDIS_ADDR", "CRON_REDIS_ADDR", "INVITE_REDIS_ADDR"},
WeekStarTimezone: firstEnv("WEEK_STAR_TIMEZONE", "", "Asia/Riyadh"), "127.0.0.1:6379",
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, RedisPassword: getEnvAny(
ShutdownTimeout: time.Duration(getEnvInt("CRON_SHUTDOWN_TIMEOUT_SECONDS", getEnvInt("INVITE_HTTP_SHUTDOWN_TIMEOUT_SECONDS", 30))) * time.Second, []string{"CHATAPP_STORE_REDIS_PASSWORD", "CRON_REDIS_PASSWORD", "INVITE_REDIS_PASSWORD"},
AutoMigrate: getEnvBool("CRON_AUTO_MIGRATE", getEnvBool("INVITE_AUTO_MIGRATE", false)), "",
),
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 { func getEnvAny(keys []string, fallback string) string {
if len(keys) == 0 { for _, key := range keys {
return "" key = strings.TrimSpace(key)
} if key == "" {
for _, key := range keys[:len(keys)-1] {
if strings.TrimSpace(key) == "" {
continue continue
} }
if value := strings.TrimSpace(os.Getenv(key)); value != "" { if value := strings.TrimSpace(os.Getenv(key)); value != "" {
return value return value
} }
} }
return keys[len(keys)-1] return fallback
} }
func getEnvInt(key string, fallback int) int { func getEnvIntAny(keys []string, fallback int) int {
value := strings.TrimSpace(os.Getenv(key)) value := getEnvAny(keys, "")
if value == "" { if value == "" {
return fallback return fallback
} }
@ -64,8 +91,8 @@ func getEnvInt(key string, fallback int) int {
return parsed return parsed
} }
func getEnvBool(key string, fallback bool) bool { func getEnvBoolAny(keys []string, fallback bool) bool {
value := strings.TrimSpace(os.Getenv(key)) value := getEnvAny(keys, "")
if value == "" { if value == "" {
return fallback return fallback
} }
@ -75,3 +102,15 @@ func getEnvBool(key string, fallback bool) bool {
} }
return parsed 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
}

110
internal/config/config_test.go Normal file
View File

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

18
internal/model/models.go
View File

@ -17,16 +17,16 @@ type UserBaseInfo struct {
// TableName 返回用户基础资料表名。 // TableName 返回用户基础资料表名。
func (UserBaseInfo) TableName() string { return "user_base_info" } func (UserBaseInfo) TableName() string { return "user_base_info" }
// WeekStarActivityConfig 对应周榜活动主配置，一条记录描述一个系统下的活动时间窗和开关状态。 // WeekStarActivityConfig 对应周榜活动主配置，一条记录描述一个系统下的一个周榜活动时间窗。
type WeekStarActivityConfig struct { type WeekStarActivityConfig struct {
ID int64 `gorm:"column:id;primaryKey"` // 配置主键 ID int64 `gorm:"column:id;primaryKey"` // 配置主键
SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_week_star_sys_origin"` // 系统标识，例如 LIKEI 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"` // 活动开关 Enabled bool `gorm:"column:enabled;index:idx_week_star_active_lookup,priority:2"` // 活动开关
StartAt time.Time `gorm:"column:start_at"` // 活动开始时间 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"` // 活动结束时间 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"` // 活动结算时区 Timezone string `gorm:"column:timezone;size:64"` // 活动结算时区
CreateTime time.Time `gorm:"column:create_time"` // 创建时间 CreateTime time.Time `gorm:"column:create_time"` // 创建时间
UpdateTime time.Time `gorm:"column:update_time"` // 更新时间 UpdateTime time.Time `gorm:"column:update_time"` // 更新时间
} }
// TableName 返回周榜活动主配置表名。 // TableName 返回周榜活动主配置表名。

241
internal/service/week_star_retry.go Normal file
View File

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

44
internal/service/week_star_settlement.go
View File

@ -82,8 +82,14 @@ func NewWeekStarSettlementService(cfg config.Config, repository *repo.Repository
} }
} }
// Start 启动定时扫描任务，按照配置周期持续检查是否存在待结算周榜。 // Start 启动周榜结算扫描和失败补发消费任务。
func (s *WeekStarSettlementService) Start(ctx context.Context) { 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 interval := s.cfg.WeekStarSettleInterval
if interval <= 0 { if interval <= 0 {
interval = time.Minute interval = time.Minute
@ -195,7 +201,7 @@ func (s *WeekStarSettlementService) settleCycle(ctx context.Context, configRow *
var rewardConfigs []model.WeekStarActivityRewardConfig var rewardConfigs []model.WeekStarActivityRewardConfig
if err := s.repo.DB.WithContext(ctx). if err := s.repo.DB.WithContext(ctx).
Where("config_id = ?", configRow.ID). Where("config_id = ?", configRow.ID).
Order("rank asc"). Order("`rank` asc").
Find(&rewardConfigs).Error; err != nil { Find(&rewardConfigs).Error; err != nil {
return err return err
} }
@ -282,14 +288,15 @@ func (s *WeekStarSettlementService) settleCycle(ctx context.Context, configRow *
var pendingRecords []model.WeekStarRewardRecord var pendingRecords []model.WeekStarRewardRecord
if err := s.repo.DB.WithContext(ctx). if err := s.repo.DB.WithContext(ctx).
Where("config_id = ? AND cycle_key = ? AND status = ?", configRow.ID, cycleKey, weekStarRewardStatusPending). Where("config_id = ? AND cycle_key = ? AND status = ?", configRow.ID, cycleKey, weekStarRewardStatusPending).
Order("rank asc"). Order("`rank` asc").
Find(&pendingRecords).Error; err != nil { Find(&pendingRecords).Error; err != nil {
return err return err
} }
var lastErr error var lastErr error
for index := range pendingRecords { 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 lastErr = err
} }
} }
@ -297,16 +304,22 @@ func (s *WeekStarSettlementService) settleCycle(ctx context.Context, configRow *
} }
// dispatchRewardRecord 调用 Java 奖励接口发奖，并回写发奖成功或失败状态。 // 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 { if record == nil {
return nil return true, nil
} }
var configRow model.WeekStarActivityConfig var configRow model.WeekStarActivityConfig
if err := s.repo.DB.WithContext(ctx).Where("id = ?", record.ConfigID).First(&configRow).Error; err != nil { if err := s.repo.DB.WithContext(ctx).Where("id = ?", record.ConfigID).First(&configRow).Error; err != nil {
return err return false, err
} }
now := time.Now() now := time.Now()
retryCount := record.RetryCount
if isRetry {
retryCount++
}
err := s.java.SendActivityReward(ctx, integration.SendActivityRewardRequest{ err := s.java.SendActivityReward(ctx, integration.SendActivityRewardRequest{
TrackID: record.TrackID, TrackID: record.TrackID,
Origin: weekStarRewardOrigin, Origin: weekStarRewardOrigin,
@ -318,7 +331,7 @@ func (s *WeekStarSettlementService) dispatchRewardRecord(ctx context.Context, re
// 发奖失败时保留错误信息，方便后台做失败补发。 // 发奖失败时保留错误信息，方便后台做失败补发。
updates := map[string]any{ updates := map[string]any{
"status": weekStarRewardStatusFailed, "status": weekStarRewardStatusFailed,
"retry_count": record.RetryCount, "retry_count": retryCount,
"last_error": truncateString(err.Error(), 1024), "last_error": truncateString(err.Error(), 1024),
"update_time": now, "update_time": now,
} }
@ -326,21 +339,24 @@ func (s *WeekStarSettlementService) dispatchRewardRecord(ctx context.Context, re
Model(&model.WeekStarRewardRecord{}). Model(&model.WeekStarRewardRecord{}).
Where("id = ?", record.ID). Where("id = ?", record.ID).
Updates(updates).Error; updateErr != nil { 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{}). Model(&model.WeekStarRewardRecord{}).
Where("id = ?", record.ID). Where("id = ?", record.ID).
Updates(map[string]any{ Updates(map[string]any{
"status": weekStarRewardStatusSuccess, "status": weekStarRewardStatusSuccess,
"retry_count": record.RetryCount, "retry_count": retryCount,
"last_error": "", "last_error": "",
"sent_at": now, "sent_at": now,
"update_time": now, "update_time": now,
}).Error }).Error; err != nil {
return false, err
}
return true, nil
} }
// loadLiveRanking 从 Redis 实时榜读取前 N 名，并补齐用户资料供快照/发奖使用。 // loadLiveRanking 从 Redis 实时榜读取前 N 名，并补齐用户资料供快照/发奖使用。
@ -464,7 +480,7 @@ func (s *WeekStarSettlementService) toStorageWallClock(t time.Time) time.Time {
local.Minute(), local.Minute(),
local.Second(), local.Second(),
local.Nanosecond(), local.Nanosecond(),
time.Local, s.location,
) )
} }