# `Sig.Flow`
[🔗](https://github.com/peeeuzin/sig/blob/v0.1.0/lib/sig/flow.ex#L1)

A small builder for workflow definitions, plus the DAG helpers used by the
engine.

A flow is an in-memory description of a workflow: a name, workflow-level
`params` and `context`, and a set of nodes connected by `depends_on`. Build
it with `new/2` and `add_node/4`, then persist it with
`Sig.create_workflow/1`.

    flow =
      Sig.Flow.new("signup", params: %{"plan" => "free"})
      |> Sig.Flow.add_node(:create_user, MyApp.Nodes.CreateUser)
      |> Sig.Flow.add_node(:send_welcome, MyApp.Nodes.SendWelcome, depends_on: [:create_user])
      |> Sig.Flow.add_node(:track, MyApp.Nodes.Track, depends_on: [:create_user])

    {:ok, workflow} = Sig.create_workflow(flow)

Nodes without a dependency between them (`send_welcome` and `track` above)
can run in parallel when the execution is started with `async: true` and
the BullMQ worker has a concurrency greater than one.

## Node definitions

Internally every node is normalized to a map with string keys:

    %{
      "name" => "send_welcome",
      "module" => "Elixir.MyApp.Nodes.SendWelcome",
      "params" => %{},
      "depends_on" => ["create_user"],
      "max_retries" => 3,
      "backoff_ms" => 500
    }

This is the shape stored in `Sig.Schemas.Workflow` and the one expected by
`topological_sort/1` and `find_runnable_nodes/2`.

# `node_def`

```elixir
@type node_def() :: %{optional(String.t()) => term()}
```

A normalized node definition, keyed by strings.

# `t`

```elixir
@type t() :: %Sig.Flow{
  context: map(),
  description: String.t() | nil,
  name: String.t() | nil,
  nodes: map(),
  params: map()
}
```

An in-memory workflow definition.

# `add_node`

```elixir
@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
    )

# `find_runnable_nodes`

```elixir
@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"]

# `new`

```elixir
@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"})

# `normalize_nodes`

```elixir
@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
      }
    }

# `to_map`

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

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

# `topological_sort`

```elixir
@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"]}

# `validate`

```elixir
@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'"}

---

*Consult [api-reference.md](api-reference.md) for complete listing*