reactor

defmodule Reactor.Planner do
  @moduledoc """
  Build an execution plan for a Reactor.

  Converts any unplanned steps into vertices in a graph with directed edges
  between them representing their dependencies (arguments).
  """

  alias Reactor.{Error.PlanError, Step}
  import Reactor, only: :macros
  import Reactor.Argument, only: :macros
  import Reactor.Utils

  @doc """
  Build an execution plan for a Reactor.

  Builds a graph of the step dependencies, resolves them and then builds an execution plan.
  """
  @spec plan(Reactor.t()) :: {:ok, Reactor.t()} | {:error, any}
  def plan(reactor) when not is_reactor(reactor),
    do: {:error, argument_error(:reactor, "not a Reactor", reactor)}

  def plan(reactor) when is_nil(reactor.plan),
    do: plan(%{reactor | plan: empty_graph()})

  def plan(reactor) do
    with {:ok, graph} <- reduce_steps_into_graph(reactor.plan, reactor.steps),
         :ok <- assert_graph_not_cyclic(reactor, graph) do
      {:ok, %{reactor | steps: [], plan: graph}}
    end
  end

  @doc """
  Raising version of `plan/1`.
  """
  @spec plan!(Reactor.t()) :: Reactor.t() | no_return
  def plan!(reactor) do
    case plan(reactor) do
      {:ok, reactor} -> reactor
      {:error, reason} -> raise reason
    end
  end

  @doc false
  def get_ref(%{ref: ref}), do: ref

  defp empty_graph, do: Graph.new(type: :directed, vertex_identifier: &__MODULE__.get_ref/1)

  defp reduce_steps_into_graph(graph, steps) do
    steps_by_name =
      graph
      |> Graph.vertices()
      |> Enum.concat(steps)
      |> Map.new(&{&1.name, &1})

    steps
    |> reduce_while_ok(graph, fn
      step, graph when is_struct(step, Step) ->
        graph
        |> Graph.add_vertex(step, step.name)
        |> reduce_arguments_into_graph(step, steps_by_name)

      not_step, _ ->
        {:error,
         PlanError.exception(
           graph: graph,
           step: not_step,
           message: "Value is not a `Reactor.Step` struct."
         )}
    end)
  end

  defp reduce_arguments_into_graph(graph, current_step, steps_by_name) do
    reduce_while_ok(current_step.arguments, graph, fn
      argument, graph when is_argument(argument) and is_from_result(argument) ->
        dependency_name = argument.source.name

        case Map.fetch(steps_by_name, dependency_name) do
          {:ok, dependency} when dependency.name == current_step.name ->
            {:ok, graph}

          {:ok, dependency} ->
            {:ok,
             Graph.add_edge(graph, dependency, current_step,
               label: {:argument, argument.name, :for, current_step.name}
             )}

          :error ->
            {:error,
             PlanError.exception(
               graph: graph,
               step: current_step,
               message:
                 "Step `#{inspect(current_step.name)}` depends on the result of a step named `#{inspect(argument.source.name)}` which cannot be found"
             )}
        end

      argument, graph
      when is_argument(argument) and (is_from_input(argument) or is_from_value(argument)) ->
        {:ok, graph}
    end)
  end

  defp assert_graph_not_cyclic(reactor, graph) do
    if Graph.is_acyclic?(graph) do
      :ok
    else
      {:error,
       PlanError.exception(
         reactor: reactor,
         graph: graph,
         message: "Reactor contains cyclic dependencies."
       )}
    end
  end
end