aslan币商数据

This commit is contained in:
zhx 2026-07-05 23:47:00 +08:00
parent 15810e891d
commit f839e44375
8 changed files with 950 additions and 99 deletions

View File

@ -96,10 +96,10 @@
- 默认使用 Go 标准库和仓库已有小包;不要为了轻量逻辑引入重框架。 - 默认使用 Go 标准库和仓库已有小包;不要为了轻量逻辑引入重框架。
- 手写代码保持高密度有效注释,所有业务逻辑、状态流转、权限判断、支付链路、异步流程都要解释工程语义和失败分支,不复述语法。 - 手写代码保持高密度有效注释,所有业务逻辑、状态流转、权限判断、支付链路、异步流程都要解释工程语义和失败分支,不复述语法。
- 新增配置必须同时考虑本地 `config.yaml`、Docker `config.docker.yaml`、以及线上配置示例。 - 新增配置必须同时考虑本地 `config.yaml`、Docker `config.docker.yaml`、以及线上配置示例。
- 所有逻辑功能配置默认开启;密钥、地址、三方账号等必须由运行环境提供的敏感配置只能使用占位符或环境配置,不能写入真实值,也不能伪造可用凭证 - 所有逻辑功能配置默认开启;密钥、地址、三方账号等敏感值只能由运行环境提供。可提交 YAML/示例配置只能写占位符;真实值只能写入已被 git 忽略的本地运行时 env 文件(例如 `.env.prod`)或部署平台环境变量,不能写入可提交文件
- 新增服务依赖必须在 `docker-compose.yml` 中体现本地可运行形态。 - 新增服务依赖必须在 `docker-compose.yml` 中体现本地可运行形态。
- 数据库表结构变更必须同步更新对应 owner service 的 `deploy/mysql/initdb` - 数据库表结构变更必须同步更新对应 owner service 的 `deploy/mysql/initdb`
- 不要把线上密钥、云数据库密码、Redis 密码写进真实配置。只能写占位示例 - 不要把线上密钥、云数据库密码、Redis 密码写进 YAML、示例配置、代码、测试或任何会进入 git 的文件;需要落地到文件时,只能写入 `.env.*` 这类已忽略的运行时 env 文件
- 保持 `gofmt`,不要手调 Go import 顺序。 - 保持 `gofmt`,不要手调 Go import 顺序。
## Test Expectations ## Test Expectations

View File

@ -5,6 +5,7 @@ import (
"database/sql" "database/sql"
"errors" "errors"
"flag" "flag"
"fmt"
"log" "log"
"log/slog" "log/slog"
"net/http" "net/http"
@ -613,16 +614,44 @@ func connectDashboardExternalSources(ctx context.Context, configs []config.Dashb
return nil, nil, err return nil, nil, err
} }
cancel() 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 err != nil {
if legacyDB != nil {
_ = legacyDB.Close()
}
_ = client.Disconnect(ctx) _ = client.Disconnect(ctx)
cleanup() cleanup()
return nil, nil, err return nil, nil, err
} }
if source == nil { if source == nil {
if legacyDB != nil {
_ = legacyDB.Close()
}
_ = client.Disconnect(ctx) _ = client.Disconnect(ctx)
continue continue
} }
if legacyDB != nil {
dbs = append(dbs, legacyDB)
}
mongoClients = append(mongoClients, client) mongoClients = append(mongoClients, client)
sources = append(sources, source) sources = append(sources, source)
continue continue

View File

@ -116,6 +116,9 @@ dashboard_external_sources:
sys_origin: "ATYOU" 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_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" 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" stat_timezone: "Asia/Shanghai"
# aslan_mongo 源按整段区间做聚合(注册/日活/充值/礼物 + 留存 cohort30 天区间比逐日查询重, # aslan_mongo 源按整段区间做聚合(注册/日活/充值/礼物 + 留存 cohort30 天区间比逐日查询重,
# 5s 会在长区间超时;线上建议 30s 起。 # 5s 会在长区间超时;线上建议 30s 起。

View File

@ -115,6 +115,9 @@ dashboard_external_sources:
sys_origin: "ATYOU" sys_origin: "ATYOU"
mongo_uri: "" mongo_uri: ""
mongo_database: "test" 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" stat_timezone: "Asia/Shanghai"
request_timeout: "5s" request_timeout: "5s"
finance_notifications: finance_notifications:

View File

@ -90,18 +90,20 @@ type StatisticsServiceConfig struct {
} }
type DashboardExternalSourceConfig struct { type DashboardExternalSourceConfig struct {
Enabled bool `yaml:"enabled"` Enabled bool `yaml:"enabled"`
SourceType string `yaml:"type"` SourceType string `yaml:"type"`
AppCode string `yaml:"app_code"` AppCode string `yaml:"app_code"`
AppName string `yaml:"app_name"` AppName string `yaml:"app_name"`
SysOrigin string `yaml:"sys_origin"` SysOrigin string `yaml:"sys_origin"`
MySQLDSN string `yaml:"mysql_dsn"` MySQLDSN string `yaml:"mysql_dsn"`
MongoURI string `yaml:"mongo_uri"` LegacyMySQLDSN string `yaml:"legacy_mysql_dsn"`
MongoDatabase string `yaml:"mongo_database"` LegacyWalletDatabase string `yaml:"legacy_wallet_database"`
BaseURL string `yaml:"base_url"` MongoURI string `yaml:"mongo_uri"`
OverviewPath string `yaml:"overview_path"` MongoDatabase string `yaml:"mongo_database"`
StatTimezone string `yaml:"stat_timezone"` BaseURL string `yaml:"base_url"`
RequestTimeout time.Duration `yaml:"request_timeout"` OverviewPath string `yaml:"overview_path"`
StatTimezone string `yaml:"stat_timezone"`
RequestTimeout time.Duration `yaml:"request_timeout"`
} }
type FinanceNotificationsConfig struct { type FinanceNotificationsConfig struct {
@ -454,6 +456,8 @@ func (cfg *Config) Normalize() {
source.AppName = strings.TrimSpace(source.AppName) source.AppName = strings.TrimSpace(source.AppName)
source.SysOrigin = strings.ToUpper(strings.TrimSpace(source.SysOrigin)) source.SysOrigin = strings.ToUpper(strings.TrimSpace(source.SysOrigin))
source.MySQLDSN = strings.TrimSpace(source.MySQLDSN) 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.MongoURI = strings.TrimSpace(source.MongoURI)
source.MongoDatabase = strings.Trim(strings.TrimSpace(source.MongoDatabase), "/") source.MongoDatabase = strings.Trim(strings.TrimSpace(source.MongoDatabase), "/")
source.BaseURL = strings.TrimRight(strings.TrimSpace(source.BaseURL), "/") source.BaseURL = strings.TrimRight(strings.TrimSpace(source.BaseURL), "/")
@ -473,6 +477,10 @@ func (cfg *Config) Normalize() {
if source.SourceType == "aslan_mongo" && source.MongoDatabase == "" { if source.SourceType == "aslan_mongo" && source.MongoDatabase == "" {
source.MongoDatabase = "test" 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) source.StatTimezone = strings.TrimSpace(source.StatTimezone)
if source.StatTimezone == "" { if source.StatTimezone == "" {
if source.SourceType == "aslan_http" || source.SourceType == "aslan_mongo" { if source.SourceType == "aslan_http" || source.SourceType == "aslan_mongo" {
@ -691,6 +699,18 @@ func (cfg Config) Validate() error {
if strings.TrimSpace(source.MongoDatabase) == "" { if strings.TrimSpace(source.MongoDatabase) == "" {
return fmt.Errorf("dashboard_external_sources[%d].mongo_database is required when aslan_mongo source is enabled", index) 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: default:
return fmt.Errorf("dashboard_external_sources[%d].type must be one of mysql, aslan_http, aslan_mongo", index) 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+"_DASHBOARD_MONGO_DATABASE",
"HYAPP_ADMIN_"+appKey+"_MONGO_DATABASE", "HYAPP_ADMIN_"+appKey+"_MONGO_DATABASE",
)); value != "" { )); value != "" {
previousMongoDatabase := source.MongoDatabase
usesDerivedWalletDatabase := source.LegacyWalletDatabase == "" || source.LegacyWalletDatabase == previousMongoDatabase+"_wallet"
source.MongoDatabase = strings.Trim(strings.TrimSpace(value), "/") 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 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 { func validEnvironment(value string) bool {
switch value { switch value {
case "local", "dev", "staging", "prod": case "local", "dev", "staging", "prod":

View File

@ -2,6 +2,7 @@ package config
import ( import (
"reflect" "reflect"
"strings"
"testing" "testing"
"time" "time"
) )
@ -75,6 +76,8 @@ func TestFinanceDingTalkEnvOverride(t *testing.T) {
func TestAslanMongoEnvOverride(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_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_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") t.Setenv("HYAPP_ADMIN_ASLAN_MONEY_REGION_MONGO_DATABASE", "tarab_region")
cfg := Default() cfg := Default()
@ -93,6 +96,12 @@ func TestAslanMongoEnvOverride(t *testing.T) {
if dashboardSource.MongoDatabase != "test_dashboard" { if dashboardSource.MongoDatabase != "test_dashboard" {
t.Fatalf("dashboard mongo database mismatch: %q", dashboardSource.MongoDatabase) 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 var regionSource MoneyRegionSourceConfig
for _, source := range cfg.MoneyRegionSources { for _, source := range cfg.MoneyRegionSources {
@ -108,3 +117,118 @@ func TestAslanMongoEnvOverride(t *testing.T) {
t.Fatalf("region mongo database mismatch: %q", regionSource.MongoDatabase) 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)
}
}

View File

@ -2,6 +2,7 @@ package dashboard
import ( import (
"context" "context"
"database/sql"
"fmt" "fmt"
"sort" "sort"
"strconv" "strconv"
@ -32,14 +33,21 @@ const (
aslanMongoSellerAgentOriginPrefix = "SELLER_AGENT" aslanMongoSellerAgentOriginPrefix = "SELLER_AGENT"
) )
var (
aslanMongoFreightCommodityTypes = bson.A{"FREIGHT_GOLD", "FREIGHT_GOLD_SUPER"}
aslanMongoFreightCommodityFields = []string{"trackCommodityType", "products.code", "products.name", "products.content"}
)
type AslanMongoDashboardSource struct { type AslanMongoDashboardSource struct {
db *mongo.Database db *mongo.Database
appCode string legacyDB *sql.DB
appName string legacyWalletDatabase string
sysOrigin string appCode string
statTimezone string appName string
requestTimeout time.Duration sysOrigin string
regionResolvers []ExternalDashboardRegionResolver statTimezone string
requestTimeout time.Duration
regionResolvers []ExternalDashboardRegionResolver
// 部分 likei 部署(如 Aslan 的 atyou 库)没有 user_gold_running_water_v2 集合; // 部分 likei 部署(如 Aslan 的 atyou 库)没有 user_gold_running_water_v2 集合;
// 探测结果缓存住,避免把"集合不存在"算成 0幸运礼物利润率会假性 100%)。 // 探测结果缓存住,避免把"集合不存在"算成 0幸运礼物利润率会假性 100%)。
@ -69,6 +77,16 @@ func (s *AslanMongoDashboardSource) hasGoldWater(ctx context.Context) bool {
// NewAslanMongoDashboardSource 直接读取 likei 平台Aslan/Yumi 同构Mongo 里的已存在事实集合。 // NewAslanMongoDashboardSource 直接读取 likei 平台Aslan/Yumi 同构Mongo 里的已存在事实集合。
// 这里只做查询聚合和字段映射,不把数据复制到 hyapp statistics-service也不回退旧 console HTTP 接口。 // 这里只做查询聚合和字段映射,不把数据复制到 hyapp statistics-service也不回退旧 console HTTP 接口。
func NewAslanMongoDashboardSource(database *mongo.Database, sourceConfig config.DashboardExternalSourceConfig, regionResolvers ...ExternalDashboardRegionResolver) (ExternalDashboardSource, error) { 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 { if database == nil || !sourceConfig.Enabled {
return nil, nil return nil, nil
} }
@ -89,13 +107,15 @@ func NewAslanMongoDashboardSource(database *mongo.Database, sourceConfig config.
timeout = 5 * time.Second timeout = 5 * time.Second
} }
return &AslanMongoDashboardSource{ return &AslanMongoDashboardSource{
db: database, db: database,
appCode: appCode, legacyDB: legacyDB,
appName: strings.TrimSpace(sourceConfig.AppName), legacyWalletDatabase: strings.TrimSpace(sourceConfig.LegacyWalletDatabase),
sysOrigin: sysOrigin, appCode: appCode,
statTimezone: statTimezone, appName: strings.TrimSpace(sourceConfig.AppName),
requestTimeout: timeout, sysOrigin: sysOrigin,
regionResolvers: compactExternalDashboardRegionResolvers(regionResolvers), statTimezone: statTimezone,
requestTimeout: timeout,
regionResolvers: compactExternalDashboardRegionResolvers(regionResolvers),
}, nil }, nil
} }
@ -162,11 +182,12 @@ func (s *AslanMongoDashboardSource) StatisticsOverview(ctx context.Context, quer
response["daily_country_breakdown"] = current.DailyCountryBreakdown response["daily_country_breakdown"] = current.DailyCountryBreakdown
goldWater := s.hasGoldWater(ctx) goldWater := s.hasGoldWater(ctx)
response["retention"] = current.Total.retentionMap() 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() response["updated_at_ms"] = time.Now().UTC().UnixMilli()
applyExternalDashboardDeltas(response, current.Total, previous.Total) applyExternalDashboardDeltas(response, current.Total, previous.Total)
response["salary_delta_rate"] = deltaRate(nullableMetricInt64(current.Total.SalaryUSDMinor), nullableMetricInt64(previous.Total.SalaryUSDMinor)) response["salary_delta_rate"] = deltaRate(nullableMetricInt64(current.Total.SalaryUSDMinor), nullableMetricInt64(previous.Total.SalaryUSDMinor))
applyAslanMongoUnavailableFields(response) applyAslanMongoUnavailableFields(response)
applyAslanMongoLegacyUnavailableFields(response, s.legacyDB != nil)
if !goldWater { if !goldWater {
applyAslanMongoGoldWaterUnavailable(response) applyAslanMongoGoldWaterUnavailable(response)
response["lucky_gift_payout_delta_rate"] = nil response["lucky_gift_payout_delta_rate"] = nil
@ -203,25 +224,26 @@ func (s *AslanMongoDashboardSource) countryFilter(ctx context.Context, query Sta
// legacyMongoMetric 是 likei Mongo 源按 天×国家 汇总出的事实指标; // legacyMongoMetric 是 likei Mongo 源按 天×国家 汇总出的事实指标;
// 消耗/产出金币是全 App 口径(按用户拆国家需要全量账变反查用户,代价过高),只挂在日行和合计上。 // 消耗/产出金币是全 App 口径(按用户拆国家需要全量账变反查用户,代价过高),只挂在日行和合计上。
type legacyMongoMetric struct { type legacyMongoMetric struct {
NewUsers int64 NewUsers int64
ActiveUsers int64 ActiveUsers int64
PaidUsers int64 PaidUsers int64
RechargeUSDMinor int64 RechargeUSDMinor int64
GoogleUSDMinor int64 GoogleUSDMinor int64
ThirdPartyUSDMinor int64 ThirdPartyUSDMinor int64
NewUserRechargeUSDMinor int64 CoinSellerRechargeUSDMinor int64
GiftCoinSpent int64 NewUserRechargeUSDMinor int64
LuckyGiftTurnover int64 GiftCoinSpent int64
LuckyGiftPayout int64 LuckyGiftTurnover int64
LuckyGiftPayers int64 LuckyGiftPayout int64
CoinSellerTransferCoin int64 LuckyGiftPayers int64
SalaryUSDMinor int64 CoinSellerTransferCoin int64
D1Users int64 SalaryUSDMinor int64
D1Base int64 D1Users int64
D7Users int64 D1Base int64
D7Base int64 D7Users int64
D30Users int64 D7Base int64
D30Base int64 D30Users int64
D30Base int64
} }
func (m *legacyMongoMetric) add(other *legacyMongoMetric) { func (m *legacyMongoMetric) add(other *legacyMongoMetric) {
@ -234,6 +256,7 @@ func (m *legacyMongoMetric) add(other *legacyMongoMetric) {
m.RechargeUSDMinor += other.RechargeUSDMinor m.RechargeUSDMinor += other.RechargeUSDMinor
m.GoogleUSDMinor += other.GoogleUSDMinor m.GoogleUSDMinor += other.GoogleUSDMinor
m.ThirdPartyUSDMinor += other.ThirdPartyUSDMinor m.ThirdPartyUSDMinor += other.ThirdPartyUSDMinor
m.CoinSellerRechargeUSDMinor += other.CoinSellerRechargeUSDMinor
m.NewUserRechargeUSDMinor += other.NewUserRechargeUSDMinor m.NewUserRechargeUSDMinor += other.NewUserRechargeUSDMinor
m.GiftCoinSpent += other.GiftCoinSpent m.GiftCoinSpent += other.GiftCoinSpent
m.LuckyGiftTurnover += other.LuckyGiftTurnover m.LuckyGiftTurnover += other.LuckyGiftTurnover
@ -255,53 +278,57 @@ type legacyMongoCoinFlow struct {
} }
func (m *legacyMongoMetric) toExternal(coinFlow *legacyMongoCoinFlow) externalDashboardMetric { func (m *legacyMongoMetric) toExternal(coinFlow *legacyMongoCoinFlow) externalDashboardMetric {
userRecharge := m.RechargeUSDMinor - m.CoinSellerRechargeUSDMinor
metric := externalDashboardMetric{ metric := externalDashboardMetric{
NewUsers: m.NewUsers, NewUsers: m.NewUsers,
ActiveUsers: m.ActiveUsers, ActiveUsers: m.ActiveUsers,
PaidUsers: m.PaidUsers, PaidUsers: m.PaidUsers,
RechargeUSDMinor: m.RechargeUSDMinor, RechargeUSDMinor: m.RechargeUSDMinor,
UserRechargeUSDMinor: m.RechargeUSDMinor, UserRechargeUSDMinor: userRecharge,
GoogleRechargeUSDMinor: m.GoogleUSDMinor, GoogleRechargeUSDMinor: m.GoogleUSDMinor,
MifaPayRechargeUSDMinor: m.ThirdPartyUSDMinor, MifaPayRechargeUSDMinor: m.ThirdPartyUSDMinor,
NewUserRechargeUSDMinor: m.NewUserRechargeUSDMinor, CoinSellerRechargeUSDMinor: m.CoinSellerRechargeUSDMinor,
GiftCoinSpent: m.GiftCoinSpent, NewUserRechargeUSDMinor: m.NewUserRechargeUSDMinor,
LuckyGiftTurnover: m.LuckyGiftTurnover, GiftCoinSpent: m.GiftCoinSpent,
LuckyGiftPayout: m.LuckyGiftPayout, LuckyGiftTurnover: m.LuckyGiftTurnover,
LuckyGiftPayers: m.LuckyGiftPayers, LuckyGiftPayout: m.LuckyGiftPayout,
CoinSellerTransferCoin: int64Ptr(m.CoinSellerTransferCoin), LuckyGiftPayers: m.LuckyGiftPayers,
SalaryUSDMinor: int64Ptr(m.SalaryUSDMinor), CoinSellerTransferCoin: int64Ptr(m.CoinSellerTransferCoin),
D1RetentionUsers: m.D1Users, SalaryUSDMinor: int64Ptr(m.SalaryUSDMinor),
D1RetentionBaseUsers: m.D1Base, D1RetentionUsers: m.D1Users,
D7RetentionUsers: m.D7Users, D1RetentionBaseUsers: m.D1Base,
D7RetentionBaseUsers: m.D7Base, D7RetentionUsers: m.D7Users,
D30RetentionUsers: m.D30Users, D7RetentionBaseUsers: m.D7Base,
D30RetentionBaseUsers: m.D30Base, D30RetentionUsers: m.D30Users,
D30RetentionBaseUsers: m.D30Base,
} }
if coinFlow != nil { if coinFlow != nil {
metric.ConsumedCoin = int64Ptr(coinFlow.ConsumedCoin) metric.ConsumedCoin = int64Ptr(coinFlow.ConsumedCoin)
metric.OutputCoin = int64Ptr(coinFlow.OutputCoin) metric.OutputCoin = int64Ptr(coinFlow.OutputCoin)
} }
metric.finalize() metric.finalize()
// finalize 会用 official/google/mifapay 组合重推总充值likei 口径 total 就是 amountUsd 求和,避免二次推导 // finalize 会按拆分字段重推总充值likei Mongo 源已经把 in-app 与 legacy 币商金额合成总额,最终以事实聚合值为准
metric.RechargeUSDMinor = m.RechargeUSDMinor metric.RechargeUSDMinor = m.RechargeUSDMinor
metric.UserRechargeUSDMinor = m.RechargeUSDMinor metric.UserRechargeUSDMinor = userRecharge
return metric return metric
} }
type legacyMongoGrid struct { type legacyMongoGrid struct {
days []string days []string
daySet map[string]struct{} daySet map[string]struct{}
metrics map[string]map[string]*legacyMongoMetric metrics map[string]map[string]*legacyMongoMetric
coinFlows map[string]*legacyMongoCoinFlow coinFlows map[string]*legacyMongoCoinFlow
rechargeUsers map[string]map[string]struct{}
// 各 loader 并行写入map 结构由锁保护;不同 loader 写 metric 的不同字段,无字段级竞争。 // 各 loader 并行写入map 结构由锁保护;不同 loader 写 metric 的不同字段,无字段级竞争。
mu sync.Mutex mu sync.Mutex
} }
func newLegacyMongoGrid(startDate time.Time, endDate time.Time) *legacyMongoGrid { func newLegacyMongoGrid(startDate time.Time, endDate time.Time) *legacyMongoGrid {
grid := &legacyMongoGrid{ grid := &legacyMongoGrid{
daySet: map[string]struct{}{}, daySet: map[string]struct{}{},
metrics: map[string]map[string]*legacyMongoMetric{}, metrics: map[string]map[string]*legacyMongoMetric{},
coinFlows: map[string]*legacyMongoCoinFlow{}, coinFlows: map[string]*legacyMongoCoinFlow{},
rechargeUsers: map[string]map[string]struct{}{},
} }
for day := startDate; day.Before(endDate); day = day.AddDate(0, 0, 1) { for day := startDate; day.Before(endDate); day = day.AddDate(0, 0, 1) {
text := dashboardSQLDate(day) 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{} dailySeries := []map[string]any{}
dailyCountries := []map[string]any{} dailyCountries := []map[string]any{}
totalFlow := legacyMongoMetric{} totalFlow := legacyMongoMetric{}
@ -440,6 +467,7 @@ func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, country
row["region_id"] = countryFilter.RegionID row["region_id"] = countryFilter.RegionID
} }
applyAslanMongoUnavailableFields(row) applyAslanMongoUnavailableFields(row)
applyAslanMongoLegacyUnavailableFields(row, legacyRecharge)
if !goldWater { if !goldWater {
applyAslanMongoGoldWaterUnavailable(row) applyAslanMongoGoldWaterUnavailable(row)
} }
@ -452,6 +480,10 @@ func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, country
} }
// 工资余额是快照:日行取当日各国快照之和,区间合计不能跨天累加。 // 工资余额是快照:日行取当日各国快照之和,区间合计不能跨天累加。
daySalary := dayFlow.SalaryUSDMinor daySalary := dayFlow.SalaryUSDMinor
if dayRechargeUsers := grid.rechargeUsers[day]; len(dayRechargeUsers) > 0 {
// 国家拆分仍保留各国家充值人数;日合计按 user_id 去重,避免同一用户因不同事实源国家码不一致而被算两次。
dayFlow.PaidUsers = int64(len(dayRechargeUsers))
}
totalFlow.add(&dayFlow) totalFlow.add(&dayFlow)
totalFlow.SalaryUSDMinor = daySalary totalFlow.SalaryUSDMinor = daySalary
@ -459,6 +491,7 @@ func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, country
dayRow["label"] = day dayRow["label"] = day
dayRow["stat_day"] = day dayRow["stat_day"] = day
applyAslanMongoUnavailableFields(dayRow) applyAslanMongoUnavailableFields(dayRow)
applyAslanMongoLegacyUnavailableFields(dayRow, legacyRecharge)
if !goldWater { if !goldWater {
applyAslanMongoGoldWaterUnavailable(dayRow) applyAslanMongoGoldWaterUnavailable(dayRow)
} }
@ -481,6 +514,7 @@ func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, country
row["region_id"] = countryFilter.RegionID row["region_id"] = countryFilter.RegionID
} }
applyAslanMongoUnavailableFields(row) applyAslanMongoUnavailableFields(row)
applyAslanMongoLegacyUnavailableFields(row, legacyRecharge)
if !goldWater { if !goldWater {
applyAslanMongoGoldWaterUnavailable(row) applyAslanMongoGoldWaterUnavailable(row)
} }
@ -579,6 +613,11 @@ func (s *AslanMongoDashboardSource) loadActiveCounts(ctx context.Context, startD
return nil return nil
} }
type aslanMongoDayCountryUsers struct {
day string
country string
}
// loadRechargeStats 一次聚合出 天×国家×渠道 的充值金额与充值用户,再按当日注册 cohort 拆出新用户充值。 // 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 { 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{ match := bson.M{
@ -588,6 +627,7 @@ func (s *AslanMongoDashboardSource) loadRechargeStats(ctx context.Context, start
"createTime": bson.M{"$gte": startDate, "$lt": endDate}, "createTime": bson.M{"$gte": startDate, "$lt": endDate},
"countryCode": bson.M{"$exists": true, "$ne": ""}, "countryCode": bson.M{"$exists": true, "$ne": ""},
} }
aslanMongoExcludeFreightGoldRecharge(match)
aslanMongoApplyCountryFilter(match, "countryCode", countryFilter) aslanMongoApplyCountryFilter(match, "countryCode", countryFilter)
// Aslan 原 Top 充值统计按 acceptUserId 聚合createUser 只在 acceptUserId 缺失的旧单据上兜底。 // Aslan 原 Top 充值统计按 acceptUserId 聚合createUser 只在 acceptUserId 缺失的旧单据上兜底。
userExpression := bson.D{{Key: "$ifNull", Value: bson.A{"$acceptUserId", "$createUser"}}} 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}}}, {Key: "users", Value: bson.D{{Key: "$addToSet", Value: userExpression}}},
}}}, }}},
} }
type dayCountryUsers struct { distinctUsers := map[aslanMongoDayCountryUsers]map[string]struct{}{}
day string
country string
}
distinctUsers := map[dayCountryUsers]map[string]struct{}{}
err := s.aggregateEach(ctx, aslanMongoInAppPurchaseCollection, pipeline, func(row bson.M) error { err := s.aggregateEach(ctx, aslanMongoInAppPurchaseCollection, pipeline, func(row bson.M) error {
key := aslanMongoDocument(row["_id"]) key := aslanMongoDocument(row["_id"])
if key == nil { if key == nil {
@ -630,7 +666,7 @@ func (s *AslanMongoDashboardSource) loadRechargeStats(ctx context.Context, start
} else { } else {
metric.ThirdPartyUSDMinor += amount metric.ThirdPartyUSDMinor += amount
} }
usersKey := dayCountryUsers{day: day, country: countryCode} usersKey := aslanMongoDayCountryUsers{day: day, country: countryCode}
userSet := distinctUsers[usersKey] userSet := distinctUsers[usersKey]
if userSet == nil { if userSet == nil {
userSet = map[string]struct{}{} userSet = map[string]struct{}{}
@ -644,9 +680,6 @@ func (s *AslanMongoDashboardSource) loadRechargeStats(ctx context.Context, start
if err != nil { if err != nil {
return fmt.Errorf("aggregate recharge stats: %w", err) 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 自带国家归属。 // 新用户充值:按 天×用户 聚合当日充值,再与当日注册 cohort 求交集cohort 自带国家归属。
perUserPipeline := mongo.Pipeline{ perUserPipeline := mongo.Pipeline{
@ -685,6 +718,28 @@ func (s *AslanMongoDashboardSource) loadRechargeStats(ctx context.Context, start
if err != nil { if err != nil {
return fmt.Errorf("aggregate per-user recharge: %w", err) 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 { for day, byCountry := range cohorts {
byUser := perUser[day] byUser := perUser[day]
if len(byUser) == 0 { 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 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 按 天×赠送人 聚合礼物价值(金币),再经用户资料把流水拆到国家; // loadGiftStats 按 天×赠送人 聚合礼物价值(金币),再经用户资料把流水拆到国家;
// luckyGift=true 的部分同时计入幸运礼物流水。 // luckyGift=true 的部分同时计入幸运礼物流水。
func (s *AslanMongoDashboardSource) loadGiftStats(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter, grid *legacyMongoGrid) error { 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 // aslanMongoDocument 把聚合结果里的嵌套文档统一转成 bson.M
// driver 默认把嵌套文档解码成 bson.D直接断言 bson.M 会静默丢行(表现为全零无报错)。 // driver 默认把嵌套文档解码成 bson.D直接断言 bson.M 会静默丢行(表现为全零无报错)。
func aslanMongoDocument(value any) bson.M { func aslanMongoDocument(value any) bson.M {
@ -1240,6 +1579,10 @@ func aslanMongoCountryCode(value any) string {
return code return code
} }
func quoteMySQLIdentifier(value string) string {
return "`" + strings.ReplaceAll(value, "`", "``") + "`"
}
func aslanMongoDistinctValueCount(value any) int64 { func aslanMongoDistinctValueCount(value any) int64 {
values, ok := value.(bson.A) values, ok := value.(bson.A)
if !ok { if !ok {
@ -1323,7 +1666,6 @@ func aslanMongoInt64Chunks(values []int64, size int) [][]int64 {
// aslanMongoUnavailableMetricFields 是 likei Mongo 事实集合仍给不出的口径; // aslanMongoUnavailableMetricFields 是 likei Mongo 事实集合仍给不出的口径;
// 游戏流水的 GoldOrigin code 枚举过杂(大量带后缀的游戏 code 与奖励类混在一起),未确认前不硬凑。 // 游戏流水的 GoldOrigin code 枚举过杂(大量带后缀的游戏 code 与奖励类混在一起),未确认前不硬凑。
var aslanMongoUnavailableMetricFields = []string{ var aslanMongoUnavailableMetricFields = []string{
"coin_seller_recharge_usd_minor",
"coin_seller_stock_coin", "coin_seller_stock_coin",
"new_dealer_recharge_usd_minor", "new_dealer_recharge_usd_minor",
"game_turnover", "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%、币商/金币口径失真。 // 金币流水集合缺失时必须显示 "--" 的字段;置 0 会让幸运礼物利润率假性 100%、币商/金币口径失真。
var aslanMongoGoldWaterMetricFields = []string{ var aslanMongoGoldWaterMetricFields = []string{
"lucky_gift_payout", "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{ goldWaterSources := []map[string]any{
{"field": "lucky_gift_payout", "available": true, "source": "likei Mongo user_gold_running_water_v2 origin=LUCKY_GIFT_GOLD_REWARD*"}, {"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*"}, {"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 集合"}, {"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": "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": "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": "paid_users", "available": true, "source": rechargeUsersSource},
{"field": "recharge_users", "available": true, "source": "likei Mongo in_app_purchase_details SUCCESS distinct acceptUserId"}, {"field": "recharge_users", "available": true, "source": rechargeUsersSource},
{"field": "recharge_usd_minor", "available": true, "source": "likei Mongo in_app_purchase_details amountUsd"}, {"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": "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": "mifapay_recharge_usd_minor", "available": true, "source": mifaPaySource},
{"field": "new_user_recharge_usd_minor", "available": true, "source": "likei Mongo 当日注册 cohort × in_app_purchase_details 当日充值"}, {"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": "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": "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": "coin_seller_stock_coin", "available": false, "missing_reason": "likei Mongo 未确认币商进货金币口径"},
{"field": "game_turnover", "available": false, "missing_reason": "likei GoldOrigin 游戏 code 枚举过多且与奖励类混排,未确认口径前不聚合"}, {"field": "game_turnover", "available": false, "missing_reason": "likei GoldOrigin 游戏 code 枚举过多且与奖励类混排,未确认口径前不聚合"},
{"field": "super_lucky_gift_turnover", "available": false, "missing_reason": "likei 平台没有超级幸运礼物玩法"}, {"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.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.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"}, {"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...)
} }

View File

@ -1,8 +1,11 @@
package dashboard package dashboard
import ( import (
"context"
"testing" "testing"
"time"
"github.com/DATA-DOG/go-sqlmock"
"go.mongodb.org/mongo-driver/v2/bson" "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) 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)
}