retort

defmodule Retort.Client.Generic do
  @moduledoc """
  A generic client for interacting with a resource over JSON API over JSON RPC over RabbitMQ.
  """

  alias Alembic.{Document, Meta, Pagination, Pagination.Page, Resource, ToEctoSchema}
  alias Retort.{Api, Channel, Client, Connection, Queue, Request, Response, Response.Error}

  require Logger

  use GenServer

  defstruct channel: nil,
            consumer_tag: nil,
            ecto_schema_module_by_type: %{},
            reply_queue: nil,
            request_by_correlation_id: %{},
            request_queue: nil,
            type: nil,
            type_by_ecto_schema_module: %{},
            meta: %{}

  # Types

  @typedoc """
  Maps Ecto.Schema Module for building resources back to JSON API `type` field, so the inverse of
  `IntepreterServer.Api.ToEctoSchema.ecto_schema_module_by_type`.
  """
  @type type_by_ecto_schema_module :: %{ToEctoSchema.ecto_schema_module => Resource.type}

  @typedoc """
  * `channel` - The channel connected to the RabbitMQ broker.  Created from the `connection` passed to `start_link/2`.
  * `consumer_tag` - The consumer tag assigned to the process when it consumes its `reply_queue` on `channel`
  * `ecto_schema_module_by_type` - Maps JSON API `type` field to the corresponding Ecto.Schema Module for building
    resources from `Retort.Response` results.
  * `meta` - Meta information about jsonapi request.
  * `reply_queue` - The name of queue on which this server gets replies
  * `request_by_correlation_id` - Tracks the `from` (calling pid and reference) that called a client function and which
    method was called as teh method influences the allowed formats for responses.
  * `request_queue` - The name of the queue on which this server publishes requests.  Matches `queue` passed to
    `start_link/2`.
  * `type` - Type used for `create/2`
  * `type_by_ecto_schema_module` - Maps model back to JSON API `type` for relationship `"type"` field.  Automatically
    derived from `ecto_schema_module_by_type`.
  """
  @type t :: %__MODULE__{
               channel: %AMQP.Channel{},
               consumer_tag: String.t,
               ecto_schema_module_by_type: ToEctoSchema.ecto_schema_module_by_type,
               meta: map,
               reply_queue: String.t,
               request_by_correlation_id: %{String.t => ClientRequest.t},
               request_queue: String.t,
               type: String.t,
               type_by_ecto_schema_module: type_by_ecto_schema_module
             }

  @typedoc """
  Common tagged tuple return format for errors.
  """
  @type error :: {:error, Error.t}

  @typedoc """
  `id` field of a `resource`
  """
  @type id :: String.t

  @typedoc """
  The name of a queue
  """
  @type queue :: String.t

  @typedoc """
  A model struct for a resource
  """
  @type resource :: struct

  # Constants

  @default_timeout 5000

  # Functions

  @doc """
  Converts `Retort.Client.Generic.error` to errors returned by `Calcinator` and `Calcinator.Resources` callbacks.
  """
  @spec error_to_calcinator_error(error) :: {:error, :bad_gateway} |
                                            {:error, :not_found} |
                                            {:error, :sandbox_access_disallowed} |
                                            {:error, Ecto.Changeset.t}
  def error_to_calcinator_error(error), do: error_to_calcinator_error(error, nil)

  @doc """
  Converts `Retort.Client.Generic.error` to errors returned by `Calcinator` and `Calcinator.Resources` callbacks.
  """
  @spec error_to_calcinator_error(error,
                                  maybe_changeset :: Ecto.Changeset.t | nil) :: {:error, :bad_gateway} |
                                                                                {:error, :not_found} |
                                                                                {:error, :sandbox_access_disallowed} |
                                                                                {:error, Ecto.Changeset.t}
  def error_to_calcinator_error({:error, error = %Error{}}, changeset), do: Error.to_calcinator_error(error, changeset)

  ## Client API

  @doc """
  Counts the number of resources
  """
  @spec count(pid) :: non_neg_integer
  @spec count(pid, map) :: non_neg_integer
  @spec count(pid, map, timeout) :: non_neg_integer
  def count(pid, params \\ %{}, timeout \\ @default_timeout) when is_map(params) do
    limited_params = Map.merge(params, Page.to_params(%Page{size: 1}))

    case index(pid, limited_params, timeout) do
      {:ok, _, %Pagination{total_size: total_size}} ->
        total_size
      {:ok, all_resources, nil} ->
        length(all_resources)
    end
  end

  @doc """
  Creates a resource.

  * `pid` - the pid of the client
  * `convertable_params` - the nested params to be converted to the data and relationships in the JSON API document.
    All keys should be `String.t` and **NOT** `atom`.
  * `mergable_params` - additional params to be merged with the JSON API document created from `convertable_params`.
    This allows setting of `include` query params or `meta` in the document.

  **All keys in `convertable_params` and `mergable_params` should be `String.t` and NOT `atom`.**
  """
  @spec create(pid, map) :: {:ok, resource} | error
  @spec create(pid, map, map) :: {:ok, resource} | error
  @spec create(pid, map, map, timeout) :: {:ok, resource} | error
  def create(pid,
             convertable_params,
             mergable_params \\ %{},
             timeout \\ @default_timeout) when is_map(convertable_params) and is_map(mergable_params) do
    GenServer.call(pid, {:create, convertable_params, mergable_params}, timeout)
  end

  @doc """
  Destroys a resource by `id`
  """
  @spec destroy(pid, id) :: :ok | error
  @spec destroy(pid, id, timeout) :: :ok | error
  def destroy(pid, id, timeout \\ @default_timeout) when is_binary(id) do
    GenServer.call(pid, {:destroy, id}, timeout)
  end

  @doc """
  Destroys all resources.  Stops on first error.
  """
  @spec destroy_all(pid) :: :ok | error
  @spec destroy_all(pid, map) :: :ok | error
  @spec destroy_all(pid, map, timeout) :: :ok | error
  def destroy_all(pid, params \\ %{}, timeout \\ @default_timeout) do
    with {:ok, resources, pagination} <- index(pid, params, timeout),
         :ok <- destroy_until_error(pid, timeout, resources) do
      destroy_rest(pid, params, timeout, pagination)
    end
  end

  @doc """
  Retrieves the index of this client's resource.
  """
  @spec index(pid) :: {:ok, [resource], Pagination.t | nil} | error
  @spec index(pid, map) :: {:ok, [resource], Pagination.t | nil} | error
  @spec index(pid, map, timeout) :: {:ok, [resource], Pagination.t | nil} | error
  def index(pid, params \\ %{}, timeout \\ @default_timeout) when is_map(params) do
    GenServer.call(pid, {:index, params}, timeout)
  end

  @doc """
  Retrieves resource with the given `id`.
  """
  @spec show(pid, id) :: {:ok, resource} | error
  @spec show(pid, id, map) :: {:ok, resource} | error
  @spec show(pid, id, map, timeout) :: {:ok, resource} | error
  def show(pid, id, params \\ %{}, timeout \\ @default_timeout) when is_binary(id) and is_map(params) do
    GenServer.call(pid, {:show, id, params}, timeout)
  end

  @doc """
  Starts a client pid to send requests and wait for responses.  To have more than one request in flight, you need
  multiple clients from multiple calls to this function.

  * `:connection` (`Retort.Connection.await/0`) - `%AMQP.Connection{}` connection.
  * `:ecto_schema_module_by_type` - Maps JSON API `type` field to the corresponsing Ecto.Schema Module for building
    resources from `Retort.Response` results.
  * `:meta` - Free form data structure to pass meta information in rpc request.  Defaults to setting `%{host: }` from
     application env var.  MUST pass `Retort.Meta.valid!/2` if there are databased-backed `ecto_schema_module` in
     `:ecto_schema_module_by_type` and `:ecto_repo_module` is sandboxed.
  * `:queue` - The name of the queue on which this server publishes requests.  Matches `queue` passed to
    `start_link/2`.
  * `:ecto_repo_module` - The `Ecto.Repo` whose `Ecto.Repo.config/0` `:pool` to check if it is
    `Ecto.Adapters.SQL.Sandbox` if any of `ecto_schema_module` in `ecto_schema_module_by_type` has a
    `__schema__(:source)` set, which is assumed to mean they are backed by this `Ect.Repo.t` and so would be sandboxed.
  * `:type` - The primary JSON API `"type"`, such as for `create/2`.
  """
  @spec start_link(connection: %AMQP.Connection{},
                   ecto_schema_module_by_type: ToEctoSchema.ecto_schema_module_by_type,
                   meta: map,
                   queue: queue,
                   ecto_repo_module: module,
                   type: String.t) :: GenServer.on_start
  def start_link(opts \\ []) do
    ecto_schema_modules = opts
                          |> opts_to_ecto_schema_module_by_type
                          |> Map.values
    meta = opts
           |> Keyword.get(:meta, %{})
           |> Retort.Meta.valid!(
                %{
                  ecto_schema_modules: ecto_schema_modules,
                  ecto_repo_module: Keyword.get(opts, :ecto_repo_module)
                }
              )
           |> Map.put_new(:host, Application.get_env(:retort, :host))

    opts = Keyword.put(opts, :meta, meta)
    GenServer.start_link(__MODULE__, opts, [])
  end

  @doc """
  Streams the index of this client's resource, automatically fetching the next page.

  Because the generated stream may make multiple requests, the `timeout` is the `timeout` for each individual `index/3`
  request.

  **NOTE: DO NOT `destroy/3` the returned resources if you plan to continuing using the stream as it will cause the
  cached pagination information to be invalid as you may have deleted enough resources that the page no longer exists.**

  If you need to destroy resources from a stream do one of the following:

  1. Convert it to a list, so all the remote `index/3` requests are complete
  2. Use `destroy_all/3` to delete the first page repeatedly until all resources are deleted.

  ## Convert Stream to List Before Destroying

  (1) has the benefit that only the resources you can see at the time of the call to `Enum.to_list` would be destoryed,
  but you need to download all the resources at once into a list.

      iex> pid
      ...> |> Retort.Client.Generic.stream
      ...> |> Enum.to_list
      ...> |> Enum.each(fn %{id: id} ->
      ...>      Retort.Client.Generic.destroy(to_string(id))
      ...>    end)

  ## Use `destroy_all/3`

      iex> Retort.Client.Generic.destroy_all(pid)

  (2) only loads a page of resources at a time, but can never finish if new resources are being created that keep the
  first page with at least one resource when it rechecks.
  """
  @spec stream(pid) :: Enumerable.t
  @spec stream(pid, map) :: Enumerable.t
  @spec stream(pid, map, timeout) :: Enumerable.t
  def stream(pid, params \\ %{}, timeout \\ @default_timeout) when is_map(params) do
    Client.Stream.new(pid, params, timeout)
  end

  @doc """
  Stops a client's `GenServer` that is waiting for responses.

  # Starting and stopping a `Retort.Client.Generic`

      iex> {:ok, pid} = Retort.Client.Generic.start_link("generic")
      iex> Retort.Client.Partner.stop(pid)
      :ok
  """
  @spec stop(pid) :: :ok
  def stop(pid) do
    :gen_server.stop(pid)
  end

  @doc """
  Update the attributes and/or associations of resource with the given `id`

  * `pid` - the pid of the client
  * `id` - the id of the resource to update as a `String.t`.
  * `convertable_params` - the nested params to be converted to the data and relationships in the JSON API document.
    All keys should be `String.t` and **NOT** `atom`.
  * `mergable_params` - additional params to be merged with the JSON API document created from `convertable_params`.
    This allows setting of `include` query params or `meta` in the document.

  **All keys in `convertable_params` and `mergable_params` should be `String.t` and NOT `atom`.**
  """
  @spec update(pid, id, map) :: {:ok, resource} | error
  @spec update(pid, id, map, map) :: {:ok, resource} | error
  @spec update(pid, id, map, map, timeout) :: {:ok, resource} | error
  def update(pid,
             id,
             convertable_params,
             mergable_params \\ %{},
             timeout \\ @default_timeout) when is_binary(id) and
                                               is_map(convertable_params) and
                                               is_map(mergable_params) do
    string_keys!(convertable_params)
    GenServer.call(pid, {:update, id, convertable_params, mergable_params}, timeout)
  end

  ## GenServer Callbacks

  @spec handle_call({:create, map, map}, GenServer.from, t) :: {:noreply, t}
  def handle_call({:create, value_by_name, params}, from, state) when is_map(value_by_name) and is_map(params) do
    data = nested_params_to_json_api_resource(value_by_name, state)

    new_state = publish(
      state,
      from,
      %Request{
        method: "create",
        params: Map.merge(
          params,
          %{
            data: data
          }
        )
      }
    )

    {:noreply, new_state}
  end

  @spec handle_call({:destroy, String.t}, GenServer.from, t) :: {:noreply, t}
  def handle_call({:destroy, id}, from, state) when is_binary(id) do
    new_state = publish(
      state,
      from,
      %Request{
        method: "destroy",
        params: %{
          "id" => id
        }
      }
    )

    {:noreply, new_state}
  end

  @spec handle_call({:index, map}, GenServer.from, t) :: {:noreply, t}
  def handle_call({:index, params}, from, state) do
    new_state = publish(state, from, %Request{method: "index", params: params})

    {:noreply, new_state}
  end

  @spec handle_call({:show, String.t, map}, GenServer.from, t) :: {:noreply, t}
  def handle_call({:show, id, params}, from, state) when is_binary(id) do
    new_state = publish(
      state,
      from,
      %Request{
        method: "show",
        params: Map.merge(
          params,
          %{
            id: id
          }
        )
      }
    )

    {:noreply, new_state}
  end

  @spec handle_call({:update, String.t, map, map}, GenServer.from, t) :: {:noreply, t}
  def handle_call({:update, id, value_by_name, params}, from, state) when is_binary(id) do
    data = value_by_name
           |> Map.put("id", id)
           |> nested_params_to_json_api_resource(state)

    new_state = publish(
      state,
      from,
      %Request{
        method: "update",
        params: Map.merge(
          params,
          %{
            data: data,
            id: id
          }
        )
      }
    )

    {:noreply, new_state}
  end

  @doc """
  The consumer process will receive the following data structures:

  * `{:basic_deliver, payload, meta}` - This is sent for each message consumed, where `payload` contains the message
    content and `meta` contains all the metadata set when sending with
    [`AMQP.Basic.publish/5`](http://hexdocs.pm/amqp/AMQP.Basic.html#publish/5) or additional info set by the broker.
  * `{:basic_consume_ok, %{consumer_tag: consumer_tag}}` - Sent when the consumer process is registered with
    [`AMQP.Basic.consume/4`](http://hexdocs.pm/amqp/AMQP.Basic.html#consume/4). The caller receives the same information
    as the return of [`AMQP.Basic.consume/4`](http://hexdocs.pm/amqp/AMQP.Basic.html#consume/4).
  * `{:basic_cancel, %{consumer_tag: consumer_tag, no_wait: no_wait}}` - Sent by the broker when the consumer is
    unexpectedly cancelled (such as after a queue deletion)
  * `{:basic_cancel_ok, %{consumer_tag: consumer_tag}}` - Sent to the consumer process after a call to
    [`AMQP.Basic.cancel`](http://hexdocs.pm/amqp/AMQP.Basic.html#cancel/3)
  """

  @spec handle_info({:basic_cancel, %{required(:consumer_tag) => String.t, optional(any) => any}}, t) ::
        {:stop, :normal, t}
  def handle_info({:basic_cancel, %{consumer_tag: consumer_tag}}, state = %__MODULE__{consumer_tag: consumer_tag}) do
    {:stop, :normal, state}
  end

  @spec handle_info({:basic_cancel_ok, %{required(:consumer_tag) => String.t, optional(any) => any}}, t) ::
        {:noreply, t}
  def handle_info({:basic_cancel_ok, %{consumer_tag: consumer_tag}}, state = %__MODULE__{consumer_tag: consumer_tag}) do
    {:noreply, state}
  end

  @spec handle_info({:basic_consume, %{required(:consumer_tag) => String.t, optional(any) => any}}, t) :: {:noreply, t}
  def handle_info({:basic_consume, %{consumer_tag: consumer_tag}}, state = %__MODULE__{consumer_tag: consumer_tag}) do
    {:noreply, state}
  end

  @spec handle_info({:basic_consume_ok, %{required(:consumer_tag) => String.t, optional(any) => any}}, t) ::
        {:noreply, t}
  def handle_info({:basic_consume_ok, %{consumer_tag: consumer_tag}},
                  state = %__MODULE__{consumer_tag: consumer_tag}) do
    {:noreply, state}
  end

  @spec handle_info(
          {:basic_deliver, String.t, %{required(:correlationed_id) => String.t, required(:delivery_tag) => binary}},
          t
        ) :: {:noreply, t}
  def handle_info({:basic_deliver,
                   encoded_response,
                   %{correlation_id: correlation_id, delivery_tag: delivery_tag}},
                  state = %__MODULE__{
                            channel: channel,
                            ecto_schema_module_by_type: ecto_schema_module_by_type,
                            request_by_correlation_id: request_by_correlation_id,
                          }) do
    Logger.debug fn ->
      ["Receiving response ", encoded_response]
    end

    new_state = case Map.fetch(request_by_correlation_id, correlation_id) do
      {:ok, request = %Client.Request{from: from}} ->
        meta = Client.Request.to_meta(request)
        encoded_reponse_reply = reply(encoded_response, correlation_id, meta, ecto_schema_module_by_type)

        GenServer.reply(from, encoded_reponse_reply)

        AMQP.Basic.ack(channel, delivery_tag)

        %__MODULE__{state | request_by_correlation_id: Map.delete(state.request_by_correlation_id, correlation_id)}
      :error ->
        from_error(correlation_id, channel, delivery_tag)
        state
    end

    {:noreply, new_state}
  end

  # catchall clause
  def handle_info(info, state) do
    Logger.error fn ->
      ["Got unexpected info ", inspect(info)]
    end

    {:noreply, state}
  end

  @doc """
  Creates a channel and anonymous queue for this GenServer to receive replies on.

  * `:connection` (`Retort.Connection.await/0`) - `%AMQP.Connection{}` connection.
  * `:ecto_schema_module_by_type` - Maps JSON API `type` field to the corresponsing Ecto.Schema Module for building
    resources from `Retort.Response` results.
  * `:meta` Free form meta data about the JSONAPI request.
  * `:queue` - The name of the queue on which this server publishes requests.  Matches `queue` passed to
    `start_link/2`.
  * `:type` - The primary JSON API `"type"`, such as for `create/2`.
  """
  @spec init([]) :: {:ok, t}
  def init(opts \\ []) do
    Process.flag(:trap_exit, true)

    state = opts_to_state(opts)

    {:ok, channel} = opts
                     |> opts_to_connection
                     |> AMQP.Channel.open

    {:ok, %{queue: reply_queue}} = AMQP.Queue.declare(channel, "", 'auto-delete': true, exclusive: true)
    # Register the GenServer process as a consumer
    {:ok, consumer_tag} = AMQP.Basic.consume(channel, reply_queue)

    {:ok, %{state | channel: channel, consumer_tag: consumer_tag, reply_queue: reply_queue}}
  end

  @doc """
  Closes the channel this GenServer instance started
  """
  @spec terminate(any, t) :: :ok
  def terminate(_reason, %__MODULE__{channel: channel, reply_queue: reply_queue}) do
    Queue.ensure_deleted(reply_queue, channel)
    Channel.ensure_closed(channel)

    :ok
  end

  ## Private functions

  defp add_id_to_json_api_resource(resource, nil), do: resource
  defp add_id_to_json_api_resource(resource, id), do: Map.put(resource, :id, to_string(id))

  defp changeset_to_json_api_resource(changeset = %Ecto.Changeset{
                                        changes: changes,
                                        data: %{__struct__: ecto_schema_module}
                                      },
                                      fields,
                                      associations,
                                      state = %__MODULE__{type: type}) do
    [primary_key] = ecto_schema_module.__schema__(:primary_key)
    attributes = changeset_to_attributes(changeset, fields, primary_key)

    %{
      attributes: attributes,
      type: type
    }
    |> add_id_to_json_api_resource(changes[primary_key])
    |> merge_associations_into_json_api_resource(associations, changeset, state)
    |> Api.serialize_keys
  end

  defp changeset_to_attributes(%Ecto.Changeset{changes: changes}, fields, primary_key) do
    attribute_fields = List.delete(fields, primary_key)
    Map.take(changes, attribute_fields)
  end

  defp castable_associations(%{__struct__: ecto_schema}, associations)
       when is_atom(ecto_schema) and is_list(associations) do
    Enum.reject associations, fn association ->
      case ecto_schema.__schema__(:association, association) do
        %Ecto.Association.BelongsTo{} ->
          true
        _ ->
          false
      end
    end
  end

  @spec decode_response(String.t, non_neg_integer) :: Response.t
  defp decode_response(encoded_response, correlation_id) do
    case Response.from_string(encoded_response) do
      matched = %{id: ^correlation_id} ->
        matched
      unmatched = %{id: id} ->
        Logger.warn fn ->
          ["JSON RPC id (", inspect(id), ") does not match RabbitMQ correlation_id (", inspect(correlation_id), ")"]
        end

        unmatched
    end
  end

  # pagination not supported then resources must have been all resources
  defp destroy_rest(_, _, _, nil), do: :ok
  # pagination supported, but this was the last page
  defp destroy_rest(_, _, _, %Pagination{next: nil}), do: :ok
  # pagination supported, delete first page again now that resources have shifted down
  defp destroy_rest(pid, params, timeout, %Pagination{}), do: destroy_all(pid, params, timeout)

  defp destroy_until_error(pid, timeout, resources) do
    Enum.reduce_while resources, :ok, fn (%{id: id}, :ok) ->
      case destroy(pid, to_string(id), timeout) do
        :ok -> {:cont, :ok}
        error -> {:halt, error}
      end
    end
  end

  defp cast_associations(changeset = %Ecto.Changeset{data: data}, associations) do
    data
    |> castable_associations(associations)
    |> Enum.reduce(changeset, fn association_name, acc ->
         Ecto.Changeset.cast_assoc(
           acc,
           association_name,
           with: fn association_schema, association_params ->
                   association_model = association_schema.__struct__

                   %{
                     associations: association_associations,
                     fields: association_fields
                   } = classify_keys(association_model)

                   nested_params_to_changeset(
                     association_params,
                     association_model,
                     association_fields,
                     association_associations
                   )
                 end
         )
       end)
  end

  defp cast_fields(ecto_schema_module, nested_params, fields) do
    fields
    # remove defaults so that any value shows as a change since the current value is not available to
    # calculate a compact update
    |> Enum.reduce(ecto_schema_module.__struct__, fn field, acc ->
         Map.update!(acc, field, fn _ -> nil end)
       end)
    |> Ecto.Changeset.cast(nested_params, fields)
  end

  defp classify_keys(primary_model) do
    associations = primary_model.__schema__(:associations)

    # primary_model.__schema__(:fields) does not include virtual fields, so
    # deduce real and virtual fields from struct keys
    keys = primary_model.__struct__ |> Map.keys
    fields = keys -- [:__meta__, :__struct__ | associations]

    %{associations: associations, fields: fields}
  end

  defp invert(map) when is_map(map), do: Enum.into map, %{}, fn {key, value} -> {value, key} end

  defp merge_association_into_json_api_resource(data,
                                                association_name,
                                                changeset = %Ecto.Changeset{data: %{__struct__: ecto_schema_module}},
                                                state) when is_atom(association_name) do
    merge_association_into_json_api_resource(data,
                                             ecto_schema_module.__schema__(:association, association_name),
                                             changeset,
                                             state)
  end

  defp merge_association_into_json_api_resource(data,
                                                %Ecto.Association.BelongsTo{
                                                  field: relationship,
                                                  owner_key: owner_key,
                                                  related: related
                                                },
                                                %Ecto.Changeset{changes: changes},
                                                %__MODULE__{type_by_ecto_schema_module: type_by_ecto_schema_module}) do
    case Map.fetch(changes, owner_key) do
      {:ok, owner_key_change} ->
        association_type = Map.fetch!(type_by_ecto_schema_module , related)
        association_resource_identifier = add_id_to_json_api_resource(%{type: association_type},
                                                                      owner_key_change)

        data
        |> update_in([:attributes], &Map.drop(&1, [owner_key]))
        |> Map.put_new(:relationships, %{})
        |> put_in([:relationships, relationship], %{data: association_resource_identifier})
      :error ->
        data
    end
  end

  defp merge_association_into_json_api_resource(data,
                                                %{field: association},
                                                %Ecto.Changeset{changes: changes},
                                                %__MODULE__{}) do
    :error = Map.fetch(changes, association)

    data
  end

  defp merge_associations_into_json_api_resource(data, association_names, changeset, state) do
    Enum.reduce association_names, data, fn (association_name, data) ->
      merge_association_into_json_api_resource(data, association_name, changeset, state)
    end
  end

  defp nested_params_to_changeset(nested_params, primary_model, fields, associations) do
    primary_model
    |> cast_fields(nested_params, fields)
    |> cast_associations(associations)
  end

  defp nested_params_to_json_api_resource(nested_params,
                                          state = %{
                                            ecto_schema_module_by_type: ecto_schema_module_by_type,
                                            type: type
                                          }) do
    primary_model = Map.fetch!(ecto_schema_module_by_type, type)
    nested_params_to_json_api_resource(nested_params, primary_model, state)
  end

  defp nested_params_to_json_api_resource(nested_params, primary_model, state) do
    %{associations: associations, fields: fields} = classify_keys(primary_model)

    nested_params_to_json_api_resource(nested_params, primary_model, fields, associations, state)
  end

  defp nested_params_to_json_api_resource(nested_params, primary_model, fields, associations, state) do
    nested_params
    |> nested_params_to_changeset(primary_model, fields, associations)
    |> changeset_to_json_api_resource(fields, associations, state)
  end

  defp opts_to_connection(opts) do
    Keyword.get_lazy opts, :connection, fn ->
      {:ok, connection} = Connection.await

      connection
    end
  end

  defp opts_to_ecto_schema_module_by_type(opts) do
    case Keyword.fetch(opts, :ecto_schema_module_by_type) do
      {:ok, non_empty_map} when map_size(non_empty_map) != 0 ->
        non_empty_map
    end
  end

  defp opts_to_request_queue(opts) do
    case Keyword.fetch(opts, :queue) do
      {:ok, string} when is_binary(string) ->
        string
    end
  end

  defp opts_to_state(opts) do
    ecto_schema_module_by_type = opts_to_ecto_schema_module_by_type(opts)
    {:ok, meta} = Keyword.fetch(opts, :meta)
    request_queue = opts_to_request_queue(opts)
    type = opts_to_type(opts)
    type_by_ecto_schema_module = invert(ecto_schema_module_by_type)

    %__MODULE__{
      ecto_schema_module_by_type: ecto_schema_module_by_type,
      meta: meta,
      request_queue: request_queue,
      type: type,
      type_by_ecto_schema_module: type_by_ecto_schema_module
    }
  end

  defp opts_to_type(opts) do
    case Keyword.fetch(opts, :type) do
      {:ok, string} when is_binary(string) ->
        string
    end
  end

  # Only adds pagination if there are multiple resources
  defp paginate_resources(document, resources) when is_list(resources) do
    pagination = Document.to_pagination(document)
    {:ok, resources, pagination}
  end

  # A singleton resource needs no pagination
  defp paginate_resources(_, resource), do: {:ok, resource}

  @spec publish(t, GenServer.from, Request.t) :: t
  defp publish(state = %__MODULE__{channel: channel,
                                   meta: meta,
                                   reply_queue: reply_queue,
                                   request_queue: request_queue},
               from,
               request = %Request{method: method}) do
    request = %Request{request | params: Map.merge(request.params, %{meta: meta})}

    correlation_id = UUID.uuid4()
    encoded_request = %Request{request | id: correlation_id}
                      |> Poison.encode!
    Logger.debug ["Sending request: ", encoded_request]

    Logger.metadata(correlation_id: correlation_id)
    Logger.debug ["Publishing request: ", encoded_request]
    Logger.metadata(correlation_id: nil)

    AMQP.Basic.publish(
      channel,
      "",
      request_queue,
      encoded_request,
      correlation_id: correlation_id,
      reply_to: reply_queue
    )

    %{
      state | request_by_correlation_id: Map.put(
                                           state.request_by_correlation_id,
                                           correlation_id,
                                           %Retort.Client.Request{
                                             from: from,
                                             method: String.to_existing_atom(method)
                                           }
                                         )
    }
  end

  @spec reply(String.t,
              non_neg_integer,
              Meta.t,
              ToEctoSchema.ecto_schema_module_by_type) :: :ok | {:ok, resource | [resource]} | error
  defp reply(encoded_response, correlation_id, meta, ecto_schema_module_by_type) when is_binary(encoded_response) do
    encoded_response
    |> decode_response(correlation_id)
    |> reply(meta, ecto_schema_module_by_type)
  end

  @spec reply(Response.t,
              Meta.t,
              ToEctoSchema.ecto_schema_module_by_type) :: :ok | {:ok, resource | [resource]} | error

  # No Content response from delete
  defp reply(%Response{error: nil, result: result},
             _meta,
             _ecto_schema_module_by_type) when map_size(result) == 0 do
    :ok
  end

  defp reply(%Response{error: nil, result: result}, meta, ecto_schema_module_by_type) do
    with {:ok, document} <- result_to_document(result, meta),
         {:ok, resource_or_resources} <- Document.to_ecto_schema(document, ecto_schema_module_by_type) do
      paginate_resources(document, resource_or_resources)
    end
  end

  defp reply(%Response{error: error, result: nil}, _, _), do: {:error, error}

  defp result_to_document(result, meta) do
    result
    |> Api.deserialize_keys
    |> Document.from_json(
         %Alembic.Error{
           meta: meta,
           source: %Alembic.Source{
             pointer: ""
           }
         }
       )
  end

  @spec string_keys!(map) :: :ok | no_return
  defp string_keys!(params) when is_map(params) do
    Enum.each params, fn
      {key, _value} when is_binary(key) -> :ok
      {key, _value} when is_atom(key) ->
        raise ArgumentError, "expected params to be %{String.t => term}, but go a map with atom key: #{inspect params}"
    end
  end

  defp from_error(correlation_id, channel, delivery_tag) do
    Logger.error "No `from` associated with correlation_id (#{inspect correlation_id})."

    AMQP.Basic.reject(channel, delivery_tag)
  end
end