package cluster

import (
	"context"
	"errors"
	"fmt"
	"strconv"

	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"gorm.io/gorm"

	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	"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/service/serverless"
)

func (svc *service) GetCluster(
	ctx context.Context,
	clusterId uint64,
	view entity.ClusterView,
) (*entity.Cluster, error) {
	ctx, span := serverless.Tracer.Start(ctx, "ClusterService.GetCluster")
	defer span.End()

	var cluster *entity.Cluster

	if clusterId == 0 {
		return nil, errors.New("cluster id is invalid")
	}

	tranXFunc := func(ctx context.Context) error {
		var err error
		// First get cluster info from database
		cluster, err = svc.clusterRepo.GetCluster(ctx, clusterId, false)
		if err != nil {
			return fmt.Errorf("failed to find cluste in database: %w", err)
		}

		if cluster.DeployType != entity.DeployTypeServerless {
			tilog.Warn(ctx, "cluster deploy type is not serverless", zap.Uint64("cluster_id", clusterId))
			return gorm.ErrRecordNotFound
		}

		if err = svc.hydrateClusterTags(ctx, []*entity.Cluster{cluster}); err != nil {
			return fmt.Errorf("failed to get cluster tags: %w", err)
		}

		if view == entity.ClusterViewClusterViewLocal {
			return nil
		}

		// Then get cluster from global serverless service
		listFilter := &globalserverlessv1.ClusterFilter{
			TenantIds:   []uint64{cluster.OrganizationId},
			ClusterIds:  []string{strconv.FormatUint(clusterId, 10)},
			ShowDetails: view == entity.ClusterViewClusterViewFull,
		}
		if cluster.OriginalProjectId != 0 {
			listFilter.OriginalProjectIds = []uint64{cluster.OriginalProjectId}
		} else {
			listFilter.ProjectIds = []uint64{cluster.ProjectId}
		}
		resp, err := svc.serverlessCli.ListClusters(ctx, &globalserverlessv1.ServerlessServiceListRequest{
			Filter: listFilter,
		})
		if err != nil {
			return err
		}

		runningDagInstances, err := svc.workflow.ListRunningDagInstances(ctx, []string{
			constant.DagServerlessCreateCluster,
			constant.DagServerlessClaimClusterFromZero,
		}, fmt.Sprintf("%d", clusterId))
		if err != nil {
			tilog.Error(ctx, "list running dag instances failed", zap.Error(err))
		}

		if len(resp.Items) == 0 {
			if len(runningDagInstances) == 0 {
				return fmt.Errorf("cluster not found in serverless global service")
			}

			tilog.Warn(
				ctx,
				"serverless cluster not found, but running dag instance found",
				zap.Uint64("cluster_id", clusterId),
			)

			// return cluster with mgmt workflow cluster info
			cluster = mergeInWorkflowState(
				ctx,
				cvtServerlessCluster(cluster, defaultServerlessCluster(ctx, cluster)),
				runningDagInstances,
			)

			return nil
		}

		if len(resp.Items) > 1 {
			return fmt.Errorf("multiple clusters found")
		}

		// In the past, there was no region and cloud provider columns in mgmt db.
		// The columns were added in the later db schema migrations.
		// For compatibility, we need to update the region and cloud provider columns in mgmt db.
		if resp.Items[0].CloudProvider != "" && resp.Items[0].Region != "" && cluster.Region == "" {
			if err := svc.clusterRepo.SetClusterRegion(ctx, clusterId, resp.Items[0].CloudProvider, resp.Items[0].Region); err != nil {
				return err
			}
		}

		cluster = cvtServerlessCluster(cluster, resp.Items[0])

		// If the claim DAG is still running (data migration in progress), keep the cluster
		// in CREATING state so that user-facing operations (update, etc.) are rejected.
		for _, inst := range runningDagInstances {
			if inst.DagID == constant.DagServerlessClaimClusterFromZero {
				sc, scErr := cluster.GetServerlessCluster()
				if scErr == nil && sc != nil {
					sc.UserStatus = globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING
					cluster = cvtServerlessCluster(cluster, sc)
				}
				break
			}
		}

		return nil
	}

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

	return cluster, nil
}