package resourcepool

import (
	"context"
	"errors"
	"maps"
	"strconv"
	"strings"

	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/protobuf/proto"

	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/entity"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/region"
	topologysvc "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/topology"
	"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 (
	defaultRootPasswdLen = 21
)

func (s *serviceImpl) CreateResourcePool(ctx context.Context,
	request *globalngv1.CreateResourcePoolRequest,
) (*entity.ResourcePoolProfile, bizerr.BizError) {
	return s.createResourcePool(ctx, request, nil)
}

func (s *serviceImpl) createResourcePool(ctx context.Context,
	request *globalngv1.CreateResourcePoolRequest,
	topologySpec *clusterngv1beta.TopologySpec,
) (*entity.ResourcePoolProfile, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "Domain.CreateResourcePool")
	defer span.End()

	if request == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "request cannot be nil")
	}

	if err := validateResourcePoolOwnership(request.GetResourcePool()); err != nil {
		return nil, err
	}
	if err := validateCloudResourceTagSet(request.GetResourcePool().GetCloudResourceTags()); err != nil {
		return nil, err
	}
	if err := validateCloudResourceTagsForProvider(
		request.GetResourcePool().GetProviderId(),
		request.GetResourcePool().GetCloudResourceTags(),
		nil,
	); err != nil {
		return nil, err
	}
	if _, err := resolveCreateSharedPoolPdDeploymentMode(request.GetPdDeploymentMode(), topologySpec != nil); err != nil {
		return nil, err
	}
	if _, err := tikvRemoteWorkersToRegional(request.GetEnabledTikvRemoteWorkers()); err != nil {
		return nil, err
	}
	if _, err := tiflashModeToRegional(request.GetTiflashMode()); err != nil {
		return nil, err
	}

	fillingCreateRequestWithDefault(ctx, request, s.serverEnv(), topologySpec == nil)
	if err := normalizeResourcePoolHighAvailability(request.GetResourcePool()); err != nil {
		return nil, err
	}
	if err := validateResourcePoolZoneConfig(request.GetResourcePool()); err != nil {
		return nil, err
	}

	cloudRegion, err := s.regionRepo.SelectRegion(ctx, region.Filter{
		CloudProvider: request.GetResourcePool().ProviderId,
		RegionId:      request.GetResourcePool().RegionId,
		LogicalName:   request.GetLogicalRegion(),
		TenantId:      resourcePoolTenantId(request.GetResourcePool()),
		ServicePlan:   request.GetResourcePool().ServicePlan,
	})

	if err != nil {
		return nil, bizerr.WrapError(err, "select region failed")
	}
	if cloudRegion == nil {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "region not found")
	}
	if err := s.checkTenantBillingPlan(ctx, request.GetResourcePool(), cloudRegion); err != nil {
		return nil, err
	}

	var providerIdentity *globalngv1.RegionProviderIdentity
	if userIdentity := request.GetResourcePool().GetProviderIdentity(); userIdentity != "" {
		// User specified a provider identity — validate it exists in this region
		regionIdentity, identityErr := s.regionProviderIdentityRepo.GetProviderIdentityByIdentity(ctx, cloudRegion.RegionID, userIdentity)
		if identityErr != nil {
			return nil, bizerr.WrapError(identityErr, "provider_identity %q is not valid for the specified region", userIdentity)
		}
		providerIdentity = regionIdentity.RegionProviderIdentity
	} else {
		// Auto-pick provider identity using default logic
		var identityErr bizerr.BizError
		providerIdentity, identityErr = s.regionService.PickRegionProviderIdentity(
			ctx,
			cloudRegion.RegionID,
			resourcePoolTenantId(request.GetResourcePool()),
		)
		if identityErr != nil {
			return nil, bizerr.WrapError(identityErr, "pick provider identity failed")
		}
	}

	resourcePool := request.GetResourcePool()
	if request.GetVcpuLimitMode() != globalngv1.ResourcePool_VCPU_LIMIT_MODE_UNSPECIFIED {
		resourcePool.VcpuLimitMode = request.GetVcpuLimitMode()
	}
	if resourcePool.GetVcpusLimit() == 0 &&
		resourcePool.GetOwnerType() == globalngv1.ResourcePool_ORGANIZATION &&
		resourcePool.GetVcpuLimitMode() != globalngv1.ResourcePool_UNLIMITED {
		rcuQuota, err := s.rcuQuotaRepo.Load(ctx, resourcePool.GetOwnerId(), resourcePool.GetServicePlan().String())
		if err != nil {
			return nil, bizerr.WrapError(err, "load rcu quota failed")
		}
		resourcePool.VcpusLimit = rcuQuota.GetVcpusLimitPerResourcePool()
	}
	if err := validateBootstrapCapacityVcpusLimit(
		request.GetBootstrapCapacity(),
		resourcePool.GetVcpusLimit(),
	); err != nil {
		return nil, err
	}
	bootstrapPreflight, err := s.preflightBootstrapCreate(ctx, request, cloudRegion)
	if err != nil {
		return nil, err
	}
	var cidrPreflight *cidrValidationResult
	if usesCustomerCIDRNamespace(resourcePool) {
		cidrPreflight, err = s.validateResourcePoolCIDR(
			ctx,
			resourcePool.GetOwnerId(),
			cloudRegion,
			resourcePool.GetCidr(),
			resourcePool.GetHighAvailabilityType(),
		)
		if err != nil {
			return nil, err
		}
		if len(cidrPreflight.findings) > 0 {
			return nil, newCIDRValidationFailure(cidrPreflight.findings)
		}
		resourcePool.Cidr = cidrPreflight.canonicalCIDR
	}
	resourcePool.CreatedBy = strings.TrimSpace(request.GetCreatedBy())
	if request.GetValidateOnly() {
		if err := s.validateOrganizationNamespaceLocked(ctx, resourcePool, 0, false); err != nil {
			return nil, err
		}
		resourcePool.Id = 0
		resourcePool.ProviderIdentityId = 0
		resourcePool.DataplaneId = ""
		var capacity *globalngv1.ResourcePoolCapacity
		var o11yClusterCIDR *string
		if bootstrapPreflight != nil {
			capacity = bootstrapPreflight.capacity
		}
		if cidrPreflight != nil {
			o11yClusterCIDR = proto.String(cidrPreflight.o11yClusterCIDR)
		}
		return &globalngv1.ResourcePoolProfile{
			ResourcePool:    resourcePool,
			Region:          cloudRegion.Region,
			Capacity:        capacity,
			O11YClusterCidr: o11yClusterCIDR,
		}, nil
	}
	// Lifecycle is persisted uniformly; the public converter exposes the real
	// state regardless of service plan.
	resourcePool.LifecycleState = globalngv1.ResourcePool_CREATING
	resourcePool.StateMessageCode = ""

	resourcePool.Id = s.idGeneratorAdapter.MustNew()
	resourcePool.ProviderIdentityId = providerIdentity.Id

	terr := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		if namespaceErr := s.lockAndValidateOrganizationNamespaceForCreate(ctx, resourcePool); namespaceErr != nil {
			return namespaceErr
		}

		// 1. create system tidb password
		rootPassword := randstr.GenWithCharset(defaultRootPasswdLen, randstr.AlphanumericCharset)
		encryptedRootPassword, err := s.encryptor.EncryptBase64(ctx, rootPassword)
		if err != nil {
			return bizerr.WrapError(err, "encrypt root password failed")
		}
		if passwdErr := s.passwordRepo.SetEncryptedPassword(
			ctx,
			resourcePool.Id,
			entity.SystemTidbId,
			entity.ClusterUserRoot,
			encryptedRootPassword,
		); passwdErr != nil {
			return passwdErr
		}

		maintenancePassword := randstr.GenWithCharset(defaultRootPasswdLen, randstr.AlphanumericCharset)
		encryptedMaintenancePassword, err := s.encryptor.EncryptBase64(ctx, maintenancePassword)
		if err != nil {
			return bizerr.WrapError(err, "encrypt maintain password failed")
		}
		if passwdErr := s.passwordRepo.SetEncryptedPassword(
			ctx,
			resourcePool.Id,
			entity.SystemTidbId,
			entity.ClusterUserMaintenanceAdmin,
			encryptedMaintenancePassword,
		); passwdErr != nil {
			return passwdErr
		}

		platformPassword := randstr.GenWithCharset(defaultRootPasswdLen, randstr.AlphanumericCharset)
		encryptedPlatformPassword, err := s.encryptor.EncryptBase64(ctx, platformPassword)
		if err != nil {
			return bizerr.WrapError(err, "encrypt platform password failed")
		}
		if passwdErr := s.passwordRepo.SetEncryptedPassword(
			ctx,
			resourcePool.Id,
			entity.SystemTidbId,
			entity.ClusterUserPlatformAdmin,
			encryptedPlatformPassword,
		); passwdErr != nil {
			return passwdErr
		}

		cloudAdminPassword := randstr.GenWithCharset(defaultRootPasswdLen, randstr.AlphanumericCharset)
		encryptedCloudAdminPassword, err := s.encryptor.EncryptBase64(ctx, cloudAdminPassword)
		if err != nil {
			return bizerr.WrapError(err, "encrypt cloud_admin password failed")
		}
		if passwdErr := s.passwordRepo.SetEncryptedPassword(
			ctx,
			resourcePool.Id,
			entity.SystemTidbId,
			entity.ClusterUserCloudAdmin,
			encryptedCloudAdminPassword,
		); passwdErr != nil {
			return passwdErr
		}

		// 2. create resource pool
		entityResourcePool := &entity.ResourcePool{
			ResourcePool:    resourcePool,
			RegionId:        cloudRegion.RegionID,
			CloudProviderId: cloudRegion.ProviderID,
		}
		err = s.resourcePoolRepo.Create(ctx, entityResourcePool)
		if err != nil {
			return bizerr.WrapError(err, "create resource pool failed")
		}

		// Only organization-owned pools participate in default-pool strategy.
		if !isSharedResourcePool(resourcePool) {
			if shouldManageResourcePoolStrategy(resourcePool.GetServicePlan()) {
				_, err = s.setDefaultPoolIfMissing(
					ctx,
					cloudRegion,
					resourcePool.OwnerId,
					resourcePool.Id,
					resourcePool.ServicePlan,
				)
				if err != nil {
					return bizerr.WrapError(err, "set default resource pool failed")
				}
			}
		}

		// 4. create resource pool dag
		// gen 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.PostCreateResourcePool,
			strconv.FormatUint(resourcePool.OwnerId, 10),
			"0", // nextgen cluster don't have project id
			strconv.FormatUint(resourcePool.Id, 10),
			map[string]string{
				workflow.KeyTraceId:      workflow.GetTraceId(ctx),
				workflow.KeyRequestMeta:  string(metadataJson),
				common.KeyResourcePoolId: strconv.FormatUint(resourcePool.Id, 10),
			},
		)
		if workflowErr != nil {
			return bizerr.WrapErrorWithCode(workflowErr, bizerr.ErrCodeUnknown, "create async workflow failed")
		}

		createSharedPoolRequest, err := buildCreateSharedPoolRequest(
			request,
			resourcePool.Id,
			providerIdentity.Identity,
			topologySpec,
		)
		if err != nil {
			return bizerr.WrapError(err, "build create shared pool request failed")
		}

		eventAggr := &entity.ResourcePoolEventAggregation{
			ResourcePool:              entityResourcePool,
			CreateResourcePoolRequest: request,
			CreateSharedPoolRequest:   createSharedPoolRequest,
		}
		events, err := eventAggr.CreateEvent(ctx)
		if err != nil {
			return bizerr.WrapError(err, "build create event failed")
		}
		if err := s.eventCommitter.BatchWrite(ctx, clusterevent.TopicClusterEvent, events); err != nil {
			return bizerr.WrapError(err, "write cluster event failed")
		}
		sharedPool, err := s.regionalCli.CreateSharedPool(ctx, cloudRegion, createSharedPoolRequest)
		if err != nil {
			return bizerr.WrapError(err, "create shared pool in regional server failed")
		}

		err = s.resourcePoolRepo.UpdateDataplaneId(ctx, resourcePool.Id, sharedPool.GetDataplaneName())
		if err != nil {
			return bizerr.WrapError(err, "update resource pool dataplane id failed")
		}

		return nil
	})

	if terr != nil {
		tilog.Error(ctx, "create resource pool failed", zap.Error(terr))
		var cidrFailure *CIDRValidationFailure
		if errors.As(terr, &cidrFailure) {
			return nil, cidrFailure
		}
		// if the terr is a bizerr.BizError
		var be bizerr.BizError
		if errors.As(terr, &be) {
			return nil, be
		}
		return nil, bizerr.WrapGormError(terr, "create resource pool failed")
	}

	resourcePoolDetail, detailErr := s.GetResourcePool(ctx, &globalngv1.GetResourcePoolRequest{
		ResourcePoolId: resourcePool.Id,
	})

	if detailErr != nil {
		tilog.Error(ctx, "create resource pool failed to get response", zap.Error(detailErr))
		return nil, detailErr
	}

	return resourcePoolDetail, nil
}

func normalizeResourcePoolHighAvailability(resourcePool *globalngv1.ResourcePool) bizerr.BizError {
	if resourcePool == nil {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool cannot be nil")
	}
	switch resourcePool.GetHighAvailabilityType() {
	case globalngv1.ResourcePool_REGIONAL, globalngv1.ResourcePool_ZONAL:
		return nil
	case globalngv1.ResourcePool_HIGH_AVAILABILITY_TYPE_UNSPECIFIED:
		if usesCustomerCIDRNamespace(resourcePool) {
			resourcePool.HighAvailabilityType = globalngv1.ResourcePool_REGIONAL
		}
		return nil
	default:
		return bizerr.New(bizerr.ErrCodeParams, "high availability type is invalid")
	}
}

func buildCreateSharedPoolRequest(crrp *globalngv1.CreateResourcePoolRequest,
	resourcePoolId uint64,
	providerIdentity string,
	topologySpec *clusterngv1beta.TopologySpec,
) (*clusterngv1beta.CreateSharedPoolRequest, bizerr.BizError) {
	pdDeploymentMode, err := resolveCreateSharedPoolPdDeploymentMode(crrp.GetPdDeploymentMode(), topologySpec != nil)
	if err != nil {
		return nil, err
	}
	enabledTikvRemoteWorkers, err := tikvRemoteWorkersToRegional(crrp.GetEnabledTikvRemoteWorkers())
	if err != nil {
		return nil, err
	}
	tiflashMode, err := tiflashModeToRegional(crrp.GetTiflashMode())
	if err != nil {
		return nil, err
	}
	if crrp.GetResourcePool().GetServicePlan() == globalcommonv1.ServicePlan_Premium_Reserved {
		tiflashMode = clusterngv1beta.TiflashMode_TIFLASH_MODE_COLUMNAR
	}

	sharedPool := &clusterngv1beta.SharedPool{
		Id:                    strconv.FormatUint(resourcePoolId, 10),
		PlacementGroupEnabled: crrp.GetResourcePool().GetPlacementGroupEnabled(),
		CloudResourceTags:     maps.Clone(crrp.GetResourcePool().GetCloudResourceTags()),
	}
	if pcaArn := crrp.GetResourcePool().GetPcaArn(); pcaArn != "" {
		sharedPool.Pca = &clusterngv1beta.SharedPool_PCA{
			CaIdentifier: pcaArn,
		}
	}

	request := &clusterngv1beta.CreateSharedPoolRequest{
		SharedPool:               sharedPool,
		TenantId:                 resourcePoolTenantID(crrp.GetResourcePool()),
		VcpusLimit:               crrp.GetResourcePool().GetVcpusLimit(),
		PdDeploymentMode:         pdDeploymentMode,
		EnabledTikvRemoteWorkers: enabledTikvRemoteWorkers,
		TiflashMode:              tiflashMode,
		EnableCapacityPlanner:    crrp.GetEnableCapacityPlanner(),
		InitSharedPoolConfig: &clusterngv1beta.CreateSharedPoolRequest_InitSharedPoolConfig{
			NetworkCidr:               crrp.GetResourcePool().GetCidr(),
			ServiceCidr:               crrp.GetResourcePool().GetK8SServiceCidr(),
			ConfigCenterMetaProfileId: crrp.GetResourcePool().GetConfigMetaProfileId(),
			CloudIdentity:             providerIdentity,
			NodeZones:                 crrp.GetResourcePool().GetZoneConfig().GetNodeZones(),
			HighAvailabilityType:      convertResourcePoolHAType(crrp.GetResourcePool().GetHighAvailabilityType()),
			TikvDataStorage: &clusterngv1beta.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
				Type:           crrp.Demand.TikvData.StorageType.String(),
				SizeGi:         crrp.Demand.TikvData.DiskSizeGi,
				Iops:           crrp.Demand.TikvData.DiskIops,
				ThroughputMbps: crrp.Demand.TikvData.DiskThroughputMbps,
			},
			TikvRaftlogStorage: &clusterngv1beta.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
				Type:           crrp.Demand.TikvRaftlog.StorageType.String(),
				SizeGi:         crrp.Demand.TikvRaftlog.DiskSizeGi,
				Iops:           crrp.Demand.TikvRaftlog.DiskIops,
				ThroughputMbps: crrp.Demand.TikvRaftlog.DiskThroughputMbps,
			},
			TiflashCnStorage: &clusterngv1beta.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
				Type:           crrp.Demand.TiflashCn.StorageType.String(),
				SizeGi:         crrp.Demand.TiflashCn.DiskSizeGi,
				Iops:           crrp.Demand.TiflashCn.DiskIops,
				ThroughputMbps: crrp.Demand.TiflashCn.DiskThroughputMbps,
			},
			TiflashWnStorage: &clusterngv1beta.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
				Type:           crrp.Demand.TiflashWn.StorageType.String(),
				SizeGi:         crrp.Demand.TiflashWn.DiskSizeGi,
				Iops:           crrp.Demand.TiflashWn.DiskIops,
				ThroughputMbps: crrp.Demand.TiflashWn.DiskThroughputMbps,
			},
		},
	}
	if topologySpec != nil {
		request.CapacityMode = clusterngv1beta.CapacityMode_TOPOLOGY_DRIVEN
		request.CapacitySpec = &clusterngv1beta.CreateSharedPoolRequest_TopologySpec{
			TopologySpec: topologySpec,
		}
	} else {
		request.CapacityMode = clusterngv1beta.CapacityMode_RCU_DRIVEN
		if crrp.GetBootstrapCapacity() != nil {
			request.EnforceO11YNetworkNonOverlap = true
			request.BootstrapCapacity = bootstrapCapacityToRegional(crrp.GetBootstrapCapacity())
		} else {
			request.CapacitySpec = &clusterngv1beta.CreateSharedPoolRequest_RequestUnit{
				RequestUnit: &clusterngv1beta.Cluster_RequestUnit{
					Min: crrp.Demand.InitialRcu,
					Max: crrp.Demand.InitialRcu,
				},
			}
		}
	}

	return request, nil
}

func convertResourcePoolHAType(
	haType globalngv1.ResourcePool_HighAvailabilityType,
) clusterngv1beta.CreateSharedPoolRequest_InitSharedPoolConfig_HighAvailabilityType {
	switch haType {
	case globalngv1.ResourcePool_REGIONAL:
		return clusterngv1beta.CreateSharedPoolRequest_InitSharedPoolConfig_REGIONAL
	case globalngv1.ResourcePool_ZONAL:
		return clusterngv1beta.CreateSharedPoolRequest_InitSharedPoolConfig_ZONAL
	default:
		return clusterngv1beta.CreateSharedPoolRequest_InitSharedPoolConfig_HIGH_AVAILABILITY_TYPE_UNSPECIFIED
	}
}

func (s *serviceImpl) createTopologyDrivenResourcePool(ctx context.Context,
	request *globalngv1.CreateResourcePoolRequest,
	fixedTopology *globalngv1.FixedTopology,
) (*entity.ResourcePoolProfile, bizerr.BizError) {
	fixedTopology = withDefaultStorageTypeForSharedPool(fixedTopology, request.GetResourcePool().GetProviderId())
	topologySpec, err := topologysvc.BuildTopologySpec(fixedTopology, topologysvc.SharedPoolComponentFilter)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "build topology spec failed")
	}
	return s.createResourcePool(ctx, request, topologySpec)
}

func withDefaultStorageTypeForSharedPool(
	fixedTopology *globalngv1.FixedTopology,
	provider globalcommonv1.CloudProvider_Name,
) *globalngv1.FixedTopology {
	if fixedTopology == nil {
		return nil
	}
	defaultStorageType := defaultStorageTypeByProvider(provider)
	if defaultStorageType == "" {
		return fixedTopology
	}
	cloned, ok := proto.Clone(fixedTopology).(*globalngv1.FixedTopology)
	if !ok {
		return fixedTopology
	}
	for _, component := range cloned.GetComponents() {
		if !topologysvc.SharedPoolComponentFilter(component.GetComponentType()) {
			continue
		}
		for _, storage := range component.GetStorages() {
			if storage != nil && storage.GetType() == "" {
				storage.Type = defaultStorageType
			}
		}
	}
	return cloned
}

func defaultStorageTypeByProvider(provider globalcommonv1.CloudProvider_Name) string {
	switch provider {
	case globalcommonv1.CloudProvider_AWS:
		return "gp3"
	case globalcommonv1.CloudProvider_ALICLOUD:
		return "cloud-auto"
	default:
		return ""
	}
}

func validateResourcePoolOwnership(resourcePool *globalngv1.ResourcePool) bizerr.BizError {
	if resourcePool == nil {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool cannot be nil")
	}

	switch resourcePool.GetOwnerType() {
	case globalngv1.ResourcePool_ORGANIZATION:
		if resourcePool.GetOwnerId() == 0 {
			return bizerr.New(bizerr.ErrCodeParams, "owner_id must be set for organization resource pool")
		}
	case globalngv1.ResourcePool_SHARED:
		if resourcePool.GetOwnerId() != 0 {
			return bizerr.New(bizerr.ErrCodeParams, "owner_id must be empty for shared resource pool")
		}
	default:
		return bizerr.New(bizerr.ErrCodeParams, "unsupported owner_type: %s", resourcePool.GetOwnerType().String())
	}

	return nil
}

func validateResourcePoolZoneConfig(resourcePool *globalngv1.ResourcePool) bizerr.BizError {
	if resourcePool == nil {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool cannot be nil")
	}

	nodeZones := resourcePool.GetZoneConfig().GetNodeZones()
	if len(nodeZones) == 0 {
		return nil
	}

	seen := make(map[string]struct{}, len(nodeZones))
	for _, zone := range nodeZones {
		if zone == "" {
			return bizerr.New(bizerr.ErrCodeParams, "zone_config.node_zones cannot contain empty zone")
		}
		if _, ok := seen[zone]; ok {
			return bizerr.New(bizerr.ErrCodeParams, "zone_config.node_zones contains duplicate zone %q", zone)
		}
		seen[zone] = struct{}{}
	}

	switch resourcePool.GetHighAvailabilityType() {
	case globalngv1.ResourcePool_ZONAL:
		if len(nodeZones) != 1 {
			return bizerr.New(bizerr.ErrCodeParams, "zone_config.node_zones must contain exactly 1 zone for zonal deployment")
		}
	case globalngv1.ResourcePool_REGIONAL:
		if len(nodeZones) != 3 {
			return bizerr.New(bizerr.ErrCodeParams, "zone_config.node_zones must contain exactly 3 zones for regional deployment")
		}
	default:
		return bizerr.New(bizerr.ErrCodeParams, "high_availability_type must be set when zone_config.node_zones is specified")
	}

	return nil
}

func isSharedResourcePool(resourcePool *globalngv1.ResourcePool) bool {
	return resourcePool.GetOwnerType() == globalngv1.ResourcePool_SHARED &&
		resourcePool.GetOwnerId() == 0
}

func resourcePoolTenantID(resourcePool *globalngv1.ResourcePool) string {
	if isSharedResourcePool(resourcePool) {
		return ""
	}
	return strconv.FormatUint(resourcePool.GetOwnerId(), 10)
}