package controller

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	orcv1 "github.com/cyrus/spark-orchestrator/api/v1alpha1"
	"github.com/cyrus/spark-orchestrator/internal/k8s"
	"github.com/cyrus/spark-orchestrator/internal/util"
)

type ClusterOperationReconciler struct {
	client.Client
	Scheme *runtime.Scheme
}

const (
	sparkNamespaceOwnerKind     = "spark"
	starRocksNamespaceOwnerKind = "starrocks"
)

func (r *ClusterOperationReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx).WithValues("clusterOperation", req.Name)

	var op orcv1.ClusterOperation
	if err := r.Get(ctx, req.NamespacedName, &op); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	if op.Status.Phase == orcv1.OperationPhaseSucceeded || op.Status.Phase == orcv1.OperationPhaseFailed {
		return ctrl.Result{}, nil
	}
	if op.Status.Phase == "" || op.Status.Phase == orcv1.OperationPhasePending {
		now := metav1.Now()
		op.Status.Phase = orcv1.OperationPhaseRunning
		op.Status.StartedAt = &now
		op.Status.Step = "Started"
		op.Status.Message = "Operation accepted"
		setCondition(&op.Status.Conditions, orcv1.NewCondition(orcv1.ConditionRunning, "Started", "Operation is running", metav1.ConditionTrue, op.Generation))
		if err := r.Status().Update(ctx, &op); err != nil {
			return ctrl.Result{}, err
		}
	}

	done, requeueAfter, resourceRef, message, err := r.handleOperation(ctx, &op)
	if err != nil {
		logger.Error(err, "operation execution failed")
		now := metav1.Now()
		op.Status.Phase = orcv1.OperationPhaseFailed
		op.Status.Message = err.Error()
		op.Status.Step = "Failed"
		op.Status.CompletedAt = &now
		op.Status.RetryCount++
		setCondition(&op.Status.Conditions, orcv1.NewCondition(orcv1.ConditionFailed, "ExecutionFailed", err.Error(), metav1.ConditionTrue, op.Generation))
		_ = r.Status().Update(ctx, &op)
		return ctrl.Result{}, nil
	}
	if resourceRef != "" {
		op.Status.ResourceRef = resourceRef
	}
	if message != "" {
		op.Status.Message = message
	}

	if done {
		now := metav1.Now()
		op.Status.Phase = orcv1.OperationPhaseSucceeded
		op.Status.Step = "Completed"
		op.Status.CompletedAt = &now
		setCondition(&op.Status.Conditions, orcv1.NewCondition(orcv1.ConditionReady, "Completed", "Operation completed successfully", metav1.ConditionTrue, op.Generation))
		if err := r.Status().Update(ctx, &op); err != nil {
			return ctrl.Result{}, err
		}
		return ctrl.Result{}, nil
	}

	op.Status.Step = "Running"
	if err := r.Status().Update(ctx, &op); err != nil {
		return ctrl.Result{}, err
	}
	if requeueAfter <= 0 {
		requeueAfter = 5 * time.Second
	}
	return ctrl.Result{RequeueAfter: requeueAfter}, nil
}

func (r *ClusterOperationReconciler) handleOperation(ctx context.Context, op *orcv1.ClusterOperation) (bool, time.Duration, string, string, error) {
	switch op.Spec.Type {
	case orcv1.OperationAddNode:
		return r.handleAddNode(ctx, op)
	case orcv1.OperationDeleteNode:
		return r.handleDeleteNode(ctx, op)
	case orcv1.OperationCreateCluster:
		return r.handleCreateCluster(ctx, op)
	case orcv1.OperationScaleOut:
		return r.handleScaleOut(ctx, op)
	case orcv1.OperationScaleIn:
		return r.handleScaleIn(ctx, op)
	case orcv1.OperationReleaseCluster:
		return r.handleReleaseCluster(ctx, op)
	case orcv1.OperationSubmitSparkJob:
		return r.handleSubmitSparkJob(ctx, op)
	case orcv1.OperationSubmitStarRocksJob:
		return r.handleSubmitStarRocksJob(ctx, op)
	case orcv1.OperationCreateStarRocksCluster:
		return r.handleCreateStarRocksCluster(ctx, op)
	case orcv1.OperationScaleOutStarRocksCluster:
		return r.handleScaleOutStarRocksCluster(ctx, op)
	case orcv1.OperationScaleInStarRocksCluster:
		return r.handleScaleInStarRocksCluster(ctx, op)
	case orcv1.OperationReleaseStarRocksCluster:
		return r.handleReleaseStarRocksCluster(ctx, op)
	default:
		return false, 0, "", "", fmt.Errorf("unsupported operation type %s", op.Spec.Type)
	}
}

func (r *ClusterOperationReconciler) handleAddNode(ctx context.Context, op *orcv1.ClusterOperation) (bool, time.Duration, string, string, error) {
	if op.Spec.AddNode == nil {
		return false, 0, "", "", fmt.Errorf("addNode payload is required")
	}
	nodeName := op.Spec.AddNode.NodeName
	if nodeName == "" {
		return false, 0, "", "", fmt.Errorf("addNode.nodeName is required")
	}

	var nodeInv orcv1.NodeInventory
	err := r.Get(ctx, types.NamespacedName{Name: nodeName}, &nodeInv)
	if apierrors.IsNotFound(err) {
		nodeInv = orcv1.NodeInventory{
			ObjectMeta: metav1.ObjectMeta{
				Name: nodeName,
				Labels: map[string]string{
					util.LabelManagedBy: util.ManagedByValue,
				},
			},
			Spec: orcv1.NodeInventorySpec{
				Address:      op.Spec.AddNode.Address,
				Capacity:     op.Spec.AddNode.Capacity,
				SSHSecretRef: op.Spec.AddNode.SSHSecretRef,
				Bootstrap: orcv1.BootstrapSpec{
					JoinCommandOverride: op.Spec.AddNode.JoinCommandOverride,
				},
			},
		}
		if err := r.Create(ctx, &nodeInv); err != nil {
			return false, 0, "", "", err
		}
		return false, 3 * time.Second, nodeName, "NodeInventory created, waiting for onboarding", nil
	}
	if err != nil {
		return false, 0, "", "", err
	}

	switch nodeInv.Status.Phase {
	case orcv1.NodePhaseIdle:
		return true, 0, nodeName, "Node is idle and ready", nil
	case orcv1.NodePhaseFailed:
		if nodeInv.Status.LastError != "" {
			return false, 8 * time.Second, nodeName, fmt.Sprintf("Node onboarding retrying: %s", nodeInv.Status.LastError), nil
		}
		return false, 8 * time.Second, nodeName, "Node onboarding retrying", nil
	default:
		return false, 5 * time.Second, nodeName, fmt.Sprintf("Node phase=%s", nodeInv.Status.Phase), nil
	}
}

func (r *ClusterOperationReconciler) handleDeleteNode(ctx context.Context, op *orcv1.ClusterOperation) (bool, time.Duration, string, string, error) {
	nodeID := op.Spec.DeleteNode
	if nodeID == "" {
		return false, 0, "", "", fmt.Errorf("deleteNode is required")
	}
	var nodeInv orcv1.NodeInventory
	err := r.Get(ctx, types.NamespacedName{Name: nodeID}, &nodeInv)
	if apierrors.IsNotFound(err) {
		return true, 0, nodeID, "Node already removed", nil
	}
	if err != nil {
		return false, 0, "", "", err
	}
	if err := validateNodeDeletable(&nodeInv); err != nil {
		return false, 0, nodeID, "", err
	}
	nodeName := nodeInv.Status.KubernetesNodeName
	if nodeName == "" {
		resolvedNodeName, err := k8s.FindNodeByInternalIP(ctx, r.Client, nodeInv.Spec.Address)
		if err == nil {
			nodeName = resolvedNodeName
		} else if !apierrors.IsNotFound(err) {
			return false, 0, "", "", err
		}
	}
	if nodeName != "" {
		if err := k8s.ClearNodeIsolation(ctx, r.Client, nodeName); err != nil {
			return false, 0, "", "", err
		}
		if err := k8s.DeleteNodeIfExists(ctx, r.Client, nodeName); err != nil {
			return false, 0, "", "", err
		}
	}
	if err := r.Delete(ctx, &nodeInv); err != nil {
		return false, 0, "", "", err
	}
	return true, 0, nodeID, "Node removed from idle pool and cluster", nil
}

func validateNodeDeletable(nodeInv *orcv1.NodeInventory) error {
	if nodeInv == nil {
		return fmt.Errorf("node inventory is empty")
	}
	phase := nodeInv.Status.Phase
	if phase == orcv1.NodePhaseIdle {
		return nil
	}
	if nodeInv.Status.ClusterRef != "" {
		return fmt.Errorf("node %s is allocated to cluster %s, current phase=%s",
			nodeInv.Name,
			nodeInv.Status.ClusterRef,
			phase)
	}
	switch phase {
	case orcv1.NodePhaseNew, orcv1.NodePhaseProvisioning, orcv1.NodePhaseFailed:
		return nil
	default:
		return fmt.Errorf("node %s is not deletable, current phase=%s", nodeInv.Name, phase)
	}
}

func (r *ClusterOperationReconciler) handleCreateCluster(ctx context.Context, op *orcv1.ClusterOperation) (bool, time.Duration, string, string, error) {
	if op.Spec.Create == nil {
		return false, 0, "", "", fmt.Errorf("create payload is required")
	}
	create := op.Spec.Create
	clusterName := create.ClusterName
	if clusterName == "" {
		return false, 0, "", "", fmt.Errorf("create.clusterName is required")
	}

	var cluster orcv1.SparkVirtualCluster
	err := r.Get(ctx, types.NamespacedName{Name: clusterName}, &cluster)
	if apierrors.IsNotFound(err) {
		cluster = orcv1.SparkVirtualCluster{
			ObjectMeta: metav1.ObjectMeta{
				Name: clusterName,
				Labels: map[string]string{
					util.LabelManagedBy: util.ManagedByValue,
				},
			},
			Spec: orcv1.SparkVirtualClusterSpec{
				Namespace:     create.Namespace,
				NodeRefs:      create.NodeIDs,
				SparkConfig:   create.SparkConfig,
				DesiredState:  orcv1.VirtualClusterDesiredActive,
				ScaleInPolicy: orcv1.ScaleInPolicyBlockIfRunning,
			},
		}
		if err := r.Create(ctx, &cluster); err != nil {
			return false, 0, "", "", err
		}
		return false, 4 * time.Second, clusterName, "Cluster resource created", nil
	}
	if err != nil {
		return false, 0, "", "", err
	}

	if cluster.Status.Phase == orcv1.SparkClusterPhaseFailed {
		return false, 0, clusterName, "", fmt.Errorf("cluster failed: %s", cluster.Status.LastError)
	}
	if cluster.Status.Phase == orcv1.SparkClusterPhaseReady {
		return true, 0, clusterName, "Cluster is ready", nil
	}
	return false, 5 * time.Second, clusterName, fmt.Sprintf("Cluster phase=%s", cluster.Status.Phase), nil
}

func (r *ClusterOperationReconciler) handleScaleOut(ctx context.Context, op *orcv1.ClusterOperation) (bool, time.Duration, string, string, error) {
	if op.Spec.Resize == nil {
		return false, 0, "", "", fmt.Errorf("resize payload is required")
	}
	clusterName := op.Spec.Resize.ClusterName
	if clusterName == "" {
		return false, 0, "", "", fmt.Errorf("resize.clusterName is required")
	}
	var cluster orcv1.SparkVirtualCluster
	if err := r.Get(ctx, types.NamespacedName{Name: clusterName}, &cluster); err != nil {
		return false, 0, "", "", err
	}
	desired := normalizeNodeList(append(cluster.Spec.NodeRefs, op.Spec.Resize.NodeIDs...))
	if !equalNodeLists(normalizeNodeList(cluster.Spec.NodeRefs), desired) {
		cluster.Spec.NodeRefs = desired
		cluster.Spec.DesiredState = orcv1.VirtualClusterDesiredActive
		if err := r.Update(ctx, &cluster); err != nil {
			return false, 0, "", "", err
		}
		return false, 4 * time.Second, clusterName, "Scale-out request applied", nil
	}
	if cluster.Status.Phase == orcv1.SparkClusterPhaseReady {
		for _, nodeID := range op.Spec.Resize.NodeIDs {
			if !util.ContainsString(cluster.Status.AssignedNodes, nodeID) {
				return false, 4 * time.Second, clusterName, "Waiting nodes to be assigned", nil
			}
		}
		return true, 0, clusterName, "Scale-out completed", nil
	}
	return false, 5 * time.Second, clusterName, fmt.Sprintf("Cluster phase=%s", cluster.Status.Phase), nil
}

func (r *ClusterOperationReconciler) handleScaleIn(ctx context.Context, op *orcv1.ClusterOperation) (bool, time.Duration, string, string, error) {
	if op.Spec.Resize == nil {
		return false, 0, "", "", fmt.Errorf("resize payload is required")
	}
	clusterName := op.Spec.Resize.ClusterName
	if clusterName == "" {
		return false, 0, "", "", fmt.Errorf("resize.clusterName is required")
	}
	var cluster orcv1.SparkVirtualCluster
	if err := r.Get(ctx, types.NamespacedName{Name: clusterName}, &cluster); err != nil {
		return false, 0, "", "", err
	}
	if err := r.ensureNoRunningPodsOnNodes(ctx, &cluster, op.Spec.Resize.NodeIDs); err != nil {
		return false, 0, clusterName, "", err
	}
	removeSet := map[string]struct{}{}
	for _, node := range op.Spec.Resize.NodeIDs {
		removeSet[node] = struct{}{}
	}
	desired := util.RemoveStrings(normalizeNodeList(cluster.Spec.NodeRefs), removeSet)
	if !equalNodeLists(normalizeNodeList(cluster.Spec.NodeRefs), desired) {
		cluster.Spec.NodeRefs = desired
		if err := r.Update(ctx, &cluster); err != nil {
			return false, 0, "", "", err
		}
		return false, 4 * time.Second, clusterName, "Scale-in request applied", nil
	}
	if cluster.Status.Phase == orcv1.SparkClusterPhaseReady {
		for _, nodeID := range op.Spec.Resize.NodeIDs {
			if util.ContainsString(cluster.Status.AssignedNodes, nodeID) {
				return false, 4 * time.Second, clusterName, "Waiting nodes to be released", nil
			}
		}
		return true, 0, clusterName, "Scale-in completed", nil
	}
	return false, 5 * time.Second, clusterName, fmt.Sprintf("Cluster phase=%s", cluster.Status.Phase), nil
}

func (r *ClusterOperationReconciler) handleReleaseCluster(ctx context.Context, op *orcv1.ClusterOperation) (bool, time.Duration, string, string, error) {
	if op.Spec.Release == nil {
		return false, 0, "", "", fmt.Errorf("release payload is required")
	}
	clusterName := op.Spec.Release.ClusterName
	if clusterName == "" {
		return false, 0, "", "", fmt.Errorf("release.clusterName is required")
	}
	var cluster orcv1.SparkVirtualCluster
	err := r.Get(ctx, types.NamespacedName{Name: clusterName}, &cluster)
	if apierrors.IsNotFound(err) {
		ns := util.SafeName("spark", clusterName)
		cleanupMessage := r.tryDeleteOwnedNamespace(ctx, ns, sparkNamespaceOwnerKind, clusterName)
		if cleanupMessage != "" {
			return true, 0, clusterName, "Cluster already released; " + cleanupMessage, nil
		}
		return true, 0, clusterName, "Cluster already released", nil
	}
	if err != nil {
		return false, 0, "", "", err
	}

	if err := r.ensureNoRunningPodsOnNodes(ctx, &cluster, normalizeNodeList(cluster.Status.AssignedNodes)); err != nil {
		return false, 0, clusterName, "", err
	}

	// Clean up SparkApplications before releasing the cluster
	ns := cluster.Status.Namespace
	if ns == "" {
		ns = cluster.Spec.Namespace
	}
	if ns != "" {
		if deleted, err := k8s.DeleteSparkApplicationsByCluster(ctx, r.Client, ns, clusterName); err != nil {
			return false, 0, clusterName, "", fmt.Errorf("failed to delete spark applications: %w", err)
		} else if deleted > 0 {
			return false, 2 * time.Second, clusterName, fmt.Sprintf("Deleted %d SparkApplications, proceeding with release", deleted), nil
		}
	}

	if cluster.Spec.DesiredState != orcv1.VirtualClusterDesiredReleased {
		cluster.Spec.DesiredState = orcv1.VirtualClusterDesiredReleased
		if err := r.Update(ctx, &cluster); err != nil {
			return false, 0, "", "", err
		}
		return false, 4 * time.Second, clusterName, "Release requested", nil
	}
	if cluster.Status.Phase != orcv1.SparkClusterPhaseReleased {
		return false, 4 * time.Second, clusterName, fmt.Sprintf("Cluster phase=%s", cluster.Status.Phase), nil
	}
	if err := r.Delete(ctx, &cluster); err != nil && !apierrors.IsNotFound(err) {
		return false, 0, "", "", err
	}
	cleanupMessage := r.tryDeleteOwnedNamespace(ctx, ns, sparkNamespaceOwnerKind, clusterName)
	if cleanupMessage != "" {
		return true, 0, clusterName, "Cluster released; " + cleanupMessage, nil
	}
	return true, 0, clusterName, "Cluster released", nil
}

func (r *ClusterOperationReconciler) handleSubmitSparkJob(ctx context.Context, op *orcv1.ClusterOperation) (bool, time.Duration, string, string, error) {
	if op.Spec.SubmitJob == nil {
		return false, 0, "", "", fmt.Errorf("submitJob payload is required")
	}
	clusterName := op.Spec.SubmitJob.ClusterName
	if clusterName == "" {
		return false, 0, "", "", fmt.Errorf("submitJob.clusterName is required")
	}
	var cluster orcv1.SparkVirtualCluster
	if err := r.Get(ctx, types.NamespacedName{Name: clusterName}, &cluster); err != nil {
		return false, 0, "", "", err
	}
	if cluster.Status.Phase != orcv1.SparkClusterPhaseReady {
		return false, 5 * time.Second, clusterName, fmt.Sprintf("Cluster phase=%s", cluster.Status.Phase), nil
	}

	resource := op.Status.ResourceRef
	if resource == "" {
		appName := util.SafeName("spark-job", op.Name)
		if err := k8s.CreateSparkApplication(ctx, r.Client, &cluster, appName, op.Spec.SubmitJob.SparkApplicationSpec); err != nil {
			if strings.Contains(err.Error(), "already exists") {
				resource = fmt.Sprintf("%s/%s", cluster.Status.Namespace, appName)
				return true, 0, resource, "SparkApplication already exists", nil
			}
			return false, 0, "", "", err
		}
		resource = fmt.Sprintf("%s/%s", cluster.Status.Namespace, appName)
		return true, 0, resource, "SparkApplication submitted", nil
	}
	return true, 0, resource, "SparkApplication submitted", nil
}

func (r *ClusterOperationReconciler) handleSubmitStarRocksJob(ctx context.Context, op *orcv1.ClusterOperation) (bool, time.Duration, string, string, error) {
	if op.Spec.SubmitStarRocks == nil {
		return false, 0, "", "", fmt.Errorf("submitStarRocks payload is required")
	}
	clusterName := op.Spec.SubmitStarRocks.ClusterName
	if clusterName == "" {
		return false, 0, "", "", fmt.Errorf("submitStarRocks.clusterName is required")
	}

	var cluster orcv1.StarRocksVirtualCluster
	if err := r.Get(ctx, types.NamespacedName{Name: clusterName}, &cluster); err != nil {
		return false, 0, "", "", err
	}
	if cluster.Status.Phase != orcv1.StarRocksClusterPhaseReady {
		return false, 5 * time.Second, clusterName, fmt.Sprintf("StarRocks cluster phase=%s", cluster.Status.Phase), nil
	}
	ns := cluster.Status.Namespace
	if ns == "" {
		ns = cluster.Spec.Namespace
	}
	if ns == "" {
		ns = util.SafeName("starrocks", cluster.Name)
	}

	resource := op.Status.ResourceRef
	if resource == "" {
		taskName := util.SafeName("starrocks-task", op.Name)
		if err := k8s.CreateStarRocksTask(ctx, r.Client, &cluster, taskName, op.Spec.SubmitStarRocks.JobSpec); err != nil {
			if strings.Contains(err.Error(), "already exists") {
				resource = fmt.Sprintf("%s/%s", ns, taskName)
				return true, 0, resource, "StarRocks task already exists", nil
			}
			return false, 0, "", "", err
		}
		resource = fmt.Sprintf("%s/%s", ns, taskName)
		return true, 0, resource, "StarRocks task submitted", nil
	}
	return true, 0, resource, "StarRocks task submitted", nil
}

func (r *ClusterOperationReconciler) handleCreateStarRocksCluster(ctx context.Context, op *orcv1.ClusterOperation) (bool, time.Duration, string, string, error) {
	if op.Spec.CreateStarRocks == nil {
		return false, 0, "", "", fmt.Errorf("createStarRocks payload is required")
	}
	create := op.Spec.CreateStarRocks
	clusterName := create.ClusterName
	if clusterName == "" {
		return false, 0, "", "", fmt.Errorf("createStarRocks.clusterName is required")
	}
	if len(create.FeNodeIDs) == 0 {
		return false, 0, "", "", fmt.Errorf("createStarRocks.feNodeIds is required")
	}
	if len(create.BeNodeIDs) == 0 {
		return false, 0, "", "", fmt.Errorf("createStarRocks.beNodeIds is required")
	}

	var cluster orcv1.StarRocksVirtualCluster
	err := r.Get(ctx, types.NamespacedName{Name: clusterName}, &cluster)
	if apierrors.IsNotFound(err) {
		cluster = orcv1.StarRocksVirtualCluster{
			ObjectMeta: metav1.ObjectMeta{
				Name: clusterName,
				Labels: map[string]string{
					util.LabelManagedBy: util.ManagedByValue,
				},
			},
			Spec: orcv1.StarRocksVirtualClusterSpec{
				Namespace:       create.Namespace,
				FeNodeRefs:      create.FeNodeIDs,
				BeNodeRefs:      create.BeNodeIDs,
				DesiredState:    orcv1.VirtualClusterDesiredActive,
				ScaleInPolicy:   orcv1.ScaleInPolicyBlockIfRunning,
				StarRocksConfig: create.StarRocksConfig,
			},
		}
		if err := r.Create(ctx, &cluster); err != nil {
			return false, 0, "", "", err
		}
		return false, 4 * time.Second, clusterName, "StarRocks cluster resource created", nil
	}
	if err != nil {
		return false, 0, "", "", err
	}

	if cluster.Status.Phase == orcv1.StarRocksClusterPhaseFailed {
		return false, 0, clusterName, "", fmt.Errorf("starrocks cluster failed: %s", cluster.Status.LastError)
	}
	if cluster.Status.Phase == orcv1.StarRocksClusterPhaseReady {
		return true, 0, clusterName, "StarRocks cluster is ready", nil
	}
	return false, 5 * time.Second, clusterName, fmt.Sprintf("StarRocks cluster phase=%s", cluster.Status.Phase), nil
}

func (r *ClusterOperationReconciler) handleScaleOutStarRocksCluster(ctx context.Context, op *orcv1.ClusterOperation) (bool, time.Duration, string, string, error) {
	if op.Spec.ResizeStarRocks == nil {
		return false, 0, "", "", fmt.Errorf("resizeStarRocks payload is required")
	}
	clusterName := op.Spec.ResizeStarRocks.ClusterName
	if clusterName == "" {
		return false, 0, "", "", fmt.Errorf("resizeStarRocks.clusterName is required")
	}
	var cluster orcv1.StarRocksVirtualCluster
	if err := r.Get(ctx, types.NamespacedName{Name: clusterName}, &cluster); err != nil {
		return false, 0, "", "", err
	}
	if cluster.Status.Phase == orcv1.StarRocksClusterPhaseFailed {
		return false, 0, clusterName, "", fmt.Errorf("starrocks cluster failed: %s", cluster.Status.LastError)
	}

	desired := normalizeNodeList(append(cluster.Spec.BeNodeRefs, op.Spec.ResizeStarRocks.NodeIDs...))
	if !equalNodeLists(normalizeNodeList(cluster.Spec.BeNodeRefs), desired) {
		cluster.Spec.BeNodeRefs = desired
		cluster.Spec.DesiredState = orcv1.VirtualClusterDesiredActive
		if err := r.Update(ctx, &cluster); err != nil {
			return false, 0, "", "", err
		}
		return false, 4 * time.Second, clusterName, "StarRocks BE scale-out request applied", nil
	}

	if cluster.Status.Phase == orcv1.StarRocksClusterPhaseReady {
		for _, nodeID := range op.Spec.ResizeStarRocks.NodeIDs {
			if !util.ContainsString(cluster.Status.AssignedBeNodes, nodeID) {
				return false, 4 * time.Second, clusterName, "Waiting StarRocks BE nodes to be assigned", nil
			}
		}
		return true, 0, clusterName, "StarRocks BE scale-out completed", nil
	}
	return false, 5 * time.Second, clusterName, fmt.Sprintf("StarRocks cluster phase=%s", cluster.Status.Phase), nil
}

func (r *ClusterOperationReconciler) handleScaleInStarRocksCluster(ctx context.Context, op *orcv1.ClusterOperation) (bool, time.Duration, string, string, error) {
	if op.Spec.ResizeStarRocks == nil {
		return false, 0, "", "", fmt.Errorf("resizeStarRocks payload is required")
	}
	clusterName := op.Spec.ResizeStarRocks.ClusterName
	if clusterName == "" {
		return false, 0, "", "", fmt.Errorf("resizeStarRocks.clusterName is required")
	}
	if len(op.Spec.ResizeStarRocks.NodeIDs) == 0 {
		return false, 0, "", "", fmt.Errorf("resizeStarRocks.nodeIds are required")
	}

	var cluster orcv1.StarRocksVirtualCluster
	if err := r.Get(ctx, types.NamespacedName{Name: clusterName}, &cluster); err != nil {
		return false, 0, "", "", err
	}
	if cluster.Status.Phase == orcv1.StarRocksClusterPhaseFailed {
		return false, 0, clusterName, "", fmt.Errorf("starrocks cluster failed: %s", cluster.Status.LastError)
	}

	ns := cluster.Status.Namespace
	if ns == "" {
		ns = cluster.Spec.Namespace
	}
	if ns == "" {
		ns = util.SafeName("starrocks", cluster.Name)
	}
	for _, nodeID := range op.Spec.ResizeStarRocks.NodeIDs {
		if util.ContainsString(cluster.Spec.BeNodeRefs, nodeID) {
			if err := r.ensureNoRunningPodsOnNodesInNamespace(ctx, ns, []string{nodeID}); err != nil {
				return false, 0, clusterName, "", err
			}
			cluster.Spec.BeNodeRefs = removeNodeFromList(cluster.Spec.BeNodeRefs, nodeID)
			cluster.Spec.DesiredState = orcv1.VirtualClusterDesiredActive
			if err := r.Update(ctx, &cluster); err != nil {
				return false, 0, "", "", err
			}
			return false, 4 * time.Second, clusterName, fmt.Sprintf("StarRocks BE scale-in step applied for node %s", nodeID), nil
		}
		if util.ContainsString(cluster.Status.AssignedBeNodes, nodeID) {
			return false, 4 * time.Second, clusterName, fmt.Sprintf("Waiting StarRocks BE node to be released: %s", nodeID), nil
		}
	}

	if cluster.Status.Phase == orcv1.StarRocksClusterPhaseReady {
		return true, 0, clusterName, "StarRocks BE scale-in completed", nil
	}
	return false, 5 * time.Second, clusterName, fmt.Sprintf("StarRocks cluster phase=%s", cluster.Status.Phase), nil
}

func (r *ClusterOperationReconciler) handleReleaseStarRocksCluster(ctx context.Context, op *orcv1.ClusterOperation) (bool, time.Duration, string, string, error) {
	if op.Spec.ReleaseStarRocks == nil {
		return false, 0, "", "", fmt.Errorf("releaseStarRocks payload is required")
	}
	clusterName := op.Spec.ReleaseStarRocks.ClusterName
	if clusterName == "" {
		return false, 0, "", "", fmt.Errorf("releaseStarRocks.clusterName is required")
	}

	var cluster orcv1.StarRocksVirtualCluster
	err := r.Get(ctx, types.NamespacedName{Name: clusterName}, &cluster)
	if apierrors.IsNotFound(err) {
		ns := util.SafeName("starrocks", clusterName)
		cleanupMessage := r.tryDeleteOwnedNamespace(ctx, ns, starRocksNamespaceOwnerKind, clusterName)
		if cleanupMessage != "" {
			return true, 0, clusterName, "StarRocks cluster already released; " + cleanupMessage, nil
		}
		return true, 0, clusterName, "StarRocks cluster already released", nil
	}
	if err != nil {
		return false, 0, "", "", err
	}

	allAssigned := normalizeNodeList(append(cluster.Status.AssignedFeNodes, cluster.Status.AssignedBeNodes...))
	ns := cluster.Status.Namespace
	if ns == "" {
		ns = cluster.Spec.Namespace
	}
	if ns == "" {
		ns = util.SafeName("starrocks", cluster.Name)
	}
	if err := r.ensureNoRunningPodsOnNodesInNamespace(ctx, ns, allAssigned); err != nil {
		return false, 0, clusterName, "", err
	}

	if cluster.Spec.DesiredState != orcv1.VirtualClusterDesiredReleased {
		cluster.Spec.DesiredState = orcv1.VirtualClusterDesiredReleased
		if err := r.Update(ctx, &cluster); err != nil {
			return false, 0, "", "", err
		}
		return false, 4 * time.Second, clusterName, "StarRocks cluster release requested", nil
	}
	if cluster.Status.Phase != orcv1.StarRocksClusterPhaseReleased {
		return false, 4 * time.Second, clusterName, fmt.Sprintf("StarRocks cluster phase=%s", cluster.Status.Phase), nil
	}
	if err := r.Delete(ctx, &cluster); err != nil && !apierrors.IsNotFound(err) {
		return false, 0, "", "", err
	}
	cleanupMessage := r.tryDeleteOwnedNamespace(ctx, ns, starRocksNamespaceOwnerKind, clusterName)
	if cleanupMessage != "" {
		return true, 0, clusterName, "StarRocks cluster released; " + cleanupMessage, nil
	}
	return true, 0, clusterName, "StarRocks cluster released", nil
}

func (r *ClusterOperationReconciler) tryDeleteOwnedNamespace(
	ctx context.Context,
	namespace string,
	ownerKind string,
	ownerName string,
) string {
	if strings.TrimSpace(namespace) == "" {
		return ""
	}

	var ns corev1.Namespace
	if err := r.Get(ctx, types.NamespacedName{Name: namespace}, &ns); err != nil {
		if apierrors.IsNotFound(err) {
			return "namespace already deleted"
		}
		return fmt.Sprintf("namespace cleanup skipped: %v", err)
	}
	labels := ns.GetLabels()
	if labels[util.LabelManagedBy] != util.ManagedByValue {
		return "namespace cleanup skipped: not managed by orchestrator"
	}
	if labels[util.LabelNamespaceAutoDelete] != "true" {
		return "namespace cleanup skipped: auto-delete disabled"
	}
	if labels[util.LabelNamespaceOwnerKind] != ownerKind || labels[util.LabelNamespaceOwnerName] != ownerName {
		return "namespace cleanup skipped: namespace owner mismatch"
	}
	if ns.DeletionTimestamp != nil {
		return "namespace deletion in progress"
	}
	if err := r.Delete(ctx, &ns); err != nil {
		if apierrors.IsNotFound(err) {
			return "namespace already deleted"
		}
		return fmt.Sprintf("namespace cleanup skipped: %v", err)
	}
	return "namespace deletion requested"
}

func (r *ClusterOperationReconciler) ensureNoRunningPodsOnNodes(ctx context.Context, cluster *orcv1.SparkVirtualCluster, nodeIDs []string) error {
	ns := cluster.Status.Namespace
	if ns == "" {
		ns = cluster.Spec.Namespace
	}
	return r.ensureNoRunningPodsOnNodesInNamespace(ctx, ns, nodeIDs)
}

func (r *ClusterOperationReconciler) ensureNoRunningPodsOnNodesInNamespace(ctx context.Context, namespace string, nodeIDs []string) error {
	if namespace == "" || len(nodeIDs) == 0 {
		return nil
	}

	nodeSet := map[string]struct{}{}
	for _, n := range nodeIDs {
		nodeSet[n] = struct{}{}
	}
	k8sNames := map[string]struct{}{}
	for _, nodeID := range nodeIDs {
		var inv orcv1.NodeInventory
		if err := r.Get(ctx, types.NamespacedName{Name: nodeID}, &inv); err == nil && inv.Status.KubernetesNodeName != "" {
			k8sNames[inv.Status.KubernetesNodeName] = struct{}{}
		}
	}

	var pods corev1.PodList
	if err := r.List(ctx, &pods, client.InNamespace(namespace)); err != nil {
		return err
	}
	blockers := make([]string, 0)
	for i := range pods.Items {
		pod := &pods.Items[i]
		if pod.Spec.NodeName == "" {
			continue
		}
		if pod.Labels != nil && pod.Labels[util.LabelManagedBy] == util.ManagedByValue {
			// Skip managed infrastructure pods: prepull DaemonSets and StarRocks FE/BE pods.
			if pod.Labels[util.LabelComponent] == util.ComponentSparkPrePull {
				continue
			}
			if pod.Labels[util.LabelComponent] == util.ComponentSparkIvyWarmup {
				continue
			}
			if _, hasStarRocksComponent := pod.Labels[util.LabelStarRocksComponent]; hasStarRocksComponent {
				continue
			}
		}
		if _, ok := nodeSet[pod.Spec.NodeName]; !ok {
			if _, ok := k8sNames[pod.Spec.NodeName]; !ok {
				continue
			}
		}
		if pod.Status.Phase == corev1.PodRunning || pod.Status.Phase == corev1.PodPending {
			blockers = append(blockers, pod.Name)
		}
	}
	if len(blockers) > 0 {
		sort.Strings(blockers)
		return errors.New("block-if-running policy violated: running pods found: " + strings.Join(blockers, ","))
	}
	return nil
}

func removeNodeFromList(nodes []string, target string) []string {
	if len(nodes) == 0 {
		return nodes
	}
	out := make([]string, 0, len(nodes))
	for _, node := range nodes {
		if node == target {
			continue
		}
		out = append(out, node)
	}
	return out
}

func (r *ClusterOperationReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&orcv1.ClusterOperation{}).
		Complete(r)
}