From c6ec9fe6aa2180dda0ef587cdc9b65c88dc3a028 Mon Sep 17 00:00:00 2001 From: zhx Date: Thu, 2 Jul 2026 19:23:36 +0800 Subject: [PATCH] =?UTF-8?q?=E9=98=BF=E6=96=AF=E5=85=B0=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/admin/cmd/server/main.go | 37 + .../admin/configs/config.tencent.example.yaml | 15 +- server/admin/configs/config.yaml | 15 +- server/admin/internal/config/config.go | 87 ++- server/admin/internal/config/config_test.go | 37 + .../modules/dashboard/aslan_dashboard.go | 3 + .../dashboard/aslan_mongo_dashboard.go | 690 ++++++++++++++++++ .../internal/service/wallet/service_test.go | 11 + .../storage/mysql/recharge_repository.go | 6 +- .../storage/mysql/recharge_repository_test.go | 12 +- 10 files changed, 898 insertions(+), 15 deletions(-) create mode 100644 server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go diff --git a/server/admin/cmd/server/main.go b/server/admin/cmd/server/main.go index e1e21957..f411884d 100644 --- a/server/admin/cmd/server/main.go +++ b/server/admin/cmd/server/main.go @@ -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() diff --git a/server/admin/configs/config.tencent.example.yaml b/server/admin/configs/config.tencent.example.yaml index b8c469a1..ed9543a2 100644 --- a/server/admin/configs/config.tencent.example.yaml +++ b/server/admin/configs/config.tencent.example.yaml @@ -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: diff --git a/server/admin/configs/config.yaml b/server/admin/configs/config.yaml index 7b013ea2..1af0d037 100644 --- a/server/admin/configs/config.yaml +++ b/server/admin/configs/config.yaml @@ -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: diff --git a/server/admin/internal/config/config.go b/server/admin/internal/config/config.go index 2da391b0..98230180 100644 --- a/server/admin/internal/config/config.go +++ b/server/admin/internal/config/config.go @@ -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) != "" { diff --git a/server/admin/internal/config/config_test.go b/server/admin/internal/config/config_test.go index 2f78b19b..d7581dca 100644 --- a/server/admin/internal/config/config_test.go +++ b/server/admin/internal/config/config_test.go @@ -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) + } +} diff --git a/server/admin/internal/modules/dashboard/aslan_dashboard.go b/server/admin/internal/modules/dashboard/aslan_dashboard.go index de66bb52..e8ea2179 100644 --- a/server/admin/internal/modules/dashboard/aslan_dashboard.go +++ b/server/admin/internal/modules/dashboard/aslan_dashboard.go @@ -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) { diff --git a/server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go b/server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go new file mode 100644 index 00000000..8e3b795b --- /dev/null +++ b/server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go @@ -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 == "" || 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 == "" || 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 活跃事实"}, + } +} diff --git a/services/wallet-service/internal/service/wallet/service_test.go b/services/wallet-service/internal/service/wallet/service_test.go index e341a6c4..69580107 100644 --- a/services/wallet-service/internal/service/wallet/service_test.go +++ b/services/wallet-service/internal/service/wallet/service_test.go @@ -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) } diff --git a/services/wallet-service/internal/storage/mysql/recharge_repository.go b/services/wallet-service/internal/storage/mysql/recharge_repository.go index c389f5ca..dc561ed6 100644 --- a/services/wallet-service/internal/storage/mysql/recharge_repository.go +++ b/services/wallet-service/internal/storage/mysql/recharge_repository.go @@ -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) diff --git a/services/wallet-service/internal/storage/mysql/recharge_repository_test.go b/services/wallet-service/internal/storage/mysql/recharge_repository_test.go index ece99c18..9dd3369a 100644 --- a/services/wallet-service/internal/storage/mysql/recharge_repository_test.go +++ b/services/wallet-service/internal/storage/mysql/recharge_repository_test.go @@ -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) + } +}