admin/dashboard-cdc-worker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Support salary transfer dashboard metrics

Browse Source
This commit is contained in:
zhx 2026-05-25 13:23:34 +08:00
parent 496e5b895e
commit af1fbde1ac
7 changed files with 89 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
internal/dashboard/mapper.go
View File

@ -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
}
})
}

1
internal/dashboard/redis.go
View File

@ -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))

2
internal/dashboard/types.go
View File

@ -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 &&

13
internal/storage/aggregate.go
View File

@ -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
}

3
internal/storage/batch.go
View File

@ -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()
}

13
internal/storage/batch_sql.go
View File

@ -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
}

58
internal/storage/schema.go
View File

@ -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,