package appuser import ( "context" "database/sql" "errors" "fmt" "log/slog" "sort" "strconv" "strings" "time" "hyapp-admin-server/internal/appctx" "hyapp-admin-server/internal/integration/activityclient" "hyapp-admin-server/internal/integration/userclient" "hyapp-admin-server/internal/security" activityv1 "hyapp.local/api/proto/activity/v1" ) var ( ErrInvalidArgument = errors.New("invalid argument") ErrNotFound = errors.New("not found") ) type Service struct { userClient userclient.Client activityClient activityclient.Client adminDB *sql.DB userDB *sql.DB walletDB *sql.DB } type AppUser struct { Avatar string `json:"avatar"` Coin int64 `json:"coin"` Country string `json:"country"` CountryDisplayName string `json:"countryDisplayName"` CountryName string `json:"countryName"` CreatedAtMs int64 `json:"createdAtMs"` DisplayUserID string `json:"displayUserId"` Gender string `json:"gender"` LastActiveAtMs int64 `json:"lastActiveAtMs"` RegionID int64 `json:"regionId"` RegionName string `json:"regionName"` Status string `json:"status"` UpdatedAtMs int64 `json:"updatedAtMs"` UserID string `json:"userId"` Username string `json:"username"` } type SetUserStatusResult struct { User AppUser `json:"user"` RevokedSessionCount int64 `json:"revokedSessionCount"` AccessTokenRevoked bool `json:"accessTokenRevoked"` AccessTokenError string `json:"accessTokenRevokeError,omitempty"` IMKicked bool `json:"imKicked"` IMKickError string `json:"imKickError,omitempty"` RoomEvicted bool `json:"roomEvicted"` RoomID string `json:"roomId,omitempty"` RTCKicked bool `json:"rtcKicked"` RTCKickError string `json:"rtcKickError,omitempty"` RoomEvictError string `json:"roomEvictError,omitempty"` } func NewService(userClient userclient.Client, activityClient activityclient.Client, adminDB *sql.DB, userDB *sql.DB, walletDB *sql.DB) *Service { return &Service{userClient: userClient, activityClient: activityClient, adminDB: adminDB, userDB: userDB, walletDB: walletDB} } func (s *Service) ListUsers(ctx context.Context, query listQuery) ([]AppUser, int64, error) { if s.userDB == nil { return nil, 0, fmt.Errorf("user mysql is not configured") } query = normalizeListQuery(query) appCode := appctx.FromContext(ctx) whereSQL := "FROM users u WHERE u.app_code = ?" args := []any{appCode} if query.Status != "" { whereSQL += " AND u.status = ?" args = append(args, query.Status) } if query.Keyword != "" { like := "%" + query.Keyword + "%" whereSQL += " AND (CAST(u.user_id AS CHAR) LIKE ? OR u.current_display_user_id LIKE ? OR u.username LIKE ?)" args = append(args, like, like, like) } total, err := countRows(ctx, s.userDB, whereSQL, args...) if err != nil { return nil, 0, err } if query.SortBy == "coin" { // 金币余额来自 wallet 库,必须先补齐当前筛选结果的余额再排序分页,避免只按当前页局部重排。 items, err := s.listUsersSortedByCoin(ctx, query, whereSQL, args) return items, total, err } rows, err := s.userDB.QueryContext(ctx, fmt.Sprintf(` SELECT u.user_id, u.current_display_user_id, COALESCE(u.username, ''), COALESCE(u.avatar, ''), COALESCE(u.gender, ''), COALESCE(u.country, ''), COALESCE((SELECT c.country_name FROM countries c WHERE c.app_code = u.app_code AND c.country_code = u.country LIMIT 1), ''), COALESCE((SELECT c.country_display_name FROM countries c WHERE c.app_code = u.app_code AND c.country_code = u.country LIMIT 1), ''), CASE WHEN COALESCE(u.region_id, 0) = 0 THEN 0 WHEN EXISTS ( SELECT 1 FROM regions rg WHERE rg.app_code = u.app_code AND rg.region_id = u.region_id AND rg.status = 'active' AND rg.region_code NOT IN ('Australia and New Zealand', 'Caribbean', 'Melanesia', 'Micronesia', 'Polynesia', 'Southern Africa', 'UNSPECIFIED') LIMIT 1 ) THEN u.region_id ELSE 0 END, CASE WHEN COALESCE(u.region_id, 0) = 0 OR NOT EXISTS ( SELECT 1 FROM regions rg WHERE rg.app_code = u.app_code AND rg.region_id = u.region_id AND rg.status = 'active' AND rg.region_code NOT IN ('Australia and New Zealand', 'Caribbean', 'Melanesia', 'Micronesia', 'Polynesia', 'Southern Africa', 'UNSPECIFIED') LIMIT 1 ) THEN 'GLOBAL' ELSE COALESCE((SELECT rg.name FROM regions rg WHERE rg.app_code = u.app_code AND rg.region_id = u.region_id LIMIT 1), '') END, u.status, u.created_at_ms, u.updated_at_ms, GREATEST( COALESCE((SELECT MAX(s.updated_at_ms) FROM auth_sessions s WHERE s.app_code = u.app_code AND s.user_id = u.user_id), 0), COALESCE((SELECT MAX(l.created_at_ms) FROM login_audit l WHERE l.app_code = u.app_code AND l.user_id = u.user_id), 0) ) %s %s LIMIT ? OFFSET ? `, whereSQL, appUserOrderSQL(query)), append(args, query.PageSize, offset(query.Page, query.PageSize))...) if err != nil { return nil, 0, err } defer rows.Close() items, userIDs, err := scanAppUserRows(rows, query.PageSize) if err != nil { return nil, 0, err } if err := s.fillBalances(ctx, items, userIDs); err != nil { return nil, 0, err } return items, total, nil } func (s *Service) listUsersSortedByCoin(ctx context.Context, query listQuery, whereSQL string, args []any) ([]AppUser, error) { rows, err := s.userDB.QueryContext(ctx, fmt.Sprintf(` SELECT u.user_id, u.current_display_user_id, COALESCE(u.username, ''), COALESCE(u.avatar, ''), COALESCE(u.gender, ''), COALESCE(u.country, ''), COALESCE((SELECT c.country_name FROM countries c WHERE c.app_code = u.app_code AND c.country_code = u.country LIMIT 1), ''), COALESCE((SELECT c.country_display_name FROM countries c WHERE c.app_code = u.app_code AND c.country_code = u.country LIMIT 1), ''), CASE WHEN COALESCE(u.region_id, 0) = 0 THEN 0 WHEN EXISTS ( SELECT 1 FROM regions rg WHERE rg.app_code = u.app_code AND rg.region_id = u.region_id AND rg.status = 'active' AND rg.region_code NOT IN ('Australia and New Zealand', 'Caribbean', 'Melanesia', 'Micronesia', 'Polynesia', 'Southern Africa', 'UNSPECIFIED') LIMIT 1 ) THEN u.region_id ELSE 0 END, CASE WHEN COALESCE(u.region_id, 0) = 0 OR NOT EXISTS ( SELECT 1 FROM regions rg WHERE rg.app_code = u.app_code AND rg.region_id = u.region_id AND rg.status = 'active' AND rg.region_code NOT IN ('Australia and New Zealand', 'Caribbean', 'Melanesia', 'Micronesia', 'Polynesia', 'Southern Africa', 'UNSPECIFIED') LIMIT 1 ) THEN 'GLOBAL' ELSE COALESCE((SELECT rg.name FROM regions rg WHERE rg.app_code = u.app_code AND rg.region_id = u.region_id LIMIT 1), '') END, u.status, u.created_at_ms, u.updated_at_ms, GREATEST( COALESCE((SELECT MAX(s.updated_at_ms) FROM auth_sessions s WHERE s.app_code = u.app_code AND s.user_id = u.user_id), 0), COALESCE((SELECT MAX(l.created_at_ms) FROM login_audit l WHERE l.app_code = u.app_code AND l.user_id = u.user_id), 0) ) %s `, whereSQL), args...) if err != nil { return nil, err } defer rows.Close() items, userIDs, err := scanAppUserRows(rows, query.PageSize) if err != nil { return nil, err } if err := s.fillBalances(ctx, items, userIDs); err != nil { return nil, err } sortAppUsersByCoin(items, query.SortDirection) return paginateAppUsers(items, query.Page, query.PageSize), nil } func scanAppUserRows(rows *sql.Rows, capacity int) ([]AppUser, []int64, error) { items := make([]AppUser, 0, capacity) userIDs := make([]int64, 0, capacity) for rows.Next() { var item AppUser var userID int64 if err := rows.Scan( &userID, &item.DisplayUserID, &item.Username, &item.Avatar, &item.Gender, &item.Country, &item.CountryName, &item.CountryDisplayName, &item.RegionID, &item.RegionName, &item.Status, &item.CreatedAtMs, &item.UpdatedAtMs, &item.LastActiveAtMs, ); err != nil { return nil, nil, err } item.UserID = strconv.FormatInt(userID, 10) items = append(items, item) userIDs = append(userIDs, userID) } if err := rows.Err(); err != nil { return nil, nil, err } return items, userIDs, nil } func (s *Service) GetUser(ctx context.Context, userID int64) (AppUser, error) { if s.userDB == nil { return AppUser{}, fmt.Errorf("user mysql is not configured") } var item AppUser var scannedUserID int64 appCode := appctx.FromContext(ctx) err := s.userDB.QueryRowContext(ctx, ` SELECT u.user_id, u.current_display_user_id, COALESCE(u.username, ''), COALESCE(u.avatar, ''), COALESCE(u.gender, ''), COALESCE(u.country, ''), COALESCE((SELECT c.country_name FROM countries c WHERE c.app_code = u.app_code AND c.country_code = u.country LIMIT 1), ''), COALESCE((SELECT c.country_display_name FROM countries c WHERE c.app_code = u.app_code AND c.country_code = u.country LIMIT 1), ''), CASE WHEN COALESCE(u.region_id, 0) = 0 THEN 0 WHEN EXISTS ( SELECT 1 FROM regions rg WHERE rg.app_code = u.app_code AND rg.region_id = u.region_id AND rg.status = 'active' AND rg.region_code NOT IN ('Australia and New Zealand', 'Caribbean', 'Melanesia', 'Micronesia', 'Polynesia', 'Southern Africa', 'UNSPECIFIED') LIMIT 1 ) THEN u.region_id ELSE 0 END, CASE WHEN COALESCE(u.region_id, 0) = 0 OR NOT EXISTS ( SELECT 1 FROM regions rg WHERE rg.app_code = u.app_code AND rg.region_id = u.region_id AND rg.status = 'active' AND rg.region_code NOT IN ('Australia and New Zealand', 'Caribbean', 'Melanesia', 'Micronesia', 'Polynesia', 'Southern Africa', 'UNSPECIFIED') LIMIT 1 ) THEN 'GLOBAL' ELSE COALESCE((SELECT rg.name FROM regions rg WHERE rg.app_code = u.app_code AND rg.region_id = u.region_id LIMIT 1), '') END, u.status, u.created_at_ms, u.updated_at_ms, GREATEST( COALESCE((SELECT MAX(s.updated_at_ms) FROM auth_sessions s WHERE s.app_code = u.app_code AND s.user_id = u.user_id), 0), COALESCE((SELECT MAX(l.created_at_ms) FROM login_audit l WHERE l.app_code = u.app_code AND l.user_id = u.user_id), 0) ) FROM users u WHERE u.app_code = ? AND u.user_id = ? `, appCode, userID).Scan( &scannedUserID, &item.DisplayUserID, &item.Username, &item.Avatar, &item.Gender, &item.Country, &item.CountryName, &item.CountryDisplayName, &item.RegionID, &item.RegionName, &item.Status, &item.CreatedAtMs, &item.UpdatedAtMs, &item.LastActiveAtMs, ) if errors.Is(err, sql.ErrNoRows) { return AppUser{}, ErrNotFound } if err != nil { return AppUser{}, err } item.UserID = strconv.FormatInt(scannedUserID, 10) items := []AppUser{item} if err := s.fillBalances(ctx, items, []int64{userID}); err != nil { return AppUser{}, err } return items[0], nil } func (s *Service) UpdateUser(ctx context.Context, userID int64, requestID string, req updateUserRequest) (AppUser, error) { if s.userDB == nil { return AppUser{}, fmt.Errorf("user mysql is not configured") } sets := make([]string, 0, 6) args := make([]any, 0, 8) if req.Username != nil { value := strings.TrimSpace(*req.Username) if len(value) > 64 { return AppUser{}, fmt.Errorf("%w: 用户名称不能超过 64 个字符", ErrInvalidArgument) } sets = append(sets, "username = ?") args = append(args, nullableString(value)) } if req.Avatar != nil { value := strings.TrimSpace(*req.Avatar) if len(value) > 512 { return AppUser{}, fmt.Errorf("%w: 头像不能超过 512 个字符", ErrInvalidArgument) } sets = append(sets, "avatar = ?") args = append(args, nullableString(value)) } if req.Gender != nil { value := strings.ToLower(strings.TrimSpace(*req.Gender)) if len(value) > 32 { return AppUser{}, fmt.Errorf("%w: 性别不能超过 32 个字符", ErrInvalidArgument) } sets = append(sets, "gender = ?") args = append(args, nullableString(value)) } oldRegionID := int64(0) if req.Country != nil { before, err := s.GetUser(ctx, userID) if err != nil { return AppUser{}, err } oldRegionID = before.RegionID country, regionID, err := s.resolveCountryRegion(ctx, *req.Country) if err != nil { return AppUser{}, err } sets = append(sets, "country = ?", "region_id = ?") args = append(args, nullableString(country), nullableInt64(regionID)) } if len(sets) == 0 { return s.GetUser(ctx, userID) } sets = append(sets, "updated_at_ms = ?") appCode := appctx.FromContext(ctx) args = append(args, nowMillis()) args = append(args, appCode, userID) result, err := s.userDB.ExecContext(ctx, "UPDATE users SET "+strings.Join(sets, ", ")+" WHERE app_code = ? AND user_id = ?", args...) if err != nil { return AppUser{}, err } affected, err := result.RowsAffected() if err != nil { return AppUser{}, err } if affected == 0 { return AppUser{}, ErrNotFound } updated, err := s.GetUser(ctx, userID) if err != nil { return AppUser{}, err } if req.Country != nil { s.removeOldRegionBroadcastMemberBestEffort(ctx, userID, oldRegionID, updated.RegionID, strings.TrimSpace(requestID), "admin_user_country_changed") } return updated, nil } func (s *Service) removeOldRegionBroadcastMemberBestEffort(ctx context.Context, userID int64, oldRegionID int64, newRegionID int64, requestID string, reason string) { if s == nil || s.activityClient == nil || userID <= 0 || oldRegionID <= 0 || oldRegionID == newRegionID { return } _, err := s.activityClient.RemoveRegionBroadcastMember(ctx, &activityv1.RemoveRegionBroadcastMemberRequest{ Meta: &activityv1.RequestMeta{ RequestId: strings.TrimSpace(requestID), Caller: "hyapp-admin-server", SentAtMs: nowMillis(), AppCode: appctx.FromContext(ctx), }, UserId: userID, RegionId: oldRegionID, Reason: reason, }) if err != nil { // 后台国家更新已经写入 users;IM 成员移除是外部副作用,失败不回滚资料,callback 会阻止用户再次加入旧区域群。 slog.WarnContext(ctx, "admin_im_region_group_member_remove_failed", slog.Int64("user_id", userID), slog.Int64("old_region_id", oldRegionID), slog.Int64("new_region_id", newRegionID), slog.String("request_id", requestID), slog.String("error", err.Error()), ) } } func (s *Service) SetUserStatus(ctx context.Context, userID int64, status string, operatorUserID int64, requestID string) (SetUserStatusResult, error) { status = strings.ToLower(strings.TrimSpace(status)) if status != "active" && status != "banned" && status != "disabled" { return SetUserStatusResult{}, fmt.Errorf("%w: 状态不正确", ErrInvalidArgument) } if s.userClient == nil { return SetUserStatusResult{}, fmt.Errorf("user service client is not configured") } result, err := s.userClient.SetUserStatus(ctx, userclient.SetUserStatusRequest{ RequestID: strings.TrimSpace(requestID), Caller: "hyapp-admin-server", TargetUserID: userID, Status: status, OperatorType: "admin", OperatorUserID: operatorUserID, Reason: "admin_app_user_status", }) if err != nil { return SetUserStatusResult{}, err } user, err := s.GetUser(ctx, userID) if err != nil { return SetUserStatusResult{}, err } output := SetUserStatusResult{User: user} if result != nil { output.RevokedSessionCount = result.RevokedSessionCount output.AccessTokenRevoked = result.AccessTokenRevoked output.AccessTokenError = result.AccessTokenError output.IMKicked = result.IMKicked output.IMKickError = result.IMKickError output.RoomEvicted = result.RoomEvicted output.RoomID = result.RoomID output.RTCKicked = result.RTCKicked output.RTCKickError = result.RTCKickError output.RoomEvictError = result.RoomEvictError } return output, nil } func (s *Service) SetPassword(ctx context.Context, userID int64, password string) error { if s.userDB == nil { return fmt.Errorf("user mysql is not configured") } password = strings.TrimSpace(password) if len(password) > 128 { return fmt.Errorf("%w: 密码不能超过 128 个字符", ErrInvalidArgument) } hash, err := security.HashPassword(password) if err != nil { return fmt.Errorf("%w: %v", ErrInvalidArgument, err) } var exists int appCode := appctx.FromContext(ctx) if err := s.userDB.QueryRowContext(ctx, `SELECT 1 FROM users WHERE app_code = ? AND user_id = ?`, appCode, userID).Scan(&exists); err != nil { if errors.Is(err, sql.ErrNoRows) { return ErrNotFound } return err } now := nowMillis() _, err = s.userDB.ExecContext(ctx, ` INSERT INTO password_accounts (app_code, user_id, password_hash, hash_alg, created_at_ms, updated_at_ms) VALUES (?, ?, ?, 'bcrypt', ?, ?) ON DUPLICATE KEY UPDATE password_hash = VALUES(password_hash), hash_alg = VALUES(hash_alg), updated_at_ms = VALUES(updated_at_ms) `, appCode, userID, hash, now, now) return err } func (s *Service) resolveCountryRegion(ctx context.Context, input string) (string, int64, error) { country := normalizeCountryCode(input) if country == "" { return "", 0, fmt.Errorf("%w: 请选择国家", ErrInvalidArgument) } if !validCountryCode(country) { return "", 0, fmt.Errorf("%w: 国家码必须是 2 到 3 位大写字母", ErrInvalidArgument) } appCode := appctx.FromContext(ctx) var canonical string var regionID int64 err := s.userDB.QueryRowContext(ctx, ` SELECT c.country_code, COALESCE(( SELECT rg.region_id FROM region_countries rc INNER JOIN regions rg ON rg.region_id = rc.region_id WHERE rc.app_code = c.app_code AND rg.app_code = c.app_code AND rc.country_code = c.country_code AND rc.status = 'active' AND rg.status = 'active' AND rg.region_code NOT IN ('Australia and New Zealand', 'Caribbean', 'Melanesia', 'Micronesia', 'Polynesia', 'Southern Africa', 'UNSPECIFIED') ORDER BY rg.sort_order ASC, rg.region_id ASC LIMIT 1 ), 0) FROM countries c WHERE c.app_code = ? AND c.country_code = ? AND c.enabled = TRUE `, appCode, country).Scan(&canonical, ®ionID) if errors.Is(err, sql.ErrNoRows) { return "", 0, fmt.Errorf("%w: 国家不存在或已停用", ErrInvalidArgument) } if err != nil { return "", 0, err } return canonical, regionID, nil } func (s *Service) fillBalances(ctx context.Context, items []AppUser, userIDs []int64) error { if s.walletDB == nil || len(userIDs) == 0 { return nil } index := make(map[int64]int, len(userIDs)) for i, id := range userIDs { index[id] = i } appCode := appctx.FromContext(ctx) const chunkSize = 500 for start := 0; start < len(userIDs); start += chunkSize { end := start + chunkSize if end > len(userIDs) { end = len(userIDs) } // 用户列表金币排序会对完整筛选集补余额,按块查询可以避免 IN 参数过长导致后台列表失败。 chunk := userIDs[start:end] placeholders := strings.TrimRight(strings.Repeat("?,", len(chunk)), ",") args := make([]any, 0, len(chunk)+2) args = append(args, appCode) for _, id := range chunk { args = append(args, id) } args = append(args, "COIN") rows, err := s.walletDB.QueryContext(ctx, fmt.Sprintf(` SELECT user_id, asset_type, available_amount FROM wallet_accounts WHERE app_code = ? AND user_id IN (%s) AND asset_type = ? `, placeholders), args...) if err != nil { return err } for rows.Next() { var userID int64 var assetType string var amount int64 if err := rows.Scan(&userID, &assetType, &amount); err != nil { _ = rows.Close() return err } i, ok := index[userID] if !ok { continue } switch strings.ToUpper(assetType) { case "COIN": items[i].Coin = amount } } if err := rows.Err(); err != nil { _ = rows.Close() return err } if err := rows.Close(); err != nil { return err } } return nil } 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.Keyword = strings.TrimSpace(query.Keyword) query.Status = strings.ToLower(strings.TrimSpace(query.Status)) query.SortBy = normalizeAppUserSortBy(query.SortBy) query.SortDirection = normalizeSortDirection(query.SortDirection) return query } func normalizeAppUserSortBy(value string) string { switch strings.ToLower(strings.TrimSpace(value)) { case "coin", "coins": return "coin" case "created_at", "createdat", "created_at_ms", "createdatms", "created": return "created_at" default: return "created_at" } } func normalizeSortDirection(value string) string { switch strings.ToLower(strings.TrimSpace(value)) { case "asc": return "asc" default: return "desc" } } func appUserOrderSQL(query listQuery) string { if query.SortBy == "created_at" && query.SortDirection == "asc" { return "ORDER BY u.created_at_ms ASC, u.user_id ASC" } return "ORDER BY u.created_at_ms DESC, u.user_id DESC" } func sortAppUsersByCoin(items []AppUser, direction string) { sort.SliceStable(items, func(i, j int) bool { left := items[i] right := items[j] if left.Coin != right.Coin { if direction == "asc" { return left.Coin < right.Coin } return left.Coin > right.Coin } // 金币相同的时候用内部用户 ID 做稳定排序,避免同余额用户在翻页和刷新时抖动。 leftID, _ := strconv.ParseInt(left.UserID, 10, 64) rightID, _ := strconv.ParseInt(right.UserID, 10, 64) if direction == "asc" { return leftID < rightID } return leftID > rightID }) } func paginateAppUsers(items []AppUser, page int, pageSize int) []AppUser { start := offset(page, pageSize) if start >= len(items) { return []AppUser{} } end := start + pageSize if end > len(items) { end = len(items) } return items[start:end] } 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 nullableString(value string) sql.Null[string] { return sql.Null[string]{V: value, Valid: value != ""} } func nullableInt64(value int64) sql.Null[int64] { return sql.Null[int64]{V: value, Valid: value > 0} } func normalizeCountryCode(value string) string { return strings.ToUpper(strings.TrimSpace(value)) } func validCountryCode(value string) bool { if len(value) < 2 || len(value) > 3 { return false } for _, char := range value { if char < 'A' || char > 'Z' { return false } } return true } func nowMillis() int64 { return time.Now().UnixMilli() } func rollbackIfOpen(tx *sql.Tx) { _ = tx.Rollback() }