package changefeed

import (
	"fmt"
	"time"

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

func NewCDCChangefeedLifecycleSmokeCase(provider CloudProviderType) cases.E2ECase {
	return &CDCChangefeedLifecycleSmokeCase{provider: provider}
}
func (c *CDCChangefeedLifecycleSmokeCase) Name() string {
	return fmt.Sprintf("%sEssentialV1CDCChangefeedLifecycleSmoke", c.provider)
}
func (c *CDCChangefeedLifecycleSmokeCase) Tags() map[string]string {
	return map[string]string{cases.TagKeyDomain: config.DomainCDC, cases.TagKeyAPIModes: common.APIModesBoth}
}
func (c *CDCChangefeedLifecycleSmokeCase) 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 *CDCChangefeedLifecycleSmokeCase) Run() {
	cluster := c.TiDBServerless("essential")
	kafka := c.Kafka("kafka")
	c.Step("verify Essential v1 cluster", func() error { return common.ValidateCluster(cluster) })
	database, topic, suffix := "e2e_essential", c.Name()+"-"+common.UniqueRunSuffix(), common.UniqueRunSuffix()
	changefeedName := c.Name() + "-" + suffix
	if _, err := common.WriteMarker(cluster, c.Name()+"-cdc-baseline-"+suffix); err != nil {
		c.Step("initialize CDC source table", func() error { return err })
		return
	}
	client := essential.NewCDCClient()
	ctx := c.Context()
	filters := []string{database + ".*"}
	c.Step("discover CDC source tables", func() error { return client.DiscoverSourceTables(ctx, cluster, filters) })
	spec := essential.ChangefeedSpec{Name: changefeedName, SinkType: essential.ChangefeedSinkKafka, Kafka: kafka, Topic: topic, Filters: filters}
	c.Step("test CDC Kafka connection", func() error { return client.TestConnection(ctx, cluster, spec) })
	var id string
	defer func() { cleanupChangefeed(&c.BaseCase, client, cluster, id) }()
	c.RetryDo("create CDC changefeed", func() error {
		var err error
		id, err = client.Create(ctx, cluster, spec)
		return err
	}, 3, time.Minute)
	c.Expect("CDC changefeed running", func() error {
		detail, err := client.Get(ctx, cluster, id)
		if err != nil {
			return err
		}
		if err := common.ValidateChangefeedState(detail.State, "RUNNING"); err != nil {
			return err
		}
		return nil
	}, 60, 30*time.Second)
	marker := c.Name() + "-cdc-event-" + suffix
	c.Step("write and consume CDC event", func() error {
		if _, err := common.WriteMarker(cluster, marker); err != nil {
			return err
		}
		return common.ConsumeKafkaEvent(kafka, topic, marker)
	})
	var checkpoint string
	c.Step("pause CDC changefeed", func() error { return client.Stop(ctx, cluster, id) })
	c.Expect("CDC changefeed paused", func() error {
		detail, err := client.Get(ctx, cluster, id)
		checkpoint = detail.Checkpoint
		if err != nil {
			return err
		}
		if err := common.ValidateChangefeedState(detail.State, "PAUSED"); err != nil {
			return err
		}
		return nil
	}, 20, 15*time.Second)
	c.Still("CDC checkpoint remains stable", func() error {
		detail, err := client.Get(ctx, cluster, id)
		if err != nil {
			return err
		}
		if detail.Checkpoint != checkpoint {
			return fmt.Errorf("paused checkpoint changed")
		}
		return nil
	}, 3, 15*time.Second)
	c.Step("resume CDC changefeed", func() error { return client.Start(ctx, cluster, id) })
	c.Expect("CDC changefeed resumed", func() error {
		detail, err := client.Get(ctx, cluster, id)
		if err != nil {
			return err
		}
		if err := common.ValidateChangefeedState(detail.State, "RUNNING"); err != nil {
			return err
		}
		return nil
	}, 30, 15*time.Second)
	resumed := c.Name() + "-cdc-resumed-event-" + suffix
	c.Step("write and consume resumed CDC event", func() error {
		if _, err := common.WriteMarker(cluster, resumed); err != nil {
			return err
		}
		return common.ConsumeKafkaEvent(kafka, topic, resumed)
	})
}