package cluster

import (
	"context"
	"fmt"
	"strconv"
	"strings"

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

	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/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/serverless"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/service/workflow"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/common"
)

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

	// filter is nil, use default filter
	if filter == nil {
		filter = &repository.ClusterFilter{
			View: entity.ClusterViewClusterViewBasic,
		}
	}

	//  make sure deploy type is serverless
	filter.DeployType = entity.DeployTypeServerless

	if page == nil {
		page = &common.Page{
			Offset:   0,
			PageSize: int64(common.ListMaxPageSize),
		}
	}

	var clusters []*entity.Cluster

	tranXFunc := func(ctx context.Context) error {
		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 serverless 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
		}

		// build global cluster filter
		globalClusterFilter := &globalserverlessv1.ClusterFilter{
			ClusterIds: lo.FlatMap(baseClusters, func(item *entity.Cluster, _ int) []string {
				return []string{strconv.FormatUint(item.ClusterId, 10)}
			}),
			States: lo.FlatMap(
				filter.States,
				func(state openapicommonv1beta1.ClusterState, _ int) []regionalserverlessv1.ClusterState {
					regionState := cvtServerlessState(state)
					if regionState == regionalserverlessv1.ClusterState_CLUSTER_STATE_UNSPECIFIED {
						return []regionalserverlessv1.ClusterState{}
					}
					return []regionalserverlessv1.ClusterState{regionState}
				},
			),
			Deleted:      &filter.QueryDeleted,
			ShowDetails:  filter.View == entity.ClusterViewClusterViewFull,
			RefreshCache: filter.RefreshCache,
		}

		// Then get cluster from global serverless service
		resp := &globalserverlessv1.ServerlessServiceListResponse{
			Items: []*globalserverlessv1.ServerlessServiceCluster{},
			Total: 0,
		}

		// skip if cluster id
		// TODO: concurrent
		for i := 0; i < len(globalClusterFilter.ClusterIds); i += int(common.ListMaxPageSize) {
			// fetch cluster ids in batch
			clusterIds := globalClusterFilter.ClusterIds[i:min(i+int(common.ListMaxPageSize), len(globalClusterFilter.ClusterIds))]
			_resp, err := svc.serverlessCli.ListClusters(ctx, &globalserverlessv1.ServerlessServiceListRequest{
				Filter: &globalserverlessv1.ClusterFilter{
					ClusterIds:   clusterIds,
					States:       globalClusterFilter.States,
					Deleted:      globalClusterFilter.Deleted,
					ShowDetails:  globalClusterFilter.ShowDetails,
					RefreshCache: globalClusterFilter.RefreshCache,
				},
			})
			if err != nil {
				return err
			}
			resp.Items = append(resp.Items, _resp.Items...)
			resp.Total += _resp.Total
		}

		clusterIDMap := lo.KeyBy(resp.Items, func(c *globalserverlessv1.ServerlessServiceCluster) string {
			return c.ClusterId
		})

		// Enrich the response with db records.
		for _, baseCluster := range baseClusters {
			strId := strconv.FormatUint(baseCluster.ClusterId, 10)
			var out *entity.Cluster
			serverlessCluster, ok := clusterIDMap[strId]
			if !ok {
				runningDagInstances, err := svc.workflow.ListRunningDagInstances(ctx, []string{
					constant.DagServerlessCreateCluster,
					constant.DagServerlessClaimClusterFromZero,
				}, 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,
						cvtServerlessCluster(baseCluster, defaultServerlessCluster(ctx, baseCluster)),
						runningDagInstances,
					)
				}
				// if cluster not found in global dedicated service, and no running dag instance, ignore it
			} else {
				out = cvtServerlessCluster(baseCluster, serverlessCluster)

				// If the claim DAG is still running (data migration phase), keep the cluster
				// in CREATING state so that user-facing operations are rejected.
				runningDagInstances, dagErr := svc.workflow.ListRunningDagInstances(ctx, []string{
					constant.DagServerlessClaimClusterFromZero,
				}, fmt.Sprintf("%d", baseCluster.ClusterId))
				if dagErr != nil {
					tilog.Error(ctx, "list running claim dag instances failed", zap.Error(dagErr))
				}
				for _, inst := range runningDagInstances {
					if inst.DagID == constant.DagServerlessClaimClusterFromZero {
						sc, scErr := out.GetServerlessCluster()
						if scErr == nil && sc != nil {
							sc.UserStatus = globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING
							out = cvtServerlessCluster(out, sc)
						}
						break
					}
				}
			}
			if out != nil {
				if out.Region == "" && serverlessCluster.GetRegion() != "" {
					if err := svc.clusterRepo.SetClusterRegion(
						ctx,
						out.ClusterId,
						serverlessCluster.GetCloudProvider(),
						serverlessCluster.GetRegion(),
					); err != nil {
						return err
					}
				}
				clusters = append(clusters, out)
			}
		}

		return nil
	}

	if err := svc.transaction.Transaction(ctx, tranXFunc); err != nil {
		return clusters, err
	}

	return clusters, nil
}

func defaultServerlessCluster(ctx context.Context, c *entity.Cluster) *globalserverlessv1.ServerlessServiceCluster {
	serverlessCluster, err := c.GetServerlessCluster()
	if err != nil {
		tilog.Warn(ctx, "failed to get serverless cluster", zap.Error(err))
		serverlessCluster = nil
	}

	return &globalserverlessv1.ServerlessServiceCluster{
		ClusterId: strconv.FormatUint(c.ClusterId, 10),
		Cluster: &regionalserverlessv1.ServerlessRegionalServiceClusterInfo{
			ClusterId: strconv.FormatUint(c.ClusterId, 10),
			TenantId:  c.OrganizationId,
			ProjectId: c.ProjectId,
			State:     regionalserverlessv1.ClusterState_CLUSTER_STATE_CREATING,
			Region: &regionalserverlessv1.Region{
				Name:          c.Region,
				DisplayName:   c.Region,
				CloudProvider: cvtRegionalCloudProvider(c.CloudProvider.String()),
			},
			TokenBasedAuthEnabled:     true,
			EnhancedEncryptionEnabled: serverlessCluster.GetCluster().GetEnhancedEncryptionEnabled(),
			ZoneRedundantEnabled:      serverlessCluster.GetCluster().GetZoneRedundantEnabled(),
			AutoScaling:               serverlessCluster.GetCluster().GetAutoScaling(),
		},
		TenantId:      c.OrganizationId,
		ProjectId:     c.ProjectId,
		HasSetPwd:     serverlessCluster.GetRootPassword() != "",
		Region:        c.Region,
		CloudProvider: c.CloudProvider.String(),
		UserStatus:    globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING,
		BackupPolicy:  serverlessCluster.GetBackupPolicy(),
		SpendLimit:    serverlessCluster.GetSpendLimit(),
		EngineType: lo.Ternary(
			c.EngineType == openapicommonv1beta1.EngineType_ENGINE_TYPE_UNSPECIFIED,
			openapicommonv1beta1.EngineType_MYSQL_COMPATIBLE.String(),
			c.EngineType.String(),
		),
	}
}

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.DagServerlessCreateCluster ||
		ins.DagID == constant.DagServerlessClaimClusterFromZero {
		out := workflow.Unmarshal(ctx, cluster.ClusterId, ins, &globalserverlessv1.ServerlessServiceCreateRequest{})
		if out != nil {
			createReq, ok := out.(*globalserverlessv1.ServerlessServiceCreateRequest)
			if !ok {
				tilog.Error(ctx, "failed to convert to ServerlessServiceCreateRequest", zap.Any("out", out))
				return cluster
			}

			globalCluster := &globalserverlessv1.ServerlessServiceCluster{
				ClusterId: strconv.FormatUint(cluster.ClusterId, 10),
				Cluster: &regionalserverlessv1.ServerlessRegionalServiceClusterInfo{
					ClusterId: strconv.FormatUint(cluster.ClusterId, 10),
					TenantId:  cluster.OrganizationId,
					ProjectId: cluster.ProjectId,
					State:     regionalserverlessv1.ClusterState_CLUSTER_STATE_CREATING,
					Region: &regionalserverlessv1.Region{
						DisplayName:   createReq.GetRegion(),
						Name:          createReq.GetRegion(),
						CloudProvider: cvtRegionalCloudProvider(createReq.GetCloudProvider()),
					},
					EnhancedEncryptionEnabled: createReq.GetEncryptionConfig().GetEnhancedEncryptionEnabled(),
					ZoneRedundantEnabled:      createReq.GetZoneRedundantEnabled(),
					AutoScaling:               createReq.GetAutoScaling(),
				},
				TenantId:      cluster.OrganizationId,
				ProjectId:     cluster.ProjectId,
				HasSetPwd:     createReq.GetRootPassword() != "",
				Region:        createReq.GetRegion(),
				CloudProvider: createReq.GetCloudProvider(),
				UserStatus:    globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING,
				SpendLimit:    createReq.GetSpendLimit(),
				BackupPolicy:  createReq.GetBackupPolicy(),
				EngineType: lo.Ternary(
					createReq.GetEngineType() == "",
					openapicommonv1beta1.EngineType_MYSQL_COMPATIBLE.String(),
					createReq.GetEngineType(),
				),
			}
			cluster.GlobalCluster = globalCluster
		}
	}
	return cluster
}

func cvtRegionalCloudProvider(provider string) regionalserverlessv1.CloudProvider {
	switch strings.ToLower(provider) {
	case "aws":
		return regionalserverlessv1.CloudProvider_CLOUD_PROVIDER_AWS
	case "gcp":
		return regionalserverlessv1.CloudProvider_CLOUD_PROVIDER_GCP
	case "azure":
		return regionalserverlessv1.CloudProvider_CLOUD_PROVIDER_AZURE
	case "alicloud":
		return regionalserverlessv1.CloudProvider_CLOUD_PROVIDER_ALICLOUD
	default:
		return regionalserverlessv1.CloudProvider_CLOUD_PROVIDER_UNSPECIFIED
	}
}