558 lines
18 KiB
Go
558 lines
18 KiB
Go
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
|
||
}
|