diff --git a/api/proto/events/room/v1/events.pb.go b/api/proto/events/room/v1/events.pb.go index 55e15b71..5276fc0b 100644 --- a/api/proto/events/room/v1/events.pb.go +++ b/api/proto/events/room/v1/events.pb.go @@ -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" + diff --git a/api/proto/events/room/v1/events.proto b/api/proto/events/room/v1/events.proto index 9b004c17..d41de4e5 100644 --- a/api/proto/events/room/v1/events.proto +++ b/api/proto/events/room/v1/events.proto @@ -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 表达用户离房成功。 diff --git a/docker-compose.yml b/docker-compose.yml index 37c4a726..9b36b36a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -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: diff --git a/server/admin/cmd/server/main.go b/server/admin/cmd/server/main.go index 325e3754..433a24e6 100644 --- a/server/admin/cmd/server/main.go +++ b/server/admin/cmd/server/main.go @@ -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), diff --git a/server/admin/configs/config.tencent.example.yaml b/server/admin/configs/config.tencent.example.yaml index c0463d37..2fa38b81 100644 --- a/server/admin/configs/config.tencent.example.yaml +++ b/server/admin/configs/config.tencent.example.yaml @@ -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" diff --git a/server/admin/configs/config.yaml b/server/admin/configs/config.yaml index 08bbb22e..f342f18a 100644 --- a/server/admin/configs/config.yaml +++ b/server/admin/configs/config.yaml @@ -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" diff --git a/server/admin/internal/config/config.go b/server/admin/internal/config/config.go index 6d17f9be..fc41004d 100644 --- a/server/admin/internal/config/config.go +++ b/server/admin/internal/config/config.go @@ -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") diff --git a/server/admin/internal/modules/dashboard/handler.go b/server/admin/internal/modules/dashboard/handler.go index 8b28e53f..def86ae8 100644 --- a/server/admin/internal/modules/dashboard/handler.go +++ b/server/admin/internal/modules/dashboard/handler.go @@ -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 +} diff --git a/server/admin/internal/modules/dashboard/routes.go b/server/admin/internal/modules/dashboard/routes.go index 74ad3955..8edd6f0d 100644 --- a/server/admin/internal/modules/dashboard/routes.go +++ b/server/admin/internal/modules/dashboard/routes.go @@ -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) } diff --git a/server/admin/internal/modules/dashboard/service.go b/server/admin/internal/modules/dashboard/service.go index 36a64580..6f70dbe5 100644 --- a/server/admin/internal/modules/dashboard/service.go +++ b/server/admin/internal/modules/dashboard/service.go @@ -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 +} diff --git a/services/room-service/internal/room/service/presence.go b/services/room-service/internal/room/service/presence.go index d5d68142..e347923d 100644 --- a/services/room-service/internal/room/service/presence.go +++ b/services/room-service/internal/room/service/presence.go @@ -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 diff --git a/services/statistics-service/configs/config.docker.yaml b/services/statistics-service/configs/config.docker.yaml index 5a5df07d..b46a2e93 100644 --- a/services/statistics-service/configs/config.docker.yaml +++ b/services/statistics-service/configs/config.docker.yaml @@ -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 diff --git a/services/statistics-service/configs/config.yaml b/services/statistics-service/configs/config.yaml index eee34bc6..4c47bf6b 100644 --- a/services/statistics-service/configs/config.yaml +++ b/services/statistics-service/configs/config.yaml @@ -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 diff --git a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql index 8c4ec284..75ed90da 100644 --- a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql +++ b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql @@ -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) diff --git a/services/statistics-service/internal/app/app.go b/services/statistics-service/internal/app/app.go index cacbffe2..36d19fac 100644 --- a/services/statistics-service/internal/app/app.go +++ b/services/statistics-service/internal/app/app.go @@ -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 } diff --git a/services/statistics-service/internal/app/query_http.go b/services/statistics-service/internal/app/query_http.go new file mode 100644 index 00000000..f1e505be --- /dev/null +++ b/services/statistics-service/internal/app/query_http.go @@ -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) +} diff --git a/services/statistics-service/internal/config/config.go b/services/statistics-service/internal/config/config.go index fa08a1ff..414b5d87 100644 --- a/services/statistics-service/internal/config/config.go +++ b/services/statistics-service/internal/config/config.go @@ -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 == "" { diff --git a/services/statistics-service/internal/storage/mysql/query.go b/services/statistics-service/internal/storage/mysql/query.go new file mode 100644 index 00000000..552e94fd --- /dev/null +++ b/services/statistics-service/internal/storage/mysql/query.go @@ -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) +} diff --git a/services/statistics-service/internal/storage/mysql/repository.go b/services/statistics-service/internal/storage/mysql/repository.go index 270facb3..eb5bc2c9 100644 --- a/services/statistics-service/internal/storage/mysql/repository.go +++ b/services/statistics-service/internal/storage/mysql/repository.go @@ -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