package impl

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

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	ticdc_mock "github.com/tidbcloud/dataflow-service/app/models/ticdc/mock"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	account_mock "github.com/tidbcloud/dataflow-service/pkg/account/accountdb/mock"
	"github.com/tidbcloud/dataflow-service/pkg/account/accounttypes"
	grpcli_mock "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli/mock"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

type BizChangefeedMeteringWorkerTestSuite struct {
	suite.Suite

	worker *BizChangefeedMeteringWorkerImpl

	mockMachineManager    *ticdc_mock.MachineManager
	mockBizChangefeedRepo *repo_mock.BizChangefeedRepo
	mockBillingRecordRepo *repo_mock.BillingRecordRepo
	mockCoreCli           *grpcli_mock.CoreGrpcClient
	mockProjectWrapper    *account_mock.ProjectWrapper
}

func (s *BizChangefeedMeteringWorkerTestSuite) SetupSuite() {
	s.mockMachineManager = ticdc_mock.NewMachineManager(s.T())
	s.mockBizChangefeedRepo = repo_mock.NewBizChangefeedRepo(s.T())
	s.mockBillingRecordRepo = repo_mock.NewBillingRecordRepo(s.T())
	s.mockCoreCli = grpcli_mock.NewCoreGrpcClient(s.T())
	s.mockProjectWrapper = account_mock.NewProjectWrapper(s.T())

	s.worker = &BizChangefeedMeteringWorkerImpl{
		machineManager:        s.mockMachineManager,
		bizChangefeedRepo:     s.mockBizChangefeedRepo,
		billingRecordRepo:     s.mockBillingRecordRepo,
		coreCli:               s.mockCoreCli,
		accountProjectWrapper: s.mockProjectWrapper,
	}
}

func (s *BizChangefeedMeteringWorkerTestSuite) TestListProjectsByBizChangefeeds() {
	mockBizChangefeeds := []entity.BizChangefeed{
		{
			ID:        1111,
			ProjectID: 11,
		},
		{
			ID:        1112,
			ProjectID: 11,
		},
	}
	projectIDs := []uint64{11}
	s.Run("list projects error", func() {
		mockErr := errors.New("network error")
		call := s.mockProjectWrapper.
			On("ListProjectMapByIDs", mock.Anything, projectIDs).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.worker.listProjectsByBizChangefeeds(context.Background(), mockBizChangefeeds)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	s.Run("list projects successfully", func() {
		mockProjects := map[uint64]accounttypes.Project{
			11: {
				Name: "11",
			},
		}
		call := s.mockProjectWrapper.
			On("ListProjectMapByIDs", mock.Anything, projectIDs).
			Return(mockProjects, nil)
		defer call.Unset()
		projects, err := s.worker.listProjectsByBizChangefeeds(context.Background(), mockBizChangefeeds)
		s.NoError(err)
		s.Equal(mockProjects, projects)
		s.mockProjectWrapper.AssertExpectations(s.T())
	})
}

func (s *BizChangefeedMeteringWorkerTestSuite) TestListSpecifications() {
	s.Run("list specs error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockMachineManager.
			On("ListSpecifications", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.worker.listSpecifications(context.Background())
		s.Error(err)
		s.Equal(mockErr, err)
	})

	s.Run("list specs successfully", func() {
		mockListSpecResp := &domain.ListSpecificationsResp{
			Specifications: []*domain.TiCDCClusterSpecification{
				{
					Code: "a",
				},
				{
					Code: "b",
				},
			},
		}
		expectedSpes := map[string]*domain.TiCDCClusterSpecification{
			"a": mockListSpecResp.Specifications[0],
			"b": mockListSpecResp.Specifications[1],
		}
		call := s.mockMachineManager.
			On("ListSpecifications", mock.Anything, mock.Anything).
			Return(mockListSpecResp, nil)
		defer call.Unset()
		specs, err := s.worker.listSpecifications(context.Background())
		s.NoError(err)
		s.Equal(expectedSpes, specs)
	})
}

func (s *BizChangefeedMeteringWorkerTestSuite) TestListClustersByBizChangefeeds() {
	mockBizChangefeeds := []entity.BizChangefeed{
		{
			ID:        1111,
			ClusterID: 111,
			ProjectID: 11,
		},
		{
			ID:        1112,
			ClusterID: 112,
			ProjectID: 11,
		},
	}

	s.Run("list clusters error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockCoreCli.
			On("ListClusters", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.worker.listClustersByBizChangefeeds(context.Background(), mockBizChangefeeds)
		s.Error(err)
		s.Contains(err.Error(), "list clusters")
	})

	s.Run("list clusters successfully", func() {
		mockClusters := map[uint64]*core.Cluster{
			111: {
				Id: 111,
			},
			112: {
				Id: 112,
			},
		}
		mockResp := &core.ListClustersResp{
			Clusters: mockClusters,
		}
		matcher := mock.MatchedBy(func(req *core.ListClustersReq) bool {
			return len(req.Filter.ClusterIds) == 2 &&
				req.Filter.ClusterIds[0] == 111 &&
				req.Filter.ClusterIds[1] == 112 &&
				len(req.Filter.ProjectIds) == 1 &&
				req.Filter.ProjectIds[0] == 11
		})
		call := s.mockCoreCli.
			On("ListClusters", mock.Anything, mock.Anything).
			Return(mockResp, nil)
		defer call.Unset()
		clusters, err := s.worker.listClustersByBizChangefeeds(context.Background(), mockBizChangefeeds)
		s.NoError(err)
		s.Equal(mockClusters, clusters)
		s.mockCoreCli.AssertCalled(s.T(), "ListClusters", mock.Anything, matcher)
	})
}

func (s *BizChangefeedMeteringWorkerTestSuite) TestGetClusterTotalRCU() {
	testcases := []struct {
		bizChangefeeds []*entity.BizChangefeed
		specs          map[string]*domain.TiCDCClusterSpecification
		totalRCU       uint32
		err            bool
	}{
		{
			bizChangefeeds: nil,
			specs:          nil,
			totalRCU:       0,
			err:            false,
		},
		{
			bizChangefeeds: []*entity.BizChangefeed{},
			specs:          nil,
			totalRCU:       0,
			err:            false,
		},
		{
			bizChangefeeds: []*entity.BizChangefeed{
				{
					SpecificationCode: "a",
				},
			},
			specs:    nil,
			totalRCU: 0,
			err:      true,
		},
		{
			bizChangefeeds: []*entity.BizChangefeed{
				{
					SpecificationCode: "a",
				},
			},
			specs:    map[string]*domain.TiCDCClusterSpecification{},
			totalRCU: 0,
			err:      true,
		},
		{
			bizChangefeeds: []*entity.BizChangefeed{
				{
					SpecificationCode: "a",
				},
				{
					SpecificationCode: "b", // is not found in specs and cause error
				},
			},
			specs: map[string]*domain.TiCDCClusterSpecification{
				"a": {
					Code:     "a",
					RCUNum:   4,
					Replicas: 2,
				},
			},
			totalRCU: 0,
			err:      true,
		},
		{
			bizChangefeeds: []*entity.BizChangefeed{
				{
					SpecificationCode: "a",
				},
			},
			specs: map[string]*domain.TiCDCClusterSpecification{
				"a": {
					Code:     "a",
					RCUNum:   4,
					Replicas: 2,
				},
			},
			totalRCU: 4,
			err:      false,
		},
		{
			bizChangefeeds: []*entity.BizChangefeed{
				{
					SpecificationCode: "a",
				},
				{
					SpecificationCode: "b",
				},
				{
					SpecificationCode: "b",
				},
			},
			specs: map[string]*domain.TiCDCClusterSpecification{
				"a": {
					Code:     "a",
					RCUNum:   4,
					Replicas: 2,
				},
				"b": {
					Code:     "b",
					RCUNum:   8,
					Replicas: 2,
				},
			},
			totalRCU: 20,
			err:      false,
		},
	}

	for _, tc := range testcases {
		total, err := s.worker.getClusterTotalRCU(tc.bizChangefeeds, tc.specs)
		if tc.err {
			s.Error(err)
		} else {
			s.NoError(err)
			s.Equal(tc.totalRCU, total)
		}
	}
}

func (s *BizChangefeedMeteringWorkerTestSuite) TestGenBillingRecord() {
	mockBizChangefeeds := []*entity.BizChangefeed{
		{
			SpecificationCode: "a",
		},
	}
	mockSpecs := map[string]*domain.TiCDCClusterSpecification{}
	mockCluster := &core.Cluster{
		Id:                111,
		ProjectId:         11,
		TenantId:          1,
		Name:              "1",
		CloudProviderName: "aws",
		RegionName:        "us-west-2",
	}
	mockProject := &accounttypes.Project{
		Name: "11",
	}
	mockMeasuredAt := time.Now()
	mockMeteringHour := 3
	mockMeteringStart := mockMeasuredAt.Add(-time.Duration(mockMeteringHour) * time.Hour)

	s.Run("get total rcu error", func() {
		_, err := s.worker.genBillingRecord(
			mockBizChangefeeds, mockMeteringStart, mockMeasuredAt, mockCluster, mockProject, mockSpecs)
		s.Error(err)
		s.Contains(err.Error(), "get cluster total cost error")
	})

	mockSpecs["a"] = &domain.TiCDCClusterSpecification{
		Code:     "a",
		RCUNum:   4,
		Replicas: 2,
	}
	s.Run("gen billing record successfully", func() {
		record, err := s.worker.genBillingRecord(
			mockBizChangefeeds, mockMeteringStart, mockMeasuredAt, mockCluster, mockProject, mockSpecs)
		s.NoError(err)
		s.Equal(int64(720), record.Total)
		s.Equal(bizChangefeedMeteringCategory, record.Category)
		s.Equal(genChangefeedSKU(mockCluster.CloudProviderName, mockCluster.RegionName), record.SKU)
	})

	s.Run("billing starts from changefeed creation time", func() {
		measuredAt := time.Date(2026, 2, 6, 12, 0, 0, 0, time.UTC)
		meteringStart := measuredAt.Add(-10 * time.Hour)
		bizChangefeeds := []*entity.BizChangefeed{
			{
				SpecificationCode: "a",
				CreatedAt:         measuredAt.Add(-20 * time.Hour),
			},
			{
				SpecificationCode: "a",
				CreatedAt:         measuredAt.Add(-30 * time.Minute),
			},
		}
		record, err := s.worker.genBillingRecord(
			bizChangefeeds, meteringStart, measuredAt, mockCluster, mockProject, mockSpecs)
		s.NoError(err)
		// 10 hours for the old changefeed + 1 hour for the new changefeed (hour-based billing)
		s.Equal(int64(2640), record.Total)
	})
}

func (s *BizChangefeedMeteringWorkerTestSuite) TestGetBizChangefeedCost() {
	spec := &domain.TiCDCClusterSpecification{
		Code:     "a",
		RCUNum:   4,
		Replicas: 2,
	}
	base := time.Date(2026, 2, 6, 12, 0, 0, 0, time.UTC)
	testcases := []struct {
		name          string
		createdAt     time.Time
		meteringStart time.Time
		measuredAt    time.Time
		expectedCost  int
	}{
		{
			name:          "created before metering start",
			createdAt:     base.Add(-48 * time.Hour),
			meteringStart: base.Add(-3 * time.Hour),
			measuredAt:    base,
			expectedCost:  720, // 4 RCU * 3h * 60
		},
		{
			name:          "created after metering start",
			createdAt:     base.Add(-90 * time.Minute),
			meteringStart: base.Add(-3 * time.Hour),
			measuredAt:    base,
			expectedCost:  480, // effectiveStart = 10:00, 2h * 4 * 60
		},
		{
			name:          "measured at effective start",
			createdAt:     base.Add(-30 * time.Minute),
			meteringStart: base.Add(-3 * time.Hour),
			measuredAt:    base.Add(-30 * time.Minute).Truncate(time.Hour),
			expectedCost:  0,
		},
		{
			name:          "measured before metering start",
			createdAt:     base.Add(-5 * time.Hour),
			meteringStart: base,
			measuredAt:    base,
			expectedCost:  0,
		},
	}

	for _, tc := range testcases {
		s.Run(tc.name, func() {
			bizChangefeed := &entity.BizChangefeed{
				SpecificationCode: "a",
				CreatedAt:         tc.createdAt,
			}
			cost := s.worker.getBizChangefeedCost(
				bizChangefeed, spec, tc.meteringStart, tc.measuredAt)
			s.Equal(tc.expectedCost, cost)
		})
	}
}

func (s *BizChangefeedMeteringWorkerTestSuite) TestGenBillingRecordForOneCluster() {
	mockBizChangefeeds := []*entity.BizChangefeed{
		{
			SpecificationCode: "a",
		},
	}
	mockSpecs := map[string]*domain.TiCDCClusterSpecification{}
	mockCluster := &core.Cluster{
		Id:                111,
		ProjectId:         11,
		TenantId:          1,
		Name:              "1",
		CloudProviderName: "aws",
		RegionName:        "us-west-2",
	}
	mockProject := &accounttypes.Project{
		Name: "11",
	}
	logger := zap.NewNop()

	s.Run("get billing record error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockBillingRecordRepo.
			On("Get", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.worker.genBillingRecordForOneCluster(
			context.Background(), mockBizChangefeeds, mockCluster, mockProject, mockSpecs, logger)
		s.Error(err)
		s.Contains(err.Error(), "get latest billing record")
	})

	mockBillingRecord := &entity.BillingRecord{
		MeasuredTS: time.Now().Truncate(time.Hour),
	}
	billingRecordCall := s.mockBillingRecordRepo.
		On("Get", mock.Anything, mock.Anything).
		Return(mockBillingRecord, nil)
	s.Run("billing record exists", func() {
		err := s.worker.genBillingRecordForOneCluster(
			context.Background(), mockBizChangefeeds, mockCluster, mockProject, mockSpecs, logger)
		s.NoError(err)
	})

	mockBillingRecord.MeasuredTS = time.Now().Add(-2 * time.Hour).Truncate(time.Hour)
	s.Run("gen billing record error", func() {
		err := s.worker.genBillingRecordForOneCluster(
			context.Background(), mockBizChangefeeds, mockCluster, mockProject, mockSpecs, logger)
		s.Error(err)
		s.Contains(err.Error(), "gen billing record error")
	})

	mockSpecs["a"] = &domain.TiCDCClusterSpecification{
		Code:     "a",
		RCUNum:   4,
		Replicas: 2,
	}
	s.Run("save billing record error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockBillingRecordRepo.On("Save", mock.Anything, mock.Anything).Return(mockErr)
		defer call.Unset()
		err := s.worker.genBillingRecordForOneCluster(
			context.Background(), mockBizChangefeeds, mockCluster, mockProject, mockSpecs, logger)
		s.Error(err)
		s.Contains(err.Error(), "save billing record")
	})

	saveCall := s.mockBillingRecordRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
	defer saveCall.Unset()
	s.Run("save billing record successfully", func() {
		err := s.worker.genBillingRecordForOneCluster(
			context.Background(), mockBizChangefeeds, mockCluster, mockProject, mockSpecs, logger)
		s.NoError(err)
		matcher := mock.MatchedBy(func(record *entity.BillingRecord) bool {
			return record.Total == 480
		})
		s.mockBillingRecordRepo.AssertCalled(s.T(), "Save", mock.Anything, matcher)
	})

	billingRecordCall.Unset()
	s.Run("billing record not found", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockBillingRecordRepo.
			On("Get", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.worker.genBillingRecordForOneCluster(
			context.Background(), mockBizChangefeeds, mockCluster, mockProject, mockSpecs, logger)
		s.NoError(err)
		matcher := mock.MatchedBy(func(record *entity.BillingRecord) bool {
			return record.Total == 240
		})
		s.mockBillingRecordRepo.AssertCalled(s.T(), "Save", mock.Anything, matcher)
	})
}

func (s *BizChangefeedMeteringWorkerTestSuite) TestGenBillingRecordsByCluster() {
	s.Run("list biz changefeed error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockBizChangefeedRepo.
			On("List", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.worker.GenBillingRecordsByCluster(context.Background())
		s.Error(err)
		s.Contains(err.Error(), "list biz_changefeeds")
	})

	s.Run("list biz changefeed empty", func() {
		call := s.mockBizChangefeedRepo.
			On("List", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer call.Unset()
		err := s.worker.GenBillingRecordsByCluster(context.Background())
		s.NoError(err)
	})

	mockBizChangefeed := []entity.BizChangefeed{
		{
			ID:                1111,
			ClusterID:         111,
			ProjectID:         11,
			SpecificationCode: "a",
		},
		{
			ID:                2222,
			ClusterID:         222,
			ProjectID:         22,
			SpecificationCode: "a",
		},
	}
	bizChangefeedCall := s.mockBizChangefeedRepo.
		On("List", mock.Anything, mock.Anything).
		Return(mockBizChangefeed, nil)
	defer bizChangefeedCall.Unset()
	s.Run("list clusters error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockCoreCli.
			On("ListClusters", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.worker.GenBillingRecordsByCluster(context.Background())
		s.Error(err)
		s.Contains(err.Error(), "list clusters error")
	})

	mockClusters := map[uint64]*core.Cluster{}
	mockClustersResp := &core.ListClustersResp{
		Clusters: mockClusters,
	}
	clusterCall := s.mockCoreCli.
		On("ListClusters", mock.Anything, mock.Anything).
		Return(mockClustersResp, nil)
	defer clusterCall.Unset()
	s.Run("list projects error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockProjectWrapper.
			On("ListProjectMapByIDs", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.worker.GenBillingRecordsByCluster(context.Background())
		s.Error(err)
		s.Contains(err.Error(), "list projects error")
	})

	mockProjects := map[uint64]accounttypes.Project{}
	projectCall := s.mockProjectWrapper.
		On("ListProjectMapByIDs", mock.Anything, mock.Anything).
		Return(mockProjects, nil)
	defer projectCall.Unset()
	s.Run("list specs error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockMachineManager.
			On("ListSpecifications", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.worker.GenBillingRecordsByCluster(context.Background())
		s.Error(err)
		s.Contains(err.Error(), "list specifications error")
	})

	var mockSpecs []*domain.TiCDCClusterSpecification
	mockSpecResp := &domain.ListSpecificationsResp{
		Specifications: mockSpecs,
	}
	specCall := s.mockMachineManager.
		On("ListSpecifications", mock.Anything, mock.Anything).
		Return(mockSpecResp, nil)
	defer specCall.Unset()
	s.Run("cluster not found", func() {
		err := s.worker.GenBillingRecordsByCluster(context.Background())
		s.Error(err)
		s.Contains(err.Error(), "cluster")
		s.Contains(err.Error(), "not found")
	})

	mockClusters[111] = &core.Cluster{
		Id:        111,
		ProjectId: 11,
	}
	mockClusters[222] = &core.Cluster{
		Id:        222,
		ProjectId: 22,
	}
	s.Run("project not found", func() {
		err := s.worker.GenBillingRecordsByCluster(context.Background())
		s.Error(err)
		s.Contains(err.Error(), "project")
		s.Contains(err.Error(), "not found")
	})

	mockProjects[11] = accounttypes.Project{Name: "11"}
	mockProjects[22] = accounttypes.Project{Name: "22"}
	mockNotFoundErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
	billingRecordCall := s.mockBillingRecordRepo.
		On("Get", mock.Anything, mock.Anything).
		Return(nil, mockNotFoundErr)
	defer billingRecordCall.Unset()
	s.Run("gen billing records error", func() {
		err := s.worker.GenBillingRecordsByCluster(context.Background())
		s.Error(err)
		s.Contains(err.Error(), "gen billing record")
	})

	mockSpecResp.Specifications = append(mockSpecResp.Specifications, &domain.TiCDCClusterSpecification{
		Code:     "a",
		RCUNum:   4,
		Replicas: 2,
	})
	saveBillingRecordCall := s.mockBillingRecordRepo.
		On("Save", mock.Anything, mock.Anything).Return(nil).Twice()
	defer saveBillingRecordCall.Unset()
	s.Run("gen billing records successfully", func() {
		err := s.worker.GenBillingRecordsByCluster(context.Background())
		s.NoError(err)
	})
}

func TestBizChangefeedMeteringWorkerTestSuite(t *testing.T) {
	suite.Run(t, new(BizChangefeedMeteringWorkerTestSuite))
}

func TestGetProject2ClustersFromBizChangefeeds(t *testing.T) {
	testcases := []struct {
		bizChangefeeds []entity.BizChangefeed
		result         map[uint64][]uint64
	}{
		{
			bizChangefeeds: nil,
			result:         map[uint64][]uint64{},
		},
		{
			bizChangefeeds: []entity.BizChangefeed{},
			result:         map[uint64][]uint64{},
		},
		{
			bizChangefeeds: []entity.BizChangefeed{
				{
					ClusterID: 111,
					ProjectID: 11,
				},
			},
			result: map[uint64][]uint64{
				11: {111},
			},
		},
		{
			bizChangefeeds: []entity.BizChangefeed{
				{
					ClusterID: 111,
					ProjectID: 11,
				},
				{
					ClusterID: 222,
					ProjectID: 22,
				},
			},
			result: map[uint64][]uint64{
				11: {111},
				22: {222},
			},
		},
		{
			bizChangefeeds: []entity.BizChangefeed{
				{
					ClusterID: 111,
					ProjectID: 11,
				},
				{
					ClusterID: 112,
					ProjectID: 11,
				},
				{
					ClusterID: 222,
					ProjectID: 22,
				},
			},
			result: map[uint64][]uint64{
				11: {111, 112},
				22: {222},
			},
		},
		{
			bizChangefeeds: []entity.BizChangefeed{
				{
					ClusterID: 111,
					ProjectID: 11,
				},
				{
					ClusterID: 111,
					ProjectID: 11,
				},
				{
					ClusterID: 112,
					ProjectID: 11,
				},
				{
					ClusterID: 222,
					ProjectID: 22,
				},
				{
					ClusterID: 222,
					ProjectID: 22,
				},
			},
			result: map[uint64][]uint64{
				11: {111, 112},
				22: {222},
			},
		},
	}

	for _, tc := range testcases {
		result := getProject2ClustersFromBizChangefeeds(tc.bizChangefeeds)
		assert.Equal(t, tc.result, result)
	}
}

func TestGetProjectIDSetFromBizChangefeeds(t *testing.T) {
	testcases := []struct {
		bizChangefeeds []entity.BizChangefeed
		result         []uint64
	}{
		{
			bizChangefeeds: nil,
			result:         []uint64{},
		},
		{
			bizChangefeeds: []entity.BizChangefeed{},
			result:         []uint64{},
		},
		{
			bizChangefeeds: []entity.BizChangefeed{
				{
					ProjectID: 11,
				},
			},
			result: []uint64{11},
		},
		{
			bizChangefeeds: []entity.BizChangefeed{
				{
					ProjectID: 11,
				},
				{
					ProjectID: 12,
				},
			},
			result: []uint64{11, 12},
		},
		{
			bizChangefeeds: []entity.BizChangefeed{
				{
					ProjectID: 11,
				},
				{
					ProjectID: 11,
				},
				{
					ProjectID: 12,
				},
			},
			result: []uint64{11, 12},
		},
	}

	for _, tc := range testcases {
		result := getProjectIDSetFromBizChangefeeds(tc.bizChangefeeds)
		assert.Equal(t, tc.result, result)
	}
}

func TestPartitionBizChangefeedsByCluster(t *testing.T) {
	testcases := []struct {
		bizChangefeeds []entity.BizChangefeed
		result         map[uint64][]*entity.BizChangefeed
	}{
		{
			bizChangefeeds: nil,
			result:         map[uint64][]*entity.BizChangefeed{},
		},
		{
			bizChangefeeds: []entity.BizChangefeed{},
			result:         map[uint64][]*entity.BizChangefeed{},
		},
		{
			bizChangefeeds: []entity.BizChangefeed{
				{
					ID:        1111,
					ClusterID: 111,
				},
			},
			result: map[uint64][]*entity.BizChangefeed{
				111: {
					{
						ID:        1111,
						ClusterID: 111,
					},
				},
			},
		},
		{
			bizChangefeeds: []entity.BizChangefeed{
				{
					ID:        1111,
					ClusterID: 111,
				},
				{
					ID:        2222,
					ClusterID: 222,
				},
			},
			result: map[uint64][]*entity.BizChangefeed{
				111: {
					{
						ID:        1111,
						ClusterID: 111,
					},
				},
				222: {
					{
						ID:        2222,
						ClusterID: 222,
					},
				},
			},
		},
		{
			bizChangefeeds: []entity.BizChangefeed{
				{
					ID:        1111,
					ClusterID: 111,
				},
				{
					ID:        1112,
					ClusterID: 111,
				},
				{
					ID:        2222,
					ClusterID: 222,
				},
			},
			result: map[uint64][]*entity.BizChangefeed{
				111: {
					{
						ID:        1111,
						ClusterID: 111,
					},
					{
						ID:        1112,
						ClusterID: 111,
					},
				},
				222: {
					{
						ID:        2222,
						ClusterID: 222,
					},
				},
			},
		},
	}

	for _, tc := range testcases {
		result := partitionBizChangefeedsByCluster(tc.bizChangefeeds)
		assert.Equal(t, tc.result, result)
	}
}