nebulex

defmodule Nebulex.Cache.Cluster do
  # The module used by cache adapters for
  # distributed caching functionality.
  @moduledoc false

  @doc """
  Joins the node where the cache `name`'s supervisor process is running to the
  `name`'s node group.
  """
  @spec join(name :: atom) :: :ok
  def join(name) do
    pid = Process.whereis(name) || self()

    if pid in pg_members(name) do
      :ok
    else
      :ok = pg_join(name, pid)
    end
  end

  @doc """
  Makes the node where the cache `name`'s supervisor process is running, leave
  the `name`'s node group.
  """
  @spec leave(name :: atom) :: :ok
  def leave(name) do
    pg_leave(name, Process.whereis(name) || self())
  end

  @doc """
  Returns the list of nodes joined to given `name`'s node group.
  """
  @spec get_nodes(name :: atom) :: [node]
  def get_nodes(name) do
    name
    |> pg_members()
    |> Enum.map(&node/1)
    |> :lists.usort()
  end

  @doc """
  Selects one node based on the computation of the `key` slot.
  """
  @spec get_node(name_or_nodes :: atom | [node], Nebulex.Cache.key(), keyslot :: module) :: node
  def get_node(name_or_nodes, key, keyslot)

  def get_node(name, key, keyslot) when is_atom(name) do
    name
    |> get_nodes()
    |> get_node(key, keyslot)
  end

  def get_node(nodes, key, keyslot) when is_list(nodes) do
    Enum.at(nodes, keyslot.hash_slot(key, length(nodes)))
  end

  ## PG

  if Code.ensure_loaded?(:pg) do
    defp pg_join(name, pid) do
      :ok = :pg.join(__MODULE__, name, pid)
    end

    defp pg_leave(name, pid) do
      _ = :pg.leave(__MODULE__, name, pid)
      :ok
    end

    defp pg_members(name) do
      :pg.get_members(__MODULE__, name)
    end
  else
    # Inline common instructions
    @compile {:inline, pg2_namespace: 1}

    defp pg_join(name, pid) do
      name
      |> ensure_namespace()
      |> :pg2.join(pid)
    end

    defp pg_leave(name, pid) do
      name
      |> ensure_namespace()
      |> :pg2.leave(pid)
    end

    defp pg_members(name) do
      name
      |> ensure_namespace()
      |> :pg2.get_members()
    end

    defp ensure_namespace(name) do
      namespace = pg2_namespace(name)
      :ok = :pg2.create(namespace)
      namespace
    end

    defp pg2_namespace(name), do: {:nbx, name}
  end
end