diff --git a/internal/dashboard/mapper.go b/internal/dashboard/mapper.go index 6214ff9..9aa3163 100644 --- a/internal/dashboard/mapper.go +++ b/internal/dashboard/mapper.go @@ -36,7 +36,7 @@ func (m *Mapper) Map(ctx context.Context, schema string, table string, action st case "order_purchase_history": return m.mapOfficialRecharge(ctx, schema, table, action, row) case "order_user_purchase_pay": - return m.mapMifapayRecharge(ctx, schema, table, action, row) + return m.mapThirdPartyRecharge(ctx, schema, table, action, row) case "game_lucky_gift_count": return m.mapLuckyGift(ctx, schema, table, action, row) case "game_lucky_box_count": @@ -96,6 +96,7 @@ func (m *Mapper) mapOfficialRecharge(ctx context.Context, schema string, table s } if payPlatform == "GOOGLE" { c.GoogleRecharge = amount + c.UserRecharge = amount } c.NewUserRechargeCandidate = amount c.RechargeDetail = rechargeDetail("OFFICIAL", rowPK(row), payPlatform, row, user, amount, @@ -103,19 +104,26 @@ func (m *Mapper) mapOfficialRecharge(ctx context.Context, schema string, table s return []Contribution{c}, nil } -func (m *Mapper) mapMifapayRecharge(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { - if row.String("evn") != "PROD" || row.String("factory_code") != "MIFA_PAY" || +func (m *Mapper) mapThirdPartyRecharge(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { + if row.String("evn") != "PROD" || row.String("pay_status") != "SUCCESSFUL" || row.String("receipt_type") != "PAYMENT" || row.String("refund_status") != "NONE" { return nil, nil } items, err := m.amountContribution(ctx, schema, table, action, row, row.Decimal("compute_usd_amount"), func(c *Contribution, amount decimal.Decimal) { - c.MifapayRecharge = amount + c.UserRecharge = amount + if row.String("factory_code") == "MIFA_PAY" { + c.MifapayRecharge = amount + } }) if err != nil || len(items) == 0 { return items, err } - items[0].RechargeDetail = rechargeDetail("MIFAPAY", rowPK(row), row.String("factory_code"), row, + sourceType := "THIRD_PARTY" + if row.String("factory_code") == "MIFA_PAY" { + sourceType = "MIFAPAY" + } + items[0].RechargeDetail = rechargeDetail(sourceType, rowPK(row), row.String("factory_code"), row, detailUser(items[0]), row.Decimal("compute_usd_amount"), decimal.Zero, "") return items, nil } @@ -126,6 +134,7 @@ func (m *Mapper) mapDealerRecharge(ctx context.Context, schema string, table str } items, err := m.amountContribution(ctx, schema, table, action, row, row.Decimal("amount"), func(c *Contribution, amount decimal.Decimal) { c.DealerRecharge = amount + c.NewDealerUserRecharge = amount }) if err != nil || len(items) == 0 { return items, err diff --git a/internal/dashboard/mapper_wallet_test.go b/internal/dashboard/mapper_wallet_test.go index d2fceec..7f2e60c 100644 --- a/internal/dashboard/mapper_wallet_test.go +++ b/internal/dashboard/mapper_wallet_test.go @@ -123,3 +123,74 @@ func TestUserCreateTimeDisplayOffsetIsApplied(t *testing.T) { t.Fatalf("expected country new user contribution, got %d", contributions[0].CountryNewUser) } } + +func TestGoogleRechargeContributesToUserRecharge(t *testing.T) { + mapper := NewMapper(walletMapperTestStore{users: map[int64]UserCountry{ + 1001: { + UserID: 1001, + SysOrigin: "LIKEI", + Country: Country{Code: "BD", Name: "Bangladesh"}, + CreatedAt: time.Date(2026, 5, 27, 8, 0, 0, 0, time.UTC), + }, + }}, []string{"LIKEI"}, "UNKNOWN", 0) + + contributions, err := mapper.Map(context.Background(), "likei", "order_purchase_history", "insert", Row{ + "id": int64(21), + "user_id": int64(1001), + "sys_origin": "LIKEI", + "evn": "PROD", + "status": "COMPLETE", + "is_trial_period": int64(0), + "unit_price": decimal.NewFromInt(99), + "pay_platform": "GOOGLE", + "create_time": time.Date(2026, 5, 27, 9, 0, 0, 0, time.UTC), + }) + if err != nil { + t.Fatal(err) + } + if len(contributions) != 1 { + t.Fatalf("expected one contribution, got %d", len(contributions)) + } + if !contributions[0].GoogleRecharge.Equal(decimal.NewFromInt(99)) { + t.Fatalf("expected google recharge 99, got %s", contributions[0].GoogleRecharge) + } + if !contributions[0].UserRecharge.Equal(decimal.NewFromInt(99)) { + t.Fatalf("expected user recharge 99, got %s", contributions[0].UserRecharge) + } +} + +func TestThirdPartyRechargeContributesToUserRechargeForAnyFactory(t *testing.T) { + mapper := NewMapper(walletMapperTestStore{users: map[int64]UserCountry{ + 1001: { + UserID: 1001, + SysOrigin: "LIKEI", + Country: Country{Code: "BD", Name: "Bangladesh"}, + CreatedAt: time.Date(2026, 5, 27, 8, 0, 0, 0, time.UTC), + }, + }}, []string{"LIKEI"}, "UNKNOWN", 0) + + contributions, err := mapper.Map(context.Background(), "likei", "order_user_purchase_pay", "insert", Row{ + "id": int64(22), + "user_id": int64(1001), + "sys_origin": "LIKEI", + "evn": "PROD", + "factory_code": "OTHER_PAY", + "pay_status": "SUCCESSFUL", + "receipt_type": "PAYMENT", + "refund_status": "NONE", + "compute_usd_amount": decimal.NewFromInt(12), + "create_time": time.Date(2026, 5, 27, 9, 0, 0, 0, time.UTC), + }) + if err != nil { + t.Fatal(err) + } + if len(contributions) != 1 { + t.Fatalf("expected one contribution, got %d", len(contributions)) + } + if !contributions[0].UserRecharge.Equal(decimal.NewFromInt(12)) { + t.Fatalf("expected user recharge 12, got %s", contributions[0].UserRecharge) + } + if !contributions[0].MifapayRecharge.IsZero() { + t.Fatalf("expected non MifaPay recharge to keep mifapay zero, got %s", contributions[0].MifapayRecharge) + } +} diff --git a/internal/dashboard/redis.go b/internal/dashboard/redis.go index 2def74c..0251585 100644 --- a/internal/dashboard/redis.go +++ b/internal/dashboard/redis.go @@ -24,10 +24,16 @@ func (p *Projector) writeRealtimeRedis(ctx context.Context, contributions []Cont c.Country.Code, ) sign := decimal.NewFromInt(int64(c.DeltaSign)) + newDealerUserRecharge := decimal.Zero + if SamePeriod(c.UserCreatedAt, c.EventTime, PeriodWindow{Type: PeriodDay}, p.storageZone) { + newDealerUserRecharge = c.NewDealerUserRecharge + } hincrDecimal(pipe, ctx, key, "official_recharge", c.OfficialRecharge.Mul(sign)) hincrDecimal(pipe, ctx, key, "mifapay_recharge", c.MifapayRecharge.Mul(sign)) hincrDecimal(pipe, ctx, key, "google_recharge", c.GoogleRecharge.Mul(sign)) hincrDecimal(pipe, ctx, key, "dealer_recharge", c.DealerRecharge.Mul(sign)) + hincrDecimal(pipe, ctx, key, "user_recharge", c.UserRecharge.Mul(sign)) + hincrDecimal(pipe, ctx, key, "new_dealer_user_recharge", newDealerUserRecharge.Mul(sign)) hincrDecimal(pipe, ctx, key, "salary_exchange", c.SalaryExchange.Mul(sign)) hincrDecimal(pipe, ctx, key, "salary_transfer", c.SalaryTransfer.Mul(sign)) hincrDecimal(pipe, ctx, key, "lucky_gift_total_flow", c.LuckyGiftTotalFlow.Mul(sign)) diff --git a/internal/dashboard/types.go b/internal/dashboard/types.go index 32788de..fb8fc1e 100644 --- a/internal/dashboard/types.go +++ b/internal/dashboard/types.go @@ -18,6 +18,7 @@ const ( MetricCountryNewUser = "COUNTRY_NEW_USER" MetricLuckyGiftUser = "LUCKY_GIFT_USER" MetricGameUser = "GAME_USER" + MetricRechargeUser = "RECHARGE_USER" ) type Country struct { @@ -46,6 +47,8 @@ type Contribution struct { MifapayRecharge decimal.Decimal GoogleRecharge decimal.Decimal DealerRecharge decimal.Decimal + UserRecharge decimal.Decimal + NewDealerUserRecharge decimal.Decimal SalaryExchange decimal.Decimal SalaryTransfer decimal.Decimal GiftConsume decimal.Decimal @@ -82,6 +85,8 @@ func (c Contribution) Empty() bool { c.MifapayRecharge.IsZero() && c.GoogleRecharge.IsZero() && c.DealerRecharge.IsZero() && + c.UserRecharge.IsZero() && + c.NewDealerUserRecharge.IsZero() && c.SalaryExchange.IsZero() && c.SalaryTransfer.IsZero() && c.GiftConsume.IsZero() && diff --git a/internal/storage/active_reconcile.go b/internal/storage/active_reconcile.go index 5e7e236..f0ef7a6 100644 --- a/internal/storage/active_reconcile.go +++ b/internal/storage/active_reconcile.go @@ -611,9 +611,9 @@ 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, country_new_user, daily_active_user, new_user_recharge, official_recharge, mifapay_recharge, - google_recharge, dealer_recharge, salary_exchange, salary_transfer, gift_consume, + google_recharge, dealer_recharge, user_recharge, new_dealer_user_recharge, salary_exchange, salary_transfer, gift_consume, lucky_gift_total_flow, lucky_gift_user, lucky_gift_payout, lucky_gift_anchor_share, - game_total_flow, game_user, game_payout, + game_total_flow, daily_recharge_user, game_user, game_payout, d1_retention_user, d1_retention_base_user, d7_retention_user, d7_retention_base_user, d30_retention_user, d30_retention_base_user, refreshed_at ) @@ -626,6 +626,8 @@ SELECT 'ALL', 'ALL', ?, '全部', NULL, NULL, IFNULL(SUM(d.mifapay_recharge), 0), IFNULL(SUM(d.google_recharge), 0), IFNULL(SUM(d.dealer_recharge), 0), + IFNULL(SUM(d.user_recharge), 0), + IFNULL(SUM(d.new_dealer_user_recharge), 0), IFNULL(SUM(d.salary_exchange), 0), IFNULL(SUM(d.salary_transfer), 0), IFNULL(SUM(d.gift_consume), 0), @@ -634,6 +636,7 @@ SELECT 'ALL', 'ALL', ?, '全部', NULL, NULL, IFNULL(SUM(d.lucky_gift_payout), 0), IFNULL(SUM(d.lucky_gift_anchor_share), 0), IFNULL(SUM(d.game_total_flow), 0), + IFNULL(MAX(u.daily_recharge_user), 0), IFNULL(MAX(u.game_user), 0), IFNULL(SUM(d.game_payout), 0), IFNULL(SUM(d.d1_retention_user), 0), @@ -648,6 +651,7 @@ LEFT JOIN ( SELECT country_code, SUM(metric_type = 'COUNTRY_NEW_USER') AS country_new_user, SUM(metric_type = 'LUCKY_GIFT_USER') AS lucky_gift_user, + SUM(metric_type = 'RECHARGE_USER') AS daily_recharge_user, SUM(metric_type = 'GAME_USER') AS game_user FROM country_dashboard_period_user_metric WHERE period_type = 'ALL' AND period_key = 'ALL' AND stat_timezone = ? AND sys_origin = ? diff --git a/internal/storage/aggregate.go b/internal/storage/aggregate.go index af55faa..7862eff 100644 --- a/internal/storage/aggregate.go +++ b/internal/storage/aggregate.go @@ -17,10 +17,12 @@ func (r *MySQLRepository) ApplyDailyContribution(ctx context.Context, tx *sql.Tx } day := dashboard.DailyDate(c.EventTime, zone) newUserRecharge := decimal.Zero + newDealerUserRecharge := decimal.Zero if dashboard.SamePeriod(c.UserCreatedAt, c.EventTime, dashboard.PeriodWindow{Type: dashboard.PeriodDay}, zone) { newUserRecharge = c.NewUserRechargeCandidate + newDealerUserRecharge = c.NewDealerUserRecharge } - if err := r.upsertDailyAmounts(ctx, tx, c, day, newUserRecharge); err != nil { + if err := r.upsertDailyAmounts(ctx, tx, c, day, newUserRecharge, newDealerUserRecharge); err != nil { return err } if c.CountryNewUser > 0 { @@ -38,6 +40,11 @@ func (r *MySQLRepository) ApplyDailyContribution(ctx context.Context, tx *sql.Tx return err } } + if c.UserRecharge.GreaterThan(decimal.Zero) { + if err := r.applyDailyUserMetric(ctx, tx, c, day, dashboard.MetricRechargeUser); err != nil { + return err + } + } return nil } @@ -48,10 +55,12 @@ func (r *MySQLRepository) ApplyPeriodContribution(ctx context.Context, tx *sql.T } for _, window := range dashboard.PeriodWindows(c.EventTime, zone) { newUserRecharge := decimal.Zero + newDealerUserRecharge := decimal.Zero if dashboard.SamePeriod(c.UserCreatedAt, c.EventTime, window, zone) { newUserRecharge = c.NewUserRechargeCandidate + newDealerUserRecharge = c.NewDealerUserRecharge } - if err := r.upsertPeriodAmounts(ctx, tx, c, window, statTimezone, newUserRecharge); err != nil { + if err := r.upsertPeriodAmounts(ctx, tx, c, window, statTimezone, newUserRecharge, newDealerUserRecharge); err != nil { return err } if c.CountryNewUser > 0 { @@ -69,6 +78,11 @@ func (r *MySQLRepository) ApplyPeriodContribution(ctx context.Context, tx *sql.T return err } } + if c.UserRecharge.GreaterThan(decimal.Zero) { + if err := r.applyPeriodUserMetric(ctx, tx, c, window, statTimezone, dashboard.MetricRechargeUser); err != nil { + return err + } + } } return nil } @@ -94,15 +108,15 @@ func (r *MySQLRepository) ApplyGamePeriodContribution(ctx context.Context, tx *s return nil } -func (r *MySQLRepository) upsertDailyAmounts(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, day time.Time, newUserRecharge decimal.Decimal) error { +func (r *MySQLRepository) upsertDailyAmounts(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, day time.Time, newUserRecharge decimal.Decimal, newDealerUserRecharge decimal.Decimal) error { sign := decimal.NewFromInt(int64(c.DeltaSign)) _, err := tx.ExecContext(ctx, ` 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, salary_transfer, gift_consume, lucky_gift_total_flow, lucky_gift_payout, + dealer_recharge, user_recharge, new_dealer_user_recharge, salary_exchange, salary_transfer, 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), @@ -110,6 +124,8 @@ ON DUPLICATE KEY UPDATE mifapay_recharge = mifapay_recharge + VALUES(mifapay_recharge), google_recharge = google_recharge + VALUES(google_recharge), dealer_recharge = dealer_recharge + VALUES(dealer_recharge), + user_recharge = user_recharge + VALUES(user_recharge), + new_dealer_user_recharge = new_dealer_user_recharge + VALUES(new_dealer_user_recharge), salary_exchange = salary_exchange + VALUES(salary_exchange), salary_transfer = salary_transfer + VALUES(salary_transfer), gift_consume = gift_consume + VALUES(gift_consume), @@ -120,20 +136,20 @@ 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.SalaryTransfer.Mul(sign), c.GiftConsume.Mul(sign), c.LuckyGiftTotalFlow.Mul(sign), c.LuckyGiftPayout.Mul(sign), + c.DealerRecharge.Mul(sign), c.UserRecharge.Mul(sign), newDealerUserRecharge.Mul(sign), c.SalaryExchange.Mul(sign), c.SalaryTransfer.Mul(sign), c.GiftConsume.Mul(sign), c.LuckyGiftTotalFlow.Mul(sign), c.LuckyGiftPayout.Mul(sign), c.GameTotalFlow.Mul(sign), c.GamePayout.Mul(sign)) return err } -func (r *MySQLRepository) upsertPeriodAmounts(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, window dashboard.PeriodWindow, statTimezone string, newUserRecharge decimal.Decimal) error { +func (r *MySQLRepository) upsertPeriodAmounts(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, window dashboard.PeriodWindow, statTimezone string, newUserRecharge decimal.Decimal, newDealerUserRecharge decimal.Decimal) error { sign := decimal.NewFromInt(int64(c.DeltaSign)) _, 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, new_user_recharge, official_recharge, mifapay_recharge, - google_recharge, dealer_recharge, salary_exchange, salary_transfer, gift_consume, lucky_gift_total_flow, lucky_gift_payout, + google_recharge, dealer_recharge, user_recharge, new_dealer_user_recharge, salary_exchange, salary_transfer, 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), @@ -144,6 +160,8 @@ ON DUPLICATE KEY UPDATE mifapay_recharge = mifapay_recharge + VALUES(mifapay_recharge), google_recharge = google_recharge + VALUES(google_recharge), dealer_recharge = dealer_recharge + VALUES(dealer_recharge), + user_recharge = user_recharge + VALUES(user_recharge), + new_dealer_user_recharge = new_dealer_user_recharge + VALUES(new_dealer_user_recharge), salary_exchange = salary_exchange + VALUES(salary_exchange), salary_transfer = salary_transfer + VALUES(salary_transfer), gift_consume = gift_consume + VALUES(gift_consume), @@ -152,9 +170,9 @@ ON DUPLICATE KEY UPDATE game_total_flow = game_total_flow + VALUES(game_total_flow), game_payout = game_payout + VALUES(game_payout), refreshed_at = NOW() -`, window.Type, window.Key, statTimezone, window.Name, window.StartDate, window.EndDate, + `, 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.MifapayRecharge.Mul(sign), c.GoogleRecharge.Mul(sign), c.DealerRecharge.Mul(sign), c.UserRecharge.Mul(sign), newDealerUserRecharge.Mul(sign), c.SalaryExchange.Mul(sign), c.SalaryTransfer.Mul(sign), c.GiftConsume.Mul(sign), c.LuckyGiftTotalFlow.Mul(sign), c.LuckyGiftPayout.Mul(sign), c.GameTotalFlow.Mul(sign), c.GamePayout.Mul(sign)) return err diff --git a/internal/storage/batch.go b/internal/storage/batch.go index 83c41dc..81f1fd2 100644 --- a/internal/storage/batch.go +++ b/internal/storage/batch.go @@ -14,18 +14,20 @@ import ( 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 + 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 { @@ -150,12 +152,14 @@ func (r *MySQLRepository) ApplyContributionBatch(ctx context.Context, tx *sql.Tx } 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) + totals.add(c, newUserRecharge, newDealerUserRecharge) dailyAmounts[dailyKey] = totals addDailyUserGroups(dailyUsers, c, day) @@ -163,8 +167,10 @@ func (r *MySQLRepository) ApplyContributionBatch(ctx context.Context, tx *sql.Tx 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, @@ -172,7 +178,7 @@ func (r *MySQLRepository) ApplyContributionBatch(ctx context.Context, tx *sql.Tx sysOrigin: c.SysOrigin, countryCode: c.Country.Code, countryName: c.Country.Name, } periodTotal := periodAmounts[periodKey] - periodTotal.add(c, newUserRecharge) + periodTotal.add(c, newUserRecharge, newDealerUserRecharge) periodAmounts[periodKey] = periodTotal addPeriodUserGroups(periodUsers, c, window, statTimezone) @@ -250,13 +256,15 @@ func (r *MySQLRepository) ApplyContributionBatch(ctx context.Context, tx *sql.Tx return nil } -func (t *amountTotals) add(c dashboard.Contribution, newUserRecharge decimal.Decimal) { +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)) @@ -268,7 +276,8 @@ 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.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() @@ -299,6 +308,9 @@ func addDailyUserGroups(groups map[dailyUserKey]map[int64]struct{}, c dashboard. 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) { @@ -314,6 +326,9 @@ func addPeriodUserGroups(groups map[periodUserKey]map[int64]struct{}, c dashboar 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) { diff --git a/internal/storage/batch_sql.go b/internal/storage/batch_sql.go index 1057452..9a051bd 100644 --- a/internal/storage/batch_sql.go +++ b/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, salary_transfer, gift_consume, lucky_gift_total_flow, lucky_gift_payout, + dealer_recharge, user_recharge, new_dealer_user_recharge, salary_exchange, salary_transfer, 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), @@ -21,6 +21,8 @@ ON DUPLICATE KEY UPDATE mifapay_recharge = mifapay_recharge + VALUES(mifapay_recharge), google_recharge = google_recharge + VALUES(google_recharge), dealer_recharge = dealer_recharge + VALUES(dealer_recharge), + user_recharge = user_recharge + VALUES(user_recharge), + new_dealer_user_recharge = new_dealer_user_recharge + VALUES(new_dealer_user_recharge), salary_exchange = salary_exchange + VALUES(salary_exchange), salary_transfer = salary_transfer + VALUES(salary_transfer), gift_consume = gift_consume + VALUES(gift_consume), @@ -31,7 +33,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.salaryTransfer, totals.giftConsume, totals.luckyGiftTotalFlow, totals.luckyGiftPayout, + totals.dealerRecharge, totals.userRecharge, totals.newDealerUserRecharge, totals.salaryExchange, totals.salaryTransfer, totals.giftConsume, totals.luckyGiftTotalFlow, totals.luckyGiftPayout, totals.gameTotalFlow, totals.gamePayout) return err } @@ -41,9 +43,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, salary_transfer, gift_consume, lucky_gift_total_flow, lucky_gift_payout, + google_recharge, dealer_recharge, user_recharge, new_dealer_user_recharge, salary_exchange, salary_transfer, 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), @@ -54,6 +56,8 @@ ON DUPLICATE KEY UPDATE mifapay_recharge = mifapay_recharge + VALUES(mifapay_recharge), google_recharge = google_recharge + VALUES(google_recharge), dealer_recharge = dealer_recharge + VALUES(dealer_recharge), + user_recharge = user_recharge + VALUES(user_recharge), + new_dealer_user_recharge = new_dealer_user_recharge + VALUES(new_dealer_user_recharge), salary_exchange = salary_exchange + VALUES(salary_exchange), salary_transfer = salary_transfer + VALUES(salary_transfer), gift_consume = gift_consume + VALUES(gift_consume), @@ -62,9 +66,9 @@ ON DUPLICATE KEY UPDATE game_total_flow = game_total_flow + VALUES(game_total_flow), game_payout = game_payout + VALUES(game_payout), refreshed_at = NOW() -`, key.periodType, key.periodKey, key.statTimezone, key.periodName, ptrOrNil(key.startDate), ptrOrNil(key.endDate), + `, 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.mifapayRecharge, totals.googleRecharge, totals.dealerRecharge, totals.userRecharge, totals.newDealerUserRecharge, totals.salaryExchange, totals.salaryTransfer, totals.giftConsume, totals.luckyGiftTotalFlow, totals.luckyGiftPayout, totals.gameTotalFlow, totals.gamePayout) return err diff --git a/internal/storage/recharge_reconcile.go b/internal/storage/recharge_reconcile.go index e124a56..1937c1b 100644 --- a/internal/storage/recharge_reconcile.go +++ b/internal/storage/recharge_reconcile.go @@ -30,12 +30,13 @@ type officialRechargeRow struct { } type mifapayRechargeRow struct { - id string - userID int64 - sysOrigin string - amount decimal.Decimal - eventTime time.Time - user dashboard.UserCountry + id string + userID int64 + sysOrigin string + factoryCode string + amount decimal.Decimal + eventTime time.Time + user dashboard.UserCountry } func (r *MySQLRepository) BuildRechargeReconcileEvents(ctx context.Context, since time.Time, @@ -79,6 +80,7 @@ func (r *MySQLRepository) BuildRechargeReconcileEvents(ctx context.Context, sinc } if row.payPlatform == "GOOGLE" { contribution.GoogleRecharge = row.unitPrice + contribution.UserRecharge = row.unitPrice } events = append(events, dashboard.Event{ EventKey: rechargeReconcileEventKey(reconcileOfficialSource, row.id), @@ -102,9 +104,12 @@ func (r *MySQLRepository) BuildRechargeReconcileEvents(ctx context.Context, sinc EventTime: row.eventTime, UserCreatedAt: row.user.CreatedAt, NewUserRechargeCandidate: row.amount, - MifapayRecharge: row.amount, + UserRecharge: row.amount, RechargeDetail: mifapayRechargeDetail(row), } + if row.factoryCode == "MIFA_PAY" { + contribution.MifapayRecharge = row.amount + } events = append(events, dashboard.Event{ EventKey: rechargeReconcileEventKey(reconcileMifapaySource, row.id), Source: reconcileMifapaySource, @@ -193,7 +198,7 @@ LIMIT ?`, r.cfg.Dashboard.UnknownCountryCode, r.cfg.Dashboard.UnknownCountryCode func (r *MySQLRepository) listMifapayRechargeRows(ctx context.Context, since time.Time, limit int) ([]mifapayRechargeRow, error) { rows, err := r.db.QueryContext(ctx, ` -SELECT CAST(o.id AS CHAR), o.user_id, o.sys_origin, o.compute_usd_amount, o.create_time, +SELECT CAST(o.id AS CHAR), o.user_id, o.sys_origin, o.factory_code, o.compute_usd_amount, o.create_time, 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 @@ -201,7 +206,6 @@ FROM order_user_purchase_pay o JOIN user_base_info u ON u.id = o.user_id WHERE o.create_time >= ? AND o.evn = 'PROD' - AND o.factory_code = 'MIFA_PAY' AND o.pay_status = 'SUCCESSFUL' AND o.receipt_type = 'PAYMENT' AND o.refund_status = 'NONE' @@ -217,7 +221,7 @@ LIMIT ?`, r.cfg.Dashboard.UnknownCountryCode, r.cfg.Dashboard.UnknownCountryCode var countryCode, countryName string var isDeleted bool var userCreatedAt, userUpdatedAt sql.NullTime - if err := rows.Scan(&item.id, &item.userID, &item.sysOrigin, &item.amount, + if err := rows.Scan(&item.id, &item.userID, &item.sysOrigin, &item.factoryCode, &item.amount, &item.eventTime, &item.user.UserID, &item.user.SysOrigin, &countryCode, &countryName, &isDeleted, &userCreatedAt, &userUpdatedAt); err != nil { return nil, err @@ -272,10 +276,14 @@ func officialRechargeDetail(row officialRechargeRow) *dashboard.RechargeDetail { } func mifapayRechargeDetail(row mifapayRechargeRow) *dashboard.RechargeDetail { + sourceType := "THIRD_PARTY" + if row.factoryCode == "MIFA_PAY" { + sourceType = "MIFAPAY" + } return &dashboard.RechargeDetail{ - SourceType: "MIFAPAY", + SourceType: sourceType, RecordID: row.id, - SourceChannel: "MIFA_PAY", + SourceChannel: row.factoryCode, SysOrigin: row.user.SysOrigin, UserID: row.userID, Country: row.user.Country, diff --git a/internal/storage/schema.go b/internal/storage/schema.go index 0dd59ce..30cd03a 100644 --- a/internal/storage/schema.go +++ b/internal/storage/schema.go @@ -64,6 +64,8 @@ const createDailyMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_dai mifapay_recharge decimal(20,2) NOT NULL DEFAULT 0, google_recharge decimal(20,2) NOT NULL DEFAULT 0, dealer_recharge decimal(20,2) NOT NULL DEFAULT 0, + user_recharge decimal(20,2) NOT NULL DEFAULT 0, + new_dealer_user_recharge decimal(20,2) NOT NULL DEFAULT 0, salary_exchange decimal(20,2) NOT NULL DEFAULT 0, salary_transfer decimal(20,2) NOT NULL DEFAULT 0, gift_consume decimal(20,2) NOT NULL DEFAULT 0, @@ -71,6 +73,7 @@ const createDailyMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_dai lucky_gift_user bigint NOT NULL DEFAULT 0, lucky_gift_payout decimal(20,2) NOT NULL DEFAULT 0, game_total_flow decimal(20,2) NOT NULL DEFAULT 0, + daily_recharge_user bigint NOT NULL DEFAULT 0, game_user bigint NOT NULL DEFAULT 0, game_payout decimal(20,2) NOT NULL DEFAULT 0, refreshed_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, @@ -110,6 +113,8 @@ const createPeriodMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_pe mifapay_recharge decimal(20,2) NOT NULL DEFAULT 0, google_recharge decimal(20,2) NOT NULL DEFAULT 0, dealer_recharge decimal(20,2) NOT NULL DEFAULT 0, + user_recharge decimal(20,2) NOT NULL DEFAULT 0, + new_dealer_user_recharge decimal(20,2) NOT NULL DEFAULT 0, salary_exchange decimal(20,2) NOT NULL DEFAULT 0, salary_transfer decimal(20,2) NOT NULL DEFAULT 0, gift_consume decimal(20,2) NOT NULL DEFAULT 0, @@ -117,6 +122,7 @@ const createPeriodMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_pe lucky_gift_user bigint NOT NULL DEFAULT 0, lucky_gift_payout decimal(20,2) NOT NULL DEFAULT 0, game_total_flow decimal(20,2) NOT NULL DEFAULT 0, + daily_recharge_user bigint NOT NULL DEFAULT 0, game_user bigint NOT NULL DEFAULT 0, game_payout decimal(20,2) NOT NULL DEFAULT 0, refreshed_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, @@ -134,6 +140,30 @@ func (r *MySQLRepository) ensureMetricColumns(ctx context.Context) error { "ALTER TABLE country_dashboard_period_metric ADD COLUMN salary_transfer decimal(20,2) NOT NULL DEFAULT 0 AFTER salary_exchange"); err != nil { return err } + if err := r.addColumnIfMissing(ctx, "country_dashboard_daily_metric", "user_recharge", + "ALTER TABLE country_dashboard_daily_metric ADD COLUMN user_recharge decimal(20,2) NOT NULL DEFAULT 0 AFTER dealer_recharge"); err != nil { + return err + } + if err := r.addColumnIfMissing(ctx, "country_dashboard_period_metric", "user_recharge", + "ALTER TABLE country_dashboard_period_metric ADD COLUMN user_recharge decimal(20,2) NOT NULL DEFAULT 0 AFTER dealer_recharge"); err != nil { + return err + } + if err := r.addColumnIfMissing(ctx, "country_dashboard_daily_metric", "new_dealer_user_recharge", + "ALTER TABLE country_dashboard_daily_metric ADD COLUMN new_dealer_user_recharge decimal(20,2) NOT NULL DEFAULT 0 AFTER user_recharge"); err != nil { + return err + } + if err := r.addColumnIfMissing(ctx, "country_dashboard_period_metric", "new_dealer_user_recharge", + "ALTER TABLE country_dashboard_period_metric ADD COLUMN new_dealer_user_recharge decimal(20,2) NOT NULL DEFAULT 0 AFTER user_recharge"); err != nil { + return err + } + if err := r.addColumnIfMissing(ctx, "country_dashboard_daily_metric", "daily_recharge_user", + "ALTER TABLE country_dashboard_daily_metric ADD COLUMN daily_recharge_user bigint NOT NULL DEFAULT 0 AFTER game_total_flow"); err != nil { + return err + } + if err := r.addColumnIfMissing(ctx, "country_dashboard_period_metric", "daily_recharge_user", + "ALTER TABLE country_dashboard_period_metric ADD COLUMN daily_recharge_user bigint NOT NULL DEFAULT 0 AFTER game_total_flow"); err != nil { + return err + } return r.addIndexIfMissing(ctx, "dashboard_cdc_user_country", "idx_dashboard_cdc_user_country_origin_created", "ALTER TABLE dashboard_cdc_user_country ADD INDEX idx_dashboard_cdc_user_country_origin_created (sys_origin, user_created_at, country_code)") } diff --git a/internal/storage/user_metrics.go b/internal/storage/user_metrics.go index 2719566..95be333 100644 --- a/internal/storage/user_metrics.go +++ b/internal/storage/user_metrics.go @@ -178,6 +178,8 @@ func dailyMetricColumn(metricType string) string { return "lucky_gift_user" case dashboard.MetricGameUser: return "game_user" + case dashboard.MetricRechargeUser: + return "daily_recharge_user" default: return "" } diff --git a/migrations/002_dashboard_recharge_user_metrics.sql b/migrations/002_dashboard_recharge_user_metrics.sql new file mode 100644 index 0000000..f2b6b05 --- /dev/null +++ b/migrations/002_dashboard_recharge_user_metrics.sql @@ -0,0 +1,85 @@ +SET @current_schema = DATABASE(); + +SET @ddl = IF( + NOT EXISTS ( + SELECT 1 FROM INFORMATION_SCHEMA.COLUMNS + WHERE TABLE_SCHEMA = @current_schema + AND TABLE_NAME = 'country_dashboard_daily_metric' + AND COLUMN_NAME = 'daily_recharge_user' + ), + 'ALTER TABLE `country_dashboard_daily_metric` ADD COLUMN `daily_recharge_user` bigint NOT NULL DEFAULT 0 COMMENT ''当日充值用户数'' AFTER `country_new_user`', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl = IF( + NOT EXISTS ( + SELECT 1 FROM INFORMATION_SCHEMA.COLUMNS + WHERE TABLE_SCHEMA = @current_schema + AND TABLE_NAME = 'country_dashboard_period_metric' + AND COLUMN_NAME = 'daily_recharge_user' + ), + 'ALTER TABLE `country_dashboard_period_metric` ADD COLUMN `daily_recharge_user` bigint NOT NULL DEFAULT 0 COMMENT ''当日充值用户数'' AFTER `daily_active_user`', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl = IF( + NOT EXISTS ( + SELECT 1 FROM INFORMATION_SCHEMA.COLUMNS + WHERE TABLE_SCHEMA = @current_schema + AND TABLE_NAME = 'country_dashboard_daily_metric' + AND COLUMN_NAME = 'user_recharge' + ), + 'ALTER TABLE `country_dashboard_daily_metric` ADD COLUMN `user_recharge` decimal(20,2) NOT NULL DEFAULT 0.00 COMMENT ''用户充值'' AFTER `dealer_recharge`', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl = IF( + NOT EXISTS ( + SELECT 1 FROM INFORMATION_SCHEMA.COLUMNS + WHERE TABLE_SCHEMA = @current_schema + AND TABLE_NAME = 'country_dashboard_period_metric' + AND COLUMN_NAME = 'user_recharge' + ), + 'ALTER TABLE `country_dashboard_period_metric` ADD COLUMN `user_recharge` decimal(20,2) NOT NULL DEFAULT 0.00 COMMENT ''用户充值'' AFTER `dealer_recharge`', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl = IF( + NOT EXISTS ( + SELECT 1 FROM INFORMATION_SCHEMA.COLUMNS + WHERE TABLE_SCHEMA = @current_schema + AND TABLE_NAME = 'country_dashboard_daily_metric' + AND COLUMN_NAME = 'new_dealer_user_recharge' + ), + 'ALTER TABLE `country_dashboard_daily_metric` ADD COLUMN `new_dealer_user_recharge` decimal(20,2) NOT NULL DEFAULT 0.00 COMMENT ''新增币商用户充值'' AFTER `user_recharge`', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl = IF( + NOT EXISTS ( + SELECT 1 FROM INFORMATION_SCHEMA.COLUMNS + WHERE TABLE_SCHEMA = @current_schema + AND TABLE_NAME = 'country_dashboard_period_metric' + AND COLUMN_NAME = 'new_dealer_user_recharge' + ), + 'ALTER TABLE `country_dashboard_period_metric` ADD COLUMN `new_dealer_user_recharge` decimal(20,2) NOT NULL DEFAULT 0.00 COMMENT ''新增币商用户充值'' AFTER `user_recharge`', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt;