package cluster

import (
	"context"
	"strconv"
	"sync"
	"time"

	"github.com/samber/lo"
	clusterdedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	regionaldedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	openapidedicatedv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/dedicated/v1beta1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/regionserver"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	clusterrepo "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/cluster"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/enum"
	"github.com/tidbcloud/tidb-management-service/utilities/spu/storagetype"
)

func (s *serviceImpl) Get(
	ctx context.Context,
	req *globaldedicatedv1.GetClusterRequest,
) (*entity.Cluster, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "Domain.GetCluster")
	defer span.End()

	if req.ClusterId == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}

	showDetails := !req.GetIgnoreDetails()
	resp, err := s.List(ctx, &globaldedicatedv1.ListClustersRequest{
		ClusterIds:  []uint64{req.ClusterId},
		ShowDeleted: false, // Get cluster does not support deleted clusters.
		ShowDetails: showDetails,
	})
	if err != nil {
		return nil, err
	}
	if len(resp) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster not found")
	}
	if len(resp) > 1 {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "multiple clusters found")
	}

	return resp[0], nil
}

func (s *serviceImpl) List(
	ctx context.Context,
	req *globaldedicatedv1.ListClustersRequest,
) ([]*entity.Cluster, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "Domain.ListClusters")
	defer span.End()

	// Load clusters from repo first.
	clusters, err := s.clusterRepo.QueryAndLoadComponents(ctx, &clusterrepo.Filter{
		ClusterIds:         req.ClusterIds,
		ProjectIds:         req.ProjectIds,
		OrganizationIds:    req.OrganizationIds,
		RegionNames:        req.RegionIds,
		CloudProviderNames: req.CloudProviderIds,
		States:             req.States,
		ExcludeStates:      req.ExcludeStates,
		ShowDeleted:        req.ShowDeleted,
	})
	if err != nil {
		return nil, err
	}
	if len(clusters) == 0 {
		return nil, nil
	}

	// Group clusters by state.
	activeClusters := make(map[uint64]*entity.Cluster)
	deletedClusters := make(map[uint64]*entity.Cluster)
	creatingClusters := make(map[uint64]*entity.Cluster)
	pauseClusters := make(map[uint64]*entity.Cluster)
	for _, cluster := range clusters {
		switch {
		case cluster.State == globaldedicatedv1.Cluster_CREATING:
			creatingClusters[cluster.ClusterId] = cluster
		case cluster.DeleteTime != nil:
			deletedClusters[cluster.ClusterId] = cluster
		default:
			activeClusters[cluster.ClusterId] = cluster
			// when cluster is pausing or paused, add to pauseClusters
			// will add the pause plan to cluster later
			if cluster.GetState() == globaldedicatedv1.Cluster_PAUSED ||
				cluster.GetState() == globaldedicatedv1.Cluster_PAUSING {
				pauseClusters[cluster.ClusterId] = cluster
			}
		}
	}

	// Load traffic filters from repo.
	clusterIdToTrafficFilters, err := s.trafficFilterRepo.LoadByClusterIds(
		ctx,
		append(lo.Keys(activeClusters), lo.Keys(creatingClusters)...),
		req.ShowDeleted,
	)
	if err != nil {
		return nil, err
	}

	// Load pause plan from repo.
	var pausePlans map[uint64]*entity.ClusterPausePlan
	if len(pauseClusters) > 0 {
		pausePlans, err = s.pausePlanRepo.LoadMany(ctx, lo.Keys(pauseClusters))
		if err != nil {
			return nil, err
		}
		pausePlans = lo.MapValues(pausePlans, func(p *entity.ClusterPausePlan, _ uint64) *entity.ClusterPausePlan {
			entity.EnsureScheduledResumeTime(p, s.cfg.BizConfig.NormalPauseScheduleDay)
			return p
		})
	}

	// Fetch resources from regional service. Caution: High RTT duration.
	var (
		clusterIdToTidbGroups          map[uint64][]*clusterdedicatedv1.TidbGroup
		clusterIdToPrivateLinkServices map[uint64][]*regionaldedicatedv1.PrivateLinkService
		clusterIdToNodeInfos           map[uint64][]*regionaldedicatedv1.ComponentNodeInfo
		regionalClusters               map[uint64]*clusterdedicatedv1.Cluster
	)

	if req.GetShowDetails() {
		// Fetch regional details for active clusters.
		{
			regionOrgToClusters, regionToClusters, regionIdToRegion, err := s.groupClusterByLogicalRegion(
				ctx,
				activeClusters,
			)
			if err != nil {
				return nil, err
			}

			// Use a wait group to wait for all goroutines to complete
			var wg sync.WaitGroup
			wg.Add(4) // We have 4 fetch operations

			var fetchErrs = make([]bizerr.BizError, 4)

			// Fetch regional clusters concurrently
			go func() {
				defer wg.Done()
				var e bizerr.BizError
				regionalClusters, e = s.fetchRegionalClusters(
					ctx,
					regionOrgToClusters,
					regionIdToRegion,
					activeClusters,
				)
				if e != nil {
					fetchErrs[0] = e
				}
			}()

			// Fetch TiDB groups concurrently
			go func() {
				defer wg.Done()
				var e bizerr.BizError
				clusterIdToTidbGroups, e = s.fetchRegionalTidbGroup(ctx, regionToClusters, regionIdToRegion)
				if e != nil {
					fetchErrs[1] = e
				}
			}()

			// Fetch PrivateLink services concurrently
			go func() {
				defer wg.Done()
				var e bizerr.BizError
				clusterIdToPrivateLinkServices, e = s.fetchRegionalPrivateLinkServices(
					ctx,
					regionToClusters,
					regionIdToRegion,
				)
				if e != nil {
					fetchErrs[2] = e
				}
			}()

			// Fetch node info for scaling clusters concurrently
			go func() {
				defer wg.Done()
				var e bizerr.BizError
				scalingClusters := lo.MapValues(
					regionToClusters,
					func(clusters []*entity.Cluster, _ *entity.Region) []*entity.Cluster {
						return lo.Filter(clusters, func(cluster *entity.Cluster, _ int) bool {
							return cluster.State == globaldedicatedv1.Cluster_SCALING
						})
					},
				)
				clusterIdToNodeInfos, e = s.fetchRegionalNodeInfos(ctx, scalingClusters, regionIdToRegion)
				if e != nil {
					fetchErrs[3] = e
				}
			}()

			// Wait for all fetch operations to complete
			wg.Wait()

			// Check if any errors occurred during fetching
			for _, e := range fetchErrs {
				if e != nil {
					return nil, e
				}
			}
		}

		// Fetch regional details for creating clusters.
		{
			regionOrgToClusters, _, regionIdToRegion, err := s.groupClusterByLogicalRegion(ctx, creatingClusters)
			if err != nil {
				return nil, err
			}
			regionalClusters1, err := s.fetchRegionalClusters(
				ctx,
				regionOrgToClusters,
				regionIdToRegion,
				creatingClusters,
			)
			if err != nil {
				return nil, err
			}
			regionalClusters = lo.Assign(regionalClusters, regionalClusters1)
		}
	}

	results := make([]*entity.Cluster, 0, len(clusters))

	// Merge results.
	for clusterId, cluster := range lo.Assign(activeClusters, creatingClusters) {
		results = append(results, s.mergeCluster(
			ctx,
			cluster,
			clusterIdToTrafficFilters[clusterId],
			regionalClusters[clusterId],
			clusterIdToTidbGroups[clusterId],
			clusterIdToPrivateLinkServices[clusterId],
			clusterIdToNodeInfos[clusterId],
			pausePlans[clusterId],
		))
	}

	// build deleted clusters
	for _, cluster := range lo.Values(deletedClusters) {
		cluster.ClusterComponents = buildComponents(
			ctx,
			cluster.State,
			cluster.ClusterComponents,
			nil,
			cluster.GetCloudProviderId(),
			cluster.GetRegionId(),
		)
		results = append(results, cluster)
	}

	return results, nil
}

func (s *serviceImpl) groupClusterByLogicalRegion(
	ctx context.Context,
	clusters map[uint64]*entity.Cluster,
) (regionOrgToClusters map[*entity.Region]map[uint64][]*entity.Cluster, regionToClusters map[*entity.Region][]*entity.Cluster, regionIdToRegion map[globalcommonv1.Region_Name]map[uint64]*entity.Region, err bizerr.BizError) {
	regionOrgToClusters = make(map[*entity.Region]map[uint64][]*entity.Cluster)
	regionToClusters = make(map[*entity.Region][]*entity.Cluster)
	regionIdToRegion = make(map[globalcommonv1.Region_Name]map[uint64]*entity.Region)
	// Group active clusters by logical region and organization.
	//   logicRegion := regionId + organizationId
	for _, cluster := range clusters {
		// create regionOrgToClusters map
		if regionIdToRegion[cluster.RegionId] == nil {
			regionIdToRegion[cluster.RegionId] = make(map[uint64]*entity.Region)
		}

		// Load region entity.
		if regionIdToRegion[cluster.RegionId][cluster.OrganizationId] == nil {
			region, err := s.regionRepo.SelectRegionById(ctx, cluster.RegionId, cluster.OrganizationId)
			if err != nil {
				if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
					tilog.Warn(ctx, "invalid region", zap.String("regionId", cluster.RegionId.String()))
					continue
				}
				return nil, nil, nil, err
			}
			regionIdToRegion[cluster.RegionId][cluster.OrganizationId] = region
		}
		// Group clusters by region and organization.
		region := regionIdToRegion[cluster.RegionId][cluster.OrganizationId]
		if regionOrgToClusters[region] == nil {
			regionOrgToClusters[region] = make(map[uint64][]*entity.Cluster)
		}
		// load region organization clusters
		regionOrgToClusters[region][cluster.OrganizationId] = append(
			regionOrgToClusters[region][cluster.OrganizationId],
			cluster,
		)
		regionToClusters[region] = append(regionToClusters[region], cluster)
	}
	return regionOrgToClusters, regionToClusters, regionIdToRegion, nil
}

func (s *serviceImpl) fetchRegionalTidbGroup(
	ctx context.Context,
	regionToClusters map[*entity.Region][]*entity.Cluster,
	regionIdToRegion map[globalcommonv1.Region_Name]map[uint64]*entity.Region,
) (map[uint64][]*clusterdedicatedv1.TidbGroup, bizerr.BizError) {
	clusterIdToTidbGroups := make(map[uint64][]*clusterdedicatedv1.TidbGroup)

	// map logical region to physical region.
	regions := lo.FlatMap(
		lo.Values(regionIdToRegion),
		func(regionMap map[uint64]*entity.Region, _ int) []*entity.Region {
			return lo.Values(regionMap)
		},
	)

	// Current by physical region.
	regionalTidbGroupResultCh := regionserver.QueryMultiRegional(
		regions,
		func(region *entity.Region, wg *sync.WaitGroup, regionalResultCh chan<- regionserver.RegionalResult) {
			ctx, span := tracer.Start(ctx, "Domain.GetTidbGroupOfCluster")
			defer span.End()

			regionClusters := regionToClusters[region]

			// Current by each cluster.
			for _, cluster := range regionClusters {
				wg.Add(1)
				go func(cluster *entity.Cluster) {
					defer wg.Done()

					listTidbGroupsRequest := &clusterdedicatedv1.ListTidbGroupsRequest{
						ClusterId: strconv.FormatUint(cluster.ClusterId, 10),
					}

					var tidbGroupsResp *clusterdedicatedv1.ListTidbGroupsResponse
					tidbGroupsResp, err := s.regionServerCli.ListTidbGroups(ctx, region, listTidbGroupsRequest)
					if err != nil {
						regionalResultCh <- regionserver.RegionalResult{Err: err}
						return
					}

					regionalResultCh <- regionserver.RegionalResult{
						Items: []any{tidbGroupsResp},
						Err:   nil,
					}
				}(cluster)
			}
		},
	)

	// Loop until the last result.
	for result := range regionalTidbGroupResultCh {
		if result.Err != nil {
			tilog.Error(ctx, "list tidb group failed", zap.Error(result.Err))
			continue
		}
		if result.Items == nil {
			continue
		}

		for _, item := range result.Items {
			tidbGroupsResp, ok := item.(*clusterdedicatedv1.ListTidbGroupsResponse)
			if !ok {
				return nil, bizerr.New(bizerr.ErrCodeUnknown, "unexpected item type")
			}
			if len(tidbGroupsResp.GetTidbGroups()) == 0 {
				tilog.Warn(ctx, "no tidb group found", zap.Any("tidbGroupsResp", tidbGroupsResp))
				continue
			}
			clusterId, rawErr := strconv.ParseUint(tidbGroupsResp.GetTidbGroups()[0].ClusterId, 10, 64)
			if rawErr != nil {
				return nil, bizerr.WrapError(rawErr, "failed to parse cluster id")
			}
			clusterIdToTidbGroups[clusterId] = tidbGroupsResp.GetTidbGroups()
		}
	}
	return clusterIdToTidbGroups, nil
}

func (s *serviceImpl) fetchRegionalPrivateLinkServices(
	ctx context.Context,
	regionToClusters map[*entity.Region][]*entity.Cluster,
	regionIdToRegion map[globalcommonv1.Region_Name]map[uint64]*entity.Region,
) (
	map[uint64][]*regionaldedicatedv1.PrivateLinkService,
	bizerr.BizError,
) {
	clusterIdToPrivateLinkServices := make(map[uint64][]*regionaldedicatedv1.PrivateLinkService)

	regions := lo.FlatMap(
		lo.Values(regionIdToRegion),
		func(regionMap map[uint64]*entity.Region, _ int) []*entity.Region {
			return lo.Values(regionMap)
		},
	)

	regionalPrivateLinkServiceResultCh := regionserver.QueryMultiRegional(
		regions,
		func(region *entity.Region, wg *sync.WaitGroup, regionalResultCh chan<- regionserver.RegionalResult) {
			ctx, span := tracer.Start(ctx, "Domain.GetPrivateLinkServiceOfCluster")
			defer span.End()

			regionClusters := regionToClusters[region]

			for _, cluster := range regionClusters {
				wg.Add(1)
				go func(cluster *entity.Cluster) {
					defer wg.Done()

					listPrivateLinkServicesRequest := &regionaldedicatedv1.ListPrivateLinkServicesRequest{
						ClusterId: strconv.FormatUint(cluster.ClusterId, 10),
					}
					var listPrivateLinkServicesResponse *regionaldedicatedv1.ListPrivateLinkServicesResponse
					listPrivateLinkServicesResponse, err := s.regionServerCli.ListPrivateLinkServices(
						ctx,
						region,
						listPrivateLinkServicesRequest,
					)
					if err != nil {
						regionalResultCh <- regionserver.RegionalResult{Err: err}
						return
					}
					regionalResultCh <- regionserver.RegionalResult{
						Items: []any{listPrivateLinkServicesResponse},
						Err:   nil,
					}
				}(cluster)
			}
		},
	)

	for result := range regionalPrivateLinkServiceResultCh {
		if result.Err != nil {
			tilog.Error(ctx, "list private link service failed", zap.Error(result.Err))
			continue
		}
		if result.Items == nil {
			continue
		}

		for _, item := range result.Items {
			listPrivateLinkServicesResponse, ok := item.(*regionaldedicatedv1.ListPrivateLinkServicesResponse)
			if !ok {
				return nil, bizerr.New(bizerr.ErrCodeUnknown, "unexpected item type")
			}
			if len(listPrivateLinkServicesResponse.PrivateLinkServices) == 0 {
				continue
			}
			clusterId, rawErr := strconv.ParseUint(
				listPrivateLinkServicesResponse.PrivateLinkServices[0].ClusterId,
				10,
				64,
			)
			if rawErr != nil {
				return nil, bizerr.WrapError(rawErr, "failed to parse cluster id")
			}
			clusterIdToPrivateLinkServices[clusterId] = listPrivateLinkServicesResponse.PrivateLinkServices
		}
	}
	return clusterIdToPrivateLinkServices, nil
}

func (s *serviceImpl) fetchRegionalNodeInfos(
	ctx context.Context,
	regionToClusters map[*entity.Region][]*entity.Cluster,
	regionIdToRegion map[globalcommonv1.Region_Name]map[uint64]*entity.Region,
) (
	map[uint64][]*regionaldedicatedv1.ComponentNodeInfo,
	bizerr.BizError,
) {
	clusterIdToNodeInfos := make(map[uint64][]*regionaldedicatedv1.ComponentNodeInfo)

	type componentNodeInfo struct {
		clusterId uint64
		nodeInfos []*regionaldedicatedv1.ComponentNodeInfo
	}

	regions := lo.FlatMap(
		lo.Values(regionIdToRegion),
		func(regionMap map[uint64]*entity.Region, _ int) []*entity.Region {
			return lo.Values(regionMap)
		},
	)

	regionalNodeInfoResultCh := regionserver.QueryMultiRegional(
		regions,
		func(region *entity.Region, wg *sync.WaitGroup, regionalResultCh chan<- regionserver.RegionalResult) {
			ctx, span := tracer.Start(ctx, "Domain.GetNodeInfoForCluster")
			defer span.End()

			regionClusters := regionToClusters[region]
			for _, cluster := range regionClusters {
				wg.Add(1)
				go func(cluster *entity.Cluster) {
					defer wg.Done()

					var nodeInfosResp *regionaldedicatedv1.GetComponentNodeInfosResponse
					nodeInfosResp, err := s.regionServerCli.GetComponentNodeInfo(
						ctx,
						region,
						&regionaldedicatedv1.GetComponentNodeInfosRequest{
							ClusterId: strconv.FormatUint(cluster.ClusterId, 10),
						},
					)
					if err != nil {
						regionalResultCh <- regionserver.RegionalResult{Err: err}
						return
					}
					regionalResultCh <- regionserver.RegionalResult{
						Items: []any{&componentNodeInfo{
							clusterId: cluster.ClusterId,
							nodeInfos: nodeInfosResp.GetComponentNodeInfos(),
						}},
						Err: nil,
					}
				}(cluster)
			}
		},
	)

	for result := range regionalNodeInfoResultCh {
		if result.Err != nil {
			tilog.Error(ctx, "get node info failed", zap.Error(result.Err))
			continue
		}
		if result.Items == nil {
			continue
		}

		for _, item := range result.Items {
			nodeInfo, ok := item.(*componentNodeInfo)
			if !ok {
				return nil, bizerr.New(bizerr.ErrCodeUnknown, "unexpected item type")
			}
			clusterIdToNodeInfos[nodeInfo.clusterId] = nodeInfo.nodeInfos
		}
	}
	return clusterIdToNodeInfos, nil
}

func (s *serviceImpl) fetchRegionalClusters(
	ctx context.Context,
	regionOrgToClusters map[*entity.Region]map[uint64][]*entity.Cluster,
	regionIdToRegion map[globalcommonv1.Region_Name]map[uint64]*entity.Region,
	clusters map[uint64]*entity.Cluster,
) (
	map[uint64]*clusterdedicatedv1.Cluster,
	bizerr.BizError,
) {
	regionalClusters := make(map[uint64]*clusterdedicatedv1.Cluster)

	regions := lo.FlatMap(
		lo.Values(regionIdToRegion),
		func(regionMap map[uint64]*entity.Region, _ int) []*entity.Region {
			return lo.Values(regionMap)
		},
	)

	regionalResultCh := regionserver.QueryMultiRegional(
		regions,
		func(region *entity.Region, wg *sync.WaitGroup, regionalResultCh chan<- regionserver.RegionalResult) {
			ctx, span := tracer.Start(ctx, "Domain.listClustersInRegion")
			defer span.End()

			orgToClusters := regionOrgToClusters[region]
			for orgId, orgClusters := range orgToClusters {
				wg.Add(1)
				go func(orgId uint64, orgClusters []*entity.Cluster) {
					defer wg.Done()

					listClustersRequest := &regionaldedicatedv1.ListClustersRequest{
						TenantId: strconv.FormatUint(orgId, 10),
						// generate project ids
						ProjectIds: lo.Union(lo.FlatMap(orgClusters, func(item *entity.Cluster, _ int) []string {
							return []string{strconv.FormatUint(item.ProjectId, 10)}
						})),
					}
					var resp *clusterdedicatedv1.ListClustersResponse
					resp, err := s.regionServerCli.ListClusters(ctx, region, listClustersRequest)
					if err != nil {
						regionalResultCh <- regionserver.RegionalResult{Err: err}
						return
					}
					regionalResultCh <- regionserver.RegionalResult{Items: lo.ToAnySlice(resp.Clusters)}
				}(orgId, orgClusters)
			}
		},
	)

	for result := range regionalResultCh {
		if result.Err != nil {
			tilog.Error(ctx, "list tidb cluster failed", zap.Error(result.Err))
			continue
		}
		if result.Items == nil {
			continue
		}

		for _, item := range result.Items {
			c, ok := item.(*clusterdedicatedv1.Cluster)
			if !ok {
				return nil, bizerr.New(bizerr.ErrCodeUnknown, "unexpected item type")
			}
			clusterId, rawErr := strconv.ParseUint(c.Id, 10, 64)
			if rawErr != nil {
				return nil, bizerr.WrapError(rawErr, "failed to parse cluster id")
			}
			if _, ok = clusters[clusterId]; !ok {
				tilog.Warn(ctx, "the cluster is not in global service", zap.String("cluster", c.Id))
				continue
			}
			regionalClusters[clusterId] = c
		}
	}
	return regionalClusters, nil
}

func (s *serviceImpl) mergeCluster(
	ctx context.Context,
	cluster *entity.Cluster,
	trafficFilters []*entity.TrafficFilter,
	regionalCluster *clusterdedicatedv1.Cluster,
	tidbGroups []*clusterdedicatedv1.TidbGroup,
	privateLinkServices []*regionaldedicatedv1.PrivateLinkService,
	nodeInfos []*regionaldedicatedv1.ComponentNodeInfo,
	pausePlan *entity.ClusterPausePlan,
) *entity.Cluster {
	if lo.Contains([]entity.State{
		globaldedicatedv1.Cluster_AVAILABLE,
		globaldedicatedv1.Cluster_MODIFYING,
		globaldedicatedv1.Cluster_SCALING,
	}, cluster.State) {
		cluster.HealthState = globaldedicatedv1.Cluster_HEALTHY
	} else {
		cluster.HealthState = globaldedicatedv1.Cluster_UNHEALTHY
	}

	if regionalCluster != nil {
		cluster.ProxyBaseUrl = regionalCluster.GetProxyBaseUrl()
		cluster.TokenBasedAuthEnabled = regionalCluster.GetTokenBasedAuthEnabled()
		cluster.AgentBaseUrl = regionalCluster.GetGatewayBaseUrl()

		// Set InsufficientVmResource flag.
		if cluster.GetState() == globaldedicatedv1.Cluster_CREATING ||
			cluster.GetState() == globaldedicatedv1.Cluster_SCALING ||
			cluster.GetState() == globaldedicatedv1.Cluster_RESUMING {
			timeout, err := time.ParseDuration(s.cfg.BizConfig.InsufficientResourceTimeout)
			if err != nil {
				tilog.Warn(ctx, "failed to parse insufficient resource timeout", zap.Error(err))
			}
			cluster.InsufficientVmResource = regionserver.IsClusterVMInsufficient(regionalCluster, timeout)
		}
	}

	cluster.TidbNodeGroupEndpoints = make([]*globaldedicatedv1.TidbNodeGroup_Endpoint, 0, 1)
	if len(tidbGroups) > 0 {
		for _, tidbGroup := range tidbGroups {
			cluster.TidbNodeGroupEndpoints = append(
				cluster.TidbNodeGroupEndpoints,
				buildTidbNodeGroupEndpoints(ctx, cluster, tidbGroup, trafficFilters, privateLinkServices),
			)
		}
	}

	cluster.ClusterComponents = buildComponents(
		ctx,
		cluster.State,
		cluster.ClusterComponents,
		nodeInfos,
		cluster.GetCloudProviderId(),
		cluster.GetRegionId(),
	)

	cluster.Features = lo.Keys(buildAvailableFeatures(cluster))

	if pausePlan != nil {
		cluster.PausePlan = pausePlan.Cluster_PausePlan
	}

	return cluster
}

func buildAvailableFeatures(c *entity.Cluster) map[globalcommonv1.AvailableFeature]bool {
	features := map[globalcommonv1.AvailableFeature]bool{}
	if c.OnNewArch {
		features[globalcommonv1.AvailableFeature_DISABLE_PUBLIC_LB] = true
	}

	switch c.CloudProviderId {
	case globalcommonv1.CloudProvider_AWS:
		// All AWS clusters support private link
		features[globalcommonv1.AvailableFeature_PRIVATELINK] = true
	case globalcommonv1.CloudProvider_GCP:
		// Only new-arch GCP clusters support private link
		if c.OnNewArch {
			features[globalcommonv1.AvailableFeature_PRIVATELINK] = true
		}
	case globalcommonv1.CloudProvider_AZURE:
		features[globalcommonv1.AvailableFeature_PRIVATELINK] = true
	}

	if c.TokenBasedAuthEnabled {
		features[globalcommonv1.AvailableFeature_DELEGATE_USER] = true
	}

	return features
}

func buildTidbNodeGroupEndpoints(
	ctx context.Context,
	cluster *entity.Cluster,
	tidbGroup *clusterdedicatedv1.TidbGroup,
	trafficFilters []*entity.TrafficFilter,
	privateLinkServices []*regionaldedicatedv1.PrivateLinkService,
) *globaldedicatedv1.TidbNodeGroup_Endpoint {
	tidbGroupId, err := strconv.ParseUint(tidbGroup.GetId(), 10, 64)
	if err != nil {
		tilog.Warn(
			ctx,
			"invalid work pool id",
			zap.String("tidbGroup Id", tidbGroup.GetId()),
			zap.Any("tidbGroup", tidbGroup),
			zap.Error(err),
		)
		return nil
	}

	tidbGroupEndpoints := &globaldedicatedv1.TidbNodeGroup_Endpoint{
		TidbGroupId: tidbGroupId,
		PublicEndpoint: &globaldedicatedv1.TidbNodeGroup_Endpoint_PublicEndpoint{
			Host:           "",  // fill later
			Port:           0,   // fill later
			IpAllowList:    nil, // fill later
			EnablePublicLb: lo.ToPtr(tidbGroup.GetEnablePublicLb()),
		},
		VpcPeeringEndpoint:   nil, // fill later
		PrivateLinkEndpoints: nil, // fill later
	}

	if tidbGroup.GetEnablePublicLb() && tidbGroup.GetPublicLb() != nil {
		tidbGroupEndpoints.PublicEndpoint.Host = tidbGroup.GetPublicLb().GetHost()
		tidbGroupEndpoints.PublicEndpoint.Port = uint32(tidbGroup.GetPublicLb().GetPort())
	}

	// merge IpAllowList settings by user
	for _, trafficFilter := range trafficFilters {
		if trafficFilter.TidbGroupID != tidbGroupId {
			continue
		}

		tidbGroupEndpoints.PublicEndpoint.IpAllowList = lo.Map(
			trafficFilter.Items,
			func(item *entity.TrafficFilterItem, _ int) *globaldedicatedv1.TidbNodeGroup_Endpoint_IpAddress {
				return &globaldedicatedv1.TidbNodeGroup_Endpoint_IpAddress{
					CidrBlock:   item.Cidr,
					Description: item.Description,
				}
			},
		)
		tidbGroupEndpoints.PublicEndpoint.EnablePublicLb = lo.ToPtr(!trafficFilter.DisablePublicLb)
	}

	// fill VpcPeeringEndpoint
	if tidbGroup.GetEnablePrivateLb() && tidbGroup.GetPrivateLb() != nil {
		tidbGroupEndpoints.VpcPeeringEndpoint = &globaldedicatedv1.TidbNodeGroup_Endpoint_VpcPeeringEndpoint{
			Host: tidbGroup.GetPrivateLb().GetHost(),
			Port: uint32(tidbGroup.GetPrivateLb().GetPort()),
		}
	}

	privateLinkService, ok := lo.Find(privateLinkServices, func(item *regionaldedicatedv1.PrivateLinkService) bool {
		return item.GetTidbGroupId() == tidbGroup.GetId()
	})
	if !ok {
		return tidbGroupEndpoints
	}

	switch privateLinkService.Info.(type) {
	case *regionaldedicatedv1.PrivateLinkService_Aws_:
		tidbGroupEndpoints.PrivateLinkEndpoints = []*globaldedicatedv1.TidbNodeGroup_Endpoint_PrivateLinkEndpoint{
			{
				Host: privateLinkService.GetServiceDnsName(),
				Port: cluster.GetPort(),
			},
		}
	case *regionaldedicatedv1.PrivateLinkService_Gcp_:
		endpoints := lo.Map(
			privateLinkService.GetConnections(),
			func(item *regionaldedicatedv1.PrivateLinkService_Connection, _ int) *globaldedicatedv1.TidbNodeGroup_Endpoint_PrivateLinkEndpoint {
				host := privateLinkService.GetHost()
				if item.GetGcp().GetHost() != "" {
					host = item.GetGcp().GetHost()
				}
				return &globaldedicatedv1.TidbNodeGroup_Endpoint_PrivateLinkEndpoint{
					Host: host,
					Port: cluster.GetPort(),
				}
			},
		)
		tidbGroupEndpoints.PrivateLinkEndpoints = append(tidbGroupEndpoints.PrivateLinkEndpoints, endpoints...)
	case *regionaldedicatedv1.PrivateLinkService_Azure_:
		host := privateLinkService.GetHost()
		endpoints := lo.Map(
			privateLinkService.GetConnections(),
			func(item *regionaldedicatedv1.PrivateLinkService_Connection, _ int) *globaldedicatedv1.TidbNodeGroup_Endpoint_PrivateLinkEndpoint {
				if item.GetAzure().GetHost() != "" {
					host = item.GetAzure().GetHost()
				}
				return &globaldedicatedv1.TidbNodeGroup_Endpoint_PrivateLinkEndpoint{
					Host: host,
					Port: cluster.GetPort(),
				}
			},
		)
		tidbGroupEndpoints.PrivateLinkEndpoints = append(tidbGroupEndpoints.PrivateLinkEndpoints, endpoints...)
	}

	return tidbGroupEndpoints
}

func buildComponents(
	ctx context.Context,
	state globaldedicatedv1.Cluster_State,
	components []*entity.ClusterComponent,
	nodeInfos []*regionaldedicatedv1.ComponentNodeInfo,
	cloudProvider globalcommonv1.CloudProvider_Name,
	region globalcommonv1.Region_Name,
) []*globaldedicatedv1.Cluster_ClusterComponent {
	results := make([]*globaldedicatedv1.Cluster_ClusterComponent, 0, len(components))
	var componentToNodeInfo = map[string]map[regionaldedicatedv1.ComponentType]*regionaldedicatedv1.ComponentNodeInfo{}
	for _, nodeInfo := range nodeInfos {
		if componentToNodeInfo[nodeInfo.GetTidbGroupId()] == nil {
			componentToNodeInfo[nodeInfo.GetTidbGroupId()] = map[regionaldedicatedv1.ComponentType]*regionaldedicatedv1.ComponentNodeInfo{}
		}
		componentToNodeInfo[nodeInfo.GetTidbGroupId()][nodeInfo.GetComponentType()] = nodeInfo
	}

	for _, comp := range components {
		if state == globaldedicatedv1.Cluster_SCALING {
			var nodeInfo *regionaldedicatedv1.ComponentNodeInfo
			rType := componentTypeGlobalToRegional(comp.GetComponentType())
			rNodeInfo := componentToNodeInfo[strconv.FormatUint(comp.GetTidbGroupId(), 10)]
			switch {
			case rNodeInfo[rType] != nil:
				nodeInfo = rNodeInfo[rType]
			case componentToNodeInfo[""][rType] != nil:
				nodeInfo = componentToNodeInfo[""][rType]
			default:
				nodeInfo = nil
			}
			upd := buildUptodateQuantity(ctx, comp, nodeInfo)
			rd := buildRemainingDeletionQuantity(ctx, comp, nodeInfo, upd)
			// no need to update progress if the progress is already completed.
			if upd != int32(comp.GetCount()) || rd != 0 {
				comp.ModifyProgress = &globaldedicatedv1.Cluster_ClusterComponent_ModifyProgress{
					UptodateQuantity:          upd,
					RemainingDeletionQuantity: rd,
				}
			}
		}

		comp.MeteringSkuKeys = entity.BuildComputeSkus(
			cloudProvider.String(),
			region.String(),
			comp.GetInstanceType(),
			comp.GetComponentType(),
		)

		if _, ok := entity.ComponentsHaveStorage[comp.GetComponentType()]; !ok {
			results = append(results, comp)
			continue
		}

		dataVt := comp.GetStorages()[globaldedicatedv1.Sku_DATA.String()].GetVolumeType()
		raftVt := comp.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()].GetVolumeType()
		var storageType openapidedicatedv1beta1.Cluster_StorageNodeSetting_StorageType
		switch comp.GetComponentType() {
		case globaldedicatedv1.ComponentType_TIKV:
			storageType = storagetype.BuildCurrentTikvStorageType(dataVt, raftVt)
		case globaldedicatedv1.ComponentType_TIFLASH:
			storageType = storagetype.BuildCurrentTiflashStorageType(dataVt, raftVt)
		}

		for diskType, storage := range comp.GetStorages() {
			storage.MeteringSkuKeys = entity.BuildStorageSkus(
				cloudProvider.String(),
				region.String(),
				storageType.String(),
				diskType,
				storage.GetVolumeType().String(),
				storage.GetSizeGi(),
				storage.GetIops(),
				storage.GetThroughputMibps(),
				storage.GetAdditionalIops(),
			)
		}

		results = append(results, comp)
	}

	return results
}

func buildUptodateQuantity(
	ctx context.Context,
	component *entity.ClusterComponent,
	nodeInfo *regionaldedicatedv1.ComponentNodeInfo,
) int32 {
	if nodeInfo == nil {
		tilog.Warn(
			ctx,
			"node info is nil",
			zap.String("component_type", component.GetComponentType().String()),
			zap.Uint64("component_id", component.Id),
		)
		return int32(component.GetCount()) // return all count
	}

	if componentTypeGlobalToRegional(component.GetComponentType()) != nodeInfo.GetComponentType() {
		tilog.Warn(
			ctx,
			"component type not match",
			zap.String("component_type", component.GetComponentType().String()),
			zap.String("node_info_component_type", nodeInfo.GetComponentType().String()),
		)
		return 0
	}

	updated := int32(0)
	for _, node := range nodeInfo.GetNodes() {
		if node.GetState() != regionaldedicatedv1.ComponentNodeInfo_Node_NORMAL {
			tilog.Warn(ctx, "node not ready", zap.String("node_state", node.GetState().String()), zap.Any("node", node))
			continue
		}
		if node.GetInstanceType() != component.GetInstanceType() {
			tilog.Warn(
				ctx,
				"instance type not match",
				zap.String("node_instance_type", node.GetInstanceType()),
				zap.String("component_instance_type", component.GetInstanceType()),
				zap.Any("node", node),
			)
			continue
		}

		if _, ok := entity.ComponentsHaveStorage[component.GetComponentType()]; !ok {
			updated++
			continue
		}

		// check component storage info.
		isUpdated := func(storage *globaldedicatedv1.Cluster_ClusterComponent_Storage) bool {
			if node.GetStorageIops() != storage.GetIops() {
				tilog.Warn(
					ctx,
					"storage iops not match",
					zap.Uint32("component_iops", storage.GetIops()),
					zap.Uint32("node_iops", node.GetStorageIops()),
					zap.Any("node", node),
				)
				return false
			}
			if node.GetStorageSizeGi() != storage.GetSizeGi() {
				tilog.Warn(
					ctx,
					"storage size not match",
					zap.Uint32("component_size", storage.GetSizeGi()),
					zap.Uint32("node_size", node.GetStorageSizeGi()),
					zap.Any("node", node),
				)
				return false
			}
			if node.GetStorageThroughput() != storage.GetThroughputMibps() {
				tilog.Warn(
					ctx,
					"storage throughput not match",
					zap.Uint32("component_throughput", storage.GetThroughputMibps()),
					zap.Uint32("node_throughput", node.GetStorageThroughput()),
					zap.Any("node", node),
				)
				return false
			}
			if cvtVolumeType(node.GetVolumeType()) != storage.GetVolumeType() {
				tilog.Warn(
					ctx,
					"volume type not match",
					zap.String("component_volume_type", storage.GetVolumeType().String()),
					zap.String("node_volume_type", cvtVolumeType(node.GetVolumeType()).String()),
					zap.Any("node", node),
				)
				return false
			}
			return true
		}

		if !isUpdated(component.GetStorages()[globaldedicatedv1.Sku_DATA.String()]) {
			continue
		}

		updated++
	}

	if updated > int32(component.GetCount()) {
		updated = int32(component.GetCount())
	}

	return updated
}

func cvtVolumeType(volumeType string) globaldedicatedv1.StorageMu_VolumeType {
	t, err := enum.FromString[globaldedicatedv1.StorageMu_VolumeType](volumeType)
	if err != nil {
		return globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED
	}
	return globaldedicatedv1.StorageMu_VolumeType(t)
}

func buildRemainingDeletionQuantity(
	ctx context.Context,
	component *entity.ClusterComponent,
	nodeInfo *regionaldedicatedv1.ComponentNodeInfo,
	uptodate int32,
) int32 {
	if nodeInfo == nil {
		tilog.Warn(
			ctx,
			"node info is nil",
			zap.String("component_type", component.GetComponentType().String()),
			zap.Uint64("component_id", component.Id),
		)
		return 0
	}

	if componentTypeGlobalToRegional(component.GetComponentType()) != nodeInfo.GetComponentType() {
		tilog.Warn(
			ctx,
			"component type not match",
			zap.String("component_type", component.GetComponentType().String()),
			zap.String("node_info_component_type", nodeInfo.GetComponentType().String()),
		)
		return 0
	}

	remainingDelete := int32(0)
	// calculate remaining delete quantity

	// scale up or scale down
	// not show scale up or scale down delete quantity @qqqdan
	// for _, node := range nodeInfo.Nodes {
	// 	// when node in creating state, node instance type is empty
	// 	if node.GetInstanceType() != component.TierUnit.InstanceType.GetInstanceType() && node.GetInstanceType() != "" {
	// 		remainingDelete++
	// 	}
	// }

	// scale in
	if remainingDelete == 0 {
		remainingDelete = int32(len(nodeInfo.Nodes)) - int32(component.GetCount())
	}

	if remainingDelete < 0 {
		remainingDelete = 0
	}

	// among all scale in/out up/down cases, only scale in has remaining delete quantity.
	//   we use uptodate != wanted count to check if the cluster is scaling in.
	if uptodate != int32(component.GetCount()) {
		remainingDelete = 0
	}

	return remainingDelete
}

func componentTypeGlobalToRegional(componentType globaldedicatedv1.ComponentType) regionaldedicatedv1.ComponentType {
	switch componentType {
	case globaldedicatedv1.ComponentType_TIKV:
		return regionaldedicatedv1.ComponentType_TIKV
	case globaldedicatedv1.ComponentType_TIDB:
		return regionaldedicatedv1.ComponentType_TIDB
	case globaldedicatedv1.ComponentType_TIFLASH:
		return regionaldedicatedv1.ComponentType_TIFLASH
	case globaldedicatedv1.ComponentType_TIFLASH_LEARNER:
		return regionaldedicatedv1.ComponentType_TIFLASH_LEARNER
	case globaldedicatedv1.ComponentType_PD:
		return regionaldedicatedv1.ComponentType_PD
	case globaldedicatedv1.ComponentType_TIPROXY:
		return regionaldedicatedv1.ComponentType_TIPROXY
	default:
		return regionaldedicatedv1.ComponentType_COMPONENT_TYPE_UNSPECIFIED
	}
}