package inner

import (
	"context"
	"testing"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	biz_changefeed_mock "github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed/mock"
	config_manager_mock "github.com/tidbcloud/dataflow-service-ng/app/models/config_manager/mock"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	ticdc_mock "github.com/tidbcloud/dataflow-service-ng/app/models/ticdc/mock"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	grpc_mock "github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli/mock"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	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"
	"google.golang.org/protobuf/types/known/structpb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

type opsInnerControllerImplTestSuite struct {
	suite.Suite
	mockBizChangefeedModel        *biz_changefeed_mock.BizChangefeedModel
	mockTiCDCClusterModel         *ticdc_mock.ClusterModel
	mockTiCDCChangefeedModel      *ticdc_mock.ChangefeedModel
	mockTiCDCClusterConfigManager *config_manager_mock.TiCDCClusterConfigManager
	mockChangefeedConfigManager   *config_manager_mock.ChangefeedConfigManager
	mockMgmtCli                   *grpc_mock.MgmtSvcGrpcClient
	controller                    *opsInnerControllerImpl
}

func (s *opsInnerControllerImplTestSuite) SetupSuite() {
	s.mockBizChangefeedModel = biz_changefeed_mock.NewBizChangefeedModel(s.T())
	s.mockTiCDCClusterModel = ticdc_mock.NewClusterModel(s.T())
	s.mockTiCDCChangefeedModel = ticdc_mock.NewChangefeedModel(s.T())
	s.mockTiCDCClusterConfigManager = config_manager_mock.NewTiCDCClusterConfigManager(s.T())
	s.mockChangefeedConfigManager = config_manager_mock.NewChangefeedConfigManager(s.T())
	s.mockMgmtCli = grpc_mock.NewMgmtSvcGrpcClient(s.T())
	s.controller = &opsInnerControllerImpl{
		bizChangefeedModel:        s.mockBizChangefeedModel,
		ticdcClusterModel:         s.mockTiCDCClusterModel,
		ticdcChangefeedModel:      s.mockTiCDCChangefeedModel,
		ticdcClusterConfigManager: s.mockTiCDCClusterConfigManager,
		changefeedConfigManager:   s.mockChangefeedConfigManager,
		mgmtCli:                   s.mockMgmtCli,
	}
}

func (s *opsInnerControllerImplTestSuite) TestDataflowConnector2Internal() {
	createdAt := time.Now()
	testcases := []struct {
		connector          *dataflow.Changefeed
		expectedChangefeed *dataflow.InternalChangefeed
	}{
		{
			connector: &dataflow.Changefeed{
				ChangefeedId:   111,
				Name:           "changefeed-1",
				ChangefeedType: dataflow.ChangefeedType_KAFKA,
				Status:         dataflow.Changefeed_RUNNING,
				CreatedAt:      timestamppb.New(createdAt),
			},
			expectedChangefeed: &dataflow.InternalChangefeed{
				Id:             111,
				DisplayId:      111,
				DisplayName:    "changefeed-1",
				DownstreamType: "kafka",
				Status:         "normal",
				CreatedAt:      timestamppb.New(createdAt),
			},
		},
		{
			connector: &dataflow.Changefeed{
				ChangefeedId:   112,
				Name:           "changefeed-2",
				ChangefeedType: dataflow.ChangefeedType_MYSQL,
				Status:         dataflow.Changefeed_SCALING,
				CreatedAt:      timestamppb.New(createdAt),
			},
			expectedChangefeed: &dataflow.InternalChangefeed{
				Id:             112,
				DisplayId:      112,
				DisplayName:    "changefeed-2",
				DownstreamType: "mysql",
				Status:         "scaling",
				CreatedAt:      timestamppb.New(createdAt),
			},
		},
	}

	for _, tc := range testcases {
		internalChangefeed := s.controller.dataflowConnector2Internal(tc.connector)
		s.Equal(tc.expectedChangefeed, internalChangefeed)
	}
}

func (s *opsInnerControllerImplTestSuite) TestTiCDCClusterEntity2DataflowInternal() {
	createdAt := time.Now()
	testcases := []struct {
		ticdcClusterEnt      *entity.TiCDCCluster
		expectedTiCDCCluster *dataflow.InternalTiCDCCluster
	}{
		{
			ticdcClusterEnt: &entity.TiCDCCluster{
				Updatable: entity.Updatable{
					Base: entity.Base{
						ID:        1111,
						CreatedAt: createdAt,
					},
				},
				Name:              "ticdc-1111",
				SharedPoolID:      "111",
				Region:            "us-west-2",
				CloudProvider:     entity.CloudProviderTypeAWS,
				Version:           "v7.5.0",
				SpecificationCode: "16rcu",
				Status:            entity.TiCDCClusterStatusRunning,
			},
			expectedTiCDCCluster: &dataflow.InternalTiCDCCluster{
				Id:                1111,
				Name:              "ticdc-1111",
				SharedPoolId:      "111",
				Region:            "us-west-2",
				ProviderType:      "aws",
				Version:           "v7.5.0",
				SpecificationCode: "16rcu",
				Status:            "running",
				CreatedAt:         timestamppb.New(createdAt),
			},
		},
		{
			ticdcClusterEnt: &entity.TiCDCCluster{
				Updatable: entity.Updatable{
					Base: entity.Base{
						ID:        1112,
						CreatedAt: createdAt,
					},
				},
				Name:              "ticdc-1112",
				SharedPoolID:      "112",
				Region:            "us-west1",
				CloudProvider:     entity.CloudProviderTypeGCP,
				Version:           "v7.5.0",
				SpecificationCode: "16rcu",
				Status:            entity.TiCDCClusterStatusScaling,
			},
			expectedTiCDCCluster: &dataflow.InternalTiCDCCluster{
				Id:                1112,
				Name:              "ticdc-1112",
				SharedPoolId:      "112",
				Region:            "us-west1",
				ProviderType:      "gcp",
				Version:           "v7.5.0",
				SpecificationCode: "16rcu",
				Status:            "scaling",
				CreatedAt:         timestamppb.New(createdAt),
			},
		},
	}

	for _, tc := range testcases {
		internalTiCDCCluster := s.controller.ticdcClusterEntity2DataflowInternal(tc.ticdcClusterEnt)
		s.Equal(tc.expectedTiCDCCluster, internalTiCDCCluster)
	}
}

func (s *opsInnerControllerImplTestSuite) TestTiCDCChangefeedEntity2DataflowInternal() {
	createdAt := time.Now()
	testcases := []struct {
		ticdcChangefeedEnt *entity.TiCDCChangefeed
		expectedChangefeed *dataflow.InternalChangefeed
	}{
		{
			ticdcChangefeedEnt: &entity.TiCDCChangefeed{
				Updatable: entity.Updatable{
					Base: entity.Base{
						ID:        1111,
						CreatedAt: createdAt,
					},
				},
				KernelChangefeedID: "changefeed-1111",
				DownStreamType:     entity.TiCDCDownStreamTypeKafka,
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusNormal,
				},
			},
			expectedChangefeed: &dataflow.InternalChangefeed{
				Id:             1111,
				DisplayId:      1111,
				DisplayName:    "changefeed-1111",
				DownstreamType: "kafka",
				Status:         "normal",
				CreatedAt:      timestamppb.New(createdAt),
			},
		},
		{
			ticdcChangefeedEnt: &entity.TiCDCChangefeed{
				Updatable: entity.Updatable{
					Base: entity.Base{
						ID:        1112,
						CreatedAt: createdAt,
					},
				},
				KernelChangefeedID: "changefeed-1112",
				DownStreamType:     entity.TiCDCDownStreamTypeMySQL,
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusStopped,
				},
			},
			expectedChangefeed: &dataflow.InternalChangefeed{
				Id:             1112,
				DisplayId:      1112,
				DisplayName:    "changefeed-1112",
				DownstreamType: "mysql",
				Status:         "stopped",
				CreatedAt:      timestamppb.New(createdAt),
			},
		},
		{
			ticdcChangefeedEnt: &entity.TiCDCChangefeed{
				Updatable: entity.Updatable{
					Base: entity.Base{
						ID:        1113,
						CreatedAt: createdAt,
					},
				},
				KernelChangefeedID: "changefeed-1113",
				DownStreamType:     entity.TiCDCDownStreamTypeKafka,
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusNormal,
				},
				TiCDCCluster: &entity.TiCDCCluster{
					Updatable: entity.Updatable{
						Base: entity.Base{
							ID:        1111,
							CreatedAt: createdAt,
						},
					},
					Name:              "ticdc-1111",
					SharedPoolID:      "111",
					Region:            "us-west-2",
					CloudProvider:     entity.CloudProviderTypeAWS,
					Version:           "v7.5.0",
					SpecificationCode: "16rcu",
					Status:            entity.TiCDCClusterStatusRunning,
				},
			},
			expectedChangefeed: &dataflow.InternalChangefeed{
				Id:             1113,
				DisplayId:      1113,
				DisplayName:    "changefeed-1113",
				DownstreamType: "kafka",
				Status:         "normal",
				CreatedAt:      timestamppb.New(createdAt),
				TicdcCluster: &dataflow.InternalTiCDCCluster{
					Id:                1111,
					Name:              "ticdc-1111",
					SharedPoolId:      "111",
					Region:            "us-west-2",
					ProviderType:      "aws",
					Version:           "v7.5.0",
					SpecificationCode: "16rcu",
					Status:            "running",
					CreatedAt:         timestamppb.New(createdAt),
				},
			},
		},
	}

	for _, tc := range testcases {
		internalChangefeed := s.controller.ticdcChangefeedEntity2DataflowInternal(tc.ticdcChangefeedEnt)
		s.Equal(tc.expectedChangefeed, internalChangefeed)
	}
}

func (s *opsInnerControllerImplTestSuite) TestAssembleTiCDCChangefeeds() {
	createdAt := time.Now()
	ticdcClusterIDs := []uint64{111, 112}
	internalTiCDCClusters := map[uint64]*dataflow.InternalTiCDCCluster{
		111: {
			Id:        111,
			Name:      "ticdc-111",
			CreatedAt: timestamppb.New(createdAt),
		},
		112: {
			Id:        112,
			Name:      "ticdc-112",
			CreatedAt: timestamppb.New(createdAt),
		},
	}
	bizChangefeeds := map[uint64]*domain.BizChangefeed{
		111111: {
			BizChangefeedID: 111111,
			Name:            "biz-changefeed-1",
		},
		111112: {
			BizChangefeedID: 111112,
			Name:            "biz-changefeed-2",
		},
	}
	s.Run("list changefeeds error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCChangefeedModel.
			On("ListChangefeeds", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()

		_, err := s.controller.assembleTiCDCChangefeeds(
			context.Background(), ticdcClusterIDs, internalTiCDCClusters, bizChangefeeds)
		s.Error(err)
		s.Contains(err.Error(), "list ticdc changefeeds")
	})

	mockListResp := &domain.ListChangefeedsResp{
		Items: []domain.ListChangefeedItem{
			{
				ChangefeedEntity: entity.TiCDCChangefeed{
					Updatable: entity.Updatable{
						Base: entity.Base{
							ID:        1111,
							CreatedAt: createdAt,
						},
					},
					KernelChangefeedID: "changefeed-1111",
					BizChangefeedID:    111111,
					TiCDCClusterID:     111,
					TiCDCCluster: &entity.TiCDCCluster{
						Updatable: entity.Updatable{
							Base: entity.Base{
								ID:        111,
								CreatedAt: createdAt,
							},
						},
						Name: "ticdc-111",
					},
				},
			},
			{
				ChangefeedEntity: entity.TiCDCChangefeed{
					Updatable: entity.Updatable{
						Base: entity.Base{
							ID:        1112,
							CreatedAt: createdAt,
						},
					},
					KernelChangefeedID: "changefeed-1112",
					BizChangefeedID:    111113,
					TiCDCClusterID:     111,
				},
			},
			{
				ChangefeedEntity: entity.TiCDCChangefeed{
					Updatable: entity.Updatable{
						Base: entity.Base{
							ID:        1114,
							CreatedAt: createdAt,
						},
					},
					KernelChangefeedID: "changefeed-1114",
					BizChangefeedID:    111112,
					TiCDCClusterID:     113,
				},
			},
			{
				ChangefeedEntity: entity.TiCDCChangefeed{
					Updatable: entity.Updatable{
						Base: entity.Base{
							ID:        1115,
							CreatedAt: createdAt,
						},
					},
					KernelChangefeedID: "changefeed-1115",
					BizChangefeedID:    111112,
					TiCDCClusterID:     112,
				},
			},
		},
	}
	listCall := s.mockTiCDCChangefeedModel.
		On("ListChangefeeds", mock.Anything, &domain.ListChangefeedsReq{
			Filter: &domain.ListChangefeedFilter{
				TiCDCClusterIDs: ticdcClusterIDs,
				Deleted:         false,
			},
			Pager: &domain.Pager{
				PageSize: 1000,
			},
		}).Return(mockListResp, nil)
	defer listCall.Unset()
}

func (s *opsInnerControllerImplTestSuite) TestListTiCDCChangefeeds() {
	createdAt := time.Now()
	req := &dataflow.ListInternalChangefeedsRequest{
		TidbClusterId: 111,
	}
	s.Run("list biz changefeeds error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockBizChangefeedModel.
			On("ListBizChangefeeds", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.controller.listTiCDCChangefeeds(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "list biz changefeeds")
	})

	s.Run("biz changefeed is empty", func() {
		call := s.mockBizChangefeedModel.
			On("ListBizChangefeeds", mock.Anything, mock.Anything).
			Return(&domain.ListBizChangefeedsResp{
				BizChangefeeds: []*domain.BizChangefeed{},
			}, nil)
		defer call.Unset()
		internalChangefeeds, err := s.controller.listTiCDCChangefeeds(context.Background(), req)
		s.NoError(err)
		s.Equal(0, len(internalChangefeeds))
	})

	mockBizChangefeedsResp := &domain.ListBizChangefeedsResp{
		BizChangefeeds: []*domain.BizChangefeed{
			{
				BizChangefeedID: 111111,
				Name:            "biz-changefeed-1",
			},
			{
				BizChangefeedID: 111112,
				Name:            "biz-changefeed-2",
			},
		},
	}
	bizChangefeedCall := s.mockBizChangefeedModel.
		On("ListBizChangefeeds", mock.Anything, &domain.ListBizChangefeedsReq{
			ClusterID: 111,
			PerPage:   1000,
		}).
		Return(mockBizChangefeedsResp, nil)
	defer bizChangefeedCall.Unset()

	s.Run("get tidb cluster error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockMgmtCli.
			On("GetClusterByID", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.controller.listTiCDCChangefeeds(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "get tidb cluster")
	})

	getTiDBClusterCall := s.mockMgmtCli.
		On("GetClusterByID", mock.Anything, mock.Anything).
		Return(&mgmt.Cluster{
			ClusterId: 111,
			ClusterInfo: &mgmt.Cluster_NextGenCluster{
				NextGenCluster: &v12.Cluster{
					SharedPoolId: "1234",
				},
			},
		}, nil)
	defer getTiDBClusterCall.Unset()

	s.Run("list ticdc clusters error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCClusterModel.
			On("List", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.controller.listTiCDCChangefeeds(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "list ticdc clusters")
	})

	s.Run("ticdc cluster is empty", func() {
		call := s.mockTiCDCClusterModel.
			On("List", mock.Anything, mock.Anything).
			Return(&domain.ListTiCDCClustersResp{Items: []entity.TiCDCCluster{}}, nil)
		defer call.Unset()
		_, err := s.controller.listTiCDCChangefeeds(context.Background(), req)
		s.NoError(err)
	})

	mockTiCDCClustersResp := &domain.ListTiCDCClustersResp{
		Items: []entity.TiCDCCluster{
			{
				Updatable: entity.Updatable{
					Base: entity.Base{
						ID:        111,
						CreatedAt: createdAt,
					},
				},
				Name: "ticdc-111",
			},
			{
				Updatable: entity.Updatable{
					Base: entity.Base{
						ID:        112,
						CreatedAt: createdAt,
					},
				},
				Name: "ticdc-112",
			},
		},
	}
	ticdcClustersCall := s.mockTiCDCClusterModel.
		On("List", mock.Anything, &domain.ListTiCDCClustersReq{
			SharedPoolID: "1234",
		}).
		Return(mockTiCDCClustersResp, nil)
	defer ticdcClustersCall.Unset()

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

		_, err := s.controller.listTiCDCChangefeeds(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "list ticdc changefeeds")
	})

	mockChangefeedsResp := &domain.ListChangefeedsResp{
		Items: []domain.ListChangefeedItem{
			{
				ChangefeedEntity: entity.TiCDCChangefeed{
					Updatable: entity.Updatable{
						Base: entity.Base{
							ID:        1111,
							CreatedAt: createdAt,
						},
					},
					KernelChangefeedID: "changefeed-1111",
					BizChangefeedID:    111111,
					TiCDCClusterID:     111,
				},
			},
		},
	}
	changefeedsCall := s.mockTiCDCChangefeedModel.
		On("ListChangefeeds", mock.Anything, mock.Anything).
		Return(mockChangefeedsResp, nil)
	defer changefeedsCall.Unset()
	s.Run("list changefeeds successfully", func() {
		internalChangefeeds, err := s.controller.listTiCDCChangefeeds(context.Background(), req)
		s.NoError(err)
		s.Equal(1, len(internalChangefeeds))
		expectedInternalChangefeed := &dataflow.InternalChangefeed{
			Id:          1111,
			DisplayId:   111111,
			DisplayName: "biz-changefeed-1",
			CreatedAt:   timestamppb.New(createdAt),
			TicdcCluster: &dataflow.InternalTiCDCCluster{
				Id:        111,
				Name:      "ticdc-111",
				CreatedAt: timestamppb.New(createdAt),
			},
		}
		s.Equal(expectedInternalChangefeed, internalChangefeeds[0])
	})
}

func (s *opsInnerControllerImplTestSuite) TestListInternalChangefeeds() {
	createdAt := time.Now()

	mockBizChangefeedsResp := &domain.ListBizChangefeedsResp{
		BizChangefeeds: []*domain.BizChangefeed{
			{
				BizChangefeedID: 111111,
				Name:            "biz-changefeed-1",
			},
			{
				BizChangefeedID: 111112,
				Name:            "biz-changefeed-2",
			},
		},
	}
	bizChangefeedCall := s.mockBizChangefeedModel.
		On("ListBizChangefeeds", mock.Anything, mock.Anything).
		Return(mockBizChangefeedsResp, nil)
	defer bizChangefeedCall.Unset()

	mockTiCDCClustersResp := &domain.ListTiCDCClustersResp{
		Items: []entity.TiCDCCluster{
			{
				Updatable: entity.Updatable{
					Base: entity.Base{
						ID:        111,
						CreatedAt: createdAt,
					},
				},
				Name: "ticdc-111",
			},
			{
				Updatable: entity.Updatable{
					Base: entity.Base{
						ID:        112,
						CreatedAt: createdAt,
					},
				},
				Name: "ticdc-112",
			},
		},
	}
	ticdcClustersCall := s.mockTiCDCClusterModel.
		On("List", mock.Anything, mock.Anything).
		Return(mockTiCDCClustersResp, nil)
	defer ticdcClustersCall.Unset()

	mockChangefeedsResp := &domain.ListChangefeedsResp{
		Items: []domain.ListChangefeedItem{
			{
				ChangefeedEntity: entity.TiCDCChangefeed{
					Updatable: entity.Updatable{
						Base: entity.Base{
							ID:        1111,
							CreatedAt: createdAt,
						},
					},
					KernelChangefeedID: "changefeed-1111",
					BizChangefeedID:    111111,
					TiCDCClusterID:     111,
				},
			},
		},
	}
	ticdcChangefeedsCall := s.mockTiCDCChangefeedModel.
		On("ListChangefeeds", mock.Anything, mock.Anything).
		Return(mockChangefeedsResp, nil)
	defer ticdcChangefeedsCall.Unset()
}

func (s *opsInnerControllerImplTestSuite) TestGetTiCDCChangefeed() {
	createdAt := time.Now()
	changefeedID := uint64(1111)
	s.Run("get ticdc changefeed error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCChangefeedModel.
			On("QueryChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.controller.getTiCDCChangefeed(context.Background(), changefeedID)
		s.Error(err)
		s.Contains(err.Error(), "get ticdc changefeed")
	})

	mockQueryResp := &domain.QueryChangefeedResp{
		ChangefeedEntity: entity.TiCDCChangefeed{
			Updatable: entity.Updatable{
				Base: entity.Base{
					ID:        1111,
					CreatedAt: createdAt,
				},
			},
			KernelChangefeedID: "changefeed-1111",
			BizChangefeedID:    111111,
			TiCDCClusterID:     111,
		},
	}
	queryCall := s.mockTiCDCChangefeedModel.
		On("QueryChangefeed", mock.Anything, &domain.QueryChangefeedReq{
			Filter: &domain.ListChangefeedFilter{
				KernelChangefeedIDs: []uint64{changefeedID},
			},
		}).Return(mockQueryResp, nil)
	defer queryCall.Unset()
	s.Run("ticdc cluster is not found", func() {
		_, err := s.controller.getTiCDCChangefeed(context.Background(), changefeedID)
		s.Error(err)
		s.Contains(err.Error(), "not found")
	})

	mockBizChangefeedResp := &domain.GetBizChangefeedResp{
		BizChangefeed: &domain.BizChangefeed{
			BizChangefeedID: 111111,
			Name:            "biz-changefeed-1",
		},
	}
	bizChangefeedCall := s.mockBizChangefeedModel.
		On("GetBizChangefeed", mock.Anything, &domain.GetBizChangefeedReq{
			BizChangefeedID: 111111,
		}).
		Return(mockBizChangefeedResp, nil)
	defer bizChangefeedCall.Unset()
}

func (s *opsInnerControllerImplTestSuite) TestGetInternalChangefeed() {
	req := &dataflow.GetInternalChangefeedRequest{
		ChangefeedId: 1111,
	}

	s.Run("cdc task not found but get ticdc changefeed error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCChangefeedModel.
			On("QueryChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.controller.GetInternalChangefeed(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "get ticdc changefeed")
	})
}

func (s *opsInnerControllerImplTestSuite) TestGetInternalTiCDCCluster() {
	createdAt := time.Now()
	req := &dataflow.GetInternalTiCDCClusterRequest{
		TicdcClusterId: 111,
	}
	s.Run("query ticdc cluster error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCClusterModel.On("Query", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		_, err := s.controller.GetInternalTiCDCCluster(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "get ticdc cluster")
	})

	mockQueryResp := &domain.QueryTiCDCClusterResp{
		TiCDCClusterEntity: &entity.TiCDCCluster{
			Updatable: entity.Updatable{
				Base: entity.Base{
					ID:        111,
					CreatedAt: createdAt,
				},
			},
			Name: "ticdc-111",
		},
	}
	call := s.mockTiCDCClusterModel.
		On("Query", mock.Anything, &domain.QueryTiCDCClusterReq{
			ID: req.TicdcClusterId,
		}).Return(mockQueryResp, nil)
	defer call.Unset()
	s.Run("query ticdc cluster successfully", func() {
		internalTiCDCCluster, err := s.controller.GetInternalTiCDCCluster(context.Background(), req)
		s.NoError(err)
		expectedTiCDCCluster := &dataflow.InternalTiCDCCluster{
			Id:        111,
			Name:      "ticdc-111",
			CreatedAt: timestamppb.New(createdAt),
		}
		s.Equal(expectedTiCDCCluster, internalTiCDCCluster)
	})
}

func (s *opsInnerControllerImplTestSuite) TestApplyTiCDCClusterConfig() {
	mockCustomParams := map[string]*dataflow.ParamItem{
		"gc-ttl": {
			Value: &structpb.Value{
				Kind: &structpb.Value_NumberValue{NumberValue: 3600},
			},
		},
	}
	mockReq := &dataflow.ApplyConfigChangesReq{
		ClusterType:       dataflow.ConfigClusterType_TICDC_CLUSTER,
		ClusterId:         1111,
		ConfigurationFile: dataflow.ConfigFileType_TICDC_FILE,
		CustomParams:      mockCustomParams,
		TemplateId:        pointer.ToUint64(11111),
		Version:           pointer.ToString("v8.1.0"),
		CustomImageTag:    pointer.ToString("v8.1.0-20250318"),
	}

	s.Run("update ticdc cluster config error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockTiCDCClusterConfigManager.On("Update", mock.Anything, mock.Anything).Return(mockErr)
		defer call.Unset()
		err := s.controller.applyTiCDCClusterConfig(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "update config center")
	})

	updateCall := s.mockTiCDCClusterConfigManager.
		On("Update", mock.Anything, &domain.UpdateTiCDCClusterConfigReq{
			TiCDCClusterID:     1111,
			TiDBClusterVersion: pointer.ToString("v8.1.0"),
			TemplateID:         pointer.ToUint64(11111),
			CustomImageTag:     pointer.ToString("v8.1.0-20250318"),
			CustomParams:       mockCustomParams,
		}).Return(nil)
	defer updateCall.Unset()

	s.Run("apply ticdc cluster config error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockTiCDCClusterModel.On("ApplyConfig", mock.Anything, mock.Anything).Return(mockErr)
		defer call.Unset()
		err := s.controller.applyTiCDCClusterConfig(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "apply ticdc cluster config")
	})

	applyCall := s.mockTiCDCClusterModel.
		On("ApplyConfig", mock.Anything, &domain.ApplyTiCDCClusterConfigReq{
			TiCDCClusterID: 1111,
		}).Return(nil)
	defer applyCall.Unset()

	s.Run("apply ticdc cluster config successfully", func() {
		err := s.controller.applyTiCDCClusterConfig(context.Background(), mockReq)
		s.NoError(err)
	})
}

func (s *opsInnerControllerImplTestSuite) TestApplyTiCDCChangefeedConfig() {
	mockCustomParams := map[string]*dataflow.ParamItem{
		"memory_quota": {
			Value: &structpb.Value{
				Kind: &structpb.Value_NumberValue{NumberValue: 3600},
			},
		},
	}
	mockReq := &dataflow.ApplyConfigChangesReq{
		ClusterType:       dataflow.ConfigClusterType_CHANGEFEED_CLUSTER,
		ClusterId:         1111,
		ConfigurationFile: dataflow.ConfigFileType_CHANGEFEED_FILE,
		CustomParams:      mockCustomParams,
	}
	mockChangefeedEnt := &entity.TiCDCChangefeed{
		Updatable:      entity.Updatable{Base: entity.Base{ID: 1111}},
		DownStreamType: entity.TiCDCDownStreamTypeKafka,
	}

	s.Run("pause changefeed error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "pause error")
		call := s.mockTiCDCChangefeedModel.
			On("PauseChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.controller.applyTiCDCChangefeedConfig(context.Background(), mockReq, mockChangefeedEnt)
		s.Error(err)
		s.Contains(err.Error(), "pause changefeed")
	})

	pauseCall := s.mockTiCDCChangefeedModel.
		On("PauseChangefeed", mock.Anything, &domain.PauseChangefeedReq{ChangefeedID: 1111}).
		Return(nil, nil)
	defer pauseCall.Unset()

	resumeErrorCall := s.mockTiCDCChangefeedModel.
		On("ResumeChangefeed", mock.Anything, mock.Anything).
		Return(nil, bizerr.New(bizerr.ErrCodeUnknown, "resume error"))

	s.Run("update config center error and resume error", func() {
		call := s.mockChangefeedConfigManager.
			On("Update", mock.Anything, mock.Anything).
			Return(bizerr.New(bizerr.ErrCodeGRPC, "network error"))
		defer call.Unset()
		err := s.controller.applyTiCDCChangefeedConfig(context.Background(), mockReq, mockChangefeedEnt)
		s.Error(err)
		s.Contains(err.Error(), "update config center")
	})

	updateConfigCall := s.mockChangefeedConfigManager.
		On("Update", mock.Anything, mock.Anything).
		Return(nil)
	defer updateConfigCall.Unset()

	s.Run("update config error and resume error", func() {
		call := s.mockBizChangefeedModel.
			On("EditBizChangefeedForOps", mock.Anything, mock.Anything).
			Return(bizerr.New(bizerr.ErrCodeUnknown, "update error"))
		defer call.Unset()
		err := s.controller.applyTiCDCChangefeedConfig(context.Background(), mockReq, mockChangefeedEnt)
		s.Error(err)
		s.Contains(err.Error(), "update changefeed config")
	})

	updateCall := s.mockBizChangefeedModel.
		On("EditBizChangefeedForOps", mock.Anything, mock.Anything).
		Return(nil)
	defer updateCall.Unset()
	s.Run("update config successfully but resume error", func() {
		err := s.controller.applyTiCDCChangefeedConfig(context.Background(), mockReq, mockChangefeedEnt)
		s.Error(err)
		s.Contains(err.Error(), "resume changefeed")
	})

	resumeErrorCall.Unset()
	resumeCall := s.mockTiCDCChangefeedModel.
		On("ResumeChangefeed", mock.Anything, &domain.ResumeChangefeedReq{ChangefeedID: 1111}).
		Return(nil, nil)
	defer resumeCall.Unset()
	s.Run("update config successfully", func() {
		err := s.controller.applyTiCDCChangefeedConfig(context.Background(), mockReq, mockChangefeedEnt)
		s.NoError(err)
	})
}

func (s *opsInnerControllerImplTestSuite) TestUpdateTiCDCChangefeed() {
	mockChangefeedEnt := &entity.TiCDCChangefeed{
		Updatable:       entity.Updatable{Base: entity.Base{ID: 1111}},
		DownStreamType:  entity.TiCDCDownStreamTypeKafka,
		BizChangefeedID: 11111,
	}

	s.Run("edit biz changefeed error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockBizChangefeedModel.
			On("EditBizChangefeedForOps", mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()
		err := s.controller.updateTiCDCChangefeed(context.Background(), mockChangefeedEnt)
		s.Error(err)
		s.Contains(err.Error(), "update biz changefeed")
	})

	editBizChangefeedCall := s.mockBizChangefeedModel.
		On("EditBizChangefeedForOps", mock.Anything, uint64(11111)).
		Return(nil)
	defer editBizChangefeedCall.Unset()
	s.Run("edit biz changefeed successfully", func() {
		err := s.controller.updateTiCDCChangefeed(context.Background(), mockChangefeedEnt)
		s.NoError(err)
	})

	updateCall := s.mockTiCDCChangefeedModel.
		On("ModifyChangefeedConfig", mock.Anything, &domain.ModifyChangefeedConfigReq{
			ChangefeedID: 1111,
			Config: &domain.TiCDCChangefeedConfig{
				URIConfig: &domain.TiCDCChangefeedMySQLURIConfigData{},
				Config:    &domain.TiCDCChangefeedConfigData{ReplicaConfig: &clusterproxy.ReplicaConfig{}},
			},
		}).Return(nil, nil)
	defer updateCall.Unset()
	s.Run("update config successfully", func() {
		err := s.controller.updateTiCDCChangefeed(context.Background(), mockChangefeedEnt)
		s.NoError(err)
	})
}

func (s *opsInnerControllerImplTestSuite) TestUpdateChangefeedConfig() {
	mockCustomParams := map[string]*dataflow.ParamItem{
		"memory_quota": {
			Value: &structpb.Value{
				Kind: &structpb.Value_NumberValue{NumberValue: 3600},
			},
		},
	}
	mockReq := &dataflow.ApplyConfigChangesReq{
		ClusterType:       dataflow.ConfigClusterType_CHANGEFEED_CLUSTER,
		ClusterId:         1111,
		ConfigurationFile: dataflow.ConfigFileType_CHANGEFEED_FILE,
		CustomParams:      mockCustomParams,
	}

	s.Run("invalid configuration file", func() {
		mockReq.ConfigurationFile = dataflow.ConfigFileType_TICDC_FILE
		defer func() { mockReq.ConfigurationFile = dataflow.ConfigFileType_CHANGEFEED_FILE }()
		err := s.controller.updateChangefeedConfig(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "invalid configuration file")
	})

	s.Run("update config error", func() {
		call := s.mockChangefeedConfigManager.
			On("Update", mock.Anything, mock.Anything).
			Return(bizerr.New(bizerr.ErrCodeGRPC, "network error"))
		defer call.Unset()
		err := s.controller.updateChangefeedConfig(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "update config center")
	})

	s.Run("update changefeed config successfully", func() {
		mockReq.ConfigurationFile = dataflow.ConfigFileType_CHANGEFEED_FILE
		call := s.mockChangefeedConfigManager.
			On("Update", mock.Anything, &domain.UpdateChangefeedConfigReq{
				ChangefeedID:  1111,
				Configuration: mockCustomParams,
			}).Return(nil)
		defer call.Unset()
		err := s.controller.updateChangefeedConfig(context.Background(), mockReq)
		s.NoError(err)
	})

	s.Run("update sink uri config successfully", func() {
		mockReq.ConfigurationFile = dataflow.ConfigFileType_SINK_URI
		call := s.mockChangefeedConfigManager.
			On("Update", mock.Anything, &domain.UpdateChangefeedConfigReq{
				ChangefeedID:     1111,
				URIConfiguration: mockCustomParams,
			}).Return(nil)
		defer call.Unset()
		err := s.controller.updateChangefeedConfig(context.Background(), mockReq)
		s.NoError(err)
	})
}

func (s *opsInnerControllerImplTestSuite) TestApplyChangefeedConfig() {
	mockCustomParams := map[string]*dataflow.ParamItem{
		"memory_quota": {
			Value: &structpb.Value{
				Kind: &structpb.Value_NumberValue{NumberValue: 3600},
			},
		},
	}
	mockReq := &dataflow.ApplyConfigChangesReq{
		ClusterType:       dataflow.ConfigClusterType_CHANGEFEED_CLUSTER,
		ClusterId:         1111,
		ConfigurationFile: dataflow.ConfigFileType_CHANGEFEED_FILE,
		CustomParams:      mockCustomParams,
	}

	s.Run("query ticdc changefeed error", func() {
		call := s.mockTiCDCChangefeedModel.
			On("QueryChangefeed", mock.Anything, mock.Anything).
			Return(nil, bizerr.New(bizerr.ErrCodeDB, "network error"))
		defer call.Unset()
		err := s.controller.applyChangefeedConfig(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "get ticdc changefeed")
	})

	mockQueryResp := &domain.QueryChangefeedResp{
		ChangefeedEntity: entity.TiCDCChangefeed{
			Updatable:      entity.Updatable{Base: entity.Base{ID: 1111}},
			DownStreamType: entity.TiCDCDownStreamTypeKafka,
		},
	}
	ticdcChangefeedCall := s.mockTiCDCChangefeedModel.
		On("QueryChangefeed", mock.Anything, &domain.QueryChangefeedReq{
			Filter: &domain.ListChangefeedFilter{KernelChangefeedIDs: []uint64{1111}},
		}).Return(mockQueryResp, nil)
	defer ticdcChangefeedCall.Unset()

	s.Run("apply ticdc changefeed conifg error", func() {
		pauseCall := s.mockTiCDCChangefeedModel.
			On("PauseChangefeed", mock.Anything, mock.Anything).
			Return(nil, bizerr.New(bizerr.ErrCodeUnknown, "pause error"))
		defer pauseCall.Unset()
		err := s.controller.applyChangefeedConfig(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "apply ticdc changefeed config")
	})

	s.Run("apply ticdc changefeed config successfully", func() {
		updateConfigCenterCall := s.mockChangefeedConfigManager.
			On("Update", mock.Anything, mock.Anything).
			Return(nil)
		defer updateConfigCenterCall.Unset()
		pauseCall := s.mockTiCDCChangefeedModel.
			On("PauseChangefeed", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer pauseCall.Unset()
		updateCall := s.mockBizChangefeedModel.
			On("EditBizChangefeedForOps", mock.Anything, mock.Anything).
			Return(nil)
		defer updateCall.Unset()
		resumeCall := s.mockTiCDCChangefeedModel.
			On("ResumeChangefeed", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer resumeCall.Unset()
		err := s.controller.applyChangefeedConfig(context.Background(), mockReq)
		s.NoError(err)
	})
}

func (s *opsInnerControllerImplTestSuite) TestRestartTiCDCChangefeed() {
	mockChangefeedID := uint64(1111)
	s.Run("pause changefeed error", func() {
		call := s.mockTiCDCChangefeedModel.
			On("PauseChangefeed", mock.Anything, mock.Anything).
			Return(nil, bizerr.New(bizerr.ErrCodeUnknown, "network error"))
		defer call.Unset()
		err := s.controller.restartTiCDCChangefeed(context.Background(), mockChangefeedID)
		s.Error(err)
		s.Contains(err.Error(), "pause changefeed")
	})

	pauseCall := s.mockTiCDCChangefeedModel.
		On("PauseChangefeed", mock.Anything, &domain.PauseChangefeedReq{ChangefeedID: 1111}).
		Return(nil, nil)
	defer pauseCall.Unset()

	s.Run("resume changefeed error", func() {
		call := s.mockTiCDCChangefeedModel.
			On("ResumeChangefeed", mock.Anything, mock.Anything).
			Return(nil, bizerr.New(bizerr.ErrCodeUnknown, "network error"))
		defer call.Unset()
		err := s.controller.restartTiCDCChangefeed(context.Background(), mockChangefeedID)
		s.Error(err)
		s.Contains(err.Error(), "resume changefeed")
	})

	resumeCall := s.mockTiCDCChangefeedModel.
		On("ResumeChangefeed", mock.Anything, &domain.ResumeChangefeedReq{ChangefeedID: 1111}).
		Return(nil, nil)
	defer resumeCall.Unset()

	s.Run("restart changefeed successfully", func() {
		err := s.controller.restartTiCDCChangefeed(context.Background(), mockChangefeedID)
		s.NoError(err)
	})
}

func (s *opsInnerControllerImplTestSuite) TestRestartChangefeed() {
	mockReq := &dataflow.RestartChangefeedRequest{
		ChangefeedId: 1111,
	}

	s.Run("query ticdc changefeed error", func() {
		call := s.mockTiCDCChangefeedModel.
			On("QueryChangefeed", mock.Anything, mock.Anything).
			Return(nil, bizerr.New(bizerr.ErrCodeDB, "network error"))
		defer call.Unset()
		_, err := s.controller.RestartChangefeed(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "get ticdc changefeed")
	})

	mockQueryResp := &domain.QueryChangefeedResp{
		ChangefeedEntity: entity.TiCDCChangefeed{
			Updatable: entity.Updatable{Base: entity.Base{ID: 1111}},
		},
	}
	ticdcChangefeedCall := s.mockTiCDCChangefeedModel.
		On("QueryChangefeed", mock.Anything, &domain.QueryChangefeedReq{
			Filter: &domain.ListChangefeedFilter{KernelChangefeedIDs: []uint64{1111}},
		}).Return(mockQueryResp, nil)
	defer ticdcChangefeedCall.Unset()

	s.Run("restart ticdc changefeed error", func() {
		call := s.mockTiCDCChangefeedModel.
			On("PauseChangefeed", mock.Anything, mock.Anything).
			Return(nil, bizerr.New(bizerr.ErrCodeUnknown, "network error"))
		defer call.Unset()
		_, err := s.controller.RestartChangefeed(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "restart ticdc changefeed")
	})

	s.Run("restart ticdc changefeed successfully", func() {
		pauseCall := s.mockTiCDCChangefeedModel.
			On("PauseChangefeed", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer pauseCall.Unset()
		resumeCall := s.mockTiCDCChangefeedModel.
			On("ResumeChangefeed", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer resumeCall.Unset()
		_, err := s.controller.RestartChangefeed(context.Background(), mockReq)
		s.NoError(err)
	})
}

func (s *opsInnerControllerImplTestSuite) TestPauseChangefeed() {
	mockReq := &dataflow.PauseChangefeedRequest{
		ChangefeedId: 1111,
	}

	s.Run("query ticdc changefeed error", func() {
		call := s.mockTiCDCChangefeedModel.
			On("QueryChangefeed", mock.Anything, mock.Anything).
			Return(nil, bizerr.New(bizerr.ErrCodeDB, "network error"))
		defer call.Unset()
		_, err := s.controller.PauseChangefeed(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "get ticdc changefeed")
	})

	mockQueryResp := &domain.QueryChangefeedResp{
		ChangefeedEntity: entity.TiCDCChangefeed{
			Updatable: entity.Updatable{Base: entity.Base{ID: 1111}},
		},
	}
	ticdcChangefeedCall := s.mockTiCDCChangefeedModel.
		On("QueryChangefeed", mock.Anything, &domain.QueryChangefeedReq{
			Filter: &domain.ListChangefeedFilter{KernelChangefeedIDs: []uint64{1111}},
		}).Return(mockQueryResp, nil)
	defer ticdcChangefeedCall.Unset()

	s.Run("pause ticdc changefeed error", func() {
		call := s.mockTiCDCChangefeedModel.
			On("PauseChangefeed", mock.Anything, mock.Anything).
			Return(nil, bizerr.New(bizerr.ErrCodeUnknown, "network error"))
		defer call.Unset()
		_, err := s.controller.PauseChangefeed(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "pause changefeed")
	})

	s.Run("pause ticdc changefeed successfully", func() {
		pauseCall := s.mockTiCDCChangefeedModel.
			On("PauseChangefeed", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer pauseCall.Unset()
		_, err := s.controller.PauseChangefeed(context.Background(), mockReq)
		s.NoError(err)
	})
}

func (s *opsInnerControllerImplTestSuite) TestResumeChangefeed() {
	mockReq := &dataflow.ResumeChangefeedRequest{
		ChangefeedId: 1111,
	}

	s.Run("query ticdc changefeed error", func() {
		call := s.mockTiCDCChangefeedModel.
			On("QueryChangefeed", mock.Anything, mock.Anything).
			Return(nil, bizerr.New(bizerr.ErrCodeDB, "network error"))
		defer call.Unset()
		_, err := s.controller.ResumeChangefeed(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "get ticdc changefeed")
	})

	mockQueryResp := &domain.QueryChangefeedResp{
		ChangefeedEntity: entity.TiCDCChangefeed{
			Updatable: entity.Updatable{Base: entity.Base{ID: 1111}},
		},
	}
	ticdcChangefeedCall := s.mockTiCDCChangefeedModel.
		On("QueryChangefeed", mock.Anything, &domain.QueryChangefeedReq{
			Filter: &domain.ListChangefeedFilter{KernelChangefeedIDs: []uint64{1111}},
		}).Return(mockQueryResp, nil)
	defer ticdcChangefeedCall.Unset()

	s.Run("resume ticdc changefeed error", func() {
		call := s.mockTiCDCChangefeedModel.
			On("ResumeChangefeed", mock.Anything, mock.Anything).
			Return(nil, bizerr.New(bizerr.ErrCodeUnknown, "network error"))
		defer call.Unset()
		_, err := s.controller.ResumeChangefeed(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "resume changefeed")
	})

	s.Run("resume ticdc changefeed successfully", func() {
		resumeCall := s.mockTiCDCChangefeedModel.
			On("ResumeChangefeed", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer resumeCall.Unset()
		_, err := s.controller.ResumeChangefeed(context.Background(), mockReq)
		s.NoError(err)
	})
}

func TestOpsInnerControllerImplTestSuite(t *testing.T) {
	suite.Run(t, new(opsInnerControllerImplTestSuite))
}