reactor.ex

defmodule Reactor do
  alias Reactor.{Dsl, Error.Validation.StateError, Executor, Step}

  @moduledoc """
  Reactor is a dynamic, concurrent, dependency resolving saga orchestrator.

  ## Usage

  You can construct a reactor using the `Reactor` Spark DSL:

  ```elixir
  defmodule HelloWorldReactor do
    @moduledoc false
    use Reactor

    input :whom

    step :greet, Greeter do
      argument :whom, input(:whom)
    end

    return :greet
  end
  ```

      iex> Reactor.run(HelloWorldReactor, %{whom: "Dear Reader"})
      {:ok, "Hello, Dear Reader!"}

  or you can build it programmatically:

      iex> reactor = Builder.new()
      ...> {:ok, reactor} = Builder.add_input(reactor, :whom)
      ...> {:ok, reactor} = Builder.add_step(reactor, :greet, Greeter, whom: {:input, :whom})
      ...> {:ok, reactor} = Builder.return(reactor, :greet)
      ...> Reactor.run(reactor, %{whom: nil})
      {:ok, "Hello, World!"}
  """

  defstruct context: %{},
            id: nil,
            input_descriptions: %{},
            inputs: [],
            intermediate_results: %{},
            middleware: [],
            plan: nil,
            return: nil,
            state: :pending,
            steps: [],
            undo: []

  use Spark.Dsl, default_extensions: [extensions: [Dsl]]

  @type context :: %{optional(atom) => any}
  @type context_arg :: Enumerable.t({atom, any})

  @typedoc """
  Specify the maximum number of asynchronous steps which can be run in parallel.

  Defaults to the result of `System.schedulers_online/0`.  Only used if
  `async?` is set to `true`.
  """
  @type max_concurrency_option :: {:max_concurrency, pos_integer()}

  @typedoc """
  Specify the amount of execution time after which to halt processing.

  Note that this is not a hard limit. The Reactor will stop when the first step
  completes _after_ the timeout has expired.

  Defaults to `:infinity`.
  """
  @type timeout_option :: {:timeout, pos_integer() | :infinity}

  @typedoc """
  The maximum number of iterations which after which the Reactor will halt.

  Defaults to `:infinity`.
  """
  @type max_iterations_option :: {:max_iterations, pos_integer() | :infinity}

  @typedoc """
  How long to wait for asynchronous steps to complete when halting.

  Defaults to 5000ms.
  """
  @type halt_timeout_option :: {:halt_timeout, pos_integer() | :infinity}

  @typedoc """
  When set to `false` forces the Reactor to run every step synchronously,
  regardless of the step configuration.

  Defaults to `true`.
  """
  @type async_option :: {:async?, boolean}

  @typedoc """
  Use a `Reactor.Executor.ConcurrencyTracker.pool_key` to allow this Reactor to
  share it's concurrency pool with other Reactor instances.

  If you do not specify one then the Reactor will initialise a new pool and
  place it in it's context for any child Reactors to re-use.

  Only used if `async?` is set to `true`.
  """
  @type concurrency_key_option :: {:concurrency_key, reference()}

  @typedoc """
  When this option is set the Reactor will return a copy of the completed Reactor
  struct for potential future undo.
  """
  @type fully_reversible_option :: {:fully_reversible?, boolean}

  @type run_options ::
          Enumerable.t(
            max_concurrency_option
            | timeout_option
            | max_iterations_option
            | halt_timeout_option
            | async_option
            | concurrency_key_option
            | fully_reversible_option
          )

  @type undo_options ::
          Enumerable.t(
            max_concurrency_option
            | timeout_option
            | max_iterations_option
            | halt_timeout_option
            | async_option
            | concurrency_key_option
          )

  @type state :: :pending | :executing | :halted | :failed | :successful
  @type inputs :: %{optional(atom) => any}

  @type t :: %Reactor{
          context: context,
          id: any,
          input_descriptions: %{atom => String.t()},
          inputs: [atom],
          intermediate_results: %{any => any},
          middleware: [Reactor.Middleware.t()],
          plan: nil | Graph.t(),
          undo: [{Step.t(), any}],
          return: any,
          state: state,
          steps: [Step.t()]
        }

  @doc "A guard which returns true if the value is a Reactor struct"
  @spec is_reactor(any) :: Macro.t()
  defguard is_reactor(reactor) when is_struct(reactor, __MODULE__)

  @run_schema [
    max_concurrency: [
      type: :pos_integer,
      required: false,
      doc: "The maximum number of processes to use to run the Reactor"
    ],
    timeout: [
      type: {:or, [:pos_integer, {:literal, :infinity}]},
      required: false,
      default: :infinity,
      doc: "How long to allow the Reactor to run for"
    ],
    max_iterations: [
      type: {:or, [:pos_integer, {:literal, :infinity}]},
      required: false,
      default: :infinity,
      doc: "The maximum number of times to allow the Reactor to loop"
    ],
    async?: [
      type: :boolean,
      required: false,
      default: true,
      doc: "Whether to allow the Reactor to start processes"
    ],
    concurrency_key: [
      type: :reference,
      required: false,
      hide: true
    ],
    run_id: [
      type: :any,
      required: false,
      doc: "A unique identifier for the Reactor run"
    ],
    fully_reversible?: [
      type: :boolean,
      required: false,
      default: false,
      doc: "Return the completed reactor as well as the result for possible later reversal"
    ]
  ]

  @doc """
  Attempt to run a Reactor.

  ## Arguments

  * `reactor` - The Reactor to run, either a Reactor DSL module, or a Reactor
    struct.
  * `inputs` - A map of values passed in to satisfy the Reactor's expected
    inputs.
  * `context` - An arbitrary map that will be merged into the Reactor context
    and passed into each step.

  ## Options

  #{Spark.Options.docs(@run_schema)}
  """
  @spec run(t | module, inputs, context_arg, run_options) ::
          {:ok, any} | {:ok, any, t} | {:error, any} | {:halted, t}
  def run(reactor, inputs \\ %{}, context \\ %{}, options \\ [])

  def run(reactor, inputs, context, options) when is_atom(reactor) do
    if Spark.Dsl.is?(reactor, Reactor) do
      run(reactor.reactor(), inputs, context, options)
    else
      {:error,
       ArgumentError.exception(message: "Module `#{inspect(reactor)}` is not a Reactor module")}
    end
  end

  def run(reactor, inputs, context, options)
      when is_reactor(reactor) and reactor.state in ~w[pending halted]a do
    Executor.run(reactor, inputs, context, options)
  end

  def run(reactor, _inputs, _context, _options) do
    {:error,
     StateError.exception(
       reactor: reactor,
       state: reactor.state,
       expected: ~w[pending halted]a
     )}
  end

  @doc "Raising version of `run/4`."
  @spec run!(t | module, inputs, context_arg, run_options) :: any | no_return
  def run!(reactor, inputs \\ %{}, context \\ %{}, options \\ [])

  def run!(reactor, inputs, context, options) do
    case run(reactor, inputs, context, options) do
      {:ok, value} -> value
      {:error, reason} -> raise reason
    end
  end

  @undo_options Keyword.drop(@run_schema, [:fully_reversible?])

  @doc """
  Attempt to undo a previously successful Reactor.

  ## Arguments

  * `reactor` - The previously successful Reactor struct.
  * `context` - An arbitrary map that will be merged into the Reactor context and passed into each undo.

  ## Options

  #{Spark.Options.docs(@undo_options)}
  """
  @spec undo(t, context_arg, undo_options) :: :ok | {:error, any}
  def undo(reactor, context, options \\ [])

  def undo(reactor, _context, _options) when not is_struct(reactor, __MODULE__) do
    {:error,
     ArgumentError.exception(
       message: "`reactor` value `#{inspect(reactor)}` is not a Reactor struct"
     )}
  end

  def undo(reactor, _context, _options) when reactor.state != :successful do
    {:error,
     StateError.exception(
       reactor: reactor,
       state: reactor.state,
       expected: ~w[successful]a
     )}
  end

  def undo(_reactor, context, _options) when not is_map(context) do
    {:error,
     ArgumentError.exception(
       message: "`context` value `#{inspect(context)}` is not valid context - must be a map"
     )}
  end

  def undo(_reactor, _context, options) when not is_list(options) do
    {:error,
     ArgumentError.exception(
       message: "`options` value `#{inspect(options)}` is not a keyword list"
     )}
  end

  def undo(reactor, context, options) do
    Reactor.Executor.undo(reactor, context, options)
  end

  @doc "A raising version of `undo/2`"
  @spec undo!(t, context_arg, undo_options) :: :ok | no_return
  def undo!(reactor, context, options) do
    with {:error, reason} <- undo(reactor, context, options) do
      raise reason
    end
  end
end