package nodepoolgc

import (
	"context"
	"errors"
	"testing"
	"time"

	ec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1"
	"github.com/crossplane-contrib/provider-aws/apis/eks/manualv1alpha1"
	gkev1beta1 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	"github.com/go-logr/logr"
	"github.com/prometheus/client_golang/prometheus/testutil"
	"github.com/stretchr/testify/require"
	azurev1beta1 "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/runtime"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/cache"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	regionalv1alpha1 "github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/metrics/collectors"
)

func gcCandidateLabels(component cloudv1.ComponentType) map[string]string {
	return map[string]string{
		consts.LabelKeyNodePoolGCManaged: consts.AnnoValTrue,
		consts.TagKeyTenant:              "tenant-1",
		consts.TagKeyProject:             "project-1",
		consts.TagKeyCluster:             "cluster-1",
		consts.TagKeyUsedBy:              consts.UsedByCustomerTiDB,
		cloudv1.TagLabelServiceType:      cloudv1.ServiceTypeDedicated,
		namer.ComponentSelectorKey:       string(component),
		consts.K8sVersionLabelKey:        "1.32",
	}
}

func gcTestScheme(t *testing.T) *runtime.Scheme {
	t.Helper()
	scheme := runtime.NewScheme()
	require.NoError(t, manualv1alpha1.SchemeBuilder.AddToScheme(scheme))
	require.NoError(t, ec2v1alpha1.SchemeBuilder.AddToScheme(scheme))
	require.NoError(t, gkev1beta1.SchemeBuilder.AddToScheme(scheme))
	require.NoError(t, azurev1beta1.SchemeBuilder.AddToScheme(scheme))
	require.NoError(t, cloudv1.AddToScheme(scheme))
	require.NoError(t, corev1.AddToScheme(scheme))
	require.NoError(t, regionalv1alpha1.AddToScheme(scheme))
	return scheme
}

func approvedNodeGroup(name string, now time.Time) *manualv1alpha1.NodeGroup {
	return &manualv1alpha1.NodeGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:   name,
			Labels: gcCandidateLabels(cloudv1.ComponentBackup),
			Annotations: map[string]string{
				consts.AnnoKeyNodePoolLastUsedTime: now.Add(-15 * 24 * time.Hour).Format(time.RFC3339),
				consts.AnnoKeyNodePoolGCApprovedAt: now.Add(-time.Hour).Format(time.RFC3339),
			},
		},
		Spec: manualv1alpha1.NodeGroupSpec{
			ForProvider: manualv1alpha1.NodeGroupParameters{Region: "us-west-2"},
		},
	}
}

func testGC(t *testing.T, cfg config.NodePoolGCConfig, objects ...client.Object) (*GC, client.Client) {
	t.Helper()
	local := fake.NewClientBuilder().WithScheme(gcTestScheme(t)).WithObjects(objects...).Build()
	gc := &GC{
		local:             local,
		logger:            logr.Discard(),
		config:            configWithDefaults(cfg),
		providers:         []CloudProvider{mustCloudProvider(t, consts.ProviderTypeEKS, local)},
		dataplaneResolver: emptyDataplaneResolver(t),
		now:               time.Now,
	}
	return gc, local
}

func emptyDataplaneResolver(t *testing.T) func(context.Context, nodePool) (client.Client, error) {
	t.Helper()
	remote := fake.NewClientBuilder().WithScheme(gcTestScheme(t)).Build()
	return func(context.Context, nodePool) (client.Client, error) {
		return remote, nil
	}
}

func matchingDataplaneResolver(t *testing.T, selector map[string]string) func(context.Context, nodePool) (client.Client, error) {
	t.Helper()
	node := &corev1.Node{
		ObjectMeta: metav1.ObjectMeta{
			Name:   "node-1",
			Labels: selector,
		},
	}
	remote := fake.NewClientBuilder().WithScheme(gcTestScheme(t)).WithObjects(node).Build()
	return func(context.Context, nodePool) (client.Client, error) {
		return remote, nil
	}
}

func mustCloudProvider(
	t *testing.T,
	providerType string,
	local client.Client,
) CloudProvider {
	t.Helper()
	provider, err := newCloudProvider(providerType, local)
	require.NoError(t, err)
	return provider
}

func TestListNodePoolsAcrossProviders(t *testing.T) {
	aws := &manualv1alpha1.NodeGroup{
		ObjectMeta: metav1.ObjectMeta{Name: "aws", Labels: gcCandidateLabels(cloudv1.ComponentBackup)},
		Spec: manualv1alpha1.NodeGroupSpec{
			ForProvider: manualv1alpha1.NodeGroupParameters{Region: "us-west-2", ClusterName: "eks-dataplane"},
		},
	}
	launchTemplate := &ec2v1alpha1.LaunchTemplate{
		ObjectMeta: metav1.ObjectMeta{Name: "aws", Labels: gcCandidateLabels(cloudv1.ComponentBackup)},
	}
	gcp := &gkev1beta1.NodePool{
		ObjectMeta: metav1.ObjectMeta{Name: "gcp", Labels: gcCandidateLabels(cloudv1.ComponentRestore)},
		Spec: gkev1beta1.NodePoolSpec{
			ForProvider: gkev1beta1.NodePoolParameters{
				Cluster: "projects/p/locations/us-central1/clusters/c",
			},
		},
	}
	azure := &azurev1beta1.KubernetesClusterNodePool{
		ObjectMeta: metav1.ObjectMeta{Name: "azure", Labels: gcCandidateLabels(cloudv1.ComponentCompactBackup)},
		Spec: azurev1beta1.KubernetesClusterNodePoolSpec{ForProvider: azurev1beta1.KubernetesClusterNodePoolParameters{
			KubernetesClusterID: ptr.To("/subscriptions/s/resourceGroups/r/providers/Microsoft.ContainerService/managedClusters/aks-dataplane"),
		}},
	}
	wrongServiceLabels := gcCandidateLabels(cloudv1.ComponentBackup)
	wrongServiceLabels[consts.TagKeyServiceType] = consts.ServiceTypePremium
	wrongService := &manualv1alpha1.NodeGroup{
		ObjectMeta: metav1.ObjectMeta{Name: "wrong-service", Labels: wrongServiceLabels},
	}
	wrongComponent := &gkev1beta1.NodePool{
		ObjectMeta: metav1.ObjectMeta{Name: "wrong-component", Labels: gcCandidateLabels(cloudv1.ComponentTiKV)},
	}
	missingComponentLabels := gcCandidateLabels(cloudv1.ComponentBackup)
	delete(missingComponentLabels, namer.ComponentSelectorKey)
	missingComponent := &azurev1beta1.KubernetesClusterNodePool{
		ObjectMeta: metav1.ObjectMeta{Name: "missing-component", Labels: missingComponentLabels},
	}
	_, local := testGC(t, config.NodePoolGCConfig{}, aws, launchTemplate, gcp, azure, wrongService, wrongComponent, missingComponent)

	tests := []struct {
		name           string
		providerType   string
		providerName   string
		wantProvider   string
		wantRegion     string
		wantKubernetes string
		wantCompanions int
	}{
		{name: "AWS", providerType: consts.ProviderTypeEKS, providerName: "AWS", wantProvider: consts.ProviderTypeEKS, wantRegion: "us-west-2", wantKubernetes: "eks-dataplane", wantCompanions: 1},
		{name: "GCP", providerType: consts.ProviderTypeGKE, providerName: "GCP", wantProvider: consts.ProviderTypeGKE, wantRegion: "us-central1", wantKubernetes: "c"},
		{name: "Azure", providerType: consts.ProviderTypeAKS, providerName: "Azure", wantProvider: consts.ProviderTypeAKS, wantKubernetes: "aks-dataplane"},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			provider := mustCloudProvider(t, tt.providerType, local)
			require.Equal(t, tt.providerName, provider.Name())

			pools, err := provider.ListNodePools(context.Background())
			require.NoError(t, err)
			require.Len(t, pools, 1)
			require.Equal(t, tt.wantProvider, pools[0].provider)
			require.Equal(t, tt.wantRegion, pools[0].region)
			require.Equal(t, tt.wantKubernetes, pools[0].kubernetesName)
			require.Len(t, pools[0].companions, tt.wantCompanions)
		})
	}
}

func TestNewCloudProvider(t *testing.T) {
	tests := []struct {
		name         string
		providerType string
		wantName     string
		wantErr      string
	}{
		{name: "AWS", providerType: consts.ProviderTypeEKS, wantName: "AWS"},
		{name: "GCP", providerType: consts.ProviderTypeGKE, wantName: "GCP"},
		{name: "Azure", providerType: consts.ProviderTypeAKS, wantName: "Azure"},
		{name: "unsupported", providerType: "unsupported", wantErr: "unsupported BR nodepool GC provider"},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			provider, err := newCloudProvider(tt.providerType, nil)
			if tt.wantErr != "" {
				require.Nil(t, provider)
				require.ErrorContains(t, err, tt.wantErr)
				return
			}
			require.NoError(t, err)
			require.Equal(t, tt.wantName, provider.Name())
		})
	}
}

func TestNewAppliesDefaultConfig(t *testing.T) {
	local := fake.NewClientBuilder().WithScheme(gcTestScheme(t)).Build()

	gc, err := New(local, logr.Discard(), config.NodePoolGCConfig{}, consts.ProviderTypeEKS, nil)

	require.NoError(t, err)
	require.Equal(t, defaultScanInterval, gc.config.ScanInterval.Duration)
	require.Equal(t, defaultExpireAfter, gc.config.ExpireAfter.Duration)
}

func TestRunOnce(t *testing.T) {
	now := time.Date(2026, 7, 10, 8, 0, 0, 0, time.UTC)
	tests := []struct {
		name string
		run  func(t *testing.T, now time.Time)
	}{
		{
			name: "manual approval deletes pool when automatic approval is disabled",
			run: func(t *testing.T, now time.Time) {
				nodeGroup := approvedNodeGroup("manual-approval", now)
				pool := newNodePool(nodeGroup.DeepCopy(), consts.ProviderTypeEKS, "us-west-2", "")
				gc, local := testGC(t, config.NodePoolGCConfig{}, nodeGroup)
				gc.now = func() time.Time { return now }
				gc.RunOnce(context.Background())
				require.True(t, apierrors.IsNotFound(local.Get(context.Background(), client.ObjectKeyFromObject(nodeGroup), &manualv1alpha1.NodeGroup{})))
				require.Equal(t, statusValueMap[StatusDeleted], metricValue(pool, ""))
			},
		},
		{
			name: "dry-run keeps approved pool without patching metadata",
			run: func(t *testing.T, now time.Time) {
				nodeGroup := approvedNodeGroup("dry-run", now)
				originalAnnotations := nodeGroup.DeepCopy().Annotations
				pool := newNodePool(nodeGroup.DeepCopy(), consts.ProviderTypeEKS, "us-west-2", "")
				gc, local := testGC(t, config.NodePoolGCConfig{DryRun: true}, nodeGroup)
				gc.now = func() time.Time { return now }
				gc.dataplaneResolver = func(context.Context, nodePool) (client.Client, error) {
					t.Fatal("dry-run must return before checking dataplane nodes")
					return nil, nil
				}

				gc.RunOnce(context.Background())

				got := &manualv1alpha1.NodeGroup{}
				require.NoError(t, local.Get(context.Background(), client.ObjectKeyFromObject(nodeGroup), got))
				require.Equal(t, originalAnnotations, got.Annotations)
				require.Equal(t, statusValueMap[StatusDeletionApproved], metricValue(pool, ""))
			},
		},
		{
			name: "delete removes pool and companion",
			run: func(t *testing.T, now time.Time) {
				nodeGroup := approvedNodeGroup("delete", now)
				launchTemplate := &ec2v1alpha1.LaunchTemplate{ObjectMeta: metav1.ObjectMeta{Name: nodeGroup.Name, Labels: gcCandidateLabels(cloudv1.ComponentBackup)}}
				gc, local := testGC(t, config.NodePoolGCConfig{AutoApproveDeletion: true}, nodeGroup, launchTemplate)
				gc.now = func() time.Time { return now }
				gc.RunOnce(context.Background())
				require.True(t, apierrors.IsNotFound(local.Get(context.Background(), client.ObjectKeyFromObject(nodeGroup), &manualv1alpha1.NodeGroup{})))
				require.True(t, apierrors.IsNotFound(local.Get(context.Background(), client.ObjectKeyFromObject(launchTemplate), &ec2v1alpha1.LaunchTemplate{})))
			},
		},
		{
			name: "matching nodes fail deletion",
			run: func(t *testing.T, now time.Time) {
				nodeGroup := approvedNodeGroup("has-nodes", now)
				pool := newNodePool(nodeGroup.DeepCopy(), consts.ProviderTypeEKS, "us-west-2", "")
				gc, local := testGC(t, config.NodePoolGCConfig{AutoApproveDeletion: true}, nodeGroup)
				gc.now = func() time.Time { return now }
				gc.dataplaneResolver = matchingDataplaneResolver(t, (&awsProvider{}).NodeSelector(pool))
				gc.RunOnce(context.Background())
				require.NoError(t, local.Get(context.Background(), client.ObjectKeyFromObject(nodeGroup), &manualv1alpha1.NodeGroup{}))
				require.Equal(t, statusValueMap[StatusFailed], metricValue(pool, failureReasonHavingNodes))
			},
		},
		{
			name: "resolver failure keeps pool",
			run: func(t *testing.T, now time.Time) {
				nodeGroup := approvedNodeGroup("resolver-failure", now)
				pool := newNodePool(nodeGroup.DeepCopy(), consts.ProviderTypeEKS, "us-west-2", "")
				gc, local := testGC(t, config.NodePoolGCConfig{AutoApproveDeletion: true}, nodeGroup)
				gc.now = func() time.Time { return now }
				gc.dataplaneResolver = func(context.Context, nodePool) (client.Client, error) {
					return nil, errors.New("dataplane unavailable")
				}
				gc.RunOnce(context.Background())
				require.NoError(t, local.Get(context.Background(), client.ObjectKeyFromObject(nodeGroup), &manualv1alpha1.NodeGroup{}))
				require.Equal(t, statusValueMap[StatusFailed], metricValue(pool, failureReasonUnknown))
			},
		},
		{
			name: "delete failure records failToDelete",
			run: func(t *testing.T, now time.Time) {
				nodeGroup := approvedNodeGroup("delete-failure", now)
				pool := newNodePool(nodeGroup.DeepCopy(), consts.ProviderTypeEKS, "us-west-2", "")
				gc, local := testGC(t, config.NodePoolGCConfig{AutoApproveDeletion: true}, nodeGroup)
				gc.now = func() time.Time { return now }
				gc.local = deleteErrorClient{Client: local}

				gc.RunOnce(context.Background())

				require.NoError(t, local.Get(context.Background(), client.ObjectKeyFromObject(nodeGroup), &manualv1alpha1.NodeGroup{}))
				require.Equal(t, statusValueMap[StatusFailed], metricValue(pool, failureReasonFailToDelete))
			},
		},
		{
			name: "malformed pool does not block later pool",
			run: func(t *testing.T, now time.Time) {
				bad := approvedNodeGroup("bad", now)
				bad.Annotations[consts.AnnoKeyNodePoolLastUsedTime] = "bad-time"
				good := approvedNodeGroup("good", now)
				gc, local := testGC(t, config.NodePoolGCConfig{AutoApproveDeletion: true}, bad, good)
				gc.now = func() time.Time { return now }
				gc.RunOnce(context.Background())
				require.NoError(t, local.Get(context.Background(), client.ObjectKeyFromObject(bad), &manualv1alpha1.NodeGroup{}))
				require.True(t, apierrors.IsNotFound(local.Get(context.Background(), client.ObjectKeyFromObject(good), &manualv1alpha1.NodeGroup{})))
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tt.run(t, now)
		})
	}
}

func TestRunOnceAPPBoxMissingOrphanCleanup(t *testing.T) {
	now := time.Date(2026, 7, 10, 8, 0, 0, 0, time.UTC)
	appBoxNotFoundErr := &appBoxNotFoundError{provider: consts.ProviderTypeEKS, kubernetesName: "missing-appbox"}
	tests := []struct {
		name                string
		primaryFinalizers   []string
		companionFinalizers []string
		resolverErr         error
		wantStatus          Status
		wantReason          failureReason
		// End state of each object after the GC run: "deleted" (gone),
		// "terminating" (Delete accepted, finalizer owner releases it), or
		// "retained" (untouched).
		wantPrimary   string
		wantCompanion string
	}{
		{
			name:          "finalizer-free orphan is deleted",
			resolverErr:   appBoxNotFoundErr,
			wantStatus:    StatusDeleted,
			wantPrimary:   "deleted",
			wantCompanion: "deleted",
		},
		{
			name:              "primary finalizer orphan is deleted via terminating",
			primaryFinalizers: []string{"crossplane.io/finalizer"},
			resolverErr:       appBoxNotFoundErr,
			wantStatus:        StatusDeleted,
			wantPrimary:       "terminating",
			wantCompanion:     "deleted",
		},
		{
			name:                "companion finalizer orphan is deleted via terminating",
			companionFinalizers: []string{"crossplane.io/finalizer"},
			resolverErr:         appBoxNotFoundErr,
			wantStatus:          StatusDeleted,
			wantPrimary:         "deleted",
			wantCompanion:       "terminating",
		},
		{
			name:          "generic resolver failure retains orphan records",
			resolverErr:   errors.New("dataplane unavailable"),
			wantStatus:    StatusFailed,
			wantReason:    failureReasonUnknown,
			wantPrimary:   "retained",
			wantCompanion: "retained",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			nodeGroup := approvedNodeGroup(tt.name, now)
			nodeGroup.Spec.ForProvider.ClusterName = "missing-appbox"
			nodeGroup.Finalizers = tt.primaryFinalizers
			launchTemplate := &ec2v1alpha1.LaunchTemplate{
				ObjectMeta: metav1.ObjectMeta{
					Name:       nodeGroup.Name,
					Labels:     gcCandidateLabels(cloudv1.ComponentBackup),
					Finalizers: tt.companionFinalizers,
				},
			}
			pool := newNodePool(nodeGroup.DeepCopy(), consts.ProviderTypeEKS, "us-west-2", nodeGroup.Spec.ForProvider.ClusterName)
			gc, local := testGC(t, config.NodePoolGCConfig{AutoApproveDeletion: true}, nodeGroup, launchTemplate)
			gc.now = func() time.Time { return now }
			gc.dataplaneResolver = func(context.Context, nodePool) (client.Client, error) {
				return nil, tt.resolverErr
			}

			gc.RunOnce(context.Background())

			var gotNodeGroup manualv1alpha1.NodeGroup
			nodeGroupErr := local.Get(context.Background(), client.ObjectKeyFromObject(nodeGroup), &gotNodeGroup)
			var gotLaunchTemplate ec2v1alpha1.LaunchTemplate
			launchTemplateErr := local.Get(context.Background(), client.ObjectKeyFromObject(launchTemplate), &gotLaunchTemplate)
			assertEndState(t, nodeGroupErr, gotNodeGroup.GetDeletionTimestamp(), tt.wantPrimary)
			assertEndState(t, launchTemplateErr, gotLaunchTemplate.GetDeletionTimestamp(), tt.wantCompanion)
			require.Equal(t, statusValueMap[tt.wantStatus], metricValue(pool, tt.wantReason))
		})
	}
}

// assertEndState verifies the observed state of an object after a GC run.
//
//	"deleted":     the object is gone (NotFound).
//	"terminating": the object exists with a deletionTimestamp; final removal is
//	               left to the finalizer owner (e.g. the Crossplane provider
//	               observing the external resource is already gone).
//	"retained":    the object exists without a deletionTimestamp.
func assertEndState(t *testing.T, getErr error, deletionTimestamp *metav1.Time, want string) {
	t.Helper()
	switch want {
	case "deleted":
		require.True(t, apierrors.IsNotFound(getErr))
	case "terminating":
		require.NoError(t, getErr)
		require.NotNil(t, deletionTimestamp, "expected object to be terminating")
	case "retained":
		require.NoError(t, getErr)
		require.Nil(t, deletionTimestamp, "expected object to be untouched")
	default:
		t.Fatalf("unknown expected end state %q", want)
	}
}

func TestStartExitsWhenContextIsCanceled(t *testing.T) {
	gc, _ := testGC(t, config.NodePoolGCConfig{ScanInterval: metav1.Duration{Duration: time.Millisecond}})
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	require.NoError(t, gc.Start(ctx))
}

func TestRemoteClientForPoolFiltersAPPBoxesByTags(t *testing.T) {
	poolObj := &manualv1alpha1.NodeGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name: "np-a",
			Labels: map[string]string{
				consts.TagKeyTenant:  "tenant-a",
				consts.TagKeyProject: "project-a",
			},
		},
	}
	pool := newNodePool(poolObj, consts.ProviderTypeEKS, "us-west-2", "dp-a")

	wrongAppBox := &regionalv1alpha1.APPBox{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "appbox-wrong",
			Namespace: "ns-a",
			Labels: map[string]string{
				regionalv1alpha1.TagTenantKey:  "tenant-b",
				regionalv1alpha1.TagProjectKey: "project-b",
				cloudv1.LabelServiceType:       consts.ServiceTypeDedicated,
			},
		},
		Spec: regionalv1alpha1.APPBoxSpec{Provider: consts.ProviderTypeEKS},
		Status: regionalv1alpha1.APPBoxStatus{
			KubernetesClusterName: "dp-a",
		},
	}

	scheme := gcTestScheme(t)
	local := fake.NewClientBuilder().WithScheme(scheme).WithObjects(wrongAppBox).Build()
	gc := &GC{
		local:     local,
		rsmanager: controllers.NewManager(local, scheme),
	}

	_, err := gc.remoteClientForPool(context.Background(), pool)
	require.ErrorContains(t, err, "no APPBox found for provider=eks kubernetes=dp-a")
	require.ErrorIs(t, err, errAPPBoxNotFound)
	var notFoundErr *appBoxNotFoundError
	require.ErrorAs(t, err, &notFoundErr)
}

func TestDataplaneNotReadySkipsCurrentScan(t *testing.T) {
	now := time.Date(2026, 7, 10, 8, 0, 0, 0, time.UTC)
	tests := []struct {
		name        string
		nodeName    string
		resolverErr error
	}{
		{
			name:        "APPBox is not registered",
			nodeName:    "appbox-unregistered",
			resolverErr: &controllers.APPBoxUnregisteredError{Name: client.ObjectKey{Name: "appbox"}},
		},
		{
			name:     "dataplane cache is not started",
			nodeName: "cache-not-started",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			nodeGroup := approvedNodeGroup(tt.nodeName, now)
			pool := newNodePool(nodeGroup.DeepCopy(), consts.ProviderTypeEKS, "us-west-2", "dp")
			g, local := testGC(t, config.NodePoolGCConfig{AutoApproveDeletion: true}, nodeGroup)
			g.now = func() time.Time { return now }
			remote := fake.NewClientBuilder().WithScheme(gcTestScheme(t)).Build()
			calls := 0
			g.dataplaneResolver = func(context.Context, nodePool) (client.Client, error) {
				calls++
				if tt.resolverErr != nil {
					return nil, tt.resolverErr
				}
				return cacheErrorClient{Client: remote}, nil
			}
			g.recordStatus(pool, StatusDeletionApproved, "")

			g.RunOnce(context.Background())

			require.Equal(t, 1, calls)
			require.NoError(t, local.Get(context.Background(), client.ObjectKeyFromObject(nodeGroup), &manualv1alpha1.NodeGroup{}))
			require.Equal(t, statusValueMap[StatusDeletionApproved], metricValue(pool, ""), "skipped scans must not overwrite the previous status")
		})
	}
}

type cacheErrorClient struct {
	client.Client
}

func (c cacheErrorClient) List(context.Context, client.ObjectList, ...client.ListOption) error {
	return &cache.ErrCacheNotStarted{}
}

type deleteErrorClient struct {
	client.Client
}

func (c deleteErrorClient) Delete(context.Context, client.Object, ...client.DeleteOption) error {
	return errors.New("delete failed")
}

func TestRecordStatusDeletesPreviousSeries(t *testing.T) {
	now := time.Date(2026, 7, 10, 8, 0, 0, 0, time.UTC)
	nodeGroup := approvedNodeGroup("replace-reason", now)
	nodeGroup.Namespace = "namespace"
	pool := newNodePool(nodeGroup.DeepCopy(), consts.ProviderTypeEKS, "us-west-2", "")

	otherNameNodeGroup := approvedNodeGroup("other-name", now)
	otherNameNodeGroup.Namespace = nodeGroup.Namespace
	otherNamePool := newNodePool(otherNameNodeGroup, consts.ProviderTypeEKS, "us-west-2", "")

	otherNamespaceNodeGroup := approvedNodeGroup(nodeGroup.Name, now)
	otherNamespaceNodeGroup.Namespace = "other-namespace"
	otherNamespacePool := newNodePool(otherNamespaceNodeGroup, consts.ProviderTypeEKS, "us-west-2", "")

	g := &GC{}
	collectors.NodePoolGCStatus.Reset()
	t.Cleanup(collectors.NodePoolGCStatus.Reset)

	legacyReason := failureReason("legacy")
	g.recordStatus(pool, StatusFailed, legacyReason)
	g.recordStatus(otherNamePool, StatusFailed, failureReasonHavingNodes)
	g.recordStatus(otherNamespacePool, StatusFailed, failureReasonFailToDelete)

	g.recordStatus(pool, StatusDeleted, "")

	require.Zero(t, metricValue(pool, legacyReason))
	require.Equal(t, statusValueMap[StatusDeleted], metricValue(pool, ""))
	require.Equal(t, statusValueMap[StatusFailed], metricValue(otherNamePool, failureReasonHavingNodes))
	require.Equal(t, statusValueMap[StatusFailed], metricValue(otherNamespacePool, failureReasonFailToDelete))
}

func metricValue(pool nodePool, reason failureReason) float64 {
	return testutil.ToFloat64(collectors.NodePoolGCStatus.WithLabelValues(
		pool.provider,
		pool.region,
		pool.component(),
		pool.primary.GetNamespace(),
		pool.primary.GetName(),
		pool.tenantID(),
		pool.projectID(),
		pool.clusterID(),
		string(reason),
	))
}