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 CDCKafkaConfigRecreateCase struct {
	cases.BaseCase
	provider CloudProviderType
}

func NewCDCKafkaConfigRecreateCase(provider CloudProviderType) cases.E2ECase {
	return &CDCKafkaConfigRecreateCase{provider: provider}
}
func (c *CDCKafkaConfigRecreateCase) Name() string {
	return fmt.Sprintf("%sEssentialV1CDCKafkaConfigEditRecreateSmoke", c.provider)
}
func (c *CDCKafkaConfigRecreateCase) Tags() map[string]string {
	return map[string]string{cases.TagKeyDomain: config.DomainCDC, cases.TagKeyAPIModes: common.APIModesBoth}
}
func (c *CDCKafkaConfigRecreateCase) 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 *CDCKafkaConfigRecreateCase) Run() {
	cluster, kafka := c.TiDBServerless("essential"), c.Kafka("kafka")
	client := essential.NewCDCClient()
	spec := essential.ChangefeedSpec{Name: c.Name(), SinkType: essential.ChangefeedSinkKafka, Kafka: kafka, Topic: c.Name() + "-topic", Filters: []string{"e2e_essential.*"}}
	var id string
	defer func() { cleanupChangefeed(&c.BaseCase, client, cluster, id) }()
	c.Step("create Kafka changefeed", func() error { var err error; id, err = client.Create(c.Context(), cluster, spec); return err })
	expectKafkaConfigRunning(&c.BaseCase, client, cluster, id)
	first := c.Name() + "-first"
	c.Step("verify initial event", func() error {
		if _, err := common.WriteMarker(cluster, first); err != nil {
			return err
		}
		return common.ConsumeKafkaEvent(kafka, spec.Topic, first)
	})
	c.Step("pause and edit Kafka changefeed", func() error {
		if err := client.Stop(c.Context(), cluster, id); err != nil {
			return err
		}
		return client.Edit(c.Context(), cluster, id, spec)
	})
	c.Step("resume edited Kafka changefeed", func() error { return client.Start(c.Context(), cluster, id) })
	second := c.Name() + "-second"
	c.Step("verify edited event", func() error {
		if _, err := common.WriteMarker(cluster, second); err != nil {
			return err
		}
		return common.ConsumeKafkaEvent(kafka, spec.Topic, second)
	})
	oldID := id
	c.Step("delete before recreate", func() error {
		return client.Delete(c.Context(), cluster, oldID)
	})
	expectKafkaConfigDeleted(&c.BaseCase, client, cluster, oldID)
	id = ""
	c.Step("recreate Kafka changefeed", func() error { var err error; id, err = client.Create(c.Context(), cluster, spec); return err })
	c.Step("validate recreated changefeed id", func() error { return validateRecreatedID(oldID, id) })
	expectKafkaConfigRunning(&c.BaseCase, client, cluster, id)
	c.Step("verify recreated changefeed event", func() error { return verifyRecreatedEvent(cluster, kafka, spec.Topic, c.Name()+"-recreated") })
}