fix: reconcile dashboard gift wallet metrics

cmd/dashboard-cdc-worker/main.go

@@ -81,6 +81,7 @@ func main() {
 
 	projector := dashboard.NewProjector(cfg, repo, redisClient, metrics)
 	repo.StartRechargeReconciler(ctx, logger, projector)
+	repo.StartGiftReconciler(ctx, logger, projector)
 
 	worker, err := cdc.NewWorker(cfg, repo, projector, logger, metrics)
 	if err != nil {

internal/cdc/handler.go

@@ -152,6 +152,8 @@ func (h *handler) applyRows(ctx context.Context, e *canal.RowsEvent, rowIndex in
 	eventKey := fmt.Sprintf("%s:%d:%s:%s:%d", h.currentFile(), e.Header.LogPos, source, e.Action, rowIndex)
 	if key := stableRechargeEventKey(source, contributions); key != "" {
 		eventKey = key
+	} else if key := stableContributionEventKey(contributions); key != "" {
+		eventKey = key
 	}
 	event := dashboard.Event{
 		EventKey:      eventKey,
@@ -166,6 +168,16 @@ func (h *handler) applyRows(ctx context.Context, e *canal.RowsEvent, rowIndex in
 	return nil
 }
 
+func stableContributionEventKey(contributions []dashboard.Contribution) string {
+	for _, contribution := range contributions {
+		if contribution.DeltaSign < 0 || contribution.EventKey == "" {
+			continue
+		}
+		return contribution.EventKey
+	}
+	return ""
+}
+
 func stableRechargeEventKey(source string, contributions []dashboard.Contribution) string {
 	if source != "likei.order_purchase_history" && source != "likei.order_user_purchase_pay" {
 		return ""

internal/config/config.go

@@ -16,6 +16,7 @@ type Config struct {
 	CDC       CDCConfig
 	Dashboard DashboardConfig
 	Recharge  RechargeReconcileConfig
+	Gift      GiftReconcileConfig
 	Health    HealthConfig
 }
 
@@ -70,6 +71,13 @@ type RechargeReconcileConfig struct {
 	BatchLimit    int
 }
 
+type GiftReconcileConfig struct {
+	Enabled       bool
+	Interval      time.Duration
+	CurrentDayLag time.Duration
+	BatchLimit    int
+}
+
 type HealthConfig struct {
 	Addr string
 }
@@ -122,6 +130,12 @@ func Load() (Config, error) {
 			CurrentDayLag: envDuration("DASHBOARD_RECHARGE_RECONCILE_CURRENT_DAY_LAG", 15*time.Minute),
 			BatchLimit:    envInt("DASHBOARD_RECHARGE_RECONCILE_BATCH_LIMIT", 5000),
 		},
+		Gift: GiftReconcileConfig{
+			Enabled:       envBool("DASHBOARD_GIFT_RECONCILE_ENABLED", true),
+			Interval:      envDuration("DASHBOARD_GIFT_RECONCILE_INTERVAL", 10*time.Minute),
+			CurrentDayLag: envDuration("DASHBOARD_GIFT_RECONCILE_CURRENT_DAY_LAG", 5*time.Minute),
+			BatchLimit:    envInt("DASHBOARD_GIFT_RECONCILE_BATCH_LIMIT", 50000),
+		},
 		Health: HealthConfig{
 			Addr: env("HEALTH_ADDR", ":2910"),
 		},
@@ -151,6 +165,9 @@ func Load() (Config, error) {
 	if cfg.Recharge.BatchLimit <= 0 {
 		cfg.Recharge.BatchLimit = 5000
 	}
+	if cfg.Gift.BatchLimit <= 0 {
+		cfg.Gift.BatchLimit = 50000
+	}
 	if cfg.CDC.StartMode != "checkpoint" && cfg.CDC.StartMode != "current" {
 		return Config{}, fmt.Errorf("unsupported CDC_START_MODE=%s", cfg.CDC.StartMode)
 	}

internal/dashboard/mapper_helpers.go

@@ -2,6 +2,7 @@ package dashboard
 
 import (
 	"context"
+	"fmt"
 	"strings"
 	"time"
 
@@ -90,6 +91,13 @@ func rowPK(row Row) string {
 	return strings.TrimSpace(row.String("event_id"))
 }
 
+func WalletMetricEventKey(schema string, table string, id string) string {
+	if strings.TrimSpace(id) == "" {
+		return ""
+	}
+	return fmt.Sprintf("wallet-metric:%s.%s:%s", schema, table, 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 {

internal/dashboard/mapper_wallet.go

@@ -18,6 +18,9 @@ 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) {
+		return m.mapWalletGift(ctx, schema, table, action, row)
+	}
 	if !isWalletGameEvent(eventType) {
 		return nil, nil
 	}
@@ -60,6 +63,47 @@ func (m *Mapper) mapWalletGame(ctx context.Context, schema string, table string,
 	return []Contribution{c}, nil
 }
 
+func (m *Mapper) mapWalletGift(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) {
+	sysOrigin := row.String("sys_origin")
+	if !m.allowed(sysOrigin) {
+		return nil, nil
+	}
+	user, ok, err := m.user(ctx, row.Int64("user_id"))
+	if err != nil || !ok || user.IsDeleted {
+		return nil, err
+	}
+	if user.SysOrigin != sysOrigin {
+		return nil, nil
+	}
+
+	amount := row.Decimal("penny_amount").Div(decimal.NewFromInt(100))
+	if amount.IsZero() {
+		return nil, nil
+	}
+	eventType := row.String("event_type")
+	recordType := row.Int64("type")
+	c := m.base(schema, table, action, row, user, row.MillisTime("create_time"))
+	if action != "delete" {
+		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:
+		return nil, nil
+	}
+	return []Contribution{c}, nil
+}
+
+func isWalletGiftConsumeEvent(eventType string) bool {
+	return strings.HasPrefix(eventType, "GIVE_GIFT") && !strings.Contains(eventType, "LUCKY_GIFT")
+}
+
 func isWalletGameEvent(eventType string) bool {
 	if _, ok := walletGameEvents[eventType]; ok {
 		return true

internal/dashboard/types.go

@@ -47,6 +47,7 @@ type Contribution struct {
 	GoogleRecharge           decimal.Decimal
 	DealerRecharge           decimal.Decimal
 	SalaryExchange           decimal.Decimal
+	GiftConsume              decimal.Decimal
 	LuckyGiftTotalFlow       decimal.Decimal
 	LuckyGiftUser            bool
 	LuckyGiftPayout          decimal.Decimal
@@ -81,6 +82,7 @@ func (c Contribution) Empty() bool {
 		c.GoogleRecharge.IsZero() &&
 		c.DealerRecharge.IsZero() &&
 		c.SalaryExchange.IsZero() &&
+		c.GiftConsume.IsZero() &&
 		c.LuckyGiftTotalFlow.IsZero() &&
 		!c.LuckyGiftUser &&
 		c.LuckyGiftPayout.IsZero() &&

internal/storage/aggregate.go

@@ -100,9 +100,9 @@ func (r *MySQLRepository) upsertDailyAmounts(ctx context.Context, tx *sql.Tx, c
 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,
+  dealer_recharge, salary_exchange, gift_consume, lucky_gift_total_flow, lucky_gift_payout,
   game_total_flow, game_payout, refreshed_at
-) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
 ON DUPLICATE KEY UPDATE
   country_name = VALUES(country_name),
   new_user_recharge = new_user_recharge + VALUES(new_user_recharge),
@@ -111,6 +111,7 @@ ON DUPLICATE KEY UPDATE
   google_recharge = google_recharge + VALUES(google_recharge),
   dealer_recharge = dealer_recharge + VALUES(dealer_recharge),
   salary_exchange = salary_exchange + VALUES(salary_exchange),
+  gift_consume = gift_consume + VALUES(gift_consume),
   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),
@@ -118,7 +119,7 @@ ON DUPLICATE KEY UPDATE
   refreshed_at = NOW()
 `, day, dashboard.DateNumber(day), c.SysOrigin, c.Country.Code, c.Country.Name,
 		newUserRecharge.Mul(sign), c.OfficialRecharge.Mul(sign), c.MifapayRecharge.Mul(sign), c.GoogleRecharge.Mul(sign),
-		c.DealerRecharge.Mul(sign), c.SalaryExchange.Mul(sign), c.LuckyGiftTotalFlow.Mul(sign), c.LuckyGiftPayout.Mul(sign),
+		c.DealerRecharge.Mul(sign), c.SalaryExchange.Mul(sign), c.GiftConsume.Mul(sign), c.LuckyGiftTotalFlow.Mul(sign), c.LuckyGiftPayout.Mul(sign),
 		c.GameTotalFlow.Mul(sign), c.GamePayout.Mul(sign))
 	return err
 }
@@ -129,9 +130,9 @@ func (r *MySQLRepository) upsertPeriodAmounts(ctx context.Context, tx *sql.Tx, c
 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,
+  google_recharge, dealer_recharge, salary_exchange, gift_consume, lucky_gift_total_flow, lucky_gift_payout,
   game_total_flow, game_payout, refreshed_at
-) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
 ON DUPLICATE KEY UPDATE
   period_name = VALUES(period_name),
   period_start_date = VALUES(period_start_date),
@@ -143,6 +144,7 @@ ON DUPLICATE KEY UPDATE
   google_recharge = google_recharge + VALUES(google_recharge),
   dealer_recharge = dealer_recharge + VALUES(dealer_recharge),
   salary_exchange = salary_exchange + VALUES(salary_exchange),
+  gift_consume = gift_consume + VALUES(gift_consume),
   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),
@@ -151,7 +153,7 @@ ON DUPLICATE KEY UPDATE
 `, window.Type, window.Key, statTimezone, window.Name, window.StartDate, window.EndDate,
 		c.SysOrigin, c.Country.Code, c.Country.Name, newUserRecharge.Mul(sign), c.OfficialRecharge.Mul(sign),
 		c.MifapayRecharge.Mul(sign), c.GoogleRecharge.Mul(sign), c.DealerRecharge.Mul(sign), c.SalaryExchange.Mul(sign),
-		c.LuckyGiftTotalFlow.Mul(sign), c.LuckyGiftPayout.Mul(sign), c.GameTotalFlow.Mul(sign), c.GamePayout.Mul(sign))
+		c.GiftConsume.Mul(sign), c.LuckyGiftTotalFlow.Mul(sign), c.LuckyGiftPayout.Mul(sign), c.GameTotalFlow.Mul(sign), c.GamePayout.Mul(sign))
 	return err
 }
 

internal/storage/batch.go

@@ -20,6 +20,7 @@ type amountTotals struct {
 	googleRecharge     decimal.Decimal
 	dealerRecharge     decimal.Decimal
 	salaryExchange     decimal.Decimal
+	giftConsume        decimal.Decimal
 	luckyGiftTotalFlow decimal.Decimal
 	luckyGiftPayout    decimal.Decimal
 	gameTotalFlow      decimal.Decimal
@@ -252,6 +253,7 @@ func (t *amountTotals) add(c dashboard.Contribution, newUserRecharge decimal.Dec
 	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.giftConsume = t.giftConsume.Add(c.GiftConsume.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))
@@ -261,7 +263,7 @@ func (t *amountTotals) add(c dashboard.Contribution, newUserRecharge decimal.Dec
 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.giftConsume.IsZero() && t.luckyGiftTotalFlow.IsZero() && t.luckyGiftPayout.IsZero() &&
 		t.gameTotalFlow.IsZero() && t.gamePayout.IsZero()
 }
 

internal/storage/batch_sql.go

@@ -11,9 +11,9 @@ func (r *MySQLRepository) upsertDailyAmountTotals(ctx context.Context, tx *sql.T
 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,
+  dealer_recharge, salary_exchange, gift_consume, lucky_gift_total_flow, lucky_gift_payout,
   game_total_flow, game_payout, refreshed_at
-) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
 ON DUPLICATE KEY UPDATE
   country_name = VALUES(country_name),
   new_user_recharge = new_user_recharge + VALUES(new_user_recharge),
@@ -22,6 +22,7 @@ ON DUPLICATE KEY UPDATE
   google_recharge = google_recharge + VALUES(google_recharge),
   dealer_recharge = dealer_recharge + VALUES(dealer_recharge),
   salary_exchange = salary_exchange + VALUES(salary_exchange),
+  gift_consume = gift_consume + VALUES(gift_consume),
   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),
@@ -29,7 +30,7 @@ ON DUPLICATE KEY UPDATE
   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.dealerRecharge, totals.salaryExchange, totals.giftConsume, totals.luckyGiftTotalFlow, totals.luckyGiftPayout,
 		totals.gameTotalFlow, totals.gamePayout)
 	return err
 }
@@ -39,9 +40,9 @@ func (r *MySQLRepository) upsertPeriodAmountTotals(ctx context.Context, tx *sql.
 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,
+  google_recharge, dealer_recharge, salary_exchange, gift_consume, lucky_gift_total_flow, lucky_gift_payout,
   game_total_flow, game_payout, refreshed_at
-) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
 ON DUPLICATE KEY UPDATE
   period_name = VALUES(period_name),
   period_start_date = VALUES(period_start_date),
@@ -53,6 +54,7 @@ ON DUPLICATE KEY UPDATE
   google_recharge = google_recharge + VALUES(google_recharge),
   dealer_recharge = dealer_recharge + VALUES(dealer_recharge),
   salary_exchange = salary_exchange + VALUES(salary_exchange),
+  gift_consume = gift_consume + VALUES(gift_consume),
   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),
@@ -61,7 +63,7 @@ ON DUPLICATE KEY UPDATE
 `, 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)
+		totals.giftConsume, totals.luckyGiftTotalFlow, totals.luckyGiftPayout, totals.gameTotalFlow, totals.gamePayout)
 	return err
 }
 

internal/storage/gift_reconcile.go

@@ -0,0 +1,250 @@
+package storage
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"log/slog"
+	"strings"
+	"time"
+
+	"github.com/shopspring/decimal"
+
+	"dashboard-cdc-worker/internal/dashboard"
+)
+
+type walletGiftRow struct {
+	table      string
+	id         string
+	idNumber   int64
+	userID     int64
+	sysOrigin  string
+	eventType  string
+	recordType int64
+	amount     decimal.Decimal
+	eventTime  time.Time
+	user       dashboard.UserCountry
+}
+
+func (r *MySQLRepository) BuildGiftReconcileEvents(ctx context.Context, since time.Time,
+	limit int) ([]dashboard.Event, error) {
+	if limit <= 0 {
+		limit = 50000
+	}
+	rows, err := r.listWalletGiftRows(ctx, since, limit)
+	if err != nil {
+		return nil, err
+	}
+	events := make([]dashboard.Event, 0, len(rows))
+	for _, row := range rows {
+		if !r.allowedOrigin(row.sysOrigin) || row.user.IsDeleted || row.user.SysOrigin != row.sysOrigin {
+			continue
+		}
+		contribution := dashboard.Contribution{
+			SourceSchema:  r.cfg.MySQL.WalletSchema,
+			SourceTable:   row.table,
+			SourcePK:      row.id,
+			SourceAction:  "insert",
+			EventKey:      dashboard.WalletMetricEventKey(r.cfg.MySQL.WalletSchema, row.table, row.id),
+			DeltaSign:     1,
+			SysOrigin:     row.sysOrigin,
+			UserID:        row.userID,
+			Country:       row.user.Country,
+			EventTime:     row.eventTime,
+			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:
+			continue
+		}
+		events = append(events, dashboard.Event{
+			EventKey:      contribution.EventKey,
+			Source:        r.cfg.MySQL.WalletSchema + "." + row.table,
+			Contributions: []dashboard.Contribution{contribution},
+		})
+	}
+	return events, nil
+}
+
+func (r *MySQLRepository) StartGiftReconciler(ctx context.Context, logger *slog.Logger,
+	projector interface {
+		ApplyEvents(context.Context, []dashboard.Event) error
+	}) {
+	if !r.cfg.Gift.Enabled || r.cfg.Gift.Interval <= 0 {
+		return
+	}
+	run := func() {
+		since := currentStatDayStart(r.cfg.Dashboard.StatTimezones, r.cfg.Gift.CurrentDayLag)
+		events, err := r.BuildGiftReconcileEvents(ctx, since, r.cfg.Gift.BatchLimit)
+		if err != nil {
+			if r.metrics != nil {
+				r.metrics.RecordError(err)
+			}
+			logger.Error("dashboard gift reconcile query failed", "error", err)
+			return
+		}
+		if len(events) == 0 {
+			return
+		}
+		if err := projector.ApplyEvents(ctx, events); err != nil {
+			r.resetGiftHighWater()
+			if r.metrics != nil {
+				r.metrics.RecordError(err)
+			}
+			logger.Error("dashboard gift reconcile apply failed", "events", len(events), "error", err)
+			return
+		}
+		logger.Info("dashboard gift reconcile applied", "events", len(events), "since", since)
+	}
+	go func() {
+		run()
+		ticker := time.NewTicker(r.cfg.Gift.Interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				run()
+			}
+		}
+	}()
+}
+
+func (r *MySQLRepository) resetGiftHighWater() {
+	r.giftMu.Lock()
+	defer r.giftMu.Unlock()
+	r.giftHighWater = make(map[string]int64)
+}
+
+func (r *MySQLRepository) listWalletGiftRows(ctx context.Context, since time.Time,
+	limit int) ([]walletGiftRow, error) {
+	tables, err := r.walletRecordTables(ctx)
+	if err != nil {
+		return nil, err
+	}
+	remaining := limit
+	var result []walletGiftRow
+	for _, table := range tables {
+		if remaining <= 0 {
+			break
+		}
+		rows, err := r.listWalletGiftRowsFromTable(ctx, table, since, remaining)
+		if err != nil {
+			return nil, err
+		}
+		result = append(result, rows...)
+		remaining -= len(rows)
+	}
+	return result, nil
+}
+
+func (r *MySQLRepository) walletRecordTables(ctx context.Context) ([]string, error) {
+	rows, err := r.db.QueryContext(ctx, `
+SELECT table_name
+FROM information_schema.tables
+WHERE table_schema = ?
+  AND table_name LIKE 'wallet_gold_asset_record\_%'
+ORDER BY table_name`, r.cfg.MySQL.WalletSchema)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var tables []string
+	for rows.Next() {
+		var table string
+		if err := rows.Scan(&table); err != nil {
+			return nil, err
+		}
+		tables = append(tables, table)
+	}
+	return tables, rows.Err()
+}
+
+func (r *MySQLRepository) listWalletGiftRowsFromTable(ctx context.Context, table string,
+	since time.Time, limit int) ([]walletGiftRow, error) {
+	highWaterKey := r.cfg.MySQL.WalletSchema + "." + table
+	r.giftMu.Lock()
+	highWater := r.giftHighWater[highWaterKey]
+	r.giftMu.Unlock()
+
+	sinceMillis := since.UnixMilli()
+	query := fmt.Sprintf(`
+SELECT CAST(w.id AS CHAR), w.id, w.user_id, w.sys_origin, w.event_type, w.type, w.penny_amount,
+       FROM_UNIXTIME(w.create_time / 1000),
+       u.id, u.origin_sys, IFNULL(NULLIF(u.country_code, ''), ?),
+       IFNULL(NULLIF(u.country_name, ''), IFNULL(NULLIF(u.country_code, ''), ?)),
+       IFNULL(u.is_del, 0), u.create_time, u.update_time
+FROM %s.%s w
+JOIN %s.user_base_info u ON u.id = w.user_id
+WHERE w.id > ?
+  AND w.create_time >= ?
+  AND (
+    w.event_type = 'LUCKY_GIFT'
+    OR (w.event_type LIKE 'GIVE_GIFT%%' AND w.event_type NOT LIKE '%%LUCKY_GIFT%%' 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,
+		r.cfg.Dashboard.UnknownCountryCode, highWater, sinceMillis, limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var result []walletGiftRow
+	maxID := highWater
+	for rows.Next() {
+		var item walletGiftRow
+		item.table = table
+		var penny int64
+		var countryCode, countryName string
+		var isDeleted bool
+		var userCreatedAt, userUpdatedAt sql.NullTime
+		if err := rows.Scan(&item.id, &item.idNumber, &item.userID, &item.sysOrigin,
+			&item.eventType, &item.recordType, &penny, &item.eventTime, &item.user.UserID,
+			&item.user.SysOrigin, &countryCode, &countryName, &isDeleted,
+			&userCreatedAt, &userUpdatedAt); err != nil {
+			return nil, err
+		}
+		item.amount = decimal.NewFromInt(penny).Div(decimal.NewFromInt(100))
+		item.user.Country = dashboard.Country{Code: countryCode, Name: countryName}
+		item.user.IsDeleted = isDeleted
+		if userCreatedAt.Valid {
+			item.user.CreatedAt = userCreatedAt.Time
+		}
+		if userUpdatedAt.Valid {
+			item.user.UpdatedAt = userUpdatedAt.Time
+		}
+		if item.idNumber > maxID {
+			maxID = item.idNumber
+		}
+		result = append(result, item)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	if maxID > highWater {
+		r.giftMu.Lock()
+		if maxID > r.giftHighWater[highWaterKey] {
+			r.giftHighWater[highWaterKey] = maxID
+		}
+		r.giftMu.Unlock()
+	}
+	return result, nil
+}
+
+func isGiftConsumeEvent(eventType string) bool {
+	return strings.HasPrefix(eventType, "GIVE_GIFT") && !strings.Contains(eventType, "LUCKY_GIFT")
+}
+
+func quoteIdent(value string) string {
+	return "`" + strings.ReplaceAll(value, "`", "``") + "`"
+}

internal/storage/repository.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"math"
+	"sync"
 	"time"
 
 	"github.com/go-mysql-org/go-mysql/mysql"
@@ -25,6 +26,8 @@ type MySQLRepository struct {
 	metrics       *monitor.Metrics
 	userCache     *ttlCache[int64, dashboard.UserCountry]
 	gameNameCache *ttlCache[string, string]
+	giftMu        sync.Mutex
+	giftHighWater map[string]int64
 }
 
 func NewMySQLRepository(db *sql.DB, cfg config.Config, metrics *monitor.Metrics) *MySQLRepository {
@@ -34,6 +37,7 @@ func NewMySQLRepository(db *sql.DB, cfg config.Config, metrics *monitor.Metrics)
 		metrics:       metrics,
 		userCache:     newTTLCache[int64, dashboard.UserCountry](30*time.Minute, 200000),
 		gameNameCache: newTTLCache[string, string](6*time.Hour, 10000),
+		giftHighWater: make(map[string]int64),
 	}
 }
 

internal/storage/schema.go

@@ -54,6 +54,7 @@ const createDailyMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_dai
   google_recharge decimal(20,2) NOT NULL DEFAULT 0,
   dealer_recharge decimal(20,2) NOT NULL DEFAULT 0,
   salary_exchange decimal(20,2) NOT NULL DEFAULT 0,
+  gift_consume decimal(20,2) NOT NULL DEFAULT 0,
   lucky_gift_total_flow decimal(20,2) NOT NULL DEFAULT 0,
   lucky_gift_user bigint NOT NULL DEFAULT 0,
   lucky_gift_payout decimal(20,2) NOT NULL DEFAULT 0,
@@ -98,6 +99,7 @@ const createPeriodMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_pe
   google_recharge decimal(20,2) NOT NULL DEFAULT 0,
   dealer_recharge decimal(20,2) NOT NULL DEFAULT 0,
   salary_exchange decimal(20,2) NOT NULL DEFAULT 0,
+  gift_consume decimal(20,2) NOT NULL DEFAULT 0,
   lucky_gift_total_flow decimal(20,2) NOT NULL DEFAULT 0,
   lucky_gift_user bigint NOT NULL DEFAULT 0,
   lucky_gift_payout decimal(20,2) NOT NULL DEFAULT 0,