From 422fab1bf329c4907d728762cd4ebdc7fe18323e Mon Sep 17 00:00:00 2001 From: zhx Date: Wed, 27 May 2026 18:11:30 +0800 Subject: [PATCH] Persist game CDC fact rows --- internal/dashboard/projector.go | 3 ++ internal/storage/batch.go | 3 ++ internal/storage/game_fact.go | 78 +++++++++++++++++++++++++++++++++ internal/storage/schema.go | 28 ++++++++++++ 4 files changed, 112 insertions(+) create mode 100644 internal/storage/game_fact.go diff --git a/internal/dashboard/projector.go b/internal/dashboard/projector.go index bedf224..6185f9f 100644 --- a/internal/dashboard/projector.go +++ b/internal/dashboard/projector.go @@ -79,6 +79,9 @@ func (p *Projector) ApplyEvents(ctx context.Context, events []Event) error { if contribution.Empty() { continue } + if contribution.EventKey == "" { + contribution.EventKey = event.EventKey + } if contribution.DeltaSign == 0 { contribution.DeltaSign = 1 } diff --git a/internal/storage/batch.go b/internal/storage/batch.go index 49978b9..83c41dc 100644 --- a/internal/storage/batch.go +++ b/internal/storage/batch.go @@ -135,6 +135,9 @@ func (r *MySQLRepository) ApplyContributionBatch(ctx context.Context, tx *sql.Tx if err := r.applyRechargeDetail(ctx, tx, c); err != nil { return err } + if err := r.upsertGameEventFact(ctx, tx, c, storageZone); err != nil { + return err + } if c.EventTime.IsZero() { continue } diff --git a/internal/storage/game_fact.go b/internal/storage/game_fact.go new file mode 100644 index 0000000..baa86ad --- /dev/null +++ b/internal/storage/game_fact.go @@ -0,0 +1,78 @@ +package storage + +import ( + "context" + "crypto/sha1" + "database/sql" + "encoding/hex" + "strings" + "time" + + "github.com/shopspring/decimal" + + "dashboard-cdc-worker/internal/dashboard" +) + +func (r *MySQLRepository) upsertGameEventFact(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, storageZone *time.Location) error { + if !isGameFact(c) { + return nil + } + if storageZone == nil { + storageZone = time.UTC + } + deltaSign := c.DeltaSign + if deltaSign == 0 { + deltaSign = 1 + } + sign := decimal.NewFromInt(int64(deltaSign)) + consume := c.GameConsume.Mul(sign) + payout := c.GamePayoutBy.Mul(sign) + profit := c.GameProfit.Mul(sign) + orderCount := c.GameOrder * int64(deltaSign) + _, err := tx.ExecContext(ctx, ` +INSERT INTO dashboard_game_event_fact ( + fact_key, source_schema, source_table, source_pk, source_action, + sys_origin, stat_date, event_time, user_id, country_code, country_name, + game_provider, game_id, game_name, consume_amount, payout_amount, profit_amount, order_count +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +ON DUPLICATE KEY UPDATE + source_action = VALUES(source_action), + sys_origin = VALUES(sys_origin), + stat_date = VALUES(stat_date), + event_time = VALUES(event_time), + user_id = VALUES(user_id), + country_code = VALUES(country_code), + country_name = VALUES(country_name), + game_provider = VALUES(game_provider), + game_id = VALUES(game_id), + game_name = VALUES(game_name), + consume_amount = VALUES(consume_amount), + payout_amount = VALUES(payout_amount), + profit_amount = VALUES(profit_amount), + order_count = VALUES(order_count), + updated_at = NOW() +`, gameFactKey(c), c.SourceSchema, c.SourceTable, c.SourcePK, c.SourceAction, + c.SysOrigin, dashboard.DailyDate(c.EventTime, storageZone), c.EventTime, c.UserID, + c.Country.Code, c.Country.Name, c.GameProvider, c.GameID, c.GameName, + consume, payout, profit, orderCount) + return err +} + +func isGameFact(c dashboard.Contribution) bool { + return !c.EventTime.IsZero() && + c.SysOrigin != "" && + c.UserID != 0 && + c.GameProvider != "" && + c.GameID != "" && + (!c.GameConsume.IsZero() || !c.GamePayoutBy.IsZero() || !c.GameProfit.IsZero() || c.GameOrder != 0 || c.GameUser) +} + +func gameFactKey(c dashboard.Contribution) string { + parts := []string{c.EventKey, c.SourceSchema, c.SourceTable, c.SourcePK, c.SourceAction} + raw := strings.Join(parts, "|") + if len(raw) <= 255 && strings.Trim(raw, "|") != "" { + return raw + } + sum := sha1.Sum([]byte(raw)) + return "game-fact:" + hex.EncodeToString(sum[:]) +} diff --git a/internal/storage/schema.go b/internal/storage/schema.go index 5d3ca4c..0dd59ce 100644 --- a/internal/storage/schema.go +++ b/internal/storage/schema.go @@ -43,6 +43,7 @@ func (r *MySQLRepository) EnsureSchema(ctx context.Context) error { createPeriodUserMetricTable, createGamePeriodMetricTable, createGamePeriodUserMetricTable, + createGameEventFactTable, createRechargeDetailTable, }); err != nil { return err @@ -265,6 +266,33 @@ const createGamePeriodUserMetricTable = `CREATE TABLE IF NOT EXISTS country_dash KEY idx_game_user (sys_origin, stat_timezone, user_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci` +const createGameEventFactTable = `CREATE TABLE IF NOT EXISTS dashboard_game_event_fact ( + fact_key varchar(255) NOT NULL, + source_schema varchar(64) NOT NULL, + source_table varchar(128) NOT NULL, + source_pk varchar(128) NOT NULL, + source_action varchar(16) NOT NULL, + sys_origin varchar(50) NOT NULL, + stat_date date NOT NULL, + event_time datetime NOT NULL, + user_id bigint NOT NULL, + country_code varchar(64) NOT NULL, + country_name varchar(128) NOT NULL, + game_provider varchar(32) NOT NULL, + game_id varchar(128) NOT NULL, + game_name varchar(256) NOT NULL, + consume_amount decimal(20,2) NOT NULL DEFAULT 0, + payout_amount decimal(20,2) NOT NULL DEFAULT 0, + profit_amount decimal(20,2) NOT NULL DEFAULT 0, + order_count bigint NOT NULL DEFAULT 0, + created_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (fact_key), + KEY idx_game_fact_date_origin_country (stat_date, sys_origin, country_code), + KEY idx_game_fact_date_game (stat_date, sys_origin, game_provider, game_id), + KEY idx_game_fact_user (stat_date, sys_origin, user_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci` + const createRechargeDetailTable = `CREATE TABLE IF NOT EXISTS country_dashboard_recharge_detail ( id bigint NOT NULL AUTO_INCREMENT, source_type varchar(32) NOT NULL,