package ops_svc

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"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/repo"
	"github.com/tidbcloud/serverless-service/internal/domain/service"
	"github.com/tidbcloud/serverless-service/internal/utils"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

const (
	checkClusterVersionChangeDeadline = 5 * time.Second
	checkClusterVersionChangeInterval = 1 * time.Second

	updateClusterMessageHeader = "Tenant: %d / %s / %s / %v\nCluster: %d/%s / %s/%s\n"
)

type clusterRpcService struct {
	mgmtSvcFacade facade.MgmtSvcFacade
	accountFacade facade.AccountFacade
	brRpc         rpc.BackupRestoreServiceServer
	billingRpc    serverless.ServerlessBillingServiceServer
	canaryRepo    repo.CanaryRepo
	rpc.UnimplementedClusterCROperationServiceServer
}

func NewClusterRpcService(
	mgmtSvcFacade facade.MgmtSvcFacade,
	accountFacade facade.AccountFacade,
	brRpc rpc.BackupRestoreServiceServer,
	billingRpc serverless.ServerlessBillingServiceServer,
	canaryRepo repo.CanaryRepo,
) rpc.ClusterCROperationServiceServer {
	return &clusterRpcService{
		mgmtSvcFacade: mgmtSvcFacade,
		accountFacade: accountFacade,
		brRpc:         brRpc,
		billingRpc:    billingRpc,
		canaryRepo:    canaryRepo,
	}
}

type clusterRequest interface {
	GetOrgId() string
	GetProjectId() string
	GetClusterId() string
}

func (c *clusterRpcService) validateRequestHeader(
	ctx context.Context,
	request clusterRequest,
) (*tidbmgmtservicev1.Cluster, bizerr.BizError) {
	if request == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "request is nil")
	}
	var (
		orgID, projectID, clusterID uint64
		err                         error
	)
	if orgID, err = strconv.ParseUint(request.GetOrgId(), 10, 64); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid organization id")
	}
	if projectID, err = strconv.ParseUint(request.GetProjectId(), 10, 64); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid project id")
	}
	if clusterID, err = strconv.ParseUint(request.GetClusterId(), 10, 64); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid cluster id")
	}
	cluster, err := c.mgmtSvcFacade.GetClusterBypassCache(
		ctx,
		clusterID,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC,
	)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeRecordNotFound, "falied to get custer")
	}
	if cluster.GetOrganizationId() != orgID {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"cluster organization id not match, expected: %d, actual: %d",
			orgID,
			cluster.GetOrganizationId(),
		)
	}
	if cluster.GetProjectId() != projectID {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"cluster project id not match, expected: %d, actual: %d",
			projectID,
			cluster.GetProjectId(),
		)
	}
	return cluster, nil
}

func (c *clusterRpcService) UpdateCluster(
	ctx context.Context,
	request *rpc.UpdateClusterRequest,
) (*rpc.ClusterDetails, error) {
	logger := log.WithContextL(ctx).With(zap.Any("request", request))
	var err error
	cluster, err := c.validateRequestHeader(ctx, request)
	if err != nil {
		return nil, err
	}
	clusterInfo := cluster.GetServerlessCluster().GetCluster()
	patch := map[string]string{}
	if err = updateClusterTiDBPool(ctx, cluster, patch, request.TidbPool, request.Canary); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "failed to parse update tidb pool request")
	}
	if err = updateClusterRuMode(clusterInfo, patch, request.RuMode); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "failed to parse update ru mode request")
	}
	if err = updateClusterTiDBReplicas(clusterInfo, patch, request.MinTidbReplicas, request.MaxTidbReplicas); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "failed to parse update tidb replicas request")
	}
	spendLimit := c.getSpendLimit(ctx, cluster)
	if err = updateClusterRCU(clusterInfo, patch, spendLimit, request.MinRcu, request.MaxRcu); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "failed to parse update rcu request")
	}
	updateClusterCustomerAlert(clusterInfo, patch, request.EnableCustomerAlert)

	if len(patch) != 0 {
		currentUpdateTS := clusterInfo.GetUpdatedTs()
		if _, err = c.mgmtSvcFacade.UpdateServerlessClusterAnnotations(ctx, cluster.GetClusterId(), patch); err != nil {
			return nil, err
		}
		cluster, err = c.getUpdatedCluster(ctx, cluster.GetClusterId(), currentUpdateTS)
		if err != nil {
			return nil, err
		}
		if cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMinRcu() > 0 ||
			cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMaxRcu() > 0 {
			/* for starter -> essential: try to delete SpendingLimit without error check */
			if _, err = c.billingRpc.DeleteSpendLimit(ctx, &serverless.DeleteSpendLimitReq{
				OrgId:     cluster.GetOrganizationId(),
				ClusterId: cluster.GetClusterId(),
			}); err != nil {
				logger.Error("failed to delete spending limit, err: %w", zap.Error(err))
			} else {
				// spending limit deleted successfully
				logger.Info("spending limit deleted successfully")
			}
		}
		if request.Canary != nil {
			canaryCluster := &repo.CanaryCluster{
				ClusterID: cluster.GetClusterId(),
			}
			if *request.Canary {
				err = c.canaryRepo.AddClusters(ctx, []*repo.CanaryCluster{canaryCluster})
				logger.Info("add canary cluseter", zap.Error(err))
			} else {
				err = c.canaryRepo.RemoveClusters(ctx, []*repo.CanaryCluster{canaryCluster})
				logger.Info("remove canary cluseter", zap.Error(err))
			}
		}
	}

	// Send notification message.
	tenant, _ := c.accountFacade.GetTenant(ctx, cluster.GetOrganizationId())
	if tenant == nil {
		tenant = &facade.Tenant{}
	}
	title := fmt.Sprintf("[%s] Update Cluster CR", strings.ToUpper(config.Global().ServerConfig.Stage))
	message := fmt.Sprintf(
		updateClusterMessageHeader,
		cluster.GetOrganizationId(), tenant.Name, tenant.Email, tenant.Kind,
		cluster.GetClusterId(), cluster.GetClusterName(), cluster.GetCloudProvider(), cluster.GetRegion(),
	)
	if tidbPool, ok := patch[constant.AnnoTidbPoolSelector]; ok {
		message += fmt.Sprintf("TiDB Pool: %s\n", tidbPool)
	}
	if request.RuMode != nil {
		ruModeVal, _ := parseRUMode(*request.RuMode)
		message += fmt.Sprintf("RU Mode: %s\n", ruModeVal)
	}
	if request.MinTidbReplicas != nil {
		message += fmt.Sprintf("Min TiDB Replicas: %d\n", *request.MinTidbReplicas)
	}
	if request.MaxTidbReplicas != nil {
		message += fmt.Sprintf("Max TiDB Replicas: %d\n", *request.MaxTidbReplicas)
	}
	if request.MinRcu != nil {
		message += fmt.Sprintf("Min RCU: %d\n", *request.MinRcu)
	}
	if request.MaxRcu != nil {
		message += fmt.Sprintf("Max RCU: %d\n", *request.MaxRcu)
	}
	if request.EnableCustomerAlert != nil {
		message += fmt.Sprintf("Enable Customer Alert: %s\n", strconv.FormatBool(*request.EnableCustomerAlert))
	}
	go func() {
		utils.SendLarkMessage(title, message, utils.LarkMsgLevelInfo)
	}()
	return c.getClusterDetails(ctx, cluster)
}

func (c *clusterRpcService) getUpdatedCluster(
	ctx context.Context,
	clusterID uint64,
	currentUpdateTS int64,
) (*tidbmgmtservicev1.Cluster, bizerr.BizError) {
	ctx, cancel := context.WithDeadline(ctx, time.Now().Add(checkClusterVersionChangeDeadline))
	defer cancel()
	// must use bypass cache to get latest result.
	cluster, err := c.mgmtSvcFacade.GetClusterBypassCache(
		ctx,
		clusterID,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC,
	)
	if err != nil {
		return nil, err
	}
	if cluster.GetServerlessCluster().GetCluster().GetUpdatedTs() != currentUpdateTS {
		return cluster, nil
	}
	ticker := time.NewTicker(checkClusterVersionChangeInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return cluster, nil
		case <-ticker.C:
			cluster, err = c.mgmtSvcFacade.GetClusterBypassCache(
				ctx,
				clusterID,
				tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC,
			)
			if err != nil {
				return nil, err
			}
			if cluster.GetServerlessCluster().GetCluster().GetUpdatedTs() != currentUpdateTS {
				return cluster, nil
			}
		}
	}
}

func parseTiDBPool(pool rpc.TiDBPool) (string, error) {
	switch pool {
	case rpc.TiDBPool_TIDB_POOL_DEFAULT:
		return constant.TidbPoolNameFree, nil
	case rpc.TiDBPool_TIDB_POOL_VIP:
		return constant.TidbPoolNameVip, nil
	case rpc.TiDBPool_TIDB_POOL_SUPER_VIP:
		return constant.TidbPoolNameSuper, nil
	default:
		return "", errors.New("invalid tidb pool type")
	}
}

var supportedTiDBPoolTransitions = map[string][]string{
	constant.TidbPoolNameFree: {
		constant.TidbPoolNameFree,                             // allow self-transition
		constant.CanaryPoolPrefix + constant.TidbPoolNameFree, // tidb-pool -> canary-tidb-pool
		constant.TidbPoolNameVip,                              // tidb-pool -> vip-tidb-pool
		constant.CanaryPoolPrefix + constant.TidbPoolNameVip,  // tidb-pool -> canary-vip-tidb-pool
	},
	constant.TidbPoolNameVip: {
		constant.TidbPoolNameVip,                               // allow self-transition
		constant.CanaryPoolPrefix + constant.TidbPoolNameVip,   // vip-tidb-pool -> canary-vip-tidb-pool
		constant.TidbPoolNameSuper,                             // vip-tidb-pool -> super-vip-tidb-pool
		constant.CanaryPoolPrefix + constant.TidbPoolNameSuper, // vip-tidb-pool -> canary-super-vip-tidb-pool
	},
	constant.TidbPoolNameSuper: {
		constant.TidbPoolNameSuper,                             // allow self-transition
		constant.CanaryPoolPrefix + constant.TidbPoolNameSuper, // super-vip-tidb-pool -> canary-super-vip-tidb-pool
	},
	constant.CanaryPoolPrefix + constant.TidbPoolNameFree: {
		constant.CanaryPoolPrefix + constant.TidbPoolNameFree, // allow self-transition
		constant.TidbPoolNameFree,                             // canary-tidb-pool -> tidb-pool (downgrade)
		constant.TidbPoolNameVip,                              // canary-tidb-pool -> vip-tidb-pool
		constant.CanaryPoolPrefix + constant.TidbPoolNameVip,  // canary-tidb-pool -> canary-vip-tidb-pool
	},
	constant.CanaryPoolPrefix + constant.TidbPoolNameVip: {
		constant.CanaryPoolPrefix + constant.TidbPoolNameVip,   // allow self-transition
		constant.TidbPoolNameVip,                               // canary-vip-tidb-pool -> vip-tidb-pool (downgrade)
		constant.TidbPoolNameSuper,                             // canary-vip-tidb-pool -> super-vip-tidb-pool
		constant.CanaryPoolPrefix + constant.TidbPoolNameSuper, // canary-vip-tidb-pool -> canary-super-vip-tidb-pool
	},
	constant.CanaryPoolPrefix + constant.TidbPoolNameSuper: {
		constant.CanaryPoolPrefix + constant.TidbPoolNameSuper, // allow self-transition
		constant.TidbPoolNameSuper,                             // canary-super-vip-tidb-pool -> super-vip-tidb-pool (downgrade)
	},
}

var supportedTiDBPoolTransitionsToCanary = map[string][]string{
	constant.TidbPoolNameFree: {
		constant.CanaryPoolPrefix + constant.TidbPoolNameVip, // tidb-pool -> canary-vip-tidb-pool
	},
	constant.TidbPoolNameVip: {
		constant.CanaryPoolPrefix + constant.TidbPoolNameVip,   // vip-tidb-pool -> canary-vip-tidb-pool
		constant.CanaryPoolPrefix + constant.TidbPoolNameSuper, // vip-tidb-pool -> canary-super-vip-tidb-pool
	},
	constant.TidbPoolNameSuper: {
		constant.CanaryPoolPrefix + constant.TidbPoolNameSuper, // super-vip-tidb-pool -> canary-super-vip-tidb-pool
	},
	constant.CanaryPoolPrefix + constant.TidbPoolNameFree: {
		constant.CanaryPoolPrefix + constant.TidbPoolNameVip, // canary-tidb-pool -> canary-vip-tidb-pool
	},
	constant.CanaryPoolPrefix + constant.TidbPoolNameVip: {
		constant.CanaryPoolPrefix + constant.TidbPoolNameSuper, // canary-vip-tidb-pool -> canary-super-vip-tidb-pool
	},
	constant.CanaryPoolPrefix + constant.TidbPoolNameSuper: {
		constant.CanaryPoolPrefix + constant.TidbPoolNameSuper, // allow self-transition
	},
}

func validatePoolTransition(allowedUpgreadPools map[string][]string, currentPool, targetPool string) error {
	supportedTargetPools := allowedUpgreadPools[currentPool]
	if len(supportedTargetPools) == 0 {
		return fmt.Errorf("current tidb pool %s is not supported via ops", currentPool)
	}
	if !slices.Contains(supportedTargetPools, targetPool) {
		return fmt.Errorf(
			"cannot change tidb pool from %s to %s, supported target pools are %v",
			currentPool,
			targetPool,
			supportedTargetPools,
		)
	}
	return nil
}

func updateClusterTiDBPool(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
	annotations map[string]string, pool *rpc.TiDBPool, canary *bool,
) (err error) {
	clusterInfo := cluster.GetServerlessCluster().GetCluster()
	targetPool := clusterInfo.GetAnnotations()[constant.AnnoTidbPoolSelector]
	currentPool := clusterInfo.GetAnnotations()[constant.AnnoTidbPoolSelector]
	if pool != nil {
		targetPool, err = parseTiDBPool(*pool)
		if err != nil {
			return err
		}
	}
	if canary != nil {
		annotations[constant.AnnoCanary] = constant.CanaryPoolUnmarkedFlag
		targetPool = constant.OutOfCanaryPool(targetPool)
		if *canary {
			annotations[constant.AnnoCanary] = constant.CanaryPoolMarkedFlag
			targetPool = constant.IntoCanaryPool(targetPool)
		}
	}
	if targetPool == currentPool {
		return nil
	}
	allowedUpgreadPools := supportedTiDBPoolTransitions
	region := cluster.GetRegion()
	provider := cluster.GetCloudProvider()
	kubernetes := clusterInfo.GetKubernetes()
	if config.GetKubernetesConfig(provider, region, kubernetes).DisableCanaryToStandardPool {
		allowedUpgreadPools = supportedTiDBPoolTransitionsToCanary
	}
	log.WithContextL(ctx).Info("upgrade tidb pools",
		zap.String("clusterID", clusterInfo.GetClusterId()),
		zap.String("currentPool", currentPool),
		zap.String("targetPool", targetPool),
		zap.String("region", region),
		zap.String("provider", provider),
		zap.String("kubernetes", kubernetes),
		zap.Any("allowedUpgreadPools", allowedUpgreadPools[currentPool]),
	)
	if err := validatePoolTransition(allowedUpgreadPools, currentPool, targetPool); err != nil {
		return err
	}
	annotations[constant.AnnoTidbPoolSelector] = targetPool
	return nil
}

func parseRUMode(ruMode rpc.RUMode) (string, error) {
	switch ruMode {
	case rpc.RUMode_RU_MODE_STANDARD:
		return constant.RuLimitModeStandard, nil
	case rpc.RUMode_RU_MODE_THROTTLED:
		return constant.RuLimitModeThrottled, nil
	default:
		return "", errors.New("invalid ru mode")
	}
}

func updateClusterRuMode(
	clusterInfo *regionalserverlessv1.ServerlessRegionalServiceClusterInfo,
	annotations map[string]string, ruMode *rpc.RUMode,
) error {
	if ruMode == nil {
		return nil
	}
	ruModeVal, err := parseRUMode(*ruMode)
	if err != nil {
		return err
	}
	currentRuMode := clusterInfo.GetAnnotations()[constant.AnnoRuLimitMode]
	if currentRuMode == ruModeVal {
		return nil
	}
	if currentRuMode == constant.RuLimitModeStandard && ruModeVal == constant.RuLimitModeThrottled {
		return errors.New("downgrade ru mode from standard to throttle is not allowed")
	}
	annotations[constant.AnnoRuLimitMode] = ruModeVal
	return nil
}

func updateClusterCustomerAlert(
	clusterInfo *regionalserverlessv1.ServerlessRegionalServiceClusterInfo,
	annotations map[string]string, enableCustomerAlert *bool,
) {
	if enableCustomerAlert == nil ||
		clusterInfo.GetAnnotations() == nil {
		return
	}

	alert := strconv.FormatBool(*enableCustomerAlert)
	if clusterInfo.GetAnnotations()[constant.AnnoKeyEnableCustomerAlert] != alert {
		annotations[constant.AnnoKeyEnableCustomerAlert] = alert
	}
}

// getAnnoUint32 parse value in cluster annotation as uint32.
func getAnnoUint32(
	clusterInfo *regionalserverlessv1.ServerlessRegionalServiceClusterInfo,
	key string,
) (*uint32, error) {
	annotations := clusterInfo.GetAnnotations()
	if annotations == nil {
		return nil, nil
	}
	valStr := annotations[key]
	if valStr == "" {
		return nil, nil
	}
	valUint, err := strconv.ParseUint(valStr, 10, 32)
	if err != nil {
		return nil, fmt.Errorf("failed to parse annotation value, key: %s, val: %s, err: %w", key, valStr, err)
	}
	valUint32 := uint32(valUint)
	return &valUint32, nil
}

func updateClusterTiDBReplicas(
	clusterInfo *regionalserverlessv1.ServerlessRegionalServiceClusterInfo,
	patch map[string]string,
	minReplicas, maxReplicas *uint32,
) error {
	if minReplicas == nil && maxReplicas == nil {
		return nil
	}
	currentMinReplicas, err := getAnnoUint32(clusterInfo, constant.AnnoMinTiDBReplicas)
	if err != nil {
		return err
	}
	currentMaxReplicas, err := getAnnoUint32(clusterInfo, constant.AnnoMaxTiDBReplicas)
	if err != nil {
		return err
	}
	var newMinReplicas, newMaxReplicas *uint32
	if minReplicas != nil {
		newMinReplicas = minReplicas
	} else {
		newMinReplicas = currentMinReplicas
	}
	if maxReplicas != nil {
		newMaxReplicas = maxReplicas
	} else {
		newMaxReplicas = currentMaxReplicas
	}
	// Check new replicas setting are legal.
	if newMinReplicas != nil && newMaxReplicas != nil && *newMinReplicas > *newMaxReplicas {
		return errors.New("min replicas can't be larger than max replicas")
	}
	// Check if updates are necessary.
	if newMinReplicas != nil && (currentMinReplicas == nil || *newMinReplicas != *currentMinReplicas) {
		patch[constant.AnnoMinTiDBReplicas] = fmt.Sprintf("%d", *newMinReplicas)
	}
	if newMaxReplicas != nil && (currentMaxReplicas == nil || *newMaxReplicas != *currentMaxReplicas) {
		patch[constant.AnnoMaxTiDBReplicas] = fmt.Sprintf("%d", *newMaxReplicas)
	}
	return nil
}

func updateClusterRCU(
	clusterInfo *regionalserverlessv1.ServerlessRegionalServiceClusterInfo,
	patch map[string]string,
	spendLimit uint32,
	minRCU, maxRCU *uint32,
) error {
	if minRCU == nil && maxRCU == nil {
		return nil
	}
	if minRCU != nil && *minRCU == 0 {
		return errors.New("min_rcu must be non-negative")
	}
	if maxRCU != nil && *maxRCU == 0 {
		return errors.New("max_rcu must be non-negative")
	}
	currentMinRCU := clusterInfo.GetAutoScaling().GetMinRcu()
	currentMaxRCU := clusterInfo.GetAutoScaling().GetMaxRcu()
	// only allow update for essential plan or cluster with Spending Limit set.
	if currentMinRCU == 0 || currentMaxRCU == 0 {
		if minRCU == nil || maxRCU == nil ||
			*minRCU == 0 || *maxRCU == 0 {
			// when upgrade from starter to essential, both min_rcu and max_rcu must be set
			return errors.New("min_rcu and max_rcu must be larger than 0")
		}
		if spendLimit == 0 {
			return errors.New("cannot set rcu for Starter/Free cluster")
		}
	}
	var newMinRCU, newMaxRCU uint32
	if minRCU != nil {
		newMinRCU = *minRCU
	} else {
		newMinRCU = currentMinRCU
	}
	if maxRCU != nil {
		newMaxRCU = *maxRCU
	} else {
		newMaxRCU = currentMaxRCU
	}
	// verify min_rcu < max_rcu
	if newMaxRCU < newMinRCU {
		return errors.New("max_rcu must be larger than min_rcu")
	}
	// Check if updates are necessary.
	if newMinRCU != currentMinRCU {
		patch[constant.AnnoMinRCU] = fmt.Sprintf("%d", newMinRCU)
	}
	if newMaxRCU != currentMaxRCU {
		patch[constant.AnnoMaxRCU] = fmt.Sprintf("%d", newMaxRCU)
	}
	return nil
}

func (c *clusterRpcService) GetCluster(
	ctx context.Context,
	request *rpc.GetClusterRequest,
) (*rpc.ClusterDetails, error) {
	cluster, err := c.validateRequestHeader(ctx, request)
	if err != nil {
		return nil, err
	}
	clusterDetails, err := c.getClusterDetails(ctx, cluster)
	if err != nil {
		return nil, err
	}
	return clusterDetails, nil
}

func (c *clusterRpcService) getClusterDetails(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
) (*rpc.ClusterDetails, bizerr.BizError) {
	clusterInfo := cluster.GetServerlessCluster().GetCluster()
	var err error
	clusterDetails := &rpc.ClusterDetails{
		OrgId:                     fmt.Sprintf("%d", cluster.GetOrganizationId()),
		ProjectId:                 fmt.Sprintf("%d", cluster.GetProjectId()),
		ClusterId:                 fmt.Sprintf("%d", cluster.GetClusterId()),
		KeyspaceToken:             clusterInfo.GetUserPrefix(),
		KeyspaceId:                clusterInfo.GetKeyspaceId(),
		Kubernetes:                clusterInfo.GetKubernetes(),
		Provider:                  cluster.GetCloudProvider(),
		EnhancedEncryptionEnabled: clusterInfo.GetEnhancedEncryptionEnabled(),
		ZoneRedundantEnabled:      clusterInfo.GetZoneRedundantEnabled(),
	}
	// Get autoscaling config.
	if clusterAutoScaling := clusterInfo.GetAutoScaling(); clusterAutoScaling != nil {
		clusterDetails.MinRcu = clusterInfo.GetAutoScaling().GetMinRcu()
		clusterDetails.MaxRcu = clusterInfo.GetAutoScaling().GetMaxRcu()
	}
	// Get info from cluster annotation.
	if clusterAnno := clusterInfo.GetAnnotations(); clusterAnno != nil {
		clusterDetails.TidbPool = clusterAnno[constant.AnnoTidbPoolSelector]
		clusterDetails.RuMode = clusterAnno[constant.AnnoRuLimitMode]
		clusterDetails.MinTidbReplicas, err = getAnnoUint32(clusterInfo, constant.AnnoMinTiDBReplicas)
		if err != nil {
			return nil, bizerr.WrapError(err, "failed to parse min tidb replicas from cluster info")
		}
		clusterDetails.MaxTidbReplicas, err = getAnnoUint32(clusterInfo, constant.AnnoMaxTiDBReplicas)
		if err != nil {
			return nil, bizerr.WrapError(err, "failed to parse max tidb replicas from cluster info")
		}
		clusterDetails.EnableCustomerAlert = clusterAnno[constant.AnnoKeyEnableCustomerAlert] == "true"
	}
	// Get backup policies.
	backupPolicies, err := c.listBackupPolicies(
		ctx,
		clusterDetails.GetOrgId(),
		clusterDetails.GetProjectId(),
		clusterDetails.GetClusterId(),
	)
	if err != nil {
		return nil, bizerr.WrapError(err, "failed to get cluster backup policies")
	}
	clusterDetails.BackupPolicies = backupPolicies
	// Get cluster plan.
	clusterPlan := service.GetClusterPlanWithSpendingLimit(
		cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMinRcu(),
		c.getSpendLimit(ctx, cluster),
	)
	clusterDetails.ClusterPlan = clusterPlan
	return clusterDetails, nil
}

func (c *clusterRpcService) listBackupPolicies(
	ctx context.Context,
	orgID, projectID, clusterID string,
) ([]*rpc.BackupPolicy, error) {
	orgIDUint, err := strconv.ParseUint(orgID, 10, 64)
	if err != nil {
		return nil, err
	}
	projectIDUint, err := strconv.ParseUint(projectID, 10, 64)
	if err != nil {
		return nil, err
	}
	req := &rpc.ListBackupPoliciesRequest{
		OrgId:     orgIDUint,
		ProjectId: projectIDUint,
		ClusterId: clusterID,
	}
	resp, err := c.brRpc.ListBackupPolicies(ctx, req)
	if err != nil {
		return nil, err
	}
	return resp.GetBackupPolicies(), nil
}

func (c *clusterRpcService) getSpendLimit(ctx context.Context, cluster *tidbmgmtservicev1.Cluster) uint32 {
	// ignore error, since some cluster might not have spent limit set.
	spendLimitResp, _ := c.billingRpc.GetSpendLimit(ctx, &serverless.GetSpendLimitReq{
		OrgId:     cluster.GetOrganizationId(),
		ClusterId: cluster.GetClusterId(),
	})
	return spendLimitResp.GetSpendLimit()
}