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

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

	cloudProvider CloudProviderType
}

// Notice: only support Azure
func NewScaleChangefeedCase(cloudProvider CloudProviderType) E2ECase {
	return &scaleChangefeedCase{cloudProvider: cloudProvider}
}

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

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

func (k *scaleChangefeedCase) Run() {
	// declare case related vars
	tidbCluster := k.TiDBDedicated("tidb")
	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 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("init database and table in %s", tidbCluster.DisplayName),
			sampleDataWriter.InitDBTable, 3, 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 = cfValidator.CreateDummyChangefeedWithPublicKafka(changefeed1, "2rcu", tidbCluster, downstream, tableFilter, changefeed1Topic)
			return err
		}, 3, time.Minute).
		Expect(changefeed1+" is running", func() error {
			return cfValidator.AreChangefeedsRunning(tidbCluster, changefeed1Id)
		}, 30, time.Minute).
		Expect(fmt.Sprintf("cdc topology changed to expected with total RCUs as %s", "2rcu"), func() error {
			return cfValidator.IsTiCDCTopologyAsExpected(tidbCluster, changefeed1Id, "2rcu")
		}, 3, time.Minute)

	var fees []float64
	// initial instance type 2rcu = 1 x 2c
	// run through all kinds of scaling in single changefeed
	RCUs := []int{
		4,  // vertical scale up to 1 x 4c
		16, // vertical scale up + horizontal scale out to 2 x 8c
		40, // horizontal scale out to 5 x 8c
		64, // vertical scale up + horizontal scale in to 4 x 16c
		40, // vertical scale down + horizontal scale out to 5 x 8c
		24, // horizontal scale in to 3 x 8c
		4,  // vertical scale down + horizontal scale in to 1 x 4c
		2,  // vertical scale down
	}

	for _, num := range RCUs {
		newInstanceType := fmt.Sprintf("%drcu", num)
		k.
			RetryDo(fmt.Sprintf("scale %s to %s", changefeed1, newInstanceType), func() error {
				return cdcApiWrapper.ScaleChangefeed(tidbCluster, changefeed1Id, newInstanceType)
			}, 3, time.Minute).
			Expect(fmt.Sprintf("scale %s to %s done", changefeed1, newInstanceType), func() error {
				return cfValidator.IsChangefeedScaleDone(tidbCluster, changefeed1Id, newInstanceType)
			}, 30, time.Minute).
			Expect(fmt.Sprintf("cdc topology changed to expected with total RCUs as %s", newInstanceType), func() error {
				return cfValidator.IsTiCDCTopologyAsExpected(tidbCluster, changefeed1Id, newInstanceType)
			}, 3, time.Minute).
			RetryDo(fmt.Sprintf("get changefeed fee with instance type as %s", newInstanceType), func() error {
				resp, err := cdcApiWrapper.CalChangefeedFee(tidbCluster, newInstanceType)
				if err != nil {
					return err
				}
				fees = append(fees, resp.Payload.RawFee)
				return nil
			}, 3, time.Minute)
	}

	k.
		Step("validate fees", func() error {
			k.Logger().Info("RCUs: %s, fees: %s")
			for i := 1; i < len(RCUs); i++ {
				if !misc.EqualFloat(fees[i-1]/float64(RCUs[i-1]), fees[i]/float64(RCUs[i])) {
					return fmt.Errorf("the fee of different RCU is not correct, RCUs: %s, fees: %s", misc.JSON(RCUs), misc.JSON(fees))
				}
			}
			return nil
		}).
		// test two changefeeds scaling composition
		RetryDo(fmt.Sprintf("create changefeed %s", changefeed2), func() (err error) {
			changefeed2Id, err = cfValidator.CreateDummyChangefeedWithPublicKafka(changefeed2, "4rcu", tidbCluster, downstream, tableFilter, changefeed2Topic)
			return err
		}, 3, time.Minute).
		Expect(changefeed2+" is running", func() error {
			return cfValidator.AreChangefeedsRunning(tidbCluster, changefeed2Id)
		}, 20, time.Minute).
		Expect(fmt.Sprintf("cdc topology changed to expected with total RCUs as %s", "6rcu"), func() error {
			return cfValidator.IsTiCDCTopologyAsExpected(tidbCluster, changefeed1Id, "6rcu")
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("scale %s to %s", changefeed1, "4rcu"), func() error {
			return cdcApiWrapper.ScaleChangefeed(tidbCluster, changefeed1Id, "4rcu")
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("scale %s to %s", changefeed2, "2rcu"), func() error {
			return cdcApiWrapper.ScaleChangefeed(tidbCluster, changefeed2Id, "2rcu")
		}, 3, time.Minute).
		Expect("changefeeds are running", func() error {
			return cfValidator.AreChangefeedsRunning(tidbCluster, changefeed1Id, changefeed2Id)
		}, 20, time.Minute).
		Expect(fmt.Sprintf("cdc topology changed to expected with total RCUs as %s", "6rcu"), func() error {
			return cfValidator.IsTiCDCTopologyAsExpected(tidbCluster, changefeed1Id, "6rcu")
		}, 3, time.Minute)
}