package impl

import (
	"context"
	"testing"

	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/config-center/proto/config_center"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	grpc_mock "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli/mock"
	"github.com/tidbcloud/pkgv2/bizerr"
)

type changefeedConfigCenterTestSuite struct {
	suite.Suite
	mockConfigCenterGrpcCli *grpc_mock.ConfigCenterGrpcClient
	configManager           *changefeedConfigCenterImpl
}

func (s *changefeedConfigCenterTestSuite) SetupSuite() {
	s.mockConfigCenterGrpcCli = grpc_mock.NewConfigCenterGrpcClient(s.T())
	s.configManager = &changefeedConfigCenterImpl{
		configCenterGrpcCli: s.mockConfigCenterGrpcCli,
	}
}

func (s *changefeedConfigCenterTestSuite) TestCreateChangefeedConfig() {
	mockChangefeedID := "1111"
	mockConfigType := "changefeed-kafka"
	s.Run("search template error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockConfigCenterGrpcCli.
			On("SearchDefaultConfigurationTemplate", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.configManager.createChangefeedConfig(
			context.Background(), mockChangefeedID, mockConfigType, changefeedConfigFileName)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "search configuration template")
	})

	mockTemplate := &config_center.ConfigurationTemplate{
		TemplateId: 1111,
	}
	templateCall := s.mockConfigCenterGrpcCli.
		On("SearchDefaultConfigurationTemplate",
			mock.Anything, &config_center.SearchDefaultConfigurationTemplateRequest{
				ClusterType:       config_center.ClusterType_CHANGEFEED,
				ConfigurationType: mockConfigType,
			}).
		Return(mockTemplate, nil)
	defer templateCall.Unset()

	s.Run("create config error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockConfigCenterGrpcCli.
			On("UpdateClusterMeta", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.configManager.createChangefeedConfig(
			context.Background(), mockChangefeedID, mockConfigType, changefeedConfigFileName)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "create configuration")
	})

	updateCall := s.mockConfigCenterGrpcCli.
		On("UpdateClusterMeta", mock.Anything, mock.Anything).
		Return(nil, nil)
	defer updateCall.Unset()

	s.Run("create config successfully", func() {
		err := s.configManager.createChangefeedConfig(
			context.Background(), mockChangefeedID, mockConfigType, changefeedConfigFileName)
		s.NoError(err)

		reqMatcher := mock.MatchedBy(func(req *config_center.UpdateClusterMetaRequest) bool {
			return req.ClusterType == config_center.ClusterType_CHANGEFEED && req.ClusterId == "1111" &&
				len(req.ConfigurationFileRequests) == 1 && len(req.ComponentRequests) == 0 &&
				req.ConfigurationFileRequests[0].ConfigurationFilename == changefeedConfigFileName &&
				*req.ConfigurationFileRequests[0].TemplateId == mockTemplate.TemplateId
		})
		s.mockConfigCenterGrpcCli.AssertCalled(s.T(), "UpdateClusterMeta", mock.Anything, reqMatcher)
	})
}

func (s *changefeedConfigCenterTestSuite) TestCreate() {
	mockTemplate := &config_center.ConfigurationTemplate{}
	templateCall := s.mockConfigCenterGrpcCli.
		On("SearchDefaultConfigurationTemplate", mock.Anything, mock.Anything).
		Return(mockTemplate, nil)
	defer templateCall.Unset()

	updateCall := s.mockConfigCenterGrpcCli.
		On("UpdateClusterMeta", mock.Anything, mock.Anything).
		Return(nil, nil)
	defer updateCall.Unset()

	mockReq := &domain.CreateChangefeedConfigReq{
		ChangefeedID:   1111,
		DownstreamType: entity.TiCDCDownStreamTypeKafka,
	}
	_, err := s.configManager.Create(context.Background(), mockReq)
	s.NoError(err)

	configTemplateReqMatcher := mock.MatchedBy(func(req *config_center.SearchDefaultConfigurationTemplateRequest) bool {
		return req.ConfigurationType == "changefeed-kafka"
	})
	uriTemplateReqMatcher := mock.MatchedBy(func(req *config_center.SearchDefaultConfigurationTemplateRequest) bool {
		return req.ConfigurationType == "changefeed-uri-kafka"
	})
	configUpdateReqMatcher := mock.MatchedBy(func(req *config_center.UpdateClusterMetaRequest) bool {
		return len(req.ConfigurationFileRequests) == 1 &&
			req.ConfigurationFileRequests[0].ConfigurationFilename == "changefeed"
	})
	uriUpdateReqMatcher := mock.MatchedBy(func(req *config_center.UpdateClusterMetaRequest) bool {
		return len(req.ConfigurationFileRequests) == 1 &&
			req.ConfigurationFileRequests[0].ConfigurationFilename == "changefeed-uri"
	})

	s.mockConfigCenterGrpcCli.AssertCalled(
		s.T(), "SearchDefaultConfigurationTemplate", mock.Anything, configTemplateReqMatcher)
	s.mockConfigCenterGrpcCli.AssertCalled(
		s.T(), "UpdateClusterMeta", mock.Anything, configUpdateReqMatcher)

	s.mockConfigCenterGrpcCli.AssertCalled(
		s.T(), "SearchDefaultConfigurationTemplate", mock.Anything, uriTemplateReqMatcher)
	s.mockConfigCenterGrpcCli.AssertCalled(
		s.T(), "UpdateClusterMeta", mock.Anything, uriUpdateReqMatcher)
}

func (s *changefeedConfigCenterTestSuite) TestUpdateChangefeedConfig() {
	mockChangefeedID := "1111"
	mockConfigFileName := "changefeed"
	mockCustomParams := map[string]*dataflow.ParamItem{}

	s.Run("update config error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockConfigCenterGrpcCli.
			On("UpdateClusterMeta", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.configManager.updateChangefeedConfig(
			context.Background(), mockChangefeedID, mockConfigFileName, mockCustomParams)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "update configuration")
	})

	updateCall := s.mockConfigCenterGrpcCli.
		On("UpdateClusterMeta", mock.Anything, &config_center.UpdateClusterMetaRequest{
			ClusterType: config_center.ClusterType_CHANGEFEED,
			ClusterId:   "1111",
			ConfigurationFileRequests: []*config_center.UpdateClusterConfigurationFileRequest{
				{
					ConfigurationFilename: "changefeed",
					CustomParams:          map[string]*config_center.ParamItem{},
				},
			},
		}).Return(nil, nil)
	defer updateCall.Unset()

	s.Run("update config successfully", func() {
		err := s.configManager.updateChangefeedConfig(
			context.Background(), mockChangefeedID, mockConfigFileName, mockCustomParams)
		s.NoError(err)
	})
}

func (s *changefeedConfigCenterTestSuite) TestUpdate() {
	mockReq := &domain.UpdateChangefeedConfigReq{
		ChangefeedID: 1111,
	}

	err := s.configManager.Update(context.Background(), mockReq)
	s.NoError(err)

	mockReq.Configuration = map[string]*dataflow.ParamItem{
		"a": {
			Comment: "abc",
		},
	}
	mockReq.URIConfiguration = map[string]*dataflow.ParamItem{}
	call := s.mockConfigCenterGrpcCli.
		On("UpdateClusterMeta", mock.Anything, mock.Anything).
		Return(nil, nil)
	defer call.Unset()

	err = s.configManager.Update(context.Background(), mockReq)
	s.NoError(err)

	configReqMatcher := mock.MatchedBy(func(req *config_center.UpdateClusterMetaRequest) bool {
		return len(req.ConfigurationFileRequests) == 1 &&
			req.ConfigurationFileRequests[0].ConfigurationFilename == "changefeed"
	})
	uriReqMatcher := mock.MatchedBy(func(req *config_center.UpdateClusterMetaRequest) bool {
		return len(req.ConfigurationFileRequests) == 1 &&
			req.ConfigurationFileRequests[0].ConfigurationFilename == "changefeed-uri"
	})
	s.mockConfigCenterGrpcCli.AssertCalled(s.T(), "UpdateClusterMeta", mock.Anything, configReqMatcher)
	s.mockConfigCenterGrpcCli.AssertCalled(s.T(), "UpdateClusterMeta", mock.Anything, uriReqMatcher)
}

func (s *changefeedConfigCenterTestSuite) TestGetChangefeedConfig() {
	mockChangefeedID := "1111"
	mockConfigFileName := "changefeed"

	s.Run("get config error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockConfigCenterGrpcCli.
			On("GetClusterConfigurationFile", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.configManager.getChangefeedConfig(context.Background(), mockChangefeedID, mockConfigFileName)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "get configuration")
	})

	mockConfig := &config_center.ConfigurationFile{
		Text: `{"a":123,"b":"abc"}`,
	}
	getCall := s.mockConfigCenterGrpcCli.
		On("GetClusterConfigurationFile", mock.Anything, &config_center.GetClusterConfigurationFileRequest{
			ClusterType:           config_center.ClusterType_CHANGEFEED,
			ClusterId:             "1111",
			ConfigurationFilename: "changefeed",
		}).Return(mockConfig, nil)
	defer getCall.Unset()

	s.Run("get config successfully", func() {
		configData, err := s.configManager.getChangefeedConfig(
			context.Background(), mockChangefeedID, mockConfigFileName)
		s.NoError(err)
		s.Equal(mockConfig.Text, configData)
	})
}

func (s *changefeedConfigCenterTestSuite) TestGet() {
	mockReq := &domain.GetChangefeedConfigReq{
		ChangefeedID: 1111,
	}
	mockConfig := &config_center.ConfigurationFile{
		Text: `{"a":123,"b":"abc"}`,
	}
	call := s.mockConfigCenterGrpcCli.
		On("GetClusterConfigurationFile", mock.Anything, mock.Anything).
		Return(mockConfig, nil)
	defer call.Unset()

	resp, err := s.configManager.Get(context.Background(), mockReq)
	s.NoError(err)
	s.Equal(mockConfig.Text, resp.Configuration)
	s.Equal(mockConfig.Text, resp.URIConfiguration)

	configReqMatcher := mock.MatchedBy(func(req *config_center.GetClusterConfigurationFileRequest) bool {
		return req.ConfigurationFilename == "changefeed"
	})
	uriReqMatcher := mock.MatchedBy(func(req *config_center.GetClusterConfigurationFileRequest) bool {
		return req.ConfigurationFilename == "changefeed-uri"
	})
	s.mockConfigCenterGrpcCli.AssertCalled(
		s.T(), "GetClusterConfigurationFile", mock.Anything, configReqMatcher)
	s.mockConfigCenterGrpcCli.AssertCalled(
		s.T(), "GetClusterConfigurationFile", mock.Anything, uriReqMatcher)
}

func TestChangefeedConfigCenterSuite(t *testing.T) {
	suite.Run(t, new(changefeedConfigCenterTestSuite))
}