package gke

import (
	"context"
	"fmt"
	"time"

	"cloud.google.com/go/container/apiv1/containerpb"
	gkev1beta1 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/tools/record"
	"k8s.io/utils/pointer"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	. "github.com/tidbcloud/eks-provider/controllers/upgradeoperation/common"
	"github.com/tidbcloud/eks-provider/controllers/utils/ticdcservicecli"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/applier"
	"github.com/tidbcloud/infra-provider/pkg/applier/mutatefuncs"
	"github.com/tidbcloud/infra-provider/pkg/cloud/gcp"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/gkemindomains"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/nodepool"
)

type upgrader struct {
	cfg           *config.RootConfig
	infraCli      client.Client
	gcpClient     gcp.Client
	eventRecorder record.EventRecorder
	cdcUpgrader   *CDCUpgrader

	// Used to apply the node pool CRs.
	// Should align with the nodepool task in the cluster controller.
	npApplier applier.Applier[*gkev1beta1.NodePool, gkev1beta1.NodePool]
}

var _ Upgrader = &upgrader{}

func (u *upgrader) trackProgress(ctx context.Context, uo *cloudv1.UpgradeOperation, phase UpgradePhase, done, total int) error {
	return TrackProgressAndUpdate(ctx, u.infraCli, uo, phase, done, total, time.Now())
}

func NewUpgrader(cfg *config.RootConfig, infraCli client.Client, gcpClient gcp.Client, eventRecorder record.EventRecorder, ticdcCli ticdcservicecli.Client) Upgrader {
	return &upgrader{
		cfg:           cfg,
		infraCli:      infraCli,
		gcpClient:     gcpClient,
		npApplier:     applier.NewApplier(infraCli, mutatefuncs.MutateGKENodePool),
		eventRecorder: eventRecorder,
		cdcUpgrader:   NewCDCUpgrader(infraCli, ticdcCli),
	}
}

func (u *upgrader) UpgradeTiDBNodes(ctx context.Context, uo *cloudv1.UpgradeOperation, envProvider dataplane.EnvProvider, cluster *cloudv1.Cluster) (ctrl.Result, error) {
	logger := log.FromContext(ctx).WithValues("cluster", cluster.Name)
	logger.Info("UpgradeTiDBNodes")

	// Fetch necessary information.
	desiredVersion := uo.Status.CurrentStep.Version
	gke, err := u.gcpClient.ClusterManager().GetCluster(ctx, envProvider.GetAccount(), envProvider.GetRegion(), envProvider.GetClusterName())
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to get the gke cluster: %w", err)
	}
	gkeCli := envProvider.GetClient()

	currentNodeVersion, err := GetTiDBNodeVersion(cluster, consts.ProviderTypeGKE)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to get TiDB node versions from the cluster CR: %w", err)
	}
	logger.Info("current node version", "version", currentNodeVersion)

	// Check if the tidb nodes have been upgraded.
	if currentNodeVersion == desiredVersion {
		logger.Info("the tidb nodes are already upgraded")

		if err = nodepool.DeleteOverProvisioningDeployments(ctx, gkeCli, consts.ClusterPrefix, cluster.PrefixedId()); err != nil {
			logger.Error(err, "failed to delete over provisioning deployments")
			return ctrl.Result{}, err
		}
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "All tidb nodes are already upgraded, delete the old version nodes", nil)
		return ctrl.Result{}, FinishUpgrading(ctx, u.infraCli, uo)
	}

	// If all tidb clusters under the network are suspended, we can upgrade all node pools directly.
	if allTiDBClustersSuspended(ctx, u.infraCli, envProvider, cluster) {
		logger.Info("All tidb clusters in the gke are suspended, will upgrade all tidb node pools directly")

		allNPs := getAllTiDBNodePoolsUnderNetwork(ctx, u.infraCli, logger, gke)
		return u.upgradeNodePools(ctx, uo, envProvider, allNPs, cloudv1.TiDBNodes)
	}

	// If the cluster is suspended, and there are other non-suspended clusters in the gke,
	// we can directly update the node version in the cluster CR's status,
	// so that when it's resumed, it will be added node selector to use new version nodes.
	// This check MUST be after the `allTiDBClustersSuspended` check.
	if cluster.Status.Phase == cloudv1.ClusterPhaseSuspend {
		logger.Info("will only update the node version in cluster CR's status", "desired version", desiredVersion)
		return ctrl.Result{Requeue: true}, UpdateClusterStatus(ctx, u.infraCli, cluster, desiredVersion)
	}

	useMinDomains := false
	if gkemindomains.Applicable(cluster) {
		decision, persisted, err := u.resolveMinDomainsUpgradeDecision(ctx, gkeCli, uo, cluster)
		if err != nil {
			return ctrl.Result{}, err
		}
		if persisted {
			return ctrl.Result{Requeue: true}, nil
		}
		useMinDomains = decision == gkemindomains.UpgradeDecisionMinDomains
	}

	// Phase 1: get the old and new version node pools.
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Begin to get the old and create new version node pools for all TCs (default + extra groups)", nil)
	var list gkev1beta1.NodePoolList
	labelSelector, err := getTiDBNodePoolLabelSelector(cluster)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to get tidb node pool label selector: %w", err)
	}
	logger.Info("tidb nodepool label selector", "selector", labelSelector.Selector.String(), "networkName", cluster.Spec.NetworkName, "use3Layer", cluster.FeatureEnabled(consts.ClusterFeatureGCP3LayerNodePool))
	if err = u.infraCli.List(ctx, &list, labelSelector); err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to list tidb node pools: %w", err)
	}
	if cluster.Spec.NetworkName == "" && !cluster.FeatureEnabled(consts.ClusterFeatureGCP3LayerNodePool) {
		before := len(list.Items)
		list = *filterLegacyNodePoolsByAntiLabels(cluster, &list, currentNodeVersion, desiredVersion)
		after := len(list.Items)
		if before != after {
			logger.Info("filtered legacy nodepools by anti-labels", "before", before, "after", after)
		}
	}
	oldNPs, newNPs, needCreate, err := getTiDBNodePoolsByCluster(&list, cluster, desiredVersion)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to get tidb node pools: %w", err)
	}
	logger.Info("tidb node pools", "old", GetCRNameList(oldNPs), "new", GetCRNameList(newNPs), "needCreate", needCreate)

	// Pause the network and cluster CR, start to upgrade the tidb node pools.
	scaleOutExtraTiDB := uo.GetCurrentStepOption(ScaleOutExtraTiDBAnnotation) != "false"
	logger.Info("start to upgrade the tidb node pools", "scaleOutExtraTiDB", scaleOutExtraTiDB)

	// Create new version node pools
	_ = u.trackProgress(ctx, uo, PhaseCreateNodePools, len(newNPs), len(oldNPs))
	if needCreate {
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Start to create new version node pools", nil)
		if err = createNewVersionNodePoolsForCluster(ctx, u.infraCli, cluster, oldNPs, desiredVersion, gke.GetCurrentMasterVersion()); err != nil {
			logger.Error(err, "failed to create new version tidb node pools")
			LogAndEmitEvent(logger, u.eventRecorder, uo, WarningEvent, "Failed to create new version node pools", err)
		}

		// If there is no errors, need requeue to get new version node pools in the next reconcile.
		// Otherwise, `Requeue` will be ignored.
		return ctrl.Result{Requeue: err == nil}, err
	}

	// Add resource labels to the new version node pools.
	// We had better do this before the new version node pools scale out from zero,
	// otherwise, all new version nodes need to be recreated to add resource labels.
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Start to add resource labels to the new version node pools", nil)
	if errCount, err := addResourceLabelsToNodePools(ctx, u.gcpClient, newNPs, getResourceLabels(cluster)); err != nil {

		_ = u.trackProgress(ctx, uo, PhaseAddResourceLabels, len(newNPs)-errCount, len(newNPs))

		// If the gke cluster is reconciling, we should wait for next reconcile.
		if gke.GetStatus() == containerpb.Cluster_RECONCILING {
			logger.Info("gke cluster is reconciling, wait for next reconcile")
			return ctrl.Result{RequeueAfter: 2 * time.Minute}, nil
		}

		logger.Error(err, "failed to add resource labels to the new version node pools")
		return ctrl.Result{}, err
	}

	// Phase 2: wait for new version node pools are ready.
	// Wait for new version node pools are ready
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Begin to wait for new version node pools are ready", nil)
	ready, npTotal := countReadyNodePools(newNPs)
	_ = u.trackProgress(ctx, uo, PhaseWaitNodesReady, ready, npTotal)
	if ready < npTotal {
		logger.Info("the new version node pools are not ready")
		return ctrl.Result{RequeueAfter: 1 * time.Minute}, nil
	}
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "All new version node pools are ready, start to create over provisioning pods", nil)

	// Phase 3: create over provisioning pods to scale out new version nodes first.
	// Otherwise, each pod will keep pending for about 70 seconds to be scheduled to a new version node.
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Begin to create over provisioning pods to scale out new version nodes first", nil)
	if err = nodepool.CreateOverProvisioningPods(
		ctx, gkeCli, cluster, consts.ClusterPrefix, cluster.PrefixedId(), desiredVersion, u.cfg, useMinDomains,
	); err != nil {
		logger.Error(err, "failed to create over provisioning pods")
		return ctrl.Result{}, err
	}
	// Wait for new version nodes ready
	opReady, opTotal := nodepool.CountOverProvisioningPodsReady(ctx, gkeCli, consts.ClusterPrefix, cluster.PrefixedId())
	_ = u.trackProgress(ctx, uo, PhaseWaitOverprovisioning, opReady, opTotal)
	if opReady < opTotal || opTotal == 0 {
		logger.Info("the new version nodes are not ready")
		return ctrl.Result{RequeueAfter: 2 * time.Minute}, nil
	}
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "All new version nodes are ready, start to update the tidbcluster CR to restart components", nil)

	// Phase 4: get all TCs (default + extra groups).
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Begin to get all TCs (default + extra groups)", nil)
	var (
		allTCs    []*opv1alpha1.TidbCluster
		defaultTC opv1alpha1.TidbCluster
	)
	if err = gkeCli.Get(ctx, types.NamespacedName{Name: consts.ClusterPrefix, Namespace: cluster.PrefixedId()}, &defaultTC); err != nil {
		logger.Error(err, "unable to get the tidbcluster CR")
		return ctrl.Result{}, err
	}
	allTCs = append(allTCs, &defaultTC)

	for _, extraGroup := range cluster.GetExtraGroups() {
		if extraGroup.Deleted() {
			logger.Info("skip zero replicas extra group", "groupName", extraGroup.Name)
			continue
		}
		extraGroupNamer := namer.NewNamer().
			ForTiDBGroup(cluster, extraGroup.Spec, "").
			WithComponent(string(cloudv1.ComponentTiDB))
		extraGroupTCName := extraGroupNamer.GetTiDBName()
		extraGroupTC := &opv1alpha1.TidbCluster{}
		if err := gkeCli.Get(ctx, types.NamespacedName{Namespace: cluster.PrefixedId(), Name: extraGroupTCName}, extraGroupTC); err != nil {
			logger.Error(err, "unable to get the extra tidbcluster CR", "groupName", extraGroupTCName)
			return ctrl.Result{}, err
		}
		allTCs = append(allTCs, extraGroupTC)
	}

	// Phase 5-7: restart components in order defined by DefaultRestartPhases.
	// Each phase: mutate restart annotations + overlay node selector → wait for pod migration → wait for ready.
	for _, ph := range DefaultRestartPhases {
		res, done, phErr := u.restartComponentsAndWait(
			ctx, gkeCli, uo, cluster, &defaultTC, allTCs, newNPs, ph,
			scaleOutExtraTiDB, desiredVersion, useMinDomains,
		)
		if !done {
			return res, phErr
		}
	}

	// Phase 8: wait until all pods have migrated to the new node pools.
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Begin to wait until ALL components are restarted + ready and pods migrated to new nodes", nil)
	onNew, podTotal, err := countTCPodsOnNewNodePools(ctx, gkeCli, cluster.PrefixedId(), newNPs)
	if err != nil {
		logger.Error(err, "unable to check if all TC pods are on new node pools")
		return ctrl.Result{}, err
	}
	_ = u.trackProgress(ctx, uo, PhaseWaitAllPodsMigrated, onNew, podTotal)
	if onNew < podTotal || podTotal == 0 {
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Waiting for all TC pods to be on new version nodes", nil)
		return ctrl.Result{RequeueAfter: 1 * time.Minute}, nil
	}

	// Phase 9: check readiness of all TCs (default + extra groups).
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Begin to check readiness of all TCs (default + extra groups)", nil)
	tcReady, tcTotal := CountTCsReady(allTCs, &defaultTC)
	_ = u.trackProgress(ctx, uo, PhaseWaitTCReady, tcReady, tcTotal)
	if tcReady < tcTotal || tcTotal == 0 {
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Waiting for all TCs to be ready", nil)
		return ctrl.Result{RequeueAfter: 1 * time.Minute}, nil
	}
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "All tidb component pods are running on the new version nodes and tidb cluster is ready", nil)

	// Phase 10: scale in the extra tidb
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Begin to scale in the extra tidb", nil)
	if scaleOutExtraTiDB {
		defaultTC.Spec.TiDB.Replicas = int32(cluster.Spec.Cluster.TiDB.Replicas)
		if err := k8s.Apply(ctx, gkeCli, &defaultTC); err != nil {
			logger.Error(err, "unable to update the tidbcluster CR")
			return ctrl.Result{}, err
		}
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Scale in the extra tidb", nil)
	}

	logger.Info("updated the node version in cluster CR's status", "nodeVersion", desiredVersion)
	return ctrl.Result{Requeue: true}, UpdateClusterStatus(ctx, u.infraCli, cluster, desiredVersion)
}

// restartComponentsAndWait handles one restart phase for all TCs:
//  1. Mutate restart annotations + overlay K8s-version node selector for ph.Components; if any TC
//     changed, persist and requeue.
//     For TiDB on the default TC, also sets the extra-TiDB replica count when scaleOutExtraTiDB is true.
//  2. For non-TiDB components: verify all pods have migrated to new node pools.
//     (TiDB pod migration is verified by areAllTCPodsOnNewNodePools in the tail.)
//  3. Verify every TC reports the component as restarted and ready.
//
// Returns (result, done=false, err) when the caller should return immediately, or (_, done=true, nil)
// when the phase is complete and the caller may proceed to the next phase.
func (u *upgrader) restartComponentsAndWait(
	ctx context.Context,
	gkeCli client.Client,
	uo *cloudv1.UpgradeOperation,
	cluster *cloudv1.Cluster,
	defaultTC *opv1alpha1.TidbCluster,
	allTCs []*opv1alpha1.TidbCluster,
	newNPs []*gkev1beta1.NodePool,
	ph RestartPhase,
	scaleOutExtraTiDB bool,
	desiredVersion string,
	useMinDomains bool,
) (result ctrl.Result, done bool, err error) {
	logger := log.FromContext(ctx)

	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent,
		fmt.Sprintf("Begin restart phase %q for ALL TCs", ph.Name), nil)

	// Step 1: write restart annotations and overlay node selectors.
	updatedAny := false
	for _, tc := range allTCs {
		updated := mutateTCForGKEUpgradeRestart(
			tc, desiredVersion, ph.Components, useMinDomains,
		)
		if updated {
			updatedAny = true
			// When restarting TiDB on the default TC, opportunistically scale out an extra TiDB pod so
			// the extra node created in Phase 1 is utilised immediately.
			for _, comp := range ph.Components {
				if comp == cloudv1.ComponentTiDB && tc.Name == defaultTC.Name && scaleOutExtraTiDB {
					tc.Spec.TiDB.Replicas = int32(cluster.Spec.Cluster.TiDB.Replicas + 1)
				}
			}
			logger.Info("updated tidbcluster CR for restart phase", "tc", tc.Name, "phase", ph.Name)
			if err = k8s.Apply(ctx, gkeCli, tc); err != nil {
				logger.Error(err, "unable to apply tidbcluster CR", "tc", tc.Name)
				return ctrl.Result{}, false, err
			}
		}
	}
	if updatedAny {
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, ph.Reason, nil)
		_ = u.trackProgress(ctx, uo, RestartPhaseID(ph), len(allTCs), len(allTCs))
		return ctrl.Result{RequeueAfter: 1 * time.Minute}, false, uo.UpdateReason(ctx, u.infraCli, ph.Reason)
	}

	// Step 2: verify pod migration to new node pools (skipped for TiDB — handled in the aggregate tail check).
	migrated, migrateTotal, migratable, migrateErr := CountRestartPhaseMigration(logger, cluster.PrefixedId(), allTCs, ph,
		func(namespace, component string, tc *opv1alpha1.TidbCluster) (int, int, error) {
			return countComponentPodsOnNewNodePools(ctx, gkeCli, namespace, component, newNPs, tc)
		})
	if migrateErr != nil {
		logger.Error(migrateErr, "unable to check pod migration", "phase", ph.Name)
		return ctrl.Result{}, false, migrateErr
	}
	_ = u.trackProgress(ctx, uo, RestartMigratePhaseID(ph), migrated, migrateTotal)
	if migratable && (migrateTotal == 0 || migrated < migrateTotal) {
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent,
			fmt.Sprintf("Waiting for %s pods to be on new version nodes for all TCs", ph.Name), nil)
		return ctrl.Result{RequeueAfter: 1 * time.Minute}, false, uo.UpdateReason(ctx, u.infraCli, ph.Reason)
	}

	// Step 3: verify restart completed and component is ready.
	readyCount, readyTotal := CountRestartPhaseReady(logger, allTCs, desiredVersion, ph)
	_ = u.trackProgress(ctx, uo, RestartReadyPhaseID(ph), readyCount, readyTotal)
	if readyCount < readyTotal {
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent,
			fmt.Sprintf("Waiting for %s restart to complete for all TCs", ph.Name), nil)
		return ctrl.Result{RequeueAfter: 1 * time.Minute}, false, uo.UpdateReason(ctx, u.infraCli, ph.Reason)
	}

	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent,
		fmt.Sprintf("Restart phase %q complete for all TCs", ph.Name), nil)
	return ctrl.Result{}, true, nil
}

func (u *upgrader) resolveMinDomainsUpgradeDecision(
	ctx context.Context,
	gkeCli client.Client,
	uo *cloudv1.UpgradeOperation,
	cluster *cloudv1.Cluster,
) (gkemindomains.UpgradeDecision, bool, error) {
	decision := gkemindomains.UpgradeDecision(
		uo.GetCurrentStepOption(gkemindomains.UpgradeDecisionOption),
	)
	if decision == gkemindomains.UpgradeDecisionMinDomains ||
		decision == gkemindomains.UpgradeDecisionWithoutMinDomains {
		return decision, false, nil
	}

	decision = gkemindomains.UpgradeDecisionMinDomains
	unlabeled, err := nodepool.HasUnlabeledUpgradeOverProvisioning(
		ctx, gkeCli, consts.ClusterPrefix, cluster.PrefixedId(),
	)
	if err != nil {
		return "", false, fmt.Errorf("check existing Kubernetes upgrade over-provisioning deployment: %w", err)
	}
	if unlabeled {
		decision = gkemindomains.UpgradeDecisionWithoutMinDomains
	}
	if uo.Status.CurrentStep.Options == nil {
		uo.Status.CurrentStep.Options = make(map[string]string)
	}
	uo.Status.CurrentStep.Options[gkemindomains.UpgradeDecisionOption] = string(decision)
	uo.Status.Reason = "Preparing Kubernetes upgrade over-provisioning capacity"
	if err := u.infraCli.Status().Update(ctx, uo); err != nil {
		return "", false, fmt.Errorf("update GKE v1 minDomains upgrade decision: %w", err)
	}
	return decision, true, nil
}

func mutateTCForGKEUpgradeRestart(
	tc *opv1alpha1.TidbCluster,
	desiredVersion string,
	components []cloudv1.ComponentType,
	useMinDomains bool,
) bool {
	updated := MutateTCForRestartingComponents(tc, desiredVersion, components...)
	if useMinDomains {
		for _, component := range components {
			if TCComponentReplicas(tc, component) > 0 {
				updated = gkemindomains.EnsureMinDomainsConstraints(tc, component) || updated
			}
		}
	}
	if updated {
		nodepool.OverlayNodeSelectorWithK8sVersionForComponents(tc, desiredVersion, components...)
	}
	return updated
}

// UpgradeCDCNodes delegates to the cloud-agnostic CDCUpgrader. See common.CDCUpgrader.Upgrade
// for the implementation details.
func (u *upgrader) UpgradeCDCNodes(ctx context.Context, uo *cloudv1.UpgradeOperation, envProvider dataplane.EnvProvider, cluster *cloudv1.Cluster) (ctrl.Result, error) {
	return u.cdcUpgrader.Upgrade(ctx, uo, envProvider, cluster)
}

func (u *upgrader) UpgradeDMNodes(ctx context.Context, uo *cloudv1.UpgradeOperation, envProvider dataplane.EnvProvider, cluster *cloudv1.Cluster) (ctrl.Result, error) {
	logger := log.FromContext(ctx).WithValues("cluster", cluster.Name)
	logger.Info("UpgradeDMNodes")

	var list gkev1beta1.NodePoolList
	if err := u.infraCli.List(ctx, &list, client.MatchingLabelsSelector{
		Selector: labels.Set{
			"cluster":   cluster.Name,
			"component": "dm",
		}.AsSelector(),
	}); err != nil && !apierrors.IsNotFound(err) {
		logger.Error(err, "failed to list tiflow node pools")
		return ctrl.Result{}, err
	}

	return u.upgradeNodePools(ctx, uo, envProvider, &list, cloudv1.DMNodes)
}

// upgradeNodePools upgrades node pools in a GKE cluster.
// It firstly checks if the control plane is upgraded to the desired version.
// If yes, it will upgrade node pools to the same version as the current control plane's version.
// Note: this function will NOT create new node pools.
func (u *upgrader) upgradeNodePools(ctx context.Context, uo *cloudv1.UpgradeOperation, envProvider dataplane.EnvProvider, npList *gkev1beta1.NodePoolList, nodeType cloudv1.UpgradeOperationTarget) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	logger.Info("upgradeNodePools")

	gke, err := u.gcpClient.ClusterManager().GetCluster(ctx, envProvider.GetAccount(), envProvider.GetRegion(), envProvider.GetClusterName())
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to get the gke cluster: %w", err)
	}

	// x.y.z-gke.xxx
	masterVersion := gke.GetCurrentMasterVersion()

	// If the gke cluster is reconciling, we should wait for next reconcile.
	if gke.GetStatus() == containerpb.Cluster_RECONCILING {
		logger.Info("gke cluster is reconciling, wait for next reconcile")
		return ctrl.Result{RequeueAfter: 2 * time.Minute}, nil
	}
	if gke.GetStatus() != containerpb.Cluster_RUNNING {
		return ctrl.Result{}, fmt.Errorf("gke cluster %s is in unexpected status %s", envProvider.GetClusterName(), gke.GetStatus())
	}

	// npSet is a set of node pool names to be upgraded.
	npSet := make(map[string]*gkev1beta1.NodePool, len(npList.Items))
	for i := range npList.Items {
		np := &npList.Items[i]
		npSet[np.Name] = np
	}
	toBeUpgradedNPs := make([]string, 0, len(npSet))

	// gke CR's status stores the current status of node pools.
	for _, npStatus := range gke.GetNodePools() {
		if npStatus == nil {
			continue
		}
		if _, ok := npSet[npStatus.GetName()]; ok {
			if npStatus.Version == masterVersion && npStatus.GetStatus() == containerpb.NodePool_RUNNING {
				logger.Info("node pool is already upgraded", "node pool", npStatus.Name, "version", npStatus.Version)
			} else {
				toBeUpgradedNPs = append(toBeUpgradedNPs, npStatus.Name)
			}
		}
	}
	// When there are no node pools left to be upgraded
	if len(toBeUpgradedNPs) == 0 {
		logger.Info(fmt.Sprintf("all node pools of %s are upgraded", nodeType))
		return ctrl.Result{}, FinishUpgrading(ctx, u.infraCli, uo)
	}

	logger.Info("node pools to be upgraded", "nodeType", nodeType, "node pools", toBeUpgradedNPs)

	for _, npName := range toBeUpgradedNPs {
		np := npSet[npName]
		logger.Info("start to upgrade the node pool", "nodeType", nodeType, "name", np.Name, "spec version", np.Spec.ForProvider.Version)

		// Upgrade the node pool via gcloud sdk.
		// Note: we can not upgrade the node pool via CR.
		if err := u.gcpClient.UpgradeNodePool(ctx, envProvider.GetAccount(), envProvider.GetRegion(), envProvider.GetClusterName(), np.Name, masterVersion); err != nil {
			logger.Error(err, "unable to upgrade the node pool via sdk", "nodeType", nodeType, "name", np.Name)
			continue
		}

		// We also need to update the node pool CR's version in spec.
		np.Spec.ForProvider.Version = pointer.String(masterVersion)
		if err = k8s.Apply(ctx, u.infraCli, np); err != nil {
			logger.Error(err, "unable to update the node pool CR", "name", np.Name)
			continue
		}
	}

	_ = u.trackProgress(ctx, uo, PhaseUpgradeNodePools, len(npSet)-len(toBeUpgradedNPs), len(npSet))
	return ctrl.Result{RequeueAfter: 2 * time.Minute}, nil
}