package cluster

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

	"github.com/samber/lo"
	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/adapter/billing"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	clusterrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/cluster"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/region"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/resourcepoolpolicy"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/rcuquota"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	"github.com/tidbcloud/tidb-management-service/utilities/randstr"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	minClusterPasswordLength = 8
	maxClusterPasswordLength = 64
	defaultRootPasswdLen     = 16 // default length of root password
	defaultTidbGroupName     = "DefaultGroup"
)

func (s *serviceImpl) Create(
	ctx context.Context,
	req *globalngv1.CreateClusterRequest,
) (*entity.Cluster, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "Domain.CreateCluster")
	defer span.End()

	// Validate request
	if req == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "request cannot be nil")
	}
	if err := validateResourcePoolSelectionRequest(req); err != nil {
		return nil, err
	}

	if err := s.validateCluster(ctx, req.GetCluster()); err != nil {
		return nil, err
	}

	cloudRegion, err := s.regionRepo.SelectRegion(ctx, region.Filter{
		CloudProvider: req.GetCluster().CloudProviderId,
		RegionId:      req.GetCluster().RegionId,
		ServicePlan:   req.GetCluster().ServicePlan,
		TenantId:      req.GetCluster().OrganizationId,
	})
	if err != nil {
		return nil, bizerr.WrapError(err, "select region failed")
	}
	if cloudRegion == nil {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "region not found")
	}

	// Check whether the organization is allowed to create a nextgen cluster.
	if err := s.checkTenantBillingPlan(ctx,
		req.GetCluster().OrganizationId,
		cloudRegion.CloudProviderId,
		cloudRegion.Id); err != nil {
		return nil, err
	}

	// Check idempotency before preparing the resource pool. This avoids repeating
	// side effects on retries, especially implicitly resuming a paused pool or
	// reprocessing encryption, while the transaction below still handles races.
	if !req.GetValidateOnly() {
		existingCluster, checkErr := s.clusterRepo.Get(ctx, req.Cluster.ClusterId, false)
		if checkErr == nil {
			if existingCluster == nil {
				return nil, bizerr.New(bizerr.GeneralInternalErrorCode, "cluster exists but could not be loaded")
			}
			tilog.Warn(ctx, "cluster already exists, skipping creation", zap.Uint64("clusterId", req.Cluster.ClusterId))
			return existingCluster, nil
		}
		if checkErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
			return nil, bizerr.WrapError(checkErr, "check cluster existence failed")
		}
	}

	var providerIdentityId uint64
	var dataplaneId string
	var resourcePoolDisplayName string
	var resourcePoolForCreate *entity.ResourcePool
	selectedResourcePool := req.ResourcePoolId != nil
	switch {
	case selectedResourcePool:
		resourcePool, err := s.loadSelectedResourcePool(ctx, req, false)
		if err != nil {
			return nil, err
		}
		if req.GetValidateOnly() {
			if err := s.validateSelectedResourcePoolInstanceLimit(ctx, resourcePool); err != nil {
				return nil, err
			}
		}
		resourcePoolForCreate = resourcePool
		if req.GetValidateOnly() {
			if err := s.validateSelectedResourcePoolVcpusLimit(ctx, req, cloudRegion, resourcePool); err != nil {
				return nil, err
			}
		}
		providerIdentityId = resourcePool.ProviderIdentityId
		dataplaneId = resourcePool.DataplaneId
		if err := setHighAvailabilityTypeFromResourcePool(req.GetCluster(), resourcePool); err != nil {
			return nil, err
		}
		req.GetCluster().SharedPoolId = strconv.FormatUint(resourcePool.GetId(), 10)
		req.GetCluster().ResourcePoolDisplayName = resourcePool.GetName()
		resourcePoolDisplayName = resourcePool.GetName()
	case req.GetValidateOnly() &&
		req.GetCluster().GetServicePlan() == globalcommonv1.ServicePlan_Essential_V2 &&
		req.GetCluster().SharedPoolId == "":
		resourcePool, err := s.pickResourcePoolForCreate(ctx, req.GetCluster())
		if err != nil {
			return nil, mapValidateOnlyEssentialV2PoolSelectionError(err)
		}
		if err := validateSharedResourcePoolForCreate(req, resourcePool); err != nil {
			return nil, err
		}
		if err := setHighAvailabilityTypeFromResourcePool(req.GetCluster(), resourcePool); err != nil {
			return nil, err
		}
		resourcePoolForCreate = resourcePool
		req.GetCluster().SharedPoolId = strconv.FormatUint(resourcePool.GetId(), 10)
		req.GetCluster().ResourcePoolDisplayName = resourcePool.GetName()
		resourcePoolDisplayName = resourcePool.GetName()
	case req.GetValidateOnly() && req.GetCluster().SharedPoolId == "":
		// Validation-only requests must not use GetOrCreateDefaultResourcePool,
		// because validation must not create or switch a default pool.
		resourcePool, err := s.resourcePoolSvc.FindDefaultResourcePool(ctx, entity.ResourcePool{
			ResourcePool: &globalngv1.ResourcePool{
				ProviderId:  req.GetCluster().GetCloudProviderId(),
				RegionId:    req.GetCluster().GetRegionId(),
				OwnerType:   globalngv1.ResourcePool_ORGANIZATION,
				OwnerId:     req.GetCluster().GetOrganizationId(),
				ServicePlan: req.GetCluster().GetServicePlan(),
			},
		})
		if err != nil {
			return nil, err
		}
		if resourcePool != nil {
			if err := validateSharedResourcePoolForCreate(req, resourcePool); err != nil {
				return nil, err
			}
			req.GetCluster().SharedPoolId = strconv.FormatUint(resourcePool.GetId(), 10)
			req.GetCluster().ResourcePoolDisplayName = resourcePool.GetName()
		}
	case !req.GetValidateOnly() && req.GetCluster().SharedPoolId == "":
		resourcePool, err := s.pickResourcePoolForCreate(ctx, req.GetCluster())
		if err != nil {
			return nil, err
		}
		if req.GetCluster().GetServicePlan() == globalcommonv1.ServicePlan_Essential_V2 {
			if err := validateSharedResourcePoolForCreate(req, resourcePool); err != nil {
				return nil, err
			}
			if err := setHighAvailabilityTypeFromResourcePool(req.GetCluster(), resourcePool); err != nil {
				return nil, err
			}
		}
		resourcePoolForCreate = resourcePool
		providerIdentityId = resourcePool.ProviderIdentityId
		dataplaneId = resourcePool.DataplaneId

		req.GetCluster().SharedPoolId = fmt.Sprintf("%d", resourcePool.GetId())
		req.GetCluster().ResourcePoolDisplayName = resourcePool.GetName()
		resourcePoolDisplayName = resourcePool.GetName()
	case !req.GetValidateOnly() && req.GetCluster().SharedPoolId != "":
		sharedPoolID, parseErr := strconv.ParseUint(req.GetCluster().SharedPoolId, 10, 64)
		if parseErr != nil || sharedPoolID == 0 {
			return nil, bizerr.New(bizerr.ErrCodeParams, "invalid shared_pool_id")
		}
		resourcePool, poolErr := s.resourcePoolRepo.Load(ctx, sharedPoolID, false)
		if poolErr != nil {
			return nil, poolErr
		}
		if resourcePool == nil {
			return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "resource pool not found")
		}
		if err := validateSharedResourcePoolForCreate(req, resourcePool); err != nil {
			return nil, err
		}
		if req.GetCluster().GetServicePlan() == globalcommonv1.ServicePlan_Essential_V2 {
			if err := setHighAvailabilityTypeFromResourcePool(req.GetCluster(), resourcePool); err != nil {
				return nil, err
			}
		}
		resourcePoolForCreate = resourcePool
		providerIdentityId = resourcePool.ProviderIdentityId
		dataplaneId = resourcePool.DataplaneId
		resourcePoolDisplayName = resourcePool.GetName()
		req.Cluster.ResourcePoolDisplayName = resourcePoolDisplayName
	}

	if err := s.validatePassword(req.Cluster.RootPassword); err != nil {
		return nil, bizerr.WrapError(err, "validate password failed")
	}

	var fromCluster *entity.Cluster
	if req.Cluster.GetSourceClusterId() != uint64(0) {
		fromCluster, err = s.clusterRepo.Get(ctx, req.Cluster.GetSourceClusterId(), true)
		if err != nil {
			return nil, bizerr.WrapError(err, "get from cluster failed")
		}
		if fromCluster == nil {
			return nil, bizerr.New(bizerr.GeneralInternalErrorCode, "from cluster not found")
		}
	}

	var dualLayerEncryptionKeyId uint64
	if req.GetValidateOnly() {
		if err := s.validateDualLayerEncryptionKeyForValidateOnly(ctx, req); err != nil {
			return nil, err
		}
	} else {
		dualLayerEncryptionKeyId, err = s.handleDualLayerEncryptionKey(ctx, req, providerIdentityId, dataplaneId, fromCluster)
		if err != nil {
			return nil, err
		}
	}

	cluster := &entity.Cluster{
		Cluster:         req.Cluster,
		RegionId:        cloudRegion.RegionID,
		CloudProviderId: cloudRegion.ProviderID,
		EncryptionKeyId: dualLayerEncryptionKeyId,
		ProjectId:       uint64(req.Cluster.GetProjectId()),
	}

	// set dataplane id
	cluster.DataplaneId = dataplaneId
	setDefaultHighAvailabilityType(cluster)

	// Save root password.
	var rootPassword string
	if cluster.RootPassword == "" {
		// generate a temp random password for root
		rootPassword = randstr.GenWithCharset(defaultRootPasswdLen, randstr.AlphanumericCharset)
	} else {
		// set root password on request, use it directly
		rootPassword = cluster.RootPassword
		// Root password not returned to the user.
		req.Cluster.RootPassword = ""
		req.Cluster.HasRootPassword = true
	}

	encryptedRootPassword, err := s.encryptor.EncryptBase64(ctx, rootPassword)
	if err != nil {
		return nil, bizerr.WrapError(err, "encrypt root password failed")
	}

	// set encrypted cloud_admin password
	switch {
	case fromCluster != nil:
		if fromCluster.EncryptedCloudAdminPassword == "" {
			return nil, bizerr.New(bizerr.GeneralInternalErrorCode, "from cluster has no encrypted cloud admin password")
		}
		cluster.EncryptedCloudAdminPassword = fromCluster.EncryptedCloudAdminPassword
	default:
		cloudAdminPassword := randstr.GenWithCharset(defaultRootPasswdLen, randstr.AlphanumericCharset)
		cluster.EncryptedCloudAdminPassword, err = s.encryptor.EncryptBase64(ctx, cloudAdminPassword)
		if err != nil {
			return nil, bizerr.WrapError(err, "failed to encrypt cloud admin password")
		}
	}

	// set encrypted platform_admin password
	switch {
	case req.Cluster.GetEncryptedPlatformAdminPassword() != "":
		cluster.EncryptedPlatformAdminPassword = req.Cluster.GetEncryptedPlatformAdminPassword()
	case fromCluster != nil:
		if fromCluster.EncryptedPlatformAdminPassword == "" {
			return nil, bizerr.New(bizerr.GeneralInternalErrorCode, "from cluster has no encrypted platform admin password")
		}
		cluster.EncryptedPlatformAdminPassword = fromCluster.EncryptedPlatformAdminPassword
	default:
		platformAdminPassword := randstr.GenWithCharset(defaultRootPasswdLen, randstr.AlphanumericCharset)
		cluster.EncryptedPlatformAdminPassword, err = s.encryptor.EncryptBase64(ctx, platformAdminPassword)
		if err != nil {
			return nil, bizerr.WrapError(err, "failed to encrypt platform admin password")
		}
	}

	// set encrypted maintenance_admin password
	switch {
	case req.Cluster.GetEncryptedMaintenanceAdminPassword() != "":
		cluster.EncryptedMaintenanceAdminPassword = req.Cluster.GetEncryptedMaintenanceAdminPassword()
	case fromCluster != nil:
		if fromCluster.EncryptedMaintenanceAdminPassword == "" {
			return nil, bizerr.New(bizerr.GeneralInternalErrorCode, "from cluster has no encrypted maintenance admin password")
		}
		cluster.EncryptedMaintenanceAdminPassword = fromCluster.EncryptedMaintenanceAdminPassword
	default:
		maintenanceAdminPassword := randstr.GenWithCharset(defaultRootPasswdLen, randstr.AlphanumericCharset)
		cluster.EncryptedMaintenanceAdminPassword, err = s.encryptor.EncryptBase64(ctx, maintenanceAdminPassword)
		if err != nil {
			return nil, bizerr.WrapError(err, "failed to encrypt maintenance admin password")
		}
	}

	if !req.GetValidateOnly() {
		if err := s.ensureResourcePoolReadyForCreate(ctx, resourcePoolForCreate); err != nil {
			return nil, err
		}
		// Resume a paused pool before querying runtime capacity.
		if resourcePoolForCreate != nil {
			if err := s.validateSelectedResourcePoolVcpusLimit(ctx, req, cloudRegion, resourcePoolForCreate); err != nil {
				return nil, err
			}
		}
	}

	// set cluster to create state
	cluster.State = globalngv1.Cluster_CREATING
	cluster.HealthState = globalngv1.Cluster_UNHEALTHY

	if req.ValidateOnly {
		// If validate only, we don't need to save the cluster to the database.
		return cluster, nil
	}

	// TODO: dns id is tidbgroup required, not cluster required
	// cluster.ClusterIdentifierHash = genClusterDnsId()

	var defaultTidbGroupId uint64
	nerr := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		var lockedResourcePool *entity.ResourcePool
		if resourcePoolForCreate != nil {
			resourcePool, lockErr := s.resourcePoolRepo.LoadForUpdate(ctx, resourcePoolForCreate.GetId(), false)
			if lockErr != nil {
				return lockErr
			}
			lockedResourcePool = resourcePool
		}

		// first check if the cluster already exists
		_, err := s.clusterRepo.Get(ctx, req.Cluster.ClusterId, false)
		if err == nil {
			tilog.Warn(ctx, "cluster already exists, skipping creation", zap.Uint64("clusterId", req.Cluster.ClusterId))
			// cluster already exists, return
			return nil
		} else if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
			return bizerr.WrapError(err, "check cluster existence failed")
		}
		switch {
		case selectedResourcePool:
			if validateErr := validateSelectedResourcePool(req, lockedResourcePool); validateErr != nil {
				return validateErr
			}
			if readyErr := validateResourcePoolReadyForCreate(lockedResourcePool); readyErr != nil {
				return readyErr
			}
			if limitErr := s.validateSelectedResourcePoolInstanceLimit(ctx, lockedResourcePool); limitErr != nil {
				return limitErr
			}
			if haErr := setHighAvailabilityTypeFromResourcePool(cluster.Cluster, lockedResourcePool); haErr != nil {
				return haErr
			}
			cluster.SharedPoolId = strconv.FormatUint(lockedResourcePool.GetId(), 10)
			cluster.DataplaneId = lockedResourcePool.GetDataplaneId()
			cluster.ResourcePoolDisplayName = lockedResourcePool.GetName()
			resourcePoolDisplayName = lockedResourcePool.GetName()
		case lockedResourcePool == nil:
			return bizerr.New(bizerr.ErrCodeParams, "resource pool is required for cluster creation")
		default:
			if associationErr := validateSharedResourcePoolForCreate(req, lockedResourcePool); associationErr != nil {
				return associationErr
			}
			if cluster.GetServicePlan() == globalcommonv1.ServicePlan_Essential_V2 {
				if haErr := setHighAvailabilityTypeFromResourcePool(cluster.Cluster, lockedResourcePool); haErr != nil {
					return haErr
				}
			}
			if validateErr := validateResourcePoolReadyForCreate(lockedResourcePool); validateErr != nil {
				return validateErr
			}
			if limitErr := s.validateSelectedResourcePoolInstanceLimit(ctx, lockedResourcePool); limitErr != nil {
				return limitErr
			}
		}

		if err := s.clusterRepo.Create(ctx, cluster); err != nil {
			return bizerr.WrapError(err, "create cluster failed")
		}

		if err := s.clusterRepo.SetEncryptedRootPassword(ctx, cluster.ClusterId, encryptedRootPassword); err != nil {
			return bizerr.WrapError(err, "set encrypted root password failed")
		}

		// A restore target must wait for Native BR to import the source encryption
		// metadata. Normal create-time encryption is ready once cluster creation
		// completes. The M2 path (SwitchClusterEncryptionKey) creates its own
		// binding with state=Modifying and later finalizes to Enabled.
		if dualLayerEncryptionKeyId != 0 {
			bindingState := entity.ClusterEncryptionBindingStateEnabled
			bindingMessage := "encryption configured at cluster creation"
			if cluster.GetSourceClusterId() != 0 {
				bindingState = entity.ClusterEncryptionBindingStatePendingRestore
				bindingMessage = "waiting for native BR restore to import encryption metadata"
			}
			if err := s.bindingRepo.Upsert(ctx, &entity.ClusterEncryptionBinding{
				ClusterId:       cluster.ClusterId,
				TenantId:        cluster.GetOrganizationId(),
				EncryptionKeyId: dualLayerEncryptionKeyId,
				State:           bindingState,
				Message:         bindingMessage,
			}); err != nil {
				return bizerr.WrapError(err, "create encryption binding failed")
			}
		}

		defaultTidbGroupId, err = s.createDefaultTidbGroup(ctx, cluster)
		if err != nil {
			return bizerr.WrapError(err, "create default tidb group failed")
		}

		events, err := cluster.CreateEvent(ctx)
		if err != nil {
			return bizerr.WrapError(err, "create cluster event failed")
		}
		if err := s.eventCommitter.BatchWrite(ctx, clusterevent.TopicClusterEvent, events); err != nil {
			return bizerr.WrapError(err, "write cluster event failed")
		}

		// metadata
		metadataJson, merr := metadata.MarshalRequestMetaFromCtx(ctx)
		if merr != nil {
			tilog.Warn(ctx, "failed to marshal request metadata", zap.Error(merr))
		}

		workflowErr := s.workflow.RunDagWithBizId(
			ctx,
			common.CreateCluster,
			strconv.FormatUint(cluster.OrganizationId, 10),
			"0", // nextgen cluster don't have project id
			strconv.FormatUint(cluster.ClusterId, 10),
			map[string]string{
				workflow.KeyTraceId:          workflow.GetTraceId(ctx),
				workflow.KeyRequestMeta:      string(metadataJson),
				common.KeyClusterId:          strconv.FormatUint(req.Cluster.ClusterId, 10),
				common.KeyClusterDisplayName: req.GetCluster().GetDisplayName(),
				common.KeyTargetStatus:       globalngv1.Cluster_AVAILABLE.String(),
				common.KeyTidbGroupId:        strconv.FormatUint(defaultTidbGroupId, 10),
				common.KeyTidbGroupIds:       strconv.FormatUint(defaultTidbGroupId, 10),
			},
		)
		if workflowErr != nil {
			return bizerr.WrapErrorWithCode(workflowErr, bizerr.ErrCodeUnknown, "create async workflow failed")
		}

		privateLinkServiceID := s.idGeneratorAdapter.MustNew()
		// create private link service
		if err := s.privatelinkRepo.CreatePrivateLinkService(ctx, &entity.PrivateLinkService{
			PrivateLinkServiceID: privateLinkServiceID,
			PrivateLinkService: &globalngv1.PrivateLinkService{
				ClusterId:         req.Cluster.ClusterId,
				TidbGroupId:       defaultTidbGroupId,
				ServicePort:       4000,
				RegionId:          cloudRegion.Id,
				RegionDisplayName: cloudRegion.DisplayName,
				CloudProviderId:   cloudRegion.CloudProviderId,
			},
		}); err != nil {
			return bizerr.WrapError(err, "create private link service failed")
		}
		workflowErr = s.workflow.RunDagWithBizId(
			ctx,
			common.CreatePrivateLinkService,
			strconv.FormatUint(cluster.OrganizationId, 10),
			"0", // nextgen cluster don't have project id
			strconv.FormatUint(cluster.ClusterId, 10),
			map[string]string{
				workflow.KeyTraceId:             workflow.GetTraceId(ctx),
				common.KeyClusterId:             strconv.FormatUint(req.Cluster.ClusterId, 10),
				common.KeyTidbGroupId:           strconv.FormatUint(defaultTidbGroupId, 10),
				common.KeyTidbGroupIds:          strconv.FormatUint(defaultTidbGroupId, 10),
				common.KeyPrivateLinkServiceIds: strconv.FormatUint(privateLinkServiceID, 10),
				common.KeyPrivateLbEnable:       "true",
			},
		)
		if workflowErr != nil {
			return bizerr.WrapErrorWithCode(workflowErr, bizerr.ErrCodeUnknown, "create async workflow failed")
		}

		return nil
	})
	if nerr != nil {
		return nil, bizerr.WrapError(nerr, "create cluster failed")
	}

	newCluster, err := s.clusterRepo.Get(ctx, req.Cluster.ClusterId, false)
	if err != nil {
		return nil, bizerr.WrapError(err, "get cluster after create failed")
	}
	if newCluster == nil {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster not found after creation")
	}
	newCluster.ResourcePoolDisplayName = resourcePoolDisplayName

	return newCluster, nil
}

func validateResourcePoolSelectionRequest(req *globalngv1.CreateClusterRequest) bizerr.BizError {
	if req.ResourcePoolId == nil {
		if req.GetCluster().GetServicePlan() == globalcommonv1.ServicePlan_BYOC {
			return bizerr.New(bizerr.ErrCodeParams, "resource_pool_id is required for BYOC")
		}
		return nil
	}
	if req.GetCluster().GetServicePlan() != globalcommonv1.ServicePlan_BYOC {
		return bizerr.New(bizerr.ErrCodeParams, "resource_pool_id is only supported for BYOC")
	}
	if req.GetResourcePoolId() == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "resource_pool_id must be greater than zero")
	}
	return nil
}

func (s *serviceImpl) loadSelectedResourcePool(
	ctx context.Context,
	req *globalngv1.CreateClusterRequest,
	forUpdate bool,
) (*entity.ResourcePool, bizerr.BizError) {
	var (
		resourcePool *entity.ResourcePool
		err          bizerr.BizError
	)
	if forUpdate {
		resourcePool, err = s.resourcePoolRepo.LoadForUpdate(ctx, req.GetResourcePoolId(), false)
	} else {
		resourcePool, err = s.resourcePoolRepo.Load(ctx, req.GetResourcePoolId(), false)
	}
	if err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "resource pool not found")
		}
		return nil, err
	}
	if err := validateSelectedResourcePool(req, resourcePool); err != nil {
		return nil, err
	}
	return resourcePool, nil
}

func validateSelectedResourcePool(
	req *globalngv1.CreateClusterRequest,
	resourcePool *entity.ResourcePool,
) bizerr.BizError {
	cluster := req.GetCluster()
	if resourcePool == nil ||
		resourcePool.GetOwnerType() != globalngv1.ResourcePool_ORGANIZATION ||
		resourcePool.GetOwnerId() != cluster.GetOrganizationId() ||
		resourcePool.GetServicePlan() != globalcommonv1.ServicePlan_BYOC {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "resource pool not found")
	}
	if resourcePool.GetProviderId() != cluster.GetCloudProviderId() ||
		resourcePool.GetRegionId() != cluster.GetRegionId() {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool provider and region must match the cluster")
	}
	return validateResourcePoolLifecycleForCreate(resourcePool)
}

func validateSharedResourcePoolForCreate(
	req *globalngv1.CreateClusterRequest,
	resourcePool *entity.ResourcePool,
) bizerr.BizError {
	cluster := req.GetCluster()
	if resourcePool == nil {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "resource pool not found")
	}
	if resourcePool.GetServicePlan() != cluster.GetServicePlan() {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "resource pool not found")
	}
	if resourcePool.GetProviderId() != cluster.GetCloudProviderId() ||
		resourcePool.GetRegionId() != cluster.GetRegionId() {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool provider and region must match the cluster")
	}
	if cluster.GetServicePlan() == globalcommonv1.ServicePlan_Essential_V2 {
		requestedHighAvailabilityType := cluster.GetHighAvailabilityType()
		if requestedHighAvailabilityType != globalngv1.Cluster_HIGH_AVAILABILITY_TYPE_UNSPECIFIED {
			poolHighAvailabilityType, err := clusterHighAvailabilityTypeFromResourcePool(resourcePool)
			if err != nil {
				return err
			}
			if requestedHighAvailabilityType != poolHighAvailabilityType {
				return bizerr.New(bizerr.ErrCodeParams, "resource pool high availability type must match the cluster")
			}
		}
	}

	switch resourcePool.GetOwnerType() {
	case globalngv1.ResourcePool_SHARED:
		if resourcePool.GetOwnerId() != 0 {
			return bizerr.New(bizerr.ErrCodeRecordNotFound, "resource pool not found")
		}
	case globalngv1.ResourcePool_ORGANIZATION:
		if resourcePool.GetOwnerId() != cluster.GetOrganizationId() {
			return bizerr.New(bizerr.ErrCodeRecordNotFound, "resource pool not found")
		}
	default:
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "resource pool not found")
	}

	return validateResourcePoolLifecycleForCreate(resourcePool)
}

func validateResourcePoolLifecycleForCreate(resourcePool *entity.ResourcePool) bizerr.BizError {
	if resourcePool == nil {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool is required for cluster creation")
	}
	if resourcePool.GetLifecycleState() == globalngv1.ResourcePool_FAILED {
		return bizerr.New(bizerr.ErrCodeGeneralBusiness, "resource pool is not available for TiDB creation")
	}

	switch resourcepoolpolicy.Decide(
		resourcepoolpolicy.OperationCreateInstance,
		resourcePool.GetLifecycleState(),
	) {
	case resourcepoolpolicy.DecisionAllowed, resourcepoolpolicy.DecisionTriggerOrReuseResume:
		return nil
	default:
		return bizerr.New(bizerr.ErrCodeGeneralBusiness, "resource pool is not available for TiDB creation")
	}
}

func validateResourcePoolReadyForCreate(resourcePool *entity.ResourcePool) bizerr.BizError {
	if resourcePool == nil {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool is required for cluster creation")
	}
	if resourcePool.GetLifecycleState() == globalngv1.ResourcePool_FAILED {
		return bizerr.New(bizerr.ErrCodeGeneralBusiness, "resource pool is not available for TiDB creation")
	}

	decision := resourcepoolpolicy.Decide(
		resourcepoolpolicy.OperationCreateInstance,
		resourcePool.GetLifecycleState(),
	)
	if decision == resourcepoolpolicy.DecisionAllowed ||
		(decision == resourcepoolpolicy.DecisionTriggerOrReuseResume &&
			resourcePool.GetLifecycleState() == globalngv1.ResourcePool_RESUMING) {
		return nil
	}
	return bizerr.New(bizerr.ErrCodeGeneralBusiness, "resource pool is not available for TiDB creation")
}

func (s *serviceImpl) ensureResourcePoolReadyForCreate(
	ctx context.Context,
	resourcePool *entity.ResourcePool,
) bizerr.BizError {
	if resourcePool == nil {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool is required for cluster creation")
	}
	if resourcePool.GetLifecycleState() == globalngv1.ResourcePool_FAILED {
		return bizerr.New(bizerr.ErrCodeGeneralBusiness, "resource pool is not available for TiDB creation")
	}

	decision := resourcepoolpolicy.Decide(
		resourcepoolpolicy.OperationCreateInstance,
		resourcePool.GetLifecycleState(),
	)
	switch decision {
	case resourcepoolpolicy.DecisionAllowed:
		return nil
	case resourcepoolpolicy.DecisionTriggerOrReuseResume:
		if resourcePool.GetLifecycleState() == globalngv1.ResourcePool_RESUMING {
			return nil
		}
		if s.resourcePoolSvc == nil || s.resourcePoolRepo == nil || s.idGeneratorAdapter == nil {
			return bizerr.New(bizerr.GeneralInternalErrorCode, "resource pool resume dependencies are unavailable")
		}

		operationID := strconv.FormatUint(s.idGeneratorAdapter.MustNew(), 10)
		_, berr := s.resourcePoolSvc.ResumeResourcePool(ctx, &globalngv1.ResumeResourcePoolRequest{
			ResourcePoolId: resourcePool.GetId(),
			OperationId:    operationID,
		})
		if berr == nil {
			return nil
		}
		if berr.ErrCode() != bizerr.ErrCode(codes.FailedPrecondition) {
			return berr
		}

		// Another request may have won the PAUSED -> RESUMING transition. The
		// row-lock in ResumeResourcePool makes RESUMING a safe admission state.
		latest, loadErr := s.resourcePoolRepo.Load(ctx, resourcePool.GetId(), false)
		if loadErr == nil && (latest.GetLifecycleState() == globalngv1.ResourcePool_RESUMING ||
			latest.GetLifecycleState() == globalngv1.ResourcePool_ACTIVE) {
			return nil
		}
		if loadErr != nil {
			return loadErr
		}
		return berr
	default:
		return bizerr.New(
			bizerr.ErrCodeGeneralBusiness,
			"resource pool is %s and not available for TiDB creation",
			strings.ToLower(resourcePool.GetLifecycleState().String()),
		)
	}
}

func (s *serviceImpl) validateSelectedResourcePoolInstanceLimit(
	ctx context.Context,
	resourcePool *entity.ResourcePool,
) bizerr.BizError {
	counts, err := s.clusterRepo.CountLiveBySharedPoolIDs(ctx, []uint64{resourcePool.GetId()})
	if err != nil {
		return err
	}
	if counts[resourcePool.GetId()] >= int64(resourcePool.GetInstanceLimit()) {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"You have reached a service limit and no more instances can be created now. Please contact support for more information.\n",
		)
	}
	return nil
}

func (s *serviceImpl) shouldValidateSelectedResourcePoolVcpusLimit(
	ctx context.Context,
	req *globalngv1.CreateClusterRequest,
	resourcePool *entity.ResourcePool,
) (bool, bizerr.BizError) {
	isNonByoc := resourcePool.GetServicePlan() != globalcommonv1.ServicePlan_BYOC
	isUnlimited := resourcePool.GetVcpusLimit() == 0
	isCreatingPool := resourcePool.GetLifecycleState() == globalngv1.ResourcePool_CREATING
	if isNonByoc || isUnlimited || isCreatingPool {
		return false, nil
	}
	if req.GetValidateOnly() {
		return true, nil
	}

	_, err := s.clusterRepo.Get(ctx, req.GetCluster().GetClusterId(), false)
	if err == nil {
		return false, nil
	}
	if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
		return true, nil
	}
	return false, bizerr.WrapError(err, "check cluster existence failed")
}

func (s *serviceImpl) validateSelectedResourcePoolVcpusLimit(
	ctx context.Context,
	req *globalngv1.CreateClusterRequest,
	cloudRegion *entity.Region,
	resourcePool *entity.ResourcePool,
) bizerr.BizError {
	shouldValidate, err := s.shouldValidateSelectedResourcePoolVcpusLimit(ctx, req, resourcePool)
	if err != nil {
		return err
	}
	if !shouldValidate {
		return nil
	}

	// This guard rejects a pool that has no remaining vCPU capacity for a new
	// TiDB. It intentionally does not reserve vCPUs or estimate the capacity
	// required by the new TiDB; enforcing future or concurrent capacity requires
	// an atomic regional admission or reservation mechanism.
	// A CREATING pool may not have a regional runtime yet, so its missing usage
	// is expected and the check is deferred until the pool becomes available.
	usage, err := s.regionalCli.GetSharedPoolVcpusUsage(ctx, cloudRegion, &clusterngv1beta.GetSharedPoolVcpusUsageRequest{
		SharedPoolId: strconv.FormatUint(resourcePool.GetId(), 10),
	})
	if err != nil {
		return bizerr.WrapError(err, "get shared pool vCPU usage failed")
	}
	if usage == nil {
		return bizerr.New(
			bizerr.ErrCodeUnknown,
			"received nil response for shared pool vCPU usage query",
		)
	}
	if usage.GetVcpusUsed() < resourcePool.GetVcpusLimit() {
		return nil
	}

	return bizerr.New(
		bizerr.ErrCode(codes.ResourceExhausted),
		"the resource pool vCPU usage (%d) reaches or exceeds its limit (%d); increase the limit before creating a TiDB instance",
		usage.GetVcpusUsed(),
		resourcePool.GetVcpusLimit(),
	)
}

func (s *serviceImpl) pickResourcePoolForCreate(
	ctx context.Context,
	cluster *globalngv1.Cluster,
) (*entity.ResourcePool, bizerr.BizError) {
	if cluster.GetServicePlan() == globalcommonv1.ServicePlan_BYOC {
		return nil, bizerr.New(bizerr.ErrCodeParams, "resource_pool_id is required for BYOC")
	}
	if cluster.GetServicePlan() == globalcommonv1.ServicePlan_Essential_V2 {
		highAvailabilityType, err := resourcePoolHighAvailabilityTypeFromCluster(cluster.GetHighAvailabilityType())
		if err != nil {
			return nil, err
		}
		return s.resourcePoolSvc.PickSharedResourcePool(ctx, entity.ResourcePool{
			ResourcePool: &globalngv1.ResourcePool{
				ProviderId:           cluster.GetCloudProviderId(),
				RegionId:             cluster.GetRegionId(),
				ServicePlan:          cluster.GetServicePlan(),
				HighAvailabilityType: highAvailabilityType,
			},
		}, cluster.GetOrganizationId())
	}
	if cluster.GetServicePlan() == globalcommonv1.ServicePlan_Premium_Reserved {
		return s.resourcePoolSvc.CreateTopologyDrivenResourcePool(ctx, entity.ResourcePool{
			ResourcePool: &globalngv1.ResourcePool{
				ProviderId:    cluster.GetCloudProviderId(),
				RegionId:      cluster.GetRegionId(),
				OwnerType:     globalngv1.ResourcePool_ORGANIZATION,
				OwnerId:       cluster.GetOrganizationId(),
				ServicePlan:   cluster.GetServicePlan(),
				InstanceLimit: 1,
			},
		}, fmt.Sprintf("reserved-for-%d", cluster.GetClusterId()), cluster.GetReservedCapacityTemplate().GetFixedTopology())
	}

	return s.resourcePoolSvc.GetOrCreateDefaultResourcePool(ctx, entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			ProviderId:  cluster.GetCloudProviderId(),
			RegionId:    cluster.GetRegionId(),
			OwnerType:   globalngv1.ResourcePool_ORGANIZATION,
			OwnerId:     cluster.GetOrganizationId(),
			ServicePlan: cluster.GetServicePlan(),
		},
	}, uint32(cluster.GetMaxRcu()))
}

func mapValidateOnlyEssentialV2PoolSelectionError(err bizerr.BizError) bizerr.BizError {
	if err == nil {
		return nil
	}
	if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
		return bizerr.New(bizerr.ErrCode(codes.ResourceExhausted), "server resources are insufficient")
	}
	return err
}

func (s *serviceImpl) createDefaultTidbGroup(ctx context.Context, cluster *entity.Cluster) (uint64, bizerr.BizError) {
	if cluster == nil {
		return 0, bizerr.New(bizerr.ErrCodeParams, "cluster cannot be nil")
	}

	id := s.idGeneratorAdapter.MustNew()
	group := &entity.TidbGroup{
		Id:          id,
		DisplayName: defaultTidbGroupName,
		Default:     true,
		ClusterId:   cluster.ClusterId,
	}
	return id, s.tidbgroupRepo.Create(ctx, group)
}

func setDefaultHighAvailabilityType(cluster *entity.Cluster) {
	if cluster == nil {
		return
	}
	if cluster.GetHighAvailabilityType() != globalngv1.Cluster_HIGH_AVAILABILITY_TYPE_UNSPECIFIED {
		return
	}

	// TODO: get this from region capabilities.
	switch cluster.GetServicePlan() {
	case globalcommonv1.ServicePlan_Premium,
		globalcommonv1.ServicePlan_Premium_Reserved,
		globalcommonv1.ServicePlan_Essential_V2:
		cluster.HighAvailabilityType = globalngv1.Cluster_REGIONAL
	}
}

func setHighAvailabilityTypeFromResourcePool(
	cluster *globalngv1.Cluster,
	resourcePool *entity.ResourcePool,
) bizerr.BizError {
	if cluster == nil || resourcePool == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "cluster or resource pool is missing")
	}
	highAvailabilityType, err := clusterHighAvailabilityTypeFromResourcePool(resourcePool)
	if err != nil {
		return err
	}
	cluster.HighAvailabilityType = highAvailabilityType
	return nil
}

func clusterHighAvailabilityTypeFromResourcePool(
	resourcePool *entity.ResourcePool,
) (globalngv1.Cluster_HighAvailabilityType, bizerr.BizError) {
	if resourcePool == nil {
		return globalngv1.Cluster_HIGH_AVAILABILITY_TYPE_UNSPECIFIED,
			bizerr.New(bizerr.ErrCodeUnknown, "resource pool is missing")
	}
	switch resourcePool.GetHighAvailabilityType() {
	case globalngv1.ResourcePool_HIGH_AVAILABILITY_TYPE_UNSPECIFIED,
		globalngv1.ResourcePool_REGIONAL:
		return globalngv1.Cluster_REGIONAL, nil
	case globalngv1.ResourcePool_ZONAL:
		return globalngv1.Cluster_ZONAL, nil
	default:
		return globalngv1.Cluster_HIGH_AVAILABILITY_TYPE_UNSPECIFIED,
			bizerr.New(bizerr.ErrCodeUnknown, "resource pool high availability type is invalid")
	}
}

func resourcePoolHighAvailabilityTypeFromCluster(
	highAvailabilityType globalngv1.Cluster_HighAvailabilityType,
) (globalngv1.ResourcePool_HighAvailabilityType, bizerr.BizError) {
	switch highAvailabilityType {
	case globalngv1.Cluster_HIGH_AVAILABILITY_TYPE_UNSPECIFIED:
		return globalngv1.ResourcePool_HIGH_AVAILABILITY_TYPE_UNSPECIFIED, nil
	case globalngv1.Cluster_REGIONAL:
		return globalngv1.ResourcePool_REGIONAL, nil
	case globalngv1.Cluster_ZONAL:
		return globalngv1.ResourcePool_ZONAL, nil
	default:
		return globalngv1.ResourcePool_HIGH_AVAILABILITY_TYPE_UNSPECIFIED,
			bizerr.New(bizerr.ErrCodeParams, "high availability type is invalid")
	}
}

func (s *serviceImpl) validatePassword(password string) bizerr.BizError {
	// means use auto-generated password
	if password == "" {
		return nil
	}

	passwordLen := len(password)
	if passwordLen < minClusterPasswordLength || passwordLen > maxClusterPasswordLength {
		return bizerr.New(bizerr.ErrCodeParams, "password must be 8-64 characters")
	}
	return nil
}

func (s *serviceImpl) handleDualLayerEncryptionKey(ctx context.Context,
	req *globalngv1.CreateClusterRequest,
	providerIdentityId uint64,
	dataplaneId string,
	sourceCluster *entity.Cluster,
) (uint64, bizerr.BizError) {
	if sourceCluster != nil {
		return s.handleDualLayerEncryptionKeyFromSourceCluster(ctx, req, providerIdentityId, dataplaneId, sourceCluster)
	}
	return s.handleDualLayerEncryptionKeyFromRequest(ctx, req, providerIdentityId, dataplaneId)
}

func (s *serviceImpl) validateDualLayerEncryptionKeyForValidateOnly(
	ctx context.Context,
	req *globalngv1.CreateClusterRequest,
) bizerr.BizError {
	encryption := req.GetCluster().GetDualLayerDataEncryption()
	switch {
	case encryption == nil ||
		encryption.GetNoEncryption() != nil ||
		encryption.GetDefaultKey() != nil:
		return nil
	case encryption.GetCmek() != nil:
		// A validate-only request must not create a pool just to run the KMS check.
		if req.GetCluster().GetSharedPoolId() == "" {
			return bizerr.New(bizerr.ErrCodeParams, "resource pool is required for CMEK validation")
		}
		return s.encryptionKeySvc.ValidateCustomerManagedEncryptionKey(
			ctx,
			encryption.GetCmek(),
			req.GetCluster().GetSharedPoolId(),
		)
	default:
		return bizerr.New(bizerr.ErrCodeParams, "invalid dual layer data encryption")
	}
}

func (s *serviceImpl) handleDualLayerEncryptionKeyFromSourceCluster(ctx context.Context,
	req *globalngv1.CreateClusterRequest,
	providerIdentityId uint64,
	dataplaneId string,
	sourceCluster *entity.Cluster,
) (uint64, bizerr.BizError) {
	if sourceCluster == nil {
		return 0, bizerr.New(bizerr.ErrCodeParams, "source cluster required")
	}
	if sourceCluster.EncryptionKeyId == 0 {
		tilog.Debug(ctx, "encryption key disabled in source cluster")
		return 0, nil
	}

	key, err := s.encryptionKeyRepo.Load(ctx, sourceCluster.EncryptionKeyId, false)
	if err != nil {
		return 0, bizerr.WrapError(err, "load encryption key failed")
	}

	if key == nil {
		return 0, bizerr.New(bizerr.ErrCodeParams, "encryption key not found")
	}

	if key.CustomerManaged {
		tilog.Info(ctx, "reuse the cmek of source cluster")
		if err := s.validateEncryptionKeyAvailableForRestore(ctx, key, req.GetCluster().GetSharedPoolId()); err != nil {
			return 0, err
		}
		return key.Id, nil
	}

	tilog.Info(ctx, "found default key used by source cluster, reuse it")
	defaultKey, err := s.encryptionKeySvc.CreateDefaultKeyOrGrantPermission(ctx,
		req.GetCluster().OrganizationId,
		providerIdentityId,
		dataplaneId)
	if err != nil {
		return 0, err
	}
	return defaultKey.Id, nil
}

func (s *serviceImpl) handleDualLayerEncryptionKeyFromRequest(ctx context.Context,
	req *globalngv1.CreateClusterRequest,
	providerIdentityId uint64,
	dataplaneId string,
) (uint64, bizerr.BizError) {
	switch {
	case req.GetCluster().GetDualLayerDataEncryption() == nil ||
		req.GetCluster().GetDualLayerDataEncryption().GetNoEncryption() != nil:
		tilog.Debug(ctx, "encryption key disabled")
		return 0, nil
	case req.GetCluster().GetDualLayerDataEncryption().GetDefaultKey() != nil:
		tilog.Info(ctx, "use default key")
		defaultKey, err := s.encryptionKeySvc.CreateDefaultKeyOrGrantPermission(ctx,
			req.GetCluster().OrganizationId,
			providerIdentityId,
			dataplaneId)
		if err != nil {
			return 0, err
		}
		return defaultKey.Id, nil
	case req.GetCluster().GetDualLayerDataEncryption().GetCmek() != nil:
		tilog.Info(ctx, "use cmek")
		cmek, err := s.encryptionKeySvc.CreateCustomerManagedEncryptionKey(ctx,
			req.GetCluster().GetDualLayerDataEncryption().GetCmek(),
			providerIdentityId,
			req.GetCluster().GetSharedPoolId(),
			dataplaneId)
		if err != nil {
			return 0, err
		}
		return cmek.CustomerManagedEncryptionKeyId, nil
	default:
		return 0, bizerr.New(bizerr.ErrCodeParams, "invalid dual layer data encryption")
	}
}

func (s *serviceImpl) validateCluster(ctx context.Context, cluster *globalngv1.Cluster) bizerr.BizError {
	if cluster == nil {
		return bizerr.New(bizerr.ErrCodeParams, "cluster cannot be nil")
	}

	if cluster.GetOrganizationId() == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "organization id cannot be 0")
	}

	if cluster.GetServicePlan() == globalcommonv1.ServicePlan_SERVICE_PLAN_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "service plan cannot be unspecified")
	}
	if cluster.GetServicePlan() == globalcommonv1.ServicePlan_BYOC && cluster.GetMaxRcu() <= 0 {
		return bizerr.New(bizerr.ErrCodeParams, "max RCU must be greater than zero for BYOC")
	}

	if cluster.GetServicePlan() == globalcommonv1.ServicePlan_Premium_Reserved {
		return s.validatePremiumReservedCluster(ctx, cluster)
	}

	rcuQuota, err := s.rcuQuotaRepo.Load(ctx, cluster.OrganizationId, cluster.ServicePlan.String())
	if err != nil {
		return bizerr.WrapError(err, "load rcu quota failed")
	}

	if cluster.GetMaxRcu() > int64(rcuQuota.InstanceMaxRcu) {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"The RCU limit for this instance is %d.",
			rcuQuota.InstanceMaxRcu,
		)
	}

	cluster.MinRcu = rcuquota.CalculateMinRcuByQuota(cluster.GetMaxRcu(), rcuQuota)

	if cluster.GetMinRcu() > cluster.GetMaxRcu() {
		return bizerr.New(bizerr.ErrCodeParams, "min RCU must be less than or equal to max RCU")
	}

	// validate total rcu of org
	clusters, err := s.clusterRepo.Query(ctx, &clusterrepo.Filter{
		OrganizationIds: []uint64{cluster.GetOrganizationId()},
		ServicePlans:    []entity.ServicePlan{cluster.GetServicePlan()},
	})

	if err != nil {
		return bizerr.WrapError(err, "query clusters for rcu validation failed")
	}

	totalMaxRcu := lo.Reduce[*entity.Cluster, uint64](clusters, func(agg uint64, item *entity.Cluster, index int) uint64 {
		return agg + uint64(item.GetMaxRcu())
	}, 0)

	if uint64(cluster.GetMaxRcu())+totalMaxRcu > rcuQuota.TotalMaxRcu {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"The RCU limit for this instance is %d.",
			rcuQuota.TotalMaxRcu,
		)
	}

	return nil
}

func (s *serviceImpl) validatePremiumReservedCluster(ctx context.Context, cluster *globalngv1.Cluster) bizerr.BizError {
	if cluster.GetReservedCapacityTemplateId() == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "reserved capacity template id cannot be empty")
	}
	template, err := s.reservedCapacityTemplateRepo.Load(ctx, cluster.GetReservedCapacityTemplateId(), false)
	if err != nil {
		return bizerr.WrapError(err, "load reserved capacity template failed")
	}
	if template.GetStatus() != globalngv1.ReservedCapacityTemplate_ACTIVE {
		return bizerr.New(bizerr.ErrCodeParams, "reserved capacity template is not active")
	}
	if template.GetScope() == globalngv1.ReservedCapacityTemplate_ORG_PRIVATE && template.GetOrganizationId() != cluster.GetOrganizationId() {
		return bizerr.New(bizerr.ErrCodeParams, "reserved capacity template does not belong to organization")
	}
	if template.GetScope() != globalngv1.ReservedCapacityTemplate_DEFAULT && template.GetScope() != globalngv1.ReservedCapacityTemplate_ORG_PRIVATE {
		return bizerr.New(bizerr.ErrCodeParams, "reserved capacity template scope is invalid")
	}
	if template.GetFixedTopology() == nil || len(template.GetFixedTopology().GetComponents()) == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "reserved capacity template fixed topology cannot be empty")
	}
	if cluster.GetMinRcu() != 0 || cluster.GetMaxRcu() != 0 {
		return bizerr.New(bizerr.ErrCodeParams, "min_rcu and max_rcu must not be set for premium reserved clusters")
	}
	cluster.ReservedCapacityTemplate = template
	cluster.MinRcu = 0
	cluster.MaxRcu = 0
	return nil
}

func (s *serviceImpl) checkTenantBillingPlan(
	ctx context.Context,
	tenantId uint64,
	cloudProvider globalcommonv1.CloudProvider_Name,
	regionName globalcommonv1.Region_Name,
) bizerr.BizError {
	result, err := s.billingCli.BillingActionCheck(ctx, billing.Action_Create, tenantId, &billing.Resource{
		Type:     billing.ResourceType_NextgenCluster,
		Region:   lo.ToPtr(regionName.String()),
		Provider: lo.ToPtr(billing.ProviderMap[cloudProvider]),
		IsFree:   false,
	})
	if err != nil {
		return err
	}
	if result.Pass {
		return nil
	}
	if result.Reason != nil {
		return bizerr.New(bizerr.ErrCodeParams, "%s", *result.Reason)
	}
	return bizerr.New(bizerr.ErrCodeParams, "Check tenant billing plan failed")
}