From 93250e5a7b79ef4104e237dc21c647cbb1dabfa8 Mon Sep 17 00:00:00 2001 From: zhx Date: Wed, 24 Jun 2026 10:43:47 +0800 Subject: [PATCH] fix statistics registration active count --- .../cmd/backfill-last7/main.go | 28 +++++++++++++------ .../internal/storage/mysql/query_test.go | 25 +++++++++++++++-- .../internal/storage/mysql/repository.go | 13 +++++++-- 3 files changed, 52 insertions(+), 14 deletions(-) diff --git a/services/statistics-service/cmd/backfill-last7/main.go b/services/statistics-service/cmd/backfill-last7/main.go index 10ba91f3..cd9614f7 100644 --- a/services/statistics-service/cmd/backfill-last7/main.go +++ b/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 diff --git a/services/statistics-service/internal/storage/mysql/query_test.go b/services/statistics-service/internal/storage/mysql/query_test.go index 5a3d671b..d8774e0f 100644 --- a/services/statistics-service/internal/storage/mysql/query_test.go +++ b/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) { diff --git a/services/statistics-service/internal/storage/mysql/repository.go b/services/statistics-service/internal/storage/mysql/repository.go index fdf4e0f4..91fac3ef 100644 --- a/services/statistics-service/internal/storage/mysql/repository.go +++ b/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) }) }