From 88584f132983d51d0d739fcfce5760ea59dc918a Mon Sep 17 00:00:00 2001 From: ZuoZuo <68836346+Mrz-sakura@users.noreply.github.com> Date: Sun, 31 May 2026 16:39:49 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=B9=B8=E8=BF=90=E7=A4=BC?= =?UTF-8?q?=E7=89=A9=E4=BF=AE=E5=A4=8D=E5=92=8C=E5=A4=A7=E5=B1=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api/proto/wallet/v1/wallet.pb.go | 38 ++- api/proto/wallet/v1/wallet.proto | 1 + server/admin/cmd/server/main.go | 2 +- .../internal/modules/luckygift/handler.go | 37 +- .../configs/config.docker.yaml | 1 + .../configs/config.tencent.example.yaml | 1 + services/activity-service/configs/config.yaml | 1 + .../mysql/initdb/001_activity_service.sql | 38 +++ services/activity-service/internal/app/app.go | 1 + .../internal/config/config.go | 8 + .../internal/service/luckygift/service.go | 90 ++++- .../service/luckygift/service_test.go | 62 +++- .../storage/mysql/lucky_gift_repository.go | 319 +++++++++++++++++- .../internal/storage/mysql/repository.go | 143 ++++++++ .../deploy/mysql/initdb/001_game_service.sql | 2 + .../game-service/internal/domain/game/game.go | 2 + .../internal/service/game/service.go | 34 ++ .../internal/service/game/service_test.go | 2 +- .../internal/service/game/yomi_adapter.go | 1 + .../internal/storage/mysql/repository.go | 38 ++- .../mysql/initdb/001_statistics_service.sql | 1 + .../statistics-service/internal/app/app.go | 48 ++- .../internal/app/local_flow_test.go | 214 ++++++++++++ .../internal/storage/mysql/query.go | 11 +- .../internal/storage/mysql/repository.go | 31 +- .../internal/service/auth/quick_account.go | 4 +- .../internal/service/auth/service_test.go | 76 +++-- .../internal/service/auth/third_party.go | 16 + .../internal/transport/grpc/server_test.go | 4 + .../internal/domain/ledger/ledger.go | 19 +- .../internal/storage/mysql/repository.go | 68 ++-- .../internal/transport/grpc/server.go | 19 +- 32 files changed, 1187 insertions(+), 145 deletions(-) create mode 100644 services/statistics-service/internal/app/local_flow_test.go diff --git a/api/proto/wallet/v1/wallet.pb.go b/api/proto/wallet/v1/wallet.pb.go index 337142f1..c712ebc6 100644 --- a/api/proto/wallet/v1/wallet.pb.go +++ b/api/proto/wallet/v1/wallet.pb.go @@ -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" + diff --git a/api/proto/wallet/v1/wallet.proto b/api/proto/wallet/v1/wallet.proto index a9c8bcc2..a06d6ba5 100644 --- a/api/proto/wallet/v1/wallet.proto +++ b/api/proto/wallet/v1/wallet.proto @@ -1155,6 +1155,7 @@ message CreditLuckyGiftRewardRequest { string gift_id = 7; string pool_id = 8; string reason = 9; + int64 visible_region_id = 10; } // CreditLuckyGiftRewardResponse 返回幸运礼物返奖入账流水和用户 COIN 账后余额。 diff --git a/server/admin/cmd/server/main.go b/server/admin/cmd/server/main.go index 99f94dff..38099096 100644 --- a/server/admin/cmd/server/main.go +++ b/server/admin/cmd/server/main.go @@ -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), diff --git a/server/admin/internal/modules/luckygift/handler.go b/server/admin/internal/modules/luckygift/handler.go index 2b7e43cb..c23ba0ea 100644 --- a/server/admin/internal/modules/luckygift/handler.go +++ b/server/admin/internal/modules/luckygift/handler.go @@ -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 { diff --git a/services/activity-service/configs/config.docker.yaml b/services/activity-service/configs/config.docker.yaml index dcce12bc..d9b553a1 100644 --- a/services/activity-service/configs/config.docker.yaml +++ b/services/activity-service/configs/config.docker.yaml @@ -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" diff --git a/services/activity-service/configs/config.tencent.example.yaml b/services/activity-service/configs/config.tencent.example.yaml index 4057b2a7..c186e4ca 100644 --- a/services/activity-service/configs/config.tencent.example.yaml +++ b/services/activity-service/configs/config.tencent.example.yaml @@ -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" diff --git a/services/activity-service/configs/config.yaml b/services/activity-service/configs/config.yaml index 884ce5c9..fdf50deb 100644 --- a/services/activity-service/configs/config.yaml +++ b/services/activity-service/configs/config.yaml @@ -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" diff --git a/services/activity-service/deploy/mysql/initdb/001_activity_service.sql b/services/activity-service/deploy/mysql/initdb/001_activity_service.sql index 11692dc6..8ac41c96 100644 --- a/services/activity-service/deploy/mysql/initdb/001_activity_service.sql +++ b/services/activity-service/deploy/mysql/initdb/001_activity_service.sql @@ -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', diff --git a/services/activity-service/internal/app/app.go b/services/activity-service/internal/app/app.go index 894a1d7e..d0b8d1ab 100644 --- a/services/activity-service/internal/app/app.go +++ b/services/activity-service/internal/app/app.go @@ -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, diff --git a/services/activity-service/internal/config/config.go b/services/activity-service/internal/config/config.go index 59fdce61..cc2320a5 100644 --- a/services/activity-service/internal/config/config.go +++ b/services/activity-service/internal/config/config.go @@ -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 } diff --git a/services/activity-service/internal/service/luckygift/service.go b/services/activity-service/internal/service/luckygift/service.go index 101e0a72..310d4a3f 100644 --- a/services/activity-service/internal/service/luckygift/service.go +++ b/services/activity-service/internal/service/luckygift/service.go @@ -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 } diff --git a/services/activity-service/internal/service/luckygift/service_test.go b/services/activity-service/internal/service/luckygift/service_test.go index 89fbe7cd..8e0ed52b 100644 --- a/services/activity-service/internal/service/luckygift/service_test.go +++ b/services/activity-service/internal/service/luckygift/service_test.go @@ -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 { diff --git a/services/activity-service/internal/storage/mysql/lucky_gift_repository.go b/services/activity-service/internal/storage/mysql/lucky_gift_repository.go index 1a731e2a..3591d6a4 100644 --- a/services/activity-service/internal/storage/mysql/lucky_gift_repository.go +++ b/services/activity-service/internal/storage/mysql/lucky_gift_repository.go @@ -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} diff --git a/services/activity-service/internal/storage/mysql/repository.go b/services/activity-service/internal/storage/mysql/repository.go index 8b140ead..da7eb6af 100644 --- a/services/activity-service/internal/storage/mysql/repository.go +++ b/services/activity-service/internal/storage/mysql/repository.go @@ -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 { diff --git a/services/game-service/deploy/mysql/initdb/001_game_service.sql b/services/game-service/deploy/mysql/initdb/001_game_service.sql index 39a3953c..655fa794 100644 --- a/services/game-service/deploy/mysql/initdb/001_game_service.sql +++ b/services/game-service/deploy/mysql/initdb/001_game_service.sql @@ -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 '业务状态', diff --git a/services/game-service/internal/domain/game/game.go b/services/game-service/internal/domain/game/game.go index f1639eb6..fc9e3957 100644 --- a/services/game-service/internal/domain/game/game.go +++ b/services/game-service/internal/domain/game/game.go @@ -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 diff --git a/services/game-service/internal/service/game/service.go b/services/game-service/internal/service/game/service.go index 30284de6..047e5304 100644 --- a/services/game-service/internal/service/game/service.go +++ b/services/game-service/internal/service/game/service.go @@ -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, diff --git a/services/game-service/internal/service/game/service_test.go b/services/game-service/internal/service/game/service_test.go index 08351335..288a5303 100644 --- a/services/game-service/internal/service/game/service_test.go +++ b/services/game-service/internal/service/game/service_test.go @@ -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) } diff --git a/services/game-service/internal/service/game/yomi_adapter.go b/services/game-service/internal/service/game/yomi_adapter.go index 47d5fb92..37edfbb8 100644 --- a/services/game-service/internal/service/game/yomi_adapter.go +++ b/services/game-service/internal/service/game/yomi_adapter.go @@ -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, diff --git a/services/game-service/internal/storage/mysql/repository.go b/services/game-service/internal/storage/mysql/repository.go index 2c2f0bdc..79de8bd6 100644 --- a/services/game-service/internal/storage/mysql/repository.go +++ b/services/game-service/internal/storage/mysql/repository.go @@ -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 diff --git a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql index 75ed90da..34e69490 100644 --- a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql +++ b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql @@ -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 '游戏参与去重人数', diff --git a/services/statistics-service/internal/app/app.go b/services/statistics-service/internal/app/app.go index 36d19fac..97cdb79a 100644 --- a/services/statistics-service/internal/app/app.go +++ b/services/statistics-service/internal/app/app.go @@ -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 +} diff --git a/services/statistics-service/internal/app/local_flow_test.go b/services/statistics-service/internal/app/local_flow_test.go new file mode 100644 index 00000000..ef679489 --- /dev/null +++ b/services/statistics-service/internal/app/local_flow_test.go @@ -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) +} diff --git a/services/statistics-service/internal/storage/mysql/query.go b/services/statistics-service/internal/storage/mysql/query.go index 552e94fd..59f81a44 100644 --- a/services/statistics-service/internal/storage/mysql/query.go +++ b/services/statistics-service/internal/storage/mysql/query.go @@ -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) diff --git a/services/statistics-service/internal/storage/mysql/repository.go b/services/statistics-service/internal/storage/mysql/repository.go index eb5bc2c9..f7025333 100644 --- a/services/statistics-service/internal/storage/mysql/repository.go +++ b/services/statistics-service/internal/storage/mysql/repository.go @@ -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) diff --git a/services/user-service/internal/service/auth/quick_account.go b/services/user-service/internal/service/auth/quick_account.go index c170aa1d..7f99cd26 100644 --- a/services/user-service/internal/service/auth/quick_account.go +++ b/services/user-service/internal/service/auth/quick_account.go @@ -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 diff --git a/services/user-service/internal/service/auth/service_test.go b/services/user-service/internal/service/auth/service_test.go index 77a8d94f..aec09df6 100644 --- a/services/user-service/internal/service/auth/service_test.go +++ b/services/user-service/internal/service/auth/service_test.go @@ -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 })) diff --git a/services/user-service/internal/service/auth/third_party.go b/services/user-service/internal/service/auth/third_party.go index 6e38e41c..417c704c 100644 --- a/services/user-service/internal/service/auth/third_party.go +++ b/services/user-service/internal/service/auth/third_party.go @@ -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 diff --git a/services/user-service/internal/transport/grpc/server_test.go b/services/user-service/internal/transport/grpc/server_test.go index 5eaa48da..5f603b20 100644 --- a/services/user-service/internal/transport/grpc/server_test.go +++ b/services/user-service/internal/transport/grpc/server_test.go @@ -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", }) diff --git a/services/wallet-service/internal/domain/ledger/ledger.go b/services/wallet-service/internal/domain/ledger/ledger.go index 8cf4a747..0cf6307f 100644 --- a/services/wallet-service/internal/domain/ledger/ledger.go +++ b/services/wallet-service/internal/domain/ledger/ledger.go @@ -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 是幸运礼物返奖入账后的稳定回执。 diff --git a/services/wallet-service/internal/storage/mysql/repository.go b/services/wallet-service/internal/storage/mysql/repository.go index fb283e6c..80da4d64 100644 --- a/services/wallet-service/internal/storage/mysql/repository.go +++ b/services/wallet-service/internal/storage/mysql/repository.go @@ -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, } diff --git a/services/wallet-service/internal/transport/grpc/server.go b/services/wallet-service/internal/transport/grpc/server.go index 7cd2dfeb..3c5dfc32 100644 --- a/services/wallet-service/internal/transport/grpc/server.go +++ b/services/wallet-service/internal/transport/grpc/server.go @@ -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)