package server

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

	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/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) ScaleSharedPool(ctx context.Context, sharedPoolId string, scaleComponent *entity.Component, force bool) (*entity.SharedPool, error) {
	if err := s.validateScaleRequest(ctx, sharedPoolId, scaleComponent); err != nil {
		return nil, err
	}

	var sharedPool *entity.SharedPool
	var dagExecId string
	var dagSubmitted bool

	if err := s.transactioner.Transaction(ctx, func(ctx context.Context) error {
		// Acquire lock at the start of transaction
		unlockFunc, err := s.lock(ctx, sharedPoolId)
		if err != nil {
			tilog.Error(ctx, "failed to lock sharedPool", zap.String("sharedPoolId", sharedPoolId), zap.Error(err))
			return err
		}
		defer func() {
			if unlockErr := unlockFunc(); unlockErr != nil {
				tilog.Error(ctx, "failed to unlock sharedPool", zap.String("sharedPoolId", sharedPoolId), zap.Error(unlockErr))
			}
		}()

		// Load and prepare shared pool for scale operation
		sharedPool, err = s.prepareScaleOperation(ctx, sharedPoolId, scaleComponent, force)
		if 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)
		}

		// HPA precondition check MUST happen BEFORE mergeScaleComponent,
		// because mergeScaleComponent mutates currComp.NodeQuota.NodeCount.
		// The gate applies only to manual scale-out, and force=true bypasses it.
		if scaleComponent.GetNodeQuota() != nil {
			currComp := findComponentByTypeAndId(sharedPool, scaleComponent.GetComponentType(), scaleComponent.GetId())
			if currComp != nil {
				if err := s.checkManualScaleHPAGate(
					ctx,
					sharedPool,
					scaleComponent.GetComponentType(),
					scaleComponent.GetId(),
					currComp.GetNodeQuota().GetNodeCount(),
					scaleComponent.GetNodeQuota().GetNodeCount(),
					force,
				); err != nil {
					return err
				}
			}
		}

		optionsMap, err := s.mergeScaleComponent(ctx, sharedPool, scaleComponent, nodeSpecMap, force)
		if err != nil {
			return err
		}

		dagExecId, dagSubmitted, err = s.submitScaleDAG(ctx, sharedPool, scaleComponent, optionsMap, force)
		return err
	}); 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) validateScaleRequest(ctx context.Context, sharedPoolId string, scaleComponent *entity.Component) error {
	if sharedPoolId == "" {
		return status.Errorf(codes.InvalidArgument, "sharedPool id is required")
	}

	if err := checkScaleComponentValid(ctx, scaleComponent, s.cloudProvider); err != nil {
		return err
	}

	return nil
}

func (s *serverImpl) prepareScaleOperation(ctx context.Context, sharedPoolId string, scaleComponent *entity.Component, force bool) (*entity.SharedPool, error) {
	sharedPool, err := s.sharedPoolRepo.Load(ctx, sharedPoolId, false)
	if err != nil {
		if status.Code(err) != codes.NotFound {
			return nil, status.Errorf(codes.Internal, "failed to load sharedPool: %s", err)
		}
		return nil, status.Errorf(codes.NotFound, "sharedPool %s not found", sharedPoolId)
	}
	if err := guardSharedPoolMutation(sharedPool, sharedPoolMutationBlockedWhilePaused); err != nil {
		return nil, err
	}

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

	if scaleComponent.GetId() == "" {
		tilog.Warn(ctx, "scale component id is empty, will use default id")
		scaleComponent.Id = sharedPool.GetDefaultComponenId()
	}

	if err := validateNodeCountForComponent(ctx, scaleComponent, sharedPool); err != nil {
		return nil, err
	}

	if err := checkSharedPoolCanBeModified(ctx, sharedPool, scaleComponent, force); err != nil {
		return nil, err
	}

	// Forced operations are an explicit recovery escape hatch and may proceed
	// while the TiKV remote-worker topology is still converging.
	if !force && isTiKVRemoteWorkerGuardedComponentType(scaleComponent.GetComponentType()) {
		if err := s.guardTiKVRemoteWorkersTopologyStable(ctx, sharedPool); err != nil {
			return nil, err
		}
	}

	if s.cloudProvider == "aws" {
		if err := s.checkVolumeCanBeModified(ctx, sharedPool, scaleComponent); err != nil {
			return nil, err
		}
	}

	return sharedPool, nil
}

func (s *serverImpl) submitScaleDAG(ctx context.Context, sharedPool *entity.SharedPool, scaleComponent *entity.Component, optionsMap map[string][]string, force bool) (string, bool, error) {
	sharedPoolStr, err := protojson.Marshal(sharedPool)
	if err != nil {
		return "", false, status.Errorf(codes.Internal, "failed to marshal sharedPool: %s", err)
	}

	args := actions.ActionArgs{
		workflow.SharedPoolIdKey:   actions.ActionArgString(sharedPool.GetId()),
		workflow.SharedPoolKey:     actions.ActionArgString(string(sharedPoolStr)),
		workflow.ComponentIdsKey:   actions.ActionArgString(scaleComponent.GetId()),
		workflow.ComponentTypesKey: actions.ActionArgString(entity.GetComponentTypeName(scaleComponent)),
		workflow.ForceKey:          actions.ActionArgString(strconv.FormatBool(force)),
		utils.TraceIdKey:           utils.GetTraceId(ctx),
	}

	if needSendMeteringMetadata(scaleComponent) {
		args[workflow.MeteringTypeKey] = actions.ActionArgString(workflow.MeteringTypeValueLogicMetaBySharedPool)
	}

	for k, v := range optionsMap {
		args[k] = actions.ActionArgString(strings.Join(v, ","))
	}

	// because global will Split the scale cluster request
	// eg.  1 4c8g  tidb -> 2 8c16g  tidb
	// global will request 2 times
	// 1. 1 4c8g  tidb -> 2 4c8g  tidb
	// 2. 2 4c8g  tidb -> 2 8c16g  tidb
	// so we need to avoid the dag exec id is the same
	dagExecId := utils.GeneratedDagExecId(ctx,
		fmt.Sprintf("scale-sharedpool-%s-%s-%s-%s",
			sharedPool.GetId(),
			entity.GetComponentTypeName(scaleComponent),
			scaleComponent.GetId(),
			utils.FnvHash(string(sharedPoolStr)),
		))

	if err := s.dagExecutor.Submit(ctx, dag.PlanUpdateComponent, args, dagExecId); err != nil {
		return "", false, status.Errorf(codes.Internal, "failed to submit scale shared pool dag: %s", err)
	}

	return dagExecId, true, nil
}

func (s *serverImpl) mergeScaleComponent(ctx context.Context, sharedPool *entity.SharedPool, scaleComponent *entity.Component, nodeSpecMap map[string][]*entity.NodeSpec, force bool) (map[string][]string, error) {
	sharedPool.Phase = apiv1.SharedPool_MODIFYING

	optionsMap := map[string][]string{
		entity.GetUpdateComponentKey(scaleComponent): {},
	}

	if scaleComponent.GetId() == "" {
		scaleComponent.Id = sharedPool.GetDefaultComponenId()
	}

	currComp, err := s.findComponentInSharedPool(sharedPool, scaleComponent)
	if err != nil {
		return optionsMap, err
	}

	if scaleComponent.GetNodeQuota() != nil {
		if err := s.applyNodeQuotaChanges(ctx, sharedPool, scaleComponent, currComp, nodeSpecMap, optionsMap, force); err != nil {
			return optionsMap, err
		}
	}

	if err := s.applyStorageChanges(ctx, sharedPool, scaleComponent, currComp, optionsMap); err != nil {
		return optionsMap, err
	}

	currComp.Phase = apiv1.Component_MODIFYING

	return optionsMap, nil
}

func (s *serverImpl) findComponentInSharedPool(sharedPool *entity.SharedPool, scaleComponent *entity.Component) (*entity.Component, error) {
	currComponentMap := make(map[string]map[apiv1.ComponentType]*entity.Component)
	for _, comp := range sharedPool.GetComponentGroups() {
		if currComponentMap[comp.GetId()] == nil {
			currComponentMap[comp.GetId()] = make(map[apiv1.ComponentType]*entity.Component)
		}
		currComponentMap[comp.GetId()][comp.GetComponentType()] = comp
	}

	if currComponentMap[scaleComponent.GetId()] == nil {
		return nil, status.Errorf(codes.NotFound, "component id %s not found", scaleComponent.GetId())
	}

	currComp := currComponentMap[scaleComponent.GetId()][scaleComponent.GetComponentType()]
	if currComp == nil {
		return nil, status.Errorf(codes.NotFound, "component %s not found", entity.GetComponentGroupName(scaleComponent))
	}

	return currComp, nil
}

func (s *serverImpl) applyNodeQuotaChanges(ctx context.Context, sharedPool *entity.SharedPool, scaleComponent, currComp *entity.Component, nodeSpecMap map[string][]*entity.NodeSpec, optionsMap map[string][]string, force bool) error {
	// Save original values before any mutation for scale-in safety check
	origNodeCount := currComp.GetNodeQuota().GetNodeCount()
	origMinNodeCount := currComp.GetNodeQuota().GetMinNodeCount()
	origMaxNodeCount := currComp.GetNodeQuota().GetMaxNodeCount()
	origNodeSpec := currComp.GetNodeQuota().GetNodeSpec()
	targetNodeCount := scaleComponent.GetNodeQuota().GetNodeCount()
	targetMinNodeCount := origMinNodeCount
	targetMaxNodeCount := origMaxNodeCount
	allowWorkerZero := isTopologyDrivenWorkerZeroCountAllowed(sharedPool, currComp.GetComponentType())

	if allowWorkerZero && targetNodeCount == 0 {
		// For topology-driven worker scale-down to zero, accept zero min/max from request.
		targetMinNodeCount = scaleComponent.GetNodeQuota().GetMinNodeCount()
		targetMaxNodeCount = scaleComponent.GetNodeQuota().GetMaxNodeCount()
	}

	if scaleComponent.GetNodeQuota().GetMinNodeCount() > 0 {
		targetMinNodeCount = scaleComponent.GetNodeQuota().GetMinNodeCount()
	}
	if scaleComponent.GetNodeQuota().GetMaxNodeCount() > 0 {
		targetMaxNodeCount = scaleComponent.GetNodeQuota().GetMaxNodeCount()
	}

	// Step 1: Deterministic validations first (local, cheap, fail fast)
	if targetMaxNodeCount > 0 && targetNodeCount > targetMaxNodeCount {
		return status.Errorf(codes.InvalidArgument, "scale component node count %d is greater than max node count %d",
			targetNodeCount, targetMaxNodeCount)
	}

	if targetNodeCount < targetMinNodeCount {
		return status.Errorf(codes.InvalidArgument, "scale component node count %d is less than min node count %d",
			targetNodeCount, targetMinNodeCount)
	}

	if targetMaxNodeCount > 0 && targetMinNodeCount > targetMaxNodeCount {
		return status.Errorf(codes.InvalidArgument, "scale component min node count %d is greater than max node count %d",
			targetMinNodeCount, targetMaxNodeCount)
	}

	// Step 2: Resolve target node spec (if spec change requested)
	var targetNodeSpec *entity.NodeSpec
	reqNodeSpec := scaleComponent.GetNodeQuota().GetNodeSpec()
	hasSpecChange := reqNodeSpec != nil && (reqNodeSpec.GetMemoryGi() != 0 && reqNodeSpec.GetCpu() != 0)

	// Reject partial node spec (only cpu or only memory specified)
	if reqNodeSpec != nil && !hasSpecChange && (reqNodeSpec.GetCpu() != 0 || reqNodeSpec.GetMemoryGi() != 0) {
		return status.Errorf(codes.InvalidArgument, "node spec must specify both cpu and memory, or neither")
	}

	if hasSpecChange {
		resolvedSpec, err := getNodeSpec(ctx, scaleComponent.GetNodeQuota().GetNodeSpec(), nodeSpecMap, origNodeSpec.GetWithLocalDisk())
		if err != nil {
			tilog.Error(ctx, "failed to get scale node spec", zap.Error(err))
			return status.Errorf(codes.InvalidArgument, "failed to get scale node spec: %s", err)
		}
		targetNodeSpec = resolvedSpec
	} else {
		targetNodeSpec = origNodeSpec
	}

	// Step 3: Run RU validation (local, cheap) before remote metrics fetch
	if hasSpecChange {
		ruConfig, err := s.ruConfigRepo.GetRuConfig(ctx, sharedPool.GetId(), apiv1.RuConfig_SHARED_POOL)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to get ru config: %s", err)
		}

		minRU, err := s.getMinRUForComponent(ctx, currComp, sharedPool)
		if err != nil {
			return err
		}

		if err := validateNodeSpecAgainstRUConfig(ctx, currComp.GetComponentType(), targetNodeSpec, ruConfig, minRU, currComp.GetId(), force); err != nil {
			return err
		}
	}

	// Step 4: Run checkScaleInSafety exactly once with resolved target spec
	// This prevents false rejection on count-down + spec-up (e.g., 3×4c → 2×16c)
	if err := s.checkScaleInSafety(ctx, sharedPool, currComp,
		origNodeCount, origNodeSpec,
		targetNodeCount, targetNodeSpec, force); err != nil {
		return err
	}

	// Step 5: Mutate state
	currComp.GetNodeQuota().MinNodeCount = targetMinNodeCount
	currComp.GetNodeQuota().MaxNodeCount = targetMaxNodeCount
	if targetMinNodeCount != origMinNodeCount {
		optionsMap[entity.GetUpdateComponentKey(scaleComponent)] = append(optionsMap[entity.GetUpdateComponentKey(scaleComponent)],
			string(tioperator.OpsUpdateMinCount))
	}
	if targetMaxNodeCount != origMaxNodeCount {
		optionsMap[entity.GetUpdateComponentKey(scaleComponent)] = append(optionsMap[entity.GetUpdateComponentKey(scaleComponent)],
			string(tioperator.OpsUpdateMaxCount))
	}

	currComp.GetNodeQuota().NodeCount = targetNodeCount
	if targetNodeCount != origNodeCount {
		optionsMap[entity.GetUpdateComponentKey(scaleComponent)] = append(optionsMap[entity.GetUpdateComponentKey(scaleComponent)],
			string(tioperator.OpsUpdateNodeCount))
	}

	if hasSpecChange {
		currComp.GetNodeQuota().NodeSpec = targetNodeSpec
		optionsMap[entity.GetUpdateComponentKey(scaleComponent)] = append(optionsMap[entity.GetUpdateComponentKey(scaleComponent)],
			string(tioperator.OpsUpdateNodeSpec))

		if err := s.handleLocalDiskChanges(ctx, scaleComponent, currComp, optionsMap); err != nil {
			return err
		}
	}

	return nil
}

func (s *serverImpl) getMinRUForComponent(ctx context.Context, currComp *entity.Component, sharedPool *entity.SharedPool) (uint32, error) {
	switch currComp.GetComponentType() {
	case apiv1.ComponentType_TIPROXY:
		if entity.HasSharedTiProxy(sharedPool) {
			return s.getTotalMinRUForSharedPool(ctx, sharedPool.GetId())
		}
		tidbCluster, err := s.tidbClusterRepo.Load(ctx, currComp.GetId(), false)
		if err != nil {
			return 0, status.Errorf(codes.Internal, "failed to load tidb cluster %s: %s", currComp.GetId(), err)
		}
		return tidbCluster.GetRequestUnit().GetMin(), nil
	case apiv1.ComponentType_TIDB:
		tidbGroup, err := s.tidbGroupRepo.Load(ctx, currComp.GetId(), false)
		if err != nil {
			return 0, status.Errorf(codes.Internal, "failed to load tidb group %s: %s", currComp.GetId(), err)
		}
		tidbCluster, err := s.tidbClusterRepo.Load(ctx, tidbGroup.GetClusterId(), false)
		if err != nil {
			return 0, status.Errorf(codes.Internal, "failed to load tidb cluster %s: %s", tidbGroup.GetClusterId(), err)
		}
		return tidbCluster.GetRequestUnit().GetMin(), nil
	default:
		return s.getTotalMinRUForSharedPool(ctx, sharedPool.GetId())
	}
}

func (s *serverImpl) handleLocalDiskChanges(ctx context.Context, scaleComponent, currComp *entity.Component, optionsMap map[string][]string) error {
	if !currComp.GetNodeQuota().GetNodeSpec().GetWithLocalDisk() {
		return nil
	}

	tilog.Info(ctx, "scale component with local disk", zap.Any("nodeSpec", currComp.GetNodeQuota().GetNodeSpec()))

	for _, storage := range currComp.GetStorages() {
		if storage.GetName() != apiv1.Component_Storage_DATA {
			continue
		}

		if storage.GetType() != entity.LocalSSD {
			return status.Errorf(codes.InvalidArgument, "%s %s storage type is not local ssd, but node spec with local disk",
				entity.GetComponentGroupName(currComp), apiv1.Component_Storage_DATA.String())
		}

		if storage.GetSizeGi() > currComp.GetNodeQuota().GetNodeSpec().GetLocalDiskSizeGi() {
			return status.Errorf(codes.InvalidArgument, "%s %s storage size %d Gi is greater than node spec local disk size %d Gi",
				entity.GetComponentGroupName(currComp), apiv1.Component_Storage_DATA.String(),
				storage.GetSizeGi(), currComp.GetNodeQuota().GetNodeSpec().GetLocalDiskSizeGi())
		}

		storage.SizeGi = currComp.GetNodeQuota().GetNodeSpec().GetLocalDiskSizeGi()
		optionsMap[entity.GetUpdateComponentKey(scaleComponent)] = append(optionsMap[entity.GetUpdateComponentKey(scaleComponent)],
			string(tioperator.OpsUpdateDataSize))
	}

	return nil
}

func (s *serverImpl) applyStorageChanges(ctx context.Context, sharedPool *entity.SharedPool, scaleComponent, currComp *entity.Component, optionsMap map[string][]string) error {
	for _, newStorage := range scaleComponent.GetStorages() {
		var found bool
		for _, currStorage := range currComp.GetStorages() {
			if newStorage.GetName() != currStorage.GetName() {
				continue
			}

			if currStorage.GetName() == apiv1.Component_Storage_DATA && currStorage.GetType() == entity.LocalSSD {
				return status.Errorf(codes.InvalidArgument, "%s %s do not support to modify local storage",
					newStorage.GetName(), entity.GetComponentGroupName(currComp))
			}

			found = true
			if err := s.applyStorageSizeChange(scaleComponent, currComp, currStorage, newStorage, optionsMap); err != nil {
				return err
			}

			if err := s.applyStoragePerformanceChange(scaleComponent, currComp, currStorage, newStorage, optionsMap); err != nil {
				return err
			}

			break
		}

		if !found {
			// allow init volume for cop worker when not found
			// TODO: remove it when all prod clusters cop-worker have migrated to use data volume
			if newStorage.GetName() == apiv1.Component_Storage_DATA &&
				scaleComponent.GetComponentType() == apiv1.ComponentType_COPROCESSOR_WORKER {
				key := entity.GetUpdateComponentKey(scaleComponent)
				optionsMap[key] = append(optionsMap[key], string(tioperator.OpsUpdateDataVolumeAttributesClass))
				storage, err := s.genDefaultCoprocessorWorkerStorages(sharedPool.GetProvider().GetType())
				if err != nil {
					tilog.Error(ctx, "failed to generate default coprocessor worker storage", zap.Error(err))
					return status.Errorf(codes.InvalidArgument, "storage %s not found in current component %s",
						newStorage.GetName(), entity.GetComponentGroupName(currComp))
				}
				if currComp.Storages == nil {
					currComp.Storages = storage
				} else {
					currComp.Storages = append(currComp.Storages, storage...)
				}
			} else {
				return status.Errorf(codes.InvalidArgument, "storage %s not found in current component %s",
					newStorage.GetName(), entity.GetComponentGroupName(currComp))
			}
		}
	}

	return nil
}

func (s *serverImpl) applyStorageSizeChange(scaleComponent, _ *entity.Component, currStorage, newStorage *entity.Storage, optionsMap map[string][]string) error {
	if newStorage.GetSizeGi() == 0 {
		return nil
	}

	if currStorage.GetSizeGi() > newStorage.GetSizeGi() {
		return status.Errorf(codes.PermissionDenied, "only support to scale up storage size, current size: %d Gi, new size: %d Gi",
			currStorage.GetSizeGi(), newStorage.GetSizeGi())
	}

	currStorage.SizeGi = newStorage.GetSizeGi()
	switch currStorage.GetName() {
	case apiv1.Component_Storage_DATA:
		optionsMap[entity.GetUpdateComponentKey(scaleComponent)] = append(optionsMap[entity.GetUpdateComponentKey(scaleComponent)],
			string(tioperator.OpsUpdateDataSize))
	case apiv1.Component_Storage_RAFTLOG:
		optionsMap[entity.GetUpdateComponentKey(scaleComponent)] = append(optionsMap[entity.GetUpdateComponentKey(scaleComponent)],
			string(tioperator.OpsUpdateRaftLogSize))
	default:
		return status.Errorf(codes.InvalidArgument, "storage %s not support to scale size", newStorage.GetName())
	}

	return nil
}

func (s *serverImpl) applyStoragePerformanceChange(scaleComponent, _ *entity.Component, currStorage, newStorage *entity.Storage, optionsMap map[string][]string) error {
	if newStorage.GetIops() == 0 && newStorage.GetThroughputMbs() == 0 && newStorage.GetType() == "" {
		return nil
	}

	if newStorage.GetType() != "" {
		currStorage.Type = newStorage.GetType()
	}
	if newStorage.GetIops() != 0 {
		currStorage.Iops = newStorage.GetIops()
	}
	if newStorage.GetThroughputMbs() != 0 {
		currStorage.ThroughputMbs = newStorage.GetThroughputMbs()
	}

	switch currStorage.GetName() {
	case apiv1.Component_Storage_DATA:
		optionsMap[entity.GetUpdateComponentKey(scaleComponent)] = append(optionsMap[entity.GetUpdateComponentKey(scaleComponent)],
			string(tioperator.OpsUpdateDataVolumeAttributesClass))
	case apiv1.Component_Storage_RAFTLOG:
		optionsMap[entity.GetUpdateComponentKey(scaleComponent)] = append(optionsMap[entity.GetUpdateComponentKey(scaleComponent)],
			string(tioperator.OpsUpdateRaftLogVolumeAttributesClass))
	default:
		return status.Errorf(codes.InvalidArgument, "storage %s not support to scale iops or throughput", newStorage.GetName())
	}

	return nil
}

func checkScaleComponentValid(ctx context.Context, scaleComponent *entity.Component, provider string) error {
	if scaleComponent == nil {
		return status.Errorf(codes.InvalidArgument, "scale components is empty")
	}

	if scaleComponent.GetComponentType() == apiv1.ComponentType_COMPONENT_TYPE_UNSPECIFIED {
		return status.Errorf(codes.InvalidArgument, "scale component type is required")
	}

	for _, storage := range scaleComponent.GetStorages() {
		storageType := storage.GetType()
		if storageType == "" {
			continue
		}

		if storageType == entity.LocalSSD {
			continue
		}

		if err := entity.ValidateStorageTypeForProvider(storage, provider, ""); err != nil {
			return status.Errorf(codes.InvalidArgument, "%s", err.Error())
		}
	}

	if scaleComponent.GetNodeQuota() != nil {
		if err := validateNodeCountForComponent(ctx, scaleComponent, nil); err != nil {
			return err
		}
	}

	return nil
}

func validateNodeCountForComponent(ctx context.Context, scaleComponent *entity.Component, sharedPool *entity.SharedPool) error {
	if scaleComponent.GetNodeQuota() == nil {
		return nil
	}

	if scaleComponent.GetNodeQuota().GetNodeCount() == 0 {
		tilog.Warn(ctx, "scale component node count is zero", zap.Any("scaleComp", scaleComponent))
		if scaleComponent.GetComponentType() == apiv1.ComponentType_TIDB_WORKER || scaleComponent.GetComponentType() == apiv1.ComponentType_TIKV_WORKER {
			// In request pre-check stage (sharedPool is nil), defer this decision to prepare stage
			// where capacity_mode is available.
			if sharedPool == nil {
				return nil
			}
		}
		if scaleComponent.GetComponentType() != apiv1.ComponentType_TIFLASH_WRITE &&
			scaleComponent.GetComponentType() != apiv1.ComponentType_TIFLASH_COMPUTE &&
			scaleComponent.GetComponentType() != apiv1.ComponentType_SCHEMA_MANAGER &&
			!isTopologyDrivenWorkerZeroCountAllowed(sharedPool, scaleComponent.GetComponentType()) {
			return status.Errorf(codes.InvalidArgument, "scale component node count is required, only TiFlash node count can be zero")
		}
	}

	if scaleComponent.GetComponentType() == apiv1.ComponentType_TIKV || scaleComponent.GetComponentType() == apiv1.ComponentType_PD {
		if scaleComponent.GetNodeQuota().GetNodeCount()%3 != 0 {
			return status.Errorf(codes.InvalidArgument, "scale component %s node count must be multiple of 3",
				entity.GetComponentTypeName(scaleComponent))
		}
	}

	if scaleComponent.GetComponentType() == apiv1.ComponentType_PD_TSO && scaleComponent.GetNodeQuota().GetNodeCount() != 2 {
		return status.Errorf(codes.InvalidArgument, "component %s node count must be 2",
			entity.GetComponentTypeName(scaleComponent))
	}

	return nil
}

func isTopologyDrivenWorkerZeroCountAllowed(sharedPool *entity.SharedPool, componentType apiv1.ComponentType) bool {
	if sharedPool == nil || sharedPool.GetCapacityMode() != apiv1.CapacityMode_TOPOLOGY_DRIVEN {
		return false
	}

	return componentType == apiv1.ComponentType_TIDB_WORKER || componentType == apiv1.ComponentType_TIKV_WORKER
}

func needSendMeteringMetadata(scaleComponent *entity.Component) bool {
	if scaleComponent == nil {
		return false
	}
	if scaleComponent.ComponentType != apiv1.ComponentType_TIKV {
		return false
	}
	for _, storage := range scaleComponent.GetStorages() {
		if storage.GetName() == apiv1.Component_Storage_DATA {
			if storage.GetType() != "" || storage.Iops != 0 || storage.ThroughputMbs != 0 {
				return true
			}
		}
	}
	return false
}

func checkSharedPoolCanBeModified(ctx context.Context, sharedPool *entity.SharedPool, targetComp *entity.Component, force bool) error {
	if sharedPool.GetPhase() == apiv1.SharedPool_CREATING {
		return status.Errorf(codes.FailedPrecondition, "shared pool is creating, please try again later")
	}

	var found bool
	for _, comp := range sharedPool.GetComponentGroups() {
		if comp.GetComponentType() != targetComp.GetComponentType() ||
			comp.GetId() != targetComp.GetId() {
			continue
		}
		found = true
		if err := checkComponentDeleteFence(comp); err != nil {
			return err
		}
		if comp.GetPhase() != apiv1.Component_AVAILABLE {
			if isTopologyDrivenWorkerCrossZeroTransition(sharedPool, comp, targetComp) {
				tilog.Info(ctx, "allow topology-driven worker scale across zero for unavailable component",
					zap.String("sharedPoolId", sharedPool.GetId()),
					zap.String("componentType", entity.GetComponentTypeName(comp)),
					zap.String("componentId", comp.GetId()),
					zap.String("componentPhase", comp.GetPhase().String()),
					zap.Uint32("currentNodeCount", comp.GetNodeQuota().GetNodeCount()),
					zap.Uint32("targetNodeCount", targetComp.GetNodeQuota().GetNodeCount()),
				)
				comp.Phase = apiv1.Component_MODIFYING
				break
			}

			tilog.Warn(ctx, "component is not available",
				zap.String("sharedPoolId", sharedPool.GetId()),
				zap.String("componentType", entity.GetComponentTypeName(comp)),
				zap.String("componentId", comp.GetId()),
				zap.String("componentPhase", comp.GetPhase().String()))

			if !force {
				return status.Errorf(codes.FailedPrecondition, "target component %s %s is not available, current phase: %s",
					entity.GetComponentTypeName(comp), comp.GetId(), comp.GetPhase().String())
			}

			if err := validateStorageNotDecreasing(comp, targetComp); err != nil {
				return err
			}

			tilog.Warn(ctx, "force modify component, ignore component phase check",
				zap.String("sharedPoolId", sharedPool.GetId()),
				zap.String("componentType", entity.GetComponentTypeName(comp)),
				zap.String("componentId", comp.GetId()),
				zap.String("componentPhase", comp.GetPhase().String()))
		}

		comp.Phase = apiv1.Component_MODIFYING
		break
	}

	if !found {
		return status.Errorf(codes.NotFound, "component %s %s not found in shared pool %s",
			entity.GetComponentTypeName(targetComp), targetComp.GetId(), sharedPool.GetId())
	}
	return nil
}

func checkComponentDeleteFence(comp *entity.Component) error {
	if comp.GetPhase() != apiv1.Component_DELETING {
		return nil
	}
	return status.Errorf(codes.FailedPrecondition, "target component %s %s is deleting, please try again later",
		entity.GetComponentTypeName(comp), comp.GetId())
}

func isTopologyDrivenWorkerCrossZeroTransition(sharedPool *entity.SharedPool, currComp, targetComp *entity.Component) bool {
	if sharedPool == nil || sharedPool.GetCapacityMode() != apiv1.CapacityMode_TOPOLOGY_DRIVEN {
		return false
	}

	if currComp == nil || targetComp == nil || currComp.GetNodeQuota() == nil || targetComp.GetNodeQuota() == nil {
		return false
	}

	if currComp.GetComponentType() != apiv1.ComponentType_TIDB_WORKER && currComp.GetComponentType() != apiv1.ComponentType_TIKV_WORKER {
		return false
	}

	currNodeCount := currComp.GetNodeQuota().GetNodeCount()
	targetNodeCount := targetComp.GetNodeQuota().GetNodeCount()

	return (currNodeCount == 0 && targetNodeCount > 0) || (currNodeCount > 0 && targetNodeCount == 0)
}

func validateStorageNotDecreasing(comp, targetComp *entity.Component) error {
	for _, target := range targetComp.GetStorages() {
		for _, curr := range comp.GetStorages() {
			if target.GetName() != curr.GetName() {
				continue
			}
			if target.GetSizeGi() != 0 && target.GetSizeGi() < curr.GetSizeGi() {
				return status.Errorf(codes.InvalidArgument, "%s sizeGi can not be decreased, current sizeGi: %v",
					target.GetName().String(),
					curr.GetSizeGi(),
				)
			}
		}
	}
	return nil
}

func (s *serverImpl) getTotalMinRUForSharedPool(ctx context.Context, sharedPoolId string) (uint32, error) {
	tidbClusters, err := s.tidbClusterRepo.List(ctx, &repo.TiDBClusterListFilter{
		SharedPoolIds: []string{sharedPoolId},
		Deleted:       false,
	})
	if err != nil {
		return 0, status.Errorf(codes.Internal, "failed to list tidb clusters: %s", err)
	}

	var total uint64
	for _, tc := range tidbClusters {
		total += uint64(tc.GetRequestUnit().GetMin())
	}
	if total > math.MaxUint32 {
		return 0, status.Errorf(codes.Internal, "total min RU overflow: %d", total)
	}
	return uint32(total), nil
}

func validateNodeSpecAgainstRUConfig(
	ctx context.Context,
	componentType apiv1.ComponentType,
	requestedNodeSpec *entity.NodeSpec,
	ruConfig *entity.RuConfig,
	minRU uint32,
	componentId string,
	force bool,
) error {
	if minRU == 0 {
		return nil
	}

	// Skip RU validation for components not governed by the RU config.
	// SCHEMA_MANAGER and other internal components are intentionally absent from
	// the RU config because they don't host user RU directly, so node-spec
	// against RU validation does not apply to them.
	if !ruConfig.HasComponent(componentType) {
		return nil
	}

	requiredNodeSpec, err := ruConfig.GetNodeSpec(componentType, minRU)
	if err != nil {
		return status.Errorf(codes.Internal, "failed to get min node spec from ru config: %s", err)
	}

	if requestedNodeSpec.GetCpu() < requiredNodeSpec.GetCpu() ||
		requestedNodeSpec.GetMemoryGi() < requiredNodeSpec.GetMemoryGi() {
		if force {
			tilog.Warn(ctx, "node spec below RU config requirement for min RU, force flag set, proceeding",
				zap.String("componentType", componentType.String()),
				zap.String("componentId", componentId),
				zap.Uint32("requestedCpu", requestedNodeSpec.GetCpu()),
				zap.Uint32("requiredCpu", requiredNodeSpec.GetCpu()),
				zap.Uint32("requestedMemory", requestedNodeSpec.GetMemoryGi()),
				zap.Uint32("requiredMemory", requiredNodeSpec.GetMemoryGi()),
				zap.Uint32("minRU", minRU),
			)
			return nil
		}
		return status.Errorf(codes.InvalidArgument,
			"component %s node spec (cpu: %d, memory: %dGi) cannot support min RU %d, required spec: cpu >= %d, memory >= %dGi",
			componentType.String(),
			requestedNodeSpec.GetCpu(), requestedNodeSpec.GetMemoryGi(),
			minRU,
			requiredNodeSpec.GetCpu(), requiredNodeSpec.GetMemoryGi())
	}

	return nil
}

// effectiveCPU calculates usable CPU capacity after system reservation
func effectiveCPU(spec *entity.NodeSpec, count uint32) float64 {
	return float64(max(int(spec.GetCpu())-1, 0)) * float64(count)
}

// effectiveMem calculates usable memory capacity after system reservation
func effectiveMem(spec *entity.NodeSpec, count uint32) float64 {
	return float64(tioperator.ComputeAvailableMemory(uint64(spec.GetMemoryGi())*1024)) * 1024 * 1024 * float64(count)
}

func (s *serverImpl) checkScaleInSafety(ctx context.Context, sharedPool *entity.SharedPool, component *entity.Component,
	currentNodeCount uint32, currentNodeSpec *entity.NodeSpec, targetNodeCount uint32, targetNodeSpec *entity.NodeSpec, force bool,
) error {
	if allowTopologyDrivenWorkerScaleAcrossZero(sharedPool, component.GetComponentType(), currentNodeCount, targetNodeCount) {
		tilog.Info(ctx, "scale protection skipped: topology-driven worker cross-zero transition",
			zap.String("componentType", component.GetComponentType().String()),
			zap.Uint32("currentNodeCount", currentNodeCount),
			zap.Uint32("targetNodeCount", targetNodeCount))
		return nil
	}

	if force {
		tilog.Info(ctx, "scale protection skipped: force flag is true")
		return nil
	}

	if currentNodeCount == 0 {
		tilog.Info(ctx, "scale protection skipped: current node count is zero")
		return nil
	}

	// Use effective capacity (after system reservation) for both trigger and enforcement
	// This ensures consistent behavior and prevents false skips on horizontal rebalancing
	currentCpuCapacity := effectiveCPU(currentNodeSpec, currentNodeCount)
	currentMemCapacity := effectiveMem(currentNodeSpec, currentNodeCount)
	targetCpuCapacity := effectiveCPU(targetNodeSpec, targetNodeCount)
	targetMemCapacity := effectiveMem(targetNodeSpec, targetNodeCount)

	if targetCpuCapacity >= currentCpuCapacity && targetMemCapacity >= currentMemCapacity {
		tilog.Info(ctx, "scale protection skipped: target capacity not decreased",
			zap.Float64("current_cpu_capacity", currentCpuCapacity),
			zap.Float64("target_cpu_capacity", targetCpuCapacity),
			zap.Float64("current_mem_capacity", currentMemCapacity/(1024*1024*1024)),
			zap.Float64("target_mem_capacity", targetMemCapacity/(1024*1024*1024)))
		return nil
	}

	metrics, err := s.clustergatewayCli.GetComponentResourceMetrics(
		ctx, sharedPool, entity.GetComponentTypeName(component), component.GetId())
	if err != nil {
		return status.Errorf(codes.FailedPrecondition, "failed to get metrics for scale protection: %v. Use force=true to bypass.", err)
	}

	// Defensive check: metrics-server may be unavailable in some environments
	// Block the operation unless force=true to prevent unsafe scale-in
	if !metrics.GetMetricsServerAvailable() {
		return status.Errorf(codes.FailedPrecondition, "metrics server not available, cannot verify scale safety. Use force=true to bypass.")
	}

	var currentCpuUsed, currentMemUsed float64
	for _, comp := range metrics.GetComponents() {
		currentCpuUsed += comp.GetCpuCoresUsed()
		currentMemUsed += float64(comp.GetMemoryBytesUsed())
	}

	// Calculate available capacity after reserving resources for system overhead
	// CPU: reserve 1 core per node for system processes (minimum 0)
	availableTargetCpu := float64(max(int(targetNodeSpec.GetCpu())-1, 0)) * float64(targetNodeCount)
	// Memory: ComputeAvailableMemory takes MiB and returns available MiB after system reservation
	// Convert: Gi -> Mi (*1024), ComputeAvailableMemory(MiB), MiB -> bytes (*1024^2)
	availableTargetMem := float64(tioperator.ComputeAvailableMemory(uint64(targetNodeSpec.GetMemoryGi())*1024)) * 1024 * 1024 * float64(targetNodeCount)

	memoryUsedForSafety := currentMemUsed
	blockCacheAdjusted := false
	if currentCpuUsed <= availableTargetCpu &&
		currentMemUsed > availableTargetMem &&
		isTiKVHorizontalScaleIn(component, currentNodeCount, currentNodeSpec, targetNodeCount, targetNodeSpec) {
		memoryBaseline, blockCacheBytes := s.getTiKVScaleInMemoryMetrics(
			ctx, sharedPool, metrics.GetComponents())
		reclaimableBlockCache := reclaimableTiKVBlockCacheBytes(
			blockCacheBytes, currentNodeCount, currentNodeCount-targetNodeCount)
		if len(blockCacheBytes) == int(currentNodeCount) {
			memoryUsedForSafety = max(memoryBaseline-reclaimableBlockCache, 0)
			blockCacheAdjusted = true
			tilog.Info(ctx, "calculated TiKV horizontal scale-in memory demand",
				zap.Float64("current_memory_gib", currentMemUsed/(1024*1024*1024)),
				zap.Float64("memory_baseline_gib", memoryBaseline/(1024*1024*1024)),
				zap.Float64("reclaimable_block_cache_gib", reclaimableBlockCache/(1024*1024*1024)),
				zap.Float64("memory_demand_gib", memoryUsedForSafety/(1024*1024*1024)),
				zap.Float64("target_memory_capacity_gib", availableTargetMem/(1024*1024*1024)))
		} else {
			tilog.Warn(ctx, "skipped TiKV block cache adjustment: pod count mismatch",
				zap.Uint32("current_node_count", currentNodeCount),
				zap.Int("metrics_pod_count", len(blockCacheBytes)))
		}
	}

	if currentCpuUsed > availableTargetCpu || memoryUsedForSafety > availableTargetMem {
		// Only claim the reported memory excludes reclaimable block cache when
		// the adjustment actually ran; otherwise it is the raw working set.
		memoryLabel := "mem"
		if blockCacheAdjusted {
			memoryLabel = "non-reclaimable memory demand"
		}
		return status.Errorf(codes.FailedPrecondition,
			"scale blocked: current load (cpu: %.2f cores, %s: %.2f GiB) exceeds target capacity (cpu: %.2f cores, mem: %.2f GiB after reserving 1 core and system memory per node). Use force=true to bypass.",
			currentCpuUsed, memoryLabel, memoryUsedForSafety/(1024*1024*1024),
			availableTargetCpu, availableTargetMem/(1024*1024*1024))
	}

	return nil
}

func allowTopologyDrivenWorkerScaleAcrossZero(sharedPool *entity.SharedPool, componentType apiv1.ComponentType, currentNodeCount, targetNodeCount uint32) bool {
	if sharedPool == nil || sharedPool.GetCapacityMode() != apiv1.CapacityMode_TOPOLOGY_DRIVEN {
		return false
	}

	if componentType != apiv1.ComponentType_TIDB_WORKER && componentType != apiv1.ComponentType_TIKV_WORKER {
		return false
	}

	return (currentNodeCount == 0 && targetNodeCount > 0) || (currentNodeCount > 0 && targetNodeCount == 0)
}