From 496e5b895e14fc0674391387b93883d22946f17d Mon Sep 17 00:00:00 2001 From: zhx Date: Mon, 25 May 2026 11:46:40 +0800 Subject: [PATCH] fix: reconcile dashboard gift wallet metrics --- cmd/dashboard-cdc-worker/main.go | 1 + internal/cdc/handler.go | 12 ++ internal/config/config.go | 17 ++ internal/dashboard/mapper_helpers.go | 8 + internal/dashboard/mapper_wallet.go | 44 +++++ internal/dashboard/types.go | 2 + internal/storage/aggregate.go | 14 +- internal/storage/batch.go | 4 +- internal/storage/batch_sql.go | 14 +- internal/storage/gift_reconcile.go | 250 +++++++++++++++++++++++++++ internal/storage/repository.go | 4 + internal/storage/schema.go | 2 + 12 files changed, 359 insertions(+), 13 deletions(-) create mode 100644 internal/storage/gift_reconcile.go diff --git a/cmd/dashboard-cdc-worker/main.go b/cmd/dashboard-cdc-worker/main.go index 6e565c4..599f314 100644 --- a/cmd/dashboard-cdc-worker/main.go +++ b/cmd/dashboard-cdc-worker/main.go @@ -81,6 +81,7 @@ func main() { projector := dashboard.NewProjector(cfg, repo, redisClient, metrics) repo.StartRechargeReconciler(ctx, logger, projector) + repo.StartGiftReconciler(ctx, logger, projector) worker, err := cdc.NewWorker(cfg, repo, projector, logger, metrics) if err != nil { diff --git a/internal/cdc/handler.go b/internal/cdc/handler.go index 6cdad30..7c6e249 100644 --- a/internal/cdc/handler.go +++ b/internal/cdc/handler.go @@ -152,6 +152,8 @@ func (h *handler) applyRows(ctx context.Context, e *canal.RowsEvent, rowIndex in eventKey := fmt.Sprintf("%s:%d:%s:%s:%d", h.currentFile(), e.Header.LogPos, source, e.Action, rowIndex) if key := stableRechargeEventKey(source, contributions); key != "" { eventKey = key + } else if key := stableContributionEventKey(contributions); key != "" { + eventKey = key } event := dashboard.Event{ EventKey: eventKey, @@ -166,6 +168,16 @@ func (h *handler) applyRows(ctx context.Context, e *canal.RowsEvent, rowIndex in return nil } +func stableContributionEventKey(contributions []dashboard.Contribution) string { + for _, contribution := range contributions { + if contribution.DeltaSign < 0 || contribution.EventKey == "" { + continue + } + return contribution.EventKey + } + return "" +} + func stableRechargeEventKey(source string, contributions []dashboard.Contribution) string { if source != "likei.order_purchase_history" && source != "likei.order_user_purchase_pay" { return "" diff --git a/internal/config/config.go b/internal/config/config.go index 77673b4..5bd76b6 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -16,6 +16,7 @@ type Config struct { CDC CDCConfig Dashboard DashboardConfig Recharge RechargeReconcileConfig + Gift GiftReconcileConfig Health HealthConfig } @@ -70,6 +71,13 @@ type RechargeReconcileConfig struct { BatchLimit int } +type GiftReconcileConfig struct { + Enabled bool + Interval time.Duration + CurrentDayLag time.Duration + BatchLimit int +} + type HealthConfig struct { Addr string } @@ -122,6 +130,12 @@ func Load() (Config, error) { CurrentDayLag: envDuration("DASHBOARD_RECHARGE_RECONCILE_CURRENT_DAY_LAG", 15*time.Minute), BatchLimit: envInt("DASHBOARD_RECHARGE_RECONCILE_BATCH_LIMIT", 5000), }, + Gift: GiftReconcileConfig{ + Enabled: envBool("DASHBOARD_GIFT_RECONCILE_ENABLED", true), + Interval: envDuration("DASHBOARD_GIFT_RECONCILE_INTERVAL", 10*time.Minute), + CurrentDayLag: envDuration("DASHBOARD_GIFT_RECONCILE_CURRENT_DAY_LAG", 5*time.Minute), + BatchLimit: envInt("DASHBOARD_GIFT_RECONCILE_BATCH_LIMIT", 50000), + }, Health: HealthConfig{ Addr: env("HEALTH_ADDR", ":2910"), }, @@ -151,6 +165,9 @@ func Load() (Config, error) { if cfg.Recharge.BatchLimit <= 0 { cfg.Recharge.BatchLimit = 5000 } + if cfg.Gift.BatchLimit <= 0 { + cfg.Gift.BatchLimit = 50000 + } if cfg.CDC.StartMode != "checkpoint" && cfg.CDC.StartMode != "current" { return Config{}, fmt.Errorf("unsupported CDC_START_MODE=%s", cfg.CDC.StartMode) } diff --git a/internal/dashboard/mapper_helpers.go b/internal/dashboard/mapper_helpers.go index 316df0f..bedfaa9 100644 --- a/internal/dashboard/mapper_helpers.go +++ b/internal/dashboard/mapper_helpers.go @@ -2,6 +2,7 @@ package dashboard import ( "context" + "fmt" "strings" "time" @@ -90,6 +91,13 @@ func rowPK(row Row) string { return strings.TrimSpace(row.String("event_id")) } +func WalletMetricEventKey(schema string, table string, id string) string { + if strings.TrimSpace(id) == "" { + return "" + } + return fmt.Sprintf("wallet-metric:%s.%s:%s", schema, table, id) +} + func rechargeDetail(sourceType string, recordID string, sourceChannel string, row Row, user UserCountry, amount decimal.Decimal, coinQuantity decimal.Decimal, remark string) *RechargeDetail { if recordID == "" || user.UserID == 0 { diff --git a/internal/dashboard/mapper_wallet.go b/internal/dashboard/mapper_wallet.go index 3c39cd1..3e381d1 100644 --- a/internal/dashboard/mapper_wallet.go +++ b/internal/dashboard/mapper_wallet.go @@ -18,6 +18,9 @@ var walletGameEvents = map[string]struct{}{ func (m *Mapper) mapWalletGame(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { eventType := row.String("event_type") + if eventType == "LUCKY_GIFT" || isWalletGiftConsumeEvent(eventType) { + return m.mapWalletGift(ctx, schema, table, action, row) + } if !isWalletGameEvent(eventType) { return nil, nil } @@ -60,6 +63,47 @@ func (m *Mapper) mapWalletGame(ctx context.Context, schema string, table string, return []Contribution{c}, nil } +func (m *Mapper) mapWalletGift(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { + sysOrigin := row.String("sys_origin") + if !m.allowed(sysOrigin) { + return nil, nil + } + user, ok, err := m.user(ctx, row.Int64("user_id")) + if err != nil || !ok || user.IsDeleted { + return nil, err + } + if user.SysOrigin != sysOrigin { + return nil, nil + } + + amount := row.Decimal("penny_amount").Div(decimal.NewFromInt(100)) + if amount.IsZero() { + return nil, nil + } + eventType := row.String("event_type") + recordType := row.Int64("type") + c := m.base(schema, table, action, row, user, row.MillisTime("create_time")) + if action != "delete" { + c.EventKey = WalletMetricEventKey(schema, table, rowPK(row)) + } + switch { + case eventType == "LUCKY_GIFT" && recordType == 1: + c.LuckyGiftTotalFlow = amount + c.LuckyGiftUser = true + case eventType == "LUCKY_GIFT" && recordType == 0: + c.LuckyGiftPayout = amount + case isWalletGiftConsumeEvent(eventType) && recordType == 1: + c.GiftConsume = amount + default: + return nil, nil + } + return []Contribution{c}, nil +} + +func isWalletGiftConsumeEvent(eventType string) bool { + return strings.HasPrefix(eventType, "GIVE_GIFT") && !strings.Contains(eventType, "LUCKY_GIFT") +} + func isWalletGameEvent(eventType string) bool { if _, ok := walletGameEvents[eventType]; ok { return true diff --git a/internal/dashboard/types.go b/internal/dashboard/types.go index 87d0240..98429ac 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 + GiftConsume decimal.Decimal LuckyGiftTotalFlow decimal.Decimal LuckyGiftUser bool LuckyGiftPayout decimal.Decimal @@ -81,6 +82,7 @@ func (c Contribution) Empty() bool { c.GoogleRecharge.IsZero() && c.DealerRecharge.IsZero() && c.SalaryExchange.IsZero() && + c.GiftConsume.IsZero() && c.LuckyGiftTotalFlow.IsZero() && !c.LuckyGiftUser && c.LuckyGiftPayout.IsZero() && diff --git a/internal/storage/aggregate.go b/internal/storage/aggregate.go index 3cd1e4e..8e9aa8d 100644 --- a/internal/storage/aggregate.go +++ b/internal/storage/aggregate.go @@ -100,9 +100,9 @@ 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, lucky_gift_total_flow, lucky_gift_payout, + dealer_recharge, salary_exchange, 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), @@ -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), + 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), game_total_flow = game_total_flow + VALUES(game_total_flow), @@ -118,7 +119,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.LuckyGiftTotalFlow.Mul(sign), c.LuckyGiftPayout.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)) return err } @@ -129,9 +130,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, lucky_gift_total_flow, lucky_gift_payout, + google_recharge, dealer_recharge, salary_exchange, 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), @@ -143,6 +144,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), + 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), game_total_flow = game_total_flow + VALUES(game_total_flow), @@ -151,7 +153,7 @@ 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.LuckyGiftTotalFlow.Mul(sign), c.LuckyGiftPayout.Mul(sign), c.GameTotalFlow.Mul(sign), c.GamePayout.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 1590849..12cf67f 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 + giftConsume decimal.Decimal luckyGiftTotalFlow decimal.Decimal luckyGiftPayout decimal.Decimal gameTotalFlow decimal.Decimal @@ -252,6 +253,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.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)) @@ -261,7 +263,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.luckyGiftTotalFlow.IsZero() && t.luckyGiftPayout.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 1f9c87d..9877710 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, lucky_gift_total_flow, lucky_gift_payout, + dealer_recharge, salary_exchange, 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), @@ -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), + 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), game_total_flow = game_total_flow + VALUES(game_total_flow), @@ -29,7 +30,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.luckyGiftTotalFlow, totals.luckyGiftPayout, + totals.dealerRecharge, totals.salaryExchange, totals.giftConsume, totals.luckyGiftTotalFlow, totals.luckyGiftPayout, totals.gameTotalFlow, totals.gamePayout) return err } @@ -39,9 +40,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, lucky_gift_total_flow, lucky_gift_payout, + google_recharge, dealer_recharge, salary_exchange, 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), @@ -53,6 +54,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), + 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), game_total_flow = game_total_flow + VALUES(game_total_flow), @@ -61,7 +63,7 @@ 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.luckyGiftTotalFlow, totals.luckyGiftPayout, totals.gameTotalFlow, totals.gamePayout) + totals.giftConsume, totals.luckyGiftTotalFlow, totals.luckyGiftPayout, totals.gameTotalFlow, totals.gamePayout) return err } diff --git a/internal/storage/gift_reconcile.go b/internal/storage/gift_reconcile.go new file mode 100644 index 0000000..55d8cdc --- /dev/null +++ b/internal/storage/gift_reconcile.go @@ -0,0 +1,250 @@ +package storage + +import ( + "context" + "database/sql" + "fmt" + "log/slog" + "strings" + "time" + + "github.com/shopspring/decimal" + + "dashboard-cdc-worker/internal/dashboard" +) + +type walletGiftRow struct { + table string + id string + idNumber int64 + userID int64 + sysOrigin string + eventType string + recordType int64 + amount decimal.Decimal + eventTime time.Time + user dashboard.UserCountry +} + +func (r *MySQLRepository) BuildGiftReconcileEvents(ctx context.Context, since time.Time, + limit int) ([]dashboard.Event, error) { + if limit <= 0 { + limit = 50000 + } + rows, err := r.listWalletGiftRows(ctx, since, limit) + if err != nil { + return nil, err + } + events := make([]dashboard.Event, 0, len(rows)) + for _, row := range rows { + if !r.allowedOrigin(row.sysOrigin) || row.user.IsDeleted || row.user.SysOrigin != row.sysOrigin { + continue + } + contribution := dashboard.Contribution{ + SourceSchema: r.cfg.MySQL.WalletSchema, + SourceTable: row.table, + SourcePK: row.id, + SourceAction: "insert", + EventKey: dashboard.WalletMetricEventKey(r.cfg.MySQL.WalletSchema, row.table, row.id), + DeltaSign: 1, + SysOrigin: row.sysOrigin, + UserID: row.userID, + Country: row.user.Country, + EventTime: row.eventTime, + UserCreatedAt: row.user.CreatedAt, + } + switch { + case row.eventType == "LUCKY_GIFT" && row.recordType == 1: + contribution.LuckyGiftTotalFlow = row.amount + contribution.LuckyGiftUser = true + case row.eventType == "LUCKY_GIFT" && row.recordType == 0: + contribution.LuckyGiftPayout = row.amount + case isGiftConsumeEvent(row.eventType) && row.recordType == 1: + contribution.GiftConsume = row.amount + default: + continue + } + events = append(events, dashboard.Event{ + EventKey: contribution.EventKey, + Source: r.cfg.MySQL.WalletSchema + "." + row.table, + Contributions: []dashboard.Contribution{contribution}, + }) + } + return events, nil +} + +func (r *MySQLRepository) StartGiftReconciler(ctx context.Context, logger *slog.Logger, + projector interface { + ApplyEvents(context.Context, []dashboard.Event) error + }) { + if !r.cfg.Gift.Enabled || r.cfg.Gift.Interval <= 0 { + return + } + run := func() { + since := currentStatDayStart(r.cfg.Dashboard.StatTimezones, r.cfg.Gift.CurrentDayLag) + events, err := r.BuildGiftReconcileEvents(ctx, since, r.cfg.Gift.BatchLimit) + if err != nil { + if r.metrics != nil { + r.metrics.RecordError(err) + } + logger.Error("dashboard gift reconcile query failed", "error", err) + return + } + if len(events) == 0 { + return + } + if err := projector.ApplyEvents(ctx, events); err != nil { + r.resetGiftHighWater() + if r.metrics != nil { + r.metrics.RecordError(err) + } + logger.Error("dashboard gift reconcile apply failed", "events", len(events), "error", err) + return + } + logger.Info("dashboard gift reconcile applied", "events", len(events), "since", since) + } + go func() { + run() + ticker := time.NewTicker(r.cfg.Gift.Interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + run() + } + } + }() +} + +func (r *MySQLRepository) resetGiftHighWater() { + r.giftMu.Lock() + defer r.giftMu.Unlock() + r.giftHighWater = make(map[string]int64) +} + +func (r *MySQLRepository) listWalletGiftRows(ctx context.Context, since time.Time, + limit int) ([]walletGiftRow, error) { + tables, err := r.walletRecordTables(ctx) + if err != nil { + return nil, err + } + remaining := limit + var result []walletGiftRow + for _, table := range tables { + if remaining <= 0 { + break + } + rows, err := r.listWalletGiftRowsFromTable(ctx, table, since, remaining) + if err != nil { + return nil, err + } + result = append(result, rows...) + remaining -= len(rows) + } + return result, nil +} + +func (r *MySQLRepository) walletRecordTables(ctx context.Context) ([]string, error) { + rows, err := r.db.QueryContext(ctx, ` +SELECT table_name +FROM information_schema.tables +WHERE table_schema = ? + AND table_name LIKE 'wallet_gold_asset_record\_%' +ORDER BY table_name`, r.cfg.MySQL.WalletSchema) + if err != nil { + return nil, err + } + defer rows.Close() + var tables []string + for rows.Next() { + var table string + if err := rows.Scan(&table); err != nil { + return nil, err + } + tables = append(tables, table) + } + return tables, rows.Err() +} + +func (r *MySQLRepository) listWalletGiftRowsFromTable(ctx context.Context, table string, + since time.Time, limit int) ([]walletGiftRow, error) { + highWaterKey := r.cfg.MySQL.WalletSchema + "." + table + r.giftMu.Lock() + highWater := r.giftHighWater[highWaterKey] + r.giftMu.Unlock() + + sinceMillis := since.UnixMilli() + query := fmt.Sprintf(` +SELECT CAST(w.id AS CHAR), w.id, w.user_id, w.sys_origin, w.event_type, w.type, w.penny_amount, + FROM_UNIXTIME(w.create_time / 1000), + 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 +FROM %s.%s w +JOIN %s.user_base_info u ON u.id = w.user_id +WHERE w.id > ? + AND w.create_time >= ? + AND ( + w.event_type = 'LUCKY_GIFT' + OR (w.event_type LIKE 'GIVE_GIFT%%' AND w.event_type NOT LIKE '%%LUCKY_GIFT%%' AND w.type = 1) + ) +ORDER BY w.id ASC +LIMIT ?`, quoteIdent(r.cfg.MySQL.WalletSchema), quoteIdent(table), quoteIdent(r.cfg.MySQL.Schema)) + rows, err := r.db.QueryContext(ctx, query, r.cfg.Dashboard.UnknownCountryCode, + r.cfg.Dashboard.UnknownCountryCode, highWater, sinceMillis, limit) + if err != nil { + return nil, err + } + defer rows.Close() + + var result []walletGiftRow + maxID := highWater + for rows.Next() { + var item walletGiftRow + item.table = table + var penny int64 + var countryCode, countryName string + var isDeleted bool + var userCreatedAt, userUpdatedAt sql.NullTime + if err := rows.Scan(&item.id, &item.idNumber, &item.userID, &item.sysOrigin, + &item.eventType, &item.recordType, &penny, &item.eventTime, &item.user.UserID, + &item.user.SysOrigin, &countryCode, &countryName, &isDeleted, + &userCreatedAt, &userUpdatedAt); err != nil { + return nil, err + } + item.amount = decimal.NewFromInt(penny).Div(decimal.NewFromInt(100)) + item.user.Country = dashboard.Country{Code: countryCode, Name: countryName} + item.user.IsDeleted = isDeleted + if userCreatedAt.Valid { + item.user.CreatedAt = userCreatedAt.Time + } + if userUpdatedAt.Valid { + item.user.UpdatedAt = userUpdatedAt.Time + } + if item.idNumber > maxID { + maxID = item.idNumber + } + result = append(result, item) + } + if err := rows.Err(); err != nil { + return nil, err + } + if maxID > highWater { + r.giftMu.Lock() + if maxID > r.giftHighWater[highWaterKey] { + r.giftHighWater[highWaterKey] = maxID + } + r.giftMu.Unlock() + } + return result, nil +} + +func isGiftConsumeEvent(eventType string) bool { + return strings.HasPrefix(eventType, "GIVE_GIFT") && !strings.Contains(eventType, "LUCKY_GIFT") +} + +func quoteIdent(value string) string { + return "`" + strings.ReplaceAll(value, "`", "``") + "`" +} diff --git a/internal/storage/repository.go b/internal/storage/repository.go index 3d314ca..b275249 100644 --- a/internal/storage/repository.go +++ b/internal/storage/repository.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "math" + "sync" "time" "github.com/go-mysql-org/go-mysql/mysql" @@ -25,6 +26,8 @@ type MySQLRepository struct { metrics *monitor.Metrics userCache *ttlCache[int64, dashboard.UserCountry] gameNameCache *ttlCache[string, string] + giftMu sync.Mutex + giftHighWater map[string]int64 } func NewMySQLRepository(db *sql.DB, cfg config.Config, metrics *monitor.Metrics) *MySQLRepository { @@ -34,6 +37,7 @@ func NewMySQLRepository(db *sql.DB, cfg config.Config, metrics *monitor.Metrics) metrics: metrics, userCache: newTTLCache[int64, dashboard.UserCountry](30*time.Minute, 200000), gameNameCache: newTTLCache[string, string](6*time.Hour, 10000), + giftHighWater: make(map[string]int64), } } diff --git a/internal/storage/schema.go b/internal/storage/schema.go index 24860fe..9323d2c 100644 --- a/internal/storage/schema.go +++ b/internal/storage/schema.go @@ -54,6 +54,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, + 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, lucky_gift_payout decimal(20,2) NOT NULL DEFAULT 0, @@ -98,6 +99,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, + 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, lucky_gift_payout decimal(20,2) NOT NULL DEFAULT 0,