package cluster

import (
	"context"
	"fmt"

	"github.com/samber/lo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	openapicommonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/constant"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/repository"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/common"
	"github.com/tidbcloud/tidb-management-service/utilities/enum"
)

const listDefaultPageSize = 100

const stateFilterBatchSize = 100

func (svc *service) ListClusters(
	ctx context.Context,
	filter *repository.ClusterFilter,
	page *common.Page,
) ([]*entity.Cluster, error) {
	ctx, span := tracer.Start(ctx, "ClusterService.ListClusters")
	defer span.End()

	if filter == nil {
		filter = &repository.ClusterFilter{
			View: entity.ClusterViewClusterViewBasic,
		}
	}

	// make sure deploy type is nextgen
	filter.DeployType = entity.DeployTypeNextGen

	if page == nil {
		page = &common.Page{
			Offset:   0,
			PageSize: listDefaultPageSize,
		}
	}

	var clusters []*entity.Cluster
	// Pool lifecycle states are projected by Global Service after the database
	// query, so state-filtered pagination must scan candidates in batches and
	// apply pagination after the projected-state filter.
	postFilterState := hasClusterStateFilter(filter) && filter.View != entity.ClusterViewClusterViewLocal
	if postFilterState {
		var err error
		clusters, err = svc.listClustersWithStateFilter(ctx, filter, page)
		if err != nil {
			return nil, fmt.Errorf("failed to list clusters: %w", err)
		}
		return clusters, nil
	}

	if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		// query database
		baseClusters, err := svc.clusterRepo.QueryClusterByFilterAndPage(ctx, filter, int(page.Offset), int(page.PageSize))
		if err != nil {
			return fmt.Errorf("failed to query clusters from database: %w", err)
		}

		// not need to query global nextgen service if no cluster found in database
		if len(baseClusters) == 0 {
			return nil
		}

		// if view is local, return directly
		if filter.View == entity.ClusterViewClusterViewLocal {
			if err := svc.hydrateClusterTags(ctx, baseClusters); err != nil {
				return fmt.Errorf("failed to list cluster tags: %w", err)
			}
			clusters = baseClusters
			return nil
		}
		clusters, err = svc.enrichNextGenClusterBatch(ctx, filter, baseClusters, false, true)
		if err != nil {
			return err
		}

		return nil
	}); err != nil {
		return nil, fmt.Errorf("failed to list clusters: %w", err)
	}

	return clusters, nil
}

func (svc *service) listClustersWithStateFilter(
	ctx context.Context,
	filter *repository.ClusterFilter,
	page *common.Page,
) ([]*entity.Cluster, error) {
	if page == nil {
		return nil, fmt.Errorf("page cannot be nil")
	}

	clusters := make([]*entity.Cluster, 0)
	remainingOffset := page.Offset
	for queryOffset := int64(0); ; {
		baseClusters, err := svc.clusterRepo.QueryClusterByFilterAndPage(
			ctx,
			filter,
			int(queryOffset),
			stateFilterBatchSize,
		)
		if err != nil {
			return nil, fmt.Errorf("failed to query clusters from database: %w", err)
		}
		if len(baseClusters) == 0 {
			break
		}

		filteredClusters, err := svc.enrichNextGenClusterBatch(ctx, filter, baseClusters, true, true)
		if err != nil {
			return nil, err
		}
		for _, cluster := range filteredClusters {
			if remainingOffset > 0 {
				remainingOffset--
				continue
			}
			if page.PageSize > 0 && int64(len(clusters)) >= page.PageSize {
				break
			}
			clusters = append(clusters, cluster)
		}

		if page.PageSize > 0 && int64(len(clusters)) >= page.PageSize {
			break
		}
		if len(baseClusters) < stateFilterBatchSize {
			break
		}
		queryOffset += int64(len(baseClusters))
	}

	return clusters, nil
}

func (svc *service) enrichNextGenClusterBatch(
	ctx context.Context,
	filter *repository.ClusterFilter,
	baseClusters []*entity.Cluster,
	postFilterState bool,
	hydrateTags bool,
) ([]*entity.Cluster, error) {
	if hydrateTags {
		if err := svc.hydrateClusterTags(ctx, baseClusters); err != nil {
			return nil, fmt.Errorf("failed to list cluster tags: %w", err)
		}
	}

	req := svc.buildGlobalListRequest(ctx, filter, lo.Map(baseClusters, func(item *entity.Cluster, _ int) uint64 {
		return item.ClusterId
	}))
	resp, err := svc.nextGenCli.ListClusters(ctx, req)
	if err != nil {
		return nil, err
	}

	clusterIDMap := lo.KeyBy(resp.Clusters, func(c *globalngv1.Cluster) uint64 {
		return c.ClusterId
	})
	clusters := make([]*entity.Cluster, 0, len(baseClusters))
	for _, baseCluster := range baseClusters {
		nextgenCluster, ok := clusterIDMap[baseCluster.ClusterId]
		var out *entity.Cluster
		if !ok {
			// if cluster not found in global nextgen service, check if there is running dag instance
			tilog.Warn(ctx, "cluster not found in nextgen global", zap.Uint64("cluster_id", baseCluster.ClusterId), zap.Any("filter", filter))
			// Only use the workflow fallback when the requested filter can match
			// CREATING; a missing Global record must not be guessed as a Pool
			// lifecycle state such as PAUSED or RESUMING.
			if svc.workflow != nil && (!postFilterState || matchesNextGenStateFilter(globalngv1.Cluster_CREATING, filter)) {
				runningDagInstances, err := svc.workflow.ListRunningDagInstances(ctx, []string{
					constant.DagNextGenCreateCluster,
				}, fmt.Sprintf("%d", baseCluster.ClusterId))
				if err != nil {
					tilog.Error(ctx, "list running dag instances failed", zap.Error(err))
				}
				tilog.Warn(
					ctx,
					"get runningDagInstances",
					zap.Uint64("cluster_id", baseCluster.ClusterId),
					zap.Any("runningDagInstances", runningDagInstances),
				)
				if len(runningDagInstances) != 0 {
					out = svc.mergeInWorkflowState(ctx, cvtNextGenCluster(baseCluster, defaultNextGenCluster(baseCluster)), runningDagInstances)
				}
			}
			// if cluster not found in nextgen global, and no running dag instance, ignore it
		} else {
			out = cvtNextGenCluster(baseCluster, nextgenCluster)
		}
		if out != nil && (!postFilterState || matchesClusterStateFilter(out, filter)) {
			clusters = append(clusters, out)
		}
	}

	return clusters, nil
}

func cvtClusterStates(states []openapicommonv1beta1.ClusterState) []globalngv1.Cluster_State {
	return lo.FlatMap(states, func(state openapicommonv1beta1.ClusterState, _ int) []globalngv1.Cluster_State {
		globalState := cvtNextGenState(state)
		if globalState == globalngv1.Cluster_STATE_UNSPECIFIED {
			return []globalngv1.Cluster_State{}
		}
		return []globalngv1.Cluster_State{globalState}
	})
}

// cvt mgmt state to nextgen global state
// global ClusterState_MODIFYING -> config change, upgrade version
// global Cluster_SCALING -> cluster scale
func cvtNextGenState(in openapicommonv1beta1.ClusterState) globalngv1.Cluster_State {
	switch in {
	case openapicommonv1beta1.ClusterState_CREATING:
		return globalngv1.Cluster_CREATING
	case openapicommonv1beta1.ClusterState_ACTIVE,
		openapicommonv1beta1.ClusterState_RESTORING,
		openapicommonv1beta1.ClusterState_MAINTENANCE,
		openapicommonv1beta1.ClusterState_IMPORTING:
		return globalngv1.Cluster_AVAILABLE
	case openapicommonv1beta1.ClusterState_MODIFYING,
		openapicommonv1beta1.ClusterState_UPGRADING:
		return globalngv1.Cluster_MODIFYING
	case openapicommonv1beta1.ClusterState_DELETING,
		openapicommonv1beta1.ClusterState_DELETED:
		return globalngv1.Cluster_DELETED
	case openapicommonv1beta1.ClusterState_PAUSING:
		return globalngv1.Cluster_PAUSING
	case openapicommonv1beta1.ClusterState_PAUSED:
		return globalngv1.Cluster_PAUSED
	case openapicommonv1beta1.ClusterState_RESUMING:
		return globalngv1.Cluster_RESUMING
	}

	return globalngv1.Cluster_STATE_UNSPECIFIED
}

func hasClusterStateFilter(filter *repository.ClusterFilter) bool {
	return filter != nil && (len(filter.States) > 0 || len(filter.ExcludeStates) > 0)
}

func (svc *service) buildGlobalListRequest(
	ctx context.Context,
	filter *repository.ClusterFilter,
	clusterIDs []uint64,
) *globalngv1.ListClustersRequest {
	req := &globalngv1.ListClustersRequest{
		ClusterIds:    clusterIDs,
		States:        cvtClusterStates(filter.States),
		ExcludeStates: cvtClusterStates(filter.ExcludeStates),
		ShowDeleted:   filter.QueryDeleted,
		ShowDetails:   filter.View == entity.ClusterViewClusterViewFull,
	}
	if filter.ServicePlans != nil {
		req.ServicePlans = lo.Map(filter.ServicePlans, func(plan string, _ int) globalcommonv1.ServicePlan {
			servicePlan, err := cvtServicePlan(plan)
			if err != nil {
				tilog.Warn(ctx, "invalid service plan", zap.String("plan", plan), zap.Error(err))
				return globalcommonv1.ServicePlan_SERVICE_PLAN_UNSPECIFIED
			}
			return servicePlan
		})
	}
	return req
}

func matchesClusterStateFilter(cluster *entity.Cluster, filter *repository.ClusterFilter) bool {
	nextgenCluster, ok := cluster.GlobalCluster.(*globalngv1.Cluster)
	if !ok || nextgenCluster == nil {
		return false
	}
	return matchesNextGenStateFilter(nextgenCluster.GetState(), filter)
}

func matchesNextGenStateFilter(state globalngv1.Cluster_State, filter *repository.ClusterFilter) bool {
	states := cvtClusterStates(filter.States)
	if len(states) > 0 && !lo.Contains(states, state) {
		return false
	}
	excludeStates := cvtClusterStates(filter.ExcludeStates)
	return !lo.Contains(excludeStates, state)
}

func cvtServicePlan(in string) (globalcommonv1.ServicePlan, error) {
	plan, err := enum.FromString[globalcommonv1.ServicePlan](in)
	if err != nil {
		return globalcommonv1.ServicePlan_SERVICE_PLAN_UNSPECIFIED, err
	}
	return globalcommonv1.ServicePlan(plan), nil
}