package billing_svc

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"html/template"
	"math"
	"regexp"
	"slices"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/dustin/go-humanize"
	"github.com/lithammer/shortuuid/v3"
	"github.com/sendgrid/sendgrid-go"
	"github.com/sendgrid/sendgrid-go/helpers/mail"
	"github.com/shopspring/decimal"
	"github.com/tidbcloud/eventbus-datastruct/event"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	"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/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"github.com/tidbcloud/serverless-service/internal/domain/service"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/infra/eventbus"
	"github.com/tidbcloud/serverless-service/internal/utils"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tiinf/sdk/auditlog"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/timestamppb"
)

var larkMsgTemplate = "Tenant: %d / %s / %s / %v\n" +
	"Cluster: %d / %s/%s\n" +
	"SpendLimit: $%.2f\n" +
	"FreeCredit: $%.2f\n" +
	"Success: %v\n" +
	"Error: %v"
var regionPattern = regexp.MustCompile(`^regions/(.+?)-(.+)$`)

type serverlessBillingService struct {
	billingFacade   facade.BillingFacade
	clusterRepo     repo.ClusterRepo
	clusterTidbPool repo.ClusterTidbPoolRepo
	billingService  service.BillingService
	accountFacade   facade.AccountFacade
	o11yFacade      facade.O11yFacade
	eventbusService *eventbus.EventbusService
	mgmtFacade      facade.MgmtSvcFacade
	serverless.UnimplementedServerlessBillingServiceServer
}

func NewBillingService(
	billingFacade facade.BillingFacade,
	clusterRepo repo.ClusterRepo,
	clusterTidbPool repo.ClusterTidbPoolRepo,
	billingService service.BillingService,
	accountFacade facade.AccountFacade,
	o11yFacade facade.O11yFacade,
	eventbusService *eventbus.EventbusService,
	mgmtFacade facade.MgmtSvcFacade,
) serverless.ServerlessBillingServiceServer {
	return &serverlessBillingService{
		billingFacade:   billingFacade,
		clusterRepo:     clusterRepo,
		clusterTidbPool: clusterTidbPool,
		billingService:  billingService,
		accountFacade:   accountFacade,
		o11yFacade:      o11yFacade,
		eventbusService: eventbusService,
		mgmtFacade:      mgmtFacade,
	}
}

func (s *serverlessBillingService) GetAllSpendLimits(
	ctx context.Context, req *serverless.GetAllSpendLimitsReq,
) (_ *serverless.GetAllSpendLimitsResp, err error) {
	ctx, span := Tracer.Start(ctx, "serverlessBilling.GetAllSpendLimits")
	logger := log.WithContextL(ctx).With(zap.Uint64("org_id", req.OrgId))
	defer func() { tracing.EndSpan(span, err) }()

	span.AddEvent("list clusters from mgmt-svc")
	// if the org has too many clusters, skip calling billing, it will break
	if slices.Contains(config.Global().BizConfig.BillingLargeTenants, req.OrgId) {
		listResp, err := s.mgmtFacade.ListClusters(ctx, &tidbmgmtservicev1.ListClustersRequest{
			Filter: &tidbmgmtservicev1.ClusterFilter{
				TenantIds:     []uint64{req.OrgId},
				View:          tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
				ExcludeStates: facade.DeletedClusterStatus,
				ClusterType:   tidbmgmtservicev1.DeployType_DEPLOY_TYPE_SERVERLESS.Enum(),
			},
			Page:     1,
			PageSize: 1,
		})
		if err != nil {
			logger.Error("failed to list clusters from mgmt-svc", zap.Error(err))
			return nil, bizerr.New(errorcode.ErrCodeSvcMgmt, errorcode.ErrMsgListClustersFailed)
		}
		return &serverless.GetAllSpendLimitsResp{
			Total:              int32(listResp.GetTotal()),
			FreeCreditMaxCount: config.Global().BizConfig.ServerlessFreeQuotaMaxCount,
			HasFreeCreditQuota: false,
		}, nil
	}

	clusters, err := s.mgmtFacade.ListAllClustersInOrg(
		ctx,
		req.OrgId,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
	)
	if err != nil {
		logger.Error("failed to list clusters from mgmt-svc", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcMgmt, errorcode.ErrMsgListClustersFailed)
	}

	clusterMap := make(map[uint64]*tidbmgmtservicev1.Cluster, len(clusters))
	for _, cluster := range clusters {
		clusterMap[cluster.ClusterId] = cluster
	}

	span.AddEvent("get billing spendlimits")
	resultMap, err := s.billingService.GetTenantSpendLimitsSettings(ctx, req.OrgId, clusterMap, true)
	if err != nil {
		logger.Error("failed to get all spendlimits from billingSvc", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgListSpendLimitsFailed)
	}

	resp := &serverless.GetAllSpendLimitsResp{
		Total:              int32(len(clusters)),
		FreeCreditMaxCount: config.Global().BizConfig.ServerlessFreeQuotaMaxCount,
		HasFreeCreditQuota: s.billingService.HasFreeCreditQuota(ctx, resultMap),
	}
	if len(clusters) <= int(config.Global().BizConfig.ServerlessFreeQuotaMaxCount) {
		resp.Items = make([]*serverless.GetSpendLimitResp, 0)
		for k, v := range resultMap {
			resp.Items = append(resp.Items, &serverless.GetSpendLimitResp{
				ClusterId:  k,
				SpendLimit: dolloarDeToCentsUint(v.InvoiceLimit),
				FreeCredit: dolloarDeToCentsUint(v.ActiveTotalCredits),
			})
		}
	}
	return resp, nil
}

func (s *serverlessBillingService) GetSpendLimit(
	ctx context.Context, req *serverless.GetSpendLimitReq,
) (_ *serverless.GetSpendLimitResp, err error) {
	ctx, span := Tracer.Start(ctx, "serverlessBilling.GetSpendLimit")
	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", req.ClusterId), zap.Uint64("org_id", req.OrgId))
	defer func() { tracing.EndSpan(span, err) }()

	span.AddEvent("list clusters from mgmt-svc")
	cluster, err := s.mgmtFacade.GetCluster(ctx, req.ClusterId, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if err != nil {
		logger.Error("get cluster from mgmt-svc failed", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcMgmt, errorcode.ErrMsgGetClusterFailed)
	}

	if cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMaxRcu() > 0 {
		// essential cluster has no Spending Limit
		return nil, bizerr.New(errorcode.ErrCodeInvalidSpendLimitValue, errorcode.ErrMsgInvalidClusterPlan)
	}

	span.AddEvent("get Spending Limit setting")
	setting, err := s.billingFacade.GetSpendLimit(ctx, req.OrgId, req.ClusterId)
	if err != nil {
		logger.Error("failed to get spendlimits from billingSvc", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgListSpendLimitsFailed)
	}
	bizData := &facade.ServerlessSpendLimitBizData{}
	if err := json.Unmarshal([]byte(setting.BizData), bizData); err != nil {
		logger.Error("failed to unmarshal bizdata", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgGetClusterFailed)
	}
	return &serverless.GetSpendLimitResp{
		ClusterId:  req.ClusterId,
		SpendLimit: dolloarDeToCentsUint(bizData.InvoiceLimit),
		FreeCredit: dolloarDeToCentsUint(bizData.ActiveTotalCredits),
	}, nil
}

func (s *serverlessBillingService) ModifySpendLimit(
	ctx context.Context, req *serverless.SetSpendLimitReq,
) (_ *serverless.GetSpendLimitResp, err error) {
	ctx, span := Tracer.Start(ctx, "serverlessBilling.ModifySpendLimit")
	defer func() { tracing.EndSpan(span, err) }()
	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", req.ClusterId),
		zap.Uint64("org_id", req.OrgId), zap.Uint32("spend_limit", req.SpendLimit))

	if err := service.ValidateSpendLimitMax(req.OrgId, req.SpendLimit); err != nil {
		return nil, err
	}
	resp, err := s.modifySpendLimit(ctx, logger, span, req)
	if err != nil {
		return nil, err
	}

	if e := s.clusterTidbPool.Save(ctx, &entity.ClusterTidbPool{
		Type:             constant.TypeSpendLimit,
		ClusterID:        req.ClusterId,
		SpendLimitStatus: constant.SpendLimitEventStatusModifySpendLimit,
	}); e != nil {
		logger.Error("failed to save cluster tidb pool settings", zap.Error(e))
	}

	if config.Global().ServerConfig.EnableAuditLog {
		/* set webconsole auditlog */
		eventBuilder := auditlog.NewEvent(ctx, auditlog.EventSetServerlessSpendLimit).
			SetResult(err == nil).
			SetOrgID(req.OrgId).
			SetClusterID(req.ClusterId).
			SetDetails(auditlog.Details{
				auditlog.KeyError:       err,
				auditlog.KeyTierType:    "Serverless",
				auditlog.KeyDescription: "spend_limit = " + fmt.Sprint(req.SpendLimit),
			}.Build())
		if e := eventBuilder.Submit(); e != nil {
			logger.Warn("failed to submit audit log", zap.Error(err))
		}
	}

	return resp, err
}

func (s *serverlessBillingService) SetSpendLimit(
	ctx context.Context, req *serverless.SetSpendLimitReq,
) (_ *serverless.GetSpendLimitResp, err error) {
	ctx, span := Tracer.Start(ctx, "serverlessBilling.SetSpendLimit")
	defer func() { tracing.EndSpan(span, err) }()
	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", req.ClusterId),
		zap.Uint64("org_id", req.OrgId), zap.Uint32("spend_limit", req.SpendLimit))

	if req.OrgId == 2130684 {
		logger.Info("skip org 2130684 to set spend limit")
		return &serverless.GetSpendLimitResp{
			ClusterId:  req.ClusterId,
			SpendLimit: req.SpendLimit,
		}, nil
	}

	if err := service.ValidateSpendLimitMax(req.OrgId, req.SpendLimit); err != nil {
		return nil, err
	}
	return s.setSpendLimit(ctx, logger, span, req)
	/* save to the clusterTidbPool repo by subscribing the cluster created events */
}

func (s *serverlessBillingService) modifySpendLimit(
	ctx context.Context, logger *zap.Logger, span trace.Span, req *serverless.SetSpendLimitReq,
) (_ *serverless.GetSpendLimitResp, err error) {
	if req.SpendLimit > 0 && !s.validTenantBillingAction(ctx, req.OrgId) {
		return nil, bizerr.New(errorcode.ErrCodeNoPaymentMethods, errorcode.ErrMsgNoPaymentMethods)
	}

	span.AddEvent("get clusters from mgmt-svc")
	cluster, err := s.mgmtFacade.GetCluster(ctx, req.ClusterId, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if err != nil {
		logger.Error("failed to get cluster from mgmt-svc", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcMgmt, errorcode.ErrMsgGetClusterFailed)
	}
	logger.Info("get clusters from mgmt-svc", zap.Any("cluster", cluster))
	if err := checkSpendLimitRestriction(req, cluster); err != nil {
		return nil, err
	}
	clusterPlan := service.GetClusterPlanWithSpendingLimit(
		cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMaxRcu(),
		req.GetSpendLimit(),
	)
	logger = logger.With(zap.String("cluster_plan", clusterPlan))
	logger.Info("parser sku plan", zap.Bool("is_modify_op", true))

	provider := cluster.CloudProvider
	region := cluster.Region
	var bizData facade.ServerlessSpendLimitBizData

	span.AddEvent("get Spending Limit setting")
	setting, e1 := s.billingFacade.GetSpendLimit(ctx, req.OrgId, req.ClusterId)
	if e1 != nil {
		logger.Error("failed to get spendlimits from billingSvc", zap.Error(e1))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgListSpendLimitsFailed)
	}
	if err := json.Unmarshal([]byte(setting.BizData), &bizData); err != nil {
		logger.Error("failed to parse biz_data.", zap.Any("content", setting), zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgGetClusterFailed)
	}

	spendLimit := decimal.NewFromInt(int64(req.SpendLimit)).Div(constant.DolloarToCents)
	if bizData.InvoiceLimit.Equal(spendLimit) { // no changes
		logger.Info("no changes for spendlimit")
		return &serverless.GetSpendLimitResp{
			ClusterId:  req.ClusterId,
			SpendLimit: req.SpendLimit,
			FreeCredit: dolloarDeToCentsUint(bizData.FreeCredits),
		}, nil
	}

	span.AddEvent("get cost from billing")
	info, err := s.billingFacade.GetSpendLimitInfo(ctx, setting)
	if err != nil {
		logger.Error("failed to get spendlimit info", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgGetSpendLimitInfoFailed)
	}

	minSpendLimit := info.Costs
	if spendLimit.LessThan(minSpendLimit) {
		return nil, bizerr.New(
			errorcode.ErrCodeLessThanMinSpendLimitValue,
			errorcode.ErrMsgLessThanMinSpendLimitValue,
			minSpendLimit.String(),
		)
	}

	if req.ValidateOnly != nil && *req.ValidateOnly {
		return &serverless.GetSpendLimitResp{
			ClusterId:  req.ClusterId,
			SpendLimit: req.SpendLimit,
		}, nil
	}

	span.AddEvent("update Spending Limit")
	bizData.InvoiceLimit = spendLimit
	body, _ := json.Marshal(&bizData)
	setting.BizData = string(body)
	setting.Limit = spendLimit.Add(bizData.FreeCredits)
	if err := s.billingFacade.ModifySpendLimit(ctx, setting); err != nil {
		logger.Error("update spendlimits setting", zap.Any("body", setting), zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgModifySpendLimitFailed)
	}

	op := "Modify Spending Limit"

	s.postUpdateSpendLimit(ctx, logger, span, req, bizData, op, provider, region, cluster, clusterPlan, err)

	return &serverless.GetSpendLimitResp{
		ClusterId:  req.ClusterId,
		SpendLimit: req.SpendLimit,
		FreeCredit: dolloarDeToCentsUint(bizData.FreeCredits),
	}, nil
}

func (s *serverlessBillingService) setSpendLimit(
	ctx context.Context, logger *zap.Logger, span trace.Span, req *serverless.SetSpendLimitReq,
) (_ *serverless.GetSpendLimitResp, err error) {
	if req.SpendLimit > 0 && !s.validTenantBillingAction(ctx, req.OrgId) {
		return nil, bizerr.New(errorcode.ErrCodeNoPaymentMethods, errorcode.ErrMsgNoPaymentMethods)
	}

	span.AddEvent("get cluster from mgmt-svc")
	cluster, err := s.mgmtFacade.GetCluster(ctx, req.ClusterId, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if err != nil {
		logger.Error("get cluster from mgmt-svc failed", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcMgmt, errorcode.ErrMsgGetClusterFailed)
	}
	logger.Info("get cluster from mgmt-svc", zap.Any("cluster", cluster))
	if err := checkSpendLimitRestriction(req, cluster); err != nil {
		return nil, err
	}
	clusterPlan := service.GetClusterPlanWithSpendingLimit(
		cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMinRcu(),
		req.GetSpendLimit(),
	)
	logger = logger.With(zap.String("cluster_plan", clusterPlan))

	// TODO: check from billing may be duplicate with check from mgmt, do more test and remove one in the future
	sl, _ := s.billingFacade.GetSpendLimit(ctx, req.OrgId, req.ClusterId)
	if sl != nil {
		logger.Info("Spending Limit already exsits", zap.Uint64("spendLimitId", sl.SpendLimitSettingID))
		return s.modifySpendLimit(ctx, logger, span, req)
	}

	span.AddEvent("get region sku price from billing")
	provider := cluster.CloudProvider
	region := cluster.Region
	skuPrice, err := s.billingService.GetSkuUnitPrices(ctx, provider, region, req.OrgId)
	if err != nil {
		logger.Error("failed to get sku unit price", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgGetSkuPricesFailed)
	}

	var (
		bizData facade.ServerlessSpendLimitBizData
		op      = "Initialize Spending Limit"
	)

	span.AddEvent("set Spending Limit")
	spendLimit := decimal.NewFromInt(int64(req.SpendLimit)).Div(constant.DolloarToCents)
	bizData.InvoiceLimit = spendLimit
	bizData.FreeCredits = decimal.Zero
	hasQuota, _ := s.hasFreeCreditQuota(ctx, logger, span, req.OrgId)
	if hasQuota {
		bizData.FreeCredits = genFreeCredits(skuPrice)
		bizData.RechargePlanID = constant.BillingRechargePlanId[provider+"/"+region]
	}
	if bizData.FreeCredits.IsZero() && spendLimit.IsZero() {
		return nil, bizerr.New(
			errorcode.ErrCodeServerlessReachMaxFreeNum,
			errorcode.ErrMsgServerlessReachFreeMaximumNum,
		)
	}
	if req.ValidateOnly != nil && *req.ValidateOnly {
		return &serverless.GetSpendLimitResp{
			ClusterId:  req.ClusterId,
			SpendLimit: req.SpendLimit,
		}, nil
	}

	data, _ := json.Marshal(&bizData)
	setting := &facade.SpendLimitSetting{
		TenantID:  req.OrgId,
		RangeID:   req.ClusterId,
		Limit:     bizData.InvoiceLimit.Add(bizData.FreeCredits),
		BizData:   string(data),
		RangeType: constant.SpendLimitRangeTypeCluster,
		Interval:  "monthly",
		CostType:  "total",
		BizType:   constant.SpendLimitBizType_ServerlessCluster,
	}
	if err := s.billingFacade.SetSpendLimit(ctx, setting); err != nil {
		logger.Error("set spendlimit setting", zap.Any("body", setting), zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgSetSpendLimitFailed)
	}

	s.postUpdateSpendLimit(ctx, logger, span, req, bizData, op, provider, region, cluster, clusterPlan, err)

	return &serverless.GetSpendLimitResp{
		ClusterId:  req.ClusterId,
		SpendLimit: req.SpendLimit,
		FreeCredit: dolloarDeToCentsUint(bizData.FreeCredits),
	}, nil
}

func (s *serverlessBillingService) postUpdateSpendLimit(
	ctx context.Context, logger *zap.Logger, span trace.Span, req *serverless.SetSpendLimitReq,
	bizData facade.ServerlessSpendLimitBizData, op, provider, region string, cluster *tidbmgmtservicev1.Cluster,
	clusterPlan string, err error,
) {
	if cluster.GetServerlessCluster().GetUserStatus() ==
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL &&
		cluster.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoServicePlan] !=
			constant.SERVERLESS_PLAN_STARTER_SCALABLE {
		span.AddEvent("update cluster cr for service plan")
		if _, err1 := s.mgmtFacade.UpdateServerlessClusterAnnotations(ctx,
			req.ClusterId, map[string]string{
				constant.AnnoServicePlan: constant.SERVERLESS_PLAN_STARTER_SCALABLE,
			}); err1 != nil {
			logger.Error("update cluster service plan annotations failed", zap.Error(err1))
		}
	} else {
		logger.Info("ignore update cluster cr for service plan", zap.Any("cluster", cluster))
	}

	if req.SpendLimit > 0 {
		if !utils.IsNotifyBlockTenantsAndEnv(req.OrgId) {
			span.AddEvent("send lark message")
			title := fmt.Sprintf("[%s] %s", strings.ToUpper(config.Global().ServerConfig.Stage), op)
			tenant, _ := s.accountFacade.GetTenant(ctx, req.OrgId)
			if tenant == nil {
				tenant = &facade.Tenant{}
			}
			message := fmt.Sprintf(
				larkMsgTemplate,
				req.OrgId,
				tenant.Name,
				tenant.Email,
				tenant.Kind,
				req.ClusterId,
				provider,
				region,
				float64(
					req.SpendLimit,
				)/100.0,
				float64(dolloarDeToCentsUint(bizData.FreeCredits))/100.0,
				err == nil,
				err,
			)
			go utils.SendLarkMessage(title, message, utils.LarkMsgLevelInfo)
		}
	}

	var userID uint64
	user, err := utils.GetUserFromContext(ctx)
	if err != nil {
		logger.Info("get user from context failed", zap.Error(err))
	} else {
		userID = user.ID
	}

	// Send events center
	if s.eventbusService != nil {
		span.AddEvent("publish events center")
		data := &event.DisplayEvent{
			DisplayEventType: constant.EventTypeUpdateSpendingLimit,
			DisplayStatus:    event.EventStatusSucceeded,
			Severity:         event.EventSeverityInfo,
			Summary:          fmt.Sprintf(constant.EventSummaryUpdateSpendingLimit, op, clusterPlan),
			HappenedAtMs:     uint64(time.Now().UTC().UnixMilli()),
			AuthorID:         userID,
			ClusterID:        cluster.ClusterId,
			ProjectID:        cluster.ProjectId,
			TenantID:         cluster.OrganizationId,
			Detail: fmt.Sprintf(
				"free_credits: $%.2f, cluster plan: %s.",
				float64(dolloarDeToCentsUint(bizData.FreeCredits))/100.0,
				clusterPlan,
			),
		}
		go func(data *event.DisplayEvent) {
			if err := s.eventbusService.Publish(&tieventbus.Event{
				ID:   shortuuid.New(),
				Type: event.DisplayEventTopic,
				Data: data,
			}); err != nil {
				logger.Error("failed to publish events center", zap.Error(err))
				return
			}
		}(data)
	}
}

func (s *serverlessBillingService) hasFreeCreditQuota(
	ctx context.Context, logger *zap.Logger, span trace.Span, tenantID uint64,
) (result bool, err error) {
	if slices.Contains(config.Global().BizConfig.BillingLargeTenants, tenantID) {
		return false, nil
	}

	span.AddEvent("get billing spendlimits")
	clusters, err := s.mgmtFacade.ListAllClustersInOrg(
		ctx,
		tenantID,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
	)
	if err != nil {
		logger.Error("failed to list clusters from mgmt-svc", zap.Error(err))
		return false, bizerr.New(errorcode.ErrCodeSvcMgmt, errorcode.ErrMsgListClustersFailed)
	}
	clusterMap := make(map[uint64]*tidbmgmtservicev1.Cluster, len(clusters))
	for _, cluster := range clusters {
		clusterMap[cluster.ClusterId] = cluster
	}

	resultMap, err := s.billingService.GetTenantSpendLimitsSettings(ctx, tenantID, clusterMap, true)
	if err != nil {
		logger.Error("failed to get all spendlimits from billingSvc", zap.Error(err))
		return false, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgListSpendLimitsFailed)
	}
	return s.billingService.HasFreeCreditQuota(ctx, resultMap), nil
}

func (s *serverlessBillingService) PredictSpendLimit(
	ctx context.Context, req *serverless.PredictSpendLimitReq,
) (_ *serverless.PredictSpendLimitResp, err error) {
	ctx, span := Tracer.Start(ctx, "serverlessBilling.PredictSpendLimit")
	defer func() { tracing.EndSpan(span, err) }()
	logger := log.WithContextL(ctx).
		With(zap.Uint64("org_id", req.OrgId), zap.Uint32("spend_limit", req.SpendLimit), zap.String("region", req.Region))

	param := &serverless.PredictClusterSpendLimitReq{
		OrgId:                     req.OrgId,
		SpendLimit:                req.SpendLimit,
		Region:                    req.Region,
		EnhancedEncryptionEnabled: req.EnhancedEncryptionEnabled,
	}

	return s.predict(ctx, logger, span, param, false)
}

func (s *serverlessBillingService) PredictClusterSpendLimit(
	ctx context.Context, req *serverless.PredictClusterSpendLimitReq,
) (_ *serverless.PredictSpendLimitResp, err error) {
	ctx, span := Tracer.Start(ctx, "serverlessBilling.PredictClusterSpendLimit")
	defer func() { tracing.EndSpan(span, err) }()
	logger := log.WithContextL(ctx).With(zap.Uint64("org_id", req.OrgId), zap.Uint64("cluster_id", req.ClusterId),
		zap.Uint32("spend_limit", req.SpendLimit), zap.String("region", req.Region))

	return s.predict(ctx, logger, span, req, true)
}

func (s *serverlessBillingService) predict(
	ctx context.Context,
	logger *zap.Logger,
	span trace.Span,
	req *serverless.PredictClusterSpendLimitReq,
	isCluster bool,
) (_ *serverless.PredictSpendLimitResp, err error) {
	if err := service.ValidateSpendLimitMax(req.OrgId, req.SpendLimit); err != nil {
		return nil, err
	}
	spendLimit := decimal.NewFromInt(int64(req.SpendLimit)).Div(constant.DolloarToCents)

	span.AddEvent("get clusters from mgmt-svc")
	var provider, region string
	if isCluster {
		cluster, err := s.mgmtFacade.GetCluster(ctx, req.ClusterId, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL)
		if err != nil {
			return nil, bizerr.New(errorcode.ErrCodeSvcMgmt, errorcode.ErrMsgGetClusterFailed)
		}
		provider = cluster.GetCloudProvider()
		region = cluster.GetRegion()
	} else {
		// console pass region in the format of "regions/{provider}-{region}" when calling org predict
		matches := regionPattern.FindStringSubmatch(req.Region)
		if len(matches) != 3 {
			return nil, bizerr.New(errorcode.ErrCodeInvalidRegion, errorcode.ErrMsgInvalidRegion, req.Region)
		}
		provider = matches[1]
		region = matches[2]
	}

	span.AddEvent("get region sku price from billing")
	skuPrice, err := s.billingService.GetSkuUnitPrices(ctx, provider, region, req.OrgId)
	if err != nil {
		logger.Error("failed to get sku unit price", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgGetSkuPricesFailed)
	}

	span.AddEvent("get spendlimits settings from billing")
	var clusterMap map[uint64]*tidbmgmtservicev1.Cluster
	resultMap, err := s.billingService.GetTenantSpendLimitsSettings(ctx, req.OrgId, clusterMap, false)
	if err != nil {
		logger.Error("failed to get all spendlimits from billingSvc", zap.Error(err))
		// return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgListSpendLimitsFailed)
	}
	freeCredit := decimal.Zero
	if isCluster {
		if res, ok := resultMap[req.ClusterId]; ok {
			freeCredit = res.ActiveTotalCredits
		}
	} else if s.billingService.HasFreeCreditQuota(ctx, resultMap) {
		freeCredit = genFreeCredits(skuPrice)
	}

	resp := &serverless.PredictSpendLimitResp{
		SpendLimit: req.SpendLimit,
	}

	if req.EnhancedEncryptionEnabled != nil && *req.EnhancedEncryptionEnabled {
		if req.SpendLimit == 0 {
			return nil, bizerr.New(errorcode.ErrCodeInvalidSpendLimitValue, errorcode.ErrMsgCheckAdvancedEncryption)
		}
		logger.Info("EnhancedEncryptionEnabled, TODO: update advanced encryption predict response ")
	}
	if freeCredit.GreaterThan(decimal.Zero) {
		resp.FreeCredit = dolloarDeToCentsUint(freeCredit)
		resp.FreeRequestUnits = constant.ServerlessFreeQuotaRequestUnit
		resp.FreeColumnStorage = constant.ServerlessFreeQuotaColumnStorage
		resp.FreeRowStorage = constant.ServerlessFreeQuotaRowStorage
	}
	resp.MaxRequestUnits = resp.FreeRequestUnits
	resp.MaxColumnStorage = resp.FreeColumnStorage
	resp.MaxRowStorage = resp.FreeRowStorage

	if skuPrice[constant.SkuTypeRequestUnit].RawPrice.GreaterThan(decimal.Zero) {
		resp.MaxRequestUnits += uint64(spendLimit.Div(skuPrice[constant.SkuTypeRequestUnit].RawPrice).IntPart())
	}
	if skuPrice[constant.SkuTypeColumnStorage].RawPrice.GreaterThan(decimal.Zero) {
		resp.MaxColumnStorage += spendLimit.Div(skuPrice[constant.SkuTypeColumnStorage].RawPrice).InexactFloat64()
	}
	if skuPrice[constant.SkuTypeRowStorage].RawPrice.GreaterThan(decimal.Zero) {
		resp.MaxRowStorage += spendLimit.Div(skuPrice[constant.SkuTypeRowStorage].RawPrice).InexactFloat64()
	}
	return resp, nil
}

func (s *serverlessBillingService) GetSpendLimitDetails(
	ctx context.Context, req *serverless.GetSpendLimitDetailsReq,
) (_ *serverless.GetSpendLimitDetailsResp, err error) {
	ctx, span := Tracer.Start(ctx, "serverlessBilling.GetSpendLimitDetails")
	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", req.ClusterId), zap.Uint64("org_id", req.OrgId))
	defer func() { tracing.EndSpan(span, err) }()

	span.AddEvent("get cluster from mgmt-svc")
	cluster, err := s.mgmtFacade.GetCluster(ctx, req.ClusterId, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if err != nil {
		return nil, bizerr.New(errorcode.ErrCodeSvcMgmt, errorcode.ErrMsgGetClusterFailed)
	}

	if cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMaxRcu() > 0 {
		// essential cluster has no Spending Limit
		return nil, bizerr.New(errorcode.ErrCodeInvalidSpendLimitValue, errorcode.ErrMsgInvalidClusterPlan)
	}

	userStatus := cluster.GetServerlessCluster().GetUserStatus()
	if userStatus != globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		return &serverless.GetSpendLimitDetailsResp{
			ClusterId:  req.ClusterId,
			SpendLimit: 0,
			FreeCredit: 0,
			SpentTotal: 0,
			SpentFree:  &serverless.MoneySpents{},
			SpentPaid:  &serverless.MoneySpents{},
			Usages:     &serverless.ResourceUsages{},
		}, nil
	}

	span.AddEvent("get Spending Limit from billing")
	setting, e := s.billingFacade.GetSpendLimit(ctx, req.OrgId, req.ClusterId)
	if e != nil {
		logger.Error("failed to get spendlimits from billingSvc", zap.Error(e))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgListSpendLimitsFailed)
	}

	var bizData facade.ServerlessSpendLimitBizData
	if err := json.Unmarshal([]byte(setting.BizData), &bizData); err != nil {
		logger.Error("failed to parse biz_data.", zap.Any("content", setting), zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgGetSpendLimitFailed)
	}
	freeCredit := bizData.ActiveTotalCredits
	spendLimit := bizData.InvoiceLimit

	var (
		skuUsage map[constant.SkuType]float64
		o11yErr  error
		wg       sync.WaitGroup
	)
	wg.Add(1)
	go func() {
		defer wg.Done()
		span.AddEvent("get sku storage usage from o11y")
		skuUsage, o11yErr = s.billingService.GetSkuStorageUsage(ctx, cluster)
	}()
	// Some of the synchronous calls below can return early. Always wait for the
	// storage usage goroutine before returning so it cannot outlive this request.
	defer wg.Wait()

	span.AddEvent("get cost from billing")
	info, err := s.billingFacade.GetSpendLimitInfo(ctx, setting)
	if err != nil {
		logger.Error("failed to get spendlimit info from billingSvc", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgGetSpendLimitInfoFailed)
	}

	span.AddEvent("get region sku price from billing")
	provider := cluster.GetCloudProvider()
	region := cluster.GetRegion()
	skuPrice, err := s.billingService.GetSkuUnitPrices(ctx, provider, region, req.OrgId)
	if err != nil {
		logger.Error("failed to get sku unit price", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgGetSkuPricesFailed)
	}

	span.AddEvent("cluster cost explorer from billing")
	costV2, err := s.billingService.CalculateClusterCostThisMonth(
		ctx,
		req.OrgId,
		strconv.FormatUint(req.ClusterId, 10),
		constant.ClusterTypeCluster,
		nil,
		"",
		"",
	)
	if err != nil {
		return nil, err
	}

	wg.Wait()
	if o11yErr != nil {
		return nil, bizerr.New(errorcode.ErrCodeSvcO11y, errorcode.ErrMsgGetSkuUsagesFailed)
	}

	var spendingLimitCreatedAT *timestamppb.Timestamp
	sql := fmt.Sprintf(
		"select * from cluster_tidb_pools where cluster_id=%d and status='synced' and spend_limit_status in ('create_cluster','modify_spend_limit') and type='spend_limit' order by created_at desc limit 1;",
		req.ClusterId,
	)
	if clusterEvents, err := s.clusterTidbPool.Query(ctx, sql); err == nil && len(clusterEvents) > 0 {
		spendingLimitCreatedAT = timestamppb.New(clusterEvents[0].CreatedAt)
	}

	res := s.billingService.GetCostExplorerServerlessSkuCost(
		costV2,
	)
	resV2 := &serverless.GetSpendLimitDetailsResp{
		ClusterId:  req.ClusterId,
		SpendLimit: dolloarDeToCentsUint(spendLimit),
		FreeCredit: dolloarDeToCentsUint(freeCredit),
		SpentTotal: dolloarDeToCentsUint(info.Costs),
		SpentFree: &serverless.MoneySpents{
			RequestUnit:   dolloarDeToCentsUint(res.RuFreeCost),
			RowStorage:    dolloarDeToCentsUint(res.RowStoreFreeCost),
			ColumnStorage: dolloarDeToCentsUint(res.ColumnarFreeCost),
		},
		SpentPaid: &serverless.MoneySpents{
			RequestUnit:   dolloarDeToCentsUint(res.RuSpentCost),
			RowStorage:    dolloarDeToCentsUint(res.RowStoreSpentCost),
			ColumnStorage: dolloarDeToCentsUint(res.ColumnarSpentCost),
		},
		Usages: &serverless.ResourceUsages{
			RequestUnit: uint64(
				res.RuFreeCost.Add(res.RuSpentCost).
					Div(skuPrice[constant.SkuTypeRequestUnit].RawPrice).
					InexactFloat64(),
			),
			RowStorage:    skuUsage[constant.SkuTypeRowStorage],
			ColumnStorage: skuUsage[constant.SkuTypeColumnStorage],
		},
		UpdateAt: spendingLimitCreatedAT,
	}
	logger.Info("GetSpendLimitDetails CostExplorerV2", zap.Any("resp_body", resV2))

	return resV2, nil
}

func (s *serverlessBillingService) GetResourceLimitMode(
	ctx context.Context, req *serverless.GetResourceLimitModeReq,
) (_ *serverless.GetResourceLimitModeResp, err error) {
	ctx, span := Tracer.Start(ctx, "serverlessBilling.GetResourceLimitMode")
	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", req.ClusterId), zap.Uint64("org_id", req.OrgId))
	defer func() { tracing.EndSpan(span, err) }()

	span.AddEvent("get cluster")
	cluster, err := s.clusterRepo.GetClusterByID(ctx, req.ClusterId)
	if err != nil {
		logger.Error("failed to get cluster", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, errorcode.ErrMsgGetClusterFailed)
	}

	mode := serverless.GetResourceLimitModeResp_UNSPECIFIED
	switch cluster.RuMode {
	case constant.ClusterRUModeStandard:
		mode = serverless.GetResourceLimitModeResp_STANDARD
	case constant.ClusterRUModeThrottled:
		mode = serverless.GetResourceLimitModeResp_THROTTLED
	}

	return &serverless.GetResourceLimitModeResp{
		ClusterId: cluster.ID,
		RuMode:    mode,
	}, nil
}

func (s *serverlessBillingService) validTenantBillingAction(ctx context.Context, tenantID uint64) bool {
	action, err := s.billingFacade.BillingActionCheck(ctx, tenantID)
	if err != nil || !action.Pass {
		log.WithContextL(ctx).With(zap.Uint64("tenant_id", tenantID)).
			Info("invalid billing action check", zap.Any("error", err), zap.Any("result", action))
		return false
	}
	return action.Pass
}

/*
	func (s *serverlessBillingService) 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 *serverlessBillingService) GetPricingDetails(
	ctx context.Context, req *serverless.GetPricingDetailsReq,
) (_ *serverless.GetPricingDetailsResp, err error) {
	region := normalizeCloudRegion(req.Region)
	for _, cfg := range config.Global().WorkerConfig.RegionalConfig {
		if cfg.Region == region && cfg.Provider == req.CloudProvider {
			return &serverless.GetPricingDetailsResp{
				Region:               cfg.Region,
				RequestUnits:         cfg.RUPrice,
				RowBasedStorage:      cfg.RowStoragePrice,
				ColumnStorage:        cfg.ColumnStoragePrice,
				CloudProvider:        req.CloudProvider,
				RequestCapacityUnits: cfg.RCUPrice,
			}, nil
		}
	}
	return nil, bizerr.New(
		errorcode.ErrCodeInvalidRegion,
		errorcode.ErrMsgInvalidRegion,
		req.CloudProvider+"/"+req.Region,
	)
}

func (s *serverlessBillingService) SendExpensiveSQLEmail(
	ctx context.Context, req *serverless.ExpensiveSqlEmailReq,
) (_ *serverless.ExpensiveSqlEmailResp, err error) {
	ctx, span := Tracer.Start(ctx, "serverlessBilling.GetSpendLimit")
	logger := log.WithContextL(ctx).With(zap.Uint64("org_id", req.OrgId))
	defer func() { tracing.EndSpan(span, err) }()
	logger.Info("SendExpensiveSQLEmail precess start", zap.Any("preview", req.Preview))

	previousMonday, currentMonday, targetDBFile := s.billingService.GetEmailTimeRange()
	if !req.SkipDownloadS3 {
		span.AddEvent("Download statements & slowlog from o11y s3")
		if err := s.billingService.DownloadS3AndInitDB(ctx, targetDBFile); err != nil {
			return nil, err
		}
	}

	span.AddEvent("Get clusters by org_id from mgmt-svc")
	allClusters, err := s.mgmtFacade.ListAllClustersInOrg(
		ctx,
		req.OrgId,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
	)
	if err != nil {
		logger.Error("Get clusters by org_id from mgmt-svc failed", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcMgmt, errorcode.ErrMsgListClustersFailed)
	}
	allClustersMap := make(map[uint64]*tidbmgmtservicev1.Cluster)
	for _, cluster := range allClusters {
		allClustersMap[cluster.ClusterId] = cluster
	}
	clusterMap := make(map[uint64]*tidbmgmtservicev1.Cluster)
	for _, clusterId := range req.ClusterIds {
		if cluster, ok := allClustersMap[clusterId]; ok {
			clusterMap[clusterId] = cluster
		}
	}
	if len(clusterMap) == 0 {
		logger.Info(errorcode.WarnMsgEmptyClustersOrg)
		return &serverless.ExpensiveSqlEmailResp{}, nil
	}
	logger.Info("Get clusters by org_id from mgmt-svc", zap.Int("cluster_number", len(clusterMap)))

	span.AddEvent("Get projects by org_id from account")
	projectsMap, err := s.accountFacade.ListProjectsMapByOrg(ctx, req.OrgId)
	if err != nil {
		logger.Error("Get projects by org_id from account failed", zap.Error(err))
		return nil, err
	}
	logger.Info("Get projects by org_id from account", zap.Int("projects_number", len(projectsMap)))

	clustersCost := make([]utils.ClusterCost, 0)
	clustersEmailData := make(map[uint64]utils.Cluster)
	for cluster_id, cluster := range clusterMap {
		span.AddEvent("Get clusters spendinglimit & usages from inner function")
		resp, err := s.GetSpendLimitDetails(ctx, &serverless.GetSpendLimitDetailsReq{
			ClusterId: cluster_id,
			OrgId:     req.OrgId,
		})
		if err != nil {
			logger.Error("Get clusters spendinglimit & usages failed.", zap.Error(err))
			return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgGetSpendLimitFailed)
		}
		clusterPlan := constant.SERVERLESS_PLAN_STARTER
		if cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMaxRcu() > 0 {
			clusterPlan = constant.SERVERLESS_PLAN_ESSENTIAL
		}
		if clusterPlan == constant.SERVERLESS_PLAN_ESSENTIAL {
			// avoid ru usage exposed to essential cluster
			continue
		}
		cost := int64(resp.SpentTotal) - int64(resp.FreeCredit)
		if cost < 0 {
			cost = 0
		}
		var freeQuotaUsedRate float64
		if resp.SpentFree != nil {
			freeQuotaUsedRate = float64(resp.SpentFree.RequestUnit+resp.SpentFree.RowStorage) / float64(resp.FreeCredit)
			if freeQuotaUsedRate >= 0.99 {
				freeQuotaUsedRate = 1
			}
		}
		if resp.Usages == nil {
			resp.Usages = &serverless.ResourceUsages{}
		}
		rss := utils.ResourceUsed{
			StorageUsed:       humanize.IBytes(uint64(resp.Usages.RowStorage * humanize.MiByte)),
			RUUsed:            humanize.Comma(int64(resp.Usages.RequestUnit)),
			FreeQuotaUsedRate: fmt.Sprintf("%.2f%%", freeQuotaUsedRate*100),
			Cost:              humanize.CommafWithDigits(float64(cost)/100, 2),
			FreeCredit:        humanize.CommafWithDigits(float64(resp.FreeCredit)/100, 2),
			SpendingLimit:     humanize.CommafWithDigits(float64(resp.SpendLimit)/100, 2),
		}

		clusterName := fmt.Sprintf("%s [%d]", cluster.ClusterName, cluster_id)
		if project, ok := projectsMap[cluster.ProjectId]; ok {
			clusterName = fmt.Sprintf("%s/%s [%d]", project.Name, cluster.ClusterName, cluster_id)
		}

		clustersEmailData[cluster_id] = utils.Cluster{
			ClusterId:   fmt.Sprint(cluster_id),
			ClusterName: clusterName,
			Region:      cluster.Region,
			PlanType:    clusterPlan,
			Resource:    rss,
		}
		clustersCost = append(clustersCost, utils.ClusterCost{
			ClusterId: cluster_id,
			Cost:      cost,
		})
		logger.Info("Get clusters spendinglimit & usages from inner function",
			zap.Uint64("cluster_id", cluster_id),
			zap.Any("clustersEmailData", clustersEmailData[cluster_id]))
	}

	// Only top 10 usage clusters to avoid email too long
	sort.Slice(clustersCost, func(i, j int) bool {
		return clustersCost[i].Cost > clustersCost[j].Cost
	})
	c := int(math.Min(float64(10), float64(len(clustersCost))))
	clustersCost = clustersCost[:c]
	topClusterIds := make(map[uint64]struct{})
	for i := 0; i < len(clustersCost); i++ {
		topClusterIds[clustersCost[i].ClusterId] = struct{}{}
	}

	span.AddEvent("Query local sqlite file")
	for cls_id := range clustersEmailData {
		cls := clustersEmailData[cls_id]
		if _, ok := topClusterIds[cls_id]; !ok {
			continue
		}
		utils.SqliteQuerysLock.RLock()
		for k, v := range utils.SqliteQuerys {
			if strings.Split(k, "_")[1] != cls.Region {
				continue
			}
			query := fmt.Sprintf(v.SqlTemplate, cls.ClusterId)
			results, err := utils.DoSqliteDBQuery(v.DB, query)
			if err != nil {
				logger.Warn("DoSqliteDBQuery failed", zap.Uint64(cls.Region, cls_id), zap.String("errors", err.Error()))
				continue
			}
			switch k {
			case utils.STATEMENT_PREFIX + cls.Region:
				for _, res := range results {
					cls.StatementTemplate = append(cls.StatementTemplate, utils.Query{
						Query:   res.Query,
						Count:   humanize.Comma(int64(res.Count)),
						TotalRU: humanize.Comma(int64(res.Value)),
						MeanRU:  humanize.Comma(int64(res.Value / float64(res.Count))),
					})
				}
			case utils.SLOWLOG_PREFIX + cls.Region:
				for _, res := range results {
					cls.SlowQueries = append(cls.SlowQueries, utils.Query{
						Query: res.Query,
						Count: humanize.Comma(int64(res.Count)),
						P99:   res.Value,
					})
				}
			}
		}
		utils.SqliteQuerysLock.RUnlock()
		clustersEmailData[cls_id] = cls
		logger.Info("Query local sqlite file", zap.Uint64(cls.Region, cls_id), zap.Any("content", cls))
	}

	span.AddEvent("Gen email html template")
	report := utils.ExpensiveSQLEmailReport{
		Clusters: make([]utils.Cluster, 0),
	}
	for _, cc := range clustersCost {
		data := clustersEmailData[cc.ClusterId]
		if len(data.SlowQueries) == 0 && len(data.StatementTemplate) == 0 {
			logger.Info("Skip clusters with empty logs.", zap.Uint64("clusterId", cc.ClusterId))
			continue
		}
		report.Clusters = append(report.Clusters, data)
	}
	if len(report.Clusters) == 0 {
		logger.Info("Skip empty org.", zap.Uint64("orgId", req.OrgId))
		return &serverless.ExpensiveSqlEmailResp{}, nil
	}
	tmpl, err := template.New("report").Funcs(template.FuncMap{"addOne": func(index int) int {
		return index + 1
	}}).Parse(utils.ExpensiveSQLEmailReportHTML)
	if err != nil {
		logger.Error("Parse email html template failed.", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.ErrMsgGenHtmlTemplateFailed)
	}

	logger.Info("Gen email html template", zap.Any("report", report))
	var buf bytes.Buffer
	err = tmpl.Execute(&buf, report)
	if err != nil {
		logger.Error("Gen email html template failed.", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.ErrMsgGenHtmlTemplateFailed)
	}

	span.AddEvent("Send email through account-svc")
	timeRange := fmt.Sprintf(
		"%s:00:00 UTC ~ %s:00:00 UTC",
		previousMonday.Format("2006-01-02 00"),
		currentMonday.Format("2006-01-02 00"),
	)
	tenant, err := s.accountFacade.GetTenant(ctx, req.OrgId)
	if err != nil {
		logger.Error("Get org name failed.", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.ErrMsgGenHtmlTemplateFailed)
	}
	orgName := tenant.Name

	/* for inner notify */
	statusCode, err := sendGrid(buf.String(), timeRange, orgName)
	if err != nil {
		logger.Error("Send email through account-svc", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.ErrMsgSendgridEmailFailed)
	}

	/* for customer org owner notify */
	if !req.Preview {
		emailUniqKey := fmt.Sprintf("%s_%d_%t", targetDBFile, req.OrgId, req.Preview)
		err = s.accountFacade.SendTenantOwnerEmailOnce(ctx, req.OrgId, emailUniqKey,
			utils.SERVERLESS_EXPENSIVE_SQL_EMAIL_TEMPLATE_ID,
			map[string]string{
				"htmlContent": buf.String(),
				"timeRange":   timeRange,
				"orgName":     orgName,
			})
		if err != nil {
			logger.Error("Send email through account-svc", zap.Error(err))
			return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.ErrMsgSendgridEmailFailed)
		}
	}

	logger.Info("Send email through account-svc", zap.Int("status_code", statusCode))
	logger.Info("SendExpensiveSQLEmail precess end")
	return &serverless.ExpensiveSqlEmailResp{}, nil
}

func genFreeCredits(skuPrice map[constant.SkuType]*facade.BillingCharge) decimal.Decimal {
	/* FreeCredit without disaccount, use RawPrice */
	ruPrice := skuPrice[constant.SkuTypeRequestUnit].RawPrice
	rsPrice := skuPrice[constant.SkuTypeRowStorage].RawPrice
	csPrice := skuPrice[constant.SkuTypeColumnStorage].RawPrice
	freeQuotaRU := ruPrice.Mul(decimal.NewFromInt(constant.ServerlessFreeQuotaRequestUnit))
	freeQuotaRS := rsPrice.Mul(decimal.NewFromInt(constant.ServerlessFreeQuotaRowStorage))
	freeQuotaCS := csPrice.Mul(decimal.NewFromInt(constant.ServerlessFreeQuotaColumnStorage))
	return freeQuotaRU.Add(freeQuotaRS).Add(freeQuotaCS)
}

func dolloarDeToCentsUint(dollar decimal.Decimal) uint32 {
	return uint32(dollar.Mul(constant.DolloarToCents).IntPart())
}

func sendGrid(body, timeRange, orgName string) (int, error) {
	apiKey := config.Global().SendGridConfig.APIKey
	from := mail.NewEmail("serverless-inner", "serverless-inner@tidbcloud.com")

	message := mail.NewV3Mail()
	message.SetFrom(from)
	message.SetTemplateID("d-cbb2bcdaabe44c3ab4c9b0d69122271d")

	p := mail.NewPersonalization()
	tos := []*mail.Email{
		mail.NewEmail("jiazhi", "jiazhi.wen@pingcap.cn"),
	}
	p.AddTos(tos...)
	p.SetDynamicTemplateData("htmlContent", body)
	p.SetDynamicTemplateData("timeRange", timeRange)
	p.SetDynamicTemplateData("orgName", orgName)

	message.AddPersonalizations(p)

	client := sendgrid.NewSendClient(apiKey)
	response, err := client.Send(message)
	if err != nil {
		return 0, err
	}
	return response.StatusCode, nil
}

func checkSpendLimitRestriction(req *serverless.SetSpendLimitReq, cluster *tidbmgmtservicev1.Cluster) error {
	// essential plan
	if cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMaxRcu() > 0 {
		return bizerr.New(errorcode.ErrCodeInvalidSpendLimitValue, errorcode.ErrMsgInvalidClusterPlan)
	}

	// starter plan
	if req.GetSpendLimit() == 0 &&
		(req.GetEnhancedEncryptionEnabled() || req.GetZoneRedundantEnabled() ||
			cluster.GetServerlessCluster().GetCluster().GetEnhancedEncryptionEnabled() ||
			cluster.GetServerlessCluster().GetCluster().GetZoneRedundantEnabled()) {
		return bizerr.New(errorcode.ErrCodeInvalidSpendLimitValue, errorcode.ErrMsgCheckAdvancedFeature)
	}
	return nil
}

func normalizeCloudRegion(region string) string {
	region = strings.TrimSpace(region)
	region = strings.TrimPrefix(region, "regions/")
	// Strip known "{provider}-" prefixes. Bare cloud regions such as
	// "us-east-1" or "us-central1" are returned unchanged.
	for _, provider := range []string{
		config.ProviderAliCloud,
		config.ProviderAWS,
		config.ProviderGCP,
		config.ProviderAzure,
	} {
		prefix := provider + "-"
		if strings.HasPrefix(region, prefix) && len(region) > len(prefix) {
			return region[len(prefix):]
		}
	}
	return region
}

func (s *serverlessBillingService) PredictCost(
	ctx context.Context,
	req *serverless.PredictCostReq,
) (*serverless.PredictCostResp, error) {
	if req.Region == "" || req.CloudProvider == "" {
		return nil, bizerr.New(errorcode.ErrCodeInvalidRegion, "empty cloud provider or region")
	}
	region := normalizeCloudRegion(req.Region)
	servicePlan := strings.ToLower(strings.TrimSpace(req.GetServicePlan()))
	var rcuPrice decimal.Decimal
	if servicePlan == "premium" {
		premiumRCUSkuName := fmt.Sprintf("rcu:%s:%s:premium:all:total", req.CloudProvider, region)
		premiumSkuPrice, err := s.billingFacade.CalcFee(ctx, []*facade.BillingReport{
			{
				TenantID: 1,
				Cost:     1,
				Sku:      premiumRCUSkuName,
				Labels: map[string]string{
					"intervalMin":   "1",
					"intervalMax":   "1",
					"intervalValue": "1",
				},
			},
		})
		if err != nil {
			log.WithContextL(ctx).Error("failed to get premium rcu sku unit price",
				zap.String("cloud_provider", req.CloudProvider),
				zap.String("region", region),
				zap.String("service_plan", servicePlan),
				zap.String("sku", premiumRCUSkuName),
				zap.Error(err))
			return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgGetSkuPricesFailed)
		}
		if premiumPrice, ok := premiumSkuPrice[premiumRCUSkuName]; ok && premiumPrice != nil {
			rcuPrice = premiumPrice.RawPrice
		} else {
			log.WithContextL(ctx).Error("premium rcu sku price not found",
				zap.String("cloud_provider", req.CloudProvider),
				zap.String("region", region),
				zap.String("service_plan", servicePlan),
				zap.String("sku", premiumRCUSkuName))
			return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgGetSkuPricesFailed)
		}
	} else {
		skuPrice, err := s.billingService.GetSkuUnitPrices(ctx, req.CloudProvider, region, 1)
		if err != nil {
			log.WithContextL(ctx).Error("failed to get sku unit price",
				zap.String("cloud_provider", req.CloudProvider),
				zap.String("region", region), zap.Error(err))
			return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgGetSkuPricesFailed)
		}
		rcuCharge, ok := skuPrice[constant.SkuTypeRequestCapacityUnit]
		if !ok || rcuCharge == nil {
			log.WithContextL(ctx).Error("essential rcu sku price not available",
				zap.String("cloud_provider", req.CloudProvider),
				zap.String("region", region),
				zap.String("service_plan", servicePlan))
			return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgGetSkuPricesFailed)
		}
		rcuPrice = rcuCharge.RawPrice
	}

	cost := rcuPrice.Mul(decimal.NewFromFloat(float64(req.Rcu))).Mul(decimal.NewFromInt(60))
	resp := &serverless.PredictCostResp{
		CloudProvider:  req.CloudProvider,
		Region:         region,
		Rcu:            req.Rcu,
		RcuCostHourly:  fmt.Sprintf("%.2f", cost.InexactFloat64()),
		RcuCostMonthly: fmt.Sprintf("%.2f", cost.Mul(decimal.NewFromInt(730)).InexactFloat64()),
	}
	return resp, nil
}

func (s *serverlessBillingService) DeleteSpendLimit(
	ctx context.Context, req *serverless.DeleteSpendLimitReq,
) (_ *serverless.GetSpendLimitResp, err error) {
	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", req.ClusterId), zap.Uint64("org_id", req.OrgId))
	setting, err := s.billingFacade.GetSpendLimit(ctx, req.OrgId, req.ClusterId)
	if err != nil {
		var e bizerr.BizError
		errors.As(err, &e)
		if strings.Contains(e.ErrMsg(), "not found") {
			return &serverless.GetSpendLimitResp{
				ClusterId: req.ClusterId,
			}, nil
		}
		logger.Error("call billing GetSpendLimit failed", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgListSpendLimitsFailed)
	}
	if err := s.billingFacade.DeleteSpendLimit(ctx, req.OrgId, setting.SpendLimitSettingID); err != nil {
		logger.Error("call billing DeleteSpendLimit failed", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcBilling, errorcode.ErrMsgModifySpendLimitFailed)
	}
	return &serverless.GetSpendLimitResp{
		ClusterId: req.ClusterId,
	}, nil
}