From ea347d29210ed16c5b5b319c87a93c808dc856bf Mon Sep 17 00:00:00 2001 From: zhx Date: Fri, 3 Jul 2026 21:03:23 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E9=93=BE=E8=B7=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../admin/configs/config.tencent.example.yaml | 4 +- .../dashboard/aslan_mongo_dashboard.go | 238 ++++++++++++++---- .../dashboard/aslan_mongo_dashboard_test.go | 58 +++++ .../admin/internal/modules/databi/service.go | 145 +++++++++-- .../internal/modules/databi/service_test.go | 20 ++ .../modules/payment/money_region_source.go | 4 +- .../repository/money_scope_repository.go | 14 +- .../repository/money_scope_repository_test.go | 29 +++ 8 files changed, 435 insertions(+), 77 deletions(-) create mode 100644 server/admin/internal/modules/dashboard/aslan_mongo_dashboard_test.go diff --git a/server/admin/configs/config.tencent.example.yaml b/server/admin/configs/config.tencent.example.yaml index e7e0993f..0c4f513d 100644 --- a/server/admin/configs/config.tencent.example.yaml +++ b/server/admin/configs/config.tencent.example.yaml @@ -117,7 +117,9 @@ dashboard_external_sources: mongo_uri: "mongodb://aslan_data:aslan_data@lb-le274w4o-wjgmcdkhye3v0ors.clb.sg-tencentclb.com:27017,lb-le274w4o-wjgmcdkhye3v0ors.clb.sg-tencentclb.com:27018/test?replicaSet=cmgo-6cztxjgr_0&authSource=admin&readPreference=secondaryPreferred" mongo_database: "atyou" stat_timezone: "Asia/Shanghai" - request_timeout: "5s" + # aslan_mongo 源按整段区间做聚合（注册/日活/充值/礼物 + 留存 cohort），30 天区间比逐日查询重， + # 5s 会在长区间超时；线上建议 30s 起。 + request_timeout: "30s" finance_notifications: dingtalk: enabled: true diff --git a/server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go b/server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go index 9c5dd387..64b5fc39 100644 --- a/server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go +++ b/server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go @@ -6,6 +6,7 @@ import ( "sort" "strconv" "strings" + "sync" "time" "hyapp-admin-server/internal/config" @@ -39,6 +40,30 @@ type AslanMongoDashboardSource struct { 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 里的已存在事实集合。 @@ -99,15 +124,28 @@ func (s *AslanMongoDashboardSource) StatisticsOverview(ctx context.Context, quer } startDate, endDate, startMS, endMS := externalDashboardDateRange(query, requestLocation) - current, err := s.loadRange(ctx, startDate, endDate, countryFilter, true) - if err != nil { - return nil, err - } days := countCalendarDays(startDate, endDate) previousStart := startDate.AddDate(0, 0, -days) - // 环比只消费 applyExternalDashboardDeltas 里的流水字段，留存链路查询昂贵且不参与环比，跳过。 - previous, err := s.loadRange(ctx, previousStart, startDate, countryFilter, false) - if err != nil { + 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 } @@ -122,12 +160,18 @@ func (s *AslanMongoDashboardSource) StatisticsOverview(ctx context.Context, quer 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() + response["report_metric_sources"] = aslanMongoDashboardMetricSources(goldWater) 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) + if !goldWater { + applyAslanMongoGoldWaterUnavailable(response) + response["lucky_gift_payout_delta_rate"] = nil + response["lucky_gift_profit_delta_rate"] = nil + } return response, nil } @@ -246,22 +290,30 @@ func (m *legacyMongoMetric) toExternal(coinFlow *legacyMongoCoinFlow) externalDa type legacyMongoGrid struct { days []string + daySet map[string]struct{} metrics map[string]map[string]*legacyMongoMetric coinFlows map[string]*legacyMongoCoinFlow + // 各 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{}, } for day := startDate; day.Before(endDate); day = day.AddDate(0, 0, 1) { - grid.days = append(grid.days, dashboardSQLDate(day)) + 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{} @@ -276,17 +328,33 @@ func (g *legacyMongoGrid) at(day string, countryCode string) *legacyMongoMetric } func (g *legacyMongoGrid) hasDay(day string) bool { - for _, item := range g.days { - if item == day { - return true - } + _, 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) } - return false + 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 @@ -297,28 +365,31 @@ func (s *AslanMongoDashboardSource) loadRange(ctx context.Context, startDate tim } } - if err := s.loadActiveCounts(ctx, startDate, endDate, countryFilter, grid); err != nil { - return aslanRangeOverview{}, err + 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 err := s.loadRechargeStats(ctx, startDate, endDate, countryFilter, cohorts, grid); err != nil { - return aslanRangeOverview{}, err - } - if err := s.loadGiftStats(ctx, startDate, endDate, countryFilter, grid); err != nil { - return aslanRangeOverview{}, err - } - if err := s.loadGoldOriginStats(ctx, startDate, endDate, countryFilter, grid); err != nil { - return aslanRangeOverview{}, err - } - if err := s.loadCoinFlows(ctx, startDate, endDate, grid); err != nil { - return aslanRangeOverview{}, err + 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 { - if err := s.applyRetention(ctx, cohorts, grid); err != nil { - return aslanRangeOverview{}, err - } + tasks = append(tasks, func() error { return s.applyRetention(ctx, cohorts, grid) }) } - salarySnapshot, err := s.loadHostSalaryBalanceByCountry(ctx, countryFilter) - if err != nil { + if err := runConcurrently(tasks); err != nil { return aslanRangeOverview{}, err } // user_bank_balance 是余额快照而不是日报表；这里按当前快照填充每一天，区间汇总只保留一份快照，避免把余额按天累加。 @@ -328,10 +399,10 @@ func (s *AslanMongoDashboardSource) loadRange(ctx context.Context, startDate tim } } - return s.assembleRange(grid, salarySnapshot, countryFilter), nil + return s.assembleRange(grid, countryFilter, goldWater), nil } -func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, salarySnapshot map[string]int64, countryFilter externalDashboardCountryFilter) aslanRangeOverview { +func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, countryFilter externalDashboardCountryFilter, goldWater bool) aslanRangeOverview { dailySeries := []map[string]any{} dailyCountries := []map[string]any{} totalFlow := legacyMongoMetric{} @@ -369,6 +440,9 @@ func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, salaryS row["region_id"] = countryFilter.RegionID } applyAslanMongoUnavailableFields(row) + if !goldWater { + applyAslanMongoGoldWaterUnavailable(row) + } dailyCountries = append(dailyCountries, row) } coinFlow := grid.coinFlows[day] @@ -385,6 +459,9 @@ func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, salaryS dayRow["label"] = day dayRow["stat_day"] = day applyAslanMongoUnavailableFields(dayRow) + if !goldWater { + applyAslanMongoGoldWaterUnavailable(dayRow) + } dailySeries = append(dailySeries, dayRow) } @@ -404,6 +481,9 @@ func (s *AslanMongoDashboardSource) assembleRange(grid *legacyMongoGrid, salaryS row["region_id"] = countryFilter.RegionID } applyAslanMongoUnavailableFields(row) + if !goldWater { + applyAslanMongoGoldWaterUnavailable(row) + } countryRows = append(countryRows, row) } sort.Slice(countryRows, func(i, j int) bool { @@ -529,8 +609,8 @@ func (s *AslanMongoDashboardSource) loadRechargeStats(ctx context.Context, start } distinctUsers := map[dayCountryUsers]map[string]struct{}{} err := s.aggregateEach(ctx, aslanMongoInAppPurchaseCollection, pipeline, func(row bson.M) error { - key, ok := row["_id"].(bson.M) - if !ok { + key := aslanMongoDocument(row["_id"]) + if key == nil { return nil } day := strings.TrimSpace(fmt.Sprint(key["day"])) @@ -581,8 +661,8 @@ func (s *AslanMongoDashboardSource) loadRechargeStats(ctx context.Context, start } perUser := map[string]map[int64]int64{} err = s.aggregateEach(ctx, aslanMongoInAppPurchaseCollection, perUserPipeline, func(row bson.M) error { - key, ok := row["_id"].(bson.M) - if !ok { + key := aslanMongoDocument(row["_id"]) + if key == nil { return nil } day := strings.TrimSpace(fmt.Sprint(key["day"])) @@ -655,8 +735,8 @@ func (s *AslanMongoDashboardSource) loadGiftStats(ctx context.Context, startDate } entries := []giftEntry{} err := s.aggregateEach(ctx, aslanMongoGiftWaterCollection, pipeline, func(row bson.M) error { - key, ok := row["_id"].(bson.M) - if !ok { + key := aslanMongoDocument(row["_id"]) + if key == nil { return nil } day := strings.TrimSpace(fmt.Sprint(key["day"])) @@ -743,8 +823,8 @@ func (s *AslanMongoDashboardSource) loadGoldOriginStats(ctx context.Context, sta } entries := []goldEntry{} err := s.aggregateEach(ctx, aslanMongoGoldWaterCollection, pipeline, func(row bson.M) error { - key, ok := row["_id"].(bson.M) - if !ok { + key := aslanMongoDocument(row["_id"]) + if key == nil { return nil } day := strings.TrimSpace(fmt.Sprint(key["day"])) @@ -806,8 +886,8 @@ func (s *AslanMongoDashboardSource) loadCoinFlows(ctx context.Context, startDate }}}, } err := s.aggregateEach(ctx, aslanMongoGoldWaterCollection, pipeline, func(row bson.M) error { - key, ok := row["_id"].(bson.M) - if !ok { + key := aslanMongoDocument(row["_id"]) + if key == nil { return nil } day := strings.TrimSpace(fmt.Sprint(key["day"])) @@ -1078,9 +1158,28 @@ func (s *AslanMongoDashboardSource) dayExpression(field string) bson.D { }}} } +// 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, ok := value.(bson.M) - if !ok { + key := aslanMongoDocument(value) + if key == nil { return "", "" } day := strings.TrimSpace(fmt.Sprint(key["day"])) @@ -1256,8 +1355,44 @@ func applyAslanMongoUnavailableFields(row map[string]any) { } } -func aslanMongoDashboardMetricSources() []map[string]any { - return []map[string]any{ +// 金币流水集合缺失时必须显示 "--" 的字段；置 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) []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 集合"}, + } + } + return append([]map[string]any{ {"field": "new_users", "available": true, "source": "likei Mongo user_run_profile createTime/countryCode"}, {"field": "active_users", "available": true, "source": "likei Mongo user_daily_active_log activeDate/registerCountryCode"}, {"field": "paid_users", "available": true, "source": "likei Mongo in_app_purchase_details SUCCESS distinct acceptUserId"}, @@ -1268,13 +1403,8 @@ func aslanMongoDashboardMetricSources() []map[string]any { {"field": "new_user_recharge_usd_minor", "available": true, "source": "likei Mongo 当日注册 cohort × in_app_purchase_details 当日充值"}, {"field": "gift_coin_spent", "available": true, "source": "likei Mongo gift_give_running_water giftValue.actualAmount（含幸运礼物）"}, {"field": "lucky_gift_turnover", "available": true, "source": "likei Mongo gift_give_running_water luckyGift=true"}, - {"field": "lucky_gift_payout", "available": true, "source": "likei Mongo user_gold_running_water_v2 origin=LUCKY_GIFT_GOLD_REWARD*"}, - {"field": "coin_seller_transfer_coin", "available": true, "source": "likei Mongo user_gold_running_water_v2 origin=SELLER_AGENT*"}, {"field": "coin_seller_recharge_usd_minor", "available": false, "missing_reason": "likei Mongo 未确认币商充值美元口径"}, {"field": "coin_seller_stock_coin", "available": false, "missing_reason": "likei Mongo 未确认币商进货金币口径"}, - {"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 口径）"}, {"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 没有可按国家低成本汇总的金币余额快照"}, @@ -1289,5 +1419,5 @@ func aslanMongoDashboardMetricSources() []map[string]any { {"field": "retention.day1_rate", "available": true, "source": "likei Mongo 注册 cohort × user_daily_active_log D+1"}, {"field": "retention.day7_rate", "available": true, "source": "likei Mongo 注册 cohort × user_daily_active_log D+7"}, {"field": "retention.day30_rate", "available": true, "source": "likei Mongo 注册 cohort × user_daily_active_log D+30"}, - } + }, goldWaterSources...) } diff --git a/server/admin/internal/modules/dashboard/aslan_mongo_dashboard_test.go b/server/admin/internal/modules/dashboard/aslan_mongo_dashboard_test.go new file mode 100644 index 00000000..2e944d50 --- /dev/null +++ b/server/admin/internal/modules/dashboard/aslan_mongo_dashboard_test.go @@ -0,0 +1,58 @@ +package dashboard + +import ( + "testing" + + "go.mongodb.org/mongo-driver/v2/bson" +) + +// 复合 $group 的 _id 子文档经 driver 解码进 bson.M 后，嵌套值默认是 bson.D 而不是 bson.M； +// 这个测试锁死我们对两种形态的兼容，防止聚合行被静默丢弃（表现为"成功但全零"）。 +func TestAslanMongoDayCountryHandlesDecodedShapes(t *testing.T) { + raw, err := bson.Marshal(bson.M{"_id": bson.M{"day": "2026-07-02", "country": "sa"}, "amount": int64(5)}) + if err != nil { + t.Fatalf("marshal fixture: %v", err) + } + row := bson.M{} + if err := bson.Unmarshal(raw, &row); err != nil { + t.Fatalf("unmarshal fixture: %v", err) + } + if _, isM := row["_id"].(bson.M); isM { + t.Log("driver decoded nested document as bson.M") + } else if _, isD := row["_id"].(bson.D); isD { + t.Log("driver decoded nested document as bson.D (must be supported)") + } else { + t.Fatalf("unexpected nested _id type %T", row["_id"]) + } + + day, country := aslanMongoDayCountry(row["_id"]) + if day != "2026-07-02" || country != "SA" { + t.Fatalf("expected day/country from decoded _id, got %q/%q", day, country) + } + + // 直接构造 bson.D 与 bson.M 两种形态都必须可读。 + for _, value := range []any{ + bson.D{{Key: "day", Value: "2026-07-01"}, {Key: "country", Value: "eg"}}, + bson.M{"day": "2026-07-01", "country": "eg"}, + } { + day, country := aslanMongoDayCountry(value) + if day != "2026-07-01" || country != "EG" { + t.Fatalf("expected 2026-07-01/EG for %T, got %q/%q", value, day, country) + } + } +} + +func TestAslanMongoDocumentConversion(t *testing.T) { + if doc := aslanMongoDocument(bson.D{{Key: "day", Value: "2026-07-02"}, {Key: "type", Value: int32(1)}}); doc == nil || doc["day"] != "2026-07-02" { + t.Fatalf("bson.D conversion failed: %v", doc) + } + if doc := aslanMongoDocument(bson.M{"user": int64(7)}); doc == nil || doc["user"] != int64(7) { + t.Fatalf("bson.M passthrough failed: %v", doc) + } + if doc := aslanMongoDocument(nil); doc != nil { + t.Fatalf("nil should convert to nil, got %v", doc) + } + if doc := aslanMongoDocument("scalar"); doc != nil { + t.Fatalf("scalar should convert to nil, got %v", doc) + } +} diff --git a/server/admin/internal/modules/databi/service.go b/server/admin/internal/modules/databi/service.go index 4a8a8f6e..aeb191fb 100644 --- a/server/admin/internal/modules/databi/service.go +++ b/server/admin/internal/modules/databi/service.go @@ -45,6 +45,53 @@ type Service struct { user userclient.Client legacyApps []LegacyAppDescriptor legacyRegionCatalogs []LegacyRegionCatalogSource + + overviewMu sync.Mutex + overviewCalls map[string]*overviewCall +} + +// overviewCall 合并同参数的统计查询：BI 页面的 overview 与 kpi 会并发请求同一窗口， +// legacy Mongo 聚合昂贵，同窗口只算一次并短暂缓存（BI 数据分钟级新鲜度足够）。 +type overviewCall struct { + done chan struct{} + data map[string]any + err error + expiresAt time.Time +} + +const overviewCacheTTL = 45 * time.Second + +func (s *Service) statisticsOverview(ctx context.Context, query dashboard.StatisticsQuery) (map[string]any, error) { + key := fmt.Sprintf("%s|%s|%d|%d|%d", appctx.Normalize(query.AppCode), strings.TrimSpace(query.StatTZ), query.StartMS, query.EndMS, query.RegionID) + s.overviewMu.Lock() + if s.overviewCalls == nil { + s.overviewCalls = map[string]*overviewCall{} + } + if call, ok := s.overviewCalls[key]; ok { + expired := !call.expiresAt.IsZero() && time.Now().After(call.expiresAt) + if !expired { + s.overviewMu.Unlock() + <-call.done + return call.data, call.err + } + delete(s.overviewCalls, key) + } + call := &overviewCall{done: make(chan struct{})} + s.overviewCalls[key] = call + s.overviewMu.Unlock() + + // 共享结果不随首个请求方取消：其余等待者可能仍需要它。 + call.data, call.err = s.dashboards.StatisticsOverview(context.WithoutCancel(ctx), query) + s.overviewMu.Lock() + if call.err != nil { + // 失败不缓存，后续请求立即重试。 + delete(s.overviewCalls, key) + } else { + call.expiresAt = time.Now().Add(overviewCacheTTL) + } + s.overviewMu.Unlock() + close(call.done) + return call.data, call.err } type ServiceOption func(*Service) @@ -142,7 +189,7 @@ func (s *Service) Master(ctx context.Context, access repository.MoneyAccess, act regions := []RegionInfo{} for _, app := range apps { allowAll, allowedIDs := allowedRegions(access, app.AppCode) - allowedSet := int64Set(allowedIDs) + allowedSet := int64Set(normalizeRegionIDs(catalog[app.AppCode], allowedIDs)) for _, region := range catalog[app.AppCode] { if !allowAll { if _, ok := allowedSet[region.RegionID]; !ok { @@ -264,6 +311,7 @@ func (s *Service) appOverview(ctx context.Context, app AppInfo, regions []Region DailyCountryBreakdown: []map[string]any{}, } allowAll, allowedIDs := allowedRegions(access, app.AppCode) + allowedIDs = normalizeRegionIDs(regions, allowedIDs) if !allowAll && len(allowedIDs) == 0 { out.Error = "没有该 App 的数据权限" return out @@ -277,7 +325,7 @@ func (s *Service) appOverview(ctx context.Context, app AppInfo, regions []Region return out } - overview, err := s.dashboards.StatisticsOverview(ctx, dashboard.StatisticsQuery{ + overview, err := s.statisticsOverview(ctx, dashboard.StatisticsQuery{ AppCode: app.AppCode, StatTZ: query.StatTZ, StartMS: query.StartMS, @@ -458,17 +506,33 @@ func (s *Service) Kpi(ctx context.Context, access repository.MoneyAccess, query defer func() { <-semaphore }() data := &appKpiData{} resolve := regionResolver(catalog[app.AppCode]) - rangeTotal, rangeRegion, err := s.kpiWindow(ctx, app, resolve, query.StatTZ, query.StartMS, query.EndMS) - if err != nil { - data.err = err - } else { - data.rangeTotal, data.rangeRegion = rangeTotal, rangeRegion - monthTotal, monthRegion, err := s.kpiWindow(ctx, app, resolve, query.StatTZ, monthStart.UnixMilli(), monthEnd.UnixMilli()) + // 展示区间与当月两个窗口互不依赖，并行取；同参数窗口会被 statisticsOverview 合并/命中缓存。 + var windowWG sync.WaitGroup + windowWG.Add(2) + var rangeErr, monthErr error + go func() { + defer windowWG.Done() + total, region, err := s.kpiWindow(ctx, app, resolve, query.StatTZ, query.StartMS, query.EndMS) if err != nil { - data.err = err - } else { - data.monthTotal, data.monthRegion = monthTotal, monthRegion + rangeErr = err + return } + data.rangeTotal, data.rangeRegion = total, region + }() + go func() { + defer windowWG.Done() + total, region, err := s.kpiWindow(ctx, app, resolve, query.StatTZ, monthStart.UnixMilli(), monthEnd.UnixMilli()) + if err != nil { + monthErr = err + return + } + data.monthTotal, data.monthRegion = total, region + }() + windowWG.Wait() + if rangeErr != nil { + data.err = rangeErr + } else if monthErr != nil { + data.err = monthErr } mu.Lock() dataByApp[app.AppCode] = data @@ -504,6 +568,8 @@ func (s *Service) Kpi(ctx context.Context, access repository.MoneyAccess, query if !ok { app = AppInfo{AppCode: appCode, AppName: appCode, Kind: appKindLegacy} } + // 历史 scope 里的 legacy 区域 ID 可能被前端 float64 圆整过，先还原成目录精确 ID 再做展示与取数。 + regionID := normalizeRegionID(catalog[appCode], scope.RegionID) row := KpiRow{ OperatorUserID: user.ID, OperatorName: user.Name, @@ -512,10 +578,10 @@ func (s *Service) Kpi(ctx context.Context, access repository.MoneyAccess, query TeamID: user.TeamID, AppCode: appCode, AppName: app.AppName, - RegionID: scope.RegionID, + RegionID: regionID, } - row.RegionCode, row.RegionName = regionDisplay(catalog[appCode], scope.RegionID) - if target, ok := targetByKey[kpiTargetKey(user.ID, appCode, scope.RegionID)]; ok { + row.RegionCode, row.RegionName = regionDisplay(catalog[appCode], regionID) + if target, ok := lookupKpiTarget(targetByKey, user.ID, appCode, scope.RegionID, regionID); ok { row.MonthTargetUSDMinor = target.TargetUSDMinor row.DailyTargetUSDMinor = target.DailyTargetUSDMinor } @@ -526,8 +592,8 @@ func (s *Service) Kpi(ctx context.Context, access repository.MoneyAccess, query row.DataError = fmt.Sprintf("获取统计数据失败: %v", data.err) appErrors[appCode] = row.DataError } else { - rangeMetrics := scopeMetrics(data.rangeTotal, data.rangeRegion, scope.RegionID) - monthMetrics := scopeMetrics(data.monthTotal, data.monthRegion, scope.RegionID) + rangeMetrics := scopeMetrics(data.rangeTotal, data.rangeRegion, regionID) + monthMetrics := scopeMetrics(data.monthTotal, data.monthRegion, regionID) row.RangeRechargeUSDMinor = metricPointer(rangeMetrics, "recharge_usd_minor") row.RangeNewUsers = metricPointer(rangeMetrics, "new_users") row.RangeActiveUsers = metricPointer(rangeMetrics, "active_users") @@ -593,7 +659,7 @@ func (s *Service) Kpi(ctx context.Context, access repository.MoneyAccess, query } func (s *Service) kpiWindow(ctx context.Context, app AppInfo, resolve func(map[string]any) (RegionInfo, bool), statTZ string, startMS int64, endMS int64) (map[string]any, map[int64]map[string]any, error) { - overview, err := s.dashboards.StatisticsOverview(ctx, dashboard.StatisticsQuery{ + overview, err := s.statisticsOverview(ctx, dashboard.StatisticsQuery{ AppCode: app.AppCode, StatTZ: statTZ, StartMS: startMS, @@ -637,6 +703,25 @@ func kpiTargetKey(userID uint, appCode string, regionID int64) string { return strconv.FormatUint(uint64(userID), 10) + ":" + appctx.Normalize(appCode) + ":" + strconv.FormatInt(regionID, 10) } +// lookupKpiTarget 依次用 scope 原始 ID 与归一后的精确 ID 查目标： +// 目标行是前端写入的，legacy 大区可能存的是圆整值，也可能是修复后的精确值。 +func lookupKpiTarget(targetByKey map[string]model.DatabiKpiTarget, userID uint, appCode string, rawRegionID int64, regionID int64) (model.DatabiKpiTarget, bool) { + if target, ok := targetByKey[kpiTargetKey(userID, appCode, rawRegionID)]; ok { + return target, true + } + if regionID != rawRegionID { + if target, ok := targetByKey[kpiTargetKey(userID, appCode, regionID)]; ok { + return target, true + } + } + if rounded := int64(float64(regionID)); rounded != rawRegionID && rounded != regionID { + if target, ok := targetByKey[kpiTargetKey(userID, appCode, rounded)]; ok { + return target, true + } + } + return model.DatabiKpiTarget{}, false +} + // resolveKpiMonth 解析目标月份并给出该月在统计时区下的起止时间；结束时间截断到明天零点， // 避免 legacy Mongo 源为未来日期空跑逐日聚合。 func resolveKpiMonth(periodMonth string, endMS int64, location *time.Location) (string, time.Time, time.Time, error) { @@ -854,19 +939,41 @@ func regionResolver(regions []RegionInfo) func(map[string]any) (RegionInfo, bool func regionInCatalog(regions []RegionInfo, regionID int64) bool { for _, region := range regions { - if region.RegionID == regionID { + if repository.RegionIDLooseEqual(region.RegionID, regionID) { return true } } return false } +func normalizeRegionIDs(regions []RegionInfo, regionIDs []int64) []int64 { + out := make([]int64, 0, len(regionIDs)) + for _, regionID := range regionIDs { + out = append(out, normalizeRegionID(regions, regionID)) + } + return out +} + +// normalizeRegionID 把前端/历史数据里被 float64 圆整过的 legacy 区域 ID 还原成目录里的精确 ID； +// 目录没有对应项时原样返回（显示与取数各自兜底）。 +func normalizeRegionID(regions []RegionInfo, regionID int64) int64 { + if regionID == 0 { + return 0 + } + for _, region := range regions { + if repository.RegionIDLooseEqual(region.RegionID, regionID) { + return region.RegionID + } + } + return regionID +} + func regionDisplay(regions []RegionInfo, regionID int64) (string, string) { if regionID == 0 { return "", "全部区域" } for _, region := range regions { - if region.RegionID == regionID { + if repository.RegionIDLooseEqual(region.RegionID, regionID) { return region.RegionCode, region.RegionName } } diff --git a/server/admin/internal/modules/databi/service_test.go b/server/admin/internal/modules/databi/service_test.go index 487830ba..7d9c6689 100644 --- a/server/admin/internal/modules/databi/service_test.go +++ b/server/admin/internal/modules/databi/service_test.go @@ -56,3 +56,23 @@ func TestTopLevelMetricsStripsStructuralKeys(t *testing.T) { t.Fatalf("expected metric kept, got %v", out["recharge_usd_minor"]) } } + +// legacy 雪花区域 ID 超出 JS 安全整数，历史 scope 里存的是 float64 圆整值； +// 归一与展示必须能容差还原（线上「区域 20453...600」事故的回归测试）。 +func TestRegionDisplayToleratesRoundedIDs(t *testing.T) { + exact := int64(2049084769873498113) + rounded := int64(float64(exact)) + if exact == rounded { + t.Fatalf("fixture should lose precision") + } + regions := []RegionInfo{{AppCode: "aslan", RegionID: exact, RegionCode: "AR", RegionName: "阿语区"}} + if _, name := regionDisplay(regions, rounded); name != "阿语区" { + t.Fatalf("expected rounded id to resolve region name, got %s", name) + } + if got := normalizeRegionID(regions, rounded); got != exact { + t.Fatalf("expected normalize to exact id, got %d", got) + } + if !regionInCatalog(regions, rounded) { + t.Fatalf("expected rounded id to match catalog") + } +} diff --git a/server/admin/internal/modules/payment/money_region_source.go b/server/admin/internal/modules/payment/money_region_source.go index 49de1269..99362d46 100644 --- a/server/admin/internal/modules/payment/money_region_source.go +++ b/server/admin/internal/modules/payment/money_region_source.go @@ -110,7 +110,7 @@ func (s *MongoMoneyRegionSource) CountryBelongsToRegion(ctx context.Context, app if err != nil { return false, false, err } - if documentRegionID != regionID { + if !repository.RegionIDLooseEqual(documentRegionID, regionID) { continue } for _, item := range splitLegacyCountryCodes(document.CountryCodes) { @@ -148,7 +148,7 @@ func (s *MongoMoneyRegionSource) CountryCodesForRegion(ctx context.Context, appC if err != nil { return nil, false, err } - if documentRegionID != regionID { + if !repository.RegionIDLooseEqual(documentRegionID, regionID) { continue } // 大屏外接源只认 legacy country_code；这里集中复用财务范围的 Mongo 区域口径，避免每个外接 App 自己维护一套区域映射。 diff --git a/server/admin/internal/repository/money_scope_repository.go b/server/admin/internal/repository/money_scope_repository.go index 20606d2b..af7366a5 100644 --- a/server/admin/internal/repository/money_scope_repository.go +++ b/server/admin/internal/repository/money_scope_repository.go @@ -28,13 +28,25 @@ func (access MoneyAccess) Allows(appCode string, regionID int64) bool { if appctx.Normalize(scope.AppCode) != appCode { continue } - if scope.RegionID == 0 || scope.RegionID == regionID { + if scope.RegionID == 0 || RegionIDLooseEqual(scope.RegionID, regionID) { return true } } return false } +// RegionIDLooseEqual 对齐 legacy 大区 ID 的浮点圆整误差：雪花/合成 ID 超出 JS 安全整数， +// 前端写回的 scope/筛选值可能是 float64 圆整后的数（相邻 ID 间距远大于圆整粒度，不会误碰）。 +func RegionIDLooseEqual(left int64, right int64) bool { + if left == right { + return true + } + if left == 0 || right == 0 { + return false + } + return int64(float64(left)) == right || left == int64(float64(right)) +} + func (access MoneyAccess) AppCodes() []string { if access.All { return nil diff --git a/server/admin/internal/repository/money_scope_repository_test.go b/server/admin/internal/repository/money_scope_repository_test.go index 87c6d143..f4702817 100644 --- a/server/admin/internal/repository/money_scope_repository_test.go +++ b/server/admin/internal/repository/money_scope_repository_test.go @@ -75,3 +75,32 @@ func newRepositorySQLMock(t *testing.T) (*Store, sqlmock.Sqlmock, func()) { _ = sqlDB.Close() } } + +func TestRegionIDLooseEqual(t *testing.T) { + exact := int64(2049039732077690882) + rounded := int64(float64(exact)) + if !RegionIDLooseEqual(exact, rounded) || !RegionIDLooseEqual(rounded, exact) { + t.Fatalf("expected rounded/exact pair to match") + } + if !RegionIDLooseEqual(7, 7) { + t.Fatalf("expected exact small ids to match") + } + if RegionIDLooseEqual(7, 8) { + t.Fatalf("small ids must not loosely match") + } + if RegionIDLooseEqual(0, rounded) || RegionIDLooseEqual(exact, 0) { + t.Fatalf("zero must never loosely match") + } +} + +func TestMoneyAccessAllowsRoundedRegion(t *testing.T) { + exact := int64(2049039732077690882) + rounded := int64(float64(exact)) + access := MoneyAccess{UserID: 1, Scopes: []model.UserMoneyScope{{UserID: 1, AppCode: "aslan", RegionID: rounded}}} + if !access.Allows("aslan", exact) { + t.Fatalf("rounded scope should allow exact region id") + } + if access.Allows("aslan", exact+999) { + t.Fatalf("unrelated region must stay denied") + } +}