From ce060acc6ed1647d60d4c0237cd889b16639028a Mon Sep 17 00:00:00 2001 From: zhx Date: Mon, 22 Jun 2026 20:43:39 +0800 Subject: [PATCH] =?UTF-8?q?=E9=80=81=E7=A4=BC=E5=A2=9E=E5=8A=A0=E9=87=91?= =?UTF-8?q?=E5=B8=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/wallet-service模块化拆分方案.md | 387 ++++++++++++++++++ .../internal/service/wallet/service_test.go | 107 +++-- .../internal/storage/mysql/repository.go | 340 +++++++++++++-- .../storage/mysql/resource_repository.go | 2 +- 4 files changed, 783 insertions(+), 53 deletions(-) create mode 100644 docs/wallet-service模块化拆分方案.md diff --git a/docs/wallet-service模块化拆分方案.md b/docs/wallet-service模块化拆分方案.md new file mode 100644 index 00000000..687fefd8 --- /dev/null +++ b/docs/wallet-service模块化拆分方案.md @@ -0,0 +1,387 @@ +# wallet-service 模块化拆分方案 + +## 结论 + +`wallet-service` 需要一起拆三层：`service` 用例层、`storage/mysql` 持久化层、`transport/grpc` 协议适配层。只拆 `repository.go` 不够，因为当前 service 层的 `Repository` 接口已经把礼物扣费、充值、红包、资源、VIP、游戏账变、币商和投影混成一个大端口，文件变小后边界仍然不清晰。 + +推荐路线是两阶段： + +| 阶段 | 目标 | 变更方式 | +| --- | --- | --- | +| 第一阶段 | 低风险整理文件，不改行为 | 保持 `walletservice.Service`、`mysql.Repository` 和 gRPC 注册方式不变，只在同 package 内按业务域拆文件，并把大 `Repository` 接口拆成嵌入式能力接口 | +| 第二阶段 | 真正目录模块化 | 抽出 `ports` 和 `usecase/*` 子包，`wallet.Service` 只保留门面方法，transport 继续依赖门面，避免 gRPC 层直接感知内部用例拆分 | + +不要第一步就把 MySQL repository 拆成多个 Go package。钱包账务依赖同一个 `*sql.Tx` 保证账户、交易、分录、outbox 和幂等结果同事务提交；过早拆 package 会迫使大量事务 helper、metadata 和 outbox 构造函数导出，风险比收益大。 + +## 当前结构和问题 + +当前 wallet-service 的主要文件规模： + +| 路径 | 行数 | 问题 | +| --- | ---: | --- | +| `internal/storage/mysql/repository.go` | 5089 | 混合核心账本、送礼、活动奖励、游戏账变、币商、工资兑换、outbox、账号锁、receipt 和 metadata | +| `internal/storage/mysql/resource_repository.go` | 3987 | 资源目录、资源组、礼物配置、发放、穿戴、商城和资源 outbox 混在一起 | +| `internal/service/wallet/service.go` | 1259 | service 门面、依赖注入、礼物、工资、币商、奖励、游戏、充值、VIP、投影混在一个文件 | +| `internal/service/wallet/service_test.go` | 5625 | 测试文件过大，说明 service 用例边界没有被稳定拆开 | +| `internal/transport/grpc/server.go` | 1389 | gRPC handler 和 proto/domain mapper 混合，资源和红包已经拆出一部分，但主文件仍覆盖过多业务 | +| `internal/app/app.go` | 576 | 进程装配、MQ 生命周期、outbox worker、投影 consumer、三方支付补偿和红包过期 worker 混合 | +| `internal/domain/ledger/ledger.go` | 1407 | 账本领域对象混合所有业务命令和回执 | +| `internal/domain/resource/resource.go` | 729 | 资源目录、礼物配置、用户权益、商城和发放对象混合 | + +当前最核心的问题不是单个文件长，而是变化原因互相交叉： + +| 变化原因 | 当前落点 | 应独立出来的边界 | +| --- | --- | --- | +| 送礼扣费和背包送礼 | `service.go` + `repository.go` + `resource_repository.go` | gift ledger | +| 后台调账、活动奖励、游戏账变 | `service.go` + `repository.go` | ledger operations | +| Google、H5、MiFaPay、V5Pay、USDT | `service.go` + `external_recharge.go` + `third_party_payment_repository.go` | recharge | +| 资源、礼物、穿戴、商城 | `resource.go` + `resource_repository.go` + `transport/grpc/resource.go` | resource | +| VIP 和钱包首页查询 | `service.go` + `app_wallet_repository.go` | app wallet | +| 红包 | `red_packet.go` + `red_packet_repository.go` + `transport/grpc/red_packet.go` | red packet | +| 钱包 outbox 和投影 | `app.go` + `service.go` + `repository.go` + projection files | outbox/projection | +| 主播工资、币商库存和工资换币 | `service.go` + `repository.go` + `host_salary_settlement.go` | host salary / coin seller | + +## 拆分原则 + +1. `wallet-service` 仍然是账务事实 owner。余额、交易、分录和 `wallet_outbox` 不拆到其他服务。 +2. `wallet_outbox` 仍然只由 wallet-service 直接读取和发布。下游服务只能消费 RocketMQ 或自己的投影表，不能回扫钱包库补业务事实。 +3. 第一阶段不改 protobuf、不改 gRPC 方法名、不改 `walletservice.New(repository, activity)` 装配方式。 +4. MySQL 第一阶段保持一个 `mysql.Repository` 类型，方法可以拆文件，但事务 helper 仍留在同 package。 +5. service 第一阶段保持一个 `wallet.Service` 门面，外部调用不变；先拆文件和接口，再考虑 usecase 子包。 +6. 每一步只搬同一业务域的方法、metadata、request hash、receipt 和 outbox event，避免把一个事务链路拆成多次提交。 + +## 第一阶段目标结构 + +### service 层 + +保留 package： + +```text +services/wallet-service/internal/service/wallet/ +``` + +第一阶段目标文件： + +```text +service.go # Service struct、New、依赖注入 setter、通用 guard +repository.go # Repository 总接口和嵌入式能力接口 +clients.go # ActivityBadgeClient、GooglePlayClient、MifaPayClient、V5PayClient、TronUSDTClient +gift.go # DebitGift、DebitRobotGift、BatchDebitGift +balance.go # GetBalances、GetWalletOverview、GetWalletValueSummary、GetUserGiftWall、ListWalletTransactions +host_salary.go # GetActiveHostSalaryPolicy、GetHostSalaryProgress、ProcessHostSalarySettlementBatch +coin_seller.go # AdminCreditCoinSellerStock、TransferCoinFromSeller、ExchangeSalaryToCoin、TransferSalaryToCoinSeller +admin_ledger.go # AdminCreditAsset +reward.go # CreditTaskReward、CreditLuckyGiftReward、CreditWheelReward、CreditRoomTurnoverReward、CreditInviteActivityReward、CreditAgencyOpeningReward +game.go # ApplyGameCoinChange +recharge.go # ListRechargeBills、ListRechargeProducts、ListAdminRechargeProducts、Create/Update/DeleteRechargeProduct、ConfirmGooglePayment +external_recharge.go # H5 三方充值、通知、查单补偿，保留现有文件但只放外部充值 +vip.go # ListVipPackages、GetMyVip、PurchaseVip、GrantVip、ListAdminVipLevels、UpdateAdminVipLevels +red_packet.go # 红包用例，保留现有文件 +resource.go # 资源目录、资源组、礼物配置、发放、穿戴、商城，后续再细拆 +projection.go # ProjectPendingGiftWallEvents、ProcessWalletProjectionMessage、ProjectPendingBadgeGrantEvents +``` + +`repository.go` 建议按 room-service 的方式拆成嵌入式能力接口： + +```go +type Repository interface { + GiftLedgerStore + BalanceStore + HostSalaryStore + CoinSellerStore + AdminLedgerStore + RewardLedgerStore + GameLedgerStore + RechargeStore + ExternalRechargeStore + VIPStore + RedPacketStore + ResourceCatalogStore + ResourceGrantStore + ResourceEquipmentStore + ResourceShopStore + WalletProjectionStore + BadgeProjectionStore +} +``` + +接口边界建议： + +| 接口 | 方法范围 | +| --- | --- | +| `GiftLedgerStore` | `DebitGift`、`BatchDebitGift` | +| `BalanceStore` | `GetBalances`、钱包首页、钱包流水、礼物墙 | +| `HostSalaryStore` | 工资政策、工资进度、工资结算批处理 | +| `CoinSellerStore` | 币商库存、币商转币、工资换币 | +| `AdminLedgerStore` | 后台调账 | +| `RewardLedgerStore` | 任务、幸运礼物、转盘、房间流水、邀请、开业奖励 | +| `GameLedgerStore` | 游戏扣款、返奖、退款、冲正 | +| `RechargeStore` | 充值档位、Google 支付、充值账单 | +| `ExternalRechargeStore` | H5 三方支付方式、下单、回调、查单补偿 | +| `VIPStore` | VIP 包、购买、发放、后台配置 | +| `RedPacketStore` | 红包配置、创建、领取、过期退款、后台重试 | +| `ResourceCatalogStore` | 资源、资源组、礼物配置、礼物类型 | +| `ResourceGrantStore` | 单资源发放、资源组发放、发放记录 | +| `ResourceEquipmentStore` | 用户权益、穿戴、卸下、批量读取穿戴 | +| `ResourceShopStore` | 资源商城档位、购买记录、购买 | +| `WalletProjectionStore` | 礼物墙投影和 wallet outbox message 投影入口 | +| `BadgeProjectionStore` | badge grant 投影 claim、delivered、failed | + +### storage/mysql 层 + +保留 package： + +```text +services/wallet-service/internal/storage/mysql/ +``` + +`repository.go` 第一阶段只保留： + +```text +Repository struct +Open / Close / Ping +通用常量中真正跨业务域使用的部分 +``` + +建议拆出的文件： + +```text +schema.go # ensure*Schema、重复列/重复索引兼容判断 +account.go # walletAccount、lockAccount、lockAccountsOrdered、applyAccountDelta +transaction.go # transactionRow、lookupTransaction、insertTransaction、transactionID、request hash 公共能力 +entry.go # walletEntry、insertEntry、balanceAfterTransaction +outbox.go # WalletOutboxRecord、WalletOutboxClaimFilter、Claim/Mark/insertWalletOutbox、事件类型过滤 +gift_ledger.go # DebitGift、BatchDebitGift、giftDebitBizType、背包送礼消耗 +gift_pricing.go # resolveGiftPrice、resolveGiftDiamondRatio、resolveActiveGiftConfig +gift_receipt.go # giftMetadata、receiptFromGiftMetadata、gift outbox events +host_diamond.go # creditHostPeriodDiamonds、host period diamond account/entry +admin_ledger.go # AdminCreditAsset、adminCreditMetadata、adminCreditRequestHash +reward_ledger.go # 各活动奖励入账、metadata、receipt、request hash、outbox event +game_ledger.go # ApplyGameCoinChange、gameCoinMetadata、gameBizTypeAndDelta +coin_seller_ledger.go # stock、seller transfer、salary exchange、salary transfer、相关 metadata/hash/receipt +``` + +已存在文件的后续整理： + +```text +app_wallet_repository.go # VIP + 钱包首页较大，后续拆 vip_repository.go / wallet_query_repository.go +resource_repository.go # 后续拆 resource_catalog.go / resource_group.go / resource_grant.go / user_resource.go / resource_shop.go / resource_outbox.go +third_party_payment_repository.go # 后续拆 payment_method_repository.go / external_recharge_order_repository.go +red_packet_repository.go # 已独立，但可继续拆 config/create/claim/refund/query +``` + +### transport/grpc 层 + +保留 `Server` 类型和 protobuf 注册方式不变，先按 RPC 业务域拆文件： + +```text +server.go # Server struct、NewServer、通用 helper +gift.go # DebitGift、BatchDebitGift、DebitRobotGift、gift response mapper +balance.go # GetBalances、钱包首页、钱包流水、礼物墙 mapper +host_salary.go # 工资政策、工资进度、cron 结果 mapper +admin_ledger.go # AdminCreditAsset +coin_seller.go # 币商库存、转币、工资换币 +reward.go # 活动奖励 RPC +game.go # ApplyGameCoinChange +recharge.go # 充值账单、Google、充值档位 +external_recharge.go # H5 三方支付 RPC 和 mapper +vip.go # VIP RPC 和 mapper +resource.go # 资源 handler，后续再拆 catalog/grant/shop/equipment +resource_mapper.go # resource proto/domain mapper +red_packet.go # 红包 handler +red_packet_mapper.go # 红包 proto/domain mapper +cron_server.go # cron gRPC server，保留独立 +``` + +`transport/grpc` 的拆分原则是：handler 和 mapper 可以同业务域放一起，但一个文件超过 500 行时应把 mapper 单独拆出。transport 只能做 proto/domain 转换和错误透传，不要在这里补账务规则。 + +### app 装配层 + +`internal/app/app.go` 可以低风险拆成： + +```text +app.go # App struct、New、Run、Close +clients.go # Activity、Google、MiFaPay、V5Pay、TronGrid 客户端装配 +mq.go # RocketMQ producer/consumer 创建、start/shutdown +outbox_worker.go # wallet outbox claim/publish/mark +projection_worker.go # wallet projection consumer +external_recharge_worker.go # 外部充值查单补偿 worker +red_packet_worker.go # 红包过期退款 worker +health.go # health HTTP run/close +``` + +app 层不承载业务规则。worker 只负责任务循环和调用 service，用例幂等必须仍在 service/storage 内保证。 + +### domain 层 + +第一阶段不拆 package，只拆文件： + +```text +internal/domain/ledger/ + ledger.go # 通用常量、AssetBalance、基础类型 + gift.go # 送礼 command/receipt + host_salary.go # 工资 policy/progress/settlement + coin_seller.go # 币商库存、转币、工资换币 + recharge.go # 充值档位、Google、充值账单 + external_recharge.go # H5 三方订单、通知、支付方式 + wallet_query.go # 钱包首页、流水、礼物墙 + vip.go # VIP + reward.go # 活动奖励 + game.go # 游戏账变 + red_packet.go # 红包 + +internal/domain/resource/ + resource.go # 通用 Resource + gift_config.go # GiftConfig、GiftTypeConfig + group.go # ResourceGroup + grant.go # GrantResource、ResourceGrant + entitlement.go # UserResourceEntitlement、穿戴 + shop.go # ResourceShopItem、PurchaseOrder +``` + +领域对象拆文件的目标是让业务域可读，不要把领域对象拆成过细 package。否则 transport、service、storage 都要同时改 import，收益不高。 + +## 第二阶段目标目录 + +第一阶段稳定后，再考虑 service 真正目录化。目标是 `wallet.Service` 作为门面，内部 usecase 子包只依赖 ports 和 domain： + +```text +services/wallet-service/internal/service/wallet/ + service.go # 门面，保留现有对外方法 + ports/ + repository.go # repository 能力接口 + clients.go # 三方 client 能力接口 + clock.go # now/clock 抽象，如需要 + usecase/ + gift/ + balance/ + hostsalary/ + coinseller/ + adminledger/ + reward/ + game/ + recharge/ + externalrecharge/ + vip/ + redpacket/ + resource/ + projection/ +``` + +依赖方向： + +```text +transport/grpc -> service/wallet facade -> service/wallet/usecase/* -> service/wallet/ports +storage/mysql -> implements service/wallet/ports.Repository +client/* -> implements service/wallet/ports third-party clients +domain/* -> shared pure data and validation helpers +``` + +第二阶段不要让 `transport/grpc` 直接调用 `usecase/*`。gRPC 层直接感知所有 usecase 会把装配复杂度扩散到 transport，后续新增 RPC 时更容易乱。 + +## 推荐实施顺序 + +### 1. service 低风险拆文件 + +先拆 `internal/service/wallet/service.go`，不改任何方法签名： + +1. 新建 `repository.go`，移动 `Repository interface`，并改成嵌入式能力接口。 +2. 新建 `clients.go`，移动 Activity、Google、MiFaPay、V5Pay、TronUSDT client 接口。 +3. 按业务域移动 Service 方法到 `gift.go`、`host_salary.go`、`coin_seller.go`、`reward.go`、`game.go`、`recharge.go`、`vip.go`、`balance.go`、`projection.go`。 +4. 每移动一个业务域，运行 `go test ./services/wallet-service/internal/service/wallet`。 + +### 2. transport 跟随拆分 + +`server.go` 只保留 `Server`、`NewServer` 和通用 helper。按 service 文件域拆 handler 文件，mapper 超过 500 行时单独放 `*_mapper.go`。 + +验证： + +```bash +go test ./services/wallet-service/internal/transport/grpc +``` + +### 3. storage/mysql 拆核心 repository + +先拆基础设施，再拆业务： + +1. `schema.go` +2. `account.go` +3. `transaction.go` +4. `entry.go` +5. `outbox.go` +6. `gift_ledger.go`、`gift_pricing.go`、`gift_receipt.go`、`host_diamond.go` +7. `admin_ledger.go`、`reward_ledger.go`、`game_ledger.go`、`coin_seller_ledger.go` + +每一步只搬代码，不改 SQL，不改事务边界，不改 outbox payload。 + +验证： + +```bash +go test ./services/wallet-service/internal/storage/mysql +go test ./services/wallet-service/... +``` + +### 4. 拆 resource_repository.go + +`resource_repository.go` 单独作为第二个大 storage 任务处理，避免和核心账本拆分混在一起： + +```text +resource_catalog.go +resource_group.go +resource_grant.go +user_resource.go +resource_equipment.go +resource_shop.go +resource_outbox.go +``` + +这一步要特别保护背包送礼链路，因为 `gift_ledger.go` 会调用用户权益消耗相关 helper。 + +### 5. app worker 拆分 + +最后拆 `internal/app/app.go`。原因是 app 层只是装配和 worker 循环，等 service/storage 边界清晰后再拆更不容易来回调整。 + +验证： + +```bash +go test ./services/wallet-service/... +``` + +## 每阶段完成标准 + +| 阶段 | 完成标准 | +| --- | --- | +| service 拆分 | `service.go` 只保留 Service 结构、构造和依赖 setter；`Repository` 变成嵌入式能力接口；service 单测通过 | +| transport 拆分 | `server.go` 不再承载大批业务 handler；mapper 按业务域归位；transport 单测通过 | +| storage 核心拆分 | `repository.go` 小于 250 行；核心账本 helper 有明确文件；wallet storage 单测通过 | +| resource storage 拆分 | `resource_repository.go` 小于 500 行；资源目录、发放、穿戴、商城各自可定位；资源 storage 单测通过 | +| app 拆分 | `app.go` 只负责生命周期；worker 文件按 MQ/outbox/projection/reconcile/expiry 分开；wallet-service 全包测试通过 | + +最终验证： + +```bash +go test ./services/wallet-service/... +make test +``` + +如果 `make test` 暴露其他服务或既有配置失败，需要单独记录为基线问题，不能把它当成钱包拆分失败，也不能因为无关失败回滚已验证的 wallet-service 拆分。 + +## 风险点 + +| 风险 | 规避方式 | +| --- | --- | +| 事务 helper 被跨 package 导出 | 第一阶段 storage 保持同 package，只拆文件 | +| request hash 或 biz type 和原逻辑不一致 | 每个业务域移动时把 command、metadata、request hash、receipt、outbox event 一起移动 | +| service 接口拆完但测试 mock 更复杂 | 总 `Repository` 保持嵌入式组合，测试按用例只依赖小接口 | +| transport 拆分后 mapper 找不到 | handler 和 mapper 同域移动，通用 mapper 留在 `server.go` 或 `mapper.go` | +| app worker 拆早导致来回改 | app 层最后拆，等 service 方法和 storage outbox 边界稳定 | +| 背包送礼跨 resource 和 gift ledger | `consumeGiftEntitlementForSend` 先留在 gift ledger 同 package；第二阶段再决定是否抽资源权益端口 | + +## 不建议做的事 + +1. 不要第一步创建多个 MySQL repository 实例。账务事务必须共享同一个连接池和同一个事务提交边界。 +2. 不要让 `room-service`、`activity-service` 或 `notice-service` 直连钱包库查询交易明细。 +3. 不要把 outbox 发布状态交给下游服务更新。wallet outbox 位点仍归 wallet-service。 +4. 不要在 transport 层补业务幂等、扣费判断或支付状态机。 +5. 不要为了拆目录改变 protobuf 字段或 gRPC 方法名；这是结构整理，不是协议重构。 + diff --git a/services/wallet-service/internal/service/wallet/service_test.go b/services/wallet-service/internal/service/wallet/service_test.go index c12c50f0..9833bfec 100644 --- a/services/wallet-service/internal/service/wallet/service_test.go +++ b/services/wallet-service/internal/service/wallet/service_test.go @@ -35,7 +35,7 @@ func walletOutboxMessageFromRecord(record mysqlstorage.WalletOutboxRecord) walle } } -// TestDebitGiftUsesServerPriceWithoutGiftPointCredit 验证送礼只信服务端价格，且礼物积分下线后不再给收礼人入账。 +// TestDebitGiftUsesServerPriceWithoutGiftPointCredit 验证送礼只信服务端价格，收礼收入进入 COIN，历史 GIFT_POINT 仍保持下线。 func TestDebitGiftUsesServerPriceWithoutGiftPointCredit(t *testing.T) { repository := mysqltest.NewRepository(t) repository.SetBalance(10001, 100) @@ -76,11 +76,21 @@ func TestDebitGiftUsesServerPriceWithoutGiftPointCredit(t *testing.T) { if balanceAmount(balances, ledger.AssetGiftPoint) != 0 { t.Fatalf("target GIFT_POINT must stay zero after point removal: %+v", balances) } - if got := repository.CountRows("wallet_entries", "transaction_id = ?", receipt.TransactionID); got != 1 { - t.Fatalf("gift debit should only write sender COIN entry, got %d", got) + balances, err = svc.GetBalances(context.Background(), 10002, []string{ledger.AssetCoin}) + if err != nil { + t.Fatalf("GetBalances target COIN failed: %v", err) } - if got := repository.CountRows("wallet_outbox", "transaction_id = ?", receipt.TransactionID); got != 2 { - t.Fatalf("gift debit should write sender balance and gift outbox events, got %d", got) + if balanceAmount(balances, ledger.AssetCoin) != 4 { + t.Fatalf("target COIN income mismatch: %+v", balances) + } + if got := repository.CountRows("wallet_entries", "transaction_id = ?", receipt.TransactionID); got != 2 { + t.Fatalf("gift debit should write sender debit and target income entries, got %d", got) + } + if got := repository.CountRows("wallet_outbox", "transaction_id = ?", receipt.TransactionID); got != 4 { + t.Fatalf("gift debit should write sender balance, target balance, debit and income outbox events, got %d", got) + } + if got := repository.CountRows("wallet_outbox", "transaction_id = ? AND event_type = ?", receipt.TransactionID, "WalletGiftIncomeCredited"); got != 1 { + t.Fatalf("gift debit should publish target income fact, got %d", got) } if receipt.HostPeriodDiamondAdded != 0 || receipt.HostPeriodCycleKey != "" { t.Fatalf("non-host gift must not credit host period diamonds: %+v", receipt) @@ -125,6 +135,13 @@ func TestBatchDebitGiftSettlesAllTargetsAtomically(t *testing.T) { if balanceAmount(balances, ledger.AssetGiftPoint) != 0 { t.Fatalf("target %d GIFT_POINT must stay zero after point removal: %+v", userID, balances) } + balances, err = svc.GetBalances(context.Background(), userID, []string{ledger.AssetCoin}) + if err != nil { + t.Fatalf("GetBalances target COIN %d failed: %v", userID, err) + } + if balanceAmount(balances, ledger.AssetCoin) != 4 { + t.Fatalf("target %d COIN income mismatch: %+v", userID, balances) + } } balances, err := svc.GetBalances(context.Background(), 10001, []string{ledger.AssetCoin}) if err != nil { @@ -136,8 +153,8 @@ func TestBatchDebitGiftSettlesAllTargetsAtomically(t *testing.T) { if got := repository.CountRows("wallet_transactions", "command_id IN (?, ?)", "cmd-gift-batch:target:10002", "cmd-gift-batch:target:10003"); got != 2 { t.Fatalf("batch gift should write one transaction per target, got %d", got) } - if got := repository.CountRows("wallet_entries", "transaction_id IN (?, ?)", receipt.Targets[0].Receipt.TransactionID, receipt.Targets[1].Receipt.TransactionID); got != 2 { - t.Fatalf("batch gift should only write sender COIN entries per target, got %d", got) + if got := repository.CountRows("wallet_entries", "transaction_id IN (?, ?)", receipt.Targets[0].Receipt.TransactionID, receipt.Targets[1].Receipt.TransactionID); got != 4 { + t.Fatalf("batch gift should write sender debit and target income entries per target, got %d", got) } again, err := svc.BatchDebitGift(context.Background(), ledger.BatchDebitGiftCommand{ @@ -159,6 +176,15 @@ func TestBatchDebitGiftSettlesAllTargetsAtomically(t *testing.T) { if again.Aggregate.CoinSpent != receipt.Aggregate.CoinSpent || again.Targets[0].Receipt.TransactionID != receipt.Targets[0].Receipt.TransactionID || repository.CountRows("wallet_transactions", "command_id IN (?, ?)", "cmd-gift-batch:target:10002", "cmd-gift-batch:target:10003") != 2 { t.Fatalf("batch idempotency mismatch: first=%+v again=%+v", receipt, again) } + for _, userID := range []int64{10002, 10003} { + balances, err := svc.GetBalances(context.Background(), userID, []string{ledger.AssetCoin}) + if err != nil { + t.Fatalf("GetBalances target COIN %d after replay failed: %v", userID, err) + } + if balanceAmount(balances, ledger.AssetCoin) != 4 { + t.Fatalf("batch replay must not duplicate target income for %d: %+v", userID, balances) + } + } } // TestDebitGiftCreditsHostPeriodDiamondsOnlyForHost 验证只有 gateway 确认的 active 主播会增加工资周期钻石。 @@ -273,17 +299,19 @@ func TestDebitGiftAppliesGlobalDefaultGiftDiamondRatioToHeatValueByGiftType(t *t repository.SetGiftDiamondRatio(8801, resourcedomain.GiftTypeSuperLucky, "50.00") cases := []struct { - giftID string - giftType string - command string - senderID int64 - coinPrice int64 - heatUnit int64 - wantHeat int64 + giftID string + giftType string + command string + senderID int64 + targetID int64 + coinPrice int64 + heatUnit int64 + wantHeat int64 + wantIncome int64 }{ - {giftID: "normal-room-ratio-gift", giftType: resourcedomain.GiftTypeNormal, command: "cmd-normal-room-ratio", senderID: 13001, coinPrice: 100, heatUnit: 999, wantHeat: 100}, - {giftID: "lucky-room-ratio-gift", giftType: resourcedomain.GiftTypeLucky, command: "cmd-lucky-room-ratio", senderID: 13002, coinPrice: 100, heatUnit: 999, wantHeat: 10}, - {giftID: "super-lucky-room-ratio-gift", giftType: resourcedomain.GiftTypeSuperLucky, command: "cmd-super-lucky-room-ratio", senderID: 13003, coinPrice: 200, heatUnit: 500, wantHeat: 2}, + {giftID: "normal-room-ratio-gift", giftType: resourcedomain.GiftTypeNormal, command: "cmd-normal-room-ratio", senderID: 13001, targetID: 14001, coinPrice: 100, heatUnit: 999, wantHeat: 100, wantIncome: 30}, + {giftID: "lucky-room-ratio-gift", giftType: resourcedomain.GiftTypeLucky, command: "cmd-lucky-room-ratio", senderID: 13002, targetID: 14002, coinPrice: 100, heatUnit: 999, wantHeat: 10, wantIncome: 10}, + {giftID: "super-lucky-room-ratio-gift", giftType: resourcedomain.GiftTypeSuperLucky, command: "cmd-super-lucky-room-ratio", senderID: 13003, targetID: 14003, coinPrice: 200, heatUnit: 500, wantHeat: 2, wantIncome: 2}, } for _, tc := range cases { @@ -296,7 +324,7 @@ func TestDebitGiftAppliesGlobalDefaultGiftDiamondRatioToHeatValueByGiftType(t *t SenderUserID: tc.senderID, SenderRegionID: 1001, RegionID: 8801, - TargetUserID: 14001, + TargetUserID: tc.targetID, GiftID: tc.giftID, GiftCount: 1, PriceVersion: "v1", @@ -307,6 +335,13 @@ func TestDebitGiftAppliesGlobalDefaultGiftDiamondRatioToHeatValueByGiftType(t *t if receipt.HeatValue != tc.wantHeat || receipt.GiftPointAdded != 0 || receipt.CoinSpent != tc.coinPrice { t.Fatalf("%s room contribution ratio mismatch: %+v want heat %d", tc.giftType, receipt, tc.wantHeat) } + balances, err := svc.GetBalances(context.Background(), tc.targetID, []string{ledger.AssetCoin}) + if err != nil { + t.Fatalf("GetBalances target income %s failed: %v", tc.giftType, err) + } + if balanceAmount(balances, ledger.AssetCoin) != tc.wantIncome { + t.Fatalf("%s target income mismatch: got %+v want %d", tc.giftType, balances, tc.wantIncome) + } } } @@ -1095,7 +1130,7 @@ func TestHostSalaryHalfMonthSettlementUsesPolicyMode(t *testing.T) { } } -// TestDebitGiftAllowsSelfGift 验证用户可以给自己送礼；礼物积分下线后只扣 COIN，不再回加 GIFT_POINT。 +// TestDebitGiftAllowsSelfGift 验证用户可以给自己送礼；同一 COIN 账户保留扣费和收礼收入两条分录，但余额事件只发净变更。 func TestDebitGiftAllowsSelfGift(t *testing.T) { repository := mysqltest.NewRepository(t) repository.SetBalance(10001, 100) @@ -1115,18 +1150,27 @@ func TestDebitGiftAllowsSelfGift(t *testing.T) { t.Fatalf("self DebitGift failed: %v", err) } - if receipt.CoinSpent != 14 || receipt.GiftPointAdded != 0 || receipt.BalanceAfter != 86 { + if receipt.CoinSpent != 14 || receipt.GiftPointAdded != 0 || receipt.BalanceAfter != 90 { t.Fatalf("self gift settlement mismatch: %+v", receipt) } balances, err := svc.GetBalances(context.Background(), 10001, []string{ledger.AssetCoin, ledger.AssetGiftPoint}) if err != nil { t.Fatalf("GetBalances failed: %v", err) } - if balanceAmount(balances, ledger.AssetCoin) != 86 || balanceAmount(balances, ledger.AssetGiftPoint) != 0 { + if balanceAmount(balances, ledger.AssetCoin) != 90 || balanceAmount(balances, ledger.AssetGiftPoint) != 0 { t.Fatalf("self gift balances mismatch: %+v", balances) } - if got := repository.CountRows("wallet_entries", "transaction_id = ?", receipt.TransactionID); got != 1 { - t.Fatalf("self gift should only write debit entry, got %d", got) + if got := repository.CountRows("wallet_entries", "transaction_id = ?", receipt.TransactionID); got != 2 { + t.Fatalf("self gift should write debit and income entries, got %d", got) + } + if got := repository.CountRows("wallet_outbox", "transaction_id = ? AND event_type = ?", receipt.TransactionID, "WalletBalanceChanged"); got != 1 { + t.Fatalf("self gift should publish one net balance event, got %d", got) + } + if got := repository.CountRows("wallet_outbox", "transaction_id = ? AND event_type = ? AND available_delta = ?", receipt.TransactionID, "WalletBalanceChanged", int64(-10)); got != 1 { + t.Fatalf("self gift balance event should carry net -10 delta, got %d", got) + } + if got := repository.CountRows("wallet_outbox", "transaction_id = ? AND event_type = ?", receipt.TransactionID, "WalletGiftIncomeCredited"); got != 1 { + t.Fatalf("self gift should still publish income fact, got %d", got) } } @@ -1305,6 +1349,9 @@ func TestDebitRobotGiftCreatesRobotCoinAccountAndAutoTopsUp(t *testing.T) { if got := repository.CountRows("wallet_accounts", "user_id = ? AND asset_type = ? AND available_amount = 0", int64(20001), ledger.AssetRobotCoin); got != 1 { t.Fatalf("robot gift should create empty ROBOT_COIN account after debit, got %d", got) } + if got := repository.CountRows("wallet_accounts", "user_id = ? AND asset_type = ?", int64(20002), ledger.AssetCoin); got != 0 { + t.Fatalf("robot gift must not create target COIN income account, got %d", got) + } if got := repository.CountRows("wallet_entries", "transaction_id = ? AND user_id = ? AND asset_type = ?", receipt.TransactionID, int64(20001), ledger.AssetRobotCoin); got != 2 { t.Fatalf("robot gift should write top-up and debit entries, got %d", got) } @@ -3078,12 +3125,22 @@ func TestDebitGiftFromBagConsumesEntitlementWithoutCoinDebit(t *testing.T) { if quantity != 3 || remaining != 1 { t.Fatalf("bag gift entitlement quantity mismatch: quantity=%d remaining=%d", quantity, remaining) } - if got := repository.CountRows("wallet_entries", "transaction_id = ?", receipt.TransactionID); got != 0 { - t.Fatalf("bag gift must not write COIN wallet entries, got %d", got) + balances, err := svc.GetBalances(ctx, 61002, []string{ledger.AssetCoin}) + if err != nil { + t.Fatalf("GetBalances bag gift target failed: %v", err) + } + if balanceAmount(balances, ledger.AssetCoin) != 3 { + t.Fatalf("bag gift target income mismatch: %+v", balances) + } + if got := repository.CountRows("wallet_entries", "transaction_id = ?", receipt.TransactionID); got != 1 { + t.Fatalf("bag gift must only write target COIN income entry, got %d", got) } if got := repository.CountRows("wallet_outbox", "transaction_id = ? AND event_type = ?", receipt.TransactionID, "WalletGiftDebited"); got != 1 { t.Fatalf("bag gift must still publish WalletGiftDebited for normal room projections, got %d", got) } + if got := repository.CountRows("wallet_outbox", "transaction_id = ? AND event_type = ?", receipt.TransactionID, "WalletGiftIncomeCredited"); got != 1 { + t.Fatalf("bag gift must publish target income fact, got %d", got) + } if got := repository.CountRows("wallet_outbox", "command_id = ? AND event_type = ?", "cmd-send-bag-spark", "UserResourceChanged"); got != 1 { t.Fatalf("bag gift must publish resource change event, got %d", got) } diff --git a/services/wallet-service/internal/storage/mysql/repository.go b/services/wallet-service/internal/storage/mysql/repository.go index a91c680c..8405e7ce 100644 --- a/services/wallet-service/internal/storage/mysql/repository.go +++ b/services/wallet-service/internal/storage/mysql/repository.go @@ -56,6 +56,8 @@ const ( giftWallChargeAssetMixed = "MIXED" giftChargeSourceCoin = "coin" giftChargeSourceBag = "bag" + normalGiftIncomeRatioPercent = "30.00" + normalGiftIncomeRatioPPM = 300_000 ) // Repository 是 wallet-service 的 MySQL v2 账本入口。 @@ -447,6 +449,15 @@ func (r *Repository) DebitGift(ctx context.Context, command ledger.DebitGiftComm if err != nil { return ledger.Receipt{}, err } + giftIncomeRatio, giftIncomeRatioRegionID := giftIncomeRatio(giftConfig.GiftTypeCode, roomContributionRatio, roomContributionRatioRegionID) + giftIncomeCoinAmount := int64(0) + if !command.RobotGift { + // 收礼金币收入只属于真人送礼经济账；背包礼物也按礼物服务端价值发放，机器人礼物继续只服务房间展示。 + giftIncomeCoinAmount, err = giftDiamondAmount(chargeAmount, giftIncomeRatio.PPM) + if err != nil { + return ledger.Receipt{}, err + } + } hostPeriodDiamondAdded := int64(0) hostPeriodCycleKey := "" giftDiamondRatioPercent := "0.00" @@ -466,16 +477,42 @@ func (r *Repository) DebitGift(ctx context.Context, command ledger.DebitGiftComm hostPeriodCycleKey = hostPeriodCycleKeyFromMS(nowMs) } - // 机器人礼物扣的是展示专用 ROBOT_COIN；本地或新机器人账号第一次送礼时可能还没有 - // 该资产账户，必须先创建空账户，再由下方 robotCoinTopUp 按本次价格补足并立即扣减。 accountAssetType := chargeAssetType if chargeSource == giftChargeSourceBag { accountAssetType = ledger.AssetCoin } - sender, err := r.lockAccount(ctx, tx, command.SenderUserID, accountAssetType, command.RobotGift || chargeAmount == 0 || chargeSource == giftChargeSourceBag, nowMs) - if err != nil { - return ledger.Receipt{}, err + var senderState *walletAccountState + var targetCoinState *walletAccountState + if command.RobotGift { + // 机器人礼物扣的是展示专用 ROBOT_COIN；它不产生收礼 COIN 收入，因此继续沿用单账户锁定路径。 + sender, err := r.lockAccount(ctx, tx, command.SenderUserID, accountAssetType, true, nowMs) + if err != nil { + return ledger.Receipt{}, err + } + senderState = &walletAccountState{account: sender} + } else { + lockRequests := []walletAccountLockRequest{{ + UserID: command.SenderUserID, + AssetType: accountAssetType, + CreateIfMissing: chargeAmount == 0 || chargeSource == giftChargeSourceBag, + }} + if giftIncomeCoinAmount > 0 { + lockRequests = append(lockRequests, walletAccountLockRequest{ + UserID: command.TargetUserID, + AssetType: ledger.AssetCoin, + CreateIfMissing: true, + }) + } + states, err := r.lockAccountsOrdered(ctx, tx, lockRequests, nowMs) + if err != nil { + return ledger.Receipt{}, err + } + senderState = states[walletAccountLockKey(command.SenderUserID, accountAssetType)] + if giftIncomeCoinAmount > 0 { + targetCoinState = states[walletAccountLockKey(command.TargetUserID, ledger.AssetCoin)] + } } + sender := senderState.account robotCoinTopUp := int64(0) if command.RobotGift && sender.AvailableAmount < chargeAmount { // 机器人房间使用专用 ROBOT_COIN 做展示账务，按次补足后立刻扣减，不要求真实充值预算。 @@ -489,6 +526,18 @@ func (r *Repository) DebitGift(ctx context.Context, command ledger.DebitGiftComm // Bag 礼物不扣 COIN；balance_after 只给客户端刷新余额使用，必须保持当前 COIN 余额。 senderAfter = sender.AvailableAmount } + giftIncomeBalanceAfter := int64(0) + if giftIncomeCoinAmount > 0 { + if targetCoinState == nil { + return ledger.Receipt{}, xerr.New(xerr.Internal, "gift income account is missing") + } + giftIncomeBalanceAfter = targetCoinState.account.AvailableAmount + giftIncomeCoinAmount + if command.TargetUserID == command.SenderUserID && accountAssetType == ledger.AssetCoin && chargeSource != giftChargeSourceBag { + // 自己送自己时同一个 COIN 账户先扣礼物价值、再加收礼收入；回执余额必须反映最终净变更。 + giftIncomeBalanceAfter = senderAfter + giftIncomeCoinAmount + senderAfter = giftIncomeBalanceAfter + } + } transactionID := transactionID(command.AppCode, command.CommandID) metadata := giftMetadata{ AppCode: command.AppCode, @@ -514,6 +563,10 @@ func (r *Repository) DebitGift(ctx context.Context, command ledger.DebitGiftComm GiftPointAdded: 0, HeatValue: heatValue, BalanceAfter: senderAfter, + GiftIncomeCoinAmount: giftIncomeCoinAmount, + GiftIncomeRatioPercent: giftIncomeRatio.Percent, + GiftIncomeRatioRegionID: giftIncomeRatioRegionID, + GiftIncomeBalanceAfter: giftIncomeBalanceAfter, BillingReceipt: billingReceiptID(command.AppCode, command.CommandID), SenderUserID: command.SenderUserID, SenderRegionID: command.SenderRegionID, @@ -537,20 +590,20 @@ func (r *Repository) DebitGift(ctx context.Context, command ledger.DebitGiftComm if err := r.insertTransaction(ctx, tx, transactionID, command.CommandID, bizType, requestHash, fmt.Sprintf("%s:%s", command.GiftID, price.PriceVersion), metadata, nowMs); err != nil { return ledger.Receipt{}, err } - events := make([]walletOutboxEvent, 0, 3) + events := make([]walletOutboxEvent, 0, 5) if robotCoinTopUp > 0 { - if err := r.applyAccountDelta(ctx, tx, sender, robotCoinTopUp, 0, nowMs); err != nil { + topUpAccount, err := r.applyTrackedAccountDelta(ctx, tx, senderState, robotCoinTopUp, 0, nowMs) + if err != nil { return ledger.Receipt{}, err } - topUpAfter := sender.AvailableAmount + robotCoinTopUp if err := r.insertEntry(ctx, tx, walletEntry{ TransactionID: transactionID, UserID: command.SenderUserID, AssetType: chargeAssetType, AvailableDelta: robotCoinTopUp, FrozenDelta: 0, - AvailableAfter: topUpAfter, - FrozenAfter: sender.FrozenAmount, + AvailableAfter: topUpAccount.AvailableAmount, + FrozenAfter: topUpAccount.FrozenAmount, CounterpartyUserID: 0, RoomID: command.RoomID, CreatedAtMS: nowMs, @@ -559,8 +612,7 @@ func (r *Repository) DebitGift(ctx context.Context, command ledger.DebitGiftComm } // 自动补足只作为机器人展示账本的内部分录；对外余额事件只发布最终扣减后的余额， // 避免同一 transaction/user/asset 生成两个 WalletBalanceChanged 事件 ID。 - sender.AvailableAmount = topUpAfter - sender.Version++ + sender = senderState.account } if chargeSource == giftChargeSourceBag { resourceEvent, err := r.consumeGiftEntitlementForSend(ctx, tx, command, giftConfig.ResourceID, int64(command.GiftCount), metadata, nowMs) @@ -569,7 +621,8 @@ func (r *Repository) DebitGift(ctx context.Context, command ledger.DebitGiftComm } events = append(events, resourceEvent) } else { - if err := r.applyAccountDelta(ctx, tx, sender, -chargeAmount, 0, nowMs); err != nil { + debitAccount, err := r.applyTrackedAccountDelta(ctx, tx, senderState, -chargeAmount, 0, nowMs) + if err != nil { return ledger.Receipt{}, err } if err := r.insertEntry(ctx, tx, walletEntry{ @@ -578,8 +631,8 @@ func (r *Repository) DebitGift(ctx context.Context, command ledger.DebitGiftComm AssetType: chargeAssetType, AvailableDelta: -chargeAmount, FrozenDelta: 0, - AvailableAfter: senderAfter, - FrozenAfter: sender.FrozenAmount, + AvailableAfter: debitAccount.AvailableAmount, + FrozenAfter: debitAccount.FrozenAmount, CounterpartyUserID: command.TargetUserID, RoomID: command.RoomID, CreatedAtMS: nowMs, @@ -587,6 +640,35 @@ func (r *Repository) DebitGift(ctx context.Context, command ledger.DebitGiftComm return ledger.Receipt{}, err } } + if giftIncomeCoinAmount > 0 { + incomeState := targetCoinState + if command.TargetUserID == command.SenderUserID && accountAssetType == ledger.AssetCoin { + // 自送同一个 COIN 账户时，收入必须基于扣费后的版本继续加账，避免用锁定初始版本写出版本冲突。 + incomeState = senderState + } + incomeAccount, err := r.applyTrackedAccountDelta(ctx, tx, incomeState, giftIncomeCoinAmount, 0, nowMs) + if err != nil { + return ledger.Receipt{}, err + } + metadata.GiftIncomeBalanceAfter = incomeAccount.AvailableAmount + if command.TargetUserID == command.SenderUserID && accountAssetType == ledger.AssetCoin { + metadata.BalanceAfter = incomeAccount.AvailableAmount + } + if err := r.insertEntry(ctx, tx, walletEntry{ + TransactionID: transactionID, + UserID: command.TargetUserID, + AssetType: ledger.AssetCoin, + AvailableDelta: giftIncomeCoinAmount, + FrozenDelta: 0, + AvailableAfter: incomeAccount.AvailableAmount, + FrozenAfter: incomeAccount.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) @@ -600,9 +682,19 @@ func (r *Repository) DebitGift(ctx context.Context, command ledger.DebitGiftComm if !command.RobotGift { // 机器人礼物只服务房间展示，不需要客户端余额私信、礼物墙投影或真实消费类下游事件。 if chargeSource != giftChargeSourceBag { - events = append(events, balanceChangedEvent(transactionID, command.CommandID, command.SenderUserID, chargeAssetType, -chargeAmount, 0, senderAfter, sender.FrozenAmount, sender.Version+1, metadata, nowMs)) + if command.TargetUserID == command.SenderUserID && chargeAssetType == ledger.AssetCoin { + events = append(events, balanceChangedEvent(transactionID, command.CommandID, command.SenderUserID, ledger.AssetCoin, giftIncomeCoinAmount-chargeAmount, 0, metadata.BalanceAfter, senderState.account.FrozenAmount, senderState.account.Version, metadata, nowMs)) + } else { + events = append(events, balanceChangedEvent(transactionID, command.CommandID, command.SenderUserID, chargeAssetType, -chargeAmount, 0, senderState.account.AvailableAmount, senderState.account.FrozenAmount, senderState.account.Version, metadata, nowMs)) + } + } + if giftIncomeCoinAmount > 0 && !(command.TargetUserID == command.SenderUserID && chargeAssetType == ledger.AssetCoin && chargeSource != giftChargeSourceBag) { + events = append(events, balanceChangedEvent(transactionID, command.CommandID, command.TargetUserID, ledger.AssetCoin, giftIncomeCoinAmount, 0, metadata.GiftIncomeBalanceAfter, targetCoinState.account.FrozenAmount, targetCoinState.account.Version, metadata, nowMs)) } events = append(events, giftDebitedEvent(transactionID, command.CommandID, command.SenderUserID, chargeAssetType, -chargeAmount, 0, metadata, nowMs)) + if giftIncomeCoinAmount > 0 { + events = append(events, giftIncomeCreditedEvent(transactionID, command.CommandID, metadata, nowMs)) + } } if hostPeriodDiamondAdded > 0 { events = append(events, hostPeriodDiamondCreditedEvent(transactionID, command.CommandID, metadata, nowMs)) @@ -668,6 +760,11 @@ func (r *Repository) BatchDebitGift(ctx context.Context, command ledger.BatchDeb if err != nil { return ledger.BatchGiftReceipt{}, err } + giftIncomeRatio, giftIncomeRatioRegionID := giftIncomeRatio(giftConfig.GiftTypeCode, roomContributionRatio, roomContributionRatioRegionID) + giftIncomeCoinAmount, err := giftDiamondAmount(chargeAmount, giftIncomeRatio.PPM) + if err != nil { + return ledger.BatchGiftReceipt{}, err + } totalChargeAmount, err := checkedMul(chargeAmount, int64(len(command.Targets))) if err != nil { return ledger.BatchGiftReceipt{}, err @@ -715,14 +812,30 @@ func (r *Repository) BatchDebitGift(ctx context.Context, command ledger.BatchDeb if chargeSource == giftChargeSourceBag { accountAssetType = ledger.AssetCoin } - sender, err := r.lockAccount(ctx, tx, command.SenderUserID, accountAssetType, totalChargeAmount == 0 || chargeSource == giftChargeSourceBag, nowMs) + lockRequests := []walletAccountLockRequest{{ + UserID: command.SenderUserID, + AssetType: accountAssetType, + CreateIfMissing: totalChargeAmount == 0 || chargeSource == giftChargeSourceBag, + }} + if giftIncomeCoinAmount > 0 { + for _, target := range command.Targets { + lockRequests = append(lockRequests, walletAccountLockRequest{ + UserID: target.TargetUserID, + AssetType: ledger.AssetCoin, + CreateIfMissing: true, + }) + } + } + accountStates, err := r.lockAccountsOrdered(ctx, tx, lockRequests, nowMs) if err != nil { return ledger.BatchGiftReceipt{}, err } + senderState := accountStates[walletAccountLockKey(command.SenderUserID, accountAssetType)] + sender := senderState.account if chargeSource != giftChargeSourceBag && sender.AvailableAmount < totalChargeAmount { return ledger.BatchGiftReceipt{}, xerr.New(xerr.InsufficientBalance, "insufficient balance") } - events := make([]walletOutboxEvent, 0, len(command.Targets)*3+1) + events := make([]walletOutboxEvent, 0, len(command.Targets)*5+1) if chargeSource == giftChargeSourceBag { resourceEvent, err := r.consumeGiftEntitlementForSend(ctx, tx, debitGiftCommandFromBatchCommand(command), giftConfig.ResourceID, int64(command.GiftCount)*int64(len(command.Targets)), giftMetadata{ AppCode: command.AppCode, @@ -776,10 +889,23 @@ func (r *Repository) BatchDebitGift(ctx context.Context, command ledger.BatchDeb } transactionID := transactionID(command.AppCode, target.CommandID) - senderAfter := sender.AvailableAmount - chargeAmount + senderAfter := senderState.account.AvailableAmount - chargeAmount if chargeSource == giftChargeSourceBag { // Bag 批量送礼按总数量扣库存，COIN 余额不变；每个目标回执仍保留原礼物价格，供房间贡献和麦位热度复用。 - senderAfter = sender.AvailableAmount + senderAfter = senderState.account.AvailableAmount + } + giftIncomeBalanceAfter := int64(0) + targetCoinState := accountStates[walletAccountLockKey(target.TargetUserID, ledger.AssetCoin)] + if giftIncomeCoinAmount > 0 { + if targetCoinState == nil { + return ledger.BatchGiftReceipt{}, xerr.New(xerr.Internal, "gift income account is missing") + } + giftIncomeBalanceAfter = targetCoinState.account.AvailableAmount + giftIncomeCoinAmount + if target.TargetUserID == command.SenderUserID && accountAssetType == ledger.AssetCoin { + // 批量里如果包含自己，当前目标的回执余额必须体现本目标扣费和收入后的净结果。 + giftIncomeBalanceAfter = senderAfter + giftIncomeCoinAmount + senderAfter = giftIncomeBalanceAfter + } } metadata := giftMetadata{ AppCode: command.AppCode, @@ -805,6 +931,10 @@ func (r *Repository) BatchDebitGift(ctx context.Context, command ledger.BatchDeb GiftPointAdded: 0, HeatValue: heatValue, BalanceAfter: senderAfter, + GiftIncomeCoinAmount: giftIncomeCoinAmount, + GiftIncomeRatioPercent: giftIncomeRatio.Percent, + GiftIncomeRatioRegionID: giftIncomeRatioRegionID, + GiftIncomeBalanceAfter: giftIncomeBalanceAfter, BillingReceipt: billingReceiptID(command.AppCode, target.CommandID), SenderUserID: command.SenderUserID, SenderRegionID: command.SenderRegionID, @@ -828,21 +958,18 @@ func (r *Repository) BatchDebitGift(ctx context.Context, command ledger.BatchDeb return ledger.BatchGiftReceipt{}, err } if chargeSource != giftChargeSourceBag { - if err := r.applyAccountDelta(ctx, tx, sender, -chargeAmount, 0, nowMs); err != nil { + debitAccount, err := r.applyTrackedAccountDelta(ctx, tx, senderState, -chargeAmount, 0, nowMs) + if 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++ - if err := r.insertEntry(ctx, tx, walletEntry{ TransactionID: transactionID, UserID: command.SenderUserID, AssetType: price.ChargeAssetType, AvailableDelta: -chargeAmount, FrozenDelta: 0, - AvailableAfter: senderAfter, - FrozenAfter: sender.FrozenAmount, + AvailableAfter: debitAccount.AvailableAmount, + FrozenAfter: debitAccount.FrozenAmount, CounterpartyUserID: target.TargetUserID, RoomID: command.RoomID, CreatedAtMS: nowMs, @@ -850,6 +977,35 @@ func (r *Repository) BatchDebitGift(ctx context.Context, command ledger.BatchDeb return ledger.BatchGiftReceipt{}, err } } + if giftIncomeCoinAmount > 0 { + incomeState := targetCoinState + if target.TargetUserID == command.SenderUserID && accountAssetType == ledger.AssetCoin { + // 自送目标复用 senderState 的当前版本，保证同一个 COIN 账户的扣费和收入顺序可审计、可重放。 + incomeState = senderState + } + incomeAccount, err := r.applyTrackedAccountDelta(ctx, tx, incomeState, giftIncomeCoinAmount, 0, nowMs) + if err != nil { + return ledger.BatchGiftReceipt{}, err + } + metadata.GiftIncomeBalanceAfter = incomeAccount.AvailableAmount + if target.TargetUserID == command.SenderUserID && accountAssetType == ledger.AssetCoin { + metadata.BalanceAfter = incomeAccount.AvailableAmount + } + if err := r.insertEntry(ctx, tx, walletEntry{ + TransactionID: transactionID, + UserID: target.TargetUserID, + AssetType: ledger.AssetCoin, + AvailableDelta: giftIncomeCoinAmount, + FrozenDelta: 0, + AvailableAfter: incomeAccount.AvailableAmount, + FrozenAfter: incomeAccount.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 { @@ -859,7 +1015,20 @@ func (r *Repository) BatchDebitGift(ctx context.Context, command ledger.BatchDeb metadata.HostPeriodDiamondVersion = hostPeriodDiamondVersion } + if chargeSource != giftChargeSourceBag { + if target.TargetUserID == command.SenderUserID && chargeAssetType == ledger.AssetCoin { + events = append(events, balanceChangedEvent(transactionID, target.CommandID, command.SenderUserID, ledger.AssetCoin, giftIncomeCoinAmount-chargeAmount, 0, metadata.BalanceAfter, senderState.account.FrozenAmount, senderState.account.Version, metadata, nowMs)) + } else { + events = append(events, balanceChangedEvent(transactionID, target.CommandID, command.SenderUserID, price.ChargeAssetType, -chargeAmount, 0, senderState.account.AvailableAmount, senderState.account.FrozenAmount, senderState.account.Version, metadata, nowMs)) + } + } + if giftIncomeCoinAmount > 0 && !(target.TargetUserID == command.SenderUserID && chargeAssetType == ledger.AssetCoin && chargeSource != giftChargeSourceBag) { + events = append(events, balanceChangedEvent(transactionID, target.CommandID, target.TargetUserID, ledger.AssetCoin, giftIncomeCoinAmount, 0, metadata.GiftIncomeBalanceAfter, targetCoinState.account.FrozenAmount, targetCoinState.account.Version, metadata, nowMs)) + } events = append(events, giftDebitedEvent(transactionID, target.CommandID, command.SenderUserID, chargeAssetType, -chargeAmount, 0, metadata, nowMs)) + if giftIncomeCoinAmount > 0 { + events = append(events, giftIncomeCreditedEvent(transactionID, target.CommandID, metadata, nowMs)) + } if hostPeriodDiamondAdded > 0 { events = append(events, hostPeriodDiamondCreditedEvent(transactionID, target.CommandID, metadata, nowMs)) } @@ -2340,6 +2509,77 @@ type walletAccount struct { UpdatedAtMS int64 } +type walletAccountLockRequest struct { + UserID int64 + AssetType string + CreateIfMissing bool +} + +type walletAccountState struct { + account walletAccount +} + +func walletAccountLockKey(userID int64, assetType string) string { + return fmt.Sprintf("%s:%d", strings.ToUpper(strings.TrimSpace(assetType)), userID) +} + +func (r *Repository) lockAccountsOrdered(ctx context.Context, tx *sql.Tx, requests []walletAccountLockRequest, nowMs int64) (map[string]*walletAccountState, error) { + merged := make(map[string]walletAccountLockRequest, len(requests)) + for _, request := range requests { + request.AssetType = ledger.NormalizeGiftChargeAssetType(request.AssetType) + if request.UserID <= 0 || strings.TrimSpace(request.AssetType) == "" { + return nil, xerr.New(xerr.InvalidArgument, "wallet account lock request is invalid") + } + key := walletAccountLockKey(request.UserID, request.AssetType) + if existing, ok := merged[key]; ok { + // 同一事务内同一用户同一资产只锁一次；任一调用方需要自动建账时必须保留 createIfMissing， + // 否则自送或背包送礼会因为去重丢掉收礼账户初始化语义。 + existing.CreateIfMissing = existing.CreateIfMissing || request.CreateIfMissing + merged[key] = existing + continue + } + merged[key] = request + } + + ordered := make([]walletAccountLockRequest, 0, len(merged)) + for _, request := range merged { + ordered = append(ordered, request) + } + sort.Slice(ordered, func(i, j int) bool { + if ordered[i].AssetType == ordered[j].AssetType { + return ordered[i].UserID < ordered[j].UserID + } + return ordered[i].AssetType < ordered[j].AssetType + }) + + states := make(map[string]*walletAccountState, len(ordered)) + for _, request := range ordered { + account, err := r.lockAccount(ctx, tx, request.UserID, request.AssetType, request.CreateIfMissing, nowMs) + if err != nil { + return nil, err + } + states[walletAccountLockKey(request.UserID, request.AssetType)] = &walletAccountState{account: account} + } + return states, nil +} + +func (r *Repository) applyTrackedAccountDelta(ctx context.Context, tx *sql.Tx, state *walletAccountState, availableDelta int64, frozenDelta int64, nowMs int64) (walletAccount, error) { + if state == nil { + return walletAccount{}, xerr.New(xerr.Internal, "wallet account state is missing") + } + before := state.account + if err := r.applyAccountDelta(ctx, tx, before, availableDelta, frozenDelta, nowMs); err != nil { + return walletAccount{}, err + } + after := before + after.AvailableAmount += availableDelta + after.FrozenAmount += frozenDelta + after.Version++ + after.UpdatedAtMS = nowMs + state.account = after + return after, nil +} + 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 { @@ -2547,6 +2787,17 @@ func giftDiamondAmount(chargeAmount int64, ratioPPM int64) (int64, error) { return product / 1_000_000, nil } +func giftIncomeRatio(giftTypeCode string, configuredRatio giftDiamondRatioSnapshot, configuredRegionID int64) (giftDiamondRatioSnapshot, int64) { + switch strings.TrimSpace(giftTypeCode) { + case resourcedomain.GiftTypeLucky, resourcedomain.GiftTypeSuperLucky: + // 幸运和超级幸运礼物继续使用后台全局比例，默认分别是 10% 和 1%；这里不复用普通礼物新增的 30% 收礼收入。 + return configuredRatio, configuredRegionID + default: + // 普通礼物收礼金币收入是本次新增固定口径；它不影响房间热度和主播周期钻石的后台配置比例。 + return giftDiamondRatioSnapshot{Percent: normalGiftIncomeRatioPercent, PPM: normalGiftIncomeRatioPPM}, 0 + } +} + func defaultGiftDiamondRatio(giftTypeCode string) giftDiamondRatioSnapshot { switch strings.TrimSpace(giftTypeCode) { case "lucky": @@ -3322,6 +3573,10 @@ type giftMetadata struct { GiftPointAdded int64 `json:"gift_point_added"` HeatValue int64 `json:"heat_value"` BalanceAfter int64 `json:"balance_after"` + GiftIncomeCoinAmount int64 `json:"gift_income_coin_amount"` + GiftIncomeRatioPercent string `json:"gift_income_ratio_percent,omitempty"` + GiftIncomeRatioRegionID int64 `json:"gift_income_ratio_region_id,omitempty"` + GiftIncomeBalanceAfter int64 `json:"gift_income_balance_after,omitempty"` BillingReceipt string `json:"billing_receipt_id"` SenderUserID int64 `json:"sender_user_id"` SenderRegionID int64 `json:"sender_region_id"` @@ -4277,6 +4532,37 @@ func giftDebitedEvent(transactionID string, commandID string, userID int64, asse } } +func giftIncomeCreditedEvent(transactionID string, commandID string, metadata giftMetadata, nowMs int64) walletOutboxEvent { + return walletOutboxEvent{ + EventID: eventID(transactionID, "WalletGiftIncomeCredited", metadata.TargetUserID, ledger.AssetCoin), + EventType: "WalletGiftIncomeCredited", + TransactionID: transactionID, + CommandID: commandID, + UserID: metadata.TargetUserID, + AssetType: ledger.AssetCoin, + AvailableDelta: metadata.GiftIncomeCoinAmount, + FrozenDelta: 0, + Payload: map[string]any{ + "transaction_id": transactionID, + "command_id": commandID, + "user_id": metadata.TargetUserID, + "sender_user_id": metadata.SenderUserID, + "room_id": metadata.RoomID, + "gift_id": metadata.GiftID, + "gift_count": metadata.GiftCount, + "gift_type_code": metadata.GiftTypeCode, + "gift_charge_amount": metadata.ChargeAmount, + "charge_source": metadata.ChargeSource, + "amount": metadata.GiftIncomeCoinAmount, + "ratio_percent": metadata.GiftIncomeRatioPercent, + "ratio_region_id": metadata.GiftIncomeRatioRegionID, + "gift_income_balance_after": metadata.GiftIncomeBalanceAfter, + "created_at_ms": nowMs, + }, + CreatedAtMS: nowMs, + } +} + func hostPeriodDiamondCreditedEvent(transactionID string, commandID string, metadata giftMetadata, nowMs int64) walletOutboxEvent { return walletOutboxEvent{ EventID: eventID(transactionID, "HostPeriodDiamondCredited", metadata.TargetUserID, ledger.AssetHostPeriodDiamond), diff --git a/services/wallet-service/internal/storage/mysql/resource_repository.go b/services/wallet-service/internal/storage/mysql/resource_repository.go index 3781e024..b91c19f5 100644 --- a/services/wallet-service/internal/storage/mysql/resource_repository.go +++ b/services/wallet-service/internal/storage/mysql/resource_repository.go @@ -168,7 +168,7 @@ func (r *Repository) CreateResource(ctx context.Context, command resourcedomain. return resource, nil } -// UpdateResource 全量更新资源配置；当前开发阶段不做兼容性合并。 +// UpdateResource 全量更新资源配置； func (r *Repository) UpdateResource(ctx context.Context, command resourcedomain.ResourceCommand) (resourcedomain.Resource, error) { if r == nil || r.db == nil { return resourcedomain.Resource{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")