package changefeed

import (
	"errors"
	"fmt"
	"strings"
	"time"

	ticdc_v1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	. "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/tidbcloud/premium/cdcapi"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/premium/consoleapi"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/premium/consoleapi/cdc/model"
	"google.golang.org/protobuf/types/known/structpb"
)

type ticdcClusterConfigOperationsCase struct {
	BaseCase
	opsApiWarpper *consoleapi.OpsClientWrapper
	cdcApiWrapper *cdcapi.CDCClientWrapper
	cfValidator   *common.ChangfeedValidator
	cloudProvider CloudProviderType
}

func NewTiCDCClusterConfigOperationsCase(cloudProvider CloudProviderType) E2ECase {
	return &ticdcClusterConfigOperationsCase{cloudProvider: cloudProvider}
}

func (c *ticdcClusterConfigOperationsCase) Name() string {
	return fmt.Sprintf("%s%s", c.cloudProvider, "TiCDCClusterConfigOperationsCase")
}

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

func (c *ticdcClusterConfigOperationsCase) Run() {
	c.Logger().Info("run %s", c.Name())
	// init reusable api client and validator
	c.opsApiWarpper = consoleapi.NewOpsClientWrapper(c.Logger())
	c.cdcApiWrapper = cdcapi.NewCDCClientWrapper(c.Logger())
	c.cfValidator = common.NewChangfeedValidator(c.Logger())

	// declare case related vars
	tidbCluster := c.TiDBPremium("premium")
	downstream := c.Kafka("kafka")
	changefeedName := "ticdcClusterConfigOperations"
	instanceType := "2rcu"
	tableFilter := []string{"*.*"}
	topic := changefeedName
	var (
		changefeedID         string
		internalChangefeedID string
		ticdcClusterID       string
	)

	// cleanup changefeeds if exists
	defer func() {
		c.
			RetryDo("delete changefeed", func() error {
				return c.cdcApiWrapper.DeleteChangefeedForCleanup(tidbCluster, changefeedID)
			}, 3, time.Second*10).
			Expect("changefeed not in list", func() error {
				return c.cfValidator.AllChangefeedsNotInListByNames(tidbCluster, changefeedName)
			}, 3, time.Second*10)
	}()

	c.
		RetryDo(fmt.Sprintf("create changefeed %s", changefeedName), func() (err error) {
			changefeedID, err = c.cfValidator.CreateDummyChangefeedWithPublicKafka(
				changefeedName, instanceType, tidbCluster, downstream, tableFilter, topic)
			return err
		}, 3, time.Minute).
		Expect("wait changefeed running", func() error {
			resp, err := c.opsApiWarpper.ListInternalChangefeeds(tidbCluster.ClusterID)
			if err != nil {
				return err
			}
			for _, internalChangefeed := range resp.Changefeeds {
				if internalChangefeed.DisplayName == changefeedName {
					if internalChangefeed.Status != "normal" {
						return fmt.Errorf("internal changefeed %s status %s is not running", internalChangefeed.ID, internalChangefeed.Status)
					}
					internalChangefeedID = internalChangefeed.ID
					return nil
				}
			}
			return errors.New("no internal changefeed found")
		}, 40, time.Second*30).
		Expect("check is new arch ticdc cluster", func() error {
			internalChangefeed, err := c.opsApiWarpper.GetInternalChangefeed(internalChangefeedID)
			if err != nil {
				return err
			}
			if internalChangefeed.TicdcCluster == nil {
				return fmt.Errorf("ticdc cluster of internal changefeed %s is not new arch", internalChangefeedID)
			}
			ticdcClusterID = internalChangefeed.TicdcCluster.ID
			return nil
		}, 3, time.Second*10).
		Expect("check ticdc cluster running", func() error {
			ticdcCluster, err := c.opsApiWarpper.GetInternalTiCDCCluster(ticdcClusterID)
			if err != nil {
				return err
			}
			if ticdcCluster.Status != "running" {
				return fmt.Errorf("ticdc cluster %s of internal changefeed %s is not running", ticdcClusterID, internalChangefeedID)
			}
			return nil
		}, 3, time.Second*10).
		RetryDo("restart ticdc cluster", func() error {
			return c.opsApiWarpper.RestartTiCDCCluster(ticdcClusterID)
		}, 3, time.Second*10).
		Expect("check ticdc cluster CR status", func() error {
			ticdcCluster, err := c.cdcApiWrapper.GetTiCDCClusterByChangefeed(tidbCluster, changefeedID)
			if err != nil {
				return err
			}
			if ticdcCluster.Phase != ticdc_v1.TiCDCCluster_RUNNING {
				return fmt.Errorf("ticdc cluster CR of biz changefeed %s phase %s is not running", changefeedID, ticdcCluster.Phase)
			}
			return nil
		}, 10, time.Second*30).
		RetryDo("update ticdc cluster config", func() error {
			clusterType := model.ConsoleapiConfigClusterTypeTICDCCLUSTER
			configFile := model.ConsoleapiConfigFileTypeTICDCFILE
			req := &model.ConsoleapiApplyConfigChangesReq{
				ClusterType:       &clusterType,
				ClusterID:         ticdcClusterID,
				ConfigurationFile: &configFile,
				CustomParams: map[string]model.ConsoleapiParamItem{
					"gc-ttl": {
						Value: &structpb.Value{
							Kind: &structpb.Value_NumberValue{
								NumberValue: 172800,
							},
						},
					},
					"kv-client.worker-concurrent": {
						Value: &structpb.Value{
							Kind: &structpb.Value_NumberValue{
								NumberValue: 4,
							},
						},
					},
				},
			}
			return c.opsApiWarpper.ApplyConfigChanges(req)
		}, 3, time.Second*10).
		Expect("check config of ticdc cluster CR", func() error {
			ticdcCluster, err := c.cdcApiWrapper.GetTiCDCClusterByChangefeed(tidbCluster, changefeedID)
			if err != nil {
				return err
			}
			configuration := strings.ReplaceAll(ticdcCluster.Configuration.Config, " ", "")
			if !strings.Contains(configuration, "gc-ttl=172800") {
				return fmt.Errorf("ticdc cluster CR of biz changefeed %s config gc-ttl is not updated", changefeedID)
			}
			if !strings.Contains(configuration, "worker-concurrent=4") {
				return fmt.Errorf("ticdc cluster CR of biz changefeed %s config kv-client.worker-concurrent is not updated", changefeedID)
			}
			return nil
		}, 3, time.Second*10)
}