// Package podmetrics scrapes Prometheus exposition endpoints directly from
// discovered Pods and writes their samples to Prometheus storage.
package podmetrics

import (
	"context"
	"errors"
	"fmt"
	"io"
	"math"
	"net"
	"net/http"
	"net/url"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	commonmodel "github.com/prometheus/common/model"
	promlabels "github.com/prometheus/prometheus/model/labels"
	"github.com/prometheus/prometheus/model/textparse"
	"github.com/prometheus/prometheus/model/value"
	"github.com/prometheus/prometheus/storage"
	corev1 "k8s.io/api/core/v1"
	"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"
)

const (
	AnnotationScrape = "prometheus.io/scrape"
	AnnotationPath   = "prometheus.io/path"
	AnnotationPort   = "prometheus.io/port"
	AnnotationScheme = "prometheus.io/scheme"

	DefaultTimeout     = 10 * time.Second
	DefaultMaxBodySize = 32 * 1024 * 1024
	DefaultConcurrency = 16

	defaultMetricsPath = "/metrics"
	maxBodySize        = 1024 * 1024 * 1024
	acceptHeader       = "application/openmetrics-text;version=1.0.0;escaping=underscores;q=0.6," +
		"application/openmetrics-text;version=0.0.1;q=0.5," +
		"text/plain;version=1.0.0;escaping=underscores;q=0.4," +
		"text/plain;version=0.0.4;q=0.3,*/*;q=0.2"
)

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

// Config configures direct Pod endpoint scraping. HTTPClient must be safe for
// concurrent use. A non-empty Metrics list is an exact metric-name allowlist;
// an empty list preserves every exposed metric. Timeout applies independently
// to each Pod request. Redirects are always disabled so a Pod endpoint cannot
// redirect the scraper outside the discovered Pod IP boundary.
type Config struct {
	HTTPClient  *http.Client
	Discovery   discovery.Discovery
	Reader      client.Reader
	TargetKind  string
	Appendable  storage.Appendable
	Clock       Clock
	Metrics     []string
	Timeout     time.Duration
	MaxBodySize int64
	Concurrency int
}

// PodMetrics resolves one discovery category and scrapes opt-in Pod endpoints
// using the conventional prometheus.io annotations.
type PodMetrics struct {
	httpClient  *http.Client
	discovery   discovery.Discovery
	reader      client.Reader
	targetKind  string
	appendable  storage.Appendable
	clock       Clock
	metrics     map[string]struct{}
	timeout     time.Duration
	maxBodySize int64
	concurrency int

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

type podScrapeResult struct {
	targetKey string
	samples   map[string]podSample
	err       error
}

type podSample struct {
	labels    promlabels.Labels
	timestamp *int64
	value     float64
}

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

// New constructs a direct Pod metrics scraper.
func New(cfg *Config) (*PodMetrics, error) {
	if cfg == nil {
		return nil, fmt.Errorf("pod metrics scraper: config is required")
	}
	if cfg.HTTPClient == nil {
		return nil, fmt.Errorf("pod metrics scraper: HTTP client is required")
	}
	if cfg.Discovery == nil {
		return nil, fmt.Errorf("pod metrics scraper: target discovery is required")
	}
	if cfg.Reader == nil {
		return nil, fmt.Errorf("pod metrics scraper: Kubernetes reader is required")
	}
	if cfg.Appendable == nil {
		return nil, fmt.Errorf("pod metrics scraper: appendable storage is required")
	}
	if cfg.Clock == nil {
		return nil, fmt.Errorf("pod metrics scraper: clock is required")
	}
	if cfg.Timeout <= 0 {
		return nil, fmt.Errorf("pod metrics scraper: timeout must be greater than zero")
	}
	if cfg.MaxBodySize <= 0 || cfg.MaxBodySize > maxBodySize {
		return nil, fmt.Errorf("pod metrics scraper: max body size must be between 1 and %d bytes", maxBodySize)
	}
	if cfg.Concurrency <= 0 {
		return nil, fmt.Errorf("pod metrics scraper: concurrency must be greater than zero")
	}
	if err := validateMetricNames(cfg.Metrics); err != nil {
		return nil, fmt.Errorf("pod metrics scraper: %w", err)
	}
	metrics := make(map[string]struct{}, len(cfg.Metrics))
	for _, name := range cfg.Metrics {
		metrics[name] = struct{}{}
	}
	httpClient := *cfg.HTTPClient
	if httpClient.Transport == nil {
		transport := http.DefaultTransport.(*http.Transport).Clone()
		transport.Proxy = nil
		httpClient.Transport = transport
	}
	httpClient.CheckRedirect = func(_ *http.Request, _ []*http.Request) error {
		return http.ErrUseLastResponse
	}
	return &PodMetrics{
		httpClient:    &httpClient,
		discovery:     cfg.Discovery,
		reader:        cfg.Reader,
		targetKind:    cfg.TargetKind,
		appendable:    cfg.Appendable,
		clock:         cfg.Clock,
		metrics:       metrics,
		timeout:       cfg.Timeout,
		maxBodySize:   cfg.MaxBodySize,
		concurrency:   cfg.Concurrency,
		activeTargets: make(map[string]map[string]promlabels.Labels),
		lastTimestamp: make(map[string]int64),
	}, nil
}

// Scrape resolves matching Pods, fetches every annotated endpoint, and commits
// all successful responses in one storage transaction. A failed endpoint does
// not block successful Pods and retains its last series until a later
// successful scrape or target removal.
func (s *PodMetrics) 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("pod metrics scraper: resolve targets: %w", err)
	}
	results := s.scrapePods(ctx, pods)

	nextTargets := make(map[string]map[string]promlabels.Labels, len(results))
	currentSamples := make(map[string]podSample)
	var scrapeErrors []error
	for i := range results {
		result := &results[i]
		if result.err != nil {
			scrapeErrors = append(scrapeErrors, result.err)
			if previous, exists := s.activeTargets[result.targetKey]; exists {
				nextTargets[result.targetKey] = previous
			}
			continue
		}
		series := make(map[string]promlabels.Labels, len(result.samples))
		for seriesKey, sample := range result.samples {
			series[seriesKey] = sample.labels
			currentSamples[seriesKey] = sample
		}
		nextTargets[result.targetKey] = series
	}

	nextSeries := flattenTargets(nextTargets)
	staleSeries := make(map[string]promlabels.Labels)
	for seriesKey, metricLabels := range flattenTargets(s.activeTargets) {
		if _, active := nextSeries[seriesKey]; !active {
			staleSeries[seriesKey] = metricLabels
		}
	}
	if err := s.append(ctx, currentSamples, staleSeries, nextSeries); err != nil {
		return errors.Join(append(scrapeErrors, err)...)
	}
	s.activeTargets = nextTargets
	return errors.Join(scrapeErrors...)
}

func (s *PodMetrics) scrapePods(ctx context.Context, pods []*corev1.Pod) []podScrapeResult {
	selected := make([]*corev1.Pod, 0, len(pods))
	for _, pod := range pods {
		if strings.EqualFold(strings.TrimSpace(pod.Annotations[AnnotationScrape]), "true") {
			selected = append(selected, pod)
		}
	}
	results := make([]podScrapeResult, len(selected))
	jobs := make(chan int)
	workerCount := min(s.concurrency, len(selected))
	var workers sync.WaitGroup
	workers.Add(workerCount)
	for range workerCount {
		go func() {
			defer workers.Done()
			for index := range jobs {
				pod := selected[index]
				key := podKey(pod)
				samples, err := s.scrapePod(ctx, pod)
				if err != nil {
					err = fmt.Errorf("pod metrics scraper: scrape Pod %s/%s: %w", pod.Namespace, pod.Name, err)
				}
				results[index] = podScrapeResult{targetKey: key, samples: samples, err: err}
			}
		}()
	}
	for index := range selected {
		jobs <- index
	}
	close(jobs)
	workers.Wait()
	return results
}

func (s *PodMetrics) scrapePod(ctx context.Context, pod *corev1.Pod) (map[string]podSample, error) {
	endpoint, err := endpointForPod(pod)
	if err != nil {
		return nil, err
	}
	requestCtx, cancel := context.WithTimeout(ctx, s.timeout)
	defer cancel()
	req, err := http.NewRequestWithContext(requestCtx, http.MethodGet, endpoint.String(), http.NoBody)
	if err != nil {
		return nil, fmt.Errorf("build request: %w", err)
	}
	req.Header.Set("Accept", acceptHeader)
	req.Header.Set("Accept-Encoding", "identity")
	req.Header.Set("User-Agent", "capacity-planner")
	req.Header.Set("X-Prometheus-Scrape-Timeout-Seconds", strconv.FormatFloat(s.timeout.Seconds(), 'f', -1, 64))

	response, err := s.httpClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("GET %s: %w", endpoint.Redacted(), err)
	}
	defer response.Body.Close()
	if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices {
		return nil, fmt.Errorf("GET %s: unexpected HTTP status %s", endpoint.Redacted(), response.Status)
	}
	body, err := io.ReadAll(io.LimitReader(response.Body, s.maxBodySize+1))
	if err != nil {
		return nil, fmt.Errorf("read %s: %w", endpoint.Redacted(), err)
	}
	if int64(len(body)) > s.maxBodySize {
		return nil, fmt.Errorf("read %s: response exceeds %d bytes", endpoint.Redacted(), s.maxBodySize)
	}
	samples, err := parseSamples(body, response.Header.Get("Content-Type"), pod, s.metrics)
	if err != nil {
		return nil, fmt.Errorf("parse %s: %w", endpoint.Redacted(), err)
	}
	return samples, nil
}

func endpointForPod(pod *corev1.Pod) (*url.URL, error) {
	podIP := strings.TrimSpace(pod.Status.PodIP)
	if net.ParseIP(podIP) == nil {
		return nil, fmt.Errorf("invalid Pod IP %q", pod.Status.PodIP)
	}
	port, err := strconv.ParseUint(strings.TrimSpace(pod.Annotations[AnnotationPort]), 10, 16)
	if err != nil || port == 0 {
		return nil, fmt.Errorf("annotation %s must be a port between 1 and 65535", AnnotationPort)
	}
	scheme := strings.ToLower(strings.TrimSpace(pod.Annotations[AnnotationScheme]))
	if scheme == "" {
		scheme = "http"
	}
	if scheme != "http" && scheme != "https" {
		return nil, fmt.Errorf("annotation %s must be http or https", AnnotationScheme)
	}
	metricsPath := strings.TrimSpace(pod.Annotations[AnnotationPath])
	if metricsPath == "" {
		metricsPath = defaultMetricsPath
	}
	parsedPath, err := url.ParseRequestURI(metricsPath)
	if err != nil || parsedPath.IsAbs() || parsedPath.Host != "" || parsedPath.Opaque != "" || parsedPath.Fragment != "" || !strings.HasPrefix(parsedPath.Path, "/") {
		return nil, fmt.Errorf("annotation %s must be an absolute request path", AnnotationPath)
	}
	return &url.URL{
		Scheme:   scheme,
		Host:     net.JoinHostPort(podIP, strconv.FormatUint(port, 10)),
		Path:     parsedPath.Path,
		RawPath:  parsedPath.RawPath,
		RawQuery: parsedPath.RawQuery,
	}, nil
}

func parseSamples(
	body []byte,
	contentType string,
	pod *corev1.Pod,
	configuredMetrics map[string]struct{},
) (map[string]podSample, error) {
	parser, warning := textparse.New(body, contentType, promlabels.NewSymbolTable(), textparse.ParserOptions{
		IgnoreNativeHistograms: true,
		FallbackContentType:    "text/plain",
	})
	if parser == nil {
		return nil, warning
	}
	samples := make(map[string]podSample)
	for {
		entry, err := parser.Next()
		if errors.Is(err, io.EOF) {
			return samples, nil
		}
		if err != nil {
			return nil, err
		}
		if entry != textparse.EntrySeries {
			if entry == textparse.EntryHistogram {
				return nil, fmt.Errorf("native histogram samples are not supported")
			}
			continue
		}
		_, timestamp, sampleValue := parser.Series()
		var sampleTimestamp *int64
		if timestamp != nil {
			copiedTimestamp := *timestamp
			sampleTimestamp = &copiedTimestamp
		}
		var parsedLabels promlabels.Labels
		parser.Labels(&parsedLabels)
		metricLabels := labelsForPod(parsedLabels, pod)
		if !metricLabels.Has(commonmodel.MetricNameLabel) {
			return nil, fmt.Errorf("sample is missing metric name")
		}
		if len(configuredMetrics) != 0 {
			if _, configured := configuredMetrics[metricLabels.Get(commonmodel.MetricNameLabel)]; !configured {
				continue
			}
		}
		seriesKey := metricLabels.String()
		if _, duplicate := samples[seriesKey]; duplicate {
			return nil, fmt.Errorf("duplicate series %s", seriesKey)
		}
		samples[seriesKey] = podSample{labels: metricLabels, timestamp: sampleTimestamp, value: sampleValue}
	}
}

func labelsForPod(scraped promlabels.Labels, pod *corev1.Pod) promlabels.Labels {
	return promlabels.NewBuilder(scraped).
		Set(decisionmetrics.LabelNamespace, pod.Namespace).
		Set(decisionmetrics.LabelPod, pod.Name).
		Set(metricstarget.LabelGroup, pod.Labels[metricstarget.PodLabelGroup]).
		Set(metricstarget.LabelComponent, pod.Labels[metricstarget.PodLabelComponent]).
		Labels()
}

func (s *PodMetrics) append(
	ctx context.Context,
	current map[string]podSample,
	stale map[string]promlabels.Labels,
	nextSeries map[string]promlabels.Labels,
) error {
	appender := s.appendable.Appender(ctx)
	appended := false
	pendingTimestamps := make(map[string]int64, len(current)+len(stale))
	now := s.clock.Now().UnixMilli()
	for _, seriesKey := range sortedKeys(current) {
		sample := current[seriesKey]
		timestamp, shouldAppend := s.sampleTimestamp(seriesKey, sample.timestamp, now)
		if !shouldAppend {
			continue
		}
		_, err := appender.Append(0, sample.labels, timestamp, sample.value)
		if isIgnorableAppendError(err) {
			pendingTimestamps[seriesKey] = timestamp
			continue
		}
		if err != nil {
			_ = appender.Rollback()
			return fmt.Errorf("append %s: %w", seriesKey, err)
		}
		appended = true
		pendingTimestamps[seriesKey] = timestamp
	}
	for _, seriesKey := range sortedKeys(stale) {
		timestamp := s.nextTimestamp(seriesKey, now)
		_, err := appender.Append(0, stale[seriesKey], timestamp, math.Float64frombits(value.StaleNaN))
		if isIgnorableAppendError(err) {
			pendingTimestamps[seriesKey] = timestamp
			continue
		}
		if err != nil {
			_ = appender.Rollback()
			return fmt.Errorf("append stale marker for %s: %w", seriesKey, err)
		}
		appended = true
		pendingTimestamps[seriesKey] = timestamp
	}
	if !appended {
		if err := appender.Rollback(); err != nil {
			return fmt.Errorf("rollback empty transaction: %w", err)
		}
	} else if err := appender.Commit(); err != nil {
		return fmt.Errorf("commit samples: %w", err)
	}

	nextTimestamps := make(map[string]int64, len(nextSeries))
	for seriesKey := range nextSeries {
		if timestamp, exists := pendingTimestamps[seriesKey]; exists {
			nextTimestamps[seriesKey] = timestamp
		} else if timestamp, exists := s.lastTimestamp[seriesKey]; exists {
			nextTimestamps[seriesKey] = timestamp
		}
	}
	s.lastTimestamp = nextTimestamps
	return nil
}

func (s *PodMetrics) sampleTimestamp(seriesKey string, explicit *int64, now int64) (int64, bool) {
	if explicit == nil {
		return s.nextTimestamp(seriesKey, now), true
	}
	previous, exists := s.lastTimestamp[seriesKey]
	return *explicit, !exists || *explicit > previous
}

func (s *PodMetrics) nextTimestamp(seriesKey string, now int64) int64 {
	if previous := s.lastTimestamp[seriesKey]; now <= previous {
		return previous + 1
	}
	return now
}

func flattenTargets(targets map[string]map[string]promlabels.Labels) map[string]promlabels.Labels {
	series := make(map[string]promlabels.Labels)
	for _, targetSeries := range targets {
		for seriesKey, metricLabels := range targetSeries {
			series[seriesKey] = metricLabels
		}
	}
	return series
}

func sortedKeys[T any](values map[string]T) []string {
	keys := make([]string, 0, len(values))
	for key := range values {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	return keys
}

func podKey(pod *corev1.Pod) string {
	return pod.Namespace + "\x00" + pod.Name
}

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