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

func NewCDCMySQLLifecycleCase(provider CloudProviderType) cases.E2ECase {
	return &CDCMySQLLifecycleCase{provider: provider}
}
func (c *CDCMySQLLifecycleCase) Name() string {
	return fmt.Sprintf("%sEssentialV1CDCMySQLChangefeedLifecycleSmoke", c.provider)
}
func (c *CDCMySQLLifecycleCase) Tags() map[string]string {
	return map[string]string{cases.TagKeyDomain: config.DomainCDC, cases.TagKeyAPIModes: common.APIModesBoth}
}
func (c *CDCMySQLLifecycleCase) 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,
			},
		},
		"mysql": &MySQLRequirement{MySQLDimensions: MySQLDimensions{
			NetworkType: DataflowNetworkTypeNetworkTypePublic,
		}},
	}
}
func (c *CDCMySQLLifecycleCase) Run() {
	cluster, source := c.TiDBServerless("essential"), c.MySQL("mysql")
	c.Step("validate source and target", func() error { return common.ValidateCluster(cluster) })
	marker := c.Name() + "-marker-" + common.UniqueRunSuffix()
	c.Step("prepare MySQL sink table", func() error { return common.PrepareMySQLMarkerTarget(source, "e2e_essential") })
	client := essential.NewCDCClient()
	spec := essential.ChangefeedSpec{Name: c.Name(), SinkType: essential.ChangefeedSinkMySQL, MySQL: source, Filters: []string{"e2e_essential.*"}}
	var id string
	defer func() { cleanupChangefeed(&c.BaseCase, client, cluster, id) }()
	c.Step("test MySQL sink", func() error { return client.TestConnection(c.Context(), cluster, spec) })
	c.Step("create MySQL changefeed", func() error { var err error; id, err = client.Create(c.Context(), cluster, spec); return err })
	expectMySQLLifecycleRunning(&c.BaseCase, client, cluster, id)
	c.Step("write source marker after CDC starts", func() error { _, err := common.WriteMarker(cluster, marker); return err })
	c.Expect("MySQL marker replicated", func() error {
		return common.VerifyMySQLFixtureMarker(source, "e2e_essential", marker)
	}, 60, 15*time.Second)
}