From d947ebbfa3c171149e7c00b911cebf73a3139d65 Mon Sep 17 00:00:00 2001 From: zhx Date: Wed, 24 Jun 2026 10:04:30 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E7=BB=9F=E8=AE=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cmd/backfill-last7/main.go | 29 +++++----- .../internal/service/auth/service_test.go | 55 ++++++++++++++++--- .../storage/mysql/auth/third_party.go | 5 -- .../internal/storage/mysql/user/repository.go | 34 +++++++++++- 4 files changed, 93 insertions(+), 30 deletions(-) diff --git a/services/statistics-service/cmd/backfill-last7/main.go b/services/statistics-service/cmd/backfill-last7/main.go index 851bdbe0..a4121b63 100644 --- a/services/statistics-service/cmd/backfill-last7/main.go +++ b/services/statistics-service/cmd/backfill-last7/main.go @@ -38,10 +38,10 @@ type appDayAgg struct { } type userRegistrationRow struct { - userID int64 - registeredDay string - dim countryRegion - createdAtMS int64 + userID int64 + registeredDay string + dim countryRegion + registeredAtMS int64 } type payerRow struct { @@ -229,8 +229,9 @@ func collectRegistrationUsers(ctx context.Context, db *sql.DB, state *rebuildSta SELECT u.user_id FROM users u WHERE u.app_code = ? - AND u.created_at_ms >= ? - AND u.created_at_ms < ? + AND u.profile_completed = 1 + AND COALESCE(NULLIF(u.profile_completed_at_ms, 0), u.created_at_ms) >= ? + AND COALESCE(NULLIF(u.profile_completed_at_ms, 0), u.created_at_ms) < ? AND LOWER(COALESCE(u.source, '')) NOT IN ('game_robot', 'quick_account')`, app, startMS, endMS) if err != nil { @@ -332,15 +333,17 @@ func loadUserDims(ctx context.Context, db *sql.DB, state *rebuildState, app stri } func rebuildRegistrations(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64, loc *time.Location) error { - // 注册国家以 users.country 映射出的真实 country_id 为准；region_id 只作为独立维度写入，不能再兜底成国家。 + // 注册国家以资料完成后的 users.country 映射为准；region_id 只作为独立维度写入，不能再兜底成国家。 // source=game_robot 是后台全站机器人池账号，source=quick_account 是后台快捷建号；两者都不是 App 自然注册，不能进入新增用户、注册 cohort 或留存口径。 rows, err := db.QueryContext(ctx, ` - SELECT u.user_id, COALESCE(c.country_id, 0), COALESCE(u.region_id, 0), u.created_at_ms + SELECT u.user_id, COALESCE(c.country_id, 0), COALESCE(u.region_id, 0), + COALESCE(NULLIF(u.profile_completed_at_ms, 0), u.created_at_ms) AS registered_at_ms FROM users u LEFT JOIN countries c ON c.app_code = u.app_code AND c.country_code = u.country WHERE u.app_code = ? - AND u.created_at_ms >= ? - AND u.created_at_ms < ? + AND u.profile_completed = 1 + AND COALESCE(NULLIF(u.profile_completed_at_ms, 0), u.created_at_ms) >= ? + AND COALESCE(NULLIF(u.profile_completed_at_ms, 0), u.created_at_ms) < ? AND LOWER(COALESCE(u.source, '')) NOT IN ('game_robot', 'quick_account')`, app, startMS, endMS) if err != nil { @@ -349,10 +352,10 @@ func rebuildRegistrations(ctx context.Context, db *sql.DB, state *rebuildState, defer rows.Close() for rows.Next() { var row userRegistrationRow - if err := rows.Scan(&row.userID, &row.dim.countryID, &row.dim.regionID, &row.createdAtMS); err != nil { + if err := rows.Scan(&row.userID, &row.dim.countryID, &row.dim.regionID, &row.registeredAtMS); err != nil { return err } - row.registeredDay = dayFromMS(row.createdAtMS, loc) + row.registeredDay = dayFromMS(row.registeredAtMS, loc) state.registrations = append(state.registrations, row) state.registeredByUserDay[userDayKey(row.registeredDay, row.userID)] = struct{}{} agg := state.ensureAgg(row.registeredDay, row.dim) @@ -619,7 +622,7 @@ func insertStageRows(ctx context.Context, tx *sql.Tx, state *rebuildState, app s for _, row := range state.registrations { if _, err := tx.ExecContext(ctx, ` INSERT INTO tmp_stat_backfill_registration (app_code, user_id, registered_day, country_id, region_id, registered_at_ms) - VALUES (?, ?, ?, ?, ?, ?)`, app, row.userID, row.registeredDay, row.dim.countryID, row.dim.regionID, row.createdAtMS); err != nil { + VALUES (?, ?, ?, ?, ?, ?)`, app, row.userID, row.registeredDay, row.dim.countryID, row.dim.regionID, row.registeredAtMS); err != nil { return err } } diff --git a/services/user-service/internal/service/auth/service_test.go b/services/user-service/internal/service/auth/service_test.go index aff44905..d63fa2fe 100644 --- a/services/user-service/internal/service/auth/service_test.go +++ b/services/user-service/internal/service/auth/service_test.go @@ -162,19 +162,21 @@ func assertUserRegisteredOutboxCount(t *testing.T, repository *mysqltest.Reposit } } -func assertUserRegisteredOutboxDimensions(t *testing.T, repository *mysqltest.Repository, userID int64, countryID int64, regionID int64) { +func assertUserRegisteredOutboxFact(t *testing.T, repository *mysqltest.Repository, userID int64, countryID int64, regionID int64, occurredAtMS int64) { t.Helper() var raw string + var createdAtMS int64 if err := repository.RawDB().QueryRowContext(context.Background(), ` - SELECT payload_json + SELECT payload_json, created_at_ms FROM user_outbox WHERE app_code = 'lalu' AND event_type = 'UserRegistered' AND aggregate_id = ? - `, userID).Scan(&raw); err != nil { + `, userID).Scan(&raw, &createdAtMS); err != nil { t.Fatalf("query UserRegistered payload failed: %v", err) } var payload struct { - CountryID int64 `json:"country_id"` - RegionID int64 `json:"region_id"` + CountryID int64 `json:"country_id"` + RegionID int64 `json:"region_id"` + ProfileCompletedAtMS int64 `json:"profile_completed_at_ms"` } if err := json.Unmarshal([]byte(raw), &payload); err != nil { t.Fatalf("decode UserRegistered payload failed: %v raw=%s", err, raw) @@ -182,6 +184,9 @@ func assertUserRegisteredOutboxDimensions(t *testing.T, repository *mysqltest.Re if payload.CountryID != countryID || payload.RegionID != regionID { t.Fatalf("UserRegistered dimensions = country:%d region:%d, want country:%d region:%d", payload.CountryID, payload.RegionID, countryID, regionID) } + if payload.ProfileCompletedAtMS != occurredAtMS || createdAtMS != occurredAtMS { + t.Fatalf("UserRegistered time = payload:%d outbox:%d, want %d", payload.ProfileCompletedAtMS, createdAtMS, occurredAtMS) + } } func thirdPartyRegistration(platform string) authdomain.ThirdPartyRegistration { @@ -240,8 +245,7 @@ func TestThirdPartyUserSetsPasswordThenLogsIn(t *testing.T) { // 覆盖三方注册、设置密码、密码登录、refresh 轮换和 logout 的完整认证生命周期。 ctx := context.Background() repository := mysqltest.NewRepository(t) - country := seedCountry(t, repository, "SG") - region := mustResolveRegionByCountry(t, repository, "SG") + seedCountry(t, repository, "SG") now := time.UnixMilli(1000) svc := newAuthService(repository, &now, []int64{900001}, []string{"100001"}) @@ -255,8 +259,7 @@ func TestThirdPartyUserSetsPasswordThenLogsIn(t *testing.T) { if thirdToken.ProfileCompleted || thirdToken.OnboardingStatus != string(userdomain.OnboardingStatusProfileRequired) { t.Fatalf("new third-party user must require onboarding: %+v", thirdToken) } - assertUserRegisteredOutboxCount(t, repository, thirdToken.UserID, 1) - assertUserRegisteredOutboxDimensions(t, repository, thirdToken.UserID, country.CountryID, region.RegionID) + assertUserRegisteredOutboxCount(t, repository, thirdToken.UserID, 0) if _, err := svc.LoginPassword(ctx, "163000", "secret-pass", "ios", authservice.Meta{}); !xerr.IsCode(err, xerr.AuthFailed) { // 未设置密码前不能通过短号密码登录，且错误必须统一 AUTH_FAILED。 @@ -306,6 +309,40 @@ func TestThirdPartyUserSetsPasswordThenLogsIn(t *testing.T) { } } +func TestCompleteOnboardingEmitsUserRegisteredOutboxWithSelectedCountry(t *testing.T) { + // UserRegistered 是“注册页资料完成”事实，不是三方登录占位账号创建事实；统计国家必须来自最终提交的国家。 + ctx := context.Background() + repository := mysqltest.NewRepository(t) + country := seedCountry(t, repository, "AE") + region := mustResolveRegionByCountry(t, repository, "AE") + loginAt := time.UnixMilli(1000) + authSvc := newAuthService(repository, &loginAt, []int64{900031}, []string{"100031"}) + + token, isNewUser, err := authSvc.LoginThirdParty(ctx, "wechat", "openid-onboarding-register", authdomain.ThirdPartyRegistration{ + DeviceID: "dev-ios", + Platform: "ios", + }, authservice.Meta{RequestID: "req-third-create"}) + if err != nil { + t.Fatalf("LoginThirdParty failed: %v", err) + } + if !isNewUser || token.ProfileCompleted { + t.Fatalf("third-party login should create profile-required user: token=%+v isNew=%v", token, isNewUser) + } + assertUserRegisteredOutboxCount(t, repository, token.UserID, 0) + + completedAtMS := int64(2500) + userSvc := newUserService(repository, userservice.WithClock(func() time.Time { return time.UnixMilli(completedAtMS).UTC() })) + user, err := userSvc.CompleteOnboarding(ctx, token.UserID, "Amina", "https://cdn.example/amina.png", "female", "ae", "", "req-complete-onboarding") + if err != nil { + t.Fatalf("CompleteOnboarding failed: %v", err) + } + if !user.ProfileCompleted || user.Country != "AE" || user.RegionID != region.RegionID || user.ProfileCompletedAtMs != completedAtMS { + t.Fatalf("completed user snapshot mismatch: %+v", user) + } + assertUserRegisteredOutboxCount(t, repository, token.UserID, 1) + assertUserRegisteredOutboxFact(t, repository, token.UserID, country.CountryID, region.RegionID, completedAtMS) +} + func TestQuickCreateAccountCreatesCompletedPasswordLogin(t *testing.T) { // 快捷建号必须一次性得到可登录短号、已完成资料快照和密码身份，服务游戏联调直接取 uid/token。 ctx := context.Background() diff --git a/services/user-service/internal/storage/mysql/auth/third_party.go b/services/user-service/internal/storage/mysql/auth/third_party.go index 222dbd6d..3fc2728e 100644 --- a/services/user-service/internal/storage/mysql/auth/third_party.go +++ b/services/user-service/internal/storage/mysql/auth/third_party.go @@ -84,11 +84,6 @@ func (r *Repository) CreateThirdPartyUser(ctx context.Context, user userdomain.U // session 插入失败会回滚用户和三方绑定。 return err } - if err := userstorage.InsertUserRegisteredOutbox(ctx, tx, user); err != nil { - // UserRegistered 是统计新增用户、留存 cohort 和注册后异步读模型的唯一事实；它必须跟首次三方注册同事务提交，避免用户已经创建但统计链路永远收不到事件。 - return err - } - return tx.Commit() } diff --git a/services/user-service/internal/storage/mysql/user/repository.go b/services/user-service/internal/storage/mysql/user/repository.go index 7d8f7dc7..452873f6 100644 --- a/services/user-service/internal/storage/mysql/user/repository.go +++ b/services/user-service/internal/storage/mysql/user/repository.go @@ -153,17 +153,28 @@ func (r *Repository) CreateUserWithIdentity(ctx context.Context, user userdomain return tx.Commit() } -// InsertUserRegisteredOutbox writes the durable UserRegistered fact in the same transaction that creates the user. +// 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, @@ -177,6 +188,7 @@ func InsertUserRegisteredOutbox(ctx context.Context, tx *sql.Tx, user userdomain "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 { @@ -185,7 +197,7 @@ func InsertUserRegisteredOutbox(ctx context.Context, tx *sql.Tx, user userdomain _, 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), user.CreatedAtMs, user.CreatedAtMs) + `, appcode.Normalize(user.AppCode), fmt.Sprintf("UserRegistered:%d", user.UserID), user.UserID, string(payloadJSON), registeredAtMS, registeredAtMS) return err } @@ -196,7 +208,7 @@ func userRegisteredCountryID(ctx context.Context, tx *sql.Tx, user userdomain.Us } country := strings.ToUpper(strings.TrimSpace(user.Country)) if country == "" { - // 无国家资料的极简注册仍允许进入全局维度，后续补资料不会回写注册当天国家快照。 + // 正常 onboarding 完成路径必须有国家；这里只给历史或测试 completed 用户保留全局兜底，避免事件写入失败。 return 0, nil } var countryID int64 @@ -377,6 +389,22 @@ func (r *Repository) CompleteOnboarding(ctx context.Context, command userdomain. 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 }