1184 lines
39 KiB
Go
1184 lines
39 KiB
Go
package coinledger
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"database/sql"
|
||
"encoding/csv"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
"unicode/utf8"
|
||
|
||
"hyapp-admin-server/internal/integration/walletclient"
|
||
"hyapp-admin-server/internal/modules/shared"
|
||
walletv1 "hyapp.local/api/proto/wallet/v1"
|
||
)
|
||
|
||
const (
|
||
coinAssetType = "COIN"
|
||
coinSellerAssetType = "COIN_SELLER_COIN"
|
||
coinManualCreditBizType = "manual_credit"
|
||
coinSellerTransferBizType = "coin_seller_transfer"
|
||
coinSellerSubTransferBizType = "coin_seller_sub_transfer"
|
||
coinSellerRechargeBizType = "coin_seller_recharge"
|
||
coinSellerStockPurchaseBizType = "coin_seller_stock_purchase"
|
||
coinSellerStockDeductionBizType = "coin_seller_stock_deduction"
|
||
coinSellerCoinCompensationBizType = "coin_seller_coin_compensation"
|
||
salaryTransferToCoinSellerBizType = "salary_transfer_to_coin_seller"
|
||
coinSellerLedgerTypeAdminStockCredit = "admin_stock_credit"
|
||
coinSellerLedgerTypeSellerTransfer = "seller_transfer"
|
||
coinSellerLedgerTypeSubSellerTransfer = "sub_seller_transfer"
|
||
coinSellerLedgerTypeSalaryTransferIncome = "salary_transfer_received"
|
||
directionIn = "income"
|
||
directionOut = "expense"
|
||
)
|
||
|
||
var (
|
||
errCoinAdjustmentTargetNotFound = errors.New("coin adjustment target user not found")
|
||
errInvalidCoinSellerLedgerType = errors.New("coin seller ledger type is invalid")
|
||
)
|
||
|
||
type Service struct {
|
||
userDB *sql.DB
|
||
walletDB *sql.DB
|
||
adminDB *sql.DB
|
||
walletClient walletclient.Client
|
||
}
|
||
|
||
type userProfile struct {
|
||
UserID int64
|
||
DisplayUserID string
|
||
Username string
|
||
Avatar string
|
||
}
|
||
|
||
func NewService(userDB *sql.DB, walletDB *sql.DB, adminDB *sql.DB, walletClient walletclient.Client) *Service {
|
||
return &Service{userDB: userDB, walletDB: walletDB, adminDB: adminDB, walletClient: walletClient}
|
||
}
|
||
|
||
func (s *Service) ListCoinLedger(ctx context.Context, appCode string, query listQuery) ([]coinLedgerEntryDTO, int64, error) {
|
||
query = normalizeListQuery(query)
|
||
if s == nil || s.walletDB == nil {
|
||
return nil, 0, fmt.Errorf("wallet mysql is not configured")
|
||
}
|
||
|
||
userIDs, userFiltered, err := s.resolveUserFilter(ctx, appCode, query.UserKeyword, query.UserFilter)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
if userFiltered && len(userIDs) == 0 {
|
||
return []coinLedgerEntryDTO{}, 0, nil
|
||
}
|
||
|
||
// 后台只做只读查询：金币流水事实仍以 wallet_entries 追加分录为准，不在 admin 库冗余账务状态。
|
||
// 普通打开列表时不带 biz_type，total 只依赖 entries 自身；避免为了一个总数对百万级 COIN 分录逐行回表 join 交易主表。
|
||
whereSQL, args := coinLedgerWhere(appCode, query, userIDs)
|
||
var total int64
|
||
if query.BizType == "" {
|
||
if err := s.walletDB.QueryRowContext(ctx, `
|
||
SELECT COUNT(*)
|
||
FROM wallet_entries e
|
||
`+whereSQL,
|
||
args...,
|
||
).Scan(&total); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
} else {
|
||
if err := s.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 := s.walletDB.QueryContext(ctx, `
|
||
SELECT e.entry_id, e.transaction_id, wt.command_id, wt.external_ref, e.user_id, wt.biz_type,
|
||
e.available_delta, e.available_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([]coinLedgerEntryDTO, 0, query.PageSize)
|
||
itemUserIDs := make([]int64, 0, query.PageSize)
|
||
for rows.Next() {
|
||
var item coinLedgerEntryDTO
|
||
var userID int64
|
||
var counterpartyUserID int64
|
||
var metadataJSON string
|
||
if err := rows.Scan(
|
||
&item.EntryID,
|
||
&item.TransactionID,
|
||
&item.CommandID,
|
||
&item.ExternalRef,
|
||
&userID,
|
||
&item.BizType,
|
||
&item.AvailableDelta,
|
||
&item.AvailableAfter,
|
||
&counterpartyUserID,
|
||
&item.RoomID,
|
||
&metadataJSON,
|
||
&item.CreatedAtMS,
|
||
); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
metadata, err := parseMetadataJSON(metadataJSON)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
item.UserID = strconv.FormatInt(userID, 10)
|
||
item.User = coinLedgerUserDTO{UserID: item.UserID}
|
||
item.CounterpartyUserID = formatOptionalID(counterpartyUserID)
|
||
item.Metadata = metadata
|
||
item.Direction = directionForDelta(item.AvailableDelta)
|
||
item.Amount = absInt64(item.AvailableDelta)
|
||
items = append(items, item)
|
||
itemUserIDs = append(itemUserIDs, userID)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
|
||
profiles, err := s.userProfiles(ctx, appCode, itemUserIDs)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
for i := range items {
|
||
userID, _ := strconv.ParseInt(items[i].UserID, 10, 64)
|
||
if profile, ok := profiles[userID]; ok {
|
||
items[i].User = coinLedgerUserDTO{
|
||
UserID: strconv.FormatInt(profile.UserID, 10),
|
||
DisplayUserID: profile.DisplayUserID,
|
||
Username: profile.Username,
|
||
Avatar: profile.Avatar,
|
||
}
|
||
}
|
||
}
|
||
return items, total, nil
|
||
}
|
||
|
||
func (s *Service) ListCoinSellerLedger(ctx context.Context, appCode string, query coinSellerLedgerQuery) ([]coinSellerLedgerDTO, int64, error) {
|
||
return s.listCoinSellerLedger(ctx, appCode, query, true)
|
||
}
|
||
|
||
func (s *Service) ExportCoinSellerLedger(ctx context.Context, appCode string, query coinSellerLedgerQuery) (CSVExport, error) {
|
||
items, total, err := s.listCoinSellerLedger(ctx, appCode, query, false)
|
||
if err != nil {
|
||
return CSVExport{}, err
|
||
}
|
||
var buf bytes.Buffer
|
||
writer := csv.NewWriter(&buf)
|
||
_ = writer.Write([]string{"币商名称", "币商短ID", "币商长ID", "流水类型", "金币变动", "转账USD金额", "USDT数量", "收款人名称", "收款人短ID", "收款人长ID", "操作人", "操作人ID", "币商余额", "转账时间", "交易ID", "命令ID"})
|
||
for _, item := range items {
|
||
operatorName, operatorID := coinSellerLedgerOperatorExportFields(item)
|
||
_ = writer.Write([]string{
|
||
item.Seller.Username,
|
||
item.Seller.DisplayUserID,
|
||
item.Seller.UserID,
|
||
coinSellerLedgerLabel(item),
|
||
strconv.FormatInt(signedCoinSellerLedgerAmount(item), 10),
|
||
formatUSDMinorForCSV(item.TransferUSDMinor),
|
||
formatUSDTMicroForCSV(item.PaidAmountMicro),
|
||
item.Receiver.Username,
|
||
item.Receiver.DisplayUserID,
|
||
item.Receiver.UserID,
|
||
operatorName,
|
||
operatorID,
|
||
strconv.FormatInt(item.SellerBalanceAfter, 10),
|
||
time.UnixMilli(item.CreatedAtMS).UTC().Format(time.RFC3339),
|
||
item.TransactionID,
|
||
item.CommandID,
|
||
})
|
||
}
|
||
writer.Flush()
|
||
return CSVExport{FileName: "hyapp-coin-seller-ledger.csv", Content: buf.Bytes(), Count: int(total)}, writer.Error()
|
||
}
|
||
|
||
func (s *Service) listCoinSellerLedger(ctx context.Context, appCode string, query coinSellerLedgerQuery, paginated bool) ([]coinSellerLedgerDTO, int64, error) {
|
||
query = normalizeCoinSellerLedgerQuery(query)
|
||
if s == nil || s.walletDB == nil {
|
||
return nil, 0, fmt.Errorf("wallet mysql is not configured")
|
||
}
|
||
|
||
// 先把前端筛选解析成币商 user_id 集合，再进入 wallet_entries 查询；
|
||
// 这样 seller_keyword 只影响币商本人范围，不会因为收款用户昵称或短 ID 命中而把其他币商流水带出来。
|
||
sellerIDs, sellerFiltered, err := s.resolveCoinSellerFilter(ctx, appCode, query)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
if sellerFiltered && len(sellerIDs) == 0 {
|
||
// 用户库已经确认没有匹配币商时直接返回空分页，避免对 wallet 库做没有结果意义的全量账本扫描。
|
||
return []coinSellerLedgerDTO{}, 0, nil
|
||
}
|
||
|
||
// 币商流水只读币商专用金币分录；wallet-service 仍是账务事实 owner，后台不回写余额、不补账、不改变交易状态，
|
||
// 这里只按运营筛选把 wallet_entries 与 wallet_transactions 的事实组装成后台展示投影。
|
||
whereSQL, args, err := coinSellerLedgerWhere(appCode, query, sellerIDs)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
var total int64
|
||
if paginated {
|
||
if err := s.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
|
||
}
|
||
}
|
||
|
||
// 明细查询只取币商侧 COIN_SELLER_COIN 分录：available_delta 表示币商库存可用变化，
|
||
// available_after 表示该币商分录落账后的库存余额，receiver 再按 biz_type 从 metadata/counterparty 补出。
|
||
querySQL := `
|
||
SELECT e.entry_id, e.transaction_id, wt.command_id, e.user_id, wt.biz_type,
|
||
e.available_delta, e.available_after, e.counterparty_user_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`
|
||
queryArgs := args
|
||
if paginated {
|
||
querySQL += `
|
||
LIMIT ? OFFSET ?`
|
||
queryArgs = append(queryArgs, query.PageSize, offset(query.Page, query.PageSize))
|
||
}
|
||
rows, err := s.walletDB.QueryContext(ctx, querySQL, queryArgs...)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
items := make([]coinSellerLedgerDTO, 0, query.PageSize)
|
||
profileIDs := make([]int64, 0, query.PageSize*2)
|
||
operatorIDs := make([]int64, 0, query.PageSize)
|
||
for rows.Next() {
|
||
var item coinSellerLedgerDTO
|
||
var sellerUserID int64
|
||
var counterpartyUserID int64
|
||
var metadataJSON string
|
||
if err := rows.Scan(
|
||
&item.EntryID,
|
||
&item.TransactionID,
|
||
&item.CommandID,
|
||
&sellerUserID,
|
||
&item.BizType,
|
||
&item.AvailableDelta,
|
||
&item.SellerBalanceAfter,
|
||
&counterpartyUserID,
|
||
&metadataJSON,
|
||
&item.CreatedAtMS,
|
||
); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
metadata, err := parseMetadataJSON(metadataJSON)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
operatorUserID := metadataInt64(metadata, "operator_user_id", "operatorUserId")
|
||
item.StockType = firstNonEmpty(metadataString(metadata, "stock_type"), metadataString(metadata, "stockType"))
|
||
item.PaidCurrencyCode = firstNonEmpty(metadataString(metadata, "paid_currency_code"), metadataString(metadata, "paidCurrencyCode"))
|
||
item.PaidAmountMicro = metadataInt64(metadata, "paid_amount_micro", "paidAmountMicro")
|
||
item.TransferUSDMinor = metadataInt64(metadata, "salary_usd_minor", "salaryUsdMinor", "transfer_usd_minor", "transferUsdMinor")
|
||
item.Reason = metadataString(metadata, "reason")
|
||
// 收款人按产品口径取实际收款人：币商转用户优先取 metadata.target_user_id，老数据没有 metadata 时再用 counterparty_user_id；
|
||
// 后台入账和工资转币商的实际收款人都是币商本人，所以不能被 counterparty 字段误导成付款方或操作方。
|
||
receiverUserID := coinSellerLedgerReceiverUserID(item.BizType, sellerUserID, counterpartyUserID, metadata)
|
||
item.LedgerType = coinSellerLedgerTypeForBizType(item.BizType)
|
||
item.Direction = directionForDelta(item.AvailableDelta)
|
||
item.Amount = absInt64(item.AvailableDelta)
|
||
item.CounterpartyUserID = formatOptionalID(counterpartyUserID)
|
||
item.OperatorUserID = formatOptionalID(operatorUserID)
|
||
item.Metadata = metadata
|
||
item.Seller = coinLedgerUserDTO{UserID: strconv.FormatInt(sellerUserID, 10)}
|
||
item.Receiver = coinLedgerUserDTO{UserID: strconv.FormatInt(receiverUserID, 10)}
|
||
items = append(items, item)
|
||
profileIDs = append(profileIDs, sellerUserID, receiverUserID)
|
||
if operatorUserID > 0 && item.LedgerType == coinSellerLedgerTypeAdminStockCredit {
|
||
operatorIDs = append(operatorIDs, operatorUserID)
|
||
}
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
if !paginated {
|
||
total = int64(len(items))
|
||
}
|
||
|
||
profiles, err := s.userProfiles(ctx, appCode, profileIDs)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
operators, err := s.adminProfiles(ctx, operatorIDs)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
for i := range items {
|
||
sellerUserID, _ := strconv.ParseInt(items[i].Seller.UserID, 10, 64)
|
||
receiverUserID, _ := strconv.ParseInt(items[i].Receiver.UserID, 10, 64)
|
||
// 用户资料只做头像、短 ID 和昵称展示补全；资料缺失时仍返回账本里的 user_id，
|
||
// 避免用户资料迁移、删除或延迟同步把真实账务事实从后台列表里吞掉。
|
||
if profile, ok := profiles[sellerUserID]; ok {
|
||
items[i].Seller = userDTOFromProfile(profile)
|
||
}
|
||
if profile, ok := profiles[receiverUserID]; ok {
|
||
items[i].Receiver = userDTOFromProfile(profile)
|
||
}
|
||
operatorUserID, _ := strconv.ParseInt(items[i].OperatorUserID, 10, 64)
|
||
if operator, ok := operators[operatorUserID]; ok {
|
||
items[i].Operator = operator
|
||
}
|
||
}
|
||
return items, total, nil
|
||
}
|
||
|
||
func (s *Service) ListCoinAdjustments(ctx context.Context, appCode string, query listQuery) ([]coinAdjustmentDTO, int64, error) {
|
||
query = normalizeListQuery(query)
|
||
if s == nil || s.walletDB == nil {
|
||
return nil, 0, fmt.Errorf("wallet mysql is not configured")
|
||
}
|
||
|
||
userIDs, userFiltered, err := s.resolveUserFilter(ctx, appCode, query.UserKeyword, query.UserFilter)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
if userFiltered && len(userIDs) == 0 {
|
||
return []coinAdjustmentDTO{}, 0, nil
|
||
}
|
||
|
||
// 金币增减列表只读 wallet-service 账本事实；后台不维护第二份余额或调账流水。
|
||
whereSQL, args := coinAdjustmentWhere(appCode, query, userIDs)
|
||
var total int64
|
||
if err := s.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 := s.walletDB.QueryContext(ctx, `
|
||
SELECT e.entry_id, e.transaction_id, wt.command_id, e.user_id,
|
||
e.available_delta, e.available_after, 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([]coinAdjustmentDTO, 0, query.PageSize)
|
||
itemUserIDs := make([]int64, 0, query.PageSize)
|
||
operatorIDs := make([]int64, 0, query.PageSize)
|
||
for rows.Next() {
|
||
var item coinAdjustmentDTO
|
||
var userID int64
|
||
var metadataJSON string
|
||
if err := rows.Scan(
|
||
&item.EntryID,
|
||
&item.TransactionID,
|
||
&item.CommandID,
|
||
&userID,
|
||
&item.AvailableDelta,
|
||
&item.AvailableAfter,
|
||
&metadataJSON,
|
||
&item.CreatedAtMS,
|
||
); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
metadata, err := parseMetadataJSON(metadataJSON)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
operatorID := metadataInt64(metadata, "operator_user_id", "operatorUserId")
|
||
item.UserID = strconv.FormatInt(userID, 10)
|
||
item.User = coinLedgerUserDTO{UserID: item.UserID}
|
||
item.Direction = directionForDelta(item.AvailableDelta)
|
||
item.Amount = item.AvailableDelta
|
||
item.Reason = metadataString(metadata, "reason")
|
||
item.Remark = metadataString(metadata, "remark")
|
||
item.OperatorUserID = formatOptionalID(operatorID)
|
||
items = append(items, item)
|
||
itemUserIDs = append(itemUserIDs, userID)
|
||
if operatorID > 0 {
|
||
operatorIDs = append(operatorIDs, operatorID)
|
||
}
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
|
||
profiles, err := s.userProfiles(ctx, appCode, itemUserIDs)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
operators, err := s.adminProfiles(ctx, operatorIDs)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
for i := range items {
|
||
userID, _ := strconv.ParseInt(items[i].UserID, 10, 64)
|
||
if profile, ok := profiles[userID]; ok {
|
||
items[i].User = coinLedgerUserDTO{
|
||
UserID: strconv.FormatInt(profile.UserID, 10),
|
||
DisplayUserID: profile.DisplayUserID,
|
||
Username: profile.Username,
|
||
Avatar: profile.Avatar,
|
||
}
|
||
}
|
||
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) LookupCoinAdjustmentTarget(ctx context.Context, appCode string, keyword string) (coinLedgerUserDTO, error) {
|
||
keyword = strings.TrimSpace(keyword)
|
||
if keyword == "" {
|
||
return coinLedgerUserDTO{}, fmt.Errorf("user_id is required")
|
||
}
|
||
if s == nil || s.userDB == nil {
|
||
return coinLedgerUserDTO{}, fmt.Errorf("user mysql is not configured")
|
||
}
|
||
nowMs := time.Now().UnixMilli()
|
||
matchSQL, matchArgs := shared.UserIdentityExactSQL("u", "u.user_id", keyword, nowMs)
|
||
orderSQL, orderArgs := shared.UserIdentityExactOrderSQL("u", "u.user_id", keyword, nowMs)
|
||
args := append([]any{appCode}, matchArgs...)
|
||
args = append(args, orderArgs...)
|
||
row := s.userDB.QueryRowContext(ctx, `
|
||
SELECT u.user_id, u.current_display_user_id, COALESCE(u.username, ''), COALESCE(u.avatar, '')
|
||
FROM users u
|
||
WHERE u.app_code = ? AND `+matchSQL+`
|
||
ORDER BY
|
||
`+orderSQL+`,
|
||
u.user_id DESC
|
||
LIMIT 1`,
|
||
args...,
|
||
)
|
||
var profile userProfile
|
||
if err := row.Scan(&profile.UserID, &profile.DisplayUserID, &profile.Username, &profile.Avatar); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return coinLedgerUserDTO{}, errCoinAdjustmentTargetNotFound
|
||
}
|
||
return coinLedgerUserDTO{}, err
|
||
}
|
||
return coinLedgerUserDTO{
|
||
UserID: strconv.FormatInt(profile.UserID, 10),
|
||
DisplayUserID: profile.DisplayUserID,
|
||
Username: profile.Username,
|
||
Avatar: profile.Avatar,
|
||
}, nil
|
||
}
|
||
|
||
func (s *Service) CreateCoinAdjustment(ctx context.Context, appCode string, actorID int64, requestID string, req coinAdjustmentRequest) (coinAdjustmentCreateDTO, error) {
|
||
if s == nil || s.walletClient == nil {
|
||
return coinAdjustmentCreateDTO{}, fmt.Errorf("wallet client is not configured")
|
||
}
|
||
targetUserID, err := parseFlexibleUserID(req.TargetUserID)
|
||
if err != nil {
|
||
return coinAdjustmentCreateDTO{}, err
|
||
}
|
||
if actorID <= 0 || targetUserID <= 0 || req.Amount == 0 {
|
||
return coinAdjustmentCreateDTO{}, fmt.Errorf("target_user_id, operator_user_id and amount are required")
|
||
}
|
||
reason := strings.TrimSpace(req.Reason)
|
||
if reason == "" {
|
||
return coinAdjustmentCreateDTO{}, fmt.Errorf("reason is required")
|
||
}
|
||
if utf8.RuneCountInString(reason) > 512 {
|
||
return coinAdjustmentCreateDTO{}, fmt.Errorf("reason is too long")
|
||
}
|
||
remark := strings.TrimSpace(req.Remark)
|
||
if remark == "" {
|
||
return coinAdjustmentCreateDTO{}, fmt.Errorf("remark is required")
|
||
}
|
||
if utf8.RuneCountInString(remark) > 512 {
|
||
return coinAdjustmentCreateDTO{}, fmt.Errorf("remark is too long")
|
||
}
|
||
user, err := s.LookupCoinAdjustmentTarget(ctx, appCode, strconv.FormatInt(targetUserID, 10))
|
||
if err != nil {
|
||
return coinAdjustmentCreateDTO{}, err
|
||
}
|
||
commandID := strings.TrimSpace(req.CommandID)
|
||
if commandID == "" {
|
||
commandID = defaultCoinAdjustmentCommandID(requestID)
|
||
}
|
||
if len(commandID) > 128 {
|
||
return coinAdjustmentCreateDTO{}, fmt.Errorf("command_id is too long")
|
||
}
|
||
evidenceRef := defaultCoinAdjustmentEvidenceRef(requestID)
|
||
resp, err := s.walletClient.AdminCreditAsset(ctx, &walletv1.AdminCreditAssetRequest{
|
||
CommandId: commandID,
|
||
TargetUserId: targetUserID,
|
||
AssetType: coinAssetType,
|
||
Amount: req.Amount,
|
||
OperatorUserId: actorID,
|
||
Reason: reason,
|
||
EvidenceRef: evidenceRef,
|
||
AppCode: appCode,
|
||
Remark: remark,
|
||
})
|
||
if err != nil {
|
||
return coinAdjustmentCreateDTO{}, err
|
||
}
|
||
balanceAfter := int64(0)
|
||
if resp.GetBalance() != nil {
|
||
balanceAfter = resp.GetBalance().GetAvailableAmount()
|
||
}
|
||
return coinAdjustmentCreateDTO{
|
||
TransactionID: resp.GetTransactionId(),
|
||
BalanceAfter: balanceAfter,
|
||
User: user,
|
||
Amount: req.Amount,
|
||
AvailableDelta: req.Amount,
|
||
Reason: reason,
|
||
Remark: remark,
|
||
}, nil
|
||
}
|
||
|
||
func (s *Service) resolveUserFilter(ctx context.Context, appCode string, keyword string, filter shared.UserIdentityFilter) ([]int64, bool, error) {
|
||
keyword = strings.TrimSpace(keyword)
|
||
if filter.IsEmpty() && keyword == "" {
|
||
return nil, false, nil
|
||
}
|
||
directUserIDs := make([]int64, 0, 1)
|
||
numericKeyword := keyword
|
||
if strings.TrimSpace(filter.UserID) != "" {
|
||
numericKeyword = filter.UserID
|
||
}
|
||
if numeric, err := strconv.ParseInt(numericKeyword, 10, 64); err == nil && numeric > 0 {
|
||
directUserIDs = append(directUserIDs, numeric)
|
||
}
|
||
if s == nil || s.userDB == nil {
|
||
if len(directUserIDs) > 0 {
|
||
return directUserIDs, true, nil
|
||
}
|
||
return nil, true, fmt.Errorf("user mysql is not configured")
|
||
}
|
||
|
||
var matchSQL string
|
||
var matchArgs []any
|
||
if !filter.IsEmpty() {
|
||
// 拆分字段来自表头明确输入：短 ID、长 ID 和昵称分别构造精确约束，避免数字短号被昵称或长 ID LIKE 扩大。
|
||
matchSQL, matchArgs = shared.UserIdentityFieldsSQL("u", "u.user_id", filter, time.Now().UnixMilli())
|
||
} else {
|
||
// 旧 keyword 入口保留兼容；数字关键字直接加入 user_id，避免用户资料缺失时查不到账务事实。
|
||
matchSQL, matchArgs = shared.UserIdentityExactSQL("u", "u.user_id", keyword, time.Now().UnixMilli())
|
||
}
|
||
args := append([]any{appCode}, matchArgs...)
|
||
|
||
rows, err := s.userDB.QueryContext(ctx, `
|
||
SELECT u.user_id
|
||
FROM users u
|
||
WHERE u.app_code = ? AND `+matchSQL+`
|
||
ORDER BY u.updated_at_ms DESC, u.user_id DESC
|
||
LIMIT 1000`,
|
||
args...,
|
||
)
|
||
if err != nil {
|
||
return nil, true, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
userIDs := append([]int64(nil), directUserIDs...)
|
||
for rows.Next() {
|
||
var userID int64
|
||
if err := rows.Scan(&userID); err != nil {
|
||
return nil, true, err
|
||
}
|
||
userIDs = append(userIDs, userID)
|
||
}
|
||
return uniqueInt64s(userIDs), true, rows.Err()
|
||
}
|
||
|
||
func (s *Service) resolveCoinSellerFilter(ctx context.Context, appCode string, query coinSellerLedgerQuery) ([]int64, bool, error) {
|
||
if query.SellerUserID > 0 {
|
||
// seller_user_id 是 Coin Saller 行按钮传入的精确条件，直接使用账本 user_id 过滤；
|
||
// 即使用户资料暂时查不到，也允许账本事实被查出来，避免抽屉入口被资料缺失阻断。
|
||
return []int64{query.SellerUserID}, true, nil
|
||
}
|
||
if !query.SellerFilter.IsEmpty() {
|
||
if s == nil || s.userDB == nil {
|
||
return nil, true, fmt.Errorf("user mysql is not configured")
|
||
}
|
||
matchSQL, matchArgs := shared.UserIdentityFieldsSQL("u", "csp.user_id", query.SellerFilter, time.Now().UnixMilli())
|
||
return s.resolveCoinSellerIDs(ctx, appCode, matchSQL, matchArgs)
|
||
}
|
||
keyword := strings.TrimSpace(query.SellerKeyword)
|
||
if keyword == "" {
|
||
// 未传币商筛选时返回 userFiltered=false，调用方会查询所有币商流水，但仍被 biz_type、asset_type 和分页限制住。
|
||
return nil, false, nil
|
||
}
|
||
if s == nil || s.userDB == nil {
|
||
return nil, true, fmt.Errorf("user mysql is not configured")
|
||
}
|
||
|
||
// 币商关键字只命中 coin_seller_profiles 里的币商本人，LEFT JOIN users 只用于短 ID、active 靓号和昵称补充；
|
||
// 这里故意不搜索 receiver/counterparty，避免“收款用户命中”把其他币商的流水带进当前币商筛选结果。
|
||
matchSQL, matchArgs := shared.UserIdentityExactSQL("u", "csp.user_id", keyword, time.Now().UnixMilli())
|
||
|
||
return s.resolveCoinSellerIDs(ctx, appCode, matchSQL, matchArgs)
|
||
}
|
||
|
||
func (s *Service) resolveCoinSellerIDs(ctx context.Context, appCode string, matchSQL string, matchArgs []any) ([]int64, bool, error) {
|
||
args := append([]any{appCode}, matchArgs...)
|
||
rows, err := s.userDB.QueryContext(ctx, `
|
||
SELECT csp.user_id
|
||
FROM coin_seller_profiles csp
|
||
LEFT JOIN users u ON u.app_code = csp.app_code AND u.user_id = csp.user_id
|
||
WHERE csp.app_code = ? AND `+matchSQL+`
|
||
ORDER BY csp.updated_at_ms DESC, csp.user_id DESC
|
||
LIMIT 1000`,
|
||
args...,
|
||
)
|
||
if err != nil {
|
||
return nil, true, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
userIDs := make([]int64, 0)
|
||
for rows.Next() {
|
||
var userID int64
|
||
if err := rows.Scan(&userID); err != nil {
|
||
return nil, true, err
|
||
}
|
||
userIDs = append(userIDs, userID)
|
||
}
|
||
return uniqueInt64s(userIDs), true, rows.Err()
|
||
}
|
||
|
||
func (s *Service) userProfiles(ctx context.Context, appCode string, userIDs []int64) (map[int64]userProfile, error) {
|
||
result := make(map[int64]userProfile, len(userIDs))
|
||
if s == nil || s.userDB == nil || len(userIDs) == 0 {
|
||
return result, nil
|
||
}
|
||
userIDs = uniqueInt64s(userIDs)
|
||
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, fmt.Sprintf(`
|
||
SELECT user_id, current_display_user_id, COALESCE(username, ''), COALESCE(avatar, '')
|
||
FROM users
|
||
WHERE app_code = ? AND user_id IN (%s)
|
||
`, placeholders(len(userIDs))), args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
for rows.Next() {
|
||
var profile userProfile
|
||
if err := rows.Scan(&profile.UserID, &profile.DisplayUserID, &profile.Username, &profile.Avatar); err != nil {
|
||
return nil, err
|
||
}
|
||
result[profile.UserID] = profile
|
||
}
|
||
return result, rows.Err()
|
||
}
|
||
|
||
func userDTOFromProfile(profile userProfile) coinLedgerUserDTO {
|
||
return coinLedgerUserDTO{
|
||
UserID: strconv.FormatInt(profile.UserID, 10),
|
||
DisplayUserID: profile.DisplayUserID,
|
||
Username: profile.Username,
|
||
Avatar: profile.Avatar,
|
||
}
|
||
}
|
||
|
||
func coinLedgerWhere(appCode string, query listQuery, userIDs []int64) (string, []any) {
|
||
where := "WHERE e.app_code = ? AND e.asset_type = ?"
|
||
args := []any{appCode, coinAssetType}
|
||
if query.BizType != "" {
|
||
// 普通金币流水的类型就是 wallet_transactions.biz_type；这里做精确参数匹配，
|
||
// 不把用户输入拼进 SQL，保证新增业务类型也能通过同一个查询入口按真实账务类型定位。
|
||
where += " AND wt.biz_type = ?"
|
||
args = append(args, query.BizType)
|
||
}
|
||
if query.StartAtMS > 0 {
|
||
where += " AND e.created_at_ms >= ?"
|
||
args = append(args, query.StartAtMS)
|
||
}
|
||
if query.EndAtMS > 0 {
|
||
where += " AND e.created_at_ms < ?"
|
||
args = append(args, query.EndAtMS)
|
||
}
|
||
if len(userIDs) > 0 {
|
||
where += fmt.Sprintf(" AND e.user_id IN (%s)", placeholders(len(userIDs)))
|
||
for _, id := range userIDs {
|
||
args = append(args, id)
|
||
}
|
||
}
|
||
return where, args
|
||
}
|
||
|
||
func coinSellerLedgerWhere(appCode string, query coinSellerLedgerQuery, sellerIDs []int64) (string, []any, error) {
|
||
// 所有币商流水都必须落在当前 app_code 和 COIN_SELLER_COIN 资产上；
|
||
// 普通 COIN 分录属于用户金币流水，不能混入币商库存流水展示。
|
||
where := "WHERE e.app_code = ? AND e.asset_type = ?"
|
||
args := []any{appCode, coinSellerAssetType}
|
||
bizTypes, err := coinSellerLedgerBizTypes(query.LedgerType)
|
||
if err != nil {
|
||
return "", nil, err
|
||
}
|
||
if len(bizTypes) == 1 {
|
||
// 单一类型用等值条件，保持 SQL 简洁，也方便后续索引按 biz_type 命中。
|
||
where += " AND wt.biz_type = ?"
|
||
args = append(args, bizTypes[0])
|
||
} else {
|
||
// 后台入账和“全部”都可能映射多个底层 biz_type，只展开白名单占位符，不拼接用户输入。
|
||
where += fmt.Sprintf(" AND wt.biz_type IN (%s)", placeholders(len(bizTypes)))
|
||
for _, bizType := range bizTypes {
|
||
args = append(args, bizType)
|
||
}
|
||
}
|
||
if query.StartAtMS > 0 {
|
||
// 开始边界包含，和所有后台列表的 [start_at_ms, end_at_ms) 约定保持一致。
|
||
where += " AND e.created_at_ms >= ?"
|
||
args = append(args, query.StartAtMS)
|
||
}
|
||
if query.EndAtMS > 0 {
|
||
// 结束边界不包含，避免按小时/天连续筛选时重复命中边界毫秒的同一条账本分录。
|
||
where += " AND e.created_at_ms < ?"
|
||
args = append(args, query.EndAtMS)
|
||
}
|
||
if len(sellerIDs) > 0 {
|
||
// sellerIDs 来自精确 seller_user_id 或 coin_seller_profiles 搜索结果，只用于限制币商分录的 e.user_id。
|
||
where += fmt.Sprintf(" AND e.user_id IN (%s)", placeholders(len(sellerIDs)))
|
||
for _, id := range sellerIDs {
|
||
args = append(args, id)
|
||
}
|
||
}
|
||
return where, args, nil
|
||
}
|
||
|
||
func coinAdjustmentWhere(appCode string, query listQuery, userIDs []int64) (string, []any) {
|
||
where := "WHERE e.app_code = ? AND e.asset_type = ? AND wt.biz_type = ?"
|
||
args := []any{appCode, coinAssetType, coinManualCreditBizType}
|
||
if query.StartAtMS > 0 {
|
||
where += " AND e.created_at_ms >= ?"
|
||
args = append(args, query.StartAtMS)
|
||
}
|
||
if query.EndAtMS > 0 {
|
||
where += " AND e.created_at_ms < ?"
|
||
args = append(args, query.EndAtMS)
|
||
}
|
||
if len(userIDs) > 0 {
|
||
where += fmt.Sprintf(" AND e.user_id IN (%s)", placeholders(len(userIDs)))
|
||
for _, id := range userIDs {
|
||
args = append(args, id)
|
||
}
|
||
}
|
||
return where, args
|
||
}
|
||
|
||
func normalizeListQuery(query listQuery) listQuery {
|
||
if query.Page < 1 {
|
||
query.Page = 1
|
||
}
|
||
if query.PageSize < 1 {
|
||
query.PageSize = 20
|
||
}
|
||
if query.PageSize > 100 {
|
||
query.PageSize = 100
|
||
}
|
||
query.UserKeyword = strings.TrimSpace(query.UserKeyword)
|
||
query.UserFilter.UserID = strings.TrimSpace(query.UserFilter.UserID)
|
||
query.UserFilter.DisplayUserID = strings.TrimSpace(query.UserFilter.DisplayUserID)
|
||
query.UserFilter.Username = strings.TrimSpace(query.UserFilter.Username)
|
||
query.BizType = strings.TrimSpace(query.BizType)
|
||
return query
|
||
}
|
||
|
||
func normalizeCoinSellerLedgerQuery(query coinSellerLedgerQuery) coinSellerLedgerQuery {
|
||
if query.Page < 1 {
|
||
query.Page = 1
|
||
}
|
||
if query.PageSize < 1 {
|
||
query.PageSize = 20
|
||
}
|
||
if query.PageSize > 100 {
|
||
query.PageSize = 100
|
||
}
|
||
query.SellerKeyword = strings.TrimSpace(query.SellerKeyword)
|
||
query.SellerFilter.UserID = strings.TrimSpace(query.SellerFilter.UserID)
|
||
query.SellerFilter.DisplayUserID = strings.TrimSpace(query.SellerFilter.DisplayUserID)
|
||
query.SellerFilter.Username = strings.TrimSpace(query.SellerFilter.Username)
|
||
query.LedgerType = strings.TrimSpace(query.LedgerType)
|
||
return query
|
||
}
|
||
|
||
func coinSellerLedgerBizTypes(ledgerType string) ([]string, error) {
|
||
switch strings.TrimSpace(ledgerType) {
|
||
case "":
|
||
// 空类型表示“全部币商流水”，但仍只允许当前产品定义的公开币商库存口径，不能把其他内部账带出来。
|
||
return []string{
|
||
coinSellerStockPurchaseBizType,
|
||
coinSellerStockDeductionBizType,
|
||
coinSellerRechargeBizType,
|
||
coinSellerCoinCompensationBizType,
|
||
coinSellerTransferBizType,
|
||
coinSellerSubTransferBizType,
|
||
salaryTransferToCoinSellerBizType,
|
||
coinManualCreditBizType,
|
||
}, nil
|
||
case coinSellerLedgerTypeAdminStockCredit:
|
||
// 进货操作是运营库存口径，底层包含后台 USDT 进货、H5 三方币商充值、金币补偿和后台扣除；资产条件仍限定在 COIN_SELLER_COIN。
|
||
return []string{coinSellerStockPurchaseBizType, coinSellerStockDeductionBizType, coinSellerRechargeBizType, coinSellerCoinCompensationBizType, coinManualCreditBizType}, nil
|
||
case coinSellerLedgerTypeSellerTransfer:
|
||
// 币商转用户只展示币商侧出账分录，收款用户资料在展示投影里补齐。
|
||
return []string{coinSellerTransferBizType}, nil
|
||
case coinSellerLedgerTypeSubSellerTransfer:
|
||
// 父子币商库存划拨按 wallet_entries 展示双边余额变动，不能混入普通玩家充值流水。
|
||
return []string{coinSellerSubTransferBizType}, nil
|
||
case coinSellerLedgerTypeSalaryTransferIncome:
|
||
// 工资转币商是用户工资资产换入币商库存，币商侧表现为 COIN_SELLER_COIN 入账。
|
||
return []string{salaryTransferToCoinSellerBizType}, nil
|
||
default:
|
||
return nil, errInvalidCoinSellerLedgerType
|
||
}
|
||
}
|
||
|
||
func coinSellerLedgerTypeForBizType(bizType string) string {
|
||
switch bizType {
|
||
case coinSellerStockPurchaseBizType, coinSellerStockDeductionBizType, coinSellerRechargeBizType, coinSellerCoinCompensationBizType:
|
||
return coinSellerLedgerTypeAdminStockCredit
|
||
case coinManualCreditBizType:
|
||
return coinSellerLedgerTypeAdminStockCredit
|
||
case coinSellerTransferBizType:
|
||
return coinSellerLedgerTypeSellerTransfer
|
||
case coinSellerSubTransferBizType:
|
||
return coinSellerLedgerTypeSubSellerTransfer
|
||
case salaryTransferToCoinSellerBizType:
|
||
return coinSellerLedgerTypeSalaryTransferIncome
|
||
default:
|
||
return bizType
|
||
}
|
||
}
|
||
|
||
func coinSellerLedgerLabel(item coinSellerLedgerDTO) string {
|
||
switch item.BizType {
|
||
case coinSellerStockPurchaseBizType:
|
||
return "USDT进货"
|
||
case coinSellerStockDeductionBizType:
|
||
return "USDT扣除"
|
||
case coinSellerRechargeBizType:
|
||
return "三方充值"
|
||
case coinSellerCoinCompensationBizType:
|
||
return "金币补偿"
|
||
case coinManualCreditBizType:
|
||
if item.AvailableDelta < 0 {
|
||
return "金币扣除"
|
||
}
|
||
return "金币增加"
|
||
case coinSellerTransferBizType:
|
||
return "币商转用户"
|
||
case coinSellerSubTransferBizType:
|
||
return "向子币商转账"
|
||
case salaryTransferToCoinSellerBizType:
|
||
return "工资转币商"
|
||
}
|
||
switch item.LedgerType {
|
||
case coinSellerLedgerTypeAdminStockCredit:
|
||
return "后台操作"
|
||
case coinSellerLedgerTypeSellerTransfer:
|
||
return "币商转用户"
|
||
case coinSellerLedgerTypeSubSellerTransfer:
|
||
return "向子币商转账"
|
||
case coinSellerLedgerTypeSalaryTransferIncome:
|
||
return "工资转币商"
|
||
default:
|
||
return firstNonEmpty(item.LedgerType, item.BizType, "-")
|
||
}
|
||
}
|
||
|
||
func signedCoinSellerLedgerAmount(item coinSellerLedgerDTO) int64 {
|
||
if item.Direction == directionOut || item.AvailableDelta < 0 {
|
||
return -absInt64(item.Amount)
|
||
}
|
||
return absInt64(item.Amount)
|
||
}
|
||
|
||
func formatUSDMinorForCSV(value int64) string {
|
||
if value <= 0 {
|
||
return ""
|
||
}
|
||
return strconv.FormatFloat(float64(value)/100, 'f', 2, 64)
|
||
}
|
||
|
||
func formatUSDTMicroForCSV(value int64) string {
|
||
if value <= 0 {
|
||
return ""
|
||
}
|
||
formatted := strconv.FormatFloat(float64(value)/1_000_000, 'f', 6, 64)
|
||
return strings.TrimRight(strings.TrimRight(formatted, "0"), ".")
|
||
}
|
||
|
||
func coinSellerLedgerOperatorExportFields(item coinSellerLedgerDTO) (string, string) {
|
||
if item.LedgerType != coinSellerLedgerTypeAdminStockCredit {
|
||
return "", ""
|
||
}
|
||
operatorID := firstNonEmpty(item.Operator.AdminID, item.OperatorUserID)
|
||
return firstNonEmpty(item.Operator.Username, item.Operator.Name, operatorID), operatorID
|
||
}
|
||
|
||
func coinSellerLedgerReceiverUserID(bizType string, sellerUserID int64, counterpartyUserID int64, metadata map[string]any) int64 {
|
||
if bizType == coinSellerTransferBizType {
|
||
// 新数据把真实收款用户写在 metadata.target_user_id，优先使用它，避免 counterparty 语义随老交易实现变化。
|
||
if targetUserID := metadataInt64(metadata, "target_user_id", "targetUserId"); targetUserID > 0 {
|
||
return targetUserID
|
||
}
|
||
// 老数据没有 target_user_id 时使用 counterparty_user_id 兜底；只有币商转用户才允许这样兜底。
|
||
if counterpartyUserID > 0 {
|
||
return counterpartyUserID
|
||
}
|
||
}
|
||
if bizType == coinSellerSubTransferBizType {
|
||
// 父子币商划拨会落父出账和子入账两条库存分录；两条记录的实际收款人都是 child_user_id。
|
||
// 子币商入账侧的 counterparty_user_id 是父币商，不能拿来展示为收款人。
|
||
if childUserID := metadataInt64(metadata, "child_user_id", "childUserId"); childUserID > 0 {
|
||
return childUserID
|
||
}
|
||
return sellerUserID
|
||
}
|
||
// 后台入账和工资转币商都展示币商本人为收款人，这是页面“实际收款人”的产品口径。
|
||
return sellerUserID
|
||
}
|
||
|
||
func directionForDelta(delta int64) string {
|
||
if delta < 0 {
|
||
return directionOut
|
||
}
|
||
return directionIn
|
||
}
|
||
|
||
func absInt64(value int64) int64 {
|
||
if value < 0 {
|
||
return -value
|
||
}
|
||
return value
|
||
}
|
||
|
||
func formatOptionalID(value int64) string {
|
||
if value <= 0 {
|
||
return ""
|
||
}
|
||
return strconv.FormatInt(value, 10)
|
||
}
|
||
|
||
func parseMetadataJSON(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 metadataString(metadata map[string]any, key string) string {
|
||
value, ok := metadata[key]
|
||
if !ok || value == nil {
|
||
return ""
|
||
}
|
||
switch typed := value.(type) {
|
||
case string:
|
||
return strings.TrimSpace(typed)
|
||
default:
|
||
return strings.TrimSpace(fmt.Sprint(typed))
|
||
}
|
||
}
|
||
|
||
func metadataInt64(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 firstNonEmpty(values ...string) string {
|
||
for _, value := range values {
|
||
value = strings.TrimSpace(value)
|
||
if value != "" {
|
||
return value
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
func parseFlexibleUserID(value any) (int64, error) {
|
||
switch typed := value.(type) {
|
||
case float64:
|
||
userID := int64(typed)
|
||
if float64(userID) != typed || userID <= 0 {
|
||
return 0, fmt.Errorf("target_user_id is invalid")
|
||
}
|
||
return userID, nil
|
||
case string:
|
||
userID, err := strconv.ParseInt(strings.TrimSpace(typed), 10, 64)
|
||
if err != nil || userID <= 0 {
|
||
return 0, fmt.Errorf("target_user_id is invalid")
|
||
}
|
||
return userID, nil
|
||
case json.Number:
|
||
userID, err := typed.Int64()
|
||
if err != nil || userID <= 0 {
|
||
return 0, fmt.Errorf("target_user_id is invalid")
|
||
}
|
||
return userID, nil
|
||
default:
|
||
return 0, fmt.Errorf("target_user_id is required")
|
||
}
|
||
}
|
||
|
||
func (s *Service) adminProfiles(ctx context.Context, adminIDs []int64) (map[int64]coinAdjustmentOperatorDTO, error) {
|
||
result := make(map[int64]coinAdjustmentOperatorDTO, len(adminIDs))
|
||
if s == nil || s.adminDB == nil || len(adminIDs) == 0 {
|
||
return result, nil
|
||
}
|
||
adminIDs = positiveUniqueInt64s(adminIDs)
|
||
if len(adminIDs) == 0 {
|
||
return result, nil
|
||
}
|
||
args := make([]any, 0, len(adminIDs))
|
||
for _, id := range adminIDs {
|
||
args = append(args, id)
|
||
}
|
||
rows, err := s.adminDB.QueryContext(ctx, fmt.Sprintf(`
|
||
SELECT id, COALESCE(username, ''), COALESCE(name, '')
|
||
FROM admin_users
|
||
WHERE id IN (%s)
|
||
`, placeholders(len(adminIDs))), args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
for rows.Next() {
|
||
var id int64
|
||
var operator coinAdjustmentOperatorDTO
|
||
if err := rows.Scan(&id, &operator.Username, &operator.Name); err != nil {
|
||
return nil, err
|
||
}
|
||
operator.AdminID = strconv.FormatInt(id, 10)
|
||
result[id] = operator
|
||
}
|
||
return result, rows.Err()
|
||
}
|
||
|
||
func positiveUniqueInt64s(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 defaultCoinAdjustmentCommandID(requestID string) string {
|
||
requestID = strings.TrimSpace(requestID)
|
||
if requestID == "" {
|
||
return "admin-coin-adjustment-" + strconv.FormatInt(time.Now().UTC().UnixNano(), 10)
|
||
}
|
||
return "admin-coin-adjustment-" + requestID
|
||
}
|
||
|
||
func defaultCoinAdjustmentEvidenceRef(requestID string) string {
|
||
requestID = strings.TrimSpace(requestID)
|
||
if requestID == "" {
|
||
return "admin-coin-adjustment"
|
||
}
|
||
return "admin-request:" + requestID
|
||
}
|
||
|
||
func offset(page int, pageSize int) int {
|
||
if page < 1 {
|
||
page = 1
|
||
}
|
||
return (page - 1) * pageSize
|
||
}
|
||
|
||
func placeholders(count int) string {
|
||
if count <= 0 {
|
||
return ""
|
||
}
|
||
return strings.TrimRight(strings.Repeat("?,", count), ",")
|
||
}
|
||
|
||
func uniqueInt64s(values []int64) []int64 {
|
||
seen := make(map[int64]struct{}, len(values))
|
||
result := make([]int64, 0, len(values))
|
||
for _, value := range values {
|
||
if _, ok := seen[value]; ok {
|
||
continue
|
||
}
|
||
seen[value] = struct{}{}
|
||
result = append(result, value)
|
||
}
|
||
return result
|
||
}
|
||
|
||
func escapeLike(value string) string {
|
||
value = strings.ReplaceAll(value, `\`, `\\`)
|
||
value = strings.ReplaceAll(value, `%`, `\%`)
|
||
value = strings.ReplaceAll(value, `_`, `\_`)
|
||
return value
|
||
}
|