# `Choreo.Dataflow`
[🔗](https://github.com/code-shoily/choreo/blob/v0.14.1/lib/choreo/dataflow.ex#L1)

Dataflow / pipeline diagram builder on top of Yog.

`Choreo.Dataflow` models stream-processing and ETL pipelines where
nodes are processing stages and edges are data streams. It is a natural
complement to `Choreo` architecture diagrams: those show *what*
infrastructure exists; dataflow shows *how data moves through it*.

## When to use

Use `Choreo.Dataflow` when designing, reviewing, or debugging pipelines
— event-driven systems, ETL jobs, stream processors, or microservice
data flows. It helps identify bottlenecks, orphan stages, and critical
paths before they hit production.

## Node types

  * `:source` — entry point that produces data
  * `:sink` — terminal consumer that persists or emits data
  * `:transform` — stateless operation (map, filter, aggregate)
  * `:buffer` — queue, topic, or back-pressure buffer
  * `:conditional` — branch / split based on a predicate
  * `:merge` — join multiple streams into one

## Further reading

  * [Dataflow Programming (Wikipedia)](https://en.wikipedia.org/wiki/Dataflow_programming)
  * [Streaming Systems (O'Reilly)](https://www.oreilly.com/library/view/streaming-systems/9781491983867/)
  * [Enterprise Integration Patterns](https://www.enterpriseintegrationpatterns.com/)

## Quick Start

    pipeline =
      Choreo.Dataflow.new()
      |> Choreo.Dataflow.add_source(:sensor, label: "IoT Sensor")
      |> Choreo.Dataflow.add_transform(:parse, label: "JSON Parser")
      |> Choreo.Dataflow.add_buffer(:kafka, label: "Kafka Topic")
      |> Choreo.Dataflow.add_transform(:aggregate, label: "Window Agg")
      |> Choreo.Dataflow.add_sink(:db, label: "TimescaleDB")
      |> Choreo.Dataflow.connect(:sensor, :parse, data_type: "raw bytes")
      |> Choreo.Dataflow.connect(:parse, :kafka, data_type: "event")
      |> Choreo.Dataflow.connect(:kafka, :aggregate, data_type: "event")
      |> Choreo.Dataflow.connect(:aggregate, :db, data_type: "metrics")

    dot = Choreo.Dataflow.to_dot(pipeline)
    mermaid = Choreo.Dataflow.to_mermaid(pipeline)

## Diagram

<div class="graphviz">
  digraph G {
    graph [rankdir=LR, splines=spline, nodesep=0.6, ranksep=1.2];
    node [shape=box, style=filled, fillcolor="white", fontname="Helvetica", fontsize=12, fontcolor="white"];
    edge [arrowhead=normal, color="#64748b", style=solid, fontname="Helvetica", fontsize=10, penwidth=1.0];

    parse [label="JSON Parser", fillcolor="#3b82f6", shape="box3d"];
    aggregate [label="Window Agg", fillcolor="#3b82f6", shape="box3d"];
    sensor [label="IoT Sensor", fillcolor="#10b981", shape="house"];
    kafka [label="Kafka Topic", fillcolor="#f59e0b", shape="cylinder"];
    db [label="TimescaleDB", fillcolor="#f43f5e", shape="invhouse"];

    parse -> kafka [style="solid", penwidth="1.0", color="#64748b", label="event"];
    aggregate -> db [style="solid", penwidth="1.0", color="#64748b", label="metrics"];
    sensor -> parse [style="solid", penwidth="1.0", color="#64748b", label="raw bytes"];
    kafka -> aggregate [style="solid", penwidth="1.0", color="#64748b", label="event"];
  }
</div>

## Analysis

    # Detect feedback loops (usually a bug in dataflow)
    Choreo.Dataflow.Analysis.cyclic?(pipeline)

    # Execution order for a DAG
    {:ok, order} = Choreo.Dataflow.Analysis.topological_sort(pipeline)

    # Find orphan stages with no upstream source
    Choreo.Dataflow.Analysis.orphan_nodes(pipeline)

    # Find dead-end stages that never reach a sink
    Choreo.Dataflow.Analysis.dead_ends(pipeline)

    # Longest latency path (critical path)
    {:ok, path, latency} = Choreo.Dataflow.Analysis.longest_path(pipeline)

    # Throughput simulation
    Choreo.Dataflow.Analysis.simulate(pipeline)

# `t`

```elixir
@type t() :: %Choreo.Dataflow{
  clusters: %{required(String.t()) =&gt; map()},
  edge_meta: %{optional({Yog.node_id(), Yog.node_id()}) =&gt; map()},
  graph: Yog.graph(),
  strict: boolean()
}
```

# `add_buffer`

Adds a buffer node (queue, topic, back-pressure reservoir).

## Options

* `:capacity`

* `:latency_ms`

* `:label` (`t:String.t/0`)

* `:description` (`t:String.t/0`)

* `:cluster` (`t:String.t/0`)

* `:shape` (`t:atom/0`)

* `:fillcolor` (`t:String.t/0`)

* `:fontcolor` (`t:String.t/0`)

* `:style` (`t:String.t/0`)

* `:penwidth`

* `:image` (`t:String.t/0`)

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = Choreo.Dataflow.add_buffer(flow, :kafka, label: "Kafka Topic", capacity: 1000)
    iex> Choreo.Dataflow.nodes(flow)
    [:kafka]
    iex> Yog.node(flow.graph, :kafka).node_type
    :buffer
    iex> Yog.node(flow.graph, :kafka).capacity
    1000

## Diagram

<div class="graphviz">
  digraph G {
    graph [rankdir=LR, splines=spline, nodesep=0.6, ranksep=1.2];
    node [shape=box, style=filled, fillcolor="white", fontname="Helvetica", fontsize=12, fontcolor="white"];
    edge [arrowhead=normal, color="#64748b", style=solid, fontname="Helvetica", fontsize=10, penwidth=1.0];

    kafka [label="Kafka Topic
(cap: 1000)", fillcolor="#f59e0b", shape="cylinder"];
  }
</div>

# `add_cluster`

```elixir
@spec add_cluster(t(), String.t(), keyword()) :: t()
```

Defines a cluster (subgraph) for grouping nodes visually.

## Options

* `:parent` (`t:String.t/0`)

* `:label` (`t:String.t/0`)

* `:style` (`t:String.t/0`)

* `:fillcolor` (`t:String.t/0`)

* `:color` (`t:String.t/0`)

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = Choreo.Dataflow.add_cluster(flow, "ingest", label: "Ingestion")
    iex> Choreo.Dataflow.cluster(flow, "ingest").label
    "Ingestion"

# `add_conditional`

Adds a conditional / split node (branching).

## Options

* `:label` (`t:String.t/0`)

* `:description` (`t:String.t/0`)

* `:cluster` (`t:String.t/0`)

* `:shape` (`t:atom/0`)

* `:fillcolor` (`t:String.t/0`)

* `:fontcolor` (`t:String.t/0`)

* `:style` (`t:String.t/0`)

* `:penwidth`

* `:image` (`t:String.t/0`)

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = Choreo.Dataflow.add_conditional(flow, :valid, label: "If valid")
    iex> Choreo.Dataflow.nodes(flow)
    [:valid]
    iex> Yog.node(flow.graph, :valid).node_type
    :conditional

## Diagram

<div class="graphviz">
  digraph G {
    graph [rankdir=LR, splines=spline, nodesep=0.6, ranksep=1.2];
    node [shape=box, style=filled, fillcolor="white", fontname="Helvetica", fontsize=12, fontcolor="white"];
    edge [arrowhead=normal, color="#64748b", style=solid, fontname="Helvetica", fontsize=10, penwidth=1.0];

    valid [label="If valid", fillcolor="#8b5cf6", shape="diamond"];
  }
</div>

# `add_dead_letter_path`

```elixir
@spec add_dead_letter_path(t(), Yog.node_id(), Yog.node_id(), keyword()) :: t()
```

Adds a dead-letter-path edge between two stages.

Dead-letter paths are rendered in grey with a dashed style.

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = flow
    ...>   |> Choreo.Dataflow.add_transform(:a)
    ...>   |> Choreo.Dataflow.add_sink(:b)
    ...>   |> Choreo.Dataflow.add_dead_letter_path(:a, :b)
    iex> flow.edge_meta[{:a, :b}].path_type
    :dead_letter

# `add_error_path`

```elixir
@spec add_error_path(t(), Yog.node_id(), Yog.node_id(), keyword()) :: t()
```

Adds an error-path edge between two stages.

Error paths are rendered in red with a dashed style.

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = flow
    ...>   |> Choreo.Dataflow.add_transform(:a)
    ...>   |> Choreo.Dataflow.add_sink(:b)
    ...>   |> Choreo.Dataflow.add_error_path(:a, :b)
    iex> flow.edge_meta[{:a, :b}].path_type
    :error

# `add_merge`

Adds a merge / join node (combining multiple streams).

## Options

* `:label` (`t:String.t/0`)

* `:description` (`t:String.t/0`)

* `:cluster` (`t:String.t/0`)

* `:shape` (`t:atom/0`)

* `:fillcolor` (`t:String.t/0`)

* `:fontcolor` (`t:String.t/0`)

* `:style` (`t:String.t/0`)

* `:penwidth`

* `:image` (`t:String.t/0`)

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = Choreo.Dataflow.add_merge(flow, :join, label: "Join")
    iex> Choreo.Dataflow.nodes(flow)
    [:join]
    iex> Yog.node(flow.graph, :join).node_type
    :merge

## Diagram

<div class="graphviz">
  digraph G {
    graph [rankdir=LR, splines=spline, nodesep=0.6, ranksep=1.2];
    node [shape=box, style=filled, fillcolor="white", fontname="Helvetica", fontsize=12, fontcolor="white"];
    edge [arrowhead=normal, color="#64748b", style=solid, fontname="Helvetica", fontsize=10, penwidth=1.0];

    join [label="Join", fillcolor="#06b6d4", shape="trapezium"];
  }
</div>

# `add_retry_path`

```elixir
@spec add_retry_path(t(), Yog.node_id(), Yog.node_id(), keyword()) :: t()
```

Adds a retry-path edge between two stages.

Retry paths are rendered in orange with a dotted style.

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = flow
    ...>   |> Choreo.Dataflow.add_transform(:a)
    ...>   |> Choreo.Dataflow.add_sink(:b)
    ...>   |> Choreo.Dataflow.add_retry_path(:a, :b)
    iex> flow.edge_meta[{:a, :b}].path_type
    :retry

# `add_sink`

Adds a sink node (data consumer / terminal).

## Options

* `:label` (`t:String.t/0`)

* `:description` (`t:String.t/0`)

* `:cluster` (`t:String.t/0`)

* `:shape` (`t:atom/0`)

* `:fillcolor` (`t:String.t/0`)

* `:fontcolor` (`t:String.t/0`)

* `:style` (`t:String.t/0`)

* `:penwidth`

* `:image` (`t:String.t/0`)

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = Choreo.Dataflow.add_sink(flow, :db, label: "TimescaleDB")
    iex> Choreo.Dataflow.nodes(flow)
    [:db]
    iex> Yog.node(flow.graph, :db).node_type
    :sink

## Diagram

<div class="graphviz">
  digraph G {
    graph [rankdir=LR, splines=spline, nodesep=0.6, ranksep=1.2];
    node [shape=box, style=filled, fillcolor="white", fontname="Helvetica", fontsize=12, fontcolor="white"];
    edge [arrowhead=normal, color="#64748b", style=solid, fontname="Helvetica", fontsize=10, penwidth=1.0];

    db [label="TimescaleDB", fillcolor="#f43f5e", shape="invhouse"];
  }
</div>

# `add_source`

Adds a source node (data producer / entry point).

## Options

* `:rate`

* `:label` (`t:String.t/0`)

* `:description` (`t:String.t/0`)

* `:cluster` (`t:String.t/0`)

* `:shape` (`t:atom/0`)

* `:fillcolor` (`t:String.t/0`)

* `:fontcolor` (`t:String.t/0`)

* `:style` (`t:String.t/0`)

* `:penwidth`

* `:image` (`t:String.t/0`)

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = Choreo.Dataflow.add_source(flow, :sensor, label: "IoT Sensor")
    iex> Choreo.Dataflow.nodes(flow)
    [:sensor]
    iex> Yog.node(flow.graph, :sensor).node_type
    :source
    iex> Yog.node(flow.graph, :sensor).label
    "IoT Sensor"

## Diagram

<div class="graphviz">
  digraph G {
    graph [rankdir=LR, splines=spline, nodesep=0.6, ranksep=1.2];
    node [shape=box, style=filled, fillcolor="white", fontname="Helvetica", fontsize=12, fontcolor="white"];
    edge [arrowhead=normal, color="#64748b", style=solid, fontname="Helvetica", fontsize=10, penwidth=1.0];

    sensor [label="IoT Sensor", fillcolor="#10b981", shape="house"];
  }
</div>

# `add_transform`

Adds a transform node (stateless processing).

## Options

* `:latency_ms`

* `:capacity`

* `:label` (`t:String.t/0`)

* `:description` (`t:String.t/0`)

* `:cluster` (`t:String.t/0`)

* `:shape` (`t:atom/0`)

* `:fillcolor` (`t:String.t/0`)

* `:fontcolor` (`t:String.t/0`)

* `:style` (`t:String.t/0`)

* `:penwidth`

* `:image` (`t:String.t/0`)

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = Choreo.Dataflow.add_transform(flow, :parse, label: "JSON Parser")
    iex> Choreo.Dataflow.nodes(flow)
    [:parse]
    iex> Yog.node(flow.graph, :parse).node_type
    :transform

## Diagram

<div class="graphviz">
  digraph G {
    graph [rankdir=LR, splines=spline, nodesep=0.6, ranksep=1.2];
    node [shape=box, style=filled, fillcolor="white", fontname="Helvetica", fontsize=12, fontcolor="white"];
    edge [arrowhead=normal, color="#64748b", style=solid, fontname="Helvetica", fontsize=10, penwidth=1.0];

    parse [label="JSON Parser", fillcolor="#3b82f6", shape="box3d"];
  }
</div>

# `cluster`

```elixir
@spec cluster(t(), String.t()) :: map() | nil
```

Returns cluster metadata by user-facing name.

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = Choreo.Dataflow.add_cluster(flow, "ingest", label: "Ingestion")
    iex> Choreo.Dataflow.cluster(flow, "ingest").label
    "Ingestion"

# `connect`

Connects two stages with a data-flow edge.

> ### Limitation
> At most one edge is allowed per `(from, to)` pair.
> Adding a second connection between the same stages raises
> `ArgumentError`. Multigraph support (parallel edges) is planned
> for a future release.

## Options

* `:data_type` (`t:String.t/0`)

* `:label` (`t:String.t/0`)

* `:rate`

* `:path_type` - The default value is `:normal`.

* `:weight`

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = flow
    ...>   |> Choreo.Dataflow.add_source(:a)
    ...>   |> Choreo.Dataflow.add_transform(:b)
    ...>   |> Choreo.Dataflow.connect(:a, :b, data_type: "event")
    iex> Choreo.Dataflow.edges(flow)
    [{:a, :b, 1}]
    iex> flow.edge_meta[{:a, :b}].label
    "event"
    iex> flow.edge_meta[{:a, :b}].path_type
    :normal

## Diagram

<div class="graphviz">
  digraph G {
    graph [rankdir=LR, splines=spline, nodesep=0.6, ranksep=1.2];
    node [shape=box, style=filled, fillcolor="white", fontname="Helvetica", fontsize=12, fontcolor="white"];
    edge [arrowhead=normal, color="#64748b", style=solid, fontname="Helvetica", fontsize=10, penwidth=1.0];

    b [label="b", fillcolor="#3b82f6", shape="box3d"];
    a [label="a", fillcolor="#10b981", shape="house"];

    a -> b [style="solid", penwidth="1.0", color="#64748b", label="event"];
  }
</div>

# `edges`

```elixir
@spec edges(t()) :: [{Yog.node_id(), Yog.node_id(), number()}]
```

Returns all edges as `{from, to, weight}` tuples.

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = flow
    ...>   |> Choreo.Dataflow.add_source(:a)
    ...>   |> Choreo.Dataflow.add_transform(:b)
    ...>   |> Choreo.Dataflow.connect(:a, :b)
    iex> Choreo.Dataflow.edges(flow)
    [{:a, :b, 1}]

# `new`

```elixir
@spec new(keyword()) :: t()
```

Creates a new empty dataflow graph.

Dataflow graphs are always directed.

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> Choreo.Dataflow.nodes(flow)
    []
    iex> Choreo.Dataflow.edges(flow)
    []

# `nodes`

```elixir
@spec nodes(t()) :: [Yog.node_id()]
```

Returns all node IDs in the dataflow.

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = flow
    ...>   |> Choreo.Dataflow.add_source(:a)
    ...>   |> Choreo.Dataflow.add_transform(:b)
    iex> Enum.sort(Choreo.Dataflow.nodes(flow))
    [:a, :b]

# `nodes_of_type`

```elixir
@spec nodes_of_type(t(), atom()) :: [Yog.node_id()]
```

Returns all nodes of a given type.

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = flow
    ...>   |> Choreo.Dataflow.add_source(:a)
    ...>   |> Choreo.Dataflow.add_source(:b)
    ...>   |> Choreo.Dataflow.add_sink(:c)
    iex> Enum.sort(Choreo.Dataflow.nodes_of_type(flow, :source))
    [:a, :b]
    iex> Choreo.Dataflow.nodes_of_type(flow, :sink)
    [:c]
    iex> Choreo.Dataflow.nodes_of_type(flow, :transform)
    []

# `theme`

```elixir
@spec theme(atom(), keyword()) :: Choreo.Theme.t()
```

Returns a theme for `Choreo.Dataflow`.

## Examples

    iex> theme = Choreo.Dataflow.theme(:default, graph_rankdir: :td)
    iex> theme.graph_rankdir
    :td

# `to_dot`

```elixir
@spec to_dot(t(), keyword()) :: String.t()
```

Renders the dataflow to DOT format.

## Options

  * `:theme` — `:default`, `:dark`, `:minimal`, `:warm`, `:forest`, `:ocean`, or a `Choreo.Theme` struct
  * `:highlighted_nodes` — list of node IDs to highlight
  * `:highlighted_edges` — list of `{from, to}` tuples to highlight
  * Any other option accepted by `Yog.Render.DOT.to_dot/2`, such as `:rankdir`

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = flow
    ...>   |> Choreo.Dataflow.add_source(:in, label: "Input")
    ...>   |> Choreo.Dataflow.add_transform(:proc, label: "Process")
    ...>   |> Choreo.Dataflow.add_sink(:out, label: "Output")
    ...>   |> Choreo.Dataflow.connect(:in, :proc, data_type: "raw")
    ...>   |> Choreo.Dataflow.connect(:proc, :out, data_type: "result")
    iex> dot = Choreo.Dataflow.to_dot(flow)
    iex> String.contains?(dot, "digraph")
    true
    iex> String.contains?(dot, "Input")
    true
    iex> String.contains?(dot, "Output")
    true

# `to_graph`

```elixir
@spec to_graph(t()) :: Yog.graph()
```

Returns the raw `Yog.Graph` struct underpinning the dataflow.

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> graph = Choreo.Dataflow.to_graph(flow)
    iex> graph.kind
    :directed

# `to_mermaid`

```elixir
@spec to_mermaid(t(), keyword()) :: String.t()
```

Renders the dataflow pipeline to Mermaid.js flowchart syntax.

## Options

  * `:theme` — `:default`, `:dark`, `:minimal`, `:warm`, `:forest`, `:ocean`, or a `Choreo.Theme` struct
  * `:direction` — `:td` (default), `:lr`, `:bt`, `:rl`
  * `:highlighted_nodes` — list of node IDs to highlight
  * `:highlighted_edges` — list of `{from, to}` tuples to highlight

## Examples

    iex> flow = Choreo.Dataflow.new()
    iex> flow = flow
    ...>   |> Choreo.Dataflow.add_source(:in, label: "Input")
    ...>   |> Choreo.Dataflow.add_transform(:proc, label: "Process")
    ...>   |> Choreo.Dataflow.add_sink(:out, label: "Output")
    ...>   |> Choreo.Dataflow.connect(:in, :proc, data_type: "raw")
    ...>   |> Choreo.Dataflow.connect(:proc, :out, data_type: "result")
    iex> mermaid = Choreo.Dataflow.to_mermaid(flow)
    iex> String.contains?(mermaid, "graph TD")
    true
    iex> String.contains?(mermaid, "Input")
    true
    iex> String.contains?(mermaid, "Output")
    true

---

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