retort

defmodule Retort.Server.Generic.RPC do
  @moduledoc """
  A request, its meta data and its response for RabbitMQ.  Lke a `Plug.Conn`, but for RPC.

  Any pure functions that write back to the `Retort.Server.Generic.RPC.t` can be in this module. Any
  code with side-effects, such as `Logger` or `Repo` calls should go in `Retort.Server.Generic`.
  """

  alias Alembic.Document
  alias Retort.{Api, Response, Request}
  alias Retort.Server.Generic

  @dialyzer {:no_match, decode_request: 1}

  # Struct

  defstruct assigns: %{},
            halted: false,
            meta: nil,
            request: %{
              decoded: nil,
              encoded: nil
            },
            response: %{
              decoded: nil,
              encoded: nil
            },
            timing: %{
              ending_monotonic_time: nil,
              starting_monotonic_time: nil
            }

  # Types

  @type t :: %__MODULE__{
               assigns: %{
                 optional(atom) => any
               },
               halted: boolean,
               meta: %{correlation_id: binary, delivery_tag: term, reply_to: term} | nil,
               request: %{
                 decoded: Request.t | nil,
                 encoded: binary
               },
               response: %{
                 decoded: Response.t | nil,
                 encoded: binary | nil
               },
               timing: %{
                 ending_monotonic_time: integer | nil,
                 starting_monotonic_time: integer | nil
               }
             }

  # Functions

  @doc """
  Adds `key` to `assigns` with `value`
  """
  @spec assign(t, key :: any, value :: any) :: t
  def assign(rpc, key, value) do
    update_in rpc.assigns, &Map.put(&1, key, value)
  end

  @doc """
  Extracts an `Alembic.Document.t` from `request.decoded.params` and assigns it to `assigns.document`.

  If there is an error extracting the document, the `response.decoded` is set to an error `Alembic.Document.t` and the
  `t` is halted.
  """
  @spec assign_document(t, atom) :: t
  def assign_document(
        rpc = %__MODULE__{
          request: %{
            decoded: %Request{
              params: params
            }
          }
        },
        action
      ) do
    case Document.from_json(
           params,
           %Alembic.Error{
             meta: %{
               "action" => action,
               "sender" => :client
             },
             source: %Alembic.Source{
               pointer: ""
             }
           }
         ) do
      {:ok, document} ->
        assign(rpc, :document, document)
      {:error, document} ->
        halt(rpc, &Response.error(&1, document))
    end
  end

  @doc """
  Decodes the `request.encoded` to an `Retort.Request` and puts it in `request.decoded`.

  If there is an error, the `response.decoded` is set to an error `Alembic.Document.t` and the `t` is halted.
  """
  @spec decode_request(t) :: t
  def decode_request(
        rpc = %__MODULE__{
          request: %{
            encoded: encoded
          }
        }
      ) do
    case Request.from_string(encoded) do
      {:ok, decoded} ->
        put_in rpc.request.decoded, decoded
      # Poison 2.0
      {:error, :invalid} ->
        halt(rpc, fn _ -> Response.error("Invalid Request") end)
      {:error, {:invalid, token}} when is_binary(token) ->
        halt(rpc, fn _ -> Response.error("Parse error") end)
      # Poison 3.0
      {:error, :invalid, position} when is_integer(position) ->
        halt(rpc, fn _ -> Response.error("Parse error") end)
      {:error, {:invalid, token, position}} when is_binary(token) and is_integer(position) ->
        halt(rpc, fn _ -> Response.error("Parse error") end)
    end
  end

  @doc """
  Makes sure that the keys in the `request.decoded.params` are underscored as required by
  [`Alembic`](https://hexdocs.pm/alembic)
  """
  @spec format_request_params(t) :: t

  def format_request_params(rpc = %__MODULE__{halted: true}), do: rpc

  def format_request_params(
        rpc = %__MODULE__{
          request: %{
            decoded: %Request{}
          }
        }
      ) do
    update_in rpc.request.decoded.params, &Api.deserialize_keys/1
  end

  @doc """
  Converts the `:basic_deliver` arguments passed to `Retort.Server.Generic.handle_info/2` by `AMQP` to
  a `t`.
  """
  @spec from_basic_deliver({:basic_deliver, binary, map}, Generic.t) :: t
  def from_basic_deliver({:basic_deliver, encoded_request, meta}, _) do
    rpc = %__MODULE__{}
    encoded_request_rpc = put_in rpc.request.encoded, encoded_request
    put_in encoded_request_rpc.meta, meta
  end

  @doc """
  Halts the `rpc` after setting the `response.decoded` to the output of `function`
  """
  @spec halt(t, (Response.t -> Response.t)) :: t
  def halt(rpc, function) do
    rpc = update_in rpc.response.decoded, function
    %{rpc | halted: true}
  end

  @doc """
  Checks that the `response.decoded.id` matches the `meta.correlation_id`
  """

  @spec match_correlation_id(%__MODULE__{halted: true}) :: t
  def match_correlation_id(rpc = %__MODULE__{halted: true}), do: rpc

  @spec match_correlation_id(
          %__MODULE__{
            halted: false,
            meta: %{
              required(:correlation_id) => binary,
              optional(any) => any
            },
            request: %{
              required(:decoded) => Response.t
            }
          }
        ) :: t
  def match_correlation_id(
        rpc = %__MODULE__{
          meta: %{
            correlation_id: correlation_id
          },
          request: %{
            decoded: decoded
          }
        }
      ) do
    case decoded do
      %Request{id: ^correlation_id} ->
        rpc
      %Request{id: id} ->
        halt rpc, fn decoded ->
          %Response{
            decoded |
            error: %Response.Error{
              code: -31_999,
              data: %{
                correlation_id: correlation_id,
                id: id
              },
              message: "JSON RPC id does not match RabbitMQ correlation_id"
            }
          }
        end
    end
  end

  @doc """
  Converts the `request.decoded` to a `response.decoded` with the same `id`
  """
  @spec request_to_response(t) :: t

  def request_to_response(rpc = %__MODULE__{halted: true}), do: rpc

  def request_to_response(
        rpc = %__MODULE__{
          request: %{
            decoded: request = %Request{id: _}
          }
        }
      ) do
    put_in rpc.response.decoded, Request.to_response(request)
  end

  @doc """
  Times duration of `function` and puts that timing information in `timing` section of `t`

  The `timing.starting_monotonic_time` will be set on the `t` passed to `function`.  The `t` returned by `function` will
  have `timing.ending_monotonic_time` set on it before being returned.
  """
  @spec time(t, (t -> t)) :: t
  def time(
        rpc = %__MODULE__{
          timing: %{
            ending_monotonic_time: nil,
            starting_monotonic_time: nil
          }
        },
        function
      ) do
    timed_rpc = put_in rpc.timing.starting_monotonic_time, :erlang.monotonic_time

    try do
      function.(timed_rpc)
    else
      function_rpc ->
        put_in function_rpc.timing.ending_monotonic_time, :erlang.monotonic_time
    end
  end
end