flowbuilder

Documentation

Index

• type Builder
  • func New(ns *sflow.NodeService, nrs *sflow.NodeRequestService, ...) *Builder
  • func (b *Builder) BuildNodes(ctx context.Context, flow mflow.Flow, nodes []mflow.Node, ...) (map[idwrap.IDWrap]node.FlowNode, []idwrap.IDWrap, error)
  • func (b *Builder) BuildVariables(ctx context.Context, workspaceID idwrap.IDWrap, flowVars []mflow.FlowVariable) (map[string]any, error)
• type SubFlowExecutorImpl
  • func NewSubFlowExecutor(builder *Builder, flowService *sflow.FlowService, ...) *SubFlowExecutorImpl
  • func (e *SubFlowExecutorImpl) ExecuteSubFlow(ctx context.Context, targetFlowID *idwrap.IDWrap, targetFlowName string, ...) (map[string]any, error)

Types

type Builder

type Builder struct {
	Node               *sflow.NodeService
	NodeRequest        *sflow.NodeRequestService
	NodeFor            *sflow.NodeForService
	NodeForEach        *sflow.NodeForEachService
	NodeIf             *sflow.NodeIfService
	NodeJS             *sflow.NodeJsService
	NodeAI             *sflow.NodeAIService
	NodeAiProvider     *sflow.NodeAiProviderService
	NodeMemory         *sflow.NodeMemoryService
	NodeGraphQL        *sflow.NodeGraphQLService
	NodeWsConnection   *sflow.NodeWsConnectionService
	NodeWsSend         *sflow.NodeWsSendService
	NodeWait           *sflow.NodeWaitService
	NodeSubFlowTrigger *sflow.NodeSubFlowTriggerService
	NodeSubFlowReturn  *sflow.NodeSubFlowReturnService
	NodeRunSubFlow     *sflow.NodeRunSubFlowService
	SubFlowExecutor    nrunsubflow.SubFlowExecutor
	WebSocket          *swebsocket.WebSocketService
	WebSocketHeader    *swebsocket.WebSocketHeaderService
	GraphQL            *sgraphql.GraphQLService
	GraphQLHeader      *sgraphql.GraphQLHeaderService

	Workspace    *sworkspace.WorkspaceService
	Variable     *senv.VariableService
	FlowVariable *sflow.FlowVariableService

	Resolver           resolver.RequestResolver
	GraphQLResolver    gqlresolver.GraphQLResolver
	Logger             *slog.Logger
	LLMProviderFactory *scredential.LLMProviderFactory
}

func New

func New(
	ns *sflow.NodeService,
	nrs *sflow.NodeRequestService,
	nfs *sflow.NodeForService,
	nfes *sflow.NodeForEachService,
	nifs *sflow.NodeIfService,
	njss *sflow.NodeJsService,
	nais *sflow.NodeAIService,
	naps *sflow.NodeAiProviderService,
	nmems *sflow.NodeMemoryService,
	ngqs *sflow.NodeGraphQLService,
	nwcs *sflow.NodeWsConnectionService,
	nwss *sflow.NodeWsSendService,
	nwaits *sflow.NodeWaitService,
	nsfts *sflow.NodeSubFlowTriggerService,
	nsfrs *sflow.NodeSubFlowReturnService,
	nrsfs *sflow.NodeRunSubFlowService,
	wsSvc *swebsocket.WebSocketService,
	wsHeaderSvc *swebsocket.WebSocketHeaderService,
	gqls *sgraphql.GraphQLService,
	gqlhs *sgraphql.GraphQLHeaderService,
	ws *sworkspace.WorkspaceService,
	vs *senv.VariableService,
	fvs *sflow.FlowVariableService,
	resolver resolver.RequestResolver,
	graphQLResolver gqlresolver.GraphQLResolver,
	logger *slog.Logger,
	llmFactory *scredential.LLMProviderFactory,
) *Builder

func (*Builder) BuildNodes

func (b *Builder) BuildNodes(
	ctx context.Context,
	flow mflow.Flow,
	nodes []mflow.Node,
	timeout time.Duration,
	httpClient httpclient.HttpClient,
	respChan chan nrequest.NodeRequestSideResp,
	gqlRespChan chan ngraphql.NodeGraphQLSideResp,
	jsClient node_js_executorv1connect.NodeJsExecutorServiceClient,
) (map[idwrap.IDWrap]node.FlowNode, []idwrap.IDWrap, error)

func (*Builder) BuildVariables

func (b *Builder) BuildVariables(
	ctx context.Context,
	workspaceID idwrap.IDWrap,
	flowVars []mflow.FlowVariable,
) (map[string]any, error)

type SubFlowExecutorImpl

type SubFlowExecutorImpl struct {
	Builder     *Builder
	FlowService *sflow.FlowService
	EdgeService *sflow.EdgeService
	JSClient    node_js_executorv1connect.NodeJsExecutorServiceClient
	Logger      *slog.Logger

	// Optional execution tracking: when set, sub-flow runs create flow version
	// history entries and persist node execution records. When nil (e.g. CLI),
	// sub-flows execute without history.
	NodeExecutionService   *sflow.NodeExecutionService
	HTTPResponseService    shttp.HttpResponseService
	GraphQLResponseService sgraphql.GraphQLResponseService
	EventPublisher         flowresult.EventPublisher
}

SubFlowExecutorImpl implements nrunsubflow.SubFlowExecutor using database services. It loads the target flow from the DB, builds nodes using the Builder, and runs the flow synchronously via FlowLocalRunner.

func NewSubFlowExecutor

func NewSubFlowExecutor(
	builder *Builder,
	flowService *sflow.FlowService,
	edgeService *sflow.EdgeService,
	jsClient node_js_executorv1connect.NodeJsExecutorServiceClient,
	logger *slog.Logger,
) *SubFlowExecutorImpl

func (*SubFlowExecutorImpl) ExecuteSubFlow

func (e *SubFlowExecutorImpl) ExecuteSubFlow(ctx context.Context, targetFlowID *idwrap.IDWrap, targetFlowName string, inputVars map[string]any) (map[string]any, error)