691 lines
25 KiB
Go
691 lines
25 KiB
Go
package dashboard
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"hyapp-admin-server/internal/config"
|
||
|
||
"go.mongodb.org/mongo-driver/v2/bson"
|
||
"go.mongodb.org/mongo-driver/v2/mongo"
|
||
"go.mongodb.org/mongo-driver/v2/mongo/options"
|
||
)
|
||
|
||
const (
|
||
aslanMongoInAppPurchaseCollection = "in_app_purchase_details"
|
||
aslanMongoUserProfileCollection = "user_run_profile"
|
||
aslanMongoUserActiveCollection = "user_daily_active_log"
|
||
aslanMongoTeamMemberCollection = "team_member"
|
||
aslanMongoBankBalanceCollection = "user_bank_balance"
|
||
aslanMongoChunkSize = 800
|
||
)
|
||
|
||
type AslanMongoDashboardSource struct {
|
||
db *mongo.Database
|
||
appCode string
|
||
appName string
|
||
sysOrigin string
|
||
statTimezone string
|
||
requestTimeout time.Duration
|
||
regionResolvers []ExternalDashboardRegionResolver
|
||
}
|
||
|
||
type aslanMongoRechargeMetric struct {
|
||
AmountUSDMinor int64
|
||
Users int64
|
||
}
|
||
|
||
// NewAslanMongoDashboardSource 直接读取 Aslan 外接项目 Mongo 里的已存在事实集合。
|
||
// 这里只做查询聚合和字段映射，不把 Aslan 数据复制到 hyapp statistics-service，也不回退旧 console HTTP 接口。
|
||
func NewAslanMongoDashboardSource(database *mongo.Database, sourceConfig config.DashboardExternalSourceConfig, regionResolvers ...ExternalDashboardRegionResolver) (ExternalDashboardSource, error) {
|
||
if database == nil || !sourceConfig.Enabled {
|
||
return nil, nil
|
||
}
|
||
appCode := strings.ToLower(strings.TrimSpace(sourceConfig.AppCode))
|
||
sysOrigin := strings.ToUpper(strings.TrimSpace(sourceConfig.SysOrigin))
|
||
if appCode == "" || sysOrigin == "" {
|
||
return nil, nil
|
||
}
|
||
statTimezone := strings.TrimSpace(sourceConfig.StatTimezone)
|
||
if statTimezone == "" {
|
||
statTimezone = "Asia/Shanghai"
|
||
}
|
||
if _, err := time.LoadLocation(statTimezone); err != nil {
|
||
return nil, fmt.Errorf("load aslan mongo storage timezone %s: %w", statTimezone, err)
|
||
}
|
||
timeout := sourceConfig.RequestTimeout
|
||
if timeout <= 0 {
|
||
timeout = 5 * time.Second
|
||
}
|
||
return &AslanMongoDashboardSource{
|
||
db: database,
|
||
appCode: appCode,
|
||
appName: strings.TrimSpace(sourceConfig.AppName),
|
||
sysOrigin: sysOrigin,
|
||
statTimezone: statTimezone,
|
||
requestTimeout: timeout,
|
||
regionResolvers: compactExternalDashboardRegionResolvers(regionResolvers),
|
||
}, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) AppCode() string {
|
||
if s == nil {
|
||
return ""
|
||
}
|
||
return s.appCode
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) StatisticsOverview(ctx context.Context, query StatisticsQuery) (map[string]any, error) {
|
||
if s == nil || s.db == nil {
|
||
return nil, fmt.Errorf("aslan mongo dashboard source is not configured")
|
||
}
|
||
countryFilter, err := s.countryFilter(ctx, query)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
requestTimezone := strings.TrimSpace(query.StatTZ)
|
||
if requestTimezone == "" {
|
||
requestTimezone = s.statTimezone
|
||
}
|
||
requestLocation, err := time.LoadLocation(requestTimezone)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("load aslan mongo request timezone %s: %w", requestTimezone, err)
|
||
}
|
||
startDate, endDate, startMS, endMS := externalDashboardDateRange(query, requestLocation)
|
||
|
||
queryCtx, cancel := context.WithTimeout(ctx, s.requestTimeout)
|
||
defer cancel()
|
||
|
||
current, err := s.loadRange(queryCtx, startDate, endDate, countryFilter)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
days := countCalendarDays(startDate, endDate)
|
||
previousStart := startDate.AddDate(0, 0, -days)
|
||
previous, err := s.loadRange(queryCtx, previousStart, startDate, countryFilter)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
response := current.Total.toMap()
|
||
response["app_code"] = s.appCode
|
||
response["app_name"] = firstExternalDashboardValue(s.appName, s.appCode)
|
||
response["sys_origin"] = s.sysOrigin
|
||
response["stat_tz"] = s.statTimezone
|
||
response["request_stat_tz"] = requestTimezone
|
||
response["start_ms"] = startMS
|
||
response["end_ms"] = endMS
|
||
response["daily_series"] = current.DailySeries
|
||
response["country_breakdown"] = current.CountryBreakdown
|
||
response["daily_country_breakdown"] = current.DailyCountryBreakdown
|
||
response["retention"] = aslanRetentionMap(current.Total)
|
||
response["report_metric_sources"] = aslanMongoDashboardMetricSources()
|
||
response["updated_at_ms"] = time.Now().UTC().UnixMilli()
|
||
applyExternalDashboardDeltas(response, current.Total, previous.Total)
|
||
response["salary_delta_rate"] = deltaRate(nullableMetricInt64(current.Total.SalaryUSDMinor), nullableMetricInt64(previous.Total.SalaryUSDMinor))
|
||
applyAslanMongoUnavailableFields(response)
|
||
return response, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) countryFilter(ctx context.Context, query StatisticsQuery) (externalDashboardCountryFilter, error) {
|
||
if query.CountryID > 0 {
|
||
return externalDashboardCountryFilter{}, fmt.Errorf("aslan mongo dashboard source %s does not support numeric country_id filter", s.appCode)
|
||
}
|
||
if query.RegionID <= 0 {
|
||
return externalDashboardCountryFilter{}, nil
|
||
}
|
||
for _, resolver := range s.regionResolvers {
|
||
codes, handled, err := resolver.CountryCodesForRegion(ctx, s.appCode, query.RegionID)
|
||
if err != nil {
|
||
return externalDashboardCountryFilter{}, err
|
||
}
|
||
if !handled {
|
||
continue
|
||
}
|
||
return externalDashboardCountryFilter{
|
||
Applied: true,
|
||
RegionID: query.RegionID,
|
||
Codes: normalizeExternalCountryCodes(codes),
|
||
}, nil
|
||
}
|
||
// Aslan Mongo 集合只有国家码，没有 hyapp 的数字 region_id；缺 legacy 区域映射时失败，避免误把全量数据当成某个区域返回。
|
||
return externalDashboardCountryFilter{}, fmt.Errorf("aslan mongo dashboard source %s cannot resolve region_id %d", s.appCode, query.RegionID)
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) loadRange(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter) (aslanRangeOverview, error) {
|
||
totalFlow := aslanDashboardMetric{}
|
||
totalSnapshot := aslanDashboardMetric{}
|
||
dailySeries := []map[string]any{}
|
||
dailyCountries := []map[string]any{}
|
||
countryAggregates := map[string]*aslanCountryAccumulator{}
|
||
|
||
// user_bank_balance 是余额快照而不是日报表；这里按当前快照填充每一天，区间汇总只保留最后一天快照，避免把余额按天累加。
|
||
salarySnapshot, err := s.loadHostSalaryBalanceByCountry(ctx, countryFilter)
|
||
if err != nil {
|
||
return aslanRangeOverview{}, err
|
||
}
|
||
|
||
for day := startDate; day.Before(endDate); day = day.AddDate(0, 0, 1) {
|
||
nextDay := day.AddDate(0, 0, 1)
|
||
registers, err := s.loadRegisteredUsersByCountry(ctx, day, nextDay, countryFilter)
|
||
if err != nil {
|
||
return aslanRangeOverview{}, err
|
||
}
|
||
activeUsers, err := s.loadActiveUsersByCountry(ctx, day, countryFilter)
|
||
if err != nil {
|
||
return aslanRangeOverview{}, err
|
||
}
|
||
recharges, err := s.loadRechargeByCountry(ctx, day, nextDay, countryFilter)
|
||
if err != nil {
|
||
return aslanRangeOverview{}, err
|
||
}
|
||
|
||
dayFlow := aslanDashboardMetric{}
|
||
daySnapshot := aslanDashboardMetric{}
|
||
dayCountryAggregates := map[string]*aslanCountryAccumulator{}
|
||
for _, countryCode := range aslanMongoCountryKeys(registers, activeUsers, recharges, salarySnapshot) {
|
||
metric := aslanDashboardMetric{
|
||
NewUsers: registers[countryCode],
|
||
ActiveUsers: activeUsers[countryCode],
|
||
PaidUsers: recharges[countryCode].Users,
|
||
RechargeUSDMinor: recharges[countryCode].AmountUSDMinor,
|
||
SalaryUSDMinor: salarySnapshot[countryCode],
|
||
}
|
||
dayFlow.addFlow(metric)
|
||
daySnapshot.addSnapshot(metric)
|
||
totalFlow.addFlow(metric)
|
||
country := dayCountryAggregates[countryCode]
|
||
if country == nil {
|
||
country = &aslanCountryAccumulator{CountryCode: countryCode, CountryName: countryCode, RegionID: countryFilter.RegionID}
|
||
dayCountryAggregates[countryCode] = country
|
||
}
|
||
country.Flow.addFlow(metric)
|
||
country.Snapshot.addSnapshot(metric)
|
||
}
|
||
|
||
totalSnapshot = daySnapshot
|
||
dayMetric := aslanCombinedMetric(dayFlow, daySnapshot)
|
||
dayRow := dayMetric.toMap()
|
||
dayLabel := dashboardSQLDate(day)
|
||
dayRow["label"] = dayLabel
|
||
dayRow["stat_day"] = dayLabel
|
||
applyAslanMongoUnavailableFields(dayRow)
|
||
dailySeries = append(dailySeries, dayRow)
|
||
mergeAslanCountryDay(countryAggregates, dayCountryAggregates)
|
||
for _, row := range aslanDailyCountryRows(dayLabel, dayCountryAggregates) {
|
||
applyAslanMongoUnavailableFields(row)
|
||
dailyCountries = append(dailyCountries, row)
|
||
}
|
||
}
|
||
|
||
countryRows := aslanCountryRows(countryAggregates)
|
||
for _, row := range countryRows {
|
||
applyAslanMongoUnavailableFields(row)
|
||
}
|
||
return aslanRangeOverview{
|
||
Total: aslanCombinedMetric(totalFlow, totalSnapshot),
|
||
DailySeries: dailySeries,
|
||
CountryBreakdown: countryRows,
|
||
DailyCountryBreakdown: dailyCountries,
|
||
}, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) loadRegisteredUsersByCountry(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter) (map[string]int64, error) {
|
||
match := bson.M{
|
||
"originSys": s.sysOrigin,
|
||
"del": bson.M{"$ne": true},
|
||
"createTime": bson.M{"$gte": startDate, "$lt": endDate},
|
||
"countryCode": bson.M{"$exists": true, "$ne": ""},
|
||
"accountStatus": bson.M{"$ne": "DELETED"},
|
||
}
|
||
aslanMongoApplyCountryFilter(match, "countryCode", countryFilter)
|
||
out, err := s.aggregateUniqueUsersByCountry(ctx, aslanMongoUserProfileCollection, match, "countryCode", "_id")
|
||
if err != nil {
|
||
return nil, fmt.Errorf("query aslan mongo registered users for %s: %w", dashboardSQLDate(startDate), err)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) loadActiveUsersByCountry(ctx context.Context, day time.Time, countryFilter externalDashboardCountryFilter) (map[string]int64, error) {
|
||
match := bson.M{
|
||
"sysOrigin": s.sysOrigin,
|
||
"activeDate": dashboardSQLDate(day),
|
||
"registerCountryCode": bson.M{"$exists": true, "$ne": ""},
|
||
}
|
||
aslanMongoApplyCountryFilter(match, "registerCountryCode", countryFilter)
|
||
out, err := s.aggregateUniqueUsersByCountry(ctx, aslanMongoUserActiveCollection, match, "registerCountryCode", "userId")
|
||
if err != nil {
|
||
return nil, fmt.Errorf("query aslan mongo active users for %s: %w", dashboardSQLDate(day), err)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) loadRechargeByCountry(ctx context.Context, startDate time.Time, endDate time.Time, countryFilter externalDashboardCountryFilter) (map[string]aslanMongoRechargeMetric, error) {
|
||
match := bson.M{
|
||
"sysOrigin": s.sysOrigin,
|
||
"status": "SUCCESS",
|
||
"trialPeriod": bson.M{"$ne": true},
|
||
"createTime": bson.M{"$gte": startDate, "$lt": endDate},
|
||
"countryCode": bson.M{"$exists": true, "$ne": ""},
|
||
}
|
||
aslanMongoApplyCountryFilter(match, "countryCode", countryFilter)
|
||
pipeline := mongo.Pipeline{
|
||
{{Key: "$match", Value: match}},
|
||
{{Key: "$group", Value: bson.D{
|
||
{Key: "_id", Value: "$countryCode"},
|
||
{Key: "amount", Value: bson.D{{Key: "$sum", Value: "$amountUsd"}}},
|
||
// Aslan 原 Top 充值统计按 acceptUserId 聚合；createUser 只在 acceptUserId 缺失的旧单据上兜底。
|
||
{Key: "users", Value: bson.D{{Key: "$addToSet", Value: bson.D{{Key: "$ifNull", Value: bson.A{"$acceptUserId", "$createUser"}}}}}},
|
||
}}},
|
||
}
|
||
cursor, err := s.db.Collection(aslanMongoInAppPurchaseCollection).Aggregate(ctx, pipeline, options.Aggregate().SetAllowDiskUse(true))
|
||
if err != nil {
|
||
return nil, fmt.Errorf("aggregate in_app_purchase_details: %w", err)
|
||
}
|
||
defer cursor.Close(ctx)
|
||
|
||
out := map[string]aslanMongoRechargeMetric{}
|
||
for cursor.Next(ctx) {
|
||
row := bson.M{}
|
||
if err := cursor.Decode(&row); err != nil {
|
||
return nil, fmt.Errorf("decode in_app_purchase_details aggregate: %w", err)
|
||
}
|
||
countryCode := aslanMongoCountryCode(row["_id"])
|
||
if countryCode == "" {
|
||
continue
|
||
}
|
||
amount, err := aslanMongoDecimalMinor(row["amount"])
|
||
if err != nil {
|
||
return nil, fmt.Errorf("parse recharge amount for country %s: %w", countryCode, err)
|
||
}
|
||
out[countryCode] = aslanMongoRechargeMetric{
|
||
AmountUSDMinor: amount,
|
||
Users: aslanMongoDistinctValueCount(row["users"]),
|
||
}
|
||
}
|
||
if err := cursor.Err(); err != nil {
|
||
return nil, fmt.Errorf("iterate in_app_purchase_details aggregate: %w", err)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) aggregateUniqueUsersByCountry(ctx context.Context, collectionName string, match bson.M, countryField string, userField string) (map[string]int64, error) {
|
||
pipeline := mongo.Pipeline{
|
||
{{Key: "$match", Value: match}},
|
||
{{Key: "$group", Value: bson.D{
|
||
{Key: "_id", Value: "$" + countryField},
|
||
{Key: "users", Value: bson.D{{Key: "$addToSet", Value: "$" + userField}}},
|
||
{Key: "count", Value: bson.D{{Key: "$sum", Value: 1}}},
|
||
}}},
|
||
}
|
||
cursor, err := s.db.Collection(collectionName).Aggregate(ctx, pipeline, options.Aggregate().SetAllowDiskUse(true))
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer cursor.Close(ctx)
|
||
|
||
out := map[string]int64{}
|
||
for cursor.Next(ctx) {
|
||
row := bson.M{}
|
||
if err := cursor.Decode(&row); err != nil {
|
||
return nil, err
|
||
}
|
||
countryCode := aslanMongoCountryCode(row["_id"])
|
||
if countryCode == "" {
|
||
continue
|
||
}
|
||
count := aslanMongoDistinctValueCount(row["users"])
|
||
if count == 0 {
|
||
count = aslanMongoWholeInt64(row["count"])
|
||
}
|
||
out[countryCode] = count
|
||
}
|
||
if err := cursor.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) loadHostSalaryBalanceByCountry(ctx context.Context, countryFilter externalDashboardCountryFilter) (map[string]int64, error) {
|
||
memberIDs, err := s.loadTeamMemberIDs(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(memberIDs) == 0 {
|
||
return map[string]int64{}, nil
|
||
}
|
||
countryByUser, err := s.loadUserCountries(ctx, memberIDs, countryFilter)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(countryByUser) == 0 {
|
||
return map[string]int64{}, nil
|
||
}
|
||
|
||
out := map[string]int64{}
|
||
for _, chunk := range aslanMongoInt64Chunks(memberIDs, aslanMongoChunkSize) {
|
||
cursor, err := s.db.Collection(aslanMongoBankBalanceCollection).Find(ctx, bson.M{
|
||
"sysOrigin": s.sysOrigin,
|
||
"_id": bson.M{"$in": chunk},
|
||
}, options.Find().SetProjection(bson.M{"_id": 1, "balance": 1}))
|
||
if err != nil {
|
||
return nil, fmt.Errorf("query user_bank_balance: %w", err)
|
||
}
|
||
for cursor.Next(ctx) {
|
||
row := bson.M{}
|
||
if err := cursor.Decode(&row); err != nil {
|
||
_ = cursor.Close(ctx)
|
||
return nil, fmt.Errorf("decode user_bank_balance: %w", err)
|
||
}
|
||
userID := aslanMongoWholeInt64(row["_id"])
|
||
countryCode := countryByUser[userID]
|
||
if countryCode == "" {
|
||
continue
|
||
}
|
||
// user_bank_balance.balance 是 PennyAmount 的 long 值；admin 响应也用 USD minor，所以不用再做美元小数换算。
|
||
out[countryCode] += aslanMongoWholeInt64(row["balance"])
|
||
}
|
||
if err := cursor.Err(); err != nil {
|
||
_ = cursor.Close(ctx)
|
||
return nil, fmt.Errorf("iterate user_bank_balance: %w", err)
|
||
}
|
||
_ = cursor.Close(ctx)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) loadTeamMemberIDs(ctx context.Context) ([]int64, error) {
|
||
cursor, err := s.db.Collection(aslanMongoTeamMemberCollection).Find(ctx, bson.M{
|
||
"sysOrigin": s.sysOrigin,
|
||
"role": bson.M{"$in": bson.A{"MEMBER", "ADMIN"}},
|
||
}, options.Find().SetProjection(bson.M{"memberId": 1}))
|
||
if err != nil {
|
||
return nil, fmt.Errorf("query team_member: %w", err)
|
||
}
|
||
defer cursor.Close(ctx)
|
||
|
||
seen := map[int64]struct{}{}
|
||
out := []int64{}
|
||
for cursor.Next(ctx) {
|
||
row := bson.M{}
|
||
if err := cursor.Decode(&row); err != nil {
|
||
return nil, fmt.Errorf("decode team_member: %w", err)
|
||
}
|
||
userID := aslanMongoWholeInt64(row["memberId"])
|
||
if userID <= 0 {
|
||
continue
|
||
}
|
||
if _, ok := seen[userID]; ok {
|
||
continue
|
||
}
|
||
seen[userID] = struct{}{}
|
||
out = append(out, userID)
|
||
}
|
||
if err := cursor.Err(); err != nil {
|
||
return nil, fmt.Errorf("iterate team_member: %w", err)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (s *AslanMongoDashboardSource) loadUserCountries(ctx context.Context, userIDs []int64, countryFilter externalDashboardCountryFilter) (map[int64]string, error) {
|
||
out := map[int64]string{}
|
||
for _, chunk := range aslanMongoInt64Chunks(userIDs, aslanMongoChunkSize) {
|
||
filter := bson.M{
|
||
"_id": bson.M{"$in": chunk},
|
||
"originSys": s.sysOrigin,
|
||
"del": bson.M{"$ne": true},
|
||
"countryCode": bson.M{"$exists": true, "$ne": ""},
|
||
}
|
||
aslanMongoApplyCountryFilter(filter, "countryCode", countryFilter)
|
||
cursor, err := s.db.Collection(aslanMongoUserProfileCollection).Find(ctx, filter, options.Find().SetProjection(bson.M{"_id": 1, "countryCode": 1}))
|
||
if err != nil {
|
||
return nil, fmt.Errorf("query user_run_profile countries: %w", err)
|
||
}
|
||
for cursor.Next(ctx) {
|
||
row := bson.M{}
|
||
if err := cursor.Decode(&row); err != nil {
|
||
_ = cursor.Close(ctx)
|
||
return nil, fmt.Errorf("decode user_run_profile country: %w", err)
|
||
}
|
||
userID := aslanMongoWholeInt64(row["_id"])
|
||
countryCode := aslanMongoCountryCode(row["countryCode"])
|
||
if userID > 0 && countryCode != "" {
|
||
out[userID] = countryCode
|
||
}
|
||
}
|
||
if err := cursor.Err(); err != nil {
|
||
_ = cursor.Close(ctx)
|
||
return nil, fmt.Errorf("iterate user_run_profile countries: %w", err)
|
||
}
|
||
_ = cursor.Close(ctx)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func aslanMongoApplyCountryFilter(match bson.M, field string, countryFilter externalDashboardCountryFilter) {
|
||
if !countryFilter.Applied {
|
||
return
|
||
}
|
||
if len(countryFilter.Codes) == 0 {
|
||
match[field] = bson.M{"$in": bson.A{"__NO_COUNTRY__"}}
|
||
return
|
||
}
|
||
codes := bson.A{}
|
||
for _, code := range countryFilter.Codes {
|
||
if normalized := aslanMongoCountryCode(code); normalized != "" {
|
||
codes = append(codes, normalized)
|
||
}
|
||
}
|
||
if len(codes) == 0 {
|
||
codes = bson.A{"__NO_COUNTRY__"}
|
||
}
|
||
match[field] = bson.M{"$in": codes}
|
||
}
|
||
|
||
func aslanMongoCountryKeys(registers map[string]int64, activeUsers map[string]int64, recharges map[string]aslanMongoRechargeMetric, salarySnapshot map[string]int64) []string {
|
||
seen := map[string]struct{}{}
|
||
for countryCode := range registers {
|
||
seen[countryCode] = struct{}{}
|
||
}
|
||
for countryCode := range activeUsers {
|
||
seen[countryCode] = struct{}{}
|
||
}
|
||
for countryCode := range recharges {
|
||
seen[countryCode] = struct{}{}
|
||
}
|
||
for countryCode := range salarySnapshot {
|
||
seen[countryCode] = struct{}{}
|
||
}
|
||
keys := make([]string, 0, len(seen))
|
||
for countryCode := range seen {
|
||
keys = append(keys, countryCode)
|
||
}
|
||
sort.Strings(keys)
|
||
return keys
|
||
}
|
||
|
||
func aslanMongoCountryCode(value any) string {
|
||
code := strings.ToUpper(strings.TrimSpace(fmt.Sprint(value)))
|
||
if code == "" || code == "<NIL>" || code == "NULL" {
|
||
return ""
|
||
}
|
||
return code
|
||
}
|
||
|
||
func aslanMongoDistinctValueCount(value any) int64 {
|
||
values, ok := value.(bson.A)
|
||
if !ok {
|
||
if raw, ok := value.([]any); ok {
|
||
values = raw
|
||
}
|
||
}
|
||
seen := map[string]struct{}{}
|
||
for _, item := range values {
|
||
key := strings.TrimSpace(fmt.Sprint(item))
|
||
if key == "" || key == "<nil>" || key == "0" {
|
||
continue
|
||
}
|
||
seen[key] = struct{}{}
|
||
}
|
||
return int64(len(seen))
|
||
}
|
||
|
||
func aslanMongoDecimalMinor(value any) (int64, error) {
|
||
switch typed := value.(type) {
|
||
case nil:
|
||
return 0, nil
|
||
case bson.Decimal128:
|
||
return decimalTextToMinor(typed.String())
|
||
case string:
|
||
return decimalTextToMinor(typed)
|
||
case int32:
|
||
return int64(typed) * 100, nil
|
||
case int64:
|
||
return typed * 100, nil
|
||
case int:
|
||
return int64(typed) * 100, nil
|
||
case float32:
|
||
return decimalTextToMinor(strconv.FormatFloat(float64(typed), 'f', -1, 32))
|
||
case float64:
|
||
return decimalTextToMinor(strconv.FormatFloat(typed, 'f', -1, 64))
|
||
default:
|
||
return decimalTextToMinor(fmt.Sprint(value))
|
||
}
|
||
}
|
||
|
||
func aslanMongoWholeInt64(value any) int64 {
|
||
switch typed := value.(type) {
|
||
case nil:
|
||
return 0
|
||
case int64:
|
||
return typed
|
||
case int32:
|
||
return int64(typed)
|
||
case int:
|
||
return int64(typed)
|
||
case float64:
|
||
return int64(typed)
|
||
case string:
|
||
parsed, _ := strconv.ParseInt(strings.TrimSpace(typed), 10, 64)
|
||
return parsed
|
||
case bson.Decimal128:
|
||
parsed, _ := strconv.ParseInt(strings.TrimSpace(typed.String()), 10, 64)
|
||
return parsed
|
||
default:
|
||
parsed, _ := strconv.ParseInt(strings.TrimSpace(fmt.Sprint(value)), 10, 64)
|
||
return parsed
|
||
}
|
||
}
|
||
|
||
func aslanMongoInt64Chunks(values []int64, size int) [][]int64 {
|
||
if size <= 0 {
|
||
size = aslanMongoChunkSize
|
||
}
|
||
out := [][]int64{}
|
||
for start := 0; start < len(values); start += size {
|
||
end := start + size
|
||
if end > len(values) {
|
||
end = len(values)
|
||
}
|
||
out = append(out, values[start:end])
|
||
}
|
||
return out
|
||
}
|
||
|
||
var aslanMongoUnavailableMetricFields = []string{
|
||
"new_user_recharge_usd_minor",
|
||
"coin_seller_recharge_usd_minor",
|
||
"coin_seller_stock_coin",
|
||
"coin_seller_transfer_coin",
|
||
"google_recharge_usd_minor",
|
||
"mifapay_recharge_usd_minor",
|
||
"game_turnover",
|
||
"game_payout",
|
||
"game_players",
|
||
"game_profit",
|
||
"game_profit_rate",
|
||
"game_payout_delta_rate",
|
||
"lucky_gift_turnover",
|
||
"lucky_gift_payout",
|
||
"lucky_gift_payout_rate",
|
||
"lucky_gift_payers",
|
||
"lucky_gift_anchor_share",
|
||
"lucky_gift_profit",
|
||
"lucky_gift_profit_rate",
|
||
"lucky_gift_payout_delta_rate",
|
||
"super_lucky_gift_turnover",
|
||
"super_lucky_gift_payout",
|
||
"super_lucky_gift_profit",
|
||
"super_lucky_gift_profit_rate",
|
||
"room_lucky_gift_turnover",
|
||
"gift_coin_spent",
|
||
"coin_total",
|
||
"coin_total_delta_rate",
|
||
"consumed_coin",
|
||
"output_coin",
|
||
"consume_output_ratio",
|
||
"platform_grant_coin",
|
||
"manual_grant_coin",
|
||
"new_dealer_recharge_usd_minor",
|
||
"salary_exchange_usd_minor",
|
||
"salary_transfer_coin",
|
||
"salary_transfer_usd_minor",
|
||
"mic_online_ms",
|
||
"avg_mic_online_ms",
|
||
"gift_coin_spent_delta_rate",
|
||
"game_turnover_delta_rate",
|
||
"game_profit_delta_rate",
|
||
"lucky_gift_turnover_delta_rate",
|
||
"lucky_gift_profit_delta_rate",
|
||
"d1_retention_rate",
|
||
"d7_retention_rate",
|
||
"d30_retention_rate",
|
||
}
|
||
|
||
func applyAslanMongoUnavailableFields(row map[string]any) {
|
||
for _, field := range aslanMongoUnavailableMetricFields {
|
||
row[field] = nil
|
||
}
|
||
}
|
||
|
||
func aslanMongoDashboardMetricSources() []map[string]any {
|
||
return []map[string]any{
|
||
{"field": "new_users", "available": true, "source": "Aslan Mongo user_run_profile createTime/countryCode"},
|
||
{"field": "active_users", "available": true, "source": "Aslan Mongo user_daily_active_log activeDate/registerCountryCode"},
|
||
{"field": "paid_users", "available": true, "source": "Aslan Mongo in_app_purchase_details SUCCESS distinct acceptUserId"},
|
||
{"field": "recharge_users", "available": true, "source": "Aslan Mongo in_app_purchase_details SUCCESS distinct acceptUserId"},
|
||
{"field": "recharge_usd_minor", "available": true, "source": "Aslan Mongo in_app_purchase_details amountUsd"},
|
||
{"field": "new_user_recharge_usd_minor", "available": false, "missing_reason": "Mongo 中未确认新用户充值 cohort 聚合口径"},
|
||
{"field": "coin_seller_recharge_usd_minor", "available": false, "missing_reason": "Mongo 中未确认币商充值金额聚合口径"},
|
||
{"field": "coin_seller_stock_coin", "available": false, "missing_reason": "Mongo 中未确认币商充值金币聚合口径"},
|
||
{"field": "coin_seller_transfer_coin", "available": false, "missing_reason": "Mongo 中未确认币商出货金币聚合口径"},
|
||
{"field": "google_recharge_usd_minor", "available": false, "missing_reason": "in_app_purchase_details 当前接入只做总充值，未拆 Google/三方渠道"},
|
||
{"field": "mifapay_recharge_usd_minor", "available": false, "missing_reason": "in_app_purchase_details 当前接入只做总充值，未拆 Google/三方渠道"},
|
||
{"field": "game_turnover", "available": false, "missing_reason": "Mongo 中未确认游戏流水和返奖集合"},
|
||
{"field": "lucky_gift_turnover", "available": false, "missing_reason": "Mongo 中未确认幸运礼物流水和返奖集合"},
|
||
{"field": "super_lucky_gift_turnover", "available": false, "missing_reason": "Mongo 中未确认超级幸运礼物流水和返奖集合"},
|
||
{"field": "gift_coin_spent", "available": false, "missing_reason": "Mongo 中未确认礼物流水总口径"},
|
||
{"field": "coin_total", "available": false, "missing_reason": "Mongo 中未发现可按国家低成本汇总的当前金币余额快照"},
|
||
{"field": "consumed_coin", "available": false, "missing_reason": "Mongo 中未确认消耗金币总口径"},
|
||
{"field": "output_coin", "available": false, "missing_reason": "Mongo 中未确认产出金币总口径"},
|
||
{"field": "consume_output_ratio", "available": false, "missing_reason": "缺少消耗金币和产出金币总口径"},
|
||
{"field": "platform_grant_coin", "available": false, "missing_reason": "Mongo 中未确认平台发放金币来源枚举"},
|
||
{"field": "manual_grant_coin", "available": false, "missing_reason": "Mongo 中未确认人工发放金币来源枚举"},
|
||
{"field": "salary_usd_minor", "available": true, "source": "Aslan Mongo team_member + user_bank_balance + user_run_profile current snapshot"},
|
||
{"field": "salary_transfer_coin", "available": false, "missing_reason": "Mongo 中未确认工资兑换金币流水口径"},
|
||
{"field": "avg_mic_online_ms", "available": false, "missing_reason": "Mongo 中未确认麦上时长集合"},
|
||
{"field": "mic_online_ms", "available": false, "missing_reason": "Mongo 中未确认麦上时长集合"},
|
||
{"field": "arpu_usd_minor", "available": true, "source": "recharge_usd_minor / active_users"},
|
||
{"field": "arppu_usd_minor", "available": true, "source": "recharge_usd_minor / paid_users"},
|
||
{"field": "paid_conversion_rate", "available": true, "source": "paid_users / active_users"},
|
||
{"field": "recharge_conversion_rate", "available": true, "source": "recharge_users / active_users"},
|
||
{"field": "retention.day1_rate", "available": false, "missing_reason": "Mongo 中未确认留存 cohort 活跃事实"},
|
||
{"field": "retention.day7_rate", "available": false, "missing_reason": "Mongo 中未确认留存 cohort 活跃事实"},
|
||
{"field": "retention.day30_rate", "available": false, "missing_reason": "Mongo 中未确认留存 cohort 活跃事实"},
|
||
}
|
||
}
|