package dm

import (
	"fmt"
	"time"

	. "github.com/tidbcloud/dataflow-service-e2e/cases"
	"github.com/tidbcloud/dataflow-service-e2e/cases/dedicated/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/dedicated/consoleapi"
	dm_model "github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/dm/model"
)

// dmJobLifecycleSmokeCase tests the complete lifecycle of a DM job.
// It verifies the entire control path: Create → Running → Pause → Paused → Resume → Running → Delete.
type dmJobLifecycleSmokeCase struct {
	BaseCase
	dmValidator   *common.DMValidator
	cloudProvider CloudProviderType
}

// NewDMJobLifecycleSmokeCase creates a new job lifecycle smoke test case.
func NewDMJobLifecycleSmokeCase(cloudProvider CloudProviderType) E2ECase {
	return &dmJobLifecycleSmokeCase{cloudProvider: cloudProvider}
}

func (m *dmJobLifecycleSmokeCase) Name() string {
	return fmt.Sprintf("%sDMJobLifecycleSmoke", m.cloudProvider)
}

func (m *dmJobLifecycleSmokeCase) Tags() map[string]string {
	return map[string]string{TagKeyDomain: TagValueDomainDM}
}

func (m *dmJobLifecycleSmokeCase) DeclareDependencyRequirements() map[string]DependencyRequirement {
	return map[string]DependencyRequirement{
		"tidb": &TiDBDedicatedRequirement{
			Dedicated: false,
			TiDBDedicatedDimensions: TiDBDedicatedDimensions{
				CloudProvider: m.cloudProvider,
				Region:        config.GlobalConfig.DefaultRegionOf(m.cloudProvider),
			},
		},
		"mysql": &MySQLRequirement{
			MySQLDimensions: MySQLDimensions{
				NetworkType: DataflowNetworkTypeNetworkTypePublic,
			},
		},
	}
}

func (m *dmJobLifecycleSmokeCase) Run() {
	tidbCluster := m.TiDBDedicated("tidb")
	mysql := m.MySQL("mysql")
	m.dmValidator = common.NewDMValidator(m.Logger())

	var jobID string
	var specID string
	var connProfileID string
	var precheckID string

	// Cleanup: delete the created DM job
	defer func() {
		if jobID != "" {
			m.RetryDo(fmt.Sprintf("cleanup: delete DM job %s", jobID), func() error {
				return m.dmValidator.Client().DeleteJob(tidbCluster, jobID)
			}, DefaultRetryAttempts, DefaultRetryInterval)
		}
	}()

	m.
		// Step 1: Get available DM job spec
		Step("get DM job spec", func() error {
			spec, err := m.dmValidator.GetFirstJobSpec()
			if err != nil {
				return err
			}
			specID = spec.ID
			return nil
		}).

		// Step 2: Register connection profile for the MySQL source
		Step("create conn profile", func() error {
			id, err := consoleapi.NewConnProfileClientWrapper(m.Logger()).CreateConnProfile(
				tidbCluster, mysql, GenerateDMJobName("cp"),
			)
			if err != nil {
				return err
			}
			connProfileID = id
			return nil
		}).

		// Step 3: Submit precheck
		Step("submit DM precheck", func() error {
			body := BuildPrecheckBody(
				GenerateDMJobName("pc"), DMModeIncremental, DMImportModeLogical, specID,
				tidbCluster, mysql, false,
			)
			id, err := m.dmValidator.Client().PrecheckJob(tidbCluster, body)
			if err != nil {
				return err
			}
			precheckID = id
			m.Logger().Info("precheck submitted: %s", precheckID)
			return nil
		}).

		// Step 4: Wait for precheck to complete
		Expect("precheck is done", func() error {
			return m.dmValidator.IsPrecheckDone(tidbCluster, precheckID)
		}, uint(PrecheckTimeout/(30*time.Second)), 30*time.Second).

		// Step 5: Create an incremental-mode DM job (incremental stays running, ideal for lifecycle testing)
		Step("create DM job", func() error {
			jobName := GenerateDMJobName("lifecycle")
			body := BuildCreateJobBody(
				jobName, DMModeIncremental, DMImportModeLogical, specID,
				tidbCluster, connProfileID, precheckID, false, []string{DefaultDBForDMTest},
			)
			resp, err := m.dmValidator.Client().CreateJob(tidbCluster, body)
			if err != nil {
				return err
			}
			jobID = resp.Payload.ID
			m.Logger().Info("created DM job for lifecycle test: %s", jobID)
			return nil
		}).

		// Step 6: Verify job can be listed
		Expect("DM job appears in list", func() error {
			resp, err := m.dmValidator.Client().ListJobs(tidbCluster, 1, 100)
			if err != nil {
				return err
			}
			for _, job := range resp.Jobs {
				if job.ID == jobID {
					return nil
				}
			}
			return fmt.Errorf("job %s not found in list", jobID)
		}, 10, 30*time.Second).

		// Step 7: Wait for job to start running
		Expect("DM job is running", func() error {
			return m.dmValidator.IsJobRunning(tidbCluster, jobID)
		}, DefaultExpectAttempts, DefaultExpectInterval).

		// Step 8: Verify job details via GetJob
		Step("verify job details", func() error {
			job, err := m.dmValidator.Client().GetJob(tidbCluster, jobID)
			if err != nil {
				return err
			}
			if job.Mode != DMModeIncremental {
				return fmt.Errorf("job mode is %s, expected %s", job.Mode, DMModeIncremental)
			}
			m.Logger().Info("job details verified: mode=%s, spec=%s", job.Mode, job.Spec.DisplayName)
			return nil
		}).

		// Step 9: Pause the job
		Step("pause DM job", func() error {
			return m.dmValidator.Client().OperateJob(tidbCluster, jobID, dm_model.DmOperateJobReqOperationPause)
		}).

		// Step 10: Wait for job to be paused
		Expect("DM job is paused", func() error {
			return m.dmValidator.IsJobPaused(tidbCluster, jobID)
		}, 20, DefaultExpectInterval).

		// Step 11: Resume the job
		Step("resume DM job", func() error {
			return m.dmValidator.Client().OperateJob(tidbCluster, jobID, dm_model.DmOperateJobReqOperationResume)
		}).

		// Step 12: Wait for job to be running again
		Expect("DM job is running after resume", func() error {
			return m.dmValidator.IsJobRunning(tidbCluster, jobID)
		}, DefaultExpectAttempts, DefaultExpectInterval).

		// Step 13: Delete the job
		RetryDo("delete DM job", func() error {
			return m.dmValidator.Client().DeleteJob(tidbCluster, jobID)
		}, DefaultRetryAttempts, DefaultRetryInterval).

		// Step 14: Verify job is deleted
		Expect("DM job is deleted", func() error {
			return m.dmValidator.IsJobDeleted(tidbCluster, jobID)
		}, 10, 30*time.Second)

	jobID = ""
}