package cron

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

	"github.com/prometheus/client_golang/prometheus"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	apiv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	repomocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

func TestWorkerNodeMeteringMetrics_CollectionSuccess(t *testing.T) {
	// Reset metrics before test
	workerNodeMeteringRunTotal.Reset()

	unlockFn := func() error { return nil }
	checkpointRepo := newFakeWorkerNodeMeteringCheckpointRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	writer := &fakeMeteringWriter{}
	dpm := &fakeDataplaneManager{clients: make(map[string]client.Client)}

	lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(unlockFn, nil).Once()
	now := lastCompletedMinute()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1").Return(int64(0), nil).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1", now).Return(nil).Once()
	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{
				Id:            "pool-1",
				DataplaneName: "dataplane-1",
			},
			TenantId: "tenant-1",
		},
	}, nil).Once()

	k8sClient := &fakeK8sClient{
		nodes: corev1.NodeList{Items: []corev1.Node{}},
		pods:  corev1.PodList{Items: []corev1.Pod{}},
	}
	dpm.clients["dataplane-1"] = k8sClient

	job := NewWorkerNodeMeteringJob(checkpointRepo, dpm, sharedPoolRepo, lockRepo, writer, "us-east-1", "aws", 30*time.Minute, "@every 1m")

	beforeCount := getCounterValue(t, "worker_node_metering_run_total", map[string]string{"result": "ok"})
	err := job.Execute(context.Background())
	afterCount := getCounterValue(t, "worker_node_metering_run_total", map[string]string{"result": "ok"})

	require.NoError(t, err)
	assert.InDelta(t, beforeCount+1, afterCount, 1e-9, "worker_node_metering_run_total{result=\"ok\"} should increment")
}

func TestWorkerNodeMeteringMetrics_CollectionError(t *testing.T) {
	// Reset metrics before test
	workerNodeMeteringRunTotal.Reset()

	unlockFn := func() error { return nil }
	checkpointRepo := newFakeWorkerNodeMeteringCheckpointRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	writer := &fakeMeteringWriter{}
	dpm := &fakeDataplaneManager{clients: make(map[string]client.Client)}

	lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(unlockFn, nil).Once()
	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{
				Id:            "pool-1",
				DataplaneName: "dataplane-1",
			},
			TenantId: "tenant-1",
		},
	}, nil).Once()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1").Return(int64(0), errors.New("checkpoint load failed")).Once()

	job := NewWorkerNodeMeteringJob(checkpointRepo, dpm, sharedPoolRepo, lockRepo, writer, "us-east-1", "aws", 30*time.Minute, "@every 1m")

	beforeCount := getCounterValue(t, "worker_node_metering_run_total", map[string]string{"result": "error"})
	err := job.Execute(context.Background())
	afterCount := getCounterValue(t, "worker_node_metering_run_total", map[string]string{"result": "error"})

	require.Error(t, err)
	assert.InDelta(t, beforeCount+1, afterCount, 1e-9, "worker_node_metering_run_total{result=\"error\"} should increment on error")
}

func TestWorkerNodeMeteringMetrics_DataplaneAccessFailure(t *testing.T) {
	// Reset metrics before test
	workerNodeMeteringDataplaneAccessTotal.Reset()

	unlockFn := func() error { return nil }
	checkpointRepo := newFakeWorkerNodeMeteringCheckpointRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	writer := &fakeMeteringWriter{}
	dpm := &fakeDataplaneManager{clients: make(map[string]client.Client)}

	lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(unlockFn, nil).Once()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1").Return(int64(0), nil).Once()
	// No SaveLastSuccessfulMinute expected due to dataplane failure
	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{
				Id:            "pool-1",
				DataplaneName: "dataplane-1",
			},
			TenantId: "tenant-1",
		},
	}, nil).Once()

	// GetClient will fail
	job := NewWorkerNodeMeteringJob(checkpointRepo, dpm, sharedPoolRepo, lockRepo, writer, "us-east-1", "aws", 30*time.Minute, "@every 1m")

	beforeCount := getCounterValue(t, "worker_node_metering_dataplane_access_total", map[string]string{"result": "error"})
	err := job.Execute(context.Background())
	afterCount := getCounterValue(t, "worker_node_metering_dataplane_access_total", map[string]string{"result": "error"})

	require.Error(t, err)
	assert.Contains(t, err.Error(), "dataplane-1")
	assert.Greater(t, afterCount, beforeCount, "worker_node_metering_dataplane_access_total{result=\"error\"} should increment on dataplane failure")
}

func TestWorkerNodeMeteringMetrics_MeteringWriteSuccess(t *testing.T) {
	// Reset metrics before test
	workerNodeMeteringWriteTotal.Reset()

	unlockFn := func() error { return nil }
	checkpointRepo := newFakeWorkerNodeMeteringCheckpointRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	writer := &fakeMeteringWriter{}
	dpm := &fakeDataplaneManager{clients: make(map[string]client.Client)}

	lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(unlockFn, nil).Once()
	now := lastCompletedMinute()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1").Return(int64(0), nil).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1", now).Return(nil).Once()
	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{
				Id:            "pool-1",
				DataplaneName: "dataplane-1",
			},
			TenantId: "tenant-1",
		},
	}, nil).Once()

	k8sClient := &fakeK8sClient{
		nodes: corev1.NodeList{
			Items: []corev1.Node{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:   "worker-1",
						Labels: map[string]string{"node-spec": "8C16G"},
					},
					Spec: corev1.NodeSpec{
						Taints: []corev1.Taint{{Key: "dedicated", Value: "worker"}},
					},
				},
			},
		},
		pods: corev1.PodList{
			Items: []corev1.Pod{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:   "pod-1",
						Labels: map[string]string{"tags.tidbcloud.com/tenant": "tenant-1", "tags.tidbcloud.com/shared-pool": "pool-1"},
					},
					Spec: corev1.PodSpec{NodeName: "worker-1"},
				},
			},
		},
	}
	dpm.clients["dataplane-1"] = k8sClient

	job := NewWorkerNodeMeteringJob(checkpointRepo, dpm, sharedPoolRepo, lockRepo, writer, "us-east-1", "aws", 30*time.Minute, "@every 1m")

	beforeCount := getCounterValue(t, "worker_node_metering_write_total", map[string]string{"result": "ok"})
	err := job.Execute(context.Background())
	afterCount := getCounterValue(t, "worker_node_metering_write_total", map[string]string{"result": "ok"})

	require.NoError(t, err)
	assert.Greater(t, afterCount, beforeCount, "worker_node_metering_write_total{result=\"ok\"} should increment on successful write")
}

func TestWorkerNodeMeteringMetrics_MeteringWriteFailure(t *testing.T) {
	// Reset metrics before test
	workerNodeMeteringWriteTotal.Reset()

	unlockFn := func() error { return nil }
	checkpointRepo := newFakeWorkerNodeMeteringCheckpointRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	writer := &fakeMeteringWriter{err: errors.New("write failed")}
	dpm := &fakeDataplaneManager{clients: make(map[string]client.Client)}

	lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(unlockFn, nil).Once()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1").Return(int64(0), nil).Once()
	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{
				Id:            "pool-1",
				DataplaneName: "dataplane-1",
			},
			TenantId: "tenant-1",
		},
	}, nil).Once()

	k8sClient := &fakeK8sClient{
		nodes: corev1.NodeList{
			Items: []corev1.Node{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:   "worker-1",
						Labels: map[string]string{"node-spec": "8C16G"},
					},
					Spec: corev1.NodeSpec{
						Taints: []corev1.Taint{{Key: "dedicated", Value: "worker"}},
					},
				},
			},
		},
		pods: corev1.PodList{
			Items: []corev1.Pod{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:   "pod-1",
						Labels: map[string]string{"tags.tidbcloud.com/tenant": "tenant-1", "tags.tidbcloud.com/shared-pool": "pool-1"},
					},
					Spec: corev1.PodSpec{NodeName: "worker-1"},
				},
			},
		},
	}
	dpm.clients["dataplane-1"] = k8sClient

	job := NewWorkerNodeMeteringJob(checkpointRepo, dpm, sharedPoolRepo, lockRepo, writer, "us-east-1", "aws", 30*time.Minute, "@every 1m")

	beforeCount := getCounterValue(t, "worker_node_metering_write_total", map[string]string{"result": "error"})
	err := job.Execute(context.Background())
	afterCount := getCounterValue(t, "worker_node_metering_write_total", map[string]string{"result": "error"})

	require.Error(t, err)
	assert.Greater(t, afterCount, beforeCount, "worker_node_metering_write_total{result=\"error\"} should increment on write failure")
}

func TestWorkerNodeMeteringMetrics_CompensationWindowProcessing(t *testing.T) {
	unlockFn := func() error { return nil }
	checkpointRepo := newFakeWorkerNodeMeteringCheckpointRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	writer := &fakeMeteringWriter{}
	dpm := &fakeDataplaneManager{clients: make(map[string]client.Client)}

	lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(unlockFn, nil).Once()
	targetMinute := lastCompletedMinute()
	checkpointMinute := targetMinute - 180 // 3 minutes ago
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1").Return(checkpointMinute, nil).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1", targetMinute).Return(nil).Once()
	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{
				Id:            "pool-1",
				DataplaneName: "dataplane-1",
			},
			TenantId: "tenant-1",
		},
	}, nil).Once()

	k8sClient := &fakeK8sClient{
		nodes: corev1.NodeList{Items: []corev1.Node{}},
		pods:  corev1.PodList{Items: []corev1.Pod{}},
	}
	dpm.clients["dataplane-1"] = k8sClient

	job := NewWorkerNodeMeteringJob(checkpointRepo, dpm, sharedPoolRepo, lockRepo, writer, "us-east-1", "aws", 30*time.Minute, "@every 1m")

	beforeCount := getCounterValue(t, "worker_node_metering_compensation_seconds_total", map[string]string{})
	err := job.Execute(context.Background())
	afterCount := getCounterValue(t, "worker_node_metering_compensation_seconds_total", map[string]string{})

	require.NoError(t, err)
	assert.InDelta(t, beforeCount+180, afterCount, 1e-9, "worker_node_metering_compensation_seconds_total should increment by 180 for 3 minutes processed")
}

func TestWorkerNodeMeteringMetrics_MissingIdentityDrop(t *testing.T) {
	// Reset metrics before test
	workerNodeMeteringNodeDropTotal.Reset()

	unlockFn := func() error { return nil }
	checkpointRepo := newFakeWorkerNodeMeteringCheckpointRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	writer := &fakeMeteringWriter{}
	dpm := &fakeDataplaneManager{clients: make(map[string]client.Client)}

	lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(unlockFn, nil).Once()
	now := lastCompletedMinute()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1").Return(int64(0), nil).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1", now).Return(nil).Once()
	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{
				Id:            "pool-1",
				DataplaneName: "dataplane-1",
			},
			TenantId: "tenant-1",
		},
	}, nil).Once()

	k8sClient := &fakeK8sClient{
		nodes: corev1.NodeList{
			Items: []corev1.Node{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:   "worker-1",
						Labels: map[string]string{"node-spec": "8C16G"},
					},
					Spec: corev1.NodeSpec{
						Taints: []corev1.Taint{{Key: "dedicated", Value: "worker"}},
					},
				},
			},
		},
		pods: corev1.PodList{Items: []corev1.Pod{}}, // No pods -> node has no identity
	}
	dpm.clients["dataplane-1"] = k8sClient

	job := NewWorkerNodeMeteringJob(checkpointRepo, dpm, sharedPoolRepo, lockRepo, writer, "us-east-1", "aws", 30*time.Minute, "@every 1m")

	beforeCount := getCounterValue(t, "worker_node_metering_node_drop_total", map[string]string{"reason": "missing_identity"})
	err := job.Execute(context.Background())
	afterCount := getCounterValue(t, "worker_node_metering_node_drop_total", map[string]string{"reason": "missing_identity"})

	require.NoError(t, err)
	assert.Greater(t, afterCount, beforeCount, "worker_node_metering_node_drop_total{reason=\"missing_identity\"} should increment for nodes without identity")
}

func TestWorkerNodeMeteringMetrics_MissingTenantButPoolLabelStillDropsAsMissingIdentity(t *testing.T) {
	workerNodeMeteringNodeDropTotal.Reset()

	unlockFn := func() error { return nil }
	checkpointRepo := newFakeWorkerNodeMeteringCheckpointRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	writer := &fakeMeteringWriter{}
	dpm := &fakeDataplaneManager{clients: make(map[string]client.Client)}

	lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(unlockFn, nil).Once()
	now := lastCompletedMinute()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1").Return(int64(0), nil).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1", now).Return(nil).Once()
	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{
				Id:            "pool-1",
				DataplaneName: "dataplane-1",
			},
			TenantId: "tenant-1",
		},
	}, nil).Once()

	k8sClient := &fakeK8sClient{
		nodes: corev1.NodeList{
			Items: []corev1.Node{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:   "worker-1",
						Labels: map[string]string{"node-spec": "8C16G"},
					},
					Spec: corev1.NodeSpec{
						Taints: []corev1.Taint{{Key: "dedicated", Value: "worker"}},
					},
				},
			},
		},
		pods: corev1.PodList{Items: []corev1.Pod{
			{
				ObjectMeta: metav1.ObjectMeta{
					Name: "pod-missing-tenant",
					Labels: map[string]string{
						"tags.tidbcloud.com/shared-pool": "pool-1",
					},
				},
				Spec:   corev1.PodSpec{NodeName: "worker-1"},
				Status: corev1.PodStatus{Phase: corev1.PodRunning},
			},
		}},
	}
	dpm.clients["dataplane-1"] = k8sClient

	job := NewWorkerNodeMeteringJob(checkpointRepo, dpm, sharedPoolRepo, lockRepo, writer, "us-east-1", "aws", 30*time.Minute, "@every 1m")

	beforeCount := getCounterValue(t, "worker_node_metering_node_drop_total", map[string]string{"reason": "missing_identity"})
	err := job.Execute(context.Background())
	afterCount := getCounterValue(t, "worker_node_metering_node_drop_total", map[string]string{"reason": "missing_identity"})

	require.NoError(t, err)
	assert.Greater(t, afterCount, beforeCount, "worker_node_metering_node_drop_total{reason=\"missing_identity\"} should increment for pods missing tenant but carrying pool identity")
	assert.Empty(t, writer.calls)
}

func TestWorkerNodeMeteringMetrics_InvalidNodeSpecDrop(t *testing.T) {
	// Reset metrics before test
	workerNodeMeteringNodeDropTotal.Reset()

	unlockFn := func() error { return nil }
	checkpointRepo := newFakeWorkerNodeMeteringCheckpointRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	writer := &fakeMeteringWriter{}
	dpm := &fakeDataplaneManager{clients: make(map[string]client.Client)}

	lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(unlockFn, nil).Once()
	now := lastCompletedMinute()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1").Return(int64(0), nil).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1", now).Return(nil).Once()
	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{
				Id:            "pool-1",
				DataplaneName: "dataplane-1",
			},
			TenantId: "tenant-1",
		},
	}, nil).Once()

	k8sClient := &fakeK8sClient{
		nodes: corev1.NodeList{
			Items: []corev1.Node{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:   "worker-1",
						Labels: map[string]string{"node-spec": "invalid-format"},
					},
					Spec: corev1.NodeSpec{
						Taints: []corev1.Taint{{Key: "dedicated", Value: "worker"}},
					},
				},
			},
		},
		pods: corev1.PodList{
			Items: []corev1.Pod{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:   "pod-1",
						Labels: map[string]string{"tags.tidbcloud.com/tenant": "tenant-1", "tags.tidbcloud.com/shared-pool": "pool-1"},
					},
					Spec: corev1.PodSpec{NodeName: "worker-1"},
				},
			},
		},
	}
	dpm.clients["dataplane-1"] = k8sClient

	job := NewWorkerNodeMeteringJob(checkpointRepo, dpm, sharedPoolRepo, lockRepo, writer, "us-east-1", "aws", 30*time.Minute, "@every 1m")

	beforeCount := getCounterValue(t, "worker_node_metering_node_drop_total", map[string]string{"reason": "invalid_node_spec"})
	err := job.Execute(context.Background())
	afterCount := getCounterValue(t, "worker_node_metering_node_drop_total", map[string]string{"reason": "invalid_node_spec"})

	require.NoError(t, err)
	assert.Greater(t, afterCount, beforeCount, "worker_node_metering_node_drop_total{reason=\"invalid_node_spec\"} should increment for invalid node-spec")
}

func TestWorkerNodeMeteringMetrics_CPUMismatchWarning(t *testing.T) {
	unlockFn := func() error { return nil }
	checkpointRepo := newFakeWorkerNodeMeteringCheckpointRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	writer := &fakeMeteringWriter{}
	dpm := &fakeDataplaneManager{clients: make(map[string]client.Client)}

	lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(unlockFn, nil).Once()
	now := lastCompletedMinute()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1").Return(int64(0), nil).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1", now).Return(nil).Once()
	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{
				Id:            "pool-1",
				DataplaneName: "dataplane-1",
			},
			TenantId: "tenant-1",
		},
	}, nil).Once()

	k8sClient := &fakeK8sClient{
		nodes: corev1.NodeList{
			Items: []corev1.Node{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:   "worker-1",
						Labels: map[string]string{"node-spec": "8C16G"},
					},
					Spec: corev1.NodeSpec{
						Taints: []corev1.Taint{{Key: "dedicated", Value: "worker"}},
					},
					Status: corev1.NodeStatus{
						Capacity: corev1.ResourceList{
							corev1.ResourceCPU: *resource.NewQuantity(16, resource.DecimalSI),
						},
					},
				},
			},
		},
		pods: corev1.PodList{
			Items: []corev1.Pod{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:   "pod-1",
						Labels: map[string]string{"tags.tidbcloud.com/tenant": "tenant-1", "tags.tidbcloud.com/shared-pool": "pool-1"},
					},
					Spec: corev1.PodSpec{NodeName: "worker-1"},
				},
			},
		},
	}
	dpm.clients["dataplane-1"] = k8sClient

	job := NewWorkerNodeMeteringJob(checkpointRepo, dpm, sharedPoolRepo, lockRepo, writer, "us-east-1", "aws", 30*time.Minute, "@every 1m")

	beforeCount := getCounterValue(t, "worker_node_metering_cpu_mismatch_total", map[string]string{})
	err := job.Execute(context.Background())
	afterCount := getCounterValue(t, "worker_node_metering_cpu_mismatch_total", map[string]string{})

	require.NoError(t, err)
	assert.Greater(t, afterCount, beforeCount, "worker_node_metering_cpu_mismatch_total should increment when node-spec CPU doesn't match capacity.cpu")
}

func TestWorkerNodeMeteringMetrics_ConflictingOwnershipDrop(t *testing.T) {
	// Reset metrics before test
	workerNodeMeteringNodeDropTotal.Reset()

	unlockFn := func() error { return nil }
	checkpointRepo := newFakeWorkerNodeMeteringCheckpointRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	writer := &fakeMeteringWriter{}
	dpm := &fakeDataplaneManager{clients: make(map[string]client.Client)}

	lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(unlockFn, nil).Once()
	now := lastCompletedMinute()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1").Return(int64(0), nil).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1", now).Return(nil).Once()
	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{
				Id:            "pool-1",
				DataplaneName: "dataplane-1",
			},
			TenantId: "tenant-1",
		},
	}, nil).Once()

	k8sClient := &fakeK8sClient{
		nodes: corev1.NodeList{
			Items: []corev1.Node{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:   "worker-1",
						Labels: map[string]string{"node-spec": "8C16G"},
					},
					Spec: corev1.NodeSpec{
						Taints: []corev1.Taint{{Key: "dedicated", Value: "worker"}},
					},
				},
			},
		},
		pods: corev1.PodList{
			Items: []corev1.Pod{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:   "pod-1",
						Labels: map[string]string{"tags.tidbcloud.com/tenant": "tenant-1", "tags.tidbcloud.com/shared-pool": "pool-1"},
					},
					Spec: corev1.PodSpec{NodeName: "worker-1"},
				},
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:   "pod-2",
						Labels: map[string]string{"tags.tidbcloud.com/tenant": "tenant-2", "tags.tidbcloud.com/shared-pool": "pool-2"},
					},
					Spec: corev1.PodSpec{NodeName: "worker-1"},
				},
			},
		},
	}
	dpm.clients["dataplane-1"] = k8sClient

	job := NewWorkerNodeMeteringJob(checkpointRepo, dpm, sharedPoolRepo, lockRepo, writer, "us-east-1", "aws", 30*time.Minute, "@every 1m")

	beforeCount := getCounterValue(t, "worker_node_metering_node_drop_total", map[string]string{"reason": "conflicting_ownership"})
	err := job.Execute(context.Background())
	afterCount := getCounterValue(t, "worker_node_metering_node_drop_total", map[string]string{"reason": "conflicting_ownership"})

	require.NoError(t, err)
	assert.Greater(t, afterCount, beforeCount, "worker_node_metering_node_drop_total{reason=\"conflicting_ownership\"} should increment for nodes with conflicting ownership")
}

// Helper function to get counter value from Prometheus metrics
func getCounterValue(t *testing.T, metricName string, labels map[string]string) float64 {
	t.Helper()
	metricFamilies, err := prometheus.DefaultGatherer.Gather()
	require.NoError(t, err)

	for _, mf := range metricFamilies {
		if mf.GetName() != metricName {
			continue
		}
		for _, metric := range mf.GetMetric() {
			if metricLabelsMatch(metric.GetLabel(), labels) {
				if metric.GetCounter() != nil {
					return metric.GetCounter().GetValue()
				}
			}
		}
	}
	return 0
}