// Package metricsserver collects Pod resource metrics from metrics-server and
// writes them to Prometheus storage.
package metricsserver

import (
	"context"
	"errors"
	"fmt"
	"math"
	"sort"
	"sync"
	"time"

	commonmodel "github.com/prometheus/common/model"
	promlabels "github.com/prometheus/prometheus/model/labels"
	"github.com/prometheus/prometheus/model/value"
	"github.com/prometheus/prometheus/storage"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	klabels "k8s.io/apimachinery/pkg/labels"
	metricsv1beta1 "k8s.io/metrics/pkg/apis/metrics/v1beta1"
	metricsclient "k8s.io/metrics/pkg/client/clientset/versioned"
	"sigs.k8s.io/controller-runtime/pkg/client"

	decisionmetrics "github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/metrics"
	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/prom/discovery"
	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/prom/scraper"
	metricstarget "github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/target"
)

// Clock abstracts time for deterministic scrape validation.
type Clock interface {
	Now() time.Time
}

// Config configures a metrics-server scraper. Resources maps the Prometheus
// metric name written to the TSDB to the corresponding Kubernetes resource.
type Config struct {
	MetricsClient metricsclient.Interface
	Discovery     discovery.Discovery
	Reader        client.Reader
	TargetKind    string
	Appendable    storage.Appendable
	Clock         Clock
	Resources     map[string]corev1.ResourceName
	MaxAge        time.Duration
}

// MetricsServer resolves its configured target category and scrapes matching
// Pods from the metrics.k8s.io PodMetrics API. It writes one time series per
// container and resource, preserving the Pod relationship as internal labels.
type MetricsServer struct {
	metricsClient metricsclient.Interface
	discovery     discovery.Discovery
	reader        client.Reader
	targetKind    string
	appendable    storage.Appendable
	clock         Clock
	resources     []resourceMetric
	maxAge        time.Duration

	mu            sync.Mutex
	lastTimestamp map[string]int64
	activeSeries  map[string]promlabels.Labels
}

type resourceMetric struct {
	name     string
	resource corev1.ResourceName
}

var _ scraper.Scraper = (*MetricsServer)(nil)

func New(cfg *Config) (*MetricsServer, error) {
	if cfg == nil {
		return nil, fmt.Errorf("metrics-server scraper: config is required")
	}
	if cfg.MetricsClient == nil {
		return nil, fmt.Errorf("metrics-server scraper: metrics client is required")
	}
	if cfg.Discovery == nil {
		return nil, fmt.Errorf("metrics-server scraper: target discovery is required")
	}
	if cfg.Reader == nil {
		return nil, fmt.Errorf("metrics-server scraper: Kubernetes reader is required")
	}
	if cfg.Appendable == nil {
		return nil, fmt.Errorf("metrics-server scraper: appendable storage is required")
	}
	if cfg.Clock == nil {
		return nil, fmt.Errorf("metrics-server scraper: clock is required")
	}
	if cfg.MaxAge <= 0 {
		return nil, fmt.Errorf("metrics-server scraper: max age must be greater than zero")
	}
	if len(cfg.Resources) == 0 {
		return nil, fmt.Errorf("metrics-server scraper: resources are required")
	}

	resources := make([]resourceMetric, 0, len(cfg.Resources))
	for name, resourceName := range cfg.Resources {
		if name == "" {
			return nil, fmt.Errorf("metrics-server scraper: metric name is required")
		}
		if resourceName == "" {
			return nil, fmt.Errorf("metrics-server scraper: metric %q resource is required", name)
		}
		resources = append(resources, resourceMetric{name: name, resource: resourceName})
	}
	sort.Slice(resources, func(i, j int) bool { return resources[i].name < resources[j].name })

	scraper := &MetricsServer{
		metricsClient: cfg.MetricsClient,
		discovery:     cfg.Discovery,
		reader:        cfg.Reader,
		targetKind:    cfg.TargetKind,
		appendable:    cfg.Appendable,
		clock:         cfg.Clock,
		resources:     resources,
		maxAge:        cfg.MaxAge,
		lastTimestamp: make(map[string]int64),
		activeSeries:  make(map[string]promlabels.Labels),
	}
	return scraper, nil
}

// Scrape resolves this scraper's target category and performs one collection
// transaction for the matching Pods.
func (s *MetricsServer) Scrape(ctx context.Context) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	targets := s.discovery.Targets(s.targetKind)
	pods, err := metricstarget.ListPods(ctx, s.reader, targets...)
	if err != nil {
		return fmt.Errorf("metrics-server scraper: resolve targets: %w", err)
	}
	podsByKey, podMetrics, err := s.collectTargetMetrics(ctx, targets, pods)
	if err != nil {
		return err
	}
	sort.Slice(podMetrics, func(i, j int) bool {
		left, right := podMetrics[i], podMetrics[j]
		if left.Namespace != right.Namespace {
			return left.Namespace < right.Namespace
		}
		return left.Name < right.Name
	})

	appender := s.appendable.Appender(ctx)
	appended := false
	pendingTimestamps := make(map[string]int64)
	now := s.clock.Now()
	scrapeTimestamp := now.UnixMilli()
	currentSeries := make(map[string]promlabels.Labels)
	for i := range podMetrics {
		item := podMetrics[i]
		pod := podsByKey[podKey(item.Namespace, item.Name)]
		if !validSample(item.Timestamp.Time, pod, now, s.maxAge) {
			continue
		}
		sort.Slice(item.Containers, func(i, j int) bool { return item.Containers[i].Name < item.Containers[j].Name })
		for j := range item.Containers {
			container := &item.Containers[j]
			for _, configured := range s.resources {
				quantity, ok := container.Usage[configured.resource]
				if !ok {
					continue
				}
				metricLabels := labelsForSample(configured.name, pod, container.Name)
				seriesKey := metricLabels.String()
				currentSeries[seriesKey] = metricLabels
				if previous, ok := s.lastTimestamp[seriesKey]; ok && scrapeTimestamp <= previous {
					continue
				}
				_, err := appender.Append(0, metricLabels, scrapeTimestamp, quantity.AsApproximateFloat64())
				if isIgnorableAppendError(err) {
					pendingTimestamps[seriesKey] = scrapeTimestamp
					continue
				}
				if err != nil {
					_ = appender.Rollback()
					return fmt.Errorf("metrics-server scraper: append %s: %w", seriesKey, err)
				}
				appended = true
				pendingTimestamps[seriesKey] = scrapeTimestamp
			}
		}
	}
	for seriesKey, metricLabels := range s.activeSeries {
		if _, ok := currentSeries[seriesKey]; ok {
			continue
		}
		if previous, ok := s.lastTimestamp[seriesKey]; ok && scrapeTimestamp <= previous {
			continue
		}
		_, err := appender.Append(0, metricLabels, scrapeTimestamp, math.Float64frombits(value.StaleNaN))
		if isIgnorableAppendError(err) {
			pendingTimestamps[seriesKey] = scrapeTimestamp
			continue
		}
		if err != nil {
			_ = appender.Rollback()
			return fmt.Errorf("metrics-server scraper: append stale marker for %s: %w", seriesKey, err)
		}
		appended = true
		pendingTimestamps[seriesKey] = scrapeTimestamp
	}

	if !appended {
		if err := appender.Rollback(); err != nil {
			return fmt.Errorf("metrics-server scraper: rollback empty scrape: %w", err)
		}
	} else if err := appender.Commit(); err != nil {
		return fmt.Errorf("metrics-server scraper: commit: %w", err)
	}
	nextTimestamps := make(map[string]int64, len(currentSeries))
	for seriesKey := range currentSeries {
		if timestamp, ok := pendingTimestamps[seriesKey]; ok {
			nextTimestamps[seriesKey] = timestamp
		} else if timestamp, ok := s.lastTimestamp[seriesKey]; ok {
			nextTimestamps[seriesKey] = timestamp
		}
	}
	s.lastTimestamp = nextTimestamps
	s.activeSeries = currentSeries
	return nil
}

func (s *MetricsServer) collectTargetMetrics(
	ctx context.Context,
	targets []metricstarget.Target,
	pods []*corev1.Pod,
) (map[string]*corev1.Pod, []*metricsv1beta1.PodMetrics, error) {
	podsByKey := make(map[string]*corev1.Pod, len(pods))
	for _, pod := range pods {
		podsByKey[podKey(pod.Namespace, pod.Name)] = pod
	}

	seenSelectors := make(map[string]struct{}, len(targets))
	podMetricsByKey := make(map[string]*metricsv1beta1.PodMetrics, len(pods))
	for _, target := range targets {
		selector := klabels.SelectorFromSet(klabels.Set(metricstarget.PodSelectorLabels(target))).String()
		selectorKey := target.Namespace() + "\x00" + selector
		if _, exists := seenSelectors[selectorKey]; exists {
			continue
		}
		seenSelectors[selectorKey] = struct{}{}

		list, err := s.metricsClient.MetricsV1beta1().PodMetricses(target.Namespace()).List(ctx, metav1.ListOptions{
			LabelSelector: selector,
		})
		if err != nil {
			return nil, nil, fmt.Errorf("metrics-server scraper: list PodMetrics for target %s: %w", selectorKey, err)
		}
		for i := range list.Items {
			item := &list.Items[i]
			key := podKey(item.Namespace, item.Name)
			if _, selected := podsByKey[key]; selected {
				podMetricsByKey[key] = item.DeepCopy()
			}
		}
	}

	podMetrics := make([]*metricsv1beta1.PodMetrics, 0, len(podMetricsByKey))
	for _, item := range podMetricsByKey {
		podMetrics = append(podMetrics, item)
	}
	return podsByKey, podMetrics, nil
}

func validSample(timestamp time.Time, pod *corev1.Pod, now time.Time, maxAge time.Duration) bool {
	if pod == nil || timestamp.IsZero() {
		return false
	}
	if !pod.CreationTimestamp.IsZero() && timestamp.Before(pod.CreationTimestamp.Time) {
		return false
	}
	return now.Sub(timestamp) <= maxAge
}

func labelsForSample(metricName string, pod *corev1.Pod, container string) promlabels.Labels {
	values := map[string]string{
		commonmodel.MetricNameLabel:    metricName,
		decisionmetrics.LabelNamespace: pod.Namespace,
		decisionmetrics.LabelPod:       pod.Name,
		decisionmetrics.LabelContainer: container,
	}
	if group := pod.Labels[metricstarget.PodLabelGroup]; group != "" {
		values[metricstarget.LabelGroup] = group
	}
	if component := pod.Labels[metricstarget.PodLabelComponent]; component != "" {
		values[metricstarget.LabelComponent] = component
	}
	return promlabels.FromMap(values)
}

func podKey(namespace, name string) string {
	return namespace + "\x00" + name
}

func isIgnorableAppendError(err error) bool {
	return errors.Is(err, storage.ErrOutOfOrderSample) ||
		errors.Is(err, storage.ErrDuplicateSampleForTimestamp) ||
		errors.Is(err, storage.ErrTooOldSample)
}