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 salaryTransfer decimal.Decimal giftConsume 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.salaryTransfer = t.salaryTransfer.Add(c.SalaryTransfer.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)) 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.salaryTransfer.IsZero() && t.giftConsume.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, ",") }