From 0b5e6873535e71e1decc035fdccc31624f896372 Mon Sep 17 00:00:00 2001 From: zhx Date: Wed, 27 May 2026 14:05:19 +0800 Subject: [PATCH] Align daily active metric with active log --- internal/storage/active_reconcile.go | 78 +++++++++++++++++++++------- 1 file changed, 59 insertions(+), 19 deletions(-) diff --git a/internal/storage/active_reconcile.go b/internal/storage/active_reconcile.go index 47192c4..0f86b44 100644 --- a/internal/storage/active_reconcile.go +++ b/internal/storage/active_reconcile.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "log/slog" + "strings" "time" "go.mongodb.org/mongo-driver/v2/bson" @@ -16,9 +17,16 @@ import ( const activeLogCollection = "user_daily_active_log" type activeLogRow struct { - UserID int64 `bson:"userId"` - ActiveDay string `bson:"activeDate"` - SysOrigin string `bson:"sysOrigin"` + UserID int64 `bson:"userId"` + ActiveDay string `bson:"activeDate"` + SysOrigin string `bson:"sysOrigin"` + RegisterCountryCode string `bson:"registerCountryCode"` +} + +type activeLogEvent struct { + UserID int64 + SysOrigin string + Country dashboard.Country } type activeDayStats struct { @@ -107,7 +115,7 @@ func (r *MySQLRepository) RebuildActiveRetentionRange(ctx context.Context, db *m }) } -func (r *MySQLRepository) loadActiveLogs(ctx context.Context, db *mongo.Database, start, end time.Time) (map[string]map[int64]struct{}, error) { +func (r *MySQLRepository) loadActiveLogs(ctx context.Context, db *mongo.Database, start, end time.Time) (map[string]map[int64]activeLogEvent, error) { filter := bson.D{ {Key: "activeDate", Value: bson.D{{Key: "$gte", Value: start.Format("2006-01-02")}, {Key: "$lt", Value: end.Format("2006-01-02")}}}, {Key: "sysOrigin", Value: bson.D{{Key: "$in", Value: r.cfg.Dashboard.SysOrigins}}}, @@ -116,6 +124,7 @@ func (r *MySQLRepository) loadActiveLogs(ctx context.Context, db *mongo.Database {Key: "userId", Value: 1}, {Key: "activeDate", Value: 1}, {Key: "sysOrigin", Value: 1}, + {Key: "registerCountryCode", Value: 1}, }) if r.cfg.Active.BatchLimit > 0 { findOptions.SetLimit(int64(r.cfg.Active.BatchLimit)) @@ -126,7 +135,7 @@ func (r *MySQLRepository) loadActiveLogs(ctx context.Context, db *mongo.Database } defer cursor.Close(ctx) - events := make(map[string]map[int64]struct{}) + events := make(map[string]map[int64]activeLogEvent) for cursor.Next(ctx) { var row activeLogRow if err := cursor.Decode(&row); err != nil { @@ -137,15 +146,27 @@ func (r *MySQLRepository) loadActiveLogs(ctx context.Context, db *mongo.Database } users := events[row.ActiveDay] if users == nil { - users = make(map[int64]struct{}) + users = make(map[int64]activeLogEvent) events[row.ActiveDay] = users } - users[row.UserID] = struct{}{} + users[row.UserID] = activeLogEvent{ + UserID: row.UserID, + SysOrigin: strings.TrimSpace(row.SysOrigin), + Country: r.activeLogCountry(row.RegisterCountryCode), + } } return events, cursor.Err() } -func collectActiveUserIDs(events map[string]map[int64]struct{}) []int64 { +func (r *MySQLRepository) activeLogCountry(registerCountryCode string) dashboard.Country { + code := strings.TrimSpace(registerCountryCode) + if code == "" { + code = r.cfg.Dashboard.UnknownCountryCode + } + return dashboard.Country{Code: code, Name: code} +} + +func collectActiveUserIDs(events map[string]map[int64]activeLogEvent) []int64 { seen := make(map[int64]struct{}) for _, users := range events { for userID := range users { @@ -159,6 +180,14 @@ func collectActiveUserIDs(events map[string]map[int64]struct{}) []int64 { return out } +func activeUserSet(events map[int64]activeLogEvent) map[int64]struct{} { + out := make(map[int64]struct{}, len(events)) + for userID := range events { + out[userID] = struct{}{} + } + return out +} + func (r *MySQLRepository) loadUserCountriesByID(ctx context.Context, ids []int64) (map[int64]dashboard.UserCountry, error) { result := make(map[int64]dashboard.UserCountry) for start := 0; start < len(ids); start += bulkInsertChunkSize { @@ -198,32 +227,43 @@ WHERE user_id IN `+inPlaceholders(end-start), args...) } func (r *MySQLRepository) replaceActiveRetentionDay(ctx context.Context, tx *sql.Tx, day time.Time, - events map[string]map[int64]struct{}, users map[int64]dashboard.UserCountry) error { + events map[string]map[int64]activeLogEvent, users map[int64]dashboard.UserCountry) error { dayKey := day.Format("2006-01-02") statsByOrigin := make(map[string]map[string]*activeDayStats) - for userID := range events[dayKey] { - user, ok := users[userID] - if !ok || user.IsDeleted || !r.allowedOrigin(user.SysOrigin) { + for userID, event := range events[dayKey] { + sysOrigin := strings.TrimSpace(event.SysOrigin) + if !r.allowedOrigin(sysOrigin) { continue } - stats := statsByOrigin[user.SysOrigin] + country := event.Country + if country.Code == "" { + if user, ok := users[userID]; ok && user.Country.Code != "" { + country = user.Country + } else { + country = dashboard.Country{Code: r.cfg.Dashboard.UnknownCountryCode, Name: r.cfg.Dashboard.UnknownCountryCode} + } + } + if country.Name == "" { + country.Name = country.Code + } + stats := statsByOrigin[sysOrigin] if stats == nil { stats = make(map[string]*activeDayStats) - statsByOrigin[user.SysOrigin] = stats + statsByOrigin[sysOrigin] = stats } - stat := stats[user.Country.Code] + stat := stats[country.Code] if stat == nil { stat = &activeDayStats{active: make(map[string]map[int64]struct{})} - stats[user.Country.Code] = stat + stats[country.Code] = stat } - active := stat.active[user.Country.Name] + active := stat.active[country.Name] if active == nil { active = make(map[int64]struct{}) - stat.active[user.Country.Name] = active + stat.active[country.Name] = active } active[userID] = struct{}{} } - retentionByOrigin, err := r.retentionStatsForDay(ctx, day, events[dayKey], users) + retentionByOrigin, err := r.retentionStatsForDay(ctx, day, activeUserSet(events[dayKey]), users) if err != nil { return err }