2026-05-15 10:00:43 +08:00

681 lines
19 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package coinledger
import (
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"strconv"
"strings"
"time"
"hyapp-admin-server/internal/integration/walletclient"
walletv1 "hyapp.local/api/proto/wallet/v1"
)
const (
coinAssetType = "COIN"
coinManualCreditBizType = "manual_credit"
directionIn = "income"
directionOut = "expense"
)
var errCoinAdjustmentTargetNotFound = errors.New("coin adjustment target user not found")
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) 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")
}
row := s.userDB.QueryRowContext(ctx, `
SELECT user_id, current_display_user_id, COALESCE(username, ''), COALESCE(avatar, '')
FROM users
WHERE app_code = ?
AND (CAST(user_id AS CHAR) = ? OR current_display_user_id = ?)
ORDER BY
CASE
WHEN CAST(user_id AS CHAR) = ? THEN 0
WHEN current_display_user_id = ? THEN 1
ELSE 2
END,
user_id DESC
LIMIT 1`,
appCode, keyword, keyword, keyword, keyword,
)
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、当前短 ID 和昵称;数字关键字直接加入 user_id避免用户资料缺失时查不到账务事实。
args := []any{appCode}
where := "WHERE app_code = ? AND (current_display_user_id = ?"
args = append(args, keyword)
if numeric, err := strconv.ParseInt(keyword, 10, 64); err == nil && numeric > 0 {
where += " OR user_id = ?"
args = append(args, numeric)
}
where += " OR username LIKE ? ESCAPE '\\\\')"
args = append(args, "%"+escapeLike(keyword)+"%")
rows, err := s.userDB.QueryContext(ctx, `
SELECT user_id
FROM users
`+where+`
ORDER BY updated_at_ms DESC, 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) 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 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 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 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 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
}