package workflow

import (
	"context"
	"fmt"

	"github.com/tidbcloud/fastflow"
	"github.com/tidbcloud/fastflow/pkg/entity"
	"github.com/tidbcloud/fastflow/pkg/mod"
)

type CreateDynamicDagOption struct {
	OrgId     string
	ProjectId string
	BizId     string
	DagVars   map[string]string
	DagTasks  []entity.Task
}

func (impl *workflowImpl) RegisterDynamicDagThenRun(
	ctx context.Context,
	prefix string,
	opt *CreateDynamicDagOption,
) error {
	for _, task := range opt.DagTasks {
		_, ok := fastflow.GetAction(task.ActionName)
		if !ok {
			return fmt.Errorf("action %s is not registered", task.ActionName)
		}
	}

	vars := make(map[string]entity.DagVar)
	for key := range opt.DagVars {
		vars[key] = entity.DagVar{}
	}

	id := fmt.Sprintf("%s_%s_%s_%d", DynamicDagPrefix, prefix, opt.BizId, impl.idGenerator.MustNew())
	dag := &entity.Dag{
		BaseInfo: entity.BaseInfo{
			ID: id,
		},
		Name:   id,
		Status: entity.DagStatusNormal,
		Vars:   vars,
		Tasks:  opt.DagTasks,
	}

	_, err := mod.BuildRootNode(mod.MapTasksToGetter(dag.Tasks))
	if err != nil {
		return err
	}

	dag.Initial()
	if err := impl.provider.WriteDB(ctx).Create(dag).Error; err != nil {
		return err
	}

	return impl.RunDagWithBizId(ctx, id, opt.OrgId, opt.ProjectId, opt.BizId, opt.DagVars)
}