package service

import (
	"context"
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"net/url"
	"strings"
	"testing"
	"time"

	pingcaplog "github.com/pingcap/log"
	"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"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/config"
)

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

func TestInjectLabelIntoLogQLSelector(t *testing.T) {
	tests := []struct {
		name      string
		logql     string
		labelName string
		labelVal  string
		want      string
	}{
		{
			name:      "empty label name returns unchanged",
			logql:     `{job="tidb"}`,
			labelName: "",
			labelVal:  "ns1",
			want:      `{job="tidb"}`,
		},
		{
			name:      "empty label value returns unchanged",
			logql:     `{job="tidb"}`,
			labelName: "namespace",
			labelVal:  "",
			want:      `{job="tidb"}`,
		},
		{
			name:      "no selector returns unchanged",
			logql:     `sum by (job) (rate(x[5m]))`,
			labelName: "namespace",
			labelVal:  "tidb123",
			want:      `sum by (job) (rate(x[5m]))`,
		},
		{
			name:      "empty selector injects label",
			logql:     `{}`,
			labelName: "namespace",
			labelVal:  "tidb123",
			want:      `{namespace="tidb123"}`,
		},
		{
			name:      "single label selector appends label",
			logql:     `{job="tidb"}`,
			labelName: "namespace",
			labelVal:  "tidb1379661944646416027",
			want:      `{job="tidb", namespace="tidb1379661944646416027"}`,
		},
		{
			name:      "multiple labels selector appends label",
			logql:     `{job="tidb", instance=~"db-tidb-.*"}`,
			labelName: "namespace",
			labelVal:  "tidb123",
			want:      `{job="tidb", instance=~"db-tidb-.*", namespace="tidb123"}`,
		},
		{
			name:      "selector with pipeline injects into first selector only",
			logql:     `{job="tidb"} | json | level="error"`,
			labelName: "namespace",
			labelVal:  "tidb456",
			want:      `{job="tidb", namespace="tidb456"} | json | level="error"`,
		},
		{
			name:      "project_id label",
			logql:     `{namespace="tidb-admin"}`,
			labelName: "project_id",
			labelVal:  "1372813089454544954",
			want:      `{namespace="tidb-admin", project_id="1372813089454544954"}`,
		},
		{
			name:      "existing label is replaced",
			logql:     `{cluster="wrong", container="clinic"}`,
			labelName: "cluster",
			labelVal:  "prod-us-east-1-f04",
			want:      `{container="clinic", cluster="prod-us-east-1-f04"}`,
		},
		{
			name:      "existing regex label is replaced",
			logql:     `{cluster=~"prod-.*", container="clinic"}`,
			labelName: "cluster",
			labelVal:  "prod-us-east-1-f04",
			want:      `{container="clinic", cluster="prod-us-east-1-f04"}`,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got := injectLabelIntoLogQLSelector(tt.logql, tt.labelName, tt.labelVal)
			assert.Equal(t, tt.want, got, "injectLabelIntoLogQLSelector(%q, %q, %q)", tt.logql, tt.labelName, tt.labelVal)
		})
	}
}

func TestInjectLogQLPipelineFilter(t *testing.T) {
	tests := []struct {
		name  string
		logql string
		value string
		want  string
	}{
		{
			name:  "empty value returns unchanged",
			logql: `{job="tidb"}`,
			value: "",
			want:  `{job="tidb"}`,
		},
		{
			name:  "no selector appends filter",
			logql: `sum(rate(x[5m]))`,
			value: "ks-abc",
			want:  `sum(rate(x[5m])) |= "ks-abc"`,
		},
		{
			name:  "single selector inserts after first }",
			logql: `{job="tidb"}`,
			value: "keyspace-1",
			want:  `{job="tidb"} |= "keyspace-1"`,
		},
		{
			name:  "selector with pipeline inserts after selector",
			logql: `{job="tidb"} | json`,
			value: "ks-xyz",
			want:  `{job="tidb"} |= "ks-xyz" | json`,
		},
		{
			name:  "selector with existing pipeline filter",
			logql: `{job="tidb"} |= "some" | level="error"`,
			value: "my-keyspace",
			want:  `{job="tidb"} |= "my-keyspace" |= "some" | level="error"`,
		},
		{
			name:  "empty selector",
			logql: `{}`,
			value: "token-value",
			want:  `{} |= "token-value"`,
		},
		{
			name:  "multiple labels in selector",
			logql: `{job="tidb", instance="db-tidb-0"}`,
			value: "o11y-keyspace",
			want:  `{job="tidb", instance="db-tidb-0"} |= "o11y-keyspace"`,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got := injectLogQLPipelineFilter(tt.logql, tt.value)
			assert.Equal(t, tt.want, got, "injectLogQLPipelineFilter(%q, %q)", tt.logql, tt.value)
		})
	}
}

func TestResolveSystemLogsURL(t *testing.T) {
	cfg := config.DefaultConfig()
	cfg.LokiProxy.DefaultSystemDatasource = "Loki-aws-us-west-2"
	cfg.LokiProxy.SystemEndpoints = []config.SystemLokiEndpointConfig{
		{
			Datasource: "Loki-aws-us-west-2",
			Provider:   "aws",
			Region:     "us-west-2",
			URL:        "https://loki-us-west-2.example.com/loki/self-monitoring",
		},
		{
			UID:      "Loki-gcp-us-central1",
			Provider: "gcp",
			Region:   "us-central1",
			URL:      "https://loki-us-central1.example.com/loki/self-monitoring",
		},
	}
	svc := NewLokiProxyService(cfg)

	datasources := svc.ListSystemLogDatasources()
	require.Len(t, datasources.Datasources, 2)
	assert.Equal(t, "Loki-aws-us-west-2", datasources.DefaultDatasource)
	assert.Equal(t, "Loki-aws-us-west-2", datasources.Datasources[0].Datasource)
	assert.True(t, datasources.Datasources[0].IsDefault)
	assert.Equal(t, "Loki-gcp-us-central1", datasources.Datasources[1].Datasource)
	assert.False(t, datasources.Datasources[1].IsDefault)

	got, err := svc.ResolveSystemLogsURL("Loki-aws-us-west-2")
	require.NoError(t, err)
	assert.Equal(t, "https://loki-us-west-2.example.com/loki/self-monitoring", got)

	got, err = svc.ResolveSystemLogsURL("Loki-gcp-us-central1")
	require.NoError(t, err)
	assert.Equal(t, "https://loki-us-central1.example.com/loki/self-monitoring", got)

	_, err = svc.ResolveSystemLogsURL("")
	require.ErrorContains(t, err, "datasource is required")

	_, err = svc.ResolveSystemLogsURL("Loki-aws-eu-central-1")
	require.ErrorContains(t, err, "system log datasource not found")

	got, err = svc.ResolveSystemLogsURLByProviderRegion("aws", "us-west-2")
	require.NoError(t, err)
	assert.Equal(t, "https://loki-us-west-2.example.com/loki/self-monitoring", got)

	got, err = svc.ResolveServerlessParentLogsURL("", ServerlessParentControlPlaneInfo{
		Vendor: "gcp",
		Region: "us-central1",
	})
	require.NoError(t, err)
	assert.Equal(t, "https://loki-us-central1.example.com/loki/self-monitoring", got)
}

func TestBuildServerlessParentLogsProxyURL(t *testing.T) {
	cfg := config.DefaultConfig()
	cfg.LokiProxy.Env = "prod"
	svc := NewLokiProxyService(cfg)
	info := ServerlessParentControlPlaneInfo{
		ProjectID:     "us-east-1-f04",
		Vendor:        "aws",
		Region:        "us-east-1",
		ApplicationID: "devtier-infra",
	}

	originalURL, err := url.Parse(`/data-proxy/serverless-parent-logs/api/v1/query_range?query=%7B%7D+%21%3D+%60operator_controller.go%60&start=1&end=2`)
	require.NoError(t, err)
	got, err := svc.BuildServerlessParentLogsProxyURL("/loki/api/v1/query_range", originalURL, info)
	require.NoError(t, err)
	assert.Equal(t, `{cluster="prod-us-east-1-f04"} != `+"`operator_controller.go`", got.Query().Get("query"))

	got, err = svc.BuildServerlessParentLogsProxyURL("/loki/api/v1/query_range", mustParseURL(t, `/data-proxy/serverless-parent-logs/api/v1/query_range?query=%7Bcluster%3D%22prod-us-east-1-f04%22%7D+%21%3D+%60operator_controller.go%60&start=1&end=2`), info)
	require.NoError(t, err)
	assert.Equal(t, `{cluster="prod-us-east-1-f04"} != `+"`operator_controller.go`", got.Query().Get("query"))

	_, err = svc.BuildServerlessParentLogsProxyURL("/loki/api/v1/query_range", mustParseURL(t, `/data-proxy/serverless-parent-logs/api/v1/query_range?query=sum%28rate%28x%5B5m%5D%29%29`), info)
	require.ErrorContains(t, err, "stream selector")

	got, err = svc.BuildServerlessParentLogsProxyURL("/loki/api/v1/series", mustParseURL(t, `/data-proxy/serverless-parent-logs/api/v1/series`), info)
	require.NoError(t, err)
	assert.Equal(t, `{cluster="prod-us-east-1-f04"}`, got.Query().Get("match[]"))

	got, err = svc.BuildServerlessParentLogsProxyURL("/loki/api/v1/labels", mustParseURL(t, `/data-proxy/serverless-parent-logs/api/v1/labels?match%5B%5D=%7Bcontainer%3D%22clinic%22%7D`), info)
	require.NoError(t, err)
	assert.Equal(t, `{container="clinic", cluster="prod-us-east-1-f04"}`, got.Query().Get("match[]"))

	got, err = svc.BuildServerlessParentLogsProxyURL("/loki/api/v1/series", mustParseURL(t, `/data-proxy/serverless-parent-logs/api/v1/series?match%5B%5D=%7Bcluster%3D%22wrong%22%2Ccontainer%3D%22clinic%22%7D`), info)
	require.NoError(t, err)
	assert.Equal(t, `{container="clinic", cluster="prod-us-east-1-f04"}`, got.Query().Get("match[]"))
}

func mustParseURL(t *testing.T, rawURL string) *url.URL {
	t.Helper()
	u, err := url.Parse(rawURL)
	require.NoError(t, err)
	return u
}

func TestProxySystemRequest(t *testing.T) {
	var gotPath string
	var gotQuery url.Values
	var gotHeader string
	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		gotPath = r.URL.Path
		gotQuery = r.URL.Query()
		gotHeader = r.Header.Get("X-Test-Header")
		w.Header().Set("Content-Type", "application/json")
		require.NoError(t, json.NewEncoder(w).Encode(map[string]any{
			"status": "success",
			"data": map[string]any{
				"resultType": "streams",
				"result":     []any{},
			},
		}))
	}))
	t.Cleanup(upstream.Close)

	cfg := config.DefaultConfig()
	cfg.LokiProxy.MaxLimit = 100
	svc := NewLokiProxyService(cfg)
	originalURL, err := url.Parse("/data-proxy/system-logs/api/v1/query_range?query=%7Bnamespace%3D%22clinic-prod-cloud%22%7D&limit=1000&start=1&end=2")
	require.NoError(t, err)
	headers := http.Header{}
	headers.Set("X-Test-Header", "keep-me")

	resp, err := svc.ProxySystemRequest(
		context.Background(),
		"/loki/api/v1/query_range",
		originalURL,
		http.MethodGet,
		nil,
		headers,
		upstream.URL+"/loki/self-monitoring",
	)
	require.NoError(t, err)
	defer resp.Body.Close()

	assert.Equal(t, http.StatusOK, resp.StatusCode)
	assert.Equal(t, "/loki/self-monitoring/loki/api/v1/query_range", gotPath)
	assert.Equal(t, `{namespace="clinic-prod-cloud"}`, gotQuery.Get("query"))
	assert.Equal(t, "100", gotQuery.Get("limit"))
	assert.Equal(t, "1", gotQuery.Get("start"))
	assert.Equal(t, "2", gotQuery.Get("end"))
	assert.Equal(t, "keep-me", gotHeader)
}

func TestProxySystemRequestDoesNotLogRawQuery(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)

	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.WriteHeader(http.StatusOK)
	}))
	t.Cleanup(upstream.Close)

	svc := NewLokiProxyService(config.DefaultConfig())
	const secretMarker = "supersecretlogqlmarker"
	originalURL, err := url.Parse("/data-proxy/system-logs/api/v1/query_range?query=" + secretMarker)
	require.NoError(t, err)

	resp, err := svc.ProxySystemRequest(
		context.Background(),
		"/loki/api/v1/query_range",
		originalURL,
		http.MethodGet,
		nil,
		nil,
		upstream.URL+"/loki/self-monitoring",
	)
	require.NoError(t, err)
	resp.Body.Close()

	entries := observed.FilterMessage("proxying to system logs").All()
	require.Len(t, entries, 1)
	fields := entries[0].ContextMap()
	assert.Equal(t, http.MethodGet, fields["method"])
	assert.Equal(t, mustParseURL(t, upstream.URL).Host, fields["upstream_host"])
	assert.Equal(t, "/loki/self-monitoring/loki/api/v1/query_range", fields["upstream_path"])
	_, hasRawURL := fields["url"]
	assert.False(t, hasRawURL)
	encodedFields, err := json.Marshal(fields)
	require.NoError(t, err)
	assert.NotContains(t, string(encodedFields), secretMarker)
}

func TestIsBYOCLogsClusterOnlyMatchesBYOC(t *testing.T) {
	tests := []struct {
		name         string
		deployType   string
		deployTypeV2 string
		want         bool
	}{
		{name: "dedicated", deployType: "dedicated", deployTypeV2: "dedicated"},
		{name: "premium", deployType: "premium", deployTypeV2: "premium"},
		{name: "starter", deployType: "shared", deployTypeV2: "starter"},
		{name: "essential", deployType: "shared", deployTypeV2: "essential"},
		{name: "BYOC logical", deployType: "byoc", deployTypeV2: "byoc", want: true},
		{name: "BYOC resource pool", deployType: "byoc-host", deployTypeV2: "byoc resource pool", want: true},
		{name: "inconsistent BYOC marker fails closed in BYOC path", deployType: "premium", deployTypeV2: "byoc", want: true},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			assert.Equal(t, tt.want, IsBYOCLogsCluster(tt.deployType, tt.deployTypeV2))
		})
	}
}

func TestProxyBYOCRequestLogicalUsesO11YRoutingAndPhysicalNamespace(t *testing.T) {
	var gotPath string
	var gotQuery url.Values
	var gotHeaders http.Header
	var gotBody string
	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		gotPath = r.URL.Path
		gotQuery = r.URL.Query()
		gotHeaders = r.Header.Clone()
		body, _ := io.ReadAll(r.Body)
		gotBody = string(body)
		w.Header().Set("Content-Type", "application/json")
		w.Header().Set("Set-Cookie", "upstream=secret")
		w.Header().Set("X-Internal-Token", "secret")
		w.Header().Set("X-Request-Id", "o11y-request")
		_, _ = io.WriteString(w, `{"status":"success"}`)
	}))
	t.Cleanup(upstream.Close)

	cfg := config.DefaultConfig()
	cfg.LokiProxy.Env = "dev"
	cfg.LokiProxy.MaxLimit = 100
	svc := NewLokiProxyService(cfg)
	query := url.Values{}
	query.Set("query", `{cluster_env="wrong",namespace="wrong",component="tidb"}`)
	query.Set("limit", "1000")
	for _, key := range []string{
		"cluster_id", "x-clinic-cluster-id", "var-cluster_id", "var-tidb_cluster",
		"tidb_cluster", "control_plane_info", "var-control_plane_info", "var-project_id",
	} {
		query.Set(key, "caller-controlled")
	}
	originalURL := mustParseURL(t, "/data-proxy/loki/logical/api/v1/query_range?"+query.Encode())
	headers := http.Header{
		"Accept":        []string{"application/json"},
		"Authorization": []string{"Bearer secret"},
		"X-Biz-Type":    []string{"caller-controlled"},
		"X-Org-Id":      []string{"999"},
		"X-Region-Id":   []string{"wrong-region"},
	}
	info := &LokiClusterInfo{
		ClusterID:    "10954487466006069299",
		ParentID:     "2034353583595458560",
		TenantID:     "100",
		Provider:     "aws",
		Region:       "us-west-2",
		DeployType:   "byoc",
		DeployTypeV2: "byoc",
		LokiURL:      upstream.URL + "/internal/loki/self-monitoring",
	}

	resp, err := svc.ProxyBYOCRequest(context.Background(), "/loki/api/v1/query_range", originalURL, http.MethodGet, strings.NewReader("query=caller-body"), headers, info)
	require.NoError(t, err)
	defer resp.Body.Close()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	assert.Equal(t, http.StatusOK, resp.StatusCode)
	assert.JSONEq(t, `{"status":"success"}`, string(body))
	assert.Equal(t, "/internal/loki/self-monitoring/loki/api/v1/query_range", gotPath)
	assert.Equal(t, `{component="tidb", cluster_env="dev", namespace="tidb2034353583595458560"}`, gotQuery.Get("query"))
	assert.Equal(t, "100", gotQuery.Get("limit"))
	for _, key := range []string{
		"cluster_id", "x-clinic-cluster-id", "var-cluster_id", "var-tidb_cluster",
		"tidb_cluster", "control_plane_info", "var-control_plane_info", "var-project_id",
	} {
		assert.Empty(t, gotQuery.Get(key), key)
	}
	assert.Equal(t, "byoc", gotHeaders.Get("X-Biz-Type"))
	assert.Equal(t, "100", gotHeaders.Get("X-Org-Id"))
	assert.Equal(t, "aws-us-west-2", gotHeaders.Get("X-Region-Id"))
	assert.Equal(t, "application/json", gotHeaders.Get("Accept"))
	assert.Empty(t, gotHeaders.Get("Authorization"))
	assert.Empty(t, gotBody)
	assert.Equal(t, "application/json", resp.Header.Get("Content-Type"))
	assert.Equal(t, "o11y-request", resp.Header.Get("X-Request-Id"))
	assert.Empty(t, resp.Header.Get("Set-Cookie"))
	assert.Empty(t, resp.Header.Get("X-Internal-Token"))
}

func TestProxyBYOCRequestResourcePoolUsesClusterNamespaceAndLabelsQuery(t *testing.T) {
	var gotQuery url.Values
	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		gotQuery = r.URL.Query()
		w.Header().Set("Content-Type", "application/json")
		_, _ = io.WriteString(w, `{"status":"success"}`)
	}))
	t.Cleanup(upstream.Close)

	cfg := config.DefaultConfig()
	cfg.LokiProxy.Env = "dev"
	svc := NewLokiProxyService(cfg)
	query := url.Values{}
	query.Set("match[]", `{component="tikv"}`)
	info := &LokiClusterInfo{
		ClusterID:    "2034353583595458560",
		TenantID:     "100",
		Provider:     "aws",
		Region:       "us-west-2",
		DeployType:   "byoc-host",
		DeployTypeV2: "byoc resource pool",
		LokiURL:      upstream.URL + "/internal/loki/self-monitoring",
	}

	resp, err := svc.ProxyBYOCRequest(
		context.Background(),
		"/loki/api/v1/labels",
		mustParseURL(t, "/data-proxy/loki/2034353583595458560/api/v1/labels?"+query.Encode()),
		http.MethodGet,
		nil,
		nil,
		info,
	)
	require.NoError(t, err)
	resp.Body.Close()

	assert.Equal(t, `{component="tikv", cluster_env="dev", namespace="tidb2034353583595458560"}`, gotQuery.Get("query"))
	assert.Empty(t, gotQuery["match[]"])
}

func TestProxyBYOCRequestFailsClosed(t *testing.T) {
	cfg := config.DefaultConfig()
	cfg.LokiProxy.Env = "dev"
	svc := NewLokiProxyService(cfg)
	baseInfo := &LokiClusterInfo{
		ClusterID:    "logical",
		ParentID:     "physical",
		TenantID:     "100",
		Provider:     "aws",
		Region:       "us-west-2",
		DeployType:   "byoc",
		DeployTypeV2: "byoc",
		LokiURL:      "https://logs.example.com/internal/loki/self-monitoring",
	}

	_, err := svc.ProxyBYOCRequest(
		context.Background(),
		"/loki/api/v1/query_range",
		mustParseURL(t, `/logs?query=%7Bcomponent%3D%22tidb%22%7D+or+%7Bcomponent%3D%22tikv%22%7D`),
		http.MethodGet,
		nil,
		nil,
		baseInfo,
	)
	require.Error(t, err)
	assert.True(t, errors.Is(err, ErrInvalidBYOCLogQuery))

	inconsistent := *baseInfo
	inconsistent.DeployType = "premium"
	_, err = svc.ProxyBYOCRequest(context.Background(), "/loki/api/v1/series", mustParseURL(t, "/logs"), http.MethodGet, nil, nil, &inconsistent)
	require.ErrorContains(t, err, "inconsistent BYOC")

	_, err = svc.ProxyBYOCRequest(context.Background(), "/loki/api/v1/label/../values", mustParseURL(t, "/logs"), http.MethodGet, nil, nil, baseInfo)
	require.Error(t, err)
	assert.True(t, errors.Is(err, ErrInvalidBYOCLogQuery))

	prodCfg := config.DefaultConfig()
	prodCfg.LokiProxy.Env = "prod"
	prodSvc := NewLokiProxyService(prodCfg)
	insecure := *baseInfo
	insecure.LokiURL = "http://logs.example.com/internal/loki/self-monitoring"
	_, err = prodSvc.ProxyBYOCRequest(context.Background(), "/loki/api/v1/series", mustParseURL(t, "/logs"), http.MethodGet, nil, nil, &insecure)
	require.ErrorContains(t, err, "must use HTTPS")
}

func TestProxyBYOCRequestDoesNotFollowRedirects(t *testing.T) {
	targetHits := 0
	target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		targetHits++
		w.WriteHeader(http.StatusOK)
	}))
	t.Cleanup(target.Close)
	redirect := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		http.Redirect(w, &http.Request{}, target.URL, http.StatusFound)
	}))
	t.Cleanup(redirect.Close)

	cfg := config.DefaultConfig()
	cfg.LokiProxy.Env = "dev"
	svc := NewLokiProxyService(cfg)
	info := &LokiClusterInfo{
		ClusterID:    "logical",
		ParentID:     "physical",
		TenantID:     "100",
		Provider:     "aws",
		Region:       "us-west-2",
		DeployType:   "byoc",
		DeployTypeV2: "byoc",
		LokiURL:      redirect.URL,
	}
	resp, err := svc.ProxyBYOCRequest(context.Background(), "/loki/api/v1/query_range", mustParseURL(t, "/logs?query=%7B%7D"), http.MethodGet, nil, nil, info)
	require.NoError(t, err)
	resp.Body.Close()
	assert.Equal(t, http.StatusFound, resp.StatusCode)
	assert.Zero(t, targetHits)
}

func TestProxyRequestNonBYOCKeepsEstablishedHeadersAndRedirects(t *testing.T) {
	finalHits := 0
	var gotAuthorization string
	var gotBizType string
	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/final" {
			http.Redirect(w, r, "/final", http.StatusFound)
			return
		}
		finalHits++
		gotAuthorization = r.Header.Get("Authorization")
		gotBizType = r.Header.Get("X-Biz-Type")
		w.WriteHeader(http.StatusOK)
	}))
	t.Cleanup(upstream.Close)

	svc := NewLokiProxyService(config.DefaultConfig())
	info := &LokiClusterInfo{
		ClusterID:    "dedicated-cluster",
		DeployType:   "dedicated",
		DeployTypeV2: "dedicated",
		LokiURL:      upstream.URL + "/loki/self-monitoring",
	}
	headers := http.Header{
		"Authorization": []string{"Bearer established-behavior"},
		"X-Biz-Type":    []string{"caller-value"},
	}
	resp, err := svc.ProxyRequest(context.Background(), "/loki/api/v1/query_range", mustParseURL(t, "/logs?query=%7B%7D"), http.MethodGet, nil, headers, info)
	require.NoError(t, err)
	resp.Body.Close()

	assert.Equal(t, http.StatusOK, resp.StatusCode)
	assert.Equal(t, 1, finalHits)
	assert.Equal(t, "Bearer established-behavior", gotAuthorization)
	assert.Equal(t, "caller-value", gotBizType)
}