1523 lines
57 KiB
Go
1523 lines
57 KiB
Go
package teamsalarysettlement
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"database/sql"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"math"
|
||
"math/big"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"hyapp-admin-server/internal/modules/shared"
|
||
)
|
||
|
||
const (
|
||
// maxPageSize 限制后台列表单次拉取规模，避免待结算计算在页面误传大 page_size 时拖垮三个库的聚合查询。
|
||
maxPageSize = 100
|
||
policyTypeBD = "bd"
|
||
policyTypeAdmin = "admin"
|
||
triggerAutomatic = "automatic"
|
||
triggerManual = "manual"
|
||
// BD/Admin 工资钱包资产类型必须与 wallet-service ledger 常量保持一致；admin-server 是独立 module，不能直接复用根 module 常量。
|
||
assetBDSalaryUSD = "BD_SALARY_USD"
|
||
assetAdminSalaryUSD = "ADMIN_SALARY_USD"
|
||
bizTypeTeamSettlement = "team_salary_settlement"
|
||
statusSucceeded = "succeeded"
|
||
)
|
||
|
||
// autoSettlementResult 是自动任务内部的聚合结果；它不直接暴露给前端，只用于日志和测试断言。
|
||
type autoSettlementResult struct {
|
||
Due bool `json:"due"`
|
||
CycleKey string `json:"cycle_key"`
|
||
Results map[string]settleResultDTO `json:"results"`
|
||
}
|
||
|
||
// Service 聚合 admin/user/wallet 三个库：政策在 admin，组织关系在 user，结算进度与钱包流水在 wallet。
|
||
type Service struct {
|
||
adminDB *sql.DB
|
||
walletDB *sql.DB
|
||
userDB *sql.DB
|
||
}
|
||
|
||
// teamPolicy 是结算时使用的政策快照；工资按等级累计值发差值，因此必须携带完整等级表。
|
||
type teamPolicy struct {
|
||
PolicyID uint64
|
||
PolicyType string
|
||
Name string
|
||
RegionID int64
|
||
TriggerMode string
|
||
Levels []teamPolicyLevel
|
||
}
|
||
|
||
// teamPolicyLevel 中金额统一转为美分整数，避免 decimal 字符串在计算差值时出现浮点误差。
|
||
type teamPolicyLevel struct {
|
||
LevelNo int
|
||
ThresholdMinor int64
|
||
SalaryMinor int64
|
||
}
|
||
|
||
// teamProgress 是幂等核心：同一用户同一周期只记录“已发到哪个累计工资”，重复结算自然得到 0 差值。
|
||
type teamProgress struct {
|
||
SettledLevelNo int
|
||
SettledSalaryUSDMinor int64
|
||
LastSettledIncomeUSDMinor int64
|
||
LastPolicyID uint64
|
||
MonthClosedAtMS int64
|
||
Version int64
|
||
}
|
||
|
||
// teamCandidate 是待结算候选人，已经完成收入聚合、区域归属、政策匹配和差值计算。
|
||
type teamCandidate struct {
|
||
PolicyType string
|
||
UserID int64
|
||
CycleKey string
|
||
RegionID int64
|
||
RegionName string
|
||
Policy teamPolicy
|
||
Level teamPolicyLevel
|
||
IncomeUSDMinor int64
|
||
Progress teamProgress
|
||
SalaryUSDMinorDelta int64
|
||
}
|
||
|
||
// settlementMetadata 被同时写入钱包交易和 outbox，后续查账可以从任一钱包流水反查政策、周期、档位和操作者。
|
||
type settlementMetadata struct {
|
||
AppCode string `json:"app_code"`
|
||
PolicyType string `json:"policy_type"`
|
||
TriggerMode string `json:"trigger_mode"`
|
||
UserID int64 `json:"user_id"`
|
||
CycleKey string `json:"cycle_key"`
|
||
RegionID int64 `json:"region_id"`
|
||
PolicyID uint64 `json:"policy_id"`
|
||
PolicyName string `json:"policy_name"`
|
||
LevelNo int `json:"level_no"`
|
||
IncomeUSDMinor int64 `json:"income_usd_minor"`
|
||
SalaryUSDMinorDelta int64 `json:"salary_usd_minor_delta"`
|
||
SalaryAssetType string `json:"salary_asset_type"`
|
||
SettledLevelNo int `json:"settled_level_no"`
|
||
SettledSalaryUSDMinor int64 `json:"settled_salary_usd_minor"`
|
||
ProcessedAtMS int64 `json:"processed_at_ms"`
|
||
OperatorAdminID int64 `json:"operator_admin_id"`
|
||
}
|
||
|
||
// userProfile 只用于后台展示，结算主链路永远以 user_id 为准，避免昵称或短 ID 变更影响账务。
|
||
type userProfile struct {
|
||
UserID int64
|
||
DisplayUserID string
|
||
Username string
|
||
Avatar string
|
||
}
|
||
|
||
// NewService 注入三个数据库连接；调用方负责连接生命周期，Service 只做业务编排。
|
||
func NewService(adminDB *sql.DB, walletDB *sql.DB, userDB *sql.DB) *Service {
|
||
return &Service{adminDB: adminDB, walletDB: walletDB, userDB: userDB}
|
||
}
|
||
|
||
// ListPending 计算当前筛选条件下“现在结算会产生正向工资差值”的 BD/Admin 列表。
|
||
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)
|
||
var matched bool
|
||
var err error
|
||
req.UserID, matched, err = s.resolveFilterUserID(ctx, appCode, req.UserID, req.UserFilter)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
if !matched {
|
||
return []pendingDTO{}, 0, nil
|
||
}
|
||
// 待结算列表复用结算候选计算，保证运营看到的金额与真正点击结算时的金额口径一致。
|
||
candidates, err := s.collectCandidates(ctx, strings.TrimSpace(appCode), req.PolicyType, req.TriggerMode, req.CycleKey, req.RegionID, req.CountryID, req.CountryCode, req.UserID, time.Now().UTC().UnixMilli())
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
// 待结算列表只展示有正向差额的用户；未达档或已结清用户留在记录/进度表里，避免运营批量操作时出现 0 金额行。
|
||
pending := make([]teamCandidate, 0, len(candidates))
|
||
for _, candidate := range candidates {
|
||
if candidate.SalaryUSDMinorDelta > 0 {
|
||
pending = append(pending, candidate)
|
||
}
|
||
}
|
||
sortTeamCandidates(pending)
|
||
total := int64(len(pending))
|
||
pageItems := paginateCandidates(pending, req.Page, req.PageSize)
|
||
// 用户信息只按分页后的结果加载，避免大区域待结算列表一次性扫全量用户资料。
|
||
profiles, err := s.userProfiles(ctx, appCode, candidateUserIDs(pageItems))
|
||
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]))
|
||
}
|
||
return items, total, nil
|
||
}
|
||
|
||
// ListRecords 查询 BD/Admin 结算记录；国家筛选会先转成区域集合，再过滤 wallet 记录里的 region_id。
|
||
func (s *Service) ListRecords(ctx context.Context, appCode string, req recordQuery) ([]recordDTO, int64, error) {
|
||
if err := s.ensureReady(ctx); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
req = normalizeRecordQuery(req)
|
||
var matched bool
|
||
var err error
|
||
req.UserID, matched, err = s.resolveFilterUserID(ctx, appCode, req.UserID, req.UserFilter)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
if !matched {
|
||
return []recordDTO{}, 0, nil
|
||
}
|
||
whereSQL, args, err := s.recordWhere(ctx, strings.TrimSpace(appCode), req)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
var total int64
|
||
if err := s.walletDB.QueryRowContext(ctx, `SELECT COUNT(*) FROM team_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, policy_type, trigger_mode, user_id,
|
||
cycle_key, region_id, policy_id, level_no, income_usd_minor, salary_usd_minor_delta,
|
||
status, reason, created_at_ms
|
||
FROM team_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([]recordDTO, 0, req.PageSize)
|
||
userIDs := make([]int64, 0, req.PageSize)
|
||
for rows.Next() {
|
||
var item recordDTO
|
||
var userID int64
|
||
if err := rows.Scan(&item.SettlementID, &item.CommandID, &item.TransactionID, &item.PolicyType, &item.TriggerMode, &userID, &item.CycleKey, &item.RegionID, &item.PolicyID, &item.LevelNo, &item.IncomeUSDMinor, &item.SalaryUSDMinorDelta, &item.Status, &item.Reason, &item.CreatedAtMS); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
item.UserID = strconv.FormatInt(userID, 10)
|
||
item.User = userDTO{UserID: item.UserID}
|
||
item.IncomeUSD = formatUSDMinor(item.IncomeUSDMinor)
|
||
item.SalaryUSDDelta = formatUSDMinor(item.SalaryUSDMinorDelta)
|
||
items = append(items, item)
|
||
userIDs = append(userIDs, userID)
|
||
}
|
||
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 id, err := strconv.ParseInt(items[i].UserID, 10, 64); err == nil {
|
||
items[i].User = userDTOFromProfile(items[i].UserID, profiles[id])
|
||
}
|
||
}
|
||
return items, total, nil
|
||
}
|
||
|
||
func (s *Service) resolveFilterUserID(ctx context.Context, appCode string, currentUserID int64, filter shared.UserIdentityFilter) (int64, bool, error) {
|
||
if currentUserID > 0 || filter.IsEmpty() {
|
||
return currentUserID, true, nil
|
||
}
|
||
userID, matched, err := shared.ResolveUserIdentityFilter(ctx, s.userDB, strings.TrimSpace(appCode), filter, time.Now().UnixMilli())
|
||
if err != nil || !matched {
|
||
return 0, matched, err
|
||
}
|
||
return userID, true, nil
|
||
}
|
||
|
||
// Settle 执行手动或自动结算；它只关心“触发方式”和“用户集合”，差值与幂等统一落在 settleCandidate。
|
||
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.PolicyType = normalizePolicyType(req.PolicyType)
|
||
req.TriggerMode = normalizeTriggerMode(req.TriggerMode)
|
||
if req.TriggerMode == "" {
|
||
req.TriggerMode = triggerManual
|
||
}
|
||
if req.CycleKey == "" {
|
||
req.CycleKey = previousMonthCycleKey(time.Now().UTC())
|
||
}
|
||
userIDs := settlementUserIDValues(req.UserIDs)
|
||
userFilter := int64(0)
|
||
if len(userIDs) == 1 {
|
||
userFilter = userIDs[0]
|
||
}
|
||
// 单用户结算在候选聚合阶段下推 userFilter；批量结算先算区域候选再做白名单过滤。
|
||
candidates, err := s.collectCandidates(ctx, strings.TrimSpace(appCode), req.PolicyType, req.TriggerMode, req.CycleKey, req.RegionID, req.CountryID, req.CountryCode, userFilter, time.Now().UTC().UnixMilli())
|
||
if err != nil {
|
||
return settleResultDTO{}, err
|
||
}
|
||
if len(userIDs) > 0 {
|
||
allowed := map[int64]struct{}{}
|
||
for _, id := range userIDs {
|
||
if id > 0 {
|
||
allowed[id] = struct{}{}
|
||
}
|
||
}
|
||
// 前端多选只允许结算勾选用户；这里二次过滤，避免客户端篡改国家/区域后误结不在选择范围内的人。
|
||
filtered := candidates[:0]
|
||
for _, candidate := range candidates {
|
||
if _, ok := allowed[candidate.UserID]; ok {
|
||
filtered = append(filtered, candidate)
|
||
}
|
||
}
|
||
candidates = filtered
|
||
}
|
||
sortTeamCandidates(candidates)
|
||
|
||
result := settleResultDTO{PolicyType: req.PolicyType, CycleKey: req.CycleKey, RequestedCount: len(userIDs), Records: []recordDTO{}}
|
||
profiles, _ := s.userProfiles(ctx, appCode, candidateUserIDs(candidates))
|
||
for _, candidate := range candidates {
|
||
result.ProcessedCount++
|
||
// 单个候选失败不阻塞同批其他用户，后台返回 failure_count 供运营重试定位。
|
||
record, settled, err := s.settleCandidate(ctx, strings.TrimSpace(appCode), operatorAdminID, req.TriggerMode, candidate)
|
||
if err != nil {
|
||
result.FailureCount++
|
||
continue
|
||
}
|
||
if !settled {
|
||
result.SkippedCount++
|
||
continue
|
||
}
|
||
if id, err := strconv.ParseInt(record.UserID, 10, 64); err == nil {
|
||
record.User = userDTOFromProfile(record.UserID, profiles[id])
|
||
}
|
||
result.SuccessCount++
|
||
result.Records = append(result.Records, record)
|
||
}
|
||
return result, nil
|
||
}
|
||
|
||
// RunAutomaticDue 是自动任务入口：只在每月 5 日结上一个完整自然月，并固定先结 BD 再结 Admin。
|
||
func (s *Service) RunAutomaticDue(ctx context.Context, appCode string, now time.Time) (autoSettlementResult, error) {
|
||
utcNow := now.UTC()
|
||
result := autoSettlementResult{
|
||
Due: utcNow.Day() == 5,
|
||
CycleKey: previousMonthCycleKey(utcNow),
|
||
Results: map[string]settleResultDTO{},
|
||
}
|
||
if !result.Due {
|
||
return result, nil
|
||
}
|
||
// BD 必须先跑：Admin 的收入基数来自同一周期内 BD 已成功入账的工资记录。
|
||
bdResult, err := s.Settle(ctx, appCode, 0, settleRequest{
|
||
PolicyType: policyTypeBD,
|
||
TriggerMode: triggerAutomatic,
|
||
CycleKey: result.CycleKey,
|
||
})
|
||
if err != nil {
|
||
return result, err
|
||
}
|
||
result.Results[policyTypeBD] = bdResult
|
||
|
||
adminResult, err := s.Settle(ctx, appCode, 0, settleRequest{
|
||
PolicyType: policyTypeAdmin,
|
||
TriggerMode: triggerAutomatic,
|
||
CycleKey: result.CycleKey,
|
||
})
|
||
if err != nil {
|
||
return result, err
|
||
}
|
||
result.Results[policyTypeAdmin] = adminResult
|
||
return result, nil
|
||
}
|
||
|
||
// collectCandidates 汇总 BD 和 Admin 两类候选；policyType 为空时用于记录页或内部全量扫描。
|
||
func (s *Service) collectCandidates(ctx context.Context, appCode string, policyType string, triggerMode string, cycleKey string, regionID int64, countryID int64, countryCode string, userID int64, nowMS int64) ([]teamCandidate, error) {
|
||
policyType = normalizePolicyType(policyType)
|
||
triggerMode = normalizeTriggerMode(triggerMode)
|
||
if cycleKey == "" {
|
||
cycleKey = previousMonthCycleKey(time.Now().UTC())
|
||
}
|
||
// 国家筛选在 user 库维护，结算记录只存 region_id，因此先把国家解析为一组可用区域。
|
||
regionFilter, err := s.regionFilter(ctx, appCode, regionID, countryID, countryCode)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var candidates []teamCandidate
|
||
if policyType == "" || policyType == policyTypeBD {
|
||
items, err := s.collectBDCandidates(ctx, appCode, triggerMode, cycleKey, regionFilter, userID, nowMS)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
candidates = append(candidates, items...)
|
||
}
|
||
if policyType == "" || policyType == policyTypeAdmin {
|
||
items, err := s.collectAdminCandidates(ctx, appCode, triggerMode, cycleKey, regionFilter, userID, nowMS)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
candidates = append(candidates, items...)
|
||
}
|
||
return candidates, nil
|
||
}
|
||
|
||
// collectBDCandidates 以 Agency 为桥接，把 Host/Agency 已成功结算的主播收入汇总到其上级 BD。
|
||
func (s *Service) collectBDCandidates(ctx context.Context, appCode string, triggerMode string, cycleKey string, regions map[int64]struct{}, userID int64, nowMS int64) ([]teamCandidate, error) {
|
||
agencyIncome, err := s.hostIncomeByAgencyOwner(ctx, appCode, cycleKey)
|
||
if err != nil || len(agencyIncome) == 0 {
|
||
return nil, err
|
||
}
|
||
// host_salary_settlement_records 只知道 agency owner；BD 归属必须回 user 库查 active agency 当前关系。
|
||
agencies, err := s.activeAgenciesByOwner(ctx, appCode, keysInt64(agencyIncome))
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
incomeByBDRegion := map[string]int64{}
|
||
for ownerID, income := range agencyIncome {
|
||
agency, ok := agencies[ownerID]
|
||
if !ok || agency.ParentBDUserID <= 0 || income <= 0 {
|
||
continue
|
||
}
|
||
// 区域来自 Agency 拥有者 users 当前区域，不从主播记录或 Agency 历史快照推导。
|
||
if len(regions) > 0 {
|
||
if _, ok := regions[agency.RegionID]; !ok {
|
||
continue
|
||
}
|
||
}
|
||
if userID > 0 && agency.ParentBDUserID != userID {
|
||
continue
|
||
}
|
||
incomeByBDRegion[teamCandidateKey(agency.ParentBDUserID, agency.RegionID)] += income
|
||
}
|
||
return s.candidatesFromIncome(ctx, appCode, policyTypeBD, triggerMode, cycleKey, incomeByBDRegion, nowMS)
|
||
}
|
||
|
||
// collectAdminCandidates 以已发 BD 工资作为 Admin 收入基数，再通过 BD profile 找到上级 Admin。
|
||
func (s *Service) collectAdminCandidates(ctx context.Context, appCode string, triggerMode string, cycleKey string, regions map[int64]struct{}, userID int64, nowMS int64) ([]teamCandidate, error) {
|
||
bdIncome, err := s.bdSalaryIncomeByUser(ctx, appCode, cycleKey)
|
||
if err != nil || len(bdIncome) == 0 {
|
||
return nil, err
|
||
}
|
||
// Admin 政策定义为“下属 BD 的 BD Salary 合计”，因此只有已成功入账的 BD 记录才进入这里。
|
||
bdProfiles, err := s.bdProfilesByUser(ctx, appCode, keysInt64(bdIncome), "bd")
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
leaderIDs := make([]int64, 0, len(bdProfiles))
|
||
for _, profile := range bdProfiles {
|
||
if profile.ParentLeaderUserID > 0 {
|
||
leaderIDs = append(leaderIDs, profile.ParentLeaderUserID)
|
||
}
|
||
}
|
||
leaders, err := s.bdProfilesByUser(ctx, appCode, uniquePositiveUserIDs(leaderIDs), "bd_leader")
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
incomeByAdminRegion := map[string]int64{}
|
||
for bdUserID, income := range bdIncome {
|
||
profile, ok := bdProfiles[bdUserID]
|
||
if !ok || profile.ParentLeaderUserID <= 0 || income <= 0 {
|
||
continue
|
||
}
|
||
leader, ok := leaders[profile.ParentLeaderUserID]
|
||
if !ok {
|
||
continue
|
||
}
|
||
// Admin 的适用区域以 BD Leader 用户当前区域为准，不再读取 bd_leader_profiles 的历史 region_id。
|
||
if len(regions) > 0 {
|
||
if _, ok := regions[leader.RegionID]; !ok {
|
||
continue
|
||
}
|
||
}
|
||
if userID > 0 && leader.UserID != userID {
|
||
continue
|
||
}
|
||
incomeByAdminRegion[teamCandidateKey(leader.UserID, leader.RegionID)] += income
|
||
}
|
||
return s.candidatesFromIncome(ctx, appCode, policyTypeAdmin, triggerMode, cycleKey, incomeByAdminRegion, nowMS)
|
||
}
|
||
|
||
// candidatesFromIncome 将“用户+区域+收入基数”转换为可结算候选，集中处理政策匹配、等级选择和差值计算。
|
||
func (s *Service) candidatesFromIncome(ctx context.Context, appCode string, policyType string, triggerMode string, cycleKey string, incomeByUserRegion map[string]int64, nowMS int64) ([]teamCandidate, error) {
|
||
if len(incomeByUserRegion) == 0 {
|
||
return nil, nil
|
||
}
|
||
// 区域名称只是展示字段；读取失败不影响结算主流程，主键 region_id 仍然完整。
|
||
regionNames, _ := s.regionNames(ctx, appCode)
|
||
out := make([]teamCandidate, 0, len(incomeByUserRegion))
|
||
for key, income := range incomeByUserRegion {
|
||
userID, regionID := parseTeamCandidateKey(key)
|
||
// 自动/手动政策同一区域可共存，triggerMode 必须参与政策解析，否则会把人工复核政策提前发放。
|
||
policy, ok, err := s.resolvePolicy(ctx, appCode, policyType, regionID, triggerMode, nowMS)
|
||
if err != nil || !ok {
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
continue
|
||
}
|
||
progress, err := s.loadProgress(ctx, appCode, policyType, userID, cycleKey)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
level := highestTeamLevel(policy.Levels, income)
|
||
// 等级薪资是“累计应得”，这里先算一次待结差值供页面展示，真正发放时还会在事务锁内重算。
|
||
delta := positiveDelta(level.SalaryMinor, progress.SettledSalaryUSDMinor)
|
||
out = append(out, teamCandidate{
|
||
PolicyType: policyType,
|
||
UserID: userID,
|
||
CycleKey: cycleKey,
|
||
RegionID: regionID,
|
||
RegionName: regionNames[regionID],
|
||
Policy: policy,
|
||
Level: level,
|
||
IncomeUSDMinor: income,
|
||
Progress: progress,
|
||
SalaryUSDMinorDelta: delta,
|
||
})
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// settleCandidate 在一个 wallet 事务内完成进度锁定、差值重算、钱包入账、流水和结算记录写入。
|
||
func (s *Service) settleCandidate(ctx context.Context, appCode string, operatorAdminID int64, triggerMode string, candidate teamCandidate) (recordDTO, bool, error) {
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
tx, err := s.walletDB.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return recordDTO{}, false, err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
|
||
progress, err := s.lockProgress(ctx, tx, appCode, candidate, nowMS)
|
||
if err != nil {
|
||
return recordDTO{}, false, err
|
||
}
|
||
// 结算表里的等级工资是累计值；每次只发“当前应得累计 - 已发累计”，因此重复提交同一批用户不会二次入账。
|
||
salaryDelta := positiveDelta(candidate.Level.SalaryMinor, progress.SettledSalaryUSDMinor)
|
||
if salaryDelta <= 0 {
|
||
// 未升档但收入基数增加时只刷新 last_settled_income，减少后续待结算列表反复出现 0 金额候选。
|
||
if candidate.IncomeUSDMinor > progress.LastSettledIncomeUSDMinor {
|
||
progress.LastSettledIncomeUSDMinor = candidate.IncomeUSDMinor
|
||
if err := s.updateProgress(ctx, tx, appCode, candidate, progress, nowMS); err != nil {
|
||
return recordDTO{}, false, err
|
||
}
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return recordDTO{}, false, err
|
||
}
|
||
return recordDTO{}, false, nil
|
||
}
|
||
|
||
// nextProgress 使用 max，确保历史已发累计不会因政策调整或候选回退被写小。
|
||
nextProgress := teamProgress{
|
||
SettledLevelNo: maxInt(progress.SettledLevelNo, candidate.Level.LevelNo),
|
||
SettledSalaryUSDMinor: maxInt64(progress.SettledSalaryUSDMinor, candidate.Level.SalaryMinor),
|
||
LastSettledIncomeUSDMinor: maxInt64(progress.LastSettledIncomeUSDMinor, candidate.IncomeUSDMinor),
|
||
LastPolicyID: candidate.Policy.PolicyID,
|
||
MonthClosedAtMS: nowMS,
|
||
Version: progress.Version,
|
||
}
|
||
salaryAssetType := teamSalaryAssetType(candidate.PolicyType)
|
||
if salaryAssetType == "" {
|
||
return recordDTO{}, false, fmt.Errorf("team salary asset type is invalid: %s", candidate.PolicyType)
|
||
}
|
||
metadata := settlementMetadata{
|
||
AppCode: appCode,
|
||
PolicyType: candidate.PolicyType,
|
||
TriggerMode: triggerMode,
|
||
UserID: candidate.UserID,
|
||
CycleKey: candidate.CycleKey,
|
||
RegionID: candidate.RegionID,
|
||
PolicyID: candidate.Policy.PolicyID,
|
||
PolicyName: candidate.Policy.Name,
|
||
LevelNo: candidate.Level.LevelNo,
|
||
IncomeUSDMinor: candidate.IncomeUSDMinor,
|
||
SalaryUSDMinorDelta: salaryDelta,
|
||
SalaryAssetType: salaryAssetType,
|
||
SettledLevelNo: nextProgress.SettledLevelNo,
|
||
SettledSalaryUSDMinor: nextProgress.SettledSalaryUSDMinor,
|
||
ProcessedAtMS: nowMS,
|
||
OperatorAdminID: operatorAdminID,
|
||
}
|
||
// commandID 带上目标等级和收入基数；同一进度重复提交会命中 progress 差值为 0，异常重试也会被唯一键兜住。
|
||
commandID := teamSalaryCommandID(candidate, triggerMode, nextProgress)
|
||
transactionID := teamSalaryTransactionID(appCode, commandID)
|
||
if err := s.insertWalletTransaction(ctx, tx, appCode, transactionID, commandID, metadata, nowMS); err != nil {
|
||
return recordDTO{}, false, err
|
||
}
|
||
balanceAfter, version, err := s.creditWallet(ctx, tx, appCode, transactionID, candidate.UserID, metadata.SalaryAssetType, salaryDelta, nowMS)
|
||
if err != nil {
|
||
return recordDTO{}, false, err
|
||
}
|
||
if err := s.insertWalletOutbox(ctx, tx, appCode, transactionID, commandID, candidate.UserID, metadata.SalaryAssetType, salaryDelta, balanceAfter, version, metadata, nowMS); err != nil {
|
||
return recordDTO{}, false, err
|
||
}
|
||
if err := s.insertRecord(ctx, tx, appCode, transactionID, commandID, metadata, nowMS); err != nil {
|
||
return recordDTO{}, false, err
|
||
}
|
||
if err := s.updateProgress(ctx, tx, appCode, candidate, nextProgress, nowMS); err != nil {
|
||
return recordDTO{}, false, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return recordDTO{}, false, err
|
||
}
|
||
return recordDTO{
|
||
SettlementID: teamSalarySettlementID(appCode, commandID),
|
||
CommandID: commandID,
|
||
TransactionID: transactionID,
|
||
PolicyType: candidate.PolicyType,
|
||
TriggerMode: triggerMode,
|
||
UserID: strconv.FormatInt(candidate.UserID, 10),
|
||
User: userDTO{UserID: strconv.FormatInt(candidate.UserID, 10)},
|
||
CycleKey: candidate.CycleKey,
|
||
RegionID: candidate.RegionID,
|
||
PolicyID: candidate.Policy.PolicyID,
|
||
LevelNo: int32(candidate.Level.LevelNo),
|
||
IncomeUSDMinor: candidate.IncomeUSDMinor,
|
||
IncomeUSD: formatUSDMinor(candidate.IncomeUSDMinor),
|
||
SalaryUSDMinorDelta: salaryDelta,
|
||
SalaryUSDDelta: formatUSDMinor(salaryDelta),
|
||
Status: statusSucceeded,
|
||
CreatedAtMS: nowMS,
|
||
}, true, nil
|
||
}
|
||
|
||
// hostIncomeByAgencyOwner 读取 Host/Agency 已成功结算记录，BD 只基于主播工资和月底剩余折美元，不计 Agency 自己工资。
|
||
func (s *Service) hostIncomeByAgencyOwner(ctx context.Context, appCode string, cycleKey string) (map[int64]int64, error) {
|
||
rows, err := s.walletDB.QueryContext(ctx, `
|
||
SELECT agency_owner_user_id, SUM(host_salary_usd_minor_delta + residual_usd_minor_delta)
|
||
FROM host_salary_settlement_records
|
||
WHERE app_code = ? AND cycle_key = ? AND status = 'succeeded' AND agency_owner_user_id > 0
|
||
GROUP BY agency_owner_user_id`, appCode, cycleKey)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
out := map[int64]int64{}
|
||
for rows.Next() {
|
||
var ownerID int64
|
||
var income sql.NullInt64
|
||
if err := rows.Scan(&ownerID, &income); err != nil {
|
||
return nil, err
|
||
}
|
||
if ownerID > 0 && income.Int64 > 0 {
|
||
out[ownerID] = income.Int64
|
||
}
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// bdSalaryIncomeByUser 读取同周期已成功发放的 BD 工资，Admin 只按真实入账 BD Salary 计算。
|
||
func (s *Service) bdSalaryIncomeByUser(ctx context.Context, appCode string, cycleKey string) (map[int64]int64, error) {
|
||
rows, err := s.walletDB.QueryContext(ctx, `
|
||
SELECT user_id, SUM(salary_usd_minor_delta)
|
||
FROM team_salary_settlement_records
|
||
WHERE app_code = ? AND policy_type = 'bd' AND cycle_key = ? AND status = 'succeeded'
|
||
GROUP BY user_id`, appCode, cycleKey)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
out := map[int64]int64{}
|
||
for rows.Next() {
|
||
var userID int64
|
||
var income sql.NullInt64
|
||
if err := rows.Scan(&userID, &income); err != nil {
|
||
return nil, err
|
||
}
|
||
if userID > 0 && income.Int64 > 0 {
|
||
out[userID] = income.Int64
|
||
}
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// agencyInfo 是 BD 汇总所需的最小组织关系视图。
|
||
type agencyInfo struct {
|
||
OwnerUserID int64
|
||
ParentBDUserID int64
|
||
RegionID int64
|
||
}
|
||
|
||
// activeAgenciesByOwner 只使用 active agency，并从 owner users 行读取当前区域，避免历史快照继续影响 BD 工资。
|
||
func (s *Service) activeAgenciesByOwner(ctx context.Context, appCode string, ownerIDs []int64) (map[int64]agencyInfo, error) {
|
||
ownerIDs = uniquePositiveUserIDs(ownerIDs)
|
||
if len(ownerIDs) == 0 {
|
||
return map[int64]agencyInfo{}, nil
|
||
}
|
||
placeholders := placeholders(len(ownerIDs))
|
||
args := make([]any, 0, len(ownerIDs)+1)
|
||
args = append(args, appCode)
|
||
for _, id := range ownerIDs {
|
||
args = append(args, id)
|
||
}
|
||
rows, err := s.userDB.QueryContext(ctx, `
|
||
SELECT a.owner_user_id, a.parent_bd_user_id, COALESCE(u.region_id, 0)
|
||
FROM agencies a
|
||
INNER JOIN users u
|
||
ON u.app_code = a.app_code AND u.user_id = a.owner_user_id
|
||
WHERE a.app_code = ? AND a.status = 'active' AND a.owner_user_id IN (`+placeholders+`)`, args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
out := map[int64]agencyInfo{}
|
||
for rows.Next() {
|
||
var item agencyInfo
|
||
if err := rows.Scan(&item.OwnerUserID, &item.ParentBDUserID, &item.RegionID); err != nil {
|
||
return nil, err
|
||
}
|
||
out[item.OwnerUserID] = item
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// bdProfile 是 Admin 汇总所需的最小 BD/BD Leader 组织关系视图。
|
||
type bdProfile struct {
|
||
UserID int64
|
||
Role string
|
||
RegionID int64
|
||
ParentLeaderUserID int64
|
||
}
|
||
|
||
// bdProfilesByUser 按角色取 active profile，RegionID 始终来自用户当前区域，调用方必须显式传 role。
|
||
func (s *Service) bdProfilesByUser(ctx context.Context, appCode string, userIDs []int64, role string) (map[int64]bdProfile, error) {
|
||
userIDs = uniquePositiveUserIDs(userIDs)
|
||
if len(userIDs) == 0 {
|
||
return map[int64]bdProfile{}, nil
|
||
}
|
||
placeholders := placeholders(len(userIDs))
|
||
args := make([]any, 0, len(userIDs)+1)
|
||
args = append(args, appCode)
|
||
for _, id := range userIDs {
|
||
args = append(args, id)
|
||
}
|
||
query := `
|
||
SELECT bp.user_id, bp.role, COALESCE(u.region_id, 0), COALESCE(bp.parent_leader_user_id, 0)
|
||
FROM bd_profiles bp
|
||
INNER JOIN users u
|
||
ON u.app_code = bp.app_code AND u.user_id = bp.user_id
|
||
WHERE bp.app_code = ? AND bp.role = 'bd' AND bp.status = 'active' AND bp.user_id IN (` + placeholders + `)`
|
||
if role == "bd_leader" {
|
||
query = `
|
||
SELECT bl.user_id, 'bd_leader', COALESCE(u.region_id, 0), 0
|
||
FROM bd_leader_profiles bl
|
||
INNER JOIN users u
|
||
ON u.app_code = bl.app_code AND u.user_id = bl.user_id
|
||
WHERE bl.app_code = ? AND bl.status = 'active' AND bl.user_id IN (` + placeholders + `)`
|
||
}
|
||
rows, err := s.userDB.QueryContext(ctx, `
|
||
`+query, args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
out := map[int64]bdProfile{}
|
||
for rows.Next() {
|
||
var item bdProfile
|
||
if err := rows.Scan(&item.UserID, &item.Role, &item.RegionID, &item.ParentLeaderUserID); err != nil {
|
||
return nil, err
|
||
}
|
||
out[item.UserID] = item
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// resolvePolicy 按 app、类型、区域、触发方式和生效时间解析当前唯一有效政策。
|
||
func (s *Service) resolvePolicy(ctx context.Context, appCode string, policyType string, regionID int64, triggerMode string, nowMS int64) (teamPolicy, bool, error) {
|
||
conditions := []string{"app_code = ?", "policy_type = ?", "region_id = ?", "status = 'active'", "effective_from_ms <= ?", "(effective_to_ms = 0 OR effective_to_ms > ?)"}
|
||
args := []any{appCode, policyType, regionID, nowMS, nowMS}
|
||
if triggerMode != "" {
|
||
conditions = append(conditions, "settlement_trigger_mode = ?")
|
||
args = append(args, triggerMode)
|
||
}
|
||
row := s.adminDB.QueryRowContext(ctx, `
|
||
SELECT id, policy_type, name, region_id, settlement_trigger_mode
|
||
FROM admin_team_salary_policies
|
||
WHERE `+strings.Join(conditions, " AND ")+`
|
||
ORDER BY effective_from_ms DESC, id DESC
|
||
LIMIT 1`, args...)
|
||
var policy teamPolicy
|
||
if err := row.Scan(&policy.PolicyID, &policy.PolicyType, &policy.Name, &policy.RegionID, &policy.TriggerMode); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
// 没有政策不是系统错误；该区域用户本轮没有可结算候选。
|
||
return teamPolicy{}, false, nil
|
||
}
|
||
return teamPolicy{}, false, err
|
||
}
|
||
levels, err := s.policyLevels(ctx, policy.PolicyID)
|
||
if err != nil {
|
||
return teamPolicy{}, false, err
|
||
}
|
||
policy.Levels = levels
|
||
return policy, true, nil
|
||
}
|
||
|
||
// policyLevels 读取并转换政策等级；排序按门槛递增，highestTeamLevel 可以线性选择最高达标档。
|
||
func (s *Service) policyLevels(ctx context.Context, policyID uint64) ([]teamPolicyLevel, error) {
|
||
rows, err := s.adminDB.QueryContext(ctx, `
|
||
SELECT level_no, CAST(threshold_usd AS CHAR), CAST(salary_usd AS CHAR)
|
||
FROM admin_team_salary_policy_levels
|
||
WHERE policy_id = ? AND status = 'active'
|
||
ORDER BY threshold_usd ASC, level_no ASC`, policyID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
var out []teamPolicyLevel
|
||
for rows.Next() {
|
||
var level teamPolicyLevel
|
||
var threshold string
|
||
var salary string
|
||
if err := rows.Scan(&level.LevelNo, &threshold, &salary); err != nil {
|
||
return nil, err
|
||
}
|
||
thresholdMinor, err := usdStringToMinor(threshold)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
salaryMinor, err := usdStringToMinor(salary)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
level.ThresholdMinor = thresholdMinor
|
||
level.SalaryMinor = salaryMinor
|
||
out = append(out, level)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// loadProgress 为列表展示读取当前进度；没有进度代表本周期尚未发过工资。
|
||
func (s *Service) loadProgress(ctx context.Context, appCode string, policyType string, userID int64, cycleKey string) (teamProgress, error) {
|
||
var progress teamProgress
|
||
err := s.walletDB.QueryRowContext(ctx, `
|
||
SELECT settled_level_no, settled_salary_usd_minor, last_settled_income_usd_minor,
|
||
last_policy_id, month_closed_at_ms, version
|
||
FROM team_salary_settlement_progress
|
||
WHERE app_code = ? AND policy_type = ? AND user_id = ? AND cycle_key = ?`,
|
||
appCode, policyType, userID, cycleKey,
|
||
).Scan(&progress.SettledLevelNo, &progress.SettledSalaryUSDMinor, &progress.LastSettledIncomeUSDMinor, &progress.LastPolicyID, &progress.MonthClosedAtMS, &progress.Version)
|
||
if err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return teamProgress{Version: 1}, nil
|
||
}
|
||
return teamProgress{}, err
|
||
}
|
||
return progress, nil
|
||
}
|
||
|
||
// lockProgress 先 INSERT IGNORE 再 SELECT FOR UPDATE，保证并发结算同一用户周期时只有一个事务能重算差值。
|
||
func (s *Service) lockProgress(ctx context.Context, tx *sql.Tx, appCode string, candidate teamCandidate, nowMS int64) (teamProgress, error) {
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT IGNORE INTO team_salary_settlement_progress (
|
||
app_code, policy_type, user_id, cycle_key, region_id, settled_level_no,
|
||
settled_salary_usd_minor, last_settled_income_usd_minor, last_policy_id,
|
||
month_closed_at_ms, version, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, 0, 0, 0, 0, 0, 1, ?, ?)`,
|
||
appCode, candidate.PolicyType, candidate.UserID, candidate.CycleKey, candidate.RegionID, nowMS, nowMS,
|
||
); err != nil {
|
||
return teamProgress{}, err
|
||
}
|
||
var progress teamProgress
|
||
if err := tx.QueryRowContext(ctx, `
|
||
SELECT settled_level_no, settled_salary_usd_minor, last_settled_income_usd_minor,
|
||
last_policy_id, month_closed_at_ms, version
|
||
FROM team_salary_settlement_progress
|
||
WHERE app_code = ? AND policy_type = ? AND user_id = ? AND cycle_key = ?
|
||
FOR UPDATE`, appCode, candidate.PolicyType, candidate.UserID, candidate.CycleKey,
|
||
).Scan(&progress.SettledLevelNo, &progress.SettledSalaryUSDMinor, &progress.LastSettledIncomeUSDMinor, &progress.LastPolicyID, &progress.MonthClosedAtMS, &progress.Version); err != nil {
|
||
return teamProgress{}, err
|
||
}
|
||
return progress, nil
|
||
}
|
||
|
||
// updateProgress 使用 version 乐观校验，防止事务外或异常并发修改覆盖已提交进度。
|
||
func (s *Service) updateProgress(ctx context.Context, tx *sql.Tx, appCode string, candidate teamCandidate, progress teamProgress, nowMS int64) error {
|
||
result, err := tx.ExecContext(ctx, `
|
||
UPDATE team_salary_settlement_progress
|
||
SET region_id = ?,
|
||
settled_level_no = ?,
|
||
settled_salary_usd_minor = ?,
|
||
last_settled_income_usd_minor = ?,
|
||
last_policy_id = ?,
|
||
month_closed_at_ms = ?,
|
||
version = version + 1,
|
||
updated_at_ms = ?
|
||
WHERE app_code = ? AND policy_type = ? AND user_id = ? AND cycle_key = ? AND version = ?`,
|
||
candidate.RegionID,
|
||
progress.SettledLevelNo,
|
||
progress.SettledSalaryUSDMinor,
|
||
progress.LastSettledIncomeUSDMinor,
|
||
progress.LastPolicyID,
|
||
progress.MonthClosedAtMS,
|
||
nowMS,
|
||
appCode,
|
||
candidate.PolicyType,
|
||
candidate.UserID,
|
||
candidate.CycleKey,
|
||
progress.Version,
|
||
)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
rows, err := result.RowsAffected()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if rows != 1 {
|
||
return fmt.Errorf("team salary settlement progress version conflict")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// insertWalletTransaction 写钱包交易主表，command_id 是业务幂等键。
|
||
func (s *Service) insertWalletTransaction(ctx context.Context, tx *sql.Tx, appCode string, transactionID string, commandID string, metadata settlementMetadata, nowMS int64) error {
|
||
payload, err := json.Marshal(metadata)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = tx.ExecContext(ctx, `
|
||
INSERT INTO wallet_transactions (
|
||
app_code, transaction_id, command_id, biz_type, status, request_hash,
|
||
external_ref, metadata_json, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, 'succeeded', ?, ?, ?, ?, ?)`,
|
||
appCode, transactionID, commandID, bizTypeTeamSettlement, stableHash(string(payload)), metadata.CycleKey, string(payload), nowMS, nowMS,
|
||
)
|
||
return err
|
||
}
|
||
|
||
// creditWallet 锁定对应角色工资钱包并追加 wallet_entries，使余额、流水和版本在同一事务内一致。
|
||
func (s *Service) creditWallet(ctx context.Context, tx *sql.Tx, appCode string, transactionID string, userID int64, assetType string, amount int64, nowMS int64) (int64, int64, error) {
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT IGNORE INTO wallet_accounts (
|
||
app_code, user_id, asset_type, available_amount, frozen_amount, version, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, 0, 0, 1, ?, ?)`, appCode, userID, assetType, nowMS, nowMS); err != nil {
|
||
return 0, 0, err
|
||
}
|
||
var available int64
|
||
var frozen int64
|
||
var version int64
|
||
if err := tx.QueryRowContext(ctx, `
|
||
SELECT available_amount, frozen_amount, version
|
||
FROM wallet_accounts
|
||
WHERE app_code = ? AND user_id = ? AND asset_type = ?
|
||
FOR UPDATE`, appCode, userID, assetType,
|
||
).Scan(&available, &frozen, &version); err != nil {
|
||
return 0, 0, err
|
||
}
|
||
after := available + amount
|
||
if after < 0 {
|
||
return 0, 0, fmt.Errorf("team salary credit overflow")
|
||
}
|
||
// 工资只会正向入账；version 条件仍保留，和其它钱包写入路径保持并发语义一致。
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE wallet_accounts
|
||
SET available_amount = ?, version = version + 1, updated_at_ms = ?
|
||
WHERE app_code = ? AND user_id = ? AND asset_type = ? AND version = ?`,
|
||
after, nowMS, appCode, userID, assetType, version,
|
||
); err != nil {
|
||
return 0, 0, err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO wallet_entries (
|
||
app_code, transaction_id, user_id, asset_type, available_delta, frozen_delta,
|
||
available_after, frozen_after, counterparty_user_id, room_id, created_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, 0, ?, ?, 0, '', ?)`,
|
||
appCode, transactionID, userID, assetType, amount, after, frozen, nowMS,
|
||
); err != nil {
|
||
return 0, 0, err
|
||
}
|
||
return after, version + 1, nil
|
||
}
|
||
|
||
// insertWalletOutbox 投递余额变更事件，供后续 MQ/通知链路异步消费。
|
||
func (s *Service) insertWalletOutbox(ctx context.Context, tx *sql.Tx, appCode string, transactionID string, commandID string, userID int64, assetType string, amount int64, balanceAfter int64, version int64, metadata settlementMetadata, nowMS int64) error {
|
||
payload, err := json.Marshal(map[string]any{
|
||
"transaction_id": transactionID,
|
||
"command_id": commandID,
|
||
"user_id": userID,
|
||
"asset_type": assetType,
|
||
"available_delta": amount,
|
||
"available_after": balanceAfter,
|
||
"version": version,
|
||
"metadata": metadata,
|
||
"created_at_ms": nowMS,
|
||
})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = tx.ExecContext(ctx, `
|
||
INSERT INTO wallet_outbox (
|
||
app_code, event_id, event_type, transaction_id, command_id, user_id, asset_type,
|
||
available_delta, frozen_delta, payload, status, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, 'WalletBalanceChanged', ?, ?, ?, ?, ?, 0, ?, 'pending', ?, ?)`,
|
||
appCode, transactionID+":"+assetType, transactionID, commandID, userID, assetType, amount, string(payload), nowMS, nowMS,
|
||
)
|
||
return err
|
||
}
|
||
|
||
// insertRecord 写业务结算记录，是后台查询 settlement_id 的权威来源。
|
||
func (s *Service) insertRecord(ctx context.Context, tx *sql.Tx, appCode string, transactionID string, commandID string, metadata settlementMetadata, nowMS int64) error {
|
||
_, err := tx.ExecContext(ctx, `
|
||
INSERT INTO team_salary_settlement_records (
|
||
app_code, settlement_id, command_id, transaction_id, policy_type, trigger_mode,
|
||
user_id, cycle_key, region_id, policy_id, level_no, income_usd_minor,
|
||
salary_usd_minor_delta, status, reason, created_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'succeeded', '', ?)`,
|
||
appCode,
|
||
teamSalarySettlementID(appCode, commandID),
|
||
commandID,
|
||
transactionID,
|
||
metadata.PolicyType,
|
||
metadata.TriggerMode,
|
||
metadata.UserID,
|
||
metadata.CycleKey,
|
||
metadata.RegionID,
|
||
metadata.PolicyID,
|
||
metadata.LevelNo,
|
||
metadata.IncomeUSDMinor,
|
||
metadata.SalaryUSDMinorDelta,
|
||
nowMS,
|
||
)
|
||
return err
|
||
}
|
||
|
||
// recordWhere 统一构造记录查询条件，避免列表 API 和后续导出 API 出现筛选口径差异。
|
||
func (s *Service) recordWhere(ctx context.Context, appCode string, req recordQuery) (string, []any, error) {
|
||
conditions := []string{"app_code = ?"}
|
||
args := []any{appCode}
|
||
if req.PolicyType != "" {
|
||
conditions = append(conditions, "policy_type = ?")
|
||
args = append(args, req.PolicyType)
|
||
}
|
||
if req.TriggerMode != "" {
|
||
conditions = append(conditions, "trigger_mode = ?")
|
||
args = append(args, req.TriggerMode)
|
||
}
|
||
if req.CycleKey != "" {
|
||
conditions = append(conditions, "cycle_key = ?")
|
||
args = append(args, req.CycleKey)
|
||
}
|
||
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.PolicyID > 0 {
|
||
conditions = append(conditions, "policy_id = ?")
|
||
args = append(args, req.PolicyID)
|
||
}
|
||
if req.RegionID > 0 {
|
||
conditions = append(conditions, "region_id = ?")
|
||
args = append(args, req.RegionID)
|
||
} else if req.CountryID > 0 || req.CountryCode != "" {
|
||
regions, err := s.regionFilter(ctx, appCode, 0, req.CountryID, req.CountryCode)
|
||
if err != nil {
|
||
return "", nil, err
|
||
}
|
||
if len(regions) == 0 {
|
||
conditions = append(conditions, "region_id = -1")
|
||
} else {
|
||
ids := keysInt64FromSet(regions)
|
||
conditions = append(conditions, "region_id IN ("+placeholders(len(ids))+")")
|
||
for _, id := range ids {
|
||
args = append(args, id)
|
||
}
|
||
}
|
||
}
|
||
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)
|
||
}
|
||
return "WHERE " + strings.Join(conditions, " AND "), args, nil
|
||
}
|
||
|
||
// regionFilter 把区域、国家 ID、国家代码归一为 region_id 集合；region_id 直传时优先级最高。
|
||
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()
|
||
}
|
||
|
||
// regionNames 批量读取区域名称，只服务后台展示，不参与任何金额计算。
|
||
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()
|
||
}
|
||
|
||
// userProfiles 批量读取用户展示信息；缺失用户资料时上层会保留 user_id。
|
||
func (s *Service) userProfiles(ctx context.Context, appCode string, userIDs []int64) (map[int64]userProfile, error) {
|
||
userIDs = uniquePositiveUserIDs(userIDs)
|
||
if len(userIDs) == 0 {
|
||
return map[int64]userProfile{}, nil
|
||
}
|
||
args := make([]any, 0, len(userIDs)+1)
|
||
args = append(args, appCode)
|
||
for _, id := range userIDs {
|
||
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(len(userIDs))+`)`, args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
out := map[int64]userProfile{}
|
||
for rows.Next() {
|
||
var item userProfile
|
||
if err := rows.Scan(&item.UserID, &item.DisplayUserID, &item.Username, &item.Avatar); err != nil {
|
||
return nil, err
|
||
}
|
||
out[item.UserID] = item
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// ensureReady 做运行时自检并补齐 wallet 运行表；本地测试可以直接跑，线上仍以标准迁移为准。
|
||
func (s *Service) ensureReady(ctx context.Context) error {
|
||
if s == nil || s.adminDB == nil || s.walletDB == nil || s.userDB == nil {
|
||
return fmt.Errorf("team salary settlement dependencies are not configured")
|
||
}
|
||
// 运行表属于 wallet 账本库；admin-server 在本地和测试环境按需补表，线上仍应由标准 DDL 发布流程先执行 initdb/migration。
|
||
for _, ddl := range teamSalaryRuntimeDDL() {
|
||
if _, err := s.walletDB.ExecContext(ctx, ddl); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// teamSalaryRuntimeDDL 返回 wallet 侧团队工资运行表 DDL；保持与 wallet-service initdb 的结构一致。
|
||
func teamSalaryRuntimeDDL() []string {
|
||
return []string{
|
||
`CREATE TABLE IF NOT EXISTS team_salary_settlement_progress (
|
||
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu',
|
||
policy_type VARCHAR(24) NOT NULL,
|
||
user_id BIGINT NOT NULL,
|
||
cycle_key VARCHAR(16) NOT NULL,
|
||
region_id BIGINT NOT NULL,
|
||
settled_level_no INT NOT NULL DEFAULT 0,
|
||
settled_salary_usd_minor BIGINT NOT NULL DEFAULT 0,
|
||
last_settled_income_usd_minor BIGINT NOT NULL DEFAULT 0,
|
||
last_policy_id BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||
month_closed_at_ms BIGINT NOT NULL DEFAULT 0,
|
||
version BIGINT NOT NULL DEFAULT 1,
|
||
created_at_ms BIGINT NOT NULL,
|
||
updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, policy_type, user_id, cycle_key),
|
||
KEY idx_team_salary_progress_cycle (app_code, policy_type, cycle_key, region_id, month_closed_at_ms)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS team_salary_settlement_records (
|
||
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu',
|
||
settlement_id VARCHAR(128) NOT NULL,
|
||
command_id VARCHAR(128) NOT NULL,
|
||
transaction_id VARCHAR(96) NOT NULL,
|
||
policy_type VARCHAR(24) NOT NULL,
|
||
trigger_mode VARCHAR(24) NOT NULL DEFAULT 'manual',
|
||
user_id BIGINT NOT NULL,
|
||
cycle_key VARCHAR(16) NOT NULL,
|
||
region_id BIGINT NOT NULL,
|
||
policy_id BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||
level_no INT NOT NULL DEFAULT 0,
|
||
income_usd_minor BIGINT NOT NULL DEFAULT 0,
|
||
salary_usd_minor_delta BIGINT NOT NULL DEFAULT 0,
|
||
status VARCHAR(24) NOT NULL DEFAULT 'succeeded',
|
||
reason VARCHAR(255) NOT NULL DEFAULT '',
|
||
created_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, settlement_id),
|
||
UNIQUE KEY uk_team_salary_settlement_command (app_code, command_id),
|
||
KEY idx_team_salary_records_user_cycle (app_code, policy_type, user_id, cycle_key, created_at_ms),
|
||
KEY idx_team_salary_records_cycle (app_code, policy_type, cycle_key, region_id, created_at_ms)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
}
|
||
}
|
||
|
||
// pendingDTOFromCandidate 将内部候选转成前端 DTO，同时保留已结进度方便运营判断为什么只发差值。
|
||
func pendingDTOFromCandidate(item teamCandidate, profile userProfile) pendingDTO {
|
||
userID := strconv.FormatInt(item.UserID, 10)
|
||
return pendingDTO{
|
||
PolicyType: item.PolicyType,
|
||
UserID: userID,
|
||
User: userDTOFromProfile(userID, profile),
|
||
CycleKey: item.CycleKey,
|
||
RegionID: item.RegionID,
|
||
RegionName: item.RegionName,
|
||
PolicyID: item.Policy.PolicyID,
|
||
PolicyName: item.Policy.Name,
|
||
LevelNo: int32(item.Level.LevelNo),
|
||
IncomeUSDMinor: item.IncomeUSDMinor,
|
||
IncomeUSD: formatUSDMinor(item.IncomeUSDMinor),
|
||
SalaryUSDMinorTarget: item.Level.SalaryMinor,
|
||
SalaryUSDTarget: formatUSDMinor(item.Level.SalaryMinor),
|
||
SalaryUSDMinorDelta: item.SalaryUSDMinorDelta,
|
||
SalaryUSDDelta: formatUSDMinor(item.SalaryUSDMinorDelta),
|
||
SettledLevelNo: int32(item.Progress.SettledLevelNo),
|
||
SettledSalaryUSDMinor: item.Progress.SettledSalaryUSDMinor,
|
||
SettledSalaryUSD: formatUSDMinor(item.Progress.SettledSalaryUSDMinor),
|
||
LastSettledIncomeUSDMinor: item.Progress.LastSettledIncomeUSDMinor,
|
||
LastSettledIncomeUSD: formatUSDMinor(item.Progress.LastSettledIncomeUSDMinor),
|
||
MonthClosedAtMS: item.Progress.MonthClosedAtMS,
|
||
}
|
||
}
|
||
|
||
// normalizePendingQuery 收敛列表参数默认值；默认看上一个完整月，符合 BD/Admin 次月 5 日结算口径。
|
||
func normalizePendingQuery(req pendingQuery) pendingQuery {
|
||
req.PolicyType = normalizePolicyType(req.PolicyType)
|
||
req.TriggerMode = normalizeTriggerMode(req.TriggerMode)
|
||
req.CycleKey = strings.TrimSpace(req.CycleKey)
|
||
req.CountryCode = strings.ToUpper(strings.TrimSpace(req.CountryCode))
|
||
if req.CycleKey == "" {
|
||
req.CycleKey = previousMonthCycleKey(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
|
||
}
|
||
|
||
// normalizeRecordQuery 收敛记录查询参数，非法枚举由 handler 提前报错，这里只做兜底。
|
||
func normalizeRecordQuery(req recordQuery) recordQuery {
|
||
req.PolicyType = normalizePolicyType(req.PolicyType)
|
||
req.TriggerMode = normalizeTriggerMode(req.TriggerMode)
|
||
req.Status = normalizeStatus(req.Status)
|
||
req.CycleKey = strings.TrimSpace(req.CycleKey)
|
||
req.CountryCode = strings.ToUpper(strings.TrimSpace(req.CountryCode))
|
||
if req.Page <= 0 {
|
||
req.Page = 1
|
||
}
|
||
if req.PageSize <= 0 {
|
||
req.PageSize = 50
|
||
}
|
||
if req.PageSize > maxPageSize {
|
||
req.PageSize = maxPageSize
|
||
}
|
||
return req
|
||
}
|
||
|
||
// normalizePolicyType 只允许 bd/admin/all，未知值归空让 handler 能统一判断是否非法。
|
||
func normalizePolicyType(value string) string {
|
||
switch strings.ToLower(strings.TrimSpace(value)) {
|
||
case "", "all":
|
||
return ""
|
||
case policyTypeBD:
|
||
return policyTypeBD
|
||
case policyTypeAdmin:
|
||
return policyTypeAdmin
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
// normalizeTriggerMode 只允许 automatic/manual/all，防止手写字符串绕过政策触发方式匹配。
|
||
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 teamSalaryAssetType(policyType string) string {
|
||
// 用户可以同时拥有 BD/Admin 等多重身份，工资资产必须按本次结算角色拆分，不能混入其它身份钱包。
|
||
switch normalizePolicyType(policyType) {
|
||
case policyTypeBD:
|
||
return assetBDSalaryUSD
|
||
case policyTypeAdmin:
|
||
return assetAdminSalaryUSD
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
// normalizeStatus 只允许记录表定义的状态值，避免前端筛选拼接任意 SQL 条件。
|
||
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 ""
|
||
}
|
||
}
|
||
|
||
// highestTeamLevel 按收入基数找最高已达档；未达任何档时返回零值，后续差值自然为 0。
|
||
func highestTeamLevel(levels []teamPolicyLevel, incomeMinor int64) teamPolicyLevel {
|
||
var selected teamPolicyLevel
|
||
for _, level := range levels {
|
||
if level.ThresholdMinor <= incomeMinor {
|
||
selected = level
|
||
}
|
||
}
|
||
return selected
|
||
}
|
||
|
||
// usdStringToMinor 把 DECIMAL 字符串转成美分整数，所有工资差值都用整数计算。
|
||
func usdStringToMinor(value string) (int64, error) {
|
||
rat, ok := new(big.Rat).SetString(strings.TrimSpace(value))
|
||
if !ok || rat.Sign() < 0 {
|
||
return 0, fmt.Errorf("invalid usd amount: %s", value)
|
||
}
|
||
rat.Mul(rat, big.NewRat(100, 1))
|
||
minor := new(big.Int).Quo(rat.Num(), rat.Denom())
|
||
if !minor.IsInt64() || minor.Int64() > math.MaxInt64 {
|
||
return 0, fmt.Errorf("usd amount overflow: %s", value)
|
||
}
|
||
return minor.Int64(), nil
|
||
}
|
||
|
||
// previousMonthCycleKey 以 UTC 自然月为结算周期，和 wallet Host 周期键保持同一口径。
|
||
func previousMonthCycleKey(now time.Time) string {
|
||
return time.Date(now.UTC().Year(), now.UTC().Month(), 1, 0, 0, 0, 0, time.UTC).AddDate(0, -1, 0).Format("2006-01")
|
||
}
|
||
|
||
// teamSalaryCommandID 是团队工资业务幂等 ID，编码维度覆盖角色、触发方式、周期、用户、等级和收入基数。
|
||
func teamSalaryCommandID(candidate teamCandidate, triggerMode string, progress teamProgress) string {
|
||
return fmt.Sprintf("team_salary:%s:%s:%s:%d:%d:%d", candidate.PolicyType, triggerMode, candidate.CycleKey, candidate.UserID, progress.SettledLevelNo, candidate.IncomeUSDMinor)
|
||
}
|
||
|
||
// teamSalarySettlementID 从 commandID 派生稳定记录 ID，方便重复请求定位到同一业务动作。
|
||
func teamSalarySettlementID(appCode string, commandID string) string {
|
||
return fmt.Sprintf("tss_%s_%s", strings.TrimSpace(appCode), stableHash(commandID))
|
||
}
|
||
|
||
// teamSalaryTransactionID 从 commandID 派生 wallet 交易 ID，保证记录和钱包交易可双向关联。
|
||
func teamSalaryTransactionID(appCode string, commandID string) string {
|
||
return fmt.Sprintf("wtx_%s_%s", strings.TrimSpace(appCode), stableHash(commandID))
|
||
}
|
||
|
||
// stableHash 使用 SHA-256 压缩长幂等键，避免 MySQL varchar 索引过长。
|
||
func stableHash(value string) string {
|
||
sum := sha256.Sum256([]byte(value))
|
||
return hex.EncodeToString(sum[:])
|
||
}
|
||
|
||
// teamCandidateKey 把用户和区域组合成 map key，BD/Admin 同一用户跨区域时必须分别匹配政策。
|
||
func teamCandidateKey(userID int64, regionID int64) string {
|
||
return strconv.FormatInt(userID, 10) + ":" + strconv.FormatInt(regionID, 10)
|
||
}
|
||
|
||
// parseTeamCandidateKey 反解 teamCandidateKey；异常 key 返回零值，后续不会命中有效政策。
|
||
func parseTeamCandidateKey(key string) (int64, int64) {
|
||
parts := strings.Split(key, ":")
|
||
if len(parts) != 2 {
|
||
return 0, 0
|
||
}
|
||
userID, _ := strconv.ParseInt(parts[0], 10, 64)
|
||
regionID, _ := strconv.ParseInt(parts[1], 10, 64)
|
||
return userID, regionID
|
||
}
|
||
|
||
// sortTeamCandidates 让后台列表稳定：先角色、区域，再按收入从高到低，最后按用户 ID。
|
||
func sortTeamCandidates(items []teamCandidate) {
|
||
sort.Slice(items, func(i, j int) bool {
|
||
if items[i].PolicyType != items[j].PolicyType {
|
||
return items[i].PolicyType < items[j].PolicyType
|
||
}
|
||
if items[i].RegionID != items[j].RegionID {
|
||
return items[i].RegionID < items[j].RegionID
|
||
}
|
||
if items[i].IncomeUSDMinor != items[j].IncomeUSDMinor {
|
||
return items[i].IncomeUSDMinor > items[j].IncomeUSDMinor
|
||
}
|
||
return items[i].UserID < items[j].UserID
|
||
})
|
||
}
|
||
|
||
// paginateCandidates 在内存候选上分页；候选已被区域/国家/角色过滤，避免 SQL 里重复复杂聚合分页。
|
||
func paginateCandidates(items []teamCandidate, page int, pageSize int) []teamCandidate {
|
||
start := offset(page, pageSize)
|
||
if start >= len(items) {
|
||
return []teamCandidate{}
|
||
}
|
||
end := start + pageSize
|
||
if end > len(items) {
|
||
end = len(items)
|
||
}
|
||
return items[start:end]
|
||
}
|
||
|
||
// candidateUserIDs 提取候选用户 ID，用于一次性加载展示资料。
|
||
func candidateUserIDs(items []teamCandidate) []int64 {
|
||
out := make([]int64, 0, len(items))
|
||
for _, item := range items {
|
||
out = append(out, item.UserID)
|
||
}
|
||
return out
|
||
}
|
||
|
||
// keysInt64 将 map key 稳定排序，保证 SQL IN 参数顺序和测试输出稳定。
|
||
func keysInt64(values map[int64]int64) []int64 {
|
||
out := make([]int64, 0, len(values))
|
||
for key := range values {
|
||
out = append(out, key)
|
||
}
|
||
sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
|
||
return out
|
||
}
|
||
|
||
// keysInt64FromSet 将 set key 稳定排序，主要用于国家转区域后的记录查询条件。
|
||
func keysInt64FromSet(values map[int64]struct{}) []int64 {
|
||
out := make([]int64, 0, len(values))
|
||
for key := range values {
|
||
out = append(out, key)
|
||
}
|
||
sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
|
||
return out
|
||
}
|
||
|
||
// uniquePositiveUserIDs 去重并过滤无效用户 ID，避免动态 IN () 和重复 profile 查询。
|
||
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)
|
||
}
|
||
sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
|
||
return out
|
||
}
|
||
|
||
// userDTOFromProfile 在资料缺失时仍返回 fallback user_id，保证结算记录永远可识别收款用户。
|
||
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,
|
||
}
|
||
}
|
||
|
||
// formatUSDMinor 只负责后端 DTO 展示格式，账务计算不使用格式化字符串。
|
||
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)
|
||
}
|
||
|
||
// placeholders 生成固定数量的 SQL 占位符，调用方已保证 n > 0。
|
||
func placeholders(n int) string {
|
||
return strings.TrimRight(strings.Repeat("?,", n), ",")
|
||
}
|
||
|
||
// offset 把页码转换为 LIMIT offset；第一页和异常页码都回到 0。
|
||
func offset(page int, pageSize int) int {
|
||
if page <= 1 {
|
||
return 0
|
||
}
|
||
return (page - 1) * pageSize
|
||
}
|
||
|
||
// positiveDelta 是所有差值发薪的最终公式：目标累计 <= 已发累计时不再发钱。
|
||
func positiveDelta(target int64, current int64) int64 {
|
||
if target <= current {
|
||
return 0
|
||
}
|
||
return target - current
|
||
}
|
||
|
||
// maxInt 防止进度等级被旧候选覆盖回退。
|
||
func maxInt(left int, right int) int {
|
||
if right > left {
|
||
return right
|
||
}
|
||
return left
|
||
}
|
||
|
||
// maxInt64 防止已发金额或收入基数被旧候选覆盖回退。
|
||
func maxInt64(left int64, right int64) int64 {
|
||
if right > left {
|
||
return right
|
||
}
|
||
return left
|
||
}
|