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

fix: reduce outbox worker mysql scans

Browse Source
This commit is contained in:
zhx 2026-06-01 12:38:30 +08:00
parent fc082d2c00
commit 815275676e
15 changed files with 292 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -19,8 +19,8 @@ red_packet_broadcast_worker:
enabled: true
lucky_gift_worker:
enabled: true
worker_poll_interval: "1s"
worker_batch_size: 100
worker_poll_interval: "60s"
worker_batch_size: 10
worker_concurrency: 4
worker_lock_ttl: "30s"
worker_max_retry: 8

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

@ -44,7 +44,9 @@ CREATE TABLE IF NOT EXISTS activity_outbox (
created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms',
updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms',
PRIMARY KEY (app_code, outbox_id),
KEY idx_activity_outbox_status_retry (app_code, status, next_retry_at_ms)
KEY idx_activity_outbox_status_retry (app_code, status, next_retry_at_ms),
KEY idx_activity_outbox_lucky_retry (app_code, event_type, status, next_retry_at_ms, created_at_ms, outbox_id),
KEY idx_activity_outbox_lucky_lock (app_code, event_type, status, lock_until_ms, created_at_ms, outbox_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='活动事件 outbox 表';
CREATE TABLE IF NOT EXISTS lucky_gift_rules (

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

@ -1359,31 +1359,35 @@ func (r *Repository) ClaimPendingLuckyGiftOutbox(ctx context.Context, workerID s
}
defer func() { _ = tx.Rollback() }()
rows, err := tx.QueryContext(ctx, `
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
WHERE event_type = 'LuckyGiftDrawn'
AND (
(status IN ('pending', 'retryable') AND next_retry_at_ms <= ?)
OR (status = 'delivering' AND lock_until_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`, nowMS, nowMS, batchSize)
FOR UPDATE`, batchSize, appCode, nowMS, batchSize)
if err != nil {
return nil, err
}
defer rows.Close()
events := make([]domain.DrawOutbox, 0, batchSize)
for rows.Next() {
var event domain.DrawOutbox
if err := rows.Scan(&event.AppCode, &event.OutboxID, &event.EventType, &event.PayloadJSON, &event.RetryCount, &event.CreatedAtMS); err != nil {
if remaining := batchSize - len(events); remaining > 0 {
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 = 'LuckyGiftDrawn'
AND status = 'delivering'
AND lock_until_ms <= ?
ORDER BY created_at_ms ASC, outbox_id ASC
LIMIT ?
FOR UPDATE`, remaining, appCode, nowMS, remaining)
if err != nil {
return nil, err
}
events = append(events, event)
}
if err := rows.Err(); err != nil {
return nil, err
events = append(events, retryEvents...)
}
lockUntilMS := nowMS + lockTTL.Milliseconds()
for _, event := range events {
@ -1402,6 +1406,23 @@ func (r *Repository) ClaimPendingLuckyGiftOutbox(ctx context.Context, workerID s
return events, nil
}
func (r *Repository) queryLuckyGiftOutboxCandidates(ctx context.Context, tx *sql.Tx, query string, capacity int, args ...any) ([]domain.DrawOutbox, error) {
rows, err := tx.QueryContext(ctx, query, args...)
if err != nil {
return nil, err
}
defer rows.Close()
events := make([]domain.DrawOutbox, 0, capacity)
for rows.Next() {
var event domain.DrawOutbox
if err := rows.Scan(&event.AppCode, &event.OutboxID, &event.EventType, &event.PayloadJSON, &event.RetryCount, &event.CreatedAtMS); err != nil {
return nil, err
}
events = append(events, event)
}
return events, rows.Err()
}
func (r *Repository) MarkLuckyGiftOutboxDelivered(ctx context.Context, event domain.DrawOutbox, nowMS int64) error {
if r == nil || r.db == nil {
return xerr.New(xerr.Unavailable, "mysql repository is not configured")

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

@ -48,6 +48,9 @@ func (r *Repository) Migrate(ctx context.Context) error {
if err := r.ensureLuckyGiftOutboxColumns(ctx); err != nil {
return err
}
if err := r.ensureLuckyGiftOutboxIndexes(ctx); err != nil {
return err
}
if err := r.ensureLuckyDrawRewardSettlementColumns(ctx); err != nil {
return err
}
@ -299,6 +302,29 @@ func (r *Repository) ensureLuckyGiftOutboxColumns(ctx context.Context) error {
return nil
}
func (r *Repository) ensureLuckyGiftOutboxIndexes(ctx context.Context) error {
// LuckyGiftDrawn worker 的 pending 和 lock-expired 两条路径分开查，索引列顺序要和各自 WHERE 匹配。
additions := []struct {
name string
sql string
}{
{"idx_activity_outbox_lucky_retry", `CREATE INDEX idx_activity_outbox_lucky_retry ON activity_outbox (app_code, event_type, status, next_retry_at_ms, created_at_ms, outbox_id)`},
{"idx_activity_outbox_lucky_lock", `CREATE INDEX idx_activity_outbox_lucky_lock ON activity_outbox (app_code, event_type, status, lock_until_ms, created_at_ms, outbox_id)`},
}
for _, addition := range additions {
exists, err := r.indexExists(ctx, "activity_outbox", addition.name)
if err != nil {
return err
}
if !exists {
if _, err := r.db.ExecContext(ctx, addition.sql); err != nil {
return err
}
}
}
return nil
}
func (r *Repository) ensureLuckyDrawRewardSettlementColumns(ctx context.Context) error {
additions := []struct {
name string

4
services/game-service/configs/config.tencent.example.yaml
View File

@ -23,8 +23,8 @@ rocketmq:
producer_group: "hyapp-game-outbox-producer"
outbox_worker:
enabled: true
poll_interval: "1s"
batch_size: 100
poll_interval: "60s"
batch_size: 10
publish_timeout: "3s"
log:
level: info

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

@ -167,6 +167,7 @@ CREATE TABLE IF NOT EXISTS game_level_event_outbox (
PRIMARY KEY(app_code, event_id),
UNIQUE KEY uk_game_level_event_order(app_code, order_id),
KEY idx_game_level_event_pending(app_code, status, next_retry_at_ms, created_at_ms),
KEY idx_game_level_event_claim(app_code, status, locked_until_ms, next_retry_at_ms, created_at_ms, event_id),
KEY idx_game_level_event_user(app_code, user_id, created_at_ms)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='游戏等级事件 outbox 表';

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

@ -214,6 +214,7 @@ func (r *Repository) Migrate(ctx context.Context) error {
PRIMARY KEY(app_code, event_id),
UNIQUE KEY uk_game_level_event_order(app_code, order_id),
KEY idx_game_level_event_pending(app_code, status, next_retry_at_ms, created_at_ms),
KEY idx_game_level_event_claim(app_code, status, locked_until_ms, next_retry_at_ms, created_at_ms, event_id),
KEY idx_game_level_event_user(app_code, user_id, created_at_ms)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
`CREATE TABLE IF NOT EXISTS game_outbox (
@ -749,20 +750,28 @@ func (r *Repository) ClaimPendingLevelEvents(ctx context.Context, workerID strin
return nil, err
}
defer func() { _ = tx.Rollback() }()
rows, err := tx.QueryContext(ctx, levelEventOutboxSelectSQL()+`
appCode := appcode.FromContext(ctx)
events, err := r.queryPendingLevelEvents(ctx, tx, `
WHERE app_code = ? AND status IN ('pending', 'failed') AND next_retry_at_ms <= ?
AND (locked_until_ms = 0 OR locked_until_ms <= ?)
AND locked_until_ms = 0
ORDER BY created_at_ms ASC, event_id ASC
LIMIT ? FOR UPDATE SKIP LOCKED`,
appcode.FromContext(ctx), nowMS, nowMS, batchSize,
)
appCode, nowMS, batchSize)
if err != nil {
return nil, err
}
events, err := scanLevelEventOutboxRows(rows)
_ = rows.Close()
if err != nil {
return nil, err
if remaining := batchSize - len(events); remaining > 0 {
retryEvents, err := r.queryPendingLevelEvents(ctx, tx, `
WHERE app_code = ? AND status IN ('pending', 'failed') AND next_retry_at_ms <= ?
AND locked_until_ms > 0 AND locked_until_ms <= ?
ORDER BY created_at_ms ASC, event_id ASC
LIMIT ? FOR UPDATE SKIP LOCKED`,
appCode, nowMS, nowMS, remaining)
if err != nil {
return nil, err
}
events = append(events, retryEvents...)
}
lockUntil := nowMS + lockTTL.Milliseconds()
for _, event := range events {
@ -781,6 +790,15 @@ func (r *Repository) ClaimPendingLevelEvents(ctx context.Context, workerID strin
return events, nil
}
func (r *Repository) queryPendingLevelEvents(ctx context.Context, tx *sql.Tx, whereSQL string, args ...any) ([]gamedomain.LevelEventOutbox, error) {
rows, err := tx.QueryContext(ctx, levelEventOutboxSelectSQL()+whereSQL, args...)
if err != nil {
return nil, err
}
defer rows.Close()
return scanLevelEventOutboxRows(rows)
}
func (r *Repository) MarkLevelEventDelivered(ctx context.Context, eventID string, nowMS int64) error {
if r == nil || r.db == nil {
return xerr.New(xerr.Unavailable, "mysql repository is not configured")
@ -1052,7 +1070,7 @@ func levelEventOutboxSelectSQL() string {
provider_round_id, coin_amount, wallet_transaction_id,
COALESCE(CAST(payload_json AS CHAR), '{}'), status, attempt_count,
failure_reason, occurred_at_ms, created_at_ms, updated_at_ms
FROM game_level_event_outbox `
FROM game_level_event_outbox FORCE INDEX (idx_game_level_event_claim) `
}
func scanCatalog(scanner catalogScanner) (gamedomain.CatalogItem, error) {

5
services/wallet-service/configs/config.docker.yaml
View File

@ -17,6 +17,11 @@ red_packet_expiry_worker:
app_code: "lalu"
poll_interval: "5s"
batch_size: 50
projection_worker:
enabled: true
poll_interval: "60s"
batch_size: 10
lock_ttl: "30s"
rocketmq:
# Docker 本地使用 compose RocketMQ 发布 wallet_outbox 事实。
enabled: true

5
services/wallet-service/configs/config.tencent.example.yaml
View File

@ -17,6 +17,11 @@ red_packet_expiry_worker:
app_code: "lalu"
poll_interval: "5s"
batch_size: 50
projection_worker:
enabled: true
poll_interval: "60s"
batch_size: 10
lock_ttl: "30s"
rocketmq:
enabled: true
name_servers:

5
services/wallet-service/configs/config.yaml
View File

@ -17,6 +17,11 @@ red_packet_expiry_worker:
app_code: "lalu"
poll_interval: "5s"
batch_size: 50
projection_worker:
enabled: true
poll_interval: "60s"
batch_size: 10
lock_ttl: "30s"
rocketmq:
# 本地默认关闭；线上由 wallet-service outbox worker 投递账务事实到 MQ。
enabled: false

21
services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql
View File

@ -75,6 +75,8 @@ CREATE TABLE IF NOT EXISTS wallet_outbox (
PRIMARY KEY (app_code, event_id),
KEY idx_wallet_outbox_pending (app_code, status, next_retry_at_ms, created_at_ms),
KEY idx_wallet_outbox_claim (app_code, status, lock_until_ms, created_at_ms),
KEY idx_wallet_outbox_event_created (app_code, event_type, created_at_ms, event_id),
KEY idx_wallet_outbox_asset_event_created (app_code, asset_type, event_type, created_at_ms, event_id),
KEY idx_wallet_outbox_tx (app_code, transaction_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='钱包事件 outbox 表';
@ -338,6 +340,25 @@ PREPARE stmt FROM @ddl;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;
-- projection worker 只扫描单租户、单事件类型的 wallet_outbox；缺少该索引会退化成百万级全表扫描。
SET @ddl := IF(
(SELECT COUNT(*) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'wallet_outbox' AND INDEX_NAME = 'idx_wallet_outbox_event_created') = 0,
'ALTER TABLE wallet_outbox ADD INDEX idx_wallet_outbox_event_created (app_code, event_type, created_at_ms, event_id)',
'SELECT 1'
);
PREPARE stmt FROM @ddl;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;
SET @ddl := IF(
(SELECT COUNT(*) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'wallet_outbox' AND INDEX_NAME = 'idx_wallet_outbox_asset_event_created') = 0,
'ALTER TABLE wallet_outbox ADD INDEX idx_wallet_outbox_asset_event_created (app_code, asset_type, event_type, created_at_ms, event_id)',
'SELECT 1'
);
PREPARE stmt FROM @ddl;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;
CREATE TABLE IF NOT EXISTS wallet_gift_prices (
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码，用于多租户隔离',
gift_id VARCHAR(96) NOT NULL COMMENT '礼物 ID',

19
services/wallet-service/internal/app/app.go
View File

@ -40,6 +40,7 @@ type App struct {
activityConn *grpc.ClientConn
outboxProducer *rocketmqx.Producer
outboxWorkerCfg config.OutboxWorkerConfig
projectionWorkerCfg config.ProjectionWorkerConfig
walletOutboxTopic string
nodeID string
redPacketExpiryWorkerCfg config.RedPacketExpiryWorkerConfig
@ -124,6 +125,7 @@ func New(cfg config.Config) (*App, error) {
activityConn: activityConn,
outboxProducer: outboxProducer,
outboxWorkerCfg: cfg.OutboxWorker,
projectionWorkerCfg: cfg.ProjectionWorker,
walletOutboxTopic: cfg.RocketMQ.WalletOutbox.Topic,
nodeID: cfg.NodeID,
redPacketExpiryWorkerCfg: cfg.RedPacketExpiryWorker,
@ -169,25 +171,28 @@ func (a *App) Close() {
}
func (a *App) runGiftWallProjectionWorker() {
if a.walletSvc == nil {
if a.walletSvc == nil || !a.projectionWorkerCfg.Enabled {
return
}
ctx, cancel := context.WithCancel(context.Background())
a.stopWorker = cancel
giftWallWorkerID := "gift-wall-" + a.nodeID
pollInterval := a.projectionWorkerCfg.PollInterval
batchSize := a.projectionWorkerCfg.BatchSize
lockTTL := a.projectionWorkerCfg.LockTTL
a.workers.Add(1)
go func() {
defer a.workers.Done()
ticker := time.NewTicker(2 * time.Second)
ticker := time.NewTicker(pollInterval)
defer ticker.Stop()
for {
// 礼物墙是展示读模型，worker 常驻在 wallet-service 内消费账务 outbox。
// 服务重启会回滚未提交的投影事务；已完成事件由 projection done 状态防重。
processed, err := a.walletSvc.ProjectPendingGiftWallEvents(ctx, giftWallWorkerID, 50, 30*time.Second)
processed, err := a.walletSvc.ProjectPendingGiftWallEvents(ctx, giftWallWorkerID, batchSize, lockTTL)
if err != nil && !errors.Is(err, context.Canceled) {
logx.Error(ctx, "gift_wall_projection_failed", err, slog.String("worker_id", giftWallWorkerID))
}
if processed >= 50 {
if processed >= batchSize {
// 批次打满说明还有积压，立即继续 drain，避免固定 tick 放大展示延迟。
continue
}
@ -202,15 +207,15 @@ func (a *App) runGiftWallProjectionWorker() {
a.workers.Add(1)
go func() {
defer a.workers.Done()
ticker := time.NewTicker(2 * time.Second)
ticker := time.NewTicker(pollInterval)
defer ticker.Stop()
for {
// 徽章展示槽位属于 activity-service 读模型；wallet 只 relay 自己 outbox 中的资源赠送事实。
processed, err := a.walletSvc.ProjectPendingBadgeGrantEvents(ctx, badgeWorkerID, 50, 30*time.Second)
processed, err := a.walletSvc.ProjectPendingBadgeGrantEvents(ctx, badgeWorkerID, batchSize, lockTTL)
if err != nil && !errors.Is(err, context.Canceled) {
logx.Error(ctx, "badge_grant_projection_failed", err, slog.String("worker_id", badgeWorkerID))
}
if processed >= 50 {
if processed >= batchSize {
continue
}
select {

28
services/wallet-service/internal/config/config.go
View File

@ -29,6 +29,8 @@ type Config struct {
GooglePlay GooglePlayConfig `yaml:"google_play"`
// OutboxWorker 控制 wallet_outbox 到 MQ 的补偿投递。
OutboxWorker OutboxWorkerConfig `yaml:"outbox_worker"`
// ProjectionWorker 控制 wallet_outbox 到本服务读模型的投影补偿扫描。
ProjectionWorker ProjectionWorkerConfig `yaml:"projection_worker"`
// MySQLAutoMigrate 保留给显式本地迁移；线上 schema 由发布流程或 initdb 管理。
MySQLAutoMigrate bool `yaml:"mysql_auto_migrate"`
Log logx.Config `yaml:"log"`
@ -75,6 +77,14 @@ type OutboxWorkerConfig struct {
MaxBackoff time.Duration `yaml:"max_backoff"`
}
// ProjectionWorkerConfig 控制礼物墙和 badge 展示读模型从 wallet_outbox 补偿投影的节奏。
type ProjectionWorkerConfig struct {
Enabled bool `yaml:"enabled"`
PollInterval time.Duration `yaml:"poll_interval"`
BatchSize int `yaml:"batch_size"`
LockTTL time.Duration `yaml:"lock_ttl"`
}
// GooglePlayConfig 保存 Google Play Developer API 服务账号和网络配置。
type GooglePlayConfig struct {
Enabled bool `yaml:"enabled"`
@ -120,6 +130,12 @@ func Default() Config {
InitialBackoff: 5 * time.Second,
MaxBackoff: 5 * time.Minute,
},
ProjectionWorker: ProjectionWorkerConfig{
Enabled: true,
PollInterval: 60 * time.Second,
BatchSize: 10,
LockTTL: 30 * time.Second,
},
MySQLAutoMigrate: false,
Log: logx.Config{
Level: "info",
@ -234,6 +250,18 @@ func Load(path string) (Config, error) {
if cfg.OutboxWorker.Enabled && !cfg.RocketMQ.WalletOutbox.Enabled {
return Config{}, errors.New("outbox_worker requires rocketmq.wallet_outbox.enabled")
}
if cfg.ProjectionWorker.PollInterval <= 0 {
cfg.ProjectionWorker.PollInterval = Default().ProjectionWorker.PollInterval
}
if cfg.ProjectionWorker.BatchSize <= 0 {
cfg.ProjectionWorker.BatchSize = Default().ProjectionWorker.BatchSize
}
if cfg.ProjectionWorker.BatchSize > 100 {
cfg.ProjectionWorker.BatchSize = 100
}
if cfg.ProjectionWorker.LockTTL <= 0 {
cfg.ProjectionWorker.LockTTL = Default().ProjectionWorker.LockTTL
}
return cfg, nil
}

74
services/wallet-service/internal/storage/mysql/badge_projection_relay.go
View File

@ -5,6 +5,7 @@ import (
"database/sql"
"encoding/json"
"errors"
"sort"
"strings"
"time"
@ -21,8 +22,9 @@ const (
)
type badgeGrantProjectionCandidate struct {
AppCode string
EventID string
AppCode string
EventID string
CreatedAtMS int64
}
type badgeGrantOutboxPayload struct {
@ -74,37 +76,81 @@ func (r *Repository) ClaimPendingBadgeGrantEvents(ctx context.Context, workerID
}
func (r *Repository) listBadgeGrantProjectionCandidates(ctx context.Context, nowMS int64, limit int) ([]badgeGrantProjectionCandidate, error) {
appCode := appcode.FromContext(ctx)
rows, err := r.db.QueryContext(ctx, `
SELECT wo.app_code, wo.event_id
FROM wallet_outbox wo
SELECT wo.app_code, wo.event_id, wo.created_at_ms
FROM wallet_outbox wo FORCE INDEX (idx_wallet_outbox_asset_event_created)
LEFT JOIN wallet_projection_events pe
ON pe.projection_name = ? AND pe.app_code = wo.app_code AND pe.event_id = wo.event_id
WHERE wo.asset_type = ?
WHERE wo.app_code = ?
AND wo.asset_type = ?
AND wo.event_type IN (?, ?)
AND (
pe.event_id IS NULL
OR (pe.status IN (?, ?) AND pe.locked_until_ms <= ?)
)
AND pe.event_id IS NULL
ORDER BY wo.created_at_ms ASC, wo.event_id ASC
LIMIT ?`,
badgeDisplayProjectionName,
appCode,
resourceOutboxAsset,
walletResourceGrantedEvent,
walletResourceGroupGrantedEvent,
projectionStatusFailed,
projectionStatusProcessing,
nowMS,
limit,
)
if err != nil {
return nil, err
}
defer rows.Close()
candidates, err := scanBadgeGrantProjectionCandidates(rows, limit)
if err != nil {
return nil, err
}
rows, err = r.db.QueryContext(ctx, `
SELECT wo.app_code, wo.event_id, wo.created_at_ms
FROM wallet_projection_events pe
JOIN wallet_outbox wo
ON wo.app_code = pe.app_code AND wo.event_id = pe.event_id
WHERE pe.projection_name = ?
AND pe.app_code = ?
AND pe.status IN (?, ?)
AND pe.locked_until_ms <= ?
AND wo.asset_type = ?
AND wo.event_type IN (?, ?)
ORDER BY wo.created_at_ms ASC, wo.event_id ASC
LIMIT ?`,
badgeDisplayProjectionName,
appCode,
projectionStatusFailed,
projectionStatusProcessing,
nowMS,
resourceOutboxAsset,
walletResourceGrantedEvent,
walletResourceGroupGrantedEvent,
limit,
)
if err != nil {
return nil, err
}
retryCandidates, err := scanBadgeGrantProjectionCandidates(rows, limit)
if err != nil {
return nil, err
}
candidates = append(candidates, retryCandidates...)
sort.SliceStable(candidates, func(i, j int) bool {
if candidates[i].CreatedAtMS == candidates[j].CreatedAtMS {
return candidates[i].EventID < candidates[j].EventID
}
return candidates[i].CreatedAtMS < candidates[j].CreatedAtMS
})
if len(candidates) > limit {
candidates = candidates[:limit]
}
return candidates, nil
}
func scanBadgeGrantProjectionCandidates(rows *sql.Rows, limit int) ([]badgeGrantProjectionCandidate, error) {
defer rows.Close()
candidates := make([]badgeGrantProjectionCandidate, 0, limit)
for rows.Next() {
var candidate badgeGrantProjectionCandidate
if err := rows.Scan(&candidate.AppCode, &candidate.EventID); err != nil {
if err := rows.Scan(&candidate.AppCode, &candidate.EventID, &candidate.CreatedAtMS); err != nil {
return nil, err
}
candidates = append(candidates, candidate)

71
services/wallet-service/internal/storage/mysql/gift_wall_projection.go
View File

@ -5,6 +5,7 @@ import (
"database/sql"
"encoding/json"
"errors"
"sort"
"strings"
"time"
@ -26,8 +27,9 @@ const (
)
type giftWallProjectionCandidate struct {
AppCode string
EventID string
AppCode string
EventID string
CreatedAtMS int64
}
type giftWallProjectionEvent struct {
@ -79,34 +81,75 @@ func (r *Repository) ProjectPendingGiftWallEvents(ctx context.Context, workerID
func (r *Repository) listGiftWallProjectionCandidates(ctx context.Context, limit int) ([]giftWallProjectionCandidate, error) {
nowMs := time.Now().UnixMilli()
appCode := appcode.FromContext(ctx)
rows, err := r.db.QueryContext(ctx, `
SELECT wo.app_code, wo.event_id
FROM wallet_outbox wo
SELECT wo.app_code, wo.event_id, wo.created_at_ms
FROM wallet_outbox wo FORCE INDEX (idx_wallet_outbox_event_created)
LEFT JOIN wallet_projection_events pe
ON pe.projection_name = ? AND pe.app_code = wo.app_code AND pe.event_id = wo.event_id
WHERE wo.event_type = ?
AND (
pe.event_id IS NULL
OR (pe.status IN (?, ?) AND pe.locked_until_ms <= ?)
)
WHERE wo.app_code = ?
AND wo.event_type = ?
AND pe.event_id IS NULL
ORDER BY wo.created_at_ms ASC, wo.event_id ASC
LIMIT ?`,
giftWallProjectionName,
appCode,
walletGiftDebitedEvent,
projectionStatusFailed,
projectionStatusProcessing,
nowMs,
limit,
)
if err != nil {
return nil, err
}
defer rows.Close()
candidates, err := scanGiftWallProjectionCandidates(rows, limit)
if err != nil {
return nil, err
}
rows, err = r.db.QueryContext(ctx, `
SELECT wo.app_code, wo.event_id, wo.created_at_ms
FROM wallet_projection_events pe
JOIN wallet_outbox wo
ON wo.app_code = pe.app_code AND wo.event_id = pe.event_id
WHERE pe.projection_name = ?
AND pe.app_code = ?
AND pe.status IN (?, ?)
AND pe.locked_until_ms <= ?
AND wo.event_type = ?
ORDER BY wo.created_at_ms ASC, wo.event_id ASC
LIMIT ?`,
giftWallProjectionName,
appCode,
projectionStatusFailed,
projectionStatusProcessing,
nowMs,
walletGiftDebitedEvent,
limit,
)
if err != nil {
return nil, err
}
retryCandidates, err := scanGiftWallProjectionCandidates(rows, limit)
if err != nil {
return nil, err
}
candidates = append(candidates, retryCandidates...)
sort.SliceStable(candidates, func(i, j int) bool {
if candidates[i].CreatedAtMS == candidates[j].CreatedAtMS {
return candidates[i].EventID < candidates[j].EventID
}
return candidates[i].CreatedAtMS < candidates[j].CreatedAtMS
})
if len(candidates) > limit {
candidates = candidates[:limit]
}
return candidates, nil
}
func scanGiftWallProjectionCandidates(rows *sql.Rows, limit int) ([]giftWallProjectionCandidate, error) {
defer rows.Close()
candidates := make([]giftWallProjectionCandidate, 0, limit)
for rows.Next() {
var candidate giftWallProjectionCandidate
if err := rows.Scan(&candidate.AppCode, &candidate.EventID); err != nil {
if err := rows.Scan(&candidate.AppCode, &candidate.EventID, &candidate.CreatedAtMS); err != nil {
return nil, err
}
candidates = append(candidates, candidate)