fix statistics backfill orphan activity

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

@@ -66,6 +66,7 @@ type rebuildState struct {
 	activeByKey         map[string]activeRow
 	userIDs             map[int64]struct{}
 	userDims            map[int64]countryRegion
+	knownUsers          map[int64]struct{}
 }
 
 func main() {
@@ -109,6 +110,7 @@ func main() {
 		activeByKey:         map[string]activeRow{},
 		userIDs:             map[int64]struct{}{},
 		userDims:            map[int64]countryRegion{},
+		knownUsers:          map[int64]struct{}{},
 	}
 	// 第一阶段只收集窗口内会参与重算的 user_id，后续统一批量加载用户国家和区域，避免按充值/活跃记录逐条访问 user-service 数据库。
 	if err := collectRegistrationUsers(ctx, userDB, state, app, windowStartMS, windowEndMS); err != nil {
@@ -324,6 +326,7 @@ func loadUserDims(ctx context.Context, db *sql.DB, state *rebuildState, app stri
 				return err
 			}
 			state.userDims[userID] = dim
+			state.knownUsers[userID] = struct{}{}
 		}
 		if err := rows.Close(); err != nil {
 			return err
@@ -382,6 +385,10 @@ func rebuildActive(ctx context.Context, db *sql.DB, state *rebuildState, app, st
 		if err := rows.Scan(&day, &userID, &firstMS); err != nil {
 			return err
 		}
+		if _, ok := state.knownUsers[userID]; !ok {
+			// 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 {