diff --git a/cmd/dashboard-cdc-worker/main.go b/cmd/dashboard-cdc-worker/main.go index f3e617e..0ae7a58 100644 --- a/cmd/dashboard-cdc-worker/main.go +++ b/cmd/dashboard-cdc-worker/main.go @@ -82,6 +82,7 @@ func main() { } projector := dashboard.NewProjector(cfg, repo, redisClient, metrics) + repo.StartUserReconciler(ctx, logger, projector) repo.StartRechargeReconciler(ctx, logger, projector) repo.StartGiftReconciler(ctx, logger, projector) diff --git a/internal/cdc/handler.go b/internal/cdc/handler.go index 7c6e249..b2ef2d9 100644 --- a/internal/cdc/handler.go +++ b/internal/cdc/handler.go @@ -36,7 +36,8 @@ func newHandler(cfg config.Config, store dashboard.Store, projector *dashboard.P cfg: cfg, store: store, projector: projector, - mapper: dashboard.NewMapper(store, cfg.Dashboard.SysOrigins, cfg.Dashboard.UnknownCountryCode), + mapper: dashboard.NewMapper(store, cfg.Dashboard.SysOrigins, + cfg.Dashboard.UnknownCountryCode, cfg.User.CreateTimeDisplayOffset), logger: logger, batch: make([]dashboard.Event, 0, cfg.CDC.BatchSize), lastFlush: time.Now(), diff --git a/internal/config/config.go b/internal/config/config.go index b0d551a..13f0412 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -18,6 +18,7 @@ type Config struct { Dashboard DashboardConfig Recharge RechargeReconcileConfig Gift GiftReconcileConfig + User UserReconcileConfig Active ActiveReconcileConfig Health HealthConfig } @@ -86,6 +87,15 @@ type GiftReconcileConfig struct { BatchLimit int } +type UserReconcileConfig struct { + Enabled bool + Interval time.Duration + LookbackDays int + CurrentDayLag time.Duration + BatchLimit int + CreateTimeDisplayOffset time.Duration +} + type ActiveReconcileConfig struct { Enabled bool Interval time.Duration @@ -156,6 +166,14 @@ func Load() (Config, error) { CurrentDayLag: envDuration("DASHBOARD_GIFT_RECONCILE_CURRENT_DAY_LAG", 5*time.Minute), BatchLimit: envInt("DASHBOARD_GIFT_RECONCILE_BATCH_LIMIT", 50000), }, + User: UserReconcileConfig{ + Enabled: envBool("DASHBOARD_USER_RECONCILE_ENABLED", true), + Interval: envDuration("DASHBOARD_USER_RECONCILE_INTERVAL", 10*time.Minute), + LookbackDays: envInt("DASHBOARD_USER_RECONCILE_LOOKBACK_DAYS", 3), + CurrentDayLag: envDuration("DASHBOARD_USER_RECONCILE_CURRENT_DAY_LAG", time.Minute), + BatchLimit: envInt("DASHBOARD_USER_RECONCILE_BATCH_LIMIT", 10000), + CreateTimeDisplayOffset: envDuration("DASHBOARD_USER_CREATE_TIME_DISPLAY_OFFSET", 8*time.Hour), + }, Active: ActiveReconcileConfig{ Enabled: envBool("DASHBOARD_ACTIVE_RECONCILE_ENABLED", false), Interval: envDuration("DASHBOARD_ACTIVE_RECONCILE_INTERVAL", 10*time.Minute), @@ -200,6 +218,12 @@ func Load() (Config, error) { if cfg.Gift.BatchLimit <= 0 { cfg.Gift.BatchLimit = 50000 } + if cfg.User.LookbackDays <= 0 { + cfg.User.LookbackDays = 3 + } + if cfg.User.BatchLimit <= 0 { + cfg.User.BatchLimit = 10000 + } if cfg.Active.LookbackDays <= 0 { cfg.Active.LookbackDays = 35 } diff --git a/internal/dashboard/mapper.go b/internal/dashboard/mapper.go index b164a41..6214ff9 100644 --- a/internal/dashboard/mapper.go +++ b/internal/dashboard/mapper.go @@ -3,22 +3,30 @@ package dashboard import ( "context" "strings" + "time" "github.com/shopspring/decimal" ) type Mapper struct { - store Store - sysOrigins map[string]struct{} - unknownCountry string + store Store + sysOrigins map[string]struct{} + unknownCountry string + userCreateDisplayOffset time.Duration } -func NewMapper(store Store, sysOrigins []string, unknownCountry string) *Mapper { +func NewMapper(store Store, sysOrigins []string, unknownCountry string, + userCreateDisplayOffset time.Duration) *Mapper { origins := make(map[string]struct{}, len(sysOrigins)) for _, origin := range sysOrigins { origins[strings.TrimSpace(origin)] = struct{}{} } - return &Mapper{store: store, sysOrigins: origins, unknownCountry: unknownCountry} + return &Mapper{ + store: store, + sysOrigins: origins, + unknownCountry: unknownCountry, + userCreateDisplayOffset: userCreateDisplayOffset, + } } func (m *Mapper) Map(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { @@ -50,6 +58,7 @@ func (m *Mapper) mapUser(_ context.Context, schema string, table string, action if !m.allowed(user.SysOrigin) || user.IsDeleted || user.UserID == 0 || user.CreatedAt.IsZero() { return nil, nil } + displayCreateTime := user.CreatedAt.Add(m.userCreateDisplayOffset) return []Contribution{{ SourceSchema: schema, SourceTable: table, @@ -59,7 +68,7 @@ func (m *Mapper) mapUser(_ context.Context, schema string, table string, action SysOrigin: user.SysOrigin, UserID: user.UserID, Country: user.Country, - EventTime: user.CreatedAt, + EventTime: displayCreateTime, CountryNewUser: 1, }}, nil } diff --git a/internal/dashboard/mapper_wallet_test.go b/internal/dashboard/mapper_wallet_test.go index 122e54d..3934784 100644 --- a/internal/dashboard/mapper_wallet_test.go +++ b/internal/dashboard/mapper_wallet_test.go @@ -52,7 +52,7 @@ func (s walletMapperTestStore) ApplyContributionBatch(context.Context, *sql.Tx, func TestWalletLuckyGiftIsNotMappedAsGiftConsume(t *testing.T) { mapper := NewMapper(walletMapperTestStore{users: map[int64]UserCountry{ 1001: {UserID: 1001, SysOrigin: "LIKEI", Country: Country{Code: "BD", Name: "Bangladesh"}}, - }}, []string{"LIKEI"}, "UNKNOWN") + }}, []string{"LIKEI"}, "UNKNOWN", 0) row := Row{ "id": int64(11), @@ -76,7 +76,7 @@ func TestWalletLuckyGiftIsNotMappedAsGiftConsume(t *testing.T) { func TestWalletGiveGiftIsMappedAsGiftConsume(t *testing.T) { mapper := NewMapper(walletMapperTestStore{users: map[int64]UserCountry{ 1001: {UserID: 1001, SysOrigin: "LIKEI", Country: Country{Code: "BD", Name: "Bangladesh"}}, - }}, []string{"LIKEI"}, "UNKNOWN") + }}, []string{"LIKEI"}, "UNKNOWN", 0) row := Row{ "id": int64(12), @@ -99,3 +99,30 @@ func TestWalletGiveGiftIsMappedAsGiftConsume(t *testing.T) { t.Fatalf("expected gift consume 100, got %s", contributions[0].GiftConsume.String()) } } + +func TestUserCreateTimeDisplayOffsetIsApplied(t *testing.T) { + mapper := NewMapper(walletMapperTestStore{}, []string{"LIKEI"}, "UNKNOWN", 8*time.Hour) + createdAt := time.Date(2026, 5, 26, 16, 2, 51, 0, time.UTC) + + contributions, err := mapper.Map(context.Background(), "likei", "user_base_info", "insert", Row{ + "id": int64(2059304265087774722), + "origin_sys": "LIKEI", + "is_del": int64(0), + "country_code": "PH", + "country_name": "Philippines", + "create_time": createdAt, + }) + if err != nil { + t.Fatal(err) + } + if len(contributions) != 1 { + t.Fatalf("expected one user contribution, got %d", len(contributions)) + } + expected := createdAt.Add(8 * time.Hour) + if !contributions[0].EventTime.Equal(expected) { + t.Fatalf("expected event time %s, got %s", expected, contributions[0].EventTime) + } + if contributions[0].CountryNewUser != 1 { + t.Fatalf("expected country new user contribution, got %d", contributions[0].CountryNewUser) + } +} diff --git a/internal/storage/user_reconcile.go b/internal/storage/user_reconcile.go new file mode 100644 index 0000000..ba50bb0 --- /dev/null +++ b/internal/storage/user_reconcile.go @@ -0,0 +1,161 @@ +package storage + +import ( + "context" + "database/sql" + "fmt" + "log/slog" + "time" + + "dashboard-cdc-worker/internal/dashboard" +) + +const userReconcileSource = "likei.user_base_info" + +type userReconcileRow struct { + user dashboard.UserCountry +} + +func (r *MySQLRepository) StartUserReconciler(ctx context.Context, logger *slog.Logger, + projector *dashboard.Projector) { + if !r.cfg.User.Enabled || r.cfg.User.Interval <= 0 { + return + } + run := func() { + location, err := time.LoadLocation(r.cfg.Dashboard.StorageTimezone) + if err != nil { + location = time.UTC + } + now := time.Now().In(location).Add(-r.cfg.User.CurrentDayLag) + end := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), now.Minute(), now.Second(), 0, location) + start := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, location). + AddDate(0, 0, -r.cfg.User.LookbackDays+1) + + events, err := r.BuildUserReconcileEvents(ctx, start, end, r.cfg.User.BatchLimit) + if err != nil { + if r.metrics != nil { + r.metrics.RecordError(err) + } + logger.Error("dashboard user reconcile failed", "error", err) + return + } + if len(events) == 0 { + logger.Info("dashboard user reconcile skipped, no events", "since", start, "until", end) + return + } + if err := projector.ApplyEvents(ctx, events); err != nil { + if r.metrics != nil { + r.metrics.RecordError(err) + } + logger.Error("dashboard user reconcile apply failed", "events", len(events), "error", err) + return + } + logger.Info("dashboard user reconcile applied", "events", len(events), "since", start, "until", end) + } + go func() { + run() + ticker := time.NewTicker(r.cfg.User.Interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + run() + } + } + }() +} + +func (r *MySQLRepository) BuildUserReconcileEvents(ctx context.Context, since, until time.Time, + limit int) ([]dashboard.Event, error) { + rows, err := r.loadUserReconcileRows(ctx, since, until, limit) + if err != nil { + return nil, err + } + events := make([]dashboard.Event, 0, len(rows)) + for _, row := range rows { + user := row.user + displayCreateTime := user.CreatedAt.Add(r.cfg.User.CreateTimeDisplayOffset) + contribution := dashboard.Contribution{ + SourceSchema: r.cfg.MySQL.Schema, + SourceTable: "user_base_info", + SourcePK: fmt.Sprint(user.UserID), + SourceAction: "insert", + DeltaSign: 1, + SysOrigin: user.SysOrigin, + UserID: user.UserID, + Country: user.Country, + EventTime: displayCreateTime, + UserCreatedAt: user.CreatedAt, + CountryNewUser: 1, + } + events = append(events, dashboard.Event{ + EventKey: userReconcileEventKey(user.UserID), + Source: userReconcileSource, + User: &user, + Contributions: []dashboard.Contribution{contribution}, + }) + } + return events, nil +} + +func (r *MySQLRepository) loadUserReconcileRows(ctx context.Context, since, until time.Time, + limit int) ([]userReconcileRow, error) { + args := []interface{}{ + r.cfg.Dashboard.UnknownCountryCode, + r.cfg.Dashboard.UnknownCountryCode, + since, + until, + } + originSQL := "" + if len(r.cfg.Dashboard.SysOrigins) > 0 { + originSQL = " AND origin_sys IN " + inPlaceholders(len(r.cfg.Dashboard.SysOrigins)) + for _, origin := range r.cfg.Dashboard.SysOrigins { + args = append(args, origin) + } + } + limitSQL := "" + if limit > 0 { + limitSQL = " LIMIT ?" + args = append(args, limit) + } + rows, err := r.db.QueryContext(ctx, ` +SELECT id, origin_sys, + IFNULL(NULLIF(country_code, ''), ?), + IFNULL(NULLIF(country_name, ''), IFNULL(NULLIF(country_code, ''), ?)), + IFNULL(is_del, 0), create_time, update_time +FROM user_base_info +WHERE create_time >= ? AND create_time < ? + AND IFNULL(is_del, 0) = 0`+originSQL+` +ORDER BY create_time ASC`+limitSQL, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + result := make([]userReconcileRow, 0) + for rows.Next() { + var user dashboard.UserCountry + var createdAt, updatedAt sql.NullTime + if err := rows.Scan(&user.UserID, &user.SysOrigin, &user.Country.Code, &user.Country.Name, + &user.IsDeleted, &createdAt, &updatedAt); err != nil { + return nil, err + } + if createdAt.Valid { + user.CreatedAt = createdAt.Time + } + if updatedAt.Valid { + user.UpdatedAt = updatedAt.Time + } + if user.UserID == 0 || user.SysOrigin == "" || user.CreatedAt.IsZero() { + continue + } + result = append(result, userReconcileRow{user: user}) + } + return result, rows.Err() +} + +func userReconcileEventKey(userID int64) string { + return fmt.Sprintf("reconcile:%s:%d:created:v2-display-time", userReconcileSource, userID) +}