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

数据大屏

Browse Source
This commit is contained in:
zhx 2026-05-25 17:50:20 +08:00
parent 597b1cf587
commit 0bfe1cf57d
19 changed files with 558 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
api/proto/events/room/v1/events.pb.go
View File

@ -280,11 +280,13 @@ func (x *RoomProfileUpdated) GetSeatCount() int32 {
// RoomUserJoined 表达用户业务态进房成功。
type RoomUserJoined struct {
state protoimpl.MessageState `protogen:"open.v1"`
UserId int64 `protobuf:"varint,1,opt,name=user_id,json=userId,proto3" json:"user_id,omitempty"`
Role string `protobuf:"bytes,2,opt,name=role,proto3" json:"role,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
state protoimpl.MessageState `protogen:"open.v1"`
UserId int64 `protobuf:"varint,1,opt,name=user_id,json=userId,proto3" json:"user_id,omitempty"`
Role string `protobuf:"bytes,2,opt,name=role,proto3" json:"role,omitempty"`
// visible_region_id 是用户进房时所在房间的国家/区域桶，统计服务不能反查 room-service。
VisibleRegionId int64 `protobuf:"varint,3,opt,name=visible_region_id,json=visibleRegionId,proto3" json:"visible_region_id,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *RoomUserJoined) Reset() {
@ -331,6 +333,13 @@ func (x *RoomUserJoined) GetRole() string {
return ""
}
func (x *RoomUserJoined) GetVisibleRegionId() int64 {
if x != nil {
return x.VisibleRegionId
}
return 0
}
// RoomUserLeft 表达用户离房成功。
type RoomUserLeft struct {
state protoimpl.MessageState `protogen:"open.v1"`
@ -1757,10 +1766,11 @@ const file_proto_events_room_v1_events_proto_rawDesc = "" +
"roomAvatar\x12)\n" +
"\x10room_description\x18\x04 \x01(\tR\x0froomDescription\x12\x1d\n" +
"\n" +
"seat_count\x18\x05 \x01(\x05R\tseatCount\"=\n" +
"seat_count\x18\x05 \x01(\x05R\tseatCount\"i\n" +
"\x0eRoomUserJoined\x12\x17\n" +
"\auser_id\x18\x01 \x01(\x03R\x06userId\x12\x12\n" +
"\x04role\x18\x02 \x01(\tR\x04role\"'\n" +
"\x04role\x18\x02 \x01(\tR\x04role\x12*\n" +
"\x11visible_region_id\x18\x03 \x01(\x03R\x0fvisibleRegionId\"'\n" +
"\fRoomUserLeft\x12\x17\n" +
"\auser_id\x18\x01 \x01(\x03R\x06userId\"H\n" +
"\n" +

2
api/proto/events/room/v1/events.proto
View File

@ -40,6 +40,8 @@ message RoomProfileUpdated {
message RoomUserJoined {
int64 user_id = 1;
string role = 2;
// visible_region_id 是用户进房时所在房间的国家/区域桶，统计服务不能反查 room-service。
int64 visible_region_id = 3;
}
// RoomUserLeft 表达用户离房成功。

11
docker-compose.yml
View File

@ -4,17 +4,17 @@ x-go-build-args: &go-build-args
services:
rocketmq-namesrv:
image: apache/rocketmq:5.3.2
image: apache/rocketmq:5.3.1
container_name: rocketmq-namesrv
environment:
TZ: UTC
JAVA_OPT_EXT: "-Xms256m -Xmx256m -Xmn128m"
command: ["sh", "mqnamesrv"]
ports:
- "9876:9876"
- "19876:9876"
rocketmq-broker:
image: apache/rocketmq:5.3.2
image: apache/rocketmq:5.3.1
container_name: rocketmq-broker
user: root
environment:
@ -23,8 +23,8 @@ services:
JAVA_OPT_EXT: "-Xms512m -Xmx512m -Xmn256m"
command: ["sh", "mqbroker", "-n", "rocketmq-namesrv:9876", "-c", "/home/rocketmq/broker.conf"]
ports:
- "10909:10909"
- "10911:10911"
- "19009:10909"
- "19011:10911"
volumes:
- ./deploy/rocketmq/broker.conf:/home/rocketmq/broker.conf:ro
- rocketmq-store:/home/rocketmq/store
@ -249,6 +249,7 @@ services:
environment:
TZ: UTC
ports:
- "13010:13010"
- "13110:13110"
depends_on:
mysql:

2
server/admin/cmd/server/main.go
View File

@ -215,7 +215,7 @@ func main() {
CoinLedger: coinledgermodule.New(userDB, walletDB, sqlDB, walletclient.NewGRPC(walletConn), auditHandler),
CountryRegion: countryregionmodule.New(userclient.NewGRPC(userConn), userDB, cfg, auditHandler),
DailyTask: dailytaskmodule.New(activityclient.NewGRPC(activityConn), auditHandler),
Dashboard: dashboardmodule.New(store),
Dashboard: dashboardmodule.New(store, cfg),
FirstRechargeReward: firstrechargerewardmodule.New(activityclient.NewGRPC(activityConn), userDB, auditHandler),
Game: gamemanagementmodule.New(gameclient.NewGRPC(gameConn), auditHandler),
Health: healthmodule.New(store, redisClient, jobStatus),

3
server/admin/configs/config.tencent.example.yaml
View File

@ -48,6 +48,9 @@ activity_service:
game_service:
addr: "10.2.1.16:13008"
request_timeout: "3s"
statistics_service:
base_url: "http://10.2.1.16:13010"
request_timeout: "3s"
tencent-cos:
enabled: true
secret-id: "REPLACE_ME"

3
server/admin/configs/config.yaml
View File

@ -48,6 +48,9 @@ activity_service:
game_service:
addr: "127.0.0.1:13008"
request_timeout: "3s"
statistics_service:
base_url: "http://127.0.0.1:13010"
request_timeout: "3s"
tencent-cos:
enabled: true
secret-id: "IKIDMchhZEfrsiNo472DAtTpzzmLjttkOnyu"

73
server/admin/internal/config/config.go
View File

@ -13,31 +13,32 @@ import (
)
type Config struct {
ServiceName string `yaml:"service_name"`
NodeID string `yaml:"node_id"`
Environment string `yaml:"environment"`
Log logging.Config `yaml:"log"`
HTTPAddr string `yaml:"http_addr"`
MySQLDSN string `yaml:"mysql_dsn"`
UserMySQLDSN string `yaml:"user_mysql_dsn"`
WalletMySQLDSN string `yaml:"wallet_mysql_dsn"`
MySQLAutoMigrate bool `yaml:"mysql_auto_migrate"`
Migrations MigrationConfig `yaml:"migrations"`
JWTSecret string `yaml:"jwt_secret"`
AccessTokenTTL time.Duration `yaml:"access_token_ttl"`
RefreshTokenTTL time.Duration `yaml:"refresh_token_ttl"`
CORSAllowedOrigins []string `yaml:"cors_allowed_origins"`
RefreshCookieSecure bool `yaml:"refresh_cookie_secure"`
Bootstrap BootstrapConfig `yaml:"bootstrap"`
BootstrapPassword string `yaml:"bootstrap_password"`
UserService UserServiceConfig `yaml:"user_service"`
TencentCOS TencentCOSConfig `yaml:"tencent-cos"`
Redis RedisConfig `yaml:"redis"`
Jobs JobsConfig `yaml:"jobs"`
WalletService WalletServiceConfig `yaml:"wallet_service"`
RoomService RoomServiceConfig `yaml:"room_service"`
ActivityService ActivityServiceConfig `yaml:"activity_service"`
GameService GameServiceConfig `yaml:"game_service"`
ServiceName string `yaml:"service_name"`
NodeID string `yaml:"node_id"`
Environment string `yaml:"environment"`
Log logging.Config `yaml:"log"`
HTTPAddr string `yaml:"http_addr"`
MySQLDSN string `yaml:"mysql_dsn"`
UserMySQLDSN string `yaml:"user_mysql_dsn"`
WalletMySQLDSN string `yaml:"wallet_mysql_dsn"`
MySQLAutoMigrate bool `yaml:"mysql_auto_migrate"`
Migrations MigrationConfig `yaml:"migrations"`
JWTSecret string `yaml:"jwt_secret"`
AccessTokenTTL time.Duration `yaml:"access_token_ttl"`
RefreshTokenTTL time.Duration `yaml:"refresh_token_ttl"`
CORSAllowedOrigins []string `yaml:"cors_allowed_origins"`
RefreshCookieSecure bool `yaml:"refresh_cookie_secure"`
Bootstrap BootstrapConfig `yaml:"bootstrap"`
BootstrapPassword string `yaml:"bootstrap_password"`
UserService UserServiceConfig `yaml:"user_service"`
TencentCOS TencentCOSConfig `yaml:"tencent-cos"`
Redis RedisConfig `yaml:"redis"`
Jobs JobsConfig `yaml:"jobs"`
WalletService WalletServiceConfig `yaml:"wallet_service"`
RoomService RoomServiceConfig `yaml:"room_service"`
ActivityService ActivityServiceConfig `yaml:"activity_service"`
GameService GameServiceConfig `yaml:"game_service"`
StatisticsService StatisticsServiceConfig `yaml:"statistics_service"`
}
type MigrationConfig struct {
@ -71,6 +72,11 @@ type GameServiceConfig struct {
RequestTimeout time.Duration `yaml:"request_timeout"`
}
type StatisticsServiceConfig struct {
BaseURL string `yaml:"base_url"`
RequestTimeout time.Duration `yaml:"request_timeout"`
}
type TencentCOSConfig struct {
Enabled bool `yaml:"enabled"`
SecretID string `yaml:"secret-id"`
@ -159,6 +165,10 @@ func Default() Config {
Addr: "127.0.0.1:13008",
RequestTimeout: 3 * time.Second,
},
StatisticsService: StatisticsServiceConfig{
BaseURL: "http://127.0.0.1:13010",
RequestTimeout: 3 * time.Second,
},
TencentCOS: TencentCOSConfig{
Enabled: false,
ObjectPrefix: "admin",
@ -267,6 +277,13 @@ func (cfg *Config) Normalize() {
if cfg.GameService.RequestTimeout <= 0 {
cfg.GameService.RequestTimeout = 3 * time.Second
}
cfg.StatisticsService.BaseURL = strings.TrimRight(strings.TrimSpace(cfg.StatisticsService.BaseURL), "/")
if cfg.StatisticsService.BaseURL == "" {
cfg.StatisticsService.BaseURL = "http://127.0.0.1:13010"
}
if cfg.StatisticsService.RequestTimeout <= 0 {
cfg.StatisticsService.RequestTimeout = 3 * time.Second
}
cfg.TencentCOS.SecretID = strings.TrimSpace(cfg.TencentCOS.SecretID)
cfg.TencentCOS.SecretKey = strings.TrimSpace(cfg.TencentCOS.SecretKey)
cfg.TencentCOS.BucketName = strings.TrimSpace(cfg.TencentCOS.BucketName)
@ -362,6 +379,12 @@ func (cfg Config) Validate() error {
if cfg.GameService.RequestTimeout <= 0 {
return errors.New("game_service.request_timeout must be greater than 0")
}
if strings.TrimSpace(cfg.StatisticsService.BaseURL) == "" {
return errors.New("statistics_service.base_url is required")
}
if cfg.StatisticsService.RequestTimeout <= 0 {
return errors.New("statistics_service.request_timeout must be greater than 0")
}
if cfg.TencentCOS.Enabled {
if cfg.TencentCOS.SecretID == "" || cfg.TencentCOS.SecretKey == "" || cfg.TencentCOS.BucketName == "" || cfg.TencentCOS.Region == "" || cfg.TencentCOS.AccessURL == "" {
return errors.New("tencent-cos config is incomplete")

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

@ -1,7 +1,10 @@
package dashboard
import (
"strconv"
"github.com/gin-gonic/gin"
"hyapp-admin-server/internal/config"
"hyapp-admin-server/internal/repository"
"hyapp-admin-server/internal/response"
)
@ -10,8 +13,8 @@ type Handler struct {
service *DashboardService
}
func New(store *repository.Store) *Handler {
return &Handler{service: NewService(store)}
func New(store *repository.Store, cfg config.Config) *Handler {
return &Handler{service: NewService(store, cfg)}
}
func (h *Handler) DashboardOverview(c *gin.Context) {
@ -22,3 +25,22 @@ func (h *Handler) DashboardOverview(c *gin.Context) {
}
response.OK(c, overview)
}
func (h *Handler) StatisticsOverview(c *gin.Context) {
overview, err := h.service.StatisticsOverview(c.Request.Context(), StatisticsQuery{
AppCode: c.DefaultQuery("app_code", "lalu"),
StartMS: parseInt64(c.Query("start_ms")),
EndMS: parseInt64(c.Query("end_ms")),
CountryID: parseInt64(c.Query("country_id")),
})
if err != nil {
response.ServerError(c, "获取统计总览失败")
return
}
response.OK(c, overview)
}
func parseInt64(value string) int64 {
parsed, _ := strconv.ParseInt(value, 10, 64)
return parsed
}

1
server/admin/internal/modules/dashboard/routes.go
View File

@ -8,4 +8,5 @@ import (
func RegisterRoutes(protected *gin.RouterGroup, h *Handler) {
protected.GET("/dashboard/overview", middleware.RequirePermission("overview:view"), h.DashboardOverview)
protected.GET("/statistics/overview", middleware.RequirePermission("overview:view"), h.StatisticsOverview)
}

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

@ -1,15 +1,70 @@
package dashboard
import "hyapp-admin-server/internal/repository"
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/url"
"strconv"
"hyapp-admin-server/internal/config"
"hyapp-admin-server/internal/repository"
)
type DashboardService struct {
store *repository.Store
store *repository.Store
cfg config.Config
httpClient *http.Client
}
func NewService(store *repository.Store) *DashboardService {
return &DashboardService{store: store}
type StatisticsQuery struct {
AppCode string
StartMS int64
EndMS int64
CountryID int64
}
func NewService(store *repository.Store, cfg config.Config) *DashboardService {
return &DashboardService{
store: store,
cfg: cfg,
httpClient: &http.Client{Timeout: cfg.StatisticsService.RequestTimeout},
}
}
func (s *DashboardService) Overview() (*repository.DashboardOverview, error) {
return s.store.DashboardOverview()
}
func (s *DashboardService) StatisticsOverview(ctx context.Context, query StatisticsQuery) (map[string]any, error) {
baseURL := s.cfg.StatisticsService.BaseURL + "/internal/v1/statistics/overview"
values := url.Values{}
values.Set("app_code", query.AppCode)
if query.StartMS > 0 {
values.Set("start_ms", strconv.FormatInt(query.StartMS, 10))
}
if query.EndMS > 0 {
values.Set("end_ms", strconv.FormatInt(query.EndMS, 10))
}
if query.CountryID > 0 {
values.Set("country_id", strconv.FormatInt(query.CountryID, 10))
}
req, err := http.NewRequestWithContext(ctx, http.MethodGet, baseURL+"?"+values.Encode(), nil)
if err != nil {
return nil, err
}
resp, err := s.httpClient.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return nil, fmt.Errorf("statistics service returned status %d", resp.StatusCode)
}
var out map[string]any
if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
return nil, err
}
return out, nil
}

5
services/room-service/internal/room/service/presence.go
View File

@ -69,8 +69,9 @@ func (s *Service) JoinRoom(ctx context.Context, req *roomv1.JoinRoomRequest) (*r
// JoinRoom 成功需要 outbox 事件，腾讯云 IM 房间系统消息由 worker 异步投递。
joinedEvent, err := outbox.Build(current.RoomID, "RoomUserJoined", current.Version, now, &roomeventsv1.RoomUserJoined{
UserId: cmd.ActorUserID(),
Role: cmd.Role,
UserId: cmd.ActorUserID(),
Role: cmd.Role,
VisibleRegionId: current.VisibleRegionID,
})
if err != nil {
return mutationResult{}, nil, err

1
services/statistics-service/configs/config.docker.yaml
View File

@ -2,6 +2,7 @@ service_name: statistics-service
node_id: statistics-docker
environment: local
health_http_addr: ":13110"
query_http_addr: ":13010"
mysql_dsn: "hyapp:hyapp@tcp(mysql:3306)/hyapp_statistics?parseTime=true&charset=utf8mb4&loc=UTC&multiStatements=true"
rocketmq:
enabled: true

1
services/statistics-service/configs/config.yaml
View File

@ -2,6 +2,7 @@ service_name: statistics-service
node_id: statistics-local
environment: local
health_http_addr: ":13110"
query_http_addr: ":13010"
mysql_dsn: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_statistics?parseTime=true&charset=utf8mb4&loc=UTC&multiStatements=true"
rocketmq:
enabled: false

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

@ -23,6 +23,7 @@ CREATE TABLE IF NOT EXISTS stat_app_day_country (
country_id BIGINT NOT NULL DEFAULT 0 COMMENT '国家或区域维度 ID，0 表示未知或全局',
new_users BIGINT NOT NULL DEFAULT 0 COMMENT '新增用户数',
active_users BIGINT NOT NULL DEFAULT 0 COMMENT '当日活跃去重用户数',
paid_users BIGINT NOT NULL DEFAULT 0 COMMENT '当日充值去重用户数',
recharge_usd_minor BIGINT NOT NULL DEFAULT 0 COMMENT '总充值美元最小单位',
new_user_recharge_usd_minor BIGINT NOT NULL DEFAULT 0 COMMENT '新增充值美元最小单位',
coin_seller_recharge_usd_minor BIGINT NOT NULL DEFAULT 0 COMMENT '币商充值美元最小单位',
@ -34,6 +35,7 @@ CREATE TABLE IF NOT EXISTS stat_app_day_country (
game_turnover BIGINT NOT NULL DEFAULT 0 COMMENT '游戏流水',
game_players BIGINT NOT NULL DEFAULT 0 COMMENT '游戏参与去重人数',
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)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='App 国家日统计聚合表';
@ -48,6 +50,26 @@ CREATE TABLE IF NOT EXISTS stat_user_day_activity (
KEY idx_stat_user_day_country (app_code, stat_day, country_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,
user_id BIGINT NOT NULL,
registered_day DATE NOT NULL,
country_id BIGINT NOT NULL DEFAULT 0,
registered_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, user_id),
KEY idx_stat_user_registration_day (app_code, registered_day, country_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='用户注册日留存 cohort 表';
CREATE TABLE IF NOT EXISTS stat_recharge_day_payers (
app_code VARCHAR(32) NOT NULL,
stat_day DATE NOT NULL,
country_id BIGINT NOT NULL DEFAULT 0,
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)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='充值日付费用户去重表';
CREATE TABLE IF NOT EXISTS stat_lucky_gift_day_payers (
app_code VARCHAR(32) NOT NULL,
stat_day DATE NOT NULL,
@ -66,6 +88,7 @@ CREATE TABLE IF NOT EXISTS stat_game_day_country (
game_id VARCHAR(96) NOT NULL,
turnover_coin BIGINT NOT NULL DEFAULT 0,
payout_coin BIGINT NOT NULL DEFAULT 0,
refund_coin BIGINT NOT NULL DEFAULT 0,
player_count BIGINT NOT NULL DEFAULT 0,
updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, stat_day, country_id, platform_code, game_id)

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

@ -28,6 +28,7 @@ type App struct {
repo *mysqlstorage.Repository
health *grpchealth.ServingChecker
healthHTTP *healthhttp.Server
queryHTTP *queryHTTPServer
consumers []*rocketmqx.Consumer
closeOnce sync.Once
}
@ -51,11 +52,19 @@ func New(cfg config.Config) (*App, error) {
_ = repo.Close()
return nil, err
}
return &App{cfg: cfg, repo: repo, health: health, healthHTTP: healthHTTP, consumers: consumers}, nil
queryHTTP, err := newQueryHTTPServer(cfg.QueryHTTPAddr, repo)
if err != nil {
shutdownConsumers(consumers)
_ = healthHTTP.Close(context.Background())
_ = repo.Close()
return nil, err
}
return &App{cfg: cfg, repo: repo, health: health, healthHTTP: healthHTTP, queryHTTP: queryHTTP, consumers: consumers}, nil
}
func (a *App) Run() error {
a.runHealthHTTP()
a.runQueryHTTP()
if err := startConsumers(a.consumers); err != nil {
a.health.MarkStopped()
return err
@ -64,6 +73,17 @@ func (a *App) Run() error {
select {}
}
func (a *App) runQueryHTTP() {
if a.queryHTTP == nil {
return
}
go func() {
if err := a.queryHTTP.Run(); err != nil {
logx.Error(context.Background(), "query_http_run_failed", err)
}
}()
}
func (a *App) Close() {
a.closeOnce.Do(func() {
a.health.MarkDraining()
@ -93,6 +113,9 @@ func (a *App) closeHealthHTTP() {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
_ = a.healthHTTP.Close(ctx)
if a.queryHTTP != nil {
_ = a.queryHTTP.Close(ctx)
}
}
func newConsumers(cfg config.Config, repo *mysqlstorage.Repository) ([]*rocketmqx.Consumer, error) {
@ -250,7 +273,7 @@ func roomJoinActiveEvent(body []byte) (mysqlstorage.UserActiveEvent, bool, error
EventID: envelope.GetEventId(),
EventType: envelope.GetEventType(),
UserID: joined.GetUserId(),
CountryID: 0,
CountryID: joined.GetVisibleRegionId(),
OccurredAtMS: envelope.GetOccurredAtMs(),
}, true, nil
}

89
services/statistics-service/internal/app/query_http.go Normal file
View File

@ -0,0 +1,89 @@
package app
import (
"context"
"encoding/json"
"errors"
"net"
"net/http"
"strconv"
"strings"
"time"
mysqlstorage "hyapp/services/statistics-service/internal/storage/mysql"
)
type queryHTTPServer struct {
repo *mysqlstorage.Repository
server *http.Server
listener net.Listener
}
func newQueryHTTPServer(addr string, repo *mysqlstorage.Repository) (*queryHTTPServer, error) {
addr = strings.TrimSpace(addr)
if addr == "" {
return nil, nil
}
listener, err := net.Listen("tcp", addr)
if err != nil {
return nil, err
}
s := &queryHTTPServer{repo: repo, listener: listener}
mux := http.NewServeMux()
mux.HandleFunc("/internal/v1/statistics/overview", s.overview)
s.server = &http.Server{Handler: mux, ReadHeaderTimeout: 2 * time.Second}
return s, nil
}
func (s *queryHTTPServer) Run() error {
if s == nil {
return nil
}
err := s.server.Serve(s.listener)
if errors.Is(err, http.ErrServerClosed) {
return nil
}
return err
}
func (s *queryHTTPServer) Close(ctx context.Context) error {
if s == nil {
return nil
}
return s.server.Shutdown(ctx)
}
func (s *queryHTTPServer) overview(w http.ResponseWriter, r *http.Request) {
query := r.URL.Query()
startMS := parseInt64(query.Get("start_ms"))
endMS := parseInt64(query.Get("end_ms"))
if startMS <= 0 {
now := time.Now().UTC()
startMS = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC).UnixMilli()
}
if endMS <= startMS {
endMS = startMS + 24*time.Hour.Milliseconds()
}
overview, err := s.repo.QueryOverview(r.Context(), mysqlstorage.OverviewQuery{
AppCode: query.Get("app_code"),
StartMS: startMS,
EndMS: endMS,
CountryID: parseInt64(query.Get("country_id")),
})
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
return
}
writeJSON(w, http.StatusOK, overview)
}
func parseInt64(value string) int64 {
parsed, _ := strconv.ParseInt(strings.TrimSpace(value), 10, 64)
return parsed
}
func writeJSON(w http.ResponseWriter, status int, payload any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(payload)
}

6
services/statistics-service/internal/config/config.go
View File

@ -14,6 +14,7 @@ type Config struct {
NodeID string `yaml:"node_id"`
Environment string `yaml:"environment"`
HealthHTTPAddr string `yaml:"health_http_addr"`
QueryHTTPAddr string `yaml:"query_http_addr"`
MySQLDSN string `yaml:"mysql_dsn"`
RocketMQ RocketMQConfig `yaml:"rocketmq"`
Log logx.Config `yaml:"log"`
@ -47,6 +48,7 @@ func Default() Config {
NodeID: "statistics-local",
Environment: "local",
HealthHTTPAddr: ":13110",
QueryHTTPAddr: ":13010",
MySQLDSN: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_statistics?parseTime=true&charset=utf8mb4&loc=UTC&multiStatements=true",
RocketMQ: defaultRocketMQConfig(),
Log: logx.Config{
@ -80,6 +82,10 @@ func Load(path string) (Config, error) {
if cfg.HealthHTTPAddr == "" {
cfg.HealthHTTPAddr = ":13110"
}
cfg.QueryHTTPAddr = strings.TrimSpace(cfg.QueryHTTPAddr)
if cfg.QueryHTTPAddr == "" {
cfg.QueryHTTPAddr = ":13010"
}
cfg.RocketMQ = normalizeRocketMQConfig(cfg.RocketMQ)
if cfg.RocketMQ.Enabled {
if len(cfg.RocketMQ.NameServers) == 0 && cfg.RocketMQ.NameServerDomain == "" {

193
services/statistics-service/internal/storage/mysql/query.go Normal file
View File

@ -0,0 +1,193 @@
package mysql
import (
"context"
"database/sql"
"time"
"hyapp/pkg/appcode"
)
type OverviewQuery struct {
AppCode string
StartMS int64
EndMS int64
CountryID int64
}
type Overview struct {
AppCode string `json:"app_code"`
StartMS int64 `json:"start_ms"`
EndMS int64 `json:"end_ms"`
CountryID int64 `json:"country_id"`
NewUsers int64 `json:"new_users"`
ActiveUsers int64 `json:"active_users"`
PaidUsers int64 `json:"paid_users"`
RechargeUSDMinor int64 `json:"recharge_usd_minor"`
NewUserRechargeUSDMinor int64 `json:"new_user_recharge_usd_minor"`
CoinSellerRechargeUSDMinor int64 `json:"coin_seller_recharge_usd_minor"`
MifaPayRechargeUSDMinor int64 `json:"mifapay_recharge_usd_minor"`
GoogleRechargeUSDMinor int64 `json:"google_recharge_usd_minor"`
GiftCoinSpent int64 `json:"gift_coin_spent"`
LuckyGiftTurnover int64 `json:"lucky_gift_turnover"`
LuckyGiftPayers int64 `json:"lucky_gift_payers"`
GameTurnover int64 `json:"game_turnover"`
GamePayout int64 `json:"game_payout"`
GameRefund int64 `json:"game_refund"`
GamePlayers int64 `json:"game_players"`
GameProfit int64 `json:"game_profit"`
GameProfitRate float64 `json:"game_profit_rate"`
ARPUUSDMinor int64 `json:"arpu_usd_minor"`
ARPPUUSDMinor int64 `json:"arppu_usd_minor"`
UPValueUSDMinor int64 `json:"up_value_usd_minor"`
Retention Retention `json:"retention"`
GameRanking []GameRank `json:"game_ranking"`
}
type Retention struct {
RegisteredUsers int64 `json:"registered_users"`
Day1Users int64 `json:"day1_users"`
Day7Users int64 `json:"day7_users"`
Day30Users int64 `json:"day30_users"`
Day1Rate float64 `json:"day1_rate"`
Day7Rate float64 `json:"day7_rate"`
Day30Rate float64 `json:"day30_rate"`
}
type GameRank struct {
PlatformCode string `json:"platform_code"`
GameID string `json:"game_id"`
TurnoverCoin int64 `json:"turnover_coin"`
PayoutCoin int64 `json:"payout_coin"`
RefundCoin int64 `json:"refund_coin"`
PlayerCount int64 `json:"player_count"`
ProfitCoin int64 `json:"profit_coin"`
ProfitRate float64 `json:"profit_rate"`
}
func (r *Repository) QueryOverview(ctx context.Context, query OverviewQuery) (Overview, error) {
app := appcode.Normalize(query.AppCode)
if app == "" {
app = "lalu"
}
startDay := statDay(query.StartMS)
endDay := statDay(query.EndMS - 1)
args := []any{app, startDay, endDay}
countryFilter := ""
if query.CountryID > 0 {
countryFilter = " AND country_id = ?"
args = append(args, query.CountryID)
}
row := r.db.QueryRowContext(ctx, `
SELECT COALESCE(SUM(new_users),0), COALESCE(SUM(active_users),0), COALESCE(SUM(paid_users),0),
COALESCE(SUM(recharge_usd_minor),0), COALESCE(SUM(new_user_recharge_usd_minor),0),
COALESCE(SUM(coin_seller_recharge_usd_minor),0), COALESCE(SUM(mifapay_recharge_usd_minor),0),
COALESCE(SUM(google_recharge_usd_minor),0), COALESCE(SUM(gift_coin_spent),0),
COALESCE(SUM(lucky_gift_turnover),0), COALESCE(SUM(lucky_gift_payers),0),
COALESCE(SUM(game_turnover),0), COALESCE(SUM(game_payout),0), COALESCE(SUM(game_refund),0),
COALESCE(SUM(game_players),0)
FROM stat_app_day_country
WHERE app_code = ? AND stat_day BETWEEN ? AND ?`+countryFilter, args...)
var overview Overview
overview.AppCode, overview.StartMS, overview.EndMS, overview.CountryID = app, query.StartMS, query.EndMS, query.CountryID
if err := row.Scan(&overview.NewUsers, &overview.ActiveUsers, &overview.PaidUsers, &overview.RechargeUSDMinor, &overview.NewUserRechargeUSDMinor, &overview.CoinSellerRechargeUSDMinor, &overview.MifaPayRechargeUSDMinor, &overview.GoogleRechargeUSDMinor, &overview.GiftCoinSpent, &overview.LuckyGiftTurnover, &overview.LuckyGiftPayers, &overview.GameTurnover, &overview.GamePayout, &overview.GameRefund, &overview.GamePlayers); err != nil {
return Overview{}, err
}
overview.GameProfit = overview.GameTurnover - overview.GamePayout - overview.GameRefund
overview.GameProfitRate = ratio(overview.GameProfit, overview.GameTurnover)
overview.ARPUUSDMinor = div(overview.RechargeUSDMinor, overview.ActiveUsers)
overview.ARPPUUSDMinor = div(overview.RechargeUSDMinor, overview.PaidUsers)
overview.UPValueUSDMinor = overview.ARPPUUSDMinor
retention, err := r.queryRetention(ctx, app, startDay, endDay, query.CountryID)
if err != nil {
return Overview{}, err
}
overview.Retention = retention
ranking, err := r.queryGameRanking(ctx, app, startDay, endDay, query.CountryID)
if err != nil {
return Overview{}, err
}
overview.GameRanking = ranking
return overview, nil
}
func (r *Repository) queryRetention(ctx context.Context, app, startDay, endDay string, countryID int64) (Retention, error) {
filter := ""
args := []any{app, startDay, endDay}
if countryID > 0 {
filter = " AND r.country_id = ?"
args = append(args, countryID)
}
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)
FROM stat_user_registration r
WHERE r.app_code = ? 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
}
out.Day1Rate = ratio(out.Day1Users, out.RegisteredUsers)
out.Day7Rate = ratio(out.Day7Users, out.RegisteredUsers)
out.Day30Rate = ratio(out.Day30Users, out.RegisteredUsers)
return out, nil
}
func (r *Repository) queryGameRanking(ctx context.Context, app, startDay, endDay string, countryID int64) ([]GameRank, error) {
args := []any{app, startDay, endDay}
filter := ""
if countryID > 0 {
filter = " AND country_id = ?"
args = append(args, countryID)
}
rows, err := r.db.QueryContext(ctx, `
SELECT platform_code, game_id, COALESCE(SUM(turnover_coin),0), COALESCE(SUM(payout_coin),0),
COALESCE(SUM(refund_coin),0), COALESCE(SUM(player_count),0)
FROM stat_game_day_country
WHERE app_code = ? AND stat_day BETWEEN ? AND ?`+filter+`
GROUP BY platform_code, game_id
ORDER BY COALESCE(SUM(turnover_coin),0) DESC
LIMIT 50`, args...)
if err != nil {
return nil, err
}
defer rows.Close()
out := []GameRank{}
for rows.Next() {
var item GameRank
if err := rows.Scan(&item.PlatformCode, &item.GameID, &item.TurnoverCoin, &item.PayoutCoin, &item.RefundCoin, &item.PlayerCount); err != nil {
return nil, err
}
item.ProfitCoin = item.TurnoverCoin - item.PayoutCoin - item.RefundCoin
item.ProfitRate = ratio(item.ProfitCoin, item.TurnoverCoin)
out = append(out, item)
}
return out, rows.Err()
}
func normalizeRange(startMS, endMS int64) (int64, int64) {
if startMS <= 0 {
now := time.Now().UTC()
startMS = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC).UnixMilli()
}
if endMS <= startMS {
endMS = startMS + 24*time.Hour.Milliseconds()
}
return startMS, endMS
}
func div(numerator, denominator int64) int64 {
if denominator <= 0 {
return 0
}
return numerator / denominator
}
func ratio(numerator, denominator int64) float64 {
if denominator <= 0 {
return 0
}
return float64(numerator) / float64(denominator)
}

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

@ -71,13 +71,13 @@ func (r *Repository) Migrate(ctx context.Context) error {
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
`CREATE TABLE IF NOT EXISTS stat_app_day_country (
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
new_users BIGINT NOT NULL DEFAULT 0, active_users BIGINT NOT NULL DEFAULT 0,
new_users BIGINT NOT NULL DEFAULT 0, active_users BIGINT NOT NULL DEFAULT 0, paid_users BIGINT NOT NULL DEFAULT 0,
recharge_usd_minor BIGINT NOT NULL DEFAULT 0, new_user_recharge_usd_minor BIGINT NOT NULL DEFAULT 0,
coin_seller_recharge_usd_minor BIGINT NOT NULL DEFAULT 0, mifapay_recharge_usd_minor BIGINT NOT NULL DEFAULT 0,
google_recharge_usd_minor BIGINT NOT NULL DEFAULT 0, gift_coin_spent BIGINT NOT NULL DEFAULT 0,
lucky_gift_turnover BIGINT NOT NULL DEFAULT 0, lucky_gift_payers BIGINT NOT NULL DEFAULT 0,
game_turnover BIGINT NOT NULL DEFAULT 0, game_players BIGINT NOT NULL DEFAULT 0,
game_payout BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL,
game_payout BIGINT NOT NULL DEFAULT 0, game_refund BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, stat_day, country_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
`CREATE TABLE IF NOT EXISTS stat_user_day_activity (
@ -86,6 +86,16 @@ func (r *Repository) Migrate(ctx context.Context) error {
PRIMARY KEY (app_code, stat_day, user_id),
KEY idx_stat_user_day_country (app_code, stat_day, country_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
`CREATE TABLE IF NOT EXISTS stat_user_registration (
app_code VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, registered_day DATE NOT NULL,
country_id BIGINT NOT NULL DEFAULT 0, registered_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, user_id), KEY idx_stat_user_registration_day (app_code, registered_day, country_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
`CREATE TABLE IF NOT EXISTS stat_recharge_day_payers (
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
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)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
`CREATE TABLE IF NOT EXISTS stat_lucky_gift_day_payers (
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
user_id BIGINT NOT NULL, first_paid_at_ms BIGINT NOT NULL,
@ -95,7 +105,7 @@ func (r *Repository) Migrate(ctx context.Context) error {
`CREATE TABLE IF NOT EXISTS stat_game_day_country (
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
platform_code VARCHAR(64) NOT NULL DEFAULT '', game_id VARCHAR(96) NOT NULL,
turnover_coin BIGINT NOT NULL DEFAULT 0, payout_coin BIGINT NOT NULL DEFAULT 0,
turnover_coin BIGINT NOT NULL DEFAULT 0, payout_coin BIGINT NOT NULL DEFAULT 0, refund_coin BIGINT NOT NULL DEFAULT 0,
player_count BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, stat_day, country_id, platform_code, game_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
@ -111,6 +121,16 @@ func (r *Repository) Migrate(ctx context.Context) error {
return err
}
}
alterStatements := []string{
`ALTER TABLE stat_app_day_country ADD COLUMN paid_users BIGINT NOT NULL DEFAULT 0 AFTER active_users`,
`ALTER TABLE stat_app_day_country ADD COLUMN game_refund BIGINT NOT NULL DEFAULT 0 AFTER game_payout`,
`ALTER TABLE stat_game_day_country ADD COLUMN refund_coin BIGINT NOT NULL DEFAULT 0 AFTER payout_coin`,
}
for _, statement := range alterStatements {
if _, err := r.db.ExecContext(ctx, statement); err != nil && !isDuplicateColumnError(err) {
return err
}
}
return nil
}
@ -172,6 +192,12 @@ func (r *Repository) ConsumeUserRegistered(ctx context.Context, event UserRegist
if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, nowMS); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `
INSERT IGNORE INTO stat_user_registration (app_code, user_id, registered_day, country_id, registered_at_ms)
VALUES (?, ?, ?, ?, ?)
`, appcode.Normalize(event.AppCode), event.UserID, day, countryID, event.OccurredAtMS); err != nil {
return err
}
_, err := tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET new_users = new_users + 1, updated_at_ms = ?
@ -188,6 +214,13 @@ func (r *Repository) ConsumeRecharge(ctx context.Context, event RechargeEvent) e
if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, nowMS); err != nil {
return err
}
paidUsers, err := insertUnique(ctx, tx, `
INSERT IGNORE INTO stat_recharge_day_payers (app_code, stat_day, country_id, user_id, first_paid_at_ms)
VALUES (?, ?, ?, ?, ?)
`, appcode.Normalize(event.AppCode), day, countryID, event.UserID, event.OccurredAtMS)
if err != nil {
return err
}
newUserUSD := int64(0)
if event.RechargeSequence == 1 {
newUserUSD = event.USDMinor
@ -201,16 +234,17 @@ func (r *Repository) ConsumeRecharge(ctx context.Context, event RechargeEvent) e
default:
coinSellerUSD = event.USDMinor
}
_, err := tx.ExecContext(ctx, `
_, err = tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET recharge_usd_minor = recharge_usd_minor + ?,
new_user_recharge_usd_minor = new_user_recharge_usd_minor + ?,
coin_seller_recharge_usd_minor = coin_seller_recharge_usd_minor + ?,
mifapay_recharge_usd_minor = mifapay_recharge_usd_minor + ?,
google_recharge_usd_minor = google_recharge_usd_minor + ?,
paid_users = paid_users + ?,
updated_at_ms = ?
WHERE app_code = ? AND stat_day = ? AND country_id = ?
`, event.USDMinor, newUserUSD, coinSellerUSD, mifapayUSD, googleUSD, nowMS, appcode.Normalize(event.AppCode), day, countryID)
`, event.USDMinor, newUserUSD, coinSellerUSD, mifapayUSD, googleUSD, paidUsers, nowMS, appcode.Normalize(event.AppCode), day, countryID)
return err
})
}
@ -260,12 +294,14 @@ func (r *Repository) ConsumeGameOrder(ctx context.Context, event GameOrderEvent)
if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, nowMS); err != nil {
return err
}
turnover, payout := int64(0), int64(0)
turnover, payout, refund := int64(0), int64(0), int64(0)
switch strings.ToLower(strings.TrimSpace(event.OpType)) {
case "debit", "reverse":
case "debit", "bet":
turnover = event.CoinAmount
case "credit", "refund":
case "credit", "payout":
payout = event.CoinAmount
case "refund", "reverse":
refund = event.CoinAmount
}
playerDelta := int64(0)
if turnover > 0 {
@ -283,18 +319,18 @@ func (r *Repository) ConsumeGameOrder(ctx context.Context, event GameOrderEvent)
}
if _, err := tx.ExecContext(ctx, `
UPDATE stat_game_day_country
SET turnover_coin = turnover_coin + ?, payout_coin = payout_coin + ?,
SET turnover_coin = turnover_coin + ?, payout_coin = payout_coin + ?, refund_coin = refund_coin + ?,
player_count = player_count + ?, updated_at_ms = ?
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND platform_code = ? AND game_id = ?
`, turnover, payout, playerDelta, nowMS, appcode.Normalize(event.AppCode), day, countryID, event.PlatformCode, event.GameID); err != nil {
`, turnover, payout, refund, playerDelta, nowMS, appcode.Normalize(event.AppCode), day, countryID, event.PlatformCode, event.GameID); err != nil {
return err
}
_, err := tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET game_turnover = game_turnover + ?, game_payout = game_payout + ?,
SET game_turnover = game_turnover + ?, game_payout = game_payout + ?, game_refund = game_refund + ?,
game_players = game_players + ?, updated_at_ms = ?
WHERE app_code = ? AND stat_day = ? AND country_id = ?
`, turnover, payout, playerDelta, nowMS, appcode.Normalize(event.AppCode), day, countryID)
`, turnover, payout, refund, playerDelta, nowMS, appcode.Normalize(event.AppCode), day, countryID)
return err
})
}
@ -364,6 +400,11 @@ func isRetryableMySQLError(err error) bool {
return mysqlErr.Number == 1213 || mysqlErr.Number == 1205
}
func isDuplicateColumnError(err error) bool {
var mysqlErr *mysqlerr.MySQLError
return errors.As(err, &mysqlErr) && mysqlErr.Number == 1060
}
func applyActive(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, userID int64, occurredAtMS int64, nowMS int64) error {
if err := ensureAppDay(ctx, tx, app, day, countryID, nowMS); err != nil {
return err