package podmetrics

import (
	"context"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"slices"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/prometheus/prometheus/tsdb"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"

	querymetrics "github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/prom/query"
	metricsstore "github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/prom/store"
	metricstarget "github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/target"
)

type mutableClock struct{ now time.Time }

func (c *mutableClock) Now() time.Time { return c.now }

type staticTarget struct {
	namespace string
	labels    map[string]string
}

func (t staticTarget) Namespace() string         { return t.namespace }
func (t staticTarget) Labels() map[string]string { return t.labels }

type staticDiscovery struct {
	kind    string
	targets []metricstarget.Target
}

func (d *staticDiscovery) Targets(kind string) []metricstarget.Target {
	d.kind = kind
	return d.targets
}

type mutableHandler struct {
	mu           sync.Mutex
	body         string
	status       int
	request      *http.Request
	requestCount int
}

func (h *mutableHandler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.request = request.Clone(request.Context())
	h.requestCount++
	writer.Header().Set("Content-Type", "text/plain; version=0.0.4")
	writer.WriteHeader(h.status)
	_, _ = io.WriteString(writer, h.body)
}

func (h *mutableHandler) update(status int, body string) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.status = status
	h.body = body
}

func (h *mutableHandler) lastRequest() *http.Request {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.request
}

func (h *mutableHandler) requests() int {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.requestCount
}

func TestPodMetricsScrapesAnnotatedPodAndMaintainsStaleness(t *testing.T) {
	handler := &mutableHandler{
		status: http.StatusOK,
		body: `# TYPE app_requests_total counter
app_requests_total{code="200",namespace="spoof",pod="spoof",group="spoof",component="spoof"} 3
# TYPE queue_depth gauge
queue_depth 2
`,
	}
	server := httptest.NewServer(handler)
	defer server.Close()
	host, port := serverAddress(t, server.URL)

	pod := targetPod("tidb-0", host, port)
	pod.Annotations[AnnotationPath] = "/custom/metrics?scope=server"
	unannotated := targetPod("tidb-1", host, port)
	delete(unannotated.Annotations, AnnotationScrape)
	target := staticTarget{namespace: "dp-x", labels: map[string]string{
		metricstarget.LabelGroup:     "tidb",
		metricstarget.LabelComponent: "tidb",
	}}
	discovered := &staticDiscovery{targets: []metricstarget.Target{target}}
	clock := &mutableClock{now: time.Date(2026, 9, 7, 8, 0, 0, 0, time.UTC)}
	db := openTestDB(t)
	scraper := newTestScraper(t, server.Client(), discovered, clock, db, pod, unannotated)
	provider := newTestProvider(t, db, clock)

	if err := scraper.Scrape(context.Background()); err != nil {
		t.Fatalf("first Scrape() error = %v", err)
	}
	if discovered.kind != "TiDBGroup" {
		t.Fatalf("discovery kind = %q, want TiDBGroup", discovered.kind)
	}
	assertSample(t, provider, target, `app_requests_total{code="200"}`, "tidb-0", 3)
	assertSample(t, provider, target, "queue_depth", "tidb-0", 2)
	assertNoSeries(t, provider, target, `app_requests_total{namespace="spoof"}`)
	request := handler.lastRequest()
	if request == nil || request.URL.RequestURI() != "/custom/metrics?scope=server" {
		t.Fatalf("request URI = %v, want /custom/metrics?scope=server", request)
	}
	if handler.requests() != 1 {
		t.Fatalf("request count = %d, want only the annotated Pod", handler.requests())
	}
	if !strings.Contains(request.Header.Get("Accept"), "application/openmetrics-text") {
		t.Fatalf("Accept = %q, want OpenMetrics support", request.Header.Get("Accept"))
	}
	if request.Header.Get("X-Prometheus-Scrape-Timeout-Seconds") != "1" {
		t.Fatalf("scrape timeout header = %q, want 1", request.Header.Get("X-Prometheus-Scrape-Timeout-Seconds"))
	}

	handler.update(http.StatusOK, "app_requests_total{code=\"200\"} 4\n")
	clock.now = clock.now.Add(time.Second)
	if err := scraper.Scrape(context.Background()); err != nil {
		t.Fatalf("second Scrape() error = %v", err)
	}
	assertSample(t, provider, target, `app_requests_total{code="200"}`, "tidb-0", 4)
	assertNoSeries(t, provider, target, "queue_depth")

	handler.update(http.StatusInternalServerError, "failed")
	clock.now = clock.now.Add(time.Second)
	err := scraper.Scrape(context.Background())
	if err == nil || !strings.Contains(err.Error(), "unexpected HTTP status 500 Internal Server Error") {
		t.Fatalf("failed Scrape() error = %v, want HTTP status error", err)
	}
	assertSample(t, provider, target, `app_requests_total{code="200"}`, "tidb-0", 4)

	discovered.targets = nil
	clock.now = clock.now.Add(time.Second)
	if err := scraper.Scrape(context.Background()); err != nil {
		t.Fatalf("removed target Scrape() error = %v", err)
	}
	assertNoSeries(t, provider, target, `app_requests_total{code="200"}`)
}

func TestPodMetricsCommitsSuccessfulPodsWhenAnotherPodFails(t *testing.T) {
	transport := roundTripFunc(func(request *http.Request) (*http.Response, error) {
		if request.URL.Hostname() == "10.0.0.1" {
			return metricsResponse(http.StatusOK, "healthy_metric 7\n"), nil
		}
		return metricsResponse(http.StatusServiceUnavailable, "unavailable"), nil
	})
	target := staticTarget{namespace: "dp-x", labels: map[string]string{
		metricstarget.LabelGroup:     "tidb",
		metricstarget.LabelComponent: "tidb",
	}}
	discovered := &staticDiscovery{targets: []metricstarget.Target{target}}
	clock := &mutableClock{now: time.Date(2026, 9, 7, 8, 0, 0, 0, time.UTC)}
	db := openTestDB(t)
	scraper := newTestScraper(t, &http.Client{Transport: transport}, discovered, clock, db,
		targetPod("tidb-0", "10.0.0.1", "10080"),
		targetPod("tidb-1", "10.0.0.2", "10080"),
	)

	err := scraper.Scrape(context.Background())
	if err == nil || !strings.Contains(err.Error(), "tidb-1") {
		t.Fatalf("Scrape() error = %v, want failed Pod identity", err)
	}
	assertSample(t, newTestProvider(t, db, clock), target, "healthy_metric", "tidb-0", 7)
}

func TestPodMetricsAppliesConfiguredMetricAllowlist(t *testing.T) {
	transport := roundTripFunc(func(_ *http.Request) (*http.Response, error) {
		return metricsResponse(http.StatusOK, "configured_metric 7\nignored_metric 9\n"), nil
	})
	target := staticTarget{namespace: "dp-x", labels: map[string]string{
		metricstarget.LabelGroup:     "tidb",
		metricstarget.LabelComponent: "tidb",
	}}
	discovered := &staticDiscovery{targets: []metricstarget.Target{target}}
	clock := &mutableClock{now: time.Date(2026, 9, 7, 8, 0, 0, 0, time.UTC)}
	db := openTestDB(t)
	scraper := newTestScraperWithMetrics(t, &http.Client{Transport: transport}, discovered, clock, db,
		[]string{"configured_metric"}, targetPod("tidb-0", "10.0.0.1", "10080"))

	if err := scraper.Scrape(context.Background()); err != nil {
		t.Fatalf("Scrape() error = %v", err)
	}
	provider := newTestProvider(t, db, clock)
	assertSample(t, provider, target, "configured_metric", "tidb-0", 7)
	assertNoSeries(t, provider, target, "ignored_metric")
}

func TestPodMetricsRejectsOversizedResponse(t *testing.T) {
	transport := roundTripFunc(func(_ *http.Request) (*http.Response, error) {
		return metricsResponse(http.StatusOK, "metric 12345\n"), nil
	})
	target := staticTarget{namespace: "dp-x", labels: map[string]string{
		metricstarget.LabelGroup:     "tidb",
		metricstarget.LabelComponent: "tidb",
	}}
	clock := &mutableClock{now: time.Now()}
	db := openTestDB(t)
	scraper := newTestScraper(t, &http.Client{Transport: transport},
		&staticDiscovery{targets: []metricstarget.Target{target}}, clock, db,
		targetPod("tidb-0", "10.0.0.1", "10080"),
	)
	scraper.maxBodySize = 4
	if err := scraper.Scrape(context.Background()); err == nil || !strings.Contains(err.Error(), "response exceeds 4 bytes") {
		t.Fatalf("Scrape() error = %v, want body size error", err)
	}
	assertNoSeries(t, newTestProvider(t, db, clock), target, "metric")
}

func TestPodMetricsDoesNotFollowRedirects(t *testing.T) {
	redirected := false
	transport := roundTripFunc(func(request *http.Request) (*http.Response, error) {
		if request.URL.Hostname() == "10.0.0.1" {
			response := metricsResponse(http.StatusFound, "")
			response.Header.Set("Location", "http://169.254.169.254/metrics")
			response.Request = request
			return response, nil
		}
		redirected = true
		return metricsResponse(http.StatusOK, "unexpected_metric 1\n"), nil
	})
	target := staticTarget{namespace: "dp-x", labels: map[string]string{
		metricstarget.LabelGroup:     "tidb",
		metricstarget.LabelComponent: "tidb",
	}}
	clock := &mutableClock{now: time.Now()}
	db := openTestDB(t)
	scraper := newTestScraper(t, &http.Client{Transport: transport},
		&staticDiscovery{targets: []metricstarget.Target{target}}, clock, db,
		targetPod("tidb-0", "10.0.0.1", "10080"),
	)
	if err := scraper.Scrape(context.Background()); err == nil || !strings.Contains(err.Error(), "302 Found") {
		t.Fatalf("Scrape() error = %v, want redirect status error", err)
	}
	if redirected {
		t.Fatal("scraper followed a redirect outside the Pod endpoint")
	}
}

func TestParseSamplesSupportsOpenMetricsAndRejectsDuplicates(t *testing.T) {
	pod := targetPod("tidb-0", "10.0.0.1", "10080")
	samples, err := parseSamples([]byte(`# TYPE request_duration_seconds histogram
request_duration_seconds_bucket{le="1"} 2
request_duration_seconds_bucket{le="+Inf"} 3
request_duration_seconds_sum 1.5
request_duration_seconds_count 3
# EOF
`), "application/openmetrics-text; version=1.0.0; charset=utf-8", pod, nil)
	if err != nil {
		t.Fatalf("parseSamples() error = %v", err)
	}
	if len(samples) != 4 {
		t.Fatalf("sample count = %d, want 4", len(samples))
	}
	for _, sample := range samples {
		if sample.labels.Get("namespace") != "dp-x" || sample.labels.Get("pod") != "tidb-0" ||
			sample.labels.Get("group") != "tidb" || sample.labels.Get("component") != "tidb" {
			t.Fatalf("sample labels = %s, want target Pod identity labels", sample.labels)
		}
	}

	_, err = parseSamples([]byte("duplicate_metric 1\nduplicate_metric 2\n"), "text/plain", pod, nil)
	if err == nil || !strings.Contains(err.Error(), "duplicate series") {
		t.Fatalf("duplicate parse error = %v, want duplicate series error", err)
	}

	timestamped, err := parseSamples([]byte("first_metric 1 1000\nsecond_metric 2 2000\n"), "text/plain", pod, nil)
	if err != nil {
		t.Fatalf("parse timestamped samples: %v", err)
	}
	var timestamps []int64
	for _, sample := range timestamped {
		timestamps = append(timestamps, *sample.timestamp)
	}
	slices.Sort(timestamps)
	if !slices.Equal(timestamps, []int64{1000, 2000}) {
		t.Fatalf("timestamps = %v, want [1000 2000]", timestamps)
	}

	filtered, err := parseSamples(
		[]byte("configured_metric 1\nignored_metric 2\n"),
		"text/plain",
		pod,
		map[string]struct{}{"configured_metric": {}},
	)
	if err != nil {
		t.Fatalf("parse configured samples: %v", err)
	}
	if len(filtered) != 1 {
		t.Fatalf("filtered sample count = %d, want 1", len(filtered))
	}
	for _, sample := range filtered {
		if sample.labels.Get("__name__") != "configured_metric" {
			t.Fatalf("filtered sample = %s, want configured_metric", sample.labels)
		}
	}
}

func TestEndpointForPod(t *testing.T) {
	defaultPod := targetPod("tidb-0", "10.0.0.1", "10080")
	defaultEndpoint, err := endpointForPod(defaultPod)
	if err != nil {
		t.Fatalf("endpointForPod(defaults) error = %v", err)
	}
	if got, want := defaultEndpoint.String(), "http://10.0.0.1:10080/metrics"; got != want {
		t.Fatalf("default endpoint = %q, want %q", got, want)
	}

	pod := targetPod("tidb-0", "2001:db8::1", "10080")
	pod.Annotations[AnnotationScheme] = "HTTPS"
	pod.Annotations[AnnotationPath] = "/metrics/component?format=prometheus"
	endpoint, err := endpointForPod(pod)
	if err != nil {
		t.Fatalf("endpointForPod() error = %v", err)
	}
	if got, want := endpoint.String(), "https://[2001:db8::1]:10080/metrics/component?format=prometheus"; got != want {
		t.Fatalf("endpoint = %q, want %q", got, want)
	}

	tests := []struct {
		name   string
		mutate func(*corev1.Pod)
	}{
		{name: "missing IP", mutate: func(pod *corev1.Pod) { pod.Status.PodIP = "" }},
		{name: "non-IP address", mutate: func(pod *corev1.Pod) { pod.Status.PodIP = "example.com" }},
		{name: "missing port", mutate: func(pod *corev1.Pod) { delete(pod.Annotations, AnnotationPort) }},
		{name: "zero port", mutate: func(pod *corev1.Pod) { pod.Annotations[AnnotationPort] = "0" }},
		{name: "overflowing port", mutate: func(pod *corev1.Pod) { pod.Annotations[AnnotationPort] = "65536" }},
		{name: "invalid scheme", mutate: func(pod *corev1.Pod) { pod.Annotations[AnnotationScheme] = "file" }},
		{name: "relative path", mutate: func(pod *corev1.Pod) { pod.Annotations[AnnotationPath] = "metrics" }},
		{name: "external URL", mutate: func(pod *corev1.Pod) { pod.Annotations[AnnotationPath] = "http://example.com/metrics" }},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			invalid := targetPod("tidb-0", "10.0.0.1", "10080")
			tt.mutate(invalid)
			if _, err := endpointForPod(invalid); err == nil {
				t.Fatal("endpointForPod() must reject invalid Pod annotations")
			}
		})
	}
}

func TestNewRejectsInvalidConfig(t *testing.T) {
	db := openTestDB(t)
	base := &Config{
		HTTPClient:  http.DefaultClient,
		Discovery:   &staticDiscovery{},
		Reader:      fake.NewClientBuilder().Build(),
		Appendable:  db,
		Clock:       &mutableClock{now: time.Now()},
		Timeout:     DefaultTimeout,
		MaxBodySize: DefaultMaxBodySize,
		Concurrency: DefaultConcurrency,
	}
	if _, err := New(base); err != nil {
		t.Fatalf("New(valid) error = %v", err)
	}

	tests := []struct {
		name   string
		mutate func(*Config)
	}{
		{name: "missing HTTP client", mutate: func(cfg *Config) { cfg.HTTPClient = nil }},
		{name: "missing discovery", mutate: func(cfg *Config) { cfg.Discovery = nil }},
		{name: "missing reader", mutate: func(cfg *Config) { cfg.Reader = nil }},
		{name: "missing storage", mutate: func(cfg *Config) { cfg.Appendable = nil }},
		{name: "missing clock", mutate: func(cfg *Config) { cfg.Clock = nil }},
		{name: "invalid timeout", mutate: func(cfg *Config) { cfg.Timeout = 0 }},
		{name: "invalid body size", mutate: func(cfg *Config) { cfg.MaxBodySize = 0 }},
		{name: "excessive body size", mutate: func(cfg *Config) { cfg.MaxBodySize = maxBodySize + 1 }},
		{name: "invalid concurrency", mutate: func(cfg *Config) { cfg.Concurrency = 0 }},
		{name: "invalid metric", mutate: func(cfg *Config) { cfg.Metrics = []string{"bad-metric"} }},
		{name: "duplicate metric", mutate: func(cfg *Config) {
			cfg.Metrics = []string{"configured_metric", "configured_metric"}
		}},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cfg := *base
			tt.mutate(&cfg)
			if _, err := New(&cfg); err == nil {
				t.Fatal("New() must reject invalid config")
			}
		})
	}
	if _, err := New(nil); err == nil {
		t.Fatal("New(nil) must reject missing config")
	}
}

func newTestScraper(
	t *testing.T,
	httpClient *http.Client,
	discovery *staticDiscovery,
	clock *mutableClock,
	db *tsdb.DB,
	pods ...*corev1.Pod,
) *PodMetrics {
	t.Helper()
	return newTestScraperWithMetrics(t, httpClient, discovery, clock, db, nil, pods...)
}

func newTestScraperWithMetrics(
	t *testing.T,
	httpClient *http.Client,
	discovery *staticDiscovery,
	clock *mutableClock,
	db *tsdb.DB,
	metrics []string,
	pods ...*corev1.Pod,
) *PodMetrics {
	t.Helper()
	scheme := runtime.NewScheme()
	if err := corev1.AddToScheme(scheme); err != nil {
		t.Fatal(err)
	}
	objects := make([]runtime.Object, len(pods))
	for i := range pods {
		objects[i] = pods[i]
	}
	configured, err := New(&Config{
		HTTPClient:  httpClient,
		Discovery:   discovery,
		Reader:      fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(objects...).Build(),
		TargetKind:  "TiDBGroup",
		Appendable:  db,
		Clock:       clock,
		Metrics:     metrics,
		Timeout:     time.Second,
		MaxBodySize: 1024 * 1024,
		Concurrency: 2,
	})
	if err != nil {
		t.Fatalf("New() error = %v", err)
	}
	return configured
}

func targetPod(name, ip, port string) *corev1.Pod {
	return &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: "dp-x",
			Labels: map[string]string{
				metricstarget.PodLabelGroup:     "tidb",
				metricstarget.PodLabelComponent: "tidb",
			},
			Annotations: map[string]string{
				AnnotationScrape: "true",
				AnnotationPort:   port,
			},
		},
		Status: corev1.PodStatus{PodIP: ip},
	}
}

func serverAddress(t *testing.T, serverURL string) (string, string) {
	t.Helper()
	hostPort := strings.TrimPrefix(serverURL, "http://")
	host, port, err := net.SplitHostPort(hostPort)
	if err != nil {
		t.Fatalf("parse test server URL: %v", err)
	}
	return host, port
}

type roundTripFunc func(request *http.Request) (*http.Response, error)

func (fn roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) {
	return fn(request)
}

func metricsResponse(status int, body string) *http.Response {
	return &http.Response{
		StatusCode: status,
		Status:     strconv.Itoa(status) + " " + http.StatusText(status),
		Header:     http.Header{"Content-Type": []string{"text/plain; version=0.0.4"}},
		Body:       io.NopCloser(strings.NewReader(body)),
	}
}

func newTestProvider(t *testing.T, db *tsdb.DB, clock *mutableClock) *querymetrics.Provider {
	t.Helper()
	provider, err := querymetrics.New(&querymetrics.Config{Queryable: db, Clock: clock})
	if err != nil {
		t.Fatalf("new PromQL provider: %v", err)
	}
	return provider
}

func assertSample(t *testing.T, provider *querymetrics.Provider, target metricstarget.Target, query, pod string, value float64) {
	t.Helper()
	result, err := provider.Query(context.Background(), target, query)
	if err != nil {
		t.Fatalf("Query(%s) error = %v", query, err)
	}
	if len(result) != 1 || len(result[0].Series) != 1 || result[0].Series[0].Ref == nil ||
		result[0].Series[0].Ref.Name != pod || len(result[0].Series[0].Samples) != 1 {
		t.Fatalf("Query(%s) = %+v, want one %s sample", query, result, pod)
	}
	if got := result[0].Series[0].Samples[0].Value; got != value {
		t.Fatalf("Query(%s) value = %v, want %v", query, got, value)
	}
}

func assertNoSeries(t *testing.T, provider *querymetrics.Provider, target metricstarget.Target, query string) {
	t.Helper()
	result, err := provider.Query(context.Background(), target, query)
	if err != nil {
		t.Fatalf("Query(%s) error = %v", query, err)
	}
	if len(result) != 1 || len(result[0].Series) != 0 {
		t.Fatalf("Query(%s) = %+v, want no series", query, result)
	}
}

func openTestDB(t *testing.T) *tsdb.DB {
	t.Helper()
	db, err := metricsstore.Open(&metricsstore.Config{Path: t.TempDir(), Retention: time.Hour})
	if err != nil {
		t.Fatalf("open TSDB: %v", err)
	}
	t.Cleanup(func() {
		if err := db.Close(); err != nil {
			t.Errorf("close TSDB: %v", err)
		}
	})
	return db
}