package service

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"os"
	"sort"
	"time"

	opv2 "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/cluster-agent/v1"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	apimeta "k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	replaceNodeOperationIDLabelKey = "cluster-agent.tidbcloud.com/replace-node-operation-id"
	replaceNodeWarmUpLabelKey      = "cluster-agent.tidbcloud.com/replace-node-warm-up"
	replaceNodeStateLabelKey       = "cluster-agent.tidbcloud.com/replace-node-state"

	// Warm-up pod annotations are the prepare-stage operation record. They keep
	// Commit/Abort recoverable without in-memory state while the warm-up pod exists.
	replaceNodeTargetPodNameAnnotationKey   = "cluster-agent.tidbcloud.com/replace-node-target-pod-name"
	replaceNodeTargetPodUIDAnnotationKey    = "cluster-agent.tidbcloud.com/replace-node-target-pod-uid"
	replaceNodeOldNodeAnnotationKey         = "cluster-agent.tidbcloud.com/replace-node-old-node"
	replaceNodeOwnerAPIVersionAnnotationKey = "cluster-agent.tidbcloud.com/replace-node-owner-api-version"
	replaceNodeOwnerKindAnnotationKey       = "cluster-agent.tidbcloud.com/replace-node-owner-kind"
	replaceNodeOwnerNameAnnotationKey       = "cluster-agent.tidbcloud.com/replace-node-owner-name"
	replaceNodeInstanceIDAnnotationKey      = "cluster-agent.tidbcloud.com/replace-node-instance-id"

	replaceNodeWarmUpImageEnvKey = "REPLACE_NODE_WARMUP_IMAGE"
	localSSDStorageClassName     = "local-ssd"

	// The state ConfigMap preserves operation metadata once Prepare mutates the
	// old node, and remains after Commit deletes the warm-up pod for status polling.
	replaceNodeStateOperationIDKey     = "operation_id"
	replaceNodeStatePodNameKey         = "pod_name"
	replaceNodeStateOldPodUIDKey       = "old_pod_uid"
	replaceNodeStateOldNodeKey         = "old_node"
	replaceNodeStateWarmUpNodeKey      = "warm_up_node"
	replaceNodeStateOwnerAPIVersionKey = "owner_api_version"
	replaceNodeStateComponentKindKey   = "component_kind"
	replaceNodeStateOwnerNameKey       = "owner_name"
	replaceNodeStateInstanceIDKey      = "instance_id"
	replaceNodeStateAbortedAtKey       = "aborted_at"

	// This label is supporting evidence only; the per-instance ownerRef decides support.
	tidbOperatorManagedLabelKey = "app.kubernetes.io/managed-by"
	tidbOperatorManagedByValue  = "tidb-operator"
)

var tidbOperatorInstanceKinds = map[string]struct{}{
	"PD":         {},
	"TiKV":       {},
	"TiDB":       {},
	"TiFlash":    {},
	"TiProxy":    {},
	"TSO":        {},
	"Scheduling": {},
	"TiKVWorker": {},
}

type replaceNodeTarget struct {
	pod           *corev1.Pod
	node          *corev1.Node
	ownerRef      metav1.OwnerReference
	componentKind string
	instanceID    string
	oldPodUID     types.UID
	oldNode       string
}

func (s *AgentService) PrepareReplaceNode(ctx context.Context, req *apiv1.PrepareReplaceNodeReq) (*apiv1.ReplaceNodeResp, error) {
	logger := replaceNodePrepareLogger(ctx, req)
	logger.Info("prepare replace node")

	if err := validatePrepareReplaceNodeReq(req); err != nil {
		return nil, err
	}
	namespace, err := replaceNodeNamespace()
	if err != nil {
		return nil, err
	}

	// Operation state is authoritative once it exists. This keeps retried Prepare
	// calls from resurrecting an operation that was already aborted or committed.
	existingState, err := s.getReplaceNodeOperationState(ctx, namespace, req.OperationId)
	hasExistingState := false
	switch {
	case err != nil && status.Code(err) != codes.NotFound:
		return nil, err
	case err == nil && existingState.isAborted():
		return replaceNodeAbortResp("replace node operation already aborted", existingState), nil
	case err == nil:
		committed, err := s.replaceNodeOperationCommitted(ctx, namespace, existingState)
		if err != nil {
			return nil, err
		}
		if committed {
			return s.buildCommittedReplaceNodeResp(ctx, namespace, existingState, "replace node already committed")
		}
		hasExistingState = true
	}

	target, err := s.resolveReplaceNodeTarget(ctx, namespace, req)
	if err != nil {
		return nil, err
	}
	if hasExistingState {
		if err := validateReplaceNodeStateMatchesTarget(existingState, target); err != nil {
			return nil, err
		}
		if target.instanceID == "" {
			target.instanceID = existingState.InstanceID
		}
	}
	if !req.GetForce() {
		if err := rejectUnreadyReplaceNodeTargetPod(target.pod); err != nil {
			return nil, err
		}
	}
	if err := s.rejectLocalSSD(ctx, target.pod); err != nil {
		return nil, err
	}

	instance := newTiDBOperatorInstanceObject(target.pod.Namespace, target.ownerRef)
	if err := s.dataplaneCli.Get(ctx, client.ObjectKeyFromObject(instance), instance); err != nil {
		return nil, fmt.Errorf("failed to get operator instance %s/%s: %w", instance.GetNamespace(), instance.GetName(), err)
	}
	// A retried Prepare after Commit should behave like a status poll, not create
	// a second warm-up pod for an already committed operation.
	if instance.GetLabels()[replaceNodeOperationIDLabelKey] == req.OperationId {
		state, err := s.getReplaceNodeOperationState(ctx, namespace, req.OperationId)
		if err != nil {
			return nil, err
		}
		return s.buildCommittedReplaceNodeResp(ctx, namespace, state, "replace node already committed")
	}

	if err := s.preflightWarmUpPod(ctx, req.OperationId, target); err != nil {
		return nil, err
	}

	// Store enough state before cordoning the old node. If warm-up creation fails
	// later, Abort can still find the old node and reverse the cordon.
	state := operationStateFromTarget(req.OperationId, target)
	if err := s.ensureReplaceNodeStateConfigMap(ctx, namespace, state); err != nil {
		return nil, err
	}

	node, err := s.cordonReplaceNode(ctx, target.oldNode)
	if err != nil {
		return nil, err
	}
	target.node = node
	if target.instanceID == "" {
		target.instanceID, _ = parseProviderInstanceIDFromNode(node)
		state.InstanceID = target.instanceID
		if err := s.ensureReplaceNodeStateConfigMap(ctx, namespace, state); err != nil {
			return nil, err
		}
	}
	logger.Info("old node cordoned", zap.String("old_node", target.oldNode))

	warmUpPod, created, err := s.ensureWarmUpPod(ctx, req.OperationId, target)
	if err != nil {
		return nil, err
	}
	if created {
		logger.Info("warm-up pod created", zap.String("pod", warmUpPod.Name))
	}

	return replaceNodeRespFromWarmUpPod(warmUpPod, target), nil
}

func (s *AgentService) GetReplaceNodeOperation(ctx context.Context, req *apiv1.ReplaceNodeOperationReq) (*apiv1.ReplaceNodeResp, error) {
	logger := replaceNodeOperationLogger(ctx, req)
	logger.Info("get replace node operation")

	if err := validateReplaceNodeOperationReq(req); err != nil {
		return nil, err
	}
	namespace, err := replaceNodeNamespace()
	if err != nil {
		return nil, err
	}

	state, err := s.getReplaceNodeOperationState(ctx, namespace, req.OperationId)
	if err != nil {
		return nil, err
	}
	return s.buildReplaceNodeOperationResp(ctx, namespace, state)
}

func (s *AgentService) CommitReplaceNode(ctx context.Context, req *apiv1.ReplaceNodeOperationReq) (*apiv1.ReplaceNodeResp, error) {
	logger := replaceNodeOperationLogger(ctx, req)
	logger.Info("commit replace node")

	if err := validateReplaceNodeOperationReq(req); err != nil {
		return nil, err
	}
	namespace, err := replaceNodeNamespace()
	if err != nil {
		return nil, err
	}

	state, err := s.getReplaceNodeOperationState(ctx, namespace, req.OperationId)
	if err != nil {
		return nil, err
	}
	if state.isAborted() {
		return nil, status.Errorf(codes.FailedPrecondition,
			"replace node operation %s is aborted and cannot be committed", req.OperationId)
	}
	committed, err := s.replaceNodeOperationCommitted(ctx, namespace, state)
	if err != nil {
		return nil, err
	}
	if committed {
		if err := s.deleteWarmUpPodIfPresent(ctx, namespace, req.OperationId); err != nil {
			return nil, err
		}
		return s.buildCommittedReplaceNodeResp(ctx, namespace, state, "replace node already committed")
	}

	warmUpPod, warmUpFound, err := s.getWarmUpPod(ctx, namespace, req.OperationId)
	if err != nil {
		return nil, err
	}
	if !warmUpFound {
		return nil, status.Errorf(codes.FailedPrecondition,
			"replace node operation %s has no warm-up pod and is not committed", req.OperationId)
	}
	if err := validateWarmUpPod(warmUpPod, req.OperationId, nil); err != nil {
		return nil, err
	}
	if warmUpPod.Status.Phase != corev1.PodRunning || warmUpPod.Spec.NodeName == "" {
		resp := replaceNodeRespFromState("waiting for warm-up pod to run", state)
		resp.Phase = apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_COMMITTING
		resp.WarmUpPod = warmUpPod.Name
		resp.WarmUpNode = warmUpPod.Spec.NodeName
		return resp, nil
	}
	state.WarmUpNode = warmUpPod.Spec.NodeName

	instance := newTiDBOperatorInstanceObject(namespace, state.ownerRef())
	// Persist commit state before deleting the warm-up pod. After the operator
	// CR is labeled, AbortReplaceNode must not auto-rollback this operation.
	if err := s.ensureReplaceNodeStateConfigMap(ctx, namespace, state); err != nil {
		return nil, err
	}
	committed, err = s.commitOperatorInstanceRestart(ctx, instance, req.OperationId)
	if err != nil {
		return nil, err
	}
	if err := s.deletePod(ctx, warmUpPod); err != nil {
		return nil, err
	}
	logger.Info("warm-up pod deleted", zap.String("pod", warmUpPod.Name))
	if committed {
		logger.Info("operator instance restart committed",
			zap.String("kind", state.ComponentKind),
			zap.String("name", state.OwnerName))
	}

	return s.buildCommittedReplaceNodeResp(ctx, namespace, state, "replace node committed")
}

func (s *AgentService) AbortReplaceNode(ctx context.Context, req *apiv1.ReplaceNodeOperationReq) (*apiv1.ReplaceNodeResp, error) {
	logger := replaceNodeOperationLogger(ctx, req)
	logger.Info("abort replace node")

	if err := validateReplaceNodeOperationReq(req); err != nil {
		return nil, err
	}
	namespace, err := replaceNodeNamespace()
	if err != nil {
		return nil, err
	}

	state, err := s.getReplaceNodeOperationState(ctx, namespace, req.OperationId)
	if err != nil {
		return nil, err
	}
	if state.isAborted() {
		return replaceNodeAbortResp("replace node operation already aborted", state), nil
	}
	committed, err := s.replaceNodeOperationCommitted(ctx, namespace, state)
	if err != nil {
		return nil, err
	}
	if committed {
		return nil, status.Errorf(codes.FailedPrecondition,
			"replace node operation %s is already committed and cannot be aborted automatically", req.OperationId)
	}

	warmUpPod, found, err := s.getWarmUpPod(ctx, namespace, req.OperationId)
	if err != nil {
		return nil, err
	}
	if found {
		if err := validateWarmUpPod(warmUpPod, req.OperationId, nil); err != nil {
			return nil, err
		}
		if state.WarmUpNode == "" {
			state.WarmUpNode = warmUpPod.Spec.NodeName
		}
	}

	if state.OldNode != "" {
		if err := s.uncordonReplaceNode(ctx, state.OldNode); err != nil {
			return nil, err
		}
	}

	if found {
		if err := s.deletePod(ctx, warmUpPod); err != nil {
			return nil, err
		}
	}
	state.AbortedAt = metav1.Now().Format(time.RFC3339)
	if err := s.ensureReplaceNodeStateConfigMap(ctx, namespace, state); err != nil {
		return nil, err
	}
	logger.Info("replace node abort completed", zap.String("old_node", state.OldNode))

	return replaceNodeAbortResp("replace node operation aborted", state), nil
}

func replaceNodePrepareLogger(ctx context.Context, req *apiv1.PrepareReplaceNodeReq) *zap.Logger {
	if req == nil {
		return tilog.WithContextL(ctx)
	}
	return tilog.WithContextL(ctx).With(
		zap.String("operation_id", req.OperationId),
		zap.String("cluster_id", req.ClusterId),
		zap.String("pod_name", req.GetPodName()),
		zap.String("instance_id", req.GetInstanceId()),
	)
}

func replaceNodeOperationLogger(ctx context.Context, req *apiv1.ReplaceNodeOperationReq) *zap.Logger {
	if req == nil {
		return tilog.WithContextL(ctx)
	}
	return tilog.WithContextL(ctx).With(
		zap.String("operation_id", req.OperationId),
		zap.String("cluster_id", req.ClusterId),
	)
}

func replaceNodeNamespace() (string, error) {
	namespace := os.Getenv("NAMESPACE")
	if namespace == "" {
		return "", fmt.Errorf("NAMESPACE is empty")
	}
	return namespace, nil
}

func validatePrepareReplaceNodeReq(req *apiv1.PrepareReplaceNodeReq) error {
	if req == nil {
		return status.Error(codes.InvalidArgument, "request is required")
	}
	if req.OperationId == "" {
		return status.Error(codes.InvalidArgument, "operation_id is required")
	}
	if req.ClusterId == "" {
		return status.Error(codes.InvalidArgument, "cluster_id is required")
	}
	switch req.GetTarget().(type) {
	case *apiv1.PrepareReplaceNodeReq_PodName:
		if req.GetPodName() == "" {
			return status.Error(codes.InvalidArgument, "pod_name must not be empty")
		}
	case *apiv1.PrepareReplaceNodeReq_InstanceId:
		if req.GetInstanceId() == "" {
			return status.Error(codes.InvalidArgument, "instance_id must not be empty")
		}
	default:
		return status.Error(codes.InvalidArgument, "exactly one of pod_name or instance_id is required")
	}
	return nil
}

func validateReplaceNodeOperationReq(req *apiv1.ReplaceNodeOperationReq) error {
	if req == nil {
		return status.Error(codes.InvalidArgument, "request is required")
	}
	if req.OperationId == "" {
		return status.Error(codes.InvalidArgument, "operation_id is required")
	}
	if req.ClusterId == "" {
		return status.Error(codes.InvalidArgument, "cluster_id is required")
	}
	return nil
}

func (s *AgentService) resolveReplaceNodeTarget(ctx context.Context, namespace string, req *apiv1.PrepareReplaceNodeReq) (*replaceNodeTarget, error) {
	switch req.GetTarget().(type) {
	case *apiv1.PrepareReplaceNodeReq_PodName:
		return s.resolveReplaceNodeTargetByPodName(ctx, namespace, req.GetPodName())
	case *apiv1.PrepareReplaceNodeReq_InstanceId:
		return s.resolveReplaceNodeTargetByInstanceID(ctx, namespace, req.GetInstanceId())
	default:
		return nil, status.Error(codes.InvalidArgument, "exactly one of pod_name or instance_id is required")
	}
}

func (s *AgentService) resolveReplaceNodeTargetByPodName(ctx context.Context, namespace, podName string) (*replaceNodeTarget, error) {
	var pod corev1.Pod
	if err := s.dataplaneCli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: podName}, &pod); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "pod %s not found", podName)
		}
		return nil, fmt.Errorf("failed to get pod %s: %w", podName, err)
	}
	return buildReplaceNodeTargetFromPod(&pod, "")
}

func (s *AgentService) resolveReplaceNodeTargetByInstanceID(ctx context.Context, namespace, instanceID string) (*replaceNodeTarget, error) {
	node, err := s.findNodeByProviderInstanceID(ctx, instanceID)
	if err != nil {
		return nil, err
	}

	var podList corev1.PodList
	if err := s.dataplaneCli.List(ctx, &podList, client.InNamespace(namespace)); err != nil {
		return nil, fmt.Errorf("failed to list pods in namespace %s: %w", namespace, err)
	}

	// An instance ID maps to a node, not directly to a TiDB component. It is safe
	// only when exactly one supported operator pod is on that node.
	var matches []replaceNodeTarget
	for i := range podList.Items {
		pod := podList.Items[i].DeepCopy()
		if pod.Spec.NodeName != node.Name {
			continue
		}
		target, err := buildReplaceNodeTargetFromPod(pod, instanceID)
		if err != nil {
			continue
		}
		target.node = node
		matches = append(matches, *target)
	}

	if len(matches) == 0 {
		return nil, status.Errorf(codes.NotFound,
			"no tidb-operator v2 managed pod found on node %s for instance_id %s", node.Name, instanceID)
	}
	if len(matches) > 1 {
		return nil, status.Errorf(codes.FailedPrecondition,
			"instance_id %s resolves to node %s with multiple tidb-operator v2 managed pods: %s",
			instanceID, node.Name, podNamesFromReplaceNodeTargets(matches))
	}
	return &matches[0], nil
}

func buildReplaceNodeTargetFromPod(pod *corev1.Pod, instanceID string) (*replaceNodeTarget, error) {
	if pod.Spec.NodeName == "" {
		return nil, status.Errorf(codes.FailedPrecondition, "pod %s is not scheduled to a node", pod.Name)
	}

	ownerRef, ok := findTiDBOperatorInstanceOwner(pod)
	if !ok {
		return nil, status.Errorf(codes.FailedPrecondition,
			"pod %s is not owned by a supported tidb-operator v2 per-instance CR", pod.Name)
	}
	return &replaceNodeTarget{
		pod:           pod,
		ownerRef:      ownerRef,
		componentKind: ownerRef.Kind,
		instanceID:    instanceID,
		oldPodUID:     pod.UID,
		oldNode:       pod.Spec.NodeName,
	}, nil
}

func rejectUnreadyReplaceNodeTargetPod(pod *corev1.Pod) error {
	if !isPodReady(pod) {
		return status.Errorf(codes.FailedPrecondition,
			"pod %s must be Running and Ready before PrepareReplaceNode", pod.Name)
	}
	return nil
}

func (s *AgentService) findNodeByProviderInstanceID(ctx context.Context, instanceID string) (*corev1.Node, error) {
	var nodeList corev1.NodeList
	if err := s.dataplaneCli.List(ctx, &nodeList); err != nil {
		return nil, fmt.Errorf("failed to list nodes: %w", err)
	}

	var matches []corev1.Node
	for i := range nodeList.Items {
		node := &nodeList.Items[i]
		providerInstanceID, err := parseProviderInstanceIDFromNode(node)
		if err != nil {
			continue
		}
		if providerInstanceID == instanceID {
			matches = append(matches, *node.DeepCopy())
		}
	}

	if len(matches) == 0 {
		return nil, status.Errorf(codes.NotFound, "node for instance_id %s not found", instanceID)
	}
	if len(matches) > 1 {
		return nil, status.Errorf(codes.FailedPrecondition,
			"instance_id %s matches multiple nodes: %s", instanceID, nodeNames(matches))
	}
	return &matches[0], nil
}

func parseProviderInstanceIDFromNode(node *corev1.Node) (string, error) {
	provider, err := getCloudProviderFromNode(node)
	if err != nil {
		return "", err
	}

	switch provider {
	case "aws":
		_, instanceID, err := parseEKSNodeProviderID(node.Spec.ProviderID)
		return instanceID, err
	case "gcp":
		_, _, instanceID, err := parseGKENodeProviderID(node.Spec.ProviderID)
		return instanceID, err
	case "azure":
		_, _, _, instanceID, err := parseAKSNodeProviderID(node.Spec.ProviderID)
		return instanceID, err
	case "alicloud":
		return parseAlicloudInstanceIDFromNode(node)
	default:
		return "", fmt.Errorf("unsupported provider %s", provider)
	}
}

func findTiDBOperatorInstanceOwner(pod *corev1.Pod) (metav1.OwnerReference, bool) {
	for _, ownerRef := range pod.OwnerReferences {
		gv, err := schema.ParseGroupVersion(ownerRef.APIVersion)
		if err != nil {
			continue
		}
		if gv.Group != CoreGroup || gv.Version != CoreVersion {
			continue
		}
		if _, ok := tidbOperatorInstanceKinds[ownerRef.Kind]; ok {
			return ownerRef, true
		}
	}
	return metav1.OwnerReference{}, false
}

func componentTypeForReplaceNodePod(pod *corev1.Pod) string {
	if componentType := normalizeComponentType(pod.Labels[K8sComponentLabelKey]); componentType != "" {
		return componentType
	}
	if componentType := normalizeComponentType(pod.Labels[PingcapComponentLabelKey]); componentType != "" {
		return componentType
	}
	if ownerRef, ok := findTiDBOperatorInstanceOwner(pod); ok {
		return componentTypeForOwnerKind(ownerRef.Kind)
	}
	return ""
}

func normalizeComponentType(componentType string) string {
	switch componentType {
	case "compute-tiflash":
		componentType = TiFlashComputeComponentType
	case "write-tiflash":
		componentType = TiFlashWriteComponentType
	case "worker-tidb":
		componentType = TiDBWorkerComponentType
	}

	if _, ok := componentMap[componentType]; !ok {
		return ""
	}
	return componentType
}

func componentTypeForOwnerKind(kind string) string {
	switch kind {
	case "PD":
		return PDComponentType
	case "TiKV":
		return TiKVComponentType
	case "TiDB":
		return TiDBComponentType
	case "TiFlash":
		return "tiflash"
	case "TiProxy":
		return TiProxyComponentType
	case "TSO":
		return TSOComponentType
	case "Scheduling":
		return SchedulingComponentType
	case "TiKVWorker":
		return TiKVWorkerComponentType
	default:
		return ""
	}
}

func newTiDBOperatorInstanceObject(namespace string, ownerRef metav1.OwnerReference) client.Object {
	obj := &unstructured.Unstructured{}
	obj.SetAPIVersion(ownerRef.APIVersion)
	obj.SetKind(ownerRef.Kind)
	obj.SetNamespace(namespace)
	obj.SetName(ownerRef.Name)
	return obj
}

func (s *AgentService) rejectLocalSSD(ctx context.Context, pod *corev1.Pod) error {
	usesLocalStorage, err := s.podUsesLocalSSD(ctx, pod)
	if err != nil {
		return err
	}
	if usesLocalStorage {
		return status.Errorf(codes.FailedPrecondition,
			"pod %s uses local-ssd storage and is not supported by ReplaceNode v2", pod.Name)
	}
	return nil
}

func (s *AgentService) podUsesLocalSSD(ctx context.Context, pod *corev1.Pod) (bool, error) {
	for _, volume := range pod.Spec.Volumes {
		if volume.PersistentVolumeClaim == nil {
			continue
		}

		var pvc corev1.PersistentVolumeClaim
		key := client.ObjectKey{Namespace: pod.Namespace, Name: volume.PersistentVolumeClaim.ClaimName}
		if err := s.dataplaneCli.Get(ctx, key, &pvc); err != nil {
			return false, fmt.Errorf("failed to get pvc %s/%s: %w", key.Namespace, key.Name, err)
		}
		if pvc.Spec.StorageClassName != nil && *pvc.Spec.StorageClassName == localSSDStorageClassName {
			return true, nil
		}
	}
	return false, nil
}

func (s *AgentService) preflightWarmUpPod(ctx context.Context, operationID string, target *replaceNodeTarget) error {
	existing, found, err := s.getWarmUpPod(ctx, target.pod.Namespace, operationID)
	if err != nil {
		return err
	}
	if found {
		return validateWarmUpPod(existing, operationID, target)
	}
	if os.Getenv(replaceNodeWarmUpImageEnvKey) == "" {
		return fmt.Errorf("%s is empty", replaceNodeWarmUpImageEnvKey)
	}
	return nil
}

func (s *AgentService) ensureWarmUpPod(ctx context.Context, operationID string, target *replaceNodeTarget) (*corev1.Pod, bool, error) {
	existing, found, err := s.getWarmUpPod(ctx, target.pod.Namespace, operationID)
	if err != nil {
		return nil, false, err
	}
	if found {
		if err := validateWarmUpPod(existing, operationID, target); err != nil {
			return nil, false, err
		}
		return existing, false, nil
	}

	image := os.Getenv(replaceNodeWarmUpImageEnvKey)
	if image == "" {
		return nil, false, fmt.Errorf("%s is empty", replaceNodeWarmUpImageEnvKey)
	}

	warmUpPod := buildWarmUpPod(operationID, image, target)
	if err := s.dataplaneCli.Create(ctx, warmUpPod); err != nil {
		if !apierrors.IsAlreadyExists(err) {
			return nil, false, fmt.Errorf("failed to create warm-up pod %s/%s: %w", warmUpPod.Namespace, warmUpPod.Name, err)
		}
		if err := s.dataplaneCli.Get(ctx, client.ObjectKeyFromObject(warmUpPod), warmUpPod); err != nil {
			return nil, false, fmt.Errorf("failed to get existing warm-up pod %s/%s: %w", warmUpPod.Namespace, warmUpPod.Name, err)
		}
		return warmUpPod, false, validateWarmUpPod(warmUpPod, operationID, target)
	}
	return warmUpPod, true, nil
}

func buildWarmUpPod(operationID, image string, target *replaceNodeTarget) *corev1.Pod {
	automountServiceAccountToken := false
	terminationGracePeriodSeconds := int64(0)

	return &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      warmUpPodName(operationID),
			Namespace: target.pod.Namespace,
			Labels: map[string]string{
				replaceNodeWarmUpLabelKey:      "true",
				replaceNodeOperationIDLabelKey: operationID,
			},
			// Commit/Abort reconstruct prepare-stage state from the warm-up pod
			// so prepared operations survive cluster-agent restarts without memory state.
			Annotations: map[string]string{
				replaceNodeTargetPodNameAnnotationKey:   target.pod.Name,
				replaceNodeTargetPodUIDAnnotationKey:    string(target.oldPodUID),
				replaceNodeOldNodeAnnotationKey:         target.oldNode,
				replaceNodeOwnerAPIVersionAnnotationKey: target.ownerRef.APIVersion,
				replaceNodeOwnerKindAnnotationKey:       target.ownerRef.Kind,
				replaceNodeOwnerNameAnnotationKey:       target.ownerRef.Name,
				replaceNodeInstanceIDAnnotationKey:      target.instanceID,
			},
		},
		Spec: corev1.PodSpec{
			// The warm-up pod only reserves scheduler capacity and never needs API access.
			AutomountServiceAccountToken: &automountServiceAccountToken,
			RestartPolicy:                corev1.RestartPolicyNever,
			// Copy only scheduler inputs needed to prove capacity for this pod class.
			// Do not copy volumes, ports, env, serviceAccount, or topology spread rules.
			NodeSelector:                  copyStringMap(target.pod.Spec.NodeSelector),
			Affinity:                      target.pod.Spec.Affinity.DeepCopy(),
			Tolerations:                   append([]corev1.Toleration(nil), target.pod.Spec.Tolerations...),
			SchedulerName:                 target.pod.Spec.SchedulerName,
			PriorityClassName:             target.pod.Spec.PriorityClassName,
			TerminationGracePeriodSeconds: &terminationGracePeriodSeconds,
			Containers: []corev1.Container{
				{
					Name:      "pause",
					Image:     image,
					Resources: corev1.ResourceRequirements{Requests: schedulerEffectiveRequests(target.pod)},
				},
			},
		},
	}
}

func warmUpPodName(operationID string) string {
	sum := sha256.Sum256([]byte(operationID))
	return "replace-node-warm-up-" + hex.EncodeToString(sum[:])[:12]
}

func replaceNodeStateConfigMapName(operationID string) string {
	sum := sha256.Sum256([]byte(operationID))
	return "replace-node-state-" + hex.EncodeToString(sum[:])[:12]
}

func schedulerEffectiveRequests(pod *corev1.Pod) corev1.ResourceList {
	appRequests := corev1.ResourceList{}
	initRequests := corev1.ResourceList{}

	// Follow kube-scheduler accounting: sum regular containers, take the max
	// init-container request, then add pod overhead.
	for _, container := range pod.Spec.Containers {
		addRequests(appRequests, container.Resources.Requests)
	}
	for _, container := range pod.Spec.InitContainers {
		maxRequests(initRequests, container.Resources.Requests)
	}

	out := corev1.ResourceList{}
	for name, quantity := range appRequests {
		out[name] = quantity.DeepCopy()
	}
	for name, quantity := range initRequests {
		if current, ok := out[name]; !ok || current.Cmp(quantity) < 0 {
			out[name] = quantity.DeepCopy()
		}
	}
	addRequests(out, pod.Spec.Overhead)

	// The floor keeps the warm-up pod meaningful when an existing pod has partial
	// or missing requests; it is intentionally conservative for scheduler proof.
	ensureMinimumRequest(out, corev1.ResourceCPU, resource.MustParse("1"))
	ensureMinimumRequest(out, corev1.ResourceMemory, resource.MustParse("1Gi"))
	return out
}

func addRequests(dst, src corev1.ResourceList) {
	for name, quantity := range src {
		current := dst[name]
		current.Add(quantity)
		dst[name] = current
	}
}

func maxRequests(dst, src corev1.ResourceList) {
	for name, quantity := range src {
		if current, ok := dst[name]; !ok || current.Cmp(quantity) < 0 {
			dst[name] = quantity.DeepCopy()
		}
	}
}

func ensureMinimumRequest(requests corev1.ResourceList, name corev1.ResourceName, minimum resource.Quantity) {
	if current, ok := requests[name]; !ok || current.Cmp(minimum) < 0 {
		requests[name] = minimum
	}
}

func validateWarmUpPod(pod *corev1.Pod, operationID string, target *replaceNodeTarget) error {
	// Deterministic names make retries idempotent, so validate ownership before
	// reusing an existing warm-up pod.
	if pod.Labels[replaceNodeOperationIDLabelKey] != operationID || pod.Labels[replaceNodeWarmUpLabelKey] != "true" {
		return status.Errorf(codes.FailedPrecondition, "warm-up pod %s/%s is not owned by operation %s", pod.Namespace, pod.Name, operationID)
	}
	if target == nil {
		// Commit/Abort can rely on the operation-owned warm-up pod when the
		// original target is expensive or impossible to resolve.
		return nil
	}
	if pod.Annotations[replaceNodeTargetPodUIDAnnotationKey] != "" &&
		pod.Annotations[replaceNodeTargetPodUIDAnnotationKey] != string(target.oldPodUID) {
		return status.Errorf(codes.FailedPrecondition,
			"warm-up pod %s/%s targets a different pod uid", pod.Namespace, pod.Name)
	}
	return nil
}

func (s *AgentService) getWarmUpPod(ctx context.Context, namespace, operationID string) (*corev1.Pod, bool, error) {
	var pod corev1.Pod
	if err := s.dataplaneCli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: warmUpPodName(operationID)}, &pod); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("failed to get warm-up pod: %w", err)
	}
	return &pod, true, nil
}

func (s *AgentService) deleteWarmUpPodIfPresent(ctx context.Context, namespace, operationID string) error {
	warmUpPod, found, err := s.getWarmUpPod(ctx, namespace, operationID)
	if err != nil {
		return err
	}
	if !found {
		return nil
	}
	if err := validateWarmUpPod(warmUpPod, operationID, nil); err != nil {
		return err
	}
	return s.deletePod(ctx, warmUpPod)
}

type replaceNodeOperationState struct {
	OperationID     string
	PodName         string
	OldPodUID       string
	OldNode         string
	WarmUpNode      string
	OwnerAPIVersion string
	ComponentKind   string
	OwnerName       string
	InstanceID      string
	AbortedAt       string
}

func operationStateFromTarget(operationID string, target *replaceNodeTarget) replaceNodeOperationState {
	return replaceNodeOperationState{
		OperationID:     operationID,
		PodName:         target.pod.Name,
		OldPodUID:       string(target.oldPodUID),
		OldNode:         target.oldNode,
		OwnerAPIVersion: target.ownerRef.APIVersion,
		ComponentKind:   target.ownerRef.Kind,
		OwnerName:       target.ownerRef.Name,
		InstanceID:      target.instanceID,
	}
}

func validateReplaceNodeStateMatchesTarget(state replaceNodeOperationState, target *replaceNodeTarget) error {
	if state.PodName == target.pod.Name &&
		state.OldPodUID == string(target.oldPodUID) &&
		state.OldNode == target.oldNode &&
		state.OwnerAPIVersion == target.ownerRef.APIVersion &&
		state.ComponentKind == target.ownerRef.Kind &&
		state.OwnerName == target.ownerRef.Name {
		return nil
	}
	return status.Errorf(codes.FailedPrecondition,
		"replace node operation %s is already bound to pod %s uid %s on node %s and cannot be reused for pod %s uid %s on node %s",
		state.OperationID, state.PodName, state.OldPodUID, state.OldNode, target.pod.Name, target.oldPodUID, target.oldNode)
}

func operationStateFromStateConfigMap(configMap *corev1.ConfigMap) replaceNodeOperationState {
	data := configMap.Data
	return replaceNodeOperationState{
		OperationID:     data[replaceNodeStateOperationIDKey],
		PodName:         data[replaceNodeStatePodNameKey],
		OldPodUID:       data[replaceNodeStateOldPodUIDKey],
		OldNode:         data[replaceNodeStateOldNodeKey],
		WarmUpNode:      data[replaceNodeStateWarmUpNodeKey],
		OwnerAPIVersion: data[replaceNodeStateOwnerAPIVersionKey],
		ComponentKind:   data[replaceNodeStateComponentKindKey],
		OwnerName:       data[replaceNodeStateOwnerNameKey],
		InstanceID:      data[replaceNodeStateInstanceIDKey],
		AbortedAt:       data[replaceNodeStateAbortedAtKey],
	}
}

func (s replaceNodeOperationState) hasOwner() bool {
	return s.OwnerAPIVersion != "" && s.ComponentKind != "" && s.OwnerName != ""
}

func (s replaceNodeOperationState) isAborted() bool {
	return s.AbortedAt != ""
}

func (s replaceNodeOperationState) ownerRef() metav1.OwnerReference {
	return metav1.OwnerReference{
		APIVersion: s.OwnerAPIVersion,
		Kind:       s.ComponentKind,
		Name:       s.OwnerName,
	}
}

func buildReplaceNodeStateConfigMap(namespace string, state replaceNodeOperationState) *corev1.ConfigMap {
	return &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      replaceNodeStateConfigMapName(state.OperationID),
			Namespace: namespace,
			Labels: map[string]string{
				replaceNodeStateLabelKey:       "true",
				replaceNodeOperationIDLabelKey: state.OperationID,
			},
		},
		Data: map[string]string{
			replaceNodeStateOperationIDKey:     state.OperationID,
			replaceNodeStatePodNameKey:         state.PodName,
			replaceNodeStateOldPodUIDKey:       state.OldPodUID,
			replaceNodeStateOldNodeKey:         state.OldNode,
			replaceNodeStateWarmUpNodeKey:      state.WarmUpNode,
			replaceNodeStateOwnerAPIVersionKey: state.OwnerAPIVersion,
			replaceNodeStateComponentKindKey:   state.ComponentKind,
			replaceNodeStateOwnerNameKey:       state.OwnerName,
			replaceNodeStateInstanceIDKey:      state.InstanceID,
			replaceNodeStateAbortedAtKey:       state.AbortedAt,
		},
	}
}

func (s *AgentService) ensureReplaceNodeStateConfigMap(ctx context.Context, namespace string, state replaceNodeOperationState) error {
	if state.OperationID == "" {
		return status.Error(codes.InvalidArgument, "operation_id is required")
	}
	if !state.hasOwner() {
		return status.Errorf(codes.FailedPrecondition, "replace node operation %s is missing operator owner", state.OperationID)
	}

	desired := buildReplaceNodeStateConfigMap(namespace, state)
	var existing corev1.ConfigMap
	if err := s.dataplaneCli.Get(ctx, client.ObjectKeyFromObject(desired), &existing); err != nil {
		if !apierrors.IsNotFound(err) {
			return fmt.Errorf("failed to get replace node state ConfigMap %s/%s: %w", namespace, desired.Name, err)
		}
		if err := s.dataplaneCli.Create(ctx, desired); err != nil {
			if !apierrors.IsAlreadyExists(err) {
				return fmt.Errorf("failed to create replace node state ConfigMap %s/%s: %w", namespace, desired.Name, err)
			}
			if err := s.dataplaneCli.Get(ctx, client.ObjectKeyFromObject(desired), &existing); err != nil {
				return fmt.Errorf("failed to get replace node state ConfigMap %s/%s after create conflict: %w", namespace, desired.Name, err)
			}
		} else {
			return nil
		}
	}

	if existing.Labels[replaceNodeOperationIDLabelKey] != state.OperationID ||
		existing.Labels[replaceNodeStateLabelKey] != "true" {
		return status.Errorf(codes.FailedPrecondition,
			"ConfigMap %s/%s is not owned by replace node operation %s", namespace, existing.Name, state.OperationID)
	}

	original := existing.DeepCopy()
	existing.Labels = desired.Labels
	existing.Data = desired.Data
	if err := s.dataplaneCli.Patch(ctx, &existing, client.MergeFrom(original)); err != nil {
		return fmt.Errorf("failed to patch replace node state ConfigMap %s/%s: %w", namespace, existing.Name, err)
	}
	return nil
}

func (s *AgentService) getReplaceNodeStateConfigMap(ctx context.Context, namespace, operationID string) (*corev1.ConfigMap, bool, error) {
	var configMap corev1.ConfigMap
	key := client.ObjectKey{Namespace: namespace, Name: replaceNodeStateConfigMapName(operationID)}
	if err := s.dataplaneCli.Get(ctx, key, &configMap); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("failed to get replace node state ConfigMap %s/%s: %w", namespace, key.Name, err)
	}
	if configMap.Labels[replaceNodeOperationIDLabelKey] != operationID ||
		configMap.Labels[replaceNodeStateLabelKey] != "true" {
		return nil, false, status.Errorf(codes.FailedPrecondition,
			"ConfigMap %s/%s is not owned by replace node operation %s", namespace, key.Name, operationID)
	}
	return &configMap, true, nil
}

func (s *AgentService) commitOperatorInstanceRestart(ctx context.Context, instance client.Object, operationID string) (bool, error) {
	key := client.ObjectKeyFromObject(instance)
	if err := s.dataplaneCli.Get(ctx, key, instance); err != nil {
		return false, fmt.Errorf("failed to get operator instance %s/%s: %w", key.Namespace, key.Name, err)
	}
	if instance.GetLabels()[replaceNodeOperationIDLabelKey] == operationID {
		return false, nil
	}

	// The op-id label is both the operator reconcile trigger and the same-operation
	// idempotency marker. A retry with the same value must not trigger another restart.
	original := instance.DeepCopyObject().(client.Object)
	labels := copyStringMap(instance.GetLabels())
	labels[replaceNodeOperationIDLabelKey] = operationID
	instance.SetLabels(labels)

	if err := s.dataplaneCli.Patch(ctx, instance, client.MergeFrom(original)); err != nil {
		return false, fmt.Errorf("failed to commit operator instance restart %s/%s: %w", key.Namespace, key.Name, err)
	}
	return true, nil
}

func (s *AgentService) getReplaceNodeOperationState(ctx context.Context, namespace, operationID string) (replaceNodeOperationState, error) {
	configMap, found, err := s.getReplaceNodeStateConfigMap(ctx, namespace, operationID)
	if err != nil {
		return replaceNodeOperationState{}, err
	}
	if !found {
		return replaceNodeOperationState{}, status.Errorf(codes.NotFound,
			"replace node operation %s state not found", operationID)
	}
	state := operationStateFromStateConfigMap(configMap)
	if !state.hasOwner() {
		return replaceNodeOperationState{}, status.Errorf(codes.FailedPrecondition,
			"replace node state %s/%s is missing operator owner", namespace, configMap.Name)
	}
	return state, nil
}

func (s *AgentService) replaceNodeOperationCommitted(ctx context.Context, namespace string, state replaceNodeOperationState) (bool, error) {
	if !state.hasOwner() {
		return false, nil
	}
	instance := newTiDBOperatorInstanceObject(namespace, state.ownerRef())
	if err := s.dataplaneCli.Get(ctx, client.ObjectKeyFromObject(instance), instance); err != nil {
		if apierrors.IsNotFound(err) {
			return false, nil
		}
		return false, fmt.Errorf("failed to get operator instance %s/%s: %w", instance.GetNamespace(), instance.GetName(), err)
	}
	return instance.GetLabels()[replaceNodeOperationIDLabelKey] == state.OperationID, nil
}

func (s *AgentService) buildReplaceNodeOperationResp(ctx context.Context, namespace string, state replaceNodeOperationState) (*apiv1.ReplaceNodeResp, error) {
	if state.isAborted() {
		return replaceNodeAbortResp("replace node operation aborted", state), nil
	}
	committed, err := s.replaceNodeOperationCommitted(ctx, namespace, state)
	if err != nil {
		return nil, err
	}
	if committed {
		return s.buildCommittedReplaceNodeResp(ctx, namespace, state, "replace node operation committed")
	}

	warmUpPod, found, err := s.getWarmUpPod(ctx, namespace, state.OperationID)
	if err != nil {
		return nil, err
	}
	resp := replaceNodeRespFromState("waiting for warm-up pod to run", state)
	resp.Phase = apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_PREPARING
	if !found {
		resp.Message = "waiting for warm-up pod to be created"
		return resp, nil
	}
	if err := validateWarmUpPod(warmUpPod, state.OperationID, nil); err != nil {
		return nil, err
	}
	resp.WarmUpPod = warmUpPod.Name
	resp.WarmUpNode = warmUpPod.Spec.NodeName
	if warmUpPod.Status.Phase == corev1.PodRunning && warmUpPod.Spec.NodeName != "" {
		resp.Phase = apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_PREPARED
		resp.Message = "replace node operation prepared"
	}
	return resp, nil
}

func (s *AgentService) buildCommittedReplaceNodeResp(ctx context.Context, namespace string, state replaceNodeOperationState, message string) (*apiv1.ReplaceNodeResp, error) {
	instance := newTiDBOperatorInstanceObject(namespace, state.ownerRef())
	if err := s.dataplaneCli.Get(ctx, client.ObjectKeyFromObject(instance), instance); err != nil {
		return nil, fmt.Errorf("failed to get operator instance %s/%s: %w", namespace, state.OwnerName, err)
	}

	resp := replaceNodeRespFromState(message, state)
	resp.Phase = apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_COMMITTED

	var replacementPod corev1.Pod
	if state.PodName == "" {
		return resp, nil
	}
	if err := s.dataplaneCli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: state.PodName}, &replacementPod); err != nil {
		if apierrors.IsNotFound(err) {
			resp.Message = "waiting for replacement pod to be created"
			return resp, nil
		}
		return nil, fmt.Errorf("failed to get replacement pod %s/%s: %w", namespace, state.PodName, err)
	}

	resp.ReplacementPod = replacementPod.Name
	resp.ReplacementNode = replacementPod.Spec.NodeName
	// The warm-up pod proves that matching capacity can be made available before
	// commit. The replacement may still land on another suitable node if the
	// autoscaler or kube-scheduler reshuffles capacity, so node mismatch is kept
	// as diagnostic output instead of a business failure.
	if state.OldPodUID != "" && string(replacementPod.UID) == state.OldPodUID {
		resp.Message = "waiting for replacement pod uid to change"
		return resp, nil
	}
	if !isPodReady(&replacementPod) {
		resp.Message = "waiting for replacement pod to become Ready"
		return resp, nil
	}
	if !operatorInstanceReadyAndSynced(instance) {
		resp.Message = "waiting for operator instance Ready/Synced"
		return resp, nil
	}

	resp.Phase = apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_SUCCEEDED
	resp.Message = "replace node succeeded"
	if state.WarmUpNode != "" && replacementPod.Spec.NodeName != "" && replacementPod.Spec.NodeName != state.WarmUpNode {
		resp.Message = fmt.Sprintf("replace node succeeded; replacement pod %s is running on node %s instead of warm-up node %s",
			replacementPod.Name, replacementPod.Spec.NodeName, state.WarmUpNode)
	}
	return resp, nil
}

func replaceNodeRespFromWarmUpPod(warmUpPod *corev1.Pod, target *replaceNodeTarget) *apiv1.ReplaceNodeResp {
	resp := newReplaceNodeResp("waiting for warm-up pod to run", target)
	resp.WarmUpPod = warmUpPod.Name
	resp.WarmUpNode = warmUpPod.Spec.NodeName
	resp.Phase = apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_PREPARING
	if warmUpPod.Status.Phase == corev1.PodRunning && warmUpPod.Spec.NodeName != "" {
		resp.Phase = apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_PREPARED
		resp.Message = "replace node prepared"
	}
	return resp
}

func replaceNodeRespFromState(message string, state replaceNodeOperationState) *apiv1.ReplaceNodeResp {
	return &apiv1.ReplaceNodeResp{
		Message:       message,
		PodName:       state.PodName,
		InstanceId:    state.InstanceID,
		ComponentKind: state.ComponentKind,
		OldNode:       state.OldNode,
		WarmUpPod:     warmUpPodNameFromState(state),
		WarmUpNode:    state.WarmUpNode,
	}
}

func replaceNodeAbortResp(message string, state replaceNodeOperationState) *apiv1.ReplaceNodeResp {
	resp := replaceNodeRespFromState(message, state)
	resp.Phase = apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_ABORTED
	return resp
}

func warmUpPodNameFromState(state replaceNodeOperationState) string {
	if state.OperationID == "" {
		return ""
	}
	return warmUpPodName(state.OperationID)
}

func newReplaceNodeResp(message string, target *replaceNodeTarget) *apiv1.ReplaceNodeResp {
	resp := &apiv1.ReplaceNodeResp{Message: message}
	if target == nil {
		return resp
	}
	resp.PodName = target.pod.Name
	resp.InstanceId = target.instanceID
	resp.ComponentKind = target.componentKind
	resp.OldNode = target.oldNode
	return resp
}

func (s *AgentService) cordonReplaceNode(ctx context.Context, nodeName string) (*corev1.Node, error) {
	var node corev1.Node
	if err := s.dataplaneCli.Get(ctx, client.ObjectKey{Name: nodeName}, &node); err != nil {
		return nil, fmt.Errorf("failed to get node %s: %w", nodeName, err)
	}

	original := node.DeepCopy()
	node.Spec.Unschedulable = true
	if err := s.dataplaneCli.Patch(ctx, &node, client.MergeFrom(original)); err != nil {
		return nil, fmt.Errorf("failed to cordon node %s: %w", node.Name, err)
	}
	return &node, nil
}

func (s *AgentService) uncordonReplaceNode(ctx context.Context, nodeName string) error {
	var node corev1.Node
	if err := s.dataplaneCli.Get(ctx, client.ObjectKey{Name: nodeName}, &node); err != nil {
		return fmt.Errorf("failed to get node %s: %w", nodeName, err)
	}

	original := node.DeepCopy()
	node.Spec.Unschedulable = false
	if err := s.dataplaneCli.Patch(ctx, &node, client.MergeFrom(original)); err != nil {
		return fmt.Errorf("failed to uncordon node %s: %w", node.Name, err)
	}
	return nil
}

func (s *AgentService) deletePod(ctx context.Context, pod *corev1.Pod) error {
	if err := s.dataplaneCli.Delete(ctx, pod); err != nil && !apierrors.IsNotFound(err) {
		return fmt.Errorf("failed to delete pod %s/%s: %w", pod.Namespace, pod.Name, err)
	}
	return nil
}

func isPodReady(pod *corev1.Pod) bool {
	if pod.Status.Phase != corev1.PodRunning {
		return false
	}
	for _, condition := range pod.Status.Conditions {
		if condition.Type == corev1.PodReady {
			return condition.Status == corev1.ConditionTrue
		}
	}
	return false
}

func operatorInstanceReadyAndSynced(instance client.Object) bool {
	unstructuredInstance, ok := instance.(*unstructured.Unstructured)
	if !ok {
		return false
	}
	items, ok, _ := unstructured.NestedSlice(unstructuredInstance.Object, "status", "conditions")
	if !ok {
		return false
	}

	conditions := make([]metav1.Condition, 0, len(items))
	for _, item := range items {
		conditionMap, ok := item.(map[string]any)
		if !ok {
			continue
		}
		var condition metav1.Condition
		if err := runtime.DefaultUnstructuredConverter.FromUnstructured(conditionMap, &condition); err != nil {
			continue
		}
		conditions = append(conditions, condition)
	}
	return apimeta.IsStatusConditionTrue(conditions, opv2.CondReady) &&
		apimeta.IsStatusConditionTrue(conditions, opv2.CondSynced)
}

func podNamesFromReplaceNodeTargets(targets []replaceNodeTarget) string {
	names := make([]string, 0, len(targets))
	for _, target := range targets {
		names = append(names, target.pod.Name)
	}
	sort.Strings(names)
	return fmt.Sprintf("%v", names)
}

func nodeNames(nodes []corev1.Node) string {
	names := make([]string, 0, len(nodes))
	for _, node := range nodes {
		names = append(names, node.Name)
	}
	sort.Strings(names)
	return fmt.Sprintf("%v", names)
}

func copyStringMap(in map[string]string) map[string]string {
	out := make(map[string]string, len(in)+1)
	for k, v := range in {
		out[k] = v
	}
	return out
}