package metricsprovider

import (
	"context"
	"fmt"
	"log/slog"
	"time"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	k8s "github.com/tidbcloud/cluster-service-ng/pkg/k8s"
	autoscalingv1 "k8s.io/api/autoscaling/v1"
	autoscalingv2 "k8s.io/api/autoscaling/v2"
	corev1 "k8s.io/api/core/v1"
	apimeta "k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/client-go/scale"
	metricsapi "k8s.io/metrics/pkg/apis/metrics/v1beta1"
)

// PodMetric contains pod metric value (the metric values are expected to be the metric as a milli-value)
type PodMetric struct {
	Timestamp time.Time
	Window    time.Duration
	Value     int64
}

// PodMetricsInfo contains pod metrics as a map from pod names to PodMetricsInfo
type PodMetricsInfo map[string]PodMetric

func drainingPodNames(pods []corev1.Pod) sets.Set[string] {
	drainingPods := sets.New[string]()
	for i := range pods {
		pod := &pods[i]
		if pod.Annotations == nil {
			continue
		}
		if _, ok := pod.Annotations[v1alpha1.AnnoKeyTiProxyGracefulShutdownBeginTime]; ok {
			drainingPods.Insert(pod.Name)
		}
	}
	return drainingPods
}

//nolint:unparam // readyPodCount is part of the pod classification result; current callers only need pod sets.
func groupPods(
	pods []corev1.Pod,
	metrics PodMetricsInfo,
	resource corev1.ResourceName,
	cpuInitializationPeriod,
	delayOfInitialReadinessStatus time.Duration,
) (readyPodCount int, unreadyPods, missingPods, ignoredPods sets.Set[string]) {
	missingPods = sets.New[string]()
	unreadyPods = sets.New[string]()
	ignoredPods = sets.New[string]()
	for i := range pods {
		pod := &pods[i]
		if pod.DeletionTimestamp != nil || pod.Status.Phase == corev1.PodFailed {
			ignoredPods.Insert(pod.Name)
			continue
		}
		// Pending pods are ignored.
		if pod.Status.Phase == corev1.PodPending {
			unreadyPods.Insert(pod.Name)
			continue
		}
		// Pods missing metrics.
		metric, found := metrics[pod.Name]
		if !found {
			missingPods.Insert(pod.Name)
			continue
		}
		// Unready pods are ignored.
		if resource == corev1.ResourceCPU {
			var unready bool
			_, condition := k8s.GetPodCondition(&pod.Status, corev1.PodReady)
			if condition == nil || pod.Status.StartTime == nil {
				unready = true
			} else {
				// Pod still within possible initialisation period.
				if pod.Status.StartTime.Add(cpuInitializationPeriod).After(time.Now()) {
					// Ignore sample if pod is unready or one window of metric wasn't collected since last state transition.
					unready = condition.Status == corev1.ConditionFalse || metric.Timestamp.Before(condition.LastTransitionTime.Add(metric.Window))
				} else {
					// Ignore metric if pod is unready and it has never been ready.
					unready = condition.Status == corev1.ConditionFalse && pod.Status.StartTime.Add(delayOfInitialReadinessStatus).After(condition.LastTransitionTime.Time)
				}
			}
			if unready {
				unreadyPods.Insert(pod.Name)
				continue
			}
		}
		readyPodCount++
	}
	return readyPodCount, unreadyPods, missingPods, ignoredPods
}

func getContainerMetrics(rawMetrics []metricsapi.PodMetrics, resource corev1.ResourceName, container string) (PodMetricsInfo, error) {
	res := make(PodMetricsInfo, len(rawMetrics))
	for i := range rawMetrics {
		m := &rawMetrics[i]
		containerFound := false
		for _, c := range m.Containers {
			if c.Name == container {
				containerFound = true
				if val, resFound := c.Usage[resource]; resFound {
					res[m.Name] = PodMetric{
						Timestamp: m.Timestamp.Time,
						Window:    m.Window.Duration,
						Value:     val.MilliValue(),
					}
				}
				break
			}
		}
		if !containerFound {
			return nil, fmt.Errorf("container %s not present in metrics for pod %s/%s", container, m.Namespace, m.Name)
		}
	}
	return res, nil
}

func getPodMetrics(rawMetrics []metricsapi.PodMetrics, resource corev1.ResourceName) PodMetricsInfo {
	res := make(PodMetricsInfo, len(rawMetrics))
	for i := range rawMetrics {
		m := &rawMetrics[i]
		podSum := int64(0)
		missing := len(m.Containers) == 0
		for _, c := range m.Containers {
			resValue, found := c.Usage[resource]
			if !found {
				missing = true
				slog.Info("missing resource metric", "resource", resource, "namespace", m.Namespace, "name", m.Name)
				break
			}
			podSum += resValue.MilliValue()
		}
		if !missing {
			res[m.Name] = PodMetric{
				Timestamp: m.Timestamp.Time,
				Window:    m.Window.Duration,
				Value:     podSum,
			}
		}
	}
	return res
}

func getResourceMetricSource(
	in []autoscalingv2.MetricSpec,
	resourceName corev1.ResourceName,
) (*autoscalingv2.ResourceMetricSource, bool) {
	for i := range in {
		metricSpec := in[i]
		if metricSpec.Type != autoscalingv2.ResourceMetricSourceType {
			continue
		}
		if metricSpec.Resource == nil {
			continue
		}
		if metricSpec.Resource.Name != resourceName {
			continue
		}
		return metricSpec.Resource, true
	}
	return nil, false
}

func getScale(
	ctx context.Context,
	scaleClient scale.ScalesGetter,
	mapper apimeta.RESTMapper,
	namespace string,
	ref autoscalingv2.CrossVersionObjectReference,
) (*autoscalingv1.Scale, error) {
	// 1. parse GroupVersion
	targetGV, err := schema.ParseGroupVersion(ref.APIVersion)
	if err != nil {
		return nil, err
	}

	// 2. build GroupKind
	targetGK := schema.GroupKind{
		Group: targetGV.Group,
		Kind:  ref.Kind,
	}

	// 3. Kind -> Resource
	mappings, err := mapper.RESTMappings(targetGK)
	if err != nil {
		return nil, err
	}
	if len(mappings) == 0 {
		return nil, fmt.Errorf("no REST mappings found for %s", targetGK.String())
	}

	// 4. get scale subresource
	var lastErr error
	for _, mapping := range mappings {
		gr := mapping.Resource.GroupResource()
		scaleObj, err := scaleClient.Scales(namespace).Get(ctx, gr, ref.Name, metav1.GetOptions{})
		if err == nil {
			return scaleObj, nil
		}
		lastErr = err
	}
	return nil, lastErr
}

func removeMetricsForPods(metrics PodMetricsInfo, pods sets.Set[string]) {
	for _, pod := range pods.UnsortedList() {
		delete(metrics, pod)
	}
}

func calculatePodRequests(pods []corev1.Pod, container string, resource corev1.ResourceName) (map[string]int64, error) {
	requests := make(map[string]int64, len(pods))
	for i := range pods {
		pod := &pods[i]
		podSum := int64(0)
		for j := range pod.Spec.Containers {
			c := &pod.Spec.Containers[j]
			if container == "" || container == c.Name {
				if containerRequest, ok := c.Resources.Requests[resource]; ok {
					podSum += containerRequest.MilliValue()
				} else {
					return nil, fmt.Errorf("missing request for %s", resource)
				}
			}
		}
		requests[pod.Name] = podSum
	}
	return requests, nil
}

func calculatePodLimits(pods []corev1.Pod, container string, resource corev1.ResourceName) (map[string]int64, error) {
	limits := make(map[string]int64, len(pods))
	for i := range pods {
		pod := &pods[i]
		podSum := int64(0)
		for j := range pod.Spec.Containers {
			c := &pod.Spec.Containers[j]
			if container == "" || container == c.Name {
				if containerLimit, ok := c.Resources.Limits[resource]; ok {
					podSum += containerLimit.MilliValue()
				} else {
					if container != "" {
						return nil, fmt.Errorf("missing limit for %s in pod %s container %s", resource, pod.Name, container)
					}
					return nil, fmt.Errorf("missing limit for %s in pod %s", resource, pod.Name)
				}
			}
		}
		limits[pod.Name] = podSum
	}
	return limits, nil
}

func GetResourceUtilizationRatio(metrics PodMetricsInfo, requests map[string]int64, targetUtilization int32) (utilizationRatio float64, currentUtilization int32, rawAverageValue int64, err error) {
	metricsTotal := int64(0)
	requestsTotal := int64(0)
	numEntries := 0

	for podName, metric := range metrics {
		request, hasRequest := requests[podName]
		if !hasRequest {
			// we check for missing requests elsewhere, so assuming missing requests == extraneous metrics
			continue
		}

		metricsTotal += metric.Value
		requestsTotal += request
		numEntries++
	}

	// if the set of requests is completely disjoint from the set of metrics,
	// then we could have an issue where the requests total is zero
	if requestsTotal == 0 {
		return 0, 0, 0, fmt.Errorf("no metrics returned matched known pods")
	}

	currentUtilization = int32((metricsTotal * 100) / requestsTotal)

	return float64(currentUtilization) / float64(targetUtilization), currentUtilization, metricsTotal / int64(numEntries), nil
}