package changefeed

import (
	"errors"
	"fmt"
	"strings"
	"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/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 changefeedConfigOperationsCase struct {
	BaseCase
	opsApiWarpper *consoleapi.OpsClientWrapper
	cdcApiWrapper *cdcapi.CDCClientWrapper
	cfValidator   *common.ChangfeedValidator
	cloudProvider CloudProviderType
}

func NewChangefeedConfigOperationsCase(cloudProvider CloudProviderType) E2ECase {
	return &changefeedConfigOperationsCase{cloudProvider: cloudProvider}
}

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

func (k *changefeedConfigOperationsCase) 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 (c *changefeedConfigOperationsCase) 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 := "changefeedConfigOperations"
	instanceType := "2rcu"
	tableFilter := []string{"*.*"}
	topic := changefeedName
	var (
		changefeedID         string
		internalChangefeedID string
		isNewArchTiCDC       bool
	)

	// 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).
		RetryDo("get changefeed", func() error {
			internalChangefeed, err := c.opsApiWarpper.GetInternalChangefeed(internalChangefeedID)
			if err != nil {
				return err
			}
			if internalChangefeed.TicdcCluster != nil {
				isNewArchTiCDC = true
			}
			return nil
		}, 3, time.Second*10).
		RetryDo("restart changefeed", func() error {
			return c.opsApiWarpper.RestartChangefeed(internalChangefeedID)
		}, 3, time.Second*10).
		Expect("check kernel changefeed running", func() error {
			if !isNewArchTiCDC {
				c.Logger().Info("the changefeed is cdc task, skip to check kernel state")
				return nil
			}
			kernelChangefeed, err := c.cdcApiWrapper.GetChangefeedKernelConfig(tidbCluster, changefeedID)
			if err != nil {
				return err
			}
			if kernelChangefeed.State != "normal" {
				return fmt.Errorf("biz changefeed %s kernel status %s is not running", changefeedID, kernelChangefeed.State)
			}
			return nil
		}, 3, time.Second*10).
		RetryDo("update changefeed config", func() error {
			clusterType := model.NewConsoleapiConfigClusterType(model.ConsoleapiConfigClusterTypeCHANGEFEEDCLUSTER)
			configFile := model.NewConsoleapiConfigFileType(model.ConsoleapiConfigFileTypeCHANGEFEEDFILE)
			req := &model.ConsoleapiApplyConfigChangesReq{
				ClusterType:       clusterType,
				ClusterID:         internalChangefeedID,
				ConfigurationFile: configFile,
				CustomParams: map[string]model.ConsoleapiParamItem{
					"scheduler": {
						Value: &structpb.Value{
							Kind: &structpb.Value_StructValue{
								StructValue: &structpb.Struct{
									Fields: map[string]*structpb.Value{
										"enable_table_across_nodes": {
											Kind: &structpb.Value_BoolValue{
												BoolValue: true,
											},
										},
									},
								},
							},
						},
					},
				},
			}
			return c.opsApiWarpper.ApplyConfigChanges(req)
		}, 3, time.Second*10).
		Expect("verify kernel changefeed config", func() error {
			if !isNewArchTiCDC {
				c.Logger().Info("the changefeed is cdc task, skip to check kernel state")
				return nil
			}
			kernelChangefeed, err := c.cdcApiWrapper.GetChangefeedKernelConfig(tidbCluster, changefeedID)
			if err != nil {
				return err
			}
			if kernelChangefeed.State != "normal" {
				return fmt.Errorf("biz changefeed %s kernel status %s is not running", changefeedID, kernelChangefeed.State)
			}
			if !kernelChangefeed.Config.GetScheduler().GetEnableTableAcrossNodes() {
				return fmt.Errorf("biz changefeed %s kernel config scheduler.enable_table_across_nodes is not true", changefeedID)
			}
			return nil
		}, 3, time.Second*10).
		RetryDo("update sink uri", func() error {
			clusterType := model.NewConsoleapiConfigClusterType(model.ConsoleapiConfigClusterTypeCHANGEFEEDCLUSTER)
			configFile := model.NewConsoleapiConfigFileType(model.ConsoleapiConfigFileTypeSINKURI)
			req := &model.ConsoleapiApplyConfigChangesReq{
				ClusterType:       clusterType,
				ClusterID:         internalChangefeedID,
				ConfigurationFile: configFile,
				CustomParams: map[string]model.ConsoleapiParamItem{
					"insecure-skip-verify": {
						Value: &structpb.Value{
							Kind: &structpb.Value_BoolValue{
								BoolValue: true,
							},
						},
					},
				},
			}
			return c.opsApiWarpper.ApplyConfigChanges(req)
		}, 3, time.Second*10).
		Expect("verify kernel sink uri", func() error {
			if !isNewArchTiCDC {
				c.Logger().Info("the changefeed is cdc task, skip to check kernel state")
				return nil
			}
			kernelChangefeed, err := c.cdcApiWrapper.GetChangefeedKernelConfig(tidbCluster, changefeedID)
			if err != nil {
				return err
			}
			if kernelChangefeed.State != "normal" {
				return fmt.Errorf("biz changefeed %s kernel status %s is not running", changefeedID, kernelChangefeed.State)
			}
			if !strings.Contains(kernelChangefeed.SinkUri, "insecure-skip-verify=true") {
				return fmt.Errorf("biz changefeed %s kernel sink uri is not applied", changefeedID)
			}
			return nil
		}, 3, time.Second*10)
}