package cluster

import (
	"context"
	"fmt"
	"strconv"

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

	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/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/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,
	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.DeployTypeNextGen {
		return nil, fmt.Errorf("cluster is not a nextgen cluster")
	}

	// check third party account provider
	thirdProviders, err := svc.accountCli.GetTenantThirdPartyAccountProviders(ctx, cluster.OrganizationId)
	if err != nil {
		return cluster, bizerr.WrapGrpcError(err, "failed to get tenant third party account providers")
	}
	if len(thirdProviders) > 0 && !lo.Contains(thirdProviders, cluster.CloudProvider) {
		return cluster, 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 org
		count, err := svc.clusterRepo.CountDisplayNameUnderOrg(ctx, cluster.OrganizationId, cluster.DisplayName)
		if err != nil {
			return err
		}
		if count > 0 {
			return constant.ErrDuplicatedClusterName
		}

		cluster.ClusterId = entity.GenClusterID()

		// Resolve ru_version from organization config
		orgConfig, err := svc.orgRuConfigRepo.GetByOrganizationId(ctx, cluster.OrganizationId)
		if err != nil {
			return err
		}
		cluster.RuVersion = orgConfig.DefaultRuVersion

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

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

		if nextgenCluster == nil {
			return fmt.Errorf("global cluster object should not be nil")
		}
		// if validate only, return temporary cluster info
		if validateOnly {
			cluster = cvtNextGenCluster(cluster, nextgenCluster)
			return nil
		}

		// First persist cluster info to database
		if _, err = svc.clusterRepo.InsertNewCluster(ctx, cluster); err != nil {
			return fmt.Errorf("failed to insert new cluster: %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.DagNextGenCreateCluster,
			strconv.FormatUint(cluster.OrganizationId, 10),
			"0",
			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 dag for create cluster: %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)
		}

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

		return nil
	}); err != nil {
		return nil, fmt.Errorf("failed to create cluster: %w", err)
	}

	return cluster, nil
}

func buildCreateRequest(
	cluster *entity.Cluster,
) (*globalngv1.CreateClusterRequest, error) {
	nextgenCluster, err := cluster.GetNextGenCluster()
	if err != nil {
		return nil, err
	}

	createReq := &globalngv1.CreateClusterRequest{
		Cluster: &globalngv1.Cluster{
			ClusterId:                  cluster.ClusterId,
			OrganizationId:             cluster.OrganizationId,
			ProjectId:                  int64(cluster.ProjectId),
			DisplayName:                cluster.DisplayName,
			CloudProviderId:            nextgenCluster.CloudProviderId,
			RegionId:                   nextgenCluster.RegionId,
			RootPassword:               nextgenCluster.RootPassword,
			Version:                    nextgenCluster.Version,
			MinRcu:                     nextgenCluster.MinRcu,
			MaxRcu:                     nextgenCluster.MaxRcu,
			ServicePlan:                nextgenCluster.ServicePlan,
			SourceClusterId:            cluster.SourceClusterId,
			DualLayerDataEncryption:    nextgenCluster.DualLayerDataEncryption,
			RuVersion:                  cluster.RuVersion,
			ReservedCapacityTemplateId: nextgenCluster.GetReservedCapacityTemplateId(),
		},
	}
	if cluster.ResourcePoolID != 0 {
		createReq.ResourcePoolId = &cluster.ResourcePoolID
	}

	return createReq, nil
}

func cvtNextGenCluster(base *entity.Cluster, nextgenCluster *globalngv1.Cluster) *entity.Cluster {
	if nextgenCluster == nil {
		return base
	}

	base.GlobalCluster = nextgenCluster
	if updateTime := nextgenCluster.GetUpdateTime(); updateTime != nil {
		base.UpdatedAt = updateTime.AsTime()
	}

	if nextgenCluster.RegionDisplayName == "" {
		nextgenCluster.RegionDisplayName = base.Region
	}

	return base
}