dep_multi

defmodule DepMulti do
  @moduledoc """
  `DepMulti` is a data structure for performing dependant asynchronous
  operations.

  To demonstrate the problem, image you have three asynchronous tasks:

  ##
    [
      Task.async(fn -> do_some_work() end),
      Task.async(fn -> do_some_other_work() end),
      Task.async(fn -> do_more_work() end),
    ] |> Enum.map(&Task.await/1)

  This works until you have a dependency / ordering issue. Say `do_more_work`
  must happen after `do_some_work`. At that point, you'll need to name each
  task and reference those names in other tasks that are dependant on them. If
  those dependant task needs data from a previous task, you'll also want pass
  those changes into the function.

  The API to solve this issue is based on `Ecto.Multi`(https://hexdocs.pm/ecto/Ecto.Multi.html),
  and specifically `run/3` and `run/5`. The differences are:
  * The tasks are executed asychronously
  * the `run` methods take a list of dependencies and a timeout
  * the changes passed into the function only include direct or indirect
  dependencies

  ## Example
    iex>   DepMulti.new()
    ...>   |> DepMulti.run(:step_1, [], fn _ ->
    ...>     {:ok, 1}
    ...>   end)
    ...>   |> DepMulti.run(:step_2a, [dependencies: [:step_1], timeout: 5000], fn %{step_1: value} ->
    ...>     {:ok, 2 + value}
    ...>   end)
    ...>   |> DepMulti.run(:step_2b, [dependencies: [:step_1]], Map, :fetch, [:step_1])
    ...>   |> DepMulti.execute()
    {:ok, %{step_1: 1, step_2a: 3, step_2b: 1}}
  """

  alias __MODULE__
  defstruct operations: [], names: MapSet.new()

  @type changes :: map
  @type run :: (changes -> {:ok | :error, any}) | {module, atom, [any]}
  @type dependencies :: list(name)
  @type run_cmd :: {:run, run}
  @type operations :: [DepMulti.Operation.t()]
  @type names :: MapSet.t()
  @type name :: any
  @type t :: %__MODULE__{operations: operations, names: names}

  @doc """
  Returns an empty `DepMulti` struct.

  ## Example

      iex> DepMulti.new() |> DepMulti.to_list()
      []
  """
  @spec new :: t
  def new do
    %DepMulti{}
  end

  @doc """
  Adds a function to run as part of the multi.

  The function should return either `{:ok, value}` or `{:error, value}`,
  and receives the changes so far as the only argument.

  NOTE: The changes will only include direct or indirect dependencies

  ## Example

      DepMulti.run(multi, :write, [:image], fn %{image: image} ->
        with :ok <- File.write(image.name, image.contents) do
          {:ok, nil}
        end
      end)
  """
  @spec run(t, name, keyword, run) :: t
  def run(multi, name, opts, run) when is_function(run, 1) do
    add_operation(multi, name, opts, {:run, run})
  end

  @doc """
  Adds a function to run as part of the multi.

  Similar to `run/4`, but allows to pass module name, function and arguments.
  The function should return either `{:ok, value}` or `{:error, value}`, and
  receives the rthe changes so far as the first argument (prepended to those
  passed in the call to the function).

  NOTE: The changes will only include direct or indirect dependencies
  """
  @spec run(t, name, keyword, module, function, args) :: t when function: atom, args: [any]
  def run(multi, name, opts, mod, fun, args)
      when is_atom(mod) and is_atom(fun) and is_list(args) do
    add_operation(multi, name, opts, {:run, {mod, fun, args}})
  end

  @spec add_operation(t, name, keyword, {:run, run}) :: t
  defp add_operation(%DepMulti{} = multi, name, opts, run_cmd) do
    %{operations: operations, names: names} = multi

    if MapSet.member?(names, name) do
      raise "#{inspect(name)} is already a member of the DepMulti: \n#{inspect(multi)}"
    end

    timeout = Keyword.get(opts, :timeout, :infinity)
    dependencies = Keyword.get(opts, :dependencies, [])

    unknown_keys = Keyword.keys(opts) -- [:timeout, :dependencies]

    unless Enum.empty?(unknown_keys) do
      raise "Unknown keys: #{inspect(unknown_keys)}"
    end

    operation = %DepMulti.Operation{
      name: name,
      dependencies: dependencies,
      run_cmd: run_cmd,
      timeout: timeout
    }

    %{
      multi
      | operations: [operation | operations],
        names: MapSet.put(names, name)
    }
  end

  @doc """
  Returns the list of operations stored in `multi`.
  """
  @spec to_list(t) :: operations
  def to_list(%DepMulti{operations: operations}) do
    operations
    |> Enum.reverse()
  end

  @spec execute(t, keyword) ::
          {:ok, changes} | {:error, name, any, changes} | {:terminate, name, any, changes}
  def execute(multi, opts \\ []) do
    operations = Enum.reverse(multi.operations)

    DepMulti.Server.execute(operations, opts)
  end
end