retort

defmodule Retort.Server.Generic do
  @moduledoc """
  Exposes an `Ecto.Schema.t` as a JSONAPI resource over JSON RPC on a RabbitMQ queue.

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

  alias Retort.{Channel, Response}
  alias __MODULE__.{Callback, RPC}

  use GenServer

  require Logger

  # Struct

  defstruct callback: nil,
            channel: nil,
            queue: nil

  # Types

  @typedoc """
  * `callback` - The `module` that implement this behaviour and its own internal state.
  * `channel` - the AMQP channel connected to `queue`
  * `queue` - name of RabbitMQ queue to consume
  """
  @type t :: %__MODULE__{
               callback: Callback.t,
               channel: %AMQP.Channel{} | nil,
               queue: String.t
             }

  # Callbacks

  @doc """
  Should match on `rpc.decoded.method` and return rpc with `rpc.response.decoded.result` set.
  """
  @callback handle_method(rpc :: RPC.t, callback_state :: any) :: RPC.t

  @callback init(generic_state :: t, callback_state) :: {:ok, callback_state} |
                                                        {:ok, callback_state, timeout | :hibernate} |
                                                        :ignore |
                                                        {:stop, reason :: any} when callback_state: any

  # Functions

  ## Callback functions

  @doc """
  Attaches consumer to `queue` in `t`
  """
  @spec start_link(t, GenServer.options) :: GenServer.on_start
  def start_link(state, gen_server_options \\ []) do
    GenServer.start_link(__MODULE__, state, gen_server_options)
  end

  @doc """
  Sets connection, channel, and queue to Rabbit
  """
  @spec init(t) :: {:ok, t} | {:stop, any} | {:ok, t, :hibernate | :infinity | non_neg_integer}
  def init(
        state = %__MODULE__{
          callback: %{
            module: callback_module,
            state: callback_state
          },
          queue: queue
        }
      )
      when not is_nil(callback_module) and is_atom(callback_module) do
    with {:ok, channel} <- Channel.init(queue) do
      Logger.info fn ->
        ["Starting ", inspect(__MODULE__), " for queue (", inspect(queue), ")"]
      end

      Logger.metadata queue: queue

      state = %__MODULE__{state | channel: channel}

      case callback_module.init(state, callback_state) do
        {:ok, callback_state} ->
          {:ok, put_in(state.callback.state, callback_state)}
        {:ok, callback_state, timeout_or_hibernate} ->
          {:ok, put_in(state.callback.state, callback_state), timeout_or_hibernate}
        other ->
          other
      end
    end
  end

  @doc """
  # Sent by the broker when the consumer is unexpectedly cancelled
  # (such as after a queue deletion)
  """
  @spec handle_info({:basic_cancel, %{required(:consumer_tag) => any, optional(term) => term}}, t) ::
          {:stop, :normal, t}
  def handle_info({:basic_cancel, %{consumer_tag: _consumer_tag}}, state) do
    {:stop, :normal, state}
  end

  @doc """
  # Confirmation sent by the broker to the consumer process after a
  # Basic.cancel
  """
  @spec handle_info({:basic_cancel_ok, %{required(:consumer_tag) => any, optional(term) => term}}, t) :: {:noreply, t}
  def handle_info({:basic_cancel_ok, %{consumer_tag: _consumer_tag}}, state) do
    {:noreply, state}
  end

  @doc """
  # Confirmation sent by the broker after registering this process as a consumer
  """
  @spec handle_info({:basic_consume_ok, %{required(:consumer_tag) => any, optional(term) => term}}, t) :: {:noreply, t}
  def handle_info({:basic_consume_ok, %{consumer_tag: _consumer_tag}}, state) do
    {:noreply, state}
  end

  @doc """
  Most received messages filter through here to find the right
  function to call based on RPC method.
  """
  @spec handle_info({:basic_deliver, any, any}, t) :: {:noreply, t}
  def handle_info(message = {:basic_deliver, _, _}, state = %__MODULE__{}) do
    message
    |> RPC.from_basic_deliver(state)
    |> log_correlation_id(
         fn correlation_id_rpc ->
           correlation_id_rpc
           |> RPC.time(
                fn timed_rpc = %RPC{
                     request: %{
                       encoded: encoded_request
                     }
                   } ->
                  Logger.debug ["Receiving payload: ", encoded_request]

                  respond_or_reject(timed_rpc, state)
                end
              )
           |> __MODULE__.Logger.call
         end
       )

    {:noreply, state}
  end

  @spec handle_info({:DOWN, reference, :process, pid, any}, t) :: {:noreply, t}
  def handle_info(
        {:DOWN, _monitor_ref, :process, pid, reason},
        state = %__MODULE__{
          channel: %AMQP.Channel{
            pid: pid
          },
          queue: queue
        }
      ) do
    case reason do
      :normal ->
        Logger.info ["AMQP channel went down due to ", to_string(reason)]
      {:shutdown, {:connection_closing, {:internal_error, 541, internal_error_message}}} ->
        Logger.error [
          "AMQP channel shutdown due to connection closing because of internal error 541\n",
          "Details: ",
          internal_error_message
        ]
      _ ->
        Logger.error [
          "#{__MODULE__}.handle_info/2 received unexpected reason (#{inspect reason}) for AMQP channel ",
          "going down. You may want to add a new clause to handle this case."
        ]
    end

    with {:ok, new_channel} <- Channel.init(queue) do
      {:noreply, %__MODULE__{state | channel: new_channel}}
    end
  end

  def handle_info(
        {
          :DOWN,
          _monitor_ref,
          :process,
          pid,
          {:shutdown, {:connection_closing, {:internal_error, 541, internal_error_message}}}
        },
        state = %__MODULE__{
          channel: %AMQP.Channel{
            pid: pid
          },
          queue: queue
        }
      ) do
    Logger.error internal_error_message

    with {:ok, new_channel} <- Channel.init(queue) do
      {:noreply, %__MODULE__{state | channel: new_channel}}
    end
  end

  # handle_info/2 MUST have a catch all clause in case it receives unexpected system messages or sends
  def handle_info(info, state) do
    Logger.error "#{__MODULE__}.handle_info/2 received unexpected info #{inspect info}. " <>
                 "You may want to add a new clause to handle this case."
    {:noreply, state}
  end

  ## Private Functions

  @spec ack(RPC.t, t) :: RPC.t
  defp ack(
         rpc = %RPC{
           meta: %{
             delivery_tag: delivery_tag
           }
         },
         %__MODULE__{channel: channel}
       ) do
    AMQP.Basic.ack channel, delivery_tag

    rpc
  end

  @spec encoded_response(RPC.t) :: binary
  defp encoded_response(
         %RPC{
           response: %{
             decoded: decoded
           }
         }
       ) do
    case Poison.encode(decoded) do
      {:ok, encoded} ->
        encoded
      {:error, {:invalid, token}} when is_binary(token) ->
        Logger.error ["Encode error on token (", inspect(token), ") when encoding response (", inspect(decoded), ")"]
        effective_decoded = Response.error("Internal error")
        # use `encode!` because if this doesn't work, we can't make a response to publish
        Poison.encode!(effective_decoded)
    end
  end

  @spec log_correlation_id(RPC.t, (RPC.t -> RPC.t)) :: RPC.t
  defp log_correlation_id(
         rpc = %RPC{
           meta: %{
             correlation_id: correlation_id
           }
         },
         function
       ) do
    Logger.metadata(correlation_id: correlation_id)

    try do
      function.(rpc)
    after
      Logger.metadata(correlation_id: nil)
    end
  end

  @spec respond_or_reject(RPC.t, t) :: RPC.t
  defp respond_or_reject(
         rpc = %RPC{},
         state = %__MODULE__{
             callback: %Callback{
             module: callback_module,
             state: callback_state
           }
         }
       ) do
    rpc
    |> RPC.decode_request()
    |> RPC.request_to_response()
    |> RPC.format_request_params()
    |> RPC.match_correlation_id()
    |> callback_module.handle_method(callback_state)
    |> publish(state)
  rescue
    exception ->
      Logger.error(Exception.format(:error, exception))
      reject(rpc, state)
  else
    published_rpc ->
      ack(published_rpc, state)
  end

  @spec reject(RPC.t, t) :: RPC.t
  defp reject(
         rpc = %RPC{
           meta: %{
             delivery_tag: delivery_tag
           }
         },
         %__MODULE__{channel: channel}
       ) do
    AMQP.Basic.reject channel, delivery_tag, requeue: false

    rpc
  end

  @spec publish(RPC.t, t) :: RPC.t
  defp publish(
         rpc = %RPC{
           meta: %{
             correlation_id: correlation_id,
             reply_to: reply_to
           }
         },
         %__MODULE__{channel: channel}
       ) do
    encoded = encoded_response(rpc)
    rpc = put_in rpc.response.encoded, encoded

    Logger.debug ["Publishing response: ", encoded]
    AMQP.Basic.publish(channel, "", reply_to, encoded, correlation_id: correlation_id)

    rpc
  end
end