retort

defmodule Retort.Client do
  @moduledoc """
  Functionality shared between all specific, non-generic `Retort.Client.*`
  """

  alias Alembic.Document
  alias Retort.{Connection, Response}

  # CONSTANTS

  @default_timeout 5_000 # ms

  # Types

  @typedoc """
  Maps client pids to the structs created with by those clients.
  """
  @type created_by_pid :: %{pid => __MODULE__.Generic.resource}

  @typedoc """
  Set of client pids where `Retort.Client.Generic.destroy_all` has already been called
  """
  @type destroyed_set :: MapSet.t

  @typedoc """
  The name of a factory in `Factory` for generating the fields for `InterpreterServrerRPC.Client.Generic.create`
  """
  @type factory :: atom

  @typedoc """
  A module that implements one of the following callbacks
  * `destroy_dag(destroyed_set, context) :: {:ok, destroyed_set} | Retort.Client.Generic.error`
  * `purge(%AMQP.Channel{})`
  * `setup(context) :: context`
  """
  @type client_module :: module

  @typedoc """
  A (non-empty) list of `client_module`
  """
  @type client_modules :: [client_module, ...]

  @typedoc """
  A map as passed to
  [`ExUnit.Callbacks.setup/2`](http://elixir-lang.org/docs/stable/ex_unit/ExUnit.Callbacks.html#setup/2).
  """
  @type context :: map

  @typedoc """
  Field value overrides of default valeus in `factory`
  """
  @type overrides :: Keyword.t

  # Functions

  @doc """
  Calls `Retort.Client.create` with fields from `factory` with `overrides` and `params` to help with
  includes
  """
  @spec create_factory(
          %{
            required(:factory) => factory,
            required(:factory_module) => module,
            optional(:overrides) => list,
            optional(:params) => map,
            required(:pid) => pid,
            optional(:timeout) => timeout
          }
        ) :: {:ok, __MODULE__.Generic.resource} | __MODULE__.Generic.error
  def create_factory(options = %{factory: factory, factory_module: factory_module, pid: pid}) do
    overrides = Map.get(options, :overrides, [])
    fields = apply(factory_module, :params_for, [factory, overrides])
    params = Map.get(options, :params, %{})
    timeout = Map.get(options, :timeout, @default_timeout)
    __MODULE__.Generic.create(pid, fields, params, timeout)
  end

  @doc """
  Only calls `Retort.Client.create` if the `pid` is not a key in `created_by_pid`.
  """
  @spec create_factory_once(
          %{
            required(:created_by_pid) => created_by_pid,
            required(:factory) => factory,
            required(:factory_module) => module,
            optional(:overrides) => list,
            optional(:params) => map,
            required(:pid) => pid,
            optional(:timeout) => timeout
          }
        ) :: {:ok, created_by_pid} | __MODULE__.Generic.error
  def create_factory_once(options = %{created_by_pid: created_by_pid, pid: pid}) do
    case Map.fetch(created_by_pid, pid) do
      {:ok, _} ->
        {:ok, created_by_pid}
      :error ->
        case create_factory(Map.delete(options, :created_by_pid)) do
          {:ok, created} ->
            {:ok, Map.put(created_by_pid, pid, created)}
          {
            :error,
            reason = %Response.Error{
              data: %Document{
                errors: errors
              }
            }
          } when is_list(errors) ->
            {:error, put_in(reason.data.meta, %{"create_factory/1 options" => options})}
          other -> other
        end
    end
  end

  @doc """
  Destroys the Directed Acyclic Graph (DAG) by destroying the DAG for all `referencing_modules` before destroying
  all referenced resources with the pid assigned to `key` in `context`

  ## Parameters
  * `destroyed_set` - set of pids that have already had `destroy_all_once/2` called on them.
  * `context` - Maps atom keys, like `location_client_pid` to the pid for that resource's client
  * `key` - An atom that selects a pid from the `context`
  * `referencing_modules` - Modules that define `destroy_dag(destroyed_dag, context)`
  """
  @spec destroy_dag(
          destroyed_set,
          context,
          key :: atom,
          referencing_modules :: client_modules
        ) :: {:ok, destroyed_set} | __MODULE__.Generic.error
  def destroy_dag(destroyed_set, context, key, referencing_modules) do
    with {:ok, referencing_modules_destroyed_set} <- destroy_dags(destroyed_set, context, referencing_modules) do
      timeout = Map.get(context, :client_timeout, @default_timeout)
      destroy_all_once(referencing_modules_destroyed_set, Map.fetch!(context, key), timeout)
    end
  end

  @doc """
  Destroys the Directed Acyclic Graph (DAG) of each module in `modules` in order as long as the previous module's
  `destroy_dag(destroyed_set, context)` returned `{:ok, new_destroyed_set}`; otherwise, returns the first error.
  """
  @spec destroy_dags(destroyed_set, context, client_modules) :: {:ok, destroyed_set} | __MODULE__.Generic.error
  def destroy_dags(destroyed_set, context, modules) do
    Enum.reduce_while modules, {:ok, destroyed_set}, fn referencing_module, {:ok, acc_destroyed_set} ->
      case referencing_module.destroy_dag(acc_destroyed_set, context) do
        acc = {:ok, _} ->
          {:cont, acc}
        acc ->
          {:halt, acc}
      end
    end
  end

  @doc """
  Destroys the Directect Acyclic Graph (DAG) of each module in `modules` in order if `destroy_dag: true` is in
  `context`.  Fails if there is an error from `destroy_dags/3`; otherwise, returns `context`, so it can be piped with
  other setup functions.
  """
  @spec destroy_dags_setup(context, client_modules) :: context | no_return

  def destroy_dags_setup(context = %{destroy_dag: true}, modules) do
    {:ok, _} = destroy_dags(MapSet.new, context, modules)

    context
  end

  def destroy_dags_setup(context, modules) when is_map(context) and is_list(modules), do: context

  @doc """
  Only calls `IntepreterServer.Rpc.Client.Generic.destroy_all` if the `pid` is not in `destroyed_set`
  """
  @spec destroy_all_once(destroyed_set, pid) :: {:ok, destroyed_set} | __MODULE__.Generic.error
  @spec destroy_all_once(destroyed_set, pid, timeout) :: {:ok, destroyed_set} | __MODULE__.Generic.error
  def destroy_all_once(destroyed_set, pid, timeout \\ @default_timeout) do
    if MapSet.member?(destroyed_set, pid) do
      {:ok, destroyed_set}
    else
      with :ok <- __MODULE__.Generic.destroy_all(pid, %{}, timeout) do
        {:ok, MapSet.put(destroyed_set, pid)}
      end
    end
  end

  @doc """
  Purges the queue for all `modules` using a shared channel
  """
  @spec purge(client_modules) :: :ok
  def purge(modules) when is_list(modules) do
    {:ok, connection} = Connection.await
    {:ok, channel} = AMQP.Channel.open(connection)

    purge(channel, modules)
  end

  @doc """
  Purges the queue for all `modules` on `channel`
  """
  @spec purge(%AMQP.Channel{}, client_modules) :: :ok
  def purge(channel = %AMQP.Channel{}, modules) when is_list(modules) do
    Enum.each modules, fn module ->
      module.purge(channel)
    end
  end

  @doc """
  Pipes `setup(context)` for each module in `modules`
  """
  @spec setup(context, client_modules) :: map
  def setup(context, modules) when is_map(context) and is_list(modules) do
    Enum.reduce modules, context, fn module, acc_context ->
      module.setup(acc_context)
    end
  end
end