package changefeed

import (
	"fmt"
	"net/url"
	"time"

	"github.com/elliotchance/pie/v2"
	. "github.com/tidbcloud/dataflow-service-e2e/cases"
	"github.com/tidbcloud/dataflow-service-e2e/cases/dedicated/common"
	"github.com/tidbcloud/dataflow-service-e2e/internal/config"
	. "github.com/tidbcloud/dataflow-service-e2e/internal/model"
	"github.com/tidbcloud/dataflow-service-e2e/internal/sample_data"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/cdcapi"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/cdc/client/connector_service"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/cdc/model"
	ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
)

const (
	CompressionTypeLZ4  = "lz4"
	CompressionTypeZLIB = "zlib"
	CompressionTypeZSTD = "zstd"

	PulsarProtocolPulsar      = "pulsar"
	PulsarProtocolPulsarSSL   = "pulsar+ssl"
	PulsarProtocolPulsarHTTP  = "pulsar+http"
	PulsarProtocolPulsarHTTPS = "pulsar+https"
)

type pulsarChangefeedSmokeCase struct {
	BaseCase
	cdcApiWrapper *cdcapi.CDCClientWrapper
	cfValidator   *common.ChangfeedValidator
	cloudProvider CloudProviderType
}

func NewPulsarChangefeedSmokeCase(cloudProvider CloudProviderType) E2ECase {
	return &pulsarChangefeedSmokeCase{cloudProvider: cloudProvider}
}

func (m *pulsarChangefeedSmokeCase) Name() string {
	return fmt.Sprintf("%s%s", m.cloudProvider, "PulsarChangefeedSmoke")
}

func (m *pulsarChangefeedSmokeCase) DeclareDependencyRequirements() map[string]DependencyRequirement {
	return map[string]DependencyRequirement{
		"tidb": &TiDBDedicatedRequirement{
			Dedicated: false,
			TiDBDedicatedDimensions: TiDBDedicatedDimensions{
				CloudProvider: m.cloudProvider,
				Region:        config.GlobalConfig.DefaultRegionOf(m.cloudProvider),
			},
		},
		"pulsar": &PulsarRequirement{
			PulsarDimensions: PulsarDimensions{
				NetworkType: DataflowNetworkTypeNetworkTypePublic,
			},
		},
	}
}

func (m *pulsarChangefeedSmokeCase) Run() {
	tidbCluster := m.TiDBDedicated("tidb")
	downstream := m.Pulsar("pulsar")
	changefeedName := m.Name()
	instanceType := "2rcu"
	dbName := m.Name()
	tableFilter := []string{fmt.Sprintf("%s.*", dbName)}

	var (
		changefeedID                     string
		createPulsarInfo, editPulsarInfo *model.DataflowPulsarInfo
	)

	// init reusable api client and validator
	m.cdcApiWrapper = cdcapi.NewCDCClientWrapper(m.Logger())
	m.cfValidator = common.NewChangfeedValidator(m.Logger())
	sampleDataWriter := sample_data.NewSampleDataWriter(m.Logger(), tidbCluster.MysqlParameter(), dbName, 3*time.Second)

	// test create request validation
	invalidCreateReqOpts := []pulsarChangefeedOpt{
		pulsarChangefeedWithInvalidTableFilter,
		pulsarChangefeedWithInvalidEventFilter,
	}

	for _, opt := range invalidCreateReqOpts {
		m.Expect("create pulsar changefeed with invalid params", func() error {
			req := genCreatePulsarChangefeedReqWithOpt(changefeedName, instanceType, dbName, downstream, opt)
			_, err := m.cdcApiWrapper.CreateChangefeed(tidbCluster, *req)
			return ExpectError(err)
		}, 3, time.Second*10)
	}

	// cleanup changefeeds if exists
	defer func() {
		m.
			Step("stop writing data", func() error {
				sampleDataWriter.StopWriting()
				return nil
			}).
			RetryDo("delete changefeeds", func() error {
				return m.cfValidator.DeleteChangefeedByName(tidbCluster, changefeedName)
			}, 3, time.Minute).
			Expect("changefeeds not in list", func() error {
				return m.cfValidator.AllChangefeedsNotInListByNames(tidbCluster, changefeedName)
			}, 3, time.Second*10)
	}()

	m.
		RetryDo(fmt.Sprintf("test connection to downstream %s", downstream.Endpoint), func() (err error) {
			return m.cdcApiWrapper.TestConnection(tidbCluster, downstream)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("init database and table in %s", tidbCluster.DisplayName),
			sampleDataWriter.InitDBTable, 3, time.Minute).
		RetryDo(fmt.Sprintf("verify tables with tableFilter %v", tableFilter), func() (err error) {
			return m.cfValidator.VerifyTables(tidbCluster, tableFilter, []string{fmt.Sprintf("%s.%s", dbName, sample_data.SAMPLE_TABLE)}, nil)
		}, 3, time.Second*10).
		RetryDo(fmt.Sprintf("create changefeed %s", changefeedName), func() (err error) {
			createReq := genCreatePulsarChangefeedReq(changefeedName, instanceType, dbName, downstream)
			resp, err := m.cdcApiWrapper.CreateChangefeed(tidbCluster, *createReq)
			if err == nil {
				changefeedID = *resp.Payload.ConnectorID
				createPulsarInfo = createReq.PulsarInfo
			}
			return err
		}, 3, time.Minute).
		Expect("changefeed in list", func() error {
			return m.cfValidator.AreChangefeedsInList(tidbCluster, changefeedID)
		}, 3, time.Second*10).
		Expect("changefeeed is running", func() error {
			return m.cfValidator.AreChangefeedsRunning(tidbCluster, changefeedID)
		}, 20, time.Minute).
		Expect("compare changefeed kernel config after create", func() error {
			cfg, err := m.cdcApiWrapper.GetChangefeedKernelConfig(tidbCluster, changefeedID)
			if err != nil {
				return err
			}
			return comparePulsarChangefeedKernelConfig(createPulsarInfo, cfg)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("start write sample data to database %s", dbName),
			sampleDataWriter.StartWriting, 3, time.Minute).
		Step("verify changefeed's CheckpointTs advance normal", func() error {
			return m.cfValidator.VerifyCheckpointTsAdvanceNormal(5*time.Minute, tidbCluster, changefeedID)
		}).
		RetryDo(fmt.Sprintf("pause changefeed %s", changefeedName), func() error {
			return m.cdcApiWrapper.PauseChangefeed(tidbCluster, changefeedID)
		}, 3, time.Minute).
		Still(fmt.Sprintf("changefeed %s is paused", changefeedName), func() error {
			return m.cfValidator.IsChangefeedInStatus(tidbCluster, changefeedID, model.DataflowConnectorStatusPAUSED)
		}, 3, time.Minute)

	// test edit request validation
	invalidEditReqOpts := []pulsarChangefeedOpt{
		pulsarChangefeedWithInvalidTableFilter,
		pulsarChangefeedWithInvalidEventFilter,
	}
	for _, opt := range invalidEditReqOpts {
		m.Expect("edit pulsar changefeed with invalid params", func() error {
			changefeedInfo, err := m.cdcApiWrapper.GetChangefeed(tidbCluster, changefeedID)
			pulsarInfo := changefeedInfo.Payload.Request.PulsarInfo
			pulsarInfo.Authentication.Password = downstream.Password
			pulsarInfo.Authentication.Token = downstream.Token
			opt(pulsarInfo)
			editReq := connector_service.EditConnectorBody{
				ConnectorID:   changefeedInfo.Payload.ConnectorID,
				ConnectorType: changefeedInfo.Payload.Request.ConnectorType,
				Name:          changefeedInfo.Payload.Name,
				PulsarInfo:    pulsarInfo,
			}
			err = m.cdcApiWrapper.EditChangefeed(tidbCluster, changefeedID, editReq)
			return ExpectError(err)
		}, 3, time.Second*10)
	}

	m.
		RetryDo(fmt.Sprintf("edit changefeed %s without password", changefeedName), func() error {
			pulsarInfo, err := m.editChangefeed(tidbCluster, changefeedID, downstream, false)
			if err == nil {
				editPulsarInfo = pulsarInfo
			}
			return err
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("edit changefeed %s with password", changefeedName), func() error {
			pulsarInfo, err := m.editChangefeed(tidbCluster, changefeedID, downstream, true)
			if err == nil {
				editPulsarInfo = pulsarInfo
			}
			return err
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("resume changefeed %s", changefeedName), func() error {
			return m.cdcApiWrapper.ResumeChangefeed(tidbCluster, changefeedID)
		}, 3, time.Minute).
		Still(fmt.Sprintf("changefeed %s is runining", changefeedName), func() error {
			return m.cfValidator.AreChangefeedsRunning(tidbCluster, changefeedID)
		}, 3, time.Minute).
		Expect("compare changefeed kernel config after edit", func() error {
			cfg, err := m.cdcApiWrapper.GetChangefeedKernelConfig(tidbCluster, changefeedID)
			if err != nil {
				return err
			}
			return comparePulsarChangefeedKernelConfig(editPulsarInfo, cfg)
		}, 3, time.Minute).
		Step(fmt.Sprintf("verify changefeed %s CheckpointTs advance normal", changefeedName), func() error {
			return m.cfValidator.VerifyCheckpointTsAdvanceNormal(2*time.Minute, tidbCluster, changefeedID)
		})
}

func (m *pulsarChangefeedSmokeCase) editChangefeed(
	tidbCluster *TiDBDedicatedEntity, changefeedID string, downstream *PulsarEntity, updatePassword bool,
) (*model.DataflowPulsarInfo, error) {
	changefeedInfo, err := m.cdcApiWrapper.GetChangefeed(tidbCluster, changefeedID)
	if err != nil {
		return nil, err
	}
	pulsarInfo := changefeedInfo.Payload.Request.PulsarInfo
	if updatePassword {
		pulsarInfo.Authentication.Password = downstream.Password
		pulsarInfo.Authentication.Token = downstream.Token
	}
	editPulsarInfoNormally(pulsarInfo)
	editReq := connector_service.EditConnectorBody{
		ConnectorID:   changefeedInfo.Payload.ConnectorID,
		ConnectorType: changefeedInfo.Payload.Request.ConnectorType,
		Name:          changefeedInfo.Payload.Name,
		PulsarInfo:    pulsarInfo,
	}
	err = m.cdcApiWrapper.EditChangefeed(tidbCluster, changefeedID, editReq)
	if err != nil {
		return nil, err
	}
	return pulsarInfo, nil
}

type pulsarChangefeedOpt func(mysqlInfo *model.DataflowPulsarInfo)

func pulsarChangefeedWithInvalidTableFilter(pulsarInfo *model.DataflowPulsarInfo) {
	pulsarInfo.TableConfig.FilterRule = []string{"abc", "def"}
}

func pulsarChangefeedWithInvalidEventFilter(pulsarInfo *model.DataflowPulsarInfo) {
	pulsarInfo.TableConfig.EventFilterRule = []*model.DataflowEventFilterRule{
		{
			IgnoreEvent: []string{"all dml", "all ddl", "update"},
			Matcher:     []string{"a.b.d", "c.d.d"},
		},
	}
}

func genCreatePulsarChangefeedReqWithOpt(
	name, instanceType, dbName string, downstream *PulsarEntity, opt pulsarChangefeedOpt,
) *connector_service.CreateConnectorBody {
	req := genCreatePulsarChangefeedReq(name, instanceType, dbName, downstream)
	opt(req.PulsarInfo)
	return req
}

func genCreatePulsarChangefeedReq(
	name, instanceType, dbName string, downstream *PulsarEntity,
) *connector_service.CreateConnectorBody {
	changefeedSpec := &connector_service.CreateConnectorBody{}
	changefeedSpec.Name = name
	connectorType := model.DataflowConnectorTypePULSAR
	changefeedSpec.ConnectorType = &connectorType
	changefeedSpec.ChangefeedInstanceType = instanceType
	pulsarInfo := &model.DataflowPulsarInfo{}
	compressionType := model.DataflowPulsarCompressionTypePULSARCOMPRESSIONLZ4
	pulsarInfo.Endpoint = downstream.Endpoint
	pulsarInfo.DestinationProtocol = model.DataflowPulsarDestinationProtocol(downstream.Protocol).Pointer()
	pulsarInfo.CompressionType = &compressionType
	pulsarInfo.Authentication = &model.DataflowPulsarAuthentication{
		AuthType: model.DataflowPulsarAuthType(downstream.AuthType).Pointer(),
		UserName: downstream.User,
		Password: downstream.Password,
		Token:    downstream.Token,
	}
	pulsarInfo.TableConfig = &model.DataflowTableConfig{
		FilterRule: []string{fmt.Sprintf("%s.*", dbName)},
		EventFilterRule: []*model.DataflowEventFilterRule{
			{
				Matcher:     []string{fmt.Sprintf("%s.*", dbName)},
				IgnoreEvent: []string{"delete"},
			},
		},
	}
	startMode := model.DataflowStartModeFROMNOW
	pulsarInfo.StartPosition = &model.DataflowStartPosition{
		Mode: &startMode,
	}
	dataFormatProtocol := model.DataflowProtocolCANALJSON
	pulsarInfo.DataFormat = &model.DataflowPulsarDataFormat{
		EnableTidbExtension: true,
		Protocol:            &dataFormatProtocol,
	}
	dispatchType := model.DataflowDispatchTypeDISPATCHTYPEBYTABLE
	pulsarInfo.TopicDistribution = &model.DataflowPulsarTopicDistribution{
		PulsarTenant:    downstream.Tenant,
		PulsarNamespace: downstream.Namespace,
		DefaultTopic:    downstream.Topic,
		DispatchType:    &dispatchType,
		TopicPrefix:     "hello",
		Separator:       "_",
		TopicSuffix:     "world",
	}
	partitionTypeTable := model.DataflowPartitionDispatcherTABLE
	partitionTypeIndex := model.DataflowPartitionDispatcherINDEXVALUE
	partitionTypeTs := model.DataflowPartitionDispatcherTS
	partitionTypeColumns := model.DataflowPartitionDispatcherCOLUMNS
	pulsarInfo.PartitionDispatchers = []*model.DataflowPartitionDispatcherConfig{
		{
			PartitionType: &partitionTypeIndex,
			Matcher:       []string{"a.*"},
			IndexName:     "id",
		},
		{
			PartitionType: &partitionTypeTable,
			Matcher:       []string{"b.*"},
		},
		{
			PartitionType: &partitionTypeColumns,
			Matcher:       []string{"c.*"},
			Columns:       []string{"name"},
		},
		{
			PartitionType: &partitionTypeTs,
			Matcher:       []string{"*.*"},
		},
	}
	pulsarInfo.AdvancedSetting = &model.DataflowPulsarAdvancedSetting{
		ConnectionTimeout:       10,
		OperationTimeout:        30,
		SendTimeout:             30,
		BatchingMaxMessages:     1000,
		BatchingMaxPublishDelay: 1000,
	}
	changefeedSpec.PulsarInfo = pulsarInfo
	return changefeedSpec
}

func comparePulsarChangefeedKernelConfig(pulsarInfo *model.DataflowPulsarInfo, kernelConfig *ticdc_v1.Changefeed) error {
	sinkUri, err := url.Parse(kernelConfig.SinkUri)
	if err != nil {
		return fmt.Errorf("parse sink uri error: %w", err)
	}
	if pulsarInfo.Endpoint != "" && sinkUri.Host != pulsarInfo.Endpoint {
		return fmt.Errorf("pulsar endpoint mismatch, expected %s, got %s", pulsarInfo.Endpoint, sinkUri.Host)
	}
	if protocol := transPulsarDestinationProtocol(*pulsarInfo.DestinationProtocol); sinkUri.Scheme != protocol {
		return fmt.Errorf("pulsar protocol mismatch, expected protocol %s, got %s", protocol, sinkUri.Scheme)
	}
	if err := checkMatched("data protocol", transferProtocol(*pulsarInfo.DataFormat.Protocol), sinkUri.Query().Get("protocol")); err != nil {
		return err
	}
	if err := checkMatched("enable tidb extension", pulsarInfo.DataFormat.EnableTidbExtension, transStringBool(sinkUri.Query().Get("enable-tidb-extension"))); err != nil {
		return err
	}
	pulsarTopic := fmt.Sprintf("/persistent://%s/%s/%s", pulsarInfo.TopicDistribution.PulsarTenant, pulsarInfo.TopicDistribution.PulsarNamespace, pulsarInfo.TopicDistribution.DefaultTopic)
	if err := checkMatched("pulsar topic", pulsarTopic, sinkUri.Path); err != nil {
		return err
	}

	if !pie.Equals(pulsarInfo.TableConfig.FilterRule, kernelConfig.Config.Filter.Rules) {
		return fmt.Errorf("table filter rules mismatch, expected %v, got %v", pulsarInfo.TableConfig.FilterRule, kernelConfig.Config.Filter.Rules)
	}

	if len(pulsarInfo.TableConfig.EventFilterRule) != len(kernelConfig.Config.Filter.EventFilters) {
		return fmt.Errorf("length of event filter rules mismatch, expected %d, got %d", len(pulsarInfo.TableConfig.EventFilterRule), len(kernelConfig.Config.Filter.EventFilters))
	}
	for i, expectedEventFilter := range pulsarInfo.TableConfig.EventFilterRule {
		actualEventFilter := kernelConfig.Config.Filter.EventFilters[i]
		if !pie.Equals(expectedEventFilter.Matcher, actualEventFilter.Matcher) {
			return fmt.Errorf("matcher of event filter %d mismatch, expected %v, got %v", i, expectedEventFilter.Matcher, actualEventFilter.Matcher)
		}
		if !pie.Equals(expectedEventFilter.IgnoreEvent, actualEventFilter.IgnoreEvent) {
			return fmt.Errorf("ignore event of event filter %d mismatch, expected %v, got %v", i, expectedEventFilter.IgnoreEvent, actualEventFilter.IgnoreEvent)
		}
	}

	pulsarConfig := kernelConfig.Config.Sink.PulsarConfig
	if compressionType := transPulsarCompressionType(*pulsarInfo.CompressionType); pulsarConfig.CompressionType != compressionType {
		return fmt.Errorf("compression type mismatch, expected %s, got %s", compressionType, pulsarConfig.CompressionType)
	}
	if err := checkMatched("token", pulsarInfo.Authentication.Token, pulsarConfig.AuthenticationToken); err != nil {
		return err
	}
	if err := checkMatched("user name", pulsarInfo.Authentication.UserName, pulsarConfig.BasicUserName); err != nil {
		return err
	}
	if err := checkMatched("password", pulsarInfo.Authentication.Password, pulsarConfig.BasicPassword); err != nil {
		return err
	}

	dispatchers := kernelConfig.Config.Sink.Dispatchers
	if err := checkMatched("dispatcher length", len(pulsarInfo.PartitionDispatchers), len(dispatchers)); err != nil {
		return err
	}

	var dispatchedTopic string
	switch *pulsarInfo.TopicDistribution.DispatchType {
	case model.DataflowDispatchTypeDISPATCHTYPEONETOPIC:
		dispatchedTopic = pulsarInfo.TopicDistribution.DefaultTopic
	case model.DataflowDispatchTypeDISPATCHTYPEBYDATABASE:
		dispatchedTopic = fmt.Sprintf("persistent://%s/%s/%s{schema}%s",
			pulsarInfo.TopicDistribution.PulsarTenant, pulsarInfo.TopicDistribution.PulsarNamespace,
			pulsarInfo.TopicDistribution.TopicPrefix, pulsarInfo.TopicDistribution.TopicSuffix)
	case model.DataflowDispatchTypeDISPATCHTYPEBYTABLE:
		dispatchedTopic = fmt.Sprintf("persistent://%s/%s/%s{schema}%s{table}%s",
			pulsarInfo.TopicDistribution.PulsarTenant, pulsarInfo.TopicDistribution.PulsarNamespace,
			pulsarInfo.TopicDistribution.TopicPrefix, pulsarInfo.TopicDistribution.Separator, pulsarInfo.TopicDistribution.TopicSuffix)
	}
	for i, dispatcher := range dispatchers {
		dispatchConf := pulsarInfo.PartitionDispatchers[i]
		if !pie.Equals(dispatcher.Matcher, dispatchConf.Matcher) {
			return fmt.Errorf("dispatcher %d matcher mismatch, expected %v, got %v", i, dispatchConf.Matcher, dispatcher.Matcher)
		}
		if !pie.Equals(dispatcher.Columns, dispatchConf.Columns) {
			return fmt.Errorf("dispatcher %d columns mismatch, expected %v, got %v", i, dispatchConf.Columns, dispatcher.Columns)
		}
		if err := checkMatched(fmt.Sprintf("dispatcher %d partition type", i), transferPartitionDispatcher(*dispatchConf.PartitionType), dispatcher.Partition); err != nil {
			return err
		}
		if err := checkMatched(fmt.Sprintf("dispatcher %d index", i), dispatchConf.IndexName, dispatcher.Index); err != nil {
			return err
		}
		if err := checkMatched(fmt.Sprintf("dispatcher %d topic", i), dispatchedTopic, dispatcher.Topic); err != nil {
			return err
		}
	}

	if err := checkMatched("connection timeout", pulsarInfo.AdvancedSetting.ConnectionTimeout, int64(pulsarConfig.ConnectionTimeout)); err != nil {
		return err
	}
	if err := checkMatched("operation timeout", pulsarInfo.AdvancedSetting.OperationTimeout, int64(pulsarConfig.OperationTimeout)); err != nil {
		return err
	}
	if err := checkMatched("send timeout", pulsarInfo.AdvancedSetting.SendTimeout, int64(pulsarConfig.SendTimeout)); err != nil {
		return err
	}
	if err := checkMatched("batching max messages", pulsarInfo.AdvancedSetting.BatchingMaxMessages, int64(pulsarConfig.BatchingMaxMessages)); err != nil {
		return err
	}
	if err := checkMatched("batching max publish delay", pulsarInfo.AdvancedSetting.BatchingMaxPublishDelay, int64(pulsarConfig.BatchingMaxPublishDelay)); err != nil {
		return err
	}
	return nil
}

func editPulsarInfoNormally(pulsarInfo *model.DataflowPulsarInfo) {
	pulsarInfo.TableConfig.FilterRule = append(pulsarInfo.TableConfig.FilterRule, "test.*")
	pulsarInfo.TableConfig.EventFilterRule = []*model.DataflowEventFilterRule{}
}

func transPulsarCompressionType(compressionType model.DataflowPulsarCompressionType) string {
	switch compressionType {
	case model.DataflowPulsarCompressionTypePULSARCOMPRESSIONNONE:
		return ""
	case model.DataflowPulsarCompressionTypePULSARCOMPRESSIONLZ4:
		return CompressionTypeLZ4
	case model.DataflowPulsarCompressionTypePULSARCOMPRESSIONZLIB:
		return CompressionTypeZLIB
	case model.DataflowPulsarCompressionTypePULSARCOMPRESSIONZSTD:
		return CompressionTypeZSTD
	default:
		return ""
	}
}

func transPulsarDestinationProtocol(protocol model.DataflowPulsarDestinationProtocol) string {
	switch protocol {
	case model.DataflowPulsarDestinationProtocolPULSARPROTOCOLPULSAR:
		return PulsarProtocolPulsar
	case model.DataflowPulsarDestinationProtocolPULSARPROTOCOLPULSARSSL:
		return PulsarProtocolPulsarSSL
	case model.DataflowPulsarDestinationProtocolPULSARPROTOCOLHTTP:
		return PulsarProtocolPulsarHTTP
	case model.DataflowPulsarDestinationProtocolPULSARPROTOCOLHTTPS:
		return PulsarProtocolPulsarHTTPS
	default:
		return ""
	}
}

func transStringBool(boolStr string) bool {
	return boolStr == "true"
}

func checkMatched[V comparable](name string, expected, actual V) error {
	if expected != actual {
		return fmt.Errorf("%s mismatch, expected %v, got %v", name, expected, actual)
	}
	return nil
}