package common

import (
	"context"
	"fmt"
	"runtime/debug"
	"slices"
	"sort"
	"strconv"
	"strings"
	"time"

	awsautoscaling "github.com/crossplane-contrib/provider-aws/apis/autoscaling/v1beta1"
	gcpcontainer "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	opv2 "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"

	azurecontainerservice "github.com/upbound/provider-azure/apis/containerservice/v1beta1"
	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/types"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/yaml"

	"github.com/tidbcloud/eks-provider/controllers/cluster/task"
	"github.com/tidbcloud/eks-provider/controllers/utils/configmanager"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/metrics"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/nodepool"
	"github.com/tidbcloud/infra-provider/pkg/scale"
	"github.com/tidbcloud/infra-provider/pkg/util/crossplane"
)

const unknownValue = "unknown"

const (
	scaleStatusConfigMapNamePrefix = "tidb-cluster-scale-status-"
	scaleStatusDataKey             = "current-scale-status"

	// PVC annotation keys for resource tracking (TidbCluster v1 only, including both status.* and spec.* annotations)
	annoKeyPVCStatusRevision     = "status.tidb.pingcap.com/revision"
	annoKeyPVCStatusStorageClass = "status.tidb.pingcap.com/storage-class"
	annoKeyPVCStatusStorageSize  = "status.tidb.pingcap.com/storage-size"
	annoKeyPVCSpecRevision       = "spec.tidb.pingcap.com/revision"
	annoKeyPVCSpecStorageClass   = "spec.tidb.pingcap.com/storage-class"
	annoKeyPVCSpecStorageSize    = "spec.tidb.pingcap.com/storage-size"

	// minReconcileIntervalSeconds is the minimum interval between two reconcile operations.
	minReconcileIntervalSeconds = 60
)

type ScaleStatusTrack struct {
	*task.BaseTask
}

// Type aliases for backward compatibility
type (
	MemberScaleType            = scale.MemberScaleType
	ScaleStatusTracked         = scale.ScaleStatusTracked
	ScaleChange                = scale.ScaleChange
	MemberConditionType        = scale.MemberConditionType
	MemberStatus               = scale.MemberStatus
	CompMemberStatus           = scale.CompMemberStatus
	StoreStatus                = scale.StoreStatus
	MemberResourceInfo         = scale.MemberResourceInfo
	CompMemberSetting          = scale.CompMemberSetting
	NodeStatus                 = scale.NodeStatus
	AutoScalingGroupListStatus = scale.AutoScalingGroupListStatus
	AutoScalingGroupStatus     = scale.AutoScalingGroupStatus
	PVInfo                     = scale.PVInfo
	PVCDetail                  = scale.PVCDetail
)

// Constants for backward compatibility
const (
	MemberScaleTypeNone          = scale.MemberScaleTypeNone
	MemberScaleTypeScaleUp       = scale.MemberScaleTypeScaleUp
	MemberScaleTypeScaleDown     = scale.MemberScaleTypeScaleDown
	MemberScaleTypeScaleOut      = scale.MemberScaleTypeScaleOut
	MemberScaleTypeScaleIn       = scale.MemberScaleTypeScaleIn
	MemberScaleStorageSizeChange = scale.MemberScaleStorageSizeChange
	MemberPodRunning             = scale.MemberPodRunning
	MemberNodeReady              = scale.MemberNodeReady
	MemberPVCReady               = scale.MemberPVCReady
	MemberStoreOnline            = scale.MemberStoreOnline
)

func NewScaleStatusTrack(ctx *task.Context) *ScaleStatusTrack {
	return &ScaleStatusTrack{
		BaseTask: &task.BaseTask{Context: ctx},
	}
}

func (t *ScaleStatusTrack) Name() string {
	return "ScaleStatusTrack"
}

func (t *ScaleStatusTrack) buildComponentGroupKey(compType cloudv1.ComponentType, groupID string) string {
	if groupID == t.Cluster.GetDefaultGroupID() {
		return string(compType)
	}
	return fmt.Sprintf("extraGroup:%s:%s", string(compType), groupID)
}

// buildWorkflowSummary builds a human-readable summary for active workflow steps.
func (t *ScaleStatusTrack) buildWorkflowSummary(ctx context.Context) string {
	if t.Cluster == nil || t.Cluster.Annotations == nil {
		return ""
	}
	wfData := t.Cluster.Annotations[consts.AnnoKeyScaleClusterWorkflow]
	if wfData == "" {
		return ""
	}

	var wfFromAnno cloudv1.Workflow
	if err := yaml.Unmarshal([]byte(wfData), &wfFromAnno); err != nil {
		return ""
	}

	curWf := &cloudv1.Workflow{}
	if err := t.Local.Get(ctx, types.NamespacedName{Namespace: wfFromAnno.Namespace, Name: wfFromAnno.Name}, curWf); err != nil {
		return ""
	}

	// Collect messages from active steps
	activePhases := map[cloudv1.WorkflowStepPhase]bool{
		cloudv1.WorkflowStepPhaseRunning:      true,
		cloudv1.WorkflowStepPhaseEnding:       true,
		cloudv1.WorkflowStepPhaseCancelEnding: true,
		cloudv1.WorkflowStepPhaseCancelling:   true,
	}

	if len(curWf.Status.Steps) == 0 {
		return ""
	}

	var builder strings.Builder
	for _, step := range curWf.Status.Steps {
		if step == nil || !activePhases[step.Phase] {
			continue
		}

		line := fmt.Sprintf(
			"workflow=%s, type=%s, step=%s, phase=%s, progress=%s",
			curWf.Name,
			curWf.Spec.Type,
			step.Name,
			step.Phase,
			curWf.Status.Progress,
		)
		if step.LastError != "" {
			line = fmt.Sprintf("%s, error=%s", line, step.LastError)
		}

		builder.WriteString(line)
		builder.WriteString("\n")
	}

	return strings.TrimSpace(builder.String())
}

func (t *ScaleStatusTrack) computeClusterMsg(ctx context.Context, curScaleStatus *ScaleStatusTracked) {
	clusterMsg := make(map[string][]string)
	allComponents := t.Cluster.GetAllComponentsV2()
	const (
		memberPrefix    = "Member"
		nodeGroupPrefix = "NodeGroup"
		ClusterPrefix   = "Cluster"
		WorkflowPrefix  = "Workflow"
	)

	makeKey := func(prefix, key string) string {
		return fmt.Sprintf("%s:%s", prefix, key)
	}
	appendMsg := func(prefix, key, format string, args ...interface{}) {
		k := makeKey(prefix, key)
		clusterMsg[k] = append(clusterMsg[k], fmt.Sprintf(format, args...))
	}

	for _, comp := range allComponents {
		compType := comp.GetComponentType()
		compGroupKey := t.buildComponentGroupKey(compType, comp.GetGroupId())
		compMemberStatus, ok := curScaleStatus.ComponentMemberStatus[compGroupKey]
		if !ok {
			continue
		}
		if compMemberStatus.ComponentStatus != cloudv1.ComponentNormal ||
			compMemberStatus.InstanceTypeScalePhase != string(cloudv1.InstanceTypeScalePhaseNone) ||
			len(compMemberStatus.NotNormalMemberStatus) != 0 {
			appendMsg(
				memberPrefix,
				compGroupKey,
				"component status: %s, scale type: %s",
				compMemberStatus.ComponentStatus,
				compMemberStatus.MemberScaleType,
			)
		}
		for _, member := range compMemberStatus.NotNormalMemberStatus {
			clusterKey := makeKey(memberPrefix, compGroupKey)
			if member.Zone != "" && member.PVInfo != nil {
				for _, pvInfo := range member.PVInfo {
					if pvInfo.PV != nil && pvInfo.PV.Zone != "" && pvInfo.PV.Zone != member.Zone {
						reason := fmt.Sprintf("member %s zone mismatch: member zone=%s, pv zone=%s", member.Name, member.Zone, pvInfo.PV.Zone)
						clusterMsg[clusterKey] = append(clusterMsg[clusterKey], reason)
					}
					if pvc := pvInfo.PVC; pvc != nil &&
						(pvc.SpecAnnoRevision != pvc.StatusAnnoRevision ||
							pvc.SpecAnnoStorageClass != pvc.StatusAnnoStorageClass ||
							pvc.SpecAnnoStorageSize != pvc.StatusAnnoStorageSize) {
						reason := fmt.Sprintf(
							"member %s pvc annotation mismatch: revision %s/%s, storageClass %s/%s, storageSize %s/%s",
							member.Name,
							pvc.SpecAnnoRevision, pvc.StatusAnnoRevision,
							pvc.SpecAnnoStorageClass, pvc.StatusAnnoStorageClass,
							pvc.SpecAnnoStorageSize, pvc.StatusAnnoStorageSize,
						)
						clusterMsg[clusterKey] = append(clusterMsg[clusterKey], reason)
					}
				}
			}
			clusterMsg[clusterKey] = append(clusterMsg[clusterKey],
				fmt.Sprintf("member %s reason: %s", member.Name, member.Reason))
		}
	}
	asgList := curScaleStatus.AutoScalingGroupListStatus
	for _, asg := range asgList.NotNormalGroupList {
		notNormalASG, ok := asgList.AutoScalingGroupStatus[asg]
		if !ok {
			continue
		}
		for _, node := range notNormalASG.NotNormalNodeStatus {
			appendMsg(
				nodeGroupPrefix,
				asg,
				"node group status: %s, node instance id: %s, node instance type: %s, node health status: %s, node lifecycle state: %s",
				notNormalASG.Status,
				node.InstanceID,
				node.InstanceType,
				node.HealthStatus,
				node.LifecycleState,
			)
		}
	}
	if len(clusterMsg) == 0 {
		clusterMsg[ClusterPrefix] = []string{"all components and node groups are normal"}
	}
	workflowSummary := t.buildWorkflowSummary(ctx)
	if workflowSummary != "" {
		clusterMsg[WorkflowPrefix] = []string{workflowSummary}
	}
	curScaleStatus.AClusterMsg = clusterMsg
}

// handlePanic safely handles panics and records metrics.
func (t *ScaleStatusTrack) handlePanic(methodName string, result *deploy.Result) {
	if err := recover(); err != nil {
		stack := debug.Stack()
		t.Logger.Error(nil, fmt.Sprintf("panic in ScaleStatusTrack.%s", methodName), "panic", err, "stack", string(stack))
		region := unknownValue
		if t.Context != nil && t.Context.Config.Network.Region != "" {
			region = t.Context.Config.Network.Region
		}
		metrics.ControllerPanic.WithLabelValues("cluster", region).Inc()
		*result = deploy.Error(fmt.Errorf("task panic: %v\n%s", err, stack))
	}
}

func (t *ScaleStatusTrack) Create(ctx context.Context) (result deploy.Result) {

	if t.Config.DryRun.Enabled || !t.Cluster.FeatureEnabled(consts.ClusterFeatureScaleStatusTrack) || t.Cluster.Suspending() || t.Cluster.Suspended() {
		if t.Cluster.Suspending() {
			t.Logger.Info("scale status track suspending", "cluster", t.Cluster.Name)
		} else if t.Cluster.Suspended() {
			t.Logger.Info("scale status track suspended", "cluster", t.Cluster.Name, "condition", t.Cluster.GetSuspendCondition())
		}
		return deploy.Done()
	}

	defer t.handlePanic("Create", &result)

	lastScaleStatus := ScaleStatusTracked{}
	lastScaleStatusCM := &corev1.ConfigMap{}
	cmName := scaleStatusConfigMapNamePrefix + t.Cluster.Name
	cmKey := client.ObjectKey{
		Namespace: t.Cluster.Namespace,
		Name:      cmName,
	}
	if err := t.Local.Get(ctx, cmKey, lastScaleStatusCM); err != nil {
		if client.IgnoreNotFound(err) != nil {
			t.Logger.Error(err, "failed to get last scale status configmap")
			return deploy.Error(err)
		}
	}

	if lastScaleStatusCM.Data != nil && len(lastScaleStatusCM.Data[scaleStatusDataKey]) > 0 {
		if err := yaml.Unmarshal([]byte(lastScaleStatusCM.Data[scaleStatusDataKey]), &lastScaleStatus); err != nil {
			t.Logger.Error(err, "failed to unmarshal last scale status")
		}
	}

	// to avoid frequent update of scale status during scale
	if !t.Cluster.FullSynced() && !lastScaleStatus.LastProbeTime.IsZero() && time.Now().Before(lastScaleStatus.LastProbeTime.Add(time.Duration(minReconcileIntervalSeconds)*time.Second)) {
		t.Logger.Info("skip scale status track, next reconcile time is: ", "next reconcile time", lastScaleStatus.LastProbeTime.Add(time.Duration(minReconcileIntervalSeconds)*time.Second))
		return deploy.Done()
	}
	curScaleStatus := ScaleStatusTracked{
		ClusterId:                   t.Cluster.Name,
		Namespace:                   t.Cluster.Namespace,
		TenantId:                    t.Cluster.GetTenantID(),
		ProjectId:                   t.Cluster.GetProjectID(),
		LastProbeTime:               time.Now(),
		ComponentMemberStatus:       make(map[string]CompMemberStatus),
		ComponentScaleChange:        make(map[string]ScaleChange),
		AClusterMsg:                 make(map[string][]string),
		ClusterFullSynced:           t.Cluster.FullSynced(),
		LastUpdateTimeFromTiDBCloud: ptr.Deref(t.TidbClusterLastUpdateTime, time.Time{}),
	}
	allComponents := t.Cluster.GetAllComponentsV2()
	for _, comp := range allComponents {
		compType := comp.GetComponentType()
		compGroupKey := t.buildComponentGroupKey(compType, comp.GetGroupId())
		lastCompScaleChange := lastScaleStatus.ComponentScaleChange[compGroupKey]
		lastCompMemberStatus, ok := lastScaleStatus.ComponentMemberStatus[compGroupKey]
		lastScaleType := MemberScaleTypeNone
		if ok {
			lastScaleType = lastCompMemberStatus.MemberScaleType
		}
		// refresh scale change
		scaleChange := t.refreshScaleChange(ctx, comp, lastCompScaleChange)
		curScaleStatus.ComponentScaleChange[compGroupKey] = scaleChange

		memberScaleType, err := t.getMemberScaleType(scaleChange, lastScaleType)
		if err != nil {
			t.Logger.Error(err, "failed to get member scale type")
			return deploy.Error(fmt.Errorf("failed to get member scale type: %w", err))
		}
		// get comp instance status
		if curCompMemberStatus, err := t.getCompMemberStatus(ctx, comp, compType, memberScaleType); err != nil {
			t.Logger.Error(err, "failed to get comp instance status")
			return deploy.Error(fmt.Errorf("failed to get comp instance status: %w", err))
		} else {
			if curCompMemberStatus.ComponentStatus != cloudv1.ComponentNormal ||
				curCompMemberStatus.InstanceTypeScalePhase != string(cloudv1.InstanceTypeScalePhaseNone) {
				curCompMemberStatus.MemberScaleType = memberScaleType
				curScaleStatus.ComponentMemberStatus[compGroupKey] = curCompMemberStatus
			}
		}
	}
	// get auto scaling group status
	if asgInfoMap, err := t.getAutoScalingGroupStatus(ctx); err != nil {
		t.Logger.Error(err, "failed to get auto scaling group info")
		return deploy.Error(fmt.Errorf("failed to get auto scaling group info: %w", err))
	} else {
		curScaleStatus.AutoScalingGroupListStatus = asgInfoMap
	}
	t.computeClusterMsg(ctx, &curScaleStatus)
	curScaleStatus.IsSynced = t.Cluster.FullSynced() && len(curScaleStatus.AutoScalingGroupListStatus.NotNormalGroupList) == 0

	yamlData, err := yaml.Marshal(curScaleStatus)
	if err != nil {
		t.Logger.Error(err, "failed to marshal scale status")
		return deploy.Error(fmt.Errorf("failed to marshal scale status: %w", err))
	}
	cm := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: t.Cluster.Namespace,
			Name:      cmName,
			Labels:    t.Cluster.Labels,
		},
	}
	cmData := map[string]string{}
	cmData[scaleStatusDataKey] = string(yamlData)
	if _, err := controllerutil.CreateOrUpdate(ctx, t.Local, cm, func() error {
		cm.Data = cmData
		return nil
	}); err != nil {
		t.Logger.Error(err, "failed to create or update scale status configmap")
		return deploy.Error(err)
	}

	if !curScaleStatus.IsSynced {
		t.Logger.Info("scale status track not synced", "cluster", t.Cluster.Name, "not normal group list", curScaleStatus.AutoScalingGroupListStatus.NotNormalGroupList)
		return deploy.Done().WithRequeuAfter(time.Duration(minReconcileIntervalSeconds*5) * time.Second)
	}
	t.Logger.Info("scale status track done", "cluster", t.Cluster.Name)
	return deploy.Done()
}

func (t *ScaleStatusTrack) Delete(ctx context.Context) (result deploy.Result) {
	if t.Config.DryRun.Enabled || !t.Cluster.FeatureEnabled(consts.ClusterFeatureScaleStatusTrack) {
		return deploy.Done()
	}

	defer t.handlePanic("Delete", &result)

	cmName := scaleStatusConfigMapNamePrefix + t.Cluster.Name
	cm := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: t.Cluster.Namespace,
			Name:      cmName,
		},
	}

	return deploy.ErrorOrDone(k8s.DeleteObject(ctx, t.Local, cm))
}

func (t *ScaleStatusTrack) GetMemberCurrentResourceInfo(ctx context.Context, comp cloudv1.ClusterComp) MemberResourceInfo {
	tcGetter, err := t.GetTCGetter(ctx, comp.GetGroupId())
	if err != nil {
		return MemberResourceInfo{}
	}
	memberResourceInfo := MemberResourceInfo{}
	desiredInstanceType := comp.GetInstanceType()
	curInstanceType := comp.GetInstanceType()
	for _, instanceType := range comp.GetInstanceTypes() {
		if instanceType == "" || instanceType == desiredInstanceType {
			continue
		}
		curInstanceType = instanceType
		break
	}
	memberResourceInfo.InstanceType = curInstanceType
	compType := comp.GetComponentType()
	memberResourceInfo.Replicas = tcGetter.ComponentReadyReplicas(compType)
	if comp.GetVolumeStatus() != nil {
		memberResourceInfo.StorageSize = comp.GetVolumeStatus().CurrentCapacity.String()
		memberResourceInfo.StorageClassName = comp.GetVolumeStatus().CurrentStorageClass
	}
	return memberResourceInfo
}

func (t *ScaleStatusTrack) refreshScaleChange(ctx context.Context, comp cloudv1.ClusterComp, lastCompScaleChange ScaleChange) ScaleChange {
	targetStorageSize, use := comp.GetStorageSize()
	if !use {
		targetStorageSize = nil
	}
	targetStorageClassName, use := comp.GetStorageClassName()
	if !use {
		targetStorageClassName = ""
	}

	storageSizeStr := ""
	if targetStorageSize != nil {
		storageSizeStr = targetStorageSize.String()
	}
	scaleChange := ScaleChange{
		SpecResourceInfo: MemberResourceInfo{
			Replicas:         int32(comp.GetReplicas()),
			InstanceType:     comp.GetInstanceType(),
			StorageSize:      storageSizeStr,
			StorageClassName: targetStorageClassName,
			ConfigVersion:    comp.GetConfigVersion(),
		},
	}
	if t.Cluster.FullSynced() {
		scaleChange.StatusResourceInfo = scaleChange.SpecResourceInfo
	} else {
		curScaleChange := t.GetMemberCurrentResourceInfo(ctx, comp)
		curScaleChange.ConfigVersion = lastCompScaleChange.SpecResourceInfo.ConfigVersion
		scaleChange.StatusResourceInfo = curScaleChange
	}
	return scaleChange
}

func (t *ScaleStatusTrack) getCompMemberStatus(ctx context.Context, comp cloudv1.ClusterComp, compType cloudv1.ComponentType, memberScaleType MemberScaleType) (CompMemberStatus, error) {
	curCompScaleStatus := CompMemberStatus{
		GroupId:                comp.GetGroupId(),
		ComponentStatus:        comp.GetComponentStatus(),
		InstanceTypeScalePhase: string(comp.GetInstanceTypeScalePhase()),
		ComponentType:          compType,
		CompMemberSetting:      t.getCompMemberSetting(ctx, comp, compType),
		Replicas:               int32(comp.GetReplicas()),
		UpdatedReplicas:        comp.GetUpdatedReplicas(),
		Condition:              comp.GetConditions(),
	}
	notNormalMemberStatus, err := t.getNotNormalMemberStatus(ctx, comp, memberScaleType)
	if err != nil {
		return CompMemberStatus{}, fmt.Errorf("failed to get member status: %w", err)
	}

	curCompScaleStatus.NotNormalMemberStatus = notNormalMemberStatus

	return curCompScaleStatus, nil
}

// get component config related to scale
func (t *ScaleStatusTrack) getCompMemberSetting(ctx context.Context, comp cloudv1.ClusterComp, compType cloudv1.ComponentType) CompMemberSetting {
	memberSetting := CompMemberSetting{}
	if t.Cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		switch compType {
		case cloudv1.ComponentPD:
			// config.TerminationGracePeriodSeconds = t.Config.ClusterController.TiDBPreStopHookSleepSeconds
		case cloudv1.ComponentTiDB:
			memberSetting.PreStopHookSleepSeconds = t.Config.ClusterController.GetTiDBPreStopHookSleepSeconds()
			memberSetting.MinReadySeconds = t.Cluster.GetMinReadySeconds(ctx, comp.GetGroupId(), cloudv1.ComponentTiDB)
		case cloudv1.ComponentTiKV:

		case cloudv1.ComponentTiFlash:
			memberSetting.GracefulShutdownSleepSeconds = t.Cluster.GetTiFlashGracefulShutdownSleepSeconds(ctx, consts.DefaultTiFlashGracefulShutdownSleepSeconds)
		case cloudv1.ComponentTiProxy:
			memberSetting.MinReadySeconds = t.Cluster.GetMinReadySeconds(ctx, comp.GetGroupId(), cloudv1.ComponentTiProxy)

		}
	} else {
		tcGetter, err := t.GetTCGetter(ctx, comp.GetGroupId())
		if err != nil {
			return memberSetting
		}
		switch compType {
		case cloudv1.ComponentPD:
			memberSetting.TerminationGracePeriodSeconds = ptr.Deref(tcGetter.ComponentTerminationGracePeriodSeconds(compType), 0)
		case cloudv1.ComponentTiDB:
			memberSetting.TerminationGracePeriodSeconds = ptr.Deref(tcGetter.ComponentTerminationGracePeriodSeconds(compType), 0)
			memberSetting.MinReadySeconds = t.Cluster.GetMinReadySeconds(ctx, comp.GetGroupId(), cloudv1.ComponentTiDB)
			memberSetting.PreStopHookSleepSeconds = t.Config.ClusterController.GetTiDBPreStopHookSleepSeconds()

		case cloudv1.ComponentTiKV:
			memberSetting.TerminationGracePeriodSeconds = ptr.Deref(tcGetter.ComponentTerminationGracePeriodSeconds(compType), 0)
		case cloudv1.ComponentTiFlash:
			memberSetting.TerminationGracePeriodSeconds = ptr.Deref(tcGetter.ComponentTerminationGracePeriodSeconds(compType), 0)
			memberSetting.GracefulShutdownSleepSeconds = t.Cluster.GetTiFlashGracefulShutdownSleepSeconds(ctx, consts.DefaultTiFlashGracefulShutdownSleepSeconds)
		case cloudv1.ComponentTiProxy:
			memberSetting.TerminationGracePeriodSeconds = ptr.Deref(tcGetter.ComponentTerminationGracePeriodSeconds(compType), 0)
			memberSetting.MinReadySeconds = t.Cluster.GetMinReadySeconds(ctx, comp.GetGroupId(), cloudv1.ComponentTiProxy)
		}
	}
	return memberSetting
}

func (t *ScaleStatusTrack) checkIfPodNormal(pod *corev1.Pod) bool {
	// TODO: consider upgrade scenario
	if pod == nil {
		return true
	}
	// Check if pod is being deleted
	if pod.DeletionTimestamp != nil {
		return false
	}
	// Check pod phase
	if pod.Status.Phase != corev1.PodRunning {
		return false
	}
	// Check if pod is ready
	if !k8s.IsPodReady(pod) {
		return false
	}

	return true
}

func (t *ScaleStatusTrack) computeMemberConditions(compType cloudv1.ComponentType, podInfo *corev1.Pod, dbStore *StoreStatus, node *corev1.Node, pvc []PVInfo, isExtra bool) ([]metav1.Condition, string) {
	var conditions []metav1.Condition
	var reasonBuilder strings.Builder
	if podInfo != nil {
		// Check all conditions and collect non-ready ones for detailed reporting
		podReady := t.checkIfPodNormal(podInfo)
		if !podReady {
			for _, cond := range podInfo.Status.Conditions {
				if cond.Status != corev1.ConditionTrue {

					conditions = append(conditions, metav1.Condition{
						Type:               fmt.Sprintf("%s-%s", string(MemberPodRunning), cond.Type),
						Status:             metav1.ConditionFalse,
						Reason:             cond.Reason,
						Message:            fmt.Sprintf("Pod %s message is: %s", podInfo.Name, cond.Message),
						LastTransitionTime: cond.LastTransitionTime,
					})
				}
			}
			reasonBuilder.WriteString(fmt.Sprintf("Pod %s is not ready, pod status is: %s. ", podInfo.Name, podInfo.Status.Phase))
			if podInfo.Status.Reason != "" {
				reasonBuilder.WriteString(fmt.Sprintf("Pod reason is: %s.", podInfo.Status.Reason))
			}
		} else {
			conditions = append(conditions, metav1.Condition{
				Type:               string(MemberPodRunning),
				Status:             metav1.ConditionTrue,
				Reason:             "PodRunning",
				Message:            fmt.Sprintf("Pod %s status is: %s", podInfo.Name, podInfo.Status.Phase),
				LastTransitionTime: metav1.NewTime(time.Now()),
			})
		}
	}
	if dbStore != nil && (compType == cloudv1.ComponentTiKV || compType == cloudv1.ComponentTiFlash) {
		storeOnline := true
		if dbStore.Status != opv1alpha1.TiKVStateUp {
			storeOnline = false
		}
		if !storeOnline {
			leaderCountBeforeUpgrade := ptr.Deref(dbStore.LeaderCountBeforeUpgrade, 0)
			msg := fmt.Sprintf("Store %s status is: %s, leader count: %d, region count: %d", dbStore.Name, dbStore.Status, dbStore.LeaderCount, dbStore.RegionCount)
			if leaderCountBeforeUpgrade > 0 {
				msg = fmt.Sprintf("Store %s status is: %s, leader count: %d, leader count before upgrade: %d, region count: %d", dbStore.Name, dbStore.Status, dbStore.LeaderCount, leaderCountBeforeUpgrade, dbStore.RegionCount)
			}
			conditions = append(conditions, metav1.Condition{
				Type:               string(MemberStoreOnline),
				Status:             metav1.ConditionFalse,
				Reason:             dbStore.Status,
				Message:            msg,
				LastTransitionTime: metav1.NewTime(time.Now()),
			})
			reasonBuilder.WriteString(fmt.Sprintf("Store %s is not online, store status is: %s, leader count: %d, region count: %d.", dbStore.Name, dbStore.Status, dbStore.LeaderCount, dbStore.RegionCount))
		} else {
			conditions = append(conditions, metav1.Condition{
				Type:               string(MemberStoreOnline),
				Status:             metav1.ConditionTrue,
				Reason:             "StoreOnline",
				Message:            fmt.Sprintf("Store %s status is: %s, leader count: %d, region count: %d", dbStore.Name, dbStore.Status, dbStore.LeaderCount, dbStore.RegionCount),
				LastTransitionTime: metav1.NewTime(time.Now()),
			})
		}
	}

	if node != nil {
		nodeReady := true
		var nodeGroup, zone string
		if node.Labels != nil {
			nodeGroup = namer.GetNodeGroupNameFromNode(node.Labels, t.Config.Provider, isExtra)
			zone = namer.GetNodeZoneFromLabels(node.Labels)
		}
		if node.Status.Conditions != nil {
			// check if node is ready, collect all non-ready conditions for detailed reporting
			for _, cond := range node.Status.Conditions {
				if cond.Status != corev1.ConditionTrue {
					nodeReady = false
					conditions = append(conditions, metav1.Condition{
						Type:               fmt.Sprintf("%s-%s", string(MemberNodeReady), cond.Type),
						Status:             metav1.ConditionFalse,
						Reason:             cond.Reason,
						Message:            fmt.Sprintf("Node Group: %s, Zone: %s, Node %s message is: %s", nodeGroup, zone, node.Name, cond.Message),
						LastTransitionTime: cond.LastTransitionTime,
					})
				}
			}
			if !nodeReady {
				reasonBuilder.WriteString(fmt.Sprintf("NodeGroup: %s, Zone: %s, Node %s is not ready.", nodeGroup, zone, node.Name))
			}
		}
		if nodeReady {
			conditions = append(conditions, metav1.Condition{
				Type:               string(MemberNodeReady),
				Status:             metav1.ConditionTrue,
				Reason:             "NodeReady",
				Message:            fmt.Sprintf("Node Group: %s, Zone: %s, Node %s is ready", nodeGroup, zone, node.Name),
				LastTransitionTime: metav1.NewTime(time.Now()),
			})
		}
	}
	for _, pvcInfo := range pvc {
		pvcReady := true
		if pvcInfo.PVC == nil {
			continue
		}
		zone := pvcInfo.PVC.Zone
		if pvcInfo.PVC.Conditions != nil {
			// check if pvc is ready, collect all non-ready conditions for detailed reporting
			for _, cond := range pvcInfo.PVC.Conditions {
				if cond.Status != corev1.ConditionTrue {
					pvcReady = false
					conditions = append(conditions, metav1.Condition{
						Type:               fmt.Sprintf("%s-%s", string(MemberPVCReady), cond.Type),
						Status:             metav1.ConditionFalse,
						Reason:             cond.Reason,
						Message:            fmt.Sprintf("PVC %s, Zone: %s, message is: %s", pvcInfo.PVC.Name, zone, cond.Message),
						LastTransitionTime: cond.LastTransitionTime,
					})
				}
			}
			if !pvcReady {
				reasonBuilder.WriteString(fmt.Sprintf("PVC %s, Zone: %s, is not ready, pvc status is: %s.", pvcInfo.PVC.Name, zone, pvcInfo.PVC.Status))
			}
		}
		if pvcReady {
			conditions = append(conditions, metav1.Condition{
				Type:               string(MemberPVCReady),
				Status:             metav1.ConditionTrue,
				Reason:             "PVCReady",
				Message:            fmt.Sprintf("PVC %s, Zone: %s, status is: %s", pvcInfo.PVC.Name, zone, pvcInfo.PVC.Status),
				LastTransitionTime: metav1.NewTime(time.Now()),
			})
		}
	}

	return conditions, reasonBuilder.String()
}

func (t *ScaleStatusTrack) getNotNormalMemberStatus(ctx context.Context, comp cloudv1.ClusterComp, memberScaleType MemberScaleType) ([]MemberStatus, error) {
	compType := comp.GetComponentType()
	pods, isExtra, err := t.listComponentPods(ctx, comp)
	if err != nil {
		return nil, err
	}

	if !t.Cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		sortPodsByOrdinal(pods)
	}

	isStoreComponent := compType == cloudv1.ComponentTiKV || compType == cloudv1.ComponentTiFlash
	podToDBStoreMap := make(map[string]StoreStatus) // podName -> StoreStatus
	if isStoreComponent {
		// build podName -> StoreStatus map for TiKV/TiFlash
		podToDBStoreMap, err = t.buildPodToDBStoreMap(ctx, compType)
		if err != nil {
			return nil, fmt.Errorf("failed to build pod to db instance map: %w", err)
		}
	}
	// Fetch PV info based on memberScaleType to optimize API calls
	var allPvInfoMap map[string][]PVInfo
	allPvInfoMap, err = t.batchGetPVInfo(ctx, pods, compType)
	if err != nil {
		t.Logger.Error(err, "failed to batch get pv info")
		return nil, err
	}
	abnormalPodsMap := make(map[string]struct{})
	for i := range pods {
		podInfo := &pods[i]
		if isStoreComponent {
			if inst, ok := podToDBStoreMap[podInfo.Name]; ok {
				if inst.Status != opv1alpha1.TiKVStateUp {
					abnormalPodsMap[podInfo.Name] = struct{}{}
					continue
				}
			}
		}
		if memberScaleType == MemberScaleStorageSizeChange {
			if pvInfoList, ok := allPvInfoMap[podInfo.Name]; ok {
				if t.hasPVCAnnotationMismatch(pvInfoList) {
					abnormalPodsMap[podInfo.Name] = struct{}{}
					continue
				}
			}
		}
		if !t.checkIfPodNormal(podInfo) {
			abnormalPodsMap[podInfo.Name] = struct{}{}
			continue
		}
	}

	// always try to include the scaling pod (by ordinal), if it exists
	replicas := int32(comp.GetReplicas())
	updatedReplicas := comp.GetUpdatedReplicas()
	if len(pods) > 0 && replicas > 0 {
		scalingPodIdx := int(replicas - 1 - updatedReplicas)
		if scalingPodIdx >= 0 && scalingPodIdx < len(pods) {
			abnormalPodsMap[pods[scalingPodIdx].Name] = struct{}{}
		}
	}

	if memberScaleType != MemberScaleTypeNone && len(abnormalPodsMap) == 0 && len(pods) > 0 {
		abnormalPodsMap[pods[len(pods)-1].Name] = struct{}{}
	}

	abnormalPods := make([]corev1.Pod, 0, len(abnormalPodsMap))
	for _, pod := range pods {
		if _, ok := abnormalPodsMap[pod.Name]; ok {
			abnormalPods = append(abnormalPods, pod)
		}
	}

	// podName -> Node
	nodeMap, err := t.batchCollectNodeInfo(ctx, abnormalPods)
	if err != nil {
		return nil, fmt.Errorf("failed to batch collect node info: %w", err)
	}

	return t.buildMemberStatusFromPods(abnormalPods, nodeMap, allPvInfoMap, podToDBStoreMap, compType, isExtra), nil
}

func (t *ScaleStatusTrack) listComponentPods(ctx context.Context, comp cloudv1.ClusterComp) ([]corev1.Pod, bool, error) {
	compType := comp.GetComponentType()
	groupID := comp.GetGroupId()
	isExtra := groupID != t.Cluster.GetDefaultGroupID()

	namer := t.Namer.WithComponent(string(compType))
	if isExtra {
		namer = namer.WithTiDBGroupName(groupID)
	}
	pods, err := k8s.SelectPods(ctx, t.EnvProvider.GetClient(), t.Namer.GetTiDBNamespace(), namer.GetPodLabels())
	if err != nil {
		return nil, false, fmt.Errorf("failed to list pods: %w", err)
	}
	return pods, isExtra, nil
}

// sortPodsByOrdinal sorts pods by numeric suffix in pod name, e.g. xxx-<number>.
func sortPodsByOrdinal(pods []corev1.Pod) {
	if len(pods) == 0 {
		return
	}

	parseIndex := func(name string) int {
		parts := strings.Split(name, "-")
		if len(parts) == 0 {
			return 0
		}
		idx, err := strconv.Atoi(parts[len(parts)-1])
		if err != nil {
			return 0
		}
		return idx
	}

	sort.Slice(pods, func(i, j int) bool {
		ai := parseIndex(pods[i].Name)
		bi := parseIndex(pods[j].Name)
		if ai != bi {
			return ai < bi
		}
		return pods[i].Name < pods[j].Name
	})
}

func (t *ScaleStatusTrack) buildMemberStatusFromPods(
	abnormalPods []corev1.Pod,
	nodeMap map[string]*corev1.Node,
	pvInfoMap map[string][]PVInfo,
	podToDBStoreMap map[string]StoreStatus,
	compType cloudv1.ComponentType,
	isExtra bool,
) []MemberStatus {
	notNormalMemberStatus := make([]MemberStatus, 0, len(abnormalPods))

	for _, pod := range abnormalPods {
		memberInfo := MemberStatus{
			Name:          pod.Name,
			PodCreateTime: pod.CreationTimestamp.Time,
		}

		// enrich with node information
		var nodeInfo *corev1.Node
		if n, ok := nodeMap[pod.Name]; ok {
			nodeInfo = n
			memberInfo.NodeName = nodeInfo.Name
			memberInfo.NodeGroup = namer.GetNodeGroupNameFromNode(nodeInfo.Labels, t.Config.Provider, isExtra)
			memberInfo.Zone = namer.GetNodeZoneFromLabels(nodeInfo.Labels)
			memberInfo.InstanceType = namer.GetNodeInstanceTypeFromLabels(nodeInfo.Labels)
		}

		// enrich with PVC/PV information
		// Explicitly check key existence and handle absent PVC/PV information gracefully
		var pvcInfoList []PVInfo
		if pvInfo, ok := pvInfoMap[pod.Name]; ok && len(pvInfo) > 0 {
			memberInfo.PVInfo = pvInfo
			pvcInfoList = pvInfo
		}

		// enrich with store information
		var dbStore *StoreStatus
		if store, ok := podToDBStoreMap[pod.Name]; ok {
			dbStore = &store
			memberInfo.LeaderCount = dbStore.LeaderCount
			memberInfo.LeaderCountBeforeUpgrade = dbStore.LeaderCountBeforeUpgrade
			memberInfo.RegionCount = dbStore.RegionCount
			memberInfo.StoreStartTime = dbStore.StartTime
		}

		memberInfo.MemberConditions, memberInfo.Reason = t.computeMemberConditions(compType, &pod, dbStore, nodeInfo, pvcInfoList, isExtra)
		notNormalMemberStatus = append(notNormalMemberStatus, memberInfo)
	}

	return notNormalMemberStatus
}

// hasPVCAnnotationMismatch checks if any PVC has annotation mismatch between spec and status.
func (t *ScaleStatusTrack) hasPVCAnnotationMismatch(pvInfoList []PVInfo) bool {
	for _, pvInfo := range pvInfoList {
		if pvInfo.PVC == nil {
			continue
		}
		if pvInfo.PVC.StatusAnnoRevision != pvInfo.PVC.SpecAnnoRevision ||
			pvInfo.PVC.StatusAnnoStorageClass != pvInfo.PVC.SpecAnnoStorageClass ||
			pvInfo.PVC.StatusAnnoStorageSize != pvInfo.PVC.SpecAnnoStorageSize {
			return true
		}
	}
	return false
}

func (t *ScaleStatusTrack) buildPodToDBStoreMapV2(ctx context.Context, compType cloudv1.ComponentType, storeIDToStoreInfoMap map[uint64]configmanager.StoreInfo) (map[string]StoreStatus, error) {
	tcV2 := t.TidbClusterV2
	result := make(map[string]StoreStatus)
	if tcV2 == nil {
		return result, nil
	}
	podLabels := t.Namer.WithComponent(string(compType)).GetPodLabels()

	switch compType {
	case cloudv1.ComponentTiKV:
		var tikvList opv2.TiKVList
		ls := client.MatchingLabels(podLabels)
		if err := t.EnvProvider.GetClient().List(ctx, &tikvList, client.InNamespace(tcV2.Namespace), ls); err != nil {
			return nil, fmt.Errorf("failed to list TiKV instances: %w", err)
		}
		for _, tikv := range tikvList.Items {
			podName := tikv.Name
			storeIDStr := tikv.Status.ID
			if storeIDStr == "" {
				continue
			}

			storeID, err := strconv.ParseUint(storeIDStr, 10, 64)
			if err != nil {
				return nil, fmt.Errorf("failed to parse store ID: %w", err)
			}
			storeInfo, ok := storeIDToStoreInfoMap[storeID]
			if !ok {
				return nil, fmt.Errorf("store info not found for store ID: %d", storeID)
			}
			if storeInfo.Store == nil || storeInfo.Status == nil {
				return nil, fmt.Errorf("store info incomplete for store ID: %d", storeID)
			}
			result[podName] = StoreStatus{
				Name:        podName,
				Status:      storeInfo.Store.State.String(),
				LeaderCount: int32(storeInfo.Status.LeaderCount),
				RegionCount: int32(storeInfo.Status.RegionCount),
			}
		}
	case cloudv1.ComponentTiFlash:
		var tiflashList opv2.TiFlashList
		ls := client.MatchingLabels(podLabels)
		if err := t.EnvProvider.GetClient().List(ctx, &tiflashList, client.InNamespace(tcV2.Namespace), ls); err != nil {
			return nil, fmt.Errorf("failed to list TiFlash instances: %w", err)
		}
		for _, tiflash := range tiflashList.Items {
			podName := tiflash.Name
			storeIDStr := tiflash.Status.ID
			if storeIDStr == "" {
				continue
			}
			storeID, err := strconv.ParseUint(storeIDStr, 10, 64)
			if err != nil {
				return nil, fmt.Errorf("failed to parse store ID: %w", err)
			}
			storeInfo, ok := storeIDToStoreInfoMap[storeID]
			if !ok {
				return nil, fmt.Errorf("store info not found for store ID: %d", storeID)
			}
			if storeInfo.Store == nil || storeInfo.Status == nil {
				return nil, fmt.Errorf("store info incomplete for store ID: %d", storeID)
			}
			result[podName] = StoreStatus{
				Name:        podName,
				Status:      storeInfo.Store.State.String(),
				LeaderCount: int32(storeInfo.Status.LeaderCount),
				RegionCount: int32(storeInfo.Status.RegionCount),
			}
		}
	}
	return result, nil
}

func (t *ScaleStatusTrack) buildPodToDBStoreMapV1(ctx context.Context, compType cloudv1.ComponentType, storeIDToStoreInfoMap map[uint64]configmanager.StoreInfo) (map[string]StoreStatus, error) {
	var storeMap map[string]opv1alpha1.TiKVStore
	result := make(map[string]StoreStatus)
	tc := t.Tidbcluster
	if tc == nil {
		return result, nil
	}
	switch compType {
	case cloudv1.ComponentTiKV:
		storeMap = tc.Status.TiKV.Stores
	case cloudv1.ComponentTiFlash:
		storeMap = tc.Status.TiFlash.Stores
	default:
		return result, nil
	}
	for _, storeStatus := range storeMap {
		if storeStatus.ID == "" {
			continue
		}
		storeID, err := strconv.ParseUint(storeStatus.ID, 10, 64)
		if err != nil {
			return nil, fmt.Errorf("failed to parse store ID %q for pod %s: %w", storeStatus.ID, storeStatus.PodName, err)
		}
		storeInfo, ok := storeIDToStoreInfoMap[storeID]
		if !ok {
			return nil, fmt.Errorf("store info not found for store ID: %d", storeID)
		}

		result[storeStatus.PodName] = StoreStatus{
			Name:                     storeStatus.PodName,
			Status:                   storeStatus.State,
			StartTime:                storeInfo.Status.StartTS,
			LeaderCount:              storeStatus.LeaderCount,
			LeaderCountBeforeUpgrade: storeStatus.LeaderCountBeforeUpgrade,
			RegionCount:              int32(storeInfo.Status.RegionCount),
		}
	}
	return result, nil
}

// podName -> DBInfo
func (t *ScaleStatusTrack) buildPodToDBStoreMap(ctx context.Context, compType cloudv1.ComponentType) (map[string]StoreStatus, error) {
	pdClient := t.Context.PDClient
	storeIDToStoreInfoMap := make(map[uint64]configmanager.StoreInfo)
	if pdClient != nil {
		stores, err := pdClient.GetStores(ctx)
		if err != nil {
			t.Logger.Error(err, "failed to get stores")
			return nil, nil
		}
		if stores.Stores != nil {
			for _, store := range stores.Stores {
				if store.Status == nil || store.Store == nil {
					continue
				}
				storeIDToStoreInfoMap[store.Store.Id] = *store
			}
		}
	}
	if t.Cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		return t.buildPodToDBStoreMapV2(ctx, compType, storeIDToStoreInfoMap)
	}
	return t.buildPodToDBStoreMapV1(ctx, compType, storeIDToStoreInfoMap)
}

func (t *ScaleStatusTrack) getPodPvInfo(podName string, pvcMap map[string][]*corev1.PersistentVolumeClaim, pvMap map[string]*corev1.PersistentVolume) []PVInfo {
	pvInfoList := make([]PVInfo, 0)
	pvcList, ok := pvcMap[podName]
	if !ok {
		return pvInfoList
	}

	for _, pvc := range pvcList {
		if pvc == nil {
			continue
		}
		pvcDetail := t.extractPVCDetail(pvc)
		pvInfo := PVInfo{PVC: pvcDetail}

		// Attach PV information if available
		if pvcDetail != nil {
			if pv, ok := pvMap[pvcDetail.Name]; ok && pv != nil {
				pvInfo.PV = t.extractPVDetail(pv)
			}
		}
		pvInfoList = append(pvInfoList, pvInfo)
	}

	return pvInfoList
}

// extractPVCDetail extracts PVC detail information from a PersistentVolumeClaim.
func (t *ScaleStatusTrack) extractPVCDetail(pvc *corev1.PersistentVolumeClaim) *PVCDetail {
	if pvc == nil {
		return nil
	}
	detail := &PVCDetail{
		Name:         pvc.Name,
		Status:       string(pvc.Status.Phase),
		Conditions:   pvc.Status.Conditions,
		StorageClass: ptr.Deref(pvc.Spec.StorageClassName, ""),
	}
	if pvc.Annotations != nil {
		detail.StatusAnnoRevision = pvc.Annotations[annoKeyPVCStatusRevision]
		detail.StatusAnnoStorageClass = pvc.Annotations[annoKeyPVCStatusStorageClass]
		detail.StatusAnnoStorageSize = pvc.Annotations[annoKeyPVCStatusStorageSize]
		detail.SpecAnnoRevision = pvc.Annotations[annoKeyPVCSpecRevision]
		detail.SpecAnnoStorageClass = pvc.Annotations[annoKeyPVCSpecStorageClass]
		detail.SpecAnnoStorageSize = pvc.Annotations[annoKeyPVCSpecStorageSize]
	}
	if pvc.Spec.Resources.Requests != nil {
		detail.StorageSize = pvc.Spec.Resources.Requests.Storage().String()
	}
	return detail
}

// extractPVDetail extracts PV detail information from a PersistentVolume.
// Returns nil if pv is nil to prevent nil pointer dereference.
func (t *ScaleStatusTrack) extractPVDetail(pv *corev1.PersistentVolume) *PVCDetail {
	if pv == nil {
		return nil
	}
	zone := k8s.ReadZoneOfPV(pv)
	resourceID := ""
	if pv.Spec.CSI != nil {
		resourceID = pv.Spec.CSI.VolumeHandle
	}
	return &PVCDetail{
		Name:       pv.Name,
		Status:     string(pv.Status.Phase),
		Zone:       zone,
		ResourceID: resourceID,
	}
}

func (t *ScaleStatusTrack) batchGetPVInfo(ctx context.Context, pods []corev1.Pod, compType cloudv1.ComponentType) (map[string][]PVInfo, error) {
	if len(pods) == 0 {
		return make(map[string][]PVInfo), nil
	}
	pvInfoMap := make(map[string][]PVInfo)
	pvcLabels := t.Namer.WithComponent(string(compType)).GetPVCLabels()
	namespace := pods[0].Namespace
	pvcList, err := k8s.SelectPVCs(ctx, t.EnvProvider.GetClient(), namespace, pvcLabels)

	if err != nil && !apierrors.IsNotFound(err) {
		return nil, fmt.Errorf("failed to list PVCs: %w", err)
	}
	if len(pvcList) == 0 {
		return pvInfoMap, nil
	}
	// pvcMap: podName->pvc
	pvcMap := make(map[string][]*corev1.PersistentVolumeClaim)
	// pvName -> pvcName
	pvNames := make(map[string]string)
	podMap := make(map[string]bool)
	for _, pod := range pods {
		podMap[pod.Name] = true
	}
	for i := range pvcList {
		pvc := &pvcList[i]
		podName := extractPodNameFromPvcName(pvc.Name, pvc.Labels)
		if _, ok := podMap[podName]; !ok {
			continue
		}
		pvcMap[podName] = append(pvcMap[podName], pvc)
		if pvc.Spec.VolumeName != "" {
			pvNames[pvc.Spec.VolumeName] = pvc.Name
		}
	}
	pvList, err := k8s.SelectPVs(ctx, t.EnvProvider.GetClient(), pvcLabels)
	if err != nil && !apierrors.IsNotFound(err) {
		return nil, fmt.Errorf("failed to list PVs: %w", err)
	}
	if len(pvList) == 0 {
		return pvInfoMap, nil
	}
	// pvMap: pvcName->pv
	pvMap := make(map[string]*corev1.PersistentVolume)
	for i := range pvList {
		pv := &pvList[i]
		if pvcName, ok := pvNames[pv.Name]; ok {
			pvMap[pvcName] = pv
		}
	}
	for _, pod := range pods {
		pvInfoMap[pod.Name] = t.getPodPvInfo(pod.Name, pvcMap, pvMap)
	}

	return pvInfoMap, nil
}

// podName -> Node
func (t *ScaleStatusTrack) batchCollectNodeInfo(ctx context.Context, pods []corev1.Pod) (map[string]*corev1.Node, error) {
	nodeMap := make(map[string]*corev1.Node)
	for _, podInfo := range pods {
		nodeName := podInfo.Spec.NodeName
		if nodeName == "" {
			t.Logger.Info("node name is empty, skipping", "pod", podInfo.Name)
			continue
		}
		node := &corev1.Node{}
		if err := t.EnvProvider.GetClient().Get(ctx, client.ObjectKey{Name: nodeName}, node); err != nil {
			if apierrors.IsNotFound(err) {
				t.Logger.Info("node not found, skipping", "node", nodeName, "pod", podInfo.Name)
				continue
			}
			return nil, fmt.Errorf("failed to get node %s for pod %s: %w", nodeName, podInfo.Name, err)
		}

		nodeMap[podInfo.Name] = node
	}

	return nodeMap, nil
}

func (t *ScaleStatusTrack) getAutoScalingGroupStatus(ctx context.Context) (AutoScalingGroupListStatus, error) {
	switch t.Config.Provider {
	case cloudv1.ProviderTypeEKS:
		return t.getEKSNodeGroupStatus(ctx)
	case cloudv1.ProviderTypeGKE:
		return t.getGKENodeGroupStatus(ctx)
	case cloudv1.ProviderTypeAKS:
		return t.getAKSNodeGroupStatus(ctx)
	}
	return AutoScalingGroupListStatus{}, fmt.Errorf("unsupported provider: %s", t.Config.Provider)
}

func (t *ScaleStatusTrack) getEKSNodeGroupStatus(ctx context.Context) (AutoScalingGroupListStatus, error) {
	allComponents := t.Cluster.GetAllComponentsV2()
	asgListStatus := AutoScalingGroupListStatus{
		AutoScalingGroupStatus: make(map[string]AutoScalingGroupStatus),
	}

	asgInfoMap := make(map[string]AutoScalingGroupStatus)
	notNormalASGList := make([]string, 0)
	for _, comp := range allComponents {
		compType := comp.GetComponentType()
		asgList := &awsautoscaling.AutoScalingGroupList{}
		if comp.GetGroupId() != t.Cluster.GetDefaultGroupID() {
			// for extra group
			clusterLabelsWithExtra := t.Namer.WithComponent(string(compType)).WithClusterName(fmt.Sprintf("%s-%s", t.Cluster.Name, comp.GetGroupId())).GetEKSAutoScalingGroupLabels()
			if err := t.Local.List(ctx, asgList, client.MatchingLabels(clusterLabelsWithExtra)); err != nil {
				if apierrors.IsNotFound(err) {
					t.Logger.Info("no EKS node groups found, skipping", "component", compType)
					continue
				}
				return AutoScalingGroupListStatus{}, fmt.Errorf("failed to list EKS node groups: %w", err)
			}
		} else {
			clusterLabels := t.Namer.WithComponent(string(compType)).GetEKSAutoScalingGroupLabels()
			if err := t.Local.List(ctx, asgList, client.MatchingLabels(clusterLabels)); err != nil {
				if apierrors.IsNotFound(err) {
					t.Logger.Info("no EKS node groups found, skipping", "component", compType)
					continue
				}
				return AutoScalingGroupListStatus{}, fmt.Errorf("failed to list EKS node groups: %w", err)
			}
		}

		for _, asg := range asgList.Items {
			asgInfo := AutoScalingGroupStatus{
				Name:            asg.Name,
				Annotations:     t.crucialXPAnnotations(asg.Annotations),
				DesiredCapacity: asg.Spec.ForProvider.DesiredCapacity,
				MinSize:         asg.Spec.ForProvider.MinSize,
				MaxSize:         asg.Spec.ForProvider.MaxSize,
				Conditions:      asg.Status.ConditionedStatus.Conditions,
				Status:          ptr.Deref(asg.Status.AtProvider.Status, ""),
			}

			for _, instance := range asg.Status.AtProvider.Instances {
				if instance.LifecycleState == nil {
					continue
				}
				if *instance.LifecycleState == string(awsautoscaling.LifecycleState_InService) &&
					(instance.HealthStatus != nil && *instance.HealthStatus == "Healthy") {
					continue
				}
				asgInfo.NotNormalNodeStatus = append(asgInfo.NotNormalNodeStatus, NodeStatus{
					InstanceID:     ptr.Deref(instance.InstanceID, ""),
					HealthStatus:   ptr.Deref(instance.HealthStatus, ""),
					LifecycleState: ptr.Deref(instance.LifecycleState, ""),
					InstanceType:   ptr.Deref(instance.InstanceType, ""),
				})
			}
			if !crossplane.IsReadyAndSynced(asg.Status.ConditionedStatus.Conditions) || len(asgInfo.NotNormalNodeStatus) > 0 {
				notNormalASGList = append(notNormalASGList, asgInfo.Name)
			}
			asgInfoMap[asgInfo.Name] = asgInfo
		}

	}
	asgListStatus.AutoScalingGroupStatus = asgInfoMap
	asgListStatus.NotNormalGroupList = notNormalASGList
	asgListStatus.AllGroupListNormal = len(notNormalASGList) == 0

	return asgListStatus, nil
}

func (t *ScaleStatusTrack) getGKENodeGroupStatus(ctx context.Context) (AutoScalingGroupListStatus, error) {
	asgInfoMap := make(map[string]AutoScalingGroupStatus)
	notNormalASGList := make([]string, 0)
	asgListStatus := AutoScalingGroupListStatus{
		AutoScalingGroupStatus: make(map[string]AutoScalingGroupStatus),
	}

	for _, c := range t.Cluster.GetAllComponentSpecs() {
		it := c.GetInstanceType()
		ngName := nodepool.GetGKENodePoolName(t.Cluster, it, t.GetNodeVersion(false))
		np := &gcpcontainer.NodePool{}
		if err := t.Local.Get(ctx, client.ObjectKey{
			Name: ngName,
		}, np); err != nil {
			if apierrors.IsNotFound(err) {
				t.Logger.Info("no GKE node pool found, skipping", "node pool", ngName)
				continue
			}
			return AutoScalingGroupListStatus{}, fmt.Errorf("failed to list GKE node pools: %w", err)
		}

		var minSize, maxSize *int64
		if np.Spec.ForProvider.Autoscaling != nil {
			minSize = np.Spec.ForProvider.Autoscaling.MinNodeCount
			maxSize = np.Spec.ForProvider.Autoscaling.MaxNodeCount
		}
		statusMsg := fmt.Sprintf("%s/%s", np.Status.AtProvider.Status, np.Status.AtProvider.StatusMessage)
		asgInfo := AutoScalingGroupStatus{
			Name:            np.Name,
			Annotations:     t.crucialXPAnnotations(np.Annotations),
			DesiredCapacity: np.Spec.ForProvider.InitialNodeCount,
			MinSize:         minSize,
			MaxSize:         maxSize,
			Conditions:      np.Status.Conditions,
			Status:          statusMsg,
		}

		if !crossplane.IsReadyAndSynced(np.Status.Conditions) {
			if _, ok := asgInfoMap[asgInfo.Name]; !ok {
				notNormalASGList = append(notNormalASGList, asgInfo.Name)
			}
		}
		// Note: GKE NodePool status doesn't provide instance-level information
		asgInfoMap[asgInfo.Name] = asgInfo
	}

	asgListStatus.AutoScalingGroupStatus = asgInfoMap
	asgListStatus.NotNormalGroupList = notNormalASGList
	asgListStatus.AllGroupListNormal = len(notNormalASGList) == 0
	return asgListStatus, nil
}

func (t *ScaleStatusTrack) getAKSNodeGroupStatus(ctx context.Context) (AutoScalingGroupListStatus, error) {
	zones := t.Zones()
	if len(zones) == 0 {
		return AutoScalingGroupListStatus{}, fmt.Errorf("zones is empty")
	}
	nodepoolZones := make([]string, 0, len(zones))
	for _, z := range zones {
		s := strings.Split(z, "-")
		if len(s) < 2 {
			t.Logger.Info("invalid zone format, skipping", "zone", z)
			continue
		}
		nodepoolZones = append(nodepoolZones, s[len(s)-1])
	}
	if len(nodepoolZones) == 0 {
		return AutoScalingGroupListStatus{}, fmt.Errorf("no valid zones found after parsing")
	}
	asgInfoMap := make(map[string]AutoScalingGroupStatus)

	notNormalASGList := make([]string, 0)
	asgListStatus := AutoScalingGroupListStatus{
		AutoScalingGroupStatus: make(map[string]AutoScalingGroupStatus),
	}
	for _, comp := range t.Cluster.GetAllComponentsV2() {
		instType := comp.GetInstanceType()
		kubernetesVersion := t.Cluster.Spec.Kubernetes.Version
		for _, zone := range nodepoolZones {
			nodepoolName := nodepool.GetAKSNodePoolName(instType, t.EnvProvider.GetClusterName(), zone, kubernetesVersion)
			np := &azurecontainerservice.KubernetesClusterNodePool{}
			if err := t.Local.Get(ctx, client.ObjectKey{
				Name: nodepoolName,
			}, np); err != nil {
				if apierrors.IsNotFound(err) {
					t.Logger.Info("no AKS node pool found, skipping", "node pool", nodepoolName)
					continue
				}
				return AutoScalingGroupListStatus{}, fmt.Errorf("failed to list AKS node pools: %w", err)
			}

			asgInfo := AutoScalingGroupStatus{
				Name:        np.Name,
				Annotations: t.crucialXPAnnotations(np.Annotations),
				Conditions:  np.Status.Conditions,
			}

			if np.Spec.ForProvider.NodeCount != nil {
				asgInfo.DesiredCapacity = ptr.To(int64(ptr.Deref(np.Spec.ForProvider.NodeCount, 0)))
			}
			if np.Spec.ForProvider.MinCount != nil {
				asgInfo.MinSize = ptr.To(int64(ptr.Deref(np.Spec.ForProvider.MinCount, 0)))
			}
			if np.Spec.ForProvider.MaxCount != nil {
				asgInfo.MaxSize = ptr.To(int64(ptr.Deref(np.Spec.ForProvider.MaxCount, 0)))
			}

			if !crossplane.IsReadyAndSynced(np.Status.Conditions) {
				if _, ok := asgInfoMap[asgInfo.Name]; !ok {
					notNormalASGList = append(notNormalASGList, asgInfo.Name)
				}
			}
			asgInfoMap[asgInfo.Name] = asgInfo
		}
	}
	asgListStatus.AutoScalingGroupStatus = asgInfoMap
	asgListStatus.NotNormalGroupList = notNormalASGList
	asgListStatus.AllGroupListNormal = len(notNormalASGList) == 0
	return asgListStatus, nil
}

func (t *ScaleStatusTrack) crucialXPAnnotations(asgAnnotations map[string]string) map[string]string {
	keysToFilter := []string{"tidbcloud.com/last-applied-configuration"}
	resMap := make(map[string]string)
	for key, value := range asgAnnotations {
		if !slices.Contains(keysToFilter, key) {
			resMap[key] = value
		}
	}
	return resMap
}

func (t *ScaleStatusTrack) isInstanceTypeScaleUp(oldInstanceType, newInstanceType string) (bool, error) {
	oldObj, err := t.Provider.FindInstanceType(oldInstanceType)
	if err != nil {
		return false, fmt.Errorf("find instance type %s: %w", oldInstanceType, err)
	}
	newObj, err := t.Provider.FindInstanceType(newInstanceType)
	if err != nil {
		return false, fmt.Errorf("find instance type %s: %w", newInstanceType, err)
	}
	return cloudv1.IsScalingUp(oldObj, newObj), nil
}

func (t *ScaleStatusTrack) getMemberScaleType(scaleChange ScaleChange, lastScaleType MemberScaleType) (MemberScaleType, error) {
	curInstanceType := scaleChange.StatusResourceInfo.InstanceType
	desiredInstanceType := scaleChange.SpecResourceInfo.InstanceType
	if curInstanceType != "" && desiredInstanceType != "" && curInstanceType != desiredInstanceType {
		isScaleUp, err := t.isInstanceTypeScaleUp(curInstanceType, desiredInstanceType)
		if err != nil {
			return lastScaleType, err
		}
		if isScaleUp {
			return MemberScaleTypeScaleUp, nil
		} else {
			return MemberScaleTypeScaleDown, nil
		}
	}
	if scaleChange.SpecResourceInfo.StorageSize != scaleChange.StatusResourceInfo.StorageSize || scaleChange.SpecResourceInfo.StorageClassName != scaleChange.StatusResourceInfo.StorageClassName {
		return MemberScaleStorageSizeChange, nil
	}

	if scaleChange.SpecResourceInfo.Replicas > scaleChange.StatusResourceInfo.Replicas {
		return MemberScaleTypeScaleOut, nil
	}
	if scaleChange.SpecResourceInfo.Replicas < scaleChange.StatusResourceInfo.Replicas {
		return MemberScaleTypeScaleIn, nil
	}

	return lastScaleType, nil
}