package server

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

	gkev1beta1 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	xpmeta "github.com/crossplane/crossplane-runtime/pkg/meta"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	shared "github.com/tidbcloud/infra-provider/pkg/controllers/common"
	"github.com/tidbcloud/infra-provider/pkg/gkeheterogeneous"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	nodepoolutil "github.com/tidbcloud/infra-provider/pkg/nodepool"
)

const (
	gkeNodePool3LayerTaintKey   = "3-layer-arch"
	gkeNodePool3LayerTaintValue = "true"
	gkeDiskTypePdStandard       = "pd-standard"
	gkeDiskTypeHyperdisk        = "hyperdisk-balanced"
)

func buildGKEHeterogeneousNodePool(cluster *cloudv1.Cluster, nodePoolName, instanceType string, minNodeCount, maxNodeCount int64) *gkev1beta1.NodePool {
	labels := gkeHeterogeneousSharedNodePoolLabels(cluster)
	configLabels := gkeHeterogeneousNodePoolConfigLabels(cluster, gkeHeterogeneousNodePoolConfigVersion(cluster))
	locations := []string(nil)
	if cluster.Spec.Kubernetes.Cloud.GCP != nil {
		locations = append(locations, cluster.Spec.Kubernetes.Cloud.GCP.NodeZones...)
	}
	providerConfigName := namer.NewName(cluster.Spec.Kubernetes.Name, "pc", "", -1)
	serviceAccountName := namer.NewName(cluster.Spec.Kubernetes.Name, "np-sa", "", namer.GKEServiceAccountNameLimit)
	if cluster.Spec.NetworkName != "" {
		networkScope := gkeHeterogeneousNetworkObjectMeta(cluster)
		providerConfigName = namer.UniqueIdentifier(networkScope)
		serviceAccountName = namer.UniqueIdentifier(networkScope, namer.GCPNodePoolServiceAccountNameOptions()...)
	}
	serviceAccountEmail := fmt.Sprintf("%s@%s.iam.gserviceaccount.com", serviceAccountName, cluster.Spec.Kubernetes.Account)
	taints := []*gkev1beta1.NodeTaint{
		{
			Key:    "dedicated",
			Value:  "tidb-" + shared.ComponentShared,
			Effect: "NO_SCHEDULE",
		},
		{
			Key:    consts.GKEHeterogeneousNodePoolIsolationTaintKey,
			Value:  consts.GKEHeterogeneousNodePoolIsolationTaintValue,
			Effect: "NO_SCHEDULE",
		},
	}
	if cluster.GetProviderType() == cloudv1.ProviderTypeGKE && cluster.FeatureEnabled(consts.ClusterFeatureGCP3LayerNodePool) {
		taints = append(taints, &gkev1beta1.NodeTaint{
			Key:    gkeNodePool3LayerTaintKey,
			Value:  gkeNodePool3LayerTaintValue,
			Effect: "NO_SCHEDULE",
		})
	}

	np := &gkev1beta1.NodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name:   nodePoolName,
			Labels: labels,
			Annotations: map[string]string{
				xpmeta.AnnotationKeyExternalName: nodePoolName,
			},
		},
		Spec: gkev1beta1.NodePoolSpec{
			ForProvider: gkev1beta1.NodePoolParameters{
				Cluster: gkeHeterogeneousNodePoolClusterRef(cluster),
				Autoscaling: &gkev1beta1.NodePoolAutoscaling{
					Enabled:      ptr.To(true),
					MaxNodeCount: ptr.To(maxNodeCount),
					MinNodeCount: ptr.To(minNodeCount),
				},
				Config: &gkev1beta1.NodeConfig{
					Labels:         configLabels,
					DiskSizeGb:     ptr.To(int64(100)),
					DiskType:       ptr.To(gkeHeterogeneousNodePoolDiskType(instanceType)),
					ImageType:      ptr.To("cos_containerd"),
					MachineType:    ptr.To(instanceType),
					ServiceAccount: ptr.To(serviceAccountEmail),
					OauthScopes: []string{
						"https://www.googleapis.com/auth/cloud-platform",
					},
					Taints: taints,
				},
				InitialNodeCount: ptr.To(int64(0)),
				Locations:        locations,
				Management: &gkev1beta1.NodeManagementSpec{
					AutoUpgrade: ptr.To(false),
					AutoRepair:  ptr.To(false),
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{Name: providerConfigName},
			},
		},
	}
	if scopeKey, ok := nodepoolutil.GKENodePoolNamingScopeKey(cluster); ok {
		np.Annotations[consts.AnnoKeyGKEHeterogeneousNodePoolScopeKey] = scopeKey
	}
	return np
}

func (s *ClusterServer) gkeHeterogeneousNodePoolClient() client.Client {
	if s.NodePoolClient != nil {
		return s.NodePoolClient
	}
	return s.KubeClient
}

func (s *ClusterServer) getGKEHeterogeneousNodePoolResource(ctx context.Context, nodePoolName string) (*gkev1beta1.NodePool, error) {
	nodePoolName = strings.TrimSpace(nodePoolName)
	if nodePoolName == "" {
		return nil, status.Error(codes.NotFound, "heterogeneous nodepool resource name is empty")
	}

	np := &gkev1beta1.NodePool{}
	if err := s.gkeHeterogeneousNodePoolClient().Get(ctx, client.ObjectKey{Name: nodePoolName}, np); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "heterogeneous nodepool resource %s not found", nodePoolName)
		}
		return nil, status.Errorf(codes.Internal, "get heterogeneous nodepool %s: %s", nodePoolName, err)
	}
	return np, nil
}

func (s *ClusterServer) ensureGKENodePoolResource(ctx context.Context, cluster *cloudv1.Cluster, nodePoolName, instanceType string, minNodeCount, maxNodeCount int64) (*gkev1beta1.NodePool, error) {
	nodePoolName = strings.TrimSpace(nodePoolName)
	if nodePoolName == "" {
		return nil, status.Error(codes.FailedPrecondition, "nodepool resource name is empty")
	}

	nodePoolClient := s.gkeHeterogeneousNodePoolClient()
	current := &gkev1beta1.NodePool{}
	if err := nodePoolClient.Get(ctx, client.ObjectKey{Name: nodePoolName}, current); err != nil {
		if !apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.Internal, "get nodepool %s: %s", nodePoolName, err)
		}

		if _, err := requireGKEHeterogeneousNodePoolScopeLabels(cluster); err != nil {
			return nil, err
		}
		desiredNodePool := buildGKEHeterogeneousNodePool(cluster, nodePoolName, instanceType, minNodeCount, maxNodeCount)
		if len(desiredNodePool.Spec.ForProvider.Locations) == 0 {
			// Legacy Cluster CRs do not carry GCP NodeZones. Ordinary NodePools
			// in the same GKE cluster share locations, so use the TiKV pool as
			// the stable source instead of looking up the new target type.
			locationSourceInstanceType := strings.TrimSpace(cluster.Spec.Cluster.TiKV.InstanceType)
			locations, err := s.ordinaryGKENodePoolLocations(ctx, cluster, locationSourceInstanceType)
			if err != nil {
				return nil, err
			}
			desiredNodePool.Spec.ForProvider.Locations = locations
		}
		if err := nodePoolClient.Create(ctx, desiredNodePool); err == nil {
			return desiredNodePool, nil
		} else if !apierrors.IsAlreadyExists(err) {
			return nil, status.Errorf(codes.Internal, "create nodepool %s: %s", nodePoolName, err)
		}

		// Another caller may create the same API-managed NodePool between our Get and Create.
		// Re-read it and validate below instead of treating the name collision as success.
		if err := nodePoolClient.Get(ctx, client.ObjectKey{Name: nodePoolName}, current); err != nil {
			return nil, status.Errorf(codes.Internal, "get nodepool %s after create conflict: %s", nodePoolName, err)
		}
	}

	// Existing resources are reused only when they already match the API-owned shape.
	// This path is intentionally validate-only: conflicting shared NodePools must surface
	// as precondition errors instead of being overwritten by the create API.
	if err := validateManagedGKEHeterogeneousNodePool(current); err != nil {
		return nil, err
	}
	if err := validateGKENodePoolMachineType(current, instanceType); err != nil {
		return nil, err
	}
	if err := validateGKEHeterogeneousNodePoolScope(current, cluster); err != nil {
		return nil, err
	}
	if err := validateGKEHeterogeneousNodePoolClusterRef(current, cluster); err != nil {
		return nil, err
	}
	if err := validateGKEHeterogeneousNodePoolAutoscalingShape(current); err != nil {
		return nil, err
	}
	// Only backfill Crossplane identity after the existing resource is proven
	// to be the API-managed NodePool for this cluster, machine type, and autoscaling shape.
	if err := ensureGKEHeterogeneousNodePoolExternalName(ctx, nodePoolClient, current, nodePoolName); err != nil {
		return nil, err
	}
	return current, nil
}

func (s *ClusterServer) ordinaryGKENodePoolLocations(ctx context.Context, cluster *cloudv1.Cluster, instanceType string) ([]string, error) {
	instanceType = strings.TrimSpace(instanceType)
	if instanceType == "" {
		return nil, status.Error(codes.FailedPrecondition, "ordinary nodepool instance type is required to resolve locations")
	}
	nodePoolName := namer.GKENodePoolNameForCluster(cluster, instanceType)
	nodePool := &gkev1beta1.NodePool{}
	if err := s.gkeHeterogeneousNodePoolClient().Get(ctx, client.ObjectKey{Name: nodePoolName}, nodePool); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.FailedPrecondition, "ordinary nodepool %s is required to resolve locations", nodePoolName)
		}
		return nil, status.Errorf(codes.Internal, "get ordinary nodepool %s to resolve locations: %s", nodePoolName, err)
	}
	if len(nodePool.Spec.ForProvider.Locations) != 3 {
		return nil, status.Errorf(
			codes.FailedPrecondition,
			"ordinary nodepool %s has %d locations, want 3",
			nodePoolName,
			len(nodePool.Spec.ForProvider.Locations),
		)
	}
	return append([]string(nil), nodePool.Spec.ForProvider.Locations...), nil
}

func ensureGKEHeterogeneousNodePoolExternalName(ctx context.Context, nodePoolClient client.Client, nodePool *gkev1beta1.NodePool, expectedExternalName string) error {
	if nodePool == nil {
		return status.Error(codes.Internal, "heterogeneous nodepool resource is nil")
	}
	expectedExternalName = strings.TrimSpace(expectedExternalName)
	if expectedExternalName == "" {
		return status.Error(codes.FailedPrecondition, "nodepool external name is empty")
	}
	actualExternalName := strings.TrimSpace(nodePool.GetAnnotations()[xpmeta.AnnotationKeyExternalName])
	if actualExternalName == expectedExternalName {
		return nil
	}
	if actualExternalName != "" {
		return status.Errorf(
			codes.FailedPrecondition,
			"nodepool %s has external name %s, want %s",
			nodePool.GetName(),
			actualExternalName,
			expectedExternalName,
		)
	}

	updated := nodePool.DeepCopy()
	annotations := updated.GetAnnotations()
	if annotations == nil {
		annotations = map[string]string{}
	}
	annotations[xpmeta.AnnotationKeyExternalName] = expectedExternalName
	updated.SetAnnotations(annotations)
	if err := nodePoolClient.Patch(ctx, updated, client.MergeFrom(nodePool)); err != nil {
		return status.Errorf(codes.Internal, "set external name for heterogeneous nodepool %s: %s", nodePool.GetName(), err)
	}
	nodePool.SetAnnotations(updated.GetAnnotations())
	return nil
}

func validateGKEHeterogeneousNodePoolAutoscalingShape(nodePool *gkev1beta1.NodePool) error {
	if nodePool == nil {
		return status.Error(codes.Internal, "heterogeneous nodepool resource is nil")
	}
	autoscaling := nodePool.Spec.ForProvider.Autoscaling
	if autoscaling == nil {
		return status.Errorf(codes.FailedPrecondition, "nodepool %s does not have autoscaling configured", nodePool.Name)
	}
	if autoscaling.Enabled == nil || !*autoscaling.Enabled {
		return status.Errorf(codes.FailedPrecondition, "nodepool %s does not have autoscaling enabled", nodePool.Name)
	}
	if autoscaling.MinNodeCount == nil || autoscaling.MaxNodeCount == nil {
		return status.Errorf(codes.FailedPrecondition, "nodepool %s does not have autoscaling min/max configured", nodePool.Name)
	}
	minNodeCount := *autoscaling.MinNodeCount
	maxNodeCount := *autoscaling.MaxNodeCount
	if minNodeCount < 0 || maxNodeCount < 0 || minNodeCount > maxNodeCount {
		return status.Errorf(
			codes.FailedPrecondition,
			"nodepool %s has invalid autoscaling min/max %d/%d",
			nodePool.Name,
			minNodeCount,
			maxNodeCount,
		)
	}
	return nil
}

func validateGKENodePoolMachineType(nodePool *gkev1beta1.NodePool, expectedInstanceType string) error {
	if nodePool == nil {
		return status.Error(codes.Internal, "heterogeneous nodepool resource is nil")
	}
	actual, ok := gkeheterogeneous.NodePoolMachineType(nodePool)
	if !ok {
		return status.Errorf(codes.FailedPrecondition, "nodepool %s does not have a machine type configured", nodePool.Name)
	}
	if actual != strings.TrimSpace(expectedInstanceType) {
		return status.Errorf(
			codes.FailedPrecondition,
			"nodepool %s has machine type %s, want %s",
			nodePool.Name,
			actual,
			expectedInstanceType,
		)
	}
	return nil
}

func validateManagedGKEHeterogeneousNodePool(nodePool *gkev1beta1.NodePool) error {
	if nodePool == nil {
		return status.Error(codes.Internal, "heterogeneous nodepool resource is nil")
	}
	if !gkeheterogeneous.IsAPIManagedNodePool(nodePool) {
		return status.Errorf(codes.FailedPrecondition, "nodepool %s is not managed by heterogeneous nodepool API", nodePool.GetName())
	}
	return nil
}

func validateGKEHeterogeneousNodePoolScope(nodePool *gkev1beta1.NodePool, cluster *cloudv1.Cluster) error {
	if cluster == nil {
		return status.Error(codes.Internal, "cluster is nil")
	}
	if nodePool == nil {
		return status.Error(codes.Internal, "heterogeneous nodepool resource is nil")
	}
	expectedLabels, err := requireGKEHeterogeneousNodePoolScopeLabels(cluster)
	if err != nil {
		return err
	}
	if err := gkeheterogeneous.ValidateNodePoolScopeLabels(nodePool, cluster.GetName(), "GKE heterogeneous nodepool scope", expectedLabels); err != nil {
		return status.Error(codes.FailedPrecondition, err.Error())
	}
	return nil
}

func validateGKEHeterogeneousNodePoolClusterRef(nodePool *gkev1beta1.NodePool, cluster *cloudv1.Cluster) error {
	if cluster == nil {
		return status.Error(codes.Internal, "cluster is nil")
	}
	if nodePool == nil {
		return status.Error(codes.Internal, "heterogeneous nodepool resource is nil")
	}
	expected := strings.TrimSpace(gkeHeterogeneousNodePoolClusterRef(cluster))
	if actual := gkeheterogeneous.NodePoolClusterRef(nodePool); actual != expected {
		return status.Errorf(
			codes.PermissionDenied,
			"nodepool %s does not belong to cluster %s GKE cluster ref: cluster=%q, want %q",
			nodePool.GetName(),
			cluster.GetName(),
			actual,
			expected,
		)
	}
	return nil
}

type gkeHeterogeneousNodePoolComputeClassMetadata struct {
	desired   string
	applied   string
	specLabel string
	specTaint string
}

func gkeHeterogeneousNodePoolComputeClassMetadataForNodePool(nodePool *gkev1beta1.NodePool) gkeHeterogeneousNodePoolComputeClassMetadata {
	return gkeHeterogeneousNodePoolComputeClassMetadata{
		desired:   nodepoolutil.GKEDesiredComputeClassNameForNodePool(nodePool),
		applied:   nodepoolutil.GKEAppliedComputeClassNameForNodePool(nodePool),
		specLabel: nodepoolutil.GKEComputeClassSpecLabelForNodePool(nodePool),
		specTaint: nodepoolutil.GKEComputeClassSpecTaintForNodePool(nodePool),
	}
}

func observedGKEHeterogeneousNodePoolComputeClass(nodePool *gkev1beta1.NodePool) string {
	metadata := gkeHeterogeneousNodePoolComputeClassMetadataForNodePool(nodePool)
	for _, computeClass := range []string{metadata.desired, metadata.applied, metadata.specLabel, metadata.specTaint} {
		if computeClass != "" {
			return computeClass
		}
	}
	return ""
}

func validateGKEHeterogeneousNodePoolComputeClass(nodePool *gkev1beta1.NodePool, expectedComputeClass string) error {
	expectedComputeClass = strings.TrimSpace(expectedComputeClass)
	if expectedComputeClass == "" {
		return nil
	}
	if nodePool == nil {
		return status.Error(codes.Internal, "heterogeneous nodepool resource is nil")
	}
	metadata := gkeHeterogeneousNodePoolComputeClassMetadataForNodePool(nodePool)
	if metadata.desired != "" && metadata.desired != expectedComputeClass {
		return status.Errorf(codes.FailedPrecondition, "nodepool %s is bound to desired compute class %s, want %s", nodePool.GetName(), metadata.desired, expectedComputeClass)
	}
	if metadata.applied != "" && metadata.applied != expectedComputeClass {
		return status.Errorf(codes.FailedPrecondition, "nodepool %s is bound to applied compute class %s, want %s", nodePool.GetName(), metadata.applied, expectedComputeClass)
	}
	if metadata.specLabel != "" && metadata.specLabel != expectedComputeClass {
		return status.Errorf(codes.FailedPrecondition, "nodepool %s has compute class spec label %s, want %s", nodePool.GetName(), metadata.specLabel, expectedComputeClass)
	}
	if metadata.specTaint != "" && metadata.specTaint != expectedComputeClass {
		return status.Errorf(codes.FailedPrecondition, "nodepool %s has compute class spec taint %s, want %s", nodePool.GetName(), metadata.specTaint, expectedComputeClass)
	}
	return nil
}

func validateGKEHeterogeneousNodePoolBindingResource(cluster *cloudv1.Cluster, state *gkeHeterogeneousNodePoolBindingState, binding cloudv1.GKEComputeClassNodePoolBinding, nodePool *gkev1beta1.NodePool) error {
	if err := validateGKEHeterogeneousNodePoolBindingIdentity(cluster, binding, nodePool); err != nil {
		return err
	}
	return validateGKEHeterogeneousNodePoolComputeClass(nodePool, state.computeClassName)
}

func validateGKEHeterogeneousNodePoolBindingIdentity(cluster *cloudv1.Cluster, binding cloudv1.GKEComputeClassNodePoolBinding, nodePool *gkev1beta1.NodePool) error {
	if err := validateManagedGKEHeterogeneousNodePool(nodePool); err != nil {
		return err
	}
	if err := validateGKENodePoolMachineType(nodePool, binding.InstanceType); err != nil {
		return err
	}
	if err := validateGKEHeterogeneousNodePoolScope(nodePool, cluster); err != nil {
		return err
	}
	if err := validateGKEHeterogeneousNodePoolClusterRef(nodePool, cluster); err != nil {
		return err
	}
	return nil
}

func gkeHeterogeneousNodePoolResourceResponse(nodePool *gkev1beta1.NodePool, computeClass string) *apiv1.GKEHeterogeneousNodePool {
	resp := &apiv1.GKEHeterogeneousNodePool{
		ComputeClassName: strings.TrimSpace(computeClass),
		Exists:           true,
	}
	if nodePool == nil {
		return resp
	}
	resp.NodePoolName = nodePool.GetName()
	if nodePool.Spec.ForProvider.Config != nil && nodePool.Spec.ForProvider.Config.MachineType != nil {
		resp.InstanceType = strings.TrimSpace(*nodePool.Spec.ForProvider.Config.MachineType)
	}
	resp.MinNodeCount, resp.MaxNodeCount = gkeHeterogeneousNodePoolAutoscalingCounts(nodePool)
	return resp
}

func (s *ClusterServer) appendPreparedGKEHeterogeneousNodePools(ctx context.Context, cluster *cloudv1.Cluster, current []*apiv1.GKEHeterogeneousNodePool) ([]*apiv1.GKEHeterogeneousNodePool, error) {
	nodePools, err := s.listPreparedGKEHeterogeneousNodePoolResources(ctx, cluster)
	if err != nil {
		return nil, err
	}

	seen := make(map[string]struct{}, len(current))
	for _, nodePool := range current {
		if nodePool == nil {
			continue
		}
		if nodePoolName := strings.TrimSpace(nodePool.NodePoolName); nodePoolName != "" {
			seen[nodePoolName] = struct{}{}
		}
	}

	prepared := make([]*apiv1.GKEHeterogeneousNodePool, 0)
	for _, nodePool := range nodePools {
		nodePoolName := strings.TrimSpace(nodePool.GetName())
		if nodePoolName == "" {
			continue
		}
		if _, ok := seen[nodePoolName]; ok {
			continue
		}
		// ComputeClass metadata is derived from Cluster intent and can outlive
		// the intent while controller cleanup is still pending. Return it as a
		// prepared/residual nodepool instead of hiding the API-managed resource.
		resp := gkeHeterogeneousNodePoolResourceResponse(nodePool, observedGKEHeterogeneousNodePoolComputeClass(nodePool))
		resp.ClusterId = cluster.GetClusterID()
		resp.IsBaseNodePool = false
		resp.ApiManaged = true
		prepared = append(prepared, resp)
		seen[nodePoolName] = struct{}{}
	}
	sort.Slice(prepared, func(i, j int) bool {
		return prepared[i].NodePoolName < prepared[j].NodePoolName
	})
	return append(current, prepared...), nil
}

func (s *ClusterServer) listPreparedGKEHeterogeneousNodePoolResources(ctx context.Context, cluster *cloudv1.Cluster) ([]*gkev1beta1.NodePool, error) {
	if cluster == nil {
		return nil, status.Error(codes.Internal, "cluster is nil")
	}
	if _, err := requireGKEHeterogeneousNodePoolScopeLabels(cluster); err != nil {
		return nil, err
	}

	nodePoolList := &gkev1beta1.NodePoolList{}
	if err := s.gkeHeterogeneousNodePoolClient().List(ctx, nodePoolList, client.MatchingLabels{
		consts.GKEHeterogeneousNodePoolManagedByLabelKey: consts.GKEHeterogeneousNodePoolManagedByLabelValue,
	}); err != nil {
		return nil, status.Errorf(codes.Internal, "list prepared heterogeneous nodepools: %s", err)
	}

	prepared := make([]*gkev1beta1.NodePool, 0, len(nodePoolList.Items))
	for i := range nodePoolList.Items {
		nodePool := &nodePoolList.Items[i]
		if strings.TrimSpace(nodePool.GetName()) == "" {
			continue
		}
		if err := validateGKEHeterogeneousNodePoolScope(nodePool, cluster); err != nil {
			continue
		}
		if err := validateGKEHeterogeneousNodePoolClusterRef(nodePool, cluster); err != nil {
			continue
		}
		prepared = append(prepared, nodePool.DeepCopy())
	}
	sort.Slice(prepared, func(i, j int) bool {
		return prepared[i].GetName() < prepared[j].GetName()
	})
	return prepared, nil
}

func gkeHeterogeneousNodePoolAutoscalingCounts(nodePool *gkev1beta1.NodePool) (minNodeCount, maxNodeCount int64) {
	if nodePool == nil || nodePool.Spec.ForProvider.Autoscaling == nil {
		return 0, 0
	}
	if nodePool.Spec.ForProvider.Autoscaling.MinNodeCount != nil {
		minNodeCount = *nodePool.Spec.ForProvider.Autoscaling.MinNodeCount
	}
	if nodePool.Spec.ForProvider.Autoscaling.MaxNodeCount != nil {
		maxNodeCount = *nodePool.Spec.ForProvider.Autoscaling.MaxNodeCount
	}
	return minNodeCount, maxNodeCount
}

func gkeHeterogeneousSharedNodePoolLabels(cluster *cloudv1.Cluster) map[string]string {
	sharedNamer := namer.NewNamer().
		WithClusterName(shared.ClustersShared).
		WithComponent(shared.ComponentShared).
		WithTiDBNamespace(shared.ClustersShared)
	if cluster != nil && cluster.Spec.NetworkName != "" {
		networkScope := gkeHeterogeneousNetworkObjectMeta(cluster)
		sharedNamer = sharedNamer.
			LegacyWithNetworkName(networkScope.Name).
			LegacyWithNetworkNamespace(networkScope.Namespace)
	}

	labels := sharedNamer.Labels()
	labels[consts.GKEHeterogeneousNodePoolManagedByLabelKey] = consts.GKEHeterogeneousNodePoolManagedByLabelValue
	scopeLabels, _ := nodepoolutil.GKEHeterogeneousNodePoolScopeLabels(cluster)
	for key, value := range scopeLabels {
		labels[key] = value
	}
	if cluster != nil && len(scopeLabels) > 0 && cluster.GetProviderType() == cloudv1.ProviderTypeGKE && cluster.FeatureEnabled(consts.ClusterFeatureGCP3LayerNodePool) {
		labels[cloudv1.TagLabelPrefix+shared.Project] = cluster.GetProjectID()
		labels[cloudv1.TagLabelPrefix+shared.Tenant] = cluster.GetTenantID()
	}
	return labels
}

func gkeHeterogeneousNodePoolConfigLabels(cluster *cloudv1.Cluster, nodeVersion string) map[string]string {
	labels := cloudv1.TagsFromLabels(map[string]string{
		cloudv1.TagLabelPrefix + shared.UsedBy:      shared.CustomerTiDB,
		cloudv1.TagLabelPrefix + shared.Cluster:     shared.ClustersShared,
		cloudv1.TagLabelPrefix + shared.ServiceType: shared.Dedicated,
		cloudv1.TagLabelPrefix + shared.Project:     cluster.GetProjectID(),
		cloudv1.TagLabelPrefix + shared.Tenant:      cluster.GetTenantID(),
	})
	if !cluster.FeatureEnabled(consts.ClusterFeatureDisableNodePoolDataPlaneLabel) {
		labels[consts.LabelKeyNodeGroupType] = consts.LabelValNodeGroupTypeDataPlane
	}
	if nodeVersion != "" {
		labels[consts.K8sVersionLabelKey] = nodeVersion
	}
	if cluster.GetProviderType() == cloudv1.ProviderTypeGKE && cluster.FeatureEnabled(consts.ClusterFeatureGCP3LayerNodePool) {
		labels[consts.LabelKeyFor3LayerArch] = consts.LabelValTrue
	}
	return labels
}

func gkeHeterogeneousNodePoolConfigVersion(cluster *cloudv1.Cluster) string {
	if cluster == nil {
		return ""
	}
	clusterStatus, err := cloudv1.GKEClusterStatusFromCluster(cluster)
	if err != nil {
		return ""
	}
	return strings.TrimSpace(clusterStatus.NodeVersion)
}

func gkeHeterogeneousNodePoolDiskType(instanceType string) string {
	if strings.HasPrefix(instanceType, "n4-") || strings.HasPrefix(instanceType, "c4a-") {
		return gkeDiskTypeHyperdisk
	}
	return gkeDiskTypePdStandard
}

func gkeHeterogeneousNodePoolClusterRef(cluster *cloudv1.Cluster) string {
	return fmt.Sprintf("projects/%s/locations/%s/clusters/%s", cluster.Spec.Kubernetes.Account, gkeHeterogeneousClusterRegion(cluster), cluster.Spec.Kubernetes.Name)
}

func gkeHeterogeneousNetworkObjectMeta(cluster *cloudv1.Cluster) metav1.ObjectMeta {
	return metav1.ObjectMeta{
		Name:      cloudv1.ProviderTypeGKE + "-" + gkeHeterogeneousClusterRegion(cluster),
		Namespace: cluster.Namespace,
	}
}

func gkeHeterogeneousClusterRegion(cluster *cloudv1.Cluster) string {
	if cluster == nil {
		return ""
	}
	return strings.TrimSpace(cluster.GetRegion())
}