主播相关
This commit is contained in:
zhx
parent
09d5747cbd
commit
18efb4ed5d
@ -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" +
|
||||
|
||||
@ -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 的执行摘要。
|
||||
|
||||
@ -3008,7 +3008,7 @@ definitions:
|
||||
- host_center
|
||||
- agency_center
|
||||
- bd_center
|
||||
- bd_leader_center
|
||||
- superadmin_center
|
||||
- coin_seller_center
|
||||
- settings
|
||||
- language
|
||||
|
||||
@ -119,7 +119,7 @@ Authorization: Bearer <access_token>
|
||||
| `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 |
|
||||
|
||||
@ -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()
|
||||
}
|
||||
|
||||
|
||||
@ -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),
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -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 == "" {
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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)
|
||||
|
||||
557
server/admin/internal/modules/hostorg/salary_wallet.go
Normal file
557
server/admin/internal/modules/hostorg/salary_wallet.go
Normal file
@ -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
|
||||
}
|
||||
@ -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{
|
||||
|
||||
@ -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"`
|
||||
}
|
||||
|
||||
@ -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 == "" {
|
||||
|
||||
@ -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)
|
||||
}
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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 ""
|
||||
}
|
||||
|
||||
@ -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},
|
||||
|
||||
@ -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(
|
||||
|
||||
@ -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 汇总一个结算批次的扫描和入账结果。
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user