package eks

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/aws/aws-sdk-go-v2/service/eks"
	"github.com/aws/aws-sdk-go-v2/service/ssm"
	autoscalingv1beta1 "github.com/crossplane-contrib/provider-aws/apis/autoscaling/v1beta1"
	ec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1"
	"github.com/crossplane-contrib/provider-aws/apis/eks/manualv1alpha1"
	"github.com/crossplane-contrib/provider-aws/apis/eks/v1alpha1"
	"github.com/crossplane-contrib/provider-aws/apis/eks/v1beta1"
	"github.com/go-logr/logr"
	"github.com/hashicorp/go-multierror"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"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/k8sversion/common"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/cloud/aws"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/extratidb"
	"github.com/tidbcloud/infra-provider/pkg/gitops"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

type EKSUpgrader struct {
	cfg           *config.RootConfig
	awsCli        aws.Client
	infraCli      client.Client
	provider      gitops.Provider
	extraTiDB     extratidb.Handler
	ssmClient     *ssm.Client
	eventRecorder record.EventRecorder

	// key: eks cluster CR's UID (do not use the name, since the CR may be recreated with the same name)
	// value: eks client
	eksClientMap sync.Map
	accountMap   sync.Map
}

var _ Upgrader = &EKSUpgrader{}

func NewUpgrader(cfg *config.RootConfig, infraCli client.Client, awsCli aws.Client, eventRecorder record.EventRecorder) *EKSUpgrader {
	return &EKSUpgrader{
		awsCli:        awsCli,
		infraCli:      infraCli,
		cfg:           cfg,
		provider:      cfg,
		extraTiDB:     extratidb.NewAWSHandler(infraCli, awsCli),
		ssmClient:     awsCli.SSM(nil),
		eventRecorder: eventRecorder,
	}
}

type upgradeInfo struct {
	desiredK8sVersion  string
	currentCPVersion   string
	currentNodeVersion string

	*namer.Namer
	logger            *logr.Logger
	uo                *cloudv1.UpgradeOperation
	network           *cloudv1.Network
	cluster           *cloudv1.Cluster
	migrateNodeGroup  bool
	scaleOutExtraTiDB bool
	useSelfManaged    bool
	eks               *v1beta1.Cluster
}

func (u *EKSUpgrader) UpgradeControlPlane(ctx context.Context, uo *cloudv1.UpgradeOperation, network *cloudv1.Network) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	desiredVersion := uo.Status.CurrentStep.Version
	eks, err := getEKSCR(ctx, u.infraCli, network)
	if err != nil {
		logger.Error(err, "unable to get the eks cluster CR")
		if apierrors.IsNotFound(err) {
			uo.Status.Phase = cloudv1.UpgradeOperationPhaseFailed
			uo.Status.Reason = fmt.Sprintf("The EKS CR %s/%s is not found: %s", uo.Namespace, uo.Spec.NetworkName, err.Error())
			return ctrl.Result{}, u.infraCli.Status().Update(ctx, uo)
		}
		return ctrl.Result{}, err
	}

	switch eks.Status.AtProvider.Status {
	case v1beta1.ClusterStatusUpdating:
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "The EKS cluster is updating, wait for the next reconcile", nil)
		return ctrl.Result{RequeueAfter: 1 * time.Minute}, nil

	case v1beta1.ClusterStatusActive:
		if eks.Status.AtProvider.Version == desiredVersion {
			LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "The EKS cluster's control plane is already upgraded", nil)
			return ctrl.Result{Requeue: true}, FinishUpgrading(ctx, u.infraCli, uo)
		} else if ExpectToBeUpgraded(cloudv1.ControlPlane, desiredVersion, eks.Status.AtProvider.Version) {
			logger.Info("start to upgrade the eks cluster's control plane")
			eks.Spec.ForProvider.Version = &desiredVersion
			if err := k8s.Apply(ctx, u.infraCli, eks); err != nil {
				logger.Error(err, "unable to upgrade eks cluster CR")
				return ctrl.Result{}, err
			}
			LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "The EKS cluster CR is updated, wait for the next reconcile", nil)
			return ctrl.Result{RequeueAfter: 5 * time.Minute}, nil
		} else {
			LogAndEmitEvent(logger, u.eventRecorder, uo, WarningEvent, "The EKS cluster is in unexpected version", nil)
		}

	default:
		LogAndEmitEvent(logger, u.eventRecorder, uo, WarningEvent, "The EKS cluster is in unexpected status", nil)
	}

	return ctrl.Result{}, nil
}

func (u *EKSUpgrader) UpgradeAdminNodes(ctx context.Context, uo *cloudv1.UpgradeOperation, network *cloudv1.Network) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	var ng manualv1alpha1.NodeGroup
	key := types.NamespacedName{Name: namer.UniqueIdentifier(network.ObjectMeta, namer.WithPrefix(consts.TiDBCloudPrefix))}
	if err := u.infraCli.Get(ctx, key, &ng); err != nil {
		logger.Error(err, "unable to get the eks admin nodegroup CR")
		return ctrl.Result{}, err
	}
	logger.Info("admin nodegroup", "name", ng.Name, "spec version", ng.Spec.ForProvider.Version, "status version", ng.Status.AtProvider.Version, "status", ng.Status.AtProvider.Status, "desired size", ng.Status.AtProvider.ScalingConfig.DesiredSize)

	return u.upgradeManagedNodeGroups(ctx, uo, network, nil, &manualv1alpha1.NodeGroupList{Items: []manualv1alpha1.NodeGroup{ng}}, cloudv1.AdminNodes)
}

func (u *EKSUpgrader) UpgradeCDCNodes(ctx context.Context, uo *cloudv1.UpgradeOperation, network *cloudv1.Network, cluster *cloudv1.Cluster) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	var list manualv1alpha1.NodeGroupList
	if err := u.infraCli.List(ctx, &list, client.MatchingLabelsSelector{
		Selector: labels.Set{
			"cluster":   cluster.Name,
			"component": "ticdc",
		}.AsSelector(),
	}); err != nil && !apierrors.IsNotFound(err) {
		logger.Error(err, "unable to list the ticdc nodegroups")
		return ctrl.Result{}, err
	}

	return u.upgradeManagedNodeGroups(ctx, uo, network, cluster, &list, cloudv1.CDCNodes)
}

func (u *EKSUpgrader) UpgradeDMNodes(ctx context.Context, uo *cloudv1.UpgradeOperation, network *cloudv1.Network, cluster *cloudv1.Cluster) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	var list manualv1alpha1.NodeGroupList
	if err := u.infraCli.List(ctx, &list, client.MatchingLabelsSelector{
		Selector: labels.Set{
			"cluster":   cluster.Name,
			"component": "tiflow",
		}.AsSelector(),
	}); err != nil && !apierrors.IsNotFound(err) {
		logger.Error(err, "unable to list the tiflow nodegroups")
		return ctrl.Result{}, err
	}

	return u.upgradeManagedNodeGroups(ctx, uo, network, cluster, &list, cloudv1.DMNodes)
}

func (u *EKSUpgrader) UpgradeTiDBNodes(ctx context.Context, uo *cloudv1.UpgradeOperation, network *cloudv1.Network, cluster *cloudv1.Cluster) (ctrl.Result, error) {
	logger := log.FromContext(ctx).WithValues("cluster", cluster.Name)

	// Get desired and current control plane versions.
	desiredCPVersion := uo.Status.CurrentStep.Version
	eksCR, err := getEKSCR(ctx, u.infraCli, network)
	if err != nil {
		logger.Error(err, "unable to get the EKS CR")
		return ctrl.Result{}, err
	}
	currentCPVersion := eksCR.Status.AtProvider.Version

	// Check if the control plane is upgraded.
	if !isControlPlaneUpgraded(eksCR, desiredCPVersion, currentCPVersion) {
		LogAndEmitEvent(logger, u.eventRecorder, uo, WarningEvent, ControlPlaneNotUpgraded, nil)
		return ctrl.Result{}, uo.Fail(ctx, u.infraCli, ControlPlaneNotUpgraded)
	}

	// Get current node version for TiDB nodes.
	currentNodeVersion, err := GetTiDBNodeVersion(cluster, network.Spec.Provider)
	if err != nil {
		logger.Error(err, "unable to get TiDB node versions from the cluster CR")
		return ctrl.Result{}, err
	}

	// Get the EKS client.
	eksCli, err := u.getEKSClient(ctx, u.infraCli, network, eksCR)
	if err != nil {
		logger.Error(err, "unable to get the EKS client")
		return ctrl.Result{}, err
	}

	// Set up upgrade information.
	info := &upgradeInfo{
		desiredK8sVersion:  desiredCPVersion,
		currentCPVersion:   currentCPVersion,
		currentNodeVersion: currentNodeVersion,
		cluster:            cluster,
		network:            network,
		uo:                 uo,
		Namer:              namer.NewNamer().LegacyWithNetworkName(network.Name).LegacyWithNetworkNamespace(network.Namespace).WithClusterName(cluster.Name).WithTiDBNamespace(cluster.PrefixedId()),
		logger:             &logger,
		eks:                eksCR,
		migrateNodeGroup:   uo.GetCurrentStepOption(MigrateNodeGroupAnnotation) == "true",
		scaleOutExtraTiDB:  uo.GetCurrentStepOption(ScaleOutExtraTiDBAnnotation) != "false",
		useSelfManaged:     util.GetAnnotation(cluster, consts.UseSelfManagedNodegroupAnnotation) == consts.AnnoValTrue,
	}

	oldManagedNGs, newManagedNGs, oldSelfManagedNGs, newSelfManagedNGs, err := getTiDBClusterNodeGroups(ctx, u.infraCli, info)
	if err != nil {
		return ctrl.Result{}, err
	}
	// In some cases, one component is on old version nodes, but the other components are on new version nodes.
	// If the migration is enabled, we should consider the new version manged node groups as old version managed node groups.
	if !info.useSelfManaged && info.migrateNodeGroup && len(newManagedNGs) > 0 {
		logger.Info("tidb components are not on the same version nodes, consider the new version managed node groups as old version managed node groups", "old managed", GetCRNameList(oldManagedNGs), "new managed", GetCRNameList(newManagedNGs))
		oldManagedNGs = append(oldManagedNGs, newManagedNGs...)
		newManagedNGs = []*manualv1alpha1.NodeGroup{}
	}

	logger.Info("tidb node groups", "old managed", GetCRNameList(oldManagedNGs), "new managed", GetCRNameList(newManagedNGs), "old self-managed", GetCRNameList(oldSelfManagedNGs), "new self-managed", GetCRNameList(newSelfManagedNGs))

	if allTiDBNodesUpgraded(info, newManagedNGs, newSelfManagedNGs) {
		if info.migrateNodeGroup {
			util.SetAnnotation(cluster, consts.UseSelfManagedNodegroupAnnotation, consts.AnnoValTrue)
			if err := k8s.Apply(ctx, u.infraCli, cluster); err != nil {
				logger.Error(err, "failed to update the cluster CR's annotation")
				return ctrl.Result{}, err
			}
		}

		// Delete the extra tidb's launch template and node group.
		if err := u.extraTiDB.DeleteNodes(ctx, eksCli, cluster); err != nil {
			logger.Error(err, "unable to delete the extra tidb's launch template and nodegroup")
			return ctrl.Result{}, err
		}

		if err := u.deleteOldVersionTiDBNodes(ctx, eksCli, oldManagedNGs, oldSelfManagedNGs); err != nil {
			logger.Error(err, "failed to delete the old version nodes")
			return ctrl.Result{}, err
		}
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Delete the old version nodes", nil)

		// unpause the network and cluster CR
		logger.Info("the tidb node groups are already upgraded")
		return ctrl.Result{}, FinishUpgrading(ctx, u.infraCli, uo)
	}

	// pause the network and cluster CR
	logger.Info("start to upgrade the tidb node groups", "scaleOutExtraTiDB", info.scaleOutExtraTiDB, "migrate", info.migrateNodeGroup)

	if cluster.Status.Phase == cloudv1.ClusterPhaseSuspend {
		logger.Info("the cluster is in suspend phase")
		return u.handleSuspendCluster(ctx, info, oldManagedNGs, newManagedNGs, oldSelfManagedNGs, newSelfManagedNGs)
	}

	// Create new version node groups
	numOldNGs, numNewNGs := len(oldManagedNGs)+len(oldSelfManagedNGs), len(newManagedNGs)+len(newSelfManagedNGs)
	if numNewNGs < numOldNGs {
		// Create a managed node group for the extra tidb.
		// `PrepareNodes` need to decide which zone to create the extra tidb node based on the current tidb nodes.
		// So we need to create the extra tidb node before creating the new version node groups.
		if info.scaleOutExtraTiDB && numNewNGs == 0 {
			extraTiDBNodegroupName, err := u.extraTiDB.PrepareNodes(ctx, eksCli, u.provider, info.network, info.cluster)
			if err != nil {
				logger.Error(err, "failed to prepare nodes for the extra tidb")
				return ctrl.Result{}, err
			}
			LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, fmt.Sprintf("Created the extra tidb node group %s", extraTiDBNodegroupName), nil)

			// Check if the extra tidb node group is ready.
			extraTiDBNodeReady, err := u.extraTiDB.AreAllNodesReady(ctx, eksCli, cluster)
			if err != nil {
				logger.Error(err, "failed to check if the extra tidb nodegroup is ready")
				return ctrl.Result{}, err
			}

			if extraTiDBNodeReady {
				// Update tc to scale out an extra tidb.
				var tc opv1alpha1.TidbCluster
				if err = eksCli.Get(ctx, types.NamespacedName{Name: info.GetTiDBClusterName(), Namespace: cluster.PrefixedId()}, &tc); err != nil {
					logger.Error(err, "unable to get the tidbcluster CR")
					return ctrl.Result{}, err
				}
				tc.Spec.TiDB.Replicas = int32(cluster.Spec.Cluster.TiDB.Replicas) + 1
				if err = k8s.Apply(ctx, eksCli, &tc); err != nil {
					logger.Error(err, "unable to update the tidbcluster CR")
					return ctrl.Result{}, err
				}
				LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Scale out an extra tidb", nil)
			} else {
				logger.Info("the extra tidb node group is not ready yet")
				return ctrl.Result{Requeue: true}, nil
			}
		}

		// If there is no errors, need requeue to get new version node groups in the next reconcile.
		// Otherwise, `Requeue` will be ignored.
		e := u.createNewVersionTiDBNodes(ctx, info, oldManagedNGs, oldSelfManagedNGs)
		return ctrl.Result{Requeue: e == nil}, e
	}

	if !allNodeGroupsReady(newManagedNGs, newSelfManagedNGs, info.desiredK8sVersion) {
		logger.Info("new version node groups are not ready")
		return ctrl.Result{RequeueAfter: 1 * time.Minute}, nil
	}
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "All new version nodes are ready", nil)

	if err := cordonTiDBNodesByNodeGroups(ctx, eksCli, cluster.Name, oldManagedNGs, oldSelfManagedNGs); err != nil {
		logger.Error(err, "unable to cordon the old node groups' nodes")
		return ctrl.Result{}, err
	}
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Cordon the old node groups' nodes", nil)

	// Update tc to restart the tidb cluster gracefully.
	var tc opv1alpha1.TidbCluster
	if err = eksCli.Get(ctx, types.NamespacedName{Name: info.GetTiDBClusterName(), Namespace: cluster.PrefixedId()}, &tc); err != nil {
		logger.Error(err, "unable to get the tidbcluster CR")
		return ctrl.Result{}, err
	}
	if MutateTCForRestarting(&tc, info.desiredK8sVersion) {
		logger.Info("update tidbcluster CR to restart the tidb cluster")

		if err = k8s.Apply(ctx, eksCli, &tc); err != nil {
			logger.Error(err, "unable to update the tidbcluster CR")
			return ctrl.Result{}, err
		}
		return ctrl.Result{RequeueAfter: 1 * time.Minute}, uo.UpdateReason(ctx, u.infraCli, "Restarting the TiDB Cluster")
	}

	yes, ngNameToPod, err := areAllTiDBPodsOnNewNodeGroups(ctx, eksCli, info.GetTiDBNamespace(), newManagedNGs, newSelfManagedNGs)
	if err != nil {
		logger.Error(err, "unable to check if all tidb component pods are running on the new node groups' nodes")
		return ctrl.Result{}, err
	}
	if !yes {
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Not all tidb component pods are running on the new version nodes", nil)
		return ctrl.Result{RequeueAfter: 2 * time.Minute}, nil
	}
	if !IsTCReady(&tc) {
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Waiting for tidb cluster to be ready", nil)
		return ctrl.Result{Requeue: true}, nil
	}
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "All tidb component pods are running on the new node groups' nodes and tidb cluster is ready", nil)

	// Scale in the extra tidb pod that is on the extra tidb node
	if info.scaleOutExtraTiDB {
		extraTiDBNodegroupName, err := u.extraTiDB.PrepareNodes(ctx, eksCli, u.provider, info.network, info.cluster)
		if err != nil {
			logger.Error(err, "failed to get the extra tidb node group name")
			return ctrl.Result{}, err
		}
		extraTiDBPod, ok := ngNameToPod[extraTiDBNodegroupName]
		if !ok {
			return ctrl.Result{}, fmt.Errorf("no pod is running on the extra tidb node group")
		}
		if err := extratidb.MutateTCToScaleInExtraTiDBPod(&tc, extraTiDBPod, int32(info.cluster.Spec.Cluster.TiDB.Replicas)); err != nil {
			logger.Error(err, "unable to update the tidbcluster CR to scale in the extra tidb pod")
			return ctrl.Result{}, err
		}
		if err := k8s.Apply(ctx, eksCli, &tc); err != nil {
			logger.Error(err, "unable to update the tidbcluster CR")
			return ctrl.Result{}, err
		}
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, fmt.Sprintf("Scale in the extra tidb pod: %s", extraTiDBPod), nil)
	}

	if err = UpdateClusterStatus(ctx, u.infraCli, cluster, desiredCPVersion); err != nil {
		logger.Error(err, "unable to update cluster status")
		return ctrl.Result{}, err
	}
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Updated the node version in cluster CR's status", nil)
	return ctrl.Result{RequeueAfter: time.Second}, nil
}

func (u *EKSUpgrader) getEKSClient(ctx context.Context, cli client.Client, network *cloudv1.Network, eks *v1beta1.Cluster) (client.Client, error) {
	c, ok := u.eksClientMap.Load(string(eks.GetUID()))
	if ok {
		return c.(client.Client), nil
	}

	scheme, err := k8s.TiDBOperatorScheme()
	if err != nil {
		return nil, err
	}

	g, err := k8s.NetworkClientGetter(ctx, cli, network, scheme)
	if err != nil {
		return nil, err
	}

	eksClient, err := g.GetClient(ctx)
	if err != nil {
		return nil, err
	}

	u.eksClientMap.Store(string(eks.GetUID()), eksClient)
	return eksClient, nil
}

// If the cluster is in suspend phase, we can directly modify its node groups to upgrade them.
func (u *EKSUpgrader) handleSuspendCluster(ctx context.Context, info *upgradeInfo, oldManaged, newManaged []*manualv1alpha1.NodeGroup, oldSelfManaged, newSelfManaged []*autoscalingv1beta1.AutoScalingGroup) (ctrl.Result, error) {
	if info.useSelfManaged {
		info.logger.Info("the cluster is in suspend phase and uses self-managed node groups")
		for _, oldAsg := range oldSelfManaged {
			info.logger.Info("delete the old version self-managed node group", "name", oldAsg.Name)
			if err := u.infraCli.Delete(ctx, oldAsg); err != nil {
				info.logger.Error(err, "unable to delete the old version self-managed node group")
			}

			info.logger.Info("deleted the old version launch template", "name", oldAsg.Name)
			if err := u.infraCli.Delete(ctx, &ec2v1alpha1.LaunchTemplate{ObjectMeta: metav1.ObjectMeta{Name: oldAsg.Name}}); err != nil {
				info.logger.Error(err, "unable to delete the old version launch template")
			}
		}
		return ctrl.Result{Requeue: true}, UpdateClusterStatus(ctx, u.infraCli, info.cluster, info.desiredK8sVersion)
	} else {
		if info.migrateNodeGroup {
			msg := "Migrating suspended clusters' node groups is not implemented yet"
			LogAndEmitEvent(*info.logger, u.eventRecorder, info.uo, WarningEvent, msg, nil)
		}

		// If the cluster uses managed node groups
		ngs := append(oldManaged, newManaged...)

		for _, ng := range ngs {
			// If the cluster is suspended, the desired size of the nodegroup should be 0.
			if ng.Status.AtProvider.ScalingConfig.DesiredSize != nil && *ng.Status.AtProvider.ScalingConfig.DesiredSize == 0 {
				if err := updateNodeGroupVersion(ctx, u.infraCli, ng, info.desiredK8sVersion); err != nil {
					info.logger.Error(err, "unable to upgrade the node group CR")
					return ctrl.Result{}, err
				}
				info.logger.Info("update the version of nodegroups", "name", ng.Name, "spec version", info.desiredK8sVersion, "status version", ng.Status.AtProvider.Version)
			}
		}
	}

	// Always requeue to update the upgrade status.
	return ctrl.Result{Requeue: true}, nil
}

func (u *EKSUpgrader) createNewVersionTiDBNodes(ctx context.Context, info *upgradeInfo, oldManagedNGs []*manualv1alpha1.NodeGroup, oldSelfManagedNGs []*autoscalingv1beta1.AutoScalingGroup) error {
	// There are three possible cases:
	// 1. from managed to self-managed
	// 2. from self-managed to self-managed
	// 3. from managed to managed
	if info.migrateNodeGroup || info.useSelfManaged {
		systemConfiguration, err := u.cfg.ClusterController.AWS.GetAWSUserdataConfigurationFromFile()
		if err != nil {
			info.logger.Error(err, "failed to get the userdata system configuration")
			return err
		}
		// only apply systemConfig to self-managed nodegroup, because the TiDB cluster nodes only use the self-managed nodegroup
		if err := createNewSelfManagedNodeGroups(ctx, u.infraCli, u.ssmClient, u.provider, info, oldManagedNGs, oldSelfManagedNGs, systemConfiguration); err != nil {
			info.logger.Error(err, "failed to create new version self-managed node groups")
			return err
		}
		info.logger.Info("created new version self-managed node groups")
	} else {
		if err := createNewManagedNodeGroups(ctx, u.infraCli, info, oldManagedNGs); err != nil {
			info.logger.Error(err, "failed to create new version managed node groups")
			return err
		}
		info.logger.Info("created new version managed node groups")
	}

	return nil
}

func (u *EKSUpgrader) deleteOldVersionTiDBNodes(ctx context.Context, eksCli client.Client, oldManagedNGs []*manualv1alpha1.NodeGroup, oldSelfManagedNGs []*autoscalingv1beta1.AutoScalingGroup) error {
	var errs *multierror.Error
	for _, ng := range oldManagedNGs {
		errs = multierror.Append(errs, deleteNodeGroupAndLaunchTemplate(ctx, u.infraCli, ng, ng.Name))
	}

	for _, ng := range oldSelfManagedNGs {
		errs = multierror.Append(errs, deleteNodeGroupAndLaunchTemplate(ctx, u.infraCli, ng, ng.Name))
	}

	return errs.ErrorOrNil()
}

func (u *EKSUpgrader) upgradeManagedNodeGroups(ctx context.Context, uo *cloudv1.UpgradeOperation, network *cloudv1.Network, cluster *cloudv1.Cluster, ngs *manualv1alpha1.NodeGroupList, nodeType cloudv1.UpgradeOperationTarget) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	if len(ngs.Items) == 0 {
		logger.Info("no node groups", "nodeType", nodeType)
		return ctrl.Result{}, FinishUpgrading(ctx, u.infraCli, uo)
	}

	desiredVersion := uo.Status.CurrentStep.Version
	eks, err := getEKSCR(ctx, u.infraCli, network)
	if err != nil {
		logger.Error(err, "unable to get the eks cluster CR")
		return ctrl.Result{}, err
	}
	currentVersion := eks.Status.AtProvider.Version

	if !readyForUpgradeNodeGroups(ctx, u.infraCli, network, desiredVersion, currentVersion) {
		logger.Info("it's not ready to upgrade the nodegroup, please check the eks cluster's version", "current version", currentVersion, "desired version", desiredVersion)
		return ctrl.Result{}, nil
	}

	done := 0
	for i := range ngs.Items {
		ng := &ngs.Items[i]
		logger.Info("node group", "nodeType", nodeType, "name", ng.Name, "spec version", ng.Spec.ForProvider.Version, "status version", ng.Status.AtProvider.Version, "status", ng.Status.AtProvider.Status)

		switch ng.Status.AtProvider.Status {
		case manualv1alpha1.NodeGroupStatusUpdating:
			LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "The node group is updating, wait for next reconcile", nil)

			// The version may be modified back by other controllers, so update it again.
			if err := updateNodeGroupVersion(ctx, u.infraCli, ng, desiredVersion); err != nil {
				logger.Error(err, "unable to upgrade the node group CR")
				return ctrl.Result{}, err
			}

		case manualv1alpha1.NodeGroupStatusActive:
			if ng.Status.AtProvider.Version == desiredVersion {
				if IsCRReadyAndSynced(ng.Status.Conditions) {
					done++
					LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, fmt.Sprintf("The node group %s is already upgraded", ng.Name), nil)
				} else {
					if err := updateNodeGroupVersion(ctx, u.infraCli, ng, desiredVersion); err != nil {
						logger.Error(err, "unable to upgrade the node group CR")
						return ctrl.Result{}, err
					}
				}
			} else {
				logger.Info("start to upgrade the node group")

				if err := updateNodeGroupVersion(ctx, u.infraCli, ng, desiredVersion); err != nil {
					logger.Error(err, "unable to upgrade the node group CR")
					return ctrl.Result{}, err
				}
			}
		default:
			logger.Info("the node group is in unexpected status", "status", ng.Status.AtProvider.Status)
		}
	}

	if done == len(ngs.Items) {
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, fmt.Sprintf("All node groups of %s are upgraded", nodeType), nil)
		return ctrl.Result{}, FinishUpgrading(ctx, u.infraCli, uo)
	}

	return ctrl.Result{Requeue: true}, nil
}

func (u *EKSUpgrader) UpgradeAddOns(ctx context.Context, uo *cloudv1.UpgradeOperation, network *cloudv1.Network) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	eksName := util.GetAnnotation(network, consts.AnnotationKeyDataplaneName)
	upgradedCount := 0
	listAddons := false
	var account cloudv1.Account
	installedAddons := make(map[string]struct{})

	for addon, version := range uo.Status.CurrentStep.Options {
		var cr v1alpha1.Addon
		crName := addonCRName(eksName, addon)
		if err := u.infraCli.Get(ctx, types.NamespacedName{Name: crName}, &cr); err != nil {
			if apierrors.IsNotFound(err) {
				// Before we create the addon CR, we need to check if the addon is already managed by AWS.
				// If so, we need to let AWS stop managing them.
				// Otherwise, addon CR will not be ready.
				if !listAddons {
					acct, ok := u.accountMap.Load(network.Spec.AccountRef.Name)
					if !ok {
						if e := u.infraCli.Get(ctx, types.NamespacedName{Name: network.Spec.AccountRef.Name}, &account); e != nil {
							logger.Error(e, "unable to get the account CR")
							return ctrl.Result{}, e
						}
						u.accountMap.Store(network.Spec.AccountRef.Name, account)
					} else {
						account = acct.(cloudv1.Account)
					}

					out, e := u.awsCli.EKS(aws.GetRoleFromAccount(&account)).ListAddons(ctx, &eks.ListAddonsInput{
						ClusterName: pointer.String(eksName),
					})
					if e != nil {
						logger.Error(e, "unable to list addons", "addon", addon)
						return ctrl.Result{}, e
					}
					logger.Info("list addons", "addons", out.Addons, "count", len(out.Addons))
					for _, name := range out.Addons {
						installedAddons[name] = struct{}{}
					}
					listAddons = true
				}

				if _, ok := installedAddons[addon]; ok {
					logger.Info("delete the addon", "addon", addon)
					if _, e := u.awsCli.EKS(aws.GetRoleFromAccount(&account)).DeleteAddon(ctx, &eks.DeleteAddonInput{
						AddonName:   pointer.String(addon),
						ClusterName: pointer.String(eksName),
						// Preserves the add-on software on the EKS cluster but AWS stops managing any settings for the add-on.
						// So that we can create the addon CR.
						Preserve: true,
					}); e != nil {
						logger.Error(e, "unable to delete the addon", "addon", addon)
						return ctrl.Result{}, e
					}
				}

				expected := getAddonCR(network, eksName, addon, version)
				if err := u.infraCli.Create(ctx, expected); err != nil {
					logger.Error(err, "unable to create the addon CR", "addon", addon)
					return ctrl.Result{}, err
				}
				logger.Info("created the addon CR", "addon", crName)
				continue
			}
			logger.Error(err, "unable to get the addon CR", "addon", addon)
			return ctrl.Result{}, err
		}

		if cr.Spec.ForProvider.AddonVersion != nil && *cr.Spec.ForProvider.AddonVersion == version {
			if IsCRReadyAndSynced(cr.Status.Conditions) && cr.Status.AtProvider.Status != nil && *cr.Status.AtProvider.Status == "ACTIVE" {
				upgradedCount++
				logger.Info("the addon CR is already upgraded", "addon", crName)
			} else {
				logger.Info("the addon CR is upgrading", "addon", crName, "status", cr.Status.AtProvider.Status)
			}
		} else {
			logger.Info("upgrade the addon CR", "addon", addon, "version", version)
			cr.Spec.ForProvider.AddonVersion = pointer.String(version)
			if err := u.infraCli.Update(ctx, &cr); err != nil {
				logger.Error(err, "unable to update the addon CR", "addon", addon)
				return ctrl.Result{}, err
			}
		}
	}

	if upgradedCount == len(uo.Status.CurrentStep.Options) {
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "All addons are upgraded", nil)
		return ctrl.Result{}, FinishUpgrading(ctx, u.infraCli, uo)
	}

	return ctrl.Result{Requeue: true}, nil
}