package changefeed

import (
	"fmt"
	"time"

	. "github.com/tidbcloud/dataflow-service-e2e/cases"
	"github.com/tidbcloud/dataflow-service-e2e/cases/premium/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/premium/cdcapi"
)

type cdcRecreateCase struct {
	BaseCase
	cdcApiWrapper *cdcapi.CDCClientWrapper
	cfValidator   *common.ChangfeedValidator

	cloudProvider CloudProviderType
}

func NewCdcRecreateCase(cloudProvider CloudProviderType) E2ECase {
	return &cdcRecreateCase{cloudProvider: cloudProvider}
}

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

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

func (k *cdcRecreateCase) Run() {
	// declare case related vars
	tidbCluster := k.TiDBPremium("premium")
	downstream := k.Kafka("kafka")
	changefeed1 := fmt.Sprintf("%s-1", k.Name())
	changefeed2 := fmt.Sprintf("%s-2", k.Name())
	var changefeed1Id, changefeed2Id string
	changefeed1Topic := changefeed1
	changefeed2Topic := changefeed2
	dbName := k.Name()
	tableFilter := []string{fmt.Sprintf("%s.*", dbName)}

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

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

	k.
		RetryDo(fmt.Sprintf("init database and table in %s", tidbCluster.DisplayName),
			sampleDataWriter.InitDBTable, 15, time.Minute).
		RetryDo(fmt.Sprintf("start write sample data to database %s", dbName),
			sampleDataWriter.StartWriting, 3, time.Minute).
		RetryDo(fmt.Sprintf("create changefeed %s", changefeed1), func() (err error) {
			changefeed1Id, err = k.cfValidator.CreateDummyChangefeedWithPublicKafka(changefeed1, "2rcu", tidbCluster, downstream, tableFilter, changefeed1Topic)
			return err
		}, 3, time.Minute).
		Expect("changefeeed1 is running", func() error {
			return cfValidator.AreChangefeedsRunning(tidbCluster, changefeed1Id)
		}, 20, time.Minute).
		RetryDo("delete changefeed1", func() error {
			return cfValidator.DeleteChangefeedByName(tidbCluster, changefeed1)
		}, 3, time.Minute).
		Expect("changefeed1 not in list", func() error {
			return cfValidator.AllChangefeedsNotInListByNames(tidbCluster, changefeed1)
		}, 3, 10*time.Second)

	if k.cloudProvider == AZURE {
		k.Expect("wait cdc cluster destroy in azure", func() (err error) {
			return k.cfValidator.NoTiCDCClusterFound(tidbCluster)
		}, 15, time.Minute)
	} else {
		k.Step("wait cdc cluster destroy in gcp&aws", func() (err error) {
			// FIXME: need to call internal api to verify
			time.Sleep(15 * time.Minute)
			return nil
		})
	}

	k.RetryDo(fmt.Sprintf("create changefeed %s", changefeed2), func() (err error) {
		changefeed2Id, err = k.cfValidator.CreateDummyChangefeedWithPublicKafka(changefeed2, "2rcu", tidbCluster, downstream, tableFilter, changefeed2Topic)
		return err
	}, 3, time.Minute).
		Expect("changefeeed2 is running", func() error {
			return cfValidator.AreChangefeedsRunning(tidbCluster, changefeed2Id)
		}, 20, time.Minute)
}