package cmd

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"testing"

	"github.com/tidbcloud/clinic-cli/internal/clinic"
	"github.com/tidbcloud/clinic-cli/internal/config"
)

const (
	testRegionalClusterID = "cluster-42"
	testRegionalProvider  = "alicloud"
	testRegionalRegion    = "ap-northeast-1"
	testRegionalUID       = "Loki-alicloud-ap-northeast-1"
)

func TestLokiQueryStarterEssentialRegionalRouting(t *testing.T) {
	t.Parallel()

	testCases := []struct {
		name            string
		profileTier     string
		refreshedTier   string
		provider        string
		region          string
		catalogProvider string
		catalogRegion   string
		output          string
		wantOutput      string
	}{
		{
			name:            "normalized starter uses case-insensitive datasource match and JSON output",
			profileTier:     " Starter ",
			refreshedTier:   " STARTER ",
			provider:        "ALICLOUD",
			region:          "AP-NORTHEAST-1",
			catalogProvider: "alicloud",
			catalogRegion:   "ap-northeast-1",
			output:          "json",
			wantOutput:      `{"status":"success","data":{"resultType":"streams","result":[]}}`,
		},
		{
			name:            "normalized essential uses regional log output",
			profileTier:     "essential",
			refreshedTier:   " Essential ",
			provider:        "alicloud",
			region:          "ap-northeast-1",
			catalogProvider: "ALICLOUD",
			catalogRegion:   "AP-NORTHEAST-1",
			wantOutput:      "[instance=dm-worker-0] routed line\n",
		},
	}

	for _, tc := range testCases {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			queryBody := []byte(`{"status":"success","data":{"resultType":"streams","result":[]}}`)
			if tc.output == "" {
				queryBody = []byte(`{"status":"success","data":{"resultType":"streams","result":[{"stream":{"instance":"dm-worker-0"},"values":[["1712123600000000000","routed line"]]}]}}`)
			}
			client := newRegionalLokiTestClient(
				dashboardClusterBody(testRegionalClusterID, tc.refreshedTier, tc.provider, tc.region),
				datasourceCatalogBody(testRegionalUID, tc.catalogProvider, tc.catalogRegion),
			)
			client.regionalResponses = []clinicResponse{{body: queryBody}}
			ctx := regionalLokiContext(client, tc.profileTier)

			args := regionalLokiArgs(`{cluster="physical-pool", component="dm-worker"} |= "routed"`)
			if tc.output != "" {
				args = append(args, "--output", tc.output)
			}
			var stdout strings.Builder
			if err := lokiCommandHandler(ctx, args, &stdout); err != nil {
				t.Fatalf("lokiCommandHandler() error = %v", err)
			}

			assertRegionalRequestSequence(t, client.requests, testRegionalUID, 1)
			queryRequest := client.requests[2]
			assertQueryValue(t, queryRequest.Query, "query", `{cluster="physical-pool", component="dm-worker", cluster_env="staging", namespace="tidb-serverless"} |= "routed"`)
			if tc.output == "json" {
				assertJSONEqual(t, stdout.String(), tc.wantOutput)
			} else if got := stdout.String(); got != tc.wantOutput {
				t.Fatalf("stdout = %q, want %q", got, tc.wantOutput)
			}
		})
	}
}

func TestLokiQueryRegionalUIDUsesDecodedPathAndEscapedRawPath(t *testing.T) {
	t.Parallel()

	const uid = "loki/team?prod#one%"
	client := newRegionalLokiTestClient(
		dashboardClusterBody(testRegionalClusterID, "starter", testRegionalProvider, testRegionalRegion),
		datasourceCatalogBody(uid, testRegionalProvider, testRegionalRegion),
	)
	client.regionalResponses = []clinicResponse{{body: emptyLokiQueryResponse()}}

	if err := lokiCommandHandler(regionalLokiContext(client, "starter"), regionalLokiArgs(`{cluster="physical-pool"}`), &strings.Builder{}); err != nil {
		t.Fatalf("lokiCommandHandler() error = %v", err)
	}

	request := requireRegionalQueryRequest(t, client.requests)
	if got, want := request.Path, "/data-proxy/api/v1/system-log-datasources/loki/team?prod#one%/logs/query-range"; got != want {
		t.Fatalf("Path = %q, want decoded %q", got, want)
	}
	if got, want := request.RawPath, "/data-proxy/api/v1/system-log-datasources/loki%2Fteam%3Fprod%23one%25/logs/query-range"; got != want {
		t.Fatalf("RawPath = %q, want %q", got, want)
	}
}

func TestLokiQueryStarterProfileRefreshMissingDeployTypeFailsClosed(t *testing.T) {
	t.Parallel()

	client := newRegionalLokiTestClient(
		dashboardClusterBody(testRegionalClusterID, "", testRegionalProvider, testRegionalRegion),
		datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion),
	)
	err := lokiCommandHandler(
		regionalLokiContext(client, "starter"),
		regionalLokiArgs(`{cluster="physical-pool"}`),
		&strings.Builder{},
	)
	if err == nil {
		t.Fatal("lokiCommandHandler() error = nil, want missing refreshed deploy-type error")
	}
	assertErrorContains(t, err, testRegionalClusterID, "missing", "deploy", "type")
	if got, want := len(client.requests), 1; got != want {
		t.Fatalf("request count = %d, want only Dashboard metadata refresh (%d); paths=%v", got, want, requestPaths(client.requests))
	}
	if got := client.requests[0].Path; got != dashboardClustersPath {
		t.Fatalf("request path = %q, want metadata refresh", got)
	}
	assertNoLokiQueryRangeRequest(t, client.requests)
}

func TestLokiQueryRegionalSelectorValidation(t *testing.T) {
	t.Parallel()

	testCases := []struct {
		name  string
		query *string
	}{
		{name: "omitted LogQL"},
		{name: "empty selector", query: stringPointer(`{}`)},
		{name: "missing physical cluster matcher", query: stringPointer(`{component="dm-worker"}`)},
		{name: "empty cluster equality", query: stringPointer(`{cluster=""}`)},
		{name: "unquoted cluster equality", query: stringPointer(`{cluster=physical-pool}`)},
		{name: "cluster regex", query: stringPointer(`{cluster=~"physical-.*"}`)},
		{name: "negative cluster matcher", query: stringPointer(`{cluster!="other"}`)},
		{name: "negative regex cluster matcher", query: stringPointer(`{cluster!~"other.*"}`)},
		{name: "multiple selectors", query: stringPointer(`{cluster="physical-pool"} or {cluster="other-pool"}`)},
	}

	for _, tc := range testCases {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			client := newRegionalLokiTestClient(
				dashboardClusterBody(testRegionalClusterID, "starter", testRegionalProvider, testRegionalRegion),
				datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion),
			)
			args := regionalLokiArgs("")
			if tc.query != nil {
				args = regionalLokiArgs(*tc.query)
			}
			err := lokiCommandHandler(regionalLokiContext(client, "starter"), args, &strings.Builder{})
			if err == nil {
				t.Fatal("lokiCommandHandler() error = nil, want selector validation error")
			}
			assertErrorContains(t, err, testRegionalClusterID, "physical", "cluster")
			if got, want := len(client.requests), 1; got != want {
				t.Fatalf("request count = %d, want only metadata refresh (%d); paths=%v", got, want, requestPaths(client.requests))
			}
			if got := client.requests[0].Path; got != dashboardClustersPath {
				t.Fatalf("request path = %q, want metadata refresh", got)
			}
		})
	}
}

func TestLokiQueryRegionalClusterMatcherValuesContainingOperatorText(t *testing.T) {
	t.Parallel()

	for _, clusterValue := range []string{
		"physical=~pool",
		"physical!=pool",
		"physical!~pool",
		"physical=pool",
	} {
		clusterValue := clusterValue
		t.Run(clusterValue, func(t *testing.T) {
			t.Parallel()

			client := newRegionalLokiTestClient(
				dashboardClusterBody(testRegionalClusterID, "starter", testRegionalProvider, testRegionalRegion),
				datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion),
			)
			client.regionalResponses = []clinicResponse{{body: emptyLokiQueryResponse()}}
			query := fmt.Sprintf(`{cluster=%q, component="dm-worker"}`, clusterValue)

			if err := lokiCommandHandler(regionalLokiContext(client, "starter"), regionalLokiArgs(query), &strings.Builder{}); err != nil {
				t.Fatalf("lokiCommandHandler() error = %v", err)
			}

			assertRegionalRequestSequence(t, client.requests, testRegionalUID, 1)
			wantQuery := fmt.Sprintf(`{cluster=%q, component="dm-worker", cluster_env="staging", namespace="tidb-serverless"}`, clusterValue)
			assertQueryValue(t, requireRegionalQueryRequest(t, client.requests).Query, "query", wantQuery)
		})
	}
}

func TestLokiQueryRegionalBacktickRawStringsDoNotCountAsSelectors(t *testing.T) {
	t.Parallel()

	testCases := []struct {
		name   string
		query  string
		suffix string
	}{
		{
			name:   "line format template",
			query:  "{cluster=\"physical-pool\", component=\"dm-worker\"} | json | line_format `{{.message}}`",
			suffix: " | json | line_format `{{.message}}`",
		},
		{
			name:   "label format template",
			query:  "{cluster=\"physical-pool\", component=\"dm-worker\"} | json | label_format source=`{{.pod}}:{region={{.region}}}`",
			suffix: " | json | label_format source=`{{.pod}}:{region={{.region}}}`",
		},
		{
			name:   "raw filter containing braces",
			query:  "{cluster=\"physical-pool\", component=\"dm-worker\"} |= `{left}{right}`",
			suffix: " |= `{left}{right}`",
		},
	}

	for _, tc := range testCases {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			client := newRegionalLokiTestClient(
				dashboardClusterBody(testRegionalClusterID, "starter", testRegionalProvider, testRegionalRegion),
				datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion),
			)
			client.regionalResponses = []clinicResponse{{body: emptyLokiQueryResponse()}}

			if err := lokiCommandHandler(regionalLokiContext(client, "starter"), regionalLokiArgs(tc.query), &strings.Builder{}); err != nil {
				t.Fatalf("lokiCommandHandler() error = %v", err)
			}

			assertRegionalRequestSequence(t, client.requests, testRegionalUID, 1)
			gotQuery := requireRegionalQueryRequest(t, client.requests).Query.Get("query")
			wantQuery := `{cluster="physical-pool", component="dm-worker", cluster_env="staging", namespace="tidb-serverless"}` + tc.suffix
			if gotQuery != wantQuery {
				t.Fatalf("regional LogQL = %q, want one injected selector and unchanged raw suffix %q", gotQuery, wantQuery)
			}
			if !strings.HasSuffix(gotQuery, tc.suffix) {
				t.Fatalf("regional LogQL suffix = %q, want byte-for-byte suffix %q", gotQuery, tc.suffix)
			}
		})
	}
}

func TestLokiQueryRegionalBacktickUnterminatedRawStringFailsClosed(t *testing.T) {
	t.Parallel()

	client := newRegionalLokiTestClient(
		dashboardClusterBody(testRegionalClusterID, "essential", testRegionalProvider, testRegionalRegion),
		datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion),
	)
	query := "{cluster=\"physical-pool\"} | json | line_format `{{.message}}"

	err := lokiCommandHandler(regionalLokiContext(client, "essential"), regionalLokiArgs(query), &strings.Builder{})
	if err == nil {
		t.Fatal("lokiCommandHandler() error = nil, want unterminated raw-string validation error")
	}
	assertErrorContains(t, err, testRegionalClusterID, "unterminated", "raw string")
	if got, want := len(client.requests), 1; got != want {
		t.Fatalf("request count = %d, want only metadata refresh (%d); paths=%v", got, want, requestPaths(client.requests))
	}
	if got := client.requests[0].Path; got != dashboardClustersPath {
		t.Fatalf("request path = %q, want metadata refresh", got)
	}
}

func TestLokiQueryRegionalCommentsDoNotAffectSelectorLexing(t *testing.T) {
	t.Parallel()

	testCases := []struct {
		name      string
		query     string
		wantQuery string
	}{
		{
			name:      "leading and trailing selector-shaped comments",
			query:     "# leading {cluster=\"commented\", cluster_env=\"commented\", namespace=\"commented\"} \"unterminated ` ignored\n{cluster=\"physical-pool\", component=\"dm-worker\"} # trailing {cluster=\"commented\"} \"unterminated ` ignored",
			wantQuery: "# leading {cluster=\"commented\", cluster_env=\"commented\", namespace=\"commented\"} \"unterminated ` ignored\n{cluster=\"physical-pool\", component=\"dm-worker\", cluster_env=\"staging\", namespace=\"tidb-serverless\"} # trailing {cluster=\"commented\"} \"unterminated ` ignored",
		},
		{
			name:      "comments and raw strings remain byte-for-byte unchanged",
			query:     "{cluster=\"physical-pool\" # preserve {cluster=\"commented\"}\n} | line_format `{{.message}} # {raw}` # tail {cluster=\"commented\"}",
			wantQuery: "{cluster=\"physical-pool\", cluster_env=\"staging\", namespace=\"tidb-serverless\" # preserve {cluster=\"commented\"}\n} | line_format `{{.message}} # {raw}` # tail {cluster=\"commented\"}",
		},
		{
			name:      "hash characters in quoted and backtick strings are literal",
			query:     "{cluster=\"physical#pool\", component=\"dm-worker\"} |= \"literal # {cluster=\\\"quoted\\\"}\" | line_format `# {cluster=\"raw\"}`",
			wantQuery: "{cluster=\"physical#pool\", component=\"dm-worker\", cluster_env=\"staging\", namespace=\"tidb-serverless\"} |= \"literal # {cluster=\\\"quoted\\\"}\" | line_format `# {cluster=\"raw\"}`",
		},
	}

	for _, tc := range testCases {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			client := newRegionalLokiTestClient(
				dashboardClusterBody(testRegionalClusterID, "starter", testRegionalProvider, testRegionalRegion),
				datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion),
			)
			client.regionalResponses = []clinicResponse{{body: emptyLokiQueryResponse()}}
			var diagnostics strings.Builder
			ctx := regionalLokiContext(client, "starter")
			ctx.DiagnosticsOut = &diagnostics

			if err := lokiCommandHandler(ctx, regionalLokiArgs(tc.query), &strings.Builder{}); err != nil {
				t.Fatalf("lokiCommandHandler() error = %v", err)
			}

			assertRegionalRequestSequence(t, client.requests, testRegionalUID, 1)
			if got := requireRegionalQueryRequest(t, client.requests).Query.Get("query"); got != tc.wantQuery {
				t.Fatalf("regional LogQL = %q, want comment-aware injection with unchanged comments and strings %q", got, tc.wantQuery)
			}
			if got := diagnostics.String(); got != "" {
				t.Fatalf("diagnostics = %q, want comments ignored by explicit-scope matcher inspection", got)
			}
		})
	}
}

func TestLokiQueryRegionalCommentsCannotProvideOrHideScope(t *testing.T) {
	t.Parallel()

	testCases := []struct {
		name  string
		query string
		want  string
	}{
		{
			name:  "comment-only selector does not provide scope",
			query: `# {cluster="comment-only"}`,
			want:  "found 0 selectors",
		},
		{
			name:  "commented cluster matcher does not provide scope",
			query: "{component=\"dm-worker\" # ignored, cluster=\"comment-only\"\n}",
			want:  "found 0",
		},
		{
			name:  "comment cannot hide a second executable selector",
			query: "{cluster=\"physical-pool\"} # {cluster=\"commented\"}\n{cluster=\"second-pool\"}",
			want:  "found 2 selectors",
		},
	}

	for _, tc := range testCases {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			client := newRegionalLokiTestClient(
				dashboardClusterBody(testRegionalClusterID, "essential", testRegionalProvider, testRegionalRegion),
				datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion),
			)
			err := lokiCommandHandler(regionalLokiContext(client, "essential"), regionalLokiArgs(tc.query), &strings.Builder{})
			if err == nil {
				t.Fatal("lokiCommandHandler() error = nil, want comment-aware selector validation error")
			}
			assertErrorContains(t, err, testRegionalClusterID, tc.want)
			if got, want := len(client.requests), 1; got != want {
				t.Fatalf("request count = %d, want only metadata refresh (%d); paths=%v", got, want, requestPaths(client.requests))
			}
			if got := client.requests[0].Path; got != dashboardClustersPath {
				t.Fatalf("request path = %q, want metadata refresh", got)
			}
		})
	}
}

func TestLokiQueryRegionalUnterminatedQuotedStringStillFailsClosed(t *testing.T) {
	t.Parallel()

	client := newRegionalLokiTestClient(
		dashboardClusterBody(testRegionalClusterID, "starter", testRegionalProvider, testRegionalRegion),
		datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion),
	)
	query := `{cluster="physical-pool"} |= "unterminated # {cluster=\"commented\"}`

	err := lokiCommandHandler(regionalLokiContext(client, "starter"), regionalLokiArgs(query), &strings.Builder{})
	if err == nil {
		t.Fatal("lokiCommandHandler() error = nil, want unterminated quoted-string validation error")
	}
	assertErrorContains(t, err, testRegionalClusterID, "unterminated", "quoted string")
	if got, want := len(client.requests), 1; got != want {
		t.Fatalf("request count = %d, want only metadata refresh (%d); paths=%v", got, want, requestPaths(client.requests))
	}
}

func TestLokiQueryRegionalMetadataValidation(t *testing.T) {
	t.Parallel()

	testCases := []struct {
		name      string
		clusterID string
		body      []byte
		want      string
	}{
		{
			name:      "missing provider",
			clusterID: testRegionalClusterID,
			body:      dashboardClusterBody(testRegionalClusterID, "starter", "", testRegionalRegion),
			want:      "provider",
		},
		{
			name:      "missing region",
			clusterID: testRegionalClusterID,
			body:      dashboardClusterBody(testRegionalClusterID, "essential", testRegionalProvider, ""),
			want:      "region",
		},
		{
			name:      "exact cluster ID absent",
			clusterID: testRegionalClusterID,
			body:      dashboardClusterBody("different-cluster", "starter", testRegionalProvider, testRegionalRegion),
			want:      "not found",
		},
	}

	for _, tc := range testCases {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			client := newRegionalLokiTestClient(tc.body, datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion))
			err := lokiCommandHandler(regionalLokiContext(client, "starter"), regionalLokiArgs(`{cluster="physical-pool"}`), &strings.Builder{})
			if err == nil {
				t.Fatal("lokiCommandHandler() error = nil, want metadata validation error")
			}
			assertErrorContains(t, err, tc.clusterID, tc.want)
			if got, want := len(client.requests), 1; got != want {
				t.Fatalf("request count = %d, want only metadata refresh (%d); paths=%v", got, want, requestPaths(client.requests))
			}
		})
	}
}

func TestLokiQueryStarterProfileRefreshedTierIsAuthoritative(t *testing.T) {
	t.Parallel()

	client := newRegionalLokiTestClient(
		dashboardClusterBody(testRegionalClusterID, "essential v2", "", ""),
		datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion),
	)
	client.legacyResponses = []clinicResponse{{body: emptyLokiQueryResponse()}}

	if err := lokiCommandHandler(regionalLokiContext(client, "starter"), regionalLokiArgs(`{component="tidb"}`), &strings.Builder{}); err != nil {
		t.Fatalf("lokiCommandHandler() error = %v", err)
	}

	if got, want := len(client.requests), 2; got != want {
		t.Fatalf("request count = %d, want metadata refresh plus legacy query (%d); paths=%v", got, want, requestPaths(client.requests))
	}
	if got, want := client.requests[1].Path, "/data-proxy/loki/cluster-42/api/v1/query_range"; got != want {
		t.Fatalf("query path = %q, want %q", got, want)
	}
	assertQueryValue(t, client.requests[1].Query, "query", `{component="tidb", cluster_env="staging", namespace="tidbcluster-42"}`)
}

func TestLokiQueryNonRegionalTiersKeepLegacyRouting(t *testing.T) {
	t.Parallel()

	testCases := []struct {
		name          string
		deployType    string
		parentID      string
		wantNamespace string
	}{
		{name: "essential v2", deployType: "essential v2", wantNamespace: "tidbcluster-42"},
		{name: "essential v2 resource pool", deployType: "essential v2 resource pool", wantNamespace: "tidbcluster-42"},
		{name: "premium", deployType: "premium", parentID: "pool-9", wantNamespace: "tidbpool-9"},
		{name: "BYOC", deployType: "byoc", parentID: "pool-9", wantNamespace: "tidbpool-9"},
		{name: "dedicated", deployType: "dedicated", wantNamespace: "tidbcluster-42"},
		{name: "other tier", deployType: "future tier", wantNamespace: "tidbcluster-42"},
	}

	for _, tc := range testCases {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			client := newRegionalLokiTestClient(nil, nil)
			client.legacyResponses = []clinicResponse{{body: emptyLokiQueryResponse()}}
			ctx := regionalLokiContext(client, tc.deployType)
			ctx.SelectedProfile.ParentID = tc.parentID

			if err := lokiCommandHandler(ctx, regionalLokiArgs(`{} |= "existing behavior"`), &strings.Builder{}); err != nil {
				t.Fatalf("lokiCommandHandler() error = %v", err)
			}

			if got, want := len(client.requests), 1; got != want {
				t.Fatalf("request count = %d, want one legacy query (%d); paths=%v", got, want, requestPaths(client.requests))
			}
			if got, want := client.requests[0].Path, "/data-proxy/loki/cluster-42/api/v1/query_range"; got != want {
				t.Fatalf("query path = %q, want %q", got, want)
			}
			assertQueryValue(t, client.requests[0].Query, "query", fmt.Sprintf(`{cluster_env="staging", namespace=%q} |= "existing behavior"`, tc.wantNamespace))
		})
	}
}

func TestLokiQueryNonRegionalOperatorTextPreservesLegacyAugmentation(t *testing.T) {
	t.Parallel()

	testCases := []struct {
		name          string
		deployType    string
		parentID      string
		wantNamespace string
	}{
		{name: "essential v2", deployType: "essential v2", wantNamespace: "tidbcluster-42"},
		{name: "essential v2 resource pool", deployType: "essential v2 resource pool", wantNamespace: "tidbcluster-42"},
		{name: "premium", deployType: "premium", parentID: "pool-9", wantNamespace: "tidbpool-9"},
		{name: "BYOC", deployType: "byoc", parentID: "pool-9", wantNamespace: "tidbpool-9"},
		{name: "dedicated", deployType: "dedicated", wantNamespace: "tidbcluster-42"},
		{name: "other tier", deployType: "future tier", wantNamespace: "tidbcluster-42"},
	}

	const query = `{cluster_env="prod=~shadow", namespace="custom!=scope!~tail"}`
	for _, tc := range testCases {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			client := newRegionalLokiTestClient(nil, nil)
			client.legacyResponses = []clinicResponse{{body: emptyLokiQueryResponse()}}
			ctx := regionalLokiContext(client, tc.deployType)
			ctx.SelectedProfile.ParentID = tc.parentID
			var diagnostics strings.Builder
			ctx.DiagnosticsOut = &diagnostics

			if err := lokiCommandHandler(ctx, regionalLokiArgs(query), &strings.Builder{}); err != nil {
				t.Fatalf("lokiCommandHandler() error = %v", err)
			}

			if got, want := len(client.requests), 1; got != want {
				t.Fatalf("request count = %d, want one legacy query (%d); paths=%v", got, want, requestPaths(client.requests))
			}
			wantQuery := fmt.Sprintf(
				`{cluster_env="prod=~shadow", namespace="custom!=scope!~tail", cluster_env="staging", namespace=%q}`,
				tc.wantNamespace,
			)
			assertQueryValue(t, client.requests[0].Query, "query", wantQuery)
			if got := diagnostics.String(); got != "" {
				t.Fatalf("diagnostics = %q, want historical no-warning behavior", got)
			}
		})
	}
}

func TestLokiQueryRegionalHydrationDoesNotPersistProfileMetadata(t *testing.T) {
	t.Setenv("XDG_CONFIG_HOME", t.TempDir())
	t.Setenv("HOME", filepath.Join(t.TempDir(), "home"))

	cfg := &config.Config{
		ActiveEnv: "staging",
		Environments: map[string]config.Environment{
			"staging": {URL: "https://example.test"},
		},
		EnvProfiles: map[string]config.EnvProfileContainer{
			"staging": {
				ActiveProfile: "starter-profile",
				Profiles: map[string]config.ProfileRecord{
					"starter-profile": {
						ClusterID:           testRegionalClusterID,
						OrgID:               "org-1",
						ClusterDeployTypeV2: "starter",
					},
				},
			},
		},
	}
	if err := cfg.Save(); err != nil {
		t.Fatalf("Config.Save() error = %v", err)
	}
	path, err := config.ConfigPath()
	if err != nil {
		t.Fatalf("ConfigPath() error = %v", err)
	}
	before, err := os.ReadFile(path)
	if err != nil {
		t.Fatalf("ReadFile(before) error = %v", err)
	}

	selected := cfg.EnvProfiles["staging"].Profiles["starter-profile"]
	client := newRegionalLokiTestClient(
		dashboardClusterBody(testRegionalClusterID, "starter", testRegionalProvider, testRegionalRegion),
		datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion),
	)
	client.regionalResponses = []clinicResponse{{body: emptyLokiQueryResponse()}}
	ctx := &RuntimeContext{
		Client:          client,
		Config:          cfg,
		SelectedEnv:     "staging",
		SelectedProfile: &selected,
	}

	if err := lokiCommandHandler(ctx, regionalLokiArgs(`{cluster="physical-pool"}`), &strings.Builder{}); err != nil {
		t.Fatalf("lokiCommandHandler() error = %v", err)
	}
	after, err := os.ReadFile(path)
	if err != nil {
		t.Fatalf("ReadFile(after) error = %v", err)
	}
	if string(after) != string(before) {
		t.Fatalf("config bytes changed during query\nbefore:\n%s\nafter:\n%s", before, after)
	}
	if got, want := ctx.SelectedProfile.ClusterDeployTypeV2, "starter"; got != want {
		t.Fatalf("selected profile deploy type = %q, want unchanged %q", got, want)
	}
	assertRegionalRequestSequence(t, client.requests, testRegionalUID, 1)
}

func TestLokiQueryRegionalDatasourceCatalogFailuresFailClosed(t *testing.T) {
	t.Parallel()

	testCases := []struct {
		name        string
		catalogBody []byte
		want        []string
	}{
		{name: "malformed JSON", catalogBody: []byte(`{"status":`), want: []string{"catalog", "contract"}},
		{name: "missing envelope", catalogBody: []byte(`{}`), want: []string{"catalog", "contract"}},
		{name: "missing datasources", catalogBody: []byte(`{"status":"success","data":{}}`), want: []string{"catalog", "contract"}},
		{name: "non-success status", catalogBody: []byte(`{"status":"error","data":{"datasources":[]}}`), want: []string{"catalog", "contract", "status"}},
		{name: "empty matched UID", catalogBody: datasourceCatalogBody("", testRegionalProvider, testRegionalRegion), want: []string{"uid"}},
		{name: "no datasource match", catalogBody: datasourceCatalogBody("other", "aws", "us-east-1"), want: []string{testRegionalProvider, testRegionalRegion}},
		{name: "ambiguous datasource match", catalogBody: []byte(`{"status":"success","data":{"datasources":[{"uid":"one","provider":"alicloud","region":"ap-northeast-1"},{"uid":"two","provider":"ALICLOUD","region":"AP-NORTHEAST-1"}]}}`), want: []string{"multiple", testRegionalProvider, testRegionalRegion}},
	}

	for _, tc := range testCases {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			client := newRegionalLokiTestClient(
				dashboardClusterBody(testRegionalClusterID, "starter", testRegionalProvider, testRegionalRegion),
				tc.catalogBody,
			)
			err := lokiCommandHandler(regionalLokiContext(client, "starter"), regionalLokiArgs(`{cluster="physical-pool"}`), &strings.Builder{})
			if err == nil {
				t.Fatal("lokiCommandHandler() error = nil, want catalog validation error")
			}
			assertErrorContains(t, err, append([]string{testRegionalClusterID}, tc.want...)...)
			assertNoLokiQueryRangeRequest(t, client.requests)
			if got, want := len(client.requests), 2; got != want {
				t.Fatalf("request count = %d, want metadata and catalog only (%d); paths=%v", got, want, requestPaths(client.requests))
			}
		})
	}
}

func TestLokiQueryRegionalHTTPFailuresPreserveHTTPErrorAndContext(t *testing.T) {
	t.Parallel()

	t.Run("catalog failure", func(t *testing.T) {
		t.Parallel()

		httpErr := &clinic.HTTPError{StatusCode: http.StatusBadGateway, Status: "502 Bad Gateway", Body: "catalog unavailable"}
		client := newRegionalLokiTestClient(
			dashboardClusterBody(testRegionalClusterID, "essential", testRegionalProvider, testRegionalRegion),
			nil,
		)
		client.catalogErr = httpErr
		err := lokiCommandHandler(regionalLokiContext(client, "essential"), regionalLokiArgs(`{cluster="physical-pool"}`), &strings.Builder{})
		assertWrappedHTTPError(t, err, httpErr, testRegionalClusterID, "catalog", testRegionalProvider, testRegionalRegion)
		assertNoLokiQueryRangeRequest(t, client.requests)
	})

	t.Run("regional query failure", func(t *testing.T) {
		t.Parallel()

		httpErr := &clinic.HTTPError{StatusCode: http.StatusBadGateway, Status: "502 Bad Gateway", Body: "query unavailable"}
		client := newRegionalLokiTestClient(
			dashboardClusterBody(testRegionalClusterID, "starter", testRegionalProvider, testRegionalRegion),
			datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion),
		)
		client.regionalResponses = []clinicResponse{{err: httpErr}}
		err := lokiCommandHandler(regionalLokiContext(client, "starter"), regionalLokiArgs(`{cluster="physical-pool"}`), &strings.Builder{})
		assertWrappedHTTPError(t, err, httpErr, testRegionalClusterID, "datasource", testRegionalUID)
		assertNoLegacyLokiRequest(t, client.requests)
	})
}

func TestLokiQueryRegionalSelectorInjection(t *testing.T) {
	t.Parallel()

	testCases := []struct {
		name          string
		query         string
		skipNamespace bool
		wantQuery     string
	}{
		{
			name:      "injects cluster env and serverless namespace",
			query:     `{cluster="physical-pool", component="dm-worker"}`,
			wantQuery: `{cluster="physical-pool", component="dm-worker", cluster_env="staging", namespace="tidb-serverless"}`,
		},
		{
			name:          "skip namespace injects only cluster env",
			query:         `{cluster="physical-pool", component="dm-worker"}`,
			skipNamespace: true,
			wantQuery:     `{cluster="physical-pool", component="dm-worker", cluster_env="staging"}`,
		},
		{
			name:      "explicit scope remains unchanged",
			query:     `{cluster="physical-pool", cluster_env="prod", namespace="custom"}`,
			wantQuery: `{cluster="physical-pool", cluster_env="prod", namespace="custom"}`,
		},
	}

	for _, tc := range testCases {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			client := newRegionalLokiTestClient(
				dashboardClusterBody(testRegionalClusterID, "essential", testRegionalProvider, testRegionalRegion),
				datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion),
			)
			client.regionalResponses = []clinicResponse{{body: emptyLokiQueryResponse()}}
			args := regionalLokiArgs(tc.query)
			if tc.skipNamespace {
				args = append(args, "--skip-namespace-injection")
			}
			if err := lokiCommandHandler(regionalLokiContext(client, "essential"), args, &strings.Builder{}); err != nil {
				t.Fatalf("lokiCommandHandler() error = %v", err)
			}
			assertQueryValue(t, requireRegionalQueryRequest(t, client.requests).Query, "query", tc.wantQuery)
		})
	}
}

func TestLokiQueryRegionalExplicitScopeWarningUsesPhysicalClusterGuidance(t *testing.T) {
	t.Parallel()

	client := newRegionalLokiTestClient(
		dashboardClusterBody(testRegionalClusterID, "essential", testRegionalProvider, testRegionalRegion),
		datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion),
	)
	client.regionalResponses = []clinicResponse{{body: emptyLokiQueryResponse()}}
	var diagnostics strings.Builder
	ctx := regionalLokiContext(client, "essential")
	ctx.DiagnosticsOut = &diagnostics
	query := `{cluster="physical-pool", cluster_env="prod", namespace=~".*"}`

	if err := lokiCommandHandler(ctx, regionalLokiArgs(query), &strings.Builder{}); err != nil {
		t.Fatalf("lokiCommandHandler() error = %v", err)
	}

	assertRegionalRequestSequence(t, client.requests, testRegionalUID, 1)
	assertQueryValue(t, requireRegionalQueryRequest(t, client.requests).Query, "query", query)
	assertContainsAll(t, diagnostics.String(), []string{
		"physical",
		`cluster="..."`,
		"cluster_env",
		"namespace",
		"omit",
		"cross-namespace",
		"--allow-explicit-scope",
	})
	assertContainsNone(t, diagnostics.String(), []string{"--cluster-id", "use {}", "from {}"})
}

func TestLokiQueryRegionalExplicitScopeValuesContainingOperatorText(t *testing.T) {
	t.Parallel()

	client := newRegionalLokiTestClient(
		dashboardClusterBody(testRegionalClusterID, "starter", testRegionalProvider, testRegionalRegion),
		datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion),
	)
	client.regionalResponses = []clinicResponse{{body: emptyLokiQueryResponse()}}
	var diagnostics strings.Builder
	ctx := regionalLokiContext(client, "starter")
	ctx.DiagnosticsOut = &diagnostics
	query := `{cluster="physical-pool", cluster_env="prod!=shadow", namespace="custom!~scope"}`

	if err := lokiCommandHandler(ctx, regionalLokiArgs(query), &strings.Builder{}); err != nil {
		t.Fatalf("lokiCommandHandler() error = %v", err)
	}

	assertRegionalRequestSequence(t, client.requests, testRegionalUID, 1)
	assertQueryValue(t, requireRegionalQueryRequest(t, client.requests).Query, "query", query)
	assertContainsAll(t, diagnostics.String(), []string{"cluster_env", "namespace"})
	if got := strings.Count(requireRegionalQueryRequest(t, client.requests).Query.Get("query"), "cluster_env="); got != 1 {
		t.Fatalf("cluster_env matcher count = %d, want 1", got)
	}
	if got := strings.Count(requireRegionalQueryRequest(t, client.requests).Query.Get("query"), "namespace="); got != 1 {
		t.Fatalf("namespace matcher count = %d, want 1", got)
	}
}

func TestLokiQueryRegionalPreservesParamsOutputAndGatewayRetry(t *testing.T) {
	t.Parallel()

	client := newRegionalLokiTestClient(
		dashboardClusterBody(testRegionalClusterID, "starter", testRegionalProvider, testRegionalRegion),
		datasourceCatalogBody(testRegionalUID, testRegionalProvider, testRegionalRegion),
	)
	client.regionalResponses = []clinicResponse{
		{err: &clinic.HTTPError{StatusCode: http.StatusGatewayTimeout, Status: "504 Gateway Timeout #1"}},
		{err: &clinic.HTTPError{StatusCode: http.StatusGatewayTimeout, Status: "504 Gateway Timeout #2"}},
		{body: []byte(`{"status":"success","data":{"resultType":"streams","result":[{"stream":{"instance":"dm-master-0"},"values":[["1712123600000000000","regional result"]]}]}}`)},
	}

	var stdout strings.Builder
	err := lokiCommandHandler(regionalLokiContext(client, "starter"), []string{
		"query",
		"--cluster-id", testRegionalClusterID,
		"--start", "1712123000",
		"--end", "1712123600",
		"--limit", "37",
		"--direction", "backward",
		"--step", "15s",
		`{cluster="physical-pool", component="dm-master"}`,
	}, &stdout)
	if err != nil {
		t.Fatalf("lokiCommandHandler() error = %v", err)
	}

	assertRegionalRequestSequence(t, client.requests, testRegionalUID, 3)
	for i, req := range client.requests[2:] {
		assertQueryValue(t, req.Query, "start", "1712123000000000000")
		assertQueryValue(t, req.Query, "end", "1712123600000000000")
		assertQueryValue(t, req.Query, "limit", "37")
		assertQueryValue(t, req.Query, "direction", "backward")
		assertQueryValue(t, req.Query, "step", "15s")
		if got, want := req.Query.Get("query"), `{cluster="physical-pool", component="dm-master", cluster_env="staging", namespace="tidb-serverless"}`; got != want {
			t.Fatalf("query request %d LogQL = %q, want %q", i+1, got, want)
		}
	}
	if got, want := stdout.String(), "[instance=dm-master-0] regional result\n"; got != want {
		t.Fatalf("stdout = %q, want %q", got, want)
	}
}

type regionalLokiTestClient struct {
	dashboardBody     []byte
	dashboardErr      error
	catalogBody       []byte
	catalogErr        error
	regionalResponses []clinicResponse
	legacyResponses   []clinicResponse
	requests          []clinic.Request
}

func newRegionalLokiTestClient(dashboardBody, catalogBody []byte) *regionalLokiTestClient {
	return &regionalLokiTestClient{
		dashboardBody:   dashboardBody,
		catalogBody:     catalogBody,
		legacyResponses: []clinicResponse{{body: emptyLokiQueryResponse()}},
	}
}

func (c *regionalLokiTestClient) Do(_ context.Context, req clinic.Request, out any) error {
	c.requests = append(c.requests, req)

	var response clinicResponse
	switch {
	case req.Path == dashboardClustersPath:
		response = clinicResponse{body: c.dashboardBody, err: c.dashboardErr}
	case req.Path == "/data-proxy/api/v1/system-log-datasources":
		response = clinicResponse{body: c.catalogBody, err: c.catalogErr}
	case strings.HasPrefix(req.Path, "/data-proxy/api/v1/system-log-datasources/"):
		response, c.regionalResponses = popClinicResponse(c.regionalResponses)
	case strings.HasPrefix(req.Path, "/data-proxy/loki/"):
		response, c.legacyResponses = popClinicResponse(c.legacyResponses)
	default:
		return fmt.Errorf("unexpected test request: %s %s", req.Method, req.Path)
	}
	if response.err != nil {
		return response.err
	}
	if raw, ok := out.(*[]byte); ok {
		*raw = append((*raw)[:0], response.body...)
		return nil
	}
	if len(response.body) == 0 || out == nil {
		return nil
	}
	return json.Unmarshal(response.body, out)
}

func popClinicResponse(responses []clinicResponse) (clinicResponse, []clinicResponse) {
	if len(responses) == 0 {
		return clinicResponse{body: emptyLokiQueryResponse()}, nil
	}
	return responses[0], responses[1:]
}

func regionalLokiContext(client Client, deployType string) *RuntimeContext {
	return &RuntimeContext{
		Client:      client,
		SelectedEnv: "staging",
		SelectedProfile: &config.ProfileRecord{
			ClusterID:           testRegionalClusterID,
			OrgID:               "org-1",
			ClusterDeployTypeV2: deployType,
		},
	}
}

func legacyLokiTestContext(client Client) *RuntimeContext {
	return regionalLokiContext(client, "dedicated")
}

func regionalLokiArgs(query string) []string {
	args := []string{
		"query",
		"--cluster-id", testRegionalClusterID,
		"--start", "1712123000",
		"--end", "1712123600",
	}
	if query != "" {
		args = append(args, query)
	}
	return args
}

func dashboardClusterBody(clusterID, deployType, provider, region string) []byte {
	return []byte(fmt.Sprintf(
		`{"items":[{"clusterID":%q,"orgID":"org-1","clusterDeployTypeV2":%q,"clusterProviderName":%q,"clusterRegionName":%q}]}`,
		clusterID,
		deployType,
		provider,
		region,
	))
}

func datasourceCatalogBody(uid, provider, region string) []byte {
	return []byte(fmt.Sprintf(
		`{"status":"success","data":{"datasources":[{"uid":%q,"provider":%q,"region":%q}]}}`,
		uid,
		provider,
		region,
	))
}

func emptyLokiQueryResponse() []byte {
	return []byte(`{"status":"success","data":{"resultType":"streams","result":[]}}`)
}

func assertRegionalRequestSequence(t *testing.T, requests []clinic.Request, uid string, queryAttempts int) {
	t.Helper()
	if got, want := len(requests), 2+queryAttempts; got != want {
		t.Fatalf("request count = %d, want %d; paths=%v", got, want, requestPaths(requests))
	}
	if got := requests[0].Path; got != dashboardClustersPath {
		t.Fatalf("metadata request path = %q, want %q", got, dashboardClustersPath)
	}
	assertQueryValue(t, requests[0].Query, "cluster_id", testRegionalClusterID)
	assertQueryValue(t, requests[0].Query, "show_deleted", "true")
	if got, want := requests[1].Path, "/data-proxy/api/v1/system-log-datasources"; got != want {
		t.Fatalf("catalog request path = %q, want %q", got, want)
	}
	wantPath := "/data-proxy/api/v1/system-log-datasources/" + uid + "/logs/query-range"
	for i, req := range requests[2:] {
		if got := req.Path; got != wantPath {
			t.Fatalf("regional query %d path = %q, want %q", i+1, got, wantPath)
		}
	}
	assertNoLegacyLokiRequest(t, requests)
}

func requireRegionalQueryRequest(t *testing.T, requests []clinic.Request) clinic.Request {
	t.Helper()
	if len(requests) < 3 {
		t.Fatalf("request count = %d, want metadata, catalog, and regional query; paths=%v", len(requests), requestPaths(requests))
	}
	return requests[2]
}

func assertNoLokiQueryRangeRequest(t *testing.T, requests []clinic.Request) {
	t.Helper()
	for _, req := range requests {
		if strings.Contains(req.Path, "/logs/query-range") || strings.Contains(req.Path, "/api/v1/query_range") {
			t.Fatalf("unexpected query-range request after routing failure: %s", req.Path)
		}
	}
}

func assertNoLegacyLokiRequest(t *testing.T, requests []clinic.Request) {
	t.Helper()
	for _, req := range requests {
		if strings.HasPrefix(req.Path, "/data-proxy/loki/") {
			t.Fatalf("unexpected legacy Loki fallback request: %s", req.Path)
		}
	}
}

func assertWrappedHTTPError(t *testing.T, err error, want *clinic.HTTPError, context ...string) {
	t.Helper()
	if err == nil {
		t.Fatal("error = nil, want wrapped *clinic.HTTPError")
	}
	var got *clinic.HTTPError
	if !errors.As(err, &got) {
		t.Fatalf("error type = %T, want wrapped *clinic.HTTPError: %v", err, err)
	}
	if got != want {
		t.Fatalf("errors.As target = %p, want original %p", got, want)
	}
	assertErrorContains(t, err, context...)
}

func assertErrorContains(t *testing.T, err error, values ...string) {
	t.Helper()
	got := strings.ToLower(err.Error())
	for _, value := range values {
		if !strings.Contains(got, strings.ToLower(value)) {
			t.Fatalf("error = %q, want substring %q", err, value)
		}
	}
}

func requestPaths(requests []clinic.Request) []string {
	paths := make([]string, 0, len(requests))
	for _, request := range requests {
		paths = append(paths, request.Path)
	}
	return paths
}

func stringPointer(value string) *string {
	return &value
}