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) }