riptide

defmodule Riptide.Processor do
  require Logger

  def process(commands, raw, from, state) when is_binary(raw) do
    msg = Jason.decode!(raw)
    process(commands, msg, from, state)
  end

  def process(commands, msg, from, state) do
    key = msg["key"]
    action = msg["action"]
    type = msg["type"] || action
    body = msg["body"]

    case type do
      "reply" ->
        GenServer.reply(state.pending[key], {:ok, body})
        {:noreply, Dynamic.delete(state, [:pending, key])}

      "error" ->
        GenServer.reply(state.pending[key], {:error, body})
        {:noreply, Dynamic.delete(state, [:pending, key])}

      result when result in ["call", "cast"] ->
        fun =
          case type do
            "call" -> :handle_call
            "cast" -> :handle_cast
          end

        process_commands(commands, %{action: action, body: body}, fun, from, state)
        |> case do
          nil -> {:error, :invalid_command, state}
          result -> result
        end
    end
    |> process_response(key, state)
  end

  def process_commands(commands, msg, fun, from, state) do
    [Riptide.Command.Echo | commands]
    |> Stream.map(&safe_trigger(&1, fun, msg, from, Dynamic.get(state, [:data])))
    |> Stream.filter(fn result -> result !== nil end)
    |> Enum.at(0)
  end

  @spec process_response(
          {:noreply, any} | {:error, any, any} | {:reply, any, any} | Task.t(),
          any,
          any
        ) :: {:noreply, any} | {:reply, binary, any}
  def process_response(response, key, state) do
    case response do
      nil ->
        {:noreply, state}

      %Task{ref: ref} ->
        {:noreply, Dynamic.put(state, [:tasks, ref], key)}

      {:noreply, data} ->
        {:noreply, Dynamic.put(state, [:data], data)}

      {:reply, action, value, data} ->
        {:reply, build_reply(nil, action, value), Dynamic.put(state, [:data], data)}

      {:reply, value, data} ->
        {:reply, build_reply(key, "reply", value), Dynamic.put(state, [:data], data)}

      {:error, error, data} ->
        {:reply, build_reply(key, "error", error), Dynamic.put(state, [:data], data)}
    end
  end

  def build_reply(key, type, body) do
    %{
      key: key,
      type: type,
      action: type,
      body: body
    }
    |> Jason.encode!()
  end

  def build_call(pid, action, body, state) do
    key = Dynamic.get(state, [:count], 0)

    {
      %{
        key: key,
        type: "call",
        action: action,
        body: body
      }
      |> Jason.encode!(),
      state
      |> Dynamic.put([:count], key)
      |> Dynamic.put([:pending, key], pid)
    }
  end

  def safe_trigger(mod, fun, msg, from, state) do
    try do
      apply(mod, fun, [msg, from, state])
    rescue
      e ->
        :error
        |> Exception.format(e, __STACKTRACE__)
        |> Logger.error(crash_reason: {e, __STACKTRACE__})

        {:error, e, state}
    catch
      _, e ->
        :throw
        |> Exception.format(e, __STACKTRACE__)
        |> Logger.error(crash_reason: {e, __STACKTRACE__})

        {:error, e, state}
    end
  end
end