package server

import (
	"context"
	"fmt"
	"math"
	"regexp"
	"strings"

	"github.com/pingcap/kvproto/pkg/keyspacepb"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"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/clustergateway"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/cluster-service-ng/pkg/tioperator"
	"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) CreateTiDBCluster(ctx context.Context, tidbCluster *entity.TiDBCluster) (*entity.TiDBCluster, error) {

	if tidbCluster.GetSharedPoolId() == "" {
		return nil, status.Errorf(codes.InvalidArgument, "shared pool id is required")
	}

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

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

	// Track the DAG exec ID for potential abort on transaction failure
	var dagExecId string
	var dagSubmitted bool

	if err := s.transactioner.Transaction(ctx, func(ctx context.Context) error {
		// check if the tidbCluster is locked
		unlocktidbClusterFunc, err := s.lock(ctx, tidbCluster.GetId())
		if err != nil {
			tilog.Error(ctx, "failed to lock tidbCluster", zap.String("tidbClusterId", tidbCluster.GetId()), zap.Error(err))
			return err
		}
		defer func() {
			if err := unlocktidbClusterFunc(); err != nil {
				tilog.Error(ctx, "failed to unlock tidbCluster", zap.String("tidbClusterId", tidbCluster.GetId()), zap.Error(err))
			}
		}()

		// set tidb cluster phase to creating
		tidbCluster.Phase = apiv1.Cluster_CREATING

		// check shared pool
		// set specified shared pool id
		sharedPool, err := s.sharedPoolRepo.Load(ctx, tidbCluster.GetSharedPoolId(), false)
		if err != nil {
			if status.Code(err) != codes.NotFound {
				return status.Errorf(codes.Internal, "failed to load sharedPool: %s", err)
			}
			// not found, check if it's in creating
			return status.Errorf(codes.NotFound, "sharedPool %s not found", tidbCluster.GetSharedPoolId())
		}
		if err := guardSharedPoolMutation(sharedPool, sharedPoolMutationCreateTiDBCluster); err != nil {
			return err
		}

		// merge sharedPool status from dag task
		err = s.mergeSharedPoolStatusFromDagTask(ctx, sharedPool)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to merge sharedPool status from dag task: %s", err)
		}

		// check sharedPool phase
		if sharedPool.GetPhase() == apiv1.SharedPool_UNAVAILABLE {
			tilog.Warn(ctx, "target sharedPool is unavailable", zap.String("sharedPoolId", sharedPool.GetId()))
			return status.Errorf(codes.FailedPrecondition, "target sharedPool is unavailable, please check the shared pool status")
		}
		if sharedPool.GetPhase() == apiv1.SharedPool_DELETING {
			return resourcePoolStatusError(codes.FailedPrecondition, reasonPauseResumeInvalidState, "shared pool is deleting")
		}

		// check if the sharedPool is locked
		unlockSharedpoolFunc, 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 := unlockSharedpoolFunc(); err != nil {
				tilog.Error(ctx, "failed to unlock sharedPool", zap.String("sharedPoolId", sharedPool.GetId()), zap.Error(err))
			}
		}()

		// set tidb cluster shared pool id and version
		tidbCluster.SharedPoolId = sharedPool.GetId()
		tidbCluster.Version = sharedPool.GetVersion()

		if tidbCluster.GetVersion() == "" {
			tilog.Error(ctx, "shared pool version is empty", zap.String("sharedPoolId", sharedPool.GetId()))
			return status.Errorf(codes.FailedPrecondition, "shared pool version is empty, please check the shared pool status")
		}

		// set default gateway identity
		if entity.HasSharedTiProxy(sharedPool) {
			tidbCluster.GatewayIdentity = utils.FnvHash(sharedPool.GetId())
		} else if tidbCluster.GetGatewayIdentity() == "" {
			tidbCluster.GatewayIdentity = newDefaultDNSIdentity()
		}

		// check tidb group keyspace
		switch sharedPool.GetPhase() {
		case apiv1.SharedPool_CREATING:
			tilog.Warn(ctx, "sharedPool is creating, will check the tidb group keyspace later", zap.String("sharedPoolId", sharedPool.GetId()))
		case apiv1.SharedPool_UNAVAILABLE:
			tilog.Warn(ctx, "sharedPool is unavailable", zap.String("sharedPoolId", sharedPool.GetId()))
			return status.Errorf(codes.FailedPrecondition, "sharedPool is unavailable, please check the shared pool status")
		default:
			if sharedPool.PauseResumeState == entity.PauseResumeStateResuming {
				tilog.Info(ctx, "sharedPool is resuming, will check the tidb group keyspace later",
					zap.String("sharedPoolId", sharedPool.GetId()))
				break
			}
			err = s.checkTiDBClusterKeyspace(ctx, sharedPool, tidbCluster)
			if err != nil {
				tilog.Error(ctx, "failed to check tidb group keyspace", zap.Error(err))
				return err
			}
		}

		cmekId, err := getCmekId(tidbCluster.GetSecurityConfiguration(), sharedPool.GetProvider().GetType())
		if err != nil {
			return status.Errorf(codes.Internal, "failed to get cmek id: %s", err)
		}

		optionsMap := map[string][]string{}
		componentTypes := []string{}
		componentIds := []string{}

		if tidbCluster.GetCapacityMode() == apiv1.CapacityMode_TOPOLOGY_DRIVEN {
			// TOPOLOGY_DRIVEN: only use caller-provided component types,
			// but complete each component fields in CSNG.
			if tidbCluster.GetTopologySpec() == nil || len(tidbCluster.GetTopologySpec().GetComponents()) == 0 {
				return status.Errorf(codes.InvalidArgument, "topology_spec with at least one component is required for TOPOLOGY_DRIVEN mode")
			}

			if err := validateTopologyDrivenTiDBClusterComponents(tidbCluster.GetTopologySpec()); err != nil {
				return err
			}

			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)
			}

			filteredTopologySpec := &apiv1.TopologySpec{Components: make([]*apiv1.Component, 0, len(tidbCluster.GetTopologySpec().GetComponents()))}
			for _, comp := range tidbCluster.GetTopologySpec().GetComponents() {
				if comp != nil && comp.GetComponentType() == apiv1.ComponentType_TIDB {
					continue
				}
				filteredTopologySpec.Components = append(filteredTopologySpec.Components, comp)
			}
			tidbCluster.TopologySpec = filteredTopologySpec

			components, err := s.generateTopologyDrivenTiDBClusterComponents(ctx, sharedPool, tidbCluster, nodeSpecMap)
			if err != nil {
				return err
			}

			for _, comp := range components {
				sharedPool.ComponentGroups = append(sharedPool.ComponentGroups, comp)
				componentIds = append(componentIds, comp.GetId())
				componentTypes = append(componentTypes, entity.GetComponentTypeName(comp))
			}
		} else {
			ruConfig, err := s.ruConfigRepo.GetRuConfig(ctx, sharedPool.GetId(), apiv1.RuConfig_SHARED_POOL)
			if err != nil {
				tilog.Error(ctx, "failed to get ru config", zap.String("sharedPoolId", sharedPool.GetId()), zap.Error(err))
				return status.Errorf(codes.Internal, "failed to get ru config: %s", err)
			}

			essentialV2 := entity.IsEssentialV2(sharedPool)
			configuredScaleSharedPoolByRu := ruConfig.GetFeatureFlags().GetEnableScaleSharedPoolByRu()
			enableScaleSharedPoolByRu := configuredScaleSharedPoolByRu && !essentialV2
			tilog.Info(ctx, "resolved create tidb cluster RU spec scaling policy",
				zap.String("tidbClusterId", tidbCluster.GetId()),
				zap.String("sharedPoolId", sharedPool.GetId()),
				zap.Bool("essentialV2", essentialV2),
				zap.Bool("configuredEnableScaleSharedPoolByRu", configuredScaleSharedPoolByRu),
				zap.Bool("effectiveEnableScaleSharedPoolByRu", enableScaleSharedPoolByRu),
			)

			var nodeSpecMap map[string][]*entity.NodeSpec
			if !essentialV2 {
				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 enableScaleSharedPoolByRu {
				tilog.Info(ctx, "create tidb cluster with ru, re-generate shared pool node spec",
					zap.String("tidbClusterId", tidbCluster.GetId()),
					zap.String("sharedPoolId", sharedPool.GetId()),
					zap.String("requestUnit", requestUnitProtoToString(tidbCluster.GetRequestUnit())),
					zap.Bool("configuredEnableScaleSharedPoolByRu", configuredScaleSharedPoolByRu),
					zap.Bool("effectiveEnableScaleSharedPoolByRu", enableScaleSharedPoolByRu),
				)
				ruTotal, err := s.getRequestUnitTotal(ctx, tidbCluster)
				if err != nil {
					return status.Errorf(codes.Internal, "failed to get request unit total: %s", err)
				}

				// Snapshot old specs for whitelist components before regeneration,
				// so ScaleOverride can compare old vs new.
				oldSpecs := snapshotComponentSpecs(sharedPool)

				// The hardcoded force=true below only relaxes the AVAILABLE-phase
				// check inside generatedSharedPoolComponent; it must not bypass the
				// tikv-remote-worker topology guard, so the guard runs unconditionally.
				if err := s.guardTiKVRemoteWorkersTopologyStable(ctx, sharedPool); err != nil {
					return err
				}

				// force=true allows modifying shared pool components even when they are not in AVAILABLE state.
				// This prevents CreateTiDBCluster from being blocked by other component changes (e.g., TiKV/TiDB Worker
				// still in CREATING/MODIFYING phase) when RU-based scaling needs to update component specs immediately.
				err = s.generatedSharedPoolComponent(ctx, sharedPool, ruTotal, nodeSpecMap, ruConfig, nil, true)
				if err != nil {
					return err
				}

				rcuSet := rcuScaleComponentSet()

				// Only submit actions for whitelist SharedPool components.
				for _, comp := range sharedPool.GetComponentGroups() {
					cfg, inWhitelist := rcuSet[comp.GetComponentType()]
					if !inWhitelist || cfg.Scope != RcuScaleScopeSharedPool {
						continue
					}
					if comp.GetPhase() != apiv1.Component_CREATING && comp.GetPhase() != apiv1.Component_MODIFYING {
						continue
					}
					// If a ScaleOverride is configured, check whether the
					// old→new transition is allowed.
					if cfg.ScaleOverride != nil {
						key := componentSpecKey{compType: comp.GetComponentType(), id: comp.GetId()}
						oldSpec, found := oldSpecs[key]
						if found && !cfg.ScaleOverride(oldSpec, comp.GetNodeQuota().GetNodeSpec()) {
							tilog.Info(ctx, "rcu_scale_up: override vetoed scale-up",
								zap.String("sharedPoolId", sharedPool.GetId()),
								zap.String("tidbClusterId", tidbCluster.GetId()),
								zap.String("componentType", comp.GetComponentType().String()),
								zap.String("componentId", comp.GetId()))
							AutoscaleRcuScaleUpTotal.WithLabelValues(sharedPool.GetId(), "create", comp.GetComponentType().String(), "vetoed").Inc()
							continue
						}
					}

					optionsMap[entity.GetUpdateComponentKey(comp)] = []string{
						string(tioperator.OpsUpdateNodeSpec),
						string(tioperator.OpsUpdateNodeCount),
					}
					componentTypes = append(componentTypes, entity.GetComponentTypeName(comp))
					componentIds = append(componentIds, comp.GetId())
					AutoscaleRcuScaleUpTotal.WithLabelValues(sharedPool.GetId(), "create", comp.GetComponentType().String(), "prepared").Inc()
				}
			}

			if !entity.HasSharedTiProxy(sharedPool) {
				err = s.generatedTiProxyGroup(ctx, sharedPool, tidbCluster, nodeSpecMap, ruConfig)
				if err != nil {
					tilog.Error(ctx, "failed to generate ti proxy group", zap.String("tidbClusterId", tidbCluster.GetId()), zap.Error(err))
					return err
				}

				componentIds = append(componentIds, tidbCluster.GetId())
				componentTypes = append(componentTypes, entity.GetComponentTypeName(&entity.Component{ComponentType: apiv1.ComponentType_TIPROXY}))
			}
		}
		// save tidb cluster to db
		if err := s.tidbClusterRepo.Save(ctx, tidbCluster); err != nil {
			tilog.Error(ctx, "failed to save tidb cluster", zap.String("tidbClusterId", tidbCluster.GetId()), zap.Error(err))
			return status.Errorf(codes.Internal, "failed to save tidb cluster: %s", err)
		}

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

		tidbClusterStr, err := protojson.Marshal(tidbCluster)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to marshal tidb cluster: %s", err)
		}

		args := actions.ActionArgs{
			workflow.TenantIdKey:       actions.ActionArgString(tidbCluster.GetTenantId()),
			workflow.SharedPoolIdKey:   actions.ActionArgString(sharedPool.GetId()),
			workflow.SharedPoolKey:     actions.ActionArgString(string(sharedPoolStr)),
			workflow.DataplaneIdKey:    actions.ActionArgString(sharedPool.GetDataplaneName()),
			workflow.TiDBClusterKey:    actions.ActionArgString(string(tidbClusterStr)),
			workflow.TiDBClusterIdKey:  actions.ActionArgString(tidbCluster.GetId()),
			workflow.ComponentIdsKey:   actions.ActionArgString(strings.Join(componentIds, ",")),
			workflow.ComponentTypesKey: actions.ActionArgString(strings.Join(componentTypes, ",")),
			workflow.CmekIdKey:         actions.ActionArgString(cmekId),
			workflow.MeteringTypeKey:   actions.ActionArgString(workflow.MeteringTypeValueLogicMetaByTiDBCluster),
			utils.TraceIdKey:           utils.GetTraceId(ctx),
		}

		if entity.IsEssentialV2(sharedPool) && len(componentIds) == 0 && len(componentTypes) == 0 {
			args[workflow.SkipComponentHPAApplyKey] = actions.ActionArgString(workflow.SkipComponentHPAApplyValue)
		}
		if entity.HasSharedTiProxy(sharedPool) {
			// A logical-cluster create does not own the shared TiProxy lifecycle.
			args[workflow.SkipSharedTiProxyApplyKey] = actions.ActionArgString(workflow.SkipSharedTiProxyApplyValue)
		}

		// build update options for shared pool components
		for k, v := range optionsMap {
			args[k] = actions.ActionArgString(strings.Join(v, ","))
		}
		tilog.Info(ctx, "rcu_scale_up: prepared create tidb cluster autoscale output",
			zap.String("sharedPoolId", sharedPool.GetId()),
			zap.String("tidbClusterId", tidbCluster.GetId()),
			zap.String("requestUnit", requestUnitProtoToString(tidbCluster.GetRequestUnit())),
			zap.String("componentIds", strings.Join(componentIds, ",")),
			zap.String("componentTypes", strings.Join(componentTypes, ",")),
			zap.String("components", describeComponentSpecsByIDs(sharedPool, componentIds)),
			zap.String("options", describeOptionsMap(optionsMap)),
		)
		// Generate exec ID before submit so we can abort it if transaction fails
		dagExecId = utils.GeneratedDagExecId(ctx, fmt.Sprintf("create-tidbcluster-%s-%s", sharedPool.GetId(), tidbCluster.GetId()))
		err = s.dagExecutor.Submit(ctx, dag.PlanCreateTiDBCluster, args, dagExecId)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to submit create tidb group dag: %s", err)
		}
		dagSubmitted = true

		return nil
	}); err != nil {
		// If DAG was submitted but transaction failed (e.g., ctx timeout during commit),
		// we need to abort the DAG task to prevent it from operating on non-existent data
		if dagSubmitted {
			tilog.Warn(ctx, "transaction failed after DAG submitted, aborting DAG task",
				zap.String("tidbClusterId", tidbCluster.GetId()),
				zap.String("dagExecId", dagExecId),
				zap.Error(err))
			if abortErr := s.dagExecutor.Abort(ctx, dagExecId); abortErr != nil {
				tilog.Error(ctx, "failed to abort DAG task after transaction failure",
					zap.String("tidbClusterId", tidbCluster.GetId()),
					zap.String("dagExecId", dagExecId),
					zap.Error(abortErr))
			}
		}
		return nil, err
	}

	return tidbCluster, nil
}

func (s *serverImpl) checkTiDBClusterKeyspace(ctx context.Context, sharedPool *entity.SharedPool, tidbCluster *entity.TiDBCluster) error {
	// check keyspace status
	keyspaceMeta, err := s.clustergatewayCli.GetKeySpace(ctx, sharedPool, tidbCluster.GetId())
	if err != nil && !clustergateway.IsNotFoundError(err) {
		return status.Errorf(codes.Internal, "failed to check keyspace status: %s", err)
	}

	if keyspaceMeta != nil && keyspaceMeta.GetState() != keyspacepb.KeyspaceState_ENABLED {
		return status.Errorf(codes.FailedPrecondition, "keyspace is not enabled, please update the keyspace status")
	}

	return nil
}

func (s *serverImpl) checkTiDBCluster(ctx context.Context, tidbCluster *entity.TiDBCluster) error {
	if tidbCluster.GetId() == "" {
		return status.Errorf(codes.InvalidArgument, "tidbCluster id is required")
	}

	if strings.Contains(tidbCluster.GetId(), "-") {
		return status.Errorf(codes.InvalidArgument, "tidbCluster id should not contain '-'")
	}

	if tidbCluster.GetTenantId() == "" {
		return status.Errorf(codes.InvalidArgument, "tidbCluster tenant id is required")
	}

	if tidbCluster.GetCapacityMode() != apiv1.CapacityMode_TOPOLOGY_DRIVEN {
		if tidbCluster.GetRequestUnit() == nil {
			return status.Errorf(codes.InvalidArgument, "tidbCluster request unit is required")
		}

		// check request unit
		if tidbCluster.GetRequestUnit().GetMax() < tidbCluster.GetRequestUnit().GetMin() {
			return status.Errorf(codes.InvalidArgument, "request unit min %d cannot be greater than max %d", tidbCluster.GetRequestUnit().GetMin(), tidbCluster.GetRequestUnit().GetMax())
		}
	}

	// check gateway identity format if provided
	if tidbCluster.GetGatewayIdentity() != "" {
		if err := validateDNSIdentity(tidbCluster.GetGatewayIdentity()); err != nil {
			return err
		}
	}

	// only allow create/update one tidb cluster in one shared pool same time
	dagIns, err := s.getRunningDagIns(ctx, []string{dag.PlanCreateTiDBCluster}, nil, nil, []string{tidbCluster.GetId()}, 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, "tidb cluster creation is already running, please try again later")
	}

	// check tidb cluster already exists
	_, err = s.tidbClusterRepo.Load(ctx, tidbCluster.GetId(), true)
	if err == nil {
		tilog.Warn(ctx, "tidb cluster already exists", zap.String("tidbClusterId", tidbCluster.GetId()))
		return status.Errorf(codes.AlreadyExists, "tidb cluster id already exists")
	} else if status.Code(err) != codes.NotFound {
		tilog.Error(ctx, "failed to load tidb cluster", zap.String("tidbClusterId", tidbCluster.GetId()), zap.Error(err))
		return status.Errorf(codes.Internal, "failed to check tidb group id: %s", tidbCluster.GetId())
	}

	return nil
}

func (s *serverImpl) getRequestUnitTotal(ctx context.Context, tidbCluster *entity.TiDBCluster) (*entity.RequestUnit, error) {
	// list all tidb clusters in shared pool
	tidbClusters, err := s.tidbClusterRepo.List(ctx, &repo.TiDBClusterListFilter{
		SharedPoolIds: []string{tidbCluster.GetSharedPoolId()},
		Deleted:       false,
	})
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to list tidb cluster: %s", err)
	}

	// Use uint64 accumulator to prevent int32 overflow for large RU values
	var ruMaxTotal, ruMinTotal uint64
	for _, tc := range tidbClusters {
		if tc.GetId() != tidbCluster.GetId() {
			ruMaxTotal += uint64(tc.GetRequestUnit().GetMax())
			ruMinTotal += uint64(tc.GetRequestUnit().GetMin())
		}
	}

	// Check for uint32 overflow
	if ruMaxTotal > math.MaxUint32 || ruMinTotal > math.MaxUint32 {
		return nil, status.Errorf(codes.InvalidArgument, "total RU overflows uint32")
	}

	return &entity.RequestUnit{
		Max: tidbCluster.GetRequestUnit().GetMax() + uint32(ruMaxTotal),
		Min: tidbCluster.GetRequestUnit().GetMin() + uint32(ruMinTotal),
	}, nil
}

func (s *serverImpl) generatedTiProxyGroup(ctx context.Context, sharedPool *entity.SharedPool, tidbCluster *entity.TiDBCluster, nodeSpecMap map[string][]*entity.NodeSpec, ruConfig *entity.RuConfig) error {
	tiproxyComp := entity.FindTiProxyComponent(sharedPool, tidbCluster.GetId())
	if tiproxyComp == nil {
		tiproxyComp = &entity.Component{
			Id:            tidbCluster.GetId(),
			ComponentType: apiv1.ComponentType_TIPROXY,
			Phase:         apiv1.Component_CREATING,
		}
		sharedPool.ComponentGroups = append(sharedPool.ComponentGroups, tiproxyComp)
	}

	err := s.genDefaultTiProxyComponent(ctx, tiproxyComp, tidbCluster.GetRequestUnit(), nodeSpecMap, ruConfig, false)
	if err != nil {
		tilog.Error(ctx, "failed to generate default tidb component", zap.Error(err))
		return err
	}
	return nil
}

func (s *serverImpl) generateTopologyDrivenTiDBClusterComponents(
	ctx context.Context,
	sharedPool *entity.SharedPool,
	tidbCluster *entity.TiDBCluster,
	nodeSpecMap map[string][]*entity.NodeSpec,
) ([]*entity.Component, error) {
	defaultComponentID := tidbCluster.GetId()
	components := make([]*entity.Component, 0, len(tidbCluster.GetTopologySpec().GetComponents()))

	for _, comp := range tidbCluster.GetTopologySpec().GetComponents() {
		if comp == nil {
			continue
		}

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

		if comp.GetNodeQuota() == nil {
			return nil, status.Errorf(codes.InvalidArgument, "component %s-%s node quota is required", entity.GetComponentTypeName(comp), comp.GetId())
		}
		if comp.GetNodeQuota().GetNodeSpec() == nil {
			return nil, 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 nil, 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 && comp.GetComponentType() == apiv1.ComponentType_TIDB {
			storages, err := s.genDefaultTiDBStorages(sharedPool.GetProvider().GetType(), tidbCluster.GetRequestUnit())
			if err != nil {
				return nil, err
			}
			comp.Storages = storages
		}

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

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

		components = append(components, comp)
	}

	return components, nil
}

func validateTopologyDrivenTiDBClusterComponents(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_TIPROXY: true,
	}

	present := make(map[apiv1.ComponentType]bool, 2)
	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
	}

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

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

	if len(unsupported) > 0 {
		return status.Errorf(codes.InvalidArgument,
			"topology_spec contains unsupported components for TOPOLOGY_DRIVEN tidb cluster mode: %s; allowed: %s",
			strings.Join(unsupported, ","),
			apiv1.ComponentType_TIPROXY.String(),
		)
	}

	return nil
}

// validateDNSIdentity validates the format of DNS identity
func validateDNSIdentity(gatewayIdentity string) error {
	if gatewayIdentity == "" {
		return nil // empty is allowed, will be auto-generated
	}

	// dns identity should be valid DNS label: lowercase letters, numbers, and hyphens
	// Cannot start or end with hyphen, max length 24 characters
	if len(gatewayIdentity) > 24 {
		return status.Errorf(codes.InvalidArgument, "dns identity must be no more than 24 characters")
	}

	// Check for valid DNS label format
	validDNSLabel := regexp.MustCompile(`^[a-z0-9]+([a-z0-9\-]*[a-z0-9])?$`)
	if !validDNSLabel.MatchString(gatewayIdentity) {
		return status.Errorf(codes.InvalidArgument, "dns identity must contain only lowercase letters, numbers, and hyphens, and cannot start or end with hyphen")
	}

	return nil
}

// getCmekId returns the KMS ARN/key identifier from the security configuration based on provider type.
// For AWS: returns the full KMS ARN (arn:aws:kms:region:account:key/key-id)
// For GCP: returns the full key name (projects/.../locations/.../keyRings/.../cryptoKeys/key-id)
// For Azure: returns the full key URL (https://{vault}.vault.azure.net/keys/{key-name}/{version})
// For AliCloud: returns the full ARN (acs:kms:region:account:key/key-id)
func getCmekId(securityConfig *entity.ClusterSecurityConfiguration, providerType entity.ProviderType) (string, error) {
	if securityConfig == nil || securityConfig.GetCmek() == nil {
		return "", nil
	}
	switch providerType {
	case apiv1.ProviderType_AWS:
		arnStr := securityConfig.GetAwsMasterKeyArn()
		if arnStr == "" {
			return "", nil
		}
		// Validate the ARN format
		_, err := utils.ParseAWSKmsKeyId(arnStr)
		if err != nil {
			return "", err
		}
		return arnStr, nil
	case apiv1.ProviderType_ALICLOUD:
		arnStr := securityConfig.GetAlicloudKmsKeyArn()
		if arnStr == "" {
			return "", nil
		}
		_, err := utils.ParseAliCloudKmsKeyId(arnStr)
		if err != nil {
			return "", err
		}
		return arnStr, nil
	// case apiv1.ProviderType_GCP:
	// 	keyName := securityConfig.GetGcpMasterKeyName()
	// 	if keyName == "" {
	// 		return "", nil
	// 	}
	// 	// Validate the key name format
	// 	_, err := utils.ParseGCPKmsKeyId(keyName)
	// 	if err != nil {
	// 		return "", err
	// 	}
	// 	return keyName, nil
	case apiv1.ProviderType_AZURE:
		keyUrl := securityConfig.GetAzureKeyVaultKeyUrl()
		if keyUrl == "" {
			return "", nil
		}
		// Validate the key URL format
		_, err := utils.ParseAzureKeyVaultKeyId(keyUrl)
		if err != nil {
			return "", err
		}
		return keyUrl, nil
	default:
		return "", fmt.Errorf("unsupported provider type: %s", providerType)
	}
}