From 1b6c77c6dca1ea2fb2a1391ba992a21265c3d049 Mon Sep 17 00:00:00 2001 From: zhx Date: Mon, 8 Jun 2026 00:55:30 +0800 Subject: [PATCH] Add regional statistics breakdown --- .../internal/modules/dashboard/handler.go | 11 +- .../internal/modules/dashboard/service.go | 20 +- .../mysql/initdb/001_statistics_service.sql | 36 +- .../statistics-service/internal/app/app.go | 65 ++- .../internal/app/local_flow_test.go | 53 ++- .../internal/app/query_http.go | 11 +- .../internal/storage/mysql/query.go | 401 +++++++++++++++++- .../internal/storage/mysql/query_test.go | 40 ++ .../internal/storage/mysql/repository.go | 207 ++++++--- .../internal/storage/mysql/user/repository.go | 1 + 10 files changed, 723 insertions(+), 122 deletions(-) diff --git a/server/admin/internal/modules/dashboard/handler.go b/server/admin/internal/modules/dashboard/handler.go index def86ae8..f4ca4b40 100644 --- a/server/admin/internal/modules/dashboard/handler.go +++ b/server/admin/internal/modules/dashboard/handler.go @@ -28,10 +28,13 @@ func (h *Handler) DashboardOverview(c *gin.Context) { func (h *Handler) StatisticsOverview(c *gin.Context) { overview, err := h.service.StatisticsOverview(c.Request.Context(), StatisticsQuery{ - AppCode: c.DefaultQuery("app_code", "lalu"), - StartMS: parseInt64(c.Query("start_ms")), - EndMS: parseInt64(c.Query("end_ms")), - CountryID: parseInt64(c.Query("country_id")), + AppCode: c.DefaultQuery("app_code", "lalu"), + StartMS: parseInt64(c.Query("start_ms")), + EndMS: parseInt64(c.Query("end_ms")), + SeriesStartMS: parseInt64(c.Query("series_start_ms")), + SeriesEndMS: parseInt64(c.Query("series_end_ms")), + RegionID: parseInt64(c.Query("region_id")), + CountryID: parseInt64(c.Query("country_id")), }) if err != nil { response.ServerError(c, "获取统计总览失败") diff --git a/server/admin/internal/modules/dashboard/service.go b/server/admin/internal/modules/dashboard/service.go index 6f70dbe5..704118c6 100644 --- a/server/admin/internal/modules/dashboard/service.go +++ b/server/admin/internal/modules/dashboard/service.go @@ -19,10 +19,13 @@ type DashboardService struct { } type StatisticsQuery struct { - AppCode string - StartMS int64 - EndMS int64 - CountryID int64 + AppCode string + StartMS int64 + EndMS int64 + SeriesStartMS int64 + SeriesEndMS int64 + RegionID int64 + CountryID int64 } func NewService(store *repository.Store, cfg config.Config) *DashboardService { @@ -47,6 +50,15 @@ func (s *DashboardService) StatisticsOverview(ctx context.Context, query Statist if query.EndMS > 0 { values.Set("end_ms", strconv.FormatInt(query.EndMS, 10)) } + if query.SeriesStartMS > 0 { + values.Set("series_start_ms", strconv.FormatInt(query.SeriesStartMS, 10)) + } + if query.SeriesEndMS > 0 { + values.Set("series_end_ms", strconv.FormatInt(query.SeriesEndMS, 10)) + } + if query.RegionID > 0 { + values.Set("region_id", strconv.FormatInt(query.RegionID, 10)) + } if query.CountryID > 0 { values.Set("country_id", strconv.FormatInt(query.CountryID, 10)) } 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 f7375e78..f8a1ac2c 100644 --- a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql +++ b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql @@ -20,7 +20,8 @@ CREATE TABLE IF NOT EXISTS statistics_event_consumption ( CREATE TABLE IF NOT EXISTS stat_app_day_country ( app_code VARCHAR(32) NOT NULL COMMENT '应用编码', stat_day DATE NOT NULL COMMENT 'UTC 统计日', - country_id BIGINT NOT NULL DEFAULT 0 COMMENT '国家或区域维度 ID，0 表示未知或全局', + country_id BIGINT NOT NULL DEFAULT 0 COMMENT '国家维度 ID，0 表示未知或全局', + region_id BIGINT NOT NULL DEFAULT 0 COMMENT '区域维度 ID，0 表示未知或全局', new_users BIGINT NOT NULL DEFAULT 0 COMMENT '新增用户数', active_users BIGINT NOT NULL DEFAULT 0 COMMENT '当日活跃去重用户数', paid_users BIGINT NOT NULL DEFAULT 0 COMMENT '当日充值去重用户数', @@ -39,17 +40,20 @@ CREATE TABLE IF NOT EXISTS stat_app_day_country ( game_payout BIGINT NOT NULL DEFAULT 0 COMMENT '游戏返奖', game_refund BIGINT NOT NULL DEFAULT 0 COMMENT '游戏退款', updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms', - PRIMARY KEY (app_code, stat_day, country_id) + PRIMARY KEY (app_code, stat_day, country_id), + KEY idx_stat_app_day_region (app_code, stat_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='App 国家日统计聚合表'; CREATE TABLE IF NOT EXISTS stat_user_day_activity ( 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, user_id BIGINT NOT NULL, first_active_at_ms BIGINT NOT NULL, PRIMARY KEY (app_code, stat_day, user_id), - KEY idx_stat_user_day_country (app_code, stat_day, country_id) + KEY idx_stat_user_day_country (app_code, stat_day, country_id), + KEY idx_stat_user_day_region (app_code, stat_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='用户日活跃去重表'; CREATE TABLE IF NOT EXISTS stat_user_registration ( @@ -57,47 +61,56 @@ CREATE TABLE IF NOT EXISTS stat_user_registration ( user_id BIGINT NOT NULL, registered_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0, + region_id BIGINT NOT NULL DEFAULT 0, registered_at_ms BIGINT NOT NULL, PRIMARY KEY (app_code, user_id), - KEY idx_stat_user_registration_day (app_code, registered_day, country_id) + KEY idx_stat_user_registration_day (app_code, registered_day, country_id), + KEY idx_stat_user_registration_region (app_code, registered_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='用户注册日留存 cohort 表'; CREATE TABLE IF NOT EXISTS stat_recharge_day_payers ( 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, user_id BIGINT NOT NULL, first_paid_at_ms BIGINT NOT NULL, PRIMARY KEY (app_code, stat_day, user_id), - KEY idx_stat_recharge_payer_country (app_code, stat_day, country_id) + KEY idx_stat_recharge_payer_country (app_code, stat_day, country_id), + KEY idx_stat_recharge_payer_region (app_code, stat_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='充值日付费用户去重表'; CREATE TABLE IF NOT EXISTS stat_lucky_gift_day_payers ( 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, user_id BIGINT NOT NULL, first_paid_at_ms BIGINT NOT NULL, PRIMARY KEY (app_code, stat_day, user_id), - KEY idx_stat_lucky_payer_country (app_code, stat_day, country_id) + KEY idx_stat_lucky_payer_country (app_code, stat_day, country_id), + KEY idx_stat_lucky_payer_region (app_code, stat_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物日付费去重表'; CREATE TABLE IF NOT EXISTS stat_lucky_gift_pool_day_country ( app_code VARCHAR(32) NOT NULL COMMENT '应用编码', stat_day DATE NOT NULL COMMENT 'UTC 统计日', - country_id BIGINT NOT NULL DEFAULT 0 COMMENT '国家或区域维度 ID，0 表示未知或全局', + country_id BIGINT NOT NULL DEFAULT 0 COMMENT '国家维度 ID，0 表示未知或全局', + region_id BIGINT NOT NULL DEFAULT 0 COMMENT '区域维度 ID，0 表示未知或全局', pool_id VARCHAR(96) NOT NULL COMMENT '幸运礼物奖池 ID', turnover_coin BIGINT NOT NULL DEFAULT 0 COMMENT '奖池幸运礼物流水金币', payout_coin BIGINT NOT NULL DEFAULT 0 COMMENT '奖池幸运礼物返奖金币', updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms', PRIMARY KEY (app_code, stat_day, country_id, pool_id), - KEY idx_stat_lucky_pool_overview (app_code, stat_day, pool_id) + KEY idx_stat_lucky_pool_overview (app_code, stat_day, pool_id), + KEY idx_stat_lucky_pool_region (app_code, stat_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物奖池国家日统计聚合表'; CREATE TABLE IF NOT EXISTS stat_game_day_country ( 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, platform_code VARCHAR(64) NOT NULL DEFAULT '', game_id VARCHAR(96) NOT NULL, turnover_coin BIGINT NOT NULL DEFAULT 0, @@ -105,16 +118,19 @@ CREATE TABLE IF NOT EXISTS stat_game_day_country ( refund_coin BIGINT NOT NULL DEFAULT 0, player_count BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL, - PRIMARY KEY (app_code, stat_day, country_id, platform_code, game_id) + PRIMARY KEY (app_code, stat_day, country_id, platform_code, game_id), + KEY idx_stat_game_day_region (app_code, stat_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='游戏国家日统计聚合表'; CREATE TABLE IF NOT EXISTS stat_game_day_players ( 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, platform_code VARCHAR(64) NOT NULL DEFAULT '', game_id VARCHAR(96) NOT NULL, user_id BIGINT NOT NULL, first_played_at_ms BIGINT NOT NULL, - PRIMARY KEY (app_code, stat_day, country_id, platform_code, game_id, user_id) + 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='游戏日参与用户去重表'; diff --git a/services/statistics-service/internal/app/app.go b/services/statistics-service/internal/app/app.go index 646c4706..336c5718 100644 --- a/services/statistics-service/internal/app/app.go +++ b/services/statistics-service/internal/app/app.go @@ -152,6 +152,13 @@ func newConsumers(cfg config.Config, repo *mysqlstorage.Repository) ([]*rocketmq if ok { return repo.ConsumeRecharge(appcode.WithContext(ctx, event.AppCode), event) } + stock, ok, err := coinSellerStockEvent(message.Body) + if err != nil { + return err + } + if ok { + return repo.ConsumeCoinSellerStock(appcode.WithContext(ctx, stock.AppCode), stock) + } reward, ok, err := luckyGiftRewardEvent(message.Body) if err != nil || !ok { return err @@ -174,7 +181,8 @@ func newConsumers(cfg config.Config, repo *mysqlstorage.Repository) ([]*rocketmq EventID: gift.EventID + ":active", EventType: "RoomGiftSentActive", UserID: gift.SenderUserID, - CountryID: gift.VisibleRegionID, + CountryID: gift.CountryID, + RegionID: gift.VisibleRegionID, OccurredAtMS: gift.OccurredAtMS, }) } @@ -213,6 +221,7 @@ func userRegisteredEvent(body []byte) (mysqlstorage.UserRegisteredEvent, bool, e AppCode: message.AppCode, EventID: message.EventID, UserID: message.AggregateID, + CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "country_id"), mysqlstorage.Int64(payload, "target_country_id")), RegionID: mysqlstorage.Int64(payload, "region_id"), OccurredAtMS: message.OccurredAtMS, }, true, nil @@ -243,12 +252,41 @@ func rechargeEvent(body []byte) (mysqlstorage.RechargeEvent, bool, error) { USDMinor: usdMinor, CoinAmount: coinAmount, RechargeSequence: mysqlstorage.Int64(payload, "recharge_sequence"), + TargetCountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "target_country_id"), mysqlstorage.Int64(payload, "country_id")), TargetRegionID: regionID, RechargeType: rechargeType, OccurredAtMS: message.OccurredAtMS, }, true, nil } +func coinSellerStockEvent(body []byte) (mysqlstorage.CoinSellerStockEvent, bool, error) { + message, err := walletmq.DecodeWalletOutboxMessage(body) + if err != nil { + return mysqlstorage.CoinSellerStockEvent{}, false, err + } + if message.EventType != "WalletCoinSellerStockPurchased" { + return mysqlstorage.CoinSellerStockEvent{}, false, nil + } + payload := mysqlstorage.DecodeJSON(message.PayloadJSON) + if !boolValue(payload, "counts_as_seller_recharge") { + return mysqlstorage.CoinSellerStockEvent{}, false, nil + } + usdMinor := amountMicroToUSDMinor(mysqlstorage.Int64(payload, "paid_amount_micro")) + if usdMinor <= 0 { + return mysqlstorage.CoinSellerStockEvent{}, false, nil + } + return mysqlstorage.CoinSellerStockEvent{ + AppCode: message.AppCode, + EventID: message.EventID, + SellerUserID: firstNonZeroInt64(mysqlstorage.Int64(payload, "seller_user_id"), message.UserID), + USDMinor: usdMinor, + CoinAmount: firstNonZeroInt64(mysqlstorage.Int64(payload, "coin_amount"), mysqlstorage.Int64(payload, "available_delta")), + CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "country_id"), mysqlstorage.Int64(payload, "seller_country_id")), + RegionID: firstNonZeroInt64(mysqlstorage.Int64(payload, "region_id"), mysqlstorage.Int64(payload, "seller_region_id")), + OccurredAtMS: message.OccurredAtMS, + }, true, nil +} + func rechargeUSDMinorFromPayload(payload map[string]any, rechargeType string) int64 { // 币商向用户转账是普通金币到账事实，只用于充值用户和金币金额统计；即使历史 payload 携带 recharge_usd_minor，也不能进入用户 USDT/USD 充值。 if isCoinSellerRechargeType(rechargeType) { @@ -303,7 +341,8 @@ func luckyGiftRewardEvent(body []byte) (mysqlstorage.LuckyGiftRewardEvent, bool, EventID: message.EventID, EventType: message.EventType, UserID: firstNonZeroInt64(mysqlstorage.Int64(payload, "user_id"), message.UserID), - CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "visible_region_id"), mysqlstorage.Int64(payload, "region_id"), mysqlstorage.Int64(payload, "target_region_id")), + CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "country_id"), mysqlstorage.Int64(payload, "target_country_id")), + RegionID: firstNonZeroInt64(mysqlstorage.Int64(payload, "visible_region_id"), mysqlstorage.Int64(payload, "region_id"), mysqlstorage.Int64(payload, "target_region_id")), PoolID: mysqlstorage.String(payload, "pool_id"), Amount: mysqlstorage.Int64(payload, "amount"), OccurredAtMS: message.OccurredAtMS, @@ -326,6 +365,7 @@ func roomGiftEvent(body []byte) (mysqlstorage.RoomGiftEvent, bool, error) { AppCode: envelope.GetAppCode(), EventID: envelope.GetEventId(), SenderUserID: gift.GetSenderUserId(), + CountryID: 0, VisibleRegionID: gift.GetVisibleRegionId(), GiftValue: gift.GetGiftValue(), CoinSpent: roomGiftCoinSpent(&gift), @@ -362,7 +402,8 @@ func roomJoinActiveEvent(body []byte) (mysqlstorage.UserActiveEvent, bool, error EventID: envelope.GetEventId(), EventType: envelope.GetEventType(), UserID: joined.GetUserId(), - CountryID: joined.GetVisibleRegionId(), + CountryID: 0, + RegionID: joined.GetVisibleRegionId(), OccurredAtMS: envelope.GetOccurredAtMs(), }, true, nil } @@ -380,7 +421,8 @@ func gameOrderEvent(body []byte) (mysqlstorage.GameOrderEvent, bool, error) { AppCode: message.AppCode, EventID: message.EventID, UserID: message.UserID, - CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "visible_region_id"), mysqlstorage.Int64(payload, "region_id")), + CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "country_id"), mysqlstorage.Int64(payload, "target_country_id")), + RegionID: firstNonZeroInt64(mysqlstorage.Int64(payload, "visible_region_id"), mysqlstorage.Int64(payload, "region_id")), PlatformCode: message.PlatformCode, GameID: message.GameID, OpType: message.OpType, @@ -443,3 +485,18 @@ func firstNonZeroInt64(values ...int64) int64 { } return 0 } + +func boolValue(payload map[string]any, key string) bool { + switch value := payload[key].(type) { + case bool: + return value + case string: + return strings.EqualFold(strings.TrimSpace(value), "true") || strings.TrimSpace(value) == "1" + case float64: + return value != 0 + case int64: + return value != 0 + default: + return false + } +} diff --git a/services/statistics-service/internal/app/local_flow_test.go b/services/statistics-service/internal/app/local_flow_test.go index 6d9ff8fa..f981acb6 100644 --- a/services/statistics-service/internal/app/local_flow_test.go +++ b/services/statistics-service/internal/app/local_flow_test.go @@ -32,9 +32,10 @@ func TestLocalStatisticsDataScreenFlow(t *testing.T) { occurredAt := time.Date(2026, 5, 31, 12, 0, 0, 0, time.UTC).UnixMilli() dayStart := time.Date(2026, 5, 31, 0, 0, 0, 0, time.UTC).UnixMilli() const ( - appCode = "lalu" - regionID = int64(210) - userID = int64(42) + appCode = "lalu" + countryID = int64(86) + regionID = int64(210) + userID = int64(42) ) googleRecharge := walletMessage(t, walletmq.WalletOutboxMessage{ @@ -46,6 +47,7 @@ func TestLocalStatisticsDataScreenFlow(t *testing.T) { UserID: userID, PayloadJSON: mustJSON(t, map[string]any{ "amount_micro": int64(1_500_000), + "country_id": countryID, "region_id": regionID, "channel": "google", "recharge_sequence": int64(1), @@ -60,6 +62,32 @@ func TestLocalStatisticsDataScreenFlow(t *testing.T) { t.Fatalf("consume google recharge failed: %v", err) } + coinSellerStock := walletMessage(t, walletmq.WalletOutboxMessage{ + AppCode: appCode, + EventID: "WalletCoinSellerStockPurchased:1", + EventType: "WalletCoinSellerStockPurchased", + TransactionID: "wallet_tx_coin_seller_stock_1", + CommandID: "coin_seller_stock_1", + UserID: userID + 10, + PayloadJSON: mustJSON(t, map[string]any{ + "seller_user_id": userID + 10, + "country_id": countryID, + "region_id": regionID, + "coin_amount": int64(400_000), + "paid_amount_micro": int64(4_000_000), + "counts_as_seller_recharge": true, + "stock_type": "usdt_purchase", + }), + OccurredAtMS: occurredAt, + }) + stock, ok, err := coinSellerStockEvent(coinSellerStock) + if err != nil || !ok { + t.Fatalf("decode coin seller stock failed: ok=%v err=%v", ok, err) + } + if err := repo.ConsumeCoinSellerStock(ctx, stock); err != nil { + t.Fatalf("consume coin seller stock failed: %v", err) + } + coinSellerRecharge := walletMessage(t, walletmq.WalletOutboxMessage{ AppCode: appCode, EventID: "WalletRechargeRecorded:coin_seller:1", @@ -70,6 +98,7 @@ func TestLocalStatisticsDataScreenFlow(t *testing.T) { PayloadJSON: mustJSON(t, map[string]any{ "amount": int64(160_000), "recharge_type": "coin_seller_transfer", + "target_country_id": countryID, "target_region_id": regionID, "recharge_sequence": int64(1), }), @@ -115,6 +144,7 @@ func TestLocalStatisticsDataScreenFlow(t *testing.T) { "room_id": "room_1", "pool_id": "lucky_100", "amount": int64(300), + "country_id": countryID, "visible_region_id": regionID, }), OccurredAtMS: occurredAt, @@ -128,8 +158,8 @@ func TestLocalStatisticsDataScreenFlow(t *testing.T) { } for _, message := range []gamemq.GameOutboxMessage{ - gameMessage(appCode, "game_order_debit_1", userID, "debit", 700, regionID, occurredAt), - gameMessage(appCode, "game_order_credit_1", userID, "credit", 200, regionID, occurredAt), + gameMessage(appCode, "game_order_debit_1", userID, "debit", 700, countryID, regionID, occurredAt), + gameMessage(appCode, "game_order_credit_1", userID, "credit", 200, countryID, regionID, occurredAt), } { body, err := gamemq.EncodeGameOutboxMessage(message) if err != nil { @@ -145,15 +175,15 @@ func TestLocalStatisticsDataScreenFlow(t *testing.T) { } overview, err := repo.QueryOverview(ctx, mysqlstorage.OverviewQuery{ - AppCode: appCode, - StartMS: dayStart, - EndMS: dayStart + int64(24*time.Hour/time.Millisecond), - CountryID: regionID, + AppCode: appCode, + StartMS: dayStart, + EndMS: dayStart + int64(24*time.Hour/time.Millisecond), + RegionID: regionID, }) if err != nil { t.Fatalf("query overview failed: %v", err) } - if overview.GoogleRechargeUSDMinor != 150 || overview.CoinSellerRechargeUSDMinor != 0 || overview.CoinSellerTransferCoin != 160_000 || overview.RechargeUSDMinor != 150 || overview.NewUserRechargeUSDMinor != 150 || overview.PaidUsers != 2 { + if overview.GoogleRechargeUSDMinor != 150 || overview.CoinSellerRechargeUSDMinor != 400 || overview.CoinSellerTransferCoin != 160_000 || overview.RechargeUSDMinor != 550 || overview.NewUserRechargeUSDMinor != 150 || overview.PaidUsers != 2 { t.Fatalf("google recharge metrics mismatch: %+v", overview) } if overview.LuckyGiftTurnover != 1_250 || overview.LuckyGiftPayout != 300 || overview.LuckyGiftProfit != 950 || overview.LuckyGiftPayers != 1 { @@ -201,7 +231,7 @@ func roomGiftMessage(t *testing.T, appCode string, eventID string, roomID string return body } -func gameMessage(appCode string, orderID string, userID int64, opType string, amount int64, regionID int64, occurredAt int64) gamemq.GameOutboxMessage { +func gameMessage(appCode string, orderID string, userID int64, opType string, amount int64, countryID int64, regionID int64, occurredAt int64) gamemq.GameOutboxMessage { return gamemq.GameOutboxMessage{ AppCode: appCode, EventID: "GameOrderSettled:" + orderID, @@ -219,6 +249,7 @@ func gameMessage(appCode string, orderID string, userID int64, opType string, am "user_id": userID, "room_id": "room_1", "visible_region_id": regionID, + "country_id": countryID, "region_id": regionID, "op_type": opType, "coin_amount": amount, diff --git a/services/statistics-service/internal/app/query_http.go b/services/statistics-service/internal/app/query_http.go index f1e505be..81a5897b 100644 --- a/services/statistics-service/internal/app/query_http.go +++ b/services/statistics-service/internal/app/query_http.go @@ -65,10 +65,13 @@ func (s *queryHTTPServer) overview(w http.ResponseWriter, r *http.Request) { endMS = startMS + 24*time.Hour.Milliseconds() } overview, err := s.repo.QueryOverview(r.Context(), mysqlstorage.OverviewQuery{ - AppCode: query.Get("app_code"), - StartMS: startMS, - EndMS: endMS, - CountryID: parseInt64(query.Get("country_id")), + AppCode: query.Get("app_code"), + StartMS: startMS, + EndMS: endMS, + SeriesStartMS: parseInt64(query.Get("series_start_ms")), + SeriesEndMS: parseInt64(query.Get("series_end_ms")), + 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()}) diff --git a/services/statistics-service/internal/storage/mysql/query.go b/services/statistics-service/internal/storage/mysql/query.go index 4c4fb228..09b5259d 100644 --- a/services/statistics-service/internal/storage/mysql/query.go +++ b/services/statistics-service/internal/storage/mysql/query.go @@ -4,16 +4,20 @@ import ( "context" "database/sql" "math" + "sort" "time" "hyapp/pkg/appcode" ) type OverviewQuery struct { - AppCode string - StartMS int64 - EndMS int64 - CountryID int64 + AppCode string + StartMS int64 + EndMS int64 + SeriesStartMS int64 + SeriesEndMS int64 + CountryID int64 + RegionID int64 } type Overview struct { @@ -24,6 +28,7 @@ type Overview struct { NewUsers int64 `json:"new_users"` ActiveUsers int64 `json:"active_users"` PaidUsers int64 `json:"paid_users"` + NewPaidUsers int64 `json:"new_paid_users"` RechargeUSDMinor int64 `json:"recharge_usd_minor"` NewUserRechargeUSDMinor int64 `json:"new_user_recharge_usd_minor"` CoinSellerRechargeUSDMinor int64 `json:"coin_seller_recharge_usd_minor"` @@ -50,6 +55,7 @@ type Overview struct { NewUserRechargeDeltaRate float64 `json:"new_user_recharge_delta_rate"` ActiveUsersDeltaRate float64 `json:"active_users_delta_rate"` PaidUsersDeltaRate float64 `json:"paid_users_delta_rate"` + NewPaidUsersDeltaRate float64 `json:"new_paid_users_delta_rate"` ARPUDeltaRate float64 `json:"arpu_delta_rate"` ARPPUDeltaRate float64 `json:"arppu_delta_rate"` GiftCoinSpentDeltaRate float64 `json:"gift_coin_spent_delta_rate"` @@ -60,8 +66,34 @@ type Overview struct { GameTurnoverDeltaRate float64 `json:"game_turnover_delta_rate"` GameProfitDeltaRate float64 `json:"game_profit_delta_rate"` Retention Retention `json:"retention"` + DailySeries []DailySeriesItem `json:"daily_series"` GameRanking []GameRank `json:"game_ranking"` LuckyGiftPools []LuckyGiftPoolStat `json:"lucky_gift_pools"` + CountryBreakdown []CountryBreakdown `json:"country_breakdown"` +} + +type DailySeriesItem struct { + StatDay string `json:"stat_day"` + Label string `json:"label"` + NewUsers int64 `json:"new_users"` + ActiveUsers int64 `json:"active_users"` + PaidUsers int64 `json:"paid_users"` + NewPaidUsers int64 `json:"new_paid_users"` + RechargeUsers int64 `json:"recharge_users"` + RechargeUSDMinor int64 `json:"recharge_usd_minor"` + NewUserRechargeUSDMinor int64 `json:"new_user_recharge_usd_minor"` + CoinSellerRechargeUSDMinor int64 `json:"coin_seller_recharge_usd_minor"` + MifaPayRechargeUSDMinor int64 `json:"mifapay_recharge_usd_minor"` + GoogleRechargeUSDMinor int64 `json:"google_recharge_usd_minor"` + CoinSellerTransferCoin int64 `json:"coin_seller_transfer_coin"` + GiftCoinSpent int64 `json:"gift_coin_spent"` + LuckyGiftTurnover int64 `json:"lucky_gift_turnover"` + LuckyGiftPayout int64 `json:"lucky_gift_payout"` + LuckyGiftProfit int64 `json:"lucky_gift_profit"` + GameTurnover int64 `json:"game_turnover"` + GameProfit int64 `json:"game_profit"` + ARPUUSDMinor int64 `json:"arpu_usd_minor"` + ARPPUUSDMinor int64 `json:"arppu_usd_minor"` } type Retention struct { @@ -94,6 +126,35 @@ type LuckyGiftPoolStat struct { ProfitRate float64 `json:"profit_rate"` } +type CountryBreakdown struct { + CountryID int64 `json:"country_id"` + RegionID int64 `json:"region_id"` + NewUsers int64 `json:"new_users"` + ActiveUsers int64 `json:"active_users"` + PaidUsers int64 `json:"paid_users"` + RechargeUSDMinor int64 `json:"recharge_usd_minor"` + NewUserRechargeUSDMinor int64 `json:"new_user_recharge_usd_minor"` + CoinSellerRechargeUSDMinor int64 `json:"coin_seller_recharge_usd_minor"` + MifaPayRechargeUSDMinor int64 `json:"mifapay_recharge_usd_minor"` + GoogleRechargeUSDMinor int64 `json:"google_recharge_usd_minor"` + CoinSellerTransferCoin int64 `json:"coin_seller_transfer_coin"` + GiftCoinSpent int64 `json:"gift_coin_spent"` + LuckyGiftTurnover int64 `json:"lucky_gift_turnover"` + LuckyGiftPayout int64 `json:"lucky_gift_payout"` + LuckyGiftPayers int64 `json:"lucky_gift_payers"` + LuckyGiftProfit int64 `json:"lucky_gift_profit"` + LuckyGiftPayoutRate float64 `json:"lucky_gift_payout_rate"` + LuckyGiftProfitRate float64 `json:"lucky_gift_profit_rate"` + GameTurnover int64 `json:"game_turnover"` + GamePayout int64 `json:"game_payout"` + GameRefund int64 `json:"game_refund"` + GamePlayers int64 `json:"game_players"` + GameProfit int64 `json:"game_profit"` + GameProfitRate float64 `json:"game_profit_rate"` + ARPUUSDMinor int64 `json:"arpu_usd_minor"` + ARPPUUSDMinor int64 `json:"arppu_usd_minor"` +} + func (r *Repository) QueryOverview(ctx context.Context, query OverviewQuery) (Overview, error) { app := appcode.Normalize(query.AppCode) if app == "" { @@ -102,31 +163,38 @@ func (r *Repository) QueryOverview(ctx context.Context, query OverviewQuery) (Ov startDay := statDay(query.StartMS) endDay := statDay(query.EndMS - 1) previousStartDay, previousEndDay := previousStatDayRange(startDay, endDay) - overview, err := r.queryAppDayOverviewTotals(ctx, app, startDay, endDay, query.CountryID) + overview, err := r.queryAppDayOverviewTotals(ctx, app, startDay, endDay, query.CountryID, query.RegionID) if err != nil { return Overview{}, err } - previous, err := r.queryAppDayOverviewTotals(ctx, app, previousStartDay, previousEndDay, query.CountryID) + previous, err := r.queryAppDayOverviewTotals(ctx, app, previousStartDay, previousEndDay, query.CountryID, query.RegionID) if err != nil { return Overview{}, err } overview.AppCode, overview.StartMS, overview.EndMS, overview.CountryID = app, query.StartMS, query.EndMS, query.CountryID previous.AppCode, previous.CountryID = app, query.CountryID + // 新增付费用户来自日充值用户去重表和注册表的交集；只读聚合小表，避免 Databi 页面回扫钱包或订单明细。 + if overview.NewPaidUsers, err = r.queryNewPaidUsers(ctx, app, startDay, endDay, query.CountryID, query.RegionID); err != nil { + return Overview{}, err + } + if previous.NewPaidUsers, err = r.queryNewPaidUsers(ctx, app, previousStartDay, previousEndDay, query.CountryID, query.RegionID); err != nil { + return Overview{}, err + } // Pool 明细和顶部汇总来自同一个统计窗口；Databi 用它来下钻展示每个 pool_id 的流水、返奖和利润。 - pools, err := r.queryLuckyGiftPools(ctx, app, startDay, endDay, query.CountryID) + pools, err := r.queryLuckyGiftPools(ctx, app, startDay, endDay, query.CountryID, query.RegionID) if err != nil { return Overview{}, err } overview.LuckyGiftPools = pools // 幸运礼物流水和返奖以 activity-service 的抽奖事实增量日汇总为准；这里仅读小表，避免 Databi 页面扫描 lucky_draw_records。 - if activityPools, ok, err := r.queryLuckyGiftPoolsFromActivityStats(ctx, app, startDay, endDay, query.CountryID); err != nil { + if activityPools, ok, err := r.queryLuckyGiftPoolsFromActivityStats(ctx, app, startDay, endDay, query.CountryID, query.RegionID); err != nil { return Overview{}, err } else if ok { overview.LuckyGiftPools = activityPools overview.LuckyGiftTurnover, overview.LuckyGiftPayout = sumLuckyGiftPools(activityPools) } // 上一周期也使用相同的日聚合口径，保证“与昨日相比/前周期相比”不额外查询明细表。 - if activityPools, ok, err := r.queryLuckyGiftPoolsFromActivityStats(ctx, app, previousStartDay, previousEndDay, query.CountryID); err != nil { + if activityPools, ok, err := r.queryLuckyGiftPoolsFromActivityStats(ctx, app, previousStartDay, previousEndDay, query.CountryID, query.RegionID); err != nil { return Overview{}, err } else if ok { previous.LuckyGiftTurnover, previous.LuckyGiftPayout = sumLuckyGiftPools(activityPools) @@ -134,26 +202,49 @@ func (r *Repository) QueryOverview(ctx context.Context, query OverviewQuery) (Ov applyOverviewDerivedMetrics(&overview) applyOverviewDerivedMetrics(&previous) overview.applyDeltaRates(previous) - retention, err := r.queryRetention(ctx, app, startDay, endDay, query.CountryID) + retention, err := r.queryRetention(ctx, app, startDay, endDay, query.CountryID, query.RegionID) if err != nil { return Overview{}, err } overview.Retention = retention - ranking, err := r.queryGameRanking(ctx, app, startDay, endDay, query.CountryID) + ranking, err := r.queryGameRanking(ctx, app, startDay, endDay, query.CountryID, query.RegionID) if err != nil { return Overview{}, err } overview.GameRanking = ranking + if query.CountryID <= 0 { + breakdown, err := r.queryCountryBreakdown(ctx, app, startDay, endDay, query.RegionID) + if err != nil { + return Overview{}, err + } + overview.CountryBreakdown = breakdown + } + seriesStartDay, seriesEndDay := startDay, endDay + if query.SeriesStartMS > 0 && query.SeriesEndMS > query.SeriesStartMS { + seriesStartDay = statDay(query.SeriesStartMS) + seriesEndDay = statDay(query.SeriesEndMS - 1) + } + // daily_series 服务卡片内 sparkline 和弹窗折线图；一条 GROUP BY stat_day 查询返回全部指标，避免前端逐日请求。 + dailySeries, err := r.queryDailySeries(ctx, app, seriesStartDay, seriesEndDay, query.CountryID, query.RegionID) + if err != nil { + return Overview{}, err + } + overview.DailySeries = dailySeries return overview, nil } -func (r *Repository) queryAppDayOverviewTotals(ctx context.Context, app, startDay, endDay string, countryID int64) (Overview, error) { +func (r *Repository) queryAppDayOverviewTotals(ctx context.Context, app, startDay, endDay string, countryID int64, regionID int64) (Overview, error) { args := []any{app, startDay, endDay} - countryFilter := "" + filter := "" if countryID > 0 { - countryFilter = " AND country_id = ?" + filter += " AND country_id = ?" args = append(args, countryID) } + if regionID > 0 { + // 旧统计行曾把 region_id 写入 country_id；兼容分支只服务历史读数，新事件写入后会走显式 region_id。 + filter += " AND (region_id = ? OR (region_id = 0 AND country_id = ?))" + args = append(args, regionID, regionID) + } row := r.db.QueryRowContext(ctx, ` SELECT COALESCE(SUM(new_users),0), COALESCE(SUM(active_users),0), COALESCE(SUM(paid_users),0), COALESCE(SUM(recharge_usd_minor),0), COALESCE(SUM(new_user_recharge_usd_minor),0), @@ -164,7 +255,7 @@ func (r *Repository) queryAppDayOverviewTotals(ctx context.Context, app, startDa COALESCE(SUM(game_turnover),0), COALESCE(SUM(game_payout),0), COALESCE(SUM(game_refund),0), COALESCE(SUM(game_players),0) FROM stat_app_day_country - WHERE app_code = ? AND stat_day BETWEEN ? AND ?`+countryFilter, args...) + WHERE app_code = ? AND stat_day BETWEEN ? AND ?`+filter, args...) var overview Overview if err := row.Scan(&overview.NewUsers, &overview.ActiveUsers, &overview.PaidUsers, &overview.RechargeUSDMinor, &overview.NewUserRechargeUSDMinor, &overview.CoinSellerRechargeUSDMinor, &overview.MifaPayRechargeUSDMinor, &overview.GoogleRechargeUSDMinor, &overview.CoinSellerTransferCoin, &overview.GiftCoinSpent, &overview.LuckyGiftTurnover, &overview.LuckyGiftPayout, &overview.LuckyGiftPayers, &overview.GameTurnover, &overview.GamePayout, &overview.GameRefund, &overview.GamePlayers); err != nil { return Overview{}, err @@ -172,12 +263,144 @@ func (r *Repository) queryAppDayOverviewTotals(ctx context.Context, app, startDa return overview, nil } +func (r *Repository) queryNewPaidUsers(ctx context.Context, app, startDay, endDay string, countryID int64, regionID int64) (int64, error) { + args := []any{app, startDay, endDay, startDay, endDay} + filter := "" + if countryID > 0 { + filter += " AND p.country_id = ?" + args = append(args, countryID) + } + if regionID > 0 { + // p 表和 r 表都已经是统计小表，这里只按充值发生地过滤，保证新增付费用户口径和付费用户卡片一致。 + filter += " AND (p.region_id = ? OR (p.region_id = 0 AND p.country_id = ?))" + args = append(args, regionID, regionID) + } + row := r.db.QueryRowContext(ctx, ` + SELECT COUNT(DISTINCT p.user_id) + FROM stat_recharge_day_payers p + JOIN stat_user_registration r ON r.app_code = p.app_code AND r.user_id = p.user_id + WHERE p.app_code = ? AND p.stat_day BETWEEN ? AND ? + AND r.registered_day BETWEEN ? AND ?`+filter, args...) + var count int64 + if err := row.Scan(&count); err != nil && err != sql.ErrNoRows { + return 0, err + } + return count, nil +} + +func (r *Repository) queryDailySeries(ctx context.Context, app, startDay, endDay string, countryID int64, regionID int64) ([]DailySeriesItem, error) { + args := []any{app, startDay, endDay} + filter := "" + if countryID > 0 { + filter += " AND country_id = ?" + args = append(args, countryID) + } + if regionID > 0 { + // 和总览一致兼容历史 region_id 写到 country_id 的聚合行；新数据优先读显式 region_id。 + filter += " AND (region_id = ? OR (region_id = 0 AND country_id = ?))" + args = append(args, regionID, regionID) + } + rows, err := r.db.QueryContext(ctx, ` + SELECT DATE_FORMAT(stat_day, '%Y-%m-%d'), + COALESCE(SUM(new_users),0), COALESCE(SUM(active_users),0), COALESCE(SUM(paid_users),0), + COALESCE(SUM(recharge_usd_minor),0), COALESCE(SUM(new_user_recharge_usd_minor),0), + COALESCE(SUM(coin_seller_recharge_usd_minor),0), COALESCE(SUM(mifapay_recharge_usd_minor),0), + COALESCE(SUM(google_recharge_usd_minor),0), COALESCE(SUM(coin_seller_transfer_coin),0), + COALESCE(SUM(gift_coin_spent),0), + COALESCE(SUM(lucky_gift_turnover),0), COALESCE(SUM(lucky_gift_payout),0), + COALESCE(SUM(game_turnover),0), COALESCE(SUM(game_payout),0), COALESCE(SUM(game_refund),0) + FROM stat_app_day_country + WHERE app_code = ? AND stat_day BETWEEN ? AND ?`+filter+` + GROUP BY stat_day + ORDER BY stat_day ASC`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := []DailySeriesItem{} + for rows.Next() { + var item DailySeriesItem + var gamePayout, gameRefund int64 + if err := rows.Scan(&item.StatDay, &item.NewUsers, &item.ActiveUsers, &item.PaidUsers, &item.RechargeUSDMinor, &item.NewUserRechargeUSDMinor, &item.CoinSellerRechargeUSDMinor, &item.MifaPayRechargeUSDMinor, &item.GoogleRechargeUSDMinor, &item.CoinSellerTransferCoin, &item.GiftCoinSpent, &item.LuckyGiftTurnover, &item.LuckyGiftPayout, &item.GameTurnover, &gamePayout, &gameRefund); err != nil { + return nil, err + } + item.Label = item.StatDay + item.GameProfit = item.GameTurnover - gamePayout - gameRefund + applyDailySeriesDerivedMetrics(&item) + out = append(out, item) + } + if err := rows.Err(); err != nil { + return nil, err + } + newPaidUsersByDay, err := r.queryDailyNewPaidUsers(ctx, app, startDay, endDay, countryID, regionID) + if err != nil { + return nil, err + } + for index := range out { + out[index].NewPaidUsers = newPaidUsersByDay[out[index].StatDay] + } + if activityDaily, ok, err := r.queryLuckyGiftDailyFromActivityStats(ctx, app, startDay, endDay, countryID, regionID); err != nil { + return nil, err + } else if ok { + out = mergeActivityLuckyGiftDaily(out, activityDaily) + } + return out, nil +} + +func (r *Repository) queryDailyNewPaidUsers(ctx context.Context, app, startDay, endDay string, countryID int64, regionID int64) (map[string]int64, error) { + args := []any{app, startDay, endDay, startDay, endDay} + filter := "" + if countryID > 0 { + filter += " AND p.country_id = ?" + args = append(args, countryID) + } + if regionID > 0 { + filter += " AND (p.region_id = ? OR (p.region_id = 0 AND p.country_id = ?))" + args = append(args, regionID, regionID) + } + rows, err := r.db.QueryContext(ctx, ` + SELECT DATE_FORMAT(p.stat_day, '%Y-%m-%d'), COUNT(DISTINCT p.user_id) + FROM stat_recharge_day_payers p + JOIN stat_user_registration r ON r.app_code = p.app_code AND r.user_id = p.user_id + WHERE p.app_code = ? AND p.stat_day BETWEEN ? AND ? + AND r.registered_day BETWEEN ? AND ?`+filter+` + GROUP BY p.stat_day`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[string]int64{} + for rows.Next() { + var day string + var count int64 + if err := rows.Scan(&day, &count); err != nil { + return nil, err + } + out[day] = count + } + return out, rows.Err() +} + +func applyDailySeriesDerivedMetrics(item *DailySeriesItem) { + if item == nil { + return + } + channelRechargeUSDMinor := item.MifaPayRechargeUSDMinor + item.GoogleRechargeUSDMinor + item.CoinSellerRechargeUSDMinor + if channelRechargeUSDMinor > item.RechargeUSDMinor { + item.RechargeUSDMinor = channelRechargeUSDMinor + } + item.RechargeUsers = item.PaidUsers + item.LuckyGiftProfit = item.LuckyGiftTurnover - item.LuckyGiftPayout + item.ARPUUSDMinor = div(item.RechargeUSDMinor, item.ActiveUsers) + item.ARPPUUSDMinor = div(item.RechargeUSDMinor, item.PaidUsers) +} + func applyOverviewDerivedMetrics(overview *Overview) { if overview == nil { return } - // 历史聚合行可能只写了真实 USD 渠道列、没有同步修正 recharge_usd_minor；查询时用渠道合计兜底，但币商转普通金币只展示金币卡片，不进入用户 USDT 充值。 - channelRechargeUSDMinor := overview.MifaPayRechargeUSDMinor + overview.GoogleRechargeUSDMinor + // 历史聚合行可能只写了真实 USD 渠道列、没有同步修正 recharge_usd_minor；查询时用渠道合计兜底。币商进货是 USDT 充值，币商转普通金币仍只展示金币卡片。 + channelRechargeUSDMinor := overview.MifaPayRechargeUSDMinor + overview.GoogleRechargeUSDMinor + overview.CoinSellerRechargeUSDMinor if channelRechargeUSDMinor > overview.RechargeUSDMinor { overview.RechargeUSDMinor = channelRechargeUSDMinor } @@ -199,6 +422,7 @@ func (overview *Overview) applyDeltaRates(previous Overview) { overview.NewUserRechargeDeltaRate = deltaRate(overview.NewUserRechargeUSDMinor, previous.NewUserRechargeUSDMinor) overview.ActiveUsersDeltaRate = deltaRate(overview.ActiveUsers, previous.ActiveUsers) overview.PaidUsersDeltaRate = deltaRate(overview.PaidUsers, previous.PaidUsers) + overview.NewPaidUsersDeltaRate = deltaRate(overview.NewPaidUsers, previous.NewPaidUsers) overview.ARPUDeltaRate = deltaRate(overview.ARPUUSDMinor, previous.ARPUUSDMinor) overview.ARPPUDeltaRate = deltaRate(overview.ARPPUUSDMinor, previous.ARPPUUSDMinor) overview.GiftCoinSpentDeltaRate = deltaRate(overview.GiftCoinSpent, previous.GiftCoinSpent) @@ -210,13 +434,17 @@ func (overview *Overview) applyDeltaRates(previous Overview) { overview.GameProfitDeltaRate = deltaRate(overview.GameProfit, previous.GameProfit) } -func (r *Repository) queryRetention(ctx context.Context, app, startDay, endDay string, countryID int64) (Retention, error) { +func (r *Repository) queryRetention(ctx context.Context, app, startDay, endDay string, countryID int64, regionID int64) (Retention, error) { filter := "" args := []any{app, startDay, endDay} if countryID > 0 { filter = " AND r.country_id = ?" args = append(args, countryID) } + if regionID > 0 { + filter += " AND (r.region_id = ? OR (r.region_id = 0 AND r.country_id = ?))" + args = append(args, regionID, regionID) + } row := r.db.QueryRowContext(ctx, ` SELECT COUNT(*), COALESCE(SUM(EXISTS(SELECT 1 FROM stat_user_day_activity a WHERE a.app_code=r.app_code AND a.user_id=r.user_id AND a.stat_day=DATE_ADD(r.registered_day, INTERVAL 1 DAY))),0), @@ -234,13 +462,17 @@ func (r *Repository) queryRetention(ctx context.Context, app, startDay, endDay s return out, nil } -func (r *Repository) queryGameRanking(ctx context.Context, app, startDay, endDay string, countryID int64) ([]GameRank, error) { +func (r *Repository) queryGameRanking(ctx context.Context, app, startDay, endDay string, countryID int64, regionID int64) ([]GameRank, error) { args := []any{app, startDay, endDay} filter := "" if countryID > 0 { filter = " AND country_id = ?" args = append(args, countryID) } + if regionID > 0 { + filter += " AND (region_id = ? OR (region_id = 0 AND country_id = ?))" + args = append(args, regionID, regionID) + } rows, err := r.db.QueryContext(ctx, ` SELECT platform_code, game_id, COALESCE(SUM(turnover_coin),0), COALESCE(SUM(payout_coin),0), COALESCE(SUM(refund_coin),0), COALESCE(SUM(player_count),0) @@ -266,13 +498,72 @@ func (r *Repository) queryGameRanking(ctx context.Context, app, startDay, endDay return out, rows.Err() } -func (r *Repository) queryLuckyGiftPools(ctx context.Context, app, startDay, endDay string, countryID int64) ([]LuckyGiftPoolStat, error) { +func (r *Repository) queryCountryBreakdown(ctx context.Context, app, startDay, endDay string, regionID int64) ([]CountryBreakdown, error) { + args := []any{app, startDay, endDay} + filter := "" + if regionID > 0 { + filter = " AND (region_id = ? OR (region_id = 0 AND country_id = ?))" + args = append(args, regionID, regionID) + } + rows, err := r.db.QueryContext(ctx, ` + SELECT country_id, COALESCE(MAX(region_id),0), + COALESCE(SUM(new_users),0), COALESCE(SUM(active_users),0), COALESCE(SUM(paid_users),0), + COALESCE(SUM(recharge_usd_minor),0), COALESCE(SUM(new_user_recharge_usd_minor),0), + COALESCE(SUM(coin_seller_recharge_usd_minor),0), COALESCE(SUM(mifapay_recharge_usd_minor),0), + COALESCE(SUM(google_recharge_usd_minor),0), COALESCE(SUM(coin_seller_transfer_coin),0), + COALESCE(SUM(gift_coin_spent),0), + COALESCE(SUM(lucky_gift_turnover),0), COALESCE(SUM(lucky_gift_payout),0), COALESCE(SUM(lucky_gift_payers),0), + COALESCE(SUM(game_turnover),0), COALESCE(SUM(game_payout),0), COALESCE(SUM(game_refund),0), + COALESCE(SUM(game_players),0) + FROM stat_app_day_country + WHERE app_code = ? AND stat_day BETWEEN ? AND ?`+filter+` + GROUP BY country_id + ORDER BY COALESCE(SUM(recharge_usd_minor),0) DESC, COALESCE(SUM(active_users),0) DESC, country_id ASC + LIMIT 500`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := []CountryBreakdown{} + for rows.Next() { + var item CountryBreakdown + if err := rows.Scan(&item.CountryID, &item.RegionID, &item.NewUsers, &item.ActiveUsers, &item.PaidUsers, &item.RechargeUSDMinor, &item.NewUserRechargeUSDMinor, &item.CoinSellerRechargeUSDMinor, &item.MifaPayRechargeUSDMinor, &item.GoogleRechargeUSDMinor, &item.CoinSellerTransferCoin, &item.GiftCoinSpent, &item.LuckyGiftTurnover, &item.LuckyGiftPayout, &item.LuckyGiftPayers, &item.GameTurnover, &item.GamePayout, &item.GameRefund, &item.GamePlayers); err != nil { + return nil, err + } + applyCountryBreakdownDerivedMetrics(&item) + out = append(out, item) + } + return out, rows.Err() +} + +func applyCountryBreakdownDerivedMetrics(item *CountryBreakdown) { + if item == nil { + return + } + channelRechargeUSDMinor := item.MifaPayRechargeUSDMinor + item.GoogleRechargeUSDMinor + item.CoinSellerRechargeUSDMinor + if channelRechargeUSDMinor > item.RechargeUSDMinor { + item.RechargeUSDMinor = channelRechargeUSDMinor + } + item.GameProfit = item.GameTurnover - item.GamePayout - item.GameRefund + item.GameProfitRate = ratio(item.GameProfit, item.GameTurnover) + item.ARPUUSDMinor = div(item.RechargeUSDMinor, item.ActiveUsers) + item.ARPPUUSDMinor = div(item.RechargeUSDMinor, item.PaidUsers) + item.LuckyGiftProfit = item.LuckyGiftTurnover - item.LuckyGiftPayout + item.LuckyGiftPayoutRate = ratio(item.LuckyGiftPayout, item.LuckyGiftTurnover) + item.LuckyGiftProfitRate = ratio(item.LuckyGiftProfit, item.LuckyGiftTurnover) +} + +func (r *Repository) queryLuckyGiftPools(ctx context.Context, app, startDay, endDay string, countryID int64, regionID int64) ([]LuckyGiftPoolStat, error) { args := []any{app, startDay, endDay} filter := "" if countryID > 0 { filter = " AND country_id = ?" args = append(args, countryID) } + if regionID > 0 { + filter += " AND (region_id = ? OR (region_id = 0 AND country_id = ?))" + args = append(args, regionID, regionID) + } rows, err := r.db.QueryContext(ctx, ` SELECT pool_id, COALESCE(SUM(turnover_coin),0), COALESCE(SUM(payout_coin),0) FROM stat_lucky_gift_pool_day_country @@ -297,13 +588,16 @@ func (r *Repository) queryLuckyGiftPools(ctx context.Context, app, startDay, end return out, rows.Err() } -func (r *Repository) queryLuckyGiftPoolsFromActivityStats(ctx context.Context, app, startDay, endDay string, countryID int64) ([]LuckyGiftPoolStat, bool, error) { +func (r *Repository) queryLuckyGiftPoolsFromActivityStats(ctx context.Context, app, startDay, endDay string, countryID int64, regionID int64) ([]LuckyGiftPoolStat, bool, error) { if r == nil || r.activityDB == nil { return nil, false, nil } args := []any{app, startDay, endDay} filter := "" - if countryID > 0 { + if regionID > 0 { + filter = " AND visible_region_id = ?" + args = append(args, regionID) + } else if countryID > 0 { filter = " AND visible_region_id = ?" args = append(args, countryID) } @@ -337,6 +631,69 @@ func (r *Repository) queryLuckyGiftPoolsFromActivityStats(ctx context.Context, a return out, len(out) > 0, nil } +func (r *Repository) queryLuckyGiftDailyFromActivityStats(ctx context.Context, app, startDay, endDay string, countryID int64, regionID int64) (map[string]LuckyGiftPoolStat, bool, error) { + if r == nil || r.activityDB == nil { + return nil, false, nil + } + args := []any{app, startDay, endDay} + filter := "" + if regionID > 0 { + filter = " AND visible_region_id = ?" + args = append(args, regionID) + } else if countryID > 0 { + filter = " AND visible_region_id = ?" + args = append(args, countryID) + } + rows, err := r.activityDB.QueryContext(ctx, ` + SELECT DATE_FORMAT(stat_day, '%Y-%m-%d'), COALESCE(SUM(turnover_coin),0), COALESCE(SUM(payout_coin),0) + FROM lucky_draw_pool_day_stats + WHERE app_code = ? AND stat_day BETWEEN ? AND ? AND gift_id = ''`+filter+` + GROUP BY stat_day`, args...) + if err != nil { + if isMissingTableError(err) { + return nil, false, nil + } + return nil, false, err + } + defer rows.Close() + out := map[string]LuckyGiftPoolStat{} + for rows.Next() { + var day string + var item LuckyGiftPoolStat + if err := rows.Scan(&day, &item.TurnoverCoin, &item.PayoutCoin); err != nil { + return nil, false, err + } + item.ProfitCoin = item.TurnoverCoin - item.PayoutCoin + out[day] = item + } + if err := rows.Err(); err != nil { + return nil, false, err + } + return out, len(out) > 0, nil +} + +func mergeActivityLuckyGiftDaily(items []DailySeriesItem, activityDaily map[string]LuckyGiftPoolStat) []DailySeriesItem { + byDay := make(map[string]int, len(items)) + for index := range items { + byDay[items[index].StatDay] = index + } + for day, activity := range activityDaily { + index, exists := byDay[day] + if !exists { + items = append(items, DailySeriesItem{StatDay: day, Label: day}) + index = len(items) - 1 + byDay[day] = index + } + items[index].LuckyGiftTurnover = activity.TurnoverCoin + items[index].LuckyGiftPayout = activity.PayoutCoin + applyDailySeriesDerivedMetrics(&items[index]) + } + sort.SliceStable(items, func(left, right int) bool { + return items[left].StatDay < items[right].StatDay + }) + return items +} + func sumLuckyGiftPools(pools []LuckyGiftPoolStat) (int64, int64) { var turnover, payout int64 for _, pool := range pools { diff --git a/services/statistics-service/internal/storage/mysql/query_test.go b/services/statistics-service/internal/storage/mysql/query_test.go index c84e4e7f..72943502 100644 --- a/services/statistics-service/internal/storage/mysql/query_test.go +++ b/services/statistics-service/internal/storage/mysql/query_test.go @@ -107,3 +107,43 @@ func TestQueryOverviewReturnsPreviousPeriodDeltaRates(t *testing.T) { t.Fatalf("game delta rates mismatch: %+v", overview) } } + +func TestQueryOverviewReturnsCountryBreakdownAndIncludesCoinSellerRecharge(t *testing.T) { + ctx := context.Background() + _, file, _, _ := runtime.Caller(0) + statsSchema := mysqlschema.New(t, mysqlschema.Config{ + InitDBPath: mysqlschema.InitDBPath(t, file, "..", "..", "..", "deploy", "mysql", "initdb", "001_statistics_service.sql"), + DatabasePrefix: "hy_stats_query_country_test", + }) + repository, err := Open(ctx, statsSchema.DSN) + if err != nil { + t.Fatalf("open repository: %v", err) + } + t.Cleanup(func() { _ = repository.Close() }) + + if _, err := repository.db.ExecContext(ctx, ` + INSERT INTO stat_app_day_country ( + app_code, stat_day, country_id, region_id, active_users, paid_users, recharge_usd_minor, + coin_seller_recharge_usd_minor, google_recharge_usd_minor, updated_at_ms + ) VALUES + ('lalu', '2026-06-06', 86, 210, 10, 2, 150, 400, 150, 1), + ('lalu', '2026-06-06', 66, 210, 5, 1, 100, 0, 100, 1), + ('lalu', '2026-06-06', 99, 220, 3, 1, 900, 0, 900, 1)`); err != nil { + t.Fatalf("seed country statistics: %v", err) + } + + start := time.Date(2026, 6, 6, 0, 0, 0, 0, time.UTC).UnixMilli() + overview, err := repository.QueryOverview(ctx, OverviewQuery{AppCode: "lalu", StartMS: start, EndMS: start + int64(24*time.Hour/time.Millisecond), RegionID: 210}) + if err != nil { + t.Fatalf("query overview: %v", err) + } + if overview.RechargeUSDMinor != 650 || overview.CoinSellerRechargeUSDMinor != 400 { + t.Fatalf("overview recharge mismatch: %+v", overview) + } + if len(overview.CountryBreakdown) != 2 { + t.Fatalf("country breakdown should come from aggregate query: %+v", overview.CountryBreakdown) + } + if overview.CountryBreakdown[0].CountryID != 86 || overview.CountryBreakdown[0].RechargeUSDMinor != 550 || overview.CountryBreakdown[0].ARPPUUSDMinor != 275 { + t.Fatalf("first country breakdown mismatch: %+v", overview.CountryBreakdown[0]) + } +} diff --git a/services/statistics-service/internal/storage/mysql/repository.go b/services/statistics-service/internal/storage/mysql/repository.go index e34f76d5..9b0ca68b 100644 --- a/services/statistics-service/internal/storage/mysql/repository.go +++ b/services/statistics-service/internal/storage/mysql/repository.go @@ -104,6 +104,7 @@ func (r *Repository) Migrate(ctx context.Context) error { ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE IF NOT EXISTS stat_app_day_country ( 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, new_users BIGINT NOT NULL DEFAULT 0, active_users BIGINT NOT NULL DEFAULT 0, paid_users BIGINT NOT NULL DEFAULT 0, recharge_usd_minor BIGINT NOT NULL DEFAULT 0, new_user_recharge_usd_minor BIGINT NOT NULL DEFAULT 0, coin_seller_recharge_usd_minor BIGINT NOT NULL DEFAULT 0, mifapay_recharge_usd_minor BIGINT NOT NULL DEFAULT 0, @@ -112,49 +113,63 @@ func (r *Repository) Migrate(ctx context.Context) error { lucky_gift_turnover BIGINT NOT NULL DEFAULT 0, lucky_gift_payout BIGINT NOT NULL DEFAULT 0, lucky_gift_payers BIGINT NOT NULL DEFAULT 0, game_turnover BIGINT NOT NULL DEFAULT 0, game_players BIGINT NOT NULL DEFAULT 0, game_payout BIGINT NOT NULL DEFAULT 0, game_refund BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL, - PRIMARY KEY (app_code, stat_day, country_id) + PRIMARY KEY (app_code, stat_day, country_id), + KEY idx_stat_app_day_region (app_code, stat_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE IF NOT EXISTS stat_user_day_activity ( 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, user_id BIGINT NOT NULL, first_active_at_ms BIGINT NOT NULL, PRIMARY KEY (app_code, stat_day, user_id), - KEY idx_stat_user_day_country (app_code, stat_day, country_id) + KEY idx_stat_user_day_country (app_code, stat_day, country_id), + KEY idx_stat_user_day_region (app_code, stat_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE IF NOT EXISTS stat_user_registration ( app_code VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, registered_day DATE NOT NULL, - country_id BIGINT NOT NULL DEFAULT 0, registered_at_ms BIGINT NOT NULL, - PRIMARY KEY (app_code, user_id), KEY idx_stat_user_registration_day (app_code, registered_day, country_id) + country_id BIGINT NOT NULL DEFAULT 0, region_id BIGINT NOT NULL DEFAULT 0, registered_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, user_id), KEY idx_stat_user_registration_day (app_code, registered_day, country_id), + KEY idx_stat_user_registration_region (app_code, registered_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE IF NOT EXISTS stat_recharge_day_payers ( 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, user_id BIGINT NOT NULL, first_paid_at_ms BIGINT NOT NULL, - PRIMARY KEY (app_code, stat_day, user_id), KEY idx_stat_recharge_payer_country (app_code, stat_day, country_id) + PRIMARY KEY (app_code, stat_day, user_id), KEY idx_stat_recharge_payer_country (app_code, stat_day, country_id), + KEY idx_stat_recharge_payer_region (app_code, stat_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE IF NOT EXISTS stat_lucky_gift_day_payers ( 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, user_id BIGINT NOT NULL, first_paid_at_ms BIGINT NOT NULL, PRIMARY KEY (app_code, stat_day, user_id), - KEY idx_stat_lucky_payer_country (app_code, stat_day, country_id) + KEY idx_stat_lucky_payer_country (app_code, stat_day, country_id), + KEY idx_stat_lucky_payer_region (app_code, stat_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE IF NOT EXISTS stat_lucky_gift_pool_day_country ( 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, pool_id VARCHAR(96) NOT NULL, turnover_coin BIGINT NOT NULL DEFAULT 0, payout_coin BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL, PRIMARY KEY (app_code, stat_day, country_id, pool_id), - KEY idx_stat_lucky_pool_overview (app_code, stat_day, pool_id) + KEY idx_stat_lucky_pool_overview (app_code, stat_day, pool_id), + KEY idx_stat_lucky_pool_region (app_code, stat_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE IF NOT EXISTS stat_game_day_country ( 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, platform_code VARCHAR(64) NOT NULL DEFAULT '', game_id VARCHAR(96) NOT NULL, turnover_coin BIGINT NOT NULL DEFAULT 0, payout_coin BIGINT NOT NULL DEFAULT 0, refund_coin BIGINT NOT NULL DEFAULT 0, player_count BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL, - PRIMARY KEY (app_code, stat_day, country_id, platform_code, game_id) + PRIMARY KEY (app_code, stat_day, country_id, platform_code, game_id), + KEY idx_stat_game_day_region (app_code, stat_day, region_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE IF NOT EXISTS stat_game_day_players ( 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, platform_code VARCHAR(64) NOT NULL DEFAULT '', game_id VARCHAR(96) NOT NULL, user_id BIGINT NOT NULL, first_played_at_ms BIGINT NOT NULL, - PRIMARY KEY (app_code, stat_day, country_id, platform_code, game_id, user_id) + 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`, } for _, statement := range statements { @@ -168,9 +183,25 @@ func (r *Repository) Migrate(ctx context.Context) error { `ALTER TABLE stat_app_day_country ADD COLUMN lucky_gift_payout BIGINT NOT NULL DEFAULT 0 AFTER lucky_gift_turnover`, `ALTER TABLE stat_app_day_country ADD COLUMN coin_seller_transfer_coin BIGINT NOT NULL DEFAULT 0 AFTER google_recharge_usd_minor`, `ALTER TABLE stat_game_day_country ADD COLUMN refund_coin BIGINT NOT NULL DEFAULT 0 AFTER payout_coin`, + `ALTER TABLE stat_app_day_country ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`, + `ALTER TABLE stat_user_day_activity ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`, + `ALTER TABLE stat_user_registration ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`, + `ALTER TABLE stat_recharge_day_payers ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`, + `ALTER TABLE stat_lucky_gift_day_payers ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`, + `ALTER TABLE stat_lucky_gift_pool_day_country ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`, + `ALTER TABLE stat_game_day_country ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`, + `ALTER TABLE stat_game_day_players ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`, + `ALTER TABLE stat_app_day_country ADD KEY idx_stat_app_day_region (app_code, stat_day, region_id)`, + `ALTER TABLE stat_user_day_activity ADD KEY idx_stat_user_day_region (app_code, stat_day, region_id)`, + `ALTER TABLE stat_user_registration ADD KEY idx_stat_user_registration_region (app_code, registered_day, region_id)`, + `ALTER TABLE stat_recharge_day_payers ADD KEY idx_stat_recharge_payer_region (app_code, stat_day, region_id)`, + `ALTER TABLE stat_lucky_gift_day_payers ADD KEY idx_stat_lucky_payer_region (app_code, stat_day, region_id)`, + `ALTER TABLE stat_lucky_gift_pool_day_country ADD KEY idx_stat_lucky_pool_region (app_code, stat_day, region_id)`, + `ALTER TABLE stat_game_day_country ADD KEY idx_stat_game_day_region (app_code, stat_day, region_id)`, + `ALTER TABLE stat_game_day_players ADD KEY idx_stat_game_player_region (app_code, stat_day, region_id)`, } for _, statement := range alterStatements { - if _, err := r.db.ExecContext(ctx, statement); err != nil && !isDuplicateColumnError(err) { + if _, err := r.db.ExecContext(ctx, statement); err != nil && !isDuplicateSchemaError(err) { return err } } @@ -181,6 +212,7 @@ type UserRegisteredEvent struct { AppCode string EventID string UserID int64 + CountryID int64 RegionID int64 OccurredAtMS int64 } @@ -193,15 +225,28 @@ type RechargeEvent struct { USDMinor int64 CoinAmount int64 RechargeSequence int64 + TargetCountryID int64 TargetRegionID int64 RechargeType string OccurredAtMS int64 } +type CoinSellerStockEvent struct { + AppCode string + EventID string + SellerUserID int64 + USDMinor int64 + CoinAmount int64 + CountryID int64 + RegionID int64 + OccurredAtMS int64 +} + type RoomGiftEvent struct { AppCode string EventID string SenderUserID int64 + CountryID int64 VisibleRegionID int64 GiftValue int64 CoinSpent int64 @@ -215,6 +260,7 @@ type LuckyGiftRewardEvent struct { EventType string UserID int64 CountryID int64 + RegionID int64 PoolID string Amount int64 OccurredAtMS int64 @@ -226,6 +272,7 @@ type UserActiveEvent struct { EventType string UserID int64 CountryID int64 + RegionID int64 OccurredAtMS int64 } @@ -234,6 +281,7 @@ type GameOrderEvent struct { EventID string UserID int64 CountryID int64 + RegionID int64 PlatformCode string GameID string OpType string @@ -244,14 +292,14 @@ type GameOrderEvent struct { 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) - countryID := normalizeID(event.RegionID) - if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, nowMS); err != nil { + countryID, regionID := normalizeDimension(event.CountryID, event.RegionID) + if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, regionID, nowMS); err != nil { return err } if _, err := tx.ExecContext(ctx, ` - INSERT IGNORE INTO stat_user_registration (app_code, user_id, registered_day, country_id, registered_at_ms) - VALUES (?, ?, ?, ?, ?) - `, appcode.Normalize(event.AppCode), event.UserID, day, countryID, event.OccurredAtMS); err != nil { + INSERT IGNORE INTO stat_user_registration (app_code, user_id, registered_day, country_id, region_id, registered_at_ms) + VALUES (?, ?, ?, ?, ?, ?) + `, appcode.Normalize(event.AppCode), event.UserID, day, countryID, regionID, event.OccurredAtMS); err != nil { return err } _, err := tx.ExecContext(ctx, ` @@ -266,14 +314,14 @@ func (r *Repository) ConsumeUserRegistered(ctx context.Context, event UserRegist func (r *Repository) ConsumeRecharge(ctx context.Context, event RechargeEvent) error { return r.withEvent(ctx, SourceWallet, event.EventID, event.EventType, func(tx *sql.Tx, nowMS int64) error { day := statDay(event.OccurredAtMS) - countryID := normalizeID(event.TargetRegionID) - if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, nowMS); err != nil { + countryID, regionID := normalizeDimension(event.TargetCountryID, event.TargetRegionID) + if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, regionID, nowMS); err != nil { return err } paidUsers, err := insertUnique(ctx, tx, ` - INSERT IGNORE INTO stat_recharge_day_payers (app_code, stat_day, country_id, user_id, first_paid_at_ms) - VALUES (?, ?, ?, ?, ?) - `, appcode.Normalize(event.AppCode), day, countryID, event.UserID, event.OccurredAtMS) + INSERT IGNORE INTO stat_recharge_day_payers (app_code, stat_day, country_id, region_id, user_id, first_paid_at_ms) + VALUES (?, ?, ?, ?, ?, ?) + `, appcode.Normalize(event.AppCode), day, countryID, regionID, event.UserID, event.OccurredAtMS) if err != nil { return err } @@ -308,11 +356,30 @@ func (r *Repository) ConsumeRecharge(ctx context.Context, event RechargeEvent) e }) } +func (r *Repository) ConsumeCoinSellerStock(ctx context.Context, event CoinSellerStockEvent) error { + return r.withEvent(ctx, SourceWallet, event.EventID, "WalletCoinSellerStockPurchased", func(tx *sql.Tx, nowMS int64) error { + day := statDay(event.OccurredAtMS) + countryID, regionID := normalizeDimension(event.CountryID, event.RegionID) + if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, regionID, nowMS); err != nil { + return err + } + // 币商进货是后台 USDT 入金事实，只进入总充值和币商充值列；它不是普通用户到账，所以不增加用户 USDT 首充和付费用户。 + _, err := tx.ExecContext(ctx, ` + UPDATE stat_app_day_country + SET recharge_usd_minor = recharge_usd_minor + ?, + coin_seller_recharge_usd_minor = coin_seller_recharge_usd_minor + ?, + updated_at_ms = ? + WHERE app_code = ? AND stat_day = ? AND country_id = ? + `, event.USDMinor, event.USDMinor, nowMS, appcode.Normalize(event.AppCode), day, countryID) + return err + }) +} + func (r *Repository) ConsumeRoomGift(ctx context.Context, event RoomGiftEvent) error { return r.withEvent(ctx, SourceRoom, event.EventID, "RoomGiftSent", func(tx *sql.Tx, nowMS int64) error { day := statDay(event.OccurredAtMS) - countryID := normalizeID(event.VisibleRegionID) - if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, nowMS); err != nil { + countryID, regionID := normalizeDimension(event.CountryID, event.VisibleRegionID) + if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, regionID, nowMS); err != nil { return err } luckyTurnover, luckyPayers := int64(0), int64(0) @@ -320,15 +387,15 @@ func (r *Repository) ConsumeRoomGift(ctx context.Context, event RoomGiftEvent) e if poolID != "" { luckyTurnover = event.CoinSpent affected, err := insertUnique(ctx, tx, ` - INSERT IGNORE INTO stat_lucky_gift_day_payers (app_code, stat_day, country_id, user_id, first_paid_at_ms) - VALUES (?, ?, ?, ?, ?) - `, appcode.Normalize(event.AppCode), day, countryID, event.SenderUserID, event.OccurredAtMS) + INSERT IGNORE INTO stat_lucky_gift_day_payers (app_code, stat_day, country_id, region_id, user_id, first_paid_at_ms) + VALUES (?, ?, ?, ?, ?, ?) + `, appcode.Normalize(event.AppCode), day, countryID, regionID, event.SenderUserID, event.OccurredAtMS) if err != nil { return err } luckyPayers = affected // Pool 明细是 Databi 点击下钻的真实口径：每笔带 pool_id 的幸运礼物送礼先按 UTC 日、国家和奖池累加流水，返奖事件稍后用相同维度补齐 payout。 - if err := ensureLuckyGiftPoolDay(ctx, tx, event.AppCode, day, countryID, poolID, nowMS); err != nil { + if err := ensureLuckyGiftPoolDay(ctx, tx, event.AppCode, day, countryID, regionID, poolID, nowMS); err != nil { return err } if _, err := tx.ExecContext(ctx, ` @@ -352,14 +419,14 @@ func (r *Repository) ConsumeRoomGift(ctx context.Context, event RoomGiftEvent) e func (r *Repository) ConsumeLuckyGiftReward(ctx context.Context, event LuckyGiftRewardEvent) error { return r.withEvent(ctx, SourceWallet, event.EventID, "WalletLuckyGiftRewardCredited", func(tx *sql.Tx, nowMS int64) error { day := statDay(event.OccurredAtMS) - countryID := normalizeID(event.CountryID) - if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, nowMS); err != nil { + countryID, regionID := normalizeDimension(event.CountryID, event.RegionID) + if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, regionID, nowMS); err != nil { return err } poolID := strings.TrimSpace(event.PoolID) if poolID != "" { // 钱包返奖事件带回 pool_id 后，按同一 UTC 日和国家写入奖池返奖；利润不落表，查询时用 turnover-payout 实时计算，避免双写漂移。 - if err := ensureLuckyGiftPoolDay(ctx, tx, event.AppCode, day, countryID, poolID, nowMS); err != nil { + if err := ensureLuckyGiftPoolDay(ctx, tx, event.AppCode, day, countryID, regionID, poolID, nowMS); err != nil { return err } if _, err := tx.ExecContext(ctx, ` @@ -381,18 +448,19 @@ func (r *Repository) ConsumeLuckyGiftReward(ctx context.Context, event LuckyGift func (r *Repository) ConsumeUserActive(ctx context.Context, event UserActiveEvent) error { return r.withEvent(ctx, SourceRoom, event.EventID, event.EventType, func(tx *sql.Tx, nowMS int64) error { - return applyActive(ctx, tx, event.AppCode, statDay(event.OccurredAtMS), normalizeID(event.CountryID), event.UserID, event.OccurredAtMS, nowMS) + countryID, regionID := normalizeDimension(event.CountryID, event.RegionID) + return applyActive(ctx, tx, event.AppCode, statDay(event.OccurredAtMS), countryID, regionID, event.UserID, event.OccurredAtMS, nowMS) }) } func (r *Repository) ConsumeGameOrder(ctx context.Context, event GameOrderEvent) error { return r.withEvent(ctx, SourceGame, event.EventID, "GameOrderSettled", func(tx *sql.Tx, nowMS int64) error { day := statDay(event.OccurredAtMS) - countryID := normalizeID(event.CountryID) - if err := applyActive(ctx, tx, event.AppCode, day, countryID, event.UserID, event.OccurredAtMS, nowMS); err != nil { + countryID, regionID := normalizeDimension(event.CountryID, event.RegionID) + if err := applyActive(ctx, tx, event.AppCode, day, countryID, regionID, event.UserID, event.OccurredAtMS, nowMS); err != nil { return err } - if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, nowMS); err != nil { + if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, regionID, nowMS); err != nil { return err } turnover, payout, refund := int64(0), int64(0), int64(0) @@ -407,15 +475,15 @@ func (r *Repository) ConsumeGameOrder(ctx context.Context, event GameOrderEvent) playerDelta := int64(0) if turnover > 0 { affected, err := insertUnique(ctx, tx, ` - INSERT IGNORE INTO stat_game_day_players (app_code, stat_day, country_id, platform_code, game_id, user_id, first_played_at_ms) - VALUES (?, ?, ?, ?, ?, ?, ?) - `, appcode.Normalize(event.AppCode), day, countryID, event.PlatformCode, event.GameID, event.UserID, event.OccurredAtMS) + INSERT IGNORE INTO stat_game_day_players (app_code, stat_day, country_id, region_id, platform_code, game_id, user_id, first_played_at_ms) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + `, appcode.Normalize(event.AppCode), day, countryID, regionID, event.PlatformCode, event.GameID, event.UserID, event.OccurredAtMS) if err != nil { return err } playerDelta = affected } - if err := ensureGameDay(ctx, tx, event.AppCode, day, countryID, event.PlatformCode, event.GameID, nowMS); err != nil { + if err := ensureGameDay(ctx, tx, event.AppCode, day, countryID, regionID, event.PlatformCode, event.GameID, nowMS); err != nil { return err } if _, err := tx.ExecContext(ctx, ` @@ -501,9 +569,9 @@ func isRetryableMySQLError(err error) bool { return mysqlErr.Number == 1213 || mysqlErr.Number == 1205 } -func isDuplicateColumnError(err error) bool { +func isDuplicateSchemaError(err error) bool { var mysqlErr *mysqlerr.MySQLError - return errors.As(err, &mysqlErr) && mysqlErr.Number == 1060 + return errors.As(err, &mysqlErr) && (mysqlErr.Number == 1060 || mysqlErr.Number == 1061) } func isMissingTableError(err error) bool { @@ -511,14 +579,14 @@ func isMissingTableError(err error) bool { return errors.As(err, &mysqlErr) && mysqlErr.Number == 1146 } -func applyActive(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, userID int64, occurredAtMS int64, nowMS int64) error { - if err := ensureAppDay(ctx, tx, app, day, countryID, nowMS); err != nil { +func applyActive(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, userID int64, occurredAtMS int64, nowMS int64) error { + if err := ensureAppDay(ctx, tx, app, day, countryID, regionID, nowMS); err != nil { return err } affected, err := insertUnique(ctx, tx, ` - INSERT IGNORE INTO stat_user_day_activity (app_code, stat_day, country_id, user_id, first_active_at_ms) - VALUES (?, ?, ?, ?, ?) - `, appcode.Normalize(app), day, countryID, userID, occurredAtMS) + INSERT IGNORE INTO stat_user_day_activity (app_code, stat_day, country_id, region_id, user_id, first_active_at_ms) + VALUES (?, ?, ?, ?, ?, ?) + `, appcode.Normalize(app), day, countryID, regionID, userID, occurredAtMS) if err != nil { return err } @@ -533,27 +601,36 @@ func applyActive(ctx context.Context, tx *sql.Tx, app string, day string, countr return err } -func ensureAppDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, nowMS int64) error { +func ensureAppDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, nowMS int64) error { _, err := tx.ExecContext(ctx, ` - INSERT IGNORE INTO stat_app_day_country (app_code, stat_day, country_id, updated_at_ms) - VALUES (?, ?, ?, ?) - `, appcode.Normalize(app), day, countryID, nowMS) - return err -} - -func ensureGameDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, platformCode string, gameID string, nowMS int64) error { - _, err := tx.ExecContext(ctx, ` - INSERT IGNORE INTO stat_game_day_country (app_code, stat_day, country_id, platform_code, game_id, updated_at_ms) - VALUES (?, ?, ?, ?, ?, ?) - `, appcode.Normalize(app), day, countryID, platformCode, gameID, nowMS) - return err -} - -func ensureLuckyGiftPoolDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, poolID string, nowMS int64) error { - _, err := tx.ExecContext(ctx, ` - INSERT IGNORE INTO stat_lucky_gift_pool_day_country (app_code, stat_day, country_id, pool_id, updated_at_ms) + INSERT INTO stat_app_day_country (app_code, stat_day, country_id, region_id, updated_at_ms) VALUES (?, ?, ?, ?, ?) - `, appcode.Normalize(app), day, countryID, strings.TrimSpace(poolID), nowMS) + ON DUPLICATE KEY UPDATE + region_id = IF(region_id = 0 AND VALUES(region_id) > 0, VALUES(region_id), region_id), + updated_at_ms = VALUES(updated_at_ms) + `, appcode.Normalize(app), day, countryID, regionID, nowMS) + return err +} + +func ensureGameDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, platformCode string, gameID string, nowMS int64) error { + _, err := tx.ExecContext(ctx, ` + INSERT INTO stat_game_day_country (app_code, stat_day, country_id, region_id, platform_code, game_id, updated_at_ms) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE + region_id = IF(region_id = 0 AND VALUES(region_id) > 0, VALUES(region_id), region_id), + updated_at_ms = VALUES(updated_at_ms) + `, appcode.Normalize(app), day, countryID, regionID, platformCode, gameID, nowMS) + return err +} + +func ensureLuckyGiftPoolDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, poolID string, nowMS int64) error { + _, err := tx.ExecContext(ctx, ` + INSERT INTO stat_lucky_gift_pool_day_country (app_code, stat_day, country_id, region_id, pool_id, updated_at_ms) + VALUES (?, ?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE + region_id = IF(region_id = 0 AND VALUES(region_id) > 0, VALUES(region_id), region_id), + updated_at_ms = VALUES(updated_at_ms) + `, appcode.Normalize(app), day, countryID, regionID, strings.TrimSpace(poolID), nowMS) return err } @@ -580,6 +657,10 @@ func normalizeID(id int64) int64 { return id } +func normalizeDimension(countryID int64, regionID int64) (int64, int64) { + return normalizeID(countryID), normalizeID(regionID) +} + func DecodeJSON(payload string) map[string]any { decoded := map[string]any{} _ = json.Unmarshal([]byte(payload), &decoded) diff --git a/services/user-service/internal/storage/mysql/user/repository.go b/services/user-service/internal/storage/mysql/user/repository.go index 2439549f..a245b88a 100644 --- a/services/user-service/internal/storage/mysql/user/repository.go +++ b/services/user-service/internal/storage/mysql/user/repository.go @@ -158,6 +158,7 @@ func insertUserRegisteredOutbox(ctx context.Context, tx *sql.Tx, user userdomain "user_id": user.UserID, "default_display_user_id": user.DefaultDisplayUserID, "country": user.Country, + "country_id": user.CountryID, "country_by_ip": user.CountryByIP, "region_id": user.RegionID, "invite_code": user.InviteCode,