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

fix: consume wallet projections from mq

Browse Source
This commit is contained in:
zhx 2026-06-01 15:01:23 +08:00
parent bd0ed78546
commit 0aef70ecd7
9 changed files with 359 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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

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

@ -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:

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

@ -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

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

@ -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),
}
}

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

@ -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
}

113
services/wallet-service/internal/service/wallet/service.go
View File

@ -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":

73
services/wallet-service/internal/service/wallet/service_test.go
View File

@ -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))
}
}

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

@ -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

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

@ -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