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

type kafkaValidateChangefeedParamsCase struct {
	BaseCase
	cdcApiWrapper     *cdcapi.CDCClientWrapper
	clusterApiWrapper *cluster_non_gen.ClusterClient
	cfValidator       *common.ChangfeedValidator
	cloudProvider     CloudProviderType
}

func NewKafkaValidateChangefeedParamsCase(cloudProvider CloudProviderType) E2ECase {
	return &kafkaValidateChangefeedParamsCase{
		cloudProvider: cloudProvider,
	}
}

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

func (k *kafkaValidateChangefeedParamsCase) 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 *kafkaValidateChangefeedParamsCase) Run() {
	// declare case related vars
	tidbCluster := k.TiDBPremium("premium")
	downstream := k.Kafka("kafka")
	changefeedName := k.Name()
	changefeedId := ""
	// init reusable api client and validator
	cdcApiWrapper := cdcapi.NewCDCClientWrapper(k.Logger())
	k.cdcApiWrapper = cdcApiWrapper
	k.cfValidator = common.NewChangfeedValidator(k.Logger())
	k.clusterApiWrapper = cluster_non_gen.NewClient(k.Logger())

	// cleanup changefeeds if exists
	defer func() {
		k.
			RetryDo("delete changefeeds", func() error {
				return k.cfValidator.DeleteChangefeedByName(tidbCluster, changefeedId)
			}, 3, time.Minute).
			RetryDo("delete changefeeds", func() error {
				return k.cfValidator.DeleteChangefeedByName(tidbCluster, changefeedId)
			}, 3, time.Minute).
			Expect("changefeeds not in list", func() error {
				return k.cfValidator.AllChangefeedsNotInListByNames(tidbCluster, changefeedId)
			}, 3, 10*time.Second)
	}()

	invalidCfg := []*model.ChangefeedServiceCreateChangefeedBody{
		addWithKafkaInfo(changefeedName, downstream, invalidColumnSelector(changefeedName, downstream)),
		addWithKafkaInfo(changefeedName, downstream, invalidTableFilterRule(changefeedName, downstream)),
		addWithKafkaInfo(changefeedName, downstream, invalidEventFilter(changefeedName, downstream)),
	}
	for _, spec := range invalidCfg {
		k.Expect(fmt.Sprintf("create changefeed %s", spec.Name), func() error {
			_, err := k.cdcApiWrapper.CreateChangefeed(tidbCluster, spec)
			return ExpectError(err)
		}, 3, time.Minute)
	}
	// create changefeed
	k.
		RetryDo(fmt.Sprintf("create changefeed %s", changefeedName), func() (err error) {
			changefeedSpec := genAddChangefeedSpec(changefeedName, "2rcu", downstream)
			resp, err := k.cdcApiWrapper.CreateChangefeed(tidbCluster, changefeedSpec)
			if err != nil {
				return err
			}
			changefeedId = *resp.Payload.ChangefeedID
			return nil
		}, 3, time.Minute).
		Expect("changefeeeds are running", func() error {
			return k.cfValidator.AreChangefeedsRunning(tidbCluster, changefeedId)
		}, 20, time.Minute).
		Expect("edit changefeeeds are running", func() error {
			return ExpectError(k.cdcApiWrapper.EditChangefeed(tidbCluster, changefeedId,
				genEditChangefeedSpec(changefeedName, "2rcu", downstream)))
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("pause changefeed %s", changefeedId), func() error {
			return cdcApiWrapper.PauseChangefeed(tidbCluster, changefeedId)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("update changefeed %s", changefeedId), func() error {
			return k.cdcApiWrapper.EditChangefeed(tidbCluster, changefeedId,
				genEditChangefeedSpec(changefeedName, "2rcu", downstream))
		}, 3, time.Minute).
		Expect("delete cluster when there is a changefeed", func() error {
			return ExpectError(k.clusterApiWrapper.DeleteCluster(tidbCluster.OrgID, tidbCluster.ClusterID))
		}, 3, time.Minute)

	invalidEditCfg := []*model.ChangefeedServiceEditChangefeedBody{
		editWithKafkaInfo(changefeedName, downstream, invalidColumnSelector(changefeedName, downstream)),
		editWithKafkaInfo(changefeedName, downstream, invalidTableFilterRule(changefeedName, downstream)),
		editWithKafkaInfo(changefeedName, downstream, invalidEventFilter(changefeedName, downstream)),
	}
	for _, spec := range invalidEditCfg {
		k.Expect(fmt.Sprintf("edit changefeed %s", spec.Name), func() error {
			return ExpectError(k.cdcApiWrapper.EditChangefeed(tidbCluster, changefeedId, spec))
		}, 3, time.Minute)
	}
}

func editWithKafkaInfo(name string, downstream *KafkaEntity, kafkaInfo *model.ConsoleapiKafkaInfo) *model.ChangefeedServiceEditChangefeedBody {
	spec := genEditChangefeedSpec(name, "2rcu", downstream)
	spec.KafkaInfo = kafkaInfo
	return spec
}

func addWithKafkaInfo(name string, downstream *KafkaEntity, kafkaInfo *model.ConsoleapiKafkaInfo) *model.ChangefeedServiceCreateChangefeedBody {
	spec := genAddChangefeedSpec(name, "2rcu", downstream)
	spec.KafkaInfo = kafkaInfo
	return spec
}

func invalidTableFilterRule(name string, downstream *KafkaEntity) *model.ConsoleapiKafkaInfo {
	spec := genAddChangefeedSpec(name, "2rcu", downstream)
	spec.KafkaInfo.TableConfig.FilterRule = []string{"abc", "def"}
	return spec.KafkaInfo
}

func invalidEventFilter(name string, downstream *KafkaEntity) *model.ConsoleapiKafkaInfo {
	spec := genAddChangefeedSpec(name, "2rcu", downstream)
	spec.KafkaInfo.TableConfig.EventFilterRule = []*model.ConsoleapiEventFilterRule{
		{
			IgnoreEvent: []string{"all dml", "all ddl", "update"},
			Matcher:     []string{"a.b.d", "c.d.d"},
		},
	}
	return spec.KafkaInfo
}

func invalidColumnSelector(name string, downstream *KafkaEntity) *model.ConsoleapiKafkaInfo {
	spec := genAddChangefeedSpec(name, "2rcu", downstream)
	spec.KafkaInfo.ColumnSelectors = []*model.ConsoleapiColumnSelector{
		{
			Columns: []string{"abc", "def"},
			Matcher: []string{"a.b.c"},
		},
	}
	return spec.KafkaInfo
}

func genAddChangefeedSpec(
	name, instanceType string,
	downstream *KafkaEntity,
) *model.ChangefeedServiceCreateChangefeedBody {
	changefeedSpec := &model.ChangefeedServiceCreateChangefeedBody{
		ChangefeedType: model.ConsoleapiChangefeedTypeKAFKA.Pointer(),
	}
	changefeedSpec.Name = name
	changefeedSpec.ChangefeedSpec = instanceType
	changefeedSpec.NetworkInfo = &model.ConsoleapiNetworkInfo{
		NetworkType: model.ConsoleapiNetworkTypeNetworkTypePublic.Pointer(),
	}
	start := time.Now()
	startTso := misc.GoTimeToTSO(start)
	changefeedSpec.KafkaInfo = &model.ConsoleapiKafkaInfo{
		Authentication: &model.ConsoleapiAuthentication{
			AuthType: model.ConsoleapiKafkaAuthType(downstream.AuthType).Pointer(),
			UserName: downstream.User,
			Password: downstream.Password,
		},
		Broker: &model.ConsoleapiBroker{
			BrokerEndpoints: downstream.BootstrapEndpoints,
			Compression:     model.ConsoleapiCompressionTypeGZIP.Pointer(),
			KafkaVersion:    model.ConsoleapiKafkaVersionKAFKAVERSION3XX.Pointer(),
			TLSEnable:       false,
		},
		ColumnSelectors: []*model.ConsoleapiColumnSelector{
			{
				Columns: []string{"abc", "def"},
				Matcher: []string{
					"test.a", "test.b"},
			},
			{
				Columns: []string{"abc"},
				Matcher: []string{
					"*.*"},
			},
		},
		DataFormat: &model.ConsoleapiDataFormat{
			Protocol: model.ConsoleapiProtocolDEBEZIUM.Pointer(),
			DebeziumConfig: &model.ConsoleapiDebeziumConfig{
				DisableSchema:  false,
				OutputOldValue: true,
			},
		},
		StartPosition: &model.ConsoleapiStartPosition{
			Mode: model.ConsoleapiStartModeFROMNOW.Pointer(),
			Tso:  fmt.Sprintf("%d", startTso),
		},
		TableConfig: &model.ConsoleapiTableConfig{
			EventFilterRule: []*model.ConsoleapiEventFilterRule{
				{
					IgnoreEvent: []string{"all dml", "all ddl", "update"},
					Matcher:     []string{"a.b", "c.d"},
				},
				{
					IgnoreEvent: []string{"insert", "update"},
					Matcher:     []string{"e.f", "h.j"},
				},
			},
			FilterRule: []string{"test.a", "test.b"},
			Mode:       model.ConsoleapiTableModeFORCESYNC.Pointer(),
		},
		TopicPartitionConfig: &model.ConsoleapiTopicPartitionConfig{
			DefaultTopic: "test2",
			DispatchType: model.ConsoleapiDispatchTypeDISPATCHTYPEONETOPIC.Pointer(),
			PartitionDispatchers: []*model.ConsoleapiPartitionDispatcherConfig{
				{
					Columns: []string{
						"abc",
					},
					Matcher:       []string{"a.b"},
					PartitionType: model.ConsoleapiPartitionDispatcherCOLUMNS.Pointer(),
				},
				{
					IndexName:     "idx",
					Matcher:       []string{"a.b"},
					PartitionType: model.ConsoleapiPartitionDispatcherINDEXVALUE.Pointer(),
				},
				{
					Matcher:       []string{"a.b"},
					PartitionType: model.ConsoleapiPartitionDispatcherTS.Pointer(),
				},
				{
					Matcher:       []string{"a.b"},
					PartitionType: model.ConsoleapiPartitionDispatcherTABLE.Pointer(),
				},
			},
			PartitionNum:      1,
			ReplicationFactor: 1,
			Separator:         "abc",
			TopicPrefix:       "efg",
			TopicSuffix:       "hij",
		},
	}
	return changefeedSpec
}

func genEditChangefeedSpec(
	name, instanceType string,
	downstream *KafkaEntity,
) *model.ChangefeedServiceEditChangefeedBody {
	changefeedSpec := &model.ChangefeedServiceEditChangefeedBody{
		ChangefeedType: model.ConsoleapiChangefeedTypeKAFKA.Pointer(),
		Name:           name,
	}
	addSpec := genAddChangefeedSpec(name, instanceType, downstream)
	changefeedSpec.KafkaInfo = addSpec.KafkaInfo
	return changefeedSpec
}

func ExpectError(err error) error {
	if err != nil {
		return nil
	}
	return fmt.Errorf("expecting error but got nil")
}