package cluster

import (
	"context"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/go-logr/logr"
	"github.com/tidbcloud/account/idl/pbgen/proto/account"
	userapiv1 "github.com/tidbcloud/aisupport/packages/sdk-go/contracts/api/core-server/user/v1"
	clusterng "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	ccproto "github.com/tidbcloud/config-center/proto/config_center"
	commonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	nextgenv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	v1beta11 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	mgmtv1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/dig"

	"github.com/tidbcloud/devops/api"
	clusterapi "github.com/tidbcloud/devops/api/cluster"
	tntent "github.com/tidbcloud/devops/entity/tenant"
	ctrl "github.com/tidbcloud/devops/pkg/controller"
	"github.com/tidbcloud/devops/pkg/infra/clients/accountcli"
	"github.com/tidbcloud/devops/pkg/infra/clients/coresvccli"
	"github.com/tidbcloud/devops/pkg/infra/clients/dataflowcli"
	"github.com/tidbcloud/devops/pkg/infra/clients/mgmthttpcli"
	"github.com/tidbcloud/devops/pkg/infra/clients/serverlesscli"
	"github.com/tidbcloud/devops/pkg/infra/consts"
	"github.com/tidbcloud/devops/pkg/infra/errs"
	"github.com/tidbcloud/devops/pkg/infra/utils/container"
	"github.com/tidbcloud/devops/pkg/service/clusterctl"
	cmxapi "github.com/tidbcloud/devops/pkg/service/clustermaintenance/api"
	jirasvcapi "github.com/tidbcloud/devops/pkg/service/jira/api"
	"github.com/tidbcloud/devops/pkg/service/k8sresource"
	"github.com/tidbcloud/devops/pkg/service/k8supgrade/infrainfo"
	taskapi "github.com/tidbcloud/devops/pkg/service/k8supgrade/task/api"
	"github.com/tidbcloud/devops/pkg/service/operation"
	"github.com/tidbcloud/devops/pkg/service/tenant"
)

type ControllerParam struct {
	dig.In

	ClusterCtlService clusterctl.ClusterCtlServiceAPI
	MaintenanceSvc    cmxapi.ServiceAPI
	CoreSvcCli        *coresvccli.CoreSvcCli
	MgmtHTTP          mgmthttpcli.MgmtHTTPClient
	AccountCli        accountcli.Client
	ServerlessCli     *serverlesscli.Client
	JiraService       jirasvcapi.JiraServiceAPI
	InfraInfoSvc      infrainfo.ServiceAPI
	K8sUpgradeTaskSvc taskapi.API
	TenantService     tenant.TenantServiceAPI
	OperationSvc      operation.OperationServiceAPI
	DataflowCli       *dataflowcli.DataflowCli
	K8sResourceSvc    k8sresource.K8SResourceServiceAPI
	Logger            logr.Logger
}

// ClusterController provide the api to do some operations for tidb cluster
type ClusterController struct {
	ctlSvc            clusterctl.ClusterCtlServiceAPI
	maintenanceSvc    cmxapi.ServiceAPI
	coreSvcCli        *coresvccli.CoreSvcCli
	mgmtHTTP          mgmthttpcli.MgmtHTTPClient
	accountCli        accountcli.Client
	serverlessCli     *serverlesscli.Client
	jiraSvc           jirasvcapi.JiraServiceAPI
	InfraInfoSvc      infrainfo.ServiceAPI
	k8sUpgradeTaskSvc taskapi.API
	tenantSvc         tenant.TenantServiceAPI
	operationSvc      operation.OperationServiceAPI
	dataflowCli       *dataflowcli.DataflowCli
	k8sResourceSvc    k8sresource.K8SResourceServiceAPI
	logger            logr.Logger
}

func (c *ClusterController) Name() string {
	return "ClusterController"
}

func NewClusterController(p ControllerParam) ctrl.Controller {
	return &ClusterController{
		ctlSvc:            p.ClusterCtlService,
		maintenanceSvc:    p.MaintenanceSvc,
		coreSvcCli:        p.CoreSvcCli,
		mgmtHTTP:          p.MgmtHTTP,
		accountCli:        p.AccountCli,
		serverlessCli:     p.ServerlessCli,
		jiraSvc:           p.JiraService,
		InfraInfoSvc:      p.InfraInfoSvc,
		k8sUpgradeTaskSvc: p.K8sUpgradeTaskSvc,
		tenantSvc:         p.TenantService,
		operationSvc:      p.OperationSvc,
		dataflowCli:       p.DataflowCli,
		k8sResourceSvc:    p.K8sResourceSvc,
		logger:            p.Logger.WithName("ClusterController"),
	}
}

func (c *ClusterController) RegisterAPI(suite *ctrl.Suite) error {
	r := suite.Router

	// ListClusters does not have a :cluster_id path param, so it can't go
	// through ClusterScopeMiddleware. Tenant scope is enforced inside the
	// handler instead, by filtering req.TenantIDs against the API key's
	// allowed tenants.
	r.GET("/clusters", c.ListClusters)

	clusters := r.Group("/clusters")
	clusters.Use(ctrl.ClusterScopeMiddleware(c.coreSvcCli))
	{
		clusters.GET("/:cluster_id", c.GetCluster)
		clusters.GET("/:cluster_id/topology", c.GetTopology)
		clusters.GET("/:cluster_id/components", c.GetClusterComponents)

		clusters.POST("/:cluster_id/ctl", c.ExecCtl)
		clusters.GET("/:cluster_id/alerts", c.ListAlerts)
		clusters.GET("/:cluster_id/alert-summary", c.GetAlertSummary)

		clusters.GET("/:cluster_id/dm-ns", c.GetDMNamespace)

		clusters.GET("/:cluster_id/controlplane", c.GetControlplaneK8SResource)
		clusters.GET("/:cluster_id/cluster-resources", c.ListClusterResources)
		clusters.GET("/k8s-resources", c.ListK8SResources)

		clusters.GET("/:cluster_id/iam-info", c.GetCloudIAMInfo)

		clusters.POST("/:cluster_id/migration/create-migration-cluster", c.CreateMigrationCluster)
		clusters.POST("/:cluster_id/migration/migrate-privatelink", c.MigratePrivateLink)

		clusters.GET("/:cluster_id/vpc-peerings", c.ListVPCPeerings)
		clusters.POST("/:cluster_id/vpc-peerings", c.CreateVPCPeering)
		clusters.DELETE("/:cluster_id/vpc-peerings/:vpc_peering_id", c.DeleteVPCPeering)

		clusters.POST("/:cluster_id/hetero-deploy/apply", c.ApplyHeterogeneousDeploy)
		clusters.GET("/:cluster_id/hetero-deploy/query", c.GetHeterogeneousDeploy)
		clusters.POST("/:cluster_id/hetero-deploy/delete", c.DeleteHeterogeneousDeploy)

	}

	group := r.Group("/cluster")
	{
		group.GET("/migration/dedicate2premium/prechecks", c.ListDedicatedToPremiumPrechecks)
		group.POST("/migration/dedicate2premium/prechecks", c.CreateDedicatedToPremiumPrecheck)
		group.GET("/migration/dedicate2premium/prechecks/:precheck_id", c.GetDedicatedToPremiumPrecheck)

		// TODO: replace GetCluster and GetSharedResourcePool
		group.GET("/resource-pools", c.ListResourcePools)
		group.GET("/resource-pools/:resource_pool_id", c.GetResourcePool)

		group.GET("/shared-resource-pools", c.ListSharedResourcePools)
		group.GET("/shared-resource-pools/:resource_pool_id", c.GetSharedResourcePool)

		group.GET("/maintenances", c.ListMaintenances)
		group.GET("/maintenances/:maintenance_id", c.GetMaintenance)
		group.POST("/maintenances", c.CreateMaintenance)
		group.POST("/maintenances/:maintenance_id/close", c.CloseMaintenance)
		group.POST("/maintenances/:maintenance_id/executions/:execution_id/start", c.StartMaintenanceExecution)
		group.POST("/maintenances/:maintenance_id/executions/:execution_id/finish", c.FinishMaintenanceExecution)
		group.POST("/maintenances/:maintenance_id/executions/:execution_id/cancel", c.CancelMaintenanceExecution)

		group.GET("/ops-maintenance-windows", c.ListClusterOpsMaintenanceWindows)
		group.GET("/ops-maintenance-windows/:cluster_id", c.GetClusterOpsMaintenanceWindow)
		group.POST("/ops-maintenance-windows", c.CreateClusterOpsMaintenanceWindow)
		group.PUT("/ops-maintenance-windows/:cluster_id", c.UpdateClusterOpsMaintenanceWindow)
		group.DELETE("/ops-maintenance-windows/:cluster_id", c.DeleteClusterOpsMaintenanceWindow)
		group.GET("/ops-maintenance-windows/:cluster_id/check", c.IsInClusterOpsMaintenanceWindow)

		group.POST("/dedicatedK8sResources", c.ListDedicatedK8sResources)

		// For k8s upgrade
		group.GET("/k8sInfos", c.ListK8SInformation)
		group.GET("/k8sInfos/:cloud_provider/:region/:k8s_name", c.GetK8SInformation)

		group.GET("/k8sUpgradeTasks", c.ListK8sUpgradeTasks)
		group.GET("/k8sUpgradeTasks/:task_id", c.GetK8sUpgradeTask)
		group.PUT("/k8sUpgradeTasks/:task_id", c.UpdateK8sUpgradeTask)
		group.POST("/k8sUpgradeTasks", c.CreateK8sUpgradeTask)
		group.POST("/k8sUpgradeTasks/:task_id/submit", c.SubmitK8sUpgradeTask)
		group.POST("/k8sUpgradeTasks/:task_id/close", c.CloseK8sUpgradeTask)
		group.GET("/k8sUpgradeTasks/:task_id/:k8s_name", c.GetK8sUpgradeTaskCluster)
		group.PUT("/k8sUpgradeTasks/:task_id/:k8s_name", c.UpdateK8sUpgradeTaskCluster)
		group.PUT("/k8sUpgradeTasks/:task_id/:k8s_name/autoGeneration", c.UpdateK8sUpgradeTaskClusterAutoGeneration)
		group.PUT("k8sUpgradeTasks/:task_id/:k8s_name/exclude", c.ExcludeK8sUpgradeTaskCluster)
		group.PUT("k8sUpgradeTasks/:task_id/:k8s_name/cancel", c.CancelUpgradeK8sCluster)
		group.DELETE("k8sUpgradeTasks/:task_id", c.DeleteK8sUpgradeTask)

		group.GET("/k8sUpgradeSupportedVersions", c.ListK8sUpgradeSupportedVersions)
		group.GET("/k8sUpgradeAdmin", c.GetK8sUpgradeAdmin)
		group.GET("/k8sUpgradeTasks/:task_id/notice", c.GetK8sUpgradeTaskNotice)
		group.PUT("/k8sUpgradeTasks/:task_id/notice", c.UpdateK8sUpgradeTaskNotice)
		group.POST("/k8sUpgradeTasks/:task_id/notify", c.NotifyK8sUpgradeTask)

		group.GET("/traffic/replays", c.ListTrafficReplays)
		group.POST("/traffic/replays", c.CreateTrafficReplay)
		group.GET("/traffic/replays/:replay_id", c.GetTrafficReplay)
		group.PUT("/traffic/replays/:replay_id", c.UpdateTrafficReplay)
		group.DELETE("/traffic/replays/:replay_id", c.DeleteTrafficReplay)
		group.POST("/traffic/replays/:replay_id/start", c.StartTrafficReplay)
		group.POST("/traffic/replays/:replay_id/stop", c.StopTrafficReplay)
	}

	return nil
}

// applyTenantScopeToListReq enforces per-role tenant restrictions on a
// ListClusters query. Restricted API keys must specify tenant_id(s)
// explicitly, and every one must pass CheckTenantScope.
func applyTenantScopeToListReq(ctx *gin.Context, req *clusterapi.ListClustersReq) error {
	if len(req.TenantIDs) == 0 {
		if ctrl.IsTenantRestricted(ctx) {
			return fmt.Errorf("tenant_id is required for this api key")
		}
		return nil
	}
	for _, s := range req.TenantIDs {
		id, err := strconv.ParseUint(s, 10, 64)
		if err != nil {
			return fmt.Errorf("invalid tenant_id %q: %w", s, err)
		}
		if err := ctrl.CheckTenantScope(ctx, id); err != nil {
			return err
		}
	}
	return nil
}

// ListClusters
//
//	@Id				ListClusters
//	@Description	list cluster
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			req	query		clusterapi.ListClustersReq	true	"ListClustersReq"
//	@Success		200	{object}	clusterapi.ListClustersResp
//	@Failure		500	{object}	api.BaseResp
//	@Router			/api/v1/devops/clusters [get]
func (c *ClusterController) ListClusters(ctx *gin.Context) {
	req := &clusterapi.ListClustersReq{}
	if err := ctx.ShouldBindQuery(req); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	if err := applyTenantScopeToListReq(ctx, req); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodePermissionDenied)
		return
	}

	resp, err := c.ListClustersCore(ctx.Request.Context(), req)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	ctrl.Success(ctx, resp)
}

func (c *ClusterController) ListClustersCore(
	ctx context.Context,
	req *clusterapi.ListClustersReq,
) (resp *clusterapi.ListClustersResp, err error) {
	dt := mgmtv1.DeployType(mgmtv1.DeployType_value[req.DeployType])

	param := coresvccli.ListClustersParam{
		Page:       int64(req.Page),
		PerPage:    int64(req.PerPage),
		DeployType: &dt,
	}

	param.TenantIDs, err = parseIDs(req.TenantIDs)
	if err != nil {
		return nil, err
	}
	param.ProjectIDs, err = parseIDs(req.ProjectIDs)
	if err != nil {
		return nil, err
	}
	param.ClusterIDs, err = parseIDs(req.ClusterIDs)
	if err != nil {
		return nil, err
	}

	if len(req.ServicePlan) > 0 {
		sp, ok := v1beta11.ServicePlan_value[req.ServicePlan]
		if !ok {
			return nil, fmt.Errorf("invalid service plan: %s", req.ServicePlan)
		}
		servicePlan := v1beta11.ServicePlan(sp)
		param.ServicePlan = &servicePlan
	}

	if req.IsResourcePool {
		return c.listTenantResourcePoolsCore(ctx, param)
	}

	return c.listClustersFromMgmtCore(ctx, param)
}

func (c *ClusterController) logListClustersTiming(stage string, start time.Time, kv ...any) {
	fields := []any{
		"stage", stage,
		"elapsed_ms", time.Since(start).Milliseconds(),
	}
	fields = append(fields, kv...)
	c.logger.Info("list clusters timing", fields...)
}

func listClustersParamLogFields(param coresvccli.ListClustersParam) []any {
	fields := []any{
		"page", param.Page,
		"per_page", param.PerPage,
		"tenant_filter_count", len(param.TenantIDs),
		"project_filter_count", len(param.ProjectIDs),
		"cluster_filter_count", len(param.ClusterIDs),
	}
	if param.DeployType != nil {
		fields = append(fields, "deploy_type", param.DeployType.String())
	}
	if param.ServicePlan != nil {
		fields = append(fields, "service_plan", param.ServicePlan.String())
	}
	return fields
}

func (c *ClusterController) listClustersFromMgmtCore(
	ctx context.Context,
	param coresvccli.ListClustersParam,
) (resp *clusterapi.ListClustersResp, err error) {
	stageStart := time.Now()
	clusters, total, err := c.coreSvcCli.Mgmt.ListClusters(ctx, param)
	if err != nil {
		c.logListClustersTiming("mgmt_list_clusters", stageStart, append(listClustersParamLogFields(param), "error", err.Error())...)
		return nil, err
	}
	c.logListClustersTiming("mgmt_list_clusters", stageStart, append(listClustersParamLogFields(param), "cluster_count", len(clusters), "total", total)...)

	if len(clusters) == 0 {
		return &clusterapi.ListClustersResp{
			Clusters: []clusterapi.Cluster{},
			Total:    0,
		}, nil
	}

	tenantIDSet := container.NewSet[uint64]()
	for _, cluster := range clusters {
		tenantIDSet.Insert(cluster.GetOrganizationId())
	}

	projectFilter := &account.ProjectFilter{
		OrgIds: tenantIDSet.Slice(),
		Status: []string{"active"},
	}

	projects, err := c.accountCli.ListProjects(ctx, projectFilter)
	if err != nil {
		return nil, err
	}
	tenantNames, err := c.tenantNamesByID(ctx, tenantIDSet.Slice())
	if err != nil {
		return nil, err
	}

	id2project := make(map[uint64]*account.Project)
	for _, p := range projects {
		id2project[p.Id] = p
	}

	clusterList := make([]clusterapi.Cluster, 0, len(clusters))

	for _, cluster := range clusters {
		project, ok := id2project[cluster.ProjectId]

		if !ok && cluster.DeployType != mgmtv1.DeployType_DEPLOY_TYPE_NEXTGEN {
			return nil, fmt.Errorf("project %d not found", cluster.ProjectId)
		}

		projectName := ""
		if project != nil {
			projectName = project.GetName()
		}

		apiCluster, err := clusterapi.BuildClusterFromMgmt(cluster, projectName)
		if err != nil {
			return nil, err
		}
		apiCluster.TenantName = tenantNames[apiCluster.TenantID]

		clusterList = append(clusterList, apiCluster)
	}
	clusterList = c.injectTiDBTagsForList(ctx, clusterList, clusters)

	return &clusterapi.ListClustersResp{
		Clusters: clusterList,
		Total:    total,
	}, nil
}

func (c *ClusterController) listTenantResourcePoolsCore(
	ctx context.Context,
	param coresvccli.ListClustersParam,
) (resp *clusterapi.ListClustersResp, err error) {
	// 参数校验
	if len(param.ProjectIDs) > 0 {
		return nil, fmt.Errorf("cannot filter resource pools by project_id")
	}
	if len(param.ClusterIDs) > 0 {
		return nil, fmt.Errorf("cannot filter resource pools by cluster_id")
	}
	if len(param.TenantIDs) != 1 {
		return nil, fmt.Errorf("must specify exactly one tenant_id to list resource pools")
	}

	tenantID := param.TenantIDs[0]

	stageStart := time.Now()
	mgmtResp, err := c.coreSvcCli.Mgmt.ResourcePool.ListOrgResourcePools(
		ctx,
		&mgmtv1.ListOrgResourcePoolsRequest{
			OrganizationId: tenantID,
		},
	)
	if err != nil {
		c.logListClustersTiming("mgmt_list_org_resource_pools", stageStart, "tenant_id", tenantID, "error", err.Error())
		return nil, err
	}
	c.logListClustersTiming("mgmt_list_org_resource_pools", stageStart, "tenant_id", tenantID, "region_count", len(mgmtResp.GetRegionResourcePools()), "resource_pool_count", countRegionResourcePools(mgmtResp))
	tenantNames, err := c.tenantNamesByID(ctx, []uint64{tenantID})
	if err != nil {
		return nil, err
	}

	resp = &clusterapi.ListClustersResp{
		Clusters: make([]clusterapi.Cluster, 0),
	}

	for _, regionList := range mgmtResp.GetRegionResourcePools() {
		for _, rp := range regionList.GetResourcePools() {
			if param.ServicePlan != nil && rp.GetResourcePool().GetServicePlan().String() != param.ServicePlan.String() {
				continue
			}

			apiCluster := c.buildResourcePoolClusterCoreWithoutTags(ctx, rp)
			apiCluster.TenantName = tenantNames[tenantID]

			resp.Clusters = append(resp.Clusters, apiCluster)
		}
	}
	resp.Clusters = c.injectTiKVTagsForList(ctx, resp.Clusters)

	resp.Total = int64(len(resp.Clusters))

	return resp, nil
}

// ListResourcePools
//
//	@Id				ListResourcePools
//	@Description	list resource pools
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			req	query		clusterapi.ListResourcePoolsReq	true	"ListResourcePoolsReq"
//	@Success		200	{object}	clusterapi.ListClustersResp
//	@Failure		500	{object}	api.BaseResp
//	@Router			/api/v1/devops/cluster/resource-pools [get]
func (c *ClusterController) ListResourcePools(ctx *gin.Context) {
	if ctrl.IsTenantRestricted(ctx) {
		ctrl.FailWithCode(ctx, fmt.Errorf("resource pools are not accessible with tenant-restricted api keys"), api.ErrCodePermissionDenied)
		return
	}

	req := &clusterapi.ListResourcePoolsReq{}
	if err := ctx.ShouldBindQuery(req); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	resp, err := c.ListResourcePoolsCore(ctx.Request.Context(), req)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	ctrl.Success(ctx, resp)
}

func (c *ClusterController) ListResourcePoolsCore(
	ctx context.Context,
	req *clusterapi.ListResourcePoolsReq,
) (*clusterapi.ListClustersResp, error) {
	var resourcePoolIDFilters []string
	page := uint32(1)
	pageSize := uint32(10)
	view := nextgenv1.ListResourcePoolsRequest_BASIC
	if req != nil {
		resourcePoolIDFilters = req.ResourcePoolIDs
		if req.Page > 0 {
			page = req.Page
		}
		if req.PageSize > 0 {
			pageSize = req.PageSize
		}
		parsedView, parseErr := parseListResourcePoolsView(req.View)
		if parseErr != nil {
			return nil, parseErr
		}
		view = parsedView
	}
	if pageSize > 50 {
		pageSize = 50
	}

	resourcePoolIDs, err := parseIDs(resourcePoolIDFilters)
	if err != nil {
		return nil, err
	}

	mgmtReq := &nextgenv1.ListResourcePoolsRequest{
		ResourcePoolIds: resourcePoolIDs,
		PageSize:        pageSize,
		Page:            page,
		View:            view,
	}
	stageStart := time.Now()
	listResp, err := c.coreSvcCli.Mgmt.ResourcePool.ListResourcePools(ctx, mgmtReq)
	if err != nil {
		c.logListClustersTiming("mgmt_list_resource_pools", stageStart, "resource_pool_id_count", len(resourcePoolIDs), "page", page, "page_size", pageSize, "view", view.String(), "error", err.Error())
		return nil, err
	}
	c.logListClustersTiming("mgmt_list_resource_pools", stageStart, "resource_pool_id_count", len(resourcePoolIDs), "page", page, "page_size", pageSize, "view", view.String(), "resource_pool_count", len(listResp.GetResourcePools()), "total", listResp.GetTotalSize())

	resp := &clusterapi.ListClustersResp{
		Clusters: make([]clusterapi.Cluster, 0, len(listResp.GetResourcePools())),
	}

	tenantNames, err := c.tenantNamesByID(ctx, resourcePoolTenantIDs(listResp.GetResourcePools()))
	if err != nil {
		return nil, err
	}
	for _, rp := range listResp.GetResourcePools() {
		apiCluster := c.buildResourcePoolClusterCoreWithoutTags(ctx, rp)
		apiCluster.TenantName = tenantNames[apiCluster.TenantID]
		resp.Clusters = append(resp.Clusters, apiCluster)
	}
	resp.Clusters = c.injectTiKVTagsForList(ctx, resp.Clusters)

	resp.Total = int64(listResp.GetTotalSize())
	if resp.Total == 0 && len(resp.Clusters) > 0 {
		resp.Total = int64(len(resp.Clusters))
	}
	return resp, nil
}

// ListSharedResourcePools
//
//	@Id				ListSharedResourcePools
//	@Description	list shared resource pools
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			req	query		clusterapi.ListSharedResourcePoolsReq	true	"ListSharedResourcePoolsReq"
//	@Success		200	{object}	clusterapi.ListClustersResp
//	@Failure		500	{object}	api.BaseResp
//	@Router			/api/v1/devops/cluster/shared-resource-pools [get]
func (c *ClusterController) ListSharedResourcePools(ctx *gin.Context) {
	if ctrl.IsTenantRestricted(ctx) {
		ctrl.FailWithCode(ctx, fmt.Errorf("shared resource pools are not accessible with tenant-restricted api keys"), api.ErrCodePermissionDenied)
		return
	}

	req := &clusterapi.ListSharedResourcePoolsReq{}
	if err := ctx.ShouldBindQuery(req); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	resp, err := c.ListSharedResourcePoolsCore(ctx.Request.Context(), req)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	ctrl.Success(ctx, resp)
}

func (c *ClusterController) ListSharedResourcePoolsCore(
	ctx context.Context,
	req *clusterapi.ListSharedResourcePoolsReq,
) (*clusterapi.ListClustersResp, error) {
	var servicePlan string
	var resourcePoolIDFilters []string
	page := uint32(1)
	pageSize := uint32(10)
	if req != nil {
		resourcePoolIDFilters = req.ResourcePoolIDs
		servicePlan = req.ServicePlan
		if req.Page > 0 {
			page = req.Page
		}
		if req.PageSize > 0 {
			pageSize = req.PageSize
		}
	}
	if pageSize > 50 {
		pageSize = 50
	}

	if servicePlan != "" {
		if _, ok := v1beta11.ServicePlan_value[servicePlan]; !ok {
			return nil, fmt.Errorf("invalid service plan: %s", servicePlan)
		}
	}
	resourcePoolIDs, err := parseIDs(resourcePoolIDFilters)
	if err != nil {
		return nil, err
	}

	mgmtReq := &nextgenv1.ListResourcePoolsRequest{
		ResourcePoolIds: resourcePoolIDs,
		OwnerType:       nextgenv1.ResourcePool_SHARED,
		PageSize:        pageSize,
		Page:            page,
		View:            nextgenv1.ListResourcePoolsRequest_DETAIL,
	}
	stageStart := time.Now()
	listResp, err := c.coreSvcCli.Mgmt.ResourcePool.ListResourcePools(ctx, mgmtReq)
	if err != nil {
		c.logListClustersTiming("mgmt_list_shared_resource_pools", stageStart, "resource_pool_id_count", len(resourcePoolIDs), "service_plan", servicePlan, "page", page, "page_size", pageSize, "error", err.Error())
		return nil, err
	}
	c.logListClustersTiming("mgmt_list_shared_resource_pools", stageStart, "resource_pool_id_count", len(resourcePoolIDs), "service_plan", servicePlan, "page", page, "page_size", pageSize, "resource_pool_count", len(listResp.GetResourcePools()), "total", listResp.GetTotalSize())

	resp := &clusterapi.ListClustersResp{
		Clusters: make([]clusterapi.Cluster, 0, len(listResp.GetResourcePools())),
	}

	for _, rp := range listResp.GetResourcePools() {
		if rp.GetResourcePool().GetOwnerType() != nextgenv1.ResourcePool_SHARED {
			continue
		}
		if servicePlan != "" && rp.GetResourcePool().GetServicePlan().String() != servicePlan {
			continue
		}
		resp.Clusters = append(resp.Clusters, c.buildResourcePoolClusterCoreWithoutTags(ctx, rp))
	}
	resp.Clusters = c.injectTiKVTagsForList(ctx, resp.Clusters)

	resp.Total = int64(len(resp.Clusters))
	return resp, nil
}

// GetResourcePool
//
//	@Id				GetResourcePool
//	@Description	get resource pool
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			resource_pool_id	path		string	true	"resource_pool_id"
//	@Success		200					{object}	clusterapi.GetClusterResp
//	@Failure		500					{object}	api.BaseResp
//	@Router			/api/v1/devops/cluster/resource-pools/{resource_pool_id} [get]
func (c *ClusterController) GetResourcePool(ctx *gin.Context) {
	resourcePoolID, err := strconv.ParseUint(ctx.Param("resource_pool_id"), 10, 64)
	if err != nil {
		ctrl.FailWithCode(ctx, fmt.Errorf("resource_pool_id is not a int"), api.ErrCodeInvalidArgument)
		return
	}

	cluster, err := c.getResourcePoolAsClusterCore(ctx.Request.Context(), resourcePoolID)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	ctrl.Success(ctx, &clusterapi.GetClusterResp{Cluster: cluster})
}

// GetSharedResourcePool
//
//	@Id				GetSharedResourcePool
//	@Description	get shared resource pool
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			resource_pool_id	path		string	true	"resource_pool_id"
//	@Success		200					{object}	clusterapi.GetClusterResp
//	@Failure		500					{object}	api.BaseResp
//	@Router			/api/v1/devops/cluster/shared-resource-pools/{resource_pool_id} [get]
func (c *ClusterController) GetSharedResourcePool(ctx *gin.Context) {
	if ctrl.IsTenantRestricted(ctx) {
		ctrl.FailWithCode(ctx, fmt.Errorf("shared resource pools are not accessible with tenant-restricted api keys"), api.ErrCodePermissionDenied)
		return
	}

	resourcePoolID, err := strconv.ParseUint(ctx.Param("resource_pool_id"), 10, 64)
	if err != nil {
		ctrl.FailWithCode(ctx, fmt.Errorf("resource_pool_id is not a int"), api.ErrCodeInvalidArgument)
		return
	}

	resp, err := c.GetSharedResourcePoolCore(ctx.Request.Context(), resourcePoolID)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	ctrl.Success(ctx, resp)
}

func (c *ClusterController) GetSharedResourcePoolCore(
	ctx context.Context,
	resourcePoolID uint64,
) (*clusterapi.GetClusterResp, error) {
	rp, err := c.coreSvcCli.Mgmt.GetResourcePool(ctx, resourcePoolID)
	if err != nil {
		return nil, err
	}
	if rp.GetResourcePool().GetOwnerType() != nextgenv1.ResourcePool_SHARED {
		return nil, errs.APIErrorf(api.ErrCodeNotFound, "shared resource pool %d not found", resourcePoolID)
	}

	cluster := c.buildResourcePoolClusterCore(ctx, rp)
	if err := c.injectLogicalClusterTiKVGroupIDs(ctx, &cluster); err != nil {
		return nil, err
	}
	return &clusterapi.GetClusterResp{Cluster: cluster}, nil
}

// GetCluster
//
//	@Id				GetCluster
//	@Description	get cluster
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			cluster_id	path		string	true	"cluster_id"
//	@Success		200			{object}	clusterapi.GetClusterResp
//	@Failure		500			{object}	api.BaseResp
//	@Router			/api/v1/devops/clusters/{cluster_id} [get]
func (c *ClusterController) GetCluster(ctx *gin.Context) {
	clusterID, err := strconv.ParseUint(ctx.Param("cluster_id"), 10, 64)
	if err != nil {
		ctrl.FailWithCode(ctx, fmt.Errorf("cluster_id is not a int"), api.ErrCodeInvalidArgument)
		return
	}

	resp, err := c.GetClusterCore(ctx.Request.Context(), clusterID)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	ctrl.Success(ctx, resp)
}

func (c *ClusterController) GetClusterCore(
	ctx context.Context,
	clusterID uint64,
) (*clusterapi.GetClusterResp, error) {

	cluster, err := c.coreSvcCli.Mgmt.GetCluster(ctx, clusterID)
	if err != nil {
		if apie, ok := errs.AsAPIError(err); ok && apie.ErrCode() == api.ErrCodeNotFound {
			rp, err := c.getResourcePoolAsClusterCore(ctx, clusterID)
			if err != nil {
				return nil, err
			}
			return &clusterapi.GetClusterResp{Cluster: rp}, nil
		}
		return nil, err
	}

	tenantEntity, err := c.tenantSvc.GetTenant(ctx, cluster.OrganizationId)
	if err != nil {
		return nil, err
	}

	if err := c.verifyRegionPermissionCore(ctx, tenantEntity.GetTenantRegion()); err != nil {
		return nil, err
	}

	projectName := ""
	if cluster.GetDeployType() != mgmtv1.DeployType_DEPLOY_TYPE_NEXTGEN {
		project, err := c.accountCli.GetProject(ctx, cluster.ProjectId)
		if err != nil {
			return nil, err
		}
		projectName = project.GetName()
	}

	apiCluster, err := clusterapi.BuildClusterFromMgmt(cluster, projectName)
	if err != nil {
		return nil, err
	}
	apiCluster.TenantName = tenantEntity.Name

	switch cluster.GetDeployType() {

	case mgmtv1.DeployType_DEPLOY_TYPE_SERVERLESS:

		clusterDetails, err := c.serverlessCli.GetCluster(
			ctx,
			strconv.FormatUint(clusterID, 10),
			strconv.FormatUint(cluster.ProjectId, 10),
			strconv.FormatUint(cluster.OrganizationId, 10),
		)
		if err != nil {
			return nil, err
		}

		apiCluster.ServerlessDetails = &clusterapi.ServerlessDetails{
			ResourceConfig: clusterapi.ResourceConfig{
				TiDBPool:             clusterDetails.TidbPool,
				RUMode:               clusterDetails.RuMode,
				MinTiDBReplicas:      clusterDetails.MinTidbReplicas,
				MaxTiDBReplicas:      clusterDetails.MaxTidbReplicas,
				MinRCU:               clusterDetails.MinRcu,
				MaxRCU:               clusterDetails.MaxRcu,
				ClusterPlan:          clusterDetails.ClusterPlan,
				Kubernetes:           clusterDetails.Kubernetes,
				ZoneRedundantEnabled: clusterDetails.ZoneRedundantEnabled,
			},
			Branches:            make([]clusterapi.ServerlessBranch, 0),
			KeyspaceToken:       clusterDetails.KeyspaceToken,
			KeyspaceID:          clusterDetails.KeyspaceId,
			EnableCustomerAlert: clusterDetails.EnableCustomerAlert,
		}

		// essential cluster has no spend limit
		if clusterDetails.GetClusterPlan() != "Essential" {
			r, err := c.serverlessCli.GetSpendLimitDetails(ctx, cluster.OrganizationId, clusterID)
			if err == nil && r != nil {
				// if spend limit details is available, set it to apiCluster
				apiCluster.ServerlessDetails.ResourceUsages = clusterapi.ResourceUsages{
					RequestUnit:   r.GetUsages().GetRequestUnit(),
					RowStorage:    r.GetUsages().GetRowStorage(),
					ColumnStorage: r.GetUsages().GetColumnStorage(),
				}
			} else if err != nil {
				c.logger.Error(err, "failed to get spend limit details for serverless cluster", "cluster_id", clusterID)
			}
		}

		branches, err := c.serverlessCli.ListBranches(ctx, clusterID)
		if err != nil {
			return nil, err
		}
		for _, branch := range branches.GetBranches() {
			apiCluster.ServerlessDetails.Branches = append(apiCluster.ServerlessDetails.Branches, clusterapi.ServerlessBranch{
				ID:          branch.GetId(),
				DisplayName: branch.GetDisplayName(),
				Status:      branch.GetState().String(),
				CreateTime:  branch.GetCreateTime().GetSeconds(),
				UpdateTime:  branch.GetUpdateTime().GetSeconds(),
			})
		}

	case mgmtv1.DeployType_DEPLOY_TYPE_NEXTGEN:

		ngCluster := cluster.GetNextGenCluster()
		if apiCluster.NextGenDetails == nil {
			apiCluster.NextGenDetails = &clusterapi.NextGenDetails{}
		}
		if ngCluster.GetState() != nextgenv1.Cluster_DELETED {
			apiCluster.NextGenDetails.SharedPool = &clusterapi.SharedPool{
				ID: ngCluster.GetSharedPoolId(),
			}
			if err := c.injectNextGenTiKVGroupID(ctx, &apiCluster, ngCluster); err != nil {
				return nil, err
			}
			break
		}
		apiCluster = c.injectTiDBTagsForList(ctx, []clusterapi.Cluster{apiCluster}, []*mgmtv1.Cluster{cluster})[0]
		sharedPool, err := c.mgmtHTTP.NextGen().GetSharedPool(ctx, strconv.FormatUint(clusterID, 10), ngCluster.GetSharedPoolId())
		if err != nil {
			return nil, err
		}
		apiCluster.NextGenDetails.SharedPool = clusterapi.BuildSharedPool(sharedPool)
	}

	return &clusterapi.GetClusterResp{Cluster: apiCluster}, nil
}

func (c *ClusterController) getResourcePoolAsClusterCore(
	ctx context.Context,
	resourcePoolID uint64,
) (clusterapi.Cluster, error) {

	rp, err := c.coreSvcCli.Mgmt.GetResourcePool(ctx, resourcePoolID)
	if err != nil {
		return clusterapi.Cluster{}, err
	}

	cluster := c.buildResourcePoolClusterCore(ctx, rp)
	if err := c.injectLogicalClusterTiKVGroupIDs(ctx, &cluster); err != nil {
		return clusterapi.Cluster{}, err
	}
	return cluster, nil
}

func (c *ClusterController) buildResourcePoolClusterCore(
	ctx context.Context,
	rp *nextgenv1.ResourcePoolProfile,
) clusterapi.Cluster {
	apiCluster := c.buildResourcePoolClusterCoreWithoutTags(ctx, rp)
	apiCluster = c.injectTiKVTagsForList(ctx, []clusterapi.Cluster{apiCluster})[0]

	return apiCluster
}

func (c *ClusterController) buildResourcePoolClusterCoreWithoutTags(
	ctx context.Context,
	rp *nextgenv1.ResourcePoolProfile,
) clusterapi.Cluster {
	resourcePoolID := rp.GetResourcePool().GetId()
	idStr := strconv.FormatUint(resourcePoolID, 10)
	apiCluster := clusterapi.BuildClusterFromResourcePool(rp)

	sp, err := c.mgmtHTTP.NextGenRegionServer(apiCluster.LogicalRegion, apiCluster.Provider).GetSharedPool(ctx, idStr)
	if err == nil {
		hiddenComponents := []clusterapi.ComponentType{
			clusterapi.ComponentTypeTidb,
			clusterapi.ComponentTypeTiproxy,
		}
		if rp.GetResourcePool().GetOwnerType() == nextgenv1.ResourcePool_SHARED {
			hiddenComponents = []clusterapi.ComponentType{clusterapi.ComponentTypeTidb}
		}
		apiCluster.ResourcePoolDetails.SharedPool = clusterapi.BuildSharedPool(sp, hiddenComponents...)
		apiCluster.Status = apiCluster.ResourcePoolDetails.SharedPool.Phase
	}

	return apiCluster
}

func (c *ClusterController) injectNextGenTiKVGroupID(
	ctx context.Context,
	apiCluster *clusterapi.Cluster,
	ngCluster *nextgenv1.Cluster,
) error {
	if apiCluster == nil || apiCluster.NextGenDetails == nil || ngCluster == nil {
		return nil
	}
	location, ok := c.nextGenPlacementPolicyLocation(ctx, ngCluster)
	if !ok {
		c.logger.Info("skip nextgen tikv group injection without regional location", "cluster_id", apiCluster.ID, "shared_pool_id", ngCluster.GetSharedPoolId())
		return nil
	}
	clusterID := strconv.FormatUint(apiCluster.ID, 10)
	tikvGroupID, err := c.placementPolicyTiKVGroupID(ctx, ngCluster.GetSharedPoolId(), clusterID, location)
	if err != nil {
		return err
	}
	apiCluster.NextGenDetails.TiKVGroupID = tikvGroupID
	return nil
}

func (c *ClusterController) injectLogicalClusterTiKVGroupIDs(
	ctx context.Context,
	apiCluster *clusterapi.Cluster,
) error {
	if apiCluster == nil || apiCluster.ResourcePoolDetails == nil || len(apiCluster.ResourcePoolDetails.LogicalClusters) == 0 {
		return nil
	}
	location := componentTagSearchLocation{
		logicalRegion: apiCluster.LogicalRegion,
		cloudProvider: normalizeCloudProvider(apiCluster.Provider),
	}
	if !location.valid() {
		c.logger.Info("skip logical cluster tikv group injection without regional location", "resource_pool_id", apiCluster.ID)
		return nil
	}

	sharedPoolID := strconv.FormatUint(apiCluster.ID, 10)
	tikvGroupIDsByClusterID, err := c.placementPolicyTiKVGroupIDs(ctx, sharedPoolID, location)
	if err != nil {
		return err
	}
	for i := range apiCluster.ResourcePoolDetails.LogicalClusters {
		logicalCluster := &apiCluster.ResourcePoolDetails.LogicalClusters[i]
		logicalCluster.TiKVGroupID = tikvGroupIDsByClusterID[logicalCluster.ID]
	}
	return nil
}

func (c *ClusterController) placementPolicyTiKVGroupIDs(
	ctx context.Context,
	sharedPoolID string,
	location componentTagSearchLocation,
) (map[string]string, error) {
	tikvGroupIDsByClusterID := make(map[string]string)
	if sharedPoolID == "" || !location.valid() || c.mgmtHTTP == nil {
		return tikvGroupIDsByClusterID, nil
	}

	resp, err := c.mgmtHTTP.NextGenRegionServer(location.logicalRegion, location.cloudProvider).
		ListPlacementPolicies(ctx, &clusterng.ListPlacementPoliciesRequest{
			SharedPoolId: sharedPoolID,
		})
	if err != nil {
		if errs.IsNotFound(err) {
			return tikvGroupIDsByClusterID, nil
		}
		return nil, err
	}
	for _, policy := range resp.GetPlacementPolicies() {
		if policy.GetTikvGroupId() == "" {
			continue
		}
		if clusterID := policy.GetClusterId(); clusterID != "" {
			tikvGroupIDsByClusterID[clusterID] = policy.GetTikvGroupId()
		}
		if placementPolicyID := policy.GetPlacementPolicyId(); placementPolicyID != "" {
			tikvGroupIDsByClusterID[placementPolicyID] = policy.GetTikvGroupId()
		}
	}
	return tikvGroupIDsByClusterID, nil
}

func (c *ClusterController) placementPolicyTiKVGroupID(
	ctx context.Context,
	sharedPoolID string,
	clusterID string,
	location componentTagSearchLocation,
) (string, error) {
	if sharedPoolID == "" || clusterID == "" || !location.valid() || c.mgmtHTTP == nil {
		return "", nil
	}
	policy, err := c.mgmtHTTP.NextGenRegionServer(location.logicalRegion, location.cloudProvider).
		GetPlacementPolicy(ctx, &clusterng.GetPlacementPolicyRequest{
			SharedPoolId:      sharedPoolID,
			PlacementPolicyId: clusterID,
		})
	if err != nil {
		if errs.IsNotFound(err) {
			return "", nil
		}
		return "", err
	}
	return policy.GetTikvGroupId(), nil
}

func (c *ClusterController) nextGenPlacementPolicyLocation(
	ctx context.Context,
	ngCluster *nextgenv1.Cluster,
) (componentTagSearchLocation, bool) {
	if ngCluster == nil {
		return componentTagSearchLocation{}, false
	}
	location := componentTagSearchLocation{
		logicalRegion: ngCluster.GetRegionLogicalName(),
		cloudProvider: normalizeCloudProvider(ngCluster.GetCloudProviderId().String()),
	}
	if location.valid() {
		return location, true
	}
	resolvedLocation, ok := c.componentTagSearchLocationByResourcePoolID(ctx, ngCluster.GetSharedPoolId(), map[string]componentTagSearchLocation{})
	if !ok {
		return location, false
	}
	if location.logicalRegion == "" {
		location.logicalRegion = resolvedLocation.logicalRegion
	}
	if location.cloudProvider == "" {
		location.cloudProvider = resolvedLocation.cloudProvider
	}
	return location, location.valid()
}

func (c *ClusterController) verifyRegionPermissionCore(
	ctx context.Context,
	region tntent.TenantRegion,
) error {
	operatorType := consts.GetOperatorTypeFromCtx(ctx)
	if operatorType != consts.CtxOperatorTypeUser && operatorType != consts.CtxOperatorTypeAPIToken {
		return nil
	}

	user, ok := ctx.Value(consts.CtxUserProfile).(*userapiv1.PrincipalProfile)
	if !ok {
		return errs.APIErrorf(api.ErrCodeUnauthenticated, "user profile not found")
	}

	if !user.AggregatedRoleScope.AllowGlobalRegion {
		for _, r := range user.AggregatedRoleScope.AllowedRegions {
			if tntent.TenantRegion(r) == region {
				return nil
			}
		}
		return errs.APIErrorf(api.ErrCodePermissionDenied, "region %s is not allowed for user", region)
	}

	return nil
}

func (c *ClusterController) tenantNamesByID(ctx context.Context, tenantIDs []uint64) (map[uint64]string, error) {
	tenantIDs = sortedUniqueUint64s(tenantIDs)
	if len(tenantIDs) == 0 {
		return map[uint64]string{}, nil
	}

	tenants, err := c.accountCli.ListTenants(ctx, &account.OrgFilter{Ids: tenantIDs})
	if err != nil {
		return nil, err
	}

	tenantNames := make(map[uint64]string, len(tenants))
	for _, tenant := range tenants {
		tenantNames[tenant.GetId()] = tenant.GetName()
	}
	return tenantNames, nil
}

func resourcePoolTenantIDs(resourcePools []*nextgenv1.ResourcePoolProfile) []uint64 {
	tenantIDs := make([]uint64, 0, len(resourcePools))
	for _, rp := range resourcePools {
		rpp := rp.GetResourcePool()
		if rpp.GetOwnerType() != nextgenv1.ResourcePool_ORGANIZATION {
			continue
		}
		if rpp.GetOwnerId() == 0 {
			continue
		}
		tenantIDs = append(tenantIDs, rpp.GetOwnerId())
	}
	return tenantIDs
}

const clusterTagSearchPageSize int32 = 500

type componentTagSearchLocation struct {
	logicalRegion string
	cloudProvider string
}

func (l componentTagSearchLocation) valid() bool {
	return l.logicalRegion != "" && normalizeCloudProvider(l.cloudProvider) != ""
}

func (c *ClusterController) injectTiDBTagsForList(
	ctx context.Context,
	clusters []clusterapi.Cluster,
	mgmtClusters []*mgmtv1.Cluster,
) []clusterapi.Cluster {
	idsByLocation := make(map[componentTagSearchLocation][]string)
	resourcePoolLocationCache := make(map[string]componentTagSearchLocation)
	for i := range clusters {
		if clusters[i].DeployType != mgmtv1.DeployType_DEPLOY_TYPE_NEXTGEN.String() {
			continue
		}
		if i >= len(mgmtClusters) {
			continue
		}
		mgmtCluster := mgmtClusters[i]
		ngCluster := mgmtCluster.GetNextGenCluster()
		location := componentTagSearchLocation{
			logicalRegion: ngCluster.GetRegionLogicalName(),
			cloudProvider: normalizeCloudProvider(ngCluster.GetCloudProviderId().String()),
		}
		if location.cloudProvider == "" {
			location.cloudProvider = normalizeCloudProvider(clusters[i].Provider)
		}
		if !location.valid() {
			resolvedLocation, ok := c.componentTagSearchLocationByResourcePoolID(ctx, ngCluster.GetSharedPoolId(), resourcePoolLocationCache)
			if !ok {
				continue
			}
			if location.logicalRegion == "" {
				location.logicalRegion = resolvedLocation.logicalRegion
			}
			if location.cloudProvider == "" {
				location.cloudProvider = resolvedLocation.cloudProvider
			}
		}
		location.cloudProvider = normalizeCloudProvider(location.cloudProvider)
		if !location.valid() {
			c.logger.Info("skip logical cluster tag injection without regional location", "cluster_id", clusters[i].ID)
			continue
		}
		idStr := strconv.FormatUint(clusters[i].ID, 10)
		idsByLocation[location] = append(idsByLocation[location], idStr)
	}

	tagsByClusterID := c.searchComponentTagsByLocation(
		ctx,
		ccproto.ClusterType_LOGICAL_CLUSTER,
		ccproto.LOGICAL_CLUSTER_COMPONENT_TIDB,
		idsByLocation,
	)
	for i := range clusters {
		idStr := strconv.FormatUint(clusters[i].ID, 10)
		tag, ok := tagsByClusterID[idStr]
		if !ok {
			continue
		}
		if clusters[i].NextGenDetails == nil {
			clusters[i].NextGenDetails = &clusterapi.NextGenDetails{}
		}
		clusters[i].NextGenDetails.TiDBTag = tag
	}
	return clusters
}

func (c *ClusterController) injectTiKVTagsForList(ctx context.Context, clusters []clusterapi.Cluster) []clusterapi.Cluster {
	idsByLocation := make(map[componentTagSearchLocation][]string)
	for i := range clusters {
		if clusters[i].DeployType != "DEPLOY_TYPE_RESOURCE_POOL" {
			continue
		}
		location := componentTagSearchLocation{
			logicalRegion: clusters[i].LogicalRegion,
			cloudProvider: normalizeCloudProvider(clusters[i].Provider),
		}
		if !location.valid() {
			c.logger.Info("skip resource pool tag injection without regional location", "resource_pool_id", clusters[i].ID)
			continue
		}
		idStr := strconv.FormatUint(clusters[i].ID, 10)
		idsByLocation[location] = append(idsByLocation[location], idStr)
	}

	tagsByClusterID := c.searchComponentTagsByLocation(
		ctx,
		ccproto.ClusterType_SHARED_POOL,
		ccproto.SHARED_POOL_COMPONENT_TIKV,
		idsByLocation,
	)
	for i := range clusters {
		idStr := strconv.FormatUint(clusters[i].ID, 10)
		tag, ok := tagsByClusterID[idStr]
		if !ok {
			continue
		}
		if clusters[i].ResourcePoolDetails == nil {
			clusters[i].ResourcePoolDetails = &clusterapi.ResourcePoolDetails{}
		}
		clusters[i].ResourcePoolDetails.TiKVTag = tag
	}
	return clusters
}

func (c *ClusterController) componentTagSearchLocationByResourcePoolID(
	ctx context.Context,
	resourcePoolID string,
	cache map[string]componentTagSearchLocation,
) (componentTagSearchLocation, bool) {
	if resourcePoolID == "" {
		return componentTagSearchLocation{}, false
	}
	if location, ok := cache[resourcePoolID]; ok {
		return location, location.valid()
	}
	if c.coreSvcCli == nil || c.coreSvcCli.Mgmt == nil {
		return componentTagSearchLocation{}, false
	}

	start := time.Now()
	rp, err := c.coreSvcCli.Mgmt.GetResourcePoolString(ctx, resourcePoolID)
	if err != nil {
		c.logListClustersTiming("mgmt_get_resource_pool", start, "resource_pool_id", resourcePoolID, "error", err.Error())
		c.logger.Error(err, "failed to get resource pool region for component tag injection", "resource_pool_id", resourcePoolID)
		cache[resourcePoolID] = componentTagSearchLocation{}
		return componentTagSearchLocation{}, false
	}
	location := componentTagSearchLocation{
		logicalRegion: rp.GetRegion().GetLogicalName(),
		cloudProvider: normalizeCloudProvider(rp.GetRegion().GetCloudProviderId().String()),
	}
	if location.cloudProvider == "" {
		location.cloudProvider = normalizeCloudProvider(rp.GetResourcePool().GetProviderId().String())
	}
	if !location.valid() {
		c.logger.Info("skip component tag injection without resource pool regional location", "resource_pool_id", resourcePoolID)
	}
	c.logListClustersTiming("mgmt_get_resource_pool", start, "resource_pool_id", resourcePoolID, "logical_region", location.logicalRegion, "cloud_provider", location.cloudProvider, "location_valid", location.valid())
	cache[resourcePoolID] = location
	return location, location.valid()
}

func (c *ClusterController) searchComponentTagsByLocation(
	ctx context.Context,
	clusterType ccproto.ClusterType,
	componentName string,
	clusterIDsByLocation map[componentTagSearchLocation][]string,
) map[string]string {
	tagsByClusterID := make(map[string]string)
	if c.mgmtHTTP == nil {
		return tagsByClusterID
	}
	configCenter := c.mgmtHTTP.ConfigCenterNgProxy()
	if configCenter == nil {
		return tagsByClusterID
	}

	locations := make([]componentTagSearchLocation, 0, len(clusterIDsByLocation))
	for location := range clusterIDsByLocation {
		if !location.valid() {
			continue
		}
		locations = append(locations, location)
	}
	sort.Slice(locations, func(i, j int) bool {
		if locations[i].logicalRegion == locations[j].logicalRegion {
			return locations[i].cloudProvider < locations[j].cloudProvider
		}
		return locations[i].logicalRegion < locations[j].logicalRegion
	})
	for _, location := range locations {
		clusterIDs := sortedUniqueStrings(clusterIDsByLocation[location])
		for _, chunk := range chunkStrings(clusterIDs, int(clusterTagSearchPageSize)) {
			for page := int32(1); ; page++ {
				req := &ccproto.SearchClusterComponentsRequest{
					ClusterType:   clusterType,
					ComponentName: componentName,
					ClusterIds:    chunk,
					PageSize:      clusterTagSearchPageSize,
					Page:          page,
					LogicalRegion: &location.logicalRegion,
				}
				resp, err := configCenter.SearchClusterComponents(ctx, location.logicalRegion, location.cloudProvider, req)
				if err != nil {
					c.logger.Error(err, "failed to search cluster component tags", "cluster_type", clusterType.String(), "component_name", componentName, "logical_region", location.logicalRegion, "cloud_provider", location.cloudProvider)
					break
				}
				if resp == nil {
					break
				}
				for _, comp := range resp.GetComponents() {
					if comp.GetComponentName() != componentName {
						continue
					}
					if comp.GetEffectiveImageTag() == "" {
						continue
					}
					tagsByClusterID[comp.GetClusterId()] = comp.GetEffectiveImageTag()
				}
				if !hasNextClusterTagSearchPage(page, clusterTagSearchPageSize, resp.GetTotal(), len(resp.GetComponents())) {
					break
				}
			}
		}
	}
	return tagsByClusterID
}

func normalizeCloudProvider(cloudProvider string) string {
	if cloudProvider == "" || cloudProvider == commonv1.CloudProvider_NAME_UNSPECIFIED.String() {
		return ""
	}
	return cloudProvider
}

func sortedUniqueStrings(items []string) []string {
	set := container.NewSetFromSlice(items)
	result := set.Slice()
	sort.Strings(result)
	return result
}

func sortedUniqueUint64s(items []uint64) []uint64 {
	set := container.NewSetFromSlice(items)
	result := set.Slice()
	sort.Slice(result, func(i, j int) bool {
		return result[i] < result[j]
	})
	return result
}

func countRegionResourcePools(resp *nextgenv1.ListOwnerResourcePoolsResponse) int {
	if resp == nil {
		return 0
	}
	count := 0
	for _, regionList := range resp.GetRegionResourcePools() {
		count += len(regionList.GetResourcePools())
	}
	return count
}

func chunkStrings(items []string, size int) [][]string {
	if len(items) == 0 || size <= 0 {
		return nil
	}
	chunks := make([][]string, 0, (len(items)+size-1)/size)
	for start := 0; start < len(items); start += size {
		end := start + size
		if end > len(items) {
			end = len(items)
		}
		chunks = append(chunks, items[start:end])
	}
	return chunks
}

func hasNextClusterTagSearchPage(page, pageSize int32, total int64, itemCount int) bool {
	if itemCount == 0 {
		return false
	}
	return int64(page)*int64(pageSize) < total
}

// GetTopology
//
//	@Id				ClusterControllerGetTopology
//	@Description	get cluster topology
//	@Tags			ClusterController
//	@Accept			json
//	@Produce		json
//	@Param			cluster_id	path		string	true	"cluster_id"
//	@Success		200			{object}	clusterapi.GetTopologyResp
//	@Failure		500			{object}	api.BaseResp
//	@Router			/api/v1/devops/clusters/{cluster_id}/topology [get]
func (c *ClusterController) GetTopology(ctx *gin.Context) {
	clusterID := ctx.Param("cluster_id")
	if clusterID == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("cluster_id is not a int"), api.ErrCodeInvalidArgument)
		return
	}

	topo, err := c.ctlSvc.GetTopology(ctx, clusterID)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	resp := &clusterapi.GetTopologyResp{Topology: topo}
	ctrl.Success(ctx, resp)
}

// GetClusterComponents
//
//	@Id				GetClusterComponents
//	@Description	get cluster components
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			cluster_id	path		string	true	"cluster_id"
//	@Success		200			{object}	clusterapi.GetClusterComponentsResp
//	@Failure		500			{object}	api.BaseResp
//	@Router			/api/v1/devops/clusters/{cluster_id}/components [get]
func (c *ClusterController) GetClusterComponents(ctx *gin.Context) {
	clusterID, err := strconv.ParseUint(ctx.Param("cluster_id"), 10, 64)
	if err != nil {
		ctrl.FailWithCode(ctx, fmt.Errorf("cluster_id is not a int"), api.ErrCodeInvalidArgument)
		return
	}

	resp, err := c.GetClusterComponentsCore(ctx.Request.Context(), clusterID)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	ctrl.Success(ctx, resp)
}

func (c *ClusterController) GetClusterComponentsCore(
	ctx context.Context,
	clusterID uint64,
) (*clusterapi.GetClusterComponentsResp, error) {

	clusters, _, err := c.coreSvcCli.Mgmt.ListClusters(ctx, coresvccli.ListClustersParam{
		ClusterIDs: []uint64{clusterID},
	})
	if err != nil {
		return nil, err
	}

	if len(clusters) == 0 {
		return nil, errs.APIErrorf(api.ErrCodeNotFound, "cluster %d not found", clusterID)
	}

	cluster := clusters[0]

	switch cluster.GetDeployType() {

	// =========================
	// Dedicated
	// =========================
	case mgmtv1.DeployType_DEPLOY_TYPE_DEDICATED:

		resp := clusterapi.BuildDedicatedComponentsDetail(
			cluster.GetDedicatedCluster().GetClusterComponents(),
		)

		if cluster.TikvStorageType.String() != "" {
			resp.TiKVStorageType = cluster.TikvStorageType.String()
		}

		if cluster.TiflashStorageType != nil {
			resp.TiFlashStorageType = cluster.TiflashStorageType.String()
		}

		return &resp, nil

	// =========================
	// NextGen
	// =========================
	case mgmtv1.DeployType_DEPLOY_TYPE_NEXTGEN:

		clusterIDStr := strconv.FormatUint(clusterID, 10)

		sharedPool, err := c.mgmtHTTP.NextGen().
			GetSharedPool(ctx, clusterIDStr, cluster.GetNextGenCluster().GetSharedPoolId())
		if err != nil {
			return nil, err
		}

		tidbGroups, err := c.mgmtHTTP.NextGen().
			ListTiDBGroups(ctx, clusterIDStr)
		if err != nil {
			return nil, err
		}

		resp := &clusterapi.GetClusterComponentsResp{
			TiDB:      make([]*clusterapi.ComponentInfo, 0),
			TiProxy:   make([]*clusterapi.ComponentInfo, 0),
			Cassandra: make([]*clusterapi.ComponentInfo, 0),
		}

		// TiProxy
		for _, comp := range sharedPool.ComponentGroups {
			if comp.ComponentType == "TIPROXY" && comp.Id == clusterIDStr {
				resp.TiProxy = append(resp.TiProxy, &clusterapi.ComponentInfo{
					ComponentType: clusterapi.ComponentTypeTiproxy,
					Count:         comp.NodeQuota.NodeCount,
					CpuNum:        comp.NodeQuota.NodeSpec.Cpu,
					MemorySizeGi:  comp.NodeQuota.NodeSpec.MemoryGi,
				})
			}
		}

		// TiDB groups
		for _, group := range tidbGroups.TidbGroups {
			resp.TiDB = append(resp.TiDB, &clusterapi.ComponentInfo{
				ComponentType: clusterapi.ComponentTypeTidb,
				TidbGroupId:   group.Id,
				Count:         group.NodeQuota.NodeCount,
				CpuNum:        group.NodeQuota.NodeSpec.Cpu,
				MemorySizeGi:  group.NodeQuota.NodeSpec.MemoryGi,
			})
		}

		if hasCassandraProxyComponent(sharedPool.ComponentGroups, clusterIDStr) {
			cassandraRegionServer, err := c.cassandraRegionServer(ctx, cluster.GetNextGenCluster().GetSharedPoolId())
			if err != nil {
				c.logger.Error(err, "failed to resolve cassandra region server", "cluster_id", clusterIDStr, "shared_pool_id", cluster.GetNextGenCluster().GetSharedPoolId())
				return resp, nil
			}

			appendCassandraProxyComponent(ctx, cassandraRegionServer, clusterIDStr, resp, c.logger)
		}

		return resp, nil

	default:
		return nil, errs.APIErrorf(
			api.ErrCodeInvalidArgument,
			"cluster is not a dedicated or nextgen cluster",
		)
	}
}

func hasCassandraProxyComponent(components []*mgmthttpcli.Component, clusterID string) bool {
	for _, component := range components {
		if component != nil && component.ComponentType == "CASSANDRA_PROXY" && component.Id == clusterID {
			return true
		}
	}
	return false
}

func appendCassandraProxyComponent(
	ctx context.Context,
	nextGen mgmthttpcli.NextGenRegionServer,
	clusterID string,
	resp *clusterapi.GetClusterComponentsResp,
	logger logr.Logger,
) {
	cassandra, err := nextGen.GetCassandraProxy(ctx, clusterID)
	if err != nil {
		if !errs.IsNotFound(err) {
			logger.Error(err, "failed to get cassandra proxy component", "cluster_id", clusterID)
		}
		return
	}

	if component := clusterapi.BuildCassandraComponentInfo(cassandra); component != nil {
		resp.Cassandra = append(resp.Cassandra, component)
	}
}

func (c *ClusterController) cassandraRegionServer(
	ctx context.Context,
	sharedPoolID string,
) (mgmthttpcli.NextGenRegionServer, error) {
	rp, err := c.coreSvcCli.Mgmt.GetResourcePoolString(ctx, sharedPoolID)
	if err != nil {
		return nil, err
	}

	region := rp.GetRegion()
	if region == nil {
		return nil, fmt.Errorf("resource pool %s has no region metadata", sharedPoolID)
	}
	if region.GetLogicalName() == "" {
		return nil, fmt.Errorf("resource pool %s has no logical region name", sharedPoolID)
	}
	if region.GetCloudProviderId() == commonv1.CloudProvider_NAME_UNSPECIFIED {
		return nil, fmt.Errorf("resource pool %s has no cloud provider", sharedPoolID)
	}

	return c.mgmtHTTP.NextGenRegionServer(region.GetLogicalName(), region.GetCloudProviderId().String()), nil
}

// ExecCtl
//
//	@Id				ClusterControllerExecCtl
//	@Description	execute ctl command on cluster
//	@Tags			ClusterController
//	@Accept			json
//	@Produce		json
//	@Param			cluster_id	path		string					true	"cluster_id"
//	@Param			req			body		clusterapi.ExecCtlReq	true	"request param"
//	@Success		200			{object}	clusterapi.ExecCtlResp
//	@Failure		500			{object}	api.BaseResp
//	@Router			/api/v1/devops/clusters/{cluster_id}/ctl [post]
func (c *ClusterController) ExecCtl(ctx *gin.Context) {
	clusterID := ctx.Param("cluster_id")
	if clusterID == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("cluster_id is empty"), api.ErrCodeInvalidArgument)
		return
	}

	req := &clusterapi.ExecCtlReq{}
	err := ctx.ShouldBindJSON(req)
	if err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	stdout, stderr, err := c.ctlSvc.ExecuteCtl(ctx, clusterID, req.Command)
	if err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInternal)
		return
	}

	resp := &clusterapi.ExecCtlResp{Stdout: stdout, Stderr: stderr}
	ctrl.Success(ctx, resp)
}

// ListAlerts
//
//	@Id				ListAlerts
//	@Description	list alerts
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			cluster_id	path		string						true	"cluster id"
//	@Param			req			query		clusterapi.ListAlertsReq	true	"request param"
//	@Success		200			{object}	clusterapi.ListAlertsResp
//	@Router			/api/v1/devops/clusters/{cluster_id}/alerts [get]
func (c *ClusterController) ListAlerts(ctx *gin.Context) {
	clusterID, err := strconv.ParseUint(ctx.Param("cluster_id"), 10, 64)
	if err != nil {
		ctrl.FailWithCode(ctx, fmt.Errorf("cluster_id is not a int"), api.ErrCodeInvalidArgument)
		return
	}

	req := &clusterapi.ListAlertsReq{}
	err = ctx.ShouldBindQuery(req)
	if err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	result, err := c.jiraSvc.ListAlerts(ctx, jirasvcapi.ListAlertsParam{
		Search: req.Search,
		Filter: jirasvcapi.ListAlertsFilter{
			Type:      req.Type,
			ClusterID: clusterID,
			StartTs:   req.StartTs,
			EndTs:     req.EndTs,
			Recovered: req.Recovered,
			Priority:  req.Priority,
		},
		PageToken: req.PageToken,
		PageSize:  req.PageSize,
	})
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}
	ctrl.Success(ctx, &clusterapi.ListAlertsResp{
		IsLast:        result.IsLast,
		NextPageToken: result.NextPageToken,
		Alerts:        result.Alerts,
	})
}

// GetAlertSummary
//
//	@Id				GetAlertSummary
//	@Description	get alert summary
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			cluster_id	path		string							true	"cluster id"
//	@Param			req			query		clusterapi.GetAlertSummaryReq	true	"request param"
//	@Success		200			{object}	clusterapi.GetAlertSummaryResp
//	@Router			/api/v1/devops/clusters/{cluster_id}/alert-summary [get]
func (c *ClusterController) GetAlertSummary(ctx *gin.Context) {
	clusterID, err := strconv.ParseUint(ctx.Param("cluster_id"), 10, 64)
	if err != nil {
		ctrl.FailWithCode(ctx, fmt.Errorf("cluster_id is not a int"), api.ErrCodeInvalidArgument)
		return
	}

	req := &clusterapi.GetAlertSummaryReq{}
	err = ctx.ShouldBindQuery(req)
	if err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	summary, err := c.jiraSvc.GetAlertSummary(ctx, jirasvcapi.GetAlertsSummaryParam{
		ClusterID: clusterID,
		StartTs:   req.StartTs,
		EndTs:     req.EndTs,
	})
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}
	ctrl.Success(ctx, &clusterapi.GetAlertSummaryResp{AlertSummary: *summary})
}

// GetDMNamespace
//
//	@Id				GetDMNamespace
//	@Description	get cluster dm namespace info
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			cluster_id	path		string	true	"cluster id"
//	@Success		200			{object}	clusterapi.GetDMNamespaceResp
//	@Failure		500			{object}	api.BaseResp
//	@Router			/api/v1/devops/clusters/{cluster_id}/dm-ns [get]
func (c *ClusterController) GetDMNamespace(ctx *gin.Context) {
	cid := ctx.Param("cluster_id")

	agentURL, err := c.operationSvc.GetOperationClusterWhiteListInfo(ctx, cid)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	dm, err := c.dataflowCli.GetClusterDMConfig(ctx, agentURL, cid)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	ctrl.Success(ctx, clusterapi.GetDMNamespaceResp{
		Namespace: dm.Namespace,
		PodNames:  dm.ExecutorPodNames,
	})
}

// GetCloudIAMInfo
//
//	@Id				GetCloudIAMInfo
//	@Description	get byoc cluster's cloud IAM info
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			cluster_id	path		string	true	"cluster id"
//	@Success		200			{object}	clusterapi.GetCloudIAMInfoResp
//	@Failure		500			{object}	api.BaseResp
//	@Router			/api/v1/devops/clusters/{cluster_id}/iam-info [get]
func (c *ClusterController) GetCloudIAMInfo(ctx *gin.Context) {
	cid := ctx.Param("cluster_id")

	info, err := c.coreSvcCli.Mgmt.GetCloudIAMInfo(ctx, cid)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	ctrl.Success(ctx, clusterapi.GetCloudIAMInfoResp{
		TiDBCloudAccountID:         info.TidbCloudAccountId,
		TiDBCloudAccountExternalID: info.TidbCloudAccountExternalId,
		CloudProvider:              info.CloudProvider.Id.String(),
		Region:                     info.Region.Id.String(),
		BucketName:                 info.Region.GetSharedBucketUrl(),
		LogicalName:                info.Region.LogicalName,
	})
}

func parseIDs(src []string) ([]uint64, error) {
	result := make([]uint64, 0, len(src))
	for _, id := range src {
		projectID, err := strconv.ParseUint(id, 10, 64)
		if err != nil {
			return nil, err
		}
		result = append(result, projectID)
	}
	return result, nil
}

func parseListResourcePoolsView(view string) (nextgenv1.ListResourcePoolsRequest_View, error) {
	if view == "" {
		return nextgenv1.ListResourcePoolsRequest_BASIC, nil
	}
	value, ok := nextgenv1.ListResourcePoolsRequest_View_value[strings.ToUpper(view)]
	if !ok {
		return nextgenv1.ListResourcePoolsRequest_VIEW_UNSPECIFIED, fmt.Errorf("invalid view: %s", view)
	}
	parsedView := nextgenv1.ListResourcePoolsRequest_View(value)
	if parsedView == nextgenv1.ListResourcePoolsRequest_VIEW_UNSPECIFIED {
		return nextgenv1.ListResourcePoolsRequest_VIEW_UNSPECIFIED, fmt.Errorf("invalid view: %s", view)
	}
	return parsedView, nil
}