eventstore

if Code.ensure_loaded?(Swarm) do
defmodule EventStore.Registration.DistributedRegistry do
  @moduledoc """
  Process registration and distribution throughout a cluster of nodes using [Swarm](https://github.com/bitwalker/swarm)
  """

  @behaviour EventStore.Registration

  @doc """
  Return an optional supervisor spec for the registry
  """
  @spec child_spec() :: [:supervisor.child_spec()]
  @impl EventStore.Registration
  def child_spec, do: []

  @doc """
  Starts a uniquely named child process of a supervisor using the given module and args.

  Registers the pid with the given name.
  """
  @spec start_child(name :: term(), supervisor :: module(), args :: [any()]) :: {:ok, pid()} | {:error, reason :: term()}
  @impl EventStore.Registration
  def start_child(name, supervisor, args) do
    case whereis_name(name) do
      :undefined ->
        case Swarm.register_name(name, Supervisor, :start_child, [supervisor, args]) do
          {:error, {:already_registered, pid}} -> {:error, {:already_started, pid}}
          reply -> reply
        end

      pid ->
        {:ok, pid}
    end
  end

  @doc """
  Sends a message to the given dest running on the current node and each
  connected node, returning `:ok`.
  """
  @callback multi_send(dest :: atom(), message :: any()) :: :ok
  @impl EventStore.Registration
  def multi_send(dest, message) do
    for node <- nodes() do
      Process.send({dest, node}, message, [:noconnect])
    end

    :ok
  end

  @doc """
  Get the pid of a registered name.
  """
  @spec whereis_name(name :: term) :: pid | :undefined
  @impl EventStore.Registration
  def whereis_name(name), do: Swarm.whereis_name(name)

  @doc """
  Return a `:via` tuple to route a message to a process by its registered name
  """
  @spec via_tuple(name :: term()) :: {:via, module(), name :: term()}
  @impl EventStore.Registration
  def via_tuple(name), do: {:via, :swarm, name}

  #
  # `GenServer` callback functions used by Swarm
  #

  @doc false
  def handle_call({:swarm, :begin_handoff}, _from, state) do
    # Stop the process when a cluster toplogy change indicates it is now running
    # on the wrong host. This is to prevent a spike in process restarts as they
    # are moved. Instead, allow the process to be started on request.
    {:stop, :shutdown, :ignore, state}
  end

  @doc false
  def handle_call(_request, _from, _state) do
    raise "attempted to call GenServer #{inspect proc()} but no handle_call/3 clause was provided"
  end

  def handle_cast({:swarm, :end_handoff, _state}, state) do
    {:noreply, state}
  end

  @doc false
  def handle_cast({:swarm, :resolve_conflict, state}, _state) do
    {:noreply, state}
  end

  @doc false
  def handle_cast(_request, _state) do
    raise "attempted to cast GenServer #{inspect proc()} but no handle_cast/2 clause was provided"
  end

  @doc false
  def handle_info({:swarm, :die}, state) do
    # Stop the process as it is being moved to another node, or there are not
    # currently enough nodes running
    {:stop, :shutdown, state}
  end

  defp nodes, do: [Node.self() | Node.list()]

  defp proc do
    case Process.info(self(), :registered_name) do
      {_, []}   -> self()
      {_, name} -> name
    end
  end
end
end