696 lines
25 KiB
Go
696 lines
25 KiB
Go
package user
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
invitedomain "hyapp/services/user-service/internal/domain/invite"
|
||
userdomain "hyapp/services/user-service/internal/domain/user"
|
||
invitestorage "hyapp/services/user-service/internal/storage/mysql/invite"
|
||
"hyapp/services/user-service/internal/storage/mysql/shared"
|
||
)
|
||
|
||
const countryChangeSessionRevokeReason = "USER_COUNTRY_CHANGED"
|
||
|
||
func (r *Repository) GetUser(ctx context.Context, userID int64) (userdomain.User, error) {
|
||
// 普通查询不加锁，靓号过期恢复由 service 层判断后调用 identity 事务。
|
||
user, err := QueryUser(ctx, r.db, "WHERE user_id = ?", userID)
|
||
if err == sql.ErrNoRows {
|
||
// users 表无记录时返回稳定 NotFound。
|
||
return userdomain.User{}, xerr.New(xerr.NotFound, "user not found")
|
||
}
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
if err := r.hydrateInviteOverview(ctx, &user); err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
|
||
return user, nil
|
||
}
|
||
|
||
// GetMyProfileStats 只读取预聚合统计表，不能实时扫描访客、关注或好友明细。
|
||
func (r *Repository) GetMyProfileStats(ctx context.Context, userID int64) (userdomain.ProfileStats, error) {
|
||
if r == nil || r.db == nil {
|
||
return userdomain.ProfileStats{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
appCode := appcode.FromContext(ctx)
|
||
stats := userdomain.ProfileStats{AppCode: appCode, UserID: userID}
|
||
err := r.db.QueryRowContext(ctx, `
|
||
SELECT visitors_count, following_count, friends_count, updated_at_ms
|
||
FROM user_profile_stats
|
||
WHERE app_code = ? AND user_id = ?`,
|
||
appCode, userID,
|
||
).Scan(&stats.VisitorsCount, &stats.FollowingCount, &stats.FriendsCount, &stats.UpdatedAtMs)
|
||
if err == sql.ErrNoRows {
|
||
return stats, nil
|
||
}
|
||
if err != nil {
|
||
return userdomain.ProfileStats{}, err
|
||
}
|
||
return stats, nil
|
||
}
|
||
|
||
// BatchGetUsers 批量读取用户主状态，缺失用户不会出现在返回 map 中。
|
||
|
||
func (r *Repository) BatchGetUsers(ctx context.Context, userIDs []int64) (map[int64]userdomain.User, error) {
|
||
// 调用方已校验 userIDs 非空且合法，这里直接构造 IN 查询。
|
||
placeholders := make([]string, 0, len(userIDs))
|
||
args := make([]any, 0, len(userIDs)+1)
|
||
args = append(args, appcode.FromContext(ctx))
|
||
for _, userID := range userIDs {
|
||
placeholders = append(placeholders, "?")
|
||
args = append(args, userID)
|
||
}
|
||
|
||
rows, err := r.db.QueryContext(ctx, fmt.Sprintf(`
|
||
SELECT %s
|
||
FROM users
|
||
WHERE app_code = ? AND user_id IN (%s)
|
||
`, userSelectColumns, strings.Join(placeholders, ",")), args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
result := make(map[int64]userdomain.User, len(userIDs))
|
||
for rows.Next() {
|
||
// 缺失用户不会返回，调用方按 map key 判断存在性。
|
||
user, err := ScanUser(rows)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
result[user.UserID] = user
|
||
}
|
||
|
||
return result, rows.Err()
|
||
}
|
||
|
||
// BatchGetRoomBasicUsers 只读取房间首屏需要的基础展示字段，避免 JoinRoom lite 触发完整用户投影。
|
||
func (r *Repository) BatchGetRoomBasicUsers(ctx context.Context, userIDs []int64) (map[int64]userdomain.RoomBasicUser, error) {
|
||
placeholders := make([]string, 0, len(userIDs))
|
||
args := make([]any, 0, len(userIDs)+2)
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
args = append(args, nowMS)
|
||
for _, userID := range userIDs {
|
||
placeholders = append(placeholders, "?")
|
||
args = append(args, userID)
|
||
}
|
||
args = append(args, appcode.FromContext(ctx))
|
||
|
||
rows, err := r.db.QueryContext(ctx, fmt.Sprintf(`
|
||
SELECT user_id,
|
||
COALESCE(username, ''),
|
||
COALESCE(avatar, ''),
|
||
CASE
|
||
WHEN current_display_user_id_kind <> 'default'
|
||
AND COALESCE(current_display_user_id_expires_at_ms, 0) > 0
|
||
AND current_display_user_id_expires_at_ms <= ?
|
||
THEN default_display_user_id
|
||
ELSE current_display_user_id
|
||
END
|
||
FROM users FORCE INDEX (PRIMARY)
|
||
WHERE user_id IN (%s)
|
||
AND app_code = ?
|
||
`, strings.Join(placeholders, ",")), args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
result := make(map[int64]userdomain.RoomBasicUser, len(userIDs))
|
||
for rows.Next() {
|
||
var user userdomain.RoomBasicUser
|
||
if err := rows.Scan(&user.UserID, &user.Username, &user.Avatar, &user.DisplayUserID); err != nil {
|
||
return nil, err
|
||
}
|
||
result[user.UserID] = user
|
||
}
|
||
return result, rows.Err()
|
||
}
|
||
|
||
// ListUserIDs returns active user IDs in ascending order for cursor-based backend fanout.
|
||
func (r *Repository) ListUserIDs(ctx context.Context, filter userdomain.UserIDPageFilter) ([]int64, error) {
|
||
conditions := []string{"app_code = ?", "status = ?", "user_id > ?"}
|
||
args := []any{appcode.FromContext(ctx), string(userdomain.StatusActive), filter.CursorUserID}
|
||
switch filter.TargetScope {
|
||
case userdomain.UserIDTargetRegion:
|
||
conditions = append(conditions, "region_id = ?")
|
||
args = append(args, filter.RegionID)
|
||
case userdomain.UserIDTargetCountry:
|
||
conditions = append(conditions, "country = ?")
|
||
args = append(args, filter.Country)
|
||
}
|
||
args = append(args, filter.PageSize)
|
||
|
||
rows, err := r.db.QueryContext(ctx, `
|
||
SELECT user_id
|
||
FROM users
|
||
WHERE `+strings.Join(conditions, " AND ")+`
|
||
ORDER BY user_id ASC
|
||
LIMIT ?`, args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
userIDs := make([]int64, 0, filter.PageSize)
|
||
for rows.Next() {
|
||
var userID int64
|
||
if err := rows.Scan(&userID); err != nil {
|
||
return nil, err
|
||
}
|
||
userIDs = append(userIDs, userID)
|
||
}
|
||
return userIDs, rows.Err()
|
||
}
|
||
|
||
// CreateUserWithIdentity 是用户创建的事务入口。
|
||
// 它同时写 users 和默认短号 active 记录，避免出现没有默认门牌号的用户。
|
||
|
||
func (r *Repository) CreateUserWithIdentity(ctx context.Context, user userdomain.User, identity userdomain.Identity) error {
|
||
// 用户和默认短号必须在一个事务里创建，避免出现孤立用户。
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer tx.Rollback()
|
||
|
||
if err := InsertUserIdentity(ctx, tx, user, identity); err != nil {
|
||
// 唯一键冲突转换成 display_user_id exists，供 service 有限重试。
|
||
return shared.MapDisplayDuplicateError(err)
|
||
}
|
||
if _, err := invitestorage.EnsurePrimaryCodeForUser(ctx, tx, user.AppCode, user.UserID, user.CreatedAtMs); err != nil {
|
||
return err
|
||
}
|
||
if err := InsertUserRegisteredOutbox(ctx, tx, user); err != nil {
|
||
return err
|
||
}
|
||
|
||
return tx.Commit()
|
||
}
|
||
|
||
// InsertUserRegisteredOutbox 写入 App 自然注册完成事实。
|
||
// 调用方必须放在用户资料完成事务内执行；三方首次登录只创建 profile_required 占位账号，不能提前进入新增用户统计。
|
||
func InsertUserRegisteredOutbox(ctx context.Context, tx *sql.Tx, user userdomain.User) error {
|
||
if userdomain.ExcludedFromRegistrationStats(user.RegisterSource) {
|
||
// 全站机器人和后台快捷账号只复用用户资料、头像和短号参与运营或联调，不是真实 App 自然注册；
|
||
// 这里不写 UserRegistered 事实，避免实时统计、留存 cohort 和注册奖励把后台造号当新增用户。
|
||
return nil
|
||
}
|
||
if !user.ProfileCompleted {
|
||
// 用户填完 username/avatar/gender/country 并固化 region_id 前，不能算注册成功；
|
||
// 否则统计服务会把三方占位账号按空国家落进“未知国家”，且后续补资料无法修正 cohort。
|
||
return nil
|
||
}
|
||
countryID, err := userRegisteredCountryID(ctx, tx, user)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
registeredAtMS := user.ProfileCompletedAtMs
|
||
if registeredAtMS <= 0 {
|
||
// 兼容少量历史/测试路径直接创建 completed 用户但未写完成时间；事件时间至少保持有效，避免 MQ 编码失败。
|
||
registeredAtMS = user.CreatedAtMs
|
||
}
|
||
payload := map[string]any{
|
||
"user_id": user.UserID,
|
||
"default_display_user_id": user.DefaultDisplayUserID,
|
||
"country": user.Country,
|
||
"country_id": countryID,
|
||
"country_by_ip": user.CountryByIP,
|
||
"region_id": user.RegionID,
|
||
"invite_code": user.InviteCode,
|
||
"register_device_id": user.RegisterDeviceID,
|
||
"register_platform": user.RegisterPlatform,
|
||
"register_install_channel": user.RegisterInstallChannel,
|
||
"register_campaign": user.RegisterCampaign,
|
||
"created_at_ms": user.CreatedAtMs,
|
||
"profile_completed_at_ms": registeredAtMS,
|
||
}
|
||
payloadJSON, err := json.Marshal(payload)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = tx.ExecContext(ctx, `
|
||
INSERT IGNORE INTO user_outbox (app_code, event_id, event_type, aggregate_type, aggregate_id, status, payload_json, created_at_ms, updated_at_ms)
|
||
VALUES (?, ?, 'UserRegistered', 'user', ?, 'pending', ?, ?, ?)
|
||
`, appcode.Normalize(user.AppCode), fmt.Sprintf("UserRegistered:%d", user.UserID), user.UserID, string(payloadJSON), registeredAtMS, registeredAtMS)
|
||
return err
|
||
}
|
||
|
||
func userRegisteredCountryID(ctx context.Context, tx *sql.Tx, user userdomain.User) (int64, error) {
|
||
if user.CountryID > 0 {
|
||
// 查询路径已经带出国家主数据时直接复用，避免多一次 SQL。
|
||
return user.CountryID, nil
|
||
}
|
||
country := strings.ToUpper(strings.TrimSpace(user.Country))
|
||
if country == "" {
|
||
// 正常 onboarding 完成路径必须有国家；这里只给历史或测试 completed 用户保留全局兜底，避免事件写入失败。
|
||
return 0, nil
|
||
}
|
||
var countryID int64
|
||
err := tx.QueryRowContext(ctx, `
|
||
SELECT country_id
|
||
FROM countries
|
||
WHERE app_code = ? AND country_code = ?
|
||
LIMIT 1
|
||
`, appcode.Normalize(user.AppCode), country).Scan(&countryID)
|
||
if err == sql.ErrNoRows {
|
||
// 国家码格式合法但历史主数据缺失时不阻断注册，统计进入全局国家维度，和查询投影的兜底语义一致。
|
||
return 0, nil
|
||
}
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return countryID, nil
|
||
}
|
||
|
||
func insertUserRegionChangedOutbox(ctx context.Context, tx *sql.Tx, command userdomain.CountryChangeCommand, oldCountry string, oldRegionID int64) error {
|
||
source := strings.TrimSpace(command.Source)
|
||
if source == "" {
|
||
source = "unknown"
|
||
}
|
||
payload := map[string]any{
|
||
"user_id": command.UserID,
|
||
"old_country": oldCountry,
|
||
"new_country": command.NewCountry,
|
||
"old_region_id": oldRegionID,
|
||
"new_region_id": command.NewRegionID,
|
||
"source": source,
|
||
"request_id": strings.TrimSpace(command.RequestID),
|
||
"changed_at_ms": command.ChangedAtMs,
|
||
}
|
||
payloadJSON, err := json.Marshal(payload)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
eventID := fmt.Sprintf("UserRegionChanged:%s:%d:%d:%d:%d", source, command.UserID, oldRegionID, command.NewRegionID, command.ChangedAtMs)
|
||
_, err = tx.ExecContext(ctx, `
|
||
INSERT IGNORE INTO user_outbox (app_code, event_id, event_type, aggregate_type, aggregate_id, status, payload_json, created_at_ms, updated_at_ms)
|
||
VALUES (?, ?, 'UserRegionChanged', 'user', ?, 'pending', ?, ?, ?)
|
||
`, appcode.Normalize(command.AppCode), eventID, command.UserID, string(payloadJSON), command.ChangedAtMs, command.ChangedAtMs)
|
||
return err
|
||
}
|
||
|
||
// UpdateUserProfile 修改用户基础资料并返回更新后的 users 快照。
|
||
|
||
func (r *Repository) UpdateUserProfile(ctx context.Context, command userdomain.ProfileUpdateCommand) (userdomain.User, error) {
|
||
// 资料更新锁定 users 行，避免和国家修改、短号懒过期覆盖 updated_at_ms。
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
defer tx.Rollback()
|
||
|
||
user, err := QueryUser(ctx, tx, "WHERE user_id = ? FOR UPDATE", command.UserID)
|
||
if err == sql.ErrNoRows {
|
||
return userdomain.User{}, xerr.New(xerr.NotFound, "user not found")
|
||
}
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
if command.Username != nil {
|
||
user.Username = *command.Username
|
||
}
|
||
if command.Avatar != nil {
|
||
user.Avatar = *command.Avatar
|
||
}
|
||
if command.Gender != nil {
|
||
user.Gender = *command.Gender
|
||
}
|
||
if command.BirthDate != nil {
|
||
user.BirthDate = *command.BirthDate
|
||
}
|
||
user.UpdatedAtMs = command.UpdatedAtMs
|
||
|
||
_, err = tx.ExecContext(ctx, `
|
||
UPDATE users
|
||
SET username = ?, avatar = ?, gender = ?, birth_date = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND user_id = ?
|
||
`, shared.NullableString(user.Username), shared.NullableString(user.Avatar), shared.NullableString(user.Gender), shared.NullableString(user.BirthDate), user.UpdatedAtMs, appcode.Normalize(command.AppCode), command.UserID)
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
|
||
return user, nil
|
||
}
|
||
|
||
// UpdateUserProfileBackground 只修改个人信息页背景图，避免通用资料 PATCH 误承担上传/背景语义。
|
||
func (r *Repository) UpdateUserProfileBackground(ctx context.Context, command userdomain.ProfileBackgroundUpdateCommand) (userdomain.User, error) {
|
||
// 背景图更新锁定 users 行，确保返回给 gateway 的用户资料就是本次写入后的快照。
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
defer tx.Rollback()
|
||
|
||
user, err := QueryUser(ctx, tx, "WHERE user_id = ? FOR UPDATE", command.UserID)
|
||
if err == sql.ErrNoRows {
|
||
return userdomain.User{}, xerr.New(xerr.NotFound, "user not found")
|
||
}
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
|
||
user.ProfileBgImg = command.ProfileBgImg
|
||
user.UpdatedAtMs = command.UpdatedAtMs
|
||
_, err = tx.ExecContext(ctx, `
|
||
UPDATE users
|
||
SET profile_bg_img = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND user_id = ?
|
||
`, shared.NullableString(user.ProfileBgImg), user.UpdatedAtMs, appcode.Normalize(command.AppCode), command.UserID)
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
|
||
return user, nil
|
||
}
|
||
|
||
// UpdateUserContactInfo 只修改用户级联系方式，避免 Host/BD/Agency/币商各自身份表保存多份互相漂移的数据。
|
||
func (r *Repository) UpdateUserContactInfo(ctx context.Context, command userdomain.ProfileContactUpdateCommand) (userdomain.User, error) {
|
||
// 联系方式展示在多个经营身份入口，写库时锁定 users 行，确保返回快照和后续列表读取都来自同一主记录。
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
defer tx.Rollback()
|
||
|
||
user, err := QueryUser(ctx, tx, "WHERE user_id = ? FOR UPDATE", command.UserID)
|
||
if err == sql.ErrNoRows {
|
||
return userdomain.User{}, xerr.New(xerr.NotFound, "user not found")
|
||
}
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
|
||
user.ContactInfo = strings.TrimSpace(command.ContactInfo)
|
||
user.UpdatedAtMs = command.UpdatedAtMs
|
||
_, err = tx.ExecContext(ctx, `
|
||
UPDATE users
|
||
SET contact_info = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND user_id = ?
|
||
`, user.ContactInfo, user.UpdatedAtMs, appcode.Normalize(command.AppCode), command.UserID)
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
|
||
return user, nil
|
||
}
|
||
|
||
// UpdateUserWithdrawAddress 只修改用户默认提现地址；提现订单自身的金额、状态和审核仍由 wallet-service 持有。
|
||
func (r *Repository) UpdateUserWithdrawAddress(ctx context.Context, command userdomain.WithdrawAddressUpdateCommand) (userdomain.User, error) {
|
||
// 提现地址会被 H5 反复读取预填，写库时锁定 users 行，确保返回值就是本次保存后的用户主资料快照。
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
defer tx.Rollback()
|
||
|
||
user, err := QueryUser(ctx, tx, "WHERE user_id = ? FOR UPDATE", command.UserID)
|
||
if err == sql.ErrNoRows {
|
||
return userdomain.User{}, xerr.New(xerr.NotFound, "user not found")
|
||
}
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
|
||
user.WithdrawUSDTTRC20Address = strings.TrimSpace(command.USDTTRC20Address)
|
||
user.UpdatedAtMs = command.UpdatedAtMs
|
||
_, err = tx.ExecContext(ctx, `
|
||
UPDATE users
|
||
SET withdraw_usdt_trc20_address = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND user_id = ?
|
||
`, user.WithdrawUSDTTRC20Address, user.UpdatedAtMs, appcode.Normalize(command.AppCode), command.UserID)
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
|
||
return user, nil
|
||
}
|
||
|
||
// CompleteOnboarding 是注册页资料固化的事务入口。
|
||
// 该方法不写 user_country_change_logs，确保首次国家选择不消耗改国家冷却窗口。
|
||
|
||
func (r *Repository) CompleteOnboarding(ctx context.Context, command userdomain.CompleteOnboardingCommand) (userdomain.User, error) {
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
defer tx.Rollback()
|
||
|
||
user, err := QueryUser(ctx, tx, "WHERE user_id = ? FOR UPDATE", command.UserID)
|
||
if err == sql.ErrNoRows {
|
||
return userdomain.User{}, xerr.New(xerr.NotFound, "user not found")
|
||
}
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
if user.ProfileCompleted {
|
||
// 已完成用户不能用注册页接口重写资料；提交入口保持幂等只返回当前事实。
|
||
if err := tx.Commit(); err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
return r.GetUser(ctx, command.UserID)
|
||
}
|
||
var inviteBindingErr error
|
||
inviteSnapshot := command.InviteCode
|
||
if command.InviteCode != "" {
|
||
user.InviteBinding, inviteBindingErr = invitestorage.BindRelation(ctx, tx, invitestorageBindCommand(command))
|
||
if inviteBindingErr != nil {
|
||
return userdomain.User{}, inviteBindingErr
|
||
}
|
||
if user.InviteBinding.InviteCode != "" {
|
||
inviteSnapshot = user.InviteBinding.InviteCode
|
||
}
|
||
}
|
||
|
||
_, err = tx.ExecContext(ctx, `
|
||
UPDATE users
|
||
SET username = ?,
|
||
avatar = ?,
|
||
gender = ?,
|
||
country = ?,
|
||
region_id = ?,
|
||
invite_code = CASE WHEN ? = '' THEN invite_code ELSE ? END,
|
||
profile_completed = ?,
|
||
profile_completed_at_ms = ?,
|
||
onboarding_status = ?,
|
||
updated_at_ms = ?
|
||
WHERE user_id = ?
|
||
AND app_code = ?
|
||
`, command.Username, command.Avatar, command.Gender, command.Country, shared.NullableRegionID(command.RegionID), inviteSnapshot, inviteSnapshot, true, command.CompletedAtMs, string(userdomain.OnboardingStatusCompleted), command.CompletedAtMs, command.UserID, appcode.Normalize(command.AppCode))
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
user.Username = command.Username
|
||
user.Avatar = command.Avatar
|
||
user.Gender = command.Gender
|
||
user.Country = command.Country
|
||
user.RegionID = command.RegionID
|
||
if inviteSnapshot != "" {
|
||
user.InviteCode = inviteSnapshot
|
||
}
|
||
user.ProfileCompleted = true
|
||
user.ProfileCompletedAtMs = command.CompletedAtMs
|
||
user.OnboardingStatus = userdomain.OnboardingStatusCompleted
|
||
user.UpdatedAtMs = command.CompletedAtMs
|
||
if err := InsertUserRegisteredOutbox(ctx, tx, user); err != nil {
|
||
// UserRegistered 是新增用户、留存 cohort 和注册奖励的事实；必须和首次资料完成同事务提交。
|
||
return userdomain.User{}, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
|
||
updated, err := r.GetUser(ctx, command.UserID)
|
||
updated.InviteBinding = user.InviteBinding
|
||
return updated, err
|
||
}
|
||
|
||
// ChangeUserCountry 修改国家并写日志；冷却期、outbox 和登录 session 撤销必须在同一事务内完成。
|
||
|
||
func (r *Repository) ChangeUserCountry(ctx context.Context, command userdomain.CountryChangeCommand) (userdomain.User, int64, []string, error) {
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return userdomain.User{}, 0, nil, err
|
||
}
|
||
defer tx.Rollback()
|
||
|
||
user, err := QueryUser(ctx, tx, "WHERE user_id = ? FOR UPDATE", command.UserID)
|
||
if err == sql.ErrNoRows {
|
||
return userdomain.User{}, 0, nil, xerr.New(xerr.NotFound, "user not found")
|
||
}
|
||
if err != nil {
|
||
return userdomain.User{}, 0, nil, err
|
||
}
|
||
lastChangedAt, err := latestCountryChangedAt(ctx, tx, command.UserID)
|
||
if err != nil {
|
||
return userdomain.User{}, 0, nil, err
|
||
}
|
||
nextAllowedAt := int64(0)
|
||
if lastChangedAt > 0 && command.CooldownMs > 0 {
|
||
nextAllowedAt = lastChangedAt + command.CooldownMs
|
||
}
|
||
revokedSessionIDs := make([]string, 0)
|
||
if user.Country == command.NewCountry {
|
||
// 修改为相同国家不写日志，也不消耗冷却期；映射变化时允许修正 stale region_id。
|
||
oldRegionID := user.RegionID
|
||
if user.RegionID != command.NewRegionID {
|
||
user.RegionID = command.NewRegionID
|
||
user.UpdatedAtMs = command.ChangedAtMs
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE users
|
||
SET region_id = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND user_id = ?
|
||
`, shared.NullableRegionID(user.RegionID), user.UpdatedAtMs, appcode.Normalize(command.AppCode), command.UserID); err != nil {
|
||
return userdomain.User{}, 0, nil, err
|
||
}
|
||
if err := insertUserRegionChangedOutbox(ctx, tx, command, user.Country, oldRegionID); err != nil {
|
||
return userdomain.User{}, 0, nil, err
|
||
}
|
||
revokedSessionIDs, err = revokeActiveUserSessionsForCountryChange(ctx, tx, command, appcode.Normalize(command.AppCode))
|
||
if err != nil {
|
||
return userdomain.User{}, 0, nil, err
|
||
}
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return userdomain.User{}, 0, nil, err
|
||
}
|
||
updated, err := r.GetUser(ctx, command.UserID)
|
||
return updated, nextAllowedAt, revokedSessionIDs, err
|
||
}
|
||
if nextAllowedAt > 0 && command.ChangedAtMs < nextAllowedAt {
|
||
return userdomain.User{}, nextAllowedAt, nil, xerr.New(xerr.CountryChangeCooldown, "country change is cooling down")
|
||
}
|
||
|
||
oldCountry := user.Country
|
||
oldRegionID := user.RegionID
|
||
user.Country = command.NewCountry
|
||
user.RegionID = command.NewRegionID
|
||
user.UpdatedAtMs = command.ChangedAtMs
|
||
_, err = tx.ExecContext(ctx, `
|
||
UPDATE users
|
||
SET country = ?, region_id = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND user_id = ?
|
||
`, shared.NullableString(user.Country), shared.NullableRegionID(user.RegionID), user.UpdatedAtMs, appcode.Normalize(command.AppCode), command.UserID)
|
||
if err != nil {
|
||
return userdomain.User{}, 0, nil, err
|
||
}
|
||
_, err = tx.ExecContext(ctx, `
|
||
INSERT INTO user_country_change_logs (app_code, user_id, old_country, new_country, request_id, changed_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
`, appcode.Normalize(command.AppCode), command.UserID, shared.NullableString(oldCountry), command.NewCountry, command.RequestID, command.ChangedAtMs)
|
||
if err != nil {
|
||
return userdomain.User{}, 0, nil, err
|
||
}
|
||
if err := insertUserRegionChangedOutbox(ctx, tx, command, oldCountry, oldRegionID); err != nil {
|
||
return userdomain.User{}, 0, nil, err
|
||
}
|
||
revokedSessionIDs, err = revokeActiveUserSessionsForCountryChange(ctx, tx, command, appcode.Normalize(command.AppCode))
|
||
if err != nil {
|
||
return userdomain.User{}, 0, nil, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return userdomain.User{}, 0, nil, err
|
||
}
|
||
|
||
updated, err := r.GetUser(ctx, command.UserID)
|
||
nextAllowedAt = int64(0)
|
||
if command.CooldownMs > 0 {
|
||
nextAllowedAt = command.ChangedAtMs + command.CooldownMs
|
||
}
|
||
return updated, nextAllowedAt, revokedSessionIDs, err
|
||
}
|
||
|
||
func revokeActiveUserSessionsForCountryChange(ctx context.Context, tx *sql.Tx, command userdomain.CountryChangeCommand, appCode string) ([]string, error) {
|
||
sessionIDs := make([]string, 0)
|
||
rows, err := tx.QueryContext(ctx, `
|
||
SELECT session_id
|
||
FROM auth_sessions
|
||
WHERE app_code = ? AND user_id = ? AND expires_at_ms > ? AND revoked_at_ms IS NULL
|
||
FOR UPDATE
|
||
`, appCode, command.UserID, command.ChangedAtMs)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
for rows.Next() {
|
||
var sessionID string
|
||
if err := rows.Scan(&sessionID); err != nil {
|
||
_ = rows.Close()
|
||
return nil, err
|
||
}
|
||
sessionIDs = append(sessionIDs, sessionID)
|
||
}
|
||
if err := rows.Close(); err != nil {
|
||
return nil, err
|
||
}
|
||
if len(sessionIDs) == 0 {
|
||
return sessionIDs, nil
|
||
}
|
||
_, err = tx.ExecContext(ctx, `
|
||
UPDATE auth_sessions
|
||
SET revoked_at_ms = ?, revoked_reason = ?, revoked_request_id = ?, revoked_by = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND user_id = ? AND expires_at_ms > ? AND revoked_at_ms IS NULL
|
||
`, command.ChangedAtMs, countryChangeSessionRevokeReason, strings.TrimSpace(command.RequestID), "country_change", command.ChangedAtMs, appCode, command.UserID, command.ChangedAtMs)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return sessionIDs, nil
|
||
}
|
||
|
||
func (r *Repository) hydrateInviteOverview(ctx context.Context, user *userdomain.User) error {
|
||
if user == nil || user.UserID <= 0 {
|
||
return nil
|
||
}
|
||
overview, err := invitestorage.New(r.db).GetOverview(ctx, user.AppCode, user.UserID, user.RegionID, timeNowMs())
|
||
if err != nil {
|
||
return err
|
||
}
|
||
user.InviteOverview = overview
|
||
return nil
|
||
}
|
||
|
||
// GetInviteAttribution 按被邀请用户读取邀请归因；主用户仓储只做委托，邀请表结构仍由 invite 仓储维护。
|
||
func (r *Repository) GetInviteAttribution(ctx context.Context, invitedUserID int64) (invitedomain.Attribution, error) {
|
||
if r == nil || r.db == nil {
|
||
return invitedomain.Attribution{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
return invitestorage.New(r.db).GetAttribution(ctx, appcode.FromContext(ctx), invitedUserID)
|
||
}
|
||
|
||
func invitestorageBindCommand(command userdomain.CompleteOnboardingCommand) invitedomain.BindCommand {
|
||
return invitedomain.BindCommand{
|
||
AppCode: command.AppCode,
|
||
InvitedUserID: command.UserID,
|
||
InvitedRegionID: command.RegionID,
|
||
InviteCode: command.InviteCode,
|
||
Source: "complete_onboarding",
|
||
RequestID: command.RequestID,
|
||
BoundAtMs: command.CompletedAtMs,
|
||
}
|
||
}
|
||
|
||
func timeNowMs() int64 {
|
||
return time.Now().UnixMilli()
|
||
}
|
||
|
||
// QueryUser 读取 users 快照字段；既服务普通查询，也服务短号事务中的 FOR UPDATE 读。
|