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"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/premium/consoleapi/cdc/model"
)

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

	cloudProvider CloudProviderType
}

func NewKafkaChangefeedSmokeCase(cloudProvider CloudProviderType) E2ECase {
	return &kafkaChangefeedSmokeCase{cloudProvider: cloudProvider}
}

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

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

func (k *kafkaChangefeedSmokeCase) 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
	changefeed1InstaneType_Initial := "2rcu"
	changefeed1InstaneType_ScaleUp := "4rcu"
	changefeed2InstaneType_Initial := "8rcu"
	changefeed2InstaneType_ScaleDown := "4rcu"
	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)

	if GlobalConfig.ApiMode == ApiModeOpenAPI {
		k.Step("verify premium changefeed CCU options are available", func() error {
			return cfValidator.VerifyCCUsAvailable(tidbCluster)
		})
	}

	// cleanup changefeeds if exists
	defer func() {
		k.
			Step("stop writing data", func() error {
				sampleDataWriter.StopWriting()
				return nil
			}).
			RetryDo("delete changefeeds", func() error {
				return cfValidator.DeleteChangefeedByName(tidbCluster, changefeed1)
			}, 3, time.Minute).
			RetryDo("delete changefeeds", 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("test connection to downstream %s", downstream.BootstrapEndpoints), func() (err error) {
			return cdcApiWrapper.TestConnection(tidbCluster, downstream)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("init database and table in %s", tidbCluster.DisplayName),
			sampleDataWriter.InitDBTable, 15, time.Minute).
		RetryDo(fmt.Sprintf("verify tables with tableFilter %v", tableFilter), func() (err error) {
			return cfValidator.VerifyTables(tidbCluster, tableFilter, []string{fmt.Sprintf("%s.%s", dbName, sample_data.SAMPLE_TABLE)}, nil)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("create changefeed %s", changefeed1), func() (err error) {
			changefeed1Id, err = k.cfValidator.CreateDummyChangefeedWithPublicKafka(changefeed1, changefeed1InstaneType_Initial, tidbCluster, downstream, tableFilter, changefeed1Topic)
			return err
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("create changefeed %s", changefeed2), func() (err error) {
			changefeed2Id, err = k.cfValidator.CreateDummyChangefeedWithPublicKafka(changefeed2, changefeed2InstaneType_Initial, tidbCluster, downstream, tableFilter, changefeed2Topic)
			return err
		}, 3, time.Minute).
		Expect("changefeeds in list", func() error {
			return cfValidator.AreChangefeedsInList(tidbCluster, changefeed1Id, changefeed2Id)
		}, 3, 10*time.Second).
		Expect("changefeeeds are running", func() error {
			return cfValidator.AreChangefeedsRunning(tidbCluster, changefeed1Id, changefeed2Id)
		}, 20, time.Minute).
		RetryDo(fmt.Sprintf("start write sample data to database %s", dbName),
			sampleDataWriter.StartWriting, 3, time.Minute).
		Step("verify realtime checkpoint is not behind cached checkpoint", func() error {
			if GlobalConfig.ApiMode != ApiModeOpenAPI {
				return nil
			}
			return cfValidator.VerifyRealtimeCheckpointNotBehind(tidbCluster, changefeed1Id)
		}).
		Step("verify changefeeds' CheckpointTs advance normal", func() error {
			return cfValidator.VerifyCheckpointTsAdvanceNormal(5*time.Minute, tidbCluster, changefeed1Id, changefeed2Id)
		}).
		RetryDo(fmt.Sprintf("scale up changefeed %s", changefeed1), func() error {
			return cdcApiWrapper.ScaleChangefeed(tidbCluster, changefeed1Id, changefeed1InstaneType_ScaleUp)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("scale down changefeed %s", changefeed2), func() error {
			return cdcApiWrapper.ScaleChangefeed(tidbCluster, changefeed2Id, changefeed2InstaneType_ScaleDown)
		}, 3, time.Minute).
		Expect(fmt.Sprintf("changefeed %s scale up done", changefeed1), func() error {
			return cfValidator.IsChangefeedScaleDone(tidbCluster, changefeed1Id, changefeed1InstaneType_ScaleUp)
		}, 20, time.Minute).
		Expect(fmt.Sprintf("changefeed %s scale down done", changefeed2), func() error {
			return cfValidator.IsChangefeedScaleDone(tidbCluster, changefeed2Id, changefeed2InstaneType_ScaleDown)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("pause changefeed %s", changefeed1), func() error {
			return cdcApiWrapper.PauseChangefeed(tidbCluster, changefeed1Id)
		}, 3, time.Minute).
		Still(fmt.Sprintf("changefeed %s is paused", changefeed1), func() error {
			return cfValidator.IsChangefeedInStatus(tidbCluster, changefeed1Id, model.ConsoleapiChangefeedStatusPAUSED)
		}, 5, time.Minute).
		RetryDo(fmt.Sprintf("edit changefeed %s", changefeed1), func() error {
			return cfValidator.ModifyKafkaChangefeedProtocol(tidbCluster, changefeed1Id, model.ConsoleapiProtocolOPENPROTOCOL)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("resume changefeed %s", changefeed1), func() error {
			return cdcApiWrapper.ResumeChangefeed(tidbCluster, changefeed1Id)
		}, 3, time.Minute).
		Expect(fmt.Sprintf("changefeed %s edit done and running", changefeed1), func() error {
			return cfValidator.IsChangefeedUseProtocol(tidbCluster, changefeed1Id, model.ConsoleapiProtocolOPENPROTOCOL)
		}, 3, time.Minute).
		Expect(fmt.Sprintf("changefeed %s changes dispatched to kernel", changefeed1), func() error {
			return cfValidator.IsKernelChangefeedUseProtocol(tidbCluster, changefeed1Id, model.ConsoleapiProtocolOPENPROTOCOL)
		}, 3, time.Minute).
		Expect(fmt.Sprintf("changefeed %s is runining", changefeed1), func() error {
			return cfValidator.AreChangefeedsRunning(tidbCluster, changefeed1Id)
		}, 20, time.Minute).
		Step(fmt.Sprintf("verify changefeed %s CheckpointTs advance normal", changefeed1), func() error {
			return cfValidator.VerifyCheckpointTsAdvanceNormal(5*time.Minute, tidbCluster, changefeed1Id)
		})
}