eidetic_projector

defmodule Eidetic.Projector do
  @moduledoc """
  Eidetic Projector provides a middleware system for consuming events and building
  projections.

  To get started, you'll need to start the Projector within your supervisor:

  ```elixir
  [
    # The second parameter is your options, documented below.
    {Eidetic.Projector, []}
  ]
  ```

  You can pass in pre and post-processors, which implement the
  `Eidetic.Projector.Middleware` behaviour.

  ```elixir
  [
    {Eidetic.Projector, [
      deserializer: MyDeserialize,
      preprocess: [
        Some.PreProcess.Middleware
      ],
      postprocess: [
        Some.PostProcess.Middleware
      ]
    ]}
  ]
  ```

  Other options accepted are:

  ```elixir
  {Eidetic.Projector, [
    name: :some_other_name
  ]}
  ```

  """
  use GenServer

  alias Eidetic.Event
  alias Eidetic.Projector.Bundle
  alias Eidetic.Projector.Bundle.ConsumerMeta
  require Logger

  @doc """
  Got to start somewhere
  """
  def start_link(options \\ []) do
    name = Keyword.get(options, :name, __MODULE__)

    _ = Logger.info(fn -> "Starting Eidetic Projector with name #{name}" end)
    GenServer.start_link(__MODULE__, options, name: name)
  end

  @doc """
  Convert the options passed through `start_link/1` into our config map
  """
  @impl GenServer
  def init(options) do
    deserializer = Keyword.get(options, :deserializer, nil)
    preprocess = Keyword.get(options, :preprocess, [])
    postprocess = Keyword.get(options, :postprocess, [])

    config = %{
      deserializer: deserializer,
      preprocess: preprocess,
      postprocess: postprocess
    }

    _ = Logger.info(fn -> "Eidetic Projector Configuration: #{inspect(config)}" end)
    {:ok, config}
  end

  @doc """
  Consumer's pass through their `ConsumerMeta` information and the received
  payload.

  It gets deserialized and passed through the preprocess middleware,
  hoping that a handle is configured.

  If no handler, exit.

  If a handler is established, send the bundle to the handler and run the
  post-process middleware.
  """
  def consume(consumer_meta = %ConsumerMeta{}, payload) do
    GenServer.call(__MODULE__, {:consume, consumer_meta, payload}, 120_000)
  end

  @doc """
  GenServer Endpoint for `consume/1`
  """
  @impl GenServer
  def handle_call({:consume, consumer_meta = %ConsumerMeta{}, payload}, _from, config) do
    event = %Event{} = deserialize(config[:deserializer], payload)

    result =
      %Bundle{
        consumer_meta: consumer_meta,
        event: event
      }
      |> preprocess(config[:preprocess])
      |> handler()
      |> postprocess(config[:postprocess])

    {:reply, result, config}
  end

  defp preprocess(bundle = %Bundle{}, middleware) do
    process(:pre, bundle, middleware)
  end

  defp postprocess({:ok, bundle = %Bundle{}}, middleware) do
    process(:post, bundle, middleware)
  end

  defp postprocess({:skip, bundle = %Bundle{}}, _middleware) do
    {:skip, bundle}
  end

  defp postprocess({:error, error}, _middleware) do
    {:error, error}
  end

  defp handler({:ok, bundle = %Bundle{handler: handler}}) when is_function(handler) do
    bundle.handler.(bundle)
  end

  defp handler({:ok, bundle = %Bundle{handler: nil}}) do
    {:ok, bundle}
  end

  defp handler({:ok, bundle = %Bundle{handler: handler}}) when is_atom(handler) do
    bundle.handler.handle(bundle)
  end

  defp handler({:skip, bundle = %Bundle{}}) do
    {:skip, bundle}
  end

  defp handler({:error, error}) do
    {:error, error}
  end

  defp process(stage, bundle = %Bundle{}, [middleware | tail]) do
    use Retry

    _ = Logger.debug(fn -> "Processing Middleware: #{inspect(middleware)}" end)

    result =
      retry with: exp_backoff() |> randomize() |> expiry(6_000),
            atoms: [:retry] do
        middleware.process(stage, bundle)
      after
        result ->
          result
      else
        _ ->
          {:error, bundle}
      end

    case result do
      {:ok, updated_bundle} ->
        process(stage, updated_bundle, tail)

      {:skip, updated_bundle} ->
        {:skip, updated_bundle}

      {:error, updated_bundle} ->
        {:error, updated_bundle}
    end
  end

  defp process(_, %Bundle{} = bundle, []) do
    {:ok, bundle}
  end

  defp deserialize(deserializer, payload) when is_function(deserializer) do
    deserializer.(payload)
  end

  defp deserialize(deserializer, payload) when is_atom(deserializer) do
    deserializer.deserialize(payload)
  end
end