package changefeed

import (
	"fmt"
	"time"

	. "github.com/tidbcloud/dataflow-service-e2e/cases"
	"github.com/tidbcloud/dataflow-service-e2e/internal/config"
	. "github.com/tidbcloud/dataflow-service-e2e/internal/model"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/premium/cdcapi"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/premium/consoleapi"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/premium/consoleapi/cdc/client/changefeed_service"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/premium/consoleapi/cdc/model"
)

// FIXME: should add more test connecton cases for different downstream and different network connection type
type TestConnectionCase struct {
	BaseCase
	cdcApiWrapper *cdcapi.CDCClientWrapper
	cloudProvider CloudProviderType
}

func NewKafkaTestConnectionCase(cloudProvider CloudProviderType) E2ECase {
	return &TestConnectionCase{
		cloudProvider: cloudProvider,
	}
}

func (k *TestConnectionCase) Name() string {
	return fmt.Sprintf("%s%s", k.cloudProvider, "TestConnection")
}

func (k *TestConnectionCase) DeclareDependencyRequirements() map[string]DependencyRequirement {
	return map[string]DependencyRequirement{
		"premium": &TiDBPremiumRequirement{
			Dedicated: false,
			TiDBPremiumDimensions: TiDBPremiumDimensions{
				CloudProvider: k.cloudProvider,
				Region:        config.GlobalConfig.DefaultRegionOf(k.cloudProvider),
			},
		},
		"kafka": &KafkaRequirement{
			KafkaDimensions: KafkaDimensions{
				NetworkType: DataflowNetworkTypeNetworkTypePublic,
				AuthType:    DataflowKafkaAuthTypeSASLPLAIN,
				TLSEnabled:  false,
			},
		},
		"mysql": &MySQLRequirement{
			MySQLDimensions: MySQLDimensions{
				NetworkType: DataflowNetworkTypeNetworkTypePublic,
			},
		},
	}
}

func (k *TestConnectionCase) Run() {
	// declare case related vars
	tidbCluster := k.TiDBPremium("premium")
	kafkaDownStream := k.Kafka("kafka")
	mysqlDownstream := k.MySQL("mysql")
	// init reusable api client and validator
	cdcApiWrapper := cdcapi.NewCDCClientWrapper(k.Logger())
	k.cdcApiWrapper = cdcApiWrapper

	k.
		Expect("kafka connection success", func() error {
			return cdcApiWrapper.TestConnection(tidbCluster, kafkaDownStream)
		}, 3, time.Minute).
		Expect("kafka connection failed since invalid password", func() error {
			params := invalidKafkaPassword(tidbCluster, kafkaDownStream)
			return k.testInvalidParams(params)
		}, 3, time.Minute).
		Expect("kafka connection failed with tls enabled", func() error {
			params := enableTLS(tidbCluster, kafkaDownStream)
			return k.testInvalidParams(params)
		}, 3, time.Minute).
		Expect("kafka connection failed since invalid endpoint", func() error {
			params := invalidBroker(tidbCluster, kafkaDownStream)
			return k.testInvalidParams(params)
		}, 3, time.Minute).
		Expect("mysql connection success", func() error {
			return cdcApiWrapper.TestConnection(tidbCluster, mysqlDownstream)
		}, 3, time.Minute).
		Expect("mysql connection failed since invalid password", func() error {
			params := invalidMySQLPassword(tidbCluster, mysqlDownstream)
			return k.testInvalidParams(params)
		}, 3, time.Minute).
		Expect("mysql connection failed since invalid endpoint", func() error {
			params := invalidMySQLEndpoint(tidbCluster, mysqlDownstream)
			return k.testInvalidParams(params)
		}, 3, time.Minute)
}

func (k *TestConnectionCase) testInvalidParams(params *changefeed_service.NgTestChangefeedParams) error {
	if err := k.cdcApiWrapper.TestConnectionWithParams(params); err != nil {
		k.Logger().Info("test invalid params failed as expected: %s", err.Error())
		return nil
	}
	return fmt.Errorf("expecting error but got nil")
}

func invalidKafkaPassword(tidbCluster *TiDBPremiumEntity, downstream *KafkaEntity) *changefeed_service.NgTestChangefeedParams {
	params := consoleapi.GetTestConnectionParams(tidbCluster, downstream)
	params.Body.KafkaInfo.Authentication.Password = "invalid"
	return params
}

func invalidBroker(tidbCluster *TiDBPremiumEntity, downstream *KafkaEntity) *changefeed_service.NgTestChangefeedParams {
	params := consoleapi.GetTestConnectionParams(tidbCluster, downstream)
	params.Body.KafkaInfo.Broker = &model.ConsoleapiBroker{
		BrokerEndpoints: "1.2.3.4:90911",
	}
	return params
}

func enableTLS(tidbCluster *TiDBPremiumEntity, downstream *KafkaEntity) *changefeed_service.NgTestChangefeedParams {
	params := consoleapi.GetTestConnectionParams(tidbCluster, downstream)
	params.Body.KafkaInfo.Broker.TLSEnable = true
	return params
}

func invalidMySQLPassword(tidbCluster *TiDBPremiumEntity, mysqlDownstream *MySQLEntity) *changefeed_service.NgTestChangefeedParams {
	params := consoleapi.GetTestConnectionParams(tidbCluster, mysqlDownstream)
	params.Body.MysqlInfo.Connection.Password = "invalid"
	return params
}

func invalidMySQLEndpoint(tidbCluster *TiDBPremiumEntity, mysqlDownstream *MySQLEntity) *changefeed_service.NgTestChangefeedParams {
	params := consoleapi.GetTestConnectionParams(tidbCluster, mysqlDownstream)
	params.Body.MysqlInfo.Connection.Endpoint = "1.2.3.4:90911"
	return params
}