983 lines
33 KiB
Go
983 lines
33 KiB
Go
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/modules/shared"
|
||
"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"`
|
||
Balances []AppUserAssetBalance `json:"balances,omitempty"`
|
||
Coin int64 `json:"coin"`
|
||
Country string `json:"country"`
|
||
CountryDisplayName string `json:"countryDisplayName"`
|
||
CountryName string `json:"countryName"`
|
||
CreatedAtMs int64 `json:"createdAtMs"`
|
||
DefaultDisplayUserID string `json:"defaultDisplayUserId"`
|
||
Diamond int64 `json:"diamond"`
|
||
DisplayUserID string `json:"displayUserId"`
|
||
EquippedResources []AppUserResource `json:"equippedResources,omitempty"`
|
||
Gender string `json:"gender"`
|
||
LastActiveAtMs int64 `json:"lastActiveAtMs"`
|
||
PrettyDisplayUserID string `json:"prettyDisplayUserId"`
|
||
PrettyID string `json:"prettyId"`
|
||
RegionID int64 `json:"regionId"`
|
||
RegionName string `json:"regionName"`
|
||
Resources []AppUserResource `json:"resources,omitempty"`
|
||
Status string `json:"status"`
|
||
UpdatedAtMs int64 `json:"updatedAtMs"`
|
||
UserID string `json:"userId"`
|
||
Username string `json:"username"`
|
||
VIP AppUserVIP `json:"vip"`
|
||
}
|
||
|
||
type AppUserAssetBalance struct {
|
||
AssetType string `json:"assetType"`
|
||
AvailableAmount int64 `json:"availableAmount"`
|
||
FrozenAmount int64 `json:"frozenAmount"`
|
||
TotalAmount int64 `json:"totalAmount"`
|
||
Version int64 `json:"version"`
|
||
UpdatedAtMs int64 `json:"updatedAtMs"`
|
||
}
|
||
|
||
type AppUserVIP struct {
|
||
Level int32 `json:"level"`
|
||
Name string `json:"name"`
|
||
Active bool `json:"active"`
|
||
StartedAtMs int64 `json:"startedAtMs"`
|
||
ExpiresAtMs int64 `json:"expiresAtMs"`
|
||
UpdatedAtMs int64 `json:"updatedAtMs"`
|
||
}
|
||
|
||
type AppUserResource struct {
|
||
EntitlementID string `json:"entitlementId"`
|
||
ResourceID int64 `json:"resourceId"`
|
||
ResourceCode string `json:"resourceCode"`
|
||
ResourceType string `json:"resourceType"`
|
||
Name string `json:"name"`
|
||
Status string `json:"status"`
|
||
Quantity int64 `json:"quantity"`
|
||
RemainingQuantity int64 `json:"remainingQuantity"`
|
||
EffectiveAtMs int64 `json:"effectiveAtMs"`
|
||
ExpiresAtMs int64 `json:"expiresAtMs"`
|
||
SourceGrantID string `json:"sourceGrantId"`
|
||
AssetURL string `json:"assetUrl"`
|
||
PreviewURL string `json:"previewUrl"`
|
||
AnimationURL string `json:"animationUrl"`
|
||
Equipped bool `json:"equipped"`
|
||
CreatedAtMs int64 `json:"createdAtMs"`
|
||
UpdatedAtMs int64 `json:"updatedAtMs"`
|
||
}
|
||
|
||
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)
|
||
nowMs := nowMillis()
|
||
whereSQL, args := appUserListWhereSQLAt(appCode, query, nowMs)
|
||
|
||
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, nowMs)
|
||
return items, total, err
|
||
}
|
||
rows, err := s.userDB.QueryContext(ctx, fmt.Sprintf(`
|
||
SELECT u.user_id,
|
||
u.current_display_user_id,
|
||
u.default_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
|
||
%s
|
||
LIMIT ? OFFSET ?
|
||
`, appUserPrettySelectColumns(), whereSQL, appUserOrderSQL(query)), append(appUserSelectArgs(nowMs, 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
|
||
}
|
||
if err := s.fillVIPLevels(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, nowMs int64) ([]AppUser, error) {
|
||
rows, err := s.userDB.QueryContext(ctx, fmt.Sprintf(`
|
||
SELECT u.user_id,
|
||
u.current_display_user_id,
|
||
u.default_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
|
||
`, appUserPrettySelectColumns(), whereSQL), appUserSelectArgs(nowMs, 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)
|
||
pagedItems := paginateAppUsers(items, query.Page, query.PageSize)
|
||
if err := s.fillVIPLevels(ctx, pagedItems, appUserNumericIDs(pagedItems)); err != nil {
|
||
return nil, err
|
||
}
|
||
return pagedItems, 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.DefaultDisplayUserID,
|
||
&item.Username,
|
||
&item.Avatar,
|
||
&item.Gender,
|
||
&item.Country,
|
||
&item.CountryName,
|
||
&item.CountryDisplayName,
|
||
&item.RegionID,
|
||
&item.RegionName,
|
||
&item.Status,
|
||
&item.CreatedAtMs,
|
||
&item.UpdatedAtMs,
|
||
&item.LastActiveAtMs,
|
||
&item.PrettyDisplayUserID,
|
||
&item.PrettyID,
|
||
); 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)
|
||
nowMs := nowMillis()
|
||
err := s.userDB.QueryRowContext(ctx, `
|
||
SELECT u.user_id,
|
||
u.current_display_user_id,
|
||
u.default_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)
|
||
)
|
||
`+appUserPrettySelectColumns()+`
|
||
FROM users u
|
||
WHERE u.app_code = ? AND u.user_id = ?
|
||
`, nowMs, nowMs, appCode, userID).Scan(
|
||
&scannedUserID,
|
||
&item.DisplayUserID,
|
||
&item.DefaultDisplayUserID,
|
||
&item.Username,
|
||
&item.Avatar,
|
||
&item.Gender,
|
||
&item.Country,
|
||
&item.CountryName,
|
||
&item.CountryDisplayName,
|
||
&item.RegionID,
|
||
&item.RegionName,
|
||
&item.Status,
|
||
&item.CreatedAtMs,
|
||
&item.UpdatedAtMs,
|
||
&item.LastActiveAtMs,
|
||
&item.PrettyDisplayUserID,
|
||
&item.PrettyID,
|
||
)
|
||
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
|
||
}
|
||
if err := s.fillVIPLevels(ctx, items, []int64{userID}); err != nil {
|
||
return AppUser{}, err
|
||
}
|
||
if err := s.fillUserResources(ctx, items, userID); err != nil {
|
||
return AppUser{}, err
|
||
}
|
||
return items[0], nil
|
||
}
|
||
|
||
func (s *Service) UpdateUser(ctx context.Context, userID int64, adminUserID int64, requestID string, req updateUserRequest) (AppUser, error) {
|
||
if s.userDB == nil {
|
||
return AppUser{}, fmt.Errorf("user mysql is not configured")
|
||
}
|
||
sets := make([]string, 0, 4)
|
||
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))
|
||
}
|
||
|
||
if len(sets) > 0 {
|
||
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
|
||
}
|
||
}
|
||
|
||
if req.Country != nil {
|
||
if s.userClient == nil {
|
||
return AppUser{}, fmt.Errorf("user client is not configured")
|
||
}
|
||
change, err := s.userClient.AdminChangeUserCountry(ctx, userclient.AdminChangeUserCountryRequest{
|
||
RequestID: strings.TrimSpace(requestID),
|
||
Caller: "hyapp-admin-server",
|
||
TargetUserID: userID,
|
||
Country: *req.Country,
|
||
AdminUserID: adminUserID,
|
||
Reason: "admin_user_country_changed",
|
||
})
|
||
if err != nil {
|
||
return AppUser{}, err
|
||
}
|
||
if change != nil && change.RegionChanged {
|
||
s.removeOldRegionBroadcastMemberBestEffort(ctx, userID, change.OldRegionID, change.NewRegionID, strings.TrimSpace(requestID), "admin_user_country_changed")
|
||
}
|
||
}
|
||
if len(sets) == 0 && req.Country == nil {
|
||
return s.GetUser(ctx, userID)
|
||
}
|
||
return s.GetUser(ctx, userID)
|
||
}
|
||
|
||
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) 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]
|
||
args := make([]any, 0, len(chunk)+1)
|
||
args = append(args, appCode)
|
||
for _, id := range chunk {
|
||
args = append(args, id)
|
||
}
|
||
rows, err := s.walletDB.QueryContext(ctx, fmt.Sprintf(`
|
||
SELECT user_id, asset_type, available_amount, frozen_amount, version, updated_at_ms
|
||
FROM wallet_accounts
|
||
WHERE app_code = ? AND user_id IN (%s)
|
||
ORDER BY user_id ASC, asset_type ASC
|
||
`, placeholders(len(chunk))), args...)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for rows.Next() {
|
||
var userID int64
|
||
var balance AppUserAssetBalance
|
||
if err := rows.Scan(
|
||
&userID,
|
||
&balance.AssetType,
|
||
&balance.AvailableAmount,
|
||
&balance.FrozenAmount,
|
||
&balance.Version,
|
||
&balance.UpdatedAtMs,
|
||
); err != nil {
|
||
_ = rows.Close()
|
||
return err
|
||
}
|
||
i, ok := index[userID]
|
||
if !ok {
|
||
continue
|
||
}
|
||
balance.AssetType = strings.ToUpper(strings.TrimSpace(balance.AssetType))
|
||
balance.TotalAmount = balance.AvailableAmount + balance.FrozenAmount
|
||
// 钱包资产明细作为详情页“全部资产”的余额来源；金币和钻石仍同步到顶层字段，兼容列表排序和旧页面读取。
|
||
items[i].Balances = append(items[i].Balances, balance)
|
||
switch balance.AssetType {
|
||
case "COIN":
|
||
items[i].Coin = balance.AvailableAmount
|
||
case "DIAMOND":
|
||
items[i].Diamond = balance.AvailableAmount
|
||
}
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
_ = rows.Close()
|
||
return err
|
||
}
|
||
if err := rows.Close(); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *Service) fillVIPLevels(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)
|
||
nowMs := nowMillis()
|
||
const chunkSize = 500
|
||
for start := 0; start < len(userIDs); start += chunkSize {
|
||
end := start + chunkSize
|
||
if end > len(userIDs) {
|
||
end = len(userIDs)
|
||
}
|
||
chunk := userIDs[start:end]
|
||
args := make([]any, 0, len(chunk)+1)
|
||
args = append(args, appCode)
|
||
for _, id := range chunk {
|
||
args = append(args, id)
|
||
}
|
||
rows, err := s.walletDB.QueryContext(ctx, fmt.Sprintf(`
|
||
SELECT m.user_id, m.level, COALESCE(v.name, m.name), m.started_at_ms, m.expires_at_ms, m.updated_at_ms
|
||
FROM user_vip_memberships m
|
||
LEFT JOIN vip_levels v ON v.app_code = m.app_code AND v.level = m.level
|
||
WHERE m.app_code = ? AND m.user_id IN (%s)
|
||
`, placeholders(len(chunk))), args...)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for rows.Next() {
|
||
var userID int64
|
||
var vip AppUserVIP
|
||
if err := rows.Scan(&userID, &vip.Level, &vip.Name, &vip.StartedAtMs, &vip.ExpiresAtMs, &vip.UpdatedAtMs); err != nil {
|
||
_ = rows.Close()
|
||
return err
|
||
}
|
||
i, ok := index[userID]
|
||
if !ok {
|
||
continue
|
||
}
|
||
// 这里刻意沿用 wallet-service GetMyVip 的展示口径：只看 level 和 expires_at_ms 计算当前 VIP，
|
||
// 过期会员仍保留历史行，但后台用户列表不展示为有效等级。
|
||
vip.Active = vip.Level > 0 && vip.ExpiresAtMs > nowMs
|
||
if !vip.Active {
|
||
vip.Level = 0
|
||
vip.Name = ""
|
||
}
|
||
items[i].VIP = vip
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
_ = rows.Close()
|
||
return err
|
||
}
|
||
if err := rows.Close(); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *Service) fillUserResources(ctx context.Context, items []AppUser, userID int64) error {
|
||
if s.walletDB == nil || len(items) == 0 || userID <= 0 {
|
||
return nil
|
||
}
|
||
appCode := appctx.FromContext(ctx)
|
||
nowMs := nowMillis()
|
||
rows, err := s.walletDB.QueryContext(ctx, `
|
||
SELECT e.entitlement_id, e.resource_id, r.resource_code, r.resource_type, r.name, e.status,
|
||
e.quantity, e.remaining_quantity, e.effective_at_ms, e.expires_at_ms, e.source_grant_id,
|
||
COALESCE(r.asset_url, ''), COALESCE(r.preview_url, ''), COALESCE(r.animation_url, ''),
|
||
CASE WHEN eq.entitlement_id IS NULL THEN FALSE ELSE TRUE END AS equipped,
|
||
e.created_at_ms, e.updated_at_ms
|
||
FROM user_resource_entitlements e
|
||
JOIN resources r ON r.app_code = e.app_code AND r.resource_id = e.resource_id
|
||
LEFT JOIN user_resource_equipment eq
|
||
ON eq.app_code = e.app_code
|
||
AND eq.user_id = e.user_id
|
||
AND eq.resource_id = e.resource_id
|
||
AND eq.entitlement_id = e.entitlement_id
|
||
WHERE e.app_code = ? AND e.user_id = ?
|
||
AND e.status = 'active'
|
||
AND e.effective_at_ms <= ? AND (e.expires_at_ms = 0 OR e.expires_at_ms > ?)
|
||
AND e.remaining_quantity > 0
|
||
AND r.status = 'active'
|
||
ORDER BY equipped DESC, r.resource_type ASC, e.updated_at_ms DESC, e.created_at_ms DESC
|
||
`, appCode, userID, nowMs, nowMs)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer rows.Close()
|
||
|
||
resources := make([]AppUserResource, 0)
|
||
equipped := make([]AppUserResource, 0)
|
||
for rows.Next() {
|
||
var resource AppUserResource
|
||
if err := rows.Scan(
|
||
&resource.EntitlementID,
|
||
&resource.ResourceID,
|
||
&resource.ResourceCode,
|
||
&resource.ResourceType,
|
||
&resource.Name,
|
||
&resource.Status,
|
||
&resource.Quantity,
|
||
&resource.RemainingQuantity,
|
||
&resource.EffectiveAtMs,
|
||
&resource.ExpiresAtMs,
|
||
&resource.SourceGrantID,
|
||
&resource.AssetURL,
|
||
&resource.PreviewURL,
|
||
&resource.AnimationURL,
|
||
&resource.Equipped,
|
||
&resource.CreatedAtMs,
|
||
&resource.UpdatedAtMs,
|
||
); err != nil {
|
||
return err
|
||
}
|
||
resources = append(resources, resource)
|
||
if resource.Equipped {
|
||
// 装备状态只从 user_resource_equipment 读取；前端可以直接展示当前佩戴，不需要再按类型猜测。
|
||
equipped = append(equipped, resource)
|
||
}
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return err
|
||
}
|
||
items[0].Resources = resources
|
||
items[0].EquippedResources = equipped
|
||
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.Country = normalizeCountryCode(query.Country)
|
||
query.Keyword = strings.TrimSpace(query.Keyword)
|
||
query.UserFilter.UserID = strings.TrimSpace(query.UserFilter.UserID)
|
||
query.UserFilter.DisplayUserID = strings.TrimSpace(query.UserFilter.DisplayUserID)
|
||
query.UserFilter.Username = strings.TrimSpace(query.UserFilter.Username)
|
||
if query.RegionID < 0 {
|
||
query.RegionID = 0
|
||
query.RegionIDSet = false
|
||
}
|
||
if query.StartMs < 0 {
|
||
query.StartMs = 0
|
||
}
|
||
if query.EndMs < 0 {
|
||
query.EndMs = 0
|
||
}
|
||
query.Status = strings.ToLower(strings.TrimSpace(query.Status))
|
||
query.SortBy = normalizeAppUserSortBy(query.SortBy)
|
||
query.SortDirection = normalizeSortDirection(query.SortDirection)
|
||
return query
|
||
}
|
||
|
||
func appUserListWhereSQL(appCode string, query listQuery) (string, []any) {
|
||
return appUserListWhereSQLAt(appCode, query, nowMillis())
|
||
}
|
||
|
||
func appUserListWhereSQLAt(appCode string, query listQuery, nowMs int64) (string, []any) {
|
||
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.UserFilter.IsEmpty() {
|
||
matchSQL, matchArgs := shared.UserIdentityFieldsSQL("u", "u.user_id", query.UserFilter, nowMs)
|
||
whereSQL += " AND " + matchSQL
|
||
args = append(args, matchArgs...)
|
||
} else if query.Keyword != "" {
|
||
matchSQL, matchArgs := shared.UserIdentityKeywordSQL("u", "u.user_id", query.Keyword, nowMs)
|
||
whereSQL += " AND " + matchSQL
|
||
args = append(args, matchArgs...)
|
||
}
|
||
if query.Country != "" {
|
||
// 国家筛选用 users.country 的 ISO code 精确命中；前端传展示名时不会被猜测转换，避免误筛到错误国家。
|
||
whereSQL += " AND UPPER(u.country) = ?"
|
||
args = append(args, query.Country)
|
||
}
|
||
if query.RegionIDSet {
|
||
if query.RegionID == 0 {
|
||
// 列表展示里无区域或无有效业务区域都会显示为 GLOBAL；筛选 GLOBAL 时保持同一套展示口径。
|
||
whereSQL += " AND (COALESCE(u.region_id, 0) = 0 OR NOT " + appUserValidRegionExistsSQL("u") + ")"
|
||
} else {
|
||
whereSQL += " AND u.region_id = ?"
|
||
args = append(args, query.RegionID)
|
||
}
|
||
}
|
||
if query.StartMs > 0 {
|
||
// 时间区间筛选的是用户创建时间，和列表“创建 / 活跃”列的第一行保持一致。
|
||
whereSQL += " AND u.created_at_ms >= ?"
|
||
args = append(args, query.StartMs)
|
||
}
|
||
if query.EndMs > 0 {
|
||
whereSQL += " AND u.created_at_ms < ?"
|
||
args = append(args, query.EndMs)
|
||
}
|
||
return whereSQL, args
|
||
}
|
||
|
||
func appUserValidRegionExistsSQL(userAlias string) string {
|
||
return fmt.Sprintf(`EXISTS (
|
||
SELECT 1 FROM regions rg
|
||
WHERE rg.app_code = %[1]s.app_code
|
||
AND rg.region_id = %[1]s.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
|
||
)`, userAlias)
|
||
}
|
||
|
||
func appUserPrettySelectColumns() string {
|
||
return `,
|
||
COALESCE((
|
||
SELECT lease.display_user_id
|
||
FROM pretty_display_user_id_leases lease
|
||
WHERE lease.app_code = u.app_code
|
||
AND lease.user_id = u.user_id
|
||
AND lease.status = 'active'
|
||
AND (COALESCE(lease.expires_at_ms, 0) = 0 OR lease.expires_at_ms > ?)
|
||
ORDER BY lease.created_at_ms DESC, lease.lease_id DESC
|
||
LIMIT 1
|
||
), ''),
|
||
COALESCE((
|
||
SELECT pdi.pretty_id
|
||
FROM pretty_display_user_id_leases lease
|
||
JOIN pretty_display_ids pdi
|
||
ON pdi.app_code = lease.app_code
|
||
AND pdi.assigned_lease_id = lease.lease_id
|
||
AND pdi.status = 'assigned'
|
||
WHERE lease.app_code = u.app_code
|
||
AND lease.user_id = u.user_id
|
||
AND lease.status = 'active'
|
||
AND (COALESCE(lease.expires_at_ms, 0) = 0 OR lease.expires_at_ms > ?)
|
||
ORDER BY lease.created_at_ms DESC, lease.lease_id DESC
|
||
LIMIT 1
|
||
), '')`
|
||
}
|
||
|
||
func appUserSelectArgs(nowMs int64, whereArgs []any) []any {
|
||
// 用户资料页展示“原始短号 + 靓号”时，靓号以租约表为准读取，不能只依赖 users.current_display_user_id。
|
||
// 本地 initdb 或历史修复脚本可能只重放 users 快照；active lease 和 pretty_display_ids 才是当前靓号归属的审计来源。
|
||
args := make([]any, 0, len(whereArgs)+2)
|
||
args = append(args, nowMs, nowMs)
|
||
args = append(args, whereArgs...)
|
||
return args
|
||
}
|
||
|
||
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 appUserNumericIDs(items []AppUser) []int64 {
|
||
ids := make([]int64, 0, len(items))
|
||
for _, item := range items {
|
||
id, err := strconv.ParseInt(item.UserID, 10, 64)
|
||
if err != nil || id <= 0 {
|
||
continue
|
||
}
|
||
ids = append(ids, id)
|
||
}
|
||
return ids
|
||
}
|
||
|
||
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 normalizeCountryCode(value string) string {
|
||
return strings.ToUpper(strings.TrimSpace(value))
|
||
}
|
||
|
||
func nowMillis() int64 {
|
||
return time.Now().UnixMilli()
|
||
}
|
||
|
||
func rollbackIfOpen(tx *sql.Tx) {
|
||
_ = tx.Rollback()
|
||
}
|