package action

import (
	"context"
	"fmt"
	"path"
	"strconv"
	"strings"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	"github.com/tidbcloud/cluster-service-ng/config"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/pkg/clusterconfig"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplane"
	"github.com/tidbcloud/cluster-service-ng/pkg/k8s"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	schemav1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	templateversion "github.com/tidbcloud/cluster-service-ng/pkg/template/version"
	"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/protobuf/encoding/protojson"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
)

type ApplyComponentGroupAction struct {
	dataplaneManager    dataplane.Manager
	sharedPoolRepo      repo.SharedPoolRepo
	templateVersionSvc  *templateversion.Service
	imageConfig         *config.ImageConfig
	clusterConfigClient clusterconfig.Client
	dataplaneClient     dplanemgrv1.DataPlaneServiceClient
	meteringConfig      *config.MeteringConfig
	bizConfig           *config.BizConfig
}

const ApplyComponentGroupActionName = "apply-component-group"

func NewApplyComponentGroupAction(
	sharedPoolRepo repo.SharedPoolRepo,
	dataplaneManager dataplane.Manager,
	clusterConfigClient clusterconfig.Client,
	dataplaneClient dplanemgrv1.DataPlaneServiceClient,
	templateVersionSvc *templateversion.Service,
	imageConfig *config.ImageConfig,
	meteringConfig *config.MeteringConfig,
	bizConfig *config.BizConfig,
) actions.Action {
	return &ApplyComponentGroupAction{
		sharedPoolRepo:      sharedPoolRepo,
		dataplaneManager:    dataplaneManager,
		dataplaneClient:     dataplaneClient,
		templateVersionSvc:  templateVersionSvc,
		clusterConfigClient: clusterConfigClient,
		imageConfig:         imageConfig,
		meteringConfig:      meteringConfig,
		bizConfig:           bizConfig,
	}
}

func (a *ApplyComponentGroupAction) Name() string {
	return ApplyComponentGroupActionName
}

func (a *ApplyComponentGroupAction) Invoke(args actions.ActionArgs) error {
	// generate context
	ctx := utils.GenerateContext(args)
	tilog.Info(ctx, "apply component group action")

	_, err := args.GetString(workflow.ComponentIdsKey)
	if err != nil {
		return fmt.Errorf("target component id is required, but not found")
	}

	sharedPoolId, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return fmt.Errorf("failed to get sharedPool id: %w", err)
	}

	// load sharedPool
	sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolId, false)
	if err != nil {
		return fmt.Errorf("failed to load sharedPool: %w", err)
	}

	sharedPoolStr, err := args.GetString(workflow.SharedPoolKey)
	if err != nil {
		return fmt.Errorf("failed to get sharedPool: %w", err)
	}

	var tempSharedPool entity.SharedPool
	err = protojson.Unmarshal([]byte(sharedPoolStr), &tempSharedPool)
	if err != nil {
		tilog.Error(ctx, "failed to unmarshal sharedPool", zap.Error(err))
		return fmt.Errorf("failed to unmarshal sharedPool: %w", err)
	}

	sharedPool.ComponentGroups = tempSharedPool.GetComponentGroups()
	if sharedPool.GetVersion() == "" {
		sharedPool.Version = tempSharedPool.GetVersion()
	}

	remoteCli, err := a.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		tilog.Error(ctx, "failed to get remote client for shared pool", zap.String("dataplaneName", sharedPool.GetDataplaneName()), zap.Error(err))
		return fmt.Errorf("failed to get remote client for shared pool: %w", err)
	}

	var nodeVersionMap map[string]string

	for _, comp := range sharedPool.GetComponentGroups() {
		switch comp.GetComponentType() {
		case apiv1.ComponentType_PD,
			apiv1.ComponentType_PD_TSO,
			apiv1.ComponentType_PD_SCHEDULING,
			apiv1.ComponentType_TIDB,
			apiv1.ComponentType_TIPROXY,
			apiv1.ComponentType_TIFLASH_LEARNER,
			apiv1.ComponentType_TIFLASH_COMPUTE_LEARNER:
			continue
		default:
			// pass
		}

		// check if the component is target component
		if !workflow.CheckIsTargetComponent(ctx, comp, args) {
			tilog.Info(ctx, "skip apply component", zap.String("componentId", comp.GetId()), zap.String("componentType", comp.GetComponentType().String()))
			continue
		}

		if comp.GetPhase() == apiv1.Component_AVAILABLE {
			tilog.Info(ctx, "component is already available, skip applying", zap.String("componentId", comp.GetId()), zap.String("componentType", entity.GetComponentTypeName(comp)))
			continue
		}

		if shouldSkipFullComponentConfigForDeletingComponent(comp) {
			tilog.Info(ctx, "delete tikv group directly",
				zap.String("componentId", comp.GetId()),
				zap.String("componentType", entity.GetComponentTypeName(comp)))
			if err := deleteTiKVGroupDirectly(ctx, remoteCli, sharedPool, comp); err != nil {
				tilog.Error(ctx, "failed to delete tikv group",
					zap.String("componentId", comp.GetId()),
					zap.String("componentType", entity.GetComponentTypeName(comp)),
					zap.Error(err))
				return fmt.Errorf("failed to delete tikv group: %w", err)
			}
			continue
		}

		err = workflow.FullComponentConfigVersion(ctx, a.clusterConfigClient, a.meteringConfig, sharedPool, comp, "")
		if err != nil {
			tilog.Error(ctx, "failed to full component config", zap.String("componentId", comp.GetId()), zap.String("componentType", entity.GetComponentTypeName(comp)), zap.Error(err))
			return fmt.Errorf("failed to full component config: %w", err)
		}

		// Set keyspace label for TiDB Worker before building template args
		if comp.GetComponentType() == apiv1.ComponentType_TIDB_WORKER {
			if comp.Labels == nil {
				comp.Labels = make(map[string]string)
			}
			comp.Labels[tioperator.LabelKeyKeyspace] = tioperator.LabelValueKeyspaceSystem
		}

		// build template args
		if nodeVersionMap == nil {
			nodeVersionMap, err = getNodeVersions(ctx, a.dataplaneClient, sharedPool.GetDataplaneName())
			if err != nil {
				tilog.Error(ctx, "failed to get node versions for shared pool", zap.String("sharedPoolId", sharedPool.GetId()), zap.Error(err))
				return fmt.Errorf("failed to get node versions for shared pool: %w", err)
			}
		}
		templArgs, err := buildCompTempArgs(a.imageConfig, comp, sharedPool, nodeVersionMap)
		if err != nil {
			tilog.Error(ctx, "failed to build component template args", zap.String("componentId", comp.GetId()), zap.String("componentType", entity.GetComponentTypeName(comp)), zap.Error(err))
			return fmt.Errorf("failed to build component template args: %w", err)
		}

		switch comp.GetComponentType() {
		case apiv1.ComponentType_TIDB_WORKER:
			templArgs["SlowLogImage"] = fmt.Sprintf("%s/%s", a.imageConfig.Registry, a.imageConfig.SlowLogImage)

			if err := workflow.AddAzureDfsArgs(sharedPool, templArgs); err != nil {
				tilog.Error(
					ctx, "failed to add Azure DFS args",
					zap.String("componentId", comp.GetId()),
					zap.String("componentType", entity.GetComponentTypeName(comp)),
					zap.Error(err),
				)
				return fmt.Errorf("failed to add Azure DFS args for component %s (%s): %w",
					comp.GetId(), entity.GetComponentTypeName(comp), err)
			}

			if a.bizConfig.EnableSemv2 {
				templArgs["Semv2Enabled"] = "true"
			}

		case apiv1.ComponentType_TIKV:
			templArgs["InitImage"] = fmt.Sprintf("%s/%s", a.imageConfig.Registry, a.imageConfig.SlowLogImage)
			templArgs["TiKVPreStopImage"] = fmt.Sprintf("%s/%s", a.imageConfig.Registry, a.imageConfig.TiKVPreStopImage)
			templArgs["DfsPrefix"] = sharedPool.GetDfsPrefix()
			templArgs["DfsBucket"] = sharedPool.GetObjectStorage().GetBucket()
			templArgs["DfsRegion"] = sharedPool.GetObjectStorage().GetRegion()

			if sharedPool.GetDfxEndpoint() != "" {
				templArgs["DfsEndpoint"] = sharedPool.GetDfxEndpoint()
			}

			if err := workflow.AddAzureDfsArgs(sharedPool, templArgs); err != nil {
				return err
			}

			if sharedPool.GetProvider().GetType() == apiv1.ProviderType_ALICLOUD {
				templArgs["AliCloudRegion"] = sharedPool.GetProvider().GetRegion()
			}

			for _, storage := range comp.GetStorages() {
				switch storage.GetName() {
				case apiv1.Component_Storage_DATA:
					templArgs["DataStorageSize"] = strconv.FormatUint(uint64(storage.GetSizeGi()), 10)
					templArgs["DataStorageClass"] = workflow.GenStorageClassName(sharedPool, storage)
					if storage.GetType() != entity.LocalSSD {
						templArgs["DataVolumeAttributesClass"] = workflow.GenVolumeAttributesClassName(storage)
					} else {
						templArgs["WithLocalDisk"] = "true"
						templArgs["LocalDiskSize"] = fmt.Sprintf("%d", storage.SizeGi)
					}
				case apiv1.Component_Storage_RAFTLOG:
					templArgs["RaftLogStorageSize"] = strconv.FormatUint(uint64(storage.GetSizeGi()), 10)
					templArgs["RaftLogStorageClass"] = workflow.GenStorageClassName(sharedPool, storage)
					templArgs["RaftLogMountPath"] = storage.GetMountPath()
					templArgs["RaftLogVolumeAttributesClass"] = workflow.GenVolumeAttributesClassName(storage)
				// raft db deprecated
				case apiv1.Component_Storage_RAFTDB:
					templArgs["RaftDBStorageSize"] = strconv.FormatUint(uint64(storage.GetSizeGi()), 10)
					templArgs["RaftDBStorageClass"] = workflow.GenStorageClassName(sharedPool, storage)
					templArgs["RaftDBMountPath"] = storage.GetMountPath()
					templArgs["RaftDBVolumeAttributesClass"] = workflow.GenVolumeAttributesClassName(storage)
				default:
					tilog.Warn(ctx, "unsupported storage type for tiflash", zap.String("storage", storage.GetName().String()))
					continue
				}
			}

			if err := setTiKVRemoteWorkerTemplateArgs(templArgs, sharedPool); err != nil {
				return err
			}
			if err := setTiKVPlacementExclusiveTemplateArg(args, templArgs); err != nil {
				return err
			}

		case apiv1.ComponentType_TIFLASH_COMPUTE,
			apiv1.ComponentType_TIFLASH_WRITE:
			// render template

			proxyConfig, err := getTiFlashLearnerConfig(ctx, a.clusterConfigClient, comp, sharedPool)
			if err != nil {
				tilog.Error(ctx, "failed to get tiflash learner config", zap.String("componentId", comp.GetId()), zap.Error(err))
				return fmt.Errorf("failed to get tiflash learner config: %w", err)
			}

			templArgs["ProxyConfig"] = proxyConfig
			templArgs["LogImage"] = fmt.Sprintf("%s/%s", a.imageConfig.Registry, a.imageConfig.SlowLogImage)
			templArgs["DfsPrefix"] = sharedPool.GetDfsPrefix()
			templArgs["DfsBucket"] = sharedPool.GetObjectStorage().GetBucket()
			templArgs["DfsRegion"] = sharedPool.GetObjectStorage().GetRegion()

			if sharedPool.GetDfxEndpoint() != "" {
				templArgs["DfsEndpoint"] = sharedPool.GetDfxEndpoint()
			}

			if err := workflow.AddAzureDfsArgs(sharedPool, templArgs); err != nil {
				return err
			}

			if sharedPool.GetProvider().GetType() == apiv1.ProviderType_ALICLOUD {
				templArgs["AliCloudRegion"] = sharedPool.GetProvider().GetRegion()
			}

			setTiFlashColumnarTemplateArg(templArgs, sharedPool)

			for _, storage := range comp.GetStorages() {
				switch storage.GetName() {
				case apiv1.Component_Storage_DATA:
					templArgs["DataStorageSize"] = strconv.FormatUint(uint64(storage.GetSizeGi()), 10)
					templArgs["DataStorageClass"] = workflow.GenStorageClassName(sharedPool, storage)
					if storage.GetType() != entity.LocalSSD {
						templArgs["DataVolumeAttributesClass"] = workflow.GenVolumeAttributesClassName(storage)
					} else {
						templArgs["WithLocalDisk"] = "true"
						templArgs["LocalDiskSize"] = fmt.Sprintf("%d", storage.SizeGi)
					}
				default:
					tilog.Warn(ctx, "unsupported storage type for tiflash", zap.String("storage", storage.GetName().String()))
					continue
				}
			}

			// get tiflash config
			conf := schemav1.NewTomlConfig(map[string]any{})
			if err := conf.UnmarshalTOML([]byte(comp.GetConfiguration().GetConfigurationFileContent())); err != nil {
				return fmt.Errorf("failed to unmarshal tidb config: %w", err)
			}

			sleepSeconds := 630
			v := conf.Get(entity.TiFlashGracefulWaitShutdownKey)
			if v != nil {
				tilog.Info(ctx, "get graceful wait before shutdown time from tiflash config", zap.Any("gracefulTime", v))
				if vInt, ok := v.(int64); ok {
					sleepSeconds = int(vInt) + 30
				}
			}
			templArgs["PreStopSleepSeconds"] = strconv.Itoa(sleepSeconds)

		case apiv1.ComponentType_TIKV_WORKER, apiv1.ComponentType_COPROCESSOR_WORKER, apiv1.ComponentType_SCHEMA_MANAGER:
			templArgs["PDAddress"] = sharedPool.GetPdAddress()
			templArgs["DfsPrefix"] = sharedPool.GetDfsPrefix()
			templArgs["DfsBucket"] = sharedPool.GetObjectStorage().GetBucket()
			templArgs["DfsRegion"] = sharedPool.GetObjectStorage().GetRegion()

			if sharedPool.GetDfxEndpoint() != "" {
				templArgs["DfsEndpoint"] = sharedPool.GetDfxEndpoint()
			}

			if err := workflow.AddAzureDfsArgs(sharedPool, templArgs); err != nil {
				return err
			}

			if sharedPool.GetProvider().GetType() == apiv1.ProviderType_ALICLOUD {
				templArgs["AliCloudRegion"] = sharedPool.GetProvider().GetRegion()
			}

			templArgs["InstanceLabel"] = fmt.Sprintf("db-%s", sharedPool.GetDefaultComponenId())
			templArgs["ComponentLabel"] = strings.ToLower(strings.ReplaceAll(comp.GetComponentType().String(), "_", "-"))

			if comp.GetComponentType() == apiv1.ComponentType_COPROCESSOR_WORKER ||
				comp.GetComponentType() == apiv1.ComponentType_SCHEMA_MANAGER {
				for _, storage := range comp.GetStorages() {
					if storage.GetName() == apiv1.Component_Storage_DATA {
						templArgs["DataStorageSize"] = strconv.FormatUint(uint64(storage.GetSizeGi()), 10)
						templArgs["DataStorageClass"] = workflow.GenStorageClassName(sharedPool, storage)
						templArgs["DataVolumeAttributesClass"] = workflow.GenVolumeAttributesClassName(storage)
					}
				}
			}

		default:
			// not need to render template
			tilog.Info(ctx, "skip render template for component", zap.String("componentId", comp.GetId()), zap.String("componentType", entity.GetComponentTypeName(comp)))
			continue
		}

		templateName, err := entity.GetComponentTemplateName(comp)
		if err != nil {
			tilog.Error(ctx, "failed to get component template name", zap.String("componentId", comp.GetId()), zap.String("componentType", entity.GetComponentTypeName(comp)), zap.Error(err))
			return fmt.Errorf("failed to get component template name: %w", err)
		}

		// if component phase is deleting, set needDelete to true
		err = templateExecuteWithVersionResolver(
			ctx, remoteCli, templateName, comp.GetTemplateVersion(), templArgs,
			comp.GetPhase() == apiv1.Component_DELETING, a.templateVersionSvc,
		)
		if err != nil {
			tilog.Error(ctx, "failed to execute template", zap.String("componentId", comp.GetId()), zap.String("componentType", entity.GetComponentTypeName(comp)), zap.Error(err))
			return fmt.Errorf("failed to execute template: %w", err)
		}
	}

	return nil
}

func shouldSkipFullComponentConfigForDeletingComponent(comp *entity.Component) bool {
	return comp.GetPhase() == apiv1.Component_DELETING && comp.GetComponentType() == apiv1.ComponentType_TIKV
}

func deleteTiKVGroupDirectly(ctx context.Context, remoteCli client.Client, sharedPool *entity.SharedPool, comp *entity.Component) error {
	return k8s.DeleteObject(ctx, remoteCli, &v1alpha1.TiKVGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      entity.GetComponentGroupName(comp),
			Namespace: sharedPool.GetDataPlaneNamespace(),
		},
	})
}

func setTiFlashColumnarTemplateArg(templArgs map[string]string, sharedPool *entity.SharedPool) {
	if entity.UsesTiFlashColumnar(sharedPool) {
		templArgs["TiFlashColumnar"] = "true"
	}
}

func buildCompTempArgs(imageConfig *config.ImageConfig, comp *entity.Component, sharedPool *entity.SharedPool, nodeVersionMap map[string]string) (map[string]string, error) {
	templArgs := make(map[string]string)
	needDelete := comp.GetPhase() == apiv1.Component_DELETING

	// set default template args
	templArgs["Image"] = fmt.Sprintf(
		"%s/%s/%s:%s",
		imageConfig.Registry,
		imageConfig.RepoPrefix,
		path.Base(comp.GetConfiguration().GetImageRepo()),
		strings.TrimSpace(comp.GetConfiguration().GetImageTag()),
	)
	templArgs["ClusterId"] = sharedPool.GetId()
	templArgs["SharedPoolId"] = sharedPool.GetId()
	templArgs["ComponentGroup"] = entity.GetComponentGroupName(comp)
	templArgs["ComponentType"] = entity.GetComponentTypeName(comp)
	templArgs["ComponentId"] = comp.GetId()
	templArgs["Version"] = comp.GetVersion()
	templArgs["Replicas"] = strconv.FormatUint(uint64(comp.GetNodeQuota().GetNodeCount()), 10)
	templArgs["Config"] = comp.GetConfiguration().GetConfigurationFileContent()
	if comp.GetComponentType() == apiv1.ComponentType_TIPROXY {
		templArgs["TiProxyGracefulShutdownDeleteDelaySeconds"] = strconv.FormatInt(entity.DefaultTiProxyGracefulShutdownDeleteDelaySeconds, 10)
		templArgs["TiProxyGracefulRestartMaxSurge"] = strconv.FormatInt(int64(entity.DefaultTiProxyGracefulRestartMaxSurge), 10)
	}

	if imageConfig.ImagePullPolicy == "" {
		// set default image pull policy if not set
		imageConfig.ImagePullPolicy = "IfNotPresent"
	}
	templArgs["ImagePullPolicy"] = imageConfig.ImagePullPolicy

	// set tenant id
	if sharedPool.GetProvider().GetTags() != nil && sharedPool.GetProvider().GetTags()[schemav1.LabelUserTenantKey] != "" {
		templArgs["UserTenantId"] = sharedPool.GetProvider().GetTags()[schemav1.LabelUserTenantKey]
	}

	// Deletion only needs stable template identity for rendering the delete manifest.
	// Resource requirements and the real node version are irrelevant in that path and
	// older records may not have the newer available resource fields populated.
	if !needDelete {
		err := workflow.SetResourceReady(comp, templArgs)
		if err != nil {
			return templArgs, fmt.Errorf("failed to set resource requirements: %w", err)
		}
	} else {
		setDeleteSafeTemplateArgs(comp, templArgs)
	}

	// Read the durable per-pool gate from the shared pool. It is the same
	// source of truth shared with the NodePool infrastructure side (DPM gRPC
	// + EKS controller), so a single value governs both scheduling labels
	// and partition placement. Re-deriving here would flip only the
	// scheduling side and pin workloads on labels the legacy nodes do not
	// carry. See docs/design/single-az-ha-placement-groups.md.
	placementGroupEnabled := sharedPool.GetProvider().GetPlacementGroupEnabled()
	err := workflow.BuildTopologies(comp.GetComponentType(), sharedPool.GetProvider().GetNodeZones(), templArgs, placementGroupEnabled)
	if err != nil {
		return templArgs, fmt.Errorf("failed to build topologies: %w", err)
	}

	injectPDLocationLabels(comp, sharedPool, templArgs)

	// set nodespec
	templArgs["NodeSpec"] = entity.GetComponentNodeSpec(comp.GetNodeQuota().GetNodeSpec())

	// if max node count not set, set it to node count
	if comp.GetNodeQuota().GetMaxNodeCount() == 0 {
		comp.GetNodeQuota().MaxNodeCount = comp.GetNodeQuota().GetNodeCount()
	}
	templArgs["MaxNodeCount"] = strconv.FormatUint(uint64(comp.GetNodeQuota().GetMaxNodeCount()), 10)
	templArgs["MinNodeCount"] = strconv.FormatUint(uint64(comp.GetNodeQuota().GetMinNodeCount()), 10)

	if needDelete {
		templArgs["NodeVersion"] = "v1.0.0"
	} else {
		nodeVersion, err := getNodeVersion(nodeVersionMap, comp)
		if err != nil {
			return templArgs, fmt.Errorf("failed to get node version: %w", err)
		}
		templArgs["NodeVersion"] = nodeVersion
	}

	if comp.GetComponentType() == apiv1.ComponentType_PD {
		pdMode, shouldRenderPDMode := resolvePDMode(sharedPool)
		if shouldRenderPDMode {
			templArgs["PDMode"] = pdMode
		}
	}

	// Build server labels if component has labels
	if len(comp.GetLabels()) > 0 {
		templArgs["ServerLabels"] = workflow.BuildServerLabelsYAML(comp.GetLabels())
	}

	return templArgs, nil
}

func setDeleteSafeTemplateArgs(comp *entity.Component, templArgs map[string]string) {
	if comp == nil || comp.GetNodeQuota() == nil {
		return
	}

	switch comp.GetComponentType() {
	case apiv1.ComponentType_TIDB, apiv1.ComponentType_TIDB_WORKER:
		memoryMiB := uint64(1024)
		if nodeSpec := comp.GetNodeQuota().GetNodeSpec(); nodeSpec != nil {
			memoryMiB = uint64(nodeSpec.GetAvailableMemoryMi())
			if memoryMiB == 0 {
				memoryGi := nodeSpec.GetMemoryGi()
				if memoryGi == 0 {
					memoryGi = 1
				}
				memoryMiB = uint64(memoryGi) * 1024
			}
		}
		templArgs["GOMEMLIMIT"] = strconv.FormatUint(uint64(float64(memoryMiB)*tioperator.GoMemlimitRate), 10)
	}
}

func injectPDLocationLabels(comp *entity.Component, sharedPool *entity.SharedPool, templArgs map[string]string) {
	// Read the same per-pool gate BuildTopologies uses so PD schedules only
	// against labels the NodePool actually carries. See the gate rationale
	// in buildCompTempArgs above.
	if !sharedPool.GetProvider().GetPlacementGroupEnabled() {
		return
	}
	if comp.GetComponentType() != apiv1.ComponentType_PD && comp.GetComponentType() != apiv1.ComponentType_PD_SCHEDULING {
		return
	}

	const locationLabelsLine = `  location-labels = ["tidbcloud.com/pg-partition"]`
	componentConfig := templArgs["Config"]
	if strings.Contains(componentConfig, "location-labels") {
		return
	}
	if strings.Contains(componentConfig, "[replication]") {
		componentConfig = strings.Replace(componentConfig, "[replication]", "[replication]\n"+locationLabelsLine, 1)
	} else {
		componentConfig += "\n[replication]\n" + locationLabelsLine
	}
	templArgs["Config"] = componentConfig
}

func resolvePDMode(sharedPool *entity.SharedPool) (string, bool) {
	if entity.GetEffectivePDDeploymentMode(sharedPool.GetPdDeploymentMode()) == apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_MONOLITHIC {
		return "", false
	}

	if sharedPool.GetCapacityMode() != apiv1.CapacityMode_TOPOLOGY_DRIVEN {
		return "ms", true
	}

	if hasSharedPoolComponentWithPositiveNodeCount(sharedPool.GetComponentGroups(), apiv1.ComponentType_PD_TSO) {
		return "ms", true
	}

	// TOPOLOGY_DRIVEN and PD_TSO node_count=0: keep PD in normal mode by omitting spec.mode.
	return "", false
}

func hasSharedPoolComponentWithPositiveNodeCount(components []*entity.Component, componentType apiv1.ComponentType) bool {
	for _, comp := range components {
		if comp == nil {
			continue
		}
		if comp.GetComponentType() == componentType && comp.GetNodeQuota().GetNodeCount() > 0 {
			return true
		}
	}
	return false
}

func setTiKVRemoteWorkerTemplateArgs(templArgs map[string]string, sharedPool *entity.SharedPool) error {
	remoteWorkers, err := entity.ResolveTiKVRemoteWorkerGroups(sharedPool)
	if err != nil {
		return fmt.Errorf("failed to resolve TiKV remote workers: %w", err)
	}
	if remoteWorkers.Worker == "" {
		return nil
	}
	templArgs["RemoteWorkerGroup"] = remoteWorkers.Worker
	templArgs["RemoteCoprocessorWorkerGroup"] = remoteWorkers.Coprocessor
	return nil
}

func setTiKVPlacementExclusiveTemplateArg(args actions.ActionArgs, templArgs map[string]string) error {
	arg, ok := args[workflow.TiKVPlacementExclusiveKey]
	if !ok {
		return nil
	}

	exclusive, err := strconv.ParseBool(arg.String())
	if err != nil {
		return fmt.Errorf("failed to parse %s: %w", workflow.TiKVPlacementExclusiveKey, err)
	}
	if exclusive {
		templArgs["TiKVPlacementExclusive"] = "true"
	}
	return nil
}

func templateExecuteWithVersionResolver(
	ctx context.Context,
	remoteCli client.Client,
	templName,
	templVersion string,
	templArgs map[string]string,
	needDelete bool,
	templateVersionSvc *templateversion.Service,
) error {
	yamlStr, resolvedName, resolvedVersion, err := templateversion.RenderWithVersion(ctx, templateVersionSvc, templName, templVersion, templArgs)
	if err != nil {
		return err
	}

	resourceCr, _, err := k8s.GenerateUnstructuredFromYaml(yamlStr)
	if err != nil {
		tilog.Error(ctx, "failed to generate unstructured from yaml", zap.String("yaml", yamlStr), zap.Error(err))
		return fmt.Errorf("failed to generate unstructured from yaml: %w", err)
	}

	templateversion.SetVersionAnnotations(resourceCr, resolvedName, resolvedVersion)

	if needDelete {
		tilog.Warn(ctx, "will delete the resource", zap.Any("resourceCr", resourceCr))
		if err = k8s.DeleteObject(ctx, remoteCli, resourceCr); err != nil {
			tilog.Error(ctx, "failed to delete resource", zap.Any("resourceCr", resourceCr), zap.Error(err))
			return fmt.Errorf("failed to delete resource: %w", err)
		}
		return nil
	}

	toCr := resourceCr.DeepCopy()
	res, err := controllerutil.CreateOrUpdate(ctx, remoteCli, toCr, func() error {
		// Handle resources that don't have spec field (like ClusterRole, ClusterRoleBinding)
		switch resourceCr.GetKind() {
		case "ClusterRole", "Role":
			toCr.Object["rules"] = resourceCr.Object["rules"]
		case "ClusterRoleBinding", "RoleBinding":
			toCr.Object["roleRef"] = resourceCr.Object["roleRef"]
			toCr.Object["subjects"] = resourceCr.Object["subjects"]
		default:
			// CreateOrUpdate fetches the live object into toCr on the update path.
			// On the create path, toCr is still the desired template copy.
			toCr.Object["spec"] = desiredSpecForUpdate(toCr, resourceCr)
		}
		toCr.SetAnnotations(resourceCr.GetAnnotations())
		toCr.SetLabels(resourceCr.GetLabels())
		return nil
	})
	if err != nil {
		tilog.Error(ctx, "failed to apply create or update", zap.Any("resourceCr", resourceCr), zap.Error(err))
		return fmt.Errorf("failed to apply create or update: %w", err)
	}

	tilog.Info(
		ctx, "create or update resource success",
		zap.String("resourceName", toCr.GetName()),
		zap.String("resourceType", toCr.GetKind()),
		zap.Any("res", res),
	)
	return nil
}

func desiredSpecForUpdate(current, desired *unstructured.Unstructured) any {
	spec := desired.Object["spec"]
	if desired.GetKind() == "TiKVGroup" {
		return desiredTiKVGroupSpecForUpdate(current, desired, spec)
	}
	if desired.GetKind() != "Deployment" {
		return spec
	}
	desiredSpec, ok := spec.(map[string]any)
	if !ok {
		return spec
	}
	if _, exists, _ := unstructured.NestedFieldNoCopy(desiredSpec, "replicas"); exists {
		return spec
	}
	// Preserve live HPA-controlled replicas when an existing Deployment is
	// updated by a template that intentionally omits spec.replicas. On create,
	// current has no live replicas to preserve, so the desired spec is used as-is.
	replicas, exists, err := unstructured.NestedFieldCopy(current.Object, "spec", "replicas")
	if err != nil || !exists {
		return spec
	}
	merged := make(map[string]any, len(desiredSpec)+1)
	for k, v := range desiredSpec {
		merged[k] = v
	}
	merged["replicas"] = replicas
	return merged
}

func desiredTiKVGroupSpecForUpdate(current, desired *unstructured.Unstructured, spec any) any {
	desiredSpec, ok := spec.(map[string]any)
	if !ok {
		return spec
	}
	if _, exists, _ := unstructured.NestedFieldNoCopy(desiredSpec, "template", "spec", "placement"); exists {
		return spec
	}
	placement, exists, err := unstructured.NestedFieldCopy(current.Object, "spec", "template", "spec", "placement")
	if err != nil || !exists {
		return spec
	}

	desiredCopy := desired.DeepCopy()
	if err := unstructured.SetNestedField(desiredCopy.Object, placement, "spec", "template", "spec", "placement"); err != nil {
		return spec
	}
	merged, _, err := unstructured.NestedMap(desiredCopy.Object, "spec")
	if err != nil {
		return spec
	}
	return merged
}

func getNodeVersions(ctx context.Context, dataplaneClient dplanemgrv1.DataPlaneServiceClient, dataplaneName string) (map[string]string, error) {
	nodePools, err := dataplaneClient.ListNodePools(ctx, &dplanemgrv1.ListNodePoolsRequest{
		DataPlaneId: dataplaneName,
	})
	if err != nil {
		return nil, err
	}
	nodeVersionMap := make(map[string]string)
	for _, np := range nodePools.GetNodePools() {
		key := fmt.Sprintf("%dC%dG", np.GetCpuCore(), np.GetMemoryGi())
		if np.GetWithLocalDisk() {
			key = fmt.Sprintf("%s-%s", key, "Local")
		}

		// if node spec already exists, use higher version
		// ali: "1.34.1-aliyun.1"
		// aws: "1.34"
		if existingVersion, ok := nodeVersionMap[key]; ok {
			cmp, err := k8s.CompareK8sVersion(np.GetK8SVersion(), existingVersion)
			if err != nil {
				tilog.Warn(ctx, "failed to compare k8s version, using new version",
					zap.String("newVersion", np.GetK8SVersion()),
					zap.String("existingVersion", existingVersion),
					zap.Error(err))
			} else if cmp <= 0 {
				// existing version is higher or equal, skip
				continue
			}
		}
		nodeVersionMap[key] = np.GetK8SVersion()
	}

	return nodeVersionMap, nil
}

func getNodeVersion(nodeVersionMap map[string]string, comp *entity.Component) (string, error) {
	// get k8s version from node pools
	nodeSpec := entity.GetComponentNodeSpec(comp.GetNodeQuota().GetNodeSpec())

	for _, storage := range comp.GetStorages() {
		if storage.GetName() != apiv1.Component_Storage_DATA {
			continue
		}
		if storage.GetType() == entity.LocalSSD {
			nodeSpec = fmt.Sprintf("%s-%s", nodeSpec, "Local")
		}
		break
	}

	k8sVersion, ok := nodeVersionMap[nodeSpec]
	if !ok {
		return "", fmt.Errorf("failed to get node version for node spec: %s", nodeSpec)
	}

	if k8sVersion == "" {
		return "", fmt.Errorf("get node version is empty")
	}
	return k8sVersion, nil
}

func getTiFlashLearnerConfig(ctx context.Context, clusterConfigClient clusterconfig.Client, comp *entity.Component, sharedPool *entity.SharedPool) (string, error) {
	var proxyComp *entity.Component
	switch comp.GetComponentType() {
	case apiv1.ComponentType_TIFLASH_WRITE:
		for _, c := range sharedPool.GetComponentGroups() {
			if c.GetComponentType() != apiv1.ComponentType_TIFLASH_LEARNER || c.GetId() != comp.GetId() {
				continue
			}
			proxyComp = c
		}

	case apiv1.ComponentType_TIFLASH_COMPUTE:
		// TODO use new comp to get proxy config
		for _, c := range sharedPool.GetComponentGroups() {
			if c.GetComponentType() != apiv1.ComponentType_TIFLASH_COMPUTE_LEARNER || c.GetId() != comp.GetId() {
				continue
			}
			proxyComp = c
		}
	case apiv1.ComponentType_TIFLASH_LEARNER,
		apiv1.ComponentType_TIFLASH_COMPUTE_LEARNER:
		proxyComp = comp
	default:
		return "", fmt.Errorf("unsupported component type for tiflash learner config: %s", comp.GetComponentType().String())
	}

	// full component config from config center
	// tiflash proxy do not need metering config
	err := workflow.FullComponentConfigVersion(ctx, clusterConfigClient, &config.MeteringConfig{Enable: false}, sharedPool, proxyComp, "")
	if err != nil {
		return "", fmt.Errorf("failed to full tiflash learner component config: %w", err)
	}
	return proxyComp.GetConfiguration().GetConfigurationFileContent(), nil
}