package changefeed

import (
	"errors"
	"fmt"
	"net/http"
	"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/tidbcloud/dedicated/cdcapi"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/cdc/client/connector_service"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/cdc/model"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/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{
		"tidb": &TiDBDedicatedRequirement{
			Dedicated: false,
			TiDBDedicatedDimensions: TiDBDedicatedDimensions{
				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.TiDBDedicated("tidb")
	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 := []connector_service.CreateConnectorBody{
		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 ExpectHTTPStatus(err, http.StatusBadRequest)
		}, 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.ConnectorID
			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.ProjectID, tidbCluster.ClusterID))
		}, 3, time.Minute)

	invalidEditCfg := []connector_service.EditConnectorBody{
		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 {
			err := k.cdcApiWrapper.EditChangefeed(tidbCluster, changefeedId, spec)
			return ExpectHTTPStatus(err, http.StatusBadRequest)
		}, 3, time.Minute)
	}
}

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

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

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

func invalidEventFilter(name string, downstream *KafkaEntity) *model.DataflowKafkaInfo {
	spec := genAddChangefeedSpec(name, "2rcu", downstream)
	spec.KafkaInfo.TableConfig.EventFilterRule = []*model.DataflowEventFilterRule{
		{
			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.DataflowKafkaInfo {
	spec := genAddChangefeedSpec(name, "2rcu", downstream)
	spec.KafkaInfo.ColumnSelectors = []*model.DataflowColumnSelector{
		{
			Columns: []string{"abc", "def"},
			Matcher: []string{"a.b.c"},
		},
	}
	return spec.KafkaInfo
}

func genAddChangefeedSpec(
	name, instanceType string,
	downstream *KafkaEntity,
) connector_service.CreateConnectorBody {
	changefeedSpec := connector_service.CreateConnectorBody{
		ConnectorType: model.NewDataflowConnectorType(model.DataflowConnectorTypeKAFKA),
	}
	changefeedSpec.Name = name
	changefeedSpec.ChangefeedInstanceType = instanceType
	changefeedSpec.NetworkInfo = &model.DataflowNetworkInfo{
		NetworkType: model.NewDataflowNetworkType(model.DataflowNetworkTypeNetworkTypePublic),
	}
	start := time.Now()
	startTso := misc.GoTimeToTSO(start)
	changefeedSpec.KafkaInfo = &model.DataflowKafkaInfo{
		Authentication: &model.DataflowAuthentication{
			AuthType: model.DataflowKafkaAuthType(downstream.AuthType).Pointer(),
			UserName: downstream.User,
			Password: downstream.Password,
		},
		Broker: &model.DataflowBroker{
			BrokerEndpoints: downstream.BootstrapEndpoints,
			Compression:     model.NewDataflowCompressionType(model.DataflowCompressionTypeGZIP),
			KafkaVersion:    model.NewDataflowKafkaVersion(model.DataflowKafkaVersionKAFKAVERSION3XX),
			TLSEnable:       false,
		},
		ColumnSelectors: []*model.DataflowColumnSelector{
			{
				Columns: []string{"abc", "def"},
				Matcher: []string{
					"test.a", "test.b"},
			},
			{
				Columns: []string{"abc"},
				Matcher: []string{
					"*.*"},
			},
		},
		DataFormat: &model.DataflowDataFormat{
			Protocol: model.NewDataflowProtocol(model.DataflowProtocolDEBEZIUM),
			DebeziumConfig: &model.DataflowDebeziumConfig{
				DisableSchema:  false,
				OutputOldValue: true,
			},
		},
		StartPosition: &model.DataflowStartPosition{
			Mode: model.NewDataflowStartMode(model.DataflowStartModeFROMNOW),
			Tso:  fmt.Sprintf("%d", startTso),
		},
		TableConfig: &model.DataflowTableConfig{
			EventFilterRule: []*model.DataflowEventFilterRule{
				{
					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.NewDataflowTableMode(model.DataflowTableModeFORCESYNC),
		},
		TopicPartitionConfig: &model.DataflowTopicPartitionConfig{
			DefaultTopic: "test2",
			DispatchType: model.NewDataflowDispatchType(model.DataflowDispatchTypeDISPATCHTYPEONETOPIC),
			PartitionDispatchers: []*model.DataflowPartitionDispatcherConfig{
				{
					Columns: []string{
						"abc",
					},
					Matcher:       []string{"a.b"},
					PartitionType: model.NewDataflowPartitionDispatcher(model.DataflowPartitionDispatcherCOLUMNS),
				},
				{
					IndexName:     "idx",
					Matcher:       []string{"a.b"},
					PartitionType: model.NewDataflowPartitionDispatcher(model.DataflowPartitionDispatcherINDEXVALUE),
				},
				{
					Matcher:       []string{"a.b"},
					PartitionType: model.NewDataflowPartitionDispatcher(model.DataflowPartitionDispatcherTS),
				},
				{
					Matcher:       []string{"a.b"},
					PartitionType: model.NewDataflowPartitionDispatcher(model.DataflowPartitionDispatcherTABLE),
				},
			},
			PartitionNum:      1,
			ReplicationFactor: 1,
			Separator:         "abc",
			TopicPrefix:       "efg",
			TopicSuffix:       "hij",
		},
	}
	return changefeedSpec
}

func genEditChangefeedSpec(
	name, instanceType string,
	downstream *KafkaEntity,
) connector_service.EditConnectorBody {
	changefeedSpec := connector_service.EditConnectorBody{
		ConnectorType: model.NewDataflowConnectorType(model.DataflowConnectorTypeKAFKA),
		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")
}

// ExpectHTTPStatus verifies that err is a go-swagger API error with the given HTTP status code.
func ExpectHTTPStatus(err error, expectedStatus int) error {
	if err == nil {
		return fmt.Errorf("expecting error with HTTP %d but got nil", expectedStatus)
	}
	var apiErr interface {
		Code() int
		Error() string
	}
	if errors.As(err, &apiErr) {
		if apiErr.Code() == expectedStatus {
			return nil
		}
		return fmt.Errorf("expected HTTP %d, got %d: %s", expectedStatus, apiErr.Code(), apiErr.Error())
	}
	return fmt.Errorf("error does not contain HTTP status code: %w", err)
}