package gke

import (
	"context"
	"fmt"
	"time"

	gkev1beta1 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	gkev1beta2 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta2"
	"go.uber.org/multierr"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	"github.com/tidbcloud/infra-provider/pkg/cloud/gcp"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/acscli"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/clustercli"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	. "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/upgradeoperation/common"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/appbox"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/upgradeoperation"
)

type upgrader struct {
	infraCli      client.Client
	gcpClient     gcp.Client
	eventRecorder record.EventRecorder
	uoWriter      upgradeoperation.Writer
	appBoxWriter  appbox.APPBoxWriter
	clusterCli    clustercli.ClusterClient
	appboxConfig  config.APPBoxConfig
	acsCli        acscli.Interface
}

var _ Upgrader = &upgrader{}

func NewUpgrader(infraCli client.Client, gcpClient gcp.Client, clusterCli clustercli.ClusterClient, eventRecorder record.EventRecorder, appboxConfig config.APPBoxConfig, acsCli acscli.Interface) Upgrader {
	return &upgrader{
		uoWriter:      upgradeoperation.NewWriter(infraCli),
		appBoxWriter:  appbox.NewAPPBoxWriter(infraCli),
		infraCli:      infraCli,
		gcpClient:     gcpClient,
		eventRecorder: eventRecorder,
		clusterCli:    clusterCli,
		appboxConfig:  appboxConfig,
		acsCli:        acsCli,
	}
}

// UpgradeControlPlane upgrades the control plane of a GKE cluster.
// Since the crossplane does not support upgrading the control plane via CR, we need to upgrade the control plane via gcloud sdk.
// See: https://github.com/crossplane-contrib/provider-gcp/issues/479
func (u *upgrader) UpgradeControlPlane(ctx context.Context, uo *v1alpha1.UpgradeOperation, appBox *v1alpha1.APPBox) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	logger.Info("UpgradeControlPlane")

	desiredVersion := uo.Status.CurrentStep.Version
	gke, err := getGKECR(ctx, u.infraCli, appBox)
	if err != nil {
		logger.Error(err, "unable to get the gke cluster CR")
		return ctrl.Result{}, err
	}
	logger.Info("get gke cluster CR", "gke", gke.Name, "current version", gke.Status.AtProvider.CurrentMasterVersion, "status", gke.Status.AtProvider.Status)

	switch gke.Status.AtProvider.Status {
	case gkev1beta2.ClusterStateReconciling:
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "The GKE cluster is reconciling, wait for the next reconcile", nil)
		return ctrl.Result{RequeueAfter: 1 * time.Minute}, nil

	case gkev1beta2.ClusterStateRunning:
		currentCPVersion := FormatMajorMinorVersion(gke.Status.AtProvider.CurrentMasterVersion)

		if currentCPVersion == desiredVersion {
			LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "The GKE cluster's control plane is already upgraded", nil)
			return FinishUpgradingResult(ctx, u.infraCli, u.uoWriter, uo)
		} else if ExpectToBeUpgraded(desiredVersion, currentCPVersion) {
			if appBox.Spec.KubernetesConfig.Version != desiredVersion {
				appBox.Spec.KubernetesConfig.Version = desiredVersion
				if err := u.appBoxWriter.Update(ctx, appBox); err != nil {
					logger.Error(err, "unable to upgrade appbox CR")
				}
			}

			projectID := getGCPProjectID(appBox.Status.IAM.ServiceAccountEmail)
			if projectID == "" {
				return ctrl.Result{}, fmt.Errorf("failed to get the GCP project ID from the service account email: %s", appBox.Status.IAM.ServiceAccountEmail)
			}
			selectedVersion, err := u.gcpClient.UpgradeCluster(ctx, projectID, appBox.Spec.Region, gke.Name, desiredVersion)
			if err != nil {
				logger.Error(err, "unable to upgrade gke cluster")
				return ctrl.Result{}, err
			}

			// It will take about 25 minutes to upgrade the control plane.
			LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, fmt.Sprintf("Start to upgrade the GKE cluster's control plane to %s", selectedVersion), nil)
			return ctrl.Result{RequeueAfter: 20 * time.Minute}, nil
		} else {
			LogAndEmitEvent(logger, u.eventRecorder, uo, WarningEvent, fmt.Sprintf("The GKE cluster is in unexpected version: %s", gke.Status.AtProvider.CurrentMasterVersion), nil)
		}

	default:
		LogAndEmitEvent(logger, u.eventRecorder, uo, WarningEvent, fmt.Sprintf("The GKE cluster is in unexpected status: %s", gke.Status.AtProvider.Status), nil)
	}

	return ctrl.Result{}, nil
}

func (u *upgrader) UpgradeAdminNodes(ctx context.Context, uo *v1alpha1.UpgradeOperation, appBox *v1alpha1.APPBox) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	logger.Info("UpgradeAdminNodes")

	gke, err := getGKECR(ctx, u.infraCli, appBox)
	if err != nil {
		logger.Error(err, "unable to get the gke cluster CR")
		return ctrl.Result{}, err
	}
	logger.Info("get gke cluster CR", "gke", gke.Name, "current version", gke.Status.AtProvider.CurrentMasterVersion, "status", gke.Status.AtProvider.Status)

	adminNPName := getAdminNodePoolName(gke)
	if adminNPName == "" {
		msg := "failed to get admin nodepool name from gke CR"
		LogAndEmitEvent(logger, u.eventRecorder, uo, WarningEvent, msg, nil)
		return ctrl.Result{}, uo.Fail(ctx, u.infraCli, msg)
	}

	np, err := fetchNodePool(ctx, u.infraCli, adminNPName)
	if err != nil {
		LogAndEmitEvent(logger, u.eventRecorder, uo, WarningEvent, "failed to get admin node pool", err)
		return ctrl.Result{}, fmt.Errorf("failed to get admin node pool %s: %w", adminNPName, err)
	}

	return u.upgradeNodePools(ctx, uo, appBox, []nodePool{np}, v1alpha1.AdminNodes)
}

func (u *upgrader) UpgradeDataNodes(ctx context.Context, uo *v1alpha1.UpgradeOperation, appBox *v1alpha1.APPBox) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	logger.Info("UpgradeDataNodes")

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

	// Check if the control plane has been upgraded to the desired version.
	if !isControlPlaneUpgraded(gke, desiredVersion) {
		LogAndEmitEvent(logger, u.eventRecorder, uo, WarningEvent, ControlPlaneNotUpgraded, nil)
		return ctrl.Result{}, uo.Fail(ctx, u.infraCli, ControlPlaneNotUpgraded)
	}

	msg, done, err := UpgradeDataNodes(ctx, u.clusterCli, uo, desiredVersion)
	if err != nil {
		LogAndEmitEvent(logger, u.eventRecorder, uo, WarningEvent, "failed to upgrade data nodes", err)
		return ctrl.Result{}, err
	}

	if done {
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "tidb nodes are already upgraded", nil)
		return FinishUpgradingResult(ctx, u.infraCli, u.uoWriter, uo)
	}

	uo.Status.Reason = msg
	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, msg, nil)
	return ctrl.Result{Requeue: true}, u.uoWriter.UpdateStatus(ctx, uo)
}

// UpgradeEmptyNodePools handles node pools that do not have nodes.
// Because every time we upgrade tidb node pools, those old version node pools will not be deleted.
// So this function checks if there are empty node pools and upgrade or delete them directly.
func (u *upgrader) UpgradeEmptyNodePools(ctx context.Context, uo *v1alpha1.UpgradeOperation, appBox *v1alpha1.APPBox) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	logger.Info("UpgradeEmptyNodePools")

	gke, err := getGKECR(ctx, u.infraCli, appBox)
	if err != nil {
		logger.Error(err, "unable to get the gke cluster CR")
		return ctrl.Result{}, err
	}

	// Check all node pools that do not have nodes, upgrade them directly.
	var emptyNodePools []nodePool
	for _, nps := range gke.Status.AtProvider.NodePools {
		if nps == nil || nps.Config == nil {
			continue
		}
		if nps.Config.Labels["cluster"] != "shared" && nps.Config.Labels["tags.tidbcloud.com/cluster"] != "shared" {
			continue
		}

		np, err := fetchNodePool(ctx, u.infraCli, nps.Name)
		if err != nil {
			logger.Error(err, "unable to get the node pool CR", "name", nps.Name)
			continue
		}
		size, err := getNodeCount(ctx, u.gcpClient, np)
		if err != nil {
			logger.Error(err, "unable to get the node count", "node pool", np.GetName())
			continue
		}
		if size == 0 {
			logger.Info("node pool does not have nodes", "node pool", np.GetName(), "version", np.GetVersion())
			emptyNodePools = append(emptyNodePools, np)
		}
	}

	// Delete empty node pools if the delete option is set.
	desiredVersion := uo.Status.CurrentStep.Version
	if uo.Status.CurrentStep != nil && uo.Status.CurrentStep.Options["delete"] == "true" {
		LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Start to delete empty node pools", nil)
		for _, np := range emptyNodePools {
			// Only delete node pools whose version is lower than the desired version.
			lower, err := IsVersionLower(FormatMajorMinorVersion(np.GetVersion()), desiredVersion)
			if err != nil {
				logger.Error(err, "skip deleting empty node pool due to invalid version comparison",
					"name", np.GetName(), "version", np.GetVersion(), "desiredVersion", desiredVersion)
				continue
			}
			if !lower {
				logger.Info("skip deleting the empty node pool because its version is not lower than the desired version",
					"name", np.GetName(), "version", np.GetVersion(), "desiredVersion", desiredVersion)
				continue
			}
			logger.Info("start to delete the empty node pool", "name", np.GetName(), "version", np.GetVersion())
			if err := np.Delete(ctx, u.infraCli); err != nil {
				logger.Error(err, "unable to delete the empty node pool", "name", np.GetName())
			}
		}
		return FinishUpgradingResult(ctx, u.infraCli, u.uoWriter, uo)
	}

	LogAndEmitEvent(logger, u.eventRecorder, uo, NormalEvent, "Start to upgrade empty node pools", nil)
	return u.upgradeNodePools(ctx, uo, appBox, emptyNodePools, v1alpha1.EmptyNodePools)
}

// 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 *v1alpha1.UpgradeOperation, appBox *v1alpha1.APPBox, nps []nodePool, nodeType v1alpha1.UpgradeOperationTarget) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	logger.Info("upgradeNodePools")

	desiredVersion := uo.Status.CurrentStep.Version
	gke, err := getGKECR(ctx, u.infraCli, appBox)
	if err != nil {
		logger.Error(err, "unable to get the gke cluster CR")
		return ctrl.Result{}, err
	}
	if !isControlPlaneUpgraded(gke, desiredVersion) {
		LogAndEmitEvent(logger, u.eventRecorder, uo, WarningEvent, ControlPlaneNotUpgraded, nil)
		return ctrl.Result{}, uo.Fail(ctx, u.infraCli, ControlPlaneNotUpgraded)
	}

	// If the gke cluster is reconciling, we should wait for next reconcile.
	if gke.Status.AtProvider.Status == gkev1beta2.ClusterStateReconciling {
		logger.Info("gke cluster is reconciling, wait for next reconcile")
		return ctrl.Result{RequeueAfter: 2 * time.Minute}, nil
	}
	if gke.Status.AtProvider.Status != gkev1beta2.ClusterStateRunning {
		return ctrl.Result{}, fmt.Errorf("gke cluster %s is in unexpected status %s", gke.Name, gke.Status.AtProvider.Status)
	}

	// npSet is a set of node pool names to be upgraded.
	npSet := make(map[string]nodePool, len(nps))
	for _, np := range nps {
		npSet[np.GetName()] = np
	}

	toBeUpgradedNPs := make([]string, 0, len(npSet))

	// gke CR's status stores the current status of node pools.
	for _, npStatus := range gke.Status.AtProvider.NodePools {
		if npStatus == nil {
			continue
		}
		if _, ok := npSet[npStatus.Name]; !ok {
			continue
		}
		if npStatus.Version == gke.Status.AtProvider.CurrentMasterVersion && npStatus.Status == gkev1beta1.NodePoolStateRunning {
			logger.Info("node pool is already upgraded", "nodeType", nodeType, "node pool", npStatus.Name, "version", npStatus.Version)
		} else {
			toBeUpgradedNPs = append(toBeUpgradedNPs, npStatus.Name)
		}
	}

	// When there are no node pools left to be upgraded, we can finish the upgrading.
	if len(toBeUpgradedNPs) == 0 {
		logger.Info(fmt.Sprintf("all node pools of %s are upgraded", nodeType))
		return FinishUpgradingResult(ctx, u.infraCli, u.uoWriter, uo)
	}

	projectID := getGCPProjectID(appBox.Status.IAM.ServiceAccountEmail)
	if projectID == "" {
		return ctrl.Result{}, fmt.Errorf("failed to get the GCP project ID from the service account email: %s", appBox.Status.IAM.ServiceAccountEmail)
	}

	logger.Info("node pools to be upgraded", "nodeType", nodeType, "node pools", toBeUpgradedNPs)
	var applyErrs []error
	for _, npName := range toBeUpgradedNPs {
		np := npSet[npName]
		logger.Info("start to upgrade the node pool", "nodeType", nodeType, "name", npName, "version", np.GetVersion())

		// Upgrade the node pool via gcloud sdk.
		// Note: we can not upgrade the node pool via CR.
		if err := u.gcpClient.UpgradeNodePool(ctx, projectID, appBox.Spec.Region, gke.Name, npName, gke.Status.AtProvider.CurrentMasterVersion); err != nil {
			logger.Error(err, "unable to upgrade the node pool via sdk", "nodeType", nodeType, "name", npName)
			applyErrs = append(applyErrs, err)
			continue
		}

		// We also need to update the node pool CR's version in spec.
		np.SetVersion(gke.Status.AtProvider.CurrentMasterVersion)
		if err := np.Apply(ctx, u.infraCli); err != nil {
			logger.Error(err, "unable to update the node pool CR", "name", npName)
			applyErrs = append(applyErrs, err)
			continue
		}
	}

	if len(applyErrs) > 0 {
		return ctrl.Result{}, multierr.Combine(applyErrs...)
	}
	return ctrl.Result{RequeueAfter: 2 * time.Minute}, nil
}

func (u *upgrader) UpgradeAddons(ctx context.Context, uo *v1alpha1.UpgradeOperation, appBox *v1alpha1.APPBox) (ctrl.Result, error) {
	return ctrl.Result{}, nil
}

func (u *upgrader) UpgradeO11yNodes(ctx context.Context, uo *v1alpha1.UpgradeOperation, appBox *v1alpha1.APPBox) (ctrl.Result, error) {
	return DoUpgradeO11yNodes(ctx, u.infraCli, u.acsCli, u.uoWriter, u.eventRecorder, uo, appBox)
}