From a31131018072ba4415dff001e6c116cc666ec80c Mon Sep 17 00:00:00 2001 From: hy Date: Fri, 17 Apr 2026 19:04:32 +0800 Subject: [PATCH] Add week star cron worker --- .gitignore | 1 + README.md | 42 ++ cmd/week-star-job/main.go | 49 +++ go.mod | 19 + go.sum | 25 ++ internal/config/config.go | 77 ++++ internal/integration/java.go | 132 ++++++ internal/model/models.go | 96 ++++ internal/repo/repo.go | 42 ++ internal/service/week_star_settlement.go | 535 +++++++++++++++++++++++ internal/util/id.go | 23 + 11 files changed, 1041 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 cmd/week-star-job/main.go create mode 100644 go.mod create mode 100644 go.sum create mode 100644 internal/config/config.go create mode 100644 internal/integration/java.go create mode 100644 internal/model/models.go create mode 100644 internal/repo/repo.go create mode 100644 internal/service/week_star_settlement.go create mode 100644 internal/util/id.go diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2f7896d --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +target/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..7d24994 --- /dev/null +++ b/README.md @@ -0,0 +1,42 @@ +# chatapp-cron + +Standalone cron workers for chatapp activities. + +Current worker: +- `week-star-job`: specified-gift weekly rank settlement and reward dispatch + +Local run: + +```bash +cd /Users/hy/Documents/hy/chatapp-cron +go mod tidy +go run ./cmd/week-star-job +``` + +Production deploy: + +```bash +cd /Users/hy/Documents/hy/chatapp3/chatapp3-deploy +./deploy.sh -c chatapp-cron +``` + +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` + +Production topology: + +- worker host: Singapore `cron-sg-1` +- dependent MySQL / Redis / Java Other: Riyadh private network +- network requirement: VPCs must be connected through Tencent Cloud Connect Network before deploy + +Detailed ops guide: + +- `/Users/hy/Documents/hy/chatapp3/chatapp3-deploy/tencent-prod/docs/CHATAPP_CRON_DEPLOYMENT.md` diff --git a/cmd/week-star-job/main.go b/cmd/week-star-job/main.go new file mode 100644 index 0000000..c033a26 --- /dev/null +++ b/cmd/week-star-job/main.go @@ -0,0 +1,49 @@ +package main + +import ( + "chatapp-cron/internal/config" + "chatapp-cron/internal/integration" + "chatapp-cron/internal/repo" + "chatapp-cron/internal/service" + "context" + "log" + "os/signal" + "syscall" + "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(), 10*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) + + workerCtx, workerCancel := context.WithCancel(context.Background()) + defer workerCancel() + settlementService.Start(workerCtx) + log.Printf( + "week-star-job started. interval=%s timezone=%s javaOther=%s", + cfg.WeekStarSettleInterval, + cfg.WeekStarTimezone, + cfg.JavaOtherBaseURL, + ) + + shutdownSignal, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + <-shutdownSignal.Done() + + log.Printf("week-star-job shutdown signal received") + workerCancel() + time.Sleep(300 * time.Millisecond) +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..415060d --- /dev/null +++ b/go.mod @@ -0,0 +1,19 @@ +module chatapp-cron + +go 1.24 + +require ( + github.com/bwmarrin/snowflake v0.3.0 + github.com/redis/go-redis/v9 v9.7.0 + gorm.io/driver/mysql v1.5.7 + gorm.io/gorm v1.30.0 +) + +require ( + github.com/cespare/xxhash/v2 v2.2.0 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect + github.com/go-sql-driver/mysql v1.7.0 // indirect + github.com/jinzhu/inflection v1.0.0 // indirect + github.com/jinzhu/now v1.1.5 // indirect + golang.org/x/text v0.20.0 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..2d3f33c --- /dev/null +++ b/go.sum @@ -0,0 +1,25 @@ +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/bwmarrin/snowflake v0.3.0 h1:xm67bEhkKh6ij1790JB83OujPR5CzNe8QuQqAgISZN0= +github.com/bwmarrin/snowflake v0.3.0/go.mod h1:NdZxfVWX+oR6y2K0o6qAYv6gIOP9rjG0/E9WsDpxqwE= +github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44= +github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= +github.com/go-sql-driver/mysql v1.7.0 h1:ueSltNNllEqE3qcWBTD0iQd3IpL/6U+mJxLkazJ7YPc= +github.com/go-sql-driver/mysql v1.7.0/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI= +github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E= +github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc= +github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ= +github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= +github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E= +github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw= +golang.org/x/text v0.20.0 h1:gK/Kv2otX8gz+wn7Rmb3vT96ZwuoxnQlY+HlJVj7Qug= +golang.org/x/text v0.20.0/go.mod h1:D4IsuqiFMhST5bX19pQ9ikHC2GsaKyk/oF+pn3ducp4= +gorm.io/driver/mysql v1.5.7 h1:MndhOPYOfEp2rHKgkZIhJ16eVUIRf2HmzgoPmh7FCWo= +gorm.io/driver/mysql v1.5.7/go.mod h1:sEtPWMiqiN1N1cMXoXmBbd8C6/l+TESwriotuRRpkDM= +gorm.io/gorm v1.25.7/go.mod h1:hbnx/Oo0ChWMn1BIhpy1oYozzpM15i4YPuHDmfYtwg8= +gorm.io/gorm v1.30.0 h1:qbT5aPv1UH8gI99OsRlvDToLxW5zR7FzS9acZDOZcgs= +gorm.io/gorm v1.30.0/go.mod h1:8Z33v652h4//uMA76KjeDH8mJXPm1QNCYrMeatR0DOE= diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..7cda770 --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,77 @@ +package config + +import ( + "os" + "strconv" + "strings" + "time" +) + +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 +} + +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)), + } +} + +func firstEnv(keys ...string) string { + if len(keys) == 0 { + return "" + } + for _, key := range keys[:len(keys)-1] { + if strings.TrimSpace(key) == "" { + continue + } + if value := strings.TrimSpace(os.Getenv(key)); value != "" { + return value + } + } + return keys[len(keys)-1] +} + +func getEnvInt(key string, fallback int) int { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback + } + parsed, err := strconv.Atoi(value) + if err != nil { + return fallback + } + return parsed +} + +func getEnvBool(key string, fallback bool) bool { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback + } + parsed, err := strconv.ParseBool(value) + if err != nil { + return fallback + } + return parsed +} diff --git a/internal/integration/java.go b/internal/integration/java.go new file mode 100644 index 0000000..25bd516 --- /dev/null +++ b/internal/integration/java.go @@ -0,0 +1,132 @@ +package integration + +import ( + "bytes" + "chatapp-cron/internal/config" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + "strings" +) + +type Client struct { + cfg config.Config + httpClient *http.Client +} + +type SendActivityRewardRequest struct { + TrackID int64 `json:"trackId"` + Origin string `json:"origin"` + SysOrigin string `json:"sysOrigin"` + SourceGroupID int64 `json:"sourceGroupId"` + AcceptUserID int64 `json:"acceptUserId"` +} + +type UserProfile struct { + ID Int64Value `json:"id"` + UserAvatar string `json:"userAvatar"` + UserNickname string `json:"userNickname"` +} + +type resultResponse[T any] struct { + Code int `json:"code"` + Message string `json:"message"` + Body T `json:"body"` + Success *bool `json:"success"` +} + +type Int64Value int64 + +func (v *Int64Value) 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, "\"") + } + parsed, err := strconv.ParseInt(raw, 10, 64) + if err != nil { + return err + } + *v = Int64Value(parsed) + return nil +} + +func New(cfg config.Config) *Client { + return &Client{ + cfg: cfg, + httpClient: &http.Client{ + Timeout: cfg.Timeout, + }, + } +} + +func (c *Client) SendActivityReward(ctx context.Context, req SendActivityRewardRequest) error { + return c.postJSON(ctx, c.cfg.JavaOtherBaseURL+"/props-activity-cnf/client/sendActivityReward", req, nil) +} + +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] + if err := c.getJSON(ctx, endpoint, &resp); err != nil { + return UserProfile{}, err + } + if int64(resp.Body.ID) == 0 { + return UserProfile{}, fmt.Errorf("empty user profile response") + } + return resp.Body, nil +} + +func (c *Client) getJSON(ctx context.Context, endpoint string, out any) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, 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 (c *Client) postJSON(ctx context.Context, endpoint string, payload any, out any) error { + data, err := json.Marshal(payload) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(data)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + 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 { + return err + } + if resp.StatusCode >= http.StatusBadRequest { + return fmt.Errorf("java api failed: status=%d body=%s", resp.StatusCode, strings.TrimSpace(string(body))) + } + if out == nil || len(body) == 0 { + return nil + } + if err := json.Unmarshal(body, out); err != nil { + return fmt.Errorf("decode response failed: %w body=%s", err, strings.TrimSpace(string(body))) + } + return nil +} diff --git a/internal/model/models.go b/internal/model/models.go new file mode 100644 index 0000000..40dd776 --- /dev/null +++ b/internal/model/models.go @@ -0,0 +1,96 @@ +package model + +import "time" + +// UserBaseInfo 对应用户基础资料表，用于在结算和展示时补齐榜单用户信息。 +type UserBaseInfo struct { + ID int64 `gorm:"column:id;primaryKey"` // 用户 ID + Account string `gorm:"column:account;size:64"` // 用户账号 + UserAvatar string `gorm:"column:user_avatar;size:255"` // 头像地址 + UserNickname string `gorm:"column:user_nickname;size:255"` // 昵称 + OriginSys string `gorm:"column:origin_sys;size:32"` // 来源系统 + CountryCode string `gorm:"column:country_code;size:32"` // 国家编码 + CountryName string `gorm:"column:country_name;size:128"` // 国家名称 + CreateTime time.Time `gorm:"column:create_time"` // 创建时间 +} + +// TableName 返回用户基础资料表名。 +func (UserBaseInfo) TableName() string { return "user_base_info" } + +// 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"` // 更新时间 +} + +// TableName 返回周榜活动主配置表名。 +func (WeekStarActivityConfig) TableName() string { return "week_star_activity_config" } + +// WeekStarActivityRewardConfig 对应周榜奖励配置，定义某个名次绑定哪个奖励组。 +type WeekStarActivityRewardConfig struct { + ID int64 `gorm:"column:id;primaryKey"` // 奖励配置主键 + ConfigID int64 `gorm:"column:config_id;uniqueIndex:uk_week_star_reward_rank,priority:1"` // 关联活动配置 ID + Rank int `gorm:"column:rank;uniqueIndex:uk_week_star_reward_rank,priority:2"` // 排名名次 + RewardGroupID int64 `gorm:"column:reward_group_id"` // 奖励组 ID + RewardGroupName string `gorm:"column:reward_group_name;size:255"` // 奖励组名称 + CreateTime time.Time `gorm:"column:create_time"` // 创建时间 + UpdateTime time.Time `gorm:"column:update_time"` // 更新时间 +} + +// TableName 返回周榜奖励配置表名。 +func (WeekStarActivityRewardConfig) TableName() string { return "week_star_activity_reward_config" } + +// WeekStarRankSnapshot 是每周结算时冻结的排行榜快照，用于后台查看和发奖依据留痕。 +type WeekStarRankSnapshot struct { + ID int64 `gorm:"column:id;primaryKey"` // 快照主键 + ConfigID int64 `gorm:"column:config_id;uniqueIndex:uk_week_star_snapshot_rank,priority:1;index:idx_week_star_snapshot_cycle,priority:1"` // 活动配置 ID + CycleKey string `gorm:"column:cycle_key;size:32;uniqueIndex:uk_week_star_snapshot_rank,priority:2;index:idx_week_star_snapshot_cycle,priority:2"` // 周期标识 + Rank int `gorm:"column:rank;uniqueIndex:uk_week_star_snapshot_rank,priority:3"` // 结算名次 + PeriodStartAt time.Time `gorm:"column:period_start_at"` // 周期开始时间 + PeriodEndAt time.Time `gorm:"column:period_end_at"` // 周期结束时间 + UserID int64 `gorm:"column:user_id"` // 上榜用户 ID + Account string `gorm:"column:account;size:64"` // 上榜用户账号 + UserAvatar string `gorm:"column:user_avatar;size:1024"` // 上榜用户头像 + UserNickname string `gorm:"column:user_nickname;size:255"` // 上榜用户昵称 + CountryCode string `gorm:"column:country_code;size:32"` // 国家编码 + CountryName string `gorm:"column:country_name;size:128"` // 国家名称 + ScoreGold int64 `gorm:"column:score_gold"` // 结算金币分值 + CreateTime time.Time `gorm:"column:create_time"` // 创建时间 + UpdateTime time.Time `gorm:"column:update_time"` // 更新时间 +} + +// TableName 返回周榜快照表名。 +func (WeekStarRankSnapshot) TableName() string { return "week_star_rank_snapshot" } + +// WeekStarRewardRecord 记录每个周期每个奖励名次的发奖状态，便于重试和审计。 +type WeekStarRewardRecord struct { + ID int64 `gorm:"column:id;primaryKey"` // 发奖记录主键 + ConfigID int64 `gorm:"column:config_id;uniqueIndex:uk_week_star_reward_record_rank,priority:1;index:idx_week_star_reward_record_cycle,priority:1"` // 活动配置 ID + CycleKey string `gorm:"column:cycle_key;size:32;uniqueIndex:uk_week_star_reward_record_rank,priority:2;index:idx_week_star_reward_record_cycle,priority:2"` // 周期标识 + Rank int `gorm:"column:rank;uniqueIndex:uk_week_star_reward_record_rank,priority:3"` // 发奖名次 + UserID int64 `gorm:"column:user_id;index:idx_week_star_reward_record_user"` // 收奖用户 ID + Account string `gorm:"column:account;size:64"` // 收奖用户账号 + UserAvatar string `gorm:"column:user_avatar;size:1024"` // 收奖用户头像 + UserNickname string `gorm:"column:user_nickname;size:255"` // 收奖用户昵称 + CountryCode string `gorm:"column:country_code;size:32"` // 国家编码 + CountryName string `gorm:"column:country_name;size:128"` // 国家名称 + RewardGroupID int64 `gorm:"column:reward_group_id"` // 奖励组 ID + RewardGroupName string `gorm:"column:reward_group_name;size:255"` // 奖励组名称 + ScoreGold int64 `gorm:"column:score_gold"` // 该用户本周期最终积分 + TrackID int64 `gorm:"column:track_id"` // 发奖请求幂等追踪 ID + Status string `gorm:"column:status;size:32;index:idx_week_star_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"` // 更新时间 +} + +// TableName 返回周榜发奖记录表名。 +func (WeekStarRewardRecord) TableName() string { return "week_star_reward_record" } diff --git a/internal/repo/repo.go b/internal/repo/repo.go new file mode 100644 index 0000000..cd90431 --- /dev/null +++ b/internal/repo/repo.go @@ -0,0 +1,42 @@ +package repo + +import ( + "context" + + "chatapp-cron/internal/config" + + "github.com/redis/go-redis/v9" + "gorm.io/driver/mysql" + "gorm.io/gorm" +) + +type Repository struct { + DB *gorm.DB + Redis *redis.Client +} + +func New(cfg config.Config) (*Repository, error) { + db, err := gorm.Open(mysql.Open(cfg.MySQLDSN), &gorm.Config{}) + if err != nil { + return nil, err + } + + client := redis.NewClient(&redis.Options{ + Addr: cfg.RedisAddr, + Password: cfg.RedisPassword, + DB: cfg.RedisDB, + }) + + return &Repository{DB: db, Redis: client}, nil +} + +func (r *Repository) Ping(ctx context.Context) error { + if err := r.Redis.Ping(ctx).Err(); err != nil { + return err + } + sqlDB, err := r.DB.DB() + if err != nil { + return err + } + return sqlDB.PingContext(ctx) +} diff --git a/internal/service/week_star_settlement.go b/internal/service/week_star_settlement.go new file mode 100644 index 0000000..922e26f --- /dev/null +++ b/internal/service/week_star_settlement.go @@ -0,0 +1,535 @@ +package service + +import ( + "context" + "fmt" + "log" + "math" + "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" + "gorm.io/gorm/clause" +) + +const ( + // weekStarRewardStatusPending 表示已生成发奖记录，但尚未调用下游发奖接口。 + weekStarRewardStatusPending = "PENDING" + // weekStarRewardStatusSuccess 表示奖励已经成功发放。 + weekStarRewardStatusSuccess = "SUCCESS" + // weekStarRewardStatusFailed 表示调用发奖失败，等待后台重试。 + weekStarRewardStatusFailed = "FAILED" + // weekStarRewardOrigin 是发奖来源标识，透传给 Java 奖励系统用于审计。 + weekStarRewardOrigin = "ACTIVITY_REWARD" +) + +// WeekStarRankingUserView 是结算过程和查询过程共用的榜单展示视图。 +type WeekStarRankingUserView struct { + UserID int64 + UserAvatar string + UserNickname string + Account string + CountryCode string + CountryName string + Rank int + ScoreGold int64 +} + +// weekStarUserProfile 是内部组装用的轻量用户资料对象。 +type weekStarUserProfile struct { + UserID int64 + Account string + UserAvatar string + UserNickname string + CountryCode string + CountryName string +} + +// weekStarRankScore 保存 Redis 排行结果归一化后的分数和名次。 +type weekStarRankScore struct { + UserID int64 + ScoreGold int64 + Rank int +} + +// WeekStarSettlementService 负责扫描周期、冻结榜单快照并发放奖励。 +type WeekStarSettlementService struct { + cfg config.Config + repo *repo.Repository + java *integration.Client + location *time.Location +} + +// NewWeekStarSettlementService 创建周榜结算服务，并初始化活动使用的时区。 +func NewWeekStarSettlementService(cfg config.Config, repository *repo.Repository, javaClient *integration.Client) *WeekStarSettlementService { + location, err := time.LoadLocation(strings.TrimSpace(cfg.WeekStarTimezone)) + if err != nil { + // 配置异常时兜底为 UTC，避免服务启动失败。 + location = time.UTC + } + return &WeekStarSettlementService{ + cfg: cfg, + repo: repository, + java: javaClient, + location: location, + } +} + +// Start 启动定时扫描任务，按照配置周期持续检查是否存在待结算周榜。 +func (s *WeekStarSettlementService) Start(ctx context.Context) { + interval := s.cfg.WeekStarSettleInterval + 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("week star settlement scan failed: %v", err) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } + }() +} + +// SettleDueCycles 扫描所有活动配置，并尝试结算已结束但尚未冻结的周期。 +func (s *WeekStarSettlementService) SettleDueCycles(ctx context.Context) error { + var configs []model.WeekStarActivityConfig + if err := s.repo.DB.WithContext(ctx).Order("id desc").Find(&configs).Error; err != nil { + return err + } + now := time.Now().In(s.location) + var lastErr error + for index := range configs { + if err := s.settleConfigCycles(ctx, &configs[index], now); err != nil { + lastErr = err + log.Printf("week star settle config failed. configId=%d err=%v", configs[index].ID, err) + } + } + return lastErr +} + +// settleConfigCycles 逐周遍历单个活动配置的时间窗，挑出所有已经结束的周期执行结算。 +func (s *WeekStarSettlementService) settleConfigCycles(ctx context.Context, configRow *model.WeekStarActivityConfig, now time.Time) error { + if configRow == nil { + return nil + } + // 周榜按周一 00:00 切周，因此先把活动开始时间对齐到所在周的起点。 + startCycle := s.cycleStart(s.fromStorageWallClock(configRow.StartAt)) + endAt := s.fromStorageWallClock(configRow.EndAt) + var lastErr error + for cycleStart := startCycle; cycleStart.Before(endAt); cycleStart = cycleStart.AddDate(0, 0, 7) { + cycleEnd := cycleStart.AddDate(0, 0, 7) + periodStart := maxTime(cycleStart, s.fromStorageWallClock(configRow.StartAt)) + periodEnd := minTime(cycleEnd, endAt) + if !periodEnd.After(periodStart) { + continue + } + if periodEnd.After(now) { + break + } + if err := s.settleCycle(ctx, configRow, cycleStart, periodStart, periodEnd); err != nil { + lastErr = err + } + } + return lastErr +} + +// settleCycle 结算单个周期，流程包含加锁、读取实时榜、冻结快照、生成发奖记录并实际发奖。 +func (s *WeekStarSettlementService) settleCycle(ctx context.Context, configRow *model.WeekStarActivityConfig, cycleStart, periodStart, periodEnd time.Time) error { + cycleKey := s.cycleKey(cycleStart) + lockKey := s.settleLockKey(configRow.ID, cycleKey) + + // 使用 Redis 分布式锁保证同一周期只会被一个实例执行结算。 + acquired, err := s.repo.Redis.SetNX(ctx, lockKey, 1, 5*time.Minute).Result() + if err != nil { + return err + } + if !acquired { + return nil + } + + // 只要快照或发奖记录已存在，就视为该周期已经执行过结算。 + var snapshotCount int64 + if err := s.repo.DB.WithContext(ctx). + Model(&model.WeekStarRankSnapshot{}). + Where("config_id = ? AND cycle_key = ?", configRow.ID, cycleKey). + Count(&snapshotCount).Error; err != nil { + return err + } + var rewardCount int64 + if err := s.repo.DB.WithContext(ctx). + Model(&model.WeekStarRewardRecord{}). + Where("config_id = ? AND cycle_key = ?", configRow.ID, cycleKey). + Count(&rewardCount).Error; err != nil { + return err + } + if snapshotCount > 0 || rewardCount > 0 { + return nil + } + + // 结算只冻结 Top3，用于历史榜单和后续奖励发放。 + liveRanking, err := s.loadLiveRanking(ctx, configRow.ID, cycleKey, 3) + if err != nil { + return err + } + if len(liveRanking) == 0 { + return nil + } + + var rewardConfigs []model.WeekStarActivityRewardConfig + if err := s.repo.DB.WithContext(ctx). + Where("config_id = ?", configRow.ID). + Order("rank asc"). + Find(&rewardConfigs).Error; err != nil { + return err + } + rewardConfigMap := make(map[int]model.WeekStarActivityRewardConfig, len(rewardConfigs)) + for _, row := range rewardConfigs { + rewardConfigMap[row.Rank] = row + } + + // 先构造要落库的快照和发奖记录，避免边查边写导致状态不一致。 + now := time.Now() + snapshots := make([]model.WeekStarRankSnapshot, 0, len(liveRanking)) + rewardRecords := make([]model.WeekStarRewardRecord, 0, len(liveRanking)) + for _, item := range liveRanking { + snapshotID, err := util.NextID() + if err != nil { + return err + } + snapshots = append(snapshots, model.WeekStarRankSnapshot{ + ID: snapshotID, + ConfigID: configRow.ID, + CycleKey: cycleKey, + Rank: item.Rank, + PeriodStartAt: s.toStorageWallClock(periodStart), + PeriodEndAt: s.toStorageWallClock(periodEnd), + UserID: item.UserID, + Account: item.Account, + UserAvatar: item.UserAvatar, + UserNickname: item.UserNickname, + CountryCode: item.CountryCode, + CountryName: item.CountryName, + ScoreGold: item.ScoreGold, + CreateTime: now, + UpdateTime: now, + }) + + // 某个名次未配置奖励时，只保留快照，不生成发奖记录。 + rewardConfig, ok := rewardConfigMap[item.Rank] + if !ok { + continue + } + recordID, err := util.NextID() + if err != nil { + return err + } + rewardRecords = append(rewardRecords, model.WeekStarRewardRecord{ + ID: recordID, + ConfigID: configRow.ID, + CycleKey: cycleKey, + Rank: item.Rank, + UserID: item.UserID, + Account: item.Account, + UserAvatar: item.UserAvatar, + UserNickname: item.UserNickname, + CountryCode: item.CountryCode, + CountryName: item.CountryName, + RewardGroupID: rewardConfig.RewardGroupID, + RewardGroupName: rewardConfig.RewardGroupName, + ScoreGold: item.ScoreGold, + TrackID: recordID, + Status: weekStarRewardStatusPending, + RetryCount: 0, + CreateTime: now, + UpdateTime: now, + }) + } + + // 快照和发奖记录在同一个事务里落库，确保结算结果成对出现。 + if err := s.repo.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if len(snapshots) > 0 { + if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&snapshots).Error; err != nil { + return err + } + } + if len(rewardRecords) > 0 { + if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&rewardRecords).Error; err != nil { + return err + } + } + return nil + }); err != nil { + return err + } + + 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"). + Find(&pendingRecords).Error; err != nil { + return err + } + var lastErr error + for index := range pendingRecords { + // 发奖逐条执行，单条失败不会阻塞同周期其他名次继续发放。 + if err := s.dispatchRewardRecord(ctx, &pendingRecords[index]); err != nil { + lastErr = err + } + } + return lastErr +} + +// dispatchRewardRecord 调用 Java 奖励接口发奖，并回写发奖成功或失败状态。 +func (s *WeekStarSettlementService) dispatchRewardRecord(ctx context.Context, record *model.WeekStarRewardRecord) error { + if record == nil { + return nil + } + var configRow model.WeekStarActivityConfig + if err := s.repo.DB.WithContext(ctx).Where("id = ?", record.ConfigID).First(&configRow).Error; err != nil { + return err + } + + now := time.Now() + err := s.java.SendActivityReward(ctx, integration.SendActivityRewardRequest{ + TrackID: record.TrackID, + Origin: weekStarRewardOrigin, + SysOrigin: configRow.SysOrigin, + SourceGroupID: record.RewardGroupID, + AcceptUserID: record.UserID, + }) + if err != nil { + // 发奖失败时保留错误信息，方便后台做失败补发。 + updates := map[string]any{ + "status": weekStarRewardStatusFailed, + "retry_count": record.RetryCount, + "last_error": truncateString(err.Error(), 1024), + "update_time": now, + } + if updateErr := s.repo.DB.WithContext(context.Background()). + Model(&model.WeekStarRewardRecord{}). + Where("id = ?", record.ID). + Updates(updates).Error; updateErr != nil { + return updateErr + } + return err + } + + return s.repo.DB.WithContext(ctx). + Model(&model.WeekStarRewardRecord{}). + Where("id = ?", record.ID). + Updates(map[string]any{ + "status": weekStarRewardStatusSuccess, + "retry_count": record.RetryCount, + "last_error": "", + "sent_at": now, + "update_time": now, + }).Error +} + +// loadLiveRanking 从 Redis 实时榜读取前 N 名，并补齐用户资料供快照/发奖使用。 +func (s *WeekStarSettlementService) loadLiveRanking(ctx context.Context, configID int64, cycleKey string, limit int64) ([]WeekStarRankingUserView, error) { + if configID <= 0 || strings.TrimSpace(cycleKey) == "" || limit <= 0 { + return []WeekStarRankingUserView{}, nil + } + + redisKey := s.rankRedisKey(configID, cycleKey) + rankScores, err := s.repo.Redis.ZRevRangeWithScores(ctx, redisKey, 0, limit-1).Result() + if err != nil { + return nil, err + } + if len(rankScores) == 0 { + return []WeekStarRankingUserView{}, nil + } + + userIDs := make([]int64, 0, len(rankScores)) + normalizedScores := make([]weekStarRankScore, 0, len(rankScores)) + for index, item := range rankScores { + userID, ok := parseRedisMemberInt64(item.Member) + if !ok || userID <= 0 { + // 跳过异常 member，避免脏数据影响正常结算。 + continue + } + userIDs = append(userIDs, userID) + normalizedScores = append(normalizedScores, weekStarRankScore{ + UserID: userID, + ScoreGold: int64(math.Round(item.Score)), + Rank: index + 1, + }) + } + + profileMap, err := s.lookupUserProfiles(ctx, userIDs) + if err != nil { + return nil, err + } + + result := make([]WeekStarRankingUserView, 0, len(normalizedScores)) + for _, item := range normalizedScores { + profile := profileMap[item.UserID] + result = append(result, WeekStarRankingUserView{ + UserID: item.UserID, + UserAvatar: profile.UserAvatar, + UserNickname: profile.UserNickname, + Account: profile.Account, + CountryCode: profile.CountryCode, + CountryName: profile.CountryName, + Rank: item.Rank, + ScoreGold: item.ScoreGold, + }) + } + return result, nil +} + +// lookupUserProfiles 优先查本地用户表，查不到再回源 Java 用户资料接口兜底。 +func (s *WeekStarSettlementService) lookupUserProfiles(ctx context.Context, userIDs []int64) (map[int64]weekStarUserProfile, error) { + result := make(map[int64]weekStarUserProfile, len(userIDs)) + if len(userIDs) == 0 { + return result, nil + } + + var rows []model.UserBaseInfo + if err := s.repo.DB.WithContext(ctx).Where("id IN ?", userIDs).Find(&rows).Error; err != nil { + return nil, err + } + for _, row := range rows { + result[row.ID] = weekStarUserProfile{ + UserID: row.ID, + Account: row.Account, + UserAvatar: row.UserAvatar, + UserNickname: row.UserNickname, + CountryCode: row.CountryCode, + CountryName: row.CountryName, + } + } + + for _, userID := range userIDs { + if _, exists := result[userID]; exists { + continue + } + profile, err := s.java.GetUserProfile(ctx, userID) + if err != nil { + // 用户资料回源失败时仍保留空壳对象，避免整个结算流程中断。 + result[userID] = weekStarUserProfile{UserID: userID} + continue + } + result[userID] = weekStarUserProfile{ + UserID: userID, + UserAvatar: profile.UserAvatar, + UserNickname: profile.UserNickname, + } + } + return result, nil +} + +// cycleStart 计算某个时间所在周的周一 00:00，作为周榜周期起点。 +func (s *WeekStarSettlementService) cycleStart(t time.Time) time.Time { + local := t.In(s.location) + weekday := int(local.Weekday()) + if weekday == 0 { + weekday = 7 + } + startDay := local.AddDate(0, 0, -(weekday - 1)) + return time.Date(startDay.Year(), startDay.Month(), startDay.Day(), 0, 0, 0, 0, s.location) +} + +// cycleKey 把周期开始时间格式化为稳定的周期标识，例如 20260414。 +func (s *WeekStarSettlementService) cycleKey(start time.Time) string { + return start.In(s.location).Format("20060102") +} + +// toStorageWallClock 把活动时区时间转换为本地库里约定的墙上时间存储格式。 +func (s *WeekStarSettlementService) toStorageWallClock(t time.Time) time.Time { + local := t.In(s.location) + return time.Date( + local.Year(), + local.Month(), + local.Day(), + local.Hour(), + local.Minute(), + local.Second(), + local.Nanosecond(), + time.Local, + ) +} + +// fromStorageWallClock 把库里保存的墙上时间恢复为活动时区时间，便于业务计算。 +func (s *WeekStarSettlementService) fromStorageWallClock(t time.Time) time.Time { + return time.Date( + t.Year(), + t.Month(), + t.Day(), + t.Hour(), + t.Minute(), + t.Second(), + t.Nanosecond(), + s.location, + ) +} + +// rankRedisKey 返回某个活动某个周期的实时排行 Redis Key。 +func (s *WeekStarSettlementService) rankRedisKey(configID int64, cycleKey string) string { + return fmt.Sprintf("week-star:rank:%d:%s", configID, cycleKey) +} + +// settleLockKey 返回某个活动某个周期的结算锁 Redis Key。 +func (s *WeekStarSettlementService) settleLockKey(configID int64, cycleKey string) string { + return fmt.Sprintf("week-star:settle-lock:%d:%s", configID, cycleKey) +} + +// parseRedisMemberInt64 兼容多种 Redis member 类型，并统一解析为用户 ID。 +func parseRedisMemberInt64(member any) (int64, bool) { + switch value := member.(type) { + case string: + parsed, err := strconv.ParseInt(strings.TrimSpace(value), 10, 64) + return parsed, err == nil + case []byte: + parsed, err := strconv.ParseInt(strings.TrimSpace(string(value)), 10, 64) + return parsed, err == nil + case int64: + return value, true + case int: + return int64(value), true + default: + return 0, false + } +} + +// truncateString 按指定长度截断错误信息，避免超出数据库字段限制。 +func truncateString(value string, limit int) string { + if limit <= 0 || len(value) <= limit { + return value + } + return value[:limit] +} + +// maxTime 返回两个时间里的较大值。 +func maxTime(left, right time.Time) time.Time { + if left.After(right) { + return left + } + return right +} + +// minTime 返回两个时间里的较小值。 +func minTime(left, right time.Time) time.Time { + if left.Before(right) { + return left + } + return right +} diff --git a/internal/util/id.go b/internal/util/id.go new file mode 100644 index 0000000..7af10d8 --- /dev/null +++ b/internal/util/id.go @@ -0,0 +1,23 @@ +package util + +import ( + "sync" + + "github.com/bwmarrin/snowflake" +) + +var ( + nodeOnce sync.Once + node *snowflake.Node + nodeErr error +) + +func NextID() (int64, error) { + nodeOnce.Do(func() { + node, nodeErr = snowflake.NewNode(1) + }) + if nodeErr != nil { + return 0, nodeErr + } + return node.Generate().Int64(), nil +}