package changefeed

import (
	"fmt"
	"net/url"
	"strings"
	"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"
)

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

func NewS3ChangefeedSmokeCase(cloudProvider CloudProviderType) E2ECase {
	return &s3ChangefeedSmokeCase{cloudProvider: cloudProvider}
}

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

func (m *s3ChangefeedSmokeCase) DeclareDependencyRequirements() map[string]DependencyRequirement {
	return map[string]DependencyRequirement{
		"tidb": &TiDBDedicatedRequirement{
			Dedicated: false,
			TiDBDedicatedDimensions: TiDBDedicatedDimensions{
				CloudProvider: m.cloudProvider,
				Region:        config.GlobalConfig.DefaultRegionOf(m.cloudProvider),
			},
		},
		"s3": &S3Requirement{
			S3Dimensions: S3Dimensions{
				Region: config.GlobalConfig.DefaultRegionOf(m.cloudProvider),
			},
		},
	}
}

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

	var (
		changefeedID             string
		createS3Info, editS3Info *model.DataflowS3Info
	)

	// 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 := []s3ChangefeedOpt{
		s3ChangefeedWithInvalidDataFormat,
		s3ChangefeedWithInvalidFlushConfig,
		s3ChangefeedWithInvalidTableFilter,
		s3ChangefeedWithInvalidEventFilter,
	}

	for _, opt := range invalidCreateReqOpts {
		m.Expect("create s3 changefeed with invalid params", func() error {
			req := genCreateS3ChangefeedReqWithOpt(changefeedName, instanceType, dbName, tidbCluster.ClusterID, 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.BucketURI), 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 := genCreateS3ChangefeedReq(changefeedName, instanceType, dbName, tidbCluster.ClusterID, downstream)
			resp, err := m.cdcApiWrapper.CreateChangefeed(tidbCluster, *createReq)
			if err == nil {
				changefeedID = *resp.Payload.ConnectorID
				createS3Info = createReq.S3Info
			}
			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 compareS3ChangefeedKernelConfig(createS3Info, 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)
		}).
		Step("verifyGetCheckpointWithoutCache", func() error {
			return m.cfValidator.VerifyGetCheckpointWithoutCache(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 := []s3ChangefeedOpt{
		s3ChangefeedWithInvalidFlushConfig,
		s3ChangefeedWithInvalidDataFormat,
		s3ChangefeedWithInvalidTableFilter,
		s3ChangefeedWithInvalidEventFilter,
	}
	for _, opt := range invalidEditReqOpts {
		m.Expect("edit mysql changefeed with invalid params", func() error {
			changefeedInfo, err := m.cdcApiWrapper.GetChangefeed(tidbCluster, changefeedID)
			s3Info := changefeedInfo.Payload.Request.S3Info
			opt(s3Info)
			editReq := connector_service.EditConnectorBody{
				ConnectorID:   changefeedInfo.Payload.ConnectorID,
				ConnectorType: changefeedInfo.Payload.Request.ConnectorType,
				Name:          changefeedInfo.Payload.Name,
				S3Info:        s3Info,
			}
			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 {
			s3Info, err := m.editChangefeed(tidbCluster, changefeedID, downstream, false)
			if err == nil {
				editS3Info = s3Info
			}
			return err
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("edit changefeed %s with password", changefeedName), func() error {
			s3Info, err := m.editChangefeed(tidbCluster, changefeedID, downstream, true)
			if err == nil {
				editS3Info = s3Info
			}
			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 compareS3ChangefeedKernelConfig(editS3Info, 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)
		}).
		Step("verifyGetCheckpointWithoutCache", func() error {
			return m.cfValidator.VerifyGetCheckpointWithoutCache(tidbCluster, changefeedID)
		})
}

func (m *s3ChangefeedSmokeCase) editChangefeed(
	tidbCluster *TiDBDedicatedEntity, changefeedID string, downstream *S3Entity, updatePassword bool,
) (*model.DataflowS3Info, error) {
	changefeedInfo, err := m.cdcApiWrapper.GetChangefeed(tidbCluster, changefeedID)
	if err != nil {
		return nil, err
	}
	s3Info := changefeedInfo.Payload.Request.S3Info
	if updatePassword {
		s3Info.Endpoint.SecretKey = downstream.SecretKey
	}
	editS3InfoNormally(s3Info)
	editReq := connector_service.EditConnectorBody{
		ConnectorID:   changefeedInfo.Payload.ConnectorID,
		ConnectorType: changefeedInfo.Payload.Request.ConnectorType,
		Name:          changefeedInfo.Payload.Name,
		S3Info:        s3Info,
	}
	err = m.cdcApiWrapper.EditChangefeed(tidbCluster, changefeedID, editReq)
	if err != nil {
		return nil, err
	}
	return s3Info, nil
}

type s3ChangefeedOpt func(s3Info *model.DataflowS3Info)

func s3ChangefeedWithInvalidDataFormat(s3Info *model.DataflowS3Info) {
	s3Info.DataFormat.Delimiter = ""
}

func s3ChangefeedWithInvalidFlushConfig(s3Info *model.DataflowS3Info) {
	s3Info.StorageFlushConfig.FlushInterval = "20m"
}

func s3ChangefeedWithInvalidTableFilter(s3Info *model.DataflowS3Info) {
	s3Info.TableConfig.FilterRule = []string{"abc", "def"}
}

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

func genCreateS3ChangefeedReqWithOpt(
	name, instanceType, dbName, clusterID string, downstream *S3Entity, opt s3ChangefeedOpt,
) *connector_service.CreateConnectorBody {
	req := genCreateS3ChangefeedReq(name, instanceType, dbName, clusterID, downstream)
	opt(req.S3Info)
	return req
}

func genCreateS3ChangefeedReq(
	name, instanceType, dbName, clusterID string, downstream *S3Entity,
) *connector_service.CreateConnectorBody {
	changefeedSpec := &connector_service.CreateConnectorBody{}
	changefeedSpec.Name = name
	connectorType := model.DataflowConnectorTypeS3
	encodingMethod := model.DataflowBinaryEncodingMethodBASE64
	s3Protocol := model.DataflowS3ProtocolS3PROTOCOLCSV
	changefeedSpec.ConnectorType = &connectorType
	changefeedSpec.ChangefeedInstanceType = instanceType
	changefeedSpec.S3Info = &model.DataflowS3Info{}
	bucketURI := downstream.BucketURI
	if strings.HasSuffix(bucketURI, "/") {
		bucketURI = bucketURI[:len(bucketURI)-1]
	}
	endpoint := &model.DataflowS3Endpoint{
		S3URI: fmt.Sprintf("%s/%s/%s", bucketURI, clusterID, name),
	}
	if downstream.UsesAssumeRole() {
		endpoint.AuthType = model.DataflowS3AuthTypeS3AUTHTYPEASSUMEROLE.Pointer()
		endpoint.IamAssumedRoleArn = downstream.RoleARN
	} else {
		endpoint.AuthType = model.DataflowS3AuthTypeS3AUTHTYPEAKSK.Pointer()
		endpoint.AccessKey = downstream.AccessKey
		endpoint.SecretKey = downstream.SecretKey
	}
	changefeedSpec.S3Info.Endpoint = endpoint
	changefeedSpec.S3Info.DataFormat = &model.DataflowS3Dataformat{
		BinaryEncodingMethod: &encodingMethod,
		DateSeparator:        "day",
		Terminator:           "\\n",
		Delimiter:            ",",
		Quote:                "\"",
		Null:                 "\\N",
		Protocol:             &s3Protocol,
	}
	changefeedSpec.S3Info.StorageFlushConfig = &model.DataflowStorageFlushConfig{
		FlushInterval: "60s",
		FileSize:      "67108864",
	}
	changefeedSpec.S3Info.TableConfig = &model.DataflowTableConfig{
		FilterRule: []string{fmt.Sprintf("%s.*", dbName)},
		EventFilterRule: []*model.DataflowEventFilterRule{
			{
				Matcher:     []string{fmt.Sprintf("%s.*", dbName)},
				IgnoreEvent: []string{"delete"},
			},
		},
	}
	startMode := model.DataflowStartModeFROMNOW
	changefeedSpec.S3Info.StartPosition = &model.DataflowStartPosition{
		Mode: &startMode,
	}
	return changefeedSpec
}

func compareS3ChangefeedKernelConfig(s3Info *model.DataflowS3Info, kernelConfig *ticdc_v1.Changefeed) error {
	sinkUri, err := url.Parse(kernelConfig.SinkUri)
	if err != nil {
		return fmt.Errorf("parse sink uri error: %w", err)
	}
	s3URI := fmt.Sprintf("%s://%s%s", sinkUri.Scheme, sinkUri.Host, sinkUri.Path)
	if s3URI != s3Info.Endpoint.S3URI {
		return fmt.Errorf("s3 uri mismatch, expected %s, got %s", s3Info.Endpoint.S3URI, s3URI)
	}
	if err := compareS3StaticCredentials(s3Info.Endpoint, sinkUri); err != nil {
		return err
	}

	return compareCloudStorageCommonKernelConfig(
		s3Info.TableConfig, s3Info.DataFormat, s3Info.StorageFlushConfig, sinkUri, kernelConfig)
}

func compareS3StaticCredentials(endpoint *model.DataflowS3Endpoint, sinkURI *url.URL) error {
	if endpoint.AuthType != nil && *endpoint.AuthType == model.DataflowS3AuthTypeS3AUTHTYPEASSUMEROLE {
		return nil
	}

	ak, sk := sinkURI.Query().Get("access-key"), sinkURI.Query().Get("secret-access-key")
	if ak != endpoint.AccessKey {
		return fmt.Errorf("access key mismatch, expected %s, got %s", endpoint.AccessKey, ak)
	}
	if sk != endpoint.SecretKey {
		return fmt.Errorf("secret key mismatch")
	}

	return nil
}

func compareCloudStorageCommonKernelConfig(
	tableConfig *model.DataflowTableConfig,
	dataFormat *model.DataflowS3Dataformat,
	flushConfig *model.DataflowStorageFlushConfig,
	sinkUri *url.URL,
	kernelConfig *ticdc_v1.Changefeed,
) error {
	flushInterval, fileSize := sinkUri.Query().Get("flush-interval"), sinkUri.Query().Get("file-size")
	if flushInterval != flushConfig.FlushInterval {
		return fmt.Errorf("flush interval mismatch, expected %s, got %s", flushConfig.FlushInterval, flushInterval)
	}
	if fileSize != flushConfig.FileSize {
		return fmt.Errorf("file size mismatch, expected %s, got %s", flushConfig.FileSize, fileSize)
	}

	if kernelConfig.Config.Sink.Terminator != dataFormat.Terminator {
		return fmt.Errorf("terminator mismatch, expected %s, got %s", dataFormat.Terminator, kernelConfig.Config.Sink.Terminator)
	}
	if !isDateSeparatorEqual(dataFormat.DateSeparator, kernelConfig.Config.Sink.DateSeparator) {
		return fmt.Errorf("date separator mismatch, expected %s, got %s", dataFormat.DateSeparator, kernelConfig.Config.Sink.DateSeparator)
	}
	expectedS3Protocol := s3ProtocolAPI2Kernel(*dataFormat.Protocol)
	if kernelConfig.Config.Sink.Protocol != expectedS3Protocol {
		return fmt.Errorf("protocol mismatch, expected %s, got %s", expectedS3Protocol, kernelConfig.Config.Sink.Protocol)
	}
	expectedEncodingMethod := s3EncodingMethodAPI2Kernel(*dataFormat.BinaryEncodingMethod)
	if kernelConfig.Config.Sink.Csv.BinaryEncodingMethod != expectedEncodingMethod {
		return fmt.Errorf("binary encoding method mismatch, expected %s, got %s", expectedEncodingMethod, kernelConfig.Config.Sink.Csv.BinaryEncodingMethod)
	}
	if kernelConfig.Config.Sink.Csv.Null != dataFormat.Null {
		return fmt.Errorf("null mismatch, expected %s, got %s", dataFormat.Null, kernelConfig.Config.Sink.Csv.Null)
	}
	if kernelConfig.Config.Sink.Csv.Delimiter != dataFormat.Delimiter {
		return fmt.Errorf("delimiter mismatch, expected %s , got %s", dataFormat.Delimiter, kernelConfig.Config.Sink.Csv.Delimiter)
	}
	if kernelConfig.Config.Sink.Csv.Quote != dataFormat.Quote {
		return fmt.Errorf("quote mismatch, expected %s , got %s", dataFormat.Quote, kernelConfig.Config.Sink.Csv.Quote)
	}

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

	if len(tableConfig.EventFilterRule) != len(kernelConfig.Config.Filter.EventFilters) {
		return fmt.Errorf("length of event filter rules mismatch, expected %d, got %d", len(tableConfig.EventFilterRule), len(kernelConfig.Config.Filter.EventFilters))
	}
	for i, expectedEventFilter := range 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)
		}
	}
	return nil
}

func isDateSeparatorEqual(expected, actual string) bool {
	return strings.EqualFold(expected, actual)
}

func editS3InfoNormally(s3Info *model.DataflowS3Info) {
	s3Info.TableConfig.FilterRule = append(s3Info.TableConfig.FilterRule, "test.*")
	s3Info.TableConfig.EventFilterRule = append(s3Info.TableConfig.EventFilterRule, &model.DataflowEventFilterRule{
		Matcher:     []string{"test.*"},
		IgnoreEvent: []string{"update", "delete"},
	})
}

func s3ProtocolAPI2Kernel(protocol model.DataflowS3Protocol) string {
	switch protocol {
	case model.DataflowS3ProtocolS3PROTOCOLCSV:
		return "csv"
	case model.DataflowS3ProtocolS3PROTOCOLCANALJSON:
		return "canal-json"
	default:
		return ""
	}
}

func s3EncodingMethodAPI2Kernel(encodingMethod model.DataflowBinaryEncodingMethod) string {
	switch encodingMethod {
	case model.DataflowBinaryEncodingMethodHEX:
		return "hex"
	case model.DataflowBinaryEncodingMethodBASE64:
		return "base64"
	default:
		return ""
	}
}