Align retention metrics with source cohorts
This commit is contained in:
zhx
parent
e6323e1635
commit
bf06fe6a8f
@ -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,12 +258,8 @@ 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}
|
||||
}
|
||||
}
|
||||
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)
|
||||
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 {
|
||||
location = time.UTC
|
||||
}
|
||||
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} {
|
||||
bases, err := r.loadRetentionBase(ctx, day.AddDate(0, 0, -days))
|
||||
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 origin, countries := range bases {
|
||||
originOut := out[origin]
|
||||
if originOut == nil {
|
||||
originOut = make(map[string]activeRetentionValues)
|
||||
out[origin] = originOut
|
||||
}
|
||||
for country, item := range countries {
|
||||
stats := originOut[country]
|
||||
stats.countryName = item.name
|
||||
switch days {
|
||||
case 1:
|
||||
stats.d1Base = item.count
|
||||
case 7:
|
||||
stats.d7Base = item.count
|
||||
case 30:
|
||||
stats.d30Base = item.count
|
||||
}
|
||||
originOut[country] = stats
|
||||
}
|
||||
}
|
||||
}
|
||||
for userID := range activeUsers {
|
||||
user, ok := users[userID]
|
||||
if !ok || user.CreatedAt.IsZero() || user.IsDeleted || !r.allowedOrigin(user.SysOrigin) {
|
||||
if len(cohorts) == 0 {
|
||||
continue
|
||||
}
|
||||
for _, days := range []int{1, 7, 30} {
|
||||
if user.CreatedAt.Format("2006-01-02") != day.AddDate(0, 0, -days).Format("2006-01-02") {
|
||||
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
|
||||
}
|
||||
originOut := out[user.SysOrigin]
|
||||
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)
|
||||
out[user.SysOrigin] = originOut
|
||||
timezoneOut[cohort.sysOrigin] = originOut
|
||||
}
|
||||
country := cohort.country.Code
|
||||
if country == "" {
|
||||
country = r.cfg.Dashboard.UnknownCountryCode
|
||||
}
|
||||
country := user.Country.Code
|
||||
stats := originOut[country]
|
||||
if stats.countryName == "" {
|
||||
stats.countryName = user.Country.Name
|
||||
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++
|
||||
if retained {
|
||||
stats.d1User++
|
||||
}
|
||||
case 7:
|
||||
stats.d7Base++
|
||||
if retained {
|
||||
stats.d7User++
|
||||
}
|
||||
case 30:
|
||||
stats.d30Base++
|
||||
if retained {
|
||||
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)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user