diff --git a/api/proto/wallet/v1/wallet.pb.go b/api/proto/wallet/v1/wallet.pb.go index ba73e0e4..f5f13732 100644 --- a/api/proto/wallet/v1/wallet.pb.go +++ b/api/proto/wallet/v1/wallet.pb.go @@ -16058,15 +16058,23 @@ func (x *RetryRedPacketRefundResponse) GetServerTimeMs() int64 { // CronBatchRequest 是 cron-service 触发 wallet-service 后台批处理的统一请求。 type CronBatchRequest struct { - state protoimpl.MessageState `protogen:"open.v1"` - RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` - AppCode string `protobuf:"bytes,2,opt,name=app_code,json=appCode,proto3" json:"app_code,omitempty"` - RunId string `protobuf:"bytes,3,opt,name=run_id,json=runId,proto3" json:"run_id,omitempty"` - WorkerId string `protobuf:"bytes,4,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty"` - BatchSize int32 `protobuf:"varint,5,opt,name=batch_size,json=batchSize,proto3" json:"batch_size,omitempty"` - LockTtlMs int64 `protobuf:"varint,6,opt,name=lock_ttl_ms,json=lockTtlMs,proto3" json:"lock_ttl_ms,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` + AppCode string `protobuf:"bytes,2,opt,name=app_code,json=appCode,proto3" json:"app_code,omitempty"` + RunId string `protobuf:"bytes,3,opt,name=run_id,json=runId,proto3" json:"run_id,omitempty"` + WorkerId string `protobuf:"bytes,4,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty"` + BatchSize int32 `protobuf:"varint,5,opt,name=batch_size,json=batchSize,proto3" json:"batch_size,omitempty"` + LockTtlMs int64 `protobuf:"varint,6,opt,name=lock_ttl_ms,json=lockTtlMs,proto3" json:"lock_ttl_ms,omitempty"` + // trigger_mode 为空时由 wallet-service 按 automatic 处理；后台手动结算显式传 manual。 + TriggerMode string `protobuf:"bytes,7,opt,name=trigger_mode,json=triggerMode,proto3" json:"trigger_mode,omitempty"` + // cycle_key 为空时由 wallet-service 按当前 UTC 月处理；后台手动补结可指定周期。 + CycleKey string `protobuf:"bytes,8,opt,name=cycle_key,json=cycleKey,proto3" json:"cycle_key,omitempty"` + // user_ids 为空表示批处理扫描全部候选；后台手动结算只传已勾选主播。 + UserIds []int64 `protobuf:"varint,9,rep,packed,name=user_ids,json=userIds,proto3" json:"user_ids,omitempty"` + // settlement_role 为空时结算完整 Host/Agency 权益；后台人工入口可传 host 或 agency 拆分钱包。 + SettlementRole string `protobuf:"bytes,10,opt,name=settlement_role,json=settlementRole,proto3" json:"settlement_role,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *CronBatchRequest) Reset() { @@ -16141,6 +16149,34 @@ func (x *CronBatchRequest) GetLockTtlMs() int64 { return 0 } +func (x *CronBatchRequest) GetTriggerMode() string { + if x != nil { + return x.TriggerMode + } + return "" +} + +func (x *CronBatchRequest) GetCycleKey() string { + if x != nil { + return x.CycleKey + } + return "" +} + +func (x *CronBatchRequest) GetUserIds() []int64 { + if x != nil { + return x.UserIds + } + return nil +} + +func (x *CronBatchRequest) GetSettlementRole() string { + if x != nil { + return x.SettlementRole + } + return "" +} + // CronBatchResponse 返回一个批处理 run 的执行摘要。 type CronBatchResponse struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -17863,7 +17899,7 @@ const file_proto_wallet_v1_wallet_proto_rawDesc = "" + "\x1cRetryRedPacketRefundResponse\x122\n" + "\x06packet\x18\x01 \x01(\v2\x1a.hyapp.wallet.v1.RedPacketR\x06packet\x12'\n" + "\x0frefunded_amount\x18\x02 \x01(\x03R\x0erefundedAmount\x12$\n" + - "\x0eserver_time_ms\x18\x03 \x01(\x03R\fserverTimeMs\"\xbf\x01\n" + + "\x0eserver_time_ms\x18\x03 \x01(\x03R\fserverTimeMs\"\xc3\x02\n" + "\x10CronBatchRequest\x12\x1d\n" + "\n" + "request_id\x18\x01 \x01(\tR\trequestId\x12\x19\n" + @@ -17872,7 +17908,12 @@ const file_proto_wallet_v1_wallet_proto_rawDesc = "" + "\tworker_id\x18\x04 \x01(\tR\bworkerId\x12\x1d\n" + "\n" + "batch_size\x18\x05 \x01(\x05R\tbatchSize\x12\x1e\n" + - "\vlock_ttl_ms\x18\x06 \x01(\x03R\tlockTtlMs\"\xc6\x01\n" + + "\vlock_ttl_ms\x18\x06 \x01(\x03R\tlockTtlMs\x12!\n" + + "\ftrigger_mode\x18\a \x01(\tR\vtriggerMode\x12\x1b\n" + + "\tcycle_key\x18\b \x01(\tR\bcycleKey\x12\x19\n" + + "\buser_ids\x18\t \x03(\x03R\auserIds\x12'\n" + + "\x0fsettlement_role\x18\n" + + " \x01(\tR\x0esettlementRole\"\xc6\x01\n" + "\x11CronBatchResponse\x12#\n" + "\rclaimed_count\x18\x01 \x01(\x05R\fclaimedCount\x12'\n" + "\x0fprocessed_count\x18\x02 \x01(\x05R\x0eprocessedCount\x12#\n" + diff --git a/api/proto/wallet/v1/wallet.proto b/api/proto/wallet/v1/wallet.proto index 5347b3bd..01fa15ae 100644 --- a/api/proto/wallet/v1/wallet.proto +++ b/api/proto/wallet/v1/wallet.proto @@ -1795,6 +1795,14 @@ message CronBatchRequest { string worker_id = 4; int32 batch_size = 5; int64 lock_ttl_ms = 6; + // trigger_mode 为空时由 wallet-service 按 automatic 处理；后台手动结算显式传 manual。 + string trigger_mode = 7; + // cycle_key 为空时由 wallet-service 按当前 UTC 月处理；后台手动补结可指定周期。 + string cycle_key = 8; + // user_ids 为空表示批处理扫描全部候选；后台手动结算只传已勾选主播。 + repeated int64 user_ids = 9; + // settlement_role 为空时结算完整 Host/Agency 权益；后台人工入口可传 host 或 agency 拆分钱包。 + string settlement_role = 10; } // CronBatchResponse 返回一个批处理 run 的执行摘要。 diff --git a/docs/openapi/gateway.swagger.yaml b/docs/openapi/gateway.swagger.yaml index c6c33c86..58c106e1 100644 --- a/docs/openapi/gateway.swagger.yaml +++ b/docs/openapi/gateway.swagger.yaml @@ -3008,7 +3008,7 @@ definitions: - host_center - agency_center - bd_center - - bd_leader_center + - superadmin_center - coin_seller_center - settings - language diff --git a/docs/我的页概览接口.md b/docs/我的页概览接口.md index 643f6dd2..e432e164 100644 --- a/docs/我的页概览接口.md +++ b/docs/我的页概览接口.md @@ -119,7 +119,7 @@ Authorization: Bearer | `agency_center` | `roles.is_agency == true` | | `manager_center` | `roles.is_manager == true` | | `bd_center` | `roles.is_bd == true` | -| `bd_leader_center` | `roles.is_bd_leader == true` | +| `superadmin_center` | `roles.is_bd_leader == true` | | `coin_seller_center` | `roles.is_coin_seller == true` | | `settings` | always true | | `language` | always true | diff --git a/pkg/rocketmqx/rocketmqx.go b/pkg/rocketmqx/rocketmqx.go index 5d947106..3dd58b59 100644 --- a/pkg/rocketmqx/rocketmqx.go +++ b/pkg/rocketmqx/rocketmqx.go @@ -9,6 +9,7 @@ import ( "fmt" "net" "strings" + "sync" "time" rocketmq "github.com/apache/rocketmq-client-go/v2" @@ -173,7 +174,9 @@ func (p *Producer) SendSync(ctx context.Context, message Message) error { // Consumer wraps a clustering push consumer. type Consumer struct { - client rocketmq.PushConsumer + mu sync.Mutex + client rocketmq.PushConsumer + started bool } // NewConsumer creates a RocketMQ push consumer. Subscribe before Start. @@ -256,7 +259,19 @@ func (c *Consumer) Start() error { if c == nil || c.client == nil { return errors.New("rocketmq consumer is not configured") } - return c.client.Start() + c.mu.Lock() + defer c.mu.Unlock() + if c.started { + return nil + } + // RocketMQ 的 PushConsumer 只有 Start 成功后才具备完整的内部统计状态。 + // 本地 namesrv 未启动时 Start 会返回错误，此时必须保持 started=false； + // 上层为了回滚已启动消费者会调用 Shutdown，未启动 consumer 直接跳过，避免依赖库在半初始化状态下 panic。 + if err := c.client.Start(); err != nil { + return err + } + c.started = true + return nil } // Shutdown releases the underlying consumer. @@ -264,6 +279,16 @@ func (c *Consumer) Shutdown() error { if c == nil || c.client == nil { return nil } + c.mu.Lock() + defer c.mu.Unlock() + if !c.started { + return nil + } + defer func() { + // 无论底层关闭是否成功，本 wrapper 都不再重复调用 Shutdown； + // RocketMQ client 的二次关闭或半初始化关闭不具备幂等保证，重复尝试只会放大退出路径风险。 + c.started = false + }() return c.client.Shutdown() } diff --git a/server/admin/cmd/server/main.go b/server/admin/cmd/server/main.go index f1628b88..a4ba9fa5 100644 --- a/server/admin/cmd/server/main.go +++ b/server/admin/cmd/server/main.go @@ -80,6 +80,7 @@ import ( "gorm.io/driver/mysql" "gorm.io/gorm" "gorm.io/gorm/logger" + walletv1 "hyapp.local/api/proto/wallet/v1" ) func main() { @@ -264,7 +265,7 @@ func main() { Health: healthmodule.New(store, redisClient, jobStatus), HostAgencyPolicy: hostagencypolicymodule.New(store, walletDB, auditHandler), HostOrg: hostorgmodule.New(userclient.NewGRPC(userConn), walletclient.NewGRPC(walletConn), userDB, walletDB, sqlDB, auditHandler), - HostSalarySettlement: hostsalarysettlementmodule.New(walletDB, userDB), + HostSalarySettlement: hostsalarysettlementmodule.New(walletDB, userDB, walletv1.NewWalletCronServiceClient(walletConn), cfg.WalletService.RequestTimeout, auditHandler), InviteActivityReward: inviteactivityrewardmodule.New(activityclient.NewGRPC(activityConn), userDB, auditHandler), Job: jobmodule.New(store, cfg, auditHandler), LevelConfig: levelconfigmodule.New(activityclient.NewGRPC(activityConn), walletclient.NewGRPC(walletConn), auditHandler), diff --git a/server/admin/internal/integration/userclient/client.go b/server/admin/internal/integration/userclient/client.go index 5c708268..d9e2bcc9 100644 --- a/server/admin/internal/integration/userclient/client.go +++ b/server/admin/internal/integration/userclient/client.go @@ -508,6 +508,7 @@ type BDProfile struct { CreatedByUsername string `json:"createdByUsername"` CreatedByAvatar string `json:"createdByAvatar"` SubBDCount int64 `json:"subBdCount"` + SalaryWallet SalaryWallet `json:"salaryWallet"` } type Agency struct { @@ -529,6 +530,7 @@ type Agency struct { ParentBDUsername string `json:"parentBdUsername"` ParentBDAvatar string `json:"parentBdAvatar"` RegionName string `json:"regionName"` + SalaryWallet SalaryWallet `json:"salaryWallet"` } type HostProfile struct { @@ -551,6 +553,16 @@ type HostProfile struct { CurrentAgencyOwnerDisplayUserID string `json:"currentAgencyOwnerDisplayUserId"` CurrentAgencyOwnerUsername string `json:"currentAgencyOwnerUsername"` CurrentAgencyOwnerAvatar string `json:"currentAgencyOwnerAvatar"` + SalaryWallet SalaryWallet `json:"salaryWallet"` +} + +type SalaryWallet struct { + Role string `json:"role"` + UserID int64 `json:"userId,string"` + AssetType string `json:"assetType"` + AvailableAmount int64 `json:"availableAmount"` + FrozenAmount int64 `json:"frozenAmount"` + UpdatedAtMs int64 `json:"updatedAtMs"` } type CoinSellerProfile struct { diff --git a/server/admin/internal/modules/hostorg/handler.go b/server/admin/internal/modules/hostorg/handler.go index e23146c1..d097740f 100644 --- a/server/admin/internal/modules/hostorg/handler.go +++ b/server/admin/internal/modules/hostorg/handler.go @@ -89,6 +89,36 @@ func (h *Handler) ListCoinSellers(c *gin.Context) { response.OK(c, response.Page{Items: items, Page: query.Page, PageSize: query.PageSize, Total: total}) } +func (h *Handler) ListSalaryWalletHistory(c *gin.Context) { + query, ok := parseSalaryWalletHistoryQuery(c) + if !ok { + return + } + items, total, err := h.service.ListSalaryWalletHistory(c.Request.Context(), query) + if err != nil { + response.BadRequest(c, err.Error()) + return + } + response.OK(c, response.Page{Items: items, Page: query.Page, PageSize: query.PageSize, Total: total}) +} + +func (h *Handler) AdjustSalaryWallet(c *gin.Context) { + var req salaryWalletAdjustRequest + if err := c.ShouldBindJSON(&req); err != nil { + response.BadRequest(c, "工资钱包操作参数不正确") + return + } + result, err := h.service.AdjustSalaryWallet(c.Request.Context(), adminActorID(c), middleware.CurrentRequestID(c), req) + if err != nil { + response.BadRequest(c, err.Error()) + return + } + writeHostOrgAuditLog(c, h.audit, "salary-wallet-adjust", "wallet_entries", req.TargetUserID.Int64(), + fmt.Sprintf("role=%s target_user_id=%d amount_minor=%d transaction_id=%s reason=%q", + result.Role, req.TargetUserID.Int64(), result.AmountMinor, result.TransactionID, strings.TrimSpace(req.Reason))) + response.Created(c, result) +} + func (h *Handler) CreateBDLeader(c *gin.Context) { var req createBDLeaderRequest if err := c.ShouldBindJSON(&req); err != nil { @@ -394,6 +424,25 @@ func parseListQuery(c *gin.Context) (listQuery, bool) { return normalizeListQuery(query), true } +func parseSalaryWalletHistoryQuery(c *gin.Context) (salaryWalletHistoryQuery, bool) { + options := shared.ListOptions(c) + userID, ok := parseOptionalInt64Query(c, "userId", "user_id") + if !ok { + return salaryWalletHistoryQuery{}, false + } + if userID <= 0 { + response.BadRequest(c, "user_id 参数不正确") + return salaryWalletHistoryQuery{}, false + } + query := salaryWalletHistoryQuery{ + Page: options.Page, + PageSize: options.PageSize, + Role: c.Query("role"), + UserID: userID, + } + return normalizeSalaryWalletHistoryQuery(query), true +} + func parseOptionalInt64Query(c *gin.Context, primary string, fallback string) (int64, bool) { raw := strings.TrimSpace(c.Query(primary)) if raw == "" { diff --git a/server/admin/internal/modules/hostorg/reader.go b/server/admin/internal/modules/hostorg/reader.go index 1f35efee..00f2c486 100644 --- a/server/admin/internal/modules/hostorg/reader.go +++ b/server/admin/internal/modules/hostorg/reader.go @@ -186,6 +186,9 @@ func (r *Reader) ListBDProfiles(ctx context.Context, query listQuery, role strin return nil, 0, err } } + if err := r.fillBDProfileSalaryWallets(ctx, role, items); err != nil { + return nil, 0, err + } return items, total, nil } @@ -283,6 +286,9 @@ func (r *Reader) listBDLeaderProfiles(ctx context.Context, query listQuery) ([]* if err := r.fillBDLeaderPositionAliases(ctx, appCode, items); err != nil { return nil, 0, err } + if err := r.fillBDProfileSalaryWallets(ctx, salaryWalletRoleBDLeader, items); err != nil { + return nil, 0, err + } return items, total, nil } @@ -342,6 +348,9 @@ func (r *Reader) GetBDLeader(ctx context.Context, userID int64) (*userclient.BDP if err := r.fillBDLeaderPositionAliases(ctx, appCode, items); err != nil { return nil, err } + if err := r.fillBDProfileSalaryWallets(ctx, salaryWalletRoleBDLeader, items); err != nil { + return nil, err + } return item, nil } @@ -614,7 +623,13 @@ func (r *Reader) ListAgencies(ctx context.Context, query listQuery) ([]*userclie } items = append(items, item) } - return items, total, rows.Err() + if err := rows.Err(); err != nil { + return nil, 0, err + } + if err := r.fillAgencySalaryWallets(ctx, items); err != nil { + return nil, 0, err + } + return items, total, nil } // ListHostProfiles 只读主播身份事实；成员变更和主播状态变更不在这个列表接口里发生。 @@ -721,6 +736,9 @@ func (r *Reader) ListHostProfiles(ctx context.Context, query listQuery) ([]*user for _, item := range items { item.Diamond = diamonds[item.UserID] } + if err := r.fillHostSalaryWallets(ctx, items); err != nil { + return nil, 0, err + } if query.SortBy == "diamond" { sortHostProfilesByDiamond(items, query.SortDirection) items = paginateHostProfiles(items, query.Page, query.PageSize) diff --git a/server/admin/internal/modules/hostorg/routes.go b/server/admin/internal/modules/hostorg/routes.go index c7df19d6..5485c916 100644 --- a/server/admin/internal/modules/hostorg/routes.go +++ b/server/admin/internal/modules/hostorg/routes.go @@ -21,6 +21,8 @@ func RegisterRoutes(protected *gin.RouterGroup, h *Handler) { protected.POST("/admin/agencies/:agency_id/delete", middleware.RequirePermission("agency:delete"), h.DeleteAgency) protected.POST("/admin/agencies/:agency_id/join-enabled", middleware.RequirePermission("agency:status"), h.SetAgencyJoinEnabled) protected.GET("/admin/hosts", middleware.RequirePermission("host:view"), h.ListHosts) + protected.GET("/admin/host-org/salary-wallets/history", middleware.RequireAnyPermission("host:view", "agency:view", "bd:view"), h.ListSalaryWalletHistory) + protected.POST("/admin/host-org/salary-wallets/adjust", middleware.RequirePermission("host-salary-settlement:settle"), h.AdjustSalaryWallet) protected.GET("/admin/coin-sellers", middleware.RequirePermission("coin-seller:view"), h.ListCoinSellers) protected.POST("/admin/coin-sellers", middleware.RequirePermission("coin-seller:create"), h.CreateCoinSeller) protected.PATCH("/admin/coin-sellers/:user_id/status", middleware.RequirePermission("coin-seller:update"), h.SetCoinSellerStatus) diff --git a/server/admin/internal/modules/hostorg/salary_wallet.go b/server/admin/internal/modules/hostorg/salary_wallet.go new file mode 100644 index 00000000..904ddd59 --- /dev/null +++ b/server/admin/internal/modules/hostorg/salary_wallet.go @@ -0,0 +1,557 @@ +package hostorg + +import ( + "context" + "encoding/json" + "fmt" + "strconv" + "strings" + "time" + + "hyapp-admin-server/internal/appctx" + "hyapp-admin-server/internal/integration/userclient" + walletv1 "hyapp.local/api/proto/wallet/v1" +) + +const ( + salaryWalletRoleHost = "host" + salaryWalletRoleAgency = "agency" + salaryWalletRoleBD = "bd" + salaryWalletRoleBDLeader = "bd_leader" + + assetHostSalaryUSD = "HOST_SALARY_USD" + assetAgencySalaryUSD = "AGENCY_SALARY_USD" + assetBDSalaryUSD = "BD_SALARY_USD" + assetAdminSalaryUSD = "ADMIN_SALARY_USD" + + salaryWalletAdjustIncrease = "increase" + salaryWalletAdjustDecrease = "decrease" +) + +type salaryWalletHistoryQuery struct { + Page int + PageSize int + Role string + UserID int64 +} + +type salaryWalletAdjustRequest struct { + CommandID string `json:"commandId"` + Role string `json:"role" binding:"required"` + TargetUserID flexibleInt64 `json:"targetUserId" binding:"required"` + Type string `json:"type" binding:"required"` + AmountMinor int64 `json:"amountMinor" binding:"required"` + Reason string `json:"reason" binding:"required"` +} + +type salaryWalletOperatorDTO struct { + AdminID string `json:"adminId"` + Username string `json:"username"` + Name string `json:"name"` +} + +type salaryWalletHistoryItem struct { + EntryID int64 `json:"entryId"` + TransactionID string `json:"transactionId"` + CommandID string `json:"commandId"` + BizType string `json:"bizType"` + ExternalRef string `json:"externalRef"` + Role string `json:"role"` + UserID string `json:"userId"` + AssetType string `json:"assetType"` + Direction string `json:"direction"` + AmountMinor int64 `json:"amountMinor"` + AvailableDelta int64 `json:"availableDelta"` + AvailableAfter int64 `json:"availableAfter"` + FrozenDelta int64 `json:"frozenDelta"` + FrozenAfter int64 `json:"frozenAfter"` + CounterpartyUserID string `json:"counterpartyUserId"` + RoomID string `json:"roomId"` + Reason string `json:"reason"` + OperatorUserID string `json:"operatorUserId"` + Operator salaryWalletOperatorDTO `json:"operator"` + Metadata map[string]any `json:"metadata"` + CreatedAtMS int64 `json:"createdAtMs"` +} + +type salaryWalletAdjustResult struct { + TransactionID string `json:"transactionId"` + Role string `json:"role"` + UserID string `json:"userId"` + AssetType string `json:"assetType"` + AmountMinor int64 `json:"amountMinor"` + Reason string `json:"reason"` + Balance userclient.SalaryWallet `json:"balance"` +} + +func (r *Reader) fillBDProfileSalaryWallets(ctx context.Context, role string, items []*userclient.BDProfile) error { + userIDs := make([]int64, 0, len(items)) + for _, item := range items { + if item.UserID > 0 { + userIDs = append(userIDs, item.UserID) + } + } + wallets, err := r.salaryWalletBalances(ctx, appctx.FromContext(ctx), role, userIDs) + if err != nil { + return err + } + for _, item := range items { + item.SalaryWallet = wallets[item.UserID] + } + return nil +} + +func (r *Reader) fillAgencySalaryWallets(ctx context.Context, items []*userclient.Agency) error { + userIDs := make([]int64, 0, len(items)) + for _, item := range items { + if item.OwnerUserID > 0 { + userIDs = append(userIDs, item.OwnerUserID) + } + } + wallets, err := r.salaryWalletBalances(ctx, appctx.FromContext(ctx), salaryWalletRoleAgency, userIDs) + if err != nil { + return err + } + for _, item := range items { + item.SalaryWallet = wallets[item.OwnerUserID] + } + return nil +} + +func (r *Reader) fillHostSalaryWallets(ctx context.Context, items []*userclient.HostProfile) error { + userIDs := make([]int64, 0, len(items)) + for _, item := range items { + if item.UserID > 0 { + userIDs = append(userIDs, item.UserID) + } + } + wallets, err := r.salaryWalletBalances(ctx, appctx.FromContext(ctx), salaryWalletRoleHost, userIDs) + if err != nil { + return err + } + for _, item := range items { + item.SalaryWallet = wallets[item.UserID] + } + return nil +} + +func (r *Reader) salaryWalletBalances(ctx context.Context, appCode string, role string, userIDs []int64) (map[int64]userclient.SalaryWallet, error) { + role = normalizeSalaryWalletRole(role) + assetType := salaryWalletAssetType(role) + result := make(map[int64]userclient.SalaryWallet, len(userIDs)) + for _, userID := range positiveUniqueIDs(userIDs) { + // 列表必须稳定返回该身份对应的钱包对象；没有 wallet_accounts 行时余额就是 0，而不是让前端猜资产类型。 + result[userID] = userclient.SalaryWallet{Role: role, UserID: userID, AssetType: assetType} + } + if r == nil || r.walletDB == nil || len(result) == 0 { + return result, nil + } + args := []any{appCode, assetType} + for userID := range result { + args = append(args, userID) + } + rows, err := r.walletDB.QueryContext(ctx, fmt.Sprintf(` + SELECT user_id, available_amount, frozen_amount, updated_at_ms + FROM wallet_accounts + WHERE app_code = ? AND asset_type = ? AND user_id IN (%s) + `, sqlPlaceholders(len(result))), args...) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var userID int64 + wallet := userclient.SalaryWallet{Role: role, AssetType: assetType} + if err := rows.Scan(&userID, &wallet.AvailableAmount, &wallet.FrozenAmount, &wallet.UpdatedAtMs); err != nil { + return nil, err + } + wallet.UserID = userID + result[userID] = wallet + } + return result, rows.Err() +} + +func (s *Service) ListSalaryWalletHistory(ctx context.Context, query salaryWalletHistoryQuery) ([]salaryWalletHistoryItem, int64, error) { + query = normalizeSalaryWalletHistoryQuery(query) + if s == nil || s.reader == nil || s.reader.walletDB == nil { + return nil, 0, fmt.Errorf("wallet mysql is not configured") + } + if query.Role == "" || query.UserID <= 0 { + return nil, 0, fmt.Errorf("role and user_id are required") + } + // 历史查询先验证用户确实拥有对应组织身份，避免拿 host 身份去看 agency/BD 的工资资产流水。 + if err := s.reader.requireSalaryWalletIdentity(ctx, query.Role, query.UserID); err != nil { + return nil, 0, err + } + return s.reader.listSalaryWalletHistory(ctx, query) +} + +func (r *Reader) listSalaryWalletHistory(ctx context.Context, query salaryWalletHistoryQuery) ([]salaryWalletHistoryItem, int64, error) { + appCode := appctx.FromContext(ctx) + assetType := salaryWalletAssetType(query.Role) + args := []any{appCode, query.UserID, assetType} + whereSQL := `WHERE e.app_code = ? AND e.user_id = ? AND e.asset_type = ?` + var total int64 + if err := r.walletDB.QueryRowContext(ctx, ` + SELECT COUNT(*) + FROM wallet_entries e + JOIN wallet_transactions wt ON wt.app_code = e.app_code AND wt.transaction_id = e.transaction_id + `+whereSQL, args..., + ).Scan(&total); err != nil { + return nil, 0, err + } + rows, err := r.walletDB.QueryContext(ctx, ` + SELECT e.entry_id, e.transaction_id, wt.command_id, wt.biz_type, wt.external_ref, + e.available_delta, e.available_after, e.frozen_delta, e.frozen_after, + e.counterparty_user_id, e.room_id, COALESCE(CAST(wt.metadata_json AS CHAR), '{}'), e.created_at_ms + FROM wallet_entries e + JOIN wallet_transactions wt ON wt.app_code = e.app_code AND wt.transaction_id = e.transaction_id + `+whereSQL+` + ORDER BY e.created_at_ms DESC, e.entry_id DESC + LIMIT ? OFFSET ?`, append(args, query.PageSize, offset(query.Page, query.PageSize))...) + if err != nil { + return nil, 0, err + } + defer rows.Close() + + items := make([]salaryWalletHistoryItem, 0, query.PageSize) + operatorIDs := make([]int64, 0, query.PageSize) + for rows.Next() { + var item salaryWalletHistoryItem + var counterpartyUserID int64 + var metadataJSON string + if err := rows.Scan( + &item.EntryID, + &item.TransactionID, + &item.CommandID, + &item.BizType, + &item.ExternalRef, + &item.AvailableDelta, + &item.AvailableAfter, + &item.FrozenDelta, + &item.FrozenAfter, + &counterpartyUserID, + &item.RoomID, + &metadataJSON, + &item.CreatedAtMS, + ); err != nil { + return nil, 0, err + } + metadata, err := parseSalaryWalletMetadata(metadataJSON) + if err != nil { + return nil, 0, err + } + operatorID := salaryWalletMetadataInt64(metadata, "operator_user_id", "operatorUserId") + if operatorID > 0 { + operatorIDs = append(operatorIDs, operatorID) + } + item.Role = query.Role + item.UserID = strconv.FormatInt(query.UserID, 10) + item.AssetType = assetType + item.Direction = salaryWalletDirection(item.AvailableDelta) + item.AmountMinor = item.AvailableDelta + item.CounterpartyUserID = formatSalaryWalletID(counterpartyUserID) + item.Reason = salaryWalletMetadataString(metadata, "reason") + item.OperatorUserID = formatSalaryWalletID(operatorID) + item.Metadata = metadata + items = append(items, item) + } + if err := rows.Err(); err != nil { + return nil, 0, err + } + operators, err := r.salaryWalletAdminProfiles(ctx, operatorIDs) + if err != nil { + return nil, 0, err + } + for i := range items { + operatorID, _ := strconv.ParseInt(items[i].OperatorUserID, 10, 64) + if operator, ok := operators[operatorID]; ok { + items[i].Operator = operator + } + } + return items, total, nil +} + +func (s *Service) AdjustSalaryWallet(ctx context.Context, actorID int64, requestID string, req salaryWalletAdjustRequest) (salaryWalletAdjustResult, error) { + if s == nil || s.walletClient == nil || s.reader == nil { + return salaryWalletAdjustResult{}, fmt.Errorf("wallet client is not configured") + } + role := normalizeSalaryWalletRole(req.Role) + targetUserID := req.TargetUserID.Int64() + if role == "" || targetUserID <= 0 || req.AmountMinor <= 0 { + return salaryWalletAdjustResult{}, fmt.Errorf("role, target_user_id and amount are required") + } + if actorID <= 0 { + return salaryWalletAdjustResult{}, fmt.Errorf("operator_user_id is required") + } + reason := strings.TrimSpace(req.Reason) + if reason == "" { + return salaryWalletAdjustResult{}, fmt.Errorf("reason is required") + } + if len(reason) > 512 { + return salaryWalletAdjustResult{}, fmt.Errorf("reason is too long") + } + // 调整工资钱包前必须确认目标用户具备该身份；同一个 user_id 可能同时有多种身份，资产类型不能由前端随意指定。 + if err := s.reader.requireSalaryWalletIdentity(ctx, role, targetUserID); err != nil { + return salaryWalletAdjustResult{}, err + } + amount := req.AmountMinor + adjustType := strings.ToLower(strings.TrimSpace(req.Type)) + switch adjustType { + case salaryWalletAdjustIncrease: + case salaryWalletAdjustDecrease: + // wallet-service 的 AdminCreditAsset 以带符号金额入账，扣减也走同一条幂等账务链路并由钱包层做余额不透支校验。 + amount = -amount + default: + return salaryWalletAdjustResult{}, fmt.Errorf("salary adjustment type is invalid") + } + commandID := strings.TrimSpace(req.CommandID) + if commandID == "" { + commandID = defaultSalaryWalletCommandID(requestID) + } + if len(commandID) > 128 { + return salaryWalletAdjustResult{}, fmt.Errorf("command_id is too long") + } + assetType := salaryWalletAssetType(role) + // 手工增减工资只提交目标身份对应的工资资产；交易、分录、余额和幂等命令统一由 wallet-service 落库。 + resp, err := s.walletClient.AdminCreditAsset(ctx, &walletv1.AdminCreditAssetRequest{ + CommandId: commandID, + TargetUserId: targetUserID, + AssetType: assetType, + Amount: amount, + OperatorUserId: actorID, + Reason: reason, + EvidenceRef: defaultSalaryWalletEvidenceRef(requestID), + AppCode: appctx.FromContext(ctx), + }) + if err != nil { + return salaryWalletAdjustResult{}, err + } + wallet := userclient.SalaryWallet{ + Role: role, + UserID: targetUserID, + AssetType: assetType, + AvailableAmount: 0, + FrozenAmount: 0, + UpdatedAtMs: time.Now().UnixMilli(), + } + if balance := resp.GetBalance(); balance != nil { + wallet.AvailableAmount = balance.GetAvailableAmount() + wallet.FrozenAmount = balance.GetFrozenAmount() + } + return salaryWalletAdjustResult{ + TransactionID: resp.GetTransactionId(), + Role: role, + UserID: strconv.FormatInt(targetUserID, 10), + AssetType: assetType, + AmountMinor: amount, + Reason: reason, + Balance: wallet, + }, nil +} + +func (r *Reader) requireSalaryWalletIdentity(ctx context.Context, role string, userID int64) error { + if r == nil || r.db == nil { + return errUserDBNotConfigured() + } + if userID <= 0 { + return fmt.Errorf("user_id is required") + } + appCode := appctx.FromContext(ctx) + var query string + args := []any{appCode, userID} + switch normalizeSalaryWalletRole(role) { + case salaryWalletRoleHost: + query = `SELECT COUNT(1) FROM host_profiles WHERE app_code = ? AND user_id = ?` + case salaryWalletRoleAgency: + query = `SELECT COUNT(1) FROM agencies WHERE app_code = ? AND owner_user_id = ? AND status <> 'deleted'` + case salaryWalletRoleBD: + query = `SELECT COUNT(1) FROM bd_profiles WHERE app_code = ? AND user_id = ? AND role = 'bd'` + case salaryWalletRoleBDLeader: + query = `SELECT COUNT(1) FROM bd_leader_profiles WHERE app_code = ? AND user_id = ?` + default: + return fmt.Errorf("salary wallet role is invalid") + } + var count int + if err := r.db.QueryRowContext(ctx, query, args...).Scan(&count); err != nil { + return err + } + if count == 0 { + return fmt.Errorf("salary wallet identity not found") + } + return nil +} + +func (r *Reader) salaryWalletAdminProfiles(ctx context.Context, adminIDs []int64) (map[int64]salaryWalletOperatorDTO, error) { + result := make(map[int64]salaryWalletOperatorDTO, len(adminIDs)) + if r == nil || r.adminDB == nil { + return result, nil + } + adminIDs = positiveUniqueIDs(adminIDs) + if len(adminIDs) == 0 { + return result, nil + } + args := make([]any, 0, len(adminIDs)) + for _, id := range adminIDs { + args = append(args, id) + } + rows, err := r.adminDB.QueryContext(ctx, fmt.Sprintf(` + SELECT id, COALESCE(username, ''), COALESCE(name, '') + FROM admin_users + WHERE id IN (%s) + `, sqlPlaceholders(len(adminIDs))), args...) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var id int64 + var item salaryWalletOperatorDTO + if err := rows.Scan(&id, &item.Username, &item.Name); err != nil { + return nil, err + } + item.AdminID = strconv.FormatInt(id, 10) + result[id] = item + } + return result, rows.Err() +} + +func normalizeSalaryWalletHistoryQuery(query salaryWalletHistoryQuery) salaryWalletHistoryQuery { + if query.Page < 1 { + query.Page = 1 + } + if query.PageSize < 1 { + query.PageSize = 20 + } + if query.PageSize > 100 { + query.PageSize = 100 + } + query.Role = normalizeSalaryWalletRole(query.Role) + return query +} + +func normalizeSalaryWalletRole(role string) string { + switch strings.ToLower(strings.TrimSpace(role)) { + case salaryWalletRoleHost: + return salaryWalletRoleHost + case salaryWalletRoleAgency: + return salaryWalletRoleAgency + case salaryWalletRoleBD: + return salaryWalletRoleBD + case salaryWalletRoleBDLeader, "leader", "bdleader", "admin": + return salaryWalletRoleBDLeader + default: + return "" + } +} + +func salaryWalletAssetType(role string) string { + switch normalizeSalaryWalletRole(role) { + case salaryWalletRoleHost: + return assetHostSalaryUSD + case salaryWalletRoleAgency: + return assetAgencySalaryUSD + case salaryWalletRoleBD: + return assetBDSalaryUSD + case salaryWalletRoleBDLeader: + // BD Leader 在工资政策里沿用 admin 结算身份，所以列表和历史都读取 ADMIN_SALARY_USD。 + return assetAdminSalaryUSD + default: + return "" + } +} + +func salaryWalletDirection(delta int64) string { + if delta < 0 { + return "expense" + } + return "income" +} + +func parseSalaryWalletMetadata(value string) (map[string]any, error) { + value = strings.TrimSpace(value) + if value == "" || value == "null" { + return map[string]any{}, nil + } + var metadata map[string]any + if err := json.Unmarshal([]byte(value), &metadata); err != nil { + return nil, err + } + if metadata == nil { + return map[string]any{}, nil + } + return metadata, nil +} + +func salaryWalletMetadataString(metadata map[string]any, key string) string { + value, ok := metadata[key] + if !ok || value == nil { + return "" + } + if text, ok := value.(string); ok { + return strings.TrimSpace(text) + } + return strings.TrimSpace(fmt.Sprint(value)) +} + +func salaryWalletMetadataInt64(metadata map[string]any, keys ...string) int64 { + for _, key := range keys { + value, ok := metadata[key] + if !ok || value == nil { + continue + } + switch typed := value.(type) { + case float64: + return int64(typed) + case int64: + return typed + case int: + return int64(typed) + case string: + parsed, err := strconv.ParseInt(strings.TrimSpace(typed), 10, 64) + if err == nil { + return parsed + } + } + } + return 0 +} + +func formatSalaryWalletID(value int64) string { + if value <= 0 { + return "" + } + return strconv.FormatInt(value, 10) +} + +func positiveUniqueIDs(values []int64) []int64 { + seen := make(map[int64]struct{}, len(values)) + result := make([]int64, 0, len(values)) + for _, value := range values { + if value <= 0 { + continue + } + if _, ok := seen[value]; ok { + continue + } + seen[value] = struct{}{} + result = append(result, value) + } + return result +} + +func defaultSalaryWalletCommandID(requestID string) string { + requestID = strings.TrimSpace(requestID) + if requestID == "" { + return "salary-wallet-adjustment-" + strconv.FormatInt(time.Now().UTC().UnixNano(), 10) + } + return "salary-wallet-adjustment-" + requestID +} + +func defaultSalaryWalletEvidenceRef(requestID string) string { + requestID = strings.TrimSpace(requestID) + if requestID == "" { + return "salary-wallet-adjustment" + } + return "admin-request:" + requestID +} diff --git a/server/admin/internal/modules/hostorg/service_test.go b/server/admin/internal/modules/hostorg/service_test.go index a4e0acae..5c671bfc 100644 --- a/server/admin/internal/modules/hostorg/service_test.go +++ b/server/admin/internal/modules/hostorg/service_test.go @@ -54,6 +54,30 @@ func TestUserCountryRegionSQLUsesCountryMapping(t *testing.T) { } } +// TestSalaryWalletAssetTypeByRole 锁定四个组织身份读取各自工资钱包资产；BD Leader 复用 Admin 工资资产，不能误读普通 BD 资产。 +func TestSalaryWalletAssetTypeByRole(t *testing.T) { + cases := map[string]string{ + "host": assetHostSalaryUSD, + "agency": assetAgencySalaryUSD, + "bd": assetBDSalaryUSD, + "bd_leader": assetAdminSalaryUSD, + "admin": assetAdminSalaryUSD, + } + for role, want := range cases { + if got := salaryWalletAssetType(role); got != want { + t.Fatalf("role %s salary wallet asset mismatch: got %q want %q", role, got, want) + } + } +} + +// TestNormalizeSalaryWalletHistoryQueryCapsPageSize 避免历史抽屉一次拉太多钱包分录，同时兼容 admin 作为 BD Leader 的旧口径。 +func TestNormalizeSalaryWalletHistoryQueryCapsPageSize(t *testing.T) { + query := normalizeSalaryWalletHistoryQuery(salaryWalletHistoryQuery{Page: 0, PageSize: 500, Role: "admin", UserID: 7}) + if query.Page != 1 || query.PageSize != 100 || query.Role != salaryWalletRoleBDLeader || query.UserID != 7 { + t.Fatalf("salary wallet history query normalized incorrectly: %+v", query) + } +} + // TestSortCoinSellerListItemsComputedFields 验证 wallet 聚合字段排序只改变展示顺序，不改动币商身份数据。 func TestSortCoinSellerListItemsComputedFields(t *testing.T) { items := []*CoinSellerListItem{ diff --git a/server/admin/internal/modules/hostsalarysettlement/dto.go b/server/admin/internal/modules/hostsalarysettlement/dto.go index 698dc167..ef45ce3d 100644 --- a/server/admin/internal/modules/hostsalarysettlement/dto.go +++ b/server/admin/internal/modules/hostsalarysettlement/dto.go @@ -1,5 +1,47 @@ package hostsalarysettlement +import ( + "encoding/json" + "fmt" + "strconv" + "strings" +) + +type settlementUserID int64 + +func (v *settlementUserID) UnmarshalJSON(data []byte) error { + raw := strings.TrimSpace(string(data)) + if raw == "" || raw == "null" { + *v = 0 + return nil + } + if strings.HasPrefix(raw, `"`) { + var text string + if err := json.Unmarshal(data, &text); err != nil { + return err + } + raw = strings.TrimSpace(text) + if raw == "" { + *v = 0 + return nil + } + } + parsed, err := strconv.ParseInt(raw, 10, 64) + if err != nil { + return fmt.Errorf("invalid settlement user id %q", raw) + } + *v = settlementUserID(parsed) + return nil +} + +func settlementUserIDValues(items []settlementUserID) []int64 { + out := make([]int64, 0, len(items)) + for _, item := range items { + out = append(out, int64(item)) + } + return out +} + type userDTO struct { UserID string `json:"user_id"` DisplayUserID string `json:"display_user_id"` @@ -7,11 +49,56 @@ type userDTO struct { Avatar string `json:"avatar"` } +type pendingDTO struct { + UserID string `json:"user_id"` + User userDTO `json:"user"` + AgencyOwnerUserID string `json:"agency_owner_user_id"` + AgencyOwner userDTO `json:"agency_owner"` + Role string `json:"role"` + CycleKey string `json:"cycle_key"` + RegionID int64 `json:"region_id"` + RegionName string `json:"region_name"` + PolicyID uint64 `json:"policy_id"` + PolicyName string `json:"policy_name"` + SettlementType string `json:"settlement_type"` + TriggerMode string `json:"trigger_mode"` + LevelNo int32 `json:"level_no"` + RequiredDiamonds int64 `json:"required_diamonds"` + TotalDiamonds int64 `json:"total_diamonds"` + HostSalaryUSDMinorTarget int64 `json:"host_salary_usd_minor_target"` + HostSalaryUSDTarget string `json:"host_salary_usd_target"` + HostSalaryUSDMinorDelta int64 `json:"host_salary_usd_minor_delta"` + HostSalaryUSDDelta string `json:"host_salary_usd_delta"` + HostCoinRewardTarget int64 `json:"host_coin_reward_target"` + HostCoinRewardDelta int64 `json:"host_coin_reward_delta"` + AgencySalaryUSDMinorTarget int64 `json:"agency_salary_usd_minor_target"` + AgencySalaryUSDTarget string `json:"agency_salary_usd_target"` + AgencySalaryUSDMinorDelta int64 `json:"agency_salary_usd_minor_delta"` + AgencySalaryUSDDelta string `json:"agency_salary_usd_delta"` + ResidualUSDMinorDelta int64 `json:"residual_usd_minor_delta"` + ResidualUSDDelta string `json:"residual_usd_delta"` + TotalUSDMinorDelta int64 `json:"total_usd_minor_delta"` + TotalUSDDelta string `json:"total_usd_delta"` + SettledLevelNo int32 `json:"settled_level_no"` + SettledHostSalaryUSDMinor int64 `json:"settled_host_salary_usd_minor"` + SettledHostSalaryUSD string `json:"settled_host_salary_usd"` + SettledHostCoinReward int64 `json:"settled_host_coin_reward"` + SettledAgencySalaryUSDMinor int64 `json:"settled_agency_salary_usd_minor"` + SettledAgencySalaryUSD string `json:"settled_agency_salary_usd"` + LastSettledTotalDiamonds int64 `json:"last_settled_total_diamonds"` + ResidualUSDMinor int64 `json:"residual_usd_minor"` + ResidualUSD string `json:"residual_usd"` + MonthEndClearedAtMS int64 `json:"month_end_cleared_at_ms"` + Status string `json:"status"` +} + type settlementDTO struct { SettlementID string `json:"settlement_id"` CommandID string `json:"command_id"` TransactionID string `json:"transaction_id"` SettlementType string `json:"settlement_type"` + TriggerMode string `json:"trigger_mode"` + SettlementRole string `json:"settlement_role"` UserID string `json:"user_id"` User userDTO `json:"user"` AgencyOwnerUserID string `json:"agency_owner_user_id"` @@ -44,3 +131,40 @@ type query struct { StartAtMS int64 EndAtMS int64 } + +type pendingQuery struct { + Page int + PageSize int + Role string + CycleKey string + SettlementType string + TriggerMode string + RegionID int64 + CountryID int64 + CountryCode string + UserID int64 +} + +type settleRequest struct { + Role string `json:"role"` + CycleKey string `json:"cycle_key"` + SettlementType string `json:"settlement_type"` + TriggerMode string `json:"trigger_mode"` + RegionID int64 `json:"region_id"` + CountryID int64 `json:"country_id"` + CountryCode string `json:"country_code"` + UserIDs []settlementUserID `json:"user_ids"` +} + +type settleResultDTO struct { + Role string `json:"role"` + CycleKey string `json:"cycle_key"` + SettlementType string `json:"settlement_type"` + TriggerMode string `json:"trigger_mode"` + RequestedCount int `json:"requested_count"` + ProcessedCount int `json:"processed_count"` + SuccessCount int `json:"success_count"` + SkippedCount int `json:"skipped_count"` + FailureCount int `json:"failure_count"` + Records []settlementDTO `json:"records"` +} diff --git a/server/admin/internal/modules/hostsalarysettlement/handler.go b/server/admin/internal/modules/hostsalarysettlement/handler.go index ddb4e3f7..b07f8d0c 100644 --- a/server/admin/internal/modules/hostsalarysettlement/handler.go +++ b/server/admin/internal/modules/hostsalarysettlement/handler.go @@ -2,22 +2,40 @@ package hostsalarysettlement import ( "database/sql" + "fmt" "strconv" "strings" + "time" "hyapp-admin-server/internal/appctx" "hyapp-admin-server/internal/modules/shared" "hyapp-admin-server/internal/response" + walletv1 "hyapp.local/api/proto/wallet/v1" + "github.com/gin-gonic/gin" ) type Handler struct { + audit shared.OperationLogger service *Service } -func New(walletDB *sql.DB, userDB *sql.DB) *Handler { - return &Handler{service: NewService(walletDB, userDB)} +func New(walletDB *sql.DB, userDB *sql.DB, walletCron walletv1.WalletCronServiceClient, requestTimeout time.Duration, audit shared.OperationLogger) *Handler { + return &Handler{audit: audit, service: NewService(walletDB, userDB, walletCron, requestTimeout)} +} + +func (h *Handler) ListPending(c *gin.Context) { + req, ok := parsePendingQuery(c) + if !ok { + return + } + items, total, err := h.service.ListPending(c.Request.Context(), appctx.FromContext(c.Request.Context()), req) + if err != nil { + response.ServerError(c, "获取 Host/Agency 待结算工资失败") + return + } + response.OK(c, response.Page{Items: items, Page: req.Page, PageSize: req.PageSize, Total: total}) } func (h *Handler) List(c *gin.Context) { @@ -33,6 +51,37 @@ func (h *Handler) List(c *gin.Context) { response.OK(c, response.Page{Items: items, Page: req.Page, PageSize: req.PageSize, Total: total}) } +func (h *Handler) Settle(c *gin.Context) { + var req settleRequest + if err := c.ShouldBindJSON(&req); err != nil { + response.BadRequest(c, "结算参数不正确") + return + } + rawSettlementType := strings.TrimSpace(req.SettlementType) + rawTriggerMode := strings.TrimSpace(req.TriggerMode) + rawRole := strings.TrimSpace(req.Role) + req = normalizeSettleRequest(req) + if rawRole != "" && normalizeSettlementRole(rawRole) == "" { + response.BadRequest(c, "结算角色不正确") + return + } + if rawSettlementType != "" && normalizeSettlementType(rawSettlementType) == "" { + response.BadRequest(c, "结算类型不正确") + return + } + if rawTriggerMode != "" && normalizeTriggerMode(rawTriggerMode) == "" { + response.BadRequest(c, "触发方式不正确") + return + } + result, err := h.service.Settle(c.Request.Context(), appctx.FromContext(c.Request.Context()), int64(shared.ActorFromContext(c).UserID), req) + if err != nil { + response.ServerError(c, "Host/Agency 工资结算失败") + return + } + shared.OperationLog(c, h.audit, "settle-host-salary", "host_salary_settlement_records", "success", fmt.Sprintf("role=%s settlement_type=%s trigger_mode=%s cycle_key=%s success=%d skipped=%d failed=%d", result.Role, result.SettlementType, result.TriggerMode, result.CycleKey, result.SuccessCount, result.SkippedCount, result.FailureCount)) + response.OK(c, result) +} + func parseQuery(c *gin.Context) (query, bool) { options := shared.ListOptions(c) userID, ok := optionalInt64(c, "user_id", "userId") @@ -87,6 +136,45 @@ func parseQuery(c *gin.Context) (query, bool) { return req, true } +func parsePendingQuery(c *gin.Context) (pendingQuery, bool) { + options := shared.ListOptions(c) + req := pendingQuery{ + Page: options.Page, + PageSize: options.PageSize, + Role: normalizeSettlementRole(firstQuery(c, "role", "settlement_role", "settlementRole")), + CycleKey: strings.TrimSpace(firstQuery(c, "cycle_key", "cycleKey")), + SettlementType: normalizeSettlementType(firstQuery(c, "settlement_type", "settlementType")), + TriggerMode: normalizeTriggerMode(firstQuery(c, "trigger_mode", "triggerMode")), + CountryCode: strings.ToUpper(strings.TrimSpace(firstQuery(c, "country_code", "countryCode"))), + } + var ok bool + if req.RegionID, ok = optionalInt64(c, "region_id", "regionId"); !ok { + response.BadRequest(c, "区域 ID 不正确") + return pendingQuery{}, false + } + if req.CountryID, ok = optionalInt64(c, "country_id", "countryId"); !ok { + response.BadRequest(c, "国家 ID 不正确") + return pendingQuery{}, false + } + if req.UserID, ok = optionalInt64(c, "user_id", "userId"); !ok { + response.BadRequest(c, "主播用户 ID 不正确") + return pendingQuery{}, false + } + if raw := strings.TrimSpace(firstQuery(c, "role", "settlement_role", "settlementRole")); raw != "" && strings.ToLower(raw) != "all" && req.Role == "" { + response.BadRequest(c, "结算角色不正确") + return pendingQuery{}, false + } + if raw := strings.TrimSpace(firstQuery(c, "settlement_type", "settlementType")); raw != "" && strings.ToLower(raw) != "all" && req.SettlementType == "" { + response.BadRequest(c, "结算类型不正确") + return pendingQuery{}, false + } + if raw := strings.TrimSpace(firstQuery(c, "trigger_mode", "triggerMode")); raw != "" && strings.ToLower(raw) != "all" && req.TriggerMode == "" { + response.BadRequest(c, "触发方式不正确") + return pendingQuery{}, false + } + return normalizePendingQuery(req), true +} + func optionalInt64(c *gin.Context, keys ...string) (int64, bool) { value := strings.TrimSpace(firstQuery(c, keys...)) if value == "" { diff --git a/server/admin/internal/modules/hostsalarysettlement/routes.go b/server/admin/internal/modules/hostsalarysettlement/routes.go index 912f784b..f1e76b52 100644 --- a/server/admin/internal/modules/hostsalarysettlement/routes.go +++ b/server/admin/internal/modules/hostsalarysettlement/routes.go @@ -11,5 +11,7 @@ func RegisterRoutes(protected *gin.RouterGroup, h *Handler) { return } + protected.GET("/admin/host-salary-settlements/pending", middleware.RequirePermission("host-salary-settlement:view"), h.ListPending) + protected.POST("/admin/host-salary-settlements/settle", middleware.RequirePermission("host-salary-settlement:settle"), h.Settle) protected.GET("/admin/host-salary-settlements", middleware.RequirePermission("host-salary-settlement:view"), h.List) } diff --git a/server/admin/internal/modules/hostsalarysettlement/service.go b/server/admin/internal/modules/hostsalarysettlement/service.go index eecd1483..7d18484b 100644 --- a/server/admin/internal/modules/hostsalarysettlement/service.go +++ b/server/admin/internal/modules/hostsalarysettlement/service.go @@ -3,16 +3,39 @@ package hostsalarysettlement import ( "context" "database/sql" + "errors" "fmt" + "math" + "math/big" + "sort" "strconv" "strings" + "time" + + walletv1 "hyapp.local/api/proto/wallet/v1" + + "google.golang.org/grpc" ) -const maxPageSize = 100 +const ( + maxPageSize = 100 + + settlementTypeDaily = "daily" + settlementTypeHalfMonth = "half_month" + settlementTypeMonthEnd = "month_end" + triggerAutomatic = "automatic" + triggerManual = "manual" + settlementRoleAll = "all" + settlementRoleHost = "host" + settlementRoleAgency = "agency" + statusPending = "pending" +) type Service struct { - walletDB *sql.DB - userDB *sql.DB + walletDB *sql.DB + userDB *sql.DB + walletCron walletCronClient + requestTimeout time.Duration } type userProfile struct { @@ -22,13 +45,321 @@ type userProfile struct { Avatar string } -func NewService(walletDB *sql.DB, userDB *sql.DB) *Service { - return &Service{walletDB: walletDB, userDB: userDB} +type walletCronClient interface { + ProcessHostSalaryDailySettlementBatch(ctx context.Context, in *walletv1.CronBatchRequest, opts ...grpc.CallOption) (*walletv1.CronBatchResponse, error) + ProcessHostSalaryHalfMonthSettlementBatch(ctx context.Context, in *walletv1.CronBatchRequest, opts ...grpc.CallOption) (*walletv1.CronBatchResponse, error) + ProcessHostSalaryMonthEndBatch(ctx context.Context, in *walletv1.CronBatchRequest, opts ...grpc.CallOption) (*walletv1.CronBatchResponse, error) +} + +type hostPolicy struct { + PolicyID uint64 + Name string + RegionID int64 + SettlementMode string + TriggerMode string + ResidualDiamondToUSDRate string + Levels []hostPolicyLevel +} + +type hostPolicyLevel struct { + LevelNo int + RequiredDiamonds int64 + HostSalaryUSDMinor int64 + HostCoinReward int64 + AgencySalaryUSDMinor int64 +} + +type hostProgress struct { + SettledLevelNo int + SettledHostSalaryUSDMinor int64 + SettledHostCoinReward int64 + SettledAgencySalaryUSDMinor int64 + LastSettledTotalDiamonds int64 + ResidualUSDMinor int64 + MonthEndClearedAtMS int64 +} + +type hostCandidate struct { + UserID int64 + AgencyOwnerUserID int64 + Role string + CycleKey string + RegionID int64 + RegionName string + SettlementType string + TotalDiamonds int64 + Policy hostPolicy + Level hostPolicyLevel + Progress hostProgress + HostSalaryUSDDelta int64 + HostCoinRewardDelta int64 + AgencySalaryUSDDelta int64 + ResidualUSDDelta int64 +} + +func NewService(walletDB *sql.DB, userDB *sql.DB, walletCron walletCronClient, requestTimeout time.Duration) *Service { + return &Service{walletDB: walletDB, userDB: userDB, walletCron: walletCron, requestTimeout: requestTimeout} +} + +func (s *Service) ListPending(ctx context.Context, appCode string, req pendingQuery) ([]pendingDTO, int64, error) { + if err := s.ensureReady(ctx); err != nil { + return nil, 0, err + } + req = normalizePendingQuery(req) + // 待结算列表只做只读重算，不写 progress、不写钱包；真正入账必须走 Settle 调 wallet-service。 + candidates, err := s.collectPendingCandidates(ctx, strings.TrimSpace(appCode), req, time.Now().UTC().UnixMilli()) + if err != nil { + return nil, 0, err + } + sortHostCandidates(candidates) + total := int64(len(candidates)) + pageItems := paginateHostCandidates(candidates, req.Page, req.PageSize) + userIDs := make([]int64, 0, len(pageItems)*2) + for _, item := range pageItems { + userIDs = append(userIDs, item.UserID, item.AgencyOwnerUserID) + } + profiles, err := s.userProfiles(ctx, appCode, userIDs) + if err != nil { + return nil, 0, err + } + items := make([]pendingDTO, 0, len(pageItems)) + for _, item := range pageItems { + items = append(items, pendingDTOFromCandidate(item, profiles[item.UserID], profiles[item.AgencyOwnerUserID])) + } + return items, total, nil +} + +func (s *Service) Settle(ctx context.Context, appCode string, operatorAdminID int64, req settleRequest) (settleResultDTO, error) { + if err := s.ensureReady(ctx); err != nil { + return settleResultDTO{}, err + } + req = normalizeSettleRequest(req) + userIDs := uniquePositiveUserIDs(settlementUserIDValues(req.UserIDs)) + result := settleResultDTO{ + CycleKey: req.CycleKey, + SettlementType: req.SettlementType, + TriggerMode: req.TriggerMode, + Role: req.Role, + RequestedCount: len(userIDs), + Records: []settlementDTO{}, + } + if len(userIDs) == 0 { + return result, nil + } + // 后台传来的 user_ids 必须先回到待结算候选里二次校验；客户端不能靠篡改区域或周期结算页面外用户。 + candidates, err := s.collectPendingCandidates(ctx, strings.TrimSpace(appCode), pendingQuery{ + Page: 1, + PageSize: maxPageSize, + Role: req.Role, + CycleKey: req.CycleKey, + SettlementType: req.SettlementType, + TriggerMode: req.TriggerMode, + RegionID: req.RegionID, + CountryID: req.CountryID, + CountryCode: req.CountryCode, + }, time.Now().UTC().UnixMilli()) + if err != nil { + return settleResultDTO{}, err + } + allowed := map[int64]struct{}{} + selected := map[int64]struct{}{} + for _, id := range userIDs { + selected[id] = struct{}{} + } + for _, candidate := range candidates { + if _, ok := selected[candidate.UserID]; ok { + allowed[candidate.UserID] = struct{}{} + } + } + settleIDs := make([]int64, 0, len(allowed)) + for _, id := range userIDs { + if _, ok := allowed[id]; ok { + settleIDs = append(settleIDs, id) + } + } + if len(settleIDs) == 0 { + result.SkippedCount = len(userIDs) + return result, nil + } + if s.requestTimeout <= 0 { + s.requestTimeout = 3 * time.Second + } + callCtx, cancel := context.WithTimeout(ctx, s.requestTimeout) + defer cancel() + // wallet-service 是 Host/Agency 工资账务 owner；admin-server 只传人工选择范围，不复制钱包入账逻辑。 + resp, err := s.callWalletSettlement(callCtx, appCode, req, operatorAdminID, settleIDs) + if err != nil { + return settleResultDTO{}, err + } + result.ProcessedCount = int(resp.GetProcessedCount()) + result.SuccessCount = int(resp.GetSuccessCount()) + result.FailureCount = int(resp.GetFailureCount()) + result.SkippedCount = len(userIDs) - len(settleIDs) + result.ProcessedCount - result.SuccessCount - result.FailureCount + if result.SkippedCount < 0 { + result.SkippedCount = 0 + } + return result, nil +} + +func (s *Service) callWalletSettlement(ctx context.Context, appCode string, req settleRequest, operatorAdminID int64, userIDs []int64) (*walletv1.CronBatchResponse, error) { + if s.walletCron == nil { + return nil, fmt.Errorf("wallet cron client is not configured") + } + batchReq := &walletv1.CronBatchRequest{ + RequestId: fmt.Sprintf("admin-host-salary-%d", time.Now().UTC().UnixNano()), + AppCode: strings.TrimSpace(appCode), + RunId: fmt.Sprintf("admin-host-salary-%s-%d", req.CycleKey, time.Now().UTC().UnixNano()), + WorkerId: fmt.Sprintf("admin-%d", operatorAdminID), + BatchSize: int32(len(userIDs)), + TriggerMode: req.TriggerMode, + SettlementRole: req.Role, + CycleKey: req.CycleKey, + UserIds: append([]int64(nil), userIDs...), + } + switch req.SettlementType { + case settlementTypeDaily: + return s.walletCron.ProcessHostSalaryDailySettlementBatch(ctx, batchReq) + case settlementTypeHalfMonth: + return s.walletCron.ProcessHostSalaryHalfMonthSettlementBatch(ctx, batchReq) + case settlementTypeMonthEnd: + return s.walletCron.ProcessHostSalaryMonthEndBatch(ctx, batchReq) + default: + return nil, fmt.Errorf("settlement_type is invalid") + } +} + +func (s *Service) collectPendingCandidates(ctx context.Context, appCode string, req pendingQuery, nowMS int64) ([]hostCandidate, error) { + req = normalizePendingQuery(req) + regions, err := s.regionFilter(ctx, appCode, req.RegionID, req.CountryID, req.CountryCode) + if err != nil { + return nil, err + } + regionClause := "" + userClause := "" + progressClause := "AND a.total_diamonds > COALESCE(progress.last_settled_total_diamonds, 0)" + if req.SettlementType == settlementTypeMonthEnd { + progressClause = "" + } + args := []any{appCode, req.CycleKey} + if len(regions) > 0 { + ids := keysInt64FromSet(regions) + regionClause = "AND a.region_id IN (" + placeholders(len(ids)) + ")" + for _, id := range ids { + args = append(args, id) + } + } else if req.RegionID > 0 || req.CountryID > 0 || req.CountryCode != "" { + regionClause = "AND a.region_id = -1" + } + if req.UserID > 0 { + userClause = "AND a.user_id = ?" + args = append(args, req.UserID) + } + args = append(args, nowMS, nowMS) + triggerClause := "" + if req.TriggerMode != "" { + triggerClause = "AND p.settlement_trigger_mode = ?" + args = append(args, req.TriggerMode) + } + modeClause := "" + if req.SettlementType != settlementTypeMonthEnd { + modeClause = "AND p.settlement_mode = ?" + args = append(args, req.SettlementType) + } + rows, err := s.walletDB.QueryContext(ctx, fmt.Sprintf(` + SELECT a.user_id, a.cycle_key, a.region_id, a.agency_owner_user_id, a.total_diamonds, + COALESCE(progress.settled_level_no, 0), COALESCE(progress.settled_host_salary_usd_minor, 0), + COALESCE(progress.settled_host_coin_reward, 0), COALESCE(progress.settled_agency_salary_usd_minor, 0), + COALESCE(progress.last_settled_total_diamonds, 0), COALESCE(progress.residual_usd_minor, 0), + COALESCE(progress.month_end_cleared_at_ms, 0) + FROM host_period_diamond_accounts a + LEFT JOIN host_salary_settlement_progress progress + ON progress.app_code = a.app_code AND progress.user_id = a.user_id AND progress.cycle_key = a.cycle_key + WHERE a.app_code = ? + AND a.cycle_key = ? + AND a.total_diamonds > 0 + AND COALESCE(progress.month_end_cleared_at_ms, 0) = 0 + %s + %s + %s + AND EXISTS ( + SELECT 1 + FROM host_agency_salary_policies p + WHERE p.app_code = a.app_code + AND p.region_id = a.region_id + AND p.status = 'active' + AND p.effective_from_ms <= ? + AND (p.effective_to_ms = 0 OR p.effective_to_ms > ?) + %s + %s + ) + ORDER BY a.updated_at_ms ASC, a.user_id ASC`, regionClause, userClause, progressClause, triggerClause, modeClause), args...) + if err != nil { + return nil, err + } + defer rows.Close() + regionNames, _ := s.regionNames(ctx, appCode) + out := []hostCandidate{} + policyCache := map[string]hostPolicy{} + for rows.Next() { + var item hostCandidate + if err := rows.Scan( + &item.UserID, + &item.CycleKey, + &item.RegionID, + &item.AgencyOwnerUserID, + &item.TotalDiamonds, + &item.Progress.SettledLevelNo, + &item.Progress.SettledHostSalaryUSDMinor, + &item.Progress.SettledHostCoinReward, + &item.Progress.SettledAgencySalaryUSDMinor, + &item.Progress.LastSettledTotalDiamonds, + &item.Progress.ResidualUSDMinor, + &item.Progress.MonthEndClearedAtMS, + ); err != nil { + return nil, err + } + cacheKey := fmt.Sprintf("%d:%s:%s", item.RegionID, req.SettlementType, req.TriggerMode) + policy, ok := policyCache[cacheKey] + if !ok { + policy, ok, err = s.resolvePolicy(ctx, appCode, item.RegionID, req.SettlementType, req.TriggerMode, nowMS) + if err != nil { + return nil, err + } + if !ok { + continue + } + policyCache[cacheKey] = policy + } + item.Policy = policy + item.RegionName = regionNames[item.RegionID] + item.Role = req.Role + item.SettlementType = req.SettlementType + item.Level = highestHostLevel(policy.Levels, item.TotalDiamonds) + item.HostSalaryUSDDelta = positiveDelta(item.Level.HostSalaryUSDMinor, item.Progress.SettledHostSalaryUSDMinor) + item.HostCoinRewardDelta = positiveDelta(item.Level.HostCoinReward, item.Progress.SettledHostCoinReward) + item.AgencySalaryUSDDelta = positiveDelta(item.Level.AgencySalaryUSDMinor, item.Progress.SettledAgencySalaryUSDMinor) + if item.AgencyOwnerUserID <= 0 { + item.AgencySalaryUSDDelta = 0 + } + if req.SettlementType == settlementTypeMonthEnd { + residualDiamonds := residualDiamondRemainder(item.TotalDiamonds, item.Level) + residualTarget, err := usdMinorFromDiamondRate(residualDiamonds, policy.ResidualDiamondToUSDRate) + if err != nil { + return nil, err + } + item.ResidualUSDDelta = positiveDelta(residualTarget, item.Progress.ResidualUSDMinor) + } + // 待结算按身份拆分：Host 只看主播工资/金币/剩余折美元，Agency 只看代理工资。 + if hostCandidateHasPendingRole(item, req.Role) { + out = append(out, item) + } + } + return out, rows.Err() } func (s *Service) List(ctx context.Context, appCode string, req query) ([]settlementDTO, int64, error) { - if s == nil || s.walletDB == nil { - return nil, 0, fmt.Errorf("wallet mysql is not configured") + if err := s.ensureReady(ctx); err != nil { + return nil, 0, err } req = normalizeQuery(req) whereSQL, args := settlementWhere(appCode, req) @@ -38,7 +369,7 @@ func (s *Service) List(ctx context.Context, appCode string, req query) ([]settle return nil, 0, err } rows, err := s.walletDB.QueryContext(ctx, ` - SELECT settlement_id, command_id, transaction_id, settlement_type, user_id, agency_owner_user_id, + SELECT settlement_id, command_id, transaction_id, settlement_type, trigger_mode, settlement_role, user_id, agency_owner_user_id, cycle_key, policy_id, level_no, total_diamonds, host_salary_usd_minor_delta, host_coin_reward_delta, agency_salary_usd_minor_delta, residual_usd_minor_delta, status, reason, created_at_ms @@ -64,6 +395,8 @@ func (s *Service) List(ctx context.Context, appCode string, req query) ([]settle &item.CommandID, &item.TransactionID, &item.SettlementType, + &item.TriggerMode, + &item.SettlementRole, &userID, &agencyOwnerUserID, &item.CycleKey, @@ -153,6 +486,130 @@ func settlementWhere(appCode string, req query) (string, []any) { return "WHERE " + strings.Join(conditions, " AND "), args } +func (s *Service) resolvePolicy(ctx context.Context, appCode string, regionID int64, settlementType string, triggerMode string, nowMS int64) (hostPolicy, bool, error) { + conditions := []string{"app_code = ?", "region_id = ?", "status = 'active'", "effective_from_ms <= ?", "(effective_to_ms = 0 OR effective_to_ms > ?)"} + args := []any{appCode, regionID, nowMS, nowMS} + if triggerMode != "" { + conditions = append(conditions, "settlement_trigger_mode = ?") + args = append(args, triggerMode) + } + if settlementType != settlementTypeMonthEnd { + conditions = append(conditions, "settlement_mode = ?") + args = append(args, settlementType) + } + row := s.walletDB.QueryRowContext(ctx, ` + SELECT policy_id, name, region_id, settlement_mode, settlement_trigger_mode, CAST(residual_diamond_to_usd_rate AS CHAR) + FROM host_agency_salary_policies + WHERE `+strings.Join(conditions, " AND ")+` + ORDER BY effective_from_ms DESC, policy_id DESC + LIMIT 1`, args...) + var policy hostPolicy + if err := row.Scan(&policy.PolicyID, &policy.Name, &policy.RegionID, &policy.SettlementMode, &policy.TriggerMode, &policy.ResidualDiamondToUSDRate); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return hostPolicy{}, false, nil + } + return hostPolicy{}, false, err + } + levels, err := s.policyLevels(ctx, appCode, policy.PolicyID) + if err != nil { + return hostPolicy{}, false, err + } + policy.Levels = levels + return policy, true, nil +} + +func (s *Service) policyLevels(ctx context.Context, appCode string, policyID uint64) ([]hostPolicyLevel, error) { + rows, err := s.walletDB.QueryContext(ctx, ` + SELECT level_no, required_diamonds, host_salary_usd_minor, host_coin_reward, agency_salary_usd_minor + FROM host_agency_salary_policy_levels + WHERE app_code = ? AND policy_id = ? AND status = 'active' + ORDER BY required_diamonds ASC, level_no ASC`, appCode, policyID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []hostPolicyLevel + for rows.Next() { + var item hostPolicyLevel + if err := rows.Scan(&item.LevelNo, &item.RequiredDiamonds, &item.HostSalaryUSDMinor, &item.HostCoinReward, &item.AgencySalaryUSDMinor); err != nil { + return nil, err + } + out = append(out, item) + } + return out, rows.Err() +} + +func (s *Service) regionFilter(ctx context.Context, appCode string, regionID int64, countryID int64, countryCode string) (map[int64]struct{}, error) { + if regionID > 0 { + return map[int64]struct{}{regionID: {}}, nil + } + countryCode = strings.ToUpper(strings.TrimSpace(countryCode)) + if countryID > 0 && countryCode == "" { + if err := s.userDB.QueryRowContext(ctx, `SELECT country_code FROM countries WHERE app_code = ? AND country_id = ?`, appCode, countryID).Scan(&countryCode); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return map[int64]struct{}{}, nil + } + return nil, err + } + } + if countryCode == "" { + return nil, nil + } + rows, err := s.userDB.QueryContext(ctx, ` + SELECT region_id + FROM region_countries + WHERE app_code = ? AND country_code = ? AND status = 'active'`, appCode, countryCode) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[int64]struct{}{} + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + out[id] = struct{}{} + } + return out, rows.Err() +} + +func (s *Service) regionNames(ctx context.Context, appCode string) (map[int64]string, error) { + rows, err := s.userDB.QueryContext(ctx, `SELECT region_id, name FROM regions WHERE app_code = ?`, appCode) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[int64]string{} + for rows.Next() { + var id int64 + var name string + if err := rows.Scan(&id, &name); err != nil { + return nil, err + } + out[id] = name + } + return out, rows.Err() +} + +func (s *Service) ensureReady(ctx context.Context) error { + if s == nil || s.walletDB == nil || s.userDB == nil { + return fmt.Errorf("host salary settlement dependencies are not configured") + } + // 老库没有 trigger_mode 时，查询和手动结算记录都会失败；这里补齐列，历史记录默认视为 automatic。 + if _, err := s.walletDB.ExecContext(ctx, ` + ALTER TABLE host_salary_settlement_records + ADD COLUMN trigger_mode VARCHAR(24) NOT NULL DEFAULT 'automatic' COMMENT '触发方式：automatic/manual' AFTER settlement_type`); err != nil && !strings.Contains(strings.ToLower(err.Error()), "duplicate column") { + return err + } + if _, err := s.walletDB.ExecContext(ctx, ` + ALTER TABLE host_salary_settlement_records + ADD COLUMN settlement_role VARCHAR(24) NOT NULL DEFAULT 'all' COMMENT '结算身份：all/host/agency' AFTER trigger_mode`); err != nil && !strings.Contains(strings.ToLower(err.Error()), "duplicate column") { + return err + } + return nil +} + func (s *Service) userProfiles(ctx context.Context, appCode string, userIDs []int64) (map[int64]userProfile, error) { if s == nil || s.userDB == nil || len(userIDs) == 0 { return map[int64]userProfile{}, nil @@ -204,6 +661,53 @@ func normalizeQuery(req query) query { return req } +func normalizePendingQuery(req pendingQuery) pendingQuery { + req.CycleKey = strings.TrimSpace(req.CycleKey) + req.Role = normalizeSettlementRole(req.Role) + if req.Role == "" { + req.Role = settlementRoleHost + } + req.SettlementType = normalizeSettlementType(req.SettlementType) + if req.SettlementType == "" { + req.SettlementType = settlementTypeHalfMonth + } + req.TriggerMode = normalizeTriggerMode(req.TriggerMode) + if req.TriggerMode == "" { + req.TriggerMode = triggerManual + } + req.CountryCode = strings.ToUpper(strings.TrimSpace(req.CountryCode)) + if req.CycleKey == "" { + req.CycleKey = currentMonthCycleKey(time.Now().UTC()) + } + if req.Page <= 0 { + req.Page = 1 + } + if req.PageSize <= 0 { + req.PageSize = 50 + } + if req.PageSize > maxPageSize { + req.PageSize = maxPageSize + } + return req +} + +func normalizeSettleRequest(req settleRequest) settleRequest { + req.CycleKey = strings.TrimSpace(req.CycleKey) + req.SettlementType = normalizeSettlementType(req.SettlementType) + if req.SettlementType == "" { + req.SettlementType = settlementTypeHalfMonth + } + req.TriggerMode = normalizeTriggerMode(req.TriggerMode) + if req.TriggerMode == "" { + req.TriggerMode = triggerManual + } + req.CountryCode = strings.ToUpper(strings.TrimSpace(req.CountryCode)) + if req.CycleKey == "" { + req.CycleKey = currentMonthCycleKey(time.Now().UTC()) + } + return req +} + func normalizeSettlementType(value string) string { switch strings.ToLower(strings.TrimSpace(value)) { case "", "all": @@ -219,6 +723,32 @@ func normalizeSettlementType(value string) string { } } +func normalizeTriggerMode(value string) string { + switch strings.ToLower(strings.TrimSpace(value)) { + case "", "all": + return "" + case triggerAutomatic: + return triggerAutomatic + case triggerManual: + return triggerManual + default: + return "" + } +} + +func normalizeSettlementRole(value string) string { + switch strings.ToLower(strings.TrimSpace(value)) { + case "", settlementRoleAll: + return "" + case settlementRoleHost: + return settlementRoleHost + case settlementRoleAgency: + return settlementRoleAgency + default: + return "" + } +} + func normalizeStatus(value string) string { switch strings.ToLower(strings.TrimSpace(value)) { case "", "all": @@ -282,6 +812,173 @@ func formatUSDMinor(value int64) string { return sign + strconv.FormatInt(whole, 10) + "." + strconv.FormatInt(fraction, 10) } +func pendingDTOFromCandidate(item hostCandidate, host userProfile, agency userProfile) pendingDTO { + userID := strconv.FormatInt(item.UserID, 10) + agencyID := formatOptionalUserID(item.AgencyOwnerUserID) + hostSalaryUSDDelta := item.HostSalaryUSDDelta + hostCoinRewardDelta := item.HostCoinRewardDelta + agencySalaryUSDDelta := item.AgencySalaryUSDDelta + residualUSDDelta := item.ResidualUSDDelta + switch item.Role { + case settlementRoleAgency: + hostSalaryUSDDelta = 0 + hostCoinRewardDelta = 0 + residualUSDDelta = 0 + case settlementRoleHost: + agencySalaryUSDDelta = 0 + } + totalUSDDelta, _ := checkedAdd(hostSalaryUSDDelta, agencySalaryUSDDelta, residualUSDDelta) + return pendingDTO{ + UserID: userID, + User: userDTOFromProfile(userID, host), + AgencyOwnerUserID: agencyID, + AgencyOwner: userDTOFromProfile(agencyID, agency), + Role: item.Role, + CycleKey: item.CycleKey, + RegionID: item.RegionID, + RegionName: item.RegionName, + PolicyID: item.Policy.PolicyID, + PolicyName: item.Policy.Name, + SettlementType: item.SettlementType, + TriggerMode: item.Policy.TriggerMode, + LevelNo: int32(item.Level.LevelNo), + RequiredDiamonds: item.Level.RequiredDiamonds, + TotalDiamonds: item.TotalDiamonds, + HostSalaryUSDMinorTarget: item.Level.HostSalaryUSDMinor, + HostSalaryUSDTarget: formatUSDMinor(item.Level.HostSalaryUSDMinor), + HostSalaryUSDMinorDelta: hostSalaryUSDDelta, + HostSalaryUSDDelta: formatUSDMinor(hostSalaryUSDDelta), + HostCoinRewardTarget: item.Level.HostCoinReward, + HostCoinRewardDelta: hostCoinRewardDelta, + AgencySalaryUSDMinorTarget: item.Level.AgencySalaryUSDMinor, + AgencySalaryUSDTarget: formatUSDMinor(item.Level.AgencySalaryUSDMinor), + AgencySalaryUSDMinorDelta: agencySalaryUSDDelta, + AgencySalaryUSDDelta: formatUSDMinor(agencySalaryUSDDelta), + ResidualUSDMinorDelta: residualUSDDelta, + ResidualUSDDelta: formatUSDMinor(residualUSDDelta), + TotalUSDMinorDelta: totalUSDDelta, + TotalUSDDelta: formatUSDMinor(totalUSDDelta), + SettledLevelNo: int32(item.Progress.SettledLevelNo), + SettledHostSalaryUSDMinor: item.Progress.SettledHostSalaryUSDMinor, + SettledHostSalaryUSD: formatUSDMinor(item.Progress.SettledHostSalaryUSDMinor), + SettledHostCoinReward: item.Progress.SettledHostCoinReward, + SettledAgencySalaryUSDMinor: item.Progress.SettledAgencySalaryUSDMinor, + SettledAgencySalaryUSD: formatUSDMinor(item.Progress.SettledAgencySalaryUSDMinor), + LastSettledTotalDiamonds: item.Progress.LastSettledTotalDiamonds, + ResidualUSDMinor: item.Progress.ResidualUSDMinor, + ResidualUSD: formatUSDMinor(item.Progress.ResidualUSDMinor), + MonthEndClearedAtMS: item.Progress.MonthEndClearedAtMS, + Status: statusPending, + } +} + +func hostCandidateHasPendingRole(item hostCandidate, role string) bool { + switch role { + case settlementRoleAgency: + return item.AgencySalaryUSDDelta > 0 + case settlementRoleHost: + return item.HostSalaryUSDDelta > 0 || item.HostCoinRewardDelta > 0 || item.ResidualUSDDelta > 0 + default: + return item.HostSalaryUSDDelta > 0 || item.HostCoinRewardDelta > 0 || item.AgencySalaryUSDDelta > 0 || item.ResidualUSDDelta > 0 + } +} + +func sortHostCandidates(items []hostCandidate) { + sort.SliceStable(items, func(i, j int) bool { + if items[i].CycleKey != items[j].CycleKey { + return items[i].CycleKey > items[j].CycleKey + } + if items[i].RegionID != items[j].RegionID { + return items[i].RegionID < items[j].RegionID + } + return items[i].UserID < items[j].UserID + }) +} + +func paginateHostCandidates(items []hostCandidate, page int, pageSize int) []hostCandidate { + start := offset(page, pageSize) + if start >= len(items) { + return []hostCandidate{} + } + end := start + pageSize + if end > len(items) { + end = len(items) + } + return items[start:end] +} + +func keysInt64FromSet(values map[int64]struct{}) []int64 { + out := make([]int64, 0, len(values)) + for id := range values { + out = append(out, id) + } + sort.Slice(out, func(i, j int) bool { return out[i] < out[j] }) + return out +} + +func placeholders(count int) string { + if count <= 0 { + return "" + } + return strings.TrimRight(strings.Repeat("?,", count), ",") +} + +func highestHostLevel(levels []hostPolicyLevel, totalDiamonds int64) hostPolicyLevel { + var selected hostPolicyLevel + for _, level := range levels { + if level.RequiredDiamonds <= totalDiamonds { + selected = level + } + } + return selected +} + +func residualDiamondRemainder(totalDiamonds int64, level hostPolicyLevel) int64 { + if level.LevelNo <= 0 || totalDiamonds <= level.RequiredDiamonds { + return 0 + } + return totalDiamonds - level.RequiredDiamonds +} + +func usdMinorFromDiamondRate(diamonds int64, rate string) (int64, error) { + if diamonds <= 0 || strings.TrimSpace(rate) == "" { + return 0, nil + } + rat, ok := new(big.Rat).SetString(strings.TrimSpace(rate)) + if !ok || rat.Sign() < 0 { + return 0, fmt.Errorf("residual diamond rate is invalid") + } + value := new(big.Rat).Mul(new(big.Rat).SetInt64(diamonds), rat) + value.Mul(value, big.NewRat(100, 1)) + result := new(big.Int).Quo(value.Num(), value.Denom()) + if !result.IsInt64() || result.Int64() > math.MaxInt64 { + return 0, fmt.Errorf("residual diamond amount overflow") + } + return result.Int64(), nil +} + +func positiveDelta(target int64, current int64) int64 { + if target <= current { + return 0 + } + return target - current +} + +func checkedAdd(values ...int64) (int64, error) { + var total int64 + for _, value := range values { + if value > 0 && total > math.MaxInt64-value { + return 0, fmt.Errorf("amount overflow") + } + total += value + } + return total, nil +} + +func currentMonthCycleKey(now time.Time) string { + return now.UTC().Format("2006-01") +} + func offset(page int, pageSize int) int { if page <= 1 { return 0 diff --git a/services/gateway-service/internal/transport/http/my_page_overview_handler_test.go b/services/gateway-service/internal/transport/http/my_page_overview_handler_test.go index bc7d8c12..1f3f58c1 100644 --- a/services/gateway-service/internal/transport/http/my_page_overview_handler_test.go +++ b/services/gateway-service/internal/transport/http/my_page_overview_handler_test.go @@ -109,13 +109,16 @@ func TestGetMyOverviewComposesProfileWalletVIPStatsRolesAndEntries(t *testing.T) !overviewEntryVisible(entries, "agency_center") || !overviewEntryVisible(entries, "manager_center") || !overviewEntryVisible(entries, "bd_center") || - !overviewEntryVisible(entries, "bd_leader_center") || + !overviewEntryVisible(entries, "superadmin_center") || !overviewEntryVisible(entries, "coin_seller_center") || !overviewEntryVisible(entries, "settings") || !overviewEntryVisible(entries, "language") || overviewEntryVisible(entries, "apply_host") { t.Fatalf("entry visibility mismatch: %+v", entries) } + if route := overviewEntryRoute(entries, "superadmin_center"); route != "/superadmin-center" { + t.Fatalf("superadmin center route mismatch: got %q entries=%+v", route, entries) + } if data["server_time_ms"].(float64) <= 0 { t.Fatalf("missing server_time_ms: %+v", data) } @@ -225,3 +228,14 @@ func overviewEntryVisible(entries []any, key string) bool { } return false } + +func overviewEntryRoute(entries []any, key string) string { + for _, entry := range entries { + item, ok := entry.(map[string]any) + if ok && item["key"] == key { + route, _ := item["route"].(string) + return route + } + } + return "" +} diff --git a/services/gateway-service/internal/transport/http/userapi/my_page_overview_handler.go b/services/gateway-service/internal/transport/http/userapi/my_page_overview_handler.go index 1b2d80f3..eaec7270 100644 --- a/services/gateway-service/internal/transport/http/userapi/my_page_overview_handler.go +++ b/services/gateway-service/internal/transport/http/userapi/my_page_overview_handler.go @@ -249,7 +249,8 @@ func overviewEntries(roles overviewRolesData) []overviewEntryData { {Key: "agency_center", Visible: roles.IsAgency, Enabled: roles.IsAgency, Route: "/agency", IconKey: "agency", Order: 70}, {Key: "manager_center", Visible: roles.IsManager, Enabled: roles.IsManager, Route: "/manager-center", IconKey: "manager_center", Order: 80}, {Key: "bd_center", Visible: roles.IsBD, Enabled: roles.IsBD, Route: "/bd", IconKey: "bd", Order: 90}, - {Key: "bd_leader_center", Visible: roles.IsBDLeader, Enabled: roles.IsBDLeader, Route: "/bd-leader", IconKey: "bd_leader", Order: 100}, + // BD Leader 和超管在 H5 里共用 superadmin-center 页面；入口 key 也统一成 superadmin_center，避免客户端再出现两个中心口径。 + {Key: "superadmin_center", Visible: roles.IsBDLeader, Enabled: roles.IsBDLeader, Route: "/superadmin-center", IconKey: "bd_leader", Order: 100}, {Key: "coin_seller_center", Visible: roles.IsCoinSeller, Enabled: roles.IsCoinSeller, Route: "/coin-seller", IconKey: "coin_seller", Order: 110}, {Key: "settings", Visible: true, Enabled: true, Route: "/settings", IconKey: "settings", Order: 120}, {Key: "language", Visible: true, Enabled: true, Route: "/language", IconKey: "language", Order: 130}, diff --git a/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql b/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql index 2b42a0a5..038c62b5 100644 --- a/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql +++ b/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql @@ -190,6 +190,8 @@ CREATE TABLE IF NOT EXISTS host_salary_settlement_records ( command_id VARCHAR(128) NOT NULL COMMENT '钱包交易幂等命令 ID', transaction_id VARCHAR(96) NOT NULL COMMENT '钱包交易 ID', settlement_type VARCHAR(24) NOT NULL COMMENT '结算类型：daily/half_month/month_end', + trigger_mode VARCHAR(24) NOT NULL DEFAULT 'automatic' COMMENT '触发方式：automatic/manual', + settlement_role VARCHAR(24) NOT NULL DEFAULT 'all' COMMENT '结算身份：all/host/agency', user_id BIGINT NOT NULL COMMENT '主播用户 ID', agency_owner_user_id BIGINT NOT NULL DEFAULT 0 COMMENT '代理收款用户 ID', cycle_key VARCHAR(16) NOT NULL COMMENT '工资周期键，UTC yyyy-MM', @@ -287,6 +289,24 @@ PREPARE stmt FROM @ddl; EXECUTE stmt; DEALLOCATE PREPARE stmt; +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_salary_settlement_records' AND COLUMN_NAME = 'trigger_mode') = 0, + 'ALTER TABLE host_salary_settlement_records ADD COLUMN trigger_mode VARCHAR(24) NOT NULL DEFAULT ''automatic'' COMMENT ''触发方式：automatic/manual'' AFTER settlement_type', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_salary_settlement_records' AND COLUMN_NAME = 'settlement_role') = 0, + 'ALTER TABLE host_salary_settlement_records ADD COLUMN settlement_role VARCHAR(24) NOT NULL DEFAULT ''all'' COMMENT ''结算身份：all/host/agency'' AFTER trigger_mode', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + -- 老环境可能已经存在 wallet_outbox，CREATE TABLE IF NOT EXISTS 不会补齐后续新增的投递锁字段。 -- 这里用 information_schema 生成幂等 ALTER，保证重复执行 initdb 时不会因重复列名失败。 SET @ddl := IF( diff --git a/services/wallet-service/internal/domain/ledger/ledger.go b/services/wallet-service/internal/domain/ledger/ledger.go index 50082010..9709bcf5 100644 --- a/services/wallet-service/internal/domain/ledger/ledger.go +++ b/services/wallet-service/internal/domain/ledger/ledger.go @@ -222,10 +222,14 @@ type HostSalarySettlementBatchCommand struct { WorkerID string BatchSize int SettlementType string - // TriggerMode 决定本次批处理读取自动政策还是手动政策；空值按 automatic 处理，兼容已有定时任务。 + // TriggerMode 只决定本次入账动作读取 automatic 还是 manual 政策；待结算账单展示可以同时读取两类政策。 TriggerMode string - CycleKey string - NowMs int64 + // SettlementRole 为空或 all 时完整结算主播和代理权益；后台人工入口传 host/agency 时只推进对应身份钱包。 + SettlementRole string + CycleKey string + // UserIDs 为空时扫描全量候选；后台人工批量结算会传入勾选主播，防止跨区域或跨周期误结算。 + UserIDs []int64 + NowMs int64 } // HostSalarySettlementBatchResult 汇总一个结算批次的扫描和入账结果。 diff --git a/services/wallet-service/internal/service/wallet/service_test.go b/services/wallet-service/internal/service/wallet/service_test.go index e7135743..4fe21480 100644 --- a/services/wallet-service/internal/service/wallet/service_test.go +++ b/services/wallet-service/internal/service/wallet/service_test.go @@ -633,6 +633,97 @@ func TestHostSalaryManualPolicyRequiresManualTrigger(t *testing.T) { assertBalance(t, svc, 30001, ledger.AssetAgencySalaryUSD, 50) } +// TestHostSalaryManualSettlementRoleSplitsHostAndAgencyWallets 验证后台人工结算按身份拆分钱包，Agency 行不会顺手发 Host 工资。 +func TestHostSalaryManualSettlementRoleSplitsHostAndAgencyWallets(t *testing.T) { + repository := mysqltest.NewRepository(t) + repository.SetBalance(10001, 1000) + repository.SetGiftPrice("rose", "salary-role-split", 120, 10, 10) + repository.SetHostSalaryPolicy(ledger.HostSalaryPolicy{ + PolicyID: 303, + Name: "manual role split salary", + RegionID: 8801, + Status: "active", + SettlementMode: ledger.HostSalarySettlementModeDaily, + SettlementTriggerMode: ledger.HostSalarySettlementTriggerManual, + Levels: []ledger.HostSalaryPolicyLevel{ + {LevelNo: 1, RequiredDiamonds: 100, HostSalaryUSDMinor: 150, HostCoinReward: 90, AgencySalaryUSDMinor: 50, SortOrder: 1}, + }, + }) + svc := walletservice.New(repository) + + receipt, err := svc.DebitGift(context.Background(), ledger.DebitGiftCommand{ + CommandID: "cmd-role-split-gift", + RoomID: "room-1", + SenderUserID: 10001, + TargetUserID: 10002, + GiftID: "rose", + GiftCount: 1, + PriceVersion: "salary-role-split", + TargetIsHost: true, + TargetHostRegionID: 8801, + TargetAgencyOwnerUserID: 30001, + }) + if err != nil { + t.Fatalf("role split gift failed: %v", err) + } + + agencyResult, err := svc.ProcessHostSalarySettlementBatch(context.Background(), ledger.HostSalarySettlementBatchCommand{ + AppCode: "lalu", + RunID: "role-split-agency", + WorkerID: "salary-test", + BatchSize: 10, + SettlementType: ledger.HostSalarySettlementModeDaily, + TriggerMode: ledger.HostSalarySettlementTriggerManual, + SettlementRole: "agency", + CycleKey: receipt.HostPeriodCycleKey, + UserIDs: []int64{10002}, + NowMs: time.Now().UnixMilli(), + }) + if err != nil { + t.Fatalf("agency-only settlement failed: %v", err) + } + if agencyResult.ProcessedCount != 1 || agencyResult.SuccessCount != 1 || agencyResult.FailureCount != 0 { + t.Fatalf("agency-only result mismatch: %+v", agencyResult) + } + assertBalance(t, svc, 30001, ledger.AssetAgencySalaryUSD, 50) + assertBalance(t, svc, 10002, ledger.AssetHostSalaryUSD, 0) + assertBalance(t, svc, 10002, ledger.AssetCoin, 0) + if got := repository.CountRows("host_salary_settlement_records", "user_id = ? AND cycle_key = ? AND settlement_role = ? AND agency_salary_usd_minor_delta = ? AND host_salary_usd_minor_delta = 0", int64(10002), receipt.HostPeriodCycleKey, "agency", int64(50)); got != 1 { + t.Fatalf("agency-only record mismatch, got %d", got) + } + if got := repository.CountRows("host_salary_settlement_progress", "user_id = ? AND cycle_key = ? AND settled_agency_salary_usd_minor = ? AND settled_host_salary_usd_minor = 0 AND last_settled_total_diamonds = 0", int64(10002), receipt.HostPeriodCycleKey, int64(50)); got != 1 { + t.Fatalf("agency-only progress should not close host side, got %d", got) + } + + hostResult, err := svc.ProcessHostSalarySettlementBatch(context.Background(), ledger.HostSalarySettlementBatchCommand{ + AppCode: "lalu", + RunID: "role-split-host", + WorkerID: "salary-test", + BatchSize: 10, + SettlementType: ledger.HostSalarySettlementModeDaily, + TriggerMode: ledger.HostSalarySettlementTriggerManual, + SettlementRole: "host", + CycleKey: receipt.HostPeriodCycleKey, + UserIDs: []int64{10002}, + NowMs: time.Now().UnixMilli(), + }) + if err != nil { + t.Fatalf("host-only settlement failed: %v", err) + } + if hostResult.ProcessedCount != 1 || hostResult.SuccessCount != 1 || hostResult.FailureCount != 0 { + t.Fatalf("host-only result mismatch: %+v", hostResult) + } + assertBalance(t, svc, 10002, ledger.AssetHostSalaryUSD, 150) + assertBalance(t, svc, 10002, ledger.AssetCoin, 90) + assertBalance(t, svc, 30001, ledger.AssetAgencySalaryUSD, 50) + if got := repository.CountRows("host_salary_settlement_records", "user_id = ? AND cycle_key = ? AND settlement_role = ? AND host_salary_usd_minor_delta = ? AND agency_salary_usd_minor_delta = 0", int64(10002), receipt.HostPeriodCycleKey, "host", int64(150)); got != 1 { + t.Fatalf("host-only record mismatch, got %d", got) + } + if got := repository.CountRows("host_salary_settlement_progress", "user_id = ? AND cycle_key = ? AND settled_host_salary_usd_minor = ? AND settled_host_coin_reward = ? AND settled_agency_salary_usd_minor = ? AND last_settled_total_diamonds = ?", int64(10002), receipt.HostPeriodCycleKey, int64(150), int64(90), int64(50), receipt.HostPeriodDiamondAdded); got != 1 { + t.Fatalf("split settlement should close progress after both roles are paid, got %d", got) + } +} + // TestHostSalaryMonthEndCreditsResidualAndClearsCycle 验证月底结算补发剩余钻石折美元，并逻辑关闭当月周期。 func TestHostSalaryMonthEndCreditsResidualAndClearsCycle(t *testing.T) { repository := mysqltest.NewRepository(t) diff --git a/services/wallet-service/internal/storage/mysql/host_salary_settlement.go b/services/wallet-service/internal/storage/mysql/host_salary_settlement.go index 2aa94320..f954f6b9 100644 --- a/services/wallet-service/internal/storage/mysql/host_salary_settlement.go +++ b/services/wallet-service/internal/storage/mysql/host_salary_settlement.go @@ -41,6 +41,8 @@ type hostSalarySettlementMetadata struct { // metadata 是审计快照：钱包交易、outbox、结算记录共享同一份口径，方便后续从任一链路追账。 AppCode string `json:"app_code"` SettlementType string `json:"settlement_type"` + TriggerMode string `json:"trigger_mode"` + SettlementRole string `json:"settlement_role"` HostUserID int64 `json:"host_user_id"` AgencyOwnerUserID int64 `json:"agency_owner_user_id"` CycleKey string `json:"cycle_key"` @@ -89,6 +91,10 @@ func (r *Repository) ProcessHostSalarySettlementBatch(ctx context.Context, comma if command.TriggerMode == "" { return ledger.HostSalarySettlementBatchResult{}, xerr.New(xerr.InvalidArgument, "trigger_mode is invalid") } + command.SettlementRole = normalizeHostSalarySettlementRole(command.SettlementRole) + if command.SettlementRole == "" { + return ledger.HostSalarySettlementBatchResult{}, xerr.New(xerr.InvalidArgument, "settlement_role is invalid") + } if command.NowMs <= 0 { return ledger.HostSalarySettlementBatchResult{}, xerr.New(xerr.InvalidArgument, "now_ms is required") } @@ -129,12 +135,24 @@ func (r *Repository) ProcessHostSalarySettlementBatch(ctx context.Context, comma func (r *Repository) listHostSalaryCandidates(ctx context.Context, command ledger.HostSalarySettlementBatchCommand, limit int) ([]hostSalaryCandidate, error) { policyModeClause := "" - args := []any{appcode.FromContext(ctx), command.CycleKey, command.NowMs, command.NowMs, command.TriggerMode} + userFilterClause := "" + args := []any{appcode.FromContext(ctx), command.CycleKey} + var policyModeArgs []any if command.SettlementType != ledger.HostSalarySettlementTypeMonthEnd { // 日结和半月结必须精确匹配政策配置的结算模式；月底清算复用有效政策处理剩余钻石和周期关闭。 policyModeClause = "AND p.settlement_mode = ?" - args = append(args, command.SettlementType) + policyModeArgs = append(policyModeArgs, command.SettlementType) } + userIDs := uniquePositiveInt64(command.UserIDs) + if len(userIDs) > 0 { + // 后台手动结算只允许处理运营勾选的主播；过滤下推到候选 SQL，避免锁定无关周期账户。 + userFilterClause = "AND a.user_id IN (" + placeholders(len(userIDs)) + ")" + for _, id := range userIDs { + args = append(args, id) + } + } + args = append(args, command.NowMs, command.NowMs, command.TriggerMode) + args = append(args, policyModeArgs...) // 日结/半月结只扫描“本周期钻石比上次结算更多”的主播，避免 cron 重跑时反复锁定无新增收益的账户。 // 月底批次必须扫描全部未关闭周期账户，因为它还负责剩余钻石折美元和周期逻辑关闭。 progressClause := "AND a.total_diamonds > COALESCE(progress.last_settled_total_diamonds, 0)" @@ -152,6 +170,7 @@ func (r *Repository) listHostSalaryCandidates(ctx context.Context, command ledge AND a.total_diamonds > 0 AND COALESCE(progress.month_end_cleared_at_ms, 0) = 0 %s + %s AND EXISTS ( SELECT 1 FROM host_agency_salary_policies p @@ -164,7 +183,7 @@ func (r *Repository) listHostSalaryCandidates(ctx context.Context, command ledge %s ) ORDER BY a.updated_at_ms ASC, a.user_id ASC - LIMIT ?`, progressClause, policyModeClause) + LIMIT ?`, progressClause, userFilterClause, policyModeClause) rows, err := r.db.QueryContext(ctx, query, args...) if err != nil { return nil, err @@ -211,6 +230,8 @@ func (r *Repository) processHostSalaryCandidate(ctx context.Context, command led if err != nil || !ok { return false, err } + shouldSettleHost := command.SettlementRole == "all" || command.SettlementRole == "host" + shouldSettleAgency := command.SettlementRole == "all" || command.SettlementRole == "agency" // 工资表是累计值，因此每次只发“目标累计值 - 已发累计值”；这对应产品规则里的 2 级只补 2 级减 1 级。 level := highestHostSalaryLevel(policy.Levels, account.TotalDiamonds) residualTarget := progress.ResidualUSDMinor @@ -224,29 +245,49 @@ func (r *Repository) processHostSalaryCandidate(ctx context.Context, command led } residualDelta = positiveDelta(residualTarget, progress.ResidualUSDMinor) } + if !shouldSettleHost { + // Agency 人工结算只处理代理工资钱包；月底剩余钻石折美元仍属于 Host 工资钱包，不能在 Agency 行触发。 + residualDelta = 0 + } - hostSalaryDelta := positiveDelta(level.HostSalaryUSDMinor, progress.SettledHostSalaryUSDMinor) - hostCoinDelta := positiveDelta(level.HostCoinReward, progress.SettledHostCoinReward) - agencyEntitlementDelta := positiveDelta(level.AgencySalaryUSDMinor, progress.SettledAgencySalaryUSDMinor) + hostSalaryDelta := int64(0) + hostCoinDelta := int64(0) + if shouldSettleHost { + hostSalaryDelta = positiveDelta(level.HostSalaryUSDMinor, progress.SettledHostSalaryUSDMinor) + hostCoinDelta = positiveDelta(level.HostCoinReward, progress.SettledHostCoinReward) + } + agencyEntitlementDelta := int64(0) + if shouldSettleAgency { + agencyEntitlementDelta = positiveDelta(level.AgencySalaryUSDMinor, progress.SettledAgencySalaryUSDMinor) + } agencyPaidDelta := agencyEntitlementDelta if account.AgencyOwnerUserID <= 0 { // 没有代理收款人时只把该等级代理权益标记为已处理，不在未来组织绑定后补发历史月份工资。 agencyPaidDelta = 0 } - nextProgress := hostSalaryProgress{ - SettledLevelNo: maxInt(progress.SettledLevelNo, level.LevelNo), - SettledHostSalaryUSDMinor: maxInt64(progress.SettledHostSalaryUSDMinor, level.HostSalaryUSDMinor), - SettledHostCoinReward: maxInt64(progress.SettledHostCoinReward, level.HostCoinReward), - SettledAgencySalaryUSDMinor: maxInt64(progress.SettledAgencySalaryUSDMinor, level.AgencySalaryUSDMinor), - LastSettledTotalDiamonds: account.TotalDiamonds, - ResidualUSDMinor: maxInt64(progress.ResidualUSDMinor, residualTarget), - MonthEndClearedAtMS: progress.MonthEndClearedAtMS, - LastPolicyID: policy.PolicyID, - Version: progress.Version, + nextProgress := progress + nextProgress.LastPolicyID = policy.PolicyID + if shouldSettleHost && (hostSalaryDelta > 0 || hostCoinDelta > 0 || residualDelta > 0) { + nextProgress.SettledLevelNo = maxInt(nextProgress.SettledLevelNo, level.LevelNo) + nextProgress.SettledHostSalaryUSDMinor = maxInt64(nextProgress.SettledHostSalaryUSDMinor, level.HostSalaryUSDMinor) + nextProgress.SettledHostCoinReward = maxInt64(nextProgress.SettledHostCoinReward, level.HostCoinReward) + nextProgress.ResidualUSDMinor = maxInt64(nextProgress.ResidualUSDMinor, residualTarget) } - if command.SettlementType == ledger.HostSalarySettlementTypeMonthEnd { - nextProgress.MonthEndClearedAtMS = command.NowMs + if shouldSettleAgency && agencyEntitlementDelta > 0 { + nextProgress.SettledLevelNo = maxInt(nextProgress.SettledLevelNo, level.LevelNo) + nextProgress.SettledAgencySalaryUSDMinor = maxInt64(nextProgress.SettledAgencySalaryUSDMinor, level.AgencySalaryUSDMinor) + } + if account.AgencyOwnerUserID <= 0 { + // 当前周期没有代理收款人时，代理权益视为已处理，避免后续绑定代理后追发旧周期代理工资。 + nextProgress.SettledAgencySalaryUSDMinor = maxInt64(nextProgress.SettledAgencySalaryUSDMinor, level.AgencySalaryUSDMinor) + } + if hostSalaryProgressCoversLevel(nextProgress, level, account, command.SettlementType, residualTarget) { + nextProgress.SettledLevelNo = maxInt(nextProgress.SettledLevelNo, level.LevelNo) + nextProgress.LastSettledTotalDiamonds = account.TotalDiamonds + if command.SettlementType == ledger.HostSalarySettlementTypeMonthEnd { + nextProgress.MonthEndClearedAtMS = command.NowMs + } } paidHostUSDDelta, err := checkedAdd(hostSalaryDelta, residualDelta) @@ -256,6 +297,8 @@ func (r *Repository) processHostSalaryCandidate(ctx context.Context, command led metadata := hostSalarySettlementMetadata{ AppCode: appcode.FromContext(ctx), SettlementType: command.SettlementType, + TriggerMode: command.TriggerMode, + SettlementRole: command.SettlementRole, HostUserID: account.UserID, AgencyOwnerUserID: account.AgencyOwnerUserID, CycleKey: account.CycleKey, @@ -282,7 +325,7 @@ func (r *Repository) processHostSalaryCandidate(ctx context.Context, command led if paidHostUSDDelta > 0 || hostCoinDelta > 0 || agencyPaidDelta > 0 { // settlement commandID 带上已结算等级和当前钻石，确保同一增量重复执行会命中交易幂等键。 - commandID := hostSalarySettlementCommandID(command, account, nextProgress) + commandID := hostSalarySettlementCommandID(command, account, policy.PolicyID, nextProgress) transactionID := transactionID(appcode.FromContext(ctx), commandID) if err := r.insertTransaction(ctx, tx, transactionID, commandID, bizTypeHostSalarySettlement, hostSalarySettlementRequestHash(metadata), fmt.Sprintf("%s:%d", account.CycleKey, account.UserID), metadata, command.NowMs); err != nil { return false, err @@ -520,16 +563,18 @@ func (r *Repository) insertHostSalarySettlementRecord(ctx context.Context, tx *s // 结算记录保存每次实际发放的增量；BD 结算后续从这里汇总主播工资和剩余钻石折美元。 _, err := tx.ExecContext(ctx, ` INSERT INTO host_salary_settlement_records ( - app_code, settlement_id, command_id, transaction_id, settlement_type, user_id, + app_code, settlement_id, command_id, transaction_id, settlement_type, trigger_mode, settlement_role, user_id, agency_owner_user_id, cycle_key, policy_id, level_no, total_diamonds, host_salary_usd_minor_delta, host_coin_reward_delta, agency_salary_usd_minor_delta, residual_usd_minor_delta, status, reason, created_at_ms - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'succeeded', '', ?)`, + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'succeeded', '', ?)`, appcode.FromContext(ctx), hostSalarySettlementID(metadata.AppCode, commandID), commandID, transactionID, metadata.SettlementType, + metadata.TriggerMode, + metadata.SettlementRole, metadata.HostUserID, metadata.AgencyOwnerUserID, metadata.CycleKey, @@ -649,9 +694,39 @@ func normalizeHostSalarySettlementTriggerMode(value string) string { } } -func hostSalarySettlementCommandID(command ledger.HostSalarySettlementBatchCommand, account hostSalaryCandidate, progress hostSalaryProgress) string { - // commandID 包含“已推进后的等级 + 当前钻石”，同一状态重放命中同一个交易，不同增量生成新交易。 - return fmt.Sprintf("host_salary:%s:%s:%d:%d:%d", command.SettlementType, account.CycleKey, account.UserID, progress.SettledLevelNo, account.TotalDiamonds) +func normalizeHostSalarySettlementRole(value string) string { + switch strings.ToLower(strings.TrimSpace(value)) { + case "", "all": + // 空值按完整结算处理，保证已有定时任务继续同时处理主播和代理权益。 + return "all" + case "host": + return "host" + case "agency": + return "agency" + default: + return "" + } +} + +func uniquePositiveInt64(values []int64) []int64 { + seen := map[int64]struct{}{} + out := make([]int64, 0, len(values)) + for _, value := range values { + if value <= 0 { + continue + } + if _, ok := seen[value]; ok { + continue + } + seen[value] = struct{}{} + out = append(out, value) + } + return out +} + +func hostSalarySettlementCommandID(command ledger.HostSalarySettlementBatchCommand, account hostSalaryCandidate, policyID uint64, progress hostSalaryProgress) string { + // commandID 包含触发方式和政策，避免 automatic/manual 共存或政策重发时把不同来源的结算误判为同一交易。 + return fmt.Sprintf("host_salary:%s:%s:%s:%s:%d:%d:%d:%d", command.SettlementType, command.TriggerMode, command.SettlementRole, account.CycleKey, account.UserID, policyID, progress.SettledLevelNo, account.TotalDiamonds) } func hostSalarySettlementID(appCode string, commandID string) string { @@ -661,9 +736,11 @@ func hostSalarySettlementID(appCode string, commandID string) string { func hostSalarySettlementRequestHash(metadata hostSalarySettlementMetadata) string { // 请求哈希覆盖金额、等级和周期口径；同 commandID 但 payload 改变会被交易层识别为冲突。 - return stableHash(fmt.Sprintf("host_salary|%s|%s|%d|%d|%s|%d|%d|%d|%d|%d|%d|%d", + return stableHash(fmt.Sprintf("host_salary|%s|%s|%s|%s|%d|%d|%s|%d|%d|%d|%d|%d|%d|%d", appcode.Normalize(metadata.AppCode), metadata.SettlementType, + metadata.TriggerMode, + metadata.SettlementRole, metadata.HostUserID, metadata.AgencyOwnerUserID, metadata.CycleKey, @@ -677,6 +754,13 @@ func hostSalarySettlementRequestHash(metadata hostSalarySettlementMetadata) stri )) } +func hostSalaryProgressCoversLevel(progress hostSalaryProgress, level ledger.HostSalaryPolicyLevel, account hostSalaryCandidate, settlementType string, residualTarget int64) bool { + hostCovered := progress.SettledHostSalaryUSDMinor >= level.HostSalaryUSDMinor && progress.SettledHostCoinReward >= level.HostCoinReward + agencyCovered := progress.SettledAgencySalaryUSDMinor >= level.AgencySalaryUSDMinor || account.AgencyOwnerUserID <= 0 + residualCovered := settlementType != ledger.HostSalarySettlementTypeMonthEnd || progress.ResidualUSDMinor >= residualTarget + return hostCovered && agencyCovered && residualCovered +} + func positiveDelta(target int64, current int64) int64 { if target <= current { return 0 diff --git a/services/wallet-service/internal/storage/mysql/repository.go b/services/wallet-service/internal/storage/mysql/repository.go index 8750c925..eade7e8f 100644 --- a/services/wallet-service/internal/storage/mysql/repository.go +++ b/services/wallet-service/internal/storage/mysql/repository.go @@ -115,10 +115,30 @@ func Open(ctx context.Context, dsn string) (*Repository, error) { _ = db.Close() return nil, err } + if err := ensureHostSalarySettlementSchema(ctx, db); err != nil { + _ = db.Close() + return nil, err + } return &Repository{db: db}, nil } +func ensureHostSalarySettlementSchema(ctx context.Context, db *sql.DB) error { + // Host/Agency 工资记录需要区分自动和手动来源；旧库没有该列时启动期补齐，避免手动结算上线后插入失败。 + if _, err := db.ExecContext(ctx, ` + ALTER TABLE host_salary_settlement_records + ADD COLUMN trigger_mode VARCHAR(24) NOT NULL DEFAULT 'automatic' COMMENT '触发方式：automatic/manual' AFTER settlement_type`); err != nil && !isDuplicateColumnError(err) { + return err + } + // 人工结算可以按 Host 或 Agency 身份拆分钱包；旧记录默认 all，表示一次完整 Host+Agency 结算。 + if _, err := db.ExecContext(ctx, ` + ALTER TABLE host_salary_settlement_records + ADD COLUMN settlement_role VARCHAR(24) NOT NULL DEFAULT 'all' COMMENT '结算身份：all/host/agency' AFTER trigger_mode`); err != nil && !isDuplicateColumnError(err) { + return err + } + return nil +} + func ensureExternalRechargeSchema(ctx context.Context, db *sql.DB) error { if _, err := db.ExecContext(ctx, ` ALTER TABLE wallet_recharge_products diff --git a/services/wallet-service/internal/transport/grpc/cron_server.go b/services/wallet-service/internal/transport/grpc/cron_server.go index 6137fda8..c9bacc4b 100644 --- a/services/wallet-service/internal/transport/grpc/cron_server.go +++ b/services/wallet-service/internal/transport/grpc/cron_server.go @@ -47,6 +47,10 @@ func (s *CronServer) processHostSalarySettlementBatch(ctx context.Context, req * WorkerID: req.GetWorkerId(), BatchSize: int(req.GetBatchSize()), SettlementType: settlementType, + TriggerMode: req.GetTriggerMode(), + SettlementRole: req.GetSettlementRole(), + CycleKey: req.GetCycleKey(), + UserIDs: append([]int64(nil), req.GetUserIds()...), }) if err != nil { return nil, xerr.ToGRPCError(err)