package common

import (
	"context"
	"maps"
	"reflect"
	"runtime/debug"
	"slices"
	"sort"
	"strings"
	"time"

	"github.com/mitchellh/mapstructure"
	"github.com/pingcap/errors"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"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"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/gkeheterogeneous"
	gkestatus "github.com/tidbcloud/infra-provider/pkg/gkeheterogeneous/status"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
)

const (
	gkeHeterogeneousStatusRetryInterval     = time.Minute
	gkeHeterogeneousStatusMinProbeInterval  = 5 * time.Minute
	gkeHeterogeneousScaleWorkflowStalePhase = "Stale"
)

type GKEHeterogeneousStatusTrack struct {
	*task.BaseTask
}

type gkeHeterogeneousIntent struct {
	componentType    cloudv1.ComponentType
	tidbGroupID      string
	computeClassName string
	baseInstanceType string
	desiredReplicas  int32
	nodePools        []cloudv1.GKEComputeClassNodePoolBinding
}

type gkeHeterogeneousComponentPods struct {
	intent gkeHeterogeneousIntent
	pods   []corev1.Pod
}

type gkeHeterogeneousScaleWorkflowInput struct {
	GroupID               string
	ComponentType         string
	ComponentSpec         string
	OriginalComponentSpec string
}

func NewGKEHeterogeneousStatusTrack(ctx *task.Context) *GKEHeterogeneousStatusTrack {
	return &GKEHeterogeneousStatusTrack{
		BaseTask: &task.BaseTask{Context: ctx},
	}
}

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

func (t *GKEHeterogeneousStatusTrack) Create(ctx context.Context) (result deploy.Result) {
	defer t.recoverCreate(ctx, &result)

	now := time.Now()
	if t.Config.DryRun.Enabled {
		return deploy.Done()
	}
	if t.Config.Provider != cloudv1.ProviderTypeGKE || t.Cluster.GetProviderType() != cloudv1.ProviderTypeGKE {
		return t.writeDisabledSnapshotIfStatusConfigMapExists(ctx)
	}
	if t.Cluster.Suspending() || t.Cluster.Suspended() {
		return deploy.Done()
	}

	oldSnapshot, err := t.readStatusSnapshot(ctx)
	if err != nil {
		return t.doneOrRetry(err, "read gke heterogeneous status snapshot")
	}
	intents, invalidErr := t.resolveIntents()
	if invalidErr != nil {
		return t.writeTerminalSnapshot(ctx, gkestatus.ProbeStatusInvalidIntent, true, false, invalidErr.Error(), nil)
	}
	if len(intents) == 0 {
		return t.writeNoIntentSnapshot(ctx, oldSnapshot)
	}

	scaleWorkflow := t.probeScaleWorkflow(ctx, oldSnapshot, intents)
	scaleWorkflowRunning := scaleWorkflow != nil && scaleWorkflow.InProgress && !scaleWorkflow.Stale
	clusterFullSynced := t.Cluster.FullSynced()
	if remaining := placementProbeThrottleRemaining(oldSnapshot, now); remaining > 0 {
		workflowUpdated, err := t.updateThrottledSnapshot(ctx, oldSnapshot, scaleWorkflow, clusterFullSynced)
		if err != nil {
			t.Logger.Error(err, "failed to update throttled gke heterogeneous status snapshot")
			return deploy.Done().WithRequeuAfter(gkeHeterogeneousStatusRetryInterval)
		}
		if oldSnapshot != nil && oldSnapshot.ProbeStatus == gkestatus.ProbeStatusFailed {
			return deploy.Done().WithRequeuAfter(remaining)
		}
		if !clusterFullSynced {
			return deploy.Done()
		}
		if oldSnapshot != nil && !oldSnapshot.ClusterFullSynced {
			return deploy.Done().WithRequeuAfter(remaining)
		}
		if scaleWorkflowRunning || workflowUpdated {
			return deploy.Done().WithRequeuAfter(remaining)
		}
		return deploy.Done()
	}

	snapshot, err := t.probePlacement(ctx, intents)
	if err != nil {
		return t.writeFailedPlacementSnapshot(ctx, oldSnapshot, scaleWorkflow, clusterFullSynced, now, err)
	}
	snapshot.ScaleWorkflow = scaleWorkflow
	snapshot.ClusterFullSynced = clusterFullSynced
	snapshot.LastPlacementProbeTime = timePtr(now)
	snapshot.LastSuccessfulPlacementProbeTime = timePtr(now)
	if err := t.writeStatusSnapshot(ctx, snapshot); err != nil {
		t.Logger.Error(err, "failed to write gke heterogeneous status snapshot")
		return deploy.Done().WithRequeuAfter(gkeHeterogeneousStatusRetryInterval)
	}
	if !clusterFullSynced {
		return deploy.Done()
	}
	if scaleWorkflowRunning {
		return deploy.Done().WithRequeuAfter(gkeHeterogeneousStatusMinProbeInterval)
	}
	return deploy.Done()
}

func (t *GKEHeterogeneousStatusTrack) Delete(ctx context.Context) (result deploy.Result) {
	defer t.recoverDelete(&result)

	if t.Config.DryRun.Enabled {
		return deploy.Done()
	}
	return t.writeDisabledSnapshotIfStatusConfigMapExists(ctx)
}

func (t *GKEHeterogeneousStatusTrack) recoverCreate(ctx context.Context, result *deploy.Result) {
	if recovered := recover(); recovered != nil {
		err := errors.Errorf("panic in GKEHeterogeneousStatusTrack.Create: %v", recovered)
		t.logRecoverError(err, "panic in GKEHeterogeneousStatusTrack.Create", "stack", string(debug.Stack()))
		t.writePanicSnapshot(ctx, err)
		*result = deploy.Done().WithRequeuAfter(gkeHeterogeneousStatusRetryInterval)
	}
}

func (t *GKEHeterogeneousStatusTrack) recoverDelete(result *deploy.Result) {
	if recovered := recover(); recovered != nil {
		err := errors.Errorf("panic in GKEHeterogeneousStatusTrack.Delete: %v", recovered)
		t.logRecoverError(err, "panic in GKEHeterogeneousStatusTrack.Delete", "stack", string(debug.Stack()))
		*result = deploy.Done().WithRequeuAfter(gkeHeterogeneousStatusRetryInterval)
	}
}

func (t *GKEHeterogeneousStatusTrack) logRecoverError(err error, msg string, keysAndValues ...interface{}) {
	if t == nil || t.BaseTask == nil || t.Context == nil {
		return
	}
	t.Logger.Error(err, msg, keysAndValues...)
}

func (t *GKEHeterogeneousStatusTrack) writePanicSnapshot(ctx context.Context, err error) {
	defer func() {
		if recovered := recover(); recovered != nil {
			t.logRecoverError(
				errors.Errorf("panic while writing gke heterogeneous status snapshot after panic: %v", recovered),
				"failed to write gke heterogeneous status snapshot after panic",
				"stack",
				string(debug.Stack()),
			)
		}
	}()

	if !t.shouldWritePanicSnapshot() {
		return
	}

	intents, intentErr := t.resolveIntents()
	var writeResult deploy.Result
	switch {
	case intentErr != nil:
		writeResult = t.writeTerminalSnapshot(ctx, gkestatus.ProbeStatusFailed, true, false, err.Error(), nil)
	case len(intents) == 0:
		writeResult = t.writeDisabledSnapshotIfStatusConfigMapExists(ctx)
	default:
		writeResult = t.writeTerminalSnapshot(ctx, gkestatus.ProbeStatusFailed, true, true, err.Error(), nil)
	}
	if writeErr := writeResult.RawError(); writeErr != nil {
		t.logRecoverError(writeErr, "failed to write gke heterogeneous status snapshot after panic")
	}
}

func (t *GKEHeterogeneousStatusTrack) shouldWritePanicSnapshot() bool {
	if t == nil || t.BaseTask == nil || t.Context == nil || t.Cluster == nil {
		return false
	}
	if t.Config.DryRun.Enabled {
		return false
	}
	if t.Config.Provider != cloudv1.ProviderTypeGKE || t.Cluster.GetProviderType() != cloudv1.ProviderTypeGKE {
		return false
	}
	return !t.Cluster.Suspending() && !t.Cluster.Suspended()
}

func (t *GKEHeterogeneousStatusTrack) writeTerminalSnapshot(
	ctx context.Context,
	probeStatus string,
	enabled, validIntent bool,
	probeError string,
	scaleWorkflow *gkestatus.ScaleWorkflowStatus,
) deploy.Result {
	snapshot := t.baseSnapshot(probeStatus, enabled, validIntent, probeError)
	snapshot.ScaleWorkflow = scaleWorkflow
	if err := t.writeStatusSnapshot(ctx, snapshot); err != nil {
		return t.doneOrRetry(err, "write gke heterogeneous terminal status snapshot")
	}
	return deploy.Done()
}

func (t *GKEHeterogeneousStatusTrack) writeDisabledSnapshotIfStatusConfigMapExists(ctx context.Context) deploy.Result {
	cm, err := t.getStatusConfigMap(ctx)
	if apierrors.IsNotFound(err) {
		return deploy.Done()
	}
	if err != nil {
		return t.doneOrRetry(errors.Wrap(err, "get gke heterogeneous status configmap"), "get gke heterogeneous status configmap for disabled intent")
	}
	if statusConfigMapAlreadyDisabled(cm) {
		return deploy.Done()
	}
	return t.writeTerminalSnapshot(ctx, gkestatus.ProbeStatusDisabled, false, true, "", nil)
}

func (t *GKEHeterogeneousStatusTrack) writeNoIntentSnapshot(ctx context.Context, oldSnapshot *gkestatus.Snapshot) deploy.Result {
	scaleWorkflow := t.probeDisableScaleWorkflow(ctx, oldSnapshot)
	if scaleWorkflow != nil {
		return t.writeDisableWorkflowSnapshot(ctx, oldSnapshot, scaleWorkflow)
	}
	if terminalDisableWorkflowSnapshot(oldSnapshot) {
		return deploy.Done()
	}
	return t.writeDisabledSnapshotIfStatusConfigMapExists(ctx)
}

func (t *GKEHeterogeneousStatusTrack) writeDisableWorkflowSnapshot(
	ctx context.Context,
	oldSnapshot *gkestatus.Snapshot,
	scaleWorkflow *gkestatus.ScaleWorkflowStatus,
) deploy.Result {
	probeStatus := gkestatus.ProbeStatusDisabling
	probeError := ""
	switch cloudv1.WorkflowPhase(scaleWorkflow.WorkflowPhase) {
	case cloudv1.WorkflowPhaseCompleted:
		probeStatus = gkestatus.ProbeStatusDisabled
	case cloudv1.WorkflowPhaseFailed, cloudv1.WorkflowPhaseCancelled:
		probeStatus = gkestatus.ProbeStatusFailed
		probeError = "disable gke heterogeneous scale workflow " + strings.ToLower(scaleWorkflow.WorkflowPhase)
	}

	snapshot := t.baseSnapshot(probeStatus, false, true, probeError)
	if oldSnapshot != nil {
		snapshot.LastPlacementProbeTime = oldSnapshot.LastPlacementProbeTime
		snapshot.LastSuccessfulPlacementProbeTime = oldSnapshot.LastSuccessfulPlacementProbeTime
		snapshot.ClusterFullSynced = oldSnapshot.ClusterFullSynced
	}
	snapshot.ScaleWorkflow = scaleWorkflow
	if err := t.writeStatusSnapshot(ctx, snapshot); err != nil {
		return t.doneOrRetry(err, "write gke heterogeneous disable workflow status snapshot")
	}
	if probeStatus == gkestatus.ProbeStatusDisabling {
		return deploy.Done().WithRequeuAfter(gkeHeterogeneousStatusRetryInterval)
	}
	return deploy.Done()
}

func (t *GKEHeterogeneousStatusTrack) doneOrRetry(err error, message string) deploy.Result {
	if err == nil {
		return deploy.Done()
	}
	t.Logger.Error(err, message)
	return deploy.Done().WithRequeuAfter(gkeHeterogeneousStatusRetryInterval)
}

func (t *GKEHeterogeneousStatusTrack) updateThrottledSnapshot(
	ctx context.Context,
	oldSnapshot *gkestatus.Snapshot,
	scaleWorkflow *gkestatus.ScaleWorkflowStatus,
	clusterFullSynced bool,
) (bool, error) {
	if oldSnapshot == nil {
		return false, nil
	}

	// Only the expensive placement probe is throttled. Workflow status and the
	// unsynced marker are cheap metadata updates.
	next := *oldSnapshot
	workflowUpdated := false
	snapshotUpdated := false
	if !reflect.DeepEqual(oldSnapshot.ScaleWorkflow, scaleWorkflow) {
		next.ScaleWorkflow = scaleWorkflow
		workflowUpdated = true
		snapshotUpdated = true
	}
	if !clusterFullSynced && next.ClusterFullSynced {
		next.ClusterFullSynced = false
		snapshotUpdated = true
	}
	if !snapshotUpdated {
		return workflowUpdated, nil
	}
	return workflowUpdated, t.writeStatusSnapshot(ctx, next)
}

func (t *GKEHeterogeneousStatusTrack) writeFailedPlacementSnapshot(
	ctx context.Context,
	oldSnapshot *gkestatus.Snapshot,
	scaleWorkflow *gkestatus.ScaleWorkflowStatus,
	clusterFullSynced bool,
	now time.Time,
	probeErr error,
) deploy.Result {
	// Placement is all-or-nothing, but workflow identity must survive failed
	// probes so a later reconcile can recover terminal duration after the
	// workflow annotation has been cleaned up.
	failedSnapshot := t.baseSnapshot(gkestatus.ProbeStatusFailed, true, true, probeErr.Error())
	failedSnapshot.ClusterFullSynced = clusterFullSynced
	failedSnapshot.LastPlacementProbeTime = timePtr(now)
	failedSnapshot.ScaleWorkflow = scaleWorkflow
	if oldSnapshot != nil {
		failedSnapshot.LastSuccessfulPlacementProbeTime = oldSnapshot.LastSuccessfulPlacementProbeTime
	}
	if err := t.writeStatusSnapshot(ctx, failedSnapshot); err != nil {
		return t.doneOrRetry(err, "write gke heterogeneous failed placement status snapshot")
	}
	return deploy.Done().WithRequeuAfter(gkeHeterogeneousStatusMinProbeInterval)
}

func (t *GKEHeterogeneousStatusTrack) baseSnapshot(probeStatus string, enabled, validIntent bool, probeError string) gkestatus.Snapshot {
	return gkestatus.Snapshot{
		ClusterID:         t.Cluster.Name,
		Namespace:         t.Cluster.Namespace,
		TenantID:          t.Cluster.GetTenantID(),
		ProjectID:         t.Cluster.GetProjectID(),
		ClusterGeneration: t.Cluster.Generation,
		ProbeStatus:       probeStatus,
		ProbeError:        probeError,
		Enabled:           enabled,
		ValidIntent:       validIntent,
		Components:        nil,
	}
}

func (t *GKEHeterogeneousStatusTrack) resolveIntents() ([]gkeHeterogeneousIntent, error) {
	out := make([]gkeHeterogeneousIntent, 0)
	appendIntent := func(componentType cloudv1.ComponentType, tidbGroupID, annotationKey, baseInstanceType string, desiredReplicas int32) error {
		if strings.TrimSpace(t.Cluster.GetAnnotations()[annotationKey]) == "" {
			return nil
		}
		annotation, explicit, err := cloudv1.ClusterGKEComputeClassBindingAnnotation(
			t.Cluster,
			annotationKey,
			baseInstanceType,
			gkeheterogeneous.NodePoolName(t.Cluster, baseInstanceType),
		)
		if err != nil {
			return err
		}
		if !explicit {
			return errors.Errorf("gke heterogeneous annotation %s is not explicit", annotationKey)
		}
		out = append(out, gkeHeterogeneousIntent{
			componentType:    componentType,
			tidbGroupID:      strings.TrimSpace(tidbGroupID),
			computeClassName: strings.TrimSpace(annotation.ComputeClassName),
			baseInstanceType: strings.TrimSpace(baseInstanceType),
			desiredReplicas:  desiredReplicas,
			nodePools:        annotation.NodePools,
		})
		return nil
	}

	if err := appendIntent(
		cloudv1.ComponentTiKV,
		"",
		cloudv1.AnnoKeyGKEComputeClassForTiKVBindings,
		strings.TrimSpace(t.Cluster.Spec.Cluster.TiKV.InstanceType),
		int32(t.Cluster.Spec.Cluster.TiKV.GetReplicas()),
	); err != nil {
		return nil, errors.Wrap(err, "resolve tikv gke heterogeneous intent")
	}

	for _, groupID := range gkeheterogeneous.ActiveTiDBGroupIDs(t.Cluster) {
		component, err := t.Cluster.GetComponentSpecInGroup(groupID, cloudv1.ComponentTiDB)
		if err != nil {
			return nil, errors.Wrapf(err, "get tidb component for group %s", groupID)
		}
		if component == nil {
			return nil, errors.Errorf("tidb component for group %s is nil", groupID)
		}
		desiredReplicas := int32(component.GetReplicas())
		if err := appendIntent(
			cloudv1.ComponentTiDB,
			groupID,
			t.Cluster.TiDBGKEComputeClassBindingsAnnoKey(groupID),
			t.Cluster.TiDBInstanceTypeForGroup(groupID),
			desiredReplicas,
		); err != nil {
			return nil, errors.Wrapf(err, "resolve tidb gke heterogeneous intent for group %s", groupID)
		}
	}

	return out, nil
}

func (t *GKEHeterogeneousStatusTrack) probeScaleWorkflow(
	ctx context.Context,
	oldSnapshot *gkestatus.Snapshot,
	intents []gkeHeterogeneousIntent,
) *gkestatus.ScaleWorkflowStatus {
	// Workflow probing is best-effort: it should enrich duration metrics without
	// changing the all-or-nothing result of the placement probe.
	workflow, ok := t.currentScaleWorkflow(ctx)
	if !ok {
		return t.recoverScaleWorkflowFromOldSnapshot(ctx, oldSnapshot)
	}
	if !t.scaleWorkflowTargetsCurrentCluster(workflow) {
		return t.recoverScaleWorkflowFromOldSnapshot(ctx, oldSnapshot)
	}

	oldWorkflow := scaleWorkflowFromSnapshot(oldSnapshot)
	matchedOld := scaleWorkflowMatches(oldWorkflow, workflow)
	component, groupID, matchedIntent := t.matchScaleWorkflowIntent(workflow, intents)
	if !matchedIntent {
		if !matchedOld || oldWorkflow == nil {
			return staleOrTerminalScaleWorkflow(oldSnapshot)
		}
		component = oldWorkflow.ComponentType
		groupID = oldWorkflow.TiDBGroupID
	}

	return scaleWorkflowStatusFromWorkflow(workflow, component, groupID, gkestatus.ScaleWorkflowOperationScaleOut)
}

func (t *GKEHeterogeneousStatusTrack) probeDisableScaleWorkflow(ctx context.Context, oldSnapshot *gkestatus.Snapshot) *gkestatus.ScaleWorkflowStatus {
	workflow, ok := t.currentScaleWorkflow(ctx)
	if ok {
		if component, groupID, matched := t.matchDisableScaleWorkflow(workflow); matched {
			return scaleWorkflowStatusFromWorkflow(workflow, component, groupID, gkestatus.ScaleWorkflowOperationDisable)
		}
		return nil
	}
	if !runningDisableWorkflowSnapshot(oldSnapshot) {
		return nil
	}
	return t.recoverScaleWorkflowFromOldSnapshot(ctx, oldSnapshot)
}

func scaleWorkflowStatusFromWorkflow(workflow *cloudv1.Workflow, component, groupID, operation string) *gkestatus.ScaleWorkflowStatus {
	if workflow == nil {
		return nil
	}
	status := &gkestatus.ScaleWorkflowStatus{
		WorkflowNamespace: workflow.Namespace,
		WorkflowName:      workflow.Name,
		WorkflowUID:       string(workflow.UID),
		WorkflowPhase:     string(workflow.Status.Phase),
		Operation:         operation,
		ComponentType:     component,
		TiDBGroupID:       groupID,
	}
	if workflow.Status.BeginAt != nil {
		beginAt := workflow.Status.BeginAt.Time
		status.BeginAt = &beginAt
	}
	if workflow.Done() {
		if workflow.Status.EndAt != nil {
			endAt := workflow.Status.EndAt.Time
			status.EndAt = &endAt
			if status.BeginAt != nil {
				duration := status.EndAt.Sub(*status.BeginAt).Seconds()
				status.DurationSeconds = &duration
			}
		}
		return status
	}

	status.InProgress = true
	return status
}

func (t *GKEHeterogeneousStatusTrack) recoverScaleWorkflowFromOldSnapshot(ctx context.Context, oldSnapshot *gkestatus.Snapshot) *gkestatus.ScaleWorkflowStatus {
	oldWorkflow := scaleWorkflowFromSnapshot(oldSnapshot)
	if oldWorkflow == nil || oldWorkflow.DurationSeconds != nil || oldWorkflow.EndAt != nil {
		return staleOrTerminalScaleWorkflow(oldSnapshot)
	}
	if oldWorkflow.WorkflowNamespace == "" || oldWorkflow.WorkflowName == "" {
		return staleOrTerminalScaleWorkflow(oldSnapshot)
	}

	workflow := &cloudv1.Workflow{}
	if err := t.Local.Get(ctx, client.ObjectKey{Namespace: oldWorkflow.WorkflowNamespace, Name: oldWorkflow.WorkflowName}, workflow); err != nil {
		if !apierrors.IsNotFound(err) {
			t.Logger.Error(err, "failed to recover terminal gke heterogeneous scale workflow", "workflow", oldWorkflow.WorkflowNamespace+"/"+oldWorkflow.WorkflowName)
		}
		return staleOrTerminalScaleWorkflow(oldSnapshot)
	}
	if !t.scaleWorkflowTargetsCurrentCluster(workflow) || !scaleWorkflowMatches(oldWorkflow, workflow) || !workflow.Done() {
		return staleOrTerminalScaleWorkflow(oldSnapshot)
	}
	operation := oldWorkflow.Operation
	if operation == "" {
		operation = scaleWorkflowOperationFromOldSnapshot(oldSnapshot)
	}
	return scaleWorkflowStatusFromWorkflow(workflow, oldWorkflow.ComponentType, oldWorkflow.TiDBGroupID, operation)
}

func scaleWorkflowOperationFromOldSnapshot(snapshot *gkestatus.Snapshot) string {
	if snapshot != nil && !snapshot.Enabled {
		return gkestatus.ScaleWorkflowOperationDisable
	}
	return gkestatus.ScaleWorkflowOperationScaleOut
}

func (t *GKEHeterogeneousStatusTrack) currentScaleWorkflow(ctx context.Context) (*cloudv1.Workflow, bool) {
	if t.Cluster.Annotations == nil {
		return nil, false
	}
	workflowData := strings.TrimSpace(t.Cluster.Annotations[consts.AnnoKeyScaleClusterWorkflow])
	if workflowData == "" {
		return nil, false
	}

	workflowRef := &cloudv1.Workflow{}
	if err := yaml.Unmarshal([]byte(workflowData), workflowRef); err != nil {
		t.Logger.Error(err, "failed to unmarshal scale workflow annotation for gke heterogeneous status")
		return nil, false
	}
	if workflowRef.Name == "" || workflowRef.Namespace == "" {
		t.Logger.Info("skip gke heterogeneous scale workflow without namespace or name", "workflow", workflowRef.NamespacedName())
		return nil, false
	}

	workflow := &cloudv1.Workflow{}
	if err := t.Local.Get(ctx, client.ObjectKey{Namespace: workflowRef.Namespace, Name: workflowRef.Name}, workflow); err != nil {
		t.Logger.Error(err, "failed to get scale workflow for gke heterogeneous status", "workflow", workflowRef.NamespacedName())
		return nil, false
	}
	if workflowRef.UID != "" && workflow.UID != "" && workflowRef.UID != workflow.UID {
		t.Logger.Info("skip gke heterogeneous scale workflow with mismatched uid", "workflow", workflowRef.NamespacedName(), "annotationUID", workflowRef.UID, "workflowUID", workflow.UID)
		return nil, false
	}
	return workflow, true
}

func (t *GKEHeterogeneousStatusTrack) matchScaleWorkflowIntent(workflow *cloudv1.Workflow, intents []gkeHeterogeneousIntent) (component, groupID string, matched bool) {
	if workflow == nil {
		return "", "", false
	}
	if !t.scaleWorkflowTargetsCurrentCluster(workflow) {
		return "", "", false
	}
	if workflow.Spec.Type != "" && workflow.Spec.Type != consts.WorkflowTypeScaleCluster {
		return "", "", false
	}
	// Real workflow step inputs keep $wf.spec.inputs.* references in spec. The
	// workflow-level inputs are the persisted source of truth for target metadata.
	input := gkeHeterogeneousScaleWorkflowInput{
		GroupID:               strings.TrimSpace(workflow.Spec.Inputs[consts.WfInputKeyGroupID]),
		ComponentType:         strings.TrimSpace(workflow.Spec.Inputs[consts.WfInputKeyComponentType]),
		ComponentSpec:         workflow.Spec.Inputs[consts.WfInputKeyScaleClusterDesiredComponentSpec],
		OriginalComponentSpec: workflow.Spec.Inputs[consts.WfInputKeyScaleClusterOriginalComponentSpec],
	}
	for _, step := range workflow.Spec.Steps {
		if step.Uses != consts.WorkflowActionScaleCluster {
			continue
		}
		stepInput := struct {
			Scopes []string `mapstructure:"scopes"`
		}{}
		if err := mapstructure.Decode(step.Inputs.Object, &stepInput); err != nil {
			t.Logger.Error(err, "failed to decode scale workflow step inputs for gke heterogeneous status", "workflow", workflow.NamespacedName(), "step", step.Name)
			continue
		}
		if !slices.Contains(stepInput.Scopes, consts.WfScaleClusterScopeReplicas) {
			continue
		}
		var matchedComponent, matchedGroupID string
		for _, intent := range intents {
			if input.ComponentType != string(intent.componentType) {
				continue
			}
			if intent.componentType == cloudv1.ComponentTiDB && input.GroupID != intent.tidbGroupID {
				continue
			}
			matchedComponent = string(intent.componentType)
			matchedGroupID = intent.tidbGroupID
			break
		}
		if matchedComponent == "" {
			continue
		}
		scaleOut, err := replicasScaleOutWorkflowInput(input)
		if err != nil {
			t.Logger.Error(err, "failed to check replicas scale-out workflow input for gke heterogeneous status", "workflow", workflow.NamespacedName(), "step", step.Name)
			continue
		}
		if !scaleOut {
			continue
		}
		return matchedComponent, matchedGroupID, true
	}
	return "", "", false
}

func (t *GKEHeterogeneousStatusTrack) matchDisableScaleWorkflow(workflow *cloudv1.Workflow) (component, groupID string, matched bool) {
	if workflow == nil {
		return "", "", false
	}
	if !t.scaleWorkflowTargetsCurrentCluster(workflow) {
		return "", "", false
	}
	if workflow.Spec.Type != "" && workflow.Spec.Type != consts.WorkflowTypeScaleCluster {
		return "", "", false
	}
	if !strings.EqualFold(strings.TrimSpace(workflow.Spec.Inputs[consts.WfInputKeyScaleClusterDisableGKEHeterogeneous]), "true") {
		return "", "", false
	}
	component = strings.TrimSpace(workflow.Spec.Inputs[consts.WfInputKeyComponentType])
	if component == "" {
		return "", "", false
	}
	return component, strings.TrimSpace(workflow.Spec.Inputs[consts.WfInputKeyGroupID]), true
}

func (t *GKEHeterogeneousStatusTrack) scaleWorkflowTargetsCurrentCluster(workflow *cloudv1.Workflow) bool {
	if workflow == nil {
		return false
	}
	if strings.TrimSpace(workflow.Spec.Inputs[consts.WfInputKeyNamespace]) != t.Cluster.Namespace {
		return false
	}
	if strings.TrimSpace(workflow.Spec.Inputs[consts.WfInputKeyClusterID]) != t.Cluster.Name {
		return false
	}
	labelCluster := strings.TrimSpace(workflow.Labels[consts.ClusterLabelKey])
	return labelCluster == "" || labelCluster == t.Cluster.Name
}

func replicasScaleOutWorkflowInput(input gkeHeterogeneousScaleWorkflowInput) (bool, error) {
	originalComp, err := newGKEHeterogeneousScaleWorkflowComponent(input.ComponentType)
	if err != nil {
		return false, err
	}
	desiredComp, err := newGKEHeterogeneousScaleWorkflowComponent(input.ComponentType)
	if err != nil {
		return false, err
	}
	if strings.TrimSpace(input.OriginalComponentSpec) == "" || strings.TrimSpace(input.ComponentSpec) == "" {
		return false, errors.New("scale workflow component specs are empty")
	}
	if err := yaml.Unmarshal([]byte(input.OriginalComponentSpec), originalComp); err != nil {
		return false, errors.Wrap(err, "unmarshal original scale workflow component spec")
	}
	if err := yaml.Unmarshal([]byte(input.ComponentSpec), desiredComp); err != nil {
		return false, errors.Wrap(err, "unmarshal desired scale workflow component spec")
	}
	return desiredComp.GetReplicas() > originalComp.GetReplicas(), nil
}

func newGKEHeterogeneousScaleWorkflowComponent(componentType string) (cloudv1.ClusterComponent, error) {
	switch componentType {
	case string(cloudv1.ComponentTiKV):
		return &cloudv1.TiKV{}, nil
	case string(cloudv1.ComponentTiDB):
		return &cloudv1.TiDB{}, nil
	default:
		return nil, errors.Errorf("unsupported gke heterogeneous scale workflow component type %s", componentType)
	}
}

func scaleWorkflowFromSnapshot(snapshot *gkestatus.Snapshot) *gkestatus.ScaleWorkflowStatus {
	if snapshot == nil || snapshot.ScaleWorkflow == nil {
		return nil
	}
	workflow := *snapshot.ScaleWorkflow
	return &workflow
}

func scaleWorkflowMatches(status *gkestatus.ScaleWorkflowStatus, workflow *cloudv1.Workflow) bool {
	if status == nil || workflow == nil {
		return false
	}
	if status.WorkflowUID != "" && string(workflow.UID) != "" {
		return status.WorkflowUID == string(workflow.UID)
	}
	return status.WorkflowNamespace == workflow.Namespace && status.WorkflowName == workflow.Name
}

func staleOrTerminalScaleWorkflow(snapshot *gkestatus.Snapshot) *gkestatus.ScaleWorkflowStatus {
	workflow := scaleWorkflowFromSnapshot(snapshot)
	if workflow == nil {
		return nil
	}
	if workflow.DurationSeconds != nil || workflow.EndAt != nil {
		workflow.InProgress = false
		workflow.Stale = false
		return workflow
	}
	if workflow.InProgress {
		workflow.Stale = true
		workflow.WorkflowPhase = gkeHeterogeneousScaleWorkflowStalePhase
		return workflow
	}
	return nil
}

func runningDisableWorkflowSnapshot(snapshot *gkestatus.Snapshot) bool {
	if snapshot == nil || snapshot.ScaleWorkflow == nil {
		return false
	}
	return snapshot.ProbeStatus == gkestatus.ProbeStatusDisabling &&
		!snapshot.Enabled &&
		snapshot.ValidIntent &&
		snapshot.ScaleWorkflow.InProgress &&
		!snapshot.ScaleWorkflow.Stale
}

func terminalDisableWorkflowSnapshot(snapshot *gkestatus.Snapshot) bool {
	if snapshot == nil || snapshot.ScaleWorkflow == nil {
		return false
	}
	if snapshot.Enabled || !snapshot.ValidIntent || snapshot.ScaleWorkflow.InProgress || snapshot.ScaleWorkflow.Stale {
		return false
	}
	if !terminalScaleWorkflow(snapshot.ScaleWorkflow) {
		return false
	}
	return snapshot.ProbeStatus == gkestatus.ProbeStatusDisabled ||
		snapshot.ProbeStatus == gkestatus.ProbeStatusFailed
}

func terminalScaleWorkflow(workflow *gkestatus.ScaleWorkflowStatus) bool {
	if workflow == nil || workflow.InProgress || workflow.Stale {
		return false
	}
	switch cloudv1.WorkflowPhase(workflow.WorkflowPhase) {
	case cloudv1.WorkflowPhaseCompleted, cloudv1.WorkflowPhaseFailed, cloudv1.WorkflowPhaseCancelled:
		return true
	default:
		return workflow.EndAt != nil || workflow.DurationSeconds != nil
	}
}

func placementProbeThrottleRemaining(snapshot *gkestatus.Snapshot, now time.Time) time.Duration {
	if snapshot == nil || snapshot.LastPlacementProbeTime == nil {
		return 0
	}
	// Throttle is based on the latest placement attempt, not the latest
	// successful placement.
	nextProbeTime := snapshot.LastPlacementProbeTime.Add(gkeHeterogeneousStatusMinProbeInterval)
	if !now.Before(nextProbeTime) {
		return 0
	}
	return nextProbeTime.Sub(now)
}

func timePtr(t time.Time) *time.Time {
	return &t
}

func (t *GKEHeterogeneousStatusTrack) probePlacement(ctx context.Context, intents []gkeHeterogeneousIntent) (gkestatus.Snapshot, error) {
	snapshot := t.baseSnapshot(gkestatus.ProbeStatusSuccess, true, true, "")
	remoteCli := t.EnvProvider.GetClient()
	componentPods := make([]gkeHeterogeneousComponentPods, 0, len(intents))
	for _, intent := range intents {
		pods, err := t.listIntentPods(ctx, intent)
		if err != nil {
			return gkestatus.Snapshot{}, err
		}
		componentPods = append(componentPods, gkeHeterogeneousComponentPods{
			intent: intent,
			pods:   pods,
		})
	}

	nodeList := &corev1.NodeList{}
	if err := remoteCli.List(ctx, nodeList); err != nil {
		return gkestatus.Snapshot{}, errors.Wrap(err, "list dataplane nodes")
	}

	nodesByName := make(map[string]*corev1.Node, len(nodeList.Items))
	for i := range nodeList.Items {
		node := &nodeList.Items[i]
		nodesByName[node.Name] = node
	}

	for _, component := range componentPods {
		componentStatus := t.probeComponent(component.intent, component.pods, nodesByName)
		if componentStatus.DesiredReplicas <= 0 && len(componentStatus.Pods) == 0 {
			continue
		}
		snapshot.Components = append(snapshot.Components, componentStatus)
	}
	sort.Slice(snapshot.Components, func(i, j int) bool {
		if snapshot.Components[i].ComponentType != snapshot.Components[j].ComponentType {
			return snapshot.Components[i].ComponentType < snapshot.Components[j].ComponentType
		}
		return snapshot.Components[i].TiDBGroupID < snapshot.Components[j].TiDBGroupID
	})
	// Counters and node usages are computed from the full placement list before
	// the persisted snapshot is compacted to pod name groups.
	snapshot.Components, snapshot.ClusterNodeUsages = gkestatus.PopulateNodeUsages(snapshot.Components)
	snapshot.ClusterNodeSummary = gkestatus.NodeSummaryFromComponents(snapshot.Components)
	return gkestatus.CompactSnapshot(snapshot), nil
}

func (t *GKEHeterogeneousStatusTrack) probeComponent(intent gkeHeterogeneousIntent, pods []corev1.Pod, nodesByName map[string]*corev1.Node) gkestatus.ComponentStatus {
	sort.Slice(pods, func(i, j int) bool {
		return pods[i].Name < pods[j].Name
	})

	status := gkestatus.ComponentStatus{
		ComponentType:    string(intent.componentType),
		TiDBGroupID:      intent.tidbGroupID,
		ComputeClassName: intent.computeClassName,
		BaseInstanceType: intent.baseInstanceType,
		DesiredReplicas:  intent.desiredReplicas,
		TotalPods:        int32(len(pods)),
	}
	for _, pod := range pods {
		placement := t.classifyPod(intent, pod, nodesByName[pod.Spec.NodeName])
		status.Pods = append(status.Pods, placement)
		switch placement.PlacementStatus {
		case gkestatus.PlacementStatusHeterogeneous:
			status.HeterogeneousPods++
			status.ScheduledPods++
		case gkestatus.PlacementStatusBase:
			status.BasePods++
			status.ScheduledPods++
		case gkestatus.PlacementStatusMismatch:
			status.MismatchPods++
			status.ScheduledPods++
		case gkestatus.PlacementStatusPending:
			status.PendingPods++
		case gkestatus.PlacementStatusUnknown:
			status.UnknownPods++
			if placement.Node != "" {
				status.ScheduledPods++
			}
		}
	}
	status.PlacementRatio = ratio(status.HeterogeneousPods, status.ScheduledPods)
	status.EffectiveRatio = ratio(status.HeterogeneousPods, status.DesiredReplicas)
	return status
}

func (t *GKEHeterogeneousStatusTrack) listIntentPods(ctx context.Context, intent gkeHeterogeneousIntent) ([]corev1.Pod, error) {
	podNamer := t.Namer.WithComponent(string(intent.componentType))
	if intent.componentType == cloudv1.ComponentTiDB && intent.tidbGroupID != t.Cluster.GetDefaultGroupID() {
		podNamer = podNamer.WithTiDBGroupName(intent.tidbGroupID)
	}
	pods, err := k8s.SelectPods(ctx, t.EnvProvider.GetClient(), t.Namer.GetTiDBNamespace(), podNamer.GetPodLabels())
	if err != nil {
		return nil, errors.Wrapf(err, "list %s pods for tidb group %s", intent.componentType, intent.tidbGroupID)
	}
	return pods, nil
}

func (t *GKEHeterogeneousStatusTrack) classifyPod(intent gkeHeterogeneousIntent, pod corev1.Pod, node *corev1.Node) gkestatus.PodPlacement {
	placement := gkestatus.PodPlacement{
		Pod:             pod.Name,
		PlacementStatus: gkestatus.PlacementStatusPending,
	}
	if pod.Spec.NodeName == "" {
		return placement
	}
	placement.Node = pod.Spec.NodeName
	if node == nil {
		placement.PlacementStatus = gkestatus.PlacementStatusUnknown
		return placement
	}

	labels := node.GetLabels()
	placement.NodePool = namer.GetNodeGroupNameFromNode(labels, t.Config.Provider, false)
	placement.Zone = namer.GetNodeZoneFromLabels(labels)
	placement.ActualInstanceType = namer.GetNodeInstanceTypeFromLabels(labels)
	placement.ActualComputeClass = labels[consts.GKEComputeClassLabelKey]
	if placement.NodePool == "" || placement.ActualInstanceType == "" {
		placement.PlacementStatus = gkestatus.PlacementStatusUnknown
		return placement
	}

	bindingByNodePool, bindingByInstanceType := intentBindingMaps(intent)
	binding, matchedNodePool := bindingByNodePool[placement.NodePool]
	if !matchedNodePool {
		if instanceBinding, ok := bindingByInstanceType[placement.ActualInstanceType]; ok {
			placement.ExpectedRole = instanceBinding.Role
			placement.ExpectedInstanceType = instanceBinding.InstanceType
			placement.ExpectedComputeClass = intent.computeClassName
		}
		placement.PlacementStatus = gkestatus.PlacementStatusMismatch
		return placement
	}

	placement.ExpectedRole = binding.Role
	placement.ExpectedInstanceType = binding.InstanceType
	placement.ExpectedComputeClass = intent.computeClassName
	if binding.Role == cloudv1.GKEComputeClassNodePoolRoleExtra &&
		placement.ActualInstanceType == binding.InstanceType &&
		placement.ActualComputeClass == intent.computeClassName {
		placement.IsHeterogeneous = true
		placement.PlacementStatus = gkestatus.PlacementStatusHeterogeneous
		return placement
	}
	if binding.Role == cloudv1.GKEComputeClassNodePoolRoleBase &&
		placement.ActualInstanceType == binding.InstanceType {
		placement.PlacementStatus = gkestatus.PlacementStatusBase
		return placement
	}
	placement.PlacementStatus = gkestatus.PlacementStatusMismatch
	return placement
}

func intentBindingMaps(intent gkeHeterogeneousIntent) (map[string]cloudv1.GKEComputeClassNodePoolBinding, map[string]cloudv1.GKEComputeClassNodePoolBinding) {
	byNodePool := make(map[string]cloudv1.GKEComputeClassNodePoolBinding, len(intent.nodePools))
	byInstanceType := make(map[string]cloudv1.GKEComputeClassNodePoolBinding, len(intent.nodePools))
	for _, binding := range intent.nodePools {
		byNodePool[binding.NodePoolName] = binding
		byInstanceType[binding.InstanceType] = binding
	}
	return byNodePool, byInstanceType
}

func ratio(numerator, denominator int32) *float64 {
	if denominator <= 0 {
		return nil
	}
	value := float64(numerator) / float64(denominator)
	return &value
}

func (t *GKEHeterogeneousStatusTrack) writeStatusSnapshot(ctx context.Context, snapshot gkestatus.Snapshot) error {
	snapshot = gkestatus.CompactSnapshot(snapshot)
	data, err := yaml.Marshal(snapshot)
	if err != nil {
		return errors.Wrap(err, "marshal gke heterogeneous status snapshot")
	}
	cm := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: t.Cluster.Namespace,
			Name:      gkestatus.ConfigMapNamePrefix + t.Cluster.Name,
		},
	}
	_, err = controllerutil.CreateOrUpdate(ctx, t.Local, cm, func() error {
		if err := controllerutil.SetOwnerReference(t.Cluster, cm, t.Local.Scheme()); err != nil {
			return errors.Wrap(err, "set cluster owner reference for gke heterogeneous status configmap")
		}
		cm.Labels = maps.Clone(t.Cluster.Labels)
		cm.Data = map[string]string{
			gkestatus.ConfigMapDataKey: string(data),
		}
		return nil
	})
	return errors.Wrap(err, "create or update gke heterogeneous status configmap")
}

func (t *GKEHeterogeneousStatusTrack) getStatusConfigMap(ctx context.Context) (*corev1.ConfigMap, error) {
	cm := &corev1.ConfigMap{}
	if err := t.Local.Get(ctx, client.ObjectKey{
		Namespace: t.Cluster.Namespace,
		Name:      gkestatus.ConfigMapNamePrefix + t.Cluster.Name,
	}, cm); err != nil {
		return nil, err
	}
	return cm, nil
}

func (t *GKEHeterogeneousStatusTrack) readStatusSnapshot(ctx context.Context) (*gkestatus.Snapshot, error) {
	cm, err := t.getStatusConfigMap(ctx)
	if apierrors.IsNotFound(err) {
		return nil, nil
	}
	if err != nil {
		return nil, errors.Wrap(err, "get gke heterogeneous status configmap")
	}
	snapshot, err := statusSnapshotFromConfigMap(cm)
	if err != nil {
		t.Logger.Error(err, "failed to unmarshal existing gke heterogeneous status snapshot")
		return nil, nil
	}
	return snapshot, nil
}

func statusConfigMapAlreadyDisabled(cm *corev1.ConfigMap) bool {
	snapshot, err := statusSnapshotFromConfigMap(cm)
	if err != nil {
		return false
	}
	return snapshot.ProbeStatus == gkestatus.ProbeStatusDisabled &&
		!snapshot.Enabled &&
		snapshot.ValidIntent &&
		snapshot.ProbeError == "" &&
		(snapshot.ScaleWorkflow == nil || terminalScaleWorkflow(snapshot.ScaleWorkflow))
}

func statusSnapshotFromConfigMap(cm *corev1.ConfigMap) (*gkestatus.Snapshot, error) {
	if cm == nil || cm.Data == nil {
		return nil, errors.New("gke heterogeneous status configmap data is empty")
	}
	data := cm.Data[gkestatus.ConfigMapDataKey]
	if data == "" {
		return nil, errors.Errorf("gke heterogeneous status configmap data key %q is empty", gkestatus.ConfigMapDataKey)
	}
	snapshot := &gkestatus.Snapshot{}
	if err := yaml.Unmarshal([]byte(data), snapshot); err != nil {
		return nil, err
	}
	return snapshot, nil
}