eidetic_projector

defmodule Eidetic.Projector.Middleware.MongoDbIdempotence do
  @moduledoc """
  Idempotence middleware for Eidetic Projector.

  This middleware will store records in mongoDB whenever a message has been
  processed by the consumer. It will pass a `replay` modifier to the handler
  for each event, which is `true` if it's been processed by this consumer before;
  `false` otherwise.
  """
  alias Eidetic.Projector.Bundle
  alias Eidetic.Projector.Middleware
  use GenServer

  require Logger

  @behaviour Middleware

  @spec start_link(any()) :: {:ok, pid()}
  def start_link(default) do
    Logger.debug fn() -> "Starting MongoDbIdempotence as ${__MODULE__} ... #{inspect default}" end

    GenServer.start_link(__MODULE__, default, name: __MODULE__)
  end

  @spec init(mongodb: atom() | pid(), collection: binary()) ::
          {:ok, %{connection: atom() | pid(), collection: binary()}}

  @impl GenServer
  def init(mongodb: mongodb, collection: collection) do
    Logger.debug fn() -> "Initialising MongoDbIdempotence ... #{mongodb} and #{collection}" end

    {:ok, %{connection: mongodb, collection: collection}}
  end

  @impl Middleware
  @doc """
  Preprocess Implementation:

  Check if this event has been seen by our consumer before. If it has, update
  modifiers to container a `replay` property, which will be `true` if it's been
  processed before; `false` if not.
  """
  def process(:pre, bundle = %Bundle{}) do
    GenServer.call(__MODULE__, {:check, bundle})
  end

  @impl Middleware
  @doc """
  Postprocess Implementation:

  If needed, update mongoDB
  """
  def process(:post, bundle = %Bundle{}) do
    GenServer.call(__MODULE__, {:persist, bundle})
  end

  @impl GenServer
  def handle_call(
        {:check, bundle = %Bundle{event: event}},
        _from,
        state = %{connection: connection, collection: collection}
      ) do
    updated_bundle =
      case Mongo.find_one(connection, collection, %{
             "$and" => [%{identifier: event.identifier}, %{serial_number: event.serial_number}]
           }) do
        nil ->
          # Update Modifiers: replay=false
          Map.put(bundle, :modifiers, Map.put(bundle.modifiers, :replay, false))

        _ ->
          # Update Modifiers: replay=true
          Map.put(bundle, :modifiers, Map.put(bundle.modifiers, :replay, true))
      end

    {:reply, {:ok, updated_bundle}, state}
  end

  @impl GenServer
  def handle_call(
        {:persist, bundle = %Bundle{event: event}},
        _from,
        state = %{connection: connection, collection: collection}
      ) do
    # We will add `{:retry, _}` returns that allow this to be retried.
    # https://github.com/safwank/ElixirRetry/issues/21
    _ =
      Mongo.insert_one(connection, collection, %{
        identifier: event.identifier,
        serial_number: event.serial_number
      })

    {:reply, {:ok, bundle}, state}
  end
end