package globalapi

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"testing"

	"github.com/gin-gonic/gin"
	"github.com/stretchr/testify/assert"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/message"
	"github.com/tidbcloud/observability/pkg/common/structs"
	globalstructs "github.com/tidbcloud/observability/pkg/globalservice/structs"
	commonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	nextgenv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	tidbmgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"google.golang.org/grpc"
)

// mockClusterServiceClient implements tidbmgmt.ClusterServiceClient for testing.
type mockClusterServiceClient struct {
	tidbmgmt.ClusterServiceClient // embed interface for methods we don't test
	listClustersFn                func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error)
}

func (m *mockClusterServiceClient) ListClusters(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
	return m.listClustersFn(ctx, in, opts...)
}

type mockResourcePoolServiceClient struct {
	tidbmgmt.ResourcePoolServiceClient
	listResourcePoolsFn func(ctx context.Context, in *nextgenv1.ListResourcePoolsRequest, opts ...grpc.CallOption) (*nextgenv1.ListResourcePoolsResponse, error)
}

func (m *mockResourcePoolServiceClient) ListResourcePools(ctx context.Context, in *nextgenv1.ListResourcePoolsRequest, opts ...grpc.CallOption) (*nextgenv1.ListResourcePoolsResponse, error) {
	return m.listResourcePoolsFn(ctx, in, opts...)
}

// --- resolveBizType ---

func Test_resolveBizType(t *testing.T) {
	tests := []struct {
		name        string
		deployType  tidbmgmt.DeployType
		servicePlan commonv1.ServicePlan
		env         string
		want        string
		wantErr     bool
	}{
		{
			name:       "dedicated prod has no prefix",
			deployType: tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED,
			env:        constants.EnvProd,
			want:       "tidbcloud",
		},
		{
			name:       "dedicated dev",
			deployType: tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED,
			env:        constants.EnvDev,
			want:       "dev-tidbcloud",
		},
		{
			name:       "dedicated staging",
			deployType: tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED,
			env:        constants.EnvStaging,
			want:       "staging-tidbcloud",
		},
		{
			name:       "nextgen prod",
			deployType: tidbmgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
			env:        constants.EnvProd,
			want:       "prod-tidbcloud-nextgen",
		},
		{
			name:       "nextgen dev",
			deployType: tidbmgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
			env:        constants.EnvDev,
			want:       "dev-tidbcloud-nextgen",
		},
		{
			name:       "nextgen staging",
			deployType: tidbmgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
			env:        constants.EnvStaging,
			want:       "staging-tidbcloud-nextgen",
		},
		{
			name:        "byoc nextgen dev uses bare byoc bizType",
			deployType:  tidbmgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
			servicePlan: commonv1.ServicePlan_BYOC,
			env:         constants.EnvDev,
			want:        "byoc",
		},
		{
			name:        "byoc nextgen prod uses bare byoc bizType",
			deployType:  tidbmgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
			servicePlan: commonv1.ServicePlan_BYOC,
			env:         constants.EnvProd,
			want:        "byoc",
		},
		{
			name:       "serverless prod",
			deployType: tidbmgmt.DeployType_DEPLOY_TYPE_SERVERLESS,
			env:        constants.EnvProd,
			want:       "prod-devtier",
		},
		{
			name:       "serverless dev",
			deployType: tidbmgmt.DeployType_DEPLOY_TYPE_SERVERLESS,
			env:        constants.EnvDev,
			want:       "dev-devtier",
		},
		{
			name:       "serverless staging",
			deployType: tidbmgmt.DeployType_DEPLOY_TYPE_SERVERLESS,
			env:        constants.EnvStaging,
			want:       "staging-devtier",
		},
		{
			name:       "unknown deploy type",
			deployType: tidbmgmt.DeployType(999),
			env:        constants.EnvProd,
			wantErr:    true,
		},
		{
			name:       "unspecified deploy type",
			deployType: tidbmgmt.DeployType_DEPLOY_TYPE_UNSPECIFIED,
			env:        constants.EnvDev,
			wantErr:    true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := resolveBizType(tt.deployType, tt.servicePlan, tt.env)
			if tt.wantErr {
				assert.Error(t, err)
				assert.Contains(t, err.Error(), "unknown deploy type")
			} else {
				assert.NoError(t, err)
				assert.Equal(t, tt.want, got)
			}
		})
	}
}

// --- parseUint64List ---

func Test_parseUint64List(t *testing.T) {
	tests := []struct {
		name    string
		input   string
		want    []uint64
		wantErr bool
	}{
		{
			name:  "empty string",
			input: "",
			want:  nil,
		},
		{
			name:  "single value",
			input: "123",
			want:  []uint64{123},
		},
		{
			name:  "multiple values",
			input: "1,2,3",
			want:  []uint64{1, 2, 3},
		},
		{
			name:  "whitespace handling",
			input: " 1 , 2 , 3 ",
			want:  []uint64{1, 2, 3},
		},
		{
			name:  "trailing comma produces empty part which is skipped",
			input: "1,2,",
			want:  []uint64{1, 2},
		},
		{
			name:  "leading comma produces empty part which is skipped",
			input: ",1,2",
			want:  []uint64{1, 2},
		},
		{
			name:    "invalid value",
			input:   "1,abc,3",
			wantErr: true,
		},
		{
			name:    "negative value",
			input:   "1,-2,3",
			wantErr: true,
		},
		{
			name:    "float value",
			input:   "1.5",
			wantErr: true,
		},
		{
			name:  "large uint64",
			input: "18446744073709551615",
			want:  []uint64{18446744073709551615},
		},
		{
			name:    "overflow uint64",
			input:   "18446744073709551616",
			wantErr: true,
		},
		{
			name:  "all whitespace parts are skipped",
			input: " , , ",
			want:  []uint64{},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := parseUint64List(tt.input)
			if tt.wantErr {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
				assert.Equal(t, tt.want, got)
			}
		})
	}
}

// --- parseAndValidateFilters ---

func Test_parseAndValidateFilters(t *testing.T) {
	tests := []struct {
		name    string
		req     message.ListClustersWithMetadataReq
		want    *parsedFilters
		wantErr bool
		errMsg  string
	}{
		{
			name: "valid tenantIds only",
			req:  message.ListClustersWithMetadataReq{TenantIds: "100,200"},
			want: &parsedFilters{TenantIds: []uint64{100, 200}},
		},
		{
			name: "valid projectIds only",
			req:  message.ListClustersWithMetadataReq{ProjectIds: "300"},
			want: &parsedFilters{ProjectIds: []uint64{300}},
		},
		{
			name: "valid clusterIds only",
			req:  message.ListClustersWithMetadataReq{ClusterIds: "400,500,600"},
			want: &parsedFilters{ClusterIds: []uint64{400, 500, 600}},
		},
		{
			name: "all three provided",
			req:  message.ListClustersWithMetadataReq{TenantIds: "1", ProjectIds: "2", ClusterIds: "3"},
			want: &parsedFilters{TenantIds: []uint64{1}, ProjectIds: []uint64{2}, ClusterIds: []uint64{3}},
		},
		{
			name:    "all empty requires at least one",
			req:     message.ListClustersWithMetadataReq{},
			wantErr: true,
			errMsg:  "at least one",
		},
		{
			name:    "invalid tenantIds",
			req:     message.ListClustersWithMetadataReq{TenantIds: "abc"},
			wantErr: true,
			errMsg:  "invalid tenantIds",
		},
		{
			name:    "invalid projectIds",
			req:     message.ListClustersWithMetadataReq{ProjectIds: "xyz"},
			wantErr: true,
			errMsg:  "invalid projectIds",
		},
		{
			name:    "invalid clusterIds",
			req:     message.ListClustersWithMetadataReq{ClusterIds: "-1"},
			wantErr: true,
			errMsg:  "invalid clusterIds",
		},
		{
			name:    "tenantIds valid but projectIds invalid",
			req:     message.ListClustersWithMetadataReq{TenantIds: "100", ProjectIds: "not-a-number"},
			wantErr: true,
			errMsg:  "invalid projectIds",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := parseAndValidateFilters(tt.req)
			if tt.wantErr {
				assert.Error(t, err)
				if tt.errMsg != "" {
					assert.Contains(t, err.Error(), tt.errMsg)
				}
			} else {
				assert.NoError(t, err)
				assert.Equal(t, tt.want.TenantIds, got.TenantIds)
				assert.Equal(t, tt.want.ProjectIds, got.ProjectIds)
				assert.Equal(t, tt.want.ClusterIds, got.ClusterIds)
			}
		})
	}
}

// --- groupClustersByRoute ---

func Test_groupClustersByRoute(t *testing.T) {
	t.Run("empty clusters", func(t *testing.T) {
		groups := groupClustersByRoute(nil, "dev")
		assert.Empty(t, groups)
	})

	t.Run("single cluster", func(t *testing.T) {
		clusters := []*tidbmgmt.Cluster{
			{
				ClusterId:      100,
				OrganizationId: 1,
				CloudProvider:  "aws",
				Region:         "us-west-2",
				DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED,
			},
		}
		groups := groupClustersByRoute(clusters, "dev")
		assert.Len(t, groups, 1)
		key := regionalQueryKey{Vendor: "aws", Region: "us-west-2", BizType: "dev-tidbcloud", TenantID: "1"}
		g, ok := groups[key]
		assert.True(t, ok)
		assert.True(t, g.clusterIDs["100"])
	})

	t.Run("multiple clusters same route", func(t *testing.T) {
		clusters := []*tidbmgmt.Cluster{
			{ClusterId: 100, OrganizationId: 1, CloudProvider: "aws", Region: "us-west-2", DeployType: tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED},
			{ClusterId: 200, OrganizationId: 1, CloudProvider: "aws", Region: "us-west-2", DeployType: tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED},
		}
		groups := groupClustersByRoute(clusters, "dev")
		assert.Len(t, groups, 1)
		key := regionalQueryKey{Vendor: "aws", Region: "us-west-2", BizType: "dev-tidbcloud", TenantID: "1"}
		g := groups[key]
		assert.Len(t, g.clusterIDs, 2)
		assert.True(t, g.clusterIDs["100"])
		assert.True(t, g.clusterIDs["200"])
	})

	t.Run("clusters with different routes", func(t *testing.T) {
		clusters := []*tidbmgmt.Cluster{
			{ClusterId: 100, OrganizationId: 1, CloudProvider: "aws", Region: "us-west-2", DeployType: tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED},
			{ClusterId: 200, OrganizationId: 2, CloudProvider: "gcp", Region: "us-central1", DeployType: tidbmgmt.DeployType_DEPLOY_TYPE_SERVERLESS},
		}
		groups := groupClustersByRoute(clusters, "dev")
		assert.Len(t, groups, 2)
	})

	t.Run("clusters with different tenants same vendor region biztype", func(t *testing.T) {
		clusters := []*tidbmgmt.Cluster{
			{ClusterId: 100, OrganizationId: 1, CloudProvider: "aws", Region: "us-west-2", DeployType: tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED},
			{ClusterId: 200, OrganizationId: 2, CloudProvider: "aws", Region: "us-west-2", DeployType: tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED},
		}
		groups := groupClustersByRoute(clusters, "dev")
		// Different tenants produce different route keys
		assert.Len(t, groups, 2)
	})

	t.Run("unknown deploy type skipped", func(t *testing.T) {
		clusters := []*tidbmgmt.Cluster{
			{ClusterId: 100, OrganizationId: 1, CloudProvider: "aws", Region: "us-west-2", DeployType: tidbmgmt.DeployType(999)},
		}
		groups := groupClustersByRoute(clusters, "dev")
		assert.Empty(t, groups)
	})

	t.Run("mixed valid and invalid deploy types", func(t *testing.T) {
		clusters := []*tidbmgmt.Cluster{
			{ClusterId: 100, OrganizationId: 1, CloudProvider: "aws", Region: "us-west-2", DeployType: tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED},
			{ClusterId: 200, OrganizationId: 1, CloudProvider: "aws", Region: "us-west-2", DeployType: tidbmgmt.DeployType(999)},
			{ClusterId: 300, OrganizationId: 1, CloudProvider: "aws", Region: "us-west-2", DeployType: tidbmgmt.DeployType_DEPLOY_TYPE_SERVERLESS},
		}
		groups := groupClustersByRoute(clusters, "dev")
		// cluster 200 with unknown type is skipped, 100 and 300 have different bizTypes
		assert.Len(t, groups, 2)
	})

	t.Run("prod dedicated has no prefix in bizType", func(t *testing.T) {
		clusters := []*tidbmgmt.Cluster{
			{ClusterId: 100, OrganizationId: 1, CloudProvider: "aws", Region: "us-west-2", DeployType: tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED},
		}
		groups := groupClustersByRoute(clusters, "prod")
		key := regionalQueryKey{Vendor: "aws", Region: "us-west-2", BizType: "tidbcloud", TenantID: "1"}
		_, ok := groups[key]
		assert.True(t, ok)
	})

	t.Run("all three deploy types in prod", func(t *testing.T) {
		clusters := []*tidbmgmt.Cluster{
			{ClusterId: 100, OrganizationId: 1, CloudProvider: "aws", Region: "us-west-2", DeployType: tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED},
			{ClusterId: 200, OrganizationId: 1, CloudProvider: "aws", Region: "us-west-2", DeployType: tidbmgmt.DeployType_DEPLOY_TYPE_NEXTGEN},
			{ClusterId: 300, OrganizationId: 1, CloudProvider: "aws", Region: "us-west-2", DeployType: tidbmgmt.DeployType_DEPLOY_TYPE_SERVERLESS},
		}
		groups := groupClustersByRoute(clusters, "prod")
		assert.Len(t, groups, 3)
		// Verify specific bizTypes
		assert.NotNil(t, groups[regionalQueryKey{Vendor: "aws", Region: "us-west-2", BizType: "tidbcloud", TenantID: "1"}])
		assert.NotNil(t, groups[regionalQueryKey{Vendor: "aws", Region: "us-west-2", BizType: "prod-tidbcloud-nextgen", TenantID: "1"}])
		assert.NotNil(t, groups[regionalQueryKey{Vendor: "aws", Region: "us-west-2", BizType: "prod-devtier", TenantID: "1"}])
	})
}

// --- aggregateRegionalResults ---

func Test_aggregateRegionalResults(t *testing.T) {
	t.Run("empty channel", func(t *testing.T) {
		ch := make(chan metadataRegionResult)
		close(ch)
		clusters, partial := aggregateRegionalResults(ch)
		assert.Empty(t, clusters)
		assert.False(t, partial)
	})

	t.Run("single success result", func(t *testing.T) {
		ch := make(chan metadataRegionResult, 1)
		ch <- metadataRegionResult{
			resp: message.QueryClustersResp{
				Clusters: []structs.ClusterBaseInfo{
					{ApplicationID: "100"},
					{ApplicationID: "200"},
				},
			},
			primary:    true,
			clusterIDs: map[string]bool{"100": true, "200": true},
		}
		close(ch)
		clusters, partial := aggregateRegionalResults(ch)
		assert.Len(t, clusters, 2)
		assert.False(t, partial)
	})

	t.Run("filters clusters by clusterIDs", func(t *testing.T) {
		ch := make(chan metadataRegionResult, 1)
		ch <- metadataRegionResult{
			resp: message.QueryClustersResp{
				Clusters: []structs.ClusterBaseInfo{
					{ApplicationID: "100"},
					{ApplicationID: "200"},
					{ApplicationID: "300"},
				},
			},
			primary:    true,
			clusterIDs: map[string]bool{"100": true, "300": true},
		}
		close(ch)
		clusters, partial := aggregateRegionalResults(ch)
		assert.Len(t, clusters, 2)
		assert.Equal(t, "100", clusters[0].ApplicationID)
		assert.Equal(t, "300", clusters[1].ApplicationID)
		assert.False(t, partial)
	})

	t.Run("nil clusterIDs includes all clusters (pushdown mode)", func(t *testing.T) {
		ch := make(chan metadataRegionResult, 1)
		ch <- metadataRegionResult{
			resp: message.QueryClustersResp{
				Clusters: []structs.ClusterBaseInfo{
					{ApplicationID: "100"},
					{ApplicationID: "200"},
				},
			},
			primary:    true,
			clusterIDs: nil,
		}
		close(ch)
		clusters, partial := aggregateRegionalResults(ch)
		assert.Len(t, clusters, 2)
		assert.False(t, partial)
	})

	t.Run("mixed pushdown and tenant-wide results", func(t *testing.T) {
		ch := make(chan metadataRegionResult, 3)
		ch <- metadataRegionResult{
			resp: message.QueryClustersResp{
				Clusters: []structs.ClusterBaseInfo{{ApplicationID: "100"}},
			},
			primary:    true,
			clusterIDs: nil,
		}
		ch <- metadataRegionResult{
			resp: message.QueryClustersResp{
				Clusters: []structs.ClusterBaseInfo{{ApplicationID: "200"}},
			},
			primary:    false,
			clusterIDs: nil,
		}
		ch <- metadataRegionResult{
			resp: message.QueryClustersResp{
				Clusters: []structs.ClusterBaseInfo{
					{ApplicationID: "300"},
					{ApplicationID: "400"},
					{ApplicationID: "500"},
					{ApplicationID: "600"},
				},
			},
			primary:    true,
			clusterIDs: map[string]bool{"300": true, "400": true, "500": true, "600": true},
		}
		close(ch)
		clusters, partial := aggregateRegionalResults(ch)
		assert.Len(t, clusters, 6)
		assert.False(t, partial)
	})

	t.Run("primary error sets partial", func(t *testing.T) {
		ch := make(chan metadataRegionResult, 1)
		ch <- metadataRegionResult{
			err:        fmt.Errorf("primary error"),
			primary:    true,
			clusterIDs: map[string]bool{"100": true},
		}
		close(ch)
		clusters, partial := aggregateRegionalResults(ch)
		assert.Empty(t, clusters)
		assert.True(t, partial)
	})

	t.Run("non-primary error sets partial", func(t *testing.T) {
		ch := make(chan metadataRegionResult, 1)
		ch <- metadataRegionResult{
			err:        fmt.Errorf("non-primary error"),
			primary:    false,
			clusterIDs: map[string]bool{"100": true},
		}
		close(ch)
		clusters, partial := aggregateRegionalResults(ch)
		assert.Empty(t, clusters)
		assert.True(t, partial)
	})

	t.Run("mixed success and error", func(t *testing.T) {
		ch := make(chan metadataRegionResult, 3)
		ch <- metadataRegionResult{
			resp: message.QueryClustersResp{
				Clusters: []structs.ClusterBaseInfo{{ApplicationID: "100"}},
			},
			primary:    true,
			clusterIDs: map[string]bool{"100": true},
		}
		ch <- metadataRegionResult{
			err:        fmt.Errorf("region-2 error"),
			primary:    false,
			clusterIDs: map[string]bool{"200": true},
		}
		ch <- metadataRegionResult{
			resp: message.QueryClustersResp{
				Clusters: []structs.ClusterBaseInfo{{ApplicationID: "300"}},
			},
			primary:    false,
			clusterIDs: map[string]bool{"300": true},
		}
		close(ch)
		clusters, partial := aggregateRegionalResults(ch)
		assert.Len(t, clusters, 2)
		assert.True(t, partial)
	})

	t.Run("multiple regions all successful", func(t *testing.T) {
		ch := make(chan metadataRegionResult, 2)
		ch <- metadataRegionResult{
			resp: message.QueryClustersResp{
				Clusters: []structs.ClusterBaseInfo{{ApplicationID: "100"}, {ApplicationID: "101"}},
			},
			primary:    true,
			clusterIDs: map[string]bool{"100": true, "101": true},
		}
		ch <- metadataRegionResult{
			resp: message.QueryClustersResp{
				Clusters: []structs.ClusterBaseInfo{{ApplicationID: "200"}},
			},
			primary:    false,
			clusterIDs: map[string]bool{"200": true},
		}
		close(ch)
		clusters, partial := aggregateRegionalResults(ch)
		assert.Len(t, clusters, 3)
		assert.False(t, partial)
	})
}

// --- fetchRegionalClusters ---

func Test_fetchRegionalClusters(t *testing.T) {
	t.Run("success", func(t *testing.T) {
		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			resp := map[string]interface{}{
				"code":    0,
				"message": "",
				"data": map[string]interface{}{
					"clusters": []map[string]interface{}{
						{"id": "c1", "applicationID": "100", "vendor": "aws", "region": "us-west-2"},
						{"id": "c2", "applicationID": "200", "vendor": "aws", "region": "us-west-2"},
					},
					"partial": false,
				},
			}
			w.WriteHeader(http.StatusOK)
			json.NewEncoder(w).Encode(resp) //nolint:errcheck
		}))
		defer srv.Close()

		ch := make(chan metadataRegionResult, 1)
		clusterIDs := map[string]bool{"100": true, "200": true}
		fetchRegionalClusters(srv.URL, map[string]string{}, clusterIDs, true, ch)
		result := <-ch

		assert.NoError(t, result.err)
		assert.True(t, result.primary)
		assert.Len(t, result.resp.Clusters, 2)
		assert.Equal(t, clusterIDs, result.clusterIDs)
	})

	t.Run("non-200 status code", func(t *testing.T) {
		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			w.WriteHeader(http.StatusInternalServerError)
			w.Write([]byte(`internal server error`)) //nolint:errcheck
		}))
		defer srv.Close()

		ch := make(chan metadataRegionResult, 1)
		fetchRegionalClusters(srv.URL, map[string]string{}, map[string]bool{}, false, ch)
		result := <-ch

		assert.Error(t, result.err)
		assert.Contains(t, result.err.Error(), "500")
		assert.False(t, result.primary)
	})

	t.Run("invalid JSON response", func(t *testing.T) {
		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			w.WriteHeader(http.StatusOK)
			w.Write([]byte(`not valid json`)) //nolint:errcheck
		}))
		defer srv.Close()

		ch := make(chan metadataRegionResult, 1)
		fetchRegionalClusters(srv.URL, map[string]string{}, map[string]bool{}, true, ch)
		result := <-ch

		assert.Error(t, result.err)
		assert.True(t, result.primary)
	})

	t.Run("network error", func(t *testing.T) {
		ch := make(chan metadataRegionResult, 1)
		fetchRegionalClusters("http://localhost:1", map[string]string{}, map[string]bool{}, true, ch)
		result := <-ch

		assert.Error(t, result.err)
		assert.True(t, result.primary)
	})

	t.Run("empty clusters response", func(t *testing.T) {
		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			resp := map[string]interface{}{
				"code":    0,
				"message": "",
				"data": map[string]interface{}{
					"clusters": []map[string]interface{}{},
					"partial":  false,
				},
			}
			w.WriteHeader(http.StatusOK)
			json.NewEncoder(w).Encode(resp) //nolint:errcheck
		}))
		defer srv.Close()

		ch := make(chan metadataRegionResult, 1)
		fetchRegionalClusters(srv.URL, map[string]string{}, map[string]bool{}, false, ch)
		result := <-ch

		assert.NoError(t, result.err)
		assert.Empty(t, result.resp.Clusters)
	})

	t.Run("headers are forwarded", func(t *testing.T) {
		var receivedAuth string
		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			receivedAuth = r.Header.Get("Authorization")
			w.WriteHeader(http.StatusOK)
			json.NewEncoder(w).Encode(map[string]interface{}{"data": map[string]interface{}{"clusters": []interface{}{}}}) //nolint:errcheck
		}))
		defer srv.Close()

		ch := make(chan metadataRegionResult, 1)
		headers := map[string]string{"Authorization": "Bearer test-token"}
		fetchRegionalClusters(srv.URL, headers, map[string]bool{}, true, ch)
		<-ch

		assert.Equal(t, "Bearer test-token", receivedAuth)
	})

	t.Run("nil clusterIDs passed through for pushdown", func(t *testing.T) {
		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			resp := map[string]interface{}{
				"code":    0,
				"message": "",
				"data": map[string]interface{}{
					"clusters": []map[string]interface{}{
						{"id": "c1", "applicationID": "100", "vendor": "aws", "region": "us-west-2"},
					},
					"partial": false,
				},
			}
			w.WriteHeader(http.StatusOK)
			json.NewEncoder(w).Encode(resp) //nolint:errcheck
		}))
		defer srv.Close()

		ch := make(chan metadataRegionResult, 1)
		fetchRegionalClusters(srv.URL, map[string]string{}, nil, true, ch)
		result := <-ch

		assert.NoError(t, result.err)
		assert.Nil(t, result.clusterIDs)
		assert.Len(t, result.resp.Clusters, 1)
	})
}

func Test_estimateResultChannelSize(t *testing.T) {
	t.Run("empty groups", func(t *testing.T) {
		size := estimateResultChannelSize(map[regionalQueryKey]*clusterGroup{})
		assert.Equal(t, 1, size)
	})

	t.Run("single group with 1 cluster uses pushdown", func(t *testing.T) {
		groups := map[regionalQueryKey]*clusterGroup{
			{}: {clusterIDs: map[string]bool{"100": true}},
		}
		assert.Equal(t, 1, estimateResultChannelSize(groups))
	})

	t.Run("single group with 3 clusters uses pushdown", func(t *testing.T) {
		groups := map[regionalQueryKey]*clusterGroup{
			{}: {clusterIDs: map[string]bool{"100": true, "200": true, "300": true}},
		}
		assert.Equal(t, 3, estimateResultChannelSize(groups))
	})

	t.Run("single group with 4 clusters uses tenant-wide", func(t *testing.T) {
		groups := map[regionalQueryKey]*clusterGroup{
			{}: {clusterIDs: map[string]bool{"100": true, "200": true, "300": true, "400": true}},
		}
		assert.Equal(t, 1, estimateResultChannelSize(groups))
	})

	t.Run("mixed pushdown and tenant-wide groups", func(t *testing.T) {
		k1 := regionalQueryKey{Vendor: "aws", Region: "us-west-2", BizType: "dev-tidbcloud", TenantID: "1"}
		k2 := regionalQueryKey{Vendor: "gcp", Region: "us-central1", BizType: "dev-devtier", TenantID: "2"}
		groups := map[regionalQueryKey]*clusterGroup{
			k1: {clusterIDs: map[string]bool{"100": true, "200": true}},
			k2: {clusterIDs: map[string]bool{"300": true, "400": true, "500": true, "600": true}},
		}
		assert.Equal(t, 3, estimateResultChannelSize(groups))
	})
}

func Test_signHeaders(t *testing.T) {
	gin.SetMode(gin.TestMode)

	t.Run("returns error when signing fails with invalid credentials", func(t *testing.T) {
		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/", nil)

		t.Setenv("AWS_ROLE_ARN", "arn:aws:iam::1234567890:role/test")
		t.Setenv("AWS_WEB_IDENTITY_TOKEN_FILE", "./main_test.go")
		t.Setenv("AWS_REGION", "us-west-2")

		routeInfo := &globalstructs.RegionalRouter{
			AccessPrincipal:  "arn:aws:iam::1234567890:role/regional-role",
			AccessExternalID: "external-id",
		}

		headers, err := signHeaders(c, routeInfo, "us-west-2", "http://example.com/api/v1/clusters")

		assert.Error(t, err)
		assert.Nil(t, headers)
	})
}

func Test_listClustersFromResourcePools(t *testing.T) {
	t.Run("empty resource pool ids returns empty cluster response", func(t *testing.T) {
		clusterClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				t.Fatal("cluster client should not be called when resourcePoolIDs is empty")
				return nil, nil
			},
		}
		resourcePoolClient := &mockResourcePoolServiceClient{
			listResourcePoolsFn: func(ctx context.Context, in *nextgenv1.ListResourcePoolsRequest, opts ...grpc.CallOption) (*nextgenv1.ListResourcePoolsResponse, error) {
				t.Fatal("resource pool client should not be called when resourcePoolIDs is empty")
				return nil, nil
			},
		}

		resp, err := listClustersFromResourcePools(context.Background(), clusterClient, resourcePoolClient, nil)
		assert.NoError(t, err)
		assert.NotNil(t, resp)
		assert.Empty(t, resp.Clusters)
	})

	t.Run("resource pool service error", func(t *testing.T) {
		clusterClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				t.Fatal("cluster client should not be called when resource pool list fails")
				return nil, nil
			},
		}
		resourcePoolClient := &mockResourcePoolServiceClient{
			listResourcePoolsFn: func(ctx context.Context, in *nextgenv1.ListResourcePoolsRequest, opts ...grpc.CallOption) (*nextgenv1.ListResourcePoolsResponse, error) {
				return nil, fmt.Errorf("rpc unavailable")
			},
		}

		resp, err := listClustersFromResourcePools(context.Background(), clusterClient, resourcePoolClient, []uint64{11})
		assert.Nil(t, resp)
		assert.Error(t, err)
	})

	t.Run("no cluster ids in resource pool response returns empty", func(t *testing.T) {
		clusterClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				t.Fatal("cluster client should not be called when no instances found")
				return nil, nil
			},
		}
		resourcePoolClient := &mockResourcePoolServiceClient{
			listResourcePoolsFn: func(ctx context.Context, in *nextgenv1.ListResourcePoolsRequest, opts ...grpc.CallOption) (*nextgenv1.ListResourcePoolsResponse, error) {
				return &nextgenv1.ListResourcePoolsResponse{
					ResourcePools: []*nextgenv1.ResourcePoolProfile{{}},
				}, nil
			},
		}

		resp, err := listClustersFromResourcePools(context.Background(), clusterClient, resourcePoolClient, []uint64{11, 22})
		assert.NoError(t, err)
		assert.NotNil(t, resp)
		assert.Empty(t, resp.Clusters)
	})

	t.Run("maps resource pool instances to cluster list request", func(t *testing.T) {
		var capturedRPReq *nextgenv1.ListResourcePoolsRequest
		var capturedClusterReq *tidbmgmt.ListClustersRequest
		clusterClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				capturedClusterReq = in
				return &tidbmgmt.ListClustersResponse{Clusters: []*tidbmgmt.Cluster{{ClusterId: 7001}}}, nil
			},
		}
		resourcePoolClient := &mockResourcePoolServiceClient{
			listResourcePoolsFn: func(ctx context.Context, in *nextgenv1.ListResourcePoolsRequest, opts ...grpc.CallOption) (*nextgenv1.ListResourcePoolsResponse, error) {
				capturedRPReq = in
				return &nextgenv1.ListResourcePoolsResponse{
					ResourcePools: []*nextgenv1.ResourcePoolProfile{
						{Instances: []*nextgenv1.ResourcePoolProfile_Instance{{ClusterId: 7001}, {ClusterId: 7002}, {ClusterId: 7001}}},
					},
				}, nil
			},
		}

		resp, err := listClustersFromResourcePools(context.Background(), clusterClient, resourcePoolClient, []uint64{9001, 9002})
		assert.NoError(t, err)
		assert.NotNil(t, resp)
		assert.Len(t, resp.Clusters, 1)

		assert.NotNil(t, capturedRPReq)
		assert.Equal(t, []uint64{9001, 9002}, capturedRPReq.ResourcePoolIds)
		assert.Equal(t, nextgenv1.ListResourcePoolsRequest_DETAIL, capturedRPReq.View)
		assert.False(t, capturedRPReq.ShowDeleted)

		assert.NotNil(t, capturedClusterReq)
		assert.NotNil(t, capturedClusterReq.Filter)
		assert.ElementsMatch(t, []uint64{7001, 7002}, capturedClusterReq.Filter.ClusterIds)
		assert.NotNil(t, capturedClusterReq.Filter.Deleted)
		assert.False(t, *capturedClusterReq.Filter.Deleted)
	})
}

// --- ListClustersWithMetadata handler ---

func Test_ListClustersWithMetadata(t *testing.T) {
	gin.SetMode(gin.TestMode)

	t.Run("missing env returns 400", func(t *testing.T) {
		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?tenantIds=100", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusBadRequest, w.Code)
		assert.Contains(t, w.Body.String(), "invalid env")
	})

	t.Run("invalid env returns 400", func(t *testing.T) {
		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?tenantIds=100&env=invalid", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusBadRequest, w.Code)
		assert.Contains(t, w.Body.String(), "invalid env")
	})

	t.Run("empty filters returns 400", func(t *testing.T) {
		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusBadRequest, w.Code)
		assert.Contains(t, w.Body.String(), "at least one")
	})

	t.Run("invalid tenantIds returns 400", func(t *testing.T) {
		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&tenantIds=abc", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusBadRequest, w.Code)
		assert.Contains(t, w.Body.String(), "invalid tenantIds")
	})

	t.Run("invalid projectIds returns 400", func(t *testing.T) {
		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&projectIds=xyz", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusBadRequest, w.Code)
		assert.Contains(t, w.Body.String(), "invalid projectIds")
	})

	t.Run("invalid clusterIds returns 400", func(t *testing.T) {
		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&clusterIds=-1", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusBadRequest, w.Code)
		assert.Contains(t, w.Body.String(), "invalid clusterIds")
	})

	t.Run("mgmt service client error returns 500", func(t *testing.T) {
		// Use an env that has no client in the map and no config
		// The TestMain meta config doesn't have envToMgmtSvcAddr for "staging"
		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=staging&tenantIds=100", nil)

		// Ensure no cached client for staging
		mgmtClients.Delete("staging")

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusInternalServerError, w.Code)
		assert.Contains(t, w.Body.String(), "no mgmt service address configured")
	})

	t.Run("mgmt service returns error", func(t *testing.T) {
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				return nil, fmt.Errorf("gRPC connection failed")
			},
		}
		mgmtClients.Store("dev", mockClient)
		defer mgmtClients.Delete("dev")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&tenantIds=100", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusInternalServerError, w.Code)
		assert.Contains(t, w.Body.String(), "failed to list clusters from mgmt service")
	})

	t.Run("mgmt service returns nil response", func(t *testing.T) {
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				return nil, nil
			},
		}
		mgmtClients.Store("dev", mockClient)
		defer mgmtClients.Delete("dev")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&tenantIds=100", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusOK, w.Code)

		var resp message.Result
		assert.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp))
		assert.Equal(t, 0, resp.Code)
	})

	t.Run("mgmt service returns empty clusters", func(t *testing.T) {
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				return &tidbmgmt.ListClustersResponse{Clusters: []*tidbmgmt.Cluster{}}, nil
			},
		}
		mockRPClient := &mockResourcePoolServiceClient{
			listResourcePoolsFn: func(ctx context.Context, in *nextgenv1.ListResourcePoolsRequest, opts ...grpc.CallOption) (*nextgenv1.ListResourcePoolsResponse, error) {
				return &nextgenv1.ListResourcePoolsResponse{ResourcePools: []*nextgenv1.ResourcePoolProfile{}}, nil
			},
		}
		mgmtClients.Store("dev", mockClient)
		mgmtResourcePoolClients.Store("dev", mockRPClient)
		defer mgmtClients.Delete("dev")
		defer mgmtResourcePoolClients.Delete("dev")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&clusterIds=100", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusOK, w.Code)
	})

	t.Run("fallback to resource pool service when cluster list is empty", func(t *testing.T) {
		callCount := 0
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				callCount++
				if callCount == 1 {
					return &tidbmgmt.ListClustersResponse{Clusters: []*tidbmgmt.Cluster{}}, nil
				}
				return &tidbmgmt.ListClustersResponse{
					Clusters: []*tidbmgmt.Cluster{{
						ClusterId:      100,
						OrganizationId: 12345,
						CloudProvider:  "aws",
						Region:         "us-west-2",
						DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED,
					}},
				}, nil
			},
		}
		mockRPClient := &mockResourcePoolServiceClient{
			listResourcePoolsFn: func(ctx context.Context, in *nextgenv1.ListResourcePoolsRequest, opts ...grpc.CallOption) (*nextgenv1.ListResourcePoolsResponse, error) {
				return &nextgenv1.ListResourcePoolsResponse{
					ResourcePools: []*nextgenv1.ResourcePoolProfile{{
						Instances: []*nextgenv1.ResourcePoolProfile_Instance{{ClusterId: 100}},
					}},
				}, nil
			},
		}
		mgmtClients.Store("dev", mockClient)
		mgmtResourcePoolClients.Store("dev", mockRPClient)
		defer mgmtClients.Delete("dev")
		defer mgmtResourcePoolClients.Delete("dev")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&clusterIds=9001", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusOK, w.Code)
		assert.Equal(t, 2, callCount)
	})

	t.Run("resource pool fallback error returns 500", func(t *testing.T) {
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				return &tidbmgmt.ListClustersResponse{Clusters: []*tidbmgmt.Cluster{}}, nil
			},
		}
		mockRPClient := &mockResourcePoolServiceClient{
			listResourcePoolsFn: func(ctx context.Context, in *nextgenv1.ListResourcePoolsRequest, opts ...grpc.CallOption) (*nextgenv1.ListResourcePoolsResponse, error) {
				return nil, fmt.Errorf("resource pool RPC failed")
			},
		}
		mgmtClients.Store("dev", mockClient)
		mgmtResourcePoolClients.Store("dev", mockRPClient)
		defer mgmtClients.Delete("dev")
		defer mgmtResourcePoolClients.Delete("dev")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&clusterIds=9001", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusInternalServerError, w.Code)
		assert.Contains(t, w.Body.String(), "failed to list clusters from resource pool service")
	})

	t.Run("mgmt service returns clusters with unknown deploy type", func(t *testing.T) {
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				return &tidbmgmt.ListClustersResponse{
					Clusters: []*tidbmgmt.Cluster{
						{
							ClusterId:      100,
							OrganizationId: 1,
							CloudProvider:  "aws",
							Region:         "us-west-2",
							DeployType:     tidbmgmt.DeployType(999),
						},
					},
				}, nil
			},
		}
		mgmtClients.Store("dev", mockClient)
		defer mgmtClients.Delete("dev")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&tenantIds=1", nil)

		ListClustersWithMetadata(c)

		// All clusters had unknown type, grouping produces empty groups → returns empty
		assert.Equal(t, http.StatusOK, w.Code)
	})

	t.Run("passes correct filters to mgmt service", func(t *testing.T) {
		var capturedReq *tidbmgmt.ListClustersRequest
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				capturedReq = in
				return &tidbmgmt.ListClustersResponse{Clusters: []*tidbmgmt.Cluster{{
					ClusterId:      40,
					OrganizationId: 12345,
					CloudProvider:  "aws",
					Region:         "us-west-2",
					DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED,
				}}}, nil
			},
		}
		mgmtClients.Store("dev", mockClient)
		defer mgmtClients.Delete("dev")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&tenantIds=10,20&projectIds=30&clusterIds=40,50,60", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusOK, w.Code)
		assert.NotNil(t, capturedReq)
		assert.NotNil(t, capturedReq.Filter)
		assert.Equal(t, []uint64{10, 20}, capturedReq.Filter.TenantIds)
		assert.Equal(t, []uint64{30}, capturedReq.Filter.ProjectIds)
		assert.Equal(t, []uint64{40, 50, 60}, capturedReq.Filter.ClusterIds)
		assert.NotNil(t, capturedReq.Filter.Deleted)
		assert.False(t, *capturedReq.Filter.Deleted)
	})

	t.Run("success with regional fan-out", func(t *testing.T) {
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				return &tidbmgmt.ListClustersResponse{
					Clusters: []*tidbmgmt.Cluster{
						{
							ClusterId:      100,
							OrganizationId: 12345,
							CloudProvider:  "aws",
							Region:         "us-west-2",
							DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED,
						},
					},
				}, nil
			},
		}
		mgmtClients.Store("dev", mockClient)
		defer mgmtClients.Delete("dev")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&tenantIds=12345", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusOK, w.Code)

		var resp message.Result
		assert.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp))
		assert.Equal(t, 0, resp.Code)
	})

	t.Run("env dev is valid", func(t *testing.T) {
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				return &tidbmgmt.ListClustersResponse{Clusters: []*tidbmgmt.Cluster{{
					ClusterId:      1,
					OrganizationId: 12345,
					CloudProvider:  "aws",
					Region:         "us-west-2",
					DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED,
				}}}, nil
			},
		}
		mgmtClients.Store("dev", mockClient)
		defer mgmtClients.Delete("dev")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&clusterIds=1", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusOK, w.Code)
	})

	t.Run("env staging is valid", func(t *testing.T) {
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				return &tidbmgmt.ListClustersResponse{Clusters: []*tidbmgmt.Cluster{{
					ClusterId:      1,
					OrganizationId: 12345,
					CloudProvider:  "aws",
					Region:         "us-west-2",
					DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED,
				}}}, nil
			},
		}
		mgmtClients.Store("staging", mockClient)
		defer mgmtClients.Delete("staging")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=staging&clusterIds=1", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusOK, w.Code)
	})

	t.Run("env prod is valid", func(t *testing.T) {
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				return &tidbmgmt.ListClustersResponse{Clusters: []*tidbmgmt.Cluster{{
					ClusterId:      1,
					OrganizationId: 12345,
					CloudProvider:  "aws",
					Region:         "us-west-2",
					DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED,
				}}}, nil
			},
		}
		mgmtClients.Store("prod", mockClient)
		defer mgmtClients.Delete("prod")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=prod&clusterIds=1", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusOK, w.Code)
	})
}

// --- showInfra validation ---

func Test_ListClustersWithMetadata_ShowInfra(t *testing.T) {
	gin.SetMode(gin.TestMode)

	t.Run("showInfra=true without clusterIds returns 400", func(t *testing.T) {
		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&tenantIds=100&showInfra=true", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusBadRequest, w.Code)
		assert.Contains(t, w.Body.String(), "showInfra requires clusterIds")
	})

	t.Run("showInfra=false returns same clusters without infraClusters", func(t *testing.T) {
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				return &tidbmgmt.ListClustersResponse{
					Clusters: []*tidbmgmt.Cluster{{
						ClusterId:      100,
						OrganizationId: 12345,
						CloudProvider:  "aws",
						Region:         "us-west-2",
						DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED,
					}},
				}, nil
			},
		}
		mgmtClients.Store("dev", mockClient)
		defer mgmtClients.Delete("dev")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&clusterIds=100", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusOK, w.Code)
		// Response must NOT contain infraClusters key (showInfra=false, omitempty applies)
		assert.NotContains(t, w.Body.String(), "infraClusters")
	})

	t.Run("showInfra=true with clusterIds passes validation", func(t *testing.T) {
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				return &tidbmgmt.ListClustersResponse{Clusters: []*tidbmgmt.Cluster{}}, nil
			},
		}
		mockRPClient := &mockResourcePoolServiceClient{
			listResourcePoolsFn: func(ctx context.Context, in *nextgenv1.ListResourcePoolsRequest, opts ...grpc.CallOption) (*nextgenv1.ListResourcePoolsResponse, error) {
				return &nextgenv1.ListResourcePoolsResponse{ResourcePools: []*nextgenv1.ResourcePoolProfile{}}, nil
			},
		}
		mgmtClients.Store("dev", mockClient)
		mgmtResourcePoolClients.Store("dev", mockRPClient)
		defer mgmtClients.Delete("dev")
		defer mgmtResourcePoolClients.Delete("dev")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&clusterIds=100&showInfra=true", nil)

		ListClustersWithMetadata(c)

		// Passes validation (empty result, but no 400)
		assert.NotEqual(t, http.StatusBadRequest, w.Code)
	})
}

// --- resolveInfraClusters ---

func Test_resolveInfraClusters(t *testing.T) {
	t.Run("serverless cluster returns nil", func(t *testing.T) {
		cli := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				t.Fatal("should not call ListClusters for serverless")
				return nil, nil
			},
		}
		mgmtCluster := &tidbmgmt.Cluster{
			ClusterId:      500,
			OrganizationId: 11,
			ProjectId:      22,
			CloudProvider:  "aws",
			Region:         "us-east-1",
			DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_SERVERLESS,
		}
		result := resolveInfraClusters(context.Background(), cli, mgmtCluster, "dev-devtier")
		assert.Nil(t, result)
	})

	t.Run("byoc cluster returns nil", func(t *testing.T) {
		cli := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				t.Fatal("should not call ListClusters for byoc")
				return nil, nil
			},
		}
		mgmtCluster := &tidbmgmt.Cluster{
			ClusterId:      600,
			OrganizationId: 11,
			ProjectId:      22,
			CloudProvider:  "aws",
			Region:         "us-east-1",
			DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
			ClusterInfo: &tidbmgmt.Cluster_NextGenCluster{
				NextGenCluster: &nextgenv1.Cluster{
					ServicePlan: commonv1.ServicePlan_BYOC,
				},
			},
		}
		result := resolveInfraClusters(context.Background(), cli, mgmtCluster, "byoc")
		assert.Nil(t, result)
	})

	t.Run("dedicated cluster returns direct k8s-infra entry", func(t *testing.T) {
		cli := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				t.Fatal("should not call ListClusters for dedicated infra lookup")
				return nil, nil
			},
		}
		mgmtCluster := &tidbmgmt.Cluster{
			ClusterId:      100,
			OrganizationId: 11,
			ProjectId:      22,
			CloudProvider:  "aws",
			Region:         "us-west-2",
			DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED,
		}
		result := resolveInfraClusters(context.Background(), cli, mgmtCluster, "tidbcloud")
		assert.Len(t, result, 1)
		assert.Equal(t, "k8s-infra", result[0].ApplicationID)
		assert.Equal(t, "tidbcloud", result[0].BizType)
		assert.Equal(t, "11", result[0].TenantID)
		assert.Equal(t, "22", result[0].ProjectID)
		assert.Equal(t, "aws", string(result[0].Vendor))
		assert.Equal(t, "us-west-2", result[0].Region)
	})

	t.Run("dedicated cluster ignores missing lookup data", func(t *testing.T) {
		cli := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				t.Fatal("should not call ListClusters for dedicated infra lookup")
				return nil, nil
			},
		}
		mgmtCluster := &tidbmgmt.Cluster{
			ClusterId:      100,
			OrganizationId: 11,
			ProjectId:      22,
			CloudProvider:  "aws",
			Region:         "us-west-2",
			DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED,
		}
		result := resolveInfraClusters(context.Background(), cli, mgmtCluster, "tidbcloud")
		assert.Len(t, result, 1)
		assert.Equal(t, "k8s-infra", result[0].ApplicationID)
	})

	t.Run("dedicated cluster ignores mgmt-svc error", func(t *testing.T) {
		cli := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				t.Fatal("should not call ListClusters for dedicated infra lookup")
				return nil, nil
			},
		}
		mgmtCluster := &tidbmgmt.Cluster{
			ClusterId:      100,
			OrganizationId: 11,
			ProjectId:      22,
			CloudProvider:  "aws",
			Region:         "us-west-2",
			DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED,
		}
		result := resolveInfraClusters(context.Background(), cli, mgmtCluster, "tidbcloud")
		assert.Len(t, result, 1)
		assert.Equal(t, "k8s-infra", result[0].ApplicationID)
	})

	t.Run("nextgen cluster with dataplaneId and sharedPoolId returns two infra entries", func(t *testing.T) {
		cli := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				t.Fatal("should not call ListClusters for nextgen infra lookup")
				return nil, nil
			},
		}
		mgmtCluster := &tidbmgmt.Cluster{
			ClusterId:      200,
			OrganizationId: 33,
			CloudProvider:  "aws",
			Region:         "ap-southeast-1",
			DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
			ClusterInfo: &tidbmgmt.Cluster_NextGenCluster{
				NextGenCluster: &nextgenv1.Cluster{
					DataplaneId:  "dp-abc123",
					SharedPoolId: "sp-xyz789",
				},
			},
		}
		result := resolveInfraClusters(context.Background(), cli, mgmtCluster, "prod-tidbcloud-nextgen")
		assert.Len(t, result, 2)

		appIDs := []string{result[0].ApplicationID, result[1].ApplicationID}
		assert.Contains(t, appIDs, "dp-abc123")
		assert.Contains(t, appIDs, "sp-xyz789")
		for _, r := range result {
			assert.Equal(t, "33", r.TenantID)
			assert.Equal(t, "aws", string(r.Vendor))
			assert.Equal(t, "ap-southeast-1", r.Region)
			assert.Equal(t, "prod-tidbcloud-nextgen", r.BizType)
		}
	})

	t.Run("nextgen cluster with only dataplaneId returns one infra entry", func(t *testing.T) {
		cli := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				t.Fatal("should not call ListClusters for nextgen")
				return nil, nil
			},
		}
		mgmtCluster := &tidbmgmt.Cluster{
			ClusterId:      300,
			OrganizationId: 44,
			CloudProvider:  "gcp",
			Region:         "us-central1",
			DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
			ClusterInfo: &tidbmgmt.Cluster_NextGenCluster{
				NextGenCluster: &nextgenv1.Cluster{
					DataplaneId: "dp-only",
					// SharedPoolId is empty
				},
			},
		}
		result := resolveInfraClusters(context.Background(), cli, mgmtCluster, "dev-tidbcloud-nextgen")
		assert.Len(t, result, 1)
		assert.Equal(t, "dp-only", result[0].ApplicationID)
	})

	t.Run("nextgen cluster with no dataplaneId or sharedPoolId returns nil", func(t *testing.T) {
		cli := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				t.Fatal("should not call ListClusters for empty nextgen")
				return nil, nil
			},
		}
		mgmtCluster := &tidbmgmt.Cluster{
			ClusterId:      400,
			OrganizationId: 55,
			DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
			ClusterInfo: &tidbmgmt.Cluster_NextGenCluster{
				NextGenCluster: &nextgenv1.Cluster{},
			},
		}
		result := resolveInfraClusters(context.Background(), cli, mgmtCluster, "dev-tidbcloud-nextgen")
		assert.Nil(t, result)
	})
}

// --- ListClustersWithMetadata showInfra handler integration ---

func Test_ListClustersWithMetadata_ShowInfraIntegration(t *testing.T) {
	gin.SetMode(gin.TestMode)

	t.Run("showInfra=true dedicated cluster returns 200 with valid response", func(t *testing.T) {
		// This test verifies the handler wiring: mgmt-svc is called once for the initial
		// cluster listing. The regional fan-out returns nothing in unit-test context, so
		// resolveInfraClusters is exercised separately in Test_resolveInfraClusters.
		callCount := 0
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				callCount++
				return &tidbmgmt.ListClustersResponse{
					Clusters: []*tidbmgmt.Cluster{{
						ClusterId:      100,
						OrganizationId: 12345,
						ProjectId:      67,
						CloudProvider:  "aws",
						Region:         "us-west-2",
						DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED,
					}},
				}, nil
			},
		}
		mgmtClients.Store("prod", mockClient)
		defer mgmtClients.Delete("prod")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=prod&clusterIds=100&showInfra=true", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusOK, w.Code)
		// Mgmt-svc called exactly once for the cluster listing
		assert.Equal(t, 1, callCount)
		// Response is valid JSON with the expected shape
		var result map[string]interface{}
		assert.NoError(t, json.Unmarshal(w.Body.Bytes(), &result))
	})

	t.Run("showInfra=true nextgen cluster populates DataplaneID and SharedPoolID without extra lookup", func(t *testing.T) {
		callCount := 0
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				callCount++
				return &tidbmgmt.ListClustersResponse{
					Clusters: []*tidbmgmt.Cluster{{
						ClusterId:      200,
						OrganizationId: 12345,
						CloudProvider:  "aws",
						Region:         "us-west-2",
						DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
						ClusterInfo: &tidbmgmt.Cluster_NextGenCluster{
							NextGenCluster: &nextgenv1.Cluster{
								DataplaneId:  "dp-test",
								SharedPoolId: "sp-test",
							},
						},
					}},
				}, nil
			},
		}
		mgmtClients.Store("dev", mockClient)
		defer mgmtClients.Delete("dev")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&clusterIds=200&showInfra=true", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusOK, w.Code)
		// Only the initial listing call; nextgen infra uses dataplaneId/sharedPoolId directly
		assert.Equal(t, 1, callCount)
	})

	t.Run("showInfra=true serverless cluster returns empty infraClusters", func(t *testing.T) {
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				return &tidbmgmt.ListClustersResponse{
					Clusters: []*tidbmgmt.Cluster{{
						ClusterId:      300,
						OrganizationId: 12345,
						CloudProvider:  "aws",
						Region:         "us-west-2",
						DeployType:     tidbmgmt.DeployType_DEPLOY_TYPE_SERVERLESS,
					}},
				}, nil
			},
		}
		mgmtClients.Store("dev", mockClient)
		defer mgmtClients.Delete("dev")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&clusterIds=300&showInfra=true", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusOK, w.Code)
		// serverless never gets infraClusters
		assert.NotContains(t, w.Body.String(), "infraClusters")
	})

	t.Run("response shape uses ListClustersWithMetadataResp (has partial, clusters keys)", func(t *testing.T) {
		mockClient := &mockClusterServiceClient{
			listClustersFn: func(ctx context.Context, in *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
				return &tidbmgmt.ListClustersResponse{Clusters: []*tidbmgmt.Cluster{}}, nil
			},
		}
		mockRPClient := &mockResourcePoolServiceClient{
			listResourcePoolsFn: func(ctx context.Context, in *nextgenv1.ListResourcePoolsRequest, opts ...grpc.CallOption) (*nextgenv1.ListResourcePoolsResponse, error) {
				return &nextgenv1.ListResourcePoolsResponse{}, nil
			},
		}
		mgmtClients.Store("dev", mockClient)
		mgmtResourcePoolClients.Store("dev", mockRPClient)
		defer mgmtClients.Delete("dev")
		defer mgmtResourcePoolClients.Delete("dev")

		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&clusterIds=1", nil)

		ListClustersWithMetadata(c)

		assert.Equal(t, http.StatusOK, w.Code)
		var result map[string]interface{}
		assert.NoError(t, json.Unmarshal(w.Body.Bytes(), &result))
		data, ok := result["data"].(map[string]interface{})
		assert.True(t, ok)
		_, hasClusters := data["clusters"]
		_, hasPartial := data["partial"]
		assert.True(t, hasClusters, "response data should contain clusters key")
		assert.True(t, hasPartial, "response data should contain partial key")
	})
}

// --- route registration ---

func Test_ListClustersWithMetadata_RouteRegistered(t *testing.T) {
	// Verify the route /api/v2/global/clusters is registered in testEngine
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodGet, "/api/v2/global/clusters?env=dev&clusterIds=1", nil)
	testEngine.ServeHTTP(w, req)

	// Should not be 404 (route exists), even though auth will fail
	assert.NotEqual(t, http.StatusNotFound, w.Code)
}