package changefeed

import (
	"fmt"
	"github.com/tidbcloud/dataflow-service-e2e/cases"
	"github.com/tidbcloud/dataflow-service-e2e/cases/essential/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/essential"
)

type CDCInvalidParamsCase struct {
	cases.BaseCase
	provider CloudProviderType
}

func NewCDCInvalidParamsCase(provider CloudProviderType) cases.E2ECase {
	return &CDCInvalidParamsCase{provider: provider}
}
func (c *CDCInvalidParamsCase) Name() string {
	return fmt.Sprintf("%sEssentialV1CDCInvalidParamsSmoke", c.provider)
}
func (c *CDCInvalidParamsCase) Tags() map[string]string {
	return map[string]string{cases.TagKeyDomain: config.DomainCDC, cases.TagKeyAPIModes: common.APIModesBoth}
}
func (c *CDCInvalidParamsCase) DeclareDependencyRequirements() map[string]DependencyRequirement {
	region := config.GlobalConfig.DefaultRegionOf(c.provider)
	return map[string]DependencyRequirement{
		"essential": &TiDBServerlessRequirement{
			Dedicated: true,
			TiDBServerlessDimensions: TiDBServerlessDimensions{
				CloudProvider: c.provider,
				Region:        region,
				ServicePlan:   essential.ServicePlanEssential,
			},
		},
		"kafka": &KafkaRequirement{KafkaDimensions: KafkaDimensions{
			NetworkType: DataflowNetworkTypeNetworkTypePublic,
			AuthType:    DataflowKafkaAuthTypeSASLPLAIN,
		}},
	}
}
func (c *CDCInvalidParamsCase) Run() {
	cluster, kafka := c.TiDBServerless("essential"), c.Kafka("kafka")
	invalid := kafka.Copy()
	invalid.BootstrapEndpoints = ""
	spec := essential.ChangefeedSpec{Name: c.Name() + "-invalid", SinkType: essential.ChangefeedSinkKafka, Kafka: invalid, Topic: "", Filters: []string{"e2e_essential.*"}}
	c.Step("reject invalid Kafka request", func() error {
		return validateConnectionError(essential.NewCDCClient().TestConnection(c.Context(), cluster, spec))
	})
}