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

func NewDMPrivateLinkCase(provider CloudProviderType) cases.E2ECase {
	return &DMPrivateLinkCase{provider: provider}
}
func (c *DMPrivateLinkCase) Name() string {
	return fmt.Sprintf("%sEssentialV1DMPrivateLinkMigrationLifecycleSmoke", c.provider)
}
func (c *DMPrivateLinkCase) Tags() map[string]string {
	return map[string]string{cases.TagKeyDomain: config.DomainDM, cases.TagKeyAPIModes: common.APIModesBoth}
}
func (c *DMPrivateLinkCase) 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:   DataflowNetworkTypeNetworkTypePrivateLink,
			CloudProvider: c.provider,
			Region:        region,
		}},
	}
}
func (c *DMPrivateLinkCase) Run() {
	cluster, source := c.TiDBServerless("essential"), c.MySQL("mysql")
	c.Step("validate DM PrivateLink fixture", func() error { return common.RequireMySQLPrivateLinkFixture(c.Name(), source) })
	plClient := essential.NewPrivateLinkClient()
	var plID string
	defer func() { common.CleanupPrivateLink(&c.BaseCase, plClient, cluster, plID) }()
	c.Step("create DM PrivateLink connection", func() error {
		detail, err := plClient.Create(c.Context(), cluster, source.PrivateLinkServiceName, c.Name())
		if err != nil {
			return err
		}
		plID = detail.ID
		return nil
	})
	common.WaitPrivateLinkActive(&c.BaseCase, plClient, cluster, plID)
	baseline, incremental := c.Name()+"-full", c.Name()+"-incremental"
	c.Step("write full migration source row", func() error { return common.WriteMySQLMarker(source, "test1", baseline) })
	client := essential.NewDMClient()
	spec := essential.NewMigrationSpec(c.Name(), source, "test1", "lifecycle_marker")
	spec.NetworkType = DataflowNetworkTypeNetworkTypePrivateLink
	spec.PrivateLinkConnection = plID
	spec.PrivateLinkPort = source.Ports[0]
	var precheckID, migrationID string
	defer func() {
		if migrationID != "" {
			c.Step("delete PrivateLink DM migration", func() error {
				return client.Delete(c.CleanupContext(), cluster, migrationID)
			})
			expectPrivateLinkDeleted(&c.BaseCase, client, cluster, migrationID)
		}
		if precheckID != "" {
			c.Step("cancel unfinished PrivateLink DM precheck", func() error {
				return client.CancelPrecheck(c.CleanupContext(), cluster, precheckID)
			})
		}
	}()
	c.Step("create PrivateLink precheck", func() error {
		var err error
		precheckID, err = client.PrecheckSpec(c.Context(), cluster, spec)
		return err
	})
	expectPrivateLinkPrecheck(&c.BaseCase, client, cluster, precheckID)
	precheckID = ""
	c.Step("create PrivateLink migration", func() error {
		var err error
		migrationID, err = client.CreateSpec(c.Context(), cluster, spec)
		return err
	})
	expectPrivateLinkRunning(&c.BaseCase, client, cluster, migrationID)
	c.Step("write incremental source row", func() error { return common.WriteMySQLMarker(source, "test1", incremental) })
	verifyPrivateLinkRows(&c.BaseCase, cluster, baseline, incremental)
}