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

Compare commits

base: hy:49d788be18935ba95351e038ee4558b29e92d892
Branches Tags
hy:main
hy:test
..
compare: hy:2fb3a1e5f5c8ecd0991d2f36b7b7c038fad9c504
Branches Tags
hy:main
hy:test

No commits in common. "49d788be18935ba95351e038ee4558b29e92d892" and "2fb3a1e5f5c8ecd0991d2f36b7b7c038fad9c504" have entirely different histories.
49d788be18 ... 2fb3a1e5f5

12 changed files with 21 additions and 868 deletions
Show Stats Expand all files Collapse all files

8
.dockerignore
View File

@ -1,8 +0,0 @@
.git
.gitignore
target
tmp
bin
dist
coverage
**/.DS_Store

22
Dockerfile
View File

@ -1,22 +0,0 @@
FROM golang:1.24 AS builder
ARG APP_VERSION="dev"
ENV CGO_ENABLED=0 GOOS=linux GOARCH=amd64
WORKDIR /src
COPY . .
RUN go build -trimpath -ldflags="-s -w" -o /out/week-star-job ./cmd/week-star-job
FROM alpine:3.20
ARG APP_VERSION="dev"
RUN apk add --no-cache ca-certificates wget
WORKDIR /application
COPY --from=builder /out/week-star-job /application/service
LABEL org.opencontainers.image.version="${APP_VERSION}"
RUN chmod +x /application/service
ENTRYPOINT ["/application/service"]

117
Jenkinsfile vendored
View File

@ -1,117 +0,0 @@
import groovy.json.JsonOutput
String shellQuote(String value) {
return "'" + value.replace("'", "'\"'\"'") + "'"
}
pipeline {
agent any
options {
timestamps()
disableConcurrentBuilds()
buildDiscarder(logRotator(numToKeepStr: '30'))
}
parameters {
string(name: 'GIT_REF', defaultValue: 'main', description: 'Git branch, tag, or commit to build.')
string(name: 'IMAGE_TAG', defaultValue: '', description: 'Optional Harbor image tag override. Defaults to <branch>-<commit12>-<buildNumber>.')
string(name: 'HARBOR_REGISTRY', defaultValue: '10.2.1.3:18082', description: 'Harbor registry address reachable from the Jenkins agent.')
string(name: 'HARBOR_PROJECT', defaultValue: 'likei', description: 'Harbor project name.')
string(name: 'HARBOR_CREDENTIALS_ID', defaultValue: 'likei-harbor-robot', description: 'Jenkins username/password credentials id for Harbor.')
string(name: 'DOCKER_PLATFORM', defaultValue: 'linux/amd64', description: 'Docker build platform.')
}
environment {
DOCKER_BUILDKIT = '1'
}
stages {
stage('Checkout') {
steps {
checkout scm
script {
if (params.GIT_REF?.trim()) {
sh """#!/usr/bin/env bash
set -euo pipefail
git fetch --all --tags --prune
git checkout --force ${shellQuote(params.GIT_REF.trim())}
"""
}
}
sh 'mkdir -p ci-out'
}
}
stage('Resolve Inputs') {
steps {
script {
def branch = (env.BRANCH_NAME ?: env.GIT_BRANCH ?: 'manual').replaceAll('[^0-9A-Za-z._-]+', '-')
env.RESOLVED_GIT_COMMIT = sh(returnStdout: true, script: 'git rev-parse HEAD').trim()
env.RESOLVED_GIT_COMMIT_SHORT = sh(returnStdout: true, script: 'git rev-parse --short=12 HEAD').trim()
env.RESOLVED_IMAGE_TAG = params.IMAGE_TAG?.trim() ? params.IMAGE_TAG.trim() : "${branch}-${env.RESOLVED_GIT_COMMIT_SHORT}-${env.BUILD_NUMBER}"
writeFile(
file: 'ci-out/request.json',
text: JsonOutput.prettyPrint(JsonOutput.toJson([
repo: 'chatapp-cron',
gitRef: params.GIT_REF?.trim() ?: 'main',
commit: env.RESOLVED_GIT_COMMIT,
imageTag: env.RESOLVED_IMAGE_TAG,
])) + '\n',
)
}
}
}
stage('Login Harbor') {
steps {
withCredentials([usernamePassword(credentialsId: params.HARBOR_CREDENTIALS_ID, usernameVariable: 'HARBOR_USER', passwordVariable: 'HARBOR_PASS')]) {
sh '''#!/usr/bin/env bash
set -euo pipefail
printf '%s' "$HARBOR_PASS" | docker login "$HARBOR_REGISTRY" -u "$HARBOR_USER" --password-stdin
'''
}
}
}
stage('Build And Push') {
steps {
script {
def imageRef = "${params.HARBOR_REGISTRY}/${params.HARBOR_PROJECT}/week-star-job:${env.RESOLVED_IMAGE_TAG}"
sh """#!/usr/bin/env bash
set -euo pipefail
DOCKER_PLATFORM='${params.DOCKER_PLATFORM}' ./ci/build-image.sh '${imageRef}' '${env.RESOLVED_GIT_COMMIT_SHORT}'
docker push '${imageRef}'
"""
def digest = sh(returnStdout: true, script: "docker image inspect '${imageRef}' --format '{{index .RepoDigests 0}}'").trim()
def serviceImages = ['week-star-job': digest]
writeFile(file: 'ci-out/service-images.json', text: JsonOutput.prettyPrint(JsonOutput.toJson(serviceImages)) + '\n')
writeFile(
file: 'ci-out/deploy-params.json',
text: JsonOutput.prettyPrint(JsonOutput.toJson([
components: ['chatapp-cron'],
ref: env.RESOLVED_GIT_COMMIT,
serviceImages: serviceImages,
])) + '\n',
)
writeFile(
file: 'ci-out/build-metadata.json',
text: JsonOutput.prettyPrint(JsonOutput.toJson([
repo: 'chatapp-cron',
gitRef: params.GIT_REF?.trim() ?: 'main',
commit: env.RESOLVED_GIT_COMMIT,
imageTag: env.RESOLVED_IMAGE_TAG,
serviceImages: serviceImages,
])) + '\n',
)
}
}
}
}
post {
always {
archiveArtifacts artifacts: 'ci-out/*.json', fingerprint: true, allowEmptyArchive: false
}
}
}

5
README.md
View File

@ -4,7 +4,6 @@ Standalone cron workers for chatapp activities.
Current worker:
- `week-star-job`: specified-gift weekly rank settlement, reward dispatch, and failed reward retry
- `week-star-job`: room turnover weekly reward settlement and backfill dispatch
Local run:
@ -28,13 +27,9 @@ Required envs:
- `CHATAPP_STORE_REDIS_PASSWORD`
- `CHATAPP_STORE_REDIS_DB`
- `CHATAPP_JAVA_OTHER_BASE_URL`
- `CHATAPP_JAVA_WALLET_BASE_URL`
- `CHATAPP_WEEK_STAR_DEFAULT_SYS_ORIGIN`
- `CHATAPP_WEEK_STAR_TIMEZONE`
- `CHATAPP_WEEK_STAR_SETTLE_INTERVAL_SECONDS`
- `CHATAPP_ROOM_TURNOVER_REWARD_DEFAULT_SYS_ORIGIN`
- `CHATAPP_ROOM_TURNOVER_REWARD_TIMEZONE`
- `CHATAPP_ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS`
- `CHATAPP_WEEK_STAR_RETRY_STREAM_KEY`
- `CHATAPP_WEEK_STAR_RETRY_CONSUMER_GROUP`
- `CHATAPP_WEEK_STAR_RETRY_CONSUMER_NAME`

15
ci/build-image.sh
View File

@ -1,15 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
ROOT_DIR="$(CDPATH= cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd)"
IMAGE_REF="${1:-}"
APP_VERSION="${APP_VERSION:-${2:-dev}}"
DOCKER_PLATFORM="${DOCKER_PLATFORM:-linux/amd64}"
if [ -z "${IMAGE_REF}" ]; then
echo "usage: ci/build-image.sh <image-ref> [app-version]" >&2
exit 1
fi
cd "${ROOT_DIR}"
docker build --platform "${DOCKER_PLATFORM}" --build-arg "APP_VERSION=${APP_VERSION}" -t "${IMAGE_REF}" .

6
cmd/week-star-job/main.go
View File

@ -28,19 +28,15 @@ func main() {
javaClient := integration.New(cfg)
settlementService := service.NewWeekStarSettlementService(cfg, repository, javaClient)
roomTurnoverRewardService := service.NewRoomTurnoverRewardSettlementService(cfg, repository, javaClient)
workerCtx, workerCancel := context.WithCancel(context.Background())
defer workerCancel()
settlementService.Start(workerCtx)
roomTurnoverRewardService.Start(workerCtx)
log.Printf(
"week-star-job started. weekStarInterval=%s roomTurnoverInterval=%s timezone=%s javaOther=%s javaWallet=%s",
"week-star-job started. interval=%s timezone=%s javaOther=%s",
cfg.WeekStarSettleInterval,
cfg.RoomTurnoverSettleInterval,
cfg.WeekStarTimezone,
cfg.JavaOtherBaseURL,
cfg.JavaWalletBaseURL,
)
shutdownSignal, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)

6
cmd/week-star-settle-once/main.go
View File

@ -26,13 +26,9 @@ func main() {
javaClient := integration.New(cfg)
settlementService := service.NewWeekStarSettlementService(cfg, repository, javaClient)
roomTurnoverRewardService := service.NewRoomTurnoverRewardSettlementService(cfg, repository, javaClient)
if err := settlementService.SettleDueCycles(ctx); err != nil {
log.Fatalf("settle due cycles failed: %v", err)
}
if err := roomTurnoverRewardService.SettleDueCycles(ctx); err != nil {
log.Fatalf("settle room turnover reward due cycles failed: %v", err)
}
log.Printf("week-star and room-turnover-reward settle once finished")
log.Printf("week-star settle once finished")
}

20
internal/config/config.go
View File

@ -13,13 +13,9 @@ type Config struct {
RedisPassword string
RedisDB int
JavaOtherBaseURL string
JavaWalletBaseURL string
WeekStarDefaultSysOrigin string
WeekStarTimezone string
WeekStarSettleInterval time.Duration
RoomTurnoverDefaultSysOrigin string
RoomTurnoverTimezone string
RoomTurnoverSettleInterval time.Duration
WeekStarRetryStreamKey string
WeekStarRetryConsumerGroup string
WeekStarRetryConsumerName string
@ -53,27 +49,11 @@ func Load() Config {
[]string{"CHATAPP_JAVA_OTHER_BASE_URL", "CRON_JAVA_OTHER_BASE_URL", "GAME_JAVA_OTHER_BASE_URL", "INVITE_JAVA_APP_BASE_URL"},
"http://127.0.0.1:2400",
),
JavaWalletBaseURL: getEnvAny(
[]string{"CHATAPP_JAVA_WALLET_BASE_URL", "CRON_JAVA_WALLET_BASE_URL", "GAME_JAVA_WALLET_BASE_URL"},
"http://127.0.0.1:2000",
),
WeekStarDefaultSysOrigin: strings.ToUpper(
getEnvAny([]string{"CHATAPP_WEEK_STAR_DEFAULT_SYS_ORIGIN", "WEEK_STAR_DEFAULT_SYS_ORIGIN"}, "LIKEI"),
),
WeekStarTimezone: getEnvAny([]string{"CHATAPP_WEEK_STAR_TIMEZONE", "WEEK_STAR_TIMEZONE"}, "Asia/Riyadh"),
WeekStarSettleInterval: time.Duration(getEnvIntAny([]string{"CHATAPP_WEEK_STAR_SETTLE_INTERVAL_SECONDS", "WEEK_STAR_SETTLE_INTERVAL_SECONDS"}, 60)) * time.Second,
RoomTurnoverDefaultSysOrigin: strings.ToUpper(getEnvAny(
[]string{"CHATAPP_ROOM_TURNOVER_REWARD_DEFAULT_SYS_ORIGIN", "ROOM_TURNOVER_REWARD_DEFAULT_SYS_ORIGIN", "CHATAPP_WEEK_STAR_DEFAULT_SYS_ORIGIN", "WEEK_STAR_DEFAULT_SYS_ORIGIN"},
"LIKEI",
)),
RoomTurnoverTimezone: getEnvAny(
[]string{"CHATAPP_ROOM_TURNOVER_REWARD_TIMEZONE", "ROOM_TURNOVER_REWARD_TIMEZONE", "CHATAPP_WEEK_STAR_TIMEZONE", "WEEK_STAR_TIMEZONE"},
"Asia/Riyadh",
),
RoomTurnoverSettleInterval: time.Duration(getEnvIntAny(
[]string{"CHATAPP_ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS", "ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS"},
60,
)) * time.Second,
WeekStarRetryStreamKey: getEnvAny([]string{"CHATAPP_WEEK_STAR_RETRY_STREAM_KEY", "WEEK_STAR_RETRY_STREAM_KEY"}, "week-star:reward-retry"),
WeekStarRetryConsumerGroup: getEnvAny([]string{"CHATAPP_WEEK_STAR_RETRY_CONSUMER_GROUP", "WEEK_STAR_RETRY_CONSUMER_GROUP"}, "week-star-retry"),
WeekStarRetryConsumerName: getEnvAny([]string{"CHATAPP_WEEK_STAR_RETRY_CONSUMER_NAME", "WEEK_STAR_RETRY_CONSUMER_NAME"}, defaultConsumerName("week-star-retry")),

20
internal/config/config_test.go
View File

@ -12,13 +12,9 @@ func TestLoadPrefersChatappKeys(t *testing.T) {
t.Setenv("CHATAPP_STORE_REDIS_PASSWORD", "chatapp-password")
t.Setenv("CHATAPP_STORE_REDIS_DB", "5")
t.Setenv("CHATAPP_JAVA_OTHER_BASE_URL", "http://chatapp-java")
t.Setenv("CHATAPP_JAVA_WALLET_BASE_URL", "http://chatapp-wallet")
t.Setenv("CHATAPP_WEEK_STAR_DEFAULT_SYS_ORIGIN", "likei")
t.Setenv("CHATAPP_WEEK_STAR_TIMEZONE", "Asia/Shanghai")
t.Setenv("CHATAPP_WEEK_STAR_SETTLE_INTERVAL_SECONDS", "120")
t.Setenv("CHATAPP_ROOM_TURNOVER_REWARD_DEFAULT_SYS_ORIGIN", "likei")
t.Setenv("CHATAPP_ROOM_TURNOVER_REWARD_TIMEZONE", "Asia/Riyadh")
t.Setenv("CHATAPP_ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS", "180")
t.Setenv("CHATAPP_WEEK_STAR_RETRY_STREAM_KEY", "chatapp-stream")
t.Setenv("CHATAPP_WEEK_STAR_RETRY_CONSUMER_GROUP", "chatapp-group")
t.Setenv("CHATAPP_WEEK_STAR_RETRY_CONSUMER_NAME", "chatapp-consumer")
@ -40,9 +36,6 @@ func TestLoadPrefersChatappKeys(t *testing.T) {
if cfg.JavaOtherBaseURL != "http://chatapp-java" {
t.Fatalf("expected chatapp java base url, got %q", cfg.JavaOtherBaseURL)
}
if cfg.JavaWalletBaseURL != "http://chatapp-wallet" {
t.Fatalf("expected chatapp java wallet base url, got %q", cfg.JavaWalletBaseURL)
}
if cfg.WeekStarDefaultSysOrigin != "LIKEI" {
t.Fatalf("expected normalized sys origin LIKEI, got %q", cfg.WeekStarDefaultSysOrigin)
}
@ -52,9 +45,6 @@ func TestLoadPrefersChatappKeys(t *testing.T) {
if cfg.WeekStarSettleInterval != 120*time.Second {
t.Fatalf("expected settle interval 120s, got %s", cfg.WeekStarSettleInterval)
}
if cfg.RoomTurnoverDefaultSysOrigin != "LIKEI" || cfg.RoomTurnoverTimezone != "Asia/Riyadh" || cfg.RoomTurnoverSettleInterval != 180*time.Second {
t.Fatalf("unexpected room turnover config: %+v", cfg)
}
if cfg.WeekStarRetryStreamKey != "chatapp-stream" || cfg.WeekStarRetryConsumerGroup != "chatapp-group" || cfg.WeekStarRetryConsumerName != "chatapp-consumer" {
t.Fatalf("unexpected retry consumer config: %+v", cfg)
}
@ -75,13 +65,9 @@ func TestLoadFallsBackToLegacyKeys(t *testing.T) {
t.Setenv("CRON_REDIS_PASSWORD", "legacy-password")
t.Setenv("CRON_REDIS_DB", "3")
t.Setenv("CRON_JAVA_OTHER_BASE_URL", "http://legacy-java")
t.Setenv("CRON_JAVA_WALLET_BASE_URL", "http://legacy-wallet")
t.Setenv("WEEK_STAR_DEFAULT_SYS_ORIGIN", "tarab")
t.Setenv("WEEK_STAR_TIMEZONE", "Asia/Riyadh")
t.Setenv("WEEK_STAR_SETTLE_INTERVAL_SECONDS", "66")
t.Setenv("ROOM_TURNOVER_REWARD_DEFAULT_SYS_ORIGIN", "tarab")
t.Setenv("ROOM_TURNOVER_REWARD_TIMEZONE", "Asia/Shanghai")
t.Setenv("ROOM_TURNOVER_REWARD_SETTLE_INTERVAL_SECONDS", "77")
t.Setenv("WEEK_STAR_RETRY_STREAM_KEY", "legacy-stream")
t.Setenv("WEEK_STAR_RETRY_CONSUMER_GROUP", "legacy-group")
t.Setenv("WEEK_STAR_RETRY_CONSUMER_NAME", "legacy-consumer")
@ -103,18 +89,12 @@ func TestLoadFallsBackToLegacyKeys(t *testing.T) {
if cfg.JavaOtherBaseURL != "http://legacy-java" {
t.Fatalf("expected legacy java base url, got %q", cfg.JavaOtherBaseURL)
}
if cfg.JavaWalletBaseURL != "http://legacy-wallet" {
t.Fatalf("expected legacy java wallet base url, got %q", cfg.JavaWalletBaseURL)
}
if cfg.WeekStarDefaultSysOrigin != "TARAB" {
t.Fatalf("expected normalized sys origin TARAB, got %q", cfg.WeekStarDefaultSysOrigin)
}
if cfg.WeekStarSettleInterval != 66*time.Second {
t.Fatalf("expected settle interval 66s, got %s", cfg.WeekStarSettleInterval)
}
if cfg.RoomTurnoverDefaultSysOrigin != "TARAB" || cfg.RoomTurnoverTimezone != "Asia/Shanghai" || cfg.RoomTurnoverSettleInterval != 77*time.Second {
t.Fatalf("unexpected legacy room turnover config: %+v", cfg)
}
if cfg.WeekStarRetryStreamKey != "legacy-stream" || cfg.WeekStarRetryConsumerGroup != "legacy-group" || cfg.WeekStarRetryConsumerName != "legacy-consumer" {
t.Fatalf("unexpected retry consumer config: %+v", cfg)
}

153
internal/integration/java.go
View File

@ -32,45 +32,6 @@ type UserProfile struct {
UserNickname string `json:"userNickname"`
}
type RoomContributionActivityCount struct {
ID string `json:"id"`
RoomID Int64Value `json:"roomId"`
ContributionValue AmountValue `json:"contributionValue"`
DateNumber int `json:"dateNumber"`
Received bool `json:"received"`
RewardCoinsSent bool `json:"rewardCoinsSent"`
}
type RoomProfile struct {
ID Int64Value `json:"id"`
RoomAccount string `json:"roomAccount"`
UserID Int64Value `json:"userId"`
RoomCover string `json:"roomCover"`
RoomName string `json:"roomName"`
SysOrigin string `json:"sysOrigin"`
CountryCode string `json:"countryCode"`
CountryName string `json:"countryName"`
}
type GoldReceiptCommand struct {
ReceiptType string `json:"receiptType"`
UserID int64 `json:"userId"`
SysOrigin string `json:"sysOrigin"`
EventID string `json:"eventId"`
Origin string `json:"origin,omitempty"`
Remark string `json:"remark,omitempty"`
Amount PennyAmountPayload `json:"amount"`
CloseDelayAsset bool `json:"closeDelayAsset"`
OpUserType string `json:"opUserType"`
CustomizeOrigin string `json:"customizeOrigin,omitempty"`
CustomizeOriginDesc string `json:"customizeOriginDesc,omitempty"`
}
type PennyAmountPayload struct {
PennyAmount int64 `json:"pennyAmount,string"`
DollarAmount int64 `json:"dollarAmount"`
}
type resultResponse[T any] struct {
Code int `json:"code"`
Message string `json:"message"`
@ -97,37 +58,6 @@ func (v *Int64Value) UnmarshalJSON(data []byte) error {
return nil
}
type AmountValue int64
func (v *AmountValue) UnmarshalJSON(data []byte) error {
raw := strings.TrimSpace(string(data))
if raw == "" || raw == "null" {
*v = 0
return nil
}
if strings.HasPrefix(raw, "\"") && strings.HasSuffix(raw, "\"") {
raw = strings.Trim(raw, "\"")
}
if raw == "" {
*v = 0
return nil
}
floatRaw := raw
if dot := strings.Index(raw, "."); dot >= 0 {
raw = raw[:dot]
}
if parsed, err := strconv.ParseInt(raw, 10, 64); err == nil {
*v = AmountValue(parsed)
return nil
}
parsed, err := strconv.ParseFloat(floatRaw, 64)
if err != nil {
return err
}
*v = AmountValue(int64(parsed))
return nil
}
func New(cfg config.Config) *Client {
return &Client{
cfg: cfg,
@ -141,66 +71,6 @@ func (c *Client) SendActivityReward(ctx context.Context, req SendActivityRewardR
return c.postJSON(ctx, c.cfg.JavaOtherBaseURL+"/props-activity-cnf/client/sendActivityReward", req, nil)
}
func (c *Client) ListLastWeekRoomContribution(ctx context.Context) ([]RoomContributionActivityCount, error) {
endpoint := c.cfg.JavaOtherBaseURL + "/room-contribution-activity-count/client/listLastWeek"
var resp resultResponse[[]RoomContributionActivityCount]
if err := c.getJSON(ctx, endpoint, &resp); err != nil {
return nil, err
}
return resp.Body, nil
}
func (c *Client) MarkRoomTurnoverRewardCoinsSent(ctx context.Context, id string) (bool, error) {
id = strings.TrimSpace(id)
if id == "" {
return false, fmt.Errorf("empty room contribution id")
}
endpoint := c.cfg.JavaOtherBaseURL + "/room-contribution-activity-count/client/rewardCoinsSent?id=" + url.QueryEscape(id)
var resp resultResponse[bool]
if err := c.postEmpty(ctx, endpoint, &resp); err != nil {
return false, err
}
return resp.Body, nil
}
func (c *Client) MapRoomProfiles(ctx context.Context, roomIDs []int64) (map[int64]RoomProfile, error) {
if len(roomIDs) == 0 {
return map[int64]RoomProfile{}, nil
}
endpoint := c.cfg.JavaOtherBaseURL + "/room-manager/client/mapProfileByRoomIds"
var resp resultResponse[map[string]RoomProfile]
if err := c.postJSON(ctx, endpoint, roomIDs, &resp); err != nil {
return nil, err
}
result := make(map[int64]RoomProfile, len(resp.Body))
for key, value := range resp.Body {
roomID, err := strconv.ParseInt(strings.TrimSpace(key), 10, 64)
if err != nil {
roomID = int64(value.ID)
}
if roomID <= 0 {
continue
}
result[roomID] = value
}
return result, nil
}
func (c *Client) ChangeGoldBalance(ctx context.Context, cmd GoldReceiptCommand) error {
var resp resultResponse[json.RawMessage]
if err := c.postJSON(ctx, c.cfg.JavaWalletBaseURL+"/wallet/gold/client/balance/change", cmd, &resp); err != nil {
return err
}
return ensureSuccessfulResult(resp.Code, resp.Message, resp.Success)
}
func NewPennyAmountPayloadFromDollar(amount int64) PennyAmountPayload {
return PennyAmountPayload{
PennyAmount: amount * 100,
DollarAmount: amount,
}
}
func (c *Client) GetUserProfile(ctx context.Context, userID int64) (UserProfile, error) {
endpoint := c.cfg.JavaOtherBaseURL + "/user-profile/client/getByUserId?userId=" + url.QueryEscape(strconv.FormatInt(userID, 10))
var resp resultResponse[UserProfile]
@ -244,19 +114,6 @@ func (c *Client) postJSON(ctx context.Context, endpoint string, payload any, out
return decodeResponse(resp, out)
}
func (c *Client) postEmpty(ctx context.Context, endpoint string, out any) error {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, nil)
if err != nil {
return err
}
resp, err := c.httpClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
return decodeResponse(resp, out)
}
func decodeResponse(resp *http.Response, out any) error {
body, err := io.ReadAll(resp.Body)
if err != nil {
@ -273,13 +130,3 @@ func decodeResponse(resp *http.Response, out any) error {
}
return nil
}
func ensureSuccessfulResult(code int, message string, success *bool) error {
if success != nil && !*success {
return fmt.Errorf("java api returned unsuccessful result: code=%d message=%s", code, strings.TrimSpace(message))
}
if code != 0 && code != 200 {
return fmt.Errorf("java api returned error result: code=%d message=%s", code, strings.TrimSpace(message))
}
return nil
}

56
internal/model/models.go
View File

@ -94,59 +94,3 @@ type WeekStarRewardRecord struct {
// TableName 返回周榜发奖记录表名。
func (WeekStarRewardRecord) TableName() string { return "week_star_reward_record" }
// RoomTurnoverRewardConfig 保存房间流水奖励的系统级配置。
type RoomTurnoverRewardConfig struct {
ID int64 `gorm:"column:id;primaryKey"`
SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_room_turnover_reward_sys_origin"`
Enabled bool `gorm:"column:enabled;index:idx_room_turnover_reward_enabled"`
Timezone string `gorm:"column:timezone;size:64"`
CreateTime time.Time `gorm:"column:create_time"`
UpdateTime time.Time `gorm:"column:update_time"`
}
func (RoomTurnoverRewardConfig) TableName() string { return "room_turnover_reward_config" }
// RoomTurnoverRewardLevel 保存达到某个流水门槛后发放的金币奖励。
type RoomTurnoverRewardLevel struct {
ID int64 `gorm:"column:id;primaryKey"`
ConfigID int64 `gorm:"column:config_id;uniqueIndex:uk_room_turnover_reward_level,priority:1;uniqueIndex:uk_room_turnover_reward_threshold,priority:1;index:idx_room_turnover_reward_level_config"`
Level int `gorm:"column:level;uniqueIndex:uk_room_turnover_reward_level,priority:2"`
TurnoverThreshold int64 `gorm:"column:turnover_threshold;uniqueIndex:uk_room_turnover_reward_threshold,priority:2"`
RewardGold int64 `gorm:"column:reward_gold"`
Enabled bool `gorm:"column:enabled"`
CreateTime time.Time `gorm:"column:create_time"`
UpdateTime time.Time `gorm:"column:update_time"`
}
func (RoomTurnoverRewardLevel) TableName() string { return "room_turnover_reward_level" }
// RoomTurnoverRewardRecord 记录每周给房主发放的最高档房间流水奖励。
type RoomTurnoverRewardRecord struct {
ID int64 `gorm:"column:id;primaryKey"`
ConfigID int64 `gorm:"column:config_id;index:idx_room_turnover_reward_record_config"`
SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_room_turnover_reward_record_room,priority:1;index:idx_room_turnover_reward_record_cycle,priority:1"`
CycleKey string `gorm:"column:cycle_key;size:32;uniqueIndex:uk_room_turnover_reward_record_room,priority:2;index:idx_room_turnover_reward_record_cycle,priority:2"`
PeriodStartAt time.Time `gorm:"column:period_start_at"`
PeriodEndAt time.Time `gorm:"column:period_end_at"`
RoomID int64 `gorm:"column:room_id;uniqueIndex:uk_room_turnover_reward_record_room,priority:3;index:idx_room_turnover_reward_record_room"`
RoomAccount string `gorm:"column:room_account;size:64"`
RoomName string `gorm:"column:room_name;size:255"`
RoomCover string `gorm:"column:room_cover;size:1024"`
OwnerUserID int64 `gorm:"column:owner_user_id;index:idx_room_turnover_reward_record_owner"`
CountryCode string `gorm:"column:country_code;size:32"`
CountryName string `gorm:"column:country_name;size:128"`
TurnoverAmount int64 `gorm:"column:turnover_amount"`
Level int `gorm:"column:level"`
TurnoverThreshold int64 `gorm:"column:turnover_threshold"`
RewardGold int64 `gorm:"column:reward_gold"`
TrackID string `gorm:"column:track_id;size:128;uniqueIndex:uk_room_turnover_reward_track"`
Status string `gorm:"column:status;size:32;index:idx_room_turnover_reward_record_status"`
RetryCount int `gorm:"column:retry_count"`
LastError string `gorm:"column:last_error;size:1024"`
SentAt *time.Time `gorm:"column:sent_at"`
CreateTime time.Time `gorm:"column:create_time"`
UpdateTime time.Time `gorm:"column:update_time"`
}
func (RoomTurnoverRewardRecord) TableName() string { return "room_turnover_reward_record" }

423
internal/service/room_turnover_reward_settlement.go
View File

@ -1,423 +0,0 @@
package service
import (
"context"
"fmt"
"log"
"sort"
"strconv"
"strings"
"time"
"chatapp-cron/internal/config"
"chatapp-cron/internal/integration"
"chatapp-cron/internal/model"
"chatapp-cron/internal/repo"
"chatapp-cron/internal/util"
"gorm.io/gorm/clause"
)
const (
roomTurnoverRewardStatusPending = "PENDING"
roomTurnoverRewardStatusSuccess = "SUCCESS"
roomTurnoverRewardStatusFailed = "FAILED"
roomTurnoverRewardGoldOrigin = "ROOM_CONTRIBUTION_REWARD"
)
type roomTurnoverRewardLevel struct {
Level int
TurnoverThreshold int64
RewardGold int64
}
// RoomTurnoverRewardSettlementService 负责房间流水奖励的周结算和补发。
type RoomTurnoverRewardSettlementService struct {
cfg config.Config
repo *repo.Repository
java *integration.Client
location *time.Location
}
func NewRoomTurnoverRewardSettlementService(cfg config.Config, repository *repo.Repository, javaClient *integration.Client) *RoomTurnoverRewardSettlementService {
location, err := time.LoadLocation(strings.TrimSpace(cfg.RoomTurnoverTimezone))
if err != nil {
location = time.UTC
}
return &RoomTurnoverRewardSettlementService{
cfg: cfg,
repo: repository,
java: javaClient,
location: location,
}
}
func (s *RoomTurnoverRewardSettlementService) Start(ctx context.Context) {
interval := s.cfg.RoomTurnoverSettleInterval
if interval <= 0 {
interval = time.Minute
}
go func() {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
if ctx.Err() != nil {
return
}
if err := s.SettleDueCycles(ctx); err != nil && ctx.Err() == nil {
log.Printf("room turnover reward settlement scan failed: %v", err)
}
select {
case <-ctx.Done():
return
case <-ticker.C:
}
}
}()
}
func (s *RoomTurnoverRewardSettlementService) SettleDueCycles(ctx context.Context) error {
var configs []model.RoomTurnoverRewardConfig
if err := s.repo.DB.WithContext(ctx).
Where("enabled = ?", true).
Order("id desc").
Find(&configs).Error; err != nil {
return err
}
if len(configs) == 0 {
return nil
}
contributions, err := s.java.ListLastWeekRoomContribution(ctx)
if err != nil {
return err
}
if len(contributions) == 0 {
return nil
}
roomProfiles, err := s.java.MapRoomProfiles(ctx, uniqueRoomTurnoverRewardRoomIDs(contributions))
if err != nil {
return err
}
var lastErr error
for index := range configs {
if err := s.settleConfig(ctx, &configs[index], contributions, roomProfiles); err != nil {
lastErr = err
log.Printf("room turnover reward settle config failed. configId=%d err=%v", configs[index].ID, err)
}
}
return lastErr
}
func (s *RoomTurnoverRewardSettlementService) settleConfig(
ctx context.Context,
configRow *model.RoomTurnoverRewardConfig,
contributions []integration.RoomContributionActivityCount,
roomProfiles map[int64]integration.RoomProfile,
) error {
if configRow == nil {
return nil
}
sysOrigin := normalizeRoomTurnoverRewardSysOrigin(configRow.SysOrigin, s.cfg.RoomTurnoverDefaultSysOrigin)
levels, err := s.loadLevels(ctx, configRow.ID)
if err != nil {
return err
}
if len(levels) == 0 {
return nil
}
byCycle := make(map[string][]integration.RoomContributionActivityCount)
for _, contribution := range contributions {
roomID := int64(contribution.RoomID)
if roomID <= 0 {
continue
}
profile, ok := roomProfiles[roomID]
if !ok {
continue
}
profileSysOrigin := strings.ToUpper(strings.TrimSpace(profile.SysOrigin))
if profileSysOrigin != "" && profileSysOrigin != sysOrigin {
continue
}
cycleKey := roomTurnoverRewardCycleKey(contribution.DateNumber, time.Now().In(s.location))
if cycleKey == "" {
continue
}
byCycle[cycleKey] = append(byCycle[cycleKey], contribution)
}
var lastErr error
for cycleKey, cycleContributions := range byCycle {
if err := s.settleCycle(ctx, configRow, sysOrigin, cycleKey, levels, cycleContributions, roomProfiles); err != nil {
lastErr = err
}
}
return lastErr
}
func (s *RoomTurnoverRewardSettlementService) settleCycle(
ctx context.Context,
configRow *model.RoomTurnoverRewardConfig,
sysOrigin string,
cycleKey string,
levels []roomTurnoverRewardLevel,
contributions []integration.RoomContributionActivityCount,
roomProfiles map[int64]integration.RoomProfile,
) error {
lockKey := fmt.Sprintf("room-turnover-reward:settle-lock:%d:%s", configRow.ID, cycleKey)
acquired, err := s.repo.Redis.SetNX(ctx, lockKey, 1, 5*time.Minute).Result()
if err != nil {
return err
}
if !acquired {
return nil
}
periodStart, periodEnd, err := s.periodBounds(cycleKey)
if err != nil {
return err
}
now := time.Now()
var lastErr error
for _, contribution := range contributions {
roomID := int64(contribution.RoomID)
profile := roomProfiles[roomID]
level, ok := pickRoomTurnoverRewardLevel(levels, int64(contribution.ContributionValue))
if !ok {
continue
}
record, err := s.ensureRewardRecord(ctx, configRow.ID, sysOrigin, cycleKey, periodStart, periodEnd, contribution, profile, level, now)
if err != nil {
lastErr = err
continue
}
if record.Status == roomTurnoverRewardStatusSuccess {
if contribution.ID != "" {
if _, markErr := s.java.MarkRoomTurnoverRewardCoinsSent(ctx, contribution.ID); markErr != nil {
lastErr = markErr
log.Printf("mark room turnover reward source failed. recordId=%d sourceId=%s err=%v", record.ID, contribution.ID, markErr)
}
}
continue
}
if record.Status != roomTurnoverRewardStatusPending && record.Status != roomTurnoverRewardStatusFailed {
continue
}
if err := s.dispatchRewardRecord(ctx, &record, contribution.ID); err != nil {
lastErr = err
}
}
return lastErr
}
func (s *RoomTurnoverRewardSettlementService) loadLevels(ctx context.Context, configID int64) ([]roomTurnoverRewardLevel, error) {
var rows []model.RoomTurnoverRewardLevel
if err := s.repo.DB.WithContext(ctx).
Where("config_id = ? AND enabled = ?", configID, true).
Order("turnover_threshold desc").
Find(&rows).Error; err != nil {
return nil, err
}
levels := make([]roomTurnoverRewardLevel, 0, len(rows))
for _, row := range rows {
if row.TurnoverThreshold <= 0 || row.RewardGold <= 0 {
continue
}
levels = append(levels, roomTurnoverRewardLevel{
Level: row.Level,
TurnoverThreshold: row.TurnoverThreshold,
RewardGold: row.RewardGold,
})
}
sort.Slice(levels, func(i, j int) bool {
return levels[i].TurnoverThreshold > levels[j].TurnoverThreshold
})
return levels, nil
}
func (s *RoomTurnoverRewardSettlementService) ensureRewardRecord(
ctx context.Context,
configID int64,
sysOrigin string,
cycleKey string,
periodStart time.Time,
periodEnd time.Time,
contribution integration.RoomContributionActivityCount,
profile integration.RoomProfile,
level roomTurnoverRewardLevel,
now time.Time,
) (model.RoomTurnoverRewardRecord, error) {
roomID := int64(contribution.RoomID)
recordID, err := util.NextID()
if err != nil {
return model.RoomTurnoverRewardRecord{}, err
}
record := model.RoomTurnoverRewardRecord{
ID: recordID,
ConfigID: configID,
SysOrigin: sysOrigin,
CycleKey: cycleKey,
PeriodStartAt: periodStart,
PeriodEndAt: periodEnd,
RoomID: roomID,
RoomAccount: profile.RoomAccount,
RoomName: profile.RoomName,
RoomCover: profile.RoomCover,
OwnerUserID: int64(profile.UserID),
CountryCode: profile.CountryCode,
CountryName: profile.CountryName,
TurnoverAmount: int64(contribution.ContributionValue),
Level: level.Level,
TurnoverThreshold: level.TurnoverThreshold,
RewardGold: level.RewardGold,
TrackID: roomTurnoverRewardTrackID(sysOrigin, cycleKey, roomID),
Status: roomTurnoverRewardStatusPending,
CreateTime: now,
UpdateTime: now,
}
if record.OwnerUserID <= 0 {
return model.RoomTurnoverRewardRecord{}, fmt.Errorf("room owner missing. roomId=%d", roomID)
}
if err := s.repo.DB.WithContext(ctx).
Clauses(clause.OnConflict{DoNothing: true}).
Create(&record).Error; err != nil {
return model.RoomTurnoverRewardRecord{}, err
}
var saved model.RoomTurnoverRewardRecord
if err := s.repo.DB.WithContext(ctx).
Where("sys_origin = ? AND cycle_key = ? AND room_id = ?", sysOrigin, cycleKey, roomID).
First(&saved).Error; err != nil {
return model.RoomTurnoverRewardRecord{}, err
}
return saved, nil
}
func (s *RoomTurnoverRewardSettlementService) dispatchRewardRecord(ctx context.Context, record *model.RoomTurnoverRewardRecord, sourceID string) error {
if record == nil {
return nil
}
now := time.Now()
retryCount := record.RetryCount
if record.Status == roomTurnoverRewardStatusFailed {
retryCount++
}
err := s.java.ChangeGoldBalance(ctx, integration.GoldReceiptCommand{
ReceiptType: "INCOME",
UserID: record.OwnerUserID,
SysOrigin: record.SysOrigin,
EventID: record.TrackID,
Origin: roomTurnoverRewardGoldOrigin,
Remark: fmt.Sprintf("room turnover reward %s room %d", record.CycleKey, record.RoomID),
Amount: integration.NewPennyAmountPayloadFromDollar(record.RewardGold),
CloseDelayAsset: false,
OpUserType: "BACK",
})
if err != nil {
updates := map[string]any{
"status": roomTurnoverRewardStatusFailed,
"retry_count": retryCount,
"last_error": truncateString(err.Error(), 1024),
"update_time": now,
}
if updateErr := s.repo.DB.WithContext(context.Background()).
Model(&model.RoomTurnoverRewardRecord{}).
Where("id = ?", record.ID).
Updates(updates).Error; updateErr != nil {
return updateErr
}
return err
}
if err := s.repo.DB.WithContext(ctx).
Model(&model.RoomTurnoverRewardRecord{}).
Where("id = ?", record.ID).
Updates(map[string]any{
"status": roomTurnoverRewardStatusSuccess,
"retry_count": retryCount,
"last_error": "",
"sent_at": now,
"update_time": now,
}).Error; err != nil {
return err
}
if strings.TrimSpace(sourceID) == "" {
return nil
}
if marked, err := s.java.MarkRoomTurnoverRewardCoinsSent(ctx, sourceID); err != nil {
return err
} else if !marked {
log.Printf("room turnover reward source was not modified. recordId=%d sourceId=%s", record.ID, sourceID)
}
return nil
}
func (s *RoomTurnoverRewardSettlementService) periodBounds(cycleKey string) (time.Time, time.Time, error) {
start, err := time.ParseInLocation("20060102", cycleKey, s.location)
if err != nil {
return time.Time{}, time.Time{}, err
}
return start, start.AddDate(0, 0, 7), nil
}
func pickRoomTurnoverRewardLevel(levels []roomTurnoverRewardLevel, turnoverAmount int64) (roomTurnoverRewardLevel, bool) {
if turnoverAmount <= 0 {
return roomTurnoverRewardLevel{}, false
}
for _, level := range levels {
if turnoverAmount >= level.TurnoverThreshold {
return level, true
}
}
return roomTurnoverRewardLevel{}, false
}
func uniqueRoomTurnoverRewardRoomIDs(contributions []integration.RoomContributionActivityCount) []int64 {
result := make([]int64, 0, len(contributions))
seen := make(map[int64]struct{}, len(contributions))
for _, item := range contributions {
roomID := int64(item.RoomID)
if roomID <= 0 {
continue
}
if _, exists := seen[roomID]; exists {
continue
}
seen[roomID] = struct{}{}
result = append(result, roomID)
}
return result
}
func normalizeRoomTurnoverRewardSysOrigin(values ...string) string {
for _, value := range values {
normalized := strings.ToUpper(strings.TrimSpace(value))
if normalized != "" {
return normalized
}
}
return "LIKEI"
}
func roomTurnoverRewardCycleKey(dateNumber int, now time.Time) string {
if dateNumber > 0 {
return fmt.Sprintf("%08d", dateNumber)
}
weekday := int(now.Weekday())
if weekday == 0 {
weekday = 7
}
monday := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()).AddDate(0, 0, 1-weekday-7)
return monday.Format("20060102")
}
func roomTurnoverRewardTrackID(sysOrigin string, cycleKey string, roomID int64) string {
return "room-turnover:" + strings.ToUpper(strings.TrimSpace(sysOrigin)) + ":" + cycleKey + ":" + strconv.FormatInt(roomID, 10)
}