433 lines
21 KiB
Go
433 lines
21 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
"hyapp/services/wallet-service/internal/domain/ledger"
|
||
)
|
||
|
||
// FreezeSalaryWithdrawal 在同一事务里把工资 available 转入 frozen，并落交易、分录和余额 outbox。
|
||
func (r *Repository) FreezeSalaryWithdrawal(ctx context.Context, command ledger.SalaryWithdrawalCommand) (ledger.SalaryWithdrawalReceipt, error) {
|
||
return r.applySalaryWithdrawal(ctx, command, salaryWithdrawalMutation{
|
||
BizType: bizTypeSalaryWithdrawalFreeze,
|
||
AvailableDelta: -command.SalaryUSDMinor,
|
||
FrozenDelta: command.SalaryUSDMinor,
|
||
ExternalRef: command.WithdrawalRef,
|
||
})
|
||
}
|
||
|
||
// SettleSalaryWithdrawal 在审核通过时只扣 frozen；这保证提交申请时已经冻结的金额不会被重复扣 available。
|
||
func (r *Repository) SettleSalaryWithdrawal(ctx context.Context, command ledger.SalaryWithdrawalCommand) (ledger.SalaryWithdrawalReceipt, error) {
|
||
return r.applySalaryWithdrawal(ctx, command, salaryWithdrawalMutation{
|
||
BizType: bizTypeSalaryWithdrawalSettle,
|
||
AvailableDelta: 0,
|
||
FrozenDelta: -command.SalaryUSDMinor,
|
||
ExternalRef: command.WithdrawalApplicationID,
|
||
})
|
||
}
|
||
|
||
// ReleaseSalaryWithdrawal 在拒绝或创建申请失败回滚时把 frozen 返还 available。
|
||
func (r *Repository) ReleaseSalaryWithdrawal(ctx context.Context, command ledger.SalaryWithdrawalCommand) (ledger.SalaryWithdrawalReceipt, error) {
|
||
return r.applySalaryWithdrawal(ctx, command, salaryWithdrawalMutation{
|
||
BizType: bizTypeSalaryWithdrawalRelease,
|
||
AvailableDelta: command.SalaryUSDMinor,
|
||
FrozenDelta: -command.SalaryUSDMinor,
|
||
ExternalRef: command.WithdrawalApplicationID,
|
||
})
|
||
}
|
||
|
||
type salaryWithdrawalMutation struct {
|
||
BizType string
|
||
AvailableDelta int64
|
||
FrozenDelta int64
|
||
ExternalRef string
|
||
}
|
||
|
||
func (r *Repository) applySalaryWithdrawal(ctx context.Context, command ledger.SalaryWithdrawalCommand, mutation salaryWithdrawalMutation) (ledger.SalaryWithdrawalReceipt, error) {
|
||
if r == nil || r.db == nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, 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.SalaryWithdrawalReceipt{}, err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
|
||
// 财务旧版 command id 带 approved/rejected，新版按 finance stage 固定。在 command 幂等前先锁定提现申请的唯一终局，
|
||
// 才能让旧 command 成功但 admin 未终态的 partial-pending 重试返回原回执，且阻止另一个 command 执行相反决策。
|
||
if receipt, exists, err := r.lookupSalaryWithdrawalTerminal(ctx, tx, command, mutation.BizType); err != nil || exists {
|
||
if err != nil || !exists {
|
||
return ledger.SalaryWithdrawalReceipt{}, err
|
||
}
|
||
return receipt, nil
|
||
}
|
||
|
||
requestHash := salaryWithdrawalRequestHash(command, mutation.BizType)
|
||
if txRow, exists, err := r.lookupTransactionWithConflictCode(ctx, tx, command.CommandID, requestHash, mutation.BizType, xerr.IdempotencyConflict); err != nil || exists {
|
||
if err != nil || !exists {
|
||
return ledger.SalaryWithdrawalReceipt{}, err
|
||
}
|
||
if mutation.BizType == bizTypeSalaryWithdrawalFreeze && salaryAssetUsesHostPolicyWithdrawal(command.SalaryAssetType) {
|
||
released, reservationErr := r.hostPolicyWithdrawalReservationWasReleased(ctx, tx, command.SalaryAssetType, command.CommandID)
|
||
if reservationErr != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, reservationErr
|
||
}
|
||
if released {
|
||
return ledger.SalaryWithdrawalReceipt{}, xerr.New(xerr.IdempotencyConflict, "withdrawal freeze was rolled back; use a new command_id")
|
||
}
|
||
}
|
||
return r.receiptForSalaryWithdrawalTransaction(ctx, tx, txRow.TransactionID)
|
||
}
|
||
|
||
nowMs := time.Now().UTC().UnixMilli()
|
||
pointPolicyInstanceCode := ""
|
||
grossUSDMinor := int64(0)
|
||
feeUSDMinor := int64(0)
|
||
netUSDMinor := int64(0)
|
||
pointPolicyID := uint64(0)
|
||
pointPolicyVersion := uint64(0)
|
||
coinsPerUSD := int64(0)
|
||
if mutation.BizType == bizTypeSalaryWithdrawalFreeze && ledger.ValidPointWithdrawalAssetType(command.SalaryAssetType) {
|
||
// Gateway 的配置读取只是页面预览。资金冻结必须在本事务内重新解析 App + 区域 wallet 政策，
|
||
// 并由 wallet 覆盖比例/手续费快照，防止 Admin 改政策后的竞态或调用方篡改。
|
||
pointsPerUSD := int64(0)
|
||
feeBPS := int64(0)
|
||
if command.SalaryAssetType == ledger.AssetPointDiamond {
|
||
policy, resolveErr := r.resolvePointDiamondHostPolicy(ctx, tx, command.RegionID, nowMs)
|
||
if resolveErr != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, resolveErr
|
||
}
|
||
if command.GrossUSDMinor < policy.MinimumWithdrawUSDMinor {
|
||
return ledger.SalaryWithdrawalReceipt{}, xerr.New(xerr.InvalidArgument, "withdrawal amount is below configured minimum")
|
||
}
|
||
pointProduct, calculateErr := checkedMul(command.GrossUSDMinor, policy.PointDiamondsPerUSD)
|
||
if calculateErr != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, calculateErr
|
||
}
|
||
command.SalaryUSDMinor = pointProduct / 100
|
||
pointsPerUSD, feeBPS = policy.PointDiamondsPerUSD, int64(policy.WithdrawFeeBPS)
|
||
pointPolicyInstanceCode = fmt.Sprintf("host:%d:%d", policy.PolicyID, policy.PolicyVersion)
|
||
pointPolicyID, pointPolicyVersion, coinsPerUSD = policy.PolicyID, policy.PolicyVersion, policy.CoinsPerUSD
|
||
feeUSDProduct, calculateErr := checkedMul(command.GrossUSDMinor, feeBPS)
|
||
if calculateErr != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, calculateErr
|
||
}
|
||
grossUSDMinor = command.GrossUSDMinor
|
||
feeUSDMinor = feeUSDProduct / 10_000
|
||
netUSDMinor = grossUSDMinor - feeUSDMinor
|
||
} else {
|
||
policy, found, resolveErr := r.resolveActiveWalletPolicy(ctx, tx, command.AppCode, command.RegionID, nowMs)
|
||
if resolveErr != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, resolveErr
|
||
}
|
||
if !found {
|
||
return ledger.SalaryWithdrawalReceipt{}, xerr.New(xerr.PermissionDenied, "point wallet policy is not configured")
|
||
}
|
||
if command.SalaryUSDMinor < policy.MinimumPoints {
|
||
return ledger.SalaryWithdrawalReceipt{}, xerr.New(xerr.InvalidArgument, "point withdrawal amount is below configured minimum")
|
||
}
|
||
pointsPerUSD, feeBPS = policy.PointsPerUSD, policy.WithdrawFeeBPS
|
||
pointPolicyInstanceCode = policy.InstanceCode
|
||
}
|
||
if command.SalaryUSDMinor <= 0 {
|
||
return ledger.SalaryWithdrawalReceipt{}, xerr.New(xerr.InvalidArgument, "withdrawal amount is below exchange precision")
|
||
}
|
||
command.PointsPerUSD = pointsPerUSD
|
||
command.PointWithdrawFeeBPS = feeBPS
|
||
if command.SalaryAssetType == ledger.AssetPointDiamond {
|
||
// 手续费先按美元分 floor，再把净美元换成积分，保证财务实际支付严格等于政策 net 美元分。
|
||
netPointProduct, calculateErr := checkedMul(netUSDMinor, pointsPerUSD)
|
||
if calculateErr != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, calculateErr
|
||
}
|
||
command.PointNetAmount = netPointProduct / 100
|
||
command.PointFeeAmount = command.SalaryUSDMinor - command.PointNetAmount
|
||
} else {
|
||
feeProduct, calculateErr := checkedMul(command.SalaryUSDMinor, feeBPS)
|
||
if calculateErr != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, calculateErr
|
||
}
|
||
command.PointFeeAmount = feeProduct / 10_000
|
||
command.PointNetAmount = command.SalaryUSDMinor - command.PointFeeAmount
|
||
}
|
||
// POINT_DIAMOND 的积分冻结额直到事务内解析 Host 政策后才确定；FreezeSalaryWithdrawal
|
||
// 入参构造 mutation 时仍是 0，必须在锁账户前用最终积分重建双边 delta。
|
||
mutation.AvailableDelta = -command.SalaryUSDMinor
|
||
mutation.FrozenDelta = command.SalaryUSDMinor
|
||
} else if command.SalaryAssetType == ledger.AssetPointDiamond {
|
||
// 审核通过、驳回和申请创建失败回滚都复制首次冻结快照；不能用审核时当前政策重算，
|
||
// 也不能把冻结积分数误写成 gross_usd_minor。
|
||
freezeMetadata, loadErr := r.loadPointDiamondFreezeMetadata(ctx, tx, command)
|
||
if loadErr != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, loadErr
|
||
}
|
||
pointPolicyInstanceCode = freezeMetadata.PointPolicyInstanceCode
|
||
grossUSDMinor, feeUSDMinor, netUSDMinor = freezeMetadata.GrossUSDMinor, freezeMetadata.FeeUSDMinor, freezeMetadata.NetUSDMinor
|
||
pointPolicyID, pointPolicyVersion = freezeMetadata.PointPolicyID, freezeMetadata.PointPolicyVersion
|
||
coinsPerUSD = freezeMetadata.CoinsPerUSD
|
||
}
|
||
account, err := r.lockAccount(ctx, tx, command.UserID, command.SalaryAssetType, false, nowMs)
|
||
if err != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, err
|
||
}
|
||
if account.AvailableAmount+mutation.AvailableDelta < 0 || account.FrozenAmount+mutation.FrozenDelta < 0 {
|
||
return ledger.SalaryWithdrawalReceipt{}, xerr.New(xerr.InsufficientBalance, "insufficient balance")
|
||
}
|
||
// Host 政策工资和 POINT 都使用同一份周期绑定提现次数规则；审核扣款与驳回不能重复计数。
|
||
if mutation.BizType == bizTypeSalaryWithdrawalFreeze && salaryAssetUsesHostPolicyWithdrawal(command.SalaryAssetType) {
|
||
missingPolicyMode := allowLegacyWithoutHostPolicy
|
||
if ledger.ValidPointWithdrawalAssetType(command.SalaryAssetType) {
|
||
// POINT 保持 Lalu 原链路：没有当前区域 Host 政策时失败关闭，不能绕过提现次数和日期限制。
|
||
missingPolicyMode = requireHostPolicy
|
||
}
|
||
policyType := ""
|
||
if command.SalaryAssetType == ledger.AssetPointDiamond {
|
||
// 永久积分限制跟随最后一份 POINT_DIAMOND 快照，不能被随后发布的工资型政策覆盖。
|
||
policyType = ledger.HostPolicyTypePointDiamond
|
||
}
|
||
if err := r.consumeHostPolicyWithdrawalLimitForPolicyType(ctx, tx, command.UserID, command.RegionID, pointWithdrawalChannelPlatform, command.CommandID, command.WithdrawalRef, nowMs, command.SalaryAssetType, policyType, missingPolicyMode); err != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, err
|
||
}
|
||
}
|
||
if mutation.BizType == bizTypeSalaryWithdrawalRelease && salaryAssetUsesHostPolicyWithdrawal(command.SalaryAssetType) && strings.TrimSpace(command.WithdrawalApplicationID) == "" {
|
||
// 只有 gateway 建单失败补偿会携带 withdrawal_ref 且没有 application_id；人工驳回必须继续占用发起次数。
|
||
if err := r.releaseHostPolicyWithdrawalLimitReservation(ctx, tx, command.UserID, command.SalaryAssetType, command.WithdrawalRef, command.CommandID, nowMs); err != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, err
|
||
}
|
||
}
|
||
|
||
transactionID := transactionID(command.AppCode, command.CommandID)
|
||
availableAfter := account.AvailableAmount + mutation.AvailableDelta
|
||
frozenAfter := account.FrozenAmount + mutation.FrozenDelta
|
||
metadata := salaryWithdrawalMetadata{
|
||
AppCode: command.AppCode,
|
||
UserID: command.UserID,
|
||
SalaryAssetType: command.SalaryAssetType,
|
||
SalaryUSDMinor: command.SalaryUSDMinor,
|
||
PointFeeAmount: command.PointFeeAmount,
|
||
PointNetAmount: command.PointNetAmount,
|
||
PointsPerUSD: command.PointsPerUSD,
|
||
PointWithdrawFeeBPS: command.PointWithdrawFeeBPS,
|
||
PointPolicyInstanceCode: pointPolicyInstanceCode,
|
||
GrossUSDMinor: grossUSDMinor,
|
||
FeeUSDMinor: feeUSDMinor,
|
||
NetUSDMinor: netUSDMinor,
|
||
PointPolicyID: pointPolicyID,
|
||
PointPolicyVersion: pointPolicyVersion,
|
||
CoinsPerUSD: coinsPerUSD,
|
||
OperatorUserID: command.OperatorUserID,
|
||
Reason: command.Reason,
|
||
WithdrawalRef: command.WithdrawalRef,
|
||
WithdrawalApplicationID: command.WithdrawalApplicationID,
|
||
RegionID: command.RegionID,
|
||
AvailableAfter: availableAfter,
|
||
FrozenAfter: frozenAfter,
|
||
Version: account.Version + 1,
|
||
CreatedAtMS: nowMs,
|
||
}
|
||
if err := r.insertTransaction(ctx, tx, transactionID, command.CommandID, mutation.BizType, requestHash, strings.TrimSpace(mutation.ExternalRef), metadata, nowMs); err != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, err
|
||
}
|
||
if err := r.applyAccountDelta(ctx, tx, account, mutation.AvailableDelta, mutation.FrozenDelta, nowMs); err != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, err
|
||
}
|
||
if err := r.insertEntry(ctx, tx, walletEntry{
|
||
TransactionID: transactionID,
|
||
UserID: command.UserID,
|
||
AssetType: command.SalaryAssetType,
|
||
AvailableDelta: mutation.AvailableDelta,
|
||
FrozenDelta: mutation.FrozenDelta,
|
||
AvailableAfter: availableAfter,
|
||
FrozenAfter: frozenAfter,
|
||
CreatedAtMS: nowMs,
|
||
}); err != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, err
|
||
}
|
||
if err := r.insertWalletOutbox(ctx, tx, []walletOutboxEvent{
|
||
balanceChangedEvent(transactionID, command.CommandID, command.UserID, command.SalaryAssetType, mutation.AvailableDelta, mutation.FrozenDelta, availableAfter, frozenAfter, account.Version+1, metadata, nowMs, mutation.BizType),
|
||
}); err != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, err
|
||
}
|
||
return receiptFromSalaryWithdrawalMetadata(transactionID, metadata), nil
|
||
}
|
||
|
||
func (r *Repository) loadPointDiamondFreezeMetadata(ctx context.Context, tx *sql.Tx, command ledger.SalaryWithdrawalCommand) (salaryWithdrawalMetadata, error) {
|
||
freezeTransactionID := strings.TrimSpace(command.FreezeTransactionID)
|
||
if freezeTransactionID == "" {
|
||
return salaryWithdrawalMetadata{}, xerr.New(xerr.InvalidArgument, "point diamond freeze transaction id is required")
|
||
}
|
||
var metadataJSON string
|
||
err := tx.QueryRowContext(ctx, `
|
||
SELECT COALESCE(CAST(metadata_json AS CHAR), '{}')
|
||
FROM wallet_transactions
|
||
WHERE app_code = ? AND transaction_id = ? AND biz_type = ?
|
||
LIMIT 1
|
||
FOR UPDATE`, appcode.FromContext(ctx), freezeTransactionID, bizTypeSalaryWithdrawalFreeze).Scan(&metadataJSON)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return salaryWithdrawalMetadata{}, xerr.New(xerr.NotFound, "point diamond freeze transaction not found")
|
||
}
|
||
if err != nil {
|
||
return salaryWithdrawalMetadata{}, err
|
||
}
|
||
var metadata salaryWithdrawalMetadata
|
||
if err := json.Unmarshal([]byte(metadataJSON), &metadata); err != nil {
|
||
return salaryWithdrawalMetadata{}, err
|
||
}
|
||
if appcode.Normalize(metadata.AppCode) != appcode.Normalize(command.AppCode) || metadata.UserID != command.UserID ||
|
||
strings.ToUpper(strings.TrimSpace(metadata.SalaryAssetType)) != ledger.AssetPointDiamond || metadata.SalaryUSDMinor != command.SalaryUSDMinor ||
|
||
metadata.PointFeeAmount != command.PointFeeAmount || metadata.PointNetAmount != command.PointNetAmount ||
|
||
metadata.PointsPerUSD != command.PointsPerUSD || metadata.PointWithdrawFeeBPS != command.PointWithdrawFeeBPS {
|
||
return salaryWithdrawalMetadata{}, xerr.New(xerr.IdempotencyConflict, "point diamond terminal snapshot conflicts with freeze transaction")
|
||
}
|
||
if metadata.PointPolicyID == 0 || metadata.PointPolicyVersion == 0 || metadata.PointsPerUSD <= 0 || metadata.CoinsPerUSD <= 0 ||
|
||
metadata.GrossUSDMinor <= 0 || metadata.FeeUSDMinor < 0 || metadata.NetUSDMinor < 0 || metadata.FeeUSDMinor+metadata.NetUSDMinor != metadata.GrossUSDMinor {
|
||
return salaryWithdrawalMetadata{}, xerr.New(xerr.Internal, "point diamond freeze policy snapshot is invalid")
|
||
}
|
||
return metadata, nil
|
||
}
|
||
|
||
func salaryAssetUsesHostPolicyWithdrawal(assetType string) bool {
|
||
return ledger.ValidPointWithdrawalAssetType(assetType) || ledger.UsesHostAgencySalaryPolicy(assetType)
|
||
}
|
||
|
||
func salaryWithdrawalRequestHash(command ledger.SalaryWithdrawalCommand, bizType string) string {
|
||
if bizType == bizTypeSalaryWithdrawalFreeze && ledger.ValidPointWithdrawalAssetType(command.SalaryAssetType) {
|
||
// POINT 的费率、换算比例和 region 均由服务端按当前用户/政策注入，同一客户端 command 重试时这些快照可能已经变化。
|
||
// 哈希只锁定用户稳定提现意图；命中旧 transaction 后直接返回首次账变快照，既不重新选政策，也不重复消耗次数。
|
||
return stableHash(fmt.Sprintf("point_withdrawal_freeze|%s|%d|%s|%d|%d|%s",
|
||
appcode.Normalize(command.AppCode),
|
||
command.UserID,
|
||
strings.ToUpper(strings.TrimSpace(command.SalaryAssetType)),
|
||
command.SalaryUSDMinor,
|
||
command.GrossUSDMinor,
|
||
strings.TrimSpace(command.WithdrawalRef),
|
||
))
|
||
}
|
||
if strings.TrimSpace(command.WithdrawalApplicationID) != "" && (bizType == bizTypeSalaryWithdrawalSettle || bizType == bizTypeSalaryWithdrawalRelease) {
|
||
// 审核人和备注是钱包 metadata 快照，不决定资金突变。将它们排除后，admin 事务失败换人/改备注重试仍命中原交易。
|
||
// bizType 仍在哈希中，且 lookupSalaryWithdrawalTerminal 会在申请维度检查相反终局，settle/release 绝不能共存。
|
||
return stableHash(fmt.Sprintf("salary_withdrawal_terminal|%s|%s|%d|%s|%d|%d|%d|%d|%d|%s",
|
||
bizType,
|
||
appcode.Normalize(command.AppCode),
|
||
command.UserID,
|
||
strings.ToUpper(strings.TrimSpace(command.SalaryAssetType)),
|
||
command.SalaryUSDMinor,
|
||
command.PointFeeAmount,
|
||
command.PointNetAmount,
|
||
command.PointsPerUSD,
|
||
command.PointWithdrawFeeBPS,
|
||
strings.TrimSpace(command.WithdrawalApplicationID),
|
||
))
|
||
}
|
||
// 冻结和申请创建失败的回滚没有 application id，继续使用旧哈希保持已有 command 兼容。
|
||
return stableHash(fmt.Sprintf("salary_withdrawal|%s|%s|%d|%s|%d|%d|%d|%d|%d|%d|%s|%s|%s",
|
||
bizType,
|
||
appcode.Normalize(command.AppCode),
|
||
command.UserID,
|
||
strings.ToUpper(strings.TrimSpace(command.SalaryAssetType)),
|
||
command.SalaryUSDMinor,
|
||
command.PointFeeAmount,
|
||
command.PointNetAmount,
|
||
command.PointsPerUSD,
|
||
command.PointWithdrawFeeBPS,
|
||
command.OperatorUserID,
|
||
strings.TrimSpace(command.Reason),
|
||
strings.TrimSpace(command.WithdrawalRef),
|
||
strings.TrimSpace(command.WithdrawalApplicationID),
|
||
))
|
||
}
|
||
|
||
// lookupSalaryWithdrawalTerminal 先用小型主键锁表串行化同一申请，再使用 (app_code, external_ref) 现有索引查找旧/新终局。
|
||
// InnoDB 纯 gap lock 之间可兼容，不能单独作为“未存在终局”的互斥；显式锁行使首次并发 settle/release 也只有一个能落账。
|
||
func (r *Repository) lookupSalaryWithdrawalTerminal(ctx context.Context, tx *sql.Tx, command ledger.SalaryWithdrawalCommand, requestedBizType string) (ledger.SalaryWithdrawalReceipt, bool, error) {
|
||
applicationID := strings.TrimSpace(command.WithdrawalApplicationID)
|
||
if applicationID == "" || (requestedBizType != bizTypeSalaryWithdrawalSettle && requestedBizType != bizTypeSalaryWithdrawalRelease) {
|
||
return ledger.SalaryWithdrawalReceipt{}, false, nil
|
||
}
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO wallet_withdrawal_terminal_locks (
|
||
app_code, withdrawal_application_id, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE withdrawal_application_id = VALUES(withdrawal_application_id)`,
|
||
appcode.FromContext(ctx), applicationID, nowMS, nowMS,
|
||
); err != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, false, err
|
||
}
|
||
rows, err := tx.QueryContext(ctx, `
|
||
SELECT transaction_id, biz_type, COALESCE(CAST(metadata_json AS CHAR), '{}')
|
||
FROM wallet_transactions
|
||
WHERE app_code = ?
|
||
AND external_ref = ?
|
||
AND biz_type IN (?, ?)
|
||
ORDER BY created_at_ms ASC, transaction_id ASC
|
||
FOR UPDATE`, appcode.FromContext(ctx), applicationID, bizTypeSalaryWithdrawalSettle, bizTypeSalaryWithdrawalRelease)
|
||
if err != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, false, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
var terminal *struct {
|
||
TransactionID string
|
||
BizType string
|
||
Metadata salaryWithdrawalMetadata
|
||
}
|
||
for rows.Next() {
|
||
var transactionID string
|
||
var bizType string
|
||
var metadataJSON string
|
||
if err := rows.Scan(&transactionID, &bizType, &metadataJSON); err != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, false, err
|
||
}
|
||
if terminal != nil {
|
||
// 历史如果已经存在多个终局，钱包不能猜测哪个是真实决策，必须 fail closed 交由人工对账。
|
||
return ledger.SalaryWithdrawalReceipt{}, true, xerr.New(xerr.IdempotencyConflict, "withdrawal application has multiple wallet terminal transactions")
|
||
}
|
||
var metadata salaryWithdrawalMetadata
|
||
if err := json.Unmarshal([]byte(metadataJSON), &metadata); err != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, false, err
|
||
}
|
||
terminal = &struct {
|
||
TransactionID string
|
||
BizType string
|
||
Metadata salaryWithdrawalMetadata
|
||
}{TransactionID: transactionID, BizType: bizType, Metadata: metadata}
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, false, err
|
||
}
|
||
if terminal == nil {
|
||
return ledger.SalaryWithdrawalReceipt{}, false, nil
|
||
}
|
||
if terminal.BizType != requestedBizType || !salaryWithdrawalTerminalMatches(command, terminal.Metadata) {
|
||
return ledger.SalaryWithdrawalReceipt{}, true, xerr.New(xerr.IdempotencyConflict, "withdrawal application terminal idempotency conflict")
|
||
}
|
||
return receiptFromSalaryWithdrawalMetadata(terminal.TransactionID, terminal.Metadata), true, nil
|
||
}
|
||
|
||
func salaryWithdrawalTerminalMatches(command ledger.SalaryWithdrawalCommand, metadata salaryWithdrawalMetadata) bool {
|
||
return appcode.Normalize(metadata.AppCode) == appcode.Normalize(command.AppCode) &&
|
||
metadata.UserID == command.UserID &&
|
||
strings.ToUpper(strings.TrimSpace(metadata.SalaryAssetType)) == strings.ToUpper(strings.TrimSpace(command.SalaryAssetType)) &&
|
||
metadata.SalaryUSDMinor == command.SalaryUSDMinor &&
|
||
metadata.PointFeeAmount == command.PointFeeAmount &&
|
||
metadata.PointNetAmount == command.PointNetAmount &&
|
||
metadata.PointsPerUSD == command.PointsPerUSD &&
|
||
metadata.PointWithdrawFeeBPS == command.PointWithdrawFeeBPS &&
|
||
strings.TrimSpace(metadata.WithdrawalApplicationID) == strings.TrimSpace(command.WithdrawalApplicationID)
|
||
}
|