阿斯兰数据
This commit is contained in:
zhx
parent
7f2387ba95
commit
c6ec9fe6aa
@ -446,12 +446,20 @@ func dashboardRegionResolvers(sources []paymentmodule.MoneyRegionSource) []dashb
|
||||
func connectDashboardExternalSources(ctx context.Context, configs []config.DashboardExternalSourceConfig, regionResolvers ...dashboardmodule.ExternalDashboardRegionResolver) ([]dashboardmodule.ExternalDashboardSource, func(), error) {
|
||||
sources := []dashboardmodule.ExternalDashboardSource{}
|
||||
dbs := []*sql.DB{}
|
||||
mongoClients := []*mongo.Client{}
|
||||
cleanup := func() {
|
||||
for _, db := range dbs {
|
||||
if db != nil {
|
||||
_ = db.Close()
|
||||
}
|
||||
}
|
||||
closeCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
for _, client := range mongoClients {
|
||||
if client != nil {
|
||||
_ = client.Disconnect(closeCtx)
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, sourceConfig := range configs {
|
||||
if !sourceConfig.Enabled {
|
||||
@ -469,6 +477,35 @@ func connectDashboardExternalSources(ctx context.Context, configs []config.Dashb
|
||||
sources = append(sources, source)
|
||||
}
|
||||
continue
|
||||
case "aslan_mongo":
|
||||
// Aslan 当前要求 admin-server 直接读外接项目 Mongo 聚合事实；启动期先 ping，
|
||||
// 避免前端选择 Aslan 后才暴露连接或凭证错误。
|
||||
client, err := mongo.Connect(options.Client().ApplyURI(sourceConfig.MongoURI).SetServerSelectionTimeout(sourceConfig.RequestTimeout))
|
||||
if err != nil {
|
||||
cleanup()
|
||||
return nil, nil, err
|
||||
}
|
||||
pingCtx, cancel := context.WithTimeout(ctx, sourceConfig.RequestTimeout)
|
||||
if err := client.Ping(pingCtx, readpref.PrimaryPreferred()); err != nil {
|
||||
cancel()
|
||||
_ = client.Disconnect(ctx)
|
||||
cleanup()
|
||||
return nil, nil, err
|
||||
}
|
||||
cancel()
|
||||
source, err := dashboardmodule.NewAslanMongoDashboardSource(client.Database(sourceConfig.MongoDatabase), sourceConfig, regionResolvers...)
|
||||
if err != nil {
|
||||
_ = client.Disconnect(ctx)
|
||||
cleanup()
|
||||
return nil, nil, err
|
||||
}
|
||||
if source == nil {
|
||||
_ = client.Disconnect(ctx)
|
||||
continue
|
||||
}
|
||||
mongoClients = append(mongoClients, client)
|
||||
sources = append(sources, source)
|
||||
continue
|
||||
case "", "mysql":
|
||||
default:
|
||||
cleanup()
|
||||
|
||||
@ -23,6 +23,14 @@ money_region_sources:
|
||||
mongo_database: "tarab_all"
|
||||
mongo_collection: "sys_region_config"
|
||||
request_timeout: "5s"
|
||||
- enabled: true
|
||||
app_code: "aslan"
|
||||
app_name: "Aslan"
|
||||
sys_origin: "ATYOU"
|
||||
mongo_uri: "mongodb://mongouser:REPLACE_ME@lb-le274w4o-wjgmcdkhye3v0ors.clb.sg-tencentclb.com:27017,lb-le274w4o-wjgmcdkhye3v0ors.clb.sg-tencentclb.com:27018/test?replicaSet=cmgo-6cztxjgr_0&authSource=admin"
|
||||
mongo_database: "tarab_all"
|
||||
mongo_collection: "sys_region_config"
|
||||
request_timeout: "5s"
|
||||
mysql_auto_migrate: false
|
||||
migrations:
|
||||
enabled: true
|
||||
@ -78,11 +86,12 @@ dashboard_external_sources:
|
||||
stat_timezone: "Asia/Riyadh"
|
||||
request_timeout: "5s"
|
||||
- enabled: true
|
||||
type: "aslan_http"
|
||||
type: "aslan_mongo"
|
||||
app_code: "aslan"
|
||||
app_name: "Aslan"
|
||||
base_url: "https://console.atuchat.com"
|
||||
overview_path: "/console/datav/aslan/region-country/statistics"
|
||||
sys_origin: "ATYOU"
|
||||
mongo_uri: "mongodb://mongouser:REPLACE_ME@lb-le274w4o-wjgmcdkhye3v0ors.clb.sg-tencentclb.com:27017,lb-le274w4o-wjgmcdkhye3v0ors.clb.sg-tencentclb.com:27018/test?replicaSet=cmgo-6cztxjgr_0&authSource=admin"
|
||||
mongo_database: "test"
|
||||
stat_timezone: "Asia/Shanghai"
|
||||
request_timeout: "5s"
|
||||
finance_notifications:
|
||||
|
||||
@ -23,6 +23,14 @@ money_region_sources:
|
||||
mongo_database: "tarab_all"
|
||||
mongo_collection: "sys_region_config"
|
||||
request_timeout: "5s"
|
||||
- enabled: false
|
||||
app_code: "aslan"
|
||||
app_name: "Aslan"
|
||||
sys_origin: "ATYOU"
|
||||
mongo_uri: ""
|
||||
mongo_database: "tarab_all"
|
||||
mongo_collection: "sys_region_config"
|
||||
request_timeout: "5s"
|
||||
mysql_auto_migrate: true
|
||||
migrations:
|
||||
enabled: true
|
||||
@ -77,11 +85,12 @@ dashboard_external_sources:
|
||||
stat_timezone: "Asia/Riyadh"
|
||||
request_timeout: "5s"
|
||||
- enabled: false
|
||||
type: "aslan_http"
|
||||
type: "aslan_mongo"
|
||||
app_code: "aslan"
|
||||
app_name: "Aslan"
|
||||
base_url: "https://console.atuchat.com"
|
||||
overview_path: "/console/datav/aslan/region-country/statistics"
|
||||
sys_origin: "ATYOU"
|
||||
mongo_uri: ""
|
||||
mongo_database: "test"
|
||||
stat_timezone: "Asia/Shanghai"
|
||||
request_timeout: "5s"
|
||||
finance_notifications:
|
||||
|
||||
@ -95,6 +95,8 @@ type DashboardExternalSourceConfig struct {
|
||||
AppName string `yaml:"app_name"`
|
||||
SysOrigin string `yaml:"sys_origin"`
|
||||
MySQLDSN string `yaml:"mysql_dsn"`
|
||||
MongoURI string `yaml:"mongo_uri"`
|
||||
MongoDatabase string `yaml:"mongo_database"`
|
||||
BaseURL string `yaml:"base_url"`
|
||||
OverviewPath string `yaml:"overview_path"`
|
||||
StatTimezone string `yaml:"stat_timezone"`
|
||||
@ -262,11 +264,11 @@ func Default() Config {
|
||||
},
|
||||
{
|
||||
Enabled: false,
|
||||
SourceType: "aslan_http",
|
||||
SourceType: "aslan_mongo",
|
||||
AppCode: "aslan",
|
||||
AppName: "Aslan",
|
||||
BaseURL: "https://console.atuchat.com",
|
||||
OverviewPath: "/console/datav/aslan/region-country/statistics",
|
||||
SysOrigin: "ATYOU",
|
||||
MongoDatabase: "test",
|
||||
StatTimezone: "Asia/Shanghai",
|
||||
RequestTimeout: 5 * time.Second,
|
||||
},
|
||||
@ -411,21 +413,28 @@ func (cfg *Config) Normalize() {
|
||||
source.AppName = strings.TrimSpace(source.AppName)
|
||||
source.SysOrigin = strings.ToUpper(strings.TrimSpace(source.SysOrigin))
|
||||
source.MySQLDSN = strings.TrimSpace(source.MySQLDSN)
|
||||
source.MongoURI = strings.TrimSpace(source.MongoURI)
|
||||
source.MongoDatabase = strings.Trim(strings.TrimSpace(source.MongoDatabase), "/")
|
||||
source.BaseURL = strings.TrimRight(strings.TrimSpace(source.BaseURL), "/")
|
||||
source.OverviewPath = "/" + strings.TrimLeft(strings.TrimSpace(source.OverviewPath), "/")
|
||||
if source.OverviewPath == "/" {
|
||||
source.OverviewPath = ""
|
||||
}
|
||||
if source.SourceType == "" {
|
||||
if source.BaseURL != "" || source.OverviewPath != "" {
|
||||
if source.MongoURI != "" || source.MongoDatabase != "" {
|
||||
source.SourceType = "aslan_mongo"
|
||||
} else if source.BaseURL != "" || source.OverviewPath != "" {
|
||||
source.SourceType = "aslan_http"
|
||||
} else {
|
||||
source.SourceType = "mysql"
|
||||
}
|
||||
}
|
||||
if source.SourceType == "aslan_mongo" && source.MongoDatabase == "" {
|
||||
source.MongoDatabase = "test"
|
||||
}
|
||||
source.StatTimezone = strings.TrimSpace(source.StatTimezone)
|
||||
if source.StatTimezone == "" {
|
||||
if source.SourceType == "aslan_http" {
|
||||
if source.SourceType == "aslan_http" || source.SourceType == "aslan_mongo" {
|
||||
source.StatTimezone = "Asia/Shanghai"
|
||||
} else {
|
||||
source.StatTimezone = "Asia/Riyadh"
|
||||
@ -435,6 +444,7 @@ func (cfg *Config) Normalize() {
|
||||
source.RequestTimeout = 5 * time.Second
|
||||
}
|
||||
}
|
||||
cfg.applyDashboardExternalSourceEnvOverrides()
|
||||
cfg.applyFinanceNotificationEnvOverrides()
|
||||
cfg.FinanceNotifications.DingTalk.WebhookURL = strings.TrimSpace(cfg.FinanceNotifications.DingTalk.WebhookURL)
|
||||
cfg.FinanceNotifications.DingTalk.Secret = strings.TrimSpace(cfg.FinanceNotifications.DingTalk.Secret)
|
||||
@ -465,6 +475,7 @@ func (cfg *Config) Normalize() {
|
||||
source.RequestTimeout = 5 * time.Second
|
||||
}
|
||||
}
|
||||
cfg.applyMoneyRegionSourceEnvOverrides()
|
||||
cfg.TencentCOS.SecretID = strings.TrimSpace(cfg.TencentCOS.SecretID)
|
||||
cfg.TencentCOS.SecretKey = strings.TrimSpace(cfg.TencentCOS.SecretKey)
|
||||
cfg.TencentCOS.BucketName = strings.TrimSpace(cfg.TencentCOS.BucketName)
|
||||
@ -605,8 +616,18 @@ func (cfg Config) Validate() error {
|
||||
if strings.TrimSpace(source.OverviewPath) == "" {
|
||||
return fmt.Errorf("dashboard_external_sources[%d].overview_path is required when aslan_http source is enabled", index)
|
||||
}
|
||||
case "aslan_mongo":
|
||||
if strings.TrimSpace(source.SysOrigin) == "" {
|
||||
return fmt.Errorf("dashboard_external_sources[%d].sys_origin is required when aslan_mongo source is enabled", index)
|
||||
}
|
||||
if strings.TrimSpace(source.MongoURI) == "" {
|
||||
return fmt.Errorf("dashboard_external_sources[%d].mongo_uri is required when aslan_mongo source is enabled", index)
|
||||
}
|
||||
if strings.TrimSpace(source.MongoDatabase) == "" {
|
||||
return fmt.Errorf("dashboard_external_sources[%d].mongo_database is required when aslan_mongo source is enabled", index)
|
||||
}
|
||||
default:
|
||||
return fmt.Errorf("dashboard_external_sources[%d].type must be one of mysql, aslan_http", index)
|
||||
return fmt.Errorf("dashboard_external_sources[%d].type must be one of mysql, aslan_http, aslan_mongo", index)
|
||||
}
|
||||
if source.RequestTimeout <= 0 {
|
||||
return fmt.Errorf("dashboard_external_sources[%d].request_timeout must be greater than 0", index)
|
||||
@ -699,6 +720,60 @@ func (cfg *Config) applyFinanceNotificationEnvOverrides() {
|
||||
}
|
||||
}
|
||||
|
||||
func (cfg *Config) applyDashboardExternalSourceEnvOverrides() {
|
||||
for index := range cfg.DashboardExternalSources {
|
||||
source := &cfg.DashboardExternalSources[index]
|
||||
appKey := envAppKey(source.AppCode)
|
||||
if appKey == "" {
|
||||
continue
|
||||
}
|
||||
if value := strings.TrimSpace(firstEnv(
|
||||
"HYAPP_ADMIN_"+appKey+"_DASHBOARD_MONGO_URI",
|
||||
"HYAPP_ADMIN_"+appKey+"_MONGO_URI",
|
||||
)); value != "" {
|
||||
// Mongo URI 含线上密码；仓库 YAML 只放占位，真实值由运行环境注入后覆盖。
|
||||
source.MongoURI = value
|
||||
}
|
||||
if value := strings.TrimSpace(firstEnv(
|
||||
"HYAPP_ADMIN_"+appKey+"_DASHBOARD_MONGO_DATABASE",
|
||||
"HYAPP_ADMIN_"+appKey+"_MONGO_DATABASE",
|
||||
)); value != "" {
|
||||
source.MongoDatabase = strings.Trim(strings.TrimSpace(value), "/")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (cfg *Config) applyMoneyRegionSourceEnvOverrides() {
|
||||
for index := range cfg.MoneyRegionSources {
|
||||
source := &cfg.MoneyRegionSources[index]
|
||||
appKey := envAppKey(source.AppCode)
|
||||
if appKey == "" {
|
||||
continue
|
||||
}
|
||||
if value := strings.TrimSpace(firstEnv(
|
||||
"HYAPP_ADMIN_"+appKey+"_MONEY_REGION_MONGO_URI",
|
||||
"HYAPP_ADMIN_"+appKey+"_MONGO_URI",
|
||||
)); value != "" {
|
||||
// legacy 区域源和 Aslan 大屏可以共享同一个 Mongo 集群；只在运行环境注入真实密码。
|
||||
source.MongoURI = value
|
||||
}
|
||||
if value := strings.TrimSpace(firstEnv(
|
||||
"HYAPP_ADMIN_"+appKey+"_MONEY_REGION_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 == "" {
|
||||
return ""
|
||||
}
|
||||
return strings.NewReplacer("-", "_", ".", "_").Replace(appCode)
|
||||
}
|
||||
|
||||
func firstEnv(keys ...string) string {
|
||||
for _, key := range keys {
|
||||
if value := os.Getenv(key); strings.TrimSpace(value) != "" {
|
||||
|
||||
@ -71,3 +71,40 @@ func TestFinanceDingTalkEnvOverride(t *testing.T) {
|
||||
t.Fatalf("at mobiles mismatch: %#v", cfg.FinanceNotifications.DingTalk.AtMobiles)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAslanMongoEnvOverride(t *testing.T) {
|
||||
t.Setenv("HYAPP_ADMIN_ASLAN_MONGO_URI", "mongodb://mongouser:secret@example:27017/test?authSource=admin")
|
||||
t.Setenv("HYAPP_ADMIN_ASLAN_DASHBOARD_MONGO_DATABASE", "test_dashboard")
|
||||
t.Setenv("HYAPP_ADMIN_ASLAN_MONEY_REGION_MONGO_DATABASE", "tarab_region")
|
||||
|
||||
cfg := Default()
|
||||
cfg.Normalize()
|
||||
|
||||
var dashboardSource DashboardExternalSourceConfig
|
||||
for _, source := range cfg.DashboardExternalSources {
|
||||
if source.AppCode == "aslan" {
|
||||
dashboardSource = source
|
||||
break
|
||||
}
|
||||
}
|
||||
if dashboardSource.MongoURI != "mongodb://mongouser:secret@example:27017/test?authSource=admin" {
|
||||
t.Fatalf("dashboard mongo uri mismatch: %q", dashboardSource.MongoURI)
|
||||
}
|
||||
if dashboardSource.MongoDatabase != "test_dashboard" {
|
||||
t.Fatalf("dashboard mongo database mismatch: %q", dashboardSource.MongoDatabase)
|
||||
}
|
||||
|
||||
var regionSource MoneyRegionSourceConfig
|
||||
for _, source := range cfg.MoneyRegionSources {
|
||||
if source.AppCode == "aslan" {
|
||||
regionSource = source
|
||||
break
|
||||
}
|
||||
}
|
||||
if regionSource.MongoURI != "mongodb://mongouser:secret@example:27017/test?authSource=admin" {
|
||||
t.Fatalf("region mongo uri mismatch: %q", regionSource.MongoURI)
|
||||
}
|
||||
if regionSource.MongoDatabase != "tarab_region" {
|
||||
t.Fatalf("region mongo database mismatch: %q", regionSource.MongoDatabase)
|
||||
}
|
||||
}
|
||||
|
||||
@ -63,6 +63,7 @@ type aslanDashboardMetric struct {
|
||||
RechargeUSDMinor int64
|
||||
ActiveUsers int64
|
||||
NewUsers int64
|
||||
PaidUsers int64
|
||||
CoinTotal int64
|
||||
SalaryUSDMinor int64
|
||||
}
|
||||
@ -407,6 +408,7 @@ func aslanCombinedMetric(flow aslanDashboardMetric, snapshot aslanDashboardMetri
|
||||
metric := externalDashboardMetric{
|
||||
NewUsers: flow.NewUsers,
|
||||
ActiveUsers: flow.ActiveUsers,
|
||||
PaidUsers: flow.PaidUsers,
|
||||
UserRechargeUSDMinor: flow.RechargeUSDMinor,
|
||||
RechargeUSDMinor: flow.RechargeUSDMinor,
|
||||
CoinTotal: int64Ptr(snapshot.CoinTotal),
|
||||
@ -426,6 +428,7 @@ func (m *aslanDashboardMetric) addFlow(item aslanDashboardMetric) {
|
||||
m.RechargeUSDMinor += item.RechargeUSDMinor
|
||||
m.ActiveUsers += item.ActiveUsers
|
||||
m.NewUsers += item.NewUsers
|
||||
m.PaidUsers += item.PaidUsers
|
||||
}
|
||||
|
||||
func (m *aslanDashboardMetric) addSnapshot(item aslanDashboardMetric) {
|
||||
|
||||
690
server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go
Normal file
690
server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go
Normal file
@ -0,0 +1,690 @@
|
||||
package dashboard
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"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"
|
||||
)
|
||||
|
||||
const (
|
||||
aslanMongoInAppPurchaseCollection = "in_app_purchase_details"
|
||||
aslanMongoUserProfileCollection = "user_run_profile"
|
||||
aslanMongoUserActiveCollection = "user_daily_active_log"
|
||||
aslanMongoTeamMemberCollection = "team_member"
|
||||
aslanMongoBankBalanceCollection = "user_bank_balance"
|
||||
aslanMongoChunkSize = 800
|
||||
)
|
||||
|
||||
type AslanMongoDashboardSource struct {
|
||||
db *mongo.Database
|
||||
appCode string
|
||||
appName string
|
||||
sysOrigin string
|
||||
statTimezone string
|
||||
requestTimeout time.Duration
|
||||
regionResolvers []ExternalDashboardRegionResolver
|
||||
}
|
||||
|
||||
type aslanMongoRechargeMetric struct {
|
||||
AmountUSDMinor int64
|
||||
Users int64
|
||||
}
|
||||
|
||||
// NewAslanMongoDashboardSource 直接读取 Aslan 外接项目 Mongo 里的已存在事实集合。
|
||||
// 这里只做查询聚合和字段映射，不把 Aslan 数据复制到 hyapp statistics-service，也不回退旧 console HTTP 接口。
|
||||
func NewAslanMongoDashboardSource(database *mongo.Database, sourceConfig config.DashboardExternalSourceConfig, regionResolvers ...ExternalDashboardRegionResolver) (ExternalDashboardSource, error) {
|
||||
if database == nil || !sourceConfig.Enabled {
|
||||
return nil, nil
|
||||
}
|
||||
appCode := strings.ToLower(strings.TrimSpace(sourceConfig.AppCode))
|
||||
sysOrigin := strings.ToUpper(strings.TrimSpace(sourceConfig.SysOrigin))
|
||||
if appCode == "" || sysOrigin == "" {
|
||||
return nil, nil
|
||||
}
|
||||
statTimezone := strings.TrimSpace(sourceConfig.StatTimezone)
|
||||
if statTimezone == "" {
|
||||
statTimezone = "Asia/Shanghai"
|
||||
}
|
||||
if _, err := time.LoadLocation(statTimezone); err != nil {
|
||||
return nil, fmt.Errorf("load aslan mongo storage timezone %s: %w", statTimezone, err)
|
||||
}
|
||||
timeout := sourceConfig.RequestTimeout
|
||||
if timeout <= 0 {
|
||||
timeout = 5 * time.Second
|
||||
}
|
||||
return &AslanMongoDashboardSource{
|
||||
db: database,
|
||||
appCode: appCode,
|
||||
appName: strings.TrimSpace(sourceConfig.AppName),
|
||||
sysOrigin: sysOrigin,
|
||||
statTimezone: statTimezone,
|
||||
requestTimeout: timeout,
|
||||
regionResolvers: compactExternalDashboardRegionResolvers(regionResolvers),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *AslanMongoDashboardSource) AppCode() string {
|
||||
if s == nil {
|
||||
return ""
|
||||
}
|
||||
return s.appCode
|
||||
}
|
||||
|
||||
func (s *AslanMongoDashboardSource) StatisticsOverview(ctx context.Context, query StatisticsQuery) (map[string]any, error) {
|
||||
if s == nil || s.db == nil {
|
||||
return nil, fmt.Errorf("aslan mongo dashboard source is not configured")
|
||||
}
|
||||
countryFilter, err := s.countryFilter(ctx, query)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
requestTimezone := strings.TrimSpace(query.StatTZ)
|
||||
if requestTimezone == "" {
|
||||
requestTimezone = s.statTimezone
|
||||
}
|
||||
requestLocation, err := time.LoadLocation(requestTimezone)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("load aslan mongo request timezone %s: %w", requestTimezone, err)
|
||||
}
|
||||
startDate, endDate, startMS, endMS := externalDashboardDateRange(query, requestLocation)
|
||||
|
||||
queryCtx, cancel := context.WithTimeout(ctx, s.requestTimeout)
|
||||
defer cancel()
|
||||
|
||||
current, err := s.loadRange(queryCtx, startDate, endDate, countryFilter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
days := countCalendarDays(startDate, endDate)
|
||||
previousStart := startDate.AddDate(0, 0, -days)
|
||||
previous, err := s.loadRange(queryCtx, previousStart, startDate, countryFilter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
response := current.Total.toMap()
|
||||
response["app_code"] = s.appCode
|
||||
response["app_name"] = firstExternalDashboardValue(s.appName, s.appCode)
|
||||
response["sys_origin"] = s.sysOrigin
|
||||
response["stat_tz"] = s.statTimezone
|
||||
response["request_stat_tz"] = requestTimezone
|
||||
response["start_ms"] = startMS
|
||||
response["end_ms"] = endMS
|
||||
response["daily_series"] = current.DailySeries
|
||||
response["country_breakdown"] = current.CountryBreakdown
|
||||
response["daily_country_breakdown"] = current.DailyCountryBreakdown
|
||||
response["retention"] = aslanRetentionMap(current.Total)
|
||||
response["report_metric_sources"] = aslanMongoDashboardMetricSources()
|
||||
response["updated_at_ms"] = time.Now().UTC().UnixMilli()
|
||||
applyExternalDashboardDeltas(response, current.Total, previous.Total)
|
||||
response["salary_delta_rate"] = deltaRate(nullableMetricInt64(current.Total.SalaryUSDMinor), nullableMetricInt64(previous.Total.SalaryUSDMinor))
|
||||
applyAslanMongoUnavailableFields(response)
|
||||
return response, nil
|
||||
}
|
||||
|
||||
func (s *AslanMongoDashboardSource) countryFilter(ctx context.Context, query StatisticsQuery) (externalDashboardCountryFilter, error) {
|
||||
if query.CountryID > 0 {
|
||||
return externalDashboardCountryFilter{}, fmt.Errorf("aslan mongo dashboard source %s does not support numeric country_id filter", s.appCode)
|
||||
}
|
||||
if query.RegionID <= 0 {
|
||||
return externalDashboardCountryFilter{}, nil
|
||||
}
|
||||
for _, resolver := range s.regionResolvers {
|
||||
codes, handled, err := resolver.CountryCodesForRegion(ctx, s.appCode, query.RegionID)
|
||||
if err != nil {
|
||||
return externalDashboardCountryFilter{}, err
|
||||
}
|
||||
if !handled {
|
||||
continue
|
||||
}
|
||||
return externalDashboardCountryFilter{
|
||||
Applied: true,
|
||||
RegionID: query.RegionID,
|
||||
Codes: normalizeExternalCountryCodes(codes),
|
||||
}, nil
|
||||
}
|
||||
// Aslan Mongo 集合只有国家码，没有 hyapp 的数字 region_id；缺 legacy 区域映射时失败，避免误把全量数据当成某个区域返回。
|
||||
return externalDashboardCountryFilter{}, fmt.Errorf("aslan mongo dashboard source %s cannot resolve region_id %d", s.appCode, query.RegionID)
|
||||
}
|
||||
|
||||
func (s *AslanMongoDashboardSource) loadRange(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter) (aslanRangeOverview, error) {
|
||||
totalFlow := aslanDashboardMetric{}
|
||||
totalSnapshot := aslanDashboardMetric{}
|
||||
dailySeries := []map[string]any{}
|
||||
dailyCountries := []map[string]any{}
|
||||
countryAggregates := map[string]*aslanCountryAccumulator{}
|
||||
|
||||
// user_bank_balance 是余额快照而不是日报表；这里按当前快照填充每一天，区间汇总只保留最后一天快照，避免把余额按天累加。
|
||||
salarySnapshot, err := s.loadHostSalaryBalanceByCountry(ctx, countryFilter)
|
||||
if err != nil {
|
||||
return aslanRangeOverview{}, err
|
||||
}
|
||||
|
||||
for day := startDate; day.Before(endDate); day = day.AddDate(0, 0, 1) {
|
||||
nextDay := day.AddDate(0, 0, 1)
|
||||
registers, err := s.loadRegisteredUsersByCountry(ctx, day, nextDay, countryFilter)
|
||||
if err != nil {
|
||||
return aslanRangeOverview{}, err
|
||||
}
|
||||
activeUsers, err := s.loadActiveUsersByCountry(ctx, day, countryFilter)
|
||||
if err != nil {
|
||||
return aslanRangeOverview{}, err
|
||||
}
|
||||
recharges, err := s.loadRechargeByCountry(ctx, day, nextDay, countryFilter)
|
||||
if err != nil {
|
||||
return aslanRangeOverview{}, err
|
||||
}
|
||||
|
||||
dayFlow := aslanDashboardMetric{}
|
||||
daySnapshot := aslanDashboardMetric{}
|
||||
dayCountryAggregates := map[string]*aslanCountryAccumulator{}
|
||||
for _, countryCode := range aslanMongoCountryKeys(registers, activeUsers, recharges, salarySnapshot) {
|
||||
metric := aslanDashboardMetric{
|
||||
NewUsers: registers[countryCode],
|
||||
ActiveUsers: activeUsers[countryCode],
|
||||
PaidUsers: recharges[countryCode].Users,
|
||||
RechargeUSDMinor: recharges[countryCode].AmountUSDMinor,
|
||||
SalaryUSDMinor: salarySnapshot[countryCode],
|
||||
}
|
||||
dayFlow.addFlow(metric)
|
||||
daySnapshot.addSnapshot(metric)
|
||||
totalFlow.addFlow(metric)
|
||||
country := dayCountryAggregates[countryCode]
|
||||
if country == nil {
|
||||
country = &aslanCountryAccumulator{CountryCode: countryCode, CountryName: countryCode, RegionID: countryFilter.RegionID}
|
||||
dayCountryAggregates[countryCode] = country
|
||||
}
|
||||
country.Flow.addFlow(metric)
|
||||
country.Snapshot.addSnapshot(metric)
|
||||
}
|
||||
|
||||
totalSnapshot = daySnapshot
|
||||
dayMetric := aslanCombinedMetric(dayFlow, daySnapshot)
|
||||
dayRow := dayMetric.toMap()
|
||||
dayLabel := dashboardSQLDate(day)
|
||||
dayRow["label"] = dayLabel
|
||||
dayRow["stat_day"] = dayLabel
|
||||
applyAslanMongoUnavailableFields(dayRow)
|
||||
dailySeries = append(dailySeries, dayRow)
|
||||
mergeAslanCountryDay(countryAggregates, dayCountryAggregates)
|
||||
for _, row := range aslanDailyCountryRows(dayLabel, dayCountryAggregates) {
|
||||
applyAslanMongoUnavailableFields(row)
|
||||
dailyCountries = append(dailyCountries, row)
|
||||
}
|
||||
}
|
||||
|
||||
countryRows := aslanCountryRows(countryAggregates)
|
||||
for _, row := range countryRows {
|
||||
applyAslanMongoUnavailableFields(row)
|
||||
}
|
||||
return aslanRangeOverview{
|
||||
Total: aslanCombinedMetric(totalFlow, totalSnapshot),
|
||||
DailySeries: dailySeries,
|
||||
CountryBreakdown: countryRows,
|
||||
DailyCountryBreakdown: dailyCountries,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *AslanMongoDashboardSource) loadRegisteredUsersByCountry(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter) (map[string]int64, error) {
|
||||
match := bson.M{
|
||||
"originSys": s.sysOrigin,
|
||||
"del": bson.M{"$ne": true},
|
||||
"createTime": bson.M{"$gte": startDate, "$lt": endDate},
|
||||
"countryCode": bson.M{"$exists": true, "$ne": ""},
|
||||
"accountStatus": bson.M{"$ne": "DELETED"},
|
||||
}
|
||||
aslanMongoApplyCountryFilter(match, "countryCode", countryFilter)
|
||||
out, err := s.aggregateUniqueUsersByCountry(ctx, aslanMongoUserProfileCollection, match, "countryCode", "_id")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query aslan mongo registered users for %s: %w", dashboardSQLDate(startDate), err)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *AslanMongoDashboardSource) loadActiveUsersByCountry(ctx context.Context, day time.Time, countryFilter externalDashboardCountryFilter) (map[string]int64, error) {
|
||||
match := bson.M{
|
||||
"sysOrigin": s.sysOrigin,
|
||||
"activeDate": dashboardSQLDate(day),
|
||||
"registerCountryCode": bson.M{"$exists": true, "$ne": ""},
|
||||
}
|
||||
aslanMongoApplyCountryFilter(match, "registerCountryCode", countryFilter)
|
||||
out, err := s.aggregateUniqueUsersByCountry(ctx, aslanMongoUserActiveCollection, match, "registerCountryCode", "userId")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query aslan mongo active users for %s: %w", dashboardSQLDate(day), err)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *AslanMongoDashboardSource) loadRechargeByCountry(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter) (map[string]aslanMongoRechargeMetric, error) {
|
||||
match := bson.M{
|
||||
"sysOrigin": s.sysOrigin,
|
||||
"status": "SUCCESS",
|
||||
"trialPeriod": bson.M{"$ne": true},
|
||||
"createTime": bson.M{"$gte": startDate, "$lt": endDate},
|
||||
"countryCode": bson.M{"$exists": true, "$ne": ""},
|
||||
}
|
||||
aslanMongoApplyCountryFilter(match, "countryCode", countryFilter)
|
||||
pipeline := mongo.Pipeline{
|
||||
{{Key: "$match", Value: match}},
|
||||
{{Key: "$group", Value: bson.D{
|
||||
{Key: "_id", Value: "$countryCode"},
|
||||
{Key: "amount", Value: bson.D{{Key: "$sum", Value: "$amountUsd"}}},
|
||||
// Aslan 原 Top 充值统计按 acceptUserId 聚合；createUser 只在 acceptUserId 缺失的旧单据上兜底。
|
||||
{Key: "users", Value: bson.D{{Key: "$addToSet", Value: bson.D{{Key: "$ifNull", Value: bson.A{"$acceptUserId", "$createUser"}}}}}},
|
||||
}}},
|
||||
}
|
||||
cursor, err := s.db.Collection(aslanMongoInAppPurchaseCollection).Aggregate(ctx, pipeline, options.Aggregate().SetAllowDiskUse(true))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("aggregate in_app_purchase_details: %w", err)
|
||||
}
|
||||
defer cursor.Close(ctx)
|
||||
|
||||
out := map[string]aslanMongoRechargeMetric{}
|
||||
for cursor.Next(ctx) {
|
||||
row := bson.M{}
|
||||
if err := cursor.Decode(&row); err != nil {
|
||||
return nil, fmt.Errorf("decode in_app_purchase_details aggregate: %w", err)
|
||||
}
|
||||
countryCode := aslanMongoCountryCode(row["_id"])
|
||||
if countryCode == "" {
|
||||
continue
|
||||
}
|
||||
amount, err := aslanMongoDecimalMinor(row["amount"])
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parse recharge amount for country %s: %w", countryCode, err)
|
||||
}
|
||||
out[countryCode] = aslanMongoRechargeMetric{
|
||||
AmountUSDMinor: amount,
|
||||
Users: aslanMongoDistinctValueCount(row["users"]),
|
||||
}
|
||||
}
|
||||
if err := cursor.Err(); err != nil {
|
||||
return nil, fmt.Errorf("iterate in_app_purchase_details aggregate: %w", err)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *AslanMongoDashboardSource) aggregateUniqueUsersByCountry(ctx context.Context, collectionName string, match bson.M, countryField string, userField string) (map[string]int64, error) {
|
||||
pipeline := mongo.Pipeline{
|
||||
{{Key: "$match", Value: match}},
|
||||
{{Key: "$group", Value: bson.D{
|
||||
{Key: "_id", Value: "$" + countryField},
|
||||
{Key: "users", Value: bson.D{{Key: "$addToSet", Value: "$" + userField}}},
|
||||
{Key: "count", Value: bson.D{{Key: "$sum", Value: 1}}},
|
||||
}}},
|
||||
}
|
||||
cursor, err := s.db.Collection(collectionName).Aggregate(ctx, pipeline, options.Aggregate().SetAllowDiskUse(true))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer cursor.Close(ctx)
|
||||
|
||||
out := map[string]int64{}
|
||||
for cursor.Next(ctx) {
|
||||
row := bson.M{}
|
||||
if err := cursor.Decode(&row); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
countryCode := aslanMongoCountryCode(row["_id"])
|
||||
if countryCode == "" {
|
||||
continue
|
||||
}
|
||||
count := aslanMongoDistinctValueCount(row["users"])
|
||||
if count == 0 {
|
||||
count = aslanMongoWholeInt64(row["count"])
|
||||
}
|
||||
out[countryCode] = count
|
||||
}
|
||||
if err := cursor.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *AslanMongoDashboardSource) loadHostSalaryBalanceByCountry(ctx context.Context, countryFilter externalDashboardCountryFilter) (map[string]int64, error) {
|
||||
memberIDs, err := s.loadTeamMemberIDs(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(memberIDs) == 0 {
|
||||
return map[string]int64{}, nil
|
||||
}
|
||||
countryByUser, err := s.loadUserCountries(ctx, memberIDs, countryFilter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(countryByUser) == 0 {
|
||||
return map[string]int64{}, nil
|
||||
}
|
||||
|
||||
out := map[string]int64{}
|
||||
for _, chunk := range aslanMongoInt64Chunks(memberIDs, aslanMongoChunkSize) {
|
||||
cursor, err := s.db.Collection(aslanMongoBankBalanceCollection).Find(ctx, bson.M{
|
||||
"sysOrigin": s.sysOrigin,
|
||||
"_id": bson.M{"$in": chunk},
|
||||
}, options.Find().SetProjection(bson.M{"_id": 1, "balance": 1}))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query user_bank_balance: %w", err)
|
||||
}
|
||||
for cursor.Next(ctx) {
|
||||
row := bson.M{}
|
||||
if err := cursor.Decode(&row); err != nil {
|
||||
_ = cursor.Close(ctx)
|
||||
return nil, fmt.Errorf("decode user_bank_balance: %w", err)
|
||||
}
|
||||
userID := aslanMongoWholeInt64(row["_id"])
|
||||
countryCode := countryByUser[userID]
|
||||
if countryCode == "" {
|
||||
continue
|
||||
}
|
||||
// user_bank_balance.balance 是 PennyAmount 的 long 值；admin 响应也用 USD minor，所以不用再做美元小数换算。
|
||||
out[countryCode] += aslanMongoWholeInt64(row["balance"])
|
||||
}
|
||||
if err := cursor.Err(); err != nil {
|
||||
_ = cursor.Close(ctx)
|
||||
return nil, fmt.Errorf("iterate user_bank_balance: %w", err)
|
||||
}
|
||||
_ = cursor.Close(ctx)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *AslanMongoDashboardSource) loadTeamMemberIDs(ctx context.Context) ([]int64, error) {
|
||||
cursor, err := s.db.Collection(aslanMongoTeamMemberCollection).Find(ctx, bson.M{
|
||||
"sysOrigin": s.sysOrigin,
|
||||
"role": bson.M{"$in": bson.A{"MEMBER", "ADMIN"}},
|
||||
}, options.Find().SetProjection(bson.M{"memberId": 1}))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query team_member: %w", err)
|
||||
}
|
||||
defer cursor.Close(ctx)
|
||||
|
||||
seen := map[int64]struct{}{}
|
||||
out := []int64{}
|
||||
for cursor.Next(ctx) {
|
||||
row := bson.M{}
|
||||
if err := cursor.Decode(&row); err != nil {
|
||||
return nil, fmt.Errorf("decode team_member: %w", err)
|
||||
}
|
||||
userID := aslanMongoWholeInt64(row["memberId"])
|
||||
if userID <= 0 {
|
||||
continue
|
||||
}
|
||||
if _, ok := seen[userID]; ok {
|
||||
continue
|
||||
}
|
||||
seen[userID] = struct{}{}
|
||||
out = append(out, userID)
|
||||
}
|
||||
if err := cursor.Err(); err != nil {
|
||||
return nil, fmt.Errorf("iterate team_member: %w", err)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *AslanMongoDashboardSource) loadUserCountries(ctx context.Context, userIDs []int64, countryFilter externalDashboardCountryFilter) (map[int64]string, error) {
|
||||
out := map[int64]string{}
|
||||
for _, chunk := range aslanMongoInt64Chunks(userIDs, aslanMongoChunkSize) {
|
||||
filter := bson.M{
|
||||
"_id": bson.M{"$in": chunk},
|
||||
"originSys": s.sysOrigin,
|
||||
"del": bson.M{"$ne": true},
|
||||
"countryCode": bson.M{"$exists": true, "$ne": ""},
|
||||
}
|
||||
aslanMongoApplyCountryFilter(filter, "countryCode", countryFilter)
|
||||
cursor, err := s.db.Collection(aslanMongoUserProfileCollection).Find(ctx, filter, options.Find().SetProjection(bson.M{"_id": 1, "countryCode": 1}))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query user_run_profile countries: %w", err)
|
||||
}
|
||||
for cursor.Next(ctx) {
|
||||
row := bson.M{}
|
||||
if err := cursor.Decode(&row); err != nil {
|
||||
_ = cursor.Close(ctx)
|
||||
return nil, fmt.Errorf("decode user_run_profile country: %w", err)
|
||||
}
|
||||
userID := aslanMongoWholeInt64(row["_id"])
|
||||
countryCode := aslanMongoCountryCode(row["countryCode"])
|
||||
if userID > 0 && countryCode != "" {
|
||||
out[userID] = countryCode
|
||||
}
|
||||
}
|
||||
if err := cursor.Err(); err != nil {
|
||||
_ = cursor.Close(ctx)
|
||||
return nil, fmt.Errorf("iterate user_run_profile countries: %w", err)
|
||||
}
|
||||
_ = cursor.Close(ctx)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func aslanMongoApplyCountryFilter(match bson.M, field string, countryFilter externalDashboardCountryFilter) {
|
||||
if !countryFilter.Applied {
|
||||
return
|
||||
}
|
||||
if len(countryFilter.Codes) == 0 {
|
||||
match[field] = bson.M{"$in": bson.A{"__NO_COUNTRY__"}}
|
||||
return
|
||||
}
|
||||
codes := bson.A{}
|
||||
for _, code := range countryFilter.Codes {
|
||||
if normalized := aslanMongoCountryCode(code); normalized != "" {
|
||||
codes = append(codes, normalized)
|
||||
}
|
||||
}
|
||||
if len(codes) == 0 {
|
||||
codes = bson.A{"__NO_COUNTRY__"}
|
||||
}
|
||||
match[field] = bson.M{"$in": codes}
|
||||
}
|
||||
|
||||
func aslanMongoCountryKeys(registers map[string]int64, activeUsers map[string]int64, recharges map[string]aslanMongoRechargeMetric, salarySnapshot map[string]int64) []string {
|
||||
seen := map[string]struct{}{}
|
||||
for countryCode := range registers {
|
||||
seen[countryCode] = struct{}{}
|
||||
}
|
||||
for countryCode := range activeUsers {
|
||||
seen[countryCode] = struct{}{}
|
||||
}
|
||||
for countryCode := range recharges {
|
||||
seen[countryCode] = struct{}{}
|
||||
}
|
||||
for countryCode := range salarySnapshot {
|
||||
seen[countryCode] = struct{}{}
|
||||
}
|
||||
keys := make([]string, 0, len(seen))
|
||||
for countryCode := range seen {
|
||||
keys = append(keys, countryCode)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
return keys
|
||||
}
|
||||
|
||||
func aslanMongoCountryCode(value any) string {
|
||||
code := strings.ToUpper(strings.TrimSpace(fmt.Sprint(value)))
|
||||
if code == "" || code == "<NIL>" || code == "NULL" {
|
||||
return ""
|
||||
}
|
||||
return code
|
||||
}
|
||||
|
||||
func aslanMongoDistinctValueCount(value any) int64 {
|
||||
values, ok := value.(bson.A)
|
||||
if !ok {
|
||||
if raw, ok := value.([]any); ok {
|
||||
values = raw
|
||||
}
|
||||
}
|
||||
seen := map[string]struct{}{}
|
||||
for _, item := range values {
|
||||
key := strings.TrimSpace(fmt.Sprint(item))
|
||||
if key == "" || key == "<nil>" || key == "0" {
|
||||
continue
|
||||
}
|
||||
seen[key] = struct{}{}
|
||||
}
|
||||
return int64(len(seen))
|
||||
}
|
||||
|
||||
func aslanMongoDecimalMinor(value any) (int64, error) {
|
||||
switch typed := value.(type) {
|
||||
case nil:
|
||||
return 0, nil
|
||||
case bson.Decimal128:
|
||||
return decimalTextToMinor(typed.String())
|
||||
case string:
|
||||
return decimalTextToMinor(typed)
|
||||
case int32:
|
||||
return int64(typed) * 100, nil
|
||||
case int64:
|
||||
return typed * 100, nil
|
||||
case int:
|
||||
return int64(typed) * 100, nil
|
||||
case float32:
|
||||
return decimalTextToMinor(strconv.FormatFloat(float64(typed), 'f', -1, 32))
|
||||
case float64:
|
||||
return decimalTextToMinor(strconv.FormatFloat(typed, 'f', -1, 64))
|
||||
default:
|
||||
return decimalTextToMinor(fmt.Sprint(value))
|
||||
}
|
||||
}
|
||||
|
||||
func aslanMongoWholeInt64(value any) int64 {
|
||||
switch typed := value.(type) {
|
||||
case nil:
|
||||
return 0
|
||||
case int64:
|
||||
return typed
|
||||
case int32:
|
||||
return int64(typed)
|
||||
case int:
|
||||
return int64(typed)
|
||||
case float64:
|
||||
return int64(typed)
|
||||
case string:
|
||||
parsed, _ := strconv.ParseInt(strings.TrimSpace(typed), 10, 64)
|
||||
return parsed
|
||||
case bson.Decimal128:
|
||||
parsed, _ := strconv.ParseInt(strings.TrimSpace(typed.String()), 10, 64)
|
||||
return parsed
|
||||
default:
|
||||
parsed, _ := strconv.ParseInt(strings.TrimSpace(fmt.Sprint(value)), 10, 64)
|
||||
return parsed
|
||||
}
|
||||
}
|
||||
|
||||
func aslanMongoInt64Chunks(values []int64, size int) [][]int64 {
|
||||
if size <= 0 {
|
||||
size = aslanMongoChunkSize
|
||||
}
|
||||
out := [][]int64{}
|
||||
for start := 0; start < len(values); start += size {
|
||||
end := start + size
|
||||
if end > len(values) {
|
||||
end = len(values)
|
||||
}
|
||||
out = append(out, values[start:end])
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
var aslanMongoUnavailableMetricFields = []string{
|
||||
"new_user_recharge_usd_minor",
|
||||
"coin_seller_recharge_usd_minor",
|
||||
"coin_seller_stock_coin",
|
||||
"coin_seller_transfer_coin",
|
||||
"google_recharge_usd_minor",
|
||||
"mifapay_recharge_usd_minor",
|
||||
"game_turnover",
|
||||
"game_payout",
|
||||
"game_players",
|
||||
"game_profit",
|
||||
"game_profit_rate",
|
||||
"game_payout_delta_rate",
|
||||
"lucky_gift_turnover",
|
||||
"lucky_gift_payout",
|
||||
"lucky_gift_payout_rate",
|
||||
"lucky_gift_payers",
|
||||
"lucky_gift_anchor_share",
|
||||
"lucky_gift_profit",
|
||||
"lucky_gift_profit_rate",
|
||||
"lucky_gift_payout_delta_rate",
|
||||
"super_lucky_gift_turnover",
|
||||
"super_lucky_gift_payout",
|
||||
"super_lucky_gift_profit",
|
||||
"super_lucky_gift_profit_rate",
|
||||
"room_lucky_gift_turnover",
|
||||
"gift_coin_spent",
|
||||
"coin_total",
|
||||
"coin_total_delta_rate",
|
||||
"consumed_coin",
|
||||
"output_coin",
|
||||
"consume_output_ratio",
|
||||
"platform_grant_coin",
|
||||
"manual_grant_coin",
|
||||
"new_dealer_recharge_usd_minor",
|
||||
"salary_exchange_usd_minor",
|
||||
"salary_transfer_coin",
|
||||
"salary_transfer_usd_minor",
|
||||
"mic_online_ms",
|
||||
"avg_mic_online_ms",
|
||||
"gift_coin_spent_delta_rate",
|
||||
"game_turnover_delta_rate",
|
||||
"game_profit_delta_rate",
|
||||
"lucky_gift_turnover_delta_rate",
|
||||
"lucky_gift_profit_delta_rate",
|
||||
"d1_retention_rate",
|
||||
"d7_retention_rate",
|
||||
"d30_retention_rate",
|
||||
}
|
||||
|
||||
func applyAslanMongoUnavailableFields(row map[string]any) {
|
||||
for _, field := range aslanMongoUnavailableMetricFields {
|
||||
row[field] = nil
|
||||
}
|
||||
}
|
||||
|
||||
func aslanMongoDashboardMetricSources() []map[string]any {
|
||||
return []map[string]any{
|
||||
{"field": "new_users", "available": true, "source": "Aslan Mongo user_run_profile createTime/countryCode"},
|
||||
{"field": "active_users", "available": true, "source": "Aslan Mongo user_daily_active_log activeDate/registerCountryCode"},
|
||||
{"field": "paid_users", "available": true, "source": "Aslan Mongo in_app_purchase_details SUCCESS distinct acceptUserId"},
|
||||
{"field": "recharge_users", "available": true, "source": "Aslan Mongo in_app_purchase_details SUCCESS distinct acceptUserId"},
|
||||
{"field": "recharge_usd_minor", "available": true, "source": "Aslan Mongo in_app_purchase_details amountUsd"},
|
||||
{"field": "new_user_recharge_usd_minor", "available": false, "missing_reason": "Mongo 中未确认新用户充值 cohort 聚合口径"},
|
||||
{"field": "coin_seller_recharge_usd_minor", "available": false, "missing_reason": "Mongo 中未确认币商充值金额聚合口径"},
|
||||
{"field": "coin_seller_stock_coin", "available": false, "missing_reason": "Mongo 中未确认币商充值金币聚合口径"},
|
||||
{"field": "coin_seller_transfer_coin", "available": false, "missing_reason": "Mongo 中未确认币商出货金币聚合口径"},
|
||||
{"field": "google_recharge_usd_minor", "available": false, "missing_reason": "in_app_purchase_details 当前接入只做总充值，未拆 Google/三方渠道"},
|
||||
{"field": "mifapay_recharge_usd_minor", "available": false, "missing_reason": "in_app_purchase_details 当前接入只做总充值，未拆 Google/三方渠道"},
|
||||
{"field": "game_turnover", "available": false, "missing_reason": "Mongo 中未确认游戏流水和返奖集合"},
|
||||
{"field": "lucky_gift_turnover", "available": false, "missing_reason": "Mongo 中未确认幸运礼物流水和返奖集合"},
|
||||
{"field": "super_lucky_gift_turnover", "available": false, "missing_reason": "Mongo 中未确认超级幸运礼物流水和返奖集合"},
|
||||
{"field": "gift_coin_spent", "available": false, "missing_reason": "Mongo 中未确认礼物流水总口径"},
|
||||
{"field": "coin_total", "available": false, "missing_reason": "Mongo 中未发现可按国家低成本汇总的当前金币余额快照"},
|
||||
{"field": "consumed_coin", "available": false, "missing_reason": "Mongo 中未确认消耗金币总口径"},
|
||||
{"field": "output_coin", "available": false, "missing_reason": "Mongo 中未确认产出金币总口径"},
|
||||
{"field": "consume_output_ratio", "available": false, "missing_reason": "缺少消耗金币和产出金币总口径"},
|
||||
{"field": "platform_grant_coin", "available": false, "missing_reason": "Mongo 中未确认平台发放金币来源枚举"},
|
||||
{"field": "manual_grant_coin", "available": false, "missing_reason": "Mongo 中未确认人工发放金币来源枚举"},
|
||||
{"field": "salary_usd_minor", "available": true, "source": "Aslan Mongo team_member + user_bank_balance + user_run_profile current snapshot"},
|
||||
{"field": "salary_transfer_coin", "available": false, "missing_reason": "Mongo 中未确认工资兑换金币流水口径"},
|
||||
{"field": "avg_mic_online_ms", "available": false, "missing_reason": "Mongo 中未确认麦上时长集合"},
|
||||
{"field": "mic_online_ms", "available": false, "missing_reason": "Mongo 中未确认麦上时长集合"},
|
||||
{"field": "arpu_usd_minor", "available": true, "source": "recharge_usd_minor / active_users"},
|
||||
{"field": "arppu_usd_minor", "available": true, "source": "recharge_usd_minor / paid_users"},
|
||||
{"field": "paid_conversion_rate", "available": true, "source": "paid_users / active_users"},
|
||||
{"field": "recharge_conversion_rate", "available": true, "source": "recharge_users / active_users"},
|
||||
{"field": "retention.day1_rate", "available": false, "missing_reason": "Mongo 中未确认留存 cohort 活跃事实"},
|
||||
{"field": "retention.day7_rate", "available": false, "missing_reason": "Mongo 中未确认留存 cohort 活跃事实"},
|
||||
{"field": "retention.day30_rate", "available": false, "missing_reason": "Mongo 中未确认留存 cohort 活跃事实"},
|
||||
}
|
||||
}
|
||||
@ -1959,6 +1959,17 @@ func TestTransferCoinFromSellerMovesDedicatedAssetToPlayerCoin(t *testing.T) {
|
||||
if got := repository.CountRows("wallet_recharge_records", "transaction_id = ?", first.TransactionID); got != 1 {
|
||||
t.Fatalf("seller transfer should write one recharge record, got %d", got)
|
||||
}
|
||||
bills, total, err := svc.ListRechargeBills(context.Background(), ledger.ListRechargeBillsQuery{
|
||||
AppCode: appcode.Default,
|
||||
Page: 1,
|
||||
PageSize: 10,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("ListRechargeBills after seller transfer failed: %v", err)
|
||||
}
|
||||
if total != 0 || len(bills) != 0 {
|
||||
t.Fatalf("finance recharge bills must hide coin seller user transfers: total=%d bills=%+v", total, bills)
|
||||
}
|
||||
if got := repository.CountRows("wallet_outbox", "transaction_id = ?", first.TransactionID); got != 4 {
|
||||
t.Fatalf("seller transfer should write two balance events, one transfer event and one recharge event, got %d", got)
|
||||
}
|
||||
|
||||
@ -102,8 +102,10 @@ func normalizeRechargeBillsQuery(query ledger.ListRechargeBillsQuery) ledger.Lis
|
||||
}
|
||||
|
||||
func rechargeBillsWhereSQL(query ledger.ListRechargeBillsQuery) (string, []any) {
|
||||
where := `WHERE rr.app_code = ?`
|
||||
args := []any{query.AppCode}
|
||||
// 财务系统的 APP 充值详情只展示用户主动向平台充值的账单；币商给用户转普通金币虽然会落充值事实供活动/统计消费，
|
||||
// 但它本质是币商出货，不应混入财务充值明细和分页总数。
|
||||
where := `WHERE rr.app_code = ? AND wt.biz_type <> ?`
|
||||
args := []any{query.AppCode, bizTypeCoinSellerTransfer}
|
||||
if query.UserID > 0 {
|
||||
where += ` AND rr.user_id = ?`
|
||||
args = append(args, query.UserID)
|
||||
|
||||
@ -16,7 +16,17 @@ func TestRechargeBillsWhereUsesExclusiveEndMs(t *testing.T) {
|
||||
if !strings.Contains(where, "rr.created_at_ms >= ?") || !strings.Contains(where, "rr.created_at_ms < ?") {
|
||||
t.Fatalf("recharge bill range must use [start_ms, end_ms): where=%s", where)
|
||||
}
|
||||
if len(args) != 3 || args[1] != int64(1778284800000) || args[2] != int64(1778371200000) {
|
||||
if len(args) != 4 || args[2] != int64(1778284800000) || args[3] != int64(1778371200000) {
|
||||
t.Fatalf("range args mismatch: %#v", args)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRechargeBillsWhereExcludesCoinSellerTransfers(t *testing.T) {
|
||||
where, args := rechargeBillsWhereSQL(ledger.ListRechargeBillsQuery{AppCode: "lalu"})
|
||||
if !strings.Contains(where, "wt.biz_type <> ?") {
|
||||
t.Fatalf("finance recharge bills must exclude coin seller user transfers: where=%s", where)
|
||||
}
|
||||
if len(args) < 2 || args[1] != bizTypeCoinSellerTransfer {
|
||||
t.Fatalf("coin seller transfer exclusion arg mismatch: %#v", args)
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user