package cluster

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

	"github.com/samber/lo"
	"google.golang.org/protobuf/encoding/protojson"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/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/pkg/apimiddleware/bizerror"
	"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 (svc *service) CreateCluster(
	ctx context.Context,
	cluster *entity.Cluster,
	clusterConfiguration *globaldedicatedv1.ClusterConfiguration,
	validateOnly bool,
) (*entity.Cluster, error) {
	spanName := "ClusterService.CreateCluster"
	ctx, span := tracer.Start(ctx, spanName)
	defer span.End()

	if cluster == nil {
		return nil, fmt.Errorf("cluster is nil")
	}

	if cluster.DeployType != entity.DeployTypeDedicated {
		return nil, fmt.Errorf("cluster deploy type is not dedicated")
	}

	// check third party account provider
	thirdProviders, err := svc.accountCli.GetTenantThirdPartyAccountProviders(ctx, cluster.OrganizationId)
	if err != nil {
		return nil, bizerr.WrapGrpcError(err, "failed to get tenant third party account providers")
	}
	if len(thirdProviders) > 0 && !lo.Contains(thirdProviders, cluster.CloudProvider) {
		return nil, bizerr.New(bizerr.ErrCodePermission, bizerr.ErrMsg3rdPartyAccount, cluster.CloudProvider)
	}

	// transaction
	if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		// Check cluster name is unique under the project
		count, err := svc.clusterRepo.CountDisplayNameUnderProject(ctx, cluster.ProjectId, cluster.DisplayName)
		if err != nil {
			return err
		}
		if count > 0 {
			return constant.ErrDuplicatedClusterName
		}

		cluster.ClusterId = entity.GenClusterID()

		// check create cluster in global dedicated service
		request, err := buildCreateRequest(cluster, clusterConfiguration)
		if err != nil {
			return err
		}

		// only validate
		request.ValidateOnly = true
		dedicatedCluster, err := svc.dedicatedCli.CreateCluster(ctx, request)
		if err != nil {
			return err
		}

		if dedicatedCluster == nil {
			return errors.New("cluster in global create response should not be nil")
		}

		// if validate only, return temporary cluster info
		if validateOnly {
			return nil
		}

		// First persist cluster info to database
		if _, err = svc.clusterRepo.InsertNewCluster(ctx, cluster); err != nil {
			return fmt.Errorf("failed to insert cluster record: %w", err)
		}

		if err = svc.accountCli.UpsertClusterRelations(ctx, cluster.ClusterId, cluster.OrganizationId, cluster.ProjectId); err != nil {
			return fmt.Errorf("failed to init cluster permissions: %w", err)
		}

		// run dag instance
		// encrypt root password
		if request.Cluster.RootPassword != "" {
			encryptedRootPassword, err := svc.encryptor.EncryptBase64(ctx, request.Cluster.RootPassword)
			if err != nil {
				return fmt.Errorf("failed to encrypt root password: %w", err)
			}
			request.Cluster.RootPassword = encryptedRootPassword
		}

		requestJson, err := protojson.Marshal(request)
		if err != nil {
			return fmt.Errorf("failed to marshal create cluster request: %w", err)
		}
		// metadata
		metadataJson, err := metadata.MarshalRequestMetaFromCtx(ctx)
		if err != nil {
			return fmt.Errorf("failed to marshal request metadata: %w", err)
		}

		err = svc.workflow.RunDagWithBizId(
			ctx,
			constant.DagDedicatedCreateCluster,
			strconv.FormatUint(cluster.OrganizationId, 10),
			strconv.FormatUint(cluster.ProjectId, 10),
			fmt.Sprintf("%d", cluster.ClusterId),
			map[string]string{
				constant.KeyRequest:     string(requestJson),
				workflow.KeyTraceId:     workflow.GetTraceId(ctx),
				workflow.KeySpanName:    spanName,
				workflow.KeyRequestMeta: string(metadataJson),
			},
		)
		if err != nil {
			return fmt.Errorf("failed to run create dedicated cluster dag task: %w", err)
		}

		// Finally, return cluster info
		cluster, err = svc.clusterRepo.GetCluster(ctx, cluster.ClusterId, false)
		if err != nil {
			return fmt.Errorf("failed to get dedicated cluster record: %w", err)
		}

		cluster = cvtDedicatedCluster(cluster, dedicatedCluster)

		return nil
	}); err != nil {
		return cluster, err
	}

	return cluster, nil
}

func buildCreateRequest(
	cluster *entity.Cluster,
	clusterConfiguration *globaldedicatedv1.ClusterConfiguration,
) (*globaldedicatedv1.CreateClusterRequest, error) {
	dedicatedCluster, err := cluster.GetDedicatedCluster()
	if err != nil {
		return nil, err
	}

	if len(dedicatedCluster.GetClusterComponents()) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster components is nil")
	}

	for _, component := range dedicatedCluster.GetClusterComponents() {
		if component.GetCount() <= 0 {
			err := fmt.Errorf("%s component count is invalid", component.ComponentType)
			return nil, bizerror.NewError(err, bizerr.ErrCodeInvalidNodeQuantity, err.Error())
		}

		if component.GetSpuName() == "" {
			err := fmt.Errorf("%s component tier is invalid", component.ComponentType)
			return nil, bizerror.NewError(err, bizerr.ErrCodeInvalidNodeSize, err.Error())
		}

		if component.GetTidbGroupId() != 0 && component.GetComponentType() == globaldedicatedv1.ComponentType_TIDB {
			err := errors.New("unable to set tidb node group id when creating cluster")
			return nil, bizerror.NewError(err, bizerr.ErrCodeParams, err.Error())
		}

		if component.GetComponentType() != globaldedicatedv1.ComponentType_TIDB &&
			component.GetComponentType() != globaldedicatedv1.ComponentType_TIPROXY {
			if component.GetStorages() == nil ||
				component.GetStorages()[globaldedicatedv1.Sku_DATA.String()].GetSizeGi() <= 0 {
				err := fmt.Errorf("%s component storage size is invalid", component.ComponentType)
				return nil, bizerror.NewError(err, bizerr.ErrCodeInvalidStorageSize, err.Error())
			}
		}
	}

	// TODO add cmek by @shirui
	createReq := &globaldedicatedv1.CreateClusterRequest{
		Cluster: &globaldedicatedv1.Cluster{
			ClusterId:                      cluster.ClusterId,
			OrganizationId:                 cluster.OrganizationId,
			ProjectId:                      cluster.ProjectId,
			CloudProviderId:                dedicatedCluster.CloudProviderId,
			RegionId:                       dedicatedCluster.RegionId,
			RootPassword:                   dedicatedCluster.RootPassword,
			Port:                           dedicatedCluster.Port,
			Version:                        dedicatedCluster.Version,
			CustomerManagedEncryptionKeyId: dedicatedCluster.CustomerManagedEncryptionKeyId,
		},
	}

	createReq.Cluster.ClusterComponents = dedicatedCluster.ClusterComponents
	createReq.ClusterConfiguration = clusterConfiguration

	return createReq, nil
}