package dm

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

func NewDMIncrementalLifecycleCase(provider CloudProviderType) cases.E2ECase {
	return &DMIncrementalLifecycleCase{provider: provider}
}
func (c *DMIncrementalLifecycleCase) Name() string {
	return fmt.Sprintf("%sEssentialV1DMIncrementalMigrationLifecycleSmoke", c.provider)
}
func (c *DMIncrementalLifecycleCase) Tags() map[string]string {
	return map[string]string{cases.TagKeyDomain: config.DomainDM, cases.TagKeyAPIModes: common.APIModesBoth}
}
func (c *DMIncrementalLifecycleCase) 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 *DMIncrementalLifecycleCase) Run() {
	cluster, source := c.TiDBServerless("essential"), c.MySQL("mysql")
	c.Step("validate Essential cluster", func() error { return common.ValidateCluster(cluster) })
	baseline := c.Name() + "-baseline"
	marker := c.Name() + "-incremental"
	c.Step("initialize incremental source table", func() error { return common.WriteMySQLMarker(source, "test1", baseline) })
	client := essential.NewDMClient()
	spec := essential.NewMigrationSpec(c.Name(), source, "test1", "lifecycle_marker")
	spec.Mode = essential.MigrationModeIncremental
	var precheckID, id string
	defer func() {
		if id != "" {
			c.Step("delete incremental DM migration", func() error {
				return client.Delete(c.CleanupContext(), cluster, id)
			})
			expectIncrementalDeleted(&c.BaseCase, client, cluster, id)
		}
		if precheckID != "" {
			c.Step("cancel unfinished incremental DM precheck", func() error {
				return client.CancelPrecheck(c.CleanupContext(), cluster, precheckID)
			})
		}
	}()
	c.Step("run incremental precheck", func() error {
		precheck, err := client.PrecheckSpec(c.Context(), cluster, spec)
		if err != nil {
			return err
		}
		precheckID = precheck
		expectIncrementalPrecheck(&c.BaseCase, client, cluster, precheck)
		precheckID = ""
		return nil
	})
	c.Step("create incremental migration", func() error { var err error; id, err = client.CreateSpec(c.Context(), cluster, spec); return err })
	expectIncrementalRunning(&c.BaseCase, client, cluster, id)
	verifyIncrementalBaseline(&c.BaseCase, cluster, "test1", baseline)
	c.Step("write incremental source marker after migration starts", func() error { return common.WriteMySQLMarker(source, "test1", marker) })
	verifyIncrementalMarker(&c.BaseCase, cluster, "test1", marker)
}