3446 lines
132 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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"
bizTypeLuckyGiftReward = "lucky_gift_reward"
bizTypeCoinSellerTransfer = "coin_seller_transfer"
bizTypeCoinSellerStockPurchase = "coin_seller_stock_purchase"
bizTypeCoinSellerCoinCompensation = "coin_seller_coin_compensation"
bizTypeSalaryExchangeToCoin = "salary_exchange_to_coin"
bizTypeSalaryTransferToCoinSeller = "salary_transfer_to_coin_seller"
bizTypeGooglePlayRecharge = "google_play_recharge"
bizTypeGameDebit = "game_debit"
bizTypeGameCredit = "game_credit"
bizTypeGameRefund = "game_refund"
bizTypeGameReverse = "game_reverse"
bizTypeHostSalarySettlement = "host_salary_settlement"
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
}
// WalletOutboxClaimFilter narrows one MQ fanout lane by event type.
type WalletOutboxClaimFilter struct {
// IncludeEventTypes lets a realtime worker own only the configured UI event facts.
IncludeEventTypes []string
// ExcludeEventTypes lets the generic wallet worker skip facts owned by the realtime lane.
ExcludeEventTypes []string
}
// 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
}
if err := ensureGiftDiamondRatioSchema(ctx, db); err != nil {
_ = db.Close()
return nil, err
}
if err := ensureCoinSellerSalaryExchangeRateSchema(ctx, db); err != nil {
_ = db.Close()
return nil, err
}
return &Repository{db: db}, nil
}
func ensureGiftDiamondRatioSchema(ctx context.Context, db *sql.DB) error {
if _, err := db.ExecContext(ctx, `
CREATE TABLE IF NOT EXISTS gift_diamond_ratio_configs (
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码,用于多租户隔离',
region_id BIGINT NOT NULL DEFAULT 0 COMMENT '房间可见区域0 表示全局默认',
gift_type_code VARCHAR(32) NOT NULL COMMENT '礼物类型编码',
status VARCHAR(32) NOT NULL DEFAULT 'active' COMMENT '业务状态',
ratio_percent DECIMAL(5,2) NOT NULL DEFAULT 100.00 COMMENT '房间贡献热度和主播周期钻石入账比例,百分比',
created_by_admin_id BIGINT NOT NULL DEFAULT 0 COMMENT '创建后台用户 ID',
updated_by_admin_id BIGINT NOT NULL DEFAULT 0 COMMENT '更新后台用户 ID',
created_at_ms BIGINT NOT NULL COMMENT '创建时间UTC epoch ms',
updated_at_ms BIGINT NOT NULL COMMENT '更新时间UTC epoch ms',
PRIMARY KEY (app_code, region_id, gift_type_code),
KEY idx_gift_diamond_ratio_region (app_code, region_id, status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='礼物钻石和房间贡献比例配置表'`); err != nil {
return err
}
_, err := db.ExecContext(ctx, `
INSERT IGNORE INTO gift_diamond_ratio_configs (
app_code, region_id, gift_type_code, status, ratio_percent,
created_by_admin_id, updated_by_admin_id, created_at_ms, updated_at_ms
) VALUES
('lalu', 0, 'normal', 'active', 100.00, 0, 0, 0, 0),
('lalu', 0, 'lucky', 'active', 100.00, 0, 0, 0, 0),
('lalu', 0, 'super_lucky', 'active', 100.00, 0, 0, 0, 0)`)
return err
}
func ensureCoinSellerSalaryExchangeRateSchema(ctx context.Context, db *sql.DB) error {
_, err := db.ExecContext(ctx, `
CREATE TABLE IF NOT EXISTS coin_seller_salary_exchange_rate_tiers (
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码,用于多租户隔离',
region_id BIGINT NOT NULL COMMENT '币商所属区域 ID',
tier_id BIGINT NOT NULL AUTO_INCREMENT COMMENT '区间 ID',
min_usd_minor BIGINT NOT NULL COMMENT '起始美元金额,单位美分,包含',
max_usd_minor BIGINT NOT NULL COMMENT '结束美元金额,单位美分,包含',
coin_per_usd BIGINT NOT NULL COMMENT '每 1 USD 工资可兑换的币商专用金币数量',
status VARCHAR(24) NOT NULL DEFAULT 'active' COMMENT '状态active/disabled',
sort_order INT NOT NULL DEFAULT 0 COMMENT '排序权重',
created_by_user_id BIGINT NOT NULL DEFAULT 0 COMMENT '创建人用户 ID',
updated_by_user_id BIGINT NOT NULL DEFAULT 0 COMMENT '更新人用户 ID',
created_at_ms BIGINT NOT NULL COMMENT '创建时间UTC epoch ms',
updated_at_ms BIGINT NOT NULL COMMENT '更新时间UTC epoch ms',
PRIMARY KEY (tier_id),
KEY idx_coin_seller_salary_rates_region (app_code, region_id, status, min_usd_minor, max_usd_minor),
KEY idx_coin_seller_salary_rates_sort (app_code, region_id, status, sort_order)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='币商工资兑换比例区间表'`)
return err
}
// 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 扣费、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")
}
if command.SenderRegionID < 0 {
return ledger.Receipt{}, xerr.New(xerr.InvalidArgument, "sender_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
}
baseHeatValue, err := checkedMul(price.HeatValue, int64(command.GiftCount))
if err != nil {
return ledger.Receipt{}, err
}
roomContributionRatio, roomContributionRatioRegionID, err := r.resolveGiftDiamondRatio(ctx, tx, command.AppCode, command.RegionID, giftConfig.GiftTypeCode)
if err != nil {
return ledger.Receipt{}, err
}
// 房间贡献由房间 visible_region_id 命中后台礼物钻石比例room-service 只消费结算后的 heat_value不再自己查钱包配置。
heatValue, err := giftDiamondAmount(baseHeatValue, roomContributionRatio.PPM)
if err != nil {
return ledger.Receipt{}, err
}
hostPeriodDiamondAdded := int64(0)
hostPeriodCycleKey := ""
giftDiamondRatioPercent := "0.00"
giftDiamondRatioRegionID := int64(0)
if command.TargetIsHost {
ratio, ratioRegionID, err := r.resolveGiftDiamondRatio(ctx, tx, command.AppCode, command.SenderRegionID, giftConfig.GiftTypeCode)
if err != nil {
return ledger.Receipt{}, err
}
// 非主播不会进入该分支,因此普通用户收礼只增加 GIFT_POINT不产生工资周期积分。
hostPeriodDiamondAdded, err = giftDiamondAmount(chargeAmount, ratio.PPM)
if err != nil {
return ledger.Receipt{}, err
}
giftDiamondRatioPercent = ratio.Percent
giftDiamondRatioRegionID = ratioRegionID
hostPeriodCycleKey = hostPeriodCycleKeyFromMS(nowMs)
}
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,
SenderRegionID: command.SenderRegionID,
TargetUserID: command.TargetUserID,
TargetIsHost: command.TargetIsHost,
TargetHostRegionID: command.TargetHostRegionID,
TargetAgencyOwnerUserID: command.TargetAgencyOwnerUserID,
HostPeriodDiamondAdded: hostPeriodDiamondAdded,
GiftDiamondRatioPercent: giftDiamondRatioPercent,
GiftDiamondRatioRegionID: giftDiamondRatioRegionID,
HostPeriodCycleKey: hostPeriodCycleKey,
RoomID: command.RoomID,
RoomRegionID: command.RegionID,
RoomContributionRatioPercent: roomContributionRatio.Percent,
RoomContributionRatioRegionID: roomContributionRatioRegionID,
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 hostPeriodDiamondAdded > 0 {
// 周期钻石与送礼扣费同事务提交;如果后续任一步失败,用户扣币、主播礼物积分和工资积分会一起回滚。
hostPeriodDiamondAfter, hostPeriodDiamondVersion, err := r.creditHostPeriodDiamonds(ctx, tx, transactionID, command.CommandID, metadata, nowMs)
if err != nil {
return ledger.Receipt{}, err
}
metadata.HostPeriodDiamondAfter = hostPeriodDiamondAfter
metadata.HostPeriodDiamondVersion = hostPeriodDiamondVersion
}
events := []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),
}
if hostPeriodDiamondAdded > 0 {
events = append(events, hostPeriodDiamondCreditedEvent(transactionID, command.CommandID, metadata, nowMs))
}
if err := r.insertWalletOutbox(ctx, tx, events); err != nil {
return ledger.Receipt{}, err
}
if err := tx.Commit(); err != nil {
return ledger.Receipt{}, err
}
return receiptFromGiftMetadata(transactionID, metadata), nil
}
// BatchDebitGift 在一个 MySQL 事务内完成多目标送礼;任一目标失败时整批回滚,避免房间命令和账务出现部分成功。
func (r *Repository) BatchDebitGift(ctx context.Context, command ledger.BatchDebitGiftCommand) (ledger.BatchGiftReceipt, error) {
if r == nil || r.db == nil {
return ledger.BatchGiftReceipt{}, 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.BatchGiftReceipt{}, err
}
defer func() {
// Commit 成功后 Rollback 会返回 sql.ErrTxDone忽略它可保证所有早退路径释放事务。
_ = tx.Rollback()
}()
nowMs := time.Now().UnixMilli()
if command.RegionID < 0 {
return ledger.BatchGiftReceipt{}, xerr.New(xerr.InvalidArgument, "region_id is invalid")
}
if command.SenderRegionID < 0 {
return ledger.BatchGiftReceipt{}, xerr.New(xerr.InvalidArgument, "sender_region_id is invalid")
}
giftConfig, err := r.resolveActiveGiftConfig(ctx, tx, command.GiftID, command.RegionID)
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
price, err := r.resolveGiftPrice(ctx, tx, command.GiftID, command.PriceVersion, nowMs)
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
chargeAmount, err := checkedMul(price.CoinPrice, int64(command.GiftCount))
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
giftPointAdded, err := checkedMul(price.GiftPointAmount, int64(command.GiftCount))
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
baseHeatValue, err := checkedMul(price.HeatValue, int64(command.GiftCount))
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
roomContributionRatio, roomContributionRatioRegionID, err := r.resolveGiftDiamondRatio(ctx, tx, command.AppCode, command.RegionID, giftConfig.GiftTypeCode)
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
// 批量多目标仍然是每个目标各送一份同款礼物;每份房间贡献先按房间区域比例折算,再由聚合回执累加。
heatValue, err := giftDiamondAmount(baseHeatValue, roomContributionRatio.PPM)
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
totalChargeAmount, err := checkedMul(chargeAmount, int64(len(command.Targets)))
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
// 批量命令的每个目标都有独立 wallet command_id幂等重试必须是“全部已存在”或“全部未存在”不接受半批历史。
existing := make(map[string]ledger.BatchGiftTargetReceipt, len(command.Targets))
for _, target := range command.Targets {
targetCommand := debitGiftCommandFromBatchTarget(command, target)
requestHash := debitRequestHash(targetCommand)
txRow, exists, err := r.lookupTransaction(ctx, tx, target.CommandID, requestHash, bizTypeGiftDebit)
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
if !exists {
continue
}
receipt, err := r.receiptForGiftTransaction(ctx, tx, txRow.TransactionID)
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
existing[target.CommandID] = ledger.BatchGiftTargetReceipt{
TargetUserID: target.TargetUserID,
CommandID: target.CommandID,
Receipt: receipt,
}
}
if len(existing) > 0 {
if len(existing) != len(command.Targets) {
// 同一个 room 命令不应该只存在部分 wallet 子交易;直接 fail-close等待人工核对账本。
return ledger.BatchGiftReceipt{}, xerr.New(xerr.IdempotencyConflict, "wallet batch command idempotency conflict")
}
targetReceipts := make([]ledger.BatchGiftTargetReceipt, 0, len(command.Targets))
for _, target := range command.Targets {
targetReceipts = append(targetReceipts, existing[target.CommandID])
}
aggregate, err := aggregateGiftReceipts(targetReceipts)
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
return ledger.BatchGiftReceipt{Aggregate: aggregate, Targets: targetReceipts}, nil
}
sender, err := r.lockAccount(ctx, tx, command.SenderUserID, price.ChargeAssetType, totalChargeAmount == 0, nowMs)
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
if sender.AvailableAmount < totalChargeAmount {
return ledger.BatchGiftReceipt{}, xerr.New(xerr.InsufficientBalance, "insufficient balance")
}
targetAccounts, err := r.lockGiftTargetAccounts(ctx, tx, command.Targets, nowMs)
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
anyHostTarget := false
for _, target := range command.Targets {
if target.TargetIsHost {
anyHostTarget = true
break
}
}
hostRatio := giftDiamondRatioSnapshot{Percent: "0.00", PPM: 0}
hostRatioRegionID := int64(0)
if anyHostTarget {
hostRatio, hostRatioRegionID, err = r.resolveGiftDiamondRatio(ctx, tx, command.AppCode, command.SenderRegionID, giftConfig.GiftTypeCode)
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
}
events := make([]walletOutboxEvent, 0, len(command.Targets)*4)
targetReceipts := make([]ledger.BatchGiftTargetReceipt, 0, len(command.Targets))
for _, target := range command.Targets {
hostPeriodDiamondAdded := int64(0)
hostPeriodCycleKey := ""
giftDiamondRatioPercent := "0.00"
giftDiamondRatioRegionID := int64(0)
if target.TargetIsHost {
// 每个目标只使用自己的 host 快照;不能把某个主播的 agency/region 套给批量里的其他用户。
hostPeriodDiamondAdded, err = giftDiamondAmount(chargeAmount, hostRatio.PPM)
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
giftDiamondRatioPercent = hostRatio.Percent
giftDiamondRatioRegionID = hostRatioRegionID
hostPeriodCycleKey = hostPeriodCycleKeyFromMS(nowMs)
}
transactionID := transactionID(command.AppCode, target.CommandID)
senderAfter := sender.AvailableAmount - chargeAmount
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: senderAfter,
BillingReceipt: billingReceiptID(command.AppCode, target.CommandID),
SenderUserID: command.SenderUserID,
SenderRegionID: command.SenderRegionID,
TargetUserID: target.TargetUserID,
TargetIsHost: target.TargetIsHost,
TargetHostRegionID: target.TargetHostRegionID,
TargetAgencyOwnerUserID: target.TargetAgencyOwnerUserID,
HostPeriodDiamondAdded: hostPeriodDiamondAdded,
GiftDiamondRatioPercent: giftDiamondRatioPercent,
GiftDiamondRatioRegionID: giftDiamondRatioRegionID,
HostPeriodCycleKey: hostPeriodCycleKey,
RoomID: command.RoomID,
RoomRegionID: command.RegionID,
RoomContributionRatioPercent: roomContributionRatio.Percent,
RoomContributionRatioRegionID: roomContributionRatioRegionID,
CoinPrice: price.CoinPrice,
GiftPointAmount: price.GiftPointAmount,
HeatUnitValue: price.HeatValue,
}
if err := r.insertTransaction(ctx, tx, transactionID, target.CommandID, bizTypeGiftDebit, debitRequestHash(debitGiftCommandFromBatchTarget(command, target)), fmt.Sprintf("%s:%s", command.GiftID, price.PriceVersion), metadata, nowMs); err != nil {
return ledger.BatchGiftReceipt{}, err
}
if err := r.applyAccountDelta(ctx, tx, sender, -chargeAmount, 0, nowMs); err != nil {
return ledger.BatchGiftReceipt{}, err
}
events = append(events, balanceChangedEvent(transactionID, target.CommandID, command.SenderUserID, price.ChargeAssetType, -chargeAmount, 0, senderAfter, sender.FrozenAmount, sender.Version+1, metadata, nowMs))
sender.AvailableAmount = senderAfter
sender.Version++
targetAccount := targetAccounts[target.TargetUserID]
if err := r.applyAccountDelta(ctx, tx, targetAccount, giftPointAdded, 0, nowMs); err != nil {
return ledger.BatchGiftReceipt{}, err
}
targetAfter := targetAccount.AvailableAmount + giftPointAdded
events = append(events, balanceChangedEvent(transactionID, target.CommandID, target.TargetUserID, ledger.AssetGiftPoint, giftPointAdded, 0, targetAfter, targetAccount.FrozenAmount, targetAccount.Version+1, metadata, nowMs))
targetAccount.AvailableAmount = targetAfter
targetAccount.Version++
targetAccounts[target.TargetUserID] = targetAccount
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: target.TargetUserID,
RoomID: command.RoomID,
CreatedAtMS: nowMs,
}); err != nil {
return ledger.BatchGiftReceipt{}, err
}
if err := r.insertEntry(ctx, tx, walletEntry{
TransactionID: transactionID,
UserID: target.TargetUserID,
AssetType: ledger.AssetGiftPoint,
AvailableDelta: giftPointAdded,
FrozenDelta: 0,
AvailableAfter: targetAfter,
FrozenAfter: targetAccount.FrozenAmount,
CounterpartyUserID: command.SenderUserID,
RoomID: command.RoomID,
CreatedAtMS: nowMs,
}); err != nil {
return ledger.BatchGiftReceipt{}, err
}
if hostPeriodDiamondAdded > 0 {
hostPeriodDiamondAfter, hostPeriodDiamondVersion, err := r.creditHostPeriodDiamonds(ctx, tx, transactionID, target.CommandID, metadata, nowMs)
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
metadata.HostPeriodDiamondAfter = hostPeriodDiamondAfter
metadata.HostPeriodDiamondVersion = hostPeriodDiamondVersion
}
events = append(events, giftDebitedEvent(transactionID, target.CommandID, command.SenderUserID, price.ChargeAssetType, -chargeAmount, 0, metadata, nowMs))
if hostPeriodDiamondAdded > 0 {
events = append(events, hostPeriodDiamondCreditedEvent(transactionID, target.CommandID, metadata, nowMs))
}
targetReceipts = append(targetReceipts, ledger.BatchGiftTargetReceipt{
TargetUserID: target.TargetUserID,
CommandID: target.CommandID,
Receipt: receiptFromGiftMetadata(transactionID, metadata),
})
}
if err := r.insertWalletOutbox(ctx, tx, events); err != nil {
return ledger.BatchGiftReceipt{}, err
}
if err := tx.Commit(); err != nil {
return ledger.BatchGiftReceipt{}, err
}
aggregate, err := aggregateGiftReceipts(targetReceipts)
if err != nil {
return ledger.BatchGiftReceipt{}, err
}
return ledger.BatchGiftReceipt{Aggregate: aggregate, Targets: targetReceipts}, 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 := chargeAmount, int64(0)
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
}
// CreditLuckyGiftReward 在同一事务里完成幸运礼物中奖 COIN 入账、交易分录和钱包 outbox。
func (r *Repository) CreditLuckyGiftReward(ctx context.Context, command ledger.LuckyGiftRewardCommand) (ledger.LuckyGiftRewardReceipt, error) {
if r == nil || r.db == nil {
return ledger.LuckyGiftRewardReceipt{}, 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.LuckyGiftRewardReceipt{}, err
}
defer func() { _ = tx.Rollback() }()
requestHash := luckyGiftRewardRequestHash(command)
if txRow, exists, err := r.lookupTransaction(ctx, tx, command.CommandID, requestHash, bizTypeLuckyGiftReward); err != nil || exists {
if err != nil || !exists {
return ledger.LuckyGiftRewardReceipt{}, err
}
return r.receiptForLuckyGiftRewardTransaction(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.LuckyGiftRewardReceipt{}, err
}
transactionID := transactionID(command.AppCode, command.CommandID)
balanceAfter := account.AvailableAmount + command.Amount
metadata := luckyGiftRewardMetadata{
AppCode: command.AppCode,
TargetUserID: command.TargetUserID,
AssetType: ledger.AssetCoin,
Amount: command.Amount,
DrawID: command.DrawID,
RoomID: command.RoomID,
VisibleRegionID: command.VisibleRegionID,
GiftID: command.GiftID,
PoolID: command.PoolID,
Reason: command.Reason,
BalanceAfter: balanceAfter,
GrantedAtMS: nowMs,
}
if err := r.insertTransaction(ctx, tx, transactionID, command.CommandID, bizTypeLuckyGiftReward, requestHash, command.DrawID, metadata, nowMs); err != nil {
return ledger.LuckyGiftRewardReceipt{}, err
}
if err := r.applyAccountDelta(ctx, tx, account, command.Amount, 0, nowMs); err != nil {
return ledger.LuckyGiftRewardReceipt{}, 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,
RoomID: command.RoomID,
CreatedAtMS: nowMs,
}); err != nil {
return ledger.LuckyGiftRewardReceipt{}, 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),
luckyGiftRewardCreditedEvent(transactionID, command.CommandID, metadata, nowMs),
}); err != nil {
return ledger.LuckyGiftRewardReceipt{}, err
}
if err := tx.Commit(); err != nil {
return ledger.LuckyGiftRewardReceipt{}, err
}
return receiptFromLuckyGiftRewardMetadata(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)
// 币商转账的事实口径只记录金币数量USD 价格策略属于外部充值定价,不参与币商库存转普通金币。
rechargeUSDMinor := int64(0)
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: "",
RechargePolicyID: 0,
RechargePolicyVersion: "",
RechargePolicyCoinAmount: 0,
RechargePolicyUSDMinorAmount: 0,
}
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
}
// ListCoinSellerSalaryExchangeRateTiers 返回指定区域工资转币商的兑换区间;默认只返回 active 区间。
func (r *Repository) ListCoinSellerSalaryExchangeRateTiers(ctx context.Context, appCode string, regionID int64, includeDisabled bool) ([]ledger.CoinSellerSalaryExchangeRateTier, error) {
if r == nil || r.db == nil {
return nil, xerr.New(xerr.Unavailable, "mysql repository is not configured")
}
ctx = contextWithCommandApp(ctx, appCode)
statusClause := "AND status = 'active'"
if includeDisabled {
statusClause = ""
}
query := fmt.Sprintf(`
SELECT region_id, min_usd_minor, max_usd_minor, coin_per_usd, status, sort_order, updated_at_ms
FROM coin_seller_salary_exchange_rate_tiers
WHERE app_code = ? AND region_id = ? %s
ORDER BY sort_order ASC, min_usd_minor ASC, tier_id ASC`, statusClause)
rows, err := r.db.QueryContext(ctx, query, appcode.FromContext(ctx), regionID)
if err != nil {
return nil, err
}
defer rows.Close()
tiers := make([]ledger.CoinSellerSalaryExchangeRateTier, 0)
for rows.Next() {
var tier ledger.CoinSellerSalaryExchangeRateTier
if err := rows.Scan(&tier.RegionID, &tier.MinUSDMinor, &tier.MaxUSDMinor, &tier.CoinPerUSD, &tier.Status, &tier.SortOrder, &tier.UpdatedAtMS); err != nil {
return nil, err
}
tiers = append(tiers, tier)
}
if err := rows.Err(); err != nil {
return nil, err
}
return tiers, nil
}
// ExchangeSalaryToCoin 在同一事务里扣工资美元钱包、给当前用户普通 COIN 入账,保证两边余额和幂等一起落库。
func (r *Repository) ExchangeSalaryToCoin(ctx context.Context, command ledger.SalaryExchangeCommand) (ledger.SalaryExchangeReceipt, error) {
if r == nil || r.db == nil {
return ledger.SalaryExchangeReceipt{}, 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.SalaryExchangeReceipt{}, err
}
defer func() { _ = tx.Rollback() }()
requestHash := salaryExchangeRequestHash(command)
if txRow, exists, err := r.lookupTransactionWithConflictCode(ctx, tx, command.CommandID, requestHash, bizTypeSalaryExchangeToCoin, xerr.IdempotencyConflict); err != nil || exists {
if err != nil || !exists {
return ledger.SalaryExchangeReceipt{}, err
}
return r.receiptForSalaryExchangeTransaction(ctx, tx, txRow.TransactionID)
}
nowMs := time.Now().UnixMilli()
coinAmount, err := calculateSalaryCoinAmount(command.SalaryUSDMinor, ledger.SalaryExchangeCoinPerUSD)
if err != nil {
return ledger.SalaryExchangeReceipt{}, err
}
salaryAccount, err := r.lockAccount(ctx, tx, command.UserID, command.SalaryAssetType, false, nowMs)
if err != nil {
return ledger.SalaryExchangeReceipt{}, err
}
if salaryAccount.AvailableAmount < command.SalaryUSDMinor {
return ledger.SalaryExchangeReceipt{}, xerr.New(xerr.InsufficientBalance, "insufficient balance")
}
coinAccount, err := r.lockAccount(ctx, tx, command.UserID, ledger.AssetCoin, true, nowMs)
if err != nil {
return ledger.SalaryExchangeReceipt{}, err
}
transactionID := transactionID(command.AppCode, command.CommandID)
salaryAfter := salaryAccount.AvailableAmount - command.SalaryUSDMinor
coinAfter, err := checkedAdd(coinAccount.AvailableAmount, coinAmount)
if err != nil {
return ledger.SalaryExchangeReceipt{}, err
}
metadata := salaryExchangeMetadata{
AppCode: command.AppCode,
UserID: command.UserID,
SalaryAssetType: command.SalaryAssetType,
CoinAssetType: ledger.AssetCoin,
SalaryUSDMinor: command.SalaryUSDMinor,
CoinPerUSD: ledger.SalaryExchangeCoinPerUSD,
CoinAmount: coinAmount,
SalaryBalanceAfter: salaryAfter,
CoinBalanceAfter: coinAfter,
Reason: command.Reason,
CreatedAtMS: nowMs,
}
if err := r.insertTransaction(ctx, tx, transactionID, command.CommandID, bizTypeSalaryExchangeToCoin, requestHash, fmt.Sprintf("salary_exchange:%d", command.UserID), metadata, nowMs); err != nil {
return ledger.SalaryExchangeReceipt{}, err
}
if err := r.applyAccountDelta(ctx, tx, salaryAccount, -command.SalaryUSDMinor, 0, nowMs); err != nil {
return ledger.SalaryExchangeReceipt{}, err
}
if err := r.insertEntry(ctx, tx, walletEntry{
TransactionID: transactionID,
UserID: command.UserID,
AssetType: command.SalaryAssetType,
AvailableDelta: -command.SalaryUSDMinor,
FrozenDelta: 0,
AvailableAfter: salaryAfter,
FrozenAfter: salaryAccount.FrozenAmount,
CreatedAtMS: nowMs,
}); err != nil {
return ledger.SalaryExchangeReceipt{}, err
}
if err := r.applyAccountDelta(ctx, tx, coinAccount, coinAmount, 0, nowMs); err != nil {
return ledger.SalaryExchangeReceipt{}, err
}
if err := r.insertEntry(ctx, tx, walletEntry{
TransactionID: transactionID,
UserID: command.UserID,
AssetType: ledger.AssetCoin,
AvailableDelta: coinAmount,
FrozenDelta: 0,
AvailableAfter: coinAfter,
FrozenAfter: coinAccount.FrozenAmount,
CreatedAtMS: nowMs,
}); err != nil {
return ledger.SalaryExchangeReceipt{}, err
}
if err := r.insertWalletOutbox(ctx, tx, []walletOutboxEvent{
balanceChangedEvent(transactionID, command.CommandID, command.UserID, command.SalaryAssetType, -command.SalaryUSDMinor, 0, salaryAfter, salaryAccount.FrozenAmount, salaryAccount.Version+1, metadata, nowMs),
balanceChangedEvent(transactionID, command.CommandID, command.UserID, ledger.AssetCoin, coinAmount, 0, coinAfter, coinAccount.FrozenAmount, coinAccount.Version+1, metadata, nowMs),
}); err != nil {
return ledger.SalaryExchangeReceipt{}, err
}
if err := tx.Commit(); err != nil {
return ledger.SalaryExchangeReceipt{}, err
}
return receiptFromSalaryExchangeMetadata(transactionID, metadata), nil
}
// TransferSalaryToCoinSeller 在同一事务里扣来源工资钱包、按当前区域配置给币商专用金币库存入账。
func (r *Repository) TransferSalaryToCoinSeller(ctx context.Context, command ledger.SalaryTransferToCoinSellerCommand) (ledger.SalaryTransferToCoinSellerReceipt, error) {
if r == nil || r.db == nil {
return ledger.SalaryTransferToCoinSellerReceipt{}, 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.SalaryTransferToCoinSellerReceipt{}, err
}
defer func() { _ = tx.Rollback() }()
requestHash := salaryTransferToCoinSellerRequestHash(command)
if txRow, exists, err := r.lookupTransactionWithConflictCode(ctx, tx, command.CommandID, requestHash, bizTypeSalaryTransferToCoinSeller, xerr.IdempotencyConflict); err != nil || exists {
if err != nil || !exists {
return ledger.SalaryTransferToCoinSellerReceipt{}, err
}
return r.receiptForSalaryTransferToCoinSellerTransaction(ctx, tx, txRow.TransactionID)
}
nowMs := time.Now().UnixMilli()
rate, err := r.resolveCoinSellerSalaryExchangeRateTier(ctx, tx, command.RegionID, command.SalaryUSDMinor)
if err != nil {
return ledger.SalaryTransferToCoinSellerReceipt{}, err
}
coinAmount, err := calculateSalaryCoinAmount(command.SalaryUSDMinor, rate.CoinPerUSD)
if err != nil {
return ledger.SalaryTransferToCoinSellerReceipt{}, err
}
source, err := r.lockAccount(ctx, tx, command.SourceUserID, command.SalaryAssetType, false, nowMs)
if err != nil {
return ledger.SalaryTransferToCoinSellerReceipt{}, err
}
if source.AvailableAmount < command.SalaryUSDMinor {
return ledger.SalaryTransferToCoinSellerReceipt{}, xerr.New(xerr.InsufficientBalance, "insufficient balance")
}
seller, err := r.lockAccount(ctx, tx, command.SellerUserID, ledger.AssetCoinSellerCoin, true, nowMs)
if err != nil {
return ledger.SalaryTransferToCoinSellerReceipt{}, err
}
transactionID := transactionID(command.AppCode, command.CommandID)
sourceAfter := source.AvailableAmount - command.SalaryUSDMinor
sellerAfter, err := checkedAdd(seller.AvailableAmount, coinAmount)
if err != nil {
return ledger.SalaryTransferToCoinSellerReceipt{}, err
}
metadata := salaryTransferToCoinSellerMetadata{
AppCode: command.AppCode,
SourceUserID: command.SourceUserID,
SellerUserID: command.SellerUserID,
RegionID: command.RegionID,
SalaryAssetType: command.SalaryAssetType,
SellerAssetType: ledger.AssetCoinSellerCoin,
SalaryUSDMinor: command.SalaryUSDMinor,
CoinPerUSD: rate.CoinPerUSD,
CoinAmount: coinAmount,
RateMinUSDMinor: rate.MinUSDMinor,
RateMaxUSDMinor: rate.MaxUSDMinor,
SourceSalaryBalanceAfter: sourceAfter,
SellerBalanceAfter: sellerAfter,
Reason: command.Reason,
CreatedAtMS: nowMs,
}
if err := r.insertTransaction(ctx, tx, transactionID, command.CommandID, bizTypeSalaryTransferToCoinSeller, requestHash, fmt.Sprintf("salary_coin_seller:%d:%d", command.SourceUserID, command.SellerUserID), metadata, nowMs); err != nil {
return ledger.SalaryTransferToCoinSellerReceipt{}, err
}
if err := r.applyAccountDelta(ctx, tx, source, -command.SalaryUSDMinor, 0, nowMs); err != nil {
return ledger.SalaryTransferToCoinSellerReceipt{}, err
}
if err := r.insertEntry(ctx, tx, walletEntry{
TransactionID: transactionID,
UserID: command.SourceUserID,
AssetType: command.SalaryAssetType,
AvailableDelta: -command.SalaryUSDMinor,
FrozenDelta: 0,
AvailableAfter: sourceAfter,
FrozenAfter: source.FrozenAmount,
CounterpartyUserID: command.SellerUserID,
CreatedAtMS: nowMs,
}); err != nil {
return ledger.SalaryTransferToCoinSellerReceipt{}, err
}
if err := r.applyAccountDelta(ctx, tx, seller, coinAmount, 0, nowMs); err != nil {
return ledger.SalaryTransferToCoinSellerReceipt{}, err
}
if err := r.insertEntry(ctx, tx, walletEntry{
TransactionID: transactionID,
UserID: command.SellerUserID,
AssetType: ledger.AssetCoinSellerCoin,
AvailableDelta: coinAmount,
FrozenDelta: 0,
AvailableAfter: sellerAfter,
FrozenAfter: seller.FrozenAmount,
CounterpartyUserID: command.SourceUserID,
CreatedAtMS: nowMs,
}); err != nil {
return ledger.SalaryTransferToCoinSellerReceipt{}, err
}
if err := r.insertWalletOutbox(ctx, tx, []walletOutboxEvent{
balanceChangedEvent(transactionID, command.CommandID, command.SourceUserID, command.SalaryAssetType, -command.SalaryUSDMinor, 0, sourceAfter, source.FrozenAmount, source.Version+1, metadata, nowMs),
balanceChangedEvent(transactionID, command.CommandID, command.SellerUserID, ledger.AssetCoinSellerCoin, coinAmount, 0, sellerAfter, seller.FrozenAmount, seller.Version+1, metadata, nowMs),
}); err != nil {
return ledger.SalaryTransferToCoinSellerReceipt{}, err
}
if err := tx.Commit(); err != nil {
return ledger.SalaryTransferToCoinSellerReceipt{}, err
}
return receiptFromSalaryTransferToCoinSellerMetadata(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
}
type giftDiamondRatioSnapshot struct {
Percent string
PPM int64
}
func (r *Repository) resolveGiftDiamondRatio(ctx context.Context, tx *sql.Tx, appCode string, regionID int64, giftTypeCode string) (giftDiamondRatioSnapshot, int64, error) {
giftTypeCode = strings.TrimSpace(giftTypeCode)
if giftTypeCode == "" {
giftTypeCode = "normal"
}
regions := []int64{regionID}
if regionID != 0 {
regions = append(regions, 0)
}
for _, regionID := range regions {
ratio, exists, err := r.getGiftDiamondRatio(ctx, tx, appCode, regionID, giftTypeCode)
if err != nil {
return giftDiamondRatioSnapshot{}, 0, err
}
if exists {
return ratio, regionID, nil
}
}
return giftDiamondRatioSnapshot{Percent: "100.00", PPM: 1_000_000}, 0, nil
}
func (r *Repository) getGiftDiamondRatio(ctx context.Context, tx *sql.Tx, appCode string, regionID int64, giftTypeCode string) (giftDiamondRatioSnapshot, bool, error) {
var percent string
var ppm int64
err := tx.QueryRowContext(ctx, `
SELECT CAST(ratio_percent AS CHAR), CAST(ROUND(ratio_percent * 10000) AS SIGNED)
FROM gift_diamond_ratio_configs
WHERE app_code = ? AND region_id = ? AND gift_type_code = ? AND status = 'active'
LIMIT 1`,
appcode.Normalize(appCode), regionID, giftTypeCode,
).Scan(&percent, &ppm)
if errors.Is(err, sql.ErrNoRows) {
return giftDiamondRatioSnapshot{}, false, nil
}
if err != nil {
return giftDiamondRatioSnapshot{}, false, err
}
if ppm < 0 || ppm > 1_000_000 {
return giftDiamondRatioSnapshot{}, false, xerr.New(xerr.InvalidArgument, "gift diamond ratio is invalid")
}
percent = strings.TrimSpace(percent)
if percent == "" {
percent = "100.00"
}
return giftDiamondRatioSnapshot{Percent: percent, PPM: ppm}, true, nil
}
func giftDiamondAmount(chargeAmount int64, ratioPPM int64) (int64, error) {
if chargeAmount < 0 || ratioPPM < 0 {
return 0, xerr.New(xerr.InvalidArgument, "gift diamond amount is invalid")
}
product, err := checkedMul(chargeAmount, ratioPPM)
if err != nil {
return 0, err
}
return product / 1_000_000, 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
}
// 金币必须完整到账;充值 USD 统计只能落到最小货币单位,不能表示的尾差向下取整。
return numerator / policy.CoinAmount, nil
}
func (r *Repository) resolveCoinSellerSalaryExchangeRateTier(ctx context.Context, tx *sql.Tx, regionID int64, salaryUSDMinor int64) (ledger.CoinSellerSalaryExchangeRateTier, error) {
row := tx.QueryRowContext(ctx,
`SELECT region_id, min_usd_minor, max_usd_minor, coin_per_usd, status, sort_order, updated_at_ms
FROM coin_seller_salary_exchange_rate_tiers
WHERE app_code = ? AND region_id = ? AND status = 'active'
AND min_usd_minor <= ? AND max_usd_minor >= ?
ORDER BY sort_order ASC, min_usd_minor ASC, tier_id ASC
LIMIT 1
FOR UPDATE`,
appcode.FromContext(ctx),
regionID,
salaryUSDMinor,
salaryUSDMinor,
)
var tier ledger.CoinSellerSalaryExchangeRateTier
if err := row.Scan(&tier.RegionID, &tier.MinUSDMinor, &tier.MaxUSDMinor, &tier.CoinPerUSD, &tier.Status, &tier.SortOrder, &tier.UpdatedAtMS); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return ledger.CoinSellerSalaryExchangeRateTier{}, xerr.New(xerr.NotFound, "exchange rate not configured")
}
return ledger.CoinSellerSalaryExchangeRateTier{}, err
}
if tier.MinUSDMinor <= 0 || tier.MaxUSDMinor < tier.MinUSDMinor || tier.CoinPerUSD <= 0 || tier.CoinPerUSD%100 != 0 {
// 工资以美分入账coin_per_usd 必须能被 100 整除,避免转账金币出现隐式四舍五入。
return ledger.CoinSellerSalaryExchangeRateTier{}, xerr.New(xerr.Internal, "coin seller salary exchange rate is invalid")
}
return tier, nil
}
func calculateSalaryCoinAmount(salaryUSDMinor int64, coinPerUSD int64) (int64, error) {
numerator, err := checkedMul(salaryUSDMinor, coinPerUSD)
if err != nil {
return 0, err
}
if numerator%100 != 0 {
return 0, xerr.New(xerr.InvalidArgument, "salary amount does not match exchange rate")
}
return numerator / 100, 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.Amount,
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
}
func (r *Repository) creditHostPeriodDiamonds(ctx context.Context, tx *sql.Tx, transactionID string, commandID string, metadata giftMetadata, nowMs int64) (int64, int64, error) {
// host_period_diamond_accounts 是工资引擎读取的投影表,不是可提现钱包资产;只有主播身份收礼会写入。
if metadata.HostPeriodDiamondAdded <= 0 {
return 0, 0, nil
}
after, version, exists, err := r.lockHostPeriodDiamondAccount(ctx, tx, metadata.TargetUserID, metadata.HostPeriodCycleKey)
if err != nil {
return 0, 0, err
}
if !exists {
// 周期账户只在主播首次收礼时创建;工资结算任务按 cycle_key 扫描,不依赖通用 wallet_accounts。
// 首笔送礼写入当时区域和代理;同周期后续收礼会刷新快照,结算以周期账户最终快照为准。
after = metadata.HostPeriodDiamondAdded
version = 1
_, err = tx.ExecContext(ctx,
`INSERT INTO host_period_diamond_accounts (
app_code, user_id, cycle_key, region_id, agency_owner_user_id, total_diamonds, gift_diamond_total,
version, first_gift_at_ms, last_gift_at_ms, created_at_ms, updated_at_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
appcode.FromContext(ctx),
metadata.TargetUserID,
metadata.HostPeriodCycleKey,
metadata.TargetHostRegionID,
metadata.TargetAgencyOwnerUserID,
after,
metadata.HostPeriodDiamondAdded,
version,
nowMs,
nowMs,
nowMs,
nowMs,
)
if err != nil && isMySQLDuplicateError(err) {
// 并发首笔送礼可能同时发现账户不存在;退回锁读更新路径,保证不丢钻石。
after, version, exists, err = r.lockHostPeriodDiamondAccount(ctx, tx, metadata.TargetUserID, metadata.HostPeriodCycleKey)
}
if err != nil {
return 0, 0, err
}
if !exists {
if err := r.insertHostPeriodDiamondEntry(ctx, tx, transactionID, commandID, metadata, after, nowMs); err != nil {
return 0, 0, err
}
return after, version, nil
}
}
after, err = checkedAdd(after, metadata.HostPeriodDiamondAdded)
if err != nil {
return 0, 0, err
}
version++
// 已存在账户时更新累计钻石,同时刷新区域/代理快照为最近一次收礼状态;这符合“当月当前归属”口径。
result, err := tx.ExecContext(ctx,
`UPDATE host_period_diamond_accounts
SET region_id = ?, agency_owner_user_id = ?, total_diamonds = ?, gift_diamond_total = gift_diamond_total + ?,
version = ?, last_gift_at_ms = ?, updated_at_ms = ?
WHERE app_code = ? AND user_id = ? AND cycle_key = ?`,
metadata.TargetHostRegionID,
metadata.TargetAgencyOwnerUserID,
after,
metadata.HostPeriodDiamondAdded,
version,
nowMs,
nowMs,
appcode.FromContext(ctx),
metadata.TargetUserID,
metadata.HostPeriodCycleKey,
)
if err != nil {
return 0, 0, err
}
rows, err := result.RowsAffected()
if err != nil {
return 0, 0, err
}
if rows != 1 {
return 0, 0, xerr.New(xerr.LedgerConflict, "host period diamond account version conflict")
}
if err := r.insertHostPeriodDiamondEntry(ctx, tx, transactionID, commandID, metadata, after, nowMs); err != nil {
return 0, 0, err
}
return after, version, nil
}
func (r *Repository) lockHostPeriodDiamondAccount(ctx context.Context, tx *sql.Tx, userID int64, cycleKey string) (int64, int64, bool, error) {
// FOR UPDATE 串行化同主播同周期的并发送礼,保证 total_diamonds 不会被覆盖丢失。
row := tx.QueryRowContext(ctx,
`SELECT total_diamonds, version
FROM host_period_diamond_accounts
WHERE app_code = ? AND user_id = ? AND cycle_key = ?
FOR UPDATE`,
appcode.FromContext(ctx),
userID,
cycleKey,
)
var totalDiamonds int64
var version int64
if err := row.Scan(&totalDiamonds, &version); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return 0, 0, false, nil
}
return 0, 0, false, err
}
return totalDiamonds, version, true, nil
}
func (r *Repository) insertHostPeriodDiamondEntry(ctx context.Context, tx *sql.Tx, transactionID string, commandID string, metadata giftMetadata, diamondAfter int64, nowMs int64) error {
// 明细表记录每一笔送礼带来的工资钻石,后续核对周期账户累计值时不需要解析钱包 metadata。
_, err := tx.ExecContext(ctx,
`INSERT INTO host_period_diamond_entries (
app_code, transaction_id, command_id, user_id, cycle_key, region_id, agency_owner_user_id,
diamond_delta, diamond_after, gift_charge_asset_type, gift_charge_amount,
gift_id, gift_count, sender_user_id, room_id, created_at_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
appcode.FromContext(ctx),
transactionID,
commandID,
metadata.TargetUserID,
metadata.HostPeriodCycleKey,
metadata.TargetHostRegionID,
metadata.TargetAgencyOwnerUserID,
metadata.HostPeriodDiamondAdded,
diamondAfter,
metadata.ChargeAssetType,
metadata.ChargeAmount,
metadata.GiftID,
metadata.GiftCount,
metadata.SenderUserID,
metadata.RoomID,
nowMs,
)
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) {
return r.ClaimPendingWalletOutboxFiltered(ctx, workerID, limit, lockUntilMS, WalletOutboxClaimFilter{})
}
// ClaimPendingWalletOutboxFiltered atomically claims committed wallet facts for one MQ fanout lane.
func (r *Repository) ClaimPendingWalletOutboxFiltered(ctx context.Context, workerID string, limit int, lockUntilMS int64, filter WalletOutboxClaimFilter) ([]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()
}
appCode := appcode.FromContext(ctx)
eventFilterSQL, eventFilterArgs := walletOutboxEventFilterSQL(filter)
pendingIndex, claimIndex := walletOutboxClaimIndexes(filter)
tx, err := r.db.BeginTx(ctx, nil)
if err != nil {
return nil, err
}
defer func() { _ = tx.Rollback() }()
records := make([]WalletOutboxRecord, 0, limit)
claimBranches := []struct {
query string
args []any
}{
{
query: walletOutboxClaimQuery(pendingIndex, `
AND next_retry_at_ms IS NULL
AND lock_until_ms IS NULL`, eventFilterSQL),
args: append([]any{appCode, outboxStatusPending}, eventFilterArgs...),
},
{
query: walletOutboxClaimQuery(pendingIndex, `
AND next_retry_at_ms <= ?
AND lock_until_ms IS NULL`, eventFilterSQL),
args: append([]any{appCode, outboxStatusRetryable, nowMS}, eventFilterArgs...),
},
{
query: walletOutboxClaimQuery(claimIndex, `
AND lock_until_ms <= ?`, eventFilterSQL),
args: append([]any{appCode, outboxStatusDelivering, nowMS}, eventFilterArgs...),
},
}
for _, branch := range claimBranches {
remaining := limit - len(records)
if remaining <= 0 {
break
}
args := append(append([]any{}, branch.args...), remaining)
branchRecords, err := queryWalletOutboxRecords(ctx, tx, branch.query, args...)
if err != nil {
return nil, err
}
records = append(records, branchRecords...)
}
for _, record := range records {
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,
record.AppCode,
record.EventID,
); err != nil {
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
}
func walletOutboxClaimQuery(indexName string, statusCondition string, eventFilterSQL string) string {
return `
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 FORCE INDEX (` + indexName + `)
WHERE app_code = ? AND status = ?` + statusCondition + eventFilterSQL + `
ORDER BY created_at_ms ASC, event_id ASC
LIMIT ?
FOR UPDATE SKIP LOCKED`
}
func walletOutboxEventFilterSQL(filter WalletOutboxClaimFilter) (string, []any) {
include := normalizeWalletOutboxEventTypes(filter.IncludeEventTypes)
if len(include) > 0 {
return " AND event_type IN (" + walletOutboxPlaceholders(len(include)) + ")", walletOutboxEventArgs(include)
}
exclude := normalizeWalletOutboxEventTypes(filter.ExcludeEventTypes)
if len(exclude) > 0 {
return " AND event_type NOT IN (" + walletOutboxPlaceholders(len(exclude)) + ")", walletOutboxEventArgs(exclude)
}
return "", nil
}
func walletOutboxClaimIndexes(filter WalletOutboxClaimFilter) (string, string) {
if len(normalizeWalletOutboxEventTypes(filter.IncludeEventTypes)) > 0 {
return "idx_wallet_outbox_event_pending", "idx_wallet_outbox_event_claim"
}
return "idx_wallet_outbox_pending", "idx_wallet_outbox_claim"
}
func normalizeWalletOutboxEventTypes(values []string) []string {
normalized := make([]string, 0, len(values))
seen := make(map[string]struct{}, len(values))
for _, value := range values {
value = strings.TrimSpace(value)
if value == "" {
continue
}
if _, exists := seen[value]; exists {
continue
}
seen[value] = struct{}{}
normalized = append(normalized, value)
}
return normalized
}
func walletOutboxPlaceholders(count int) string {
if count <= 0 {
return ""
}
placeholders := make([]string, count)
for index := range placeholders {
placeholders[index] = "?"
}
return strings.Join(placeholders, ",")
}
func walletOutboxEventArgs(values []string) []any {
args := make([]any, 0, len(values))
for _, value := range values {
args = append(args, value)
}
return args
}
func queryWalletOutboxRecords(ctx context.Context, tx *sql.Tx, query string, args ...any) ([]WalletOutboxRecord, error) {
rows, err := tx.QueryContext(ctx, query, args...)
if err != nil {
return nil, err
}
defer rows.Close()
records := make([]WalletOutboxRecord, 0)
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 {
return nil, err
}
records = append(records, record)
}
if err := rows.Err(); err != nil {
return nil, err
}
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"`
SenderRegionID int64 `json:"sender_region_id"`
TargetUserID int64 `json:"target_user_id"`
TargetIsHost bool `json:"target_is_host"`
TargetHostRegionID int64 `json:"target_host_region_id"`
TargetAgencyOwnerUserID int64 `json:"target_agency_owner_user_id"`
HostPeriodDiamondAdded int64 `json:"host_period_diamond_added"`
GiftDiamondRatioPercent string `json:"gift_diamond_ratio_percent"`
GiftDiamondRatioRegionID int64 `json:"gift_diamond_ratio_region_id"`
HostPeriodDiamondAfter int64 `json:"host_period_diamond_after"`
HostPeriodDiamondVersion int64 `json:"host_period_diamond_version"`
HostPeriodCycleKey string `json:"host_period_cycle_key"`
RoomID string `json:"room_id"`
RoomRegionID int64 `json:"room_region_id"`
RoomContributionRatioPercent string `json:"room_contribution_ratio_percent"`
RoomContributionRatioRegionID int64 `json:"room_contribution_ratio_region_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 luckyGiftRewardMetadata struct {
AppCode string `json:"app_code"`
TargetUserID int64 `json:"target_user_id"`
AssetType string `json:"asset_type"`
Amount int64 `json:"amount"`
DrawID string `json:"draw_id"`
RoomID string `json:"room_id"`
VisibleRegionID int64 `json:"visible_region_id"`
GiftID string `json:"gift_id"`
PoolID string `json:"pool_id"`
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 salaryExchangeMetadata struct {
AppCode string `json:"app_code"`
UserID int64 `json:"user_id"`
SalaryAssetType string `json:"salary_asset_type"`
CoinAssetType string `json:"coin_asset_type"`
SalaryUSDMinor int64 `json:"salary_usd_minor"`
CoinPerUSD int64 `json:"coin_per_usd"`
CoinAmount int64 `json:"coin_amount"`
SalaryBalanceAfter int64 `json:"salary_balance_after"`
CoinBalanceAfter int64 `json:"coin_balance_after"`
Reason string `json:"reason"`
CreatedAtMS int64 `json:"created_at_ms"`
}
type salaryTransferToCoinSellerMetadata struct {
AppCode string `json:"app_code"`
SourceUserID int64 `json:"source_user_id"`
SellerUserID int64 `json:"seller_user_id"`
RegionID int64 `json:"region_id"`
SalaryAssetType string `json:"salary_asset_type"`
SellerAssetType string `json:"seller_asset_type"`
SalaryUSDMinor int64 `json:"salary_usd_minor"`
CoinPerUSD int64 `json:"coin_per_usd"`
CoinAmount int64 `json:"coin_amount"`
RateMinUSDMinor int64 `json:"rate_min_usd_minor"`
RateMaxUSDMinor int64 `json:"rate_max_usd_minor"`
SourceSalaryBalanceAfter int64 `json:"source_salary_balance_after"`
SellerBalanceAfter int64 `json:"seller_balance_after"`
Reason string `json:"reason"`
CreatedAtMS int64 `json:"created_at_ms"`
}
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,
HostPeriodDiamondAdded: metadata.HostPeriodDiamondAdded,
HostPeriodCycleKey: metadata.HostPeriodCycleKey,
}
}
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) receiptForLuckyGiftRewardTransaction(ctx context.Context, tx *sql.Tx, transactionID string) (ledger.LuckyGiftRewardReceipt, 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.LuckyGiftRewardReceipt{}, err
}
var metadata luckyGiftRewardMetadata
if err := json.Unmarshal([]byte(metadataJSON), &metadata); err != nil {
return ledger.LuckyGiftRewardReceipt{}, err
}
return receiptFromLuckyGiftRewardMetadata(transactionID, metadata), nil
}
func receiptFromLuckyGiftRewardMetadata(transactionID string, metadata luckyGiftRewardMetadata) ledger.LuckyGiftRewardReceipt {
return ledger.LuckyGiftRewardReceipt{
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 (r *Repository) receiptForSalaryExchangeTransaction(ctx context.Context, tx *sql.Tx, transactionID string) (ledger.SalaryExchangeReceipt, 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.SalaryExchangeReceipt{}, err
}
var metadata salaryExchangeMetadata
if err := json.Unmarshal([]byte(metadataJSON), &metadata); err != nil {
return ledger.SalaryExchangeReceipt{}, err
}
return receiptFromSalaryExchangeMetadata(transactionID, metadata), nil
}
func receiptFromSalaryExchangeMetadata(transactionID string, metadata salaryExchangeMetadata) ledger.SalaryExchangeReceipt {
return ledger.SalaryExchangeReceipt{
TransactionID: transactionID,
UserID: metadata.UserID,
SalaryAssetType: metadata.SalaryAssetType,
SalaryBalanceAfter: metadata.SalaryBalanceAfter,
CoinBalanceAfter: metadata.CoinBalanceAfter,
SalaryUSDMinor: metadata.SalaryUSDMinor,
CoinAmount: metadata.CoinAmount,
CoinPerUSD: metadata.CoinPerUSD,
CreatedAtMS: metadata.CreatedAtMS,
}
}
func (r *Repository) receiptForSalaryTransferToCoinSellerTransaction(ctx context.Context, tx *sql.Tx, transactionID string) (ledger.SalaryTransferToCoinSellerReceipt, 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.SalaryTransferToCoinSellerReceipt{}, err
}
var metadata salaryTransferToCoinSellerMetadata
if err := json.Unmarshal([]byte(metadataJSON), &metadata); err != nil {
return ledger.SalaryTransferToCoinSellerReceipt{}, err
}
return receiptFromSalaryTransferToCoinSellerMetadata(transactionID, metadata), nil
}
func receiptFromSalaryTransferToCoinSellerMetadata(transactionID string, metadata salaryTransferToCoinSellerMetadata) ledger.SalaryTransferToCoinSellerReceipt {
return ledger.SalaryTransferToCoinSellerReceipt{
TransactionID: transactionID,
SourceUserID: metadata.SourceUserID,
SellerUserID: metadata.SellerUserID,
SalaryAssetType: metadata.SalaryAssetType,
SourceSalaryBalanceAfter: metadata.SourceSalaryBalanceAfter,
SellerBalanceAfter: metadata.SellerBalanceAfter,
SalaryUSDMinor: metadata.SalaryUSDMinor,
CoinAmount: metadata.CoinAmount,
CoinPerUSD: metadata.CoinPerUSD,
RateMinUSDMinor: metadata.RateMinUSDMinor,
RateMaxUSDMinor: metadata.RateMaxUSDMinor,
CreatedAtMS: metadata.CreatedAtMS,
}
}
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 luckyGiftRewardCreditedEvent(transactionID string, commandID string, metadata luckyGiftRewardMetadata, nowMs int64) walletOutboxEvent {
return walletOutboxEvent{
EventID: eventID(transactionID, "WalletLuckyGiftRewardCredited", metadata.TargetUserID, ledger.AssetCoin),
EventType: "WalletLuckyGiftRewardCredited",
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,
"draw_id": metadata.DrawID,
"room_id": metadata.RoomID,
"visible_region_id": metadata.VisibleRegionID,
"region_id": metadata.VisibleRegionID,
"gift_id": metadata.GiftID,
"pool_id": metadata.PoolID,
"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 hostPeriodDiamondCreditedEvent(transactionID string, commandID string, metadata giftMetadata, nowMs int64) walletOutboxEvent {
return walletOutboxEvent{
EventID: eventID(transactionID, "HostPeriodDiamondCredited", metadata.TargetUserID, ledger.AssetHostPeriodDiamond),
EventType: "HostPeriodDiamondCredited",
TransactionID: transactionID,
CommandID: commandID,
UserID: metadata.TargetUserID,
AssetType: ledger.AssetHostPeriodDiamond,
AvailableDelta: metadata.HostPeriodDiamondAdded,
FrozenDelta: 0,
Payload: map[string]any{
"transaction_id": transactionID,
"command_id": commandID,
"host_user_id": metadata.TargetUserID,
"cycle_key": metadata.HostPeriodCycleKey,
"region_id": metadata.TargetHostRegionID,
"agency_owner_user_id": metadata.TargetAgencyOwnerUserID,
"diamond_delta": metadata.HostPeriodDiamondAdded,
"diamond_after": metadata.HostPeriodDiamondAfter,
"gift_charge_asset_type": metadata.ChargeAssetType,
"gift_charge_amount": metadata.ChargeAmount,
"gift_id": metadata.GiftID,
"gift_count": metadata.GiftCount,
"sender_user_id": metadata.SenderUserID,
"sender_region_id": metadata.SenderRegionID,
"room_id": metadata.RoomID,
"gift_diamond_ratio_percent": metadata.GiftDiamondRatioPercent,
"gift_diamond_ratio_region_id": metadata.GiftDiamondRatioRegionID,
"created_at_ms": nowMs,
},
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 debitGiftCommandFromBatchTarget(command ledger.BatchDebitGiftCommand, target ledger.DebitGiftTargetCommand) ledger.DebitGiftCommand {
// 批量目标最终仍落成独立 gift_debit 交易;复用单目标 hash 可以保证账务语义变化时照样触发幂等冲突。
return ledger.DebitGiftCommand{
AppCode: command.AppCode,
CommandID: target.CommandID,
RoomID: command.RoomID,
SenderUserID: command.SenderUserID,
TargetUserID: target.TargetUserID,
GiftID: command.GiftID,
GiftCount: command.GiftCount,
PriceVersion: command.PriceVersion,
RegionID: command.RegionID,
SenderRegionID: command.SenderRegionID,
TargetIsHost: target.TargetIsHost,
TargetHostRegionID: target.TargetHostRegionID,
TargetAgencyOwnerUserID: target.TargetAgencyOwnerUserID,
}
}
func (r *Repository) lockGiftTargetAccounts(ctx context.Context, tx *sql.Tx, targets []ledger.DebitGiftTargetCommand, nowMs int64) (map[int64]walletAccount, error) {
ids := make([]int64, 0, len(targets))
seen := make(map[int64]struct{}, len(targets))
for _, target := range targets {
if _, exists := seen[target.TargetUserID]; exists {
continue
}
seen[target.TargetUserID] = struct{}{}
ids = append(ids, target.TargetUserID)
}
sort.Slice(ids, func(left, right int) bool {
return ids[left] < ids[right]
})
accounts := make(map[int64]walletAccount, len(ids))
for _, targetUserID := range ids {
// 固定按 user_id 升序锁目标账户,降低并发批量送礼互相等待时形成死锁的概率。
account, err := r.lockAccount(ctx, tx, targetUserID, ledger.AssetGiftPoint, true, nowMs)
if err != nil {
return nil, err
}
accounts[targetUserID] = account
}
return accounts, nil
}
func aggregateGiftReceipts(targets []ledger.BatchGiftTargetReceipt) (ledger.Receipt, error) {
aggregate := ledger.Receipt{}
billingReceiptIDs := make([]string, 0, len(targets))
transactionIDs := make([]string, 0, len(targets))
for _, target := range targets {
receipt := target.Receipt
if receipt.BillingReceiptID != "" {
billingReceiptIDs = append(billingReceiptIDs, receipt.BillingReceiptID)
}
if receipt.TransactionID != "" {
transactionIDs = append(transactionIDs, receipt.TransactionID)
}
var err error
if aggregate.CoinSpent, err = checkedAdd(aggregate.CoinSpent, receipt.CoinSpent); err != nil {
return ledger.Receipt{}, err
}
if aggregate.ChargeAmount, err = checkedAdd(aggregate.ChargeAmount, receipt.ChargeAmount); err != nil {
return ledger.Receipt{}, err
}
if aggregate.GiftPointAdded, err = checkedAdd(aggregate.GiftPointAdded, receipt.GiftPointAdded); err != nil {
return ledger.Receipt{}, err
}
if aggregate.HeatValue, err = checkedAdd(aggregate.HeatValue, receipt.HeatValue); err != nil {
return ledger.Receipt{}, err
}
if aggregate.HostPeriodDiamondAdded, err = checkedAdd(aggregate.HostPeriodDiamondAdded, receipt.HostPeriodDiamondAdded); err != nil {
return ledger.Receipt{}, err
}
if aggregate.ChargeAssetType == "" {
aggregate.ChargeAssetType = receipt.ChargeAssetType
} else if receipt.ChargeAssetType != "" && aggregate.ChargeAssetType != receipt.ChargeAssetType {
aggregate.ChargeAssetType = giftWallChargeAssetMixed
}
if aggregate.GiftTypeCode == "" {
aggregate.GiftTypeCode = receipt.GiftTypeCode
}
if aggregate.PriceVersion == "" {
aggregate.PriceVersion = receipt.PriceVersion
}
if aggregate.HostPeriodCycleKey == "" {
aggregate.HostPeriodCycleKey = receipt.HostPeriodCycleKey
}
// sender 账后余额随目标顺序逐笔递减;聚合响应返回最后一笔后的余额,表示整批完成后的余额。
aggregate.BalanceAfter = receipt.BalanceAfter
}
aggregate.BillingReceiptID = strings.Join(billingReceiptIDs, ",")
aggregate.TransactionID = strings.Join(transactionIDs, ",")
return aggregate, nil
}
func hostPeriodCycleKeyFromMS(ms int64) string {
// 工资周期用 UTC 月份锚定,避免服务节点本地时区差异导致月底礼物落到不同周期。
return time.UnixMilli(ms).UTC().Format("2006-01")
}
func debitRequestHash(command ledger.DebitGiftCommand) string {
// 哈希覆盖所有账务语义字段;相同 command_id 只能重放完全相同的业务命令。
return stableHash(fmt.Sprintf("gift|%s|%s|%d|%d|%s|%d|%s|%d|%d|%t|%d|%d",
appcode.Normalize(command.AppCode),
command.RoomID,
command.SenderUserID,
command.TargetUserID,
command.GiftID,
command.GiftCount,
strings.TrimSpace(command.PriceVersion),
command.RegionID,
command.SenderRegionID,
command.TargetIsHost,
command.TargetHostRegionID,
command.TargetAgencyOwnerUserID,
))
}
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 luckyGiftRewardRequestHash(command ledger.LuckyGiftRewardCommand) string {
return stableHash(fmt.Sprintf("lucky_gift_reward|%s|%d|%d|%s|%s|%s|%s|%s",
appcode.Normalize(command.AppCode),
command.TargetUserID,
command.Amount,
strings.TrimSpace(command.DrawID),
strings.TrimSpace(command.RoomID),
strings.TrimSpace(command.GiftID),
strings.TrimSpace(command.PoolID),
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 salaryExchangeRequestHash(command ledger.SalaryExchangeCommand) string {
return stableHash(fmt.Sprintf("salary_exchange|%s|%d|%s|%d|%s",
appcode.Normalize(command.AppCode),
command.UserID,
strings.ToUpper(strings.TrimSpace(command.SalaryAssetType)),
command.SalaryUSDMinor,
strings.TrimSpace(command.Reason),
))
}
func salaryTransferToCoinSellerRequestHash(command ledger.SalaryTransferToCoinSellerCommand) string {
return stableHash(fmt.Sprintf("salary_transfer_coin_seller|%s|%d|%d|%s|%d|%d|%s",
appcode.Normalize(command.AppCode),
command.SourceUserID,
command.SellerUserID,
strings.ToUpper(strings.TrimSpace(command.SalaryAssetType)),
command.SalaryUSDMinor,
command.RegionID,
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
}