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/require"
	"github.com/stretchr/testify/suite"
	ticdc_v1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/dataflow-service-ng/app/config"
	cluster_job_mock "github.com/tidbcloud/dataflow-service-ng/app/internal/cluster_job/mock"
	biz_changefeed_mock "github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed/mock"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	sink_endpoint_mock "github.com/tidbcloud/dataflow-service-ng/app/models/sink_endpoint/mock"
	ticdc_mock "github.com/tidbcloud/dataflow-service-ng/app/models/ticdc/mock"
	"github.com/tidbcloud/dataflow-service-ng/app/repo"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service-ng/app/repo/mock"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	"github.com/tidbcloud/dataflow-service-ng/app/workflow"
	workflow_mock "github.com/tidbcloud/dataflow-service-ng/app/workflow/mock"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/dataflow-service-ng/pkg/billing"
	billing_mock "github.com/tidbcloud/dataflow-service-ng/pkg/billing/mock"
	grpcli_mock "github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli/mock"
	regionalapi_mock "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api/mock"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/db"
	commonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	v12 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
)

type bizChangefeedModelTestSuite struct {
	suite.Suite

	mockBizChangefeedRepo *repo_mock.BizChangefeedRepo

	mockMachineManager        *ticdc_mock.MachineManager
	mockKernelChangefeedModel *ticdc_mock.ChangefeedModel

	mockCoreCli           *grpcli_mock.MgmtSvcGrpcClient
	mockWorkflow          *workflow_mock.Workflow
	mockBilling           *billing_mock.BillingHandler
	mockSinkEndpointModel *sink_endpoint_mock.SinkEndpointModel

	mockMysqlAdaptor *biz_changefeed_mock.DownstreamAdaptor
	mockHooker       *biz_changefeed_mock.BizChangefeedHooker
	mockClusterJob   *cluster_job_mock.Model
	mockPreCheckCli  *regionalapi_mock.TiCDCPreCheckClient

	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.mockCoreCli = grpcli_mock.NewMgmtSvcGrpcClient(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.mockPreCheckCli = regionalapi_mock.NewTiCDCPreCheckClient(s.T())
	s.mockSinkEndpointModel = sink_endpoint_mock.NewSinkEndpointModel(s.T())

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

		machineManager:                s.mockMachineManager,
		regionalKernelChangefeedModel: s.mockKernelChangefeedModel,

		mgmtCli:           s.mockCoreCli,
		workflowCli:       s.mockWorkflow,
		billingHandler:    s.mockBilling,
		sinkEndpointModel: s.mockSinkEndpointModel,

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

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 := utils.NewCluster(&mgmt.Cluster{
		ClusterId:      111,
		ProjectId:      11,
		ClusterName:    "11",
		CloudProvider:  "aws",
		OrganizationId: 1,
		Region:         "us-west-2",
	})
	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 := utils.NewCluster(&mgmt.Cluster{
		ClusterId:      111,
		ProjectId:      11,
		ClusterName:    "11",
		CloudProvider:  "aws",
		OrganizationId: 1,
		Region:         "us-west-2",
	})
	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 := utils.NewCluster(&mgmt.Cluster{
		ClusterId:      111,
		ProjectId:      11,
		ClusterName:    "11",
		CloudProvider:  "aws",
		OrganizationId: 1,
		Region:         "us-west-2",
	})
	mockReq := &domain.CreateBizChangefeedReq{
		Name:              "test1",
		SpecificationCode: "code1",
		UserID:            1,
	}

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

	mockReq.DownstreamType = dataflow.ChangefeedType_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())
	// Omitting the identity leaves it for the repo to allocate.
	s.Zero(ent.ID.Uint64())

	// A caller that names one gets exactly that one. This is what makes an
	// orchestrated create resolvable by Get after a lost response.
	chosen := uint64(9007199254740993)
	mockReq.ID = &chosen
	ent, err = s.model.genBizChangefeedEntity(context.Background(), mockReq, mockCluster)
	s.NoError(err)
	s.Equal(chosen, ent.ID.Uint64())
}

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",
		DownstreamParameter: &dataflow.MySQLInfo{},
	}

	s.Run("get downstream adaptor error", func() {
		req.DownstreamType = dataflow.ChangefeedType(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.ChangefeedType_MYSQL
		mockClusterErr := bizerr.New(bizerr.ErrCodeGRPC, "grpc error")
		call := s.mockCoreCli.
			On("GetClusterByID", mock.Anything, req.ClusterID).
			Return((*mgmt.Cluster)(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() {
		// Empty cluster has state UNSPECIFIED → fails checkUpstreamCluster
		call := s.mockCoreCli.
			On("GetClusterByID", mock.Anything, req.ClusterID).
			Return(&mgmt.Cluster{}, nil)
		defer call.Unset()
		_, err := s.model.CreateBizChangefeed(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "precheck")
	})

	// make pre-check pass: cluster in AVAILABLE state
	mockCluster := &mgmt.Cluster{
		ClusterId: req.ClusterID,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &v12.Cluster{
				State: v12.Cluster_AVAILABLE,
			},
		},
	}
	clusterCall := s.mockCoreCli.
		On("GetClusterByID", 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()
	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("dry run completes prechecks without persistence", func() {
		req.DryRun = true
		resp, err := s.model.CreateBizChangefeed(context.Background(), req)
		s.NoError(err)
		s.NotNil(resp)
		s.Zero(resp.BizChangefeedID)
		s.Zero(resp.WorkflowID)
		req.DryRun = false
	})

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

	s.Run("create changefeed entity error", func() {
		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, 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, 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 TestCreateBizChangefeedReturnsSinkEndpointUsageError(t *testing.T) {
	testDB, dbErr := repo.GetFakeDB()
	require.NoError(t, dbErr)

	mockBizChangefeedRepo := repo_mock.NewBizChangefeedRepo(t)
	mockMachineManager := ticdc_mock.NewMachineManager(t)
	mockCoreCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	mockWorkflow := workflow_mock.NewWorkflow(t)
	mockBilling := billing_mock.NewBillingHandler(t)
	mockSinkEndpointModel := sink_endpoint_mock.NewSinkEndpointModel(t)
	mockMysqlAdaptor := biz_changefeed_mock.NewDownstreamAdaptor(t)
	mockHooker := biz_changefeed_mock.NewBizChangefeedHooker(t)
	mockClusterJob := cluster_job_mock.NewModel(t)
	mockPreCheckCli := regionalapi_mock.NewTiCDCPreCheckClient(t)

	model := &BizChangefeedModelImpl{
		transProvider:                 db.NewProvider(testDB),
		bizChangefeedRepo:             mockBizChangefeedRepo,
		machineManager:                mockMachineManager,
		mgmtCli:                       mockCoreCli,
		workflowCli:                   mockWorkflow,
		billingHandler:                mockBilling,
		sinkEndpointModel:             mockSinkEndpointModel,
		downstreamAdaptorFactory:      NewMockDownstreamAdaptorFactory(mockMysqlAdaptor),
		hooker:                        mockHooker,
		statusMatrix:                  NewBizChangefeedStatusMatrix(),
		clusterJob:                    mockClusterJob,
		cdcPreCheckCli:                mockPreCheckCli,
		regionalKernelChangefeedModel: ticdc_mock.NewChangefeedModel(t),
	}

	req := &domain.CreateBizChangefeedReq{
		ClusterID:             111,
		SpecificationCode:     "code1",
		DownstreamType:        dataflow.ChangefeedType_MYSQL,
		DownstreamParameter:   &dataflow.MySQLInfo{},
		PrivateLinkEndpointID: entity.ID(999),
		NetworkInfo:           &dataflow.NetworkInfo{NetworkType: dataflow.NetworkType_NetworkType_Private_Link},
	}
	cluster := &mgmt.Cluster{
		ClusterId: req.ClusterID,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &v12.Cluster{State: v12.Cluster_AVAILABLE},
		},
	}
	mockCoreCli.
		On("GetClusterByID", mock.Anything, req.ClusterID).
		Return(cluster, nil).
		Once()
	mockMachineManager.
		On("Query", mock.Anything, mock.Anything).
		Return(nil, nil).
		Once()
	mockBizChangefeedRepo.
		On("List", mock.Anything, mock.Anything).
		Return([]entity.BizChangefeed{}, nil).
		Once()
	mockSinkEndpointModel.
		On("QuerySinkEndpoint", mock.Anything, mock.MatchedBy(func(queryReq *domain.QuerySinkEndpointReq) bool {
			return queryReq.ID == req.PrivateLinkEndpointID.Uint64() && queryReq.ClusterID == req.ClusterID
		})).
		Return(&domain.QuerySinkEndpointResp{
			SinkEndpoint: &entity.BizSinkEndpoint{Status: entity.BizPLEndpointStatusAvailable},
		}, nil).
		Once()
	mockMysqlAdaptor.
		On("TestConnection", mock.Anything, mock.Anything, req.DownstreamParameter).
		Return(nil).
		Once()
	mockMysqlAdaptor.
		On("PreCheckChangefeedParameters", mock.Anything, mock.Anything, req.DownstreamParameter).
		Return(nil).
		Once()
	mockMysqlAdaptor.
		On("EncryptPasswordInPlace", mock.Anything, req.DownstreamParameter).
		Return(nil).
		Once()
	mockBizChangefeedRepo.
		On("Create", mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) {
			args.Get(1).(*entity.BizChangefeed).ID = entity.ID(1111)
		}).
		Return(nil).
		Once()
	mockSinkEndpointModel.
		On("CreateSinkEndpointUsage", mock.Anything, mock.Anything).
		Return((*domain.CreateSinkEndpointUsageResp)(nil), bizerr.New(bizerr.ErrCodeDB, "sink endpoint db error")).
		Once()

	resp, err := model.CreateBizChangefeed(context.Background(), req)
	require.Nil(t, resp)
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeDB, err.RawErrCode())
	require.Contains(t, err.Error(), "create sink endpoint usage")
}

func TestCheckCreateChangefeedNetwork(t *testing.T) {
	t.Run("essential v2 rejects vpc peering", func(t *testing.T) {
		req := &domain.CreateBizChangefeedReq{
			NetworkInfo: &dataflow.NetworkInfo{
				NetworkType: dataflow.NetworkType_NetworkType_VPC_Peering,
			},
		}
		cluster := utils.NewCluster(&mgmt.Cluster{
			ClusterInfo: &mgmt.Cluster_NextGenCluster{
				NextGenCluster: &v12.Cluster{ServicePlan: commonv1.ServicePlan_Essential_V2},
			},
		})
		err := checkCreateChangefeedNetwork(context.Background(), req, cluster, false)
		require.Error(t, err)
		require.Contains(t, err.Error(), "vpc peering")
	})

	t.Run("premium allows vpc peering", func(t *testing.T) {
		req := &domain.CreateBizChangefeedReq{
			NetworkInfo: &dataflow.NetworkInfo{
				NetworkType: dataflow.NetworkType_NetworkType_VPC_Peering,
			},
		}
		cluster := utils.NewCluster(&mgmt.Cluster{
			ClusterInfo: &mgmt.Cluster_NextGenCluster{
				NextGenCluster: &v12.Cluster{ServicePlan: commonv1.ServicePlan_Premium},
			},
		})
		err := checkCreateChangefeedNetwork(context.Background(), req, cluster, false)
		require.NoError(t, err)
	})

	t.Run("public rejects private mysql endpoint", func(t *testing.T) {
		req := &domain.CreateBizChangefeedReq{
			DownstreamType: dataflow.ChangefeedType_MYSQL,
			DownstreamParameter: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{Endpoint: "10.1.2.3:4000"},
			},
			NetworkInfo: &dataflow.NetworkInfo{NetworkType: dataflow.NetworkType_NetworkType_Public},
		}
		cluster := utils.NewCluster(&mgmt.Cluster{
			ClusterInfo: &mgmt.Cluster_NextGenCluster{
				NextGenCluster: &v12.Cluster{ServicePlan: commonv1.ServicePlan_Essential_V2},
			},
		})
		err := checkCreateChangefeedNetwork(context.Background(), req, cluster, false)
		require.Error(t, err)
		require.Contains(t, err.Error(), "non-public ip")
	})

	t.Run("public skips private mysql endpoint validation when allowlisted", func(t *testing.T) {
		req := &domain.CreateBizChangefeedReq{
			DownstreamType: dataflow.ChangefeedType_MYSQL,
			DownstreamParameter: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{Endpoint: "10.1.2.3:4000"},
			},
			NetworkInfo: &dataflow.NetworkInfo{NetworkType: dataflow.NetworkType_NetworkType_Public},
		}
		cluster := utils.NewCluster(&mgmt.Cluster{
			ClusterInfo: &mgmt.Cluster_NextGenCluster{
				NextGenCluster: &v12.Cluster{ServicePlan: commonv1.ServicePlan_Essential_V2},
			},
		})
		err := checkCreateChangefeedNetwork(context.Background(), req, cluster, true)
		require.NoError(t, err)
	})

	t.Run("nil network info defaults to public and rejects private mysql endpoint", func(t *testing.T) {
		req := &domain.CreateBizChangefeedReq{
			DownstreamType: dataflow.ChangefeedType_MYSQL,
			DownstreamParameter: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{Endpoint: "10.1.2.3:4000"},
			},
		}
		cluster := utils.NewCluster(&mgmt.Cluster{
			ClusterInfo: &mgmt.Cluster_NextGenCluster{
				NextGenCluster: &v12.Cluster{ServicePlan: commonv1.ServicePlan_Essential_V2},
			},
		})
		err := checkCreateChangefeedNetwork(context.Background(), req, cluster, false)
		require.Error(t, err)
		require.Contains(t, err.Error(), "non-public ip")
	})

	t.Run("public rejects private ipv6 mysql endpoint", func(t *testing.T) {
		req := &domain.CreateBizChangefeedReq{
			DownstreamType: dataflow.ChangefeedType_MYSQL,
			DownstreamParameter: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{Endpoint: "[fd00::1]:4000"},
			},
			NetworkInfo: &dataflow.NetworkInfo{NetworkType: dataflow.NetworkType_NetworkType_Public},
		}
		cluster := utils.NewCluster(&mgmt.Cluster{
			ClusterInfo: &mgmt.Cluster_NextGenCluster{
				NextGenCluster: &v12.Cluster{ServicePlan: commonv1.ServicePlan_Essential_V2},
			},
		})
		err := checkCreateChangefeedNetwork(context.Background(), req, cluster, false)
		require.Error(t, err)
		require.Contains(t, err.Error(), "non-public ip")
	})

	t.Run("premium skips public private ip validation", func(t *testing.T) {
		req := &domain.CreateBizChangefeedReq{
			DownstreamType: dataflow.ChangefeedType_MYSQL,
			DownstreamParameter: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{Endpoint: "10.1.2.3:4000"},
			},
			NetworkInfo: &dataflow.NetworkInfo{NetworkType: dataflow.NetworkType_NetworkType_Public},
		}
		cluster := utils.NewCluster(&mgmt.Cluster{
			ClusterInfo: &mgmt.Cluster_NextGenCluster{
				NextGenCluster: &v12.Cluster{ServicePlan: commonv1.ServicePlan_Premium},
			},
		})
		require.NoError(t, checkCreateChangefeedNetwork(context.Background(), req, cluster, false))
	})

	t.Run("public allows public kafka and pulsar endpoints", func(t *testing.T) {
		cluster := utils.NewCluster(&mgmt.Cluster{
			ClusterInfo: &mgmt.Cluster_NextGenCluster{
				NextGenCluster: &v12.Cluster{ServicePlan: commonv1.ServicePlan_Essential_V2},
			},
		})
		kafkaReq := &domain.CreateBizChangefeedReq{
			DownstreamType: dataflow.ChangefeedType_KAFKA,
			DownstreamParameter: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{BrokerEndpoints: "8.8.8.8:9092"},
			},
			NetworkInfo: &dataflow.NetworkInfo{NetworkType: dataflow.NetworkType_NetworkType_Public},
		}
		require.NoError(t, checkCreateChangefeedNetwork(context.Background(), kafkaReq, cluster, false))

		pulsarReq := &domain.CreateBizChangefeedReq{
			DownstreamType:      dataflow.ChangefeedType_PULSAR,
			DownstreamParameter: &dataflow.PulsarInfo{Endpoint: "8.8.4.4:6650"},
			NetworkInfo:         &dataflow.NetworkInfo{NetworkType: dataflow.NetworkType_NetworkType_Public},
		}
		require.NoError(t, checkCreateChangefeedNetwork(context.Background(), pulsarReq, cluster, false))
	})
}

func TestCheckSinkEndpointUsesClusterScope(t *testing.T) {
	sinkEndpointModel := sink_endpoint_mock.NewSinkEndpointModel(t)
	model := &BizChangefeedModelImpl{sinkEndpointModel: sinkEndpointModel}
	cluster := utils.NewCluster(&mgmt.Cluster{
		ClusterId: 111,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &v12.Cluster{},
		},
	})
	req := &domain.CreateBizChangefeedReq{
		PrivateLinkEndpointID: entity.ID(999),
		NetworkInfo:           &dataflow.NetworkInfo{NetworkType: dataflow.NetworkType_NetworkType_Private_Link},
	}
	queryMatcher := mock.MatchedBy(func(req *domain.QuerySinkEndpointReq) bool {
		return req.ID == 999 && req.ClusterID == cluster.GetClusterID()
	})
	sinkEndpointModel.
		On("QuerySinkEndpoint", mock.Anything, queryMatcher).
		Return(&domain.QuerySinkEndpointResp{
			SinkEndpoint: &entity.BizSinkEndpoint{Status: entity.BizPLEndpointStatusAvailable},
		}, nil).
		Once()

	require.NoError(t, model.checkSinkEndpoint(context.Background(), req, cluster))
}

func TestCheckSinkEndpointSkipsNonPrivateLinkNetwork(t *testing.T) {
	sinkEndpointModel := sink_endpoint_mock.NewSinkEndpointModel(t)
	model := &BizChangefeedModelImpl{sinkEndpointModel: sinkEndpointModel}
	cluster := utils.NewCluster(&mgmt.Cluster{ClusterId: 111})
	req := &domain.CreateBizChangefeedReq{
		PrivateLinkEndpointID: entity.ID(999),
		NetworkInfo:           &dataflow.NetworkInfo{NetworkType: dataflow.NetworkType_NetworkType_Public},
	}

	require.NoError(t, model.checkSinkEndpoint(context.Background(), req, cluster))
	sinkEndpointModel.AssertNotCalled(t, "QuerySinkEndpoint", mock.Anything, mock.Anything)
}

func (s *bizChangefeedModelTestSuite) TestCreateSinkEndpointUsageIfNeeded() {
	const mockEndpointID = entity.ID(999)
	const mockChangefeedID = entity.ID(1111)

	s.Run("no endpoint id - skips creation", func() {
		ent := &entity.BizChangefeed{ID: mockChangefeedID, BizPLEndpointID: 0}
		// Track call count before to detect new calls only (AssertNotCalled checks all-time history).
		callsBefore := len(s.mockSinkEndpointModel.Calls)
		err := s.model.createSinkEndpointUsageIfNeeded(context.Background(), ent)
		s.NoError(err)
		// Verify CreateSinkEndpointUsage is NOT called when no endpoint is set.
		s.Equal(callsBefore, len(s.mockSinkEndpointModel.Calls), "CreateSinkEndpointUsage should not have been called")
	})

	s.Run("with endpoint id - create usage succeeds", func() {
		ent := &entity.BizChangefeed{ID: mockChangefeedID, BizPLEndpointID: mockEndpointID}
		expectedReq := &domain.CreateSinkEndpointUsageReq{
			SinkEndpointID: mockEndpointID.Uint64(),
			OwnerID:        strconv.FormatUint(mockChangefeedID.Uint64(), 10),
			UsageType:      entity.BizSinkEndpointUsageTypeChangefeed,
		}
		call := s.mockSinkEndpointModel.
			On("CreateSinkEndpointUsage", mock.Anything, expectedReq).
			Return(&domain.CreateSinkEndpointUsageResp{}, nil)
		defer call.Unset()

		err := s.model.createSinkEndpointUsageIfNeeded(context.Background(), ent)
		s.NoError(err)
		s.mockSinkEndpointModel.AssertCalled(s.T(), "CreateSinkEndpointUsage", mock.Anything, expectedReq)
	})

	s.Run("with endpoint id - create usage fails", func() {
		ent := &entity.BizChangefeed{ID: mockChangefeedID, BizPLEndpointID: mockEndpointID}
		mockErr := bizerr.New(bizerr.ErrCodeDB, "db error")
		call := s.mockSinkEndpointModel.
			On("CreateSinkEndpointUsage", mock.Anything, mock.Anything).
			Return((*domain.CreateSinkEndpointUsageResp)(nil), mockErr)
		defer call.Unset()

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

func (s *bizChangefeedModelTestSuite) TestDeleteSinkEndpointUsageIfNeeded() {
	const mockEndpointID = entity.ID(999)
	const mockChangefeedID = entity.ID(1111)

	s.Run("no endpoint id - skips deletion", func() {
		ent := &entity.BizChangefeed{ID: mockChangefeedID, BizPLEndpointID: 0}
		// Track call count before to detect new calls only (AssertNotCalled checks all-time history).
		callsBefore := len(s.mockSinkEndpointModel.Calls)
		err := s.model.deleteSinkEndpointUsageIfNeeded(context.Background(), ent)
		s.NoError(err)
		s.Equal(callsBefore, len(s.mockSinkEndpointModel.Calls), "DeleteSinkEndpointUsage should not have been called")
	})

	s.Run("with endpoint id - delete usage succeeds", func() {
		ent := &entity.BizChangefeed{ID: mockChangefeedID, BizPLEndpointID: mockEndpointID}
		expectedReq := &domain.DeleteSinkEndpointUsageReq{
			SinkEndpointID: mockEndpointID.Uint64(),
			OwnerID:        strconv.FormatUint(mockChangefeedID.Uint64(), 10),
			UsageType:      entity.BizSinkEndpointUsageTypeChangefeed,
		}
		call := s.mockSinkEndpointModel.
			On("DeleteSinkEndpointUsage", mock.Anything, expectedReq).
			Return(nil)
		defer call.Unset()

		err := s.model.deleteSinkEndpointUsageIfNeeded(context.Background(), ent)
		s.NoError(err)
		s.mockSinkEndpointModel.AssertCalled(s.T(), "DeleteSinkEndpointUsage", mock.Anything, expectedReq)
	})

	s.Run("with endpoint id - delete usage fails", func() {
		ent := &entity.BizChangefeed{ID: mockChangefeedID, BizPLEndpointID: mockEndpointID}
		mockErr := bizerr.New(bizerr.ErrCodeDB, "db error")
		call := s.mockSinkEndpointModel.
			On("DeleteSinkEndpointUsage", mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()

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

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("paused changefeed remains rejected by public scale", func() {
		req.SpecificationCode = "code2"
		ent.OperationStatus = entity.BizChangefeedOperationStatusPaused
		err := s.model.preCheckScaleBizChangefeed(context.Background(), req, ent)
		s.Error(err)
		s.Equal(bizerr.ErrCodeParams, err.RawErrCode())
		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() {
	bizChangefeedID := uint64(111)

	matcher := func(queryReq *domain.QueryChangefeedReq) bool {
		return queryReq.Filter.BizChangefeedIDs[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.ChangefeedKernelStatusFailed
	req.DownstreamType = dataflow.ChangefeedType_MYSQL
	ent.DownstreamType = entity.TiCDCDownStreamTypeMySQL
	ent.OperationStatus = entity.BizChangefeedOperationStatusNormal
	s.Run("failed kernel changefeed is editable", func() {
		err := s.model.preCheckEditBizChangefeed(context.Background(), req, ent, kernelChangefeed)
		s.NoError(err)
	})
	kernelChangefeed.KernelMetrics.FeedState = entity.ChangefeedKernelStatusStopped
	s.Run("invalid downstream type", func() {
		req.DownstreamType = dataflow.ChangefeedType_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.ChangefeedType_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)
	})
}

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,
		SpecificationCode: "original-spec",
	}
	matcher := func(updateEnt *entity.BizChangefeed) bool {
		return updateEnt.Name == req.Name &&
			updateEnt.DownstreamParams == downstreamParamsStr &&
			updateEnt.SpecificationCode == "original-spec"
	}

	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 := utils.NewCluster(&mgmt.Cluster{
		ClusterId:      111,
		ProjectId:      11,
		ClusterName:    "11",
		CloudProvider:  "aws",
		OrganizationId: 1,
		Region:         "us-west-2",
	})
	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.ChangefeedType_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")
	})

	req.DownstreamType = dataflow.ChangefeedType_MYSQL
	s.Run("get cluster error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockCoreCli.
			On("GetClusterByID", 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 := &mgmt.Cluster{}
	clusterCall := s.mockCoreCli.
		On("GetClusterByID", mock.Anything, ent.ClusterID.Uint64()).
		Return(cluster, nil)
	defer clusterCall.Unset()
	s.Run("dry run completes params without persistence", func() {
		const storedParams = `{"connection":{"password":"stored-secret"}}`
		currentParams := ent.DownstreamParams
		ent.DownstreamParams = storedParams
		defer func() { ent.DownstreamParams = currentParams }()
		dryRunParams := &dataflow.MySQLInfo{Connection: &dataflow.Connection{Username: "root"}}
		dryRunReq := &domain.EditBizChangefeedReq{
			BizChangefeedID:     req.BizChangefeedID,
			DownstreamType:      dataflow.ChangefeedType_MYSQL,
			DownstreamParameter: dryRunParams,
			DryRun:              true,
		}
		call := s.mockMysqlAdaptor.
			On("CompleteEditParams", mock.Anything, dryRunParams, storedParams).
			Run(func(args mock.Arguments) {
				params := args.Get(1).(*dataflow.MySQLInfo)
				params.Connection.Password = "stored-secret"
			}).
			Return(nil)
		defer call.Unset()

		err := s.model.EditBizChangefeed(context.Background(), dryRunReq)
		s.NoError(err)
		s.Equal("stored-secret", dryRunParams.GetConnection().GetPassword())
	})
	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.ChangefeedType_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.ChangefeedType_MYSQL
	s.Run("get cluster error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockCoreCli.
			On("GetClusterByID", 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 := &mgmt.Cluster{
		ClusterInfo: &mgmt.Cluster_NextGenCluster{NextGenCluster: &v12.Cluster{}},
	}
	clusterCall := s.mockCoreCli.
		On("GetClusterByID", 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, mock.Anything, 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, mock.Anything, req.DownstreamParameter).
			Return(nil)
		defer call.Unset()
		err := s.model.TestBizChangefeed(context.Background(), req)
		s.NoError(err)
	})
}

func (s *bizChangefeedModelTestSuite) TestDescribeSchemaTable() {
	req := &domain.DescribeSchemaTableReq{
		ClusterID: 111,
		Username:  "sss",
		Password:  "aaa",
	}

	s.Run("pass filter error", func() {
		req.Filters = []string{"a.b.c.d"}
		_, err := s.model.DescribeSchemaTable(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "invalid table filter")
	})

	req.Filters = []string{"a.b"}
	s.Run("get cluster error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockCoreCli.
			On("GetClusterByID", mock.Anything, req.ClusterID).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.DescribeSchemaTable(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "network error")
	})

	cluster := &mgmt.Cluster{
		ClusterInfo: &mgmt.Cluster_NextGenCluster{NextGenCluster: &v12.Cluster{}},
	}
	clusterCall := s.mockCoreCli.
		On("GetClusterByID", mock.Anything, req.ClusterID).
		Return(cluster, nil)
	defer clusterCall.Unset()
	s.Run("verify table error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "verify table")
		call := s.mockPreCheckCli.
			On("VerifyTable", mock.Anything, mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.DescribeSchemaTable(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "verify table")
	})

	s.Run("verify table successfully", func() {
		call := s.mockPreCheckCli.
			On("VerifyTable", mock.Anything, mock.Anything, mock.Anything).
			Return(&ticdc_v1.VerifyTableResponse{}, nil)
		defer call.Unset()
		_, err := s.model.DescribeSchemaTable(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.BizChangefeedIDs) != len(mockBizChangefeedIDs) {
				return false
			}
			for i := range req.Filter.BizChangefeedIDs {
				if req.Filter.BizChangefeedIDs[i] != mockBizChangefeedIDs[i] {
					return false
				}
			}
			return req.Pager.PageSize == int64(len(mockBizChangefeedIDs))
		}
		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",
				BizChangefeedID:    1,
			},
		},
		{
			ChangefeedEntity: entity.TiCDCChangefeed{
				KernelChangefeedID: "2",
				BizChangefeedID:    2,
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState:     entity.ChangefeedKernelStatusNormal,
					CheckpointTSO: 20,
				},
			},
		},
		{
			ChangefeedEntity: entity.TiCDCChangefeed{
				KernelChangefeedID: "3",
				BizChangefeedID:    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.BizChangefeedIDs) != len(ids) {
				return false
			}
			for i := range ids {
				if ids[i] != req.Filter.BizChangefeedIDs[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("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",
			CpuNum:   2,
			Replicas: 1,
		},
		{
			Code:     "rcu4",
			CpuNum:   2,
			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",
		CpuNum:   4,
		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{
		ClusterID:         1,
		SpecificationCode: "rcu4",
	}
	mockCluster := &mgmt.Cluster{
		ClusterId:      1,
		OrganizationId: 123,
		CloudProvider:  string(entity.CloudProviderTypeAWS),
		ClusterInfo:    &mgmt.Cluster_NextGenCluster{NextGenCluster: &v12.Cluster{}},
	}

	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",
		CpuNum:   2,
		Replicas: 2,
	}
	mockSpecResp := &domain.QuerySpecificationResp{
		Specification: mockSpec,
	}
	specCall := s.mockMachineManager.
		On("Query", mock.Anything, mock.Anything).
		Return(mockSpecResp, nil)
	defer specCall.Unset()

	s.Run("query cluster error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockCoreCli.
			On("GetClusterByID", mock.Anything, mockReq.ClusterID).
			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(), "not found")
	})
	call := s.mockCoreCli.
		On("GetClusterByID", mock.Anything, mockReq.ClusterID).
		Return(mockCluster, nil)
	defer call.Unset()

	s.Run("cal fee error", func() {
		mockErr := errors.New("network error")
		call := s.mockBilling.
			On("CalFee", mock.Anything, mockCluster.OrganizationId, 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, mockCluster.OrganizationId, 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("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,
			CpuNum:   2,
		},
		{
			Code:     "rcu4",
			Replicas: 2,
			RPSLimit: 40000,
			CpuNum:   2,
		},
		{
			Code:     "rcu8",
			Replicas: 2,
			RPSLimit: 80000,
			CpuNum:   4,
		},
	}
	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.ChangefeedType
		err            bool
	}{
		{
			downstreamType: dataflow.ChangefeedType_TOTAL,
			err:            true,
		},
		{
			downstreamType: dataflow.ChangefeedType_KAFKA,
			err:            false,
		},
		{
			downstreamType: dataflow.ChangefeedType_MYSQL,
			err:            false,
		},
		{
			downstreamType: dataflow.ChangefeedType_SERVERLESS_TIDB,
			err:            false,
		},
		{
			downstreamType: dataflow.ChangefeedType_S3,
			err:            false,
		},
		{
			downstreamType: dataflow.ChangefeedType_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])
}

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

// TestPreCheckRejectsAnExplicitZeroChangefeedID covers the one value a caller
// must not choose. Zero is what the repo reads as "allocate for me", so
// accepting it would answer an explicit choice with a different id - and an
// orchestrator that supplied it to make its create resolvable by Get would
// never find the row again. Nil still means "allocate".
func TestPreCheckRejectsAnExplicitZeroChangefeedID(t *testing.T) {
	// The check runs before any dependency is consulted, so a bare model is
	// enough to reach it.
	model := &BizChangefeedModelImpl{}
	zero := uint64(0)

	err := model.preCheckCreateBizChangefeed(
		context.Background(), &domain.CreateBizChangefeedReq{ID: &zero}, nil, nil)
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode())
	require.Contains(t, err.Error(), "must not be zero")
}