package cluster

import (
	"context"
	"strconv"
	"sync"

	"github.com/samber/lo"
	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dataplanemanagerv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/adapter/regionalserver"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	clusterrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/cluster"
	regionrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/region"
	reservedcapacitytemplaterepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/reservedcapacitytemplate"
	resourcepoolrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/resourcepool"
	tidbgrouprepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/tidbgroup"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/doh"
)

var checkDNSPropagation = doh.CheckDnsPropagation

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

	if req.ClusterId == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}
	cluster, err := s.clusterRepo.Get(ctx, req.ClusterId, req.ShowDeleted)
	if err != nil {
		return nil, err
	}
	if cluster == nil {
		tilog.Warn(ctx, "cluster not found", zap.Uint64("cluster_id", req.ClusterId))
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster not found")
	}

	err = s.fullClusterInfo(ctx, []*entity.Cluster{cluster}, req.ShowDeleted)
	if err != nil {
		return nil, err
	}
	if err := s.fullResourcePoolLifecycle(ctx, []*entity.Cluster{cluster}, req.ShowDeleted); err != nil {
		return nil, err
	}
	if err := s.fullResourcePoolDisplayNames(ctx, []*entity.Cluster{cluster}, req.ShowDeleted); err != nil {
		return nil, err
	}

	err = s.fullClusterDualLayerEncryption(ctx, cluster)
	if err != nil {
		return nil, err
	}

	err = s.fullReservedCapacityTemplate(ctx, cluster)
	if err != nil {
		return nil, err
	}

	s.fullConnectionReachability(ctx, cluster)

	return cluster, nil
}

func (s *serviceImpl) fullReservedCapacityTemplate(ctx context.Context, cluster *entity.Cluster) bizerr.BizError {
	if cluster.GetReservedCapacityTemplateId() == 0 {
		return nil
	}
	template, err := s.reservedCapacityTemplateRepo.Load(ctx, cluster.GetReservedCapacityTemplateId(), false)
	if err != nil {
		return bizerr.WrapError(err, "load reserved capacity template")
	}
	cluster.ReservedCapacityTemplate = template
	return nil
}

func (s *serviceImpl) fullReservedCapacityTemplates(ctx context.Context, clusters []*entity.Cluster) bizerr.BizError {
	templateIds := lo.Uniq(lo.FilterMap(clusters, func(cluster *entity.Cluster, _ int) (uint64, bool) {
		return cluster.GetReservedCapacityTemplateId(), cluster.GetReservedCapacityTemplateId() != 0
	}))
	if len(templateIds) == 0 {
		return nil
	}
	templates, _, err := s.reservedCapacityTemplateRepo.Query(ctx, &reservedcapacitytemplaterepo.Filter{
		Ids:         templateIds,
		ShowDeleted: true,
	})
	if err != nil {
		return bizerr.WrapError(err, "load reserved capacity templates")
	}
	templateById := lo.SliceToMap(templates, func(template *entity.ReservedCapacityTemplate) (uint64, *entity.ReservedCapacityTemplate) {
		templateId := template.GetId()
		return templateId, template
	})
	for _, cluster := range clusters {
		if cluster.GetReservedCapacityTemplateId() == 0 {
			continue
		}
		template, ok := templateById[cluster.GetReservedCapacityTemplateId()]
		if !ok {
			return bizerr.New(bizerr.ErrCodeRecordNotFound, "reserved capacity template not found")
		}
		cluster.ReservedCapacityTemplate = template
	}
	return nil
}

func (s *serviceImpl) fullResourcePoolDisplayNames(
	ctx context.Context,
	clusters []*entity.Cluster,
	showDeleted bool,
) bizerr.BizError {
	resourcePoolIDs := lo.Uniq(lo.FilterMap(clusters, func(cluster *entity.Cluster, _ int) (uint64, bool) {
		if cluster == nil ||
			cluster.GetServicePlan() != globalcommonv1.ServicePlan_BYOC ||
			cluster.GetSharedPoolId() == "" {
			return 0, false
		}
		resourcePoolID, err := strconv.ParseUint(cluster.GetSharedPoolId(), 10, 64)
		if err != nil {
			tilog.Warn(ctx, "failed to parse resource pool id",
				zap.Uint64("cluster_id", cluster.GetClusterId()),
				zap.String("shared_pool_id", cluster.GetSharedPoolId()),
				zap.Error(err))
			return 0, false
		}
		return resourcePoolID, true
	}))
	if len(resourcePoolIDs) == 0 {
		return nil
	}
	organizationIDs := lo.Uniq(lo.FilterMap(clusters, func(cluster *entity.Cluster, _ int) (uint64, bool) {
		if cluster == nil ||
			cluster.GetServicePlan() != globalcommonv1.ServicePlan_BYOC ||
			cluster.GetSharedPoolId() == "" {
			return 0, false
		}
		return cluster.GetOrganizationId(), cluster.GetOrganizationId() != 0
	}))

	resourcePools, _, err := s.resourcePoolRepo.Query(ctx, &resourcepoolrepo.Filter{
		Ids:          resourcePoolIDs,
		ServicePlans: []entity.ServicePlan{globalcommonv1.ServicePlan_BYOC},
		OwnerType:    int(globalngv1.ResourcePool_ORGANIZATION),
		OwnerIds:     organizationIDs,
		ShowDeleted:  showDeleted,
		PageSize:     int64(len(resourcePoolIDs)),
	})
	if err != nil {
		return bizerr.WrapError(err, "load resource pools for clusters")
	}
	resourcePoolNames := lo.SliceToMap(resourcePools, func(resourcePool *entity.ResourcePool) (uint64, string) {
		return resourcePool.GetId(), resourcePool.GetName()
	})
	for _, cluster := range clusters {
		if cluster == nil || cluster.GetServicePlan() != globalcommonv1.ServicePlan_BYOC {
			continue
		}
		resourcePoolID, err := strconv.ParseUint(cluster.GetSharedPoolId(), 10, 64)
		if err != nil {
			continue
		}
		cluster.ResourcePoolDisplayName = resourcePoolNames[resourcePoolID]
	}
	return nil
}

func (s *serviceImpl) fullResourcePoolLifecycle(
	ctx context.Context,
	clusters []*entity.Cluster,
	showDeleted bool,
) bizerr.BizError {
	resourcePoolIDs := lo.Uniq(lo.FilterMap(clusters, func(cluster *entity.Cluster, _ int) (uint64, bool) {
		if cluster == nil || cluster.GetSharedPoolId() == "" {
			return 0, false
		}
		resourcePoolID, err := strconv.ParseUint(cluster.GetSharedPoolId(), 10, 64)
		if err != nil {
			tilog.Warn(ctx, "failed to parse resource pool id for lifecycle projection",
				zap.Uint64("cluster_id", cluster.GetClusterId()),
				zap.String("shared_pool_id", cluster.GetSharedPoolId()),
				zap.Error(err))
			return 0, false
		}
		return resourcePoolID, true
	}))
	if len(resourcePoolIDs) == 0 {
		return nil
	}

	resourcePools, _, err := s.resourcePoolRepo.Query(ctx, &resourcepoolrepo.Filter{
		Ids:         resourcePoolIDs,
		ShowDeleted: showDeleted,
	})
	if err != nil {
		return bizerr.WrapError(err, "query resource pool lifecycle")
	}
	resourcePoolsByID := lo.SliceToMap(resourcePools, func(pool *entity.ResourcePool) (uint64, *entity.ResourcePool) {
		poolID := pool.GetId()
		return poolID, pool
	})

	for _, cluster := range clusters {
		if cluster == nil || cluster.GetState() != globalngv1.Cluster_AVAILABLE {
			continue
		}
		resourcePoolID, err := strconv.ParseUint(cluster.GetSharedPoolId(), 10, 64)
		if err != nil {
			continue
		}
		resourcePool, ok := resourcePoolsByID[resourcePoolID]
		if !ok || resourcePool == nil {
			continue
		}

		cluster.AutomaticResumeAt = nil
		switch resourcePool.GetLifecycleState() {
		case globalngv1.ResourcePool_PAUSING:
			cluster.State = globalngv1.Cluster_PAUSING
		case globalngv1.ResourcePool_PAUSED:
			cluster.State = globalngv1.Cluster_PAUSED
			cluster.AutomaticResumeAt = resourcePool.GetAutomaticResumeAt()
		case globalngv1.ResourcePool_RESUMING:
			cluster.State = globalngv1.Cluster_RESUMING
		}
	}
	return nil
}

func (s *serviceImpl) fullClusterInfo(
	ctx context.Context,
	clusters []*entity.Cluster,
	showDeleted bool,
) bizerr.BizError {
	regionalClusters := make(map[uint64][]*entity.Cluster)

	regionIds := lo.UniqMap(clusters, func(cluster *entity.Cluster, index int) uint64 {
		return cluster.RegionId
	})

	regions, err := s.regionRepo.ListRegions(ctx, regionrepo.Filter{
		DBRegionIds: regionIds,
		ShowDeleted: showDeleted,
	})

	if err != nil {
		return bizerr.WrapError(err, "failed to list regions")
	}

	regionsMap := lo.SliceToMap(regions, func(region *entity.Region) (uint64, *entity.Region) {
		return region.RegionID, region
	})

	for _, cluster := range clusters {
		if _, ok := regionsMap[cluster.RegionId]; ok {
			regionalClusters[cluster.RegionId] = append(regionalClusters[cluster.RegionId], cluster)
			continue
		}
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "region not found for cluster")
	}

	regionalResultCh := regionalserver.QueryMultiRegional(
		lo.Values(regionsMap),
		func(region *entity.Region, wg *sync.WaitGroup, regionalResultCh chan<- regionalserver.RegionalResult) {
			ctx, span := tracer.Start(ctx, "ClusterService.fullClusterInfo")
			defer span.End()
			clusters := regionalClusters[region.RegionID]
			for _, cluster := range clusters {
				if cluster.State == globalngv1.Cluster_DELETED {
					tilog.Warn(ctx, "cluster is deleted, skip fullClusterInfo",
						zap.Uint64("cluster_id", cluster.ClusterId))
					continue
				}

				wg.Add(1)

				go func(cluster *entity.Cluster) {
					defer wg.Done()
					tidbGroups, err := s.tidbgroupRepo.Query(ctx, &tidbgrouprepo.Filter{
						ClusterIds:  []uint64{cluster.ClusterId},
						ShowDeleted: showDeleted,
					})
					if err != nil {
						regionalResultCh <- regionalserver.RegionalResult{Err: err}
						return
					}
					if len(tidbGroups) == 0 {
						regionalResultCh <- regionalserver.RegionalResult{Err: bizerr.New(bizerr.ErrCodeRecordNotFound, "tidbgroup not found")}
						return
					}
					for _, tidbGroup := range tidbGroups {
						rtg, err := s.regionalCli.GetTidbGroup(ctx, region, &clusterngv1beta.GetTiDBGroupRequest{
							ClusterId:   strconv.FormatUint(cluster.ClusterId, 10),
							TidbGroupId: strconv.FormatUint(tidbGroup.Id, 10),
						})
						if err != nil {
							regionalResultCh <- regionalserver.RegionalResult{Err: err}
							return
						}
						regionalResultCh <- regionalserver.RegionalResult{
							Items: []any{rtg},
						}
					}
				}(cluster)
			}
		},
	)

	clusterIdMaps := lo.SliceToMap(clusters, func(cluster *entity.Cluster) (uint64, *entity.Cluster) {
		return cluster.ClusterId, cluster
	})
	for regionalResult := range regionalResultCh {
		if regionalResult.Err != nil {
			tilog.Error(ctx, "get tidb group failed", zap.Error(regionalResult.Err))
			continue
		}
		if regionalResult.Items == nil {
			continue
		}
		for _, item := range regionalResult.Items {
			rtg, ok := item.(*clusterngv1beta.TiDBGroup)
			if !ok {
				return bizerr.New(bizerr.ErrCodeUnknown, "unexpected item type")
			}
			tidbgroupIdUint64, err := strconv.ParseUint(rtg.GetId(), 10, 64)
			if err != nil {
				tilog.Error(ctx, "failed to parse tidb group id", zap.Error(err))
				continue
			}
			endpoints := make([]*globalngv1.Endpoint, 0, 2)
			if rtg.GetSqlEndpoint() != nil {
				if rtg.GetSqlEndpoint().GetPublicLb() != nil && rtg.GetSqlEndpoint().GetPublicLb().GetEnabled() {
					publicEndpoint := &globalngv1.Endpoint{
						TidbGroupId: tidbgroupIdUint64,
						PublicEndpoint: &globalngv1.Endpoint_PublicEndpoint{
							EnablePublicLb: &rtg.GetSqlEndpoint().GetPublicLb().Enabled,
							Host:           rtg.GetSqlEndpoint().GetPublicLb().GetHost(),
							Port:           uint32(rtg.GetSqlEndpoint().GetPublicLb().GetPort()),
						},
					}
					if len(rtg.GetSqlEndpoint().GetPublicLb().GetIpAllowList()) > 0 {
						publicEndpoint.PublicEndpoint.IpAllowList = make(
							[]*globalngv1.Endpoint_IpAddress,
							len(rtg.GetSqlEndpoint().GetPublicLb().GetIpAllowList()),
						)
						for i, ip := range rtg.GetSqlEndpoint().GetPublicLb().GetIpAllowList() {
							publicEndpoint.PublicEndpoint.IpAllowList[i] = &globalngv1.Endpoint_IpAddress{
								CidrBlock: ip.GetCidrBlock(),
							}
						}
					}
					endpoints = append(endpoints, publicEndpoint)
				}
				if rtg.GetSqlEndpoint().GetPrivateLb() != nil && rtg.GetSqlEndpoint().GetPrivateLb().GetEnabled() {
					endpoints = append(endpoints, &globalngv1.Endpoint{
						TidbGroupId: tidbgroupIdUint64,
						VpcPeeringEndpoint: &globalngv1.Endpoint_VpcPeeringEndpoint{
							Host: rtg.GetSqlEndpoint().GetPrivateLb().GetHost(),
							Port: uint32(rtg.GetSqlEndpoint().GetPrivateLb().GetPort()),
						},
					})
				}
			}
			if len(endpoints) == 0 {
				endpoints = append(endpoints, &globalngv1.Endpoint{
					TidbGroupId: tidbgroupIdUint64,
				})
			}
			clusterIdUint64, err := strconv.ParseUint(rtg.GetClusterId(), 10, 64)
			if err != nil {
				tilog.Error(ctx, "failed to parse cluster id", zap.Error(err))
				continue
			}
			cluster, ok := clusterIdMaps[clusterIdUint64]
			if !ok {
				tilog.Error(ctx, "cluster not found", zap.Uint64("cluster_id", clusterIdUint64))
				continue
			}
			cluster.Endpoints = append(cluster.Endpoints, endpoints...)
			cluster.SharedPoolId = rtg.GetSharedPoolId()
			cluster.Version = rtg.GetVersion()
			cluster.DataplaneId = rtg.GetDataplaneName()
		}
	}

	s.fullPublicEndpointEgressIPs(ctx, lo.Values(regionsMap), clusters)

	s.fullClusterHighAvailabilityTypes(ctx, clusters, regionsMap)

	return nil
}

func (s *serviceImpl) fullPublicEndpointEgressIPs(
	ctx context.Context,
	regions []*entity.Region,
	clusters []*entity.Cluster,
) {
	type dataplaneKey struct {
		regionID    uint64
		dataplaneID string
	}

	clustersByDataplane := make(map[dataplaneKey][]*entity.Cluster)
	regionsByID := lo.SliceToMap(regions, func(region *entity.Region) (uint64, *entity.Region) {
		return region.RegionID, region
	})

	for _, cluster := range clusters {
		if cluster == nil || cluster.GetDataplaneId() == "" {
			continue
		}
		if !lo.SomeBy(cluster.GetEndpoints(), func(endpoint *globalngv1.Endpoint) bool {
			return endpoint.GetPublicEndpoint() != nil
		}) {
			continue
		}
		key := dataplaneKey{regionID: cluster.RegionId, dataplaneID: cluster.GetDataplaneId()}
		clustersByDataplane[key] = append(clustersByDataplane[key], cluster)
	}

	for key, keyClusters := range clustersByDataplane {
		region, ok := regionsByID[key.regionID]
		if !ok {
			continue
		}

		dataplane, err := s.regionalCli.GetDataPlane(ctx, region, &dataplanemanagerv1.GetDataPlaneRequest{
			DataPlaneId: key.dataplaneID,
		})
		if err != nil {
			tilog.Warn(ctx, "failed to get dataplane for egress ips",
				zap.Uint64("region_id", key.regionID),
				zap.String("dataplane_id", key.dataplaneID),
				zap.Error(err))
			continue
		}

		egressIPs := append([]string(nil), dataplane.GetVpc().GetNatIps()...)
		for _, cluster := range keyClusters {
			for _, endpoint := range cluster.GetEndpoints() {
				if endpoint.GetPublicEndpoint() == nil {
					continue
				}
				endpoint.PublicEndpoint.EgressIps = append([]string(nil), egressIPs...)
			}
		}
	}
}

func (s *serviceImpl) fullClusterHighAvailabilityTypes(
	ctx context.Context,
	clusters []*entity.Cluster,
	regionsMap map[uint64]*entity.Region,
) {
	clustersByRegion := make(map[uint64][]*entity.Cluster)
	sharedPoolIdsByRegion := make(map[uint64][]string)

	for _, cluster := range clusters {
		if cluster.GetState() == globalngv1.Cluster_DELETED ||
			cluster.GetServicePlan() != globalcommonv1.ServicePlan_Essential_V2 ||
			cluster.GetSharedPoolId() == "" {
			continue
		}

		region, ok := regionsMap[cluster.RegionId]
		if !ok {
			tilog.Warn(ctx, "skip high availability type derivation because region is missing",
				zap.Uint64("cluster_id", cluster.GetClusterId()),
				zap.Uint64("region_id", cluster.RegionId))
			continue
		}

		clustersByRegion[region.RegionID] = append(clustersByRegion[region.RegionID], cluster)
		sharedPoolIdsByRegion[region.RegionID] = append(sharedPoolIdsByRegion[region.RegionID], cluster.GetSharedPoolId())
	}

	for regionID, regionClusters := range clustersByRegion {
		region := regionsMap[regionID]
		sharedPoolIds := lo.Uniq(sharedPoolIdsByRegion[regionID])

		sharedPoolsResp, err := s.regionalCli.ListSharedPools(ctx, region, &clusterngv1beta.ListSharedPoolsRequest{
			Ids: sharedPoolIds,
		})
		if err != nil {
			tilog.Warn(ctx, "failed to derive high availability type from shared pools",
				zap.Uint64("region_id", regionID),
				zap.Strings("shared_pool_ids", sharedPoolIds),
				zap.Error(err))
			continue
		}

		sharedPoolById := lo.SliceToMap(sharedPoolsResp.GetSharedPools(),
			func(sharedPool *clusterngv1beta.SharedPool) (string, *clusterngv1beta.SharedPool) {
				sharedPoolId := sharedPool.GetId()
				return sharedPoolId, sharedPool
			})
		for _, cluster := range regionClusters {
			sharedPool, ok := sharedPoolById[cluster.GetSharedPoolId()]
			if !ok {
				tilog.Warn(ctx, "skip high availability type derivation because shared pool is missing",
					zap.Uint64("cluster_id", cluster.GetClusterId()),
					zap.String("shared_pool_id", cluster.GetSharedPoolId()))
				continue
			}

			// DataPlane Manager returns all availability zones as node zones when no zonal restriction is set.
			if len(sharedPool.GetProvider().GetNodeZones()) == 1 {
				cluster.HighAvailabilityType = globalngv1.Cluster_ZONAL
				continue
			}
			cluster.HighAvailabilityType = globalngv1.Cluster_REGIONAL
		}
	}
}

func (s *serviceImpl) fullClusterDualLayerEncryption(ctx context.Context,
	cluster *entity.Cluster) bizerr.BizError {
	if cluster.EncryptionKeyId == 0 {
		cluster.DualLayerDataEncryption = &globalngv1.Cluster_DualLayerDataEncryption{
			Key: &globalngv1.Cluster_DualLayerDataEncryption_NoEncryption{},
		}
		return nil
	}

	keys, err := s.encryptionKeySvc.ListEncryptionByIds(ctx, []uint64{cluster.EncryptionKeyId})
	if err != nil {
		return err
	}
	switch {
	case len(keys) == 0:
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "encryption key not found")
	case keys[0].CustomerManaged:
		providerIdentity, err := s.regionProviderIdentityRepo.Load(ctx, keys[0].ProviderIdentityId)
		if err != nil {
			return err
		}
		cmek, err := s.encryptionKeySvc.CvtEncryptionKeyToCmek(ctx, keys[0], providerIdentity)
		if err != nil {
			return err
		}
		cluster.DualLayerDataEncryption = &globalngv1.Cluster_DualLayerDataEncryption{
			Key: &globalngv1.Cluster_DualLayerDataEncryption_Cmek{
				Cmek: cmek,
			},
		}
	case !keys[0].CustomerManaged:
		cluster.DualLayerDataEncryption = &globalngv1.Cluster_DualLayerDataEncryption{
			Key: &globalngv1.Cluster_DualLayerDataEncryption_DefaultKey{},
		}
	}

	return nil
}

func (s *serviceImpl) fullClustersDualLayerEncryption(ctx context.Context,
	clusters []*entity.Cluster) bizerr.BizError {
	encryptionKeyIds := make([]uint64, 0, len(clusters))
	seenEncryptionKeyIds := make(map[uint64]struct{}, len(clusters))
	for _, cluster := range clusters {
		if cluster.EncryptionKeyId == 0 {
			cluster.DualLayerDataEncryption = &globalngv1.Cluster_DualLayerDataEncryption{
				Key: &globalngv1.Cluster_DualLayerDataEncryption_NoEncryption{},
			}
			continue
		}
		if _, ok := seenEncryptionKeyIds[cluster.EncryptionKeyId]; ok {
			continue
		}
		seenEncryptionKeyIds[cluster.EncryptionKeyId] = struct{}{}
		encryptionKeyIds = append(encryptionKeyIds, cluster.EncryptionKeyId)
	}
	if len(encryptionKeyIds) == 0 {
		return nil
	}

	keys, err := s.encryptionKeySvc.ListEncryptionByIds(ctx, encryptionKeyIds)
	if err != nil {
		return err
	}
	keysById := make(map[uint64]*entity.EncryptionKey, len(keys))
	for _, key := range keys {
		keysById[key.Id] = key
	}

	providerIdentitiesById := make(map[uint64]*entity.RegionProviderIdentity)
	for _, cluster := range clusters {
		if cluster.EncryptionKeyId == 0 {
			continue
		}
		key, ok := keysById[cluster.EncryptionKeyId]
		if !ok {
			return bizerr.New(bizerr.ErrCodeRecordNotFound, "encryption key not found")
		}
		switch {
		case key.CustomerManaged:
			providerIdentity, ok := providerIdentitiesById[key.ProviderIdentityId]
			if !ok {
				var loadErr bizerr.BizError
				providerIdentity, loadErr = s.regionProviderIdentityRepo.Load(ctx, key.ProviderIdentityId)
				if loadErr != nil {
					return loadErr
				}
				providerIdentitiesById[key.ProviderIdentityId] = providerIdentity
			}
			cmek, cvtErr := s.encryptionKeySvc.CvtEncryptionKeyToCmek(ctx, key, providerIdentity)
			if cvtErr != nil {
				return cvtErr
			}
			cluster.DualLayerDataEncryption = &globalngv1.Cluster_DualLayerDataEncryption{
				Key: &globalngv1.Cluster_DualLayerDataEncryption_Cmek{
					Cmek: cmek,
				},
			}
		case !key.CustomerManaged:
			cluster.DualLayerDataEncryption = &globalngv1.Cluster_DualLayerDataEncryption{
				Key: &globalngv1.Cluster_DualLayerDataEncryption_DefaultKey{},
			}
		}
	}

	return nil
}

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

	queryStates := resourcePoolLifecycleQueryStates(req.States)
	queryExcludeStates := resourcePoolLifecycleQueryExcludeStates(req.ExcludeStates)
	resp, err := s.clusterRepo.Query(ctx, &clusterrepo.Filter{
		ClusterIds:         req.ClusterIds,
		OrganizationIds:    req.OrganizationIds,
		SharedPoolIds:      req.ResourcePoolIds,
		RegionNames:        req.RegionIds,
		CloudProviderNames: req.CloudProviderIds,
		States:             queryStates,
		ExcludeStates:      queryExcludeStates,
		ShowDeleted:        req.ShowDeleted,
		ServicePlans:       req.ServicePlans,
	})
	if err != nil {
		return nil, err
	}
	if resp == nil {
		return nil, nil
	}

	err = s.fullClusterInfo(ctx, resp, req.ShowDeleted)
	if err != nil {
		return nil, err
	}
	if err := s.fullResourcePoolLifecycle(ctx, resp, req.ShowDeleted); err != nil {
		return nil, err
	}
	resp = filterResourcePoolLifecycleClusters(resp, req.States, req.ExcludeStates)
	if err := s.fullResourcePoolDisplayNames(ctx, resp, req.ShowDeleted); err != nil {
		return nil, err
	}
	if err := s.fullClustersDualLayerEncryption(ctx, resp); err != nil {
		return nil, err
	}
	if err := s.fullReservedCapacityTemplates(ctx, resp); err != nil {
		return nil, err
	}
	for _, cluster := range resp {
		s.fullConnectionReachability(ctx, cluster)
	}

	return resp, nil
}

func resourcePoolLifecycleQueryStates(states []globalngv1.Cluster_State) []globalngv1.Cluster_State {
	if !lo.SomeBy(states, isResourcePoolLifecycleFilterState) {
		return states
	}

	queryStates := lo.Filter(states, func(state globalngv1.Cluster_State, _ int) bool {
		return !isResourcePoolLifecycleClusterState(state)
	})
	if !lo.Contains(queryStates, globalngv1.Cluster_AVAILABLE) {
		queryStates = append(queryStates, globalngv1.Cluster_AVAILABLE)
	}
	return lo.Uniq(queryStates)
}

func resourcePoolLifecycleQueryExcludeStates(states []globalngv1.Cluster_State) []globalngv1.Cluster_State {
	if !lo.SomeBy(states, isResourcePoolLifecycleFilterState) {
		return states
	}

	// AVAILABLE is also projected from Resource Pool after the database query.
	// Do not exclude it here: paused and active clusters both have AVAILABLE as
	// their durable database state, so that would remove both before the
	// projected-state filter can run.
	return lo.Filter(states, func(state globalngv1.Cluster_State, _ int) bool {
		return !isResourcePoolLifecycleFilterState(state)
	})
}

func filterResourcePoolLifecycleClusters(
	clusters []*entity.Cluster,
	states []globalngv1.Cluster_State,
	excludeStates []globalngv1.Cluster_State,
) []*entity.Cluster {
	if !lo.SomeBy(states, isResourcePoolLifecycleFilterState) &&
		!lo.SomeBy(excludeStates, isResourcePoolLifecycleFilterState) {
		return clusters
	}

	return lo.Filter(clusters, func(cluster *entity.Cluster, _ int) bool {
		if len(states) > 0 && !lo.Contains(states, cluster.GetState()) {
			return false
		}
		return !lo.Contains(excludeStates, cluster.GetState())
	})
}

func isResourcePoolLifecycleClusterState(state globalngv1.Cluster_State) bool {
	switch state {
	case globalngv1.Cluster_PAUSING, globalngv1.Cluster_PAUSED, globalngv1.Cluster_RESUMING:
		return true
	default:
		return false
	}
}

func isResourcePoolLifecycleFilterState(state globalngv1.Cluster_State) bool {
	return state == globalngv1.Cluster_AVAILABLE || isResourcePoolLifecycleClusterState(state)
}

func (s *serviceImpl) fullConnectionReachability(
	ctx context.Context,
	cluster *entity.Cluster,
) {
	for _, endpoint := range cluster.Endpoints {
		switch {
		case endpoint.PublicEndpoint != nil:
			dnsReachable, message := checkDNSPropagation(ctx, endpoint.PublicEndpoint.Host)
			endpoint.PublicEndpoint.Reachability = &globalngv1.ConnectionReachability{
				Reachable: dnsReachable,
				Detail: &globalngv1.ConnectionReachability_Detail{
					ServiceActive:  true,
					EndpointActive: true,
					DnsReachable:   dnsReachable,
					Message:        message,
				},
			}
		case len(endpoint.PrivateLinkEndpoints) > 0:
			for _, privateLinkEndpoint := range endpoint.PrivateLinkEndpoints {
				privateLinkEndpoint.Reachability = &globalngv1.ConnectionReachability{
					Reachable: true,
					Detail: &globalngv1.ConnectionReachability_Detail{
						ServiceActive:  true,
						EndpointActive: true,
						DnsReachable:   true,
						Message:        "",
					},
				}
			}
		case endpoint.VpcPeeringEndpoint != nil:
			endpoint.VpcPeeringEndpoint.Reachability = &globalngv1.ConnectionReachability{
				Reachable: true,
				Detail: &globalngv1.ConnectionReachability_Detail{
					ServiceActive:  true,
					EndpointActive: true,
					DnsReachable:   true,
					Message:        "",
				},
			}
		}
	}
}