Align new user metric with admin display date

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)
 

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(),

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
 	}

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
 }

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)
+	}
+}

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)
+}