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

func NewCDCKafkaPrivateLinkCase(provider CloudProviderType) cases.E2ECase {
	return &CDCKafkaPrivateLinkCase{provider: provider}
}
func (c *CDCKafkaPrivateLinkCase) Name() string {
	return fmt.Sprintf("%sEssentialV1CDCKafkaPrivateLinkLifecycleSmoke", c.provider)
}
func (c *CDCKafkaPrivateLinkCase) Tags() map[string]string {
	return map[string]string{cases.TagKeyDomain: config.DomainCDC, cases.TagKeyAPIModes: common.APIModesBoth}
}
func (c *CDCKafkaPrivateLinkCase) 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:   DataflowNetworkTypeNetworkTypePrivateLink,
			AuthType:      DataflowKafkaAuthTypeSASLPLAIN,
			CloudProvider: c.provider,
			Region:        region,
		}},
	}
}
func (c *CDCKafkaPrivateLinkCase) Run() {
	cluster, kafka := c.TiDBServerless("essential"), c.Kafka("kafka")
	c.Step("validate Kafka PrivateLink fixture", func() error {
		return validateKafkaPrivateLinkFixture(c.Name(), kafka)
	})
	plClient := essential.NewPrivateLinkClient()
	var plID string
	defer func() { common.CleanupPrivateLink(&c.BaseCase, plClient, cluster, plID) }()
	c.Step("create Kafka PrivateLink connection", func() error {
		detail, err := plClient.Create(c.Context(), cluster, kafka.PrivateLinkServiceName, c.Name())
		if err != nil {
			return err
		}
		plID = detail.ID
		return nil
	})
	common.WaitPrivateLinkActive(&c.BaseCase, plClient, cluster, plID)
	client := essential.NewCDCClient()
	spec := essential.ChangefeedSpec{
		Name:                  c.Name(),
		SinkType:              essential.ChangefeedSinkKafka,
		Kafka:                 kafka,
		Topic:                 c.Name() + "-topic",
		Filters:               []string{"e2e_essential.*"},
		NetworkType:           DataflowNetworkTypeNetworkTypePrivateLink,
		PrivateLinkConnection: plID,
		PrivateLinkPort:       kafka.Ports[0],
	}
	var id string
	defer func() { cleanupChangefeed(&c.BaseCase, client, cluster, id) }()
	c.Step("test Kafka PrivateLink connection", func() error { return client.TestConnection(c.Context(), cluster, spec) })
	c.Step("create Kafka PrivateLink changefeed", func() error { var err error; id, err = client.Create(c.Context(), cluster, spec); return err })
	expectKafkaPrivateLinkRunning(&c.BaseCase, client, cluster, id)
	marker := c.Name() + "-marker"
	c.Step("verify Kafka PrivateLink event", func() error { return verifyKafkaPrivateLinkEvent(cluster, kafka, spec.Topic, marker) })
}