package coinledger import ( "bytes" "context" "database/sql" "encoding/csv" "encoding/json" "errors" "fmt" "strconv" "strings" "time" "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" coinSellerRechargeBizType = "coin_seller_recharge" coinSellerStockPurchaseBizType = "coin_seller_stock_purchase" coinSellerCoinCompensationBizType = "coin_seller_coin_compensation" salaryTransferToCoinSellerBizType = "salary_transfer_to_coin_seller" coinSellerLedgerTypeAdminStockCredit = "admin_stock_credit" coinSellerLedgerTypeSellerTransfer = "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) if err != nil { return nil, 0, err } if userFiltered && len(userIDs) == 0 { return []coinLedgerEntryDTO{}, 0, nil } // 后台只做只读查询:金币流水事实仍以 wallet_entries 追加分录为准,不在 admin 库冗余账务状态。 whereSQL, args := coinLedgerWhere(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, 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) 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.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 len(reason) > 512 { return coinAdjustmentCreateDTO{}, fmt.Errorf("reason 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, }) 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, }, nil } func (s *Service) resolveUserFilter(ctx context.Context, appCode string, keyword string) ([]int64, bool, error) { keyword = strings.TrimSpace(keyword) if keyword == "" { return nil, false, nil } directUserIDs := make([]int64, 0, 1) if numeric, err := strconv.ParseInt(keyword, 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") } // 用户列筛选同时支持长 ID、默认短号、当前短号、active 靓号和昵称;数字关键字直接加入 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 } 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()) 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.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) 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.LedgerType = strings.TrimSpace(query.LedgerType) return query } func coinSellerLedgerBizTypes(ledgerType string) ([]string, error) { switch strings.TrimSpace(ledgerType) { case "": // 空类型表示“全部币商流水”,但仍只允许当前产品定义的公开币商库存口径,不能把其他内部账带出来。 return []string{ coinSellerStockPurchaseBizType, coinSellerRechargeBizType, coinSellerCoinCompensationBizType, coinSellerTransferBizType, salaryTransferToCoinSellerBizType, coinManualCreditBizType, }, nil case coinSellerLedgerTypeAdminStockCredit: // 进货操作是运营库存口径,底层包含后台 USDT 进货、H5 三方币商充值、金币补偿和后台扣除;资产条件仍限定在 COIN_SELLER_COIN。 return []string{coinSellerStockPurchaseBizType, coinSellerRechargeBizType, coinSellerCoinCompensationBizType, coinManualCreditBizType}, nil case coinSellerLedgerTypeSellerTransfer: // 币商转用户只展示币商侧出账分录,收款用户资料在展示投影里补齐。 return []string{coinSellerTransferBizType}, nil case coinSellerLedgerTypeSalaryTransferIncome: // 工资转币商是用户工资资产换入币商库存,币商侧表现为 COIN_SELLER_COIN 入账。 return []string{salaryTransferToCoinSellerBizType}, nil default: return nil, errInvalidCoinSellerLedgerType } } func coinSellerLedgerTypeForBizType(bizType string) string { switch bizType { case coinSellerStockPurchaseBizType, coinSellerRechargeBizType, coinSellerCoinCompensationBizType: return coinSellerLedgerTypeAdminStockCredit case coinManualCreditBizType: return coinSellerLedgerTypeAdminStockCredit case coinSellerTransferBizType: return coinSellerLedgerTypeSellerTransfer case salaryTransferToCoinSellerBizType: return coinSellerLedgerTypeSalaryTransferIncome default: return bizType } } func coinSellerLedgerLabel(item coinSellerLedgerDTO) string { switch item.BizType { case coinSellerStockPurchaseBizType: return "USDT进货" case coinSellerRechargeBizType: return "三方充值" case coinSellerCoinCompensationBizType: return "金币补偿" case coinManualCreditBizType: if item.AvailableDelta < 0 { return "金币扣除" } return "金币增加" case coinSellerTransferBizType: return "币商转用户" case salaryTransferToCoinSellerBizType: return "工资转币商" } switch item.LedgerType { case coinSellerLedgerTypeAdminStockCredit: return "后台操作" case coinSellerLedgerTypeSellerTransfer: 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 } } // 后台入账和工资转币商都展示币商本人为收款人,这是页面“实际收款人”的产品口径。 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 }