fix statistics registration active count

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

@@ -363,12 +363,15 @@ func rebuildRegistrations(ctx context.Context, db *sql.DB, state *rebuildState,
 		state.registeredByUserDay[userDayKey(row.registeredDay, row.userID)] = struct{}{}
 		agg := state.ensureAgg(row.registeredDay, row.dim)
 		agg.newUsers++
+		// 产品口径要求完成注册即算当日活跃；补数时直接把注册事实写入活跃重建集合，
+		// 后续房间/游戏活跃若命中同一用户同一天，只会更新更早的 first_active_at_ms，不会重复计数。
+		state.addActive(row.registeredDay, row.userID, row.dim, row.registeredAtMS)
 	}
 	return rows.Err()
 }
 
 func rebuildActive(ctx context.Context, db *sql.DB, state *rebuildState, app, startDay, endDay string) error {
-	// 活跃的原始房间事件已经被实时消费进 stat_user_day_activity；离线回填只重刷该表的国家/区域快照，避免在大窗口内扫描房间流水。
+	// 房间/游戏活跃已经被实时消费进 stat_user_day_activity；离线回填在注册活跃基础上重刷这张表的国家/区域快照。
 	rows, err := db.QueryContext(ctx, `
 		SELECT DATE_FORMAT(stat_day, '%Y-%m-%d'), user_id, MIN(first_active_at_ms)
 		FROM stat_user_day_activity
@@ -389,13 +392,7 @@ func rebuildActive(ctx context.Context, db *sql.DB, state *rebuildState, app, st
 			// stat_user_day_activity 是历史活跃读模型，不是用户事实 owner；回填时必须丢弃 user-service 已不存在的孤儿活跃，避免脏事实继续汇总到 country_id=0。
 			continue
 		}
-		dim := state.userDims[userID]
-		key := fmt.Sprintf("%s:%d", day, userID)
-		if _, exists := state.activeByKey[key]; exists {
-			continue
-		}
-		state.activeByKey[key] = activeRow{day: day, userID: userID, dim: dim, firstMS: firstMS}
-		state.ensureAgg(day, dim).activeUsers++
+		state.addActive(day, userID, state.userDims[userID], firstMS)
 	}
 	return rows.Err()
 }
@@ -486,6 +483,21 @@ func (s *rebuildState) addPayer(day string, userID int64, dim countryRegion, fir
 	s.payersByKey[key] = payerRow{day: day, userID: userID, dim: dim, firstMS: firstMS}
 }
 
+func (s *rebuildState) addActive(day string, userID int64, dim countryRegion, firstMS int64) {
+	key := userDayKey(day, userID)
+	if existing, exists := s.activeByKey[key]; exists {
+		// 注册、进房和游戏都能证明同一用户同一天活跃；主键按用户日收敛，聚合只加一次，
+		// 但 first_active_at_ms 要保留最早事实，便于后续排查“为什么算活跃”时能回到第一条证据。
+		if firstMS > 0 && (existing.firstMS == 0 || firstMS < existing.firstMS) {
+			existing.firstMS = firstMS
+			s.activeByKey[key] = existing
+		}
+		return
+	}
+	s.activeByKey[key] = activeRow{day: day, userID: userID, dim: dim, firstMS: firstMS}
+	s.ensureAgg(day, dim).activeUsers++
+}
+
 func (s *rebuildState) userRegisteredOn(userID int64, day string) bool {
 	_, ok := s.registeredByUserDay[userDayKey(day, userID)]
 	return ok

services/statistics-service/internal/storage/mysql/query_test.go

@@ -95,17 +95,36 @@ func TestConsumeUserRegisteredCountsOnlyNewRegistration(t *testing.T) {
 		t.Fatalf("consume registration replay: %v", err)
 	}
 
-	var newUsers int64
+	var newUsers, activeUsers int64
 	if err := repository.db.QueryRowContext(ctx, `
-		SELECT new_users
+		SELECT new_users, active_users
 		FROM stat_app_day_country
 		WHERE app_code = 'lalu' AND stat_day = '2026-06-22' AND country_id = 15 AND region_id = 10
-	`).Scan(&newUsers); err != nil {
+	`).Scan(&newUsers, &activeUsers); err != nil {
 		t.Fatalf("query app day country: %v", err)
 	}
 	if newUsers != 1 {
 		t.Fatalf("registration replay must not duplicate new users, got %d", newUsers)
 	}
+	if activeUsers != 1 {
+		t.Fatalf("registration must count as one same-day active user, got %d", activeUsers)
+	}
+
+	var activeRows int64
+	if err := repository.db.QueryRowContext(ctx, `
+		SELECT COUNT(*)
+		FROM stat_user_day_activity
+		WHERE app_code = 'lalu'
+		  AND stat_day = '2026-06-22'
+		  AND country_id = 15
+		  AND region_id = 10
+		  AND user_id = 327300172394536960
+	`).Scan(&activeRows); err != nil {
+		t.Fatalf("query registration active row: %v", err)
+	}
+	if activeRows != 1 {
+		t.Fatalf("registration replay must keep one user-day active row, got %d", activeRows)
+	}
 }
 
 func TestQueryOverviewReturnsPreviousPeriodDeltaRates(t *testing.T) {

services/statistics-service/internal/storage/mysql/repository.go

@@ -514,12 +514,19 @@ func (r *Repository) ConsumeUserRegistered(ctx context.Context, event UserRegist
 		}
 		// MQ 重投和补偿任务可能带来新的 event_id，但同一用户注册 cohort 只能入库一次；
 		// new_users 必须按 INSERT IGNORE 的真实插入行数累加，避免回放把国家新增人数重复放大。
-		_, err = tx.ExecContext(ctx, `
+		if _, err = tx.ExecContext(ctx, `
 			UPDATE stat_app_day_country
 			SET new_users = new_users + ?, updated_at_ms = ?
 			WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
-		`, inserted, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID)
-		return err
+		`, inserted, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID); err != nil {
+			return err
+		}
+		if inserted == 0 {
+			return nil
+		}
+		// 产品口径要求“完成注册即产生当日活跃”：注册完成本身已经证明用户打开并完成 App 主流程；
+		// 这里复用活跃唯一表收敛，若同日已有进房或游戏事件，不会重复增加 active_users。
+		return applyActive(ctx, tx, event.AppCode, day, countryID, regionID, event.UserID, event.OccurredAtMS, nowMS)
 	})
 }