fix: stabilize dashboard cdc worker

cmd/dashboard-cdc-worker/main.go

@@ -17,6 +17,7 @@ import (
 	"dashboard-cdc-worker/internal/config"
 	"dashboard-cdc-worker/internal/dashboard"
 	"dashboard-cdc-worker/internal/health"
+	"dashboard-cdc-worker/internal/monitor"
 	"dashboard-cdc-worker/internal/storage"
 )
 
@@ -42,7 +43,8 @@ func main() {
 
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer stop()
-	healthServer := health.New(cfg.Health.Addr)
+	metrics := monitor.New()
+	healthServer := health.New(cfg.Health.Addr, metrics)
 	healthServer.Start(logger)
 	defer func() {
 		shutdownCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
@@ -55,11 +57,12 @@ func main() {
 		os.Exit(1)
 	}
 
-	repo := storage.NewMySQLRepository(db, cfg)
+	repo := storage.NewMySQLRepository(db, cfg, metrics)
 	if err := repo.EnsureSchema(ctx); err != nil {
 		logger.Error("ensure schema failed", "error", err)
 		os.Exit(1)
 	}
+	repo.StartCleanup(ctx, logger)
 
 	var redisClient *redis.Client
 	if cfg.Redis.Enabled {
@@ -76,8 +79,8 @@ func main() {
 		}
 	}
 
-	projector := dashboard.NewProjector(cfg, repo, redisClient)
+	projector := dashboard.NewProjector(cfg, repo, redisClient, metrics)
-	worker, err := cdc.NewWorker(cfg, repo, projector, logger)
+	worker, err := cdc.NewWorker(cfg, repo, projector, logger, metrics)
 	if err != nil {
 		logger.Error("create cdc worker failed", "error", err)
 		os.Exit(1)

env.example

@@ -5,9 +5,17 @@ DASHBOARD_MYSQL_DSN=dashboard_cdc:password@tcp(127.0.0.1:3306)/likei?charset=utf
 CDC_START_MODE=checkpoint
 CDC_SERVER_ID=29301
 CDC_DRY_RUN=true
+CDC_BATCH_SIZE=1000
+CDC_FLUSH_INTERVAL=3s
+CDC_BACKFILL_LOCK_WAIT=1s
+CDC_MASTER_POSITION_POLL_INTERVAL=1m
 DASHBOARD_SYS_ORIGINS=LIKEI
 DASHBOARD_STAT_TIMEZONES=Asia/Riyadh
 DASHBOARD_STORAGE_TIMEZONE=Asia/Riyadh
+DASHBOARD_CLEANUP_INTERVAL=30m
+DASHBOARD_DEDUP_RETENTION=72h
+DASHBOARD_USER_METRIC_RETENTION_DAYS=120
+DASHBOARD_CLEANUP_BATCH_SIZE=5000
 REDIS_ENABLED=false
 REDIS_ADDR=127.0.0.1:6379
 HEALTH_ADDR=:2910

internal/cdc/handler.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log/slog"
 	"sync"
+	"time"
 
 	"github.com/go-mysql-org/go-mysql/canal"
 	"github.com/go-mysql-org/go-mysql/mysql"
@@ -12,6 +13,7 @@ import (
 
 	"dashboard-cdc-worker/internal/config"
 	"dashboard-cdc-worker/internal/dashboard"
+	"dashboard-cdc-worker/internal/monitor"
 )
 
 type handler struct {
@@ -23,15 +25,22 @@ type handler struct {
 	logger    *slog.Logger
 	mu        sync.RWMutex
 	file      string
+	batch     []dashboard.Event
+	lastFlush time.Time
+	metrics   *monitor.Metrics
 }
 
-func newHandler(cfg config.Config, store dashboard.Store, projector *dashboard.Projector, logger *slog.Logger) *handler {
+func newHandler(cfg config.Config, store dashboard.Store, projector *dashboard.Projector,
+	logger *slog.Logger, metrics *monitor.Metrics) *handler {
 	return &handler{
 		cfg:       cfg,
 		store:     store,
 		projector: projector,
 		mapper:    dashboard.NewMapper(store, cfg.Dashboard.SysOrigins, cfg.Dashboard.UnknownCountryCode),
 		logger:    logger,
+		batch:     make([]dashboard.Event, 0, cfg.CDC.BatchSize),
+		lastFlush: time.Now(),
+		metrics:   metrics,
 	}
 }
 
@@ -47,18 +56,21 @@ func (h *handler) OnRow(e *canal.RowsEvent) error {
 	case canal.InsertAction:
 		for index, row := range e.Rows {
 			if err := h.applyRows(ctx, e, index, nil, row); err != nil {
+				h.recordError(err)
 				return err
 			}
 		}
 	case canal.DeleteAction:
 		for index, row := range e.Rows {
 			if err := h.applyRows(ctx, e, index, row, nil); err != nil {
+				h.recordError(err)
 				return err
 			}
 		}
 	case canal.UpdateAction:
 		for index := 0; index+1 < len(e.Rows); index += 2 {
 			if err := h.applyRows(ctx, e, index/2, e.Rows[index], e.Rows[index+1]); err != nil {
+				h.recordError(err)
 				return err
 			}
 		}
@@ -72,10 +84,24 @@ func (h *handler) OnPosSynced(_ *replication.EventHeader, pos mysql.Position, _
 	if h.cfg.CDC.DryRun {
 		return nil
 	}
-	if !force && h.cfg.CDC.FlushInterval > 0 {
+	if len(h.batch) > 0 {
-		// canal already throttles calls; keep this hook simple and always persist when invoked.
+		if force || h.shouldFlush() {
+			if err := h.flushBatch(context.Background()); err != nil {
+				h.recordError(err)
+				return err
 			}
-	return h.store.SaveCheckpoint(context.Background(), pos)
+		} else {
+			return nil
+		}
+	}
+	if err := h.store.SaveCheckpoint(context.Background(), pos); err != nil {
+		h.recordError(err)
+		return err
+	}
+	if h.metrics != nil {
+		h.metrics.SetLastSyncedPosition(pos)
+	}
+	return nil
 }
 
 func (h *handler) String() string {
@@ -120,9 +146,55 @@ func (h *handler) applyRows(ctx context.Context, e *canal.RowsEvent, rowIndex in
 		contributions = append(contributions, items...)
 	}
 
+	if user == nil && len(contributions) == 0 {
+		return nil
+	}
 	eventKey := fmt.Sprintf("%s:%d:%s:%s:%d", h.currentFile(), e.Header.LogPos, source, e.Action, rowIndex)
-	if user != nil {
+	event := dashboard.Event{
-		return h.projector.ApplyUser(ctx, eventKey, source, *user, contributions)
+		EventKey:      eventKey,
+		Source:        source,
+		User:          user,
+		Contributions: contributions,
+	}
+	h.batch = append(h.batch, event)
+	if h.shouldFlush() {
+		return h.flushBatch(ctx)
+	}
+	return nil
+}
+
+func (h *handler) shouldFlush() bool {
+	if len(h.batch) == 0 {
+		return false
+	}
+	if len(h.batch) >= h.cfg.CDC.BatchSize {
+		return true
+	}
+	return h.cfg.CDC.FlushInterval > 0 && time.Since(h.lastFlush) >= h.cfg.CDC.FlushInterval
+}
+
+func (h *handler) flushBatch(ctx context.Context) error {
+	if len(h.batch) == 0 {
+		h.lastFlush = time.Now()
+		return nil
+	}
+	events := make([]dashboard.Event, len(h.batch))
+	copy(events, h.batch)
+	start := time.Now()
+	if err := h.projector.ApplyEvents(ctx, events); err != nil {
+		return err
+	}
+	if h.metrics != nil {
+		h.metrics.RecordBatch(len(events), time.Since(start))
+	}
+	h.logger.Info("dashboard cdc batch flushed", "events", len(events))
+	h.batch = h.batch[:0]
+	h.lastFlush = time.Now()
+	return nil
+}
+
+func (h *handler) recordError(err error) {
+	if h.metrics != nil {
+		h.metrics.RecordError(err)
 	}
-	return h.projector.Apply(ctx, eventKey, source, contributions)
 }

internal/cdc/worker.go

@@ -12,6 +12,7 @@ import (
 
 	"dashboard-cdc-worker/internal/config"
 	"dashboard-cdc-worker/internal/dashboard"
+	"dashboard-cdc-worker/internal/monitor"
 )
 
 type Worker struct {
@@ -20,9 +21,11 @@ type Worker struct {
 	projector *dashboard.Projector
 	logger    *slog.Logger
 	canal     *canal.Canal
+	metrics   *monitor.Metrics
 }
 
-func NewWorker(cfg config.Config, store dashboard.Store, projector *dashboard.Projector, logger *slog.Logger) (*Worker, error) {
+func NewWorker(cfg config.Config, store dashboard.Store, projector *dashboard.Projector,
+	logger *slog.Logger, metrics *monitor.Metrics) (*Worker, error) {
 	canalCfg := canal.NewDefaultConfig()
 	canalCfg.Addr = cfg.MySQL.CDCAddr
 	canalCfg.User = cfg.MySQL.CDCUser
@@ -51,16 +54,18 @@ func NewWorker(cfg config.Config, store dashboard.Store, projector *dashboard.Pr
 		projector: projector,
 		logger:    logger,
 		canal:     instance,
+		metrics:   metrics,
 	}, nil
 }
 
 func (w *Worker) Run(ctx context.Context) error {
-	handler := newHandler(w.cfg, w.store, w.projector, w.logger)
+	handler := newHandler(w.cfg, w.store, w.projector, w.logger, w.metrics)
 	w.canal.SetEventHandler(handler)
 	go func() {
 		<-ctx.Done()
 		w.canal.Close()
 	}()
+	w.startMasterPositionReporter(ctx)
 
 	if w.cfg.CDC.StartMode == "checkpoint" {
 		pos, ok, err := w.store.LoadCheckpoint(ctx)
@@ -69,6 +74,9 @@ func (w *Worker) Run(ctx context.Context) error {
 		}
 		if ok && pos.Name != "" && pos.Pos > 0 {
 			handler.setCurrentFile(pos.Name)
+			if w.metrics != nil {
+				w.metrics.SetStartPosition(pos)
+			}
 			w.logger.Info("starting cdc from checkpoint", "position", pos.String())
 			return w.canal.RunFrom(pos)
 		}
@@ -79,10 +87,42 @@ func (w *Worker) Run(ctx context.Context) error {
 		return fmt.Errorf("get master position: %w", err)
 	}
 	handler.setCurrentFile(pos.Name)
+	if w.metrics != nil {
+		w.metrics.SetStartPosition(pos)
+		w.metrics.SetMasterPosition(pos)
+	}
+	if w.cfg.CDC.StartMode == "current" {
+		if err := w.store.SaveCheckpoint(ctx, pos); err != nil {
+			return fmt.Errorf("save current checkpoint: %w", err)
+		}
+	}
 	w.logger.Info("starting cdc from current master position", "position", pos.String())
 	return w.canal.RunFrom(pos)
 }
 
+func (w *Worker) startMasterPositionReporter(ctx context.Context) {
+	if w.metrics == nil || w.cfg.CDC.MasterPollEvery <= 0 {
+		return
+	}
+	go func() {
+		ticker := time.NewTicker(w.cfg.CDC.MasterPollEvery)
+		defer ticker.Stop()
+		for {
+			pos, err := w.canal.GetMasterPos()
+			if err == nil {
+				w.metrics.SetMasterPosition(pos)
+			} else {
+				w.metrics.RecordError(err)
+			}
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
+		}
+	}()
+}
+
 func includeRegex(tables []string) []string {
 	out := make([]string, 0, len(tables))
 	for _, table := range tables {

internal/config/config.go

@@ -44,6 +44,9 @@ type CDCConfig struct {
 	IncludeTables    []string
 	FlushInterval    time.Duration
 	HeartbeatEvery   time.Duration
+	BatchSize        int
+	BackfillLockWait time.Duration
+	MasterPollEvery  time.Duration
 }
 
 type DashboardConfig struct {
@@ -53,6 +56,10 @@ type DashboardConfig struct {
 	RedisKeyPrefix     string
 	RealtimeRedisTTL   time.Duration
 	UnknownCountryCode string
+	CleanupInterval    time.Duration
+	DedupRetention     time.Duration
+	UserMetricDays     int
+	CleanupBatchSize   int
 }
 
 type HealthConfig struct {
@@ -85,6 +92,9 @@ func Load() (Config, error) {
 			IncludeTables:    csv(env("CDC_INCLUDE_TABLES", defaultTables())),
 			FlushInterval:    envDuration("CDC_FLUSH_INTERVAL", 3*time.Second),
 			HeartbeatEvery:   envDuration("CDC_HEARTBEAT_EVERY", 30*time.Second),
+			BatchSize:        envInt("CDC_BATCH_SIZE", 1000),
+			BackfillLockWait: envDuration("CDC_BACKFILL_LOCK_WAIT", time.Second),
+			MasterPollEvery:  envDuration("CDC_MASTER_POSITION_POLL_INTERVAL", time.Minute),
 		},
 		Dashboard: DashboardConfig{
 			SysOrigins:         csv(env("DASHBOARD_SYS_ORIGINS", "LIKEI")),
@@ -93,6 +103,10 @@ func Load() (Config, error) {
 			RedisKeyPrefix:     env("DASHBOARD_REDIS_KEY_PREFIX", "country_dashboard"),
 			RealtimeRedisTTL:   envDuration("DASHBOARD_REALTIME_REDIS_TTL", 48*time.Hour),
 			UnknownCountryCode: env("DASHBOARD_UNKNOWN_COUNTRY_CODE", "UNKNOWN"),
+			CleanupInterval:    envDuration("DASHBOARD_CLEANUP_INTERVAL", 30*time.Minute),
+			DedupRetention:     envDuration("DASHBOARD_DEDUP_RETENTION", 72*time.Hour),
+			UserMetricDays:     envInt("DASHBOARD_USER_METRIC_RETENTION_DAYS", 120),
+			CleanupBatchSize:   envInt("DASHBOARD_CLEANUP_BATCH_SIZE", 5000),
 		},
 		Health: HealthConfig{
 			Addr: env("HEALTH_ADDR", ":2910"),
@@ -114,6 +128,12 @@ func Load() (Config, error) {
 	if cfg.CDC.ServerID == 0 {
 		return Config{}, errors.New("CDC_SERVER_ID must be greater than 0")
 	}
+	if cfg.CDC.BatchSize <= 0 {
+		cfg.CDC.BatchSize = 1000
+	}
+	if cfg.Dashboard.CleanupBatchSize <= 0 {
+		cfg.Dashboard.CleanupBatchSize = 5000
+	}
 	if cfg.CDC.StartMode != "checkpoint" && cfg.CDC.StartMode != "current" {
 		return Config{}, fmt.Errorf("unsupported CDC_START_MODE=%s", cfg.CDC.StartMode)
 	}

internal/dashboard/mapper.go

@@ -89,6 +89,8 @@ func (m *Mapper) mapOfficialRecharge(ctx context.Context, schema string, table s
 		c.GoogleRecharge = amount
 	}
 	c.NewUserRechargeCandidate = amount
+	c.RechargeDetail = rechargeDetail("OFFICIAL", rowPK(row), payPlatform, row, user, amount,
+		row.Decimal("coin_quantity"), "")
 	return []Contribution{c}, nil
 }
 
@@ -98,18 +100,30 @@ func (m *Mapper) mapMifapayRecharge(ctx context.Context, schema string, table st
 		row.String("refund_status") != "NONE" {
 		return nil, nil
 	}
-	return m.amountContribution(ctx, schema, table, action, row, row.Decimal("compute_usd_amount"), func(c *Contribution, amount decimal.Decimal) {
+	items, err := m.amountContribution(ctx, schema, table, action, row, row.Decimal("compute_usd_amount"), func(c *Contribution, amount decimal.Decimal) {
 		c.MifapayRecharge = amount
 	})
+	if err != nil || len(items) == 0 {
+		return items, err
+	}
+	items[0].RechargeDetail = rechargeDetail("MIFAPAY", rowPK(row), row.String("factory_code"), row,
+		detailUser(items[0]), row.Decimal("compute_usd_amount"), decimal.Zero, "")
+	return items, nil
 }
 
 func (m *Mapper) mapDealerRecharge(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) {
 	if row.String("origin") != "PURCHASE" || row.Int64("type") != 0 || !row.Decimal("amount").GreaterThan(decimal.Zero) {
 		return nil, nil
 	}
-	return m.amountContribution(ctx, schema, table, action, row, row.Decimal("amount"), func(c *Contribution, amount decimal.Decimal) {
+	items, err := m.amountContribution(ctx, schema, table, action, row, row.Decimal("amount"), func(c *Contribution, amount decimal.Decimal) {
 		c.DealerRecharge = amount
 	})
+	if err != nil || len(items) == 0 {
+		return items, err
+	}
+	items[0].RechargeDetail = rechargeDetail("DEALER", rowPK(row), row.String("recharge_type"), row,
+		detailUser(items[0]), row.Decimal("amount"), row.Decimal("quantity"), row.String("remark"))
+	return items, nil
 }
 
 func (m *Mapper) mapSalaryExchange(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) {

internal/dashboard/mapper_helpers.go

@@ -90,6 +90,33 @@ func rowPK(row Row) string {
 	return strings.TrimSpace(row.String("event_id"))
 }
 
+func rechargeDetail(sourceType string, recordID string, sourceChannel string, row Row, user UserCountry,
+	amount decimal.Decimal, coinQuantity decimal.Decimal, remark string) *RechargeDetail {
+	if recordID == "" || user.UserID == 0 {
+		return nil
+	}
+	return &RechargeDetail{
+		SourceType:    sourceType,
+		RecordID:      recordID,
+		SourceChannel: strings.TrimSpace(sourceChannel),
+		SysOrigin:     user.SysOrigin,
+		UserID:        user.UserID,
+		Country:       user.Country,
+		Amount:        amount,
+		CoinQuantity:  coinQuantity,
+		Remark:        strings.TrimSpace(remark),
+		EventTime:     row.Time("create_time"),
+	}
+}
+
+func detailUser(c Contribution) UserCountry {
+	return UserCountry{
+		UserID:    c.UserID,
+		SysOrigin: c.SysOrigin,
+		Country:   c.Country,
+	}
+}
+
 func signForAction(action string) int {
 	if action == "delete" {
 		return -1

internal/dashboard/projector.go

@@ -3,12 +3,14 @@ package dashboard
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"log/slog"
 	"time"
 
 	"github.com/redis/go-redis/v9"
 
 	"dashboard-cdc-worker/internal/config"
+	"dashboard-cdc-worker/internal/monitor"
 )
 
 type Projector struct {
@@ -16,21 +18,17 @@ type Projector struct {
 	store       Store
 	redisClient *redis.Client
 	storageZone *time.Location
-	statZones   map[string]*time.Location
+	metrics     *monitor.Metrics
 }
 
-func NewProjector(cfg config.Config, store Store, redisClient *redis.Client) *Projector {
+func NewProjector(cfg config.Config, store Store, redisClient *redis.Client, metrics *monitor.Metrics) *Projector {
 	storageZone := mustLocation(cfg.Dashboard.StorageTimezone)
-	statZones := make(map[string]*time.Location, len(cfg.Dashboard.StatTimezones))
-	for _, zone := range cfg.Dashboard.StatTimezones {
-		statZones[zone] = mustLocation(zone)
-	}
 	return &Projector{
 		cfg:         cfg,
 		store:       store,
 		redisClient: redisClient,
 		storageZone: storageZone,
-		statZones:   statZones,
+		metrics:     metrics,
 	}
 }
 
@@ -43,53 +41,93 @@ func (p *Projector) ApplyUser(ctx context.Context, eventKey string, source strin
 }
 
 func (p *Projector) apply(ctx context.Context, eventKey string, source string, user *UserCountry, contributions []Contribution) error {
-	if user == nil && len(contributions) == 0 {
+	event := Event{EventKey: eventKey, Source: source, User: user, Contributions: contributions}
+	return p.ApplyEvents(ctx, []Event{event})
+}
+
+func (p *Projector) ApplyEvents(ctx context.Context, events []Event) error {
+	if len(events) == 0 {
 		return nil
 	}
-	if len(contributions) == 0 {
+	events = compactEvents(events)
-		contributions = nil
+	if len(events) == 0 {
+		return nil
 	}
 	if p.cfg.CDC.DryRun {
-		slog.Info("dashboard cdc dry run", "eventKey", eventKey, "source", source, "user", user != nil, "contributions", len(contributions))
+		slog.Info("dashboard cdc dry run batch", "events", len(events))
 		return nil
 	}
-	if err := p.store.WithTx(ctx, func(tx *sql.Tx) error {
+
-		applied, err := p.store.MarkEventApplied(ctx, tx, eventKey, source)
+	var appliedContributions []Contribution
+	err := p.withDashboardLock(ctx, func() error {
+		return p.store.WithTx(ctx, func(tx *sql.Tx) error {
+			batchContributions := make([]Contribution, 0, len(events))
+			for _, event := range events {
+				applied, err := p.store.MarkEventApplied(ctx, tx, event.EventKey, event.Source)
 				if err != nil {
 					return err
 				}
 				if !applied {
-			return nil
+					continue
 				}
-		if user != nil {
+				if event.User != nil {
-			if err := p.store.UpsertUserCountry(ctx, tx, *user); err != nil {
+					if err := p.store.UpsertUserCountry(ctx, tx, *event.User); err != nil {
 						return err
 					}
 				}
-		for _, contribution := range contributions {
+				for _, contribution := range event.Contributions {
 					if contribution.Empty() {
 						continue
 					}
 					if contribution.DeltaSign == 0 {
 						contribution.DeltaSign = 1
 					}
-			if err := p.store.ApplyDailyContribution(ctx, tx, contribution, p.cfg.Dashboard.StorageTimezone); err != nil {
+					batchContributions = append(batchContributions, contribution)
+				}
+			}
+			if err := p.store.ApplyContributionBatch(ctx, tx, batchContributions,
+				p.cfg.Dashboard.StorageTimezone, p.cfg.Dashboard.StatTimezones); err != nil {
 				return err
 			}
-			for statTimezone := range p.statZones {
+			appliedContributions = batchContributions
-				if err := p.store.ApplyPeriodContribution(ctx, tx, contribution, statTimezone); err != nil {
-					return err
-				}
-				if err := p.store.ApplyGamePeriodContribution(ctx, tx, contribution, statTimezone); err != nil {
-					return err
-				}
-			}
-		}
 			return nil
-	}); err != nil {
+		})
+	})
+	if err != nil {
 		return err
 	}
-	return p.writeRealtimeRedis(ctx, contributions)
+	return p.writeRealtimeRedis(ctx, appliedContributions)
+}
+
+func (p *Projector) withDashboardLock(ctx context.Context, fn func() error) error {
+	for {
+		err := p.store.WithDashboardLock(ctx, p.cfg.CDC.BackfillLockWait, fn)
+		if !errors.Is(err, ErrDashboardBackfillLocked) {
+			return err
+		}
+		if p.metrics != nil {
+			p.metrics.RecordLockWait()
+		}
+		slog.Warn("dashboard cdc paused because backfill lock is held")
+		timer := time.NewTimer(p.cfg.CDC.FlushInterval)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return ctx.Err()
+		case <-timer.C:
+		}
+	}
+}
+
+func compactEvents(events []Event) []Event {
+	out := make([]Event, 0, len(events))
+	for _, event := range events {
+		if event.User == nil && len(event.Contributions) == 0 {
+			continue
+		}
+		out = append(out, event)
+	}
+	return out
 }
 
 func mustLocation(name string) *time.Location {

internal/dashboard/store.go

@@ -3,15 +3,18 @@ package dashboard
 import (
 	"context"
 	"database/sql"
+	"time"
 
 	"github.com/go-mysql-org/go-mysql/mysql"
 )
 
 type Store interface {
 	WithTx(ctx context.Context, fn func(*sql.Tx) error) error
+	WithDashboardLock(ctx context.Context, timeout time.Duration, fn func() error) error
 	MarkEventApplied(ctx context.Context, tx *sql.Tx, eventKey string, source string) (bool, error)
 	SaveCheckpoint(ctx context.Context, pos mysql.Position) error
 	LoadCheckpoint(ctx context.Context) (mysql.Position, bool, error)
+	Cleanup(ctx context.Context) (int64, error)
 
 	UpsertUserCountry(ctx context.Context, tx *sql.Tx, user UserCountry) error
 	UserCountry(ctx context.Context, userID int64) (UserCountry, bool, error)
@@ -20,4 +23,5 @@ type Store interface {
 	ApplyDailyContribution(ctx context.Context, tx *sql.Tx, c Contribution, storageZone string) error
 	ApplyPeriodContribution(ctx context.Context, tx *sql.Tx, c Contribution, statTimezone string) error
 	ApplyGamePeriodContribution(ctx context.Context, tx *sql.Tx, c Contribution, statTimezone string) error
+	ApplyContributionBatch(ctx context.Context, tx *sql.Tx, contributions []Contribution, storageTimezone string, statTimezones []string) error
 }

internal/dashboard/types.go

@@ -1,11 +1,14 @@
 package dashboard
 
 import (
+	"errors"
 	"time"
 
 	"github.com/shopspring/decimal"
 )
 
+var ErrDashboardBackfillLocked = errors.New("dashboard backfill lock is held")
+
 const (
 	PeriodDay   = "DAY"
 	PeriodWeek  = "WEEK"
@@ -58,6 +61,15 @@ type Contribution struct {
 	GamePayoutBy decimal.Decimal
 	GameProfit   decimal.Decimal
 	GameOrder    int64
+
+	RechargeDetail *RechargeDetail
+}
+
+type Event struct {
+	EventKey      string
+	Source        string
+	User          *UserCountry
+	Contributions []Contribution
 }
 
 func (c Contribution) Empty() bool {
@@ -78,7 +90,21 @@ func (c Contribution) Empty() bool {
 		c.GameConsume.IsZero() &&
 		c.GamePayoutBy.IsZero() &&
 		c.GameProfit.IsZero() &&
-		c.GameOrder == 0
+		c.GameOrder == 0 &&
+		c.RechargeDetail == nil
+}
+
+type RechargeDetail struct {
+	SourceType    string
+	RecordID      string
+	SourceChannel string
+	SysOrigin     string
+	UserID        int64
+	Country       Country
+	Amount        decimal.Decimal
+	CoinQuantity  decimal.Decimal
+	Remark        string
+	EventTime     time.Time
 }
 
 type PeriodWindow struct {

internal/health/server.go

@@ -7,17 +7,31 @@ import (
 	"log/slog"
 	"net/http"
 	"time"
+
+	"dashboard-cdc-worker/internal/monitor"
 )
 
 type Server struct {
 	server *http.Server
 }
 
-func New(addr string) *Server {
+func New(addr string, metrics *monitor.Metrics) *Server {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) {
 		w.Header().Set("Content-Type", "application/json")
+		payload := map[string]interface{}{"code": "ok"}
+		if metrics != nil {
+			payload["metrics"] = metrics.Snapshot()
+		}
+		_ = json.NewEncoder(w).Encode(payload)
+	})
+	mux.HandleFunc("/metrics", func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		if metrics == nil {
 			_ = json.NewEncoder(w).Encode(map[string]string{"code": "ok"})
+			return
+		}
+		_ = json.NewEncoder(w).Encode(metrics.Snapshot())
 	})
 	return &Server{server: &http.Server{Addr: addr, Handler: mux, ReadHeaderTimeout: 5 * time.Second}}
 }

internal/monitor/metrics.go

@@ -0,0 +1,173 @@
+package monitor
+
+import (
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/go-mysql-org/go-mysql/mysql"
+)
+
+type Metrics struct {
+	mu sync.RWMutex
+
+	startedAt time.Time
+
+	masterPosition     mysql.Position
+	checkpointPosition mysql.Position
+	lastSyncedPosition mysql.Position
+	startPosition      mysql.Position
+
+	lastFlushAt       time.Time
+	lastFlushEvents   int
+	totalFlushes      uint64
+	totalEvents       uint64
+	totalRetries      uint64
+	totalLockWaits    uint64
+	totalCleanupRows  uint64
+	lastCleanupAt     time.Time
+	lastCleanupRows   int64
+	lastError         string
+	lastErrorAt       time.Time
+	lastBatchDuration time.Duration
+}
+
+func New() *Metrics {
+	return &Metrics{startedAt: time.Now()}
+}
+
+func (m *Metrics) SetStartPosition(pos mysql.Position) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.startPosition = pos
+}
+
+func (m *Metrics) SetMasterPosition(pos mysql.Position) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.masterPosition = pos
+}
+
+func (m *Metrics) SetCheckpointPosition(pos mysql.Position) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.checkpointPosition = pos
+}
+
+func (m *Metrics) SetLastSyncedPosition(pos mysql.Position) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.lastSyncedPosition = pos
+}
+
+func (m *Metrics) RecordBatch(events int, duration time.Duration) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.lastFlushAt = time.Now()
+	m.lastFlushEvents = events
+	m.totalFlushes++
+	m.totalEvents += uint64(events)
+	m.lastBatchDuration = duration
+}
+
+func (m *Metrics) RecordRetry() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.totalRetries++
+}
+
+func (m *Metrics) RecordLockWait() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.totalLockWaits++
+}
+
+func (m *Metrics) RecordCleanup(rows int64) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.lastCleanupAt = time.Now()
+	m.lastCleanupRows = rows
+	if rows > 0 {
+		m.totalCleanupRows += uint64(rows)
+	}
+}
+
+func (m *Metrics) RecordError(err error) {
+	if err == nil {
+		return
+	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.lastError = err.Error()
+	m.lastErrorAt = time.Now()
+}
+
+func (m *Metrics) Snapshot() map[string]interface{} {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	return map[string]interface{}{
+		"started_at":             formatTime(m.startedAt),
+		"start_position":         positionMap(m.startPosition),
+		"master_position":        positionMap(m.masterPosition),
+		"checkpoint_position":    positionMap(m.checkpointPosition),
+		"last_synced_position":   positionMap(m.lastSyncedPosition),
+		"binlog_file_gap":        fileGap(m.masterPosition, m.checkpointPosition),
+		"binlog_position_gap":    positionGap(m.masterPosition, m.checkpointPosition),
+		"last_flush_at":          formatTime(m.lastFlushAt),
+		"last_flush_events":      m.lastFlushEvents,
+		"last_batch_duration_ms": m.lastBatchDuration.Milliseconds(),
+		"total_flushes":          m.totalFlushes,
+		"total_events":           m.totalEvents,
+		"total_retries":          m.totalRetries,
+		"total_lock_waits":       m.totalLockWaits,
+		"last_cleanup_at":        formatTime(m.lastCleanupAt),
+		"last_cleanup_rows":      m.lastCleanupRows,
+		"total_cleanup_rows":     m.totalCleanupRows,
+		"last_error":             m.lastError,
+		"last_error_at":          formatTime(m.lastErrorAt),
+	}
+}
+
+func positionMap(pos mysql.Position) map[string]interface{} {
+	return map[string]interface{}{
+		"file": pos.Name,
+		"pos":  pos.Pos,
+	}
+}
+
+func positionGap(master mysql.Position, checkpoint mysql.Position) interface{} {
+	if master.Name == "" || checkpoint.Name == "" || master.Name != checkpoint.Name || master.Pos < checkpoint.Pos {
+		return nil
+	}
+	return master.Pos - checkpoint.Pos
+}
+
+func fileGap(master mysql.Position, checkpoint mysql.Position) interface{} {
+	masterSeq, ok := binlogSequence(master.Name)
+	if !ok {
+		return nil
+	}
+	checkpointSeq, ok := binlogSequence(checkpoint.Name)
+	if !ok {
+		return nil
+	}
+	return masterSeq - checkpointSeq
+}
+
+func binlogSequence(file string) (int64, bool) {
+	index := strings.LastIndex(file, ".")
+	if index < 0 || index == len(file)-1 {
+		return 0, false
+	}
+	value, err := strconv.ParseInt(file[index+1:], 10, 64)
+	return value, err == nil
+}
+
+func formatTime(value time.Time) string {
+	if value.IsZero() {
+		return ""
+	}
+	return value.Format(time.RFC3339)
+}

internal/storage/batch.go

@@ -0,0 +1,380 @@
+package storage
+
+import (
+	"context"
+	"database/sql"
+	"strings"
+	"time"
+
+	"github.com/shopspring/decimal"
+
+	"dashboard-cdc-worker/internal/dashboard"
+)
+
+const bulkInsertChunkSize = 500
+
+type amountTotals struct {
+	newUserRecharge    decimal.Decimal
+	officialRecharge   decimal.Decimal
+	mifapayRecharge    decimal.Decimal
+	googleRecharge     decimal.Decimal
+	dealerRecharge     decimal.Decimal
+	salaryExchange     decimal.Decimal
+	luckyGiftTotalFlow decimal.Decimal
+	luckyGiftPayout    decimal.Decimal
+	gameTotalFlow      decimal.Decimal
+	gamePayout         decimal.Decimal
+}
+
+type gameAmountTotals struct {
+	consume decimal.Decimal
+	payout  decimal.Decimal
+	profit  decimal.Decimal
+	orders  int64
+}
+
+type dailyAmountKey struct {
+	day         time.Time
+	sysOrigin   string
+	countryCode string
+	countryName string
+}
+
+type periodAmountKey struct {
+	periodType   string
+	periodKey    string
+	statTimezone string
+	periodName   string
+	startDate    time.Time
+	endDate      time.Time
+	sysOrigin    string
+	countryCode  string
+	countryName  string
+}
+
+type gameAmountKey struct {
+	periodType   string
+	periodKey    string
+	statTimezone string
+	periodName   string
+	startDate    time.Time
+	endDate      time.Time
+	sysOrigin    string
+	countryCode  string
+	countryName  string
+	gameProvider string
+	gameID       string
+	gameName     string
+}
+
+type dailyUserKey struct {
+	day         time.Time
+	sysOrigin   string
+	countryCode string
+	countryName string
+	metricType  string
+}
+
+type periodUserKey struct {
+	periodType   string
+	periodKey    string
+	statTimezone string
+	periodName   string
+	startDate    time.Time
+	endDate      time.Time
+	sysOrigin    string
+	countryCode  string
+	countryName  string
+	metricType   string
+}
+
+type gameUserKey struct {
+	periodType   string
+	periodKey    string
+	statTimezone string
+	periodName   string
+	startDate    time.Time
+	endDate      time.Time
+	sysOrigin    string
+	countryCode  string
+	countryName  string
+	gameProvider string
+	gameID       string
+	gameName     string
+}
+
+func (r *MySQLRepository) ApplyContributionBatch(ctx context.Context, tx *sql.Tx,
+	contributions []dashboard.Contribution, storageTimezone string, statTimezones []string) error {
+	if len(contributions) == 0 {
+		return nil
+	}
+	storageZone, err := time.LoadLocation(storageTimezone)
+	if err != nil {
+		storageZone = time.UTC
+	}
+	statZones := make(map[string]*time.Location, len(statTimezones))
+	for _, statTimezone := range statTimezones {
+		zone, err := time.LoadLocation(statTimezone)
+		if err != nil {
+			zone = time.UTC
+		}
+		statZones[statTimezone] = zone
+	}
+
+	dailyAmounts := make(map[dailyAmountKey]amountTotals)
+	periodAmounts := make(map[periodAmountKey]amountTotals)
+	gameAmounts := make(map[gameAmountKey]gameAmountTotals)
+	dailyUsers := make(map[dailyUserKey]map[int64]struct{})
+	periodUsers := make(map[periodUserKey]map[int64]struct{})
+	gameUsers := make(map[gameUserKey]map[int64]struct{})
+	var negative []dashboard.Contribution
+
+	for _, c := range contributions {
+		if err := r.applyRechargeDetail(ctx, tx, c); err != nil {
+			return err
+		}
+		if c.EventTime.IsZero() {
+			continue
+		}
+		if c.DeltaSign < 0 {
+			negative = append(negative, c)
+			continue
+		}
+		if c.DeltaSign == 0 {
+			c.DeltaSign = 1
+		}
+		day := dashboard.DailyDate(c.EventTime, storageZone)
+		newUserRecharge := decimal.Zero
+		if dashboard.SamePeriod(c.UserCreatedAt, c.EventTime, dashboard.PeriodWindow{Type: dashboard.PeriodDay}, storageZone) {
+			newUserRecharge = c.NewUserRechargeCandidate
+		}
+		dailyKey := dailyAmountKey{day: day, sysOrigin: c.SysOrigin, countryCode: c.Country.Code, countryName: c.Country.Name}
+		totals := dailyAmounts[dailyKey]
+		totals.add(c, newUserRecharge)
+		dailyAmounts[dailyKey] = totals
+		addDailyUserGroups(dailyUsers, c, day)
+
+		for statTimezone, zone := range statZones {
+			for _, window := range dashboard.PeriodWindows(c.EventTime, zone) {
+				startDate, endDate := windowDates(window)
+				newUserRecharge = decimal.Zero
+				if dashboard.SamePeriod(c.UserCreatedAt, c.EventTime, window, zone) {
+					newUserRecharge = c.NewUserRechargeCandidate
+				}
+				periodKey := periodAmountKey{
+					periodType: window.Type, periodKey: window.Key, statTimezone: statTimezone,
+					periodName: window.Name, startDate: startDate, endDate: endDate,
+					sysOrigin: c.SysOrigin, countryCode: c.Country.Code, countryName: c.Country.Name,
+				}
+				periodTotal := periodAmounts[periodKey]
+				periodTotal.add(c, newUserRecharge)
+				periodAmounts[periodKey] = periodTotal
+				addPeriodUserGroups(periodUsers, c, window, statTimezone)
+
+				if c.GameProvider != "" && c.GameID != "" {
+					gameKey := gameAmountKey{
+						periodType: window.Type, periodKey: window.Key, statTimezone: statTimezone,
+						periodName: window.Name, startDate: startDate, endDate: endDate,
+						sysOrigin: c.SysOrigin, countryCode: c.Country.Code, countryName: c.Country.Name,
+						gameProvider: c.GameProvider, gameID: c.GameID, gameName: c.GameName,
+					}
+					gameTotal := gameAmounts[gameKey]
+					gameTotal.add(c)
+					gameAmounts[gameKey] = gameTotal
+					if c.GameUser {
+						addGameUserGroup(gameUsers, c, window, statTimezone)
+					}
+				}
+			}
+		}
+	}
+
+	for key, totals := range dailyAmounts {
+		if totals.empty() {
+			continue
+		}
+		if err := r.upsertDailyAmountTotals(ctx, tx, key, totals); err != nil {
+			return err
+		}
+	}
+	for key, totals := range periodAmounts {
+		if totals.empty() {
+			continue
+		}
+		if err := r.upsertPeriodAmountTotals(ctx, tx, key, totals); err != nil {
+			return err
+		}
+	}
+	for key, totals := range gameAmounts {
+		if totals.empty() {
+			continue
+		}
+		if err := r.upsertGameAmountTotals(ctx, tx, key, totals); err != nil {
+			return err
+		}
+	}
+	for key, users := range dailyUsers {
+		if err := r.applyDailyUserGroup(ctx, tx, key, users); err != nil {
+			return err
+		}
+	}
+	for key, users := range periodUsers {
+		if err := r.applyPeriodUserGroup(ctx, tx, key, users); err != nil {
+			return err
+		}
+	}
+	for key, users := range gameUsers {
+		if err := r.applyGameUserGroup(ctx, tx, key, users); err != nil {
+			return err
+		}
+	}
+	for _, c := range negative {
+		if err := r.ApplyDailyContribution(ctx, tx, c, storageTimezone); err != nil {
+			return err
+		}
+		for _, statTimezone := range statTimezones {
+			if err := r.ApplyPeriodContribution(ctx, tx, c, statTimezone); err != nil {
+				return err
+			}
+			if err := r.ApplyGamePeriodContribution(ctx, tx, c, statTimezone); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (t *amountTotals) add(c dashboard.Contribution, newUserRecharge decimal.Decimal) {
+	sign := decimal.NewFromInt(int64(c.DeltaSign))
+	t.newUserRecharge = t.newUserRecharge.Add(newUserRecharge.Mul(sign))
+	t.officialRecharge = t.officialRecharge.Add(c.OfficialRecharge.Mul(sign))
+	t.mifapayRecharge = t.mifapayRecharge.Add(c.MifapayRecharge.Mul(sign))
+	t.googleRecharge = t.googleRecharge.Add(c.GoogleRecharge.Mul(sign))
+	t.dealerRecharge = t.dealerRecharge.Add(c.DealerRecharge.Mul(sign))
+	t.salaryExchange = t.salaryExchange.Add(c.SalaryExchange.Mul(sign))
+	t.luckyGiftTotalFlow = t.luckyGiftTotalFlow.Add(c.LuckyGiftTotalFlow.Mul(sign))
+	t.luckyGiftPayout = t.luckyGiftPayout.Add(c.LuckyGiftPayout.Mul(sign))
+	t.gameTotalFlow = t.gameTotalFlow.Add(c.GameTotalFlow.Mul(sign))
+	t.gamePayout = t.gamePayout.Add(c.GamePayout.Mul(sign))
+}
+
+func (t amountTotals) empty() bool {
+	return t.newUserRecharge.IsZero() && t.officialRecharge.IsZero() && t.mifapayRecharge.IsZero() &&
+		t.googleRecharge.IsZero() && t.dealerRecharge.IsZero() && t.salaryExchange.IsZero() &&
+		t.luckyGiftTotalFlow.IsZero() && t.luckyGiftPayout.IsZero() &&
+		t.gameTotalFlow.IsZero() && t.gamePayout.IsZero()
+}
+
+func (t *gameAmountTotals) add(c dashboard.Contribution) {
+	sign := decimal.NewFromInt(int64(c.DeltaSign))
+	t.consume = t.consume.Add(c.GameConsume.Mul(sign))
+	t.payout = t.payout.Add(c.GamePayoutBy.Mul(sign))
+	t.profit = t.profit.Add(c.GameProfit.Mul(sign))
+	t.orders += c.GameOrder * int64(c.DeltaSign)
+}
+
+func (t gameAmountTotals) empty() bool {
+	return t.consume.IsZero() && t.payout.IsZero() && t.profit.IsZero() && t.orders == 0
+}
+
+func addDailyUserGroups(groups map[dailyUserKey]map[int64]struct{}, c dashboard.Contribution, day time.Time) {
+	if c.UserID == 0 {
+		return
+	}
+	if c.CountryNewUser > 0 {
+		addDailyUser(groups, c, day, dashboard.MetricCountryNewUser)
+	}
+	if c.LuckyGiftUser {
+		addDailyUser(groups, c, day, dashboard.MetricLuckyGiftUser)
+	}
+	if c.GameUser {
+		addDailyUser(groups, c, day, dashboard.MetricGameUser)
+	}
+}
+
+func addPeriodUserGroups(groups map[periodUserKey]map[int64]struct{}, c dashboard.Contribution, window dashboard.PeriodWindow, statTimezone string) {
+	if c.UserID == 0 {
+		return
+	}
+	if c.CountryNewUser > 0 {
+		addPeriodUser(groups, c, window, statTimezone, dashboard.MetricCountryNewUser)
+	}
+	if c.LuckyGiftUser {
+		addPeriodUser(groups, c, window, statTimezone, dashboard.MetricLuckyGiftUser)
+	}
+	if c.GameUser {
+		addPeriodUser(groups, c, window, statTimezone, dashboard.MetricGameUser)
+	}
+}
+
+func addDailyUser(groups map[dailyUserKey]map[int64]struct{}, c dashboard.Contribution, day time.Time, metricType string) {
+	key := dailyUserKey{day: day, sysOrigin: c.SysOrigin, countryCode: c.Country.Code, countryName: c.Country.Name, metricType: metricType}
+	addUserID(groups, key, c.UserID)
+}
+
+func addPeriodUser(groups map[periodUserKey]map[int64]struct{}, c dashboard.Contribution, window dashboard.PeriodWindow, statTimezone string, metricType string) {
+	startDate, endDate := windowDates(window)
+	key := periodUserKey{
+		periodType: window.Type, periodKey: window.Key, statTimezone: statTimezone, periodName: window.Name,
+		startDate: startDate, endDate: endDate, sysOrigin: c.SysOrigin, countryCode: c.Country.Code,
+		countryName: c.Country.Name, metricType: metricType,
+	}
+	addUserID(groups, key, c.UserID)
+}
+
+func addGameUserGroup(groups map[gameUserKey]map[int64]struct{}, c dashboard.Contribution, window dashboard.PeriodWindow, statTimezone string) {
+	if c.UserID == 0 {
+		return
+	}
+	startDate, endDate := windowDates(window)
+	key := gameUserKey{
+		periodType: window.Type, periodKey: window.Key, statTimezone: statTimezone, periodName: window.Name,
+		startDate: startDate, endDate: endDate, sysOrigin: c.SysOrigin, countryCode: c.Country.Code,
+		countryName: c.Country.Name, gameProvider: c.GameProvider, gameID: c.GameID, gameName: c.GameName,
+	}
+	addUserID(groups, key, c.UserID)
+}
+
+func addUserID[K comparable](groups map[K]map[int64]struct{}, key K, userID int64) {
+	users := groups[key]
+	if users == nil {
+		users = make(map[int64]struct{})
+		groups[key] = users
+	}
+	users[userID] = struct{}{}
+}
+
+func windowDates(window dashboard.PeriodWindow) (time.Time, time.Time) {
+	var startDate, endDate time.Time
+	if window.StartDate != nil {
+		startDate = *window.StartDate
+	}
+	if window.EndDate != nil {
+		endDate = *window.EndDate
+	}
+	return startDate, endDate
+}
+
+func ptrOrNil(value time.Time) interface{} {
+	if value.IsZero() {
+		return nil
+	}
+	return value
+}
+
+func userIDs(users map[int64]struct{}) []int64 {
+	out := make([]int64, 0, len(users))
+	for userID := range users {
+		out = append(out, userID)
+	}
+	return out
+}
+
+func placeholders(count, columns int) string {
+	groups := make([]string, count)
+	item := "(" + strings.TrimRight(strings.Repeat("?,", columns), ",") + ")"
+	for i := range groups {
+		groups[i] = item
+	}
+	return strings.Join(groups, ",")
+}

internal/storage/batch_sql.go

@@ -0,0 +1,231 @@
+package storage
+
+import (
+	"context"
+	"database/sql"
+	"time"
+)
+
+func (r *MySQLRepository) upsertDailyAmountTotals(ctx context.Context, tx *sql.Tx, key dailyAmountKey, totals amountTotals) error {
+	_, err := tx.ExecContext(ctx, `
+INSERT INTO country_dashboard_daily_metric (
+  stat_date, date_number, sys_origin, country_code, country_name,
+  new_user_recharge, official_recharge, mifapay_recharge, google_recharge,
+  dealer_recharge, salary_exchange, lucky_gift_total_flow, lucky_gift_payout,
+  game_total_flow, game_payout, refreshed_at
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
+ON DUPLICATE KEY UPDATE
+  country_name = VALUES(country_name),
+  new_user_recharge = new_user_recharge + VALUES(new_user_recharge),
+  official_recharge = official_recharge + VALUES(official_recharge),
+  mifapay_recharge = mifapay_recharge + VALUES(mifapay_recharge),
+  google_recharge = google_recharge + VALUES(google_recharge),
+  dealer_recharge = dealer_recharge + VALUES(dealer_recharge),
+  salary_exchange = salary_exchange + VALUES(salary_exchange),
+  lucky_gift_total_flow = lucky_gift_total_flow + VALUES(lucky_gift_total_flow),
+  lucky_gift_payout = lucky_gift_payout + VALUES(lucky_gift_payout),
+  game_total_flow = game_total_flow + VALUES(game_total_flow),
+  game_payout = game_payout + VALUES(game_payout),
+  refreshed_at = NOW()
+`, key.day, dateNumber(key.day), key.sysOrigin, key.countryCode, key.countryName,
+		totals.newUserRecharge, totals.officialRecharge, totals.mifapayRecharge, totals.googleRecharge,
+		totals.dealerRecharge, totals.salaryExchange, totals.luckyGiftTotalFlow, totals.luckyGiftPayout,
+		totals.gameTotalFlow, totals.gamePayout)
+	return err
+}
+
+func (r *MySQLRepository) upsertPeriodAmountTotals(ctx context.Context, tx *sql.Tx, key periodAmountKey, totals amountTotals) error {
+	_, 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, new_user_recharge, official_recharge, mifapay_recharge,
+  google_recharge, dealer_recharge, salary_exchange, lucky_gift_total_flow, lucky_gift_payout,
+  game_total_flow, game_payout, refreshed_at
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
+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),
+  new_user_recharge = new_user_recharge + VALUES(new_user_recharge),
+  official_recharge = official_recharge + VALUES(official_recharge),
+  mifapay_recharge = mifapay_recharge + VALUES(mifapay_recharge),
+  google_recharge = google_recharge + VALUES(google_recharge),
+  dealer_recharge = dealer_recharge + VALUES(dealer_recharge),
+  salary_exchange = salary_exchange + VALUES(salary_exchange),
+  lucky_gift_total_flow = lucky_gift_total_flow + VALUES(lucky_gift_total_flow),
+  lucky_gift_payout = lucky_gift_payout + VALUES(lucky_gift_payout),
+  game_total_flow = game_total_flow + VALUES(game_total_flow),
+  game_payout = game_payout + VALUES(game_payout),
+  refreshed_at = NOW()
+`, key.periodType, key.periodKey, key.statTimezone, key.periodName, ptrOrNil(key.startDate), ptrOrNil(key.endDate),
+		key.sysOrigin, key.countryCode, key.countryName, totals.newUserRecharge, totals.officialRecharge,
+		totals.mifapayRecharge, totals.googleRecharge, totals.dealerRecharge, totals.salaryExchange,
+		totals.luckyGiftTotalFlow, totals.luckyGiftPayout, totals.gameTotalFlow, totals.gamePayout)
+	return err
+}
+
+func (r *MySQLRepository) upsertGameAmountTotals(ctx context.Context, tx *sql.Tx, key gameAmountKey, totals gameAmountTotals) error {
+	_, err := tx.ExecContext(ctx, `
+INSERT INTO country_dashboard_game_period_metric (
+  period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date,
+  sys_origin, country_code, country_name, game_provider, game_id, game_name,
+  consume_amount, payout_amount, profit_amount, order_count, refreshed_at
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
+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),
+  game_name = VALUES(game_name),
+  consume_amount = consume_amount + VALUES(consume_amount),
+  payout_amount = payout_amount + VALUES(payout_amount),
+  profit_amount = profit_amount + VALUES(profit_amount),
+  order_count = order_count + VALUES(order_count),
+  refreshed_at = NOW()
+`, key.periodType, key.periodKey, key.statTimezone, key.periodName, ptrOrNil(key.startDate), ptrOrNil(key.endDate),
+		key.sysOrigin, key.countryCode, key.countryName, key.gameProvider, key.gameID, key.gameName,
+		totals.consume, totals.payout, totals.profit, totals.orders)
+	return err
+}
+
+func (r *MySQLRepository) applyDailyUserGroup(ctx context.Context, tx *sql.Tx, key dailyUserKey, users map[int64]struct{}) error {
+	ids := userIDs(users)
+	inserted, err := bulkInsertDailyUsers(ctx, tx, key, ids)
+	if err != nil || inserted == 0 {
+		return err
+	}
+	column := dailyMetricColumn(key.metricType)
+	if column == "" {
+		return nil
+	}
+	_, err = tx.ExecContext(ctx, `
+INSERT INTO country_dashboard_daily_metric (
+  stat_date, date_number, sys_origin, country_code, country_name, refreshed_at
+) VALUES (?, ?, ?, ?, ?, NOW())
+ON DUPLICATE KEY UPDATE `+column+` = `+column+` + ?, country_name = VALUES(country_name), refreshed_at = NOW()
+`, key.day, dateNumber(key.day), key.sysOrigin, key.countryCode, key.countryName, inserted)
+	return err
+}
+
+func (r *MySQLRepository) applyPeriodUserGroup(ctx context.Context, tx *sql.Tx, key periodUserKey, users map[int64]struct{}) error {
+	ids := userIDs(users)
+	inserted, err := bulkInsertPeriodUsers(ctx, tx, key, ids)
+	if err != nil || inserted == 0 {
+		return err
+	}
+	column := periodMetricColumn(key.metricType)
+	if column == "" {
+		return nil
+	}
+	_, 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, refreshed_at
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
+ON DUPLICATE KEY UPDATE `+column+` = `+column+` + ?, country_name = VALUES(country_name), refreshed_at = NOW()
+`, key.periodType, key.periodKey, key.statTimezone, key.periodName, ptrOrNil(key.startDate), ptrOrNil(key.endDate),
+		key.sysOrigin, key.countryCode, key.countryName, inserted)
+	return err
+}
+
+func (r *MySQLRepository) applyGameUserGroup(ctx context.Context, tx *sql.Tx, key gameUserKey, users map[int64]struct{}) error {
+	ids := userIDs(users)
+	inserted, err := bulkInsertGameUsers(ctx, tx, key, ids)
+	if err != nil || inserted == 0 {
+		return err
+	}
+	_, err = tx.ExecContext(ctx, `
+INSERT INTO country_dashboard_game_period_metric (
+  period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date,
+  sys_origin, country_code, country_name, game_provider, game_id, game_name, user_count, refreshed_at
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
+ON DUPLICATE KEY UPDATE user_count = user_count + VALUES(user_count), country_name = VALUES(country_name),
+  game_name = VALUES(game_name), refreshed_at = NOW()
+`, key.periodType, key.periodKey, key.statTimezone, key.periodName, ptrOrNil(key.startDate), ptrOrNil(key.endDate),
+		key.sysOrigin, key.countryCode, key.countryName, key.gameProvider, key.gameID, key.gameName, inserted)
+	return err
+}
+
+func bulkInsertDailyUsers(ctx context.Context, tx *sql.Tx, key dailyUserKey, ids []int64) (int64, error) {
+	var inserted int64
+	for start := 0; start < len(ids); start += bulkInsertChunkSize {
+		end := min(start+bulkInsertChunkSize, len(ids))
+		args := make([]interface{}, 0, (end-start)*6)
+		for _, userID := range ids[start:end] {
+			args = append(args, key.day, key.sysOrigin, key.countryCode, key.countryName, userID, key.metricType)
+		}
+		result, err := tx.ExecContext(ctx, `
+INSERT IGNORE INTO country_dashboard_daily_user_metric (
+  stat_date, sys_origin, country_code, country_name, user_id, metric_type
+) VALUES `+placeholders(end-start, 6), args...)
+		if err != nil {
+			return 0, err
+		}
+		affected, err := result.RowsAffected()
+		if err != nil {
+			return 0, err
+		}
+		inserted += affected
+	}
+	return inserted, nil
+}
+
+func bulkInsertPeriodUsers(ctx context.Context, tx *sql.Tx, key periodUserKey, ids []int64) (int64, error) {
+	var inserted int64
+	for start := 0; start < len(ids); start += bulkInsertChunkSize {
+		end := min(start+bulkInsertChunkSize, len(ids))
+		args := make([]interface{}, 0, (end-start)*11)
+		for _, userID := range ids[start:end] {
+			args = append(args, key.periodType, key.periodKey, key.statTimezone, key.periodName,
+				ptrOrNil(key.startDate), ptrOrNil(key.endDate), key.sysOrigin, key.countryCode,
+				key.countryName, userID, key.metricType)
+		}
+		result, 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
+) VALUES `+placeholders(end-start, 11), args...)
+		if err != nil {
+			return 0, err
+		}
+		affected, err := result.RowsAffected()
+		if err != nil {
+			return 0, err
+		}
+		inserted += affected
+	}
+	return inserted, nil
+}
+
+func bulkInsertGameUsers(ctx context.Context, tx *sql.Tx, key gameUserKey, ids []int64) (int64, error) {
+	var inserted int64
+	for start := 0; start < len(ids); start += bulkInsertChunkSize {
+		end := min(start+bulkInsertChunkSize, len(ids))
+		args := make([]interface{}, 0, (end-start)*12)
+		for _, userID := range ids[start:end] {
+			args = append(args, key.periodType, key.periodKey, key.statTimezone, key.periodName,
+				ptrOrNil(key.startDate), ptrOrNil(key.endDate), key.sysOrigin, key.countryCode,
+				key.countryName, key.gameProvider, key.gameID, userID)
+		}
+		result, err := tx.ExecContext(ctx, `
+INSERT IGNORE INTO country_dashboard_game_period_user_metric (
+  period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date,
+  sys_origin, country_code, country_name, game_provider, game_id, user_id
+) VALUES `+placeholders(end-start, 12), args...)
+		if err != nil {
+			return 0, err
+		}
+		affected, err := result.RowsAffected()
+		if err != nil {
+			return 0, err
+		}
+		inserted += affected
+	}
+	return inserted, nil
+}
+
+func dateNumber(day time.Time) int {
+	year, month, date := day.Date()
+	return year*10000 + int(month)*100 + date
+}

internal/storage/cache.go

@@ -0,0 +1,58 @@
+package storage
+
+import (
+	"sync"
+	"time"
+)
+
+type ttlCache[K comparable, V any] struct {
+	mu      sync.RWMutex
+	values  map[K]ttlCacheEntry[V]
+	ttl     time.Duration
+	maxSize int
+}
+
+type ttlCacheEntry[V any] struct {
+	value     V
+	expiresAt time.Time
+}
+
+func newTTLCache[K comparable, V any](ttl time.Duration, maxSize int) *ttlCache[K, V] {
+	return &ttlCache[K, V]{
+		values:  make(map[K]ttlCacheEntry[V]),
+		ttl:     ttl,
+		maxSize: maxSize,
+	}
+}
+
+func (c *ttlCache[K, V]) Get(key K) (V, bool) {
+	c.mu.RLock()
+	entry, ok := c.values[key]
+	c.mu.RUnlock()
+	var zero V
+	if !ok {
+		return zero, false
+	}
+	if time.Now().After(entry.expiresAt) {
+		c.mu.Lock()
+		delete(c.values, key)
+		c.mu.Unlock()
+		return zero, false
+	}
+	return entry.value, true
+}
+
+func (c *ttlCache[K, V]) Set(key K, value V) {
+	if c.ttl <= 0 {
+		return
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.maxSize > 0 && len(c.values) >= c.maxSize {
+		c.values = make(map[K]ttlCacheEntry[V])
+	}
+	c.values[key] = ttlCacheEntry[V]{
+		value:     value,
+		expiresAt: time.Now().Add(c.ttl),
+	}
+}

internal/storage/cleanup.go

@@ -0,0 +1,119 @@
+package storage
+
+import (
+	"context"
+	"log/slog"
+	"time"
+)
+
+func (r *MySQLRepository) StartCleanup(ctx context.Context, logger *slog.Logger) {
+	if r.cfg.Dashboard.CleanupInterval <= 0 {
+		return
+	}
+	go func() {
+		r.cleanupOnce(ctx, logger)
+		ticker := time.NewTicker(r.cfg.Dashboard.CleanupInterval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				r.cleanupOnce(ctx, logger)
+			}
+		}
+	}()
+}
+
+func (r *MySQLRepository) cleanupOnce(ctx context.Context, logger *slog.Logger) {
+	rows, err := r.Cleanup(ctx)
+	if err != nil {
+		if r.metrics != nil {
+			r.metrics.RecordError(err)
+		}
+		logger.Error("dashboard cleanup failed", "error", err)
+		return
+	}
+	if r.metrics != nil {
+		r.metrics.RecordCleanup(rows)
+	}
+	if rows > 0 {
+		logger.Info("dashboard cleanup finished", "rows", rows)
+	}
+}
+
+func (r *MySQLRepository) Cleanup(ctx context.Context) (int64, error) {
+	var total int64
+	if r.cfg.Dashboard.DedupRetention > 0 {
+		cutoff := time.Now().Add(-r.cfg.Dashboard.DedupRetention)
+		rows, err := r.deleteLimit(ctx, `
+DELETE FROM dashboard_cdc_event_dedup
+WHERE created_at < ?
+LIMIT ?`, cutoff, r.cfg.Dashboard.CleanupBatchSize)
+		if err != nil {
+			return total, err
+		}
+		total += rows
+	}
+
+	if r.cfg.Dashboard.UserMetricDays > 0 {
+		cutoffDate := time.Now().AddDate(0, 0, -r.cfg.Dashboard.UserMetricDays).Format("2006-01-02")
+		deletes := []struct {
+			sql  string
+			args []interface{}
+		}{
+			{
+				sql: `
+DELETE FROM country_dashboard_daily_user_metric
+WHERE stat_date < ?
+LIMIT ?`,
+				args: []interface{}{cutoffDate, r.cfg.Dashboard.CleanupBatchSize},
+			},
+			{
+				sql: `
+DELETE FROM country_dashboard_period_user_metric
+WHERE period_type <> 'ALL'
+  AND period_end_date IS NOT NULL
+  AND period_end_date < ?
+LIMIT ?`,
+				args: []interface{}{cutoffDate, r.cfg.Dashboard.CleanupBatchSize},
+			},
+			{
+				sql: `
+DELETE FROM country_dashboard_game_period_user_metric
+WHERE period_type <> 'ALL'
+  AND period_end_date IS NOT NULL
+  AND period_end_date < ?
+LIMIT ?`,
+				args: []interface{}{cutoffDate, r.cfg.Dashboard.CleanupBatchSize},
+			},
+			{
+				sql: `
+DELETE FROM country_dashboard_recharge_detail
+WHERE create_time < ?
+LIMIT ?`,
+				args: []interface{}{cutoffDate, r.cfg.Dashboard.CleanupBatchSize},
+			},
+		}
+		for _, item := range deletes {
+			rows, err := r.execRows(ctx, item.sql, item.args...)
+			if err != nil {
+				return total, err
+			}
+			total += rows
+		}
+	}
+	return total, nil
+}
+
+func (r *MySQLRepository) deleteLimit(ctx context.Context, query string, args ...interface{}) (int64, error) {
+	return r.execRows(ctx, query, args...)
+}
+
+func (r *MySQLRepository) execRows(ctx context.Context, query string, args ...interface{}) (int64, error) {
+	result, err := r.db.ExecContext(ctx, query, args...)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}

internal/storage/game.go

@@ -8,12 +8,19 @@ import (
 )
 
 func (r *MySQLRepository) ResolveWalletGame(ctx context.Context, sysOrigin string, eventType string, eventID string) (string, string, string, error) {
-	if provider, gameID, gameName, ok, err := r.resolveBaishunByWalletEvent(ctx, sysOrigin, eventID); ok || err != nil {
+	if provider, gameID, fallback, ok := directGameMeta(eventType); ok {
+		gameName, err := r.lookupGameName(ctx, sysOrigin, provider, gameID)
+		if gameName == "" {
+			gameName = fallback
+		}
 		return provider, gameID, gameName, err
 	}
 	if provider, gameID, gameName, ok, err := r.resolveGameOpenByWalletEvent(ctx, sysOrigin, eventType, eventID); ok || err != nil {
 		return provider, gameID, gameName, err
 	}
+	if provider, gameID, gameName, ok, err := r.resolveBaishunByWalletEvent(ctx, sysOrigin, eventID); ok || err != nil {
+		return provider, gameID, gameName, err
+	}
 
 	provider, gameID, fallback := fallbackGameMeta(eventType)
 	gameName, err := r.lookupGameName(ctx, sysOrigin, provider, gameID)
@@ -87,6 +94,10 @@ LIMIT 1
 }
 
 func (r *MySQLRepository) lookupGameName(ctx context.Context, sysOrigin string, provider string, gameID string) (string, error) {
+	cacheKey := strings.Join([]string{sysOrigin, provider, gameID}, "|")
+	if name, ok := r.gameNameCache.Get(cacheKey); ok {
+		return name, nil
+	}
 	var name string
 	if provider == "BAISHUN" {
 		err := r.db.QueryRowContext(ctx, `
@@ -96,6 +107,9 @@ WHERE sys_origin = ? AND profile = 'PROD' AND vendor_game_id = ?
 LIMIT 1
 `, sysOrigin, gameID).Scan(&name)
 		if err == nil || err != sql.ErrNoRows {
+			if err == nil {
+				r.gameNameCache.Set(cacheKey, name)
+			}
 			return name, err
 		}
 	}
@@ -106,6 +120,9 @@ WHERE sys_origin = ? AND profile = 'PROD' AND vendor_game_id = ?
 LIMIT 1
 `, sysOrigin, gameID).Scan(&name)
 	if err == nil || err != sql.ErrNoRows {
+		if err == nil {
+			r.gameNameCache.Set(cacheKey, name)
+		}
 		return name, err
 	}
 	err = r.db.QueryRowContext(ctx, `
@@ -115,11 +132,33 @@ WHERE sys_origin = ? AND game_id = ?
 LIMIT 1
 `, sysOrigin, gameID).Scan(&name)
 	if err == sql.ErrNoRows {
+		r.gameNameCache.Set(cacheKey, "")
 		return "", nil
 	}
+	if err == nil {
+		r.gameNameCache.Set(cacheKey, name)
+	}
 	return name, err
 }
 
+func directGameMeta(eventType string) (string, string, string, bool) {
+	switch {
+	case strings.HasPrefix(eventType, "BAISHUN_GAME_"),
+		strings.HasPrefix(eventType, "GAME_OPEN_"),
+		strings.HasPrefix(eventType, "YOMI_GAME_"),
+		strings.HasPrefix(eventType, "HKYS_GAME_"),
+		strings.HasPrefix(eventType, "HOT_GAME_"),
+		in(eventType, "FRUIT_GAME", "DOUBLE_LAYER_FRUIT", "GAME_FRUIT_BET", "GAME_FRUIT_AWARD",
+			"LUDO_GAME", "LUDO_VICTORY", "LUDO_REFUND", "GAME_LUDO_REFUND",
+			"GAME_BARBECUE", "DOUBLE_LAYER_BARBECUE", "GAME_SLOT_MACHINE", "TURNTABLE_GAME",
+			"GAME_BURST_CRYSTAL", "TEEN_PATTI", "BET_TEEN_PATTI", "LOTTERY_NUMBER_BET", "LOTTERY_REWARD"):
+		provider, gameID, fallback := fallbackGameMeta(eventType)
+		return provider, gameID, fallback, true
+	default:
+		return "", "", "", false
+	}
+}
+
 func fallbackGameMeta(eventType string) (string, string, string) {
 	switch {
 	case strings.HasPrefix(eventType, "BAISHUN_GAME_"):

internal/storage/recharge_detail.go

@@ -0,0 +1,45 @@
+package storage
+
+import (
+	"context"
+	"database/sql"
+
+	"dashboard-cdc-worker/internal/dashboard"
+)
+
+func (r *MySQLRepository) applyRechargeDetail(ctx context.Context, tx *sql.Tx, c dashboard.Contribution) error {
+	detail := c.RechargeDetail
+	if detail == nil || detail.RecordID == "" || detail.SourceType == "" {
+		return nil
+	}
+	if c.DeltaSign < 0 {
+		_, err := tx.ExecContext(ctx, `
+DELETE FROM country_dashboard_recharge_detail
+WHERE source_type = ? AND record_id = ?
+`, detail.SourceType, detail.RecordID)
+		return err
+	}
+	if detail.EventTime.IsZero() {
+		return nil
+	}
+	_, err := tx.ExecContext(ctx, `
+INSERT INTO country_dashboard_recharge_detail (
+  source_type, record_id, source_channel, sys_origin, user_id, country_code, country_name,
+  amount, coin_quantity, remark, create_time, update_time
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
+ON DUPLICATE KEY UPDATE
+  source_channel = VALUES(source_channel),
+  sys_origin = VALUES(sys_origin),
+  user_id = VALUES(user_id),
+  country_code = VALUES(country_code),
+  country_name = VALUES(country_name),
+  amount = VALUES(amount),
+  coin_quantity = VALUES(coin_quantity),
+  remark = VALUES(remark),
+  create_time = VALUES(create_time),
+  update_time = NOW()
+`, detail.SourceType, detail.RecordID, detail.SourceChannel, detail.SysOrigin, detail.UserID,
+		detail.Country.Code, detail.Country.Name, detail.Amount, detail.CoinQuantity,
+		detail.Remark, detail.EventTime)
+	return err
+}

internal/storage/repository.go

@@ -3,34 +3,108 @@ package storage
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"fmt"
+	"math"
+	"time"
 
 	"github.com/go-mysql-org/go-mysql/mysql"
+	driverMysql "github.com/go-sql-driver/mysql"
 
 	"dashboard-cdc-worker/internal/config"
+	"dashboard-cdc-worker/internal/dashboard"
+	"dashboard-cdc-worker/internal/monitor"
 )
 
 const serviceName = "dashboard-cdc-worker"
+const dashboardBackfillLockName = "country_dashboard_backfill_lock"
 
 type MySQLRepository struct {
 	db            *sql.DB
 	cfg           config.Config
+	metrics       *monitor.Metrics
+	userCache     *ttlCache[int64, dashboard.UserCountry]
+	gameNameCache *ttlCache[string, string]
 }
 
-func NewMySQLRepository(db *sql.DB, cfg config.Config) *MySQLRepository {
+func NewMySQLRepository(db *sql.DB, cfg config.Config, metrics *monitor.Metrics) *MySQLRepository {
-	return &MySQLRepository{db: db, cfg: cfg}
+	return &MySQLRepository{
+		db:            db,
+		cfg:           cfg,
+		metrics:       metrics,
+		userCache:     newTTLCache[int64, dashboard.UserCountry](30*time.Minute, 200000),
+		gameNameCache: newTTLCache[string, string](6*time.Hour, 10000),
+	}
 }
 
 func (r *MySQLRepository) WithTx(ctx context.Context, fn func(*sql.Tx) error) error {
+	const maxAttempts = 6
+	var lastErr error
+	for attempt := 1; attempt <= maxAttempts; attempt++ {
 		tx, err := r.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
 		if err != nil {
 			return err
 		}
-	if err := fn(tx); err != nil {
+		err = fn(tx)
+		if err != nil {
 			_ = tx.Rollback()
+		} else {
+			err = tx.Commit()
+		}
+		if err == nil {
+			return nil
+		}
+		lastErr = err
+		if !isRetryableMySQLError(err) || attempt == maxAttempts {
 			return err
 		}
-	return tx.Commit()
+		if r.metrics != nil {
+			r.metrics.RecordRetry()
+		}
+		backoff := time.Duration(math.Pow(2, float64(attempt-1))) * 100 * time.Millisecond
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(backoff):
+		}
+	}
+	return lastErr
+}
+
+func (r *MySQLRepository) WithDashboardLock(ctx context.Context, timeout time.Duration, fn func() error) error {
+	conn, err := r.db.Conn(ctx)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	timeoutSeconds := int(math.Ceil(timeout.Seconds()))
+	if timeoutSeconds < 0 {
+		timeoutSeconds = 0
+	}
+	var acquired sql.NullInt64
+	if err := conn.QueryRowContext(ctx, `SELECT GET_LOCK(?, ?)`, dashboardBackfillLockName, timeoutSeconds).Scan(&acquired); err != nil {
+		return err
+	}
+	if !acquired.Valid || acquired.Int64 != 1 {
+		if r.metrics != nil {
+			r.metrics.RecordLockWait()
+		}
+		return dashboard.ErrDashboardBackfillLocked
+	}
+	defer func() {
+		var released sql.NullInt64
+		_ = conn.QueryRowContext(context.Background(), `SELECT RELEASE_LOCK(?)`, dashboardBackfillLockName).Scan(&released)
+	}()
+	return fn()
+}
+
+func isRetryableMySQLError(err error) bool {
+	var mysqlErr *driverMysql.MySQLError
+	if !errors.As(err, &mysqlErr) {
+		return false
+	}
+	return mysqlErr.Number == 1213 || mysqlErr.Number == 1205
 }
 
 func (r *MySQLRepository) LoadCheckpoint(ctx context.Context) (mysql.Position, bool, error) {
@@ -47,7 +121,11 @@ WHERE service_name = ?
 	if err != nil {
 		return mysql.Position{}, false, err
 	}
-	return mysql.Position{Name: file, Pos: pos}, true, nil
+	position := mysql.Position{Name: file, Pos: pos}
+	if r.metrics != nil {
+		r.metrics.SetCheckpointPosition(position)
+	}
+	return position, true, nil
 }
 
 func (r *MySQLRepository) SaveCheckpoint(ctx context.Context, pos mysql.Position) error {
@@ -59,6 +137,9 @@ ON DUPLICATE KEY UPDATE
   binlog_pos = VALUES(binlog_pos),
   updated_at = NOW()
 `, serviceName, pos.Name, pos.Pos)
+	if err == nil && r.metrics != nil {
+		r.metrics.SetCheckpointPosition(pos)
+	}
 	return err
 }
 

internal/storage/schema.go

@@ -36,6 +36,7 @@ func (r *MySQLRepository) EnsureSchema(ctx context.Context) error {
 		createPeriodUserMetricTable,
 		createGamePeriodMetricTable,
 		createGamePeriodUserMetricTable,
+		createRechargeDetailTable,
 	})
 }
 
@@ -172,3 +173,23 @@ const createGamePeriodUserMetricTable = `CREATE TABLE IF NOT EXISTS country_dash
   UNIQUE KEY uk_game_period_user (period_type, period_key, stat_timezone, sys_origin, country_code, game_provider, game_id, user_id),
   KEY idx_game_user (sys_origin, stat_timezone, user_id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`
+
+const createRechargeDetailTable = `CREATE TABLE IF NOT EXISTS country_dashboard_recharge_detail (
+  id bigint NOT NULL AUTO_INCREMENT,
+  source_type varchar(32) NOT NULL,
+  record_id varchar(128) NOT NULL,
+  source_channel varchar(64) NOT NULL DEFAULT '',
+  sys_origin varchar(50) NOT NULL,
+  user_id bigint NOT NULL,
+  country_code varchar(64) NOT NULL DEFAULT 'UNKNOWN',
+  country_name varchar(128) NOT NULL DEFAULT 'UNKNOWN',
+  amount decimal(20,2) NOT NULL DEFAULT 0,
+  coin_quantity decimal(20,2) NULL,
+  remark varchar(512) NOT NULL DEFAULT '',
+  create_time datetime NOT NULL,
+  update_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  PRIMARY KEY (id),
+  UNIQUE KEY uk_recharge_source_record (source_type, record_id),
+  KEY idx_recharge_query (sys_origin, create_time, country_code),
+  KEY idx_recharge_user (user_id, create_time)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`

internal/storage/user.go

@@ -22,10 +22,16 @@ ON DUPLICATE KEY UPDATE
   user_updated_at = VALUES(user_updated_at),
   updated_at = NOW()
 `, user.UserID, user.SysOrigin, user.Country.Code, user.Country.Name, user.IsDeleted, nullTime(user.CreatedAt), nullTime(user.UpdatedAt))
+	if err == nil {
+		r.userCache.Set(user.UserID, user)
+	}
 	return err
 }
 
 func (r *MySQLRepository) UserCountry(ctx context.Context, userID int64) (dashboard.UserCountry, bool, error) {
+	if user, ok := r.userCache.Get(userID); ok {
+		return user, true, nil
+	}
 	var user dashboard.UserCountry
 	var countryCode, countryName string
 	var isDeleted bool
@@ -49,6 +55,7 @@ WHERE user_id = ?
 	if updatedAt.Valid {
 		user.UpdatedAt = updatedAt.Time
 	}
+	r.userCache.Set(userID, user)
 	return user, true, nil
 }
 
@@ -84,6 +91,7 @@ INSERT INTO dashboard_cdc_user_country (
 ) VALUES (?, ?, ?, ?, ?, ?, ?, NOW())
 ON DUPLICATE KEY UPDATE updated_at = updated_at
 `, user.UserID, user.SysOrigin, user.Country.Code, user.Country.Name, user.IsDeleted, nullTime(user.CreatedAt), nullTime(user.UpdatedAt))
+	r.userCache.Set(userID, user)
 	return user, true, nil
 }
 

migrations/001_dashboard_cdc_worker.sql

@@ -40,3 +40,23 @@ CREATE TABLE IF NOT EXISTS country_dashboard_daily_user_metric (
   UNIQUE KEY uk_daily_user_metric (stat_date, sys_origin, country_code, user_id, metric_type),
   KEY idx_daily_user_metric (sys_origin, user_id, metric_type)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+
+CREATE TABLE IF NOT EXISTS country_dashboard_recharge_detail (
+  id bigint NOT NULL AUTO_INCREMENT,
+  source_type varchar(32) NOT NULL,
+  record_id varchar(128) NOT NULL,
+  source_channel varchar(64) NOT NULL DEFAULT '',
+  sys_origin varchar(50) NOT NULL,
+  user_id bigint NOT NULL,
+  country_code varchar(64) NOT NULL DEFAULT 'UNKNOWN',
+  country_name varchar(128) NOT NULL DEFAULT 'UNKNOWN',
+  amount decimal(20,2) NOT NULL DEFAULT 0,
+  coin_quantity decimal(20,2) NULL,
+  remark varchar(512) NOT NULL DEFAULT '',
+  create_time datetime NOT NULL,
+  update_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  PRIMARY KEY (id),
+  UNIQUE KEY uk_recharge_source_record (source_type, record_id),
+  KEY idx_recharge_query (sys_origin, create_time, country_code),
+  KEY idx_recharge_user (user_id, create_time)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;