Persist game CDC fact rows

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
 					}

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
 		}

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[:])
+}

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,