package service

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/shopspring/decimal"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/serverless"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/utils"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

var (
	RegionalSkuPriceCache sync.Map                       // make(map[string]interface{})
	priceMinToMonth       = decimal.NewFromInt(730 * 60) /* 730h per month*/
)

const defaultSkuRegion = "us-east-1"

type BillingService interface {
	GetTenantSpendLimitsSettings(
		ctx context.Context,
		tenantID uint64,
		clusterMap map[uint64]*tidbmgmtservicev1.Cluster,
		checkCluster bool,
	) (map[uint64]*facade.ServerlessSpendLimitBizData, error)
	GetSkuUnitPrices(
		ctx context.Context,
		provider, region string,
		tenantId uint64,
	) (map[constant.SkuType]*facade.BillingCharge, error)
	GetSkuStorageUsage(ctx context.Context, cluster *tidbmgmtservicev1.Cluster) (map[constant.SkuType]float64, error)
	GetBranchSkuUsage(ctx context.Context, branch *vo.Branch) (map[constant.SkuType]float64, error)
	HasFreeCreditQuota(ctx context.Context, spendlimits map[uint64]*facade.ServerlessSpendLimitBizData) bool
	ValidTenantPlan(ctx context.Context, tenantID uint64) bool
	CalculateClusterCostThisMonth(
		ctx context.Context,
		tenantId uint64,
		id string,
		clusterType constant.ClusterType,
		skuTypes []constant.SkuType,
		provider string,
		region string,
	) (*facade.CompleteCostExplorerResult, bizerr.BizError)
	GetCostExplorerServerlessSkuCost(
		cost *facade.CompleteCostExplorerResult,
	) *facade.BillingSubCostExplorerResult
	GetEmailTimeRange() (previousMonday, currentMonday time.Time, targetDBFile string)
	DownloadS3AndInitDB(ctx context.Context, targetDBFile string) error
	// list orgs with statement/slow logs, used to filter inactive orgs and clusters when send expensive sql email
	ListOrgsWithActiveLogs(ctx context.Context) (map[uint64][]uint64, error)
}

type billingService struct {
	billingFacade facade.BillingFacade
	o11yFacade    facade.O11yFacade
	serverless.UnimplementedServerlessBillingServiceServer
}

func NewBillingService(
	billingFacade facade.BillingFacade,
	o11yFacade facade.O11yFacade,
) BillingService {
	svc := &billingService{
		billingFacade: billingFacade,
		o11yFacade:    o11yFacade,
	}

	// prefill sku price cache
	for _, cfg := range config.Global().WorkerConfig.RegionalConfig {
		for i := 0; i < 3; i++ {
			_, err := svc.GetSkuUnitPrices(context.Background(), cfg.Provider, cfg.Region, 1)
			if err == nil {
				break
			}
		}
		log.WithContextL(context.Background()).
			Info("prefill sku price cache", zap.String("provider", cfg.Provider), zap.String("region", cfg.Region))
	}

	return svc
}

func (s *billingService) GetTenantSpendLimitsSettings(
	ctx context.Context, tenantID uint64, clusterMap map[uint64]*tidbmgmtservicev1.Cluster, checkCluster bool,
) (map[uint64]*facade.ServerlessSpendLimitBizData, error) {
	settings, err := s.billingFacade.GetSpendLimitSettings(ctx, tenantID)
	if err != nil {
		return nil, err
	}
	resultMap := make(map[uint64]*facade.ServerlessSpendLimitBizData, 0)
	for _, s := range settings {
		if s.RangeType != constant.SpendLimitRangeTypeCluster {
			continue
		}
		/* skip invalid cluster spendlimit settings */
		if _, ok := clusterMap[s.RangeID]; !ok && checkCluster {
			continue
		}
		bizData := &facade.ServerlessSpendLimitBizData{}
		if err := json.Unmarshal([]byte(s.BizData), bizData); err != nil {
			return nil, err
		}
		resultMap[s.RangeID] = bizData
	}

	return resultMap, nil
}

func rewriteRegion(region string) string {
	for _, cfg := range config.Global().WorkerConfig.RegionalConfig {
		if cfg.Region == region {
			return region
		}
	}
	// without billing sku config, hack region
	return defaultSkuRegion
}

func isDefaultSkuLocation(provider, region string) bool {
	return provider == config.ProviderAWS && region == defaultSkuRegion
}

func notifyMissingSkuFallback(provider, region string, tenantID uint64, cause error) {
	title := fmt.Sprintf(
		"[%s] Billing SKU fallback to aws/%s",
		strings.ToUpper(config.Global().ServerConfig.Stage),
		defaultSkuRegion,
	)
	message := fmt.Sprintf(
		"SKU is not configured for a new region, falling back to aws/%s prices.\n"+
			"provider=%s region=%s org_id=%d\nerror=%v",
		defaultSkuRegion,
		provider,
		region,
		tenantID,
		cause,
	)
	go utils.SendLarkMessage(title, message, utils.LarkMsgLevelWarn)
}

func buildSkuPriceRecords(
	tenantId uint64, skuTypeToStringList map[constant.SkuType]string,
) []*facade.BillingReport {
	records := make([]*facade.BillingReport, 0, len(skuTypeToStringList))
	for _, v := range skuTypeToStringList {
		records = append(records, &facade.BillingReport{
			TenantID: tenantId,
			Cost:     1,
			Sku:      v,
		})
	}
	return records
}

func isSkuObjectNotExists(err error) bool {
	return err != nil && strings.Contains(err.Error(), "ObjectNotExists")
}

func (s *billingService) GetSkuUnitPrices(
	ctx context.Context,
	provider, region string,
	tenantId uint64,
) (map[constant.SkuType]*facade.BillingCharge, error) {
	logger := log.WithContextL(ctx).
		With(zap.String("provider", provider), zap.String("region", region), zap.Uint64("org_id", tenantId))
	cacheKey := provider + "-" + region
	if value, ok := RegionalSkuPriceCache.Load(cacheKey); ok {
		if res, ok := value.(map[constant.SkuType]*facade.BillingCharge); ok {
			logger.Info("getSkuUnitPrices for local cache", zap.Any("body", res))
			return res, nil
		}
		RegionalSkuPriceCache.Delete(cacheKey)
	}
	region = rewriteRegion(region)

	skuNameBuilder := &vo.SkuNameBuilder{CloudProvider: provider, Region: region}
	skuTypeToStringList := map[constant.SkuType]string{
		constant.SkuTypeRowStorage:          skuNameBuilder.Build(constant.SkuTypeRowStorage),
		constant.SkuTypeColumnStorage:       skuNameBuilder.Build(constant.SkuTypeColumnStorage),
		constant.SkuTypeRequestUnit:         skuNameBuilder.Build(constant.SkuTypeRequestUnit),
		constant.SkuTypeRequestCapacityUnit: skuNameBuilder.Build(constant.SkuTypeRequestCapacityUnit),
	}
	skuStringToTypeList := map[string]constant.SkuType{
		skuNameBuilder.Build(constant.SkuTypeRowStorage):          constant.SkuTypeRowStorage,
		skuNameBuilder.Build(constant.SkuTypeColumnStorage):       constant.SkuTypeColumnStorage,
		skuNameBuilder.Build(constant.SkuTypeRequestUnit):         constant.SkuTypeRequestUnit,
		skuNameBuilder.Build(constant.SkuTypeRequestCapacityUnit): constant.SkuTypeRequestCapacityUnit,
	}

	records := buildSkuPriceRecords(tenantId, skuTypeToStringList)
	logger.Info("getSkuUnitPrices request", zap.Any("records", records))
	skuPrice, err := s.billingFacade.CalcFee(ctx, records)
	// Some regions (e.g. aws/sa-east-1) only provide Starter SKUs and have no Essential RCU SKU.
	// Retry without Essential so Starter flows like SetSpendLimit can still resolve unit prices.
	if isSkuObjectNotExists(err) {
		essentialSku := skuTypeToStringList[constant.SkuTypeRequestCapacityUnit]
		logger.Warn("essential sku not available, retry with starter skus only",
			zap.String("essential_sku", essentialSku), zap.Error(err))
		delete(skuTypeToStringList, constant.SkuTypeRequestCapacityUnit)
		delete(skuStringToTypeList, essentialSku)
		records = buildSkuPriceRecords(tenantId, skuTypeToStringList)
		logger.Info("getSkuUnitPrices request without essential", zap.Any("records", records))
		skuPrice, err = s.billingFacade.CalcFee(ctx, records)
	}
	if err != nil {
		if isSkuObjectNotExists(err) && !isDefaultSkuLocation(provider, region) {
			logger.Warn("sku not configured, fallback to aws us-east-1",
				zap.String("fallback_provider", config.ProviderAWS),
				zap.String("fallback_region", defaultSkuRegion),
				zap.Error(err))
			notifyMissingSkuFallback(provider, region, tenantId, err)
			res, fallbackErr := s.GetSkuUnitPrices(ctx, config.ProviderAWS, defaultSkuRegion, tenantId)
			if fallbackErr != nil {
				logger.Error("getSkuUnitPrices fallback failed", zap.Error(fallbackErr))
				return nil, fallbackErr
			}
			RegionalSkuPriceCache.Store(cacheKey, res)
			return res, nil
		}
		logger.Error("getSkuUnitPrices for billing-svc", zap.Error(err), zap.Any("records", records))
		return nil, err
	}

	res := make(map[constant.SkuType]*facade.BillingCharge)
	for k := range skuPrice {
		if skuStringToTypeList[k] == constant.SkuTypeColumnStorage ||
			skuStringToTypeList[k] == constant.SkuTypeRowStorage {
			skuPrice[k].RawPrice = skuPrice[k].RawPrice.Mul(priceMinToMonth) // Mib/min to MiB/month
		}
		res[skuStringToTypeList[k]] = skuPrice[k]
	}
	RegionalSkuPriceCache.Store(cacheKey, res)
	logger.Info("getSkuUnitPrices for billing-svc", zap.Any("body", res))
	return res, nil
}

func (s *billingService) GetSkuStorageUsage(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
) (map[constant.SkuType]float64, error) {
	clusterId := cluster.GetClusterId()
	keyspaceId := cluster.GetServerlessCluster().GetCluster().GetKeyspaceId()
	logger := log.WithContextL(ctx).
		With(zap.Uint64("cluster_id", clusterId)).
		With(zap.Uint32("keyspace_id", keyspaceId))

	o11yAddr := config.GetO11yUrl(
		cluster.CloudProvider,
		cluster.Region,
		cluster.GetServerlessCluster().GetCluster().GetKubernetes(),
	)
	if o11yAddr == "" {
		logger.Error(
			"can't get o11y regional url",
			zap.String("kubernetes", cluster.GetServerlessCluster().GetCluster().GetKubernetes()),
		)
		return nil, errors.New("can't get o11y regional url")
	}

	logger.Info("get storage usage from o11y")
	usages, err := s.o11yFacade.GetClusterSkuUsageByRegion(ctx, strconv.FormatUint(clusterId, 10), keyspaceId, o11yAddr)
	if err != nil {
		logger.Error("get cluster sku usage failed.", zap.Error(err))
		return nil, err
	}

	res := make(map[constant.SkuType]float64, 0)
	for k, v := range usages {
		res[constant.SkuAliasNameStringToType[k]] = v
	}

	// Include branch storage in cluster's storage. We don't need this any more. Comment to reserve logic for reference.
	//
	// branches, err := s.infraFacade.ListBillableBranchesUnderCluster(ctx, cluster.ProjectId, clusterId)
	// if err != nil {
	// 	logger.Error("failed to list billable branches under cluster", zap.Error(err))
	// 	return nil, err
	// }
	// var keyspaceIds []uint32
	// for _, branch := range branches {
	// 	branchStat, err := branch.AwsFreeTierClusterStatusFromBranch()
	// 	if err != nil {
	// 		logger.Error("failed to get aws free tier cluster status from branch", zap.Error(err), zap.String("branch", branch.Name))
	// 		return nil, err
	// 	}
	// 	keyspaceIds = append(keyspaceIds, branchStat.KeyspaceId)
	// }

	// branchUsages, err := s.o11yFacade.GetBranchesSkuUsagesByRegion(ctx, keyspaceIds, constant.MetricTypeTikvStorage, o11yAddr)
	// if err != nil {
	// 	logger.Error("get branch sku usage failed.", zap.Error(err), zap.Uint32s("keyspace_ids", keyspaceIds))
	// 	return nil, err
	// }
	// for _, v := range branchUsages {
	// 	res[constant.SkuTypeRowStorage] += v
	// }

	return res, nil
}

// GetBranchSkuUsage returns row storage/column storage/request units
func (s *billingService) GetBranchSkuUsage(
	ctx context.Context,
	branch *vo.Branch,
) (map[constant.SkuType]float64, error) {
	branchId := branch.ID
	keyspaceId := branch.KeyspaceId
	logger := log.WithContextL(ctx).With(zap.String("branch_id", branchId), zap.Uint32("keyspace_id", keyspaceId))

	if branch.AutoScaling.MinRcu > 0 {
		// essential cluster has no Spending Limit
		return nil, bizerr.New(errorcode.ErrCodeInvalidSpendLimitValue, errorcode.ErrMsgInvalidClusterPlan)
	}

	o11yAddr := config.GetO11yUrl(branch.CloudProviderName, branch.RegionName, branch.Kubernetes)
	if o11yAddr == "" {
		logger.Error("can't get o11y regional url", zap.String("kubernetes", branch.Kubernetes))
		return nil, errors.New("can't get o11y regional url")
	}

	logger.Info("get storage usage from o11y")
	usages, err := s.o11yFacade.GetClusterSkuUsageByRegion(ctx, branchId, keyspaceId, o11yAddr)
	if err != nil {
		logger.Error("get branch sku usage failed.", zap.Error(err))
		return nil, err
	}

	res := make(map[constant.SkuType]float64, 0)
	for k, v := range usages {
		res[constant.SkuAliasNameStringToType[k]] = v
	}

	skuPrice, err := s.GetSkuUnitPrices(ctx, branch.CloudProviderName, branch.RegionName, branch.TenantID)
	if err != nil {
		logger.Error("failed to get sku unit price", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgGetSkuPricesFailed)
	}
	logger.Info("calculate ru usages with cost/price")
	ruRawPrice := skuPrice[constant.SkuTypeRequestUnit].RawPrice
	if ruRawPrice.GreaterThan(decimal.Zero) {
		// branch needs to set ignoreBeta to false since it's beta now. And it will return 0 if below $0.01
		costV2, err := s.CalculateClusterCostThisMonth(
			ctx,
			branch.TenantID,
			branchId+"/"+branch.MainBranchID,
			constant.ClusterTypeBranch,
			[]constant.SkuType{constant.SkuTypeRequestUnit},
			branch.CloudProviderName,
			branch.RegionName,
		)
		if err != nil {
			return nil, err
		}
		subCost := s.GetCostExplorerServerlessSkuCost(costV2)
		res[constant.SkuTypeRequestUnit] = subCost.RuFreeCost.Add(subCost.RuSpentCost).Div(ruRawPrice).InexactFloat64()
	}
	return res, nil
}

func (s *billingService) HasFreeCreditQuota(
	ctx context.Context, spendlimits map[uint64]*facade.ServerlessSpendLimitBizData,
) bool {
	freeClusterCount := int32(0)
	for _, v := range spendlimits {
		if v.FreeCredits.GreaterThan(decimal.Zero) {
			freeClusterCount++
		}
	}
	return freeClusterCount < config.Global().BizConfig.ServerlessFreeQuotaMaxCount
}

func (s *billingService) ValidTenantPlan(ctx context.Context, tenantID uint64) bool {
	plan, err := s.billingFacade.GetTenantPlan(ctx, tenantID)
	if err == nil && plan != nil && (plan.EffectivePlan == "on_demand" ||
		(plan.EffectivePlan == "poc" && plan.POCCredits.GreaterThan(decimal.Zero))) {
		/* Spending Limit can only be set by on_demand and poc tenant */
		return true
	}
	return false
}

func (s *billingService) CalculateClusterCostThisMonth(
	ctx context.Context,
	tenantId uint64,
	id string, // cluster_id or branch_id/cluster_id
	clusterType constant.ClusterType,
	skuTypes []constant.SkuType,
	provider string,
	region string,
) (*facade.CompleteCostExplorerResult, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(
		zap.String("id", id),
		zap.Uint64("org_id", tenantId),
		zap.String("type", string(clusterType)),
	)

	now := time.Now()
	firstDay := now.AddDate(0, 0, -now.Day()+1)
	endDay := firstDay.AddDate(0, 1, 0)
	query := &facade.CostExplorerInternalQuery{
		StartDate:  time.Date(firstDay.Year(), firstDay.Month(), firstDay.Day(), 0, 0, 0, 0, firstDay.Location()),
		EndDate:    time.Date(endDay.Year(), endDay.Month(), endDay.Day(), 0, 0, 0, 0, endDay.Location()),
		IgnoreBeta: clusterType != constant.ClusterTypeBranch,
		Labels:     make(map[string]string, 1),
	}
	clusterID := id
	if clusterType == constant.ClusterTypeBranch {
		item := strings.Split(id, "/")
		query.Labels["branchID"] = item[0]
		clusterID = item[1]
	}
	query.Labels["clusterID"] = clusterID

	for _, skuType := range skuTypes {
		skuNameBuilder := &vo.SkuNameBuilder{CloudProvider: provider, Region: region}
		query.Units = append(query.Units, skuNameBuilder.Build(skuType))
	}

	cost, err := s.billingFacade.CostExplorerV2(ctx, tenantId, query)
	if err != nil {
		logger.Error("call CostExplorerV2 failed.", zap.Error(err))
	}
	logger.Debug("CalculateClusterCostThisMonth", zap.Any("cost", cost))
	return cost, err
}

func (s *billingService) GetCostExplorerServerlessSkuCost(
	cost *facade.CompleteCostExplorerResult,
) *facade.BillingSubCostExplorerResult {
	res := &facade.BillingSubCostExplorerResult{
		RuFreeCost:        decimal.NewFromInt(0),
		RowStoreFreeCost:  decimal.NewFromInt(0),
		ColumnarFreeCost:  decimal.NewFromInt(0),
		RuSpentCost:       decimal.NewFromInt(0),
		RowStoreSpentCost: decimal.NewFromInt(0),
		ColumnarSpentCost: decimal.NewFromInt(0),
	}
	if cost == nil {
		return res
	}
	if item, ok := cost.InternalCosts.CreditsCosts.KindCategoryUnitsCost["TiDB Cloud Starter"]; ok {
		if c, ok := item.CategoryItemUnitsCosts["Request Units"]; ok {
			res.RuFreeCost = res.RuFreeCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Row-based Storage"]; ok {
			res.RowStoreFreeCost = res.RowStoreFreeCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Row-based Storage with Dual-layer Encryption"]; ok {
			res.RowStoreFreeCost = res.RowStoreFreeCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Columnar Storage"]; ok {
			res.ColumnarFreeCost = res.ColumnarFreeCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Columnar Storage with Dual-layer Encryption"]; ok {
			res.ColumnarFreeCost = res.ColumnarFreeCost.Add(c.RawPrice)
		}
		// beta sku
		if c, ok := item.CategoryItemUnitsCosts["Request Units (Beta)"]; ok {
			res.RuFreeCost = res.RuFreeCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Row-based Storage (Beta)"]; ok {
			res.RowStoreFreeCost = res.RowStoreFreeCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Row-based Storage with Dual-layer Encryption (Beta)"]; ok {
			res.RowStoreFreeCost = res.RowStoreFreeCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Columnar Storage (Beta)"]; ok {
			res.ColumnarFreeCost = res.ColumnarFreeCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Columnar Storage with Dual-layer Encryption (Beta)"]; ok {
			res.ColumnarFreeCost = res.ColumnarFreeCost.Add(c.RawPrice)
		}
	}
	if item, ok := cost.ExternalCosts.TotalCosts.KindCategoryUnitsCost["TiDB Cloud Starter"]; ok {
		if c, ok := item.CategoryItemUnitsCosts["Request Units"]; ok {
			res.RuSpentCost = res.RuSpentCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Row-based Storage"]; ok {
			res.RowStoreSpentCost = res.RowStoreSpentCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Row-based Storage with Dual-layer Encryption"]; ok {
			res.RowStoreSpentCost = res.RowStoreSpentCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Columnar Storage"]; ok {
			res.ColumnarSpentCost = res.ColumnarSpentCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Columnar Storage with Dual-layer Encryption"]; ok {
			res.ColumnarSpentCost = res.ColumnarSpentCost.Add(c.RawPrice)
		}
		// beta sku
		if c, ok := item.CategoryItemUnitsCosts["Request Units (Beta)"]; ok {
			res.RuSpentCost = res.RuSpentCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Row-based Storage (Beta)"]; ok {
			res.RowStoreSpentCost = res.RowStoreSpentCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Row-based Storage with Dual-layer Encryption (Beta)"]; ok {
			res.RowStoreSpentCost = res.RowStoreSpentCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Columnar Storage (Beta)"]; ok {
			res.ColumnarSpentCost = res.ColumnarSpentCost.Add(c.RawPrice)
		}
		if c, ok := item.CategoryItemUnitsCosts["Columnar Storage with Dual-layer Encryption (Beta)"]; ok {
			res.ColumnarSpentCost = res.ColumnarSpentCost.Add(c.RawPrice)
		}
	}
	return res
}

func (s *billingService) GetEmailTimeRange() (previousMonday, currentMonday time.Time, targetDBFile string) {
	now := time.Now()
	// Calculate the current Monday
	currentMonday = now.AddDate(0, 0, -int(now.Weekday()-time.Monday))
	if now.Weekday() == time.Sunday {
		currentMonday = now.AddDate(0, 0, -6)
	}
	// Calculate the previous Monday
	previousMonday = currentMonday.AddDate(0, 0, -7)
	targetDBFile = previousMonday.Format("2006010200") + ".db"
	return previousMonday, currentMonday, targetDBFile
}

func (s *billingService) DownloadS3AndInitDB(ctx context.Context, targetDBFile string) error {
	logger := log.WithContextL(ctx)
	for _, regionConfig := range config.Global().WorkerConfig.RegionalConfig {
		if regionConfig.Provider == config.ProviderAWS {
			// statmentsFile: us-west-2_statements_168h_2024041400.db
			// slowlogFile: us-west-2_slowlog-stats_168h_2024041400.db
			statmentsFile, slowlogFile, err := s.o11yFacade.GetSQLiteDBFileFromS3(
				ctx,
				regionConfig.Region,
				targetDBFile,
			)
			if err != nil {
				logger.Warn("Get statements & slowlog from o11y s3 failed.",
					zap.String("region", regionConfig.Region), zap.String("error", err.Error()))
				return bizerr.New(errorcode.ErrCodeInternalError, errorcode.ErrMsgInitS3Failed)
			}
			if err = utils.InitSQLiteDB(regionConfig.Region, statmentsFile, slowlogFile); err != nil {
				logger.Error(
					"Init statements & slowlog from o11y s3 failed.",
					zap.String("region", regionConfig.Region),
					zap.Error(err),
				)
				return bizerr.New(errorcode.ErrCodeInternalError, errorcode.ErrMsgInitS3Failed)
			}
			logger.Info("Download statements & slowlog from o11y s3 success",
				zap.String("region", regionConfig.Region),
				zap.Any("SqliteQuerys", utils.SqliteQuerys))
		}
	}
	return nil
}

// list orgs with statement/slow logs, used to filter inactive orgs and clusters when send expensive sql email
func (s *billingService) ListOrgsWithActiveLogs(ctx context.Context) (map[uint64][]uint64, error) {
	_, _, targetDBFile := s.GetEmailTimeRange()
	if err := s.DownloadS3AndInitDB(ctx, targetDBFile); err != nil {
		return nil, err
	}

	result := make(map[uint64][]uint64)
	query := "select serverless_tenant_id, group_concat(distinct serverless_cluster_id) from %s where serverless_cluster_id not like ? group by serverless_tenant_id"
	for key, meta := range utils.SqliteQuerys {
		if err := func() error {
			tableName := "slow_stats"
			if strings.HasPrefix(key, utils.STATEMENT_PREFIX) {
				tableName = "statements"
			}
			rows, err := meta.DB.Query(fmt.Sprintf(query, tableName), "bran-%")
			if err != nil {
				return err
			}
			defer rows.Close()
			for rows.Next() {
				var orgIdStr, clusterIdsStr string
				if err := rows.Scan(&orgIdStr, &clusterIdsStr); err != nil {
					return err
				}
				orgId, err := strconv.ParseUint(orgIdStr, 10, 64)
				if err != nil {
					return err
				}
				clusterIdStrList := strings.Split(clusterIdsStr, ",")
				clusterIds := make([]uint64, len(clusterIdStrList))
				for i, clusterIdStr := range clusterIdStrList {
					clusterId, err := strconv.ParseUint(clusterIdStr, 10, 64)
					if err != nil {
						return err
					}
					clusterIds[i] = clusterId
				}
				if _, ok := result[orgId]; !ok {
					result[orgId] = clusterIds
				} else {
					result[orgId] = append(result[orgId], clusterIds...)
				}
			}
			return nil
		}(); err != nil {
			return nil, err
		}
	}
	return result, nil
}

// ValidateSpendLimitMax checks whether spendLimit (cents) is within the org's configured ceiling.
// Orgs listed in BizConfig.SpendLimitMaxOverrideTenants may exceed the default $10000 max.
func ValidateSpendLimitMax(orgID uint64, spendLimit uint32) error {
	limitMax := config.Global().BizConfig.MaxSpendLimitForOrg(orgID)
	if spendLimit > limitMax {
		return bizerr.New(
			errorcode.ErrCodeGreaterThanMaxSpendLimitValue,
			errorcode.ErrMsgGreaterThanMaxSpendLimitValue,
			strconv.FormatUint(uint64(limitMax)/100, 10),
		)
	}
	return nil
}