package impl

import (
	"context"
	"encoding/json"
	"errors"
	"reflect"
	"strconv"
	"testing"

	"github.com/AlekSi/pointer"
	workflow_entity "github.com/shiningrush/fastflow/pkg/entity"
	"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/config"
	"github.com/tidbcloud/dataflow-service/app/internal/cluster_job"
	cluster_job_mock "github.com/tidbcloud/dataflow-service/app/internal/cluster_job/mock"
	biz_changefeed_mock "github.com/tidbcloud/dataflow-service/app/models/biz_changefeed/mock"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	ticdc_impl "github.com/tidbcloud/dataflow-service/app/models/ticdc/impl"
	ticdc_mock "github.com/tidbcloud/dataflow-service/app/models/ticdc/mock"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	workflow_mock "github.com/tidbcloud/dataflow-service/app/workflow/mock"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/pkg/billing"
	billing_mock "github.com/tidbcloud/dataflow-service/pkg/billing/mock"
	grpcli_mock "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli/mock"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/db"
)

type bizChangefeedModelTestSuite struct {
	suite.Suite

	mockBizChangefeedRepo *repo_mock.BizChangefeedRepo

	mockMachineManager        *ticdc_mock.MachineManager
	mockKernelChangefeedModel *ticdc_mock.ChangefeedModel

	mockImportCli *grpcli_mock.ImportGrpcClient
	mockCoreCli   *grpcli_mock.CoreGrpcClient
	mockWorkflow  *workflow_mock.Workflow
	mockBilling   *billing_mock.BillingHandler

	mockMysqlAdaptor *biz_changefeed_mock.DownstreamAdaptor
	mockHooker       *biz_changefeed_mock.BizChangefeedHooker
	mockClusterJob   *cluster_job_mock.Model

	model *BizChangefeedModelImpl
}

func (s *bizChangefeedModelTestSuite) SetupSuite() {
	testDB, err := repo.GetFakeDB()
	s.NoError(err)

	dbProvider := db.NewProvider(testDB)
	s.mockBizChangefeedRepo = repo_mock.NewBizChangefeedRepo(s.T())

	s.mockMachineManager = ticdc_mock.NewMachineManager(s.T())
	s.mockKernelChangefeedModel = ticdc_mock.NewChangefeedModel(s.T())

	s.mockImportCli = grpcli_mock.NewImportGrpcClient(s.T())
	s.mockCoreCli = grpcli_mock.NewCoreGrpcClient(s.T())
	s.mockWorkflow = workflow_mock.NewWorkflow(s.T())
	s.mockBilling = billing_mock.NewBillingHandler(s.T())

	s.mockMysqlAdaptor = biz_changefeed_mock.NewDownstreamAdaptor(s.T())
	s.mockHooker = biz_changefeed_mock.NewBizChangefeedHooker(s.T())
	s.mockClusterJob = cluster_job_mock.NewModel(s.T())

	s.model = &BizChangefeedModelImpl{
		transProvider:     dbProvider,
		bizChangefeedRepo: s.mockBizChangefeedRepo,

		machineManager:                s.mockMachineManager,
		regionalKernelChangefeedModel: s.mockKernelChangefeedModel,
		kernelChangefeedModelFactory: ticdc_impl.NewChangefeedModelFactory(
			s.mockKernelChangefeedModel),

		importCli:      s.mockImportCli,
		coreCli:        s.mockCoreCli,
		workflowCli:    s.mockWorkflow,
		billingHandler: s.mockBilling,

		downstreamAdaptorFactory: NewMockDownstreamAdaptorFactory(s.mockMysqlAdaptor),
		hooker:                   s.mockHooker,
		statusMatrix:             NewBizChangefeedStatusMatrix(),
		clusterJob:               s.mockClusterJob,
	}
}

func (s *bizChangefeedModelTestSuite) TestCheckUpstreamCluster() {
	testcases := []struct {
		cluster *core.Cluster
		err     bool
		errCode bizerr.ErrCode
	}{
		{
			cluster: &core.Cluster{IsDevTier: true},
			err:     true,
			errCode: config.ErrCodeCDCNotallowDevTierErr,
		},
		{
			cluster: &core.Cluster{Status: core.ClusterStatus_CLUSTER_STATUS_INITIALIZING},
			err:     true,
			errCode: config.ErrCodeCDCCreatingErr,
		},
		{
			cluster: &core.Cluster{Status: core.ClusterStatus_CLUSTER_STATUS_PREPARING},
			err:     true,
			errCode: config.ErrCodeCDCCreatingErr,
		},
		{
			cluster: &core.Cluster{Status: core.ClusterStatus_CLUSTER_STATUS_LAUNCHING},
			err:     true,
			errCode: config.ErrCodeCDCCreatingErr,
		},
		{
			cluster: &core.Cluster{Status: core.ClusterStatus_CLUSTER_STATUS_RESUMING},
			err:     true,
			errCode: config.ErrCodeCDCCreatingErr,
		},
		{
			cluster: &core.Cluster{Status: core.ClusterStatus_CLUSTER_STATUS_RECOVERING},
			err:     true,
			errCode: config.ErrCodeCDCRecoveringErr,
		},
		{
			cluster: &core.Cluster{Status: core.ClusterStatus_CLUSTER_STATUS_PAUSING},
			err:     true,
			errCode: config.ErrCodeClusterPausing,
		},
		{
			cluster: &core.Cluster{Status: core.ClusterStatus_CLUSTER_STATUS_PAUSED},
			err:     true,
			errCode: config.ErrCodeClusterPausing,
		},
		{
			cluster: &core.Cluster{Status: core.ClusterStatus_CLUSTER_STATUS_NORMAL},
			err:     false,
		},
		{
			cluster: &core.Cluster{Status: core.ClusterStatus_CLUSTER_STATUS_SCALING},
			err:     false,
		},
	}

	for _, tc := range testcases {
		err := s.model.checkUpstreamCluster(context.Background(), tc.cluster)
		if tc.err {
			s.Error(err)
			s.Equal(tc.errCode, err.RawErrCode())
		} else {
			s.NoError(err)
		}
	}
}

func (s *bizChangefeedModelTestSuite) TestCheckSpecificationCode() {
	code := "code1"
	reqMatcher := mock.MatchedBy(func(req *domain.QuerySpecificationReq) bool {
		return req.Code == code
	})

	s.Run("query error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "record not found")
		call := s.mockMachineManager.
			On("Query", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()

		err := s.model.checkSpecificationCode(context.Background(), code)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.mockMachineManager.AssertCalled(s.T(), "Query", mock.Anything, reqMatcher)
	})

	s.Run("query successfully", func() {
		call := s.mockMachineManager.
			On("Query", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer call.Unset()

		err := s.model.checkSpecificationCode(context.Background(), code)
		s.NoError(err)
		s.mockMachineManager.AssertCalled(s.T(), "Query", mock.Anything, reqMatcher)
	})
}

func (s *bizChangefeedModelTestSuite) TestCheckRunningImports() {
	mockCluster := &core.Cluster{
		Id:        111,
		ProjectId: 11,
		TenantId:  1,
	}
	reqMatcher := mock.MatchedBy(func(req *dataflow.ListImportReq) bool {
		return req.ClusterId == mockCluster.Id &&
			req.ProjectId == mockCluster.ProjectId &&
			req.OrgId == mockCluster.TenantId
	})

	s.Run("list import error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "grpc error")
		call := s.mockImportCli.
			On("List", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()

		err := s.model.checkRunningImports(context.Background(), mockCluster)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.mockImportCli.AssertCalled(s.T(), "List", mock.Anything, reqMatcher)
	})

	s.Run("multiple import tasks", func() {
		mockResp := &dataflow.ListImportResp{
			Imports: []*dataflow.ImportGetResp{{Id: 1}, {Id: 2}},
		}
		call := s.mockImportCli.
			On("List", mock.Anything, mock.Anything).
			Return(mockResp, nil)
		defer call.Unset()

		err := s.model.checkRunningImports(context.Background(), mockCluster)
		s.Error(err)
		s.Equal(config.ErrCodeCDCImportingErr, err.RawErrCode())
		s.mockImportCli.AssertCalled(s.T(), "List", mock.Anything, reqMatcher)
	})

	s.Run("resp is nil", func() {
		call := s.mockImportCli.
			On("List", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer call.Unset()

		err := s.model.checkRunningImports(context.Background(), mockCluster)
		s.NoError(err)
		s.mockImportCli.AssertCalled(s.T(), "List", mock.Anything, reqMatcher)
	})

	s.Run("import tasks is nil", func() {
		mockResp := &dataflow.ListImportResp{
			Imports: nil,
		}
		call := s.mockImportCli.
			On("List", mock.Anything, mock.Anything).
			Return(mockResp, nil)
		defer call.Unset()

		err := s.model.checkRunningImports(context.Background(), mockCluster)
		s.NoError(err)
		s.mockImportCli.AssertCalled(s.T(), "List", mock.Anything, reqMatcher)
	})

	s.Run("import tasks is empty", func() {
		mockResp := &dataflow.ListImportResp{
			Imports: []*dataflow.ImportGetResp{},
		}
		call := s.mockImportCli.
			On("List", mock.Anything, mock.Anything).
			Return(mockResp, nil)
		defer call.Unset()

		err := s.model.checkRunningImports(context.Background(), mockCluster)
		s.NoError(err)
		s.mockImportCli.AssertCalled(s.T(), "List", mock.Anything, reqMatcher)
	})
}

func (s *bizChangefeedModelTestSuite) TestCheckBizChangefeedQuota() {
	mockCluster := &core.Cluster{
		Id:        111,
		ProjectId: 11,
		TenantId:  1,
	}
	changefeedEnts := make([]entity.BizChangefeed, 0, 150)
	for i := 0; i < 150; i++ {
		changefeedEnts = append(changefeedEnts, entity.BizChangefeed{ID: entity.ID(i)})
	}

	s.Run("list changefeeds error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "db error")
		call := s.mockBizChangefeedRepo.
			On("List", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()

		err := s.model.checkBizChangefeedQuota(context.Background(), mockCluster)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
	})

	testcases := []struct {
		name        string
		existingNum int
		err         bool
	}{
		{
			name:        "exceed quota",
			existingNum: 101,
			err:         true,
		},
		{
			name:        "equal to quota",
			existingNum: 100,
			err:         true,
		},
		{
			name:        "not exceed quota",
			existingNum: 99,
			err:         false,
		},
	}
	for _, tc := range testcases {
		s.Run(tc.name, func() {
			call := s.mockBizChangefeedRepo.
				On("List", mock.Anything, mock.Anything).
				Return(changefeedEnts[:tc.existingNum], nil)
			defer call.Unset()

			err := s.model.checkBizChangefeedQuota(context.Background(), mockCluster)
			if tc.err {
				s.Error(err)
				s.Equal(config.ErrCodeChangefeedPerClusterExceeded, err.RawErrCode())
			} else {
				s.NoError(err)
			}
		})
	}
}

func (s *bizChangefeedModelTestSuite) TestCheckDownstream() {
	mockCluster := &core.Cluster{}
	mockReq := &domain.CreateBizChangefeedReq{}
	mockErr := bizerr.New(bizerr.ErrCodeUnknown, "mock error")

	s.Run("test connection failed", func() {
		connCall := s.mockMysqlAdaptor.
			On("TestConnection", mock.Anything, mock.Anything, mockReq.DownstreamParameter).
			Return(mockErr)
		defer connCall.Unset()

		err := s.model.checkDownstream(context.Background(), mockReq, mockCluster, s.mockMysqlAdaptor)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
	})

	s.Run("pre check changefeed params failed", func() {
		connCall := s.mockMysqlAdaptor.
			On("TestConnection", mock.Anything, mock.Anything, mockReq.DownstreamParameter).
			Return(nil)
		defer connCall.Unset()
		tsoCall := s.mockMysqlAdaptor.
			On("PreCheckChangefeedParameters", mock.Anything, mock.Anything, mockReq.DownstreamParameter).
			Return(mockErr)
		defer tsoCall.Unset()

		err := s.model.checkDownstream(context.Background(), mockReq, mockCluster, s.mockMysqlAdaptor)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
	})

	s.Run("check pass", func() {
		connCall := s.mockMysqlAdaptor.
			On("TestConnection", mock.Anything, mock.Anything, mockReq.DownstreamParameter).
			Return(nil)
		defer connCall.Unset()
		tsoCall := s.mockMysqlAdaptor.
			On("PreCheckChangefeedParameters", mock.Anything, mock.Anything, mockReq.DownstreamParameter).
			Return(nil)
		defer tsoCall.Unset()

		err := s.model.checkDownstream(context.Background(), mockReq, mockCluster, s.mockMysqlAdaptor)
		s.NoError(err)
	})
}

func (s *bizChangefeedModelTestSuite) TestGenBizChangefeedEntity() {
	mockCluster := &core.Cluster{}
	mockReq := &domain.CreateBizChangefeedReq{
		Name:              "test1",
		SpecificationCode: "code1",
		UserID:            1,
		DeployType:        entity.TiCDCClusterDeployTypeShared,
	}

	// convert downstream type error
	_, err := s.model.genBizChangefeedEntity(context.Background(), mockReq, mockCluster)
	s.Error(err)
	s.Equal(bizerr.ErrCodeParams, err.RawErrCode())

	mockReq.DownstreamType = dataflow.ConnectorType_MYSQL
	mockReq.DownstreamParameter = &dataflow.MySQLInfo{}
	ent, err := s.model.genBizChangefeedEntity(context.Background(), mockReq, mockCluster)
	s.NoError(err)
	s.Equal(mockReq.Name, ent.Name)
	s.Equal(mockReq.SpecificationCode, ent.SpecificationCode)
	s.Equal(mockReq.UserID, ent.CreatorID.Uint64())
	s.Equal(entity.TiCDCClusterDeployTypeShared, ent.DeployType)
	s.True(ent.IsRegional)
}

func (s *bizChangefeedModelTestSuite) TestCreateBizChangefeedWorkflow() {
	mockEnt := &entity.BizChangefeed{}
	mockDag := "d"
	mockAction := "a"

	s.Run("create workflow error", func() {
		mockErr := errors.New("run error")
		call := s.mockWorkflow.
			On("RunDagWithTags", mockDag, mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()

		_, err := s.model.createBizChangefeedWorkflow(context.Background(), mockEnt, mockDag, mockAction)
		s.Error(err)
		s.Equal(bizerr.ErrCodeUnknown, err.RawErrCode())
		s.mockWorkflow.AssertExpectations(s.T())
	})

	s.Run("create workflow successfully", func() {
		mockDagIns := &workflow_entity.DagInstance{
			BaseInfo: workflow_entity.BaseInfo{ID: "1"},
		}
		call := s.mockWorkflow.
			On("RunDagWithTags", mockDag, mock.Anything, mock.Anything).
			Return(mockDagIns, nil)
		defer call.Unset()

		dagInsID, err := s.model.createBizChangefeedWorkflow(context.Background(), mockEnt, mockDag, mockAction)
		s.NoError(err)
		s.Equal(mockDagIns.ID, dagInsID)
		s.mockWorkflow.AssertExpectations(s.T())
	})
}

func (s *bizChangefeedModelTestSuite) TestCreateBizChangefeed() {
	req := &domain.CreateBizChangefeedReq{
		ClusterID:         111,
		SpecificationCode: "code1",
	}
	s.Run("get downstream adaptor error", func() {
		req.DownstreamType = dataflow.ConnectorType(10)
		_, err := s.model.CreateBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "get downstream adaptor")
	})

	s.Run("get cluster error", func() {
		req.DownstreamType = dataflow.ConnectorType_MYSQL
		mockClusterErr := bizerr.New(bizerr.ErrCodeGRPC, "grpc error")
		call := s.mockCoreCli.
			On("GetAvailableClusterByID", mock.Anything, req.ClusterID).
			Return(nil, mockClusterErr)
		defer call.Unset()
		_, err := s.model.CreateBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockClusterErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "get cluster")
	})

	s.Run("precheck error", func() {
		mockCluster := &core.Cluster{IsDevTier: true}
		call := s.mockCoreCli.
			On("GetAvailableClusterByID", mock.Anything, req.ClusterID).
			Return(mockCluster, nil)
		defer call.Unset()
		_, err := s.model.CreateBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "precheck")
	})

	s.Run("dry run precheck error", func() {
		mockCluster := &core.Cluster{IsDevTier: true}
		call := s.mockCoreCli.
			On("GetAvailableClusterByID", mock.Anything, req.ClusterID).
			Return(mockCluster, nil)
		defer call.Unset()
		req := &domain.CreateBizChangefeedReq{
			ClusterID:           111,
			DownstreamType:      dataflow.ConnectorType_MYSQL,
			DownstreamParameter: &dataflow.MySQLInfo{},
			SpecificationCode:   "code1",
			DryRun:              true,
		}
		_, err := s.model.CreateBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(config.ErrCodeCDCNotallowDevTierErr, err.RawErrCode())
		s.Contains(err.Error(), "check upstream cluster")
	})

	s.Run("dry run successful", func() {
		req := &domain.CreateBizChangefeedReq{
			ClusterID:           111,
			DownstreamType:      dataflow.ConnectorType_MYSQL,
			DownstreamParameter: &dataflow.MySQLInfo{},
			SpecificationCode:   "code1",
			DryRun:              true,
		}
		cluster := &core.Cluster{
			Id:        req.ClusterID,
			ProjectId: 11,
			TenantId:  1,
			Status:    core.ClusterStatus_CLUSTER_STATUS_NORMAL,
		}
		clusterCall := s.mockCoreCli.
			On("GetAvailableClusterByID", mock.Anything, req.ClusterID).
			Return(cluster, nil)
		defer clusterCall.Unset()

		specCall := s.mockMachineManager.
			On("Query", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer specCall.Unset()

		importCall := s.mockImportCli.
			On("List", mock.Anything, mock.Anything).
			Return(&dataflow.ListImportResp{Imports: []*dataflow.ImportGetResp{}}, nil)
		defer importCall.Unset()

		changefeedCall := s.mockBizChangefeedRepo.
			On("List", mock.Anything, mock.Anything).
			Return([]entity.BizChangefeed{}, nil)
		defer changefeedCall.Unset()

		connCall := s.mockMysqlAdaptor.
			On("TestConnection", mock.Anything, mock.Anything, mock.Anything).
			Return(nil)
		defer connCall.Unset()

		preCheckCall := s.mockMysqlAdaptor.
			On("PreCheckChangefeedParameters", mock.Anything, mock.Anything, mock.Anything).
			Return(nil)
		defer preCheckCall.Unset()

		encryptCall := s.mockMysqlAdaptor.
			On("EncryptPasswordInPlace", mock.Anything, mock.Anything).
			Return(nil)
		defer encryptCall.Unset()

		idCall := s.mockMysqlAdaptor.
			On("GetPrivateLinkEndpointID", mock.Anything).
			Return(entity.ID(0))
		defer idCall.Unset()

		resp, err := s.model.CreateBizChangefeed(context.Background(), req)
		s.NoError(err)
		s.NotNil(resp)
		s.mockBizChangefeedRepo.AssertNotCalled(s.T(), "Create", mock.Anything, mock.Anything)
		s.mockWorkflow.AssertNotCalled(s.T(), "RunDagWithTags", mock.Anything, mock.Anything, mock.Anything)
		s.mockClusterJob.AssertNotCalled(s.T(), "RegisterCDCClusterJob", mock.Anything, mock.Anything, mock.Anything)
	})

	// make pre check pass
	mockCluster := &core.Cluster{
		Id:     req.ClusterID,
		Status: core.ClusterStatus_CLUSTER_STATUS_NORMAL,
	}
	clusterCall := s.mockCoreCli.
		On("GetAvailableClusterByID", mock.Anything, mock.Anything).
		Return(mockCluster, nil)
	defer clusterCall.Unset()
	specCall := s.mockMachineManager.
		On("Query", mock.Anything, mock.Anything).
		Return(nil, nil)
	defer specCall.Unset()
	mockResp := &dataflow.ListImportResp{
		Imports: []*dataflow.ImportGetResp{},
	}
	importCall := s.mockImportCli.
		On("List", mock.Anything, mock.Anything).
		Return(mockResp, nil)
	defer importCall.Unset()
	changefeedCall := s.mockBizChangefeedRepo.
		On("List", mock.Anything, mock.Anything).
		Return([]entity.BizChangefeed{}, nil)
	defer changefeedCall.Unset()
	connCall := s.mockMysqlAdaptor.
		On("TestConnection", mock.Anything, mock.Anything, mock.Anything).
		Return(nil)
	defer connCall.Unset()
	tsoCall := s.mockMysqlAdaptor.
		On("PreCheckChangefeedParameters", mock.Anything, mock.Anything, mock.Anything).
		Return(nil)
	defer tsoCall.Unset()

	s.Run("encrypt password error", func() {
		mockEncryptErr := bizerr.New(bizerr.ErrCodeUnknown, "invalid password")
		encryptCall := s.mockMysqlAdaptor.
			On("EncryptPasswordInPlace", mock.Anything, mock.Anything).
			Return(mockEncryptErr)
		defer encryptCall.Unset()
		_, err := s.model.CreateBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockEncryptErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "encrypt password")
	})

	encryptCall := s.mockMysqlAdaptor.
		On("EncryptPasswordInPlace", mock.Anything, mock.Anything).
		Return(nil)
	defer encryptCall.Unset()

	idCall := s.mockMysqlAdaptor.
		On("GetPrivateLinkEndpointID", mock.Anything).
		Return(entity.ID(0))
	defer idCall.Unset()
	s.Run("create changefeed entity error", func() {
		req.DownstreamParameter = &dataflow.MySQLInfo{}
		mockCreateErr := bizerr.New(bizerr.ErrCodeRecordExist, "duplicate record")
		createChangefeedCall := s.mockBizChangefeedRepo.
			On("Create", mock.Anything, mock.Anything).
			Return(mockCreateErr)
		defer createChangefeedCall.Unset()
		_, err := s.model.CreateBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockCreateErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "create biz_changefeed entity")
	})

	mockBizChangefeedID := uint64(1111)
	createChangefeedCall := s.mockBizChangefeedRepo.
		On("Create", mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) {
			mockEnt := args[1].(*entity.BizChangefeed)
			mockEnt.ID = entity.ID(mockBizChangefeedID)
		}).
		Return(nil)
	defer createChangefeedCall.Unset()

	s.Run("create workflow error", func() {
		mockWorkflowErr := errors.New("network error")
		workflowCall := s.mockWorkflow.
			On("RunDagWithTags", mock.Anything, mock.Anything, mock.Anything).
			Return(nil, mockWorkflowErr)
		defer workflowCall.Unset()
		_, err := s.model.CreateBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(bizerr.ErrCodeUnknown, err.RawErrCode())
		s.Contains(err.Error(), "run dag")
	})

	mockDagIns := &workflow_entity.DagInstance{
		BaseInfo: workflow_entity.BaseInfo{
			ID: "dag_ins_1",
		},
	}
	workflowCall := s.mockWorkflow.
		On("RunDagWithTags", mock.Anything, mock.Anything, mock.Anything).
		Return(mockDagIns, nil)
	defer workflowCall.Unset()
	s.Run("register cluster job error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockClusterJob.
			On("RegisterCDCClusterJob", mock.Anything, req.ClusterID,
				cluster_job.GetBizChangefeedTaskID(context.Background(), mockBizChangefeedID)).
			Return(mockErr)
		defer call.Unset()
		_, err := s.model.CreateBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(bizerr.ErrCodeGRPC, err.RawErrCode())
		s.Contains(err.Error(), "register cluster job")
	})

	clusterJobCall := s.mockClusterJob.
		On("RegisterCDCClusterJob", mock.Anything, req.ClusterID,
			cluster_job.GetBizChangefeedTaskID(context.Background(), mockBizChangefeedID)).
		Return(nil)
	defer clusterJobCall.Unset()
	s.Run("create successfully", func() {
		hookCall := s.mockHooker.
			On("HandleBizChangefeedEvent", mock.Anything, entity.CreateChangefeed, req.UserID, mock.Anything).
			Return()
		defer hookCall.Unset()
		resp, err := s.model.CreateBizChangefeed(context.Background(), req)
		s.NoError(err)
		s.NotNil(resp)
		s.Equal(mockDagIns.ID, resp.WorkflowID)
	})
}

func (s *bizChangefeedModelTestSuite) TestCheckBizChangefeedStatus() {
	testcases := []struct {
		currentStatus entity.BizChangefeedOperationStatus
		action        string
		err           bool
	}{
		{
			currentStatus: entity.BizChangefeedOperationStatusCreating,
			action:        bizChangefeedActionScale,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusCreating,
			action:        bizChangefeedActionDelete,
			err:           false,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusCreating,
			action:        bizChangefeedActionPause,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusCreating,
			action:        bizChangefeedActionResume,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusCreating,
			action:        bizChangefeedActionEdit,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusNormal,
			action:        bizChangefeedActionScale,
			err:           false,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusNormal,
			action:        bizChangefeedActionDelete,
			err:           false,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusNormal,
			action:        bizChangefeedActionPause,
			err:           false,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusNormal,
			action:        bizChangefeedActionResume,
			err:           false,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusNormal,
			action:        bizChangefeedActionEdit,
			err:           false,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusScaling,
			action:        bizChangefeedActionScale,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusScaling,
			action:        bizChangefeedActionDelete,
			err:           false,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusScaling,
			action:        bizChangefeedActionPause,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusScaling,
			action:        bizChangefeedActionResume,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusScaling,
			action:        bizChangefeedActionEdit,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusDeleting,
			action:        bizChangefeedActionScale,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusDeleting,
			action:        bizChangefeedActionDelete,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusDeleting,
			action:        bizChangefeedActionPause,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusDeleting,
			action:        bizChangefeedActionResume,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusDeleting,
			action:        bizChangefeedActionEdit,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusDeleted,
			action:        bizChangefeedActionScale,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusDeleted,
			action:        bizChangefeedActionDelete,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusDeleted,
			action:        bizChangefeedActionPause,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusDeleted,
			action:        bizChangefeedActionResume,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusDeleted,
			action:        bizChangefeedActionEdit,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusPaused,
			action:        bizChangefeedActionScale,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusPaused,
			action:        bizChangefeedActionPause,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusPaused,
			action:        bizChangefeedActionResume,
			err:           false,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusPaused,
			action:        bizChangefeedActionEdit,
			err:           false,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusPaused,
			action:        bizChangefeedActionDelete,
			err:           false,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusFailed,
			action:        bizChangefeedActionScale,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusFailed,
			action:        bizChangefeedActionPause,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusFailed,
			action:        bizChangefeedActionResume,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusFailed,
			action:        bizChangefeedActionEdit,
			err:           true,
		},
		{
			currentStatus: entity.BizChangefeedOperationStatusFailed,
			action:        bizChangefeedActionDelete,
			err:           false,
		},
	}

	ctx := context.Background()
	for _, tc := range testcases {
		err := s.model.checkBizChangefeedStatus(ctx, tc.currentStatus, tc.action)
		if tc.err {
			s.Error(err)
		} else {
			s.NoError(err)
		}
	}
}

func (s *bizChangefeedModelTestSuite) TestPreCheckScaleBizChangefeed() {
	ent := &entity.BizChangefeed{
		ID:                entity.ID(123),
		SpecificationCode: "code1",
	}
	req := &domain.ScaleBizChangefeedReq{
		BizChangefeedID: ent.ID.Uint64(),
	}

	s.Run("specification codes are equal", func() {
		req.SpecificationCode = ent.SpecificationCode
		err := s.model.preCheckScaleBizChangefeed(context.Background(), req, ent)
		s.Error(err)
		s.Contains(err.Error(), "specification code is not changed")
	})

	s.Run("check status error", func() {
		req.SpecificationCode = "code2"
		ent.OperationStatus = entity.BizChangefeedOperationStatusCreating
		err := s.model.preCheckScaleBizChangefeed(context.Background(), req, ent)
		s.Error(err)
		s.Contains(err.Error(), "check biz_changefeed status")
	})

	s.Run("check specification code error", func() {
		ent.OperationStatus = entity.BizChangefeedOperationStatusNormal
		mockSpecErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		specCall := s.mockMachineManager.
			On("Query", mock.Anything, mock.Anything).
			Return(nil, mockSpecErr)
		defer specCall.Unset()

		err := s.model.preCheckScaleBizChangefeed(context.Background(), req, ent)
		s.Error(err)
		s.Equal(mockSpecErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "check specification code")
	})

	s.Run("check passed", func() {
		specCall := s.mockMachineManager.
			On("Query", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer specCall.Unset()

		err := s.model.preCheckScaleBizChangefeed(context.Background(), req, ent)
		s.NoError(err)
	})
}

func (s *bizChangefeedModelTestSuite) TestUpdateBizChangefeedEntityForScale() {
	ent := &entity.BizChangefeed{
		ID:                entity.ID(123),
		SpecificationCode: "code1",
	}
	req := &domain.ScaleBizChangefeedReq{
		BizChangefeedID:   ent.ID.Uint64(),
		SpecificationCode: "code2",
	}

	s.Run("calculate next status error", func() {
		ent.OperationStatus = entity.BizChangefeedOperationStatusCreating
		err := s.model.updateBizChangefeedEntityForScale(context.Background(), req, ent)
		s.Error(err)
		s.Contains(err.Error(), "calculate next status")
	})

	s.Run("update entity error", func() {
		ent.OperationStatus = entity.BizChangefeedOperationStatusNormal
		mockUpdateErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		updateCall := s.mockBizChangefeedRepo.
			On("Update", mock.Anything, mock.Anything).
			Return(mockUpdateErr)
		defer updateCall.Unset()
		err := s.model.updateBizChangefeedEntityForScale(context.Background(), req, ent)
		s.Error(err)
		s.Contains(err.Error(), "update biz_changefeed entity")
	})

	entMatcher := func(updatedEnt *entity.BizChangefeed) bool {
		return updatedEnt.SpecificationCode == req.SpecificationCode &&
			updatedEnt.OperationStatus == entity.BizChangefeedOperationStatusScaling
	}
	s.Run("update successfully", func() {
		ent.OperationStatus = entity.BizChangefeedOperationStatusNormal
		updateCall := s.mockBizChangefeedRepo.
			On("Update", mock.Anything, mock.Anything).
			Return(nil)
		defer updateCall.Unset()
		err := s.model.updateBizChangefeedEntityForScale(context.Background(), req, ent)
		s.NoError(err)
		s.mockBizChangefeedRepo.AssertCalled(s.T(), "Update", mock.Anything, mock.MatchedBy(entMatcher))
	})
}

func (s *bizChangefeedModelTestSuite) TestScaleBizChangefeed() {
	mockEnt := &entity.BizChangefeed{
		ID:                entity.ID(123),
		SpecificationCode: "code1",
		OperationStatus:   entity.BizChangefeedOperationStatusNormal,
	}
	req := &domain.ScaleBizChangefeedReq{
		BizChangefeedID: mockEnt.ID.Uint64(),
	}

	s.Run("load entity error", func() {
		mockLoadErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		loadCall := s.mockBizChangefeedRepo.
			On("Load", mock.Anything, req.BizChangefeedID).
			Return(nil, mockLoadErr)
		defer loadCall.Unset()
		_, err := s.model.ScaleBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockLoadErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "load biz_changefeed")
		s.mockBizChangefeedRepo.AssertExpectations(s.T())
	})

	loadCall := s.mockBizChangefeedRepo.
		On("Load", mock.Anything, req.BizChangefeedID).
		Return(mockEnt, nil)
	defer loadCall.Unset()
	s.Run("pre check error", func() {
		req.SpecificationCode = mockEnt.SpecificationCode
		_, err := s.model.ScaleBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "precheck scale biz_changefeed")
	})

	specCall := s.mockMachineManager.
		On("Query", mock.Anything, mock.Anything).
		Return(nil, nil)
	defer specCall.Unset()
	s.Run("update entity error", func() {
		req.SpecificationCode = "code2"
		mockUpdateErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		updateCall := s.mockBizChangefeedRepo.
			On("Update", mock.Anything, mock.Anything).
			Return(mockUpdateErr)
		defer updateCall.Unset()
		_, err := s.model.ScaleBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockUpdateErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "update biz_changefeed entity")
	})

	updateCall := s.mockBizChangefeedRepo.
		On("Update", mock.Anything, mock.Anything).
		Return(nil)
	defer updateCall.Unset()
	s.Run("create workflow error", func() {
		mockEnt.OperationStatus = entity.BizChangefeedOperationStatusNormal
		mockEnt.SpecificationCode = "code1"
		mockWorkflowErr := errors.New("network error")
		workflowCall := s.mockWorkflow.
			On("RunDagWithTags", mock.Anything, mock.Anything, mock.Anything).
			Return(nil, mockWorkflowErr)
		defer workflowCall.Unset()
		_, err := s.model.ScaleBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "create biz_changefeed workflow")
	})

	s.Run("scale successfully", func() {
		mockEnt.OperationStatus = entity.BizChangefeedOperationStatusNormal
		mockEnt.SpecificationCode = "code1"
		mockDagIns := &workflow_entity.DagInstance{
			BaseInfo: workflow_entity.BaseInfo{
				ID: "dag_ins_1",
			},
		}
		workflowCall := s.mockWorkflow.
			On("RunDagWithTags", mock.Anything, mock.Anything, mock.Anything).
			Return(mockDagIns, nil)
		defer workflowCall.Unset()
		hookCall := s.mockHooker.
			On("HandleBizChangefeedEvent", mock.Anything, entity.ScaleChangefeed, req.UserID, mock.Anything).
			Return()
		defer hookCall.Unset()

		resp, err := s.model.ScaleBizChangefeed(context.Background(), req)
		s.NoError(err)
		s.NotNil(resp)
		s.Equal(mockDagIns.ID, resp.WorkflowID)
	})
}

func (s *bizChangefeedModelTestSuite) TestPreCheckDeleteBizChangefeed() {
	mockEnt := &entity.BizChangefeed{}
	s.Run("pre check failed", func() {
		mockEnt.OperationStatus = entity.BizChangefeedOperationStatusDeleting
		err := s.model.preCheckDeleteBizChangefeed(context.Background(), mockEnt)
		s.Error(err)
	})

	s.Run("pre check successfully", func() {
		mockEnt.OperationStatus = entity.BizChangefeedOperationStatusNormal
		err := s.model.preCheckDeleteBizChangefeed(context.Background(), mockEnt)
		s.NoError(err)
	})
}

func (s *bizChangefeedModelTestSuite) TestUpdateBizChangefeedEntityForDelete() {
	mockEnt := &entity.BizChangefeed{}

	s.Run("calculate next status error", func() {
		mockEnt.OperationStatus = entity.BizChangefeedOperationStatusDeleted
		err := s.model.updateBizChangefeedEntityForDelete(context.Background(), mockEnt)
		s.Error(err)
		s.Contains(err.Error(), "calculate next status")
	})

	s.Run("update entity error", func() {
		mockEnt.OperationStatus = entity.BizChangefeedOperationStatusNormal
		mockUpdateErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		updateCall := s.mockBizChangefeedRepo.
			On("Update", mock.Anything, mock.Anything).
			Return(mockUpdateErr)
		defer updateCall.Unset()
		err := s.model.updateBizChangefeedEntityForDelete(context.Background(), mockEnt)
		s.Error(err)
		s.Equal(mockUpdateErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "update biz_changefeed entity")
	})

	entMatcher := func(updatedEnt *entity.BizChangefeed) bool {
		return updatedEnt.OperationStatus == entity.BizChangefeedOperationStatusDeleting
	}
	s.Run("update entity successfully", func() {
		mockEnt.OperationStatus = entity.BizChangefeedOperationStatusNormal
		updateCall := s.mockBizChangefeedRepo.
			On("Update", mock.Anything, mock.Anything).
			Return(nil)
		defer updateCall.Unset()
		err := s.model.updateBizChangefeedEntityForDelete(context.Background(), mockEnt)
		s.NoError(err)
		s.mockBizChangefeedRepo.AssertCalled(s.T(), "Update", mock.Anything, mock.MatchedBy(entMatcher))
	})
}

func (s *bizChangefeedModelTestSuite) TestDeleteBizChangefeed() {
	mockEnt := &entity.BizChangefeed{
		ID: entity.ID(123),
	}
	req := &domain.DeleteBizChangefeedReq{
		BizChangefeedID: mockEnt.ID.Uint64(),
	}

	s.Run("load entity error", func() {
		mockLoadErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		loadCall := s.mockBizChangefeedRepo.
			On("Load", mock.Anything, req.BizChangefeedID).
			Return(nil, mockLoadErr)
		defer loadCall.Unset()
		_, err := s.model.DeleteBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockLoadErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "load biz_changefeed")
		s.mockBizChangefeedRepo.AssertExpectations(s.T())
	})

	loadCall := s.mockBizChangefeedRepo.
		On("Load", mock.Anything, req.BizChangefeedID).
		Return(mockEnt, nil)
	defer loadCall.Unset()
	s.Run("pre check error", func() {
		mockEnt.OperationStatus = entity.BizChangefeedOperationStatusDeleted
		_, err := s.model.DeleteBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "precheck delete biz_changefeed")
	})

	s.Run("update entity error", func() {
		mockEnt.OperationStatus = entity.BizChangefeedOperationStatusNormal
		mockUpdateErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		updateCall := s.mockBizChangefeedRepo.
			On("Update", mock.Anything, mock.Anything).
			Return(mockUpdateErr)
		defer updateCall.Unset()
		_, err := s.model.DeleteBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockUpdateErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "update biz_changefeed entity")
	})

	updateCall := s.mockBizChangefeedRepo.
		On("Update", mock.Anything, mock.Anything).
		Return(nil)
	defer updateCall.Unset()
	s.Run("create workflow error", func() {
		mockEnt.OperationStatus = entity.BizChangefeedOperationStatusNormal
		mockWorkflowErr := errors.New("network error")
		workflowCall := s.mockWorkflow.
			On("RunDagWithTags", mock.Anything, mock.Anything, mock.Anything).
			Return(nil, mockWorkflowErr)
		defer workflowCall.Unset()
		_, err := s.model.DeleteBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "create biz_changefeed workflow")
	})

	s.Run("delete successfully", func() {
		mockEnt.OperationStatus = entity.BizChangefeedOperationStatusNormal
		mockDagIns := &workflow_entity.DagInstance{
			BaseInfo: workflow_entity.BaseInfo{
				ID: "dag_ins_1",
			},
		}
		workflowCall := s.mockWorkflow.
			On("RunDagWithTags", mock.Anything, mock.Anything, mock.Anything).
			Return(mockDagIns, nil)
		defer workflowCall.Unset()
		hookCall := s.mockHooker.
			On("HandleBizChangefeedEvent", mock.Anything, entity.DeleteChangefeed, req.UserID, mock.Anything).
			Return()
		defer hookCall.Unset()

		resp, err := s.model.DeleteBizChangefeed(context.Background(), req)
		s.NoError(err)
		s.NotNil(resp)
		s.Equal(mockDagIns.ID, resp.WorkflowID)
	})
}

func (s *bizChangefeedModelTestSuite) TestQueryKernelChangefeed() {
	bizType := domain.TiCDCClusterBizTypeChangefeed
	bizChangefeedID := uint64(111)

	matcher := func(queryReq *domain.QueryChangefeedReq) bool {
		return *queryReq.Filter.BizType == bizType && queryReq.Filter.BusinessIDs[0] == bizChangefeedID
	}

	s.Run("query kernel changefeed error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockKernelChangefeedModel.
			On("QueryChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.queryKernelChangefeed(context.Background(), bizChangefeedID)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.mockKernelChangefeedModel.AssertCalled(
			s.T(), "QueryChangefeed", mock.Anything, mock.MatchedBy(matcher))
	})

	s.Run("query kernel changefeed successfully", func() {
		mockResp := &domain.QueryChangefeedResp{
			ChangefeedEntity: entity.TiCDCChangefeed{
				Updatable: entity.Updatable{
					Base: entity.Base{
						ID: 111,
					},
				},
			},
		}
		call := s.mockKernelChangefeedModel.
			On("QueryChangefeed", mock.Anything, mock.Anything).
			Return(mockResp, nil)
		defer call.Unset()
		kernelChangefeed, err := s.model.queryKernelChangefeed(context.Background(), bizChangefeedID)
		s.NoError(err)
		s.Equal(&mockResp.ChangefeedEntity, kernelChangefeed)
	})
}

func (s *bizChangefeedModelTestSuite) TestPauseKernelChangefeed() {
	mockEnt := &entity.BizChangefeed{
		ID: entity.ID(111),
	}
	s.Run("query kernel changefeed error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockKernelChangefeedModel.
			On("QueryChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.pauseKernelChangefeed(context.Background(), mockEnt)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "query kernel changefeed")
	})

	mockResp := &domain.QueryChangefeedResp{
		ChangefeedEntity: entity.TiCDCChangefeed{},
	}
	mockResp.ChangefeedEntity.ID = entity.ID(1111)
	call := s.mockKernelChangefeedModel.
		On("QueryChangefeed", mock.Anything, mock.Anything).
		Return(mockResp, nil)
	defer call.Unset()
	s.Run("kernel changefeed already stopped", func() {
		mockResp.ChangefeedEntity.KernelMetrics.FeedState = entity.ChangefeedKernelStatusStopped
		err := s.model.pauseKernelChangefeed(context.Background(), mockEnt)
		s.NoError(err)
	})

	s.Run("kernel changefeed failed", func() {
		mockResp.ChangefeedEntity.KernelMetrics.FeedState = entity.ChangefeedKernelStatusFailed
		call := s.mockKernelChangefeedModel.
			On("PauseChangefeed", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer call.Unset()
		err := s.model.pauseKernelChangefeed(context.Background(), mockEnt)
		s.NoError(err)
	})

	mockResp.ChangefeedEntity.KernelMetrics.FeedState = entity.ChangefeedKernelStatusNormal
	matcher := func(req *domain.PauseChangefeedReq) bool {
		return req.ChangefeedID == mockResp.ChangefeedEntity.ID.Uint64()
	}
	s.Run("pause kernel changefeed error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockKernelChangefeedModel.
			On("PauseChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.pauseKernelChangefeed(context.Background(), mockEnt)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "pause changefeed")
		s.mockKernelChangefeedModel.AssertCalled(
			s.T(), "PauseChangefeed", mock.Anything, mock.MatchedBy(matcher))
	})

	s.Run("pause kernel changefeed successfully", func() {
		call := s.mockKernelChangefeedModel.
			On("PauseChangefeed", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer call.Unset()
		err := s.model.pauseKernelChangefeed(context.Background(), mockEnt)
		s.NoError(err)
		s.mockKernelChangefeedModel.AssertCalled(
			s.T(), "PauseChangefeed", mock.Anything, mock.MatchedBy(matcher))
	})
}

func (s *bizChangefeedModelTestSuite) TestPauseBizChangefeed() {
	req := &domain.PauseBizChangefeedReq{
		BizChangefeedID: 111,
		UserID:          11,
	}
	s.Run("load entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockBizChangefeedRepo.
			On("Load", mock.Anything, req.BizChangefeedID).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.PauseBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "load biz_changefeed")
	})

	mockEntity := &entity.BizChangefeed{
		ID: 111,
	}
	loadCall := s.mockBizChangefeedRepo.
		On("Load", mock.Anything, req.BizChangefeedID).
		Return(mockEntity, nil)
	defer loadCall.Unset()
	s.Run("pre check error", func() {
		mockEntity.OperationStatus = entity.BizChangefeedOperationStatusScaling
		err := s.model.PauseBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(bizerr.ErrCodeParams, err.RawErrCode())
		s.Contains(err.Error(), "precheck pause biz_changefeed")
	})

	s.Run("update entity error", func() {
		mockEntity.OperationStatus = entity.BizChangefeedOperationStatusNormal
		mockErr := bizerr.New(bizerr.ErrCodeDB, "version conflict")
		call := s.mockBizChangefeedRepo.On("Update", mock.Anything, mock.Anything).Return(mockErr)
		defer call.Unset()
		err := s.model.PauseBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "update biz_changefeed entity")
		s.Equal(entity.BizChangefeedOperationStatusPaused, mockEntity.OperationStatus)
	})

	updateCall := s.mockBizChangefeedRepo.On("Update", mock.Anything, mock.Anything).Return(nil)
	defer updateCall.Unset()
	mockResp := &domain.QueryChangefeedResp{
		ChangefeedEntity: entity.TiCDCChangefeed{},
	}
	queryCall := s.mockKernelChangefeedModel.
		On("QueryChangefeed", mock.Anything, mock.Anything).
		Return(mockResp, nil)
	defer queryCall.Unset()
	s.Run("pause kernel changefeed error", func() {
		mockEntity.OperationStatus = entity.BizChangefeedOperationStatusNormal
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockKernelChangefeedModel.
			On("PauseChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.PauseBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "pause kernel changefeed")
	})

	s.Run("pause successfully", func() {
		mockEntity.OperationStatus = entity.BizChangefeedOperationStatusNormal
		pauseCall := s.mockKernelChangefeedModel.
			On("PauseChangefeed", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer pauseCall.Unset()
		hookCall := s.mockHooker.
			On("HandleBizChangefeedEvent", mock.Anything, entity.PauseChangefeed, req.UserID, mock.Anything).
			Return()
		defer hookCall.Unset()
		err := s.model.PauseBizChangefeed(context.Background(), req)
		s.NoError(err)
	})
}

func (s *bizChangefeedModelTestSuite) TestResumeKernelChangefeed() {
	mockEnt := &entity.BizChangefeed{
		ID: entity.ID(111),
	}
	s.Run("query kernel changefeed error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockKernelChangefeedModel.
			On("QueryChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.resumeKernelChangefeed(context.Background(), mockEnt)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "query kernel changefeed")
	})

	mockResp := &domain.QueryChangefeedResp{
		ChangefeedEntity: entity.TiCDCChangefeed{},
	}
	mockResp.ChangefeedEntity.ID = entity.ID(1111)
	call := s.mockKernelChangefeedModel.
		On("QueryChangefeed", mock.Anything, mock.Anything).
		Return(mockResp, nil)
	defer call.Unset()
	s.Run("kernel changefeed is normal", func() {
		mockResp.ChangefeedEntity.KernelMetrics.FeedState = entity.ChangefeedKernelStatusNormal
		err := s.model.resumeKernelChangefeed(context.Background(), mockEnt)
		s.NoError(err)
	})

	mockResp.ChangefeedEntity.KernelMetrics.FeedState = entity.ChangefeedKernelStatusStopped
	matcher := func(req *domain.ResumeChangefeedReq) bool {
		return req.ChangefeedID == mockResp.ChangefeedEntity.ID.Uint64()
	}
	s.Run("resume kernel changefeed error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockKernelChangefeedModel.
			On("ResumeChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.resumeKernelChangefeed(context.Background(), mockEnt)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "resume changefeed")
		s.mockKernelChangefeedModel.AssertCalled(
			s.T(), "ResumeChangefeed", mock.Anything, mock.MatchedBy(matcher))
	})

	resumeCall := s.mockKernelChangefeedModel.
		On("ResumeChangefeed", mock.Anything, mock.Anything).
		Return(nil, nil).Twice()
	defer resumeCall.Unset()
	s.Run("resume kernel changefeed successfully", func() {
		err := s.model.resumeKernelChangefeed(context.Background(), mockEnt)
		s.NoError(err)
		s.mockKernelChangefeedModel.AssertCalled(
			s.T(), "ResumeChangefeed", mock.Anything, mock.MatchedBy(matcher))
	})

	s.Run("resume failed kernel changefeed successfully", func() {
		mockResp.ChangefeedEntity.KernelMetrics.FeedState = entity.ChangefeedKernelStatusFailed
		err := s.model.resumeKernelChangefeed(context.Background(), mockEnt)
		s.NoError(err)
		s.mockKernelChangefeedModel.AssertCalled(
			s.T(), "ResumeChangefeed", mock.Anything, mock.MatchedBy(matcher))
	})
}

func (s *bizChangefeedModelTestSuite) TestResumeBizChangefeed() {
	req := &domain.ResumeBizChangefeedReq{
		BizChangefeedID: 111,
		UserID:          11,
	}
	s.Run("load entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockBizChangefeedRepo.
			On("Load", mock.Anything, req.BizChangefeedID).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.ResumeBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "load biz_changefeed")
	})

	mockEntity := &entity.BizChangefeed{
		ID: 111,
	}
	loadCall := s.mockBizChangefeedRepo.
		On("Load", mock.Anything, req.BizChangefeedID).
		Return(mockEntity, nil)
	defer loadCall.Unset()
	s.Run("pre check error", func() {
		mockEntity.OperationStatus = entity.BizChangefeedOperationStatusScaling
		err := s.model.ResumeBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(bizerr.ErrCodeParams, err.RawErrCode())
		s.Contains(err.Error(), "precheck resume biz_changefeed")
	})

	s.Run("update entity error", func() {
		mockEntity.OperationStatus = entity.BizChangefeedOperationStatusPaused
		mockErr := bizerr.New(bizerr.ErrCodeDB, "version conflict")
		call := s.mockBizChangefeedRepo.On("Update", mock.Anything, mock.Anything).Return(mockErr)
		defer call.Unset()
		err := s.model.ResumeBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "update biz_changefeed entity")
		s.Equal(entity.BizChangefeedOperationStatusNormal, mockEntity.OperationStatus)
	})

	updateCall := s.mockBizChangefeedRepo.On("Update", mock.Anything, mock.Anything).Return(nil)
	defer updateCall.Unset()
	mockResp := &domain.QueryChangefeedResp{
		ChangefeedEntity: entity.TiCDCChangefeed{},
	}
	queryCall := s.mockKernelChangefeedModel.
		On("QueryChangefeed", mock.Anything, mock.Anything).
		Return(mockResp, nil)
	defer queryCall.Unset()
	s.Run("resume kernel changefeed error", func() {
		mockResp.ChangefeedEntity.KernelMetrics.FeedState = entity.ChangefeedKernelStatusStopped
		mockEntity.OperationStatus = entity.BizChangefeedOperationStatusPaused
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockKernelChangefeedModel.
			On("ResumeChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.ResumeBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "resume kernel changefeed")
	})

	s.Run("resume successfully", func() {
		mockEntity.OperationStatus = entity.BizChangefeedOperationStatusPaused
		resumeCall := s.mockKernelChangefeedModel.
			On("ResumeChangefeed", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer resumeCall.Unset()
		hookCall := s.mockHooker.
			On("HandleBizChangefeedEvent", mock.Anything, entity.ResumeChangefeed, req.UserID, mock.Anything).
			Return()
		defer hookCall.Unset()
		err := s.model.ResumeBizChangefeed(context.Background(), req)
		s.NoError(err)
	})
}

func (s *bizChangefeedModelTestSuite) TestPreCheckEditBizChangefeed() {
	req := &domain.EditBizChangefeedReq{
		BizChangefeedID: 111,
		UserID:          11,
	}
	ent := &entity.BizChangefeed{
		ID: 111,
	}
	kernelChangefeed := &entity.TiCDCChangefeed{}
	kernelChangefeed.ID = entity.ID(1111)

	s.Run("kernel changefeed is not stopped", func() {
		kernelChangefeed.KernelMetrics.FeedState = entity.ChangefeedKernelStatusNormal
		err := s.model.preCheckEditBizChangefeed(context.Background(), req, ent, kernelChangefeed)
		s.Error(err)
		s.Equal(bizerr.ErrCodeParams, err.RawErrCode())
		s.Contains(err.Error(), "kernel changefeed is not stopped")
	})

	kernelChangefeed.KernelMetrics.FeedState = entity.ChangefeedKernelStatusStopped
	s.Run("invalid downstream type", func() {
		req.DownstreamType = dataflow.ConnectorType_TOTAL
		err := s.model.preCheckEditBizChangefeed(context.Background(), req, ent, kernelChangefeed)
		s.Error(err)
		s.Equal(bizerr.ErrCodeParams, err.RawErrCode())
		s.Contains(err.Error(), "convert downstream type")
	})

	s.Run("downstream type is not equal", func() {
		req.DownstreamType = dataflow.ConnectorType_MYSQL
		ent.DownstreamType = "kafka"
		err := s.model.preCheckEditBizChangefeed(context.Background(), req, ent, kernelChangefeed)
		s.Error(err)
		s.Equal(bizerr.ErrCodeParams, err.RawErrCode())
		s.Contains(err.Error(), "can't edit downstream type")
	})

	ent.DownstreamType = entity.TiCDCDownStreamTypeMySQL
	s.Run("check status error", func() {
		ent.OperationStatus = entity.BizChangefeedOperationStatusScaling
		err := s.model.preCheckEditBizChangefeed(context.Background(), req, ent, kernelChangefeed)
		s.Error(err)
		s.Equal(bizerr.ErrCodeParams, err.RawErrCode())
		s.Contains(err.Error(), "check biz_changefeed status")
	})

	ent.OperationStatus = entity.BizChangefeedOperationStatusPaused
	s.Run("pre check successful", func() {
		err := s.model.preCheckEditBizChangefeed(context.Background(), req, ent, kernelChangefeed)
		s.NoError(err)
	})

	s.Run("failed kernel changefeed can be edited", func() {
		kernelChangefeed.KernelMetrics.FeedState = entity.ChangefeedKernelStatusFailed
		ent.OperationStatus = entity.BizChangefeedOperationStatusNormal
		err := s.model.preCheckEditBizChangefeed(context.Background(), req, ent, kernelChangefeed)
		s.NoError(err)
	})
}

func (s *bizChangefeedModelTestSuite) TestUpdateBizChangefeedEntityForEdit() {
	req := &domain.EditBizChangefeedReq{
		BizChangefeedID: 111,
		Name:            "abc",
		DownstreamParameter: &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{Endpoint: "a"},
		},
	}
	downstreamParamsBytes, err := json.Marshal(req.DownstreamParameter)
	s.NoError(err)
	downstreamParamsStr := string(downstreamParamsBytes)
	ent := &entity.BizChangefeed{
		ID: 111,
	}
	matcher := func(updateEnt *entity.BizChangefeed) bool {
		return updateEnt.Name == req.Name && updateEnt.DownstreamParams == downstreamParamsStr
	}

	s.Run("update entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockBizChangefeedRepo.On("Update", mock.Anything, mock.Anything).Return(mockErr)
		defer call.Unset()
		err := s.model.updateBizChangefeedEntityForEdit(context.Background(), req, ent)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "update biz_changefeed entity")
		s.mockBizChangefeedRepo.AssertCalled(s.T(), "Update", mock.Anything, mock.MatchedBy(matcher))
	})

	s.Run("update entity successfully", func() {
		call := s.mockBizChangefeedRepo.On("Update", mock.Anything, mock.Anything).Return(nil)
		defer call.Unset()
		err := s.model.updateBizChangefeedEntityForEdit(context.Background(), req, ent)
		s.NoError(err)
		s.mockBizChangefeedRepo.AssertCalled(s.T(), "Update", mock.Anything, mock.MatchedBy(matcher))
	})
}

func (s *bizChangefeedModelTestSuite) TestEditKernelChangefeed() {
	mockEnt := &entity.BizChangefeed{
		ID: entity.ID(111),
	}
	mockCluster := &core.Cluster{}
	mockKernelChangefeed := &entity.TiCDCChangefeed{}
	mockKernelChangefeed.ID = entity.ID(1111)
	s.Run("assemble downstream params error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "assemble failed")
		call := s.mockMysqlAdaptor.
			On("AssembleChangefeedConfig", mock.Anything, mockCluster, mockEnt).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.editKernelChangefeed(
			context.Background(), mockCluster, s.mockMysqlAdaptor, mockEnt, mockKernelChangefeed)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "assemble kernel changefeed config")
	})

	mockConfig := &domain.TiCDCChangefeedConfig{
		URIConfig: &domain.TiCDCChangefeedMySQLURIConfigData{Endpoint: pointer.ToString("a")},
		Config: &domain.TiCDCChangefeedConfigData{
			ReplicaConfig: &clusterproxy.ReplicaConfig{
				Filter: &clusterproxy.FilterConfig{IgnoreTxnStartTs: []uint64{123, 456}},
			},
		},
	}
	assembleCall := s.mockMysqlAdaptor.
		On("AssembleChangefeedConfig", mock.Anything, mockCluster, mockEnt).
		Return(mockConfig, nil)
	defer assembleCall.Unset()
	matcher := func(modifyReq *domain.ModifyChangefeedConfigReq) bool {
		return modifyReq.ChangefeedID == mockKernelChangefeed.ID.Uint64() &&
			reflect.DeepEqual(mockConfig, modifyReq.Config)
	}
	s.Run("modify kernel config error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockKernelChangefeedModel.
			On("ModifyChangefeedConfig", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.editKernelChangefeed(
			context.Background(), mockCluster, s.mockMysqlAdaptor, mockEnt, mockKernelChangefeed)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "edit kernel changefeed")
		s.mockKernelChangefeedModel.AssertCalled(
			s.T(), "ModifyChangefeedConfig", mock.Anything, mock.MatchedBy(matcher))
	})

	s.Run("modify kernel config successfully", func() {
		call := s.mockKernelChangefeedModel.
			On("ModifyChangefeedConfig", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer call.Unset()
		err := s.model.editKernelChangefeed(
			context.Background(), mockCluster, s.mockMysqlAdaptor, mockEnt, mockKernelChangefeed)
		s.NoError(err)
		s.mockKernelChangefeedModel.AssertCalled(
			s.T(), "ModifyChangefeedConfig", mock.Anything, mock.MatchedBy(matcher))
	})
}

func (s *bizChangefeedModelTestSuite) TestEditBizChangefeed() {
	req := &domain.EditBizChangefeedReq{
		BizChangefeedID:     111,
		Name:                "a",
		DownstreamParameter: &dataflow.MySQLInfo{},
	}
	s.Run("load entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockBizChangefeedRepo.
			On("Load", mock.Anything, req.BizChangefeedID).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.EditBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "load biz_changefeed")
	})

	ent := &entity.BizChangefeed{
		ID:              entity.ID(req.BizChangefeedID),
		DownstreamType:  entity.TiCDCDownStreamTypeMySQL,
		OperationStatus: entity.BizChangefeedOperationStatusPaused,
		ClusterID:       111,
	}
	loadCall := s.mockBizChangefeedRepo.
		On("Load", mock.Anything, req.BizChangefeedID).
		Return(ent, nil)
	defer loadCall.Unset()

	s.Run("query kernel changefeed error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockKernelChangefeedModel.
			On("QueryChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.EditBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "query kernel changefeed")
	})

	mockResp := &domain.QueryChangefeedResp{
		ChangefeedEntity: entity.TiCDCChangefeed{},
	}
	mockResp.ChangefeedEntity.ID = entity.ID(1111)
	mockResp.ChangefeedEntity.KernelMetrics.FeedState = entity.ChangefeedKernelStatusStopped
	queryCall := s.mockKernelChangefeedModel.
		On("QueryChangefeed", mock.Anything, mock.Anything).
		Return(mockResp, nil)
	defer queryCall.Unset()
	s.Run("pre check error", func() {
		req.DownstreamType = dataflow.ConnectorType_TOTAL
		err := s.model.EditBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(bizerr.ErrCodeParams, err.RawErrCode())
		s.Contains(err.Error(), "precheck edit biz_changefeed")
	})

	s.Run("dry run precheck error", func() {
		dryReq := &domain.EditBizChangefeedReq{
			BizChangefeedID:     111,
			DownstreamType:      dataflow.ConnectorType_TOTAL,
			DownstreamParameter: &dataflow.MySQLInfo{},
			DryRun:              true,
		}
		err := s.model.EditBizChangefeed(context.Background(), dryReq)
		s.Error(err)
		s.Contains(err.Error(), "precheck edit biz_changefeed")
	})

	s.Run("dry run successful", func() {
		dryReq := &domain.EditBizChangefeedReq{
			BizChangefeedID:     111,
			Name:                "updated-name",
			DownstreamType:      dataflow.ConnectorType_MYSQL,
			DownstreamParameter: &dataflow.MySQLInfo{},
			DryRun:              true,
		}
		clusterCall := s.mockCoreCli.
			On("GetAvailableClusterByID", mock.Anything, ent.ClusterID.Uint64()).
			Return(&core.Cluster{}, nil)
		defer clusterCall.Unset()
		completeCall := s.mockMysqlAdaptor.
			On("CompleteEditParams", mock.Anything, mock.Anything, mock.Anything).
			Return(nil)
		defer completeCall.Unset()
		err := s.model.EditBizChangefeed(context.Background(), dryReq)
		s.NoError(err)
	})

	req.DownstreamType = dataflow.ConnectorType_MYSQL
	s.Run("get cluster error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockCoreCli.
			On("GetAvailableClusterByID", mock.Anything, ent.ClusterID.Uint64()).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.EditBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "get cluster")
	})

	cluster := &core.Cluster{}
	clusterCall := s.mockCoreCli.
		On("GetAvailableClusterByID", mock.Anything, ent.ClusterID.Uint64()).
		Return(cluster, nil)
	defer clusterCall.Unset()
	s.Run("complete edit param error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "parse param")
		call := s.mockMysqlAdaptor.
			On("CompleteEditParams", mock.Anything, req.DownstreamParameter, ent.DownstreamParams).
			Return(mockErr)
		defer call.Unset()
		err := s.model.EditBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "manage edit param")
	})

	completeParamsCall := s.mockMysqlAdaptor.
		On("CompleteEditParams", mock.Anything, mock.Anything, mock.Anything).
		Return(nil)
	defer completeParamsCall.Unset()
	s.Run("update entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "version conflict")
		call := s.mockBizChangefeedRepo.On("Update", mock.Anything, mock.Anything).Return(mockErr)
		defer call.Unset()
		err := s.model.EditBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "update biz_changefeed entity")
	})

	updateCall := s.mockBizChangefeedRepo.On("Update", mock.Anything, mock.Anything).Return(nil)
	defer updateCall.Unset()
	mockConfig := &domain.TiCDCChangefeedConfig{
		URIConfig: &domain.TiCDCChangefeedMySQLURIConfigData{Endpoint: pointer.ToString("a")},
	}
	assembleCall := s.mockMysqlAdaptor.
		On("AssembleChangefeedConfig", mock.Anything, mock.Anything, mock.Anything).
		Return(mockConfig, nil)
	defer assembleCall.Unset()
	s.Run("edit kernel changefeed error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockKernelChangefeedModel.
			On("ModifyChangefeedConfig", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.EditBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "edit kernel changefeed")
	})

	s.Run("edit changefeed successfully", func() {
		modifyCall := s.mockKernelChangefeedModel.
			On("ModifyChangefeedConfig", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer modifyCall.Unset()
		hookCall := s.mockHooker.
			On("HandleBizChangefeedEvent", mock.Anything, entity.EditChangefeed, req.UserID, mock.Anything).
			Return()
		defer hookCall.Unset()
		err := s.model.EditBizChangefeed(context.Background(), req)
		s.NoError(err)
	})
}

func (s *bizChangefeedModelTestSuite) TestTestBizChangefeed() {
	req := &domain.TestBizChangefeedReq{
		ClusterID:           111,
		DownstreamParameter: &dataflow.MySQLInfo{},
	}

	s.Run("get downstream adaptor error", func() {
		req.DownstreamType = dataflow.ConnectorType_TOTAL
		err := s.model.TestBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(bizerr.ErrCodeParams, err.RawErrCode())
		s.Contains(err.Error(), "get downstream adaptor")
	})

	req.DownstreamType = dataflow.ConnectorType_MYSQL
	s.Run("get cluster error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockCoreCli.
			On("GetAvailableClusterByID", mock.Anything, req.ClusterID).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.TestBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "get cluster")
	})

	cluster := &core.Cluster{}
	clusterCall := s.mockCoreCli.
		On("GetAvailableClusterByID", mock.Anything, req.ClusterID).
		Return(cluster, nil)
	defer clusterCall.Unset()
	s.Run("test connection error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockMysqlAdaptor.
			On("TestConnection", mock.Anything, cluster, req.DownstreamParameter).
			Return(mockErr)
		defer call.Unset()
		err := s.model.TestBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "test connection")
	})

	s.Run("test connection successfully", func() {
		call := s.mockMysqlAdaptor.
			On("TestConnection", mock.Anything, cluster, req.DownstreamParameter).
			Return(nil)
		defer call.Unset()
		err := s.model.TestBizChangefeed(context.Background(), req)
		s.NoError(err)
	})
}

func (s *bizChangefeedModelTestSuite) TestGetBizChangefeed() {
	req := &domain.GetBizChangefeedReq{
		BizChangefeedID: 111,
	}
	s.Run("load entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockBizChangefeedRepo.
			On("Load", mock.Anything, req.BizChangefeedID).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.GetBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "load biz_changefeed")
	})

	ent := &entity.BizChangefeed{
		ID: entity.ID(req.BizChangefeedID),
	}
	loadCall := s.mockBizChangefeedRepo.
		On("Load", mock.Anything, req.BizChangefeedID).
		Return(ent, nil)
	defer loadCall.Unset()
	s.Run("query kernel changefeed error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockKernelChangefeedModel.
			On("QueryChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.GetBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "query kernel changefeed")
	})

	s.Run("query kernel changefeed not found", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockKernelChangefeedModel.
			On("QueryChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		resp, err := s.model.GetBizChangefeed(context.Background(), req)
		s.NoError(err)
		s.Empty(resp.BizChangefeed.ErrorMessage)
		s.Empty(resp.BizChangefeed.CheckpointTSO)
		s.Empty(resp.BizChangefeed.ChangefeedStatus)
	})

	mockKernelChangefeed := entity.TiCDCChangefeed{
		KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
			FeedState:     entity.ChangefeedKernelStatusNormal,
			CheckpointTSO: 10,
		},
	}
	mockResp := &domain.QueryChangefeedResp{
		ChangefeedEntity: mockKernelChangefeed,
	}
	queryCall := s.mockKernelChangefeedModel.
		On("QueryChangefeed", mock.Anything, mock.Anything).
		Return(mockResp, nil)
	defer queryCall.Unset()
	s.Run("query kernel changefeed successfully", func() {
		resp, err := s.model.GetBizChangefeed(context.Background(), req)
		s.NoError(err)
		s.Empty(resp.BizChangefeed.ErrorMessage)
		s.Equal(mockKernelChangefeed.KernelMetrics.CheckpointTSO, resp.BizChangefeed.CheckpointTSO)
		s.Equal(mockKernelChangefeed.KernelMetrics.FeedState, resp.BizChangefeed.ChangefeedStatus)
	})
}

func (s *bizChangefeedModelTestSuite) TestListKernelChangefeeds() {
	mockBizChangefeedIDs := []uint64{1, 2, 3}
	s.Run("list error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockKernelChangefeedModel.
			On("ListChangefeeds", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.listKernelChangefeeds(context.Background(), mockBizChangefeedIDs)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	mockKernelChangefeeds := []domain.ListChangefeedItem{
		{
			ChangefeedEntity: entity.TiCDCChangefeed{
				KernelChangefeedID: "1",
			},
		},
		{
			ChangefeedEntity: entity.TiCDCChangefeed{
				KernelChangefeedID: "2",
			},
		},
		{
			ChangefeedEntity: entity.TiCDCChangefeed{
				KernelChangefeedID: "3",
			},
		},
	}
	s.Run("list successfully", func() {
		mockResp := &domain.ListChangefeedsResp{
			Items: mockKernelChangefeeds,
			Total: int64(len(mockKernelChangefeeds)),
		}
		call := s.mockKernelChangefeedModel.
			On("ListChangefeeds", mock.Anything, mock.Anything).
			Return(mockResp, nil)
		defer call.Unset()
		changefeedEnts, err := s.model.listKernelChangefeeds(context.Background(), mockBizChangefeedIDs)
		s.NoError(err)

		matcher := func(req *domain.ListChangefeedsReq) bool {
			if len(req.Filter.BusinessIDs) != len(mockBizChangefeedIDs) {
				return false
			}
			for i := range req.Filter.BusinessIDs {
				if req.Filter.BusinessIDs[i] != mockBizChangefeedIDs[i] {
					return false
				}
			}
			return req.Pager.PageSize == int64(len(mockBizChangefeedIDs)) &&
				*req.Filter.BizType == domain.TiCDCClusterBizTypeChangefeed
		}
		s.mockKernelChangefeedModel.AssertCalled(
			s.T(), "ListChangefeeds", mock.Anything, mock.MatchedBy(matcher))
		s.Equal(len(mockKernelChangefeeds), len(changefeedEnts))
		for i := range mockKernelChangefeeds {
			s.Equal(&mockKernelChangefeeds[i].ChangefeedEntity, changefeedEnts[i])
		}
	})
}

func (s *bizChangefeedModelTestSuite) TestListBizChangefeeds() {
	mockReq := &domain.ListBizChangefeedsReq{
		ClusterID: 111,
		Page:      2,
		PerPage:   3,
	}
	s.Run("count biz_changefeed entities error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockBizChangefeedRepo.On("Count", mock.Anything, mock.Anything).Return(int64(0), mockErr)
		defer call.Unset()
		_, err := s.model.ListBizChangefeeds(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "count biz_changefeeds")
	})

	mockTotal := int64(100)
	countCall := s.mockBizChangefeedRepo.On("Count", mock.Anything, mock.Anything).Return(mockTotal, nil)
	defer countCall.Unset()
	s.Run("list biz_changefeed entities error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockBizChangefeedRepo.
			On("List", mock.Anything, mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.ListBizChangefeeds(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "list biz_changefeeds")
	})

	mockBizChangefeedEnts := []entity.BizChangefeed{
		{
			ID:              1,
			OperationStatus: entity.BizChangefeedOperationStatusFailed,
			ErrMsg:          "create changefeed error",
		},
		{
			ID:              2,
			OperationStatus: entity.BizChangefeedOperationStatusNormal,
		},
		{
			ID:              3,
			OperationStatus: entity.BizChangefeedOperationStatusNormal,
		},
	}
	listCall := s.mockBizChangefeedRepo.
		On("List", mock.Anything, mock.Anything, mock.Anything).
		Return(mockBizChangefeedEnts, nil)
	defer listCall.Unset()
	s.Run("list kernel changefeeds error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockKernelChangefeedModel.
			On("ListChangefeeds", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.ListBizChangefeeds(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "list kernel changefeeds")
	})

	mockKernelChangefeeds := []domain.ListChangefeedItem{
		{
			ChangefeedEntity: entity.TiCDCChangefeed{
				KernelChangefeedID: "1",
				BusinessID:         1,
			},
		},
		{
			ChangefeedEntity: entity.TiCDCChangefeed{
				KernelChangefeedID: "2",
				BusinessID:         2,
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState:     entity.ChangefeedKernelStatusNormal,
					CheckpointTSO: 20,
				},
			},
		},
		{
			ChangefeedEntity: entity.TiCDCChangefeed{
				KernelChangefeedID: "3",
				BusinessID:         3,
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState:     entity.ChangefeedKernelStatusError,
					CheckpointTSO: 30,
					RunningError: &entity.TiCDCChangefeedKernelRunningError{
						Code:    "1",
						Message: "changefeed failed",
					},
				},
			},
		},
	}
	mockListChangefeedsResp := &domain.ListChangefeedsResp{
		Items: mockKernelChangefeeds,
		Total: int64(len(mockKernelChangefeeds)),
	}
	listChangefeedsCall := s.mockKernelChangefeedModel.
		On("ListChangefeeds", mock.Anything, mock.Anything).
		Return(mockListChangefeedsResp, nil)
	defer listChangefeedsCall.Unset()
	s.Run("list kernel changefeeds successfully", func() {
		resp, err := s.model.ListBizChangefeeds(context.Background(), mockReq)
		s.NoError(err)
		matcher := func(req *domain.ListChangefeedsReq) bool {
			ids := []uint64{1, 2, 3}
			if len(req.Filter.BusinessIDs) != len(ids) {
				return false
			}
			for i := range ids {
				if ids[i] != req.Filter.BusinessIDs[i] {
					return false
				}
			}
			return true
		}
		s.mockKernelChangefeedModel.AssertCalled(
			s.T(), "ListChangefeeds", mock.Anything, mock.MatchedBy(matcher))
		s.Equal(mockTotal, int64(resp.Total))
		s.Equal(len(mockBizChangefeedEnts), len(resp.BizChangefeeds))
		s.Equal(len(mockKernelChangefeeds), len(resp.BizChangefeeds))
		for i := range resp.BizChangefeeds {
			s.Equal(mockBizChangefeedEnts[i].ID.Uint64(), resp.BizChangefeeds[i].BizChangefeedID)
			s.Equal(mockBizChangefeedEnts[i].OperationStatus, resp.BizChangefeeds[i].OperationStatus)
			s.Equal(
				mockKernelChangefeeds[i].ChangefeedEntity.KernelMetrics.FeedState,
				resp.BizChangefeeds[i].ChangefeedStatus)
			s.Equal(
				mockKernelChangefeeds[i].ChangefeedEntity.KernelMetrics.CheckpointTSO,
				resp.BizChangefeeds[i].CheckpointTSO)
			if mockKernelChangefeeds[i].ChangefeedEntity.KernelMetrics.RunningError != nil {
				s.Contains(
					resp.BizChangefeeds[i].ErrorMessage,
					mockKernelChangefeeds[i].ChangefeedEntity.KernelMetrics.RunningError.Message)
			} else {
				s.Equal(mockBizChangefeedEnts[i].ErrMsg, resp.BizChangefeeds[i].ErrorMessage)
			}
		}
	})
}

func (s *bizChangefeedModelTestSuite) TestGetClusterBizChangefeedRCU() {
	mockReq := &domain.GetClusterBizChangefeedRCUReq{
		ClusterID: 111,
	}
	s.Run("list biz_changefeed entities 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.model.GetClusterBizChangefeedRCU(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "list biz_changefeeds")
	})

	mockBizChangefeedEnts := []entity.BizChangefeed{
		{
			ID:                1,
			SpecificationCode: "rcu2",
		},
		{
			ID:                2,
			SpecificationCode: "rcu4",
		},
		{
			ID:                3,
			SpecificationCode: "rcu8",
		},
	}
	listCall := s.mockBizChangefeedRepo.
		On("List", mock.Anything, mock.Anything).
		Return(mockBizChangefeedEnts, nil)
	defer listCall.Unset()
	s.Run("get cluster error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockCoreCli.
			On("GetAvailableClusterByID", mock.Anything, mockReq.ClusterID).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.GetClusterBizChangefeedRCU(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "get cluster")
	})

	mockCluster := &core.Cluster{
		Id:                111,
		InfraProviderType: "eks",
	}
	clusterCall := s.mockCoreCli.
		On("GetAvailableClusterByID", mock.Anything, mockReq.ClusterID).
		Return(mockCluster, nil)
	defer clusterCall.Unset()
	s.Run("list spec 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.model.GetClusterBizChangefeedRCU(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "list specifications")
	})

	mockSpecEnts := []*domain.TiCDCClusterSpecification{
		{
			Code:     "rcu2",
			RCUNum:   2,
			Replicas: 1,
		},
		{
			Code:     "rcu4",
			RCUNum:   4,
			Replicas: 2,
		},
	}
	mockSpecResp := &domain.ListSpecificationsResp{
		Specifications: mockSpecEnts,
	}
	specCall := s.mockMachineManager.
		On("ListSpecifications", mock.Anything, mock.Anything).
		Return(mockSpecResp, nil)
	defer specCall.Unset()
	s.Run("specs are not full", func() {
		_, err := s.model.GetClusterBizChangefeedRCU(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "unknown specification code")
	})

	mockSpecResp.Specifications = append(mockSpecResp.Specifications, &domain.TiCDCClusterSpecification{
		Code:     "rcu8",
		RCUNum:   8,
		Replicas: 2,
	})
	s.Run("specs are full", func() {
		resp, err := s.model.GetClusterBizChangefeedRCU(context.Background(), mockReq)
		s.NoError(err)
		s.Equal(uint32(14), resp.RCUNumber)
	})
}

func (s *bizChangefeedModelTestSuite) TestCalBizChangefeedFee() {
	mockReq := &domain.CalBizChangefeedFeeReq{
		TenantID:          1,
		ProjectID:         11,
		SpecificationCode: "rcu4",
	}
	s.Run("query spec error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockMachineManager.
			On("Query", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.CalBizChangefeedFee(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "query specification code")
	})

	mockSpec := &domain.TiCDCClusterSpecification{
		Code:     "rcu4",
		RCUNum:   4,
		Replicas: 2,
	}
	mockSpecResp := &domain.QuerySpecificationResp{
		Specification: mockSpec,
	}
	specCall := s.mockMachineManager.
		On("Query", mock.Anything, mock.Anything).
		Return(mockSpecResp, nil)
	defer specCall.Unset()
	s.Run("cal fee error", func() {
		mockErr := errors.New("network error")
		call := s.mockBilling.
			On("CalFee", mock.Anything, mockReq.TenantID, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.CalBizChangefeedFee(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), mockErr.Error())
		s.Contains(err.Error(), "calculate changefeed fee")
	})

	mockFee := &billing.BillingCharge{
		RawFee: 1.0,
		Fee:    1.0,
	}
	feeCall := s.mockBilling.
		On("CalFee", mock.Anything, mockReq.TenantID, mock.Anything).
		Return(mockFee, nil)
	defer feeCall.Unset()
	s.Run("cal fee successfully", func() {
		resp, err := s.model.CalBizChangefeedFee(context.Background(), mockReq)
		s.NoError(err)
		s.Equal(mockFee.RawFee, resp.RawFee)
		s.Equal(mockFee.Fee, resp.Fee)
		matcher := func(req *domain.QuerySpecificationReq) bool {
			return req.Code == mockReq.SpecificationCode
		}
		s.mockMachineManager.AssertCalled(s.T(), "Query", mock.Anything, mock.MatchedBy(matcher))
	})
}

func (s *bizChangefeedModelTestSuite) TestListBizChangefeedSpecifications() {
	mockReq := &domain.ListBizChangefeedSpecificationsReq{
		ClusterID: 111,
	}
	s.Run("get cluster error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockCoreCli.
			On("GetAvailableClusterByID", mock.Anything, mockReq.ClusterID).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.ListBizChangefeedSpecifications(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "get cluster")
	})

	mockCluster := &core.Cluster{
		Id: 111,
	}
	clusterCall := s.mockCoreCli.
		On("GetAvailableClusterByID", mock.Anything, mockReq.ClusterID).
		Return(mockCluster, nil)
	defer clusterCall.Unset()
	s.Run("list spec 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.model.ListBizChangefeedSpecifications(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "list specifications")
	})

	mockSpecs := []*domain.TiCDCClusterSpecification{
		{
			Code:     "rcu2",
			Replicas: 1,
			RPSLimit: 20000,
			RCUNum:   2,
		},
		{
			Code:     "rcu4",
			Replicas: 2,
			RPSLimit: 40000,
			RCUNum:   4,
		},
		{
			Code:     "rcu8",
			Replicas: 2,
			RPSLimit: 80000,
			RCUNum:   8,
		},
	}
	mockSpecResp := &domain.ListSpecificationsResp{
		Specifications: mockSpecs,
	}
	specCall := s.mockMachineManager.
		On("ListSpecifications", mock.Anything, mock.Anything).
		Return(mockSpecResp, nil)
	defer specCall.Unset()
	s.Run("list spec successfully", func() {
		resp, err := s.model.ListBizChangefeedSpecifications(context.Background(), mockReq)
		s.NoError(err)
		s.Equal(len(mockSpecs), len(resp.Specifications))
		s.Equal(len(mockSpecs), int(resp.Total))
		for i := range mockSpecs {
			s.Equal(mockSpecs[i].Code, resp.Specifications[i].SpecificationCode)
			s.Equal(mockSpecs[i].RPSLimit, resp.Specifications[i].RPSLimit)
			s.Equal(mockSpecs[i].Replicas, resp.Specifications[i].Replicas)
			s.Equal(mockSpecs[i].GetRCUNumber(), resp.Specifications[i].RCUNumber)
		}
	})
}

func TestBizChangefeedModelTestSuite(t *testing.T) {
	suite.Run(t, new(bizChangefeedModelTestSuite))
}

func TestConvertTransactionError(t *testing.T) {
	testcases := []struct {
		inputErr        error
		expectedErrCode bizerr.ErrCode
	}{
		{
			inputErr:        nil,
			expectedErrCode: 0,
		},
		{
			inputErr:        errors.New("transaction error"),
			expectedErrCode: bizerr.ErrCodeDB,
		},
		{
			inputErr:        bizerr.New(bizerr.ErrCodeRecordExist, "duplicate row"),
			expectedErrCode: bizerr.ErrCodeRecordExist,
		},
	}

	for _, tc := range testcases {
		err := convertTransactionError(tc.inputErr)
		if tc.inputErr == nil {
			assert.NoError(t, err)
		} else {
			assert.Error(t, err)
			assert.Equal(t, tc.expectedErrCode, err.RawErrCode())
		}
	}
}

func TestConvertDownstreamType(t *testing.T) {
	testcases := []struct {
		downstreamType dataflow.ConnectorType
		err            bool
	}{
		{
			downstreamType: dataflow.ConnectorType_TOTAL,
			err:            true,
		},
		{
			downstreamType: dataflow.ConnectorType_KAFKA,
			err:            false,
		},
		{
			downstreamType: dataflow.ConnectorType_MYSQL,
			err:            false,
		},
		{
			downstreamType: dataflow.ConnectorType_SERVERLESS_TIDB,
			err:            false,
		},
		{
			downstreamType: dataflow.ConnectorType_S3,
			err:            false,
		},
		{
			downstreamType: dataflow.ConnectorType_GCS,
			err:            false,
		},
	}

	for _, tc := range testcases {
		_, err := convertDownstreamType(tc.downstreamType)
		if tc.err {
			assert.Error(t, err)
			assert.Equal(t, bizerr.ErrCodeParams, err.RawErrCode())
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestBuildBizChangefeedDagVarsTags(t *testing.T) {
	bizChangefeedID := uint64(1)
	dagAction := string(workflow.BusinessActionCreateBizChangefeed)
	vars, tags := buildBizChangefeedDagVarsTags(bizChangefeedID, dagAction)
	assert.Equal(t, 1, len(vars))
	assert.Equal(t, 3, len(tags))
	bizChangefeedIDStr := strconv.FormatUint(bizChangefeedID, 10)
	assert.Equal(t, bizChangefeedIDStr, vars[workflow.KeyBizChangefeedID])
	assert.Equal(t, dagAction, tags[workflow.BusinessActionKey])
	assert.Equal(t, bizChangefeedIDStr, tags[workflow.BusinessID])
}

func TestFillKernelChangefeedInfo(t *testing.T) {
	testcases := []struct {
		bizDomain *domain.BizChangefeed
		kernelEnt *entity.TiCDCChangefeed
		err       bool
	}{
		{
			bizDomain: &domain.BizChangefeed{},
			kernelEnt: &entity.TiCDCChangefeed{
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState:     entity.ChangefeedKernelStatusNormal,
					CheckpointTSO: 1,
				},
			},
			err: false,
		},
		{
			bizDomain: &domain.BizChangefeed{},
			kernelEnt: &entity.TiCDCChangefeed{
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState:     entity.ChangefeedKernelStatusError,
					CheckpointTSO: 10,
					RunningError: &entity.TiCDCChangefeedKernelRunningError{
						Addr:    "a",
						Code:    "1",
						Message: "err",
					},
				},
			},
			err: true,
		},
	}

	for _, tc := range testcases {
		fillKernelChangefeedInfo(tc.bizDomain, tc.kernelEnt)
		assert.Equal(t, tc.kernelEnt.KernelMetrics.FeedState, tc.bizDomain.ChangefeedStatus)
		assert.Equal(t, tc.kernelEnt.KernelMetrics.CheckpointTSO, tc.bizDomain.CheckpointTSO)
		if tc.err {
			assert.NotEmpty(t, tc.bizDomain.ErrorMessage)
		} else {
			assert.Empty(t, tc.bizDomain.ErrorMessage)
		}
	}
}