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

增加幸运礼物修复和大屏

Browse Source
This commit is contained in:
ZuoZuo 2026-05-31 16:39:49 +08:00
parent fedd5c8f41
commit 88584f1329
32 changed files with 1187 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
api/proto/wallet/v1/wallet.pb.go
View File

@ -10339,18 +10339,19 @@ func (x *CreditTaskRewardResponse) GetGrantedAtMs() int64 {
// CreditLuckyGiftRewardRequest 是 activity-service 抽奖 outbox 补偿 worker 调用的钱包入账命令。
type CreditLuckyGiftRewardRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
CommandId string `protobuf:"bytes,1,opt,name=command_id,json=commandId,proto3" json:"command_id,omitempty"`
AppCode string `protobuf:"bytes,2,opt,name=app_code,json=appCode,proto3" json:"app_code,omitempty"`
TargetUserId int64 `protobuf:"varint,3,opt,name=target_user_id,json=targetUserId,proto3" json:"target_user_id,omitempty"`
Amount int64 `protobuf:"varint,4,opt,name=amount,proto3" json:"amount,omitempty"`
DrawId string `protobuf:"bytes,5,opt,name=draw_id,json=drawId,proto3" json:"draw_id,omitempty"`
RoomId string `protobuf:"bytes,6,opt,name=room_id,json=roomId,proto3" json:"room_id,omitempty"`
GiftId string `protobuf:"bytes,7,opt,name=gift_id,json=giftId,proto3" json:"gift_id,omitempty"`
PoolId string `protobuf:"bytes,8,opt,name=pool_id,json=poolId,proto3" json:"pool_id,omitempty"`
Reason string `protobuf:"bytes,9,opt,name=reason,proto3" json:"reason,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
state protoimpl.MessageState `protogen:"open.v1"`
CommandId string `protobuf:"bytes,1,opt,name=command_id,json=commandId,proto3" json:"command_id,omitempty"`
AppCode string `protobuf:"bytes,2,opt,name=app_code,json=appCode,proto3" json:"app_code,omitempty"`
TargetUserId int64 `protobuf:"varint,3,opt,name=target_user_id,json=targetUserId,proto3" json:"target_user_id,omitempty"`
Amount int64 `protobuf:"varint,4,opt,name=amount,proto3" json:"amount,omitempty"`
DrawId string `protobuf:"bytes,5,opt,name=draw_id,json=drawId,proto3" json:"draw_id,omitempty"`
RoomId string `protobuf:"bytes,6,opt,name=room_id,json=roomId,proto3" json:"room_id,omitempty"`
GiftId string `protobuf:"bytes,7,opt,name=gift_id,json=giftId,proto3" json:"gift_id,omitempty"`
PoolId string `protobuf:"bytes,8,opt,name=pool_id,json=poolId,proto3" json:"pool_id,omitempty"`
Reason string `protobuf:"bytes,9,opt,name=reason,proto3" json:"reason,omitempty"`
VisibleRegionId int64 `protobuf:"varint,10,opt,name=visible_region_id,json=visibleRegionId,proto3" json:"visible_region_id,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *CreditLuckyGiftRewardRequest) Reset() {
@ -10446,6 +10447,13 @@ func (x *CreditLuckyGiftRewardRequest) GetReason() string {
return ""
}
func (x *CreditLuckyGiftRewardRequest) GetVisibleRegionId() int64 {
if x != nil {
return x.VisibleRegionId
}
return 0
}
// CreditLuckyGiftRewardResponse 返回幸运礼物返奖入账流水和用户 COIN 账后余额。
type CreditLuckyGiftRewardResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
@ -13566,7 +13574,7 @@ const file_proto_wallet_v1_wallet_proto_rawDesc = "" +
"\x0etransaction_id\x18\x01 \x01(\tR\rtransactionId\x127\n" +
"\abalance\x18\x02 \x01(\v2\x1d.hyapp.wallet.v1.AssetBalanceR\abalance\x12\x16\n" +
"\x06amount\x18\x03 \x01(\x03R\x06amount\x12\"\n" +
"\rgranted_at_ms\x18\x04 \x01(\x03R\vgrantedAtMs\"\x92\x02\n" +
"\rgranted_at_ms\x18\x04 \x01(\x03R\vgrantedAtMs\"\xbe\x02\n" +
"\x1cCreditLuckyGiftRewardRequest\x12\x1d\n" +
"\n" +
"command_id\x18\x01 \x01(\tR\tcommandId\x12\x19\n" +
@ -13577,7 +13585,9 @@ const file_proto_wallet_v1_wallet_proto_rawDesc = "" +
"\aroom_id\x18\x06 \x01(\tR\x06roomId\x12\x17\n" +
"\agift_id\x18\a \x01(\tR\x06giftId\x12\x17\n" +
"\apool_id\x18\b \x01(\tR\x06poolId\x12\x16\n" +
"\x06reason\x18\t \x01(\tR\x06reason\"\xbb\x01\n" +
"\x06reason\x18\t \x01(\tR\x06reason\x12*\n" +
"\x11visible_region_id\x18\n" +
" \x01(\x03R\x0fvisibleRegionId\"\xbb\x01\n" +
"\x1dCreditLuckyGiftRewardResponse\x12%\n" +
"\x0etransaction_id\x18\x01 \x01(\tR\rtransactionId\x127\n" +
"\abalance\x18\x02 \x01(\v2\x1d.hyapp.wallet.v1.AssetBalanceR\abalance\x12\x16\n" +

1
api/proto/wallet/v1/wallet.proto
View File

@ -1155,6 +1155,7 @@ message CreditLuckyGiftRewardRequest {
string gift_id = 7;
string pool_id = 8;
string reason = 9;
int64 visible_region_id = 10;
}
// CreditLuckyGiftRewardResponse 返回幸运礼物返奖入账流水和用户 COIN 账后余额。

2
server/admin/cmd/server/main.go
View File

@ -235,7 +235,7 @@ func main() {
HostSalarySettlement: hostsalarysettlementmodule.New(walletDB, userDB),
Job: jobmodule.New(store, cfg, auditHandler),
LevelConfig: levelconfigmodule.New(activityclient.NewGRPC(activityConn), auditHandler),
LuckyGift: luckygiftmodule.New(activityclient.NewGRPC(activityConn), auditHandler),
LuckyGift: luckygiftmodule.New(activityclient.NewGRPC(activityConn), cfg.ActivityService.RequestTimeout, auditHandler),
Menu: menumodule.New(store, auditHandler),
Payment: paymentmodule.New(walletclient.NewGRPC(walletConn), auditHandler),
RBAC: rbacmodule.New(store, auditHandler),

37
server/admin/internal/modules/luckygift/handler.go
View File

@ -1,6 +1,7 @@
package luckygift
import (
"context"
"math"
"strings"
"time"
@ -16,12 +17,16 @@ import (
)
type Handler struct {
activity activityclient.Client
audit shared.OperationLogger
activity activityclient.Client
requestTimeout time.Duration
audit shared.OperationLogger
}
func New(activity activityclient.Client, audit shared.OperationLogger) *Handler {
return &Handler{activity: activity, audit: audit}
func New(activity activityclient.Client, requestTimeout time.Duration, audit shared.OperationLogger) *Handler {
if requestTimeout <= 0 {
requestTimeout = 3 * time.Second
}
return &Handler{activity: activity, requestTimeout: requestTimeout, audit: audit}
}
type configRequest struct {
@ -108,7 +113,9 @@ type drawSummaryDTO struct {
}
func (h *Handler) GetLuckyGiftConfig(c *gin.Context) {
resp, err := h.activity.GetLuckyGiftConfig(c.Request.Context(), &activityv1.GetLuckyGiftConfigRequest{
ctx, cancel := h.activityContext(c)
defer cancel()
resp, err := h.activity.GetLuckyGiftConfig(ctx, &activityv1.GetLuckyGiftConfigRequest{
Meta: h.meta(c),
PoolId: strings.TrimSpace(c.Query("pool_id")),
})
@ -128,7 +135,9 @@ func (h *Handler) UpsertLuckyGiftConfig(c *gin.Context) {
if req.PoolID == "" {
req.PoolID = strings.TrimSpace(c.Query("pool_id"))
}
resp, err := h.activity.UpsertLuckyGiftConfig(c.Request.Context(), &activityv1.UpsertLuckyGiftConfigRequest{
ctx, cancel := h.activityContext(c)
defer cancel()
resp, err := h.activity.UpsertLuckyGiftConfig(ctx, &activityv1.UpsertLuckyGiftConfigRequest{
Meta: h.meta(c),
Config: configToProto(req),
OperatorAdminId: int64(middleware.CurrentUserID(c)),
@ -143,7 +152,9 @@ func (h *Handler) UpsertLuckyGiftConfig(c *gin.Context) {
}
func (h *Handler) ListLuckyGiftConfigs(c *gin.Context) {
resp, err := h.activity.ListLuckyGiftConfigs(c.Request.Context(), &activityv1.ListLuckyGiftConfigsRequest{
ctx, cancel := h.activityContext(c)
defer cancel()
resp, err := h.activity.ListLuckyGiftConfigs(ctx, &activityv1.ListLuckyGiftConfigsRequest{
Meta: h.meta(c),
})
if err != nil {
@ -159,7 +170,9 @@ func (h *Handler) ListLuckyGiftConfigs(c *gin.Context) {
func (h *Handler) ListLuckyGiftDraws(c *gin.Context) {
options := shared.ListOptions(c)
resp, err := h.activity.ListLuckyGiftDraws(c.Request.Context(), &activityv1.ListLuckyGiftDrawsRequest{
ctx, cancel := h.activityContext(c)
defer cancel()
resp, err := h.activity.ListLuckyGiftDraws(ctx, &activityv1.ListLuckyGiftDrawsRequest{
Meta: h.meta(c),
PoolId: strings.TrimSpace(c.Query("pool_id")),
GiftId: strings.TrimSpace(c.Query("gift_id")),
@ -181,7 +194,9 @@ func (h *Handler) ListLuckyGiftDraws(c *gin.Context) {
}
func (h *Handler) GetLuckyGiftDrawSummary(c *gin.Context) {
resp, err := h.activity.GetLuckyGiftDrawSummary(c.Request.Context(), &activityv1.GetLuckyGiftDrawSummaryRequest{
ctx, cancel := h.activityContext(c)
defer cancel()
resp, err := h.activity.GetLuckyGiftDrawSummary(ctx, &activityv1.GetLuckyGiftDrawSummaryRequest{
Meta: h.meta(c),
PoolId: strings.TrimSpace(c.Query("pool_id")),
GiftId: strings.TrimSpace(c.Query("gift_id")),
@ -205,6 +220,10 @@ func (h *Handler) meta(c *gin.Context) *activityv1.RequestMeta {
}
}
func (h *Handler) activityContext(c *gin.Context) (context.Context, context.CancelFunc) {
return context.WithTimeout(c.Request.Context(), h.requestTimeout)
}
func configToProto(req configRequest) *activityv1.LuckyGiftRuleConfig {
stages := make([]*activityv1.LuckyGiftRuleStage, 0, len(req.Stages))
for _, stage := range req.Stages {

1
services/activity-service/configs/config.docker.yaml
View File

@ -21,6 +21,7 @@ lucky_gift_worker:
enabled: true
worker_poll_interval: "1s"
worker_batch_size: 100
worker_concurrency: 4
worker_lock_ttl: "30s"
worker_max_retry: 8
publish_timeout: "5s"

1
services/activity-service/configs/config.tencent.example.yaml
View File

@ -21,6 +21,7 @@ lucky_gift_worker:
enabled: true
worker_poll_interval: "1s"
worker_batch_size: 100
worker_concurrency: 4
worker_lock_ttl: "30s"
worker_max_retry: 8
publish_timeout: "5s"

1
services/activity-service/configs/config.yaml
View File

@ -21,6 +21,7 @@ lucky_gift_worker:
enabled: true
worker_poll_interval: "1s"
worker_batch_size: 100
worker_concurrency: 4
worker_lock_ttl: "30s"
worker_max_retry: 8
publish_timeout: "5s"

38
services/activity-service/deploy/mysql/initdb/001_activity_service.sql
View File

@ -256,6 +256,44 @@ CREATE TABLE IF NOT EXISTS lucky_draw_records (
KEY idx_lucky_draw_status (app_code, reward_status, updated_at_ms)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物抽奖事实';
CREATE TABLE IF NOT EXISTS lucky_draw_pool_stats (
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码',
pool_id VARCHAR(96) NOT NULL COMMENT '幸运礼物奖池 ID',
gift_id VARCHAR(96) NOT NULL DEFAULT '' COMMENT '礼物 ID，空值表示奖池汇总',
total_draws BIGINT NOT NULL DEFAULT 0 COMMENT '抽奖次数',
unique_users BIGINT NOT NULL DEFAULT 0 COMMENT '参与用户数',
unique_rooms BIGINT NOT NULL DEFAULT 0 COMMENT '参与房间数',
total_spent_coins BIGINT NOT NULL DEFAULT 0 COMMENT '消耗金币',
total_reward_coins BIGINT NOT NULL DEFAULT 0 COMMENT '返还金币',
base_reward_coins BIGINT NOT NULL DEFAULT 0 COMMENT '基础 RTP 返奖',
room_atmosphere_reward_coins BIGINT NOT NULL DEFAULT 0 COMMENT '房间气氛支出',
activity_subsidy_coins BIGINT NOT NULL DEFAULT 0 COMMENT '活动补贴支出',
pending_draws BIGINT NOT NULL DEFAULT 0 COMMENT '待发放数量',
granted_draws BIGINT NOT NULL DEFAULT 0 COMMENT '已发放数量',
failed_draws BIGINT NOT NULL DEFAULT 0 COMMENT '发放失败数量',
created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms',
updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms',
PRIMARY KEY (app_code, pool_id, gift_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物池级汇总统计';
CREATE TABLE IF NOT EXISTS lucky_draw_pool_stat_users (
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码',
pool_id VARCHAR(96) NOT NULL COMMENT '幸运礼物奖池 ID',
gift_id VARCHAR(96) NOT NULL DEFAULT '' COMMENT '礼物 ID，空值表示奖池汇总',
user_id BIGINT NOT NULL COMMENT '用户 ID',
created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms',
PRIMARY KEY (app_code, pool_id, gift_id, user_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物汇总用户去重';
CREATE TABLE IF NOT EXISTS lucky_draw_pool_stat_rooms (
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码',
pool_id VARCHAR(96) NOT NULL COMMENT '幸运礼物奖池 ID',
gift_id VARCHAR(96) NOT NULL DEFAULT '' COMMENT '礼物 ID，空值表示奖池汇总',
room_id VARCHAR(96) NOT NULL COMMENT '房间 ID',
created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms',
PRIMARY KEY (app_code, pool_id, gift_id, room_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物汇总房间去重';
CREATE TABLE IF NOT EXISTS im_broadcast_outbox (
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码，用于多租户隔离',
event_id VARCHAR(128) NOT NULL COMMENT '事件幂等 ID',

1
services/activity-service/internal/app/app.go
View File

@ -532,6 +532,7 @@ func luckyGiftWorkerOptions(nodeID string, cfg config.LuckyGiftWorkerConfig) luc
WorkerID: nodeID + "-lucky-gift",
PollInterval: cfg.WorkerPollInterval,
BatchSize: cfg.WorkerBatchSize,
Concurrency: cfg.WorkerConcurrency,
LockTTL: cfg.WorkerLockTTL,
MaxRetry: cfg.WorkerMaxRetry,
PublishTimeout: cfg.PublishTimeout,

8
services/activity-service/internal/config/config.go
View File

@ -72,6 +72,7 @@ type LuckyGiftWorkerConfig struct {
Enabled bool `yaml:"enabled"`
WorkerPollInterval time.Duration `yaml:"worker_poll_interval"`
WorkerBatchSize int `yaml:"worker_batch_size"`
WorkerConcurrency int `yaml:"worker_concurrency"`
WorkerLockTTL time.Duration `yaml:"worker_lock_ttl"`
WorkerMaxRetry int `yaml:"worker_max_retry"`
PublishTimeout time.Duration `yaml:"publish_timeout"`
@ -169,6 +170,7 @@ func Default() Config {
Enabled: false,
WorkerPollInterval: time.Second,
WorkerBatchSize: 100,
WorkerConcurrency: 4,
WorkerLockTTL: 30 * time.Second,
WorkerMaxRetry: 8,
PublishTimeout: 5 * time.Second,
@ -297,6 +299,12 @@ func Load(path string) (Config, error) {
if cfg.LuckyGiftWorker.WorkerBatchSize <= 0 {
cfg.LuckyGiftWorker.WorkerBatchSize = 100
}
if cfg.LuckyGiftWorker.WorkerConcurrency <= 0 {
cfg.LuckyGiftWorker.WorkerConcurrency = 4
}
if cfg.LuckyGiftWorker.WorkerConcurrency > cfg.LuckyGiftWorker.WorkerBatchSize {
cfg.LuckyGiftWorker.WorkerConcurrency = cfg.LuckyGiftWorker.WorkerBatchSize
}
if cfg.LuckyGiftWorker.WorkerLockTTL <= 0 {
cfg.LuckyGiftWorker.WorkerLockTTL = 30 * time.Second
}

90
services/activity-service/internal/service/luckygift/service.go
View File

@ -6,6 +6,7 @@ import (
"fmt"
"log/slog"
"strings"
"sync"
"time"
"hyapp/pkg/appcode"
@ -142,6 +143,7 @@ type WorkerOptions struct {
WorkerID string
PollInterval time.Duration
BatchSize int
Concurrency int
LockTTL time.Duration
MaxRetry int
PublishTimeout time.Duration
@ -174,15 +176,54 @@ func (s *Service) ProcessPendingDrawOutbox(ctx context.Context, options WorkerOp
if err != nil {
return 0, err
}
processed := 0
if len(events) == 0 {
return 0, nil
}
workerCount := options.Concurrency
if workerCount > len(events) {
workerCount = len(events)
}
jobs := make(chan domain.DrawOutbox)
errs := make(chan error, len(events))
var wg sync.WaitGroup
for index := 0; index < workerCount; index++ {
wg.Add(1)
go func() {
defer wg.Done()
for event := range jobs {
if err := s.processDrawOutbox(ctx, event, options); err != nil {
errs <- err
continue
}
errs <- nil
}
}()
}
for _, event := range events {
// 单条失败立即返回，让外层记录批次错误；已处理成功的事件已经各自落 delivered，不会被回滚。
if err := s.processDrawOutbox(ctx, event, options); err != nil {
return processed, err
select {
case <-ctx.Done():
close(jobs)
wg.Wait()
close(errs)
return 0, ctx.Err()
case jobs <- event:
}
}
close(jobs)
wg.Wait()
close(errs)
processed := 0
var firstErr error
for err := range errs {
if err != nil {
if firstErr == nil {
firstErr = err
}
continue
}
processed++
}
return processed, nil
return processed, firstErr
}
func (s *Service) processDrawOutbox(ctx context.Context, event domain.DrawOutbox, options WorkerOptions) error {
@ -200,15 +241,16 @@ func (s *Service) processDrawOutbox(ctx context.Context, event domain.DrawOutbox
}
// 返奖 command_id 派生自聚合 draw_id，钱包侧可用它做幂等；批量抽奖不按子抽逐条加币，避免待发放和钱包流水膨胀。
resp, err := s.wallet.CreditLuckyGiftReward(eventCtx, &walletv1.CreditLuckyGiftRewardRequest{
CommandId: "lucky_reward:" + payload.DrawID,
AppCode: event.AppCode,
TargetUserId: payload.UserID,
Amount: payload.EffectiveRewardCoins,
DrawId: payload.DrawID,
RoomId: payload.RoomID,
GiftId: payload.GiftID,
PoolId: payload.PoolID,
Reason: "lucky_gift_reward",
CommandId: "lucky_reward:" + payload.DrawID,
AppCode: event.AppCode,
TargetUserId: payload.UserID,
Amount: payload.EffectiveRewardCoins,
DrawId: payload.DrawID,
RoomId: payload.RoomID,
GiftId: payload.GiftID,
PoolId: payload.PoolID,
Reason: "lucky_gift_reward",
VisibleRegionId: payload.VisibleRegionID,
})
if err != nil {
return s.markOutboxFailed(eventCtx, event, options, err, false)
@ -216,6 +258,12 @@ func (s *Service) processDrawOutbox(ctx context.Context, event domain.DrawOutbox
walletTransactionID = resp.GetTransactionId()
credited = true
}
if credited {
nowMS := s.now().UTC().UnixMilli()
if err := s.repository.MarkLuckyGiftDrawsGranted(eventCtx, event.AppCode, payload.DrawIDs, walletTransactionID, nowMS); err != nil {
return err
}
}
if s.publisher == nil {
// 已返奖但缺少房间发布依赖时不能把 draw 标记 failed，否则会误导账务；只让 outbox 保持可重试。
return s.markOutboxFailed(eventCtx, event, options, fmt.Errorf("lucky gift room publisher is not configured"), credited)
@ -233,10 +281,10 @@ func (s *Service) processDrawOutbox(ctx context.Context, event domain.DrawOutbox
return s.markOutboxFailed(eventCtx, event, options, err, credited)
}
nowMS := s.now().UTC().UnixMilli()
// 所有外部副作用都成功后再把整批 draw 从 pending 收敛到 granted。
// 这一步必须晚于钱包和 IM，否则客户端可能看到“已发放”但余额或房间表现还没落地。
if err := s.repository.MarkLuckyGiftDrawsGranted(eventCtx, event.AppCode, payload.DrawIDs, walletTransactionID, nowMS); err != nil {
return err
if !credited {
if err := s.repository.MarkLuckyGiftDrawsGranted(eventCtx, event.AppCode, payload.DrawIDs, walletTransactionID, nowMS); err != nil {
return err
}
}
return s.repository.MarkLuckyGiftOutboxDelivered(eventCtx, event, nowMS)
}
@ -490,6 +538,12 @@ func normalizeWorkerOptions(options WorkerOptions) WorkerOptions {
if options.BatchSize > 500 {
options.BatchSize = 500
}
if options.Concurrency <= 0 {
options.Concurrency = 4
}
if options.Concurrency > options.BatchSize {
options.Concurrency = options.BatchSize
}
if options.LockTTL <= 0 {
options.LockTTL = 30 * time.Second
}

62
services/activity-service/internal/service/luckygift/service_test.go
View File

@ -3,6 +3,7 @@ package luckygift
import (
"context"
"encoding/json"
"errors"
"strings"
"testing"
"time"
@ -124,6 +125,7 @@ func TestProcessPendingBatchDrawOutboxCreditsWalletOnceAndGrantsAllDraws(t *test
"room_id": "room-1",
"gift_id": "rose",
"gift_count": 3,
"visible_region_id": 210,
"coin_spent": 300,
"effective_reward_coins": 800,
"multiplier_ppm": 8000000,
@ -139,7 +141,8 @@ func TestProcessPendingBatchDrawOutboxCreditsWalletOnceAndGrantsAllDraws(t *test
}
wallet := &fakeLuckyGiftWallet{}
publisher := &fakeLuckyGiftPublisher{}
service := New(repository, WithWallet(wallet), WithRoomPublisher(publisher))
broadcaster := &fakeLuckyGiftRegionBroadcaster{}
service := New(repository, WithWallet(wallet), WithRoomPublisher(publisher), WithRegionBroadcaster(broadcaster))
service.SetClock(func() time.Time { return time.UnixMilli(1760000001000).UTC() })
processed, err := service.ProcessPendingDrawOutbox(context.Background(), WorkerOptions{WorkerID: "worker-1"})
@ -149,7 +152,7 @@ func TestProcessPendingBatchDrawOutboxCreditsWalletOnceAndGrantsAllDraws(t *test
if processed != 1 {
t.Fatalf("processed count mismatch: %d", processed)
}
if wallet.last == nil || wallet.last.GetAmount() != 800 || wallet.last.GetDrawId() != "lucky_draw_batch_1" || wallet.last.GetCommandId() != "lucky_reward:lucky_draw_batch_1" {
if wallet.last == nil || wallet.last.GetAmount() != 800 || wallet.last.GetDrawId() != "lucky_draw_batch_1" || wallet.last.GetCommandId() != "lucky_reward:lucky_draw_batch_1" || wallet.last.GetVisibleRegionId() != 210 {
t.Fatalf("wallet should receive one aggregate reward request: %+v", wallet.last)
}
if got := strings.Join(repository.grantedDrawIDs, ","); got != "lucky_draw_batch_1,lucky_draw_batch_2,lucky_draw_batch_3" {
@ -157,6 +160,54 @@ func TestProcessPendingBatchDrawOutboxCreditsWalletOnceAndGrantsAllDraws(t *test
}
}
func TestProcessPendingDrawOutboxGrantsRewardWhenRoomIMFails(t *testing.T) {
payload, _ := json.Marshal(map[string]any{
"event_type": "lucky_gift_drawn",
"event_id": "lucky_gift_drawn:lucky_draw_im_fail",
"app_code": "lalu",
"draw_id": "lucky_draw_im_fail",
"command_id": "cmd-gift-im-fail",
"pool_id": "super_lucky",
"user_id": 42,
"room_id": "room-1",
"gift_id": "rose",
"gift_count": 1,
"coin_spent": 100,
"effective_reward_coins": 300,
"multiplier_ppm": 3000000,
"created_at_ms": 1760000000000,
})
repository := &fakeLuckyGiftRepository{
outbox: []domain.DrawOutbox{{
AppCode: "lalu",
OutboxID: "lucky_lucky_draw_im_fail",
EventType: "LuckyGiftDrawn",
PayloadJSON: string(payload),
}},
}
wallet := &fakeLuckyGiftWallet{}
publisher := &fakeLuckyGiftPublisher{err: errors.New("im unavailable")}
service := New(repository, WithWallet(wallet), WithRoomPublisher(publisher), WithRegionBroadcaster(&fakeLuckyGiftRegionBroadcaster{}))
service.SetClock(func() time.Time { return time.UnixMilli(1760000001000).UTC() })
processed, err := service.ProcessPendingDrawOutbox(context.Background(), WorkerOptions{WorkerID: "worker-1"})
if err != nil {
t.Fatalf("ProcessPendingDrawOutbox should schedule retry without failing batch: %v", err)
}
if processed != 1 {
t.Fatalf("processed count mismatch: %d", processed)
}
if repository.grantedDrawID != "lucky_draw_im_fail" {
t.Fatalf("wallet-granted draw should be settled before IM retry, got %q", repository.grantedDrawID)
}
if repository.deliveredOutboxID != "" {
t.Fatalf("outbox must not be delivered while IM failed")
}
if repository.retryableOutboxID != "lucky_lucky_draw_im_fail" {
t.Fatalf("IM failure should keep outbox retryable, got %q", repository.retryableOutboxID)
}
}
type fakeLuckyGiftRepository struct {
getConfigCalls int
upserted domain.RuleConfig
@ -165,6 +216,7 @@ type fakeLuckyGiftRepository struct {
grantedDrawID string
grantedDrawIDs []string
deliveredOutboxID string
retryableOutboxID string
}
func (r *fakeLuckyGiftRepository) GetLuckyGiftRuleConfig(context.Context, string) (domain.RuleConfig, bool, error) {
@ -206,7 +258,8 @@ func (r *fakeLuckyGiftRepository) MarkLuckyGiftOutboxDelivered(_ context.Context
return nil
}
func (r *fakeLuckyGiftRepository) MarkLuckyGiftOutboxRetryable(context.Context, domain.DrawOutbox, int, int64, string, int64) error {
func (r *fakeLuckyGiftRepository) MarkLuckyGiftOutboxRetryable(_ context.Context, event domain.DrawOutbox, _ int, _ int64, _ string, _ int64) error {
r.retryableOutboxID = event.OutboxID
return nil
}
@ -246,11 +299,12 @@ func (w *fakeLuckyGiftWallet) CreditLuckyGiftReward(_ context.Context, req *wall
type fakeLuckyGiftPublisher struct {
last tencentim.CustomGroupMessage
err error
}
func (p *fakeLuckyGiftPublisher) PublishGroupCustomMessage(_ context.Context, message tencentim.CustomGroupMessage) error {
p.last = message
return nil
return p.err
}
type fakeLuckyGiftRegionBroadcaster struct {

319
services/activity-service/internal/storage/mysql/lucky_gift_repository.go
View File

@ -576,6 +576,9 @@ func (r *Repository) persistLuckyBatchDraw(ctx context.Context, tx *sql.Tx, appC
if err := r.insertLuckyDrawRecords(ctx, tx, state.Records); err != nil {
return err
}
if err := r.incrementLuckyDrawPoolStats(ctx, tx, state.Records, nowMS); err != nil {
return err
}
if state.NeedsOutbox {
// 只有本批存在钱包/IM/反馈副作用时才产生 outbox；纯 0 倍且无反馈的批次直接终态 granted。
return r.insertLuckyAggregateDrawOutbox(ctx, tx, appCode, cmd, state, nowMS)
@ -653,6 +656,143 @@ func (r *Repository) insertLuckyDrawRecords(ctx context.Context, tx *sql.Tx, rec
return nil
}
type luckyDrawPoolStatKey struct {
AppCode string
PoolID string
GiftID string
}
type luckyDrawPoolStatDelta struct {
TotalDraws int64
TotalSpentCoins int64
TotalRewardCoins int64
BaseRewardCoins int64
RoomAtmosphereRewardCoins int64
ActivitySubsidyCoins int64
PendingDraws int64
GrantedDraws int64
FailedDraws int64
UserIDs map[int64]struct{}
RoomIDs map[string]struct{}
}
func (r *Repository) incrementLuckyDrawPoolStats(ctx context.Context, tx *sql.Tx, records []luckyDrawRecordInput, nowMS int64) error {
deltas := map[luckyDrawPoolStatKey]*luckyDrawPoolStatDelta{}
for _, record := range records {
poolID := luckyPoolID(record.Config.GiftID)
giftID := strings.TrimSpace(record.Command.GiftID)
keys := []luckyDrawPoolStatKey{
{AppCode: record.AppCode, PoolID: poolID, GiftID: ""},
}
if giftID != "" {
keys = append(keys, luckyDrawPoolStatKey{AppCode: record.AppCode, PoolID: poolID, GiftID: giftID})
}
for _, key := range keys {
delta := deltas[key]
if delta == nil {
delta = &luckyDrawPoolStatDelta{UserIDs: map[int64]struct{}{}, RoomIDs: map[string]struct{}{}}
deltas[key] = delta
}
delta.TotalDraws++
delta.TotalSpentCoins += record.Command.CoinSpent
delta.TotalRewardCoins += record.Candidate.effectiveReward()
delta.BaseRewardCoins += record.Candidate.BaseReward
delta.RoomAtmosphereRewardCoins += record.Candidate.RoomReward
delta.ActivitySubsidyCoins += record.Candidate.ActivityReward
switch record.RewardStatus {
case domain.StatusPending:
delta.PendingDraws++
case domain.StatusGranted:
delta.GrantedDraws++
case domain.StatusFailed:
delta.FailedDraws++
}
if record.Command.UserID > 0 {
delta.UserIDs[record.Command.UserID] = struct{}{}
}
roomID := strings.TrimSpace(record.Command.RoomID)
if roomID != "" {
delta.RoomIDs[roomID] = struct{}{}
}
}
}
for key, delta := range deltas {
uniqueUsers, err := r.insertLuckyDrawStatUsers(ctx, tx, key, delta.UserIDs, nowMS)
if err != nil {
return err
}
uniqueRooms, err := r.insertLuckyDrawStatRooms(ctx, tx, key, delta.RoomIDs, nowMS)
if err != nil {
return err
}
if err := r.upsertLuckyDrawPoolStatDelta(ctx, tx, key, *delta, uniqueUsers, uniqueRooms, nowMS); err != nil {
return err
}
}
return nil
}
func (r *Repository) insertLuckyDrawStatUsers(ctx context.Context, tx *sql.Tx, key luckyDrawPoolStatKey, userIDs map[int64]struct{}, nowMS int64) (int64, error) {
var inserted int64
for userID := range userIDs {
result, err := tx.ExecContext(ctx, `
INSERT IGNORE INTO lucky_draw_pool_stat_users (app_code, pool_id, gift_id, user_id, created_at_ms)
VALUES (?, ?, ?, ?, ?)`,
key.AppCode, key.PoolID, key.GiftID, userID, nowMS,
)
if err != nil {
return 0, err
}
rows, _ := result.RowsAffected()
inserted += rows
}
return inserted, nil
}
func (r *Repository) insertLuckyDrawStatRooms(ctx context.Context, tx *sql.Tx, key luckyDrawPoolStatKey, roomIDs map[string]struct{}, nowMS int64) (int64, error) {
var inserted int64
for roomID := range roomIDs {
result, err := tx.ExecContext(ctx, `
INSERT IGNORE INTO lucky_draw_pool_stat_rooms (app_code, pool_id, gift_id, room_id, created_at_ms)
VALUES (?, ?, ?, ?, ?)`,
key.AppCode, key.PoolID, key.GiftID, roomID, nowMS,
)
if err != nil {
return 0, err
}
rows, _ := result.RowsAffected()
inserted += rows
}
return inserted, nil
}
func (r *Repository) upsertLuckyDrawPoolStatDelta(ctx context.Context, tx *sql.Tx, key luckyDrawPoolStatKey, delta luckyDrawPoolStatDelta, uniqueUsers int64, uniqueRooms int64, nowMS int64) error {
_, err := tx.ExecContext(ctx, `
INSERT INTO lucky_draw_pool_stats (
app_code, pool_id, gift_id, total_draws, unique_users, unique_rooms, total_spent_coins,
total_reward_coins, base_reward_coins, room_atmosphere_reward_coins, activity_subsidy_coins,
pending_draws, granted_draws, failed_draws, created_at_ms, updated_at_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
total_draws = total_draws + VALUES(total_draws),
unique_users = unique_users + VALUES(unique_users),
unique_rooms = unique_rooms + VALUES(unique_rooms),
total_spent_coins = total_spent_coins + VALUES(total_spent_coins),
total_reward_coins = total_reward_coins + VALUES(total_reward_coins),
base_reward_coins = base_reward_coins + VALUES(base_reward_coins),
room_atmosphere_reward_coins = room_atmosphere_reward_coins + VALUES(room_atmosphere_reward_coins),
activity_subsidy_coins = activity_subsidy_coins + VALUES(activity_subsidy_coins),
pending_draws = pending_draws + VALUES(pending_draws),
granted_draws = granted_draws + VALUES(granted_draws),
failed_draws = failed_draws + VALUES(failed_draws),
updated_at_ms = VALUES(updated_at_ms)`,
key.AppCode, key.PoolID, key.GiftID, delta.TotalDraws, uniqueUsers, uniqueRooms, delta.TotalSpentCoins,
delta.TotalRewardCoins, delta.BaseRewardCoins, delta.RoomAtmosphereRewardCoins, delta.ActivitySubsidyCoins,
delta.PendingDraws, delta.GrantedDraws, delta.FailedDraws, nowMS, nowMS,
)
return err
}
// luckyDrawRecordArgs 固定 lucky_draw_records 的列顺序；集中维护可以避免批量 INSERT 和单条 INSERT 语义漂移。
func luckyDrawRecordArgs(input luckyDrawRecordInput) []any {
candidateSnapshot, poolSnapshot, rtpSnapshot := luckyDrawRecordSnapshots(input)
@ -1132,6 +1272,27 @@ func (r *Repository) applyLuckyDraw(ctx context.Context, tx *sql.Tx, input lucky
); err != nil {
return err
}
if err := r.incrementLuckyDrawPoolStats(ctx, tx, []luckyDrawRecordInput{{
AppCode: input.AppCode,
DrawID: input.DrawID,
Command: input.Command,
Config: input.Config,
ExperiencePool: input.ExperiencePool,
Candidate: input.Candidate,
Limited: input.Limited,
StageFeedback: input.StageFeedback,
GlobalWindow: input.GlobalWindow,
GiftWindow: input.GiftWindow,
PlatformPool: input.PlatformPool,
RoomPool: input.RoomPool,
GiftPool: input.GiftPool,
Atmosphere: input.Atmosphere,
BudgetSourcesJSON: input.BudgetSourcesJSON,
RewardStatus: input.RewardStatus,
NowMS: input.NowMS,
}}, input.NowMS); err != nil {
return err
}
if luckyDrawNeedsOutbox(input.Candidate, input.StageFeedback) {
// outbox 是钱包入账和房间 IM 的唯一异步出口；事务内只写事实，不调用外部 RPC。
payload, _ := json.Marshal(map[string]any{
@ -1326,6 +1487,15 @@ func (r *Repository) markLuckyGiftDraws(ctx context.Context, appCode string, dra
end = len(drawIDs)
}
chunk := drawIDs[start:end]
tx, err := r.db.BeginTx(ctx, nil)
if err != nil {
return err
}
deltas, err := r.selectLuckyDrawStatusDeltas(ctx, tx, appCode, chunk, status, onlyPending)
if err != nil {
_ = tx.Rollback()
return err
}
args := []any{status, walletTransactionID, failureReason, nowMS, appCode}
for _, drawID := range chunk {
args = append(args, drawID)
@ -1333,18 +1503,117 @@ func (r *Repository) markLuckyGiftDraws(ctx context.Context, appCode string, dra
query := `
UPDATE lucky_draw_records
SET reward_status = ?, reward_transaction_id = ?, reward_failure_reason = ?, updated_at_ms = ?
WHERE app_code = ? AND draw_id IN (` + luckySQLPlaceholders(len(chunk)) + `)`
WHERE app_code = ? AND draw_id IN (` + luckySQLPlaceholders(len(chunk)) + `) AND reward_status <> ?`
args = append(args, status)
if onlyPending {
query += ` AND reward_status = ?`
args = append(args, domain.StatusPending)
}
if _, err := r.db.ExecContext(ctx, query, args...); err != nil {
if _, err := tx.ExecContext(ctx, query, args...); err != nil {
_ = tx.Rollback()
return err
}
if err := r.applyLuckyDrawStatusDeltas(ctx, tx, appCode, deltas, status, nowMS); err != nil {
_ = tx.Rollback()
return err
}
if err := tx.Commit(); err != nil {
return err
}
}
return nil
}
type luckyDrawStatusDelta struct {
PoolID string
GiftID string
OldStatus string
Count int64
}
func (r *Repository) selectLuckyDrawStatusDeltas(ctx context.Context, tx *sql.Tx, appCode string, drawIDs []string, targetStatus string, onlyPending bool) ([]luckyDrawStatusDelta, error) {
args := []any{appCode}
for _, drawID := range drawIDs {
args = append(args, drawID)
}
args = append(args, targetStatus)
query := `
SELECT pool_id, gift_id, reward_status, COUNT(*)
FROM lucky_draw_records
WHERE app_code = ? AND draw_id IN (` + luckySQLPlaceholders(len(drawIDs)) + `) AND reward_status <> ?`
if onlyPending {
query += ` AND reward_status = ?`
args = append(args, domain.StatusPending)
}
query += ` GROUP BY pool_id, gift_id, reward_status`
rows, err := tx.QueryContext(ctx, query, args...)
if err != nil {
return nil, err
}
defer rows.Close()
deltas := []luckyDrawStatusDelta{}
for rows.Next() {
var delta luckyDrawStatusDelta
if err := rows.Scan(&delta.PoolID, &delta.GiftID, &delta.OldStatus, &delta.Count); err != nil {
return nil, err
}
deltas = append(deltas, delta)
}
return deltas, rows.Err()
}
func (r *Repository) applyLuckyDrawStatusDeltas(ctx context.Context, tx *sql.Tx, appCode string, deltas []luckyDrawStatusDelta, targetStatus string, nowMS int64) error {
for _, delta := range deltas {
keys := []luckyDrawPoolStatKey{{AppCode: appCode, PoolID: delta.PoolID, GiftID: ""}}
if strings.TrimSpace(delta.GiftID) != "" {
keys = append(keys, luckyDrawPoolStatKey{AppCode: appCode, PoolID: delta.PoolID, GiftID: delta.GiftID})
}
for _, key := range keys {
if err := r.applyLuckyDrawStatusDelta(ctx, tx, key, delta.OldStatus, targetStatus, delta.Count, nowMS); err != nil {
return err
}
}
}
return nil
}
func (r *Repository) applyLuckyDrawStatusDelta(ctx context.Context, tx *sql.Tx, key luckyDrawPoolStatKey, oldStatus string, targetStatus string, count int64, nowMS int64) error {
pendingDelta, grantedDelta, failedDelta := luckyStatusCounterDelta(oldStatus, targetStatus, count)
if pendingDelta == 0 && grantedDelta == 0 && failedDelta == 0 {
return nil
}
_, err := tx.ExecContext(ctx, `
UPDATE lucky_draw_pool_stats
SET pending_draws = pending_draws + ?,
granted_draws = granted_draws + ?,
failed_draws = failed_draws + ?,
updated_at_ms = ?
WHERE app_code = ? AND pool_id = ? AND gift_id = ?`,
pendingDelta, grantedDelta, failedDelta, nowMS, key.AppCode, key.PoolID, key.GiftID,
)
return err
}
func luckyStatusCounterDelta(oldStatus string, targetStatus string, count int64) (pending int64, granted int64, failed int64) {
switch oldStatus {
case domain.StatusPending:
pending -= count
case domain.StatusGranted:
granted -= count
case domain.StatusFailed:
failed -= count
}
switch targetStatus {
case domain.StatusPending:
pending += count
case domain.StatusGranted:
granted += count
case domain.StatusFailed:
failed += count
}
return pending, granted, failed
}
// normalizeLuckyDrawIDs 去掉空值和重复值；outbox payload 是补偿输入，必须先清洗再拼 SQL IN。
func normalizeLuckyDrawIDs(drawIDs []string) []string {
seen := map[string]bool{}
@ -1421,6 +1690,13 @@ func (r *Repository) GetLuckyGiftDrawSummary(ctx context.Context, query domain.D
if r == nil || r.db == nil {
return domain.DrawSummary{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
}
if luckyDrawSummaryCanUseStats(query) {
return r.getLuckyGiftDrawSummaryFromStats(ctx, query)
}
return r.getLuckyGiftDrawSummaryFromRecords(ctx, query)
}
func (r *Repository) getLuckyGiftDrawSummaryFromRecords(ctx context.Context, query domain.DrawQuery) (domain.DrawSummary, error) {
whereSQL, args := luckyDrawWhereClause(appcode.FromContext(ctx), query)
var summary domain.DrawSummary
summary.PoolID = query.PoolID
@ -1459,6 +1735,45 @@ func (r *Repository) GetLuckyGiftDrawSummary(ctx context.Context, query domain.D
return summary, nil
}
func luckyDrawSummaryCanUseStats(query domain.DrawQuery) bool {
return query.UserID <= 0 && strings.TrimSpace(query.RoomID) == "" && strings.TrimSpace(query.Status) == ""
}
func (r *Repository) getLuckyGiftDrawSummaryFromStats(ctx context.Context, query domain.DrawQuery) (domain.DrawSummary, error) {
var summary domain.DrawSummary
summary.PoolID = query.PoolID
err := r.db.QueryRowContext(ctx, `
SELECT total_draws, unique_users, unique_rooms, total_spent_coins, total_reward_coins,
base_reward_coins, room_atmosphere_reward_coins, activity_subsidy_coins,
pending_draws, granted_draws, failed_draws
FROM lucky_draw_pool_stats
WHERE app_code = ? AND pool_id = ? AND gift_id = ?`,
appcode.FromContext(ctx), query.PoolID, strings.TrimSpace(query.GiftID),
).Scan(
&summary.TotalDraws,
&summary.UniqueUsers,
&summary.UniqueRooms,
&summary.TotalSpentCoins,
&summary.TotalRewardCoins,
&summary.BaseRewardCoins,
&summary.RoomAtmosphereRewardCoins,
&summary.ActivitySubsidyCoins,
&summary.PendingDraws,
&summary.GrantedDraws,
&summary.FailedDraws,
)
if errors.Is(err, sql.ErrNoRows) {
return r.getLuckyGiftDrawSummaryFromRecords(ctx, query)
}
if err != nil {
return domain.DrawSummary{}, err
}
if summary.TotalSpentCoins > 0 {
summary.ActualRTPPPM = summary.TotalRewardCoins * luckyPPMScale / summary.TotalSpentCoins
}
return summary, nil
}
func luckyDrawWhereClause(appCode string, query domain.DrawQuery) (string, []any) {
where := []string{"app_code = ?"}
args := []any{appCode}

143
services/activity-service/internal/storage/mysql/repository.go
View File

@ -51,6 +51,9 @@ func (r *Repository) Migrate(ctx context.Context) error {
if err := r.ensureLuckyDrawRewardSettlementColumns(ctx); err != nil {
return err
}
if err := r.ensureLuckyDrawPoolStatsTables(ctx); err != nil {
return err
}
if err := r.ensureLuckyGiftRuleVersionTables(ctx); err != nil {
return err
}
@ -63,6 +66,146 @@ func (r *Repository) Migrate(ctx context.Context) error {
return nil
}
func (r *Repository) ensureLuckyDrawPoolStatsTables(ctx context.Context) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS lucky_draw_pool_stats (
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码',
pool_id VARCHAR(96) NOT NULL COMMENT '幸运礼物奖池 ID',
gift_id VARCHAR(96) NOT NULL DEFAULT '' COMMENT '礼物 ID，空值表示奖池汇总',
total_draws BIGINT NOT NULL DEFAULT 0 COMMENT '抽奖次数',
unique_users BIGINT NOT NULL DEFAULT 0 COMMENT '参与用户数',
unique_rooms BIGINT NOT NULL DEFAULT 0 COMMENT '参与房间数',
total_spent_coins BIGINT NOT NULL DEFAULT 0 COMMENT '消耗金币',
total_reward_coins BIGINT NOT NULL DEFAULT 0 COMMENT '返还金币',
base_reward_coins BIGINT NOT NULL DEFAULT 0 COMMENT '基础 RTP 返奖',
room_atmosphere_reward_coins BIGINT NOT NULL DEFAULT 0 COMMENT '房间气氛支出',
activity_subsidy_coins BIGINT NOT NULL DEFAULT 0 COMMENT '活动补贴支出',
pending_draws BIGINT NOT NULL DEFAULT 0 COMMENT '待发放数量',
granted_draws BIGINT NOT NULL DEFAULT 0 COMMENT '已发放数量',
failed_draws BIGINT NOT NULL DEFAULT 0 COMMENT '发放失败数量',
created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms',
updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms',
PRIMARY KEY (app_code, pool_id, gift_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物池级汇总统计'`,
`CREATE TABLE IF NOT EXISTS lucky_draw_pool_stat_users (
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码',
pool_id VARCHAR(96) NOT NULL COMMENT '幸运礼物奖池 ID',
gift_id VARCHAR(96) NOT NULL DEFAULT '' COMMENT '礼物 ID，空值表示奖池汇总',
user_id BIGINT NOT NULL COMMENT '用户 ID',
created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms',
PRIMARY KEY (app_code, pool_id, gift_id, user_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物汇总用户去重'`,
`CREATE TABLE IF NOT EXISTS lucky_draw_pool_stat_rooms (
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码',
pool_id VARCHAR(96) NOT NULL COMMENT '幸运礼物奖池 ID',
gift_id VARCHAR(96) NOT NULL DEFAULT '' COMMENT '礼物 ID，空值表示奖池汇总',
room_id VARCHAR(96) NOT NULL COMMENT '房间 ID',
created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms',
PRIMARY KEY (app_code, pool_id, gift_id, room_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物汇总房间去重'`,
}
for _, statement := range statements {
if _, err := r.db.ExecContext(ctx, statement); err != nil {
return err
}
}
return r.backfillLuckyDrawPoolStatsIfEmpty(ctx)
}
func (r *Repository) backfillLuckyDrawPoolStatsIfEmpty(ctx context.Context) error {
var existing int64
if err := r.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM lucky_draw_pool_stats LIMIT 1`).Scan(&existing); err != nil {
return err
}
if existing > 0 {
return nil
}
return r.backfillLuckyDrawPoolStats(ctx)
}
func (r *Repository) backfillLuckyDrawPoolStats(ctx context.Context) error {
statements := []string{
`INSERT IGNORE INTO lucky_draw_pool_stat_users (app_code, pool_id, gift_id, user_id, created_at_ms)
SELECT app_code, pool_id, '', user_id, MIN(created_at_ms)
FROM lucky_draw_records
GROUP BY app_code, pool_id, user_id`,
`INSERT IGNORE INTO lucky_draw_pool_stat_users (app_code, pool_id, gift_id, user_id, created_at_ms)
SELECT app_code, pool_id, gift_id, user_id, MIN(created_at_ms)
FROM lucky_draw_records
GROUP BY app_code, pool_id, gift_id, user_id`,
`INSERT IGNORE INTO lucky_draw_pool_stat_rooms (app_code, pool_id, gift_id, room_id, created_at_ms)
SELECT app_code, pool_id, '', room_id, MIN(created_at_ms)
FROM lucky_draw_records
GROUP BY app_code, pool_id, room_id`,
`INSERT IGNORE INTO lucky_draw_pool_stat_rooms (app_code, pool_id, gift_id, room_id, created_at_ms)
SELECT app_code, pool_id, gift_id, room_id, MIN(created_at_ms)
FROM lucky_draw_records
GROUP BY app_code, pool_id, gift_id, room_id`,
`INSERT INTO lucky_draw_pool_stats (
app_code, pool_id, gift_id, total_draws, unique_users, unique_rooms, total_spent_coins,
total_reward_coins, base_reward_coins, room_atmosphere_reward_coins, activity_subsidy_coins,
pending_draws, granted_draws, failed_draws, created_at_ms, updated_at_ms
)
SELECT app_code, pool_id, '', COUNT(*), COUNT(DISTINCT user_id), COUNT(DISTINCT room_id),
COALESCE(SUM(coin_spent), 0), COALESCE(SUM(effective_reward_coins), 0),
COALESCE(SUM(base_reward_coins), 0), COALESCE(SUM(room_atmosphere_reward_coins), 0),
COALESCE(SUM(activity_subsidy_coins), 0),
COALESCE(SUM(CASE WHEN reward_status = 'pending' THEN 1 ELSE 0 END), 0),
COALESCE(SUM(CASE WHEN reward_status = 'granted' THEN 1 ELSE 0 END), 0),
COALESCE(SUM(CASE WHEN reward_status = 'failed' THEN 1 ELSE 0 END), 0),
MIN(created_at_ms), MAX(updated_at_ms)
FROM lucky_draw_records
GROUP BY app_code, pool_id
ON DUPLICATE KEY UPDATE
total_draws = VALUES(total_draws),
unique_users = VALUES(unique_users),
unique_rooms = VALUES(unique_rooms),
total_spent_coins = VALUES(total_spent_coins),
total_reward_coins = VALUES(total_reward_coins),
base_reward_coins = VALUES(base_reward_coins),
room_atmosphere_reward_coins = VALUES(room_atmosphere_reward_coins),
activity_subsidy_coins = VALUES(activity_subsidy_coins),
pending_draws = VALUES(pending_draws),
granted_draws = VALUES(granted_draws),
failed_draws = VALUES(failed_draws),
updated_at_ms = VALUES(updated_at_ms)`,
`INSERT INTO lucky_draw_pool_stats (
app_code, pool_id, gift_id, total_draws, unique_users, unique_rooms, total_spent_coins,
total_reward_coins, base_reward_coins, room_atmosphere_reward_coins, activity_subsidy_coins,
pending_draws, granted_draws, failed_draws, created_at_ms, updated_at_ms
)
SELECT app_code, pool_id, gift_id, COUNT(*), COUNT(DISTINCT user_id), COUNT(DISTINCT room_id),
COALESCE(SUM(coin_spent), 0), COALESCE(SUM(effective_reward_coins), 0),
COALESCE(SUM(base_reward_coins), 0), COALESCE(SUM(room_atmosphere_reward_coins), 0),
COALESCE(SUM(activity_subsidy_coins), 0),
COALESCE(SUM(CASE WHEN reward_status = 'pending' THEN 1 ELSE 0 END), 0),
COALESCE(SUM(CASE WHEN reward_status = 'granted' THEN 1 ELSE 0 END), 0),
COALESCE(SUM(CASE WHEN reward_status = 'failed' THEN 1 ELSE 0 END), 0),
MIN(created_at_ms), MAX(updated_at_ms)
FROM lucky_draw_records
GROUP BY app_code, pool_id, gift_id
ON DUPLICATE KEY UPDATE
total_draws = VALUES(total_draws),
unique_users = VALUES(unique_users),
unique_rooms = VALUES(unique_rooms),
total_spent_coins = VALUES(total_spent_coins),
total_reward_coins = VALUES(total_reward_coins),
base_reward_coins = VALUES(base_reward_coins),
room_atmosphere_reward_coins = VALUES(room_atmosphere_reward_coins),
activity_subsidy_coins = VALUES(activity_subsidy_coins),
pending_draws = VALUES(pending_draws),
granted_draws = VALUES(granted_draws),
failed_draws = VALUES(failed_draws),
updated_at_ms = VALUES(updated_at_ms)`,
}
for _, statement := range statements {
if _, err := r.db.ExecContext(ctx, statement); err != nil {
return err
}
}
return nil
}
func (r *Repository) ensureLuckyUserStateFlowColumns(ctx context.Context) error {
// 用户阶段改按金币流水折算等价抽数，状态表必须持久保存累计流水和连续未中奖次数，不能再只依赖 paid_draws。
additions := []struct {

2
services/game-service/deploy/mysql/initdb/001_game_service.sql
View File

@ -67,6 +67,7 @@ CREATE TABLE IF NOT EXISTS game_launch_sessions (
user_id BIGINT NOT NULL COMMENT '用户 ID',
display_user_id VARCHAR(32) NOT NULL COMMENT '展示用户 ID',
room_id VARCHAR(96) NOT NULL DEFAULT '' COMMENT '房间 ID',
region_id BIGINT NOT NULL DEFAULT 0 COMMENT '用户启动游戏时的区域 ID',
scene VARCHAR(64) NOT NULL COMMENT '场景',
platform_code VARCHAR(64) NOT NULL COMMENT '平台编码',
game_id VARCHAR(96) NOT NULL COMMENT '游戏 ID',
@ -91,6 +92,7 @@ CREATE TABLE IF NOT EXISTS game_orders (
provider_game_id VARCHAR(128) NOT NULL COMMENT '提供方游戏 ID',
user_id BIGINT NOT NULL COMMENT '用户 ID',
room_id VARCHAR(96) NOT NULL DEFAULT '' COMMENT '房间 ID',
region_id BIGINT NOT NULL DEFAULT 0 COMMENT '用户启动游戏时的区域 ID',
op_type VARCHAR(32) NOT NULL COMMENT 'op类型',
coin_amount BIGINT NOT NULL COMMENT '金币数量',
status VARCHAR(32) NOT NULL COMMENT '业务状态',

2
services/game-service/internal/domain/game/game.go
View File

@ -112,6 +112,7 @@ type LaunchSession struct {
// DisplayUserID 用于对外 uid_mode，避免直接暴露内部数字 ID。
DisplayUserID string
RoomID string
RegionID int64
Scene string
PlatformCode string
GameID string
@ -137,6 +138,7 @@ type GameOrder struct {
ProviderGameID string
UserID int64
RoomID string
RegionID int64
// OpType 只允许钱包语义 debit/credit，厂商枚举在适配器层转换。
OpType string
CoinAmount int64

34
services/game-service/internal/service/game/service.go
View File

@ -107,6 +107,7 @@ type LaunchCommand struct {
GameID string
Scene string
RoomID string
RegionID int64
AccessToken string
}
@ -188,6 +189,14 @@ func (s *Service) LaunchGame(ctx context.Context, command LaunchCommand) (Launch
if game.PlatformStatus != gamedomain.StatusActive || game.Status != gamedomain.StatusActive {
return LaunchResult{}, xerr.New(xerr.Conflict, "game is not launchable")
}
regionID := command.RegionID
if regionID <= 0 {
var err error
regionID, err = s.resolveUserRegion(ctx, command.AppCode, command.RequestID, command.UserID)
if err != nil {
return LaunchResult{}, err
}
}
now := s.now()
sessionID := "game_sess_" + strconv.FormatInt(now.UnixMilli(), 10) + "_" + randomHex(6)
@ -216,6 +225,7 @@ func (s *Service) LaunchGame(ctx context.Context, command LaunchCommand) (Launch
UserID: command.UserID,
DisplayUserID: command.DisplayUserID,
RoomID: strings.TrimSpace(command.RoomID),
RegionID: regionID,
Scene: command.Scene,
PlatformCode: game.PlatformCode,
GameID: game.GameID,
@ -409,6 +419,25 @@ func (s *Service) getCallbackUserInfo(ctx context.Context, app string, req *game
}, nil
}
func (s *Service) resolveUserRegion(ctx context.Context, app string, requestID string, userID int64) (int64, error) {
if userID <= 0 || s.user == nil {
return 0, nil
}
userResp, err := s.user.GetUser(ctx, &userv1.GetUserRequest{
Meta: &userv1.RequestMeta{
RequestId: requestID,
Caller: "game-service",
SentAtMs: s.now().UnixMilli(),
AppCode: app,
},
UserId: userID,
})
if err != nil {
return 0, err
}
return userResp.GetUser().GetRegionId(), nil
}
func (s *Service) applyCallbackCoinChange(ctx context.Context, app string, req *gamev1.CallbackRequest, requestHash string) (map[string]any, error) {
if s.wallet == nil {
return nil, xerr.New(xerr.Unavailable, "wallet client is not configured")
@ -439,6 +468,10 @@ func (s *Service) applyCallbackCoinChange(ctx context.Context, app string, req *
}
body.ProviderGameID = game.ProviderGameID
}
regionID, err := s.resolveUserRegion(ctx, app, req.GetMeta().GetRequestId(), body.UserID)
if err != nil {
return nil, err
}
order := gamedomain.GameOrder{
AppCode: app,
OrderID: "gord_" + stableHash(app+"|"+req.GetPlatformCode()+"|"+body.ProviderOrderID),
@ -449,6 +482,7 @@ func (s *Service) applyCallbackCoinChange(ctx context.Context, app string, req *
ProviderGameID: body.ProviderGameID,
UserID: body.UserID,
RoomID: strings.TrimSpace(body.RoomID),
RegionID: regionID,
OpType: body.OpType,
CoinAmount: body.CoinAmount,
Status: gamedomain.OrderStatusWalletApplying,

2
services/game-service/internal/service/game/service_test.go
View File

@ -57,7 +57,7 @@ func TestLaunchGameCreatesSessionAndRejectsMaintenance(t *testing.T) {
if result.SessionID == "" || result.ExpiresAtMS != 1700000900000 || !strings.Contains(result.LaunchURL, "session_token=") {
t.Fatalf("launch result mismatch: %+v", result)
}
if repo.session.GameID != "demo_rocket_001" || repo.session.UserID != 42 || repo.session.LaunchTokenHash == "" {
if repo.session.GameID != "demo_rocket_001" || repo.session.UserID != 42 || repo.session.RegionID != 100 || repo.session.LaunchTokenHash == "" {
t.Fatalf("launch session was not persisted: %+v", repo.session)
}

1
services/game-service/internal/service/game/yomi_adapter.go
View File

@ -308,6 +308,7 @@ func (s *Service) applyYomiCoinChange(ctx context.Context, app string, req *game
ProviderGameID: session.ProviderGameID,
UserID: session.UserID,
RoomID: strings.TrimSpace(firstNonEmpty(roomID, session.RoomID)),
RegionID: session.RegionID,
OpType: opType,
CoinAmount: amount,
Status: gamedomain.OrderStatusWalletApplying,

38
services/game-service/internal/storage/mysql/repository.go
View File

@ -118,6 +118,7 @@ func (r *Repository) Migrate(ctx context.Context) error {
user_id BIGINT NOT NULL,
display_user_id VARCHAR(32) NOT NULL,
room_id VARCHAR(96) NOT NULL DEFAULT '',
region_id BIGINT NOT NULL DEFAULT 0,
scene VARCHAR(64) NOT NULL,
platform_code VARCHAR(64) NOT NULL,
game_id VARCHAR(96) NOT NULL,
@ -141,6 +142,7 @@ func (r *Repository) Migrate(ctx context.Context) error {
provider_game_id VARCHAR(128) NOT NULL,
user_id BIGINT NOT NULL,
room_id VARCHAR(96) NOT NULL DEFAULT '',
region_id BIGINT NOT NULL DEFAULT 0,
op_type VARCHAR(32) NOT NULL,
coin_amount BIGINT NOT NULL,
status VARCHAR(32) NOT NULL,
@ -241,6 +243,12 @@ func (r *Repository) Migrate(ctx context.Context) error {
if err := r.ensureColumn(ctx, "game_platforms", "adapter_type", "adapter_type VARCHAR(64) NOT NULL DEFAULT 'demo' AFTER api_base_url"); err != nil {
return err
}
if err := r.ensureColumn(ctx, "game_launch_sessions", "region_id", "region_id BIGINT NOT NULL DEFAULT 0 AFTER room_id"); err != nil {
return err
}
if err := r.ensureColumn(ctx, "game_orders", "region_id", "region_id BIGINT NOT NULL DEFAULT 0 AFTER room_id"); err != nil {
return err
}
return r.seedDefaults(ctx)
}
@ -380,11 +388,11 @@ func (r *Repository) GetPlatform(ctx context.Context, appCode string, platformCo
func (r *Repository) CreateLaunchSession(ctx context.Context, session gamedomain.LaunchSession) error {
_, err := r.db.ExecContext(ctx,
`INSERT INTO game_launch_sessions (
app_code, session_id, user_id, display_user_id, room_id, scene, platform_code, game_id,
app_code, session_id, user_id, display_user_id, room_id, region_id, scene, platform_code, game_id,
provider_game_id, launch_token_hash, status, expires_at_ms, created_at_ms, updated_at_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
appcode.Normalize(session.AppCode), session.SessionID, session.UserID, session.DisplayUserID, session.RoomID, session.Scene,
session.PlatformCode, session.GameID, session.ProviderGameID, session.LaunchTokenHash, session.Status,
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
appcode.Normalize(session.AppCode), session.SessionID, session.UserID, session.DisplayUserID, session.RoomID, session.RegionID,
session.Scene, session.PlatformCode, session.GameID, session.ProviderGameID, session.LaunchTokenHash, session.Status,
session.ExpiresAtMS, session.CreatedAtMS, session.UpdatedAtMS,
)
return err
@ -392,14 +400,14 @@ func (r *Repository) CreateLaunchSession(ctx context.Context, session gamedomain
func (r *Repository) GetLaunchSessionByToken(ctx context.Context, appCode string, token string) (gamedomain.LaunchSession, error) {
row := r.db.QueryRowContext(ctx,
`SELECT app_code, session_id, user_id, display_user_id, room_id, scene, platform_code, game_id,
`SELECT app_code, session_id, user_id, display_user_id, room_id, region_id, scene, platform_code, game_id,
provider_game_id, launch_token_hash, status, expires_at_ms, created_at_ms, updated_at_ms
FROM game_launch_sessions
WHERE app_code = ? AND launch_token_hash = ?`,
appcode.Normalize(appCode), stableHash(strings.TrimSpace(token)),
)
var item gamedomain.LaunchSession
if err := row.Scan(&item.AppCode, &item.SessionID, &item.UserID, &item.DisplayUserID, &item.RoomID, &item.Scene, &item.PlatformCode, &item.GameID, &item.ProviderGameID, &item.LaunchTokenHash, &item.Status, &item.ExpiresAtMS, &item.CreatedAtMS, &item.UpdatedAtMS); err != nil {
if err := row.Scan(&item.AppCode, &item.SessionID, &item.UserID, &item.DisplayUserID, &item.RoomID, &item.RegionID, &item.Scene, &item.PlatformCode, &item.GameID, &item.ProviderGameID, &item.LaunchTokenHash, &item.Status, &item.ExpiresAtMS, &item.CreatedAtMS, &item.UpdatedAtMS); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return gamedomain.LaunchSession{}, xerr.New(xerr.Unauthorized, "game session token invalid")
}
@ -454,10 +462,10 @@ func (r *Repository) CreateGameOrder(ctx context.Context, order gamedomain.GameO
if _, err := tx.ExecContext(ctx,
`INSERT INTO game_orders (
app_code, order_id, platform_code, provider_order_id, provider_round_id, game_id, provider_game_id,
user_id, room_id, op_type, coin_amount, status, request_hash, created_at_ms, updated_at_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
user_id, room_id, region_id, op_type, coin_amount, status, request_hash, created_at_ms, updated_at_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
order.AppCode, order.OrderID, order.PlatformCode, order.ProviderOrderID, order.ProviderRoundID, order.GameID, order.ProviderGameID,
order.UserID, order.RoomID, order.OpType, order.CoinAmount, order.Status, order.RequestHash, order.CreatedAtMS, order.UpdatedAtMS,
order.UserID, order.RoomID, order.RegionID, order.OpType, order.CoinAmount, order.Status, order.RequestHash, order.CreatedAtMS, order.UpdatedAtMS,
); err != nil {
return gamedomain.GameOrder{}, false, err
}
@ -470,7 +478,7 @@ func (r *Repository) CreateGameOrder(ctx context.Context, order gamedomain.GameO
func queryOrderForUpdate(ctx context.Context, tx *sql.Tx, appCode string, platformCode string, providerOrderID string) (gamedomain.GameOrder, bool, error) {
row := tx.QueryRowContext(ctx,
`SELECT app_code, order_id, platform_code, provider_order_id, provider_round_id, game_id, provider_game_id,
user_id, room_id, op_type, coin_amount, status, wallet_transaction_id, wallet_balance_after,
user_id, room_id, region_id, op_type, coin_amount, status, wallet_transaction_id, wallet_balance_after,
request_hash, failure_code, failure_message, created_at_ms, updated_at_ms
FROM game_orders
WHERE app_code = ? AND platform_code = ? AND provider_order_id = ?
@ -478,7 +486,7 @@ func queryOrderForUpdate(ctx context.Context, tx *sql.Tx, appCode string, platfo
appcode.Normalize(appCode), strings.TrimSpace(platformCode), strings.TrimSpace(providerOrderID),
)
var order gamedomain.GameOrder
if err := row.Scan(&order.AppCode, &order.OrderID, &order.PlatformCode, &order.ProviderOrderID, &order.ProviderRoundID, &order.GameID, &order.ProviderGameID, &order.UserID, &order.RoomID, &order.OpType, &order.CoinAmount, &order.Status, &order.WalletTransactionID, &order.WalletBalanceAfter, &order.RequestHash, &order.FailureCode, &order.FailureMessage, &order.CreatedAtMS, &order.UpdatedAtMS); err != nil {
if err := row.Scan(&order.AppCode, &order.OrderID, &order.PlatformCode, &order.ProviderOrderID, &order.ProviderRoundID, &order.GameID, &order.ProviderGameID, &order.UserID, &order.RoomID, &order.RegionID, &order.OpType, &order.CoinAmount, &order.Status, &order.WalletTransactionID, &order.WalletBalanceAfter, &order.RequestHash, &order.FailureCode, &order.FailureMessage, &order.CreatedAtMS, &order.UpdatedAtMS); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return gamedomain.GameOrder{}, false, nil
}
@ -576,7 +584,7 @@ func (r *Repository) getRepairOrderByProvider(ctx context.Context, appCode strin
func queryOrderByIDForUpdate(ctx context.Context, tx *sql.Tx, appCode string, orderID string) (gamedomain.GameOrder, bool, error) {
row := tx.QueryRowContext(ctx,
`SELECT app_code, order_id, platform_code, provider_order_id, provider_round_id, game_id, provider_game_id,
user_id, room_id, op_type, coin_amount, status, wallet_transaction_id, wallet_balance_after,
user_id, room_id, region_id, op_type, coin_amount, status, wallet_transaction_id, wallet_balance_after,
request_hash, failure_code, failure_message, created_at_ms, updated_at_ms
FROM game_orders
WHERE app_code = ? AND order_id = ?
@ -584,7 +592,7 @@ func queryOrderByIDForUpdate(ctx context.Context, tx *sql.Tx, appCode string, or
appcode.Normalize(appCode), strings.TrimSpace(orderID),
)
var order gamedomain.GameOrder
if err := row.Scan(&order.AppCode, &order.OrderID, &order.PlatformCode, &order.ProviderOrderID, &order.ProviderRoundID, &order.GameID, &order.ProviderGameID, &order.UserID, &order.RoomID, &order.OpType, &order.CoinAmount, &order.Status, &order.WalletTransactionID, &order.WalletBalanceAfter, &order.RequestHash, &order.FailureCode, &order.FailureMessage, &order.CreatedAtMS, &order.UpdatedAtMS); err != nil {
if err := row.Scan(&order.AppCode, &order.OrderID, &order.PlatformCode, &order.ProviderOrderID, &order.ProviderRoundID, &order.GameID, &order.ProviderGameID, &order.UserID, &order.RoomID, &order.RegionID, &order.OpType, &order.CoinAmount, &order.Status, &order.WalletTransactionID, &order.WalletBalanceAfter, &order.RequestHash, &order.FailureCode, &order.FailureMessage, &order.CreatedAtMS, &order.UpdatedAtMS); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return gamedomain.GameOrder{}, false, nil
}
@ -632,6 +640,8 @@ func insertGameStatisticsOutbox(ctx context.Context, tx *sql.Tx, order gamedomai
'provider_round_id', ?,
'user_id', ?,
'room_id', ?,
'visible_region_id', ?,
'region_id', ?,
'op_type', ?,
'coin_amount', ?,
'wallet_transaction_id', ?,
@ -639,7 +649,7 @@ func insertGameStatisticsOutbox(ctx context.Context, tx *sql.Tx, order gamedomai
), 'pending', ?, ?)`,
order.AppCode, eventID, order.OrderID, order.UserID, order.PlatformCode, order.GameID, order.OpType, order.CoinAmount,
order.OrderID, order.PlatformCode, order.GameID, order.ProviderGameID, order.ProviderOrderID, order.ProviderRoundID,
order.UserID, order.RoomID, order.OpType, order.CoinAmount, walletTransactionID, balanceAfter,
order.UserID, order.RoomID, order.RegionID, order.RegionID, order.OpType, order.CoinAmount, walletTransactionID, balanceAfter,
nowMs, nowMs,
)
return err

1
services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql
View File

@ -31,6 +31,7 @@ CREATE TABLE IF NOT EXISTS stat_app_day_country (
google_recharge_usd_minor BIGINT NOT NULL DEFAULT 0 COMMENT 'Google 充值美元最小单位',
gift_coin_spent BIGINT NOT NULL DEFAULT 0 COMMENT '礼物消耗金币',
lucky_gift_turnover BIGINT NOT NULL DEFAULT 0 COMMENT '幸运礼物流水',
lucky_gift_payout BIGINT NOT NULL DEFAULT 0 COMMENT '幸运礼物返奖',
lucky_gift_payers BIGINT NOT NULL DEFAULT 0 COMMENT '幸运礼物付费去重人数',
game_turnover BIGINT NOT NULL DEFAULT 0 COMMENT '游戏流水',
game_players BIGINT NOT NULL DEFAULT 0 COMMENT '游戏参与去重人数',

48
services/statistics-service/internal/app/app.go
View File

@ -146,10 +146,17 @@ func newConsumers(cfg config.Config, repo *mysqlstorage.Repository) ([]*rocketmq
}
if err := add(cfg.RocketMQ.WalletOutbox, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
event, ok, err := rechargeEvent(message.Body)
if err != nil {
return err
}
if ok {
return repo.ConsumeRecharge(appcode.WithContext(ctx, event.AppCode), event)
}
reward, ok, err := luckyGiftRewardEvent(message.Body)
if err != nil || !ok {
return err
}
return repo.ConsumeRecharge(appcode.WithContext(ctx, event.AppCode), event)
return repo.ConsumeLuckyGiftReward(appcode.WithContext(ctx, reward.AppCode), reward)
}); err != nil {
shutdownConsumers(consumers)
return nil, err
@ -220,19 +227,41 @@ func rechargeEvent(body []byte) (mysqlstorage.RechargeEvent, bool, error) {
return mysqlstorage.RechargeEvent{}, false, nil
}
payload := mysqlstorage.DecodeJSON(message.PayloadJSON)
usdMinor := firstNonZeroInt64(mysqlstorage.Int64(payload, "recharge_usd_minor"), mysqlstorage.Int64(payload, "amount_micro"))
regionID := firstNonZeroInt64(mysqlstorage.Int64(payload, "target_region_id"), mysqlstorage.Int64(payload, "region_id"))
return mysqlstorage.RechargeEvent{
AppCode: message.AppCode,
EventID: message.EventID,
EventType: message.EventType,
UserID: message.UserID,
USDMinor: mysqlstorage.Int64(payload, "recharge_usd_minor"),
USDMinor: usdMinor,
RechargeSequence: mysqlstorage.Int64(payload, "recharge_sequence"),
TargetRegionID: mysqlstorage.Int64(payload, "target_region_id"),
RechargeType: firstNonEmpty(mysqlstorage.String(payload, "recharge_type"), "coin_seller_transfer"),
TargetRegionID: regionID,
RechargeType: firstNonEmpty(mysqlstorage.String(payload, "recharge_type"), mysqlstorage.String(payload, "channel"), mysqlstorage.String(payload, "provider"), "coin_seller_transfer"),
OccurredAtMS: message.OccurredAtMS,
}, true, nil
}
func luckyGiftRewardEvent(body []byte) (mysqlstorage.LuckyGiftRewardEvent, bool, error) {
message, err := walletmq.DecodeWalletOutboxMessage(body)
if err != nil {
return mysqlstorage.LuckyGiftRewardEvent{}, false, err
}
if message.EventType != "WalletLuckyGiftRewardCredited" {
return mysqlstorage.LuckyGiftRewardEvent{}, false, nil
}
payload := mysqlstorage.DecodeJSON(message.PayloadJSON)
return mysqlstorage.LuckyGiftRewardEvent{
AppCode: message.AppCode,
EventID: message.EventID,
EventType: message.EventType,
UserID: firstNonZeroInt64(mysqlstorage.Int64(payload, "user_id"), message.UserID),
CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "visible_region_id"), mysqlstorage.Int64(payload, "region_id"), mysqlstorage.Int64(payload, "target_region_id")),
Amount: mysqlstorage.Int64(payload, "amount"),
OccurredAtMS: message.OccurredAtMS,
}, true, nil
}
func roomGiftEvent(body []byte) (mysqlstorage.RoomGiftEvent, bool, error) {
envelope, _, err := roommq.DecodeRoomOutboxMessage(body)
if err != nil {
@ -291,7 +320,7 @@ func gameOrderEvent(body []byte) (mysqlstorage.GameOrderEvent, bool, error) {
AppCode: message.AppCode,
EventID: message.EventID,
UserID: message.UserID,
CountryID: mysqlstorage.Int64(payload, "visible_region_id"),
CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "visible_region_id"), mysqlstorage.Int64(payload, "region_id")),
PlatformCode: message.PlatformCode,
GameID: message.GameID,
OpType: message.OpType,
@ -345,3 +374,12 @@ func firstNonEmpty(values ...string) string {
}
return ""
}
func firstNonZeroInt64(values ...int64) int64 {
for _, value := range values {
if value != 0 {
return value
}
}
return 0
}

214
services/statistics-service/internal/app/local_flow_test.go Normal file
View File

@ -0,0 +1,214 @@
package app
import (
"context"
"encoding/json"
"runtime"
"testing"
"time"
"google.golang.org/protobuf/proto"
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
"hyapp/internal/testutil/mysqlschema"
"hyapp/pkg/gamemq"
"hyapp/pkg/roommq"
"hyapp/pkg/walletmq"
mysqlstorage "hyapp/services/statistics-service/internal/storage/mysql"
)
func TestLocalStatisticsDataScreenFlow(t *testing.T) {
ctx := context.Background()
_, file, _, _ := runtime.Caller(0)
schema := mysqlschema.New(t, mysqlschema.Config{
InitDBPath: mysqlschema.InitDBPath(t, file, "../../deploy/mysql/initdb/001_statistics_service.sql"),
DatabasePrefix: "hy_stats_flow",
})
repo, err := mysqlstorage.Open(ctx, schema.DSN)
if err != nil {
t.Fatalf("open statistics repository failed: %v", err)
}
t.Cleanup(func() { _ = repo.Close() })
occurredAt := time.Date(2026, 5, 31, 12, 0, 0, 0, time.UTC).UnixMilli()
dayStart := time.Date(2026, 5, 31, 0, 0, 0, 0, time.UTC).UnixMilli()
const (
appCode = "lalu"
regionID = int64(210)
userID = int64(42)
)
googleRecharge := walletMessage(t, walletmq.WalletOutboxMessage{
AppCode: appCode,
EventID: "WalletRechargeRecorded:google:1",
EventType: "WalletRechargeRecorded",
TransactionID: "wallet_tx_google_1",
CommandID: "google_order_1",
UserID: userID,
PayloadJSON: mustJSON(t, map[string]any{
"amount_micro": int64(1_500_000),
"region_id": regionID,
"channel": "google",
"recharge_sequence": int64(1),
}),
OccurredAtMS: occurredAt,
})
recharge, ok, err := rechargeEvent(googleRecharge)
if err != nil || !ok {
t.Fatalf("decode google recharge failed: ok=%v err=%v", ok, err)
}
if err := repo.ConsumeRecharge(ctx, recharge); err != nil {
t.Fatalf("consume google recharge failed: %v", err)
}
roomGiftBody := roomGiftMessage(t, appCode, "room_gift_1", "room_1", occurredAt, &roomeventsv1.RoomGiftSent{
SenderUserId: userID,
TargetUserId: 99,
GiftId: "rose",
GiftCount: 10,
GiftValue: 1_000,
VisibleRegionId: regionID,
CommandId: "gift_command_1",
PoolId: "lucky_100",
})
gift, ok, err := roomGiftEvent(roomGiftBody)
if err != nil || !ok {
t.Fatalf("decode room gift failed: ok=%v err=%v", ok, err)
}
if err := repo.ConsumeRoomGift(ctx, gift); err != nil {
t.Fatalf("consume room gift failed: %v", err)
}
luckyReward := walletMessage(t, walletmq.WalletOutboxMessage{
AppCode: appCode,
EventID: "WalletLuckyGiftRewardCredited:1",
EventType: "WalletLuckyGiftRewardCredited",
TransactionID: "wallet_tx_lucky_1",
CommandID: "lucky_reward:lucky_draw_1",
UserID: userID,
PayloadJSON: mustJSON(t, map[string]any{
"user_id": userID,
"draw_id": "lucky_draw_1",
"room_id": "room_1",
"pool_id": "lucky_100",
"amount": int64(300),
"visible_region_id": regionID,
}),
OccurredAtMS: occurredAt,
})
reward, ok, err := luckyGiftRewardEvent(luckyReward)
if err != nil || !ok {
t.Fatalf("decode lucky reward failed: ok=%v err=%v", ok, err)
}
if err := repo.ConsumeLuckyGiftReward(ctx, reward); err != nil {
t.Fatalf("consume lucky reward failed: %v", err)
}
for _, message := range []gamemq.GameOutboxMessage{
gameMessage(appCode, "game_order_debit_1", userID, "debit", 700, regionID, occurredAt),
gameMessage(appCode, "game_order_credit_1", userID, "credit", 200, regionID, occurredAt),
} {
body, err := gamemq.EncodeGameOutboxMessage(message)
if err != nil {
t.Fatalf("encode game message failed: %v", err)
}
event, ok, err := gameOrderEvent(body)
if err != nil || !ok {
t.Fatalf("decode game order failed: ok=%v err=%v", ok, err)
}
if err := repo.ConsumeGameOrder(ctx, event); err != nil {
t.Fatalf("consume game order failed: %v", err)
}
}
overview, err := repo.QueryOverview(ctx, mysqlstorage.OverviewQuery{
AppCode: appCode,
StartMS: dayStart,
EndMS: dayStart + int64(24*time.Hour/time.Millisecond),
CountryID: regionID,
})
if err != nil {
t.Fatalf("query overview failed: %v", err)
}
if overview.GoogleRechargeUSDMinor != 1_500_000 || overview.RechargeUSDMinor != 1_500_000 || overview.NewUserRechargeUSDMinor != 1_500_000 {
t.Fatalf("google recharge metrics mismatch: %+v", overview)
}
if overview.LuckyGiftTurnover != 1_000 || overview.LuckyGiftPayout != 300 || overview.LuckyGiftProfit != 700 || overview.LuckyGiftPayers != 1 {
t.Fatalf("lucky gift metrics mismatch: %+v", overview)
}
if overview.GameTurnover != 700 || overview.GamePayout != 200 || overview.GameProfit != 500 || overview.GamePlayers != 1 {
t.Fatalf("game metrics mismatch: %+v", overview)
}
}
func walletMessage(t *testing.T, message walletmq.WalletOutboxMessage) []byte {
t.Helper()
body, err := walletmq.EncodeWalletOutboxMessage(message)
if err != nil {
t.Fatalf("encode wallet message failed: %v", err)
}
return body
}
func roomGiftMessage(t *testing.T, appCode string, eventID string, roomID string, occurredAt int64, gift *roomeventsv1.RoomGiftSent) []byte {
t.Helper()
giftBody, err := proto.Marshal(gift)
if err != nil {
t.Fatalf("marshal gift failed: %v", err)
}
body, err := roommq.EncodeRoomOutboxMessage(&roomeventsv1.EventEnvelope{
EventId: eventID,
RoomId: roomID,
EventType: "RoomGiftSent",
RoomVersion: 1,
OccurredAtMs: occurredAt,
Body: giftBody,
AppCode: appCode,
})
if err != nil {
t.Fatalf("encode room message failed: %v", err)
}
return body
}
func gameMessage(appCode string, orderID string, userID int64, opType string, amount int64, regionID int64, occurredAt int64) gamemq.GameOutboxMessage {
return gamemq.GameOutboxMessage{
AppCode: appCode,
EventID: "GameOrderSettled:" + orderID,
EventType: "GameOrderSettled",
OrderID: orderID,
UserID: userID,
PlatformCode: "yomi",
GameID: "fish_001",
OpType: opType,
CoinAmount: amount,
PayloadJSON: mustJSONNoT(map[string]any{
"order_id": orderID,
"platform_code": "yomi",
"game_id": "fish_001",
"user_id": userID,
"room_id": "room_1",
"visible_region_id": regionID,
"region_id": regionID,
"op_type": opType,
"coin_amount": amount,
}),
OccurredAtMS: occurredAt,
}
}
func mustJSON(t *testing.T, value map[string]any) string {
t.Helper()
body, err := json.Marshal(value)
if err != nil {
t.Fatalf("marshal json failed: %v", err)
}
return string(body)
}
func mustJSONNoT(value map[string]any) string {
body, err := json.Marshal(value)
if err != nil {
panic(err)
}
return string(body)
}

11
services/statistics-service/internal/storage/mysql/query.go
View File

@ -30,7 +30,11 @@ type Overview struct {
GoogleRechargeUSDMinor int64 `json:"google_recharge_usd_minor"`
GiftCoinSpent int64 `json:"gift_coin_spent"`
LuckyGiftTurnover int64 `json:"lucky_gift_turnover"`
LuckyGiftPayout int64 `json:"lucky_gift_payout"`
LuckyGiftPayers int64 `json:"lucky_gift_payers"`
LuckyGiftProfit int64 `json:"lucky_gift_profit"`
LuckyGiftPayoutRate float64 `json:"lucky_gift_payout_rate"`
LuckyGiftProfitRate float64 `json:"lucky_gift_profit_rate"`
GameTurnover int64 `json:"game_turnover"`
GamePayout int64 `json:"game_payout"`
GameRefund int64 `json:"game_refund"`
@ -83,16 +87,19 @@ func (r *Repository) QueryOverview(ctx context.Context, query OverviewQuery) (Ov
COALESCE(SUM(recharge_usd_minor),0), COALESCE(SUM(new_user_recharge_usd_minor),0),
COALESCE(SUM(coin_seller_recharge_usd_minor),0), COALESCE(SUM(mifapay_recharge_usd_minor),0),
COALESCE(SUM(google_recharge_usd_minor),0), COALESCE(SUM(gift_coin_spent),0),
COALESCE(SUM(lucky_gift_turnover),0), COALESCE(SUM(lucky_gift_payers),0),
COALESCE(SUM(lucky_gift_turnover),0), COALESCE(SUM(lucky_gift_payout),0), COALESCE(SUM(lucky_gift_payers),0),
COALESCE(SUM(game_turnover),0), COALESCE(SUM(game_payout),0), COALESCE(SUM(game_refund),0),
COALESCE(SUM(game_players),0)
FROM stat_app_day_country
WHERE app_code = ? AND stat_day BETWEEN ? AND ?`+countryFilter, args...)
var overview Overview
overview.AppCode, overview.StartMS, overview.EndMS, overview.CountryID = app, query.StartMS, query.EndMS, query.CountryID
if err := row.Scan(&overview.NewUsers, &overview.ActiveUsers, &overview.PaidUsers, &overview.RechargeUSDMinor, &overview.NewUserRechargeUSDMinor, &overview.CoinSellerRechargeUSDMinor, &overview.MifaPayRechargeUSDMinor, &overview.GoogleRechargeUSDMinor, &overview.GiftCoinSpent, &overview.LuckyGiftTurnover, &overview.LuckyGiftPayers, &overview.GameTurnover, &overview.GamePayout, &overview.GameRefund, &overview.GamePlayers); err != nil {
if err := row.Scan(&overview.NewUsers, &overview.ActiveUsers, &overview.PaidUsers, &overview.RechargeUSDMinor, &overview.NewUserRechargeUSDMinor, &overview.CoinSellerRechargeUSDMinor, &overview.MifaPayRechargeUSDMinor, &overview.GoogleRechargeUSDMinor, &overview.GiftCoinSpent, &overview.LuckyGiftTurnover, &overview.LuckyGiftPayout, &overview.LuckyGiftPayers, &overview.GameTurnover, &overview.GamePayout, &overview.GameRefund, &overview.GamePlayers); err != nil {
return Overview{}, err
}
overview.LuckyGiftProfit = overview.LuckyGiftTurnover - overview.LuckyGiftPayout
overview.LuckyGiftPayoutRate = ratio(overview.LuckyGiftPayout, overview.LuckyGiftTurnover)
overview.LuckyGiftProfitRate = ratio(overview.LuckyGiftProfit, overview.LuckyGiftTurnover)
overview.GameProfit = overview.GameTurnover - overview.GamePayout - overview.GameRefund
overview.GameProfitRate = ratio(overview.GameProfit, overview.GameTurnover)
overview.ARPUUSDMinor = div(overview.RechargeUSDMinor, overview.ActiveUsers)

31
services/statistics-service/internal/storage/mysql/repository.go
View File

@ -75,7 +75,7 @@ func (r *Repository) Migrate(ctx context.Context) error {
recharge_usd_minor BIGINT NOT NULL DEFAULT 0, new_user_recharge_usd_minor BIGINT NOT NULL DEFAULT 0,
coin_seller_recharge_usd_minor BIGINT NOT NULL DEFAULT 0, mifapay_recharge_usd_minor BIGINT NOT NULL DEFAULT 0,
google_recharge_usd_minor BIGINT NOT NULL DEFAULT 0, gift_coin_spent BIGINT NOT NULL DEFAULT 0,
lucky_gift_turnover BIGINT NOT NULL DEFAULT 0, lucky_gift_payers BIGINT NOT NULL DEFAULT 0,
lucky_gift_turnover BIGINT NOT NULL DEFAULT 0, lucky_gift_payout BIGINT NOT NULL DEFAULT 0, lucky_gift_payers BIGINT NOT NULL DEFAULT 0,
game_turnover BIGINT NOT NULL DEFAULT 0, game_players BIGINT NOT NULL DEFAULT 0,
game_payout BIGINT NOT NULL DEFAULT 0, game_refund BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, stat_day, country_id)
@ -124,6 +124,7 @@ func (r *Repository) Migrate(ctx context.Context) error {
alterStatements := []string{
`ALTER TABLE stat_app_day_country ADD COLUMN paid_users BIGINT NOT NULL DEFAULT 0 AFTER active_users`,
`ALTER TABLE stat_app_day_country ADD COLUMN game_refund BIGINT NOT NULL DEFAULT 0 AFTER game_payout`,
`ALTER TABLE stat_app_day_country ADD COLUMN lucky_gift_payout BIGINT NOT NULL DEFAULT 0 AFTER lucky_gift_turnover`,
`ALTER TABLE stat_game_day_country ADD COLUMN refund_coin BIGINT NOT NULL DEFAULT 0 AFTER payout_coin`,
}
for _, statement := range alterStatements {
@ -164,6 +165,16 @@ type RoomGiftEvent struct {
OccurredAtMS int64
}
type LuckyGiftRewardEvent struct {
AppCode string
EventID string
EventType string
UserID int64
CountryID int64
Amount int64
OccurredAtMS int64
}
type UserActiveEvent struct {
AppCode string
EventID string
@ -227,7 +238,7 @@ func (r *Repository) ConsumeRecharge(ctx context.Context, event RechargeEvent) e
}
coinSellerUSD, mifapayUSD, googleUSD := int64(0), int64(0), int64(0)
switch strings.ToLower(strings.TrimSpace(event.RechargeType)) {
case "google":
case "google", "google_play":
googleUSD = event.USDMinor
case "mifapay":
mifapayUSD = event.USDMinor
@ -278,6 +289,22 @@ func (r *Repository) ConsumeRoomGift(ctx context.Context, event RoomGiftEvent) e
})
}
func (r *Repository) ConsumeLuckyGiftReward(ctx context.Context, event LuckyGiftRewardEvent) error {
return r.withEvent(ctx, SourceWallet, event.EventID, "WalletLuckyGiftRewardCredited", func(tx *sql.Tx, nowMS int64) error {
day := statDay(event.OccurredAtMS)
countryID := normalizeID(event.CountryID)
if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, nowMS); err != nil {
return err
}
_, err := tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET lucky_gift_payout = lucky_gift_payout + ?, updated_at_ms = ?
WHERE app_code = ? AND stat_day = ? AND country_id = ?
`, event.Amount, nowMS, appcode.Normalize(event.AppCode), day, countryID)
return err
})
}
func (r *Repository) ConsumeUserActive(ctx context.Context, event UserActiveEvent) error {
return r.withEvent(ctx, SourceRoom, event.EventID, event.EventType, func(tx *sql.Tx, nowMS int64) error {
return applyActive(ctx, tx, event.AppCode, statDay(event.OccurredAtMS), normalizeID(event.CountryID), event.UserID, event.OccurredAtMS, nowMS)

4
services/user-service/internal/service/auth/quick_account.go
View File

@ -28,9 +28,9 @@ func (s *Service) QuickCreateAccount(ctx context.Context, password string, regis
// refresh session 必须绑定设备，平台也会进入注册来源和风控审计。
return authdomain.Token{}, xerr.New(xerr.InvalidArgument, "device_id and platform are required")
}
if registration.Username == "" || registration.Avatar == "" || registration.Gender == "" || registration.Country == "" {
if err := validateThirdPartyRegistrationRequiredForCreate(registration); err != nil {
// 快捷账号直接跳过 onboarding，因此必须一次性写齐客户端准入所需的最小资料。
return authdomain.Token{}, xerr.New(xerr.InvalidArgument, "username, avatar, gender and country are required")
return authdomain.Token{}, err
}
if err := validateThirdPartyRegistration(registration); err != nil {
return authdomain.Token{}, err

76
services/user-service/internal/service/auth/service_test.go
View File

@ -126,8 +126,15 @@ func newAuthService(repository *mysqltest.Repository, now *time.Time, ids []int6
}
func thirdPartyRegistration(platform string) authdomain.ThirdPartyRegistration {
// 三方登录是注册入口，测试默认同时提供设备和平台，避免生成缺失注册上下文的用户。
return authdomain.ThirdPartyRegistration{DeviceID: "dev-" + platform, Platform: platform}
// 三方登录是注册入口，测试默认提供完整最小资料，避免生成后台不可运营的空资料用户。
return authdomain.ThirdPartyRegistration{
Username: "hy",
Avatar: "https://cdn.example/avatar.png",
Gender: "male",
Country: "SG",
DeviceID: "dev-" + platform,
Platform: platform,
}
}
func seedCountry(t *testing.T, repository *mysqltest.Repository, code string) userdomain.Country {
@ -174,7 +181,7 @@ func TestThirdPartyUserSetsPasswordThenLogsIn(t *testing.T) {
// 覆盖三方注册、设置密码、密码登录、refresh 轮换和 logout 的完整认证生命周期。
ctx := context.Background()
repository := mysqltest.NewRepository(t)
seedCountry(t, repository, "CN")
seedCountry(t, repository, "SG")
now := time.UnixMilli(1000)
svc := newAuthService(repository, &now, []int64{900001}, []string{"100001"})
@ -185,8 +192,8 @@ func TestThirdPartyUserSetsPasswordThenLogsIn(t *testing.T) {
if !isNewUser || thirdToken.UserID != 900001 || thirdToken.DisplayUserID != "163000" || thirdToken.RefreshToken == "" || thirdToken.AccessToken == "" {
t.Fatalf("unexpected third-party token: token=%+v isNew=%v", thirdToken, isNewUser)
}
if thirdToken.ProfileCompleted || thirdToken.OnboardingStatus != string(userdomain.OnboardingStatusProfileRequired) {
t.Fatalf("new third-party user must require onboarding: %+v", thirdToken)
if !thirdToken.ProfileCompleted || thirdToken.OnboardingStatus != string(userdomain.OnboardingStatusCompleted) {
t.Fatalf("new third-party user must be completed: %+v", thirdToken)
}
if _, err := svc.LoginPassword(ctx, "163000", "secret-pass", "ios", authservice.Meta{}); !xerr.IsCode(err, xerr.AuthFailed) {
@ -287,25 +294,19 @@ func TestQuickCreateAccountCreatesCompletedPasswordLogin(t *testing.T) {
}
func TestRefreshTokenCarriesCompletedOnboardingStatus(t *testing.T) {
// gateway profile gate 依赖 access token 快照；refresh 必须在用户完成注册后带出最新状态。
// gateway profile gate 依赖 access token 快照；注册资料齐全的新用户必须直接带出完成态。
ctx := context.Background()
repository := mysqltest.NewRepository(t)
seedCountry(t, repository, "SG")
now := time.UnixMilli(1000)
authSvc := newAuthService(repository, &now, []int64{900009}, []string{"100009"})
userSvc := newUserService(repository, userservice.WithClock(func() time.Time { return now }))
token, _, err := authSvc.LoginThirdParty(ctx, "wechat", "openid-onboarding", thirdPartyRegistration("ios"), authservice.Meta{RequestID: "req-login"})
if err != nil {
t.Fatalf("LoginThirdParty failed: %v", err)
}
if token.ProfileCompleted || token.OnboardingStatus != string(userdomain.OnboardingStatusProfileRequired) {
t.Fatalf("new token should require onboarding: %+v", token)
}
now = time.UnixMilli(2000)
if _, err := userSvc.CompleteOnboarding(ctx, token.UserID, "hy", "https://cdn.example/a.png", "male", "SG", "", ""); err != nil {
t.Fatalf("CompleteOnboarding failed: %v", err)
if !token.ProfileCompleted || token.OnboardingStatus != string(userdomain.OnboardingStatusCompleted) {
t.Fatalf("new token should be completed: %+v", token)
}
now = time.UnixMilli(3000)
@ -322,6 +323,7 @@ func TestLoginIPRiskWorkerRevokesBlockedCountrySession(t *testing.T) {
ctx := context.Background()
repository := mysqltest.NewRepository(t)
seedCountry(t, repository, "CN")
seedCountry(t, repository, "SG")
now := time.UnixMilli(1000)
cache := newMemoryDecisionCache()
authSvc := newAuthService(repository, &now, []int64{900020}, []string{"100020"},
@ -392,6 +394,7 @@ func TestLoginIPRiskWorkerUsesDynamicCountryBlocks(t *testing.T) {
ctx := context.Background()
repository := mysqltest.NewRepository(t)
seedCountry(t, repository, "MY")
seedCountry(t, repository, "SG")
now := time.UnixMilli(1000)
if _, err := repository.RawDB().ExecContext(ctx, `
INSERT INTO login_risk_country_blocks (
@ -444,22 +447,17 @@ func TestLoginIPRiskWorkerUsesDynamicCountryBlocks(t *testing.T) {
}
func TestIssueAccessTokenForSessionCarriesCompletedOnboardingStatusWithoutRotatingRefresh(t *testing.T) {
// CompleteOnboarding 后只需要替换 access token；refresh token 原文不可从数据库反查，也不应被静默轮换。
// 重新签发 access token 只能复用已有 session；refresh token 原文不可从数据库反查，也不应被静默轮换。
ctx := context.Background()
repository := mysqltest.NewRepository(t)
seedCountry(t, repository, "SG")
now := time.UnixMilli(1000)
authSvc := newAuthService(repository, &now, []int64{900010}, []string{"100010"})
userSvc := newUserService(repository, userservice.WithClock(func() time.Time { return now }))
token, _, err := authSvc.LoginThirdParty(ctx, "wechat", "openid-onboarding-resign", thirdPartyRegistration("ios"), authservice.Meta{RequestID: "req-login"})
if err != nil {
t.Fatalf("LoginThirdParty failed: %v", err)
}
now = time.UnixMilli(2000)
if _, err := userSvc.CompleteOnboarding(ctx, token.UserID, "hy", "https://cdn.example/a.png", "male", "SG", "", ""); err != nil {
t.Fatalf("CompleteOnboarding failed: %v", err)
}
now = time.UnixMilli(3000)
reissued, err := authSvc.IssueAccessTokenForSession(ctx, token.UserID, token.SessionID, authservice.Meta{RequestID: "req-resign"})
@ -479,6 +477,7 @@ func TestDisabledUserCannotLogin(t *testing.T) {
ctx := context.Background()
repository := mysqltest.NewRepository(t)
seedCountry(t, repository, "CN")
seedCountry(t, repository, "SG")
now := time.UnixMilli(1000)
svc := newAuthService(repository, &now, []int64{900001}, []string{"100001"})
@ -507,6 +506,7 @@ func TestThirdPartyLoginOrRegister(t *testing.T) {
// 首次三方登录创建用户，再次同 provider subject 登录应复用原用户。
ctx := context.Background()
repository := mysqltest.NewRepository(t)
seedCountry(t, repository, "SG")
now := time.UnixMilli(1000)
svc := newAuthService(repository, &now, []int64{900002}, []string{"100002"})
@ -517,8 +517,11 @@ func TestThirdPartyLoginOrRegister(t *testing.T) {
if !isNewUser || firstToken.UserID != 900002 || firstToken.DisplayUserID != "163000" {
t.Fatalf("unexpected first third-party result: token=%+v isNew=%v", firstToken, isNewUser)
}
if !firstToken.ProfileCompleted || firstToken.OnboardingStatus != string(userdomain.OnboardingStatusCompleted) {
t.Fatalf("first third-party registration must complete profile: %+v", firstToken)
}
secondToken, isNewUser, err := svc.LoginThirdParty(ctx, "wechat", "openid-1", thirdPartyRegistration("ios"), authservice.Meta{RequestID: "req-third-second"})
secondToken, isNewUser, err := svc.LoginThirdParty(ctx, "wechat", "openid-1", authdomain.ThirdPartyRegistration{DeviceID: "dev-ios", Platform: "ios"}, authservice.Meta{RequestID: "req-third-second"})
if err != nil {
t.Fatalf("second LoginThirdParty failed: %v", err)
}
@ -532,6 +535,32 @@ func TestThirdPartyLoginOrRegister(t *testing.T) {
}
}
func TestThirdPartyRegisterRequiresProfileFields(t *testing.T) {
// 三方首次注册必须一次性写齐后台列表和业务准入依赖的最小资料，不能再落 profile_required 空用户。
ctx := context.Background()
repository := mysqltest.NewRepository(t)
seedCountry(t, repository, "SG")
now := time.UnixMilli(1000)
svc := newAuthService(repository, &now, []int64{900012, 900013, 900014, 900015}, []string{"100012", "100013", "100014", "100015"})
base := thirdPartyRegistration("ios")
cases := []struct {
name string
registration authdomain.ThirdPartyRegistration
}{
{name: "username", registration: authdomain.ThirdPartyRegistration{Avatar: base.Avatar, Gender: base.Gender, Country: base.Country, DeviceID: base.DeviceID, Platform: base.Platform}},
{name: "country", registration: authdomain.ThirdPartyRegistration{Username: base.Username, Avatar: base.Avatar, Gender: base.Gender, DeviceID: base.DeviceID, Platform: base.Platform}},
{name: "gender", registration: authdomain.ThirdPartyRegistration{Username: base.Username, Avatar: base.Avatar, Country: base.Country, DeviceID: base.DeviceID, Platform: base.Platform}},
{name: "avatar", registration: authdomain.ThirdPartyRegistration{Username: base.Username, Gender: base.Gender, Country: base.Country, DeviceID: base.DeviceID, Platform: base.Platform}},
}
for _, tc := range cases {
_, _, err := svc.LoginThirdParty(ctx, "wechat", "openid-missing-"+tc.name, tc.registration, authservice.Meta{})
if !xerr.IsCode(err, xerr.InvalidArgument) {
t.Fatalf("expected INVALID_ARGUMENT for missing %s, got %v", tc.name, err)
}
}
}
func TestThirdPartyRegisterStoresRegistrationProfile(t *testing.T) {
// 三方首次注册要把资料字段和注册来源快照随 users 主记录同事务写入。
ctx := context.Background()
@ -624,6 +653,9 @@ func TestThirdPartyRegisterAssignsRegionAndExistingLoginKeepsOriginalCountry(t *
svc := newAuthService(repository, &now, []int64{900006}, []string{"100006"})
firstToken, isNew, err := svc.LoginThirdParty(ctx, "wechat", "openid-region", authdomain.ThirdPartyRegistration{
Username: "hy",
Avatar: "https://cdn.example/avatar.png",
Gender: "male",
DeviceID: "ios",
Platform: "ios",
Country: " sg ",
@ -756,6 +788,7 @@ func TestThirdPartyRegisterRetriesDisplayUserIDConflict(t *testing.T) {
// 默认短号候选冲突时，三方注册应在有限次数内重试下一个候选。
ctx := context.Background()
repository := mysqltest.NewRepository(t)
seedCountry(t, repository, "SG")
repository.PutUser(userdomain.User{UserID: 1, CurrentDisplayUserID: "163000", Status: userdomain.StatusActive})
now := time.UnixMilli(1000)
svc := newAuthService(repository, &now, []int64{900001, 900002}, []string{"100001", "100002"})
@ -779,6 +812,7 @@ func TestPasswordLoginUsesCurrentDisplayUserIDWithPrettyLease(t *testing.T) {
// 密码登录必须使用当前 active display_user_id；靓号 active 时默认号不能登录。
ctx := context.Background()
repository := mysqltest.NewRepository(t)
seedCountry(t, repository, "SG")
now := time.UnixMilli(1000)
authSvc := newAuthService(repository, &now, []int64{900001}, []string{"100001"})
userSvc := newUserService(repository, userservice.WithClock(func() time.Time { return now }))

16
services/user-service/internal/service/auth/third_party.go
View File

@ -83,6 +83,10 @@ func (s *Service) LoginThirdParty(ctx context.Context, provider string, credenti
s.audit(ctx, meta, 0, loginThird, provider, resultFailed, string(xerr.CodeOf(err)))
return authdomain.Token{}, false, err
}
if err := validateThirdPartyRegistrationRequiredForCreate(registration); err != nil {
// 只有首次创建用户才要求注册资料完整；已有三方绑定登录不能被请求体资料缺失阻断。
return authdomain.Token{}, false, err
}
registration, err = s.resolveRegistrationCountry(ctx, registration)
if err != nil {
return authdomain.Token{}, false, err
@ -129,6 +133,9 @@ func (s *Service) createThirdPartyUser(ctx context.Context, provider string, pro
// 三方首次登录要同时生成 user_id 和默认短号，短号冲突时有限重试。
user, displayIdentity := s.newUserWithIdentity(attempt)
user = applyThirdPartyRegistration(user, registration)
user.ProfileCompleted = true
user.ProfileCompletedAtMs = user.CreatedAtMs
user.OnboardingStatus = userdomain.OnboardingStatusCompleted
displayUserID, err := s.authRepository.AllocateDisplayUserID(ctx, registration.AppCode)
if err != nil {
return authdomain.Token{}, false, err
@ -292,6 +299,15 @@ func validateThirdPartyRegistration(registration authdomain.ThirdPartyRegistrati
return nil
}
func validateThirdPartyRegistrationRequiredForCreate(registration authdomain.ThirdPartyRegistration) error {
if registration.Username == "" || registration.Country == "" || registration.Gender == "" || registration.Avatar == "" {
// 注册即落用户主数据；最小公开资料不完整时拒绝创建，避免后台出现空昵称、空头像或 GLOBAL 国家用户。
return xerr.New(xerr.InvalidArgument, "username, country, gender and avatar are required")
}
return nil
}
func validateRegistrationLengths(registration authdomain.ThirdPartyRegistration) error {
checks := []struct {
name string

4
services/user-service/internal/transport/grpc/server_test.go
View File

@ -157,6 +157,10 @@ func TestCompleteOnboardingReturnsReplacementAccessToken(t *testing.T) {
Meta: &userv1.RequestMeta{RequestId: "req-login", DeviceId: "ios"},
Provider: "wechat",
Credential: "openid-complete-token",
Username: "hy",
Avatar: "https://cdn.example/a.png",
Gender: "male",
Country: "SG",
DeviceId: "ios",
Platform: "ios",
})

19
services/wallet-service/internal/domain/ledger/ledger.go
View File

@ -629,15 +629,16 @@ type TaskRewardReceipt struct {
// LuckyGiftRewardCommand 是 activity-service 用抽奖 draw_id 发起的 COIN 入账命令。
type LuckyGiftRewardCommand struct {
AppCode string
CommandID string
TargetUserID int64
Amount int64
DrawID string
RoomID string
GiftID string
PoolID string
Reason string
AppCode string
CommandID string
TargetUserID int64
Amount int64
DrawID string
RoomID string
VisibleRegionID int64
GiftID string
PoolID string
Reason string
}
// LuckyGiftRewardReceipt 是幸运礼物返奖入账后的稳定回执。

68
services/wallet-service/internal/storage/mysql/repository.go
View File

@ -593,17 +593,18 @@ func (r *Repository) CreditLuckyGiftReward(ctx context.Context, command ledger.L
transactionID := transactionID(command.AppCode, command.CommandID)
balanceAfter := account.AvailableAmount + command.Amount
metadata := luckyGiftRewardMetadata{
AppCode: command.AppCode,
TargetUserID: command.TargetUserID,
AssetType: ledger.AssetCoin,
Amount: command.Amount,
DrawID: command.DrawID,
RoomID: command.RoomID,
GiftID: command.GiftID,
PoolID: command.PoolID,
Reason: command.Reason,
BalanceAfter: balanceAfter,
GrantedAtMS: nowMs,
AppCode: command.AppCode,
TargetUserID: command.TargetUserID,
AssetType: ledger.AssetCoin,
Amount: command.Amount,
DrawID: command.DrawID,
RoomID: command.RoomID,
VisibleRegionID: command.VisibleRegionID,
GiftID: command.GiftID,
PoolID: command.PoolID,
Reason: command.Reason,
BalanceAfter: balanceAfter,
GrantedAtMS: nowMs,
}
if err := r.insertTransaction(ctx, tx, transactionID, command.CommandID, bizTypeLuckyGiftReward, requestHash, command.DrawID, metadata, nowMs); err != nil {
return ledger.LuckyGiftRewardReceipt{}, err
@ -1794,17 +1795,18 @@ type taskRewardMetadata struct {
}
type luckyGiftRewardMetadata struct {
AppCode string `json:"app_code"`
TargetUserID int64 `json:"target_user_id"`
AssetType string `json:"asset_type"`
Amount int64 `json:"amount"`
DrawID string `json:"draw_id"`
RoomID string `json:"room_id"`
GiftID string `json:"gift_id"`
PoolID string `json:"pool_id"`
Reason string `json:"reason"`
BalanceAfter int64 `json:"balance_after"`
GrantedAtMS int64 `json:"granted_at_ms"`
AppCode string `json:"app_code"`
TargetUserID int64 `json:"target_user_id"`
AssetType string `json:"asset_type"`
Amount int64 `json:"amount"`
DrawID string `json:"draw_id"`
RoomID string `json:"room_id"`
VisibleRegionID int64 `json:"visible_region_id"`
GiftID string `json:"gift_id"`
PoolID string `json:"pool_id"`
Reason string `json:"reason"`
BalanceAfter int64 `json:"balance_after"`
GrantedAtMS int64 `json:"granted_at_ms"`
}
type gameCoinMetadata struct {
@ -2140,16 +2142,18 @@ func luckyGiftRewardCreditedEvent(transactionID string, commandID string, metada
AvailableDelta: metadata.Amount,
FrozenDelta: 0,
Payload: map[string]any{
"transaction_id": transactionID,
"command_id": commandID,
"user_id": metadata.TargetUserID,
"draw_id": metadata.DrawID,
"room_id": metadata.RoomID,
"gift_id": metadata.GiftID,
"pool_id": metadata.PoolID,
"amount": metadata.Amount,
"reason": metadata.Reason,
"created_at_ms": nowMs,
"transaction_id": transactionID,
"command_id": commandID,
"user_id": metadata.TargetUserID,
"draw_id": metadata.DrawID,
"room_id": metadata.RoomID,
"visible_region_id": metadata.VisibleRegionID,
"region_id": metadata.VisibleRegionID,
"gift_id": metadata.GiftID,
"pool_id": metadata.PoolID,
"amount": metadata.Amount,
"reason": metadata.Reason,
"created_at_ms": nowMs,
},
CreatedAtMS: nowMs,
}

19
services/wallet-service/internal/transport/grpc/server.go
View File

@ -661,15 +661,16 @@ func (s *Server) CreditTaskReward(ctx context.Context, req *walletv1.CreditTaskR
func (s *Server) CreditLuckyGiftReward(ctx context.Context, req *walletv1.CreditLuckyGiftRewardRequest) (*walletv1.CreditLuckyGiftRewardResponse, error) {
ctx = appcode.WithContext(ctx, req.GetAppCode())
receipt, err := s.svc.CreditLuckyGiftReward(ctx, ledger.LuckyGiftRewardCommand{
AppCode: req.GetAppCode(),
CommandID: req.GetCommandId(),
TargetUserID: req.GetTargetUserId(),
Amount: req.GetAmount(),
DrawID: req.GetDrawId(),
RoomID: req.GetRoomId(),
GiftID: req.GetGiftId(),
PoolID: req.GetPoolId(),
Reason: req.GetReason(),
AppCode: req.GetAppCode(),
CommandID: req.GetCommandId(),
TargetUserID: req.GetTargetUserId(),
Amount: req.GetAmount(),
DrawID: req.GetDrawId(),
RoomID: req.GetRoomId(),
VisibleRegionID: req.GetVisibleRegionId(),
GiftID: req.GetGiftId(),
PoolID: req.GetPoolId(),
Reason: req.GetReason(),
})
if err != nil {
return nil, xerr.ToGRPCError(err)