diff --git a/services/activity-service/configs/config.docker.yaml b/services/activity-service/configs/config.docker.yaml index 8db07f36..8062e942 100644 --- a/services/activity-service/configs/config.docker.yaml +++ b/services/activity-service/configs/config.docker.yaml @@ -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 diff --git a/services/activity-service/configs/config.tencent.example.yaml b/services/activity-service/configs/config.tencent.example.yaml index 1687a4b2..010cfe7c 100644 --- a/services/activity-service/configs/config.tencent.example.yaml +++ b/services/activity-service/configs/config.tencent.example.yaml @@ -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 diff --git a/services/activity-service/configs/config.yaml b/services/activity-service/configs/config.yaml index fdf50deb..86d3c302 100644 --- a/services/activity-service/configs/config.yaml +++ b/services/activity-service/configs/config.yaml @@ -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 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 8ed8b138..d9da299a 100644 --- a/services/activity-service/deploy/mysql/initdb/001_activity_service.sql +++ b/services/activity-service/deploy/mysql/initdb/001_activity_service.sql @@ -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', diff --git a/services/activity-service/internal/app/app.go b/services/activity-service/internal/app/app.go index aa3b86b1..1737839b 100644 --- a/services/activity-service/internal/app/app.go +++ b/services/activity-service/internal/app/app.go @@ -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, } } diff --git a/services/activity-service/internal/config/config.go b/services/activity-service/internal/config/config.go index ab3cef1a..c139957f 100644 --- a/services/activity-service/internal/config/config.go +++ b/services/activity-service/internal/config/config.go @@ -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, diff --git a/services/activity-service/internal/domain/luckygift/lucky_gift.go b/services/activity-service/internal/domain/luckygift/lucky_gift.go index 62cc60bd..e50fe196 100644 --- a/services/activity-service/internal/domain/luckygift/lucky_gift.go +++ b/services/activity-service/internal/domain/luckygift/lucky_gift.go @@ -28,6 +28,9 @@ const ( BroadcastRoom = "room" BroadcastRegion = "region" BroadcastGlobal = "global" + + EventTypeLuckyGiftDrawn = "LuckyGiftDrawn" + EventTypeLuckyGiftRewardSettlement = "LuckyGiftRewardSettlement" ) // Tier 是单个体验池里的候选奖档；权重只塑造体验节奏，基础成本仍由 RTP 窗口兜住。 diff --git a/services/activity-service/internal/service/luckygift/service.go b/services/activity-service/internal/service/luckygift/service.go index 8c0eba38..9471b555 100644 --- a/services/activity-service/internal/service/luckygift/service.go +++ b/services/activity-service/internal/service/luckygift/service.go @@ -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< 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 { diff --git a/services/activity-service/internal/storage/mysql/lucky_gift_repository_test.go b/services/activity-service/internal/storage/mysql/lucky_gift_repository_test.go index 28e0e160..a2cb3aa0 100644 --- a/services/activity-service/internal/storage/mysql/lucky_gift_repository_test.go +++ b/services/activity-service/internal/storage/mysql/lucky_gift_repository_test.go @@ -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) } }) } diff --git a/services/activity-service/internal/storage/mysql/lucky_gift_settlement_pressure_test.go b/services/activity-service/internal/storage/mysql/lucky_gift_settlement_pressure_test.go new file mode 100644 index 00000000..7f9e2db4 --- /dev/null +++ b/services/activity-service/internal/storage/mysql/lucky_gift_settlement_pressure_test.go @@ -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 +} diff --git a/services/activity-service/internal/storage/mysql/repository.go b/services/activity-service/internal/storage/mysql/repository.go index 781f7542..2d9c43a2 100644 --- a/services/activity-service/internal/storage/mysql/repository.go +++ b/services/activity-service/internal/storage/mysql/repository.go @@ -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 }