405 lines
12 KiB
Go
405 lines
12 KiB
Go
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
|
|
userRecharge decimal.Decimal
|
|
newDealerUserRecharge 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 err := r.upsertGameEventFact(ctx, tx, c, storageZone); 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
|
|
newDealerUserRecharge := decimal.Zero
|
|
if dashboard.SamePeriod(c.UserCreatedAt, c.EventTime, dashboard.PeriodWindow{Type: dashboard.PeriodDay}, storageZone) {
|
|
newUserRecharge = c.NewUserRechargeCandidate
|
|
newDealerUserRecharge = c.NewDealerUserRecharge
|
|
}
|
|
dailyKey := dailyAmountKey{day: day, sysOrigin: c.SysOrigin, countryCode: c.Country.Code, countryName: c.Country.Name}
|
|
totals := dailyAmounts[dailyKey]
|
|
totals.add(c, newUserRecharge, newDealerUserRecharge)
|
|
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
|
|
newDealerUserRecharge = decimal.Zero
|
|
if dashboard.SamePeriod(c.UserCreatedAt, c.EventTime, window, zone) {
|
|
newUserRecharge = c.NewUserRechargeCandidate
|
|
newDealerUserRecharge = c.NewDealerUserRecharge
|
|
}
|
|
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, newDealerUserRecharge)
|
|
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 _, 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
|
|
}
|
|
}
|
|
}
|
|
|
|
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
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (t *amountTotals) add(c dashboard.Contribution, newUserRecharge decimal.Decimal, newDealerUserRecharge 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.userRecharge = t.userRecharge.Add(c.UserRecharge.Mul(sign))
|
|
t.newDealerUserRecharge = t.newDealerUserRecharge.Add(newDealerUserRecharge.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.userRecharge.IsZero() &&
|
|
t.newDealerUserRecharge.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)
|
|
}
|
|
if c.UserRecharge.GreaterThan(decimal.Zero) {
|
|
addDailyUser(groups, c, day, dashboard.MetricRechargeUser)
|
|
}
|
|
}
|
|
|
|
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)
|
|
}
|
|
if c.UserRecharge.GreaterThan(decimal.Zero) {
|
|
addPeriodUser(groups, c, window, statTimezone, dashboard.MetricRechargeUser)
|
|
}
|
|
}
|
|
|
|
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, ",")
|
|
}
|