From f839e44375e3617eaa7d422b3c065559365031e6 Mon Sep 17 00:00:00 2001 From: zhx Date: Sun, 5 Jul 2026 23:47:00 +0800 Subject: [PATCH] =?UTF-8?q?aslan=E5=B8=81=E5=95=86=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 4 +- server/admin/cmd/server/main.go | 31 +- .../admin/configs/config.tencent.example.yaml | 3 + server/admin/configs/config.yaml | 3 + server/admin/internal/config/config.go | 80 ++- server/admin/internal/config/config_test.go | 124 ++++ .../dashboard/aslan_mongo_dashboard.go | 541 +++++++++++++++--- .../dashboard/aslan_mongo_dashboard_test.go | 263 +++++++++ 8 files changed, 950 insertions(+), 99 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index a9d2ae6e..961c42c8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -96,10 +96,10 @@ - 默认使用 Go 标准库和仓库已有小包；不要为了轻量逻辑引入重框架。 - 手写代码保持高密度有效注释，所有业务逻辑、状态流转、权限判断、支付链路、异步流程都要解释工程语义和失败分支，不复述语法。 - 新增配置必须同时考虑本地 `config.yaml`、Docker `config.docker.yaml`、以及线上配置示例。 -- 所有逻辑功能配置默认开启；密钥、地址、三方账号等必须由运行环境提供的敏感配置只能使用占位符或环境配置，不能写入真实值，也不能伪造可用凭证。 +- 所有逻辑功能配置默认开启；密钥、地址、三方账号等敏感值只能由运行环境提供。可提交 YAML/示例配置只能写占位符；真实值只能写入已被 git 忽略的本地运行时 env 文件（例如 `.env.prod`）或部署平台环境变量，不能写入可提交文件。 - 新增服务依赖必须在 `docker-compose.yml` 中体现本地可运行形态。 - 数据库表结构变更必须同步更新对应 owner service 的 `deploy/mysql/initdb`。 -- 不要把线上密钥、云数据库密码、Redis 密码写进真实配置。只能写占位示例。 +- 不要把线上密钥、云数据库密码、Redis 密码写进 YAML、示例配置、代码、测试或任何会进入 git 的文件；需要落地到文件时，只能写入 `.env.*` 这类已忽略的运行时 env 文件。 - 保持 `gofmt`，不要手调 Go import 顺序。 ## Test Expectations diff --git a/server/admin/cmd/server/main.go b/server/admin/cmd/server/main.go index 345d1f14..57f9b3ae 100644 --- a/server/admin/cmd/server/main.go +++ b/server/admin/cmd/server/main.go @@ -5,6 +5,7 @@ import ( "database/sql" "errors" "flag" + "fmt" "log" "log/slog" "net/http" @@ -613,16 +614,44 @@ func connectDashboardExternalSources(ctx context.Context, configs []config.Dashb return nil, nil, err } cancel() - source, err := dashboardmodule.NewAslanMongoDashboardSource(client.Database(sourceConfig.MongoDatabase), sourceConfig, regionResolvers...) + var legacyDB *sql.DB + if strings.TrimSpace(sourceConfig.LegacyMySQLDSN) != "" { + // 金币代理发货充值金额只存在 legacy RDS；它是 Aslan Mongo 大屏的补充事实源，失败应在启动期暴露。 + legacyDB, err = sql.Open("mysql", sourceConfig.LegacyMySQLDSN) + if err != nil { + _ = client.Disconnect(ctx) + cleanup() + return nil, nil, fmt.Errorf("open dashboard legacy mysql for app %s: %w", sourceConfig.AppCode, err) + } + legacyCtx, legacyCancel := context.WithTimeout(ctx, sourceConfig.RequestTimeout) + if err := legacyDB.PingContext(legacyCtx); err != nil { + legacyCancel() + _ = legacyDB.Close() + _ = client.Disconnect(ctx) + cleanup() + return nil, nil, fmt.Errorf("ping dashboard legacy mysql for app %s: %w", sourceConfig.AppCode, err) + } + legacyCancel() + } + source, err := dashboardmodule.NewAslanMongoDashboardSourceWithLegacyDB(client.Database(sourceConfig.MongoDatabase), legacyDB, sourceConfig, regionResolvers...) if err != nil { + if legacyDB != nil { + _ = legacyDB.Close() + } _ = client.Disconnect(ctx) cleanup() return nil, nil, err } if source == nil { + if legacyDB != nil { + _ = legacyDB.Close() + } _ = client.Disconnect(ctx) continue } + if legacyDB != nil { + dbs = append(dbs, legacyDB) + } mongoClients = append(mongoClients, client) sources = append(sources, source) continue diff --git a/server/admin/configs/config.tencent.example.yaml b/server/admin/configs/config.tencent.example.yaml index 0c4f513d..3cd4ab32 100644 --- a/server/admin/configs/config.tencent.example.yaml +++ b/server/admin/configs/config.tencent.example.yaml @@ -116,6 +116,9 @@ dashboard_external_sources: sys_origin: "ATYOU" mongo_uri: "mongodb://aslan_data:aslan_data@lb-le274w4o-wjgmcdkhye3v0ors.clb.sg-tencentclb.com:27017,lb-le274w4o-wjgmcdkhye3v0ors.clb.sg-tencentclb.com:27018/test?replicaSet=cmgo-6cztxjgr_0&authSource=admin&readPreference=secondaryPreferred" mongo_database: "atyou" + # 金币代理发货充值金额来自 legacy RDS；真实密码通过 HYAPP_ADMIN_ASLAN_DASHBOARD_LEGACY_MYSQL_DSN 注入覆盖。 + legacy_mysql_dsn: "aslan:REPLACE_ME@tcp(sg-cdb-pa8hgyh1.sql.tencentcdb.com:29850)/atyou?parseTime=true&charset=utf8mb4&loc=Asia%2FShanghai&timeout=10s&readTimeout=30s&writeTimeout=30s" + legacy_wallet_database: "atyou_wallet" stat_timezone: "Asia/Shanghai" # aslan_mongo 源按整段区间做聚合（注册/日活/充值/礼物 + 留存 cohort），30 天区间比逐日查询重， # 5s 会在长区间超时；线上建议 30s 起。 diff --git a/server/admin/configs/config.yaml b/server/admin/configs/config.yaml index 0bdcc01b..d5beab1c 100644 --- a/server/admin/configs/config.yaml +++ b/server/admin/configs/config.yaml @@ -115,6 +115,9 @@ dashboard_external_sources: sys_origin: "ATYOU" mongo_uri: "" mongo_database: "test" + # 金币代理发货充值金额来自 legacy RDS；真实密码通过环境配置注入，不能写入仓库 YAML。 + legacy_mysql_dsn: "aslan:REPLACE_ME@tcp(sg-cdb-pa8hgyh1.sql.tencentcdb.com:29850)/atyou?parseTime=true&charset=utf8mb4&loc=Asia%2FShanghai&timeout=10s&readTimeout=30s&writeTimeout=30s" + legacy_wallet_database: "atyou_wallet" 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 31f44015..f5c94c2e 100644 --- a/server/admin/internal/config/config.go +++ b/server/admin/internal/config/config.go @@ -90,18 +90,20 @@ type StatisticsServiceConfig struct { } type DashboardExternalSourceConfig struct { - Enabled bool `yaml:"enabled"` - SourceType string `yaml:"type"` - AppCode string `yaml:"app_code"` - 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"` - RequestTimeout time.Duration `yaml:"request_timeout"` + Enabled bool `yaml:"enabled"` + SourceType string `yaml:"type"` + AppCode string `yaml:"app_code"` + AppName string `yaml:"app_name"` + SysOrigin string `yaml:"sys_origin"` + MySQLDSN string `yaml:"mysql_dsn"` + LegacyMySQLDSN string `yaml:"legacy_mysql_dsn"` + LegacyWalletDatabase string `yaml:"legacy_wallet_database"` + 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"` + RequestTimeout time.Duration `yaml:"request_timeout"` } type FinanceNotificationsConfig struct { @@ -454,6 +456,8 @@ 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.LegacyMySQLDSN = strings.TrimSpace(source.LegacyMySQLDSN) + source.LegacyWalletDatabase = strings.Trim(strings.TrimSpace(source.LegacyWalletDatabase), "/") source.MongoURI = strings.TrimSpace(source.MongoURI) source.MongoDatabase = strings.Trim(strings.TrimSpace(source.MongoDatabase), "/") source.BaseURL = strings.TrimRight(strings.TrimSpace(source.BaseURL), "/") @@ -473,6 +477,10 @@ func (cfg *Config) Normalize() { if source.SourceType == "aslan_mongo" && source.MongoDatabase == "" { source.MongoDatabase = "test" } + if source.SourceType == "aslan_mongo" && source.LegacyWalletDatabase == "" && source.MongoDatabase != "" { + // Aslan legacy RDS 把用户资料放在 atyou，把金币代理钱包流水放在 atyou_wallet；默认按主库名派生。 + source.LegacyWalletDatabase = source.MongoDatabase + "_wallet" + } source.StatTimezone = strings.TrimSpace(source.StatTimezone) if source.StatTimezone == "" { if source.SourceType == "aslan_http" || source.SourceType == "aslan_mongo" { @@ -691,6 +699,18 @@ func (cfg Config) Validate() error { if strings.TrimSpace(source.MongoDatabase) == "" { return fmt.Errorf("dashboard_external_sources[%d].mongo_database is required when aslan_mongo source is enabled", index) } + if strings.TrimSpace(source.LegacyMySQLDSN) == "" { + return fmt.Errorf("dashboard_external_sources[%d].legacy_mysql_dsn is required when aslan_mongo source is enabled", index) + } + if containsConfigPlaceholder(source.LegacyMySQLDSN) { + return fmt.Errorf("dashboard_external_sources[%d].legacy_mysql_dsn must be injected by environment, not left as a placeholder", index) + } + if strings.TrimSpace(source.LegacyWalletDatabase) == "" { + return fmt.Errorf("dashboard_external_sources[%d].legacy_wallet_database is required when aslan_mongo source is enabled", index) + } + if !isMySQLIdentifier(source.LegacyWalletDatabase) { + return fmt.Errorf("dashboard_external_sources[%d].legacy_wallet_database must be a mysql identifier", index) + } default: return fmt.Errorf("dashboard_external_sources[%d].type must be one of mysql, aslan_http, aslan_mongo", index) } @@ -827,7 +847,25 @@ func (cfg *Config) applyDashboardExternalSourceEnvOverrides() { "HYAPP_ADMIN_"+appKey+"_DASHBOARD_MONGO_DATABASE", "HYAPP_ADMIN_"+appKey+"_MONGO_DATABASE", )); value != "" { + previousMongoDatabase := source.MongoDatabase + usesDerivedWalletDatabase := source.LegacyWalletDatabase == "" || source.LegacyWalletDatabase == previousMongoDatabase+"_wallet" source.MongoDatabase = strings.Trim(strings.TrimSpace(value), "/") + if source.SourceType == "aslan_mongo" && usesDerivedWalletDatabase && source.MongoDatabase != "" { + source.LegacyWalletDatabase = source.MongoDatabase + "_wallet" + } + } + if value := strings.TrimSpace(firstEnv( + "HYAPP_ADMIN_"+appKey+"_DASHBOARD_LEGACY_MYSQL_DSN", + "HYAPP_ADMIN_"+appKey+"_LEGACY_MYSQL_DSN", + )); value != "" { + // Aslan Mongo 大屏只把 legacy MySQL 作为补充事实源，真实 likei RDS 凭证由运行环境注入。 + source.LegacyMySQLDSN = value + } + if value := strings.TrimSpace(firstEnv( + "HYAPP_ADMIN_"+appKey+"_DASHBOARD_LEGACY_WALLET_DATABASE", + "HYAPP_ADMIN_"+appKey+"_LEGACY_WALLET_DATABASE", + )); value != "" { + source.LegacyWalletDatabase = strings.Trim(value, "/") } } } @@ -924,6 +962,24 @@ func compactStrings(values []string) []string { return out } +func containsConfigPlaceholder(value string) bool { + upper := strings.ToUpper(strings.TrimSpace(value)) + return strings.Contains(upper, "REPLACE_ME") || strings.Contains(upper, "CHANGEME") || strings.Contains(upper, "<") +} + +func isMySQLIdentifier(value string) bool { + if value == "" { + return false + } + for _, r := range value { + if r == '_' || r == '$' || (r >= '0' && r <= '9') || (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') { + continue + } + return false + } + return true +} + func validEnvironment(value string) bool { switch value { case "local", "dev", "staging", "prod": diff --git a/server/admin/internal/config/config_test.go b/server/admin/internal/config/config_test.go index d7581dca..8a58963c 100644 --- a/server/admin/internal/config/config_test.go +++ b/server/admin/internal/config/config_test.go @@ -2,6 +2,7 @@ package config import ( "reflect" + "strings" "testing" "time" ) @@ -75,6 +76,8 @@ func TestFinanceDingTalkEnvOverride(t *testing.T) { 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_DASHBOARD_LEGACY_MYSQL_DSN", "likei:secret@tcp(example:3306)/likei?parseTime=true&loc=Asia%2FShanghai") + t.Setenv("HYAPP_ADMIN_ASLAN_DASHBOARD_LEGACY_WALLET_DATABASE", "test_wallet") t.Setenv("HYAPP_ADMIN_ASLAN_MONEY_REGION_MONGO_DATABASE", "tarab_region") cfg := Default() @@ -93,6 +96,12 @@ func TestAslanMongoEnvOverride(t *testing.T) { if dashboardSource.MongoDatabase != "test_dashboard" { t.Fatalf("dashboard mongo database mismatch: %q", dashboardSource.MongoDatabase) } + if dashboardSource.LegacyMySQLDSN != "likei:secret@tcp(example:3306)/likei?parseTime=true&loc=Asia%2FShanghai" { + t.Fatalf("dashboard legacy mysql dsn mismatch: %q", dashboardSource.LegacyMySQLDSN) + } + if dashboardSource.LegacyWalletDatabase != "test_wallet" { + t.Fatalf("dashboard legacy wallet database mismatch: %q", dashboardSource.LegacyWalletDatabase) + } var regionSource MoneyRegionSourceConfig for _, source := range cfg.MoneyRegionSources { @@ -108,3 +117,118 @@ func TestAslanMongoEnvOverride(t *testing.T) { t.Fatalf("region mongo database mismatch: %q", regionSource.MongoDatabase) } } + +func TestAslanMongoLegacyMySQLFallbackEnvOverride(t *testing.T) { + t.Setenv("HYAPP_ADMIN_ASLAN_LEGACY_MYSQL_DSN", "likei:secret@tcp(example:3306)/likei?parseTime=true&loc=Asia%2FShanghai") + t.Setenv("HYAPP_ADMIN_ASLAN_LEGACY_WALLET_DATABASE", "atyou_wallet") + + cfg := Default() + cfg.Normalize() + + for _, source := range cfg.DashboardExternalSources { + if source.AppCode != "aslan" { + continue + } + if source.LegacyMySQLDSN != "likei:secret@tcp(example:3306)/likei?parseTime=true&loc=Asia%2FShanghai" { + t.Fatalf("dashboard legacy mysql fallback dsn mismatch: %q", source.LegacyMySQLDSN) + } + if source.LegacyWalletDatabase != "atyou_wallet" { + t.Fatalf("dashboard legacy wallet database fallback mismatch: %q", source.LegacyWalletDatabase) + } + return + } + t.Fatal("aslan dashboard source not found") +} + +func TestTencentExampleRequiresRuntimeLegacyMySQLDSN(t *testing.T) { + t.Setenv("HYAPP_ADMIN_ASLAN_DASHBOARD_LEGACY_MYSQL_DSN", "") + t.Setenv("HYAPP_ADMIN_ASLAN_LEGACY_MYSQL_DSN", "") + + _, err := Load("../../configs/config.tencent.example.yaml") + if err == nil || !strings.Contains(err.Error(), "legacy_mysql_dsn must be injected by environment") { + t.Fatalf("Load() error = %v, want legacy mysql placeholder rejection", err) + } + + t.Setenv("HYAPP_ADMIN_ASLAN_DASHBOARD_LEGACY_MYSQL_DSN", "aslan:runtime-secret@tcp(example:3306)/atyou?parseTime=true&loc=Asia%2FShanghai") + cfg, err := Load("../../configs/config.tencent.example.yaml") + if err != nil { + t.Fatalf("Load() with runtime legacy dsn error = %v", err) + } + for _, source := range cfg.DashboardExternalSources { + if source.AppCode != "aslan" { + continue + } + if source.LegacyMySQLDSN != "aslan:runtime-secret@tcp(example:3306)/atyou?parseTime=true&loc=Asia%2FShanghai" { + t.Fatalf("dashboard legacy mysql dsn mismatch: %q", source.LegacyMySQLDSN) + } + if source.LegacyWalletDatabase != "atyou_wallet" { + t.Fatalf("dashboard legacy wallet database mismatch: %q", source.LegacyWalletDatabase) + } + return + } + t.Fatal("aslan dashboard source not found") +} + +func TestAslanMongoLegacyWalletDatabaseFollowsMongoDatabaseEnv(t *testing.T) { + t.Setenv("HYAPP_ADMIN_ASLAN_DASHBOARD_MONGO_DATABASE", "atyou") + + cfg := Default() + cfg.Normalize() + + for _, source := range cfg.DashboardExternalSources { + if source.AppCode != "aslan" { + continue + } + if source.MongoDatabase != "atyou" { + t.Fatalf("dashboard mongo database mismatch: %q", source.MongoDatabase) + } + if source.LegacyWalletDatabase != "atyou_wallet" { + t.Fatalf("dashboard legacy wallet database should follow mongo database, got %q", source.LegacyWalletDatabase) + } + return + } + t.Fatal("aslan dashboard source not found") +} + +func TestAslanMongoRequiresLegacyMySQLDSNWhenEnabled(t *testing.T) { + cfg := Default() + for index := range cfg.DashboardExternalSources { + source := &cfg.DashboardExternalSources[index] + if source.AppCode != "aslan" { + continue + } + source.Enabled = true + source.MongoURI = "mongodb://example:27017/test" + source.MongoDatabase = "atyou" + source.LegacyMySQLDSN = "" + break + } + cfg.Normalize() + + err := cfg.Validate() + if err == nil || !strings.Contains(err.Error(), "legacy_mysql_dsn is required") { + t.Fatalf("Validate() error = %v, want legacy mysql requirement", err) + } +} + +func TestAslanMongoRejectsPlaceholderLegacyMySQLDSN(t *testing.T) { + cfg := Default() + for index := range cfg.DashboardExternalSources { + source := &cfg.DashboardExternalSources[index] + if source.AppCode != "aslan" { + continue + } + source.Enabled = true + source.MongoURI = "mongodb://example:27017/test" + source.MongoDatabase = "atyou" + source.LegacyMySQLDSN = "aslan:REPLACE_ME@tcp(example:3306)/atyou?parseTime=true&loc=Asia%2FShanghai" + source.LegacyWalletDatabase = "atyou_wallet" + break + } + cfg.Normalize() + + err := cfg.Validate() + if err == nil || !strings.Contains(err.Error(), "not left as a placeholder") { + t.Fatalf("Validate() error = %v, want placeholder rejection", err) + } +} diff --git a/server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go b/server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go index 097736f4..2871af1c 100644 --- a/server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go +++ b/server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go @@ -2,6 +2,7 @@ package dashboard import ( "context" + "database/sql" "fmt" "sort" "strconv" @@ -32,14 +33,21 @@ const ( aslanMongoSellerAgentOriginPrefix = "SELLER_AGENT" ) +var ( + aslanMongoFreightCommodityTypes = bson.A{"FREIGHT_GOLD", "FREIGHT_GOLD_SUPER"} + aslanMongoFreightCommodityFields = []string{"trackCommodityType", "products.code", "products.name", "products.content"} +) + type AslanMongoDashboardSource struct { - db *mongo.Database - appCode string - appName string - sysOrigin string - statTimezone string - requestTimeout time.Duration - regionResolvers []ExternalDashboardRegionResolver + db *mongo.Database + legacyDB *sql.DB + legacyWalletDatabase string + appCode string + appName string + sysOrigin string + statTimezone string + requestTimeout time.Duration + regionResolvers []ExternalDashboardRegionResolver // 部分 likei 部署（如 Aslan 的 atyou 库）没有 user_gold_running_water_v2 集合； // 探测结果缓存住，避免把"集合不存在"算成 0（幸运礼物利润率会假性 100%）。 @@ -69,6 +77,16 @@ func (s *AslanMongoDashboardSource) hasGoldWater(ctx context.Context) bool { // NewAslanMongoDashboardSource 直接读取 likei 平台（Aslan/Yumi 同构）Mongo 里的已存在事实集合。 // 这里只做查询聚合和字段映射，不把数据复制到 hyapp statistics-service，也不回退旧 console HTTP 接口。 func NewAslanMongoDashboardSource(database *mongo.Database, sourceConfig config.DashboardExternalSourceConfig, regionResolvers ...ExternalDashboardRegionResolver) (ExternalDashboardSource, error) { + return newAslanMongoDashboardSource(database, nil, sourceConfig, regionResolvers...) +} + +// NewAslanMongoDashboardSourceWithLegacyDB 在 Mongo 事实之外接入 likei legacy MySQL 的补充口径。 +// legacyDB 由调用方负责生命周期；nil 表示对应指标不可用，而不是按 0 处理。 +func NewAslanMongoDashboardSourceWithLegacyDB(database *mongo.Database, legacyDB *sql.DB, sourceConfig config.DashboardExternalSourceConfig, regionResolvers ...ExternalDashboardRegionResolver) (ExternalDashboardSource, error) { + return newAslanMongoDashboardSource(database, legacyDB, sourceConfig, regionResolvers...) +} + +func newAslanMongoDashboardSource(database *mongo.Database, legacyDB *sql.DB, sourceConfig config.DashboardExternalSourceConfig, regionResolvers ...ExternalDashboardRegionResolver) (ExternalDashboardSource, error) { if database == nil || !sourceConfig.Enabled { return nil, nil } @@ -89,13 +107,15 @@ func NewAslanMongoDashboardSource(database *mongo.Database, sourceConfig config. timeout = 5 * time.Second } return &AslanMongoDashboardSource{ - db: database, - appCode: appCode, - appName: strings.TrimSpace(sourceConfig.AppName), - sysOrigin: sysOrigin, - statTimezone: statTimezone, - requestTimeout: timeout, - regionResolvers: compactExternalDashboardRegionResolvers(regionResolvers), + db: database, + legacyDB: legacyDB, + legacyWalletDatabase: strings.TrimSpace(sourceConfig.LegacyWalletDatabase), + appCode: appCode, + appName: strings.TrimSpace(sourceConfig.AppName), + sysOrigin: sysOrigin, + statTimezone: statTimezone, + requestTimeout: timeout, + regionResolvers: compactExternalDashboardRegionResolvers(regionResolvers), }, nil } @@ -162,11 +182,12 @@ func (s *AslanMongoDashboardSource) StatisticsOverview(ctx context.Context, quer response["daily_country_breakdown"] = current.DailyCountryBreakdown goldWater := s.hasGoldWater(ctx) response["retention"] = current.Total.retentionMap() - response["report_metric_sources"] = aslanMongoDashboardMetricSources(goldWater) + response["report_metric_sources"] = aslanMongoDashboardMetricSources(goldWater, s.legacyDB != nil) 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) + applyAslanMongoLegacyUnavailableFields(response, s.legacyDB != nil) if !goldWater { applyAslanMongoGoldWaterUnavailable(response) response["lucky_gift_payout_delta_rate"] = nil @@ -203,25 +224,26 @@ func (s *AslanMongoDashboardSource) countryFilter(ctx context.Context, query Sta // legacyMongoMetric 是 likei Mongo 源按 天×国家 汇总出的事实指标； // 消耗/产出金币是全 App 口径（按用户拆国家需要全量账变反查用户，代价过高），只挂在日行和合计上。 type legacyMongoMetric struct { - NewUsers int64 - ActiveUsers int64 - PaidUsers int64 - RechargeUSDMinor int64 - GoogleUSDMinor int64 - ThirdPartyUSDMinor int64 - NewUserRechargeUSDMinor int64 - GiftCoinSpent int64 - LuckyGiftTurnover int64 - LuckyGiftPayout int64 - LuckyGiftPayers int64 - CoinSellerTransferCoin int64 - SalaryUSDMinor int64 - D1Users int64 - D1Base int64 - D7Users int64 - D7Base int64 - D30Users int64 - D30Base int64 + NewUsers int64 + ActiveUsers int64 + PaidUsers int64 + RechargeUSDMinor int64 + GoogleUSDMinor int64 + ThirdPartyUSDMinor int64 + CoinSellerRechargeUSDMinor int64 + NewUserRechargeUSDMinor int64 + GiftCoinSpent int64 + LuckyGiftTurnover int64 + LuckyGiftPayout int64 + LuckyGiftPayers int64 + CoinSellerTransferCoin int64 + SalaryUSDMinor int64 + D1Users int64 + D1Base int64 + D7Users int64 + D7Base int64 + D30Users int64 + D30Base int64 } func (m *legacyMongoMetric) add(other *legacyMongoMetric) { @@ -234,6 +256,7 @@ func (m *legacyMongoMetric) add(other *legacyMongoMetric) { m.RechargeUSDMinor += other.RechargeUSDMinor m.GoogleUSDMinor += other.GoogleUSDMinor m.ThirdPartyUSDMinor += other.ThirdPartyUSDMinor + m.CoinSellerRechargeUSDMinor += other.CoinSellerRechargeUSDMinor m.NewUserRechargeUSDMinor += other.NewUserRechargeUSDMinor m.GiftCoinSpent += other.GiftCoinSpent m.LuckyGiftTurnover += other.LuckyGiftTurnover @@ -255,53 +278,57 @@ type legacyMongoCoinFlow struct { } func (m *legacyMongoMetric) toExternal(coinFlow *legacyMongoCoinFlow) externalDashboardMetric { + userRecharge := m.RechargeUSDMinor - m.CoinSellerRechargeUSDMinor metric := externalDashboardMetric{ - NewUsers: m.NewUsers, - ActiveUsers: m.ActiveUsers, - PaidUsers: m.PaidUsers, - RechargeUSDMinor: m.RechargeUSDMinor, - UserRechargeUSDMinor: m.RechargeUSDMinor, - GoogleRechargeUSDMinor: m.GoogleUSDMinor, - MifaPayRechargeUSDMinor: m.ThirdPartyUSDMinor, - NewUserRechargeUSDMinor: m.NewUserRechargeUSDMinor, - GiftCoinSpent: m.GiftCoinSpent, - LuckyGiftTurnover: m.LuckyGiftTurnover, - LuckyGiftPayout: m.LuckyGiftPayout, - LuckyGiftPayers: m.LuckyGiftPayers, - CoinSellerTransferCoin: int64Ptr(m.CoinSellerTransferCoin), - SalaryUSDMinor: int64Ptr(m.SalaryUSDMinor), - D1RetentionUsers: m.D1Users, - D1RetentionBaseUsers: m.D1Base, - D7RetentionUsers: m.D7Users, - D7RetentionBaseUsers: m.D7Base, - D30RetentionUsers: m.D30Users, - D30RetentionBaseUsers: m.D30Base, + NewUsers: m.NewUsers, + ActiveUsers: m.ActiveUsers, + PaidUsers: m.PaidUsers, + RechargeUSDMinor: m.RechargeUSDMinor, + UserRechargeUSDMinor: userRecharge, + GoogleRechargeUSDMinor: m.GoogleUSDMinor, + MifaPayRechargeUSDMinor: m.ThirdPartyUSDMinor, + CoinSellerRechargeUSDMinor: m.CoinSellerRechargeUSDMinor, + NewUserRechargeUSDMinor: m.NewUserRechargeUSDMinor, + GiftCoinSpent: m.GiftCoinSpent, + LuckyGiftTurnover: m.LuckyGiftTurnover, + LuckyGiftPayout: m.LuckyGiftPayout, + LuckyGiftPayers: m.LuckyGiftPayers, + CoinSellerTransferCoin: int64Ptr(m.CoinSellerTransferCoin), + SalaryUSDMinor: int64Ptr(m.SalaryUSDMinor), + D1RetentionUsers: m.D1Users, + D1RetentionBaseUsers: m.D1Base, + D7RetentionUsers: m.D7Users, + D7RetentionBaseUsers: m.D7Base, + D30RetentionUsers: m.D30Users, + D30RetentionBaseUsers: m.D30Base, } if coinFlow != nil { metric.ConsumedCoin = int64Ptr(coinFlow.ConsumedCoin) metric.OutputCoin = int64Ptr(coinFlow.OutputCoin) } metric.finalize() - // finalize 会用 official/google/mifapay 组合重推总充值；likei 口径 total 就是 amountUsd 求和，避免二次推导。 + // finalize 会按拆分字段重推总充值；likei Mongo 源已经把 in-app 与 legacy 币商金额合成总额，最终以事实聚合值为准。 metric.RechargeUSDMinor = m.RechargeUSDMinor - metric.UserRechargeUSDMinor = m.RechargeUSDMinor + metric.UserRechargeUSDMinor = userRecharge return metric } type legacyMongoGrid struct { - days []string - daySet map[string]struct{} - metrics map[string]map[string]*legacyMongoMetric - coinFlows map[string]*legacyMongoCoinFlow + days []string + daySet map[string]struct{} + metrics map[string]map[string]*legacyMongoMetric + coinFlows map[string]*legacyMongoCoinFlow + rechargeUsers map[string]map[string]struct{} // 各 loader 并行写入：map 结构由锁保护；不同 loader 写 metric 的不同字段，无字段级竞争。 mu sync.Mutex } func newLegacyMongoGrid(startDate time.Time, endDate time.Time) *legacyMongoGrid { grid := &legacyMongoGrid{ - daySet: map[string]struct{}{}, - metrics: map[string]map[string]*legacyMongoMetric{}, - coinFlows: map[string]*legacyMongoCoinFlow{}, + daySet: map[string]struct{}{}, + metrics: map[string]map[string]*legacyMongoMetric{}, + coinFlows: map[string]*legacyMongoCoinFlow{}, + rechargeUsers: map[string]map[string]struct{}{}, } for day := startDate; day.Before(endDate); day = day.AddDate(0, 0, 1) { text := dashboardSQLDate(day) @@ -399,10 +426,10 @@ func (s *AslanMongoDashboardSource) loadRange(ctx context.Context, startDate tim } } - return s.assembleRange(grid, countryFilter, goldWater), nil + return s.assembleRange(grid, countryFilter, goldWater, s.legacyDB != nil), nil } -func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, countryFilter externalDashboardCountryFilter, goldWater bool) aslanRangeOverview { +func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, countryFilter externalDashboardCountryFilter, goldWater bool, legacyRecharge bool) aslanRangeOverview { dailySeries := []map[string]any{} dailyCountries := []map[string]any{} totalFlow := legacyMongoMetric{} @@ -440,6 +467,7 @@ func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, country row["region_id"] = countryFilter.RegionID } applyAslanMongoUnavailableFields(row) + applyAslanMongoLegacyUnavailableFields(row, legacyRecharge) if !goldWater { applyAslanMongoGoldWaterUnavailable(row) } @@ -452,6 +480,10 @@ func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, country } // 工资余额是快照：日行取当日各国快照之和，区间合计不能跨天累加。 daySalary := dayFlow.SalaryUSDMinor + if dayRechargeUsers := grid.rechargeUsers[day]; len(dayRechargeUsers) > 0 { + // 国家拆分仍保留各国家充值人数；日合计按 user_id 去重，避免同一用户因不同事实源国家码不一致而被算两次。 + dayFlow.PaidUsers = int64(len(dayRechargeUsers)) + } totalFlow.add(&dayFlow) totalFlow.SalaryUSDMinor = daySalary @@ -459,6 +491,7 @@ func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, country dayRow["label"] = day dayRow["stat_day"] = day applyAslanMongoUnavailableFields(dayRow) + applyAslanMongoLegacyUnavailableFields(dayRow, legacyRecharge) if !goldWater { applyAslanMongoGoldWaterUnavailable(dayRow) } @@ -481,6 +514,7 @@ func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, country row["region_id"] = countryFilter.RegionID } applyAslanMongoUnavailableFields(row) + applyAslanMongoLegacyUnavailableFields(row, legacyRecharge) if !goldWater { applyAslanMongoGoldWaterUnavailable(row) } @@ -579,6 +613,11 @@ func (s *AslanMongoDashboardSource) loadActiveCounts(ctx context.Context, startD return nil } +type aslanMongoDayCountryUsers struct { + day string + country string +} + // loadRechargeStats 一次聚合出 天×国家×渠道 的充值金额与充值用户，再按当日注册 cohort 拆出新用户充值。 func (s *AslanMongoDashboardSource) loadRechargeStats(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter, cohorts map[string]map[string][]int64, grid *legacyMongoGrid) error { match := bson.M{ @@ -588,6 +627,7 @@ func (s *AslanMongoDashboardSource) loadRechargeStats(ctx context.Context, start "createTime": bson.M{"$gte": startDate, "$lt": endDate}, "countryCode": bson.M{"$exists": true, "$ne": ""}, } + aslanMongoExcludeFreightGoldRecharge(match) aslanMongoApplyCountryFilter(match, "countryCode", countryFilter) // Aslan 原 Top 充值统计按 acceptUserId 聚合；createUser 只在 acceptUserId 缺失的旧单据上兜底。 userExpression := bson.D{{Key: "$ifNull", Value: bson.A{"$acceptUserId", "$createUser"}}} @@ -603,11 +643,7 @@ func (s *AslanMongoDashboardSource) loadRechargeStats(ctx context.Context, start {Key: "users", Value: bson.D{{Key: "$addToSet", Value: userExpression}}}, }}}, } - type dayCountryUsers struct { - day string - country string - } - distinctUsers := map[dayCountryUsers]map[string]struct{}{} + distinctUsers := map[aslanMongoDayCountryUsers]map[string]struct{}{} err := s.aggregateEach(ctx, aslanMongoInAppPurchaseCollection, pipeline, func(row bson.M) error { key := aslanMongoDocument(row["_id"]) if key == nil { @@ -630,7 +666,7 @@ func (s *AslanMongoDashboardSource) loadRechargeStats(ctx context.Context, start } else { metric.ThirdPartyUSDMinor += amount } - usersKey := dayCountryUsers{day: day, country: countryCode} + usersKey := aslanMongoDayCountryUsers{day: day, country: countryCode} userSet := distinctUsers[usersKey] if userSet == nil { userSet = map[string]struct{}{} @@ -644,9 +680,6 @@ func (s *AslanMongoDashboardSource) loadRechargeStats(ctx context.Context, start if err != nil { return fmt.Errorf("aggregate recharge stats: %w", err) } - for key, userSet := range distinctUsers { - grid.at(key.day, key.country).PaidUsers = int64(len(userSet)) - } // 新用户充值：按 天×用户 聚合当日充值，再与当日注册 cohort 求交集；cohort 自带国家归属。 perUserPipeline := mongo.Pipeline{ @@ -685,6 +718,28 @@ func (s *AslanMongoDashboardSource) loadRechargeStats(ctx context.Context, start if err != nil { return fmt.Errorf("aggregate per-user recharge: %w", err) } + if err := s.loadLegacyOtherRechargeStats(ctx, startDate, endDate, countryFilter, grid, distinctUsers, perUser); err != nil { + return err + } + if err := s.loadLegacyCoinSellerRechargeStats(ctx, startDate, endDate, countryFilter, grid, distinctUsers, perUser); err != nil { + return err + } + applyAslanRechargeUserMetrics(cohorts, grid, distinctUsers, perUser) + return nil +} + +func applyAslanRechargeUserMetrics(cohorts map[string]map[string][]int64, grid *legacyMongoGrid, distinctUsers map[aslanMongoDayCountryUsers]map[string]struct{}, perUser map[string]map[int64]int64) { + for key, userSet := range distinctUsers { + grid.at(key.day, key.country).PaidUsers = int64(len(userSet)) + dayUsers := grid.rechargeUsers[key.day] + if dayUsers == nil { + dayUsers = map[string]struct{}{} + grid.rechargeUsers[key.day] = dayUsers + } + for userID := range userSet { + dayUsers[userID] = struct{}{} + } + } for day, byCountry := range cohorts { byUser := perUser[day] if len(byUser) == 0 { @@ -700,9 +755,281 @@ func (s *AslanMongoDashboardSource) loadRechargeStats(ctx context.Context, start } } } +} + +func (s *AslanMongoDashboardSource) loadLegacyOtherRechargeStats(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter, grid *legacyMongoGrid, distinctUsers map[aslanMongoDayCountryUsers]map[string]struct{}, perUser map[string]map[int64]int64) error { + if s.legacyDB == nil { + return nil + } + where := []string{ + "oor.sys_origin = ?", + "oor.payment_date_number >= ?", + "oor.payment_date_number < ?", + "ubi.country_code IS NOT NULL", + "ubi.country_code != ''", + } + args := []any{s.sysOrigin, dashboardDateNumber(startDate), dashboardDateNumber(endDate)} + if countryFilter.Applied { + if len(countryFilter.Codes) == 0 { + return nil + } + placeholders := make([]string, 0, len(countryFilter.Codes)) + for _, code := range countryFilter.Codes { + normalized := aslanMongoCountryCode(code) + if normalized == "" { + continue + } + placeholders = append(placeholders, "?") + args = append(args, normalized) + } + if len(placeholders) == 0 { + return nil + } + where = append(where, "UPPER(ubi.country_code) IN ("+strings.Join(placeholders, ",")+")") + } + + queryCtx, cancel := context.WithTimeout(ctx, s.requestTimeout) + defer cancel() + rows, err := s.legacyDB.QueryContext(queryCtx, ` +SELECT DATE_FORMAT(STR_TO_DATE(oor.payment_date_number, '%Y%m%d'), '%Y-%m-%d') AS stat_day, + UPPER(ubi.country_code) AS country_code, + oor.user_id, + CAST(COALESCE(SUM(oor.amount), 0) AS CHAR) AS amount +FROM order_other_recharge oor + INNER JOIN user_base_info ubi ON ubi.id = oor.user_id +WHERE `+strings.Join(where, " AND ")+` +GROUP BY DATE_FORMAT(STR_TO_DATE(oor.payment_date_number, '%Y%m%d'), '%Y-%m-%d'), UPPER(ubi.country_code), oor.user_id`, args...) + if err != nil { + return fmt.Errorf("query legacy other recharge: %w", err) + } + defer rows.Close() + + for rows.Next() { + var day string + var countryCode string + var userID int64 + var amountText string + if err := rows.Scan(&day, &countryCode, &userID, &amountText); err != nil { + return fmt.Errorf("scan legacy other recharge: %w", err) + } + day = strings.TrimSpace(day) + countryCode = aslanMongoCountryCode(countryCode) + if day == "" || countryCode == "" || userID <= 0 || !grid.hasDay(day) { + continue + } + amount, err := decimalTextToMinor(amountText) + if err != nil { + return fmt.Errorf("parse legacy other recharge amount for %s/%s/%d: %w", day, countryCode, userID, err) + } + addLegacyStandardRecharge(grid, distinctUsers, perUser, day, countryCode, userID, amount) + } + if err := rows.Err(); err != nil { + return fmt.Errorf("iterate legacy other recharge: %w", err) + } return nil } +func (s *AslanMongoDashboardSource) loadLegacyCoinSellerRechargeStats(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter, grid *legacyMongoGrid, distinctUsers map[aslanMongoDayCountryUsers]map[string]struct{}, perUser map[string]map[int64]int64) error { + if s.legacyDB == nil { + return nil + } + if err := s.loadLegacyGoldCoinRechargeStats(ctx, startDate, endDate, countryFilter, grid, distinctUsers, perUser); err != nil { + return err + } + if err := s.loadLegacyFreightPurchaseRechargeStats(ctx, startDate, endDate, countryFilter, grid, distinctUsers, perUser); err != nil { + return err + } + return nil +} + +func (s *AslanMongoDashboardSource) loadLegacyGoldCoinRechargeStats(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter, grid *legacyMongoGrid, distinctUsers map[aslanMongoDayCountryUsers]map[string]struct{}, perUser map[string]map[int64]int64) error { + where := []string{ + "ugcr.sys_origin = ?", + "ugcr.gold_coin_source = 2", + "ugcr.create_time >= ?", + "ugcr.create_time < ?", + "ubi.country_code IS NOT NULL", + "ubi.country_code != ''", + } + args := []any{s.sysOrigin, startDate, endDate} + if countryFilter.Applied { + if len(countryFilter.Codes) == 0 { + return nil + } + placeholders := make([]string, 0, len(countryFilter.Codes)) + for _, code := range countryFilter.Codes { + normalized := aslanMongoCountryCode(code) + if normalized == "" { + continue + } + placeholders = append(placeholders, "?") + args = append(args, normalized) + } + if len(placeholders) == 0 { + return nil + } + where = append(where, "UPPER(ubi.country_code) IN ("+strings.Join(placeholders, ",")+")") + } + + queryCtx, cancel := context.WithTimeout(ctx, s.requestTimeout) + defer cancel() + rows, err := s.legacyDB.QueryContext(queryCtx, ` +SELECT DATE_FORMAT(ugcr.create_time, '%Y-%m-%d') AS stat_day, + UPPER(ubi.country_code) AS country_code, + ugcr.user_id, + CAST(COALESCE(SUM(ugcr.recharge_amount), 0) AS CHAR) AS amount +FROM user_gold_coin_recharge ugcr + INNER JOIN user_base_info ubi ON ubi.id = ugcr.user_id +WHERE `+strings.Join(where, " AND ")+` +GROUP BY DATE_FORMAT(ugcr.create_time, '%Y-%m-%d'), UPPER(ubi.country_code), ugcr.user_id`, args...) + if err != nil { + return fmt.Errorf("query legacy coin seller recharge: %w", err) + } + defer rows.Close() + + for rows.Next() { + var day string + var countryCode string + var userID int64 + var amountText string + if err := rows.Scan(&day, &countryCode, &userID, &amountText); err != nil { + return fmt.Errorf("scan legacy coin seller recharge: %w", err) + } + day = strings.TrimSpace(day) + countryCode = aslanMongoCountryCode(countryCode) + if day == "" || countryCode == "" || userID <= 0 || !grid.hasDay(day) { + continue + } + amount, err := decimalTextToMinor(amountText) + if err != nil { + return fmt.Errorf("parse legacy coin seller recharge amount for %s/%s/%d: %w", day, countryCode, userID, err) + } + addLegacyCoinSellerRecharge(grid, distinctUsers, perUser, day, countryCode, userID, amount) + } + if err := rows.Err(); err != nil { + return fmt.Errorf("iterate legacy coin seller recharge: %w", err) + } + return nil +} + +func (s *AslanMongoDashboardSource) loadLegacyFreightPurchaseRechargeStats(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter, grid *legacyMongoGrid, distinctUsers map[aslanMongoDayCountryUsers]map[string]struct{}, perUser map[string]map[int64]int64) error { + walletDatabase := strings.TrimSpace(s.legacyWalletDatabase) + if walletDatabase == "" { + return nil + } + // 后台“金币代理 > 发货”实际是给代理进货：origin=PURCHASE/type=0，表单金额落 amount； + // App/经销商路径可能只落 usd_quantity，因此金额优先取 amount，空值再回退 usd_quantity。 + where := []string{ + "fbrw.sys_origin = ?", + "fbrw.origin = 'PURCHASE'", + "fbrw.type = 0", + "fbrw.create_time >= ?", + "fbrw.create_time < ?", + "(COALESCE(fbrw.amount, 0) <> 0 OR COALESCE(fbrw.usd_quantity, 0) <> 0)", + "ubi.country_code IS NOT NULL", + "ubi.country_code != ''", + } + args := []any{s.sysOrigin, startDate, endDate} + if countryFilter.Applied { + if len(countryFilter.Codes) == 0 { + return nil + } + placeholders := make([]string, 0, len(countryFilter.Codes)) + for _, code := range countryFilter.Codes { + normalized := aslanMongoCountryCode(code) + if normalized == "" { + continue + } + placeholders = append(placeholders, "?") + args = append(args, normalized) + } + if len(placeholders) == 0 { + return nil + } + where = append(where, "UPPER(ubi.country_code) IN ("+strings.Join(placeholders, ",")+")") + } + + queryCtx, cancel := context.WithTimeout(ctx, s.requestTimeout) + defer cancel() + rows, err := s.legacyDB.QueryContext(queryCtx, ` +SELECT DATE_FORMAT(fbrw.create_time, '%Y-%m-%d') AS stat_day, + UPPER(ubi.country_code) AS country_code, + fbrw.user_id, + CAST(COALESCE(SUM(CASE + WHEN fbrw.amount IS NOT NULL AND fbrw.amount <> 0 THEN fbrw.amount + ELSE COALESCE(fbrw.usd_quantity, 0) + END), 0) AS CHAR) AS amount +FROM `+quoteMySQLIdentifier(walletDatabase)+`.user_freight_balance_running_water fbrw + INNER JOIN user_base_info ubi ON ubi.id = fbrw.user_id +WHERE `+strings.Join(where, " AND ")+` +GROUP BY DATE_FORMAT(fbrw.create_time, '%Y-%m-%d'), UPPER(ubi.country_code), fbrw.user_id`, args...) + if err != nil { + return fmt.Errorf("query legacy freight purchase recharge: %w", err) + } + defer rows.Close() + + for rows.Next() { + var day string + var countryCode string + var userID int64 + var amountText string + if err := rows.Scan(&day, &countryCode, &userID, &amountText); err != nil { + return fmt.Errorf("scan legacy freight purchase recharge: %w", err) + } + day = strings.TrimSpace(day) + countryCode = aslanMongoCountryCode(countryCode) + if day == "" || countryCode == "" || userID <= 0 || !grid.hasDay(day) { + continue + } + amount, err := decimalTextToMinor(amountText) + if err != nil { + return fmt.Errorf("parse legacy freight purchase recharge amount for %s/%s/%d: %w", day, countryCode, userID, err) + } + addLegacyCoinSellerRecharge(grid, distinctUsers, perUser, day, countryCode, userID, amount) + } + if err := rows.Err(); err != nil { + return fmt.Errorf("iterate legacy freight purchase recharge: %w", err) + } + return nil +} + +func addLegacyCoinSellerRecharge(grid *legacyMongoGrid, distinctUsers map[aslanMongoDayCountryUsers]map[string]struct{}, perUser map[string]map[int64]int64, day string, countryCode string, userID int64, amount int64) { + if amount <= 0 { + return + } + metric := grid.at(day, countryCode) + metric.RechargeUSDMinor += amount + metric.CoinSellerRechargeUSDMinor += amount + addLegacyRechargeUser(distinctUsers, perUser, day, countryCode, userID, amount) +} + +func addLegacyStandardRecharge(grid *legacyMongoGrid, distinctUsers map[aslanMongoDayCountryUsers]map[string]struct{}, perUser map[string]map[int64]int64, day string, countryCode string, userID int64, amount int64) { + if amount <= 0 { + return + } + metric := grid.at(day, countryCode) + metric.RechargeUSDMinor += amount + metric.ThirdPartyUSDMinor += amount + addLegacyRechargeUser(distinctUsers, perUser, day, countryCode, userID, amount) +} + +func addLegacyRechargeUser(distinctUsers map[aslanMongoDayCountryUsers]map[string]struct{}, perUser map[string]map[int64]int64, day string, countryCode string, userID int64, amount int64) { + usersKey := aslanMongoDayCountryUsers{day: day, country: countryCode} + userSet := distinctUsers[usersKey] + if userSet == nil { + userSet = map[string]struct{}{} + distinctUsers[usersKey] = userSet + } + userSet[strconv.FormatInt(userID, 10)] = struct{}{} + + byUser := perUser[day] + if byUser == nil { + byUser = map[int64]int64{} + perUser[day] = byUser + } + byUser[userID] += amount +} + // loadGiftStats 按 天×赠送人 聚合礼物价值（金币），再经用户资料把流水拆到国家； // luckyGift=true 的部分同时计入幸运礼物流水。 func (s *AslanMongoDashboardSource) loadGiftStats(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter, grid *legacyMongoGrid) error { @@ -1158,6 +1485,18 @@ func (s *AslanMongoDashboardSource) dayExpression(field string) bson.D { }}} } +func aslanMongoExcludeFreightGoldRecharge(match bson.M) { + // FREIGHT_GOLD 成功单会落入 atyou_wallet.user_freight_balance_running_water 进货流水； + // legacy 版本商品字段不统一（顶层类型、products.code/name/content 均出现过），必须全部排除避免双算。 + for _, field := range aslanMongoFreightCommodityFields { + match[field] = bson.M{"$nin": aslanMongoFreightCommodityTypes} + } +} + +func dashboardDateNumber(day time.Time) string { + return day.Format("20060102") +} + // aslanMongoDocument 把聚合结果里的嵌套文档统一转成 bson.M； // driver 默认把嵌套文档解码成 bson.D，直接断言 bson.M 会静默丢行（表现为全零无报错）。 func aslanMongoDocument(value any) bson.M { @@ -1240,6 +1579,10 @@ func aslanMongoCountryCode(value any) string { return code } +func quoteMySQLIdentifier(value string) string { + return "`" + strings.ReplaceAll(value, "`", "``") + "`" +} + func aslanMongoDistinctValueCount(value any) int64 { values, ok := value.(bson.A) if !ok { @@ -1323,7 +1666,6 @@ func aslanMongoInt64Chunks(values []int64, size int) [][]int64 { // aslanMongoUnavailableMetricFields 是 likei Mongo 事实集合仍给不出的口径； // 游戏流水的 GoldOrigin code 枚举过杂（大量带后缀的游戏 code 与奖励类混在一起），未确认前不硬凑。 var aslanMongoUnavailableMetricFields = []string{ - "coin_seller_recharge_usd_minor", "coin_seller_stock_coin", "new_dealer_recharge_usd_minor", "game_turnover", @@ -1355,6 +1697,19 @@ func applyAslanMongoUnavailableFields(row map[string]any) { } } +var aslanMongoLegacyMetricFields = []string{ + "coin_seller_recharge_usd_minor", +} + +func applyAslanMongoLegacyUnavailableFields(row map[string]any, legacyRecharge bool) { + if legacyRecharge { + return + } + for _, field := range aslanMongoLegacyMetricFields { + row[field] = nil + } +} + // 金币流水集合缺失时必须显示 "--" 的字段；置 0 会让幸运礼物利润率假性 100%、币商/金币口径失真。 var aslanMongoGoldWaterMetricFields = []string{ "lucky_gift_payout", @@ -1375,7 +1730,7 @@ func applyAslanMongoGoldWaterUnavailable(row map[string]any) { } } -func aslanMongoDashboardMetricSources(goldWater bool) []map[string]any { +func aslanMongoDashboardMetricSources(goldWater bool, legacyRecharge bool) []map[string]any { goldWaterSources := []map[string]any{ {"field": "lucky_gift_payout", "available": true, "source": "likei Mongo user_gold_running_water_v2 origin=LUCKY_GIFT_GOLD_REWARD*"}, {"field": "coin_seller_transfer_coin", "available": true, "source": "likei Mongo user_gold_running_water_v2 origin=SELLER_AGENT*"}, @@ -1392,18 +1747,34 @@ func aslanMongoDashboardMetricSources(goldWater bool) []map[string]any { {"field": "consume_output_ratio", "available": false, "missing_reason": "该部署的 Mongo 没有 user_gold_running_water_v2 集合"}, } } - return append([]map[string]any{ + legacySources := []map[string]any{ + {"field": "coin_seller_recharge_usd_minor", "available": true, "source": "likei legacy MySQL user_gold_coin_recharge gold_coin_source=2 recharge_amount + user_freight_balance_running_water PURCHASE amount/usd_quantity"}, + } + rechargeUsersSource := "likei Mongo in_app_purchase_details SUCCESS distinct acceptUserId（排除 FREIGHT_GOLD）" + rechargeAmountSource := "likei Mongo in_app_purchase_details amountUsd（排除 FREIGHT_GOLD）" + mifaPaySource := "likei Mongo in_app_purchase_details factory.factoryCode!=GOOGLE（排除 FREIGHT_GOLD，全部三方渠道合计）" + newUserRechargeSource := "likei Mongo 当日注册 cohort × in_app_purchase_details（排除 FREIGHT_GOLD）" + if !legacyRecharge { + legacySources = []map[string]any{ + {"field": "coin_seller_recharge_usd_minor", "available": false, "missing_reason": "未配置 legacy_mysql_dsn，无法读取 Aslan 金币代理发货充值金额"}, + } + } else { + rechargeUsersSource += " + legacy MySQL order_other_recharge/user_gold_coin_recharge/user_freight_balance_running_water distinct user_id" + rechargeAmountSource += " + legacy MySQL order_other_recharge amount + user_gold_coin_recharge recharge_amount + user_freight_balance_running_water PURCHASE amount/usd_quantity" + mifaPaySource += " + legacy MySQL order_other_recharge amount" + newUserRechargeSource += "/legacy 其他充值/legacy 币商充值" + } + baseSources := []map[string]any{ {"field": "new_users", "available": true, "source": "likei Mongo user_run_profile createTime/countryCode"}, {"field": "active_users", "available": true, "source": "likei Mongo user_daily_active_log activeDate/registerCountryCode"}, - {"field": "paid_users", "available": true, "source": "likei Mongo in_app_purchase_details SUCCESS distinct acceptUserId"}, - {"field": "recharge_users", "available": true, "source": "likei Mongo in_app_purchase_details SUCCESS distinct acceptUserId"}, - {"field": "recharge_usd_minor", "available": true, "source": "likei Mongo in_app_purchase_details amountUsd"}, + {"field": "paid_users", "available": true, "source": rechargeUsersSource}, + {"field": "recharge_users", "available": true, "source": rechargeUsersSource}, + {"field": "recharge_usd_minor", "available": true, "source": rechargeAmountSource}, {"field": "google_recharge_usd_minor", "available": true, "source": "likei Mongo in_app_purchase_details factory.factoryCode=GOOGLE"}, - {"field": "mifapay_recharge_usd_minor", "available": true, "source": "likei Mongo in_app_purchase_details factory.factoryCode!=GOOGLE（全部三方渠道合计）"}, - {"field": "new_user_recharge_usd_minor", "available": true, "source": "likei Mongo 当日注册 cohort × in_app_purchase_details 当日充值"}, + {"field": "mifapay_recharge_usd_minor", "available": true, "source": mifaPaySource}, + {"field": "new_user_recharge_usd_minor", "available": true, "source": newUserRechargeSource}, {"field": "gift_coin_spent", "available": true, "source": "likei Mongo gift_give_running_water giftValue.actualAmount（含幸运礼物）"}, {"field": "lucky_gift_turnover", "available": true, "source": "likei Mongo gift_give_running_water luckyGift=true"}, - {"field": "coin_seller_recharge_usd_minor", "available": false, "missing_reason": "likei Mongo 未确认币商充值美元口径"}, {"field": "coin_seller_stock_coin", "available": false, "missing_reason": "likei Mongo 未确认币商进货金币口径"}, {"field": "game_turnover", "available": false, "missing_reason": "likei GoldOrigin 游戏 code 枚举过多且与奖励类混排，未确认口径前不聚合"}, {"field": "super_lucky_gift_turnover", "available": false, "missing_reason": "likei 平台没有超级幸运礼物玩法"}, @@ -1419,5 +1790,7 @@ func aslanMongoDashboardMetricSources(goldWater bool) []map[string]any { {"field": "retention.day1_rate", "available": true, "source": "likei Mongo 注册 cohort × user_daily_active_log D+1"}, {"field": "retention.day7_rate", "available": true, "source": "likei Mongo 注册 cohort × user_daily_active_log D+7"}, {"field": "retention.day30_rate", "available": true, "source": "likei Mongo 注册 cohort × user_daily_active_log D+30"}, - }, goldWaterSources...) + } + out := append(baseSources, legacySources...) + return append(out, goldWaterSources...) } diff --git a/server/admin/internal/modules/dashboard/aslan_mongo_dashboard_test.go b/server/admin/internal/modules/dashboard/aslan_mongo_dashboard_test.go index 2e944d50..1f2070e2 100644 --- a/server/admin/internal/modules/dashboard/aslan_mongo_dashboard_test.go +++ b/server/admin/internal/modules/dashboard/aslan_mongo_dashboard_test.go @@ -1,8 +1,11 @@ package dashboard import ( + "context" "testing" + "time" + "github.com/DATA-DOG/go-sqlmock" "go.mongodb.org/mongo-driver/v2/bson" ) @@ -56,3 +59,263 @@ func TestAslanMongoDocumentConversion(t *testing.T) { t.Fatalf("scalar should convert to nil, got %v", doc) } } + +func TestAslanMongoLegacyCoinSellerRechargeFeedsRechargeChannel(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock: %v", err) + } + defer db.Close() + + source := &AslanMongoDashboardSource{ + legacyDB: db, + sysOrigin: "ATYOU", + requestTimeout: time.Second, + } + start := time.Date(2026, 7, 5, 0, 0, 0, 0, time.UTC) + end := start.AddDate(0, 0, 1) + grid := newLegacyMongoGrid(start, end) + cohorts := map[string]map[string][]int64{ + "2026-07-05": { + "SA": {101}, + }, + } + distinctUsers := map[aslanMongoDayCountryUsers]map[string]struct{}{} + perUser := map[string]map[int64]int64{} + + mock.ExpectQuery("FROM user_gold_coin_recharge[\\s\\S]*ugcr\\.gold_coin_source = 2[\\s\\S]*ugcr\\.create_time >= \\?[\\s\\S]*ugcr\\.create_time < \\?"). + WithArgs("ATYOU", start, end). + WillReturnRows(sqlmock.NewRows([]string{"stat_day", "country_code", "user_id", "amount"}). + AddRow("2026-07-05", "sa", int64(101), "42.00"). + AddRow("2026-07-05", "AE", int64(202), "10.25")) + + if err := source.loadLegacyCoinSellerRechargeStats(context.Background(), start, end, externalDashboardCountryFilter{}, grid, distinctUsers, perUser); err != nil { + t.Fatalf("loadLegacyCoinSellerRechargeStats: %v", err) + } + applyAslanRechargeUserMetrics(cohorts, grid, distinctUsers, perUser) + + saMetric := grid.metrics["2026-07-05"]["SA"] + if saMetric == nil { + t.Fatalf("missing SA metric: %#v", grid.metrics) + } + saRow := saMetric.toExternal(nil).toMap() + assertInt64(t, saRow["coin_seller_recharge_usd_minor"], 4200) + assertInt64(t, saRow["recharge_usd_minor"], 4200) + assertInt64(t, saRow["total_recharge_usd_minor"], 4200) + assertInt64(t, saRow["user_recharge_usd_minor"], 0) + assertInt64(t, saRow["mifapay_recharge_usd_minor"], 0) + assertInt64(t, saRow["paid_users"], 1) + assertInt64(t, saRow["new_user_recharge_usd_minor"], 4200) + + aeMetric := grid.metrics["2026-07-05"]["AE"] + if aeMetric == nil { + t.Fatalf("missing AE metric: %#v", grid.metrics) + } + aeRow := aeMetric.toExternal(nil).toMap() + assertInt64(t, aeRow["coin_seller_recharge_usd_minor"], 1025) + assertInt64(t, aeRow["recharge_usd_minor"], 1025) + assertInt64(t, aeRow["new_user_recharge_usd_minor"], 0) + + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + +func TestAslanMongoLegacyOtherRechargeFeedsThirdPartyChannel(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock: %v", err) + } + defer db.Close() + + source := &AslanMongoDashboardSource{ + legacyDB: db, + sysOrigin: "ATYOU", + requestTimeout: time.Second, + } + start := time.Date(2026, 7, 5, 0, 0, 0, 0, time.UTC) + end := start.AddDate(0, 0, 1) + grid := newLegacyMongoGrid(start, end) + cohorts := map[string]map[string][]int64{ + "2026-07-05": { + "SA": {101, 202, 303}, + }, + } + distinctUsers := map[aslanMongoDayCountryUsers]map[string]struct{}{} + perUser := map[string]map[int64]int64{} + + mock.ExpectQuery("FROM order_other_recharge oor[\\s\\S]*INNER JOIN user_base_info ubi ON ubi\\.id = oor\\.user_id[\\s\\S]*oor\\.payment_date_number >= \\?[\\s\\S]*oor\\.payment_date_number < \\?"). + WithArgs("ATYOU", "20260705", "20260706"). + WillReturnRows(sqlmock.NewRows([]string{"stat_day", "country_code", "user_id", "amount"}). + AddRow("2026-07-05", "SA", int64(101), "12.34"). + AddRow("2026-07-05", "SA", int64(202), "0.00"). + AddRow("2026-07-05", "SA", int64(303), "-3.00")) + + if err := source.loadLegacyOtherRechargeStats(context.Background(), start, end, externalDashboardCountryFilter{}, grid, distinctUsers, perUser); err != nil { + t.Fatalf("loadLegacyOtherRechargeStats: %v", err) + } + applyAslanRechargeUserMetrics(cohorts, grid, distinctUsers, perUser) + + row := grid.metrics["2026-07-05"]["SA"].toExternal(nil).toMap() + assertInt64(t, row["recharge_usd_minor"], 1234) + assertInt64(t, row["mifapay_recharge_usd_minor"], 1234) + assertInt64(t, row["coin_seller_recharge_usd_minor"], 0) + assertInt64(t, row["user_recharge_usd_minor"], 1234) + assertInt64(t, row["paid_users"], 1) + assertInt64(t, row["new_user_recharge_usd_minor"], 1234) + + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + +func TestAslanMongoLegacyFreightPurchaseRechargeFeedsCoinSellerChannel(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock: %v", err) + } + defer db.Close() + + source := &AslanMongoDashboardSource{ + legacyDB: db, + legacyWalletDatabase: "atyou_wallet", + sysOrigin: "ATYOU", + requestTimeout: time.Second, + } + start := time.Date(2026, 7, 5, 0, 0, 0, 0, time.UTC) + end := start.AddDate(0, 0, 1) + grid := newLegacyMongoGrid(start, end) + cohorts := map[string]map[string][]int64{ + "2026-07-05": { + "PH": {2069828597070528514}, + }, + } + distinctUsers := map[aslanMongoDayCountryUsers]map[string]struct{}{} + perUser := map[string]map[int64]int64{} + + mock.ExpectQuery("FROM user_gold_coin_recharge[\\s\\S]*ugcr\\.gold_coin_source = 2"). + WithArgs("ATYOU", start, end). + WillReturnRows(sqlmock.NewRows([]string{"stat_day", "country_code", "user_id", "amount"})) + mock.ExpectQuery("FROM `atyou_wallet`\\.user_freight_balance_running_water fbrw[\\s\\S]*fbrw\\.origin = 'PURCHASE'[\\s\\S]*fbrw\\.type = 0"). + WithArgs("ATYOU", start, end). + WillReturnRows(sqlmock.NewRows([]string{"stat_day", "country_code", "user_id", "amount"}). + AddRow("2026-07-05", "PH", int64(2069828597070528514), "2000.00"). + AddRow("2026-07-05", "PK", int64(2071589123112779777), "25.47")) + + if err := source.loadLegacyCoinSellerRechargeStats(context.Background(), start, end, externalDashboardCountryFilter{}, grid, distinctUsers, perUser); err != nil { + t.Fatalf("loadLegacyCoinSellerRechargeStats: %v", err) + } + applyAslanRechargeUserMetrics(cohorts, grid, distinctUsers, perUser) + + phRow := grid.metrics["2026-07-05"]["PH"].toExternal(nil).toMap() + assertInt64(t, phRow["coin_seller_recharge_usd_minor"], 200000) + assertInt64(t, phRow["recharge_usd_minor"], 200000) + assertInt64(t, phRow["user_recharge_usd_minor"], 0) + assertInt64(t, phRow["paid_users"], 1) + assertInt64(t, phRow["new_user_recharge_usd_minor"], 200000) + + pkRow := grid.metrics["2026-07-05"]["PK"].toExternal(nil).toMap() + assertInt64(t, pkRow["coin_seller_recharge_usd_minor"], 2547) + assertInt64(t, pkRow["recharge_usd_minor"], 2547) + + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + +func TestAslanMongoRechargeMatchExcludesFreightGold(t *testing.T) { + match := bson.M{} + aslanMongoExcludeFreightGoldRecharge(match) + + for _, field := range []string{"trackCommodityType", "products.code", "products.name", "products.content"} { + condition, ok := match[field].(bson.M) + if !ok { + t.Fatalf("%s condition = %#v", field, match[field]) + } + values, ok := condition["$nin"].(bson.A) + if !ok { + t.Fatalf("%s $nin = %#v", field, condition["$nin"]) + } + got := map[any]bool{} + for _, value := range values { + got[value] = true + } + if !got["FREIGHT_GOLD"] || !got["FREIGHT_GOLD_SUPER"] { + t.Fatalf("%s should exclude freight commodity types, got %#v", field, values) + } + } +} + +func TestAslanMongoLegacyRechargeSkipsNonPositiveAmount(t *testing.T) { + start := time.Date(2026, 7, 5, 0, 0, 0, 0, time.UTC) + grid := newLegacyMongoGrid(start, start.AddDate(0, 0, 1)) + distinctUsers := map[aslanMongoDayCountryUsers]map[string]struct{}{} + perUser := map[string]map[int64]int64{} + + addLegacyStandardRecharge(grid, distinctUsers, perUser, "2026-07-05", "SA", 101, 0) + addLegacyCoinSellerRecharge(grid, distinctUsers, perUser, "2026-07-05", "SA", 102, -100) + + row := grid.metrics["2026-07-05"]["SA"] + if row != nil { + t.Fatalf("non-positive legacy recharge should not create metric row: %#v", row) + } + if len(distinctUsers) != 0 { + t.Fatalf("non-positive legacy recharge should not create paid users: %#v", distinctUsers) + } + if len(perUser) != 0 { + t.Fatalf("non-positive legacy recharge should not create per-user recharge: %#v", perUser) + } +} + +func TestAslanMongoCoinSellerRechargeSourceAvailability(t *testing.T) { + available := aslanMongoDashboardMetricSources(false, true) + assertMetricSourceAvailable(t, available, "coin_seller_recharge_usd_minor") + + unavailable := aslanMongoDashboardMetricSources(false, false) + assertMetricSourceUnavailable(t, unavailable, "coin_seller_recharge_usd_minor") +} + +func TestAslanMongoDailyRechargeUsersDeduplicateAcrossCountries(t *testing.T) { + start := time.Date(2026, 7, 5, 0, 0, 0, 0, time.UTC) + grid := newLegacyMongoGrid(start, start.AddDate(0, 0, 1)) + grid.at("2026-07-05", "SA").RechargeUSDMinor = 4200 + grid.at("2026-07-05", "AE").RechargeUSDMinor = 1000 + distinctUsers := map[aslanMongoDayCountryUsers]map[string]struct{}{ + {day: "2026-07-05", country: "SA"}: {"101": {}}, + {day: "2026-07-05", country: "AE"}: {"101": {}}, + } + applyAslanRechargeUserMetrics(nil, grid, distinctUsers, nil) + + overview := (&AslanMongoDashboardSource{}).assembleRange(grid, externalDashboardCountryFilter{}, false, true) + if overview.Total.PaidUsers != 1 { + t.Fatalf("total paid users = %d, want 1", overview.Total.PaidUsers) + } + if len(overview.DailySeries) != 1 { + t.Fatalf("daily series mismatch: %#v", overview.DailySeries) + } + assertInt64(t, overview.DailySeries[0]["paid_users"], 1) + + countries := map[string]int64{} + for _, row := range overview.CountryBreakdown { + country, _ := row["country_code"].(string) + paid, _ := row["paid_users"].(int64) + countries[country] = paid + } + if countries["SA"] != 1 || countries["AE"] != 1 { + t.Fatalf("country paid users should preserve country buckets, got %#v", countries) + } +} + +func assertMetricSourceAvailable(t *testing.T, sources []map[string]any, field string) { + t.Helper() + for _, source := range sources { + if source["field"] != field { + continue + } + if source["available"] != true { + t.Fatalf("metric source %s should be available, got %#v", field, source) + } + return + } + t.Fatalf("metric source %s not found in %#v", field, sources) +}