固化机器人钱包通知抑制快照
This commit is contained in:
zhx
parent
5db48e1bcc
commit
175ddbea7e
@ -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 {
|
||||
|
||||
@ -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{{
|
||||
|
||||
@ -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
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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)
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user