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

Public API of the Sig workflow engine.

Sig runs workflows defined as a Directed Acyclic Graph (DAG) of nodes. Each
node is a module implementing `Sig.Node`. Workflows and their executions are
persisted in PostgreSQL through your application's Ecto repo, and executions
can run either inline (synchronously) or distributed across workers through
BullMQ (Redis).

## Setup

    # config/config.exs
    config :sig, repo: MyApp.Repo

and create Sig's tables with a migration that calls `Sig.Migration.up/1`.
See the README for the full installation guide.

## Concepts

  * **Workflow** (`Sig.Schemas.Workflow`) - a persisted DAG definition, with
    workflow-level `params` and `context`. Build one with `Sig.Flow` and save
    it with `create_workflow/1`.
  * **Execution** (`Sig.Schemas.WorkflowExecution`) - a single run of a
    workflow, with its own `params`, `context` and accumulated
    `node_results`.
  * **Node execution** (`Sig.Schemas.NodeExecution`) - the record of one node
    within an execution, tracking attempts, result and errors.

## Params and context

When a node runs it receives a `Sig.NodeContext`. Its `params` are deep-merged
in this order, with later levels winning:

    workflow params -> execution params -> node params

and its `context` is the deep merge of the workflow and execution contexts.
The outputs of completed nodes are available in `context.results`, keyed by
node name.

> #### Keys are strings at runtime {: .warning}
>
> Params, context and results are stored in JSONB columns, so by the time a
> node runs all map keys are strings, even if atoms were used when defining
> the workflow. Read them with `params["key"]`, not `params[:key]`.

## Example

    flow =
      Sig.Flow.new("payment_pipeline", params: %{"currency" => "BRL"})
      |> Sig.Flow.add_node(:fetch_user, MyApp.Nodes.FetchUser)
      |> Sig.Flow.add_node(:debit, MyApp.Nodes.DebitAccount,
        depends_on: [:fetch_user],
        params: %{"amount" => 50}
      )

    {:ok, workflow} = Sig.create_workflow(flow)
    {:ok, execution} = Sig.run_execution(workflow, %{"user_id" => 10})

    execution.status
    #=> "completed"

## Return values

Functions that look up a record return `{:error, :not_found}` (or
`{:error, :workflow_not_found}` in `create_execution/4`) when it does not
exist. Lifecycle functions also return `{:error, reason}` when the
transition is not allowed, for example
`{:error, {:cannot_resume_in_state, "completed"}}`.

# `execution_ref`

```elixir
@type execution_ref() :: Sig.Schemas.WorkflowExecution.t() | String.t()
```

An execution struct or its ID.

# `workflow_ref`

```elixir
@type workflow_ref() :: Sig.Schemas.Workflow.t() | String.t()
```

A workflow struct or its ID.

# `create_execution`

```elixir
@spec create_execution(workflow_ref(), map(), map(), keyword()) ::
  {:ok, Sig.Schemas.WorkflowExecution.t()} | {:error, term()}
```

Creates a `pending` execution for a workflow without running it.

Start it afterwards with `start_execution/2`, or use `run_execution/4` to
create and start in a single call.

`execution_params` are merged over the workflow params (and under each
node's params); `execution_context` is merged over the workflow context.

## Options

  * `:idempotency_key` - when an execution with the same key already exists
    for this workflow, that execution is returned instead of creating a
    duplicate.

## Errors

  * `{:error, :workflow_not_found}`
  * `{:error, :workflow_suspended}`
  * `{:error, :workflow_stopped}`

## Examples

    {:ok, execution} = Sig.create_execution(workflow.id, %{"batch_id" => 123}, %{"user_id" => 42})
    {:ok, execution} = Sig.create_execution(workflow, %{}, %{}, idempotency_key: "tx-999")

# `create_workflow`

```elixir
@spec create_workflow(Sig.Flow.t() | map()) ::
  {:ok, Sig.Schemas.Workflow.t()} | {:error, Ecto.Changeset.t()}
```

Creates and persists a workflow definition.

Accepts either a `Sig.Flow` struct or a map of attributes. The node graph is
validated (missing modules, missing dependencies and cycles) before the
workflow is inserted; validation errors are returned in the changeset under
`:nodes`.

## Attributes

  * `:name` (required) - the workflow name.
  * `:description` - a free-form description.
  * `:params` - workflow-level params, merged into every node's params.
  * `:context` - workflow-level context, shared by every node.
  * `:nodes` - a map of node name to node definition (or a list of node
    definitions with a `:name` key). See `Sig.Flow.add_node/4` for the
    supported fields.

## Examples

    # Using the Sig.Flow DSL:
    flow =
      Sig.Flow.new("data_pipeline", params: %{"env" => "prod"}, context: %{"source" => "s3"})
      |> Sig.Flow.add_node(:fetch, FetchDataNode, params: %{"bucket" => "my-bucket"})
      |> Sig.Flow.add_node(:process, ProcessDataNode, depends_on: [:fetch])
      |> Sig.Flow.add_node(:notify, NotifyNode, depends_on: [:process])

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

    # Using a map:
    {:ok, workflow} =
      Sig.create_workflow(%{
        name: "data_pipeline",
        params: %{"env" => "prod"},
        context: %{"source" => "s3"},
        nodes: %{
          "fetch" => %{module: FetchDataNode, params: %{"bucket" => "my-bucket"}},
          "process" => %{module: ProcessDataNode, depends_on: ["fetch"]}
        }
      })

    # Invalid graphs are rejected:
    {:error, %Ecto.Changeset{errors: [nodes: {"Workflow contains circular dependencies" <> _, _}]}} =
      Sig.create_workflow(cyclic_flow)

# `create_workflow`

```elixir
@spec create_workflow(String.t() | atom(), map() | list(), keyword()) ::
  {:ok, Sig.Schemas.Workflow.t()} | {:error, Ecto.Changeset.t()}
```

Creates a workflow from a name and a map of node definitions.

A shortcut for `create_workflow/1` that does not require building a
`Sig.Flow` first.

## Options

  * `:description` - a free-form description.
  * `:params` - workflow-level params.
  * `:context` - workflow-level context.

## Examples

    {:ok, workflow} =
      Sig.create_workflow("etl", %{
        "extract" => %{module: Extract},
        "load" => %{module: Load, depends_on: ["extract"]}
      })

# `delete_workflow`

```elixir
@spec delete_workflow(workflow_ref()) ::
  {:ok, Sig.Schemas.Workflow.t()} | {:error, :not_found | Ecto.Changeset.t()}
```

Deletes a workflow and all of its executions.

Returns `{:error, :not_found}` if the workflow does not exist.

# `get_execution`

```elixir
@spec get_execution(String.t()) :: Sig.Schemas.WorkflowExecution.t() | nil
```

Fetches an execution by ID, preloading its `:workflow` and
`:node_executions`. Returns `nil` if it does not exist.

# `get_execution!`

```elixir
@spec get_execution!(String.t()) :: Sig.Schemas.WorkflowExecution.t()
```

Fetches an execution by ID, preloading its `:workflow` and
`:node_executions`. Raises `Ecto.NoResultsError` if it does not exist.

# `get_node_results`

```elixir
@spec get_node_results(execution_ref()) :: {:ok, map()} | {:error, :not_found}
```

Returns the results of all completed nodes of an execution, keyed by node
name.

Results that were not maps when returned by the node are wrapped as
`%{"value" => result}` (see `Sig.Serializer.wrap_result/1`).

## Examples

    {:ok, %{"fetch_user" => %{"name" => "Ana"}}} = Sig.get_node_results(execution)

# `get_workflow`

```elixir
@spec get_workflow(String.t()) :: Sig.Schemas.Workflow.t() | nil
```

Fetches a workflow by ID. Returns `nil` if it does not exist.

# `get_workflow!`

```elixir
@spec get_workflow!(String.t()) :: Sig.Schemas.Workflow.t()
```

Fetches a workflow by ID. Raises `Ecto.NoResultsError` if it does not exist.

# `list_executions`

```elixir
@spec list_executions(String.t() | nil) :: [Sig.Schemas.WorkflowExecution.t()]
```

Lists executions, most recently created first, with `:workflow` preloaded.

When `workflow_id` is given, only executions of that workflow are returned.

# `list_workflows`

```elixir
@spec list_workflows() :: [Sig.Schemas.Workflow.t()]
```

Lists all workflows, most recently created first.

# `resume_execution`

```elixir
@spec resume_execution(execution_ref(), keyword()) ::
  {:ok, Sig.Schemas.WorkflowExecution.t()} | {:error, term()}
```

Resumes a suspended execution from where it stopped.

Nodes that already completed are not run again; only the remaining nodes
are dispatched.

## Options

  * `:async` - same as in `start_execution/2`. Defaults to `false`.

## Errors

  * `{:error, :not_found}`
  * `{:error, :workflow_suspended}` / `{:error, :workflow_stopped}`
  * `{:error, {:cannot_resume_in_state, status}}` - the execution is not
    `suspended`.

# `resume_workflow`

```elixir
@spec resume_workflow(workflow_ref()) ::
  {:ok, Sig.Schemas.Workflow.t()} | {:error, term()}
```

Resumes a suspended workflow, allowing new executions to be created again.

Executions suspended by `suspend_workflow/1` stay suspended; resume them
individually with `resume_execution/2`.

Returns `{:error, :workflow_already_stopped}` for stopped workflows.

# `run_execution`

```elixir
@spec run_execution(workflow_ref(), map(), map(), keyword()) ::
  {:ok, Sig.Schemas.WorkflowExecution.t()} | {:error, term()}
```

Creates an execution and starts it right away.

Equivalent to `create_execution/4` followed by `start_execution/2`. Accepts
the options of both functions (`:idempotency_key` and `:async`).

## Examples

    {:ok, execution} = Sig.run_execution(workflow, %{"input" => "hello"}, %{"auth" => token})

    {:ok, execution} =
      Sig.run_execution(workflow, %{"order_id" => 1}, %{},
        idempotency_key: "order-1",
        async: true
      )

# `start_execution`

```elixir
@spec start_execution(execution_ref(), keyword()) ::
  {:ok, Sig.Schemas.WorkflowExecution.t()} | {:error, term()}
```

Starts an execution.

## Options

  * `:async` - when `false` (the default), runs the execution in the calling
    process with `Sig.Engine.execute_sync/1` and returns once it reaches a
    final state (or is suspended/stopped). When `true`, marks the execution
    as `running`, enqueues its runnable nodes to BullMQ and returns
    immediately; `Sig.BullMQ.Supervisor` must be running for the jobs to be
    processed.

## Examples

    {:ok, %{status: "completed"}} = Sig.start_execution(execution)

    {:ok, %{status: "running"}} = Sig.start_execution(execution, async: true)

# `stop_execution`

```elixir
@spec stop_execution(execution_ref()) ::
  {:ok, Sig.Schemas.WorkflowExecution.t()} | {:error, term()}
```

Stops an execution permanently. A stopped execution cannot be resumed.

Returns `{:error, {:cannot_stop_in_state, status}}` for executions that are
already `completed` or `failed`.

# `stop_workflow`

```elixir
@spec stop_workflow(workflow_ref()) ::
  {:ok, Sig.Schemas.Workflow.t()} | {:error, term()}
```

Stops a workflow permanently.

All `pending`, `running` and `suspended` executions are moved to `stopped`,
and no new executions can be created. A stopped workflow cannot be resumed.

# `suspend_execution`

```elixir
@spec suspend_execution(execution_ref()) ::
  {:ok, Sig.Schemas.WorkflowExecution.t()} | {:error, term()}
```

Suspends an execution.

Nodes that are already running finish, but no further nodes are
dispatched. Continue it later with `resume_execution/2`.

Returns `{:error, {:cannot_suspend_in_state, status}}` for executions that
are `completed`, `failed` or `stopped`.

# `suspend_workflow`

```elixir
@spec suspend_workflow(workflow_ref()) ::
  {:ok, Sig.Schemas.Workflow.t()} | {:error, term()}
```

Suspends a workflow.

While suspended, new executions cannot be created, and all `pending` and
`running` executions are moved to `suspended`. Nodes that are already
running finish, but no further nodes are dispatched.

Use `resume_workflow/1` to reactivate the workflow, then
`resume_execution/2` for each execution you want to continue.

Returns `{:error, :workflow_already_stopped}` for stopped workflows.

---

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