package cluster

import (
	"context"
	"fmt"

	"braces.dev/errtrace"
	"github.com/samber/lo"
	workflowentity "github.com/tidbcloud/fastflow/pkg/entity"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/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/domain/service/workflow"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/common"
)

const listDefaultPageSize = 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 dedicated
	filter.DeployType = entity.DeployTypeDedicated

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

	var clusters []*entity.Cluster

	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 in database: %w", err)
		}

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

		if err := svc.hydrateClusterTags(ctx, baseClusters); err != nil {
			return fmt.Errorf("failed to list cluster tags: %w", err)
		}

		// if view is local, return directly
		if filter.View == entity.ClusterViewClusterViewLocal {
			clusters = baseClusters
			return nil
		}

		// query global dedicated service
		req := &globaldedicatedv1.ListClustersRequest{
			ClusterIds: lo.FlatMap(baseClusters, func(item *entity.Cluster, _ int) []uint64 {
				return []uint64{item.ClusterId}
			}),
			States:        cvtClusterStates(filter.States),
			ExcludeStates: cvtClusterStates(filter.ExcludeStates),
			ShowDeleted:   filter.QueryDeleted,
			ShowDetails:   filter.View == entity.ClusterViewClusterViewFull,
		}
		resp, err := svc.dedicatedCli.ListClusters(ctx, req)
		if err != nil {
			return errtrace.Wrap(err)
		}

		clusterIDMap := lo.KeyBy(resp.Clusters, func(c *globaldedicatedv1.Cluster) uint64 {
			return c.ClusterId
		})

		// Enrich the response with db records.
		for _, baseCluster := range baseClusters {
			dedicatedCluster, ok := clusterIDMap[baseCluster.ClusterId]
			var out *entity.Cluster
			if !ok {
				// if cluster not found in global dedicated service, check if there is running dag instance
				tilog.Warn(ctx, "cluster not found in dedicated global", zap.Uint64("cluster_id", baseCluster.ClusterId), zap.Any("filter", filter))
				runningDagInstances, err := svc.workflow.ListRunningDagInstances(ctx, []string{
					constant.DagDedicatedCreateCluster,
					constant.DagDedicatedRestoreCluster,
				}, 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 = mergeInWorkflowState(ctx, cvtDedicatedCluster(baseCluster, defaultDedicatedCluster(baseCluster)), runningDagInstances)
				}
				// if cluster not found in global dedicated service, and no running dag instance, ignore it
			} else {
				out = cvtDedicatedCluster(baseCluster, dedicatedCluster)
			}
			if out != nil {
				clusters = append(clusters, out)
			}
		}

		return nil
	}); err != nil {
		return clusters, errtrace.Wrap(err)
	}

	return clusters, nil
}

func defaultDedicatedCluster(c *entity.Cluster) *globaldedicatedv1.Cluster {
	return &globaldedicatedv1.Cluster{
		ClusterId:         c.ClusterId,
		OrganizationId:    c.OrganizationId,
		ProjectId:         c.ProjectId,
		State:             globaldedicatedv1.Cluster_CREATING,
		HealthState:       globaldedicatedv1.Cluster_UNHEALTHY,
		RegionDisplayName: c.Region,
	}
}

func mergeInWorkflowState(
	ctx context.Context,
	cluster *entity.Cluster,
	runningDagInstances []*workflowentity.DagInstance,
) *entity.Cluster {
	if len(runningDagInstances) == 0 {
		return cluster
	}

	ins := runningDagInstances[0]

	if ins.DagID == constant.DagDedicatedCreateCluster || ins.DagID == constant.DagDedicatedRestoreCluster {
		out := workflow.Unmarshal(ctx, cluster.ClusterId, ins, &globaldedicatedv1.CreateClusterRequest{})
		if out != nil {
			globalCluster := out.(*globaldedicatedv1.CreateClusterRequest).Cluster
			globalCluster.State = globaldedicatedv1.Cluster_CREATING
			components := make(
				[]*globaldedicatedv1.Cluster_ClusterComponent,
				0,
				len(globalCluster.GetClusterComponents()),
			)
			for _, comp := range globalCluster.GetClusterComponents() {
				comp.IsDefaultTidbGroup = true
				components = append(components, comp)
			}
			globalCluster.RegionDisplayName = globalCluster.RegionId.String()
			globalCluster.ClusterComponents = components
			cluster.GlobalCluster = globalCluster
		}
	}
	return cluster
}

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

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

	return globaldedicatedv1.Cluster_STATE_UNSPECIFIED
}