添加埋点以及Yumi数据
This commit is contained in:
zhx
parent
0480304483
commit
0cd30e222d
1
.gitignore
vendored
1
.gitignore
vendored
@ -31,3 +31,4 @@ redis-data/
|
||||
|
||||
.tmp/
|
||||
/tmp/
|
||||
.claude/settings.local.json
|
||||
|
||||
108
docs/flutter对接/App埋点Flutter对接.md
Normal file
108
docs/flutter对接/App埋点Flutter对接.md
Normal file
@ -0,0 +1,108 @@
|
||||
# App 埋点 Flutter 对接
|
||||
|
||||
## 地址
|
||||
|
||||
```http
|
||||
POST /api/v1/app/events
|
||||
Content-Type: application/json
|
||||
```
|
||||
|
||||
可登录也可未登录调用。已登录时建议带：
|
||||
|
||||
```http
|
||||
Authorization: Bearer <access_token>
|
||||
```
|
||||
|
||||
未登录时必须传 `device_id` 或 Header：
|
||||
|
||||
```http
|
||||
X-Device-ID: <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
|
||||
|
||||
无。
|
||||
@ -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 {
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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 == "" {
|
||||
|
||||
@ -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 {
|
||||
|
||||
401
server/admin/internal/modules/payment/legacy_bill_source.go
Normal file
401
server/admin/internal/modules/payment/legacy_bill_source.go
Normal file
@ -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))
|
||||
}
|
||||
165
server/admin/internal/modules/payment/legacy_bill_source_test.go
Normal file
165
server/admin/internal/modules/payment/legacy_bill_source_test.go
Normal file
@ -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")
|
||||
}
|
||||
}
|
||||
@ -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())
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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 ""
|
||||
}
|
||||
@ -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)
|
||||
}
|
||||
|
||||
@ -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)
|
||||
}
|
||||
|
||||
|
||||
@ -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()
|
||||
|
||||
|
||||
@ -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(),
|
||||
|
||||
@ -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',
|
||||
|
||||
@ -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"})
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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)
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user