2055 lines
73 KiB
Go
2055 lines
73 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"database/sql"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"math"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
"hyapp/services/wallet-service/internal/domain/ledger"
|
||
|
||
mysqlDriver "github.com/go-sql-driver/mysql"
|
||
)
|
||
|
||
const (
|
||
transactionStatusSucceeded = "succeeded"
|
||
bizTypeGiftDebit = "gift_debit"
|
||
bizTypeManualCredit = "manual_credit"
|
||
bizTypeTaskReward = "task_reward"
|
||
bizTypeCoinSellerTransfer = "coin_seller_transfer"
|
||
bizTypeCoinSellerStockPurchase = "coin_seller_stock_purchase"
|
||
bizTypeCoinSellerCoinCompensation = "coin_seller_coin_compensation"
|
||
bizTypeGooglePlayRecharge = "google_play_recharge"
|
||
bizTypeGameDebit = "game_debit"
|
||
bizTypeGameCredit = "game_credit"
|
||
bizTypeGameRefund = "game_refund"
|
||
bizTypeGameReverse = "game_reverse"
|
||
outboxStatusPending = "pending"
|
||
outboxStatusDelivering = "delivering"
|
||
outboxStatusDelivered = "delivered"
|
||
outboxStatusRetryable = "retryable"
|
||
outboxStatusFailed = "failed"
|
||
giftWallChargeAssetMixed = "MIXED"
|
||
)
|
||
|
||
// Repository 是 wallet-service 的 MySQL v2 账本入口。
|
||
type Repository struct {
|
||
// db 是账户、交易、分录和 outbox 的同一事务边界。
|
||
db *sql.DB
|
||
}
|
||
|
||
// WalletOutboxRecord is a committed wallet fact ready for external fanout.
|
||
type WalletOutboxRecord struct {
|
||
AppCode string
|
||
EventID string
|
||
EventType string
|
||
TransactionID string
|
||
CommandID string
|
||
UserID int64
|
||
AssetType string
|
||
AvailableDelta int64
|
||
FrozenDelta int64
|
||
PayloadJSON string
|
||
CreatedAtMS int64
|
||
RetryCount int
|
||
LastError string
|
||
WorkerID string
|
||
LockUntilMS int64
|
||
}
|
||
|
||
// Open 创建 MySQL 连接池并做启动 ping。
|
||
func Open(ctx context.Context, dsn string) (*Repository, error) {
|
||
if strings.TrimSpace(dsn) == "" {
|
||
return nil, xerr.New(xerr.InvalidArgument, "mysql_dsn is required")
|
||
}
|
||
|
||
db, err := sql.Open("mysql", dsn)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
if err := db.PingContext(ctx); err != nil {
|
||
_ = db.Close()
|
||
return nil, err
|
||
}
|
||
|
||
return &Repository{db: db}, nil
|
||
}
|
||
|
||
// Close 释放 MySQL 连接池。
|
||
func (r *Repository) Close() error {
|
||
if r == nil || r.db == nil {
|
||
return nil
|
||
}
|
||
|
||
return r.db.Close()
|
||
}
|
||
|
||
// Ping 验证 MySQL 当前可用。
|
||
func (r *Repository) Ping(ctx context.Context) error {
|
||
if r == nil || r.db == nil {
|
||
return xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
|
||
return r.db.PingContext(ctx)
|
||
}
|
||
|
||
// DebitGift 在一个 MySQL 事务内完成 sender COIN 扣减、target GIFT_POINT 入账、分录和 outbox。
|
||
func (r *Repository) DebitGift(ctx context.Context, command ledger.DebitGiftCommand) (ledger.Receipt, error) {
|
||
if r == nil || r.db == nil {
|
||
return ledger.Receipt{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
ctx = contextWithCommandApp(ctx, command.AppCode)
|
||
command.AppCode = appcode.FromContext(ctx)
|
||
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
defer func() {
|
||
// Commit 成功后 Rollback 会返回 sql.ErrTxDone；忽略它可保证所有早退路径释放事务。
|
||
_ = tx.Rollback()
|
||
}()
|
||
|
||
requestHash := debitRequestHash(command)
|
||
if txRow, exists, err := r.lookupTransaction(ctx, tx, command.CommandID, requestHash, bizTypeGiftDebit); err != nil || exists {
|
||
if err != nil || !exists {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
receipt, receiptErr := r.receiptForGiftTransaction(ctx, tx, txRow.TransactionID)
|
||
return receipt, receiptErr
|
||
}
|
||
|
||
nowMs := time.Now().UnixMilli()
|
||
if command.RegionID < 0 {
|
||
return ledger.Receipt{}, xerr.New(xerr.InvalidArgument, "region_id is invalid")
|
||
}
|
||
giftConfig, err := r.resolveActiveGiftConfig(ctx, tx, command.GiftID, command.RegionID)
|
||
if err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
price, err := r.resolveGiftPrice(ctx, tx, command.GiftID, command.PriceVersion, nowMs)
|
||
if err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
chargeAmount, err := checkedMul(price.CoinPrice, int64(command.GiftCount))
|
||
if err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
giftPointAdded, err := checkedMul(price.GiftPointAmount, int64(command.GiftCount))
|
||
if err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
heatValue, err := checkedMul(price.HeatValue, int64(command.GiftCount))
|
||
if err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
|
||
sender, err := r.lockAccount(ctx, tx, command.SenderUserID, price.ChargeAssetType, chargeAmount == 0, nowMs)
|
||
if err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
if sender.AvailableAmount < chargeAmount {
|
||
return ledger.Receipt{}, xerr.New(xerr.InsufficientBalance, "insufficient balance")
|
||
}
|
||
target, err := r.lockAccount(ctx, tx, command.TargetUserID, ledger.AssetGiftPoint, true, nowMs)
|
||
if err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
|
||
transactionID := transactionID(command.AppCode, command.CommandID)
|
||
metadata := giftMetadata{
|
||
AppCode: command.AppCode,
|
||
GiftID: command.GiftID,
|
||
GiftName: giftConfig.Name,
|
||
ResourceID: giftConfig.ResourceID,
|
||
ResourceSnapshot: mustJSON(giftConfig.Resource),
|
||
GiftTypeCode: giftConfig.GiftTypeCode,
|
||
PresentationJSON: giftConfig.PresentationJSON,
|
||
SortOrder: giftConfig.SortOrder,
|
||
GiftCount: command.GiftCount,
|
||
PriceVersion: price.PriceVersion,
|
||
ChargeAssetType: price.ChargeAssetType,
|
||
ChargeAmount: chargeAmount,
|
||
CoinSpent: chargeAmount,
|
||
GiftPointAdded: giftPointAdded,
|
||
HeatValue: heatValue,
|
||
BalanceAfter: sender.AvailableAmount - chargeAmount,
|
||
BillingReceipt: billingReceiptID(command.AppCode, command.CommandID),
|
||
SenderUserID: command.SenderUserID,
|
||
TargetUserID: command.TargetUserID,
|
||
RoomID: command.RoomID,
|
||
CoinPrice: price.CoinPrice,
|
||
GiftPointAmount: price.GiftPointAmount,
|
||
HeatUnitValue: price.HeatValue,
|
||
}
|
||
if err := r.insertTransaction(ctx, tx, transactionID, command.CommandID, bizTypeGiftDebit, requestHash, fmt.Sprintf("%s:%s", command.GiftID, price.PriceVersion), metadata, nowMs); err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
if err := r.applyAccountDelta(ctx, tx, sender, -chargeAmount, 0, nowMs); err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
senderAfter := sender.AvailableAmount - chargeAmount
|
||
if err := r.insertEntry(ctx, tx, walletEntry{
|
||
TransactionID: transactionID,
|
||
UserID: command.SenderUserID,
|
||
AssetType: price.ChargeAssetType,
|
||
AvailableDelta: -chargeAmount,
|
||
FrozenDelta: 0,
|
||
AvailableAfter: senderAfter,
|
||
FrozenAfter: sender.FrozenAmount,
|
||
CounterpartyUserID: command.TargetUserID,
|
||
RoomID: command.RoomID,
|
||
CreatedAtMS: nowMs,
|
||
}); err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
if err := r.applyAccountDelta(ctx, tx, target, giftPointAdded, 0, nowMs); err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
targetAfter := target.AvailableAmount + giftPointAdded
|
||
if err := r.insertEntry(ctx, tx, walletEntry{
|
||
TransactionID: transactionID,
|
||
UserID: command.TargetUserID,
|
||
AssetType: ledger.AssetGiftPoint,
|
||
AvailableDelta: giftPointAdded,
|
||
FrozenDelta: 0,
|
||
AvailableAfter: targetAfter,
|
||
FrozenAfter: target.FrozenAmount,
|
||
CounterpartyUserID: command.SenderUserID,
|
||
RoomID: command.RoomID,
|
||
CreatedAtMS: nowMs,
|
||
}); err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
if err := r.insertWalletOutbox(ctx, tx, []walletOutboxEvent{
|
||
balanceChangedEvent(transactionID, command.CommandID, command.SenderUserID, price.ChargeAssetType, -chargeAmount, 0, senderAfter, sender.FrozenAmount, sender.Version+1, metadata, nowMs),
|
||
balanceChangedEvent(transactionID, command.CommandID, command.TargetUserID, ledger.AssetGiftPoint, giftPointAdded, 0, targetAfter, target.FrozenAmount, target.Version+1, metadata, nowMs),
|
||
giftDebitedEvent(transactionID, command.CommandID, command.SenderUserID, price.ChargeAssetType, -chargeAmount, 0, metadata, nowMs),
|
||
}); err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
|
||
return receiptFromGiftMetadata(transactionID, metadata), nil
|
||
}
|
||
|
||
// GetBalances 读取用户多资产余额；指定资产不存在时返回 0 投影，便于客户端稳定渲染。
|
||
func (r *Repository) GetBalances(ctx context.Context, userID int64, assetTypes []string) ([]ledger.AssetBalance, error) {
|
||
if r == nil || r.db == nil {
|
||
return nil, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
assetTypes = normalizeAssetTypes(assetTypes)
|
||
appCode := appcode.FromContext(ctx)
|
||
|
||
query := `SELECT app_code, user_id, asset_type, available_amount, frozen_amount, version, updated_at_ms
|
||
FROM wallet_accounts WHERE app_code = ? AND user_id = ?`
|
||
args := []any{appCode, userID}
|
||
if len(assetTypes) > 0 {
|
||
placeholders := strings.TrimRight(strings.Repeat("?,", len(assetTypes)), ",")
|
||
query += " AND asset_type IN (" + placeholders + ")"
|
||
for _, assetType := range assetTypes {
|
||
args = append(args, assetType)
|
||
}
|
||
}
|
||
query += " ORDER BY asset_type"
|
||
|
||
rows, err := r.db.QueryContext(ctx, query, args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
byAsset := make(map[string]ledger.AssetBalance)
|
||
for rows.Next() {
|
||
var balance ledger.AssetBalance
|
||
if err := rows.Scan(&balance.AppCode, &balance.UserID, &balance.AssetType, &balance.AvailableAmount, &balance.FrozenAmount, &balance.Version, &balance.UpdatedAtMs); err != nil {
|
||
return nil, err
|
||
}
|
||
byAsset[balance.AssetType] = balance
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
if len(assetTypes) == 0 {
|
||
balances := make([]ledger.AssetBalance, 0, len(byAsset))
|
||
for _, balance := range byAsset {
|
||
balances = append(balances, balance)
|
||
}
|
||
sort.Slice(balances, func(i, j int) bool { return balances[i].AssetType < balances[j].AssetType })
|
||
return balances, nil
|
||
}
|
||
|
||
balances := make([]ledger.AssetBalance, 0, len(assetTypes))
|
||
for _, assetType := range assetTypes {
|
||
if balance, ok := byAsset[assetType]; ok {
|
||
balances = append(balances, balance)
|
||
continue
|
||
}
|
||
balances = append(balances, ledger.AssetBalance{AppCode: appCode, UserID: userID, AssetType: assetType})
|
||
}
|
||
return balances, nil
|
||
}
|
||
|
||
func (r *Repository) upsertUserGiftWall(ctx context.Context, tx *sql.Tx, metadata giftMetadata, nowMs int64) error {
|
||
chargeAssetType := ledger.NormalizeGiftChargeAssetType(metadata.ChargeAssetType)
|
||
chargeAmount := metadata.ChargeAmount
|
||
if chargeAmount == 0 {
|
||
// metadata 同时保留 coin_spent 和 charge_amount；投影使用最终收费金额作为礼物价值。
|
||
chargeAmount = metadata.CoinSpent
|
||
}
|
||
totalCoinValue, totalDiamondValue := int64(0), int64(0)
|
||
if chargeAssetType == ledger.AssetDiamond {
|
||
totalDiamondValue = chargeAmount
|
||
} else {
|
||
totalCoinValue = chargeAmount
|
||
}
|
||
resourceSnapshotJSON := strings.TrimSpace(metadata.ResourceSnapshot)
|
||
if resourceSnapshotJSON == "" {
|
||
resourceSnapshotJSON = "{}"
|
||
}
|
||
presentationJSON := strings.TrimSpace(metadata.PresentationJSON)
|
||
if presentationJSON == "" {
|
||
presentationJSON = "{}"
|
||
}
|
||
giftTypeCode := strings.TrimSpace(metadata.GiftTypeCode)
|
||
if giftTypeCode == "" {
|
||
giftTypeCode = "normal"
|
||
}
|
||
|
||
_, err := tx.ExecContext(ctx, `
|
||
INSERT INTO user_gift_wall (
|
||
app_code, user_id, gift_id, gift_name, resource_id, resource_snapshot_json,
|
||
gift_type_code, presentation_json, gift_count, total_value, total_coin_value,
|
||
total_diamond_value, total_gift_point, total_heat_value, charge_asset_type,
|
||
last_price_version, first_received_at_ms, last_received_at_ms, sort_order,
|
||
created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
gift_name = VALUES(gift_name),
|
||
resource_id = VALUES(resource_id),
|
||
resource_snapshot_json = VALUES(resource_snapshot_json),
|
||
gift_type_code = VALUES(gift_type_code),
|
||
presentation_json = VALUES(presentation_json),
|
||
gift_count = gift_count + VALUES(gift_count),
|
||
total_value = total_value + VALUES(total_value),
|
||
total_coin_value = total_coin_value + VALUES(total_coin_value),
|
||
total_diamond_value = total_diamond_value + VALUES(total_diamond_value),
|
||
total_gift_point = total_gift_point + VALUES(total_gift_point),
|
||
total_heat_value = total_heat_value + VALUES(total_heat_value),
|
||
charge_asset_type = IF(charge_asset_type = VALUES(charge_asset_type), charge_asset_type, ?),
|
||
last_price_version = VALUES(last_price_version),
|
||
last_received_at_ms = VALUES(last_received_at_ms),
|
||
sort_order = VALUES(sort_order),
|
||
updated_at_ms = VALUES(updated_at_ms)
|
||
`,
|
||
appcode.FromContext(ctx),
|
||
metadata.TargetUserID,
|
||
metadata.GiftID,
|
||
metadata.GiftName,
|
||
metadata.ResourceID,
|
||
resourceSnapshotJSON,
|
||
giftTypeCode,
|
||
presentationJSON,
|
||
int64(metadata.GiftCount),
|
||
chargeAmount,
|
||
totalCoinValue,
|
||
totalDiamondValue,
|
||
metadata.GiftPointAdded,
|
||
metadata.HeatValue,
|
||
chargeAssetType,
|
||
metadata.PriceVersion,
|
||
nowMs,
|
||
nowMs,
|
||
metadata.SortOrder,
|
||
nowMs,
|
||
nowMs,
|
||
giftWallChargeAssetMixed,
|
||
)
|
||
return err
|
||
}
|
||
|
||
// AdminCreditAsset 在一个事务内写入后台调账交易、分录和 outbox。
|
||
func (r *Repository) AdminCreditAsset(ctx context.Context, command ledger.AdminCreditAssetCommand) (ledger.AssetBalance, string, error) {
|
||
if r == nil || r.db == nil {
|
||
return ledger.AssetBalance{}, "", xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
ctx = contextWithCommandApp(ctx, command.AppCode)
|
||
command.AppCode = appcode.FromContext(ctx)
|
||
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return ledger.AssetBalance{}, "", err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
|
||
requestHash := adminCreditRequestHash(command)
|
||
if txRow, exists, err := r.lookupTransaction(ctx, tx, command.CommandID, requestHash, bizTypeManualCredit); err != nil || exists {
|
||
if err != nil || !exists {
|
||
return ledger.AssetBalance{}, "", err
|
||
}
|
||
balance, balanceErr := r.balanceAfterTransaction(ctx, tx, txRow.TransactionID, command.TargetUserID, command.AssetType)
|
||
return balance, txRow.TransactionID, balanceErr
|
||
}
|
||
|
||
nowMs := time.Now().UnixMilli()
|
||
// 扣账不能隐式创建 0 余额账户；入账可以创建账户，保证运营补发能落到新用户。
|
||
account, err := r.lockAccount(ctx, tx, command.TargetUserID, command.AssetType, command.Amount > 0, nowMs)
|
||
if err != nil {
|
||
return ledger.AssetBalance{}, "", err
|
||
}
|
||
if command.Amount < 0 && (command.Amount == math.MinInt64 || account.AvailableAmount < -command.Amount) {
|
||
return ledger.AssetBalance{}, "", xerr.New(xerr.InsufficientBalance, "insufficient balance")
|
||
}
|
||
transactionID := transactionID(command.AppCode, command.CommandID)
|
||
metadata := adminCreditMetadata{
|
||
AppCode: command.AppCode,
|
||
TargetUserID: command.TargetUserID,
|
||
AssetType: command.AssetType,
|
||
Amount: command.Amount,
|
||
OperatorUserID: command.OperatorUserID,
|
||
Reason: command.Reason,
|
||
EvidenceRef: command.EvidenceRef,
|
||
}
|
||
if err := r.insertTransaction(ctx, tx, transactionID, command.CommandID, bizTypeManualCredit, requestHash, command.EvidenceRef, metadata, nowMs); err != nil {
|
||
return ledger.AssetBalance{}, "", err
|
||
}
|
||
if err := r.applyAccountDelta(ctx, tx, account, command.Amount, 0, nowMs); err != nil {
|
||
return ledger.AssetBalance{}, "", err
|
||
}
|
||
balance := ledger.AssetBalance{
|
||
AppCode: command.AppCode,
|
||
UserID: command.TargetUserID,
|
||
AssetType: command.AssetType,
|
||
AvailableAmount: account.AvailableAmount + command.Amount,
|
||
FrozenAmount: account.FrozenAmount,
|
||
Version: account.Version + 1,
|
||
UpdatedAtMs: nowMs,
|
||
}
|
||
if err := r.insertEntry(ctx, tx, walletEntry{
|
||
TransactionID: transactionID,
|
||
UserID: command.TargetUserID,
|
||
AssetType: command.AssetType,
|
||
AvailableDelta: command.Amount,
|
||
FrozenDelta: 0,
|
||
AvailableAfter: balance.AvailableAmount,
|
||
FrozenAfter: balance.FrozenAmount,
|
||
CreatedAtMS: nowMs,
|
||
}); err != nil {
|
||
return ledger.AssetBalance{}, "", err
|
||
}
|
||
if err := r.insertWalletOutbox(ctx, tx, []walletOutboxEvent{
|
||
balanceChangedEvent(transactionID, command.CommandID, command.TargetUserID, command.AssetType, command.Amount, 0, balance.AvailableAmount, balance.FrozenAmount, balance.Version, metadata, nowMs),
|
||
}); err != nil {
|
||
return ledger.AssetBalance{}, "", err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return ledger.AssetBalance{}, "", err
|
||
}
|
||
|
||
return balance, transactionID, nil
|
||
}
|
||
|
||
// CreditTaskReward 在一个事务内完成任务奖励 COIN 入账、交易分录和 outbox 事件。
|
||
func (r *Repository) CreditTaskReward(ctx context.Context, command ledger.TaskRewardCommand) (ledger.TaskRewardReceipt, error) {
|
||
if r == nil || r.db == nil {
|
||
return ledger.TaskRewardReceipt{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
ctx = contextWithCommandApp(ctx, command.AppCode)
|
||
command.AppCode = appcode.FromContext(ctx)
|
||
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return ledger.TaskRewardReceipt{}, err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
|
||
requestHash := taskRewardRequestHash(command)
|
||
if txRow, exists, err := r.lookupTransaction(ctx, tx, command.CommandID, requestHash, bizTypeTaskReward); err != nil || exists {
|
||
if err != nil || !exists {
|
||
return ledger.TaskRewardReceipt{}, err
|
||
}
|
||
return r.receiptForTaskRewardTransaction(ctx, tx, txRow.TransactionID)
|
||
}
|
||
|
||
nowMs := time.Now().UnixMilli()
|
||
account, err := r.lockAccount(ctx, tx, command.TargetUserID, ledger.AssetCoin, true, nowMs)
|
||
if err != nil {
|
||
return ledger.TaskRewardReceipt{}, err
|
||
}
|
||
transactionID := transactionID(command.AppCode, command.CommandID)
|
||
balanceAfter := account.AvailableAmount + command.Amount
|
||
metadata := taskRewardMetadata{
|
||
AppCode: command.AppCode,
|
||
TargetUserID: command.TargetUserID,
|
||
AssetType: ledger.AssetCoin,
|
||
Amount: command.Amount,
|
||
TaskType: command.TaskType,
|
||
TaskID: command.TaskID,
|
||
CycleKey: command.CycleKey,
|
||
Reason: command.Reason,
|
||
BalanceAfter: balanceAfter,
|
||
GrantedAtMS: nowMs,
|
||
}
|
||
if err := r.insertTransaction(ctx, tx, transactionID, command.CommandID, bizTypeTaskReward, requestHash, fmt.Sprintf("%s:%s", command.TaskID, command.CycleKey), metadata, nowMs); err != nil {
|
||
return ledger.TaskRewardReceipt{}, err
|
||
}
|
||
if err := r.applyAccountDelta(ctx, tx, account, command.Amount, 0, nowMs); err != nil {
|
||
return ledger.TaskRewardReceipt{}, err
|
||
}
|
||
if err := r.insertEntry(ctx, tx, walletEntry{
|
||
TransactionID: transactionID,
|
||
UserID: command.TargetUserID,
|
||
AssetType: ledger.AssetCoin,
|
||
AvailableDelta: command.Amount,
|
||
FrozenDelta: 0,
|
||
AvailableAfter: balanceAfter,
|
||
FrozenAfter: account.FrozenAmount,
|
||
CreatedAtMS: nowMs,
|
||
}); err != nil {
|
||
return ledger.TaskRewardReceipt{}, err
|
||
}
|
||
if err := r.insertWalletOutbox(ctx, tx, []walletOutboxEvent{
|
||
balanceChangedEvent(transactionID, command.CommandID, command.TargetUserID, ledger.AssetCoin, command.Amount, 0, balanceAfter, account.FrozenAmount, account.Version+1, metadata, nowMs),
|
||
taskRewardCreditedEvent(transactionID, command.CommandID, metadata, nowMs),
|
||
}); err != nil {
|
||
return ledger.TaskRewardReceipt{}, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return ledger.TaskRewardReceipt{}, err
|
||
}
|
||
|
||
return receiptFromTaskRewardMetadata(transactionID, metadata), nil
|
||
}
|
||
|
||
// ApplyGameCoinChange 在同一事务内完成游戏 COIN 改账、分录和 outbox。
|
||
func (r *Repository) ApplyGameCoinChange(ctx context.Context, command ledger.GameCoinChangeCommand) (ledger.GameCoinChangeReceipt, error) {
|
||
if r == nil || r.db == nil {
|
||
return ledger.GameCoinChangeReceipt{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
ctx = contextWithCommandApp(ctx, command.AppCode)
|
||
command.AppCode = appcode.FromContext(ctx)
|
||
|
||
bizType, delta, err := gameBizTypeAndDelta(command.OpType, command.CoinAmount)
|
||
if err != nil {
|
||
return ledger.GameCoinChangeReceipt{}, err
|
||
}
|
||
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return ledger.GameCoinChangeReceipt{}, err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
|
||
if txRow, exists, err := r.lookupTransactionWithConflictCode(ctx, tx, command.CommandID, command.RequestHash, bizType, xerr.IdempotencyConflict); err != nil || exists {
|
||
if err != nil || !exists {
|
||
return ledger.GameCoinChangeReceipt{}, err
|
||
}
|
||
receipt, receiptErr := r.receiptForGameTransaction(ctx, tx, txRow.TransactionID)
|
||
receipt.IdempotentReplay = true
|
||
return receipt, receiptErr
|
||
}
|
||
|
||
nowMs := time.Now().UnixMilli()
|
||
createIfMissing := delta > 0
|
||
account, err := r.lockAccount(ctx, tx, command.UserID, ledger.AssetCoin, createIfMissing, nowMs)
|
||
if err != nil {
|
||
return ledger.GameCoinChangeReceipt{}, err
|
||
}
|
||
if delta < 0 && account.AvailableAmount < -delta {
|
||
return ledger.GameCoinChangeReceipt{}, xerr.New(xerr.InsufficientBalance, "insufficient balance")
|
||
}
|
||
|
||
transactionID := transactionID(command.AppCode, command.CommandID)
|
||
balanceAfter := account.AvailableAmount + delta
|
||
metadata := gameCoinMetadata{
|
||
AppCode: command.AppCode,
|
||
UserID: command.UserID,
|
||
PlatformCode: command.PlatformCode,
|
||
GameID: command.GameID,
|
||
ProviderOrderID: command.ProviderOrderID,
|
||
ProviderRoundID: command.ProviderRoundID,
|
||
OpType: command.OpType,
|
||
AssetType: ledger.AssetCoin,
|
||
CoinAmount: command.CoinAmount,
|
||
AvailableDelta: delta,
|
||
RoomID: command.RoomID,
|
||
BalanceAfter: balanceAfter,
|
||
AppliedAtMS: nowMs,
|
||
}
|
||
if err := r.insertTransaction(ctx, tx, transactionID, command.CommandID, bizType, command.RequestHash, command.ProviderOrderID, metadata, nowMs); err != nil {
|
||
return ledger.GameCoinChangeReceipt{}, err
|
||
}
|
||
if err := r.applyAccountDelta(ctx, tx, account, delta, 0, nowMs); err != nil {
|
||
return ledger.GameCoinChangeReceipt{}, err
|
||
}
|
||
if err := r.insertEntry(ctx, tx, walletEntry{
|
||
TransactionID: transactionID,
|
||
UserID: command.UserID,
|
||
AssetType: ledger.AssetCoin,
|
||
AvailableDelta: delta,
|
||
FrozenDelta: 0,
|
||
AvailableAfter: balanceAfter,
|
||
FrozenAfter: account.FrozenAmount,
|
||
RoomID: command.RoomID,
|
||
CreatedAtMS: nowMs,
|
||
}); err != nil {
|
||
return ledger.GameCoinChangeReceipt{}, err
|
||
}
|
||
if err := r.insertWalletOutbox(ctx, tx, []walletOutboxEvent{
|
||
balanceChangedEvent(transactionID, command.CommandID, command.UserID, ledger.AssetCoin, delta, 0, balanceAfter, account.FrozenAmount, account.Version+1, metadata, nowMs),
|
||
}); err != nil {
|
||
return ledger.GameCoinChangeReceipt{}, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return ledger.GameCoinChangeReceipt{}, err
|
||
}
|
||
|
||
return receiptFromGameMetadata(transactionID, metadata, false), nil
|
||
}
|
||
|
||
// AdminCreditCoinSellerStock 在同一事务里给币商 COIN_SELLER_COIN 库存入账，并记录专用进货明细。
|
||
func (r *Repository) AdminCreditCoinSellerStock(ctx context.Context, command ledger.CoinSellerStockCreditCommand) (ledger.CoinSellerStockCreditReceipt, error) {
|
||
if r == nil || r.db == nil {
|
||
return ledger.CoinSellerStockCreditReceipt{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
ctx = contextWithCommandApp(ctx, command.AppCode)
|
||
command.AppCode = appcode.FromContext(ctx)
|
||
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return ledger.CoinSellerStockCreditReceipt{}, err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
|
||
bizType := coinSellerStockBizType(command.StockType)
|
||
if bizType == "" {
|
||
return ledger.CoinSellerStockCreditReceipt{}, xerr.New(xerr.CoinSellerStockTypeInvalid, "stock_type is invalid")
|
||
}
|
||
requestHash := coinSellerStockRequestHash(command)
|
||
if txRow, exists, err := r.lookupTransactionWithConflictCode(ctx, tx, command.CommandID, requestHash, bizType, xerr.IdempotencyConflict); err != nil || exists {
|
||
if err != nil || !exists {
|
||
return ledger.CoinSellerStockCreditReceipt{}, err
|
||
}
|
||
return r.receiptForCoinSellerStockTransaction(ctx, tx, txRow.TransactionID)
|
||
}
|
||
if command.PaymentRef != "" {
|
||
if duplicated, err := r.coinSellerStockPaymentRefExists(ctx, tx, command.StockType, command.PaymentRef); err != nil || duplicated {
|
||
if err != nil {
|
||
return ledger.CoinSellerStockCreditReceipt{}, err
|
||
}
|
||
return ledger.CoinSellerStockCreditReceipt{}, xerr.New(xerr.CoinSellerPaymentRefDuplicated, "payment_ref is duplicated")
|
||
}
|
||
}
|
||
|
||
nowMs := time.Now().UnixMilli()
|
||
account, err := r.lockAccount(ctx, tx, command.SellerUserID, ledger.AssetCoinSellerCoin, true, nowMs)
|
||
if err != nil {
|
||
return ledger.CoinSellerStockCreditReceipt{}, err
|
||
}
|
||
balanceAfter, err := checkedAdd(account.AvailableAmount, command.CoinAmount)
|
||
if err != nil {
|
||
return ledger.CoinSellerStockCreditReceipt{}, err
|
||
}
|
||
|
||
transactionID := transactionID(command.AppCode, command.CommandID)
|
||
countsAsSellerRecharge := command.StockType == ledger.StockTypeUSDTPurchase
|
||
metadata := coinSellerStockMetadata{
|
||
AppCode: command.AppCode,
|
||
SellerUserID: command.SellerUserID,
|
||
StockType: command.StockType,
|
||
CoinAmount: command.CoinAmount,
|
||
PaidCurrencyCode: command.PaidCurrencyCode,
|
||
PaidAmountMicro: command.PaidAmountMicro,
|
||
PaymentRef: command.PaymentRef,
|
||
EvidenceRef: command.EvidenceRef,
|
||
OperatorUserID: command.OperatorUserID,
|
||
Reason: command.Reason,
|
||
AssetType: ledger.AssetCoinSellerCoin,
|
||
CountsAsSellerRecharge: countsAsSellerRecharge,
|
||
BalanceAfter: balanceAfter,
|
||
CreatedAtMS: nowMs,
|
||
}
|
||
externalRef := command.EvidenceRef
|
||
if command.PaymentRef != "" {
|
||
externalRef = command.PaymentRef
|
||
}
|
||
if err := r.insertTransaction(ctx, tx, transactionID, command.CommandID, bizType, requestHash, externalRef, metadata, nowMs); err != nil {
|
||
return ledger.CoinSellerStockCreditReceipt{}, err
|
||
}
|
||
if err := r.applyAccountDelta(ctx, tx, account, command.CoinAmount, 0, nowMs); err != nil {
|
||
return ledger.CoinSellerStockCreditReceipt{}, err
|
||
}
|
||
if err := r.insertEntry(ctx, tx, walletEntry{
|
||
TransactionID: transactionID,
|
||
UserID: command.SellerUserID,
|
||
AssetType: ledger.AssetCoinSellerCoin,
|
||
AvailableDelta: command.CoinAmount,
|
||
FrozenDelta: 0,
|
||
AvailableAfter: balanceAfter,
|
||
FrozenAfter: account.FrozenAmount,
|
||
CreatedAtMS: nowMs,
|
||
}); err != nil {
|
||
return ledger.CoinSellerStockCreditReceipt{}, err
|
||
}
|
||
if err := r.insertCoinSellerStockRecord(ctx, tx, transactionID, command.CommandID, metadata, nowMs); err != nil {
|
||
return ledger.CoinSellerStockCreditReceipt{}, err
|
||
}
|
||
if err := r.insertWalletOutbox(ctx, tx, []walletOutboxEvent{
|
||
balanceChangedEvent(transactionID, command.CommandID, command.SellerUserID, ledger.AssetCoinSellerCoin, command.CoinAmount, 0, balanceAfter, account.FrozenAmount, account.Version+1, metadata, nowMs),
|
||
coinSellerStockCreditedEvent(transactionID, command.CommandID, metadata, nowMs),
|
||
}); err != nil {
|
||
return ledger.CoinSellerStockCreditReceipt{}, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return ledger.CoinSellerStockCreditReceipt{}, err
|
||
}
|
||
|
||
return receiptFromCoinSellerStockMetadata(transactionID, metadata), nil
|
||
}
|
||
|
||
// TransferCoinFromSeller 在同一事务里扣币商专用金币、给玩家普通 COIN 入账，并记录充值口径。
|
||
func (r *Repository) TransferCoinFromSeller(ctx context.Context, command ledger.CoinSellerTransferCommand) (ledger.CoinSellerTransferReceipt, error) {
|
||
if r == nil || r.db == nil {
|
||
return ledger.CoinSellerTransferReceipt{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
ctx = contextWithCommandApp(ctx, command.AppCode)
|
||
command.AppCode = appcode.FromContext(ctx)
|
||
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
|
||
requestHash := coinSellerTransferRequestHash(command)
|
||
if txRow, exists, err := r.lookupTransaction(ctx, tx, command.CommandID, requestHash, bizTypeCoinSellerTransfer); err != nil || exists {
|
||
if err != nil || !exists {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
return r.receiptForCoinSellerTransferTransaction(ctx, tx, txRow.TransactionID)
|
||
}
|
||
|
||
nowMs := time.Now().UnixMilli()
|
||
seller, err := r.lockAccount(ctx, tx, command.SellerUserID, ledger.AssetCoinSellerCoin, false, nowMs)
|
||
if err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
if seller.AvailableAmount < command.Amount {
|
||
return ledger.CoinSellerTransferReceipt{}, xerr.New(xerr.InsufficientBalance, "insufficient balance")
|
||
}
|
||
target, err := r.lockAccount(ctx, tx, command.TargetUserID, ledger.AssetCoin, true, nowMs)
|
||
if err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
|
||
transactionID := transactionID(command.AppCode, command.CommandID)
|
||
policy, err := r.resolveRechargePolicy(ctx, tx, command.TargetRegionID, nowMs)
|
||
if err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
rechargeUSDMinor, err := calculateRechargeUSDMinor(command.Amount, policy)
|
||
if err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
rechargeSequence, err := r.reserveUserRechargeSequence(ctx, tx, command.AppCode, command.TargetUserID, transactionID, command.Amount, rechargeUSDMinor, nowMs)
|
||
if err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
sellerAfter := seller.AvailableAmount - command.Amount
|
||
targetAfter := target.AvailableAmount + command.Amount
|
||
metadata := coinSellerTransferMetadata{
|
||
AppCode: command.AppCode,
|
||
SellerUserID: command.SellerUserID,
|
||
TargetUserID: command.TargetUserID,
|
||
SellerRegionID: command.SellerRegionID,
|
||
TargetRegionID: command.TargetRegionID,
|
||
Amount: command.Amount,
|
||
Reason: command.Reason,
|
||
SellerAssetType: ledger.AssetCoinSellerCoin,
|
||
TargetAssetType: ledger.AssetCoin,
|
||
SellerBalanceAfter: sellerAfter,
|
||
TargetBalanceAfter: targetAfter,
|
||
RechargeSequence: rechargeSequence,
|
||
RechargeUSDMinor: rechargeUSDMinor,
|
||
RechargeCurrencyCode: policy.CurrencyCode,
|
||
RechargePolicyID: policy.PolicyID,
|
||
RechargePolicyVersion: policy.PolicyVersion,
|
||
RechargePolicyCoinAmount: policy.CoinAmount,
|
||
RechargePolicyUSDMinorAmount: policy.USDMinorAmount,
|
||
}
|
||
if err := r.insertTransaction(ctx, tx, transactionID, command.CommandID, bizTypeCoinSellerTransfer, requestHash, fmt.Sprintf("coin_seller:%d:%d", command.SellerUserID, command.TargetUserID), metadata, nowMs); err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
if err := r.applyAccountDelta(ctx, tx, seller, -command.Amount, 0, nowMs); err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
if err := r.insertEntry(ctx, tx, walletEntry{
|
||
TransactionID: transactionID,
|
||
UserID: command.SellerUserID,
|
||
AssetType: ledger.AssetCoinSellerCoin,
|
||
AvailableDelta: -command.Amount,
|
||
FrozenDelta: 0,
|
||
AvailableAfter: sellerAfter,
|
||
FrozenAfter: seller.FrozenAmount,
|
||
CounterpartyUserID: command.TargetUserID,
|
||
CreatedAtMS: nowMs,
|
||
}); err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
if err := r.applyAccountDelta(ctx, tx, target, command.Amount, 0, nowMs); err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
if err := r.insertEntry(ctx, tx, walletEntry{
|
||
TransactionID: transactionID,
|
||
UserID: command.TargetUserID,
|
||
AssetType: ledger.AssetCoin,
|
||
AvailableDelta: command.Amount,
|
||
FrozenDelta: 0,
|
||
AvailableAfter: targetAfter,
|
||
FrozenAfter: target.FrozenAmount,
|
||
CounterpartyUserID: command.SellerUserID,
|
||
CreatedAtMS: nowMs,
|
||
}); err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
if err := r.insertRechargeRecord(ctx, tx, transactionID, metadata, nowMs); err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
if err := r.insertWalletOutbox(ctx, tx, []walletOutboxEvent{
|
||
balanceChangedEvent(transactionID, command.CommandID, command.SellerUserID, ledger.AssetCoinSellerCoin, -command.Amount, 0, sellerAfter, seller.FrozenAmount, seller.Version+1, metadata, nowMs),
|
||
balanceChangedEvent(transactionID, command.CommandID, command.TargetUserID, ledger.AssetCoin, command.Amount, 0, targetAfter, target.FrozenAmount, target.Version+1, metadata, nowMs),
|
||
coinSellerTransferredEvent(transactionID, command.CommandID, command.SellerUserID, -command.Amount, metadata, nowMs),
|
||
rechargeRecordedEvent(transactionID, command.CommandID, command.TargetUserID, command.Amount, metadata, nowMs),
|
||
}); err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
|
||
return receiptFromCoinSellerTransferMetadata(transactionID, metadata), nil
|
||
}
|
||
|
||
type transactionRow struct {
|
||
TransactionID string
|
||
MetadataJSON string
|
||
}
|
||
|
||
func (r *Repository) lookupTransaction(ctx context.Context, tx *sql.Tx, commandID string, requestHash string, bizType string) (transactionRow, bool, error) {
|
||
return r.lookupTransactionWithConflictCode(ctx, tx, commandID, requestHash, bizType, xerr.LedgerConflict)
|
||
}
|
||
|
||
func (r *Repository) lookupTransactionWithConflictCode(ctx context.Context, tx *sql.Tx, commandID string, requestHash string, bizType string, conflictCode xerr.Code) (transactionRow, bool, error) {
|
||
row := tx.QueryRowContext(ctx,
|
||
`SELECT transaction_id, request_hash, biz_type, COALESCE(CAST(metadata_json AS CHAR), '{}')
|
||
FROM wallet_transactions
|
||
WHERE app_code = ? AND command_id = ?
|
||
FOR UPDATE`,
|
||
appcode.FromContext(ctx),
|
||
commandID,
|
||
)
|
||
|
||
var txRow transactionRow
|
||
var storedHash string
|
||
var storedBizType string
|
||
if err := row.Scan(&txRow.TransactionID, &storedHash, &storedBizType, &txRow.MetadataJSON); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return transactionRow{}, false, nil
|
||
}
|
||
return transactionRow{}, false, err
|
||
}
|
||
if storedHash != requestHash || storedBizType != bizType {
|
||
// command_id 是钱包幂等主键，业务 payload 或命令类型变化必须 fail-closed。
|
||
return transactionRow{}, true, xerr.New(conflictCode, "wallet command idempotency conflict")
|
||
}
|
||
return txRow, true, nil
|
||
}
|
||
|
||
type walletAccount struct {
|
||
AppCode string
|
||
UserID int64
|
||
AssetType string
|
||
AvailableAmount int64
|
||
FrozenAmount int64
|
||
Version int64
|
||
UpdatedAtMS int64
|
||
}
|
||
|
||
func (r *Repository) lockAccount(ctx context.Context, tx *sql.Tx, userID int64, assetType string, createIfMissing bool, nowMs int64) (walletAccount, error) {
|
||
account, exists, err := r.queryAccountForUpdate(ctx, tx, userID, assetType)
|
||
if err != nil || exists || !createIfMissing {
|
||
if err != nil {
|
||
return walletAccount{}, err
|
||
}
|
||
if !exists {
|
||
return walletAccount{}, xerr.New(xerr.InsufficientBalance, "insufficient balance")
|
||
}
|
||
return account, nil
|
||
}
|
||
|
||
if _, err := tx.ExecContext(ctx,
|
||
`INSERT INTO wallet_accounts (app_code, user_id, asset_type, available_amount, frozen_amount, version, created_at_ms, updated_at_ms)
|
||
VALUES (?, ?, ?, 0, 0, 1, ?, ?)`,
|
||
appcode.FromContext(ctx), userID, assetType, nowMs, nowMs,
|
||
); err != nil {
|
||
return walletAccount{}, err
|
||
}
|
||
return r.queryRequiredAccountForUpdate(ctx, tx, userID, assetType)
|
||
}
|
||
|
||
func (r *Repository) queryAccountForUpdate(ctx context.Context, tx *sql.Tx, userID int64, assetType string) (walletAccount, bool, error) {
|
||
row := tx.QueryRowContext(ctx,
|
||
`SELECT app_code, user_id, asset_type, available_amount, frozen_amount, version, updated_at_ms
|
||
FROM wallet_accounts
|
||
WHERE app_code = ? AND user_id = ? AND asset_type = ?
|
||
FOR UPDATE`,
|
||
appcode.FromContext(ctx),
|
||
userID, assetType,
|
||
)
|
||
var account walletAccount
|
||
if err := row.Scan(&account.AppCode, &account.UserID, &account.AssetType, &account.AvailableAmount, &account.FrozenAmount, &account.Version, &account.UpdatedAtMS); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return walletAccount{}, false, nil
|
||
}
|
||
return walletAccount{}, false, err
|
||
}
|
||
return account, true, nil
|
||
}
|
||
|
||
func (r *Repository) queryRequiredAccountForUpdate(ctx context.Context, tx *sql.Tx, userID int64, assetType string) (walletAccount, error) {
|
||
account, exists, err := r.queryAccountForUpdate(ctx, tx, userID, assetType)
|
||
if err != nil {
|
||
return walletAccount{}, err
|
||
}
|
||
if !exists {
|
||
return walletAccount{}, xerr.New(xerr.Internal, "wallet account creation failed")
|
||
}
|
||
return account, nil
|
||
}
|
||
|
||
func (r *Repository) applyAccountDelta(ctx context.Context, tx *sql.Tx, account walletAccount, availableDelta int64, frozenDelta int64, nowMs int64) error {
|
||
availableAfter := account.AvailableAmount + availableDelta
|
||
frozenAfter := account.FrozenAmount + frozenDelta
|
||
if availableAfter < 0 || frozenAfter < 0 || availableAfter > math.MaxInt64-frozenAfter {
|
||
// 钱包负债不能出现负数，也不能因溢出绕过余额约束。
|
||
return xerr.New(xerr.LedgerConflict, "wallet balance delta is invalid")
|
||
}
|
||
|
||
result, err := tx.ExecContext(ctx,
|
||
`UPDATE wallet_accounts
|
||
SET available_amount = ?, frozen_amount = ?, version = version + 1, updated_at_ms = ?
|
||
WHERE app_code = ? AND user_id = ? AND asset_type = ? AND version = ?`,
|
||
availableAfter,
|
||
frozenAfter,
|
||
nowMs,
|
||
account.AppCode,
|
||
account.UserID,
|
||
account.AssetType,
|
||
account.Version,
|
||
)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
rows, err := result.RowsAffected()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if rows != 1 {
|
||
return xerr.New(xerr.LedgerConflict, "wallet account version conflict")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
type giftPrice struct {
|
||
GiftID string
|
||
PriceVersion string
|
||
ChargeAssetType string
|
||
CoinPrice int64
|
||
GiftPointAmount int64
|
||
HeatValue int64
|
||
}
|
||
|
||
type rechargePolicy struct {
|
||
PolicyID int64
|
||
RegionID int64
|
||
PolicyVersion string
|
||
CurrencyCode string
|
||
CoinAmount int64
|
||
USDMinorAmount int64
|
||
EffectiveFromMs int64
|
||
}
|
||
|
||
func (r *Repository) resolveGiftPrice(ctx context.Context, tx *sql.Tx, giftID string, priceVersion string, nowMs int64) (giftPrice, error) {
|
||
var row *sql.Row
|
||
if strings.TrimSpace(priceVersion) != "" {
|
||
row = tx.QueryRowContext(ctx,
|
||
`SELECT gift_id, price_version, COALESCE(charge_asset_type, 'COIN'), coin_price, gift_point_amount, heat_value
|
||
FROM wallet_gift_prices
|
||
WHERE app_code = ? AND gift_id = ? AND price_version = ? AND status = 'active' AND effective_at_ms <= ?`,
|
||
appcode.FromContext(ctx), giftID, priceVersion, nowMs,
|
||
)
|
||
} else {
|
||
row = tx.QueryRowContext(ctx,
|
||
`SELECT gift_id, price_version, COALESCE(charge_asset_type, 'COIN'), coin_price, gift_point_amount, heat_value
|
||
FROM wallet_gift_prices
|
||
WHERE app_code = ? AND gift_id = ? AND status = 'active' AND effective_at_ms <= ?
|
||
ORDER BY effective_at_ms DESC, price_version DESC
|
||
LIMIT 1`,
|
||
appcode.FromContext(ctx), giftID, nowMs,
|
||
)
|
||
}
|
||
|
||
var price giftPrice
|
||
if err := row.Scan(&price.GiftID, &price.PriceVersion, &price.ChargeAssetType, &price.CoinPrice, &price.GiftPointAmount, &price.HeatValue); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return giftPrice{}, xerr.New(xerr.NotFound, "gift price is not active")
|
||
}
|
||
return giftPrice{}, err
|
||
}
|
||
if price.CoinPrice < 0 || price.GiftPointAmount < 0 || price.HeatValue < 0 {
|
||
return giftPrice{}, xerr.New(xerr.Internal, "gift price is invalid")
|
||
}
|
||
price.ChargeAssetType = ledger.NormalizeGiftChargeAssetType(price.ChargeAssetType)
|
||
return price, nil
|
||
}
|
||
|
||
func (r *Repository) resolveRechargePolicy(ctx context.Context, tx *sql.Tx, regionID int64, nowMs int64) (rechargePolicy, error) {
|
||
row := tx.QueryRowContext(ctx,
|
||
`SELECT policy_id, region_id, policy_version, currency_code, coin_amount, usd_minor_amount, effective_from_ms
|
||
FROM wallet_recharge_policies
|
||
WHERE app_code = ? AND region_id = ? AND status = 'active'
|
||
AND effective_from_ms <= ?
|
||
AND (effective_to_ms IS NULL OR effective_to_ms > ?)
|
||
ORDER BY effective_from_ms DESC, policy_id DESC
|
||
LIMIT 1`,
|
||
appcode.FromContext(ctx),
|
||
regionID,
|
||
nowMs,
|
||
nowMs,
|
||
)
|
||
var policy rechargePolicy
|
||
if err := row.Scan(&policy.PolicyID, &policy.RegionID, &policy.PolicyVersion, &policy.CurrencyCode, &policy.CoinAmount, &policy.USDMinorAmount, &policy.EffectiveFromMs); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return rechargePolicy{}, xerr.New(xerr.NotFound, "recharge policy not found")
|
||
}
|
||
return rechargePolicy{}, err
|
||
}
|
||
if policy.CoinAmount <= 0 || policy.USDMinorAmount <= 0 || strings.TrimSpace(policy.CurrencyCode) == "" {
|
||
return rechargePolicy{}, xerr.New(xerr.Internal, "recharge policy is invalid")
|
||
}
|
||
return policy, nil
|
||
}
|
||
|
||
func calculateRechargeUSDMinor(coinAmount int64, policy rechargePolicy) (int64, error) {
|
||
numerator, err := checkedMul(coinAmount, policy.USDMinorAmount)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
if numerator%policy.CoinAmount != 0 {
|
||
// 充值金额按最小货币单位入账；不能精确换算时拒绝，避免财务口径出现隐式四舍五入。
|
||
return 0, xerr.New(xerr.InvalidArgument, "coin amount does not match recharge policy")
|
||
}
|
||
return numerator / policy.CoinAmount, nil
|
||
}
|
||
|
||
func (r *Repository) insertTransaction(ctx context.Context, tx *sql.Tx, transactionID string, commandID string, bizType string, requestHash string, externalRef string, metadata any, nowMs int64) error {
|
||
metadataJSON, err := json.Marshal(metadata)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = tx.ExecContext(ctx,
|
||
`INSERT INTO wallet_transactions (
|
||
app_code, transaction_id, command_id, biz_type, status, request_hash, external_ref, metadata_json, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||
appcode.FromContext(ctx),
|
||
transactionID,
|
||
commandID,
|
||
bizType,
|
||
transactionStatusSucceeded,
|
||
requestHash,
|
||
externalRef,
|
||
string(metadataJSON),
|
||
nowMs,
|
||
nowMs,
|
||
)
|
||
return err
|
||
}
|
||
|
||
func (r *Repository) insertRechargeRecord(ctx context.Context, tx *sql.Tx, transactionID string, metadata coinSellerTransferMetadata, nowMs int64) error {
|
||
_, err := tx.ExecContext(ctx,
|
||
`INSERT INTO wallet_recharge_records (
|
||
app_code, transaction_id, user_id, recharge_sequence, seller_user_id, seller_region_id, target_region_id,
|
||
policy_id, policy_version, currency_code, coin_amount, usd_minor_amount,
|
||
exchange_coin_amount, exchange_usd_minor_amount, created_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||
normalizedEntryAppCode(ctx, metadata.AppCode),
|
||
transactionID,
|
||
metadata.TargetUserID,
|
||
metadata.RechargeSequence,
|
||
metadata.SellerUserID,
|
||
metadata.SellerRegionID,
|
||
metadata.TargetRegionID,
|
||
metadata.RechargePolicyID,
|
||
metadata.RechargePolicyVersion,
|
||
metadata.RechargeCurrencyCode,
|
||
metadata.Amount,
|
||
metadata.RechargeUSDMinor,
|
||
metadata.RechargePolicyCoinAmount,
|
||
metadata.RechargePolicyUSDMinorAmount,
|
||
nowMs,
|
||
)
|
||
return err
|
||
}
|
||
|
||
func (r *Repository) reserveUserRechargeSequence(ctx context.Context, tx *sql.Tx, appCode string, userID int64, transactionID string, coinAmount int64, usdMinorAmount int64, nowMs int64) (int64, error) {
|
||
normalizedApp := normalizedEntryAppCode(ctx, appCode)
|
||
var count int64
|
||
err := tx.QueryRowContext(ctx, `
|
||
SELECT recharge_count
|
||
FROM wallet_user_recharge_stats
|
||
WHERE app_code = ? AND user_id = ?
|
||
FOR UPDATE`,
|
||
normalizedApp, userID,
|
||
).Scan(&count)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
_, err = tx.ExecContext(ctx, `
|
||
INSERT INTO wallet_user_recharge_stats (
|
||
app_code, user_id, recharge_count, first_transaction_id, first_recharged_at_ms,
|
||
total_coin_amount, total_usd_minor_amount, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, 1, ?, ?, ?, ?, ?, ?)`,
|
||
normalizedApp, userID, transactionID, nowMs, coinAmount, usdMinorAmount, nowMs, nowMs,
|
||
)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return 1, nil
|
||
}
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
next := count + 1
|
||
_, err = tx.ExecContext(ctx, `
|
||
UPDATE wallet_user_recharge_stats
|
||
SET recharge_count = ?,
|
||
total_coin_amount = total_coin_amount + ?,
|
||
total_usd_minor_amount = total_usd_minor_amount + ?,
|
||
updated_at_ms = ?
|
||
WHERE app_code = ? AND user_id = ?`,
|
||
next, coinAmount, usdMinorAmount, nowMs, normalizedApp, userID,
|
||
)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return next, nil
|
||
}
|
||
|
||
func (r *Repository) insertCoinSellerStockRecord(ctx context.Context, tx *sql.Tx, transactionID string, commandID string, metadata coinSellerStockMetadata, nowMs int64) error {
|
||
metadataJSON, err := json.Marshal(metadata)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
var paymentRef any
|
||
if metadata.PaymentRef != "" {
|
||
paymentRef = metadata.PaymentRef
|
||
}
|
||
_, err = tx.ExecContext(ctx,
|
||
`INSERT INTO coin_seller_stock_records (
|
||
app_code, stock_id, transaction_id, command_id, seller_user_id, stock_type,
|
||
counts_as_seller_recharge, coin_amount, paid_currency_code, paid_amount_micro,
|
||
payment_ref, evidence_ref, operator_user_id, reason, balance_after, metadata_json, created_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||
normalizedEntryAppCode(ctx, metadata.AppCode),
|
||
coinSellerStockID(metadata.AppCode, commandID),
|
||
transactionID,
|
||
commandID,
|
||
metadata.SellerUserID,
|
||
metadata.StockType,
|
||
metadata.CountsAsSellerRecharge,
|
||
metadata.CoinAmount,
|
||
metadata.PaidCurrencyCode,
|
||
metadata.PaidAmountMicro,
|
||
paymentRef,
|
||
metadata.EvidenceRef,
|
||
metadata.OperatorUserID,
|
||
metadata.Reason,
|
||
metadata.BalanceAfter,
|
||
string(metadataJSON),
|
||
nowMs,
|
||
)
|
||
if err != nil && isMySQLDuplicateError(err) && metadata.PaymentRef != "" {
|
||
return xerr.New(xerr.CoinSellerPaymentRefDuplicated, "payment_ref is duplicated")
|
||
}
|
||
return err
|
||
}
|
||
|
||
func (r *Repository) coinSellerStockPaymentRefExists(ctx context.Context, tx *sql.Tx, stockType string, paymentRef string) (bool, error) {
|
||
var stockID string
|
||
err := tx.QueryRowContext(ctx,
|
||
`SELECT stock_id
|
||
FROM coin_seller_stock_records
|
||
WHERE app_code = ? AND stock_type = ? AND payment_ref = ?
|
||
LIMIT 1`,
|
||
appcode.FromContext(ctx),
|
||
stockType,
|
||
paymentRef,
|
||
).Scan(&stockID)
|
||
if err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return false, nil
|
||
}
|
||
return false, err
|
||
}
|
||
return true, nil
|
||
}
|
||
|
||
type walletEntry struct {
|
||
AppCode string
|
||
TransactionID string
|
||
UserID int64
|
||
AssetType string
|
||
AvailableDelta int64
|
||
FrozenDelta int64
|
||
AvailableAfter int64
|
||
FrozenAfter int64
|
||
CounterpartyUserID int64
|
||
RoomID string
|
||
CreatedAtMS int64
|
||
}
|
||
|
||
func (r *Repository) insertEntry(ctx context.Context, tx *sql.Tx, entry walletEntry) error {
|
||
_, err := tx.ExecContext(ctx,
|
||
`INSERT INTO wallet_entries (
|
||
app_code, transaction_id, user_id, asset_type, available_delta, frozen_delta,
|
||
available_after, frozen_after, counterparty_user_id, room_id, created_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||
normalizedEntryAppCode(ctx, entry.AppCode),
|
||
entry.TransactionID,
|
||
entry.UserID,
|
||
entry.AssetType,
|
||
entry.AvailableDelta,
|
||
entry.FrozenDelta,
|
||
entry.AvailableAfter,
|
||
entry.FrozenAfter,
|
||
entry.CounterpartyUserID,
|
||
entry.RoomID,
|
||
entry.CreatedAtMS,
|
||
)
|
||
return err
|
||
}
|
||
|
||
type walletOutboxEvent struct {
|
||
AppCode string
|
||
EventID string
|
||
EventType string
|
||
TransactionID string
|
||
CommandID string
|
||
UserID int64
|
||
AssetType string
|
||
AvailableDelta int64
|
||
FrozenDelta int64
|
||
Payload any
|
||
CreatedAtMS int64
|
||
}
|
||
|
||
func (r *Repository) insertWalletOutbox(ctx context.Context, tx *sql.Tx, events []walletOutboxEvent) error {
|
||
for _, event := range events {
|
||
payloadJSON, err := json.Marshal(event.Payload)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx,
|
||
`INSERT INTO wallet_outbox (
|
||
app_code, event_id, event_type, transaction_id, command_id, user_id, asset_type,
|
||
available_delta, frozen_delta, payload, status, retry_count, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)`,
|
||
normalizedEntryAppCode(ctx, event.AppCode),
|
||
event.EventID,
|
||
event.EventType,
|
||
event.TransactionID,
|
||
event.CommandID,
|
||
event.UserID,
|
||
event.AssetType,
|
||
event.AvailableDelta,
|
||
event.FrozenDelta,
|
||
string(payloadJSON),
|
||
outboxStatusPending,
|
||
event.CreatedAtMS,
|
||
event.CreatedAtMS,
|
||
); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ClaimPendingWalletOutbox atomically claims committed wallet facts for MQ fanout.
|
||
func (r *Repository) ClaimPendingWalletOutbox(ctx context.Context, workerID string, limit int, lockUntilMS int64) ([]WalletOutboxRecord, error) {
|
||
if r == nil || r.db == nil {
|
||
return nil, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
workerID = strings.TrimSpace(workerID)
|
||
if workerID == "" {
|
||
return nil, xerr.New(xerr.InvalidArgument, "worker_id is required")
|
||
}
|
||
if limit <= 0 {
|
||
limit = 100
|
||
}
|
||
if limit > 500 {
|
||
limit = 500
|
||
}
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
if lockUntilMS <= nowMS {
|
||
lockUntilMS = time.Now().UTC().Add(30 * time.Second).UnixMilli()
|
||
}
|
||
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
rows, err := tx.QueryContext(ctx, `
|
||
SELECT app_code, event_id, event_type, transaction_id, command_id, user_id, asset_type,
|
||
available_delta, frozen_delta, COALESCE(CAST(payload AS CHAR), '{}'), created_at_ms,
|
||
retry_count, COALESCE(last_error, ''), COALESCE(worker_id, ''), COALESCE(lock_until_ms, 0)
|
||
FROM wallet_outbox
|
||
WHERE status IN (?, ?)
|
||
AND (next_retry_at_ms IS NULL OR next_retry_at_ms <= ?)
|
||
AND (lock_until_ms IS NULL OR lock_until_ms <= ?)
|
||
ORDER BY created_at_ms ASC, event_id ASC
|
||
LIMIT ?
|
||
FOR UPDATE`,
|
||
outboxStatusPending,
|
||
outboxStatusRetryable,
|
||
nowMS,
|
||
nowMS,
|
||
limit,
|
||
)
|
||
if err != nil {
|
||
_ = tx.Rollback()
|
||
return nil, err
|
||
}
|
||
|
||
records := make([]WalletOutboxRecord, 0, limit)
|
||
eventIDs := make([]string, 0, limit)
|
||
for rows.Next() {
|
||
var record WalletOutboxRecord
|
||
if err := rows.Scan(
|
||
&record.AppCode,
|
||
&record.EventID,
|
||
&record.EventType,
|
||
&record.TransactionID,
|
||
&record.CommandID,
|
||
&record.UserID,
|
||
&record.AssetType,
|
||
&record.AvailableDelta,
|
||
&record.FrozenDelta,
|
||
&record.PayloadJSON,
|
||
&record.CreatedAtMS,
|
||
&record.RetryCount,
|
||
&record.LastError,
|
||
&record.WorkerID,
|
||
&record.LockUntilMS,
|
||
); err != nil {
|
||
_ = rows.Close()
|
||
_ = tx.Rollback()
|
||
return nil, err
|
||
}
|
||
records = append(records, record)
|
||
eventIDs = append(eventIDs, record.EventID)
|
||
}
|
||
if err := rows.Close(); err != nil {
|
||
_ = tx.Rollback()
|
||
return nil, err
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
_ = tx.Rollback()
|
||
return nil, err
|
||
}
|
||
|
||
for index, eventID := range eventIDs {
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE wallet_outbox
|
||
SET status = ?, worker_id = ?, lock_until_ms = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND event_id = ?`,
|
||
outboxStatusDelivering,
|
||
workerID,
|
||
lockUntilMS,
|
||
nowMS,
|
||
records[index].AppCode,
|
||
eventID,
|
||
); err != nil {
|
||
_ = tx.Rollback()
|
||
return nil, err
|
||
}
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return nil, err
|
||
}
|
||
for index := range records {
|
||
records[index].WorkerID = workerID
|
||
records[index].LockUntilMS = lockUntilMS
|
||
}
|
||
return records, nil
|
||
}
|
||
|
||
// MarkWalletOutboxDelivered marks one wallet fact as accepted by the MQ broker.
|
||
func (r *Repository) MarkWalletOutboxDelivered(ctx context.Context, eventID string) error {
|
||
if r == nil || r.db == nil {
|
||
return xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
_, err := r.db.ExecContext(ctx, `
|
||
UPDATE wallet_outbox
|
||
SET status = ?, worker_id = '', lock_until_ms = NULL, next_retry_at_ms = NULL, last_error = NULL, updated_at_ms = ?
|
||
WHERE app_code = ? AND event_id = ?`,
|
||
outboxStatusDelivered,
|
||
time.Now().UTC().UnixMilli(),
|
||
appcode.FromContext(ctx),
|
||
eventID,
|
||
)
|
||
return err
|
||
}
|
||
|
||
// MarkWalletOutboxRetryable releases one failed MQ publish with a bounded retry schedule.
|
||
func (r *Repository) MarkWalletOutboxRetryable(ctx context.Context, eventID string, lastErr string, nextRetryAtMS int64) error {
|
||
if r == nil || r.db == nil {
|
||
return xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
_, err := r.db.ExecContext(ctx, `
|
||
UPDATE wallet_outbox
|
||
SET status = ?, worker_id = '', lock_until_ms = NULL, retry_count = retry_count + 1, next_retry_at_ms = ?, last_error = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND event_id = ?`,
|
||
outboxStatusRetryable,
|
||
nextRetryAtMS,
|
||
trimWalletOutboxError(lastErr),
|
||
time.Now().UTC().UnixMilli(),
|
||
appcode.FromContext(ctx),
|
||
eventID,
|
||
)
|
||
return err
|
||
}
|
||
|
||
// MarkWalletOutboxDead stops retrying a poison wallet fact after the configured retry ceiling.
|
||
func (r *Repository) MarkWalletOutboxDead(ctx context.Context, eventID string, lastErr string) error {
|
||
if r == nil || r.db == nil {
|
||
return xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
_, err := r.db.ExecContext(ctx, `
|
||
UPDATE wallet_outbox
|
||
SET status = ?, worker_id = '', lock_until_ms = NULL, last_error = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND event_id = ?`,
|
||
outboxStatusFailed,
|
||
trimWalletOutboxError(lastErr),
|
||
time.Now().UTC().UnixMilli(),
|
||
appcode.FromContext(ctx),
|
||
eventID,
|
||
)
|
||
return err
|
||
}
|
||
|
||
func trimWalletOutboxError(value string) string {
|
||
value = strings.TrimSpace(value)
|
||
if len(value) > 1024 {
|
||
return value[:1024]
|
||
}
|
||
return value
|
||
}
|
||
|
||
func (r *Repository) receiptForGiftTransaction(ctx context.Context, tx *sql.Tx, transactionID string) (ledger.Receipt, error) {
|
||
var metadataJSON string
|
||
if err := tx.QueryRowContext(ctx,
|
||
`SELECT COALESCE(CAST(metadata_json AS CHAR), '{}')
|
||
FROM wallet_transactions
|
||
WHERE app_code = ? AND transaction_id = ?`,
|
||
appcode.FromContext(ctx),
|
||
transactionID,
|
||
).Scan(&metadataJSON); err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
var metadata giftMetadata
|
||
if err := json.Unmarshal([]byte(metadataJSON), &metadata); err != nil {
|
||
return ledger.Receipt{}, err
|
||
}
|
||
return receiptFromGiftMetadata(transactionID, metadata), nil
|
||
}
|
||
|
||
func (r *Repository) balanceAfterTransaction(ctx context.Context, tx *sql.Tx, transactionID string, userID int64, assetType string) (ledger.AssetBalance, error) {
|
||
row := tx.QueryRowContext(ctx,
|
||
`SELECT app_code, user_id, asset_type, available_after, frozen_after
|
||
FROM wallet_entries
|
||
WHERE app_code = ? AND transaction_id = ? AND user_id = ? AND asset_type = ?
|
||
ORDER BY entry_id DESC LIMIT 1`,
|
||
appcode.FromContext(ctx), transactionID, userID, assetType,
|
||
)
|
||
var balance ledger.AssetBalance
|
||
if err := row.Scan(&balance.AppCode, &balance.UserID, &balance.AssetType, &balance.AvailableAmount, &balance.FrozenAmount); err != nil {
|
||
return ledger.AssetBalance{}, err
|
||
}
|
||
return balance, nil
|
||
}
|
||
|
||
type giftMetadata struct {
|
||
AppCode string `json:"app_code"`
|
||
GiftID string `json:"gift_id"`
|
||
GiftName string `json:"gift_name"`
|
||
ResourceID int64 `json:"resource_id"`
|
||
ResourceSnapshot string `json:"resource_snapshot_json"`
|
||
GiftTypeCode string `json:"gift_type_code"`
|
||
PresentationJSON string `json:"presentation_json"`
|
||
SortOrder int32 `json:"sort_order"`
|
||
GiftCount int32 `json:"gift_count"`
|
||
PriceVersion string `json:"price_version"`
|
||
CoinPrice int64 `json:"coin_price"`
|
||
GiftPointAmount int64 `json:"gift_point_amount"`
|
||
HeatUnitValue int64 `json:"heat_unit_value"`
|
||
ChargeAssetType string `json:"charge_asset_type"`
|
||
ChargeAmount int64 `json:"charge_amount"`
|
||
CoinSpent int64 `json:"coin_spent"`
|
||
GiftPointAdded int64 `json:"gift_point_added"`
|
||
HeatValue int64 `json:"heat_value"`
|
||
BalanceAfter int64 `json:"balance_after"`
|
||
BillingReceipt string `json:"billing_receipt_id"`
|
||
SenderUserID int64 `json:"sender_user_id"`
|
||
TargetUserID int64 `json:"target_user_id"`
|
||
RoomID string `json:"room_id"`
|
||
}
|
||
|
||
type adminCreditMetadata struct {
|
||
AppCode string `json:"app_code"`
|
||
TargetUserID int64 `json:"target_user_id"`
|
||
AssetType string `json:"asset_type"`
|
||
Amount int64 `json:"amount"`
|
||
OperatorUserID int64 `json:"operator_user_id"`
|
||
Reason string `json:"reason"`
|
||
EvidenceRef string `json:"evidence_ref"`
|
||
}
|
||
|
||
type taskRewardMetadata struct {
|
||
AppCode string `json:"app_code"`
|
||
TargetUserID int64 `json:"target_user_id"`
|
||
AssetType string `json:"asset_type"`
|
||
Amount int64 `json:"amount"`
|
||
TaskType string `json:"task_type"`
|
||
TaskID string `json:"task_id"`
|
||
CycleKey string `json:"cycle_key"`
|
||
Reason string `json:"reason"`
|
||
BalanceAfter int64 `json:"balance_after"`
|
||
GrantedAtMS int64 `json:"granted_at_ms"`
|
||
}
|
||
|
||
type gameCoinMetadata struct {
|
||
AppCode string `json:"app_code"`
|
||
UserID int64 `json:"user_id"`
|
||
PlatformCode string `json:"platform_code"`
|
||
GameID string `json:"game_id"`
|
||
ProviderOrderID string `json:"provider_order_id"`
|
||
ProviderRoundID string `json:"provider_round_id"`
|
||
OpType string `json:"op_type"`
|
||
AssetType string `json:"asset_type"`
|
||
CoinAmount int64 `json:"coin_amount"`
|
||
AvailableDelta int64 `json:"available_delta"`
|
||
RoomID string `json:"room_id"`
|
||
BalanceAfter int64 `json:"balance_after"`
|
||
AppliedAtMS int64 `json:"applied_at_ms"`
|
||
}
|
||
|
||
type coinSellerTransferMetadata struct {
|
||
AppCode string `json:"app_code"`
|
||
SellerUserID int64 `json:"seller_user_id"`
|
||
TargetUserID int64 `json:"target_user_id"`
|
||
SellerRegionID int64 `json:"seller_region_id"`
|
||
TargetRegionID int64 `json:"target_region_id"`
|
||
Amount int64 `json:"amount"`
|
||
Reason string `json:"reason"`
|
||
SellerAssetType string `json:"seller_asset_type"`
|
||
TargetAssetType string `json:"target_asset_type"`
|
||
SellerBalanceAfter int64 `json:"seller_balance_after"`
|
||
TargetBalanceAfter int64 `json:"target_balance_after"`
|
||
RechargeSequence int64 `json:"recharge_sequence"`
|
||
RechargeUSDMinor int64 `json:"recharge_usd_minor"`
|
||
RechargeCurrencyCode string `json:"recharge_currency_code"`
|
||
RechargePolicyID int64 `json:"recharge_policy_id"`
|
||
RechargePolicyVersion string `json:"recharge_policy_version"`
|
||
RechargePolicyCoinAmount int64 `json:"recharge_policy_coin_amount"`
|
||
RechargePolicyUSDMinorAmount int64 `json:"recharge_policy_usd_minor_amount"`
|
||
}
|
||
|
||
type coinSellerStockMetadata struct {
|
||
AppCode string `json:"app_code"`
|
||
SellerUserID int64 `json:"seller_user_id"`
|
||
StockType string `json:"stock_type"`
|
||
CoinAmount int64 `json:"coin_amount"`
|
||
PaidCurrencyCode string `json:"paid_currency_code"`
|
||
PaidAmountMicro int64 `json:"paid_amount_micro"`
|
||
PaymentRef string `json:"payment_ref"`
|
||
EvidenceRef string `json:"evidence_ref"`
|
||
OperatorUserID int64 `json:"operator_user_id"`
|
||
Reason string `json:"reason"`
|
||
AssetType string `json:"asset_type"`
|
||
CountsAsSellerRecharge bool `json:"counts_as_seller_recharge"`
|
||
BalanceAfter int64 `json:"balance_after"`
|
||
CreatedAtMS int64 `json:"created_at_ms"`
|
||
}
|
||
|
||
func receiptFromGiftMetadata(transactionID string, metadata giftMetadata) ledger.Receipt {
|
||
chargeAssetType := ledger.NormalizeGiftChargeAssetType(metadata.ChargeAssetType)
|
||
chargeAmount := metadata.ChargeAmount
|
||
if chargeAmount == 0 {
|
||
chargeAmount = metadata.CoinSpent
|
||
}
|
||
return ledger.Receipt{
|
||
BillingReceiptID: metadata.BillingReceipt,
|
||
TransactionID: transactionID,
|
||
CoinSpent: metadata.CoinSpent,
|
||
ChargeAssetType: chargeAssetType,
|
||
ChargeAmount: chargeAmount,
|
||
GiftPointAdded: metadata.GiftPointAdded,
|
||
HeatValue: metadata.HeatValue,
|
||
GiftTypeCode: metadata.GiftTypeCode,
|
||
PriceVersion: metadata.PriceVersion,
|
||
BalanceAfter: metadata.BalanceAfter,
|
||
}
|
||
}
|
||
|
||
func (r *Repository) receiptForTaskRewardTransaction(ctx context.Context, tx *sql.Tx, transactionID string) (ledger.TaskRewardReceipt, error) {
|
||
var metadataJSON string
|
||
if err := tx.QueryRowContext(ctx,
|
||
`SELECT COALESCE(CAST(metadata_json AS CHAR), '{}')
|
||
FROM wallet_transactions
|
||
WHERE app_code = ? AND transaction_id = ?`,
|
||
appcode.FromContext(ctx),
|
||
transactionID,
|
||
).Scan(&metadataJSON); err != nil {
|
||
return ledger.TaskRewardReceipt{}, err
|
||
}
|
||
var metadata taskRewardMetadata
|
||
if err := json.Unmarshal([]byte(metadataJSON), &metadata); err != nil {
|
||
return ledger.TaskRewardReceipt{}, err
|
||
}
|
||
return receiptFromTaskRewardMetadata(transactionID, metadata), nil
|
||
}
|
||
|
||
func receiptFromTaskRewardMetadata(transactionID string, metadata taskRewardMetadata) ledger.TaskRewardReceipt {
|
||
return ledger.TaskRewardReceipt{
|
||
TransactionID: transactionID,
|
||
Balance: ledger.AssetBalance{
|
||
AppCode: metadata.AppCode,
|
||
UserID: metadata.TargetUserID,
|
||
AssetType: metadata.AssetType,
|
||
AvailableAmount: metadata.BalanceAfter,
|
||
},
|
||
Amount: metadata.Amount,
|
||
GrantedAtMS: metadata.GrantedAtMS,
|
||
}
|
||
}
|
||
|
||
func (r *Repository) receiptForGameTransaction(ctx context.Context, tx *sql.Tx, transactionID string) (ledger.GameCoinChangeReceipt, error) {
|
||
var metadataJSON string
|
||
if err := tx.QueryRowContext(ctx,
|
||
`SELECT COALESCE(CAST(metadata_json AS CHAR), '{}')
|
||
FROM wallet_transactions
|
||
WHERE app_code = ? AND transaction_id = ?`,
|
||
appcode.FromContext(ctx),
|
||
transactionID,
|
||
).Scan(&metadataJSON); err != nil {
|
||
return ledger.GameCoinChangeReceipt{}, err
|
||
}
|
||
var metadata gameCoinMetadata
|
||
if err := json.Unmarshal([]byte(metadataJSON), &metadata); err != nil {
|
||
return ledger.GameCoinChangeReceipt{}, err
|
||
}
|
||
return receiptFromGameMetadata(transactionID, metadata, false), nil
|
||
}
|
||
|
||
func receiptFromGameMetadata(transactionID string, metadata gameCoinMetadata, idempotentReplay bool) ledger.GameCoinChangeReceipt {
|
||
return ledger.GameCoinChangeReceipt{
|
||
TransactionID: transactionID,
|
||
BalanceAfter: metadata.BalanceAfter,
|
||
IdempotentReplay: idempotentReplay,
|
||
}
|
||
}
|
||
|
||
func (r *Repository) receiptForCoinSellerStockTransaction(ctx context.Context, tx *sql.Tx, transactionID string) (ledger.CoinSellerStockCreditReceipt, error) {
|
||
row := tx.QueryRowContext(ctx,
|
||
`SELECT seller_user_id, stock_type, coin_amount, paid_currency_code, paid_amount_micro,
|
||
counts_as_seller_recharge, balance_after, created_at_ms
|
||
FROM coin_seller_stock_records
|
||
WHERE app_code = ? AND transaction_id = ?`,
|
||
appcode.FromContext(ctx),
|
||
transactionID,
|
||
)
|
||
var receipt ledger.CoinSellerStockCreditReceipt
|
||
if err := row.Scan(
|
||
&receipt.SellerUserID,
|
||
&receipt.StockType,
|
||
&receipt.CoinAmount,
|
||
&receipt.PaidCurrencyCode,
|
||
&receipt.PaidAmountMicro,
|
||
&receipt.CountsAsSellerRecharge,
|
||
&receipt.BalanceAfter,
|
||
&receipt.CreatedAtMS,
|
||
); err != nil {
|
||
return ledger.CoinSellerStockCreditReceipt{}, err
|
||
}
|
||
receipt.TransactionID = transactionID
|
||
return receipt, nil
|
||
}
|
||
|
||
func receiptFromCoinSellerStockMetadata(transactionID string, metadata coinSellerStockMetadata) ledger.CoinSellerStockCreditReceipt {
|
||
return ledger.CoinSellerStockCreditReceipt{
|
||
TransactionID: transactionID,
|
||
SellerUserID: metadata.SellerUserID,
|
||
StockType: metadata.StockType,
|
||
CoinAmount: metadata.CoinAmount,
|
||
PaidCurrencyCode: metadata.PaidCurrencyCode,
|
||
PaidAmountMicro: metadata.PaidAmountMicro,
|
||
CountsAsSellerRecharge: metadata.CountsAsSellerRecharge,
|
||
BalanceAfter: metadata.BalanceAfter,
|
||
CreatedAtMS: metadata.CreatedAtMS,
|
||
}
|
||
}
|
||
|
||
func (r *Repository) receiptForCoinSellerTransferTransaction(ctx context.Context, tx *sql.Tx, transactionID string) (ledger.CoinSellerTransferReceipt, error) {
|
||
var metadataJSON string
|
||
if err := tx.QueryRowContext(ctx,
|
||
`SELECT COALESCE(CAST(metadata_json AS CHAR), '{}')
|
||
FROM wallet_transactions
|
||
WHERE app_code = ? AND transaction_id = ?`,
|
||
appcode.FromContext(ctx),
|
||
transactionID,
|
||
).Scan(&metadataJSON); err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
var metadata coinSellerTransferMetadata
|
||
if err := json.Unmarshal([]byte(metadataJSON), &metadata); err != nil {
|
||
return ledger.CoinSellerTransferReceipt{}, err
|
||
}
|
||
return receiptFromCoinSellerTransferMetadata(transactionID, metadata), nil
|
||
}
|
||
|
||
func receiptFromCoinSellerTransferMetadata(transactionID string, metadata coinSellerTransferMetadata) ledger.CoinSellerTransferReceipt {
|
||
return ledger.CoinSellerTransferReceipt{
|
||
TransactionID: transactionID,
|
||
SellerBalanceAfter: metadata.SellerBalanceAfter,
|
||
TargetBalanceAfter: metadata.TargetBalanceAfter,
|
||
Amount: metadata.Amount,
|
||
RechargeSequence: metadata.RechargeSequence,
|
||
RechargeUSDMinor: metadata.RechargeUSDMinor,
|
||
RechargeCurrencyCode: metadata.RechargeCurrencyCode,
|
||
RechargePolicyID: metadata.RechargePolicyID,
|
||
RechargePolicyVersion: metadata.RechargePolicyVersion,
|
||
RechargePolicyCoinAmount: metadata.RechargePolicyCoinAmount,
|
||
RechargePolicyUSDMinorUnit: metadata.RechargePolicyUSDMinorAmount,
|
||
}
|
||
}
|
||
|
||
func balanceChangedEvent(transactionID string, commandID string, userID int64, assetType string, availableDelta int64, frozenDelta int64, availableAfter int64, frozenAfter int64, balanceVersion int64, payload any, nowMs int64) walletOutboxEvent {
|
||
return walletOutboxEvent{
|
||
EventID: eventID(transactionID, "WalletBalanceChanged", userID, assetType),
|
||
EventType: "WalletBalanceChanged",
|
||
TransactionID: transactionID,
|
||
CommandID: commandID,
|
||
UserID: userID,
|
||
AssetType: assetType,
|
||
AvailableDelta: availableDelta,
|
||
FrozenDelta: frozenDelta,
|
||
Payload: map[string]any{
|
||
"transaction_id": transactionID,
|
||
"command_id": commandID,
|
||
"user_id": userID,
|
||
"asset_type": assetType,
|
||
"available_delta": availableDelta,
|
||
"frozen_delta": frozenDelta,
|
||
"available_after": availableAfter,
|
||
"frozen_after": frozenAfter,
|
||
"balance_version": balanceVersion,
|
||
"metadata": payload,
|
||
"created_at_ms": nowMs,
|
||
},
|
||
CreatedAtMS: nowMs,
|
||
}
|
||
}
|
||
|
||
func coinSellerTransferredEvent(transactionID string, commandID string, sellerUserID int64, availableDelta int64, payload any, nowMs int64) walletOutboxEvent {
|
||
return walletOutboxEvent{
|
||
EventID: eventID(transactionID, "WalletCoinSellerTransferred", sellerUserID, ledger.AssetCoinSellerCoin),
|
||
EventType: "WalletCoinSellerTransferred",
|
||
TransactionID: transactionID,
|
||
CommandID: commandID,
|
||
UserID: sellerUserID,
|
||
AssetType: ledger.AssetCoinSellerCoin,
|
||
AvailableDelta: availableDelta,
|
||
FrozenDelta: 0,
|
||
Payload: payload,
|
||
CreatedAtMS: nowMs,
|
||
}
|
||
}
|
||
|
||
func rechargeRecordedEvent(transactionID string, commandID string, targetUserID int64, availableDelta int64, payload any, nowMs int64) walletOutboxEvent {
|
||
return walletOutboxEvent{
|
||
EventID: eventID(transactionID, "WalletRechargeRecorded", targetUserID, ledger.AssetCoin),
|
||
EventType: "WalletRechargeRecorded",
|
||
TransactionID: transactionID,
|
||
CommandID: commandID,
|
||
UserID: targetUserID,
|
||
AssetType: ledger.AssetCoin,
|
||
AvailableDelta: availableDelta,
|
||
FrozenDelta: 0,
|
||
Payload: payload,
|
||
CreatedAtMS: nowMs,
|
||
}
|
||
}
|
||
|
||
func taskRewardCreditedEvent(transactionID string, commandID string, metadata taskRewardMetadata, nowMs int64) walletOutboxEvent {
|
||
return walletOutboxEvent{
|
||
EventID: eventID(transactionID, "WalletTaskRewardCredited", metadata.TargetUserID, ledger.AssetCoin),
|
||
EventType: "WalletTaskRewardCredited",
|
||
TransactionID: transactionID,
|
||
CommandID: commandID,
|
||
UserID: metadata.TargetUserID,
|
||
AssetType: ledger.AssetCoin,
|
||
AvailableDelta: metadata.Amount,
|
||
FrozenDelta: 0,
|
||
Payload: map[string]any{
|
||
"transaction_id": transactionID,
|
||
"command_id": commandID,
|
||
"user_id": metadata.TargetUserID,
|
||
"task_type": metadata.TaskType,
|
||
"task_id": metadata.TaskID,
|
||
"cycle_key": metadata.CycleKey,
|
||
"amount": metadata.Amount,
|
||
"reason": metadata.Reason,
|
||
"created_at_ms": nowMs,
|
||
},
|
||
CreatedAtMS: nowMs,
|
||
}
|
||
}
|
||
|
||
func coinSellerStockCreditedEvent(transactionID string, commandID string, metadata coinSellerStockMetadata, nowMs int64) walletOutboxEvent {
|
||
eventType := "WalletCoinSellerCoinCompensated"
|
||
if metadata.StockType == ledger.StockTypeUSDTPurchase {
|
||
eventType = "WalletCoinSellerStockPurchased"
|
||
}
|
||
return walletOutboxEvent{
|
||
EventID: eventID(transactionID, eventType, metadata.SellerUserID, ledger.AssetCoinSellerCoin),
|
||
EventType: eventType,
|
||
TransactionID: transactionID,
|
||
CommandID: commandID,
|
||
UserID: metadata.SellerUserID,
|
||
AssetType: ledger.AssetCoinSellerCoin,
|
||
AvailableDelta: metadata.CoinAmount,
|
||
FrozenDelta: 0,
|
||
Payload: map[string]any{
|
||
"transaction_id": transactionID,
|
||
"command_id": commandID,
|
||
"seller_user_id": metadata.SellerUserID,
|
||
"stock_type": metadata.StockType,
|
||
"coin_amount": metadata.CoinAmount,
|
||
"paid_currency_code": metadata.PaidCurrencyCode,
|
||
"paid_amount_micro": metadata.PaidAmountMicro,
|
||
"counts_as_seller_recharge": metadata.CountsAsSellerRecharge,
|
||
"operator_user_id": metadata.OperatorUserID,
|
||
"created_at_ms": nowMs,
|
||
},
|
||
CreatedAtMS: nowMs,
|
||
}
|
||
}
|
||
|
||
func giftDebitedEvent(transactionID string, commandID string, userID int64, assetType string, availableDelta int64, frozenDelta int64, payload any, nowMs int64) walletOutboxEvent {
|
||
return walletOutboxEvent{
|
||
EventID: eventID(transactionID, "WalletGiftDebited", userID, assetType),
|
||
EventType: "WalletGiftDebited",
|
||
TransactionID: transactionID,
|
||
CommandID: commandID,
|
||
UserID: userID,
|
||
AssetType: assetType,
|
||
AvailableDelta: availableDelta,
|
||
FrozenDelta: frozenDelta,
|
||
Payload: payload,
|
||
CreatedAtMS: nowMs,
|
||
}
|
||
}
|
||
|
||
func normalizeAssetTypes(assetTypes []string) []string {
|
||
seen := make(map[string]bool, len(assetTypes))
|
||
result := make([]string, 0, len(assetTypes))
|
||
for _, assetType := range assetTypes {
|
||
assetType = strings.ToUpper(strings.TrimSpace(assetType))
|
||
if assetType == "" || seen[assetType] {
|
||
continue
|
||
}
|
||
seen[assetType] = true
|
||
result = append(result, assetType)
|
||
}
|
||
return result
|
||
}
|
||
|
||
func contextWithCommandApp(ctx context.Context, value string) context.Context {
|
||
if strings.TrimSpace(value) == "" {
|
||
return appcode.WithContext(ctx, appcode.FromContext(ctx))
|
||
}
|
||
|
||
return appcode.WithContext(ctx, value)
|
||
}
|
||
|
||
func normalizedEntryAppCode(ctx context.Context, value string) string {
|
||
if strings.TrimSpace(value) == "" {
|
||
return appcode.FromContext(ctx)
|
||
}
|
||
|
||
return appcode.Normalize(value)
|
||
}
|
||
|
||
func checkedMul(unit int64, count int64) (int64, error) {
|
||
if unit < 0 || count <= 0 {
|
||
return 0, xerr.New(xerr.InvalidArgument, "amount is invalid")
|
||
}
|
||
if unit != 0 && count > math.MaxInt64/unit {
|
||
return 0, xerr.New(xerr.InvalidArgument, "amount overflow")
|
||
}
|
||
return unit * count, nil
|
||
}
|
||
|
||
func checkedAdd(left int64, right int64) (int64, error) {
|
||
if right < 0 || left > math.MaxInt64-right {
|
||
return 0, xerr.New(xerr.CoinSellerStockAmountInvalid, "coin amount overflow")
|
||
}
|
||
return left + right, nil
|
||
}
|
||
|
||
func debitRequestHash(command ledger.DebitGiftCommand) string {
|
||
// 哈希覆盖所有账务语义字段；相同 command_id 只能重放完全相同的业务命令。
|
||
return stableHash(fmt.Sprintf("gift|%s|%s|%d|%d|%s|%d|%s",
|
||
appcode.Normalize(command.AppCode),
|
||
command.RoomID,
|
||
command.SenderUserID,
|
||
command.TargetUserID,
|
||
command.GiftID,
|
||
command.GiftCount,
|
||
strings.TrimSpace(command.PriceVersion),
|
||
))
|
||
}
|
||
|
||
func adminCreditRequestHash(command ledger.AdminCreditAssetCommand) string {
|
||
return stableHash(fmt.Sprintf("admin_credit|%s|%d|%s|%d|%d|%s|%s",
|
||
appcode.Normalize(command.AppCode),
|
||
command.TargetUserID,
|
||
command.AssetType,
|
||
command.Amount,
|
||
command.OperatorUserID,
|
||
command.Reason,
|
||
command.EvidenceRef,
|
||
))
|
||
}
|
||
|
||
func taskRewardRequestHash(command ledger.TaskRewardCommand) string {
|
||
return stableHash(fmt.Sprintf("task_reward|%s|%d|%d|%s|%s|%s|%s",
|
||
appcode.Normalize(command.AppCode),
|
||
command.TargetUserID,
|
||
command.Amount,
|
||
command.TaskType,
|
||
command.TaskID,
|
||
command.CycleKey,
|
||
strings.TrimSpace(command.Reason),
|
||
))
|
||
}
|
||
|
||
func gameBizTypeAndDelta(opType string, coinAmount int64) (string, int64, error) {
|
||
switch ledger.NormalizeGameOpType(opType) {
|
||
case ledger.GameOpDebit:
|
||
return bizTypeGameDebit, -coinAmount, nil
|
||
case ledger.GameOpCredit:
|
||
return bizTypeGameCredit, coinAmount, nil
|
||
case ledger.GameOpRefund:
|
||
return bizTypeGameRefund, coinAmount, nil
|
||
case ledger.GameOpReverse:
|
||
// 冲正首版按“从用户 COIN 扣回”处理；后续接入平台若语义不同，必须在 adapter 层拆成 debit/refund。
|
||
return bizTypeGameReverse, -coinAmount, nil
|
||
default:
|
||
return "", 0, xerr.New(xerr.InvalidArgument, "game op_type is invalid")
|
||
}
|
||
}
|
||
|
||
func coinSellerStockRequestHash(command ledger.CoinSellerStockCreditCommand) string {
|
||
return stableHash(fmt.Sprintf("coin_seller_stock|%s|%d|%s|%d|%s|%d|%s|%s|%d|%s",
|
||
appcode.Normalize(command.AppCode),
|
||
command.SellerUserID,
|
||
command.StockType,
|
||
command.CoinAmount,
|
||
command.PaidCurrencyCode,
|
||
command.PaidAmountMicro,
|
||
command.PaymentRef,
|
||
command.EvidenceRef,
|
||
command.OperatorUserID,
|
||
command.Reason,
|
||
))
|
||
}
|
||
|
||
func coinSellerTransferRequestHash(command ledger.CoinSellerTransferCommand) string {
|
||
return stableHash(fmt.Sprintf("coin_seller_transfer|%s|%d|%d|%d|%d|%d|%s",
|
||
appcode.Normalize(command.AppCode),
|
||
command.SellerUserID,
|
||
command.TargetUserID,
|
||
command.SellerRegionID,
|
||
command.TargetRegionID,
|
||
command.Amount,
|
||
strings.TrimSpace(command.Reason),
|
||
))
|
||
}
|
||
|
||
func coinSellerStockBizType(stockType string) string {
|
||
switch stockType {
|
||
case ledger.StockTypeUSDTPurchase:
|
||
return bizTypeCoinSellerStockPurchase
|
||
case ledger.StockTypeCoinCompensation:
|
||
return bizTypeCoinSellerCoinCompensation
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
func stableHash(value string) string {
|
||
sum := sha256.Sum256([]byte(value))
|
||
return hex.EncodeToString(sum[:])
|
||
}
|
||
|
||
func mustJSON(value any) string {
|
||
body, err := json.Marshal(value)
|
||
if err != nil {
|
||
return "{}"
|
||
}
|
||
return string(body)
|
||
}
|
||
|
||
func transactionID(appCode string, commandID string) string {
|
||
return "wtx_" + stableHash(appcode.Normalize(appCode)+"|"+commandID)
|
||
}
|
||
|
||
func coinSellerStockID(appCode string, commandID string) string {
|
||
return "cstock_" + stableHash(appcode.Normalize(appCode)+"|"+commandID)
|
||
}
|
||
|
||
func billingReceiptID(appCode string, commandID string) string {
|
||
return "bill_" + stableHash(appcode.Normalize(appCode)+"|"+commandID)
|
||
}
|
||
|
||
func eventID(transactionID string, eventType string, userID int64, assetType string) string {
|
||
return "wev_" + stableHash(fmt.Sprintf("%s|%s|%d|%s", transactionID, eventType, userID, assetType))
|
||
}
|
||
|
||
func isMySQLDuplicateError(err error) bool {
|
||
var mysqlErr *mysqlDriver.MySQLError
|
||
return errors.As(err, &mysqlErr) && mysqlErr.Number == 1062
|
||
}
|