package cluster

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

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

	bffclusterv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/bff/cluster_management/v1beta1"
	openapidedicatedv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/dedicated/v1beta1"
	openapiserverlessv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/serverless/v1beta1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/app/bff/orchestrator"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

func (s *Server) GetCluster(
	ctx context.Context,
	request *bffclusterv1beta1.GetClusterRequest,
) (*bffclusterv1beta1.Cluster, error) {
	// Extract request meta.
	_, err := metadata.ExtractRequestMetaWithCheckPermission(ctx)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "failed to extract request metadata")
	}

	// check if the request is valid
	if request.ClusterId == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}

	clusterId, err := strconv.ParseUint(request.ClusterId, 10, 64)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid cluster_id")
	}

	// get the cluster from repo
	cls, err := s.clusterRepo.GetCluster(ctx, clusterId, false)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster not found")
		}
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "failed to get cluster")
	}

	switch cls.DeployType {
	case entity.DeployTypeServerless:
		// first try to get the cluster from serverless
		serverlessCluster, err := s.openAPIServerlessClusterServer.GetCluster(
			ctx,
			&openapiserverlessv1beta1.GetClusterRequest{
				ClusterId: request.GetClusterId(),
			},
		)
		if err != nil {
			return nil, bizerr.WrapGrpcError(err, "failed to get serverless cluster")
		}

		return &bffclusterv1beta1.Cluster{
			ClusterType: bffclusterv1beta1.Cluster_SERVERLESS,
			Cluster:     &bffclusterv1beta1.Cluster_ServerlessCluster{ServerlessCluster: serverlessCluster},
		}, nil

	case entity.DeployTypeDedicated:
		// if the cluster is not found in serverless, try to get it from dedicated
		dedicatedCluster, err := s.openAPIDedicatedClusterServer.GetCluster(
			ctx,
			&openapidedicatedv1beta1.GetClusterRequest{
				ClusterId: request.GetClusterId(),
			},
		)
		if err != nil {
			return nil, bizerr.WrapGrpcError(err, "failed to get dedicated cluster")
		}

		// merge in workflow state
		dagIns, err := s.orchestrator.ListRunningDagInstancesByNamePrefix(
			ctx,
			fmt.Sprintf("%s_%s_%s", workflow.DynamicDagPrefix, orchestrator.DagConsoleAPIScale, request.GetClusterId()),
			request.GetClusterId())

		// if list running dag instances failed, log the error and continue
		if err != nil {
			tilog.Error(ctx, "list running dag instances failed", zap.Error(err))
		}
		dedicatedCluster = s.mergeInWorkflowState(ctx, dedicatedCluster, dagIns)

		return &bffclusterv1beta1.Cluster{
			ClusterType: bffclusterv1beta1.Cluster_DEDICATED,
			Cluster:     &bffclusterv1beta1.Cluster_DedicatedCluster{DedicatedCluster: dedicatedCluster},
		}, nil
	}

	return nil, bizerr.New(bizerr.ErrCodeUnknown, "unknown cluster type")
}