package dataplane

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

	"github.com/crossplane/crossplane-runtime/pkg/meta"
	"github.com/pingcap/log"
	ec2v1beta1 "github.com/upbound/provider-aws/apis/ec2/v1beta1"
	eksv1beta1 "github.com/upbound/provider-aws/apis/eks/v1beta1"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/emptypb"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	dmpbv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/utils"
	rsconfig "github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/endpointresolver"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/k8s"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/namer"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/networkcapacity"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/repository"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/util"
)

type DataPlaneService struct {
	dmpbv1.UnimplementedDataPlaneServiceServer

	localCli               client.Client
	dataPlaneRW            repository.DataPlaneReaderWriter
	vpcRW                  repository.VPCReaderWriter
	internetRW             repository.InternetReaderWriter
	appBoxRW               repository.APPBoxReaderWriter
	nodePoolRW             repository.NodePoolReaderWriter
	config                 rsconfig.RootConfig
	tmplConfig             rsconfig.TemplatesConfig
	cloudProviderExtractor CloudProviderExtractor
	endpointZoneResolver   endpointresolver.PrivateLinkEndpointZoneResolver
}

type DataPlaneServiceOption func(*DataPlaneService)

func WithPrivateLinkEndpointZoneResolver(resolver endpointresolver.PrivateLinkEndpointZoneResolver) DataPlaneServiceOption {
	return func(service *DataPlaneService) {
		service.endpointZoneResolver = resolver
	}
}

func NewDataPlaneService(
	cli client.Client,
	dataPlaneRW repository.DataPlaneReaderWriter,
	vpcRW repository.VPCReaderWriter,
	internetRW repository.InternetReaderWriter,
	appBoxRW repository.APPBoxReaderWriter,
	cloudProviderExtractor CloudProviderExtractor,
	config rsconfig.RootConfig,
	tmplConfig rsconfig.TemplatesConfig,
	options ...DataPlaneServiceOption,
) *DataPlaneService {
	service := &DataPlaneService{
		localCli:               cli,
		dataPlaneRW:            dataPlaneRW,
		vpcRW:                  vpcRW,
		internetRW:             internetRW,
		appBoxRW:               appBoxRW,
		nodePoolRW:             repository.NewNodePoolReaderWriter(cli),
		config:                 config,
		tmplConfig:             tmplConfig,
		cloudProviderExtractor: cloudProviderExtractor,
	}
	for _, option := range options {
		option(service)
	}
	return service
}

func (s *DataPlaneService) EvaluateDataPlaneNetworkCapacity(
	ctx context.Context,
	req *dmpbv1.EvaluateDataPlaneNetworkCapacityRequest,
) (*dmpbv1.EvaluateDataPlaneNetworkCapacityResponse, error) {
	if req == nil || req.CidrPrefixLength == nil {
		return nil, status.Error(codes.InvalidArgument, "cidr_prefix_length is required")
	}
	prefixLength := int(req.GetCidrPrefixLength())
	if prefixLength < 0 || prefixLength > 32 {
		return nil, status.Error(codes.InvalidArgument, "cidr_prefix_length must be between 0 and 32")
	}

	var highAvailability networkcapacity.HighAvailabilityType
	switch req.GetHighAvailabilityType() {
	case dmpbv1.DataPlane_REGIONAL:
		highAvailability = networkcapacity.HighAvailabilityRegional
	case dmpbv1.DataPlane_ZONAL:
		highAvailability = networkcapacity.HighAvailabilityZonal
	case dmpbv1.DataPlane_HIGH_AVAILABILITY_TYPE_UNSPECIFIED:
		return nil, status.Error(codes.InvalidArgument, "high_availability_type is required")
	default:
		return nil, status.Errorf(codes.InvalidArgument, "unknown high_availability_type %d", req.GetHighAvailabilityType())
	}
	requestedNodeZone := ""
	switch highAvailability {
	case networkcapacity.HighAvailabilityRegional:
		if len(req.GetNodeZones()) == 1 {
			return nil, status.Error(codes.InvalidArgument, "node_zones with 1 zone means a zonal evaluation; use high_availability_type=ZONAL or pass 2+ zones for REGIONAL")
		}
	case networkcapacity.HighAvailabilityZonal:
		if len(req.GetNodeZones()) > 1 {
			return nil, status.Error(codes.InvalidArgument, "node_zones must contain at most 1 zone for zonal evaluation")
		}
		if len(req.GetNodeZones()) == 1 {
			requestedNodeZone = strings.TrimSpace(req.GetNodeZones()[0])
			if requestedNodeZone == "" {
				return nil, status.Error(codes.InvalidArgument, "node_zones must not contain an empty zone")
			}
		}
	}
	if s.config.Provider != consts.ProviderTypeEKS {
		return nil, status.Errorf(
			codes.FailedPrecondition,
			"network capacity evaluation is unsupported for provider %q",
			s.config.Provider,
		)
	}

	if len(s.config.RegionConfig.Zones) == 0 {
		return nil, status.Error(codes.FailedPrecondition, "regionConfig.zones is not configured")
	}
	if err := s.config.RegionConfig.ValidateNodeZones(); err != nil {
		return nil, status.Errorf(codes.FailedPrecondition, "invalid region topology: %v", err)
	}

	var nodeZones []string
	if requestedNodeZone != "" {
		if !slices.Contains(s.config.RegionConfig.Zones, requestedNodeZone) {
			return nil, status.Errorf(codes.InvalidArgument, "node_zones contains zone %q not configured in regionConfig.zones", requestedNodeZone)
		}
		nodeZones = []string{requestedNodeZone}
	} else if highAvailability == networkcapacity.HighAvailabilityZonal {
		if len(s.config.RegionConfig.NodeZones) != 1 {
			return nil, status.Error(codes.FailedPrecondition, "zonal evaluation requires one node_zones entry or exactly one regionConfig.nodeZones entry")
		}
		nodeZones = s.config.RegionConfig.NodeZones
	}

	var endpointZones []string
	needsEndpointDiscovery := highAvailability == networkcapacity.HighAvailabilityZonal &&
		s.config.ShouldProvisionPrivateLinkEndpointSubnets(nodeZones)
	if needsEndpointDiscovery {
		accountID := strings.TrimSpace(req.GetAccountId())
		if accountID == "" {
			return nil, status.Error(codes.FailedPrecondition, "account_id is required for endpoint-zone discovery")
		}
		roleName := strings.TrimSpace(s.config.DataPlaneRoleName)
		if roleName == "" {
			return nil, status.Error(codes.FailedPrecondition, "dataPlaneRoleName is required for endpoint-zone discovery")
		}
		if s.endpointZoneResolver == nil {
			return nil, status.Error(codes.Internal, "private link endpoint zone resolver is not configured")
		}
		var err error
		endpointZones, err = s.endpointZoneResolver.Resolve(
			ctx,
			util.GetRoleARN(accountID, roleName),
			s.config.AWSDataPlaneExternalID,
			s.config.RegionConfig.Name,
		)
		if err != nil {
			switch {
			case errors.Is(err, context.DeadlineExceeded), errors.Is(ctx.Err(), context.DeadlineExceeded):
				return nil, status.Errorf(codes.DeadlineExceeded, "endpoint-zone discovery timed out: %v", err)
			case errors.Is(err, context.Canceled), errors.Is(ctx.Err(), context.Canceled):
				return nil, status.Errorf(codes.Canceled, "endpoint-zone discovery canceled: %v", err)
			default:
				return nil, status.Errorf(codes.Unavailable, "endpoint-zone discovery failed: %v", err)
			}
		}
		if len(endpointZones) == 0 {
			return nil, status.Error(codes.Unavailable, "endpoint-zone discovery returned no usable zones")
		}
	}

	result, err := networkcapacity.Evaluate(networkcapacity.Input{
		Provider:                 s.config.Provider,
		PrefixLength:             prefixLength,
		HighAvailability:         highAvailability,
		RegionZones:              s.config.RegionConfig.Zones,
		ConfiguredNodeZones:      nodeZones,
		EndpointSubnetsEnabled:   needsEndpointDiscovery,
		PrivateLinkEndpointZones: endpointZones,
		ReservedAdminPodIPs:      s.config.EffectiveReservedAdminPodIPs(),
	})
	if err != nil {
		return nil, status.Errorf(codes.FailedPrecondition, "invalid network capacity deployment context: %v", err)
	}
	if err := networkcapacity.ValidateResult(result); err != nil {
		return nil, status.Errorf(codes.Internal, "invalid network capacity result: %v", err)
	}
	return &dmpbv1.EvaluateDataPlaneNetworkCapacityResponse{
		Supported:          result.Supported,
		SupportedNodeCount: result.SupportedNodeCount,
	}, nil
}

func (s *DataPlaneService) CreateDataPlane(
	ctx context.Context,
	req *dmpbv1.CreateDataPlaneRequest,
) (*dmpbv1.DataPlane, error) {
	log.Info("create dataplane", zap.Any("req", req))
	if req.DataPlaneId == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "data_plane_id is required")
	}
	if req.DataPlane == nil {
		return nil, bizerr.New(bizerr.ParameterInvalid, "data_plane is required")
	}
	requestedOwnershipMode, err := apiDataPlaneOwnershipModeToCR(req.DataPlane.GetOwnershipMode())
	if err != nil {
		return nil, bizerr.New(bizerr.ParameterInvalid, err.Error())
	}
	if req.DataPlane.Vpc == nil || req.DataPlane.Vpc.Cidr == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "cidr is required")
	}
	nodeZones, err := s.resolveNodeZonesForCreate(req.DataPlane.NodeZones, req.DataPlane.GetHighAvailabilityType())
	if err != nil {
		return nil, bizerr.New(bizerr.ParameterInvalid, err.Error())
	}
	if err := s.validatePrivateLinkEndpointSubnetCIDR(req.DataPlane.Vpc.Cidr, nodeZones); err != nil {
		return nil, bizerr.New(bizerr.ParameterInvalid, err.Error())
	}
	if req.DataPlane.Tags[consts.TenantLabelKey] == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "missing required tag: tags.tidbcloud.com/tenant")
	}
	pca := req.DataPlane.GetPca()
	if pca != nil {
		pcaCAIdentifier := pca.GetCaIdentifier()
		if pcaCAIdentifier == "" {
			return nil, bizerr.New(bizerr.ParameterInvalid, "data_plane.pca.ca_identifier is required")
		}
		if _, err := util.ParseAWSPCAARN(pcaCAIdentifier); err != nil {
			return nil, bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("invalid data_plane.pca.ca_identifier: %v", err))
		}
	}
	resourceGroupName := strings.TrimSpace(req.DataPlane.ResourceGroupName)
	if resourceGroupName != "" && s.config.Provider != consts.ProviderTypeAKS {
		return nil, bizerr.New(
			bizerr.ParameterInvalid,
			"resource_group_name is only supported for AKS dataplanes",
		)
	}

	configuredZones := append([]string(nil), s.config.RegionConfig.Zones...)

	labels := map[string]string{
		v1alpha1.RegionKey: s.config.RegionConfig.Name,
		v1alpha1.ManageKey: "dataplane-manager",
	}

	labels = util.MergeMaps(labels, req.DataPlane.Tags)
	dataPlane := &v1alpha1.DataPlane{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.DataPlaneId,
			Namespace: s.getNamespace(),
			Labels:    labels,
		},
		Spec: v1alpha1.DataPlaneSpec{
			CloudTags:     req.DataPlane.Tags,
			VpcCidr:       req.DataPlane.Vpc.Cidr,
			OwnershipMode: requestedOwnershipMode,
			// AvailabilityZones is the region-level AZ baseline for control-plane/network resources,
			// sourced from regionConfig.zones.
			AvailabilityZones: configuredZones,
			// NodeZones is the workload AZ restriction (optional):
			// - request data_plane.node_zones when provided;
			// - otherwise regionConfig.nodeZones when configured;
			// - left empty when no restriction is intended.
			NodeZones: nodeZones,
			AccountRef: v1alpha1.AccountReference{
				ProviderConfigName: req.DataPlane.AccountId,
				AccountID:          req.DataPlane.AccountId,
				ResourceGroupName:  resourceGroupName,
				Role:               s.config.DataPlaneRoleName,
			},
		},
	}
	if req.GetEnableCapacityPlanner() {
		dataPlane.Annotations = map[string]string{
			v1alpha1.AnnotationKeyEnableCapacityPlanner: "true",
		}
	}
	if pca != nil {
		dataPlane.Spec.PCA = &v1alpha1.PCAConfig{
			CAIdentifier: pca.GetCaIdentifier(),
		}
	}
	if k := req.DataPlane.Kubernetes; k != nil {
		dataPlane.Spec.ServiceCidr = k.ServiceCidr
	}
	if err := s.dataPlaneRW.Create(ctx, dataPlane); err != nil {
		if apierrors.IsAlreadyExists(err) {
			log.Info("dataplane already exists",
				zap.String("name", dataPlane.Name), zap.String("namespace", dataPlane.Namespace))
			existing := new(v1alpha1.DataPlane)
			if getErr := s.dataPlaneRW.Get(ctx, client.ObjectKeyFromObject(dataPlane), existing); getErr != nil {
				return nil, bizerr.Wrap(bizerr.K8SClientError, "get existing dataplane error", getErr)
			}
			existingOwnershipMode, modeErr := v1alpha1.EffectiveDataPlaneOwnershipMode(existing.Spec.OwnershipMode)
			if modeErr != nil || existingOwnershipMode != requestedOwnershipMode {
				return nil, bizerr.New(bizerr.ResourceConflict, "dataplane ownership mode conflicts with existing dataplane")
			}
			return nil, bizerr.New(bizerr.ResourceConflict, "dataplane already exists")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "create dataplane error", err)
	}

	response, err := dataPlaneCR2API(dataPlane)
	if err != nil {
		return nil, bizerr.New(bizerr.InternalServerError, err.Error())
	}
	return response, nil
}

func (s *DataPlaneService) validatePrivateLinkEndpointSubnetCIDR(cidr string, nodeZones []string) error {
	if s.config.Provider != consts.ProviderTypeACK ||
		!s.config.ShouldProvisionPrivateLinkEndpointSubnets(nodeZones) {
		return nil
	}

	ip, network, err := net.ParseCIDR(cidr)
	if err == nil {
		_, bits := network.Mask.Size()
		if ip.To4() != nil && bits == net.IPv4len*8 {
			return nil
		}
	}
	return errors.New("ACK private link endpoint subnets require an IPv4 VPC CIDR")
}

func (s *DataPlaneService) GetDataPlane(
	ctx context.Context,
	req *dmpbv1.GetDataPlaneRequest,
) (*dmpbv1.DataPlane, error) {
	dataPlaneCR := new(v1alpha1.DataPlane)
	dataPlaneKey := client.ObjectKey{
		Namespace: s.getNamespace(),
		Name:      req.DataPlaneId,
	}
	if err := s.dataPlaneRW.Get(ctx, dataPlaneKey, dataPlaneCR); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "dataplane not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get dataplane error", err)
	}

	var (
		vpcCR      *v1alpha1.VPC
		internetCR *v1alpha1.Internet
		appBoxCR   *v1alpha1.APPBox
	)
	if vpcRef := dataPlaneCR.Status.Vpc; vpcRef != nil && vpcRef.Name != "" {
		vpcCR = new(v1alpha1.VPC)
		vpcKey := client.ObjectKey{
			Namespace: vpcRef.Namespace,
			Name:      vpcRef.Name,
		}
		if err := s.vpcRW.Get(ctx, vpcKey, vpcCR); err != nil {
			return nil, bizerr.Wrap(bizerr.K8SClientError, "get vpc error", err)
		}
	}
	if internetRef := dataPlaneCR.Status.Internet; internetRef != nil && internetRef.Name != "" {
		internetCR = new(v1alpha1.Internet)
		internetKey := client.ObjectKey{
			Namespace: internetRef.Namespace,
			Name:      internetRef.Name,
		}
		if err := s.internetRW.Get(ctx, internetKey, internetCR); err != nil {
			return nil, bizerr.Wrap(bizerr.K8SClientError, "get internet error", err)
		}
	}
	if appBoxRef := dataPlaneCR.Status.AppBox; appBoxRef != nil && appBoxRef.Name != "" {
		appBoxCR = new(v1alpha1.APPBox)
		appBoxKey := client.ObjectKey{
			Namespace: appBoxRef.Namespace,
			Name:      appBoxRef.Name,
		}
		if err := s.appBoxRW.Get(ctx, appBoxKey, appBoxCR); err != nil {
			return nil, bizerr.Wrap(bizerr.K8SClientError, "get appbox error", err)
		}
	}

	return s.convertDataPlane(ctx, dataPlaneCR, vpcCR, internetCR, appBoxCR)
}

func (s *DataPlaneService) convertDataPlane(
	ctx context.Context, dataPlaneCR *v1alpha1.DataPlane,
	vpcCR *v1alpha1.VPC, internetCR *v1alpha1.Internet, appBoxCR *v1alpha1.APPBox,
) (*dmpbv1.DataPlane, error) {
	dataPlane, err := dataPlaneCR2API(dataPlaneCR)
	if err != nil {
		return nil, bizerr.New(bizerr.InternalServerError, err.Error())
	}
	dataPlane.Domain = s.cloudProviderExtractor.GetDataPlaneDomain()
	dataPlane.Capabilities = dataPlaneCapabilities()
	if vpcCR != nil {
		s.fillVPC2DataPlane(vpcCR, dataPlane)
	}
	if internetCR != nil {
		s.fillInternet2DataPlane(internetCR, dataPlane)
	}
	if appBoxCR != nil {
		if err := s.fillAppBox2DataPlane(ctx, appBoxCR, dataPlane); err != nil {
			return nil, err
		}
	}
	return dataPlane, nil
}

// dataPlaneCapabilities lists runtime feature flags advertised by this build.
// Clients probe this set through GetDataPlane before sending requests that
// depend on a conditional feature.
func dataPlaneCapabilities() []string {
	return []string{"loadbalancer.multi-port"}
}

func (s *DataPlaneService) fillVPC2DataPlane(vpcCR *v1alpha1.VPC, dataPlane *dmpbv1.DataPlane) {
	if vpcCR.Status.Phase != v1alpha1.VPCPhaseActive {
		log.Info("vpc is not active",
			zap.String("name", vpcCR.Name),
			zap.String("namespace", vpcCR.Namespace),
			zap.String("phase", string(vpcCR.Status.Phase)))
		return
	}

	zoneToAZID := make(map[string]string, len(vpcCR.Status.NodeSubnets))
	for _, subnet := range vpcCR.Status.NodeSubnets {
		zoneToAZID[subnet.AvailabilityZone] = subnet.AvailabilityZoneID
	}

	nodeZones := s.resolveNodeZonesFromVPC(vpcCR)
	azs, azIDs := make([]string, 0, len(nodeZones)), make([]string, 0, len(nodeZones))
	for _, zone := range nodeZones {
		azID, ok := zoneToAZID[zone]
		if !ok {
			continue
		}
		azs = append(azs, zone)
		azIDs = append(azIDs, azID)
	}
	dataPlane.Vpc.AvailabilityZones = azs
	dataPlane.Vpc.AvailabilityZoneIds = azIDs

	for _, subnet := range vpcCR.Status.PublicSubnets {
		if subnet.CIDRBlock != "" {
			dataPlane.Vpc.PublicSubnetCidrs = append(dataPlane.Vpc.PublicSubnetCidrs, subnet.CIDRBlock)
		}
	}
}

func (s *DataPlaneService) resolveNodeZonesForCreate(
	reqNodeZones []string,
	haType dmpbv1.DataPlane_HighAvailabilityType,
) ([]string, error) {
	// Pre-validate caller input shape before any auto-fill or strict validation
	// kicks in. The check below fails fast with a message that points directly
	// at the misconfigured caller side instead of bubbling up as a confusing
	// regionConfig error.
	//
	// ZONAL + empty req_node_zones is intentionally NOT pre-validated: it falls
	// through to the existing regionConfig.nodeZones auto-fill below so callers
	// can rely on the template-provided default zone. UNSPECIFIED HA is also
	// intentionally not pre-validated so the existing defaults path keeps
	// working.
	switch haType {
	case dmpbv1.DataPlane_REGIONAL:
		if len(reqNodeZones) == 1 {
			return nil, fmt.Errorf("data_plane.node_zones with 1 zone means a zonal deployment; use high_availability_type=ZONAL or pass 2+ zones for REGIONAL")
		}
	}

	if len(reqNodeZones) == 0 {
		switch haType {
		case dmpbv1.DataPlane_ZONAL:
			if len(s.config.RegionConfig.NodeZones) != 1 {
				return nil, fmt.Errorf("regionConfig.nodeZones must contain exactly 1 zone for zonal dataplane")
			}
			return append([]string(nil), s.config.RegionConfig.NodeZones...), nil
		case dmpbv1.DataPlane_REGIONAL:
			// Bring-up layer sources the multi-AZ baseline from regionConfig.zones.
			return nil, nil
		case dmpbv1.DataPlane_HIGH_AVAILABILITY_TYPE_UNSPECIFIED:
		default:
			return nil, fmt.Errorf("invalid data_plane.high_availability_type %s", haType.String())
		}
		if len(s.config.RegionConfig.NodeZones) == 0 {
			return nil, nil
		}
		return append([]string(nil), s.config.RegionConfig.EffectiveNodeZones()...), nil
	}

	nodeZones := make([]string, 0, len(reqNodeZones))
	seen := make(map[string]struct{}, len(reqNodeZones))
	for _, zone := range reqNodeZones {
		if _, ok := seen[zone]; ok {
			return nil, fmt.Errorf("data_plane.node_zones contains duplicate zone %q", zone)
		}
		seen[zone] = struct{}{}
		nodeZones = append(nodeZones, zone)
	}

	allZones := append([]string(nil), s.config.RegionConfig.Zones...)
	allowedZones := make(map[string]struct{}, len(allZones))
	for _, zone := range allZones {
		allowedZones[zone] = struct{}{}
	}
	for _, zone := range nodeZones {
		if _, ok := allowedZones[zone]; !ok {
			return nil, fmt.Errorf("data_plane.node_zones contains zone %q not in configured regionConfig.zones", zone)
		}
	}
	switch haType {
	case dmpbv1.DataPlane_ZONAL:
		if len(nodeZones) != 1 {
			return nil, fmt.Errorf("data_plane.node_zones must contain exactly 1 zone for zonal dataplane")
		}
	case dmpbv1.DataPlane_REGIONAL:
		// Pre-validated upstream at the top of this function:
		// REGIONAL callers are expected to pass 2+ zones (explicit multi-AZ layout).
		// 1 zone is rejected because it semantically means zonal — see
		// dataPlaneCR2API which infers HA type from len(nodeZones). This empty
		// case arm is load-bearing: without it, REGIONAL would fall into the
		// `default:` arm and emit "invalid data_plane.high_availability_type
		// REGIONAL".
	case dmpbv1.DataPlane_HIGH_AVAILABILITY_TYPE_UNSPECIFIED:
		if len(nodeZones) >= len(allZones) {
			return nil, fmt.Errorf("data_plane.node_zones must be a strict subset of configured regionConfig.zones")
		}
	default:
		return nil, fmt.Errorf("invalid data_plane.high_availability_type %s", haType.String())
	}

	return nodeZones, nil
}

func (s *DataPlaneService) resolveNodeZonesFromVPC(vpcCR *v1alpha1.VPC) []string {
	if vpcCR == nil {
		return nil
	}
	if len(vpcCR.Spec.NodeZones) > 0 {
		return append([]string(nil), vpcCR.Spec.NodeZones...)
	}
	return append([]string(nil), vpcCR.Spec.Zones...)
}

func (s *DataPlaneService) fillInternet2DataPlane(internetCR *v1alpha1.Internet, dataPlane *dmpbv1.DataPlane) {
	if internetCR.Status.Phase != v1alpha1.InternetPhaseActive {
		log.Info("internet is not active",
			zap.String("name", internetCR.Name),
			zap.String("namespace", internetCR.Namespace),
			zap.String("phase", string(internetCR.Status.Phase)))
		return
	}

	dataPlane.Vpc.NatIps = internetCR.Status.ExternalIPs
}

func (s *DataPlaneService) fillAppBox2DataPlane(
	ctx context.Context, appBoxCR *v1alpha1.APPBox, dataPlane *dmpbv1.DataPlane,
) error {
	dataPlane.Kubernetes.Version = appBoxCR.Spec.KubernetesConfig.Version
	dataPlane.Kubernetes.Name = appBoxCR.Status.KubernetesClusterName
	kubeconfig, err := s.cloudProviderExtractor.BuildKubeconfig(ctx, dataPlane.AccountId, appBoxCR)
	if err != nil {
		if errors.Is(err, errK8sNotActive) {
			log.Info("the Kubernetes cluster of appbox is not active",
				zap.String("appbox_name", appBoxCR.Name),
				zap.String("namespace", appBoxCR.Namespace),
				zap.String("appbox_phase", string(appBoxCR.Status.Phase)))
			return nil
		}
		return err
	}
	log.Debug("kubeconfig generated", zap.String("kubeconfig", string(kubeconfig)))
	dataPlane.Kubernetes.Kubeconfig = string(kubeconfig)

	if appBoxCR.Status.Phase != v1alpha1.APPBoxPhaseActive {
		log.Info("appbox is not active",
			zap.String("name", appBoxCR.Name),
			zap.String("namespace", appBoxCR.Namespace),
			zap.String("phase", string(appBoxCR.Status.Phase)))
		return nil
	}

	dataPlane.NodeRole = s.cloudProviderExtractor.ParseNodeRole(appBoxCR)
	dataPlane.NodeClientId = s.cloudProviderExtractor.ParseNodeClientID(appBoxCR)
	if err := s.fillAuthTokenApiGateway2DataPlane(ctx, appBoxCR, dataPlane); err != nil {
		return err
	}

	return nil
}

func (s *DataPlaneService) fillAuthTokenApiGateway2DataPlane(
	ctx context.Context, appBox *v1alpha1.APPBox, dataPlane *dmpbv1.DataPlane,
) error {
	apiGateway, err := s.cloudProviderExtractor.GetDataPlaneApiGateway(ctx, appBox)
	if err != nil {
		return err
	}
	dataPlane.ApiGateway = apiGateway

	authToken, err := s.cloudProviderExtractor.GetDataPlaneAuthToken(dataPlane.AccountId, appBox)
	if err != nil {
		return err
	}
	dataPlane.AuthToken = authToken
	return nil
}

func (s *DataPlaneService) ListDataPlanes(
	ctx context.Context,
	req *dmpbv1.ListDataPlanesRequest,
) (*dmpbv1.ListDataPlanesResponse, error) {
	inNamespace := client.InNamespace(s.getNamespace())
	dataPlaneListCR := new(v1alpha1.DataPlaneList)
	if err := s.dataPlaneRW.List(ctx, dataPlaneListCR, inNamespace); err != nil {
		return nil, bizerr.Wrap(bizerr.K8SClientError, "list dataplanes error", err)
	}
	vpcListCR := new(v1alpha1.VPCList)
	if err := s.vpcRW.List(ctx, vpcListCR, inNamespace); err != nil {
		return nil, bizerr.Wrap(bizerr.K8SClientError, "list vpc error", err)
	}
	internetListCR := new(v1alpha1.InternetList)
	if err := s.internetRW.List(ctx, internetListCR, inNamespace); err != nil {
		return nil, bizerr.Wrap(bizerr.K8SClientError, "list internets error", err)
	}
	appBoxListCR := new(v1alpha1.APPBoxList)
	if err := s.appBoxRW.List(ctx, appBoxListCR, inNamespace); err != nil {
		return nil, bizerr.Wrap(bizerr.K8SClientError, "list appboxes error", err)
	}

	vpcMap := make(map[string]*v1alpha1.VPC, len(vpcListCR.Items))
	for i := range vpcListCR.Items {
		vpcCR := &vpcListCR.Items[i]
		vpcMap[vpcCR.Name] = vpcCR
	}
	internetMap := make(map[string]*v1alpha1.Internet, len(internetListCR.Items))
	for i := range internetListCR.Items {
		internetCR := &internetListCR.Items[i]
		internetMap[internetCR.Name] = internetCR
	}
	appBoxMap := make(map[string]*v1alpha1.APPBox, len(appBoxListCR.Items))
	for i := range appBoxListCR.Items {
		appBoxCR := &appBoxListCR.Items[i]
		appBoxMap[appBoxCR.Name] = appBoxCR
	}

	dataPlanes := make([]*dmpbv1.DataPlane, 0, len(dataPlaneListCR.Items))
	for i := range dataPlaneListCR.Items {
		dataPlaneCR := &dataPlaneListCR.Items[i]
		var (
			vpcCR      *v1alpha1.VPC
			internetCR *v1alpha1.Internet
			appBoxCR   *v1alpha1.APPBox
		)
		if vpcRef := dataPlaneCR.Status.Vpc; vpcRef != nil && vpcRef.Name != "" {
			vpcCR = vpcMap[vpcRef.Name]
		}
		if internetRef := dataPlaneCR.Status.Internet; internetRef != nil && internetRef.Name != "" {
			internetCR = internetMap[internetRef.Name]
		}
		if appBoxRef := dataPlaneCR.Status.AppBox; appBoxRef != nil && appBoxRef.Name != "" {
			appBoxCR = appBoxMap[appBoxRef.Name]
		}
		dataPlane, err := s.convertDataPlane(ctx, dataPlaneCR, vpcCR, internetCR, appBoxCR)
		if err != nil {
			return nil, err
		}
		dataPlanes = append(dataPlanes, dataPlane)
	}
	return &dmpbv1.ListDataPlanesResponse{
		DataPlanes: dataPlanes,
	}, nil
}

func (s *DataPlaneService) UpdateDataPlaneCloudTags(
	ctx context.Context,
	req *dmpbv1.UpdateDataPlaneCloudTagsRequest,
) (*dmpbv1.DataPlane, error) {
	if req.DataPlaneId == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "data_plane_id is required")
	}
	if req.Tags == nil {
		return nil, bizerr.New(bizerr.ParameterInvalid, "tags is required")
	}
	for key, value := range req.Tags {
		if strings.TrimSpace(key) == "" {
			return nil, bizerr.New(bizerr.ParameterInvalid, "tags contains empty key")
		}
		if strings.TrimSpace(value) == "" {
			return nil, bizerr.New(bizerr.ParameterInvalid, "tags contains empty value")
		}
	}
	if req.Tags[consts.TenantLabelKey] == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "missing required tag: tags.tidbcloud.com/tenant")
	}
	dataPlaneCR := &v1alpha1.DataPlane{}
	if err := s.dataPlaneRW.Get(ctx, client.ObjectKey{
		Namespace: s.getNamespace(),
		Name:      req.DataPlaneId,
	}, dataPlaneCR); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "dataplane not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get dataplane error", err)
	}
	if _, err := v1alpha1.EffectiveDataPlaneOwnershipMode(dataPlaneCR.Spec.OwnershipMode); err != nil {
		return nil, bizerr.New(bizerr.InternalServerError, err.Error())
	}
	dataPlaneCR.Spec.CloudTags = util.DeepCopyMap(req.Tags)
	dataPlaneCR.Labels = util.MergeMaps(dataPlaneCR.Labels, req.Tags)
	if err := s.dataPlaneRW.Update(ctx, dataPlaneCR); err != nil {
		return nil, bizerr.Wrap(bizerr.K8SClientError, "update dataplane cloud tags error", err)
	}
	return s.convertDataPlane(ctx, dataPlaneCR, nil, nil, nil)
}

func (s *DataPlaneService) DeleteDataPlane(
	ctx context.Context, req *dmpbv1.DeleteDataPlaneRequest,
) (*emptypb.Empty, error) {
	nsName := s.getNamespace()
	dataPlane := &v1alpha1.DataPlane{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: nsName,
			Name:      req.DataPlaneId,
		},
	}
	if err := s.dataPlaneRW.Delete(ctx, dataPlane); err != nil {
		if apierrors.IsNotFound(err) {
			log.Info("dataplane is not found",
				zap.String("namespace", nsName),
				zap.String("name", req.DataPlaneId))
			return new(emptypb.Empty), nil
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "delete dataplane error", err)
	}
	return new(emptypb.Empty), nil
}

func (s *DataPlaneService) ListNodePools(
	ctx context.Context,
	req *dmpbv1.ListNodePoolsRequest,
) (*dmpbv1.ListNodePoolsResponse, error) {
	dataPlane := new(v1alpha1.DataPlane)
	dataPlaneKey := client.ObjectKey{
		Namespace: s.getNamespace(),
		Name:      req.DataPlaneId,
	}
	if err := s.dataPlaneRW.Get(ctx, dataPlaneKey, dataPlane); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "dataplane not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get dataplane error", err)
	}
	if dataPlane.Status.Phase != v1alpha1.DataPlanePhaseActive {
		return &dmpbv1.ListNodePoolsResponse{
			NodePools: []*dmpbv1.NodePool{},
		}, nil
	}

	appBox := new(v1alpha1.APPBox)
	appBoxKey := client.ObjectKey{
		Namespace: dataPlane.Status.AppBox.Namespace,
		Name:      dataPlane.Status.AppBox.Name,
	}
	if err := s.appBoxRW.Get(ctx, appBoxKey, appBox); err != nil {
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get appbox error", err)
	}
	nodePoolCRs, err := s.listNodePoolCRs(ctx, req.DataPlaneId)
	if err != nil {
		return nil, bizerr.Wrap(bizerr.K8SClientError, "list nodepools error", err)
	}
	nodePools := make([]*dmpbv1.NodePool, 0, len(nodePoolCRs))
	for i := range nodePoolCRs {
		nodePools = append(nodePools, nodePoolCR2API(nodePoolCRs[i]))
	}
	return &dmpbv1.ListNodePoolsResponse{
		NodePools: nodePools,
	}, nil
}

func (s *DataPlaneService) CreateNodePool(
	ctx context.Context,
	req *dmpbv1.CreateNodePoolRequest,
) (*dmpbv1.NodePool, error) {
	log.Info("create nodepool", zap.Any("req", req))
	if req.DataPlaneId == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "data_plane_id is required")
	}
	if req.NodePool == nil {
		return nil, bizerr.New(bizerr.ParameterInvalid, "node_pool is required")
	}
	if req.NodePool.InstanceType == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "instance_type are required")
	}
	if req.NodePool.CpuCore == 0 || req.NodePool.MemoryGi == 0 {
		return nil, bizerr.New(bizerr.ParameterInvalid, "cpu or memory are required")
	}

	dataPlane := new(v1alpha1.DataPlane)
	dataPlaneKey := client.ObjectKey{
		Namespace: s.getNamespace(),
		Name:      req.DataPlaneId,
	}
	if err := s.dataPlaneRW.Get(ctx, dataPlaneKey, dataPlane); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "dataplane not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get dataplane error", err)
	}
	if dataPlane.Status.Phase != v1alpha1.DataPlanePhaseActive {
		return nil, bizerr.New(bizerr.ParameterInvalid, "dataplane is not active")
	}

	if dataPlane.Status.Vpc == nil {
		return nil, bizerr.New(bizerr.ParameterInvalid, "dataplane does not have vpc reference")
	}

	// Honour placement_group_enabled from the gRPC request: it carries the
	// CSNG-resolved durable value (per-dataplane SharedPool gate, or
	// inherited from a sibling NodePool when one exists). The value is
	// persisted on NodePool Spec as the durable record that the EKS
	// controllers (Create / PostCreate / Delete / PostDelete / sibling
	// counts / NodeGroup partition branch) read on every reconcile.
	//
	// Topology sanity check: when the caller asks for `true`, the dataplane
	// must be a single-AZ EKS dataplane — that is the only topology that
	// supports partition placement groups. Anything else (multi-AZ, or any
	// non-EKS provider) is rejected so a stale or buggy caller cannot pin
	// an unsupported feature flag onto the spec. `false` is always accepted:
	// the legacy per-zone path is safe on every topology, and `false` is
	// also the explicit durable record for a legacy single-AZ, PG-disabled
	// pool whose CSNG sibling chain must stay on the legacy path even
	// though the topology itself is eligible.
	//
	// Built-in NodePools are not created via this gRPC path (they are
	// provisioned by pkg/controllers/dataplane/task/nodepool.go), so the
	// built-in annotation check is unnecessary here — built-ins reach the
	// EKS controller with spec=false by default and the controller's
	// annotation gate keeps them outside the PG scope.
	topologyEligible, err := s.resolvePlacementGroupEnabled(ctx, dataPlane)
	if err != nil {
		return nil, bizerr.Wrap(bizerr.K8SClientError, "resolve placement group topology eligibility error", err)
	}
	wirePlacementGroupEnabled := req.NodePool.GetPlacementGroupEnabled()
	if wirePlacementGroupEnabled && !topologyEligible {
		return nil, bizerr.New(bizerr.ParameterInvalid,
			fmt.Sprintf("placement_group_enabled=true requested but dataplane %s topology is not single-AZ EKS", dataPlane.Name))
	}
	placementGroupEnabled := wirePlacementGroupEnabled

	if dataPlane.Status.AppBox == nil {
		return nil, bizerr.New(bizerr.ParameterInvalid, "dataplane does not have appbox reference")
	}

	version := req.NodePool.K8SVersion
	if version == "" {
		appBoxRef := dataPlane.Status.AppBox
		appBoxCR := new(v1alpha1.APPBox)
		appBoxKey := client.ObjectKey{
			Namespace: appBoxRef.Namespace,
			Name:      appBoxRef.Name,
		}

		if err := s.appBoxRW.Get(ctx, appBoxKey, appBoxCR); err != nil {
			return nil, bizerr.Wrap(bizerr.K8SClientError, "get appbox error", err)
		}
		version = appBoxCR.Spec.KubernetesConfig.Version
	}

	// ACK upgrades node pools in-place, so k8s version should not be part of the name
	// (it would become stale after an upgrade). EKS creates a new node group per version.
	nameK8sVersion := version
	if s.config.Provider == consts.ProviderTypeACK {
		nameK8sVersion = ""
	}
	name := buildNodePoolName(
		effectiveResourceNamePrefix(req.GetResourcePrefix(), s.config.ResourceNamePrefix),
		req.DataPlaneId,
		req.NodePool.InstanceType,
		nameK8sVersion,
		req.NodePool.CpuCore,
		req.NodePool.MemoryGi,
	)
	labels := map[string]string{
		v1alpha1.RegionKey:      s.config.RegionConfig.Name,
		v1alpha1.VPCNameKey:     dataPlane.Status.Vpc.Name,
		v1alpha1.ManageKey:      v1alpha1.ManageFull,
		v1alpha1.DataPlaneIDKey: dataPlane.Name,
		v1alpha1.APPBoxNameKey:  dataPlane.Status.AppBox.Name,
	}
	labels = util.MergeMaps(
		s.config.ResourceLabels, dataPlane.Spec.CloudTags, labels)
	nodePoolCR := &v1alpha1.NodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: s.getNamespace(),
			Labels:    labels,
		},
		Spec: v1alpha1.NodePoolSpec{
			ResourceNamePrefix: req.GetResourcePrefix(),
			DataPlane: v1alpha1.Reference{
				Namespace: s.getNamespace(),
				Name:      req.DataPlaneId,
			},
			InstanceType:          req.NodePool.InstanceType,
			KubernetesVersion:     version,
			CpuCore:               req.NodePool.CpuCore,
			AvailableCpuMilliCore: req.NodePool.AvailableCpuMilliCore,
			MaxCpuMilli:           nodePoolAPIMaxCPUMilli(req.NodePool),
			MemoryGi:              req.NodePool.MemoryGi,
			AvailableMemoryMi:     req.NodePool.AvailableMemoryMi,
			MaxMemoryMi:           nodePoolAPIMaxMemoryMi(req.NodePool),
			WithLocalDisk:         req.NodePool.WithLocalDisk,
			LocalDiskSizeGi:       req.NodePool.LocalDiskSizeGi,
			Labels:                req.NodePool.Labels,
			Taints:                nodePoolTaintsAPI2CR(req.NodePool.Taints),
			PlacementGroupEnabled: placementGroupEnabled,
		},
	}
	if req.NodePool.NodeType != dmpbv1.NodeType_NodeType_Unspecified {
		nodePoolCR.Spec.NodeType = req.NodePool.NodeType.String()
	}
	if err := s.nodePoolRW.Create(ctx, nodePoolCR); err != nil {
		if apierrors.IsAlreadyExists(err) {
			return nil, bizerr.New(bizerr.ResourceConflict, "nodepool already exists")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "create nodepool error", err)
	}

	return nodePoolCR2API(nodePoolCR), nil
}

// resolvePlacementGroupEnabled reports whether the dataplane's topology
// supports the partition placement-group feature. It is the topology
// sanity check used by CreateNodePool when the caller requests
// placement_group_enabled=true: the request is only honoured when this
// helper returns true.
//
// Topology rule: single-AZ EKS = eligible; multi-AZ, or any non-EKS
// provider (ACK / AKS), = ineligible.
//
// Topology source: DataPlane.Spec.NodeZones when populated (the
// workload-declared topology, which the user explicitly committed to);
// otherwise VPC.Spec.NodeZones / VPC.Spec.Zones for dataplane CRs where
// Spec.NodeZones was not set.
//
// Unresolved topology (no NodeZones on either DataPlane or VPC) returns
// an error rather than false-with-no-error: false-with-no-error would
// silently force every NodePool into the legacy per-zone path on a
// dataplane that may eventually resolve to single-AZ, masking the
// divergence from CSNG's sibling-inheritance chain.
func (s *DataPlaneService) resolvePlacementGroupEnabled(ctx context.Context, dataPlane *v1alpha1.DataPlane) (bool, error) {
	if dataPlane == nil {
		return false, fmt.Errorf("dataplane is required to resolve placement group eligibility")
	}
	if s.config.Provider != consts.ProviderTypeEKS {
		return false, nil
	}
	if dataPlane.Status.Vpc == nil {
		return false, fmt.Errorf("dataplane %s has no vpc reference; cannot resolve placement group eligibility", dataPlane.Name)
	}
	if len(dataPlane.Spec.NodeZones) > 0 {
		return len(dataPlane.Spec.NodeZones) == 1, nil
	}
	if s.vpcRW == nil {
		return false, fmt.Errorf("vpc reader is not configured; cannot resolve placement group eligibility for dataplane %s", dataPlane.Name)
	}
	vpc := new(v1alpha1.VPC)
	if err := s.vpcRW.Get(ctx, client.ObjectKey{
		Namespace: dataPlane.Status.Vpc.Namespace,
		Name:      dataPlane.Status.Vpc.Name,
	}, vpc); err != nil {
		if apierrors.IsNotFound(err) {
			return false, fmt.Errorf("vpc %s/%s not found yet; cannot resolve placement group eligibility",
				dataPlane.Status.Vpc.Namespace, dataPlane.Status.Vpc.Name)
		}
		return false, err
	}
	if len(vpc.Spec.NodeZones) > 0 {
		return len(vpc.Spec.NodeZones) == 1, nil
	}
	if len(vpc.Spec.Zones) == 0 {
		return false, fmt.Errorf("vpc %s/%s has no zones; cannot resolve placement group eligibility",
			vpc.Namespace, vpc.Name)
	}
	return len(vpc.Spec.Zones) == 1, nil
}

func (s *DataPlaneService) LegacyDeleteNodePool(
	ctx context.Context,
	req *dmpbv1.LegacyDeleteNodePoolRequest,
) (*emptypb.Empty, error) {
	// for specified instance_type
	dataPlaneId := req.GetDataPlaneId()
	version := req.GetK8SVersion()
	if dataPlaneId == "" || version == "" {
		log.Error("data_plane_id and k8s_version are required for DeleteNodeGroups")
		return nil, bizerr.New(bizerr.ParameterInvalid, "data_plane_id and k8s_version are required")
	}
	instanceType := req.GetInstanceType()

	var dataPlane v1alpha1.DataPlane
	if err := s.dataPlaneRW.Get(ctx, client.ObjectKey{
		Namespace: s.config.Namespace,
		Name:      dataPlaneId,
	}, &dataPlane); err != nil {
		log.Error("failed to get dataplane", zap.String("data_plane_id", dataPlaneId), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to get dataplane", err)
	}
	if dataPlane.Status.AppBox == nil {
		log.Error("dataplane does not have appbox reference", zap.String("data_plane_id", dataPlaneId))
		return nil, bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("dataplane %s does not have appbox reference", dataPlaneId))
	}

	var appBox v1alpha1.APPBox
	if err := s.appBoxRW.Get(ctx, client.ObjectKey{
		Namespace: dataPlane.Status.AppBox.Namespace,
		Name:      dataPlane.Status.AppBox.Name,
	}, &appBox); err != nil {
		log.Error("failed to get appbox", zap.String("name", dataPlane.Status.AppBox.Name), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to get appbox", err)
	}

	// Delete node groups/pools with the specified version based on provider type
	switch appBox.Spec.Provider {
	case consts.ProviderTypeEKS:
		var ngList eksv1beta1.NodeGroupList
		if err := s.localCli.List(ctx, &ngList, utils.GetNodePoolsLabelSelector(dataPlaneId)); err != nil {
			log.Error("failed to list eks nodegroups", zap.String("data_plane_id", dataPlaneId), zap.Error(err))
			return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to list eks nodegroups", err)
		}

		// Collect worker node groups with the specified version and check if they have NO nodes.
		// Skip APPBox platform pools (admin/shared, o11y): they are upgraded in-place and must
		// not be deleted by the blue/green worker cleanup path.
		var ngsToDelete []*eksv1beta1.NodeGroup
		for i := range ngList.Items {
			ng := &ngList.Items[i]
			if comp := ng.Spec.ForProvider.Labels[consts.LabelKeyComponent]; comp != nil && consts.IsPlatformManagedNodeComponent(*comp) {
				continue
			}
			if ptr.Deref(ng.Spec.ForProvider.Version, "") == version {
				// ensure the version in the spec and in the status are the same
				// to avoid deleting node groups that are still upgrading
				if ptr.Deref(ng.Status.AtProvider.Version, "") != version {
					log.Error("node group version in status does not match the specified version, cannot delete",
						zap.String("data_plane_id", dataPlaneId),
						zap.String("nodegroup", ng.Name),
						zap.String("version", version),
						zap.String("status_version", ptr.Deref(ng.Status.AtProvider.Version, "")))
					return nil, bizerr.New(bizerr.ParameterInvalid,
						fmt.Sprintf("node group %s version in status (%s) does not match the specified version (%s), cannot delete",
							ng.Name, ptr.Deref(ng.Status.AtProvider.Version, ""), version))
				}

				if len(ng.Status.AtProvider.ScalingConfig) > 0 && ng.Status.AtProvider.ScalingConfig[0].DesiredSize != nil &&
					*ng.Status.AtProvider.ScalingConfig[0].DesiredSize > 0 {
					log.Error("node group still has nodes, cannot delete",
						zap.String("data_plane_id", dataPlaneId),
						zap.String("nodegroup", ng.Name),
						zap.String("version", version),
						zap.Float64("desired_size", *ng.Status.AtProvider.ScalingConfig[0].DesiredSize))
					return nil, bizerr.New(bizerr.ParameterInvalid,
						fmt.Sprintf("node group %s still has nodes (desired size: %d), cannot delete",
							ng.Name, int64(*ng.Status.AtProvider.ScalingConfig[0].DesiredSize)))
				}

				// if instanceType is specified, filter by instance type
				if instanceType != "" {
					found := false
					for _, it := range ng.Spec.ForProvider.InstanceTypes {
						if ptr.Deref(it, "") == instanceType {
							found = true
							break
						}
					}
					if !found {
						continue
					}
				}
				ngsToDelete = append(ngsToDelete, ng)
			}
		}

		patches := []k8s.ObjectPatch{
			k8s.RemoveAnnotationPatch(meta.AnnotationKeyExternalCreatePending),
			k8s.SetAnnotationPatch(consts.AnnotationKeyDeletionTrigger, time.Now().Format(time.RFC3339)),
		}

		for _, ng := range ngsToDelete {
			if err := k8s.PatchObject(ctx, s.localCli, ng, patches...); err != nil {
				log.Error("failed to patch eks nodegroup", zap.String("data_plane_id", dataPlaneId),
					zap.String("nodegroup", ng.Name), zap.String("version", version), zap.Error(err))
				return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to patch eks nodegroup", err)
			}
			if err := controllers.DeleteUpboundResource(ctx, s.localCli, ng); err != nil {
				log.Error("failed to delete eks nodegroup", zap.String("data_plane_id", dataPlaneId),
					zap.String("nodegroup", ng.Name), zap.String("version", version), zap.Error(err))
				return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to delete eks nodegroup", err)
			}

			// delete the LaunchTemplate with the same name as the NodeGroup
			lt := &ec2v1beta1.LaunchTemplate{
				ObjectMeta: metav1.ObjectMeta{
					Name: ng.Name,
				},
			}
			if err := k8s.PatchObject(ctx, s.localCli, lt, patches...); err != nil {
				log.Error("failed to patch ec2 launchtemplate", zap.String("data_plane_id", dataPlaneId),
					zap.String("launchtemplate", lt.Name), zap.String("version", version), zap.Error(err))
				return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to patch ec2 launchtemplate", err)
			}
			if err := controllers.DeleteUpboundResource(ctx, s.localCli, lt); err != nil {
				log.Error("failed to delete ec2 launchtemplate", zap.String("data_plane_id", dataPlaneId),
					zap.String("launchtemplate", lt.Name), zap.String("version", version), zap.Error(err))
				return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to delete ec2 launchtemplate", err)
			}

			log.Info("deleted eks nodegroup", zap.String("data_plane_id", dataPlaneId),
				zap.String("nodegroup", ng.Name), zap.String("version", version))
		}
		log.Info("deleted eks nodegroups by version", zap.String("data_plane_id", dataPlaneId),
			zap.String("version", version), zap.Int("count", len(ngsToDelete)))

	case consts.ProviderTypeACK:
		// ACK upgrade NodePools in place, no need to delete NodePools in the K8s upgrade process
		// so we choose to not implement this method for ACK provider
		return nil, bizerr.New(bizerr.InternalServerError, "method is not implemented")

	default:
		return nil, bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("unsupported provider: %s", appBox.Spec.Provider))
	}

	return &emptypb.Empty{}, nil
}

func (s *DataPlaneService) DeleteNodePool(
	ctx context.Context,
	req *dmpbv1.DeleteNodePoolRequest,
) (*emptypb.Empty, error) {
	dataPlaneID := req.GetDataPlaneId()
	nodePoolID := req.GetNodePoolId()
	if dataPlaneID == "" || nodePoolID == "" {
		log.Error("data_plane_id and node_pool_id are required for DeleteNodePool")
		return nil, bizerr.New(bizerr.ParameterInvalid, "data_plane_id and node_pool_id are required")
	}

	nodePoolCR := &v1alpha1.NodePool{}
	nodePoolKey := client.ObjectKey{
		Namespace: s.getNamespace(),
		Name:      nodePoolID,
	}
	if err := s.nodePoolRW.Get(ctx, nodePoolKey, nodePoolCR); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "nodepool not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get nodepool error", err)
	}

	if nodePoolCR.Labels[v1alpha1.DataPlaneIDKey] != dataPlaneID {
		return nil, bizerr.New(bizerr.ResourceNotFound, "nodepool not found")
	}

	if err := s.nodePoolRW.Delete(ctx, nodePoolCR); err != nil && !apierrors.IsNotFound(err) {
		log.Error("failed to delete nodepool cr", zap.String("data_plane_id", dataPlaneID),
			zap.String("nodepool", nodePoolCR.Name), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to delete nodepool", err)
	}

	return &emptypb.Empty{}, nil
}

func (s *DataPlaneService) UpdateNodePool(
	ctx context.Context,
	req *dmpbv1.UpdateNodePoolRequest,
) (*dmpbv1.NodePool, error) {
	if req.GetDataPlaneId() == "" || req.GetNodePoolId() == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "data_plane_id and node_pool_id are required")
	}
	if s.cloudProviderExtractor == nil {
		return nil, bizerr.New(bizerr.InternalServerError, "cloud provider extractor is not initialized")
	}
	if req.NodePool == nil {
		return nil, bizerr.New(bizerr.ParameterInvalid, "node_pool is required")
	}

	nodePoolCR := &v1alpha1.NodePool{}
	nodePoolKey := client.ObjectKey{
		Namespace: s.getNamespace(),
		Name:      req.NodePoolId,
	}
	if err := s.nodePoolRW.Get(ctx, nodePoolKey, nodePoolCR); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "nodepool not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get nodepool error", err)
	}

	if nodePoolCR.Labels[v1alpha1.DataPlaneIDKey] != req.DataPlaneId {
		return nil, bizerr.New(bizerr.ResourceNotFound, "nodepool not found")
	}

	nodePoolCR.Spec.Labels = req.NodePool.Labels
	nodePoolCR.Spec.Taints = nodePoolTaintsAPI2CR(req.NodePool.Taints)

	if err := s.nodePoolRW.Update(ctx, nodePoolCR); err != nil {
		return nil, bizerr.Wrap(bizerr.K8SClientError, "update nodepool error", err)
	}

	return nodePoolCR2API(nodePoolCR), nil
}

func (s *DataPlaneService) CheckNodePoolScaleActivities(
	ctx context.Context,
	req *dmpbv1.CheckNodePoolScaleActivitiesRequest,
) (*dmpbv1.CheckNodePoolScaleActivitiesResponse, error) {
	if req.GetDataPlaneId() == "" || req.GetNodePoolId() == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "data_plane_id and node_pool_id are required")
	}
	nodePoolCR := &v1alpha1.NodePool{}
	if err := s.nodePoolRW.Get(ctx, types.NamespacedName{
		Namespace: s.getNamespace(),
		Name:      req.NodePoolId,
	}, nodePoolCR); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "nodepool not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get nodepool error", err)
	}
	if nodePoolCR.Labels[v1alpha1.DataPlaneIDKey] != req.DataPlaneId {
		return nil, bizerr.New(bizerr.ResourceNotFound, "nodepool not found")
	}

	dataPlane := &v1alpha1.DataPlane{}
	if err := s.dataPlaneRW.Get(ctx, types.NamespacedName{
		Namespace: s.getNamespace(),
		Name:      req.DataPlaneId,
	}, dataPlane); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "dataplane not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get dataplane error", err)
	}
	if dataPlane.Status.AppBox == nil {
		return nil, bizerr.New(bizerr.ResourceNotReady, "dataplane dependencies are not ready")
	}

	appBox := &v1alpha1.APPBox{}
	if err := s.appBoxRW.Get(ctx, types.NamespacedName{
		Namespace: dataPlane.Status.AppBox.Namespace,
		Name:      dataPlane.Status.AppBox.Name,
	}, appBox); err != nil {
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get appbox error", err)
	}
	if appBox.Spec.Provider != consts.ProviderTypeEKS {
		return nil, bizerr.New(bizerr.UnsupportedBizType, "unsupported provider")
	}
	if appBox.Status.KubernetesClusterName == "" {
		return nil, bizerr.New(bizerr.ResourceNotReady, "kubernetes cluster name is empty")
	}
	if dataPlane.Spec.AccountRef.Role == "" || dataPlane.Spec.AccountRef.AccountID == "" {
		return nil, bizerr.New(bizerr.ResourceNotReady, "account role is not ready")
	}

	resp, err := s.cloudProviderExtractor.CheckNodePoolScaleActivities(ctx, nodePoolCR, dataPlane, appBox)
	if err != nil {
		return nil, bizerr.Wrap(bizerr.AWSClientError, "check nodepool scale activities error", err)
	}
	return resp, nil
}

func (s *DataPlaneService) ListNodePoolConfigs(
	ctx context.Context,
	req *dmpbv1.ListNodePoolConfigsRequest,
) (*dmpbv1.ListNodePoolConfigsResponse, error) {
	appBoxConf := s.config.APPBoxConfig
	nodePoolConfigs := make([]*dmpbv1.NodePoolConfig, 0, len(appBoxConf.SharedNodePools)+len(appBoxConf.NodePools))
	for i := range appBoxConf.SharedNodePools {
		nodePoolConfigs = append(
			nodePoolConfigs, nodePoolConfig2API(&appBoxConf.SharedNodePools[i], dmpbv1.NodeType_Shared))
	}
	for i := range appBoxConf.NodePools {
		nodePoolConfigs = append(
			nodePoolConfigs, nodePoolConfig2API(&appBoxConf.NodePools[i], dmpbv1.NodeType_Dedicated))
	}
	return &dmpbv1.ListNodePoolConfigsResponse{
		NodePoolConfigs:    nodePoolConfigs,
		AvailableNodeZones: append([]string(nil), s.config.RegionConfig.Zones...),
	}, nil
}

func (s *DataPlaneService) getNamespace() string {
	return s.config.Namespace
}

func (s *DataPlaneService) listNodePoolCRs(ctx context.Context, dataPlaneID string) ([]*v1alpha1.NodePool, error) {
	var nodePoolList v1alpha1.NodePoolList
	if err := s.nodePoolRW.List(ctx, &nodePoolList, client.InNamespace(s.getNamespace()), client.MatchingLabels{
		v1alpha1.DataPlaneIDKey: dataPlaneID,
	}); err != nil {
		return nil, err
	}

	ns := make([]*v1alpha1.NodePool, 0, len(nodePoolList.Items))
	for i := range nodePoolList.Items {
		np := &nodePoolList.Items[i]
		ns = append(ns, np)
	}

	return ns, nil
}

func dataPlaneCR2API(cr *v1alpha1.DataPlane) (*dmpbv1.DataPlane, error) {
	nodeZones := append([]string(nil), cr.Spec.NodeZones...)
	if len(nodeZones) == 0 {
		nodeZones = append([]string(nil), cr.Spec.AvailabilityZones...)
	}

	ownershipMode, err := dataPlaneOwnershipModeCR2API(cr.Spec.OwnershipMode)
	if err != nil {
		return nil, err
	}

	return &dmpbv1.DataPlane{
		Id:        cr.Name,
		AccountId: cr.Spec.AccountRef.AccountID,
		Tags:      cr.Spec.CloudTags,
		Vpc: &dmpbv1.DataPlane_Vpc{
			Cidr: cr.Spec.VpcCidr,
		},
		Kubernetes: &dmpbv1.DataPlane_Kubernetes{},
		Pca:        pcaConfigCR2API(cr.Spec.PCA),
		NodeZones:  nodeZones,
		HighAvailabilityType: func() dmpbv1.DataPlane_HighAvailabilityType {
			if len(nodeZones) == 1 {
				return dmpbv1.DataPlane_ZONAL
			}
			return dmpbv1.DataPlane_REGIONAL
		}(),
		ResourceGroupName: cr.Spec.AccountRef.ResourceGroupName,
		Status:            dataPlanePhase2APIStatus(cr.Status.Phase),
		OwnershipMode:     ownershipMode,
	}, nil
}

func apiDataPlaneOwnershipModeToCR(mode dmpbv1.DataPlaneOwnershipMode) (v1alpha1.DataPlaneOwnershipMode, error) {
	switch mode {
	case dmpbv1.DataPlaneOwnershipMode_DATA_PLANE_OWNERSHIP_MODE_UNSPECIFIED,
		dmpbv1.DataPlaneOwnershipMode_SINGLE_POOL_OWNERSHIP:
		return v1alpha1.DataPlaneOwnershipModeSinglePool, nil
	case dmpbv1.DataPlaneOwnershipMode_MULTI_POOL_OWNERSHIP:
		return v1alpha1.DataPlaneOwnershipModeMultiPool, nil
	default:
		return "", fmt.Errorf("unknown dataplane ownership mode %d", mode)
	}
}

func dataPlaneOwnershipModeCR2API(mode v1alpha1.DataPlaneOwnershipMode) (dmpbv1.DataPlaneOwnershipMode, error) {
	effectiveMode, err := v1alpha1.EffectiveDataPlaneOwnershipMode(mode)
	if err != nil {
		return dmpbv1.DataPlaneOwnershipMode_DATA_PLANE_OWNERSHIP_MODE_UNSPECIFIED, err
	}
	switch effectiveMode {
	case v1alpha1.DataPlaneOwnershipModeSinglePool:
		return dmpbv1.DataPlaneOwnershipMode_SINGLE_POOL_OWNERSHIP, nil
	case v1alpha1.DataPlaneOwnershipModeMultiPool:
		return dmpbv1.DataPlaneOwnershipMode_MULTI_POOL_OWNERSHIP, nil
	default:
		return dmpbv1.DataPlaneOwnershipMode_DATA_PLANE_OWNERSHIP_MODE_UNSPECIFIED,
			fmt.Errorf("unknown effective dataplane ownership mode %q", effectiveMode)
	}
}

func pcaConfigCR2API(cfg *v1alpha1.PCAConfig) *dmpbv1.DataPlane_PCA {
	if cfg == nil {
		return nil
	}
	return &dmpbv1.DataPlane_PCA{
		CaIdentifier: cfg.CAIdentifier,
	}
}

func nodePoolCR2API(cr *v1alpha1.NodePool) *dmpbv1.NodePool {
	if cr == nil {
		return nil
	}
	out := &dmpbv1.NodePool{
		Name:                  cr.Name,
		InstanceType:          cr.Spec.InstanceType,
		K8SVersion:            cr.Spec.KubernetesVersion,
		CpuCore:               cr.Spec.CpuCore,
		AvailableCpuMilliCore: cr.Spec.AvailableCpuMilliCore,
		MaxCpuMilli:           cr.Spec.MaxCpuMilli,
		MemoryGi:              cr.Spec.MemoryGi,
		AvailableMemoryMi:     cr.Spec.AvailableMemoryMi,
		MaxMemoryMi:           cr.Spec.MaxMemoryMi,
		WithLocalDisk:         cr.Spec.WithLocalDisk,
		LocalDiskSizeGi:       cr.Spec.LocalDiskSizeGi,
		// Round-trip the durable placement_group_enabled value so a single
		// version of the field drives both producers and consumers, and so
		// a stale CSNG/DPM pair cannot silently disagree.
		PlacementGroupEnabled: cr.Spec.PlacementGroupEnabled,
		Labels:                cr.Spec.Labels,
		Taints:                nodePoolTaintsCR2API(cr.Spec.Taints),
		CurrentReplicas:       cr.Status.CurrentReplicas,
		Status:                nodePoolPhase2APIStatus(cr.Status.Phase),
	}
	if cr.Spec.NodeType != "" {
		if nodeType, ok := dmpbv1.NodeType_value[cr.Spec.NodeType]; ok {
			out.NodeType = dmpbv1.NodeType(nodeType)
		}
	}
	return out
}

func nodePoolPhase2APIStatus(phase v1alpha1.NodePoolPhase) dmpbv1.NodePoolStatus {
	switch phase {
	case v1alpha1.NodePoolPhaseCreating, "":
		return dmpbv1.NodePoolStatus_NODEPOOL_CREATING
	case v1alpha1.NodePoolPhaseActive:
		return dmpbv1.NodePoolStatus_NODEPOOL_ACTIVE
	case v1alpha1.NodePoolPhaseFailed:
		return dmpbv1.NodePoolStatus_NODEPOOL_FAILED
	case v1alpha1.NodePoolPhaseDeleting:
		return dmpbv1.NodePoolStatus_NODEPOOL_DELETING
	default:
		return dmpbv1.NodePoolStatus_NODEPOOL_STATUS_UNSPECIFIED
	}
}

func nodePoolTaintsAPI2CR(taints []*dmpbv1.NodePoolTaint) []v1alpha1.NodePoolTaint {
	if len(taints) == 0 {
		return nil
	}
	out := make([]v1alpha1.NodePoolTaint, 0, len(taints))
	for _, t := range taints {
		if t == nil {
			continue
		}
		out = append(out, v1alpha1.NodePoolTaint{
			Key:    t.Key,
			Value:  t.Value,
			Effect: t.Effect,
		})
	}
	return out
}

func nodePoolTaintsCR2API(taints []v1alpha1.NodePoolTaint) []*dmpbv1.NodePoolTaint {
	if len(taints) == 0 {
		return nil
	}
	out := make([]*dmpbv1.NodePoolTaint, 0, len(taints))
	for i := range taints {
		t := taints[i]
		out = append(out, &dmpbv1.NodePoolTaint{
			Key:    t.Key,
			Value:  t.Value,
			Effect: t.Effect,
		})
	}
	return out
}

func buildNodePoolName(resourceNamePrefix, dataPlaneID, instanceType, k8sVersion string, cpu, memory uint32) string {
	identity := dataPlaneID
	instancePart := strings.ToLower(strings.NewReplacer(".", "-", "_", "-").Replace(instanceType))
	versionPart := strings.ReplaceAll(k8sVersion, ".", "-")
	specPart := fmt.Sprintf("%dc%dg", cpu, memory)
	prefix := fmt.Sprintf("%s-%s-%s", resourceNamePrefix, instancePart, specPart)
	return namer.NewName(identity, prefix, versionPart, -1)
}

func effectiveResourceNamePrefix(override, fallback string) string {
	if override != "" {
		return override
	}
	return fallback
}

func dataPlanePhase2APIStatus(phase v1alpha1.DataPlanePhase) dmpbv1.DataPlane_Status {
	switch phase {
	case v1alpha1.DataPlanePhaseCreating:
		return dmpbv1.DataPlane_Creating
	case v1alpha1.DataPlanePhaseActive:
		return dmpbv1.DataPlane_Active
	case v1alpha1.DataPlanePhaseFailed:
		return dmpbv1.DataPlane_Failed
	case v1alpha1.DataPlanePhaseDeleting:
		return dmpbv1.DataPlane_Deleting
	default:
		return dmpbv1.DataPlane_Creating
	}
}

func nodePoolConfig2API(conf *rsconfig.NodePoolConfig, nodeType dmpbv1.NodeType) *dmpbv1.NodePoolConfig {
	availMilli := nodePoolConfigAvailableCPUMilli(conf)
	return &dmpbv1.NodePoolConfig{
		InstanceType:          conf.InstanceType,
		CpuCore:               uint32(conf.CPU),
		AvailableCpuMilliCore: availMilli,
		MaxCpuMilli:           nodePoolConfigMaxCPUMilli(conf, availMilli),
		MemoryGi:              uint32(conf.MemoryGi),
		AvailableMemoryMi:     uint32(conf.AvailableMemoryMi),
		MaxMemoryMi:           nodePoolConfigMaxMemoryMi(conf),
		WithLocalDisk:         conf.HasLocalDisk,
		LocalDiskSizeGi:       uint32(conf.LocalDiskSize),
		NodeType:              nodeType,
	}
}

func nodePoolConfigAvailableCPUMilli(conf *rsconfig.NodePoolConfig) uint32 {
	if conf == nil {
		return 0
	}
	if conf.AvailableCPUMilli > 0 {
		return uint32(conf.AvailableCPUMilli)
	}
	if conf.AvailableCPU > 0 {
		return uint32(conf.AvailableCPU) * 1000
	}
	if conf.CPU > 1 {
		return uint32(conf.CPU-1) * 1000
	}
	return 0
}

func nodePoolConfigMaxCPUMilli(conf *rsconfig.NodePoolConfig, availMilli uint32) uint32 {
	if conf != nil && conf.MaxCPUMilli > 0 {
		return uint32(conf.MaxCPUMilli)
	}
	return availMilli
}

func nodePoolConfigMaxMemoryMi(conf *rsconfig.NodePoolConfig) uint32 {
	if conf != nil && conf.MaxMemoryMi > 0 {
		return uint32(conf.MaxMemoryMi)
	}
	if conf == nil {
		return 0
	}
	return uint32(conf.AvailableMemoryMi)
}

func nodePoolAPIMaxCPUMilli(np *dmpbv1.NodePool) uint32 {
	if np == nil {
		return 0
	}
	if np.MaxCpuMilli > 0 {
		return np.MaxCpuMilli
	}
	return np.AvailableCpuMilliCore
}

func nodePoolAPIMaxMemoryMi(np *dmpbv1.NodePool) uint32 {
	if np == nil {
		return 0
	}
	if np.MaxMemoryMi > 0 {
		return np.MaxMemoryMi
	}
	return np.AvailableMemoryMi
}