eventstore

defmodule EventStore.MonitoredServer do
  @moduledoc false

  # Starts a `GenServer` process using a given module-fun-args tuple. Monitors
  # the started process and attempts to restart it on terminate using an
  # exponential backoff strategy. Allows interested processes to be informed
  # when the process terminates.

  use GenServer

  require Logger

  alias DBConnection.Backoff

  defmodule State do
    @moduledoc false

    defstruct [:mfa, :after_exit, :after_restart, :backoff, :pid, :shutdown, :queue]
  end

  def start_link([{_module, _fun, _args} = mfa, opts]) do
    {start_opts, monitor_opts} = Keyword.split(opts, [:name, :timeout, :debug, :spawn_opt])

    GenServer.start_link(__MODULE__, [mfa, monitor_opts], start_opts)
  end

  def init([mfa, opts]) do
    Process.flag(:trap_exit, true)

    state = %State{
      after_exit: Keyword.get(opts, :after_exit),
      after_restart: Keyword.get(opts, :after_restart),
      backoff: Backoff.new(backoff_type: :exp),
      mfa: mfa,
      queue: :queue.new(),
      shutdown: Keyword.get(opts, :shutdown, 100)
    }

    {:ok, start_process(:start, state)}
  end

  def handle_call(msg, from, %State{pid: nil} = state) do
    {:noreply, enqueue({:call, msg, from}, state)}
  end

  def handle_call(msg, from, %State{pid: pid} = state) do
    forward_call(pid, msg, from)

    {:noreply, state}
  end

  def handle_cast(msg, %State{pid: nil} = state) do
    {:noreply, enqueue({:cast, msg}, state)}
  end

  def handle_cast(msg, %State{pid: pid} = state) do
    forward_cast(pid, msg)

    {:noreply, state}
  end

  def handle_info(:start_process, %State{} = state) do
    {:noreply, start_process(:restart, state)}
  end

  @doc """
  Handle process terminate by attempting to restart, after a delay.
  """
  def handle_info({:EXIT, pid, reason}, %State{pid: pid} = state) do
    Logger.debug(fn -> "Monitored process EXIT due to: #{inspect(reason)}" end)

    after_callback(:exit, state)

    state = %State{state | pid: nil}

    {:noreply, delayed_start(state)}
  end

  def handle_info({:EXIT, _pid, reason}, %State{} = state) do
    Logger.debug(fn -> "Monitored process EXIT due to: #{inspect(reason)}" end)

    {:noreply, state}
  end

  def handle_info(msg, %State{pid: nil} = state) do
    {:noreply, enqueue({:info, msg}, state)}
  end

  def handle_info(msg, %State{pid: pid} = state) do
    forward_info(pid, msg)

    {:noreply, state}
  end

  def terminate(_, %State{pid: nil}), do: :ok

  def terminate(reason, %State{} = state) do
    %State{pid: pid, shutdown: shutdown, mfa: {module, _fun, _args}} = state

    Logger.debug(fn ->
      "Monitored server #{inspect(module)} terminate due to: #{inspect(reason)}"
    end)

    Process.exit(pid, reason)

    receive do
      {:EXIT, ^pid, _} -> :ok
    after
      shutdown ->
        Logger.warn(
          "Monitored server #{inspect(module)} failed to terminate within #{shutdown}, killing it brutally"
        )

        Process.exit(pid, :kill)

        receive do
          {:EXIT, ^pid, _} -> :ok
        end
    end
  end

  # Attempt to start the process, retry after a delay on failure
  defp start_process(start_type, %State{} = state) do
    %State{mfa: {module, fun, args}, queue: queue} = state

    Logger.debug(fn -> "Attempting to start #{inspect(module)}" end)

    case apply(module, fun, args) do
      {:ok, pid} ->
        Logger.debug(fn -> "Successfully started #{inspect(module)} (#{inspect(pid)})" end)

        :ok = forward_queued_msgs(pid, queue)

        after_callback(start_type, state)

        %State{state | pid: pid, queue: :queue.new()}

      {:error, reason} ->
        Logger.info(fn -> "Failed to start #{inspect(module)} due to: #{inspect(reason)}" end)

        delayed_start(state)
    end
  end

  defp enqueue(item, %State{queue: queue} = state) do
    %State{state | queue: :queue.in(item, queue)}
  end

  defp forward_call(pid, msg, from) do
    :erlang.send(pid, {:"$gen_call", from, msg}, [:noconnect])
  end

  defp forward_cast(pid, msg) do
    :erlang.send(pid, {:"$gen_cast", msg}, [:noconnect])
  end

  defp forward_info(pid, msg) do
    :erlang.send(pid, msg, [:noconnect])
  end

  defp forward_queued_msgs(pid, queue) do
    case :queue.out(queue) do
      {{:value, item}, new_queue} ->
        forward_queued_msg(pid, item)
        forward_queued_msgs(pid, new_queue)

      {:empty, _new_queue} ->
        :ok
    end
  end

  defp forward_queued_msg(pid, {:call, msg, from}), do: forward_call(pid, msg, from)

  defp forward_queued_msg(pid, {:cast, msg}), do: forward_cast(pid, msg)

  defp forward_queued_msg(pid, {:info, msg}), do: forward_info(pid, msg)

  # Invoke `after_restart/0` callback function
  defp after_callback(:restart, %State{after_restart: after_restart})
       when is_function(after_restart, 0) do
    Task.start(after_restart)
  end

  # Invoke `after_exit/0` callback function
  defp after_callback(:exit, %State{after_exit: after_exit})
       when is_function(after_exit, 0) do
    Task.start(after_exit)
  end

  defp after_callback(_type, _state), do: :ok

  defp delayed_start(%State{backoff: backoff} = state) do
    {delay, backoff} = Backoff.backoff(backoff)

    Process.send_after(self(), :start_process, delay)

    %State{state | backoff: backoff}
  end
end