564 lines
18 KiB
Go
Raw Permalink 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 appuser
import (
"context"
"database/sql"
"errors"
"fmt"
"log/slog"
"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"`
Diamond int64 `json:"diamond"`
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
}
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
ORDER BY u.created_at_ms DESC, u.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([]AppUser, 0, query.PageSize)
userIDs := make([]int64, 0, query.PageSize)
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, 0, err
}
item.UserID = strconv.FormatInt(userID, 10)
items = append(items, item)
userIDs = append(userIDs, userID)
}
if err := rows.Err(); 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) 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 {
// 后台国家更新已经写入 usersIM 成员移除是外部副作用失败不回滚资料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, &regionID)
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
}
placeholders := strings.TrimRight(strings.Repeat("?,", len(userIDs)), ",")
args := make([]any, 0, len(userIDs)+2)
args = append(args, appctx.FromContext(ctx))
for _, id := range userIDs {
args = append(args, id)
}
args = append(args, "COIN", "DIAMOND")
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 IN (?, ?)
`, placeholders), args...)
if err != nil {
return err
}
defer rows.Close()
index := make(map[int64]int, len(userIDs))
for i, id := range userIDs {
index[id] = i
}
for rows.Next() {
var userID int64
var assetType string
var amount int64
if err := rows.Scan(&userID, &assetType, &amount); err != nil {
return err
}
i, ok := index[userID]
if !ok {
continue
}
switch strings.ToUpper(assetType) {
case "COIN":
items[i].Coin = amount
case "DIAMOND":
items[i].Diamond = amount
}
}
return rows.Err()
}
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))
return query
}
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()
}