@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")

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

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

@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.

@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.

@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.

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

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

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

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

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

@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.

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

Lists all workflows, most recently created first.

@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.

@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.

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

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

@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.

@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.

@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.

@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.