2026-06-10 11:14:49 +08:00

1136 lines
36 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 hostorg
import (
"context"
"database/sql"
"fmt"
"sort"
"strings"
"time"
"hyapp-admin-server/internal/appctx"
"hyapp-admin-server/internal/integration/userclient"
)
type Reader struct {
db *sql.DB
walletDB *sql.DB
adminDB *sql.DB
}
func NewReader(db *sql.DB, walletDB *sql.DB, adminDB *sql.DB) *Reader {
return &Reader{db: db, walletDB: walletDB, adminDB: adminDB}
}
type CoinSellerListItem struct {
UserID int64 `json:"userId,string"`
Status string `json:"status"`
MerchantAssetType string `json:"merchantAssetType"`
MerchantBalance int64 `json:"merchantBalance"`
// TotalRechargeUSDTMicro 是币商 USDT 进货累计付款微单位,只统计 counts_as_seller_recharge 的库存记录。
TotalRechargeUSDTMicro int64 `json:"totalRechargeUsdtMicro"`
Contact string `json:"contact"`
DisplayUserID string `json:"displayUserId"`
Username string `json:"username"`
Avatar string `json:"avatar"`
RegionID int64 `json:"regionId"`
RegionName string `json:"regionName"`
CreatedByUserID int64 `json:"createdByUserId"`
CreatedAtMs int64 `json:"createdAtMs"`
UpdatedAtMs int64 `json:"updatedAtMs"`
}
type CoinSellerSalaryRateTier struct {
RegionID int64 `json:"regionId,string"`
MinUSDMinor int64 `json:"minUsdMinor"`
MaxUSDMinor int64 `json:"maxUsdMinor"`
MinUSD string `json:"minUsd"`
MaxUSD string `json:"maxUsd"`
CoinPerUSD int64 `json:"coinPerUsd"`
Status string `json:"status"`
Enabled bool `json:"enabled"`
SortOrder int `json:"sortOrder"`
UpdatedAtMs int64 `json:"updatedAtMs"`
}
// ListBDProfiles 只读 user-service 的 BD/BD Leader 事实表,后台列表按角色选择物理表,避免把两种身份重新混在一起。
func (r *Reader) ListBDProfiles(ctx context.Context, query listQuery, role string) ([]*userclient.BDProfile, int64, error) {
if r == nil || r.db == nil {
return nil, 0, errUserDBNotConfigured()
}
if role == "bd_leader" {
return r.listBDLeaderProfiles(ctx, query)
}
appCode := appctx.FromContext(ctx)
whereSQL := `
FROM bd_profiles bp
LEFT JOIN users u ON u.app_code = bp.app_code AND u.user_id = bp.user_id
LEFT JOIN users parent_leader ON parent_leader.app_code = bp.app_code AND parent_leader.user_id = bp.parent_leader_user_id
LEFT JOIN users creator ON creator.app_code = bp.app_code AND creator.user_id = bp.created_by_user_id
LEFT JOIN regions r ON r.app_code = bp.app_code AND r.region_id = bp.region_id
WHERE bp.app_code = ? AND bp.role = ?`
args := []any{appCode, role}
if query.Status != "" {
whereSQL += " AND bp.status = ?"
args = append(args, query.Status)
}
if query.RegionID > 0 {
whereSQL += " AND bp.region_id = ?"
args = append(args, query.RegionID)
}
if query.ParentLeaderUserID > 0 {
whereSQL += " AND bp.parent_leader_user_id = ?"
args = append(args, query.ParentLeaderUserID)
}
if keyword := strings.TrimSpace(query.Keyword); keyword != "" {
like := "%" + keyword + "%"
whereSQL += " AND (CAST(bp.user_id AS CHAR) LIKE ? OR u.current_display_user_id LIKE ? OR u.username LIKE ? OR r.name LIKE ?)"
args = append(args, like, like, like, like)
}
total, err := countRows(ctx, r.db, whereSQL, args...)
if err != nil {
return nil, 0, err
}
rows, err := r.db.QueryContext(ctx, fmt.Sprintf(`
SELECT bp.user_id, bp.role, bp.region_id, COALESCE(bp.parent_leader_user_id, 0),
bp.status, bp.created_by_user_id, bp.created_at_ms, bp.updated_at_ms,
COALESCE(u.current_display_user_id, ''), COALESCE(u.username, ''),
COALESCE(u.avatar, ''), COALESCE(r.name, ''),
COALESCE(parent_leader.current_display_user_id, ''), COALESCE(parent_leader.username, ''),
COALESCE(parent_leader.avatar, ''),
COALESCE(creator.current_display_user_id, ''), COALESCE(creator.username, ''),
COALESCE(creator.avatar, ''),
(
SELECT COUNT(1)
FROM bd_profiles child
WHERE child.app_code = bp.app_code
AND child.role = 'bd'
AND child.parent_leader_user_id = bp.user_id
)
%s
ORDER BY bp.created_at_ms DESC, bp.user_id DESC
LIMIT ? OFFSET ?
`, whereSQL), append(args, query.PageSize, offset(query.Page, query.PageSize))...)
if err != nil {
return nil, 0, err
}
defer rows.Close()
items := make([]*userclient.BDProfile, 0, query.PageSize)
for rows.Next() {
item := &userclient.BDProfile{}
if err := rows.Scan(
&item.UserID,
&item.Role,
&item.RegionID,
&item.ParentLeaderUserID,
&item.Status,
&item.CreatedByUserID,
&item.CreatedAtMs,
&item.UpdatedAtMs,
&item.DisplayUserID,
&item.Username,
&item.Avatar,
&item.RegionName,
&item.ParentLeaderDisplayID,
&item.ParentLeaderUsername,
&item.ParentLeaderAvatar,
&item.CreatedByDisplayUserID,
&item.CreatedByUsername,
&item.CreatedByAvatar,
&item.SubBDCount,
); err != nil {
return nil, 0, err
}
items = append(items, item)
}
if err := rows.Err(); err != nil {
return nil, 0, err
}
if err := r.fillBDProfileCreators(ctx, items); err != nil {
return nil, 0, err
}
if role == "bd_leader" {
if err := r.fillBDLeaderPositionAliases(ctx, appCode, items); err != nil {
return nil, 0, err
}
}
return items, total, nil
}
func (r *Reader) listBDLeaderProfiles(ctx context.Context, query listQuery) ([]*userclient.BDProfile, int64, error) {
appCode := appctx.FromContext(ctx)
if query.ParentLeaderUserID > 0 {
// Leader 身份没有上级 Leader 字段;带父级筛选时直接返回空结果,避免误用普通 BD 的父级语义。
return []*userclient.BDProfile{}, 0, nil
}
whereSQL := `
FROM bd_leader_profiles bp
LEFT JOIN users u ON u.app_code = bp.app_code AND u.user_id = bp.user_id
LEFT JOIN users creator ON creator.app_code = bp.app_code AND creator.user_id = bp.created_by_user_id
LEFT JOIN regions r ON r.app_code = bp.app_code AND r.region_id = bp.region_id
WHERE bp.app_code = ?`
args := []any{appCode}
if query.Status != "" {
whereSQL += " AND bp.status = ?"
args = append(args, query.Status)
}
if query.RegionID > 0 {
whereSQL += " AND bp.region_id = ?"
args = append(args, query.RegionID)
}
if keyword := strings.TrimSpace(query.Keyword); keyword != "" {
like := "%" + keyword + "%"
whereSQL += " AND (CAST(bp.user_id AS CHAR) LIKE ? OR u.current_display_user_id LIKE ? OR u.username LIKE ? OR r.name LIKE ?)"
args = append(args, like, like, like, like)
}
total, err := countRows(ctx, r.db, whereSQL, args...)
if err != nil {
return nil, 0, err
}
rows, err := r.db.QueryContext(ctx, fmt.Sprintf(`
SELECT bp.user_id, 'bd_leader', bp.region_id, 0,
bp.status, bp.created_by_user_id, bp.created_at_ms, bp.updated_at_ms,
COALESCE(u.current_display_user_id, ''), COALESCE(u.username, ''),
COALESCE(u.avatar, ''), COALESCE(r.name, ''),
'', '', '',
COALESCE(creator.current_display_user_id, ''), COALESCE(creator.username, ''),
COALESCE(creator.avatar, ''),
(
SELECT COUNT(1)
FROM bd_profiles child
WHERE child.app_code = bp.app_code
AND child.role = 'bd'
AND child.parent_leader_user_id = bp.user_id
)
%s
ORDER BY bp.created_at_ms DESC, bp.user_id DESC
LIMIT ? OFFSET ?
`, whereSQL), append(args, query.PageSize, offset(query.Page, query.PageSize))...)
if err != nil {
return nil, 0, err
}
defer rows.Close()
items := make([]*userclient.BDProfile, 0, query.PageSize)
for rows.Next() {
item := &userclient.BDProfile{}
if err := rows.Scan(
&item.UserID,
&item.Role,
&item.RegionID,
&item.ParentLeaderUserID,
&item.Status,
&item.CreatedByUserID,
&item.CreatedAtMs,
&item.UpdatedAtMs,
&item.DisplayUserID,
&item.Username,
&item.Avatar,
&item.RegionName,
&item.ParentLeaderDisplayID,
&item.ParentLeaderUsername,
&item.ParentLeaderAvatar,
&item.CreatedByDisplayUserID,
&item.CreatedByUsername,
&item.CreatedByAvatar,
&item.SubBDCount,
); err != nil {
return nil, 0, err
}
items = append(items, item)
}
if err := rows.Err(); err != nil {
return nil, 0, err
}
if err := r.fillBDProfileCreators(ctx, items); err != nil {
return nil, 0, err
}
if err := r.fillBDLeaderPositionAliases(ctx, appCode, items); err != nil {
return nil, 0, err
}
return items, total, nil
}
func (r *Reader) GetBDLeader(ctx context.Context, userID int64) (*userclient.BDProfile, error) {
if r == nil || r.db == nil {
return nil, errUserDBNotConfigured()
}
appCode := appctx.FromContext(ctx)
item := &userclient.BDProfile{}
err := r.db.QueryRowContext(ctx, `
SELECT bp.user_id, 'bd_leader', bp.region_id, 0,
bp.status, bp.created_by_user_id, bp.created_at_ms, bp.updated_at_ms,
COALESCE(u.current_display_user_id, ''), COALESCE(u.username, ''),
COALESCE(u.avatar, ''), COALESCE(r.name, ''),
COALESCE(creator.current_display_user_id, ''), COALESCE(creator.username, ''),
COALESCE(creator.avatar, ''),
(
SELECT COUNT(1)
FROM bd_profiles child
WHERE child.app_code = bp.app_code
AND child.role = 'bd'
AND child.parent_leader_user_id = bp.user_id
)
FROM bd_leader_profiles bp
LEFT JOIN users u ON u.app_code = bp.app_code AND u.user_id = bp.user_id
LEFT JOIN users creator ON creator.app_code = bp.app_code AND creator.user_id = bp.created_by_user_id
LEFT JOIN regions r ON r.app_code = bp.app_code AND r.region_id = bp.region_id
WHERE bp.app_code = ? AND bp.user_id = ?
`, appCode, userID).Scan(
&item.UserID,
&item.Role,
&item.RegionID,
&item.ParentLeaderUserID,
&item.Status,
&item.CreatedByUserID,
&item.CreatedAtMs,
&item.UpdatedAtMs,
&item.DisplayUserID,
&item.Username,
&item.Avatar,
&item.RegionName,
&item.CreatedByDisplayUserID,
&item.CreatedByUsername,
&item.CreatedByAvatar,
&item.SubBDCount,
)
if err == sql.ErrNoRows {
return nil, fmt.Errorf("bd leader not found")
}
if err != nil {
return nil, err
}
items := []*userclient.BDProfile{item}
if err := r.fillBDProfileCreators(ctx, items); err != nil {
return nil, err
}
if err := r.fillBDLeaderPositionAliases(ctx, appCode, items); err != nil {
return nil, err
}
return item, nil
}
func (r *Reader) fillBDProfileCreators(ctx context.Context, items []*userclient.BDProfile) error {
if r == nil || r.adminDB == nil || len(items) == 0 {
return nil
}
seen := make(map[int64]struct{}, len(items))
ids := make([]any, 0, len(items))
for _, item := range items {
if item.CreatedByUserID <= 0 {
continue
}
if _, ok := seen[item.CreatedByUserID]; ok {
continue
}
seen[item.CreatedByUserID] = struct{}{}
ids = append(ids, item.CreatedByUserID)
}
if len(ids) == 0 {
return nil
}
rows, err := r.adminDB.QueryContext(ctx, `
SELECT id, username, name
FROM admin_users
WHERE id IN (`+sqlPlaceholders(len(ids))+`)
`, ids...)
if err != nil {
return err
}
defer rows.Close()
type creator struct {
account string
name string
}
creators := make(map[int64]creator, len(ids))
for rows.Next() {
var id int64
var item creator
if err := rows.Scan(&id, &item.account, &item.name); err != nil {
return err
}
creators[id] = item
}
if err := rows.Err(); err != nil {
return err
}
for _, item := range items {
creator, ok := creators[item.CreatedByUserID]
if !ok {
continue
}
item.CreatedByDisplayUserID = creator.account
item.CreatedByUsername = firstNonBlank(creator.name, creator.account)
}
return nil
}
func (r *Reader) fillBDLeaderPositionAliases(ctx context.Context, appCode string, items []*userclient.BDProfile) error {
if r == nil || r.adminDB == nil || len(items) == 0 {
return nil
}
ids := make([]any, 0, len(items))
for _, item := range items {
if item.UserID > 0 {
ids = append(ids, item.UserID)
}
}
if len(ids) == 0 {
return nil
}
args := append([]any{appCode}, ids...)
rows, err := r.adminDB.QueryContext(ctx, `
SELECT user_id, position_alias
FROM admin_bd_leader_position_aliases
WHERE app_code = ? AND user_id IN (`+sqlPlaceholders(len(ids))+`)
`, args...)
if err != nil {
return err
}
defer rows.Close()
aliases := make(map[int64]string, len(ids))
for rows.Next() {
var userID int64
var alias string
if err := rows.Scan(&userID, &alias); err != nil {
return err
}
aliases[userID] = strings.TrimSpace(alias)
}
if err := rows.Err(); err != nil {
return err
}
for _, item := range items {
item.PositionAlias = aliases[item.UserID]
}
return nil
}
func (r *Reader) SaveBDLeaderPositionAlias(ctx context.Context, userID int64, actorID int64, alias string) error {
if r == nil || r.adminDB == nil {
return fmt.Errorf("admin mysql is not configured")
}
if userID <= 0 {
return fmt.Errorf("user_id is required")
}
appCode := appctx.FromContext(ctx)
positionAlias := strings.TrimSpace(alias)
if positionAlias == "" {
// 空别名表示恢复 H5 配置表里的默认 admin 展示名,删除行比保留空值更清晰。
_, err := r.adminDB.ExecContext(ctx, `
DELETE FROM admin_bd_leader_position_aliases
WHERE app_code = ? AND user_id = ?
`, appCode, userID)
return err
}
nowMs := time.Now().UnixMilli()
_, err := r.adminDB.ExecContext(ctx, `
INSERT INTO admin_bd_leader_position_aliases (
app_code, user_id, position_alias, created_by_admin_id, updated_by_admin_id, created_at_ms, updated_at_ms
) VALUES (?, ?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
position_alias = VALUES(position_alias),
updated_by_admin_id = VALUES(updated_by_admin_id),
updated_at_ms = VALUES(updated_at_ms)
`, appCode, userID, positionAlias, actorID, actorID, nowMs, nowMs)
return err
}
func (r *Reader) DeleteBDLeader(ctx context.Context, userID int64) (*userclient.BDProfile, error) {
if r == nil || r.db == nil {
return nil, errUserDBNotConfigured()
}
appCode := appctx.FromContext(ctx)
item := &userclient.BDProfile{}
err := r.db.QueryRowContext(ctx, `
SELECT bp.user_id, 'bd_leader', bp.region_id, 0,
bp.status, bp.created_by_user_id, bp.created_at_ms, bp.updated_at_ms
FROM bd_leader_profiles bp
WHERE bp.app_code = ? AND bp.user_id = ?
`, appCode, userID).Scan(
&item.UserID,
&item.Role,
&item.RegionID,
&item.ParentLeaderUserID,
&item.Status,
&item.CreatedByUserID,
&item.CreatedAtMs,
&item.UpdatedAtMs,
)
if err == sql.ErrNoRows {
return nil, fmt.Errorf("bd leader not found")
}
if err != nil {
return nil, err
}
result, err := r.db.ExecContext(ctx, `
DELETE FROM bd_leader_profiles
WHERE app_code = ? AND user_id = ?
`, appCode, userID)
if err != nil {
return nil, err
}
affected, err := result.RowsAffected()
if err != nil {
return nil, err
}
if affected == 0 {
return nil, fmt.Errorf("bd leader not found")
}
if r.adminDB != nil {
_, _ = r.adminDB.ExecContext(ctx, `
DELETE FROM admin_bd_leader_position_aliases
WHERE app_code = ? AND user_id = ?
`, appCode, userID)
}
return item, nil
}
// ListAgencies 只读 user-service 的 Agency 事实表;后台关闭和入会开关仍走写命令。
func (r *Reader) ListAgencies(ctx context.Context, query listQuery) ([]*userclient.Agency, int64, error) {
if r == nil || r.db == nil {
return nil, 0, errUserDBNotConfigured()
}
appCode := appctx.FromContext(ctx)
whereSQL := `
FROM agencies a
LEFT JOIN users owner ON owner.app_code = a.app_code AND owner.user_id = a.owner_user_id
LEFT JOIN users parent_bd ON parent_bd.app_code = a.app_code AND parent_bd.user_id = a.parent_bd_user_id
LEFT JOIN regions r ON r.app_code = a.app_code AND r.region_id = a.region_id
WHERE a.app_code = ?`
args := []any{appCode}
if query.Status != "" {
whereSQL += " AND a.status = ?"
args = append(args, query.Status)
} else {
whereSQL += " AND a.status <> ?"
args = append(args, "deleted")
}
if query.RegionID > 0 {
whereSQL += " AND a.region_id = ?"
args = append(args, query.RegionID)
}
if query.ParentBDUserID > 0 {
whereSQL += " AND a.parent_bd_user_id = ?"
args = append(args, query.ParentBDUserID)
}
if keyword := strings.TrimSpace(query.Keyword); keyword != "" {
like := "%" + keyword + "%"
whereSQL += " AND (a.name LIKE ? OR CAST(a.agency_id AS CHAR) LIKE ? OR CAST(a.owner_user_id AS CHAR) LIKE ? OR owner.current_display_user_id LIKE ? OR owner.username LIKE ? OR CAST(a.parent_bd_user_id AS CHAR) LIKE ? OR parent_bd.current_display_user_id LIKE ? OR parent_bd.username LIKE ? OR r.name LIKE ?)"
args = append(args, like, like, like, like, like, like, like, like, like)
}
total, err := countRows(ctx, r.db, whereSQL, args...)
if err != nil {
return nil, 0, err
}
rows, err := r.db.QueryContext(ctx, fmt.Sprintf(`
SELECT a.agency_id, a.owner_user_id, a.region_id, a.parent_bd_user_id,
a.name, a.status, a.join_enabled, a.max_hosts,
a.created_by_user_id, a.created_at_ms, a.updated_at_ms,
COALESCE(owner.current_display_user_id, ''), COALESCE(owner.username, ''),
COALESCE(owner.avatar, ''),
COALESCE(parent_bd.current_display_user_id, ''), COALESCE(parent_bd.username, ''),
COALESCE(parent_bd.avatar, ''), COALESCE(r.name, '')
%s
ORDER BY a.created_at_ms DESC, a.agency_id DESC
LIMIT ? OFFSET ?
`, whereSQL), append(args, query.PageSize, offset(query.Page, query.PageSize))...)
if err != nil {
return nil, 0, err
}
defer rows.Close()
items := make([]*userclient.Agency, 0, query.PageSize)
for rows.Next() {
item := &userclient.Agency{}
if err := rows.Scan(
&item.AgencyID,
&item.OwnerUserID,
&item.RegionID,
&item.ParentBDUserID,
&item.Name,
&item.Status,
&item.JoinEnabled,
&item.MaxHosts,
&item.CreatedByUserID,
&item.CreatedAtMs,
&item.UpdatedAtMs,
&item.OwnerDisplayUserID,
&item.OwnerUsername,
&item.OwnerAvatar,
&item.ParentBDDisplayUserID,
&item.ParentBDUsername,
&item.ParentBDAvatar,
&item.RegionName,
); err != nil {
return nil, 0, err
}
items = append(items, item)
}
return items, total, rows.Err()
}
// ListHostProfiles 只读主播身份事实;成员变更和主播状态变更不在这个列表接口里发生。
func (r *Reader) ListHostProfiles(ctx context.Context, query listQuery) ([]*userclient.HostProfile, int64, error) {
if r == nil || r.db == nil {
return nil, 0, errUserDBNotConfigured()
}
appCode := appctx.FromContext(ctx)
whereSQL := `
FROM host_profiles hp
LEFT JOIN users u ON u.app_code = hp.app_code AND u.user_id = hp.user_id
LEFT JOIN agencies a ON a.app_code = hp.app_code AND a.agency_id = hp.current_agency_id
LEFT JOIN users agency_owner ON agency_owner.app_code = a.app_code AND agency_owner.user_id = a.owner_user_id
LEFT JOIN regions r ON r.app_code = hp.app_code AND r.region_id = hp.region_id
WHERE hp.app_code = ?`
args := []any{appCode}
if query.Status != "" {
whereSQL += " AND hp.status = ?"
args = append(args, query.Status)
}
if query.RegionID > 0 {
whereSQL += " AND hp.region_id = ?"
args = append(args, query.RegionID)
}
if query.AgencyID > 0 {
whereSQL += " AND hp.current_agency_id = ?"
args = append(args, query.AgencyID)
}
if keyword := strings.TrimSpace(query.Keyword); keyword != "" {
like := "%" + keyword + "%"
whereSQL += " AND (CAST(hp.user_id AS CHAR) LIKE ? OR u.current_display_user_id LIKE ? OR u.username LIKE ? OR a.name LIKE ? OR agency_owner.current_display_user_id LIKE ? OR agency_owner.username LIKE ? OR r.name LIKE ?)"
args = append(args, like, like, like, like, like, like, like)
}
total, err := countRows(ctx, r.db, whereSQL, args...)
if err != nil {
return nil, 0, err
}
limitSQL := "LIMIT ? OFFSET ?"
queryArgs := append(args, query.PageSize, offset(query.Page, query.PageSize))
if query.SortBy == "diamond" {
limitSQL = ""
queryArgs = args
}
rows, err := r.db.QueryContext(ctx, fmt.Sprintf(`
SELECT hp.user_id, hp.status, hp.region_id, COALESCE(hp.current_agency_id, 0),
COALESCE(hp.current_membership_id, 0), hp.source,
hp.first_became_host_at_ms, hp.created_at_ms, hp.updated_at_ms,
COALESCE(u.current_display_user_id, ''), COALESCE(u.username, ''),
COALESCE(u.avatar, ''), COALESCE(r.name, ''), COALESCE(a.name, ''),
COALESCE(a.owner_user_id, 0), COALESCE(agency_owner.current_display_user_id, ''),
COALESCE(agency_owner.username, ''), COALESCE(agency_owner.avatar, '')
%s
ORDER BY hp.created_at_ms DESC, hp.user_id DESC
%s
`, whereSQL, limitSQL), queryArgs...)
if err != nil {
return nil, 0, err
}
defer rows.Close()
items := make([]*userclient.HostProfile, 0, query.PageSize)
userIDs := make([]int64, 0, query.PageSize)
for rows.Next() {
item := &userclient.HostProfile{}
if err := rows.Scan(
&item.UserID,
&item.Status,
&item.RegionID,
&item.CurrentAgencyID,
&item.CurrentMembershipID,
&item.Source,
&item.FirstBecameHostAtMs,
&item.CreatedAtMs,
&item.UpdatedAtMs,
&item.DisplayUserID,
&item.Username,
&item.Avatar,
&item.RegionName,
&item.CurrentAgencyName,
&item.CurrentAgencyOwnerUserID,
&item.CurrentAgencyOwnerDisplayUserID,
&item.CurrentAgencyOwnerUsername,
&item.CurrentAgencyOwnerAvatar,
); err != nil {
return nil, 0, err
}
items = append(items, item)
userIDs = append(userIDs, item.UserID)
}
if err := rows.Err(); err != nil {
return nil, 0, err
}
diamonds, err := r.hostPeriodDiamonds(ctx, appCode, userIDs)
if err != nil {
return nil, 0, err
}
for _, item := range items {
item.Diamond = diamonds[item.UserID]
}
if query.SortBy == "diamond" {
sortHostProfilesByDiamond(items, query.SortDirection)
items = paginateHostProfiles(items, query.Page, query.PageSize)
}
return items, total, nil
}
// ListCoinSellers 聚合 user-service 身份事实和 wallet-service 币商余额,只做后台展示读模型。
func (r *Reader) ListCoinSellers(ctx context.Context, query listQuery) ([]*CoinSellerListItem, int64, error) {
if r == nil || r.db == nil {
return nil, 0, errUserDBNotConfigured()
}
appCode := appctx.FromContext(ctx)
if err := r.ensureCoinSellerContactColumn(ctx); err != nil {
return nil, 0, err
}
whereSQL := `
FROM coin_seller_profiles csp
LEFT JOIN users u ON u.app_code = csp.app_code AND u.user_id = csp.user_id
LEFT JOIN regions r ON r.app_code = csp.app_code AND r.region_id = u.region_id
WHERE csp.app_code = ?`
args := []any{appCode}
if query.Status != "" {
whereSQL += " AND csp.status = ?"
args = append(args, query.Status)
}
if query.RegionID > 0 {
whereSQL += " AND u.region_id = ?"
args = append(args, query.RegionID)
}
if keyword := strings.TrimSpace(query.Keyword); keyword != "" {
like := "%" + keyword + "%"
whereSQL += " AND (CAST(csp.user_id AS CHAR) LIKE ? OR u.current_display_user_id LIKE ? OR u.username LIKE ? OR csp.contact_info LIKE ?)"
args = append(args, like, like, like, like)
}
total, err := countRows(ctx, r.db, whereSQL, args...)
if err != nil {
return nil, 0, err
}
limitSQL := "LIMIT ? OFFSET ?"
queryArgs := append(args, query.PageSize, offset(query.Page, query.PageSize))
if isCoinSellerComputedSort(query.SortBy) {
// 币商余额和累充 USDT 都来自 wallet-service 聚合,必须先读取所有匹配币商再排序分页,避免只排序当前页。
limitSQL = ""
queryArgs = args
}
rows, err := r.db.QueryContext(ctx, fmt.Sprintf(`
SELECT csp.user_id, csp.status, csp.merchant_asset_type,
COALESCE(csp.contact_info, ''),
csp.created_by_user_id, csp.created_at_ms, csp.updated_at_ms,
COALESCE(u.current_display_user_id, ''), COALESCE(u.username, ''),
COALESCE(u.avatar, ''), COALESCE(u.region_id, 0),
COALESCE(r.name, '')
%s
ORDER BY csp.created_at_ms DESC, csp.user_id DESC
%s
`, whereSQL, limitSQL), queryArgs...)
if err != nil {
return nil, 0, err
}
defer rows.Close()
items := make([]*CoinSellerListItem, 0, query.PageSize)
userIDs := make([]int64, 0, query.PageSize)
for rows.Next() {
item := &CoinSellerListItem{}
if err := rows.Scan(
&item.UserID,
&item.Status,
&item.MerchantAssetType,
&item.Contact,
&item.CreatedByUserID,
&item.CreatedAtMs,
&item.UpdatedAtMs,
&item.DisplayUserID,
&item.Username,
&item.Avatar,
&item.RegionID,
&item.RegionName,
); err != nil {
return nil, 0, err
}
items = append(items, item)
userIDs = append(userIDs, item.UserID)
}
if err := rows.Err(); err != nil {
return nil, 0, err
}
balances, err := r.coinSellerBalances(ctx, appCode, userIDs)
if err != nil {
return nil, 0, err
}
rechargeTotals, err := r.coinSellerRechargeTotals(ctx, appCode, userIDs)
if err != nil {
return nil, 0, err
}
for _, item := range items {
item.MerchantBalance = balances[item.UserID]
item.TotalRechargeUSDTMicro = rechargeTotals[item.UserID]
}
if isCoinSellerComputedSort(query.SortBy) {
sortCoinSellerListItems(items, query.SortBy, query.SortDirection)
items = paginateCoinSellerListItems(items, query.Page, query.PageSize)
}
return items, total, nil
}
func (r *Reader) UpdateCoinSellerContact(ctx context.Context, userID int64, contact string) error {
if r == nil || r.db == nil {
return errUserDBNotConfigured()
}
if err := r.ensureCoinSellerContactColumn(ctx); err != nil {
return err
}
_, err := r.db.ExecContext(ctx, `
UPDATE coin_seller_profiles
SET contact_info = ?, updated_at_ms = ?
WHERE app_code = ? AND user_id = ?
`, strings.TrimSpace(contact), time.Now().UnixMilli(), appctx.FromContext(ctx), userID)
return err
}
func (r *Reader) ListCoinSellerSalaryRates(ctx context.Context, regionID int64) ([]CoinSellerSalaryRateTier, error) {
if r == nil || r.walletDB == nil {
return nil, fmt.Errorf("wallet mysql is not configured")
}
if regionID <= 0 {
return nil, fmt.Errorf("region_id is required")
}
if err := r.ensureCoinSellerSalaryRateTable(ctx); err != nil {
return nil, err
}
rows, err := r.walletDB.QueryContext(ctx, `
SELECT region_id, min_usd_minor, max_usd_minor, coin_per_usd, status, sort_order, updated_at_ms
FROM coin_seller_salary_exchange_rate_tiers
WHERE app_code = ? AND region_id = ?
ORDER BY sort_order ASC, min_usd_minor ASC, tier_id ASC
`, appctx.FromContext(ctx), regionID)
if err != nil {
return nil, err
}
defer rows.Close()
items := make([]CoinSellerSalaryRateTier, 0)
for rows.Next() {
var item CoinSellerSalaryRateTier
if err := rows.Scan(&item.RegionID, &item.MinUSDMinor, &item.MaxUSDMinor, &item.CoinPerUSD, &item.Status, &item.SortOrder, &item.UpdatedAtMs); err != nil {
return nil, err
}
item.Enabled = item.Status == "active"
item.MinUSD = formatUSDMinor(item.MinUSDMinor)
item.MaxUSD = formatUSDMinor(item.MaxUSDMinor)
items = append(items, item)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
func (r *Reader) ReplaceCoinSellerSalaryRates(ctx context.Context, regionID int64, actorID int64, tiers []CoinSellerSalaryRateTier) ([]CoinSellerSalaryRateTier, error) {
if r == nil || r.walletDB == nil {
return nil, fmt.Errorf("wallet mysql is not configured")
}
if regionID <= 0 {
return nil, fmt.Errorf("region_id is required")
}
if err := r.ensureCoinSellerSalaryRateTable(ctx); err != nil {
return nil, err
}
tx, err := r.walletDB.BeginTx(ctx, nil)
if err != nil {
return nil, err
}
defer func() { _ = tx.Rollback() }()
appCode := appctx.FromContext(ctx)
nowMs := time.Now().UnixMilli()
if _, err := tx.ExecContext(ctx, `
DELETE FROM coin_seller_salary_exchange_rate_tiers
WHERE app_code = ? AND region_id = ?
`, appCode, regionID); err != nil {
return nil, err
}
for index, tier := range tiers {
status := strings.ToLower(strings.TrimSpace(tier.Status))
if status == "" {
if tier.Enabled {
status = "active"
} else {
status = "disabled"
}
}
if status != "active" && status != "disabled" {
return nil, fmt.Errorf("rate status is invalid")
}
sortOrder := tier.SortOrder
if sortOrder == 0 {
sortOrder = (index + 1) * 10
}
_, err := tx.ExecContext(ctx, `
INSERT INTO coin_seller_salary_exchange_rate_tiers (
app_code, region_id, min_usd_minor, max_usd_minor, coin_per_usd,
status, sort_order, created_by_user_id, updated_by_user_id, created_at_ms, updated_at_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`, appCode, regionID, tier.MinUSDMinor, tier.MaxUSDMinor, tier.CoinPerUSD, status, sortOrder, actorID, actorID, nowMs, nowMs)
if err != nil {
return nil, err
}
}
if err := tx.Commit(); err != nil {
return nil, err
}
return r.ListCoinSellerSalaryRates(ctx, regionID)
}
func (r *Reader) ensureCoinSellerSalaryRateTable(ctx context.Context) error {
if _, err := r.walletDB.ExecContext(ctx, `
CREATE TABLE IF NOT EXISTS coin_seller_salary_exchange_rate_tiers (
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码,用于多租户隔离',
region_id BIGINT NOT NULL COMMENT '币商所属区域 ID',
tier_id BIGINT NOT NULL AUTO_INCREMENT COMMENT '区间 ID',
min_usd_minor BIGINT NOT NULL COMMENT '起始美元金额,单位美分,包含',
max_usd_minor BIGINT NOT NULL COMMENT '结束美元金额,单位美分,包含',
coin_per_usd BIGINT NOT NULL COMMENT '每 1 USD 工资可兑换的币商专用金币数量',
status VARCHAR(24) NOT NULL DEFAULT 'active' COMMENT '状态active/disabled',
sort_order INT NOT NULL DEFAULT 0 COMMENT '排序权重',
created_by_user_id BIGINT NOT NULL DEFAULT 0 COMMENT '创建人用户 ID',
updated_by_user_id BIGINT NOT NULL DEFAULT 0 COMMENT '更新人用户 ID',
created_at_ms BIGINT NOT NULL COMMENT '创建时间UTC epoch ms',
updated_at_ms BIGINT NOT NULL COMMENT '更新时间UTC epoch ms',
PRIMARY KEY (tier_id),
KEY idx_coin_seller_salary_rates_region (app_code, region_id, status, min_usd_minor, max_usd_minor),
KEY idx_coin_seller_salary_rates_sort (app_code, region_id, status, sort_order)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='币商工资兑换比例区间表'`); err != nil {
return err
}
return nil
}
func (r *Reader) ensureCoinSellerContactColumn(ctx context.Context) error {
var count int
if err := r.db.QueryRowContext(ctx, `
SELECT COUNT(*)
FROM information_schema.columns
WHERE table_schema = DATABASE()
AND table_name = 'coin_seller_profiles'
AND column_name = 'contact_info'
`).Scan(&count); err != nil {
return err
}
if count > 0 {
return nil
}
_, err := r.db.ExecContext(ctx, `
ALTER TABLE coin_seller_profiles
ADD COLUMN contact_info VARCHAR(128) NOT NULL DEFAULT '' AFTER merchant_asset_type
`)
return err
}
func (r *Reader) coinSellerBalances(ctx context.Context, appCode string, userIDs []int64) (map[int64]int64, error) {
result := make(map[int64]int64, len(userIDs))
if r == nil || r.walletDB == nil || len(userIDs) == 0 {
return result, nil
}
placeholders := strings.TrimRight(strings.Repeat("?,", len(userIDs)), ",")
args := []any{appCode, "COIN_SELLER_COIN"}
for _, userID := range userIDs {
args = append(args, userID)
}
rows, err := r.walletDB.QueryContext(ctx, fmt.Sprintf(`
SELECT user_id, available_amount
FROM wallet_accounts
WHERE app_code = ? AND asset_type = ? AND user_id IN (%s)
`, placeholders), args...)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var userID int64
var amount int64
if err := rows.Scan(&userID, &amount); err != nil {
return nil, err
}
result[userID] = amount
}
return result, rows.Err()
}
// coinSellerRechargeTotals 从币商进货专表汇总 USDT 付款,金币补偿不会计入累充口径。
func (r *Reader) coinSellerRechargeTotals(ctx context.Context, appCode string, userIDs []int64) (map[int64]int64, error) {
result := make(map[int64]int64, len(userIDs))
if r == nil || r.walletDB == nil || len(userIDs) == 0 {
return result, nil
}
placeholders := sqlPlaceholders(len(userIDs))
args := []any{appCode, paidCurrencyUSDT}
for _, userID := range userIDs {
args = append(args, userID)
}
rows, err := r.walletDB.QueryContext(ctx, fmt.Sprintf(`
SELECT seller_user_id, COALESCE(SUM(paid_amount_micro), 0)
FROM coin_seller_stock_records
WHERE app_code = ?
AND counts_as_seller_recharge = TRUE
AND paid_currency_code = ?
AND seller_user_id IN (%s)
GROUP BY seller_user_id
`, placeholders), args...)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var userID int64
var amount int64
if err := rows.Scan(&userID, &amount); err != nil {
return nil, err
}
result[userID] = amount
}
return result, rows.Err()
}
func (r *Reader) hostPeriodDiamonds(ctx context.Context, appCode string, userIDs []int64) (map[int64]int64, error) {
result := make(map[int64]int64, len(userIDs))
if r == nil || r.walletDB == nil || len(userIDs) == 0 {
return result, nil
}
placeholders := strings.TrimRight(strings.Repeat("?,", len(userIDs)), ",")
args := []any{appCode, time.Now().UTC().Format("2006-01")}
for _, userID := range userIDs {
args = append(args, userID)
}
rows, err := r.walletDB.QueryContext(ctx, fmt.Sprintf(`
SELECT user_id, total_diamonds
FROM host_period_diamond_accounts
WHERE app_code = ? AND cycle_key = ? AND user_id IN (%s)
`, placeholders), args...)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var userID int64
var amount int64
if err := rows.Scan(&userID, &amount); err != nil {
return nil, err
}
result[userID] = amount
}
return result, rows.Err()
}
func isCoinSellerComputedSort(sortBy string) bool {
return sortBy == sortByMerchantBalance || sortBy == sortByTotalRechargeUSDT
}
// sortCoinSellerListItems 只处理 wallet 聚合字段排序,基础身份字段仍使用 SQL 默认创建时间排序。
func sortCoinSellerListItems(items []*CoinSellerListItem, sortBy string, direction string) {
sort.SliceStable(items, func(i, j int) bool {
left := coinSellerSortValue(items[i], sortBy)
right := coinSellerSortValue(items[j], sortBy)
if left == right {
return false
}
if direction == "asc" {
return left < right
}
return left > right
})
}
func coinSellerSortValue(item *CoinSellerListItem, sortBy string) int64 {
if item == nil {
return 0
}
switch sortBy {
case sortByTotalRechargeUSDT:
return item.TotalRechargeUSDTMicro
case sortByMerchantBalance:
return item.MerchantBalance
default:
return 0
}
}
func sortHostProfilesByDiamond(items []*userclient.HostProfile, direction string) {
sort.SliceStable(items, func(i, j int) bool {
if items[i].Diamond == items[j].Diamond {
return false
}
if direction == "asc" {
return items[i].Diamond < items[j].Diamond
}
return items[i].Diamond > items[j].Diamond
})
}
func paginateCoinSellerListItems(items []*CoinSellerListItem, page int, pageSize int) []*CoinSellerListItem {
if page < 1 {
page = 1
}
if pageSize < 1 {
pageSize = 20
}
start := (page - 1) * pageSize
if start >= len(items) {
return []*CoinSellerListItem{}
}
end := start + pageSize
if end > len(items) {
end = len(items)
}
return items[start:end]
}
func paginateHostProfiles(items []*userclient.HostProfile, page int, pageSize int) []*userclient.HostProfile {
if page < 1 {
page = 1
}
if pageSize < 1 {
pageSize = 20
}
start := (page - 1) * pageSize
if start >= len(items) {
return []*userclient.HostProfile{}
}
end := start + pageSize
if end > len(items) {
end = len(items)
}
return items[start:end]
}
func sqlPlaceholders(count int) string {
if count <= 0 {
return ""
}
return strings.TrimRight(strings.Repeat("?,", count), ",")
}
func countRows(ctx context.Context, db *sql.DB, whereSQL string, args ...any) (int64, error) {
var total int64
err := db.QueryRowContext(ctx, "SELECT COUNT(*) "+whereSQL, args...).Scan(&total)
return total, err
}
func offset(page int, pageSize int) int {
return (page - 1) * pageSize
}
func errUserDBNotConfigured() error {
return fmt.Errorf("user mysql is not configured")
}