package region

import (
	"context"
	"encoding/json"
	"errors"
	"strconv"
	"strings"
	"time"

	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"google.golang.org/grpc/codes"
	"google.golang.org/protobuf/proto"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/adapter/regionalserver"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/resourcepoolpolicy"
	servicecommon "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/common"
	workflowcommon "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	utilworkflow "github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

func (s *service) GetByocSharedComponentScaleConstraints(
	ctx context.Context,
	req *globalngv1.GetByocSharedComponentScaleConstraintsRequest,
) (*globalngv1.ByocSharedComponentScaleConstraints, bizerr.BizError) {
	state, err := s.loadByocSharedPoolScaleState(ctx, req.GetResourcePoolId(), req.GetOrganizationId())
	if err != nil {
		return nil, err
	}

	preview, components := state.buildPreview(nil)
	if state.modifying {
		blockPreviewForModifyingPool(preview)
	}
	blockScalePreviewForPauseResumeLifecycle(preview, state.resourcePool.GetLifecycleState())
	applyConstraintBlockingReasonsToPreview(preview, components)

	return &globalngv1.ByocSharedComponentScaleConstraints{
		ResourcePoolId: req.GetResourcePoolId(),
		VcpusLimit:     state.resourcePool.GetVcpusLimit(),
		Preview:        preview,
		Components:     loMapConstraintComponents(components),
	}, nil
}

func (s *service) ScaleByocSharedComponents(
	ctx context.Context,
	req *globalngv1.ScaleByocSharedComponentsRequest,
) (*globalngv1.ScaleByocSharedComponentsResponse, bizerr.BizError) {
	state, err := s.loadByocSharedPoolScaleState(ctx, req.GetResourcePoolId(), req.GetOrganizationId())
	if err != nil {
		return nil, err
	}

	requestTargets := byocSharedComponentRequestTargets(req)
	if len(requestTargets) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "at least one shared component target is required")
	}
	if err := validateScaleRequestTargets(requestTargets); err != nil {
		return nil, err
	}
	preview, components := state.buildPreview(requestTargets)
	if state.modifying {
		blockPreviewForModifyingPool(preview)
	}
	lifecycleBlocked := blockScalePreviewForPauseResumeLifecycle(preview, state.resourcePool.GetLifecycleState())

	resp := &globalngv1.ScaleByocSharedComponentsResponse{
		Preview:    preview,
		Components: loMapPlanComponents(components),
	}
	if state.modifying {
		if req.GetValidateOnly() {
			return resp, nil
		}
		return nil, newScaleAlreadyInProgressError(req.GetResourcePoolId())
	}
	if lifecycleBlocked && req.GetValidateOnly() {
		return resp, nil
	}
	if !hasAnyEffectiveScaleChange(components) {
		return nil, bizerr.New(bizerr.ErrCodeParams, "scale request has no effective changes")
	}
	if lifecycleBlocked {
		return nil, newScaleBlockedByLifecycleError(state.resourcePool.GetLifecycleState())
	}
	if req.GetValidateOnly() {
		return resp, nil
	}
	if !preview.GetAllowed() {
		return nil, bizerr.New(bizerr.ErrCodeParams, "scale plan blocked: %s", strings.Join(preview.GetBlockingReasons(), "; "))
	}
	if s.workflow == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "workflow is not configured")
	}

	operationID, err := s.submitScaleWorkflow(ctx, state, components)
	if err != nil {
		return nil, err
	}
	resp.OperationId = operationID

	return resp, nil
}

type byocSharedPoolScaleState struct {
	resourcePool *entity.ResourcePool
	region       *entity.Region
	sharedPool   *clusterngv1beta.SharedPool
	constraints  *clusterngv1beta.GetSharedPoolScaleConstraintsResponse
	currentUsage uint64
	modifying    bool
	nextgenCli   regionalserver.Client
}

type byocSharedPoolPlanComponent struct {
	componentType        globalngv1.ByocSharedComponentType
	regionalType         clusterngv1beta.ComponentType
	componentID          string
	exists               bool
	phase                string
	current              *globalngv1.ByocSharedComponentScaleTarget
	target               *globalngv1.ByocSharedComponentScaleTarget
	requestTarget        *globalngv1.ByocSharedComponentScaleTarget
	requested            bool
	allowed              bool
	blockingReasons      []string
	constraintComponent  *clusterngv1beta.SharedPoolScaleConstraintComponent
	currentSpecCandidate *clusterngv1beta.SharedPoolScaleConstraintNodeSpecCandidate
	targetSpecCandidate  *clusterngv1beta.SharedPoolScaleConstraintNodeSpecCandidate
}

func blockScalePreviewForPauseResumeLifecycle(
	preview *globalngv1.ByocSharedComponentScalePreviewSummary,
	lifecycleState globalngv1.ResourcePool_LifecycleState,
) bool {
	if resourcepoolpolicy.Decide(resourcepoolpolicy.OperationScale, lifecycleState) !=
		resourcepoolpolicy.DecisionAllowed {
		if preview != nil {
			preview.Allowed = false
			preview.BlockingReasons = stableUniqueStrings(append(
				preview.GetBlockingReasons(),
				scaleBlockedByLifecycleReason(lifecycleState),
			))
		}
		return true
	}
	return false
}

func scaleBlockedByLifecycleReason(lifecycleState globalngv1.ResourcePool_LifecycleState) string {
	return "resource pool cannot be scaled while lifecycle state is " + lifecycleState.String()
}

func newScaleBlockedByLifecycleError(lifecycleState globalngv1.ResourcePool_LifecycleState) bizerr.BizError {
	return bizerr.New(
		bizerr.ErrCode(codes.FailedPrecondition),
		"resource pool cannot be scaled while lifecycle state is %s",
		lifecycleState.String(),
	)
}

func (s *service) loadByocSharedPoolScaleState(
	ctx context.Context,
	resourcePoolID uint64,
	organizationID uint64,
) (*byocSharedPoolScaleState, bizerr.BizError) {
	if resourcePoolID == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "resource_pool_id is required")
	}
	if organizationID == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "organization_id is required")
	}

	resourcePool, err := s.resourcePoolRepo.Load(ctx, resourcePoolID, false)
	if err != nil {
		return nil, err
	}
	if resourcePool.GetOwnerType() != globalngv1.ResourcePool_ORGANIZATION || resourcePool.GetOwnerId() != organizationID {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "resource pool not found")
	}
	if resourcePool.GetServicePlan() != globalcommonv1.ServicePlan_BYOC {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "resource pool not found")
	}

	region, err := s.regionRepo.Get(ctx, resourcePool.RegionId)
	if err != nil {
		return nil, err
	}
	sharedPoolID := strconv.FormatUint(resourcePoolID, 10)
	sharedPool, err := s.nextgenCli.GetSharedPool(ctx, region, &clusterngv1beta.GetSharedPoolRequest{
		SharedPoolId: sharedPoolID,
	})
	if err != nil {
		return nil, err
	}
	modifying := sharedPool.GetPhase() == clusterngv1beta.SharedPool_MODIFYING
	hasRunningWorkflow, err := s.hasRunningScaleWorkflow(ctx, resourcePoolID)
	if err != nil {
		return nil, err
	}
	modifying = modifying || hasRunningWorkflow
	constraints, err := s.nextgenCli.GetSharedPoolScaleConstraints(ctx, region, &clusterngv1beta.GetSharedPoolScaleConstraintsRequest{
		SharedPoolId: sharedPoolID,
	})
	if err != nil {
		return nil, err
	}
	usage, err := s.nextgenCli.GetSharedPoolVcpusUsage(ctx, region, &clusterngv1beta.GetSharedPoolVcpusUsageRequest{
		SharedPoolId: sharedPoolID,
	})
	if err != nil {
		return nil, err
	}

	return &byocSharedPoolScaleState{
		resourcePool: resourcePool,
		region:       region,
		sharedPool:   sharedPool,
		constraints:  constraints,
		currentUsage: usage.GetVcpusUsed(),
		modifying:    modifying,
		nextgenCli:   s.nextgenCli,
	}, nil
}

func (s *byocSharedPoolScaleState) buildPreview(
	requestTargets map[globalngv1.ByocSharedComponentType]*globalngv1.ByocSharedComponentScaleTarget,
) (*globalngv1.ByocSharedComponentScalePreviewSummary, []*byocSharedPoolPlanComponent) {
	plan := make([]*byocSharedPoolPlanComponent, 0, 3)
	scalingVcpusUsed := s.currentUsage
	blockingReasons := make([]string, 0)

	for _, item := range supportedSharedComponentTypes() {
		currentComponent := findSharedPoolComponent(s.sharedPool, item.regionalType)
		currentTarget := convertSharedPoolComponentToTarget(currentComponent)
		requestTarget := requestTargets[item.publicType]
		requested := requestTarget != nil
		target := mergeScaleTarget(currentTarget, requestTarget)
		if !requested {
			target = cloneScaleTarget(currentTarget)
		}

		component := &byocSharedPoolPlanComponent{
			componentType:       item.publicType,
			regionalType:        item.regionalType,
			current:             currentTarget,
			target:              target,
			requestTarget:       requestTarget,
			requested:           requested,
			allowed:             true,
			constraintComponent: findConstraintComponent(s.constraints, item.regionalType),
		}
		if currentComponent != nil {
			component.exists = true
			component.componentID = currentComponent.GetId()
			component.phase = currentComponent.GetPhase().String()
		}

		component.currentSpecCandidate = selectCurrentSpecCandidate(component.constraintComponent, currentTarget)
		component.targetSpecCandidate = selectTargetSpecCandidate(component.constraintComponent, target)
		for _, reason := range validatePlanComponent(component) {
			component.allowed = false
			component.blockingReasons = append(component.blockingReasons, reason)
			blockingReasons = append(blockingReasons, reason)
		}
		if requested {
			var ok bool
			scalingVcpusUsed, ok = applyComponentVcpusDelta(
				scalingVcpusUsed,
				estimateTargetVcpus(currentTarget),
				estimateTargetVcpus(target),
			)
			if !ok {
				blockingReasons = append(blockingReasons, workflowcommon.ByocSharedComponentScaleUsageMismatchReason)
			}
		}
		plan = append(plan, component)
	}

	allowed := len(blockingReasons) == 0
	if scalingVcpusUsed < s.currentUsage && !isScaleInAllowedByPlan(plan) {
		allowed = false
		blockingReasons = append(blockingReasons, workflowcommon.ByocSharedComponentScaleInBlockingReason)
	}
	if s.resourcePool.GetVcpusLimit() > 0 && scalingVcpusUsed > s.resourcePool.GetVcpusLimit() {
		allowed = false
		blockingReasons = append(blockingReasons, workflowcommon.ByocSharedComponentScaleVcpusLimitBlockingReason)
	}

	return &globalngv1.ByocSharedComponentScalePreviewSummary{
		CurrentVcpusUsed: s.currentUsage,
		ScalingVcpusUsed: scalingVcpusUsed,
		MaxVcpusLimit:    s.resourcePool.GetVcpusLimit(),
		Allowed:          allowed,
		BlockingReasons:  stableUniqueStrings(blockingReasons),
	}, plan
}

func (s *service) submitScaleWorkflow(
	ctx context.Context,
	state *byocSharedPoolScaleState,
	components []*byocSharedPoolPlanComponent,
) (string, bizerr.BizError) {
	if s.transaction == nil {
		return "", bizerr.New(bizerr.ErrCodeUnknown, "transaction is not configured")
	}

	var operationID string
	if err := s.transaction.Transaction(ctx, func(txCtx context.Context) error {
		// The row lock serializes scale admission with Pause/Resume lifecycle changes
		// after preview/state loading.
		// Regional reads and preview construction stay outside this short transaction;
		// action-level revalidation is responsible for catching stale component state.
		lockedResourcePool, err := s.resourcePoolRepo.LoadForUpdate(txCtx, state.resourcePool.GetId(), false)
		if err != nil {
			return err
		}
		hasRunningWorkflow, err := s.hasRunningScaleWorkflow(txCtx, state.resourcePool.GetId())
		if err != nil {
			return err
		}
		if hasRunningWorkflow {
			return newScaleAlreadyInProgressError(state.resourcePool.GetId())
		}
		if lifecycleErr := servicecommon.RequireResourcePoolLifecycle(
			lockedResourcePool,
			resourcepoolpolicy.OperationScale,
		); lifecycleErr != nil {
			return lifecycleErr
		}
		if err := s.resourcePoolRepo.UpdateLifecycle(
			txCtx,
			state.resourcePool.GetId(),
			globalngv1.ResourcePool_MODIFYING,
			"",
		); err != nil {
			return err
		}

		var bizErr bizerr.BizError
		operationID, bizErr = s.createScaleWorkflow(txCtx, state, components)
		if bizErr != nil {
			return bizErr
		}
		return nil
	}); err != nil {
		var bizErr bizerr.BizError
		if errors.As(err, &bizErr) {
			return "", bizErr
		}
		return "", bizerr.WrapError(err, "submit scale workflow failed")
	}

	return operationID, nil
}

func (s *service) createScaleWorkflow(
	ctx context.Context,
	state *byocSharedPoolScaleState,
	components []*byocSharedPoolPlanComponent,
) (string, bizerr.BizError) {
	operationID := s.newScaleOperationID(ctx)
	payload := workflowcommon.ByocSharedComponentScaleWorkflowPayload{
		ResourcePoolID: state.resourcePool.GetId(),
		Components:     make([]workflowcommon.ByocSharedComponentScaleWorkflowComponent, 0, len(components)),
	}
	for _, component := range components {
		if !component.requested {
			continue
		}
		payload.Components = append(payload.Components, workflowcommon.ByocSharedComponentScaleWorkflowComponent{
			ComponentType: component.componentType.String(),
			ComponentID:   component.componentID,
			Current:       cloneScaleTarget(component.current),
			Target:        component.target,
		})
	}

	data, err := json.Marshal(payload)
	if err != nil {
		return "", bizerr.WrapError(err, "marshal workflow payload failed")
	}
	metadataJSON, metaErr := metadata.MarshalRequestMetaFromCtx(ctx)
	if metaErr != nil {
		metadataJSON = nil
	}

	orgID := strconv.FormatUint(state.resourcePool.GetOwnerId(), 10)
	bizID := strconv.FormatUint(state.resourcePool.GetId(), 10)
	if err := s.workflow.RunDagWithBizId(
		ctx,
		workflowcommon.ScaleByocSharedComponents,
		orgID,
		"0",
		bizID,
		map[string]string{
			utilworkflow.KeyTraceId:                utilworkflow.GetTraceId(ctx),
			utilworkflow.KeyRequestMeta:            string(metadataJSON),
			workflowcommon.KeyResourcePoolId:       bizID,
			workflowcommon.KeyOperationId:          operationID,
			workflowcommon.ScaleTaskKeyRequestBody: string(data),
		},
	); err != nil {
		return "", bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "create async workflow failed")
	}

	return operationID, nil
}

func byocSharedComponentRequestTargets(
	req *globalngv1.ScaleByocSharedComponentsRequest,
) map[globalngv1.ByocSharedComponentType]*globalngv1.ByocSharedComponentScaleTarget {
	targets := make(map[globalngv1.ByocSharedComponentType]*globalngv1.ByocSharedComponentScaleTarget, 3)
	if target := req.GetTikv(); target != nil {
		targets[globalngv1.ByocSharedComponentType_BYOC_SHARED_COMPONENT_TYPE_TIKV] = target
	}
	if target := req.GetTiflashWrite(); target != nil {
		targets[globalngv1.ByocSharedComponentType_BYOC_SHARED_COMPONENT_TYPE_TIFLASH_WRITE] = target
	}
	if target := req.GetTiflashCompute(); target != nil {
		targets[globalngv1.ByocSharedComponentType_BYOC_SHARED_COMPONENT_TYPE_TIFLASH_COMPUTE] = target
	}
	return targets
}

func validateScaleRequestTargets(
	targets map[globalngv1.ByocSharedComponentType]*globalngv1.ByocSharedComponentScaleTarget,
) bizerr.BizError {
	for _, target := range targets {
		seenStorageNames := make(map[string]struct{}, len(target.GetStorages()))
		for _, storage := range target.GetStorages() {
			name := strings.TrimSpace(storage.GetStorageName())
			if name == "" {
				return bizerr.New(bizerr.ErrCodeParams, "storage_name is required")
			}
			key := strings.ToUpper(name)
			if _, ok := seenStorageNames[key]; ok {
				return bizerr.New(bizerr.ErrCodeParams, "duplicate storage_name %s", name)
			}
			seenStorageNames[key] = struct{}{}
		}
	}
	return nil
}

func supportedSharedComponentTypes() []struct {
	publicType   globalngv1.ByocSharedComponentType
	regionalType clusterngv1beta.ComponentType
} {
	return []struct {
		publicType   globalngv1.ByocSharedComponentType
		regionalType clusterngv1beta.ComponentType
	}{
		{globalngv1.ByocSharedComponentType_BYOC_SHARED_COMPONENT_TYPE_TIKV, clusterngv1beta.ComponentType_TIKV},
		{globalngv1.ByocSharedComponentType_BYOC_SHARED_COMPONENT_TYPE_TIFLASH_WRITE, clusterngv1beta.ComponentType_TIFLASH_WRITE},
		{globalngv1.ByocSharedComponentType_BYOC_SHARED_COMPONENT_TYPE_TIFLASH_COMPUTE, clusterngv1beta.ComponentType_TIFLASH_COMPUTE},
	}
}

func validatePlanComponent(component *byocSharedPoolPlanComponent) []string {
	if !component.requested {
		return nil
	}

	reasons := provisioningBlockingReasons(component)
	reasons = append(reasons, constraintReasonsToMessages(component.constraintComponent.GetBlockingReasons())...)
	reasons = append(reasons, validateNodeCount(component)...)
	reasons = append(reasons, validateNodeSpec(component)...)
	reasons = append(reasons, validateComponentVcpus(component)...)
	reasons = append(reasons, validateStorages(component)...)

	return stableUniqueStrings(reasons)
}

func validateNodeCount(component *byocSharedPoolPlanComponent) []string {
	if component.target == nil || component.target.NodeCount == nil {
		if component.current == nil {
			return []string{"node_count is required"}
		}
		return nil
	}

	constraints := component.constraintComponent.GetNodeCountConstraints()
	if constraints == nil {
		return nil
	}

	targetCount := component.target.GetNodeCount()
	reasons := make([]string, 0)
	if isTiflashScaleToZeroIntent(component) &&
		scaleToZeroRequestChangesCapacity(component.current, component.requestTarget) {
		reasons = append(reasons, "tiflash scale-to-zero only supports node_count=0")
	}
	if component.componentType == globalngv1.ByocSharedComponentType_BYOC_SHARED_COMPONENT_TYPE_TIKV &&
		component.current != nil &&
		component.current.NodeCount != nil &&
		component.current.GetNodeCount() == 0 &&
		targetCount > 0 {
		reasons = append(reasons, "tikv scale-from-zero is not supported")
	}
	if component.current != nil &&
		component.current.NodeCount != nil &&
		targetCount < component.current.GetNodeCount() &&
		!isTiflashScaleToZeroIntent(component) {
		reasons = append(reasons, "scale-in for node_count is not supported in phase 1")
	}
	if targetCount < constraints.GetMin() || targetCount > constraints.GetMax() {
		reasons = append(reasons, "target node_count is outside the allowed range")
	}
	if !constraints.GetAllowCrossZero() {
		current := constraints.GetCurrent()
		if (current == 0 && targetCount > 0) || (current > 0 && targetCount == 0) {
			reasons = append(reasons, "cross-zero scaling is not allowed")
		}
	}
	if rule := constraints.GetRule(); rule != nil {
		switch rule.GetType() {
		case "MULTIPLE_OF":
			if value := rule.GetValue(); value > 0 && targetCount%value != 0 {
				reasons = append(reasons, "target node_count violates the multiple-of rule")
			}
		case "FIXED":
			if value := rule.GetValue(); value > 0 && targetCount != value {
				reasons = append(reasons, "target node_count violates the fixed-count rule")
			}
		}
	}

	return reasons
}

func validateNodeSpec(component *byocSharedPoolPlanComponent) []string {
	if component.target == nil {
		return nil
	}
	if component.componentType == globalngv1.ByocSharedComponentType_BYOC_SHARED_COMPONENT_TYPE_TIKV &&
		component.requestTarget != nil &&
		nodeSpecRequestChanges(component.current, component.requestTarget) {
		return []string{"tikv node spec changes are not supported"}
	}
	if component.current != nil {
		if component.current.MemoryGi != nil && component.target.MemoryGi != nil && component.target.GetMemoryGi() < component.current.GetMemoryGi() {
			return []string{"scale-in for memory_gi is not supported in phase 1"}
		}
	}
	if component.targetSpecCandidate == nil {
		if component.target.Cpu == nil && component.target.MemoryGi == nil && component.current != nil {
			return nil
		}
		return []string{"target node spec is not supported"}
	}
	if !component.targetSpecCandidate.GetAvailable() && !component.targetSpecCandidate.GetIsCurrent() {
		reasons := constraintReasonsToMessages(component.targetSpecCandidate.GetReasons())
		if len(reasons) == 0 {
			return []string{"target node spec is not available"}
		}
		return reasons
	}
	return nil
}

func nodeSpecRequestChanges(
	current *globalngv1.ByocSharedComponentScaleTarget,
	requested *globalngv1.ByocSharedComponentScaleTarget,
) bool {
	if requested == nil {
		return false
	}
	if current == nil {
		return requested.Cpu != nil || requested.MemoryGi != nil
	}
	return requested.Cpu != nil && requested.GetCpu() != current.GetCpu() ||
		requested.MemoryGi != nil && requested.GetMemoryGi() != current.GetMemoryGi()
}

func validateComponentVcpus(component *byocSharedPoolPlanComponent) []string {
	if component == nil || component.current == nil || component.target == nil {
		return nil
	}
	if estimateTargetVcpus(component.target) < estimateTargetVcpus(component.current) &&
		!isTiflashScaleToZeroIntent(component) {
		return []string{"scale-in for component vcpus is not supported in phase 1"}
	}
	return nil
}

func isScaleInAllowedByPlan(components []*byocSharedPoolPlanComponent) bool {
	hasVcpuDecrease := false
	for _, component := range components {
		if component == nil || !component.requested {
			continue
		}
		if estimateTargetVcpus(component.target) >= estimateTargetVcpus(component.current) {
			continue
		}
		hasVcpuDecrease = true
		if !isTiflashScaleToZeroIntent(component) {
			return false
		}
	}
	return hasVcpuDecrease
}

func isTiflashScaleToZeroIntent(component *byocSharedPoolPlanComponent) bool {
	if component == nil || !isTiflashSharedComponent(component.componentType) {
		return false
	}
	if component.current == nil || component.current.NodeCount == nil || component.current.GetNodeCount() == 0 {
		return false
	}
	if component.requestTarget == nil || component.requestTarget.NodeCount == nil || component.requestTarget.GetNodeCount() != 0 {
		return false
	}
	return component.target != nil && component.target.NodeCount != nil && component.target.GetNodeCount() == 0
}

func isTiflashSharedComponent(componentType globalngv1.ByocSharedComponentType) bool {
	return componentType == globalngv1.ByocSharedComponentType_BYOC_SHARED_COMPONENT_TYPE_TIFLASH_WRITE ||
		componentType == globalngv1.ByocSharedComponentType_BYOC_SHARED_COMPONENT_TYPE_TIFLASH_COMPUTE
}

func scaleToZeroRequestChangesCapacity(
	current *globalngv1.ByocSharedComponentScaleTarget,
	requested *globalngv1.ByocSharedComponentScaleTarget,
) bool {
	if requested == nil {
		return false
	}
	if nodeSpecRequestChanges(current, requested) {
		return true
	}
	currentStorages := make(map[string]*globalngv1.ByocSharedComponentStorage, len(current.GetStorages()))
	for _, storage := range current.GetStorages() {
		currentStorages[strings.ToUpper(storage.GetStorageName())] = storage
	}
	for _, storage := range requested.GetStorages() {
		if storageRequestChanges(currentStorages[strings.ToUpper(storage.GetStorageName())], storage) {
			return true
		}
	}
	return false
}

func validateStorages(component *byocSharedPoolPlanComponent) []string {
	if component.target == nil || len(component.target.GetStorages()) == 0 {
		return nil
	}
	candidate := component.targetSpecCandidate
	if candidate == nil {
		return []string{"target storage constraints are unavailable"}
	}

	currentByName := make(map[string]*globalngv1.ByocSharedComponentStorage, len(component.current.GetStorages()))
	if component.current != nil {
		for _, storage := range component.current.GetStorages() {
			currentByName[strings.ToUpper(storage.GetStorageName())] = storage
		}
	}
	constraintsByName := make(map[string]*clusterngv1beta.SharedPoolScaleConstraintStorageConstraint, len(candidate.GetStorageConstraints()))
	for _, storage := range candidate.GetStorageConstraints() {
		constraintsByName[storage.GetName().String()] = storage
	}

	reasons := make([]string, 0)
	for _, storage := range component.target.GetStorages() {
		name := strings.ToUpper(storage.GetStorageName())
		requestedStorage := findRequestedStorage(component.requestTarget, name)
		if component.componentType == globalngv1.ByocSharedComponentType_BYOC_SHARED_COMPONENT_TYPE_TIKV &&
			requestedStorage != nil &&
			name != clusterngv1beta.Component_Storage_DATA.String() &&
			storageRequestChanges(currentByName[name], requestedStorage) {
			reasons = append(reasons, "tikv only supports resizing DATA storage")
			continue
		}
		constraint := constraintsByName[name]
		if constraint == nil {
			reasons = append(reasons, "storage "+name+" is not supported")
			continue
		}
		current := currentByName[name]
		targetSize := constraint.GetCurrentSizeGi()
		if storage.SizeGi != nil {
			targetSize = storage.GetSizeGi()
		} else if current != nil && current.SizeGi != nil {
			targetSize = current.GetSizeGi()
		}
		if current != nil && current.SizeGi != nil && storage.SizeGi != nil && storage.GetSizeGi() < current.GetSizeGi() {
			reasons = append(reasons, "scale-in for storage "+name+" size is not supported in phase 1")
		}
		if !constraint.GetResizable() && targetSize != constraint.GetCurrentSizeGi() {
			reasons = append(reasons, "storage "+name+" is not resizable")
		}
		if constraint.MaxSizeGi != nil && targetSize > constraint.GetMaxSizeGi() {
			reasons = append(reasons, "storage "+name+" exceeds the max allowed size")
		}
		if requestedStorage != nil && requestedStorage.Iops != nil &&
			(current == nil || requestedStorage.GetIops() != current.GetIops()) {
			reasons = append(reasons, "storage "+name+" iops changes are not supported")
		}
		if requestedStorage != nil && requestedStorage.ThroughputMbs != nil &&
			(current == nil || requestedStorage.GetThroughputMbs() != current.GetThroughputMbs()) {
			reasons = append(reasons, "storage "+name+" throughput_mbs changes are not supported")
		}
	}

	return reasons
}

func storageRequestChanges(
	current *globalngv1.ByocSharedComponentStorage,
	requested *globalngv1.ByocSharedComponentStorage,
) bool {
	if requested == nil {
		return false
	}
	if current == nil {
		return requested.SizeGi != nil || requested.Iops != nil || requested.ThroughputMbs != nil
	}
	return requested.SizeGi != nil && requested.GetSizeGi() != current.GetSizeGi() ||
		requested.Iops != nil && requested.GetIops() != current.GetIops() ||
		requested.ThroughputMbs != nil && requested.GetThroughputMbs() != current.GetThroughputMbs()
}

func findRequestedStorage(
	target *globalngv1.ByocSharedComponentScaleTarget,
	storageName string,
) *globalngv1.ByocSharedComponentStorage {
	if target == nil {
		return nil
	}
	for _, storage := range target.GetStorages() {
		if strings.EqualFold(storage.GetStorageName(), storageName) {
			return storage
		}
	}
	return nil
}

func findSharedPoolComponent(sharedPool *clusterngv1beta.SharedPool, componentType clusterngv1beta.ComponentType) *clusterngv1beta.Component {
	if sharedPool == nil {
		return nil
	}
	for _, component := range sharedPool.GetComponentGroups() {
		if component.GetComponentType() == componentType {
			return component
		}
	}
	return nil
}

func findConstraintComponent(
	constraints *clusterngv1beta.GetSharedPoolScaleConstraintsResponse,
	componentType clusterngv1beta.ComponentType,
) *clusterngv1beta.SharedPoolScaleConstraintComponent {
	if constraints == nil {
		return nil
	}
	for _, component := range constraints.GetComponents() {
		if component.GetComponentType() == componentType {
			return component
		}
	}
	return nil
}

func convertSharedPoolComponentToTarget(component *clusterngv1beta.Component) *globalngv1.ByocSharedComponentScaleTarget {
	if component == nil {
		return nil
	}
	target := &globalngv1.ByocSharedComponentScaleTarget{
		Storages: make([]*globalngv1.ByocSharedComponentStorage, 0, len(component.GetStorages())),
	}
	if quota := component.GetNodeQuota(); quota != nil {
		target.NodeCount = proto.Uint32(quota.GetNodeCount())
		if spec := quota.GetNodeSpec(); spec != nil {
			target.Cpu = proto.Uint32(spec.GetCpu())
			target.MemoryGi = proto.Uint32(spec.GetMemoryGi())
		}
	}
	for _, storage := range component.GetStorages() {
		item := &globalngv1.ByocSharedComponentStorage{
			StorageName: storage.GetName().String(),
		}
		item.SizeGi = proto.Uint32(storage.GetSizeGi())
		if storage.Iops != 0 {
			item.Iops = proto.Uint32(storage.GetIops())
		}
		if storage.ThroughputMbs != 0 {
			item.ThroughputMbs = proto.Uint32(storage.GetThroughputMbs())
		}
		target.Storages = append(target.Storages, item)
	}
	if target.NodeCount == nil && target.Cpu == nil && target.MemoryGi == nil && len(target.Storages) == 0 {
		return nil
	}
	return target
}

func cloneScaleTarget(in *globalngv1.ByocSharedComponentScaleTarget) *globalngv1.ByocSharedComponentScaleTarget {
	if in == nil {
		return nil
	}
	out := &globalngv1.ByocSharedComponentScaleTarget{
		Storages: make([]*globalngv1.ByocSharedComponentStorage, 0, len(in.GetStorages())),
	}
	if in.NodeCount != nil {
		out.NodeCount = proto.Uint32(in.GetNodeCount())
	}
	if in.Cpu != nil {
		out.Cpu = proto.Uint32(in.GetCpu())
	}
	if in.MemoryGi != nil {
		out.MemoryGi = proto.Uint32(in.GetMemoryGi())
	}
	for _, storage := range in.GetStorages() {
		item := &globalngv1.ByocSharedComponentStorage{
			StorageName: storage.GetStorageName(),
		}
		if storage.SizeGi != nil {
			item.SizeGi = proto.Uint32(storage.GetSizeGi())
		}
		if storage.Iops != nil {
			item.Iops = proto.Uint32(storage.GetIops())
		}
		if storage.ThroughputMbs != nil {
			item.ThroughputMbs = proto.Uint32(storage.GetThroughputMbs())
		}
		out.Storages = append(out.Storages, item)
	}
	return out
}

func mergeScaleTarget(current, requested *globalngv1.ByocSharedComponentScaleTarget) *globalngv1.ByocSharedComponentScaleTarget {
	if requested == nil {
		return cloneScaleTarget(current)
	}

	out := cloneScaleTarget(current)
	if out == nil {
		out = &globalngv1.ByocSharedComponentScaleTarget{}
	}
	if requested.NodeCount != nil {
		out.NodeCount = proto.Uint32(requested.GetNodeCount())
	}
	if requested.Cpu != nil {
		out.Cpu = proto.Uint32(requested.GetCpu())
	}
	if requested.MemoryGi != nil {
		out.MemoryGi = proto.Uint32(requested.GetMemoryGi())
	}
	if len(requested.GetStorages()) > 0 {
		currentStorages := make(map[string]*globalngv1.ByocSharedComponentStorage, len(out.GetStorages()))
		for _, storage := range out.GetStorages() {
			currentStorages[strings.ToUpper(storage.GetStorageName())] = storage
		}

		merged := make([]*globalngv1.ByocSharedComponentStorage, 0, len(currentStorages)+len(requested.GetStorages()))
		seen := make(map[string]bool, len(requested.GetStorages()))
		for _, requestedStorage := range requested.GetStorages() {
			name := strings.ToUpper(requestedStorage.GetStorageName())
			base := cloneStorage(currentStorages[name])
			if base == nil {
				base = &globalngv1.ByocSharedComponentStorage{StorageName: requestedStorage.GetStorageName()}
			}
			if requestedStorage.SizeGi != nil {
				base.SizeGi = proto.Uint32(requestedStorage.GetSizeGi())
			}
			if requestedStorage.Iops != nil {
				base.Iops = proto.Uint32(requestedStorage.GetIops())
			}
			if requestedStorage.ThroughputMbs != nil {
				base.ThroughputMbs = proto.Uint32(requestedStorage.GetThroughputMbs())
			}
			merged = append(merged, base)
			seen[name] = true
		}
		for _, storage := range out.GetStorages() {
			if !seen[strings.ToUpper(storage.GetStorageName())] {
				merged = append(merged, cloneStorage(storage))
			}
		}
		out.Storages = merged
	}

	return out
}

func cloneStorage(in *globalngv1.ByocSharedComponentStorage) *globalngv1.ByocSharedComponentStorage {
	if in == nil {
		return nil
	}
	out := &globalngv1.ByocSharedComponentStorage{
		StorageName: in.GetStorageName(),
	}
	if in.SizeGi != nil {
		out.SizeGi = proto.Uint32(in.GetSizeGi())
	}
	if in.Iops != nil {
		out.Iops = proto.Uint32(in.GetIops())
	}
	if in.ThroughputMbs != nil {
		out.ThroughputMbs = proto.Uint32(in.GetThroughputMbs())
	}
	return out
}

func estimateTargetVcpus(target *globalngv1.ByocSharedComponentScaleTarget) uint64 {
	if target == nil || target.NodeCount == nil || target.Cpu == nil {
		return 0
	}
	return uint64(target.GetNodeCount()) * uint64(target.GetCpu())
}

func selectCurrentSpecCandidate(
	component *clusterngv1beta.SharedPoolScaleConstraintComponent,
	current *globalngv1.ByocSharedComponentScaleTarget,
) *clusterngv1beta.SharedPoolScaleConstraintNodeSpecCandidate {
	if component == nil || current == nil {
		return nil
	}
	return selectCandidateBySpec(component.GetNodeSpecConstraints(), current.GetCpu(), current.GetMemoryGi())
}

func selectTargetSpecCandidate(
	component *clusterngv1beta.SharedPoolScaleConstraintComponent,
	target *globalngv1.ByocSharedComponentScaleTarget,
) *clusterngv1beta.SharedPoolScaleConstraintNodeSpecCandidate {
	if component == nil || target == nil || target.Cpu == nil || target.MemoryGi == nil {
		return nil
	}
	return selectCandidateBySpec(component.GetNodeSpecConstraints(), target.GetCpu(), target.GetMemoryGi())
}

func selectCandidateBySpec(
	constraints *clusterngv1beta.SharedPoolScaleConstraintNodeSpecConstraints,
	cpu, memoryGi uint32,
) *clusterngv1beta.SharedPoolScaleConstraintNodeSpecCandidate {
	if constraints == nil {
		return nil
	}
	for _, candidate := range constraints.GetCandidates() {
		if candidate.GetCpu() == cpu && candidate.GetMemoryGi() == memoryGi {
			return candidate
		}
	}
	return nil
}

func constraintReasonsToMessages(reasons []*clusterngv1beta.SharedPoolScaleConstraintReason) []string {
	out := make([]string, 0, len(reasons))
	for _, reason := range reasons {
		if msg := strings.TrimSpace(reason.GetMessage()); msg != "" {
			out = append(out, msg)
		}
	}
	return out
}

func stableUniqueStrings(in []string) []string {
	if len(in) == 0 {
		return nil
	}
	out := make([]string, 0, len(in))
	seen := make(map[string]bool, len(in))
	for _, item := range in {
		item = strings.TrimSpace(item)
		if item == "" || seen[item] {
			continue
		}
		seen[item] = true
		out = append(out, item)
	}
	return out
}

func hasAnyEffectiveScaleChange(components []*byocSharedPoolPlanComponent) bool {
	for _, component := range components {
		if !component.requested {
			continue
		}
		if !proto.Equal(component.current, component.target) {
			return true
		}
	}
	return false
}

func (s *service) newScaleOperationID(ctx context.Context) string {
	if s.idGeneratorAdapter != nil {
		return strconv.FormatUint(s.idGeneratorAdapter.MustNew(), 10)
	}
	if traceID := strings.TrimSpace(utilworkflow.GetTraceId(ctx)); traceID != "" {
		return traceID
	}
	return strconv.FormatInt(time.Now().UnixNano(), 10)
}

func applyComponentVcpusDelta(total, current, target uint64) (uint64, bool) {
	if total < current {
		return total, false
	}
	return total - current + target, true
}

func applyConstraintBlockingReasonsToPreview(
	preview *globalngv1.ByocSharedComponentScalePreviewSummary,
	components []*byocSharedPoolPlanComponent,
) {
	if preview == nil {
		return
	}

	blockingReasons := append([]string{}, preview.GetBlockingReasons()...)
	for _, component := range components {
		for _, reason := range constraintBlockingReasons(component) {
			if msg := strings.TrimSpace(reason.GetMessage()); msg != "" {
				blockingReasons = append(blockingReasons, msg)
			}
		}
	}
	blockingReasons = stableUniqueStrings(blockingReasons)
	if len(blockingReasons) == 0 {
		return
	}
	preview.Allowed = false
	preview.BlockingReasons = blockingReasons
}

func loMapConstraintComponents(in []*byocSharedPoolPlanComponent) []*globalngv1.ByocSharedComponentScaleConstraint {
	components := appendMissingConstraintComponents(in)
	out := make([]*globalngv1.ByocSharedComponentScaleConstraint, 0, len(components))
	for _, component := range components {
		out = append(out, &globalngv1.ByocSharedComponentScaleConstraint{
			ComponentType:        component.componentType,
			ComponentId:          component.componentID,
			Exists:               component.exists,
			Phase:                component.phase,
			Current:              component.current,
			BlockingReasons:      convertConstraintReasonsToGlobal(constraintBlockingReasons(component)),
			NodeCountConstraints: convertNodeCountConstraintsToGlobal(component.constraintComponent.GetNodeCountConstraints()),
			NodeSpecConstraints:  convertNodeSpecConstraintsToGlobal(component.constraintComponent.GetNodeSpecConstraints()),
		})
	}
	return out
}

func appendMissingConstraintComponents(in []*byocSharedPoolPlanComponent) []*byocSharedPoolPlanComponent {
	out := make([]*byocSharedPoolPlanComponent, 0, len(in)+3)
	seen := make(map[globalngv1.ByocSharedComponentType]struct{}, len(in))
	for _, component := range in {
		if component == nil {
			continue
		}
		out = append(out, component)
		seen[component.componentType] = struct{}{}
	}
	for _, item := range supportedSharedComponentTypes() {
		if _, ok := seen[item.publicType]; ok {
			continue
		}
		component := &byocSharedPoolPlanComponent{
			componentType: item.publicType,
			regionalType:  item.regionalType,
			allowed:       true,
		}
		out = append(out, component)
	}
	return out
}

func constraintBlockingReasons(component *byocSharedPoolPlanComponent) []*clusterngv1beta.SharedPoolScaleConstraintReason {
	reasons := make([]*clusterngv1beta.SharedPoolScaleConstraintReason, 0)
	if component.constraintComponent != nil {
		reasons = append(reasons, component.constraintComponent.GetBlockingReasons()...)
	}
	for _, reason := range provisioningBlockingReasons(component) {
		reasons = append(reasons, &clusterngv1beta.SharedPoolScaleConstraintReason{
			Message: reason,
		})
	}
	for _, reason := range component.blockingReasons {
		reasons = append(reasons, &clusterngv1beta.SharedPoolScaleConstraintReason{
			Message: reason,
		})
	}
	return reasons
}

func provisioningBlockingReasons(component *byocSharedPoolPlanComponent) []string {
	if component == nil {
		return nil
	}

	componentID := strings.TrimSpace(component.componentID)
	if componentID == "" {
		switch component.componentType {
		case globalngv1.ByocSharedComponentType_BYOC_SHARED_COMPONENT_TYPE_TIKV:
			return []string{"tikv scale-from-zero is not supported"}
		case globalngv1.ByocSharedComponentType_BYOC_SHARED_COMPONENT_TYPE_TIFLASH_WRITE:
			return []string{"tiflash_write component is not provisioned in the shared pool"}
		case globalngv1.ByocSharedComponentType_BYOC_SHARED_COMPONENT_TYPE_TIFLASH_COMPUTE:
			return []string{"tiflash_compute component is not provisioned in the shared pool"}
		default:
			return []string{"target component is not provisioned in the shared pool"}
		}
	}

	if component.componentType == globalngv1.ByocSharedComponentType_BYOC_SHARED_COMPONENT_TYPE_TIKV &&
		component.current != nil &&
		component.current.NodeCount != nil &&
		component.current.GetNodeCount() == 0 {
		return []string{"tikv scale-from-zero is not supported"}
	}

	return nil
}

func loMapPlanComponents(in []*byocSharedPoolPlanComponent) []*globalngv1.ByocSharedComponentScalePlanComponent {
	out := make([]*globalngv1.ByocSharedComponentScalePlanComponent, 0, len(in))
	for _, component := range in {
		out = append(out, &globalngv1.ByocSharedComponentScalePlanComponent{
			ComponentType:   component.componentType,
			ComponentId:     component.componentID,
			Current:         component.current,
			Target:          component.target,
			Allowed:         component.allowed,
			BlockingReasons: component.blockingReasons,
		})
	}
	return out
}

func convertConstraintReasonsToGlobal(
	in []*clusterngv1beta.SharedPoolScaleConstraintReason,
) []*globalngv1.ByocSharedComponentScaleConstraintReason {
	out := make([]*globalngv1.ByocSharedComponentScaleConstraintReason, 0, len(in))
	for _, item := range in {
		out = append(out, &globalngv1.ByocSharedComponentScaleConstraintReason{
			Code:    item.GetCode(),
			Message: item.GetMessage(),
		})
	}
	return out
}

func convertNodeCountConstraintsToGlobal(
	in *clusterngv1beta.SharedPoolScaleConstraintNodeCountConstraints,
) *globalngv1.ByocSharedComponentScaleNodeCountConstraints {
	if in == nil {
		return nil
	}
	return &globalngv1.ByocSharedComponentScaleNodeCountConstraints{
		Current:        in.GetCurrent(),
		Min:            in.GetMin(),
		Max:            in.GetMax(),
		Rule:           convertNodeCountRuleToGlobal(in.GetRule()),
		AllowCrossZero: in.GetAllowCrossZero(),
	}
}

func convertNodeCountRuleToGlobal(
	in *clusterngv1beta.SharedPoolScaleConstraintNodeCountRule,
) *globalngv1.ByocSharedComponentScaleNodeCountRule {
	if in == nil {
		return nil
	}
	return &globalngv1.ByocSharedComponentScaleNodeCountRule{
		Type:  in.GetType(),
		Value: in.GetValue(),
	}
}

func convertNodeSpecConstraintsToGlobal(
	in *clusterngv1beta.SharedPoolScaleConstraintNodeSpecConstraints,
) *globalngv1.ByocSharedComponentScaleNodeSpecConstraints {
	if in == nil {
		return nil
	}
	out := &globalngv1.ByocSharedComponentScaleNodeSpecConstraints{
		Current: in.GetCurrent(),
	}
	for _, candidate := range in.GetCandidates() {
		out.Candidates = append(out.Candidates, &globalngv1.ByocSharedComponentScaleNodeSpecCandidate{
			Name:               candidate.GetName(),
			Cpu:                candidate.GetCpu(),
			MemoryGi:           candidate.GetMemoryGi(),
			Available:          candidate.GetAvailable(),
			IsCurrent:          candidate.GetIsCurrent(),
			Reasons:            convertConstraintReasonsToGlobal(candidate.GetReasons()),
			RequiredMinSpec:    convertRequiredMinSpecToGlobal(candidate.GetRequiredMinSpec()),
			StorageConstraints: convertStorageConstraintsToGlobal(candidate.GetStorageConstraints()),
		})
	}
	return out
}

func convertRequiredMinSpecToGlobal(
	in *clusterngv1beta.SharedPoolScaleConstraintRequiredMinSpec,
) *globalngv1.ByocSharedComponentScaleRequiredMinSpec {
	if in == nil {
		return nil
	}
	return &globalngv1.ByocSharedComponentScaleRequiredMinSpec{
		Name:     in.GetName(),
		Cpu:      in.GetCpu(),
		MemoryGi: in.GetMemoryGi(),
	}
}

func convertStorageConstraintsToGlobal(
	in []*clusterngv1beta.SharedPoolScaleConstraintStorageConstraint,
) []*globalngv1.ByocSharedComponentScaleStorageConstraint {
	out := make([]*globalngv1.ByocSharedComponentScaleStorageConstraint, 0, len(in))
	for _, item := range in {
		storage := &globalngv1.ByocSharedComponentScaleStorageConstraint{
			StorageName:   item.GetName().String(),
			Resizable:     item.GetResizable(),
			CurrentSizeGi: item.GetCurrentSizeGi(),
		}
		if item.MaxSizeGi != nil {
			storage.MaxSizeGi = proto.Uint32(item.GetMaxSizeGi())
		}
		out = append(out, storage)
	}
	return out
}