package dag

import (
	"context"
	"fmt"
	"strconv"

	flowentity "github.com/tidbcloud/fastflow/pkg/entity"

	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/constant"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

// ProjectMigrationDagParams carries the inputs required to build and launch the
// project-migration dynamic DAG.
type ProjectMigrationDagParams struct {
	OrgID  uint64
	TaskID uint64
	// Migrations is the ordered list of source→target mappings; the DAG creates
	// one migrate-step per entry.
	Migrations []*entity.ProjectMigration
}

// RunProjectMigrationDag constructs and starts a dynamic DAG that:
//  1. Ensures every target project exists.
//  2. Moves Starter/Essential clusters for each source project.
//  3. Marks the ProjectMigrationTask as SUCCEED.
func RunProjectMigrationDag(ctx context.Context, wflow workflow.Workflow, params *ProjectMigrationDagParams) error {
	if params == nil {
		return fmt.Errorf("params is nil")
	}

	metadataJSON, err := metadata.MarshalRequestMetaFromCtx(ctx)
	if err != nil {
		return fmt.Errorf("marshal request metadata: %w", err)
	}

	orgIDStr := strconv.FormatUint(params.OrgID, 10)
	taskIDStr := strconv.FormatUint(params.TaskID, 10)

	// Step 1: ensure all target projects exist.
	tasks := []flowentity.Task{
		{
			ID:         constant.TaskProjectMigrationEnsureTargets,
			ActionName: constant.TaskProjectMigrationEnsureTargets,
		},
	}

	// Step 2: one migrate-step per source project (sequential chain).
	prevTaskID := constant.TaskProjectMigrationEnsureTargets
	for i := range params.Migrations {
		taskID := fmt.Sprintf("%s_%d", constant.TaskProjectMigrationMigrateProject, i)
		tasks = append(tasks, flowentity.Task{
			ID:         taskID,
			ActionName: constant.TaskProjectMigrationMigrateProject,
			DependOn:   []string{prevTaskID},
			// Pass the migration index so the action knows which entry to process.
			Params: map[string]any{
				constant.KeyProjectMigrationProjectIdx: strconv.Itoa(i),
			},
		})
		prevTaskID = taskID
	}

	// Step 3: mark the whole task as SUCCEED.
	tasks = append(tasks, flowentity.Task{
		ID:         constant.TaskProjectMigrationComplete,
		ActionName: constant.TaskProjectMigrationComplete,
		DependOn:   []string{prevTaskID},
	})

	dagVars := map[string]string{
		constant.KeyProjectMigrationTaskID: taskIDStr,
		workflow.KeyTraceId:                workflow.GetTraceId(ctx),
		workflow.KeyRequestMeta:            string(metadataJSON),
	}

	return wflow.RegisterDynamicDagThenRun(ctx, constant.DagPrefixProjectMigration, &workflow.CreateDynamicDagOption{
		OrgId:     orgIDStr,
		ProjectId: "", // org-level task; no single project scope
		BizId:     taskIDStr,
		DagVars:   dagVars,
		DagTasks:  tasks,
	})
}