diff --git a/services/notice-service/internal/modules/walletnotice/service.go b/services/notice-service/internal/modules/walletnotice/service.go index c0647eb8..2dd1d8a8 100644 --- a/services/notice-service/internal/modules/walletnotice/service.go +++ b/services/notice-service/internal/modules/walletnotice/service.go @@ -59,6 +59,15 @@ func (s *Service) ProcessWalletBalanceNotices(ctx context.Context, options Walle } processed := 0 for _, event := range events { + if walletPrivateNoticeSuppressed(event.AssetType, event.PayloadJSON) { + // owner 事件已经固化“不投私信”时，把本地 delivery 位点收敛为 delivered； + // 这条终态既阻止 pull fallback 反复认领，也不需要 notice 反查 user-service 猜账号类型。 + if err := s.repository.MarkWalletBalanceDelivered(ctx, event, json.RawMessage(event.PayloadJSON), time.Now().UnixMilli()); err != nil { + return processed, err + } + processed++ + continue + } if err := s.publishWalletBalanceEvent(ctx, event, options); err != nil { return processed, err } @@ -73,8 +82,9 @@ func (s *Service) ProcessWalletOutboxMessage(ctx context.Context, message wallet if !isWalletPrivateNoticeEvent(message.EventType) { return false, nil } - if isRobotCoinWalletNotice(message.AssetType) { - // ROBOT_COIN 只服务机器人房间展示扣费，机器人账号不需要也不能接收用户 C2C 余额通知。 + if walletPrivateNoticeSuppressed(message.AssetType, message.PayloadJSON) { + // MQ 主链路在写 notice delivery 前确认 owner 快照；ROBOT_COIN 是历史兼容条件， + // suppress_private_notice 才是不依赖资产类型的显式契约。 return true, nil } options = normalizeWalletNoticeWorkerOptions(options, s.cfg.NodeID) @@ -170,6 +180,21 @@ func isRobotCoinWalletNotice(assetType string) bool { return assetType == robotCoinAssetType } +func walletPrivateNoticeSuppressed(assetType string, payloadJSON string) bool { + if isRobotCoinWalletNotice(assetType) { + return true + } + var control struct { + SuppressPrivateNotice bool `json:"suppress_private_notice"` + } + // 旧事件没有控制字段时保持原投递语义；非法 payload 也不能在这里被静默确认， + // 后续正常 payload 校验会把它收敛到 retry/failed，保留可观测性。 + if err := json.Unmarshal([]byte(payloadJSON), &control); err != nil { + return false + } + return control.SuppressPrivateNotice +} + func walletBalanceNoticePayload(event WalletBalanceEvent) (json.RawMessage, error) { var payload map[string]any if err := json.Unmarshal([]byte(event.PayloadJSON), &payload); err != nil { diff --git a/services/notice-service/internal/modules/walletnotice/service_test.go b/services/notice-service/internal/modules/walletnotice/service_test.go index 73425bc7..91d6ac25 100644 --- a/services/notice-service/internal/modules/walletnotice/service_test.go +++ b/services/notice-service/internal/modules/walletnotice/service_test.go @@ -140,6 +140,56 @@ func TestProcessWalletOutboxMessageSkipsRobotCoinPrivateIM(t *testing.T) { } } +func TestProcessWalletOutboxMessageSkipsOwnerSuppressedPrivateIM(t *testing.T) { + repo := &fakeRepository{} + publisher := &fakePublisher{} + service := New(Config{NodeID: "node-a"}, repo, publisher) + + handled, err := service.ProcessWalletOutboxMessage(context.Background(), walletmq.WalletOutboxMessage{ + AppCode: "huwaa", + EventID: "suppressed-wallet-1", + EventType: "WalletBalanceChanged", + TransactionID: "suppressed-tx-1", + CommandID: "suppressed-cmd-1", + UserID: 100002, + AssetType: "COIN", + AvailableDelta: -1, + PayloadJSON: `{"available_after":999,"suppress_private_notice":true}`, + OccurredAtMS: 1710000000000, + }, WalletNoticeWorkerOptions{MaxRetryCount: 3}) + if err != nil { + t.Fatalf("ProcessWalletOutboxMessage failed: %v", err) + } + if !handled || len(repo.claimedMessages) != 0 || len(repo.delivered) != 0 || len(publisher.messages) != 0 { + t.Fatalf("owner-suppressed notice must be acked before delivery claim: handled=%v claimed=%d delivered=%d messages=%d", handled, len(repo.claimedMessages), len(repo.delivered), len(publisher.messages)) + } +} + +func TestProcessWalletBalanceNoticesTerminatesOwnerSuppressedFallback(t *testing.T) { + repo := &fakeRepository{events: []WalletBalanceEvent{{ + AppCode: "huwaa", + EventID: "suppressed-wallet-fallback-1", + EventType: "WalletBalanceChanged", + TransactionID: "suppressed-tx-fallback-1", + CommandID: "suppressed-cmd-fallback-1", + UserID: 100003, + AssetType: "COIN", + AvailableDelta: -1, + PayloadJSON: `{"available_after":999,"suppress_private_notice":true}`, + CreatedAtMS: 1710000000000, + }}} + publisher := &fakePublisher{} + service := New(Config{NodeID: "node-a"}, repo, publisher) + + processed, err := service.ProcessWalletBalanceNotices(context.Background(), WalletNoticeWorkerOptions{MaxRetryCount: 3}) + if err != nil { + t.Fatalf("ProcessWalletBalanceNotices failed: %v", err) + } + if processed != 1 || len(repo.delivered) != 1 || len(publisher.messages) != 0 { + t.Fatalf("fallback suppression must terminate locally without C2C: processed=%d delivered=%d messages=%d", processed, len(repo.delivered), len(publisher.messages)) + } +} + func TestProcessWalletBalanceNoticesMarksRetryable(t *testing.T) { repo := &fakeRepository{ events: []WalletBalanceEvent{{ diff --git a/services/wallet-service/internal/storage/mysql/outbox.go b/services/wallet-service/internal/storage/mysql/outbox.go index 87d9c79f..2bdfe80b 100644 --- a/services/wallet-service/internal/storage/mysql/outbox.go +++ b/services/wallet-service/internal/storage/mysql/outbox.go @@ -373,6 +373,25 @@ func balanceChangedEvent(transactionID string, commandID string, userID int64, a if len(bizTypes) > 0 { bizType = strings.TrimSpace(bizTypes[0]) } + eventPayload := map[string]any{ + "transaction_id": transactionID, + "command_id": commandID, + "user_id": userID, + "asset_type": assetType, + "biz_type": bizType, + "available_delta": availableDelta, + "frozen_delta": frozenDelta, + "available_after": availableAfter, + "frozen_after": frozenAfter, + "balance_version": balanceVersion, + "metadata": payload, + "created_at_ms": nowMs, + } + if suppressWalletPrivateNotice(assetType, payload) { + // 抑制决策必须和账务事实一起固化，notice-service 只能消费这份不可变快照， + // 不能跨库读取用户 source 或根据当前资料反推交易发生时的账号语义。 + eventPayload["suppress_private_notice"] = true + } return walletOutboxEvent{ EventID: eventID(transactionID, "WalletBalanceChanged", userID, assetType), EventType: "WalletBalanceChanged", @@ -382,21 +401,24 @@ func balanceChangedEvent(transactionID string, commandID string, userID int64, a AssetType: assetType, AvailableDelta: availableDelta, FrozenDelta: frozenDelta, - Payload: map[string]any{ - "transaction_id": transactionID, - "command_id": commandID, - "user_id": userID, - "asset_type": assetType, - "biz_type": bizType, - "available_delta": availableDelta, - "frozen_delta": frozenDelta, - "available_after": availableAfter, - "frozen_after": frozenAfter, - "balance_version": balanceVersion, - "metadata": payload, - "created_at_ms": nowMs, - }, - CreatedAtMS: nowMs, + Payload: eventPayload, + CreatedAtMS: nowMs, + } +} + +func suppressWalletPrivateNotice(assetType string, payload any) bool { + if strings.EqualFold(strings.TrimSpace(assetType), ledger.AssetRobotCoin) { + return true + } + // giftMetadata.RobotGift 是 wallet 扣费时的 owner 快照；只认可强类型业务元数据， + // 不在通用 JSON/map 里猎猜 source、command_id 或 biz_type。 + switch metadata := payload.(type) { + case giftMetadata: + return metadata.RobotGift + case *giftMetadata: + return metadata != nil && metadata.RobotGift + default: + return false } } diff --git a/services/wallet-service/internal/storage/mysql/outbox_private_notice_test.go b/services/wallet-service/internal/storage/mysql/outbox_private_notice_test.go new file mode 100644 index 00000000..195fc235 --- /dev/null +++ b/services/wallet-service/internal/storage/mysql/outbox_private_notice_test.go @@ -0,0 +1,69 @@ +package mysql + +import ( + "testing" + + "hyapp/services/wallet-service/internal/domain/ledger" +) + +func TestBalanceChangedEventPersistsPrivateNoticeSuppressionSnapshot(t *testing.T) { + event := balanceChangedEvent( + "tx-robot-1", + "cmd-robot-1", + 100001, + // COIN 刻意不依赖 ROBOT_COIN 兼容分支，证明强类型 giftMetadata.RobotGift 本身能固化决策。 + ledger.AssetCoin, + -10, + 0, + 90, + 0, + 2, + giftMetadata{RobotGift: true}, + 1710000000000, + bizTypeRobotGiftDebit, + ) + payload, ok := event.Payload.(map[string]any) + if !ok { + t.Fatalf("payload type = %T, want map[string]any", event.Payload) + } + if suppressed, ok := payload["suppress_private_notice"].(bool); !ok || !suppressed { + t.Fatalf("suppress_private_notice = %#v, want true", payload["suppress_private_notice"]) + } + legacyRobotAsset := balanceChangedEvent( + "tx-robot-asset-1", + "cmd-robot-asset-1", + 100001, + ledger.AssetRobotCoin, + -10, + 0, + 90, + 0, + 2, + struct{}{}, + 1710000000000, + bizTypeRobotGiftDebit, + ) + legacyRobotPayload := legacyRobotAsset.Payload.(map[string]any) + if suppressed, ok := legacyRobotPayload["suppress_private_notice"].(bool); !ok || !suppressed { + t.Fatalf("ROBOT_COIN compatibility suppression = %#v, want true", legacyRobotPayload["suppress_private_notice"]) + } + + human := balanceChangedEvent( + "tx-human-1", + "cmd-human-1", + 100002, + ledger.AssetCoin, + -10, + 0, + 90, + 0, + 2, + giftMetadata{}, + 1710000000000, + bizTypeGiftDebit, + ) + humanPayload := human.Payload.(map[string]any) + if _, exists := humanPayload["suppress_private_notice"]; exists { + t.Fatalf("human wallet event must not persist suppression control: %#v", humanPayload) + } +}