package input_k8s_v1

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/tidbcloud/cluster-service-ng/pkg/autoscale/framework"
	"github.com/tidbcloud/cluster-service-ng/pkg/autoscale/modules/common"
	k8s "github.com/tidbcloud/cluster-service-ng/pkg/k8s"
	"gopkg.in/yaml.v2"
	autoscalingv2 "k8s.io/api/autoscaling/v2"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/client-go/rest"
	metricsapi "k8s.io/metrics/pkg/apis/metrics/v1beta1"
	resourceclient "k8s.io/metrics/pkg/client/clientset/versioned/typed/metrics/v1beta1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const InputK8sV1IdStr framework.InputProviderType = "InputK8sV1"

var InputK8sV1Meta = framework.InputProviderMeta{
	Type:             InputK8sV1IdStr,
	InterfaceVersion: framework.InputInterfaceV1IdStr,
}

type InputK8sV1Config struct {
	K8sClient                 client.Client
	ResourceMetricsClient     resourceclient.MetricsV1beta1Interface
	ComponentType             common.HPAComponentType
	DownsamplingIntervalInMin int `yaml:"downsamplingIntervalInMin"`
	CacheRetentionMinutes     int `yaml:"cacheRetentionMinutes"`
}

func (c *InputK8sV1Config) FillBlankValuesFrom(another framework.ConfigData) framework.ConfigData {
	if another == nil {
		return c
	}
	other, ok := another.(*InputK8sV1Config)
	if !ok {
		return c
	}
	merged := *c
	if merged.ComponentType == "" {
		merged.ComponentType = other.ComponentType
	}
	if merged.K8sClient == nil && other.K8sClient != nil {
		merged.K8sClient = other.K8sClient
	}
	if merged.ResourceMetricsClient == nil && other.ResourceMetricsClient != nil {
		merged.ResourceMetricsClient = other.ResourceMetricsClient
	}
	if merged.DownsamplingIntervalInMin == 0 && other.DownsamplingIntervalInMin != 0 {
		merged.DownsamplingIntervalInMin = other.DownsamplingIntervalInMin
	}
	if merged.CacheRetentionMinutes == 0 && other.CacheRetentionMinutes != 0 {
		merged.CacheRetentionMinutes = other.CacheRetentionMinutes
	}
	return &merged
}

func (c *InputK8sV1Config) UniqStr() string {
	return string(c.ComponentType)
}

func ParseConfig(cfg map[string]any) (framework.ConfigData, error) {
	if cfg == nil {
		return &InputK8sV1Config{}, nil
	}
	content, err := yaml.Marshal(cfg)
	if err != nil {
		return nil, err
	}
	var c InputK8sV1Config
	if err := yaml.UnmarshalStrict(content, &c); err != nil {
		return nil, err
	}

	restCfg, err := rest.InClusterConfig()
	if err != nil {
		return nil, fmt.Errorf("failed to get in-cluster config: %w", err)
	}
	c.K8sClient, err = k8s.NewClient(context.Background(), restCfg)
	if err != nil {
		return nil, fmt.Errorf("failed to create k8s client: %w", err)
	}
	c.ResourceMetricsClient = resourceclient.NewForConfigOrDie(restCfg)

	return &c, nil
}

type CachedMetricPoint struct {
	Time  time.Time
	Value int64
}

type MetricsCache struct {
	mu        sync.RWMutex
	data      map[string]map[string][]CachedMetricPoint
	retention time.Duration
}

func NewMetricsCache(retention time.Duration) *MetricsCache {
	return &MetricsCache{
		data:      make(map[string]map[string][]CachedMetricPoint),
		retention: retention,
	}
}

func (c *MetricsCache) Add(clusterId, metricName string, point CachedMetricPoint) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.data[clusterId] == nil {
		c.data[clusterId] = make(map[string][]CachedMetricPoint)
	}

	points := c.data[clusterId][metricName]
	points = append(points, point)

	cutoff := point.Time.Add(-c.retention)
	var validPoints []CachedMetricPoint
	for _, p := range points {
		if p.Time.After(cutoff) {
			validPoints = append(validPoints, p)
		}
	}

	c.data[clusterId][metricName] = validPoints
}

func (c *MetricsCache) GetRange(clusterId, metricName string, from, to time.Time) []CachedMetricPoint {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.data[clusterId] == nil {
		return nil
	}
	points := c.data[clusterId][metricName]

	var result []CachedMetricPoint
	for _, p := range points {
		if (p.Time.Equal(from) || p.Time.After(from)) &&
			(p.Time.Equal(to) || p.Time.Before(to)) {
			result = append(result, p)
		}
	}
	return result
}

type InputK8sV1 struct {
	eventReceiver             framework.EventReceiver
	k8sClient                 client.Client
	resourceMetricsClient     resourceclient.MetricsV1beta1Interface
	componentType             common.HPAComponentType
	metricsCache              *MetricsCache
	downsamplingIntervalInMin int
	stopCh                    chan struct{}
	wg                        sync.WaitGroup
	clusterIds                sync.Map
}

func NewInputK8sV1(eventReceiver framework.EventReceiver, config framework.ConfigData) (any, error) {
	cfg, ok := config.(*InputK8sV1Config)
	if !ok || cfg == nil {
		return nil, fmt.Errorf("[NewInputK8sV1] config type mismatch or nil")
	}
	if cfg.K8sClient == nil {
		return nil, fmt.Errorf("[NewInputK8sV1] k8sClient is required")
	}
	if cfg.ResourceMetricsClient == nil {
		return nil, fmt.Errorf("[NewInputK8sV1] resourceMetricsClient is required")
	}

	downsamplingIntervalInMin := cfg.DownsamplingIntervalInMin
	if downsamplingIntervalInMin == 0 {
		downsamplingIntervalInMin = 2
	}

	cacheRetentionMinutes := cfg.CacheRetentionMinutes
	if cacheRetentionMinutes == 0 {
		cacheRetentionMinutes = 30
	}

	this := &InputK8sV1{
		eventReceiver:             eventReceiver,
		k8sClient:                 cfg.K8sClient,
		resourceMetricsClient:     cfg.ResourceMetricsClient,
		componentType:             cfg.ComponentType,
		metricsCache:              NewMetricsCache(time.Duration(cacheRetentionMinutes) * time.Minute),
		downsamplingIntervalInMin: downsamplingIntervalInMin,
		stopCh:                    make(chan struct{}),
	}

	this.wg.Add(1)
	go this.runBackgroundCollector()

	return this, nil
}

func (this *InputK8sV1) Stop() {
	close(this.stopCh)
	this.wg.Wait()
}

func (this *InputK8sV1) runBackgroundCollector() {
	defer this.wg.Done()

	interval := time.Duration(this.downsamplingIntervalInMin) * time.Minute
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for {
		select {
		case <-this.stopCh:
			return
		case <-ticker.C:
			this.collectMetrics()
		}
	}
}

func (this *InputK8sV1) collectMetrics() {
	ctx := context.Background()

	this.clusterIds.Range(func(key, value any) bool {
		clusterId := key.(string)

		for _, metricName := range []string{"cpu_utilization", "memory_utilization"} {
			val, err := this.fetchCurrentMetric(ctx, clusterId, metricName)
			if err != nil {
				this.eventReceiver.OnEvent(
					EventInputK8sV1CollectMetricsError,
					framework.EventDataWithName("Error", framework.CloneableString(fmt.Sprintf("clusterId=%s, metric=%s: %s", clusterId, metricName, err.Error()))),
				)
				continue
			}
			this.metricsCache.Add(clusterId, metricName, CachedMetricPoint{
				Time:  time.Now(),
				Value: val,
			})
		}
		return true
	})
}

func (this *InputK8sV1) fetchCurrentMetric(ctx context.Context, clusterId, metricName string) (int64, error) {
	namespace := common.GetNamespace(clusterId)
	lbls := common.BuildLabels(clusterId, this.componentType)
	resourceName, err := parseResourceName(metricName)
	if err != nil {
		return 0, err
	}
	return this.getResourceMetricValue(ctx, namespace, lbls, resourceName)
}

func (this *InputK8sV1) GetRangeMetrics(
	clusterId framework.ClusterId,
	name framework.MetricName,
	fromTo framework.TimeRange,
) (framework.MetricTimeValueInt64s, error) {
	this.eventReceiver.OnEvent(
		EventInputK8sV1GetRangeMetrics,
		framework.EventDataWithName("ClusterId", framework.CloneableString(clusterId)),
		framework.EventDataWithName("MetricName", framework.CloneableString(string(name))),
	)

	this.clusterIds.Store(string(clusterId), struct{}{})

	metricName := string(name)
	points := this.metricsCache.GetRange(string(clusterId), metricName, fromTo.From, fromTo.To)

	if len(points) == 0 {
		value, err := this.getResourceMetric(string(clusterId), metricName)
		if err != nil {
			this.eventReceiver.OnEvent(
				EventInputK8sV1GetMetricsError,
				framework.EventDataWithName("Error", framework.CloneableString(err.Error())),
			)
			return nil, fmt.Errorf("[InputK8sV1.GetRangeMetrics] failed: %w", err)
		}

		now := time.Now()
		this.metricsCache.Add(string(clusterId), metricName, CachedMetricPoint{Time: now, Value: value})
		points = []CachedMetricPoint{{Time: now, Value: value}}
	}

	result := make(framework.MetricTimeValueInt64s, len(points))
	for idx, p := range points {
		result[idx] = framework.MetricTimeValueInt64{
			Time:  p.Time,
			Value: framework.MetricValueInt64(p.Value),
		}
	}

	this.eventReceiver.OnEvent(
		EventInputK8sV1GetMetricsSuccess,
		framework.EventDataWithName("PointsCount", framework.CloneableInt(len(result))),
	)

	return result, nil
}

func (this *InputK8sV1) getResourceMetric(clusterId, metricName string) (int64, error) {
	namespace := common.GetNamespace(clusterId)
	lbls := common.BuildLabels(clusterId, this.componentType)
	resourceName, err := parseResourceName(metricName)
	if err != nil {
		return 0, err
	}
	return this.getResourceMetricValue(context.Background(), namespace, lbls, resourceName)
}

func (this *InputK8sV1) getResourceMetricValue(ctx context.Context, namespace string, lbls map[string]string, resourceName corev1.ResourceName) (int64, error) {
	selector := labels.SelectorFromSet(lbls)
	metrics, err := this.resourceMetricsClient.PodMetricses(namespace).List(ctx, metav1.ListOptions{
		LabelSelector: selector.String(),
	})
	if err != nil {
		return 0, fmt.Errorf("unable to fetch metrics from resource metrics API: %w", err)
	}

	if len(metrics.Items) == 0 {
		return 0, fmt.Errorf("no metrics returned from resource metrics API")
	}

	pods := &corev1.PodList{}
	if err := this.k8sClient.List(ctx, pods,
		client.InNamespace(namespace),
		client.MatchingLabelsSelector{Selector: selector},
	); err != nil {
		return 0, fmt.Errorf("failed to list pods: %w", err)
	}

	if len(pods.Items) == 0 {
		return 0, fmt.Errorf("no pods found for selector %s", selector.String())
	}

	podMetrics := this.getPodMetrics(metrics.Items, resourceName)
	requests, err := this.calculatePodRequests(pods.Items, resourceName)
	if err != nil {
		return 0, fmt.Errorf("failed to calculate pod requests: %w", err)
	}

	ignoredPods := this.groupPods(pods.Items, podMetrics)
	this.removeMetricsForPods(podMetrics, ignoredPods)

	if len(podMetrics) == 0 {
		return 0, fmt.Errorf("no ready pods with metrics found")
	}

	metricsTotal := int64(0)
	requestsTotal := int64(0)
	for podName, metric := range podMetrics {
		if request, ok := requests[podName]; ok {
			metricsTotal += metric.Value
			requestsTotal += request
		}
	}

	if requestsTotal == 0 {
		return 0, fmt.Errorf("no requests found for pods")
	}

	return int64(float64(metricsTotal) / float64(requestsTotal) * 100), nil
}

func (this *InputK8sV1) GetCPUUtilization(clusterId framework.ClusterId) (float64, error) {
	return this.GetUtilization(clusterId, corev1.ResourceCPU)
}

func (this *InputK8sV1) GetMemoryUtilization(clusterId framework.ClusterId) (float64, error) {
	return this.GetUtilization(clusterId, corev1.ResourceMemory)
}

func (this *InputK8sV1) GetUtilization(clusterId framework.ClusterId, resourceName corev1.ResourceName) (float64, error) {
	namespace := common.GetNamespace(string(clusterId))
	lbls := common.BuildLabels(string(clusterId), this.componentType)

	selector := labels.SelectorFromSet(lbls)
	metrics, err := this.resourceMetricsClient.PodMetricses(namespace).List(context.Background(), metav1.ListOptions{
		LabelSelector: selector.String(),
	})
	if err != nil {
		return 0, fmt.Errorf("unable to fetch metrics from resource metrics API: %w", err)
	}

	if len(metrics.Items) == 0 {
		return 0, fmt.Errorf("no metrics returned from resource metrics API")
	}

	pods := &corev1.PodList{}
	if err := this.k8sClient.List(context.Background(), pods,
		client.InNamespace(namespace),
		client.MatchingLabelsSelector{Selector: selector},
	); err != nil {
		return 0, fmt.Errorf("failed to list pods: %w", err)
	}

	if len(pods.Items) == 0 {
		return 0, fmt.Errorf("no pods found for selector %s", selector.String())
	}

	podMetrics := this.getPodMetrics(metrics.Items, resourceName)
	requests, err := this.calculatePodRequests(pods.Items, resourceName)
	if err != nil {
		return 0, fmt.Errorf("failed to calculate pod requests: %w", err)
	}

	ignoredPods := this.groupPods(pods.Items, podMetrics)
	this.removeMetricsForPods(podMetrics, ignoredPods)

	if len(podMetrics) == 0 {
		return 0, fmt.Errorf("no ready pods with metrics found")
	}

	metricsTotal := int64(0)
	requestsTotal := int64(0)
	for podName, metric := range podMetrics {
		if request, ok := requests[podName]; ok {
			metricsTotal += metric.Value
			requestsTotal += request
		}
	}

	if requestsTotal == 0 {
		return 0, fmt.Errorf("no requests found for pods")
	}

	return float64(metricsTotal) / float64(requestsTotal) * 100, nil
}

func (this *InputK8sV1) getPodMetrics(rawMetrics []metricsapi.PodMetrics, res corev1.ResourceName) PodMetricsInfo {
	result := make(PodMetricsInfo, len(rawMetrics))
	for idx := range rawMetrics {
		m := &rawMetrics[idx]
		if len(m.Containers) == 0 {
			continue
		}
		podSum := int64(0)
		missing := false
		for _, c := range m.Containers {
			resValue, found := c.Usage[res]
			if !found {
				missing = true
				break
			}
			podSum += resValue.MilliValue()
		}
		if !missing {
			result[m.Name] = PodMetric{
				Timestamp: m.Timestamp.Time,
				Window:    m.Window.Duration,
				Value:     podSum,
			}
		}
	}
	return result
}

func (this *InputK8sV1) calculatePodRequests(pods []corev1.Pod, res corev1.ResourceName) (map[string]int64, error) {
	requests := make(map[string]int64, len(pods))
	for idx := range pods {
		pod := &pods[idx]
		podSum := int64(0)
		hasRequest := false
		for j := range pod.Spec.Containers {
			c := &pod.Spec.Containers[j]
			if containerRequest, ok := c.Resources.Requests[res]; ok {
				podSum += containerRequest.MilliValue()
				hasRequest = true
			}
		}
		if !hasRequest {
			return nil, fmt.Errorf("no %s request found for pod %s", res, pod.Name)
		}
		requests[pod.Name] = podSum
	}
	return requests, nil
}

func (this *InputK8sV1) groupPods(pods []corev1.Pod, metrics PodMetricsInfo) sets.Set[string] {
	ignoredPods := sets.New[string]()
	for idx := range pods {
		pod := &pods[idx]
		if pod.DeletionTimestamp != nil || pod.Status.Phase == corev1.PodFailed {
			ignoredPods.Insert(pod.Name)
			continue
		}
		if pod.Status.Phase == corev1.PodPending {
			ignoredPods.Insert(pod.Name)
			continue
		}
		if _, found := metrics[pod.Name]; !found {
			ignoredPods.Insert(pod.Name)
			continue
		}
	}
	return ignoredPods
}

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

func parseResourceName(metricName string) (corev1.ResourceName, error) {
	switch metricName {
	case "cpu", "cpu_utilization":
		return corev1.ResourceCPU, nil
	case "memory", "memory_utilization":
		return corev1.ResourceMemory, nil
	default:
		return "", fmt.Errorf("unsupported metric name: %s", metricName)
	}
}

func (this *InputK8sV1) GetClusterTopology(clusterId framework.ClusterId) (framework.ClusterTopologyV1, error) {
	this.eventReceiver.OnEvent(
		EventInputK8sV1GetTopology,
		framework.EventDataWithName("ClusterId", framework.CloneableString(clusterId)),
	)

	namespace := common.GetNamespace(string(clusterId))
	lbls := common.BuildLabels(string(clusterId), this.componentType)

	hpa, err := this.getHPA(namespace, lbls)
	if err != nil {
		this.eventReceiver.OnEvent(
			EventInputK8sV1GetHPAError,
			framework.EventDataWithName("Error", framework.CloneableString(err.Error())),
		)
		return nil, fmt.Errorf("[InputK8sV1.GetClusterTopology] failed to get HPA: %w", err)
	}

	currentReplicas := hpa.Status.CurrentReplicas
	topology := framework.ClusterTopologyV1{
		framework.InstanceName(this.componentType): {
			Name:     framework.InstanceName(this.componentType),
			Type:     framework.InstanceType(this.componentType),
			Replicas: uint(currentReplicas),
		},
	}

	this.eventReceiver.OnEvent(
		EventInputK8sV1GetTopologySuccess,
		framework.EventDataWithName("Replicas", framework.CloneableInt(int(currentReplicas))),
	)

	return topology, nil
}

func (this *InputK8sV1) getHPA(namespace string, lbls map[string]string) (*autoscalingv2.HorizontalPodAutoscaler, error) {
	hpas := &autoscalingv2.HorizontalPodAutoscalerList{}
	labelSelector := labels.SelectorFromSet(lbls)

	if err := this.k8sClient.List(context.Background(), hpas,
		client.InNamespace(namespace),
		client.MatchingLabelsSelector{Selector: labelSelector},
	); err != nil {
		return nil, fmt.Errorf("failed to list HPA: %w", err)
	}

	if len(hpas.Items) == 0 {
		return nil, errors.NewNotFound(
			autoscalingv2.Resource("horizontalpodautoscalers"),
			fmt.Sprintf("hpa for component %s", this.componentType),
		)
	}

	if len(hpas.Items) > 1 {
		return nil, fmt.Errorf("expected exactly one HPA for component %s in namespace %s, found %d",
			this.componentType, namespace, len(hpas.Items))
	}

	return &hpas.Items[0], nil
}

type PodMetric struct {
	Timestamp time.Time
	Window    time.Duration
	Value     int64
}

type PodMetricsInfo map[string]PodMetric

const (
	EventInputK8sV1GetRangeMetrics     framework.EventName = "InputK8sV1.GetRangeMetrics"
	EventInputK8sV1GetTopology         framework.EventName = "InputK8sV1.GetTopology"
	EventInputK8sV1GetTopologySuccess  framework.EventName = "InputK8sV1.GetTopology.Success"
	EventInputK8sV1GetHPAError         framework.EventName = "InputK8sV1.GetHPA.Error"
	EventInputK8sV1GetMetricsError     framework.EventName = "InputK8sV1.GetMetrics.Error"
	EventInputK8sV1GetMetricsSuccess   framework.EventName = "InputK8sV1.GetMetrics.Success"
	EventInputK8sV1CollectMetrics      framework.EventName = "InputK8sV1.CollectMetrics"
	EventInputK8sV1CollectMetricsError framework.EventName = "InputK8sV1.CollectMetrics.Error"
)