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/utilities/grpc/metadata"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

type CreateDataSynchronizationDagTasks struct {
	TenantId              uint64
	ProjectId             uint64
	DataSynchronizationId uint64
	CrossProject          bool
	SpanName              string
}

// RunCreateDataSynchronizationDag builds a dynamic DAG based on whether it is cross-project.
//
// - Same project: create changefeed -> wait changefeed running
// - Cross project: create endpoint -> approve endpoint -> wait endpoint active -> create changefeed -> wait changefeed running
func RunCreateDataSynchronizationDag(ctx context.Context, wflow workflow.Workflow, params *CreateDataSynchronizationDagTasks) error {
	if params == nil {
		return fmt.Errorf("params is nil")
	}

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

	var tasks []flowentity.Task
	if params.CrossProject {
		tasks = []flowentity.Task{
			{
				ID:         constant.TaskDataSynchronizationCreateSinkEndpoint,
				ActionName: constant.TaskDataSynchronizationCreateSinkEndpoint,
			},
			{
				ID:         constant.TaskDataSynchronizationApproveSinkEndpoint,
				ActionName: constant.TaskDataSynchronizationApproveSinkEndpoint,
				DependOn:   []string{constant.TaskDataSynchronizationCreateSinkEndpoint},
			},
			{
				ID:         constant.TaskDataSynchronizationWaitSinkEndpointActive,
				ActionName: constant.TaskDataSynchronizationWaitSinkEndpointActive,
				DependOn:   []string{constant.TaskDataSynchronizationApproveSinkEndpoint},
			},
			{
				ID:         constant.TaskDataSynchronizationCreateChangefeed,
				ActionName: constant.TaskDataSynchronizationCreateChangefeed,
				DependOn:   []string{constant.TaskDataSynchronizationWaitSinkEndpointActive},
			},
			{
				ID:         constant.TaskDataSynchronizationWaitChangefeedRunning,
				ActionName: constant.TaskDataSynchronizationWaitChangefeedRunning,
				DependOn:   []string{constant.TaskDataSynchronizationCreateChangefeed},
			},
		}
	} else {
		tasks = []flowentity.Task{
			{
				ID:         constant.TaskDataSynchronizationCreateChangefeed,
				ActionName: constant.TaskDataSynchronizationCreateChangefeed,
			},
			{
				ID:         constant.TaskDataSynchronizationWaitChangefeedRunning,
				ActionName: constant.TaskDataSynchronizationWaitChangefeedRunning,
				DependOn:   []string{constant.TaskDataSynchronizationCreateChangefeed},
			},
		}
	}

	dagVars := map[string]string{
		constant.KeyDataSynchronizationID: strconv.FormatUint(params.DataSynchronizationId, 10),
		workflow.KeyTraceId:               workflow.GetTraceId(ctx),
		workflow.KeySpanName:              params.SpanName,
		workflow.KeyRequestMeta:           string(metadataJson),
	}

	return wflow.RegisterDynamicDagThenRun(ctx, constant.DagPrefixDataSynchronizationCreate, &workflow.CreateDynamicDagOption{
		OrgId:     strconv.FormatUint(params.TenantId, 10),
		ProjectId: strconv.FormatUint(params.ProjectId, 10),
		BizId:     strconv.FormatUint(params.DataSynchronizationId, 10),
		DagVars:   dagVars,
		DagTasks:  tasks,
	})
}