package cloudconfig

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

	ccproto "github.com/tidbcloud/config-center/proto/config_center"
	devopsapi "github.com/tidbcloud/devops/api"
	ccent "github.com/tidbcloud/devops/entity/cloudconfig"
	"github.com/tidbcloud/devops/pkg/infra/clients/coresvccli"
	"github.com/tidbcloud/devops/pkg/infra/errs"
	"github.com/tidbcloud/devops/pkg/infra/utils/workqueue"
	ccsvcapi "github.com/tidbcloud/devops/pkg/service/cloudconfig/api"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/cvt"
	mgmtv1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
)

const (
	defaultClusterComponentSearchPage          int32 = 1
	defaultClusterComponentSearchPerPage       int32 = 100
	configCenterClusterComponentPageSize       int32 = 500
	maxTenantClusterComponentRows                    = 10000
	clusterComponentSearchConcurrency                = 4
	configCenterSearchClusterComponentsTimeout       = 5 * time.Second
)

type clusterComponentSearchPlan struct {
	ClusterType   ccent.ClusterType
	LogicalRegion string
	ClusterIDs    []string
}

func (s *svc) SearchClusterComponents(ctx context.Context, param ccsvcapi.SearchClusterComponentsParam) (*ccent.ClusterComponentSearchResult, error) {
	param = normalizeSearchClusterComponentsParam(param)
	if param.TenantID != nil && param.ClusterID != "" {
		return nil, errs.APIErrorf(devopsapi.ErrCodeInvalidArgument, "tenant_id and cluster_id are mutually exclusive")
	}
	if cvt.To.ConfigCenter.ClusterType(param.ClusterType) == ccproto.ClusterType_CLUSTER_TYPE_UNSPECIFIED {
		return nil, errs.APIErrorf(devopsapi.ErrCodeInvalidArgument, "invalid cluster_type: %s", param.ClusterType)
	}
	if err := validateRegionMode(param); err != nil {
		return nil, err
	}

	switch {
	case param.TenantID != nil:
		return s.searchTenantClusterComponents(ctx, param)
	case param.ClusterID != "":
		return s.searchSingleClusterComponents(ctx, param)
	default:
		return s.searchGlobalClusterComponents(ctx, param)
	}
}

func normalizeSearchClusterComponentsParam(param ccsvcapi.SearchClusterComponentsParam) ccsvcapi.SearchClusterComponentsParam {
	if param.RegionMode == "" {
		param.RegionMode = ccsvcapi.RegionModeAuto
	}
	if param.Page <= 0 {
		param.Page = defaultClusterComponentSearchPage
	}
	if param.PerPage <= 0 {
		param.PerPage = defaultClusterComponentSearchPerPage
	}
	if param.PerPage > configCenterClusterComponentPageSize {
		param.PerPage = configCenterClusterComponentPageSize
	}
	return param
}

func validateRegionMode(param ccsvcapi.SearchClusterComponentsParam) error {
	switch param.RegionMode {
	case ccsvcapi.RegionModeAuto, ccsvcapi.RegionModeAll, ccsvcapi.RegionModeSingle:
		return nil
	default:
		return errs.APIErrorf(devopsapi.ErrCodeInvalidArgument, "invalid region_mode: %s", param.RegionMode)
	}
}

func (s *svc) searchTenantClusterComponents(ctx context.Context, param ccsvcapi.SearchClusterComponentsParam) (*ccent.ClusterComponentSearchResult, error) {
	var plans []clusterComponentSearchPlan
	var err error
	switch param.ClusterType {
	case ccent.ClusterTypeLogicalCluster:
		plans, err = s.buildTenantLogicalClusterComponentPlans(ctx, *param.TenantID, param.LogicalRegion)
	case ccent.ClusterTypeSharedPool:
		plans, err = s.buildTenantSharedPoolComponentPlans(ctx, *param.TenantID, param.LogicalRegion)
	default:
		return nil, errs.APIErrorf(devopsapi.ErrCodeInvalidArgument, "tenant scope only supports %s or %s", ccent.ClusterTypeLogicalCluster, ccent.ClusterTypeSharedPool)
	}
	if err != nil {
		return nil, err
	}
	if len(plans) == 0 {
		return &ccent.ClusterComponentSearchResult{
			Items: []*ccent.ClusterComponentVersion{},
		}, nil
	}

	items, err := s.fetchAllClusterComponents(ctx, plans, param)
	if err != nil {
		return nil, err
	}
	sortClusterComponentVersions(items)

	total := int64(len(items))
	items = paginateClusterComponentVersions(items, param.Page, param.PerPage)
	return &ccent.ClusterComponentSearchResult{
		Total: total,
		Items: items,
	}, nil
}

func (s *svc) searchSingleClusterComponents(ctx context.Context, param ccsvcapi.SearchClusterComponentsParam) (*ccent.ClusterComponentSearchResult, error) {
	plan := clusterComponentSearchPlan{
		ClusterType: param.ClusterType,
		ClusterIDs:  []string{param.ClusterID},
	}
	if isRegionalClusterType(param.ClusterType) {
		logicalRegion, err := s.resolveClusterComponentLogicalRegion(ctx, param.ClusterType, param.ClusterID)
		if err != nil {
			return nil, err
		}
		plan.LogicalRegion = logicalRegion
	}

	regionResult, err := s.searchClusterComponentsByPlan(ctx, plan, param, param.Page, param.PerPage)
	if err != nil {
		return nil, err
	}
	return &ccent.ClusterComponentSearchResult{
		Total: regionResult.Total,
		Items: regionResult.Items,
	}, nil
}

func (s *svc) searchGlobalClusterComponents(ctx context.Context, param ccsvcapi.SearchClusterComponentsParam) (*ccent.ClusterComponentSearchResult, error) {
	if param.ComponentName == "" {
		return nil, errs.APIErrorf(devopsapi.ErrCodeInvalidArgument, "component_name is required for global search")
	}
	if !isRegionalClusterType(param.ClusterType) {
		regionResult, err := s.searchClusterComponentsByPlan(ctx, clusterComponentSearchPlan{ClusterType: param.ClusterType}, param, param.Page, param.PerPage)
		if err != nil {
			return nil, err
		}
		return &ccent.ClusterComponentSearchResult{
			Total: regionResult.Total,
			Items: regionResult.Items,
		}, nil
	}

	if param.LogicalRegion != "" {
		regionResult, err := s.searchClusterComponentsByPlan(ctx, clusterComponentSearchPlan{
			ClusterType:   param.ClusterType,
			LogicalRegion: param.LogicalRegion,
		}, param, param.Page, param.PerPage)
		if err != nil {
			return nil, err
		}
		return &ccent.ClusterComponentSearchResult{
			Total: regionResult.Total,
			Items: regionResult.Items,
		}, nil
	}
	if param.RegionMode == ccsvcapi.RegionModeSingle {
		return nil, errs.APIErrorf(devopsapi.ErrCodeInvalidArgument, "logical_region is required when region_mode=single")
	}
	if param.Page > 1 {
		return nil, errs.APIErrorf(devopsapi.ErrCodeInvalidArgument, "all-region search only supports page=1; use logical_region and region_mode=single to load more for one region")
	}

	regions, err := s.listNextGenLogicalRegions(ctx)
	if err != nil {
		return nil, err
	}
	plans := make([]clusterComponentSearchPlan, 0, len(regions))
	for _, region := range regions {
		plans = append(plans, clusterComponentSearchPlan{
			ClusterType:   param.ClusterType,
			LogicalRegion: region,
		})
	}
	return s.searchAllRegionClusterComponents(ctx, plans, param)
}

func (s *svc) buildTenantLogicalClusterComponentPlans(ctx context.Context, tenantID uint64, logicalRegionFilter string) ([]clusterComponentSearchPlan, error) {
	deployType := mgmtv1.DeployType_DEPLOY_TYPE_NEXTGEN
	clusters := make([]*mgmtv1.Cluster, 0)
	for page := int64(1); ; page++ {
		batch, total, err := s.coreSvc.Mgmt.ListClusters(ctx, coresvccli.ListClustersParam{
			TenantIDs:  []uint64{tenantID},
			DeployType: &deployType,
			Page:       page,
			PerPage:    int64(configCenterClusterComponentPageSize),
		})
		if err != nil {
			return nil, err
		}
		clusters = append(clusters, batch...)
		if len(batch) == 0 || int64(len(clusters)) >= total {
			break
		}
	}

	regionBySharedPoolID := make(map[string]string)
	clusterIDsByRegion := make(map[string][]string)
	for _, cluster := range clusters {
		sharedPoolID := cluster.GetNextGenCluster().GetSharedPoolId()
		if sharedPoolID == "" {
			return nil, fmt.Errorf("shared pool id is empty for logical cluster %d", cluster.GetClusterId())
		}
		logicalRegion, ok := regionBySharedPoolID[sharedPoolID]
		if !ok {
			var err error
			logicalRegion, err = s.logicalRegionByResourcePoolID(ctx, sharedPoolID)
			if err != nil {
				return nil, fmt.Errorf("resolve logical region for logical cluster %d failed: %w", cluster.GetClusterId(), err)
			}
			regionBySharedPoolID[sharedPoolID] = logicalRegion
		}
		if logicalRegionFilter != "" && logicalRegionFilter != logicalRegion {
			continue
		}
		clusterIDsByRegion[logicalRegion] = append(clusterIDsByRegion[logicalRegion], strconv.FormatUint(cluster.GetClusterId(), 10))
	}
	return clusterComponentPlansFromRegionMap(ccent.ClusterTypeLogicalCluster, clusterIDsByRegion), nil
}

func (s *svc) buildTenantSharedPoolComponentPlans(ctx context.Context, tenantID uint64, logicalRegionFilter string) ([]clusterComponentSearchPlan, error) {
	resp, err := s.coreSvc.Mgmt.ResourcePool.ListOrgResourcePools(ctx, &mgmtv1.ListOrgResourcePoolsRequest{
		OrganizationId: tenantID,
	})
	if err != nil {
		return nil, err
	}

	seen := make(map[string]struct{})
	clusterIDsByRegion := make(map[string][]string)
	for _, regionList := range resp.GetRegionResourcePools() {
		for _, rp := range regionList.GetResourcePools() {
			id := strconv.FormatUint(rp.GetResourcePool().GetId(), 10)
			if _, ok := seen[id]; ok {
				continue
			}
			seen[id] = struct{}{}

			logicalRegion := rp.GetRegion().GetLogicalName()
			if logicalRegion == "" {
				return nil, fmt.Errorf("logical region is empty for resource pool %s", id)
			}
			if logicalRegionFilter != "" && logicalRegionFilter != logicalRegion {
				continue
			}
			clusterIDsByRegion[logicalRegion] = append(clusterIDsByRegion[logicalRegion], id)
		}
	}
	return clusterComponentPlansFromRegionMap(ccent.ClusterTypeSharedPool, clusterIDsByRegion), nil
}

func clusterComponentPlansFromRegionMap(clusterType ccent.ClusterType, clusterIDsByRegion map[string][]string) []clusterComponentSearchPlan {
	regions := make([]string, 0, len(clusterIDsByRegion))
	for region := range clusterIDsByRegion {
		regions = append(regions, region)
	}
	sort.Strings(regions)

	plans := make([]clusterComponentSearchPlan, 0, len(regions))
	for _, region := range regions {
		clusterIDs := clusterIDsByRegion[region]
		sort.Strings(clusterIDs)
		plans = append(plans, clusterComponentSearchPlan{
			ClusterType:   clusterType,
			LogicalRegion: region,
			ClusterIDs:    clusterIDs,
		})
	}
	return plans
}

func (s *svc) resolveClusterComponentLogicalRegion(ctx context.Context, clusterType ccent.ClusterType, clusterID string) (string, error) {
	switch clusterType {
	case ccent.ClusterTypeLogicalCluster:
		clusterIDUint, err := strconv.ParseUint(clusterID, 10, 64)
		if err != nil {
			return "", errs.APIErrorf(devopsapi.ErrCodeInvalidArgument, "invalid cluster_id %q: %v", clusterID, err)
		}
		cluster, err := s.coreSvc.Mgmt.GetCluster(ctx, clusterIDUint)
		if err != nil {
			return "", err
		}
		if cluster.GetDeployType() != mgmtv1.DeployType_DEPLOY_TYPE_NEXTGEN {
			return "", errs.APIErrorf(devopsapi.ErrCodeInvalidArgument, "cluster %s is not a logical cluster", clusterID)
		}
		sharedPoolID := cluster.GetNextGenCluster().GetSharedPoolId()
		if sharedPoolID == "" {
			return "", fmt.Errorf("shared pool id is empty for logical cluster %s", clusterID)
		}
		return s.logicalRegionByResourcePoolID(ctx, sharedPoolID)
	case ccent.ClusterTypeSharedPool:
		return s.logicalRegionByResourcePoolID(ctx, clusterID)
	default:
		return "", nil
	}
}

func (s *svc) logicalRegionByResourcePoolID(ctx context.Context, resourcePoolID string) (string, error) {
	rp, err := s.coreSvc.Mgmt.GetResourcePoolString(ctx, resourcePoolID)
	if err != nil {
		return "", err
	}
	logicalRegion := rp.GetRegion().GetLogicalName()
	if logicalRegion == "" {
		return "", fmt.Errorf("logical region is empty for resource pool %s", resourcePoolID)
	}
	return logicalRegion, nil
}

func (s *svc) listNextGenLogicalRegions(ctx context.Context) ([]string, error) {
	resp, err := s.coreSvc.Mgmt.Region.ListRegions(ctx, &mgmtv1.ListRegionsRequest{
		DeployType: mgmtv1.DeployType_DEPLOY_TYPE_NEXTGEN,
		Page:       1,
		PageSize:   500,
	})
	if err != nil {
		return nil, err
	}

	seen := make(map[string]struct{})
	regions := make([]string, 0, len(resp.GetRegions()))
	for _, region := range resp.GetRegions() {
		logicalRegion := region.GetRegion().GetLogicalName()
		if logicalRegion == "" {
			continue
		}
		if _, ok := seen[logicalRegion]; ok {
			continue
		}
		seen[logicalRegion] = struct{}{}
		regions = append(regions, logicalRegion)
	}
	sort.Strings(regions)
	return regions, nil
}

func (s *svc) fetchAllClusterComponents(ctx context.Context, plans []clusterComponentSearchPlan, param ccsvcapi.SearchClusterComponentsParam) ([]*ccent.ClusterComponentVersion, error) {
	var mu sync.Mutex
	var firstErr error
	items := make([]*ccent.ClusterComponentVersion, 0)

	wq := workqueue.NewParallelizer(plans, clusterComponentSearchConcurrency, 1)
	wq.Do(ctx, func(ctx context.Context, job []clusterComponentSearchPlan) {
		for _, plan := range job {
			planItems, err := s.fetchAllClusterComponentsByPlan(ctx, plan, param)
			mu.Lock()
			if err != nil && firstErr == nil {
				firstErr = err
			}
			items = append(items, planItems...)
			if len(items) > maxTenantClusterComponentRows && firstErr == nil {
				firstErr = errs.APIErrorf(devopsapi.ErrCodeInvalidArgument, "tenant cluster component search exceeds internal limit %d", maxTenantClusterComponentRows)
			}
			mu.Unlock()
		}
	})
	if firstErr != nil {
		return nil, firstErr
	}
	return items, nil
}

func (s *svc) fetchAllClusterComponentsByPlan(ctx context.Context, plan clusterComponentSearchPlan, param ccsvcapi.SearchClusterComponentsParam) ([]*ccent.ClusterComponentVersion, error) {
	items := make([]*ccent.ClusterComponentVersion, 0)
	for page := int32(1); ; page++ {
		regionResult, err := s.searchClusterComponentsByPlan(ctx, plan, param, page, configCenterClusterComponentPageSize)
		if err != nil {
			return nil, err
		}
		items = append(items, regionResult.Items...)
		if int64(len(items)) >= regionResult.Total || len(regionResult.Items) == 0 {
			break
		}
	}
	return items, nil
}

func (s *svc) searchAllRegionClusterComponents(ctx context.Context, plans []clusterComponentSearchPlan, param ccsvcapi.SearchClusterComponentsParam) (*ccent.ClusterComponentSearchResult, error) {
	if len(plans) == 0 {
		return &ccent.ClusterComponentSearchResult{
			RegionResults: []*ccent.ClusterComponentSearchRegionResult{},
		}, nil
	}

	var mu sync.Mutex
	var firstErr error
	total := int64(0)
	failed := 0
	regionResults := make([]*ccent.ClusterComponentSearchRegionResult, 0, len(plans))

	wq := workqueue.NewParallelizer(plans, clusterComponentSearchConcurrency, 1)
	wq.Do(ctx, func(ctx context.Context, job []clusterComponentSearchPlan) {
		for _, plan := range job {
			regionResult, err := s.searchClusterComponentsByPlan(ctx, plan, param, defaultClusterComponentSearchPage, param.PerPage)
			mu.Lock()
			if err != nil {
				failed++
				if firstErr == nil {
					firstErr = err
				}
				regionResults = append(regionResults, &ccent.ClusterComponentSearchRegionResult{
					LogicalRegion: plan.LogicalRegion,
					ClusterType:   plan.ClusterType,
					Items:         []*ccent.ClusterComponentVersion{},
					Error:         err.Error(),
				})
			} else {
				total += regionResult.Total
				regionResults = append(regionResults, regionResult)
			}
			mu.Unlock()
		}
	})

	sort.Slice(regionResults, func(i, j int) bool {
		return regionResults[i].LogicalRegion < regionResults[j].LogicalRegion
	})
	if failed == len(plans) {
		return nil, fmt.Errorf("all regional config center searches failed: %w", firstErr)
	}
	return &ccent.ClusterComponentSearchResult{
		Total:          total,
		RegionResults:  regionResults,
		PartialFailure: failed > 0,
	}, nil
}

func (s *svc) searchClusterComponentsByPlan(ctx context.Context, plan clusterComponentSearchPlan, param ccsvcapi.SearchClusterComponentsParam, page, perPage int32) (*ccent.ClusterComponentSearchRegionResult, error) {
	req := &ccproto.SearchClusterComponentsRequest{
		ClusterType:   cvt.To.ConfigCenter.ClusterType(plan.ClusterType),
		ComponentName: param.ComponentName,
		ClusterIds:    plan.ClusterIDs,
		PageSize:      perPage,
		Page:          page,
	}
	if param.ComponentVersion != "" {
		req.ComponentVersion = param.ComponentVersion
	}
	if param.ImageTag != "" {
		req.ImageTag = param.ImageTag
	}
	if plan.LogicalRegion != "" {
		req.LogicalRegion = &plan.LogicalRegion
	}

	reqCtx, cancel := context.WithTimeout(ctx, configCenterSearchClusterComponentsTimeout)
	defer cancel()

	resp, err := s.cfgCenter.ConfigAPI.SearchClusterComponents(reqCtx, req)
	if err != nil {
		return nil, err
	}
	items := clusterComponentVersionsFromConfigCenter(resp.GetComponents(), plan)
	hasMore := hasMoreClusterComponentPage(page, perPage, resp.GetTotal())
	result := &ccent.ClusterComponentSearchRegionResult{
		LogicalRegion: plan.LogicalRegion,
		ClusterType:   plan.ClusterType,
		Total:         resp.GetTotal(),
		HasMore:       hasMore,
		Items:         items,
	}
	return result, nil
}

func clusterComponentVersionsFromConfigCenter(components []*ccproto.ClusterComponent, plan clusterComponentSearchPlan) []*ccent.ClusterComponentVersion {
	items := make([]*ccent.ClusterComponentVersion, 0, len(components))
	for _, comp := range components {
		clusterType := cvt.From.ConfigCenter.ClusterType(comp.GetClusterType())
		if clusterType == ccent.ClusterTypeUnknown {
			clusterType = plan.ClusterType
		}
		items = append(items, &ccent.ClusterComponentVersion{
			ClusterType:       clusterType,
			ClusterID:         comp.GetClusterId(),
			ComponentName:     comp.GetComponentName(),
			Version:           comp.GetVersion(),
			CloudProvider:     comp.GetCloudProvider(),
			NodeInstance:      comp.GetNodeInstance(),
			OverwriteImageTag: comp.GetOverwriteImageTag(),
			BaseImage:         comp.GetBaseImage(),
			EffectiveImageTag: comp.GetEffectiveImageTag(),
			UpdateStrategy:    comp.GetUpdateStrategy().String(),
			LogicalRegion:     plan.LogicalRegion,
		})
	}
	return items
}

func isRegionalClusterType(clusterType ccent.ClusterType) bool {
	return cvt.To.ConfigCenter.ClusterType(clusterType).UseRegionalMode()
}

func sortClusterComponentVersions(items []*ccent.ClusterComponentVersion) {
	sort.SliceStable(items, func(i, j int) bool {
		left := items[i]
		right := items[j]
		if left.ClusterType != right.ClusterType {
			return left.ClusterType < right.ClusterType
		}
		if left.LogicalRegion != right.LogicalRegion {
			return left.LogicalRegion < right.LogicalRegion
		}
		if left.ClusterID != right.ClusterID {
			return left.ClusterID < right.ClusterID
		}
		if left.ComponentName != right.ComponentName {
			return left.ComponentName < right.ComponentName
		}
		if left.Version != right.Version {
			return left.Version < right.Version
		}
		if left.EffectiveImageTag != right.EffectiveImageTag {
			return left.EffectiveImageTag < right.EffectiveImageTag
		}
		if left.OverwriteImageTag != right.OverwriteImageTag {
			return left.OverwriteImageTag < right.OverwriteImageTag
		}
		if left.BaseImage != right.BaseImage {
			return left.BaseImage < right.BaseImage
		}
		if left.CloudProvider != right.CloudProvider {
			return left.CloudProvider < right.CloudProvider
		}
		if left.NodeInstance != right.NodeInstance {
			return left.NodeInstance < right.NodeInstance
		}
		return left.UpdateStrategy < right.UpdateStrategy
	})
}

func hasMoreClusterComponentPage(page, perPage int32, total int64) bool {
	if page <= 0 || perPage <= 0 || total <= 0 {
		return false
	}
	return int64(page)*int64(perPage) < total
}

func paginateClusterComponentVersions(items []*ccent.ClusterComponentVersion, page, perPage int32) []*ccent.ClusterComponentVersion {
	if len(items) == 0 || page <= 0 || perPage <= 0 {
		return []*ccent.ClusterComponentVersion{}
	}
	start := (int64(page) - 1) * int64(perPage)
	if start >= int64(len(items)) {
		return []*ccent.ClusterComponentVersion{}
	}
	end := start + int64(perPage)
	if end > int64(len(items)) {
		end = int64(len(items))
	}
	return items[int(start):int(end)]
}