diff --git a/.gitignore b/.gitignore index 65402c2e..c0661130 100644 --- a/.gitignore +++ b/.gitignore @@ -31,3 +31,4 @@ redis-data/ .tmp/ /tmp/ +.claude/settings.local.json diff --git a/docs/flutter对接/App埋点Flutter对接.md b/docs/flutter对接/App埋点Flutter对接.md new file mode 100644 index 00000000..7784983d --- /dev/null +++ b/docs/flutter对接/App埋点Flutter对接.md @@ -0,0 +1,108 @@ +# App 埋点 Flutter 对接 + +## 地址 + +```http +POST /api/v1/app/events +Content-Type: application/json +``` + +可登录也可未登录调用。已登录时建议带： + +```http +Authorization: Bearer +``` + +未登录时必须传 `device_id` 或 Header： + +```http +X-Device-ID: +``` + +## 参数 + +Body： + +| 字段 | 类型 | 必填 | 说明 | +| --- | --- | --- | --- | +| `events` | array | 是 | 埋点列表，单次最多 50 条。 | + +`events[]`： + +| 字段 | 类型 | 必填 | 说明 | +| --- | --- | --- | --- | +| `event_id` | string | 是 | 客户端生成的唯一事件 ID。同一事件重试必须复用同一个值。 | +| `event_name` | string | 是 | 事件名，例如 `banner_view`、`banner_click`、`page_open`。 | +| `event_type` | string | 否 | 事件类型，例如 `view`、`click`、`submit`。 | +| `screen` | string | 否 | 页面或模块，例如 `home`、`room`、`me`。 | +| `target_type` | string | 否 | 目标类型，例如 `banner`、`button`、`room`。 | +| `target_id` | string | 否 | 目标 ID，例如 banner ID、房间 ID。 | +| `device_id` | string | 未登录必填 | 设备 ID。也可以放到 `X-Device-ID`。 | +| `session_id` | string | 否 | 客户端本地会话 ID。登录态下服务端优先使用 token 里的 session。 | +| `platform` | string | 否 | `android` 或 `ios`。不传时读取 `X-App-Platform` / `X-Platform`。 | +| `app_version` | string | 否 | App 版本。不传时读取 `X-App-Version` / `X-Client-Version`。 | +| `language` | string | 否 | 语言。不传时读取 `X-App-Language` / `X-Language` / `Accept-Language`。 | +| `timezone` | string | 否 | 客户端时区，只做属性记录，不参与业务切日。 | +| `duration_ms` | int64 | 否 | 客户端记录的耗时毫秒。 | +| `success` | bool | 否 | 动作是否成功。 | +| `error_code` | string | 否 | 失败时的客户端错误码。 | +| `properties` | object | 否 | 扩展属性，最大 4KB。 | +| `occurred_at_ms` | int64 | 否 | 事件发生时间，UTC epoch milliseconds。不传时使用服务端接收时间。 | + +服务端会自己识别 `app_code`、`user_id`、`country_id`、`region_id`，客户端不要传这些字段。 + +示例： + +```json +{ + "events": [ + { + "event_id": "evt_20260703_100001_banner_17_view", + "event_name": "banner_view", + "event_type": "view", + "screen": "home", + "target_type": "banner", + "target_id": "17", + "device_id": "device_xxx", + "platform": "android", + "app_version": "1.2.3", + "language": "en-US", + "timezone": "Asia/Shanghai", + "success": true, + "properties": { + "position": "top" + }, + "occurred_at_ms": 1778000005000 + } + ] +} +``` + +## 返回值 + +```json +{ + "code": "OK", + "message": "ok", + "request_id": "req_xxx", + "data": { + "accepted": true, + "received": 1, + "stored": 1, + "duplicated": 0, + "server_time_ms": 1778000005000 + } +} +``` + +| 字段 | 说明 | +| --- | --- | +| `data.accepted` | 是否接受本次上报。 | +| `data.received` | 本次收到的事件数。 | +| `data.stored` | 本次新写入的事件数。 | +| `data.duplicated` | 已存在的事件数。通常是重试导致的重复 `event_id`。 | +| `data.server_time_ms` | 服务端时间，UTC epoch milliseconds。 | + +## 相关 IM + +无。 diff --git a/server/admin/cmd/server/main.go b/server/admin/cmd/server/main.go index f411884d..37e02897 100644 --- a/server/admin/cmd/server/main.go +++ b/server/admin/cmd/server/main.go @@ -184,6 +184,11 @@ func main() { fatalRuntime("connect_money_region_sources_failed", err) } defer closeMoneyRegionSources(context.Background()) + financeBillSources, closeFinanceBillSources, err := connectFinanceBillSources(context.Background(), cfg.FinanceBillSources, moneyRegionSources) + if err != nil { + fatalRuntime("connect_finance_bill_sources_failed", err) + } + defer closeFinanceBillSources(context.Background()) dashboardExternalSources, closeDashboardExternalSources, err := connectDashboardExternalSources(context.Background(), cfg.DashboardExternalSources, dashboardRegionResolvers(moneyRegionSources)...) if err != nil { fatalRuntime("connect_dashboard_external_sources_failed", err) @@ -327,7 +332,7 @@ func main() { LevelConfig: levelconfigmodule.New(activityclient.NewGRPC(activityConn), walletclient.NewGRPC(walletConn), auditHandler), LuckyGift: luckygiftmodule.New(activityclient.NewGRPC(activityConn), cfg.ActivityService.RequestTimeout, auditHandler), Menu: menumodule.New(store, auditHandler), - Payment: paymentmodule.New(walletclient.NewGRPC(walletConn), userDB, walletDB, store, auditHandler, paymentmodule.WithMoneyRegionSources(moneyRegionSources...)), + Payment: paymentmodule.New(walletclient.NewGRPC(walletConn), userDB, walletDB, store, auditHandler, paymentmodule.WithMoneyRegionSources(moneyRegionSources...), paymentmodule.WithRechargeBillSources(financeBillSources...)), PrettyID: prettyidmodule.New(userclient.NewGRPC(userConn), auditHandler), RBAC: rbacmodule.New(store, auditHandler), RedPacket: redpacketmodule.New(walletclient.NewGRPC(walletConn), auditHandler), @@ -432,6 +437,44 @@ func connectMoneyRegionSources(ctx context.Context, configs []config.MoneyRegion return sources, cleanup, nil } +// connectFinanceBillSources 连接 legacy App 的账单 Mongo；启动阶段 ping 失败直接退出，避免财务明细静默漏掉 Yumi 账单。 +func connectFinanceBillSources(ctx context.Context, configs []config.FinanceBillSourceConfig, regionSources []paymentmodule.MoneyRegionSource) ([]paymentmodule.RechargeBillSource, func(context.Context), error) { + sources := []paymentmodule.RechargeBillSource{} + clients := []*mongo.Client{} + cleanup := func(closeCtx context.Context) { + for _, client := range clients { + if client != nil { + _ = client.Disconnect(closeCtx) + } + } + } + for _, sourceConfig := range configs { + if !sourceConfig.Enabled { + continue + } + connectCtx, cancel := context.WithTimeout(ctx, sourceConfig.RequestTimeout) + client, err := mongo.Connect(options.Client().ApplyURI(sourceConfig.MongoURI).SetServerSelectionTimeout(sourceConfig.RequestTimeout)) + if err != nil { + cancel() + cleanup(ctx) + return nil, nil, err + } + if err := client.Ping(connectCtx, readpref.Primary()); err != nil { + cancel() + _ = client.Disconnect(ctx) + cleanup(ctx) + return nil, nil, err + } + cancel() + clients = append(clients, client) + source := paymentmodule.NewMongoRechargeBillSource(client.Database(sourceConfig.MongoDatabase), sourceConfig, regionSources...) + if source != nil { + sources = append(sources, source) + } + } + return sources, cleanup, nil +} + func dashboardRegionResolvers(sources []paymentmodule.MoneyRegionSource) []dashboardmodule.ExternalDashboardRegionResolver { resolvers := []dashboardmodule.ExternalDashboardRegionResolver{} for _, source := range sources { diff --git a/server/admin/configs/config.tencent.example.yaml b/server/admin/configs/config.tencent.example.yaml index b09e9dbc..e8532bc2 100644 --- a/server/admin/configs/config.tencent.example.yaml +++ b/server/admin/configs/config.tencent.example.yaml @@ -15,11 +15,11 @@ robot_profile_source: mysql_dsn: "likei_reader:REPLACE_ME@tcp(10.2.21.3:3306)/likei?parseTime=true&charset=utf8mb4&loc=UTC" request_timeout: "5s" money_region_sources: - - enabled: false + - enabled: true app_code: "yumi" app_name: "Yumi" sys_origin: "LIKEI" - mongo_uri: "mongodb://REPLACE_ME" + mongo_uri: "mongodb://mongouser:REPLACE_ME@10.2.21.9:27017/?authSource=admin" mongo_database: "tarab_all" mongo_collection: "sys_region_config" request_timeout: "5s" @@ -31,6 +31,16 @@ money_region_sources: mongo_database: "tarab_all" mongo_collection: "sys_region_config" request_timeout: "5s" +# Yumi 的充值账单在利雅得内网 likei Mongo（与 hyapp-server 同 VPC）；启用后 APP 充值详情直接读该库。 +finance_bill_sources: + - enabled: true + app_code: "yumi" + app_name: "Yumi" + sys_origin: "LIKEI" + mongo_uri: "mongodb://mongouser:REPLACE_ME@10.2.21.9:27017/?authSource=admin" + mongo_database: "tarab_all" + mongo_collection: "in_app_purchase_details" + request_timeout: "5s" mysql_auto_migrate: false migrations: enabled: true diff --git a/server/admin/configs/config.yaml b/server/admin/configs/config.yaml index 1af0d037..2f85044b 100644 --- a/server/admin/configs/config.yaml +++ b/server/admin/configs/config.yaml @@ -31,6 +31,16 @@ money_region_sources: mongo_database: "tarab_all" mongo_collection: "sys_region_config" request_timeout: "5s" +# legacy App 的财务充值账单外部源：命中 app_code 后，APP 充值详情直接读 likei Mongo 的内购明细。 +finance_bill_sources: + - enabled: false + app_code: "yumi" + app_name: "Yumi" + sys_origin: "LIKEI" + mongo_uri: "" + mongo_database: "tarab_all" + mongo_collection: "in_app_purchase_details" + request_timeout: "5s" mysql_auto_migrate: true migrations: enabled: true diff --git a/server/admin/internal/config/config.go b/server/admin/internal/config/config.go index 98230180..57a46c5a 100644 --- a/server/admin/internal/config/config.go +++ b/server/admin/internal/config/config.go @@ -23,6 +23,7 @@ type Config struct { WalletMySQLDSN string `yaml:"wallet_mysql_dsn"` RobotProfileSource RobotProfileSourceConfig `yaml:"robot_profile_source"` MoneyRegionSources []MoneyRegionSourceConfig `yaml:"money_region_sources"` + FinanceBillSources []FinanceBillSourceConfig `yaml:"finance_bill_sources"` MySQLAutoMigrate bool `yaml:"mysql_auto_migrate"` Migrations MigrationConfig `yaml:"migrations"` JWTSecret string `yaml:"jwt_secret"` @@ -132,6 +133,19 @@ type MoneyRegionSourceConfig struct { RequestTimeout time.Duration `yaml:"request_timeout"` } +// FinanceBillSourceConfig 描述 legacy App（Yumi/Aslan 等 likei 平台）的充值账单外部 Mongo 源； +// 财务系统 APP 充值详情按 app_code 命中后直接读 in_app_purchase_details，不经过 wallet-service。 +type FinanceBillSourceConfig struct { + Enabled bool `yaml:"enabled"` + AppCode string `yaml:"app_code"` + AppName string `yaml:"app_name"` + SysOrigin string `yaml:"sys_origin"` + MongoURI string `yaml:"mongo_uri"` + MongoDatabase string `yaml:"mongo_database"` + MongoCollection string `yaml:"mongo_collection"` + RequestTimeout time.Duration `yaml:"request_timeout"` +} + type TencentCOSConfig struct { Enabled bool `yaml:"enabled"` SecretID string `yaml:"secret-id"` @@ -200,6 +214,17 @@ func Default() Config { RequestTimeout: 5 * time.Second, }, }, + FinanceBillSources: []FinanceBillSourceConfig{ + { + Enabled: false, + AppCode: "yumi", + AppName: "Yumi", + SysOrigin: "LIKEI", + MongoDatabase: "tarab_all", + MongoCollection: "in_app_purchase_details", + RequestTimeout: 5 * time.Second, + }, + }, MySQLAutoMigrate: true, Migrations: MigrationConfig{ Enabled: true, @@ -476,6 +501,27 @@ func (cfg *Config) Normalize() { } } cfg.applyMoneyRegionSourceEnvOverrides() + for index := range cfg.FinanceBillSources { + source := &cfg.FinanceBillSources[index] + // legacy 账单源只服务财务充值明细；配置层统一大小写和默认集合，handler 只按 app_code 命中。 + source.AppCode = strings.ToLower(strings.TrimSpace(source.AppCode)) + source.AppName = strings.TrimSpace(source.AppName) + source.SysOrigin = strings.ToUpper(strings.TrimSpace(source.SysOrigin)) + source.MongoURI = strings.TrimSpace(source.MongoURI) + source.MongoDatabase = strings.Trim(strings.TrimSpace(source.MongoDatabase), "/") + if source.MongoDatabase == "" { + // 线上 likei Mongo（10.2.21.9）的账单和区域目录都在 tarab_all 库，与 money_region_sources 保持一致。 + source.MongoDatabase = "tarab_all" + } + source.MongoCollection = strings.TrimSpace(source.MongoCollection) + if source.MongoCollection == "" { + source.MongoCollection = "in_app_purchase_details" + } + if source.RequestTimeout <= 0 { + source.RequestTimeout = 5 * time.Second + } + } + cfg.applyFinanceBillSourceEnvOverrides() cfg.TencentCOS.SecretID = strings.TrimSpace(cfg.TencentCOS.SecretID) cfg.TencentCOS.SecretKey = strings.TrimSpace(cfg.TencentCOS.SecretKey) cfg.TencentCOS.BucketName = strings.TrimSpace(cfg.TencentCOS.BucketName) @@ -671,6 +717,30 @@ func (cfg Config) Validate() error { return fmt.Errorf("money_region_sources[%d].request_timeout must be greater than 0", index) } } + for index, source := range cfg.FinanceBillSources { + if !source.Enabled { + continue + } + // legacy 账单源启用但配置不全时必须启动失败，避免财务充值明细静默漏掉 Yumi 账单。 + if strings.TrimSpace(source.AppCode) == "" { + return fmt.Errorf("finance_bill_sources[%d].app_code is required when enabled", index) + } + if strings.TrimSpace(source.SysOrigin) == "" { + return fmt.Errorf("finance_bill_sources[%d].sys_origin is required when enabled", index) + } + if strings.TrimSpace(source.MongoURI) == "" { + return fmt.Errorf("finance_bill_sources[%d].mongo_uri is required when enabled", index) + } + if strings.TrimSpace(source.MongoDatabase) == "" { + return fmt.Errorf("finance_bill_sources[%d].mongo_database is required when enabled", index) + } + if strings.TrimSpace(source.MongoCollection) == "" { + return fmt.Errorf("finance_bill_sources[%d].mongo_collection is required when enabled", index) + } + if source.RequestTimeout <= 0 { + return fmt.Errorf("finance_bill_sources[%d].request_timeout must be greater than 0", index) + } + } 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") @@ -766,6 +836,29 @@ func (cfg *Config) applyMoneyRegionSourceEnvOverrides() { } } +func (cfg *Config) applyFinanceBillSourceEnvOverrides() { + for index := range cfg.FinanceBillSources { + source := &cfg.FinanceBillSources[index] + appKey := envAppKey(source.AppCode) + if appKey == "" { + continue + } + if value := strings.TrimSpace(firstEnv( + "HYAPP_ADMIN_"+appKey+"_FINANCE_BILL_MONGO_URI", + "HYAPP_ADMIN_"+appKey+"_MONGO_URI", + )); value != "" { + // legacy 账单源与区域源/大屏共享同一个 likei Mongo 集群；真实密码只由运行环境注入。 + source.MongoURI = value + } + if value := strings.TrimSpace(firstEnv( + "HYAPP_ADMIN_"+appKey+"_FINANCE_BILL_MONGO_DATABASE", + "HYAPP_ADMIN_"+appKey+"_MONGO_DATABASE", + )); value != "" { + source.MongoDatabase = strings.Trim(strings.TrimSpace(value), "/") + } + } +} + func envAppKey(appCode string) string { appCode = strings.ToUpper(strings.TrimSpace(appCode)) if appCode == "" { diff --git a/server/admin/internal/modules/payment/handler.go b/server/admin/internal/modules/payment/handler.go index 4ba2f5c5..9829ebbb 100644 --- a/server/admin/internal/modules/payment/handler.go +++ b/server/admin/internal/modules/payment/handler.go @@ -32,6 +32,7 @@ type Handler struct { audit shared.OperationLogger exchangeRates exchangeRateClient moneyRegionSources []MoneyRegionSource + billSources map[string]RechargeBillSource } type HandlerOption func(*Handler) @@ -46,6 +47,21 @@ func WithMoneyRegionSources(sources ...MoneyRegionSource) HandlerOption { } } +// WithRechargeBillSources 注册 legacy App（Yumi 等）的外部账单源；命中 app_code 的充值明细和汇总直接读外部源。 +func WithRechargeBillSources(sources ...RechargeBillSource) HandlerOption { + return func(h *Handler) { + for _, source := range sources { + if source == nil { + continue + } + if h.billSources == nil { + h.billSources = map[string]RechargeBillSource{} + } + h.billSources[appctx.Normalize(source.AppCode())] = source + } + } +} + // New 组装支付后台 handler；wallet 负责账单和商品事实，userDB 只补后台列表需要的用户展示资料。 func New(wallet walletclient.Client, userDB *sql.DB, walletDB *sql.DB, store *repository.Store, audit shared.OperationLogger, options ...HandlerOption) *Handler { handler := &Handler{wallet: wallet, userDB: userDB, walletDB: walletDB, store: store, audit: audit, exchangeRates: newDefaultExchangeRateClient()} @@ -61,6 +77,10 @@ func New(wallet walletclient.Client, userDB *sql.DB, walletDB *sql.DB, store *re func (h *Handler) ListRechargeBills(c *gin.Context) { options := shared.ListOptions(c) appCode := appctx.FromContext(c.Request.Context()) + if source, ok := h.billSources[appCode]; ok { + h.listLegacyRechargeBills(c, source) + return + } userFilter := shared.UserIdentityFilterFromQuery(c, "") userID, userMatched, ok := h.resolveRechargeBillUserFilter(c, appCode, userFilter, "user_id 参数不正确", "user_id", "userId") if !ok { diff --git a/server/admin/internal/modules/payment/legacy_bill_source.go b/server/admin/internal/modules/payment/legacy_bill_source.go new file mode 100644 index 00000000..50440744 --- /dev/null +++ b/server/admin/internal/modules/payment/legacy_bill_source.go @@ -0,0 +1,401 @@ +package payment + +import ( + "context" + "fmt" + "math" + "strconv" + "strings" + "time" + + "hyapp-admin-server/internal/appctx" + "hyapp-admin-server/internal/config" + + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" +) + +// legacyRechargeBillQuery 是 legacy 账单源支持的筛选子集；用户/币商筛选属于 hyapp 语义，legacy 源不适用。 +type legacyRechargeBillQuery struct { + Keyword string + RegionID int64 + StartAtMS int64 + EndAtMS int64 + Page int + PageSize int +} + +// RechargeBillSource 是 legacy App（Yumi 等 likei 平台）充值账单的外部只读来源； +// 命中 app_code 后，财务充值明细和汇总直接走该来源，不再请求 wallet-service。 +type RechargeBillSource interface { + AppCode() string + AppName() string + ListRechargeBills(ctx context.Context, query legacyRechargeBillQuery) ([]rechargeBillDTO, int64, error) + SummarizeRechargeBills(ctx context.Context, query legacyRechargeBillQuery) (rechargeBillSummaryDTO, error) +} + +// legacyRegionCountryResolver 复用财务范围的区域→国家映射（MongoMoneyRegionSource 已实现）。 +type legacyRegionCountryResolver interface { + CountryCodesForRegion(ctx context.Context, appCode string, regionID int64) ([]string, bool, error) +} + +// MongoRechargeBillSource 读取 likei Mongo 的 in_app_purchase_details 集合作为充值账单事实。 +type MongoRechargeBillSource struct { + collection *mongo.Collection + config config.FinanceBillSourceConfig + regionResolvers []legacyRegionCountryResolver +} + +// NewMongoRechargeBillSource 创建 legacy 账单源；regionSources 用于把后台区域 ID 解析成国家码过滤条件。 +func NewMongoRechargeBillSource(database *mongo.Database, sourceConfig config.FinanceBillSourceConfig, regionSources ...MoneyRegionSource) *MongoRechargeBillSource { + if database == nil || !sourceConfig.Enabled { + return nil + } + collectionName := strings.TrimSpace(sourceConfig.MongoCollection) + if collectionName == "" { + collectionName = "in_app_purchase_details" + } + sourceConfig.AppCode = appctx.Normalize(sourceConfig.AppCode) + sourceConfig.AppName = strings.TrimSpace(sourceConfig.AppName) + sourceConfig.SysOrigin = strings.ToUpper(strings.TrimSpace(sourceConfig.SysOrigin)) + source := &MongoRechargeBillSource{collection: database.Collection(collectionName), config: sourceConfig} + for _, regionSource := range regionSources { + if resolver, ok := regionSource.(legacyRegionCountryResolver); ok && resolver != nil { + source.regionResolvers = append(source.regionResolvers, resolver) + } + } + return source +} + +func (s *MongoRechargeBillSource) AppCode() string { + return s.config.AppCode +} + +func (s *MongoRechargeBillSource) AppName() string { + if s.config.AppName != "" { + return s.config.AppName + } + return s.config.AppCode +} + +// legacyPurchaseDocument 对应 chatapp3-java 的 InAppPurchaseDetails Mongo 文档（只取展示需要的字段）。 +type legacyPurchaseDocument struct { + ID any `bson:"_id"` + OrderID string `bson:"orderId"` + AcceptUserID int64 `bson:"acceptUserId"` + CountryCode string `bson:"countryCode"` + Currency string `bson:"currency"` + Amount any `bson:"amount"` + AmountUSD any `bson:"amountUsd"` + Status string `bson:"status"` + Factory legacyPurchaseFactory `bson:"factory"` + Products []legacyPurchaseProduct `bson:"products"` + CreateTime time.Time `bson:"createTime"` + PurchaseDateMS time.Time `bson:"purchaseDateMs"` +} + +type legacyPurchaseFactory struct { + Platform string `bson:"platform"` + FactoryCode string `bson:"factoryCode"` +} + +type legacyPurchaseProduct struct { + Name string `bson:"name"` + Content string `bson:"content"` +} + +func (s *MongoRechargeBillSource) ListRechargeBills(ctx context.Context, query legacyRechargeBillQuery) ([]rechargeBillDTO, int64, error) { + if s == nil || s.collection == nil { + return nil, 0, fmt.Errorf("legacy bill source is not configured") + } + filter, matchable, err := s.billFilter(ctx, query) + if err != nil { + return nil, 0, err + } + if !matchable { + return []rechargeBillDTO{}, 0, nil + } + total, err := s.collection.CountDocuments(ctx, filter) + if err != nil { + return nil, 0, fmt.Errorf("count legacy recharge bills for %s: %w", s.config.AppCode, err) + } + page := query.Page + if page < 1 { + page = 1 + } + pageSize := query.PageSize + if pageSize < 1 { + pageSize = 10 + } + cursor, err := s.collection.Find(ctx, filter, options.Find(). + SetSort(bson.D{{Key: "createTime", Value: -1}, {Key: "_id", Value: -1}}). + SetSkip(int64(page-1)*int64(pageSize)). + SetLimit(int64(pageSize))) + if err != nil { + return nil, 0, fmt.Errorf("query legacy recharge bills for %s: %w", s.config.AppCode, err) + } + defer cursor.Close(ctx) + + items := make([]rechargeBillDTO, 0, pageSize) + for cursor.Next(ctx) { + var document legacyPurchaseDocument + if err := cursor.Decode(&document); err != nil { + return nil, 0, fmt.Errorf("decode legacy recharge bill for %s: %w", s.config.AppCode, err) + } + items = append(items, legacyRechargeBillDTO(s.config, document)) + } + if err := cursor.Err(); err != nil { + return nil, 0, fmt.Errorf("iterate legacy recharge bills for %s: %w", s.config.AppCode, err) + } + return items, total, nil +} + +func (s *MongoRechargeBillSource) SummarizeRechargeBills(ctx context.Context, query legacyRechargeBillQuery) (rechargeBillSummaryDTO, error) { + if s == nil || s.collection == nil { + return rechargeBillSummaryDTO{}, fmt.Errorf("legacy bill source is not configured") + } + filter, matchable, err := s.billFilter(ctx, query) + if err != nil { + return rechargeBillSummaryDTO{}, err + } + if !matchable { + return rechargeBillSummaryDTO{}, nil + } + // 金币数只统计 GOLD/CANDY 商品内容，口径对齐 chatapp3 的 firstGoldProduct；content 非数值时按 0 计入。 + coinExpr := bson.M{"$reduce": bson.M{ + "input": bson.M{"$filter": bson.M{ + "input": bson.M{"$ifNull": bson.A{"$products", bson.A{}}}, + "cond": bson.M{"$in": bson.A{"$$this.name", bson.A{"GOLD", "CANDY"}}}, + }}, + "initialValue": 0, + "in": bson.M{"$add": bson.A{ + "$$value", + bson.M{"$convert": bson.M{"input": "$$this.content", "to": "long", "onError": 0, "onNull": 0}}, + }}, + }} + cursor, err := s.collection.Aggregate(ctx, mongo.Pipeline{ + {{Key: "$match", Value: filter}}, + {{Key: "$group", Value: bson.M{ + "_id": bson.M{"$cond": bson.A{ + bson.M{"$eq": bson.A{"$factory.factoryCode", "GOOGLE"}}, + "google", + "third_party", + }}, + "billCount": bson.M{"$sum": 1}, + "usd": bson.M{"$sum": bson.M{"$ifNull": bson.A{"$amountUsd", 0}}}, + "coin": bson.M{"$sum": coinExpr}, + }}}, + }) + if err != nil { + return rechargeBillSummaryDTO{}, fmt.Errorf("summarize legacy recharge bills for %s: %w", s.config.AppCode, err) + } + defer cursor.Close(ctx) + + summary := rechargeBillSummaryDTO{} + for cursor.Next(ctx) { + var row struct { + ID string `bson:"_id"` + BillCount int64 `bson:"billCount"` + USD any `bson:"usd"` + Coin int64 `bson:"coin"` + } + if err := cursor.Decode(&row); err != nil { + return rechargeBillSummaryDTO{}, fmt.Errorf("decode legacy recharge summary for %s: %w", s.config.AppCode, err) + } + bucket := rechargeBillSummaryBucketDTO{ + BillCount: row.BillCount, + CoinAmount: row.Coin, + USDMinorAmount: legacyDecimalToMinor(row.USD), + } + if row.ID == "google" { + summary.GooglePlay = bucket + } else { + summary.ThirdParty.BillCount += bucket.BillCount + summary.ThirdParty.CoinAmount += bucket.CoinAmount + summary.ThirdParty.USDMinorAmount += bucket.USDMinorAmount + } + } + if err := cursor.Err(); err != nil { + return rechargeBillSummaryDTO{}, fmt.Errorf("iterate legacy recharge summary for %s: %w", s.config.AppCode, err) + } + summary.Total = rechargeBillSummaryBucketDTO{ + BillCount: summary.GooglePlay.BillCount + summary.ThirdParty.BillCount, + CoinAmount: summary.GooglePlay.CoinAmount + summary.ThirdParty.CoinAmount, + USDMinorAmount: summary.GooglePlay.USDMinorAmount + summary.ThirdParty.USDMinorAmount, + } + return summary, nil +} + +// billFilter 构造与 hyapp 账单列表同口径的 Mongo 过滤条件；返回 matchable=false 表示筛选注定无结果。 +func (s *MongoRechargeBillSource) billFilter(ctx context.Context, query legacyRechargeBillQuery) (bson.M, bool, error) { + filter := bson.M{ + "sysOrigin": s.config.SysOrigin, + "status": "SUCCESS", + // 订阅免费试用不产生真实付款，与 Aslan 大屏充值口径保持一致。 + "trialPeriod": bson.M{"$ne": true}, + } + timeRange := bson.M{} + if query.StartAtMS > 0 { + timeRange["$gte"] = time.UnixMilli(query.StartAtMS) + } + if query.EndAtMS > 0 { + timeRange["$lt"] = time.UnixMilli(query.EndAtMS) + } + if len(timeRange) > 0 { + filter["createTime"] = timeRange + } + if keyword := strings.TrimSpace(query.Keyword); keyword != "" { + // legacy 集合没有 hyapp 交易号；关键词只精确匹配内部订单 ID 和三方订单号，避免大集合正则扫描。 + filter["$or"] = bson.A{ + bson.M{"_id": keyword}, + bson.M{"orderId": keyword}, + } + } + if query.RegionID > 0 { + codes, handled, err := s.regionCountryCodes(ctx, query.RegionID) + if err != nil { + return nil, false, err + } + if !handled || len(codes) == 0 { + return nil, false, nil + } + filter["countryCode"] = bson.M{"$in": codes} + } + return filter, true, nil +} + +func (s *MongoRechargeBillSource) regionCountryCodes(ctx context.Context, regionID int64) ([]string, bool, error) { + for _, resolver := range s.regionResolvers { + codes, handled, err := resolver.CountryCodesForRegion(ctx, s.config.AppCode, regionID) + if err != nil { + return nil, false, err + } + if handled { + normalized := make([]string, 0, len(codes)) + for _, code := range codes { + if code = strings.ToUpper(strings.TrimSpace(code)); code != "" { + normalized = append(normalized, code) + } + } + return normalized, true, nil + } + } + return nil, false, nil +} + +// legacyRechargeBillDTO 把 likei 内购明细收敛成财务充值账单 DTO；金额统一按实付币种微单位、美金最小单位表达。 +func legacyRechargeBillDTO(sourceConfig config.FinanceBillSourceConfig, document legacyPurchaseDocument) rechargeBillDTO { + createdAtMS := int64(0) + if !document.CreateTime.IsZero() { + createdAtMS = document.CreateTime.UnixMilli() + } else if !document.PurchaseDateMS.IsZero() { + createdAtMS = document.PurchaseDateMS.UnixMilli() + } + factoryCode := strings.ToUpper(strings.TrimSpace(document.Factory.FactoryCode)) + dto := rechargeBillDTO{ + AppCode: sourceConfig.AppCode, + TransactionID: legacyDocumentID(document.ID), + RechargeType: legacyRechargeType(factoryCode), + Status: "succeeded", + ExternalRef: strings.TrimSpace(document.OrderID), + UserID: document.AcceptUserID, + CurrencyCode: strings.ToUpper(strings.TrimSpace(document.Currency)), + CoinAmount: legacyGoldCoinAmount(document.Products), + USDMinorAmount: legacyDecimalToMinor(document.AmountUSD), + CreatedAtMS: createdAtMS, + ProviderCode: legacyProviderCode(factoryCode), + UserPaidCurrencyCode: strings.ToUpper(strings.TrimSpace(document.Currency)), + UserPaidAmountMicro: legacyDecimalToMicro(document.Amount), + // likei 平台没有可反推的三方扣款数据；标记已同步避免前端把 legacy 谷歌账单当成待查询。 + PaidSyncedAtMS: createdAtMS, + } + return dto +} + +func legacyDocumentID(value any) string { + if value == nil { + return "" + } + if text, ok := value.(string); ok { + return text + } + if objectID, ok := value.(bson.ObjectID); ok { + return objectID.Hex() + } + return strings.TrimSpace(fmt.Sprint(value)) +} + +func legacyRechargeType(factoryCode string) string { + switch factoryCode { + case "GOOGLE": + return "google_play_recharge" + case "APPLE": + return "apple_recharge" + case "HUAWEI": + return "huawei_recharge" + case "TELEGRAM": + return "telegram_recharge" + default: + return "web_recharge" + } +} + +func legacyProviderCode(factoryCode string) string { + if factoryCode == "" { + return "legacy" + } + return strings.ToLower(factoryCode) +} + +// legacyGoldCoinAmount 只统计 GOLD/CANDY 商品内容，口径对齐 chatapp3 的 firstGoldProduct。 +func legacyGoldCoinAmount(products []legacyPurchaseProduct) int64 { + var total int64 + for _, product := range products { + if product.Name != "GOLD" && product.Name != "CANDY" { + continue + } + if amount, err := strconv.ParseInt(strings.TrimSpace(product.Content), 10, 64); err == nil { + total += amount + } + } + return total +} + +func legacyDecimalToMicro(value any) int64 { + return legacyDecimalScaled(value, 1_000_000) +} + +func legacyDecimalToMinor(value any) int64 { + return legacyDecimalScaled(value, 100) +} + +// legacyDecimalScaled 把 Mongo 里 Decimal128/数值/字符串形式的十进制金额换算成目标最小单位；非法值按 0 处理。 +func legacyDecimalScaled(value any, scale float64) int64 { + var text string + switch typed := value.(type) { + case nil: + return 0 + case bson.Decimal128: + text = typed.String() + case string: + text = typed + case int32: + return int64(typed) * int64(scale) + case int64: + return typed * int64(scale) + case int: + return int64(typed) * int64(scale) + case float64: + text = strconv.FormatFloat(typed, 'f', -1, 64) + case float32: + text = strconv.FormatFloat(float64(typed), 'f', -1, 32) + default: + text = fmt.Sprint(value) + } + parsed, err := strconv.ParseFloat(strings.TrimSpace(text), 64) + if err != nil || math.IsNaN(parsed) || math.IsInf(parsed, 0) { + return 0 + } + return int64(math.Round(parsed * scale)) +} diff --git a/server/admin/internal/modules/payment/legacy_bill_source_test.go b/server/admin/internal/modules/payment/legacy_bill_source_test.go new file mode 100644 index 00000000..8383f322 --- /dev/null +++ b/server/admin/internal/modules/payment/legacy_bill_source_test.go @@ -0,0 +1,165 @@ +package payment + +import ( + "context" + "testing" + "time" + + "hyapp-admin-server/internal/config" + + "go.mongodb.org/mongo-driver/v2/bson" +) + +func yumiBillSourceConfig() config.FinanceBillSourceConfig { + return config.FinanceBillSourceConfig{ + Enabled: true, + AppCode: "yumi", + AppName: "Yumi", + SysOrigin: "LIKEI", + MongoDatabase: "test", + MongoCollection: "in_app_purchase_details", + RequestTimeout: time.Second, + } +} + +func TestLegacyRechargeBillDTOMapsGooglePurchase(t *testing.T) { + amount, err := bson.ParseDecimal128("164.99") + if err != nil { + t.Fatalf("parse amount: %v", err) + } + amountUSD, err := bson.ParseDecimal128("2.99") + if err != nil { + t.Fatalf("parse amount usd: %v", err) + } + createTime := time.UnixMilli(1_751_000_000_000).UTC() + dto := legacyRechargeBillDTO(yumiBillSourceConfig(), legacyPurchaseDocument{ + ID: "175100000000012345", + OrderID: "GPA.1234-5678-9012-34567", + AcceptUserID: 42, + CountryCode: "TR", + Currency: "try", + Amount: amount, + AmountUSD: amountUSD, + Status: "SUCCESS", + Factory: legacyPurchaseFactory{Platform: "Android", FactoryCode: "GOOGLE"}, + Products: []legacyPurchaseProduct{ + {Name: "GOLD", Content: "560000"}, + {Name: "PROPS", Content: "AVATAR_FRAME"}, + }, + CreateTime: createTime, + }) + + if dto.AppCode != "yumi" || dto.TransactionID != "175100000000012345" { + t.Fatalf("unexpected identity: %+v", dto) + } + if dto.RechargeType != "google_play_recharge" || dto.ProviderCode != "google" { + t.Fatalf("unexpected channel mapping: %+v", dto) + } + if dto.Status != "succeeded" || dto.ExternalRef != "GPA.1234-5678-9012-34567" { + t.Fatalf("unexpected status/external ref: %+v", dto) + } + if dto.CoinAmount != 560000 { + t.Fatalf("coin amount = %d, want 560000", dto.CoinAmount) + } + if dto.USDMinorAmount != 299 { + t.Fatalf("usd minor = %d, want 299", dto.USDMinorAmount) + } + if dto.UserPaidCurrencyCode != "TRY" || dto.UserPaidAmountMicro != 164_990_000 { + t.Fatalf("unexpected user paid: %+v", dto) + } + if dto.CreatedAtMS != createTime.UnixMilli() || dto.PaidSyncedAtMS != dto.CreatedAtMS { + t.Fatalf("unexpected timestamps: %+v", dto) + } +} + +func TestLegacyRechargeTypeMapping(t *testing.T) { + cases := map[string]string{ + "GOOGLE": "google_play_recharge", + "APPLE": "apple_recharge", + "HUAWEI": "huawei_recharge", + "TELEGRAM": "telegram_recharge", + "PAYER_MAX": "web_recharge", + "": "web_recharge", + } + for factoryCode, want := range cases { + if got := legacyRechargeType(factoryCode); got != want { + t.Fatalf("legacyRechargeType(%q) = %q, want %q", factoryCode, got, want) + } + } +} + +func TestLegacyDecimalScaledHandlesTypes(t *testing.T) { + decimal, err := bson.ParseDecimal128("6.99") + if err != nil { + t.Fatalf("parse decimal: %v", err) + } + if got := legacyDecimalToMinor(decimal); got != 699 { + t.Fatalf("decimal minor = %d, want 699", got) + } + if got := legacyDecimalToMicro("54.99"); got != 54_990_000 { + t.Fatalf("string micro = %d, want 54990000", got) + } + if got := legacyDecimalToMinor(int64(3)); got != 300 { + t.Fatalf("int64 minor = %d, want 300", got) + } + if got := legacyDecimalToMinor(nil); got != 0 { + t.Fatalf("nil minor = %d, want 0", got) + } + if got := legacyDecimalToMinor("not-a-number"); got != 0 { + t.Fatalf("invalid minor = %d, want 0", got) + } +} + +type staticLegacyRegionResolver struct { + codes []string + handled bool +} + +func (r staticLegacyRegionResolver) CountryCodesForRegion(context.Context, string, int64) ([]string, bool, error) { + return r.codes, r.handled, nil +} + +func TestLegacyBillFilterBuildsQuery(t *testing.T) { + source := &MongoRechargeBillSource{ + config: yumiBillSourceConfig(), + regionResolvers: []legacyRegionCountryResolver{staticLegacyRegionResolver{codes: []string{"tr", "SA"}, handled: true}}, + } + filter, matchable, err := source.billFilter(context.Background(), legacyRechargeBillQuery{ + Keyword: "GPA.1", + RegionID: 9_000_000_000_000_000_001, + StartAtMS: 1_000, + EndAtMS: 2_000, + }) + if err != nil || !matchable { + t.Fatalf("billFilter err=%v matchable=%v", err, matchable) + } + if filter["sysOrigin"] != "LIKEI" || filter["status"] != "SUCCESS" { + t.Fatalf("unexpected base filter: %v", filter) + } + timeRange, ok := filter["createTime"].(bson.M) + if !ok || !timeRange["$gte"].(time.Time).Equal(time.UnixMilli(1_000)) || !timeRange["$lt"].(time.Time).Equal(time.UnixMilli(2_000)) { + t.Fatalf("unexpected time range: %v", filter["createTime"]) + } + countries, ok := filter["countryCode"].(bson.M) + if !ok { + t.Fatalf("country filter missing: %v", filter) + } + codes, ok := countries["$in"].([]string) + if !ok || len(codes) != 2 || codes[0] != "TR" || codes[1] != "SA" { + t.Fatalf("unexpected country codes: %v", countries["$in"]) + } + if _, ok := filter["$or"]; !ok { + t.Fatalf("keyword filter missing: %v", filter) + } +} + +func TestLegacyBillFilterUnresolvedRegionMatchesNothing(t *testing.T) { + source := &MongoRechargeBillSource{config: yumiBillSourceConfig()} + _, matchable, err := source.billFilter(context.Background(), legacyRechargeBillQuery{RegionID: 123}) + if err != nil { + t.Fatalf("billFilter err=%v", err) + } + if matchable { + t.Fatalf("unresolved region should not be matchable") + } +} diff --git a/server/admin/internal/modules/payment/recharge_bill_stats.go b/server/admin/internal/modules/payment/recharge_bill_stats.go index 1b25d058..2d2b4141 100644 --- a/server/admin/internal/modules/payment/recharge_bill_stats.go +++ b/server/admin/internal/modules/payment/recharge_bill_stats.go @@ -2,6 +2,7 @@ package payment import ( "context" + "sort" "strings" "time" @@ -47,10 +48,42 @@ type googleRechargePaidDTO struct { Error string `json:"error,omitempty"` } +// listLegacyRechargeBills 处理 legacy App 的充值明细；分页与展示口径对齐 wallet-service 账单列表。 +func (h *Handler) listLegacyRechargeBills(c *gin.Context, source RechargeBillSource) { + options := shared.ListOptions(c) + items, total, err := source.ListRechargeBills(c.Request.Context(), legacyRechargeBillQuery{ + Keyword: options.Keyword, + RegionID: queryInt64(c, "region_id", "regionId"), + StartAtMS: queryInt64(c, "start_at_ms", "startAtMs"), + EndAtMS: queryInt64(c, "end_at_ms", "endAtMs"), + Page: options.Page, + PageSize: options.PageSize, + }) + if err != nil { + response.ServerError(c, "获取账单列表失败") + return + } + response.OK(c, response.Page{Items: items, Page: options.Page, PageSize: options.PageSize, Total: total}) +} + // GetRechargeBillSummary 返回与账单列表同一筛选口径的充值总和、Google 充值与三方充值聚合。 func (h *Handler) GetRechargeBillSummary(c *gin.Context) { options := shared.ListOptions(c) appCode := appctx.FromContext(c.Request.Context()) + if source, ok := h.billSources[appCode]; ok { + summary, err := source.SummarizeRechargeBills(c.Request.Context(), legacyRechargeBillQuery{ + Keyword: options.Keyword, + RegionID: queryInt64(c, "region_id", "regionId"), + StartAtMS: queryInt64(c, "start_at_ms", "startAtMs"), + EndAtMS: queryInt64(c, "end_at_ms", "endAtMs"), + }) + if err != nil { + response.ServerError(c, "获取充值汇总失败") + return + } + response.OK(c, summary) + return + } userFilter := shared.UserIdentityFilterFromQuery(c, "") userID, userMatched, ok := h.resolveRechargeBillUserFilter(c, appCode, userFilter, "user_id 参数不正确", "user_id", "userId") if !ok { @@ -111,6 +144,10 @@ func (h *Handler) RefreshGoogleRechargePaidDetails(c *gin.Context) { return } appCode := appctx.FromContext(c.Request.Context()) + if _, ok := h.billSources[appCode]; ok { + response.BadRequest(c, "该 App 的账单来自外部系统，不支持谷歌实付同步") + return + } resp, err := h.wallet.RefreshGooglePaymentPrices(c.Request.Context(), &walletv1.RefreshGooglePaymentPricesRequest{ RequestId: middleware.CurrentRequestID(c), AppCode: appCode, @@ -136,6 +173,54 @@ func (h *Handler) RefreshGoogleRechargePaidDetails(c *gin.Context) { response.OK(c, gin.H{"items": items}) } +type rechargeBillAppDTO struct { + AppCode string `json:"appCode"` + AppName string `json:"appName"` +} + +// ListRechargeBillApps 返回充值详情可选的 App 目录：hyapp 在册应用 + 配置的 legacy 账单源（Yumi 等）。 +func (h *Handler) ListRechargeBillApps(c *gin.Context) { + items := []rechargeBillAppDTO{} + seen := map[string]struct{}{} + if h.userDB != nil { + rows, err := h.userDB.QueryContext(c.Request.Context(), ` + SELECT app_code, app_name + FROM apps + WHERE status = 'active' + ORDER BY app_name ASC, app_code ASC`) + if err != nil { + response.ServerError(c, "获取 App 列表失败") + return + } + defer rows.Close() + for rows.Next() { + var item rechargeBillAppDTO + if err := rows.Scan(&item.AppCode, &item.AppName); err != nil { + response.ServerError(c, "获取 App 列表失败") + return + } + item.AppCode = appctx.Normalize(item.AppCode) + seen[item.AppCode] = struct{}{} + items = append(items, item) + } + if err := rows.Err(); err != nil { + response.ServerError(c, "获取 App 列表失败") + return + } + } + legacyApps := make([]rechargeBillAppDTO, 0, len(h.billSources)) + for _, source := range h.billSources { + appCode := appctx.Normalize(source.AppCode()) + if _, ok := seen[appCode]; ok { + continue + } + legacyApps = append(legacyApps, rechargeBillAppDTO{AppCode: appCode, AppName: source.AppName()}) + } + sort.Slice(legacyApps, func(i, j int) bool { return legacyApps[i].AppName < legacyApps[j].AppName }) + items = append(items, legacyApps...) + response.OK(c, gin.H{"items": items, "total": len(items)}) +} + // ListRechargeBillRegions 返回当前 App 的区域目录，用于充值详情按区域筛选；不受财务范围授权限制。 func (h *Handler) ListRechargeBillRegions(c *gin.Context) { appCode := appctx.FromContext(c.Request.Context()) diff --git a/server/admin/internal/modules/payment/routes.go b/server/admin/internal/modules/payment/routes.go index 2f7a4f66..223a9ecc 100644 --- a/server/admin/internal/modules/payment/routes.go +++ b/server/admin/internal/modules/payment/routes.go @@ -15,6 +15,7 @@ func RegisterRoutes(protected *gin.RouterGroup, h *Handler) { protected.GET("/admin/payment/recharge-bills/summary", middleware.RequirePermission("payment-bill:view"), h.GetRechargeBillSummary) protected.POST("/admin/payment/recharge-bills/google-paid/refresh", middleware.RequirePermission("payment-bill:view"), h.RefreshGoogleRechargePaidDetails) protected.GET("/admin/payment/recharge-regions", middleware.RequirePermission("payment-bill:view"), h.ListRechargeBillRegions) + protected.GET("/admin/payment/recharge-apps", middleware.RequirePermission("payment-bill:view"), h.ListRechargeBillApps) protected.GET("/admin/payment/third-party-channels", middleware.RequirePermission("payment-third-party:view"), h.ListThirdPartyPaymentChannels) protected.GET("/admin/finance/scope", middleware.RequirePermission(financeViewPermission), h.GetMoneyScope) protected.GET("/admin/money/scope", middleware.RequireAnyPermission(financeViewPermission, legacyMoneyViewPermission), h.GetMoneyScope) diff --git a/services/gateway-service/internal/client/statistics_client.go b/services/gateway-service/internal/client/statistics_client.go index 0b3c3d6f..ae037763 100644 --- a/services/gateway-service/internal/client/statistics_client.go +++ b/services/gateway-service/internal/client/statistics_client.go @@ -11,6 +11,38 @@ import ( "time" ) +type AppTrackingEvent struct { + AppCode string `json:"app_code"` + EventID string `json:"event_id"` + EventName string `json:"event_name"` + EventType string `json:"event_type"` + Screen string `json:"screen"` + TargetType string `json:"target_type"` + TargetID string `json:"target_id"` + UserID int64 `json:"user_id"` + DeviceID string `json:"device_id"` + SessionID string `json:"session_id"` + Platform string `json:"platform"` + AppVersion string `json:"app_version"` + Language string `json:"language"` + Timezone string `json:"timezone"` + CountryID int64 `json:"country_id"` + RegionID int64 `json:"region_id"` + DurationMS int64 `json:"duration_ms"` + Success bool `json:"success"` + ErrorCode string `json:"error_code"` + Properties json.RawMessage `json:"properties,omitempty"` + OccurredAtMS int64 `json:"occurred_at_ms"` +} + +type AppTrackingReportResult struct { + Accepted bool `json:"accepted"` + Received int `json:"received"` + Stored int `json:"stored"` + Duplicated int `json:"duplicated"` + ServerTimeMS int64 `json:"server_time_ms"` +} + type SelfGameH5Event struct { AppCode string `json:"app_code"` EventID string `json:"event_id"` @@ -34,6 +66,7 @@ type SelfGameH5Event struct { type StatisticsClient interface { ReportSelfGameH5Event(ctx context.Context, event SelfGameH5Event) error + ReportAppTrackingEvents(ctx context.Context, events []AppTrackingEvent) (AppTrackingReportResult, error) } type HTTPStatisticsClient struct { @@ -79,3 +112,32 @@ func (c *HTTPStatisticsClient) ReportSelfGameH5Event(ctx context.Context, event payload, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) return fmt.Errorf("statistics report failed: status=%d body=%s", resp.StatusCode, strings.TrimSpace(string(payload))) } + +func (c *HTTPStatisticsClient) ReportAppTrackingEvents(ctx context.Context, events []AppTrackingEvent) (AppTrackingReportResult, error) { + if c == nil || c.httpClient == nil || c.baseURL == "" { + return AppTrackingReportResult{}, fmt.Errorf("statistics client is not configured") + } + body, err := json.Marshal(map[string]any{"events": events}) + if err != nil { + return AppTrackingReportResult{}, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/internal/v1/statistics/app/events", bytes.NewReader(body)) + if err != nil { + return AppTrackingReportResult{}, err + } + req.Header.Set("Content-Type", "application/json") + resp, err := c.httpClient.Do(req) + if err != nil { + return AppTrackingReportResult{}, err + } + defer resp.Body.Close() + payload, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return AppTrackingReportResult{}, fmt.Errorf("statistics app event report failed: status=%d body=%s", resp.StatusCode, strings.TrimSpace(string(payload))) + } + var result AppTrackingReportResult + if err := json.Unmarshal(payload, &result); err != nil { + return AppTrackingReportResult{}, err + } + return result, nil +} diff --git a/services/gateway-service/internal/transport/http/appapi/app_event_handler.go b/services/gateway-service/internal/transport/http/appapi/app_event_handler.go new file mode 100644 index 00000000..3c7ba23b --- /dev/null +++ b/services/gateway-service/internal/transport/http/appapi/app_event_handler.go @@ -0,0 +1,215 @@ +package appapi + +import ( + "bytes" + "encoding/json" + "net/http" + "strings" + "time" + + userv1 "hyapp.local/api/proto/user/v1" + "hyapp/pkg/appcode" + "hyapp/services/gateway-service/internal/auth" + "hyapp/services/gateway-service/internal/client" + "hyapp/services/gateway-service/internal/transport/http/httpkit" +) + +const ( + appEventMaxBatchSize = 50 + appEventMaxPropertiesBytes = 4096 +) + +type appEventsRequest struct { + Events []appEventRequest `json:"events"` +} + +type appEventRequest struct { + EventID string `json:"event_id"` + EventName string `json:"event_name"` + EventType string `json:"event_type"` + Screen string `json:"screen"` + TargetType string `json:"target_type"` + TargetID string `json:"target_id"` + DeviceID string `json:"device_id"` + SessionID string `json:"session_id"` + Platform string `json:"platform"` + AppVersion string `json:"app_version"` + Language string `json:"language"` + Timezone string `json:"timezone"` + DurationMS int64 `json:"duration_ms"` + Success bool `json:"success"` + ErrorCode string `json:"error_code"` + Properties json.RawMessage `json:"properties"` + OccurredAtMS int64 `json:"occurred_at_ms"` +} + +func (h *Handler) reportAppEvents(writer http.ResponseWriter, request *http.Request) { + if h.statisticsClient == nil { + // App 埋点事实由 statistics-service 持久化；gateway 未注入写入边界时不能假成功，否则客户端会丢掉重试机会。 + httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error") + return + } + var body appEventsRequest + if !httpkit.Decode(writer, request, &body) { + return + } + if len(body.Events) == 0 || len(body.Events) > appEventMaxBatchSize { + httpkit.WriteError(writer, request, http.StatusBadRequest, httpkit.CodeInvalidArgument, "invalid argument") + return + } + + nowMS := time.Now().UTC().UnixMilli() + userID := auth.UserIDFromContext(request.Context()) + countryID, regionID, ok := h.appEventUserDimensions(writer, request, userID) + if !ok { + return + } + headerDeviceID := httpkit.FirstHeader(request, "X-Device-ID", "X-HY-Device-ID") + headerPlatform := httpkit.FirstHeader(request, "X-App-Platform", "X-Platform") + headerAppVersion := httpkit.FirstHeader(request, "X-App-Version", "X-Client-Version") + headerLanguage := appEventRequestLanguage(request) + headerTimezone := httpkit.FirstHeader(request, "X-Timezone", "X-App-Timezone") + + events := make([]client.AppTrackingEvent, 0, len(body.Events)) + for _, item := range body.Events { + event, valid := appTrackingEventFromRequest(item, appTrackingEventDefaults{ + appCode: appcode.FromContext(request.Context()), + userID: userID, + countryID: countryID, + regionID: regionID, + deviceID: headerDeviceID, + sessionID: auth.SessionIDFromContext(request.Context()), + platform: headerPlatform, + appVersion: headerAppVersion, + language: headerLanguage, + timezone: headerTimezone, + receivedAtMS: nowMS, + }) + if !valid { + httpkit.WriteError(writer, request, http.StatusBadRequest, httpkit.CodeInvalidArgument, "invalid argument") + return + } + events = append(events, event) + } + + result, err := h.statisticsClient.ReportAppTrackingEvents(request.Context(), events) + if err != nil { + httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error") + return + } + httpkit.WriteOK(writer, request, result) +} + +func (h *Handler) appEventUserDimensions(writer http.ResponseWriter, request *http.Request, userID int64) (int64, int64, bool) { + if userID <= 0 { + return 0, 0, true + } + if h.userProfileClient == nil { + httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error") + return 0, 0, false + } + resp, err := h.userProfileClient.GetUser(request.Context(), &userv1.GetUserRequest{ + Meta: httpkit.UserMeta(request, httpkit.FirstHeader(request, "X-Device-ID", "X-HY-Device-ID")), + UserId: userID, + }) + if err != nil { + httpkit.WriteRPCError(writer, request, err) + return 0, 0, false + } + user := resp.GetUser() + if user == nil { + // 登录态缺用户主数据属于服务端依赖异常；不能退化成匿名埋点，否则会污染用户维度。 + httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error") + return 0, 0, false + } + return user.GetCountryId(), user.GetRegionId(), true +} + +type appTrackingEventDefaults struct { + appCode string + userID int64 + countryID int64 + regionID int64 + deviceID string + sessionID string + platform string + appVersion string + language string + timezone string + receivedAtMS int64 +} + +func appTrackingEventFromRequest(item appEventRequest, defaults appTrackingEventDefaults) (client.AppTrackingEvent, bool) { + item.EventID = strings.TrimSpace(item.EventID) + item.EventName = strings.TrimSpace(item.EventName) + if item.EventID == "" || item.EventName == "" { + return client.AppTrackingEvent{}, false + } + deviceID := firstTrimmed(item.DeviceID, defaults.deviceID) + if defaults.userID <= 0 && deviceID == "" { + // 匿名埋点没有服务端 user_id，只能依赖稳定 device_id 做排障和后续粗粒度去重。 + return client.AppTrackingEvent{}, false + } + properties, ok := normalizeAppEventProperties(item.Properties) + if !ok { + return client.AppTrackingEvent{}, false + } + occurredAtMS := item.OccurredAtMS + if occurredAtMS <= 0 { + occurredAtMS = defaults.receivedAtMS + } + return client.AppTrackingEvent{ + AppCode: defaults.appCode, + EventID: item.EventID, + EventName: item.EventName, + EventType: strings.TrimSpace(item.EventType), + Screen: strings.TrimSpace(item.Screen), + TargetType: strings.TrimSpace(item.TargetType), + TargetID: strings.TrimSpace(item.TargetID), + UserID: defaults.userID, + DeviceID: deviceID, + SessionID: firstTrimmed(defaults.sessionID, item.SessionID), + Platform: strings.ToLower(firstTrimmed(item.Platform, defaults.platform)), + AppVersion: firstTrimmed(item.AppVersion, defaults.appVersion), + Language: firstTrimmed(item.Language, defaults.language), + Timezone: firstTrimmed(item.Timezone, defaults.timezone), + CountryID: defaults.countryID, + RegionID: defaults.regionID, + DurationMS: item.DurationMS, + Success: item.Success, + ErrorCode: strings.TrimSpace(item.ErrorCode), + Properties: properties, + OccurredAtMS: occurredAtMS, + }, true +} + +func normalizeAppEventProperties(raw json.RawMessage) (json.RawMessage, bool) { + trimmed := bytes.TrimSpace(raw) + if len(trimmed) == 0 || bytes.Equal(trimmed, []byte("null")) { + return nil, true + } + if len(trimmed) > appEventMaxPropertiesBytes || !json.Valid(trimmed) { + return nil, false + } + return json.RawMessage(trimmed), true +} + +func appEventRequestLanguage(request *http.Request) string { + value := httpkit.FirstHeader(request, "X-App-Language", "X-Language") + if value == "" { + value = request.Header.Get("Accept-Language") + } + if index := strings.Index(value, ","); index >= 0 { + value = value[:index] + } + return strings.TrimSpace(value) +} + +func firstTrimmed(values ...string) string { + for _, value := range values { + if value = strings.TrimSpace(value); value != "" { + return value + } + } + return "" +} diff --git a/services/gateway-service/internal/transport/http/appapi/handler.go b/services/gateway-service/internal/transport/http/appapi/handler.go index b4b98727..ee563228 100644 --- a/services/gateway-service/internal/transport/http/appapi/handler.go +++ b/services/gateway-service/internal/transport/http/appapi/handler.go @@ -32,6 +32,7 @@ type Handler struct { userAuthClient client.UserAuthClient userProfileClient client.UserProfileClient userDeviceClient client.UserDeviceClient + statisticsClient client.StatisticsClient objectUploader ObjectUploader } @@ -40,6 +41,7 @@ type Config struct { UserAuthClient client.UserAuthClient UserProfileClient client.UserProfileClient UserDeviceClient client.UserDeviceClient + StatisticsClient client.StatisticsClient ObjectUploader ObjectUploader } @@ -49,6 +51,7 @@ func New(config Config) *Handler { userAuthClient: config.UserAuthClient, userProfileClient: config.UserProfileClient, userDeviceClient: config.UserDeviceClient, + statisticsClient: config.StatisticsClient, objectUploader: config.ObjectUploader, } } @@ -104,3 +107,7 @@ func (h *Handler) HandlePushToken(writer http.ResponseWriter, request *http.Requ func (h *Handler) AppHeartbeat(writer http.ResponseWriter, request *http.Request) { h.appHeartbeat(writer, request) } + +func (h *Handler) ReportAppEvents(writer http.ResponseWriter, request *http.Request) { + h.reportAppEvents(writer, request) +} diff --git a/services/gateway-service/internal/transport/http/httproutes/router.go b/services/gateway-service/internal/transport/http/httproutes/router.go index 34ef22df..bd1d393a 100644 --- a/services/gateway-service/internal/transport/http/httproutes/router.go +++ b/services/gateway-service/internal/transport/http/httproutes/router.go @@ -65,6 +65,7 @@ type AppHandlers struct { ProxyEffectMedia http.HandlerFunc HandlePushToken http.HandlerFunc AppHeartbeat http.HandlerFunc + ReportAppEvents http.HandlerFunc } type UserHandlers struct { @@ -413,6 +414,7 @@ func (r routes) registerAppRoutes() { r.public("/files/registration/avatar/upload", http.MethodPost, h.UploadRegistrationAvatar) r.public("/media/effects/proxy", http.MethodGet, h.ProxyEffectMedia) r.auth("/app/heartbeat", http.MethodPost, h.AppHeartbeat) + r.public("/app/events", http.MethodPost, h.ReportAppEvents) r.profile("/devices/push-token", "", h.HandlePushToken) } diff --git a/services/gateway-service/internal/transport/http/response_test.go b/services/gateway-service/internal/transport/http/response_test.go index 3499b58e..7881c0ce 100644 --- a/services/gateway-service/internal/transport/http/response_test.go +++ b/services/gateway-service/internal/transport/http/response_test.go @@ -31,6 +31,7 @@ import ( "hyapp/pkg/xerr" "hyapp/services/gateway-service/internal/appconfig" "hyapp/services/gateway-service/internal/auth" + gatewayclient "hyapp/services/gateway-service/internal/client" "hyapp/services/gateway-service/internal/financewithdrawal" ) @@ -413,6 +414,30 @@ type fakeUserDeviceClient struct { deleteErr error } +type fakeStatisticsClient struct { + lastAppEvents []gatewayclient.AppTrackingEvent + appResult gatewayclient.AppTrackingReportResult + appErr error + selfEvents []gatewayclient.SelfGameH5Event + selfErr error +} + +func (f *fakeStatisticsClient) ReportSelfGameH5Event(_ context.Context, event gatewayclient.SelfGameH5Event) error { + f.selfEvents = append(f.selfEvents, event) + return f.selfErr +} + +func (f *fakeStatisticsClient) ReportAppTrackingEvents(_ context.Context, events []gatewayclient.AppTrackingEvent) (gatewayclient.AppTrackingReportResult, error) { + f.lastAppEvents = append([]gatewayclient.AppTrackingEvent(nil), events...) + if f.appErr != nil { + return gatewayclient.AppTrackingReportResult{}, f.appErr + } + if f.appResult.ServerTimeMS == 0 { + f.appResult = gatewayclient.AppTrackingReportResult{Accepted: true, Received: len(events), Stored: len(events), Duplicated: 0, ServerTimeMS: 1_778_000_005_000} + } + return f.appResult, nil +} + type fakeUserCountryQueryClient struct { last *userv1.ListRegistrationCountriesRequest resp *userv1.ListRegistrationCountriesResponse @@ -3776,6 +3801,115 @@ func TestAppHeartbeatUsesAuthenticatedSession(t *testing.T) { } } +func TestReportAppEventsUsesAuthenticatedUserDimensions(t *testing.T) { + profileClient := &fakeUserProfileClient{regionID: 210} + statisticsClient := &fakeStatisticsClient{} + handler := NewHandlerWithClients(&fakeRoomClient{}, nil, nil, profileClient) + handler.SetStatisticsClient(statisticsClient) + router := handler.Routes(auth.NewVerifier("secret")) + body := []byte(`{"events":[{"event_id":"evt-login-1","event_name":"banner_click","event_type":"click","screen":"home","target_type":"banner","target_id":"17","device_id":"dev-body","user_id":999,"country_id":1,"region_id":2,"platform":"ios","app_version":"1.2.3","language":"en-US","timezone":"Asia/Shanghai","duration_ms":12,"success":true,"properties":{"slot":"home"}}]}`) + request := httptest.NewRequest(http.MethodPost, "/api/v1/app/events", bytes.NewReader(body)) + request.Header.Set("Authorization", "Bearer "+signGatewayTokenWithAppCode(t, "secret", 42, "fami")) + request.Header.Set("X-Device-ID", "dev-header") + request.Header.Set("X-Request-ID", "req-app-events") + recorder := httptest.NewRecorder() + + router.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusOK { + t.Fatalf("status mismatch: got %d body=%s", recorder.Code, recorder.Body.String()) + } + if len(statisticsClient.lastAppEvents) != 1 { + t.Fatalf("statistics app events not called: %+v", statisticsClient.lastAppEvents) + } + event := statisticsClient.lastAppEvents[0] + if event.AppCode != "fami" || event.UserID != 42 || event.CountryID != 840 || event.RegionID != 210 { + t.Fatalf("authenticated event must use server dimensions: %+v", event) + } + if event.DeviceID != "dev-body" || event.EventName != "banner_click" || event.Platform != "ios" || string(event.Properties) != `{"slot":"home"}` { + t.Fatalf("authenticated event fields mismatch: %+v properties=%s", event, string(event.Properties)) + } + var response httpkit.ResponseEnvelope + if err := json.NewDecoder(recorder.Body).Decode(&response); err != nil { + t.Fatalf("decode app events response failed: %v", err) + } + data := response.Data.(map[string]any) + if data["accepted"] != true || data["received"].(float64) != 1 || data["stored"].(float64) != 1 || data["duplicated"].(float64) != 0 { + t.Fatalf("app events response mismatch: %+v", data) + } +} + +func TestReportAppEventsAllowsAnonymousWithDeviceID(t *testing.T) { + statisticsClient := &fakeStatisticsClient{} + handler := NewHandlerWithClients(&fakeRoomClient{}, nil, nil, &fakeUserProfileClient{}) + handler.SetStatisticsClient(statisticsClient) + router := handler.Routes(auth.NewVerifier("secret")) + body := []byte(`{"events":[{"event_id":"evt-anon-1","event_name":"splash_view"}]}`) + request := httptest.NewRequest(http.MethodPost, "/api/v1/app/events", bytes.NewReader(body)) + request.Header.Set("X-Device-ID", "dev-anon") + request.Header.Set("X-App-Platform", "android") + request.Header.Set("X-App-Version", "2.0.1") + request.Header.Set("Accept-Language", "ar-EG,ar;q=0.9") + recorder := httptest.NewRecorder() + + router.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusOK { + t.Fatalf("status mismatch: got %d body=%s", recorder.Code, recorder.Body.String()) + } + if len(statisticsClient.lastAppEvents) != 1 { + t.Fatalf("statistics app events not called: %+v", statisticsClient.lastAppEvents) + } + event := statisticsClient.lastAppEvents[0] + if event.UserID != 0 || event.DeviceID != "dev-anon" || event.Platform != "android" || event.AppVersion != "2.0.1" || event.Language != "ar-EG" { + t.Fatalf("anonymous event must use device/header fields: %+v", event) + } +} + +func TestReportAppEventsRejectsInvalidRequests(t *testing.T) { + for _, test := range []struct { + name string + body []byte + }{ + {name: "missing device anonymous", body: []byte(`{"events":[{"event_id":"evt-1","event_name":"open"}]}`)}, + {name: "missing event id", body: []byte(`{"events":[{"event_name":"open","device_id":"dev-1"}]}`)}, + {name: "missing event name", body: []byte(`{"events":[{"event_id":"evt-1","device_id":"dev-1"}]}`)}, + {name: "too many events", body: mustAppEventsBody(t, 51)}, + } { + t.Run(test.name, func(t *testing.T) { + statisticsClient := &fakeStatisticsClient{} + handler := NewHandlerWithClients(&fakeRoomClient{}, nil, nil, &fakeUserProfileClient{}) + handler.SetStatisticsClient(statisticsClient) + router := handler.Routes(auth.NewVerifier("secret")) + request := httptest.NewRequest(http.MethodPost, "/api/v1/app/events", bytes.NewReader(test.body)) + request.Header.Set("X-Request-ID", "req-invalid-app-events") + recorder := httptest.NewRecorder() + + router.ServeHTTP(recorder, request) + + assertEnvelope(t, recorder, http.StatusBadRequest, httpkit.CodeInvalidArgument, "req-invalid-app-events") + if len(statisticsClient.lastAppEvents) != 0 { + t.Fatalf("invalid app events must not reach statistics-service: %+v", statisticsClient.lastAppEvents) + } + }) + } +} + +func TestReportAppEventsUpstreamFailure(t *testing.T) { + statisticsClient := &fakeStatisticsClient{appErr: errors.New("statistics unavailable")} + handler := NewHandlerWithClients(&fakeRoomClient{}, nil, nil, &fakeUserProfileClient{}) + handler.SetStatisticsClient(statisticsClient) + router := handler.Routes(auth.NewVerifier("secret")) + body := []byte(`{"events":[{"event_id":"evt-upstream","event_name":"open","device_id":"dev-1"}]}`) + request := httptest.NewRequest(http.MethodPost, "/api/v1/app/events", bytes.NewReader(body)) + request.Header.Set("X-Request-ID", "req-app-events-upstream") + recorder := httptest.NewRecorder() + + router.ServeHTTP(recorder, request) + + assertEnvelope(t, recorder, http.StatusBadGateway, httpkit.CodeUpstreamError, "req-app-events-upstream") +} + func TestDeletePushTokenUsesAuthenticatedUser(t *testing.T) { deviceClient := &fakeUserDeviceClient{} handler := NewHandlerWithClients(&fakeRoomClient{}, nil, nil, &fakeUserProfileClient{}) @@ -10364,6 +10498,23 @@ func assertEnvelope(t *testing.T, recorder *httptest.ResponseRecorder, statusCod } } +func mustAppEventsBody(t *testing.T, count int) []byte { + t.Helper() + events := make([]map[string]any, 0, count) + for i := 0; i < count; i++ { + events = append(events, map[string]any{ + "event_id": "evt-" + strconv.Itoa(i), + "event_name": "open", + "device_id": "dev-1", + }) + } + payload, err := json.Marshal(map[string]any{"events": events}) + if err != nil { + t.Fatalf("marshal app events body failed: %v", err) + } + return payload +} + func assertEnvelopeMessage(t *testing.T, recorder *httptest.ResponseRecorder, statusCode int, code string, message string, requestID string) { t.Helper() diff --git a/services/gateway-service/internal/transport/http/router.go b/services/gateway-service/internal/transport/http/router.go index 92fef72e..7f3cccb9 100644 --- a/services/gateway-service/internal/transport/http/router.go +++ b/services/gateway-service/internal/transport/http/router.go @@ -119,6 +119,7 @@ func (h *Handler) Routes(jwtVerifier *auth.Verifier) http.Handler { UserAuthClient: h.userClient, UserProfileClient: h.userProfileClient, UserDeviceClient: h.userDeviceClient, + StatisticsClient: h.statisticsClient, ObjectUploader: h.objectUploader, }) callbackAPI := callbackapi.New(callbackapi.Config{ @@ -204,6 +205,7 @@ func (h *Handler) Routes(jwtVerifier *auth.Verifier) http.Handler { ProxyEffectMedia: appAPI.ProxyEffectMedia, HandlePushToken: appAPI.HandlePushToken, AppHeartbeat: appAPI.AppHeartbeat, + ReportAppEvents: appAPI.ReportAppEvents, }, User: userHandlers, Manager: managerAPI.ManagerHandlers(), 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 cdc2f9b9..d45961de 100644 --- a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql +++ b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql @@ -17,6 +17,37 @@ CREATE TABLE IF NOT EXISTS statistics_event_consumption ( 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 app_tracking_events ( + app_code VARCHAR(32) NOT NULL COMMENT '应用编码', + event_id VARCHAR(160) NOT NULL COMMENT '客户端生成的埋点幂等 ID', + event_name VARCHAR(96) NOT NULL COMMENT '事件名', + event_type VARCHAR(64) NOT NULL DEFAULT '' COMMENT '事件类型，如 click、view、submit', + screen VARCHAR(128) NOT NULL DEFAULT '' COMMENT '页面或模块', + target_type VARCHAR(64) NOT NULL DEFAULT '' COMMENT '目标类型，如 banner、button、room', + target_id VARCHAR(128) NOT NULL DEFAULT '' COMMENT '目标 ID', + user_id BIGINT NOT NULL DEFAULT 0 COMMENT '服务端识别的用户 ID，匿名为 0', + device_id VARCHAR(128) NOT NULL DEFAULT '' COMMENT '设备 ID', + session_id VARCHAR(160) NOT NULL DEFAULT '' COMMENT '会话 ID', + platform VARCHAR(32) NOT NULL DEFAULT '' COMMENT '平台', + app_version VARCHAR(64) NOT NULL DEFAULT '' COMMENT 'App 版本', + language VARCHAR(32) NOT NULL DEFAULT '' COMMENT '语言', + timezone VARCHAR(64) NOT NULL DEFAULT '' COMMENT '客户端时区，仅作属性记录', + country_id BIGINT NOT NULL DEFAULT 0 COMMENT '服务端用户国家维度', + region_id BIGINT NOT NULL DEFAULT 0 COMMENT '服务端用户区域维度', + duration_ms BIGINT NOT NULL DEFAULT 0 COMMENT '客户端记录的耗时毫秒', + success TINYINT(1) NOT NULL DEFAULT 0 COMMENT '客户端动作是否成功', + error_code VARCHAR(64) NOT NULL DEFAULT '' COMMENT '客户端错误码', + properties_json JSON NOT NULL COMMENT '事件扩展属性', + occurred_at_ms BIGINT NOT NULL COMMENT '客户端事件发生时间，UTC epoch ms', + received_at_ms BIGINT NOT NULL COMMENT '服务端接收时间，UTC epoch ms', + stat_day DATE NOT NULL COMMENT 'UTC 统计日', + PRIMARY KEY (app_code, event_id), + KEY idx_app_tracking_event_name (app_code, stat_day, event_name), + KEY idx_app_tracking_screen (app_code, stat_day, screen), + KEY idx_app_tracking_user (app_code, user_id, occurred_at_ms), + KEY idx_app_tracking_device (app_code, device_id, occurred_at_ms) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='App 埋点原始明细表'; + CREATE TABLE IF NOT EXISTS stat_app_day_country ( app_code VARCHAR(32) NOT NULL COMMENT '应用编码', stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai', diff --git a/services/statistics-service/internal/app/query_http.go b/services/statistics-service/internal/app/query_http.go index 0b9f4088..4d39ace5 100644 --- a/services/statistics-service/internal/app/query_http.go +++ b/services/statistics-service/internal/app/query_http.go @@ -34,6 +34,7 @@ func newQueryHTTPServer(addr string, repo *mysqlstorage.Repository) (*queryHTTPS mux.HandleFunc("/internal/v1/statistics/overview", s.overview) mux.HandleFunc("/internal/v1/statistics/platform-grants/users", s.platformGrantUsers) mux.HandleFunc("/internal/v1/statistics/platform-grants/records", s.platformGrantRecords) + mux.HandleFunc("/internal/v1/statistics/app/events", s.appEvents) mux.HandleFunc("/internal/v1/statistics/self-game/events", s.selfGameEvents) mux.HandleFunc("/internal/v1/statistics/self-games/overview", s.selfGameOverview) s.server = &http.Server{Handler: mux, ReadHeaderTimeout: 2 * time.Second} @@ -163,6 +164,84 @@ type selfGameH5EventRequest struct { OccurredAtMS int64 `json:"occurred_at_ms"` } +type appEventsRequest struct { + Events []appTrackingEventRequest `json:"events"` +} + +type appTrackingEventRequest struct { + AppCode string `json:"app_code"` + EventID string `json:"event_id"` + EventName string `json:"event_name"` + EventType string `json:"event_type"` + Screen string `json:"screen"` + TargetType string `json:"target_type"` + TargetID string `json:"target_id"` + UserID int64 `json:"user_id"` + DeviceID string `json:"device_id"` + SessionID string `json:"session_id"` + Platform string `json:"platform"` + AppVersion string `json:"app_version"` + Language string `json:"language"` + Timezone string `json:"timezone"` + CountryID int64 `json:"country_id"` + RegionID int64 `json:"region_id"` + DurationMS int64 `json:"duration_ms"` + Success bool `json:"success"` + ErrorCode string `json:"error_code"` + Properties json.RawMessage `json:"properties"` + OccurredAtMS int64 `json:"occurred_at_ms"` +} + +func (s *queryHTTPServer) appEvents(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeJSON(w, http.StatusMethodNotAllowed, map[string]any{"error": "method not allowed"}) + return + } + defer r.Body.Close() + var body appEventsRequest + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]any{"error": "invalid json"}) + return + } + if len(body.Events) == 0 || len(body.Events) > 50 { + writeJSON(w, http.StatusBadRequest, map[string]any{"error": "events batch is invalid"}) + return + } + events := make([]mysqlstorage.AppTrackingEvent, 0, len(body.Events)) + for _, item := range body.Events { + // Internal ingest 只接受 gateway 已解析过的 App、用户和维度；statistics-service 只负责幂等落表，不回查 owner service。 + events = append(events, mysqlstorage.AppTrackingEvent{ + AppCode: appcode.Normalize(item.AppCode), + EventID: item.EventID, + EventName: item.EventName, + EventType: item.EventType, + Screen: item.Screen, + TargetType: item.TargetType, + TargetID: item.TargetID, + UserID: item.UserID, + DeviceID: item.DeviceID, + SessionID: item.SessionID, + Platform: item.Platform, + AppVersion: item.AppVersion, + Language: item.Language, + Timezone: item.Timezone, + CountryID: item.CountryID, + RegionID: item.RegionID, + DurationMS: item.DurationMS, + Success: item.Success, + ErrorCode: item.ErrorCode, + Properties: item.Properties, + OccurredAtMS: item.OccurredAtMS, + }) + } + result, err := s.repo.ConsumeAppTrackingEvents(r.Context(), events) + if err != nil { + writeJSON(w, http.StatusBadRequest, map[string]any{"error": err.Error()}) + return + } + writeJSON(w, http.StatusOK, result) +} + func (s *queryHTTPServer) selfGameEvents(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { writeJSON(w, http.StatusMethodNotAllowed, map[string]any{"error": "method not allowed"}) diff --git a/services/statistics-service/internal/storage/mysql/query_test.go b/services/statistics-service/internal/storage/mysql/query_test.go index 1b47a469..d44d306b 100644 --- a/services/statistics-service/internal/storage/mysql/query_test.go +++ b/services/statistics-service/internal/storage/mysql/query_test.go @@ -2,11 +2,14 @@ package mysql import ( "context" + "encoding/json" "runtime" + "strings" "testing" "time" "hyapp/internal/testutil/mysqlschema" + "hyapp/pkg/appcode" ) func TestQueryOverviewUsesActivityLuckyGiftDayStats(t *testing.T) { @@ -65,6 +68,120 @@ func TestQueryOverviewUsesActivityLuckyGiftDayStats(t *testing.T) { } } +func TestConsumeAppTrackingEventsStoresRawEventsAndDeduplicates(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_app_tracking_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, 7, 2, 23, 59, 59, 0, time.UTC).UnixMilli() + result, err := repository.ConsumeAppTrackingEvents(appcode.WithContext(ctx, "lalu"), []AppTrackingEvent{ + { + AppCode: "lalu", + EventID: "evt-banner-1", + EventName: "banner_view", + EventType: "view", + Screen: "home", + TargetType: "banner", + TargetID: "17", + UserID: 42, + DeviceID: "dev-1", + SessionID: "sess-1", + Platform: "ios", + AppVersion: "1.2.3", + Language: "en-US", + Timezone: "Asia/Shanghai", + CountryID: 86, + RegionID: 210, + DurationMS: 12, + Success: true, + Properties: json.RawMessage(`{"slot":"top"}`), + OccurredAtMS: occurredAt, + }, + }) + if err != nil { + t.Fatalf("consume app tracking event: %v", err) + } + if !result.Accepted || result.Received != 1 || result.Stored != 1 || result.Duplicated != 0 || result.ServerTimeMS <= 0 { + t.Fatalf("unexpected first consume result: %+v", result) + } + + duplicate, err := repository.ConsumeAppTrackingEvents(appcode.WithContext(ctx, "lalu"), []AppTrackingEvent{ + { + AppCode: "lalu", + EventID: "evt-banner-1", + EventName: "banner_view", + Screen: "changed", + UserID: 42, + DeviceID: "dev-1", + OccurredAtMS: occurredAt, + }, + }) + if err != nil { + t.Fatalf("consume duplicate app tracking event: %v", err) + } + if duplicate.Received != 1 || duplicate.Stored != 0 || duplicate.Duplicated != 1 { + t.Fatalf("duplicate consume result mismatch: %+v", duplicate) + } + + var count int64 + var statDay, screen, propertiesJSON string + var userID, countryID, regionID, durationMS int64 + var success bool + if err := repository.db.QueryRowContext(ctx, ` + SELECT COUNT(*), MAX(CAST(stat_day AS CHAR)), MAX(screen), MAX(user_id), MAX(country_id), MAX(region_id), + MAX(duration_ms), MAX(success), COALESCE(MAX(CAST(properties_json AS CHAR)), '{}') + FROM app_tracking_events + WHERE app_code = 'lalu' AND event_id = 'evt-banner-1' + `).Scan(&count, &statDay, &screen, &userID, &countryID, ®ionID, &durationMS, &success, &propertiesJSON); err != nil { + t.Fatalf("query app tracking event: %v", err) + } + if count != 1 || statDay != "2026-07-02" || screen != "home" || userID != 42 || countryID != 86 || regionID != 210 || durationMS != 12 || !success { + t.Fatalf("stored event mismatch: count=%d statDay=%s screen=%s userID=%d countryID=%d regionID=%d durationMS=%d success=%v", count, statDay, screen, userID, countryID, regionID, durationMS, success) + } + if DecodeJSON(propertiesJSON)["slot"] != "top" { + t.Fatalf("properties_json mismatch: %s", propertiesJSON) + } +} + +func TestConsumeAppTrackingEventsRejectsInvalidProperties(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_app_tracking_invalid_test", + }) + repository, err := Open(ctx, statsSchema.DSN) + if err != nil { + t.Fatalf("open repository: %v", err) + } + t.Cleanup(func() { _ = repository.Close() }) + + for _, test := range []struct { + name string + properties json.RawMessage + }{ + {name: "invalid json", properties: json.RawMessage(`{"bad"`)}, + {name: "too large json", properties: json.RawMessage(`"` + strings.Repeat("a", appTrackingMaxPropertiesBytes+1) + `"`)}, + } { + t.Run(test.name, func(t *testing.T) { + _, err := repository.ConsumeAppTrackingEvents(appcode.WithContext(ctx, "lalu"), []AppTrackingEvent{ + {EventID: "evt-" + test.name, EventName: "open", DeviceID: "dev-1", Properties: test.properties, OccurredAtMS: time.Date(2026, 7, 2, 0, 0, 0, 0, time.UTC).UnixMilli()}, + }) + if err == nil { + t.Fatalf("invalid properties must be rejected") + } + }) + } +} + func TestConsumeUserRegisteredCountsOnlyNewRegistration(t *testing.T) { ctx := context.Background() _, file, _, _ := runtime.Caller(0) diff --git a/services/statistics-service/internal/storage/mysql/repository.go b/services/statistics-service/internal/storage/mysql/repository.go index 459053e5..576635db 100644 --- a/services/statistics-service/internal/storage/mysql/repository.go +++ b/services/statistics-service/internal/storage/mysql/repository.go @@ -1,6 +1,7 @@ package mysql import ( + "bytes" "context" "database/sql" "encoding/json" @@ -21,6 +22,11 @@ const ( SourceH5 = "h5" ) +const ( + appTrackingMaxBatchSize = 50 + appTrackingMaxPropertiesBytes = 4096 +) + // Repository owns the statistics read model. Online queries must use aggregate // tables; the optional activity connection only reads lucky-gift day snapshots, // never the high-cardinality draw fact table. @@ -103,6 +109,24 @@ func (r *Repository) Migrate(ctx context.Context) error { PRIMARY KEY (app_code, source, event_id), KEY idx_event_consumption_type (app_code, source, event_type, consumed_at_ms) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, + `CREATE TABLE IF NOT EXISTS app_tracking_events ( + app_code VARCHAR(32) NOT NULL, event_id VARCHAR(160) NOT NULL, event_name VARCHAR(96) NOT NULL, + event_type VARCHAR(64) NOT NULL DEFAULT '', screen VARCHAR(128) NOT NULL DEFAULT '', + target_type VARCHAR(64) NOT NULL DEFAULT '', target_id VARCHAR(128) NOT NULL DEFAULT '', + user_id BIGINT NOT NULL DEFAULT 0, device_id VARCHAR(128) NOT NULL DEFAULT '', + session_id VARCHAR(160) NOT NULL DEFAULT '', platform VARCHAR(32) NOT NULL DEFAULT '', + app_version VARCHAR(64) NOT NULL DEFAULT '', language VARCHAR(32) NOT NULL DEFAULT '', + timezone VARCHAR(64) NOT NULL DEFAULT '', country_id BIGINT NOT NULL DEFAULT 0, + region_id BIGINT NOT NULL DEFAULT 0, duration_ms BIGINT NOT NULL DEFAULT 0, + success TINYINT(1) NOT NULL DEFAULT 0, error_code VARCHAR(64) NOT NULL DEFAULT '', + properties_json JSON NOT NULL, occurred_at_ms BIGINT NOT NULL, received_at_ms BIGINT NOT NULL, + stat_day DATE NOT NULL, + PRIMARY KEY (app_code, event_id), + KEY idx_app_tracking_event_name (app_code, stat_day, event_name), + KEY idx_app_tracking_screen (app_code, stat_day, screen), + KEY idx_app_tracking_user (app_code, user_id, occurred_at_ms), + KEY idx_app_tracking_device (app_code, device_id, occurred_at_ms) + ) 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0, region_id BIGINT NOT NULL DEFAULT 0, @@ -352,6 +376,7 @@ func (r *Repository) Migrate(ctx context.Context) error { } } for table, columns := range map[string][]string{ + "app_tracking_events": {"app_code", "event_id"}, "stat_app_day_country": {"app_code", "stat_tz", "stat_day", "country_id", "region_id"}, "stat_user_day_activity": {"app_code", "stat_tz", "stat_day", "user_id"}, "stat_user_registration": {"app_code", "stat_tz", "user_id"}, @@ -380,6 +405,10 @@ func (r *Repository) Migrate(ctx context.Context) error { name string columns []string }{ + {"app_tracking_events", "idx_app_tracking_event_name", []string{"app_code", "stat_day", "event_name"}}, + {"app_tracking_events", "idx_app_tracking_screen", []string{"app_code", "stat_day", "screen"}}, + {"app_tracking_events", "idx_app_tracking_user", []string{"app_code", "user_id", "occurred_at_ms"}}, + {"app_tracking_events", "idx_app_tracking_device", []string{"app_code", "device_id", "occurred_at_ms"}}, {"stat_app_day_country", "idx_stat_app_day_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}}, {"stat_user_day_activity", "idx_stat_user_day_country", []string{"app_code", "stat_tz", "stat_day", "country_id"}}, {"stat_user_day_activity", "idx_stat_user_day_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}}, @@ -657,6 +686,38 @@ type SelfGameH5Event struct { OccurredAtMS int64 } +type AppTrackingEvent struct { + AppCode string + EventID string + EventName string + EventType string + Screen string + TargetType string + TargetID string + UserID int64 + DeviceID string + SessionID string + Platform string + AppVersion string + Language string + Timezone string + CountryID int64 + RegionID int64 + DurationMS int64 + Success bool + ErrorCode string + Properties json.RawMessage + OccurredAtMS int64 +} + +type AppTrackingReportResult struct { + Accepted bool `json:"accepted"` + Received int `json:"received"` + Stored int `json:"stored"` + Duplicated int `json:"duplicated"` + ServerTimeMS int64 `json:"server_time_ms"` +} + type SelfGameMatchParticipantEvent struct { UserID int64 ParticipantType string @@ -1286,6 +1347,52 @@ func (r *Repository) ConsumeSelfGameH5Event(ctx context.Context, event SelfGameH }) } +func (r *Repository) ConsumeAppTrackingEvents(ctx context.Context, events []AppTrackingEvent) (AppTrackingReportResult, error) { + if r == nil || r.db == nil { + return AppTrackingReportResult{}, xerr.New(xerr.Unavailable, "mysql repository is not configured") + } + nowMS := time.Now().UTC().UnixMilli() + result := AppTrackingReportResult{Accepted: true, Received: len(events), ServerTimeMS: nowMS} + if len(events) == 0 || len(events) > appTrackingMaxBatchSize { + return AppTrackingReportResult{}, xerr.New(xerr.InvalidArgument, "app tracking events batch is invalid") + } + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return AppTrackingReportResult{}, err + } + defer func() { _ = tx.Rollback() }() + for _, event := range events { + normalized, propertiesJSON, err := normalizeAppTrackingEvent(ctx, event, nowMS) + if err != nil { + return AppTrackingReportResult{}, err + } + affected, err := insertUnique(ctx, tx, ` + INSERT IGNORE INTO app_tracking_events ( + app_code, event_id, event_name, event_type, screen, target_type, target_id, + user_id, device_id, session_id, platform, app_version, language, timezone, + country_id, region_id, duration_ms, success, error_code, properties_json, + occurred_at_ms, received_at_ms, stat_day + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), ?, ?, ?) + `, normalized.AppCode, normalized.EventID, normalized.EventName, normalized.EventType, normalized.Screen, + normalized.TargetType, normalized.TargetID, normalized.UserID, normalized.DeviceID, normalized.SessionID, + normalized.Platform, normalized.AppVersion, normalized.Language, normalized.Timezone, normalized.CountryID, + normalized.RegionID, normalized.DurationMS, normalized.Success, normalized.ErrorCode, propertiesJSON, + normalized.OccurredAtMS, nowMS, statDay(normalized.OccurredAtMS)) + if err != nil { + return AppTrackingReportResult{}, err + } + if affected > 0 { + result.Stored++ + } else { + result.Duplicated++ + } + } + if err := tx.Commit(); err != nil { + return AppTrackingReportResult{}, err + } + return result, nil +} + func (r *Repository) ConsumeSelfGameMatch(ctx context.Context, event SelfGameMatchEvent) error { return r.withEvent(ctx, SourceGame, event.EventID, event.EventType, func(tx *sql.Tx, nowMS int64) error { countryID, regionID := normalizeDimension(event.CountryID, event.RegionID) @@ -1855,6 +1962,102 @@ func normalizeShortText(value string) string { return strings.TrimSpace(value) } +func normalizeAppTrackingEvent(ctx context.Context, event AppTrackingEvent, receivedAtMS int64) (AppTrackingEvent, string, error) { + var err error + if strings.TrimSpace(event.AppCode) == "" { + event.AppCode = appcode.FromContext(ctx) + } + event.AppCode = appcode.Normalize(event.AppCode) + event.EventID, err = normalizeAppTrackingText(event.EventID, 160) + if err != nil { + return AppTrackingEvent{}, "", err + } + event.EventName, err = normalizeAppTrackingText(event.EventName, 96) + if err != nil { + return AppTrackingEvent{}, "", err + } + if event.EventID == "" || event.EventName == "" { + return AppTrackingEvent{}, "", xerr.New(xerr.InvalidArgument, "app tracking event_id and event_name are required") + } + event.EventType, err = normalizeAppTrackingText(event.EventType, 64) + if err != nil { + return AppTrackingEvent{}, "", err + } + event.Screen, err = normalizeAppTrackingText(event.Screen, 128) + if err != nil { + return AppTrackingEvent{}, "", err + } + event.TargetType, err = normalizeAppTrackingText(event.TargetType, 64) + if err != nil { + return AppTrackingEvent{}, "", err + } + event.TargetID, err = normalizeAppTrackingText(event.TargetID, 128) + if err != nil { + return AppTrackingEvent{}, "", err + } + event.DeviceID, err = normalizeAppTrackingText(event.DeviceID, 128) + if err != nil { + return AppTrackingEvent{}, "", err + } + if event.UserID <= 0 && event.DeviceID == "" { + return AppTrackingEvent{}, "", xerr.New(xerr.InvalidArgument, "anonymous app tracking event requires device_id") + } + event.SessionID, err = normalizeAppTrackingText(event.SessionID, 160) + if err != nil { + return AppTrackingEvent{}, "", err + } + event.Platform, err = normalizeAppTrackingText(strings.ToLower(event.Platform), 32) + if err != nil { + return AppTrackingEvent{}, "", err + } + event.AppVersion, err = normalizeAppTrackingText(event.AppVersion, 64) + if err != nil { + return AppTrackingEvent{}, "", err + } + event.Language, err = normalizeAppTrackingText(event.Language, 32) + if err != nil { + return AppTrackingEvent{}, "", err + } + event.Timezone, err = normalizeAppTrackingText(event.Timezone, 64) + if err != nil { + return AppTrackingEvent{}, "", err + } + event.ErrorCode, err = normalizeAppTrackingText(event.ErrorCode, 64) + if err != nil { + return AppTrackingEvent{}, "", err + } + event.UserID = normalizeID(event.UserID) + event.CountryID, event.RegionID = normalizeDimension(event.CountryID, event.RegionID) + event.DurationMS = normalizeID(event.DurationMS) + if event.OccurredAtMS <= 0 { + event.OccurredAtMS = receivedAtMS + } + propertiesJSON, err := normalizeAppTrackingProperties(event.Properties) + if err != nil { + return AppTrackingEvent{}, "", err + } + return event, propertiesJSON, nil +} + +func normalizeAppTrackingText(value string, maxBytes int) (string, error) { + value = strings.TrimSpace(value) + if maxBytes > 0 && len(value) > maxBytes { + return "", xerr.New(xerr.InvalidArgument, "app tracking text field is too long") + } + return value, nil +} + +func normalizeAppTrackingProperties(raw json.RawMessage) (string, error) { + trimmed := bytes.TrimSpace(raw) + if len(trimmed) == 0 || bytes.Equal(trimmed, []byte("null")) { + return "{}", nil + } + if len(trimmed) > appTrackingMaxPropertiesBytes || !json.Valid(trimmed) { + return "", xerr.New(xerr.InvalidArgument, "app tracking properties_json is invalid") + } + return string(trimmed), nil +} + func normalizeDimension(countryID int64, regionID int64) (int64, int64) { return normalizeID(countryID), normalizeID(regionID) }