package cron

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

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplanemgr/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/nodegroup"
	repomnocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/emptypb"
)

//nolint:unparam // nodeSpec is always "8C16G" in current tests but kept for future spec coverage
func newTestGroup(nodeSpec, primaryInstance string, st apiv1.NodeInstanceGroup_Status, fallbacks ...string) *entity.NodeInstanceGroup {
	cpu, mem, err := nodegroup.ParseSpec(nodeSpec)
	if err != nil {
		// Fallback for tests that pass non-parseable specs.
		cpu, mem = 8, 16
	}
	fbs := make([]*apiv1.NodeInstanceGroup_FallbackInstance, 0, len(fallbacks))
	for _, fb := range fallbacks {
		fbs = append(fbs, &apiv1.NodeInstanceGroup_FallbackInstance{
			InstanceType: fb,
			NodeSpec:     &apiv1.NodeSpec{Cpu: cpu, MemoryGi: mem},
		})
	}
	return &entity.NodeInstanceGroup{
		NodeSpec:            &apiv1.NodeSpec{Cpu: cpu, MemoryGi: mem},
		PrimaryInstanceType: primaryInstance,
		Status:              st,
		FallbackInstances:   fbs,
	}
}

func noopUnlock() error { return nil }

func TestRetiringProcessorJob_Name(t *testing.T) {
	job := NewRetiringProcessorJob(nil, nil, nil, nil, "0 */5 * * * *")
	assert.Equal(t, "RetiringProcessorJob", job.Name())
}

func TestRetiringProcessorJob_CronExpression(t *testing.T) {
	job := NewRetiringProcessorJob(nil, nil, nil, nil, "0 */5 * * * *")
	assert.Equal(t, "0 */5 * * * *", job.CronExpression())
}

func TestRetiringProcessorJob_Execute_LockFails(t *testing.T) {
	lockRepo := repomnocks.NewLockRepo(t)
	lockRepo.On("Lock", mock.Anything, &entity.Lock{Key: retiringProcessorLockKey, TimeoutSecs: 0}).
		Return((func() error)(nil), errors.New("lock failed"))

	job := NewRetiringProcessorJob(nil, nil, nil, lockRepo, "0 */5 * * * *")

	err := job.Execute(context.Background())
	require.NoError(t, err) // should skip, not fail
}

func TestRetiringProcessorJob_Execute_NoRetiringGroups(t *testing.T) {
	lockRepo := repomnocks.NewLockRepo(t)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(noopUnlock, nil)

	nodeGroupRepo := repomnocks.NewNodeInstanceGroupRepo(t)
	nodeGroupRepo.On("ListByStatus", mock.Anything, "retiring").Return([]*entity.NodeInstanceGroup{}, nil)

	job := NewRetiringProcessorJob(nodeGroupRepo, nil, nil, lockRepo, "0 */5 * * * *")

	err := job.Execute(context.Background())
	require.NoError(t, err)
}

func TestRetiringProcessorJob_Execute_ActiveReplicasBlockRetirement(t *testing.T) {
	lockRepo := repomnocks.NewLockRepo(t)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(noopUnlock, nil)

	group := newTestGroup("8C16G", "m5.2xlarge", apiv1.NodeInstanceGroup_RETIRING)

	nodeGroupRepo := repomnocks.NewNodeInstanceGroupRepo(t)
	nodeGroupRepo.On("ListByStatus", mock.Anything, "retiring").Return([]*entity.NodeInstanceGroup{group}, nil)

	dpmClient := mocks.NewClient(t)
	dpmClient.On("ListDataPlanes", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListDataPlanesResponse{
		DataPlanes: []*dplanemgrv1.DataPlane{{Id: "dp-1"}},
	}, nil)
	dpmClient.On("ListNodePools", mock.Anything, &dplanemgrv1.ListNodePoolsRequest{DataPlaneId: "dp-1"}).
		Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{
					Name:            "pool-1",
					InstanceType:    "m5.2xlarge",
					CpuCore:         8,
					MemoryGi:        16,
					CurrentReplicas: 2, // still active
				},
			},
		}, nil)

	dataplaneNodeGroupRepo := repomnocks.NewDataplaneNodeInstanceGroupRepo(t)
	dataplaneNodeGroupRepo.On("LoadByDataplaneAndSpec", mock.Anything, "dp-1", "8C16G", "m5.2xlarge").
		Return((*entity.DataplaneNodeInstanceGroup)(nil), nil)

	job := NewRetiringProcessorJob(nodeGroupRepo, dataplaneNodeGroupRepo, dpmClient, lockRepo, "0 */5 * * * *")

	err := job.Execute(context.Background())
	require.NoError(t, err)
	// Group should NOT have been saved as RETIRED.
	nodeGroupRepo.AssertNotCalled(t, "Save", mock.Anything, mock.Anything)
}

func TestRetiringProcessorJob_Execute_DeleteNotFoundIsIdempotent(t *testing.T) {
	lockRepo := repomnocks.NewLockRepo(t)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(noopUnlock, nil)

	group := newTestGroup("8C16G", "m5.2xlarge", apiv1.NodeInstanceGroup_RETIRING)

	nodeGroupRepo := repomnocks.NewNodeInstanceGroupRepo(t)
	nodeGroupRepo.On("ListByStatus", mock.Anything, "retiring").Return([]*entity.NodeInstanceGroup{group}, nil)
	nodeGroupRepo.On("Save", mock.Anything, mock.Anything).Return(nil)

	dpmClient := mocks.NewClient(t)
	dpmClient.On("ListDataPlanes", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListDataPlanesResponse{
		DataPlanes: []*dplanemgrv1.DataPlane{{Id: "dp-1"}},
	}, nil)
	dpmClient.On("ListNodePools", mock.Anything, &dplanemgrv1.ListNodePoolsRequest{DataPlaneId: "dp-1"}).
		Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{
					Name:            "pool-1",
					InstanceType:    "m5.2xlarge",
					CpuCore:         8,
					MemoryGi:        16,
					CurrentReplicas: 0,
				},
			},
		}, nil)
	dpmClient.On("DeleteNodePool", mock.Anything, &dplanemgrv1.DeleteNodePoolRequest{
		DataPlaneId: "dp-1",
		NodePoolId:  "pool-1",
	}).Return(nil, status.Errorf(codes.NotFound, "not found"))

	dataplaneNodeGroupRepo := repomnocks.NewDataplaneNodeInstanceGroupRepo(t)
	dataplaneNodeGroupRepo.On("LoadByDataplaneAndSpec", mock.Anything, "dp-1", "8C16G", "m5.2xlarge").
		Return((*entity.DataplaneNodeInstanceGroup)(nil), nil)
	dataplaneNodeGroupRepo.On("Save", mock.Anything, mock.Anything).Return(nil)

	job := NewRetiringProcessorJob(nodeGroupRepo, dataplaneNodeGroupRepo, dpmClient, lockRepo, "0 */5 * * * *")
	// Pre-populate in-memory zero-since timestamp (>10 min ago) to pass confirmation.
	job.replicasZeroSince["dp-1/8C16G/m5.2xlarge"] = time.Now().Add(-15 * time.Minute)

	err := job.Execute(context.Background())
	require.NoError(t, err)
	// NotFound should be treated as success, group should be RETIRED.
	nodeGroupRepo.AssertCalled(t, "Save", mock.Anything, mock.Anything)
}

func TestRetiringProcessorJob_Execute_AllDataplanesRetired(t *testing.T) {
	lockRepo := repomnocks.NewLockRepo(t)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(noopUnlock, nil)

	group := newTestGroup("8C16G", "m5.2xlarge", apiv1.NodeInstanceGroup_RETIRING)

	nodeGroupRepo := repomnocks.NewNodeInstanceGroupRepo(t)
	nodeGroupRepo.On("ListByStatus", mock.Anything, "retiring").Return([]*entity.NodeInstanceGroup{group}, nil)
	nodeGroupRepo.On("Save", mock.Anything, mock.MatchedBy(func(g *entity.NodeInstanceGroup) bool {
		return g.GetStatus() == apiv1.NodeInstanceGroup_RETIRED
	})).Return(nil)

	dpmClient := mocks.NewClient(t)
	dpmClient.On("ListDataPlanes", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListDataPlanesResponse{
		DataPlanes: []*dplanemgrv1.DataPlane{{Id: "dp-1"}, {Id: "dp-2"}},
	}, nil)
	// dp-1: no matching pools
	dpmClient.On("ListNodePools", mock.Anything, &dplanemgrv1.ListNodePoolsRequest{DataPlaneId: "dp-1"}).
		Return(&dplanemgrv1.ListNodePoolsResponse{NodePools: []*dplanemgrv1.NodePool{}}, nil)
	// dp-2: no matching pools
	dpmClient.On("ListNodePools", mock.Anything, &dplanemgrv1.ListNodePoolsRequest{DataPlaneId: "dp-2"}).
		Return(&dplanemgrv1.ListNodePoolsResponse{NodePools: []*dplanemgrv1.NodePool{}}, nil)

	dataplaneNodeGroupRepo := repomnocks.NewDataplaneNodeInstanceGroupRepo(t)
	dataplaneNodeGroupRepo.On("LoadByDataplaneAndSpec", mock.Anything, "dp-1", "8C16G", "m5.2xlarge").
		Return((*entity.DataplaneNodeInstanceGroup)(nil), nil)
	dataplaneNodeGroupRepo.On("LoadByDataplaneAndSpec", mock.Anything, "dp-2", "8C16G", "m5.2xlarge").
		Return((*entity.DataplaneNodeInstanceGroup)(nil), nil)
	dataplaneNodeGroupRepo.On("Save", mock.Anything, mock.Anything).Return(nil)

	job := NewRetiringProcessorJob(nodeGroupRepo, dataplaneNodeGroupRepo, dpmClient, lockRepo, "0 */5 * * * *")
	// Pre-populate in-memory zero-since timestamps (>10 min ago) for both dataplanes.
	job.replicasZeroSince["dp-1/8C16G/m5.2xlarge"] = time.Now().Add(-15 * time.Minute)
	job.replicasZeroSince["dp-2/8C16G/m5.2xlarge"] = time.Now().Add(-15 * time.Minute)

	err := job.Execute(context.Background())
	require.NoError(t, err)
	// Group should be saved as RETIRED.
	nodeGroupRepo.AssertCalled(t, "Save", mock.Anything, mock.MatchedBy(func(g *entity.NodeInstanceGroup) bool {
		return g.GetStatus() == apiv1.NodeInstanceGroup_RETIRED
	}))
}

func TestRetiringProcessorJob_Execute_SkipsAlreadyRetiredDataplanes(t *testing.T) {
	lockRepo := repomnocks.NewLockRepo(t)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(noopUnlock, nil)

	group := newTestGroup("8C16G", "m5.2xlarge", apiv1.NodeInstanceGroup_RETIRING)

	nodeGroupRepo := repomnocks.NewNodeInstanceGroupRepo(t)
	nodeGroupRepo.On("ListByStatus", mock.Anything, "retiring").Return([]*entity.NodeInstanceGroup{group}, nil)
	nodeGroupRepo.On("Save", mock.Anything, mock.Anything).Return(nil)

	dpmClient := mocks.NewClient(t)
	dpmClient.On("ListDataPlanes", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListDataPlanesResponse{
		DataPlanes: []*dplanemgrv1.DataPlane{{Id: "dp-1"}},
	}, nil)

	// dp-1 already RETIRED — should skip ListNodePools entirely.
	dataplaneNodeGroupRepo := repomnocks.NewDataplaneNodeInstanceGroupRepo(t)
	dataplaneNodeGroupRepo.On("LoadByDataplaneAndSpec", mock.Anything, "dp-1", "8C16G", "m5.2xlarge").
		Return(&entity.DataplaneNodeInstanceGroup{
			DataplaneName:       "dp-1",
			NodeSpec:            &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16},
			PrimaryInstanceType: "m5.2xlarge",
			Status:              apiv1.DataplaneNodeInstanceGroup_RETIRED,
		}, nil)

	job := NewRetiringProcessorJob(nodeGroupRepo, dataplaneNodeGroupRepo, dpmClient, lockRepo, "0 */5 * * * *")

	err := job.Execute(context.Background())
	require.NoError(t, err)
	dpmClient.AssertNotCalled(t, "ListNodePools", mock.Anything, mock.Anything)
	// Group set to RETIRED since all dataplanes are retired.
	nodeGroupRepo.AssertCalled(t, "Save", mock.Anything, mock.MatchedBy(func(g *entity.NodeInstanceGroup) bool {
		return g.GetStatus() == apiv1.NodeInstanceGroup_RETIRED
	}))
}

func TestRetiringProcessorJob_Execute_ManagedPoolMatching(t *testing.T) {
	lockRepo := repomnocks.NewLockRepo(t)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(noopUnlock, nil)

	group := newTestGroup("8C16G", "m5.2xlarge", apiv1.NodeInstanceGroup_RETIRING, "m5a.2xlarge")

	nodeGroupRepo := repomnocks.NewNodeInstanceGroupRepo(t)
	nodeGroupRepo.On("ListByStatus", mock.Anything, "retiring").Return([]*entity.NodeInstanceGroup{group}, nil)
	nodeGroupRepo.On("Save", mock.Anything, mock.Anything).Return(nil)

	dpmClient := mocks.NewClient(t)
	dpmClient.On("ListDataPlanes", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListDataPlanesResponse{
		DataPlanes: []*dplanemgrv1.DataPlane{{Id: "dp-1"}},
	}, nil)
	dpmClient.On("ListNodePools", mock.Anything, &dplanemgrv1.ListNodePoolsRequest{DataPlaneId: "dp-1"}).
		Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{
					Name:            "managed-pool",
					InstanceType:    "m5a.2xlarge",
					CpuCore:         8,
					MemoryGi:        16,
					CurrentReplicas: 0,
					Labels: map[string]string{
						"tidbcloud.com/managed-by": "instance-type-manager",
						"tidbcloud.com/node-spec":  "8C16G",
					},
				},
				{
					Name:            "other-pool",
					InstanceType:    "c5.2xlarge",
					CpuCore:         8,
					MemoryGi:        16,
					CurrentReplicas: 5,
					Labels: map[string]string{
						"tidbcloud.com/managed-by": "instance-type-manager",
						"tidbcloud.com/node-spec":  "8C16G",
					},
				},
			},
		}, nil)
	// Only managed-pool should match (fallback instance type), other-pool has wrong instance type.
	dpmClient.On("DeleteNodePool", mock.Anything, &dplanemgrv1.DeleteNodePoolRequest{
		DataPlaneId: "dp-1",
		NodePoolId:  "managed-pool",
	}).Return(&emptypb.Empty{}, nil)

	dataplaneNodeGroupRepo := repomnocks.NewDataplaneNodeInstanceGroupRepo(t)
	dataplaneNodeGroupRepo.On("LoadByDataplaneAndSpec", mock.Anything, "dp-1", "8C16G", "m5.2xlarge").
		Return((*entity.DataplaneNodeInstanceGroup)(nil), nil)
	dataplaneNodeGroupRepo.On("Save", mock.Anything, mock.Anything).Return(nil)

	job := NewRetiringProcessorJob(nodeGroupRepo, dataplaneNodeGroupRepo, dpmClient, lockRepo, "0 */5 * * * *")
	// Pre-populate in-memory zero-since timestamp (>10 min ago) to pass confirmation.
	job.replicasZeroSince["dp-1/8C16G/m5.2xlarge"] = time.Now().Add(-15 * time.Minute)

	err := job.Execute(context.Background())
	require.NoError(t, err)
	// Only managed-pool deleted (the other pool has wrong instance type).
	dpmClient.AssertNumberOfCalls(t, "DeleteNodePool", 1)
}

func TestCollectInstanceTypes(t *testing.T) {
	group := newTestGroup("8C16G", "m5.2xlarge", apiv1.NodeInstanceGroup_RETIRING, "m5.2xlarge", "m5a.2xlarge", "m6i.2xlarge")
	types := collectInstanceTypes(group)

	assert.Len(t, types, 3)
	assert.Equal(t, nodegroup.NodePoolRolePrimary, types["m5.2xlarge"])
	assert.Equal(t, nodegroup.NodePoolRoleFallback, types["m5a.2xlarge"])
	assert.Equal(t, nodegroup.NodePoolRoleFallback, types["m6i.2xlarge"])
}

func TestMatchesNodeGroup_ManagedPool(t *testing.T) {
	instanceTypes := map[string]nodegroup.NodePoolRole{"m5.2xlarge": nodegroup.NodePoolRolePrimary}

	// Managed pool with matching spec and instance type.
	np := &dplanemgrv1.NodePool{
		Name:         "pool-1",
		InstanceType: "m5.2xlarge",
		CpuCore:      8,
		MemoryGi:     16,
		Labels: map[string]string{
			"tidbcloud.com/managed-by": "instance-type-manager",
			"tidbcloud.com/node-spec":  "8C16G",
		},
	}
	assert.True(t, matchesNodeGroup(np, "8C16G", 8, 16, instanceTypes))

	// Managed pool with wrong node-spec label.
	np2 := &dplanemgrv1.NodePool{
		Name:         "pool-2",
		InstanceType: "m5.2xlarge",
		CpuCore:      8,
		MemoryGi:     16,
		Labels: map[string]string{
			"tidbcloud.com/managed-by": "instance-type-manager",
			"tidbcloud.com/node-spec":  "16C32G",
		},
	}
	assert.False(t, matchesNodeGroup(np2, "8C16G", 8, 16, instanceTypes))

	// Managed pool with node-spec label in different casing should still match.
	np3 := &dplanemgrv1.NodePool{
		Name:         "pool-3",
		InstanceType: "m5.2xlarge",
		CpuCore:      8,
		MemoryGi:     16,
		Labels: map[string]string{
			"tidbcloud.com/managed-by": "instance-type-manager",
			"tidbcloud.com/node-spec":  "8c16g",
		},
	}
	assert.True(t, matchesNodeGroup(np3, "8C16G", 8, 16, instanceTypes))

	fallbackPoolWithSameInstanceType := &dplanemgrv1.NodePool{
		Name:         "fallback-other-primary-m5-2xlarge",
		InstanceType: "m5.2xlarge",
		CpuCore:      8,
		MemoryGi:     16,
		Labels: map[string]string{
			"tidbcloud.com/managed-by": "instance-type-manager",
			"tidbcloud.com/node-spec":  "8C16G",
			nodegroup.LabelPriority:    "100",
		},
	}
	assert.False(t, matchesNodeGroup(fallbackPoolWithSameInstanceType, "8C16G", 8, 16, instanceTypes))
}

func TestMatchesNodeGroup_DefaultPool(t *testing.T) {
	instanceTypes := map[string]nodegroup.NodePoolRole{"m5.2xlarge": nodegroup.NodePoolRolePrimary}

	// Default pool with matching CPU/Memory and instance type.
	np := &dplanemgrv1.NodePool{
		Name:         "pool-1",
		InstanceType: "m5.2xlarge",
		CpuCore:      8,
		MemoryGi:     16,
	}
	assert.True(t, matchesNodeGroup(np, "8C16G", 8, 16, instanceTypes))

	// Default pool with wrong CPU.
	np2 := &dplanemgrv1.NodePool{
		Name:         "pool-2",
		InstanceType: "m5.2xlarge",
		CpuCore:      4,
		MemoryGi:     16,
	}
	assert.False(t, matchesNodeGroup(np2, "8C16G", 8, 16, instanceTypes))

	// Default pool with wrong instance type.
	np3 := &dplanemgrv1.NodePool{
		Name:         "pool-3",
		InstanceType: "c5.2xlarge",
		CpuCore:      8,
		MemoryGi:     16,
	}
	assert.False(t, matchesNodeGroup(np3, "8C16G", 8, 16, instanceTypes))
}

func TestFilterMatchingNodePools_InvalidSpec(t *testing.T) {
	job := &RetiringProcessorJob{}
	_, err := job.filterMatchingNodePools(nil, "invalid", nil)
	require.Error(t, err)
	assert.Contains(t, err.Error(), "failed to parse nodeSpec")
}

func TestRetiringProcessorJob_ZeroConfirmation_FirstObservation_Skips(t *testing.T) {
	lockRepo := repomnocks.NewLockRepo(t)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(noopUnlock, nil)

	group := newTestGroup("8C16G", "m5.2xlarge", apiv1.NodeInstanceGroup_RETIRING)
	nodeGroupRepo := repomnocks.NewNodeInstanceGroupRepo(t)
	nodeGroupRepo.On("ListByStatus", mock.Anything, "retiring").Return([]*entity.NodeInstanceGroup{group}, nil)

	dpmClient := mocks.NewClient(t)
	dpmClient.On("ListDataPlanes", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListDataPlanesResponse{
		DataPlanes: []*dplanemgrv1.DataPlane{{Id: "dp-1"}},
	}, nil)
	dpmClient.On("ListNodePools", mock.Anything, &dplanemgrv1.ListNodePoolsRequest{DataPlaneId: "dp-1"}).
		Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{Name: "pool-1", InstanceType: "m5.2xlarge", CpuCore: 8, MemoryGi: 16, CurrentReplicas: 0},
			},
		}, nil)

	dataplaneNodeGroupRepo := repomnocks.NewDataplaneNodeInstanceGroupRepo(t)
	dataplaneNodeGroupRepo.On("LoadByDataplaneAndSpec", mock.Anything, "dp-1", "8C16G", "m5.2xlarge").
		Return((*entity.DataplaneNodeInstanceGroup)(nil), nil)

	job := NewRetiringProcessorJob(nodeGroupRepo, dataplaneNodeGroupRepo, dpmClient, lockRepo, "0 */10 * * * *")

	err := job.Execute(context.Background())
	require.NoError(t, err)
	// First observation: should record timestamp but NOT proceed with deletion
	assert.Contains(t, job.replicasZeroSince, "dp-1/8C16G/m5.2xlarge", "should record zero-since timestamp")
	nodeGroupRepo.AssertNotCalled(t, "Save", mock.Anything, mock.Anything)
}

func TestRetiringProcessorJob_ZeroConfirmation_NotElapsed_Skips(t *testing.T) {
	lockRepo := repomnocks.NewLockRepo(t)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(noopUnlock, nil)

	group := newTestGroup("8C16G", "m5.2xlarge", apiv1.NodeInstanceGroup_RETIRING)
	nodeGroupRepo := repomnocks.NewNodeInstanceGroupRepo(t)
	nodeGroupRepo.On("ListByStatus", mock.Anything, "retiring").Return([]*entity.NodeInstanceGroup{group}, nil)

	dpmClient := mocks.NewClient(t)
	dpmClient.On("ListDataPlanes", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListDataPlanesResponse{
		DataPlanes: []*dplanemgrv1.DataPlane{{Id: "dp-1"}},
	}, nil)
	dpmClient.On("ListNodePools", mock.Anything, &dplanemgrv1.ListNodePoolsRequest{DataPlaneId: "dp-1"}).
		Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{Name: "pool-1", InstanceType: "m5.2xlarge", CpuCore: 8, MemoryGi: 16, CurrentReplicas: 0},
			},
		}, nil)

	dataplaneNodeGroupRepo := repomnocks.NewDataplaneNodeInstanceGroupRepo(t)
	dataplaneNodeGroupRepo.On("LoadByDataplaneAndSpec", mock.Anything, "dp-1", "8C16G", "m5.2xlarge").
		Return((*entity.DataplaneNodeInstanceGroup)(nil), nil)

	job := NewRetiringProcessorJob(nodeGroupRepo, dataplaneNodeGroupRepo, dpmClient, lockRepo, "0 */10 * * * *")
	// Set zero-since to 5 minutes ago (< 10 minute threshold)
	job.replicasZeroSince["dp-1/8C16G/m5.2xlarge"] = time.Now().Add(-5 * time.Minute)

	err := job.Execute(context.Background())
	require.NoError(t, err)
	// Should NOT proceed with deletion
	nodeGroupRepo.AssertNotCalled(t, "Save", mock.Anything, mock.Anything)
}

func TestRetiringProcessorJob_ZeroConfirmation_NonZeroReplicas_ClearsTimestamp(t *testing.T) {
	lockRepo := repomnocks.NewLockRepo(t)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(noopUnlock, nil)

	group := newTestGroup("8C16G", "m5.2xlarge", apiv1.NodeInstanceGroup_RETIRING)
	nodeGroupRepo := repomnocks.NewNodeInstanceGroupRepo(t)
	nodeGroupRepo.On("ListByStatus", mock.Anything, "retiring").Return([]*entity.NodeInstanceGroup{group}, nil)

	dpmClient := mocks.NewClient(t)
	dpmClient.On("ListDataPlanes", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListDataPlanesResponse{
		DataPlanes: []*dplanemgrv1.DataPlane{{Id: "dp-1"}},
	}, nil)
	dpmClient.On("ListNodePools", mock.Anything, &dplanemgrv1.ListNodePoolsRequest{DataPlaneId: "dp-1"}).
		Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{Name: "pool-1", InstanceType: "m5.2xlarge", CpuCore: 8, MemoryGi: 16, CurrentReplicas: 3},
			},
		}, nil)

	dataplaneNodeGroupRepo := repomnocks.NewDataplaneNodeInstanceGroupRepo(t)
	dataplaneNodeGroupRepo.On("LoadByDataplaneAndSpec", mock.Anything, "dp-1", "8C16G", "m5.2xlarge").
		Return((*entity.DataplaneNodeInstanceGroup)(nil), nil)

	job := NewRetiringProcessorJob(nodeGroupRepo, dataplaneNodeGroupRepo, dpmClient, lockRepo, "0 */10 * * * *")
	// Pre-populate a timestamp that should get cleared when replicas are non-zero
	job.replicasZeroSince["dp-1/8C16G/m5.2xlarge"] = time.Now().Add(-15 * time.Minute)

	err := job.Execute(context.Background())
	require.NoError(t, err)
	assert.Empty(t, job.replicasZeroSince, "timestamp should be cleared when replicas go back to non-zero")
	nodeGroupRepo.AssertNotCalled(t, "Save", mock.Anything, mock.Anything)
}