package upgradeoperation

import (
	"context"
	"fmt"
	"time"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/client-go/tools/record"
	"k8s.io/client-go/util/workqueue"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	. "github.com/tidbcloud/eks-provider/controllers/upgradeoperation/common"
	"github.com/tidbcloud/eks-provider/controllers/upgradeoperation/eks"
	"github.com/tidbcloud/eks-provider/controllers/upgradeoperation/gke"
	"github.com/tidbcloud/eks-provider/controllers/utils/ticdcservicecli"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/cloud/aws"
	"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"
	utiltime "github.com/tidbcloud/infra-provider/pkg/util/time"
)

const (
	defaultBaseDelay = 5 * time.Second
	defaultMaxDelay  = 2 * time.Minute
)

// Reconciler reconciles a UpgradeOperation object and upgrades the k8s version of the control plane and nodes.
type Reconciler struct {
	client         client.Client
	cache          client.Reader
	upgrader       Upgrader
	clock          utiltime.Clock
	eventRecorder  record.EventRecorder
	cachedProvider dataplane.EnvProviderManager
	cfg            *config.RootConfig
}

func NewReconciler(cachedProvider dataplane.EnvProviderManager, client client.Client, cache client.Reader, awsCli aws.Client, gcpCli gcp.Client, cfg *config.RootConfig, clock utiltime.Clock, eventRecorder record.EventRecorder) *Reconciler {
	var upgrader Upgrader
	ticdcCli := ticdcservicecli.NewClient(cfg.GetTiCDCServiceEndpoint())
	switch cfg.Predicate.MatchLabels[cloudv1.ProviderTypeLabel] {
	case consts.ProviderTypeEKS:
		upgrader = eks.NewUpgrader(cfg, client, awsCli, eventRecorder, ticdcCli)
	case consts.ProviderTypeGKE:
		upgrader = gke.NewUpgrader(cfg, client, gcpCli, eventRecorder, ticdcCli)
	}

	return &Reconciler{
		cfg:            cfg,
		client:         client,
		cache:          cache,
		upgrader:       upgrader,
		clock:          clock,
		eventRecorder:  eventRecorder,
		cachedProvider: cachedProvider,
	}
}

func (r *Reconciler) SetupWithManager(mgr ctrl.Manager, pds ...predicate.Predicate) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&cloudv1.UpgradeOperation{}, builder.WithPredicates(pds...)).
		WithOptions(
			controller.Options{
				MaxConcurrentReconciles: 1,
				RateLimiter:             workqueue.NewTypedItemExponentialFailureRateLimiter[ctrl.Request](defaultBaseDelay, defaultMaxDelay),
			}).
		Complete(r)
}

func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (res ctrl.Result, _ error) {
	logger := log.FromContext(ctx)
	logger.Info("Reconciling UpgradeOperation")
	defer func() { logger.Info("Finish reconcile", "Requeue", res.Requeue, "RequeueAfter", res.RequeueAfter) }()

	var uo cloudv1.UpgradeOperation
	if err := r.cache.Get(ctx, req.NamespacedName, &uo); err != nil {
		if apierrors.IsNotFound(err) {
			logger.Info("UpgradeOperation is not found, may has been deleted")
		} else {
			LogAndEmitEvent(logger, r.eventRecorder, &uo, WarningEvent, "Failed to get the UpgradeOperation CR", err)
		}
		return ctrl.Result{}, nil
	}

	// Handle the deletion reconciliation loop
	if uo.GetDeletionTimestamp() != nil {
		if uo.Status.Phase == cloudv1.UpgradeOperationPhaseRunning {
			// We need to wait for the UpgradeOperation to be completed or failed before deleting it
			logger.Info("wait for UpgradeOperation to complete/fail before deleting it")
			// Will go to `handleRunning`
		} else {
			updated := controllerutil.RemoveFinalizer(&uo, cloudv1.UpgradeOperationFinalizer)
			if err := r.client.Update(ctx, &uo); err != nil {
				logger.Error(err, "failed to remove the finalizer from UpgradeOperation")
				return ctrl.Result{}, err
			}
			logger.Info("remove the finalizer from UpgradeOperation", "updated", updated)
			return ctrl.Result{}, nil
		}
	}

	if uo.Spec.Paused {
		LogAndEmitEvent(logger, r.eventRecorder, &uo, NormalEvent, "UpgradeOperation is paused, will not process it", nil)
		return ctrl.Result{}, nil
	}

	switch uo.Status.Phase {
	case cloudv1.UpgradeOperationPhaseNone:
		uo.Status.Phase = cloudv1.UpgradeOperationPhasePending
		uo.Status.FinishedSteps = cloudv1.NewSteps()
		uo.Status.NextSteps = cloudv1.NewSteps()
		return ctrl.Result{}, r.client.Status().Update(ctx, &uo)

	case cloudv1.UpgradeOperationPhasePending:
		return r.handlePending(ctx, &uo)

	case cloudv1.UpgradeOperationPhaseRunning:
		return r.handleRunning(ctx, &uo)

	case cloudv1.UpgradeOperationPhaseCompleted, cloudv1.UpgradeOperationPhaseFailed:
		return r.handleCompletedOrFailed(ctx, &uo)
	default:
		logger.Info("unknown phase, ignore it")
		return ctrl.Result{}, nil
	}
}

func (r *Reconciler) handlePending(ctx context.Context, uo *cloudv1.UpgradeOperation) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	// We have already checked the `schedule` field in the webhook, so we can ignore the error here.
	schedule, _ := time.Parse(cloudv1.UpgradeOperationScheduleLayout, uo.Spec.Schedule)
	now := r.clock.Now()
	if now.After(schedule) {
		uo.Status.NextSteps = uo.Spec.Steps
		uo.Status.Phase = cloudv1.UpgradeOperationPhaseRunning
		uo.Status.Reason = "Begin to upgrade"
		LogAndEmitEvent(logger, r.eventRecorder, uo, NormalEvent, "The schedule is reached", nil)
	} else {
		if uo.Status.Reason == ReasonPending {
			interval := schedule.Sub(now)
			LogAndEmitEvent(logger, r.eventRecorder, uo, NormalEvent, "Wait for the schedule", nil)
			return ctrl.Result{RequeueAfter: interval}, nil
		} else {
			uo.Status.Reason = ReasonPending
		}
	}

	return ctrl.Result{}, r.client.Status().Update(ctx, uo)
}

func (r *Reconciler) handleRunning(ctx context.Context, uo *cloudv1.UpgradeOperation) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	if uo.Status.CurrentStep == nil {
		uo.UpdateCurrentStep(r.clock.Now())
		if uo.Status.Phase == cloudv1.UpgradeOperationPhaseCompleted {
			if err := UnpauseCluster(ctx, r.client, uo); err != nil {
				logger.Error(err, "failed to unpause network and cluster")
				return ctrl.Result{}, err
			}
			LogAndEmitEvent(logger, r.eventRecorder, uo, NormalEvent, "Unpaused the cluster", nil)
		}
		return ctrl.Result{}, r.client.Status().Update(ctx, uo)
	}

	ep, cluster, err := r.getEnvProviderAndCluster(ctx, uo)
	if err != nil {
		logger.Error(err, "failed to get related env provider and cluster")
		return ctrl.Result{}, err
	}
	updated, err := PauseCluster(ctx, r.client, cluster)
	if err != nil {
		logger.Error(err, "failed to pause the cluster")
		return ctrl.Result{}, err
	}
	if updated {
		LogAndEmitEvent(logger, r.eventRecorder, uo, NormalEvent, "Paused the cluster", nil)
	}

	switch target := uo.Status.CurrentStep.Target; target {
	case cloudv1.CDCNodes:
		return r.upgrader.UpgradeCDCNodes(ctx, uo, ep, cluster)

	case cloudv1.DMNodes:
		return r.upgrader.UpgradeDMNodes(ctx, uo, ep, cluster)

	case cloudv1.TiDBNodes:
		return r.upgrader.UpgradeTiDBNodes(ctx, uo, ep, cluster)
	}

	logger.Info("should not reach here")
	return ctrl.Result{}, nil
}

func (r *Reconciler) handleCompletedOrFailed(ctx context.Context, uo *cloudv1.UpgradeOperation) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	now := r.clock.Now()
	var timeToDelete time.Time
	if uo.Status.EndTime != nil {
		timeToDelete = uo.Status.EndTime.Add(uo.Spec.TTL.Duration)
	} else if uo.Status.BeginTime != nil {
		timeToDelete = uo.Status.BeginTime.Add(uo.Spec.TTL.Duration)
	} else {
		timeToDelete = uo.GetCreationTimestamp().Add(uo.Spec.TTL.Duration)
	}

	logger.WithValues("TTL", uo.Spec.TTL.Duration, "now", now, "timeToDelete", timeToDelete)

	if now.After(timeToDelete) {
		logger.Info("UpgradeOperation can be deleted")
		return ctrl.Result{}, r.client.Delete(ctx, uo)
	}

	interval := timeToDelete.Sub(now)
	LogAndEmitEvent(logger, r.eventRecorder, uo, NormalEvent, "Wait for the UpgradeOperation to be deleted", nil)
	return ctrl.Result{RequeueAfter: interval}, nil
}

func (r *Reconciler) getEnvProviderAndCluster(ctx context.Context, uo *cloudv1.UpgradeOperation) (dataplane.EnvProvider, *cloudv1.Cluster, error) {
	if uo.Status.CurrentStep == nil || uo.Status.CurrentStep.Cluster == "" {
		return nil, nil, fmt.Errorf("the current step is not set or the cluster is not set")
	}

	var cluster cloudv1.Cluster
	if err := r.cache.Get(ctx, client.ObjectKey{Namespace: uo.Namespace, Name: uo.Status.CurrentStep.Cluster}, &cluster); err != nil {
		if apierrors.IsNotFound(err) {
			if err := uo.Fail(ctx, r.client, fmt.Sprintf("The cluster CR %s/%s is not found", uo.Namespace, uo.Spec.NetworkName)); err != nil {
				return nil, nil, fmt.Errorf("failed to update UpgradeOperation status, err: %w", err)
			}
		}
		return nil, nil, fmt.Errorf("failed to get cluster %s/%s, err: %w", uo.Namespace, uo.Status.CurrentStep.Cluster, err)
	}

	ep, err := r.cachedProvider.GetEnvProvider(ctx, &cluster)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to create dataplane provider, err: %w", err)
	}

	return ep, &cluster, nil
}