331 lines
16 KiB
Go
331 lines
16 KiB
Go
package auth
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
authdomain "hyapp/services/user-service/internal/domain/auth"
|
||
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"
|
||
userstorage "hyapp/services/user-service/internal/storage/mysql/user"
|
||
)
|
||
|
||
func (r *Repository) FindThirdPartyIdentity(ctx context.Context, provider string, providerSubject string) (authdomain.ThirdPartyIdentity, error) {
|
||
var identity authdomain.ThirdPartyIdentity
|
||
err := r.db.QueryRowContext(ctx, `
|
||
SELECT app_code, user_id, provider, provider_subject, COALESCE(provider_union_id, ''), created_at_ms, updated_at_ms
|
||
FROM third_party_identities
|
||
WHERE app_code = ? AND provider = ? AND provider_subject = ?
|
||
`, appcode.FromContext(ctx), provider, providerSubject).Scan(&identity.AppCode, &identity.UserID, &identity.Provider, &identity.ProviderSubject, &identity.ProviderUnionID, &identity.CreatedAtMs, &identity.UpdatedAtMs)
|
||
if err == sql.ErrNoRows {
|
||
// 未绑定三方身份时由 service 转入注册路径。
|
||
return authdomain.ThirdPartyIdentity{}, xerr.New(xerr.NotFound, "third-party identity not found")
|
||
}
|
||
if err != nil {
|
||
return authdomain.ThirdPartyIdentity{}, err
|
||
}
|
||
|
||
return identity, nil
|
||
}
|
||
|
||
func (r *Repository) FindUserIDByRegisterDeviceID(ctx context.Context, appCode string, deviceID string) (int64, error) {
|
||
var userID int64
|
||
err := r.db.QueryRowContext(ctx, `
|
||
SELECT user_id
|
||
FROM users
|
||
WHERE app_code = ? AND register_device_id = ?
|
||
ORDER BY created_at_ms ASC, user_id ASC
|
||
LIMIT 1
|
||
`, appcode.Normalize(appCode), deviceID).Scan(&userID)
|
||
if err == sql.ErrNoRows {
|
||
// 未查到占用者时，service 才允许新三方身份走注册事务。
|
||
return 0, xerr.New(xerr.NotFound, "registered device not found")
|
||
}
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
|
||
return userID, nil
|
||
}
|
||
|
||
func (r *Repository) ListDisplayUserIDsByRegisterDeviceID(ctx context.Context, appCode string, deviceID string) ([]string, error) {
|
||
rows, err := r.db.QueryContext(ctx, `
|
||
SELECT current_display_user_id
|
||
FROM users
|
||
WHERE app_code = ? AND register_device_id = ?
|
||
ORDER BY created_at_ms ASC, user_id ASC
|
||
`, appcode.Normalize(appCode), deviceID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
displayUserIDs := make([]string, 0)
|
||
for rows.Next() {
|
||
var displayUserID string
|
||
if err := rows.Scan(&displayUserID); err != nil {
|
||
return nil, err
|
||
}
|
||
// 这里不去重，账号数量以 users 行为准；文案 helper 会只清洗展示列表。
|
||
displayUserIDs = append(displayUserIDs, displayUserID)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return displayUserIDs, nil
|
||
}
|
||
|
||
func (r *Repository) CountUsersByRegisterDeviceID(ctx context.Context, appCode string, deviceID string) (int64, error) {
|
||
var count int64
|
||
if err := r.db.QueryRowContext(ctx, `
|
||
SELECT COUNT(*)
|
||
FROM users
|
||
WHERE app_code = ? AND register_device_id = ?
|
||
`, appcode.Normalize(appCode), deviceID).Scan(&count); err != nil {
|
||
return 0, err
|
||
}
|
||
|
||
return count, nil
|
||
}
|
||
|
||
func (r *Repository) UpsertPendingThirdPartyRegistration(ctx context.Context, pending authdomain.PendingThirdPartyRegistration) (authdomain.PendingThirdPartyRegistration, error) {
|
||
// 同一个 provider subject 重复登录只刷新待注册来源快照，不新建 users，也不制造多条 pending 垃圾数据。
|
||
_, err := r.db.ExecContext(ctx, `
|
||
INSERT INTO pending_third_party_registrations (
|
||
app_code, pending_id, provider, provider_subject, provider_union_id,
|
||
username, gender, country, invite_code, register_ip, register_user_agent, country_by_ip,
|
||
device_id, device, os_version, avatar, birth_date, app_version, build_number, source,
|
||
install_channel, campaign, platform, language, timezone, status, completed_user_id,
|
||
expires_at_ms, completed_at_ms, created_at_ms, updated_at_ms
|
||
)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, 0, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
provider_union_id = IF(status = 'pending', VALUES(provider_union_id), provider_union_id),
|
||
username = IF(status = 'pending', VALUES(username), username),
|
||
gender = IF(status = 'pending', VALUES(gender), gender),
|
||
country = IF(status = 'pending', VALUES(country), country),
|
||
invite_code = IF(status = 'pending', VALUES(invite_code), invite_code),
|
||
register_ip = IF(status = 'pending', VALUES(register_ip), register_ip),
|
||
register_user_agent = IF(status = 'pending', VALUES(register_user_agent), register_user_agent),
|
||
country_by_ip = IF(status = 'pending', VALUES(country_by_ip), country_by_ip),
|
||
device_id = IF(status = 'pending', VALUES(device_id), device_id),
|
||
device = IF(status = 'pending', VALUES(device), device),
|
||
os_version = IF(status = 'pending', VALUES(os_version), os_version),
|
||
avatar = IF(status = 'pending', VALUES(avatar), avatar),
|
||
birth_date = IF(status = 'pending', VALUES(birth_date), birth_date),
|
||
app_version = IF(status = 'pending', VALUES(app_version), app_version),
|
||
build_number = IF(status = 'pending', VALUES(build_number), build_number),
|
||
source = IF(status = 'pending', VALUES(source), source),
|
||
install_channel = IF(status = 'pending', VALUES(install_channel), install_channel),
|
||
campaign = IF(status = 'pending', VALUES(campaign), campaign),
|
||
platform = IF(status = 'pending', VALUES(platform), platform),
|
||
language = IF(status = 'pending', VALUES(language), language),
|
||
timezone = IF(status = 'pending', VALUES(timezone), timezone),
|
||
expires_at_ms = IF(status = 'pending', VALUES(expires_at_ms), expires_at_ms),
|
||
updated_at_ms = IF(status = 'pending', VALUES(updated_at_ms), updated_at_ms)
|
||
`, appcode.Normalize(pending.AppCode), pending.PendingID, pending.Provider, pending.ProviderSubject, shared.NullableString(pending.ProviderUnionID),
|
||
shared.NullableString(pending.Registration.Username), shared.NullableString(pending.Registration.Gender), shared.NullableString(pending.Registration.Country), shared.NullableString(pending.Registration.InviteCode), shared.NullableString(pending.Registration.IP), shared.NullableString(pending.Registration.UserAgent), shared.NullableString(pending.Registration.CountryByIP),
|
||
shared.NullableString(pending.Registration.DeviceID), shared.NullableString(pending.Registration.Device), shared.NullableString(pending.Registration.OSVersion), shared.NullableString(pending.Registration.Avatar), shared.NullableString(pending.Registration.BirthDate), shared.NullableString(pending.Registration.AppVersion), shared.NullableString(pending.Registration.BuildNumber), shared.NullableString(pending.Registration.Source),
|
||
shared.NullableString(pending.Registration.InstallChannel), shared.NullableString(pending.Registration.Campaign), shared.NullableString(pending.Registration.Platform), shared.NullableString(pending.Registration.Language), shared.NullableString(pending.Registration.Timezone), authdomain.PendingThirdPartyRegistrationStatusPending,
|
||
pending.ExpiresAtMs, pending.CreatedAtMs, pending.UpdatedAtMs)
|
||
if err != nil {
|
||
return authdomain.PendingThirdPartyRegistration{}, mapAuthDuplicateError(err)
|
||
}
|
||
|
||
return r.findPendingThirdPartyRegistrationByProvider(ctx, pending.AppCode, pending.Provider, pending.ProviderSubject)
|
||
}
|
||
|
||
func (r *Repository) FindPendingThirdPartyRegistration(ctx context.Context, appCode string, pendingID string) (authdomain.PendingThirdPartyRegistration, error) {
|
||
return r.scanPendingThirdPartyRegistration(ctx, r.db.QueryRowContext(ctx, pendingThirdPartyRegistrationSelectSQL()+`
|
||
WHERE app_code = ? AND pending_id = ?
|
||
`, appcode.Normalize(appCode), pendingID))
|
||
}
|
||
|
||
func (r *Repository) findPendingThirdPartyRegistrationByProvider(ctx context.Context, appCode string, provider string, providerSubject string) (authdomain.PendingThirdPartyRegistration, error) {
|
||
return r.scanPendingThirdPartyRegistration(ctx, r.db.QueryRowContext(ctx, pendingThirdPartyRegistrationSelectSQL()+`
|
||
WHERE app_code = ? AND provider = ? AND provider_subject = ?
|
||
`, appcode.Normalize(appCode), provider, providerSubject))
|
||
}
|
||
|
||
// CreateThirdPartyUser 原子创建用户、默认短号、三方身份和首个 session。
|
||
|
||
func (r *Repository) CreateThirdPartyUser(ctx context.Context, user userdomain.User, identity userdomain.Identity, thirdParty authdomain.ThirdPartyIdentity, session authdomain.Session, inviteBind invitedomain.BindCommand, maxAccountsPerDevice int32) error {
|
||
// 三方首次登录注册必须同时写用户、默认短号、三方绑定和首个 session。
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer tx.Rollback()
|
||
|
||
if err := ensureRegisterDeviceQuotaTx(ctx, tx, user.AppCode, user.RegisterDeviceID, maxAccountsPerDevice); err != nil {
|
||
// 事务内再次校验设备上限，避免两个并发注册都在事务外读到旧计数后同时插入。
|
||
return err
|
||
}
|
||
if err := userstorage.InsertUserIdentity(ctx, tx, user, identity); err != nil {
|
||
// 默认短号冲突由 service 重试。
|
||
return mapAuthDuplicateError(err)
|
||
}
|
||
if _, err := invitestorage.EnsurePrimaryCodeForUser(ctx, tx, user.AppCode, user.UserID, user.CreatedAtMs); err != nil {
|
||
return err
|
||
}
|
||
if _, err := invitestorage.BindRelation(ctx, tx, inviteBind); err != nil {
|
||
return err
|
||
}
|
||
if err := userstorage.InsertUserRegisteredOutbox(ctx, tx, user); err != nil {
|
||
// 直接带完整资料注册时，UserRegistered 必须与用户/三方身份同事务提交，统计国家不能事后补猜。
|
||
return err
|
||
}
|
||
|
||
_, err = tx.ExecContext(ctx, `
|
||
INSERT INTO third_party_identities (app_code, user_id, provider, provider_subject, provider_union_id, created_at_ms, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||
`, appcode.Normalize(thirdParty.AppCode), thirdParty.UserID, thirdParty.Provider, thirdParty.ProviderSubject, shared.NullableString(thirdParty.ProviderUnionID), thirdParty.CreatedAtMs, thirdParty.UpdatedAtMs)
|
||
if err != nil {
|
||
return mapAuthDuplicateError(err)
|
||
}
|
||
|
||
if err := insertSession(ctx, tx, session); err != nil {
|
||
// session 插入失败会回滚用户和三方绑定。
|
||
return err
|
||
}
|
||
return tx.Commit()
|
||
}
|
||
|
||
func (r *Repository) CompletePendingThirdPartyUser(ctx context.Context, pendingID string, user userdomain.User, identity userdomain.Identity, thirdParty authdomain.ThirdPartyIdentity, session authdomain.Session, inviteBind invitedomain.BindCommand, maxAccountsPerDevice int32) (userdomain.User, error) {
|
||
// pending 消费、真实用户创建、默认短号、三方绑定、session、邀请码和注册事件必须同事务；
|
||
// 任何一步失败都不能留下半个真实账号。
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
defer tx.Rollback()
|
||
|
||
pending, err := r.scanPendingThirdPartyRegistration(ctx, tx.QueryRowContext(ctx, pendingThirdPartyRegistrationSelectSQL()+`
|
||
WHERE app_code = ? AND pending_id = ?
|
||
FOR UPDATE
|
||
`, appcode.Normalize(user.AppCode), pendingID))
|
||
if err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
if pending.Status != authdomain.PendingThirdPartyRegistrationStatusPending {
|
||
return userdomain.User{}, xerr.New(xerr.Conflict, "pending registration already completed")
|
||
}
|
||
if pending.ExpiresAtMs > 0 && pending.ExpiresAtMs <= user.CreatedAtMs {
|
||
return userdomain.User{}, xerr.New(xerr.SessionExpired, "pending registration expired")
|
||
}
|
||
if pending.Provider != thirdParty.Provider || pending.ProviderSubject != thirdParty.ProviderSubject {
|
||
return userdomain.User{}, xerr.New(xerr.Conflict, "pending registration identity mismatch")
|
||
}
|
||
if err := ensureRegisterDeviceQuotaTx(ctx, tx, user.AppCode, user.RegisterDeviceID, maxAccountsPerDevice); err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
if err := userstorage.InsertUserIdentity(ctx, tx, user, identity); err != nil {
|
||
return userdomain.User{}, mapAuthDuplicateError(err)
|
||
}
|
||
if _, err := invitestorage.EnsurePrimaryCodeForUser(ctx, tx, user.AppCode, user.UserID, user.CreatedAtMs); err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
if binding, err := invitestorage.BindRelation(ctx, tx, inviteBind); err != nil {
|
||
return userdomain.User{}, err
|
||
} else {
|
||
user.InviteBinding = binding
|
||
if binding.InviteCode != "" {
|
||
user.InviteCode = binding.InviteCode
|
||
}
|
||
}
|
||
if err := userstorage.InsertUserRegisteredOutbox(ctx, tx, user); err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO third_party_identities (app_code, user_id, provider, provider_subject, provider_union_id, created_at_ms, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||
`, appcode.Normalize(thirdParty.AppCode), thirdParty.UserID, thirdParty.Provider, thirdParty.ProviderSubject, shared.NullableString(thirdParty.ProviderUnionID), thirdParty.CreatedAtMs, thirdParty.UpdatedAtMs); err != nil {
|
||
return userdomain.User{}, mapAuthDuplicateError(err)
|
||
}
|
||
if err := insertSession(ctx, tx, session); err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE pending_third_party_registrations
|
||
SET status = ?, completed_user_id = ?, completed_at_ms = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND pending_id = ?
|
||
`, authdomain.PendingThirdPartyRegistrationStatusCompleted, user.UserID, user.CreatedAtMs, user.CreatedAtMs, appcode.Normalize(user.AppCode), pendingID); err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
|
||
if err := tx.Commit(); err != nil {
|
||
return userdomain.User{}, err
|
||
}
|
||
return user, nil
|
||
}
|
||
|
||
func pendingThirdPartyRegistrationSelectSQL() string {
|
||
return `
|
||
SELECT app_code, pending_id, provider, provider_subject, COALESCE(provider_union_id, ''),
|
||
COALESCE(username, ''), COALESCE(gender, ''), COALESCE(country, ''), COALESCE(invite_code, ''),
|
||
COALESCE(register_ip, ''), COALESCE(register_user_agent, ''), COALESCE(country_by_ip, ''),
|
||
COALESCE(device_id, ''), COALESCE(device, ''), COALESCE(os_version, ''), COALESCE(avatar, ''),
|
||
COALESCE(birth_date, ''), COALESCE(app_version, ''), COALESCE(build_number, ''), COALESCE(source, ''),
|
||
COALESCE(install_channel, ''), COALESCE(campaign, ''), COALESCE(platform, ''), COALESCE(language, ''),
|
||
COALESCE(timezone, ''), status, completed_user_id, expires_at_ms, completed_at_ms, created_at_ms, updated_at_ms
|
||
FROM pending_third_party_registrations
|
||
`
|
||
}
|
||
|
||
type pendingThirdPartyScanner interface {
|
||
Scan(dest ...any) error
|
||
}
|
||
|
||
func (r *Repository) scanPendingThirdPartyRegistration(_ context.Context, row pendingThirdPartyScanner) (authdomain.PendingThirdPartyRegistration, error) {
|
||
var pending authdomain.PendingThirdPartyRegistration
|
||
err := row.Scan(
|
||
&pending.AppCode,
|
||
&pending.PendingID,
|
||
&pending.Provider,
|
||
&pending.ProviderSubject,
|
||
&pending.ProviderUnionID,
|
||
&pending.Registration.Username,
|
||
&pending.Registration.Gender,
|
||
&pending.Registration.Country,
|
||
&pending.Registration.InviteCode,
|
||
&pending.Registration.IP,
|
||
&pending.Registration.UserAgent,
|
||
&pending.Registration.CountryByIP,
|
||
&pending.Registration.DeviceID,
|
||
&pending.Registration.Device,
|
||
&pending.Registration.OSVersion,
|
||
&pending.Registration.Avatar,
|
||
&pending.Registration.BirthDate,
|
||
&pending.Registration.AppVersion,
|
||
&pending.Registration.BuildNumber,
|
||
&pending.Registration.Source,
|
||
&pending.Registration.InstallChannel,
|
||
&pending.Registration.Campaign,
|
||
&pending.Registration.Platform,
|
||
&pending.Registration.Language,
|
||
&pending.Registration.Timezone,
|
||
&pending.Status,
|
||
&pending.CompletedUserID,
|
||
&pending.ExpiresAtMs,
|
||
&pending.CompletedAtMs,
|
||
&pending.CreatedAtMs,
|
||
&pending.UpdatedAtMs,
|
||
)
|
||
if err == sql.ErrNoRows {
|
||
return authdomain.PendingThirdPartyRegistration{}, xerr.New(xerr.NotFound, "pending registration not found")
|
||
}
|
||
if err != nil {
|
||
return authdomain.PendingThirdPartyRegistration{}, err
|
||
}
|
||
pending.AppCode = appcode.Normalize(pending.AppCode)
|
||
pending.Registration.AppCode = pending.AppCode
|
||
return pending, nil
|
||
}
|
||
|
||
// RecordLoginAudit 写登录审计；调用方不会让审计失败影响主链路。
|