From 0aef70ecd7917dc41adb221cac38e753dd72b221 Mon Sep 17 00:00:00 2001 From: zhx Date: Mon, 1 Jun 2026 15:01:23 +0800 Subject: [PATCH] fix: consume wallet projections from mq --- .../wallet-service/configs/config.docker.yaml | 2 + .../configs/config.tencent.example.yaml | 2 + services/wallet-service/configs/config.yaml | 4 +- services/wallet-service/internal/app/app.go | 141 ++++++++++-------- .../wallet-service/internal/config/config.go | 33 ++-- .../internal/service/wallet/service.go | 113 ++++++++++---- .../internal/service/wallet/service_test.go | 73 +++++++-- .../storage/mysql/badge_projection_relay.go | 61 +++++++- .../storage/mysql/gift_wall_projection.go | 43 ++++++ 9 files changed, 359 insertions(+), 113 deletions(-) diff --git a/services/wallet-service/configs/config.docker.yaml b/services/wallet-service/configs/config.docker.yaml index eaf79a1f..fa176a70 100644 --- a/services/wallet-service/configs/config.docker.yaml +++ b/services/wallet-service/configs/config.docker.yaml @@ -22,6 +22,8 @@ projection_worker: poll_interval: "60s" batch_size: 10 lock_ttl: "30s" + consumer_group: "hyapp-wallet-projection" + consumer_max_reconsume_times: 16 rocketmq: # Docker 本地使用 compose RocketMQ 发布 wallet_outbox 事实。 enabled: true diff --git a/services/wallet-service/configs/config.tencent.example.yaml b/services/wallet-service/configs/config.tencent.example.yaml index 1c9e5935..39c8a358 100644 --- a/services/wallet-service/configs/config.tencent.example.yaml +++ b/services/wallet-service/configs/config.tencent.example.yaml @@ -22,6 +22,8 @@ projection_worker: poll_interval: "60s" batch_size: 10 lock_ttl: "30s" + consumer_group: "hyapp-wallet-projection" + consumer_max_reconsume_times: 16 rocketmq: enabled: true name_servers: diff --git a/services/wallet-service/configs/config.yaml b/services/wallet-service/configs/config.yaml index 0ebeff83..dfc32b74 100644 --- a/services/wallet-service/configs/config.yaml +++ b/services/wallet-service/configs/config.yaml @@ -18,10 +18,12 @@ red_packet_expiry_worker: poll_interval: "5s" batch_size: 50 projection_worker: - enabled: true + enabled: false poll_interval: "60s" batch_size: 10 lock_ttl: "30s" + consumer_group: "hyapp-wallet-projection" + consumer_max_reconsume_times: 16 rocketmq: # 本地默认关闭；线上由 wallet-service outbox worker 投递账务事实到 MQ。 enabled: false diff --git a/services/wallet-service/internal/app/app.go b/services/wallet-service/internal/app/app.go index 8290ced1..20d7b358 100644 --- a/services/wallet-service/internal/app/app.go +++ b/services/wallet-service/internal/app/app.go @@ -39,6 +39,7 @@ type App struct { walletSvc *walletservice.Service activityConn *grpc.ClientConn outboxProducer *rocketmqx.Producer + projectionConsumer *rocketmqx.Consumer outboxWorkerCfg config.OutboxWorkerConfig projectionWorkerCfg config.ProjectionWorkerConfig walletOutboxTopic string @@ -114,6 +115,16 @@ func New(cfg config.Config) (*App, error) { _ = repository.Close() return nil, err } + projectionConsumer, err := newWalletProjectionConsumer(cfg, svc) + if err != nil { + if outboxProducer != nil { + _ = outboxProducer.Shutdown() + } + _ = activityConn.Close() + _ = listener.Close() + _ = repository.Close() + return nil, err + } return &App{ server: server, @@ -124,6 +135,7 @@ func New(cfg config.Config) (*App, error) { walletSvc: svc, activityConn: activityConn, outboxProducer: outboxProducer, + projectionConsumer: projectionConsumer, outboxWorkerCfg: cfg.OutboxWorker, projectionWorkerCfg: cfg.ProjectionWorker, walletOutboxTopic: cfg.RocketMQ.WalletOutbox.Topic, @@ -139,11 +151,11 @@ func (a *App) Run() error { a.health.MarkStopped() return err } - a.runGiftWallProjectionWorker() + a.runBackgroundWorkers() a.health.MarkServing() err := a.server.Serve(a.listener) a.health.MarkStopped() - a.closeGiftWallProjectionWorker() + a.closeBackgroundWorkers() a.shutdownMQ() if errors.Is(err, grpc.ErrServerStopped) { return nil @@ -156,7 +168,7 @@ func (a *App) Run() error { func (a *App) Close() { a.closeOnce.Do(func() { a.health.MarkDraining() - a.closeGiftWallProjectionWorker() + a.closeBackgroundWorkers() grpcshutdown.GracefulStop(a.server, 15*time.Second) a.closeHealthHTTP() if a.activityConn != nil { @@ -170,61 +182,12 @@ func (a *App) Close() { }) } -func (a *App) runGiftWallProjectionWorker() { - if a.walletSvc == nil || !a.projectionWorkerCfg.Enabled { +func (a *App) runBackgroundWorkers() { + if a.walletSvc == nil { 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(pollInterval) - defer ticker.Stop() - for { - // 礼物墙是展示读模型，worker 常驻在 wallet-service 内消费账务 outbox。 - // 服务重启会回滚未提交的投影事务；已完成事件由 projection done 状态防重。 - 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 >= batchSize { - // 批次打满说明还有积压，立即继续 drain，避免固定 tick 放大展示延迟。 - continue - } - select { - case <-ctx.Done(): - return - case <-ticker.C: - } - } - }() - badgeWorkerID := "badge-grant-" + a.nodeID - a.workers.Add(1) - go func() { - defer a.workers.Done() - ticker := time.NewTicker(pollInterval) - defer ticker.Stop() - for { - // 徽章展示槽位属于 activity-service 读模型；wallet 只 relay 自己 outbox 中的资源赠送事实。 - 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 >= batchSize { - continue - } - select { - case <-ctx.Done(): - return - case <-ticker.C: - } - } - }() if a.outboxWorkerCfg.Enabled && a.outboxProducer != nil { outboxWorkerID := "wallet-outbox-" + a.nodeID a.workers.Add(1) @@ -373,7 +336,7 @@ func walletOutboxBackoff(retryCount int, options config.OutboxWorkerConfig) time return backoff } -func (a *App) closeGiftWallProjectionWorker() { +func (a *App) closeBackgroundWorkers() { if a.stopWorker != nil { a.stopWorker() } @@ -381,18 +344,72 @@ func (a *App) closeGiftWallProjectionWorker() { } func (a *App) startMQ() error { - if a.outboxProducer == nil { - return nil + if a.outboxProducer != nil { + if err := a.outboxProducer.Start(); err != nil { + return err + } } - return a.outboxProducer.Start() + if a.projectionConsumer != nil { + if err := a.projectionConsumer.Start(); err != nil { + if a.outboxProducer != nil { + _ = a.outboxProducer.Shutdown() + } + return err + } + } + return nil } func (a *App) shutdownMQ() { - if a.outboxProducer == nil { - return + if a.projectionConsumer != nil { + if err := a.projectionConsumer.Shutdown(); err != nil { + logx.Warn(context.Background(), "rocketmq_projection_consumer_shutdown_failed", slog.String("error", err.Error())) + } } - if err := a.outboxProducer.Shutdown(); err != nil { - logx.Warn(context.Background(), "rocketmq_producer_shutdown_failed", slog.String("error", err.Error())) + if a.outboxProducer != nil { + if err := a.outboxProducer.Shutdown(); err != nil { + logx.Warn(context.Background(), "rocketmq_producer_shutdown_failed", slog.String("error", err.Error())) + } + } +} + +func newWalletProjectionConsumer(cfg config.Config, svc *walletservice.Service) (*rocketmqx.Consumer, error) { + if svc == nil || !cfg.ProjectionWorker.Enabled { + return nil, nil + } + consumer, err := rocketmqx.NewConsumer(rocketMQProjectionConsumerConfig(cfg.RocketMQ, cfg.ProjectionWorker)) + if err != nil { + return nil, err + } + workerID := "wallet-projection-" + cfg.NodeID + if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { + walletMessage, err := walletmq.DecodeWalletOutboxMessage(message.Body) + if err != nil { + return err + } + _, err = svc.ProcessWalletProjectionMessage(appcode.WithContext(ctx, walletMessage.AppCode), workerID, walletMessage, cfg.ProjectionWorker.LockTTL) + return err + }); err != nil { + _ = consumer.Shutdown() + return nil, err + } + return consumer, nil +} + +func rocketMQProjectionConsumerConfig(cfg config.RocketMQConfig, projection config.ProjectionWorkerConfig) rocketmqx.ConsumerConfig { + return rocketmqx.ConsumerConfig{ + EndpointConfig: rocketmqx.EndpointConfig{ + NameServers: cfg.NameServers, + NameServerDomain: cfg.NameServerDomain, + AccessKey: cfg.AccessKey, + SecretKey: cfg.SecretKey, + SecurityToken: cfg.SecurityToken, + Namespace: cfg.Namespace, + VIPChannel: cfg.VIPChannel, + }, + GroupName: projection.ConsumerGroup, + MaxReconsumeTimes: projection.ConsumerMaxReconsumeTimes, + ConsumePullBatch: int32(projection.BatchSize), } } diff --git a/services/wallet-service/internal/config/config.go b/services/wallet-service/internal/config/config.go index 68d7fc68..b89f2ab3 100644 --- a/services/wallet-service/internal/config/config.go +++ b/services/wallet-service/internal/config/config.go @@ -29,7 +29,7 @@ type Config struct { GooglePlay GooglePlayConfig `yaml:"google_play"` // OutboxWorker 控制 wallet_outbox 到 MQ 的补偿投递。 OutboxWorker OutboxWorkerConfig `yaml:"outbox_worker"` - // ProjectionWorker 控制 wallet_outbox 到本服务读模型的投影补偿扫描。 + // ProjectionWorker 控制 wallet_outbox MQ 到本服务读模型的投影消费。 ProjectionWorker ProjectionWorkerConfig `yaml:"projection_worker"` // MySQLAutoMigrate 保留给显式本地迁移；线上 schema 由发布流程或 initdb 管理。 MySQLAutoMigrate bool `yaml:"mysql_auto_migrate"` @@ -77,12 +77,14 @@ type OutboxWorkerConfig struct { MaxBackoff time.Duration `yaml:"max_backoff"` } -// ProjectionWorkerConfig 控制礼物墙和 badge 展示读模型从 wallet_outbox 补偿投影的节奏。 +// ProjectionWorkerConfig 控制礼物墙和 badge 展示读模型从 wallet_outbox MQ 投影的节奏。 type ProjectionWorkerConfig struct { - Enabled bool `yaml:"enabled"` - PollInterval time.Duration `yaml:"poll_interval"` - BatchSize int `yaml:"batch_size"` - LockTTL time.Duration `yaml:"lock_ttl"` + Enabled bool `yaml:"enabled"` + PollInterval time.Duration `yaml:"poll_interval"` // Deprecated: runtime projection is MQ-driven. + BatchSize int `yaml:"batch_size"` + LockTTL time.Duration `yaml:"lock_ttl"` + ConsumerGroup string `yaml:"consumer_group"` + ConsumerMaxReconsumeTimes int32 `yaml:"consumer_max_reconsume_times"` } // GooglePlayConfig 保存 Google Play Developer API 服务账号和网络配置。 @@ -131,10 +133,12 @@ func Default() Config { MaxBackoff: 5 * time.Minute, }, ProjectionWorker: ProjectionWorkerConfig{ - Enabled: true, - PollInterval: 60 * time.Second, - BatchSize: 10, - LockTTL: 30 * time.Second, + Enabled: false, + PollInterval: 60 * time.Second, + BatchSize: 10, + LockTTL: 30 * time.Second, + ConsumerGroup: "hyapp-wallet-projection", + ConsumerMaxReconsumeTimes: 16, }, MySQLAutoMigrate: false, Log: logx.Config{ @@ -262,6 +266,15 @@ func Load(path string) (Config, error) { if cfg.ProjectionWorker.LockTTL <= 0 { cfg.ProjectionWorker.LockTTL = Default().ProjectionWorker.LockTTL } + if cfg.ProjectionWorker.ConsumerGroup = strings.TrimSpace(cfg.ProjectionWorker.ConsumerGroup); cfg.ProjectionWorker.ConsumerGroup == "" { + cfg.ProjectionWorker.ConsumerGroup = Default().ProjectionWorker.ConsumerGroup + } + if cfg.ProjectionWorker.ConsumerMaxReconsumeTimes <= 0 { + cfg.ProjectionWorker.ConsumerMaxReconsumeTimes = Default().ProjectionWorker.ConsumerMaxReconsumeTimes + } + if cfg.ProjectionWorker.Enabled && !cfg.RocketMQ.WalletOutbox.Enabled { + return Config{}, errors.New("projection_worker requires rocketmq.wallet_outbox.enabled") + } return cfg, nil } diff --git a/services/wallet-service/internal/service/wallet/service.go b/services/wallet-service/internal/service/wallet/service.go index 6bd2274c..5309c5d5 100644 --- a/services/wallet-service/internal/service/wallet/service.go +++ b/services/wallet-service/internal/service/wallet/service.go @@ -8,6 +8,7 @@ import ( "google.golang.org/grpc" activityv1 "hyapp.local/api/proto/activity/v1" "hyapp/pkg/appcode" + "hyapp/pkg/walletmq" "hyapp/pkg/xerr" "hyapp/services/wallet-service/internal/domain/ledger" resourcedomain "hyapp/services/wallet-service/internal/domain/resource" @@ -81,7 +82,9 @@ type Repository interface { SetResourceShopItemStatus(ctx context.Context, command resourcedomain.ResourceShopItemStatusCommand) (resourcedomain.ResourceShopItem, error) PurchaseResourceShopItem(ctx context.Context, command resourcedomain.ResourceShopPurchaseCommand) (resourcedomain.ResourceShopPurchaseReceipt, error) ProjectPendingGiftWallEvents(ctx context.Context, workerID string, limit int, lockTTL time.Duration) (int, error) + ProjectGiftWallMessage(ctx context.Context, workerID string, message walletmq.WalletOutboxMessage, lockTTL time.Duration) (bool, error) ClaimPendingBadgeGrantEvents(ctx context.Context, workerID string, nowMS int64, lockTTL time.Duration, batchSize int) ([]resourcedomain.BadgeGrantOutbox, error) + ClaimBadgeGrantMessage(ctx context.Context, workerID string, message walletmq.WalletOutboxMessage, nowMS int64, lockTTL time.Duration) (resourcedomain.BadgeGrantOutbox, bool, error) MarkBadgeGrantEventDelivered(ctx context.Context, event resourcedomain.BadgeGrantOutbox, nowMS int64) error MarkBadgeGrantEventFailed(ctx context.Context, event resourcedomain.BadgeGrantOutbox, failureReason string, nowMS int64) error } @@ -426,6 +429,30 @@ func (s *Service) ProjectPendingGiftWallEvents(ctx context.Context, workerID str return s.repository.ProjectPendingGiftWallEvents(ctx, workerID, limit, lockTTL) } +// ProcessWalletProjectionMessage consumes one wallet_outbox MQ fact for wallet-owned projections. +func (s *Service) ProcessWalletProjectionMessage(ctx context.Context, workerID string, message walletmq.WalletOutboxMessage, lockTTL time.Duration) (bool, error) { + workerID = strings.TrimSpace(workerID) + if workerID == "" { + return false, xerr.New(xerr.InvalidArgument, "worker_id is required") + } + if s.repository == nil { + return false, xerr.New(xerr.Unavailable, "wallet repository is not configured") + } + ctx = appcode.WithContext(ctx, message.AppCode) + handled := false + projected, err := s.repository.ProjectGiftWallMessage(ctx, workerID+"-gift-wall", message, lockTTL) + if err != nil { + return handled, err + } + handled = handled || projected + projected, err = s.projectBadgeGrantMessage(ctx, workerID+"-badge-grant", message, lockTTL) + if err != nil { + return handled, err + } + handled = handled || projected + return handled, nil +} + // ProjectPendingBadgeGrantEvents relays wallet resource grant outbox facts to activity badge display projection. func (s *Service) ProjectPendingBadgeGrantEvents(ctx context.Context, workerID string, limit int, lockTTL time.Duration) (int, error) { workerID = strings.TrimSpace(workerID) @@ -452,41 +479,69 @@ func (s *Service) ProjectPendingBadgeGrantEvents(ctx context.Context, workerID s processed := 0 for _, event := range events { processed++ - reqCtx := appcode.WithContext(ctx, event.AppCode) - resp, consumeErr := s.activity.ConsumeAchievementEvent(reqCtx, &activityv1.ConsumeAchievementEventRequest{ - Meta: &activityv1.RequestMeta{ - Caller: "wallet-service", - AppCode: event.AppCode, - SentAtMs: s.now().UnixMilli(), - }, - EventId: event.EventID, - EventType: event.EventType, - SourceService: "wallet-service", - UserId: event.UserID, - MetricType: walletBadgeGrantMetric, - Value: 1, - OccurredAtMs: event.CreatedAtMS, - DimensionsJson: event.PayloadJSON, - }) - if consumeErr != nil { - if markErr := s.repository.MarkBadgeGrantEventFailed(ctx, event, xerr.MessageOf(consumeErr), s.now().UnixMilli()); markErr != nil { - return processed, markErr - } - continue - } - if !isBadgeGrantProjectionStatusTerminal(resp.GetStatus()) { - if markErr := s.repository.MarkBadgeGrantEventFailed(ctx, event, "unexpected activity status: "+resp.GetStatus(), s.now().UnixMilli()); markErr != nil { - return processed, markErr - } - continue - } - if err := s.repository.MarkBadgeGrantEventDelivered(ctx, event, s.now().UnixMilli()); err != nil { + if err := s.deliverBadgeGrantProjection(ctx, event); err != nil { return processed, err } } return processed, nil } +func (s *Service) projectBadgeGrantMessage(ctx context.Context, workerID string, message walletmq.WalletOutboxMessage, lockTTL time.Duration) (bool, error) { + if !isBadgeGrantWalletProjectionMessage(message) { + return false, nil + } + if s.activity == nil { + return false, xerr.New(xerr.Unavailable, "activity badge client is not configured") + } + nowMS := s.now().UnixMilli() + event, claimed, err := s.repository.ClaimBadgeGrantMessage(ctx, workerID, message, nowMS, lockTTL) + if err != nil || !claimed { + return claimed, err + } + if err := s.deliverBadgeGrantProjection(ctx, event); err != nil { + return true, err + } + return true, nil +} + +func (s *Service) deliverBadgeGrantProjection(ctx context.Context, event resourcedomain.BadgeGrantOutbox) error { + reqCtx := appcode.WithContext(ctx, event.AppCode) + resp, consumeErr := s.activity.ConsumeAchievementEvent(reqCtx, &activityv1.ConsumeAchievementEventRequest{ + Meta: &activityv1.RequestMeta{ + Caller: "wallet-service", + AppCode: event.AppCode, + SentAtMs: s.now().UnixMilli(), + }, + EventId: event.EventID, + EventType: event.EventType, + SourceService: "wallet-service", + UserId: event.UserID, + MetricType: walletBadgeGrantMetric, + Value: 1, + OccurredAtMs: event.CreatedAtMS, + DimensionsJson: event.PayloadJSON, + }) + if consumeErr != nil { + return s.repository.MarkBadgeGrantEventFailed(ctx, event, xerr.MessageOf(consumeErr), s.now().UnixMilli()) + } + if !isBadgeGrantProjectionStatusTerminal(resp.GetStatus()) { + return s.repository.MarkBadgeGrantEventFailed(ctx, event, "unexpected activity status: "+resp.GetStatus(), s.now().UnixMilli()) + } + return s.repository.MarkBadgeGrantEventDelivered(ctx, event, s.now().UnixMilli()) +} + +func isBadgeGrantWalletProjectionMessage(message walletmq.WalletOutboxMessage) bool { + if strings.TrimSpace(message.AssetType) != "RESOURCE" { + return false + } + switch message.EventType { + case "ResourceGranted", "ResourceGroupGranted": + return true + default: + return false + } +} + func isBadgeGrantProjectionStatusTerminal(status string) bool { switch status { case "consumed", "skipped", "duplicate": diff --git a/services/wallet-service/internal/service/wallet/service_test.go b/services/wallet-service/internal/service/wallet/service_test.go index 0a6f7e4d..0b03cd92 100644 --- a/services/wallet-service/internal/service/wallet/service_test.go +++ b/services/wallet-service/internal/service/wallet/service_test.go @@ -9,13 +9,31 @@ import ( "google.golang.org/grpc" activityv1 "hyapp.local/api/proto/activity/v1" "hyapp/pkg/appcode" + "hyapp/pkg/walletmq" "hyapp/pkg/xerr" "hyapp/services/wallet-service/internal/domain/ledger" resourcedomain "hyapp/services/wallet-service/internal/domain/resource" walletservice "hyapp/services/wallet-service/internal/service/wallet" + mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql" "hyapp/services/wallet-service/internal/testutil/mysqltest" ) +func walletOutboxMessageFromRecord(record mysqlstorage.WalletOutboxRecord) walletmq.WalletOutboxMessage { + return walletmq.WalletOutboxMessage{ + AppCode: record.AppCode, + EventID: record.EventID, + EventType: record.EventType, + TransactionID: record.TransactionID, + CommandID: record.CommandID, + UserID: record.UserID, + AssetType: record.AssetType, + AvailableDelta: record.AvailableDelta, + FrozenDelta: record.FrozenDelta, + PayloadJSON: record.PayloadJSON, + OccurredAtMS: record.CreatedAtMS, + } +} + // TestDebitGiftUsesServerPriceAndCreditsGiftPoint 验证送礼只信服务端价格并产生双边分录。 func TestDebitGiftUsesServerPriceAndCreditsGiftPoint(t *testing.T) { repository := mysqltest.NewRepository(t) @@ -756,16 +774,32 @@ func TestGetUserGiftWallAggregatesSettledGifts(t *testing.T) { if got := repository.CountRows("user_gift_wall", "user_id = ?", 10002); got != 0 { t.Fatalf("gift wall must not be updated inside DebitGift transaction, got %d rows", got) } - processed, err := svc.ProjectPendingGiftWallEvents(context.Background(), "test-worker", 10, time.Minute) + records, err := repository.ClaimPendingWalletOutbox(context.Background(), "gift-wall-mq-test", 20, time.Now().Add(time.Minute).UnixMilli()) if err != nil { - t.Fatalf("ProjectPendingGiftWallEvents failed: %v", err) + t.Fatalf("ClaimPendingWalletOutbox failed: %v", err) + } + processed := 0 + for _, record := range records { + handled, err := svc.ProcessWalletProjectionMessage(context.Background(), "test-worker", walletOutboxMessageFromRecord(record), time.Minute) + if err != nil { + t.Fatalf("ProcessWalletProjectionMessage failed: %v", err) + } + if handled { + processed++ + } } if processed != 3 { t.Fatalf("projection should process three gift events, got %d", processed) } - processed, err = svc.ProjectPendingGiftWallEvents(context.Background(), "test-worker", 10, time.Minute) - if err != nil { - t.Fatalf("ProjectPendingGiftWallEvents retry failed: %v", err) + processed = 0 + for _, record := range records { + handled, err := svc.ProcessWalletProjectionMessage(context.Background(), "test-worker", walletOutboxMessageFromRecord(record), time.Minute) + if err != nil { + t.Fatalf("ProcessWalletProjectionMessage retry failed: %v", err) + } + if handled { + processed++ + } } if processed != 0 { t.Fatalf("projection must be idempotent after done events, got %d", processed) @@ -2041,9 +2075,19 @@ func TestProjectPendingBadgeGrantEventsRelaysWalletOutbox(t *testing.T) { t.Fatalf("GrantResource failed: %v", err) } - processed, err := svc.ProjectPendingBadgeGrantEvents(ctx, "test-badge-worker", 10, time.Minute) + records, err := repository.ClaimPendingWalletOutbox(ctx, "badge-mq-test", 20, time.Now().Add(time.Minute).UnixMilli()) if err != nil { - t.Fatalf("ProjectPendingBadgeGrantEvents failed: %v", err) + t.Fatalf("ClaimPendingWalletOutbox failed: %v", err) + } + processed := 0 + for _, record := range records { + handled, err := svc.ProcessWalletProjectionMessage(ctx, "test-badge-worker", walletOutboxMessageFromRecord(record), time.Minute) + if err != nil { + t.Fatalf("ProcessWalletProjectionMessage failed: %v", err) + } + if handled { + processed++ + } } if processed != 1 || len(activity.requests) != 1 { t.Fatalf("badge projection relay mismatch: processed=%d requests=%d", processed, len(activity.requests)) @@ -2074,9 +2118,18 @@ func TestProjectPendingBadgeGrantEventsRelaysWalletOutbox(t *testing.T) { if got := repository.CountRows("wallet_projection_events", "projection_name = ? AND status = ?", "activity_badge_display", "done"); got != 1 { t.Fatalf("badge projection should mark event done, got %d", got) } - processed, err = svc.ProjectPendingBadgeGrantEvents(ctx, "test-badge-worker", 10, time.Minute) - if err != nil || processed != 0 || len(activity.requests) != 1 { - t.Fatalf("badge projection must be idempotent: processed=%d requests=%d err=%v", processed, len(activity.requests), err) + processed = 0 + for _, record := range records { + handled, err := svc.ProcessWalletProjectionMessage(ctx, "test-badge-worker", walletOutboxMessageFromRecord(record), time.Minute) + if err != nil { + t.Fatalf("ProcessWalletProjectionMessage retry failed: %v", err) + } + if handled { + processed++ + } + } + if processed != 0 || len(activity.requests) != 1 { + t.Fatalf("badge projection must be idempotent: processed=%d requests=%d", processed, len(activity.requests)) } } diff --git a/services/wallet-service/internal/storage/mysql/badge_projection_relay.go b/services/wallet-service/internal/storage/mysql/badge_projection_relay.go index 3196dfe8..af4477ec 100644 --- a/services/wallet-service/internal/storage/mysql/badge_projection_relay.go +++ b/services/wallet-service/internal/storage/mysql/badge_projection_relay.go @@ -10,6 +10,7 @@ import ( "time" "hyapp/pkg/appcode" + "hyapp/pkg/walletmq" "hyapp/pkg/xerr" resourcedomain "hyapp/services/wallet-service/internal/domain/resource" ) @@ -75,6 +76,47 @@ func (r *Repository) ClaimPendingBadgeGrantEvents(ctx context.Context, workerID return events, nil } +// ClaimBadgeGrantMessage claims one wallet_outbox MQ fact for badge display projection. +// Runtime projection consumes MQ and does not scan wallet_outbox for candidates. +func (r *Repository) ClaimBadgeGrantMessage(ctx context.Context, workerID string, message walletmq.WalletOutboxMessage, nowMS int64, lockTTL time.Duration) (resourcedomain.BadgeGrantOutbox, bool, error) { + if r == nil || r.db == nil { + return resourcedomain.BadgeGrantOutbox{}, false, xerr.New(xerr.Unavailable, "mysql repository is not configured") + } + if !isBadgeGrantWalletMessage(message) { + return resourcedomain.BadgeGrantOutbox{}, false, nil + } + workerID = strings.TrimSpace(workerID) + if workerID == "" { + return resourcedomain.BadgeGrantOutbox{}, false, xerr.New(xerr.InvalidArgument, "worker_id is required") + } + if nowMS <= 0 { + nowMS = time.Now().UnixMilli() + } + if lockTTL <= 0 { + lockTTL = 30 * time.Second + } + event := resourcedomain.BadgeGrantOutbox{ + AppCode: appcode.Normalize(message.AppCode), + EventID: strings.TrimSpace(message.EventID), + EventType: strings.TrimSpace(message.EventType), + CommandID: strings.TrimSpace(message.CommandID), + UserID: message.UserID, + PayloadJSON: strings.TrimSpace(message.PayloadJSON), + CreatedAtMS: message.OccurredAtMS, + UpdatedAtMS: message.OccurredAtMS, + } + if event.PayloadJSON == "" { + event.PayloadJSON = "{}" + } + ctx = appcode.WithContext(ctx, event.AppCode) + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return resourcedomain.BadgeGrantOutbox{}, false, err + } + defer func() { _ = tx.Rollback() }() + return r.claimBadgeGrantEventInTx(ctx, tx, event, workerID, nowMS, lockTTL) +} + func (r *Repository) listBadgeGrantProjectionCandidates(ctx context.Context, nowMS int64, limit int) ([]badgeGrantProjectionCandidate, error) { appCode := appcode.FromContext(ctx) rows, err := r.db.QueryContext(ctx, ` @@ -145,6 +187,18 @@ func (r *Repository) listBadgeGrantProjectionCandidates(ctx context.Context, now return candidates, nil } +func isBadgeGrantWalletMessage(message walletmq.WalletOutboxMessage) bool { + if strings.TrimSpace(message.AssetType) != resourceOutboxAsset { + return false + } + switch message.EventType { + case walletResourceGrantedEvent, walletResourceGroupGrantedEvent: + return true + default: + return false + } +} + func scanBadgeGrantProjectionCandidates(rows *sql.Rows, limit int) ([]badgeGrantProjectionCandidate, error) { defer rows.Close() candidates := make([]badgeGrantProjectionCandidate, 0, limit) @@ -159,7 +213,6 @@ func scanBadgeGrantProjectionCandidates(rows *sql.Rows, limit int) ([]badgeGrant } func (r *Repository) claimBadgeGrantEvent(ctx context.Context, candidate badgeGrantProjectionCandidate, workerID string, nowMS int64, lockTTL time.Duration) (resourcedomain.BadgeGrantOutbox, bool, error) { - lockUntilMS := time.UnixMilli(nowMS).Add(lockTTL).UnixMilli() ctx = appcode.WithContext(ctx, candidate.AppCode) tx, err := r.db.BeginTx(ctx, nil) @@ -175,6 +228,12 @@ func (r *Repository) claimBadgeGrantEvent(ctx context.Context, candidate badgeGr } return resourcedomain.BadgeGrantOutbox{}, false, err } + return r.claimBadgeGrantEventInTx(ctx, tx, event, workerID, nowMS, lockTTL) +} + +func (r *Repository) claimBadgeGrantEventInTx(ctx context.Context, tx *sql.Tx, event resourcedomain.BadgeGrantOutbox, workerID string, nowMS int64, lockTTL time.Duration) (resourcedomain.BadgeGrantOutbox, bool, error) { + lockUntilMS := time.UnixMilli(nowMS).Add(lockTTL).UnixMilli() + claimed, err := r.claimBadgeProjectionEvent(ctx, tx, event, workerID, lockUntilMS, nowMS) if err != nil || !claimed { return resourcedomain.BadgeGrantOutbox{}, false, err diff --git a/services/wallet-service/internal/storage/mysql/gift_wall_projection.go b/services/wallet-service/internal/storage/mysql/gift_wall_projection.go index 9d7c878d..d69f4c62 100644 --- a/services/wallet-service/internal/storage/mysql/gift_wall_projection.go +++ b/services/wallet-service/internal/storage/mysql/gift_wall_projection.go @@ -10,6 +10,7 @@ import ( "time" "hyapp/pkg/appcode" + "hyapp/pkg/walletmq" "hyapp/pkg/xerr" ) @@ -79,6 +80,44 @@ func (r *Repository) ProjectPendingGiftWallEvents(ctx context.Context, workerID return processed, nil } +// ProjectGiftWallMessage projects one wallet_outbox MQ fact into user_gift_wall. +// Runtime projection consumes MQ and does not scan wallet_outbox for candidates. +func (r *Repository) ProjectGiftWallMessage(ctx context.Context, workerID string, message walletmq.WalletOutboxMessage, lockTTL time.Duration) (bool, error) { + if r == nil || r.db == nil { + return false, xerr.New(xerr.Unavailable, "mysql repository is not configured") + } + if message.EventType != walletGiftDebitedEvent { + return false, nil + } + workerID = strings.TrimSpace(workerID) + if workerID == "" { + return false, xerr.New(xerr.InvalidArgument, "worker_id is required") + } + if lockTTL <= 0 { + lockTTL = 30 * time.Second + } + nowMs := time.Now().UnixMilli() + lockUntilMS := time.Now().Add(lockTTL).UnixMilli() + event := giftWallProjectionEvent{ + AppCode: appcode.Normalize(message.AppCode), + EventID: strings.TrimSpace(message.EventID), + TransactionID: strings.TrimSpace(message.TransactionID), + CommandID: strings.TrimSpace(message.CommandID), + PayloadJSON: strings.TrimSpace(message.PayloadJSON), + CreatedAtMS: message.OccurredAtMS, + } + if event.PayloadJSON == "" { + event.PayloadJSON = "{}" + } + ctx = appcode.WithContext(ctx, event.AppCode) + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return false, err + } + defer func() { _ = tx.Rollback() }() + return r.projectGiftWallLockedEvent(ctx, tx, event, workerID, lockUntilMS, nowMs) +} + func (r *Repository) listGiftWallProjectionCandidates(ctx context.Context, limit int) ([]giftWallProjectionCandidate, error) { nowMs := time.Now().UnixMilli() appCode := appcode.FromContext(ctx) @@ -176,6 +215,10 @@ func (r *Repository) projectGiftWallEvent(ctx context.Context, candidate giftWal } return false, err } + return r.projectGiftWallLockedEvent(ctx, tx, event, workerID, lockUntilMS, nowMs) +} + +func (r *Repository) projectGiftWallLockedEvent(ctx context.Context, tx *sql.Tx, event giftWallProjectionEvent, workerID string, lockUntilMS int64, nowMs int64) (bool, error) { claimed, err := r.claimProjectionEvent(ctx, tx, event, workerID, lockUntilMS, nowMs) if err != nil || !claimed { return false, err