From ac767d1c7e2d74ada08eb1dab0e11d13c21df444 Mon Sep 17 00:00:00 2001 From: local Date: Tue, 23 Jun 2026 23:33:28 +0800 Subject: [PATCH] fix coin seller stock deduction accounting --- api/proto/wallet/v1/wallet.pb.go | 2 +- api/proto/wallet/v1/wallet.proto | 2 +- .../internal/modules/coinledger/service.go | 8 ++- .../modules/coinledger/service_test.go | 9 ++-- .../admin/internal/modules/hostorg/handler.go | 6 +-- .../admin/internal/modules/hostorg/request.go | 9 ++-- .../internal/modules/hostorg/response.go | 36 +++++++------ .../admin/internal/modules/hostorg/service.go | 45 +++++++++------- .../cmd/backfill-last7/main.go | 2 +- .../statistics-service/internal/app/app.go | 4 +- .../internal/app/local_flow_test.go | 27 +++++++++- .../internal/domain/ledger/coin_seller.go | 2 + .../internal/domain/ledger/constants.go | 2 + .../internal/service/wallet/coin_seller.go | 6 +-- .../internal/service/wallet/service_test.go | 52 +++++++++++++++++++ .../storage/mysql/coin_seller_ledger.go | 40 ++++++++++---- .../internal/storage/mysql/constants.go | 1 + .../internal/storage/mysql/outbox.go | 2 +- 18 files changed, 190 insertions(+), 65 deletions(-) diff --git a/api/proto/wallet/v1/wallet.pb.go b/api/proto/wallet/v1/wallet.pb.go index 81c5aee6..fbcc6da8 100644 --- a/api/proto/wallet/v1/wallet.pb.go +++ b/api/proto/wallet/v1/wallet.pb.go @@ -1749,7 +1749,7 @@ func (x *AdminCreditAssetResponse) GetBalance() *AssetBalance { return nil } -// AdminCreditCoinSellerStockRequest 是后台给 active 币商专用库存入账的账务命令。 +// AdminCreditCoinSellerStockRequest 是后台调整币商专用库存的账务命令。 type AdminCreditCoinSellerStockRequest struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache diff --git a/api/proto/wallet/v1/wallet.proto b/api/proto/wallet/v1/wallet.proto index e719a7a9..10871578 100644 --- a/api/proto/wallet/v1/wallet.proto +++ b/api/proto/wallet/v1/wallet.proto @@ -212,7 +212,7 @@ message AdminCreditAssetResponse { AssetBalance balance = 2; } -// AdminCreditCoinSellerStockRequest 是后台给 active 币商专用库存入账的账务命令。 +// AdminCreditCoinSellerStockRequest 是后台调整币商专用库存的账务命令。 message AdminCreditCoinSellerStockRequest { string command_id = 1; int64 seller_user_id = 2; diff --git a/server/admin/internal/modules/coinledger/service.go b/server/admin/internal/modules/coinledger/service.go index 13a99ad5..264565f0 100644 --- a/server/admin/internal/modules/coinledger/service.go +++ b/server/admin/internal/modules/coinledger/service.go @@ -24,6 +24,7 @@ const ( coinSellerTransferBizType = "coin_seller_transfer" coinSellerRechargeBizType = "coin_seller_recharge" coinSellerStockPurchaseBizType = "coin_seller_stock_purchase" + coinSellerStockDeductionBizType = "coin_seller_stock_deduction" coinSellerCoinCompensationBizType = "coin_seller_coin_compensation" salaryTransferToCoinSellerBizType = "salary_transfer_to_coin_seller" coinSellerLedgerTypeAdminStockCredit = "admin_stock_credit" @@ -782,6 +783,7 @@ func coinSellerLedgerBizTypes(ledgerType string) ([]string, error) { // 空类型表示“全部币商流水”，但仍只允许当前产品定义的公开币商库存口径，不能把其他内部账带出来。 return []string{ coinSellerStockPurchaseBizType, + coinSellerStockDeductionBizType, coinSellerRechargeBizType, coinSellerCoinCompensationBizType, coinSellerTransferBizType, @@ -790,7 +792,7 @@ func coinSellerLedgerBizTypes(ledgerType string) ([]string, error) { }, nil case coinSellerLedgerTypeAdminStockCredit: // 进货操作是运营库存口径，底层包含后台 USDT 进货、H5 三方币商充值、金币补偿和后台扣除；资产条件仍限定在 COIN_SELLER_COIN。 - return []string{coinSellerStockPurchaseBizType, coinSellerRechargeBizType, coinSellerCoinCompensationBizType, coinManualCreditBizType}, nil + return []string{coinSellerStockPurchaseBizType, coinSellerStockDeductionBizType, coinSellerRechargeBizType, coinSellerCoinCompensationBizType, coinManualCreditBizType}, nil case coinSellerLedgerTypeSellerTransfer: // 币商转用户只展示币商侧出账分录，收款用户资料在展示投影里补齐。 return []string{coinSellerTransferBizType}, nil @@ -804,7 +806,7 @@ func coinSellerLedgerBizTypes(ledgerType string) ([]string, error) { func coinSellerLedgerTypeForBizType(bizType string) string { switch bizType { - case coinSellerStockPurchaseBizType, coinSellerRechargeBizType, coinSellerCoinCompensationBizType: + case coinSellerStockPurchaseBizType, coinSellerStockDeductionBizType, coinSellerRechargeBizType, coinSellerCoinCompensationBizType: return coinSellerLedgerTypeAdminStockCredit case coinManualCreditBizType: return coinSellerLedgerTypeAdminStockCredit @@ -821,6 +823,8 @@ func coinSellerLedgerLabel(item coinSellerLedgerDTO) string { switch item.BizType { case coinSellerStockPurchaseBizType: return "USDT进货" + case coinSellerStockDeductionBizType: + return "USDT扣除" case coinSellerRechargeBizType: return "三方充值" case coinSellerCoinCompensationBizType: diff --git a/server/admin/internal/modules/coinledger/service_test.go b/server/admin/internal/modules/coinledger/service_test.go index 42e764fd..aeda45ee 100644 --- a/server/admin/internal/modules/coinledger/service_test.go +++ b/server/admin/internal/modules/coinledger/service_test.go @@ -51,10 +51,10 @@ func TestCoinSellerLedgerWhereMapsAdminStockCredit(t *testing.T) { if err != nil { t.Fatalf("coin seller ledger where failed: %v", err) } - if want := "WHERE e.app_code = ? AND e.asset_type = ? AND wt.biz_type IN (?,?,?,?) AND e.user_id IN (?,?)"; where != want { + if want := "WHERE e.app_code = ? AND e.asset_type = ? AND wt.biz_type IN (?,?,?,?,?) AND e.user_id IN (?,?)"; where != want { t.Fatalf("where mismatch:\nwant %s\n got %s", want, where) } - if len(args) != 8 || args[0] != "lalu" || args[1] != coinSellerAssetType || args[2] != coinSellerStockPurchaseBizType || args[3] != coinSellerRechargeBizType || args[4] != coinSellerCoinCompensationBizType || args[5] != coinManualCreditBizType || args[6] != int64(3001) || args[7] != int64(3002) { + if len(args) != 9 || args[0] != "lalu" || args[1] != coinSellerAssetType || args[2] != coinSellerStockPurchaseBizType || args[3] != coinSellerStockDeductionBizType || args[4] != coinSellerRechargeBizType || args[5] != coinSellerCoinCompensationBizType || args[6] != coinManualCreditBizType || args[7] != int64(3001) || args[8] != int64(3002) { t.Fatalf("args mismatch: %#v", args) } } @@ -64,10 +64,10 @@ func TestCoinSellerLedgerWhereEmptyTypeUsesAllPublicTypes(t *testing.T) { if err != nil { t.Fatalf("coin seller ledger where failed: %v", err) } - if want := "WHERE e.app_code = ? AND e.asset_type = ? AND wt.biz_type IN (?,?,?,?,?,?)"; where != want { + if want := "WHERE e.app_code = ? AND e.asset_type = ? AND wt.biz_type IN (?,?,?,?,?,?,?)"; where != want { t.Fatalf("where mismatch:\nwant %s\n got %s", want, where) } - if len(args) != 8 || args[0] != "lalu" || args[1] != coinSellerAssetType || args[2] != coinSellerStockPurchaseBizType || args[3] != coinSellerRechargeBizType || args[4] != coinSellerCoinCompensationBizType || args[5] != coinSellerTransferBizType || args[6] != salaryTransferToCoinSellerBizType || args[7] != coinManualCreditBizType { + if len(args) != 9 || args[0] != "lalu" || args[1] != coinSellerAssetType || args[2] != coinSellerStockPurchaseBizType || args[3] != coinSellerStockDeductionBizType || args[4] != coinSellerRechargeBizType || args[5] != coinSellerCoinCompensationBizType || args[6] != coinSellerTransferBizType || args[7] != salaryTransferToCoinSellerBizType || args[8] != coinManualCreditBizType { t.Fatalf("args mismatch: %#v", args) } } @@ -117,6 +117,7 @@ func TestCoinSellerLedgerLabelUsesConcreteStockAndDebitType(t *testing.T) { want string }{ {name: "usdt purchase", item: coinSellerLedgerDTO{BizType: coinSellerStockPurchaseBizType, LedgerType: coinSellerLedgerTypeAdminStockCredit}, want: "USDT进货"}, + {name: "usdt deduction", item: coinSellerLedgerDTO{BizType: coinSellerStockDeductionBizType, LedgerType: coinSellerLedgerTypeAdminStockCredit}, want: "USDT扣除"}, {name: "third party recharge", item: coinSellerLedgerDTO{BizType: coinSellerRechargeBizType, LedgerType: coinSellerLedgerTypeAdminStockCredit}, want: "三方充值"}, {name: "compensation", item: coinSellerLedgerDTO{BizType: coinSellerCoinCompensationBizType, LedgerType: coinSellerLedgerTypeAdminStockCredit}, want: "金币补偿"}, {name: "debit", item: coinSellerLedgerDTO{BizType: coinManualCreditBizType, LedgerType: coinSellerLedgerTypeAdminStockCredit, AvailableDelta: -100}, want: "金币扣除"}, diff --git a/server/admin/internal/modules/hostorg/handler.go b/server/admin/internal/modules/hostorg/handler.go index 244ef5f3..3486444b 100644 --- a/server/admin/internal/modules/hostorg/handler.go +++ b/server/admin/internal/modules/hostorg/handler.go @@ -340,9 +340,9 @@ func (h *Handler) DebitCoinSellerStock(c *gin.Context) { response.BadRequest(c, err.Error()) return } - writeHostOrgAuditLog(c, h.audit, "coin-seller-stock-debit", "wallet_entries", sellerUserID, - fmt.Sprintf("command_id=%s seller_user_id=%d coin_amount=%d transaction_id=%s reason=%q", - strings.TrimSpace(req.CommandID), sellerUserID, req.CoinAmount, result.GetTransactionId(), strings.TrimSpace(req.Reason))) + writeHostOrgAuditLog(c, h.audit, "coin-seller-stock-debit", "coin_seller_stock_records", sellerUserID, + fmt.Sprintf("command_id=%s seller_user_id=%d stock_type=%s coin_amount=%d paid_currency_code=%s paid_amount_micro=%d transaction_id=%s reason=%q", + strings.TrimSpace(req.CommandID), sellerUserID, result.GetStockType(), result.GetCoinAmount(), result.GetPaidCurrencyCode(), result.GetPaidAmountMicro(), result.GetTransactionId(), strings.TrimSpace(req.Reason))) response.Created(c, coinSellerStockDebitFromProto(sellerUserID, req.CoinAmount, result)) } diff --git a/server/admin/internal/modules/hostorg/request.go b/server/admin/internal/modules/hostorg/request.go index b7116681..307172a1 100644 --- a/server/admin/internal/modules/hostorg/request.go +++ b/server/admin/internal/modules/hostorg/request.go @@ -171,10 +171,11 @@ type coinSellerStockCreditRequest struct { } type coinSellerStockDebitRequest struct { - CommandID string `json:"commandId" binding:"required"` - CoinAmount int64 `json:"coinAmount" binding:"required"` - EvidenceRef string `json:"evidenceRef"` - Reason string `json:"reason" binding:"required"` + CommandID string `json:"commandId" binding:"required"` + CoinAmount int64 `json:"coinAmount" binding:"required"` + RechargeAmount string `json:"rechargeAmount" binding:"required"` + EvidenceRef string `json:"evidenceRef"` + Reason string `json:"reason" binding:"required"` } type replaceCoinSellerSalaryRatesRequest struct { diff --git a/server/admin/internal/modules/hostorg/response.go b/server/admin/internal/modules/hostorg/response.go index df08b5e6..733c3925 100644 --- a/server/admin/internal/modules/hostorg/response.go +++ b/server/admin/internal/modules/hostorg/response.go @@ -15,11 +15,16 @@ type coinSellerStockCreditResponse struct { } type coinSellerStockDebitResponse struct { - TransactionID string `json:"transactionId"` - SellerUserID int64 `json:"sellerUserId,string"` - CoinAmount int64 `json:"coinAmount"` - AvailableDelta int64 `json:"availableDelta"` - BalanceAfter int64 `json:"balanceAfter"` + TransactionID string `json:"transactionId"` + SellerUserID int64 `json:"sellerUserId,string"` + StockType string `json:"stockType"` + CoinAmount int64 `json:"coinAmount"` + AvailableDelta int64 `json:"availableDelta"` + PaidCurrencyCode string `json:"paidCurrencyCode"` + PaidAmountMicro int64 `json:"paidAmountMicro"` + CountsAsSellerRecharge bool `json:"countsAsSellerRecharge"` + BalanceAfter int64 `json:"balanceAfter"` + CreatedAtMS int64 `json:"createdAtMs"` } func coinSellerStockCreditFromProto(item *walletv1.AdminCreditCoinSellerStockResponse) coinSellerStockCreditResponse { @@ -39,19 +44,20 @@ func coinSellerStockCreditFromProto(item *walletv1.AdminCreditCoinSellerStockRes } } -func coinSellerStockDebitFromProto(sellerUserID int64, coinAmount int64, item *walletv1.AdminCreditAssetResponse) coinSellerStockDebitResponse { +func coinSellerStockDebitFromProto(sellerUserID int64, coinAmount int64, item *walletv1.AdminCreditCoinSellerStockResponse) coinSellerStockDebitResponse { if item == nil { return coinSellerStockDebitResponse{SellerUserID: sellerUserID, CoinAmount: coinAmount, AvailableDelta: -coinAmount} } - balanceAfter := int64(0) - if item.GetBalance() != nil { - balanceAfter = item.GetBalance().GetAvailableAmount() - } return coinSellerStockDebitResponse{ - TransactionID: item.GetTransactionId(), - SellerUserID: sellerUserID, - CoinAmount: coinAmount, - AvailableDelta: -coinAmount, - BalanceAfter: balanceAfter, + TransactionID: item.GetTransactionId(), + SellerUserID: sellerUserID, + StockType: item.GetStockType(), + CoinAmount: item.GetCoinAmount(), + AvailableDelta: item.GetCoinAmount(), + PaidCurrencyCode: item.GetPaidCurrencyCode(), + PaidAmountMicro: item.GetPaidAmountMicro(), + CountsAsSellerRecharge: item.GetCountsAsSellerRecharge(), + BalanceAfter: item.GetBalanceAfter(), + CreatedAtMS: item.GetCreatedAtMs(), } } diff --git a/server/admin/internal/modules/hostorg/service.go b/server/admin/internal/modules/hostorg/service.go index d5940b7b..a9fe8364 100644 --- a/server/admin/internal/modules/hostorg/service.go +++ b/server/admin/internal/modules/hostorg/service.go @@ -13,9 +13,9 @@ import ( ) const ( - coinSellerStatusActive = "active" coinSellerMerchantAssetType = "COIN_SELLER_COIN" coinSellerStockTypePurchase = "usdt_purchase" + coinSellerStockTypeDeduction = "usdt_deduction" coinSellerStockTypeCompensate = "coin_compensation" paidCurrencyUSDT = "USDT" sortByDiamond = "diamond" @@ -312,7 +312,7 @@ func (s *Service) CreditCoinSellerStock(ctx context.Context, actorID int64, sell return nil, fmt.Errorf("stock type is invalid") } - seller, err := s.activeCoinSellerUser(ctx, sellerUserID, requestID) + seller, err := s.coinSellerUser(ctx, sellerUserID, requestID) if err != nil { return nil, err } @@ -334,7 +334,7 @@ func (s *Service) CreditCoinSellerStock(ctx context.Context, actorID int64, sell }) } -func (s *Service) DebitCoinSellerStock(ctx context.Context, actorID int64, sellerUserID int64, requestID string, req coinSellerStockDebitRequest) (*walletv1.AdminCreditAssetResponse, error) { +func (s *Service) DebitCoinSellerStock(ctx context.Context, actorID int64, sellerUserID int64, requestID string, req coinSellerStockDebitRequest) (*walletv1.AdminCreditCoinSellerStockResponse, error) { if s.walletClient == nil { return nil, fmt.Errorf("wallet client is not configured") } @@ -353,24 +353,33 @@ func (s *Service) DebitCoinSellerStock(ctx context.Context, actorID int64, selle if len(reason) > 512 || len(commandID) > 128 || len(evidenceRef) > 256 { return nil, fmt.Errorf("coin seller debit text fields are too long") } - if _, err := s.activeCoinSellerUser(ctx, sellerUserID, requestID); err != nil { + paidAmountMicro, err := parseUSDTAmountMicro(req.RechargeAmount) + if err != nil { return nil, err } - // 币商扣除是运营对币商专用库存的负向调账：只扣 COIN_SELLER_COIN，不写普通 COIN，也不写进货记录。 - // wallet-service 的 AdminCreditAsset 负责余额不足校验、幂等冲突检测、wallet_entries 和 outbox 原子落账。 - return s.walletClient.AdminCreditAsset(ctx, &walletv1.AdminCreditAssetRequest{ - CommandId: commandID, - TargetUserId: sellerUserID, - AssetType: coinSellerMerchantAssetType, - Amount: -req.CoinAmount, - OperatorUserId: actorID, - Reason: reason, - EvidenceRef: evidenceRef, - AppCode: appctx.FromContext(ctx), + seller, err := s.coinSellerUser(ctx, sellerUserID, requestID) + if err != nil { + return nil, err + } + // 币商扣除按 USDT 进货冲回处理：钱包事务会同时写负向库存分录、负向进货记录和统计 outbox， + // 这样后台列表、币商流水和充值聚合都使用同一条账务事实，避免只扣余额却不冲收入。 + return s.walletClient.AdminCreditCoinSellerStock(ctx, &walletv1.AdminCreditCoinSellerStockRequest{ + CommandId: commandID, + SellerUserId: sellerUserID, + SellerCountryId: seller.CountryID, + SellerRegionId: seller.RegionID, + StockType: coinSellerStockTypeDeduction, + CoinAmount: req.CoinAmount, + PaidCurrencyCode: paidCurrencyUSDT, + PaidAmountMicro: paidAmountMicro, + EvidenceRef: evidenceRef, + OperatorUserId: actorID, + Reason: reason, + AppCode: appctx.FromContext(ctx), }) } -func (s *Service) activeCoinSellerUser(ctx context.Context, sellerUserID int64, requestID string) (*userclient.User, error) { +func (s *Service) coinSellerUser(ctx context.Context, sellerUserID int64, requestID string) (*userclient.User, error) { if s == nil || s.userClient == nil { return nil, fmt.Errorf("user client is not configured") } @@ -382,8 +391,8 @@ func (s *Service) activeCoinSellerUser(ctx context.Context, sellerUserID int64, if err != nil { return nil, err } - if profile == nil || profile.Status != coinSellerStatusActive || profile.MerchantAssetType != coinSellerMerchantAssetType { - return nil, fmt.Errorf("target user is not an active coin seller") + if profile == nil || profile.MerchantAssetType != coinSellerMerchantAssetType { + return nil, fmt.Errorf("target user is not a coin seller") } seller, err := s.userClient.GetUser(ctx, userclient.GetUserRequest{ RequestID: requestID, diff --git a/services/statistics-service/cmd/backfill-last7/main.go b/services/statistics-service/cmd/backfill-last7/main.go index 851bdbe0..47f9bc60 100644 --- a/services/statistics-service/cmd/backfill-last7/main.go +++ b/services/statistics-service/cmd/backfill-last7/main.go @@ -436,7 +436,7 @@ func rebuildCoinSellerStock(ctx context.Context, db *sql.DB, state *rebuildState rows, err := db.QueryContext(ctx, ` SELECT seller_user_id, paid_amount_micro, created_at_ms FROM coin_seller_stock_records - WHERE app_code = ? AND counts_as_seller_recharge = 1 AND paid_amount_micro > 0 AND created_at_ms >= ? AND created_at_ms < ?`, + WHERE app_code = ? AND counts_as_seller_recharge = 1 AND paid_amount_micro <> 0 AND created_at_ms >= ? AND created_at_ms < ?`, app, startMS, endMS) if err != nil { return err diff --git a/services/statistics-service/internal/app/app.go b/services/statistics-service/internal/app/app.go index cc144814..426e2789 100644 --- a/services/statistics-service/internal/app/app.go +++ b/services/statistics-service/internal/app/app.go @@ -287,7 +287,7 @@ func coinSellerStockEvent(body []byte) (mysqlstorage.CoinSellerStockEvent, bool, return mysqlstorage.CoinSellerStockEvent{}, false, nil } usdMinor := amountMicroToUSDMinor(mysqlstorage.Int64(payload, "paid_amount_micro")) - if usdMinor <= 0 { + if usdMinor == 0 { return mysqlstorage.CoinSellerStockEvent{}, false, nil } return mysqlstorage.CoinSellerStockEvent{ @@ -336,7 +336,7 @@ func isCoinSellerRechargeType(value string) bool { } func amountMicroToUSDMinor(amountMicro int64) int64 { - if amountMicro <= 0 { + if amountMicro == 0 { return 0 } return amountMicro / 10_000 diff --git a/services/statistics-service/internal/app/local_flow_test.go b/services/statistics-service/internal/app/local_flow_test.go index a9450441..e8b18dc3 100644 --- a/services/statistics-service/internal/app/local_flow_test.go +++ b/services/statistics-service/internal/app/local_flow_test.go @@ -88,6 +88,31 @@ func TestLocalStatisticsDataScreenFlow(t *testing.T) { if err := repo.ConsumeCoinSellerStock(ctx, stock); err != nil { t.Fatalf("consume coin seller stock failed: %v", err) } + coinSellerStockDeduction := walletMessage(t, walletmq.WalletOutboxMessage{ + AppCode: appCode, + EventID: "WalletCoinSellerStockPurchased:deduction:1", + EventType: "WalletCoinSellerStockPurchased", + TransactionID: "wallet_tx_coin_seller_stock_deduction_1", + CommandID: "coin_seller_stock_deduction_1", + UserID: userID + 10, + PayloadJSON: mustJSON(t, map[string]any{ + "seller_user_id": userID + 10, + "country_id": countryID, + "region_id": regionID, + "coin_amount": int64(-50_000), + "paid_amount_micro": int64(-500_000), + "counts_as_seller_recharge": true, + "stock_type": "usdt_deduction", + }), + OccurredAtMS: occurredAt, + }) + stockDeduction, ok, err := coinSellerStockEvent(coinSellerStockDeduction) + if err != nil || !ok { + t.Fatalf("decode coin seller stock deduction failed: ok=%v err=%v", ok, err) + } + if err := repo.ConsumeCoinSellerStock(ctx, stockDeduction); err != nil { + t.Fatalf("consume coin seller stock deduction failed: %v", err) + } coinSellerRecharge := walletMessage(t, walletmq.WalletOutboxMessage{ AppCode: appCode, @@ -206,7 +231,7 @@ func TestLocalStatisticsDataScreenFlow(t *testing.T) { if err != nil { t.Fatalf("query overview failed: %v", err) } - if overview.GoogleRechargeUSDMinor != 150 || overview.CoinSellerRechargeUSDMinor != 400 || overview.CoinSellerTransferCoin != 160_000 || overview.RechargeUSDMinor != 550 || overview.NewUserRechargeUSDMinor != 150 || overview.PaidUsers != 2 { + if overview.GoogleRechargeUSDMinor != 150 || overview.CoinSellerRechargeUSDMinor != 350 || overview.CoinSellerTransferCoin != 160_000 || overview.RechargeUSDMinor != 500 || overview.NewUserRechargeUSDMinor != 150 || overview.PaidUsers != 2 { t.Fatalf("google recharge metrics mismatch: %+v", overview) } if overview.LuckyGiftTurnover != 1_250 || overview.LuckyGiftPayout != 300 || overview.LuckyGiftProfit != 950 || overview.LuckyGiftPayers != 1 { diff --git a/services/wallet-service/internal/domain/ledger/coin_seller.go b/services/wallet-service/internal/domain/ledger/coin_seller.go index c7bf8887..39b4469b 100644 --- a/services/wallet-service/internal/domain/ledger/coin_seller.go +++ b/services/wallet-service/internal/domain/ledger/coin_seller.go @@ -130,6 +130,8 @@ func NormalizeCoinSellerStockType(stockType string) string { switch strings.ToLower(strings.TrimSpace(stockType)) { case StockTypeUSDTPurchase: return StockTypeUSDTPurchase + case StockTypeUSDTDeduction: + return StockTypeUSDTDeduction case StockTypeCoinCompensation: return StockTypeCoinCompensation default: diff --git a/services/wallet-service/internal/domain/ledger/constants.go b/services/wallet-service/internal/domain/ledger/constants.go index 41597c65..96e30d6c 100644 --- a/services/wallet-service/internal/domain/ledger/constants.go +++ b/services/wallet-service/internal/domain/ledger/constants.go @@ -26,6 +26,8 @@ const ( // StockTypeUSDTPurchase 表示币商线下 USDT 进货后发放专用金币库存。 StockTypeUSDTPurchase = "usdt_purchase" + // StockTypeUSDTDeduction 表示运营按 USDT 入金口径冲回币商专用金币库存和充值金额。 + StockTypeUSDTDeduction = "usdt_deduction" // StockTypeCoinCompensation 表示平台人工补偿币商专用金币库存，不计入进货金额。 StockTypeCoinCompensation = "coin_compensation" // PaidCurrencyUSDT 是首版币商进货支持的唯一线下付款币种。 diff --git a/services/wallet-service/internal/service/wallet/coin_seller.go b/services/wallet-service/internal/service/wallet/coin_seller.go index 27a0bdff..d826dd38 100644 --- a/services/wallet-service/internal/service/wallet/coin_seller.go +++ b/services/wallet-service/internal/service/wallet/coin_seller.go @@ -8,7 +8,7 @@ import ( "strings" ) -// AdminCreditCoinSellerStock 只给 COIN_SELLER_COIN 库存入账；USDT 进货和金币补偿使用不同账务类型。 +// AdminCreditCoinSellerStock 只调整 COIN_SELLER_COIN 库存；USDT 进货、USDT 冲回和金币补偿使用不同账务类型。 func (s *Service) AdminCreditCoinSellerStock(ctx context.Context, command ledger.CoinSellerStockCreditCommand) (ledger.CoinSellerStockCreditReceipt, error) { if command.CommandID == "" || command.SellerUserID <= 0 || command.OperatorUserID <= 0 { return ledger.CoinSellerStockCreditReceipt{}, xerr.New(xerr.InvalidArgument, "coin seller stock command is incomplete") @@ -34,12 +34,12 @@ func (s *Service) AdminCreditCoinSellerStock(ctx context.Context, command ledger return ledger.CoinSellerStockCreditReceipt{}, xerr.New(xerr.InvalidArgument, "coin seller stock text fields are too long") } switch command.StockType { - case ledger.StockTypeUSDTPurchase: + case ledger.StockTypeUSDTPurchase, ledger.StockTypeUSDTDeduction: if command.PaidCurrencyCode == "" { command.PaidCurrencyCode = ledger.PaidCurrencyUSDT } if command.PaidCurrencyCode != ledger.PaidCurrencyUSDT || command.PaidAmountMicro <= 0 { - return ledger.CoinSellerStockCreditReceipt{}, xerr.New(xerr.CoinSellerStockAmountInvalid, "usdt purchase requires paid amount") + return ledger.CoinSellerStockCreditReceipt{}, xerr.New(xerr.CoinSellerStockAmountInvalid, "usdt stock adjustment requires paid amount") } case ledger.StockTypeCoinCompensation: if command.PaidCurrencyCode != "" || command.PaidAmountMicro != 0 || command.PaymentRef != "" { diff --git a/services/wallet-service/internal/service/wallet/service_test.go b/services/wallet-service/internal/service/wallet/service_test.go index 4bd0225a..3a3921f0 100644 --- a/services/wallet-service/internal/service/wallet/service_test.go +++ b/services/wallet-service/internal/service/wallet/service_test.go @@ -2983,6 +2983,58 @@ func TestAdminCreditCoinSellerStockPurchaseCreditsDedicatedBalance(t *testing.T) } } +// TestAdminCreditCoinSellerStockDeductionDebitsDedicatedBalanceAndRechargeAmount 验证 USDT 扣除会同时冲回库存和币商充值金额。 +func TestAdminCreditCoinSellerStockDeductionDebitsDedicatedBalanceAndRechargeAmount(t *testing.T) { + repository := mysqltest.NewRepository(t) + svc := walletservice.New(repository) + credit := ledger.CoinSellerStockCreditCommand{ + CommandID: "cmd-stock-usdt-deduct-seed", + SellerUserID: 33011, + SellerCountryID: 63, + SellerRegionID: 7, + StockType: ledger.StockTypeUSDTPurchase, + CoinAmount: 8000000, + PaidAmountMicro: 100000000, + OperatorUserID: 90001, + } + if _, err := svc.AdminCreditCoinSellerStock(context.Background(), credit); err != nil { + t.Fatalf("seed stock purchase failed: %v", err) + } + + deduction := ledger.CoinSellerStockCreditCommand{ + CommandID: "cmd-stock-usdt-deduct", + SellerUserID: 33011, + SellerCountryID: 63, + SellerRegionID: 7, + StockType: ledger.StockTypeUSDTDeduction, + CoinAmount: 3000000, + PaidAmountMicro: 37500000, + OperatorUserID: 90002, + Reason: "reverse bad stock", + } + first, err := svc.AdminCreditCoinSellerStock(context.Background(), deduction) + if err != nil { + t.Fatalf("AdminCreditCoinSellerStock deduction failed: %v", err) + } + second, err := svc.AdminCreditCoinSellerStock(context.Background(), deduction) + if err != nil { + t.Fatalf("AdminCreditCoinSellerStock deduction retry failed: %v", err) + } + if first.TransactionID == "" || first.TransactionID != second.TransactionID || first.BalanceAfter != 5000000 || first.CoinAmount != -3000000 || first.PaidAmountMicro != -37500000 || !first.CountsAsSellerRecharge { + t.Fatalf("stock deduction receipt mismatch: first=%+v second=%+v", first, second) + } + assertBalance(t, svc, 33011, ledger.AssetCoinSellerCoin, 5000000) + if got := repository.CountRows("coin_seller_stock_records", "transaction_id = ? AND stock_type = ? AND counts_as_seller_recharge = TRUE AND coin_amount = ? AND paid_amount_micro = ?", first.TransactionID, ledger.StockTypeUSDTDeduction, int64(-3000000), int64(-37500000)); got != 1 { + t.Fatalf("stock deduction should write one negative seller recharge record, got %d", got) + } + if got := repository.CountRows("wallet_entries", "transaction_id = ? AND available_delta = ?", first.TransactionID, int64(-3000000)); got != 1 { + t.Fatalf("stock deduction should write one negative wallet entry, got %d", got) + } + if got := repository.CountRows("wallet_outbox", "transaction_id = ? AND event_type = ? AND available_delta = ? AND JSON_EXTRACT(payload_json, '$.paid_amount_micro') = ?", first.TransactionID, "WalletCoinSellerStockPurchased", int64(-3000000), int64(-37500000)); got != 1 { + t.Fatalf("stock deduction should emit negative seller stock outbox event, got %d", got) + } +} + // TestAdminCreditCoinSellerStockCompensationRejectsPaidAmount 锁定补偿不能携带充值金额的边界。 func TestAdminCreditCoinSellerStockCompensationRejectsPaidAmount(t *testing.T) { repository := mysqltest.NewRepository(t) diff --git a/services/wallet-service/internal/storage/mysql/coin_seller_ledger.go b/services/wallet-service/internal/storage/mysql/coin_seller_ledger.go index f8c6bb74..1d880009 100644 --- a/services/wallet-service/internal/storage/mysql/coin_seller_ledger.go +++ b/services/wallet-service/internal/storage/mysql/coin_seller_ledger.go @@ -14,7 +14,7 @@ import ( "hyapp/services/wallet-service/internal/domain/ledger" ) -// AdminCreditCoinSellerStock 在同一事务里给币商 COIN_SELLER_COIN 库存入账，并记录专用进货明细。 +// 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") @@ -49,26 +49,32 @@ func (r *Repository) AdminCreditCoinSellerStock(ctx context.Context, command led } nowMs := time.Now().UnixMilli() - account, err := r.lockAccount(ctx, tx, command.SellerUserID, ledger.AssetCoinSellerCoin, true, nowMs) + availableDelta := coinSellerStockAvailableDelta(command) + account, err := r.lockAccount(ctx, tx, command.SellerUserID, ledger.AssetCoinSellerCoin, availableDelta > 0, nowMs) if err != nil { return ledger.CoinSellerStockCreditReceipt{}, err } - balanceAfter, err := checkedAdd(account.AvailableAmount, command.CoinAmount) + if availableDelta < 0 && account.AvailableAmount < command.CoinAmount { + return ledger.CoinSellerStockCreditReceipt{}, xerr.New(xerr.InsufficientBalance, "insufficient balance") + } + balanceAfter, err := checkedAdd(account.AvailableAmount, availableDelta) if err != nil { return ledger.CoinSellerStockCreditReceipt{}, err } transactionID := transactionID(command.AppCode, command.CommandID) - countsAsSellerRecharge := command.StockType == ledger.StockTypeUSDTPurchase + countsAsSellerRecharge := command.StockType == ledger.StockTypeUSDTPurchase || command.StockType == ledger.StockTypeUSDTDeduction + recordCoinAmount := availableDelta + recordPaidAmountMicro := coinSellerStockPaidAmountMicro(command) metadata := coinSellerStockMetadata{ AppCode: command.AppCode, SellerUserID: command.SellerUserID, SellerCountryID: command.SellerCountryID, SellerRegionID: command.SellerRegionID, StockType: command.StockType, - CoinAmount: command.CoinAmount, + CoinAmount: recordCoinAmount, PaidCurrencyCode: command.PaidCurrencyCode, - PaidAmountMicro: command.PaidAmountMicro, + PaidAmountMicro: recordPaidAmountMicro, PaymentRef: command.PaymentRef, EvidenceRef: command.EvidenceRef, OperatorUserID: command.OperatorUserID, @@ -85,14 +91,14 @@ func (r *Repository) AdminCreditCoinSellerStock(ctx context.Context, command led 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 { + if err := r.applyAccountDelta(ctx, tx, account, availableDelta, 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, + AvailableDelta: availableDelta, FrozenDelta: 0, AvailableAfter: balanceAfter, FrozenAfter: account.FrozenAmount, @@ -104,7 +110,7 @@ func (r *Repository) AdminCreditCoinSellerStock(ctx context.Context, command led 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), + balanceChangedEvent(transactionID, command.CommandID, command.SellerUserID, ledger.AssetCoinSellerCoin, availableDelta, 0, balanceAfter, account.FrozenAmount, account.Version+1, metadata, nowMs), coinSellerStockCreditedEvent(transactionID, command.CommandID, metadata, nowMs), }); err != nil { return ledger.CoinSellerStockCreditReceipt{}, err @@ -669,9 +675,25 @@ func coinSellerStockBizType(stockType string) string { switch stockType { case ledger.StockTypeUSDTPurchase: return bizTypeCoinSellerStockPurchase + case ledger.StockTypeUSDTDeduction: + return bizTypeCoinSellerStockDeduction case ledger.StockTypeCoinCompensation: return bizTypeCoinSellerCoinCompensation default: return "" } } + +func coinSellerStockAvailableDelta(command ledger.CoinSellerStockCreditCommand) int64 { + if command.StockType == ledger.StockTypeUSDTDeduction { + return -command.CoinAmount + } + return command.CoinAmount +} + +func coinSellerStockPaidAmountMicro(command ledger.CoinSellerStockCreditCommand) int64 { + if command.StockType == ledger.StockTypeUSDTDeduction { + return -command.PaidAmountMicro + } + return command.PaidAmountMicro +} diff --git a/services/wallet-service/internal/storage/mysql/constants.go b/services/wallet-service/internal/storage/mysql/constants.go index 36cc9153..ca577d0f 100644 --- a/services/wallet-service/internal/storage/mysql/constants.go +++ b/services/wallet-service/internal/storage/mysql/constants.go @@ -13,6 +13,7 @@ const ( bizTypeAgencyOpeningReward = "agency_opening_reward" bizTypeCoinSellerTransfer = "coin_seller_transfer" bizTypeCoinSellerStockPurchase = "coin_seller_stock_purchase" + bizTypeCoinSellerStockDeduction = "coin_seller_stock_deduction" bizTypeCoinSellerCoinCompensation = "coin_seller_coin_compensation" bizTypeSalaryExchangeToCoin = "salary_exchange_to_coin" bizTypeSalaryTransferToCoinSeller = "salary_transfer_to_coin_seller" diff --git a/services/wallet-service/internal/storage/mysql/outbox.go b/services/wallet-service/internal/storage/mysql/outbox.go index f042180b..999b6f9a 100644 --- a/services/wallet-service/internal/storage/mysql/outbox.go +++ b/services/wallet-service/internal/storage/mysql/outbox.go @@ -608,7 +608,7 @@ func agencyOpeningRewardCreditedEvent(transactionID string, commandID string, me func coinSellerStockCreditedEvent(transactionID string, commandID string, metadata coinSellerStockMetadata, nowMs int64) walletOutboxEvent { eventType := "WalletCoinSellerCoinCompensated" - if metadata.StockType == ledger.StockTypeUSDTPurchase { + if metadata.StockType == ledger.StockTypeUSDTPurchase || metadata.StockType == ledger.StockTypeUSDTDeduction { eventType = "WalletCoinSellerStockPurchased" } return walletOutboxEvent{