988 lines
32 KiB
Go
988 lines
32 KiB
Go
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
|
||
|
||
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
|
||
walletCron walletCronClient
|
||
requestTimeout time.Duration
|
||
}
|
||
|
||
type userProfile struct {
|
||
UserID int64
|
||
DisplayUserID string
|
||
Username string
|
||
Avatar string
|
||
}
|
||
|
||
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 err := s.ensureReady(ctx); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
req = normalizeQuery(req)
|
||
whereSQL, args := settlementWhere(appCode, req)
|
||
|
||
var total int64
|
||
if err := s.walletDB.QueryRowContext(ctx, `SELECT COUNT(*) FROM host_salary_settlement_records `+whereSQL, args...).Scan(&total); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
rows, err := s.walletDB.QueryContext(ctx, `
|
||
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
|
||
FROM host_salary_settlement_records
|
||
`+whereSQL+`
|
||
ORDER BY created_at_ms DESC, settlement_id DESC
|
||
LIMIT ? OFFSET ?`,
|
||
append(args, req.PageSize, offset(req.Page, req.PageSize))...,
|
||
)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
items := make([]settlementDTO, 0, req.PageSize)
|
||
userIDs := make([]int64, 0, req.PageSize*2)
|
||
for rows.Next() {
|
||
var item settlementDTO
|
||
var userID int64
|
||
var agencyOwnerUserID int64
|
||
if err := rows.Scan(
|
||
&item.SettlementID,
|
||
&item.CommandID,
|
||
&item.TransactionID,
|
||
&item.SettlementType,
|
||
&item.TriggerMode,
|
||
&item.SettlementRole,
|
||
&userID,
|
||
&agencyOwnerUserID,
|
||
&item.CycleKey,
|
||
&item.PolicyID,
|
||
&item.LevelNo,
|
||
&item.TotalDiamonds,
|
||
&item.HostSalaryUSDMinorDelta,
|
||
&item.HostCoinRewardDelta,
|
||
&item.AgencySalaryUSDMinorDelta,
|
||
&item.ResidualUSDMinorDelta,
|
||
&item.Status,
|
||
&item.Reason,
|
||
&item.CreatedAtMS,
|
||
); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
item.UserID = strconv.FormatInt(userID, 10)
|
||
item.AgencyOwnerUserID = formatOptionalUserID(agencyOwnerUserID)
|
||
item.User = userDTO{UserID: item.UserID}
|
||
item.AgencyOwner = userDTO{UserID: item.AgencyOwnerUserID}
|
||
item.HostSalaryUSDDelta = formatUSDMinor(item.HostSalaryUSDMinorDelta)
|
||
item.AgencySalaryUSDDelta = formatUSDMinor(item.AgencySalaryUSDMinorDelta)
|
||
item.ResidualUSDDelta = formatUSDMinor(item.ResidualUSDMinorDelta)
|
||
items = append(items, item)
|
||
if userID > 0 {
|
||
userIDs = append(userIDs, userID)
|
||
}
|
||
if agencyOwnerUserID > 0 {
|
||
userIDs = append(userIDs, agencyOwnerUserID)
|
||
}
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
|
||
profiles, err := s.userProfiles(ctx, appCode, userIDs)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
for i := range items {
|
||
if userID, err := strconv.ParseInt(items[i].UserID, 10, 64); err == nil {
|
||
items[i].User = userDTOFromProfile(items[i].UserID, profiles[userID])
|
||
}
|
||
if agencyID, err := strconv.ParseInt(items[i].AgencyOwnerUserID, 10, 64); err == nil && agencyID > 0 {
|
||
items[i].AgencyOwner = userDTOFromProfile(items[i].AgencyOwnerUserID, profiles[agencyID])
|
||
}
|
||
}
|
||
return items, total, nil
|
||
}
|
||
|
||
func settlementWhere(appCode string, req query) (string, []any) {
|
||
conditions := []string{"app_code = ?"}
|
||
args := []any{strings.TrimSpace(appCode)}
|
||
if req.CycleKey != "" {
|
||
conditions = append(conditions, "cycle_key = ?")
|
||
args = append(args, req.CycleKey)
|
||
}
|
||
if req.SettlementType != "" {
|
||
conditions = append(conditions, "settlement_type = ?")
|
||
args = append(args, req.SettlementType)
|
||
}
|
||
if req.Status != "" {
|
||
conditions = append(conditions, "status = ?")
|
||
args = append(args, req.Status)
|
||
}
|
||
if req.UserID > 0 {
|
||
conditions = append(conditions, "user_id = ?")
|
||
args = append(args, req.UserID)
|
||
}
|
||
if req.AgencyOwnerUserID > 0 {
|
||
conditions = append(conditions, "agency_owner_user_id = ?")
|
||
args = append(args, req.AgencyOwnerUserID)
|
||
}
|
||
if req.PolicyID > 0 {
|
||
conditions = append(conditions, "policy_id = ?")
|
||
args = append(args, req.PolicyID)
|
||
}
|
||
if req.StartAtMS > 0 {
|
||
conditions = append(conditions, "created_at_ms >= ?")
|
||
args = append(args, req.StartAtMS)
|
||
}
|
||
if req.EndAtMS > 0 {
|
||
conditions = append(conditions, "created_at_ms < ?")
|
||
args = append(args, req.EndAtMS)
|
||
}
|
||
// 查询只读 wallet-service 结算事实表，不回推钱包分录；账务对账需要单独走 entries/transactions 明细。
|
||
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
|
||
}
|
||
uniqueIDs := uniquePositiveUserIDs(userIDs)
|
||
if len(uniqueIDs) == 0 {
|
||
return map[int64]userProfile{}, nil
|
||
}
|
||
placeholders := strings.TrimRight(strings.Repeat("?,", len(uniqueIDs)), ",")
|
||
args := make([]any, 0, len(uniqueIDs)+1)
|
||
args = append(args, strings.TrimSpace(appCode))
|
||
for _, id := range uniqueIDs {
|
||
args = append(args, id)
|
||
}
|
||
rows, err := s.userDB.QueryContext(ctx, `
|
||
SELECT user_id, current_display_user_id, COALESCE(username, ''), COALESCE(avatar, '')
|
||
FROM users
|
||
WHERE app_code = ? AND user_id IN (`+placeholders+`)`,
|
||
args...,
|
||
)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
profiles := make(map[int64]userProfile, len(uniqueIDs))
|
||
for rows.Next() {
|
||
var item userProfile
|
||
if err := rows.Scan(&item.UserID, &item.DisplayUserID, &item.Username, &item.Avatar); err != nil {
|
||
return nil, err
|
||
}
|
||
profiles[item.UserID] = item
|
||
}
|
||
return profiles, rows.Err()
|
||
}
|
||
|
||
func normalizeQuery(req query) query {
|
||
req.CycleKey = strings.TrimSpace(req.CycleKey)
|
||
req.SettlementType = normalizeSettlementType(req.SettlementType)
|
||
req.Status = normalizeStatus(req.Status)
|
||
if req.Page <= 0 {
|
||
req.Page = 1
|
||
}
|
||
if req.PageSize <= 0 {
|
||
req.PageSize = 50
|
||
}
|
||
if req.PageSize > maxPageSize {
|
||
req.PageSize = maxPageSize
|
||
}
|
||
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":
|
||
return ""
|
||
case "daily", "day":
|
||
return "daily"
|
||
case "half_month", "half-month", "halfmonth":
|
||
return "half_month"
|
||
case "month_end", "month-end", "monthend":
|
||
return "month_end"
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
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":
|
||
return ""
|
||
case "succeeded", "skipped", "failed":
|
||
return strings.ToLower(strings.TrimSpace(value))
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
func uniquePositiveUserIDs(ids []int64) []int64 {
|
||
seen := map[int64]struct{}{}
|
||
out := make([]int64, 0, len(ids))
|
||
for _, id := range ids {
|
||
if id <= 0 {
|
||
continue
|
||
}
|
||
if _, ok := seen[id]; ok {
|
||
continue
|
||
}
|
||
seen[id] = struct{}{}
|
||
out = append(out, id)
|
||
}
|
||
return out
|
||
}
|
||
|
||
func userDTOFromProfile(fallbackID string, profile userProfile) userDTO {
|
||
if profile.UserID <= 0 {
|
||
return userDTO{UserID: fallbackID}
|
||
}
|
||
return userDTO{
|
||
UserID: strconv.FormatInt(profile.UserID, 10),
|
||
DisplayUserID: profile.DisplayUserID,
|
||
Username: profile.Username,
|
||
Avatar: profile.Avatar,
|
||
}
|
||
}
|
||
|
||
func formatOptionalUserID(value int64) string {
|
||
if value <= 0 {
|
||
return ""
|
||
}
|
||
return strconv.FormatInt(value, 10)
|
||
}
|
||
|
||
func formatUSDMinor(value int64) string {
|
||
sign := ""
|
||
if value < 0 {
|
||
sign = "-"
|
||
value = -value
|
||
}
|
||
whole := value / 100
|
||
fraction := value % 100
|
||
if fraction == 0 {
|
||
return sign + strconv.FormatInt(whole, 10)
|
||
}
|
||
if fraction < 10 {
|
||
return sign + strconv.FormatInt(whole, 10) + ".0" + strconv.FormatInt(fraction, 10)
|
||
}
|
||
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
|
||
}
|
||
return (page - 1) * pageSize
|
||
}
|