hyapp-server/server/admin/internal/modules/databi/service.go

package databi

import (
	"context"
	"fmt"
	"log/slog"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"hyapp-admin-server/internal/appctx"
	"hyapp-admin-server/internal/integration/userclient"
	"hyapp-admin-server/internal/model"
	"hyapp-admin-server/internal/modules/appregistry"
	"hyapp-admin-server/internal/modules/dashboard"
	"hyapp-admin-server/internal/modules/payment"
	"hyapp-admin-server/internal/repository"
)

const (
	appKindHyapp       = "hyapp"
	appKindLegacy      = "legacy"
	unknownCountryName = "未知国家"

	overviewConcurrency = 4
)

var appTrackingFunnelSupportedApps = map[string]struct{}{
	"fami":  {},
	"huwaa": {},
	"lalu":  {},
}

// LegacyRegionCatalogSource 由 payment.MongoMoneyRegionSource 实现；
// 社交 BI 与财务范围共用同一套 legacy 区域目录与合成 region_id 口径。
type LegacyRegionCatalogSource interface {
	RegionCatalog(ctx context.Context) ([]payment.RegionCatalogEntry, error)
}

// LegacyAppDescriptor 描述一个外接 App（Yumi/Aslan），来源于 dashboard 外部源配置。
type LegacyAppDescriptor struct {
	AppCode string
	AppName string
	LogoURL string
}

type Service struct {
	store                *repository.Store
	dashboards           *dashboard.DashboardService
	apps                 *appregistry.Service
	user                 userclient.Client
	financeRecharge      payment.FinanceRechargeOverviewSource
	luckyGiftStats       LuckyGiftDailyStatsSource
	luckyGiftTimeout     time.Duration
	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)

func WithLegacyApps(apps ...LegacyAppDescriptor) ServiceOption {
	return func(s *Service) {
		for _, app := range apps {
			appCode := appctx.Normalize(app.AppCode)
			if appCode == "" {
				continue
			}
			app.AppCode = appCode
			s.legacyApps = append(s.legacyApps, app)
		}
	}
}

// WithFinanceRechargeOverviewSource 强制 Social BI 的充值渠道复用 Finance 口径；
// 不在 Databi 内复制订单、H5 或 legacy 账单 SQL，避免两个页面再次漂移。
func WithFinanceRechargeOverviewSource(source payment.FinanceRechargeOverviewSource) ServiceOption {
	return func(s *Service) {
		s.financeRecharge = source
	}
}

func WithLegacyRegionCatalogs(catalogs ...LegacyRegionCatalogSource) ServiceOption {
	return func(s *Service) {
		for _, catalog := range catalogs {
			if catalog != nil {
				s.legacyRegionCatalogs = append(s.legacyRegionCatalogs, catalog)
			}
		}
	}
}

func NewService(store *repository.Store, dashboards *dashboard.DashboardService, apps *appregistry.Service, user userclient.Client, options ...ServiceOption) *Service {
	service := &Service{
		store:      store,
		dashboards: dashboards,
		apps:       apps,
		user:       user,
	}
	for _, option := range options {
		if option != nil {
			option(service)
		}
	}
	return service
}

type AppInfo struct {
	AppCode string `json:"app_code"`
	AppName string `json:"app_name"`
	LogoURL string `json:"logo_url"`
	Kind    string `json:"kind"`
}

type RegionInfo struct {
	AppCode    string   `json:"app_code"`
	RegionID   int64    `json:"region_id"`
	RegionCode string   `json:"region_code"`
	RegionName string   `json:"region_name"`
	Countries  []string `json:"countries"`
}

type ScopeInfo struct {
	AppCode  string `json:"app_code"`
	RegionID int64  `json:"region_id"`
}

type OperatorInfo struct {
	UserID  uint        `json:"user_id"`
	Name    string      `json:"name"`
	Account string      `json:"account"`
	Team    string      `json:"team"`
	TeamID  *uint       `json:"team_id"`
	Status  string      `json:"status"`
	Scopes  []ScopeInfo `json:"scopes"`
}

type AccessInfo struct {
	All    bool        `json:"all"`
	Scopes []ScopeInfo `json:"scopes"`
}

type MasterData struct {
	Apps      []AppInfo      `json:"apps"`
	Regions   []RegionInfo   `json:"regions"`
	Operators []OperatorInfo `json:"operators"`
	Access    AccessInfo     `json:"access"`
}

// Master 返回社交 BI 的筛选主数据：当前用户可见的 App、区域目录、运营人员及各自负责范围。
func (s *Service) Master(ctx context.Context, access repository.MoneyAccess, actorUserID uint, viewAll bool) (*MasterData, error) {
	apps, err := s.listAllowedApps(ctx, nil, access)
	if err != nil {
		return nil, err
	}
	catalog, err := s.regionCatalog(ctx, apps)
	if err != nil {
		return nil, err
	}
	regions := []RegionInfo{}
	for _, app := range apps {
		allowAll, allowedIDs := allowedRegions(access, app.AppCode)
		allowedSet := int64Set(normalizeRegionIDs(catalog[app.AppCode], allowedIDs))
		for _, region := range catalog[app.AppCode] {
			if !allowAll {
				if _, ok := allowedSet[region.RegionID]; !ok {
					continue
				}
			}
			regions = append(regions, region)
		}
	}

	operators, err := s.listOperators(actorUserID, viewAll)
	if err != nil {
		return nil, err
	}
	return &MasterData{
		Apps:      apps,
		Regions:   regions,
		Operators: operators,
		Access:    AccessInfo{All: access.All, Scopes: scopeInfos(access.Scopes)},
	}, nil
}

func (s *Service) listOperators(actorUserID uint, viewAll bool) ([]OperatorInfo, error) {
	users, scopesByUser, err := s.store.ListMoneyScopeOperators()
	if err != nil {
		return nil, err
	}
	operators := make([]OperatorInfo, 0, len(users))
	for _, user := range users {
		if !viewAll && user.ID != actorUserID {
			continue
		}
		scopes := scopesByUser[user.ID]
		operators = append(operators, OperatorInfo{
			UserID:  user.ID,
			Name:    user.Name,
			Account: user.Username,
			Team:    user.Team,
			TeamID:  user.TeamID,
			Status:  user.Status,
			Scopes:  scopeInfos(scopes),
		})
	}
	return operators, nil
}

type OverviewQuery struct {
	StatTZ    string
	StartMS   int64
	EndMS     int64
	AppCodes  []string
	RegionID  int64
	RegionIDs []int64
}

type AppOverview struct {
	AppCode               string           `json:"app_code"`
	AppName               string           `json:"app_name"`
	Kind                  string           `json:"kind"`
	Restricted            bool             `json:"restricted"`
	AllowedRegionIDs      []int64          `json:"allowed_region_ids,omitempty"`
	Total                 map[string]any   `json:"total"`
	DailySeries           []map[string]any `json:"daily_series"`
	RegionBreakdown       []map[string]any `json:"region_breakdown"`
	DailyRegionBreakdown  []map[string]any `json:"daily_region_breakdown"`
	CountryBreakdown      []map[string]any `json:"country_breakdown"`
	DailyCountryBreakdown []map[string]any `json:"daily_country_breakdown"`
	ReportMetricSources   any              `json:"report_metric_sources,omitempty"`
	UpdatedAtMS           any              `json:"updated_at_ms,omitempty"`
	Error                 string           `json:"error,omitempty"`
}

type OverviewResult struct {
	Apps   []AppOverview `json:"apps"`
	Access AccessInfo    `json:"access"`
}

type FunnelQuery struct {
	StatTZ   string
	StartMS  int64
	EndMS    int64
	AppCodes []string
	RegionID int64
}

type AppFunnel struct {
	AppCode          string           `json:"app_code"`
	AppName          string           `json:"app_name"`
	Kind             string           `json:"kind"`
	Restricted       bool             `json:"restricted"`
	AllowedRegionIDs []int64          `json:"allowed_region_ids,omitempty"`
	Steps            []map[string]any `json:"steps"`
	Cohorts          []map[string]any `json:"cohorts"`
	Totals           map[string]any   `json:"totals"`
	UpdatedAtMS      int64            `json:"updated_at_ms"`
	Error            string           `json:"error,omitempty"`
}

type FunnelResult struct {
	Apps   []AppFunnel `json:"apps"`
	Access AccessInfo  `json:"access"`
}

type RequirementsQuery struct {
	StatTZ      string
	StartMS     int64
	EndMS       int64
	AppCodes    []string
	RegionID    int64
	RegionIDs   []int64
	Section     string
	UserRole    string
	PayerType   string
	NewUserType string
}

type AppRequirements struct {
	AppCode          string           `json:"app_code"`
	AppName          string           `json:"app_name"`
	Kind             string           `json:"kind"`
	Restricted       bool             `json:"restricted"`
	AllowedRegionIDs []int64          `json:"allowed_region_ids,omitempty"`
	Sections         []map[string]any `json:"sections"`
	UpdatedAtMS      int64            `json:"updated_at_ms"`
	Error            string           `json:"error,omitempty"`
}

type RequirementsResult struct {
	Apps   []AppRequirements `json:"apps"`
	Access AccessInfo        `json:"access"`
}

// Overview 服务端并发聚合各 App 的统计总览：hyapp App 走 statistics-service，
// legacy App 走 dashboard 外部源；国家行按区域目录卷积出大区行，并按用户数据范围裁剪。
func (s *Service) Overview(ctx context.Context, access repository.MoneyAccess, query OverviewQuery) (*OverviewResult, error) {
	registeredApps, err := s.listApps(ctx)
	if err != nil {
		return nil, err
	}
	apps := filterAllowedApps(registeredApps, query.AppCodes, access)
	catalog, err := s.regionCatalog(ctx, apps)
	if err != nil {
		return nil, err
	}
	// 国家主数据属于 App 业务域；即使 country_id 是物理全局自增，也必须按当前可查询 App 建目录，
	// 否则跨 App 污染会被另一个 App 的主数据包装成合法国家，掩盖上游事实错误。
	countryDirectories := s.countryDirectories(ctx, apps)

	results := make([]AppOverview, len(apps))
	var wg sync.WaitGroup
	semaphore := make(chan struct{}, overviewConcurrency)
	for index, app := range apps {
		wg.Add(1)
		go func(index int, app AppInfo) {
			defer wg.Done()
			semaphore <- struct{}{}
			defer func() { <-semaphore }()
			results[index] = s.appOverview(ctx, app, catalog[app.AppCode], countryDirectories[app.AppCode], access, query)
		}(index, app)
	}
	wg.Wait()

	return &OverviewResult{
		Apps:   results,
		Access: AccessInfo{All: access.All, Scopes: scopeInfos(access.Scopes)},
	}, nil
}

func (s *Service) Funnel(ctx context.Context, access repository.MoneyAccess, query FunnelQuery) (*FunnelResult, error) {
	apps, err := s.listAllowedApps(ctx, query.AppCodes, access)
	if err != nil {
		return nil, err
	}
	apps = filterAppTrackingFunnelApps(apps)
	catalog, err := s.regionCatalog(ctx, apps)
	if err != nil {
		return nil, err
	}

	results := make([]AppFunnel, len(apps))
	var wg sync.WaitGroup
	semaphore := make(chan struct{}, overviewConcurrency)
	for index, app := range apps {
		wg.Add(1)
		go func(index int, app AppInfo) {
			defer wg.Done()
			semaphore <- struct{}{}
			defer func() { <-semaphore }()
			results[index] = s.appFunnel(ctx, app, catalog[app.AppCode], access, query)
		}(index, app)
	}
	wg.Wait()

	return &FunnelResult{
		Apps:   results,
		Access: AccessInfo{All: access.All, Scopes: scopeInfos(access.Scopes)},
	}, nil
}

func (s *Service) Requirements(ctx context.Context, access repository.MoneyAccess, query RequirementsQuery) (*RequirementsResult, error) {
	apps, err := s.listAllowedApps(ctx, query.AppCodes, access)
	if err != nil {
		return nil, err
	}
	catalog, err := s.regionCatalog(ctx, apps)
	if err != nil {
		return nil, err
	}

	results := make([]AppRequirements, len(apps))
	var wg sync.WaitGroup
	semaphore := make(chan struct{}, overviewConcurrency)
	for index, app := range apps {
		wg.Add(1)
		go func(index int, app AppInfo) {
			defer wg.Done()
			semaphore <- struct{}{}
			defer func() { <-semaphore }()
			results[index] = s.appRequirements(ctx, app, catalog[app.AppCode], access, query)
		}(index, app)
	}
	wg.Wait()

	return &RequirementsResult{
		Apps:   results,
		Access: AccessInfo{All: access.All, Scopes: scopeInfos(access.Scopes)},
	}, nil
}

func (s *Service) appRequirements(ctx context.Context, app AppInfo, regions []RegionInfo, access repository.MoneyAccess, query RequirementsQuery) AppRequirements {
	out := AppRequirements{
		AppCode:  app.AppCode,
		AppName:  app.AppName,
		Kind:     app.Kind,
		Sections: []map[string]any{},
	}
	if app.Kind != appKindHyapp {
		out.Error = "该 App 暂未接入数据需求宽表"
		return out
	}
	allowAll, allowedIDs := allowedRegions(access, app.AppCode)
	allowedIDs = normalizeRegionIDs(regions, allowedIDs)
	if !allowAll && len(allowedIDs) == 0 {
		out.Error = "没有该 App 的数据权限"
		return out
	}
	requestRegionIDs := normalizeRegionIDs(regions, query.RegionIDs)
	if len(query.RegionIDs) > 0 && len(requestRegionIDs) == 0 {
		out.Error = "该 App 未配置所选区域"
		return out
	}
	if query.RegionID > 0 && !access.Allows(app.AppCode, query.RegionID) {
		out.Error = "没有该区域的数据权限"
		return out
	}
	for _, regionID := range requestRegionIDs {
		if !access.Allows(app.AppCode, regionID) {
			out.Error = "没有该区域的数据权限"
			return out
		}
	}
	if query.RegionID > 0 && !regionInCatalog(regions, query.RegionID) {
		out.Error = "该 App 未配置此区域"
		return out
	}
	restricted := !allowAll && query.RegionID == 0 && len(requestRegionIDs) == 0
	regionIDs := []int64(nil)
	if restricted {
		// 用户没有全局范围时，admin 只把允许的大区 ID 传给 statistics-service；查询层仍只读聚合投影，不回扫业务库补权限。
		out.Restricted = true
		out.AllowedRegionIDs = allowedIDs
		regionIDs = allowedIDs
	} else if len(requestRegionIDs) > 0 {
		// 顶栏多选区域时保留多区域筛选；单区域仍走 region_id 以兼容旧统计入口。
		regionIDs = requestRegionIDs
	}
	if requirementsOnlyLuckyGiftV3(query.Section) {
		// 只要幸运礼物V3板块时跳过 statistics-service：它不认识该 section key，
		// 透传会触发"未知 key 回退全量六板块"的最重查询路径。
		if s.luckyGiftStats == nil {
			out.Error = "幸运礼物V3统计源未配置"
			return out
		}
		luckySection, err := s.luckyGiftV3Section(ctx, app.AppCode, query, luckyGiftV3RegionIDs(query, regions, regionIDs))
		if err != nil {
			out.Error = fmt.Sprintf("获取幸运礼物V3统计失败: %v", err)
			return out
		}
		out.Sections = append(out.Sections, luckySection)
		out.UpdatedAtMS = time.Now().UnixMilli()
		return out
	}
	requirements, err := s.dashboards.SocialRequirements(ctx, dashboard.SocialRequirementsQuery{
		AppCode:     app.AppCode,
		StatTZ:      query.StatTZ,
		StartMS:     query.StartMS,
		EndMS:       query.EndMS,
		RegionID:    query.RegionID,
		RegionIDs:   regionIDs,
		Section:     query.Section,
		UserRole:    query.UserRole,
		PayerType:   query.PayerType,
		NewUserType: query.NewUserType,
	})
	if err != nil {
		out.Error = fmt.Sprintf("获取数据需求指标失败: %v", err)
		return out
	}
	out.Sections = anyRowSlice(requirements["sections"])
	if s.luckyGiftStats != nil && requirementsIncludesLuckyGiftV3(query.Section) {
		luckySection, err := s.luckyGiftV3Section(ctx, app.AppCode, query, luckyGiftV3RegionIDs(query, regions, regionIDs))
		if err != nil {
			out.Error = fmt.Sprintf("获取幸运礼物V3统计失败: %v", err)
			return out
		}
		out.Sections = append(out.Sections, luckySection)
	}
	if s.financeRecharge != nil && requirementsIncludesRevenue(query.Section) {
		if requirementsHasCohortFilter(query) {
			// Finance 不具备角色/付费身份/新老用户维度；筛选后必须把金额标为不可用，
			// 不能把全量 Finance 金额灌进 cohort，也不能静默展示旧宽表金额冒充 Finance。
			markFinanceRechargeRequirementsUnavailable(out.Sections)
		} else {
			financeRegionIDs := regionIDs
			if query.RegionID > 0 {
				financeRegionIDs = []int64{normalizeRegionID(regions, query.RegionID)}
			}
			financeOverview, financeErr := s.financeRecharge.FinanceRechargeOverview(ctx, payment.FinanceRechargeOverviewQuery{
				AppCode:         app.AppCode,
				StartAtMS:       query.StartMS,
				EndAtMS:         query.EndMS,
				ScopeRegionIDs:  financeRegionIDs,
				IncludeRegions:  false,
				TZOffsetMinutes: financeRechargeTZOffsetMinutes,
			})
			if financeErr != nil {
				out.Error = fmt.Sprintf("获取 Finance 充值统计失败: %v", financeErr)
				return out
			}
			overlayFinanceRechargeRequirements(out.Sections, financeOverview)
		}
	}
	out.UpdatedAtMS = time.Now().UnixMilli()
	return out
}

func (s *Service) appFunnel(ctx context.Context, app AppInfo, regions []RegionInfo, access repository.MoneyAccess, query FunnelQuery) AppFunnel {
	out := AppFunnel{
		AppCode: app.AppCode,
		AppName: app.AppName,
		Kind:    app.Kind,
		Steps:   []map[string]any{},
		Cohorts: []map[string]any{},
		Totals:  map[string]any{},
	}
	if app.Kind != appKindHyapp {
		out.Error = "该 App 暂未接入 App 埋点漏斗"
		return out
	}
	if !supportsAppTrackingFunnel(app.AppCode) {
		out.Error = "该 App 暂未接入 App 埋点漏斗"
		return out
	}
	allowAll, allowedIDs := allowedRegions(access, app.AppCode)
	allowedIDs = normalizeRegionIDs(regions, allowedIDs)
	if !allowAll && len(allowedIDs) == 0 {
		out.Error = "没有该 App 的数据权限"
		return out
	}
	if query.RegionID > 0 && !access.Allows(app.AppCode, query.RegionID) {
		out.Error = "没有该区域的数据权限"
		return out
	}
	if query.RegionID > 0 && !regionInCatalog(regions, query.RegionID) {
		out.Error = "该 App 未配置此区域"
		return out
	}
	restricted := !allowAll && query.RegionID == 0
	if restricted {
		out.Restricted = true
		out.AllowedRegionIDs = allowedIDs
	}
	regionIDs := []int64(nil)
	if restricted {
		regionIDs = allowedIDs
	}
	funnel, err := s.dashboards.AppTrackingFunnel(ctx, dashboard.AppTrackingFunnelQuery{
		AppCode:   app.AppCode,
		StatTZ:    query.StatTZ,
		StartMS:   query.StartMS,
		EndMS:     query.EndMS,
		RegionID:  query.RegionID,
		RegionIDs: regionIDs,
	})
	if err != nil {
		out.Error = fmt.Sprintf("获取埋点漏斗失败: %v", err)
		return out
	}
	out.Steps = anyRowSlice(funnel["steps"])
	out.Cohorts = anyRowSlice(funnel["cohorts"])
	if totals, ok := funnel["totals"].(map[string]any); ok {
		out.Totals = totals
	}
	out.UpdatedAtMS = time.Now().UnixMilli()
	return out
}

func filterAppTrackingFunnelApps(apps []AppInfo) []AppInfo {
	out := make([]AppInfo, 0, len(apps))
	for _, app := range apps {
		if app.Kind == appKindHyapp && supportsAppTrackingFunnel(app.AppCode) {
			out = append(out, app)
		}
	}
	return out
}

func supportsAppTrackingFunnel(appCode string) bool {
	_, ok := appTrackingFunnelSupportedApps[appctx.Normalize(appCode)]
	return ok
}

func (s *Service) appOverview(ctx context.Context, app AppInfo, regions []RegionInfo, countries map[int64]countryDirectoryEntry, access repository.MoneyAccess, query OverviewQuery) AppOverview {
	out := AppOverview{
		AppCode:               app.AppCode,
		AppName:               app.AppName,
		Kind:                  app.Kind,
		Total:                 map[string]any{},
		DailySeries:           []map[string]any{},
		RegionBreakdown:       []map[string]any{},
		DailyRegionBreakdown:  []map[string]any{},
		CountryBreakdown:      []map[string]any{},
		DailyCountryBreakdown: []map[string]any{},
	}
	allowAll, allowedIDs := allowedRegions(access, app.AppCode)
	allowedIDs = normalizeRegionIDs(regions, allowedIDs)
	if !allowAll && len(allowedIDs) == 0 {
		out.Error = "没有该 App 的数据权限"
		return out
	}
	if query.RegionID > 0 && !access.Allows(app.AppCode, query.RegionID) {
		out.Error = "没有该区域的数据权限"
		return out
	}
	if query.RegionID > 0 && !regionInCatalog(regions, query.RegionID) {
		out.Error = "该 App 未配置此区域"
		return out
	}
	requestRegionIDs := normalizeRegionIDs(regions, query.RegionIDs)
	if len(query.RegionIDs) > 0 && len(requestRegionIDs) == 0 {
		out.Error = "该 App 未配置所选区域"
		return out
	}
	for _, regionID := range requestRegionIDs {
		if !access.Allows(app.AppCode, regionID) {
			out.Error = "没有该区域的数据权限"
			return out
		}
	}

	overview, err := s.statisticsOverview(ctx, dashboard.StatisticsQuery{
		AppCode:       app.AppCode,
		StatTZ:        query.StatTZ,
		StartMS:       query.StartMS,
		EndMS:         query.EndMS,
		SeriesStartMS: query.StartMS,
		SeriesEndMS:   query.EndMS,
		RegionID:      query.RegionID,
	})
	if err != nil {
		out.Error = fmt.Sprintf("获取统计数据失败: %v", err)
		return out
	}

	// statisticsOverview 会短时缓存并共享底层 map；补展示字段时必须复制行，避免并发请求互相改写缓存。
	preserveSourceNames := app.Kind == appKindLegacy
	countryRows := enrichCountryRows(anyRowSlice(overview["country_breakdown"]), countries, preserveSourceNames)
	dailyCountryRows := enrichCountryRows(anyRowSlice(overview["daily_country_breakdown"]), countries, preserveSourceNames)
	resolve := regionResolver(regions)

	restricted := !allowAll && query.RegionID == 0 && len(requestRegionIDs) == 0
	filterRegionIDs := requestRegionIDs
	if restricted {
		filterRegionIDs = allowedIDs
		out.Restricted = true
		out.AllowedRegionIDs = allowedIDs
	}
	if query.RegionID == 0 && len(filterRegionIDs) > 0 {
		// statistics-service 目前只有单区域参数；多区域在已返回的国家投影上裁剪并重算非充值指标，
		// 充值随后由 Finance 按同一组区域重新覆盖，避免顶栏多选退回旧统计口径。
		allowedSet := int64Set(filterRegionIDs)
		countryRows = filterRowsByRegion(countryRows, resolve, allowedSet)
		dailyCountryRows = filterRowsByRegion(dailyCountryRows, resolve, allowedSet)

		totalAcc := newMetricAccumulator()
		for _, row := range countryRows {
			totalAcc.addRow(row)
		}
		out.Total = totalAcc.toMap()
		out.DailySeries = rollupDaily(dailyCountryRows)
	} else {
		out.Total = topLevelMetrics(overview)
		out.DailySeries = anyRowSlice(overview["daily_series"])
	}

	out.RegionBreakdown = rollupByRegion(countryRows, resolve)
	out.DailyRegionBreakdown = rollupDailyByRegion(dailyCountryRows, resolve)
	out.CountryBreakdown = countryRows
	out.DailyCountryBreakdown = dailyCountryRows
	out.ReportMetricSources = overview["report_metric_sources"]
	out.UpdatedAtMS = overview["updated_at_ms"]
	if s.financeRecharge != nil {
		// Social BI 的充值卡片、趋势和区域下钻必须与 Finance 同源；
		// Finance 查询失败时整 App 显式报错，不能静默回退到旧后台库存流水。
		if err := s.applyFinanceRechargeOverview(ctx, &out, query, regions, restricted, allowedIDs); err != nil {
			out.Error = fmt.Sprintf("获取 Finance 充值统计失败: %v", err)
		}
	}
	return out
}

type KpiQuery struct {
	StatTZ         string
	StartMS        int64
	EndMS          int64
	PeriodMonth    string
	AppCodes       []string
	OperatorUserID uint
	ActorUserID    uint
	ViewAll        bool
}

type KpiRow struct {
	OperatorUserID        uint   `json:"operator_user_id"`
	OperatorName          string `json:"operator_name"`
	OperatorAccount       string `json:"operator_account"`
	Team                  string `json:"team"`
	TeamID                *uint  `json:"team_id"`
	AppCode               string `json:"app_code"`
	AppName               string `json:"app_name"`
	RegionID              int64  `json:"region_id"`
	RegionCode            string `json:"region_code"`
	RegionName            string `json:"region_name"`
	RangeRechargeUSDMinor *int64 `json:"range_recharge_usd_minor"`
	RangeNewUsers         *int64 `json:"range_new_users"`
	RangeActiveUsers      *int64 `json:"range_active_users"`
	MonthRechargeUSDMinor *int64 `json:"month_recharge_usd_minor"`
	DataError             string `json:"data_error,omitempty"`
}

// KpiOperatorAppRow 是 人×App 级小计：该运营在此 App 下全部负责区域的充值合计。
type KpiOperatorAppRow struct {
	OperatorUserID        uint   `json:"operator_user_id"`
	OperatorName          string `json:"operator_name"`
	OperatorAccount       string `json:"operator_account"`
	Team                  string `json:"team"`
	TeamID                *uint  `json:"team_id"`
	AppCode               string `json:"app_code"`
	AppName               string `json:"app_name"`
	RegionCount           int    `json:"region_count"`
	RegionNames           string `json:"region_names"`
	WholeAppScope         bool   `json:"whole_app_scope"`
	RangeRechargeUSDMinor *int64 `json:"range_recharge_usd_minor"`
	RangeNewUsers         *int64 `json:"range_new_users"`
	MonthRechargeUSDMinor *int64 `json:"month_recharge_usd_minor"`
	DataError             string `json:"data_error,omitempty"`
}

type KpiSummary struct {
	RangeRechargeUSDMinor int64 `json:"range_recharge_usd_minor"`
	MonthRechargeUSDMinor int64 `json:"month_recharge_usd_minor"`
	OperatorCount         int   `json:"operator_count"`
}

type KpiResult struct {
	PeriodMonth     string              `json:"period_month"`
	MonthStartMS    int64               `json:"month_start_ms"`
	MonthEndMS      int64               `json:"month_end_ms"`
	Items           []KpiRow            `json:"items"`
	OperatorAppRows []KpiOperatorAppRow `json:"operator_app_rows"`
	Summary         KpiSummary          `json:"summary"`
	AppErrors       map[string]any      `json:"app_errors,omitempty"`
}

// Kpi 按运营人员负责范围（App×区域）聚合充值/新增；无全量数据范围时只返回本人行。
func (s *Service) Kpi(ctx context.Context, access repository.MoneyAccess, query KpiQuery) (*KpiResult, error) {
	users, scopesByUser, err := s.store.ListMoneyScopeOperators()
	if err != nil {
		return nil, err
	}
	operators := make([]model.User, 0, len(users))
	for _, user := range users {
		if !query.ViewAll && user.ID != query.ActorUserID {
			continue
		}
		if query.OperatorUserID > 0 && user.ID != query.OperatorUserID {
			continue
		}
		operators = append(operators, user)
	}

	requestedApps := map[string]struct{}{}
	for _, appCode := range query.AppCodes {
		appCode = appctx.Normalize(appCode)
		if appCode != "" {
			requestedApps[appCode] = struct{}{}
		}
	}
	appCodeSet := map[string]struct{}{}
	for _, user := range operators {
		for _, scope := range scopesByUser[user.ID] {
			appCode := appctx.Normalize(scope.AppCode)
			if appCode == "" {
				continue
			}
			if len(requestedApps) > 0 {
				if _, ok := requestedApps[appCode]; !ok {
					continue
				}
			}
			appCodeSet[appCode] = struct{}{}
		}
	}

	apps, err := s.listApps(ctx)
	if err != nil {
		return nil, err
	}
	appsByCode := map[string]AppInfo{}
	relevantApps := []AppInfo{}
	for _, app := range apps {
		appsByCode[app.AppCode] = app
		if _, ok := appCodeSet[app.AppCode]; ok {
			relevantApps = append(relevantApps, app)
		}
	}
	catalog, err := s.regionCatalog(ctx, relevantApps)
	if err != nil {
		return nil, err
	}

	location, err := time.LoadLocation(firstNonEmptyString(query.StatTZ, "Asia/Shanghai"))
	if err != nil {
		return nil, fmt.Errorf("load kpi timezone %s: %w", query.StatTZ, err)
	}
	periodMonth, monthStart, monthEnd, err := resolveKpiMonth(query.PeriodMonth, query.EndMS, location)
	if err != nil {
		return nil, err
	}

	dataByApp := make(map[string]*appKpiData, len(relevantApps))
	var mu sync.Mutex
	var wg sync.WaitGroup
	semaphore := make(chan struct{}, overviewConcurrency)
	for _, app := range relevantApps {
		wg.Add(1)
		go func(app AppInfo) {
			defer wg.Done()
			semaphore <- struct{}{}
			defer func() { <-semaphore }()
			data := &appKpiData{}
			resolve := regionResolver(catalog[app.AppCode])
			// 展示区间与当月两个窗口互不依赖，并行取；同参数窗口会被 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 {
					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
			mu.Unlock()
		}(app)
	}
	wg.Wait()

	rows := []KpiRow{}
	appErrors := map[string]any{}
	for _, user := range operators {
		for _, scope := range scopesByUser[user.ID] {
			appCode := appctx.Normalize(scope.AppCode)
			if len(requestedApps) > 0 {
				if _, ok := requestedApps[appCode]; !ok {
					continue
				}
			}
			app, ok := appsByCode[appCode]
			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,
				OperatorAccount: user.Username,
				Team:            user.Team,
				TeamID:          user.TeamID,
				AppCode:         appCode,
				AppName:         app.AppName,
				RegionID:        regionID,
			}
			row.RegionCode, row.RegionName = regionDisplay(catalog[appCode], regionID)
			data := dataByApp[appCode]
			if data == nil {
				row.DataError = "该 App 未接入统计数据"
			} else if data.err != nil {
				row.DataError = fmt.Sprintf("获取统计数据失败: %v", data.err)
				appErrors[appCode] = row.DataError
			} else {
				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")
				row.MonthRechargeUSDMinor = metricPointer(monthMetrics, "recharge_usd_minor")
			}
			rows = append(rows, row)
		}
	}
	sort.Slice(rows, func(i, j int) bool {
		left, right := rows[i], rows[j]
		leftMonth := int64(0)
		if left.MonthRechargeUSDMinor != nil {
			leftMonth = *left.MonthRechargeUSDMinor
		}
		rightMonth := int64(0)
		if right.MonthRechargeUSDMinor != nil {
			rightMonth = *right.MonthRechargeUSDMinor
		}
		if leftMonth != rightMonth {
			return leftMonth > rightMonth
		}
		if left.OperatorUserID != right.OperatorUserID {
			return left.OperatorUserID < right.OperatorUserID
		}
		if left.AppCode != right.AppCode {
			return left.AppCode < right.AppCode
		}
		return left.RegionID < right.RegionID
	})

	operatorAppRows := buildOperatorAppRows(rows)
	summary := buildKpiSummary(rows, operatorAppRows)

	result := &KpiResult{
		PeriodMonth:     periodMonth,
		MonthStartMS:    monthStart.UnixMilli(),
		MonthEndMS:      monthEnd.UnixMilli(),
		Items:           rows,
		OperatorAppRows: operatorAppRows,
		Summary:         summary,
	}
	if len(appErrors) > 0 {
		result.AppErrors = appErrors
	}
	return result, nil
}

type appKpiData struct {
	rangeTotal  map[string]any
	rangeRegion map[int64]map[string]any
	monthTotal  map[string]any
	monthRegion map[int64]map[string]any
	err         error
}

func nullableAdd(sum *int64, value *int64) *int64 {
	if value == nil {
		return sum
	}
	if sum == nil {
		out := *value
		return &out
	}
	out := *sum + *value
	return &out
}

// buildOperatorAppRows 把 人×App×区域 明细行归并成 人×App 小计：
// 该运营有整 App 范围（region 0）时直接取那一行（其值即整 App 口径），否则各区域可空求和。
func buildOperatorAppRows(rows []KpiRow) []KpiOperatorAppRow {
	type group struct {
		row         KpiOperatorAppRow
		regionNames []string
		wholeRow    *KpiRow
	}
	groups := map[string]*group{}
	order := []string{}
	for index := range rows {
		row := rows[index]
		key := strconv.FormatUint(uint64(row.OperatorUserID), 10) + "|" + row.AppCode
		item := groups[key]
		if item == nil {
			item = &group{row: KpiOperatorAppRow{
				AppCode:         row.AppCode,
				AppName:         row.AppName,
				OperatorAccount: row.OperatorAccount,
				OperatorName:    row.OperatorName,
				OperatorUserID:  row.OperatorUserID,
				Team:            row.Team,
				TeamID:          row.TeamID,
			}}
			groups[key] = item
			order = append(order, key)
		}
		item.row.RegionCount++
		item.regionNames = append(item.regionNames, row.RegionName)
		if row.DataError != "" && item.row.DataError == "" {
			item.row.DataError = row.DataError
		}
		if row.RegionID == 0 {
			item.row.WholeAppScope = true
			item.wholeRow = &rows[index]
			continue
		}
		item.row.RangeRechargeUSDMinor = nullableAdd(item.row.RangeRechargeUSDMinor, row.RangeRechargeUSDMinor)
		item.row.RangeNewUsers = nullableAdd(item.row.RangeNewUsers, row.RangeNewUsers)
		item.row.MonthRechargeUSDMinor = nullableAdd(item.row.MonthRechargeUSDMinor, row.MonthRechargeUSDMinor)
	}

	out := make([]KpiOperatorAppRow, 0, len(order))
	for _, key := range order {
		item := groups[key]
		if item.wholeRow != nil {
			// 整 App 范围行的取数已是整 App 口径，区域行是它的子集，避免相加重复计数。
			item.row.RangeRechargeUSDMinor = item.wholeRow.RangeRechargeUSDMinor
			item.row.RangeNewUsers = item.wholeRow.RangeNewUsers
			item.row.MonthRechargeUSDMinor = item.wholeRow.MonthRechargeUSDMinor
		}
		item.row.RegionNames = joinRegionNames(item.regionNames)
		out = append(out, item.row)
	}
	sort.Slice(out, func(i, j int) bool {
		left := int64(0)
		if out[i].MonthRechargeUSDMinor != nil {
			left = *out[i].MonthRechargeUSDMinor
		}
		right := int64(0)
		if out[j].MonthRechargeUSDMinor != nil {
			right = *out[j].MonthRechargeUSDMinor
		}
		if left != right {
			return left > right
		}
		if out[i].OperatorUserID != out[j].OperatorUserID {
			return out[i].OperatorUserID < out[j].OperatorUserID
		}
		return out[i].AppCode < out[j].AppCode
	})
	return out
}

func joinRegionNames(names []string) string {
	if len(names) <= 3 {
		return strings.Join(names, " / ")
	}
	return strings.Join(names[:3], " / ") + fmt.Sprintf(" 等%d个", len(names))
}

// buildKpiSummary 只服务列表头部元信息；充值合计按人×App 小计相加，避免同一运营多区域重复。
func buildKpiSummary(rows []KpiRow, operatorAppRows []KpiOperatorAppRow) KpiSummary {
	summary := KpiSummary{}
	operatorSeen := map[uint]struct{}{}
	for _, row := range rows {
		operatorSeen[row.OperatorUserID] = struct{}{}
	}
	summary.OperatorCount = len(operatorSeen)

	for _, row := range operatorAppRows {
		if row.RangeRechargeUSDMinor != nil {
			summary.RangeRechargeUSDMinor += *row.RangeRechargeUSDMinor
		}
		if row.MonthRechargeUSDMinor != nil {
			summary.MonthRechargeUSDMinor += *row.MonthRechargeUSDMinor
		}
	}
	return summary
}

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.statisticsOverview(ctx, dashboard.StatisticsQuery{
		AppCode:       app.AppCode,
		StatTZ:        statTZ,
		StartMS:       startMS,
		EndMS:         endMS,
		SeriesStartMS: startMS,
		SeriesEndMS:   endMS,
	})
	if err != nil {
		return nil, nil, err
	}
	regionRows := rollupByRegion(anyRowSlice(overview["country_breakdown"]), resolve)
	regionMetrics := map[int64]map[string]any{}
	for _, row := range regionRows {
		if regionID, ok := rowInt64(row, "region_id"); ok {
			regionMetrics[regionID] = row
		}
	}
	total := topLevelMetrics(overview)
	if s.financeRecharge != nil {
		// 运营中心的人×App×区域充值 KPI 与经营概览共用 Finance 口径；
		// 否则同一路径切换菜单会出现两套充值金额。
		if err := s.applyFinanceRechargeKpiWindow(ctx, app.AppCode, startMS, endMS, total, regionMetrics); err != nil {
			return nil, nil, err
		}
	}
	return total, regionMetrics, nil
}

func scopeMetrics(total map[string]any, regions map[int64]map[string]any, regionID int64) map[string]any {
	if regionID == 0 {
		return total
	}
	return regions[regionID]
}

func metricPointer(row map[string]any, key string) *int64 {
	if row == nil {
		return nil
	}
	value, ok := rowInt64(row, key)
	if !ok {
		return nil
	}
	out := value
	return &out
}

// resolveKpiMonth 解析统计月份并给出该月在统计时区下的起止时间；结束时间截断到明天零点，
// 避免 legacy Mongo 源为未来日期空跑逐日聚合。
func resolveKpiMonth(periodMonth string, endMS int64, location *time.Location) (string, time.Time, time.Time, error) {
	now := time.Now().In(location)
	reference := now
	if endMS > 0 {
		reference = time.UnixMilli(endMS - 1).In(location)
	}
	month := strings.TrimSpace(periodMonth)
	if month == "" {
		month = reference.Format("2006-01")
	}
	parsed, err := time.ParseInLocation("2006-01", month, location)
	if err != nil {
		return "", time.Time{}, time.Time{}, fmt.Errorf("period_month must be formatted as YYYY-MM")
	}
	monthStart := time.Date(parsed.Year(), parsed.Month(), 1, 0, 0, 0, 0, location)
	monthEnd := monthStart.AddDate(0, 1, 0)
	tomorrow := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, location).AddDate(0, 0, 1)
	if monthEnd.After(tomorrow) {
		monthEnd = tomorrow
	}
	if !monthEnd.After(monthStart) {
		monthEnd = monthStart.AddDate(0, 0, 1)
	}
	return month, monthStart, monthEnd, nil
}

func (s *Service) listApps(ctx context.Context) ([]AppInfo, error) {
	out := []AppInfo{}
	seen := map[string]struct{}{}
	if s.apps != nil {
		hyappApps, err := s.apps.ListApps(ctx)
		if err != nil {
			return nil, fmt.Errorf("list hyapp apps: %w", err)
		}
		for _, app := range hyappApps {
			appCode := appctx.Normalize(app.AppCode)
			if appCode == "" {
				continue
			}
			if _, ok := seen[appCode]; ok {
				continue
			}
			seen[appCode] = struct{}{}
			out = append(out, AppInfo{
				AppCode: appCode,
				AppName: firstNonEmptyString(app.AppName, appCode),
				LogoURL: strings.TrimSpace(app.LogoURL),
				Kind:    appKindHyapp,
			})
		}
	}
	for _, app := range s.legacyApps {
		if _, ok := seen[app.AppCode]; ok {
			continue
		}
		seen[app.AppCode] = struct{}{}
		out = append(out, AppInfo{
			AppCode: app.AppCode,
			AppName: firstNonEmptyString(app.AppName, app.AppCode),
			LogoURL: strings.TrimSpace(app.LogoURL),
			Kind:    appKindLegacy,
		})
	}
	return out, nil
}

func (s *Service) listAllowedApps(ctx context.Context, requested []string, access repository.MoneyAccess) ([]AppInfo, error) {
	apps, err := s.listApps(ctx)
	if err != nil {
		return nil, err
	}
	return filterAllowedApps(apps, requested, access), nil
}

// filterAllowedApps 只裁剪可查询的 App，不改变 appregistry 给出的稳定顺序。
// Overview 会保留裁剪前的注册 App 集合，供全局 country_id 展示目录使用。
func filterAllowedApps(apps []AppInfo, requested []string, access repository.MoneyAccess) []AppInfo {
	requestedSet := map[string]struct{}{}
	for _, appCode := range requested {
		appCode = appctx.Normalize(appCode)
		if appCode != "" {
			requestedSet[appCode] = struct{}{}
		}
	}
	out := []AppInfo{}
	for _, app := range apps {
		if len(requestedSet) > 0 {
			if _, ok := requestedSet[app.AppCode]; !ok {
				continue
			}
		}
		if !access.All {
			allowAll, allowedIDs := allowedRegions(access, app.AppCode)
			if !allowAll && len(allowedIDs) == 0 {
				continue
			}
		}
		out = append(out, app)
	}
	return out
}

type countryDirectoryEntry struct {
	AppCode     string
	ID          int64
	Code        string
	Name        string
	DisplayName string
	Flag        string
	Enabled     bool
}

// countryDirectories 按 app_code + country_id 汇总当前可查询 HyApp 的国家主数据。
// country_id 虽是全局自增主键，但国家启停和归属仍受 app_code 隔离；目录不能跨 App 复用。
func (s *Service) countryDirectories(ctx context.Context, apps []AppInfo) map[string]map[int64]countryDirectoryEntry {
	byApp := make(map[string]map[int64]countryDirectoryEntry)
	if s.user == nil {
		return byApp
	}

	// appregistry 当前按名称排序；改按 app_code 查询让下游调用顺序和日志保持稳定。
	hyappApps := make([]AppInfo, 0, len(apps))
	for _, app := range apps {
		if app.Kind == appKindHyapp {
			hyappApps = append(hyappApps, app)
		}
	}
	sort.Slice(hyappApps, func(i, j int) bool {
		return hyappApps[i].AppCode < hyappApps[j].AppCode
	})

	for _, app := range hyappApps {
		// user-service 依赖 RequestMeta.app_code 选择 App 数据域；每次调用都必须显式覆盖请求上下文。
		appContext := appctx.WithContext(ctx, app.AppCode)
		countries, err := s.user.ListCountries(appContext, userclient.ListCountriesRequest{
			Caller: "admin-server",
			// nil 表示同时读取 enabled 与 disabled。已停用国家仍可能存在于历史统计事实中，不能从展示目录丢失。
			Enabled: nil,
		})
		if err != nil {
			// 国家目录只用于展示；单 App 故障时该 App 降级为未知，不能拖垮整个 Social BI Overview 接口。
			slog.Warn("databi_list_hyapp_countries_failed", "app_code", app.AppCode, "error", err)
			continue
		}
		byID := make(map[int64]countryDirectoryEntry, len(countries))
		mergeCountryDirectory(byID, app.AppCode, countries)
		byApp[app.AppCode] = byID
	}
	return byApp
}

// mergeCountryDirectory 将一次 App 查询投影为 BI 展示字段；ID=0 是统计未知值，不伪造国家主数据。
// 同一 App 返回重复 ID 时保留第一项，避免异常目录响应在同一次请求内反复覆盖展示语义。
func mergeCountryDirectory(byID map[int64]countryDirectoryEntry, appCode string, countries []*userclient.Country) {
	for _, country := range countries {
		if country == nil || country.CountryID <= 0 {
			continue
		}
		if _, exists := byID[country.CountryID]; exists {
			continue
		}
		byID[country.CountryID] = countryDirectoryEntry{
			AppCode:     appctx.Normalize(appCode),
			ID:          country.CountryID,
			Code:        strings.TrimSpace(country.CountryCode),
			Name:        strings.TrimSpace(country.CountryName),
			DisplayName: strings.TrimSpace(country.CountryDisplayName),
			Flag:        strings.TrimSpace(country.Flag),
			Enabled:     country.Enabled,
		}
	}
}

// enrichCountryRows 复制统计行后补展示字段，不能原地修改 statisticsOverview 的共享缓存。
func enrichCountryRows(rows []map[string]any, countries map[int64]countryDirectoryEntry, preserveSourceNames bool) []map[string]any {
	out := make([]map[string]any, 0, len(rows))
	for _, source := range rows {
		row := make(map[string]any, len(source)+4)
		for key, value := range source {
			row[key] = value
		}
		enrichCountryRow(row, countries, preserveSourceNames)
		out = append(out, row)
	}
	return out
}

func enrichCountryRow(row map[string]any, countries map[int64]countryDirectoryEntry, preserveSourceNames bool) {
	// 外接 legacy 源可能使用 country_name、country 或 country_code 标识国家；任一字段存在时都以源数据为准，
	// 避免它携带的本地数字 ID 碰巧等于某个 HyApp 全局 ID 后被错误覆盖。
	if preserveSourceNames {
		sourceDisplayName := rowString(row, "country_display_name")
		sourceName := rowString(row, "country_name", "country")
		sourceCode := rowString(row, "country_code")
		if firstNonEmptyString(sourceDisplayName, sourceName, sourceCode) != "" {
			row["country_name"] = firstNonEmptyString(sourceDisplayName, sourceName, sourceCode)
			setCountryFieldIfEmpty(row, "country_code", "")
			// 字段必须存在供前端统一读取，但不把 legacy 的英文名伪装成原始中文 display_name。
			setCountryFieldIfEmpty(row, "country_display_name", "")
			setCountryFieldIfEmpty(row, "flag", "")
			return
		}
		// legacy 的数字 ID 由外部数据源定义，不能碰巧命中 HyApp ID 后借用其名称；
		// 源没有返回名称/国家码时明确保持未知，等待该源补齐自己的展示契约。
		row["country_name"] = unknownCountryName
		setCountryFieldIfEmpty(row, "country_code", "")
		setCountryFieldIfEmpty(row, "country_display_name", unknownCountryName)
		setCountryFieldIfEmpty(row, "flag", "")
		return
	}
	countryID, hasCountryID := rowInt64(row, "country_id")
	country, found := countries[countryID]
	if !hasCountryID || countryID <= 0 || !found {
		// 0 和目录未命中都属于未知维度；保留原 country_id 便于继续追查事实来源，不伪造 ID 或国家码。
		setCountryFieldIfEmpty(row, "country_code", "")
		row["country_name"] = unknownCountryName
		setCountryFieldIfEmpty(row, "country_display_name", unknownCountryName)
		setCountryFieldIfEmpty(row, "flag", "")
		return
	}

	row["country_code"] = country.Code
	// HyApp 数字统计行以 country_id 对应的主数据为准；country_display_name 保持原始字段，
	// country_name 则按中文展示名、英文名、国家码依次 fallback，供现有前端直接渲染。
	row["country_display_name"] = country.DisplayName
	row["flag"] = country.Flag
	row["country_name"] = firstNonEmptyString(
		country.DisplayName,
		country.Name,
		country.Code,
		unknownCountryName,
	)
}

func setCountryFieldIfEmpty(row map[string]any, key string, fallback string) {
	if rowString(row, key) == "" {
		row[key] = fallback
	}
}

// regionCatalog 汇总每个 App 的区域目录：hyapp App 共用 user-service 的区域主数据，
// legacy App 读各自 Mongo sys_region_config（与财务范围同一套合成 region_id）。
// 单个目录源故障时降级返回其余目录（行仍可按行内 region_id 分组），不让整个 BI 查询失败。
func (s *Service) regionCatalog(ctx context.Context, apps []AppInfo) (map[string][]RegionInfo, error) {
	out := map[string][]RegionInfo{}
	hasHyapp := false
	legacyNeeded := map[string]struct{}{}
	for _, app := range apps {
		if app.Kind == appKindHyapp {
			hasHyapp = true
		} else {
			legacyNeeded[app.AppCode] = struct{}{}
		}
	}

	hyappRegions := []RegionInfo{}
	if hasHyapp && s.user != nil {
		regions, err := s.user.ListRegions(ctx, userclient.ListRegionsRequest{Caller: "admin-server", Status: "active"})
		if err != nil {
			slog.Warn("databi_list_hyapp_regions_failed", "error", err)
		}
		for _, region := range regions {
			if region == nil {
				continue
			}
			hyappRegions = append(hyappRegions, RegionInfo{
				RegionID:   region.RegionID,
				RegionCode: region.RegionCode,
				RegionName: firstNonEmptyString(region.Name, region.RegionCode),
				Countries:  normalizeCountryCodes(region.Countries),
			})
		}
	}

	legacyRegions := map[string][]RegionInfo{}
	if len(legacyNeeded) > 0 {
		for _, catalog := range s.legacyRegionCatalogs {
			entries, err := catalog.RegionCatalog(ctx)
			if err != nil {
				slog.Warn("databi_list_legacy_regions_failed", "error", err)
				continue
			}
			for _, entry := range entries {
				appCode := appctx.Normalize(entry.AppCode)
				if _, ok := legacyNeeded[appCode]; !ok {
					continue
				}
				legacyRegions[appCode] = append(legacyRegions[appCode], RegionInfo{
					AppCode:    appCode,
					RegionID:   entry.RegionID,
					RegionCode: entry.RegionCode,
					RegionName: firstNonEmptyString(entry.RegionName, entry.RegionCode),
					Countries:  normalizeCountryCodes(entry.Countries),
				})
			}
		}
	}

	for _, app := range apps {
		if app.Kind == appKindHyapp {
			regions := make([]RegionInfo, 0, len(hyappRegions))
			for _, region := range hyappRegions {
				region.AppCode = app.AppCode
				regions = append(regions, region)
			}
			out[app.AppCode] = regions
			continue
		}
		out[app.AppCode] = legacyRegions[app.AppCode]
	}
	return out, nil
}

func allowedRegions(access repository.MoneyAccess, appCode string) (bool, []int64) {
	if access.All {
		return true, nil
	}
	appCode = appctx.Normalize(appCode)
	ids := []int64{}
	for _, scope := range access.Scopes {
		if appctx.Normalize(scope.AppCode) != appCode {
			continue
		}
		if scope.RegionID == 0 {
			return true, nil
		}
		ids = append(ids, scope.RegionID)
	}
	return false, ids
}

func regionResolver(regions []RegionInfo) func(map[string]any) (RegionInfo, bool) {
	byID := map[int64]RegionInfo{}
	byCountry := map[string]RegionInfo{}
	for _, region := range regions {
		byID[region.RegionID] = region
		for _, code := range region.Countries {
			if _, ok := byCountry[code]; !ok {
				byCountry[code] = region
			}
		}
	}
	return func(row map[string]any) (RegionInfo, bool) {
		if regionID, ok := rowInt64(row, "region_id"); ok && regionID > 0 {
			if region, ok := byID[regionID]; ok {
				return region, true
			}
			return RegionInfo{RegionID: regionID, RegionName: "区域 " + strconv.FormatInt(regionID, 10)}, true
		}
		countryCode := strings.ToUpper(rowString(row, "country_code", "countryCode"))
		if countryCode != "" {
			if region, ok := byCountry[countryCode]; ok {
				return region, true
			}
		}
		return RegionInfo{}, false
	}
}

func regionInCatalog(regions []RegionInfo, regionID int64) bool {
	for _, region := range regions {
		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 repository.RegionIDLooseEqual(region.RegionID, regionID) {
			return region.RegionCode, region.RegionName
		}
	}
	return "", "区域 " + strconv.FormatInt(regionID, 10)
}

func filterRowsByRegion(rows []map[string]any, resolve func(map[string]any) (RegionInfo, bool), allowed map[int64]struct{}) []map[string]any {
	out := make([]map[string]any, 0, len(rows))
	for _, row := range rows {
		region, ok := resolve(row)
		if !ok {
			continue
		}
		if _, ok := allowed[region.RegionID]; !ok {
			continue
		}
		out = append(out, row)
	}
	return out
}

// rollupByRegion 把国家行按区域目录归并成大区行；无法归属任何区域的国家进入
// region_id=0 的"未划分区域"行，保证与整体合计能对账。
func rollupByRegion(rows []map[string]any, resolve func(map[string]any) (RegionInfo, bool)) []map[string]any {
	type regionGroup struct {
		region RegionInfo
		acc    *metricAccumulator
	}
	groups := map[int64]*regionGroup{}
	order := []int64{}
	for _, row := range rows {
		region, ok := resolve(row)
		if !ok {
			region = RegionInfo{RegionID: 0, RegionCode: "UNASSIGNED", RegionName: "未划分区域"}
		}
		group := groups[region.RegionID]
		if group == nil {
			group = &regionGroup{region: region, acc: newMetricAccumulator()}
			groups[region.RegionID] = group
			order = append(order, region.RegionID)
		}
		group.acc.addRow(row)
	}
	out := make([]map[string]any, 0, len(order))
	for _, regionID := range order {
		group := groups[regionID]
		row := group.acc.toMap()
		row["region_id"] = group.region.RegionID
		row["region_code"] = group.region.RegionCode
		row["region_name"] = group.region.RegionName
		out = append(out, row)
	}
	sort.Slice(out, func(i, j int) bool {
		left, _ := rowInt64(out[i], "recharge_usd_minor")
		right, _ := rowInt64(out[j], "recharge_usd_minor")
		if left != right {
			return left > right
		}
		leftID, _ := rowInt64(out[i], "region_id")
		rightID, _ := rowInt64(out[j], "region_id")
		return leftID < rightID
	})
	return out
}

func rollupDailyByRegion(rows []map[string]any, resolve func(map[string]any) (RegionInfo, bool)) []map[string]any {
	byDay := map[string][]map[string]any{}
	days := []string{}
	for _, row := range rows {
		day := rowString(row, "stat_day", "label")
		if day == "" {
			continue
		}
		if _, ok := byDay[day]; !ok {
			days = append(days, day)
		}
		byDay[day] = append(byDay[day], row)
	}
	sort.Strings(days)
	out := []map[string]any{}
	for _, day := range days {
		for _, row := range rollupByRegion(byDay[day], resolve) {
			row["stat_day"] = day
			row["label"] = day
			out = append(out, row)
		}
	}
	return out
}

func rollupDaily(rows []map[string]any) []map[string]any {
	byDay := map[string]*metricAccumulator{}
	days := []string{}
	for _, row := range rows {
		day := rowString(row, "stat_day", "label")
		if day == "" {
			continue
		}
		acc := byDay[day]
		if acc == nil {
			acc = newMetricAccumulator()
			byDay[day] = acc
			days = append(days, day)
		}
		acc.addRow(row)
	}
	sort.Strings(days)
	out := make([]map[string]any, 0, len(days))
	for _, day := range days {
		row := byDay[day].toMap()
		row["stat_day"] = day
		row["label"] = day
		out = append(out, row)
	}
	return out
}

// topLevelMetrics 复制统计源顶层指标并剥离结构性字段，避免响应里重复携带明细数组。
func topLevelMetrics(overview map[string]any) map[string]any {
	out := make(map[string]any, len(overview))
	for key, value := range overview {
		switch key {
		case "daily_series", "country_breakdown", "daily_country_breakdown", "report_metric_sources", "game_ranking", "lucky_gift_pools", "user_risk", "daily_region_breakdown", "region_breakdown":
			continue
		}
		out[key] = value
	}
	flattenRetentionMetrics(out, overview["retention"])
	return out
}

func flattenRetentionMetrics(out map[string]any, retention any) {
	retentionRow, ok := retention.(map[string]any)
	if !ok {
		return
	}
	for _, item := range []struct {
		day      string
		rateKey  string
		usersKey string
		baseKey  string
	}{
		{day: "day1", rateKey: "d1_retention_rate", usersKey: "d1_retention_users", baseKey: "d1_retention_base_users"},
		{day: "day7", rateKey: "d7_retention_rate", usersKey: "d7_retention_users", baseKey: "d7_retention_base_users"},
		{day: "day30", rateKey: "d30_retention_rate", usersKey: "d30_retention_users", baseKey: "d30_retention_base_users"},
	} {
		// statistics-service 的历史契约把整段留存放在 retention 子对象里；
		// Databi v2 的 App 行统一读取扁平字段，这里只做字段搬运，不重算口径。
		out[item.rateKey] = retentionRow[item.day+"_rate"]
		out[item.usersKey] = retentionRow[item.day+"_users"]
		out[item.baseKey] = retentionRow[item.day+"_base_users"]
	}
}

func scopeInfos(scopes []model.UserMoneyScope) []ScopeInfo {
	out := make([]ScopeInfo, 0, len(scopes))
	for _, scope := range scopes {
		out = append(out, ScopeInfo{AppCode: appctx.Normalize(scope.AppCode), RegionID: scope.RegionID})
	}
	return out
}

func int64Set(values []int64) map[int64]struct{} {
	out := make(map[int64]struct{}, len(values))
	for _, value := range values {
		out[value] = struct{}{}
	}
	return out
}

func normalizeCountryCodes(codes []string) []string {
	out := make([]string, 0, len(codes))
	seen := map[string]struct{}{}
	for _, code := range codes {
		normalized := strings.ToUpper(strings.TrimSpace(code))
		if normalized == "" {
			continue
		}
		if _, ok := seen[normalized]; ok {
			continue
		}
		seen[normalized] = struct{}{}
		out = append(out, normalized)
	}
	return out
}

func firstNonEmptyString(values ...string) string {
	for _, value := range values {
		if trimmed := strings.TrimSpace(value); trimmed != "" {
			return trimmed
		}
	}
	return ""
}