dashboard-cdc-worker/internal/storage/gift_reconcile.go

package storage

import (
	"context"
	"database/sql"
	"log/slog"
	"time"

	"github.com/shopspring/decimal"
	"go.mongodb.org/mongo-driver/v2/mongo"

	"dashboard-cdc-worker/internal/dashboard"
)

const giftRunningWaterCollection = "gift_give_running_water"

type giftMetricRow struct {
	day                  string
	sysOrigin            string
	country              dashboard.Country
	giftConsume          decimal.Decimal
	luckyGiftAnchorShare decimal.Decimal
	luckyGiftTotalFlow   decimal.Decimal
	luckyGiftPayout      decimal.Decimal
	luckyGiftUsers       map[int64]struct{}
}

type mongoGiftAmount struct {
	userID int64
	day    string
	amount decimal.Decimal
}

type sqlLuckyGiftRow struct {
	id        string
	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)
		if r.cfg.Gift.LookbackDays > 1 {
			since = since.AddDate(0, 0, -(r.cfg.Gift.LookbackDays - 1))
		}
		until := since.AddDate(0, 0, 1)
		if r.cfg.Gift.LookbackDays > 1 {
			until = since.AddDate(0, 0, r.cfg.Gift.LookbackDays)
		}
		if err := r.RebuildGiftMetricsRange(ctx, db, since, until); err != nil {
			if r.metrics != nil {
				r.metrics.RecordError(err)
			}
			logger.Error("dashboard gift reconcile failed", "since", since, "until", until, "error", err)
			return
		}
		logger.Info("dashboard gift reconcile applied", "since", since, "until", until)
	}
	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) RebuildGiftMetricsRange(ctx context.Context, db *mongo.Database, start, end time.Time) error {
	rowsByTimezone := make(map[string]map[string]*giftMetricRow)
	userIDs := make(map[int64]struct{})
	for _, statTimezone := range r.cfg.Dashboard.StatTimezones {
		location, err := time.LoadLocation(statTimezone)
		if err != nil {
			location = time.UTC
		}
		rows := make(map[string]*giftMetricRow)
		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
	}
	users, err := r.loadSourceUserCountriesByID(ctx, int64SetToSlice(userIDs))
	if err != nil {
		return err
	}
	for _, rows := range rowsByTimezone {
		applyGiftUserCountries(rows, users, r.cfg.Dashboard.UnknownCountryCode)
	}
	return r.WithDashboardLock(ctx, r.cfg.CDC.BackfillLockWait, func() error {
		return r.WithTx(ctx, func(tx *sql.Tx) error {
			for statTimezone, rows := range rowsByTimezone {
				if err := r.replaceGiftMetricRows(ctx, tx, statTimezone, start, end, rows); err != nil {
					return err
				}
				if err := r.rebuildAllPeriodFromDays(ctx, tx, statTimezone); err != nil {
					return err
				}
			}
			return nil
		})
	})
}