@spec add_node(t(), String.t() | atom(), module() | String.t(), keyword()) :: t()

Adds a node to the flow.

name identifies the node within the workflow (and is the key of its output in context.results). module is a module implementing Sig.Node. Adding a node with an existing name replaces it.

Options

• :depends_on - list of node names that must complete before this node runs. Defaults to [].
• :params - params for this node only. They take precedence over the execution and workflow params. Defaults to %{}.
• :max_retries - total number of attempts before the node (and the execution) is marked as failed. Defaults to 3.
• :backoff_ms - base delay between attempts. The delay doubles after every failed attempt (backoff_ms * 2^(attempt - 1)), capped at 30 seconds. Use 0 to retry immediately. Defaults to 500.

The graph is not validated here; validation happens in Sig.create_workflow/1 (see validate/1).

Examples

flow
|> Sig.Flow.add_node(:charge, MyApp.Nodes.Charge,
  depends_on: [:fetch_user],
  params: %{"amount" => 50},
  max_retries: 5,
  backoff_ms: 1_000
)

@spec find_runnable_nodes(%{required(String.t()) => node_def()}, [String.t()]) :: [
  {String.t(), node_def()}
]

Returns the nodes that are ready to run.

A node is runnable when it is not in completed_names and all of its dependencies are. nodes must already be normalized (see normalize_nodes/1).

Examples

iex> nodes = Sig.Flow.normalize_nodes(%{"a" => %{module: A}, "b" => %{module: B, depends_on: ["a"]}})
iex> nodes |> Sig.Flow.find_runnable_nodes(["a"]) |> Enum.map(&elem(&1, 0))
["b"]

@spec new(String.t() | atom() | nil, keyword()) :: t()

Creates an empty flow.

Options

• :description - a free-form description.
• :params - workflow-level params, merged into the params of every node.
• :context - workflow-level context, shared by every node.

Examples

Sig.Flow.new("reports", params: %{"format" => "pdf"}, context: %{"tenant" => "acme"})

@spec normalize_nodes(map() | [map()]) :: %{required(String.t()) => node_def()}

Normalizes node definitions to the string-keyed shape described in the module docs, filling in defaults for missing fields.

Accepts either a map of node name to definition, or a list of definitions that carry a :name (or "name") key. Atom and string keys are both accepted.

Examples

iex> Sig.Flow.normalize_nodes(%{fetch: %{module: MyNode}})
%{
  "fetch" => %{
    "name" => "fetch",
    "module" => "Elixir.MyNode",
    "params" => %{},
    "depends_on" => [],
    "max_retries" => 3,
    "backoff_ms" => 500
  }
}

@spec to_map(t()) :: map()

Converts a flow to an attributes map accepted by Sig.create_workflow/1.

@spec topological_sort(map()) :: {:ok, [String.t()]} | {:error, String.t()}

Sorts node names so every node comes after its dependencies, using Kahn's algorithm.

Returns {:ok, sorted_names}, or {:error, message} if the graph contains a cycle.

Examples

iex> Sig.Flow.topological_sort(%{
...>   "b" => %{module: B, depends_on: ["a"]},
...>   "a" => %{module: A}
...> })
{:ok, ["a", "b"]}

@spec validate(term()) :: :ok | {:error, String.t()}

Validates a set of node definitions.

Checks that:

• every node has a non-blank name and a module;
• every entry in depends_on refers to an existing node;
• the graph has no cycles.

Returns :ok or {:error, message} describing the first problem found.

Examples

iex> Sig.Flow.validate(%{"a" => %{module: A, depends_on: ["b"]}})
{:error, "Node 'a' depends on non-existent node 'b'"}