From bf06fe6a8fbe4c5566b5692280b2853ea451c0c7 Mon Sep 17 00:00:00 2001 From: zhx Date: Wed, 27 May 2026 16:35:25 +0800 Subject: [PATCH] Align retention metrics with source cohorts --- internal/storage/active_reconcile.go | 265 ++++++++++++++++++++------- 1 file changed, 202 insertions(+), 63 deletions(-) diff --git a/internal/storage/active_reconcile.go b/internal/storage/active_reconcile.go index 0f86b44..5e7e236 100644 --- a/internal/storage/active_reconcile.go +++ b/internal/storage/active_reconcile.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "log/slog" + "strconv" "strings" "time" @@ -49,6 +50,13 @@ type countryCount struct { count int64 } +type retentionCohort struct { + userID int64 + sysOrigin string + country dashboard.Country + registerDate string +} + func (r *MySQLRepository) StartActiveReconciler(ctx context.Context, db *mongo.Database, logger *slog.Logger) { if !r.cfg.Active.Enabled || r.cfg.Active.Interval <= 0 { return @@ -94,14 +102,10 @@ func (r *MySQLRepository) RebuildActiveRetentionRange(ctx context.Context, db *m if err != nil { return err } - users, err := r.loadUserCountriesByID(ctx, collectActiveUserIDs(events)) - if err != nil { - return err - } return r.WithDashboardLock(ctx, r.cfg.CDC.BackfillLockWait, func() error { return r.WithTx(ctx, func(tx *sql.Tx) error { for day := start; day.Before(end); day = day.AddDate(0, 0, 1) { - if err := r.replaceActiveRetentionDay(ctx, tx, day, events, users); err != nil { + if err := r.replaceActiveRetentionDay(ctx, tx, db, day, events); err != nil { return err } } @@ -126,9 +130,6 @@ func (r *MySQLRepository) loadActiveLogs(ctx context.Context, db *mongo.Database {Key: "sysOrigin", Value: 1}, {Key: "registerCountryCode", Value: 1}, }) - if r.cfg.Active.BatchLimit > 0 { - findOptions.SetLimit(int64(r.cfg.Active.BatchLimit)) - } cursor, err := db.Collection(activeLogCollection).Find(ctx, filter, findOptions) if err != nil { return nil, err @@ -188,6 +189,26 @@ func activeUserSet(events map[int64]activeLogEvent) map[int64]struct{} { return out } +func stringSetToSlice(values map[string]struct{}) []string { + out := make([]string, 0, len(values)) + for value := range values { + out = append(out, value) + } + return out +} + +func retentionTargetDate(registerDate string, retentionDays int) string { + parsed, err := time.Parse("2006-01-02", strings.TrimSpace(registerDate)) + if err != nil { + return "" + } + return parsed.AddDate(0, 0, retentionDays).Format("2006-01-02") +} + +func activeUserDateKey(userID int64, activeDate string) string { + return strconv.FormatInt(userID, 10) + "|" + activeDate +} + 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 { @@ -226,8 +247,8 @@ WHERE user_id IN `+inPlaceholders(end-start), args...) return result, nil } -func (r *MySQLRepository) replaceActiveRetentionDay(ctx context.Context, tx *sql.Tx, day time.Time, - events map[string]map[int64]activeLogEvent, users map[int64]dashboard.UserCountry) error { +func (r *MySQLRepository) replaceActiveRetentionDay(ctx context.Context, tx *sql.Tx, db *mongo.Database, + day time.Time, events map[string]map[int64]activeLogEvent) error { dayKey := day.Format("2006-01-02") statsByOrigin := make(map[string]map[string]*activeDayStats) for userID, event := range events[dayKey] { @@ -237,11 +258,7 @@ func (r *MySQLRepository) replaceActiveRetentionDay(ctx context.Context, tx *sql } 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} - } + country = dashboard.Country{Code: r.cfg.Dashboard.UnknownCountryCode, Name: r.cfg.Dashboard.UnknownCountryCode} } if country.Name == "" { country.Name = country.Code @@ -263,7 +280,7 @@ func (r *MySQLRepository) replaceActiveRetentionDay(ctx context.Context, tx *sql } active[userID] = struct{}{} } - retentionByOrigin, err := r.retentionStatsForDay(ctx, day, activeUserSet(events[dayKey]), users) + retentionByTimezone, err := r.retentionStatsForDay(ctx, db, day) if err != nil { return err } @@ -280,7 +297,7 @@ WHERE period_type = 'DAY' AND period_key = ? AND stat_timezone = ? AND sys_origi return err } if err := r.upsertActiveRetentionRows(ctx, tx, day, statTimezone, sysOrigin, - statsByOrigin[sysOrigin], retentionByOrigin[sysOrigin]); err != nil { + statsByOrigin[sysOrigin], retentionByTimezone[statTimezone][sysOrigin]); err != nil { return err } } @@ -288,65 +305,104 @@ WHERE period_type = 'DAY' AND period_key = ? AND stat_timezone = ? AND sys_origi return nil } -func (r *MySQLRepository) retentionStatsForDay(ctx context.Context, day time.Time, - activeUsers map[int64]struct{}, users map[int64]dashboard.UserCountry) (map[string]map[string]activeRetentionValues, error) { - out := make(map[string]map[string]activeRetentionValues) - for _, days := range []int{1, 7, 30} { - bases, err := r.loadRetentionBase(ctx, day.AddDate(0, 0, -days)) +func (r *MySQLRepository) retentionStatsForDay(ctx context.Context, db *mongo.Database, + day time.Time) (map[string]map[string]map[string]activeRetentionValues, error) { + out := make(map[string]map[string]map[string]activeRetentionValues) + for _, statTimezone := range r.cfg.Dashboard.StatTimezones { + location, err := time.LoadLocation(statTimezone) if err != nil { - return nil, err + location = time.UTC } - for origin, countries := range bases { - originOut := out[origin] - if originOut == nil { - originOut = make(map[string]activeRetentionValues) - out[origin] = originOut + statDay := day.In(location) + dayStart := time.Date(statDay.Year(), statDay.Month(), statDay.Day(), 0, 0, 0, 0, location) + today := time.Now().In(location).Format("2006-01-02") + for _, days := range []int{1, 7, 30} { + cohortStart := dayStart.AddDate(0, 0, -days) + cohortEnd := cohortStart.AddDate(0, 0, 1) + cohorts, err := r.loadRetentionCohorts(ctx, cohortStart, cohortEnd, statTimezone) + if err != nil { + return nil, err } - for country, item := range countries { + if len(cohorts) == 0 { + continue + } + userIDsByOrigin := make(map[string]map[int64]struct{}) + activeDatesByOrigin := make(map[string]map[string]struct{}) + targetByUser := make(map[int64]string, len(cohorts)) + for _, cohort := range cohorts { + targetDate := retentionTargetDate(cohort.registerDate, days) + if targetDate == "" || targetDate > today { + continue + } + targetByUser[cohort.userID] = targetDate + if userIDsByOrigin[cohort.sysOrigin] == nil { + userIDsByOrigin[cohort.sysOrigin] = make(map[int64]struct{}) + } + if activeDatesByOrigin[cohort.sysOrigin] == nil { + activeDatesByOrigin[cohort.sysOrigin] = make(map[string]struct{}) + } + userIDsByOrigin[cohort.sysOrigin][cohort.userID] = struct{}{} + activeDatesByOrigin[cohort.sysOrigin][targetDate] = struct{}{} + } + activeByOrigin := make(map[string]map[string]struct{}) + for sysOrigin, ids := range userIDsByOrigin { + activeDates, err := r.listActiveUserDateSet(ctx, db, int64SetToSlice(ids), stringSetToSlice(activeDatesByOrigin[sysOrigin]), sysOrigin) + if err != nil { + return nil, err + } + activeByOrigin[sysOrigin] = activeDates + } + for _, cohort := range cohorts { + targetDate := targetByUser[cohort.userID] + if targetDate == "" { + continue + } + timezoneOut := out[statTimezone] + if timezoneOut == nil { + timezoneOut = make(map[string]map[string]activeRetentionValues) + out[statTimezone] = timezoneOut + } + originOut := timezoneOut[cohort.sysOrigin] + if originOut == nil { + originOut = make(map[string]activeRetentionValues) + timezoneOut[cohort.sysOrigin] = originOut + } + country := cohort.country.Code + if country == "" { + country = r.cfg.Dashboard.UnknownCountryCode + } stats := originOut[country] - stats.countryName = item.name + if stats.countryName == "" { + stats.countryName = cohort.country.Name + if stats.countryName == "" { + stats.countryName = country + } + } + retained := false + if activeByOrigin[cohort.sysOrigin] != nil { + _, retained = activeByOrigin[cohort.sysOrigin][activeUserDateKey(cohort.userID, targetDate)] + } switch days { case 1: - stats.d1Base = item.count + stats.d1Base++ + if retained { + stats.d1User++ + } case 7: - stats.d7Base = item.count + stats.d7Base++ + if retained { + stats.d7User++ + } case 30: - stats.d30Base = item.count + stats.d30Base++ + if retained { + stats.d30User++ + } } originOut[country] = stats } } } - for userID := range activeUsers { - user, ok := users[userID] - if !ok || user.CreatedAt.IsZero() || user.IsDeleted || !r.allowedOrigin(user.SysOrigin) { - continue - } - for _, days := range []int{1, 7, 30} { - if user.CreatedAt.Format("2006-01-02") != day.AddDate(0, 0, -days).Format("2006-01-02") { - continue - } - originOut := out[user.SysOrigin] - if originOut == nil { - originOut = make(map[string]activeRetentionValues) - out[user.SysOrigin] = originOut - } - country := user.Country.Code - stats := originOut[country] - if stats.countryName == "" { - stats.countryName = user.Country.Name - } - switch days { - case 1: - stats.d1User++ - case 7: - stats.d7User++ - case 30: - stats.d30User++ - } - originOut[country] = stats - } - } return out, nil } @@ -382,6 +438,89 @@ GROUP BY sys_origin, country_code`, args...) return result, rows.Err() } +func (r *MySQLRepository) loadRetentionCohorts(ctx context.Context, start, end time.Time, + statTimezone string) ([]retentionCohort, error) { + args := []interface{}{timezoneOffset(statTimezone)} + args = append(args, stringArgs(r.cfg.Dashboard.SysOrigins)...) + args = append(args, start, end) + rows, err := r.db.QueryContext(ctx, ` +SELECT u.id, + u.origin_sys, + NULLIF(u.country_code, ''), + NULLIF(u.country_name, ''), + DATE_FORMAT(CONVERT_TZ(u.create_time, '+03:00', ?), '%Y-%m-%d') AS register_date +FROM `+quoteIdent(r.cfg.MySQL.Schema)+`.user_base_info u +WHERE (u.is_del = 0 OR u.is_del IS NULL) + AND u.origin_sys IN `+inStringPlaceholders(len(r.cfg.Dashboard.SysOrigins))+` + AND u.create_time >= ? + AND u.create_time < ?`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var result []retentionCohort + for rows.Next() { + var item retentionCohort + var countryCode, countryName sql.NullString + if err := rows.Scan(&item.userID, &item.sysOrigin, &countryCode, &countryName, &item.registerDate); err != nil { + return nil, err + } + item.sysOrigin = strings.TrimSpace(item.sysOrigin) + if !r.allowedOrigin(item.sysOrigin) || item.userID == 0 || item.registerDate == "" { + continue + } + if countryCode.Valid && strings.TrimSpace(countryCode.String) != "" { + item.country.Code = strings.TrimSpace(countryCode.String) + } else { + item.country.Code = r.cfg.Dashboard.UnknownCountryCode + } + if countryName.Valid && strings.TrimSpace(countryName.String) != "" { + item.country.Name = strings.TrimSpace(countryName.String) + } else { + item.country.Name = item.country.Code + } + result = append(result, item) + } + return result, rows.Err() +} + +func (r *MySQLRepository) listActiveUserDateSet(ctx context.Context, db *mongo.Database, + userIDs []int64, activeDates []string, sysOrigin string) (map[string]struct{}, error) { + result := make(map[string]struct{}) + if len(userIDs) == 0 || len(activeDates) == 0 { + return result, nil + } + for start := 0; start < len(userIDs); start += bulkInsertChunkSize { + end := min(start+bulkInsertChunkSize, len(userIDs)) + filter := bson.D{ + {Key: "userId", Value: bson.D{{Key: "$in", Value: userIDs[start:end]}}}, + {Key: "activeDate", Value: bson.D{{Key: "$in", Value: activeDates}}}, + {Key: "sysOrigin", Value: sysOrigin}, + } + cursor, err := db.Collection(activeLogCollection).Find(ctx, filter, options.Find().SetProjection(bson.D{ + {Key: "userId", Value: 1}, + {Key: "activeDate", Value: 1}, + })) + if err != nil { + return nil, err + } + for cursor.Next(ctx) { + var row activeLogRow + if err := cursor.Decode(&row); err != nil { + _ = cursor.Close(ctx) + return nil, err + } + if row.UserID != 0 && row.ActiveDay != "" { + result[activeUserDateKey(row.UserID, row.ActiveDay)] = struct{}{} + } + } + if err := cursor.Close(ctx); err != nil { + return nil, err + } + } + return result, nil +} + func (r *MySQLRepository) upsertActiveRetentionRows(ctx context.Context, tx *sql.Tx, day time.Time, statTimezone string, sysOrigin string, active map[string]*activeDayStats, retention map[string]activeRetentionValues) error { values := make(map[string]activeRetentionValues)