package cluster

import (
	"context"
	"fmt"
	"sort"
	"strings"

	"github.com/tidbcloud/core-portal/internal/central/config"
	"github.com/tidbcloud/core-portal/internal/central/db"
	"github.com/tidbcloud/core-portal/internal/central/db/dbutil"
	"github.com/tidbcloud/core-portal/internal/central/grpc/grpcli"
	infraapi "github.com/tidbcloud/core-portal/internal/central/infra-api"
	"github.com/tidbcloud/core-portal/internal/central/model/billing"
	centralClusterModel "github.com/tidbcloud/core-portal/internal/central/model/cluster"
	"github.com/tidbcloud/core-portal/internal/constant"
	"github.com/tidbcloud/core-portal/internal/openapi/service"
	"github.com/tidbcloud/core-portal/internal/openapi/service/cvt"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/base"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	entity "github.com/tidbcloud/core-svc/model"
	"github.com/tidbcloud/idl/pbgen/proto/openapi/cm"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/emptypb"
	"k8s.io/utils/pointer"
)

type clusterService struct {
	service.OpenAPIService
	infraAPI      *infraapi.API
	clustersModel centralClusterModel.Model
	billingModel  *billing.Model
	cm.UnimplementedClusterServer
}

// NewService constructs a ClusterService.
func NewService(s service.OpenAPIService, infraAPI *infraapi.API, clustersModel centralClusterModel.Model, billingModel *billing.Model) cm.ClusterServer {
	return &clusterService{
		OpenAPIService: s,
		infraAPI:       infraAPI,
		clustersModel:  clustersModel,
		billingModel:   billingModel,
	}
}

func (s *clusterService) userCluster(ctx context.Context) centralClusterModel.UserContext {
	user := s.CurrentUser(ctx)
	return s.clustersModel.User(ctx, user)
}

func (s *clusterService) checkProjectPermission(ctx context.Context, projectID uint64) bizerr.BizError {
	user := s.CurrentUser(ctx)

	err := dbutil.CheckProjectID(user.Projects, entity.ID(projectID))
	if err != nil {
		return bizerr.New(bizerr.ErrCodePermission, err.Error())
	}
	return nil
}

func (s *clusterService) checkClusterPermission(ctx context.Context, cluster *core.Cluster) bizerr.BizError {
	clusterProjectID := cluster.ProjectId
	return s.checkProjectPermission(ctx, clusterProjectID)
}

func filterCloudProviderByMarketplace(cloudProvider string, tenant *entity.Tenant) bool {
	if tenant != nil && tenant.ThirdPartyAccountProvider != nil && *tenant.ThirdPartyAccountProvider == constant.AWSAccountProvider && cloudProvider != entity.Aws.String() {
		return true
	}
	if tenant != nil && tenant.ThirdPartyAccountProvider != nil && *tenant.ThirdPartyAccountProvider == constant.GCPAccountProvider && cloudProvider != entity.Gcp.String() {
		return true
	}
	return false
}

type NodeProfile struct {
	ClusterType      cm.ClusterType
	CloudProvider    cm.CloudProvider
	Region           string
	ComponentType    core.ComponentType
	NodeSize         string
	StorageSizeRange *cm.NodeStorageSizeRange
	CpuNum           int32
	MemoryGi         int32
}

type ProviderRegionsItemWrapper struct {
	*cm.ListProviderRegionsItem

	CpuNum   int32
	MemoryGi int32
}

func (n *NodeProfile) groupKey() string {
	return fmt.Sprintf("%s-%s-%s", n.ClusterType, n.CloudProvider, n.Region)
}

func getNodeQuantityRange(clusterType cm.ClusterType, componentType core.ComponentType) *cm.NodeQuantityRange {
	var min, step int32
	if clusterType == cm.ClusterType_DEVELOPER {
		min = constant.DevTierMinNodeQuantity
		step = constant.DevTierNodeQuantityStep

		return &cm.NodeQuantityRange{
			Min:  min,
			Step: step,
		}
	}

	switch componentType {
	case core.ComponentType_COMPONENT_TYPE_TIDB:
		min = constant.MinTiDBNodeQuantity
		step = constant.TiDBNodeQuantityStep
	case core.ComponentType_COMPONENT_TYPE_TIKV:
		min = constant.MinTiKVNodeQuantity
		step = constant.TiKVNodeQuantityStep
	case core.ComponentType_COMPONENT_TYPE_TIFLASH:
		min = constant.MinTiFlashNodeQuantity
		step = constant.TiFlashNodeQuantityStep
	}
	return &cm.NodeQuantityRange{
		Min:  min,
		Step: step,
	}
}

func buildNodeProfile(profile *core.NodeProfile4Tenant) *NodeProfile {
	return &NodeProfile{
		ClusterType:   cvt.DeployTypeCore2OpenAPI(profile.DeployType),
		ComponentType: profile.ComponentType,
		NodeSize:      profile.Name,
		CloudProvider: cvt.CloudProviderCore2OpenAPI(profile.CloudProvider),
		Region:        profile.Region,
		StorageSizeRange: &cm.NodeStorageSizeRange{
			Min: profile.MinStorageSizeGi,
			Max: profile.MaxStorageSizeGi,
		},
		CpuNum:   profile.CpuNum,
		MemoryGi: profile.MemoryGi,
	}
}

var (
	clusterTypePriority = map[cm.ClusterType]int{
		cm.ClusterType_DEDICATED: 1,
		cm.ClusterType_DEVELOPER: 2,
	}
)

func (s *clusterService) ListProviderRegions(ctx context.Context, _ *emptypb.Empty) (*cm.ListProviderRegionsResp, error) {
	user := s.CurrentUser(ctx)
	logger := log.WithContextS(ctx)
	orgID := user.TenantID.Uint64()
	listNodeProfiles4TenantReq := &core.ListNodeProfiles4TenantReq{
		TenantId: orgID,
	}
	profiles4TenantResp, err := grpcli.ListNodeProfiles4Tenant(ctx, listNodeProfiles4TenantReq)
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}

	tx := db.NewProvider().ReadDB(ctx)
	tenant, dbErr := dbutil.GetTenantByID(tx, entity.ID(orgID))
	if err != nil {
		logger.Error("failed to get tenant", zap.Uint64("tenant_id", orgID), zap.Error(dbErr))
		st := status.New(codes.Code(bizerr.ErrCodeDB), "failed to get organization info")
		return nil, st.Err()
	}

	nodeProfileMap := make(map[string][]*NodeProfile)
	for _, each := range profiles4TenantResp.Profiles {
		if filterCloudProviderByMarketplace(each.CloudProvider, &tenant) {
			continue
		}

		nodeProfile := buildNodeProfile(each)
		nodeProfileMap[nodeProfile.groupKey()] = append(nodeProfileMap[nodeProfile.groupKey()], nodeProfile)
	}

	items := buildListProviderRegionsItems(nodeProfileMap)
	return &cm.ListProviderRegionsResp{
		Items: items,
	}, nil
}

func sortNodeProfiles(profiles []*NodeProfile) {
	sort.Slice(profiles, func(i, j int) bool {
		if profiles[i].CpuNum != profiles[j].CpuNum {
			return profiles[i].CpuNum < profiles[j].CpuNum
		}
		if profiles[i].MemoryGi != profiles[j].MemoryGi {
			return profiles[i].MemoryGi < profiles[j].MemoryGi
		}
		return profiles[i].NodeSize < profiles[j].NodeSize
	})
}

func buildTiDBProfiles(profilesMapByComponentType map[core.ComponentType][]*NodeProfile) []*cm.TiDBProfile {
	profiles, ok := profilesMapByComponentType[core.ComponentType_COMPONENT_TYPE_TIDB]
	if !ok {
		return nil
	}
	sortNodeProfiles(profiles)

	tidbProfiles := make([]*cm.TiDBProfile, 0)
	for _, each := range profiles {
		item := &cm.TiDBProfile{
			NodeSize:          each.NodeSize,
			NodeQuantityRange: getNodeQuantityRange(each.ClusterType, each.ComponentType),
		}
		tidbProfiles = append(tidbProfiles, item)
	}

	return tidbProfiles
}

func buildTiKVProfiles(profilesMapByComponentType map[core.ComponentType][]*NodeProfile) []*cm.TiKVProfile {
	profiles, ok := profilesMapByComponentType[core.ComponentType_COMPONENT_TYPE_TIKV]
	if !ok {
		return nil
	}
	sortNodeProfiles(profiles)

	tikvProfiles := make([]*cm.TiKVProfile, 0)
	for _, each := range profiles {
		item := &cm.TiKVProfile{
			NodeSize:            each.NodeSize,
			NodeQuantityRange:   getNodeQuantityRange(each.ClusterType, each.ComponentType),
			StorageSizeGibRange: each.StorageSizeRange,
		}
		tikvProfiles = append(tikvProfiles, item)
	}

	return tikvProfiles
}

func buildTiFlashProfiles(profilesMapByComponentType map[core.ComponentType][]*NodeProfile) []*cm.TiFlashProfile {
	profiles, ok := profilesMapByComponentType[core.ComponentType_COMPONENT_TYPE_TIFLASH]
	if !ok {
		return nil
	}
	sortNodeProfiles(profiles)

	tiflashProfiles := make([]*cm.TiFlashProfile, 0)
	for _, each := range profiles {
		item := &cm.TiFlashProfile{
			NodeSize:            each.NodeSize,
			NodeQuantityRange:   getNodeQuantityRange(each.ClusterType, each.ComponentType),
			StorageSizeGibRange: each.StorageSizeRange,
		}
		tiflashProfiles = append(tiflashProfiles, item)
	}

	return tiflashProfiles
}

func sortListProviderRegionsItem(items []*cm.ListProviderRegionsItem) []*cm.ListProviderRegionsItem {
	sort.Slice(items, func(i, j int) bool {
		return clusterTypePriority[items[i].ClusterType] < clusterTypePriority[items[j].ClusterType]
	})
	return items
}

func buildListProviderRegionsItems(groupKey2Profile map[string][]*NodeProfile) []*cm.ListProviderRegionsItem {
	items := make([]*cm.ListProviderRegionsItem, 0)
	for _, profiles := range groupKey2Profile {
		profilesMapByComponentType := make(map[core.ComponentType][]*NodeProfile)

		for _, each := range profiles {
			profilesMapByComponentType[each.ComponentType] = append(profilesMapByComponentType[each.ComponentType], each)
		}

		item := &cm.ListProviderRegionsItem{
			ClusterType:   profiles[0].ClusterType,
			CloudProvider: profiles[0].CloudProvider,
			Region:        profiles[0].Region,
			Tidb:          buildTiDBProfiles(profilesMapByComponentType),
			Tikv:          buildTiKVProfiles(profilesMapByComponentType),
			Tiflash:       buildTiFlashProfiles(profilesMapByComponentType),
		}

		items = append(items, item)
	}

	sortListProviderRegionsItem(items)
	return items
}

func buildDevTierClusterComponents() []*core.ClusterComponent {
	out := make([]*core.ClusterComponent, 0)
	nodeSize := config.Global().ClusterConfig.DevTierNodeSize
	storageSize := config.Global().ClusterConfig.DevTierStorageSizeGib

	tidbComponent := &core.ClusterComponent{
		ComponentType:   core.ComponentType_COMPONENT_TYPE_TIDB,
		NodeProfileName: nodeSize,
		Replicas:        constant.DevTierMinNodeQuantity,
	}

	tikvComponent := &core.ClusterComponent{
		ComponentType:   core.ComponentType_COMPONENT_TYPE_TIKV,
		NodeProfileName: nodeSize,
		Replicas:        constant.DevTierMinNodeQuantity,
		StorageSizeGi:   storageSize,
	}

	tiflashComponent := &core.ClusterComponent{
		ComponentType:   core.ComponentType_COMPONENT_TYPE_TIFLASH,
		NodeProfileName: nodeSize,
		Replicas:        constant.DevTierMinNodeQuantity,
		StorageSizeGi:   storageSize,
	}

	out = append(out, tidbComponent, tikvComponent, tiflashComponent)
	return out
}

func buildDedicatedClusterComponents(req *cm.CreateClusterReq) []*core.ClusterComponent {
	out := make([]*core.ClusterComponent, 0)
	clusterSpec := req.Config
	tidbSpec := clusterSpec.Components.Tidb
	tikvSpec := clusterSpec.Components.Tikv
	tiflashSpec := clusterSpec.Components.Tiflash

	tidbComponent := &core.ClusterComponent{
		ComponentType:   core.ComponentType_COMPONENT_TYPE_TIDB,
		NodeProfileName: tidbSpec.NodeSize,
		Replicas:        tidbSpec.NodeQuantity,
	}

	tikvComponent := &core.ClusterComponent{
		ComponentType:   core.ComponentType_COMPONENT_TYPE_TIKV,
		NodeProfileName: tikvSpec.NodeSize,
		Replicas:        tikvSpec.NodeQuantity,
		StorageSizeGi:   int64(tikvSpec.StorageSizeGib),
	}
	out = append(out, tidbComponent, tikvComponent)

	if tiflashSpec != nil {
		tiflashComponent := &core.ClusterComponent{
			ComponentType:   core.ComponentType_COMPONENT_TYPE_TIFLASH,
			NodeProfileName: tiflashSpec.NodeSize,
			Replicas:        tiflashSpec.NodeQuantity,
			StorageSizeGi:   int64(tiflashSpec.StorageSizeGib),
		}
		out = append(out, tiflashComponent)
	}

	return out
}

// TODO(Louis): try to use validator middleware to validate the input
// https://github.com/grpc-ecosystem/go-grpc-middleware/tree/master/validator
// https://github.com/envoyproxy/protoc-gen-validate
// https://buf.build/envoyproxy/protoc-gen-validate
func validateCreateClusterReq(req *cm.CreateClusterReq) bizerr.BizError {
	clusterConfig := req.Config
	if clusterConfig == nil {
		return bizerr.New(bizerr.ErrCodeParams, "config field cannot be empty")
	}

	clusterType := req.ClusterType
	if clusterType == cm.ClusterType_DEDICATED {
		clusterComponents := clusterConfig.Components

		if clusterComponents == nil {
			return bizerr.New(bizerr.ErrCodeParams, "config.components field cannot be empty")
		}

		if clusterComponents.Tidb == nil {
			return bizerr.New(bizerr.ErrCodeParams, "config.components.tidb field cannot be empty")
		}

		if clusterComponents.Tikv == nil {
			return bizerr.New(bizerr.ErrCodeParams, "config.components.tikv field cannot be empty")
		}
	}

	if len(clusterConfig.RootPassword) == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "root password cannot be empty")
	}
	return nil
}

func (s *clusterService) CreateCluster(ctx context.Context, req *cm.CreateClusterReq) (*cm.CreateClusterResp, error) {
	projectID := req.ProjectId
	err := s.checkProjectPermission(ctx, projectID)
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}

	user := s.CurrentUser(ctx)
	orgID := user.TenantID.Uint64()
	userID := user.ID.Uint64()
	logger := log.WithContextS(ctx)

	err = validateCreateClusterReq(req)
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}

	cloudProvider, err := cvt.CloudProviderOpenAPI2Core(req.CloudProvider)
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}
	deployType, err := cvt.DeployTypeOpenAPI2Core(req.ClusterType)
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}

	createClusterReq := &core.CreateClusterReq{
		TenantId:         orgID,
		ProjectId:        projectID,
		Name:             req.Name,
		RootPassword:     req.Config.RootPassword,
		CloudProvider:    cloudProvider,
		Region:           req.Region,
		DeployType:       deployType,
		IsDevTier:        false, // fill later
		Components:       nil,   // fill later
		CreateById:       userID,
		PrimaryClusterId: 0,
		Port:             req.Config.Port,
		XaccountDeployId: 0,
		TrafficFilter:    &core.TrafficFilter{}, // fill later
		Dryrun:           false,
		BaseReq:          nil,
	}

	// fill IsDevTier
	if createClusterReq.DeployType == core.DeployType_DEPLOY_TYPE_SHARED {
		createClusterReq.IsDevTier = true
	}

	// fill Components
	if createClusterReq.IsDevTier {
		createClusterReq.Components = buildDevTierClusterComponents()
	} else {
		createClusterReq.Components = buildDedicatedClusterComponents(req)
	}

	// fill TrafficFilter
	for _, each := range req.Config.IpAccessList {
		createClusterReq.TrafficFilter.Items = append(createClusterReq.TrafficFilter.Items, &core.TrafficFilterItem{
			Cidr:        each.Cidr,
			Description: each.Description,
		})
	}

	createClusterResp, err := grpcli.CreateCluster(ctx, createClusterReq)
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}

	billingCtx := s.billingModel.User(user)
	if createClusterReq.IsDevTier {
		if err := billingCtx.EnableOrgDevTier(); err != nil {
			// skip error
			logger.Error("EnableOrgDevTier", zap.Uint64("tenantID", orgID), zap.Error(err))
		}
	}

	return &cm.CreateClusterResp{
		Id: createClusterResp.Id,
	}, nil
}

func getCluster(ctx context.Context, projectID, clusterID uint64) (*core.Cluster, bizerr.BizError) {
	filter := &core.ClusterFilter{
		Deleted:    pointer.BoolPtr(false),
		ProjectIds: []uint64{projectID},
		ClusterIds: []uint64{clusterID},
	}

	listReq := &core.ListClustersReq{
		Filter:  filter,
		Pager:   nil,
		Sorters: nil,
		BaseReq: nil,
	}
	listResp, err := grpcli.ListClusters(ctx, listReq)
	if err != nil {
		return nil, err
	}
	cluster, ok := listResp.Clusters[clusterID]
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster is not found, ID=%d", clusterID)
	}
	return cluster, nil
}

func (s *clusterService) GetCluster(ctx context.Context, req *cm.GetClusterReq) (*cm.ClusterItem, error) {
	clusterID := req.ClusterId
	projectID := req.ProjectId
	cluster, err := getCluster(ctx, projectID, clusterID)
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}
	err = s.checkClusterPermission(ctx, cluster)
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}
	resp, err := s.buildClusterItem(ctx, cluster)
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}
	return resp, nil
}

func getClusterType(cluster *core.Cluster) cm.ClusterType {
	if cluster.IsDevTier {
		return cm.ClusterType_DEVELOPER
	}
	return cm.ClusterType_DEDICATED
}

// getClusterNodeName trims the tc name prefix
// "{tcName}-tidb-0" -> "tidb-0"
func getClusterNodeName(podName string) string {
	ar := strings.Split(podName, "-")
	if len(ar) > 2 {
		ar = ar[len(ar)-2:]
	}
	return strings.Join(ar, "-")
}

func buildClusterConfig(cluster *core.Cluster) *cm.GetClusterConfig {
	clusterConfig := &cm.GetClusterConfig{
		Port:       0,   // fill later
		Components: nil, // fill later
	}

	// fill Port
	for _, conn := range cluster.Connections {
		if conn.Type == core.ClusterConnectionType_CLUSTER_CONNECTION_TYPE_PUBLIC {
			clusterConfig.Port = conn.Port
		}
	}

	// fill Components
	componentsSpec := &cm.ClusterComponents{
		Tidb:    nil, // fill later
		Tikv:    nil, // fill later
		Tiflash: nil, // fill later
	}

	for _, component := range cluster.Components {
		switch component.ComponentType {
		case core.ComponentType_COMPONENT_TYPE_TIDB:
			componentsSpec.Tidb = &cm.TiDBComponent{
				NodeSize:     component.NodeProfileName,
				NodeQuantity: component.Replicas,
			}
		case core.ComponentType_COMPONENT_TYPE_TIKV:
			componentsSpec.Tikv = &cm.TiKVComponent{
				NodeSize:       component.NodeProfileName,
				StorageSizeGib: int32(component.StorageSizeGi),
				NodeQuantity:   component.Replicas,
			}
		case core.ComponentType_COMPONENT_TYPE_TIFLASH:
			componentsSpec.Tiflash = &cm.TiFlashComponent{
				NodeSize:       component.NodeProfileName,
				StorageSizeGib: int32(component.StorageSizeGi),
				NodeQuantity:   component.Replicas,
			}
		}
	}
	clusterConfig.Components = componentsSpec

	return clusterConfig
}

func buildConnectionString(cluster *core.Cluster) *cm.ClusterConnectionStrings {
	connectionStrings := &cm.ClusterConnectionStrings{
		DefaultUser: "",  // fill later
		Standard:    nil, // fill later
		VpcPeering:  nil, // fill later
	}

	var defaultUser string
	for _, conn := range cluster.Connections {
		if !conn.CreateSuccess {
			continue
		}

		switch conn.Type {
		case core.ClusterConnectionType_CLUSTER_CONNECTION_TYPE_PUBLIC:
			connectionStrings.Standard = &cm.StandardConnection{
				Host: conn.Host,
				Port: conn.Port,
			}
			// Developer tier cluster default user
			if cluster.DeployType == core.DeployType_DEPLOY_TYPE_SHARED {
				defaultUser = fmt.Sprintf("%s.root", conn.Token)
			} else {
				defaultUser = "root"
			}

		case core.ClusterConnectionType_CLUSTER_CONNECTION_TYPE_PRIVATE:
			connectionStrings.VpcPeering = &cm.VPCPeeringConnection{
				Host: conn.Host,
				Port: conn.Port,
			}
		}
	}
	connectionStrings.DefaultUser = defaultUser

	return connectionStrings
}

func (s *clusterService) buildClusterItemStatus(ctx context.Context, cluster *core.Cluster) (*cm.ClusterItemStatus, bizerr.BizError) {
	clusterItemStatus := &cm.ClusterItemStatus{
		TidbVersion:       cluster.Version,
		ClusterStatus:     cvt.ClusterStatusCore2OpenAPI(cluster),
		NodeMap:           nil, // fill later
		ConnectionStrings: buildConnectionString(cluster),
	}

	// fill NodeMap
	nodeMap, err := s.buildClusterNodeMap(ctx, cluster)
	if err != nil {
		return nil, err
	}
	clusterItemStatus.NodeMap = nodeMap

	return clusterItemStatus, nil
}

func (s *clusterService) buildClusterItem(ctx context.Context, cluster *core.Cluster) (*cm.ClusterItem, bizerr.BizError) {
	item := &cm.ClusterItem{
		Id:              cluster.Id,
		ProjectId:       cluster.ProjectId,
		Name:            cluster.Name,
		ClusterType:     getClusterType(cluster),
		CloudProvider:   cvt.CloudProviderCore2OpenAPI(cluster.CloudProviderName),
		Region:          cluster.RegionName,
		CreateTimestamp: cluster.CreatedAt,
		Config:          buildClusterConfig(cluster),
		Status:          nil, // fill later
	}

	itemStatus, err := s.buildClusterItemStatus(ctx, cluster)
	if err != nil {
		return nil, err
	}
	item.Status = itemStatus

	return item, nil
}

func (s *clusterService) buildClusterNodeMap(ctx context.Context, cluster *core.Cluster) (*cm.ClusterNodeMap, bizerr.BizError) {
	nodeMap := &cm.ClusterNodeMap{
		Tidb:    nil,
		Tikv:    nil,
		Tiflash: nil,
	}

	clusterStatus := cvt.ClusterStatusCore2OpenAPI(cluster)
	if !constant.DisplayNodeMap(clusterStatus) {
		return nodeMap, nil
	}

	logger := log.WithContextS(ctx)
	cr, err := s.infraAPI.GetCluster(ctx, cluster.ProjectId, cluster.Id)
	if err != nil {
		logger.Error("failed to get cluster by infra API", zap.Uint64("cluster_id", cluster.Id), zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeSvcInfraAPI, "failed to get cluster")
	}
	if cr.Status.DB == nil {
		return nodeMap, nil
	}

	var tidbComponent, tikvComponent, tiflashComponent *core.ClusterComponent
	for _, component := range cluster.Components {
		if !config.UserVisibleClusterComponentsContains(component.ComponentType) {
			continue
		}

		switch component.ComponentType {
		case core.ComponentType_COMPONENT_TYPE_TIDB:
			tidbComponent = component
		case core.ComponentType_COMPONENT_TYPE_TIKV:
			tikvComponent = component
		case core.ComponentType_COMPONENT_TYPE_TIFLASH:
			tiflashComponent = component
		}
	}

	TiDBStatus := cr.Status.DB.TiDB
	TiKVStatus := cr.Status.DB.TiKV
	TiFlashStatus := cr.Status.DB.TiFlash
	if TiDBStatus != nil {
		for _, m := range TiDBStatus.Members {
			node := &cm.TiDBNodeMap{
				NodeName:         getClusterNodeName(m.Name),
				AvailabilityZone: m.Labels["cloud.pingcap.com/zone"],
				NodeSize:         tidbComponent.NodeProfileName,
				VcpuNum:          tidbComponent.CpuNum,
				RamBytes:         tidbComponent.MemoryGi * 1024 * 1024 * 1024,
				Status:           cvt.ConvertNodeStatus(m.Status),
			}
			nodeMap.Tidb = append(nodeMap.Tidb, node)
		}
	}
	if TiKVStatus != nil {
		for _, m := range TiKVStatus.Members {
			node := &cm.TiKVNodeMap{
				NodeName:         getClusterNodeName(m.Name),
				AvailabilityZone: m.Labels["cloud.pingcap.com/zone"],
				NodeSize:         tikvComponent.NodeProfileName,
				VcpuNum:          tikvComponent.CpuNum,
				RamBytes:         tikvComponent.MemoryGi * 1024 * 1024 * 1024,
				StorageSizeGib:   int32(tikvComponent.StorageSizeGi),
				Status:           cvt.ConvertNodeStatus(m.Status),
			}
			nodeMap.Tikv = append(nodeMap.Tikv, node)
		}
	}
	if TiFlashStatus != nil {
		for _, m := range TiFlashStatus.Members {
			node := &cm.TiFlashNodeMap{
				NodeName:         getClusterNodeName(m.Name),
				AvailabilityZone: m.Labels["cloud.pingcap.com/zone"],
				NodeSize:         "",
				VcpuNum:          0,
				RamBytes:         0,
				StorageSizeGib:   0,
				Status:           cvt.ConvertNodeStatus(m.Status),
			}
			if tiflashComponent != nil {
				node.NodeSize = tiflashComponent.NodeProfileName
				node.VcpuNum = tiflashComponent.CpuNum
				node.RamBytes = tiflashComponent.MemoryGi * 1024 * 1024 * 1024
				node.StorageSizeGib = int32(tiflashComponent.StorageSizeGi)
			}
			nodeMap.Tiflash = append(nodeMap.Tiflash, node)
		}
	}
	return nodeMap, nil
}

func (s *clusterService) ListClustersOfProject(ctx context.Context, req *cm.ListClustersOfProjectReq) (*cm.ListClustersOfProjectResp, error) {
	projectID := req.ProjectId
	err := s.checkProjectPermission(ctx, projectID)
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}

	page := req.Page
	pageSize := req.PageSize
	if page == 0 {
		page = 1
	}
	if pageSize == 0 {
		pageSize = 10
	}
	pager := &base.Pager{
		Page:     int64(page),
		PageSize: int64(pageSize),
	}
	sorter := &base.Sorter{
		OrderBy: "created_at",
		Sort:    base.Sort_SORT_DESC,
	}
	filter := &core.ClusterFilter{
		ProjectIds: []uint64{projectID},
		Deleted:    pointer.BoolPtr(false),
	}

	listReq := &core.ListClustersReq{
		Filter:  filter,
		Pager:   pager,
		Sorters: []*base.Sorter{sorter},
		BaseReq: nil,
	}
	listResp, err := grpcli.ListClusters(ctx, listReq)
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}
	items := make([]*cm.ClusterItem, 0, len(listResp.Clusters))
	for _, clusterID := range listResp.OrderedClusterIds {
		item, err := s.buildClusterItem(ctx, listResp.Clusters[clusterID])
		if err != nil {
			return nil, service.ConvertBizErr2gRPCErr(err)
		}
		items = append(items, item)
	}
	countResp, err := grpcli.CountClusters(ctx, &core.CountClustersReq{Filter: filter})
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}
	return &cm.ListClustersOfProjectResp{
		Items: items,
		Total: uint32(countResp.Count),
	}, nil
}

func buildUpdateClusterComponents(cluster *core.Cluster, req *cm.UpdateClusterReq) (out []*core.ClusterComponent, changed bool) {
	componentsConfig := req.Config.Components
	tidbSpec := componentsConfig.Tidb
	tikvSpec := componentsConfig.Tikv
	tiflashSpec := componentsConfig.Tiflash

	requestComponentsMap := make(map[core.ComponentType]*core.ClusterComponent)
	existingComponentsMap := make(map[core.ComponentType]bool)

	if tidbSpec != nil {
		requestComponentsMap[core.ComponentType_COMPONENT_TYPE_TIDB] = &core.ClusterComponent{
			ComponentType: core.ComponentType_COMPONENT_TYPE_TIDB,
			Replicas:      tidbSpec.NodeQuantity,
		}
	}
	if tikvSpec != nil {
		requestComponentsMap[core.ComponentType_COMPONENT_TYPE_TIKV] = &core.ClusterComponent{
			ComponentType: core.ComponentType_COMPONENT_TYPE_TIKV,
			Replicas:      tikvSpec.NodeQuantity,
		}
	}
	if tiflashSpec != nil {
		requestComponentsMap[core.ComponentType_COMPONENT_TYPE_TIFLASH] = &core.ClusterComponent{
			ComponentType:   core.ComponentType_COMPONENT_TYPE_TIFLASH,
			NodeProfileName: tiflashSpec.NodeSize,
			Replicas:        tiflashSpec.NodeQuantity,
			StorageSizeGi:   int64(tiflashSpec.StorageSizeGib),
		}
	}

	for _, each := range cluster.Components {
		if !config.UserVisibleClusterComponentsContains(each.ComponentType) {
			continue
		}

		existingComponentsMap[each.ComponentType] = true
		component := &core.ClusterComponent{
			ComponentType:   each.ComponentType,
			Replicas:        each.Replicas,
			NodeProfileName: each.NodeProfileName,
			StorageSizeGi:   each.StorageSizeGi,
		}

		requestComponent, ok := requestComponentsMap[each.ComponentType]
		if ok && requestComponent.Replicas != component.Replicas {
			component.Replicas = requestComponent.Replicas
			changed = true
		}
		out = append(out, component)
	}

	for _, each := range requestComponentsMap {
		if !existingComponentsMap[each.ComponentType] {
			out = append(out, each)
			changed = true
		}
	}

	return out, changed
}

func updateClusterComponents(ctx context.Context, cluster *core.Cluster, req *cm.UpdateClusterReq) bizerr.BizError {
	newComponents, changed := buildUpdateClusterComponents(cluster, req)
	// if no change, return directly
	if !changed {
		return nil
	}

	scaleClusterReq := &core.ScaleClusterReq{
		ClusterId:           cluster.Id,
		NewComponents:       newComponents,
		AllowScaleIn:        false,
		AllowScaleSecondary: false,
		AllowScaleDown:      false,
		BaseReq:             nil,
	}
	_, err := grpcli.ScaleCluster(ctx, scaleClusterReq)
	return err
}

func pauseOrResumeCluster(ctx context.Context, paused bool, clusterID uint64) bizerr.BizError {
	var err bizerr.BizError
	if paused {
		_, err = grpcli.PauseCluster(ctx, clusterID)
	} else {
		_, err = grpcli.ResumeCluster(ctx, clusterID)
	}
	return err
}

func (s *clusterService) UpdateCluster(ctx context.Context, req *cm.UpdateClusterReq) (*emptypb.Empty, error) {
	clusterID := req.ClusterId
	projectID := req.ProjectId
	cluster, err := getCluster(ctx, projectID, clusterID)
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}
	err = s.checkClusterPermission(ctx, cluster)
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}

	clusterConfig := req.Config
	if clusterConfig == nil {
		return &emptypb.Empty{}, nil
	}
	paused := clusterConfig.Paused
	componentsConfig := clusterConfig.Components
	if paused != nil && componentsConfig != nil {
		err = bizerr.New(bizerr.ErrCodeParams, "paused and components cannot be set at the same time")
		return nil, service.ConvertBizErr2gRPCErr(err)
	}

	if componentsConfig != nil {
		err = updateClusterComponents(ctx, cluster, req)
		if err != nil {
			return nil, service.ConvertBizErr2gRPCErr(err)
		}
	}

	if paused != nil {
		err = pauseOrResumeCluster(ctx, *paused, clusterID)
		if err != nil {
			return nil, service.ConvertBizErr2gRPCErr(err)
		}
	}
	return &emptypb.Empty{}, nil
}

func (s *clusterService) DeleteCluster(ctx context.Context, req *cm.DeleteClusterReq) (*emptypb.Empty, error) {
	clusterID := req.ClusterId
	projectID := req.ProjectId
	cluster, err := getCluster(ctx, projectID, clusterID)
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}
	err = s.checkClusterPermission(ctx, cluster)
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}

	user := s.CurrentUser(ctx)
	orgID := user.TenantID
	userContext := s.userCluster(ctx)

	err = userContext.Delete(ctx, orgID, entity.ID(projectID), entity.ID(clusterID))
	if err != nil {
		return nil, service.ConvertBizErr2gRPCErr(err)
	}
	return &emptypb.Empty{}, nil
}