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"
)

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

func NewMySQLChangefeedSmokeCase(cloudProvider CloudProviderType) E2ECase {
	return &mysqlChangefeedSmokeCase{cloudProvider: cloudProvider}
}

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

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

func (m *mysqlChangefeedSmokeCase) Run() {
	tidbCluster := m.TiDBDedicated("tidb")
	downstream := m.MySQL("mysql")
	changefeedName := m.Name()
	instanceType := "2rcu"
	dbName := m.Name()
	tableFilter := []string{fmt.Sprintf("%s.*", dbName)}
	runningCheckRetry := runningCheckRetryForCloud(m.cloudProvider)

	var (
		changefeedID                   string
		createMySQLInfo, editMySQLInfo *model.DataflowMySQLInfo
	)

	// 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)
	mysqlParams := &sample_data.MySQLParameter{
		Endpoint:   downstream.Endpoint,
		User:       downstream.User,
		Password:   downstream.Password,
		TLSEnabled: false,
	}
	downstreamMgr := sample_data.NewMySQLDownstreamManager(m.Logger(), mysqlParams, dbName)

	// test create request validation
	invalidCreateReqOpts := []mysqlChangefeedOpt{
		mysqlChangefeedWithInvalidTableFilter,
		mysqlChangefeedWithInvalidEventFilter,
	}

	for _, opt := range invalidCreateReqOpts {
		m.Expect("create mysql changefeed with invalid params", func() error {
			req := genCreateMySQLChangefeedReqWithOpt(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).
			RetryDo("delete database and tables in downstream", downstreamMgr.DropTables, 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("init database and table in downstream", downstreamMgr.InitTables, 3, time.Minute).
		RetryDo(fmt.Sprintf("create changefeed %s", changefeedName), func() (err error) {
			createReq := genCreateMySQLChangefeedReq(changefeedName, instanceType, dbName, downstream)
			resp, err := m.cdcApiWrapper.CreateChangefeed(tidbCluster, *createReq)
			if err == nil {
				changefeedID = *resp.Payload.ConnectorID
				createMySQLInfo = createReq.MysqlInfo
			}
			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)
		}, runningCheckRetry, time.Minute).
		Expect("compare changefeed kernel config after create", func() error {
			cfg, err := m.cdcApiWrapper.GetChangefeedKernelConfig(tidbCluster, changefeedID)
			if err != nil {
				return err
			}
			return compareMySQLChangefeedKernelConfig(createMySQLInfo, 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 := []mysqlChangefeedOpt{
		mysqlChangefeedWithInvalidTableFilter,
		mysqlChangefeedWithInvalidEventFilter,
	}
	for _, opt := range invalidEditReqOpts {
		m.Expect("edit mysql changefeed with invalid params", func() error {
			changefeedInfo, err := m.cdcApiWrapper.GetChangefeed(tidbCluster, changefeedID)
			mysqlInfo := changefeedInfo.Payload.Request.MysqlInfo
			mysqlInfo.Connection.Password = downstream.Password
			opt(mysqlInfo)
			editReq := connector_service.EditConnectorBody{
				ConnectorID:   changefeedInfo.Payload.ConnectorID,
				ConnectorType: changefeedInfo.Payload.Request.ConnectorType,
				Name:          changefeedInfo.Payload.Name,
				MysqlInfo:     mysqlInfo,
			}
			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 {
			mysqlInfo, err := m.editChangefeed(tidbCluster, changefeedID, downstream, false)
			if err == nil {
				editMySQLInfo = mysqlInfo
			}
			return err
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("edit changefeed %s with password", changefeedName), func() error {
			mysqlInfo, err := m.editChangefeed(tidbCluster, changefeedID, downstream, true)
			if err == nil {
				editMySQLInfo = mysqlInfo
			}
			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 compareMySQLChangefeedKernelConfig(editMySQLInfo, 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 *mysqlChangefeedSmokeCase) editChangefeed(
	tidbCluster *TiDBDedicatedEntity, changefeedID string, downstream *MySQLEntity, updatePassword bool,
) (*model.DataflowMySQLInfo, error) {
	changefeedInfo, err := m.cdcApiWrapper.GetChangefeed(tidbCluster, changefeedID)
	if err != nil {
		return nil, err
	}
	mysqlInfo := changefeedInfo.Payload.Request.MysqlInfo
	if updatePassword {
		mysqlInfo.Connection.Password = downstream.Password
	}
	editMySQLInfoNormally(mysqlInfo)
	editReq := connector_service.EditConnectorBody{
		ConnectorID:   changefeedInfo.Payload.ConnectorID,
		ConnectorType: changefeedInfo.Payload.Request.ConnectorType,
		Name:          changefeedInfo.Payload.Name,
		MysqlInfo:     mysqlInfo,
	}
	err = m.cdcApiWrapper.EditChangefeed(tidbCluster, changefeedID, editReq)
	if err != nil {
		return nil, err
	}
	return mysqlInfo, nil
}

type mysqlChangefeedOpt func(mysqlInfo *model.DataflowMySQLInfo)

func mysqlChangefeedWithInvalidTableFilter(mysqlInfo *model.DataflowMySQLInfo) {
	mysqlInfo.TableConfig.FilterRule = []string{"abc", "def"}
}

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

func genCreateMySQLChangefeedReqWithOpt(
	name, instanceType, dbName string, downstream *MySQLEntity, opt mysqlChangefeedOpt,
) *connector_service.CreateConnectorBody {
	req := genCreateMySQLChangefeedReq(name, instanceType, dbName, downstream)
	opt(req.MysqlInfo)
	return req
}

func genCreateMySQLChangefeedReq(
	name, instanceType, dbName string, downstream *MySQLEntity,
) *connector_service.CreateConnectorBody {
	changefeedSpec := &connector_service.CreateConnectorBody{}
	changefeedSpec.Name = name
	connectorType := model.DataflowConnectorTypeMYSQL
	changefeedSpec.ConnectorType = &connectorType
	changefeedSpec.ChangefeedInstanceType = instanceType
	changefeedSpec.MysqlInfo = &model.DataflowMySQLInfo{}
	changefeedSpec.MysqlInfo.Connection = &model.DataflowConnection{
		Endpoint: downstream.Endpoint,
		Username: downstream.User,
		Password: downstream.Password,
	}
	changefeedSpec.MysqlInfo.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.MysqlInfo.StartPosition = &model.DataflowStartPosition{
		Mode: &startMode,
	}
	return changefeedSpec
}

func compareMySQLChangefeedKernelConfig(mysqlInfo *model.DataflowMySQLInfo, kernelConfig *ticdc_v1.Changefeed) error {
	sinkUri, err := url.Parse(kernelConfig.SinkUri)
	if err != nil {
		return fmt.Errorf("parse sink uri error: %w", err)
	}
	if mysqlInfo.Connection.Endpoint != "" && sinkUri.Host != mysqlInfo.Connection.Endpoint {
		return fmt.Errorf("mysql endpoint mismatch, expected %s, got %s", mysqlInfo.Connection.Endpoint, sinkUri.Host)
	}
	if sinkUri.User.Username() != mysqlInfo.Connection.Username {
		return fmt.Errorf("mysql user mismatch, expected %s, got %s", mysqlInfo.Connection.Username, sinkUri.User.Username())
	}

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

	if len(mysqlInfo.TableConfig.EventFilterRule) != len(kernelConfig.Config.Filter.EventFilters) {
		return fmt.Errorf("length of event filter rules mismatch, expected %d, got %d", len(mysqlInfo.TableConfig.EventFilterRule), len(kernelConfig.Config.Filter.EventFilters))
	}
	for i, expectedEventFilter := range mysqlInfo.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 editMySQLInfoNormally(mysqlInfo *model.DataflowMySQLInfo) {
	mysqlInfo.TableConfig.FilterRule = append(mysqlInfo.TableConfig.FilterRule, "test.*")
	mysqlInfo.TableConfig.EventFilterRule = append(mysqlInfo.TableConfig.EventFilterRule, &model.DataflowEventFilterRule{
		Matcher:     []string{"test.*"},
		IgnoreEvent: []string{"update", "delete"},
	})
}