package node_controller

import (
	"context"
	"fmt"
	"reflect"
	"regexp"
	"strings"
	"time"

	ec2api "github.com/aws/aws-sdk-go-v2/service/ec2"
	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	"github.com/go-logr/logr"
	"github.com/pingcap/errors"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/pkg/gardener"
	metricsctrl "github.com/tidbcloud/infra-api/pkg/metrics/controller"
	"github.com/tidbcloud/infra-api/pkg/pingcap/label"
	"github.com/tidbcloud/infra-api/pkg/providers/aws"
	"github.com/tidbcloud/infra-api/pkg/providers/gcp"
)

const (
	machineHighPriority    = "4"
	machineDefaultPriority = "3"

	machinePriorityKey = "machinepriority.machine.sapcloud.io"

	protectionKey      = "tidbcloud.com/node-protection"
	protectionEnabled  = "enabled"
	protectionDisabled = "disabled"
	protectionUnknown  = "unknown"

	tidbclusterName = "tidb-cluster"
)

var (
	gcpInstanceIdRegx = regexp.MustCompile(`gce://(.+)/(.+)/(.+)`)
)

type NodeProtectionEditor interface {
	EditNodeProtection(node *corev1.Node, deletionProtection bool) error
}

type NoopProtectionEditor struct{}

func (p *NoopProtectionEditor) EditNodeProtection(_ *corev1.Node, _ bool) error {
	return nil
}

type AWSNodeProtectionEditor struct {
	awsAPI aws.Client
	region string
}

func (p *AWSNodeProtectionEditor) EditNodeProtection(node *corev1.Node, deletionProtection bool) error {
	providerId := node.Spec.ProviderID
	if providerId == "" {
		return errors.New("Empty providerID")
	}

	instanceID := providerId[strings.LastIndex(providerId, "/")+1:]

	_, err := p.awsAPI.Ec2().ModifyInstanceAttribute(context.TODO(), &ec2api.ModifyInstanceAttributeInput{
		InstanceId: pointer.String(instanceID),
		DisableApiTermination: &ec2types.AttributeBooleanValue{
			Value: pointer.Bool(deletionProtection),
		},
	}, aws.WithEC2Region(p.region))
	return err
}

type GCPNodeProtectionEditor struct {
	gcpAPI gcp.GCPAPI
}

func (p *GCPNodeProtectionEditor) EditNodeProtection(node *corev1.Node, deletionProtection bool) error {
	providerID := node.Spec.ProviderID
	if providerID == "" {
		return errors.New("Empty providerID")
	}

	project, zone, nodeName, err := parseInstanceID(providerID)
	if err != nil {
		return errors.Annotate(err, "ParseGCPInstanceID")
	}

	return p.gcpAPI.SetDeleteProtection(project, zone, nodeName, deletionProtection)
}

func parseInstanceID(instance string) (project, zone, nodeName string, err error) {
	// e.g.: gce://pingcap-gardener/us-west1-a/gke-dev-seed-us-west-gardener-node-po-1cbd90b1-pbf6
	matches := gcpInstanceIdRegx.FindStringSubmatch(instance)
	if len(matches) != 4 {
		err = fmt.Errorf("invalid provider id %s", instance)
		return
	}
	project = matches[1]
	zone = matches[2]
	nodeName = matches[3]
	return
}

// CLusterNodeProtector protects the storage node inuse
type ClusterNodeProtector struct {
	InfraClient client.Client

	log         logr.Logger
	shootClient client.Client

	protectionEditor NodeProtectionEditor
}

func (c *ClusterNodeProtector) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	start := time.Now()
	defer func() {
		metricsctrl.ReconcileDuration.WithLabelValues("node_protector").Observe(time.Since(start).Seconds())
	}()
	node := &corev1.Node{}
	err := c.shootClient.Get(ctx, req.NamespacedName, node)
	if err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}
	if _, ok := node.Labels[gardener.NodeGroupSetOwnerKey]; !ok {
		// node is not managed by any nodegroup, shortcut to skip
		return ctrl.Result{}, nil
	}
	podList := &corev1.PodList{}
	err = c.shootClient.List(ctx, podList, client.MatchingFields(map[string]string{
		"spec.nodeName": node.Name,
	}))
	if err != nil {
		return ctrl.Result{}, errors.AddStack(err)
	}
	inUse := false
	usedByStorage := false
	for _, pod := range podList.Items {
		if component, ok := getPodComponentType(&pod); ok {
			inUse = true
			if *component == cloudv1.ComponentTiKV || *component == cloudv1.ComponentTiFlash {
				usedByStorage = true
			}
			break
		}
	}

	// set machine priority, so that the node that is not used by any cluster component is scaled in first
	err = c.mutateNode(ctx, node, func(n *corev1.Node) error {
		if node.Annotations == nil {
			node.Annotations = map[string]string{}
		}
		if inUse {
			// increase machine priority of that node
			node.Annotations[machinePriorityKey] = machineHighPriority
		} else {
			// set the priority back to default if the node is not in use
			node.Annotations[machinePriorityKey] = machineDefaultPriority
		}
		return nil
	})
	if err != nil {
		return ctrl.Result{}, errors.AddStack(err)
	}

	if err := c.processNodeProtection(ctx, node, usedByStorage); err != nil {
		return ctrl.Result{}, errors.Annotate(err, "processNodeProtection")
	}

	// TODO: add node protection
	return ctrl.Result{}, nil
}

func (c *ClusterNodeProtector) processNodeProtection(ctx context.Context, node *corev1.Node, usedByStorage bool) error {
	// load state
	state, processed := node.Annotations[protectionKey]
	enabled := state == protectionEnabled
	disabled := state == protectionDisabled

	switch {
	case usedByStorage && !enabled:
		err := c.transProtectionState(ctx, node, protectionUnknown)
		if err != nil {
			return err
		}
		err = c.protectionEditor.EditNodeProtection(node, true)
		if err != nil {
			return err
		}
		err = c.transProtectionState(ctx, node, protectionEnabled)
		if err != nil {
			return err
		}
	// processed but not disabled, then there might be delete protection in the node that we did not aware
	case !usedByStorage && !disabled && processed:
		err := c.transProtectionState(ctx, node, protectionUnknown)
		if err != nil {
			return err
		}
		err = c.protectionEditor.EditNodeProtection(node, false)
		if err != nil {
			return err
		}
		err = c.transProtectionState(ctx, node, protectionDisabled)
		if err != nil {
			return err
		}
	}
	return nil
}

func (c *ClusterNodeProtector) transProtectionState(ctx context.Context, node *corev1.Node, state string) error {
	return c.mutateNode(ctx, node, func(n *corev1.Node) error {
		if node.Annotations == nil {
			node.Annotations = map[string]string{}
		}
		node.Annotations[protectionKey] = state
		return nil
	})
}

func (c *ClusterNodeProtector) mutateNode(ctx context.Context, node *corev1.Node, mutateFn func(*corev1.Node) error) error {
	// get the latest node object to reduce the possibility of conflict
	err := c.shootClient.Get(ctx, types.NamespacedName{Name: node.Name}, node)
	if err != nil {
		return err
	}
	oldNode := node.DeepCopy()
	err = mutateFn(node)
	if err != nil {
		return err
	}
	if !reflect.DeepEqual(oldNode, node) {
		return c.shootClient.Update(ctx, node)
	}
	return nil
}

func getPodComponentType(pod *corev1.Pod) (*cloudv1.ComponentType, bool) {
	var component cloudv1.ComponentType
	if name, ok := pod.Labels[label.NameLabelKey]; ok {
		if name != tidbclusterName {
			return nil, false
		}
	} else {
		return nil, false
	}
	if componentName, ok := pod.Labels[label.ComponentLabelKey]; ok {
		component = cloudv1.ComponentType(componentName)
	} else {
		return nil, false
	}
	return &component, true
}