Add week star cron worker
This commit is contained in:
commit
a311310180
1
.gitignore
vendored
Normal file
1
.gitignore
vendored
Normal file
@ -0,0 +1 @@
|
|||||||
|
target/
|
||||||
42
README.md
Normal file
42
README.md
Normal file
@ -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`
|
||||||
49
cmd/week-star-job/main.go
Normal file
49
cmd/week-star-job/main.go
Normal file
@ -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)
|
||||||
|
}
|
||||||
19
go.mod
Normal file
19
go.mod
Normal file
@ -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
|
||||||
|
)
|
||||||
25
go.sum
Normal file
25
go.sum
Normal file
@ -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=
|
||||||
77
internal/config/config.go
Normal file
77
internal/config/config.go
Normal file
@ -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
|
||||||
|
}
|
||||||
132
internal/integration/java.go
Normal file
132
internal/integration/java.go
Normal file
@ -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
|
||||||
|
}
|
||||||
96
internal/model/models.go
Normal file
96
internal/model/models.go
Normal file
@ -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" }
|
||||||
42
internal/repo/repo.go
Normal file
42
internal/repo/repo.go
Normal file
@ -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)
|
||||||
|
}
|
||||||
535
internal/service/week_star_settlement.go
Normal file
535
internal/service/week_star_settlement.go
Normal file
@ -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
|
||||||
|
}
|
||||||
23
internal/util/id.go
Normal file
23
internal/util/id.go
Normal file
@ -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
|
||||||
|
}
|
||||||
Loading…
x
Reference in New Issue
Block a user