From cc84e14860e059648c4be0fab1cb7052b4c6f6a0 Mon Sep 17 00:00:00 2001 From: zhx Date: Thu, 11 Jun 2026 19:24:51 +0800 Subject: [PATCH] =?UTF-8?q?=E9=A6=96=E5=85=85=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/gamemq/messages.go | 28 +- .../mysql/dice_matchmaking_repository.go | 51 +- .../internal/storage/mysql/dice_repository.go | 179 ++++++- .../configs/config.docker.yaml | 3 + .../configs/config.tencent.example.yaml | 3 + services/gateway-service/configs/config.yaml | 3 + services/gateway-service/internal/app/app.go | 2 + .../internal/client/statistics_client.go | 81 +++ .../gateway-service/internal/config/config.go | 18 + .../first_recharge_reward_handler.go | 8 +- .../first_recharge_reward_handler_test.go | 33 ++ .../transport/http/gameapi/handler.go | 4 + .../http/gameapi/self_game_report_handler.go | 95 ++++ .../internal/transport/http/handler.go | 6 + .../transport/http/httproutes/router.go | 2 + .../internal/transport/http/router.go | 1 + .../mysql/initdb/001_statistics_service.sql | 123 +++++ .../statistics-service/internal/app/app.go | 131 ++++- .../internal/app/local_flow_test.go | 161 ++++++ .../internal/app/query_http.go | 102 ++++ .../internal/storage/mysql/query.go | 415 ++++++++++++++++ .../internal/storage/mysql/repository.go | 465 ++++++++++++++++++ 22 files changed, 1897 insertions(+), 17 deletions(-) create mode 100644 services/gateway-service/internal/client/statistics_client.go create mode 100644 services/gateway-service/internal/transport/http/activityapi/first_recharge_reward_handler_test.go create mode 100644 services/gateway-service/internal/transport/http/gameapi/self_game_report_handler.go diff --git a/pkg/gamemq/messages.go b/pkg/gamemq/messages.go index c74a4e61..2b3060cf 100644 --- a/pkg/gamemq/messages.go +++ b/pkg/gamemq/messages.go @@ -11,6 +11,14 @@ const ( MessageTypeGameOutboxEvent = "game_outbox_event" TagGameOutboxEvent = "game_outbox_event" + + EventTypeGameOrderSettled = "GameOrderSettled" + EventTypeSelfGameMatchCreated = "SelfGameMatchCreated" + EventTypeSelfGameMatchReady = "SelfGameMatchReady" + EventTypeSelfGameMatchSettled = "SelfGameMatchSettled" + EventTypeSelfGameMatchCanceled = "SelfGameMatchCanceled" + EventTypeSelfGameMatchFailed = "SelfGameMatchFailed" + EventTypeSelfGamePoolAdjusted = "SelfGamePoolAdjusted" ) // GameOutboxMessage is the MQ representation of one committed game_outbox fact. @@ -35,14 +43,16 @@ func EncodeGameOutboxMessage(message GameOutboxMessage) ([]byte, error) { if strings.TrimSpace(message.AppCode) == "" || strings.TrimSpace(message.EventID) == "" || strings.TrimSpace(message.EventType) == "" || - strings.TrimSpace(message.OrderID) == "" || - message.UserID <= 0 || strings.TrimSpace(message.GameID) == "" || - strings.TrimSpace(message.OpType) == "" || - message.CoinAmount <= 0 || message.OccurredAtMS <= 0 { return nil, errors.New("game outbox message is incomplete") } + if message.EventType == EventTypeGameOrderSettled && (strings.TrimSpace(message.OrderID) == "" || + message.UserID <= 0 || + strings.TrimSpace(message.OpType) == "" || + message.CoinAmount <= 0) { + return nil, errors.New("game order outbox message is incomplete") + } if strings.TrimSpace(message.PayloadJSON) == "" { message.PayloadJSON = "{}" } @@ -61,14 +71,16 @@ func DecodeGameOutboxMessage(body []byte) (GameOutboxMessage, error) { if strings.TrimSpace(message.AppCode) == "" || strings.TrimSpace(message.EventID) == "" || strings.TrimSpace(message.EventType) == "" || - strings.TrimSpace(message.OrderID) == "" || - message.UserID <= 0 || strings.TrimSpace(message.GameID) == "" || - strings.TrimSpace(message.OpType) == "" || - message.CoinAmount <= 0 || message.OccurredAtMS <= 0 { return GameOutboxMessage{}, errors.New("game outbox message is incomplete") } + if message.EventType == EventTypeGameOrderSettled && (strings.TrimSpace(message.OrderID) == "" || + message.UserID <= 0 || + strings.TrimSpace(message.OpType) == "" || + message.CoinAmount <= 0) { + return GameOutboxMessage{}, errors.New("game order outbox message is incomplete") + } if strings.TrimSpace(message.PayloadJSON) == "" { message.PayloadJSON = "{}" } diff --git a/services/game-service/internal/storage/mysql/dice_matchmaking_repository.go b/services/game-service/internal/storage/mysql/dice_matchmaking_repository.go index 831b3a7a..140db81f 100644 --- a/services/game-service/internal/storage/mysql/dice_matchmaking_repository.go +++ b/services/game-service/internal/storage/mysql/dice_matchmaking_repository.go @@ -4,6 +4,7 @@ import ( "context" cryptorand "crypto/rand" "database/sql" + "encoding/json" "errors" "fmt" "math/big" @@ -11,6 +12,7 @@ import ( "time" "hyapp/pkg/appcode" + "hyapp/pkg/gamemq" "hyapp/pkg/xerr" dicedomain "hyapp/services/game-service/internal/domain/dice" ) @@ -252,6 +254,13 @@ func (r *Repository) CancelDiceMatch(ctx context.Context, appCode string, matchI ); err != nil { return dicedomain.Match{}, err } + match.Status = dicedomain.MatchStatusCanceled + match.CanceledAtMS = nowMs + match.UpdatedAtMS = nowMs + match.Participants = participants + if err := insertSelfGameMatchOutbox(ctx, tx, gamemq.EventTypeSelfGameMatchCanceled, match, nowMs); err != nil { + return dicedomain.Match{}, err + } if err := tx.Commit(); err != nil { return dicedomain.Match{}, err } @@ -385,6 +394,34 @@ func (r *Repository) AdjustDicePool(ctx context.Context, adjustment dicedomain.P return dicedomain.PoolAdjustment{}, err } } + payload, err := json.Marshal(map[string]any{ + "adjustment_id": adjustment.AdjustmentID, + "game_id": adjustment.GameID, + "match_id": adjustment.MatchID, + "user_id": adjustment.UserID, + "direction": adjustment.Direction, + "amount_coin": adjustment.AmountCoin, + "reason": adjustment.Reason, + "balance_after": adjustment.BalanceAfter, + "created_by": adjustment.CreatedBy, + "created_at_ms": adjustment.CreatedAtMS, + }) + if err != nil { + return dicedomain.PoolAdjustment{}, err + } + // 奖池调整和奖池流水同事务写入 outbox；统计服务消费 owner fact 后才能同时得到入池、出池和余额快照。 + if _, err := tx.ExecContext(ctx, ` + INSERT INTO game_outbox ( + app_code, event_id, event_type, order_id, user_id, platform_code, game_id, + op_type, coin_amount, payload_json, status, attempts, next_retry_at_ms, + created_at_ms, updated_at_ms, published_at_ms + ) VALUES (?, ?, ?, '', ?, 'self', ?, '', 0, CAST(? AS JSON), 'pending', 0, 0, ?, ?, 0) + ON DUPLICATE KEY UPDATE updated_at_ms = VALUES(updated_at_ms)`, + adjustment.AppCode, gamemq.EventTypeSelfGamePoolAdjusted+":"+adjustment.AdjustmentID, gamemq.EventTypeSelfGamePoolAdjusted, + adjustment.UserID, adjustment.GameID, string(payload), adjustment.CreatedAtMS, adjustment.CreatedAtMS, + ); err != nil { + return dicedomain.PoolAdjustment{}, err + } if err := tx.Commit(); err != nil { return dicedomain.PoolAdjustment{}, err } @@ -672,7 +709,19 @@ func updateDiceMatchAfterJoinLocked(ctx context.Context, tx *sql.Tx, match diced match.CurrentPlayers, nextStatus, readyAtMS, strings.TrimSpace(matchMode), strings.TrimSpace(forcedResult), nowMs, match.AppCode, match.MatchID, ) - return err + if err != nil { + return err + } + match.Status = nextStatus + match.ReadyAtMS = readyAtMS + match.MatchMode = strings.TrimSpace(matchMode) + match.ForcedResult = strings.TrimSpace(forcedResult) + match.UpdatedAtMS = nowMs + if nextStatus == dicedomain.MatchStatusReady { + // ready 事件只用 INSERT IGNORE 写一次；真人 join 和机器人补位可能都走这里，幂等键保证不会重复统计匹配成功局。 + return insertSelfGameMatchOutbox(ctx, tx, gamemq.EventTypeSelfGameMatchReady, match, nowMs) + } + return nil } func normalizeDiceGameID(gameID string) string { diff --git a/services/game-service/internal/storage/mysql/dice_repository.go b/services/game-service/internal/storage/mysql/dice_repository.go index e8dea804..8185fd3b 100644 --- a/services/game-service/internal/storage/mysql/dice_repository.go +++ b/services/game-service/internal/storage/mysql/dice_repository.go @@ -8,6 +8,7 @@ import ( "strings" "hyapp/pkg/appcode" + "hyapp/pkg/gamemq" "hyapp/pkg/xerr" dicedomain "hyapp/services/game-service/internal/domain/dice" ) @@ -55,6 +56,9 @@ func (r *Repository) CreateDiceMatch(ctx context.Context, match dicedomain.Match return dicedomain.Match{}, err } } + if err := insertSelfGameMatchOutbox(ctx, tx, gamemq.EventTypeSelfGameMatchCreated, match, match.CreatedAtMS); err != nil { + return dicedomain.Match{}, err + } if err := tx.Commit(); err != nil { return dicedomain.Match{}, err } @@ -139,6 +143,20 @@ func (r *Repository) JoinDiceMatch(ctx context.Context, appCode string, matchID ); err != nil { return dicedomain.Match{}, err } + match.CurrentPlayers = nextPlayers + match.Status = nextStatus + match.ReadyAtMS = readyAtMS + match.UpdatedAtMS = nowMs + participants, err := queryDiceParticipants(ctx, tx, app, match.MatchID, false) + if err != nil { + return dicedomain.Match{}, err + } + match.Participants = participants + if nextStatus == dicedomain.MatchStatusReady { + if err := insertSelfGameMatchOutbox(ctx, tx, gamemq.EventTypeSelfGameMatchReady, match, nowMs); err != nil { + return dicedomain.Match{}, err + } + } if err := tx.Commit(); err != nil { return dicedomain.Match{}, err } @@ -394,19 +412,176 @@ func (r *Repository) MarkDiceMatchSettled(ctx context.Context, appCode string, m ); err != nil { return err } + match, err := queryDiceMatch(ctx, tx, app, strings.TrimSpace(matchID), true) + if err != nil { + return err + } + participants, err := queryDiceParticipants(ctx, tx, app, match.MatchID, true) + if err != nil { + return err + } + match.Participants = participants + if err := insertSelfGameMatchOutbox(ctx, tx, gamemq.EventTypeSelfGameMatchSettled, match, nowMs); err != nil { + return err + } return tx.Commit() } func (r *Repository) MarkDiceMatchFailed(ctx context.Context, appCode string, matchID string, result string, nowMs int64) error { - _, err := r.db.ExecContext(ctx, + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + app := appcode.Normalize(appCode) + if _, err := tx.ExecContext(ctx, `UPDATE game_dice_matches SET status = ?, result = ?, updated_at_ms = ? WHERE app_code = ? AND match_id = ?`, - dicedomain.MatchStatusFailed, strings.TrimSpace(result), nowMs, appcode.Normalize(appCode), strings.TrimSpace(matchID), + dicedomain.MatchStatusFailed, strings.TrimSpace(result), nowMs, app, strings.TrimSpace(matchID), + ); err != nil { + return err + } + match, err := queryDiceMatch(ctx, tx, app, strings.TrimSpace(matchID), true) + if err != nil { + return err + } + participants, err := queryDiceParticipants(ctx, tx, app, match.MatchID, true) + if err != nil { + return err + } + match.Participants = participants + if err := insertSelfGameMatchOutbox(ctx, tx, gamemq.EventTypeSelfGameMatchFailed, match, nowMs); err != nil { + return err + } + return tx.Commit() +} + +func insertSelfGameMatchOutbox(ctx context.Context, tx *sql.Tx, eventType string, match dicedomain.Match, nowMs int64) error { + match.AppCode = appcode.Normalize(match.AppCode) + match.GameID = strings.TrimSpace(match.GameID) + if match.AppCode == "" || match.MatchID == "" || match.GameID == "" || nowMs <= 0 { + return xerr.New(xerr.InvalidArgument, "self game match event is incomplete") + } + payload := selfGameMatchOutboxPayload(match, nowMs) + raw, err := json.Marshal(payload) + if err != nil { + return err + } + userID := int64(0) + if len(match.Participants) > 0 { + userID = match.Participants[0].UserID + } + // 自研局级事件没有单笔订单语义，order_id/op_type/coin_amount 只保留空值；统计服务按 event_type 解析 payload。 + _, err = tx.ExecContext(ctx, ` + INSERT IGNORE INTO game_outbox ( + app_code, event_id, event_type, order_id, user_id, platform_code, game_id, op_type, + coin_amount, payload_json, status, created_at_ms, updated_at_ms + ) VALUES (?, ?, ?, '', ?, ?, ?, '', 0, CAST(? AS JSON), 'pending', ?, ?)`, + match.AppCode, eventType+":"+match.MatchID, eventType, userID, strings.TrimSpace(match.PlatformCode), + match.GameID, string(raw), nowMs, nowMs, ) return err } +func selfGameMatchOutboxPayload(match dicedomain.Match, nowMs int64) map[string]any { + participants := make([]map[string]any, 0, len(match.Participants)) + userStakeCoin, robotStakeCoin, payoutCoin, refundCoin := int64(0), int64(0), int64(0), int64(0) + userParticipants, robotParticipants, winners, losers, draws := int64(0), int64(0), int64(0), int64(0), int64(0) + for _, participant := range match.Participants { + isRobot := participant.ParticipantType == dicedomain.ParticipantTypeRobot + if isRobot { + robotParticipants++ + robotStakeCoin += participant.StakeCoin + } else { + userParticipants++ + userStakeCoin += participant.StakeCoin + } + switch participant.Result { + case dicedomain.ParticipantResultWin: + winners++ + case dicedomain.ParticipantResultLose: + losers++ + case dicedomain.ParticipantResultDraw: + draws++ + if !isRobot { + refundCoin += participant.PayoutCoin + } + } + if !isRobot && participant.Result != dicedomain.ParticipantResultDraw { + payoutCoin += participant.PayoutCoin + } + netWinCoin := participant.PayoutCoin + if !isRobot { + netWinCoin -= participant.StakeCoin + } + participants = append(participants, map[string]any{ + "user_id": participant.UserID, + "participant_type": participant.ParticipantType, + "seat_no": participant.SeatNo, + "status": participant.Status, + "stake_coin": participant.StakeCoin, + "dice_points": participant.DicePoints, + "rps_gesture": strings.TrimSpace(participant.RPSGesture), + "result": strings.TrimSpace(participant.Result), + "payout_coin": participant.PayoutCoin, + "net_win_coin": netWinCoin, + "joined_at_ms": participant.JoinedAtMS, + "updated_at_ms": participant.UpdatedAtMS, + }) + } + waitMS, durationMS := int64(0), int64(0) + if match.ReadyAtMS > 0 && match.CreatedAtMS > 0 { + waitMS = match.ReadyAtMS - match.CreatedAtMS + } + finishedAtMS := match.SettledAtMS + if finishedAtMS == 0 { + finishedAtMS = match.CanceledAtMS + } + if finishedAtMS > 0 && match.CreatedAtMS > 0 { + durationMS = finishedAtMS - match.CreatedAtMS + } + return map[string]any{ + "match_id": match.MatchID, + "game_id": strings.TrimSpace(match.GameID), + "platform_code": strings.TrimSpace(match.PlatformCode), + "provider_game_id": strings.TrimSpace(match.ProviderGameID), + "room_id": strings.TrimSpace(match.RoomID), + "region_id": match.RegionID, + "stake_coin": match.StakeCoin, + "min_players": match.MinPlayers, + "max_players": match.MaxPlayers, + "current_players": match.CurrentPlayers, + "status": strings.TrimSpace(match.Status), + "result": strings.TrimSpace(match.Result), + "match_mode": strings.TrimSpace(match.MatchMode), + "forced_result": strings.TrimSpace(match.ForcedResult), + "fee_bps": match.FeeBPS, + "pool_bps": match.PoolBPS, + "pool_delta_coin": match.PoolDeltaCoin, + "user_participants": userParticipants, + "robot_participants": robotParticipants, + "winner_count": winners, + "loser_count": losers, + "draw_count": draws, + "user_stake_coin": userStakeCoin, + "robot_stake_coin": robotStakeCoin, + "payout_coin": payoutCoin, + "refund_coin": refundCoin, + "platform_profit_coin": userStakeCoin - payoutCoin - refundCoin, + "join_deadline_ms": match.JoinDeadlineMS, + "ready_at_ms": match.ReadyAtMS, + "created_at_ms": match.CreatedAtMS, + "updated_at_ms": match.UpdatedAtMS, + "settled_at_ms": match.SettledAtMS, + "canceled_at_ms": match.CanceledAtMS, + "wait_ms": waitMS, + "duration_ms": durationMS, + "participants": participants, + "event_recorded_at_ms": nowMs, + } +} + func queryDiceMatch(ctx context.Context, q diceQuerier, appCode string, matchID string, forUpdate bool) (dicedomain.Match, error) { // forUpdate 只给状态推进路径使用；读路径不加锁，避免普通刷新阻塞 join/roll。 query := `SELECT app_code, match_id, game_id, platform_code, provider_game_id, room_id, region_id, diff --git a/services/gateway-service/configs/config.docker.yaml b/services/gateway-service/configs/config.docker.yaml index a6b24894..d2d0be9b 100644 --- a/services/gateway-service/configs/config.docker.yaml +++ b/services/gateway-service/configs/config.docker.yaml @@ -34,6 +34,9 @@ user_service_addr: "user-service:13005" wallet_service_addr: "wallet-service:13004" activity_service_addr: "activity-service:13006" game_service_addr: "game-service:13008" +statistics: + base_url: "http://statistics-service:13010" + timeout: "2s" grpc_client: # Docker 本地也使用生产同形态的 deadline、keepalive 和窄重试策略。 default_timeout: "5s" diff --git a/services/gateway-service/configs/config.tencent.example.yaml b/services/gateway-service/configs/config.tencent.example.yaml index 57011b7d..d12d2655 100644 --- a/services/gateway-service/configs/config.tencent.example.yaml +++ b/services/gateway-service/configs/config.tencent.example.yaml @@ -38,6 +38,9 @@ user_service_addr: "user-service.internal:13005" wallet_service_addr: "wallet-service.internal:13004" activity_service_addr: "activity-service.internal:13006" game_service_addr: "game-service.internal:13008" +statistics: + base_url: "http://statistics-service.internal:13010" + timeout: "2s" grpc_client: # 线上通过内网 CLB/DNS 访问内部服务；重试只覆盖瞬时 UNAVAILABLE，避免放大业务副作用。 default_timeout: "5s" diff --git a/services/gateway-service/configs/config.yaml b/services/gateway-service/configs/config.yaml index 243d2a3b..8601f69f 100644 --- a/services/gateway-service/configs/config.yaml +++ b/services/gateway-service/configs/config.yaml @@ -34,6 +34,9 @@ user_service_addr: "127.0.0.1:13005" wallet_service_addr: "127.0.0.1:13004" activity_service_addr: "127.0.0.1:13006" game_service_addr: "127.0.0.1:13008" +statistics: + base_url: "http://127.0.0.1:13010" + timeout: "2s" grpc_client: # gateway 内部 gRPC 默认 deadline 包含一次 UNAVAILABLE 重试；业务长调用应在 handler 上下文显式收紧或放宽。 default_timeout: "5s" diff --git a/services/gateway-service/internal/app/app.go b/services/gateway-service/internal/app/app.go index e5924211..c0cacb5d 100644 --- a/services/gateway-service/internal/app/app.go +++ b/services/gateway-service/internal/app/app.go @@ -109,6 +109,7 @@ func New(cfg config.Config) (*App, error) { var weeklyStarClient client.WeeklyStarClient = client.NewGRPCWeeklyStarClient(activityConn) var broadcastClient client.BroadcastClient = client.NewGRPCBroadcastClient(activityConn) var gameClient client.GameClient = client.NewGRPCGameClient(gameConn) + var statisticsClient client.StatisticsClient = client.NewHTTPStatisticsClient(cfg.Statistics.BaseURL, cfg.Statistics.Timeout) appConfigReader, err := openAppConfigReader(cfg.AppConfig) if err != nil { _ = roomConn.Close() @@ -179,6 +180,7 @@ func New(cfg config.Config) (*App, error) { handler.SetWeeklyStarClient(weeklyStarClient) handler.SetBroadcastClient(broadcastClient) handler.SetGameClient(gameClient) + handler.SetStatisticsClient(statisticsClient) handler.SetLeaderboardWalletDB(leaderboardWalletDB) if appConfigReader != nil { handler.SetAppConfigReader(appConfigReader) diff --git a/services/gateway-service/internal/client/statistics_client.go b/services/gateway-service/internal/client/statistics_client.go new file mode 100644 index 00000000..0b3c3d6f --- /dev/null +++ b/services/gateway-service/internal/client/statistics_client.go @@ -0,0 +1,81 @@ +package client + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +type SelfGameH5Event struct { + AppCode string `json:"app_code"` + EventID string `json:"event_id"` + EventName string `json:"event_name"` + GameID string `json:"game_id"` + Page string `json:"page"` + SessionID string `json:"session_id"` + MatchID string `json:"match_id"` + UserID int64 `json:"user_id"` + CountryID int64 `json:"country_id"` + RegionID int64 `json:"region_id"` + Language string `json:"language"` + ClientVersion string `json:"client_version"` + H5Version string `json:"h5_version"` + EntrySource string `json:"entry_source"` + DurationMS int64 `json:"duration_ms"` + Success bool `json:"success"` + ErrorCode string `json:"error_code"` + OccurredAtMS int64 `json:"occurred_at_ms"` +} + +type StatisticsClient interface { + ReportSelfGameH5Event(ctx context.Context, event SelfGameH5Event) error +} + +type HTTPStatisticsClient struct { + baseURL string + httpClient *http.Client +} + +func NewHTTPStatisticsClient(baseURL string, timeout time.Duration) *HTTPStatisticsClient { + baseURL = strings.TrimRight(strings.TrimSpace(baseURL), "/") + if baseURL == "" { + return nil + } + if timeout <= 0 { + timeout = 2 * time.Second + } + return &HTTPStatisticsClient{ + baseURL: baseURL, + httpClient: &http.Client{Timeout: timeout}, + } +} + +func (c *HTTPStatisticsClient) ReportSelfGameH5Event(ctx context.Context, event SelfGameH5Event) error { + if c == nil || c.httpClient == nil || c.baseURL == "" { + return fmt.Errorf("statistics client is not configured") + } + body, err := json.Marshal(event) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/internal/v1/statistics/self-game/events", bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + resp, err := c.httpClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + return nil + } + payload, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return fmt.Errorf("statistics report failed: status=%d body=%s", resp.StatusCode, strings.TrimSpace(string(payload))) +} diff --git a/services/gateway-service/internal/config/config.go b/services/gateway-service/internal/config/config.go index 05c3d678..96e52ed5 100644 --- a/services/gateway-service/internal/config/config.go +++ b/services/gateway-service/internal/config/config.go @@ -26,6 +26,7 @@ type Config struct { WalletServiceAddr string `yaml:"wallet_service_addr"` ActivityServiceAddr string `yaml:"activity_service_addr"` GameServiceAddr string `yaml:"game_service_addr"` + Statistics StatisticsConfig `yaml:"statistics"` GRPCClient GRPCClientConfig `yaml:"grpc_client"` AuthRateLimit AuthRateLimitConfig `yaml:"auth_rate_limit"` LoginRisk LoginRiskConfig `yaml:"login_risk"` @@ -122,6 +123,12 @@ type LeaderboardConfig struct { WalletMySQLDSN string `yaml:"wallet_mysql_dsn"` } +// StatisticsConfig 描述 gateway 转发客户端 H5 埋点到 statistics-service 的内部 HTTP 入口。 +type StatisticsConfig struct { + BaseURL string `yaml:"base_url"` + Timeout time.Duration `yaml:"timeout"` +} + // AuthRateLimitConfig 控制 gateway public auth 入口的 Redis 频控。 type AuthRateLimitConfig struct { // Enabled 为 false 时关闭 gateway auth 入口频控。 @@ -273,6 +280,10 @@ func Default() Config { WalletServiceAddr: "127.0.0.1:13004", ActivityServiceAddr: "127.0.0.1:13006", GameServiceAddr: "127.0.0.1:13008", + Statistics: StatisticsConfig{ + BaseURL: "http://127.0.0.1:13010", + Timeout: 2 * time.Second, + }, GRPCClient: GRPCClientConfig{ DefaultTimeout: 5 * time.Second, ConnectTimeout: 2 * time.Second, @@ -384,6 +395,7 @@ func (cfg *Config) Normalize() error { cfg.WalletServiceAddr = strings.TrimSpace(cfg.WalletServiceAddr) cfg.ActivityServiceAddr = strings.TrimSpace(cfg.ActivityServiceAddr) cfg.GameServiceAddr = strings.TrimSpace(cfg.GameServiceAddr) + cfg.Statistics.BaseURL = strings.TrimRight(strings.TrimSpace(cfg.Statistics.BaseURL), "/") if cfg.WalletServiceAddr == "" { cfg.WalletServiceAddr = "127.0.0.1:13004" } @@ -393,6 +405,12 @@ func (cfg *Config) Normalize() error { if cfg.GameServiceAddr == "" { cfg.GameServiceAddr = "127.0.0.1:13008" } + if cfg.Statistics.BaseURL == "" { + cfg.Statistics.BaseURL = "http://127.0.0.1:13010" + } + if cfg.Statistics.Timeout <= 0 { + cfg.Statistics.Timeout = 2 * time.Second + } runtimeGRPC := cfg.GRPCClient.Runtime() cfg.GRPCClient = GRPCClientConfig{ DefaultTimeout: runtimeGRPC.DefaultTimeout, diff --git a/services/gateway-service/internal/transport/http/activityapi/first_recharge_reward_handler.go b/services/gateway-service/internal/transport/http/activityapi/first_recharge_reward_handler.go index 4d0cef8f..6debeb97 100644 --- a/services/gateway-service/internal/transport/http/activityapi/first_recharge_reward_handler.go +++ b/services/gateway-service/internal/transport/http/activityapi/first_recharge_reward_handler.go @@ -30,6 +30,7 @@ type firstRechargeRewardItemData struct { ResourceType string `json:"resource_type"` ResourceName string `json:"resource_name"` CoverURL string `json:"cover_url"` + AssetURL string `json:"asset_url"` EffectURL string `json:"effect_url"` DurationMS int64 `json:"duration_ms"` Quantity int64 `json:"quantity"` @@ -196,8 +197,11 @@ func firstRechargeRewardItemsFromGroup(group checkinResourceGroupData) []firstRe if item.Resource != nil { reward.ResourceType = item.Resource.ResourceType reward.ResourceName = item.Resource.Name - reward.CoverURL = firstRechargeFirstNonEmpty(item.Resource.AssetURL, item.Resource.PreviewURL, item.Resource.AnimationURL) - reward.EffectURL = firstRechargeFirstNonEmpty(item.Resource.AnimationURL, item.Resource.PreviewURL, item.Resource.AssetURL) + // 首充弹窗格子只展示静态封面，点击后才播放资源资产；因此优先把 preview_url 下发为 cover_url， + // asset_url 保持资源主资产原值，effect_url 继续兜底旧客户端使用的动画预览字段。 + reward.CoverURL = firstRechargeFirstNonEmpty(item.Resource.PreviewURL, item.Resource.AssetURL, item.Resource.AnimationURL) + reward.AssetURL = firstRechargeFirstNonEmpty(item.Resource.AssetURL, item.Resource.AnimationURL) + reward.EffectURL = firstRechargeFirstNonEmpty(item.Resource.AnimationURL, item.Resource.AssetURL, item.Resource.PreviewURL) } else { reward.ResourceType = firstRechargeWalletAssetType(item.WalletAssetType) reward.ResourceName = firstRechargeWalletAssetType(item.WalletAssetType) diff --git a/services/gateway-service/internal/transport/http/activityapi/first_recharge_reward_handler_test.go b/services/gateway-service/internal/transport/http/activityapi/first_recharge_reward_handler_test.go new file mode 100644 index 00000000..100af070 --- /dev/null +++ b/services/gateway-service/internal/transport/http/activityapi/first_recharge_reward_handler_test.go @@ -0,0 +1,33 @@ +package activityapi + +import "testing" + +func TestFirstRechargeRewardItemsSplitCoverImageAndPreviewAsset(t *testing.T) { + items := firstRechargeRewardItemsFromGroup(checkinResourceGroupData{ + Items: []checkinResourceGroupItemData{{ + ResourceID: 88001, + Resource: &checkinResourceData{ + ResourceID: 88001, + ResourceType: "vehicle", + Name: "Rocket Car", + PreviewURL: "https://cdn.example.test/first-recharge/rocket-cover.png", + AssetURL: "https://cdn.example.test/first-recharge/rocket-preview.mp4", + }, + Quantity: 1, + DurationMS: 604800000, + }}, + }) + + if len(items) != 1 { + t.Fatalf("reward item count = %d", len(items)) + } + if items[0].CoverURL != "https://cdn.example.test/first-recharge/rocket-cover.png" { + t.Fatalf("cover_url should stay static image, got %q", items[0].CoverURL) + } + if items[0].AssetURL != "https://cdn.example.test/first-recharge/rocket-preview.mp4" { + t.Fatalf("asset_url should carry mp4 preview asset, got %q", items[0].AssetURL) + } + if items[0].EffectURL != "https://cdn.example.test/first-recharge/rocket-preview.mp4" { + t.Fatalf("effect_url should keep old preview fallback, got %q", items[0].EffectURL) + } +} diff --git a/services/gateway-service/internal/transport/http/gameapi/handler.go b/services/gateway-service/internal/transport/http/gameapi/handler.go index 642b50d7..e9377b53 100644 --- a/services/gateway-service/internal/transport/http/gameapi/handler.go +++ b/services/gateway-service/internal/transport/http/gameapi/handler.go @@ -14,6 +14,7 @@ type Handler struct { userAuthClient client.UserAuthClient userProfileClient client.UserProfileClient walletClient client.WalletClient + statisticsClient client.StatisticsClient // 热游固定回调地址先进入 hyapp gateway；chatapp3 老 token 需要原样转给 chatapp3，hyapp token 继续落到 game-service。 hotgameCompatHTTPClient *http.Client } @@ -23,6 +24,7 @@ type Config struct { UserAuthClient client.UserAuthClient UserProfileClient client.UserProfileClient WalletClient client.WalletClient + StatisticsClient client.StatisticsClient } func New(config Config) *Handler { @@ -31,6 +33,7 @@ func New(config Config) *Handler { userAuthClient: config.UserAuthClient, userProfileClient: config.UserProfileClient, walletClient: config.WalletClient, + statisticsClient: config.StatisticsClient, hotgameCompatHTTPClient: hotgameCompatHTTPClient(), } } @@ -56,6 +59,7 @@ func (h *Handler) Handlers() httproutes.GameHandlers { GetRPSMatch: h.getRPSMatch, RollRPSMatch: h.rollRPSMatch, CancelRPSMatch: h.cancelRPSMatch, + ReportSelfGameH5Event: h.reportSelfGameH5Event, GetRoomRPSConfig: h.getRoomRPSConfig, ListRoomRPSChallenges: h.listRoomRPSChallenges, CreateRoomRPSChallenge: h.createRoomRPSChallenge, diff --git a/services/gateway-service/internal/transport/http/gameapi/self_game_report_handler.go b/services/gateway-service/internal/transport/http/gameapi/self_game_report_handler.go new file mode 100644 index 00000000..31508664 --- /dev/null +++ b/services/gateway-service/internal/transport/http/gameapi/self_game_report_handler.go @@ -0,0 +1,95 @@ +package gameapi + +import ( + "net/http" + "strings" + "time" + + userv1 "hyapp.local/api/proto/user/v1" + "hyapp/pkg/appcode" + "hyapp/services/gateway-service/internal/auth" + "hyapp/services/gateway-service/internal/client" + "hyapp/services/gateway-service/internal/transport/http/httpkit" +) + +type selfGameH5EventRequest struct { + EventID string `json:"event_id"` + EventName string `json:"event_name"` + GameID string `json:"game_id"` + Page string `json:"page"` + SessionID string `json:"session_id"` + MatchID string `json:"match_id"` + Language string `json:"language"` + ClientVersion string `json:"client_version"` + H5Version string `json:"h5_version"` + EntrySource string `json:"entry_source"` + DurationMS int64 `json:"duration_ms"` + Success bool `json:"success"` + ErrorCode string `json:"error_code"` + OccurredAtMS int64 `json:"occurred_at_ms"` +} + +func (h *Handler) reportSelfGameH5Event(writer http.ResponseWriter, request *http.Request) { + if h.statisticsClient == nil || h.userProfileClient == nil { + httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error") + return + } + var body selfGameH5EventRequest + if err := httpkit.DecodeJSON(request.Body, &body); err != nil { + httpkit.WriteError(writer, request, http.StatusBadRequest, httpkit.CodeInvalidJSON, "invalid request body") + return + } + body.EventID = strings.TrimSpace(body.EventID) + body.EventName = strings.TrimSpace(body.EventName) + body.GameID = strings.TrimSpace(body.GameID) + if body.EventID == "" || body.EventName == "" || body.GameID == "" { + httpkit.WriteError(writer, request, http.StatusBadRequest, httpkit.CodeInvalidArgument, "invalid argument") + return + } + userID := auth.UserIDFromContext(request.Context()) + userResp, err := h.userProfileClient.GetUser(request.Context(), &userv1.GetUserRequest{ + Meta: httpkit.UserMeta(request, ""), + UserId: userID, + }) + if err != nil { + httpkit.WriteRPCError(writer, request, err) + return + } + if body.OccurredAtMS <= 0 { + body.OccurredAtMS = time.Now().UTC().UnixMilli() + } + language := strings.TrimSpace(body.Language) + if language == "" { + language = requestLanguage(request) + } + clientVersion := strings.TrimSpace(body.ClientVersion) + if clientVersion == "" { + clientVersion = httpkit.FirstHeader(request, "X-App-Version", "X-Client-Version") + } + // H5 上报只信任客户端的行为事实和耗时；用户、App、国家和区域由 gateway 根据登录态补齐，避免前端伪造维度污染统计口径。 + err = h.statisticsClient.ReportSelfGameH5Event(request.Context(), client.SelfGameH5Event{ + AppCode: appcode.FromContext(request.Context()), + EventID: body.EventID, + EventName: body.EventName, + GameID: body.GameID, + Page: body.Page, + SessionID: body.SessionID, + MatchID: body.MatchID, + UserID: userID, + CountryID: userResp.GetUser().GetCountryId(), + RegionID: userResp.GetUser().GetRegionId(), + Language: language, + ClientVersion: clientVersion, + H5Version: body.H5Version, + EntrySource: body.EntrySource, + DurationMS: body.DurationMS, + Success: body.Success, + ErrorCode: body.ErrorCode, + OccurredAtMS: body.OccurredAtMS, + }) + if err != nil { + httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error") + return + } + httpkit.WriteOK(writer, request, map[string]any{"ok": true}) +} diff --git a/services/gateway-service/internal/transport/http/handler.go b/services/gateway-service/internal/transport/http/handler.go index 531b4b43..17d178c8 100644 --- a/services/gateway-service/internal/transport/http/handler.go +++ b/services/gateway-service/internal/transport/http/handler.go @@ -50,6 +50,7 @@ type Handler struct { weeklyStar client.WeeklyStarClient broadcastClient client.BroadcastClient gameClient client.GameClient + statisticsClient client.StatisticsClient appConfigReader appapi.ConfigReader leaderboardWalletDB *sql.DB tencentIM TencentIMConfig @@ -268,6 +269,11 @@ func (h *Handler) SetGameClient(gameClient client.GameClient) { h.gameClient = gameClient } +// SetStatisticsClient 注入 statistics-service 内部 HTTP client；gateway 只转发 H5 埋点，不在入口层做统计聚合。 +func (h *Handler) SetStatisticsClient(statisticsClient client.StatisticsClient) { + h.statisticsClient = statisticsClient +} + // SetLeaderboardWalletDB 注入 App 榜单只读钱包库连接。 func (h *Handler) SetLeaderboardWalletDB(db *sql.DB) { h.leaderboardWalletDB = db diff --git a/services/gateway-service/internal/transport/http/httproutes/router.go b/services/gateway-service/internal/transport/http/httproutes/router.go index 00f5abc7..f1e557ed 100644 --- a/services/gateway-service/internal/transport/http/httproutes/router.go +++ b/services/gateway-service/internal/transport/http/httproutes/router.go @@ -272,6 +272,7 @@ type GameHandlers struct { GetRPSMatch http.HandlerFunc RollRPSMatch http.HandlerFunc CancelRPSMatch http.HandlerFunc + ReportSelfGameH5Event http.HandlerFunc GetRoomRPSConfig http.HandlerFunc ListRoomRPSChallenges http.HandlerFunc CreateRoomRPSChallenge http.HandlerFunc @@ -593,6 +594,7 @@ func (r routes) registerGameRoutes() { r.profile("/games/rps/matches/{match_id}", http.MethodGet, h.GetRPSMatch) r.profile("/games/rps/matches/{match_id}/roll", http.MethodPost, h.RollRPSMatch) r.profile("/games/rps/matches/{match_id}/cancel", http.MethodPost, h.CancelRPSMatch) + r.profile("/games/self/report-events", http.MethodPost, h.ReportSelfGameH5Event) r.profile("/games/room-rps/config", http.MethodGet, h.GetRoomRPSConfig) r.profile("/games/room-rps/challenges", http.MethodGet, h.ListRoomRPSChallenges) // 当前 ServeMux 包装按 path 注册，不能把 GET 列表和 POST 创建同时挂在 /challenges；创建动作单独放 /create，避免路由覆盖。 diff --git a/services/gateway-service/internal/transport/http/router.go b/services/gateway-service/internal/transport/http/router.go index e7649d05..04fd7d2c 100644 --- a/services/gateway-service/internal/transport/http/router.go +++ b/services/gateway-service/internal/transport/http/router.go @@ -90,6 +90,7 @@ func (h *Handler) Routes(jwtVerifier *auth.Verifier) http.Handler { UserAuthClient: h.userClient, UserProfileClient: h.userProfileClient, WalletClient: h.walletClient, + StatisticsClient: h.statisticsClient, }) resourceAPI := resourceapi.New(resourceapi.Config{ WalletClient: h.walletClient, diff --git a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql index b4b707e6..e998a2ce 100644 --- a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql +++ b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql @@ -134,3 +134,126 @@ CREATE TABLE IF NOT EXISTS stat_game_day_players ( PRIMARY KEY (app_code, stat_day, country_id, platform_code, game_id, user_id), KEY idx_stat_game_player_region (app_code, stat_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='游戏日参与用户去重表'; + +CREATE TABLE IF NOT EXISTS stat_self_game_h5_event_day ( + app_code VARCHAR(32) NOT NULL, + stat_day DATE NOT NULL, + country_id BIGINT NOT NULL DEFAULT 0, + region_id BIGINT NOT NULL DEFAULT 0, + game_id VARCHAR(96) NOT NULL, + event_name VARCHAR(64) NOT NULL, + language VARCHAR(16) NOT NULL DEFAULT '', + client_version VARCHAR(64) NOT NULL DEFAULT '', + h5_version VARCHAR(64) NOT NULL DEFAULT '', + entry_source VARCHAR(64) NOT NULL DEFAULT '', + event_count BIGINT NOT NULL DEFAULT 0, + user_count BIGINT NOT NULL DEFAULT 0, + success_count BIGINT NOT NULL DEFAULT 0, + error_count BIGINT NOT NULL DEFAULT 0, + duration_ms_sum BIGINT NOT NULL DEFAULT 0, + updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, stat_day, country_id, region_id, game_id, event_name, language, client_version, h5_version, entry_source), + KEY idx_stat_self_game_h5_day (app_code, stat_day, game_id, event_name), + KEY idx_stat_self_game_h5_region (app_code, stat_day, region_id, game_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='自研游戏 H5 事件日统计表'; + +CREATE TABLE IF NOT EXISTS stat_self_game_h5_event_users ( + app_code VARCHAR(32) NOT NULL, + stat_day DATE NOT NULL, + game_id VARCHAR(96) NOT NULL, + event_name VARCHAR(64) NOT NULL, + user_id BIGINT NOT NULL, + first_seen_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, stat_day, game_id, event_name, user_id), + KEY idx_stat_self_game_h5_users_day (app_code, stat_day, game_id, event_name) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='自研游戏 H5 事件用户去重表'; + +CREATE TABLE IF NOT EXISTS stat_self_game_match_day ( + app_code VARCHAR(32) NOT NULL, + stat_day DATE NOT NULL, + country_id BIGINT NOT NULL DEFAULT 0, + region_id BIGINT NOT NULL DEFAULT 0, + game_id VARCHAR(96) NOT NULL, + stake_coin BIGINT NOT NULL DEFAULT 0, + created_matches BIGINT NOT NULL DEFAULT 0, + matched_matches BIGINT NOT NULL DEFAULT 0, + settled_matches BIGINT NOT NULL DEFAULT 0, + canceled_matches BIGINT NOT NULL DEFAULT 0, + failed_matches BIGINT NOT NULL DEFAULT 0, + human_matches BIGINT NOT NULL DEFAULT 0, + robot_matches BIGINT NOT NULL DEFAULT 0, + human_participants BIGINT NOT NULL DEFAULT 0, + robot_participants BIGINT NOT NULL DEFAULT 0, + winner_count BIGINT NOT NULL DEFAULT 0, + loser_count BIGINT NOT NULL DEFAULT 0, + draw_count BIGINT NOT NULL DEFAULT 0, + user_stake_coin BIGINT NOT NULL DEFAULT 0, + robot_stake_coin BIGINT NOT NULL DEFAULT 0, + payout_coin BIGINT NOT NULL DEFAULT 0, + refund_coin BIGINT NOT NULL DEFAULT 0, + platform_profit_coin BIGINT NOT NULL DEFAULT 0, + pool_delta_coin BIGINT NOT NULL DEFAULT 0, + forced_player_lose_matches BIGINT NOT NULL DEFAULT 0, + wait_ms_sum BIGINT NOT NULL DEFAULT 0, + duration_ms_sum BIGINT NOT NULL DEFAULT 0, + updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, stat_day, country_id, region_id, game_id, stake_coin), + KEY idx_stat_self_game_match_day (app_code, stat_day, game_id), + KEY idx_stat_self_game_match_region (app_code, stat_day, region_id, game_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='自研游戏对局日统计表'; + +CREATE TABLE IF NOT EXISTS stat_self_game_pool_day ( + app_code VARCHAR(32) NOT NULL, + stat_day DATE NOT NULL, + game_id VARCHAR(96) NOT NULL, + direction VARCHAR(16) NOT NULL DEFAULT '', + reason VARCHAR(64) NOT NULL DEFAULT '', + transaction_count BIGINT NOT NULL DEFAULT 0, + in_coin BIGINT NOT NULL DEFAULT 0, + out_coin BIGINT NOT NULL DEFAULT 0, + balance_after_coin BIGINT NOT NULL DEFAULT 0, + updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, stat_day, game_id, direction, reason), + KEY idx_stat_self_game_pool_day (app_code, stat_day, game_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='自研游戏奖池日统计表'; + +CREATE TABLE IF NOT EXISTS stat_self_game_participant_day ( + app_code VARCHAR(32) NOT NULL, + stat_day DATE NOT NULL, + country_id BIGINT NOT NULL DEFAULT 0, + region_id BIGINT NOT NULL DEFAULT 0, + game_id VARCHAR(96) NOT NULL, + stake_coin BIGINT NOT NULL DEFAULT 0, + participant_type VARCHAR(32) NOT NULL DEFAULT 'user', + result VARCHAR(32) NOT NULL DEFAULT '', + rps_gesture VARCHAR(32) NOT NULL DEFAULT '', + dice_point INT NOT NULL DEFAULT 0, + participant_count BIGINT NOT NULL DEFAULT 0, + payout_coin BIGINT NOT NULL DEFAULT 0, + net_win_coin BIGINT NOT NULL DEFAULT 0, + updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, stat_day, country_id, region_id, game_id, stake_coin, participant_type, result, rps_gesture, dice_point), + KEY idx_stat_self_game_participant_day (app_code, stat_day, game_id), + KEY idx_stat_self_game_participant_region (app_code, stat_day, region_id, game_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='自研游戏参与者结果分布日统计表'; + +CREATE TABLE IF NOT EXISTS stat_self_game_user_day ( + app_code VARCHAR(32) NOT NULL, + stat_day DATE NOT NULL, + country_id BIGINT NOT NULL DEFAULT 0, + region_id BIGINT NOT NULL DEFAULT 0, + game_id VARCHAR(96) NOT NULL, + user_id BIGINT NOT NULL, + match_count BIGINT NOT NULL DEFAULT 0, + win_count BIGINT NOT NULL DEFAULT 0, + lose_count BIGINT NOT NULL DEFAULT 0, + draw_count BIGINT NOT NULL DEFAULT 0, + cancel_count BIGINT NOT NULL DEFAULT 0, + stake_coin BIGINT NOT NULL DEFAULT 0, + payout_coin BIGINT NOT NULL DEFAULT 0, + net_win_coin BIGINT NOT NULL DEFAULT 0, + updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, stat_day, game_id, user_id), + KEY idx_stat_self_game_user_win (app_code, stat_day, game_id, net_win_coin), + KEY idx_stat_self_game_user_region (app_code, stat_day, region_id, game_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='自研游戏用户风险日统计表'; diff --git a/services/statistics-service/internal/app/app.go b/services/statistics-service/internal/app/app.go index 83a5f2c3..32d67553 100644 --- a/services/statistics-service/internal/app/app.go +++ b/services/statistics-service/internal/app/app.go @@ -196,11 +196,21 @@ func newConsumers(cfg config.Config, repo *mysqlstorage.Repository) ([]*rocketmq return nil, err } if err := add(cfg.RocketMQ.GameOutbox, gamemq.TagGameOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { - event, ok, err := gameOrderEvent(message.Body) - if err != nil || !ok { - return err + if event, ok, err := gameOrderEvent(message.Body); err != nil || ok { + if err != nil || !ok { + return err + } + return repo.ConsumeGameOrder(appcode.WithContext(ctx, event.AppCode), event) } - return repo.ConsumeGameOrder(appcode.WithContext(ctx, event.AppCode), event) + match, ok, err := selfGameMatchEvent(message.Body) + if err != nil || !ok { + pool, poolOK, poolErr := selfGamePoolAdjustmentEvent(message.Body) + if poolErr != nil || !poolOK { + return poolErr + } + return repo.ConsumeSelfGamePoolAdjustment(appcode.WithContext(ctx, pool.AppCode), pool) + } + return repo.ConsumeSelfGameMatch(appcode.WithContext(ctx, match.AppCode), match) }); err != nil { shutdownConsumers(consumers) return nil, err @@ -431,6 +441,119 @@ func gameOrderEvent(body []byte) (mysqlstorage.GameOrderEvent, bool, error) { }, true, nil } +func selfGameMatchEvent(body []byte) (mysqlstorage.SelfGameMatchEvent, bool, error) { + message, err := gamemq.DecodeGameOutboxMessage(body) + if err != nil { + return mysqlstorage.SelfGameMatchEvent{}, false, err + } + switch message.EventType { + case gamemq.EventTypeSelfGameMatchCreated, + gamemq.EventTypeSelfGameMatchReady, + gamemq.EventTypeSelfGameMatchSettled, + gamemq.EventTypeSelfGameMatchCanceled, + gamemq.EventTypeSelfGameMatchFailed: + default: + return mysqlstorage.SelfGameMatchEvent{}, false, nil + } + payload := mysqlstorage.DecodeJSON(message.PayloadJSON) + event := mysqlstorage.SelfGameMatchEvent{ + AppCode: message.AppCode, + EventID: message.EventID, + EventType: message.EventType, + MatchID: firstNonEmpty(mysqlstorage.String(payload, "match_id"), message.OrderID), + GameID: firstNonEmpty(mysqlstorage.String(payload, "game_id"), message.GameID), + PlatformCode: firstNonEmpty(mysqlstorage.String(payload, "platform_code"), message.PlatformCode), + RoomID: mysqlstorage.String(payload, "room_id"), + RegionID: mysqlstorage.Int64(payload, "region_id"), + StakeCoin: mysqlstorage.Int64(payload, "stake_coin"), + Status: mysqlstorage.String(payload, "status"), + Result: mysqlstorage.String(payload, "result"), + MatchMode: mysqlstorage.String(payload, "match_mode"), + ForcedResult: mysqlstorage.String(payload, "forced_result"), + UserParticipants: mysqlstorage.Int64(payload, "user_participants"), + RobotParticipants: mysqlstorage.Int64(payload, "robot_participants"), + WinnerCount: mysqlstorage.Int64(payload, "winner_count"), + LoserCount: mysqlstorage.Int64(payload, "loser_count"), + DrawCount: mysqlstorage.Int64(payload, "draw_count"), + UserStakeCoin: mysqlstorage.Int64(payload, "user_stake_coin"), + RobotStakeCoin: mysqlstorage.Int64(payload, "robot_stake_coin"), + PayoutCoin: mysqlstorage.Int64(payload, "payout_coin"), + RefundCoin: mysqlstorage.Int64(payload, "refund_coin"), + PlatformProfitCoin: mysqlstorage.Int64(payload, "platform_profit_coin"), + PoolDeltaCoin: mysqlstorage.Int64(payload, "pool_delta_coin"), + WaitMS: mysqlstorage.Int64(payload, "wait_ms"), + DurationMS: mysqlstorage.Int64(payload, "duration_ms"), + Participants: selfGameParticipants(payload["participants"]), + OccurredAtMS: message.OccurredAtMS, + } + return event, true, nil +} + +func selfGamePoolAdjustmentEvent(body []byte) (mysqlstorage.SelfGamePoolAdjustmentEvent, bool, error) { + message, err := gamemq.DecodeGameOutboxMessage(body) + if err != nil { + return mysqlstorage.SelfGamePoolAdjustmentEvent{}, false, err + } + if message.EventType != gamemq.EventTypeSelfGamePoolAdjusted { + return mysqlstorage.SelfGamePoolAdjustmentEvent{}, false, nil + } + payload := mysqlstorage.DecodeJSON(message.PayloadJSON) + return mysqlstorage.SelfGamePoolAdjustmentEvent{ + AppCode: message.AppCode, + EventID: message.EventID, + GameID: firstNonEmpty(mysqlstorage.String(payload, "game_id"), message.GameID), + Direction: mysqlstorage.String(payload, "direction"), + Reason: mysqlstorage.String(payload, "reason"), + AmountCoin: mysqlstorage.Int64(payload, "amount_coin"), + BalanceAfter: mysqlstorage.Int64(payload, "balance_after"), + OccurredAtMS: message.OccurredAtMS, + }, true, nil +} + +func selfGameParticipants(raw any) []mysqlstorage.SelfGameMatchParticipantEvent { + values, ok := raw.([]any) + if !ok { + return nil + } + participants := make([]mysqlstorage.SelfGameMatchParticipantEvent, 0, len(values)) + for _, value := range values { + payload, ok := value.(map[string]any) + if !ok { + continue + } + participants = append(participants, mysqlstorage.SelfGameMatchParticipantEvent{ + UserID: mysqlstorage.Int64(payload, "user_id"), + ParticipantType: mysqlstorage.String(payload, "participant_type"), + StakeCoin: mysqlstorage.Int64(payload, "stake_coin"), + DicePoints: int64Slice(payload["dice_points"]), + RPSGesture: mysqlstorage.String(payload, "rps_gesture"), + Result: mysqlstorage.String(payload, "result"), + PayoutCoin: mysqlstorage.Int64(payload, "payout_coin"), + NetWinCoin: mysqlstorage.Int64(payload, "net_win_coin"), + }) + } + return participants +} + +func int64Slice(raw any) []int64 { + values, ok := raw.([]any) + if !ok { + return nil + } + out := make([]int64, 0, len(values)) + for _, value := range values { + switch typed := value.(type) { + case float64: + out = append(out, int64(typed)) + case int64: + out = append(out, typed) + case int: + out = append(out, int64(typed)) + } + } + return out +} + func startConsumers(consumers []*rocketmqx.Consumer) error { for _, consumer := range consumers { if err := consumer.Start(); err != nil { diff --git a/services/statistics-service/internal/app/local_flow_test.go b/services/statistics-service/internal/app/local_flow_test.go index f981acb6..13c53cb5 100644 --- a/services/statistics-service/internal/app/local_flow_test.go +++ b/services/statistics-service/internal/app/local_flow_test.go @@ -10,6 +10,7 @@ import ( "google.golang.org/protobuf/proto" roomeventsv1 "hyapp.local/api/proto/events/room/v1" "hyapp/internal/testutil/mysqlschema" + "hyapp/pkg/appcode" "hyapp/pkg/gamemq" "hyapp/pkg/roommq" "hyapp/pkg/walletmq" @@ -201,6 +202,166 @@ func TestLocalStatisticsDataScreenFlow(t *testing.T) { } } +func TestLocalSelfGameStatisticsFlow(t *testing.T) { + ctx := context.Background() + _, file, _, _ := runtime.Caller(0) + schema := mysqlschema.New(t, mysqlschema.Config{ + InitDBPath: mysqlschema.InitDBPath(t, file, "../../deploy/mysql/initdb/001_statistics_service.sql"), + DatabasePrefix: "hy_self_game_stats", + }) + repo, err := mysqlstorage.Open(ctx, schema.DSN) + if err != nil { + t.Fatalf("open statistics repository failed: %v", err) + } + t.Cleanup(func() { _ = repo.Close() }) + + occurredAt := time.Date(2026, 6, 11, 12, 0, 0, 0, time.UTC).UnixMilli() + dayStart := time.Date(2026, 6, 11, 0, 0, 0, 0, time.UTC).UnixMilli() + const ( + appCode = "lalu" + regionID = int64(210) + userID = int64(42) + ) + appCtx := appcode.WithContext(ctx, appCode) + for _, eventName := range []string{"page_open", "match_click", "match_success", "settlement_show", "play_again"} { + if err := repo.ConsumeSelfGameH5Event(appCtx, mysqlstorage.SelfGameH5Event{ + AppCode: appCode, + EventID: "h5:" + eventName, + EventName: eventName, + GameID: "dice", + UserID: userID, + RegionID: regionID, + H5Version: "dice-h5-test", + EntrySource: "room", + DurationMS: 20, + Success: true, + OccurredAtMS: occurredAt, + }); err != nil { + t.Fatalf("consume h5 event %s failed: %v", eventName, err) + } + } + + body, err := gamemq.EncodeGameOutboxMessage(gamemq.GameOutboxMessage{ + AppCode: appCode, + EventID: "SelfGameMatchSettled:dice_match_1", + EventType: gamemq.EventTypeSelfGameMatchSettled, + OrderID: "", + UserID: userID, + PlatformCode: "self", + GameID: "dice", + PayloadJSON: mustJSON(t, map[string]any{ + "match_id": "dice_match_1", + "game_id": "dice", + "platform_code": "self", + "region_id": regionID, + "stake_coin": int64(500), + "status": "settled", + "result": "win", + "match_mode": "robot", + "forced_result": "", + "user_participants": int64(1), + "robot_participants": int64(1), + "winner_count": int64(1), + "loser_count": int64(1), + "draw_count": int64(0), + "user_stake_coin": int64(500), + "robot_stake_coin": int64(500), + "payout_coin": int64(900), + "refund_coin": int64(0), + "platform_profit_coin": int64(100), + "pool_delta_coin": int64(50), + "wait_ms": int64(1200), + "duration_ms": int64(4800), + "participants": []map[string]any{ + { + "user_id": userID, + "participant_type": "user", + "stake_coin": int64(500), + "dice_points": []int64{6}, + "result": "win", + "payout_coin": int64(900), + "net_win_coin": int64(400), + }, + { + "user_id": int64(0), + "participant_type": "robot", + "stake_coin": int64(500), + "dice_points": []int64{2}, + "result": "lose", + "payout_coin": int64(0), + "net_win_coin": int64(-500), + }, + }, + }), + OccurredAtMS: occurredAt, + }) + if err != nil { + t.Fatalf("encode self game match failed: %v", err) + } + match, ok, err := selfGameMatchEvent(body) + if err != nil || !ok { + t.Fatalf("decode self game match failed: ok=%v err=%v", ok, err) + } + if err := repo.ConsumeSelfGameMatch(appCtx, match); err != nil { + t.Fatalf("consume self game match failed: %v", err) + } + poolBody, err := gamemq.EncodeGameOutboxMessage(gamemq.GameOutboxMessage{ + AppCode: appCode, + EventID: "SelfGamePoolAdjusted:pool_adjust_1", + EventType: gamemq.EventTypeSelfGamePoolAdjusted, + UserID: userID, + PlatformCode: "self", + GameID: "dice", + PayloadJSON: mustJSON(t, map[string]any{ + "adjustment_id": "pool_adjust_1", + "game_id": "dice", + "match_id": "dice_match_1", + "user_id": userID, + "direction": "in", + "amount_coin": int64(50), + "reason": "match_pool_in", + "balance_after": int64(1050), + }), + OccurredAtMS: occurredAt, + }) + if err != nil { + t.Fatalf("encode self game pool failed: %v", err) + } + pool, ok, err := selfGamePoolAdjustmentEvent(poolBody) + if err != nil || !ok { + t.Fatalf("decode self game pool failed: ok=%v err=%v", ok, err) + } + if err := repo.ConsumeSelfGamePoolAdjustment(appCtx, pool); err != nil { + t.Fatalf("consume self game pool failed: %v", err) + } + + overview, err := repo.QuerySelfGameOverview(ctx, mysqlstorage.SelfGameOverviewQuery{ + AppCode: appCode, + StartMS: dayStart, + EndMS: dayStart + int64(24*time.Hour/time.Millisecond), + GameID: "dice", + RegionID: regionID, + }) + if err != nil { + t.Fatalf("query self game overview failed: %v", err) + } + if overview.Conversion.PageToMatchClickRate != 1 || overview.Conversion.MatchClickToSuccessRate != 1 || overview.Conversion.SuccessToSettlementRate != 1 || overview.Conversion.SettlementToPlayAgainRate != 1 { + t.Fatalf("self game funnel mismatch: %+v", overview.Conversion) + } + if overview.MatchTotals.SettledMatches != 1 || overview.MatchTotals.RobotMatches != 1 || overview.MatchTotals.UserStakeCoin != 500 || overview.MatchTotals.PayoutCoin != 900 || overview.MatchTotals.PlatformProfitCoin != 100 || overview.MatchTotals.PoolDeltaCoin != 50 { + t.Fatalf("self game match totals mismatch: %+v", overview.MatchTotals) + } + if len(overview.StakeBreakdown) != 1 || overview.StakeBreakdown[0].StakeCoin != 500 || overview.StakeBreakdown[0].PlatformProfitCoin != 100 { + t.Fatalf("self game stake breakdown mismatch: %+v", overview.StakeBreakdown) + } + if len(overview.PoolStats) != 1 || overview.PoolStats[0].InCoin != 50 || overview.PoolStats[0].BalanceAfterCoin != 1050 { + t.Fatalf("self game pool stats mismatch: %+v", overview.PoolStats) + } + if len(overview.UserRisk) != 1 || overview.UserRisk[0].UserID != userID || overview.UserRisk[0].WinCount != 1 || overview.UserRisk[0].NetWinCoin != 400 { + t.Fatalf("self game user risk mismatch: %+v", overview.UserRisk) + } +} + func walletMessage(t *testing.T, message walletmq.WalletOutboxMessage) []byte { t.Helper() body, err := walletmq.EncodeWalletOutboxMessage(message) diff --git a/services/statistics-service/internal/app/query_http.go b/services/statistics-service/internal/app/query_http.go index 81a5897b..716be60e 100644 --- a/services/statistics-service/internal/app/query_http.go +++ b/services/statistics-service/internal/app/query_http.go @@ -10,6 +10,7 @@ import ( "strings" "time" + "hyapp/pkg/appcode" mysqlstorage "hyapp/services/statistics-service/internal/storage/mysql" ) @@ -31,6 +32,8 @@ func newQueryHTTPServer(addr string, repo *mysqlstorage.Repository) (*queryHTTPS s := &queryHTTPServer{repo: repo, listener: listener} mux := http.NewServeMux() mux.HandleFunc("/internal/v1/statistics/overview", s.overview) + mux.HandleFunc("/internal/v1/statistics/self-game/events", s.selfGameEvents) + mux.HandleFunc("/internal/v1/statistics/self-games/overview", s.selfGameOverview) s.server = &http.Server{Handler: mux, ReadHeaderTimeout: 2 * time.Second} return s, nil } @@ -80,6 +83,105 @@ func (s *queryHTTPServer) overview(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, overview) } +type selfGameH5EventRequest struct { + AppCode string `json:"app_code"` + EventID string `json:"event_id"` + EventName string `json:"event_name"` + GameID string `json:"game_id"` + Page string `json:"page"` + SessionID string `json:"session_id"` + MatchID string `json:"match_id"` + UserID int64 `json:"user_id"` + CountryID int64 `json:"country_id"` + RegionID int64 `json:"region_id"` + Language string `json:"language"` + ClientVersion string `json:"client_version"` + H5Version string `json:"h5_version"` + EntrySource string `json:"entry_source"` + DurationMS int64 `json:"duration_ms"` + Success bool `json:"success"` + ErrorCode string `json:"error_code"` + OccurredAtMS int64 `json:"occurred_at_ms"` +} + +func (s *queryHTTPServer) selfGameEvents(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeJSON(w, http.StatusMethodNotAllowed, map[string]any{"error": "method not allowed"}) + return + } + defer r.Body.Close() + var body selfGameH5EventRequest + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]any{"error": "invalid json"}) + return + } + body.EventID = strings.TrimSpace(body.EventID) + body.EventName = strings.TrimSpace(body.EventName) + if body.EventID == "" || body.EventName == "" { + writeJSON(w, http.StatusBadRequest, map[string]any{"error": "event_id and event_name are required"}) + return + } + if body.OccurredAtMS <= 0 { + body.OccurredAtMS = time.Now().UTC().UnixMilli() + } + app := appcode.Normalize(body.AppCode) + if app == "" { + app = "lalu" + } + // 内部 ingest 只承接 gateway 已解析过的用户、App 和维度字段；statistics-service 只做幂等聚合，不反查业务库。 + err := s.repo.ConsumeSelfGameH5Event(appcode.WithContext(r.Context(), app), mysqlstorage.SelfGameH5Event{ + AppCode: app, + EventID: body.EventID, + EventName: body.EventName, + GameID: body.GameID, + Page: body.Page, + SessionID: body.SessionID, + MatchID: body.MatchID, + UserID: body.UserID, + CountryID: body.CountryID, + RegionID: body.RegionID, + Language: body.Language, + ClientVersion: body.ClientVersion, + H5Version: body.H5Version, + EntrySource: body.EntrySource, + DurationMS: body.DurationMS, + Success: body.Success, + ErrorCode: body.ErrorCode, + OccurredAtMS: body.OccurredAtMS, + }) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()}) + return + } + writeJSON(w, http.StatusOK, map[string]any{"ok": true}) +} + +func (s *queryHTTPServer) selfGameOverview(w http.ResponseWriter, r *http.Request) { + query := r.URL.Query() + startMS := parseInt64(query.Get("start_ms")) + endMS := parseInt64(query.Get("end_ms")) + if startMS <= 0 { + now := time.Now().UTC() + startMS = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC).UnixMilli() + } + if endMS <= startMS { + endMS = startMS + 24*time.Hour.Milliseconds() + } + overview, err := s.repo.QuerySelfGameOverview(r.Context(), mysqlstorage.SelfGameOverviewQuery{ + AppCode: query.Get("app_code"), + StartMS: startMS, + EndMS: endMS, + GameID: query.Get("game_id"), + CountryID: parseInt64(query.Get("country_id")), + RegionID: parseInt64(query.Get("region_id")), + }) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()}) + return + } + writeJSON(w, http.StatusOK, overview) +} + func parseInt64(value string) int64 { parsed, _ := strconv.ParseInt(strings.TrimSpace(value), 10, 64) return parsed diff --git a/services/statistics-service/internal/storage/mysql/query.go b/services/statistics-service/internal/storage/mysql/query.go index 7544bc03..1475f361 100644 --- a/services/statistics-service/internal/storage/mysql/query.go +++ b/services/statistics-service/internal/storage/mysql/query.go @@ -155,6 +155,144 @@ type CountryBreakdown struct { ARPPUUSDMinor int64 `json:"arppu_usd_minor"` } +type SelfGameOverviewQuery struct { + AppCode string + StartMS int64 + EndMS int64 + GameID string + CountryID int64 + RegionID int64 +} + +type SelfGameOverview struct { + AppCode string `json:"app_code"` + StartMS int64 `json:"start_ms"` + EndMS int64 `json:"end_ms"` + GameID string `json:"game_id"` + CountryID int64 `json:"country_id"` + RegionID int64 `json:"region_id"` + Events []SelfGameEventStat `json:"events"` + MatchTotals SelfGameMatchTotals `json:"match_totals"` + StakeBreakdown []SelfGameStakeStat `json:"stake_breakdown"` + PoolStats []SelfGamePoolStat `json:"pool_stats"` + ParticipantDistribution []SelfGameParticipantStat `json:"participant_distribution"` + UserRisk []SelfGameUserRiskStat `json:"user_risk"` + Funnel []SelfGameFunnelStep `json:"funnel"` + Conversion SelfGameConversion `json:"conversion"` +} + +type SelfGameEventStat struct { + GameID string `json:"game_id"` + EventName string `json:"event_name"` + EventCount int64 `json:"event_count"` + UserCount int64 `json:"user_count"` + SuccessCount int64 `json:"success_count"` + ErrorCount int64 `json:"error_count"` + DurationMSSum int64 `json:"duration_ms_sum"` + AvgDurationMS int64 `json:"avg_duration_ms"` +} + +type SelfGameMatchTotals struct { + CreatedMatches int64 `json:"created_matches"` + MatchedMatches int64 `json:"matched_matches"` + SettledMatches int64 `json:"settled_matches"` + CanceledMatches int64 `json:"canceled_matches"` + FailedMatches int64 `json:"failed_matches"` + HumanMatches int64 `json:"human_matches"` + RobotMatches int64 `json:"robot_matches"` + HumanParticipants int64 `json:"human_participants"` + RobotParticipants int64 `json:"robot_participants"` + WinnerCount int64 `json:"winner_count"` + LoserCount int64 `json:"loser_count"` + DrawCount int64 `json:"draw_count"` + UserStakeCoin int64 `json:"user_stake_coin"` + RobotStakeCoin int64 `json:"robot_stake_coin"` + PayoutCoin int64 `json:"payout_coin"` + RefundCoin int64 `json:"refund_coin"` + PlatformProfitCoin int64 `json:"platform_profit_coin"` + PoolDeltaCoin int64 `json:"pool_delta_coin"` + ForcedPlayerLoseMatches int64 `json:"forced_player_lose_matches"` + WaitMSSum int64 `json:"wait_ms_sum"` + DurationMSSum int64 `json:"duration_ms_sum"` + AvgWaitMS int64 `json:"avg_wait_ms"` + AvgDurationMS int64 `json:"avg_duration_ms"` + CancelRate float64 `json:"cancel_rate"` + FailRate float64 `json:"fail_rate"` + RobotMatchRate float64 `json:"robot_match_rate"` + PlatformProfitRate float64 `json:"platform_profit_rate"` +} + +type SelfGameStakeStat struct { + GameID string `json:"game_id"` + StakeCoin int64 `json:"stake_coin"` + CreatedMatches int64 `json:"created_matches"` + MatchedMatches int64 `json:"matched_matches"` + SettledMatches int64 `json:"settled_matches"` + CanceledMatches int64 `json:"canceled_matches"` + FailedMatches int64 `json:"failed_matches"` + HumanParticipants int64 `json:"human_participants"` + RobotParticipants int64 `json:"robot_participants"` + UserStakeCoin int64 `json:"user_stake_coin"` + RobotStakeCoin int64 `json:"robot_stake_coin"` + PayoutCoin int64 `json:"payout_coin"` + RefundCoin int64 `json:"refund_coin"` + PlatformProfitCoin int64 `json:"platform_profit_coin"` + PoolDeltaCoin int64 `json:"pool_delta_coin"` + PlatformProfitRate float64 `json:"platform_profit_rate"` +} + +type SelfGamePoolStat struct { + GameID string `json:"game_id"` + Direction string `json:"direction"` + Reason string `json:"reason"` + TransactionCount int64 `json:"transaction_count"` + InCoin int64 `json:"in_coin"` + OutCoin int64 `json:"out_coin"` + BalanceAfterCoin int64 `json:"balance_after_coin"` +} + +type SelfGameParticipantStat struct { + GameID string `json:"game_id"` + StakeCoin int64 `json:"stake_coin"` + ParticipantType string `json:"participant_type"` + Result string `json:"result"` + RPSGesture string `json:"rps_gesture"` + DicePoint int64 `json:"dice_point"` + ParticipantCount int64 `json:"participant_count"` + PayoutCoin int64 `json:"payout_coin"` + NetWinCoin int64 `json:"net_win_coin"` + WinRate float64 `json:"win_rate"` +} + +type SelfGameUserRiskStat struct { + GameID string `json:"game_id"` + UserID int64 `json:"user_id"` + MatchCount int64 `json:"match_count"` + WinCount int64 `json:"win_count"` + LoseCount int64 `json:"lose_count"` + DrawCount int64 `json:"draw_count"` + CancelCount int64 `json:"cancel_count"` + StakeCoin int64 `json:"stake_coin"` + PayoutCoin int64 `json:"payout_coin"` + NetWinCoin int64 `json:"net_win_coin"` + WinRate float64 `json:"win_rate"` + CancelRate float64 `json:"cancel_rate"` +} + +type SelfGameFunnelStep struct { + EventName string `json:"event_name"` + EventCount int64 `json:"event_count"` + UserCount int64 `json:"user_count"` +} + +type SelfGameConversion struct { + PageToMatchClickRate float64 `json:"page_to_match_click_rate"` + MatchClickToSuccessRate float64 `json:"match_click_to_success_rate"` + SuccessToSettlementRate float64 `json:"success_to_settlement_rate"` + SettlementToPlayAgainRate float64 `json:"settlement_to_play_again_rate"` + PageToSettlementRate float64 `json:"page_to_settlement_rate"` +} + func (r *Repository) QueryOverview(ctx context.Context, query OverviewQuery) (Overview, error) { app := appcode.Normalize(query.AppCode) if app == "" { @@ -233,6 +371,57 @@ func (r *Repository) QueryOverview(ctx context.Context, query OverviewQuery) (Ov return overview, nil } +func (r *Repository) QuerySelfGameOverview(ctx context.Context, query SelfGameOverviewQuery) (SelfGameOverview, error) { + startMS, endMS := normalizeRange(query.StartMS, query.EndMS) + app := appcode.Normalize(query.AppCode) + if app == "" { + app = "lalu" + } + startDay := statDay(startMS) + endDay := statDay(endMS - 1) + gameID := normalizeShortText(query.GameID) + overview := SelfGameOverview{ + AppCode: app, + StartMS: startMS, + EndMS: endMS, + GameID: gameID, + CountryID: query.CountryID, + RegionID: query.RegionID, + } + events, err := r.querySelfGameEvents(ctx, app, startDay, endDay, gameID, query.CountryID, query.RegionID) + if err != nil { + return SelfGameOverview{}, err + } + overview.Events = events + overview.Funnel, overview.Conversion = buildSelfGameFunnel(events) + totals, err := r.querySelfGameMatchTotals(ctx, app, startDay, endDay, gameID, query.CountryID, query.RegionID) + if err != nil { + return SelfGameOverview{}, err + } + overview.MatchTotals = totals + stakes, err := r.querySelfGameStakeBreakdown(ctx, app, startDay, endDay, gameID, query.CountryID, query.RegionID) + if err != nil { + return SelfGameOverview{}, err + } + overview.StakeBreakdown = stakes + pools, err := r.querySelfGamePools(ctx, app, startDay, endDay, gameID) + if err != nil { + return SelfGameOverview{}, err + } + overview.PoolStats = pools + participants, err := r.querySelfGameParticipantDistribution(ctx, app, startDay, endDay, gameID, query.CountryID, query.RegionID) + if err != nil { + return SelfGameOverview{}, err + } + overview.ParticipantDistribution = participants + risk, err := r.querySelfGameUserRisk(ctx, app, startDay, endDay, gameID, query.CountryID, query.RegionID) + if err != nil { + return SelfGameOverview{}, err + } + overview.UserRisk = risk + return overview, nil +} + func (r *Repository) queryAppDayOverviewTotals(ctx context.Context, app, startDay, endDay string, countryID int64, regionID int64) (Overview, error) { args := []any{app, startDay, endDay} filter := "" @@ -701,6 +890,232 @@ func sumLuckyGiftPools(pools []LuckyGiftPoolStat) (int64, int64) { return turnover, payout } +func (r *Repository) querySelfGameEvents(ctx context.Context, app, startDay, endDay, gameID string, countryID int64, regionID int64) ([]SelfGameEventStat, error) { + filter, args := selfGameFilter(app, startDay, endDay, gameID, countryID, regionID) + rows, err := r.db.QueryContext(ctx, ` + SELECT game_id, event_name, COALESCE(SUM(event_count),0), COALESCE(SUM(user_count),0), + COALESCE(SUM(success_count),0), COALESCE(SUM(error_count),0), COALESCE(SUM(duration_ms_sum),0) + FROM stat_self_game_h5_event_day + WHERE `+filter+` + GROUP BY game_id, event_name + ORDER BY game_id, event_name`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := []SelfGameEventStat{} + for rows.Next() { + var item SelfGameEventStat + if err := rows.Scan(&item.GameID, &item.EventName, &item.EventCount, &item.UserCount, &item.SuccessCount, &item.ErrorCount, &item.DurationMSSum); err != nil { + return nil, err + } + item.AvgDurationMS = div(item.DurationMSSum, item.EventCount) + out = append(out, item) + } + return out, rows.Err() +} + +func (r *Repository) querySelfGameMatchTotals(ctx context.Context, app, startDay, endDay, gameID string, countryID int64, regionID int64) (SelfGameMatchTotals, error) { + filter, args := selfGameFilter(app, startDay, endDay, gameID, countryID, regionID) + row := r.db.QueryRowContext(ctx, ` + SELECT COALESCE(SUM(created_matches),0), COALESCE(SUM(matched_matches),0), COALESCE(SUM(settled_matches),0), + COALESCE(SUM(canceled_matches),0), COALESCE(SUM(failed_matches),0), COALESCE(SUM(human_matches),0), + COALESCE(SUM(robot_matches),0), COALESCE(SUM(human_participants),0), COALESCE(SUM(robot_participants),0), + COALESCE(SUM(winner_count),0), COALESCE(SUM(loser_count),0), COALESCE(SUM(draw_count),0), + COALESCE(SUM(user_stake_coin),0), COALESCE(SUM(robot_stake_coin),0), COALESCE(SUM(payout_coin),0), + COALESCE(SUM(refund_coin),0), COALESCE(SUM(platform_profit_coin),0), COALESCE(SUM(pool_delta_coin),0), + COALESCE(SUM(forced_player_lose_matches),0), COALESCE(SUM(wait_ms_sum),0), COALESCE(SUM(duration_ms_sum),0) + FROM stat_self_game_match_day + WHERE `+filter, args...) + var totals SelfGameMatchTotals + if err := row.Scan(&totals.CreatedMatches, &totals.MatchedMatches, &totals.SettledMatches, &totals.CanceledMatches, &totals.FailedMatches, &totals.HumanMatches, &totals.RobotMatches, &totals.HumanParticipants, &totals.RobotParticipants, &totals.WinnerCount, &totals.LoserCount, &totals.DrawCount, &totals.UserStakeCoin, &totals.RobotStakeCoin, &totals.PayoutCoin, &totals.RefundCoin, &totals.PlatformProfitCoin, &totals.PoolDeltaCoin, &totals.ForcedPlayerLoseMatches, &totals.WaitMSSum, &totals.DurationMSSum); err != nil { + return SelfGameMatchTotals{}, err + } + applySelfGameMatchDerived(&totals) + return totals, nil +} + +func (r *Repository) querySelfGameStakeBreakdown(ctx context.Context, app, startDay, endDay, gameID string, countryID int64, regionID int64) ([]SelfGameStakeStat, error) { + filter, args := selfGameFilter(app, startDay, endDay, gameID, countryID, regionID) + rows, err := r.db.QueryContext(ctx, ` + SELECT game_id, stake_coin, COALESCE(SUM(created_matches),0), COALESCE(SUM(matched_matches),0), + COALESCE(SUM(settled_matches),0), COALESCE(SUM(canceled_matches),0), COALESCE(SUM(failed_matches),0), + COALESCE(SUM(human_participants),0), COALESCE(SUM(robot_participants),0), COALESCE(SUM(user_stake_coin),0), + COALESCE(SUM(robot_stake_coin),0), COALESCE(SUM(payout_coin),0), COALESCE(SUM(refund_coin),0), + COALESCE(SUM(platform_profit_coin),0), COALESCE(SUM(pool_delta_coin),0) + FROM stat_self_game_match_day + WHERE `+filter+` + GROUP BY game_id, stake_coin + ORDER BY game_id, stake_coin`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := []SelfGameStakeStat{} + for rows.Next() { + var item SelfGameStakeStat + if err := rows.Scan(&item.GameID, &item.StakeCoin, &item.CreatedMatches, &item.MatchedMatches, &item.SettledMatches, &item.CanceledMatches, &item.FailedMatches, &item.HumanParticipants, &item.RobotParticipants, &item.UserStakeCoin, &item.RobotStakeCoin, &item.PayoutCoin, &item.RefundCoin, &item.PlatformProfitCoin, &item.PoolDeltaCoin); err != nil { + return nil, err + } + item.PlatformProfitRate = ratio(item.PlatformProfitCoin, item.UserStakeCoin+item.RobotStakeCoin) + out = append(out, item) + } + return out, rows.Err() +} + +func (r *Repository) querySelfGamePools(ctx context.Context, app, startDay, endDay, gameID string) ([]SelfGamePoolStat, error) { + filter := "app_code = ? AND stat_day BETWEEN ? AND ?" + args := []any{app, startDay, endDay} + if gameID != "" { + filter += " AND game_id = ?" + args = append(args, gameID) + } + rows, err := r.db.QueryContext(ctx, ` + SELECT game_id, direction, reason, COALESCE(SUM(transaction_count),0), + COALESCE(SUM(in_coin),0), COALESCE(SUM(out_coin),0), COALESCE(MAX(balance_after_coin),0) + FROM stat_self_game_pool_day + WHERE `+filter+` + GROUP BY game_id, direction, reason + ORDER BY game_id, direction, reason`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := []SelfGamePoolStat{} + for rows.Next() { + var item SelfGamePoolStat + if err := rows.Scan(&item.GameID, &item.Direction, &item.Reason, &item.TransactionCount, &item.InCoin, &item.OutCoin, &item.BalanceAfterCoin); err != nil { + return nil, err + } + out = append(out, item) + } + return out, rows.Err() +} + +func (r *Repository) querySelfGameParticipantDistribution(ctx context.Context, app, startDay, endDay, gameID string, countryID int64, regionID int64) ([]SelfGameParticipantStat, error) { + filter, args := selfGameFilter(app, startDay, endDay, gameID, countryID, regionID) + rows, err := r.db.QueryContext(ctx, ` + SELECT game_id, stake_coin, participant_type, result, rps_gesture, dice_point, + COALESCE(SUM(participant_count),0), COALESCE(SUM(payout_coin),0), COALESCE(SUM(net_win_coin),0) + FROM stat_self_game_participant_day + WHERE `+filter+` + GROUP BY game_id, stake_coin, participant_type, result, rps_gesture, dice_point + ORDER BY game_id, stake_coin, participant_type, result, rps_gesture, dice_point`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + items := []SelfGameParticipantStat{} + totals := map[string]int64{} + for rows.Next() { + var item SelfGameParticipantStat + if err := rows.Scan(&item.GameID, &item.StakeCoin, &item.ParticipantType, &item.Result, &item.RPSGesture, &item.DicePoint, &item.ParticipantCount, &item.PayoutCoin, &item.NetWinCoin); err != nil { + return nil, err + } + key := item.GameID + "\x00" + item.ParticipantType + "\x00" + item.RPSGesture + if item.DicePoint > 0 { + key = item.GameID + "\x00" + item.ParticipantType + "\x00" + string(rune(item.DicePoint)) + } + totals[key] += item.ParticipantCount + items = append(items, item) + } + if err := rows.Err(); err != nil { + return nil, err + } + for index := range items { + key := items[index].GameID + "\x00" + items[index].ParticipantType + "\x00" + items[index].RPSGesture + if items[index].DicePoint > 0 { + key = items[index].GameID + "\x00" + items[index].ParticipantType + "\x00" + string(rune(items[index].DicePoint)) + } + if items[index].Result == "win" { + items[index].WinRate = ratio(items[index].ParticipantCount, totals[key]) + } + } + return items, nil +} + +func (r *Repository) querySelfGameUserRisk(ctx context.Context, app, startDay, endDay, gameID string, countryID int64, regionID int64) ([]SelfGameUserRiskStat, error) { + filter, args := selfGameFilter(app, startDay, endDay, gameID, countryID, regionID) + rows, err := r.db.QueryContext(ctx, ` + SELECT game_id, user_id, COALESCE(SUM(match_count),0), COALESCE(SUM(win_count),0), + COALESCE(SUM(lose_count),0), COALESCE(SUM(draw_count),0), COALESCE(SUM(cancel_count),0), + COALESCE(SUM(stake_coin),0), COALESCE(SUM(payout_coin),0), COALESCE(SUM(net_win_coin),0) + FROM stat_self_game_user_day + WHERE `+filter+` + GROUP BY game_id, user_id + ORDER BY SUM(match_count) DESC, SUM(net_win_coin) DESC, SUM(cancel_count) DESC + LIMIT 100`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := []SelfGameUserRiskStat{} + for rows.Next() { + var item SelfGameUserRiskStat + if err := rows.Scan(&item.GameID, &item.UserID, &item.MatchCount, &item.WinCount, &item.LoseCount, &item.DrawCount, &item.CancelCount, &item.StakeCoin, &item.PayoutCoin, &item.NetWinCoin); err != nil { + return nil, err + } + item.WinRate = ratio(item.WinCount, item.MatchCount) + item.CancelRate = ratio(item.CancelCount, item.MatchCount+item.CancelCount) + out = append(out, item) + } + return out, rows.Err() +} + +func selfGameFilter(app, startDay, endDay, gameID string, countryID int64, regionID int64) (string, []any) { + filter := "app_code = ? AND stat_day BETWEEN ? AND ?" + args := []any{app, startDay, endDay} + if gameID != "" { + filter += " AND game_id = ?" + args = append(args, gameID) + } + if countryID > 0 { + filter += " AND country_id = ?" + args = append(args, countryID) + } + if regionID > 0 { + filter += " AND region_id = ?" + args = append(args, regionID) + } + return filter, args +} + +func applySelfGameMatchDerived(totals *SelfGameMatchTotals) { + if totals == nil { + return + } + totals.AvgWaitMS = div(totals.WaitMSSum, totals.MatchedMatches+totals.SettledMatches+totals.CanceledMatches+totals.FailedMatches) + totals.AvgDurationMS = div(totals.DurationMSSum, totals.SettledMatches+totals.CanceledMatches+totals.FailedMatches) + totals.CancelRate = ratio(totals.CanceledMatches, totals.CreatedMatches) + totals.FailRate = ratio(totals.FailedMatches, totals.CreatedMatches) + totals.RobotMatchRate = ratio(totals.RobotMatches, totals.SettledMatches) + totals.PlatformProfitRate = ratio(totals.PlatformProfitCoin, totals.UserStakeCoin+totals.RobotStakeCoin) +} + +func buildSelfGameFunnel(events []SelfGameEventStat) ([]SelfGameFunnelStep, SelfGameConversion) { + order := []string{"page_open", "match_click", "match_success", "settlement_show", "play_again"} + byEvent := map[string]SelfGameFunnelStep{} + for _, event := range events { + step := byEvent[event.EventName] + step.EventName = event.EventName + step.EventCount += event.EventCount + step.UserCount += event.UserCount + byEvent[event.EventName] = step + } + funnel := make([]SelfGameFunnelStep, 0, len(order)) + for _, name := range order { + funnel = append(funnel, byEvent[name]) + funnel[len(funnel)-1].EventName = name + } + return funnel, SelfGameConversion{ + PageToMatchClickRate: ratio(byEvent["match_click"].UserCount, byEvent["page_open"].UserCount), + MatchClickToSuccessRate: ratio(byEvent["match_success"].UserCount, byEvent["match_click"].UserCount), + SuccessToSettlementRate: ratio(byEvent["settlement_show"].UserCount, byEvent["match_success"].UserCount), + SettlementToPlayAgainRate: ratio(byEvent["play_again"].UserCount, byEvent["settlement_show"].UserCount), + PageToSettlementRate: ratio(byEvent["settlement_show"].UserCount, byEvent["page_open"].UserCount), + } +} + func normalizeRange(startMS, endMS int64) (int64, int64) { if startMS <= 0 { now := time.Now().UTC() diff --git a/services/statistics-service/internal/storage/mysql/repository.go b/services/statistics-service/internal/storage/mysql/repository.go index 721b758b..8db79657 100644 --- a/services/statistics-service/internal/storage/mysql/repository.go +++ b/services/statistics-service/internal/storage/mysql/repository.go @@ -18,6 +18,7 @@ const ( SourceWallet = "wallet" SourceRoom = "room" SourceGame = "game" + SourceH5 = "h5" ) // Repository owns the statistics read model. Online queries must use aggregate @@ -171,6 +172,74 @@ func (r *Repository) Migrate(ctx context.Context) error { PRIMARY KEY (app_code, stat_day, country_id, platform_code, game_id, user_id), KEY idx_stat_game_player_region (app_code, stat_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, + `CREATE TABLE IF NOT EXISTS stat_self_game_h5_event_day ( + app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0, + region_id BIGINT NOT NULL DEFAULT 0, game_id VARCHAR(96) NOT NULL, event_name VARCHAR(64) NOT NULL, + language VARCHAR(16) NOT NULL DEFAULT '', client_version VARCHAR(64) NOT NULL DEFAULT '', + h5_version VARCHAR(64) NOT NULL DEFAULT '', entry_source VARCHAR(64) NOT NULL DEFAULT '', + event_count BIGINT NOT NULL DEFAULT 0, user_count BIGINT NOT NULL DEFAULT 0, + success_count BIGINT NOT NULL DEFAULT 0, error_count BIGINT NOT NULL DEFAULT 0, + duration_ms_sum BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, stat_day, country_id, region_id, game_id, event_name, language, client_version, h5_version, entry_source), + KEY idx_stat_self_game_h5_day (app_code, stat_day, game_id, event_name), + KEY idx_stat_self_game_h5_region (app_code, stat_day, region_id, game_id) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, + `CREATE TABLE IF NOT EXISTS stat_self_game_h5_event_users ( + app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, game_id VARCHAR(96) NOT NULL, + event_name VARCHAR(64) NOT NULL, user_id BIGINT NOT NULL, first_seen_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, stat_day, game_id, event_name, user_id), + KEY idx_stat_self_game_h5_users_day (app_code, stat_day, game_id, event_name) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, + `CREATE TABLE IF NOT EXISTS stat_self_game_match_day ( + app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0, + region_id BIGINT NOT NULL DEFAULT 0, game_id VARCHAR(96) NOT NULL, stake_coin BIGINT NOT NULL DEFAULT 0, + created_matches BIGINT NOT NULL DEFAULT 0, matched_matches BIGINT NOT NULL DEFAULT 0, + settled_matches BIGINT NOT NULL DEFAULT 0, canceled_matches BIGINT NOT NULL DEFAULT 0, + failed_matches BIGINT NOT NULL DEFAULT 0, human_matches BIGINT NOT NULL DEFAULT 0, + robot_matches BIGINT NOT NULL DEFAULT 0, human_participants BIGINT NOT NULL DEFAULT 0, + robot_participants BIGINT NOT NULL DEFAULT 0, winner_count BIGINT NOT NULL DEFAULT 0, + loser_count BIGINT NOT NULL DEFAULT 0, draw_count BIGINT NOT NULL DEFAULT 0, + user_stake_coin BIGINT NOT NULL DEFAULT 0, robot_stake_coin BIGINT NOT NULL DEFAULT 0, + payout_coin BIGINT NOT NULL DEFAULT 0, refund_coin BIGINT NOT NULL DEFAULT 0, + platform_profit_coin BIGINT NOT NULL DEFAULT 0, pool_delta_coin BIGINT NOT NULL DEFAULT 0, + forced_player_lose_matches BIGINT NOT NULL DEFAULT 0, wait_ms_sum BIGINT NOT NULL DEFAULT 0, + duration_ms_sum BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, stat_day, country_id, region_id, game_id, stake_coin), + KEY idx_stat_self_game_match_day (app_code, stat_day, game_id), + KEY idx_stat_self_game_match_region (app_code, stat_day, region_id, game_id) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, + `CREATE TABLE IF NOT EXISTS stat_self_game_pool_day ( + app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, game_id VARCHAR(96) NOT NULL, + direction VARCHAR(16) NOT NULL DEFAULT '', reason VARCHAR(64) NOT NULL DEFAULT '', + transaction_count BIGINT NOT NULL DEFAULT 0, in_coin BIGINT NOT NULL DEFAULT 0, + out_coin BIGINT NOT NULL DEFAULT 0, balance_after_coin BIGINT NOT NULL DEFAULT 0, + updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, stat_day, game_id, direction, reason), + KEY idx_stat_self_game_pool_day (app_code, stat_day, game_id) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, + `CREATE TABLE IF NOT EXISTS stat_self_game_participant_day ( + app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0, + region_id BIGINT NOT NULL DEFAULT 0, game_id VARCHAR(96) NOT NULL, stake_coin BIGINT NOT NULL DEFAULT 0, + participant_type VARCHAR(32) NOT NULL DEFAULT 'user', result VARCHAR(32) NOT NULL DEFAULT '', + rps_gesture VARCHAR(32) NOT NULL DEFAULT '', dice_point INT NOT NULL DEFAULT 0, + participant_count BIGINT NOT NULL DEFAULT 0, payout_coin BIGINT NOT NULL DEFAULT 0, + net_win_coin BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, stat_day, country_id, region_id, game_id, stake_coin, participant_type, result, rps_gesture, dice_point), + KEY idx_stat_self_game_participant_day (app_code, stat_day, game_id), + KEY idx_stat_self_game_participant_region (app_code, stat_day, region_id, game_id) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, + `CREATE TABLE IF NOT EXISTS stat_self_game_user_day ( + app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0, + region_id BIGINT NOT NULL DEFAULT 0, game_id VARCHAR(96) NOT NULL, user_id BIGINT NOT NULL, + match_count BIGINT NOT NULL DEFAULT 0, win_count BIGINT NOT NULL DEFAULT 0, + lose_count BIGINT NOT NULL DEFAULT 0, draw_count BIGINT NOT NULL DEFAULT 0, + cancel_count BIGINT NOT NULL DEFAULT 0, stake_coin BIGINT NOT NULL DEFAULT 0, + payout_coin BIGINT NOT NULL DEFAULT 0, net_win_coin BIGINT NOT NULL DEFAULT 0, + updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, stat_day, game_id, user_id), + KEY idx_stat_self_game_user_win (app_code, stat_day, game_id, net_win_coin), + KEY idx_stat_self_game_user_region (app_code, stat_day, region_id, game_id) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, } for _, statement := range statements { if _, err := r.db.ExecContext(ctx, statement); err != nil { @@ -340,6 +409,80 @@ type GameOrderEvent struct { OccurredAtMS int64 } +type SelfGameH5Event struct { + AppCode string + EventID string + EventName string + GameID string + Page string + SessionID string + MatchID string + UserID int64 + CountryID int64 + RegionID int64 + Language string + ClientVersion string + H5Version string + EntrySource string + DurationMS int64 + Success bool + ErrorCode string + OccurredAtMS int64 +} + +type SelfGameMatchParticipantEvent struct { + UserID int64 + ParticipantType string + StakeCoin int64 + DicePoints []int64 + RPSGesture string + Result string + PayoutCoin int64 + NetWinCoin int64 +} + +type SelfGameMatchEvent struct { + AppCode string + EventID string + EventType string + MatchID string + GameID string + PlatformCode string + RoomID string + RegionID int64 + StakeCoin int64 + Status string + Result string + MatchMode string + ForcedResult string + UserParticipants int64 + RobotParticipants int64 + WinnerCount int64 + LoserCount int64 + DrawCount int64 + UserStakeCoin int64 + RobotStakeCoin int64 + PayoutCoin int64 + RefundCoin int64 + PlatformProfitCoin int64 + PoolDeltaCoin int64 + WaitMS int64 + DurationMS int64 + Participants []SelfGameMatchParticipantEvent + OccurredAtMS int64 +} + +type SelfGamePoolAdjustmentEvent struct { + AppCode string + EventID string + GameID string + Direction string + Reason string + AmountCoin int64 + BalanceAfter int64 + OccurredAtMS int64 +} + func (r *Repository) ConsumeUserRegistered(ctx context.Context, event UserRegisteredEvent) error { return r.withEvent(ctx, SourceUser, event.EventID, "UserRegistered", func(tx *sql.Tx, nowMS int64) error { day := statDay(event.OccurredAtMS) @@ -558,6 +701,274 @@ func (r *Repository) ConsumeGameOrder(ctx context.Context, event GameOrderEvent) }) } +func (r *Repository) ConsumeSelfGameH5Event(ctx context.Context, event SelfGameH5Event) error { + eventName := normalizeShortText(event.EventName) + if eventName == "" { + eventName = "unknown" + } + return r.withEvent(ctx, SourceH5, event.EventID, eventName, func(tx *sql.Tx, nowMS int64) error { + day := statDay(event.OccurredAtMS) + countryID, regionID := normalizeDimension(event.CountryID, event.RegionID) + gameID := normalizeShortText(event.GameID) + if gameID == "" { + gameID = "unknown" + } + language := normalizeShortText(event.Language) + clientVersion := normalizeShortText(event.ClientVersion) + h5Version := normalizeShortText(event.H5Version) + entrySource := normalizeShortText(event.EntrySource) + if err := ensureSelfGameH5EventDay(ctx, tx, event.AppCode, day, countryID, regionID, gameID, eventName, language, clientVersion, h5Version, entrySource, nowMS); err != nil { + return err + } + userDelta := int64(0) + if event.UserID > 0 { + affected, err := insertUnique(ctx, tx, ` + INSERT IGNORE INTO stat_self_game_h5_event_users (app_code, stat_day, game_id, event_name, user_id, first_seen_at_ms) + VALUES (?, ?, ?, ?, ?, ?) + `, appcode.Normalize(event.AppCode), day, gameID, eventName, event.UserID, event.OccurredAtMS) + if err != nil { + return err + } + userDelta = affected + } + successDelta, errorDelta := int64(0), int64(0) + if event.Success { + successDelta = 1 + } + if normalizeShortText(event.ErrorCode) != "" { + errorDelta = 1 + } + // H5 事件按 event_id 做幂等；PV 用事件次数，UV 用同日同玩法同事件的用户去重，接口耗时只累加客户端实际传入的 duration。 + _, err := tx.ExecContext(ctx, ` + UPDATE stat_self_game_h5_event_day + SET event_count = event_count + 1, + user_count = user_count + ?, + success_count = success_count + ?, + error_count = error_count + ?, + duration_ms_sum = duration_ms_sum + ?, + updated_at_ms = ? + WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ? + AND game_id = ? AND event_name = ? AND language = ? AND client_version = ? + AND h5_version = ? AND entry_source = ? + `, userDelta, successDelta, errorDelta, normalizeID(event.DurationMS), nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID, gameID, eventName, language, clientVersion, h5Version, entrySource) + return err + }) +} + +func (r *Repository) ConsumeSelfGameMatch(ctx context.Context, event SelfGameMatchEvent) error { + return r.withEvent(ctx, SourceGame, event.EventID, event.EventType, func(tx *sql.Tx, nowMS int64) error { + day := statDay(event.OccurredAtMS) + countryID, regionID := normalizeDimension(0, event.RegionID) + gameID := normalizeShortText(event.GameID) + if gameID == "" { + gameID = "unknown" + } + stakeCoin := normalizeID(event.StakeCoin) + if err := ensureSelfGameMatchDay(ctx, tx, event.AppCode, day, countryID, regionID, gameID, stakeCoin, nowMS); err != nil { + return err + } + created, matched, settled, canceled, failed := selfGameMatchCounters(event.EventType) + humanMatch, robotMatch := int64(0), int64(0) + humanParticipants, robotParticipants := int64(0), int64(0) + userStake, robotStake, payout, refund, platformProfit, poolDelta := int64(0), int64(0), int64(0), int64(0), int64(0), int64(0) + winners, losers, draws := int64(0), int64(0), int64(0) + forcedLose, waitMS, durationMS := int64(0), int64(0), int64(0) + if settled > 0 { + if event.UserParticipants > 0 { + humanMatch = 1 + } + if event.RobotParticipants > 0 { + robotMatch = 1 + } + humanParticipants = normalizeID(event.UserParticipants) + robotParticipants = normalizeID(event.RobotParticipants) + winners = normalizeID(event.WinnerCount) + losers = normalizeID(event.LoserCount) + draws = normalizeID(event.DrawCount) + userStake = normalizeID(event.UserStakeCoin) + robotStake = normalizeID(event.RobotStakeCoin) + payout = normalizeID(event.PayoutCoin) + refund = normalizeID(event.RefundCoin) + platformProfit = event.PlatformProfitCoin + poolDelta = event.PoolDeltaCoin + waitMS = normalizeID(event.WaitMS) + durationMS = normalizeID(event.DurationMS) + if strings.EqualFold(strings.TrimSpace(event.ForcedResult), "player_lose") { + forcedLose = 1 + } + } + if canceled > 0 || failed > 0 || matched > 0 { + waitMS = normalizeID(event.WaitMS) + durationMS = normalizeID(event.DurationMS) + } + // 对局状态事件只按 match 级事实累加一次；明细用户维度只在结算/取消时落表，避免创建和 ready 阶段重复计算下注与输赢。 + if _, err := tx.ExecContext(ctx, ` + UPDATE stat_self_game_match_day + SET created_matches = created_matches + ?, + matched_matches = matched_matches + ?, + settled_matches = settled_matches + ?, + canceled_matches = canceled_matches + ?, + failed_matches = failed_matches + ?, + human_matches = human_matches + ?, + robot_matches = robot_matches + ?, + human_participants = human_participants + ?, + robot_participants = robot_participants + ?, + winner_count = winner_count + ?, + loser_count = loser_count + ?, + draw_count = draw_count + ?, + user_stake_coin = user_stake_coin + ?, + robot_stake_coin = robot_stake_coin + ?, + payout_coin = payout_coin + ?, + refund_coin = refund_coin + ?, + platform_profit_coin = platform_profit_coin + ?, + pool_delta_coin = pool_delta_coin + ?, + forced_player_lose_matches = forced_player_lose_matches + ?, + wait_ms_sum = wait_ms_sum + ?, + duration_ms_sum = duration_ms_sum + ?, + updated_at_ms = ? + WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ? + AND game_id = ? AND stake_coin = ? + `, created, matched, settled, canceled, failed, humanMatch, robotMatch, humanParticipants, robotParticipants, winners, losers, draws, userStake, robotStake, payout, refund, platformProfit, poolDelta, forcedLose, waitMS, durationMS, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID, gameID, stakeCoin); err != nil { + return err + } + if settled > 0 { + for _, participant := range event.Participants { + if err := consumeSelfGameSettledParticipant(ctx, tx, event, participant, day, countryID, regionID, gameID, stakeCoin, nowMS); err != nil { + return err + } + } + } + if canceled > 0 { + for _, participant := range event.Participants { + if participant.UserID <= 0 || !strings.EqualFold(strings.TrimSpace(participant.ParticipantType), "user") { + continue + } + if err := ensureSelfGameUserDay(ctx, tx, event.AppCode, day, countryID, regionID, gameID, participant.UserID, nowMS); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, ` + UPDATE stat_self_game_user_day + SET cancel_count = cancel_count + 1, updated_at_ms = ? + WHERE app_code = ? AND stat_day = ? AND game_id = ? AND user_id = ? + `, nowMS, appcode.Normalize(event.AppCode), day, gameID, participant.UserID); err != nil { + return err + } + } + } + return nil + }) +} + +func (r *Repository) ConsumeSelfGamePoolAdjustment(ctx context.Context, event SelfGamePoolAdjustmentEvent) error { + return r.withEvent(ctx, SourceGame, event.EventID, "SelfGamePoolAdjusted", func(tx *sql.Tx, nowMS int64) error { + day := statDay(event.OccurredAtMS) + gameID := normalizeShortText(event.GameID) + if gameID == "" { + gameID = "unknown" + } + direction := normalizeShortText(event.Direction) + reason := normalizeShortText(event.Reason) + if err := ensureSelfGamePoolDay(ctx, tx, event.AppCode, day, gameID, direction, reason, nowMS); err != nil { + return err + } + inCoin, outCoin := int64(0), int64(0) + switch direction { + case "in": + inCoin = normalizeID(event.AmountCoin) + case "out": + outCoin = normalizeID(event.AmountCoin) + } + _, err := tx.ExecContext(ctx, ` + UPDATE stat_self_game_pool_day + SET transaction_count = transaction_count + 1, + in_coin = in_coin + ?, + out_coin = out_coin + ?, + balance_after_coin = ?, + updated_at_ms = ? + WHERE app_code = ? AND stat_day = ? AND game_id = ? AND direction = ? AND reason = ? + `, inCoin, outCoin, event.BalanceAfter, nowMS, appcode.Normalize(event.AppCode), day, gameID, direction, reason) + return err + }) +} + +func consumeSelfGameSettledParticipant(ctx context.Context, tx *sql.Tx, event SelfGameMatchEvent, participant SelfGameMatchParticipantEvent, day string, countryID int64, regionID int64, gameID string, stakeCoin int64, nowMS int64) error { + participantType := normalizeShortText(participant.ParticipantType) + if participantType == "" { + participantType = "user" + } + result := normalizeShortText(participant.Result) + gesture := normalizeShortText(participant.RPSGesture) + points := participant.DicePoints + if len(points) == 0 { + points = []int64{0} + } + for _, point := range points { + dicePoint := int64(0) + if point > 0 { + dicePoint = point + } + if err := ensureSelfGameParticipantDay(ctx, tx, event.AppCode, day, countryID, regionID, gameID, stakeCoin, participantType, result, gesture, dicePoint, nowMS); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, ` + UPDATE stat_self_game_participant_day + SET participant_count = participant_count + 1, + payout_coin = payout_coin + ?, + net_win_coin = net_win_coin + ?, + updated_at_ms = ? + WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ? + AND game_id = ? AND stake_coin = ? AND participant_type = ? AND result = ? + AND rps_gesture = ? AND dice_point = ? + `, normalizeID(participant.PayoutCoin), participant.NetWinCoin, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID, gameID, stakeCoin, participantType, result, gesture, dicePoint); err != nil { + return err + } + } + if participant.UserID <= 0 || participantType != "user" { + return nil + } + if err := ensureSelfGameUserDay(ctx, tx, event.AppCode, day, countryID, regionID, gameID, participant.UserID, nowMS); err != nil { + return err + } + win, lose, draw := int64(0), int64(0), int64(0) + switch result { + case "win": + win = 1 + case "lose": + lose = 1 + case "draw": + draw = 1 + } + _, err := tx.ExecContext(ctx, ` + UPDATE stat_self_game_user_day + SET match_count = match_count + 1, + win_count = win_count + ?, + lose_count = lose_count + ?, + draw_count = draw_count + ?, + stake_coin = stake_coin + ?, + payout_coin = payout_coin + ?, + net_win_coin = net_win_coin + ?, + updated_at_ms = ? + WHERE app_code = ? AND stat_day = ? AND game_id = ? AND user_id = ? + `, win, lose, draw, normalizeID(participant.StakeCoin), normalizeID(participant.PayoutCoin), participant.NetWinCoin, nowMS, appcode.Normalize(event.AppCode), day, gameID, participant.UserID) + return err +} + +func selfGameMatchCounters(eventType string) (created, matched, settled, canceled, failed int64) { + switch strings.TrimSpace(eventType) { + case "SelfGameMatchCreated": + created = 1 + case "SelfGameMatchReady": + matched = 1 + case "SelfGameMatchSettled": + settled = 1 + case "SelfGameMatchCanceled": + canceled = 1 + case "SelfGameMatchFailed": + failed = 1 + } + return created, matched, settled, canceled, failed +} + func (r *Repository) withEvent(ctx context.Context, source string, eventID string, eventType string, apply func(tx *sql.Tx, nowMS int64) error) error { var lastErr error for attempt := 0; attempt < 5; attempt++ { @@ -685,6 +1096,56 @@ func ensureLuckyGiftPoolDay(ctx context.Context, tx *sql.Tx, app string, day str return err } +func ensureSelfGameH5EventDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, gameID string, eventName string, language string, clientVersion string, h5Version string, entrySource string, nowMS int64) error { + _, err := tx.ExecContext(ctx, ` + INSERT INTO stat_self_game_h5_event_day (app_code, stat_day, country_id, region_id, game_id, event_name, language, client_version, h5_version, entry_source, updated_at_ms) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE + updated_at_ms = VALUES(updated_at_ms) + `, appcode.Normalize(app), day, countryID, regionID, gameID, eventName, language, clientVersion, h5Version, entrySource, nowMS) + return err +} + +func ensureSelfGameMatchDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, gameID string, stakeCoin int64, nowMS int64) error { + _, err := tx.ExecContext(ctx, ` + INSERT INTO stat_self_game_match_day (app_code, stat_day, country_id, region_id, game_id, stake_coin, updated_at_ms) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE + updated_at_ms = VALUES(updated_at_ms) + `, appcode.Normalize(app), day, countryID, regionID, gameID, stakeCoin, nowMS) + return err +} + +func ensureSelfGamePoolDay(ctx context.Context, tx *sql.Tx, app string, day string, gameID string, direction string, reason string, nowMS int64) error { + _, err := tx.ExecContext(ctx, ` + INSERT INTO stat_self_game_pool_day (app_code, stat_day, game_id, direction, reason, updated_at_ms) + VALUES (?, ?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE + updated_at_ms = VALUES(updated_at_ms) + `, appcode.Normalize(app), day, gameID, direction, reason, nowMS) + return err +} + +func ensureSelfGameParticipantDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, gameID string, stakeCoin int64, participantType string, result string, gesture string, dicePoint int64, nowMS int64) error { + _, err := tx.ExecContext(ctx, ` + INSERT INTO stat_self_game_participant_day (app_code, stat_day, country_id, region_id, game_id, stake_coin, participant_type, result, rps_gesture, dice_point, updated_at_ms) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE + updated_at_ms = VALUES(updated_at_ms) + `, appcode.Normalize(app), day, countryID, regionID, gameID, stakeCoin, participantType, result, gesture, dicePoint, nowMS) + return err +} + +func ensureSelfGameUserDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, gameID string, userID int64, nowMS int64) error { + _, err := tx.ExecContext(ctx, ` + INSERT INTO stat_self_game_user_day (app_code, stat_day, country_id, region_id, game_id, user_id, updated_at_ms) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE + updated_at_ms = VALUES(updated_at_ms) + `, appcode.Normalize(app), day, countryID, regionID, gameID, userID, nowMS) + return err +} + func insertUnique(ctx context.Context, tx *sql.Tx, query string, args ...any) (int64, error) { result, err := tx.ExecContext(ctx, query, args...) if err != nil { @@ -708,6 +1169,10 @@ func normalizeID(id int64) int64 { return id } +func normalizeShortText(value string) string { + return strings.TrimSpace(value) +} + func normalizeDimension(countryID int64, regionID int64) (int64, int64) { return normalizeID(countryID), normalizeID(regionID) }