From 1fe23326c6027dfca12ec53bf8f4de7da6b92083 Mon Sep 17 00:00:00 2001 From: zhx Date: Sat, 11 Jul 2026 04:34:26 +0800 Subject: [PATCH] feat: add lossless typed wallet MQ tags --- pkg/walletmq/messages.go | 77 ++++++++++++ pkg/walletmq/messages_test.go | 61 +++++++++- services/activity-service/internal/app/mq.go | 30 +++-- .../internal/app/task_events.go | 2 +- .../internal/app/vip_rebate_notice.go | 2 +- .../internal/app/wallet_events.go | 16 +-- services/notice-service/internal/app/app.go | 13 ++- .../modules/walletnotice/mysql_repository.go | 4 +- .../statistics-service/internal/app/app.go | 47 ++++++-- .../internal/app/local_flow_test.go | 39 +++++++ services/user-service/internal/app/app.go | 8 +- .../wallet-service/configs/config.docker.yaml | 2 + .../configs/config.tencent.example.yaml | 2 + services/wallet-service/configs/config.yaml | 2 + ...60711_retry_incomplete_resource_outbox.sql | 110 ++++++++++++++++++ services/wallet-service/internal/app/app.go | 2 + services/wallet-service/internal/app/mq.go | 12 +- .../internal/app/outbox_worker.go | 34 +++++- .../internal/app/outbox_worker_test.go | 85 ++++++++++++++ .../wallet-service/internal/config/config.go | 54 ++++++--- .../internal/config/config_test.go | 49 ++++++++ .../internal/service/wallet/projection.go | 2 +- .../storage/mysql/badge_projection_relay.go | 4 +- .../storage/mysql/gift_wall_projection.go | 2 +- .../internal/storage/mysql/resource_outbox.go | 10 +- .../storage/mysql/resource_outbox_test.go | 98 ++++++++++++++++ 26 files changed, 706 insertions(+), 61 deletions(-) create mode 100644 services/wallet-service/deploy/mysql/maintenance/20260711_retry_incomplete_resource_outbox.sql create mode 100644 services/wallet-service/internal/app/outbox_worker_test.go create mode 100644 services/wallet-service/internal/storage/mysql/resource_outbox_test.go diff --git a/pkg/walletmq/messages.go b/pkg/walletmq/messages.go index 062a8f86..8370637b 100644 --- a/pkg/walletmq/messages.go +++ b/pkg/walletmq/messages.go @@ -4,15 +4,92 @@ package walletmq import ( "encoding/json" "errors" + "fmt" + "sort" "strings" ) const ( MessageTypeWalletOutboxEvent = "wallet_outbox_event" + // TagWalletOutboxEvent 是 typed tag 上线前所有钱包事实共用的 legacy Tag。 + // 兼容窗口内消费者必须同时订阅它和自己需要的 EventType Tag，确保 producer + // 切换前已经进入 Broker 的旧消息仍会被同一 consumer group 完整消费。 TagWalletOutboxEvent = "wallet_outbox_event" + + // 以下常量只描述当前 wallet topic 消费者实际处理的事件类型。producer 使用 + // outbox.event_type 作为 Tag，因此消费者和消息体分派必须复用同一拼写，避免 + // Tag 过滤成功但 handler 因事件名漂移而静默跳过。 + EventTypeWalletRechargeRecorded = "WalletRechargeRecorded" + EventTypeWalletCoinSellerStockPurchased = "WalletCoinSellerStockPurchased" + EventTypeWalletLuckyGiftRewardCredited = "WalletLuckyGiftRewardCredited" + EventTypeWalletTaskRewardCredited = "WalletTaskRewardCredited" + EventTypeWalletWheelRewardCredited = "WalletWheelRewardCredited" + EventTypeWalletRoomTurnoverRewardCredited = "WalletRoomTurnoverRewardCredited" + EventTypeWalletInviteActivityRewardCredited = "WalletInviteActivityRewardCredited" + EventTypeWalletAgencyOpeningRewardCredited = "WalletAgencyOpeningRewardCredited" + EventTypeWalletGiftIncomeCredited = "WalletGiftIncomeCredited" + EventTypeWalletBalanceChanged = "WalletBalanceChanged" + EventTypeWalletGiftDebited = "WalletGiftDebited" + EventTypeWalletRedPacketCreated = "WalletRedPacketCreated" + EventTypeWalletRedPacketClaimed = "WalletRedPacketClaimed" + EventTypeWalletRedPacketRefunded = "WalletRedPacketRefunded" + EventTypeVIPActivated = "VipActivated" + EventTypeVIPDailyCoinRebateAvailable = "VipDailyCoinRebateAvailable" + EventTypeResourceGranted = "ResourceGranted" + EventTypeResourceGroupGranted = "ResourceGroupGranted" ) +// EventTypeTag converts one committed wallet event type into its RocketMQ Tag. +// Event types are controlled identifiers rather than user input; rejecting expression +// operators and punctuation here prevents a malformed outbox row from widening a +// subscription or being published under a Tag no consumer can express safely. +func EventTypeTag(eventType string) (string, error) { + tag := strings.TrimSpace(eventType) + if tag == "" { + return "", errors.New("wallet event_type tag is required") + } + if tag == TagWalletOutboxEvent { + return "", errors.New("wallet event_type cannot reuse the legacy wallet tag") + } + for _, char := range tag { + if (char >= 'a' && char <= 'z') || + (char >= 'A' && char <= 'Z') || + (char >= '0' && char <= '9') || + char == '_' || char == '-' { + continue + } + return "", fmt.Errorf("wallet event_type %q cannot be used as a RocketMQ tag", eventType) + } + return tag, nil +} + +// LegacyCompatibleTagExpression builds the compatibility selector used during +// the typed-tag migration: legacy messages remain consumable, while newly published +// messages are filtered by the exact event types handled by this consumer. +// Sorting and deduplication make the expression deterministic so every process in +// the same consumer group registers byte-for-byte identical subscription metadata. +func LegacyCompatibleTagExpression(eventTypes ...string) (string, error) { + typedTags := make([]string, 0, len(eventTypes)) + seen := make(map[string]struct{}, len(eventTypes)) + for _, eventType := range eventTypes { + tag, err := EventTypeTag(eventType) + if err != nil { + return "", err + } + if _, exists := seen[tag]; exists { + continue + } + seen[tag] = struct{}{} + typedTags = append(typedTags, tag) + } + if len(typedTags) == 0 { + return "", errors.New("at least one typed wallet event tag is required") + } + sort.Strings(typedTags) + return TagWalletOutboxEvent + " || " + strings.Join(typedTags, " || "), nil +} + // WalletOutboxMessage is the MQ representation of one committed wallet_outbox fact. type WalletOutboxMessage struct { MessageType string `json:"message_type"` diff --git a/pkg/walletmq/messages_test.go b/pkg/walletmq/messages_test.go index c31f021c..c638ee68 100644 --- a/pkg/walletmq/messages_test.go +++ b/pkg/walletmq/messages_test.go @@ -1,6 +1,9 @@ package walletmq -import "testing" +import ( + "strings" + "testing" +) func TestWalletOutboxMessageRoundTrip(t *testing.T) { encoded, err := EncodeWalletOutboxMessage(WalletOutboxMessage{ @@ -27,3 +30,59 @@ func TestWalletOutboxMessageRoundTrip(t *testing.T) { t.Fatalf("decoded message mismatch: %+v", decoded) } } + +func TestEventTypeTagUsesExactTrimmedEventType(t *testing.T) { + tag, err := EventTypeTag(" WalletRechargeRecorded ") + if err != nil { + t.Fatalf("EventTypeTag failed: %v", err) + } + if tag != EventTypeWalletRechargeRecorded { + t.Fatalf("event tag = %q, want %q", tag, EventTypeWalletRechargeRecorded) + } + + for _, invalid := range []string{"", TagWalletOutboxEvent, "Wallet RechargeRecorded", "WalletRechargeRecorded || *", "Wallet.RechargeRecorded"} { + if _, err := EventTypeTag(invalid); err == nil { + t.Fatalf("EventTypeTag(%q) unexpectedly succeeded", invalid) + } + } +} + +func TestLegacyCompatibleTagExpressionIsDeterministicAndDeduplicated(t *testing.T) { + first, err := LegacyCompatibleTagExpression( + EventTypeWalletGiftDebited, + EventTypeWalletRechargeRecorded, + EventTypeWalletGiftDebited, + ) + if err != nil { + t.Fatalf("LegacyCompatibleTagExpression failed: %v", err) + } + second, err := LegacyCompatibleTagExpression( + EventTypeWalletRechargeRecorded, + EventTypeWalletGiftDebited, + ) + if err != nil { + t.Fatalf("LegacyCompatibleTagExpression failed: %v", err) + } + if first != second { + t.Fatalf("expressions differ by input order: first=%q second=%q", first, second) + } + want := TagWalletOutboxEvent + " || " + EventTypeWalletGiftDebited + " || " + EventTypeWalletRechargeRecorded + if first != want { + t.Fatalf("expression = %q, want %q", first, want) + } + if strings.Count(first, EventTypeWalletGiftDebited) != 1 { + t.Fatalf("typed tag was not deduplicated: %q", first) + } +} + +func TestLegacyCompatibleTagExpressionRejectsUnsafeOrMissingTypedTags(t *testing.T) { + for _, eventTypes := range [][]string{ + nil, + {TagWalletOutboxEvent}, + {"WalletRechargeRecorded || *"}, + } { + if _, err := LegacyCompatibleTagExpression(eventTypes...); err == nil { + t.Fatalf("LegacyCompatibleTagExpression(%v) unexpectedly succeeded", eventTypes) + } + } +} diff --git a/services/activity-service/internal/app/mq.go b/services/activity-service/internal/app/mq.go index b31077ad..755ca000 100644 --- a/services/activity-service/internal/app/mq.go +++ b/services/activity-service/internal/app/mq.go @@ -186,7 +186,7 @@ func newFirstRechargeWalletConsumer(cfg config.Config, services *serviceBundle) if err != nil { return nil, err } - if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { + if err := subscribeWalletOutboxEvents(consumer, cfg.RocketMQ.WalletOutbox.Topic, []string{walletmq.EventTypeWalletRechargeRecorded}, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { event, ok, err := firstRechargeEventFromWalletMessage(message.Body) if err != nil || !ok { return err @@ -205,7 +205,7 @@ func newCumulativeRechargeWalletConsumer(cfg config.Config, services *serviceBun if err != nil { return nil, err } - if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { + if err := subscribeWalletOutboxEvents(consumer, cfg.RocketMQ.WalletOutbox.Topic, []string{walletmq.EventTypeWalletRechargeRecorded}, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { event, ok, err := cumulativeRechargeEventFromWalletMessage(message.Body) if err != nil || !ok { return err @@ -224,7 +224,7 @@ func newInviteActivityWalletConsumer(cfg config.Config, services *serviceBundle) if err != nil { return nil, err } - if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { + if err := subscribeWalletOutboxEvents(consumer, cfg.RocketMQ.WalletOutbox.Topic, []string{walletmq.EventTypeWalletRechargeRecorded}, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { event, ok, err := inviteActivityRechargeEventFromWalletMessage(message.Body) if err != nil || !ok { return err @@ -268,7 +268,11 @@ func newRedPacketWalletConsumer(cfg config.Config, services *serviceBundle) (*ro if err != nil { return nil, err } - if err := consumer.Subscribe(redPacketWalletOutboxTopic(cfg.RocketMQ.WalletOutbox), walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { + if err := subscribeWalletOutboxEvents(consumer, redPacketWalletOutboxTopic(cfg.RocketMQ.WalletOutbox), []string{ + walletmq.EventTypeWalletRedPacketCreated, + walletmq.EventTypeWalletRedPacketClaimed, + walletmq.EventTypeWalletRedPacketRefunded, + }, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { event, ok, err := redPacketEventFromWalletMessage(message.Body) if err != nil || !ok { return err @@ -288,7 +292,7 @@ func newVIPRebateNoticeWalletConsumer(cfg config.Config, services *serviceBundle if err != nil { return nil, err } - if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { + if err := subscribeWalletOutboxEvents(consumer, cfg.RocketMQ.WalletOutbox.Topic, []string{walletmq.EventTypeVIPDailyCoinRebateAvailable}, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { event, ok, err := vipRebateAvailableEventFromWalletMessage(message.Body) if err != nil || !ok { return err @@ -310,7 +314,7 @@ func newUserLeaderboardWalletConsumer(cfg config.Config, services *serviceBundle if err != nil { return nil, err } - if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { + if err := subscribeWalletOutboxEvents(consumer, cfg.RocketMQ.WalletOutbox.Topic, []string{walletmq.EventTypeWalletGiftDebited}, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { event, ok, err := userLeaderboardGiftEventFromWalletMessage(message.Body) if err != nil || !ok { return err @@ -330,7 +334,7 @@ func newTaskWalletConsumer(cfg config.Config, services *serviceBundle) (*rocketm if err != nil { return nil, err } - if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { + if err := subscribeWalletOutboxEvents(consumer, cfg.RocketMQ.WalletOutbox.Topic, []string{walletmq.EventTypeWalletRechargeRecorded}, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { // 钱包任务消费者使用独立 consumer group，不能复用红包或其他钱包消费者进度；否则充值任务会被其他业务确认掉。 event, appCode, ok, err := taskEventFromWalletMessage(message.Body) if err != nil || !ok { @@ -346,6 +350,18 @@ func newTaskWalletConsumer(cfg config.Config, services *serviceBundle) (*rocketm return consumer, nil } +// subscribeWalletOutboxEvents is the only activity-service entry point for wallet +// subscriptions during the typed-tag compatibility phase. Keeping selector assembly +// centralized guarantees every replica in a consumer group registers the same sorted +// expression and that the legacy Tag remains present until the old backlog is drained. +func subscribeWalletOutboxEvents(consumer *rocketmqx.Consumer, topic string, eventTypes []string, handler rocketmqx.Handler) error { + tagExpression, err := walletmq.LegacyCompatibleTagExpression(eventTypes...) + if err != nil { + return err + } + return consumer.Subscribe(topic, tagExpression, handler) +} + func newTaskUserOutboxConsumer(cfg config.Config, services *serviceBundle) (*rocketmqx.Consumer, error) { consumer, err := rocketmqx.NewConsumer(userOutboxTaskConsumerConfig(cfg.RocketMQ)) if err != nil { diff --git a/services/activity-service/internal/app/task_events.go b/services/activity-service/internal/app/task_events.go index bbf142e3..300020c5 100644 --- a/services/activity-service/internal/app/task_events.go +++ b/services/activity-service/internal/app/task_events.go @@ -13,7 +13,7 @@ import ( ) const ( - taskWalletRechargeRecorded = "WalletRechargeRecorded" + taskWalletRechargeRecorded = walletmq.EventTypeWalletRechargeRecorded taskCPRelationshipCreated = "UserCPRelationshipCreated" taskMicDailyStatsUpdated = "UserMicDailyStatsUpdated" taskMoodPublished = "UserMoodPublished" diff --git a/services/activity-service/internal/app/vip_rebate_notice.go b/services/activity-service/internal/app/vip_rebate_notice.go index 167f2d70..a478c2e6 100644 --- a/services/activity-service/internal/app/vip_rebate_notice.go +++ b/services/activity-service/internal/app/vip_rebate_notice.go @@ -13,7 +13,7 @@ import ( messageservice "hyapp/services/activity-service/internal/service/message" ) -const vipDailyCoinRebateAvailableEventType = "VipDailyCoinRebateAvailable" +const vipDailyCoinRebateAvailableEventType = walletmq.EventTypeVIPDailyCoinRebateAvailable // vipRebateAvailableEvent 是 wallet outbox 的逐用户返现资格快照。 // activity-service 只把这个已提交事实投影为 inbox；资格、金额、过期和最终入账仍由 wallet owner 判定。 diff --git a/services/activity-service/internal/app/wallet_events.go b/services/activity-service/internal/app/wallet_events.go index 7781b3a4..1a9009b2 100644 --- a/services/activity-service/internal/app/wallet_events.go +++ b/services/activity-service/internal/app/wallet_events.go @@ -18,7 +18,7 @@ func firstRechargeEventFromWalletMessage(body []byte) (firstrechargedomain.Recha if err != nil { return firstrechargedomain.RechargeEvent{}, false, err } - if message.EventType != firstrechargedomain.RechargeEventWalletRecorded { + if message.EventType != walletmq.EventTypeWalletRechargeRecorded { return firstrechargedomain.RechargeEvent{}, false, nil } fields := firstRechargeFactFromWalletPayload(message.PayloadJSON) @@ -49,7 +49,7 @@ func cumulativeRechargeEventFromWalletMessage(body []byte) (cumulativerechargedo if err != nil { return cumulativerechargedomain.RechargeEvent{}, false, err } - if message.EventType != cumulativerechargedomain.RechargeEventWalletRecorded { + if message.EventType != walletmq.EventTypeWalletRechargeRecorded { return cumulativerechargedomain.RechargeEvent{}, false, nil } // WalletRechargeRecorded 是累充唯一事实源；payload 里没有可折算 USD 金额时直接跳过，避免把普通钱包流水计入活动。 @@ -78,7 +78,7 @@ func inviteActivityRechargeEventFromWalletMessage(body []byte) (inviteactivitydo if err != nil { return inviteactivitydomain.RechargeEvent{}, false, err } - if message.EventType != inviteactivitydomain.EventTypeWalletRechargeRecorded || message.AssetType != "COIN" || message.AvailableDelta <= 0 { + if message.EventType != walletmq.EventTypeWalletRechargeRecorded || message.AssetType != "COIN" || message.AvailableDelta <= 0 { return inviteactivitydomain.RechargeEvent{}, false, nil } fields := cumulativeRechargeFactFromWalletPayload(message.PayloadJSON, message.AvailableDelta) @@ -101,7 +101,9 @@ func redPacketEventFromWalletMessage(body []byte) (broadcastdomain.RedPacketWall if err != nil { return broadcastdomain.RedPacketWalletEvent{}, false, err } - if message.EventType != "WalletRedPacketCreated" && message.EventType != "WalletRedPacketClaimed" && message.EventType != "WalletRedPacketRefunded" { + if message.EventType != walletmq.EventTypeWalletRedPacketCreated && + message.EventType != walletmq.EventTypeWalletRedPacketClaimed && + message.EventType != walletmq.EventTypeWalletRedPacketRefunded { return broadcastdomain.RedPacketWalletEvent{}, false, nil } fields := redPacketFieldsFromWalletPayload(message.PayloadJSON, message.EventType) @@ -138,7 +140,7 @@ func userLeaderboardGiftEventFromWalletMessage(body []byte) (userleaderboard.Gif if err != nil { return userleaderboard.GiftEvent{}, false, err } - if message.EventType != "WalletGiftDebited" { + if message.EventType != walletmq.EventTypeWalletGiftDebited { return userleaderboard.GiftEvent{}, false, nil } var decoded map[string]any @@ -310,12 +312,12 @@ func redPacketFieldsFromWalletPayload(payload string, eventType string) redPacke status := stringFromDecoded(decoded, "status") openAtMS := int64FromDecoded(decoded, "open_at_ms") expiresAtMS := int64FromDecoded(decoded, "expires_at_ms") - if eventType == "WalletRedPacketClaimed" || eventType == "WalletRedPacketRefunded" { + if eventType == walletmq.EventTypeWalletRedPacketClaimed || eventType == walletmq.EventTypeWalletRedPacketRefunded { status = stringFromDecoded(decoded, "packet_status") } totalCount := int32FromDecoded(decoded, "packet_count") remainingCount := int32FromDecoded(decoded, "remaining_count") - if remainingCount == 0 && eventType == "WalletRedPacketCreated" { + if remainingCount == 0 && eventType == walletmq.EventTypeWalletRedPacketCreated { remainingCount = totalCount } return redPacketPayloadFields{ diff --git a/services/notice-service/internal/app/app.go b/services/notice-service/internal/app/app.go index 20d95ea7..75cf4e9b 100644 --- a/services/notice-service/internal/app/app.go +++ b/services/notice-service/internal/app/app.go @@ -148,13 +148,24 @@ func New(cfg config.Config) (*App, error) { confirmOptions := confirmNoticeWorkerOptions(cfg.ConfirmNoticeWorker) mqConsumers := make([]*rocketmqx.Consumer, 0, 4) if cfg.RocketMQ.WalletOutbox.Enabled { + // notice-service 只投递余额变化和 VIP 激活私信。保留 legacy Tag 是迁移期 + // 的零丢失边界，新 producer 的其他钱包事实由 Broker 直接过滤。 + walletTagExpression, err := walletmq.LegacyCompatibleTagExpression( + walletmq.EventTypeWalletBalanceChanged, + walletmq.EventTypeVIPActivated, + ) + if err != nil { + _ = listener.Close() + _ = store.Close() + return nil, err + } consumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.WalletOutbox.ConsumerGroup, cfg.RocketMQ.WalletOutbox.ConsumerMaxReconsumeTimes)) if err != nil { _ = listener.Close() _ = store.Close() return nil, err } - if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { + if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletTagExpression, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { walletMessage, err := walletmq.DecodeWalletOutboxMessage(message.Body) if err != nil { return err diff --git a/services/notice-service/internal/modules/walletnotice/mysql_repository.go b/services/notice-service/internal/modules/walletnotice/mysql_repository.go index ab2441aa..9407605d 100644 --- a/services/notice-service/internal/modules/walletnotice/mysql_repository.go +++ b/services/notice-service/internal/modules/walletnotice/mysql_repository.go @@ -18,8 +18,8 @@ import ( const ( sourceWalletOutbox = "wallet_outbox" channelTencentIMC2C = "tencent_im_c2c" - eventWalletBalance = "WalletBalanceChanged" - eventVIPActivated = "VipActivated" + eventWalletBalance = walletmq.EventTypeWalletBalanceChanged + eventVIPActivated = walletmq.EventTypeVIPActivated robotCoinAssetType = "ROBOT_COIN" deliveryStatusDelivering = "delivering" deliveryStatusDelivered = "delivered" diff --git a/services/statistics-service/internal/app/app.go b/services/statistics-service/internal/app/app.go index 7dd67fdd..b77743a7 100644 --- a/services/statistics-service/internal/app/app.go +++ b/services/statistics-service/internal/app/app.go @@ -145,7 +145,15 @@ func newConsumers(cfg config.Config, repo *mysqlstorage.Repository) ([]*rocketmq }); err != nil { return nil, err } - if err := add(cfg.RocketMQ.WalletOutbox, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { + // 统计服务需要多个钱包事实，但不应为未建模的 gift debit、红包等事件支付 + // 消费 TPS。兼容表达式继续接住存量 legacy 消息，新消息由 Broker 按下列 + // 精确事件集合过滤。 + walletTagExpression, err := walletmq.LegacyCompatibleTagExpression(statisticsWalletEventTypes()...) + if err != nil { + shutdownConsumers(consumers) + return nil, err + } + if err := add(cfg.RocketMQ.WalletOutbox, walletTagExpression, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { return ConsumeWalletOutboxBody(ctx, repo, message.Body) }); err != nil { shutdownConsumers(consumers) @@ -166,6 +174,21 @@ func newConsumers(cfg config.Config, repo *mysqlstorage.Repository) ([]*rocketmq return consumers, nil } +func statisticsWalletEventTypes() []string { + return []string{ + walletmq.EventTypeWalletRechargeRecorded, + walletmq.EventTypeWalletCoinSellerStockPurchased, + walletmq.EventTypeWalletLuckyGiftRewardCredited, + walletmq.EventTypeWalletTaskRewardCredited, + walletmq.EventTypeWalletWheelRewardCredited, + walletmq.EventTypeWalletRoomTurnoverRewardCredited, + walletmq.EventTypeWalletInviteActivityRewardCredited, + walletmq.EventTypeWalletAgencyOpeningRewardCredited, + walletmq.EventTypeWalletGiftIncomeCredited, + walletmq.EventTypeWalletBalanceChanged, + } +} + // ConsumeUserOutboxBody 是 user outbox 的唯一统计投影入口；实时 MQ 和离线回灌共用它，避免同一 payload 出现两套解析口径。 func ConsumeUserOutboxBody(ctx context.Context, repo *mysqlstorage.Repository, body []byte) error { if event, ok, err := userRegisteredEvent(body); err != nil || ok { @@ -359,7 +382,7 @@ func rechargeEvent(body []byte) (mysqlstorage.RechargeEvent, bool, error) { if err != nil { return mysqlstorage.RechargeEvent{}, false, err } - if message.EventType != "WalletRechargeRecorded" { + if message.EventType != walletmq.EventTypeWalletRechargeRecorded { return mysqlstorage.RechargeEvent{}, false, nil } payload := mysqlstorage.DecodeJSON(message.PayloadJSON) @@ -391,7 +414,7 @@ func coinSellerStockEvent(body []byte) (mysqlstorage.CoinSellerStockEvent, bool, if err != nil { return mysqlstorage.CoinSellerStockEvent{}, false, err } - if message.EventType != "WalletCoinSellerStockPurchased" { + if message.EventType != walletmq.EventTypeWalletCoinSellerStockPurchased { return mysqlstorage.CoinSellerStockEvent{}, false, nil } payload := mysqlstorage.DecodeJSON(message.PayloadJSON) @@ -459,7 +482,7 @@ func luckyGiftRewardEvent(body []byte) (mysqlstorage.LuckyGiftRewardEvent, bool, if err != nil { return mysqlstorage.LuckyGiftRewardEvent{}, false, err } - if message.EventType != "WalletLuckyGiftRewardCredited" { + if message.EventType != walletmq.EventTypeWalletLuckyGiftRewardCredited { return mysqlstorage.LuckyGiftRewardEvent{}, false, nil } payload := mysqlstorage.DecodeJSON(message.PayloadJSON) @@ -481,7 +504,7 @@ func platformGrantCoinEvent(body []byte) (mysqlstorage.PlatformGrantCoinEvent, b if err != nil { return mysqlstorage.PlatformGrantCoinEvent{}, false, err } - if !isPlatformGrantCoinEventType(message.EventType) || message.EventType == "WalletLuckyGiftRewardCredited" { + if !isPlatformGrantCoinEventType(message.EventType) || message.EventType == walletmq.EventTypeWalletLuckyGiftRewardCredited { return mysqlstorage.PlatformGrantCoinEvent{}, false, nil } payload := mysqlstorage.DecodeJSON(message.PayloadJSON) @@ -502,7 +525,7 @@ func giftIncomeOutputEvent(body []byte) (mysqlstorage.CoinOutputEvent, bool, err if err != nil { return mysqlstorage.CoinOutputEvent{}, false, err } - if message.EventType != "WalletGiftIncomeCredited" { + if message.EventType != walletmq.EventTypeWalletGiftIncomeCredited { return mysqlstorage.CoinOutputEvent{}, false, nil } payload := mysqlstorage.DecodeJSON(message.PayloadJSON) @@ -523,7 +546,7 @@ func walletBalanceChangedEvent(body []byte) (mysqlstorage.WalletBalanceChangedEv if err != nil { return mysqlstorage.WalletBalanceChangedEvent{}, false, err } - if message.EventType != "WalletBalanceChanged" { + if message.EventType != walletmq.EventTypeWalletBalanceChanged { return mysqlstorage.WalletBalanceChangedEvent{}, false, nil } payload := mysqlstorage.DecodeJSON(message.PayloadJSON) @@ -553,11 +576,11 @@ func walletBalanceChangedEvent(body []byte) (mysqlstorage.WalletBalanceChangedEv func isPlatformGrantCoinEventType(value string) bool { switch strings.TrimSpace(value) { - case "WalletTaskRewardCredited", - "WalletWheelRewardCredited", - "WalletRoomTurnoverRewardCredited", - "WalletInviteActivityRewardCredited", - "WalletAgencyOpeningRewardCredited": + case walletmq.EventTypeWalletTaskRewardCredited, + walletmq.EventTypeWalletWheelRewardCredited, + walletmq.EventTypeWalletRoomTurnoverRewardCredited, + walletmq.EventTypeWalletInviteActivityRewardCredited, + walletmq.EventTypeWalletAgencyOpeningRewardCredited: return true default: return false diff --git a/services/statistics-service/internal/app/local_flow_test.go b/services/statistics-service/internal/app/local_flow_test.go index 8a99c381..eb25d728 100644 --- a/services/statistics-service/internal/app/local_flow_test.go +++ b/services/statistics-service/internal/app/local_flow_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "runtime" + "strings" "testing" "time" @@ -17,6 +18,44 @@ import ( mysqlstorage "hyapp/services/statistics-service/internal/storage/mysql" ) +func TestStatisticsWalletTagExpressionMatchesModeledEvents(t *testing.T) { + eventTypes := statisticsWalletEventTypes() + want := map[string]struct{}{ + walletmq.EventTypeWalletRechargeRecorded: {}, + walletmq.EventTypeWalletCoinSellerStockPurchased: {}, + walletmq.EventTypeWalletLuckyGiftRewardCredited: {}, + walletmq.EventTypeWalletTaskRewardCredited: {}, + walletmq.EventTypeWalletWheelRewardCredited: {}, + walletmq.EventTypeWalletRoomTurnoverRewardCredited: {}, + walletmq.EventTypeWalletInviteActivityRewardCredited: {}, + walletmq.EventTypeWalletAgencyOpeningRewardCredited: {}, + walletmq.EventTypeWalletGiftIncomeCredited: {}, + walletmq.EventTypeWalletBalanceChanged: {}, + } + if len(eventTypes) != len(want) { + t.Fatalf("statistics wallet event count = %d, want %d: %v", len(eventTypes), len(want), eventTypes) + } + for _, eventType := range eventTypes { + if _, ok := want[eventType]; !ok { + t.Fatalf("statistics selector contains unmodeled event %q", eventType) + } + delete(want, eventType) + } + if len(want) != 0 { + t.Fatalf("statistics selector misses modeled events: %v", want) + } + expression, err := walletmq.LegacyCompatibleTagExpression(eventTypes...) + if err != nil { + t.Fatalf("LegacyCompatibleTagExpression failed: %v", err) + } + if !strings.Contains(expression, walletmq.TagWalletOutboxEvent) { + t.Fatalf("compatibility expression misses legacy tag: %q", expression) + } + if strings.Contains(expression, walletmq.EventTypeWalletGiftDebited) { + t.Fatalf("statistics selector must not receive high-volume gift debit facts: %q", expression) + } +} + // 这个用例只覆盖 statistics-service 对 owner outbox payload 的解析和聚合边界； // 本地真实业务链路用 scripts/databi-real-flow-check.sh 从各 owner service 的 outbox 重放验证。 func TestConsumeOutboxMessagesProjectsDataScreenMetrics(t *testing.T) { diff --git a/services/user-service/internal/app/app.go b/services/user-service/internal/app/app.go index dfa1db93..713f8836 100644 --- a/services/user-service/internal/app/app.go +++ b/services/user-service/internal/app/app.go @@ -641,11 +641,17 @@ func (i tencentIMAccountImporter) ImportAccount(ctx context.Context, userID int6 func newMQConsumers(cfg config.Config, inviteSvc *inviteservice.Service, micTimeSvc *mictimeservice.Service, roomTimeSvc *roomtimeservice.Service, cpSvc *cpservice.Service) ([]*rocketmqx.Consumer, error) { consumers := make([]*rocketmqx.Consumer, 0, 4) if cfg.InviteRechargeWorker.Enabled && cfg.RocketMQ.WalletOutbox.Enabled { + // 邀请关系只消费充值成功事实。兼容期仍订阅 legacy Tag 以排空 producer + // 切换前的存量消息，typed 消息则在 Broker 侧只投递充值事件。 + walletTagExpression, err := walletmq.LegacyCompatibleTagExpression(walletmq.EventTypeWalletRechargeRecorded) + if err != nil { + return nil, err + } consumer, err := rocketmqx.NewConsumer(walletOutboxConsumerConfig(cfg.RocketMQ)) if err != nil { return nil, err } - if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { + if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletTagExpression, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { event, ok, err := inviteRechargeEventFromWalletMessage(message.Body) if err != nil || !ok { return err diff --git a/services/wallet-service/configs/config.docker.yaml b/services/wallet-service/configs/config.docker.yaml index c24bc4db..88b8d617 100644 --- a/services/wallet-service/configs/config.docker.yaml +++ b/services/wallet-service/configs/config.docker.yaml @@ -34,6 +34,8 @@ rocketmq: namespace: "" send_timeout: "3s" retry: 2 + # Phase A 保持 legacy；所有消费者完成兼容滚动后才允许切换为 event_type。 + wallet_event_tag_mode: "legacy" wallet_outbox: enabled: true topic: "hyapp_wallet_outbox" diff --git a/services/wallet-service/configs/config.tencent.example.yaml b/services/wallet-service/configs/config.tencent.example.yaml index bb21a836..8efa94f0 100644 --- a/services/wallet-service/configs/config.tencent.example.yaml +++ b/services/wallet-service/configs/config.tencent.example.yaml @@ -34,6 +34,8 @@ rocketmq: namespace: "hyapp-prod" send_timeout: "3s" retry: 2 + # Phase A 必须保持 legacy；Phase B 经消费者订阅与积压验证后，仅将该值切为 event_type。 + wallet_event_tag_mode: "legacy" wallet_outbox: enabled: true topic: "hyapp_wallet_outbox" diff --git a/services/wallet-service/configs/config.yaml b/services/wallet-service/configs/config.yaml index 7ea71547..0c5eeb43 100644 --- a/services/wallet-service/configs/config.yaml +++ b/services/wallet-service/configs/config.yaml @@ -34,6 +34,8 @@ rocketmq: namespace: "" send_timeout: "3s" retry: 2 + # Phase A 保持 legacy；确认所有 wallet consumer 已双订阅且稳定后，Phase B 只把此值改为 event_type。 + wallet_event_tag_mode: "legacy" wallet_outbox: enabled: true topic: "hyapp_wallet_outbox" diff --git a/services/wallet-service/deploy/mysql/maintenance/20260711_retry_incomplete_resource_outbox.sql b/services/wallet-service/deploy/mysql/maintenance/20260711_retry_incomplete_resource_outbox.sql new file mode 100644 index 00000000..ba28ad8e --- /dev/null +++ b/services/wallet-service/deploy/mysql/maintenance/20260711_retry_incomplete_resource_outbox.sql @@ -0,0 +1,110 @@ +-- wallet resource outbox transaction_id repair and controlled retry. +-- +-- This file is deliberately outside initdb/migrations: Phase A must not replay any +-- historical message. Run it only after Phase B (`wallet_event_tag_mode=event_type`) +-- is live, all wallet consumer subscriptions are healthy, and the target rows below +-- have been reviewed. One execution handles at most 100 unlocked rows; repeat while +-- observing MQ lag and consumer errors. +-- +-- Preview-only (default, never mutates rows): +-- mysql ... < 20260711_retry_incomplete_resource_outbox.sql +-- +-- Explicit Phase-B execution in the same mysql session: +-- SET @execute_wallet_resource_outbox_retry = 1; +-- SOURCE /absolute/path/20260711_retry_incomplete_resource_outbox.sql; + +SET @execute_wallet_resource_outbox_retry := + IF(COALESCE(@execute_wallet_resource_outbox_retry, 0) = 1, 1, 0); + +DROP TEMPORARY TABLE IF EXISTS tmp_wallet_resource_outbox_retry; +CREATE TEMPORARY TABLE tmp_wallet_resource_outbox_retry ( + app_code VARCHAR(32) NOT NULL, + event_id VARCHAR(128) NOT NULL, + PRIMARY KEY (app_code, event_id) +) ENGINE=MEMORY; + +-- The event list is the complete set currently created through resourceOutboxEvent. +-- Empty transaction_id plus the exact encoder error prevents this maintenance action +-- from touching financial facts or resource failures caused by any other condition. +INSERT INTO tmp_wallet_resource_outbox_retry (app_code, event_id) +SELECT app_code, event_id +FROM wallet_outbox +WHERE asset_type = 'RESOURCE' + AND event_type IN ( + 'GiftConfigChanged', + 'ResourceChanged', + 'ResourceGrantRevoked', + 'ResourceGranted', + 'ResourceGroupChanged', + 'ResourceGroupGranted', + 'UserResourceChanged', + 'VipEffectiveChanged', + 'VipTrialCardGranted' + ) + AND transaction_id = '' + AND BINARY last_error = BINARY 'wallet outbox message is incomplete' + AND status IN ('failed', 'retryable') + AND worker_id = '' + AND lock_until_ms IS NULL +ORDER BY created_at_ms, event_id +LIMIT 100; + +SELECT + wo.app_code, + wo.event_id, + wo.event_type, + wo.status, + wo.retry_count, + wo.created_at_ms +FROM wallet_outbox wo +JOIN tmp_wallet_resource_outbox_retry target + ON target.app_code = wo.app_code AND target.event_id = wo.event_id +ORDER BY wo.created_at_ms, wo.event_id; + +START TRANSACTION; + +-- @execute_wallet_resource_outbox_retry defaults to 0. When explicitly set to 1, +-- transaction_id=event_id restores a stable correlation ID and status=pending asks +-- the already-running Phase-B worker to publish exactly the original outbox fact. +UPDATE wallet_outbox wo +JOIN tmp_wallet_resource_outbox_retry target + ON target.app_code = wo.app_code AND target.event_id = wo.event_id +SET wo.transaction_id = wo.event_id, + wo.status = 'pending', + wo.worker_id = '', + wo.lock_until_ms = NULL, + wo.retry_count = 0, + wo.next_retry_at_ms = NULL, + wo.last_error = NULL, + wo.updated_at_ms = CAST(UNIX_TIMESTAMP(CURRENT_TIMESTAMP(3)) * 1000 AS UNSIGNED) +WHERE @execute_wallet_resource_outbox_retry = 1 + AND wo.asset_type = 'RESOURCE' + AND wo.event_type IN ( + 'GiftConfigChanged', + 'ResourceChanged', + 'ResourceGrantRevoked', + 'ResourceGranted', + 'ResourceGroupChanged', + 'ResourceGroupGranted', + 'UserResourceChanged', + 'VipEffectiveChanged', + 'VipTrialCardGranted' + ) + AND wo.transaction_id = '' + AND BINARY wo.last_error = BINARY 'wallet outbox message is incomplete' + AND wo.status IN ('failed', 'retryable') + AND wo.worker_id = '' + AND wo.lock_until_ms IS NULL; + +SET @wallet_resource_outbox_requeued_rows := ROW_COUNT(); +COMMIT; + +SELECT + @execute_wallet_resource_outbox_retry AS execution_enabled, + @wallet_resource_outbox_requeued_rows AS requeued_rows, + CASE + WHEN @execute_wallet_resource_outbox_retry = 1 THEN 'phase_b_batch_requeued' + ELSE 'preview_only_no_rows_changed' + END AS result; + +DROP TEMPORARY TABLE IF EXISTS tmp_wallet_resource_outbox_retry; diff --git a/services/wallet-service/internal/app/app.go b/services/wallet-service/internal/app/app.go index e7bfd5f0..1b4bda64 100644 --- a/services/wallet-service/internal/app/app.go +++ b/services/wallet-service/internal/app/app.go @@ -48,6 +48,7 @@ type App struct { externalRechargeReconcileWorkerCfg config.ExternalRechargeReconcileWorkerConfig walletOutboxTopic string realtimeWalletOutboxTopic string + walletEventTagMode config.WalletEventTagMode nodeID string redPacketExpiryWorkerCfg config.RedPacketExpiryWorkerConfig stopWorker context.CancelFunc @@ -208,6 +209,7 @@ func New(cfg config.Config) (*App, error) { externalRechargeReconcileWorkerCfg: cfg.ExternalRechargeReconcileWorker, walletOutboxTopic: cfg.RocketMQ.WalletOutbox.Topic, realtimeWalletOutboxTopic: cfg.RocketMQ.RealtimeOutbox.Topic, + walletEventTagMode: cfg.RocketMQ.WalletEventTagMode, nodeID: cfg.NodeID, redPacketExpiryWorkerCfg: cfg.RedPacketExpiryWorker, }, nil diff --git a/services/wallet-service/internal/app/mq.go b/services/wallet-service/internal/app/mq.go index 140d3c68..de71e791 100644 --- a/services/wallet-service/internal/app/mq.go +++ b/services/wallet-service/internal/app/mq.go @@ -31,12 +31,22 @@ func newWalletProjectionConsumer(cfg config.Config, svc *walletservice.Service) if svc == nil || !cfg.ProjectionWorker.Enabled { return nil, nil } + // 钱包自有投影只关心礼物墙和徽章发放事实；兼容 selector 同时保留 + // legacy Tag，待旧消息排空后可在后续阶段只留下这三个 typed Tag。 + tagExpression, err := walletmq.LegacyCompatibleTagExpression( + walletmq.EventTypeWalletGiftDebited, + walletmq.EventTypeResourceGranted, + walletmq.EventTypeResourceGroupGranted, + ) + if err != nil { + return nil, err + } 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 { + if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, tagExpression, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { walletMessage, err := walletmq.DecodeWalletOutboxMessage(message.Body) if err != nil { return err diff --git a/services/wallet-service/internal/app/outbox_worker.go b/services/wallet-service/internal/app/outbox_worker.go index 40aa06da..4950c47c 100644 --- a/services/wallet-service/internal/app/outbox_worker.go +++ b/services/wallet-service/internal/app/outbox_worker.go @@ -94,10 +94,34 @@ func (a *App) processWalletOutboxBatch(ctx context.Context, workerID string, opt } func (a *App) publishWalletOutboxRecord(ctx context.Context, producer *rocketmqx.Producer, topic string, record mysqlstorage.WalletOutboxRecord) error { + message, err := walletOutboxRocketMessage(topic, a.walletEventTagMode, record) + if err != nil { + return err + } + return producer.SendSync(ctx, message) +} + +// walletOutboxRocketMessage keeps the envelope event_type canonical in both phases. +// Phase A deliberately publishes the legacy Tag; Phase B uses that same event_type as +// the Broker routing Tag, avoiding any derived category that could drift from handlers. +func walletOutboxRocketMessage(topic string, tagMode config.WalletEventTagMode, record mysqlstorage.WalletOutboxRecord) (rocketmqx.Message, error) { + eventTag, err := walletmq.EventTypeTag(record.EventType) + if err != nil { + return rocketmqx.Message{}, err + } + publishTag := walletmq.TagWalletOutboxEvent + switch tagMode { + case "", config.WalletEventTagModeLegacy: + // 空值属于旧配置或测试构造，必须保持 legacy；只有显式 event_type 才能进入 Phase B。 + case config.WalletEventTagModeEventType: + publishTag = eventTag + default: + return rocketmqx.Message{}, fmt.Errorf("unsupported wallet event tag mode %q", tagMode) + } body, err := walletmq.EncodeWalletOutboxMessage(walletmq.WalletOutboxMessage{ AppCode: record.AppCode, EventID: record.EventID, - EventType: record.EventType, + EventType: eventTag, TransactionID: record.TransactionID, CommandID: record.CommandID, UserID: record.UserID, @@ -108,14 +132,14 @@ func (a *App) publishWalletOutboxRecord(ctx context.Context, producer *rocketmqx OccurredAtMS: record.CreatedAtMS, }) if err != nil { - return err + return rocketmqx.Message{}, err } - return producer.SendSync(ctx, rocketmqx.Message{ + return rocketmqx.Message{ Topic: topic, - Tag: walletmq.TagWalletOutboxEvent, + Tag: publishTag, Keys: []string{record.EventID, record.TransactionID, record.CommandID}, Body: body, - }) + }, nil } func deadWalletOutboxReason(record mysqlstorage.WalletOutboxRecord) string { diff --git a/services/wallet-service/internal/app/outbox_worker_test.go b/services/wallet-service/internal/app/outbox_worker_test.go new file mode 100644 index 00000000..e6c3075e --- /dev/null +++ b/services/wallet-service/internal/app/outbox_worker_test.go @@ -0,0 +1,85 @@ +package app + +import ( + "testing" + + "hyapp/pkg/walletmq" + "hyapp/services/wallet-service/internal/config" + mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql" +) + +func TestWalletOutboxRocketMessageUsesConfiguredTagMode(t *testing.T) { + tests := []struct { + name string + mode config.WalletEventTagMode + wantTag string + }{ + { + name: "missing config remains legacy", + mode: "", + wantTag: walletmq.TagWalletOutboxEvent, + }, + { + name: "phase a publishes legacy", + mode: config.WalletEventTagModeLegacy, + wantTag: walletmq.TagWalletOutboxEvent, + }, + { + name: "phase b publishes event type", + mode: config.WalletEventTagModeEventType, + wantTag: walletmq.EventTypeWalletGiftDebited, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + message, err := walletOutboxRocketMessage("hyapp_wallet_outbox", test.mode, validWalletOutboxRecord()) + if err != nil { + t.Fatalf("walletOutboxRocketMessage failed: %v", err) + } + if message.Tag != test.wantTag { + t.Fatalf("message tag = %q, want %q", message.Tag, test.wantTag) + } + decoded, err := walletmq.DecodeWalletOutboxMessage(message.Body) + if err != nil { + t.Fatalf("DecodeWalletOutboxMessage failed: %v", err) + } + if decoded.EventType != walletmq.EventTypeWalletGiftDebited { + t.Fatalf("body event_type = %q", decoded.EventType) + } + if message.Topic != "hyapp_wallet_outbox" || len(message.Keys) != 3 { + t.Fatalf("message routing metadata mismatch: %+v", message) + } + }) + } +} + +func TestWalletOutboxRocketMessageRejectsUnsupportedTagMode(t *testing.T) { + _, err := walletOutboxRocketMessage("hyapp_wallet_outbox", config.WalletEventTagMode("all"), validWalletOutboxRecord()) + if err == nil { + t.Fatal("walletOutboxRocketMessage unexpectedly accepted an unsupported tag mode") + } +} + +func TestWalletOutboxRocketMessageRejectsUnsafeEventTypeTag(t *testing.T) { + record := validWalletOutboxRecord() + record.EventType = "WalletGiftDebited || *" + _, err := walletOutboxRocketMessage("hyapp_wallet_outbox", config.WalletEventTagModeLegacy, record) + if err == nil { + t.Fatal("walletOutboxRocketMessage unexpectedly accepted an unsafe event_type") + } +} + +func validWalletOutboxRecord() mysqlstorage.WalletOutboxRecord { + return mysqlstorage.WalletOutboxRecord{ + AppCode: "lalu", + EventID: "event-1", + EventType: " " + walletmq.EventTypeWalletGiftDebited + " ", + TransactionID: "transaction-1", + CommandID: "command-1", + UserID: 1001, + AssetType: "COIN", + AvailableDelta: -100, + PayloadJSON: `{"gift_id":"gift-1"}`, + CreatedAtMS: 1_700_000_000_000, + } +} diff --git a/services/wallet-service/internal/config/config.go b/services/wallet-service/internal/config/config.go index 3c3eed77..3a327aa2 100644 --- a/services/wallet-service/internal/config/config.go +++ b/services/wallet-service/internal/config/config.go @@ -66,20 +66,33 @@ type RedPacketExpiryWorkerConfig struct { // RocketMQConfig 描述 wallet-service 发布账务事实的 MQ 连接。 type RocketMQConfig struct { - Enabled bool `yaml:"enabled"` - NameServers []string `yaml:"name_servers"` - NameServerDomain string `yaml:"name_server_domain"` - AccessKey string `yaml:"access_key"` - SecretKey string `yaml:"secret_key"` - SecurityToken string `yaml:"security_token"` - Namespace string `yaml:"namespace"` - VIPChannel bool `yaml:"vip_channel"` - SendTimeout time.Duration `yaml:"send_timeout"` - Retry int `yaml:"retry"` - WalletOutbox WalletOutboxMQConfig `yaml:"wallet_outbox"` - RealtimeOutbox WalletOutboxMQConfig `yaml:"realtime_outbox"` + Enabled bool `yaml:"enabled"` + NameServers []string `yaml:"name_servers"` + NameServerDomain string `yaml:"name_server_domain"` + AccessKey string `yaml:"access_key"` + SecretKey string `yaml:"secret_key"` + SecurityToken string `yaml:"security_token"` + Namespace string `yaml:"namespace"` + VIPChannel bool `yaml:"vip_channel"` + SendTimeout time.Duration `yaml:"send_timeout"` + Retry int `yaml:"retry"` + // WalletEventTagMode 是 wallet topic producer 的两阶段发布开关： + // legacy 用统一 Tag 保证旧消费者可见，event_type 仅在所有消费者完成双订阅后启用。 + WalletEventTagMode WalletEventTagMode `yaml:"wallet_event_tag_mode"` + WalletOutbox WalletOutboxMQConfig `yaml:"wallet_outbox"` + RealtimeOutbox WalletOutboxMQConfig `yaml:"realtime_outbox"` } +// WalletEventTagMode 控制钱包事实的 Broker 路由 Tag，不改变消息体协议。 +type WalletEventTagMode string + +const ( + // WalletEventTagModeLegacy 是安全默认值；旧配置缺字段时不能因普通滚动发布提前切 typed Tag。 + WalletEventTagModeLegacy WalletEventTagMode = "legacy" + // WalletEventTagModeEventType 在 Phase B 使用消息 event_type 作为 Tag，让 Broker 精确过滤。 + WalletEventTagModeEventType WalletEventTagMode = "event_type" +) + // WalletOutboxMQConfig 保存 wallet_outbox topic 的生产者配置。 type WalletOutboxMQConfig struct { Enabled bool `yaml:"enabled"` @@ -367,9 +380,10 @@ func Default() Config { func defaultRocketMQConfig() RocketMQConfig { return RocketMQConfig{ - Enabled: false, - SendTimeout: 3 * time.Second, - Retry: 2, + Enabled: false, + SendTimeout: 3 * time.Second, + Retry: 2, + WalletEventTagMode: WalletEventTagModeLegacy, WalletOutbox: WalletOutboxMQConfig{ Enabled: false, Topic: "hyapp_wallet_outbox", @@ -769,6 +783,16 @@ func normalizeRocketMQConfig(cfg RocketMQConfig) (RocketMQConfig, error) { if cfg.Retry <= 0 { cfg.Retry = defaults.Retry } + cfg.WalletEventTagMode = WalletEventTagMode(strings.ToLower(strings.TrimSpace(string(cfg.WalletEventTagMode)))) + if cfg.WalletEventTagMode == "" { + // 配置升级必须显式进入 Phase B；历史线上 YAML 缺少新字段时始终继续发布 legacy Tag。 + cfg.WalletEventTagMode = defaults.WalletEventTagMode + } + switch cfg.WalletEventTagMode { + case WalletEventTagModeLegacy, WalletEventTagModeEventType: + default: + return RocketMQConfig{}, errors.New("rocketmq.wallet_event_tag_mode must be legacy or event_type") + } if cfg.WalletOutbox.Topic = strings.TrimSpace(cfg.WalletOutbox.Topic); cfg.WalletOutbox.Topic == "" { cfg.WalletOutbox.Topic = defaults.WalletOutbox.Topic } diff --git a/services/wallet-service/internal/config/config_test.go b/services/wallet-service/internal/config/config_test.go index bf95fb3a..26448579 100644 --- a/services/wallet-service/internal/config/config_test.go +++ b/services/wallet-service/internal/config/config_test.go @@ -4,6 +4,8 @@ import ( "crypto/rsa" "crypto/x509" "encoding/base64" + "os" + "path/filepath" "strings" "testing" ) @@ -17,6 +19,9 @@ func TestLoadLocalEnablesOutboxMQ(t *testing.T) { if !cfg.RocketMQ.Enabled || !cfg.RocketMQ.WalletOutbox.Enabled || !cfg.RocketMQ.RealtimeOutbox.Enabled { t.Fatalf("local config must enable wallet outbox MQ producers: %+v", cfg.RocketMQ) } + if cfg.RocketMQ.WalletEventTagMode != WalletEventTagModeLegacy { + t.Fatalf("Phase A config must publish legacy wallet tags: %q", cfg.RocketMQ.WalletEventTagMode) + } if !cfg.OutboxWorker.Enabled || !cfg.RealtimeOutboxWorker.Enabled || !cfg.ProjectionWorker.Enabled { t.Fatalf("local config must enable wallet outbox workers: outbox=%+v realtime=%+v projection=%+v", cfg.OutboxWorker, cfg.RealtimeOutboxWorker, cfg.ProjectionWorker) } @@ -31,6 +36,50 @@ func TestLoadLocalEnablesOutboxMQ(t *testing.T) { } } +func TestDefaultWalletEventTagModeIsLegacy(t *testing.T) { + if got := Default().RocketMQ.WalletEventTagMode; got != WalletEventTagModeLegacy { + t.Fatalf("default wallet event tag mode = %q, want %q", got, WalletEventTagModeLegacy) + } +} + +func TestWalletEventTagModeMissingConfigDefaultsToLegacy(t *testing.T) { + path := writeWalletConfig(t, "rocketmq:\n enabled: false\n") + cfg, err := Load(path) + if err != nil { + t.Fatalf("Load missing tag mode config failed: %v", err) + } + if cfg.RocketMQ.WalletEventTagMode != WalletEventTagModeLegacy { + t.Fatalf("missing wallet_event_tag_mode = %q, want %q", cfg.RocketMQ.WalletEventTagMode, WalletEventTagModeLegacy) + } +} + +func TestWalletEventTagModeParsesExplicitPhaseB(t *testing.T) { + path := writeWalletConfig(t, "rocketmq:\n enabled: false\n wallet_event_tag_mode: ' EVENT_TYPE '\n") + cfg, err := Load(path) + if err != nil { + t.Fatalf("Load Phase B tag mode config failed: %v", err) + } + if cfg.RocketMQ.WalletEventTagMode != WalletEventTagModeEventType { + t.Fatalf("wallet_event_tag_mode = %q, want %q", cfg.RocketMQ.WalletEventTagMode, WalletEventTagModeEventType) + } +} + +func TestWalletEventTagModeRejectsUnknownValue(t *testing.T) { + path := writeWalletConfig(t, "rocketmq:\n enabled: false\n wallet_event_tag_mode: wildcard\n") + if _, err := Load(path); err == nil || !strings.Contains(err.Error(), "wallet_event_tag_mode") { + t.Fatalf("Load unknown wallet_event_tag_mode error = %v", err) + } +} + +func writeWalletConfig(t *testing.T, body string) string { + t.Helper() + path := filepath.Join(t.TempDir(), "config.yaml") + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + t.Fatalf("write config: %v", err) + } + return path +} + func TestLoadTencentExampleSplitsRealtimeOutboxAndRaisesThroughput(t *testing.T) { cfg, err := Load("../../configs/config.tencent.example.yaml") if err != nil { diff --git a/services/wallet-service/internal/service/wallet/projection.go b/services/wallet-service/internal/service/wallet/projection.go index 574900b1..234a153a 100644 --- a/services/wallet-service/internal/service/wallet/projection.go +++ b/services/wallet-service/internal/service/wallet/projection.go @@ -129,7 +129,7 @@ func isBadgeGrantWalletProjectionMessage(message walletmq.WalletOutboxMessage) b return false } switch message.EventType { - case "ResourceGranted", "ResourceGroupGranted": + case walletmq.EventTypeResourceGranted, walletmq.EventTypeResourceGroupGranted: return true default: return false 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 af4477ec..7aeef24f 100644 --- a/services/wallet-service/internal/storage/mysql/badge_projection_relay.go +++ b/services/wallet-service/internal/storage/mysql/badge_projection_relay.go @@ -17,8 +17,8 @@ import ( const ( badgeDisplayProjectionName = "activity_badge_display" - walletResourceGrantedEvent = "ResourceGranted" - walletResourceGroupGrantedEvent = "ResourceGroupGranted" + walletResourceGrantedEvent = walletmq.EventTypeResourceGranted + walletResourceGroupGrantedEvent = walletmq.EventTypeResourceGroupGranted grantSourceVIPPurchase = "vip_purchase" ) 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 d69f4c62..3966ee9e 100644 --- a/services/wallet-service/internal/storage/mysql/gift_wall_projection.go +++ b/services/wallet-service/internal/storage/mysql/gift_wall_projection.go @@ -16,7 +16,7 @@ import ( const ( giftWallProjectionName = "user_gift_wall" - walletGiftDebitedEvent = "WalletGiftDebited" + walletGiftDebitedEvent = walletmq.EventTypeWalletGiftDebited // projection 状态机只描述“某个 outbox 事件是否已经投影进礼物墙”。 // 这里不复用 wallet_outbox.status，避免展示读模型反向污染账务事件分发语义。 diff --git a/services/wallet-service/internal/storage/mysql/resource_outbox.go b/services/wallet-service/internal/storage/mysql/resource_outbox.go index fbf3260f..785e1cb6 100644 --- a/services/wallet-service/internal/storage/mysql/resource_outbox.go +++ b/services/wallet-service/internal/storage/mysql/resource_outbox.go @@ -8,10 +8,14 @@ import ( func resourceOutboxEvent(eventType string, commandID string, userID int64, resourceID int64, payload any, nowMs int64) walletOutboxEvent { eventKey := fmt.Sprintf("%s|%s|%d|%d", eventType, commandID, userID, resourceID) + eventID := "wev_" + stableHash(eventKey) return walletOutboxEvent{ - EventID: "wev_" + stableHash(eventKey), - EventType: eventType, - TransactionID: "", + EventID: eventID, + EventType: eventType, + // 资源目录、赠送和装备变更没有金融 ledger transaction；使用稳定 event_id + // 作为 correlation transaction_id，既满足统一 MQ envelope，又能从消息直接 + // 回查唯一 outbox 行。相同业务命令重试会得到同一 ID，不会制造新事实。 + TransactionID: eventID, CommandID: commandID, UserID: userID, AssetType: resourceOutboxAsset, diff --git a/services/wallet-service/internal/storage/mysql/resource_outbox_test.go b/services/wallet-service/internal/storage/mysql/resource_outbox_test.go new file mode 100644 index 00000000..b269d805 --- /dev/null +++ b/services/wallet-service/internal/storage/mysql/resource_outbox_test.go @@ -0,0 +1,98 @@ +package mysql + +import ( + "os" + "path/filepath" + "runtime" + "strings" + "testing" + + "hyapp/pkg/walletmq" +) + +func TestResourceOutboxEventsUseStableEventIDAsTransactionID(t *testing.T) { + // 这些是当前所有通过 resourceOutboxEvent 创建的资源事实；helper 对事件类型 + // 本身无分支，因此该约束也自动覆盖以后接入同一入口的新资源事实。 + eventTypes := []string{ + "GiftConfigChanged", + "ResourceChanged", + "ResourceGrantRevoked", + "ResourceGranted", + "ResourceGroupChanged", + "ResourceGroupGranted", + "UserResourceChanged", + "VipEffectiveChanged", + "VipTrialCardGranted", + } + for _, eventType := range eventTypes { + t.Run(eventType, func(t *testing.T) { + first := resourceOutboxEvent(eventType, "command-1", 1001, 2001, map[string]any{"source": "test"}, 1_700_000_000_000) + second := resourceOutboxEvent(eventType, "command-1", 1001, 2001, map[string]any{"source": "retry"}, 1_700_000_000_999) + if first.TransactionID == "" || first.TransactionID != first.EventID { + t.Fatalf("resource correlation ids mismatch: event_id=%q transaction_id=%q", first.EventID, first.TransactionID) + } + if second.EventID != first.EventID || second.TransactionID != first.TransactionID { + t.Fatalf("resource correlation id changed on retry: first=%+v second=%+v", first, second) + } + if len(first.TransactionID) > 96 { + t.Fatalf("resource transaction_id exceeds wallet_outbox column: %d", len(first.TransactionID)) + } + if _, err := walletmq.EncodeWalletOutboxMessage(walletmq.WalletOutboxMessage{ + AppCode: "lalu", + EventID: first.EventID, + EventType: first.EventType, + TransactionID: first.TransactionID, + CommandID: first.CommandID, + UserID: first.UserID, + AssetType: first.AssetType, + PayloadJSON: `{"source":"test"}`, + OccurredAtMS: first.CreatedAtMS, + }); err != nil { + t.Fatalf("resource event is still rejected by wallet MQ encoder: %v", err) + } + }) + } +} + +func TestResourceOutboxHistoricalRetryIsOptInAndNarrowlyScoped(t *testing.T) { + _, file, _, ok := runtime.Caller(0) + if !ok { + t.Fatal("resolve test source path") + } + path := filepath.Join(filepath.Dir(file), "../../../deploy/mysql/maintenance/20260711_retry_incomplete_resource_outbox.sql") + body, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read resource outbox maintenance SQL: %v", err) + } + sql := string(body) + for _, required := range []string{ + "COALESCE(@execute_wallet_resource_outbox_retry, 0)", + "WHERE @execute_wallet_resource_outbox_retry = 1", + "AND wo.asset_type = 'RESOURCE'", + "AND wo.transaction_id = ''", + "AND BINARY wo.last_error = BINARY 'wallet outbox message is incomplete'", + "AND wo.status IN ('failed', 'retryable')", + "LIMIT 100", + "SET wo.transaction_id = wo.event_id", + "wo.status = 'pending'", + } { + if !strings.Contains(sql, required) { + t.Fatalf("maintenance SQL misses safety contract %q", required) + } + } + for _, eventType := range []string{ + "GiftConfigChanged", + "ResourceChanged", + "ResourceGrantRevoked", + "ResourceGranted", + "ResourceGroupChanged", + "ResourceGroupGranted", + "UserResourceChanged", + "VipEffectiveChanged", + "VipTrialCardGranted", + } { + if !strings.Contains(sql, "'"+eventType+"'") { + t.Fatalf("maintenance SQL misses resource event type %q", eventType) + } + } +}