From 3666a9ff02723fa6ec1812894c99879b09cf35fd Mon Sep 17 00:00:00 2001 From: zhx Date: Sat, 11 Jul 2026 02:55:31 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E9=81=BF=E5=85=8D=20VIP=20=E9=80=9A?= =?UTF-8?q?=E7=9F=A5=E9=87=8D=E6=94=BE=E5=8E=86=E5=8F=B2=E9=92=B1=E5=8C=85?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- services/activity-service/internal/app/mq.go | 8 +++----- services/activity-service/internal/app/mq_test.go | 13 +++++++++++++ 2 files changed, 16 insertions(+), 5 deletions(-) diff --git a/services/activity-service/internal/app/mq.go b/services/activity-service/internal/app/mq.go index f5ac8e09..b31077ad 100644 --- a/services/activity-service/internal/app/mq.go +++ b/services/activity-service/internal/app/mq.go @@ -409,11 +409,9 @@ func walletOutboxUserLeaderboardConsumerConfig(cfg config.RocketMQConfig) rocket } func walletOutboxVIPRebateNoticeConsumerConfig(cfg config.RocketMQConfig) rocketmqx.ConsumerConfig { - consumerConfig := walletOutboxConsumerConfig(cfg, cfg.WalletOutbox.VIPRebateNoticeConsumerGroup) - // 新 consumer group 上线时必须从 broker 仍保留的最早事实恢复；否则 wallet/cron 先发布的 - // 当日资格会永久没有系统通知。非目标钱包事实会被解析器安全跳过，不会写 inbox。 - consumerConfig.ConsumeFromFirst = true - return consumerConfig + // VIP 返币通知会写用户 inbox，属于不可批量重放的业务副作用。新消费组必须沿用 SDK 的 + // latest-offset 默认值；上线窗口内需要补偿的少量资格由明确的位点重置处理，不能扫描全部历史钱包事实。 + return walletOutboxConsumerConfig(cfg, cfg.WalletOutbox.VIPRebateNoticeConsumerGroup) } func messageActionProducerConfig(cfg config.RocketMQConfig) rocketmqx.ProducerConfig { diff --git a/services/activity-service/internal/app/mq_test.go b/services/activity-service/internal/app/mq_test.go index 6be07b57..4131bc61 100644 --- a/services/activity-service/internal/app/mq_test.go +++ b/services/activity-service/internal/app/mq_test.go @@ -18,3 +18,16 @@ func TestRoomOutboxConsumerConfigSerializesCompositeProjection(t *testing.T) { t.Fatalf("room outbox group = %q, want %q", consumerConfig.GroupName, cfg.RoomOutbox.ConsumerGroup) } } + +func TestVIPRebateNoticeConsumerDoesNotReplayHistoricalWalletEvents(t *testing.T) { + cfg := config.Default().RocketMQ + cfg.WalletOutbox.VIPRebateNoticeConsumerGroup = "activity-vip-rebate-notice-test" + + consumerConfig := walletOutboxVIPRebateNoticeConsumerConfig(cfg) + if consumerConfig.ConsumeFromFirst { + t.Fatal("VIP rebate notice consumer must start at the latest offset for a new group") + } + if consumerConfig.GroupName != cfg.WalletOutbox.VIPRebateNoticeConsumerGroup { + t.Fatalf("VIP rebate notice group = %q, want %q", consumerConfig.GroupName, cfg.WalletOutbox.VIPRebateNoticeConsumerGroup) + } +}