1839 lines
67 KiB
Go
1839 lines
67 KiB
Go
package dashboard
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"fmt"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"hyapp-admin-server/internal/config"
|
||
|
||
"go.mongodb.org/mongo-driver/v2/bson"
|
||
"go.mongodb.org/mongo-driver/v2/mongo"
|
||
"go.mongodb.org/mongo-driver/v2/mongo/options"
|
||
)
|
||
|
||
const (
|
||
aslanMongoInAppPurchaseCollection = "in_app_purchase_details"
|
||
aslanMongoUserProfileCollection = "user_run_profile"
|
||
aslanMongoUserActiveCollection = "user_daily_active_log"
|
||
aslanMongoTeamMemberCollection = "team_member"
|
||
aslanMongoBankBalanceCollection = "user_bank_balance"
|
||
aslanMongoGiftWaterCollection = "gift_give_running_water"
|
||
aslanMongoGoldWaterCollection = "user_gold_running_water_v2"
|
||
aslanMongoChunkSize = 800
|
||
|
||
// likei GoldOrigin 枚举以 code 形式落在 user_gold_running_water_v2.origin；
|
||
// 历史上部分 code 带数字后缀（如 HOT_GAME_200），所以统一用前缀匹配。
|
||
aslanMongoLuckyGiftPayoutOriginPrefix = "LUCKY_GIFT_GOLD_REWARD"
|
||
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
|
||
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%）。
|
||
goldWaterMu sync.Mutex
|
||
goldWaterChecked bool
|
||
goldWaterAvailable bool
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) hasGoldWater(ctx context.Context) bool {
|
||
s.goldWaterMu.Lock()
|
||
defer s.goldWaterMu.Unlock()
|
||
if s.goldWaterChecked {
|
||
return s.goldWaterAvailable
|
||
}
|
||
queryCtx, cancel := context.WithTimeout(ctx, s.requestTimeout)
|
||
defer cancel()
|
||
count, err := s.db.Collection(aslanMongoGoldWaterCollection).EstimatedDocumentCount(queryCtx)
|
||
if err != nil {
|
||
// 探测失败不缓存结论（下次请求重试）；本次按不可用处理，宁缺毋错。
|
||
return false
|
||
}
|
||
s.goldWaterChecked = true
|
||
s.goldWaterAvailable = count > 0
|
||
return s.goldWaterAvailable
|
||
}
|
||
|
||
// 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
|
||
}
|
||
appCode := strings.ToLower(strings.TrimSpace(sourceConfig.AppCode))
|
||
sysOrigin := strings.ToUpper(strings.TrimSpace(sourceConfig.SysOrigin))
|
||
if appCode == "" || sysOrigin == "" {
|
||
return nil, nil
|
||
}
|
||
statTimezone := strings.TrimSpace(sourceConfig.StatTimezone)
|
||
if statTimezone == "" {
|
||
statTimezone = "Asia/Shanghai"
|
||
}
|
||
if _, err := time.LoadLocation(statTimezone); err != nil {
|
||
return nil, fmt.Errorf("load aslan mongo storage timezone %s: %w", statTimezone, err)
|
||
}
|
||
timeout := sourceConfig.RequestTimeout
|
||
if timeout <= 0 {
|
||
timeout = 5 * time.Second
|
||
}
|
||
return &AslanMongoDashboardSource{
|
||
db: database,
|
||
legacyDB: legacyDB,
|
||
legacyWalletDatabase: strings.TrimSpace(sourceConfig.LegacyWalletDatabase),
|
||
appCode: appCode,
|
||
appName: strings.TrimSpace(sourceConfig.AppName),
|
||
sysOrigin: sysOrigin,
|
||
statTimezone: statTimezone,
|
||
requestTimeout: timeout,
|
||
regionResolvers: compactExternalDashboardRegionResolvers(regionResolvers),
|
||
}, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) AppCode() string {
|
||
if s == nil {
|
||
return ""
|
||
}
|
||
return s.appCode
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) StatisticsOverview(ctx context.Context, query StatisticsQuery) (map[string]any, error) {
|
||
if s == nil || s.db == nil {
|
||
return nil, fmt.Errorf("aslan mongo dashboard source is not configured")
|
||
}
|
||
countryFilter, err := s.countryFilter(ctx, query)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
requestTimezone := strings.TrimSpace(query.StatTZ)
|
||
if requestTimezone == "" {
|
||
requestTimezone = s.statTimezone
|
||
}
|
||
requestLocation, err := time.LoadLocation(requestTimezone)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("load aslan mongo request timezone %s: %w", requestTimezone, err)
|
||
}
|
||
startDate, endDate, startMS, endMS := externalDashboardDateRange(query, requestLocation)
|
||
|
||
days := countCalendarDays(startDate, endDate)
|
||
previousStart := startDate.AddDate(0, 0, -days)
|
||
var current, previous aslanRangeOverview
|
||
// 当前周期与环比周期互不依赖，并行加载；环比只消费流水字段，昂贵的留存链路跳过。
|
||
if err := runConcurrently([]func() error{
|
||
func() error {
|
||
out, err := s.loadRange(ctx, startDate, endDate, countryFilter, true)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
current = out
|
||
return nil
|
||
},
|
||
func() error {
|
||
out, err := s.loadRange(ctx, previousStart, startDate, countryFilter, false)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
previous = out
|
||
return nil
|
||
},
|
||
}); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
response := current.Total.toMap()
|
||
response["app_code"] = s.appCode
|
||
response["app_name"] = firstExternalDashboardValue(s.appName, s.appCode)
|
||
response["sys_origin"] = s.sysOrigin
|
||
response["stat_tz"] = s.statTimezone
|
||
response["request_stat_tz"] = requestTimezone
|
||
response["start_ms"] = startMS
|
||
response["end_ms"] = endMS
|
||
response["daily_series"] = current.DailySeries
|
||
response["country_breakdown"] = current.CountryBreakdown
|
||
response["daily_country_breakdown"] = current.DailyCountryBreakdown
|
||
goldWater := s.hasGoldWater(ctx)
|
||
response["retention"] = current.Total.retentionMap()
|
||
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
|
||
response["lucky_gift_profit_delta_rate"] = nil
|
||
}
|
||
return response, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) countryFilter(ctx context.Context, query StatisticsQuery) (externalDashboardCountryFilter, error) {
|
||
if query.CountryID > 0 {
|
||
return externalDashboardCountryFilter{}, fmt.Errorf("aslan mongo dashboard source %s does not support numeric country_id filter", s.appCode)
|
||
}
|
||
if query.RegionID <= 0 {
|
||
return externalDashboardCountryFilter{}, nil
|
||
}
|
||
for _, resolver := range s.regionResolvers {
|
||
codes, handled, err := resolver.CountryCodesForRegion(ctx, s.appCode, query.RegionID)
|
||
if err != nil {
|
||
return externalDashboardCountryFilter{}, err
|
||
}
|
||
if !handled {
|
||
continue
|
||
}
|
||
return externalDashboardCountryFilter{
|
||
Applied: true,
|
||
RegionID: query.RegionID,
|
||
Codes: normalizeExternalCountryCodes(codes),
|
||
}, nil
|
||
}
|
||
// likei Mongo 集合只有国家码，没有 hyapp 的数字 region_id；缺 legacy 区域映射时失败，避免误把全量数据当成某个区域返回。
|
||
return externalDashboardCountryFilter{}, fmt.Errorf("aslan mongo dashboard source %s cannot resolve region_id %d", s.appCode, query.RegionID)
|
||
}
|
||
|
||
// legacyMongoMetric 是 likei Mongo 源按 天×国家 汇总出的事实指标；
|
||
// 消耗/产出金币是全 App 口径（按用户拆国家需要全量账变反查用户，代价过高），只挂在日行和合计上。
|
||
type legacyMongoMetric struct {
|
||
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) {
|
||
if other == nil {
|
||
return
|
||
}
|
||
m.NewUsers += other.NewUsers
|
||
m.ActiveUsers += other.ActiveUsers
|
||
m.PaidUsers += other.PaidUsers
|
||
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
|
||
m.LuckyGiftPayout += other.LuckyGiftPayout
|
||
m.LuckyGiftPayers += other.LuckyGiftPayers
|
||
m.CoinSellerTransferCoin += other.CoinSellerTransferCoin
|
||
m.SalaryUSDMinor += other.SalaryUSDMinor
|
||
m.D1Users += other.D1Users
|
||
m.D1Base += other.D1Base
|
||
m.D7Users += other.D7Users
|
||
m.D7Base += other.D7Base
|
||
m.D30Users += other.D30Users
|
||
m.D30Base += other.D30Base
|
||
}
|
||
|
||
type legacyMongoCoinFlow struct {
|
||
ConsumedCoin int64
|
||
OutputCoin int64
|
||
}
|
||
|
||
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: 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 会按拆分字段重推总充值；likei Mongo 源已经把 in-app 与 legacy 币商金额合成总额，最终以事实聚合值为准。
|
||
metric.RechargeUSDMinor = 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
|
||
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{},
|
||
rechargeUsers: map[string]map[string]struct{}{},
|
||
}
|
||
for day := startDate; day.Before(endDate); day = day.AddDate(0, 0, 1) {
|
||
text := dashboardSQLDate(day)
|
||
grid.days = append(grid.days, text)
|
||
grid.daySet[text] = struct{}{}
|
||
}
|
||
return grid
|
||
}
|
||
|
||
func (g *legacyMongoGrid) at(day string, countryCode string) *legacyMongoMetric {
|
||
g.mu.Lock()
|
||
defer g.mu.Unlock()
|
||
byCountry := g.metrics[day]
|
||
if byCountry == nil {
|
||
byCountry = map[string]*legacyMongoMetric{}
|
||
g.metrics[day] = byCountry
|
||
}
|
||
metric := byCountry[countryCode]
|
||
if metric == nil {
|
||
metric = &legacyMongoMetric{}
|
||
byCountry[countryCode] = metric
|
||
}
|
||
return metric
|
||
}
|
||
|
||
func (g *legacyMongoGrid) hasDay(day string) bool {
|
||
_, ok := g.daySet[day]
|
||
return ok
|
||
}
|
||
|
||
// runConcurrently 并发执行各 loader 并返回第一个错误；每个 loader 内部有独立的查询超时。
|
||
func runConcurrently(tasks []func() error) error {
|
||
var wg sync.WaitGroup
|
||
errs := make(chan error, len(tasks))
|
||
for _, task := range tasks {
|
||
wg.Add(1)
|
||
go func(task func() error) {
|
||
defer wg.Done()
|
||
if err := task(); err != nil {
|
||
errs <- err
|
||
}
|
||
}(task)
|
||
}
|
||
wg.Wait()
|
||
close(errs)
|
||
return <-errs
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) loadRange(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter, includeRetention bool) (aslanRangeOverview, error) {
|
||
grid := newLegacyMongoGrid(startDate, endDate)
|
||
|
||
// 注册 cohort 是新用户充值与留存的前置依赖，先串行取；其余 loader 集合互不相交，并发执行，
|
||
// 让整段耗时收敛到最慢一条聚合而不是全部相加（跨地域 Mongo 时差异是数量级的）。
|
||
cohorts, err := s.loadRegistrationCohorts(ctx, startDate, endDate, countryFilter)
|
||
if err != nil {
|
||
return aslanRangeOverview{}, err
|
||
}
|
||
for day, byCountry := range cohorts {
|
||
for countryCode, userIDs := range byCountry {
|
||
grid.at(day, countryCode).NewUsers = int64(len(userIDs))
|
||
}
|
||
}
|
||
|
||
goldWater := s.hasGoldWater(ctx)
|
||
var salarySnapshot map[string]int64
|
||
tasks := []func() error{
|
||
func() error { return s.loadActiveCounts(ctx, startDate, endDate, countryFilter, grid) },
|
||
func() error { return s.loadRechargeStats(ctx, startDate, endDate, countryFilter, cohorts, grid) },
|
||
func() error { return s.loadGiftStats(ctx, startDate, endDate, countryFilter, grid) },
|
||
func() error {
|
||
snapshot, err := s.loadHostSalaryBalanceByCountry(ctx, countryFilter)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
salarySnapshot = snapshot
|
||
return nil
|
||
},
|
||
}
|
||
if goldWater {
|
||
tasks = append(tasks,
|
||
func() error { return s.loadGoldOriginStats(ctx, startDate, endDate, countryFilter, grid) },
|
||
func() error { return s.loadCoinFlows(ctx, startDate, endDate, grid) },
|
||
)
|
||
}
|
||
if includeRetention {
|
||
tasks = append(tasks, func() error { return s.applyRetention(ctx, cohorts, grid) })
|
||
}
|
||
if err := runConcurrently(tasks); err != nil {
|
||
return aslanRangeOverview{}, err
|
||
}
|
||
// user_bank_balance 是余额快照而不是日报表；这里按当前快照填充每一天，区间汇总只保留一份快照，避免把余额按天累加。
|
||
for _, day := range grid.days {
|
||
for countryCode, balance := range salarySnapshot {
|
||
grid.at(day, countryCode).SalaryUSDMinor = balance
|
||
}
|
||
}
|
||
|
||
return s.assembleRange(grid, countryFilter, goldWater, s.legacyDB != nil), nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, countryFilter externalDashboardCountryFilter, goldWater bool, legacyRecharge bool) aslanRangeOverview {
|
||
dailySeries := []map[string]any{}
|
||
dailyCountries := []map[string]any{}
|
||
totalFlow := legacyMongoMetric{}
|
||
totalCoinFlow := legacyMongoCoinFlow{}
|
||
countryTotals := map[string]*legacyMongoMetric{}
|
||
|
||
for _, day := range grid.days {
|
||
byCountry := grid.metrics[day]
|
||
dayFlow := legacyMongoMetric{}
|
||
countryCodes := make([]string, 0, len(byCountry))
|
||
for countryCode := range byCountry {
|
||
countryCodes = append(countryCodes, countryCode)
|
||
}
|
||
sort.Strings(countryCodes)
|
||
for _, countryCode := range countryCodes {
|
||
metric := byCountry[countryCode]
|
||
dayFlow.add(metric)
|
||
countryTotal := countryTotals[countryCode]
|
||
if countryTotal == nil {
|
||
countryTotal = &legacyMongoMetric{}
|
||
countryTotals[countryCode] = countryTotal
|
||
}
|
||
salary := metric.SalaryUSDMinor
|
||
countryTotal.add(metric)
|
||
countryTotal.SalaryUSDMinor = salary
|
||
|
||
row := metric.toExternal(nil).toMap()
|
||
row["label"] = day
|
||
row["stat_day"] = day
|
||
row["country_id"] = 0
|
||
row["country_code"] = countryCode
|
||
row["country_name"] = countryCode
|
||
row["country"] = countryCode
|
||
if countryFilter.RegionID > 0 {
|
||
row["region_id"] = countryFilter.RegionID
|
||
}
|
||
applyAslanMongoUnavailableFields(row)
|
||
applyAslanMongoLegacyUnavailableFields(row, legacyRecharge)
|
||
if !goldWater {
|
||
applyAslanMongoGoldWaterUnavailable(row)
|
||
}
|
||
dailyCountries = append(dailyCountries, row)
|
||
}
|
||
coinFlow := grid.coinFlows[day]
|
||
if coinFlow != nil {
|
||
totalCoinFlow.ConsumedCoin += coinFlow.ConsumedCoin
|
||
totalCoinFlow.OutputCoin += coinFlow.OutputCoin
|
||
}
|
||
// 工资余额是快照：日行取当日各国快照之和，区间合计不能跨天累加。
|
||
daySalary := dayFlow.SalaryUSDMinor
|
||
if dayRechargeUsers := grid.rechargeUsers[day]; len(dayRechargeUsers) > 0 {
|
||
// 国家拆分仍保留各国家充值人数；日合计按 user_id 去重，避免同一用户因不同事实源国家码不一致而被算两次。
|
||
dayFlow.PaidUsers = int64(len(dayRechargeUsers))
|
||
}
|
||
totalFlow.add(&dayFlow)
|
||
totalFlow.SalaryUSDMinor = daySalary
|
||
|
||
dayRow := dayFlow.toExternal(coinFlow).toMap()
|
||
dayRow["label"] = day
|
||
dayRow["stat_day"] = day
|
||
applyAslanMongoUnavailableFields(dayRow)
|
||
applyAslanMongoLegacyUnavailableFields(dayRow, legacyRecharge)
|
||
if !goldWater {
|
||
applyAslanMongoGoldWaterUnavailable(dayRow)
|
||
}
|
||
dailySeries = append(dailySeries, dayRow)
|
||
}
|
||
|
||
countryRows := make([]map[string]any, 0, len(countryTotals))
|
||
countryCodes := make([]string, 0, len(countryTotals))
|
||
for countryCode := range countryTotals {
|
||
countryCodes = append(countryCodes, countryCode)
|
||
}
|
||
sort.Strings(countryCodes)
|
||
for _, countryCode := range countryCodes {
|
||
row := countryTotals[countryCode].toExternal(nil).toMap()
|
||
row["country_id"] = 0
|
||
row["country_code"] = countryCode
|
||
row["country_name"] = countryCode
|
||
row["country"] = countryCode
|
||
if countryFilter.RegionID > 0 {
|
||
row["region_id"] = countryFilter.RegionID
|
||
}
|
||
applyAslanMongoUnavailableFields(row)
|
||
applyAslanMongoLegacyUnavailableFields(row, legacyRecharge)
|
||
if !goldWater {
|
||
applyAslanMongoGoldWaterUnavailable(row)
|
||
}
|
||
countryRows = append(countryRows, row)
|
||
}
|
||
sort.Slice(countryRows, func(i, j int) bool {
|
||
left, _ := countryRows[i]["recharge_usd_minor"].(int64)
|
||
right, _ := countryRows[j]["recharge_usd_minor"].(int64)
|
||
if left == right {
|
||
return fmt.Sprint(countryRows[i]["country_code"]) < fmt.Sprint(countryRows[j]["country_code"])
|
||
}
|
||
return left > right
|
||
})
|
||
|
||
return aslanRangeOverview{
|
||
Total: totalFlow.toExternal(&totalCoinFlow),
|
||
DailySeries: dailySeries,
|
||
CountryBreakdown: countryRows,
|
||
DailyCountryBreakdown: dailyCountries,
|
||
}
|
||
}
|
||
|
||
// loadRegistrationCohorts 返回 天→国家→注册用户ID；同时用于新增数、新用户充值和留存 cohort。
|
||
func (s *AslanMongoDashboardSource) loadRegistrationCohorts(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter) (map[string]map[string][]int64, error) {
|
||
match := bson.M{
|
||
"originSys": s.sysOrigin,
|
||
"del": bson.M{"$ne": true},
|
||
"createTime": bson.M{"$gte": startDate, "$lt": endDate},
|
||
"countryCode": bson.M{"$exists": true, "$ne": ""},
|
||
"accountStatus": bson.M{"$ne": "DELETED"},
|
||
}
|
||
aslanMongoApplyCountryFilter(match, "countryCode", countryFilter)
|
||
pipeline := mongo.Pipeline{
|
||
{{Key: "$match", Value: match}},
|
||
{{Key: "$group", Value: bson.D{
|
||
{Key: "_id", Value: bson.D{
|
||
{Key: "day", Value: s.dayExpression("$createTime")},
|
||
{Key: "country", Value: "$countryCode"},
|
||
}},
|
||
{Key: "users", Value: bson.D{{Key: "$addToSet", Value: "$_id"}}},
|
||
}}},
|
||
}
|
||
out := map[string]map[string][]int64{}
|
||
err := s.aggregateEach(ctx, aslanMongoUserProfileCollection, pipeline, func(row bson.M) error {
|
||
day, countryCode := aslanMongoDayCountry(row["_id"])
|
||
if day == "" || countryCode == "" {
|
||
return nil
|
||
}
|
||
userIDs := aslanMongoInt64Values(row["users"])
|
||
if len(userIDs) == 0 {
|
||
return nil
|
||
}
|
||
byCountry := out[day]
|
||
if byCountry == nil {
|
||
byCountry = map[string][]int64{}
|
||
out[day] = byCountry
|
||
}
|
||
byCountry[countryCode] = userIDs
|
||
return nil
|
||
})
|
||
if err != nil {
|
||
return nil, fmt.Errorf("aggregate registration cohorts: %w", err)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) loadActiveCounts(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter, grid *legacyMongoGrid) error {
|
||
// user_daily_active_log 的 activeDate 是 yyyy-MM-dd 字符串（likei 固定按其统计时区落盘），ISO 日期字符串可直接做范围比较。
|
||
match := bson.M{
|
||
"sysOrigin": s.sysOrigin,
|
||
"activeDate": bson.M{"$gte": dashboardSQLDate(startDate), "$lt": dashboardSQLDate(endDate)},
|
||
"registerCountryCode": bson.M{"$exists": true, "$ne": ""},
|
||
}
|
||
aslanMongoApplyCountryFilter(match, "registerCountryCode", countryFilter)
|
||
pipeline := mongo.Pipeline{
|
||
{{Key: "$match", Value: match}},
|
||
{{Key: "$group", Value: bson.D{
|
||
{Key: "_id", Value: bson.D{
|
||
{Key: "day", Value: "$activeDate"},
|
||
{Key: "country", Value: "$registerCountryCode"},
|
||
}},
|
||
{Key: "users", Value: bson.D{{Key: "$addToSet", Value: "$userId"}}},
|
||
}}},
|
||
}
|
||
err := s.aggregateEach(ctx, aslanMongoUserActiveCollection, pipeline, func(row bson.M) error {
|
||
day, countryCode := aslanMongoDayCountry(row["_id"])
|
||
if day == "" || countryCode == "" || !grid.hasDay(day) {
|
||
return nil
|
||
}
|
||
grid.at(day, countryCode).ActiveUsers = aslanMongoDistinctValueCount(row["users"])
|
||
return nil
|
||
})
|
||
if err != nil {
|
||
return fmt.Errorf("aggregate active users: %w", err)
|
||
}
|
||
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{
|
||
"sysOrigin": s.sysOrigin,
|
||
"status": "SUCCESS",
|
||
"trialPeriod": bson.M{"$ne": true},
|
||
"createTime": bson.M{"$gte": startDate, "$lt": endDate},
|
||
}
|
||
aslanMongoExcludeFreightGoldRecharge(match)
|
||
// Aslan 原 Top 充值统计按 acceptUserId 聚合；createUser 只在 acceptUserId 缺失的旧单据上兜底。
|
||
userExpression := bson.D{{Key: "$ifNull", Value: bson.A{"$acceptUserId", "$createUser"}}}
|
||
pipeline := mongo.Pipeline{
|
||
{{Key: "$match", Value: match}},
|
||
}
|
||
pipeline = append(pipeline, aslanMongoRechargeCountryStages(userExpression, countryFilter)...)
|
||
pipeline = append(pipeline, bson.D{{Key: "$group", Value: bson.D{
|
||
{Key: "_id", Value: bson.D{
|
||
{Key: "day", Value: s.dayExpression("$createTime")},
|
||
{Key: "country", Value: "$aslanBillCountry"},
|
||
{Key: "factory", Value: bson.D{{Key: "$toUpper", Value: bson.D{{Key: "$ifNull", Value: bson.A{"$factory.factoryCode", ""}}}}}},
|
||
}},
|
||
{Key: "amount", Value: bson.D{{Key: "$sum", Value: "$amountUsd"}}},
|
||
{Key: "users", Value: bson.D{{Key: "$addToSet", Value: "$aslanBillUserID"}}},
|
||
}}})
|
||
distinctUsers := map[aslanMongoDayCountryUsers]map[string]struct{}{}
|
||
err := s.aggregateEach(ctx, aslanMongoInAppPurchaseCollection, pipeline, func(row bson.M) error {
|
||
key := aslanMongoDocument(row["_id"])
|
||
if key == nil {
|
||
return nil
|
||
}
|
||
day := strings.TrimSpace(fmt.Sprint(key["day"]))
|
||
countryCode := aslanMongoCountryCode(key["country"])
|
||
factory := strings.TrimSpace(fmt.Sprint(key["factory"]))
|
||
if day == "" || countryCode == "" || !grid.hasDay(day) {
|
||
return nil
|
||
}
|
||
amount, err := aslanMongoDecimalMinor(row["amount"])
|
||
if err != nil {
|
||
return fmt.Errorf("parse recharge amount for %s/%s: %w", day, countryCode, err)
|
||
}
|
||
metric := grid.at(day, countryCode)
|
||
metric.RechargeUSDMinor += amount
|
||
if factory == "GOOGLE" {
|
||
metric.GoogleUSDMinor += amount
|
||
} else {
|
||
metric.ThirdPartyUSDMinor += amount
|
||
}
|
||
usersKey := aslanMongoDayCountryUsers{day: day, country: countryCode}
|
||
userSet := distinctUsers[usersKey]
|
||
if userSet == nil {
|
||
userSet = map[string]struct{}{}
|
||
distinctUsers[usersKey] = userSet
|
||
}
|
||
for _, value := range aslanMongoInt64Values(row["users"]) {
|
||
userSet[strconv.FormatInt(value, 10)] = struct{}{}
|
||
}
|
||
return nil
|
||
})
|
||
if err != nil {
|
||
return fmt.Errorf("aggregate recharge stats: %w", err)
|
||
}
|
||
|
||
// 新用户充值：按 天×用户 聚合当日充值，再与当日注册 cohort 求交集；cohort 自带国家归属。
|
||
perUserPipeline := mongo.Pipeline{
|
||
{{Key: "$match", Value: match}},
|
||
}
|
||
perUserPipeline = append(perUserPipeline, aslanMongoRechargeCountryStages(userExpression, countryFilter)...)
|
||
perUserPipeline = append(perUserPipeline, bson.D{{Key: "$group", Value: bson.D{
|
||
{Key: "_id", Value: bson.D{
|
||
{Key: "day", Value: s.dayExpression("$createTime")},
|
||
{Key: "user", Value: "$aslanBillUserID"},
|
||
}},
|
||
{Key: "amount", Value: bson.D{{Key: "$sum", Value: "$amountUsd"}}},
|
||
}}})
|
||
perUser := map[string]map[int64]int64{}
|
||
err = s.aggregateEach(ctx, aslanMongoInAppPurchaseCollection, perUserPipeline, func(row bson.M) error {
|
||
key := aslanMongoDocument(row["_id"])
|
||
if key == nil {
|
||
return nil
|
||
}
|
||
day := strings.TrimSpace(fmt.Sprint(key["day"]))
|
||
userID := aslanMongoWholeInt64(key["user"])
|
||
if day == "" || userID <= 0 || !grid.hasDay(day) {
|
||
return nil
|
||
}
|
||
amount, err := aslanMongoDecimalMinor(row["amount"])
|
||
if err != nil {
|
||
return fmt.Errorf("parse per-user recharge amount for %s: %w", day, err)
|
||
}
|
||
byUser := perUser[day]
|
||
if byUser == nil {
|
||
byUser = map[int64]int64{}
|
||
perUser[day] = byUser
|
||
}
|
||
byUser[userID] += amount
|
||
return nil
|
||
})
|
||
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 aslanMongoRechargeCountryStages(userExpression any, countryFilter externalDashboardCountryFilter) mongo.Pipeline {
|
||
// Google 账单普遍没有 countryCode；finance 口径已按付款用户资料兜底，dashboard/social 也必须一致，
|
||
// 否则全 App、按 App 对比和国家拆分会漏掉这些真实充值。
|
||
stages := mongo.Pipeline{
|
||
bson.D{{Key: "$addFields", Value: bson.M{"aslanBillUserID": userExpression}}},
|
||
bson.D{{Key: "$lookup", Value: bson.M{
|
||
"from": aslanMongoUserProfileCollection,
|
||
"localField": "aslanBillUserID",
|
||
"foreignField": "_id",
|
||
"as": "aslanBillUser",
|
||
}}},
|
||
bson.D{{Key: "$addFields", Value: bson.M{
|
||
"aslanBillCountry": bson.M{"$toUpper": bson.M{"$cond": bson.A{
|
||
bson.M{"$ne": bson.A{bson.M{"$ifNull": bson.A{"$countryCode", ""}}, ""}},
|
||
"$countryCode",
|
||
bson.M{"$ifNull": bson.A{bson.M{"$first": "$aslanBillUser.countryCode"}, ""}},
|
||
}}},
|
||
}}},
|
||
bson.D{{Key: "$match", Value: bson.M{"aslanBillCountry": bson.M{"$ne": ""}}}},
|
||
}
|
||
if countryFilter.Applied {
|
||
codes := make([]string, 0, len(countryFilter.Codes))
|
||
for _, code := range countryFilter.Codes {
|
||
if normalized := aslanMongoCountryCode(code); normalized != "" {
|
||
codes = append(codes, normalized)
|
||
}
|
||
}
|
||
if len(codes) == 0 {
|
||
codes = append(codes, "__NO_COUNTRY__")
|
||
}
|
||
stages = append(stages, bson.D{{Key: "$match", Value: bson.M{"aslanBillCountry": bson.M{"$in": codes}}}})
|
||
}
|
||
return stages
|
||
}
|
||
|
||
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 {
|
||
continue
|
||
}
|
||
for countryCode, userIDs := range byCountry {
|
||
total := int64(0)
|
||
for _, userID := range userIDs {
|
||
total += byUser[userID]
|
||
}
|
||
if total != 0 {
|
||
grid.at(day, countryCode).NewUserRechargeUSDMinor = total
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
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.loadLegacyFreightStockRechargeStats(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) loadLegacyFreightStockRechargeStats(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；
|
||
// “扣减”是库存/USDT 冲减：origin=DEDUCTION/type=1，amount 在源系统仍是正数，统计时必须转成负数。
|
||
// App/经销商路径可能只落 usd_quantity，因此金额优先取 amount，空值再回退 usd_quantity。
|
||
where := []string{
|
||
"fbrw.sys_origin = ?",
|
||
"((fbrw.origin = 'PURCHASE' AND fbrw.type = 0) OR (fbrw.origin = 'DEDUCTION' AND fbrw.type = 1))",
|
||
"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.origin = 'DEDUCTION' OR fbrw.type = 1 THEN -ABS(CASE
|
||
WHEN fbrw.amount IS NOT NULL AND fbrw.amount <> 0 THEN fbrw.amount
|
||
ELSE COALESCE(fbrw.usd_quantity, 0)
|
||
END)
|
||
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 stock 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 stock 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 stock 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 stock 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
|
||
// 扣减流水按负数冲减金额，但不是一次新的付费行为，不能增加付费用户或新用户充值。
|
||
if amount > 0 {
|
||
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 {
|
||
match := bson.M{
|
||
"sysOrigin": s.sysOrigin,
|
||
"createTime": bson.M{"$gte": startDate, "$lt": endDate},
|
||
"refunded": bson.M{"$ne": true},
|
||
}
|
||
giftValueExpression := bson.D{{Key: "$ifNull", Value: bson.A{"$giftValue.actualAmount", bson.D{{Key: "$ifNull", Value: bson.A{"$giftValue.giftValue", 0}}}}}}
|
||
pipeline := mongo.Pipeline{
|
||
{{Key: "$match", Value: match}},
|
||
{{Key: "$group", Value: bson.D{
|
||
{Key: "_id", Value: bson.D{
|
||
{Key: "day", Value: s.dayExpression("$createTime")},
|
||
{Key: "user", Value: "$userId"},
|
||
}},
|
||
{Key: "total", Value: bson.D{{Key: "$sum", Value: giftValueExpression}}},
|
||
{Key: "lucky", Value: bson.D{{Key: "$sum", Value: bson.D{{Key: "$cond", Value: bson.A{
|
||
bson.D{{Key: "$eq", Value: bson.A{"$luckyGift", true}}},
|
||
giftValueExpression,
|
||
0,
|
||
}}}}}},
|
||
}}},
|
||
}
|
||
type giftEntry struct {
|
||
day string
|
||
userID int64
|
||
total int64
|
||
lucky int64
|
||
}
|
||
entries := []giftEntry{}
|
||
err := s.aggregateEach(ctx, aslanMongoGiftWaterCollection, pipeline, func(row bson.M) error {
|
||
key := aslanMongoDocument(row["_id"])
|
||
if key == nil {
|
||
return nil
|
||
}
|
||
day := strings.TrimSpace(fmt.Sprint(key["day"]))
|
||
userID := aslanMongoWholeInt64(key["user"])
|
||
if day == "" || userID <= 0 || !grid.hasDay(day) {
|
||
return nil
|
||
}
|
||
total, err := aslanMongoDecimalMinor(row["total"])
|
||
if err != nil {
|
||
return fmt.Errorf("parse gift total for %s: %w", day, err)
|
||
}
|
||
lucky, err := aslanMongoDecimalMinor(row["lucky"])
|
||
if err != nil {
|
||
return fmt.Errorf("parse lucky gift total for %s: %w", day, err)
|
||
}
|
||
// 礼物价值以金币计，amount 十进制只是 Decimal128 载体，这里换回整数金币。
|
||
entries = append(entries, giftEntry{day: day, userID: userID, total: roundMinorToWhole(total), lucky: roundMinorToWhole(lucky)})
|
||
return nil
|
||
})
|
||
if err != nil {
|
||
return fmt.Errorf("aggregate gift stats: %w", err)
|
||
}
|
||
if len(entries) == 0 {
|
||
return nil
|
||
}
|
||
userIDs := make([]int64, 0, len(entries))
|
||
seen := map[int64]struct{}{}
|
||
for _, entry := range entries {
|
||
if _, ok := seen[entry.userID]; ok {
|
||
continue
|
||
}
|
||
seen[entry.userID] = struct{}{}
|
||
userIDs = append(userIDs, entry.userID)
|
||
}
|
||
countryByUser, err := s.loadUserCountries(ctx, userIDs, countryFilter)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for _, entry := range entries {
|
||
countryCode := countryByUser[entry.userID]
|
||
if countryCode == "" {
|
||
continue
|
||
}
|
||
metric := grid.at(entry.day, countryCode)
|
||
metric.GiftCoinSpent += entry.total
|
||
metric.LuckyGiftTurnover += entry.lucky
|
||
if entry.lucky > 0 {
|
||
metric.LuckyGiftPayers++
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// loadGoldOriginStats 聚合金币流水里两类来源：幸运礼物返奖（LUCKY_GIFT_GOLD_REWARD*）与币商出货（SELLER_AGENT*）。
|
||
func (s *AslanMongoDashboardSource) loadGoldOriginStats(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter, grid *legacyMongoGrid) error {
|
||
assigners := []struct {
|
||
originPrefix string
|
||
assign func(metric *legacyMongoMetric, amount int64)
|
||
}{
|
||
{aslanMongoLuckyGiftPayoutOriginPrefix, func(metric *legacyMongoMetric, amount int64) { metric.LuckyGiftPayout += amount }},
|
||
{aslanMongoSellerAgentOriginPrefix, func(metric *legacyMongoMetric, amount int64) { metric.CoinSellerTransferCoin += amount }},
|
||
}
|
||
for _, item := range assigners {
|
||
match := bson.M{
|
||
"sysOrigin": s.sysOrigin,
|
||
"createTime": bson.M{"$gte": startDate, "$lt": endDate},
|
||
"type": 0,
|
||
"origin": bson.M{"$regex": "^" + item.originPrefix},
|
||
}
|
||
pipeline := mongo.Pipeline{
|
||
{{Key: "$match", Value: match}},
|
||
{{Key: "$group", Value: bson.D{
|
||
{Key: "_id", Value: bson.D{
|
||
{Key: "day", Value: s.dayExpression("$createTime")},
|
||
{Key: "user", Value: "$userId"},
|
||
}},
|
||
{Key: "amount", Value: bson.D{{Key: "$sum", Value: "$quantity"}}},
|
||
}}},
|
||
}
|
||
type goldEntry struct {
|
||
day string
|
||
userID int64
|
||
amount int64
|
||
}
|
||
entries := []goldEntry{}
|
||
err := s.aggregateEach(ctx, aslanMongoGoldWaterCollection, pipeline, func(row bson.M) error {
|
||
key := aslanMongoDocument(row["_id"])
|
||
if key == nil {
|
||
return nil
|
||
}
|
||
day := strings.TrimSpace(fmt.Sprint(key["day"]))
|
||
userID := aslanMongoWholeInt64(key["user"])
|
||
if day == "" || userID <= 0 || !grid.hasDay(day) {
|
||
return nil
|
||
}
|
||
amount, err := aslanMongoDecimalMinor(row["amount"])
|
||
if err != nil {
|
||
return fmt.Errorf("parse gold origin %s amount for %s: %w", item.originPrefix, day, err)
|
||
}
|
||
entries = append(entries, goldEntry{day: day, userID: userID, amount: roundMinorToWhole(amount)})
|
||
return nil
|
||
})
|
||
if err != nil {
|
||
return fmt.Errorf("aggregate gold origin %s: %w", item.originPrefix, err)
|
||
}
|
||
if len(entries) == 0 {
|
||
continue
|
||
}
|
||
userIDs := make([]int64, 0, len(entries))
|
||
seen := map[int64]struct{}{}
|
||
for _, entry := range entries {
|
||
if _, ok := seen[entry.userID]; ok {
|
||
continue
|
||
}
|
||
seen[entry.userID] = struct{}{}
|
||
userIDs = append(userIDs, entry.userID)
|
||
}
|
||
countryByUser, err := s.loadUserCountries(ctx, userIDs, countryFilter)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for _, entry := range entries {
|
||
countryCode := countryByUser[entry.userID]
|
||
if countryCode == "" {
|
||
continue
|
||
}
|
||
item.assign(grid.at(entry.day, countryCode), entry.amount)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// loadCoinFlows 汇总每日全 App 消耗/产出金币；type=1 是消耗、type=0 是收入（产出）。
|
||
// 区域筛选时无法低成本把全量账变拆到国家，保持整段 nil 交由上层展示 "--"。
|
||
func (s *AslanMongoDashboardSource) loadCoinFlows(ctx context.Context, startDate time.Time, endDate time.Time, grid *legacyMongoGrid) error {
|
||
pipeline := mongo.Pipeline{
|
||
{{Key: "$match", Value: bson.M{
|
||
"sysOrigin": s.sysOrigin,
|
||
"createTime": bson.M{"$gte": startDate, "$lt": endDate},
|
||
}}},
|
||
{{Key: "$group", Value: bson.D{
|
||
{Key: "_id", Value: bson.D{
|
||
{Key: "day", Value: s.dayExpression("$createTime")},
|
||
{Key: "type", Value: "$type"},
|
||
}},
|
||
{Key: "amount", Value: bson.D{{Key: "$sum", Value: "$quantity"}}},
|
||
}}},
|
||
}
|
||
err := s.aggregateEach(ctx, aslanMongoGoldWaterCollection, pipeline, func(row bson.M) error {
|
||
key := aslanMongoDocument(row["_id"])
|
||
if key == nil {
|
||
return nil
|
||
}
|
||
day := strings.TrimSpace(fmt.Sprint(key["day"]))
|
||
if day == "" || !grid.hasDay(day) {
|
||
return nil
|
||
}
|
||
amount, err := aslanMongoDecimalMinor(row["amount"])
|
||
if err != nil {
|
||
return fmt.Errorf("parse coin flow amount for %s: %w", day, err)
|
||
}
|
||
flow := grid.coinFlows[day]
|
||
if flow == nil {
|
||
flow = &legacyMongoCoinFlow{}
|
||
grid.coinFlows[day] = flow
|
||
}
|
||
if aslanMongoWholeInt64(key["type"]) == 1 {
|
||
flow.ConsumedCoin += roundMinorToWhole(amount)
|
||
} else {
|
||
flow.OutputCoin += roundMinorToWhole(amount)
|
||
}
|
||
return nil
|
||
})
|
||
if err != nil {
|
||
return fmt.Errorf("aggregate coin flows: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// applyRetention 用注册 cohort × user_daily_active_log 推 D1/D7/D30：目标观察日未到时基数保持 0，
|
||
// 上层会把 0 基数转成 nil，避免把“还不能观察”展示成真实 0% 留存。
|
||
func (s *AslanMongoDashboardSource) applyRetention(ctx context.Context, cohorts map[string]map[string][]int64, grid *legacyMongoGrid) error {
|
||
location, err := time.LoadLocation(s.statTimezone)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
today := dashboardSQLDate(time.Now().In(location))
|
||
offsets := []struct {
|
||
days int
|
||
assign func(metric *legacyMongoMetric, users int64, base int64)
|
||
}{
|
||
{1, func(metric *legacyMongoMetric, users, base int64) { metric.D1Users, metric.D1Base = users, base }},
|
||
{7, func(metric *legacyMongoMetric, users, base int64) { metric.D7Users, metric.D7Base = users, base }},
|
||
{30, func(metric *legacyMongoMetric, users, base int64) { metric.D30Users, metric.D30Base = users, base }},
|
||
}
|
||
for day, byCountry := range cohorts {
|
||
if !grid.hasDay(day) {
|
||
continue
|
||
}
|
||
dayTime, err := time.ParseInLocation("2006-01-02", day, location)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
allIDs := []int64{}
|
||
for _, userIDs := range byCountry {
|
||
allIDs = append(allIDs, userIDs...)
|
||
}
|
||
if len(allIDs) == 0 {
|
||
continue
|
||
}
|
||
for _, offset := range offsets {
|
||
target := dashboardSQLDate(dayTime.AddDate(0, 0, offset.days))
|
||
if target > today {
|
||
continue
|
||
}
|
||
activeSet, err := s.loadActiveUserSet(ctx, target, allIDs)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for countryCode, userIDs := range byCountry {
|
||
active := int64(0)
|
||
for _, userID := range userIDs {
|
||
if _, ok := activeSet[userID]; ok {
|
||
active++
|
||
}
|
||
}
|
||
offset.assign(grid.at(day, countryCode), active, int64(len(userIDs)))
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) loadActiveUserSet(ctx context.Context, activeDate string, userIDs []int64) (map[int64]struct{}, error) {
|
||
out := map[int64]struct{}{}
|
||
for _, chunk := range aslanMongoInt64Chunks(userIDs, aslanMongoChunkSize) {
|
||
queryCtx, cancel := context.WithTimeout(ctx, s.requestTimeout)
|
||
cursor, err := s.db.Collection(aslanMongoUserActiveCollection).Find(queryCtx, bson.M{
|
||
"sysOrigin": s.sysOrigin,
|
||
"activeDate": activeDate,
|
||
"userId": bson.M{"$in": chunk},
|
||
}, options.Find().SetProjection(bson.M{"userId": 1}))
|
||
if err != nil {
|
||
cancel()
|
||
return nil, fmt.Errorf("query retention actives for %s: %w", activeDate, err)
|
||
}
|
||
for cursor.Next(queryCtx) {
|
||
row := bson.M{}
|
||
if err := cursor.Decode(&row); err != nil {
|
||
_ = cursor.Close(queryCtx)
|
||
cancel()
|
||
return nil, fmt.Errorf("decode retention active row for %s: %w", activeDate, err)
|
||
}
|
||
if userID := aslanMongoWholeInt64(row["userId"]); userID > 0 {
|
||
out[userID] = struct{}{}
|
||
}
|
||
}
|
||
err = cursor.Err()
|
||
_ = cursor.Close(queryCtx)
|
||
cancel()
|
||
if err != nil {
|
||
return nil, fmt.Errorf("iterate retention actives for %s: %w", activeDate, err)
|
||
}
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) loadHostSalaryBalanceByCountry(ctx context.Context, countryFilter externalDashboardCountryFilter) (map[string]int64, error) {
|
||
memberIDs, err := s.loadTeamMemberIDs(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(memberIDs) == 0 {
|
||
return map[string]int64{}, nil
|
||
}
|
||
countryByUser, err := s.loadUserCountries(ctx, memberIDs, countryFilter)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(countryByUser) == 0 {
|
||
return map[string]int64{}, nil
|
||
}
|
||
|
||
out := map[string]int64{}
|
||
for _, chunk := range aslanMongoInt64Chunks(memberIDs, aslanMongoChunkSize) {
|
||
queryCtx, cancel := context.WithTimeout(ctx, s.requestTimeout)
|
||
cursor, err := s.db.Collection(aslanMongoBankBalanceCollection).Find(queryCtx, bson.M{
|
||
"sysOrigin": s.sysOrigin,
|
||
"_id": bson.M{"$in": chunk},
|
||
}, options.Find().SetProjection(bson.M{"_id": 1, "balance": 1}))
|
||
if err != nil {
|
||
cancel()
|
||
return nil, fmt.Errorf("query user_bank_balance: %w", err)
|
||
}
|
||
for cursor.Next(queryCtx) {
|
||
row := bson.M{}
|
||
if err := cursor.Decode(&row); err != nil {
|
||
_ = cursor.Close(queryCtx)
|
||
cancel()
|
||
return nil, fmt.Errorf("decode user_bank_balance: %w", err)
|
||
}
|
||
userID := aslanMongoWholeInt64(row["_id"])
|
||
countryCode := countryByUser[userID]
|
||
if countryCode == "" {
|
||
continue
|
||
}
|
||
// user_bank_balance.balance 是 PennyAmount 的 long 值；admin 响应也用 USD minor，所以不用再做美元小数换算。
|
||
out[countryCode] += aslanMongoWholeInt64(row["balance"])
|
||
}
|
||
err = cursor.Err()
|
||
_ = cursor.Close(queryCtx)
|
||
cancel()
|
||
if err != nil {
|
||
return nil, fmt.Errorf("iterate user_bank_balance: %w", err)
|
||
}
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) loadTeamMemberIDs(ctx context.Context) ([]int64, error) {
|
||
queryCtx, cancel := context.WithTimeout(ctx, s.requestTimeout)
|
||
defer cancel()
|
||
cursor, err := s.db.Collection(aslanMongoTeamMemberCollection).Find(queryCtx, bson.M{
|
||
"sysOrigin": s.sysOrigin,
|
||
"role": bson.M{"$in": bson.A{"MEMBER", "ADMIN"}},
|
||
}, options.Find().SetProjection(bson.M{"memberId": 1}))
|
||
if err != nil {
|
||
return nil, fmt.Errorf("query team_member: %w", err)
|
||
}
|
||
defer cursor.Close(queryCtx)
|
||
|
||
seen := map[int64]struct{}{}
|
||
out := []int64{}
|
||
for cursor.Next(queryCtx) {
|
||
row := bson.M{}
|
||
if err := cursor.Decode(&row); err != nil {
|
||
return nil, fmt.Errorf("decode team_member: %w", err)
|
||
}
|
||
userID := aslanMongoWholeInt64(row["memberId"])
|
||
if userID <= 0 {
|
||
continue
|
||
}
|
||
if _, ok := seen[userID]; ok {
|
||
continue
|
||
}
|
||
seen[userID] = struct{}{}
|
||
out = append(out, userID)
|
||
}
|
||
if err := cursor.Err(); err != nil {
|
||
return nil, fmt.Errorf("iterate team_member: %w", err)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) loadUserCountries(ctx context.Context, userIDs []int64, countryFilter externalDashboardCountryFilter) (map[int64]string, error) {
|
||
out := map[int64]string{}
|
||
for _, chunk := range aslanMongoInt64Chunks(userIDs, aslanMongoChunkSize) {
|
||
filter := bson.M{
|
||
"_id": bson.M{"$in": chunk},
|
||
"originSys": s.sysOrigin,
|
||
"del": bson.M{"$ne": true},
|
||
"countryCode": bson.M{"$exists": true, "$ne": ""},
|
||
}
|
||
aslanMongoApplyCountryFilter(filter, "countryCode", countryFilter)
|
||
queryCtx, cancel := context.WithTimeout(ctx, s.requestTimeout)
|
||
cursor, err := s.db.Collection(aslanMongoUserProfileCollection).Find(queryCtx, filter, options.Find().SetProjection(bson.M{"_id": 1, "countryCode": 1}))
|
||
if err != nil {
|
||
cancel()
|
||
return nil, fmt.Errorf("query user_run_profile countries: %w", err)
|
||
}
|
||
for cursor.Next(queryCtx) {
|
||
row := bson.M{}
|
||
if err := cursor.Decode(&row); err != nil {
|
||
_ = cursor.Close(queryCtx)
|
||
cancel()
|
||
return nil, fmt.Errorf("decode user_run_profile country: %w", err)
|
||
}
|
||
userID := aslanMongoWholeInt64(row["_id"])
|
||
countryCode := aslanMongoCountryCode(row["countryCode"])
|
||
if userID > 0 && countryCode != "" {
|
||
out[userID] = countryCode
|
||
}
|
||
}
|
||
err = cursor.Err()
|
||
_ = cursor.Close(queryCtx)
|
||
cancel()
|
||
if err != nil {
|
||
return nil, fmt.Errorf("iterate user_run_profile countries: %w", err)
|
||
}
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) aggregateEach(ctx context.Context, collectionName string, pipeline mongo.Pipeline, handle func(bson.M) error) error {
|
||
queryCtx, cancel := context.WithTimeout(ctx, s.requestTimeout)
|
||
defer cancel()
|
||
cursor, err := s.db.Collection(collectionName).Aggregate(queryCtx, pipeline, options.Aggregate().SetAllowDiskUse(true))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer cursor.Close(queryCtx)
|
||
for cursor.Next(queryCtx) {
|
||
row := bson.M{}
|
||
if err := cursor.Decode(&row); err != nil {
|
||
return err
|
||
}
|
||
if err := handle(row); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return cursor.Err()
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) dayExpression(field string) bson.D {
|
||
return bson.D{{Key: "$dateToString", Value: bson.D{
|
||
{Key: "format", Value: "%Y-%m-%d"},
|
||
{Key: "date", Value: field},
|
||
{Key: "timezone", Value: s.statTimezone},
|
||
}}}
|
||
}
|
||
|
||
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 {
|
||
switch typed := value.(type) {
|
||
case bson.M:
|
||
return typed
|
||
case bson.D:
|
||
out := make(bson.M, len(typed))
|
||
for _, element := range typed {
|
||
out[element.Key] = element.Value
|
||
}
|
||
return out
|
||
case map[string]any:
|
||
return bson.M(typed)
|
||
default:
|
||
return nil
|
||
}
|
||
}
|
||
|
||
func aslanMongoDayCountry(value any) (string, string) {
|
||
key := aslanMongoDocument(value)
|
||
if key == nil {
|
||
return "", ""
|
||
}
|
||
day := strings.TrimSpace(fmt.Sprint(key["day"]))
|
||
if day == "" || day == "<nil>" {
|
||
return "", ""
|
||
}
|
||
return day, aslanMongoCountryCode(key["country"])
|
||
}
|
||
|
||
func aslanMongoInt64Values(value any) []int64 {
|
||
values, ok := value.(bson.A)
|
||
if !ok {
|
||
if raw, ok := value.([]any); ok {
|
||
values = raw
|
||
}
|
||
}
|
||
out := make([]int64, 0, len(values))
|
||
seen := map[int64]struct{}{}
|
||
for _, item := range values {
|
||
parsed := aslanMongoWholeInt64(item)
|
||
if parsed <= 0 {
|
||
continue
|
||
}
|
||
if _, ok := seen[parsed]; ok {
|
||
continue
|
||
}
|
||
seen[parsed] = struct{}{}
|
||
out = append(out, parsed)
|
||
}
|
||
return out
|
||
}
|
||
|
||
func aslanMongoApplyCountryFilter(match bson.M, field string, countryFilter externalDashboardCountryFilter) {
|
||
if !countryFilter.Applied {
|
||
return
|
||
}
|
||
if len(countryFilter.Codes) == 0 {
|
||
match[field] = bson.M{"$in": bson.A{"__NO_COUNTRY__"}}
|
||
return
|
||
}
|
||
codes := bson.A{}
|
||
for _, code := range countryFilter.Codes {
|
||
if normalized := aslanMongoCountryCode(code); normalized != "" {
|
||
codes = append(codes, normalized)
|
||
}
|
||
}
|
||
if len(codes) == 0 {
|
||
codes = bson.A{"__NO_COUNTRY__"}
|
||
}
|
||
match[field] = bson.M{"$in": codes}
|
||
}
|
||
|
||
func aslanMongoCountryCode(value any) string {
|
||
code := strings.ToUpper(strings.TrimSpace(fmt.Sprint(value)))
|
||
if code == "" || code == "<NIL>" || code == "NULL" {
|
||
return ""
|
||
}
|
||
return code
|
||
}
|
||
|
||
func quoteMySQLIdentifier(value string) string {
|
||
return "`" + strings.ReplaceAll(value, "`", "``") + "`"
|
||
}
|
||
|
||
func aslanMongoDistinctValueCount(value any) int64 {
|
||
values, ok := value.(bson.A)
|
||
if !ok {
|
||
if raw, ok := value.([]any); ok {
|
||
values = raw
|
||
}
|
||
}
|
||
seen := map[string]struct{}{}
|
||
for _, item := range values {
|
||
key := strings.TrimSpace(fmt.Sprint(item))
|
||
if key == "" || key == "<nil>" || key == "0" {
|
||
continue
|
||
}
|
||
seen[key] = struct{}{}
|
||
}
|
||
return int64(len(seen))
|
||
}
|
||
|
||
func aslanMongoDecimalMinor(value any) (int64, error) {
|
||
switch typed := value.(type) {
|
||
case nil:
|
||
return 0, nil
|
||
case bson.Decimal128:
|
||
return decimalTextToMinor(typed.String())
|
||
case string:
|
||
return decimalTextToMinor(typed)
|
||
case int32:
|
||
return int64(typed) * 100, nil
|
||
case int64:
|
||
return typed * 100, nil
|
||
case int:
|
||
return int64(typed) * 100, nil
|
||
case float32:
|
||
return decimalTextToMinor(strconv.FormatFloat(float64(typed), 'f', -1, 32))
|
||
case float64:
|
||
return decimalTextToMinor(strconv.FormatFloat(typed, 'f', -1, 64))
|
||
default:
|
||
return decimalTextToMinor(fmt.Sprint(value))
|
||
}
|
||
}
|
||
|
||
func aslanMongoWholeInt64(value any) int64 {
|
||
switch typed := value.(type) {
|
||
case nil:
|
||
return 0
|
||
case int64:
|
||
return typed
|
||
case int32:
|
||
return int64(typed)
|
||
case int:
|
||
return int64(typed)
|
||
case float64:
|
||
return int64(typed)
|
||
case string:
|
||
parsed, _ := strconv.ParseInt(strings.TrimSpace(typed), 10, 64)
|
||
return parsed
|
||
case bson.Decimal128:
|
||
parsed, _ := strconv.ParseInt(strings.TrimSpace(typed.String()), 10, 64)
|
||
return parsed
|
||
default:
|
||
parsed, _ := strconv.ParseInt(strings.TrimSpace(fmt.Sprint(value)), 10, 64)
|
||
return parsed
|
||
}
|
||
}
|
||
|
||
func aslanMongoInt64Chunks(values []int64, size int) [][]int64 {
|
||
if size <= 0 {
|
||
size = aslanMongoChunkSize
|
||
}
|
||
out := [][]int64{}
|
||
for start := 0; start < len(values); start += size {
|
||
end := start + size
|
||
if end > len(values) {
|
||
end = len(values)
|
||
}
|
||
out = append(out, values[start:end])
|
||
}
|
||
return out
|
||
}
|
||
|
||
// aslanMongoUnavailableMetricFields 是 likei Mongo 事实集合仍给不出的口径；
|
||
// 游戏流水的 GoldOrigin code 枚举过杂（大量带后缀的游戏 code 与奖励类混在一起），未确认前不硬凑。
|
||
var aslanMongoUnavailableMetricFields = []string{
|
||
"coin_seller_stock_coin",
|
||
"new_dealer_recharge_usd_minor",
|
||
"game_turnover",
|
||
"game_payout",
|
||
"game_players",
|
||
"game_profit",
|
||
"game_profit_rate",
|
||
"game_payout_delta_rate",
|
||
"game_turnover_delta_rate",
|
||
"game_profit_delta_rate",
|
||
"super_lucky_gift_turnover",
|
||
"super_lucky_gift_payout",
|
||
"super_lucky_gift_profit",
|
||
"super_lucky_gift_profit_rate",
|
||
"coin_total",
|
||
"coin_total_delta_rate",
|
||
"platform_grant_coin",
|
||
"manual_grant_coin",
|
||
"salary_exchange_usd_minor",
|
||
"salary_transfer_usd_minor",
|
||
"salary_transfer_coin",
|
||
"mic_online_ms",
|
||
"avg_mic_online_ms",
|
||
}
|
||
|
||
func applyAslanMongoUnavailableFields(row map[string]any) {
|
||
for _, field := range aslanMongoUnavailableMetricFields {
|
||
row[field] = nil
|
||
}
|
||
}
|
||
|
||
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",
|
||
"lucky_gift_payout_rate",
|
||
"lucky_gift_profit",
|
||
"lucky_gift_profit_rate",
|
||
"lucky_gift_anchor_share",
|
||
"coin_seller_transfer_coin",
|
||
"consumed_coin",
|
||
"output_coin",
|
||
"consume_output_ratio",
|
||
"consume_output_delta",
|
||
}
|
||
|
||
func applyAslanMongoGoldWaterUnavailable(row map[string]any) {
|
||
for _, field := range aslanMongoGoldWaterMetricFields {
|
||
row[field] = nil
|
||
}
|
||
}
|
||
|
||
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*"},
|
||
{"field": "consumed_coin", "available": true, "source": "likei Mongo user_gold_running_water_v2 type=1（仅全 App 口径，不拆国家）"},
|
||
{"field": "output_coin", "available": true, "source": "likei Mongo user_gold_running_water_v2 type=0（仅全 App 口径，不拆国家）"},
|
||
{"field": "consume_output_ratio", "available": true, "source": "consumed_coin / output_coin（仅全 App 口径）"},
|
||
}
|
||
if !goldWater {
|
||
goldWaterSources = []map[string]any{
|
||
{"field": "lucky_gift_payout", "available": false, "missing_reason": "该部署的 Mongo 没有 user_gold_running_water_v2 集合"},
|
||
{"field": "coin_seller_transfer_coin", "available": false, "missing_reason": "该部署的 Mongo 没有 user_gold_running_water_v2 集合"},
|
||
{"field": "consumed_coin", "available": false, "missing_reason": "该部署的 Mongo 没有 user_gold_running_water_v2 集合"},
|
||
{"field": "output_coin", "available": false, "missing_reason": "该部署的 Mongo 没有 user_gold_running_water_v2 集合"},
|
||
{"field": "consume_output_ratio", "available": false, "missing_reason": "该部署的 Mongo 没有 user_gold_running_water_v2 集合"},
|
||
}
|
||
}
|
||
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 - DEDUCTION amount/usd_quantity"},
|
||
}
|
||
rechargeUsersSource := "likei Mongo in_app_purchase_details SUCCESS distinct acceptUserId（排除 FREIGHT_GOLD，国家码按账单 countryCode 优先、user_run_profile.countryCode 兜底）"
|
||
rechargeAmountSource := "likei Mongo in_app_purchase_details amountUsd（排除 FREIGHT_GOLD，国家码按账单 countryCode 优先、user_run_profile.countryCode 兜底）"
|
||
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 - DEDUCTION 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": 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": 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_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 平台没有超级幸运礼物玩法"},
|
||
{"field": "coin_total", "available": false, "missing_reason": "likei Mongo 没有可按国家低成本汇总的金币余额快照"},
|
||
{"field": "platform_grant_coin", "available": false, "missing_reason": "likei Mongo 未确认平台发放金币来源枚举"},
|
||
{"field": "manual_grant_coin", "available": false, "missing_reason": "likei Mongo 未确认人工发放金币来源枚举"},
|
||
{"field": "salary_usd_minor", "available": true, "source": "likei Mongo team_member + user_bank_balance + user_run_profile 当前快照"},
|
||
{"field": "mic_online_ms", "available": false, "missing_reason": "likei Mongo 未确认麦上时长集合"},
|
||
{"field": "arpu_usd_minor", "available": true, "source": "recharge_usd_minor / active_users"},
|
||
{"field": "arppu_usd_minor", "available": true, "source": "recharge_usd_minor / paid_users"},
|
||
{"field": "paid_conversion_rate", "available": true, "source": "paid_users / active_users"},
|
||
{"field": "recharge_conversion_rate", "available": true, "source": "recharge_users / active_users"},
|
||
{"field": "retention.day1_rate", "available": 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"},
|
||
}
|
||
out := append(baseSources, legacySources...)
|
||
return append(out, goldWaterSources...)
|
||
}
|