package changefeed

import (
	"fmt"
	"time"

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

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

func NewKafkaAvroChangefeedCase(cloudProvider CloudProviderType) E2ECase {
	return &kafkaAvroChangefeedCase{
		cloudProvider: cloudProvider,
	}
}

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

func (k *kafkaAvroChangefeedCase) DeclareDependencyRequirements() map[string]DependencyRequirement {
	return map[string]DependencyRequirement{
		"tidb": &TiDBDedicatedRequirement{
			Dedicated: false,
			TiDBDedicatedDimensions: TiDBDedicatedDimensions{
				CloudProvider: k.cloudProvider,
				Region:        GlobalConfig.DefaultRegionOf(k.cloudProvider),
			},
		},
		"kafka": &KafkaRequirement{
			KafkaDimensions: KafkaDimensions{
				NetworkType: DataflowNetworkTypeNetworkTypePublic,
				AuthType:    DataflowKafkaAuthTypeSASLPLAIN,
				TLSEnabled:  false,
			},
		},
	}
}

func (k *kafkaAvroChangefeedCase) Run() {
	// declare case related vars
	tidbCluster := k.TiDBDedicated("tidb")
	downstream := k.Kafka("kafka")
	changefeedName := k.Name()
	changefeedId := ""
	// init reusable api client and validator
	cdcApiWrapper := cdcapi.NewCDCClientWrapper(k.Logger())
	k.cdcApiWrapper = cdcApiWrapper
	k.cfValidator = common.NewChangfeedValidator(k.Logger())

	schemaRegistry := GlobalConfig.GetSchemaRegistry()
	changefeedSpec := connector_service.CreateConnectorBody{
		ConnectorType:          model.NewDataflowConnectorType(model.DataflowConnectorTypeKAFKA),
		Name:                   changefeedName,
		ChangefeedInstanceType: "2rcu",
		NetworkInfo: &model.DataflowNetworkInfo{
			NetworkType: model.NewDataflowNetworkType(model.DataflowNetworkTypeNetworkTypePublic),
		},
		KafkaInfo: k.genKafkaInfo(downstream, schemaRegistry),
	}

	editSpec := connector_service.EditConnectorBody{
		ConnectorType: model.NewDataflowConnectorType(model.DataflowConnectorTypeKAFKA),
		Name:          changefeedName,
	}

	// cleanup changefeeds if exists
	defer func() {
		k.
			RetryDo("delete changefeeds", func() error {
				return k.cfValidator.DeleteChangefeedByName(tidbCluster, changefeedName)
			}, 3, time.Minute).
			Expect("changefeeds not in list", func() error {
				return k.cfValidator.AllChangefeedsNotInListByNames(tidbCluster, changefeedName)
			}, 3, 10*time.Second)
	}()

	// create changefeed
	k.
		RetryDo(fmt.Sprintf("create changefeed %s", changefeedName), func() (err error) {
			resp, err := k.cdcApiWrapper.CreateChangefeed(tidbCluster, changefeedSpec)
			if err != nil {
				return err
			}
			changefeedId = *resp.Payload.ConnectorID
			editSpec.ConnectorID = resp.Payload.ConnectorID
			return nil
		}, 3, time.Minute).
		Expect("changefeeeds are running", func() error {
			return k.cfValidator.AreChangefeedsRunning(tidbCluster, changefeedId)
		}, 20, time.Minute).
		RetryDo(fmt.Sprintf("pause changefeed %s", changefeedId), func() error {
			return cdcApiWrapper.PauseChangefeed(tidbCluster, changefeedId)
		}, 3, time.Minute).
		Expect(fmt.Sprintf("invalid  DispatchType  %s", changefeedId), func() error {
			kafkaInfo := k.genKafkaInfo(downstream, schemaRegistry)
			editSpec.KafkaInfo = kafkaInfo
			editSpec.KafkaInfo.TopicPartitionConfig.DispatchType = model.NewDataflowDispatchType(model.DataflowDispatchTypeDISPATCHTYPEONETOPIC)
			return ExpectError(cdcApiWrapper.EditChangefeed(tidbCluster, changefeedId, editSpec))
		}, 3, time.Minute).
		Expect(fmt.Sprintf("invalid  DispatchType %s", changefeedId), func() error {
			kafkaInfo := k.genKafkaInfo(downstream, schemaRegistry)
			editSpec.KafkaInfo = kafkaInfo
			editSpec.KafkaInfo.TopicPartitionConfig.DispatchType = model.NewDataflowDispatchType(model.DataflowDispatchTypeDISPATCHTYPEBYDATABASE)
			return ExpectError(cdcApiWrapper.EditChangefeed(tidbCluster, changefeedId, editSpec))
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("edit changefeed sucessfully %s", changefeedId), func() error {
			kafkaInfo := k.genKafkaInfo(downstream, schemaRegistry)
			editSpec.KafkaInfo = kafkaInfo
			editSpec.ConnectorID = &changefeedId
			kafkaInfo.DataFormat.BigintUnsignedHandlingMode = model.NewDataflowBigintUnsignedHandlingMode(model.DataflowBigintUnsignedHandlingModeBIGINTUNSIGNEDHANDLINGMODELONG)
			kafkaInfo.DataFormat.EnableTidbExtension = false
			kafkaInfo.DataFormat.DecimalHandlingMode = model.NewDataflowDecimalHandlingMode(model.DataflowDecimalHandlingModeDECIMALHANDLINGMODEPRECISE)
			return cdcApiWrapper.EditChangefeed(tidbCluster, changefeedId, editSpec)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("resume changefeed %s", changefeedId), func() error {
			return cdcApiWrapper.ResumeChangefeed(tidbCluster, changefeedId)
		}, 3, time.Minute).
		Expect("changefeeeds are running", func() error {
			return k.cfValidator.AreChangefeedsRunning(tidbCluster, changefeedId)
		}, 3, time.Minute)
}

func (k *kafkaAvroChangefeedCase) genKafkaInfo(
	downstream *KafkaEntity, registry *SchemaRegistry) *model.DataflowKafkaInfo {
	kafkaInfo := common.GenDefaultKafkaChangefeedSpec().KafkaInfo
	kafkaInfo.Authentication = &model.DataflowAuthentication{
		AuthType: model.DataflowKafkaAuthType(downstream.AuthType).Pointer(),
		UserName: downstream.User,
		Password: downstream.Password,
	}
	kafkaInfo.Broker.BrokerEndpoints = downstream.BootstrapEndpoints
	kafkaInfo.DataFormat = &model.DataflowDataFormat{
		Protocol:                   model.NewDataflowProtocol(model.DataflowProtocolAVRO),
		BigintUnsignedHandlingMode: model.NewDataflowBigintUnsignedHandlingMode(model.DataflowBigintUnsignedHandlingModeBIGINTUNSIGNEDHANDLINGMODESTRING),
		EnableTidbExtension:        true,
		DecimalHandlingMode:        model.NewDataflowDecimalHandlingMode(model.DataflowDecimalHandlingModeDECIMALHANDLINGMODESTRING),
		SchemaRegistry: &model.DataflowSchemaRegistry{
			EnableHTTPAuth:          true,
			Password:                registry.Password,
			SchemaRegistryEndpoints: registry.Endpoints,
			UserName:                registry.Username,
		},
	}
	kafkaInfo.TopicPartitionConfig.DispatchType = model.NewDataflowDispatchType(model.DataflowDispatchTypeDISPATCHTYPEBYTABLE)
	return kafkaInfo
}