Fix dashboard lucky gift and active retention CDC

cmd/dashboard-cdc-worker/main.go

@@ -12,6 +12,8 @@ import (
 
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/redis/go-redis/v9"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
 
 	"dashboard-cdc-worker/internal/cdc"
 	"dashboard-cdc-worker/internal/config"
@@ -83,6 +85,24 @@ func main() {
 	repo.StartRechargeReconciler(ctx, logger, projector)
 	repo.StartGiftReconciler(ctx, logger, projector)
 
+	if cfg.Mongo.Enabled {
+		mongoClient, err := mongo.Connect(options.Client().ApplyURI(cfg.Mongo.URI))
+		if err != nil {
+			logger.Error("connect mongo failed", "error", err)
+			os.Exit(1)
+		}
+		defer func() {
+			shutdownCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+			defer cancel()
+			_ = mongoClient.Disconnect(shutdownCtx)
+		}()
+		if err := mongoClient.Ping(ctx, nil); err != nil {
+			logger.Error("ping mongo failed", "error", err)
+			os.Exit(1)
+		}
+		repo.StartActiveReconciler(ctx, mongoClient.Database(cfg.Mongo.Database), logger)
+	}
+
 	worker, err := cdc.NewWorker(cfg, repo, projector, logger, metrics)
 	if err != nil {
 		logger.Error("create cdc worker failed", "error", err)

env.example

@@ -9,6 +9,9 @@ CDC_BATCH_SIZE=1000
 CDC_FLUSH_INTERVAL=3s
 CDC_BACKFILL_LOCK_WAIT=1s
 CDC_MASTER_POSITION_POLL_INTERVAL=1m
+MONGO_ENABLED=false
+MONGO_URI=
+MONGO_DATABASE=
 DASHBOARD_SYS_ORIGINS=LIKEI
 DASHBOARD_STAT_TIMEZONES=Asia/Riyadh
 DASHBOARD_STORAGE_TIMEZONE=Asia/Riyadh
@@ -16,6 +19,10 @@ DASHBOARD_CLEANUP_INTERVAL=30m
 DASHBOARD_DEDUP_RETENTION=72h
 DASHBOARD_USER_METRIC_RETENTION_DAYS=120
 DASHBOARD_CLEANUP_BATCH_SIZE=5000
+DASHBOARD_ACTIVE_RECONCILE_ENABLED=false
+DASHBOARD_ACTIVE_RECONCILE_INTERVAL=10m
+DASHBOARD_ACTIVE_RECONCILE_LOOKBACK_DAYS=35
+DASHBOARD_ACTIVE_RECONCILE_BATCH_LIMIT=200000
 REDIS_ENABLED=false
 REDIS_ADDR=127.0.0.1:6379
 HEALTH_ADDR=:2910

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/go-sql-driver/mysql v1.9.3
 	github.com/redis/go-redis/v9 v9.12.1
 	github.com/shopspring/decimal v1.4.0
+	go.mongodb.org/mongo-driver/v2 v2.3.1
 )
 
 require (
@@ -15,15 +16,22 @@ require (
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/goccy/go-json v0.10.2 // indirect
+	github.com/golang/snappy v1.0.0 // indirect
 	github.com/google/uuid v1.3.0 // indirect
 	github.com/klauspost/compress v1.17.8 // indirect
 	github.com/pingcap/errors v0.11.5-0.20250318082626-8f80e5cb09ec // indirect
 	github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 // indirect
 	github.com/pingcap/log v1.1.1-0.20241212030209-7e3ff8601a2a // indirect
 	github.com/pingcap/tidb/pkg/parser v0.0.0-20250421232622-526b2c79173d // indirect
+	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
+	github.com/xdg-go/scram v1.1.2 // indirect
+	github.com/xdg-go/stringprep v1.0.4 // indirect
+	github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect
 	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect
+	golang.org/x/crypto v0.33.0 // indirect
+	golang.org/x/sync v0.13.0 // indirect
 	golang.org/x/text v0.24.0 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
 )

go.sum

@@ -20,6 +20,10 @@ github.com/go-sql-driver/mysql v1.9.3 h1:U/N249h2WzJ3Ukj8SowVFjdtZKfu9vlLZxjPXV1
 github.com/go-sql-driver/mysql v1.9.3/go.mod h1:qn46aNg1333BRMNU69Lq93t8du/dwxI64Gl8i5p1WMU=
 github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU=
 github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
+github.com/golang/snappy v1.0.0 h1:Oy607GVXHs7RtbggtPBnr2RmDArIsAefDwvrdWvRhGs=
+github.com/golang/snappy v1.0.0/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/klauspost/compress v1.17.8 h1:YcnTYrq7MikUT7k0Yb5eceMmALQPYBW/Xltxn0NAMnU=
@@ -49,6 +53,17 @@ github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81P
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
+github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY=
+github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
+github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8=
+github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
+github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 h1:ilQV1hzziu+LLM3zUTJ0trRztfwgjqKnBWNtSRkbmwM=
+github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78/go.mod h1:aL8wCCfTfSfmXjznFBSZNN13rSJjlIOI1fUNAtF7rmI=
+github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
+go.mongodb.org/mongo-driver/v2 v2.3.1 h1:WrCgSzO7dh1/FrePud9dK5fKNZOE97q5EQimGkos7Wo=
+go.mongodb.org/mongo-driver/v2 v2.3.1/go.mod h1:jHeEDJHJq7tm6ZF45Issun9dbogjfnPySb1vXA7EeAI=
 go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
@@ -65,17 +80,38 @@ go.uber.org/zap v1.19.0/go.mod h1:xg/QME4nWcxGxrpdeYfq7UvYrLh66cuVKdrbD1XF/NI=
 go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
 go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.33.0 h1:IOBPskki6Lysi0lo9qQvbxiQ+FvsCC/YWOecCHAixus=
+golang.org/x/crypto v0.33.0/go.mod h1:bVdXmD7IV/4GdElGPozy6U7lWdRXA4qyRVGJV57uQ5M=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
+golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.13.0 h1:AauUjRAJ9OSnvULf/ARrrVywoJDy0YS2AwQ98I37610=
+golang.org/x/sync v0.13.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
 golang.org/x/text v0.24.0 h1:dd5Bzh4yt5KYA8f9CJHCP4FB4D51c2c6JvN37xJJkJ0=
 golang.org/x/text v0.24.0/go.mod h1:L8rBsPeo2pSS+xqN0d5u2ikmjtmoJbDBT1b7nHvFCdU=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

internal/config/config.go

@@ -12,11 +12,13 @@ import (
 
 type Config struct {
 	MySQL     MySQLConfig
+	Mongo     MongoConfig
 	Redis     RedisConfig
 	CDC       CDCConfig
 	Dashboard DashboardConfig
 	Recharge  RechargeReconcileConfig
 	Gift      GiftReconcileConfig
+	Active    ActiveReconcileConfig
 	Health    HealthConfig
 }
 
@@ -31,6 +33,12 @@ type MySQLConfig struct {
 	MaxIdleConns int
 }
 
+type MongoConfig struct {
+	Enabled  bool
+	URI      string
+	Database string
+}
+
 type RedisConfig struct {
 	Enabled  bool
 	Addr     string
@@ -78,6 +86,13 @@ type GiftReconcileConfig struct {
 	BatchLimit    int
 }
 
+type ActiveReconcileConfig struct {
+	Enabled      bool
+	Interval     time.Duration
+	LookbackDays int
+	BatchLimit   int
+}
+
 type HealthConfig struct {
 	Addr string
 }
@@ -94,6 +109,11 @@ func Load() (Config, error) {
 			MaxOpenConns: envInt("MYSQL_MAX_OPEN_CONNS", 8),
 			MaxIdleConns: envInt("MYSQL_MAX_IDLE_CONNS", 4),
 		},
+		Mongo: MongoConfig{
+			Enabled:  envBool("MONGO_ENABLED", false),
+			URI:      env("MONGO_URI", env("LIKEI_MONGO_URI", "")),
+			Database: env("MONGO_DATABASE", env("LIKEI_MONGO_DB", "")),
+		},
 		Redis: RedisConfig{
 			Enabled:  envBool("REDIS_ENABLED", false),
 			Addr:     env("REDIS_ADDR", ""),
@@ -136,6 +156,12 @@ func Load() (Config, error) {
 			CurrentDayLag: envDuration("DASHBOARD_GIFT_RECONCILE_CURRENT_DAY_LAG", 5*time.Minute),
 			BatchLimit:    envInt("DASHBOARD_GIFT_RECONCILE_BATCH_LIMIT", 50000),
 		},
+		Active: ActiveReconcileConfig{
+			Enabled:      envBool("DASHBOARD_ACTIVE_RECONCILE_ENABLED", false),
+			Interval:     envDuration("DASHBOARD_ACTIVE_RECONCILE_INTERVAL", 10*time.Minute),
+			LookbackDays: envInt("DASHBOARD_ACTIVE_RECONCILE_LOOKBACK_DAYS", 35),
+			BatchLimit:   envInt("DASHBOARD_ACTIVE_RECONCILE_BATCH_LIMIT", 200000),
+		},
 		Health: HealthConfig{
 			Addr: env("HEALTH_ADDR", ":2910"),
 		},
@@ -153,6 +179,12 @@ func Load() (Config, error) {
 	if cfg.Redis.Enabled && cfg.Redis.Addr == "" {
 		return Config{}, errors.New("REDIS_ADDR is required when REDIS_ENABLED=true")
 	}
+	if cfg.Mongo.Enabled && cfg.Mongo.URI == "" {
+		return Config{}, errors.New("MONGO_URI or LIKEI_MONGO_URI is required when MONGO_ENABLED=true")
+	}
+	if cfg.Mongo.Enabled && cfg.Mongo.Database == "" {
+		return Config{}, errors.New("MONGO_DATABASE or LIKEI_MONGO_DB is required when MONGO_ENABLED=true")
+	}
 	if cfg.CDC.ServerID == 0 {
 		return Config{}, errors.New("CDC_SERVER_ID must be greater than 0")
 	}
@@ -168,6 +200,12 @@ func Load() (Config, error) {
 	if cfg.Gift.BatchLimit <= 0 {
 		cfg.Gift.BatchLimit = 50000
 	}
+	if cfg.Active.LookbackDays <= 0 {
+		cfg.Active.LookbackDays = 35
+	}
+	if cfg.Active.BatchLimit <= 0 {
+		cfg.Active.BatchLimit = 200000
+	}
 	if cfg.CDC.StartMode != "checkpoint" && cfg.CDC.StartMode != "current" {
 		return Config{}, fmt.Errorf("unsupported CDC_START_MODE=%s", cfg.CDC.StartMode)
 	}

internal/dashboard/mapper_wallet.go

@@ -18,7 +18,7 @@ var walletGameEvents = map[string]struct{}{
 
 func (m *Mapper) mapWalletGame(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) {
 	eventType := row.String("event_type")
-	if eventType == "LUCKY_GIFT" || isWalletGiftConsumeEvent(eventType) {
+	if isWalletGiftConsumeEvent(eventType) {
 		return m.mapWalletGift(ctx, schema, table, action, row)
 	}
 	if !isWalletGameEvent(eventType) {
@@ -87,11 +87,6 @@ func (m *Mapper) mapWalletGift(ctx context.Context, schema string, table string,
 		c.EventKey = WalletMetricEventKey(schema, table, rowPK(row))
 	}
 	switch {
-	case eventType == "LUCKY_GIFT" && recordType == 1:
-		c.LuckyGiftTotalFlow = amount
-		c.LuckyGiftUser = true
-	case eventType == "LUCKY_GIFT" && recordType == 0:
-		c.LuckyGiftPayout = amount
 	case isWalletGiftConsumeEvent(eventType) && recordType == 1:
 		c.GiftConsume = amount
 	default:

internal/dashboard/mapper_wallet_test.go

@@ -0,0 +1,101 @@
+package dashboard
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+	"time"
+
+	"github.com/go-mysql-org/go-mysql/mysql"
+	"github.com/shopspring/decimal"
+)
+
+type walletMapperTestStore struct {
+	users map[int64]UserCountry
+}
+
+func (s walletMapperTestStore) WithTx(context.Context, func(*sql.Tx) error) error { return nil }
+func (s walletMapperTestStore) WithDashboardLock(context.Context, time.Duration, func() error) error {
+	return nil
+}
+func (s walletMapperTestStore) MarkEventApplied(context.Context, *sql.Tx, string, string) (bool, error) {
+	return true, nil
+}
+func (s walletMapperTestStore) SaveCheckpoint(context.Context, mysql.Position) error { return nil }
+func (s walletMapperTestStore) LoadCheckpoint(context.Context) (mysql.Position, bool, error) {
+	return mysql.Position{}, false, nil
+}
+func (s walletMapperTestStore) Cleanup(context.Context) (int64, error) { return 0, nil }
+func (s walletMapperTestStore) UpsertUserCountry(context.Context, *sql.Tx, UserCountry) error {
+	return nil
+}
+func (s walletMapperTestStore) UserCountry(_ context.Context, userID int64) (UserCountry, bool, error) {
+	user, ok := s.users[userID]
+	return user, ok, nil
+}
+func (s walletMapperTestStore) ResolveWalletGame(context.Context, string, string, string) (string, string, string, error) {
+	return "TEST", "1", "Test", nil
+}
+func (s walletMapperTestStore) ApplyDailyContribution(context.Context, *sql.Tx, Contribution, string) error {
+	return nil
+}
+func (s walletMapperTestStore) ApplyPeriodContribution(context.Context, *sql.Tx, Contribution, string) error {
+	return nil
+}
+func (s walletMapperTestStore) ApplyGamePeriodContribution(context.Context, *sql.Tx, Contribution, string) error {
+	return nil
+}
+func (s walletMapperTestStore) ApplyContributionBatch(context.Context, *sql.Tx, []Contribution, string, []string) error {
+	return nil
+}
+
+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")
+
+	row := Row{
+		"id":           int64(11),
+		"user_id":      int64(1001),
+		"sys_origin":   "LIKEI",
+		"event_type":   "LUCKY_GIFT",
+		"type":         int64(1),
+		"penny_amount": decimal.NewFromInt(10000),
+		"create_time":  time.Now().UnixMilli(),
+	}
+
+	contributions, err := mapper.Map(context.Background(), "likei_wallet", "wallet_gold_asset_record_1", "insert", row)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(contributions) != 0 {
+		t.Fatalf("expected wallet LUCKY_GIFT to be ignored, got %#v", contributions)
+	}
+}
+
+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")
+
+	row := Row{
+		"id":           int64(12),
+		"user_id":      int64(1001),
+		"sys_origin":   "LIKEI",
+		"event_type":   "GIVE_GIFT_V3",
+		"type":         int64(1),
+		"penny_amount": decimal.NewFromInt(10000),
+		"create_time":  time.Now().UnixMilli(),
+	}
+
+	contributions, err := mapper.Map(context.Background(), "likei_wallet", "wallet_gold_asset_record_1", "insert", row)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(contributions) != 1 {
+		t.Fatalf("expected one gift contribution, got %d", len(contributions))
+	}
+	if !contributions[0].GiftConsume.Equal(decimal.NewFromInt(100)) {
+		t.Fatalf("expected gift consume 100, got %s", contributions[0].GiftConsume.String())
+	}
+}

internal/storage/active_reconcile.go

@@ -0,0 +1,500 @@
+package storage
+
+import (
+	"context"
+	"database/sql"
+	"log/slog"
+	"time"
+
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
+
+	"dashboard-cdc-worker/internal/dashboard"
+)
+
+const activeLogCollection = "user_daily_active_log"
+
+type activeLogRow struct {
+	UserID    int64  `bson:"userId"`
+	ActiveDay string `bson:"activeDate"`
+	SysOrigin string `bson:"sysOrigin"`
+}
+
+type activeDayStats struct {
+	active map[string]map[int64]struct{}
+}
+
+type activeRetentionValues struct {
+	countryName string
+	activeUsers int64
+	d1User      int64
+	d1Base      int64
+	d7User      int64
+	d7Base      int64
+	d30User     int64
+	d30Base     int64
+}
+
+type countryCount struct {
+	name  string
+	count int64
+}
+
+func (r *MySQLRepository) StartActiveReconciler(ctx context.Context, db *mongo.Database, logger *slog.Logger) {
+	if !r.cfg.Active.Enabled || r.cfg.Active.Interval <= 0 {
+		return
+	}
+	run := func() {
+		if err := r.RebuildRecentActiveRetention(ctx, db); err != nil {
+			if r.metrics != nil {
+				r.metrics.RecordError(err)
+			}
+			logger.Error("dashboard active reconcile failed", "error", err)
+			return
+		}
+		logger.Info("dashboard active reconcile applied", "lookbackDays", r.cfg.Active.LookbackDays)
+	}
+	go func() {
+		run()
+		ticker := time.NewTicker(r.cfg.Active.Interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				run()
+			}
+		}
+	}()
+}
+
+func (r *MySQLRepository) RebuildRecentActiveRetention(ctx context.Context, db *mongo.Database) error {
+	location, err := time.LoadLocation(r.cfg.Dashboard.StorageTimezone)
+	if err != nil {
+		location = time.UTC
+	}
+	today := time.Now().In(location)
+	end := time.Date(today.Year(), today.Month(), today.Day(), 0, 0, 0, 0, location).AddDate(0, 0, 1)
+	start := end.AddDate(0, 0, -r.cfg.Active.LookbackDays)
+	return r.RebuildActiveRetentionRange(ctx, db, start, end)
+}
+
+func (r *MySQLRepository) RebuildActiveRetentionRange(ctx context.Context, db *mongo.Database, start, end time.Time) error {
+	events, err := r.loadActiveLogs(ctx, db, start, end)
+	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 {
+					return err
+				}
+			}
+			for _, statTimezone := range r.cfg.Dashboard.StatTimezones {
+				if err := r.rebuildAllPeriodFromDays(ctx, tx, statTimezone); err != nil {
+					return err
+				}
+			}
+			return nil
+		})
+	})
+}
+
+func (r *MySQLRepository) loadActiveLogs(ctx context.Context, db *mongo.Database, start, end time.Time) (map[string]map[int64]struct{}, 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}}},
+	}
+	findOptions := options.Find().SetProjection(bson.D{
+		{Key: "userId", Value: 1},
+		{Key: "activeDate", Value: 1},
+		{Key: "sysOrigin", 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
+	}
+	defer cursor.Close(ctx)
+
+	events := make(map[string]map[int64]struct{})
+	for cursor.Next(ctx) {
+		var row activeLogRow
+		if err := cursor.Decode(&row); err != nil {
+			return nil, err
+		}
+		if row.UserID == 0 || row.ActiveDay == "" || !r.allowedOrigin(row.SysOrigin) {
+			continue
+		}
+		users := events[row.ActiveDay]
+		if users == nil {
+			users = make(map[int64]struct{})
+			events[row.ActiveDay] = users
+		}
+		users[row.UserID] = struct{}{}
+	}
+	return events, cursor.Err()
+}
+
+func collectActiveUserIDs(events map[string]map[int64]struct{}) []int64 {
+	seen := make(map[int64]struct{})
+	for _, users := range events {
+		for userID := range users {
+			seen[userID] = struct{}{}
+		}
+	}
+	out := make([]int64, 0, len(seen))
+	for userID := range seen {
+		out = append(out, userID)
+	}
+	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 {
+		end := min(start+bulkInsertChunkSize, len(ids))
+		args := make([]interface{}, 0, end-start)
+		for _, id := range ids[start:end] {
+			args = append(args, id)
+		}
+		rows, err := r.db.QueryContext(ctx, `
+SELECT user_id, sys_origin, country_code, country_name, is_deleted, user_created_at, user_updated_at
+FROM dashboard_cdc_user_country
+WHERE user_id IN `+inPlaceholders(end-start), args...)
+		if err != nil {
+			return nil, err
+		}
+		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 {
+				_ = rows.Close()
+				return nil, err
+			}
+			if createdAt.Valid {
+				user.CreatedAt = createdAt.Time
+			}
+			if updatedAt.Valid {
+				user.UpdatedAt = updatedAt.Time
+			}
+			result[user.UserID] = user
+		}
+		if err := rows.Close(); err != nil {
+			return nil, err
+		}
+	}
+	return result, nil
+}
+
+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 {
+	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) {
+			continue
+		}
+		stats := statsByOrigin[user.SysOrigin]
+		if stats == nil {
+			stats = make(map[string]*activeDayStats)
+			statsByOrigin[user.SysOrigin] = stats
+		}
+		stat := stats[user.Country.Code]
+		if stat == nil {
+			stat = &activeDayStats{active: make(map[string]map[int64]struct{})}
+			stats[user.Country.Code] = stat
+		}
+		active := stat.active[user.Country.Name]
+		if active == nil {
+			active = make(map[int64]struct{})
+			stat.active[user.Country.Name] = active
+		}
+		active[userID] = struct{}{}
+	}
+	retentionByOrigin, err := r.retentionStatsForDay(ctx, day, events[dayKey], users)
+	if err != nil {
+		return err
+	}
+	for _, statTimezone := range r.cfg.Dashboard.StatTimezones {
+		for _, sysOrigin := range r.cfg.Dashboard.SysOrigins {
+			if _, err := tx.ExecContext(ctx, `
+UPDATE country_dashboard_period_metric
+SET daily_active_user = 0,
+    d1_retention_user = 0, d1_retention_base_user = 0,
+    d7_retention_user = 0, d7_retention_base_user = 0,
+    d30_retention_user = 0, d30_retention_base_user = 0,
+    refreshed_at = NOW()
+WHERE period_type = 'DAY' AND period_key = ? AND stat_timezone = ? AND sys_origin = ?`, dayKey, statTimezone, sysOrigin); err != nil {
+				return err
+			}
+			if err := r.upsertActiveRetentionRows(ctx, tx, day, statTimezone, sysOrigin,
+				statsByOrigin[sysOrigin], retentionByOrigin[sysOrigin]); err != nil {
+				return err
+			}
+		}
+	}
+	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))
+		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) {
+			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
+}
+
+func (r *MySQLRepository) loadRetentionBase(ctx context.Context, cohortDay time.Time) (map[string]map[string]countryCount, error) {
+	args := stringArgs(r.cfg.Dashboard.SysOrigins)
+	args = append(args, cohortDay, cohortDay.AddDate(0, 0, 1))
+	rows, err := r.db.QueryContext(ctx, `
+SELECT sys_origin, country_code, MAX(country_name), COUNT(1)
+FROM dashboard_cdc_user_country
+WHERE is_deleted = 0
+  AND sys_origin IN `+inStringPlaceholders(len(r.cfg.Dashboard.SysOrigins))+`
+  AND user_created_at >= ?
+  AND user_created_at < ?
+GROUP BY sys_origin, country_code`, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	result := make(map[string]map[string]countryCount)
+	for rows.Next() {
+		var origin, country, name string
+		var count int64
+		if err := rows.Scan(&origin, &country, &name, &count); err != nil {
+			return nil, err
+		}
+		countries := result[origin]
+		if countries == nil {
+			countries = make(map[string]countryCount)
+			result[origin] = countries
+		}
+		countries[country] = countryCount{name: name, count: count}
+	}
+	return result, rows.Err()
+}
+
+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)
+	for country, stat := range active {
+		item := values[country]
+		for name, users := range stat.active {
+			item.countryName = name
+			item.activeUsers += int64(len(users))
+		}
+		values[country] = item
+	}
+	for country, item := range retention {
+		existing := values[country]
+		if existing.countryName == "" {
+			existing.countryName = item.countryName
+		}
+		existing.d1User = item.d1User
+		existing.d1Base = item.d1Base
+		existing.d7User = item.d7User
+		existing.d7Base = item.d7Base
+		existing.d30User = item.d30User
+		existing.d30Base = item.d30Base
+		values[country] = existing
+	}
+	if len(values) == 0 {
+		return nil
+	}
+	dayKey := day.Format("2006-01-02")
+	args := make([]interface{}, 0, len(values)*16)
+	for country, item := range values {
+		name := item.countryName
+		if name == "" {
+			name = country
+		}
+		args = append(args, dashboard.PeriodDay, dayKey, statTimezone, dayKey, day,
+			day.AddDate(0, 0, 1), sysOrigin, country, name, item.activeUsers,
+			item.d1User, item.d1Base, item.d7User, item.d7Base, item.d30User, item.d30Base)
+	}
+	_, err := tx.ExecContext(ctx, `
+INSERT INTO country_dashboard_period_metric (
+  period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date,
+  sys_origin, country_code, country_name, daily_active_user,
+  d1_retention_user, d1_retention_base_user,
+  d7_retention_user, d7_retention_base_user,
+  d30_retention_user, d30_retention_base_user,
+  refreshed_at
+) VALUES `+placeholders(len(values), 16)+`
+ON DUPLICATE KEY UPDATE
+  period_name = VALUES(period_name),
+  period_start_date = VALUES(period_start_date),
+  period_end_date = VALUES(period_end_date),
+  country_name = VALUES(country_name),
+  daily_active_user = VALUES(daily_active_user),
+  d1_retention_user = VALUES(d1_retention_user),
+  d1_retention_base_user = VALUES(d1_retention_base_user),
+  d7_retention_user = VALUES(d7_retention_user),
+  d7_retention_base_user = VALUES(d7_retention_base_user),
+  d30_retention_user = VALUES(d30_retention_user),
+  d30_retention_base_user = VALUES(d30_retention_base_user),
+  refreshed_at = NOW()`, args...)
+	return err
+}
+
+func (r *MySQLRepository) rebuildAllPeriodFromDays(ctx context.Context, tx *sql.Tx, statTimezone string) error {
+	for _, sysOrigin := range r.cfg.Dashboard.SysOrigins {
+		if _, err := tx.ExecContext(ctx, `
+DELETE FROM country_dashboard_period_user_metric
+WHERE period_type = 'ALL' AND period_key = 'ALL' AND stat_timezone = ? AND sys_origin = ?`, statTimezone, sysOrigin); err != nil {
+			return err
+		}
+		if _, err := tx.ExecContext(ctx, `
+INSERT IGNORE INTO country_dashboard_period_user_metric (
+  period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date,
+  sys_origin, country_code, country_name, user_id, metric_type
+)
+SELECT DISTINCT 'ALL', 'ALL', ?, '全部', NULL, NULL,
+  sys_origin, country_code, country_name, user_id, metric_type
+FROM country_dashboard_period_user_metric
+WHERE period_type = 'DAY' AND stat_timezone = ? AND sys_origin = ?`, statTimezone, statTimezone, sysOrigin); err != nil {
+			return err
+		}
+		if _, err := tx.ExecContext(ctx, `
+DELETE FROM country_dashboard_period_metric
+WHERE period_type = 'ALL' AND period_key = 'ALL' AND stat_timezone = ? AND sys_origin = ?`, statTimezone, sysOrigin); err != nil {
+			return err
+		}
+		if _, err := tx.ExecContext(ctx, `
+INSERT INTO country_dashboard_period_metric (
+  period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date,
+  sys_origin, country_code, country_name,
+  country_new_user, daily_active_user, new_user_recharge, official_recharge, mifapay_recharge,
+  google_recharge, dealer_recharge, salary_exchange, salary_transfer, gift_consume,
+  lucky_gift_total_flow, lucky_gift_user, lucky_gift_payout, lucky_gift_anchor_share,
+  game_total_flow, game_user, game_payout,
+  d1_retention_user, d1_retention_base_user, d7_retention_user, d7_retention_base_user,
+  d30_retention_user, d30_retention_base_user, refreshed_at
+)
+SELECT 'ALL', 'ALL', ?, '全部', NULL, NULL,
+  d.sys_origin, d.country_code, MAX(d.country_name),
+  IFNULL(MAX(u.country_new_user), 0),
+  IFNULL(SUM(d.daily_active_user), 0),
+  IFNULL(SUM(d.new_user_recharge), 0),
+  IFNULL(SUM(d.official_recharge), 0),
+  IFNULL(SUM(d.mifapay_recharge), 0),
+  IFNULL(SUM(d.google_recharge), 0),
+  IFNULL(SUM(d.dealer_recharge), 0),
+  IFNULL(SUM(d.salary_exchange), 0),
+  IFNULL(SUM(d.salary_transfer), 0),
+  IFNULL(SUM(d.gift_consume), 0),
+  IFNULL(SUM(d.lucky_gift_total_flow), 0),
+  IFNULL(MAX(u.lucky_gift_user), 0),
+  IFNULL(SUM(d.lucky_gift_payout), 0),
+  IFNULL(SUM(d.lucky_gift_anchor_share), 0),
+  IFNULL(SUM(d.game_total_flow), 0),
+  IFNULL(MAX(u.game_user), 0),
+  IFNULL(SUM(d.game_payout), 0),
+  IFNULL(SUM(d.d1_retention_user), 0),
+  IFNULL(SUM(d.d1_retention_base_user), 0),
+  IFNULL(SUM(d.d7_retention_user), 0),
+  IFNULL(SUM(d.d7_retention_base_user), 0),
+  IFNULL(SUM(d.d30_retention_user), 0),
+  IFNULL(SUM(d.d30_retention_base_user), 0),
+  NOW()
+FROM country_dashboard_period_metric d
+LEFT JOIN (
+  SELECT country_code,
+    SUM(metric_type = 'COUNTRY_NEW_USER') AS country_new_user,
+    SUM(metric_type = 'LUCKY_GIFT_USER') AS lucky_gift_user,
+    SUM(metric_type = 'GAME_USER') AS game_user
+  FROM country_dashboard_period_user_metric
+  WHERE period_type = 'ALL' AND period_key = 'ALL' AND stat_timezone = ? AND sys_origin = ?
+  GROUP BY country_code
+) u ON u.country_code = d.country_code
+WHERE d.period_type = 'DAY' AND d.stat_timezone = ? AND d.sys_origin = ?
+GROUP BY d.sys_origin, d.country_code`, statTimezone, statTimezone, sysOrigin, statTimezone, sysOrigin); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func inPlaceholders(count int) string {
+	return "(" + placeholders(count, 1) + ")"
+}
+
+func inStringPlaceholders(count int) string {
+	return "(" + placeholders(count, 1) + ")"
+}
+
+func stringArgs(values []string) []interface{} {
+	args := make([]interface{}, 0, len(values))
+	for _, value := range values {
+		args = append(args, value)
+	}
+	return args
+}

internal/storage/gift_reconcile.go

@@ -54,11 +54,6 @@ func (r *MySQLRepository) BuildGiftReconcileEvents(ctx context.Context, since ti
 			UserCreatedAt: row.user.CreatedAt,
 		}
 		switch {
-		case row.eventType == "LUCKY_GIFT" && row.recordType == 1:
-			contribution.LuckyGiftTotalFlow = row.amount
-			contribution.LuckyGiftUser = true
-		case row.eventType == "LUCKY_GIFT" && row.recordType == 0:
-			contribution.LuckyGiftPayout = row.amount
 		case isGiftConsumeEvent(row.eventType) && row.recordType == 1:
 			contribution.GiftConsume = row.amount
 		default:
@@ -186,10 +181,9 @@ FROM %s.%s w
 JOIN %s.user_base_info u ON u.id = w.user_id
 WHERE w.id > ?
   AND w.create_time >= ?
-  AND (
+  AND w.event_type LIKE 'GIVE_GIFT%%'
-    w.event_type = 'LUCKY_GIFT'
+  AND w.event_type NOT LIKE '%%LUCKY_GIFT%%'
-    OR (w.event_type LIKE 'GIVE_GIFT%%' AND w.event_type NOT LIKE '%%LUCKY_GIFT%%' AND w.type = 1)
+  AND w.type = 1
-  )
 ORDER BY w.id ASC
 LIMIT ?`, quoteIdent(r.cfg.MySQL.WalletSchema), quoteIdent(table), quoteIdent(r.cfg.MySQL.Schema))
 	rows, err := r.db.QueryContext(ctx, query, r.cfg.Dashboard.UnknownCountryCode,

internal/storage/schema.go

@@ -34,7 +34,8 @@ func (r *MySQLRepository) EnsureSchema(ctx context.Context) error {
   user_updated_at datetime NULL,
   updated_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
   PRIMARY KEY (user_id),
-  KEY idx_dashboard_cdc_user_country_origin_country (sys_origin, country_code)
+  KEY idx_dashboard_cdc_user_country_origin_country (sys_origin, country_code),
+  KEY idx_dashboard_cdc_user_country_origin_created (sys_origin, user_created_at, country_code)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
 		createDailyMetricTable,
 		createDailyUserMetricTable,
@@ -128,8 +129,12 @@ func (r *MySQLRepository) ensureMetricColumns(ctx context.Context) error {
 		"ALTER TABLE country_dashboard_daily_metric ADD COLUMN salary_transfer decimal(20,2) NOT NULL DEFAULT 0 AFTER salary_exchange"); err != nil {
 		return err
 	}
-	return r.addColumnIfMissing(ctx, "country_dashboard_period_metric", "salary_transfer",
+	if err := r.addColumnIfMissing(ctx, "country_dashboard_period_metric", "salary_transfer",
-		"ALTER TABLE country_dashboard_period_metric ADD COLUMN salary_transfer decimal(20,2) NOT NULL DEFAULT 0 AFTER salary_exchange")
+		"ALTER TABLE country_dashboard_period_metric ADD COLUMN salary_transfer decimal(20,2) NOT NULL DEFAULT 0 AFTER salary_exchange"); err != nil {
+		return err
+	}
+	return r.addIndexIfMissing(ctx, "dashboard_cdc_user_country", "idx_dashboard_cdc_user_country_origin_created",
+		"ALTER TABLE dashboard_cdc_user_country ADD INDEX idx_dashboard_cdc_user_country_origin_created (sys_origin, user_created_at, country_code)")
 }
 
 func (r *MySQLRepository) addColumnIfMissing(ctx context.Context, table string, column string, ddl string) error {
@@ -153,6 +158,27 @@ WHERE table_schema = DATABASE()
 	return nil
 }
 
+func (r *MySQLRepository) addIndexIfMissing(ctx context.Context, table string, indexName string, ddl string) error {
+	var exists int
+	err := r.db.QueryRowContext(ctx, `
+SELECT COUNT(1)
+FROM information_schema.statistics
+WHERE table_schema = DATABASE()
+  AND table_name = ?
+  AND index_name = ?
+`, table, indexName).Scan(&exists)
+	if err != nil {
+		return err
+	}
+	if exists > 0 {
+		return nil
+	}
+	if _, err := r.db.ExecContext(ctx, ddl); err != nil && !isDuplicateKeyError(err) {
+		return err
+	}
+	return nil
+}
+
 func isDuplicateColumnError(err error) bool {
 	var mysqlErr *driverMysql.MySQLError
 	if err == nil {
@@ -164,6 +190,17 @@ func isDuplicateColumnError(err error) bool {
 	return fmt.Sprint(err) == "Error 1060 (42S21): Duplicate column name 'salary_transfer'"
 }
 
+func isDuplicateKeyError(err error) bool {
+	var mysqlErr *driverMysql.MySQLError
+	if err == nil {
+		return false
+	}
+	if errors.As(err, &mysqlErr) {
+		return mysqlErr.Number == 1061
+	}
+	return false
+}
+
 const createPeriodUserMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_period_user_metric (
   id bigint NOT NULL AUTO_INCREMENT,
   period_type varchar(16) NOT NULL,