package impl

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"testing"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	encryptor_mock "github.com/tidbcloud/dataflow-service/pkg/encryptor-wrapper/mock"
	regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api"
	regionalapi_mock "github.com/tidbcloud/dataflow-service/pkg/regional-api/mock"
	ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
)

type pulsarDownstreamAdaptorTestSuite struct {
	suite.Suite
	pulsarAdaptor *pulsarDownstreamAdaptor

	mockPreCheckCli *regionalapi_mock.TiCDCPreCheckClient
	mockEncryptor   *encryptor_mock.Encryptor
}

func (s *pulsarDownstreamAdaptorTestSuite) SetupSuite() {
	s.mockPreCheckCli = regionalapi_mock.NewTiCDCPreCheckClient(s.T())
	s.mockEncryptor = encryptor_mock.NewEncryptor(s.T())

	s.pulsarAdaptor = &pulsarDownstreamAdaptor{
		cdcPreCheckCli: s.mockPreCheckCli,
		encryptor:      s.mockEncryptor,
	}
}

func (s *pulsarDownstreamAdaptorTestSuite) TestParsePulsarParams() {
	testcases := []struct {
		name             string
		downstreamParams interface{}
		expectedError    bool
	}{
		{
			name:             "nil param",
			downstreamParams: nil,
			expectedError:    true,
		},
		{
			name:             "wrong type param",
			downstreamParams: &dataflow.MySQLInfo{},
			expectedError:    true,
		},
		{
			name:             "wrong type param2",
			downstreamParams: dataflow.PulsarInfo{},
			expectedError:    true,
		},
		{
			name:             "right type param",
			downstreamParams: &dataflow.PulsarInfo{},
			expectedError:    false,
		},
	}

	for _, tc := range testcases {
		s.Run(tc.name, func() {
			_, err := s.pulsarAdaptor.parsePulsarParams(tc.downstreamParams)
			if tc.expectedError {
				s.Error(err)
			} else {
				s.NoError(err)
			}
		})
	}
}

func (s *pulsarDownstreamAdaptorTestSuite) TestParsePulsarParamsStr() {
	testcases := []struct {
		name             string
		downstreamParams interface{}
		expectedError    bool
	}{
		{
			name: "wrong type param",
			downstreamParams: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "abc:3306",
				},
			},
			expectedError: true,
		},
		{
			name: "right type param",
			downstreamParams: &dataflow.PulsarInfo{
				Endpoint: "abc:9092",
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN,
					Token:    "abc",
				},
			},
			expectedError: false,
		},
	}

	for _, tc := range testcases {
		s.Run(tc.name, func() {
			paramsBytes, err := json.Marshal(tc.downstreamParams)
			s.Require().NoError(err)
			_, err = s.pulsarAdaptor.parsePulsarParamsStr(string(paramsBytes))
			if tc.expectedError {
				s.Error(err)
			} else {
				s.NoError(err)
			}
		})
	}
}

func (s *pulsarDownstreamAdaptorTestSuite) TestTestConnection() {
	cluster := &core.Cluster{
		Id:                111,
		ProjectId:         11,
		TenantId:          1,
		RegionName:        "us-west-2",
		InfraProviderType: "eks",
	}

	s.Run("downstream param error", func() {
		downstreamParams := &dataflow.MySQLInfo{}
		err := s.pulsarAdaptor.TestConnection(context.Background(), cluster, downstreamParams)
		s.Error(err)
	})

	mockDownstreamParams := &dataflow.PulsarInfo{
		DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR_SSL,
		Endpoint:            "abc:9092",
		Authentication: &dataflow.PulsarAuthentication{
			AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN,
			Token:    "abc",
		},
		AdvancedSetting: &dataflow.PulsarAdvancedSetting{
			ConnectionTimeout: 10,
		},
	}
	s.Run("test connection error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockPreCheckCli.
			On("TestNetworkConnection", mock.Anything, mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()
		err := s.pulsarAdaptor.TestConnection(context.Background(), cluster, mockDownstreamParams)
		s.Error(err)
	})

	expectedRegion := &regionalapi.Region{
		Name:          "us-west-2",
		TenantID:      1,
		CloudProvider: "aws",
	}
	expectedReq := &ticdc_v1.TestNetworkConnectionRequest{
		ClusterId: "111",
		Type:      "pulsar",
		PulsarParam: &ticdc_v1.TestNetworkConnectionRequest_PulsarParameter{
			ServiceUrl:         "pulsar+ssl://abc:9092",
			AuthenticationType: "token",
			Token:              "abc",
			ConnectionTimeout:  10,
		},
	}
	call := s.mockPreCheckCli.
		On("TestNetworkConnection", mock.Anything, expectedRegion, expectedReq).
		Return(nil)
	defer call.Unset()
	s.Run("test connection successfully", func() {
		err := s.pulsarAdaptor.TestConnection(context.Background(), cluster, mockDownstreamParams)
		s.NoError(err)
	})
}

func (s *pulsarDownstreamAdaptorTestSuite) TestPreCheckChangefeedParameters() {
	cluster := &core.Cluster{
		Id:                111,
		ProjectId:         11,
		TenantId:          1,
		RegionName:        "us-west-2",
		InfraProviderType: "eks",
	}

	s.Run("downstream param error", func() {
		downstreamParams := &dataflow.MySQLInfo{}
		err := s.pulsarAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Error(err)
	})

	s.Run("parse start tso error", func() {
		downstreamParams := &dataflow.PulsarInfo{
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_UTC,
				Utc:  time.Now().Format(time.RFC1123),
			},
		}
		err := s.pulsarAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Error(err)
	})

	downstreamParams := &dataflow.PulsarInfo{
		StartPosition: &dataflow.StartPosition{
			Mode: dataflow.StartMode_FROM_TSO,
			Tso:  123123,
		},
	}
	s.Run("pre check error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockPreCheckCli.
			On("ValidateChangefeedParameters", mock.Anything, mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()

		err := s.pulsarAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Error(err)
	})

	expectedRegion := &regionalapi.Region{
		Name:          "us-west-2",
		TenantID:      1,
		CloudProvider: "aws",
	}
	expectedReq := &ticdc_v1.ValidateChangefeedParametersRequest{
		ClusterId: "111",
		Type:      "pulsar",
		StartTso:  123123,
	}
	call := s.mockPreCheckCli.
		On("ValidateChangefeedParameters", mock.Anything, expectedRegion, expectedReq).
		Return(nil)
	defer call.Unset()
	s.Run("pre check successfully", func() {
		err := s.pulsarAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.NoError(err)
	})
}

func (s *pulsarDownstreamAdaptorTestSuite) TestEncryptPasswordInPlace() {
	s.Run("downstream param error", func() {
		downstreamParams := &dataflow.MySQLInfo{}
		err := s.pulsarAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.Error(err)
	})

	s.Run("encrypt authentication password error", func() {
		downstreamParams := &dataflow.PulsarInfo{
			Authentication: &dataflow.PulsarAuthentication{
				AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
				UserName: "user",
				Password: "abc",
			},
		}
		encryptErr := bizerr.New(bizerr.ErrCodeUnknown, "encrypt error")
		call := s.mockEncryptor.On("EncryptBase64Inplace", mock.Anything).Return(encryptErr)
		defer call.Unset()

		err := s.pulsarAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.Error(err)
		s.Contains(err.Error(), "encrypt password of authentication")
	})

	s.Run("encrypt authentication token error", func() {
		downstreamParams := &dataflow.PulsarInfo{
			Authentication: &dataflow.PulsarAuthentication{
				AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN,
				Token:    "token",
			},
		}

		encryptErr := bizerr.New(bizerr.ErrCodeUnknown, "encrypt error")
		call := s.mockEncryptor.On("EncryptBase64Inplace", mock.Anything).Return(encryptErr)
		defer call.Unset()
		err := s.pulsarAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.Error(err)
		s.Contains(err.Error(), "encrypt token of authentication")
	})

	s.Run("encrypt authentication password successfully", func() {
		downstreamParams := &dataflow.PulsarInfo{
			Authentication: &dataflow.PulsarAuthentication{
				AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
				UserName: "user",
				Password: "abc",
			},
		}
		encryptedPassword := base64.StdEncoding.EncodeToString([]byte(downstreamParams.Authentication.Password))
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", &downstreamParams.Authentication.Password).
			Return(nil).
			Run(func(args mock.Arguments) {
				password := args.Get(0).(*string)
				*password = encryptedPassword
			})
		defer call.Unset()

		err := s.pulsarAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.NoError(err)
		s.Equal(encryptedPassword, downstreamParams.Authentication.Password)
	})

	s.Run("encrypt authentication token successfully", func() {
		downstreamParams := &dataflow.PulsarInfo{
			Authentication: &dataflow.PulsarAuthentication{
				AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN,
				Token:    "token",
			},
		}

		encryptedPassword := base64.StdEncoding.EncodeToString([]byte(downstreamParams.Authentication.Token))
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", &downstreamParams.Authentication.Token).
			Return(nil).
			Run(func(args mock.Arguments) {
				password := args.Get(0).(*string)
				*password = encryptedPassword
			})
		defer call.Unset()
		err := s.pulsarAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.NoError(err)
		s.Equal(encryptedPassword, downstreamParams.Authentication.Token)
	})
}

func (s *pulsarDownstreamAdaptorTestSuite) TestCompleteEditParams() {
	encryptedPassword := base64.StdEncoding.EncodeToString([]byte("password"))
	testcases := []struct {
		name          string
		editParams    interface{}
		currentParams interface{}
		encryptErr    bool
		err           bool
		assertFun     func(pulsarInfo *dataflow.PulsarInfo)
	}{
		{
			name:          "invalid edit params",
			editParams:    nil,
			currentParams: &dataflow.PulsarInfo{},
			encryptErr:    false,
			err:           true,
		},
		{
			name:          "invalid current params",
			editParams:    &dataflow.PulsarInfo{},
			currentParams: &dataflow.MySQLInfo{Connection: &dataflow.Connection{Endpoint: "abc:3306"}},
			encryptErr:    false,
			err:           true,
		},
		{
			name: "password empty",
			editParams: &dataflow.PulsarInfo{
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					Password: "",
				},
			},
			currentParams: &dataflow.PulsarInfo{
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN,
					Token:    "token",
				},
			},
			encryptErr: false,
			err:        true,
		},
		{
			name: "token empty",
			editParams: &dataflow.PulsarInfo{
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN,
					Token:    "",
				},
			},
			currentParams: &dataflow.PulsarInfo{
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					Password: "password",
				},
			},
			encryptErr: false,
			err:        true,
		},
		{
			name: "use password of current params",
			editParams: &dataflow.PulsarInfo{
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					Password: "",
				},
			},
			currentParams: &dataflow.PulsarInfo{
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					Password: "password",
				},
			},
			encryptErr: false,
			err:        false,
			assertFun: func(pulsarInfo *dataflow.PulsarInfo) {
				s.Equal("password", pulsarInfo.Authentication.Password)
			},
		},
		{
			name: "use token of current params",
			editParams: &dataflow.PulsarInfo{
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN,
					Token:    "",
				},
			},
			currentParams: &dataflow.PulsarInfo{
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN,
					Token:    "token",
				},
			},
			encryptErr: false,
			err:        false,
			assertFun: func(pulsarInfo *dataflow.PulsarInfo) {
				s.Equal("token", pulsarInfo.Authentication.Token)
			},
		},
		{
			name: "encrypt password error",
			editParams: &dataflow.PulsarInfo{
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					Password: "password",
				},
			},
			currentParams: &dataflow.PulsarInfo{},
			encryptErr:    true,
			err:           true,
		},
		{
			name: "encrypt token error",
			editParams: &dataflow.PulsarInfo{
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN,
					Token:    "token",
				},
			},
			currentParams: &dataflow.PulsarInfo{},
			encryptErr:    true,
			err:           true,
		},
		{
			name: "encrypt password successfully",
			editParams: &dataflow.PulsarInfo{
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					Password: "password",
				},
			},
			currentParams: &dataflow.PulsarInfo{},
			encryptErr:    false,
			err:           false,
			assertFun: func(pulsarInfo *dataflow.PulsarInfo) {
				s.Equal(encryptedPassword, pulsarInfo.Authentication.Password)
			},
		},
		{
			name: "encrypt token successfully",
			editParams: &dataflow.PulsarInfo{
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN,
					Token:    "token",
				},
			},
			currentParams: &dataflow.PulsarInfo{},
			encryptErr:    false,
			err:           false,
			assertFun: func(pulsarInfo *dataflow.PulsarInfo) {
				s.Equal(encryptedPassword, pulsarInfo.Authentication.Token)
			},
		},
	}

	for _, tc := range testcases {
		s.Run(tc.name, func() {
			currentParamsBytes, err := json.Marshal(tc.currentParams)
			s.NoError(err)
			if tc.encryptErr {
				call := s.mockEncryptor.
					On("EncryptBase64Inplace", mock.Anything).
					Return(bizerr.New(bizerr.ErrCodeUnknown, "encrypt error"))
				defer call.Unset()
			} else {
				call := s.mockEncryptor.
					On("EncryptBase64Inplace", mock.Anything).
					Return(nil).
					Run(func(args mock.Arguments) {
						password := args.Get(0).(*string)
						*password = encryptedPassword
					})
				defer call.Unset()
			}

			err = s.pulsarAdaptor.CompleteEditParams(context.Background(), tc.editParams, string(currentParamsBytes))
			if tc.err {
				s.Error(err)
			} else {
				s.NoError(err)
				if tc.assertFun != nil {
					pulsarInfo, ok := tc.editParams.(*dataflow.PulsarInfo)
					s.True(ok)
					tc.assertFun(pulsarInfo)
				}
			}
		})
	}
}

func (s *pulsarDownstreamAdaptorTestSuite) TestAssembleURIConfigData() {
	pulsarInfo := &dataflow.PulsarInfo{
		DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR_SSL,
		Endpoint:            "abc:9093",
		TopicDistribution: &dataflow.PulsarTopicDistribution{
			PulsarTenant:    "tenant",
			PulsarNamespace: "namespace",
			DefaultTopic:    "topic",
		},
		DataFormat: &dataflow.PulsarDataFormat{
			Protocol:            dataflow.Protocol_CANAL_JSON,
			EnableTidbExtension: true,
		},
	}
	expectedURIData := &domain.TiCDCChangefeedPulsarURIConfigData{
		PulsarProtocol:      "pulsar+ssl",
		Endpoint:            "abc:9093",
		PulsarTenant:        "tenant",
		PulsarNamespace:     "namespace",
		DefaultTopic:        "topic",
		DataProtocol:        "canal-json",
		EnableTiDBExtension: pointer.ToBool(true),
	}
	uriData := s.pulsarAdaptor.assembleURIConfigData(pulsarInfo)
	s.Equal(expectedURIData, uriData)
}

func (s *pulsarDownstreamAdaptorTestSuite) TestAssemblePulsarConfig() {
	testcases := []struct {
		pulsarInfo     *dataflow.PulsarInfo
		expectedConfig *clusterproxy.PulsarConfig
	}{
		{
			pulsarInfo: &dataflow.PulsarInfo{
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					UserName: "user",
					Password: "password",
				},
				CompressionType: dataflow.PulsarCompressionType_PULSAR_COMPRESSION_LZ4,
				AdvancedSetting: &dataflow.PulsarAdvancedSetting{
					ConnectionTimeout:       5,
					OperationTimeout:        4,
					SendTimeout:             2,
					BatchingMaxPublishDelay: 1000,
					BatchingMaxMessages:     100,
				},
				DataFormat: &dataflow.PulsarDataFormat{
					OutputRawChangeEvent: false,
				},
			},
			expectedConfig: &clusterproxy.PulsarConfig{
				AuthenticationToken:     "",
				BasicUserName:           "user",
				BasicPassword:           "raw-string",
				CompressionType:         "lz4",
				ConnectionTimeout:       5,
				OperationTimeout:        4,
				SendTimeout:             2,
				BatchingMaxPublishDelay: 1000,
				BatchingMaxMessages:     100,
				OutputRawChangeEvent:    pointer.ToBool(false),
			},
		},
		{
			pulsarInfo: &dataflow.PulsarInfo{
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN,
					Token:    "token",
				},
				CompressionType: dataflow.PulsarCompressionType_PULSAR_COMPRESSION_LZ4,
				AdvancedSetting: &dataflow.PulsarAdvancedSetting{
					ConnectionTimeout:       5,
					OperationTimeout:        4,
					SendTimeout:             2,
					BatchingMaxPublishDelay: 1000,
					BatchingMaxMessages:     100,
				},
				DataFormat: &dataflow.PulsarDataFormat{
					OutputRawChangeEvent: true,
				},
			},
			expectedConfig: &clusterproxy.PulsarConfig{
				AuthenticationToken:     "raw-string",
				BasicUserName:           "",
				BasicPassword:           "",
				CompressionType:         "lz4",
				ConnectionTimeout:       5,
				OperationTimeout:        4,
				SendTimeout:             2,
				BatchingMaxPublishDelay: 1000,
				BatchingMaxMessages:     100,
				OutputRawChangeEvent:    pointer.ToBool(true),
			},
		},
	}

	call := s.mockEncryptor.On("DecryptBase64", mock.Anything).Return("raw-string", nil)
	defer call.Unset()
	for _, tc := range testcases {
		pulsarConfig, err := s.pulsarAdaptor.assemblePulsarConfig(tc.pulsarInfo)
		s.NoError(err)
		s.Equal(tc.expectedConfig, pulsarConfig)
	}
}

func (s *pulsarDownstreamAdaptorTestSuite) TestGeneratePulsarTopicExpression() {
	testcases := []struct {
		distribution *dataflow.PulsarTopicDistribution
		expectedExpr string
	}{
		{
			distribution: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "namespace",
				DefaultTopic:    "topic",
				DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
			},
			expectedExpr: "",
		},
		{
			distribution: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "namespace",
				DefaultTopic:    "topic",
				DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_BY_DATABASE,
				TopicPrefix:     "prefix",
				Separator:       "sep",
				TopicSuffix:     "suffix",
			},
			expectedExpr: "persistent://tenant/namespace/prefix{schema}suffix",
		},
		{
			distribution: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "namespace",
				DefaultTopic:    "topic",
				DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
				TopicPrefix:     "prefix",
				Separator:       "sep",
				TopicSuffix:     "suffix",
			},
			expectedExpr: "persistent://tenant/namespace/prefix{schema}sep{table}suffix",
		},
	}

	for _, tc := range testcases {
		topicExpr := s.pulsarAdaptor.generatePulsarTopicExpression(tc.distribution)
		s.Equal(tc.expectedExpr, topicExpr)
	}
}

func (s *pulsarDownstreamAdaptorTestSuite) TestAssemblePulsarDispatchers() {
	testcases := []struct {
		td            *dataflow.PulsarTopicDistribution
		pds           []*dataflow.PartitionDispatcherConfig
		expectedRules []*clusterproxy.DispatchRuleV2
	}{
		{
			td: &dataflow.PulsarTopicDistribution{
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
			},
			pds: []*dataflow.PartitionDispatcherConfig{
				{
					Matcher:       []string{"*.*"},
					PartitionType: dataflow.PartitionDispatcher_TABLE,
				},
			},
			expectedRules: []*clusterproxy.DispatchRuleV2{
				{
					Matcher:   []string{"*.*"},
					Topic:     "",
					Partition: "table",
				},
			},
		},
		{
			td: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "namespace",
				DefaultTopic:    "topic",
				DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
				TopicPrefix:     "prefix",
				Separator:       "sep",
				TopicSuffix:     "suffix",
			},
			pds: []*dataflow.PartitionDispatcherConfig{
				{
					Matcher:       []string{"*.*"},
					PartitionType: dataflow.PartitionDispatcher_TS,
				},
			},
			expectedRules: []*clusterproxy.DispatchRuleV2{
				{
					Matcher:   []string{"*.*"},
					Topic:     "persistent://tenant/namespace/prefix{schema}sep{table}suffix",
					Partition: "ts",
				},
			},
		},
		{
			td: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "namespace",
				DefaultTopic:    "topic",
				DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
				TopicPrefix:     "prefix",
				Separator:       "sep",
				TopicSuffix:     "suffix",
			},
			pds: []*dataflow.PartitionDispatcherConfig{
				{
					Matcher:       []string{"a.*"},
					PartitionType: dataflow.PartitionDispatcher_TS,
				},
				{
					Matcher:       []string{"b.*"},
					PartitionType: dataflow.PartitionDispatcher_INDEX_VALUE,
					IndexName:     "index",
				},
				{
					Matcher:       []string{"*.*"},
					PartitionType: dataflow.PartitionDispatcher_TABLE,
				},
			},
			expectedRules: []*clusterproxy.DispatchRuleV2{
				{
					Matcher:   []string{"a.*"},
					Topic:     "persistent://tenant/namespace/prefix{schema}sep{table}suffix",
					Partition: "ts",
				},
				{
					Matcher:   []string{"b.*"},
					Topic:     "persistent://tenant/namespace/prefix{schema}sep{table}suffix",
					Partition: "index-value",
					IndexName: "index",
				},
				{
					Matcher:   []string{"*.*"},
					Topic:     "persistent://tenant/namespace/prefix{schema}sep{table}suffix",
					Partition: "table",
				},
			},
		},
	}

	for _, tc := range testcases {
		rules := s.pulsarAdaptor.assemblePulsarDispatchers(tc.td, tc.pds)
		s.Equal(tc.expectedRules, rules)
	}
}

func (s *pulsarDownstreamAdaptorTestSuite) TestAssembleChangefeedConfigData() {
	s.Run("parse star tso error", func() {
		pulsarInfo := &dataflow.PulsarInfo{
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_UTC,
				Utc:  time.Now().Format(time.RFC1123),
			},
		}
		_, err := s.pulsarAdaptor.assembleChangefeedConfigData(pulsarInfo)
		s.Error(err)
	})

	s.Run("assemble config data successfully", func() {
		call := s.mockEncryptor.On("DecryptBase64", mock.Anything).Return("raw-string", nil)
		defer call.Unset()

		pulsarInfo := &dataflow.PulsarInfo{
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_TSO,
				Tso:  123123,
			},
			DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR_SSL,
			Endpoint:            "abc:9093",
			CompressionType:     dataflow.PulsarCompressionType_PULSAR_COMPRESSION_LZ4,
			Authentication: &dataflow.PulsarAuthentication{
				AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
				UserName: "user",
				Password: "password",
			},
			TableConfig: &dataflow.TableConfig{
				Mode:       dataflow.TableMode_FORCE_SYNC,
				FilterRule: []string{"*.*"},
				EventFilterRule: []*dataflow.EventFilterRule{
					{
						Matcher:     []string{"*.*"},
						IgnoreEvent: []string{"delete"},
					},
				},
			},
			DataFormat: &dataflow.PulsarDataFormat{
				Protocol:             dataflow.Protocol_CANAL_JSON,
				EnableTidbExtension:  true,
				OutputRawChangeEvent: true,
			},
			TopicDistribution: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "namespace",
				DefaultTopic:    "topic",
				DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
			},
			PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
				{
					PartitionType: dataflow.PartitionDispatcher_TS,
					Matcher:       []string{"*.*"},
				},
			},
			AdvancedSetting: &dataflow.PulsarAdvancedSetting{
				ConnectionTimeout:       5,
				OperationTimeout:        4,
				SendTimeout:             3,
				BatchingMaxMessages:     100,
				BatchingMaxPublishDelay: 1000,
			},
		}
		expectedConfigData := &domain.TiCDCChangefeedConfigData{
			StartTs: pointer.ToUint64(123123),
			ReplicaConfig: &clusterproxy.ReplicaConfig{
				CaseSensitive:         pointer.ToBool(false),
				ForceReplicate:        pointer.ToBool(true),
				IgnoreIneligibleTable: pointer.ToBool(false),
				Filter: &clusterproxy.FilterConfig{
					Rules: []string{"*.*"},
					EventFilters: []clusterproxy.EventFilterRule{
						{
							Matcher:     []string{"*.*"},
							IgnoreEvent: []string{"delete"},
						},
					},
				},
				Sink: &clusterproxy.SinkConfigV2{
					DispatchRules: []*clusterproxy.DispatchRuleV2{
						{
							Matcher:   []string{"*.*"},
							Topic:     "",
							Partition: "ts",
						},
					},
					PulsarConfig: &clusterproxy.PulsarConfig{
						AuthenticationToken:     "",
						BasicUserName:           "user",
						BasicPassword:           "raw-string",
						CompressionType:         "lz4",
						ConnectionTimeout:       5,
						OperationTimeout:        4,
						SendTimeout:             3,
						BatchingMaxMessages:     100,
						BatchingMaxPublishDelay: 1000,
						OutputRawChangeEvent:    pointer.ToBool(true),
					},
				},
			},
		}

		changefeedConfigData, err := s.pulsarAdaptor.assembleChangefeedConfigData(pulsarInfo)
		s.NoError(err)
		s.Equal(expectedConfigData, changefeedConfigData)
	})
}

func (s *pulsarDownstreamAdaptorTestSuite) TestAssembleChangefeedConfig() {
	mockCluster := &core.Cluster{}
	s.Run("downstream param error", func() {
		downstreamParams := &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Endpoint: "abc:3306",
			},
		}
		downstreamParamsBytes, err := json.Marshal(downstreamParams)
		s.NoError(err)
		_, err = s.pulsarAdaptor.AssembleChangefeedConfig(context.Background(), mockCluster, &entity.BizChangefeed{DownstreamParams: string(downstreamParamsBytes)})
		s.Error(err)
		s.Contains(err.Error(), "unmarshal pulsar params error")
	})

	s.Run("assemble changefeed config", func() {
		downstreamParams := &dataflow.PulsarInfo{
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_UTC,
				Utc:  time.Now().Format(time.RFC1123),
			},
		}
		downstreamParamsBytes, err := json.Marshal(downstreamParams)
		s.NoError(err)
		_, err = s.pulsarAdaptor.AssembleChangefeedConfig(context.Background(), mockCluster, &entity.BizChangefeed{DownstreamParams: string(downstreamParamsBytes)})
		s.Error(err)
		s.Contains(err.Error(), "assemble changefeed config")
	})

	s.Run("assemble successfully", func() {
		call := s.mockEncryptor.On("DecryptBase64", mock.Anything).Return("raw-string", nil)
		defer call.Unset()

		downstreamParams := &dataflow.PulsarInfo{
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_TSO,
				Tso:  123123,
			},
			DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR_SSL,
			Endpoint:            "abc:9093",
			CompressionType:     dataflow.PulsarCompressionType_PULSAR_COMPRESSION_LZ4,
			Authentication: &dataflow.PulsarAuthentication{
				AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
				UserName: "user",
				Password: "password",
			},
			TableConfig: &dataflow.TableConfig{
				Mode:       dataflow.TableMode_FORCE_SYNC,
				FilterRule: []string{"*.*"},
				EventFilterRule: []*dataflow.EventFilterRule{
					{
						Matcher:     []string{"*.*"},
						IgnoreEvent: []string{"delete"},
					},
				},
				CaseSensitive: true,
			},
			DataFormat: &dataflow.PulsarDataFormat{
				Protocol:            dataflow.Protocol_CANAL_JSON,
				EnableTidbExtension: true,
			},
			TopicDistribution: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "namespace",
				DefaultTopic:    "topic",
				DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
			},
			PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
				{
					PartitionType: dataflow.PartitionDispatcher_TS,
					Matcher:       []string{"*.*"},
				},
			},
			AdvancedSetting: &dataflow.PulsarAdvancedSetting{
				ConnectionTimeout:       5,
				OperationTimeout:        4,
				SendTimeout:             3,
				BatchingMaxMessages:     100,
				BatchingMaxPublishDelay: 1000,
			},
		}
		expectedChangefeedConfig := &domain.TiCDCChangefeedConfig{
			URIConfig: &domain.TiCDCChangefeedPulsarURIConfigData{
				PulsarProtocol:      "pulsar+ssl",
				Endpoint:            "abc:9093",
				PulsarTenant:        "tenant",
				PulsarNamespace:     "namespace",
				DefaultTopic:        "topic",
				DataProtocol:        "canal-json",
				EnableTiDBExtension: pointer.ToBool(true),
			},
			Config: &domain.TiCDCChangefeedConfigData{
				StartTs: pointer.ToUint64(123123),
				ReplicaConfig: &clusterproxy.ReplicaConfig{
					CaseSensitive:         pointer.ToBool(true),
					ForceReplicate:        pointer.ToBool(true),
					IgnoreIneligibleTable: pointer.ToBool(false),
					Filter: &clusterproxy.FilterConfig{
						Rules: []string{"*.*"},
						EventFilters: []clusterproxy.EventFilterRule{
							{
								Matcher:     []string{"*.*"},
								IgnoreEvent: []string{"delete"},
							},
						},
					},
					Sink: &clusterproxy.SinkConfigV2{
						DispatchRules: []*clusterproxy.DispatchRuleV2{
							{
								Matcher:   []string{"*.*"},
								Topic:     "",
								Partition: "ts",
							},
						},
						PulsarConfig: &clusterproxy.PulsarConfig{
							AuthenticationToken:     "",
							BasicUserName:           "user",
							BasicPassword:           "raw-string",
							CompressionType:         "lz4",
							ConnectionTimeout:       5,
							OperationTimeout:        4,
							SendTimeout:             3,
							BatchingMaxMessages:     100,
							BatchingMaxPublishDelay: 1000,
							OutputRawChangeEvent:    pointer.ToBool(false),
						},
					},
				},
			},
		}

		downstreamParamsBytes, err := json.Marshal(downstreamParams)
		s.NoError(err)
		changefeedConfig, err := s.pulsarAdaptor.AssembleChangefeedConfig(
			context.Background(), mockCluster, &entity.BizChangefeed{DownstreamParams: string(downstreamParamsBytes)})
		s.NoError(err)
		s.Equal(expectedChangefeedConfig, changefeedConfig)
	})
}

func TestPulsarDownstreamAdaptorTestSuite(t *testing.T) {
	suite.Run(t, new(pulsarDownstreamAdaptorTestSuite))
}