Reconcile gift metrics from source tables

cmd/dashboard-cdc-worker/main.go

@@ -84,7 +84,6 @@ func main() {
 	projector := dashboard.NewProjector(cfg, repo, redisClient, metrics)
 	repo.StartUserReconciler(ctx, logger, projector)
 	repo.StartRechargeReconciler(ctx, logger, projector)
-	repo.StartGiftReconciler(ctx, logger, projector)
 
 	if cfg.Mongo.Enabled {
 		mongoClient, err := mongo.Connect(options.Client().ApplyURI(cfg.Mongo.URI))
@@ -101,7 +100,9 @@ func main() {
 			logger.Error("ping mongo failed", "error", err)
 			os.Exit(1)
 		}
-		repo.StartActiveReconciler(ctx, mongoClient.Database(cfg.Mongo.Database), logger)
+		mongoDB := mongoClient.Database(cfg.Mongo.Database)
+		repo.StartGiftReconciler(ctx, mongoDB, logger)
+		repo.StartActiveReconciler(ctx, mongoDB, logger)
 	}
 
 	worker, err := cdc.NewWorker(cfg, repo, projector, logger, metrics)

internal/storage/gift_helpers.go

@@ -0,0 +1,183 @@
+package storage
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/shopspring/decimal"
+	"go.mongodb.org/mongo-driver/v2/bson"
+
+	"dashboard-cdc-worker/internal/dashboard"
+)
+
+func mergeMongoGiftRows(rows map[string]*giftMetricRow, metrics []mongoGiftAmount, anchorShare bool) {
+	for _, item := range metrics {
+		key := giftMetricKey(item.day, "", item.userID)
+		row := rows[key]
+		if row == nil {
+			row = &giftMetricRow{day: item.day, luckyGiftUsers: make(map[int64]struct{})}
+			rows[key] = row
+		}
+		if anchorShare {
+			row.luckyGiftAnchorShare = row.luckyGiftAnchorShare.Add(item.amount)
+		} else {
+			row.giftConsume = row.giftConsume.Add(item.amount)
+		}
+	}
+}
+
+func mergeSQLLuckyGiftRows(rows map[string]*giftMetricRow, metrics []sqlLuckyGiftRow) {
+	for _, item := range metrics {
+		key := giftMetricKey(item.day, item.country.Code, 0)
+		row := rows[key]
+		if row == nil {
+			row = &giftMetricRow{
+				day:            item.day,
+				sysOrigin:      item.sysOrigin,
+				country:        item.country,
+				luckyGiftUsers: make(map[int64]struct{}),
+			}
+			rows[key] = row
+		}
+		row.sysOrigin = item.sysOrigin
+		row.country = item.country
+		row.luckyGiftTotalFlow = row.luckyGiftTotalFlow.Add(item.payAmount)
+		row.luckyGiftPayout = row.luckyGiftPayout.Add(item.payout)
+		row.luckyGiftUsers[item.userID] = struct{}{}
+	}
+}
+
+func applyGiftUserCountries(rows map[string]*giftMetricRow, users map[int64]dashboard.UserCountry, unknownCountry string) {
+	for key, row := range rows {
+		if row.country.Code != "" {
+			continue
+		}
+		parts := strings.Split(key, "|")
+		if len(parts) != 3 {
+			continue
+		}
+		userID := parseInt64(parts[2])
+		user, ok := users[userID]
+		if !ok {
+			delete(rows, key)
+			continue
+		}
+		row.sysOrigin = user.SysOrigin
+		row.country = user.Country
+		countryKey := giftMetricKey(row.day, row.country.Code, 0)
+		countryRow := rows[countryKey]
+		if countryRow == nil {
+			countryRow = &giftMetricRow{
+				day:            row.day,
+				sysOrigin:      row.sysOrigin,
+				country:        row.country,
+				luckyGiftUsers: make(map[int64]struct{}),
+			}
+			rows[countryKey] = countryRow
+		}
+		countryRow.giftConsume = countryRow.giftConsume.Add(row.giftConsume)
+		countryRow.luckyGiftAnchorShare = countryRow.luckyGiftAnchorShare.Add(row.luckyGiftAnchorShare)
+		delete(rows, key)
+	}
+}
+
+func (row *giftMetricRow) empty() bool {
+	return row.giftConsume.IsZero() &&
+		row.luckyGiftAnchorShare.IsZero() &&
+		row.luckyGiftTotalFlow.IsZero() &&
+		row.luckyGiftPayout.IsZero() &&
+		len(row.luckyGiftUsers) == 0
+}
+
+func giftMetricKey(day, countryCode string, userID int64) string {
+	return fmt.Sprintf("%s|%s|%d", day, countryCode, userID)
+}
+
+func int64SetToSlice(values map[int64]struct{}) []int64 {
+	out := make([]int64, 0, len(values))
+	for value := range values {
+		out = append(out, value)
+	}
+	return out
+}
+
+func bsonInt64(value interface{}) int64 {
+	switch typed := value.(type) {
+	case int64:
+		return typed
+	case int32:
+		return int64(typed)
+	case int:
+		return int64(typed)
+	case float64:
+		return int64(typed)
+	case string:
+		return parseInt64(typed)
+	default:
+		return parseInt64(fmt.Sprint(typed))
+	}
+}
+
+func bsonDecimal(value interface{}) decimal.Decimal {
+	switch typed := value.(type) {
+	case decimal.Decimal:
+		return typed
+	case bson.Decimal128:
+		parsed, err := decimal.NewFromString(typed.String())
+		if err == nil {
+			return parsed
+		}
+	case int64:
+		return decimal.NewFromInt(typed)
+	case int32:
+		return decimal.NewFromInt(int64(typed))
+	case int:
+		return decimal.NewFromInt(int64(typed))
+	case float64:
+		return decimal.NewFromFloat(typed)
+	case string:
+		parsed, err := decimal.NewFromString(typed)
+		if err == nil {
+			return parsed
+		}
+	}
+	parsed, err := decimal.NewFromString(fmt.Sprint(value))
+	if err != nil {
+		return decimal.Zero
+	}
+	return parsed
+}
+
+func parseInt64(value string) int64 {
+	parsed, _ := strconv.ParseInt(strings.TrimSpace(value), 10, 64)
+	return parsed
+}
+
+func defaultSysOrigin(values []string) string {
+	if len(values) == 0 {
+		return ""
+	}
+	return strings.TrimSpace(values[0])
+}
+
+func timezoneOffset(timezone string) string {
+	location, err := time.LoadLocation(timezone)
+	if err != nil {
+		return "+00:00"
+	}
+	_, offset := time.Now().In(location).Zone()
+	sign := "+"
+	if offset < 0 {
+		sign = "-"
+		offset = -offset
+	}
+	hours := offset / 3600
+	minutes := (offset % 3600) / 60
+	return fmt.Sprintf("%s%02d:%02d", sign, hours, minutes)
+}
+
+func quoteIdent(value string) string {
+	return "`" + strings.ReplaceAll(value, "`", "``") + "`"
+}

internal/storage/gift_mongo.go

@@ -0,0 +1,80 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+)
+
+func (r *MySQLRepository) listMongoGiftAmounts(ctx context.Context, db *mongo.Database,
+	start, end time.Time, statTimezone string, luckyGift bool) ([]mongoGiftAmount, error) {
+	filter := bson.D{
+		{Key: "refunded", Value: bson.D{{Key: "$ne", Value: true}}},
+		{Key: "userId", Value: bson.D{{Key: "$ne", Value: nil}}},
+		{Key: "createTime", Value: bson.D{{Key: "$gte", Value: start}, {Key: "$lt", Value: end}}},
+		{Key: "sysOrigin", Value: bson.D{{Key: "$in", Value: r.cfg.Dashboard.SysOrigins}}},
+	}
+	amountPath := "$giftValue.actualAmount"
+	unwindAcceptUsers := false
+	if luckyGift {
+		filter = append(filter, bson.E{Key: "luckyGift", Value: true})
+		amountPath = "$acceptUsers.targetAmount"
+		unwindAcceptUsers = true
+	} else {
+		filter = append(filter,
+			bson.E{Key: "luckyGift", Value: bson.D{{Key: "$ne", Value: true}}},
+			bson.E{Key: "originId", Value: bson.D{{Key: "$regex", Value: "^\\d+$"}}},
+			bson.E{Key: "$or", Value: bson.A{
+				bson.D{{Key: "dynamicContentId", Value: bson.D{{Key: "$exists", Value: false}}}},
+				bson.D{{Key: "dynamicContentId", Value: nil}},
+				bson.D{{Key: "dynamicContentId", Value: ""}},
+			}},
+		)
+	}
+	pipeline := mongo.Pipeline{
+		bson.D{{Key: "$match", Value: filter}},
+	}
+	if unwindAcceptUsers {
+		pipeline = append(pipeline, bson.D{{Key: "$unwind", Value: "$acceptUsers"}})
+	}
+	pipeline = append(pipeline,
+		bson.D{{Key: "$project", Value: bson.D{
+			{Key: "userId", Value: "$userId"},
+			{Key: "amount", Value: amountPath},
+			{Key: "day", Value: bson.D{{Key: "$dateToString", Value: bson.D{
+				{Key: "format", Value: "%Y-%m-%d"},
+				{Key: "date", Value: "$createTime"},
+				{Key: "timezone", Value: statTimezone},
+			}}}},
+		}}},
+		bson.D{{Key: "$group", Value: bson.D{
+			{Key: "_id", Value: bson.D{{Key: "userId", Value: "$userId"}, {Key: "day", Value: "$day"}}},
+			{Key: "amount", Value: bson.D{{Key: "$sum", Value: "$amount"}}},
+		}}},
+	)
+	cursor, err := db.Collection(giftRunningWaterCollection).Aggregate(ctx, pipeline)
+	if err != nil {
+		return nil, err
+	}
+	defer cursor.Close(ctx)
+	var result []mongoGiftAmount
+	for cursor.Next(ctx) {
+		var doc bson.M
+		if err := cursor.Decode(&doc); err != nil {
+			return nil, err
+		}
+		idDoc, _ := doc["_id"].(bson.M)
+		userID := bsonInt64(idDoc["userId"])
+		day := strings.TrimSpace(fmt.Sprint(idDoc["day"]))
+		amount := bsonDecimal(doc["amount"])
+		if userID == 0 || day == "" {
+			continue
+		}
+		result = append(result, mongoGiftAmount{userID: userID, day: day, amount: amount})
+	}
+	return result, cursor.Err()
+}

internal/storage/gift_reconcile.go

@@ -3,100 +3,59 @@ package storage
 import (
 	"context"
 	"database/sql"
-	"fmt"
 	"log/slog"
-	"strings"
 	"time"
 
 	"github.com/shopspring/decimal"
+	"go.mongodb.org/mongo-driver/v2/mongo"
 
 	"dashboard-cdc-worker/internal/dashboard"
 )
 
-type walletGiftRow struct {
+const giftRunningWaterCollection = "gift_give_running_water"
-	table      string
+
-	id         string
+type giftMetricRow struct {
-	idNumber   int64
+	day                  string
-	userID     int64
+	sysOrigin            string
-	sysOrigin  string
+	country              dashboard.Country
-	eventType  string
+	giftConsume          decimal.Decimal
-	recordType int64
+	luckyGiftAnchorShare decimal.Decimal
-	amount     decimal.Decimal
+	luckyGiftTotalFlow   decimal.Decimal
-	eventTime  time.Time
+	luckyGiftPayout      decimal.Decimal
-	user       dashboard.UserCountry
+	luckyGiftUsers       map[int64]struct{}
 }
 
-func (r *MySQLRepository) BuildGiftReconcileEvents(ctx context.Context, since time.Time,
+type mongoGiftAmount struct {
-	limit int) ([]dashboard.Event, error) {
+	userID int64
-	if limit <= 0 {
+	day    string
-		limit = 50000
+	amount decimal.Decimal
-	}
-	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 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,
+type sqlLuckyGiftRow struct {
-	projector interface {
+	id        string
-		ApplyEvents(context.Context, []dashboard.Event) error
+	userID    int64
-	}) {
+	day       string
+	sysOrigin string
+	country   dashboard.Country
+	payAmount decimal.Decimal
+	payout    decimal.Decimal
+}
+
+func (r *MySQLRepository) StartGiftReconciler(ctx context.Context, db *mongo.Database, logger *slog.Logger) {
 	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)
+		until := since.AddDate(0, 0, 1)
-		if err != nil {
+		if err := r.RebuildGiftMetricsRange(ctx, db, since, until); err != nil {
 			if r.metrics != nil {
 				r.metrics.RecordError(err)
 			}
-			logger.Error("dashboard gift reconcile query failed", "error", err)
+			logger.Error("dashboard gift reconcile failed", "since", since, "until", until, "error", err)
 			return
 		}
-		if len(events) == 0 {
+		logger.Info("dashboard gift reconcile applied", "since", since, "until", until)
-			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()
@@ -113,132 +72,63 @@ func (r *MySQLRepository) StartGiftReconciler(ctx context.Context, logger *slog.
 	}()
 }
 
-func (r *MySQLRepository) resetGiftHighWater() {
+func (r *MySQLRepository) RebuildGiftMetricsRange(ctx context.Context, db *mongo.Database, start, end time.Time) error {
-	r.giftMu.Lock()
+	rowsByTimezone := make(map[string]map[string]*giftMetricRow)
-	defer r.giftMu.Unlock()
+	userIDs := make(map[int64]struct{})
-	r.giftHighWater = make(map[string]int64)
+	for _, statTimezone := range r.cfg.Dashboard.StatTimezones {
-}
+		location, err := time.LoadLocation(statTimezone)
-
-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
+			location = time.UTC
 		}
-		result = append(result, rows...)
+		rows := make(map[string]*giftMetricRow)
-		remaining -= len(rows)
+		for day := start.In(location); day.Before(end); day = day.AddDate(0, 0, 1) {
+			dayStart := time.Date(day.Year(), day.Month(), day.Day(), 0, 0, 0, 0, location)
+			dayEnd := dayStart.AddDate(0, 0, 1)
+			giftConsume, err := r.listMongoGiftAmounts(ctx, db, dayStart, dayEnd, statTimezone, false)
+			if err != nil {
+				return err
+			}
+			for _, item := range giftConsume {
+				userIDs[item.userID] = struct{}{}
+			}
+			luckyAnchorShare, err := r.listMongoGiftAmounts(ctx, db, dayStart, dayEnd, statTimezone, true)
+			if err != nil {
+				return err
+			}
+			for _, item := range luckyAnchorShare {
+				userIDs[item.userID] = struct{}{}
+			}
+			luckyRows, err := r.listSQLLuckyGiftRows(ctx, dayStart, dayEnd, statTimezone)
+			if err != nil {
+				return err
+			}
+			for _, item := range luckyRows {
+				userIDs[item.userID] = struct{}{}
+			}
+			mergeMongoGiftRows(rows, giftConsume, false)
+			mergeMongoGiftRows(rows, luckyAnchorShare, true)
+			mergeSQLLuckyGiftRows(rows, luckyRows)
+		}
+		rowsByTimezone[statTimezone] = rows
 	}
-	return result, nil
+	users, err := r.loadSourceUserCountriesByID(ctx, int64SetToSlice(userIDs))
-}
-
-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
+		return err
 	}
-	defer rows.Close()
+	for _, rows := range rowsByTimezone {
-	var tables []string
+		applyGiftUserCountries(rows, users, r.cfg.Dashboard.UnknownCountryCode)
-	for rows.Next() {
-		var table string
-		if err := rows.Scan(&table); err != nil {
-			return nil, err
-		}
-		tables = append(tables, table)
 	}
-	return tables, rows.Err()
+	return r.WithDashboardLock(ctx, r.cfg.CDC.BackfillLockWait, func() error {
-}
+		return r.WithTx(ctx, func(tx *sql.Tx) error {
-
+			for statTimezone, rows := range rowsByTimezone {
-func (r *MySQLRepository) listWalletGiftRowsFromTable(ctx context.Context, table string,
+				if err := r.replaceGiftMetricRows(ctx, tx, statTimezone, start, end, rows); err != nil {
-	since time.Time, limit int) ([]walletGiftRow, error) {
+					return err
-	highWaterKey := r.cfg.MySQL.WalletSchema + "." + table
+				}
-	r.giftMu.Lock()
+				if err := r.rebuildAllPeriodFromDays(ctx, tx, statTimezone); err != nil {
-	highWater := r.giftHighWater[highWaterKey]
+					return err
-	r.giftMu.Unlock()
+				}
-
+			}
-	sinceMillis := since.UnixMilli()
+			return nil
-	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 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/gift_sql.go

@@ -0,0 +1,193 @@
+package storage
+
+import (
+	"context"
+	"database/sql"
+	"time"
+
+	"github.com/shopspring/decimal"
+
+	"dashboard-cdc-worker/internal/dashboard"
+)
+
+func (r *MySQLRepository) listSQLLuckyGiftRows(ctx context.Context, start, end time.Time,
+	statTimezone string) ([]sqlLuckyGiftRow, error) {
+	query := `
+SELECT CAST(t.id AS CHAR), t.user_id, DATE_FORMAT(CONVERT_TZ(t.create_time, '+03:00', ?), '%Y-%m-%d'),
+       t.sys_origin, IFNULL(NULLIF(u.country_code, ''), ?),
+       IFNULL(NULLIF(u.country_name, ''), IFNULL(NULLIF(u.country_code, ''), ?)),
+       t.pay_amount, t.award_amount
+FROM ` + quoteIdent(r.cfg.MySQL.Schema) + `.game_lucky_gift_count t
+JOIN ` + quoteIdent(r.cfg.MySQL.Schema) + `.user_base_info u ON u.id = t.user_id
+WHERE t.create_time >= ?
+  AND t.create_time < ?
+  AND t.sys_origin IN ` + inStringPlaceholders(len(r.cfg.Dashboard.SysOrigins)) + `
+  AND u.origin_sys = t.sys_origin
+  AND (u.is_del = 0 OR u.is_del IS NULL)`
+	args := []interface{}{timezoneOffset(statTimezone), r.cfg.Dashboard.UnknownCountryCode,
+		r.cfg.Dashboard.UnknownCountryCode, start, end}
+	args = append(args, stringArgs(r.cfg.Dashboard.SysOrigins)...)
+	rows, err := r.db.QueryContext(ctx, query, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var result []sqlLuckyGiftRow
+	for rows.Next() {
+		var item sqlLuckyGiftRow
+		var pay, payout decimal.Decimal
+		if err := rows.Scan(&item.id, &item.userID, &item.day, &item.sysOrigin, &item.country.Code,
+			&item.country.Name, &pay, &payout); err != nil {
+			return nil, err
+		}
+		item.payAmount = pay
+		item.payout = payout
+		result = append(result, item)
+	}
+	return result, rows.Err()
+}
+
+func (r *MySQLRepository) replaceGiftMetricRows(ctx context.Context, tx *sql.Tx,
+	statTimezone string, start, end time.Time, rows map[string]*giftMetricRow) error {
+	location, err := time.LoadLocation(statTimezone)
+	if err != nil {
+		location = time.UTC
+	}
+	for day := start.In(location); day.Before(end); day = day.AddDate(0, 0, 1) {
+		dayStart := time.Date(day.Year(), day.Month(), day.Day(), 0, 0, 0, 0, location)
+		dayEnd := dayStart.AddDate(0, 0, 1)
+		dayKey := dayStart.Format("2006-01-02")
+		for _, sysOrigin := range r.cfg.Dashboard.SysOrigins {
+			if _, err := tx.ExecContext(ctx, `
+UPDATE country_dashboard_period_metric
+SET gift_consume = 0,
+    lucky_gift_total_flow = 0,
+    lucky_gift_user = 0,
+    lucky_gift_payout = 0,
+    lucky_gift_anchor_share = 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 := tx.ExecContext(ctx, `
+DELETE FROM country_dashboard_period_user_metric
+WHERE period_type = 'DAY' AND period_key = ? AND stat_timezone = ? AND sys_origin = ?
+  AND metric_type = ?`, dayKey, statTimezone, sysOrigin, dashboard.MetricLuckyGiftUser); err != nil {
+				return err
+			}
+		}
+		if err := r.upsertGiftMetricRows(ctx, tx, statTimezone, dayStart, dayEnd, rows); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (r *MySQLRepository) upsertGiftMetricRows(ctx context.Context, tx *sql.Tx,
+	statTimezone string, dayStart, dayEnd time.Time, rows map[string]*giftMetricRow) error {
+	dayKey := dayStart.Format("2006-01-02")
+	var metricRows []*giftMetricRow
+	for _, row := range rows {
+		if row.day != dayKey || row.empty() {
+			continue
+		}
+		if row.sysOrigin == "" {
+			row.sysOrigin = defaultSysOrigin(r.cfg.Dashboard.SysOrigins)
+		}
+		if !r.allowedOrigin(row.sysOrigin) {
+			continue
+		}
+		if row.country.Code == "" {
+			row.country.Code = r.cfg.Dashboard.UnknownCountryCode
+		}
+		if row.country.Name == "" {
+			row.country.Name = row.country.Code
+		}
+		metricRows = append(metricRows, row)
+	}
+	if len(metricRows) == 0 {
+		return nil
+	}
+	args := make([]interface{}, 0, len(metricRows)*14)
+	for _, row := range metricRows {
+		args = append(args, dashboard.PeriodDay, dayKey, statTimezone, dayKey, dayStart, dayEnd,
+			row.sysOrigin, row.country.Code, row.country.Name, row.giftConsume,
+			row.luckyGiftTotalFlow, len(row.luckyGiftUsers), row.luckyGiftPayout,
+			row.luckyGiftAnchorShare)
+	}
+	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, gift_consume, lucky_gift_total_flow,
+  lucky_gift_user, lucky_gift_payout, lucky_gift_anchor_share
+) VALUES `+placeholders(len(metricRows), 14)+`
+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),
+  gift_consume = VALUES(gift_consume),
+  lucky_gift_total_flow = VALUES(lucky_gift_total_flow),
+  lucky_gift_user = VALUES(lucky_gift_user),
+  lucky_gift_payout = VALUES(lucky_gift_payout),
+  lucky_gift_anchor_share = VALUES(lucky_gift_anchor_share),
+  refreshed_at = NOW()`, args...); err != nil {
+		return err
+	}
+	for _, row := range metricRows {
+		if len(row.luckyGiftUsers) == 0 {
+			continue
+		}
+		key := periodUserKey{
+			periodType: dashboard.PeriodDay, periodKey: dayKey, statTimezone: statTimezone,
+			periodName: dayKey, startDate: dayStart, endDate: dayEnd, sysOrigin: row.sysOrigin,
+			countryCode: row.country.Code, countryName: row.country.Name, metricType: dashboard.MetricLuckyGiftUser,
+		}
+		if _, err := bulkInsertPeriodUsers(ctx, tx, key, int64SetToSlice(row.luckyGiftUsers)); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (r *MySQLRepository) loadSourceUserCountriesByID(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+2)
+		args = append(args, r.cfg.Dashboard.UnknownCountryCode, r.cfg.Dashboard.UnknownCountryCode)
+		for _, id := range ids[start:end] {
+			args = append(args, id)
+		}
+		rows, err := r.db.QueryContext(ctx, `
+SELECT id, origin_sys, IFNULL(NULLIF(country_code, ''), ?),
+       IFNULL(NULLIF(country_name, ''), IFNULL(NULLIF(country_code, ''), ?)),
+       IFNULL(is_del, 0), create_time, update_time
+FROM `+quoteIdent(r.cfg.MySQL.Schema)+`.user_base_info
+WHERE 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
+}