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

feat(statistics): add stat_tz dashboard aggregation

Browse Source
This commit is contained in:
zhx 2026-06-25 14:31:15 +08:00
parent d57419c99e
commit 13f07d800f
11 changed files with 1441 additions and 635 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
server/admin/internal/modules/dashboard/handler.go
View File

@ -31,6 +31,7 @@ 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"),
StatTZ: c.Query("stat_tz"),
StartMS: parseInt64(c.Query("start_ms")),
EndMS: parseInt64(c.Query("end_ms")),
SeriesStartMS: parseInt64(c.Query("series_start_ms")),
@ -48,6 +49,7 @@ func (h *Handler) StatisticsOverview(c *gin.Context) {
func (h *Handler) SelfGameStatisticsOverview(c *gin.Context) {
overview, err := h.service.SelfGameStatisticsOverview(c.Request.Context(), SelfGameStatisticsQuery{
AppCode: c.DefaultQuery("app_code", "lalu"),
StatTZ: c.Query("stat_tz"),
RequestID: middleware.CurrentRequestID(c),
StartMS: parseInt64(c.Query("start_ms")),
EndMS: parseInt64(c.Query("end_ms")),

8
server/admin/internal/modules/dashboard/service.go
View File

@ -23,6 +23,7 @@ type DashboardService struct {
type StatisticsQuery struct {
AppCode string
StatTZ string
StartMS int64
EndMS int64
SeriesStartMS int64
@ -33,6 +34,7 @@ type StatisticsQuery struct {
type SelfGameStatisticsQuery struct {
AppCode string
StatTZ string
RequestID string
StartMS int64
EndMS int64
@ -57,6 +59,9 @@ func (s *DashboardService) Overview() (*repository.DashboardOverview, error) {
func (s *DashboardService) StatisticsOverview(ctx context.Context, query StatisticsQuery) (map[string]any, error) {
values := url.Values{}
values.Set("app_code", query.AppCode)
if statTZ := strings.TrimSpace(query.StatTZ); statTZ != "" {
values.Set("stat_tz", statTZ)
}
if query.StartMS > 0 {
values.Set("start_ms", strconv.FormatInt(query.StartMS, 10))
}
@ -81,6 +86,9 @@ func (s *DashboardService) StatisticsOverview(ctx context.Context, query Statist
func (s *DashboardService) SelfGameStatisticsOverview(ctx context.Context, query SelfGameStatisticsQuery) (map[string]any, error) {
values := url.Values{}
values.Set("app_code", query.AppCode)
if statTZ := strings.TrimSpace(query.StatTZ); statTZ != "" {
values.Set("stat_tz", statTZ)
}
if query.StartMS > 0 {
values.Set("start_ms", strconv.FormatInt(query.StartMS, 10))
}

124
services/statistics-service/cmd/backfill-last7/main.go
View File

@ -33,6 +33,7 @@ type appDayAgg struct {
rechargeUSDMinor int64
newUserRechargeUSDMinor int64
coinSellerRechargeUSDMinor int64
mifaPayRechargeUSDMinor int64
googleRechargeUSDMinor int64
coinSellerTransferCoin int64
}
@ -90,6 +91,10 @@ func main() {
if err != nil {
log.Fatalf("load timezone: %v", err)
}
statTZ, err := normalizeBackfillStatTZ(loc.String())
if err != nil {
log.Fatal(err)
}
startDay, endDay, windowStartMS, windowEndMS, err := resolveWindow(*startFlag, *endFlag, loc)
if err != nil {
log.Fatal(err)
@ -116,7 +121,7 @@ func main() {
if err := collectRegistrationUsers(ctx, userDB, state, app, windowStartMS, windowEndMS); err != nil {
log.Fatalf("collect registrations: %v", err)
}
if err := collectActiveUserIDs(ctx, statsDB, state, app, startDay, endDay); err != nil {
if err := collectActiveUserIDs(ctx, statsDB, state, app, statTZ, startDay, endDay); err != nil {
log.Fatalf("collect active ids: %v", err)
}
if err := collectRechargeUserIDs(ctx, walletDB, state, app, windowStartMS, windowEndMS); err != nil {
@ -128,7 +133,7 @@ func main() {
if err := rebuildRegistrations(ctx, userDB, state, app, windowStartMS, windowEndMS, loc); err != nil {
log.Fatalf("rebuild registrations: %v", err)
}
if err := rebuildActive(ctx, statsDB, state, app, startDay, endDay); err != nil {
if err := rebuildActive(ctx, statsDB, state, app, statTZ, startDay, endDay); err != nil {
log.Fatalf("rebuild active: %v", err)
}
if err := rebuildWalletRecharge(ctx, walletDB, state, app, windowStartMS, windowEndMS, loc); err != nil {
@ -138,7 +143,7 @@ func main() {
log.Fatalf("rebuild coin seller stock: %v", err)
}
if err := stageAndMaybeApply(ctx, statsDB, state, app, startDay, endDay, *applyFlag); err != nil {
if err := stageAndMaybeApply(ctx, statsDB, state, app, statTZ, startDay, endDay, *applyFlag); err != nil {
log.Fatalf("stage/apply: %v", err)
}
mode := "dry-run"
@ -146,7 +151,18 @@ func main() {
mode = "applied"
}
fmt.Printf("%s app=%s timezone=%s window=%s..%s app_day_rows=%d registrations=%d active=%d payers=%d\n",
mode, app, loc.String(), startDay, endDay, len(state.appDays), len(state.registrations), len(state.activeByKey), len(state.payersByKey))
mode, app, statTZ, startDay, endDay, len(state.appDays), len(state.registrations), len(state.activeByKey), len(state.payersByKey))
}
func normalizeBackfillStatTZ(value string) (string, error) {
switch strings.TrimSpace(value) {
case "UTC":
return "UTC", nil
case "Asia/Shanghai":
return "Asia/Shanghai", nil
default:
return "", fmt.Errorf("unsupported stat_tz %q; only UTC and Asia/Shanghai are queryable", value)
}
}
func envDefault(key string, fallback string) string {
@ -250,12 +266,12 @@ func collectRegistrationUsers(ctx context.Context, db *sql.DB, state *rebuildSta
return rows.Err()
}
func collectActiveUserIDs(ctx context.Context, db *sql.DB, state *rebuildState, app, startDay, endDay string) error {
func collectActiveUserIDs(ctx context.Context, db *sql.DB, state *rebuildState, app, statTZ, startDay, endDay string) error {
rows, err := db.QueryContext(ctx, `
SELECT DISTINCT user_id
FROM stat_user_day_activity
WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
app, startDay, endDay)
WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`,
app, statTZ, startDay, endDay)
if err != nil {
return err
}
@ -370,14 +386,14 @@ func rebuildRegistrations(ctx context.Context, db *sql.DB, state *rebuildState,
return rows.Err()
}
func rebuildActive(ctx context.Context, db *sql.DB, state *rebuildState, app, startDay, endDay string) error {
// 房间/游戏活跃已经被实时消费进 stat_user_day_activity；离线回填在注册活跃基础上重刷这张表的国家/区域快照。
func rebuildActive(ctx context.Context, db *sql.DB, state *rebuildState, app, statTZ, startDay, endDay string) error {
// 房间/游戏活跃已经被实时消费进对应 stat_tz 的 stat_user_day_activity；回填只读取目标时区，避免用 UTC 活跃去污染中国自然日。
rows, err := db.QueryContext(ctx, `
SELECT DATE_FORMAT(stat_day, '%Y-%m-%d'), user_id, MIN(first_active_at_ms)
FROM stat_user_day_activity
WHERE app_code = ? AND stat_day BETWEEN ? AND ?
WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?
GROUP BY stat_day, user_id`,
app, startDay, endDay)
app, statTZ, startDay, endDay)
if err != nil {
return err
}
@ -426,6 +442,9 @@ func rebuildWalletRecharge(ctx context.Context, db *sql.DB, state *rebuildState,
case "google_play_recharge":
agg.rechargeUSDMinor += usdMinor
agg.googleRechargeUSDMinor += usdMinor
case "mifapay", "v5pay":
agg.rechargeUSDMinor += usdMinor
agg.mifaPayRechargeUSDMinor += usdMinor
default:
if usdMinor > 0 {
agg.rechargeUSDMinor += usdMinor
@ -507,7 +526,7 @@ func userDayKey(day string, userID int64) string {
return fmt.Sprintf("%s:%d", day, userID)
}
func stageAndMaybeApply(ctx context.Context, db *sql.DB, state *rebuildState, app, startDay, endDay string, apply bool) error {
func stageAndMaybeApply(ctx context.Context, db *sql.DB, state *rebuildState, app, statTZ, startDay, endDay string, apply bool) error {
conn, err := db.Conn(ctx)
if err != nil {
return err
@ -521,7 +540,7 @@ func stageAndMaybeApply(ctx context.Context, db *sql.DB, state *rebuildState, ap
if err := createTempTables(ctx, tx); err != nil {
return err
}
if err := insertStageRows(ctx, tx, state, app); err != nil {
if err := insertStageRows(ctx, tx, state, app, statTZ); err != nil {
return err
}
if err := validateStage(ctx, tx); err != nil {
@ -531,60 +550,61 @@ func stageAndMaybeApply(ctx context.Context, db *sql.DB, state *rebuildState, ap
return tx.Rollback()
}
nowMS := time.Now().UTC().UnixMilli()
// 本 job 只覆盖能从当前源表确定的近 7 天列；礼物、游戏、MifaPay 等不在本次源表范围内的指标保留原值，避免修国家维度时误删其他业务数据。
// 本 job 只覆盖能从当前源表确定的列；礼物和游戏等不在本次源表范围内的指标保留原值，且只替换目标 stat_tz，避免重跑中国日时覆盖 UTC 读模型。
if _, err := tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET new_users = 0, active_users = 0, paid_users = 0,
recharge_usd_minor = 0, new_user_recharge_usd_minor = 0,
coin_seller_recharge_usd_minor = 0,
google_recharge_usd_minor = 0, coin_seller_transfer_coin = 0,
mifapay_recharge_usd_minor = 0, google_recharge_usd_minor = 0, coin_seller_transfer_coin = 0,
updated_at_ms = ?
WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
nowMS, app, startDay, endDay); err != nil {
WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`,
nowMS, app, statTZ, startDay, endDay); err != nil {
return err
}
for _, statement := range []string{
`DELETE FROM stat_user_registration WHERE app_code = ? AND registered_day BETWEEN ? AND ?`,
`DELETE FROM stat_user_day_activity WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
`DELETE FROM stat_recharge_day_payers WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
`DELETE FROM stat_user_registration WHERE app_code = ? AND stat_tz = ? AND registered_day BETWEEN ? AND ?`,
`DELETE FROM stat_user_day_activity WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`,
`DELETE FROM stat_recharge_day_payers WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`,
} {
if _, err := tx.ExecContext(ctx, statement, app, startDay, endDay); err != nil {
if _, err := tx.ExecContext(ctx, statement, app, statTZ, startDay, endDay); err != nil {
return err
}
}
if _, err := tx.ExecContext(ctx, `
INSERT INTO stat_app_day_country (
app_code, stat_day, country_id, region_id, new_users, active_users, paid_users,
app_code, stat_tz, stat_day, country_id, region_id, new_users, active_users, paid_users,
recharge_usd_minor, new_user_recharge_usd_minor, coin_seller_recharge_usd_minor,
google_recharge_usd_minor, coin_seller_transfer_coin, updated_at_ms
mifapay_recharge_usd_minor, google_recharge_usd_minor, coin_seller_transfer_coin, updated_at_ms
)
SELECT app_code, stat_day, country_id, region_id, new_users, active_users, paid_users,
SELECT app_code, stat_tz, stat_day, country_id, region_id, new_users, active_users, paid_users,
recharge_usd_minor, new_user_recharge_usd_minor, coin_seller_recharge_usd_minor,
google_recharge_usd_minor, coin_seller_transfer_coin, ?
mifapay_recharge_usd_minor, google_recharge_usd_minor, coin_seller_transfer_coin, ?
FROM tmp_stat_backfill_app_day
ON DUPLICATE KEY UPDATE
new_users = VALUES(new_users), active_users = VALUES(active_users), paid_users = VALUES(paid_users),
recharge_usd_minor = VALUES(recharge_usd_minor),
new_user_recharge_usd_minor = VALUES(new_user_recharge_usd_minor),
coin_seller_recharge_usd_minor = VALUES(coin_seller_recharge_usd_minor),
mifapay_recharge_usd_minor = VALUES(mifapay_recharge_usd_minor),
google_recharge_usd_minor = VALUES(google_recharge_usd_minor),
coin_seller_transfer_coin = VALUES(coin_seller_transfer_coin),
updated_at_ms = VALUES(updated_at_ms)`, nowMS); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `
INSERT INTO stat_user_registration (app_code, user_id, registered_day, country_id, region_id, registered_at_ms)
SELECT app_code, user_id, registered_day, country_id, region_id, registered_at_ms FROM tmp_stat_backfill_registration`); err != nil {
INSERT INTO stat_user_registration (app_code, stat_tz, user_id, registered_day, country_id, region_id, registered_at_ms)
SELECT app_code, stat_tz, user_id, registered_day, country_id, region_id, registered_at_ms FROM tmp_stat_backfill_registration`); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `
INSERT INTO stat_user_day_activity (app_code, stat_day, country_id, region_id, user_id, first_active_at_ms)
SELECT app_code, stat_day, country_id, region_id, user_id, first_active_at_ms FROM tmp_stat_backfill_active`); err != nil {
INSERT INTO stat_user_day_activity (app_code, stat_tz, stat_day, country_id, region_id, user_id, first_active_at_ms)
SELECT app_code, stat_tz, stat_day, country_id, region_id, user_id, first_active_at_ms FROM tmp_stat_backfill_active`); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `
INSERT INTO stat_recharge_day_payers (app_code, stat_day, country_id, region_id, user_id, first_paid_at_ms)
SELECT app_code, stat_day, country_id, region_id, user_id, first_paid_at_ms FROM tmp_stat_backfill_payers`); err != nil {
INSERT INTO stat_recharge_day_payers (app_code, stat_tz, stat_day, country_id, region_id, user_id, first_paid_at_ms)
SELECT app_code, stat_tz, stat_day, country_id, region_id, user_id, first_paid_at_ms FROM tmp_stat_backfill_payers`); err != nil {
return err
}
return tx.Commit()
@ -593,27 +613,27 @@ func stageAndMaybeApply(ctx context.Context, db *sql.DB, state *rebuildState, ap
func createTempTables(ctx context.Context, tx *sql.Tx) error {
statements := []string{
`CREATE TEMPORARY TABLE tmp_stat_backfill_app_day (
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL, region_id BIGINT NOT NULL,
app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL, region_id BIGINT NOT NULL,
new_users BIGINT NOT NULL, active_users BIGINT NOT NULL, paid_users BIGINT NOT NULL,
recharge_usd_minor BIGINT NOT NULL, new_user_recharge_usd_minor BIGINT NOT NULL,
coin_seller_recharge_usd_minor BIGINT NOT NULL, google_recharge_usd_minor BIGINT NOT NULL,
coin_seller_recharge_usd_minor BIGINT NOT NULL, mifapay_recharge_usd_minor BIGINT NOT NULL, google_recharge_usd_minor BIGINT NOT NULL,
coin_seller_transfer_coin BIGINT NOT NULL,
PRIMARY KEY(app_code, stat_day, country_id, region_id)
PRIMARY KEY(app_code, stat_tz, stat_day, country_id, region_id)
)`,
`CREATE TEMPORARY TABLE tmp_stat_backfill_registration (
app_code VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, registered_day DATE NOT NULL,
app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL, user_id BIGINT NOT NULL, registered_day DATE NOT NULL,
country_id BIGINT NOT NULL, region_id BIGINT NOT NULL, registered_at_ms BIGINT NOT NULL,
PRIMARY KEY(app_code, user_id)
PRIMARY KEY(app_code, stat_tz, user_id)
)`,
`CREATE TEMPORARY TABLE tmp_stat_backfill_active (
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL,
app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL,
region_id BIGINT NOT NULL, user_id BIGINT NOT NULL, first_active_at_ms BIGINT NOT NULL,
PRIMARY KEY(app_code, stat_day, user_id)
PRIMARY KEY(app_code, stat_tz, stat_day, user_id)
)`,
`CREATE TEMPORARY TABLE tmp_stat_backfill_payers (
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL,
app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL,
region_id BIGINT NOT NULL, user_id BIGINT NOT NULL, first_paid_at_ms BIGINT NOT NULL,
PRIMARY KEY(app_code, stat_day, user_id)
PRIMARY KEY(app_code, stat_tz, stat_day, user_id)
)`,
}
for _, statement := range statements {
@ -624,38 +644,38 @@ func createTempTables(ctx context.Context, tx *sql.Tx) error {
return nil
}
func insertStageRows(ctx context.Context, tx *sql.Tx, state *rebuildState, app string) error {
func insertStageRows(ctx context.Context, tx *sql.Tx, state *rebuildState, app string, statTZ string) error {
for key, agg := range state.appDays {
if _, err := tx.ExecContext(ctx, `
INSERT INTO tmp_stat_backfill_app_day (
app_code, stat_day, country_id, region_id, new_users, active_users, paid_users,
app_code, stat_tz, stat_day, country_id, region_id, new_users, active_users, paid_users,
recharge_usd_minor, new_user_recharge_usd_minor, coin_seller_recharge_usd_minor,
google_recharge_usd_minor, coin_seller_transfer_coin
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
app, key.day, key.countryID, key.regionID, agg.newUsers, agg.activeUsers, agg.paidUsers,
mifapay_recharge_usd_minor, google_recharge_usd_minor, coin_seller_transfer_coin
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
app, statTZ, key.day, key.countryID, key.regionID, agg.newUsers, agg.activeUsers, agg.paidUsers,
agg.rechargeUSDMinor, agg.newUserRechargeUSDMinor, agg.coinSellerRechargeUSDMinor,
agg.googleRechargeUSDMinor, agg.coinSellerTransferCoin); err != nil {
agg.mifaPayRechargeUSDMinor, agg.googleRechargeUSDMinor, agg.coinSellerTransferCoin); err != nil {
return err
}
}
for _, row := range state.registrations {
if _, err := tx.ExecContext(ctx, `
INSERT INTO tmp_stat_backfill_registration (app_code, user_id, registered_day, country_id, region_id, registered_at_ms)
VALUES (?, ?, ?, ?, ?, ?)`, app, row.userID, row.registeredDay, row.dim.countryID, row.dim.regionID, row.registeredAtMS); err != nil {
INSERT INTO tmp_stat_backfill_registration (app_code, stat_tz, user_id, registered_day, country_id, region_id, registered_at_ms)
VALUES (?, ?, ?, ?, ?, ?, ?)`, app, statTZ, row.userID, row.registeredDay, row.dim.countryID, row.dim.regionID, row.registeredAtMS); err != nil {
return err
}
}
for _, row := range state.activeByKey {
if _, err := tx.ExecContext(ctx, `
INSERT INTO tmp_stat_backfill_active (app_code, stat_day, country_id, region_id, user_id, first_active_at_ms)
VALUES (?, ?, ?, ?, ?, ?)`, app, row.day, row.dim.countryID, row.dim.regionID, row.userID, row.firstMS); err != nil {
INSERT INTO tmp_stat_backfill_active (app_code, stat_tz, stat_day, country_id, region_id, user_id, first_active_at_ms)
VALUES (?, ?, ?, ?, ?, ?, ?)`, app, statTZ, row.day, row.dim.countryID, row.dim.regionID, row.userID, row.firstMS); err != nil {
return err
}
}
for _, row := range state.payersByKey {
if _, err := tx.ExecContext(ctx, `
INSERT INTO tmp_stat_backfill_payers (app_code, stat_day, country_id, region_id, user_id, first_paid_at_ms)
VALUES (?, ?, ?, ?, ?, ?)`, app, row.day, row.dim.countryID, row.dim.regionID, row.userID, row.firstMS); err != nil {
INSERT INTO tmp_stat_backfill_payers (app_code, stat_tz, stat_day, country_id, region_id, user_id, first_paid_at_ms)
VALUES (?, ?, ?, ?, ?, ?, ?)`, app, statTZ, row.day, row.dim.countryID, row.dim.regionID, row.userID, row.firstMS); err != nil {
return err
}
}

414
services/statistics-service/cmd/replay-stat-tz/main.go Normal file
View File

@ -0,0 +1,414 @@
package main
import (
"context"
"crypto/sha256"
"database/sql"
"encoding/hex"
"flag"
"fmt"
"log"
"os"
"strings"
"time"
"google.golang.org/protobuf/proto"
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
"hyapp/pkg/gamemq"
"hyapp/pkg/roommq"
"hyapp/pkg/usermq"
"hyapp/pkg/walletmq"
statapp "hyapp/services/statistics-service/internal/app"
mysqlstorage "hyapp/services/statistics-service/internal/storage/mysql"
_ "github.com/go-sql-driver/mysql"
)
const replayEventIDMaxLen = 160
type replayWindow struct {
startDay string
endDay string
startMS int64
endMS int64
}
type replaySummary struct {
userRows int64
walletRows int64
roomRows int64
gameRows int64
}
func main() {
var (
appCode = flag.String("app", envDefault("APP_CODE", "lalu"), "app_code to replay")
statTZ = flag.String("timezone", envDefault("STAT_TZ", mysqlstorage.StatTZAsiaShanghai), "stat_tz to rebuild: UTC or Asia/Shanghai")
startDay = flag.String("start-day", "", "inclusive stat day in selected timezone, yyyy-mm-dd")
endDay = flag.String("end-day", "", "inclusive stat day in selected timezone, yyyy-mm-dd")
apply = flag.Bool("apply", false, "apply mutations; dry-run only counts source rows by default")
statisticsDSN = flag.String("statistics-dsn", firstNonEmptyEnv("STATISTICS_MYSQL_DSN", "MYSQL_DSN"), "statistics mysql dsn")
userDSN = flag.String("user-dsn", firstNonEmptyEnv("USER_MYSQL_DSN", "MYSQL_USER_DSN"), "user-service mysql dsn")
walletDSN = flag.String("wallet-dsn", firstNonEmptyEnv("WALLET_MYSQL_DSN", "MYSQL_WALLET_DSN"), "wallet-service mysql dsn")
roomDSN = flag.String("room-dsn", firstNonEmptyEnv("ROOM_MYSQL_DSN", "MYSQL_ROOM_DSN"), "room-service mysql dsn")
gameDSN = flag.String("game-dsn", firstNonEmptyEnv("GAME_MYSQL_DSN", "MYSQL_GAME_DSN"), "game-service mysql dsn")
)
flag.Parse()
statTZValue, loc, err := normalizeReplayStatTZ(*statTZ)
if err != nil {
log.Fatal(err)
}
window, err := resolveReplayWindow(*startDay, *endDay, loc)
if err != nil {
log.Fatal(err)
}
if *apply && strings.TrimSpace(*statisticsDSN) == "" {
log.Fatal("statistics-dsn is required when apply is true")
}
if *apply && (strings.TrimSpace(*userDSN) == "" || strings.TrimSpace(*walletDSN) == "" || strings.TrimSpace(*roomDSN) == "" || strings.TrimSpace(*gameDSN) == "") {
log.Fatal("apply requires user-dsn, wallet-dsn, room-dsn and game-dsn so the rebuilt stat_tz window is complete for owner outbox facts")
}
ctx := context.Background()
dbs := map[string]*sql.DB{}
for name, dsn := range map[string]string{"user": *userDSN, "wallet": *walletDSN, "room": *roomDSN, "game": *gameDSN} {
if strings.TrimSpace(dsn) == "" {
continue
}
db, err := openDB(ctx, dsn)
if err != nil {
log.Fatalf("open %s db: %v", name, err)
}
defer db.Close()
dbs[name] = db
}
summary, err := countSources(ctx, dbs, *appCode, window)
if err != nil {
log.Fatalf("count sources: %v", err)
}
if !*apply {
fmt.Printf("dry-run app=%s stat_tz=%s days=%s..%s source_rows user=%d wallet=%d room=%d game=%d\n",
*appCode, statTZValue, window.startDay, window.endDay, summary.userRows, summary.walletRows, summary.roomRows, summary.gameRows)
return
}
statsDB, err := openDB(ctx, *statisticsDSN)
if err != nil {
log.Fatalf("open statistics db: %v", err)
}
defer statsDB.Close()
repo, err := mysqlstorage.Open(ctx, *statisticsDSN)
if err != nil {
log.Fatalf("open statistics repository: %v", err)
}
defer repo.Close()
if err := resetReplayTarget(ctx, statsDB, *appCode, statTZValue, window); err != nil {
log.Fatalf("reset target: %v", err)
}
scopeCtx := mysqlstorage.WithStatTZScope(ctx, statTZValue)
if err := replayUser(ctx, scopeCtx, repo, dbs["user"], *appCode, statTZValue, window); err != nil {
log.Fatalf("replay user outbox: %v", err)
}
if err := replayWallet(ctx, scopeCtx, repo, dbs["wallet"], *appCode, statTZValue, window); err != nil {
log.Fatalf("replay wallet outbox: %v", err)
}
if err := replayRoom(ctx, scopeCtx, repo, dbs["room"], *appCode, statTZValue, window); err != nil {
log.Fatalf("replay room outbox: %v", err)
}
if err := replayGame(ctx, scopeCtx, repo, dbs["game"], *appCode, statTZValue, window); err != nil {
log.Fatalf("replay game outbox: %v", err)
}
fmt.Printf("applied app=%s stat_tz=%s days=%s..%s source_rows user=%d wallet=%d room=%d game=%d\n",
*appCode, statTZValue, window.startDay, window.endDay, summary.userRows, summary.walletRows, summary.roomRows, summary.gameRows)
}
func normalizeReplayStatTZ(value string) (string, *time.Location, error) {
switch strings.TrimSpace(value) {
case mysqlstorage.StatTZUTC:
return mysqlstorage.StatTZUTC, time.UTC, nil
case mysqlstorage.StatTZAsiaShanghai:
return mysqlstorage.StatTZAsiaShanghai, time.FixedZone(mysqlstorage.StatTZAsiaShanghai, 8*60*60), nil
default:
return "", nil, fmt.Errorf("unsupported stat_tz %q; only UTC and Asia/Shanghai are supported", value)
}
}
func resolveReplayWindow(startDay string, endDay string, loc *time.Location) (replayWindow, error) {
now := time.Now().In(loc)
if strings.TrimSpace(startDay) == "" || strings.TrimSpace(endDay) == "" {
// 默认重跑“最近 30 个已完成自然日”，避开当天实时消费者仍在写入造成 replay 后缀和原 event_id 同时入账。
yesterday := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, loc).AddDate(0, 0, -1)
start := yesterday.AddDate(0, 0, -29)
startDay = start.Format("2006-01-02")
endDay = yesterday.Format("2006-01-02")
}
start, err := time.ParseInLocation("2006-01-02", startDay, loc)
if err != nil {
return replayWindow{}, fmt.Errorf("parse start-day: %w", err)
}
end, err := time.ParseInLocation("2006-01-02", endDay, loc)
if err != nil {
return replayWindow{}, fmt.Errorf("parse end-day: %w", err)
}
if end.Before(start) {
return replayWindow{}, fmt.Errorf("end-day %s is before start-day %s", endDay, startDay)
}
return replayWindow{
startDay: start.Format("2006-01-02"),
endDay: end.Format("2006-01-02"),
startMS: start.UnixMilli(),
endMS: end.AddDate(0, 0, 1).UnixMilli(),
}, nil
}
func openDB(ctx context.Context, dsn string) (*sql.DB, error) {
db, err := sql.Open("mysql", strings.TrimSpace(dsn))
if err != nil {
return nil, err
}
if err := db.PingContext(ctx); err != nil {
_ = db.Close()
return nil, err
}
return db, nil
}
func countSources(ctx context.Context, dbs map[string]*sql.DB, app string, window replayWindow) (replaySummary, error) {
var out replaySummary
var err error
if db := dbs["user"]; db != nil {
out.userRows, err = countOutboxRows(ctx, db, "user_outbox", app, []string{"UserRegistered"}, window)
if err != nil {
return replaySummary{}, err
}
}
if db := dbs["wallet"]; db != nil {
out.walletRows, err = countOutboxRows(ctx, db, "wallet_outbox", app, []string{"WalletRechargeRecorded", "WalletCoinSellerStockPurchased", "WalletLuckyGiftRewardCredited"}, window)
if err != nil {
return replaySummary{}, err
}
}
if db := dbs["room"]; db != nil {
out.roomRows, err = countOutboxRows(ctx, db, "room_outbox", app, []string{"RoomGiftSent", "RoomUserJoined"}, window)
if err != nil {
return replaySummary{}, err
}
}
if db := dbs["game"]; db != nil {
out.gameRows, err = countOutboxRows(ctx, db, "game_outbox", app, gameReplayEventTypes(), window)
if err != nil {
return replaySummary{}, err
}
}
return out, nil
}
func countOutboxRows(ctx context.Context, db *sql.DB, table string, app string, eventTypes []string, window replayWindow) (int64, error) {
query, args := outboxWindowQuery("COUNT(*)", table, app, eventTypes, window, "")
row := db.QueryRowContext(ctx, query, args...)
var count int64
return count, row.Scan(&count)
}
func resetReplayTarget(ctx context.Context, db *sql.DB, app string, statTZ string, window replayWindow) error {
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
// 目标表只删除当前 stat_tz 的可重建 owner-outbox 聚合；H5 上报没有 owner outbox，不在这里删除，避免抹掉部署后实时进入的 H5 事件。
for _, spec := range []struct {
table string
day string
}{
{"stat_app_day_country", "stat_day"},
{"stat_user_day_activity", "stat_day"},
{"stat_user_registration", "registered_day"},
{"stat_recharge_day_payers", "stat_day"},
{"stat_lucky_gift_day_payers", "stat_day"},
{"stat_lucky_gift_pool_day_country", "stat_day"},
{"stat_real_room_robot_gift_day_rooms", "stat_day"},
{"stat_game_day_country", "stat_day"},
{"stat_game_day_players", "stat_day"},
{"stat_self_game_match_day", "stat_day"},
{"stat_self_game_pool_day", "stat_day"},
{"stat_self_game_participant_day", "stat_day"},
{"stat_self_game_user_day", "stat_day"},
} {
if _, err := tx.ExecContext(ctx, "DELETE FROM "+spec.table+" WHERE app_code = ? AND stat_tz = ? AND "+spec.day+" BETWEEN ? AND ?", app, statTZ, window.startDay, window.endDay); err != nil {
return err
}
}
token := replayToken(statTZ)
for _, source := range []string{mysqlstorage.SourceUser, mysqlstorage.SourceWallet, mysqlstorage.SourceRoom, mysqlstorage.SourceGame} {
if _, err := tx.ExecContext(ctx, `
DELETE FROM statistics_event_consumption
WHERE app_code = ? AND source = ? AND (event_id LIKE ? OR event_id LIKE ?)`,
app, source, "%:stat_tz:"+token, "replay:"+token+":%"); err != nil {
return err
}
}
return tx.Commit()
}
func replayUser(ctx context.Context, scopeCtx context.Context, repo *mysqlstorage.Repository, db *sql.DB, app string, statTZ string, window replayWindow) error {
rows, err := db.QueryContext(ctx, `
SELECT app_code, event_id, event_type, aggregate_type, aggregate_id, CAST(payload_json AS CHAR), created_at_ms
FROM user_outbox
WHERE app_code = ? AND event_type = 'UserRegistered' AND created_at_ms >= ? AND created_at_ms < ?
ORDER BY created_at_ms ASC, event_id ASC`, app, window.startMS, window.endMS)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var msg usermq.UserOutboxMessage
if err := rows.Scan(&msg.AppCode, &msg.EventID, &msg.EventType, &msg.AggregateType, &msg.AggregateID, &msg.PayloadJSON, &msg.OccurredAtMS); err != nil {
return err
}
msg.EventID = replayEventID(msg.EventID, statTZ)
body, err := usermq.EncodeUserOutboxMessage(msg)
if err != nil {
return err
}
if err := statapp.ConsumeUserOutboxBody(scopeCtx, repo, body); err != nil {
return err
}
}
return rows.Err()
}
func replayWallet(ctx context.Context, scopeCtx context.Context, repo *mysqlstorage.Repository, db *sql.DB, app string, statTZ string, window replayWindow) error {
query, args := outboxWindowQuery(`
app_code, event_id, event_type, transaction_id, command_id, user_id, asset_type,
available_delta, frozen_delta, CAST(payload AS CHAR), created_at_ms`, "wallet_outbox", app,
[]string{"WalletRechargeRecorded", "WalletCoinSellerStockPurchased", "WalletLuckyGiftRewardCredited"}, window, "ORDER BY created_at_ms ASC, event_id ASC")
rows, err := db.QueryContext(ctx, query, args...)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var msg walletmq.WalletOutboxMessage
if err := rows.Scan(&msg.AppCode, &msg.EventID, &msg.EventType, &msg.TransactionID, &msg.CommandID, &msg.UserID, &msg.AssetType, &msg.AvailableDelta, &msg.FrozenDelta, &msg.PayloadJSON, &msg.OccurredAtMS); err != nil {
return err
}
msg.EventID = replayEventID(msg.EventID, statTZ)
body, err := walletmq.EncodeWalletOutboxMessage(msg)
if err != nil {
return err
}
if err := statapp.ConsumeWalletOutboxBody(scopeCtx, repo, body); err != nil {
return err
}
}
return rows.Err()
}
func replayRoom(ctx context.Context, scopeCtx context.Context, repo *mysqlstorage.Repository, db *sql.DB, app string, statTZ string, window replayWindow) error {
query, args := outboxWindowQuery("envelope", "room_outbox", app, []string{"RoomGiftSent", "RoomUserJoined"}, window, "ORDER BY created_at_ms ASC, event_id ASC")
rows, err := db.QueryContext(ctx, query, args...)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var raw []byte
if err := rows.Scan(&raw); err != nil {
return err
}
var envelope roomeventsv1.EventEnvelope
if err := proto.Unmarshal(raw, &envelope); err != nil {
return err
}
envelope.EventId = replayEventID(envelope.GetEventId(), statTZ)
body, err := roommq.EncodeRoomOutboxMessage(&envelope)
if err != nil {
return err
}
if err := statapp.ConsumeRoomOutboxBody(scopeCtx, repo, body); err != nil {
return err
}
}
return rows.Err()
}
func replayGame(ctx context.Context, scopeCtx context.Context, repo *mysqlstorage.Repository, db *sql.DB, app string, statTZ string, window replayWindow) error {
query, args := outboxWindowQuery(`
app_code, event_id, event_type, order_id, user_id, platform_code, game_id,
op_type, coin_amount, CAST(payload_json AS CHAR), created_at_ms`, "game_outbox", app, gameReplayEventTypes(), window, "ORDER BY created_at_ms ASC, event_id ASC")
rows, err := db.QueryContext(ctx, query, args...)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var msg gamemq.GameOutboxMessage
if err := rows.Scan(&msg.AppCode, &msg.EventID, &msg.EventType, &msg.OrderID, &msg.UserID, &msg.PlatformCode, &msg.GameID, &msg.OpType, &msg.CoinAmount, &msg.PayloadJSON, &msg.OccurredAtMS); err != nil {
return err
}
msg.EventID = replayEventID(msg.EventID, statTZ)
body, err := gamemq.EncodeGameOutboxMessage(msg)
if err != nil {
return err
}
if err := statapp.ConsumeGameOutboxBody(scopeCtx, repo, body); err != nil {
return err
}
}
return rows.Err()
}
func outboxWindowQuery(selectExpr string, table string, app string, eventTypes []string, window replayWindow, suffix string) (string, []any) {
placeholders := strings.TrimRight(strings.Repeat("?,", len(eventTypes)), ",")
args := []any{app}
for _, eventType := range eventTypes {
args = append(args, eventType)
}
args = append(args, window.startMS, window.endMS)
return fmt.Sprintf("SELECT %s FROM %s WHERE app_code = ? AND event_type IN (%s) AND created_at_ms >= ? AND created_at_ms < ? %s", selectExpr, table, placeholders, suffix), args
}
func gameReplayEventTypes() []string {
return []string{
gamemq.EventTypeGameOrderSettled,
gamemq.EventTypeSelfGameMatchCreated,
gamemq.EventTypeSelfGameMatchReady,
gamemq.EventTypeSelfGameMatchSettled,
gamemq.EventTypeSelfGameMatchCanceled,
gamemq.EventTypeSelfGameMatchFailed,
gamemq.EventTypeSelfGamePoolAdjusted,
}
}
func replayEventID(eventID string, statTZ string) string {
token := replayToken(statTZ)
suffixed := eventID + ":stat_tz:" + token
if len(suffixed) <= replayEventIDMaxLen {
return suffixed
}
sum := sha256.Sum256([]byte(eventID + "|" + statTZ))
return "replay:" + token + ":" + hex.EncodeToString(sum[:])[:40]
}
func replayToken(statTZ string) string {
return strings.NewReplacer("/", "_", ":", "_").Replace(strings.ToLower(strings.TrimSpace(statTZ)))
}
func envDefault(key string, fallback string) string {
if value := strings.TrimSpace(os.Getenv(key)); value != "" {
return value
}
return fallback
}
func firstNonEmptyEnv(keys ...string) string {
for _, key := range keys {
if value := strings.TrimSpace(os.Getenv(key)); value != "" {
return value
}
}
return ""
}

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

@ -14,12 +14,13 @@ CREATE TABLE IF NOT EXISTS statistics_event_consumption (
created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms',
updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms',
PRIMARY KEY (app_code, source, event_id),
KEY idx_statistics_event_type (app_code, source, event_type, consumed_at_ms)
KEY idx_event_consumption_type (app_code, source, event_type, consumed_at_ms)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='统计事件消费幂等表';
CREATE TABLE IF NOT EXISTS stat_app_day_country (
app_code VARCHAR(32) NOT NULL COMMENT '应用编码',
stat_day DATE NOT NULL COMMENT 'UTC 统计日',
stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
stat_day DATE NOT NULL COMMENT '统计日，随 stat_tz 切换',
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 '新增用户数',
@ -40,75 +41,81 @@ 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, region_id),
KEY idx_stat_app_day_region (app_code, stat_day, region_id)
PRIMARY KEY (app_code, stat_tz, stat_day, country_id, region_id),
KEY idx_stat_app_day_region (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
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_region (app_code, stat_day, region_id)
PRIMARY KEY (app_code, stat_tz, stat_day, user_id),
KEY idx_stat_user_day_country (app_code, stat_tz, stat_day, country_id),
KEY idx_stat_user_day_region (app_code, stat_tz, stat_day, region_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='用户日活跃去重表';
CREATE TABLE IF NOT EXISTS stat_user_registration (
app_code VARCHAR(32) NOT NULL,
stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
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_region (app_code, registered_day, region_id)
PRIMARY KEY (app_code, stat_tz, user_id),
KEY idx_stat_user_registration_day (app_code, stat_tz, registered_day, country_id),
KEY idx_stat_user_registration_region (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
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_region (app_code, stat_day, region_id)
PRIMARY KEY (app_code, stat_tz, stat_day, user_id),
KEY idx_stat_recharge_payer_country (app_code, stat_tz, stat_day, country_id),
KEY idx_stat_recharge_payer_region (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
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_region (app_code, stat_day, region_id)
PRIMARY KEY (app_code, stat_tz, stat_day, user_id),
KEY idx_stat_lucky_payer_country (app_code, stat_tz, stat_day, country_id),
KEY idx_stat_lucky_payer_region (app_code, stat_tz, 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 统计日',
stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
stat_day DATE NOT NULL COMMENT '统计日，随 stat_tz 切换',
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, region_id, 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)
PRIMARY KEY (app_code, stat_tz, stat_day, country_id, region_id, pool_id),
KEY idx_stat_lucky_pool_overview (app_code, stat_tz, stat_day, pool_id),
KEY idx_stat_lucky_pool_region (app_code, stat_tz, stat_day, region_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物奖池国家日统计聚合表';
CREATE TABLE IF NOT EXISTS stat_real_room_robot_gift_day_rooms (
app_code VARCHAR(32) NOT NULL COMMENT '应用编码',
stat_day DATE NOT NULL COMMENT 'UTC 统计日',
stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
stat_day DATE NOT NULL COMMENT '统计日，随 stat_tz 切换',
country_id BIGINT NOT NULL DEFAULT 0 COMMENT '国家维度 ID，0 表示未知或全局',
region_id BIGINT NOT NULL DEFAULT 0 COMMENT '区域维度 ID，0 表示未知或全局',
room_id VARCHAR(96) NOT NULL COMMENT '真人房间 ID',
@ -116,13 +123,14 @@ CREATE TABLE IF NOT EXISTS stat_real_room_robot_gift_day_rooms (
lucky_gift_coin BIGINT NOT NULL DEFAULT 0 COMMENT '机器人在真人房送出的幸运礼物金币价值',
super_lucky_gift_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, region_id, room_id),
KEY idx_stat_real_room_robot_gift_region (app_code, stat_day, region_id),
KEY idx_stat_real_room_robot_gift_country (app_code, stat_day, country_id)
PRIMARY KEY (app_code, stat_tz, stat_day, country_id, region_id, room_id),
KEY idx_stat_real_room_robot_gift_region (app_code, stat_tz, stat_day, region_id),
KEY idx_stat_real_room_robot_gift_country (app_code, stat_tz, stat_day, country_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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
stat_day DATE NOT NULL,
country_id BIGINT NOT NULL DEFAULT 0,
region_id BIGINT NOT NULL DEFAULT 0,
@ -133,12 +141,13 @@ 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, region_id, platform_code, game_id),
KEY idx_stat_game_day_region (app_code, stat_day, region_id)
PRIMARY KEY (app_code, stat_tz, stat_day, country_id, region_id, platform_code, game_id),
KEY idx_stat_game_day_region (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
stat_day DATE NOT NULL,
country_id BIGINT NOT NULL DEFAULT 0,
region_id BIGINT NOT NULL DEFAULT 0,
@ -146,12 +155,13 @@ CREATE TABLE IF NOT EXISTS stat_game_day_players (
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),
KEY idx_stat_game_player_region (app_code, stat_day, region_id)
PRIMARY KEY (app_code, stat_tz, stat_day, country_id, platform_code, game_id, user_id),
KEY idx_stat_game_player_region (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
stat_day DATE NOT NULL,
country_id BIGINT NOT NULL DEFAULT 0,
region_id BIGINT NOT NULL DEFAULT 0,
@ -167,24 +177,26 @@ CREATE TABLE IF NOT EXISTS stat_self_game_h5_event_day (
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)
PRIMARY KEY (app_code, stat_tz, stat_day, country_id, region_id, game_id, event_name, language, client_version, h5_version, entry_source),
KEY idx_stat_self_game_h5_event_name (app_code, stat_tz, stat_day, game_id, event_name),
KEY idx_stat_self_game_h5_users_day (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
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)
PRIMARY KEY (app_code, stat_tz, stat_day, game_id, event_name, user_id),
KEY idx_stat_self_game_h5_users_day (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
stat_day DATE NOT NULL,
country_id BIGINT NOT NULL DEFAULT 0,
region_id BIGINT NOT NULL DEFAULT 0,
@ -212,13 +224,14 @@ CREATE TABLE IF NOT EXISTS stat_self_game_match_day (
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)
PRIMARY KEY (app_code, stat_tz, stat_day, country_id, region_id, game_id, stake_coin),
KEY idx_stat_self_game_match_game (app_code, stat_tz, stat_day, game_id),
KEY idx_stat_self_game_match_region (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
stat_day DATE NOT NULL,
game_id VARCHAR(96) NOT NULL,
direction VARCHAR(16) NOT NULL DEFAULT '',
@ -228,12 +241,13 @@ CREATE TABLE IF NOT EXISTS stat_self_game_pool_day (
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)
PRIMARY KEY (app_code, stat_tz, stat_day, game_id, direction, reason),
KEY idx_stat_self_game_pool_game (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
stat_day DATE NOT NULL,
country_id BIGINT NOT NULL DEFAULT 0,
region_id BIGINT NOT NULL DEFAULT 0,
@ -247,13 +261,14 @@ CREATE TABLE IF NOT EXISTS stat_self_game_participant_day (
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)
PRIMARY KEY (app_code, stat_tz, stat_day, country_id, region_id, game_id, stake_coin, participant_type, result, rps_gesture, dice_point),
KEY idx_stat_self_game_participant_game (app_code, stat_tz, stat_day, game_id),
KEY idx_stat_self_game_participant_region (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
stat_day DATE NOT NULL,
country_id BIGINT NOT NULL DEFAULT 0,
region_id BIGINT NOT NULL DEFAULT 0,
@ -268,7 +283,7 @@ CREATE TABLE IF NOT EXISTS stat_self_game_user_day (
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)
PRIMARY KEY (app_code, stat_tz, stat_day, game_id, user_id),
KEY idx_stat_self_game_user_win (app_code, stat_tz, stat_day, game_id, net_win_coin),
KEY idx_stat_self_game_user_region (app_code, stat_tz, stat_day, region_id, game_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='自研游戏用户风险日统计表';

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

@ -141,81 +141,24 @@ func newConsumers(cfg config.Config, repo *mysqlstorage.Repository) ([]*rocketmq
return nil
}
if err := add(cfg.RocketMQ.UserOutbox, usermq.TagUserOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
event, ok, err := userRegisteredEvent(message.Body)
if err != nil || !ok {
return err
}
return repo.ConsumeUserRegistered(appcode.WithContext(ctx, event.AppCode), event)
return ConsumeUserOutboxBody(ctx, repo, message.Body)
}); err != nil {
return nil, err
}
if err := add(cfg.RocketMQ.WalletOutbox, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
event, ok, err := rechargeEvent(message.Body)
if err != nil {
return err
}
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
}
return repo.ConsumeLuckyGiftReward(appcode.WithContext(ctx, reward.AppCode), reward)
return ConsumeWalletOutboxBody(ctx, repo, message.Body)
}); err != nil {
shutdownConsumers(consumers)
return nil, err
}
if err := add(cfg.RocketMQ.RoomOutbox, roommq.TagRoomOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
if gift, ok, err := roomGiftEvent(message.Body); err != nil || ok {
if err != nil || !ok {
return err
}
if err := repo.ConsumeRoomGift(appcode.WithContext(ctx, gift.AppCode), gift); err != nil {
return err
}
return repo.ConsumeUserActive(appcode.WithContext(ctx, gift.AppCode), mysqlstorage.UserActiveEvent{
AppCode: gift.AppCode,
EventID: gift.EventID + ":active",
EventType: "RoomGiftSentActive",
UserID: gift.SenderUserID,
CountryID: gift.CountryID,
RegionID: gift.VisibleRegionID,
OccurredAtMS: gift.OccurredAtMS,
})
}
active, ok, err := roomJoinActiveEvent(message.Body)
if err != nil || !ok {
return err
}
return repo.ConsumeUserActive(appcode.WithContext(ctx, active.AppCode), active)
return ConsumeRoomOutboxBody(ctx, repo, message.Body)
}); err != nil {
shutdownConsumers(consumers)
return nil, err
}
if err := add(cfg.RocketMQ.GameOutbox, gamemq.TagGameOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
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)
}
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)
return ConsumeGameOutboxBody(ctx, repo, message.Body)
}); err != nil {
shutdownConsumers(consumers)
return nil, err
@ -223,6 +166,83 @@ func newConsumers(cfg config.Config, repo *mysqlstorage.Repository) ([]*rocketmq
return consumers, nil
}
// ConsumeUserOutboxBody 是 user outbox 的唯一统计投影入口；实时 MQ 和离线回灌共用它，避免同一 payload 出现两套解析口径。
func ConsumeUserOutboxBody(ctx context.Context, repo *mysqlstorage.Repository, body []byte) error {
event, ok, err := userRegisteredEvent(body)
if err != nil || !ok {
return err
}
return repo.ConsumeUserRegistered(appcode.WithContext(ctx, event.AppCode), event)
}
// ConsumeWalletOutboxBody 按钱包事实类型分派充值、币商补货和幸运礼物返奖；未知事件只提交消费位点，不写统计读模型。
func ConsumeWalletOutboxBody(ctx context.Context, repo *mysqlstorage.Repository, body []byte) error {
event, ok, err := rechargeEvent(body)
if err != nil {
return err
}
if ok {
return repo.ConsumeRecharge(appcode.WithContext(ctx, event.AppCode), event)
}
stock, ok, err := coinSellerStockEvent(body)
if err != nil {
return err
}
if ok {
return repo.ConsumeCoinSellerStock(appcode.WithContext(ctx, stock.AppCode), stock)
}
reward, ok, err := luckyGiftRewardEvent(body)
if err != nil || !ok {
return err
}
return repo.ConsumeLuckyGiftReward(appcode.WithContext(ctx, reward.AppCode), reward)
}
// ConsumeRoomOutboxBody 只投影房间统计需要的礼物和真人进房事件；礼物事实额外产生 sender 当日活跃，保持实时和回灌一致。
func ConsumeRoomOutboxBody(ctx context.Context, repo *mysqlstorage.Repository, body []byte) error {
if gift, ok, err := roomGiftEvent(body); err != nil || ok {
if err != nil || !ok {
return err
}
if err := repo.ConsumeRoomGift(appcode.WithContext(ctx, gift.AppCode), gift); err != nil {
return err
}
return repo.ConsumeUserActive(appcode.WithContext(ctx, gift.AppCode), mysqlstorage.UserActiveEvent{
AppCode: gift.AppCode,
EventID: gift.EventID + ":active",
EventType: "RoomGiftSentActive",
UserID: gift.SenderUserID,
CountryID: gift.CountryID,
RegionID: gift.VisibleRegionID,
OccurredAtMS: gift.OccurredAtMS,
})
}
active, ok, err := roomJoinActiveEvent(body)
if err != nil || !ok {
return err
}
return repo.ConsumeUserActive(appcode.WithContext(ctx, active.AppCode), active)
}
// ConsumeGameOutboxBody 同时承接厂商游戏订单和自研游戏局级事实；回灌复用它可保证利润、奖池和参与者分布口径一致。
func ConsumeGameOutboxBody(ctx context.Context, repo *mysqlstorage.Repository, body []byte) error {
if event, ok, err := gameOrderEvent(body); err != nil || ok {
if err != nil || !ok {
return err
}
return repo.ConsumeGameOrder(appcode.WithContext(ctx, event.AppCode), event)
}
match, ok, err := selfGameMatchEvent(body)
if err != nil || !ok {
pool, poolOK, poolErr := selfGamePoolAdjustmentEvent(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)
}
func userRegisteredEvent(body []byte) (mysqlstorage.UserRegisteredEvent, bool, error) {
message, err := usermq.DecodeUserOutboxMessage(body)
if err != nil {

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

@ -69,6 +69,7 @@ func (s *queryHTTPServer) overview(w http.ResponseWriter, r *http.Request) {
}
overview, err := s.repo.QueryOverview(r.Context(), mysqlstorage.OverviewQuery{
AppCode: query.Get("app_code"),
StatTZ: query.Get("stat_tz"),
StartMS: startMS,
EndMS: endMS,
SeriesStartMS: parseInt64(query.Get("series_start_ms")),
@ -169,6 +170,7 @@ func (s *queryHTTPServer) selfGameOverview(w http.ResponseWriter, r *http.Reques
}
overview, err := s.repo.QuerySelfGameOverview(r.Context(), mysqlstorage.SelfGameOverviewQuery{
AppCode: query.Get("app_code"),
StatTZ: query.Get("stat_tz"),
StartMS: startMS,
EndMS: endMS,
GameID: query.Get("game_id"),

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

@ -13,6 +13,7 @@ import (
type OverviewQuery struct {
AppCode string
StatTZ string
StartMS int64
EndMS int64
SeriesStartMS int64
@ -23,6 +24,7 @@ type OverviewQuery struct {
type Overview struct {
AppCode string `json:"app_code"`
StatTZ string `json:"stat_tz"`
StartMS int64 `json:"start_ms"`
EndMS int64 `json:"end_ms"`
CountryID int64 `json:"country_id"`
@ -164,6 +166,7 @@ type CountryBreakdown struct {
type SelfGameOverviewQuery struct {
AppCode string
StatTZ string
StartMS int64
EndMS int64
GameID string
@ -173,6 +176,7 @@ type SelfGameOverviewQuery struct {
type SelfGameOverview struct {
AppCode string `json:"app_code"`
StatTZ string `json:"stat_tz"`
StartMS int64 `json:"start_ms"`
EndMS int64 `json:"end_ms"`
GameID string `json:"game_id"`
@ -320,27 +324,28 @@ func (r *Repository) QueryOverview(ctx context.Context, query OverviewQuery) (Ov
if app == "" {
app = "lalu"
}
startDay := statDay(query.StartMS)
endDay := statDay(query.EndMS - 1)
statTZ := normalizeStatTZ(query.StatTZ)
startDay := statDayIn(query.StartMS, statTZ)
endDay := statDayIn(query.EndMS-1, statTZ)
previousStartDay, previousEndDay := previousStatDayRange(startDay, endDay)
overview, err := r.queryAppDayOverviewTotals(ctx, app, startDay, endDay, query.CountryID, query.RegionID)
overview, err := r.queryAppDayOverviewTotals(ctx, app, statTZ, startDay, endDay, query.CountryID, query.RegionID)
if err != nil {
return Overview{}, err
}
previous, err := r.queryAppDayOverviewTotals(ctx, app, previousStartDay, previousEndDay, query.CountryID, query.RegionID)
previous, err := r.queryAppDayOverviewTotals(ctx, app, statTZ, 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
overview.AppCode, overview.StatTZ, overview.StartMS, overview.EndMS, overview.CountryID = app, statTZ, query.StartMS, query.EndMS, query.CountryID
previous.AppCode, previous.StatTZ, previous.CountryID = app, statTZ, query.CountryID
// 新增付费用户来自日充值用户去重表和注册表的交集；只读聚合小表，避免 Databi 页面回扫钱包或订单明细。
if overview.NewPaidUsers, err = r.queryNewPaidUsers(ctx, app, startDay, endDay, query.CountryID, query.RegionID); err != nil {
if overview.NewPaidUsers, err = r.queryNewPaidUsers(ctx, app, statTZ, 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 {
if previous.NewPaidUsers, err = r.queryNewPaidUsers(ctx, app, statTZ, previousStartDay, previousEndDay, query.CountryID, query.RegionID); err != nil {
return Overview{}, err
}
if robotGifts, err := r.queryRealRoomRobotGiftOverview(ctx, app, startDay, endDay, query.CountryID, query.RegionID); err != nil {
if robotGifts, err := r.queryRealRoomRobotGiftOverview(ctx, app, statTZ, startDay, endDay, query.CountryID, query.RegionID); err != nil {
return Overview{}, err
} else {
overview.RealRoomRobotNormalGiftCoin = robotGifts.RealRoomRobotNormalGiftCoin
@ -351,39 +356,41 @@ func (r *Repository) QueryOverview(ctx context.Context, query OverviewQuery) (Ov
overview.RealRoomRobotAvgRoomGiftCoin = robotGifts.RealRoomRobotAvgRoomGiftCoin
}
// Pool 明细和顶部汇总来自同一个统计窗口；Databi 用它来下钻展示每个 pool_id 的流水、返奖和利润。
pools, err := r.queryLuckyGiftPools(ctx, app, startDay, endDay, query.CountryID, query.RegionID)
pools, err := r.queryLuckyGiftPools(ctx, app, statTZ, 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, 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, query.RegionID); err != nil {
return Overview{}, err
} else if ok {
previous.LuckyGiftTurnover, previous.LuckyGiftPayout = sumLuckyGiftPools(activityPools)
if statTZ == StatTZUTC {
// activity-service 的 lucky_draw_pool_day_stats 仍是 UTC 日表；只有 UTC 查询能拿它覆盖旧统计行，北京日必须使用本服务 stat_tz=Asia/Shanghai 的读模型。
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, query.RegionID); err != nil {
return Overview{}, err
} else if ok {
previous.LuckyGiftTurnover, previous.LuckyGiftPayout = sumLuckyGiftPools(activityPools)
}
}
applyOverviewDerivedMetrics(&overview)
applyOverviewDerivedMetrics(&previous)
overview.applyDeltaRates(previous)
retention, err := r.queryRetention(ctx, app, startDay, endDay, query.CountryID, query.RegionID)
retention, err := r.queryRetention(ctx, app, statTZ, 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, query.RegionID)
ranking, err := r.queryGameRanking(ctx, app, statTZ, 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)
breakdown, err := r.queryCountryBreakdown(ctx, app, statTZ, startDay, endDay, query.RegionID)
if err != nil {
return Overview{}, err
}
@ -391,11 +398,11 @@ func (r *Repository) QueryOverview(ctx context.Context, query OverviewQuery) (Ov
}
seriesStartDay, seriesEndDay := startDay, endDay
if query.SeriesStartMS > 0 && query.SeriesEndMS > query.SeriesStartMS {
seriesStartDay = statDay(query.SeriesStartMS)
seriesEndDay = statDay(query.SeriesEndMS - 1)
seriesStartDay = statDayIn(query.SeriesStartMS, statTZ)
seriesEndDay = statDayIn(query.SeriesEndMS-1, statTZ)
}
// daily_series 服务卡片内 sparkline 和弹窗折线图；一条 GROUP BY stat_day 查询返回全部指标，避免前端逐日请求。
dailySeries, err := r.queryDailySeries(ctx, app, seriesStartDay, seriesEndDay, query.CountryID, query.RegionID)
dailySeries, err := r.queryDailySeries(ctx, app, statTZ, seriesStartDay, seriesEndDay, query.CountryID, query.RegionID)
if err != nil {
return Overview{}, err
}
@ -409,49 +416,51 @@ func (r *Repository) QuerySelfGameOverview(ctx context.Context, query SelfGameOv
if app == "" {
app = "lalu"
}
startDay := statDay(startMS)
endDay := statDay(endMS - 1)
statTZ := normalizeStatTZ(query.StatTZ)
startDay := statDayIn(startMS, statTZ)
endDay := statDayIn(endMS-1, statTZ)
gameID := normalizeShortText(query.GameID)
overview := SelfGameOverview{
AppCode: app,
StatTZ: statTZ,
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)
events, err := r.querySelfGameEvents(ctx, app, statTZ, 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)
totals, err := r.querySelfGameMatchTotals(ctx, app, statTZ, 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)
stakes, err := r.querySelfGameStakeBreakdown(ctx, app, statTZ, 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)
pools, err := r.querySelfGamePools(ctx, app, statTZ, 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)
participants, err := r.querySelfGameParticipantDistribution(ctx, app, statTZ, 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)
risk, err := r.querySelfGameUserRisk(ctx, app, statTZ, startDay, endDay, gameID, query.CountryID, query.RegionID)
if err != nil {
return SelfGameOverview{}, err
}
overview.UserRisk = risk
retention, err := r.querySelfGameRetention(ctx, app, startDay, endDay, gameID, query.CountryID, query.RegionID)
retention, err := r.querySelfGameRetention(ctx, app, statTZ, startDay, endDay, gameID, query.CountryID, query.RegionID)
if err != nil {
return SelfGameOverview{}, err
}
@ -460,8 +469,8 @@ func (r *Repository) QuerySelfGameOverview(ctx context.Context, query SelfGameOv
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}
func (r *Repository) queryAppDayOverviewTotals(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) (Overview, error) {
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
filter := ""
if countryID > 0 {
filter += " AND country_id = ?"
@ -481,7 +490,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 ?`+filter, args...)
WHERE app_code = ? AND stat_tz = ? 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
@ -489,8 +498,8 @@ 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}
func (r *Repository) queryNewPaidUsers(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) (int64, error) {
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay, normalizeStatTZ(statTZ), startDay, endDay}
filter := ""
if countryID > 0 {
filter += " AND p.country_id = ?"
@ -504,9 +513,9 @@ func (r *Repository) queryNewPaidUsers(ctx context.Context, app, startDay, endDa
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...)
JOIN stat_user_registration r ON r.app_code = p.app_code AND r.stat_tz = p.stat_tz AND r.user_id = p.user_id
WHERE p.app_code = ? AND p.stat_tz = ? AND p.stat_day BETWEEN ? AND ?
AND r.stat_tz = ? AND r.registered_day BETWEEN ? AND ?`+filter, args...)
var count int64
if err := row.Scan(&count); err != nil && err != sql.ErrNoRows {
return 0, err
@ -514,8 +523,8 @@ func (r *Repository) queryNewPaidUsers(ctx context.Context, app, startDay, endDa
return count, nil
}
func (r *Repository) queryRealRoomRobotGiftOverview(ctx context.Context, app, startDay, endDay string, countryID int64, regionID int64) (Overview, error) {
args := []any{app, startDay, endDay}
func (r *Repository) queryRealRoomRobotGiftOverview(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) (Overview, error) {
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
filter := ""
if countryID > 0 {
filter += " AND country_id = ?"
@ -531,7 +540,7 @@ func (r *Repository) queryRealRoomRobotGiftOverview(ctx context.Context, app, st
COALESCE(SUM(super_lucky_gift_coin),0),
COUNT(*)
FROM stat_real_room_robot_gift_day_rooms
WHERE app_code = ? AND stat_day BETWEEN ? AND ?`+filter, args...)
WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`+filter, args...)
var overview Overview
if err := row.Scan(&overview.RealRoomRobotNormalGiftCoin, &overview.RealRoomRobotLuckyGiftCoin, &overview.RealRoomRobotSuperGiftCoin, &overview.RealRoomRobotGiftRoomCount); err != nil {
return Overview{}, err
@ -541,8 +550,8 @@ func (r *Repository) queryRealRoomRobotGiftOverview(ctx context.Context, app, st
return overview, nil
}
func (r *Repository) queryDailySeries(ctx context.Context, app, startDay, endDay string, countryID int64, regionID int64) ([]DailySeriesItem, error) {
args := []any{app, startDay, endDay}
func (r *Repository) queryDailySeries(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) ([]DailySeriesItem, error) {
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
filter := ""
if countryID > 0 {
filter += " AND country_id = ?"
@ -562,7 +571,7 @@ func (r *Repository) queryDailySeries(ctx context.Context, app, startDay, endDay
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+`
WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`+filter+`
GROUP BY stat_day
ORDER BY stat_day ASC`, args...)
if err != nil {
@ -584,23 +593,25 @@ func (r *Repository) queryDailySeries(ctx context.Context, app, startDay, endDay
if err := rows.Err(); err != nil {
return nil, err
}
newPaidUsersByDay, err := r.queryDailyNewPaidUsers(ctx, app, startDay, endDay, countryID, regionID)
newPaidUsersByDay, err := r.queryDailyNewPaidUsers(ctx, app, statTZ, 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)
if normalizeStatTZ(statTZ) == StatTZUTC {
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}
func (r *Repository) queryDailyNewPaidUsers(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) (map[string]int64, error) {
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay, normalizeStatTZ(statTZ), startDay, endDay}
filter := ""
if countryID > 0 {
filter += " AND p.country_id = ?"
@ -613,9 +624,9 @@ func (r *Repository) queryDailyNewPaidUsers(ctx context.Context, app, startDay,
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+`
JOIN stat_user_registration r ON r.app_code = p.app_code AND r.stat_tz = p.stat_tz AND r.user_id = p.user_id
WHERE p.app_code = ? AND p.stat_tz = ? AND p.stat_day BETWEEN ? AND ?
AND r.stat_tz = ? AND r.registered_day BETWEEN ? AND ?`+filter+`
GROUP BY p.stat_day`, args...)
if err != nil {
return nil, err
@ -686,9 +697,9 @@ 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, regionID int64) (Retention, error) {
func (r *Repository) queryRetention(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) (Retention, error) {
filter := ""
args := []any{app, startDay, endDay}
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
if countryID > 0 {
filter = " AND r.country_id = ?"
args = append(args, countryID)
@ -699,11 +710,11 @@ func (r *Repository) queryRetention(ctx context.Context, app, startDay, endDay s
}
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),
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 7 DAY))),0),
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 30 DAY))),0)
COALESCE(SUM(EXISTS(SELECT 1 FROM stat_user_day_activity a WHERE a.app_code=r.app_code AND a.stat_tz=r.stat_tz AND a.user_id=r.user_id AND a.stat_day=DATE_ADD(r.registered_day, INTERVAL 1 DAY))),0),
COALESCE(SUM(EXISTS(SELECT 1 FROM stat_user_day_activity a WHERE a.app_code=r.app_code AND a.stat_tz=r.stat_tz AND a.user_id=r.user_id AND a.stat_day=DATE_ADD(r.registered_day, INTERVAL 7 DAY))),0),
COALESCE(SUM(EXISTS(SELECT 1 FROM stat_user_day_activity a WHERE a.app_code=r.app_code AND a.stat_tz=r.stat_tz AND a.user_id=r.user_id AND a.stat_day=DATE_ADD(r.registered_day, INTERVAL 30 DAY))),0)
FROM stat_user_registration r
WHERE r.app_code = ? AND r.registered_day BETWEEN ? AND ?`+filter, args...)
WHERE r.app_code = ? AND r.stat_tz = ? AND r.registered_day BETWEEN ? AND ?`+filter, args...)
var out Retention
if err := row.Scan(&out.RegisteredUsers, &out.Day1Users, &out.Day7Users, &out.Day30Users); err != nil && err != sql.ErrNoRows {
return Retention{}, err
@ -714,8 +725,8 @@ 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, regionID int64) ([]GameRank, error) {
args := []any{app, startDay, endDay}
func (r *Repository) queryGameRanking(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) ([]GameRank, error) {
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
filter := ""
if countryID > 0 {
filter = " AND country_id = ?"
@ -729,7 +740,7 @@ func (r *Repository) queryGameRanking(ctx context.Context, app, startDay, endDay
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)
FROM stat_game_day_country
WHERE app_code = ? AND stat_day BETWEEN ? AND ?`+filter+`
WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`+filter+`
GROUP BY platform_code, game_id
ORDER BY COALESCE(SUM(turnover_coin),0) DESC
LIMIT 50`, args...)
@ -750,8 +761,8 @@ func (r *Repository) queryGameRanking(ctx context.Context, app, startDay, endDay
return out, rows.Err()
}
func (r *Repository) queryCountryBreakdown(ctx context.Context, app, startDay, endDay string, regionID int64) ([]CountryBreakdown, error) {
args := []any{app, startDay, endDay}
func (r *Repository) queryCountryBreakdown(ctx context.Context, app, statTZ, startDay, endDay string, regionID int64) ([]CountryBreakdown, error) {
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
filter := ""
if regionID > 0 {
filter = " AND region_id = ?"
@ -768,7 +779,7 @@ func (r *Repository) queryCountryBreakdown(ctx context.Context, app, startDay, e
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+`
WHERE app_code = ? AND stat_tz = ? 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...)
@ -805,8 +816,8 @@ func applyCountryBreakdownDerivedMetrics(item *CountryBreakdown) {
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}
func (r *Repository) queryLuckyGiftPools(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) ([]LuckyGiftPoolStat, error) {
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
filter := ""
if countryID > 0 {
filter = " AND country_id = ?"
@ -819,7 +830,7 @@ func (r *Repository) queryLuckyGiftPools(ctx context.Context, app, startDay, end
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
WHERE app_code = ? AND stat_day BETWEEN ? AND ?`+filter+`
WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`+filter+`
GROUP BY pool_id
ORDER BY COALESCE(SUM(turnover_coin),0) DESC, pool_id ASC`, args...)
if err != nil {
@ -955,8 +966,8 @@ 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)
func (r *Repository) querySelfGameEvents(ctx context.Context, app, statTZ, startDay, endDay, gameID string, countryID int64, regionID int64) ([]SelfGameEventStat, error) {
filter, args := selfGameFilter(app, statTZ, 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)
@ -980,8 +991,8 @@ func (r *Repository) querySelfGameEvents(ctx context.Context, app, startDay, end
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)
func (r *Repository) querySelfGameMatchTotals(ctx context.Context, app, statTZ, startDay, endDay, gameID string, countryID int64, regionID int64) (SelfGameMatchTotals, error) {
filter, args := selfGameFilter(app, statTZ, 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),
@ -1000,8 +1011,8 @@ func (r *Repository) querySelfGameMatchTotals(ctx context.Context, app, startDay
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)
func (r *Repository) querySelfGameStakeBreakdown(ctx context.Context, app, statTZ, startDay, endDay, gameID string, countryID int64, regionID int64) ([]SelfGameStakeStat, error) {
filter, args := selfGameFilter(app, statTZ, 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),
@ -1030,15 +1041,16 @@ func (r *Repository) querySelfGameStakeBreakdown(ctx context.Context, app, start
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}
func (r *Repository) querySelfGamePools(ctx context.Context, app, statTZ, startDay, endDay, gameID string) ([]SelfGamePoolStat, error) {
statTZ = normalizeStatTZ(statTZ)
filter := "app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?"
args := []any{app, statTZ, startDay, endDay}
if gameID != "" {
filter += " AND game_id = ?"
args = append(args, gameID)
}
latestArgs := append([]any{}, args...)
latestArgs = append(latestArgs, app)
latestArgs = append(latestArgs, app, statTZ)
latestArgs = append(latestArgs, args...)
rows, err := r.db.QueryContext(ctx, `
SELECT agg.game_id, agg.direction, agg.reason, agg.transaction_count,
@ -1050,7 +1062,7 @@ func (r *Repository) querySelfGamePools(ctx context.Context, app, startDay, endD
WHERE `+filter+`
GROUP BY game_id, direction, reason
) agg
LEFT JOIN stat_self_game_pool_day latest ON latest.app_code = ? AND latest.game_id = agg.game_id
LEFT JOIN stat_self_game_pool_day latest ON latest.app_code = ? AND latest.stat_tz = ? AND latest.game_id = agg.game_id
AND latest.direction = agg.direction AND latest.reason = agg.reason
AND latest.stat_day = (
SELECT p2.stat_day
@ -1075,8 +1087,8 @@ func (r *Repository) querySelfGamePools(ctx context.Context, app, startDay, endD
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)
func (r *Repository) querySelfGameParticipantDistribution(ctx context.Context, app, statTZ, startDay, endDay, gameID string, countryID int64, regionID int64) ([]SelfGameParticipantStat, error) {
filter, args := selfGameFilter(app, statTZ, 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)
@ -1121,8 +1133,8 @@ func selfGameParticipantRateKey(item SelfGameParticipantStat) string {
return item.GameID + "\x00" + strconv.FormatInt(item.StakeCoin, 10) + "\x00" + item.ParticipantType + "\x00" + playKey
}
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)
func (r *Repository) querySelfGameUserRisk(ctx context.Context, app, statTZ, startDay, endDay, gameID string, countryID int64, regionID int64) ([]SelfGameUserRiskStat, error) {
filter, args := selfGameFilter(app, statTZ, 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),
@ -1149,9 +1161,9 @@ func (r *Repository) querySelfGameUserRisk(ctx context.Context, app, startDay, e
return out, rows.Err()
}
func (r *Repository) querySelfGameRetention(ctx context.Context, app, startDay, endDay, gameID string, countryID int64, regionID int64) (SelfGameRetention, error) {
cohortFilter, cohortArgs := selfGameUserDayFilter(app, startDay, endDay, gameID, countryID, regionID)
returnFilter, returnArgs := selfGameUserDayFilter(app, addStatDays(startDay, 1), addStatDays(endDay, 7), gameID, countryID, regionID)
func (r *Repository) querySelfGameRetention(ctx context.Context, app, statTZ, startDay, endDay, gameID string, countryID int64, regionID int64) (SelfGameRetention, error) {
cohortFilter, cohortArgs := selfGameUserDayFilter(app, statTZ, startDay, endDay, gameID, countryID, regionID)
returnFilter, returnArgs := selfGameUserDayFilter(app, statTZ, addStatDays(startDay, 1), addStatDays(endDay, 7), gameID, countryID, regionID)
args := make([]any, 0, len(cohortArgs)+len(returnArgs)*2)
args = append(args, cohortArgs...)
args = append(args, returnArgs...)
@ -1184,9 +1196,9 @@ func (r *Repository) querySelfGameRetention(ctx context.Context, app, startDay,
return retention, nil
}
func selfGameUserDayFilter(app, startDay, endDay, gameID string, countryID int64, regionID int64) (string, []any) {
filter := "app_code = ? AND match_count > 0"
args := []any{app}
func selfGameUserDayFilter(app, statTZ, startDay, endDay, gameID string, countryID int64, regionID int64) (string, []any) {
filter := "app_code = ? AND stat_tz = ? AND match_count > 0"
args := []any{app, normalizeStatTZ(statTZ)}
if startDay != "" && endDay != "" {
filter += " AND stat_day BETWEEN ? AND ?"
args = append(args, startDay, endDay)
@ -1206,9 +1218,9 @@ func selfGameUserDayFilter(app, startDay, endDay, gameID string, countryID int64
return filter, args
}
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}
func selfGameFilter(app, statTZ, startDay, endDay, gameID string, countryID int64, regionID int64) (string, []any) {
filter := "app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?"
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
if gameID != "" {
filter += " AND game_id = ?"
args = append(args, gameID)

106
services/statistics-service/internal/storage/mysql/query_test.go
View File

@ -99,7 +99,7 @@ func TestConsumeUserRegisteredCountsOnlyNewRegistration(t *testing.T) {
if err := repository.db.QueryRowContext(ctx, `
SELECT new_users, active_users
FROM stat_app_day_country
WHERE app_code = 'lalu' AND stat_day = '2026-06-22' AND country_id = 15 AND region_id = 10
WHERE app_code = 'lalu' AND stat_tz = 'UTC' AND stat_day = '2026-06-22' AND country_id = 15 AND region_id = 10
`).Scan(&newUsers, &activeUsers); err != nil {
t.Fatalf("query app day country: %v", err)
}
@ -115,6 +115,7 @@ func TestConsumeUserRegisteredCountsOnlyNewRegistration(t *testing.T) {
SELECT COUNT(*)
FROM stat_user_day_activity
WHERE app_code = 'lalu'
AND stat_tz = 'UTC'
AND stat_day = '2026-06-22'
AND country_id = 15
AND region_id = 10
@ -127,6 +128,109 @@ func TestConsumeUserRegisteredCountsOnlyNewRegistration(t *testing.T) {
}
}
func TestConsumeRechargeWritesUTCAndChinaStatDays(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_stat_tz_test",
})
repository, err := Open(ctx, statsSchema.DSN)
if err != nil {
t.Fatalf("open repository: %v", err)
}
t.Cleanup(func() { _ = repository.Close() })
occurredAt := time.Date(2026, 6, 23, 16, 30, 0, 0, time.UTC).UnixMilli()
if err := repository.ConsumeRecharge(ctx, RechargeEvent{
AppCode: "lalu",
EventID: "recharge:tz-boundary:1",
EventType: "WalletRechargeRecorded",
UserID: 9001,
TargetCountryID: 63,
TargetRegionID: 210,
USDMinor: 80_000,
RechargeType: "mifapay",
OccurredAtMS: occurredAt,
}); err != nil {
t.Fatalf("consume recharge: %v", err)
}
utcStart := time.Date(2026, 6, 23, 0, 0, 0, 0, time.UTC).UnixMilli()
utcOverview, err := repository.QueryOverview(ctx, OverviewQuery{AppCode: "lalu", StartMS: utcStart, EndMS: utcStart + int64(24*time.Hour/time.Millisecond), CountryID: 63})
if err != nil {
t.Fatalf("query utc overview: %v", err)
}
if utcOverview.StatTZ != StatTZUTC || utcOverview.RechargeUSDMinor != 80_000 || utcOverview.MifaPayRechargeUSDMinor != 80_000 {
t.Fatalf("UTC stat day mismatch: %+v", utcOverview)
}
beijingStart := time.Date(2026, 6, 23, 16, 0, 0, 0, time.UTC).UnixMilli()
chinaOverview, err := repository.QueryOverview(ctx, OverviewQuery{AppCode: "lalu", StatTZ: StatTZAsiaShanghai, StartMS: beijingStart, EndMS: beijingStart + int64(24*time.Hour/time.Millisecond), CountryID: 63})
if err != nil {
t.Fatalf("query china overview: %v", err)
}
if chinaOverview.StatTZ != StatTZAsiaShanghai || chinaOverview.RechargeUSDMinor != 80_000 || len(chinaOverview.DailySeries) != 1 || chinaOverview.DailySeries[0].StatDay != "2026-06-24" {
t.Fatalf("China stat day mismatch: %+v", chinaOverview)
}
nextUTCStart := time.Date(2026, 6, 24, 0, 0, 0, 0, time.UTC).UnixMilli()
nextUTCOverview, err := repository.QueryOverview(ctx, OverviewQuery{AppCode: "lalu", StartMS: nextUTCStart, EndMS: nextUTCStart + int64(24*time.Hour/time.Millisecond), CountryID: 63})
if err != nil {
t.Fatalf("query next utc overview: %v", err)
}
if nextUTCOverview.RechargeUSDMinor != 0 {
t.Fatalf("UTC 2026-06-24 must not include previous UTC day recharge: %+v", nextUTCOverview)
}
}
func TestConsumeRechargeWithStatTZScopeDoesNotTouchUTC(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_stat_tz_scope_test",
})
repository, err := Open(ctx, statsSchema.DSN)
if err != nil {
t.Fatalf("open repository: %v", err)
}
t.Cleanup(func() { _ = repository.Close() })
occurredAt := time.Date(2026, 6, 23, 16, 30, 0, 0, time.UTC).UnixMilli()
if err := repository.ConsumeRecharge(WithStatTZScope(ctx, StatTZAsiaShanghai), RechargeEvent{
AppCode: "lalu",
EventID: "recharge:tz-scope:1",
EventType: "WalletRechargeRecorded",
UserID: 9002,
TargetCountryID: 63,
TargetRegionID: 210,
USDMinor: 90_000,
RechargeType: "mifapay",
OccurredAtMS: occurredAt,
}); err != nil {
t.Fatalf("consume scoped recharge: %v", err)
}
utcStart := time.Date(2026, 6, 23, 0, 0, 0, 0, time.UTC).UnixMilli()
utcOverview, err := repository.QueryOverview(ctx, OverviewQuery{AppCode: "lalu", StartMS: utcStart, EndMS: utcStart + int64(24*time.Hour/time.Millisecond), CountryID: 63})
if err != nil {
t.Fatalf("query utc overview: %v", err)
}
if utcOverview.RechargeUSDMinor != 0 {
t.Fatalf("scoped replay must not mutate UTC rows: %+v", utcOverview)
}
beijingStart := time.Date(2026, 6, 23, 16, 0, 0, 0, time.UTC).UnixMilli()
chinaOverview, err := repository.QueryOverview(ctx, OverviewQuery{AppCode: "lalu", StatTZ: StatTZAsiaShanghai, StartMS: beijingStart, EndMS: beijingStart + int64(24*time.Hour/time.Millisecond), CountryID: 63})
if err != nil {
t.Fatalf("query china overview: %v", err)
}
if chinaOverview.RechargeUSDMinor != 90_000 || chinaOverview.StatTZ != StatTZAsiaShanghai {
t.Fatalf("scoped replay must write China stat_tz only: %+v", chinaOverview)
}
}
func TestQueryOverviewReturnsPreviousPeriodDeltaRates(t *testing.T) {
ctx := context.Background()
_, file, _, _ := runtime.Caller(0)

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

File diff suppressed because it is too large Load Diff

68
services/statistics-service/internal/storage/mysql/stat_tz.go Normal file
View File

@ -0,0 +1,68 @@
package mysql
import (
"context"
"strings"
"time"
)
const (
// StatTZUTC 是历史默认口径，保留它可以让旧接口、旧报表和跨服务 UTC 统计规则继续稳定工作。
StatTZUTC = "UTC"
// StatTZAsiaShanghai 是 Databi 需要的中国自然日口径；中国当前无夏令时，固定 +08:00 能避免容器缺 tzdata 时切日失败。
StatTZAsiaShanghai = "Asia/Shanghai"
)
var asiaShanghaiLocation = time.FixedZone(StatTZAsiaShanghai, 8*60*60)
type statDayScope struct {
tz string
day string
}
type statTZScopeContextKey struct{}
// WithStatTZScope 限制当前聚合只写一个 stat_tz；实时消费不使用它，回灌任务用它避免重放历史事件时污染 UTC 口径。
func WithStatTZScope(ctx context.Context, statTZ string) context.Context {
return context.WithValue(ctx, statTZScopeContextKey{}, normalizeStatTZ(statTZ))
}
func normalizeStatTZ(value string) string {
switch strings.TrimSpace(value) {
case StatTZAsiaShanghai:
return StatTZAsiaShanghai
case StatTZUTC, "":
return StatTZUTC
default:
return StatTZUTC
}
}
func statLocation(statTZ string) *time.Location {
if normalizeStatTZ(statTZ) == StatTZAsiaShanghai {
return asiaShanghaiLocation
}
return time.UTC
}
func statDayIn(ms int64, statTZ string) string {
return time.UnixMilli(ms).In(statLocation(statTZ)).Format("2006-01-02")
}
func statDayScopes(ms int64) []statDayScope {
// 同一条业务事实必须一次事务内写出两份读模型：UTC 行维持历史聚合，Asia/Shanghai 行给中国自然日大屏使用。
return []statDayScope{
{tz: StatTZUTC, day: statDayIn(ms, StatTZUTC)},
{tz: StatTZAsiaShanghai, day: statDayIn(ms, StatTZAsiaShanghai)},
}
}
func statDayScopesFromContext(ctx context.Context, ms int64) []statDayScope {
if ctx != nil {
if statTZ, ok := ctx.Value(statTZScopeContextKey{}).(string); ok {
statTZ = normalizeStatTZ(statTZ)
return []statDayScope{{tz: statTZ, day: statDayIn(ms, statTZ)}}
}
}
return statDayScopes(ms)
}