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

feat: simplify lucky gift settlement outbox

Browse Source
This commit is contained in:
zhx 2026-06-02 23:45:46 +08:00
parent 10a3df3a5a
commit 858b0993bc
13 changed files with 903 additions and 451 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -21,10 +21,12 @@ lucky_gift_worker:
enabled: true
worker_poll_interval: "1s"
worker_batch_size: 100
worker_concurrency: 4
worker_concurrency: 8
worker_lock_ttl: "30s"
worker_max_retry: 8
publish_timeout: "5s"
stats_refresh_interval: "10m"
stats_batch_size: 5000
tencent_im:
# Docker 本地联调全局/区域播报群；必须与 gateway/room-service 使用同一组腾讯 IM 应用配置。
enabled: true

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

@ -19,12 +19,14 @@ red_packet_broadcast_worker:
enabled: true
lucky_gift_worker:
enabled: true
worker_poll_interval: "60s"
worker_batch_size: 10
worker_concurrency: 4
worker_poll_interval: "1s"
worker_batch_size: 100
worker_concurrency: 8
worker_lock_ttl: "30s"
worker_max_retry: 8
publish_timeout: "5s"
stats_refresh_interval: "10m"
stats_batch_size: 5000
tencent_im:
# 腾讯云 IM 应用配置；必须和 gateway 的 UserSig、room-service 房间群配置属于同一个 SDKAppID。
enabled: true

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

@ -21,10 +21,12 @@ lucky_gift_worker:
enabled: true
worker_poll_interval: "1s"
worker_batch_size: 100
worker_concurrency: 4
worker_concurrency: 8
worker_lock_ttl: "30s"
worker_max_retry: 8
publish_timeout: "5s"
stats_refresh_interval: "10m"
stats_batch_size: 5000
tencent_im:
# activity-service 只负责全局/区域播报群；必须与 gateway/room-service 使用同一组腾讯 IM 应用配置。
enabled: true

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

@ -256,7 +256,10 @@ CREATE TABLE IF NOT EXISTS lucky_draw_records (
KEY idx_lucky_draw_user (app_code, user_id, created_at_ms),
KEY idx_lucky_draw_gift (app_code, gift_id, created_at_ms),
KEY idx_lucky_draw_room (app_code, room_id, created_at_ms),
KEY idx_lucky_draw_status (app_code, reward_status, updated_at_ms)
KEY idx_lucky_draw_status (app_code, reward_status, updated_at_ms),
KEY idx_lucky_draw_created (app_code, created_at_ms, draw_id),
KEY idx_lucky_draw_pool_status_summary (app_code, pool_id, reward_status, created_at_ms, draw_id),
KEY idx_lucky_draw_gift_status_summary (app_code, pool_id, gift_id, reward_status, created_at_ms, draw_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物抽奖事实';
CREATE TABLE IF NOT EXISTS lucky_draw_pool_stats (
@ -297,6 +300,16 @@ CREATE TABLE IF NOT EXISTS lucky_draw_pool_stat_rooms (
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 lucky_draw_pool_stat_cursors (
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码',
cursor_name VARCHAR(64) NOT NULL COMMENT '游标名称',
last_draw_created_at_ms BIGINT NOT NULL DEFAULT 0 COMMENT '最后已聚合 draw 创建时间',
last_draw_id VARCHAR(128) NOT NULL DEFAULT '' COMMENT '同一毫秒内最后已聚合 draw ID',
created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms',
updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms',
PRIMARY KEY (app_code, cursor_name)
) 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',

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

@ -586,5 +586,7 @@ func luckyGiftWorkerOptions(nodeID string, cfg config.LuckyGiftWorkerConfig) luc
LockTTL: cfg.WorkerLockTTL,
MaxRetry: cfg.WorkerMaxRetry,
PublishTimeout: cfg.PublishTimeout,
StatsInterval: cfg.StatsRefreshInterval,
StatsBatchSize: cfg.StatsBatchSize,
}
}

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

@ -69,13 +69,15 @@ type RedPacketBroadcastWorkerConfig struct {
// LuckyGiftWorkerConfig 保存幸运礼物抽奖副作用补偿策略。
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"`
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"`
StatsRefreshInterval time.Duration `yaml:"stats_refresh_interval"`
StatsBatchSize int `yaml:"stats_batch_size"`
}
// TencentIMConfig 描述 activity-service 调用腾讯云 IM REST API 的配置。
@ -168,13 +170,15 @@ func Default() Config {
Enabled: false,
},
LuckyGiftWorker: LuckyGiftWorkerConfig{
Enabled: false,
WorkerPollInterval: time.Second,
WorkerBatchSize: 100,
WorkerConcurrency: 4,
WorkerLockTTL: 30 * time.Second,
WorkerMaxRetry: 8,
PublishTimeout: 5 * time.Second,
Enabled: false,
WorkerPollInterval: time.Second,
WorkerBatchSize: 100,
WorkerConcurrency: 4,
WorkerLockTTL: 30 * time.Second,
WorkerMaxRetry: 8,
PublishTimeout: 5 * time.Second,
StatsRefreshInterval: 10 * time.Minute,
StatsBatchSize: 5000,
},
TencentIM: TencentIMConfig{
Enabled: false,

3
services/activity-service/internal/domain/luckygift/lucky_gift.go
View File

@ -28,6 +28,9 @@ const (
BroadcastRoom = "room"
BroadcastRegion = "region"
BroadcastGlobal = "global"
EventTypeLuckyGiftDrawn = "LuckyGiftDrawn"
EventTypeLuckyGiftRewardSettlement = "LuckyGiftRewardSettlement"
)
// Tier 是单个体验池里的候选奖档；权重只塑造体验节奏，基础成本仍由 RTP 窗口兜住。

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

@ -35,6 +35,7 @@ type Repository interface {
MarkLuckyGiftOutboxDelivered(ctx context.Context, event domain.DrawOutbox, nowMS int64) error
MarkLuckyGiftOutboxRetryable(ctx context.Context, event domain.DrawOutbox, retryCount int, nextRetryAtMS int64, lastErr string, nowMS int64) error
MarkLuckyGiftOutboxFailed(ctx context.Context, event domain.DrawOutbox, retryCount int, lastErr string, nowMS int64) error
RefreshLuckyGiftAdminStatsSnapshots(ctx context.Context, nowMS int64, batchSize int) (int, error)
MarkLuckyGiftDrawGranted(ctx context.Context, appCode string, drawID string, walletTransactionID string, nowMS int64) error
MarkLuckyGiftDrawFailed(ctx context.Context, appCode string, drawID string, failureReason string, nowMS int64) error
// 批量抽奖只产生一条聚合 outbox；worker 处理后需要用 draw_ids 一次性收敛所有明细状态。
@ -152,17 +153,27 @@ type WorkerOptions struct {
LockTTL time.Duration
MaxRetry int
PublishTimeout time.Duration
StatsInterval time.Duration
StatsBatchSize int
}
func (s *Service) RunWorker(ctx context.Context, options WorkerOptions) {
options = normalizeWorkerOptions(options)
ticker := time.NewTicker(options.PollInterval)
defer ticker.Stop()
nextStatsRefresh := time.Time{}
for {
if _, err := s.ProcessPendingDrawOutbox(ctx, options); err != nil && ctx.Err() == nil {
// 批处理失败不删除 outbox；锁过期后下一轮或其他实例会继续补偿。
logx.Error(ctx, "lucky_gift_outbox_batch_failed", err, slog.String("worker_id", options.WorkerID))
}
now := s.now().UTC()
if options.StatsInterval > 0 && (nextStatsRefresh.IsZero() || !now.Before(nextStatsRefresh)) {
if _, err := s.repository.RefreshLuckyGiftAdminStatsSnapshots(ctx, now.UnixMilli(), options.StatsBatchSize); err != nil && ctx.Err() == nil {
logx.Error(ctx, "lucky_gift_admin_stats_refresh_failed", err, slog.String("worker_id", options.WorkerID))
}
nextStatsRefresh = now.Add(options.StatsInterval)
}
select {
case <-ctx.Done():
return
@ -176,7 +187,7 @@ func (s *Service) ProcessPendingDrawOutbox(ctx context.Context, options WorkerOp
return 0, err
}
options = normalizeWorkerOptions(options)
// outbox 先抢占再处理，避免多实例同时给同一 draw 返奖或重复投递房间 IM。
// outbox 只承载钱包返奖结算，避免多实例同时给同一 draw 返奖。
events, err := s.repository.ClaimPendingLuckyGiftOutbox(ctx, options.WorkerID, s.now().UTC().UnixMilli(), options.LockTTL, options.BatchSize)
if err != nil {
return 0, err
@ -238,56 +249,55 @@ func (s *Service) processDrawOutbox(ctx context.Context, event domain.DrawOutbox
return s.markOutboxFailed(ctx, event, options, err, false)
}
eventCtx := appcode.WithContext(ctx, event.AppCode)
switch event.EventType {
case domain.EventTypeLuckyGiftRewardSettlement, domain.EventTypeLuckyGiftDrawn, "":
return s.processRewardSettlementOutbox(eventCtx, event, payload, options)
default:
return s.markOutboxFailed(eventCtx, event, options, fmt.Errorf("unsupported lucky gift outbox event type: %s", event.EventType), false)
}
}
func (s *Service) processRewardSettlementOutbox(ctx context.Context, event domain.DrawOutbox, payload luckyGiftDrawnPayload, options WorkerOptions) error {
if _, _, err := s.settleLuckyGiftReward(ctx, event, payload, options); err != nil {
return err
}
return s.repository.MarkLuckyGiftOutboxDelivered(ctx, event, s.now().UTC().UnixMilli())
}
func (s *Service) settleLuckyGiftReward(ctx context.Context, event domain.DrawOutbox, payload luckyGiftDrawnPayload, options WorkerOptions) (bool, bool, error) {
walletTransactionID := ""
credited := false
alreadyGranted := false
if payload.EffectiveRewardCoins > 0 {
state, err := s.repository.GetLuckyGiftDrawRewardState(eventCtx, event.AppCode, payload.DrawIDs)
state, err := s.repository.GetLuckyGiftDrawRewardState(ctx, event.AppCode, payload.DrawIDs)
if err != nil {
return s.markOutboxFailed(eventCtx, event, options, err, false)
return false, false, s.markOutboxFailed(ctx, event, options, err, false)
}
alreadyGranted = state.AllGranted
if alreadyGranted {
walletTransactionID = state.WalletTransactionID
credited = true
} else {
receipt, err := s.creditLuckyGiftReward(eventCtx, event.AppCode, payload)
receipt, err := s.creditLuckyGiftReward(ctx, event.AppCode, payload)
if err != nil {
return s.markOutboxFailed(eventCtx, event, options, err, false)
return false, false, s.markOutboxFailed(ctx, event, options, err, false)
}
walletTransactionID = receipt.WalletTransactionID
credited = true
}
}
if credited && !alreadyGranted {
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)
}
if shouldPublishLuckyGiftRegionBroadcast(payload) {
if s.broadcaster == nil {
return s.markOutboxFailed(eventCtx, event, options, fmt.Errorf("lucky gift region broadcaster is not configured"), credited)
}
// 区域播报先写 broadcast outbox，不直接发 IM；这里失败意味着大额表现未落地，需要保留重试。
if err := s.publishLuckyGiftRegionBroadcast(eventCtx, payload); err != nil {
return s.markOutboxFailed(eventCtx, event, options, err, credited)
}
}
if err := s.publishLuckyGiftDrawn(eventCtx, payload, options); err != nil {
return s.markOutboxFailed(eventCtx, event, options, err, credited)
}
nowMS := s.now().UTC().UnixMilli()
if !credited {
if err := s.repository.MarkLuckyGiftDrawsGranted(eventCtx, event.AppCode, payload.DrawIDs, walletTransactionID, nowMS); err != nil {
return err
if err := s.markLuckyGiftDrawsGrantedAfterWallet(ctx, event.AppCode, payload.DrawIDs, walletTransactionID); err != nil {
return false, false, err
}
return false, false, nil
}
if !alreadyGranted {
if err := s.markLuckyGiftDrawsGrantedAfterWallet(ctx, event.AppCode, payload.DrawIDs, walletTransactionID); err != nil {
return credited, alreadyGranted, err
}
}
return s.repository.MarkLuckyGiftOutboxDelivered(eventCtx, event, nowMS)
return credited, alreadyGranted, nil
}
type luckyGiftRewardReceipt struct {
@ -337,7 +347,7 @@ func (s *Service) creditLuckyGiftRewardFastPath(ctx context.Context, cmd domain.
result.RewardStatus = domain.StatusGranted
result.WalletTransactionID = receipt.WalletTransactionID
result.CoinBalanceAfter = receipt.CoinBalanceAfter
if err := s.repository.MarkLuckyGiftDrawsGranted(ctx, payload.AppCode, payload.DrawIDs, receipt.WalletTransactionID, s.now().UTC().UnixMilli()); err != nil {
if err := s.markLuckyGiftDrawsGrantedAfterWallet(ctx, payload.AppCode, payload.DrawIDs, receipt.WalletTransactionID); err != nil {
logx.Error(ctx, "lucky_gift_reward_fast_path_mark_granted_failed", err,
slog.String("draw_id", result.DrawID),
slog.String("command_id", result.CommandID),
@ -347,6 +357,25 @@ func (s *Service) creditLuckyGiftRewardFastPath(ctx context.Context, cmd domain.
return result
}
func (s *Service) markLuckyGiftDrawsGrantedAfterWallet(ctx context.Context, appCode string, drawIDs []string, walletTransactionID string) error {
baseCtx := context.WithoutCancel(ctx)
var lastErr error
for attempt := 0; attempt < 3; attempt++ {
markCtx, cancel := context.WithTimeout(appcode.WithContext(baseCtx, appCode), 5*time.Second)
err := s.repository.MarkLuckyGiftDrawsGranted(markCtx, appCode, drawIDs, walletTransactionID, s.now().UTC().UnixMilli())
cancel()
if err == nil {
return nil
}
lastErr = err
if !isRetryableLuckyGrantMarkError(err) {
return err
}
time.Sleep(time.Duration(attempt+1) * 50 * time.Millisecond)
}
return lastErr
}
func (s *Service) creditLuckyGiftReward(ctx context.Context, appCode string, payload luckyGiftDrawnPayload) (luckyGiftRewardReceipt, error) {
if s.wallet == nil {
return luckyGiftRewardReceipt{}, fmt.Errorf("lucky gift wallet client is not configured")
@ -377,7 +406,7 @@ func (s *Service) publishLuckyGiftDrawn(ctx context.Context, payload luckyGiftDr
if err != nil {
return err
}
// 房间 IM 只是表现层事实同步，不修改 Room Cell；超时由 outbox 重试兜底。
// 房间 IM 只是表现层事实同步，不参与钱包结算和 reward_status 收敛。
publishCtx, cancel := context.WithTimeout(ctx, options.PublishTimeout)
defer cancel()
return s.publisher.PublishGroupCustomMessage(publishCtx, tencentim.CustomGroupMessage{
@ -411,12 +440,12 @@ func (s *Service) markOutboxFailed(ctx context.Context, event domain.DrawOutbox,
nextRetryCount := event.RetryCount + 1
if nextRetryCount >= options.MaxRetry {
if !alreadyCredited {
// 未返奖的 draw 可以整批标记 failed；已返奖但 IM 失败时不能覆盖账务事实，只停 outbox 并告警。
// 未返奖的 draw 可以整批标记 failed；钱包已返奖时不能覆盖账务事实。
_ = s.repository.MarkLuckyGiftDrawsFailed(ctx, event.AppCode, drawIDsFromPayload(event.PayloadJSON), cause.Error(), nowMS)
}
return s.repository.MarkLuckyGiftOutboxFailed(ctx, event, nextRetryCount, cause.Error(), nowMS)
}
// 指数退避降低故障期间对钱包和 IM 的压力，同时保留 pending 状态给客户端展示“处理中”。
// 指数退避降低故障期间对钱包的压力，同时保留 pending 状态给客户端展示“处理中”。
nextRetryAtMS := nowMS + luckyGiftBackoff(nextRetryCount).Milliseconds()
return s.repository.MarkLuckyGiftOutboxRetryable(ctx, event, nextRetryCount, nextRetryAtMS, cause.Error(), nowMS)
}
@ -676,6 +705,15 @@ func normalizeWorkerOptions(options WorkerOptions) WorkerOptions {
if options.PublishTimeout <= 0 {
options.PublishTimeout = 5 * time.Second
}
if options.StatsInterval <= 0 {
options.StatsInterval = 10 * time.Minute
}
if options.StatsBatchSize <= 0 {
options.StatsBatchSize = 5000
}
if options.StatsBatchSize > 50000 {
options.StatsBatchSize = 50000
}
return options
}
@ -688,3 +726,14 @@ func luckyGiftBackoff(retryCount int) time.Duration {
}
return time.Duration(1<<retryCount) * time.Second
}
func isRetryableLuckyGrantMarkError(err error) bool {
if err == nil {
return false
}
message := err.Error()
return strings.Contains(message, "Deadlock") ||
strings.Contains(message, "1213") ||
strings.Contains(message, "40001") ||
strings.Contains(message, "Lock wait timeout")
}

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

@ -127,7 +127,7 @@ func TestDrawKeepsPendingWhenFastPathWalletFails(t *testing.T) {
}
}
func TestProcessPendingDrawOutboxCreditsWalletAndPublishesRoomIM(t *testing.T) {
func TestProcessPendingDrawOutboxCreditsWalletOnly(t *testing.T) {
payload, _ := json.Marshal(map[string]any{
"event_type": "lucky_gift_drawn",
"event_id": "lucky_gift_drawn:lucky_draw_test",
@ -172,18 +172,8 @@ func TestProcessPendingDrawOutboxCreditsWalletAndPublishesRoomIM(t *testing.T) {
if wallet.last == nil || wallet.last.GetDrawId() != "lucky_draw_test" || wallet.last.GetTargetUserId() != 42 || wallet.last.GetAmount() != 300 {
t.Fatalf("wallet reward request mismatch: %+v", wallet.last)
}
if publisher.last.GroupID != "room-1" || publisher.last.Desc != "lucky_gift_drawn" || publisher.last.EventID != "lucky_gift_drawn:lucky_draw_test" {
t.Fatalf("room im message mismatch: %+v", publisher.last)
}
if broadcaster.last.EventID != "lucky_gift_big_win:lucky_draw_test" || broadcaster.last.BroadcastType != broadcastdomain.TypeLuckyGiftBigWin || broadcaster.last.RegionID != 210 {
t.Fatalf("region lucky broadcast mismatch: %+v", broadcaster.last)
}
var broadcastPayload map[string]any
if err := json.Unmarshal([]byte(broadcaster.last.PayloadJSON), &broadcastPayload); err != nil {
t.Fatalf("decode region broadcast payload failed: %v", err)
}
if broadcastPayload["event_type"] != "lucky_gift_drawn" || broadcastPayload["broadcast_type"] != broadcastdomain.TypeLuckyGiftBigWin || broadcastPayload["room_id"] != "room-1" {
t.Fatalf("region broadcast payload mismatch: %+v", broadcastPayload)
if publisher.last.EventID != "" || broadcaster.last.EventID != "" {
t.Fatalf("settlement worker must not publish presentation: publisher=%+v broadcaster=%+v", publisher.last, broadcaster.last)
}
if repository.grantedDrawID != "lucky_draw_test" || repository.deliveredOutboxID != "lucky_lucky_draw_test" {
t.Fatalf("repository settlement mismatch: granted=%s delivered=%s", repository.grantedDrawID, repository.deliveredOutboxID)
@ -240,6 +230,51 @@ func TestProcessPendingBatchDrawOutboxCreditsWalletOnceAndGrantsAllDraws(t *test
}
}
func TestProcessRewardSettlementOutboxDoesNotPublishPresentation(t *testing.T) {
payload, _ := json.Marshal(map[string]any{
"event_type": "lucky_gift_drawn",
"event_id": "lucky_gift_drawn:lucky_draw_settle",
"app_code": "lalu",
"draw_id": "lucky_draw_settle",
"command_id": "cmd-gift-settle",
"pool_id": "super_lucky",
"user_id": 42,
"visible_region_id": 210,
"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_reward_lucky_draw_settle",
EventType: domain.EventTypeLuckyGiftRewardSettlement,
PayloadJSON: string(payload),
}},
}
wallet := &fakeLuckyGiftWallet{}
publisher := &fakeLuckyGiftPublisher{}
service := New(repository, WithWallet(wallet), WithRoomPublisher(publisher), WithRegionBroadcaster(&fakeLuckyGiftRegionBroadcaster{}))
processed, err := service.ProcessPendingDrawOutbox(context.Background(), WorkerOptions{WorkerID: "worker-1"})
if err != nil {
t.Fatalf("ProcessPendingDrawOutbox failed: %v", err)
}
if processed != 1 || repository.deliveredOutboxID != "lucky_reward_lucky_draw_settle" {
t.Fatalf("settlement outbox should be delivered: processed=%d delivered=%s", processed, repository.deliveredOutboxID)
}
if wallet.last == nil || repository.grantedDrawID != "lucky_draw_settle" {
t.Fatalf("settlement should credit wallet and mark draw granted: wallet=%+v granted=%s", wallet.last, repository.grantedDrawID)
}
if publisher.last.EventID != "" {
t.Fatalf("settlement outbox must not publish presentation: %+v", publisher.last)
}
}
func TestPublishLuckyGiftDrawnOmitsBatchDrawIDsFromRoomIM(t *testing.T) {
drawIDs := make([]string, 0, 999)
for index := 0; index < 999; index++ {
@ -292,7 +327,7 @@ func TestPublishLuckyGiftDrawnOmitsBatchDrawIDsFromRoomIM(t *testing.T) {
}
}
func TestProcessPendingDrawOutboxGrantsRewardWhenRoomIMFails(t *testing.T) {
func TestProcessPendingDrawOutboxIgnoresRoomIMPublisher(t *testing.T) {
payload, _ := json.Marshal(map[string]any{
"event_type": "lucky_gift_drawn",
"event_id": "lucky_gift_drawn:lucky_draw_im_fail",
@ -324,19 +359,16 @@ func TestProcessPendingDrawOutboxGrantsRewardWhenRoomIMFails(t *testing.T) {
processed, err := service.ProcessPendingDrawOutbox(context.Background(), WorkerOptions{WorkerID: "worker-1"})
if err != nil {
t.Fatalf("ProcessPendingDrawOutbox should schedule retry without failing batch: %v", err)
t.Fatalf("ProcessPendingDrawOutbox failed: %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)
t.Fatalf("draw should be settled through wallet, 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)
if repository.deliveredOutboxID != "lucky_lucky_draw_im_fail" || repository.retryableOutboxID != "" {
t.Fatalf("IM publisher must not affect settlement outbox: delivered=%s retryable=%s", repository.deliveredOutboxID, repository.retryableOutboxID)
}
}
@ -383,8 +415,8 @@ func TestProcessPendingDrawOutboxSkipsWalletWhenDrawAlreadyGranted(t *testing.T)
if repository.grantedDrawID != "" {
t.Fatalf("already granted draw should not be marked again, got %q", repository.grantedDrawID)
}
if publisher.last.EventID != "lucky_gift_drawn:lucky_draw_granted" || repository.deliveredOutboxID != "lucky_lucky_draw_granted" {
t.Fatalf("already granted draw should still publish IM and deliver outbox: publisher=%+v delivered=%s", publisher.last, repository.deliveredOutboxID)
if publisher.last.EventID != "" || repository.deliveredOutboxID != "lucky_lucky_draw_granted" {
t.Fatalf("already granted draw should only deliver settlement outbox: publisher=%+v delivered=%s", publisher.last, repository.deliveredOutboxID)
}
}
@ -455,6 +487,10 @@ func (r *fakeLuckyGiftRepository) MarkLuckyGiftOutboxFailed(context.Context, dom
return nil
}
func (r *fakeLuckyGiftRepository) RefreshLuckyGiftAdminStatsSnapshots(context.Context, int64, int) (int, error) {
return 0, nil
}
func (r *fakeLuckyGiftRepository) MarkLuckyGiftDrawGranted(_ context.Context, _ string, drawID string, _ string, _ int64) error {
r.grantedDrawID = drawID
return nil

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

@ -134,11 +134,11 @@ type luckyBatchDrawState struct {
AtmosphereOut int64
ActivityReward int64
EffectiveReward int64
// NeedsOutbox 表示本批存在任意真实异步副作用：钱包加币、房间展示或阶段反馈。
// 只要为 true，本批所有子抽奖先统一 pending，worker 成功后一起 granted。
NeedsOutbox bool
StageFeedback bool
HighMultiplier bool
// 只有钱包结算会影响 reward_status=pending；IM/广播表现层不进入可靠 outbox。
NeedsRewardSettlement bool
StageFeedback bool
Presentation bool
HighMultiplier bool
// Final* 是用户体验状态的最终值；批量抽奖不能每抽 UPDATE lucky_user_states。
FinalLossStreak int64
FinalCumulative int64
@ -370,13 +370,9 @@ func (r *Repository) executeOptimizedLuckyGiftDrawBatch(ctx context.Context, tx
state.AggregateSources = aggregate.BudgetSourcesJSON
state.AggregateStatus = aggregate.RewardStatus
state.AggregateCommandID = cmd.CommandID
if state.NeedsOutbox {
// 批量命令只需要一次钱包发放和一次房间表现；所有子抽奖共享这条 outbox 的最终收敛状态。
if state.NeedsRewardSettlement {
// 批量命令只需要一次钱包发放；所有子抽奖共享这条 settlement outbox 的最终收敛状态。
state.AggregateResult.RewardStatus = domain.StatusPending
for index := range state.Records {
state.Records[index].RewardStatus = domain.StatusPending
results[index].RewardStatus = domain.StatusPending
}
} else {
state.AggregateResult.RewardStatus = domain.StatusGranted
}
@ -435,9 +431,10 @@ func (r *Repository) stageLuckyBatchDraw(appCode string, baseConfig domain.Confi
state.FinalLossStreak++
}
state.StageFeedback = state.StageFeedback || stageFeedback
state.Presentation = state.Presentation || candidate.Presentation
state.HighMultiplier = state.HighMultiplier || candidate.HighMultiplier
if luckyDrawNeedsOutbox(candidate, stageFeedback) {
state.NeedsOutbox = true
if luckyDrawNeedsRewardSettlementOutbox(candidate) {
state.NeedsRewardSettlement = true
}
// 每个子抽必须有独立 draw_id，保证审计、列表和重试读回都能定位到具体一次随机结果。
@ -576,12 +573,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)
if state.NeedsRewardSettlement {
// 主事务只写返奖结算 outbox；后台统计由低频快照任务从 lucky_draw_records 增量刷新。
return r.insertLuckyAggregateRewardSettlementOutbox(ctx, tx, appCode, cmd, state, nowMS)
}
return nil
}
@ -669,69 +663,10 @@ type luckyDrawPoolStatDelta struct {
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 {
@ -766,7 +701,7 @@ func (r *Repository) insertLuckyDrawStatRooms(ctx context.Context, tx *sql.Tx, k
return inserted, nil
}
func (r *Repository) upsertLuckyDrawPoolStatDelta(ctx context.Context, tx *sql.Tx, key luckyDrawPoolStatKey, delta luckyDrawPoolStatDelta, uniqueUsers int64, uniqueRooms int64, nowMS int64) error {
func (r *Repository) upsertLuckyDrawPoolStatsSnapshotDelta(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,
@ -778,21 +713,186 @@ func (r *Repository) upsertLuckyDrawPoolStatDelta(ctx context.Context, tx *sql.T
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)`,
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),
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,
int64(0), int64(0), int64(0), nowMS, nowMS,
)
return err
}
type luckyDrawStatsCursor struct {
LastDrawCreatedAtMS int64
LastDrawID string
}
type luckyDrawStatsRecord struct {
AppCode string
DrawID string
PoolID string
GiftID string
UserID int64
RoomID string
TotalSpentCoins int64
TotalRewardCoins int64
BaseRewardCoins int64
RoomAtmosphereRewardCoins int64
ActivitySubsidyCoins int64
CreatedAtMS int64
}
func (r *Repository) RefreshLuckyGiftAdminStatsSnapshots(ctx context.Context, nowMS int64, batchSize int) (int, error) {
if r == nil || r.db == nil {
return 0, xerr.New(xerr.Unavailable, "mysql repository is not configured")
}
if batchSize <= 0 {
batchSize = 5000
}
if batchSize > 50000 {
batchSize = 50000
}
appCode := appcode.FromContext(ctx)
tx, err := r.db.BeginTx(ctx, nil)
if err != nil {
return 0, err
}
defer func() { _ = tx.Rollback() }()
cursor, initialized, err := r.lockLuckyDrawStatsCursor(ctx, tx, appCode, nowMS)
if err != nil {
return 0, err
}
if initialized {
if err := tx.Commit(); err != nil {
return 0, err
}
return 0, nil
}
rows, err := tx.QueryContext(ctx, `
SELECT app_code, draw_id, pool_id, gift_id, user_id, room_id, coin_spent,
total_reward_coins, base_reward_coins, room_atmosphere_reward_coins, activity_subsidy_coins,
created_at_ms
FROM lucky_draw_records FORCE INDEX (idx_lucky_draw_created)
WHERE app_code = ?
AND (created_at_ms > ? OR (created_at_ms = ? AND draw_id > ?))
ORDER BY created_at_ms ASC, draw_id ASC
LIMIT ?`, appCode, cursor.LastDrawCreatedAtMS, cursor.LastDrawCreatedAtMS, cursor.LastDrawID, batchSize)
if err != nil {
return 0, err
}
defer rows.Close()
records := make([]luckyDrawStatsRecord, 0, batchSize)
for rows.Next() {
var record luckyDrawStatsRecord
if err := rows.Scan(&record.AppCode, &record.DrawID, &record.PoolID, &record.GiftID, &record.UserID, &record.RoomID,
&record.TotalSpentCoins, &record.TotalRewardCoins, &record.BaseRewardCoins,
&record.RoomAtmosphereRewardCoins, &record.ActivitySubsidyCoins, &record.CreatedAtMS); err != nil {
return 0, err
}
records = append(records, record)
}
if err := rows.Err(); err != nil {
return 0, err
}
if len(records) == 0 {
if err := tx.Commit(); err != nil {
return 0, err
}
return 0, nil
}
grouped := map[luckyDrawPoolStatKey]*luckyDrawPoolStatDelta{}
last := records[len(records)-1]
for _, record := range records {
keys := []luckyDrawPoolStatKey{{AppCode: record.AppCode, PoolID: record.PoolID, GiftID: ""}}
if giftID := strings.TrimSpace(record.GiftID); giftID != "" {
keys = append(keys, luckyDrawPoolStatKey{AppCode: record.AppCode, PoolID: record.PoolID, GiftID: giftID})
}
for _, key := range keys {
delta := grouped[key]
if delta == nil {
delta = &luckyDrawPoolStatDelta{UserIDs: map[int64]struct{}{}, RoomIDs: map[string]struct{}{}}
grouped[key] = delta
}
delta.TotalDraws++
delta.TotalSpentCoins += record.TotalSpentCoins
delta.TotalRewardCoins += record.TotalRewardCoins
delta.BaseRewardCoins += record.BaseRewardCoins
delta.RoomAtmosphereRewardCoins += record.RoomAtmosphereRewardCoins
delta.ActivitySubsidyCoins += record.ActivitySubsidyCoins
if record.UserID > 0 {
delta.UserIDs[record.UserID] = struct{}{}
}
if roomID := strings.TrimSpace(record.RoomID); roomID != "" {
delta.RoomIDs[roomID] = struct{}{}
}
}
}
for key, delta := range grouped {
uniqueUsers, err := r.insertLuckyDrawStatUsers(ctx, tx, key, delta.UserIDs, nowMS)
if err != nil {
return 0, err
}
uniqueRooms, err := r.insertLuckyDrawStatRooms(ctx, tx, key, delta.RoomIDs, nowMS)
if err != nil {
return 0, err
}
if err := r.upsertLuckyDrawPoolStatsSnapshotDelta(ctx, tx, key, *delta, uniqueUsers, uniqueRooms, nowMS); err != nil {
return 0, err
}
}
if _, err := tx.ExecContext(ctx, `
UPDATE lucky_draw_pool_stat_cursors
SET last_draw_created_at_ms = ?, last_draw_id = ?, updated_at_ms = ?
WHERE app_code = ? AND cursor_name = 'draw_records'`,
last.CreatedAtMS, last.DrawID, nowMS, appCode,
); err != nil {
return 0, err
}
if err := tx.Commit(); err != nil {
return 0, err
}
return len(records), nil
}
func (r *Repository) lockLuckyDrawStatsCursor(ctx context.Context, tx *sql.Tx, appCode string, nowMS int64) (luckyDrawStatsCursor, bool, error) {
var cursor luckyDrawStatsCursor
err := tx.QueryRowContext(ctx, `
SELECT last_draw_created_at_ms, last_draw_id
FROM lucky_draw_pool_stat_cursors
WHERE app_code = ? AND cursor_name = 'draw_records'
FOR UPDATE`, appCode,
).Scan(&cursor.LastDrawCreatedAtMS, &cursor.LastDrawID)
if err == nil {
return cursor, false, nil
}
if !errors.Is(err, sql.ErrNoRows) {
return luckyDrawStatsCursor{}, false, err
}
if err := tx.QueryRowContext(ctx, `
SELECT COALESCE(created_at_ms, 0), COALESCE(draw_id, '')
FROM lucky_draw_records FORCE INDEX (idx_lucky_draw_created)
WHERE app_code = ?
ORDER BY created_at_ms DESC, draw_id DESC
LIMIT 1`, appCode,
).Scan(&cursor.LastDrawCreatedAtMS, &cursor.LastDrawID); err != nil && !errors.Is(err, sql.ErrNoRows) {
return luckyDrawStatsCursor{}, false, err
}
if _, err := tx.ExecContext(ctx, `
INSERT INTO lucky_draw_pool_stat_cursors (
app_code, cursor_name, last_draw_created_at_ms, last_draw_id, created_at_ms, updated_at_ms
) VALUES (?, 'draw_records', ?, ?, ?, ?)`,
appCode, cursor.LastDrawCreatedAtMS, cursor.LastDrawID, nowMS, nowMS,
); err != nil {
return luckyDrawStatsCursor{}, false, err
}
return cursor, true, nil
}
// luckyDrawRecordArgs 固定 lucky_draw_records 的列顺序；集中维护可以避免批量 INSERT 和单条 INSERT 语义漂移。
func luckyDrawRecordArgs(input luckyDrawRecordInput) []any {
candidateSnapshot, poolSnapshot, rtpSnapshot := luckyDrawRecordSnapshots(input)
@ -830,9 +930,9 @@ func luckyDrawRecordSnapshots(input luckyDrawRecordInput) (string, string, strin
return string(candidateSnapshot), string(poolSnapshot), string(rtpSnapshot)
}
// insertLuckyAggregateDrawOutbox 为整批抽奖只写一条副作用事件。
// draw_id 使用聚合响应的首个 draw，draw_ids 保存全部子抽；worker 用总 effective_reward_coins 发一次奖，再批量 granted。
func (r *Repository) insertLuckyAggregateDrawOutbox(ctx context.Context, tx *sql.Tx, appCode string, cmd domain.DrawCommand, state luckyBatchDrawState, nowMS int64) error {
// insertLuckyAggregateRewardSettlementOutbox 为整批抽奖写聚合返奖结算事件。
// draw_id 使用聚合响应的首个 draw，draw_ids 保存全部子抽；settlement 用总 effective_reward_coins 发一次奖，再批量 granted。
func (r *Repository) insertLuckyAggregateRewardSettlementOutbox(ctx context.Context, tx *sql.Tx, appCode string, cmd domain.DrawCommand, state luckyBatchDrawState, nowMS int64) error {
aggregate := state.AggregateResult
payload, _ := json.Marshal(map[string]any{
"event_type": "lucky_gift_drawn",
@ -862,14 +962,7 @@ func (r *Repository) insertLuckyAggregateDrawOutbox(ctx context.Context, tx *sql
"high_multiplier": aggregate.HighMultiplier,
"created_at_ms": nowMS,
})
_, err := tx.ExecContext(ctx, `
INSERT INTO activity_outbox (
app_code, outbox_id, event_type, payload, status, retry_count, next_retry_at_ms,
locked_by, lock_until_ms, last_error, created_at_ms, updated_at_ms
) VALUES (?, ?, 'LuckyGiftDrawn', ?, 'pending', 0, ?, '', 0, '', ?, ?)`,
appCode, "lucky_"+aggregate.DrawID, payload, nowMS, nowMS, nowMS,
)
return err
return r.insertLuckyRewardSettlementOutbox(ctx, tx, appCode, aggregate.DrawID, payload, nowMS)
}
func (r *Repository) executeSingleLuckyGiftDraw(ctx context.Context, tx *sql.Tx, cmd domain.DrawCommand, nowMS int64) (domain.DrawResult, error) {
@ -1278,29 +1371,8 @@ 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。
if luckyDrawNeedsRewardSettlementOutbox(input.Candidate) {
// outbox 只记录钱包返奖结算；表现层 IM/广播不参与可靠补偿链路。
payload, _ := json.Marshal(map[string]any{
"event_type": "lucky_gift_drawn",
"event_id": "lucky_gift_drawn:" + input.DrawID,
@ -1328,20 +1400,29 @@ func (r *Repository) applyLuckyDraw(ctx context.Context, tx *sql.Tx, input lucky
"high_multiplier": input.Candidate.HighMultiplier,
"created_at_ms": input.NowMS,
})
if _, err := tx.ExecContext(ctx, `
INSERT INTO activity_outbox (
app_code, outbox_id, event_type, payload, status, retry_count, next_retry_at_ms,
locked_by, lock_until_ms, last_error, created_at_ms, updated_at_ms
) VALUES (?, ?, 'LuckyGiftDrawn', ?, 'pending', 0, ?, '', 0, '', ?, ?)`,
input.AppCode, "lucky_"+input.DrawID, payload, input.NowMS, input.NowMS, input.NowMS,
); err != nil {
if err := r.insertLuckyRewardSettlementOutbox(ctx, tx, input.AppCode, input.DrawID, payload, input.NowMS); err != nil {
return err
}
}
return nil
}
// ClaimPendingLuckyGiftOutbox 抢占幸运礼物抽奖副作用。抽奖事实已在同一事务落库，worker 只补偿钱包和房间 IM。
func (r *Repository) insertLuckyRewardSettlementOutbox(ctx context.Context, tx *sql.Tx, appCode string, drawID string, payload []byte, nowMS int64) error {
return r.insertLuckyDrawOutbox(ctx, tx, appCode, "lucky_reward_"+drawID, domain.EventTypeLuckyGiftRewardSettlement, payload, nowMS)
}
func (r *Repository) insertLuckyDrawOutbox(ctx context.Context, tx *sql.Tx, appCode string, outboxID string, eventType string, payload []byte, nowMS int64) error {
_, err := tx.ExecContext(ctx, `
INSERT INTO activity_outbox (
app_code, outbox_id, event_type, payload, status, retry_count, next_retry_at_ms,
locked_by, lock_until_ms, last_error, created_at_ms, updated_at_ms
) VALUES (?, ?, ?, ?, 'pending', 0, ?, '', 0, '', ?, ?)`,
appCode, outboxID, eventType, payload, nowMS, nowMS, nowMS,
)
return err
}
// ClaimPendingLuckyGiftOutbox 抢占幸运礼物抽奖副作用。抽奖事实已在同一事务落库，worker 优先补偿影响 reward_status 的结算事件。
func (r *Repository) ClaimPendingLuckyGiftOutbox(ctx context.Context, workerID string, nowMS int64, lockTTL time.Duration, batchSize int) ([]domain.DrawOutbox, error) {
if r == nil || r.db == nil {
return nil, xerr.New(xerr.Unavailable, "mysql repository is not configured")
@ -1366,34 +1447,50 @@ func (r *Repository) ClaimPendingLuckyGiftOutbox(ctx context.Context, workerID s
defer func() { _ = tx.Rollback() }()
appCode := appcode.FromContext(ctx)
events, err := r.queryLuckyGiftOutboxCandidates(ctx, tx, `
SELECT app_code, outbox_id, event_type, COALESCE(CAST(payload AS CHAR), '{}'), retry_count, created_at_ms
FROM activity_outbox FORCE INDEX (idx_activity_outbox_lucky_retry)
WHERE app_code = ?
AND event_type = 'LuckyGiftDrawn'
AND status IN ('pending', 'retryable')
AND next_retry_at_ms <= ?
ORDER BY created_at_ms ASC, outbox_id ASC
LIMIT ?
FOR UPDATE`, batchSize, appCode, nowMS, batchSize)
if err != nil {
return nil, err
events := make([]domain.DrawOutbox, 0, batchSize)
priorityTypes := []string{
domain.EventTypeLuckyGiftRewardSettlement,
domain.EventTypeLuckyGiftDrawn,
}
if remaining := batchSize - len(events); remaining > 0 {
retryEvents, err := r.queryLuckyGiftOutboxCandidates(ctx, tx, `
for _, eventType := range priorityTypes {
candidates, err := r.queryLuckyGiftOutboxCandidates(ctx, tx, `
SELECT app_code, outbox_id, event_type, COALESCE(CAST(payload AS CHAR), '{}'), retry_count, created_at_ms
FROM activity_outbox FORCE INDEX (idx_activity_outbox_lucky_lock)
FROM activity_outbox FORCE INDEX (idx_activity_outbox_lucky_retry)
WHERE app_code = ?
AND event_type = 'LuckyGiftDrawn'
AND status = 'delivering'
AND lock_until_ms <= ?
AND event_type = ?
AND status IN ('pending', 'retryable')
AND next_retry_at_ms <= ?
ORDER BY created_at_ms ASC, outbox_id ASC
LIMIT ?
FOR UPDATE`, remaining, appCode, nowMS, remaining)
FOR UPDATE`, batchSize, appCode, eventType, nowMS, batchSize)
if err != nil {
return nil, err
}
events = append(events, retryEvents...)
if len(candidates) > 0 {
events = append(events, candidates...)
break
}
}
if len(events) == 0 {
for _, eventType := range priorityTypes {
retryEvents, err := r.queryLuckyGiftOutboxCandidates(ctx, tx, `
SELECT app_code, outbox_id, event_type, COALESCE(CAST(payload AS CHAR), '{}'), retry_count, created_at_ms
FROM activity_outbox FORCE INDEX (idx_activity_outbox_lucky_lock)
WHERE app_code = ?
AND event_type = ?
AND status = 'delivering'
AND lock_until_ms <= ?
ORDER BY created_at_ms ASC, outbox_id ASC
LIMIT ?
FOR UPDATE`, batchSize, appCode, eventType, nowMS, batchSize)
if err != nil {
return nil, err
}
if len(retryEvents) > 0 {
events = append(events, retryEvents...)
break
}
}
}
lockUntilMS := nowMS + lockTTL.Milliseconds()
for _, event := range events {
@ -1566,11 +1663,6 @@ func (r *Repository) markLuckyGiftDraws(ctx context.Context, appCode string, dra
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)
@ -1588,10 +1680,6 @@ func (r *Repository) markLuckyGiftDraws(ctx context.Context, appCode string, dra
_ = 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
}
@ -1599,96 +1687,6 @@ func (r *Repository) markLuckyGiftDraws(ctx context.Context, appCode string, dra
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{}
@ -1820,8 +1818,7 @@ func (r *Repository) getLuckyGiftDrawSummaryFromStats(ctx context.Context, query
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
base_reward_coins, room_atmosphere_reward_coins, activity_subsidy_coins
FROM lucky_draw_pool_stats
WHERE app_code = ? AND pool_id = ? AND gift_id = ?`,
appcode.FromContext(ctx), query.PoolID, strings.TrimSpace(query.GiftID),
@ -1834,22 +1831,88 @@ func (r *Repository) getLuckyGiftDrawSummaryFromStats(ctx context.Context, query
&summary.BaseRewardCoins,
&summary.RoomAtmosphereRewardCoins,
&summary.ActivitySubsidyCoins,
&summary.PendingDraws,
&summary.GrantedDraws,
&summary.FailedDraws,
)
if errors.Is(err, sql.ErrNoRows) {
return r.getLuckyGiftDrawSummaryFromRecords(ctx, query)
return summary, nil
}
if err != nil {
return domain.DrawSummary{}, err
}
if err := r.loadLuckyGiftDrawStatusCounts(ctx, query, &summary); err != nil {
return domain.DrawSummary{}, err
}
if summary.TotalSpentCoins > 0 {
summary.ActualRTPPPM = summary.TotalRewardCoins * luckyPPMScale / summary.TotalSpentCoins
}
return summary, nil
}
func (r *Repository) loadLuckyGiftDrawStatusCounts(ctx context.Context, query domain.DrawQuery, summary *domain.DrawSummary) error {
if summary == nil {
return nil
}
appCode := appcode.FromContext(ctx)
where := []string{"app_code = ?"}
args := []any{appCode}
if strings.TrimSpace(query.PoolID) != "" {
where = append(where, "pool_id = ?")
args = append(args, query.PoolID)
}
indexName := "idx_lucky_draw_pool_status_summary"
if strings.TrimSpace(query.GiftID) != "" {
indexName = "idx_lucky_draw_gift_status_summary"
where = append(where, "gift_id = ?")
args = append(args, strings.TrimSpace(query.GiftID))
}
if cursor, exists, err := r.getLuckyDrawStatsCursor(ctx, appCode); err != nil {
return err
} else if exists && (cursor.LastDrawCreatedAtMS > 0 || cursor.LastDrawID != "") {
where = append(where, "(created_at_ms < ? OR (created_at_ms = ? AND draw_id <= ?))")
args = append(args, cursor.LastDrawCreatedAtMS, cursor.LastDrawCreatedAtMS, cursor.LastDrawID)
}
rows, err := r.db.QueryContext(ctx, `
SELECT reward_status, COUNT(*)
FROM lucky_draw_records FORCE INDEX (`+indexName+`)
WHERE `+strings.Join(where, " AND ")+`
GROUP BY reward_status`, args...)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var status string
var count int64
if err := rows.Scan(&status, &count); err != nil {
return err
}
switch status {
case domain.StatusPending:
summary.PendingDraws = count
case domain.StatusGranted:
summary.GrantedDraws = count
case domain.StatusFailed:
summary.FailedDraws = count
}
}
return rows.Err()
}
func (r *Repository) getLuckyDrawStatsCursor(ctx context.Context, appCode string) (luckyDrawStatsCursor, bool, error) {
var cursor luckyDrawStatsCursor
err := r.db.QueryRowContext(ctx, `
SELECT last_draw_created_at_ms, last_draw_id
FROM lucky_draw_pool_stat_cursors
WHERE app_code = ? AND cursor_name = 'draw_records'`, appCode,
).Scan(&cursor.LastDrawCreatedAtMS, &cursor.LastDrawID)
if errors.Is(err, sql.ErrNoRows) {
return luckyDrawStatsCursor{}, false, nil
}
if err != nil {
return luckyDrawStatsCursor{}, false, err
}
return cursor, true, nil
}
func luckyDrawWhereClause(appCode string, query domain.DrawQuery) (string, []any) {
where := []string{"app_code = ?"}
args := []any{appCode}
@ -2747,17 +2810,15 @@ func luckyBudgetSourcesJSON(candidate luckyCandidate) string {
return string(raw)
}
func luckyInitialRewardStatus(candidate luckyCandidate, stageFeedback bool) string {
// 没有金币返还、也没有房间展示副作用时，抽奖事实已经终态化；不能留给不存在的 outbox 消费。
if !luckyDrawNeedsOutbox(candidate, stageFeedback) {
return domain.StatusGranted
func luckyInitialRewardStatus(candidate luckyCandidate, _ bool) string {
if luckyDrawNeedsRewardSettlementOutbox(candidate) {
return domain.StatusPending
}
return domain.StatusPending
return domain.StatusGranted
}
func luckyDrawNeedsOutbox(candidate luckyCandidate, stageFeedback bool) bool {
// outbox 只承载真实异步副作用：钱包入账、房间 IM 展示或阶段反馈展示。
return candidate.effectiveReward() > 0 || stageFeedback || candidate.Presentation
func luckyDrawNeedsRewardSettlementOutbox(candidate luckyCandidate) bool {
return candidate.effectiveReward() > 0
}
func luckyEnabledReason(enabled bool) string {

166
services/activity-service/internal/storage/mysql/lucky_gift_repository_test.go
View File

@ -1,9 +1,13 @@
package mysql
import (
"context"
"strings"
"testing"
"time"
"hyapp/internal/testutil/mysqlschema"
"hyapp/pkg/appcode"
domain "hyapp/services/activity-service/internal/domain/luckygift"
)
@ -36,6 +40,148 @@ func TestLuckyRuntimeConfigScalesTierRewardsByActualSpendButKeepsRiskCapsAbsolut
}
}
func TestClaimPendingLuckyGiftOutboxPrioritizesRewardSettlement(t *testing.T) {
schema := mysqlschema.New(t, mysqlschema.Config{
EnvVar: "ACTIVITY_SERVICE_MYSQL_TEST_DSN",
InitDBPath: mysqlschema.InitDBPath(t, mysqlschema.CallerFile(t, 1), "..", "..", "..", "deploy", "mysql", "initdb", "001_activity_service.sql"),
DatabasePrefix: "hyapp_activity_lucky_outbox_test",
})
repository, err := Open(context.Background(), schema.DSN)
if err != nil {
t.Fatalf("open repository: %v", err)
}
t.Cleanup(func() { _ = repository.Close() })
ctx := appcode.WithContext(context.Background(), "lalu")
nowMS := int64(1_700_000_000_000)
_, err = repository.db.ExecContext(ctx, `
INSERT INTO activity_outbox (
app_code, outbox_id, event_type, payload, status, retry_count, next_retry_at_ms,
locked_by, lock_until_ms, last_error, created_at_ms, updated_at_ms
) VALUES
('lalu', 'lucky_legacy_draw_1', ?, '{}', 'pending', 0, ?, '', 0, '', ?, ?),
('lalu', 'lucky_reward_draw_1', ?, '{}', 'pending', 0, ?, '', 0, '', ?, ?)`,
domain.EventTypeLuckyGiftDrawn, nowMS-100, nowMS-200, nowMS-200,
domain.EventTypeLuckyGiftRewardSettlement, nowMS-100, nowMS-100, nowMS-100,
)
if err != nil {
t.Fatalf("seed activity_outbox: %v", err)
}
events, err := repository.ClaimPendingLuckyGiftOutbox(ctx, "worker-priority", nowMS, 30*time.Second, 10)
if err != nil {
t.Fatalf("ClaimPendingLuckyGiftOutbox failed: %v", err)
}
if len(events) != 1 || events[0].EventType != domain.EventTypeLuckyGiftRewardSettlement || events[0].OutboxID != "lucky_reward_draw_1" {
t.Fatalf("claim should only return reward settlement first, got %+v", events)
}
var legacyStatus string
if err := repository.db.QueryRowContext(ctx, `
SELECT status FROM activity_outbox WHERE app_code = 'lalu' AND outbox_id = 'lucky_legacy_draw_1'`).Scan(&legacyStatus); err != nil {
t.Fatalf("read legacy outbox status: %v", err)
}
if legacyStatus != "pending" {
t.Fatalf("legacy outbox should remain pending while reward settlement exists, got %s", legacyStatus)
}
}
func TestRefreshLuckyGiftAdminStatsSnapshotsAdvancesCursorByBatch(t *testing.T) {
schema := mysqlschema.New(t, mysqlschema.Config{
EnvVar: "ACTIVITY_SERVICE_MYSQL_TEST_DSN",
InitDBPath: mysqlschema.InitDBPath(t, mysqlschema.CallerFile(t, 1), "..", "..", "..", "deploy", "mysql", "initdb", "001_activity_service.sql"),
DatabasePrefix: "hyapp_activity_lucky_stats_test",
})
repository, err := Open(context.Background(), schema.DSN)
if err != nil {
t.Fatalf("open repository: %v", err)
}
t.Cleanup(func() { _ = repository.Close() })
ctx := appcode.WithContext(context.Background(), "lalu")
if _, err := repository.db.ExecContext(ctx, `
INSERT INTO lucky_draw_pool_stat_cursors (
app_code, cursor_name, last_draw_created_at_ms, last_draw_id, created_at_ms, updated_at_ms
) VALUES ('lalu', 'draw_records', 0, '', 1, 1)`); err != nil {
t.Fatalf("seed stats cursor: %v", err)
}
seedLuckyDrawRecordForStats(t, repository, "draw_001", "cmd_001", domain.StatusGranted, 100, 100, 300)
seedLuckyDrawRecordForStats(t, repository, "draw_002", "cmd_002", domain.StatusPending, 200, 150, 0)
processed, err := repository.RefreshLuckyGiftAdminStatsSnapshots(ctx, 300, 1)
if err != nil {
t.Fatalf("first refresh failed: %v", err)
}
if processed != 1 {
t.Fatalf("first refresh should process one row, got %d", processed)
}
assertLuckyStatsTotal(t, repository, "", 1, 100, 300)
assertLuckyStatsCursor(t, repository, 100, "draw_001")
processed, err = repository.RefreshLuckyGiftAdminStatsSnapshots(ctx, 400, 10)
if err != nil {
t.Fatalf("second refresh failed: %v", err)
}
if processed != 1 {
t.Fatalf("second refresh should process remaining row, got %d", processed)
}
assertLuckyStatsTotal(t, repository, "", 2, 250, 300)
assertLuckyStatsCursor(t, repository, 200, "draw_002")
}
func seedLuckyDrawRecordForStats(t *testing.T, repository *Repository, drawID string, commandID string, status string, createdAtMS int64, spent int64, reward int64) {
t.Helper()
_, err := repository.db.ExecContext(context.Background(), `
INSERT INTO lucky_draw_records (
app_code, draw_id, command_id, user_id, device_id, room_id, anchor_id, pool_id, gift_id,
coin_spent, rule_version, experience_pool, rtp_window_index, gift_rtp_window_index,
selected_tier_id, base_reward_coins, room_atmosphere_reward_coins, activity_subsidy_coins,
effective_reward_coins, budget_sources_json, stage_feedback, high_multiplier,
candidate_tiers_json, pool_snapshot_json, rtp_snapshot_json, reward_status,
reward_transaction_id, reward_failure_reason, paid_at_ms, created_at_ms, updated_at_ms
) VALUES (
'lalu', ?, ?, 42, 'device-1', 'room-1', 'anchor-1', 'lucky', 'rose',
?, 1, 'novice', 1, 1,
'tier', ?, 0, 0,
?, '{}', 0, 0,
'{}', '{}', '{}', ?,
'', '', ?, ?, ?
)`, drawID, commandID, spent, reward, reward, status, createdAtMS, createdAtMS, createdAtMS)
if err != nil {
t.Fatalf("seed draw record %s: %v", drawID, err)
}
}
func assertLuckyStatsTotal(t *testing.T, repository *Repository, giftID string, wantDraws int64, wantSpent int64, wantReward int64) {
t.Helper()
var draws, spent, reward int64
if err := repository.db.QueryRowContext(context.Background(), `
SELECT total_draws, total_spent_coins, total_reward_coins
FROM lucky_draw_pool_stats
WHERE app_code = 'lalu' AND pool_id = 'lucky' AND gift_id = ?`, giftID,
).Scan(&draws, &spent, &reward); err != nil {
t.Fatalf("read stats total: %v", err)
}
if draws != wantDraws || spent != wantSpent || reward != wantReward {
t.Fatalf("stats mismatch: draws=%d spent=%d reward=%d want draws=%d spent=%d reward=%d", draws, spent, reward, wantDraws, wantSpent, wantReward)
}
}
func assertLuckyStatsCursor(t *testing.T, repository *Repository, wantCreatedAtMS int64, wantDrawID string) {
t.Helper()
var createdAtMS int64
var drawID string
if err := repository.db.QueryRowContext(context.Background(), `
SELECT last_draw_created_at_ms, last_draw_id
FROM lucky_draw_pool_stat_cursors
WHERE app_code = 'lalu' AND cursor_name = 'draw_records'`,
).Scan(&createdAtMS, &drawID); err != nil {
t.Fatalf("read stats cursor: %v", err)
}
if createdAtMS != wantCreatedAtMS || drawID != wantDrawID {
t.Fatalf("cursor mismatch: created_at=%d draw_id=%s want created_at=%d draw_id=%s", createdAtMS, drawID, wantCreatedAtMS, wantDrawID)
}
}
func TestLuckyRuntimeConfigFromRuleConfigUsesV2PoolAndStageTiers(t *testing.T) {
ruleConfig := domain.RuleConfig{
AppCode: "lalu",
@ -399,29 +545,31 @@ func TestLuckyInitialRewardStatusGrantsZeroRewardWithoutOutbox(t *testing.T) {
if status := luckyInitialRewardStatus(luckyCandidate{TierID: "none"}, false); status != domain.StatusGranted {
t.Fatalf("zero reward without presentation should be granted, got %s", status)
}
if luckyDrawNeedsOutbox(luckyCandidate{TierID: "none"}, false) {
if luckyDrawNeedsRewardSettlementOutbox(luckyCandidate{TierID: "none"}) {
t.Fatalf("zero reward without presentation must not emit outbox")
}
}
func TestLuckyInitialRewardStatusKeepsAsyncSideEffectsPending(t *testing.T) {
func TestLuckyInitialRewardStatusOnlyWaitsForWalletSettlement(t *testing.T) {
tests := []struct {
name string
candidate luckyCandidate
stageFeedback bool
wantStatus string
wantReward bool
}{
{name: "wallet credit", candidate: luckyCandidate{TierID: "hit", BaseReward: 10}},
{name: "presentation candidate", candidate: luckyCandidate{TierID: "stage", Presentation: true}},
{name: "stage feedback", candidate: luckyCandidate{TierID: "none"}, stageFeedback: true},
{name: "wallet credit", candidate: luckyCandidate{TierID: "hit", BaseReward: 10}, wantStatus: domain.StatusPending, wantReward: true},
{name: "presentation candidate", candidate: luckyCandidate{TierID: "stage", Presentation: true}, wantStatus: domain.StatusGranted},
{name: "stage feedback", candidate: luckyCandidate{TierID: "none"}, stageFeedback: true, wantStatus: domain.StatusGranted},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
if status := luckyInitialRewardStatus(test.candidate, test.stageFeedback); status != domain.StatusPending {
t.Fatalf("async side effect should stay pending, got %s", status)
if status := luckyInitialRewardStatus(test.candidate, test.stageFeedback); status != test.wantStatus {
t.Fatalf("reward status mismatch: got %s want %s", status, test.wantStatus)
}
if !luckyDrawNeedsOutbox(test.candidate, test.stageFeedback) {
t.Fatalf("async side effect should emit outbox")
if got := luckyDrawNeedsRewardSettlementOutbox(test.candidate); got != test.wantReward {
t.Fatalf("reward settlement outbox mismatch: got %v want %v", got, test.wantReward)
}
})
}

196
services/activity-service/internal/storage/mysql/lucky_gift_settlement_pressure_test.go Normal file
View File

@ -0,0 +1,196 @@
package mysql
import (
"context"
"encoding/json"
"fmt"
"os"
"strings"
"sync/atomic"
"testing"
"time"
"hyapp/internal/testutil/mysqlschema"
"hyapp/pkg/appcode"
domain "hyapp/services/activity-service/internal/domain/luckygift"
luckygiftservice "hyapp/services/activity-service/internal/service/luckygift"
"google.golang.org/grpc"
walletv1 "hyapp.local/api/proto/wallet/v1"
)
func TestLuckyGiftRewardSettlementOutboxPressureLocal(t *testing.T) {
if os.Getenv("RUN_LUCKY_GIFT_SETTLEMENT_PRESSURE") != "1" {
t.Skip("set RUN_LUCKY_GIFT_SETTLEMENT_PRESSURE=1 to run local settlement outbox pressure test")
}
cases := []struct {
name string
total int
batchSize int
concurrency int
}{
{name: "1k_batch100_c4", total: 1000, batchSize: 100, concurrency: 4},
{name: "5k_batch100_c4", total: 5000, batchSize: 100, concurrency: 4},
{name: "10k_batch200_c8", total: 10000, batchSize: 200, concurrency: 8},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
runLuckyGiftSettlementPressureCase(t, tc.total, tc.batchSize, tc.concurrency)
})
}
}
func runLuckyGiftSettlementPressureCase(t *testing.T, total int, batchSize int, concurrency int) {
t.Helper()
schema := mysqlschema.New(t, mysqlschema.Config{
EnvVar: "ACTIVITY_SERVICE_MYSQL_TEST_DSN",
InitDBPath: mysqlschema.InitDBPath(t, mysqlschema.CallerFile(t, 1), "..", "..", "..", "deploy", "mysql", "initdb", "001_activity_service.sql"),
DatabasePrefix: "hy_lucky_press",
})
repository, err := Open(context.Background(), schema.DSN)
if err != nil {
t.Fatalf("open repository: %v", err)
}
t.Cleanup(func() { _ = repository.Close() })
seedStart := time.Now()
seedLuckyGiftSettlementPressureRows(t, repository, total)
seedElapsed := time.Since(seedStart)
wallet := &pressureLuckyGiftWallet{}
service := luckygiftservice.New(repository, luckygiftservice.WithWallet(wallet))
ctx := appcode.WithContext(context.Background(), "lalu")
options := luckygiftservice.WorkerOptions{
WorkerID: "pressure-worker",
BatchSize: batchSize,
Concurrency: concurrency,
LockTTL: time.Minute,
MaxRetry: 3,
}
start := time.Now()
processedTotal := 0
loops := 0
for processedTotal < total {
processed, err := service.ProcessPendingDrawOutbox(ctx, options)
if err != nil {
t.Fatalf("process settlement outbox: %v", err)
}
if processed == 0 {
break
}
processedTotal += processed
loops++
}
elapsed := time.Since(start)
granted := countLuckyGiftPressureRows(t, repository, "lucky_draw_records", "reward_status = 'granted'")
delivered := countLuckyGiftPressureRows(t, repository, "activity_outbox", "status = 'delivered'")
if processedTotal != total || granted != int64(total) || delivered != int64(total) || wallet.calls.Load() != int64(total) {
t.Fatalf("pressure result mismatch: processed=%d granted=%d delivered=%d wallet_calls=%d total=%d",
processedTotal, granted, delivered, wallet.calls.Load(), total)
}
t.Logf("settlement_pressure total=%d batch=%d concurrency=%d loops=%d seed_ms=%d process_ms=%d throughput=%.2f events/s avg_event_ms=%.4f",
total, batchSize, concurrency, loops, seedElapsed.Milliseconds(), elapsed.Milliseconds(),
float64(total)/elapsed.Seconds(), float64(elapsed.Microseconds())/1000.0/float64(total))
}
func seedLuckyGiftSettlementPressureRows(t *testing.T, repository *Repository, total int) {
t.Helper()
const chunkSize = 500
for start := 0; start < total; start += chunkSize {
end := start + chunkSize
if end > total {
end = total
}
if err := seedLuckyGiftSettlementPressureChunk(repository, start, end); err != nil {
t.Fatalf("seed pressure rows %d-%d: %v", start, end, err)
}
}
}
func seedLuckyGiftSettlementPressureChunk(repository *Repository, start int, end int) error {
nowMS := time.Now().UTC().UnixMilli()
drawPlaceholders := make([]string, 0, end-start)
drawArgs := make([]any, 0, (end-start)*31)
outboxPlaceholders := make([]string, 0, end-start)
outboxArgs := make([]any, 0, (end-start)*12)
for index := start; index < end; index++ {
drawID := fmt.Sprintf("pressure_draw_%08d", index)
commandID := fmt.Sprintf("pressure_cmd_%08d", index)
userID := int64(100000 + index%1000)
roomID := fmt.Sprintf("room_%03d", index%100)
createdAtMS := nowMS + int64(index)
drawPlaceholders = append(drawPlaceholders, "("+luckySQLPlaceholders(31)+")")
drawArgs = append(drawArgs,
"lalu", drawID, commandID, userID, "pressure-device", roomID, "pressure-anchor", "lucky", "rose",
int64(100), int64(1), domain.PoolNovice, int64(1), int64(1),
"hit", int64(10), int64(0), int64(0),
int64(10), "{}", false, false,
"{}", "{}", "{}", domain.StatusPending, "", "",
createdAtMS, createdAtMS, createdAtMS,
)
payload, _ := json.Marshal(map[string]any{
"event_type": "lucky_gift_drawn",
"event_id": "lucky_gift_drawn:" + drawID,
"app_code": "lalu",
"draw_id": drawID,
"command_id": commandID,
"pool_id": "lucky",
"user_id": userID,
"room_id": roomID,
"gift_id": "rose",
"gift_count": 1,
"coin_spent": 100,
"effective_reward_coins": 10,
"created_at_ms": createdAtMS,
})
outboxPlaceholders = append(outboxPlaceholders, "("+luckySQLPlaceholders(12)+")")
outboxArgs = append(outboxArgs,
"lalu", "lucky_reward_"+drawID, domain.EventTypeLuckyGiftRewardSettlement, payload,
"pending", 0, createdAtMS, "", int64(0), "", createdAtMS, createdAtMS,
)
}
ctx := context.Background()
if _, err := repository.db.ExecContext(ctx, `
INSERT INTO lucky_draw_records (
app_code, draw_id, command_id, user_id, device_id, room_id, anchor_id, pool_id, gift_id,
coin_spent, rule_version, experience_pool, rtp_window_index, gift_rtp_window_index,
selected_tier_id, base_reward_coins, room_atmosphere_reward_coins, activity_subsidy_coins,
effective_reward_coins, budget_sources_json, stage_feedback, high_multiplier,
candidate_tiers_json, pool_snapshot_json, rtp_snapshot_json, reward_status,
reward_transaction_id, reward_failure_reason, paid_at_ms, created_at_ms, updated_at_ms
) VALUES `+strings.Join(drawPlaceholders, ","), drawArgs...); err != nil {
return err
}
if _, err := repository.db.ExecContext(ctx, `
INSERT INTO activity_outbox (
app_code, outbox_id, event_type, payload, status, retry_count, next_retry_at_ms,
locked_by, lock_until_ms, last_error, created_at_ms, updated_at_ms
) VALUES `+strings.Join(outboxPlaceholders, ","), outboxArgs...); err != nil {
return err
}
return nil
}
func countLuckyGiftPressureRows(t *testing.T, repository *Repository, table string, where string) int64 {
t.Helper()
var count int64
if err := repository.db.QueryRowContext(context.Background(), "SELECT COUNT(*) FROM "+table+" WHERE app_code = 'lalu' AND "+where).Scan(&count); err != nil {
t.Fatalf("count %s: %v", table, err)
}
return count
}
type pressureLuckyGiftWallet struct {
calls atomic.Int64
}
func (w *pressureLuckyGiftWallet) CreditLuckyGiftReward(_ context.Context, req *walletv1.CreditLuckyGiftRewardRequest, _ ...grpc.CallOption) (*walletv1.CreditLuckyGiftRewardResponse, error) {
w.calls.Add(1)
return &walletv1.CreditLuckyGiftRewardResponse{
TransactionId: "pressure_tx_" + req.GetDrawId(),
Balance: &walletv1.AssetBalance{AvailableAmount: 1_000_000},
Amount: req.GetAmount(),
GrantedAtMs: time.Now().UTC().UnixMilli(),
}, nil
}

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

@ -102,107 +102,22 @@ func (r *Repository) ensureLuckyDrawPoolStatsTables(ctx context.Context) error {
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)`,
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 lucky_draw_pool_stat_cursors (
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码',
cursor_name VARCHAR(64) NOT NULL COMMENT '游标名称',
last_draw_created_at_ms BIGINT NOT NULL DEFAULT 0 COMMENT '最后已聚合 draw 创建时间',
last_draw_id VARCHAR(128) NOT NULL DEFAULT '' COMMENT '同一毫秒内最后已聚合 draw ID',
created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms',
updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms',
PRIMARY KEY (app_code, cursor_name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物后台统计快照游标'`,
}
for _, statement := range statements {
if _, err := r.db.ExecContext(ctx, statement); err != nil {
@ -279,6 +194,25 @@ func (r *Repository) ensureLuckyDrawPoolID(ctx context.Context) error {
return err
}
}
indexes := []struct {
name string
sql string
}{
{"idx_lucky_draw_created", `CREATE INDEX idx_lucky_draw_created ON lucky_draw_records (app_code, created_at_ms, draw_id)`},
{"idx_lucky_draw_pool_status_summary", `CREATE INDEX idx_lucky_draw_pool_status_summary ON lucky_draw_records (app_code, pool_id, reward_status, created_at_ms, draw_id)`},
{"idx_lucky_draw_gift_status_summary", `CREATE INDEX idx_lucky_draw_gift_status_summary ON lucky_draw_records (app_code, pool_id, gift_id, reward_status, created_at_ms, draw_id)`},
}
for _, addition := range indexes {
exists, err := r.indexExists(ctx, table, addition.name)
if err != nil {
return err
}
if !exists {
if _, err := r.db.ExecContext(ctx, addition.sql); err != nil {
return err
}
}
}
return nil
}