Documentation
¶
Overview ¶
Package nodes provides the generic, deterministic node types every dataflow graph can use regardless of tenant/product (http_request, filter, if, switch, set, merge, wait) — no credentials, no multi-tenant concerns. Database and SaaS connector node types build on this package but live separately (see pkg/dataflow/nodes/database, Phase 2).
Index ¶
Constants ¶
const ( ScheduleTriggerModeCron = "cron" ScheduleTriggerModeInterval = "interval" ScheduleTriggerModeOnce = "once" )
const ( WebhookResponseModeImmediate = "immediate" WebhookResponseModeWhenFinished = "whenFinished" )
Variables ¶
This section is empty.
Functions ¶
Types ¶
type Filter ¶
type Filter struct {
// contains filtered or unexported fields
}
Filter is the "filter" node type: keeps only items where expression evaluates true, evaluated per item with $json bound to that item — same goja/"$"-prefixed convention as internal/inbox/event_filter.go's $event, so an LLM authoring a graph reuses one expression language across the SDK instead of two.
func (*Filter) Description ¶
func (n *Filter) Description() dataflow.NodeDescription
Description returns the catalog entry of the filter node: it keeps only the items for which a boolean JS expression is true.
func (*Filter) Execute ¶
func (n *Filter) Execute(ctx context.Context, rt *dataflow.Runtime, input []dataflow.Item, params map[string]any) (dataflow.Output, error)
Execute evaluates the expression for each item, with the item bound as $json, and passes on to the main port only the items for which it is true. An expression that fails to evaluate fails the node.
type HTTPRequest ¶
type HTTPRequest struct {
// contains filtered or unexported fields
}
HTTPRequest is the "http_request" node type: a single HTTP call per Run (not per input item — chain it after a node that produces one item per call if per-item requests are needed).
Parameters: method (default GET), url (required), headers (map[string]string), body (string, sent as-is with Content-Type left to headers).
func NewHTTPRequest ¶
func NewHTTPRequest() *HTTPRequest
NewHTTPRequest creates an HTTP Request node with a 30-second timeout that refuses private, internal and cloud metadata addresses, also when a redirect or a DNS answer leads to one.
func (*HTTPRequest) Description ¶
func (n *HTTPRequest) Description() dataflow.NodeDescription
Description returns the catalog entry of the http_request node: it makes a single HTTP call and returns the response as one item.
func (*HTTPRequest) Execute ¶
func (n *HTTPRequest) Execute(ctx context.Context, rt *dataflow.Runtime, input []dataflow.Item, params map[string]any) (dataflow.Output, error)
Execute makes the call described by url, method, headers and body, whose values may be expressions resolved against the first input item, and returns one item with the status, the headers and either the parsed JSON ("json") or the text ("body") of the response, which is read up to 10 MB.
func (*HTTPRequest) ValidateParameters ¶
func (n *HTTPRequest) ValidateParameters(params map[string]any) error
ValidateParameters requires a url and refuses one that points at a private, internal or cloud metadata address.
type If ¶
type If struct {
// contains filtered or unexported fields
}
If is the "if" node type: splits items into "true"/"false" output ports by expression — unlike Filter (which drops non-matching items), both branches are wired to different downstream nodes.
func (*If) Description ¶
func (n *If) Description() dataflow.NodeDescription
Description returns the catalog entry of the if node: it routes each item to the "true" or the "false" port according to a boolean JS expression.
func (*If) Execute ¶
func (n *If) Execute(ctx context.Context, rt *dataflow.Runtime, input []dataflow.Item, params map[string]any) (dataflow.Output, error)
Execute evaluates the expression for each item, with the item bound as $json, and sends the item to the "true" port when it is true and to the "false" port otherwise. An expression that fails to evaluate fails the node.
type Merge ¶
type Merge struct{}
Merge is the "merge" node type. The engine already concatenates items from every upstream connection into one input slice (see engine.go), so by default this is an identity pass-through — its value is as an explicit, labeled join point in a graph rather than as a transformation. Fancier merge strategies (pair-by-index, join-by-key) are deliberately not built yet; add them here if/when a workflow actually needs one.
func (Merge) Description ¶
func (Merge) Description() dataflow.NodeDescription
Description returns the catalog entry of the merge node: it joins the items of several upstream branches into one list.
type ScheduleTrigger ¶
type ScheduleTrigger struct{}
ScheduleTrigger is the "schedule_trigger" node type: a graph's time-based entry point. It carries schedule config (mode/cronExpr/intervalSeconds/ runAt) rather than transforming data - a caller that hosts the graph (e.g. seshat-server's job scheduler) reads these parameters to decide when to run the graph at all, mirroring the same three trigger modes a scheduled job already supports. Execute is a pure passthrough: Run already seeds every node with no predecessors (this one included) with the run's input, so there is no real per-node work to do here - the node exists for its config and its IsTrigger marker, not its Execute logic.
func NewScheduleTrigger ¶
func NewScheduleTrigger() ScheduleTrigger
NewScheduleTrigger creates a ScheduleTrigger node.
func (ScheduleTrigger) Description ¶
func (ScheduleTrigger) Description() dataflow.NodeDescription
Description returns the catalog entry of the schedule trigger node: it starts the graph on a schedule, given as a cron expression, a fixed interval or a single future time.
func (ScheduleTrigger) Execute ¶
func (ScheduleTrigger) Execute(_ context.Context, _ *dataflow.Runtime, input []dataflow.Item, _ map[string]any) (dataflow.Output, error)
Execute passes its input on unchanged: the schedule itself is handled by the automation that owns the graph.
func (ScheduleTrigger) ValidateParameters ¶
func (ScheduleTrigger) ValidateParameters(params map[string]any) error
ValidateParameters deliberately checks presence/shape only, not full cron syntax - pkg/automation (which can actually parse a cron expression) isn't importable here without an import cycle (pkg/automation pulls in internal/automation, whose own test suite imports pkg/dataflow/nodes to exercise the dataflow adapter). Real cron validation still happens where it already did, server-side, when the job is saved.
type Set ¶
type Set struct {
// contains filtered or unexported fields
}
Set is the "set" node type: adds/overwrites fields on every item. Parameters: fields (map[string]any) — a string value starting with "=" (n8n's convention for "this is an expression, not a literal") is evaluated per item with $json bound to the item being modified; everything else is copied through as a literal.
func (*Set) Description ¶
func (n *Set) Description() dataflow.NodeDescription
Description returns the catalog entry of the set node: it adds or overwrites fields on every item, a string value that starts with "=" being evaluated as a JS expression.
func (*Set) Execute ¶
func (n *Set) Execute(ctx context.Context, rt *dataflow.Runtime, input []dataflow.Item, params map[string]any) (dataflow.Output, error)
Execute returns the items with the fields added or overwritten and the other fields kept. A field value that is a string starting with "=" is evaluated as a JS expression with the item bound as $json; any other value is used as it is.
type Switch ¶
type Switch struct {
// contains filtered or unexported fields
}
Switch is the "switch" node type: like If but with more than two branches. Parameters: cases (map[string]string, port name -> boolean expression), evaluated in the order given by casesOrder ([]string, since map iteration order isn't stable and case order matters — first match wins). An item matching no case routes to the "default" port.
func NewSwitch ¶
NewSwitch creates a Switch node that evaluates the expressions of its cases with pool.
func (*Switch) Description ¶
func (n *Switch) Description() dataflow.NodeDescription
Description returns the catalog entry of the switch node: it routes each item to the port of the first case whose expression is true, or to the "default" port when none is.
func (*Switch) Execute ¶
func (n *Switch) Execute(ctx context.Context, rt *dataflow.Runtime, input []dataflow.Item, params map[string]any) (dataflow.Output, error)
Execute tries the cases in the order of casesOrder for each item, with the item bound as $json, and sends the item to the port of the first case whose expression is true, or to the "default" port when none is.
type Wait ¶
type Wait struct{}
Wait is the "wait" node type: pauses before passing items through unchanged. Parameters: seconds (int, required, 0 < seconds <= 600).
func (Wait) Description ¶
func (Wait) Description() dataflow.NodeDescription
Description returns the catalog entry of the wait node: it pauses for a fixed number of seconds, then passes the items through unchanged.
func (Wait) Execute ¶
func (Wait) Execute(ctx context.Context, _ *dataflow.Runtime, input []dataflow.Item, params map[string]any) (dataflow.Output, error)
Execute waits for the given number of seconds, or until ctx is cancelled, in which case it returns the error of the context, then passes the items through unchanged.
type WebhookTrigger ¶
type WebhookTrigger struct{}
WebhookTrigger is the "webhook_trigger" node type: a graph's HTTP entry point. It carries the expected HTTP method rather than transforming data - the caller that hosts the graph (seshat-server) reads this parameter to decide which incoming requests are allowed to start a run, and mints/owns the actual URL (a per-job random token, not a user-chosen path - see the job-side trigger extraction for why). Execute is a pure passthrough, same reasoning as ScheduleTrigger: Run already seeds every node with no predecessors (this one included) with the run's input.
func NewWebhookTrigger ¶
func NewWebhookTrigger() WebhookTrigger
NewWebhookTrigger creates a WebhookTrigger node.
func (WebhookTrigger) Description ¶
func (WebhookTrigger) Description() dataflow.NodeDescription
Description returns the catalog entry of the webhook trigger node: it starts the graph when an external HTTP call reaches the webhook URL of the job, and gives the request body, headers and query to the graph under a "webhook" key.
func (WebhookTrigger) Execute ¶
func (WebhookTrigger) Execute(_ context.Context, _ *dataflow.Runtime, input []dataflow.Item, _ map[string]any) (dataflow.Output, error)
Execute passes its input on unchanged: the request has already been turned into the input of the run.
func (WebhookTrigger) ValidateParameters ¶
func (WebhookTrigger) ValidateParameters(params map[string]any) error
ValidateParameters checks that method, when given, is one of GET, POST, PUT, PATCH or DELETE.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package database provides deterministic node executors for querying external databases from a dataflow graph — Postgres, MySQL, SQLite (pkg/database/sql, three thin drivers sharing one executor), Redis, and MongoDB (native drivers), and Elasticsearch (its own REST API, no client SDK needed).
|
Package database provides deterministic node executors for querying external databases from a dataflow graph — Postgres, MySQL, SQLite (pkg/database/sql, three thin drivers sharing one executor), Redis, and MongoDB (native drivers), and Elasticsearch (its own REST API, no client SDK needed). |