hy/hyapp-server
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

首充修复

Browse Source
This commit is contained in:
zhx 2026-06-11 19:24:51 +08:00
parent eea564ac56
commit cc84e14860
22 changed files with 1897 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
pkg/gamemq/messages.go
View File

@ -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 = "{}"
}

51
services/game-service/internal/storage/mysql/dice_matchmaking_repository.go
View File

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

179
services/game-service/internal/storage/mysql/dice_repository.go
View File

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

3
services/gateway-service/configs/config.docker.yaml
View File

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

3
services/gateway-service/configs/config.tencent.example.yaml
View File

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

3
services/gateway-service/configs/config.yaml
View File

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

2
services/gateway-service/internal/app/app.go
View File

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

81
services/gateway-service/internal/client/statistics_client.go Normal file
View File

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

18
services/gateway-service/internal/config/config.go
View File

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

8
services/gateway-service/internal/transport/http/activityapi/first_recharge_reward_handler.go
View File

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

33
services/gateway-service/internal/transport/http/activityapi/first_recharge_reward_handler_test.go Normal file
View File

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

4
services/gateway-service/internal/transport/http/gameapi/handler.go
View File

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

95
services/gateway-service/internal/transport/http/gameapi/self_game_report_handler.go Normal file
View File

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

6
services/gateway-service/internal/transport/http/handler.go
View File

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

2
services/gateway-service/internal/transport/http/httproutes/router.go
View File

@ -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，避免路由覆盖。

1
services/gateway-service/internal/transport/http/router.go
View File

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

123
services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql
View File

@ -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='自研游戏用户风险日统计表';

131
services/statistics-service/internal/app/app.go
View File

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

161
services/statistics-service/internal/app/local_flow_test.go
View File

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

102
services/statistics-service/internal/app/query_http.go
View File

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

415
services/statistics-service/internal/storage/mysql/query.go
View File

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

465
services/statistics-service/internal/storage/mysql/repository.go
View File

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