package tidbpod

import (
	"context"
	"fmt"
	"time"

	"github.com/go-logr/logr"
	"github.com/tidbcloud/aws-shared-provider/pkg/manager"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbmanager"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/kubeapi"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/labels"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	k8slabels "k8s.io/apimachinery/pkg/labels"
	"k8s.io/utils/ptr"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"sigs.k8s.io/controller-runtime/pkg/source"
)

const (
	MaxConcurrentSyncCacheReconciles = 10
)

// SyncCacheReconciler reconciles a Tidb Pod object.
type SyncCacheReconciler struct {
	controller.Controller
	logr.Logger

	Client      client.Client
	TidbManager tidbmanager.TidbManager
	Namespace   string
}

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
func (r *SyncCacheReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	logger.V(2).Info("start reconcile tidb pod for sync cache")
	start := time.Now()
	defer func() {
		logger.V(2).Info("finish reconcile tidb pod for sync cache", "time", time.Since(start))
	}()

	pod := &corev1.Pod{}
	if err := r.Client.Get(ctx, req.NamespacedName, pod); err != nil {
		if apierrors.IsNotFound(err) {
			return ctrl.Result{}, nil
		}
		logger.Error(err, "failed to get tidb pod")
		return ctrl.Result{}, err
	}

	if pod.Status.Phase == corev1.PodPending && pod.DeletionTimestamp == nil {
		return ctrl.Result{}, nil
	}

	podStatus := utils.PodNotReady
	if pod.DeletionTimestamp != nil {
		podStatus = utils.PodTerminating
	} else if kubeapi.IsPodReady(pod) {
		podStatus = utils.PodReady
	}

	token := r.TidbManager.GetAssignedToken(pod.Name, pod.Status.PodIP)

	logger.Info("[pod lifecycle] tidb pod status",
		"ip", pod.Status.PodIP,
		"node", pod.Spec.NodeName,
		"age", time.Since(pod.CreationTimestamp.Time).Round(100*time.Millisecond),
		"status", podStatus,
		"abnormal", kubeapi.IsPodAbnormal(pod),
		"disruption", kubeapi.GetPodDisruptionMsg(pod),
		"msg", pod.Status.Message,
		"token", token,
	)

	return ctrl.Result{}, r.TidbManager.SyncCache(ctx, pod.Spec.NodeName, pod.Name, pod.Status.PodIP, podStatus)
}

// podLabelSelectorPredicate creates a TypedPredicate for *corev1.Pod using a LabelSelector.
func podLabelSelectorPredicate(selector metav1.LabelSelector) (predicate.TypedPredicate[*corev1.Pod], error) {
	labelSelector, err := metav1.LabelSelectorAsSelector(&selector)
	if err != nil {
		return nil, err
	}
	return predicate.NewTypedPredicateFuncs(func(pod *corev1.Pod) bool {
		return labelSelector.Matches(k8slabels.Set(pod.GetLabels()))
	}), nil
}

// SetupWithManager sets up the controller with the Manager.
func (r *SyncCacheReconciler) SetupWithManager(mgr manager.Manager) error {
	c, err := controller.New("Tidb-Pod-SyncCache-"+r.Namespace, mgr, controller.Options{
		Reconciler:              r,
		MaxConcurrentReconciles: MaxConcurrentSyncCacheReconciles,
		NeedLeaderElection:      ptr.To(false),
	})
	if err != nil {
		return fmt.Errorf("can't new a controller: %w", err)
	}
	r.Controller = c
	r.Logger = c.GetLogger()

	byNamespaceFilter := func(pod *corev1.Pod) bool {
		return pod.GetNamespace() == r.Namespace
	}
	byLabelSelector := metav1.LabelSelector{
		MatchLabels: map[string]string{
			labels.NameLabelKey:      labels.TidbPodNameLabelValue,
			labels.ComponentLabelKey: labels.TidbPodComponentLabelValue,
		},
	}
	byNamespaceFuncs := predicate.NewTypedPredicateFuncs(byNamespaceFilter)
	byTidbPod, err := podLabelSelectorPredicate(byLabelSelector)
	if err != nil {
		return fmt.Errorf("invalid label selector: %w", err)
	}
	selector, err := metav1.LabelSelectorAsSelector(&byLabelSelector)
	if err != nil {
		return fmt.Errorf("invalid selector: %w", err)
	}

	err = mgr.GetFieldIndexer().IndexField(context.TODO(), &corev1.Pod{},
		"tidbpool.abnormal."+r.Namespace, func(obj client.Object) []string {
			if byNamespaceFilter(obj.(*corev1.Pod)) && selector.Matches(k8slabels.Set(obj.GetLabels())) &&
				kubeapi.IsPodAbnormal(obj.(*corev1.Pod)) {
				return []string{"abnormal"}
			}
			return nil
		})
	if err != nil {
		return fmt.Errorf("unable to add index for Tidb-Pod-SyncCache: %w", err)
	}

	if err := c.Watch(
		source.Kind(mgr.GetCache(), &corev1.Pod{}, &handler.TypedEnqueueRequestForObject[*corev1.Pod]{},
			byNamespaceFuncs, byTidbPod)); err != nil {
		return fmt.Errorf("can't watch Tidb-Pod-SyncCache: %w", err)
	}

	return nil
}