package cron

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

	meteringsdkcommon "github.com/pingcap/metering_sdk/common"
	"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/meta"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/rest"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

type selectiveFailMeteringWriter struct {
	calls             []*meteringsdkcommon.MeteringData
	errBySharedPoolID map[string]error
}

func (w *selectiveFailMeteringWriter) Write(_ context.Context, data any) error {
	meteringData := data.(*meteringsdkcommon.MeteringData)
	if err := w.errBySharedPoolID[meteringData.SharedPoolID]; err != nil {
		return err
	}
	w.calls = append(w.calls, meteringData)
	return nil
}

type fakeWorkerNodeMeteringCheckpointRepo struct {
	mock.Mock
}

type fakeWorkerNodeMeteringCheckpointRepoExpecter struct {
	mock *mock.Mock
}

type fakeWorkerNodeMeteringCheckpointRepoLoadCall struct {
	*mock.Call
}

type fakeWorkerNodeMeteringCheckpointRepoSaveCall struct {
	*mock.Call
}

func newFakeWorkerNodeMeteringCheckpointRepo(t *testing.T) *fakeWorkerNodeMeteringCheckpointRepo {
	t.Helper()
	repo := &fakeWorkerNodeMeteringCheckpointRepo{}
	t.Cleanup(func() {
		repo.AssertExpectations(t)
	})
	return repo
}

func (r *fakeWorkerNodeMeteringCheckpointRepo) EXPECT() *fakeWorkerNodeMeteringCheckpointRepoExpecter {
	return &fakeWorkerNodeMeteringCheckpointRepoExpecter{mock: &r.Mock}
}

func (e *fakeWorkerNodeMeteringCheckpointRepoExpecter) LoadLastSuccessfulMinute(ctx, region, cloudProvider, dataplane any) *fakeWorkerNodeMeteringCheckpointRepoLoadCall {
	return &fakeWorkerNodeMeteringCheckpointRepoLoadCall{Call: e.mock.On("LoadLastSuccessfulMinute", ctx, region, cloudProvider, dataplane)}
}

func (c *fakeWorkerNodeMeteringCheckpointRepoLoadCall) Return(minute int64, err error) *fakeWorkerNodeMeteringCheckpointRepoLoadCall {
	c.Call.Return(minute, err)
	return c
}

func (e *fakeWorkerNodeMeteringCheckpointRepoExpecter) SaveLastSuccessfulMinute(ctx, region, cloudProvider, dataplane, minuteTs any) *fakeWorkerNodeMeteringCheckpointRepoSaveCall {
	return &fakeWorkerNodeMeteringCheckpointRepoSaveCall{Call: e.mock.On("SaveLastSuccessfulMinute", ctx, region, cloudProvider, dataplane, minuteTs)}
}

func (c *fakeWorkerNodeMeteringCheckpointRepoSaveCall) Return(err error) *fakeWorkerNodeMeteringCheckpointRepoSaveCall {
	c.Call.Return(err)
	return c
}

func (r *fakeWorkerNodeMeteringCheckpointRepo) LoadLastSuccessfulMinute(ctx context.Context, region, cloudProvider, dataplane string) (int64, error) {
	args := r.Called(ctx, region, cloudProvider, dataplane)
	return args.Get(0).(int64), args.Error(1)
}

func (r *fakeWorkerNodeMeteringCheckpointRepo) SaveLastSuccessfulMinute(ctx context.Context, region, cloudProvider, dataplane string, minuteTs int64) error {
	args := r.Called(ctx, region, cloudProvider, dataplane, minuteTs)
	return args.Error(0)
}

type fakeDataplaneManager struct {
	clients map[string]client.Client
}

func (f *fakeDataplaneManager) GetClient(name string) (client.Client, error) {
	c, ok := f.clients[name]
	if !ok {
		return nil, errors.New("client not found")
	}
	return c, nil
}

func (f *fakeDataplaneManager) GetRestConfig(name string) (*rest.Config, error) {
	return nil, errors.New("not implemented")
}

type fakeK8sClient struct {
	nodes corev1.NodeList
	pods  corev1.PodList
	err   error
}

func (f *fakeK8sClient) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error {
	if f.err != nil {
		return f.err
	}
	switch v := list.(type) {
	case *corev1.NodeList:
		*v = f.nodes
	case *corev1.PodList:
		*v = f.pods
	}
	return nil
}

func (f *fakeK8sClient) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
	return errors.New("not implemented")
}

func (f *fakeK8sClient) Create(ctx context.Context, obj client.Object, opts ...client.CreateOption) error {
	return errors.New("not implemented")
}

func (f *fakeK8sClient) Delete(ctx context.Context, obj client.Object, opts ...client.DeleteOption) error {
	return errors.New("not implemented")
}

func (f *fakeK8sClient) Update(ctx context.Context, obj client.Object, opts ...client.UpdateOption) error {
	return errors.New("not implemented")
}

func (f *fakeK8sClient) Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
	return errors.New("not implemented")
}

func (f *fakeK8sClient) DeleteAllOf(ctx context.Context, obj client.Object, opts ...client.DeleteAllOfOption) error {
	return errors.New("not implemented")
}

func (f *fakeK8sClient) Status() client.SubResourceWriter {
	return nil
}

func (f *fakeK8sClient) SubResource(subResource string) client.SubResourceClient {
	return nil
}

func (f *fakeK8sClient) GroupVersionKindFor(obj runtime.Object) (schema.GroupVersionKind, error) {
	return schema.GroupVersionKind{}, errors.New("not implemented")
}

func (f *fakeK8sClient) Scheme() *runtime.Scheme {
	return nil
}

func (f *fakeK8sClient) RESTMapper() meta.RESTMapper {
	return nil
}

func (f *fakeK8sClient) IsObjectNamespaced(obj runtime.Object) (bool, error) {
	return false, errors.New("not implemented")
}

func TestWorkerNodeMeteringJob_Execute(t *testing.T) {
	unlockFn := func() error { return nil }

	tests := []struct {
		name             string
		setupMocks       func(*fakeWorkerNodeMeteringCheckpointRepo, *repomocks.SharedPoolRepo, *repomocks.LockRepo, *fakeDataplaneManager)
		expectedPayloads map[string][]map[string]any
		expectedCalls    int
		expectError      bool
	}{
		{
			name: "single billable worker node with one tenant pod",
			setupMocks: func(checkpointRepo *fakeWorkerNodeMeteringCheckpointRepo, sharedPoolRepo *repomocks.SharedPoolRepo, lockRepo *repomocks.LockRepo, dpm *fakeDataplaneManager) {
				lockRepo.EXPECT().Lock(mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
				})).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()

				// Setup k8s client with one worker node and one billable pod
				k8sClient := &fakeK8sClient{
					nodes: corev1.NodeList{
						Items: []corev1.Node{
							{
								ObjectMeta: metav1.ObjectMeta{
									Name: "worker-node-1",
									Labels: map[string]string{
										"node-spec": "8C16G",
									},
								},
								Spec: corev1.NodeSpec{
									Taints: []corev1.Taint{
										{
											Key:    "dedicated",
											Value:  "worker",
											Effect: corev1.TaintEffectNoSchedule,
										},
									},
								},
							},
						},
					},
					pods: corev1.PodList{
						Items: []corev1.Pod{
							{
								ObjectMeta: metav1.ObjectMeta{
									Name:      "tenant-pod-1",
									Namespace: "default",
									Labels: map[string]string{
										"tags.tidbcloud.com/tenant":      "tenant-1",
										"tags.tidbcloud.com/shared-pool": "pool-1",
									},
								},
								Spec: corev1.PodSpec{
									NodeName: "worker-node-1",
								},
							},
						},
					},
				}
				dpm.clients["dataplane-1"] = k8sClient
			},
			expectedCalls: 1,
			expectedPayloads: map[string][]map[string]any{
				"pool-1": {
					{
						"tenant_id":       "tenant-1",
						"resourcepool_id": "pool-1",
						"region":          "us-east-1",
						"meter_name":      "vcpu_usage",
						"json_version":    "1.0.0",
						"usage": map[string]any{
							"value": float64(8),
							"unit":  "vCPU",
						},
					},
				},
			},
		},
		{
			name: "exclude idle worker nodes without billable pods",
			setupMocks: func(checkpointRepo *fakeWorkerNodeMeteringCheckpointRepo, sharedPoolRepo *repomocks.SharedPoolRepo, lockRepo *repomocks.LockRepo, dpm *fakeDataplaneManager) {
				lockRepo.EXPECT().Lock(mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
				})).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-node-1",
									Labels: map[string]string{
										"node-spec": "8C16G",
									},
								},
								Spec: corev1.NodeSpec{
									Taints: []corev1.Taint{
										{
											Key:    "dedicated",
											Value:  "worker",
											Effect: corev1.TaintEffectNoSchedule,
										},
									},
								},
							},
						},
					},
					pods: corev1.PodList{
						Items: []corev1.Pod{},
					},
				}
				dpm.clients["dataplane-1"] = k8sClient
			},
			expectedCalls:    0,
			expectedPayloads: map[string][]map[string]any{},
		},
		{
			name: "exclude nodes with conflicting resource pool ownership",
			setupMocks: func(checkpointRepo *fakeWorkerNodeMeteringCheckpointRepo, sharedPoolRepo *repomocks.SharedPoolRepo, lockRepo *repomocks.LockRepo, dpm *fakeDataplaneManager) {
				lockRepo.EXPECT().Lock(mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
				})).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-node-1",
									Labels: map[string]string{
										"node-spec": "8C16G",
									},
								},
								Spec: corev1.NodeSpec{
									Taints: []corev1.Taint{
										{
											Key:    "dedicated",
											Value:  "worker",
											Effect: corev1.TaintEffectNoSchedule,
										},
									},
								},
							},
						},
					},
					pods: corev1.PodList{
						Items: []corev1.Pod{
							{
								ObjectMeta: metav1.ObjectMeta{
									Name:      "tenant-pod-1",
									Namespace: "default",
									Labels: map[string]string{
										"tags.tidbcloud.com/tenant":      "tenant-1",
										"tags.tidbcloud.com/shared-pool": "pool-1",
									},
								},
								Spec: corev1.PodSpec{
									NodeName: "worker-node-1",
								},
							},
							{
								ObjectMeta: metav1.ObjectMeta{
									Name:      "tenant-pod-2",
									Namespace: "default",
									Labels: map[string]string{
										"tags.tidbcloud.com/tenant":      "tenant-2",
										"tags.tidbcloud.com/shared-pool": "pool-2",
									},
								},
								Spec: corev1.PodSpec{
									NodeName: "worker-node-1",
								},
							},
						},
					},
				}
				dpm.clients["dataplane-1"] = k8sClient
			},
			expectedCalls:    0,
			expectedPayloads: map[string][]map[string]any{},
		},
		{
			name: "parse vCPU from node spec label formats",
			setupMocks: func(checkpointRepo *fakeWorkerNodeMeteringCheckpointRepo, sharedPoolRepo *repomocks.SharedPoolRepo, lockRepo *repomocks.LockRepo, dpm *fakeDataplaneManager) {
				lockRepo.EXPECT().Lock(mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
				})).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-node-1",
									Labels: map[string]string{
										"node-spec": "16C32G",
									},
								},
								Spec: corev1.NodeSpec{
									Taints: []corev1.Taint{
										{
											Key:    "dedicated",
											Value:  "worker",
											Effect: corev1.TaintEffectNoSchedule,
										},
									},
								},
							},
						},
					},
					pods: corev1.PodList{
						Items: []corev1.Pod{
							{
								ObjectMeta: metav1.ObjectMeta{
									Name:      "tenant-pod-1",
									Namespace: "default",
									Labels: map[string]string{
										"tags.tidbcloud.com/tenant":      "tenant-1",
										"tags.tidbcloud.com/shared-pool": "pool-1",
									},
								},
								Spec: corev1.PodSpec{
									NodeName: "worker-node-1",
								},
							},
						},
					},
				}
				dpm.clients["dataplane-1"] = k8sClient
			},
			expectedCalls: 1,
			expectedPayloads: map[string][]map[string]any{
				"pool-1": {
					{
						"tenant_id":       "tenant-1",
						"resourcepool_id": "pool-1",
						"region":          "us-east-1",
						"meter_name":      "vcpu_usage",
						"json_version":    "1.0.0",
						"usage": map[string]any{
							"value": float64(16),
							"unit":  "vCPU",
						},
					},
				},
			},
		},
		{
			name: "exclude DaemonSet-owned pods from ownership calculation",
			setupMocks: func(checkpointRepo *fakeWorkerNodeMeteringCheckpointRepo, sharedPoolRepo *repomocks.SharedPoolRepo, lockRepo *repomocks.LockRepo, dpm *fakeDataplaneManager) {
				lockRepo.EXPECT().Lock(mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
				})).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-node-1",
									Labels: map[string]string{
										"node-spec": "8C16G",
									},
								},
								Spec: corev1.NodeSpec{
									Taints: []corev1.Taint{
										{
											Key:    "dedicated",
											Value:  "worker",
											Effect: corev1.TaintEffectNoSchedule,
										},
									},
								},
							},
						},
					},
					pods: corev1.PodList{
						Items: []corev1.Pod{
							{
								ObjectMeta: metav1.ObjectMeta{
									Name:      "daemonset-pod",
									Namespace: "default",
									Labels: map[string]string{
										"tags.tidbcloud.com/tenant":      "tenant-2",
										"tags.tidbcloud.com/shared-pool": "pool-2",
									},
									OwnerReferences: []metav1.OwnerReference{
										{
											Kind: "DaemonSet",
											Name: "monitoring-agent",
										},
									},
								},
								Spec: corev1.PodSpec{
									NodeName: "worker-node-1",
								},
							},
							{
								ObjectMeta: metav1.ObjectMeta{
									Name:      "tenant-pod-1",
									Namespace: "default",
									Labels: map[string]string{
										"tags.tidbcloud.com/tenant":      "tenant-1",
										"tags.tidbcloud.com/shared-pool": "pool-1",
									},
								},
								Spec: corev1.PodSpec{
									NodeName: "worker-node-1",
								},
							},
						},
					},
				}
				dpm.clients["dataplane-1"] = k8sClient
			},
			expectedCalls: 1,
			expectedPayloads: map[string][]map[string]any{
				"pool-1": {
					{
						"tenant_id":       "tenant-1",
						"resourcepool_id": "pool-1",
						"region":          "us-east-1",
						"meter_name":      "vcpu_usage",
						"json_version":    "1.0.0",
						"usage": map[string]any{
							"value": float64(8),
							"unit":  "vCPU",
						},
					},
				},
			},
		},
		{
			name: "exclude non-worker nodes from billing",
			setupMocks: func(checkpointRepo *fakeWorkerNodeMeteringCheckpointRepo, sharedPoolRepo *repomocks.SharedPoolRepo, lockRepo *repomocks.LockRepo, dpm *fakeDataplaneManager) {
				lockRepo.EXPECT().Lock(mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
				})).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: "control-plane-node",
									Labels: map[string]string{
										"node-spec": "8C16G",
									},
								},
								Spec: corev1.NodeSpec{
									Taints: []corev1.Taint{
										{
											Key:    "dedicated",
											Value:  "control-plane",
											Effect: corev1.TaintEffectNoSchedule,
										},
									},
								},
							},
						},
					},
					pods: corev1.PodList{
						Items: []corev1.Pod{
							{
								ObjectMeta: metav1.ObjectMeta{
									Name:      "tenant-pod-1",
									Namespace: "default",
									Labels: map[string]string{
										"tags.tidbcloud.com/tenant":      "tenant-1",
										"tags.tidbcloud.com/shared-pool": "pool-1",
									},
								},
								Spec: corev1.PodSpec{
									NodeName: "control-plane-node",
								},
							},
						},
					},
				}
				dpm.clients["dataplane-1"] = k8sClient
			},
			expectedCalls:    0,
			expectedPayloads: map[string][]map[string]any{},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			checkpointRepo := newFakeWorkerNodeMeteringCheckpointRepo(t)
			sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
			lockRepo := repomocks.NewLockRepo(t)
			writer := &fakeMeteringWriter{}
			dpm := &fakeDataplaneManager{clients: make(map[string]client.Client)}

			tt.setupMocks(checkpointRepo, sharedPoolRepo, lockRepo, dpm)

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

			if tt.expectError {
				require.Error(t, err)
			} else {
				require.NoError(t, err)
			}

			assert.Len(t, writer.calls, tt.expectedCalls)
			assertWorkerNodeMeteringPayloads(t, writer.calls, tt.expectedPayloads)
		})
	}
}

func TestWorkerNodeMeteringJob_Execute_CompensationWindow(t *testing.T) {
	unlockFn := func() error { return nil }

	t.Run("backfill missing minutes within compensation window", func(t *testing.T) {
		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.MatchedBy(func(lock *entity.Lock) bool {
			return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
		})).Return(unlockFn, nil).Once()

		targetMinute := lastCompletedMinute()
		checkpointMinute := targetMinute - 180

		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()

		oldCreationTime := metav1.Unix(checkpointMinute-300, 0)

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

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

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

		require.Len(t, writer.calls, 3)

		expectedTimestamps := []int64{
			targetMinute - 120,
			targetMinute - 60,
			targetMinute,
		}

		actualTimestamps := make([]int64, len(writer.calls))
		for i, call := range writer.calls {
			actualTimestamps[i] = call.Timestamp
		}

		for i, expectedTs := range expectedTimestamps {
			assert.Equal(t, expectedTs, actualTimestamps[i])
		}

		for _, call := range writer.calls {
			assert.Equal(t, "worker-node", call.Category)
			assert.Equal(t, "pool-1", call.SharedPoolID)
			require.Len(t, call.Data, 1)
			assert.Equal(t, "tenant-1", call.Data[0]["tenant_id"])
			assert.InDelta(t, float64(8), call.Data[0]["usage"].(map[string]any)["value"], 1e-9)
		}
	})

	t.Run("exclude objects not visible at target minute based on creationTimestamp", func(t *testing.T) {
		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.MatchedBy(func(lock *entity.Lock) bool {
			return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
		})).Return(unlockFn, nil).Once()

		targetMinute := lastCompletedMinute()
		checkpointMinute := targetMinute - 120

		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()

		oldNodeCreation := metav1.Unix(checkpointMinute-300, 0)
		podCreatedAfterPreviousMinute := metav1.Unix(targetMinute, 0)

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

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

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

		require.Len(t, writer.calls, 1)
		assert.Equal(t, targetMinute, writer.calls[0].Timestamp)
	})

	t.Run("backfill bills minutes before deletion starts", func(t *testing.T) {
		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.MatchedBy(func(lock *entity.Lock) bool {
			return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
		})).Return(unlockFn, nil).Once()

		targetMinute := lastCompletedMinute()
		checkpointMinute := targetMinute - 180

		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()

		oldCreationTime := metav1.Unix(checkpointMinute-300, 0)
		deletingTime := metav1.Unix(targetMinute-30, 0)

		k8sClient := &fakeK8sClient{
			nodes: corev1.NodeList{
				Items: []corev1.Node{
					{
						ObjectMeta: metav1.ObjectMeta{
							Name:              "worker-node-1",
							CreationTimestamp: oldCreationTime,
							Labels: map[string]string{
								"node-spec": "8C16G",
							},
						},
						Spec: corev1.NodeSpec{
							Taints: []corev1.Taint{
								{
									Key:    "dedicated",
									Value:  "worker",
									Effect: corev1.TaintEffectNoSchedule,
								},
							},
						},
					},
				},
			},
			pods: corev1.PodList{
				Items: []corev1.Pod{
					{
						ObjectMeta: metav1.ObjectMeta{
							Name:              "tenant-pod-1",
							Namespace:         "default",
							CreationTimestamp: oldCreationTime,
							DeletionTimestamp: &deletingTime,
							Labels: map[string]string{
								"tags.tidbcloud.com/tenant":      "tenant-1",
								"tags.tidbcloud.com/shared-pool": "pool-1",
							},
						},
						Spec: corev1.PodSpec{
							NodeName: "worker-node-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",
		)

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

		require.Len(t, writer.calls, 1)
		assert.Equal(t, targetMinute-120, writer.calls[0].Timestamp)
		assert.Equal(t, "pool-1", writer.calls[0].SharedPoolID)
		require.Len(t, writer.calls[0].Data, 1)
		assert.Equal(t, "tenant-1", writer.calls[0].Data[0]["tenant_id"])
		assert.InDelta(t, float64(8), writer.calls[0].Data[0]["usage"].(map[string]any)["value"], 1e-9)
	})
}

func TestWorkerNodeMeteringJob_Execute_DataplaneFailureDoesNotSaveCheckpoint(t *testing.T) {
	unlockFn := func() error { return nil }

	t.Run("GetClient failure does not save checkpoint", func(t *testing.T) {
		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.MatchedBy(func(lock *entity.Lock) bool {
			return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
		})).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()

		// GetClient will fail - no checkpoint should be saved
		job := NewWorkerNodeMeteringJob(
			checkpointRepo,
			dpm,
			sharedPoolRepo,
			lockRepo,
			writer,
			"us-east-1",
			"aws",
			30*time.Minute,
			"@every 1m",
		)
		err := job.Execute(context.Background())

		require.Error(t, err)
		assert.Contains(t, err.Error(), "dataplane-1")
		assert.Empty(t, writer.calls)
	})

	t.Run("List nodes failure does not save checkpoint", func(t *testing.T) {
		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.MatchedBy(func(lock *entity.Lock) bool {
			return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
		})).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{
			err: errors.New("kubernetes api unavailable"),
		}
		dpm.clients["dataplane-1"] = k8sClient

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

		require.Error(t, err)
		assert.Contains(t, err.Error(), "dataplane-1")
		assert.Empty(t, writer.calls)
	})
}

func TestWorkerNodeMeteringJob_Execute_UsesDataplaneScopedCheckpoints(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.MatchedBy(func(lock *entity.Lock) bool {
		return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
	})).Return(unlockFn, nil).Once()

	targetMinute := lastCompletedMinute()
	dp1Checkpoint := targetMinute - 60
	dp2Checkpoint := targetMinute - 120

	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1").Return(dp1Checkpoint, nil).Once()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-2").Return(dp2Checkpoint, nil).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-1", targetMinute).Return(nil).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-2", 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",
		},
		{
			SharedPool: apiv1beta.SharedPool{Id: "pool-2", DataplaneName: "dataplane-2"},
			TenantId:   "tenant-2",
		},
	}, nil).Once()

	oldCreationTime := metav1.Unix(targetMinute-300, 0)

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

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

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

	err := job.Execute(context.Background())
	require.NoError(t, err)
	require.Len(t, writer.calls, 3)

	actualCounts := map[string]int{}
	for _, call := range writer.calls {
		actualCounts[call.SharedPoolID]++
	}
	assert.Equal(t, map[string]int{"pool-1": 1, "pool-2": 2}, actualCounts)
}

func TestWorkerNodeMeteringJob_Execute_CompensatesFailedDataplaneIndependently(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.MatchedBy(func(lock *entity.Lock) bool {
		return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
	})).Return(unlockFn, nil).Once()

	targetMinute := lastCompletedMinute()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-ok").Return(targetMinute-60, nil).Once()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-fail").Return(targetMinute-120, nil).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-ok", targetMinute).Return(nil).Once()

	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{Id: "pool-ok", DataplaneName: "dataplane-ok"},
			TenantId:   "tenant-ok",
		},
		{
			SharedPool: apiv1beta.SharedPool{Id: "pool-fail", DataplaneName: "dataplane-fail"},
			TenantId:   "tenant-fail",
		},
	}, nil).Once()

	oldCreationTime := metav1.Unix(targetMinute-300, 0)
	dpm.clients["dataplane-ok"] = &fakeK8sClient{
		nodes: corev1.NodeList{Items: []corev1.Node{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "worker-node-ok",
				CreationTimestamp: oldCreationTime,
				Labels:            map[string]string{"node-spec": "8C16G"},
			},
			Spec: corev1.NodeSpec{Taints: []corev1.Taint{{Key: "dedicated", Value: "worker", Effect: corev1.TaintEffectNoSchedule}}},
		}}},
		pods: corev1.PodList{Items: []corev1.Pod{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "tenant-pod-ok",
				Namespace:         "default",
				CreationTimestamp: oldCreationTime,
				Labels: map[string]string{
					"tags.tidbcloud.com/tenant":      "tenant-ok",
					"tags.tidbcloud.com/shared-pool": "pool-ok",
				},
			},
			Spec: corev1.PodSpec{NodeName: "worker-node-ok"},
		}}},
	}

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

	err := job.Execute(context.Background())
	require.Error(t, err)
	assert.Contains(t, err.Error(), "dataplane-fail")
	require.Len(t, writer.calls, 1)
	assert.Equal(t, "pool-ok", writer.calls[0].SharedPoolID)
	assert.Equal(t, targetMinute, writer.calls[0].Timestamp)
}

func TestWorkerNodeMeteringJob_Execute_LoggingAndDropBehavior(t *testing.T) {
	unlockFn := func() error { return nil }

	t.Run("identity-missing non-DaemonSet pod does not produce billing", func(t *testing.T) {
		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.MatchedBy(func(lock *entity.Lock) bool {
			return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
		})).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-node-1",
							Labels: map[string]string{
								"node-spec": "8C16G",
							},
						},
						Spec: corev1.NodeSpec{
							Taints: []corev1.Taint{
								{
									Key:    "dedicated",
									Value:  "worker",
									Effect: corev1.TaintEffectNoSchedule,
								},
							},
						},
					},
				},
			},
			pods: corev1.PodList{
				Items: []corev1.Pod{
					{
						ObjectMeta: metav1.ObjectMeta{
							Name:      "pod-missing-tenant-label",
							Namespace: "default",
							Labels: map[string]string{
								"tags.tidbcloud.com/shared-pool": "pool-1",
							},
						},
						Spec: corev1.PodSpec{
							NodeName: "worker-node-1",
						},
					},
					{
						ObjectMeta: metav1.ObjectMeta{
							Name:      "pod-missing-pool-label",
							Namespace: "default",
							Labels: map[string]string{
								"tags.tidbcloud.com/tenant": "tenant-1",
							},
						},
						Spec: corev1.PodSpec{
							NodeName: "worker-node-1",
						},
					},
				},
			},
		}
		dpm.clients["dataplane-1"] = k8sClient

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

		require.NoError(t, err)
		assert.Empty(t, writer.calls)
	})

	t.Run("conflicting ownership does not produce billing", func(t *testing.T) {
		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.MatchedBy(func(lock *entity.Lock) bool {
			return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
		})).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-node-1",
							Labels: map[string]string{
								"node-spec": "8C16G",
							},
						},
						Spec: corev1.NodeSpec{
							Taints: []corev1.Taint{
								{
									Key:    "dedicated",
									Value:  "worker",
									Effect: corev1.TaintEffectNoSchedule,
								},
							},
						},
					},
				},
			},
			pods: corev1.PodList{
				Items: []corev1.Pod{
					{
						ObjectMeta: metav1.ObjectMeta{
							Name:      "tenant1-pod",
							Namespace: "default",
							Labels: map[string]string{
								"tags.tidbcloud.com/tenant":      "tenant-1",
								"tags.tidbcloud.com/shared-pool": "pool-1",
							},
						},
						Spec: corev1.PodSpec{
							NodeName: "worker-node-1",
						},
					},
					{
						ObjectMeta: metav1.ObjectMeta{
							Name:      "tenant2-pod",
							Namespace: "default",
							Labels: map[string]string{
								"tags.tidbcloud.com/tenant":      "tenant-2",
								"tags.tidbcloud.com/shared-pool": "pool-2",
							},
						},
						Spec: corev1.PodSpec{
							NodeName: "worker-node-1",
						},
					},
				},
			},
		}
		dpm.clients["dataplane-1"] = k8sClient

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

		require.NoError(t, err)
		assert.Empty(t, writer.calls)
	})

	t.Run("invalid node-spec does not produce billing", func(t *testing.T) {
		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.MatchedBy(func(lock *entity.Lock) bool {
			return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
		})).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-node-1",
							Labels: map[string]string{
								"node-spec": "invalid-format",
							},
						},
						Spec: corev1.NodeSpec{
							Taints: []corev1.Taint{
								{
									Key:    "dedicated",
									Value:  "worker",
									Effect: corev1.TaintEffectNoSchedule,
								},
							},
						},
					},
					{
						ObjectMeta: metav1.ObjectMeta{
							Name:   "worker-node-2",
							Labels: map[string]string{},
						},
						Spec: corev1.NodeSpec{
							Taints: []corev1.Taint{
								{
									Key:    "dedicated",
									Value:  "worker",
									Effect: corev1.TaintEffectNoSchedule,
								},
							},
						},
					},
				},
			},
			pods: corev1.PodList{
				Items: []corev1.Pod{
					{
						ObjectMeta: metav1.ObjectMeta{
							Name:      "tenant-pod-1",
							Namespace: "default",
							Labels: map[string]string{
								"tags.tidbcloud.com/tenant":      "tenant-1",
								"tags.tidbcloud.com/shared-pool": "pool-1",
							},
						},
						Spec: corev1.PodSpec{
							NodeName: "worker-node-1",
						},
					},
					{
						ObjectMeta: metav1.ObjectMeta{
							Name:      "tenant-pod-2",
							Namespace: "default",
							Labels: map[string]string{
								"tags.tidbcloud.com/tenant":      "tenant-1",
								"tags.tidbcloud.com/shared-pool": "pool-1",
							},
						},
						Spec: corev1.PodSpec{
							NodeName: "worker-node-2",
						},
					},
				},
			},
		}
		dpm.clients["dataplane-1"] = k8sClient

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

		require.NoError(t, err)
		assert.Empty(t, writer.calls)
	})

	t.Run("node-spec vs status.capacity.cpu mismatch emits warning but still bills using node-spec", func(t *testing.T) {
		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.MatchedBy(func(lock *entity.Lock) bool {
			return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
		})).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-node-1",
							Labels: map[string]string{
								"node-spec": "8C16G",
							},
						},
						Spec: corev1.NodeSpec{
							Taints: []corev1.Taint{
								{
									Key:    "dedicated",
									Value:  "worker",
									Effect: corev1.TaintEffectNoSchedule,
								},
							},
						},
						Status: corev1.NodeStatus{
							Capacity: corev1.ResourceList{
								corev1.ResourceCPU: *resource.NewQuantity(16, resource.DecimalSI),
							},
						},
					},
				},
			},
			pods: corev1.PodList{
				Items: []corev1.Pod{
					{
						ObjectMeta: metav1.ObjectMeta{
							Name:      "tenant-pod-1",
							Namespace: "default",
							Labels: map[string]string{
								"tags.tidbcloud.com/tenant":      "tenant-1",
								"tags.tidbcloud.com/shared-pool": "pool-1",
							},
						},
						Spec: corev1.PodSpec{
							NodeName: "worker-node-1",
						},
					},
				},
			},
		}
		dpm.clients["dataplane-1"] = k8sClient

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

		require.NoError(t, err)
		require.Len(t, writer.calls, 1)
		assert.InDelta(t, float64(8), writer.calls[0].Data[0]["usage"].(map[string]any)["value"], 1e-9)
	})
}

func TestWorkerNodeMeteringJob_Execute_CheckpointLoadFailureContinuesWithOtherDataplanes(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.MatchedBy(func(lock *entity.Lock) bool {
		return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
	})).Return(unlockFn, nil).Once()

	targetMinute := lastCompletedMinute()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-fail").
		Return(int64(0), errors.New("checkpoint load failed")).Once()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-ok").
		Return(int64(0), nil).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-ok", targetMinute).
		Return(nil).Once()

	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{Id: "pool-fail", DataplaneName: "dataplane-fail"},
			TenantId:   "tenant-fail",
		},
		{
			SharedPool: apiv1beta.SharedPool{Id: "pool-ok", DataplaneName: "dataplane-ok"},
			TenantId:   "tenant-ok",
		},
	}, nil).Once()

	oldCreationTime := metav1.Unix(targetMinute-300, 0)
	dpm.clients["dataplane-ok"] = &fakeK8sClient{
		nodes: corev1.NodeList{Items: []corev1.Node{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "worker-node-ok",
				CreationTimestamp: oldCreationTime,
				Labels:            map[string]string{"node-spec": "8C16G"},
			},
			Spec: corev1.NodeSpec{Taints: []corev1.Taint{{Key: "dedicated", Value: "worker", Effect: corev1.TaintEffectNoSchedule}}},
		}}},
		pods: corev1.PodList{Items: []corev1.Pod{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "tenant-pod-ok",
				Namespace:         "default",
				CreationTimestamp: oldCreationTime,
				Labels: map[string]string{
					"tags.tidbcloud.com/tenant":      "tenant-ok",
					"tags.tidbcloud.com/shared-pool": "pool-ok",
				},
			},
			Spec: corev1.PodSpec{NodeName: "worker-node-ok"},
		}}},
	}

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

	err := job.Execute(context.Background())

	require.Error(t, err)
	assert.Contains(t, err.Error(), "dataplane-fail")

	require.Len(t, writer.calls, 1)
	assert.Equal(t, "pool-ok", writer.calls[0].SharedPoolID)
	assert.Equal(t, targetMinute, writer.calls[0].Timestamp)
}

func TestWorkerNodeMeteringJob_Execute_CheckpointSaveFailureContinuesWithOtherDataplanes(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.MatchedBy(func(lock *entity.Lock) bool {
		return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
	})).Return(unlockFn, nil).Once()

	targetMinute := lastCompletedMinute()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-fail").
		Return(int64(0), nil).Once()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-ok").
		Return(int64(0), nil).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-fail", targetMinute).
		Return(errors.New("checkpoint save failed")).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-ok", targetMinute).
		Return(nil).Once()

	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{Id: "pool-fail", DataplaneName: "dataplane-fail"},
			TenantId:   "tenant-fail",
		},
		{
			SharedPool: apiv1beta.SharedPool{Id: "pool-ok", DataplaneName: "dataplane-ok"},
			TenantId:   "tenant-ok",
		},
	}, nil).Once()

	oldCreationTime := metav1.Unix(targetMinute-300, 0)
	dpm.clients["dataplane-fail"] = &fakeK8sClient{
		nodes: corev1.NodeList{Items: []corev1.Node{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "worker-node-fail",
				CreationTimestamp: oldCreationTime,
				Labels:            map[string]string{"node-spec": "16C32G"},
			},
			Spec: corev1.NodeSpec{Taints: []corev1.Taint{{Key: "dedicated", Value: "worker", Effect: corev1.TaintEffectNoSchedule}}},
		}}},
		pods: corev1.PodList{Items: []corev1.Pod{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "tenant-pod-fail",
				Namespace:         "default",
				CreationTimestamp: oldCreationTime,
				Labels: map[string]string{
					"tags.tidbcloud.com/tenant":      "tenant-fail",
					"tags.tidbcloud.com/shared-pool": "pool-fail",
				},
			},
			Spec: corev1.PodSpec{NodeName: "worker-node-fail"},
		}}},
	}
	dpm.clients["dataplane-ok"] = &fakeK8sClient{
		nodes: corev1.NodeList{Items: []corev1.Node{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "worker-node-ok",
				CreationTimestamp: oldCreationTime,
				Labels:            map[string]string{"node-spec": "8C16G"},
			},
			Spec: corev1.NodeSpec{Taints: []corev1.Taint{{Key: "dedicated", Value: "worker", Effect: corev1.TaintEffectNoSchedule}}},
		}}},
		pods: corev1.PodList{Items: []corev1.Pod{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "tenant-pod-ok",
				Namespace:         "default",
				CreationTimestamp: oldCreationTime,
				Labels: map[string]string{
					"tags.tidbcloud.com/tenant":      "tenant-ok",
					"tags.tidbcloud.com/shared-pool": "pool-ok",
				},
			},
			Spec: corev1.PodSpec{NodeName: "worker-node-ok"},
		}}},
	}

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

	err := job.Execute(context.Background())

	require.Error(t, err)
	assert.Contains(t, err.Error(), "dataplane-fail")

	require.Len(t, writer.calls, 2)
	poolIDs := make(map[string]bool)
	for _, call := range writer.calls {
		poolIDs[call.SharedPoolID] = true
	}
	assert.True(t, poolIDs["pool-fail"])
	assert.True(t, poolIDs["pool-ok"])
}

func TestWorkerNodeMeteringJob_Execute_MeteringWriteFailureContinuesWithOtherDataplanes(t *testing.T) {
	unlockFn := func() error { return nil }

	checkpointRepo := newFakeWorkerNodeMeteringCheckpointRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	writer := &selectiveFailMeteringWriter{
		errBySharedPoolID: map[string]error{
			"pool-fail": errors.New("metering write failed"),
		},
	}
	dpm := &fakeDataplaneManager{clients: make(map[string]client.Client)}

	lockRepo.EXPECT().Lock(mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
		return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
	})).Return(unlockFn, nil).Once()

	targetMinute := lastCompletedMinute()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-fail").
		Return(int64(0), nil).Once()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-ok").
		Return(int64(0), nil).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-ok", targetMinute).
		Return(nil).Once()

	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{Id: "pool-fail", DataplaneName: "dataplane-fail"},
			TenantId:   "tenant-fail",
		},
		{
			SharedPool: apiv1beta.SharedPool{Id: "pool-ok", DataplaneName: "dataplane-ok"},
			TenantId:   "tenant-ok",
		},
	}, nil).Once()

	oldCreationTime := metav1.Unix(targetMinute-300, 0)
	dpm.clients["dataplane-fail"] = &fakeK8sClient{
		nodes: corev1.NodeList{Items: []corev1.Node{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "worker-node-fail",
				CreationTimestamp: oldCreationTime,
				Labels:            map[string]string{"node-spec": "16C32G"},
			},
			Spec: corev1.NodeSpec{Taints: []corev1.Taint{{Key: "dedicated", Value: "worker", Effect: corev1.TaintEffectNoSchedule}}},
		}}},
		pods: corev1.PodList{Items: []corev1.Pod{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "tenant-pod-fail",
				Namespace:         "default",
				CreationTimestamp: oldCreationTime,
				Labels: map[string]string{
					"tags.tidbcloud.com/tenant":      "tenant-fail",
					"tags.tidbcloud.com/shared-pool": "pool-fail",
				},
			},
			Spec: corev1.PodSpec{NodeName: "worker-node-fail"},
		}}},
	}
	dpm.clients["dataplane-ok"] = &fakeK8sClient{
		nodes: corev1.NodeList{Items: []corev1.Node{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "worker-node-ok",
				CreationTimestamp: oldCreationTime,
				Labels:            map[string]string{"node-spec": "8C16G"},
			},
			Spec: corev1.NodeSpec{Taints: []corev1.Taint{{Key: "dedicated", Value: "worker", Effect: corev1.TaintEffectNoSchedule}}},
		}}},
		pods: corev1.PodList{Items: []corev1.Pod{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "tenant-pod-ok",
				Namespace:         "default",
				CreationTimestamp: oldCreationTime,
				Labels: map[string]string{
					"tags.tidbcloud.com/tenant":      "tenant-ok",
					"tags.tidbcloud.com/shared-pool": "pool-ok",
				},
			},
			Spec: corev1.PodSpec{NodeName: "worker-node-ok"},
		}}},
	}

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

	err := job.Execute(context.Background())

	require.Error(t, err)
	assert.Contains(t, err.Error(), "dataplane-fail")

	require.Len(t, writer.calls, 1)
	assert.Equal(t, "pool-ok", writer.calls[0].SharedPoolID)
	assert.Equal(t, targetMinute, writer.calls[0].Timestamp)
}

func TestWorkerNodeMeteringJob_Execute_DataplaneAccessFailuresContinueWithOtherDataplanes(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.MatchedBy(func(lock *entity.Lock) bool {
		return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
	})).Return(unlockFn, nil).Once()

	targetMinute := lastCompletedMinute()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-client-fail").
		Return(int64(0), nil).Once()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-fetch-fail").
		Return(int64(0), nil).Once()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-ok").
		Return(int64(0), nil).Once()
	checkpointRepo.EXPECT().SaveLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-ok", targetMinute).
		Return(nil).Once()

	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{Id: "pool-client-fail", DataplaneName: "dataplane-client-fail"},
			TenantId:   "tenant-client-fail",
		},
		{
			SharedPool: apiv1beta.SharedPool{Id: "pool-fetch-fail", DataplaneName: "dataplane-fetch-fail"},
			TenantId:   "tenant-fetch-fail",
		},
		{
			SharedPool: apiv1beta.SharedPool{Id: "pool-ok", DataplaneName: "dataplane-ok"},
			TenantId:   "tenant-ok",
		},
	}, nil).Once()

	oldCreationTime := metav1.Unix(targetMinute-300, 0)
	dpm.clients["dataplane-fetch-fail"] = &fakeK8sClient{err: errors.New("list failed")}
	dpm.clients["dataplane-ok"] = &fakeK8sClient{
		nodes: corev1.NodeList{Items: []corev1.Node{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "worker-node-ok",
				CreationTimestamp: oldCreationTime,
				Labels:            map[string]string{"node-spec": "8C16G"},
			},
			Spec: corev1.NodeSpec{Taints: []corev1.Taint{{Key: "dedicated", Value: "worker", Effect: corev1.TaintEffectNoSchedule}}},
		}}},
		pods: corev1.PodList{Items: []corev1.Pod{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "tenant-pod-ok",
				Namespace:         "default",
				CreationTimestamp: oldCreationTime,
				Labels: map[string]string{
					"tags.tidbcloud.com/tenant":      "tenant-ok",
					"tags.tidbcloud.com/shared-pool": "pool-ok",
				},
			},
			Spec: corev1.PodSpec{NodeName: "worker-node-ok"},
		}}},
	}

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

	err := job.Execute(context.Background())

	require.Error(t, err)
	require.ErrorContains(t, err, "dataplane-client-fail")
	require.ErrorContains(t, err, "dataplane-fetch-fail")

	require.Len(t, writer.calls, 1)
	assert.Equal(t, "pool-ok", writer.calls[0].SharedPoolID)
	assert.Equal(t, targetMinute, writer.calls[0].Timestamp)
}

func TestWorkerNodeMeteringJob_Execute_JoinsMultipleDataplaneErrors(t *testing.T) {
	unlockFn := func() error { return nil }

	checkpointLoadErr := errors.New("checkpoint load failed")
	meteringWriteErr := errors.New("metering write failed")

	checkpointRepo := newFakeWorkerNodeMeteringCheckpointRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	writer := &selectiveFailMeteringWriter{
		errBySharedPoolID: map[string]error{
			"pool-write-fail": meteringWriteErr,
		},
	}
	dpm := &fakeDataplaneManager{clients: make(map[string]client.Client)}

	lockRepo.EXPECT().Lock(mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
		return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
	})).Return(unlockFn, nil).Once()

	targetMinute := lastCompletedMinute()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-checkpoint-fail").
		Return(int64(0), checkpointLoadErr).Once()
	checkpointRepo.EXPECT().LoadLastSuccessfulMinute(mock.Anything, "us-east-1", "aws", "dataplane-write-fail").
		Return(int64(0), nil).Once()

	sharedPoolRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.SharedPool{
		{
			SharedPool: apiv1beta.SharedPool{Id: "pool-checkpoint-fail", DataplaneName: "dataplane-checkpoint-fail"},
			TenantId:   "tenant-checkpoint-fail",
		},
		{
			SharedPool: apiv1beta.SharedPool{Id: "pool-write-fail", DataplaneName: "dataplane-write-fail"},
			TenantId:   "tenant-write-fail",
		},
	}, nil).Once()

	oldCreationTime := metav1.Unix(targetMinute-300, 0)
	dpm.clients["dataplane-write-fail"] = &fakeK8sClient{
		nodes: corev1.NodeList{Items: []corev1.Node{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "worker-node-write-fail",
				CreationTimestamp: oldCreationTime,
				Labels:            map[string]string{"node-spec": "8C16G"},
			},
			Spec: corev1.NodeSpec{Taints: []corev1.Taint{{Key: "dedicated", Value: "worker", Effect: corev1.TaintEffectNoSchedule}}},
		}}},
		pods: corev1.PodList{Items: []corev1.Pod{{
			ObjectMeta: metav1.ObjectMeta{
				Name:              "tenant-pod-write-fail",
				Namespace:         "default",
				CreationTimestamp: oldCreationTime,
				Labels: map[string]string{
					"tags.tidbcloud.com/tenant":      "tenant-write-fail",
					"tags.tidbcloud.com/shared-pool": "pool-write-fail",
				},
			},
			Spec: corev1.PodSpec{NodeName: "worker-node-write-fail"},
		}}},
	}

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

	err := job.Execute(context.Background())

	require.Error(t, err)
	require.ErrorIs(t, err, checkpointLoadErr)
	require.ErrorIs(t, err, meteringWriteErr)
	assert.Contains(t, err.Error(), "dataplane-checkpoint-fail")
	assert.Contains(t, err.Error(), "dataplane-write-fail")
}

func TestMinutesToProcess_OnlyReturnsCompletedMinutes(t *testing.T) {
	now := time.Now().Unix()
	currentMinute := now / 60 * 60
	lastCompletedMinute := currentMinute - 60

	t.Run("initial run bills the last completed minute", func(t *testing.T) {
		minutes := minutesToProcess(0, lastCompletedMinute, 30*time.Minute)
		require.Equal(t, []int64{lastCompletedMinute}, minutes)
	})

	t.Run("incremental run stops at the last completed minute", func(t *testing.T) {
		minutes := minutesToProcess(lastCompletedMinute-120, lastCompletedMinute, 30*time.Minute)
		require.Equal(t, []int64{lastCompletedMinute - 60, lastCompletedMinute}, minutes)
	})
}

func TestWorkerNodeMeteringJob_Execute_IgnoresFinishedAndDeletingPods(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.MatchedBy(func(lock *entity.Lock) bool {
		return lock.Key == "worker_node_metering" && lock.TimeoutSecs == 10
	})).Return(unlockFn, nil).Once()

	targetMinute := lastCompletedMinute()
	oldCreationTime := metav1.Unix(targetMinute-300, 0)
	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", 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()

	deletingTime := metav1.Unix(targetMinute-30, 0)
	dpm.clients["dataplane-1"] = &fakeK8sClient{
		nodes: corev1.NodeList{
			Items: []corev1.Node{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:              "worker-node-1",
						CreationTimestamp: oldCreationTime,
						Labels: map[string]string{
							"node-spec": "8C16G",
						},
					},
					Spec: corev1.NodeSpec{
						Taints: []corev1.Taint{
							{
								Key:    "dedicated",
								Value:  "worker",
								Effect: corev1.TaintEffectNoSchedule,
							},
						},
					},
				},
			},
		},
		pods: corev1.PodList{
			Items: []corev1.Pod{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:              "finished-pod",
						Namespace:         "default",
						CreationTimestamp: oldCreationTime,
						Labels: map[string]string{
							"tags.tidbcloud.com/tenant":      "tenant-1",
							"tags.tidbcloud.com/shared-pool": "pool-1",
						},
					},
					Spec: corev1.PodSpec{
						NodeName: "worker-node-1",
					},
					Status: corev1.PodStatus{
						Phase: corev1.PodSucceeded,
					},
				},
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:              "deleting-pod",
						Namespace:         "default",
						CreationTimestamp: oldCreationTime,
						DeletionTimestamp: &deletingTime,
						Labels: map[string]string{
							"tags.tidbcloud.com/tenant":      "tenant-1",
							"tags.tidbcloud.com/shared-pool": "pool-1",
						},
					},
					Spec: corev1.PodSpec{
						NodeName: "worker-node-1",
					},
					Status: corev1.PodStatus{
						Phase: corev1.PodRunning,
					},
				},
			},
		},
	}

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

	err := job.Execute(context.Background())

	require.NoError(t, err)
	assert.Empty(t, writer.calls)
}

func TestWorkerNodeMeteringJob_PodIdentitySupportsResourcePoolLabel(t *testing.T) {
	job := &WorkerNodeMeteringJob{}
	pod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Labels: map[string]string{
				"tags.tidbcloud.com/tenant":        "tenant-1",
				"tags.tidbcloud.com/resource-pool": "pool-1",
			},
		},
	}

	identity, ok := job.podIdentity(pod)

	require.True(t, ok)
	assert.Equal(t, podIdentity{
		tenantID: "tenant-1",
		poolID:   "pool-1",
	}, identity)
}

func assertWorkerNodeMeteringPayloads(t *testing.T, calls []*meteringsdkcommon.MeteringData, expectedPayloads map[string][]map[string]any) {
	t.Helper()
	if !assert.Len(t, calls, len(expectedPayloads)) {
		return
	}

	actualByPool := make(map[string]*meteringsdkcommon.MeteringData, len(calls))
	for _, call := range calls {
		actualByPool[call.SharedPoolID] = call
	}

	if !assert.Len(t, actualByPool, len(expectedPayloads)) {
		return
	}

	for sharedPoolID, expectedRows := range expectedPayloads {
		call, ok := actualByPool[sharedPoolID]
		if !assert.True(t, ok, "missing payload for shared pool %s", sharedPoolID) {
			continue
		}

		assert.Equal(t, "worker-node", call.Category)
		assert.Equal(t, sharedPoolID, call.SharedPoolID)
		assert.NotEmpty(t, call.SelfID)
		assert.NotContains(t, call.SelfID, "-")
		assert.NotZero(t, call.Timestamp)
		assert.EqualValues(t, 0, call.Timestamp%60)
		assert.Equal(t, expectedRows, call.Data)
	}
}

func lastCompletedMinute() int64 {
	return time.Now().Unix()/60*60 - 60
}