package service

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"net/http/httptest"
	"sync/atomic"
	"testing"
	"time"

	pingcaplog "github.com/pingcap/log"
	"github.com/prometheus/prometheus/model/labels"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"go.uber.org/zap/zaptest/observer"

	clinic_internal_api "github.com/pingcap-inc/clinic/clinic_data_service/pkg/clinic-internal-api"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/config"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
)

func TestNewMetricsQueryServiceUsesQueryTimeout(t *testing.T) {
	svc := NewMetricsQueryService(config.DefaultConfig())
	require.Equal(t, 2*time.Minute, svc.httpClient.Timeout)
}

type metricsRoundTripFunc func(*http.Request) (*http.Response, error)

func (f metricsRoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
	return f(req)
}

func TestMetricsQueryServiceBYOCUsesAuthoritativeRoutingForAllPrometheusEndpoints(t *testing.T) {
	tiers := []struct {
		name         string
		clusterID    string
		deployType   string
		deployTypeV2 string
	}{
		{name: "logical", clusterID: "logical-cluster", deployType: "byoc", deployTypeV2: "byoc"},
		{name: "physical", clusterID: "physical-cluster", deployType: "byoc-host", deployTypeV2: "byoc resource pool"},
	}
	sources := []struct {
		name       string
		metric     string
		pathPrefix string
	}{
		{name: "cluster", metric: "cluster_metric", pathPrefix: "/cluster/api/v1/"},
		{name: "data-plane", metric: `kube_pod_info{namespace="tidbphysical-cluster"}`, pathPrefix: "/data-plane/api/v1/"},
	}
	operations := []struct {
		name       string
		pathSuffix string
		execute    func(*MetricsQueryService, string, string, *clinic_internal_api.ClusterDataURLs) *dto.PrometheusResponse
	}{
		{
			name:       "instant",
			pathSuffix: "query",
			execute: func(svc *MetricsQueryService, clusterID, metric string, target *clinic_internal_api.ClusterDataURLs) *dto.PrometheusResponse {
				return svc.InstantQueryWithResolvedTarget(context.Background(), &dto.InstantQueryRequest{ClusterID: clusterID, Query: metric}, target)
			},
		},
		{
			name:       "range",
			pathSuffix: "query_range",
			execute: func(svc *MetricsQueryService, clusterID, metric string, target *clinic_internal_api.ClusterDataURLs) *dto.PrometheusResponse {
				return svc.RangeQueryWithResolvedTarget(context.Background(), &dto.RangeQueryRequest{
					ClusterID: clusterID,
					Query:     metric,
					Start:     time.Unix(100, 0),
					End:       time.Unix(200, 0),
					Step:      time.Minute,
				}, target)
			},
		},
		{
			name:       "series",
			pathSuffix: "series",
			execute: func(svc *MetricsQueryService, clusterID, metric string, target *clinic_internal_api.ClusterDataURLs) *dto.PrometheusResponse {
				return svc.SeriesWithResolvedTarget(context.Background(), &dto.SeriesRequest{ClusterID: clusterID, Match: []string{metric}}, target)
			},
		},
	}

	for _, tier := range tiers {
		for _, source := range sources {
			for _, operation := range operations {
				t.Run(tier.name+"/"+source.name+"/"+operation.name, func(t *testing.T) {
					var gotPath string
					var gotHeaders http.Header
					var gotPromQL string
					upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
						gotPath = r.URL.Path
						gotHeaders = r.Header.Clone()
						gotPromQL = r.URL.Query().Get("query")
						if gotPromQL == "" {
							gotPromQL = r.URL.Query().Get("match[]")
						}
						writePrometheusSuccess(t, w, map[string]any{"resultType": "vector", "result": []any{}})
					}))
					t.Cleanup(upstream.Close)

					cluster := &clinic_internal_api.ClusterDataURLs{
						ClusterID:           tier.clusterID,
						OrgID:               "different-dbaas-org",
						ClinicTenantID:      "123",
						Provider:            "aws",
						Region:              "us-west-2",
						DeployType:          tier.deployType,
						DeployTypeV2:        tier.deployTypeV2,
						ParentID:            "physical-parent",
						ClusterMetricsURL:   upstream.URL + "/cluster/api/v1/",
						DataPlaneMetricsURL: upstream.URL + "/data-plane/api/v1/",
					}
					svc := &MetricsQueryService{
						httpClient:       upstream.Client(),
						dataPlaneMetrics: map[string]bool{"kube_pod_info": true},
					}

					resp := operation.execute(svc, cluster.ClusterID, source.metric, cluster)

					require.Equal(t, "success", resp.Status, "response: %#v", resp)
					assert.Equal(t, source.pathPrefix+operation.pathSuffix, gotPath)
					assert.Equal(t, "byoc", gotHeaders.Get("X-Biz-Type"))
					assert.Equal(t, "123", gotHeaders.Get("X-Org-Id"))
					assert.Equal(t, "aws-us-west-2", gotHeaders.Get("X-Region-Id"))
					assert.Empty(t, gotHeaders.Get("X-Env"))
					assert.Empty(t, gotHeaders.Get("X-Application-Id"))
					if source.name == "data-plane" {
						assert.Equal(t, source.metric, gotPromQL)
					} else {
						assert.Equal(t, source.metric+`{tidb_cluster_id="`+tier.clusterID+`"}`, gotPromQL)
					}
				})
			}
		}
	}
}

func TestMetricsQueryServiceBYOCDataPlaneExceptionDoesNotChangePremium(t *testing.T) {
	svc := &MetricsQueryService{}
	target := &clinic_internal_api.ClusterDataURLs{
		ClusterID:    "premium-cluster",
		ParentID:     "premium-resource-pool",
		DeployType:   "premium",
		DeployTypeV2: "premium",
	}
	query := `kube_pod_info{namespace="tidbpremium-resource-pool"}`

	got, err := svc.buildQueryWithLabels(
		context.Background(),
		query,
		target.ClusterID,
		target,
		dto.MetricCategoryDataPlane,
		time.Time{},
		time.Time{},
	)

	require.NoError(t, err)
	assert.Equal(t, `kube_pod_info{namespace="tidbpremium-resource-pool",tidb_cluster_id="premium-resource-pool"}`, got)
}

func TestMetricsQueryServiceDoesNotAddBYOCHeadersForOtherTiers(t *testing.T) {
	for _, tier := range []struct {
		name         string
		deployType   string
		deployTypeV2 string
	}{
		{name: "dedicated", deployType: "dedicated", deployTypeV2: "dedicated"},
		{name: "premium", deployType: "premium", deployTypeV2: "premium"},
		{name: "starter", deployType: "serverless", deployTypeV2: "starter"},
		{name: "essential", deployType: "serverless", deployTypeV2: "essential"},
		{name: "inconsistent-byoc", deployType: "byoc", deployTypeV2: "byoc resource pool"},
		{name: "inconsistent-byoc-host", deployType: "byoc-host", deployTypeV2: "byoc"},
	} {
		t.Run(tier.name, func(t *testing.T) {
			var gotHeaders http.Header
			upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				gotHeaders = r.Header.Clone()
				writePrometheusSuccess(t, w, map[string]any{"resultType": "vector", "result": []any{}})
			}))
			t.Cleanup(upstream.Close)
			target := &clinic_internal_api.ClusterDataURLs{
				ClusterID:         "cluster-a",
				OrgID:             "123",
				ClinicTenantID:    "456",
				Provider:          "aws",
				Region:            "us-west-2",
				DeployType:        tier.deployType,
				DeployTypeV2:      tier.deployTypeV2,
				ClusterMetricsURL: upstream.URL + "/api/v1/",
			}
			svc := &MetricsQueryService{httpClient: upstream.Client()}

			resp := svc.executeQueryWithValues(
				context.Background(),
				target.ClusterMetricsURL+"query",
				nil,
				target,
			)

			require.Equal(t, "success", resp.Status, "response: %#v", resp)
			assert.Empty(t, gotHeaders.Get("X-Biz-Type"))
			assert.Empty(t, gotHeaders.Get("X-Org-Id"))
			assert.Empty(t, gotHeaders.Get("X-Region-Id"))
		})
	}
}

func TestMetricsQueryServiceBYOCInvalidServerRoutingFailsClosed(t *testing.T) {
	var upstreamCalls atomic.Int32
	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		upstreamCalls.Add(1)
		writePrometheusSuccess(t, w, map[string]any{"resultType": "vector", "result": []any{}})
	}))
	t.Cleanup(upstream.Close)
	baseTarget := clinic_internal_api.ClusterDataURLs{
		ClusterID:         "logical-cluster",
		OrgID:             "123",
		ClinicTenantID:    "456",
		Provider:          "aws",
		Region:            "us-west-2",
		DeployType:        "byoc",
		DeployTypeV2:      "byoc",
		ClusterMetricsURL: upstream.URL + "/api/v1/",
	}
	svc := &MetricsQueryService{httpClient: upstream.Client()}

	for _, tt := range []struct {
		name   string
		mutate func(*clinic_internal_api.ClusterDataURLs)
	}{
		{name: "missing tenant", mutate: func(target *clinic_internal_api.ClusterDataURLs) { target.ClinicTenantID = "" }},
		{name: "non-numeric tenant", mutate: func(target *clinic_internal_api.ClusterDataURLs) { target.ClinicTenantID = "tenant" }},
		{name: "zero tenant", mutate: func(target *clinic_internal_api.ClusterDataURLs) { target.ClinicTenantID = "0" }},
		{name: "malformed provider", mutate: func(target *clinic_internal_api.ClusterDataURLs) { target.Provider = "aws/us" }},
		{name: "malformed region", mutate: func(target *clinic_internal_api.ClusterDataURLs) { target.Region = "us west 2" }},
	} {
		t.Run(tt.name, func(t *testing.T) {
			target := baseTarget
			tt.mutate(&target)
			resp := svc.InstantQueryWithResolvedTarget(context.Background(), &dto.InstantQueryRequest{
				ClusterID: target.ClusterID,
				Query:     "up",
			}, &target)

			require.Equal(t, "error", resp.Status)
			assert.Equal(t, "internal", resp.ErrorType)
			assert.Contains(t, resp.Error, errInvalidBYOCMetricsTarget.Error())
			assert.Equal(t, int32(0), upstreamCalls.Load())
		})
	}
}

func TestInjectLabelIntoPromQL(t *testing.T) {
	tests := []struct {
		name      string
		query     string
		labelName string
		labelVal  string
		want      string
	}{
		{
			name:      "empty label name returns query unchanged",
			query:     "up",
			labelName: "",
			labelVal:  "v1",
			want:      "up",
		},
		{
			name:      "empty label value returns query unchanged",
			query:     "up",
			labelName: "tidb_cluster_id",
			labelVal:  "",
			want:      "up",
		},
		{
			name:      "bare metric no selector",
			query:     "up",
			labelName: "tidb_cluster_id",
			labelVal:  "123",
			want:      `up{tidb_cluster_id="123"}`,
		},
		{
			name:      "metric with empty selector",
			query:     "tidb_server_connections{}",
			labelName: "keyspace_id",
			labelVal:  "ks-1",
			want:      `tidb_server_connections{keyspace_id="ks-1"}`,
		},
		{
			name:      "metric with existing single label",
			query:     `tidb_server_connections{cluster_id="c1"}`,
			labelName: "keyspace_id",
			labelVal:  "ks-1",
			want:      `tidb_server_connections{cluster_id="c1",keyspace_id="ks-1"}`,
		},
		{
			name:      "metric with existing multiple labels",
			query:     `tidb_server_query_total{instance="db-tidb-0", job="tidb"}`,
			labelName: "tidb_cluster_id",
			labelVal:  "1379661944646416027",
			want:      `tidb_server_query_total{instance="db-tidb-0",job="tidb",tidb_cluster_id="1379661944646416027"}`,
		},
		{
			name:      "wrapped in rate() no selector",
			query:     "rate(tidb_server_handle_query_duration_seconds_bucket[5m])",
			labelName: "keyspace_id",
			labelVal:  "ks-1",
			want:      `rate(tidb_server_handle_query_duration_seconds_bucket{keyspace_id="ks-1"}[5m])`,
		},
		{
			name:      "wrapped in rate() with selector",
			query:     `rate(tidb_server_handle_query_duration_seconds_bucket{instance=~"db-tidb-.*"}[5m])`,
			labelName: "tidb_cluster_id",
			labelVal:  "123",
			want:      `rate(tidb_server_handle_query_duration_seconds_bucket{instance=~"db-tidb-.*",tidb_cluster_id="123"}[5m])`,
		},
		{
			name:      "sum() wrapped metric",
			query:     "sum(tidb_server_query_total)",
			labelName: "keyspace_id",
			labelVal:  "ks-2",
			want:      `sum(tidb_server_query_total{keyspace_id="ks-2"})`,
		},
		{
			name:      "sum rate with selector injects into inner metric",
			query:     `sum(rate(tikv_grpc_msg_duration_seconds_bucket{type="coprocessor"}[2m]))`,
			labelName: "tidb_cluster_id",
			labelVal:  "456",
			// Parser outputs labels in sorted order
			want: `sum(rate(tikv_grpc_msg_duration_seconds_bucket{tidb_cluster_id="456",type="coprocessor"}[2m]))`,
		},
		{
			name:      "metric name with underscores",
			query:     "node_memory_MemTotal_bytes",
			labelName: "tidb_cluster_id",
			labelVal:  "789",
			want:      `node_memory_MemTotal_bytes{tidb_cluster_id="789"}`,
		},
		{
			name:      "label value with alphanumeric",
			query:     "up",
			labelName: "keyspace_id",
			labelVal:  "ks-abc-123",
			want:      `up{keyspace_id="ks-abc-123"}`,
		},
		{
			name:      "multiple metrics each get label",
			query:     "metric_a + metric_b * 2",
			labelName: "tidb_cluster_id",
			labelVal:  "cid-1",
			want:      `metric_a{tidb_cluster_id="cid-1"} + metric_b{tidb_cluster_id="cid-1"} * 2`,
		},
		{
			name:      "multiple metrics with selectors",
			query:     `sum(rate(foo{job="a"}[5m])) / sum(rate(bar{job="b"}[5m]))`,
			labelName: "keyspace_id",
			labelVal:  "ks",
			want:      `sum(rate(foo{job="a",keyspace_id="ks"}[5m])) / sum(rate(bar{job="b",keyspace_id="ks"}[5m]))`,
		},
		{
			name:      "by(name) and without(label) are not metrics",
			query:     `sum(increase(foo{ns="x"}[1m])) by(name)`,
			labelName: "keyspace_id",
			labelVal:  "554458",
			// Parser normalizes to "sum by (name) (...)" and may reorder labels
			want: `sum by (name) (increase(foo{keyspace_id="554458",ns="x"}[1m]))`,
		},
		{
			name:      "on(instance) modifier: parser returns query unchanged (grouping modifier not expanded)",
			query:     `metric_a + metric_b on(instance)`,
			labelName: "tidb_cluster_id",
			labelVal:  "c1",
			// Parser may return expression unchanged when grouping modifier is present
			want: `metric_a + metric_b on(instance)`,
		},
		{
			name:      "selector already has label is unchanged",
			query:     `up{keyspace_id="existing"}`,
			labelName: "keyspace_id",
			labelVal:  "new",
			want:      `up{keyspace_id="existing"}`,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			defaultLabel := MetricLabel{Name: tt.labelName, MatchType: labels.MatchEqual, Value: tt.labelVal}
			got := injectLabelIntoPromQL(tt.query, defaultLabel, nil)
			assert.Equal(t, tt.want, got, "injectLabelIntoPromQL(%q, %+v, nil)", tt.query, defaultLabel)
		})
	}
}

func TestInjectLabelIntoPromQL_PerMetricMap(t *testing.T) {
	// Per-metric map: metric_a gets keyspace_id=ks-a, metric_b gets keyspace_id=ks-b; others use default keyspace_id=default.
	perMetric := MetricLabelMap{
		"metric_a": {Name: "keyspace_id", MatchType: labels.MatchRegexp, Value: "ks-a"},
		"metric_b": {Name: "keyspace_id", MatchType: labels.MatchEqual, Value: "ks-b"},
		"tikv_*":   {Name: "control_plane_info", MatchType: labels.MatchEqual, Value: "us-east-1-f01/aws-us-east-1"},
	}
	defaultLabel := MetricLabel{Name: "keyspace_id", MatchType: labels.MatchEqual, Value: "default"}

	got := injectLabelIntoPromQL("metric_a + metric_b + other_metric", defaultLabel, perMetric)
	// metric_a -> keyspace_id=ks-a, metric_b -> keyspace_id=ks-b, other_metric -> keyspace_id=default
	assert.Contains(t, got, `metric_a{keyspace_id=~"ks-a"}`)
	assert.Contains(t, got, `metric_b{keyspace_id="ks-b"}`)
	assert.Contains(t, got, `other_metric{keyspace_id="default"}`)

	got = injectLabelIntoPromQL("tikv_grpc_msg_duration_seconds_count + tikv_grpc_msg_duration_seconds_sum", defaultLabel, perMetric)
	assert.Contains(t, got, `tikv_grpc_msg_duration_seconds_count{control_plane_info="us-east-1-f01/aws-us-east-1"}`)
	assert.Contains(t, got, `tikv_grpc_msg_duration_seconds_sum{control_plane_info="us-east-1-f01/aws-us-east-1"}`)
}

func TestMetricsQueryService_AzureInstantQueryUsesClusterMetricsURL(t *testing.T) {
	var gotPath, gotQuery, gotTime string
	queryTime := time.Date(2026, 3, 6, 22, 0, 0, 0, time.UTC)
	svc := newAzureMetricsQueryTestService(t, clinic_internal_api.ClusterDataURLs{
		ClusterID:    "azure-cluster",
		Provider:     "azure",
		Region:       "southeastasia",
		DeployTypeV2: "dedicated",
	}, func(w http.ResponseWriter, r *http.Request) {
		gotPath = r.URL.Path
		gotQuery = r.URL.Query().Get("query")
		gotTime = r.URL.Query().Get("time")
		writePrometheusSuccess(t, w, map[string]any{
			"resultType": "vector",
			"result":     []any{},
		})
	})

	resp := svc.InstantQuery(context.Background(), &dto.InstantQueryRequest{
		ClusterID: "azure-cluster",
		Query:     "sum(tidb_server_query_total)",
		Time:      queryTime,
	})

	require.Equal(t, "success", resp.Status)
	assert.Equal(t, "/api/v1/query", gotPath)
	assert.Equal(t, "sum(tidb_server_query_total)", gotQuery)
	assert.Equal(t, "1772834400", gotTime)
	assert.Equal(t, "azure", resp.Provider)
	assert.Equal(t, "southeastasia", resp.Region)
	assert.Equal(t, "dedicated", resp.DeployTypeV2)
}

func TestMetricsQueryService_AzureRangeQueryForwardsRangeParams(t *testing.T) {
	var gotPath, gotQuery, gotStart, gotEnd, gotStep string
	start := time.Date(2026, 3, 6, 21, 0, 0, 0, time.UTC)
	end := time.Date(2026, 3, 6, 22, 0, 0, 0, time.UTC)
	svc := newAzureMetricsQueryTestService(t, clinic_internal_api.ClusterDataURLs{
		ClusterID:    "azure-cluster",
		Provider:     "azure",
		Region:       "southeastasia",
		DeployTypeV2: "dedicated",
	}, func(w http.ResponseWriter, r *http.Request) {
		gotPath = r.URL.Path
		gotQuery = r.URL.Query().Get("query")
		gotStart = r.URL.Query().Get("start")
		gotEnd = r.URL.Query().Get("end")
		gotStep = r.URL.Query().Get("step")
		writePrometheusSuccess(t, w, map[string]any{
			"resultType": "matrix",
			"result":     []any{},
		})
	})

	resp := svc.RangeQuery(context.Background(), &dto.RangeQueryRequest{
		ClusterID: "azure-cluster",
		Query:     "sum(tidb_server_query_total)",
		Start:     start,
		End:       end,
		Step:      15 * time.Second,
	})

	require.Equal(t, "success", resp.Status)
	assert.Equal(t, "/api/v1/query_range", gotPath)
	assert.Equal(t, "sum(tidb_server_query_total)", gotQuery)
	assert.Equal(t, "1772830800", gotStart)
	assert.Equal(t, "1772834400", gotEnd)
	assert.Equal(t, "15s", gotStep)
	assert.Equal(t, "azure", resp.Provider)
	assert.Equal(t, "southeastasia", resp.Region)
}

func TestMetricsQueryService_AzureDataPlaneMetricUsesDataPlaneURL(t *testing.T) {
	clusterCalled := false
	clusterMetricsServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		clusterCalled = true
		writePrometheusSuccess(t, w, map[string]any{
			"resultType": "vector",
			"result":     []any{},
		})
	}))
	t.Cleanup(clusterMetricsServer.Close)

	var dataPlanePath, dataPlaneQuery string
	dataPlaneServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		dataPlanePath = r.URL.Path
		dataPlaneQuery = r.URL.Query().Get("query")
		writePrometheusSuccess(t, w, map[string]any{
			"resultType": "vector",
			"result":     []any{},
		})
	}))
	t.Cleanup(dataPlaneServer.Close)

	cluster := clinic_internal_api.ClusterDataURLs{
		ClusterID:           "azure-cluster",
		Provider:            "azure",
		Region:              "southeastasia",
		DeployTypeV2:        "dedicated",
		ClusterMetricsURL:   clusterMetricsServer.URL + "/api/v1/",
		DataPlaneMetricsURL: dataPlaneServer.URL + "/api/v1/",
	}
	clinicServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, "/clinic/api/v1/internal/clusters/azure-cluster/cluster-dataurls", r.URL.Path)
		w.Header().Set("Content-Type", "application/json")
		require.NoError(t, json.NewEncoder(w).Encode(cluster))
	}))
	t.Cleanup(clinicServer.Close)

	svc := &MetricsQueryService{
		clinicClient: clinic_internal_api.NewClient(clinicServer.URL),
		httpClient:   dataPlaneServer.Client(),
		dataPlaneMetrics: map[string]bool{
			"node_memory_MemTotal_bytes": true,
		},
	}

	resp := svc.InstantQuery(context.Background(), &dto.InstantQueryRequest{
		ClusterID: "azure-cluster",
		Query:     "node_memory_MemTotal_bytes",
	})

	require.Equal(t, "success", resp.Status)
	assert.False(t, clusterCalled)
	assert.Equal(t, "/api/v1/query", dataPlanePath)
	assert.Equal(t, "node_memory_MemTotal_bytes", dataPlaneQuery)
	assert.Equal(t, "azure", resp.Provider)
	assert.Equal(t, "southeastasia", resp.Region)
}

func TestMetricsQueryService_GetMetricCategoryUsesPromQLSelectors(t *testing.T) {
	svc := &MetricsQueryService{
		dataPlaneMetrics: map[string]bool{
			"kube_*":        true,
			"node_*":        true,
			"aws_ebs_csi_*": true,
		},
	}

	tests := []struct {
		name  string
		query string
		want  dto.MetricCategory
	}{
		{
			name:  "bare kube metric",
			query: `kube_pod_info{namespace="tidb10518506906906971664",pod=~"db-tikv-.*"}`,
			want:  dto.MetricCategoryDataPlane,
		},
		{
			name:  "group by wrapped kube metric",
			query: `group by(node) (kube_pod_info{namespace="tidb10518506906906971664", pod=~"db-tikv-.*"})`,
			want:  dto.MetricCategoryDataPlane,
		},
		{
			name:  "sum by wrapped node metric",
			query: `sum by(instance) (rate(node_disk_reads_completed_total{device!~"dm.*"}[5m]))`,
			want:  dto.MetricCategoryDataPlane,
		},
		{
			name: "project node to pod join",
			query: `sum(
				sum by(instance) (rate(node_disk_reads_completed_total{device!~"dm.*"}[5m]))
				* on(instance) group_right()
				label_replace(
					kube_pod_info{namespace="tidb10518506906906971664", pod=~"db-tikv-.*"},
					"instance", "$1", "node", "(.*)"
				)
			)`,
			want: dto.MetricCategoryDataPlane,
		},
		{
			name:  "topk with scalar argument",
			query: `topk(3, rate(aws_ebs_csi_exceeded_iops_seconds_total[5m]))`,
			want:  dto.MetricCategoryDataPlane,
		},
		{
			name:  "cluster metric remains cluster",
			query: `sum(rate(tidb_server_query_total[1m])) by (type)`,
			want:  dto.MetricCategoryCluster,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			assert.Equal(t, tt.want, svc.GetMetricCategory(tt.query))
		})
	}
}

func TestMetricsQueryService_ProjectJoinMetricUsesDataPlaneURL(t *testing.T) {
	clusterCalled := false
	clusterMetricsServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		clusterCalled = true
		writePrometheusSuccess(t, w, map[string]any{
			"resultType": "matrix",
			"result":     []any{},
		})
	}))
	t.Cleanup(clusterMetricsServer.Close)

	var dataPlanePath, dataPlaneQuery string
	dataPlaneServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		dataPlanePath = r.URL.Path
		dataPlaneQuery = r.URL.Query().Get("query")
		writePrometheusSuccess(t, w, map[string]any{
			"resultType": "matrix",
			"result":     []any{},
		})
	}))
	t.Cleanup(dataPlaneServer.Close)

	cluster := clinic_internal_api.ClusterDataURLs{
		ClusterID:           "dedicated-cluster",
		Provider:            "aws",
		Region:              "us-east-1",
		DeployTypeV2:        "dedicated",
		ClusterMetricsURL:   clusterMetricsServer.URL + "/api/v1/",
		DataPlaneMetricsURL: dataPlaneServer.URL + "/api/v1/",
	}
	clinicServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, "/clinic/api/v1/internal/clusters/dedicated-cluster/cluster-dataurls", r.URL.Path)
		w.Header().Set("Content-Type", "application/json")
		require.NoError(t, json.NewEncoder(w).Encode(cluster))
	}))
	t.Cleanup(clinicServer.Close)

	svc := &MetricsQueryService{
		clinicClient: clinic_internal_api.NewClient(clinicServer.URL),
		httpClient:   dataPlaneServer.Client(),
		dataPlaneMetrics: map[string]bool{
			"kube_*": true,
			"node_*": true,
		},
	}

	query := `sum(
		sum by(instance) (rate(node_disk_reads_completed_total{device!~"dm.*"}[5m]))
		* on(instance) group_right()
		label_replace(
			kube_pod_info{namespace="tidb10518506906906971664", pod=~"db-tikv-.*"},
			"instance", "$1", "node", "(.*)"
		)
	)`
	resp := svc.RangeQuery(context.Background(), &dto.RangeQueryRequest{
		ClusterID: "dedicated-cluster",
		Query:     query,
		Start:     time.Date(2026, 5, 27, 2, 0, 0, 0, time.UTC),
		End:       time.Date(2026, 5, 27, 2, 30, 0, 0, time.UTC),
		Step:      time.Minute,
	})

	require.Equal(t, "success", resp.Status)
	assert.False(t, clusterCalled)
	assert.Equal(t, "/api/v1/query_range", dataPlanePath)
	assert.Equal(t, query, dataPlaneQuery)
}

func TestMetricsQueryService_CrossSourceMetricRejectedWithReason(t *testing.T) {
	clusterCalled := false
	clusterMetricsServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		clusterCalled = true
		writePrometheusSuccess(t, w, map[string]any{
			"resultType": "matrix",
			"result":     []any{},
		})
	}))
	t.Cleanup(clusterMetricsServer.Close)

	dataPlaneCalled := false
	dataPlaneServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		dataPlaneCalled = true
		writePrometheusSuccess(t, w, map[string]any{
			"resultType": "matrix",
			"result":     []any{},
		})
	}))
	t.Cleanup(dataPlaneServer.Close)

	cluster := clinic_internal_api.ClusterDataURLs{
		ClusterID:           "dedicated-cluster",
		Provider:            "aws",
		Region:              "us-east-1",
		DeployTypeV2:        "dedicated",
		ClusterMetricsURL:   clusterMetricsServer.URL + "/api/v1/",
		DataPlaneMetricsURL: dataPlaneServer.URL + "/api/v1/",
	}
	clinicServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, "/clinic/api/v1/internal/clusters/dedicated-cluster/cluster-dataurls", r.URL.Path)
		w.Header().Set("Content-Type", "application/json")
		require.NoError(t, json.NewEncoder(w).Encode(cluster))
	}))
	t.Cleanup(clinicServer.Close)

	svc := &MetricsQueryService{
		clinicClient: clinic_internal_api.NewClient(clinicServer.URL),
		httpClient:   dataPlaneServer.Client(),
		dataPlaneMetrics: map[string]bool{
			"node_*": true,
		},
	}

	resp := svc.RangeQuery(context.Background(), &dto.RangeQueryRequest{
		ClusterID: "dedicated-cluster",
		Query:     `sum(rate(node_disk_reads_completed_total[5m])) + sum(rate(tidb_server_query_total[5m]))`,
		Start:     time.Date(2026, 5, 27, 2, 0, 0, 0, time.UTC),
		End:       time.Date(2026, 5, 27, 2, 30, 0, 0, time.UTC),
		Step:      time.Minute,
	})

	require.Equal(t, "error", resp.Status)
	assert.Equal(t, "unsupported_query", resp.ErrorType)
	assert.Contains(t, resp.Error, "unsupported cross-source metrics query")
	assert.Contains(t, resp.Error, "Clinic Data Proxy cannot execute one PromQL expression across both data-plane and cluster metrics")
	assert.Contains(t, resp.Error, "node_disk_reads_completed_total")
	assert.Contains(t, resp.Error, "tidb_server_query_total")
	assert.False(t, clusterCalled)
	assert.False(t, dataPlaneCalled)
	assert.Equal(t, "aws", resp.Provider)
	assert.Equal(t, "us-east-1", resp.Region)
}

func TestMetricsQueryService_AzurePremiumInjectsTidbClusterID(t *testing.T) {
	var gotQuery string
	svc := newAzureMetricsQueryTestService(t, clinic_internal_api.ClusterDataURLs{
		ClusterID:    "azure-cluster",
		Provider:     "azure",
		Region:       "southeastasia",
		DeployTypeV2: "premium",
	}, func(w http.ResponseWriter, r *http.Request) {
		gotQuery = r.URL.Query().Get("query")
		writePrometheusSuccess(t, w, map[string]any{
			"resultType": "vector",
			"result":     []any{},
		})
	})

	resp := svc.InstantQuery(context.Background(), &dto.InstantQueryRequest{
		ClusterID: "azure-cluster",
		Query:     "sum(tidb_server_query_total)",
	})

	require.Equal(t, "success", resp.Status)
	assert.Equal(t, `sum(tidb_server_query_total{tidb_cluster_id="azure-cluster"})`, gotQuery)
	assert.Equal(t, "azure", resp.Provider)
	assert.Equal(t, "southeastasia", resp.Region)
	assert.Equal(t, "premium", resp.DeployTypeV2)
}

func TestMetricsQueryService_StarterInstantQueryReturnsEmptyVectorWhenMetricLabelsNoSeries(t *testing.T) {
	var queryCalled atomic.Bool
	svc := newAzureMetricsQueryTestService(t, clinic_internal_api.ClusterDataURLs{
		ClusterID:    "starter-cluster",
		Provider:     "aws",
		Region:       "us-west-2",
		DeployTypeV2: "starter",
	}, func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/api/v1/series":
			assert.Equal(t, `tidb_server_connections{cluster_id="starter-cluster"}`, r.URL.Query().Get("match[]"))
			writePrometheusSuccess(t, w, []map[string]string{})
		case "/api/v1/query":
			queryCalled.Store(true)
			writePrometheusSuccess(t, w, map[string]any{
				"resultType": "vector",
				"result":     []any{},
			})
		default:
			require.Failf(t, "unexpected upstream path", "path=%s", r.URL.Path)
		}
	})

	resp := svc.InstantQuery(context.Background(), &dto.InstantQueryRequest{
		ClusterID: "starter-cluster",
		Query:     "sum(tidb_server_query_total)",
	})

	require.Equal(t, "success", resp.Status)
	assertEmptyPrometheusQueryResult(t, resp.Data, "vector")
	assert.False(t, queryCalled.Load())
	assert.Equal(t, "aws", resp.Provider)
	assert.Equal(t, "us-west-2", resp.Region)
	assert.Equal(t, "starter", resp.DeployTypeV2)
}

func TestMetricsQueryService_StarterRangeQueryReturnsEmptyMatrixWhenMetricLabelsNoSeries(t *testing.T) {
	var queryRangeCalled atomic.Bool
	start := time.Date(2026, 3, 6, 21, 0, 0, 0, time.UTC)
	end := time.Date(2026, 3, 6, 22, 0, 0, 0, time.UTC)
	svc := newAzureMetricsQueryTestService(t, clinic_internal_api.ClusterDataURLs{
		ClusterID:    "starter-cluster",
		Provider:     "aws",
		Region:       "us-west-2",
		DeployTypeV2: "essential",
	}, func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/api/v1/series":
			assert.Equal(t, `tidb_server_connections{cluster_id="starter-cluster"}`, r.URL.Query().Get("match[]"))
			assert.Equal(t, "1772830800", r.URL.Query().Get("start"))
			assert.Equal(t, "1772834400", r.URL.Query().Get("end"))
			writePrometheusSuccess(t, w, []map[string]string{})
		case "/api/v1/query_range":
			queryRangeCalled.Store(true)
			writePrometheusSuccess(t, w, map[string]any{
				"resultType": "matrix",
				"result":     []any{},
			})
		default:
			require.Failf(t, "unexpected upstream path", "path=%s", r.URL.Path)
		}
	})

	resp := svc.RangeQuery(context.Background(), &dto.RangeQueryRequest{
		ClusterID: "starter-cluster",
		Query:     "sum(rate(tidb_server_query_total[1m]))",
		Start:     start,
		End:       end,
		Step:      time.Minute,
	})

	require.Equal(t, "success", resp.Status)
	assertEmptyPrometheusQueryResult(t, resp.Data, "matrix")
	assert.False(t, queryRangeCalled.Load())
	assert.Equal(t, "essential", resp.DeployTypeV2)
}

func TestMetricsQueryService_StarterSeriesReturnsEmptyArrayWhenMetricLabelsNoSeries(t *testing.T) {
	var downstreamSeriesCalls atomic.Int32
	start := time.Date(2026, 3, 6, 21, 0, 0, 0, time.UTC)
	end := time.Date(2026, 3, 6, 22, 0, 0, 0, time.UTC)
	svc := newAzureMetricsQueryTestService(t, clinic_internal_api.ClusterDataURLs{
		ClusterID:    "starter-cluster",
		Provider:     "aws",
		Region:       "us-west-2",
		DeployTypeV2: "starter",
	}, func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/api/v1/series":
			downstreamSeriesCalls.Add(1)
			assert.Equal(t, `tidb_server_connections{cluster_id="starter-cluster"}`, r.URL.Query().Get("match[]"))
			writePrometheusSuccess(t, w, []map[string]string{})
		default:
			require.Failf(t, "unexpected upstream path", "path=%s", r.URL.Path)
		}
	})

	resp := svc.Series(context.Background(), &dto.SeriesRequest{
		ClusterID: "starter-cluster",
		Match:     []string{"tidb_server_query_total"},
		Start:     start,
		End:       end,
	})

	require.Equal(t, "success", resp.Status)
	assert.Equal(t, []any{}, resp.Data)
	assert.Equal(t, int32(1), downstreamSeriesCalls.Load())
	assert.Equal(t, "starter", resp.DeployTypeV2)
}

func TestMetricsQueryServiceDoesNotLogRawPromQLOnTransportFailure(t *testing.T) {
	core, observed := observer.New(zapcore.InfoLevel)
	restoreLogger := pingcaplog.ReplaceGlobals(
		zap.New(core),
		&pingcaplog.ZapProperties{Core: core, Level: zap.NewAtomicLevelAt(zapcore.InfoLevel)},
	)
	t.Cleanup(restoreLogger)

	const secretMarker = "supersecretpromqlmarker"
	svc := &MetricsQueryService{
		httpClient: &http.Client{Transport: metricsRoundTripFunc(func(*http.Request) (*http.Response, error) {
			return nil, errors.New("dial failed")
		})},
	}
	response := svc.executeQueryWithValues(
		context.Background(),
		"https://metrics.internal.example.com/api/v1/query",
		map[string][]string{"query": {secretMarker}},
		nil,
	)
	require.Equal(t, "error", response.Status)

	entries := observed.FilterMessage("HTTP request failed").All()
	require.Len(t, entries, 1)
	encodedFields, err := json.Marshal(entries[0].ContextMap())
	require.NoError(t, err)
	assert.NotContains(t, string(encodedFields), secretMarker)
	assert.NotContains(t, string(encodedFields), "metrics.internal.example.com")
}

func newAzureMetricsQueryTestService(t *testing.T, cluster clinic_internal_api.ClusterDataURLs, upstream http.HandlerFunc) *MetricsQueryService {
	t.Helper()

	upstreamServer := httptest.NewServer(upstream)
	t.Cleanup(upstreamServer.Close)

	cluster.ClusterMetricsURL = upstreamServer.URL + "/api/v1/"
	if cluster.DataPlaneMetricsURL == "" {
		cluster.DataPlaneMetricsURL = upstreamServer.URL + "/api/v1/"
	}

	clinicServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, "/clinic/api/v1/internal/clusters/"+cluster.ClusterID+"/cluster-dataurls", r.URL.Path)
		require.Equal(t, http.MethodGet, r.Method)
		w.Header().Set("Content-Type", "application/json")
		require.NoError(t, json.NewEncoder(w).Encode(cluster))
	}))
	t.Cleanup(clinicServer.Close)

	return &MetricsQueryService{
		clinicClient: clinic_internal_api.NewClient(clinicServer.URL),
		httpClient:   upstreamServer.Client(),
		dataPlaneMetrics: map[string]bool{
			"node_memory_MemTotal_bytes": true,
		},
	}
}

func assertEmptyPrometheusQueryResult(t *testing.T, data any, resultType string) {
	t.Helper()

	result, ok := data.(map[string]any)
	require.True(t, ok, "expected Prometheus query data to be an object")
	assert.Equal(t, resultType, result["resultType"])
	assert.Equal(t, []any{}, result["result"])
}

func writePrometheusSuccess(t *testing.T, w http.ResponseWriter, data any) {
	t.Helper()

	w.Header().Set("Content-Type", "application/json")
	require.NoError(t, json.NewEncoder(w).Encode(dto.PrometheusResponse{
		Status: "success",
		Data:   data,
	}))
}