增加统计

services/statistics-service/cmd/backfill-last7/main.go

@@ -38,10 +38,10 @@ type appDayAgg struct {
 }
 
 type userRegistrationRow struct {
-	userID        int64
+	userID         int64
-	registeredDay string
+	registeredDay  string
-	dim           countryRegion
+	dim            countryRegion
-	createdAtMS   int64
+	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.profile_completed = 1
-		  AND u.created_at_ms < ?
+		  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.profile_completed = 1
-		  AND u.created_at_ms < ?
+		  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
 		}
 	}

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"`
+		CountryID            int64 `json:"country_id"`
-		RegionID  int64 `json:"region_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")
+	seedCountry(t, repository, "SG")
-	region := mustResolveRegionByCountry(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)
+	assertUserRegisteredOutboxCount(t, repository, thirdToken.UserID, 0)
-	assertUserRegisteredOutboxDimensions(t, repository, thirdToken.UserID, country.CountryID, region.RegionID)
 
 	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()

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()
 }
 

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
 	}