diff --git a/services/wallet-service/internal/app/outbox_worker.go b/services/wallet-service/internal/app/outbox_worker.go index da4893f4..408ddeec 100644 --- a/services/wallet-service/internal/app/outbox_worker.go +++ b/services/wallet-service/internal/app/outbox_worker.go @@ -151,11 +151,19 @@ func walletOutboxRocketMessage(topic string, tagMode config.WalletEventTagMode, default: return rocketmqx.Message{}, fmt.Errorf("unsupported wallet event tag mode %q", tagMode) } + transactionID := strings.TrimSpace(record.TransactionID) + if transactionID == "" && isHistoricalResourceOutboxRecord(record) { + // 2026-07-11 以前的资源事实没有金融 transaction_id;event_id 已由 + // event_type+command_id+user_id+resource_id 稳定派生。发布时使用它作为 + // correlation id 与当前写入逻辑一致,避免为了回放改写已提交的 payload, + // 同时严格限制为 RESOURCE 和已知资源事件,绝不替金融事实补造交易号。 + transactionID = strings.TrimSpace(record.EventID) + } body, err := walletmq.EncodeWalletOutboxMessage(walletmq.WalletOutboxMessage{ AppCode: record.AppCode, EventID: record.EventID, EventType: eventTag, - TransactionID: record.TransactionID, + TransactionID: transactionID, CommandID: record.CommandID, UserID: record.UserID, AssetType: record.AssetType, @@ -170,11 +178,31 @@ func walletOutboxRocketMessage(topic string, tagMode config.WalletEventTagMode, return rocketmqx.Message{ Topic: topic, Tag: publishTag, - Keys: []string{record.EventID, record.TransactionID, record.CommandID}, + Keys: []string{record.EventID, transactionID, record.CommandID}, Body: body, }, nil } +func isHistoricalResourceOutboxRecord(record mysqlstorage.WalletOutboxRecord) bool { + if strings.TrimSpace(record.EventID) == "" || !strings.EqualFold(strings.TrimSpace(record.AssetType), "RESOURCE") { + return false + } + switch strings.TrimSpace(record.EventType) { + case "GiftConfigChanged", + "ResourceChanged", + "ResourceGrantRevoked", + "ResourceGranted", + "ResourceGroupChanged", + "ResourceGroupGranted", + "UserResourceChanged", + "VipEffectiveChanged", + "VipTrialCardGranted": + return true + default: + return false + } +} + func deadWalletOutboxReason(record mysqlstorage.WalletOutboxRecord) string { last := strings.TrimSpace(record.LastError) if last == "" { diff --git a/services/wallet-service/internal/app/outbox_worker_test.go b/services/wallet-service/internal/app/outbox_worker_test.go index 1e6cae45..e32c5403 100644 --- a/services/wallet-service/internal/app/outbox_worker_test.go +++ b/services/wallet-service/internal/app/outbox_worker_test.go @@ -79,6 +79,53 @@ func TestWalletOutboxRocketMessageRejectsUnsafeEventTypeTag(t *testing.T) { } } +func TestWalletOutboxRocketMessageRestoresHistoricalResourceCorrelationID(t *testing.T) { + for _, eventType := range []string{ + "GiftConfigChanged", + "ResourceChanged", + "ResourceGrantRevoked", + "ResourceGranted", + "ResourceGroupChanged", + "ResourceGroupGranted", + "UserResourceChanged", + "VipEffectiveChanged", + "VipTrialCardGranted", + } { + t.Run(eventType, func(t *testing.T) { + record := validWalletOutboxRecord() + record.EventID = "wev-historical-resource" + record.EventType = eventType + record.TransactionID = "" + record.AssetType = "RESOURCE" + + message, err := walletOutboxRocketMessage("hyapp_wallet_outbox", config.WalletEventTagModeEventType, record) + if err != nil { + t.Fatalf("historical resource fact was rejected: %v", err) + } + decoded, err := walletmq.DecodeWalletOutboxMessage(message.Body) + if err != nil { + t.Fatalf("DecodeWalletOutboxMessage failed: %v", err) + } + if decoded.TransactionID != record.EventID || message.Keys[1] != record.EventID { + t.Fatalf("correlation id mismatch: decoded=%q keys=%v", decoded.TransactionID, message.Keys) + } + }) + } +} + +func TestWalletOutboxRocketMessageDoesNotInventFinancialTransactionID(t *testing.T) { + record := validWalletOutboxRecord() + record.TransactionID = "" + if _, err := walletOutboxRocketMessage("hyapp_wallet_outbox", config.WalletEventTagModeEventType, record); err == nil { + t.Fatal("financial fact without transaction_id must remain invalid") + } + record.EventType = "UnknownResourceEvent" + record.AssetType = "RESOURCE" + if _, err := walletOutboxRocketMessage("hyapp_wallet_outbox", config.WalletEventTagModeEventType, record); err == nil { + t.Fatal("unknown resource fact must not receive a synthetic transaction_id") + } +} + func validWalletOutboxRecord() mysqlstorage.WalletOutboxRecord { return mysqlstorage.WalletOutboxRecord{ AppCode: "lalu",