package common

import (
	"context"
	stderrors "errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	"github.com/pingcap/errors"
	gcpcontainer "google.golang.org/api/container/v1"
	"k8s.io/apimachinery/pkg/api/equality"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/labels"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/yaml"

	cloudv1 "github.com/tidbcloud/infra-api/api/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/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	nodepoolutil "github.com/tidbcloud/infra-provider/pkg/nodepool"
	"github.com/tidbcloud/infra-provider/util/collections"
)

const gkeComputeClassNodeMetadataRequeueAfter = 30 * time.Second

type desiredNodePoolRequest struct {
	nodePoolName  string
	instanceType  string
	componentType cloudv1.ComponentType
}

type gkeNodePoolComputeClassMetadata struct {
	desired string
	applied string
	// markEverBound is set only for API-managed heterogeneous nodepools after an
	// explicit heterogeneous intent references them. Idle GC uses it to avoid
	// deleting pure prepared nodepools that were created but never used.
	markEverBound bool
}

type malformedSharedGKEComputeClassIntentError struct {
	err error
}

func (e *malformedSharedGKEComputeClassIntentError) Error() string {
	return e.err.Error()
}

func (e *malformedSharedGKEComputeClassIntentError) Unwrap() error {
	return e.err
}

func (t *NodePool) mergeDesiredNodePool(current, desired *v1beta1.NodePool, req desiredNodePoolRequest) error {
	oldInstanceType := ""
	if current.Spec.ForProvider.Config != nil && current.Spec.ForProvider.Config.MachineType != nil {
		oldInstanceType = *current.Spec.ForProvider.Config.MachineType
	}
	if oldInstanceType != req.instanceType {
		return errors.Errorf("inconsistent nodepool instance type for nodepool %s, current component %s: %s vs %s", desired.Name, req.componentType, oldInstanceType, req.instanceType)
	}

	if !equality.Semantic.DeepEqual(current.Spec, desired.Spec) {
		return errors.Errorf("inconsistent nodepool spec for instance type %s, current component %s: %+v vs %+v", req.instanceType, req.componentType, current.Spec, desired.Spec)
	}
	return nil
}

func (t *NodePool) resolveDisableGKEHeterogeneousOPNodeSelector(comp cloudv1.ClusterComponentSpecWrapped, currentNodeSelector map[string]string) (map[string]string, string, bool, error) {
	if t.Cluster == nil || t.Cluster.GetProviderType() != cloudv1.ProviderTypeGKE {
		return nil, "", false, nil
	}

	rawWorkflow := strings.TrimSpace(t.Cluster.GetAnnotations()[consts.AnnoKeyScaleClusterWorkflow])
	if rawWorkflow == "" || !strings.Contains(rawWorkflow, consts.WfInputKeyScaleClusterDisableGKEHeterogeneous) {
		return nil, "", false, nil
	}

	workflow := &cloudv1.Workflow{}
	if err := yaml.Unmarshal([]byte(rawWorkflow), workflow); err != nil {
		return nil, "", false, fmt.Errorf("failed to unmarshal scale cluster workflow for disable gke heterogeneous fallback: %v", err)
	}
	if workflow.Spec.Type != "" && workflow.Spec.Type != consts.WorkflowTypeScaleCluster {
		return nil, "", false, nil
	}

	inputs := workflow.Spec.Inputs
	if !strings.EqualFold(strings.TrimSpace(inputs[consts.WfInputKeyScaleClusterDisableGKEHeterogeneous]), "true") {
		return nil, "", false, nil
	}
	if strings.TrimSpace(inputs[consts.WfInputKeyNamespace]) != strings.TrimSpace(t.Cluster.Namespace) ||
		strings.TrimSpace(inputs[consts.WfInputKeyClusterID]) != strings.TrimSpace(t.Cluster.Name) ||
		strings.TrimSpace(inputs[consts.WfInputKeyComponentType]) != string(comp.GetComponentType()) ||
		strings.TrimSpace(inputs[consts.WfInputKeyGroupID]) != strings.TrimSpace(comp.GetGroupId()) {
		return nil, "", false, nil
	}

	originalComp, err := newGKEHeterogeneousScaleWorkflowComponent(inputs[consts.WfInputKeyComponentType])
	if err != nil {
		return nil, "", false, err
	}
	if err := yaml.Unmarshal([]byte(inputs[consts.WfInputKeyScaleClusterOriginalComponentSpec]), originalComp); err != nil {
		return nil, "", false, fmt.Errorf("failed to unmarshal original component spec for disable gke heterogeneous fallback: %v", err)
	}

	nodeSelector := collections.DeepCopyMap(currentNodeSelector)
	delete(nodeSelector, consts.GKEComputeClassLabelKey)
	nodeSelector[consts.InstanceTypeLabelKey] = comp.GetInstanceType()
	return nodeSelector, originalComp.GetInstanceType(), true, nil
}

type gkeComputeClassNodePool struct {
	nodePoolName  string
	instanceType  string
	componentType cloudv1.ComponentType
	computeClass  string
	apiManaged    bool
}

type sharedGKEComputeClassNodePoolIntent struct {
	instanceType string
	computeClass string
}

type sharedGKEComputeClassPlan struct {
	byNodePool         map[string]sharedGKEComputeClassNodePoolIntent
	peerViewIncomplete bool
}

type gkeComputeClassMetadataReconcileGate struct {
	hasCurrentBinding         bool
	hasAPIManagedNodePool     bool
	hasSharedCleanupCandidate bool
	apiManagedNodePoolListErr bool
}

// explicitGKEComputeClassNodePoolsFromAnnotations is the single parse/normalize
// point for binding annotations used by reconcile and shared cleanup planning.
// It follows raw annotation presence so malformed configured intents still fail closed.
func (t *NodePool) explicitGKEComputeClassNodePoolsFromAnnotations() ([]gkeComputeClassNodePool, error) {
	return t.explicitGKEComputeClassNodePoolsFromAnnotationsForCluster(t.Cluster)
}

func (t *NodePool) explicitGKEComputeClassNodePoolsFromAnnotationsForCluster(cluster *cloudv1.Cluster) ([]gkeComputeClassNodePool, error) {
	out := []gkeComputeClassNodePool{}
	if cluster == nil {
		return out, nil
	}
	appendBindings := func(componentType cloudv1.ComponentType, bindingKey, baseInstanceType string) error {
		if strings.TrimSpace(cluster.GetAnnotations()[bindingKey]) == "" {
			return nil
		}
		annotation, explicit, err := cloudv1.ClusterGKEComputeClassBindingAnnotation(
			cluster,
			bindingKey,
			baseInstanceType,
			gkeheterogeneous.NodePoolName(cluster, baseInstanceType),
		)
		if err != nil {
			return err
		}
		if !explicit {
			return errors.Errorf("gke compute class binding annotation %s is required to explicitly specify all heterogeneous nodepool names", bindingKey)
		}
		for _, binding := range annotation.NodePools {
			if strings.TrimSpace(binding.NodePoolName) == "" {
				return errors.Errorf("gke compute class binding annotation %s contains an empty nodepool name", bindingKey)
			}
			out = append(out, gkeComputeClassNodePool{
				nodePoolName:  binding.NodePoolName,
				instanceType:  binding.InstanceType,
				componentType: componentType,
				computeClass:  annotation.ComputeClassName,
				apiManaged:    true,
			})
		}
		return nil
	}

	if err := appendBindings(cloudv1.ComponentTiKV, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, strings.TrimSpace(cluster.Spec.Cluster.TiKV.InstanceType)); err != nil {
		return nil, &malformedSharedGKEComputeClassIntentError{err: err}
	}
	defaultGroupID := strings.TrimSpace(cluster.GetDefaultGroupID())
	if err := appendBindings(cloudv1.ComponentTiDB, cluster.TiDBGKEComputeClassBindingsAnnoKey(defaultGroupID), cluster.TiDBInstanceTypeForGroup(defaultGroupID)); err != nil {
		return nil, &malformedSharedGKEComputeClassIntentError{err: err}
	}
	for _, extraGroup := range cluster.GetExtraGroups() {
		groupID := strings.TrimSpace(extraGroup.Name)
		if groupID == "" {
			continue
		}
		if err := appendBindings(cloudv1.ComponentTiDB, cluster.TiDBGKEComputeClassBindingsAnnoKey(groupID), cluster.TiDBInstanceTypeForGroup(groupID)); err != nil {
			return nil, &malformedSharedGKEComputeClassIntentError{err: err}
		}
	}
	return out, nil
}

// buildSharedGKEComputeClassPlan scans clusters that share the same persisted
// nodepool scope and folds their compute-class intents by nodepool name. The
// plan is used only by heterogeneous metadata cleanup and API-managed idle GC.
func (t *NodePool) buildSharedGKEComputeClassPlan(ctx context.Context) (sharedGKEComputeClassPlan, error) {
	plan := sharedGKEComputeClassPlan{byNodePool: map[string]sharedGKEComputeClassNodePoolIntent{}}
	if t.Cluster == nil || t.Cluster.GetProviderType() != cloudv1.ProviderTypeGKE || t.Local == nil {
		return plan, nil
	}

	currentBindings, err := t.explicitGKEComputeClassNodePoolsFromAnnotations()
	if err != nil {
		return sharedGKEComputeClassPlan{}, err
	}
	_, missingScopeLabels := nodepoolutil.GKEHeterogeneousNodePoolScopeLabels(t.Cluster)
	if len(missingScopeLabels) > 0 {
		if len(currentBindings) > 0 {
			return sharedGKEComputeClassPlan{}, errors.Errorf("cluster %s/%s is missing required GKE heterogeneous nodepool scope labels: %s", t.Cluster.Namespace, t.Cluster.Name, strings.Join(missingScopeLabels, ","))
		}
		plan.peerViewIncomplete = true
		return plan, nil
	}

	peers, err := t.listClustersInSharedNodePoolScope(ctx)
	if err != nil {
		return sharedGKEComputeClassPlan{}, err
	}

	currentSeen := false
	// A malformed peer cluster should not block the current cluster from reconciling
	// its own shared labels, but we must remember that the scope view is incomplete
	// so later cleanup cannot safely remove labels that a skipped peer may still need.
	for _, cluster := range peers {
		isCurrent := cluster.Namespace == t.Cluster.Namespace && cluster.Name == t.Cluster.Name
		if isCurrent {
			currentSeen = true
		}

		bindings := currentBindings
		if !isCurrent {
			bindings, err = t.explicitGKEComputeClassNodePoolsFromAnnotationsForCluster(cluster)
			if err != nil {
				var malformedErr *malformedSharedGKEComputeClassIntentError
				if !stderrors.As(err, &malformedErr) {
					return sharedGKEComputeClassPlan{}, err
				}
				plan.peerViewIncomplete = true
				continue
			}
		}
		if err := t.addGKEComputeClassNodePoolIntents(plan.byNodePool, bindings); err != nil {
			var malformedErr *malformedSharedGKEComputeClassIntentError
			if isCurrent || !stderrors.As(err, &malformedErr) {
				return sharedGKEComputeClassPlan{}, err
			}
			plan.peerViewIncomplete = true
		}
	}

	if !currentSeen {
		if err := t.addGKEComputeClassNodePoolIntents(plan.byNodePool, currentBindings); err != nil {
			return sharedGKEComputeClassPlan{}, err
		}
	}
	return plan, nil
}

func (t *NodePool) listClustersInSharedNodePoolScope(ctx context.Context) ([]*cloudv1.Cluster, error) {
	if t.Cluster == nil || t.Cluster.GetProviderType() != cloudv1.ProviderTypeGKE || t.Local == nil {
		return nil, nil
	}
	if _, err := t.requiredGKEHeterogeneousNodePoolScopeLabels(); err != nil {
		return nil, err
	}
	scopeLabels, err := t.requiredGKEKubernetesClusterScopeLabels()
	if err != nil {
		return nil, err
	}

	clusterList := &cloudv1.ClusterList{}
	if err := t.Local.List(ctx, clusterList, client.MatchingLabels(scopeLabels)); err != nil {
		return nil, errors.Wrap(err, "list clusters in same GKE heterogeneous nodepool scope for gke shared nodepool peers")
	}

	out := make([]*cloudv1.Cluster, 0, len(clusterList.Items))
	for i := range clusterList.Items {
		peer := &clusterList.Items[i]
		if !nodepoolutil.GKEClustersShareHeterogeneousNodePoolScope(t.Cluster, peer) {
			continue
		}
		out = append(out, peer)
	}
	return out, nil
}

func (t *NodePool) requiredGKEKubernetesClusterScopeLabels() (map[string]string, error) {
	if t.Cluster == nil {
		return nil, errors.Errorf("cluster is nil")
	}
	scopeLabels, missingScopeLabels := nodepoolutil.GKEKubernetesClusterScopeLabels(t.Cluster)
	if len(missingScopeLabels) > 0 {
		return nil, errors.Errorf("cluster %s/%s is missing required GKE Kubernetes cluster scope labels: %s", t.Cluster.Namespace, t.Cluster.Name, strings.Join(missingScopeLabels, ","))
	}
	return scopeLabels, nil
}

func (t *NodePool) requiredGKEHeterogeneousNodePoolScopeLabels() (map[string]string, error) {
	if t.Cluster == nil {
		return nil, errors.Errorf("cluster is nil")
	}
	scopeLabels, missingScopeLabels := nodepoolutil.GKEHeterogeneousNodePoolScopeLabels(t.Cluster)
	if len(missingScopeLabels) > 0 {
		return nil, errors.Errorf("cluster %s/%s is missing required GKE heterogeneous nodepool scope labels: %s", t.Cluster.Namespace, t.Cluster.Name, strings.Join(missingScopeLabels, ","))
	}
	return scopeLabels, nil
}

func (t *NodePool) addGKEComputeClassNodePoolIntents(out map[string]sharedGKEComputeClassNodePoolIntent, bindings []gkeComputeClassNodePool) error {
	// Shared compute-class retention follows the surviving heterogeneous binding intent.
	// Pause and replicas=0 only stop runtime nodepool generation; they do not revoke the binding.
	for _, binding := range bindings {
		nodePoolName := binding.nodePoolName
		instanceType := binding.instanceType
		current, ok := out[nodePoolName]
		if ok && current.computeClass != binding.computeClass {
			return errors.Errorf("gke shared nodepool %s has conflicting compute class %s and %s for component %s", nodePoolName, current.computeClass, binding.computeClass, binding.componentType)
		}
		if ok && current.instanceType != instanceType {
			return errors.Errorf("gke shared nodepool %s has conflicting instance type %s and %s for component %s", nodePoolName, current.instanceType, instanceType, binding.componentType)
		}
		if ok {
			continue
		}
		current = sharedGKEComputeClassNodePoolIntent{
			instanceType: instanceType,
			computeClass: binding.computeClass,
		}
		out[nodePoolName] = current
	}
	return nil
}

func (t *NodePool) refreshExistingNodePoolResourceLabels(ctx context.Context, obj *v1beta1.NodePool) error {
	if obj.Status.AtProvider.Status != v1beta1.NodePoolStateRunning {
		return fmt.Errorf("non-active nodepool %s has the state %q", obj.Name, obj.Status.AtProvider.Status)
	}
	if _, ok := obj.Annotations[annoKeyResourceLabels]; ok {
		return nil
	}
	ls, err := t.ensureGKENodePoolResourceLabels(ctx, obj)
	if err != nil {
		return err
	}
	if len(ls) == 0 {
		return nil
	}

	updated := obj.DeepCopy()
	if updated.Annotations == nil {
		updated.Annotations = map[string]string{}
	}
	updated.Annotations[annoKeyResourceLabels] = resourceLabelsAnnotationValue(ls)
	if err := t.Local.Patch(ctx, updated, client.MergeFrom(obj)); err != nil {
		return fmt.Errorf("record resource labels of node pool %s in anno failed: %w", obj.Name, err)
	}
	if err := t.Local.Get(ctx, client.ObjectKey{Name: obj.Name}, obj); err != nil {
		return fmt.Errorf("get node pool %s after recording resource labels: %w", obj.Name, err)
	}
	return nil
}

func (t *NodePool) ensureGKENodePoolResourceLabels(ctx context.Context, obj *v1beta1.NodePool) (map[string]string, error) {
	// v1beta1.NodeConfig does not expose resourceLabels, so existing GKE
	// NodePools need a direct GCP API update before we cache the value locally.
	nodePoolName, err := t.ParseNameFromSelfLink(obj.Status.AtProvider.SelfLink)
	if err != nil {
		return nil, fmt.Errorf("cannot parse self link of nodepool %s", obj.Name)
	}
	ls, err := t.gcpCli.GetNodePoolResourceLabels(ctx, nodePoolName)
	if err != nil {
		return nil, fmt.Errorf("failed get node pool %s resource labels: %w", obj.Name, err)
	}
	if ls == nil {
		ls = make(map[string]string)
	}
	labelsChanged, err := mergeMissingResourceLabels(ls, t.resourceLabels(), obj.Name)
	if err != nil {
		return nil, err
	}
	if !labelsChanged {
		return ls, nil
	}
	op, err := t.gcpCli.UpdateNodePoolResourceLabels(ctx, nodePoolName, ls)
	if err != nil {
		return nil, fmt.Errorf("failed update node pool %s resource labels: %w", obj.Name, err)
	}
	if op != nil {
		return nil, fmt.Errorf("set resource labels for nodepool %s", obj.Name)
	}
	return ls, nil
}

func mergeMissingResourceLabels(actual, expected map[string]string, nodePoolName string) (bool, error) {
	changed := false
	for k, v := range expected {
		if _, ok := actual[k]; !ok {
			actual[k] = v
			changed = true
			continue
		}
		if actual[k] != v {
			return false, fmt.Errorf("resource label %s of node pool %s is already set to %s, but expected %s", k, nodePoolName, actual[k], v)
		}
	}
	return changed, nil
}

func resourceLabelsAnnotationValue(ls map[string]string) string {
	var kvs []string
	for k, v := range ls {
		kvs = append(kvs, k+":"+v)
	}
	sort.Strings(kvs)
	return strings.Join(kvs, ",")
}

func (t *NodePool) reconcileGKEComputeClassNodePoolMetadata(ctx context.Context) (bool, error) {
	gate, err := t.gkeComputeClassMetadataReconcileGate(ctx)
	if err != nil {
		if gate.apiManagedNodePoolListErr {
			return false, t.gkeComputeClassMetadataCleanupError(err, gate.hasCurrentBinding)
		}
		return false, err
	}
	if !gate.hasCurrentBinding && !gate.hasAPIManagedNodePool {
		return false, nil
	}

	// Keep the order explicit: first project current cluster bindings, then
	// backfill prepared API-managed nodepools, cleanup stale heterogeneous
	// metadata after all expected owners have had a chance to refresh state,
	// then run idle GC. Cleanup and GC share the same plan so one reconcile
	// round uses one consistent view of peer intent.
	if gate.hasCurrentBinding {
		if err := t.reconcileCurrentGKEComputeClassBindingNodePoolMetadata(ctx); err != nil {
			return false, err
		}
	}
	if gate.hasAPIManagedNodePool {
		if err := t.refreshUnboundPreparedNodePoolResourceLabels(ctx); err != nil {
			return false, t.gkeComputeClassMetadataCleanupError(err, gate.hasCurrentBinding)
		}
	}
	needsSharedCleanup := gate.hasCurrentBinding || gate.hasSharedCleanupCandidate
	if !needsSharedCleanup {
		return false, nil
	}
	plan, err := t.buildSharedGKEComputeClassPlan(ctx)
	if err != nil {
		return false, t.gkeComputeClassMetadataCleanupError(err, gate.hasCurrentBinding)
	}
	if err := t.cleanupStaleSharedGKEComputeClassNodePoolMetadataWithPlan(ctx, plan); err != nil {
		return false, t.gkeComputeClassMetadataCleanupError(err, gate.hasCurrentBinding)
	}
	deleted, err := t.cleanupUnusedAPIManagedGKEHeterogeneousNodePoolsWithPlan(ctx, plan)
	if err != nil && !gate.hasCurrentBinding {
		return false, t.gkeComputeClassMetadataCleanupError(err, gate.hasCurrentBinding)
	}
	return deleted, err
}

func (t *NodePool) gkeComputeClassMetadataReconcileGate(ctx context.Context) (gkeComputeClassMetadataReconcileGate, error) {
	gate := gkeComputeClassMetadataReconcileGate{}
	if t.Cluster == nil || t.Cluster.GetProviderType() != cloudv1.ProviderTypeGKE || t.Local == nil {
		return gate, nil
	}

	currentBindings, err := t.explicitGKEComputeClassNodePoolsFromAnnotations()
	if err != nil {
		return gate, err
	}
	gate.hasCurrentBinding = len(currentBindings) > 0

	selector, ok := t.apiManagedGKEHeterogeneousNodePoolLabelSelectorIfScoped()
	if !ok {
		return gate, nil
	}
	nodePools := &v1beta1.NodePoolList{}
	if err := t.Local.List(ctx, nodePools, selector); err != nil {
		gate.apiManagedNodePoolListErr = true
		return gate, err
	}
	gate.hasAPIManagedNodePool = len(nodePools.Items) > 0
	for i := range nodePools.Items {
		nodePool := &nodePools.Items[i]
		if nodepoolutil.HasGKEComputeClassMetadataForNodePool(nodePool) || gkeHeterogeneousNodePoolEverBound(nodePool) {
			gate.hasSharedCleanupCandidate = true
			break
		}
	}
	return gate, nil
}

// reconcileCurrentGKEComputeClassBindingNodePoolMetadata only patches nodepools
// explicitly bound by the current cluster's heterogeneous intent. It does not
// create missing nodepools; all bindings are API-managed heterogeneous NodePools
// and must already exist before they can receive ComputeClass metadata.
func (t *NodePool) reconcileCurrentGKEComputeClassBindingNodePoolMetadata(ctx context.Context) error {
	if t.Cluster == nil || t.Cluster.GetProviderType() != cloudv1.ProviderTypeGKE || t.Local == nil {
		return nil
	}

	boundNodePools, err := t.explicitGKEComputeClassNodePoolsFromAnnotations()
	if err != nil {
		return err
	}
	for _, expected := range boundNodePools {
		if strings.TrimSpace(expected.nodePoolName) == "" {
			continue
		}
		nodePool := &v1beta1.NodePool{}
		if err := t.Local.Get(ctx, client.ObjectKey{Name: expected.nodePoolName}, nodePool); err != nil {
			if apierrors.IsNotFound(err) {
				return newGKEComputeClassNodeMetadataRequeueError(errors.Errorf(
					"gke compute class binding nodepool %s for component %s instance type %s compute class %s is missing",
					expected.nodePoolName,
					expected.componentType,
					expected.instanceType,
					expected.computeClass,
				))
			}
			return err
		}
		if expected.apiManaged {
			if err := validateAPIManagedGKEHeterogeneousNodePoolPatchTarget(nodePool); err != nil {
				return err
			}
		}
		if err := t.validateGKEComputeClassNodePoolPatchTarget(nodePool, expected.nodePoolName, expected.instanceType, t.GetGCPClusterRef()); err != nil {
			return err
		}
		if err := t.patchGKENodePoolComputeClassMetadata(ctx, nodePool, gkeNodePoolComputeClassMetadata{
			desired:       expected.computeClass,
			markEverBound: expected.apiManaged,
		}); err != nil {
			return errors.Wrapf(err, "patch desired compute class for bound nodepool %s", expected.nodePoolName)
		}
		if expected.apiManaged {
			// API-managed heterogeneous nodepools bypass desiredNodePools(), so
			// they also need resource-label backfill here. This runs after desired
			// metadata is recorded, because resource-label backfill requires
			// external readiness.
			if err := t.refreshExistingNodePoolResourceLabels(ctx, nodePool); err != nil {
				return err
			}
		}
		synced, submitted, err := t.ensureGKENodePoolComputeClassNodeMetadataWithRequeue(ctx, nodePool, expected.computeClass)
		if err != nil {
			return err
		}
		if submitted {
			return newGKEComputeClassNodeMetadataRequeueError(errors.Errorf("set compute class node metadata for nodepool %s", nodePool.Name))
		}
		if !synced {
			return errors.Errorf("gke compute class nodepool %s is not ready to sync node metadata", nodePool.Name)
		}
		if err := t.patchGKENodePoolComputeClassMetadata(ctx, nodePool, gkeNodePoolComputeClassMetadata{desired: expected.computeClass, applied: expected.computeClass}); err != nil {
			return errors.Wrapf(err, "patch compute class for bound nodepool %s", expected.nodePoolName)
		}
	}
	return nil
}

// refreshUnboundPreparedNodePoolResourceLabels handles API-managed heterogeneous
// nodepools that were created before any deploy intent references them.
func (t *NodePool) refreshUnboundPreparedNodePoolResourceLabels(ctx context.Context) error {
	if t.Cluster == nil || t.Cluster.GetProviderType() != cloudv1.ProviderTypeGKE || t.Local == nil {
		return nil
	}
	labelSelector, err := t.apiManagedGKEHeterogeneousNodePoolLabelSelector()
	if err != nil {
		return err
	}

	nodePools := &v1beta1.NodePoolList{}
	if err := t.Local.List(ctx, nodePools, labelSelector); err != nil {
		return err
	}

	for i := range nodePools.Items {
		nodePool := &nodePools.Items[i]
		if strings.TrimSpace(nodepoolutil.GKEComputeClassNameForNodePool(nodePool)) != "" {
			continue
		}
		if err := t.validateGKEComputeClassNodePoolPatchTarget(nodePool, nodePool.Name, "", t.GetGCPClusterRef()); err != nil {
			continue
		}
		if err := t.patchPreparedAPIManagedGKEHeterogeneousIsolationTaint(ctx, nodePool); err != nil {
			return err
		}
		if nodePool.Status.AtProvider.Status != v1beta1.NodePoolStateRunning {
			continue
		}
		if err := t.refreshExistingNodePoolResourceLabels(ctx, nodePool); err != nil {
			return err
		}
	}
	return nil
}

// cleanupStaleSharedGKEComputeClassNodePoolMetadata removes stale compute-class
// metadata from API-managed heterogeneous nodepools only when no cluster in the
// shared scope still references the nodepool. Ordinary shared nodepools are not
// part of this cleanup path.
func (t *NodePool) cleanupStaleSharedGKEComputeClassNodePoolMetadata(ctx context.Context) error {
	if t.Cluster == nil || t.Cluster.GetProviderType() != cloudv1.ProviderTypeGKE {
		return nil
	}
	plan, err := t.buildSharedGKEComputeClassPlan(ctx)
	if err != nil {
		return err
	}
	return t.cleanupStaleSharedGKEComputeClassNodePoolMetadataWithPlan(ctx, plan)
}

func (t *NodePool) cleanupStaleSharedGKEComputeClassNodePoolMetadataWithPlan(ctx context.Context, plan sharedGKEComputeClassPlan) error {
	if t.Cluster == nil || t.Cluster.GetProviderType() != cloudv1.ProviderTypeGKE || t.Local == nil {
		return nil
	}
	if plan.peerViewIncomplete {
		// A skipped peer may still reference the nodepool, so cleanup must wait
		// until the shared-scope intent view is complete again.
		return nil
	}
	labelSelector, err := t.apiManagedGKEHeterogeneousNodePoolLabelSelector()
	if err != nil {
		return err
	}

	nodePools := &v1beta1.NodePoolList{}
	if err := t.Local.List(ctx, nodePools, labelSelector); err != nil {
		return err
	}
	for i := range nodePools.Items {
		nodePool := &nodePools.Items[i]
		if _, ok := plan.byNodePool[nodePool.Name]; ok {
			continue
		}
		if err := t.validateSharedGKENodePoolComputeClassCleanupTarget(nodePool, t.GetGCPClusterRef()); err != nil {
			continue
		}
		if err := t.cleanupGKENodePoolComputeClassMetadata(ctx, nodePool); err != nil {
			return err
		}
	}

	return nil
}

func (t *NodePool) validateSharedGKENodePoolComputeClassCleanupTarget(nodePool *v1beta1.NodePool, expectedGCPClusterRef string) error {
	// Cleanup only verifies ownership scope. The original binding may already be
	// gone, so there is no authoritative instance type to compare against here.
	return t.validateGKEComputeClassNodePoolPatchTarget(nodePool, nodePool.Name, "", expectedGCPClusterRef)
}

func (t *NodePool) cleanupGKENodePoolComputeClassMetadata(ctx context.Context, nodePool *v1beta1.NodePool) error {
	if !nodepoolutil.HasGKEComputeClassMetadataForNodePool(nodePool) {
		return nil
	}
	computeClassGuard := gkeNodePoolComputeClassCleanupGuard(nodePool)
	if err := t.patchGKENodePoolComputeClassMetadata(ctx, nodePool, gkeNodePoolComputeClassMetadata{applied: computeClassGuard}); err != nil {
		return errors.Wrapf(err, "cleanup stale compute class for nodepool %s", nodePool.Name)
	}
	synced, submitted, err := t.ensureGKENodePoolComputeClassNodeMetadataWithRequeue(ctx, nodePool, "")
	if err != nil {
		return err
	}
	if submitted {
		return newGKEComputeClassNodeMetadataRequeueError(errors.Errorf("remove compute class node metadata for nodepool %s", nodePool.Name))
	}
	if !synced {
		return errors.Errorf("gke compute class nodepool %s is not ready to sync node metadata", nodePool.Name)
	}
	if err := t.patchGKENodePoolComputeClassMetadata(ctx, nodePool, gkeNodePoolComputeClassMetadata{}); err != nil {
		return errors.Wrapf(err, "clear applied compute class for nodepool %s", nodePool.Name)
	}
	_ = gkeheterogeneous.RecordComputeClassCleanupCandidate(t.Cluster, computeClassGuard)
	return nil
}

// cleanupUnusedAPIManagedGKEHeterogeneousNodePools deletes only API-managed
// heterogeneous nodepools that were previously referenced by a heterogeneous
// compute-class intent. Ordinary shared nodepools and pure prepared pools are
// never deleted here.
func (t *NodePool) cleanupUnusedAPIManagedGKEHeterogeneousNodePools(ctx context.Context) (bool, error) {
	if t.Cluster == nil || t.Cluster.GetProviderType() != cloudv1.ProviderTypeGKE || t.Local == nil || t.gcpCli == nil {
		return false, nil
	}
	plan, err := t.buildSharedGKEComputeClassPlan(ctx)
	if err != nil {
		return false, err
	}
	return t.cleanupUnusedAPIManagedGKEHeterogeneousNodePoolsWithPlan(ctx, plan)
}

func (t *NodePool) cleanupUnusedAPIManagedGKEHeterogeneousNodePoolsWithPlan(ctx context.Context, plan sharedGKEComputeClassPlan) (bool, error) {
	if t.Cluster == nil || t.Cluster.GetProviderType() != cloudv1.ProviderTypeGKE || t.Local == nil || t.gcpCli == nil {
		return false, nil
	}
	if plan.peerViewIncomplete {
		// Deleting an API-managed extra pool requires a complete shared-scope
		// view; otherwise a malformed or temporarily invisible peer could still
		// depend on it.
		return false, nil
	}
	labelSelector, err := t.apiManagedGKEHeterogeneousNodePoolLabelSelector()
	if err != nil {
		return false, err
	}

	nodePools := &v1beta1.NodePoolList{}
	if err := t.Local.List(ctx, nodePools, labelSelector); err != nil {
		return false, err
	}
	for i := range nodePools.Items {
		nodePool := &nodePools.Items[i]
		if _, ok := plan.byNodePool[nodePool.Name]; ok {
			continue
		}
		if err := t.validateSharedGKENodePoolComputeClassCleanupTarget(nodePool, t.GetGCPClusterRef()); err != nil {
			continue
		}
		if !gkeHeterogeneousNodePoolEverBound(nodePool) {
			// Pure prepared pools are intentionally stable resources. Only pools
			// that were once bound by compute-class intent can be reclaimed.
			continue
		}
		if nodepoolutil.HasGKEComputeClassMetadataForNodePool(nodePool) {
			// Metadata cleanup is a separate, safer phase because it first clears
			// the external GKE node metadata and confirms the applied annotation.
			continue
		}
		if len(nodePool.Status.AtProvider.InstanceGroupUrls) == 0 {
			// Instance groups are the source for actual node count. Without them
			// we cannot distinguish "empty" from "not fully observed yet".
			continue
		}
		synced, submitted, err := t.ensureGKENodePoolComputeClassNodeMetadataWithRequeue(ctx, nodePool, "")
		if err != nil {
			return false, err
		}
		if submitted {
			return false, newGKEComputeClassNodeMetadataRequeueError(errors.Errorf("remove compute class node metadata for nodepool %s", nodePool.Name))
		}
		if !synced {
			// GKE has not yet reported the node metadata as clean. Try again after a
			// later provider/GKE observation instead of deleting immediately.
			continue
		}
		// Delete only after both GCP views agree the nodepool is empty.
		managedNodeCount, err := nodepoolutil.GetManagedInstanceCount(ctx, t.gcpCli, nodePool.Status.AtProvider.InstanceGroupUrls)
		if err != nil {
			return false, errors.Wrapf(err, "get managed node count for nodepool %s", nodePool.Name)
		}
		if managedNodeCount > 0 {
			// Autoscaling min/max is not authoritative here; delete only after
			// the backing instance groups report no nodes.
			continue
		}
		instanceGroupNodeCount, err := nodepoolutil.GetNodeCount(ctx, t.gcpCli, nodePool.Status.AtProvider.InstanceGroupUrls)
		if err != nil {
			return false, errors.Wrapf(err, "get instance group node count for nodepool %s", nodePool.Name)
		}
		if instanceGroupNodeCount > 0 {
			continue
		}
		if err := client.IgnoreNotFound(t.Local.Delete(ctx, nodePool)); err != nil {
			return false, err
		}
		return true, nil
	}
	return false, nil
}

func newGKEComputeClassNodeMetadataRequeueError(err error) error {
	return k8s.NewRetriableError(err, ctrl.Result{Requeue: true, RequeueAfter: gkeComputeClassNodeMetadataRequeueAfter})
}

func (t *NodePool) gkeComputeClassMetadataCleanupError(err error, active bool) error {
	if err == nil {
		return nil
	}
	if active {
		return err
	}
	t.Logger.Error(err, "retry inactive gke compute class nodepool metadata cleanup")
	return k8s.NewRetriableError(k8s.RequeueOnlyError, ctrl.Result{Requeue: true, RequeueAfter: gkeComputeClassNodeMetadataRequeueAfter})
}

// ensureGKENodePoolComputeClassNodeMetadataWithRequeue is used only by
// heterogeneous compute-class paths. It returns synced only after GKE GET
// confirms the desired node label and taint state, and asks the controller to
// recheck shortly after an update is submitted or fails.
func (t *NodePool) ensureGKENodePoolComputeClassNodeMetadataWithRequeue(ctx context.Context, nodePool *v1beta1.NodePool, computeClass string) (synced bool, operationSubmitted bool, _ error) {
	if nodePool == nil {
		return false, false, errors.Errorf("gke compute class nodepool is nil")
	}
	if nodePool.Status.AtProvider.Status != v1beta1.NodePoolStateRunning {
		return false, false, nil
	}
	if strings.TrimSpace(nodePool.Status.AtProvider.SelfLink) == "" {
		return false, false, nil
	}
	if t.gcpCli == nil {
		return false, false, errors.Errorf("gcp client is nil")
	}

	nodePoolName, err := t.ParseNameFromSelfLink(nodePool.Status.AtProvider.SelfLink)
	if err != nil {
		return false, false, errors.Wrapf(err, "cannot parse self link of nodepool %s", nodePool.Name)
	}
	labels, err := t.gcpCli.GetNodePoolNodeLabels(ctx, nodePoolName)
	if err != nil {
		return false, false, errors.Wrapf(err, "failed get node labels of nodepool %s", nodePool.Name)
	}
	if labels == nil {
		labels = map[string]string{}
	}
	taints, err := t.gcpCli.GetNodePoolNodeTaints(ctx, nodePoolName)
	if err != nil {
		return false, false, errors.Wrapf(err, "failed get node taints of nodepool %s", nodePool.Name)
	}

	computeClass = strings.TrimSpace(computeClass)
	apiManaged := gkeheterogeneous.IsAPIManagedNodePool(nodePool)
	if computeClass != "" {
		computeClassLabelSynced := strings.TrimSpace(labels[consts.GKEComputeClassLabelKey]) == computeClass
		computeClassTaintSynced := hasGKENodePoolTaint(taints, consts.GKEComputeClassLabelKey, computeClass)
		if !computeClassLabelSynced || !computeClassTaintSynced {
			nextLabels := cloneGKENodePoolLabels(labels)
			nextLabels[consts.GKEComputeClassLabelKey] = computeClass
			nextTaints := setGKENodePoolTaint(taints, consts.GKEComputeClassLabelKey, computeClass)
			if apiManaged {
				nextTaints = setGKENodePoolTaint(nextTaints, consts.GKEHeterogeneousNodePoolIsolationTaintKey, consts.GKEHeterogeneousNodePoolIsolationTaintValue)
			}
			if !gkeNodePoolLabelsEqual(labels, nextLabels) || !gkeNodePoolTaintsEqual(taints, nextTaints) {
				if _, err = t.gcpCli.UpdateNodePoolNodeLabelsAndTaints(ctx, nodePoolName, nextLabels, nextTaints); err != nil {
					return false, false, newGKEComputeClassNodeMetadataRequeueError(errors.Wrapf(err, "failed update compute class node metadata for nodepool %s", nodePool.Name))
				}
				return false, true, nil
			}
		}
		if apiManaged && hasGKENodePoolTaint(taints, consts.GKEHeterogeneousNodePoolIsolationTaintKey, consts.GKEHeterogeneousNodePoolIsolationTaintValue) {
			nextTaints := clearGKENodePoolTaint(taints, consts.GKEHeterogeneousNodePoolIsolationTaintKey)
			if _, err = t.gcpCli.UpdateNodePoolNodeTaints(ctx, nodePoolName, nextTaints); err != nil {
				return false, false, newGKEComputeClassNodeMetadataRequeueError(errors.Wrapf(err, "failed remove heterogeneous isolation taint for nodepool %s", nodePool.Name))
			}
			return false, true, nil
		}
		return true, false, nil
	}

	computeClassLabelPresent := strings.TrimSpace(labels[consts.GKEComputeClassLabelKey]) != ""
	computeClassTaintPresent := hasGKENodePoolTaintKey(taints, consts.GKEComputeClassLabelKey)
	isolationTaintPresent := hasGKENodePoolTaint(taints, consts.GKEHeterogeneousNodePoolIsolationTaintKey, consts.GKEHeterogeneousNodePoolIsolationTaintValue)
	if apiManaged && (computeClassLabelPresent || computeClassTaintPresent) && !isolationTaintPresent {
		nextTaints := setGKENodePoolTaint(taints, consts.GKEHeterogeneousNodePoolIsolationTaintKey, consts.GKEHeterogeneousNodePoolIsolationTaintValue)
		if _, err = t.gcpCli.UpdateNodePoolNodeTaints(ctx, nodePoolName, nextTaints); err != nil {
			return false, false, newGKEComputeClassNodeMetadataRequeueError(errors.Wrapf(err, "failed restore heterogeneous isolation taint for nodepool %s", nodePool.Name))
		}
		return false, true, nil
	}
	if computeClassLabelPresent || computeClassTaintPresent {
		nextLabels := cloneGKENodePoolLabels(labels)
		delete(nextLabels, consts.GKEComputeClassLabelKey)
		nextTaints := clearGKENodePoolTaint(taints, consts.GKEComputeClassLabelKey)
		if apiManaged {
			nextTaints = setGKENodePoolTaint(nextTaints, consts.GKEHeterogeneousNodePoolIsolationTaintKey, consts.GKEHeterogeneousNodePoolIsolationTaintValue)
		}
		if !gkeNodePoolLabelsEqual(labels, nextLabels) || !gkeNodePoolTaintsEqual(taints, nextTaints) {
			if _, err = t.gcpCli.UpdateNodePoolNodeLabelsAndTaints(ctx, nodePoolName, nextLabels, nextTaints); err != nil {
				return false, false, newGKEComputeClassNodeMetadataRequeueError(errors.Wrapf(err, "failed remove compute class node metadata for nodepool %s", nodePool.Name))
			}
			return false, true, nil
		}
	}
	if apiManaged && !isolationTaintPresent {
		nextTaints := setGKENodePoolTaint(taints, consts.GKEHeterogeneousNodePoolIsolationTaintKey, consts.GKEHeterogeneousNodePoolIsolationTaintValue)
		if _, err = t.gcpCli.UpdateNodePoolNodeTaints(ctx, nodePoolName, nextTaints); err != nil {
			return false, false, newGKEComputeClassNodeMetadataRequeueError(errors.Wrapf(err, "failed restore heterogeneous isolation taint for nodepool %s", nodePool.Name))
		}
		return false, true, nil
	}
	return true, false, nil
}

func cloneGKENodePoolLabels(labels map[string]string) map[string]string {
	out := make(map[string]string, len(labels))
	for key, value := range labels {
		out[key] = value
	}
	return out
}

func gkeNodePoolLabelsEqual(a, b map[string]string) bool {
	if len(a) != len(b) {
		return false
	}
	for key, value := range a {
		if b[key] != value {
			return false
		}
	}
	return true
}

func gkeNodePoolTaintsEqual(a, b []*gcpcontainer.NodeTaint) bool {
	left := canonicalGKENodePoolTaints(a)
	right := canonicalGKENodePoolTaints(b)
	if len(left) != len(right) {
		return false
	}
	for i := range left {
		if left[i] != right[i] {
			return false
		}
	}
	return true
}

func canonicalGKENodePoolTaints(taints []*gcpcontainer.NodeTaint) []string {
	out := make([]string, 0, len(taints))
	for _, taint := range taints {
		if taint == nil {
			continue
		}
		out = append(out, strings.TrimSpace(taint.Key)+"\x00"+strings.TrimSpace(taint.Value)+"\x00"+strings.TrimSpace(taint.Effect))
	}
	sort.Strings(out)
	return out
}

func hasGKENodePoolTaint(taints []*gcpcontainer.NodeTaint, key, value string) bool {
	key = strings.TrimSpace(key)
	value = strings.TrimSpace(value)
	if key == "" || value == "" {
		return false
	}
	for _, taint := range taints {
		if taint == nil {
			continue
		}
		if strings.TrimSpace(taint.Key) == key &&
			strings.TrimSpace(taint.Value) == value &&
			strings.TrimSpace(taint.Effect) == "NO_SCHEDULE" {
			return true
		}
	}
	return false
}

func hasGKENodePoolTaintKey(taints []*gcpcontainer.NodeTaint, key string) bool {
	key = strings.TrimSpace(key)
	if key == "" {
		return false
	}
	for _, taint := range taints {
		if taint != nil && strings.TrimSpace(taint.Key) == key {
			return true
		}
	}
	return false
}

func setGKENodePoolTaint(taints []*gcpcontainer.NodeTaint, key, value string) []*gcpcontainer.NodeTaint {
	key = strings.TrimSpace(key)
	value = strings.TrimSpace(value)
	out := clearGKENodePoolTaint(taints, key)
	if key == "" || value == "" {
		return out
	}
	return append(out, &gcpcontainer.NodeTaint{
		Key:    key,
		Value:  value,
		Effect: "NO_SCHEDULE",
	})
}

func clearGKENodePoolTaint(taints []*gcpcontainer.NodeTaint, key string) []*gcpcontainer.NodeTaint {
	key = strings.TrimSpace(key)
	out := make([]*gcpcontainer.NodeTaint, 0, len(taints))
	for _, taint := range taints {
		if taint == nil || strings.TrimSpace(taint.Key) == key {
			continue
		}
		copied := *taint
		out = append(out, &copied)
	}
	return out
}

func (t *NodePool) apiManagedGKEHeterogeneousNodePoolLabelSelector() (client.MatchingLabelsSelector, error) {
	if t.Cluster == nil {
		return client.MatchingLabelsSelector{}, errors.Errorf("cluster is nil")
	}
	scopeLabels, err := t.requiredGKEHeterogeneousNodePoolScopeLabels()
	if err != nil {
		return client.MatchingLabelsSelector{}, err
	}

	ls := map[string]string{
		namer.ComponentSelectorKey:                       shared.ComponentShared,
		namer.ClusterSelectorKey:                         shared.ClustersShared,
		consts.GKEHeterogeneousNodePoolManagedByLabelKey: consts.GKEHeterogeneousNodePoolManagedByLabelValue,
	}
	for key, value := range scopeLabels {
		ls[key] = value
	}

	return client.MatchingLabelsSelector{
		Selector: labels.SelectorFromSet(ls),
	}, nil
}

func (t *NodePool) apiManagedGKEHeterogeneousNodePoolLabelSelectorIfScoped() (client.MatchingLabelsSelector, bool) {
	if t.Cluster == nil {
		return client.MatchingLabelsSelector{}, false
	}
	scopeLabels, missingScopeLabels := nodepoolutil.GKEHeterogeneousNodePoolScopeLabels(t.Cluster)
	if len(missingScopeLabels) > 0 {
		return client.MatchingLabelsSelector{}, false
	}

	ls := map[string]string{
		namer.ComponentSelectorKey:                       shared.ComponentShared,
		namer.ClusterSelectorKey:                         shared.ClustersShared,
		consts.GKEHeterogeneousNodePoolManagedByLabelKey: consts.GKEHeterogeneousNodePoolManagedByLabelValue,
	}
	for key, value := range scopeLabels {
		ls[key] = value
	}

	return client.MatchingLabelsSelector{
		Selector: labels.SelectorFromSet(ls),
	}, true
}

func validateAPIManagedGKEHeterogeneousNodePoolPatchTarget(nodePool *v1beta1.NodePool) error {
	if nodePool == nil {
		return errors.Errorf("gke compute class nodepool is nil")
	}
	if gkeheterogeneous.IsAPIManagedNodePool(nodePool) {
		return nil
	}
	return errors.Errorf("nodepool %s is not managed by heterogeneous nodepool API", nodePool.Name)
}

func (t *NodePool) validateGKEComputeClassNodePoolPatchTarget(nodePool *v1beta1.NodePool, nodePoolName, expectedInstanceType, expectedGCPClusterRef string) error {
	if nodePool == nil {
		return errors.Errorf("gke compute class nodepool %s is nil", nodePoolName)
	}
	if t.Cluster == nil {
		return errors.Errorf("cluster is nil")
	}
	nodePoolName = strings.TrimSpace(nodePoolName)
	if nodePoolName != "" && nodePool.Name != nodePoolName {
		return errors.Errorf("gke compute class nodepool %s does not match expected nodepool %s", nodePool.Name, nodePoolName)
	}
	labels := nodePool.GetLabels()
	if labels[namer.ComponentSelectorKey] != shared.ComponentShared || labels[namer.ClusterSelectorKey] != shared.ClustersShared {
		return errors.Errorf("nodepool %s is not a shared GKE nodepool", nodePool.Name)
	}
	if err := t.validateGKENodePoolKubernetesClusterScope(nodePool); err != nil {
		return err
	}
	if err := t.validateGKENodePoolGCPClusterRef(nodePool, expectedGCPClusterRef); err != nil {
		return err
	}
	expectedInstanceType = strings.TrimSpace(expectedInstanceType)
	if expectedInstanceType == "" {
		return nil
	}
	actualInstanceType, ok := gkeheterogeneous.NodePoolMachineType(nodePool)
	if !ok {
		return errors.Errorf("gke compute class nodepool %s does not have a machine type configured", nodePool.Name)
	}
	if actualInstanceType != expectedInstanceType {
		return errors.Errorf("gke compute class nodepool %s has machine type %s, want %s", nodePool.Name, actualInstanceType, expectedInstanceType)
	}
	return nil
}

func (t *NodePool) validateGKENodePoolKubernetesClusterScope(nodePool *v1beta1.NodePool) error {
	scopeLabels, err := t.requiredGKEHeterogeneousNodePoolScopeLabels()
	if err != nil {
		return err
	}
	return gkeheterogeneous.ValidateNodePoolScopeLabels(nodePool, t.Cluster.Name, "GKE heterogeneous nodepool scope", scopeLabels)
}

func (t *NodePool) validateGKENodePoolGCPClusterRef(nodePool *v1beta1.NodePool, expectedGCPClusterRef string) error {
	if nodePool == nil {
		return errors.Errorf("gke nodepool is nil")
	}
	expectedGCPClusterRef = strings.TrimSpace(expectedGCPClusterRef)
	if expectedGCPClusterRef == "" {
		return errors.Errorf("expected GCP cluster ref is empty")
	}
	actualGCPClusterRef := gkeheterogeneous.NodePoolClusterRef(nodePool)
	if actualGCPClusterRef != expectedGCPClusterRef {
		return errors.Errorf("nodepool %s belongs to GCP cluster %q, want %q", nodePool.Name, actualGCPClusterRef, expectedGCPClusterRef)
	}
	return nil
}

func (t *NodePool) patchGKENodePoolComputeClassMetadata(ctx context.Context, nodePool *v1beta1.NodePool, metadata gkeNodePoolComputeClassMetadata) error {
	if nodePool == nil {
		return errors.Errorf("gke compute class nodepool is nil")
	}
	metadataMatches := nodepoolutil.GKEComputeClassMetadataMatchesNodePool(nodePool, metadata.desired, metadata.applied)
	markerMatches := !metadata.markEverBound || gkeHeterogeneousNodePoolEverBound(nodePool)
	isolationMatches := apiManagedGKEHeterogeneousIsolationTaintMatches(nodePool, metadata)
	if metadataMatches && markerMatches && isolationMatches {
		return nil
	}

	updated := nodePool.DeepCopy()
	if !metadataMatches {
		nodepoolutil.ApplyGKEComputeClassMetadataForNodePool(updated, metadata.desired, metadata.applied)
	}
	if !isolationMatches {
		applyAPIManagedGKEHeterogeneousIsolationTaintExpectation(updated, metadata)
	}
	if metadata.markEverBound && !gkeHeterogeneousNodePoolEverBound(updated) {
		markGKEHeterogeneousNodePoolEverBound(updated)
	}
	if err := t.Local.Patch(ctx, updated, client.MergeFrom(nodePool)); err != nil {
		return err
	}
	nodePool.Annotations = updated.Annotations
	nodePool.Spec.ForProvider.Config = updated.Spec.ForProvider.Config
	return nil
}

func (t *NodePool) patchPreparedAPIManagedGKEHeterogeneousIsolationTaint(ctx context.Context, nodePool *v1beta1.NodePool) error {
	if nodePool == nil {
		return errors.Errorf("gke compute class nodepool is nil")
	}
	if !gkeheterogeneous.IsAPIManagedNodePool(nodePool) || nodepoolutil.HasGKEHeterogeneousNodePoolIsolationTaint(nodePool) {
		return nil
	}

	updated := nodePool.DeepCopy()
	nodepoolutil.SetGKEHeterogeneousNodePoolIsolationTaint(updated)
	if err := t.Local.Patch(ctx, updated, client.MergeFrom(nodePool)); err != nil {
		return err
	}
	nodePool.Spec.ForProvider.Config = updated.Spec.ForProvider.Config
	return nil
}

func apiManagedGKEHeterogeneousIsolationTaintMatches(nodePool *v1beta1.NodePool, metadata gkeNodePoolComputeClassMetadata) bool {
	want, applies := apiManagedGKEHeterogeneousIsolationTaintExpectation(nodePool, metadata)
	if !applies {
		return true
	}
	return nodepoolutil.HasGKEHeterogeneousNodePoolIsolationTaint(nodePool) == want
}

func applyAPIManagedGKEHeterogeneousIsolationTaintExpectation(nodePool *v1beta1.NodePool, metadata gkeNodePoolComputeClassMetadata) {
	want, applies := apiManagedGKEHeterogeneousIsolationTaintExpectation(nodePool, metadata)
	if !applies {
		return
	}
	if want {
		nodepoolutil.SetGKEHeterogeneousNodePoolIsolationTaint(nodePool)
		return
	}
	nodepoolutil.ClearGKEHeterogeneousNodePoolIsolationTaint(nodePool)
}

func apiManagedGKEHeterogeneousIsolationTaintExpectation(nodePool *v1beta1.NodePool, metadata gkeNodePoolComputeClassMetadata) (want bool, applies bool) {
	if !gkeheterogeneous.IsAPIManagedNodePool(nodePool) {
		return false, false
	}

	desired := strings.TrimSpace(metadata.desired)
	applied := strings.TrimSpace(metadata.applied)
	if desired != "" && applied != "" {
		return false, true
	}
	if desired != "" && applied == "" &&
		nodepoolutil.GKEComputeClassMetadataMatchesNodePool(nodePool, desired, desired) {
		return false, true
	}
	return true, true
}

func gkeNodePoolComputeClassCleanupGuard(nodePool *v1beta1.NodePool) string {
	// Keep an applied reference during cleanup so ComputeClass deletion remains
	// blocked until the GKE actual node label/taint is confirmed removed.
	for _, computeClass := range []string{
		nodepoolutil.GKEAppliedComputeClassNameForNodePool(nodePool),
		nodepoolutil.GKEDesiredComputeClassNameForNodePool(nodePool),
		nodepoolutil.GKEComputeClassSpecLabelForNodePool(nodePool),
		nodepoolutil.GKEComputeClassSpecTaintForNodePool(nodePool),
	} {
		if computeClass = strings.TrimSpace(computeClass); computeClass != "" {
			return computeClass
		}
	}
	return ""
}

func gkeHeterogeneousNodePoolEverBound(nodePool *v1beta1.NodePool) bool {
	if nodePool == nil {
		return false
	}
	return strings.TrimSpace(nodePool.GetAnnotations()[consts.AnnoKeyGKEHeterogeneousNodePoolEverBound]) == "true"
}

func markGKEHeterogeneousNodePoolEverBound(nodePool *v1beta1.NodePool) {
	if nodePool == nil {
		return
	}
	annotations := nodePool.GetAnnotations()
	if annotations == nil {
		annotations = map[string]string{}
	}
	annotations[consts.AnnoKeyGKEHeterogeneousNodePoolEverBound] = "true"
	nodePool.SetAnnotations(annotations)
}