71 lines
2.4 KiB
Go
71 lines
2.4 KiB
Go
// Package walletmq defines RocketMQ payloads owned by wallet-service.
|
|
package walletmq
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"strings"
|
|
)
|
|
|
|
const (
|
|
MessageTypeWalletOutboxEvent = "wallet_outbox_event"
|
|
|
|
TagWalletOutboxEvent = "wallet_outbox_event"
|
|
)
|
|
|
|
// WalletOutboxMessage is the MQ representation of one committed wallet_outbox fact.
|
|
type WalletOutboxMessage struct {
|
|
MessageType string `json:"message_type"`
|
|
AppCode string `json:"app_code"`
|
|
EventID string `json:"event_id"`
|
|
EventType string `json:"event_type"`
|
|
TransactionID string `json:"transaction_id"`
|
|
CommandID string `json:"command_id"`
|
|
UserID int64 `json:"user_id"`
|
|
AssetType string `json:"asset_type"`
|
|
AvailableDelta int64 `json:"available_delta"`
|
|
FrozenDelta int64 `json:"frozen_delta"`
|
|
PayloadJSON string `json:"payload_json"`
|
|
OccurredAtMS int64 `json:"occurred_at_ms"`
|
|
}
|
|
|
|
// EncodeWalletOutboxMessage serializes a wallet fact for fanout through RocketMQ.
|
|
func EncodeWalletOutboxMessage(message WalletOutboxMessage) ([]byte, error) {
|
|
message.MessageType = MessageTypeWalletOutboxEvent
|
|
if strings.TrimSpace(message.AppCode) == "" ||
|
|
strings.TrimSpace(message.EventID) == "" ||
|
|
strings.TrimSpace(message.EventType) == "" ||
|
|
strings.TrimSpace(message.TransactionID) == "" ||
|
|
strings.TrimSpace(message.CommandID) == "" ||
|
|
message.OccurredAtMS <= 0 {
|
|
return nil, errors.New("wallet outbox message is incomplete")
|
|
}
|
|
if strings.TrimSpace(message.PayloadJSON) == "" {
|
|
message.PayloadJSON = "{}"
|
|
}
|
|
return json.Marshal(message)
|
|
}
|
|
|
|
// DecodeWalletOutboxMessage validates the MQ body and restores the wallet fact.
|
|
func DecodeWalletOutboxMessage(body []byte) (WalletOutboxMessage, error) {
|
|
var message WalletOutboxMessage
|
|
if err := json.Unmarshal(body, &message); err != nil {
|
|
return WalletOutboxMessage{}, err
|
|
}
|
|
if message.MessageType != MessageTypeWalletOutboxEvent {
|
|
return WalletOutboxMessage{}, errors.New("unexpected wallet outbox message_type")
|
|
}
|
|
if strings.TrimSpace(message.AppCode) == "" ||
|
|
strings.TrimSpace(message.EventID) == "" ||
|
|
strings.TrimSpace(message.EventType) == "" ||
|
|
strings.TrimSpace(message.TransactionID) == "" ||
|
|
strings.TrimSpace(message.CommandID) == "" ||
|
|
message.OccurredAtMS <= 0 {
|
|
return WalletOutboxMessage{}, errors.New("wallet outbox message is incomplete")
|
|
}
|
|
if strings.TrimSpace(message.PayloadJSON) == "" {
|
|
message.PayloadJSON = "{}"
|
|
}
|
|
return message, nil
|
|
}
|