package globalapi

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"
	"sync"

	"github.com/ddliu/go-httpclient"
	"github.com/gin-gonic/gin"
	"github.com/pingcap/log"
	"github.com/tidbcloud/observability/pkg/common/awsauth"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/errors"
	"github.com/tidbcloud/observability/pkg/common/message"
	"github.com/tidbcloud/observability/pkg/common/structs"
	"github.com/tidbcloud/observability/pkg/globalservice"
	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"
	tigrpc "github.com/tidbcloud/tiinf/transport/tigrpc/client"
	"go.uber.org/zap"
)

var mgmtClients sync.Map
var mgmtResourcePoolClients sync.Map

type metadataRegionResult struct {
	resp       message.QueryClustersResp
	err        error
	primary    bool
	clusterIDs map[string]bool
}

type regionalQueryKey struct {
	Vendor   string
	Region   string
	BizType  string
	TenantID string
}

type clusterGroup struct {
	key        regionalQueryKey
	clusterIDs map[string]bool
}

type parsedFilters struct {
	TenantIds  []uint64
	ProjectIds []uint64
	ClusterIds []uint64
}

// resolveBizType maps a tidb-management-service DeployType (and optional ServicePlan) to the o11y
// bizType string, applying the environment prefix per the convention:
//
//	dedicated  -> tidbcloud          (prod has no prefix)
//	nextgen    -> tidbcloud-nextgen
//	serverless -> devtier
//	BYOC       -> byoc               (no env prefix; route table uses bare "byoc" prefix)
func resolveBizType(deployType tidbmgmt.DeployType, servicePlan commonv1.ServicePlan, env string) (string, error) {
	if servicePlan == commonv1.ServicePlan_BYOC {
		return "byoc", nil
	}
	var base string
	switch deployType {
	case tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED:
		base = "tidbcloud"
	case tidbmgmt.DeployType_DEPLOY_TYPE_NEXTGEN:
		base = "tidbcloud-nextgen"
	case tidbmgmt.DeployType_DEPLOY_TYPE_SERVERLESS:
		base = "devtier"
	default:
		return "", fmt.Errorf("unknown deploy type: %v", deployType)
	}
	if env == constants.EnvProd && base == "tidbcloud" {
		return base, nil
	}
	return fmt.Sprintf("%s-%s", env, base), nil
}

func getMgmtSvcClient(env string) (tidbmgmt.ClusterServiceClient, error) {
	if cli, ok := mgmtClients.Load(env); ok {
		return cli.(tidbmgmt.ClusterServiceClient), nil
	}

	config := globalservice.GetGlobalManager().GetGlobalMetaCache().ServerConfig
	addr, ok := config.EnvToMgmtSvcAddr[env]
	if !ok || addr == "" {
		return nil, fmt.Errorf("no mgmt service address configured for env %q", env)
	}

	conn, err := tigrpc.NewClientConn(addr, tigrpc.WithDefault())
	if err != nil {
		return nil, fmt.Errorf("failed to create gRPC connection to %s: %w", addr, err)
	}
	cli := tidbmgmt.NewClusterServiceClient(conn)
	mgmtClients.Store(env, cli)
	return cli, nil
}

func getResourcePoolSvcClient(env string) (tidbmgmt.ResourcePoolServiceClient, error) {
	if cli, ok := mgmtResourcePoolClients.Load(env); ok {
		return cli.(tidbmgmt.ResourcePoolServiceClient), nil
	}

	config := globalservice.GetGlobalManager().GetGlobalMetaCache().ServerConfig
	addr, ok := config.EnvToMgmtSvcAddr[env]
	if !ok || addr == "" {
		return nil, fmt.Errorf("no mgmt service address configured for env %q", env)
	}

	conn, err := tigrpc.NewClientConn(addr, tigrpc.WithDefault())
	if err != nil {
		return nil, fmt.Errorf("failed to create gRPC connection to %s: %w", addr, err)
	}
	cli := tidbmgmt.NewResourcePoolServiceClient(conn)
	mgmtResourcePoolClients.Store(env, cli)
	return cli, nil
}

func listClustersFromResourcePools(ctx context.Context, clusterCli tidbmgmt.ClusterServiceClient, resourcePoolCli tidbmgmt.ResourcePoolServiceClient, resourcePoolIDs []uint64) (*tidbmgmt.ListClustersResponse, error) {
	if len(resourcePoolIDs) == 0 {
		return &tidbmgmt.ListClustersResponse{Clusters: []*tidbmgmt.Cluster{}}, nil
	}

	rpResp, err := resourcePoolCli.ListResourcePools(ctx, &nextgenv1.ListResourcePoolsRequest{
		ResourcePoolIds: resourcePoolIDs,
		ShowDeleted:     false,
		View:            nextgenv1.ListResourcePoolsRequest_DETAIL,
		PageSize:        100,
	})
	if err != nil {
		return nil, err
	}

	clusterIDSet := make(map[uint64]struct{})
	for _, profile := range rpResp.GetResourcePools() {
		for _, instance := range profile.GetInstances() {
			if instance.GetClusterId() == 0 {
				continue
			}
			clusterIDSet[instance.GetClusterId()] = struct{}{}
		}
	}

	if len(clusterIDSet) == 0 {
		return &tidbmgmt.ListClustersResponse{Clusters: []*tidbmgmt.Cluster{}}, nil
	}

	clusterIDs := make([]uint64, 0, len(clusterIDSet))
	for id := range clusterIDSet {
		clusterIDs = append(clusterIDs, id)
	}

	deleted := false
	return clusterCli.ListClusters(ctx, &tidbmgmt.ListClustersRequest{
		Filter: &tidbmgmt.ClusterFilter{
			ClusterIds: clusterIDs,
			Deleted:    &deleted,
		},
		PageSize: 100,
	})
}

func parseUint64List(s string) ([]uint64, error) {
	if s == "" {
		return nil, nil
	}
	parts := strings.Split(s, ",")
	result := make([]uint64, 0, len(parts))
	for _, p := range parts {
		p = strings.TrimSpace(p)
		if p == "" {
			continue
		}
		v, err := strconv.ParseUint(p, 10, 64)
		if err != nil {
			return nil, fmt.Errorf("invalid uint64 value %q: %w", p, err)
		}
		result = append(result, v)
	}
	return result, nil
}

func parseAndValidateFilters(req message.ListClustersWithMetadataReq) (*parsedFilters, error) {
	tenantIds, err := parseUint64List(req.TenantIds)
	if err != nil {
		return nil, fmt.Errorf("invalid tenantIds: %w", err)
	}
	projectIds, err := parseUint64List(req.ProjectIds)
	if err != nil {
		return nil, fmt.Errorf("invalid projectIds: %w", err)
	}
	clusterIds, err := parseUint64List(req.ClusterIds)
	if err != nil {
		return nil, fmt.Errorf("invalid clusterIds: %w", err)
	}
	if len(tenantIds) == 0 && len(projectIds) == 0 && len(clusterIds) == 0 {
		return nil, fmt.Errorf("at least one of tenantIds, projectIds, or clusterIds must be provided")
	}
	return &parsedFilters{TenantIds: tenantIds, ProjectIds: projectIds, ClusterIds: clusterIds}, nil
}

func groupClustersByRoute(clusters []*tidbmgmt.Cluster, env string) map[regionalQueryKey]*clusterGroup {
	groups := make(map[regionalQueryKey]*clusterGroup)
	for _, cluster := range clusters {
		bizType, resolveErr := resolveBizType(cluster.DeployType, cluster.GetNextGenCluster().GetServicePlan(), env)
		if resolveErr != nil {
			log.Warn("skipping cluster with unknown deploy type",
				zap.Uint64("clusterId", cluster.ClusterId),
				zap.Int32("deployType", int32(cluster.DeployType)),
				zap.Error(resolveErr))
			continue
		}
		key := regionalQueryKey{
			Vendor:   cluster.CloudProvider,
			Region:   cluster.Region,
			BizType:  bizType,
			TenantID: fmt.Sprintf("%d", cluster.OrganizationId),
		}
		g, ok := groups[key]
		if !ok {
			g = &clusterGroup{key: key, clusterIDs: make(map[string]bool)}
			groups[key] = g
		}
		g.clusterIDs[fmt.Sprintf("%d", cluster.ClusterId)] = true
	}
	return groups
}

const maxPushdownClusters = 3

func fanOutToRegionalServers(c *gin.Context, groups map[regionalQueryKey]*clusterGroup) <-chan metadataRegionResult {
	resultCh := make(chan metadataRegionResult, estimateResultChannelSize(groups))
	wg := &sync.WaitGroup{}
	globalManager := globalservice.GetGlobalManager()

	for _, group := range groups {
		routeInfo, ok := resolveRoute(c, globalManager, group, resultCh)
		if !ok {
			continue
		}

		baseURL := fmt.Sprintf("%s/api/v1/clusters/?vendor=%s&region=%s&bizType=%s&tenantID=%s",
			routeInfo.DnsAddress, group.key.Vendor, group.key.Region, group.key.BizType, group.key.TenantID)

		if len(group.clusterIDs) <= maxPushdownClusters {
			dispatchPerCluster(c, wg, group, baseURL, routeInfo, resultCh)
		} else {
			dispatchTenantWide(c, wg, group, baseURL, routeInfo, resultCh)
		}
	}

	go func() {
		wg.Wait()
		close(resultCh)
	}()

	return resultCh
}

func estimateResultChannelSize(groups map[regionalQueryKey]*clusterGroup) int {
	n := 0
	for _, g := range groups {
		if len(g.clusterIDs) <= maxPushdownClusters {
			n += len(g.clusterIDs)
		} else {
			n++
		}
	}
	if n == 0 {
		n = 1
	}
	return n
}

func resolveRoute(c *gin.Context, globalManager *globalservice.GlobalManager, group *clusterGroup, resultCh chan<- metadataRegionResult) (*globalstructs.RegionalRouter, bool) {
	routingCtx := globalstructs.RoutingContext{
		Vendor:   constants.ProvidedVendor(group.key.Vendor),
		Region:   group.key.Region,
		BizType:  group.key.BizType,
		TenantID: group.key.TenantID,
	}
	routeInfo, routeErr := globalManager.GetRegionalRouteInfo(c, routingCtx)
	if routeErr != nil {
		log.Error("failed to get regional route",
			zap.String("vendor", group.key.Vendor),
			zap.String("region", group.key.Region),
			zap.String("bizType", group.key.BizType),
			zap.Error(routeErr))
		resultCh <- metadataRegionResult{err: routeErr, primary: true, clusterIDs: group.clusterIDs}
		return nil, false
	}
	if routeInfo.DnsAddress == "" {
		resultCh <- metadataRegionResult{
			err:        fmt.Errorf("empty DNS address for %s/%s/%s", group.key.Vendor, group.key.Region, group.key.BizType),
			primary:    routeInfo.Primary,
			clusterIDs: group.clusterIDs,
		}
		return nil, false
	}
	return routeInfo, true
}

// dispatchPerCluster sends one request per applicationID (pushdown).
// clusterIDs=nil on results signals no global-side filtering needed.
func dispatchPerCluster(c *gin.Context, wg *sync.WaitGroup, group *clusterGroup, baseURL string, routeInfo *globalstructs.RegionalRouter, resultCh chan<- metadataRegionResult) {
	for id := range group.clusterIDs {
		url := fmt.Sprintf("%s&applicationID=%s", baseURL, id)

		headers := map[string]string{}
		if routeInfo.AccessPrincipal != "" {
			var err error
			headers, err = signHeaders(c, routeInfo, group.key.Region, url)
			if err != nil {
				log.Error("failed to generate SigV4 headers", zap.String("url", url), zap.Error(err))
				resultCh <- metadataRegionResult{err: err, primary: routeInfo.Primary}
				continue
			}
		}

		log.Debug("querying regional server (pushdown)",
			zap.String("url", url),
			zap.String("applicationID", id))

		u := url
		h := headers
		wg.Add(1)
		go func() {
			defer wg.Done()
			fetchRegionalClusters(u, h, nil, routeInfo.Primary, resultCh)
		}()
	}
}

func dispatchTenantWide(c *gin.Context, wg *sync.WaitGroup, group *clusterGroup, baseURL string, routeInfo *globalstructs.RegionalRouter, resultCh chan<- metadataRegionResult) {
	headers := map[string]string{}
	if routeInfo.AccessPrincipal != "" {
		var err error
		headers, err = signHeaders(c, routeInfo, group.key.Region, baseURL)
		if err != nil {
			log.Error("failed to generate SigV4 headers", zap.String("url", baseURL), zap.Error(err))
			resultCh <- metadataRegionResult{err: err, primary: routeInfo.Primary, clusterIDs: group.clusterIDs}
			return
		}
	}

	log.Debug("querying regional server (tenant-wide)",
		zap.String("url", baseURL),
		zap.String("vendor", group.key.Vendor),
		zap.String("region", group.key.Region),
		zap.String("bizType", group.key.BizType))

	wg.Add(1)
	go func() {
		defer wg.Done()
		fetchRegionalClusters(baseURL, headers, group.clusterIDs, routeInfo.Primary, resultCh)
	}()
}

func signHeaders(c *gin.Context, routeInfo *globalstructs.RegionalRouter, region, url string) (map[string]string, error) {
	return awsauth.GetSigV4Headers(c.Request.Context(), awsauth.SigV4AuthInfo{
		RoleArn:    routeInfo.AccessPrincipal,
		ExternalID: routeInfo.AccessExternalID,
		Region:     region,
	}, http.MethodGet, url, nil)
}

func aggregateRegionalResults(resultCh <-chan metadataRegionResult) ([]structs.ClusterBaseInfo, bool) {
	allClusters := make([]structs.ClusterBaseInfo, 0)
	partial := false
	for result := range resultCh {
		if result.err != nil {
			partial = true
			if result.primary {
				log.Error("primary region query failed", zap.Error(result.err))
			} else {
				log.Warn("non-primary region query failed", zap.Error(result.err))
			}
			continue
		}
		// clusterIDs == nil means the regional server already filtered (pushdown);
		// include all returned clusters directly.
		if result.clusterIDs == nil {
			allClusters = append(allClusters, result.resp.Clusters...)
			continue
		}
		for _, cluster := range result.resp.Clusters {
			if result.clusterIDs[cluster.ApplicationID] {
				allClusters = append(allClusters, cluster)
			}
		}
	}
	return allClusters, partial
}

// resolveInfraClusters returns the infra ClusterBaseInfo entries for a given
// mgmt-svc cluster based on its deploy type.
//
// Dedicated: a cluster named "k8s-infra" in the same tenant+project
//
// NextGen (including BYOC service plan): builds two infra entries from the
// cluster's DataplaneId and SharedPoolId without an additional mgmt-svc round-trip.
//
// All other types (serverless, unknown): returns nil.
func resolveInfraClusters(ctx context.Context, cli tidbmgmt.ClusterServiceClient, mgmtCluster *tidbmgmt.Cluster, bizType string) []structs.ClusterBaseInfo {
	switch mgmtCluster.DeployType {
	case tidbmgmt.DeployType_DEPLOY_TYPE_DEDICATED:
		return []structs.ClusterBaseInfo{
			{
				CommonReqInfo: structs.CommonReqInfo{
					Vendor: constants.ProvidedVendor(mgmtCluster.CloudProvider),
					Region: mgmtCluster.Region,
				},
				TenantID:      fmt.Sprintf("%d", mgmtCluster.OrganizationId),
				ProjectID:     fmt.Sprintf("%d", mgmtCluster.ProjectId),
				ApplicationID: "k8s-infra",
				BizType:       bizType,
			},
		}

	case tidbmgmt.DeployType_DEPLOY_TYPE_NEXTGEN:
		dataplaneID := mgmtCluster.GetNextGenCluster().GetDataplaneId()
		sharedPoolID := mgmtCluster.GetNextGenCluster().GetSharedPoolId()
		var result []structs.ClusterBaseInfo
		if dataplaneID != "" {
			result = append(result, structs.ClusterBaseInfo{
				CommonReqInfo: structs.CommonReqInfo{
					Vendor: constants.ProvidedVendor(mgmtCluster.CloudProvider),
					Region: mgmtCluster.Region,
				},
				TenantID:      fmt.Sprintf("%d", mgmtCluster.OrganizationId),
				ApplicationID: dataplaneID,
				BizType:       bizType,
			})
		}
		if sharedPoolID != "" {
			result = append(result, structs.ClusterBaseInfo{
				CommonReqInfo: structs.CommonReqInfo{
					Vendor: constants.ProvidedVendor(mgmtCluster.CloudProvider),
					Region: mgmtCluster.Region,
				},
				TenantID:      fmt.Sprintf("%d", mgmtCluster.OrganizationId),
				ApplicationID: sharedPoolID,
				BizType:       bizType,
			})
		}
		return result

	default:
		return nil
	}
}

// ListClustersWithMetadata handles GET /api/v2/global/clusters.
// @Summary list clusters with metadata
// @Description list clusters by querying tidb-management-service and o11y regional servers. At least one of tenantIds, projectIds, or clusterIds must be provided. When showInfra=true, clusterIds is required and each cluster row includes an infraClusters array.
// @Tags cluster
// @Accept json
// @Produce json
// @Param tenantIds query string false "comma-separated tenant IDs; at least one of tenantIds/projectIds/clusterIds is required"
// @Param projectIds query string false "comma-separated project IDs; at least one of tenantIds/projectIds/clusterIds is required"
// @Param clusterIds query string false "comma-separated cluster IDs; at least one of tenantIds/projectIds/clusterIds is required"
// @Param env query string true "environment: dev, staging, prod"
// @Param showInfra query bool false "when true, each cluster row includes an infraClusters array; requires clusterIds to be set"
// @Success 200 {object} message.Result{data=message.ListClustersWithMetadataResp}
// @Failure 400 {object} message.Result
// @Failure 500 {object} message.Result
// @Router /api/v2/global/clusters [get]
func ListClustersWithMetadata(c *gin.Context) {
	var req message.ListClustersWithMetadataReq
	if err := c.ShouldBindQuery(&req); err != nil {
		c.JSON(errors.ParameterInvalid.GetHttpCode(), &message.Result{Code: int(errors.ParameterInvalid), Message: err.Error()})
		return
	}

	switch req.Env {
	case constants.EnvDev, constants.EnvStaging, constants.EnvProd:
	default:
		c.JSON(errors.ParameterInvalid.GetHttpCode(), &message.Result{
			Code: int(errors.ParameterInvalid), Message: fmt.Sprintf("invalid env %q, must be one of: dev, staging, prod", req.Env),
		})
		return
	}

	if req.ShowInfra && strings.TrimSpace(req.ClusterIds) == "" {
		c.JSON(errors.ParameterInvalid.GetHttpCode(), &message.Result{
			Code: int(errors.ParameterInvalid), Message: "showInfra requires clusterIds to be specified",
		})
		return
	}

	filters, err := parseAndValidateFilters(req)
	if err != nil {
		c.JSON(errors.ParameterInvalid.GetHttpCode(), &message.Result{Code: int(errors.ParameterInvalid), Message: err.Error()})
		return
	}

	cli, err := getMgmtSvcClient(req.Env)
	if err != nil {
		log.Error("failed to get mgmt service client", zap.String("env", req.Env), zap.Error(err))
		c.JSON(errors.InternalServerError.GetHttpCode(), &message.Result{Code: int(errors.InternalServerError), Message: err.Error()})
		return
	}

	deleted := false
	mgmtResp, err := cli.ListClusters(c.Request.Context(), &tidbmgmt.ListClustersRequest{
		Filter: &tidbmgmt.ClusterFilter{
			TenantIds: filters.TenantIds, ProjectIds: filters.ProjectIds,
			ClusterIds: filters.ClusterIds, Deleted: &deleted,
		},
		PageSize: 100,
	})
	if err != nil {
		log.Error("failed to list clusters from mgmt service", zap.Error(err))
		c.JSON(errors.InternalServerError.GetHttpCode(), &message.Result{
			Code: int(errors.InternalServerError), Message: fmt.Sprintf("failed to list clusters from mgmt service: %s", err.Error()),
		})
		return
	}

	emptyResp := message.Result{Data: &message.ListClustersWithMetadataResp{Clusters: []message.ClusterWithInfra{}}}
	if (mgmtResp == nil || len(mgmtResp.Clusters) == 0) && len(filters.ClusterIds) > 0 {
		resourcePoolCli, resourcePoolErr := getResourcePoolSvcClient(req.Env)
		if resourcePoolErr != nil {
			log.Error("failed to get resource pool service client", zap.String("env", req.Env), zap.Error(resourcePoolErr))
			c.JSON(errors.InternalServerError.GetHttpCode(), &message.Result{Code: int(errors.InternalServerError), Message: resourcePoolErr.Error()})
			return
		}

		mgmtResp, err = listClustersFromResourcePools(c.Request.Context(), cli, resourcePoolCli, filters.ClusterIds)
		if err != nil {
			log.Error("failed to list clusters from resource pool service", zap.Error(err))
			c.JSON(errors.InternalServerError.GetHttpCode(), &message.Result{
				Code: int(errors.InternalServerError), Message: fmt.Sprintf("failed to list clusters from resource pool service: %s", err.Error()),
			})
			return
		}
	}

	if mgmtResp == nil || len(mgmtResp.Clusters) == 0 {
		c.JSON(http.StatusOK, emptyResp)
		return
	}

	log.Info("received clusters from mgmt service", zap.Int("count", len(mgmtResp.Clusters)))

	// Build a lookup from applicationId (string of ClusterId) to mgmt-svc cluster for
	// enriching the o11y response with nextgen-specific IDs and infra clusters.
	mgmtClusterByAppID := make(map[string]*tidbmgmt.Cluster, len(mgmtResp.Clusters))
	for _, mc := range mgmtResp.Clusters {
		mgmtClusterByAppID[fmt.Sprintf("%d", mc.ClusterId)] = mc
	}

	groups := groupClustersByRoute(mgmtResp.Clusters, req.Env)
	if len(groups) == 0 {
		c.JSON(http.StatusOK, emptyResp)
		return
	}

	resultCh := fanOutToRegionalServers(c, groups)
	allClusters, partial := aggregateRegionalResults(resultCh)

	rows := make([]message.ClusterWithInfra, 0, len(allClusters))
	for _, base := range allClusters {
		row := message.ClusterWithInfra{ClusterBaseInfo: base}
		mc, hasMgmt := mgmtClusterByAppID[base.ApplicationID]
		if hasMgmt {
			ngCluster := mc.GetNextGenCluster()
			if ngCluster != nil {
				row.DataplaneID = ngCluster.GetDataplaneId()
				row.SharedPoolID = ngCluster.GetSharedPoolId()
			}
			if req.ShowInfra {
				bizType, resolveErr := resolveBizType(mc.DeployType, mc.GetNextGenCluster().GetServicePlan(), req.Env)
				if resolveErr == nil {
					row.InfraClusters = resolveInfraClusters(c.Request.Context(), cli, mc, bizType)
				}
			}
		}
		rows = append(rows, row)
	}

	c.JSON(http.StatusOK, message.Result{
		Data: &message.ListClustersWithMetadataResp{Clusters: rows, Partial: partial},
	})
}

func fetchRegionalClusters(url string, headers map[string]string, clusterIDs map[string]bool, primary bool, resultCh chan<- metadataRegionResult) {
	httpResp, err := httpclient.Begin().WithHeaders(headers).Get(url)
	if err != nil {
		resultCh <- metadataRegionResult{err: err, primary: primary, clusterIDs: clusterIDs}
		return
	}
	defer httpResp.Body.Close()

	body, err := io.ReadAll(httpResp.Body)
	if err != nil {
		resultCh <- metadataRegionResult{err: errors.New(errors.IOReadFileError, err.Error()), primary: primary, clusterIDs: clusterIDs}
		return
	}

	if httpResp.StatusCode != http.StatusOK {
		resultCh <- metadataRegionResult{
			err:     errors.NewErrorf(errors.UnrecognizedError, "regional server returned %d: %s", httpResp.StatusCode, string(body)),
			primary: primary, clusterIDs: clusterIDs,
		}
		return
	}

	var result UnmarshalResult
	if err = json.Unmarshal(body, &result); err != nil {
		resultCh <- metadataRegionResult{err: errors.New(errors.UnmarshalError, err.Error()), primary: primary, clusterIDs: clusterIDs}
		return
	}

	resultCh <- metadataRegionResult{resp: result.Data, primary: primary, clusterIDs: clusterIDs}
}