// Package podwatcher converts configured fields from target Pods into
// Prometheus time series.
package podwatcher

import (
	"context"
	"errors"
	"fmt"
	"maps"
	"math"
	"reflect"
	"sort"
	"strings"
	"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"
	"k8s.io/client-go/util/jsonpath"
	"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 samples.
type Clock interface {
	Now() time.Time
}

// MetricConfig maps JSONPath results from a Pod to a gauge. JSONPath omits the
// outer template braces, for example ".status.phase". The emitted value is 1
// when any scalar result equals MatchValue and 0 otherwise.
type MetricConfig struct {
	Name       string
	JSONPath   string
	MatchValue string
}

// Config configures Pod field conversion for one-shot scrapes.
type Config struct {
	Discovery  discovery.Discovery
	Reader     client.Reader
	TargetKind string
	Appendable storage.Appendable
	Clock      Clock
	Metrics    []MetricConfig
}

// PodWatcher resolves one discovery category and converts its Pods to
// configured metrics. Scrape scheduling is owned by scraper.Manager.
type PodWatcher struct {
	discovery  discovery.Discovery
	reader     client.Reader
	targetKind string
	appendable storage.Appendable
	clock      Clock
	metrics    []statusMetric

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

type statusMetric struct {
	name       string
	jsonPath   *jsonpath.JSONPath
	matchValue string
}

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

// New constructs a PodWatcher with immutable metric configurations.
func New(cfg *Config) (*PodWatcher, error) {
	if cfg == nil {
		return nil, fmt.Errorf("pod watcher scraper: config is required")
	}
	if cfg.Discovery == nil {
		return nil, fmt.Errorf("pod watcher scraper: target discovery is required")
	}
	if cfg.Reader == nil {
		return nil, fmt.Errorf("pod watcher scraper: Kubernetes reader is required")
	}
	if cfg.Appendable == nil {
		return nil, fmt.Errorf("pod watcher scraper: appendable storage is required")
	}
	if cfg.Clock == nil {
		return nil, fmt.Errorf("pod watcher scraper: clock is required")
	}
	if len(cfg.Metrics) == 0 {
		return nil, fmt.Errorf("pod watcher scraper: metrics are required")
	}

	metrics := make([]statusMetric, 0, len(cfg.Metrics))
	names := make(map[string]struct{}, len(cfg.Metrics))
	for i := range cfg.Metrics {
		configured := &cfg.Metrics[i]
		compiled, err := compileMetricConfig(configured)
		if err != nil {
			return nil, fmt.Errorf("pod watcher scraper: metrics[%d]: %w", i, err)
		}
		if _, exists := names[configured.Name]; exists {
			return nil, fmt.Errorf("pod watcher scraper: duplicate metric name %q", configured.Name)
		}
		names[configured.Name] = struct{}{}
		metrics = append(metrics, compiled)
	}
	sort.Slice(metrics, func(i, j int) bool { return metrics[i].name < metrics[j].name })

	return &PodWatcher{
		discovery:     cfg.Discovery,
		reader:        cfg.Reader,
		targetKind:    cfg.TargetKind,
		appendable:    cfg.Appendable,
		clock:         cfg.Clock,
		metrics:       metrics,
		activeSeries:  make(map[string]promlabels.Labels),
		lastTimestamp: make(map[string]int64),
	}, nil
}

func compileMetricConfig(metric *MetricConfig) (statusMetric, error) {
	if !commonmodel.LegacyValidation.IsValidMetricName(metric.Name) {
		return statusMetric{}, fmt.Errorf("invalid metric name %q", metric.Name)
	}
	expression := strings.TrimSpace(metric.JSONPath)
	if expression == "" {
		return statusMetric{}, fmt.Errorf("metric %q JSONPath is required", metric.Name)
	}
	if strings.HasPrefix(expression, "{") || strings.HasSuffix(expression, "}") {
		return statusMetric{}, fmt.Errorf("metric %q JSONPath must omit outer braces", metric.Name)
	}
	compiled := jsonpath.New(metric.Name).AllowMissingKeys(true)
	if err := compiled.Parse("{" + expression + "}"); err != nil {
		return statusMetric{}, fmt.Errorf("metric %q has invalid JSONPath: %w", metric.Name, err)
	}
	return statusMetric{
		name:       metric.Name,
		jsonPath:   compiled,
		matchValue: metric.MatchValue,
	}, nil
}

func (m *statusMetric) value(pod *corev1.Pod) (float64, error) {
	results, err := m.jsonPath.FindResults(pod)
	if err != nil {
		return 0, err
	}
	for i := range results {
		for j := range results[i] {
			if scalarString(results[i][j]) == m.matchValue {
				return 1, nil
			}
		}
	}
	return 0, nil
}

func scalarString(value reflect.Value) string {
	for value.IsValid() && (value.Kind() == reflect.Interface || value.Kind() == reflect.Pointer) {
		if value.IsNil() {
			return ""
		}
		value = value.Elem()
	}
	if !value.IsValid() || !value.CanInterface() {
		return ""
	}
	return fmt.Sprint(value.Interface())
}

// Scrape resolves this scraper's target category and converts the matching Pods
// in one storage transaction. An empty target set marks prior series stale.
func (w *PodWatcher) Scrape(ctx context.Context) error {
	w.mu.Lock()
	defer w.mu.Unlock()

	targets := w.discovery.Targets(w.targetKind)
	pods, err := metricstarget.ListPods(ctx, w.reader, targets...)
	if err != nil {
		return fmt.Errorf("pod watcher scraper: resolve targets: %w", err)
	}
	desiredSeries := make(map[string]promlabels.Labels, len(pods)*len(w.metrics))
	values := make(map[string]float64, len(pods)*len(w.metrics))
	for _, pod := range pods {
		for i := range w.metrics {
			metricValue, err := w.metrics[i].value(pod)
			if err != nil {
				return fmt.Errorf("pod watcher scraper: evaluate metric %q for Pod %s/%s: %w", w.metrics[i].name, pod.Namespace, pod.Name, err)
			}
			metricLabels := labelsForPodStatus(w.metrics[i].name, pod)
			seriesKey := metricLabels.String()
			desiredSeries[seriesKey] = metricLabels
			values[seriesKey] = metricValue
		}
	}

	currentKeys := sortedSeriesKeys(desiredSeries)
	staleKeys := make([]string, 0, len(w.activeSeries))
	for seriesKey := range w.activeSeries {
		if _, exists := desiredSeries[seriesKey]; !exists {
			staleKeys = append(staleKeys, seriesKey)
		}
	}
	sort.Strings(staleKeys)

	appender := w.appendable.Appender(ctx)
	appended := false
	pendingTimestamps := make(map[string]int64, len(currentKeys)+len(staleKeys))
	now := w.clock.Now().UnixMilli()
	for _, seriesKey := range currentKeys {
		timestamp := w.nextTimestamp(seriesKey, now)
		_, err := appender.Append(0, desiredSeries[seriesKey], timestamp, values[seriesKey])
		if isIgnorableAppendError(err) {
			pendingTimestamps[seriesKey] = timestamp
			continue
		}
		if err != nil {
			_ = appender.Rollback()
			return fmt.Errorf("pod watcher scraper: append %s: %w", seriesKey, err)
		}
		appended = true
		pendingTimestamps[seriesKey] = timestamp
	}
	for _, seriesKey := range staleKeys {
		timestamp := w.nextTimestamp(seriesKey, now)
		_, err := appender.Append(0, w.activeSeries[seriesKey], timestamp, math.Float64frombits(value.StaleNaN))
		if isIgnorableAppendError(err) {
			pendingTimestamps[seriesKey] = timestamp
			continue
		}
		if err != nil {
			_ = appender.Rollback()
			return fmt.Errorf("pod watcher scraper: append stale marker for %s: %w", seriesKey, err)
		}
		appended = true
		pendingTimestamps[seriesKey] = timestamp
	}
	if err := finishAppend(appender, appended); err != nil {
		return fmt.Errorf("pod watcher scraper: %w", err)
	}
	maps.Copy(w.lastTimestamp, pendingTimestamps)
	w.activeSeries = desiredSeries
	return nil
}

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

func sortedSeriesKeys(series map[string]promlabels.Labels) []string {
	keys := make([]string, 0, len(series))
	for key := range series {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	return keys
}

func labelsForPodStatus(metricName string, pod *corev1.Pod) promlabels.Labels {
	return promlabels.FromMap(map[string]string{
		commonmodel.MetricNameLabel:    metricName,
		decisionmetrics.LabelNamespace: pod.Namespace,
		metricstarget.LabelGroup:       pod.Labels[metricstarget.PodLabelGroup],
		metricstarget.LabelComponent:   pod.Labels[metricstarget.PodLabelComponent],
		decisionmetrics.LabelPod:       pod.Name,
	})
}

func finishAppend(appender storage.Appender, appended bool) error {
	if !appended {
		if err := appender.Rollback(); err != nil {
			return fmt.Errorf("rollback empty transaction: %w", err)
		}
		return nil
	}
	if err := appender.Commit(); err != nil {
		return fmt.Errorf("commit samples: %w", err)
	}
	return nil
}

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