package server

import (
	"context"
	"encoding/json"
	"fmt"
	"net"
	"sort"
	"strconv"
	"strings"

	"github.com/google/uuid"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	"github.com/tidbcloud/cluster-service-ng/pkg/clusterconfig"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/schema"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
)

func (s *serverImpl) CreateSharedPool(
	ctx context.Context,
	sharedPool *entity.SharedPool,
	requestUint *entity.RequestUnit,
	initConfig *entity.InitSharedPoolConfig,
	optionList ...*CreateSharedPoolOptions,
) (result *entity.SharedPool, resultErr error) {
	options, err := normalizeCreateSharedPoolOptions(optionList)
	if err != nil {
		return nil, err
	}
	if sharedPool == nil {
		return nil, status.Error(codes.InvalidArgument, "sharedPool is required")
	}
	bootstrapCapacity := options.BootstrapCapacity
	if bootstrapCapacity != nil {
		compute := "disabled"
		if bootstrapCapacity.TiFlashCompute != nil {
			compute = "enabled"
		}
		defer func() {
			outcome := "success"
			if resultErr != nil {
				outcome = "error"
			}
			tilog.Info(ctx, "completed shared-pool create with bootstrap capacity",
				zap.String("sharedPoolId", sharedPool.GetId()),
				zap.String("tiflashCompute", compute),
				zap.String("outcome", outcome),
				zap.Error(resultErr),
			)
		}()
	}

	if sharedPool.GetId() == "" {
		return nil, status.Errorf(codes.InvalidArgument, "sharedPool id is required")
	}

	// Validate sharedPool id format
	if err := entity.ValidateLowercaseAlphanumericId("shared_pool_id", sharedPool.GetId()); err != nil {
		return nil, status.Errorf(codes.InvalidArgument, "%s", err.Error())
	}

	if err := validateCreateSharedPoolCapacityMode(sharedPool); err != nil {
		return nil, err
	}

	// set sharedPool status to CREATING
	sharedPool.Phase = apiv1.SharedPool_CREATING
	sharedPool.CapacityMode = entity.GetEffectiveCapacityMode(sharedPool.GetCapacityMode())
	if err := validateCreateSharedPoolCapacityInput(sharedPool, requestUint, bootstrapCapacity); err != nil {
		return nil, err
	}
	// CreateSharedPool owns this persisted marker. Start from Legacy; the RCU
	// path resolves it later from the request and effective TiKV image.
	sharedPool.TiflashMode = schema.SharedPoolTiflashModeDefault
	if err := validateCreateSharedPoolTiflashMode(sharedPool); err != nil {
		return nil, err
	}
	// Topology-driven pools skip resolveCreateSharedPoolTiflashMode, so persist
	// their explicit Columnar choice before topology generation reads the marker.
	if sharedPool.GetCapacityMode() == apiv1.CapacityMode_TOPOLOGY_DRIVEN &&
		sharedPool.RequestedTiflashMode == apiv1.TiflashMode_TIFLASH_MODE_COLUMNAR {
		sharedPool.TiflashMode = schema.SharedPoolTiflashModeColumnar
	}

	if err := normalizeAndValidateCreateSharedPoolEnabledTiKVRemoteWorkers(sharedPool); err != nil {
		return nil, err
	}
	sharedPool.PdDeploymentMode = entity.GetEffectivePDDeploymentMode(sharedPool.GetPdDeploymentMode())
	if err := validateCreateSharedPoolPDDeploymentMode(sharedPool); err != nil {
		return nil, err
	}
	profileID := initConfig.GetConfigCenterMetaProfileId()
	if _, err := clusterconfig.ParseMetaProfileID(profileID); err != nil {
		return nil, status.Errorf(codes.InvalidArgument, "invalid config center meta profile id: %s", err)
	}
	sharedPool.ConfigCenterMetaProfileId = profileID

	// check sharedPool
	err = s.checkCreateSharedpool(ctx, sharedPool, requestUint)
	if err != nil {
		return nil, err
	}

	err = checkInitSharedPoolConfig(s.cloudProvider, sharedPool.GetCapacityMode(), sharedPool.GetTopologySpec(), initConfig)
	if err != nil {
		return nil, err
	}
	haType := s.resolveCreateSharedPoolHAType(sharedPool, initConfig.GetHighAvailabilityType())
	sharedPool.HighAvailabilityType = haType
	initConfig.HighAvailabilityType = sharedPoolRequestHAFromCanonical(haType)

	// Reject illegal HA + zone combinations upfront so the SharedPool is
	// never persisted with a wrong gate and the error message names the
	// caller-side cause. Mirrors DPM resolveNodeZonesForCreate at
	// dataplaneservice.go:438-441 and 484-488, which would otherwise reject
	// later at CreateDataPlane time. CSNG's pre-validation is also what lets
	// the gate derivation below assume a legal shape (REGIONAL is always
	// multi-AZ; ZONAL is always 0 or 1 zone).
	switch haType {
	case apiv1.SharedPool_REGIONAL:
		if len(initConfig.GetNodeZones()) == 1 {
			return nil, status.Errorf(codes.InvalidArgument,
				"init_shared_pool_config.node_zones with 1 zone means a zonal deployment; use high_availability_type=ZONAL or pass 2+ zones for REGIONAL")
		}
	case apiv1.SharedPool_ZONAL:
		if len(initConfig.GetNodeZones()) > 1 {
			return nil, status.Errorf(codes.InvalidArgument,
				"init_shared_pool_config.node_zones must contain at most 1 zone for zonal deployment")
		}
	}

	if len(sharedPool.GetCloudResourceTags()) > 0 {
		if err := validateCloudResourceTagsPatch(sharedPool.GetCloudResourceTags(), nil); err != nil {
			return nil, err
		}
	}
	var bootstrapNodeSpecMap map[string][]*entity.NodeSpec
	var bootstrapRuConfig *entity.RuConfig
	if bootstrapCapacity != nil {
		nodeSpecs, dropped, err := s.listDefaultNodeSpecs(ctx)
		if err != nil {
			return nil, resourcePoolStatusError(codes.Unavailable, reasonNodeSpecsUnavailable, "node specs are unavailable")
		}
		logNodeSpecDrops(ctx, dropped)
		if len(nodeSpecs) == 0 {
			return nil, resourcePoolStatusError(codes.FailedPrecondition, reasonNodeSpecsUnavailable, "no usable node specs are configured")
		}
		bootstrapRuConfig, err = s.resolveSharedPoolRuConfig(ctx, sharedPool.TenantId)
		if err != nil {
			return nil, err
		}
		resolvedBootstrap, err := validateAndResolveBootstrapCapacity(bootstrapCapacity, initConfig, nodeSpecs, bootstrapRuConfig)
		if err != nil {
			return nil, err
		}
		if _, err := deriveBootstrapRURange(resolvedBootstrap, bootstrapRuConfig); err != nil {
			return nil, err
		}
		bootstrapCapacity = resolvedBootstrap
		bootstrapNodeSpecMap = nodeSpecMapFromList(nodeSpecs)
		tilog.Info(ctx, "validated shared-pool bootstrap capacity",
			zap.String("sharedPoolId", sharedPool.GetId()),
			zap.Any("tikvNodeSpec", resolvedBootstrap.TiKV.NodeSpec),
			zap.Uint32("tikvNodeCount", resolvedBootstrap.TiKV.NodeCount),
			zap.Bool("tiflashComputeEnabled", resolvedBootstrap.TiFlashCompute != nil),
		)
	}
	if err := s.validateO11yNetworkNonOverlap(ctx, initConfig.GetNetworkCidr(), options.EnforceO11yNetworkNonOverlap); err != nil {
		return nil, err
	}

	if sharedPool.GetDataplaneName() == "" {
		if sharedPool.GetProvider().GetName() != "" {
			sharedPool.DataplaneName = sharedPool.GetProvider().GetName()
		} else {
			sharedPool.DataplaneName = genDataplaneName()
			sharedPool.Provider = &entity.DataPlaneProvider{Name: sharedPool.GetDataplaneName()}

			switch strings.ToUpper(s.cloudProvider) {
			case apiv1.ProviderType_AWS.String():
				sharedPool.Provider = &apiv1.Provider{
					Type:   apiv1.ProviderType_AWS,
					Region: s.region,
				}
			case apiv1.ProviderType_GCP.String():
				return nil, status.Errorf(codes.Unimplemented, "GCP cloud provider is not supported yet")
			case apiv1.ProviderType_AZURE.String():
				sharedPool.Provider = &apiv1.Provider{
					Type:   apiv1.ProviderType_AZURE,
					Region: s.region,
				}
			case apiv1.ProviderType_ALICLOUD.String():
				sharedPool.Provider = &apiv1.Provider{
					Type:   apiv1.ProviderType_ALICLOUD,
					Region: s.region,
				}
			default:
				return nil, status.Errorf(codes.Unimplemented, "cloud provider %s is not supported", s.cloudProvider)
			}
			tilog.Warn(ctx, "dataplane name is empty, will auto create new one",
				zap.String("sharedPoolId", sharedPool.GetId()),
				zap.String("dataplaneName", sharedPool.GetDataplaneName()),
			)
		}
	}
	if sharedPool.Provider == nil {
		sharedPool.Provider = &apiv1.Provider{}
	}
	// Provider.PlacementGroupEnabled is the durable per-pool gate shared by
	// NodePool infrastructure, component topology, NIG creation, and
	// shortage-detector fallback. PR #3040 derives it from the canonical
	// public input at creation so every consumer of the shared pool reads
	// the same value. The check is "is this AWS and single-AZ?" — only EKS
	// supports EC2 partition placement groups, and single-AZ is the
	// precondition. Single-AZ pools on Azure/GCP/ACK are valid (the cloud's
	// native zone/availability-set handles intra-AZ placement), they just
	// must not opt into the partition path.
	//
	// The HA + caller-zone shape is pre-validated above (REGIONAL+1 and
	// ZONAL+>1 are rejected with InvalidArgument), so by the time we reach
	// the gate derivation we know:
	//   - REGIONAL: caller sent 0 or 2+ zones. Effective topology is
	//     multi-AZ (caller zones dropped before the DAG, then DPM auto-fills
	//     from regionConfig.zones), so gate=false.
	//   - ZONAL: caller sent 0 or 1 zone. DPM auto-fills from
	//     regionConfig.nodeZones (1 zone) when caller sent 0; effective
	//     topology is single-AZ. gate=true iff provider is AWS.
	//
	// The caller-supplied SharedPool.placement_group_enabled field is
	// accepted but ignored: PR #3040 makes the topology check the source of
	// truth regardless of any input flag. Legacy shared pools (created
	// before this change) keep their original persisted value: this code
	// only runs for new CreateSharedPool calls, so the persisted field is
	// never overwritten for an existing pool.
	sharedPool.Provider.PlacementGroupEnabled = false
	if haType != apiv1.SharedPool_REGIONAL {
		// ZONAL + (0 or 1 caller zone) → effective single-AZ → gate follows provider.
		providerType := sharedPool.GetProvider().GetType()
		if providerType == apiv1.ProviderType_PROVIDER_TYPE_UNSPECIFIED {
			switch strings.ToUpper(s.cloudProvider) {
			case apiv1.ProviderType_AWS.String():
				providerType = apiv1.ProviderType_AWS
			case apiv1.ProviderType_GCP.String():
				providerType = apiv1.ProviderType_GCP
			case apiv1.ProviderType_AZURE.String():
				providerType = apiv1.ProviderType_AZURE
			case apiv1.ProviderType_ALICLOUD.String():
				providerType = apiv1.ProviderType_ALICLOUD
			}
		}
		sharedPool.Provider.PlacementGroupEnabled = providerType == apiv1.ProviderType_AWS
	}

	sharedPool.ObjectStorage = &entity.ObjectStorage{
		Bucket: entity.GetDataBucketId(sharedPool.GetId(), s.region),
		Prefix: "",
		Region: s.region,
	}

	sharedPool.InternalStorage = &entity.ObjectStorage{
		Bucket: entity.GetInternalBucketId(sharedPool.GetId(), s.region),
		Prefix: "",
		Region: s.region,
	}

	var dagExecId string
	var dagSubmitted bool
	if err := s.transactioner.Transaction(ctx, func(ctx context.Context) error {
		// check if the sharedPool is locked
		unlockFunc, err := s.lock(ctx, sharedPool.GetId())
		if err != nil {
			tilog.Error(ctx, "failed to lock sharedPool", zap.String("sharedPoolId", sharedPool.GetId()), zap.Error(err))
			return err
		}
		defer func() {
			if err := unlockFunc(); err != nil {
				tilog.Error(ctx, "failed to unlock sharedPool", zap.String("sharedPoolId", sharedPool.GetId()), zap.Error(err))
			}
		}()

		// first create the sharedPool, dataplane is not created yet
		// so we need get default nodespec map
		enableTici := false
		if sharedPool.GetCapacityMode() == apiv1.CapacityMode_TOPOLOGY_DRIVEN {
			nodeSpecMap, err := s.genNodeSpecMap(ctx, sharedPool)
			if err != nil {
				tilog.Error(ctx, "failed to get node spec map", zap.String("dataplaneId", sharedPool.GetDataplaneName()), zap.Error(err))
				return status.Errorf(codes.Internal, "failed to get node spec map: %s", err)
			}

			if err := s.generateTopologyDrivenComponents(ctx, sharedPool, requestUint, nodeSpecMap, initConfig); err != nil {
				return err
			}
		} else {
			nodeSpecMap := bootstrapNodeSpecMap
			if nodeSpecMap == nil {
				nodeSpecMap, err = s.genNodeSpecMap(ctx, sharedPool)
				if err != nil {
					tilog.Error(ctx, "failed to get node spec map", zap.String("dataplaneId", sharedPool.GetDataplaneName()), zap.Error(err))
					return status.Errorf(codes.Internal, "failed to get node spec map: %s", err)
				}
			}

			resolvedRuConfig := bootstrapRuConfig
			if resolvedRuConfig == nil {
				resolvedRuConfig, err = s.resolveSharedPoolRuConfig(ctx, sharedPool.TenantId)
				if err != nil {
					return err
				}
			}

			version, err := s.clusterConfigClient.GetDefaultVersion(ctx, sharedPool.TenantId)
			if err != nil {
				return status.Errorf(codes.Internal, "failed to get default version: %s", err)
			}
			sharedPool.Version = version

			if err := s.resolveCreateSharedPoolTiflashMode(ctx, sharedPool); err != nil {
				return err
			}

			// generate sharedPool component pools from the RESOLVED
			// ru config and node spec map (so component generation
			// sees the frozen tenant-vs-default values, not the raw
			// tenant row).
			err = s.generatedSharedPoolComponent(ctx, sharedPool, requestUint, nodeSpecMap, resolvedRuConfig, initConfig, false, bootstrapCapacity)
			if err != nil {
				return err
			}

			// Persist the RESOLVED ru config (tenant-vs-default
			// frozen + bundle-table policy flags overlaid), not the
			// raw tenant row — otherwise the freeze computed above is
			// materialized into the component pools but never written
			// to the shared pool's persisted ru_configs row, so the
			// RCU cron (which reads the persisted SHARED_POOL row)
			// silently ignores bundle-set policy flags.
			if err := s.ruConfigRepo.SetRuConfig(ctx, sharedPool.GetId(), apiv1.RuConfig_SHARED_POOL, resolvedRuConfig); err != nil {
				tilog.Error(ctx, "failed to save ruConfig to sharedPool", zap.Error(err))
				return fmt.Errorf("failed to save ruConfig to sharedPool: %w", err)
			}
			enableTici = resolvedRuConfig.GetFeatureFlags().GetEnableTici()
		}

		// Topology-driven pools do not use the TiFlash Columnar gate, so their
		// default version is resolved after topology generation as before.
		if sharedPool.GetVersion() == "" {
			version, err := s.clusterConfigClient.GetDefaultVersion(ctx, sharedPool.TenantId)
			if err != nil {
				return status.Errorf(codes.Internal, "failed to get default version: %s", err)
			}
			sharedPool.Version = version
		}
		sharedPool.Phase = apiv1.SharedPool_CREATING

		dataplaneAccountId := s.bizCfg.DefaultDataplaneAccount
		if initConfig.GetCloudIdentity() != "" {
			dataplaneAccountId = initConfig.GetCloudIdentity()
		}

		if err := s.sharedPoolRepo.Save(ctx, sharedPool); err != nil {
			tilog.Error(ctx, "failed to save sharedPool meta", zap.Error(err))
			return fmt.Errorf("failed to save sharedPool meta: %w", err)
		}

		// create sharedPool
		sharedPoolStr, err := protojson.Marshal(sharedPool)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to marshal sharedPool: %s", err)
		}

		args := actions.ActionArgs{
			workflow.DataplaneAccountIdKey:   actions.ActionArgString(dataplaneAccountId),
			workflow.DataplaneIdKey:          actions.ActionArgString(sharedPool.GetDataplaneName()),
			workflow.SharedPoolIdKey:         actions.ActionArgString(sharedPool.GetId()),
			workflow.SharedPoolKey:           actions.ActionArgString(string(sharedPoolStr)),
			workflow.SharedPoolDeployTypeKey: actions.ActionArgString(sharedPool.DeployType),
			// Enumerate every per-pool HPA-eligible component so
			// ApplyComponentHpaAction reaches each worker.
			// GetDefaultComponenId() returns the PD id and would
			// cause CheckIsTargetComponent to filter every worker
			// out — workers would never get HPAs in K8s.
			workflow.ComponentIdsKey:              actions.ActionArgString(strings.Join(sharedPool.GetPerPoolHPAComponentIds(), ",")),
			workflow.MeteringTypeKey:              actions.ActionArgString(workflow.MeteringTypeValueSharedPoolMeta),
			workflow.DataplaneCidrKey:             actions.ActionArgString(initConfig.GetNetworkCidr()),
			workflow.DataplaneServiceCidrKey:      actions.ActionArgString(initConfig.GetServiceCidr()),
			workflow.CapacityModeKey:              actions.ActionArgString(sharedPool.GetCapacityMode().String()),
			workflow.EnableTiCIKey:                actions.ActionArgString(strconv.FormatBool(enableTici)),
			workflow.ConfigCenterMetaProfileIdKey: actions.ActionArgString(initConfig.GetConfigCenterMetaProfileId()),
			workflow.EnableCapacityPlannerKey:     actions.ActionArgString(strconv.FormatBool(sharedPool.EnableCapacityPlanner)),
			utils.TraceIdKey:                      utils.GetTraceId(ctx),
		}

		if sharedPool.GetPca() != nil {
			pcaStr, err := protojson.Marshal(sharedPool.GetPca())
			if err != nil {
				return status.Errorf(codes.Internal, "failed to marshal pca: %s", err)
			}
			args[workflow.DataplanePCAKey] = actions.ActionArgString(string(pcaStr))
		}

		// For REGIONAL HA, drop caller-supplied node_zones. The DPM pre-validator
		// (dataplaneservice.go:298) rejects REGIONAL SharedPools when caller zones > 1, while
		// the new REGIONAL HA contract says caller sends 0 zones and DPM auto-fills from
		// regionConfig.zones. resolveCreateSharedPoolHAType resolves the HA type but does not
		// update this arg-build block, so without dropping caller zones here the executor DAG
		// passes them through verbatim and a REGIONAL pool can wedge in the DPM pre-validator.
		// ZONAL still passes caller zones through unchanged; REGIONAL logs a Warn so the silent
		// override is discoverable.
		switch {
		case len(initConfig.GetNodeZones()) == 0:
			// No caller zones. REGIONAL relies on DPM regionConfig.zones auto-fill (see above);
			// ZONAL without zones is the default and DPM will fail-downstream if invalid.
		case haType == apiv1.SharedPool_REGIONAL:
			tilog.Warn(ctx, "dropping caller-supplied node_zones for REGIONAL HA; DPM will auto-fill from regionConfig.zones",
				zap.String("sharedPoolId", sharedPool.GetId()),
				zap.Strings("callerZones", initConfig.GetNodeZones()))
		default:
			nodeZones, err := json.Marshal(initConfig.GetNodeZones())
			if err != nil {
				return status.Errorf(codes.Internal, "failed to marshal dataplane node zones: %s", err)
			}
			args[workflow.DataplaneNodeZonesKey] = actions.ActionArgString(string(nodeZones))
		}
		args[workflow.DataplaneHATypeKey] = actions.ActionArgString(haType.String())

		if sharedPool.TenantId != "" {
			args[workflow.TenantIdKey] = actions.ActionArgString(sharedPool.TenantId)
		}

		dagExecId = utils.GeneratedDagExecId(ctx, fmt.Sprintf("create-sharedpool-%s", sharedPool.GetId()))
		err = s.dagExecutor.Submit(ctx, dag.PlanCreateSharedPool, args, dagExecId)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to submit create sharedpool dag: %s", err)
		}
		dagSubmitted = true
		return nil
	}); err != nil {
		if dagSubmitted {
			if abortErr := s.dagExecutor.Abort(ctx, dagExecId); abortErr != nil {
				tilog.Error(ctx, "failed to abort dag after transaction failure",
					zap.String("dagExecId", dagExecId),
					zap.Error(abortErr))
			}
		}
		return nil, err
	}

	return sharedPool, nil
}

func (s *serverImpl) resolveCreateSharedPoolHAType(
	sharedPool *entity.SharedPool,
	requested apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_HighAvailabilityType,
) apiv1.SharedPool_HighAvailabilityType {
	switch requested {
	case apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_REGIONAL:
		return apiv1.SharedPool_REGIONAL
	case apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_ZONAL:
		return apiv1.SharedPool_ZONAL
	}

	plan := ""
	if s.meteringConfig != nil {
		plan = s.meteringConfig.MetadataPlan()
	}
	servicePlan := entity.ResolveServicePlan(sharedPool, sharedPool.GetCapacityMode(), plan)
	if servicePlan == entity.EssentialServicePlan {
		return apiv1.SharedPool_ZONAL
	}
	return apiv1.SharedPool_REGIONAL
}

// sharedPoolRequestHAFromCanonical translates the canonical output enum to the
// request enum used by initConfig and the executor DAG arg. The two enums share
// the same numeric values today, so the cast is identity at the wire level.
func sharedPoolRequestHAFromCanonical(canonical apiv1.SharedPool_HighAvailabilityType) apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_HighAvailabilityType {
	return apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_HighAvailabilityType(canonical)
}

func validateCreateSharedPoolPDDeploymentMode(sharedPool *entity.SharedPool) error {
	if entity.GetEffectivePDDeploymentMode(sharedPool.GetPdDeploymentMode()) != apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_MONOLITHIC {
		return nil
	}

	if sharedPool.GetCapacityMode() == apiv1.CapacityMode_TOPOLOGY_DRIVEN {
		return status.Errorf(codes.InvalidArgument, "pd_deployment_mode MONOLITHIC is only supported for RCU_DRIVEN mode")
	}

	for _, comp := range sharedPool.GetComponentGroups() {
		if comp == nil {
			continue
		}
		switch comp.GetComponentType() {
		case apiv1.ComponentType_PD_TSO, apiv1.ComponentType_PD_SCHEDULING, apiv1.ComponentType_RESOURCE_MANAGER:
			return status.Errorf(codes.InvalidArgument, "pd_deployment_mode MONOLITHIC does not support component %s", comp.GetComponentType().String())
		}
	}

	return nil
}

func validateCreateSharedPoolCapacityMode(sharedPool *entity.SharedPool) error {
	switch sharedPool.GetCapacityMode() {
	case apiv1.CapacityMode_CAPACITY_MODE_UNSPECIFIED,
		apiv1.CapacityMode_RCU_DRIVEN,
		apiv1.CapacityMode_TOPOLOGY_DRIVEN:
		return nil
	default:
		return status.Errorf(codes.InvalidArgument, "invalid shared pool capacity mode %s", sharedPool.GetCapacityMode().String())
	}
}

func validateCreateSharedPoolCapacityInput(sharedPool *entity.SharedPool, requestUnit *entity.RequestUnit, bootstrap *SharedPoolBootstrapCapacity) error {
	if bootstrap != nil {
		if sharedPool.GetCapacityMode() != apiv1.CapacityMode_RCU_DRIVEN {
			return resourcePoolStatusError(codes.InvalidArgument, reasonInvalidBootstrapCapacityMode,
				"bootstrap_capacity requires RCU_DRIVEN capacity_mode")
		}
		if requestUnit != nil {
			return resourcePoolStatusError(codes.InvalidArgument, reasonInvalidBootstrapCapacityMode,
				"bootstrap_capacity must not be combined with request_unit")
		}
		if sharedPool.GetTopologySpec() != nil {
			return resourcePoolStatusError(codes.InvalidArgument, reasonInvalidBootstrapCapacityMode,
				"bootstrap_capacity must not be combined with topology_spec")
		}
		return nil
	}

	if sharedPool.GetCapacityMode() == apiv1.CapacityMode_RCU_DRIVEN && requestUnit == nil {
		return status.Error(codes.InvalidArgument, "request_unit is required for RCU_DRIVEN capacity_mode when bootstrap_capacity is absent")
	}
	return nil
}

func validateCreateSharedPoolTiflashMode(sharedPool *entity.SharedPool) error {
	schemaManagerCount := 0
	hasTiflashWrite := false
	for _, comp := range sharedPool.GetTopologySpec().GetComponents() {
		switch comp.GetComponentType() {
		case apiv1.ComponentType_SCHEMA_MANAGER:
			schemaManagerCount++
		case apiv1.ComponentType_TIFLASH_WRITE:
			hasTiflashWrite = true
		}
	}

	switch sharedPool.RequestedTiflashMode {
	case apiv1.TiflashMode_TIFLASH_MODE_UNSPECIFIED, apiv1.TiflashMode_TIFLASH_MODE_AUTO:
		if sharedPool.GetCapacityMode() == apiv1.CapacityMode_TOPOLOGY_DRIVEN && schemaManagerCount > 0 {
			return status.Error(codes.InvalidArgument, "topology_spec must not contain SCHEMA_MANAGER unless tiflash_mode is COLUMNAR")
		}
		return nil
	case apiv1.TiflashMode_TIFLASH_MODE_LEGACY:
		if entity.HasSharedTiProxy(sharedPool) {
			return status.Error(codes.InvalidArgument, "tiflash_mode LEGACY is not supported for Essential v2 pools")
		}
		if sharedPool.GetCapacityMode() != apiv1.CapacityMode_RCU_DRIVEN {
			return status.Error(codes.InvalidArgument, "tiflash_mode LEGACY is only supported for RCU_DRIVEN mode")
		}
		return nil
	case apiv1.TiflashMode_TIFLASH_MODE_COLUMNAR:
		if sharedPool.GetCapacityMode() != apiv1.CapacityMode_RCU_DRIVEN &&
			sharedPool.GetCapacityMode() != apiv1.CapacityMode_TOPOLOGY_DRIVEN {
			return status.Error(codes.InvalidArgument, "tiflash_mode COLUMNAR is only supported for RCU_DRIVEN or TOPOLOGY_DRIVEN mode")
		}
		if sharedPool.GetCapacityMode() == apiv1.CapacityMode_TOPOLOGY_DRIVEN {
			if hasTiflashWrite {
				return status.Error(codes.InvalidArgument, "topology_spec must not contain TIFLASH_WRITE when tiflash_mode is COLUMNAR")
			}
			if schemaManagerCount != 1 {
				return status.Errorf(codes.InvalidArgument, "topology_spec must contain exactly one SCHEMA_MANAGER when tiflash_mode is COLUMNAR; got %d", schemaManagerCount)
			}
		}
		return nil
	default:
		return status.Errorf(codes.InvalidArgument, "invalid tiflash_mode %s", sharedPool.RequestedTiflashMode.String())
	}
}

func (s *serverImpl) resolveCreateSharedPoolTiflashMode(ctx context.Context, sharedPool *entity.SharedPool) error {
	if entity.HasSharedTiProxy(sharedPool) {
		sharedPool.TiflashMode = schema.SharedPoolTiflashModeColumnar
		return nil
	}

	if sharedPool.RequestedTiflashMode == apiv1.TiflashMode_TIFLASH_MODE_LEGACY {
		return nil
	}
	if sharedPool.RequestedTiflashMode == apiv1.TiflashMode_TIFLASH_MODE_COLUMNAR {
		sharedPool.TiflashMode = schema.SharedPoolTiflashModeColumnar
		return nil
	}

	tikvImageTag, err := s.clusterConfigClient.GetEffectiveComponentImageTag(ctx, sharedPool, apiv1.ComponentType_TIKV)
	if err != nil {
		return status.Errorf(codes.Internal, "failed to get effective TiKV image tag: %s", err)
	}
	if supportsColumnarByTiKVImage(tikvImageTag) {
		sharedPool.TiflashMode = schema.SharedPoolTiflashModeColumnar
		return nil
	}

	tilog.Warn(ctx, "TiKV image does not support Columnar; creating legacy topology",
		zap.String("sharedPoolId", sharedPool.GetId()),
		zap.String("tikvImageTag", tikvImageTag),
		zap.String("minSupportedVersion", minTiKVColumnarImageVersion))
	return nil
}

func normalizeAndValidateCreateSharedPoolEnabledTiKVRemoteWorkers(sharedPool *entity.SharedPool) error {
	capacityMode := entity.GetEffectiveCapacityMode(sharedPool.GetCapacityMode())
	if capacityMode != apiv1.CapacityMode_RCU_DRIVEN {
		if len(sharedPool.GetEnabledTikvRemoteWorkers()) > 0 {
			return status.Errorf(codes.InvalidArgument, "enabled_tikv_remote_workers is only supported for RCU_DRIVEN mode")
		}
		sharedPool.EnabledTikvRemoteWorkers = []apiv1.ComponentType{}
		return nil
	}

	enabledWorkers, err := entity.NormalizeEnabledTiKVRemoteWorkers(sharedPool.GetEnabledTikvRemoteWorkers())
	if err != nil {
		return status.Errorf(codes.InvalidArgument, "%s", err.Error())
	}
	sharedPool.EnabledTikvRemoteWorkers = enabledWorkers

	if entity.HasEnabledTiKVRemoteWorker(enabledWorkers, apiv1.ComponentType_COPROCESSOR_WORKER) {
		return nil
	}

	for _, comp := range sharedPool.GetComponentGroups() {
		if comp == nil {
			continue
		}
		if comp.GetComponentType() == apiv1.ComponentType_COPROCESSOR_WORKER {
			return status.Errorf(
				codes.InvalidArgument,
				"component %s is not enabled by enabled_tikv_remote_workers",
				comp.GetComponentType().String(),
			)
		}
	}

	return nil
}

func (s *serverImpl) checkCreateSharedpool(ctx context.Context, sharedPool *entity.SharedPool, requestUint *entity.RequestUnit) error {
	// check sharedpool already exists
	_, err := s.sharedPoolRepo.Load(ctx, sharedPool.GetId(), true)
	if err == nil {
		tilog.Warn(ctx, "shared pool id already exists", zap.String("sharedPoolId", sharedPool.GetId()))
		return status.Errorf(codes.AlreadyExists, "shared pool id %s already exists", sharedPool.GetId())
	} else if status.Code(err) != codes.NotFound {
		return status.Errorf(codes.Internal, "failed to load shared pool: %s", err)
	}

	dagIns, err := s.getRunningDagIns(ctx, []string{dag.PlanCreateSharedPool}, []string{sharedPool.GetId()}, nil, nil, nil)
	if err != nil {
		return status.Errorf(codes.Internal, "failed to get running dag instance: %s", err)
	}

	if len(dagIns) > 0 {
		return status.Errorf(codes.AlreadyExists, "shared pool creation is already running")
	}

	// check request unit (only for RCU_DRIVEN mode)
	if sharedPool.GetCapacityMode() != apiv1.CapacityMode_TOPOLOGY_DRIVEN && requestUint != nil {
		if requestUint.GetMax() < requestUint.GetMin() {
			return status.Errorf(codes.InvalidArgument, "request unit min %d cannot be greater than max %d", requestUint.GetMin(), requestUint.GetMax())
		}
	}

	return nil
}

func (s *serverImpl) genNodeSpecMap(ctx context.Context, sharedPool *entity.SharedPool) (map[string][]*entity.NodeSpec, error) {
	nodeSpecMap := make(map[string][]*entity.NodeSpec)

	if sharedPool.GetDataplaneName() == "" || sharedPool.GetPhase() == apiv1.SharedPool_CREATING {
		tilog.Warn(ctx, "dataplane name is empty or shared pool is creating, will use default node spec",
			zap.String("sharedPoolId", sharedPool.GetId()),
			zap.String("dataplaneName", sharedPool.GetDataplaneName()),
			zap.String("phase", sharedPool.GetPhase().String()),
		)
		// if dataplaneId is empty, we need to get the default node spec
		nodeSpecs, dropped, err := s.listDefaultNodeSpecs(ctx)
		if err != nil {
			tilog.Error(ctx, "failed to list node pool configs", zap.Error(err))
			return nil, status.Errorf(codes.Internal, "failed to list node pool configs: %s", err)
		}
		if len(nodeSpecs) == 0 {
			return nil, status.Errorf(codes.NotFound, "no node spec found for default dataplane")
		}
		logNodeSpecDrops(ctx, dropped)
		nodeSpecMap = nodeSpecMapFromList(nodeSpecs)
	} else {
		dataplaneNodeGroups, nodeGroups, err := s.loadNodeInstanceGroups(ctx, sharedPool.GetDataplaneName())
		if err != nil {
			return nil, err
		}

		nodePoolResp, err := s.dataplaneClient.ListNodePools(ctx, &dplanemgrv1.ListNodePoolsRequest{DataPlaneId: sharedPool.GetDataplaneName()})
		if err != nil {
			return nil, err
		}
		for _, nodePool := range nodePoolResp.GetNodePools() {
			if nodePool.GetNodeType() != dplanemgrv1.NodeType_Dedicated {
				// only support dedicated node type
				continue
			}

			nodeSpec := entity.GetComponentNodeSpec(&entity.NodeSpec{Cpu: nodePool.GetCpuCore(), MemoryGi: nodePool.GetMemoryGi()})
			if isInstanceTypeDeprecated(dataplaneNodeGroups, nodeGroups, nodeSpec, nodePool.GetInstanceType()) {
				tilog.Info(ctx, "skipping deprecated instance type",
					zap.String("dataplaneName", sharedPool.GetDataplaneName()),
					zap.String("nodeSpec", nodeSpec),
					zap.String("instanceType", nodePool.GetInstanceType()),
				)
				continue
			}

			key := nodeSpec
			arr, ok := nodeSpecMap[key]
			if !ok {
				arr = make([]*entity.NodeSpec, 0)
			}
			arr = append(arr, &entity.NodeSpec{
				Cpu:               nodePool.GetCpuCore(),
				MemoryGi:          nodePool.GetMemoryGi(),
				AvailableCpu:      nodePool.AvailableCpuMilliCore / 1000, // convert milliCore to core
				AvailableCpuMilli: nodePool.AvailableCpuMilliCore,
				AvailableMemoryMi: nodePool.AvailableMemoryMi,
				MaxCpuMilli:       nodePool.MaxCpuMilli,
				MaxMemoryMi:       nodePool.MaxMemoryMi,
				WithLocalDisk:     nodePool.WithLocalDisk,
				LocalDiskSizeGi:   nodePool.LocalDiskSizeGi,
			})
			nodeSpecMap[key] = arr
		}
	}

	if len(nodeSpecMap) == 0 {
		return nil, status.Errorf(codes.NotFound, "no node spec found from dataplane %s", sharedPool.GetDataplaneName())
	}

	return nodeSpecMap, nil
}

type nodeInstanceTypeKey struct {
	nodeSpec     string
	instanceType string
}

func (s *serverImpl) loadNodeInstanceGroups(ctx context.Context, dataplaneName string) (map[nodeInstanceTypeKey]*entity.DataplaneNodeInstanceGroup, map[nodeInstanceTypeKey]*entity.NodeInstanceGroup, error) {
	dataplaneNodeGroups := make(map[nodeInstanceTypeKey]*entity.DataplaneNodeInstanceGroup)
	if s.dataplaneNodeGroupRepo != nil {
		overrides, err := s.dataplaneNodeGroupRepo.ListByDataplane(ctx, dataplaneName)
		if err != nil {
			return nil, nil, fmt.Errorf("failed to list dataplane node instance groups for %s: %w", dataplaneName, err)
		}
		for _, override := range overrides {
			dataplaneNodeGroups[nodeInstanceTypeKey{
				nodeSpec:     entity.GetComponentNodeSpec(override.GetNodeSpec()),
				instanceType: override.GetPrimaryInstanceType(),
			}] = override
		}
	}

	nodeGroups := make(map[nodeInstanceTypeKey]*entity.NodeInstanceGroup)
	if s.nodeGroupRepo != nil {
		groups, err := s.nodeGroupRepo.ListAll(ctx)
		if err != nil {
			return nil, nil, fmt.Errorf("failed to list node instance groups: %w", err)
		}
		for _, group := range groups {
			nodeGroups[nodeInstanceTypeKey{
				nodeSpec:     entity.GetComponentNodeSpec(group.GetNodeSpec()),
				instanceType: group.GetPrimaryInstanceType(),
			}] = group
		}
	}

	return dataplaneNodeGroups, nodeGroups, nil
}

func isInstanceTypeDeprecated(
	dataplaneNodeGroups map[nodeInstanceTypeKey]*entity.DataplaneNodeInstanceGroup,
	nodeGroups map[nodeInstanceTypeKey]*entity.NodeInstanceGroup,
	nodeSpec string,
	instanceType string,
) bool {
	key := nodeInstanceTypeKey{
		nodeSpec:     nodeSpec,
		instanceType: instanceType,
	}
	if override, ok := dataplaneNodeGroups[key]; ok {
		return override.GetStatus() != apiv1.DataplaneNodeInstanceGroup_ACTIVE
	}
	if group, ok := nodeGroups[key]; ok {
		return group.GetStatus() != apiv1.NodeInstanceGroup_ACTIVE
	}

	return false
}

func genDataplaneName() string {
	return fmt.Sprintf("dp-%s", uuid.New().String())
}

func (s *serverImpl) generateTopologyDrivenComponents(
	ctx context.Context,
	sharedPool *entity.SharedPool,
	requestUint *entity.RequestUnit,
	nodeSpecMap map[string][]*entity.NodeSpec,
	initConfig *entity.InitSharedPoolConfig,
) error {
	if sharedPool.GetTopologySpec() == nil || len(sharedPool.GetTopologySpec().GetComponents()) == 0 {
		return status.Errorf(codes.InvalidArgument, "topology_spec with at least one component is required for TOPOLOGY_DRIVEN mode")
	}

	if err := validateTopologyDrivenRequiredComponents(sharedPool.GetTopologySpec()); err != nil {
		return err
	}

	defaultComponentID := utils.FnvHash(sharedPool.GetId())
	columnar := sharedPool.TiflashMode == schema.SharedPoolTiflashModeColumnar
	topologyComponents, err := ensureTopologyDrivenTiFlashComponents(
		sharedPool.GetTopologySpec().GetComponents(),
		defaultComponentID,
		nodeSpecMap,
		initConfig,
		columnar,
	)
	if err != nil {
		return err
	}

	components := make([]*entity.Component, 0, len(topologyComponents))

	for _, comp := range topologyComponents {
		if comp == nil {
			continue
		}

		if comp.GetId() == "" {
			comp.Id = defaultComponentID
		}
		if comp.GetPhase() == apiv1.Component_PHASE_UNSPECIFIED {
			comp.Phase = apiv1.Component_CREATING
		}

		if comp.GetComponentType() == apiv1.ComponentType_TIFLASH_LEARNER ||
			comp.GetComponentType() == apiv1.ComponentType_TIFLASH_COMPUTE_LEARNER {
			if comp.GetPhase() == apiv1.Component_CREATING {
				comp.Phase = apiv1.Component_UNAVAILABLE
			}
			components = append(components, comp)
			continue
		}

		if comp.GetNodeQuota() == nil {
			return status.Errorf(codes.InvalidArgument, "component %s-%s node quota is required", entity.GetComponentTypeName(comp), comp.GetId())
		}
		if comp.GetNodeQuota().GetNodeSpec() == nil {
			return status.Errorf(codes.InvalidArgument, "component %s-%s node spec is required", entity.GetComponentTypeName(comp), comp.GetId())
		}

		nodeSpec := comp.GetNodeQuota().GetNodeSpec()
		withLocalDisk := nodeSpec.GetWithLocalDisk()
		for _, st := range comp.GetStorages() {
			if st.GetName() == apiv1.Component_Storage_DATA && st.GetType() == entity.LocalSSD {
				withLocalDisk = true
				break
			}
		}

		resolvedSpec, err := getNodeSpec(ctx, nodeSpec, nodeSpecMap, withLocalDisk)
		if err != nil {
			return status.Errorf(codes.InvalidArgument, "component %s-%s invalid node spec: %s", entity.GetComponentTypeName(comp), comp.GetId(), err)
		}
		comp.GetNodeQuota().NodeSpec = resolvedSpec

		if comp.GetNodeQuota().GetMaxNodeCount() == 0 {
			comp.GetNodeQuota().MaxNodeCount = comp.GetNodeQuota().GetNodeCount()
		}
		if comp.GetNodeQuota().GetMinNodeCount() == 0 {
			comp.GetNodeQuota().MinNodeCount = comp.GetNodeQuota().GetNodeCount()
		}

		if len(comp.GetStorages()) == 0 {
			switch comp.GetComponentType() {
			case apiv1.ComponentType_TIKV:
				storages, err := s.genDefaultTiKVStorages(sharedPool.GetProvider().GetType(), comp.GetNodeQuota().GetNodeSpec(), initConfig)
				if err != nil {
					return err
				}
				comp.Storages = storages
			case apiv1.ComponentType_PD:
				storages, err := s.genDefaultPDStorages(sharedPool.GetProvider().GetType(), requestUint)
				if err != nil {
					return err
				}
				comp.Storages = storages
			case apiv1.ComponentType_COPROCESSOR_WORKER:
				storages, err := s.genDefaultCoprocessorWorkerStorages(sharedPool.GetProvider().GetType())
				if err != nil {
					return err
				}
				comp.Storages = storages
			case apiv1.ComponentType_TIFLASH_COMPUTE:
				storages, err := s.genDefaultTiFlashCNStorages(sharedPool.GetProvider().GetType(), comp.GetNodeQuota().GetNodeSpec(), initConfig)
				if err != nil {
					return err
				}
				comp.Storages = storages
			case apiv1.ComponentType_TIFLASH_WRITE:
				storages, err := s.genDefaultTiFlashWNStorages(sharedPool.GetProvider().GetType(), initConfig)
				if err != nil {
					return err
				}
				comp.Storages = storages
			case apiv1.ComponentType_SCHEMA_MANAGER:
				storages, err := s.genDefaultSchemaManagerStorages(sharedPool.GetProvider().GetType())
				if err != nil {
					return err
				}
				comp.Storages = storages
			}
		}

		for _, storage := range comp.GetStorages() {
			if storage.GetName() == apiv1.Component_Storage_RAFTLOG && storage.GetMountPath() == "" {
				storage.MountPath = entity.DefaultRaftlogMountPath
			}
		}

		if err := s.checkComponent(comp); err != nil {
			return err
		}

		components = append(components, comp)
	}

	sharedPool.ComponentGroups = components
	return nil
}

func ensureTopologyDrivenTiFlashComponents(
	components []*entity.Component,
	defaultComponentID string,
	nodeSpecMap map[string][]*entity.NodeSpec,
	initConfig *entity.InitSharedPoolConfig,
	columnar bool,
) ([]*entity.Component, error) {
	result := append([]*entity.Component{}, components...)

	findByType := func(componentType apiv1.ComponentType) *entity.Component {
		for _, comp := range result {
			if comp != nil && comp.GetComponentType() == componentType {
				return comp
			}
		}
		return nil
	}

	writeComp := findByType(apiv1.ComponentType_TIFLASH_WRITE)
	if columnar && writeComp != nil {
		return nil, status.Error(codes.InvalidArgument, "topology_spec must not contain TIFLASH_WRITE when tiflash_mode is COLUMNAR")
	}
	if !columnar && writeComp == nil {
		nodeSpec, err := pickDefaultNodeSpecForTiFlash(result, nodeSpecMap, false)
		if err != nil {
			return nil, err
		}
		writeComp = &entity.Component{
			Id:            defaultComponentID,
			ComponentType: apiv1.ComponentType_TIFLASH_WRITE,
			Phase:         apiv1.Component_CREATING,
			NodeQuota: &entity.NodeQuota{
				NodeCount:    0,
				MaxNodeCount: 0,
				MinNodeCount: 0,
				NodeSpec:     nodeSpec,
			},
		}
		result = append(result, writeComp)
	}

	computeComp := findByType(apiv1.ComponentType_TIFLASH_COMPUTE)
	if computeComp == nil {
		preferLocalDisk := initConfig != nil &&
			initConfig.GetTiflashCnStorage() != nil &&
			initConfig.GetTiflashCnStorage().GetType() == entity.LocalSSD

		nodeSpec, err := pickDefaultNodeSpecForTiFlash(result, nodeSpecMap, preferLocalDisk)
		if err != nil {
			return nil, err
		}
		computeComp = &entity.Component{
			Id:            defaultComponentID,
			ComponentType: apiv1.ComponentType_TIFLASH_COMPUTE,
			Phase:         apiv1.Component_CREATING,
			NodeQuota: &entity.NodeQuota{
				NodeCount:    0,
				MaxNodeCount: 0,
				MinNodeCount: 0,
				NodeSpec:     nodeSpec,
			},
		}
		result = append(result, computeComp)
	}

	hasLearner := make(map[string]bool)
	hasComputeLearner := make(map[string]bool)

	for _, comp := range result {
		if comp == nil {
			continue
		}
		switch comp.GetComponentType() {
		case apiv1.ComponentType_TIFLASH_LEARNER:
			hasLearner[comp.GetId()] = true
		case apiv1.ComponentType_TIFLASH_COMPUTE_LEARNER:
			hasComputeLearner[comp.GetId()] = true
		}
	}

	for _, comp := range result {
		if comp == nil {
			continue
		}

		switch comp.GetComponentType() {
		case apiv1.ComponentType_TIFLASH_WRITE:
			if !hasLearner[comp.GetId()] {
				result = append(result, &entity.Component{
					Id:            comp.GetId(),
					ComponentType: apiv1.ComponentType_TIFLASH_LEARNER,
					Phase:         apiv1.Component_UNAVAILABLE,
					NodeQuota: &apiv1.NodeQuota{
						NodeCount: 0,
					},
				})
				hasLearner[comp.GetId()] = true
			}
		case apiv1.ComponentType_TIFLASH_COMPUTE:
			if !hasComputeLearner[comp.GetId()] {
				result = append(result, &entity.Component{
					Id:            comp.GetId(),
					ComponentType: apiv1.ComponentType_TIFLASH_COMPUTE_LEARNER,
					Phase:         apiv1.Component_UNAVAILABLE,
					NodeQuota: &apiv1.NodeQuota{
						NodeCount: 0,
					},
				})
				hasComputeLearner[comp.GetId()] = true
			}
		}
	}

	return result, nil
}

func pickDefaultNodeSpecForTiFlash(
	components []*entity.Component,
	nodeSpecMap map[string][]*entity.NodeSpec,
	preferLocalDisk bool,
) (*entity.NodeSpec, error) {
	for _, comp := range components {
		if comp == nil || comp.GetNodeQuota() == nil || comp.GetNodeQuota().GetNodeSpec() == nil {
			continue
		}
		spec := comp.GetNodeQuota().GetNodeSpec()
		if spec.GetWithLocalDisk() == preferLocalDisk {
			return spec, nil
		}
	}

	keys := make([]string, 0, len(nodeSpecMap))
	for key := range nodeSpecMap {
		keys = append(keys, key)
	}
	sort.Strings(keys)

	for _, key := range keys {
		for _, spec := range nodeSpecMap[key] {
			if spec.GetWithLocalDisk() == preferLocalDisk {
				return spec, nil
			}
		}
	}

	for _, key := range keys {
		if len(nodeSpecMap[key]) > 0 {
			return nodeSpecMap[key][0], nil
		}
	}

	return nil, status.Errorf(codes.InvalidArgument, "no node spec found for auto-generated tiflash components")
}

func validateTopologyDrivenRequiredComponents(topologySpec *apiv1.TopologySpec) error {
	if topologySpec == nil || len(topologySpec.GetComponents()) == 0 {
		return status.Errorf(codes.InvalidArgument, "topology_spec with at least one component is required for TOPOLOGY_DRIVEN mode")
	}

	allowedTypes := map[apiv1.ComponentType]bool{
		apiv1.ComponentType_PD:              true,
		apiv1.ComponentType_TIKV:            true,
		apiv1.ComponentType_TIKV_WORKER:     true,
		apiv1.ComponentType_TIDB_WORKER:     true,
		apiv1.ComponentType_TIFLASH_WRITE:   true,
		apiv1.ComponentType_TIFLASH_COMPUTE: true,
		apiv1.ComponentType_SCHEMA_MANAGER:  true,
	}

	requiredTypes := []apiv1.ComponentType{
		apiv1.ComponentType_PD,
		apiv1.ComponentType_TIKV,
		apiv1.ComponentType_TIKV_WORKER,
		apiv1.ComponentType_TIDB_WORKER,
	}

	present := make(map[apiv1.ComponentType]bool, len(topologySpec.GetComponents()))
	unsupported := make([]string, 0)
	for _, comp := range topologySpec.GetComponents() {
		if comp == nil {
			continue
		}
		if !allowedTypes[comp.GetComponentType()] {
			unsupported = append(unsupported, comp.GetComponentType().String())
			continue
		}
		present[comp.GetComponentType()] = true
	}

	missing := make([]string, 0)
	for _, componentType := range requiredTypes {
		if !present[componentType] {
			missing = append(missing, componentType.String())
		}
	}

	if len(missing) > 0 {
		return status.Errorf(codes.InvalidArgument,
			"topology_spec missing required components for TOPOLOGY_DRIVEN mode: %s",
			strings.Join(missing, ","),
		)
	}

	if len(unsupported) > 0 {
		return status.Errorf(codes.InvalidArgument,
			"topology_spec contains unsupported components for TOPOLOGY_DRIVEN mode: %s; allowed: %s",
			strings.Join(unsupported, ","),
			strings.Join([]string{
				apiv1.ComponentType_PD.String(),
				apiv1.ComponentType_TIKV.String(),
				apiv1.ComponentType_TIKV_WORKER.String(),
				apiv1.ComponentType_TIDB_WORKER.String(),
				apiv1.ComponentType_TIFLASH_WRITE.String(),
				apiv1.ComponentType_TIFLASH_COMPUTE.String(),
				apiv1.ComponentType_SCHEMA_MANAGER.String(),
			}, ","),
		)
	}

	return nil
}

func checkInitSharedPoolConfig(provider string, capacityMode apiv1.CapacityMode, topologySpec *apiv1.TopologySpec, initConfig *entity.InitSharedPoolConfig) error {
	if initConfig == nil {
		return status.Error(codes.InvalidArgument, "initSharedPoolConfig is required")
	}

	if initConfig.GetNetworkCidr() == "" {
		return status.Error(codes.InvalidArgument, "network cidr is required")
	}

	// check cidr
	_, networkIPNet, err := net.ParseCIDR(initConfig.GetNetworkCidr())
	if err != nil {
		return status.Errorf(codes.InvalidArgument, "network cidr %s is invalid: %s", initConfig.GetNetworkCidr(), err)
	}

	if initConfig.GetServiceCidr() != "" {
		_, serviceIPNet, err := net.ParseCIDR(initConfig.GetServiceCidr())
		if err != nil {
			return status.Errorf(codes.InvalidArgument, "service cidr %s is invalid: %s", initConfig.GetServiceCidr(), err)
		}

		// check if service cidr and network cidr overlap
		if networkIPNet.Contains(serviceIPNet.IP) || serviceIPNet.Contains(networkIPNet.IP) {
			return status.Errorf(codes.InvalidArgument, "service cidr %s and network cidr %s cannot overlap", initConfig.GetServiceCidr(), initConfig.GetNetworkCidr())
		}
	}

	// Validate storage configurations using unified validation.
	if capacityMode == apiv1.CapacityMode_TOPOLOGY_DRIVEN {
		if err := validateTopologyStorageConfigs(provider, topologySpec); err != nil {
			return err
		}
		return nil
	}

	if err := validateInitStorageConfigs(provider, initConfig); err != nil {
		return err
	}

	return nil
}

func validateTopologyStorageConfigs(provider string, topologySpec *apiv1.TopologySpec) error {
	if topologySpec == nil || len(topologySpec.GetComponents()) == 0 {
		return status.Errorf(codes.InvalidArgument, "topology_spec with at least one component is required for TOPOLOGY_DRIVEN mode")
	}

	for _, comp := range topologySpec.GetComponents() {
		for _, st := range comp.GetStorages() {
			// The Schema Manager TiKVWorkerGroup template only supports PVC-backed storage.
			if comp.GetComponentType() == apiv1.ComponentType_SCHEMA_MANAGER && st.GetType() == entity.LocalSSD {
				return status.Error(codes.InvalidArgument, "SCHEMA_MANAGER does not support local-ssd storage")
			}
			if err := entity.ValidateStorageTypeForProvider(st, provider, fmt.Sprintf("%s.%s", comp.GetComponentType().String(), st.GetName().String())); err != nil {
				return status.Errorf(codes.InvalidArgument, "%s", err.Error())
			}
		}
	}

	return nil
}

// validateInitStorageConfigs validates all storage configurations in InitSharedPoolConfig
func validateInitStorageConfigs(provider string, initConfig *entity.InitSharedPoolConfig) error {
	upperProvider := strings.ToUpper(provider)

	// Define storage configs with their validation options
	type storageConfig struct {
		name         string
		storage      *apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage
		opts         *entity.StorageValidationOptions
		allowedTypes map[string]bool // nil means use default validation
	}

	var configs []storageConfig

	switch upperProvider {
	case apiv1.ProviderType_AWS.String():
		configs = []storageConfig{
			{
				name:         "tikv data",
				storage:      initConfig.GetTikvDataStorage(),
				opts:         &entity.StorageValidationOptions{MinSize: entity.MinTiKVDataStorageSize},
				allowedTypes: map[string]bool{entity.GP3: true, entity.IO2: true, entity.LocalSSD: true},
			},
			{
				name:         "tikv raftlog",
				storage:      initConfig.GetTikvRaftlogStorage(),
				opts:         &entity.StorageValidationOptions{MinSize: entity.MinTiKVRaftlogStorageSize},
				allowedTypes: map[string]bool{entity.GP3: true, entity.IO2: true}, // LocalSSD not allowed for raftlog
			},
			{
				name:         "tiflash cn",
				storage:      initConfig.GetTiflashCnStorage(),
				opts:         &entity.StorageValidationOptions{MinSize: entity.MinTiFlashCNStorageSize},
				allowedTypes: map[string]bool{entity.GP3: true, entity.IO2: true, entity.LocalSSD: true},
			},
			{
				name:         "tiflash wn",
				storage:      initConfig.GetTiflashWnStorage(),
				opts:         &entity.StorageValidationOptions{MinSize: entity.MinTiFlashWNStorageSize},
				allowedTypes: map[string]bool{entity.GP3: true, entity.IO2: true}, // LocalSSD not allowed for tiflash wn
			},
		}
	case apiv1.ProviderType_ALICLOUD.String():
		configs = []storageConfig{
			{
				name:         "tikv data",
				storage:      initConfig.GetTikvDataStorage(),
				opts:         &entity.StorageValidationOptions{MinSize: entity.MinTiKVDataStorageSize},
				allowedTypes: map[string]bool{entity.CloudAuto: true, entity.LocalSSD: true},
			},
			{
				name:         "tikv raftlog",
				storage:      initConfig.GetTikvRaftlogStorage(),
				opts:         &entity.StorageValidationOptions{MinSize: entity.MinTiKVRaftlogStorageSize},
				allowedTypes: map[string]bool{entity.CloudAuto: true}, // LocalSSD not allowed for raftlog
			},
			{
				name:         "tiflash cn",
				storage:      initConfig.GetTiflashCnStorage(),
				opts:         &entity.StorageValidationOptions{MinSize: entity.MinTiFlashCNStorageSize},
				allowedTypes: map[string]bool{entity.CloudAuto: true, entity.LocalSSD: true},
			},
			{
				name:         "tiflash wn",
				storage:      initConfig.GetTiflashWnStorage(),
				opts:         &entity.StorageValidationOptions{MinSize: entity.MinTiFlashWNStorageSize},
				allowedTypes: map[string]bool{entity.CloudAuto: true}, // LocalSSD not allowed for tiflash wn
			},
		}
	case apiv1.ProviderType_GCP.String():
		return status.Errorf(codes.Unimplemented, "GCP cloud provider is not supported yet")
	case apiv1.ProviderType_AZURE.String():
		configs = []storageConfig{
			{
				name:         "tikv data",
				storage:      initConfig.GetTikvDataStorage(),
				opts:         &entity.StorageValidationOptions{MinSize: entity.MinTiKVDataStorageSize},
				allowedTypes: map[string]bool{entity.PremiumSSDv2: true},
			},
			{
				name:         "tikv raftlog",
				storage:      initConfig.GetTikvRaftlogStorage(),
				opts:         &entity.StorageValidationOptions{MinSize: entity.MinTiKVRaftlogStorageSize},
				allowedTypes: map[string]bool{entity.PremiumSSDv2: true},
			},
			{
				name:         "tiflash cn",
				storage:      initConfig.GetTiflashCnStorage(),
				opts:         &entity.StorageValidationOptions{MinSize: entity.MinTiFlashCNStorageSize},
				allowedTypes: map[string]bool{entity.PremiumSSDv2: true},
			},
			{
				name:         "tiflash wn",
				storage:      initConfig.GetTiflashWnStorage(),
				opts:         &entity.StorageValidationOptions{MinSize: entity.MinTiFlashWNStorageSize},
				allowedTypes: map[string]bool{entity.PremiumSSDv2: true},
			},
		}
	default:
		return status.Errorf(codes.Unimplemented, "cloud provider %s is not supported yet", provider)
	}

	// Validate each storage config
	for _, cfg := range configs {
		if cfg.storage == nil {
			return status.Errorf(codes.InvalidArgument, "%s storage config is required", cfg.name)
		}

		// Check if storage type is allowed for this config
		if cfg.allowedTypes != nil && !cfg.allowedTypes[cfg.storage.GetType()] {
			return status.Errorf(codes.InvalidArgument, "%s storage type %s not supported for %s",
				cfg.name, cfg.storage.GetType(), strings.ToLower(upperProvider))
		}

		// For LocalSSD, skip size validation but ensure size is 0
		if cfg.storage.GetType() == entity.LocalSSD {
			cfg.opts = &entity.StorageValidationOptions{AllowZeroSize: true}
		}

		// Validate using unified function
		params := entity.StorageParamsFromInitConfig(cfg.storage)
		if err := entity.ValidateStorageParams(params, provider, cfg.name, cfg.opts); err != nil {
			return status.Errorf(codes.InvalidArgument, "%s", err.Error())
		}
	}

	return nil
}

// (mergeFeatureFlagsIntoResolved was removed when CreateSharedPool
// was migrated to `resolve.PriorityResolveForSharedPoolFromBundle`,
// which reads DEFAULT + TENANT bundle rows itself in a single pass.
// The double-resolution bug-class — where the previous code ran
// `PriorityResolveForSharedPool` (TENANT-only) and then overlaid
// chain-resolved flags via this helper — is no longer reachable.)