package common

import (
	"context"
	"fmt"
	"sort"
	"strings"

	"github.com/go-logr/logr"
	opv2 "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"go.uber.org/multierr"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/eks-provider/controllers/cluster/task/common/feature"
	"github.com/tidbcloud/eks-provider/controllers/cluster/task/common/status"
	"github.com/tidbcloud/eks-provider/utils/tidb-operator/v2utils"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/gitops"
	"github.com/tidbcloud/infra-provider/pkg/namer"
)

type TiDBClusterStatusResolverV2 struct {
	// Provider specific
	client.Client
	gitops.Provider

	// Provider agnostic
	Target       *cloudv1.Cluster
	TiDBCluster  *opv2.Cluster
	PDGroup      *opv2.PDGroup
	TiDBGroup    *opv2.TiDBGroup
	TiKVGroup    *opv2.TiKVGroup
	TiFlashGroup *opv2.TiFlashGroup
	TiProxyGroup *opv2.TiProxyGroup

	Mapper cloudv1.CMEKStorageClassMapperV2

	// Version is current tidb version from config
	Version string
	Gates   feature.Gates

	Log   logr.Logger
	Namer *namer.Namer
}

func (r *TiDBClusterStatusResolverV2) Resolve(ctx context.Context) error {
	var errs error
	tidb, err := r.computeTiDBStatus(ctx)
	if err != nil {
		r.Log.Error(err, "failed to compute TiDB status")
		errs = multierr.Append(errs, err)
	}
	pd, err := r.computePDStatus(ctx)
	if err != nil {
		r.Log.Error(err, "failed to compute PD status")
		errs = multierr.Append(errs, err)
	}
	tikv, err := r.computeTiKVStatus(ctx)
	if err != nil {
		r.Log.Error(err, "failed to compute TiKV status")
		errs = multierr.Append(errs, err)
	}
	tiflash, err := r.computeTiFlashStatus(ctx)
	if err != nil {
		r.Log.Error(err, "failed to compute TiFlash status")
		errs = multierr.Append(errs, err)
	}
	tiproxy, err := r.computeTiProxyStatus(ctx)
	if err != nil {
		r.Log.Error(err, "failed to compute TiProxy status")
		errs = multierr.Append(errs, err)
	}
	if errs != nil {
		return errs
	}

	// Do not overwrite not own field.
	if r.Target.Status.DB != nil && r.Target.Status.DB.TiDB != nil {
		tidb.AuditReady = r.Target.Status.DB.TiDB.AuditReady
	}

	var phase cloudv1.ClusterPhase
	secondaryStatus := cloudv1.DBSecondaryStatusHealthy

	// PD partial failure cannot be felt by user, so the status is hidden
	if tiproxy.Status == cloudv1.ComponentUnhealthy ||
		tikv.Status == cloudv1.ComponentUnhealthy ||
		tidb.Status == cloudv1.ComponentUnhealthy ||
		tiflash.Status == cloudv1.ComponentUnhealthy {
		secondaryStatus = cloudv1.DBSecondaryStatusUnhealthy
	}

	switch {
	// cluster is considered as unavailable when one of TiKV, PD, TiDB, TiProxy is unavailable
	case tidb.Status == cloudv1.ComponentUnavailable || pd.Status == cloudv1.ComponentUnavailable || tikv.Status == cloudv1.ComponentUnavailable || tiproxy.Status == cloudv1.ComponentUnavailable:
		if r.Target.Status.Phase != cloudv1.ClusterPhaseSuspend {
			phase = cloudv1.ClusterPhaseUnavailable
			secondaryStatus = cloudv1.DBSecondaryStatusUnhealthy
		} else {
			phase = cloudv1.ClusterPhaseSuspend
		}
	// if both upgrading and scaling are ongoing (e.g. failover when upgrading), show Upgrading
	case tidb.Status == cloudv1.ComponentUpgrading || tikv.Status == cloudv1.ComponentUpgrading || tiflash.Status == cloudv1.ComponentUpgrading || tiproxy.Status == cloudv1.ComponentUpgrading:
		phase = cloudv1.ClusterPhaseUpgrading
	case tidb.Status == cloudv1.ComponentScaling || tikv.Status == cloudv1.ComponentScaling || tiflash.Status == cloudv1.ComponentScaling || tiproxy.Status == cloudv1.ComponentScaling:
		phase = cloudv1.ClusterPhaseScaling
	default:
		phase = cloudv1.ClusterPhaseRunning
	}
	reason := fmt.Sprintf("pd: %s, tidb: %s, tikv: %s, tiflash: %s, tiproxy: %s", pd.Status, tidb.Status, tikv.Status, tiflash.Status, tiproxy.Status)
	dbStatus := &cloudv1.DBStatus{
		Version:         r.Version,
		TiDB:            tidb,
		TiKV:            tikv,
		PD:              pd,
		SecondaryStatus: secondaryStatus,
	}

	if r.Target.Spec.Cluster.TiFlash.Replicas > 0 {
		dbStatus.TiFlash = tiflash
	}
	if r.Target.Spec.Cluster.TiProxy.Replicas > 0 {
		dbStatus.TiProxy = tiproxy
	}

	if r.Target.Status.Conditions == nil {
		r.Target.Status.Conditions = []cloudv1.ClusterCondition{}
	}
	cloudv1.SetStatusCondition(&r.Target.Status.Conditions, r.buildSuspendCondition(dbStatus))

	if r.Target.Suspending() || r.Target.Suspended() {
		phase = cloudv1.ClusterPhaseSuspend
	}

	r.Target.Status.Phase = phase
	r.Target.Status.Reason = reason
	r.Target.Status.DB = dbStatus

	return nil
}

func (r *TiDBClusterStatusResolverV2) buildSuspendCondition(dbStatus *cloudv1.DBStatus) cloudv1.ClusterCondition {
	suspendCondition := cloudv1.ClusterCondition{
		Type:    cloudv1.ClusterSuspended,
		Status:  cloudv1.ConditionTrue,
		Reason:  "ClusterSuspended",
		Message: "The cluster has been suspended",
	}

	var msg []string
	if dbStatus == nil {
		msg = append(msg, "Cluster.Status.DB is nil")
	} else {
		if dbStatus.PD == nil {
			msg = append(msg, "Cluster.Status.DB.PD is nil")
		} else if !meta.IsStatusConditionTrue(dbStatus.PD.Conditions, cloudv1.ComponentSuspended) {
			msg = append(msg, "PD has not been suspended")
		}

		if dbStatus.TiDB == nil {
			msg = append(msg, "Cluster.Status.DB.TiDB is nil")
		} else if !meta.IsStatusConditionTrue(dbStatus.TiDB.Conditions, cloudv1.ComponentSuspended) {
			msg = append(msg, "TiDB has not been suspended")
		}

		if dbStatus.TiKV == nil {
			msg = append(msg, "Cluster.Status.DB.TiKV is nil")
		} else if !meta.IsStatusConditionTrue(dbStatus.TiKV.Conditions, cloudv1.ComponentSuspended) {
			msg = append(msg, "TiKV has not been suspended")
		}

		if r.Target.Spec.Cluster.TiFlash.Replicas > 0 {
			if dbStatus.TiFlash == nil {
				msg = append(msg, "Cluster.Status.DB.TiFlash is nil")
			} else if !meta.IsStatusConditionTrue(dbStatus.TiFlash.Conditions, cloudv1.ComponentSuspended) {
				msg = append(msg, "TiFlash has not been suspended")
			}
		}
	}

	if len(msg) > 0 {
		suspendCondition = cloudv1.ClusterCondition{
			Type:    cloudv1.ClusterSuspended,
			Status:  cloudv1.ConditionFalse,
			Reason:  "ClusterNotSuspended",
			Message: strings.Join(msg, ";"),
		}
	}
	return suspendCondition
}

func (r *TiDBClusterStatusResolverV2) computeComponentConditions(cType cloudv1.ComponentType, nodeCount int) []metav1.Condition {
	var conditions []metav1.Condition
	memberSuspended := false

	switch cType {
	case cloudv1.ComponentTiDB:
		// TiDB use status.TiDBResolver to compute component conditions
	case cloudv1.ComponentTiKV:
		if r.Target.Status.DB != nil && r.Target.Status.DB.TiKV != nil {
			conditions = r.Target.Status.DB.TiKV.Conditions
		}
		memberSuspended = meta.IsStatusConditionTrue(r.TiKVGroup.Status.Conditions, opv2.CondSuspended)
	case cloudv1.ComponentPD:
		if r.Target.Status.DB != nil && r.Target.Status.DB.PD != nil {
			conditions = r.Target.Status.DB.PD.Conditions
		}
		memberSuspended = meta.IsStatusConditionTrue(r.PDGroup.Status.Conditions, opv2.CondSuspended)
	case cloudv1.ComponentTiFlash:
		if r.Target.Status.DB != nil && r.Target.Status.DB.TiFlash != nil {
			conditions = r.Target.Status.DB.TiFlash.Conditions
		}
		memberSuspended = meta.IsStatusConditionTrue(r.TiFlashGroup.Status.Conditions, opv2.CondSuspended)
	case cloudv1.ComponentTiProxy:
		if r.Target.Status.DB != nil && r.Target.Status.DB.TiProxy != nil {
			conditions = r.Target.Status.DB.TiProxy.Conditions
		}
	default:
		return nil
	}

	if conditions == nil {
		conditions = []metav1.Condition{}
	}

	// The flow for the condition update is as below:
	// - If `suspendAction` is set in the spec, add the `ComponentSuspended`
	//   condition first to indicate that we're going to suspend the cluster
	// - Update TC with the `suspendAction` in the next reconcile,
	//   in case that the user may resume the cluster immediately
	//   after pausing the cluster
	// - Update `ComponentSuspended` condition to `True` after the cluster is suspended
	// - Remove the `ComponentSuspended` condition when the user resumes the cluster
	needSuspend := r.Target.GetComponent(cType).NeedSuspend(r.Target.Spec.Cluster.SuspendAction)
	memberSuspending := r.TiDBCluster.Spec.SuspendAction != nil && r.TiDBCluster.Spec.SuspendAction.SuspendCompute
	if !needSuspend && !memberSuspending {
		meta.RemoveStatusCondition(&conditions, cloudv1.ComponentSuspended)
		return conditions
	}

	var suspendCondition metav1.Condition
	if memberSuspended && nodeCount == 0 {
		suspendCondition = metav1.Condition{
			Type:    cloudv1.ComponentSuspended,
			Status:  metav1.ConditionTrue,
			Reason:  "ResourcesSuspended",
			Message: "All Pods and nodes are suspended",
		}
	} else {
		var msg []string
		if !memberSuspended {
			msg = append(msg, "Pods not suspended")
		}
		if nodeCount != 0 {
			msg = append(msg, fmt.Sprintf("%d Nodes not suspended", nodeCount))
		}
		suspendCondition = metav1.Condition{
			Type:    cloudv1.ComponentSuspended,
			Status:  metav1.ConditionFalse,
			Reason:  "ResourcesNotSuspended",
			Message: strings.Join(msg, ";"),
		}
	}
	meta.SetStatusCondition(&conditions, suspendCondition)
	return conditions
}

func (r *TiDBClusterStatusResolverV2) getInstanceResourceStatus(ctx context.Context, comp cloudv1.ComponentType, groupName, instanceName, instanceType string) (res cloudv1.Resources, curStorageClass, curVAC string) {
	if instanceType != "" {
		it, err := r.FindInstanceType(instanceType)
		if err != nil {
			r.Log.Error(err, "failed to find instance type", "instanceType", instanceType)
		} else {
			res.CPU = it.CPU
			res.Memory = it.Memory
		}
	}

	var pvcList corev1.PersistentVolumeClaimList
	if err := r.Client.List(ctx, &pvcList, client.InNamespace(r.TiDBCluster.Namespace), client.MatchingLabels(map[string]string{
		opv2.LabelKeyGroup:     groupName,
		opv2.LabelKeyInstance:  instanceName,
		opv2.LabelKeyComponent: string(comp),
		// Only list data volumes
		opv2.LabelKeyVolumeName: "data",
	})); err != nil {
		r.Log.Error(err, "failed to list PVCs", "component", comp, "namespace", r.TiDBCluster.Namespace, "group", groupName, "instance", instanceName)
	}
	if len(pvcList.Items) > 0 {
		// TODO: handle multiple PVCs
		pvc := pvcList.Items[0]
		res.Storage = pvc.Status.Capacity.Storage()

		// Since the `StorageClassName` field can not be changed in PVC,
		// tidb operator will store the actual storage class in the annotation.
		if scName, ok := pvc.Annotations["status.tidb.pingcap.com/storage-class"]; ok {
			curStorageClass = scName
		} else {
			curStorageClass = ptr.Deref(pvc.Spec.StorageClassName, "")
		}

		curVAC = ptr.Deref(pvc.Status.CurrentVolumeAttributesClassName, "")
	}

	return res, curStorageClass, curVAC
}

func (r *TiDBClusterStatusResolverV2) getPodsAndNodes(ctx context.Context, comp cloudv1.ComponentType, ls client.MatchingLabels) (instanceToPod map[string]*corev1.Pod, nameToNode map[string]*corev1.Node, err error) {
	var podList corev1.PodList
	if err = r.Client.List(ctx, &podList, client.InNamespace(r.TiDBCluster.Namespace), ls); err != nil {
		return nil, nil, fmt.Errorf("failed to list %s pods: %w", comp, err)
	}
	instanceToPod = make(map[string]*corev1.Pod, len(podList.Items))
	nameToNode = make(map[string]*corev1.Node, len(podList.Items))
	for i, pod := range podList.Items {
		for _, owner := range pod.GetOwnerReferences() {
			if strings.ToLower(owner.Kind) == string(comp) && owner.APIVersion == opv2.SchemeGroupVersion.String() {
				instanceToPod[owner.Name] = &podList.Items[i]
			}
		}
		if len(pod.Spec.NodeName) < 1 {
			continue
		}
		var node corev1.Node
		if err = r.Client.Get(ctx, client.ObjectKey{Name: pod.Spec.NodeName}, &node); err != nil {
			r.Log.Error(err, "failed to get node", "node", pod.Spec.NodeName)
			continue
		}
		nameToNode[node.Name] = &node
	}

	// We have to combine the nodes got from the pods and the nodes got from the lister.
	// Because in GCP and Azure, we can not list nodes by the component label.
	var nodes corev1.NodeList
	if err = r.Client.List(ctx, &nodes, client.MatchingLabels(r.Namer.WithComponent(string(comp)).Labels())); err != nil {
		return nil, nil, fmt.Errorf("failed to list %s nodes: %w", comp, err)
	}
	for i, node := range nodes.Items {
		nameToNode[node.Name] = &nodes.Items[i]
	}

	return instanceToPod, nameToNode, nil
}

func (r *TiDBClusterStatusResolverV2) computePDStatus(ctx context.Context) (*cloudv1.PDStatus, error) {
	var pdGroup opv2.PDGroup
	if err := r.Client.Get(ctx, client.ObjectKey{
		Name:      consts.ClusterPrefix,
		Namespace: r.TiDBCluster.Namespace,
	}, &pdGroup); err != nil {
		return nil, fmt.Errorf("failed to get the PDGroup: %w", err)
	}
	r.PDGroup = &pdGroup

	ls := client.MatchingLabels(getComponentPodLabels(cloudv1.ComponentPD, r.TiDBCluster.Name, pdGroup.Name))
	var pdList opv2.PDList
	if err := r.Client.List(ctx, &pdList, client.InNamespace(r.TiDBCluster.Namespace), ls); err != nil {
		return nil, fmt.Errorf("failed to list PD instances: %w", err)
	}

	instanceToPod, nameToNode, err := r.getPodsAndNodes(ctx, cloudv1.ComponentPD, ls)
	if err != nil {
		return nil, fmt.Errorf("failed to get pods and nodes: %w", err)
	}

	members := make([]cloudv1.PDMember, 0, len(pdList.Items))
	// storageSizeMap is used to count the number of PD member with the same storage size.
	// The key is the storage size string, and the value is the count.
	storageSizeMap := make(map[string]int, len(pdList.Items))
	// scNameMap is used to count the number of PD member with the same StorageClass.
	// The key is the storage class name, and the value is the count.
	scNameMap := make(map[string]int, len(pdList.Items))
	// vacNameMap is used to count the number of PD member with the same VolumeAttributesClass name.
	vacNameMap := make(map[string]int, len(pdList.Items))
	allPDMembersReady := true
	for _, pd := range pdList.Items {
		memberStatus := cloudv1.MemberStatusNormal
		if !v2utils.IsReady(pd.Status.Conditions) {
			allPDMembersReady = false
			memberStatus = cloudv1.MemberStatusUnavailable
		}

		var instanceType, zone string
		pod, ok := instanceToPod[pd.Name]
		if ok && pod.Spec.NodeName != "" {
			if node, ok2 := nameToNode[pod.Spec.NodeName]; ok2 {
				zone = node.Labels[consts.ZoneLabelKey]
				instanceType = node.Labels[corev1.LabelInstanceTypeStable]
			}
		}

		res, curSC, curVAC := r.getInstanceResourceStatus(ctx, cloudv1.ComponentPD, pdGroup.Name, pd.Name, instanceType)
		m := cloudv1.PDMember{
			Name:         status.InstanceNameToPodName(pd.Name, "pd"),
			Labels:       map[string]string{cloudv1.MemberLabelAvailableZone: zone},
			Resources:    res,
			Status:       memberStatus,
			InstanceType: instanceType,
		}
		scNameMap[curSC]++
		if curVAC != "" {
			vacNameMap[curVAC]++
		}
		if m.Resources.Storage != nil && !m.Resources.Storage.IsZero() {
			storageSizeMap[m.Resources.Storage.String()]++
		}
		members = append(members, m)
	}
	sort.Slice(members, func(i, j int) bool {
		return members[i].Name < members[j].Name
	})

	pdNodeScaling := r.componentNodeScaling(cloudv1.ComponentPD)

	if pdGroup.Status.ObservedGeneration != pdGroup.GetGeneration() {
		return nil, fmt.Errorf("wait until change is observed")
	}

	var prevVol *cloudv1.VolumeStatus
	if r.Target.Status.DB != nil && r.Target.Status.DB.PD != nil {
		prevVol = r.Target.Status.DB.PD.Volume
	}

	pdSpec := r.Target.Spec.Cluster.PD
	r.Log.Info("PD storageMap", "storageSizeMap", storageSizeMap, "scNameMap", scNameMap, "vacNameMap", vacNameMap)
	volume, volModifying, err := r.computeVolumeStatus(&pdSpec, prevVol, int(pdGroup.Status.Replicas), int(pdGroup.Status.CurrentReplicas), int(pdGroup.Status.UpdatedReplicas), storageSizeMap, scNameMap, vacNameMap)
	if err != nil {
		return nil, fmt.Errorf("failed to compute PD volume status: %w", err)
	}

	var instanceTypeScalingPhase cloudv1.InstanceTypeScalePhase
	if r.Target.Status.DB != nil && r.Target.Status.DB.PD != nil {
		instanceTypeScalingPhase = r.Target.Status.DB.PD.InstanceTypeScalePhase
	}

	itScaling := instanceTypeScalingFromPDMembers(members, pdSpec.GetInstanceType()) || instanceTypeScalingPhase != cloudv1.InstanceTypeScalePhaseNone

	r.Log.Info("PD status", "scaling", pdNodeScaling, "modifying", volModifying, "itScaling", itScaling, "allPDMembersReady", allPDMembersReady)
	var status cloudv1.ComponentStatus
	switch {
	case (pdGroup.Status.ReadyReplicas == 0 && pdGroup.Spec.Replicas != nil && *pdGroup.Spec.Replicas > 0) || meta.IsStatusConditionTrue(pdGroup.Status.Conditions, opv2.CondSuspended):
		status = cloudv1.ComponentUnavailable
	case pdGroup.Status.Version != pdGroup.Spec.Template.Spec.Version && !itScaling:
		status = cloudv1.ComponentUpgrading
	case pdNodeScaling || volModifying || itScaling:
		status = cloudv1.ComponentScaling
	case !allPDMembersReady:
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	return &cloudv1.PDStatus{
		Members:                members,
		Status:                 status,
		Volume:                 volume,
		UpdatedReplicas:        pdGroup.Status.UpdatedReplicas,
		InstanceTypeScalePhase: instanceTypeScalingPhase, // do not overwrite it as it update by the cluster controller
		Conditions:             r.computeComponentConditions(cloudv1.ComponentPD, len(nameToNode)),
	}, nil
}

func (r *TiDBClusterStatusResolverV2) computeTiDBStatus(ctx context.Context) (*cloudv1.TiDBStatus, error) {
	dbg := &opv2.TiDBGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      r.Namer.GetTiDBName(),
			Namespace: r.Namer.GetTiDBNamespace(),
		},
	}
	if err := r.Get(ctx, client.ObjectKeyFromObject(dbg), dbg); err != nil {
		return nil, err
	}
	r.TiDBGroup = dbg

	ls := client.MatchingLabels(getComponentPodLabels(cloudv1.ComponentTiDB, r.TiDBCluster.Name, r.TiDBGroup.Name))
	var tidbList opv2.TiDBList
	if err := r.Client.List(ctx, &tidbList, client.InNamespace(r.TiDBCluster.Namespace), ls); err != nil {
		return nil, fmt.Errorf("failed to list TiDB instances: %w", err)
	}
	dbs := make([]*opv2.TiDB, 0, len(tidbList.Items))
	for i := range tidbList.Items {
		dbs = append(dbs, &tidbList.Items[i])
	}

	_, nameToNode, err := r.getPodsAndNodes(ctx, cloudv1.ComponentTiDB, ls)
	if err != nil {
		return nil, fmt.Errorf("failed to get pods and nodes: %w", err)
	}

	var tidbStatus *cloudv1.TiDBStatus
	if r.Target.Status.DB != nil {
		tidbStatus = r.Target.Status.DB.TiDB
	}
	needSuspend := r.Target.Spec.Cluster.TiDB.NeedSuspend(r.Target.Spec.Cluster.SuspendAction)
	instanceType := r.Target.Spec.Cluster.TiDB.GetInstanceType()
	return status.NewTiDBResolver(
		r.Log,
		r.Gates,
		r.Provider,
		len(nameToNode),
		r.Target.Spec.Cluster.TiDB.GetReplicas(),
		needSuspend,
		instanceType,
		r.Target.TiDBMemberInstanceTypeAllowed,
	).ComputeTiDBStatus(tidbStatus, r.TiDBGroup, dbs)
}

func (r *TiDBClusterStatusResolverV2) computeTiProxyStatus(ctx context.Context) (*cloudv1.TiProxyStatus, error) {
	tiproxySpec := r.Target.Spec.Cluster.TiProxy
	if tiproxySpec.Replicas == 0 {
		// dummy status
		return &cloudv1.TiProxyStatus{Status: cloudv1.ComponentNormal}, nil
	}

	pg := &opv2.TiProxyGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      r.Namer.GetTiDBName(),
			Namespace: r.Namer.GetTiDBNamespace(),
		},
	}
	if err := r.Get(ctx, client.ObjectKeyFromObject(pg), pg); err != nil {
		return nil, err
	}
	r.TiProxyGroup = pg

	ls := client.MatchingLabels(getComponentPodLabels(cloudv1.ComponentTiProxy, r.TiDBCluster.Name, r.TiProxyGroup.Name))
	var tiproxyList opv2.TiProxyList
	if err := r.Client.List(ctx, &tiproxyList, client.InNamespace(r.TiDBCluster.Namespace), ls); err != nil {
		return nil, fmt.Errorf("failed to list TiProxy instances: %w", err)
	}
	ps := make([]*opv2.TiProxy, 0, len(tiproxyList.Items))
	for i := range tiproxyList.Items {
		ps = append(ps, &tiproxyList.Items[i])
	}

	_, nameToNode, err := r.getPodsAndNodes(ctx, cloudv1.ComponentTiProxy, ls)
	if err != nil {
		return nil, fmt.Errorf("failed to get pods and nodes: %w", err)
	}

	var tiproxyStatus *cloudv1.TiProxyStatus
	if r.Target.Status.DB != nil {
		tiproxyStatus = r.Target.Status.DB.TiProxy
	}
	needSuspend := r.Target.Spec.Cluster.TiProxy.NeedSuspend(r.Target.Spec.Cluster.SuspendAction)
	instanceType := r.Target.Spec.Cluster.TiProxy.GetInstanceType()
	return status.NewTiProxyResolver(r.Log, r.Gates, r.Provider, len(nameToNode), r.Target.Spec.Cluster.TiProxy.GetReplicas(), needSuspend, instanceType).ComputeTiProxyStatus(tiproxyStatus, r.TiProxyGroup, ps)
}

func (r *TiDBClusterStatusResolverV2) computeTiKVStatus(ctx context.Context) (*cloudv1.TiKVStatus, error) {
	var tikvGroup opv2.TiKVGroup
	if err := r.Client.Get(ctx, client.ObjectKey{
		Name:      consts.ClusterPrefix,
		Namespace: r.TiDBCluster.Namespace,
	}, &tikvGroup); err != nil {
		return nil, fmt.Errorf("failed to get the TiKVGroup: %w", err)
	}
	r.TiKVGroup = &tikvGroup

	ls := client.MatchingLabels(getComponentPodLabels(cloudv1.ComponentTiKV, r.TiDBCluster.Name, tikvGroup.Name))
	var tikvList opv2.TiKVList
	if err := r.Client.List(ctx, &tikvList, client.InNamespace(r.TiDBCluster.Namespace), ls); err != nil {
		return nil, fmt.Errorf("failed to list TiKV instances: %w", err)
	}

	nameToPod, nameToNode, err := r.getPodsAndNodes(ctx, cloudv1.ComponentTiKV, ls)
	if err != nil {
		return nil, fmt.Errorf("failed to get pods and nodes: %w", err)
	}

	healthyCount := 0
	members := make([]cloudv1.StoreMember, 0, len(tikvList.Items))
	// storageSizeMap is used to count the number of TiKV member with the same storage size.
	// The key is the storage size string, and the value is the count.
	storageSizeMap := make(map[string]int, len(tikvList.Items))
	// scNameMap is used to count the number of TiKV member with the same storage class.
	// The key is the storage class name, and the value is the count.
	scNameMap := make(map[string]int, len(tikvList.Items))
	// vacNameMap is used to count the number of TiKV member with the same VolumeAttributesClass name.
	vacNameMap := make(map[string]int, len(tikvList.Items))
	tikvSpec := r.Target.Spec.Cluster.TiKV
	for _, tikv := range tikvList.Items {
		readyAndUpToDate := v2utils.ReadyAndUpToDate(tikv.GetGeneration(), tikv.Status.CommonStatus)
		if readyAndUpToDate {
			healthyCount++
		}
		m, curSc, curVAC := r.computeStoreMember(ctx, &tikvSpec, tikvGroup.Name, tikv.Name, tikv.Namespace, tikv.Status.State, readyAndUpToDate, nameToPod, nameToNode)
		if m.Resources.Storage != nil && !m.Resources.Storage.IsZero() {
			storageSizeMap[m.Resources.Storage.String()]++
		}
		scNameMap[curSc]++
		if curVAC != "" {
			vacNameMap[curVAC]++
		}
		members = append(members, m)
	}
	sort.Slice(members, func(i, j int) bool {
		return members[i].Name < members[j].Name
	})

	tikvNodeScaling := r.componentNodeScaling(cloudv1.ComponentTiKV)

	if tikvGroup.Status.ObservedGeneration != tikvGroup.GetGeneration() {
		return nil, fmt.Errorf("wait until change is observed")
	}

	var prevVol *cloudv1.VolumeStatus
	if r.Target.Status.DB != nil && r.Target.Status.DB.TiKV != nil {
		prevVol = r.Target.Status.DB.TiKV.Volume
	}

	r.Log.Info("TiKV storageMap", "storageMap", storageSizeMap, "scNameMap", scNameMap, "desiredSize", tikvSpec.StorageSize.String(), "vacNameMap", vacNameMap)
	volume, volModifying, err := r.computeVolumeStatus(&tikvSpec, prevVol, int(tikvGroup.Status.Replicas), int(tikvGroup.Status.CurrentReplicas), int(tikvGroup.Status.UpdatedReplicas), storageSizeMap, scNameMap, vacNameMap)
	if err != nil {
		return nil, fmt.Errorf("failed to compute TiKV volume status: %w", err)
	}
	additionalVolumes, additionalVolModifying, err := r.computeTiKVAdditinalVolumesStatus(ctx, tikvGroup)
	if err != nil {
		return nil, fmt.Errorf("failed to compute additional volumes status: %w", err)
	}
	volModifying = volModifying || additionalVolModifying

	var instanceTypeScalingPhase cloudv1.InstanceTypeScalePhase
	if r.Target.Status.DB != nil && r.Target.Status.DB.TiKV != nil {
		instanceTypeScalingPhase = r.Target.Status.DB.TiKV.InstanceTypeScalePhase
	}
	itScaling := storeMembersHaveDisallowedInstanceType(members, r.Target.TiKVMemberInstanceTypeAllowed) || instanceTypeScalingPhase != cloudv1.InstanceTypeScalePhaseNone

	r.Log.Info("TiKV status", "scaling", tikvNodeScaling, "modifying", volModifying, "itScaling", itScaling, "healthyCount", healthyCount)
	var status cloudv1.ComponentStatus
	switch {
	case healthyCount == 0:
		status = cloudv1.ComponentUnavailable
	case tikvGroup.Status.Version != tikvGroup.Spec.Template.Spec.Version:
		status = cloudv1.ComponentUpgrading
	case tikvNodeScaling || volModifying || itScaling:
		status = cloudv1.ComponentScaling
	case tikvGroup.Spec.Replicas != nil && int(*tikvGroup.Spec.Replicas) != healthyCount:
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	return &cloudv1.TiKVStatus{
		Members:                members,
		Status:                 status,
		Volume:                 volume,
		StorageVolumes:         additionalVolumes,
		UpdatedReplicas:        tikvGroup.Status.UpdatedReplicas,
		InstanceTypeScalePhase: instanceTypeScalingPhase,
		Conditions:             r.computeComponentConditions(cloudv1.ComponentTiKV, len(nameToNode)),
	}, nil
}

func (r *TiDBClusterStatusResolverV2) computeTiFlashStatus(ctx context.Context) (*cloudv1.TiFlashStatus, error) {
	tiflashSpec := r.Target.Spec.Cluster.TiFlash
	if tiflashSpec.Replicas == 0 {
		// dummy status
		return &cloudv1.TiFlashStatus{Status: cloudv1.ComponentNormal}, nil
	}

	var tiflashGroup opv2.TiFlashGroup
	if err := r.Client.Get(ctx, client.ObjectKey{
		Name:      consts.ClusterPrefix,
		Namespace: r.TiDBCluster.Namespace,
	}, &tiflashGroup); err != nil {
		return nil, fmt.Errorf("failed to get the TiFlashGroup: %w", err)
	}
	r.TiFlashGroup = &tiflashGroup

	ls := client.MatchingLabels(getComponentPodLabels(cloudv1.ComponentTiFlash, r.TiDBCluster.Name, tiflashGroup.Name))
	var tiflashList opv2.TiFlashList
	if err := r.Client.List(ctx, &tiflashList, client.InNamespace(r.TiDBCluster.Namespace), ls); err != nil {
		return nil, fmt.Errorf("failed to list TiFlash instances: %w", err)
	}

	nameToPod, nameToNode, err := r.getPodsAndNodes(ctx, cloudv1.ComponentTiFlash, ls)
	if err != nil {
		return nil, fmt.Errorf("failed to get pods and nodes: %w", err)
	}

	healthyCount := 0
	members := make([]cloudv1.StoreMember, 0, len(tiflashList.Items))
	// storageSizeMap is used to count the number of TiFlash member with the same storage size.
	// The key is the storage size string, and the value is the count.
	storageSizeMap := make(map[string]int, len(tiflashList.Items))
	// scNameMap is used to count the number of TiFlash member with the same storage class.
	// The key is the storage class name, and the value is the count.
	scNameMap := make(map[string]int, len(tiflashList.Items))
	// vacNameMap is used to count the number of TiFlash member with the same VolumeAttributesClass name.
	vacNameMap := make(map[string]int, len(tiflashList.Items))
	for _, tiflash := range tiflashList.Items {
		readyAndUpToDate := v2utils.ReadyAndUpToDate(tiflash.GetGeneration(), tiflash.Status.CommonStatus)
		if readyAndUpToDate {
			healthyCount++
		}
		m, curSc, curVAC := r.computeStoreMember(ctx, &tiflashSpec, tiflashGroup.Name, tiflash.Name, tiflash.Namespace, tiflash.Status.State, readyAndUpToDate, nameToPod, nameToNode)
		if m.Resources.Storage != nil && !m.Resources.Storage.IsZero() {
			storageSizeMap[m.Resources.Storage.String()]++
		}
		scNameMap[curSc]++
		if curVAC != "" {
			vacNameMap[curVAC]++
		}
		members = append(members, m)
	}
	sort.Slice(members, func(i, j int) bool {
		return members[i].Name < members[j].Name
	})

	tiflashNodeScaling := r.componentNodeScaling(cloudv1.ComponentTiFlash)

	if tiflashGroup.Status.ObservedGeneration != tiflashGroup.GetGeneration() {
		return nil, fmt.Errorf("wait until change is observed")
	}

	var prevVol *cloudv1.VolumeStatus
	if r.Target.Status.DB != nil && r.Target.Status.DB.TiFlash != nil {
		prevVol = r.Target.Status.DB.TiFlash.Volume
	}

	r.Log.Info("TiFlash storageMap", "storageSizeMap", storageSizeMap, "scNameMap", scNameMap, "vacNameMap", vacNameMap)
	volume, volModifying, err := r.computeVolumeStatus(&tiflashSpec, prevVol, int(tiflashGroup.Status.Replicas), int(tiflashGroup.Status.CurrentReplicas), int(tiflashGroup.Status.UpdatedReplicas), storageSizeMap, scNameMap, vacNameMap)
	if err != nil {
		return nil, fmt.Errorf("failed to compute TiFlash volume status: %w", err)
	}

	var instanceTypeScalingPhase cloudv1.InstanceTypeScalePhase
	if r.Target.Status.DB != nil && r.Target.Status.DB.TiFlash != nil {
		instanceTypeScalingPhase = r.Target.Status.DB.TiFlash.InstanceTypeScalePhase
	}
	itScaling := instanceTypeScalingFromStoreMembers(members, tiflashSpec.InstanceType) || instanceTypeScalingPhase != cloudv1.InstanceTypeScalePhaseNone

	r.Log.Info("TiFlash status", "scaling", tiflashNodeScaling, "modifying", volModifying, "itScaling", itScaling, "healthyCount", healthyCount)
	var compStatus cloudv1.ComponentStatus
	switch {
	case tiflashGroup.Status.Version != tiflashGroup.Spec.Template.Spec.Version:
		compStatus = cloudv1.ComponentUpgrading
	case tiflashNodeScaling || volModifying || itScaling:
		compStatus = cloudv1.ComponentScaling
	case healthyCount == 0:
		compStatus = cloudv1.ComponentUnavailable
	case tiflashGroup.Spec.Replicas != nil && int(*tiflashGroup.Spec.Replicas) != healthyCount:
		compStatus = cloudv1.ComponentUnhealthy
	default:
		compStatus = cloudv1.ComponentNormal
	}

	return &cloudv1.TiFlashStatus{
		Members:                members,
		Status:                 compStatus,
		Volume:                 volume,
		UpdatedReplicas:        tiflashGroup.Status.UpdatedReplicas,
		InstanceTypeScalePhase: instanceTypeScalingPhase,
		Conditions:             r.computeComponentConditions(cloudv1.ComponentTiFlash, len(nameToNode)),
	}, nil
}

func (r *TiDBClusterStatusResolverV2) computeStoreMember(
	ctx context.Context,
	comp cloudv1.ClusterComponent,
	groupName, instanceName, ns, storeState string,
	isNormal bool,
	instanceToPod map[string]*corev1.Pod,
	nameToNode map[string]*corev1.Node,
) (_ cloudv1.StoreMember, curStorageClass, curVAC string) {
	compType := comp.GetComponentType()
	pod := instanceToPod[instanceName]
	podIsDeleting := pod != nil && !pod.DeletionTimestamp.IsZero()

	var memberStatus cloudv1.MemberStatus
	switch {
	case isNormal:
		memberStatus = cloudv1.MemberStatusNormal
	case podIsDeleting:
		memberStatus = cloudv1.MemberStatusTerminating
	case storeState == opv2.StoreStateRemoving:
		memberStatus = cloudv1.MemberStatusLeaving
	case pod == nil || pod.Status.Phase == corev1.PodPending:
		memberStatus = cloudv1.MemberStatusCreating
	default:
		r.Log.Info("unknown store state", "component", compType, "group", groupName, "instance", instanceName, "namespace", ns, "state", storeState, "deleting", podIsDeleting)
		memberStatus = cloudv1.MemberStatusUnavailable
	}

	var instanceType, zone string
	if pod != nil && pod.Spec.NodeName != "" && nameToNode[pod.Spec.NodeName] != nil {
		zone = nameToNode[pod.Spec.NodeName].Labels[consts.ZoneLabelKey]
		instanceType = nameToNode[pod.Spec.NodeName].Labels[corev1.LabelInstanceTypeStable]
	}

	res, curSC, curVAC := r.getInstanceResourceStatus(ctx, compType, groupName, instanceName, instanceType)
	return cloudv1.StoreMember{
		Name:         status.InstanceNameToPodName(instanceName, string(compType)),
		Labels:       map[string]string{cloudv1.MemberLabelAvailableZone: zone},
		Resources:    res,
		Status:       memberStatus,
		InstanceType: instanceType,
	}, curSC, curVAC
}

func getComponentPodLabels(compType cloudv1.ComponentType, tcName, groupName string) map[string]string {
	return map[string]string{
		opv2.LabelKeyCluster:   tcName,
		opv2.LabelKeyGroup:     groupName,
		opv2.LabelKeyComponent: string(compType),
	}
}

func (r *TiDBClusterStatusResolverV2) componentNodeScaling(compType cloudv1.ComponentType) bool {
	var groupStatus opv2.GroupStatus
	switch compType {
	case cloudv1.ComponentTiDB:
		// tidb group status will be resolved by status.TiDBResolver
	case cloudv1.ComponentPD:
		groupStatus = r.PDGroup.Status.GroupStatus
	case cloudv1.ComponentTiKV:
		groupStatus = r.TiKVGroup.Status.GroupStatus
	case cloudv1.ComponentTiFlash:
		groupStatus = r.TiFlashGroup.Status.GroupStatus
	default:
		// other components does not consider node scaling
		return false
	}

	desiredReplicas := int32(r.Target.GetComponent(compType).GetReplicas())
	r.Log.Info("componentNodeScaling", "component", compType, "actualReplicas", groupStatus.Replicas, "readyReplicas", groupStatus.ReadyReplicas, "desiredReplicas", desiredReplicas)
	return groupStatus.Replicas != desiredReplicas
}

func (r *TiDBClusterStatusResolverV2) computeVolumeStatus(
	comp cloudv1.ClusterComponent,
	prev *cloudv1.VolumeStatus,
	replicas, currentReplicas, updateReplicas int,
	storageSizeMap, scNameMap, vacNameMap map[string]int,
) (vs *cloudv1.VolumeStatus, modifying bool, err error) {
	if prev != nil {
		vs = prev.DeepCopy()
		vs.BoundCount = 0
	} else {
		vs = &cloudv1.VolumeStatus{}
	}

	for _, v := range storageSizeMap {
		vs.BoundCount += v
	}

	// always set modified cap and sc to expected
	size, ok := comp.GetStorageSize()
	if ok {
		vs.ModifiedCapacity = *size
	}
	vs.ModifiedStorageClass, _ = comp.GetStorageClassName()
	vs.ModifiedStorageClass = r.Mapper.Name(vs.ModifiedStorageClass)

	// init current cap and sc if it's not set
	if vs.CurrentCapacity.IsZero() {
		vs.CurrentCapacity = vs.ModifiedCapacity
	} else {
		for k, v := range storageSizeMap {
			// If all replicas have completed the capacity change,
			// set the current capacity to the modified capacity.
			if k == vs.ModifiedCapacity.String() && v == comp.GetReplicas() && v == replicas {
				vs.CurrentCapacity = vs.ModifiedCapacity
				break
			}
			// The volume capacity is changing.
			if v > 0 && vs.ModifiedCapacity.String() != k {
				vs.CurrentCapacity = resource.MustParse(k)
				break
			}
		}
	}

	if vs.CurrentStorageClass == "" {
		vs.CurrentStorageClass = vs.ModifiedStorageClass
	} else {
		for k, v := range scNameMap {
			// If all replicas have completed the storage class change,
			// set the current storage class to the modified storage class.
			if k == vs.ModifiedStorageClass && v == comp.GetReplicas() && v == replicas {
				vs.CurrentStorageClass = vs.ModifiedStorageClass
				break
			}
			// The storage class is changing.
			if v > 0 && vs.ModifiedStorageClass != k {
				vs.CurrentStorageClass = k
				break
			}
		}
	}

	if vs.CurrentVolumeAttributesClass == "" {
		vs.CurrentVolumeAttributesClass = vs.ModifiedVolumeAttributesClass
	} else {
		for k, v := range vacNameMap {
			// If all replicas have completed the VolumeAttributesClass change,
			// set the current VAC to the modified VAC.
			if k == vs.ModifiedVolumeAttributesClass && v == comp.GetReplicas() && v == replicas {
				vs.CurrentVolumeAttributesClass = vs.ModifiedVolumeAttributesClass
				break
			}
			// The volume attributes class is changing.
			if v > 0 && vs.ModifiedVolumeAttributesClass != k {
				vs.CurrentVolumeAttributesClass = k
				break
			}
		}
	}

	// if cap or sc is changed, update count
	if vs.CurrentCapacity.Cmp(vs.ModifiedCapacity) != 0 || vs.CurrentStorageClass != vs.ModifiedStorageClass {
		// only when modifying, current count and update count are equals with the currentReplicas and updateReplicas
		vs.CurrentCount = currentReplicas
		vs.ModifiedCount = updateReplicas
		modifying = true
	}

	if !modifying {
		vs.CurrentCount = replicas
		vs.ModifiedCount = replicas
	}

	// Deprecated, just set as modified
	// See https://github.com/pingcap/tidb-operator/blob/v1.7.0-alpha.7/pkg/manager/volumes/sync_volume_status.go#L150-L151
	vs.ResizedCapacity = vs.ModifiedCapacity
	vs.ResizedCount = vs.ModifiedCount

	return vs, modifying, nil
}

func (r *TiDBClusterStatusResolverV2) computeTiKVAdditinalVolumesStatus(
	ctx context.Context, tikvGroup opv2.TiKVGroup) (
	[]cloudv1.StorageVolumeStatus, bool /*modifying*/, error,
) {
	if len(r.Target.Spec.Cluster.TiKV.StorageVolumes) == 0 {
		// we do not support to remove additional volumes now.
		// so if the additional volumes is empty, we do not need to check the status.
		return nil, false, nil
	}

	var (
		status    []cloudv1.StorageVolumeStatus
		modifying bool
	)
	for _, vol := range r.Target.Spec.Cluster.TiKV.StorageVolumes {
		var pvcList corev1.PersistentVolumeClaimList
		if err := r.Client.List(ctx, &pvcList,
			client.InNamespace(tikvGroup.Namespace),
			client.MatchingLabels(map[string]string{
				opv2.LabelKeyGroup:      tikvGroup.Name,
				opv2.LabelKeyComponent:  string(cloudv1.ComponentTiKV),
				opv2.LabelKeyVolumeName: vol.Name,
			})); err != nil {
			return nil, false, err
		}

		desiredSC := r.Mapper.Name(vol.StorageClassName)

		scCountMap := make(map[string]int)
		sizeCountMap := make(map[string]int)
		for _, pvc := range pvcList.Items {
			currentSC := ""
			if scName, ok := pvc.Annotations["status.tidb.pingcap.com/storage-class"]; ok {
				currentSC = scName
			} else {
				currentSC = ptr.Deref(pvc.Spec.StorageClassName, "")
			}
			scCountMap[currentSC]++
			sizeCountMap[pvc.Status.Capacity.Storage().String()]++
		}
		currentSC := desiredSC
		for k, v := range scCountMap {
			if k != desiredSC && v > 0 {
				currentSC = k
				break
			}
		}
		currentSize := vol.StorageSize
		for k, v := range sizeCountMap {
			if k != vol.StorageSize.String() && v > 0 {
				currentSize = resource.MustParse(k)
				break
			}
		}

		vs := cloudv1.VolumeStatus{
			BoundCount:           len(pvcList.Items),
			CurrentCount:         int(tikvGroup.Status.CurrentReplicas),
			ModifiedCount:        int(tikvGroup.Status.UpdatedReplicas),
			CurrentCapacity:      currentSize,
			ModifiedCapacity:     vol.StorageSize,
			CurrentStorageClass:  currentSC,
			ModifiedStorageClass: desiredSC,
			ResizedCount:         int(tikvGroup.Status.UpdatedReplicas),
			ResizedCapacity:      vol.StorageSize,
		}

		if vs.CurrentCapacity.Cmp(vs.ModifiedCapacity) != 0 || vs.CurrentStorageClass != vs.ModifiedStorageClass {
			modifying = true
		}

		status = append(status, cloudv1.StorageVolumeStatus{
			Name:         vol.Name,
			VolumeStatus: vs,
		})
	}

	sort.Slice(status, func(i, j int) bool {
		return status[i].Name < status[j].Name
	})

	return status, modifying, nil
}