package dispatcher

import (
	"context"
	"fmt"

	"github.com/go-logr/logr"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/shared-provider/pkg/utils"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	apitypes "k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/tools/record"
	"sigs.k8s.io/controller-runtime/pkg/cache"
	"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/source"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/shared-provider/pkg/manager"
	"github.com/tidbcloud/shared-provider/pkg/metrics"
	"github.com/tidbcloud/shared-provider/pkg/types"
	"github.com/tidbcloud/shared-provider/pkg/utils/labels"
)

type ClusterDispatcherConfig struct {
	Manager manager.Manager

	RescheduleThreshold int

	KubeClusterConfigMapName      string
	KubeClusterConfigMapNamespace string
}

func NewClusterDispatcher(cfg *ClusterDispatcherConfig) Dispatcher {
	d := &cluster{
		KubeClusterConfigMapName:      cfg.KubeClusterConfigMapName,
		KubeClusterConfigMapNamespace: cfg.KubeClusterConfigMapNamespace,

		RescheduleThreshold: cfg.RescheduleThreshold,
	}

	d.InfraClient = cfg.Manager.InfraCluster().GetClient()
	d.InfraCache = cfg.Manager.InfraCluster().GetCache()
	d.Cache = cfg.Manager.GetCache()
	d.InfraRecorder = cfg.Manager.InfraCluster().GetEventRecorderFor("DispatcherCluster")
	d.Logger = cfg.Manager.GetLogger().WithValues("dispatcher", "Cluster")

	return d
}

type cluster struct {
	Logger        logr.Logger
	InfraClient   client.Client
	InfraCache    cache.Cache
	InfraRecorder record.EventRecorder

	Cache cache.Cache

	RescheduleThreshold           int
	KubeClusterConfigMapName      string
	KubeClusterConfigMapNamespace string
}

func (d *cluster) New() client.Object {
	return &cloudv1.Cluster{}
}

func (d *cluster) Watch(c controller.Controller) error {
	ptp := utils.GetPredicateForProviderTypes(types.ProviderTypeGCPSharedTier, types.ProviderTypeAWSSharedTier)

	obj := d.New()

	if err := c.Watch(
		source.NewKindWithCache(obj, d.InfraCache),
		&handler.EnqueueRequestForObject{},
		ptp,
	); err != nil {
		return fmt.Errorf("can't watch resource: %w", err)
	}

	if err := c.Watch(
		source.NewKindWithCache(&cloudv1.Network{}, d.InfraCache),
		&handler.Funcs{
			UpdateFunc: d.handleNetworkLabelsUpdation,
		},
		ptp,
	); err != nil {
		return fmt.Errorf("can't watch network: %w", err)
	}

	return nil
}

func (d *cluster) Schedule(ctx context.Context, obj client.Object) (string, error) {
	scheduled := labels.GetKubeClusterFromLabels(obj.GetLabels())
	if scheduled != "" {
		return "", nil
	}

	net, err := GetNetworkFromCluster(ctx, d.InfraCache, obj)
	if err != nil {
		return "", err
	}
	if net.Spec.Provider != types.ProviderTypeGCPSharedTier && net.Spec.Provider != types.ProviderTypeAWSSharedTier {
		return "", nil
	}

	kubeClusterName := labels.GetKubeClusterFromLabels(net.Labels)
	if kubeClusterName == "" {
		return "", fmt.Errorf("network %s/%s is unscheduled, waiting", net.Namespace, net.Name)
	}

	selector, err := metav1.LabelSelectorAsSelector(&metav1.LabelSelector{
		MatchLabels: map[string]string{
			labels.LabelKeyKubernetesCluster: kubeClusterName,
		},
		MatchExpressions: []metav1.LabelSelectorRequirement{{
			Key:      cloudv1.ProviderTypeLabel,
			Operator: metav1.LabelSelectorOpIn,
			Values:   []string{types.ProviderTypeGCPSharedTier, types.ProviderTypeAWSSharedTier},
		}},
	})
	if err != nil {
		return "", fmt.Errorf("can't get LabelSelectorAsSelector in %s: %w", kubeClusterName, err)
	}

	cl := cloudv1.ClusterList{}
	if err := d.InfraCache.List(ctx, &cl, client.MatchingLabelsSelector{Selector: selector}); err != nil {
		return "", fmt.Errorf("can't list tidb clusters in %s: %w", kubeClusterName, err)
	}

	if len(cl.Items) > d.RescheduleThreshold {
		reason, isFailed := isRescheduleFailed(net)
		if isFailed {
			d.Logger.Info("reschedule is failed previously, so just create cluster in current network",
				"namespace", net.Namespace, "name", net.Name, "reason", reason)
			return kubeClusterName, nil
		}

		kubeClusterNames, err := getAvailableKubeClusters(
			ctx,
			d.Cache,
			d.KubeClusterConfigMapNamespace,
			d.KubeClusterConfigMapName,
			net.Spec.Region,
		)
		if err != nil {
			return "", err
		}

		others := []string{}
		for _, c := range kubeClusterNames {
			if c != kubeClusterName {
				others = append(others, c)
			}
		}

		to, err := getPreferedKubeCluster(ctx, d.InfraCache, len(cl.Items), others)
		if err != nil {
			return "", err
		}
		if to == "" {
			d.Logger.Info("current network is the best choice, skip rescheduling",
				"namespace", net.Namespace, "name", net.Name)
			return kubeClusterName, nil
		}

		d.Logger.Info("try to reschedule network", "namespace",
			net.Namespace, "name", net.Name, "from", kubeClusterName, "to", to)
		if err := reschedule(ctx, d.InfraClient, net, to); err != nil {
			return "", fmt.Errorf("failed to reschedule network: %w", err)
		}

		return "", nil
	}

	return kubeClusterName, nil
}

func (d *cluster) DeleteMetrics(key client.ObjectKey) {
	d.Logger.Info("delete metrics", "namespace", key.Namespace, "name", key.Name)

	// TODO: refactor to avoid new an object
	obj := d.New()
	obj.SetNamespace(key.Namespace)
	obj.SetName(key.Name)
	labels := getClusterMetricLabels(obj, true)

	metrics.ClusterPhase.Delete(prometheus.Labels(labels))
	metrics.ClusterCreationTime.Delete(prometheus.Labels(labels))
}

func (d *cluster) UpdateMetrics(obj client.Object) {
	d.Logger.Info("update metrics", "namespace", obj.GetNamespace(), "name", obj.GetName())

	labels := getClusterMetricLabels(obj, false)
	metrics.ClusterPhase.Set(prometheus.Labels(labels), 1)

	keyLabels := getClusterMetricLabels(obj, true)
	metrics.ClusterCreationTime.With(prometheus.Labels(keyLabels)).Set(float64(obj.GetCreationTimestamp().Unix()))
}

func GetNetworkFromCluster(ctx context.Context, r client.Reader, obj client.Object) (*cloudv1.Network, error) {
	c, ok := obj.(*cloudv1.Cluster)
	if !ok {
		return nil, fmt.Errorf("can't assert obj as cluster")
	}

	return getNetworkFromCluster(ctx, r, c)
}

func getNetworkFromCluster(ctx context.Context, r client.Reader, c *cloudv1.Cluster) (*cloudv1.Network, error) {
	netKey := client.ObjectKey{
		Namespace: c.GetNamespace(),
		Name:      c.Spec.NetworkName,
	}

	net := cloudv1.Network{}

	if err := r.Get(ctx, netKey, &net); err != nil {
		return nil, err
	}
	return &net, nil
}

func getClusterMetricLabels(obj client.Object, onlyKey bool) map[string]string {
	c, ok := obj.(*cloudv1.Cluster)
	if !ok {
		panic("can't assert obj as cluster")
	}

	m := map[string]string{
		"provider_type": labels.GetProviderTypeFromLabels(c.Labels),
		"namespace":     c.Namespace,
		"tidb":          c.Name,
	}
	if !onlyKey {
		m["kubernetes"] = labels.GetKubeClusterFromLabels(c.Labels)
		m["network"] = c.Spec.NetworkName
		m["phase"] = string(c.Status.Phase)
	}
	return m
}

func reschedule(ctx context.Context, c client.Client, net *cloudv1.Network, to string) error {
	patch := fmt.Sprintf(
		`{"metadata":{"annotations":{"%s":"%s","%s":null}}}`,
		types.AnnoKeyReschedule,
		to,
		types.AnnoKeyRescheduleFailedReason,
	)

	return c.Patch(ctx, net, client.RawPatch(apitypes.MergePatchType, []byte(patch)))
}

func isRescheduleFailed(net *cloudv1.Network) (string, bool) {
	if net.Annotations == nil {
		return "", false
	}
	reason, ok := net.Annotations[types.AnnoKeyRescheduleFailedReason]
	return reason, ok
}