virta

defmodule Virta.Core.Workflow do
  @moduledoc """
  Virta.Core.Worflow is a special component which allows us to invoke a different workflow from the
  current workflow.

  A workflow node can be represented as %Node{ module: Virta.Core.Workflow, id: 1, ref: "adder" }

  Notice the :ref property. It refers to the registered workflow with the name adder.

  Lets see a code example for a complex worflow which invokes other workflows:

  ```elixir
  alias Virta.Node
  alias Virta.Registry
  alias Virta.EdgeData
  alias Virta.Instance

  adder = Graph.new(type: :directed)
  |> Graph.add_edge(
    %Node{ module: Virta.Core.In, id: 0 },
    %Node{ module: Virta.Math.Add, id: 1 },
    label: %EdgeData{ from: :addend, to: :addend }
  )
  |> Graph.add_edge(
    %Node{ module: Virta.Core.In, id: 0 },
    %Node{ module: Virta.Math.Add, id: 1 },
    label: %EdgeData{ from: :augend, to: :augend }
  )
  |> Graph.add_edge(
    %Node{ module: Virta.Math.Add, id: 1 },
    %Node{ module: Virta.Core.Out, id: 2 },
    label: %EdgeData{ from: :sum, to: :sum }
  )

  multiplier = Graph.new(type: :directed)
  |> Graph.add_edge(
    %Node{ module: Virta.Core.In, id: 0 },
    %Node{ module: Virta.Math.Multiply, id: 1 },
    label: %EdgeData{ from: :multiplicand, to: :multiplicand }
  )
  |> Graph.add_edge(
    %Node{ module: Virta.Core.In, id: 0 },
    %Node{ module: Virta.Math.Multiply, id: 1 },
    label: %EdgeData{ from: :multiplier, to: :multiplier }
  )
  |> Graph.add_edge(
    %Node{ module: Virta.Math.Multiply, id: 1 },
    %Node{ module: Virta.Core.Out, id: 2 },
    label: %EdgeData{ from: :product, to: :product }
  )

  complex_graph = Graph.new(type: :directed)
  |> Graph.add_edge(
    %Node{ module: Virta.Core.In, id: 0 },
    %Node{ module: Virta.Core.Workflow, id: 1, ref: "adder" },
    label: %EdgeData{ from: :augend, to: :augend }
  )
  |> Graph.add_edge(
    %Node{ module: Virta.Core.In, id: 0 },
    %Node{ module: Virta.Core.Workflow, id: 1, ref: "adder" },
    label: %EdgeData{ from: :addend, to: :addend }
  )
  |> Graph.add_edge(
    %Node{ module: Virta.Core.Workflow, id: 1, ref: "adder" },
    %Node{ module: Virta.Core.Workflow, id: 2, ref: "multiplier" },
    label: %EdgeData{ from: :sum, to: :multiplicand }
  )
  |> Graph.add_edge(
    %Node{ module: Virta.Core.Workflow, id: 1, ref: "adder" },
    %Node{ module: Virta.Core.Workflow, id: 2, ref: "multiplier" },
    label: %EdgeData{ from: :sum, to: :multiplier }
  )
  |> Graph.add_edge(
    %Node{ module: Virta.Core.Workflow, id: 2, ref: "multiplier" },
    %Node{ module: Virta.Core.Out, id: 3 },
    label: %EdgeData{ from: :product, to: :product }
  )

  Registry.register("adder", adder)
  Registry.register("multiplier", multiplier)
  Registry.register("complex_graph", complex_graph)
  ```

  This can then be executed as follows:
  ```elixir
  data = %{
    %Node{ module: Virta.Core.In, id: 0 } => [
      { 1, :augend, 10 }, { 1, :addend, 20 }
    ]
  }

  { requst_id, output } = Virta.Executor.call("complex_graph", data)
  ```
  """
  @inports []
  @outports []

  alias Virta.Node
  alias Virta.Instance

  use Virta.Component

  @doc false
  def workflow do true end

  @impl true
  def loop(inport_args, outport_args, instance_pid, rinports \\ nil, rgraph_name \\ nil) do
    receive do
      { request_id, port, value } ->
        { inports, graph_name } = if(port == :graph) do
          get_ports_from_graph(value)
        else
          { rinports, rgraph_name }
        end
        inport_args = Map.put(inport_args, port, value)
        required_fields = check_required_fields(outport_args)
        if(required_fields |> Enum.all?(&(Map.has_key?(inport_args, &1)))) do
          run(request_id, inport_args, outport_args, instance_pid)
          loop(%{}, outport_args, instance_pid, inports, graph_name)
        else
          loop(inport_args, outport_args, instance_pid, inports, graph_name)
        end
    end
  end

  @impl true
  def run(request_id, inport_args, outport_args, _instance_pid) do
    { ref, message_configs } = outport_args
    |> Enum.filter(fn arg -> !Map.has_key?(arg, :pid) end)
    |> Enum.reduce({ nil, [] }, fn(arg, { _ref, message_configs }) ->
      to_port = Map.get(arg, :to)
      from_port = Map.get(arg, :from)
      ref = arg |> Map.get(:ref) |> Map.get(:ref)
      message_configs = message_configs ++ [{ from_port, to_port }]
      { ref, message_configs }
    end)

    data = message_configs
    |> Enum.reduce(%{}, fn({ _from, to}, acc) ->
      value = Map.get(inport_args, to)
      messages = (Map.get(acc, %Node{ module: Virta.Core.In, id: 0 }) || []) ++ [{ request_id, to, value }]
      Map.put(acc, %Node{ module: Virta.Core.In, id: 0 }, messages)
    end)

    { _requst_id, response } = Virta.Executor.call(ref, data)
    outport_args
    |> Enum.filter(fn arg -> Map.has_key?(arg, :pid) end)
    |> Enum.map(fn(arg) ->
      %{ from: from, to: to, pid: pid } = arg
      send(pid, { request_id, to , Map.get(response, from) })
    end)

    { request_id, :noreply }
  end

  defp get_ports_from_graph(value) do
    :poolboy.transaction(String.to_existing_atom(value), fn (server) ->
      inports = Instance.inports(server)
      { inports, value }
    end)
  end

  defp check_required_fields(outport_args) do
    outport_args
    |> Enum.filter(fn arg -> !Map.has_key?(arg, :pid) end)
    |> Enum.map(&Map.get(&1, :to))
  end
end