From af1fbde1ac2d374ce6945b22cba9e1854fe9b048 Mon Sep 17 00:00:00 2001 From: zhx Date: Mon, 25 May 2026 13:23:34 +0800 Subject: [PATCH] Support salary transfer dashboard metrics --- internal/dashboard/mapper.go | 17 +++++++--- internal/dashboard/redis.go | 1 + internal/dashboard/types.go | 2 ++ internal/storage/aggregate.go | 13 +++++--- internal/storage/batch.go | 3 ++ internal/storage/batch_sql.go | 13 +++++--- internal/storage/schema.go | 58 +++++++++++++++++++++++++++++++++-- 7 files changed, 89 insertions(+), 18 deletions(-) diff --git a/internal/dashboard/mapper.go b/internal/dashboard/mapper.go index a92663c..b164a41 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 "user_freight_balance_running_water": return m.mapDealerRecharge(ctx, schema, table, action, row) case "user_salary_account_running_water", "user_salary_diamond_running_water": - return m.mapSalaryExchange(ctx, schema, table, action, row) + return m.mapSalary(ctx, schema, table, action, row) default: if strings.HasPrefix(table, "wallet_gold_asset_record_") { return m.mapWalletGame(ctx, schema, table, action, row) @@ -126,17 +126,24 @@ func (m *Mapper) mapDealerRecharge(ctx context.Context, schema string, table str return items, nil } -func (m *Mapper) mapSalaryExchange(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { +func (m *Mapper) mapSalary(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { if row.Int64("type") != 1 { return nil, nil } event := row.String("salary_event") - if event != "SALARY_EXCHANGE" && event != "SALARY_EXCHANGE_GOLD_COINS" && - event != "EXCHANGE_GOLD_COINS" && event != "BILL_EXCHANGE_GOLD_COINS" { + isExchange := event == "SALARY_EXCHANGE" || event == "SALARY_EXCHANGE_GOLD_COINS" || + event == "EXCHANGE_GOLD_COINS" || event == "BILL_EXCHANGE_GOLD_COINS" + isTransfer := event == "SALARY_TRANSFER" + if !isExchange && !isTransfer { return nil, nil } return m.amountContribution(ctx, schema, table, action, row, row.Decimal("amount"), func(c *Contribution, amount decimal.Decimal) { - c.SalaryExchange = amount + if isExchange { + c.SalaryExchange = amount + } + if isTransfer { + c.SalaryTransfer = amount + } }) } diff --git a/internal/dashboard/redis.go b/internal/dashboard/redis.go index ff390a2..2def74c 100644 --- a/internal/dashboard/redis.go +++ b/internal/dashboard/redis.go @@ -29,6 +29,7 @@ func (p *Projector) writeRealtimeRedis(ctx context.Context, contributions []Cont hincrDecimal(pipe, ctx, key, "google_recharge", c.GoogleRecharge.Mul(sign)) hincrDecimal(pipe, ctx, key, "dealer_recharge", c.DealerRecharge.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)) hincrDecimal(pipe, ctx, key, "lucky_gift_payout", c.LuckyGiftPayout.Mul(sign)) hincrDecimal(pipe, ctx, key, "game_total_flow", c.GameTotalFlow.Mul(sign)) diff --git a/internal/dashboard/types.go b/internal/dashboard/types.go index 98429ac..32788de 100644 --- a/internal/dashboard/types.go +++ b/internal/dashboard/types.go @@ -47,6 +47,7 @@ type Contribution struct { GoogleRecharge decimal.Decimal DealerRecharge decimal.Decimal SalaryExchange decimal.Decimal + SalaryTransfer decimal.Decimal GiftConsume decimal.Decimal LuckyGiftTotalFlow decimal.Decimal LuckyGiftUser bool @@ -82,6 +83,7 @@ func (c Contribution) Empty() bool { c.GoogleRecharge.IsZero() && c.DealerRecharge.IsZero() && c.SalaryExchange.IsZero() && + c.SalaryTransfer.IsZero() && c.GiftConsume.IsZero() && c.LuckyGiftTotalFlow.IsZero() && !c.LuckyGiftUser && diff --git a/internal/storage/aggregate.go b/internal/storage/aggregate.go index 8e9aa8d..05cb261 100644 --- a/internal/storage/aggregate.go +++ b/internal/storage/aggregate.go @@ -100,7 +100,7 @@ func (r *MySQLRepository) upsertDailyAmounts(ctx context.Context, tx *sql.Tx, c 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, gift_consume, lucky_gift_total_flow, lucky_gift_payout, + dealer_recharge, salary_exchange, salary_transfer, gift_consume, lucky_gift_total_flow, lucky_gift_payout, game_total_flow, game_payout, refreshed_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()) ON DUPLICATE KEY UPDATE @@ -111,6 +111,7 @@ ON DUPLICATE KEY UPDATE google_recharge = google_recharge + VALUES(google_recharge), dealer_recharge = dealer_recharge + VALUES(dealer_recharge), salary_exchange = salary_exchange + VALUES(salary_exchange), + salary_transfer = salary_transfer + VALUES(salary_transfer), gift_consume = gift_consume + VALUES(gift_consume), lucky_gift_total_flow = lucky_gift_total_flow + VALUES(lucky_gift_total_flow), lucky_gift_payout = lucky_gift_payout + VALUES(lucky_gift_payout), @@ -119,7 +120,7 @@ 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.GiftConsume.Mul(sign), c.LuckyGiftTotalFlow.Mul(sign), c.LuckyGiftPayout.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.GameTotalFlow.Mul(sign), c.GamePayout.Mul(sign)) return err } @@ -130,9 +131,9 @@ func (r *MySQLRepository) upsertPeriodAmounts(ctx context.Context, tx *sql.Tx, c 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, gift_consume, lucky_gift_total_flow, lucky_gift_payout, + google_recharge, dealer_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 +145,7 @@ ON DUPLICATE KEY UPDATE google_recharge = google_recharge + VALUES(google_recharge), dealer_recharge = dealer_recharge + VALUES(dealer_recharge), salary_exchange = salary_exchange + VALUES(salary_exchange), + salary_transfer = salary_transfer + VALUES(salary_transfer), gift_consume = gift_consume + VALUES(gift_consume), lucky_gift_total_flow = lucky_gift_total_flow + VALUES(lucky_gift_total_flow), lucky_gift_payout = lucky_gift_payout + VALUES(lucky_gift_payout), @@ -153,7 +155,8 @@ ON DUPLICATE KEY UPDATE `, 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.GiftConsume.Mul(sign), c.LuckyGiftTotalFlow.Mul(sign), c.LuckyGiftPayout.Mul(sign), c.GameTotalFlow.Mul(sign), c.GamePayout.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 12cf67f..bf9a00f 100644 --- a/internal/storage/batch.go +++ b/internal/storage/batch.go @@ -20,6 +20,7 @@ type amountTotals struct { googleRecharge decimal.Decimal dealerRecharge decimal.Decimal salaryExchange decimal.Decimal + salaryTransfer decimal.Decimal giftConsume decimal.Decimal luckyGiftTotalFlow decimal.Decimal luckyGiftPayout decimal.Decimal @@ -253,6 +254,7 @@ func (t *amountTotals) add(c dashboard.Contribution, newUserRecharge decimal.Dec 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)) @@ -263,6 +265,7 @@ 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.salaryTransfer.IsZero() && t.giftConsume.IsZero() && t.luckyGiftTotalFlow.IsZero() && t.luckyGiftPayout.IsZero() && t.gameTotalFlow.IsZero() && t.gamePayout.IsZero() } diff --git a/internal/storage/batch_sql.go b/internal/storage/batch_sql.go index 9877710..9386719 100644 --- a/internal/storage/batch_sql.go +++ b/internal/storage/batch_sql.go @@ -11,7 +11,7 @@ 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, gift_consume, lucky_gift_total_flow, lucky_gift_payout, + dealer_recharge, salary_exchange, salary_transfer, gift_consume, lucky_gift_total_flow, lucky_gift_payout, game_total_flow, game_payout, refreshed_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()) ON DUPLICATE KEY UPDATE @@ -22,6 +22,7 @@ ON DUPLICATE KEY UPDATE google_recharge = google_recharge + VALUES(google_recharge), dealer_recharge = dealer_recharge + VALUES(dealer_recharge), salary_exchange = salary_exchange + VALUES(salary_exchange), + salary_transfer = salary_transfer + VALUES(salary_transfer), gift_consume = gift_consume + VALUES(gift_consume), lucky_gift_total_flow = lucky_gift_total_flow + VALUES(lucky_gift_total_flow), lucky_gift_payout = lucky_gift_payout + VALUES(lucky_gift_payout), @@ -30,7 +31,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.giftConsume, totals.luckyGiftTotalFlow, totals.luckyGiftPayout, + totals.dealerRecharge, totals.salaryExchange, totals.salaryTransfer, totals.giftConsume, totals.luckyGiftTotalFlow, totals.luckyGiftPayout, totals.gameTotalFlow, totals.gamePayout) return err } @@ -40,9 +41,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, gift_consume, lucky_gift_total_flow, lucky_gift_payout, + google_recharge, dealer_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 +55,7 @@ ON DUPLICATE KEY UPDATE google_recharge = google_recharge + VALUES(google_recharge), dealer_recharge = dealer_recharge + VALUES(dealer_recharge), salary_exchange = salary_exchange + VALUES(salary_exchange), + salary_transfer = salary_transfer + VALUES(salary_transfer), gift_consume = gift_consume + VALUES(gift_consume), lucky_gift_total_flow = lucky_gift_total_flow + VALUES(lucky_gift_total_flow), lucky_gift_payout = lucky_gift_payout + VALUES(lucky_gift_payout), @@ -63,7 +65,8 @@ ON DUPLICATE KEY UPDATE `, 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.giftConsume, totals.luckyGiftTotalFlow, totals.luckyGiftPayout, totals.gameTotalFlow, totals.gamePayout) + totals.salaryTransfer, totals.giftConsume, totals.luckyGiftTotalFlow, totals.luckyGiftPayout, + totals.gameTotalFlow, totals.gamePayout) return err } diff --git a/internal/storage/schema.go b/internal/storage/schema.go index 9323d2c..3560ce1 100644 --- a/internal/storage/schema.go +++ b/internal/storage/schema.go @@ -1,9 +1,15 @@ package storage -import "context" +import ( + "context" + "errors" + "fmt" + + driverMysql "github.com/go-sql-driver/mysql" +) func (r *MySQLRepository) EnsureSchema(ctx context.Context) error { - return execAll(ctx, r.db, []string{ + if err := execAll(ctx, r.db, []string{ `CREATE TABLE IF NOT EXISTS dashboard_cdc_checkpoint ( service_name varchar(64) NOT NULL, binlog_file varchar(255) NOT NULL, @@ -37,7 +43,10 @@ func (r *MySQLRepository) EnsureSchema(ctx context.Context) error { createGamePeriodMetricTable, createGamePeriodUserMetricTable, createRechargeDetailTable, - }) + }); err != nil { + return err + } + return r.ensureMetricColumns(ctx) } const createDailyMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_daily_metric ( @@ -54,6 +63,7 @@ const createDailyMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_dai google_recharge decimal(20,2) NOT NULL DEFAULT 0, dealer_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, lucky_gift_total_flow decimal(20,2) NOT NULL DEFAULT 0, lucky_gift_user bigint NOT NULL DEFAULT 0, @@ -99,6 +109,7 @@ const createPeriodMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_pe google_recharge decimal(20,2) NOT NULL DEFAULT 0, dealer_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, lucky_gift_total_flow decimal(20,2) NOT NULL DEFAULT 0, lucky_gift_user bigint NOT NULL DEFAULT 0, @@ -112,6 +123,47 @@ const createPeriodMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_pe KEY idx_origin_period (sys_origin, stat_timezone, period_type, period_key) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci` +func (r *MySQLRepository) ensureMetricColumns(ctx context.Context) error { + if err := r.addColumnIfMissing(ctx, "country_dashboard_daily_metric", "salary_transfer", + "ALTER TABLE country_dashboard_daily_metric ADD COLUMN salary_transfer decimal(20,2) NOT NULL DEFAULT 0 AFTER salary_exchange"); err != nil { + return err + } + return r.addColumnIfMissing(ctx, "country_dashboard_period_metric", "salary_transfer", + "ALTER TABLE country_dashboard_period_metric ADD COLUMN salary_transfer decimal(20,2) NOT NULL DEFAULT 0 AFTER salary_exchange") +} + +func (r *MySQLRepository) addColumnIfMissing(ctx context.Context, table string, column string, ddl string) error { + var exists int + err := r.db.QueryRowContext(ctx, ` +SELECT COUNT(1) +FROM information_schema.columns +WHERE table_schema = DATABASE() + AND table_name = ? + AND column_name = ? +`, table, column).Scan(&exists) + if err != nil { + return err + } + if exists > 0 { + return nil + } + if _, err := r.db.ExecContext(ctx, ddl); err != nil && !isDuplicateColumnError(err) { + return err + } + return nil +} + +func isDuplicateColumnError(err error) bool { + var mysqlErr *driverMysql.MySQLError + if err == nil { + return false + } + if errors.As(err, &mysqlErr) { + return mysqlErr.Number == 1060 + } + return fmt.Sprint(err) == "Error 1060 (42S21): Duplicate column name 'salary_transfer'" +} + const createPeriodUserMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_period_user_metric ( id bigint NOT NULL AUTO_INCREMENT, period_type varchar(16) NOT NULL,