package dm

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"
)

const dmDatabase = "test1"
const dmTable = "lifecycle_marker"

type FullMigrationLifecycleSmokeCase struct {
	cases.BaseCase
	provider CloudProviderType
}

func NewFullMigrationLifecycleSmokeCase(provider CloudProviderType) cases.E2ECase {
	return &FullMigrationLifecycleSmokeCase{provider: provider}
}
func (c *FullMigrationLifecycleSmokeCase) Name() string {
	return fmt.Sprintf("%sEssentialV1DMFullMigrationLifecycleSmoke", c.provider)
}
func (c *FullMigrationLifecycleSmokeCase) Tags() map[string]string {
	return map[string]string{cases.TagKeyDomain: config.DomainDM, cases.TagKeyAPIModes: common.APIModesBoth}
}
func (c *FullMigrationLifecycleSmokeCase) 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 *FullMigrationLifecycleSmokeCase) Run() {
	cluster, source := c.TiDBServerless("essential"), c.MySQL("mysql")
	c.Step("verify Essential v1 cluster", func() error { return common.ValidateCluster(cluster) })
	suffix := common.UniqueRunSuffix()
	baseline, incremental := c.Name()+"-dm-baseline-"+suffix, c.Name()+"-dm-incremental-"+suffix
	c.Step("initialize DM source table", func() error { return common.WriteMySQLMarker(source, dmDatabase, baseline) })
	client := essential.NewDMClient()
	ctx := c.Context()
	spec := essential.NewMigrationSpec(c.Name()+"-migration", source, dmDatabase, dmTable)
	var precheckID string
	defer func() {
		if precheckID != "" {
			c.RetryDo("cancel unfinished DM precheck", func() error {
				return cancelUnfinishedPrecheck(c.CleanupContext(), client, cluster, precheckID)
			}, 3, time.Minute)
		}
	}()
	var migrationID string
	defer func() {
		if migrationID != "" {
			c.RetryDo("delete DM migration", func() error { return client.Delete(c.CleanupContext(), cluster, migrationID) }, 3, time.Minute)
			c.Expect("DM migration deleted", func() error {
				exists, err := client.Exists(c.CleanupContext(), cluster, migrationID)
				if err != nil {
					return err
				}
				if exists {
					return fmt.Errorf("migration %s still exists", migrationID)
				}
				return nil
			}, 30, 15*time.Second)
		}
	}()
	c.Step("prepare DM source and target", func() error { return client.Prepare(ctx, cluster, source, dmDatabase, dmTable) })
	c.Step("run DM precheck", func() error {
		var err error
		precheckID, err = client.PrecheckSpec(ctx, cluster, spec)
		return err
	})
	c.Expect("DM precheck has no blocking failures", func() error {
		state, failed, err := client.PrecheckState(ctx, cluster, precheckID)
		if err != nil {
			return err
		}
		return common.ValidateDMPrecheckState(state, failed)
	}, 80, 15*time.Second)
	precheckID = ""
	c.Step("create MODE_ALL DM migration", func() error {
		var err error
		migrationID, err = client.CreateSpec(ctx, cluster, spec)
		return err
	})
	c.Expect("DM migration is running", func() error {
		state, err := client.State(ctx, cluster, migrationID)
		if err != nil {
			return err
		}
		return common.ValidateDMState(state, "RUNNING")
	}, 80, 15*time.Second)
	c.Expect("DM full data reached target", func() error { return common.VerifyMySQLMarker(cluster, dmDatabase, baseline) }, 80, 15*time.Second)
	c.Step("write DM incremental source data", func() error { return common.WriteMySQLMarker(source, dmDatabase, incremental) })
	c.Expect("DM incremental data reached target", func() error { return common.VerifyMySQLMarker(cluster, dmDatabase, incremental) }, 80, 15*time.Second)
	c.RetryDo("pause DM migration", func() error { return client.Pause(ctx, cluster, migrationID) }, 3, time.Minute)
	c.Expect("DM migration is paused", func() error {
		state, err := client.State(ctx, cluster, migrationID)
		if err != nil {
			return err
		}
		return common.ValidateDMState(state, "PAUSED")
	}, 80, 15*time.Second)
	c.RetryDo("resume DM migration", func() error { return client.Resume(ctx, cluster, migrationID) }, 3, time.Minute)
	c.Expect("DM migration is running after resume", func() error {
		state, err := client.State(ctx, cluster, migrationID)
		if err != nil {
			return err
		}
		return common.ValidateDMState(state, "RUNNING")
	}, 80, 15*time.Second)
}