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"` 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 事实表,避免后台列表需求改动 user-service 发布物。 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() } 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) 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, 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 FROM bd_profiles bp WHERE bp.app_code = ? AND bp.user_id = ? AND bp.role = 'bd_leader' `, 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_profiles WHERE app_code = ? AND user_id = ? AND role = 'bd_leader' `, 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) } 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 } 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 LIMIT ? OFFSET ? `, whereSQL), append(args, query.PageSize, offset(query.Page, query.PageSize))...) 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 } for _, item := range items { item.MerchantBalance = balances[item.UserID] } 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() } 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 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 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") }