package zoneredundantevicter

import (
	"context"
	"fmt"
	"os"
	"sort"
	"time"

	"github.com/tidbcloud/aws-shared-provider/pkg/metrics"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/kubeapi"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/labels"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"
)

const (
	RetryTooManyRequestsTime     = 3 * time.Second
	LongUnreachableTime          = 6 * time.Minute
	EvictUnschedulablePodPVCTime = 6 * time.Minute
)

var (
	noEvictNewNodeDuration = 24 * time.Hour
)

func init() {
	if d, err := time.ParseDuration(os.Getenv("NoEvictNewNodeDuration")); err == nil {
		noEvictNewNodeDuration = d
	}
}

func ignoredPod(pod *corev1.Pod) bool {
	return pod.Labels[labels.ZREvicterIgnoredKey] != ""
}

func ignoredComponent(componentName string) bool {
	switch componentName {
	case "tikv", "tiflash-wn":
		return true
	default:
		return false
	}
}

func pvcEvictable(pod *corev1.Pod) bool {
	return pod.Labels[labels.ZREvicterPVCEvictableKey] != ""
}

func filterEvictNodeBySort(allNodes *corev1.NodeList, mainZone string) map[string][]*corev1.Node {
	var componentNodeMap = make(map[string][]*corev1.Node)
	for i := range allNodes.Items {
		node := &allNodes.Items[i]
		if zone, exist := node.Labels[labels.ServerlessZoneLabelKey]; !exist || zone == mainZone {
			continue
		}
		if node.DeletionTimestamp != nil {
			continue
		}
		if !kubeapi.IsNodeLongUnready(node, LongUnreachableTime) &&
			time.Since(node.CreationTimestamp.Time) < noEvictNewNodeDuration {
			continue
		}
		componentName := node.Labels[labels.ServerlessComponentNodeLabelKey]
		if componentName == "" {
			continue
		}
		if ignoredComponent(componentName) {
			continue
		}
		if (componentName == "general" || componentName == "critical-services") && !kubeapi.IsNodeLongUnready(node, LongUnreachableTime) {
			continue
		}
		componentNodeMap[componentName] = append(componentNodeMap[componentName], node)
	}

	for _, nodes := range componentNodeMap {
		sort.Slice(nodes, func(i, j int) bool {
			return nodes[i].CreationTimestamp.Time.Before(nodes[j].CreationTimestamp.Time)
		})
		// and unready nodes are in the front
		for target, i := 0, 0; i < len(nodes); i++ {
			if !kubeapi.IsNodeReady(nodes[i]) {
				nodes[target], nodes[i] = nodes[i], nodes[target]
				target++
			}
		}
	}
	return componentNodeMap
}

func (r *Reconciler) evictZoneRedundant(ctx context.Context, mainZone string) (err error) {
	logger := log.FromContext(ctx)

	if err = r.evictUnschedulablePodPVC(ctx); err != nil {
		logger.Error(err, "failed to evictUnschedulablePodPVC, skip")
	}

	if r.ProviderType == types.ProviderTypeAliCloudFreeTier {
		return
	}

	allNodes := &corev1.NodeList{}
	if err = r.Client.List(ctx, allNodes); err != nil {
		logger.Error(err, "failed to list all nodes")
		return
	}
	if err = r.deleteMainZoneEvicterTaint(ctx, allNodes, mainZone); err != nil {
		logger.Error(err, "failed to delete evicter taint in main zone")
		return
	}
	evictNodeMap := filterEvictNodeBySort(allNodes, mainZone) // not in mainZone

	if err = r.evictCoreComponents(ctx, evictNodeMap); err == nil {
		err = r.evictOther(ctx, evictNodeMap)
	}
	return
}

func (r *Reconciler) deleteMainZoneEvicterTaint(ctx context.Context, allNodes *corev1.NodeList, mainZone string) error {
	for i := range allNodes.Items {
		node := &allNodes.Items[i]
		if node.DeletionTimestamp != nil {
			continue
		}
		if node.Labels[labels.ServerlessZoneLabelKey] != mainZone {
			continue
		}
		var deleted bool
		newTaints := make([]corev1.Taint, 0, len(node.Spec.Taints))
		for i := range node.Spec.Taints {
			if node.Spec.Taints[i].Key == types.ZoneRedundantEvicterTaintKey {
				deleted = true
				continue
			}
			newTaints = append(newTaints, node.Spec.Taints[i])
		}
		if deleted {
			if err := kubeapi.UpdateNode(ctx, r.Client, func() *corev1.Node {
				node.Spec.Taints = newTaints
				return node
			}); err != nil {
				return err
			}
			log.FromContext(ctx).Info("delete zr evicter taint", "nodeName", node.Name)
			metrics.ZRNodeEvicterTaintUpdateCounter.WithLabelValues(node.Name, "del").Inc()
		}
	}
	return nil
}

func (r *Reconciler) evictUnschedulablePodPVC(ctx context.Context) error {
	var podList corev1.PodList
	if err := r.Client.List(ctx, &podList, client.MatchingFields{"status.unschedulable": "unschedulable"}); err != nil {
		return err
	}
	for i := range podList.Items {
		pod := &podList.Items[i]
		if !ignoredPod(pod) && time.Since(pod.CreationTimestamp.Time) > EvictUnschedulablePodPVCTime {
			if err := r.evictPodPVC(ctx, pod); err != nil {
				return err
			}
		}
	}
	return nil
}

func (r *Reconciler) evictCoreComponents(ctx context.Context, evictNodeMap map[string][]*corev1.Node) error {
	retryTidb, err := r.evictZoneRedundantTidbPods(ctx, evictNodeMap["tidb"])
	if err != nil {
		return err
	}
	// tidb-gateway-privatelink is in critical-services component.
	wait, retry, err := r.evictGatewayPrivatelink(ctx, evictNodeMap["critical-services"])
	if err != nil {
		return err
	}
	if wait {
		return &triggerAfterError{triggerAfter: r.CheckInterval}
	}
	if retryTidb || retry {
		return &triggerAfterError{triggerAfter: RetryTooManyRequestsTime}
	}
	return nil
}

func (r *Reconciler) evictGatewayPrivatelink(ctx context.Context, gatewayNodes []*corev1.Node) (bool, bool, error) {
	return r.evictComponent(ctx, "critical-services", gatewayNodes, map[string]string{
		"control-plane": "tidb-gateway-privatelink-zr",
	})
}

func (r *Reconciler) evictOther(ctx context.Context, evictNodeMap map[string][]*corev1.Node) error {
	var wait, retry bool
	for component, nodes := range evictNodeMap {
		_wait, _retry, _ := r.evictComponent(ctx, component, nodes, nil)
		wait = wait || _wait
		retry = retry || _retry
	}
	if wait {
		return &triggerAfterError{triggerAfter: r.CheckInterval}
	}
	if retry {
		return &triggerAfterError{triggerAfter: RetryTooManyRequestsTime}
	}
	return nil
}

func (r *Reconciler) trySetEvicterTaint(ctx context.Context, node *corev1.Node) (updated bool, err error) {
	if !kubeapi.IsNodeReady(node) || kubeapi.IsNodeUnreachable(node) {
		return false, nil
	}

	for i := range node.Spec.Taints {
		taint := &node.Spec.Taints[i]
		if taint.Key == types.ZoneRedundantEvicterTaintKey && taint.Effect == corev1.TaintEffectNoSchedule {
			return
		}
	}
	if err = kubeapi.UpdateNode(ctx, r.Client, func() *corev1.Node {
		node.Spec.Taints = append(node.Spec.Taints, corev1.Taint{
			Key:    types.ZoneRedundantEvicterTaintKey,
			Value:  fmt.Sprint(time.Now().Unix()),
			Effect: corev1.TaintEffectNoSchedule,
		})
		return node
	}); err == nil {
		updated = true
		metrics.ZRNodeEvicterTaintUpdateCounter.WithLabelValues(node.Name, "set").Inc()
	}
	return
}

func podName(pod *corev1.Pod) string {
	return pod.Namespace + "/" + pod.Name
}

func isControlledByNonJob(ownerRef *metav1.OwnerReference) bool {
	return ownerRef != nil && ownerRef.Kind != "Job"
}