admin/dashboard-cdc-worker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Align daily active metric with active log

Browse Source
This commit is contained in:
zhx 2026-05-27 14:05:19 +08:00
parent 0c0abcfe3c
commit 0b5e687353
1 changed files with 59 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

78
internal/storage/active_reconcile.go
View File

@ -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
}