package vpa

import (
	"context"
	"fmt"
	"time"

	"github.com/aws/aws-node-termination-handler/pkg/ec2metadata"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	v1 "k8s.io/client-go/informers/core/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/scheme"
	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
)

const (
	checkInterval = 3 * time.Second
	metadataURL   = "http://169.254.169.254"
	metadataTries = 2
)

func StartMonitorRebalanceSignal(clientset *kubernetes.Clientset, nodeInformer v1.NodeInformer, nodeName string) error {
	broadcaster := record.NewBroadcaster()
	broadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: clientset.CoreV1().Events("")})
	recorder := broadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: "RebalanceSignal"})

	imds := ec2metadata.New(metadataURL, metadataTries)
	if _, err := imds.GetMetadataInfo(ec2metadata.IdentityDocPath); err != nil {
		return err
	}

	checkTicker := time.NewTicker(checkInterval)
	go func() {
		for range checkTicker.C {
			rebalanceRec, err := imds.GetRebalanceRecommendationEvent()
			if err != nil {
				ctrl.Log.Error(err, "failed to get rebalance recommendation event", "node", nodeName)
				continue
			}
			if rebalanceRec == nil {
				continue
			}
			noticeTime, err := time.Parse(time.RFC3339, rebalanceRec.NoticeTime)
			if err != nil {
				ctrl.Log.Error(err, "failed to parse noticeTime", "node", nodeName)
				continue
			}

			node, err := nodeInformer.Lister().Get(nodeName)
			if err != nil {
				ctrl.Log.Error(err, "failed to get node", "node", nodeName)
				continue
			}
			if node.DeletionTimestamp != nil {
				continue
			}

			needUpdate := true
			for i := range node.Spec.Taints {
				if node.Spec.Taints[i].Key == types.RebalanceSignalTaintKey {
					needUpdate = false
					break
				}
			}
			if !needUpdate {
				continue
			}

			node = node.DeepCopy()
			node.Spec.Taints = append(node.Spec.Taints, corev1.Taint{
				Key:    types.RebalanceSignalTaintKey,
				Value:  fmt.Sprint(noticeTime.Unix()),
				Effect: corev1.TaintEffectNoSchedule,
			})
			ctx, cancel := context.WithTimeout(context.Background(), checkInterval)
			_, err = clientset.CoreV1().Nodes().Update(ctx, node, metav1.UpdateOptions{})
			cancel()
			if err != nil {
				ctrl.Log.Error(err, "failed to update rebalance-signal taint",
					"node", nodeName, "providerID", node.Spec.ProviderID)
				continue
			}

			ctrl.Log.Info("receive rebalance signal and update taint",
				"node", nodeName, "providerID", node.Spec.ProviderID, "noticeTime", rebalanceRec.NoticeTime)
			recorder.Eventf(node, corev1.EventTypeWarning, "RebalanceSignal",
				"%s %s receive rebalance signal and update taint", node.Spec.ProviderID, rebalanceRec.NoticeTime)
			RebalanceSignalCounter.Inc()
		}
	}()

	return nil
}