cachex

defmodule Cachex.Router do
  @moduledoc """
  Module controlling routing behaviour definitions.

  This module defines the router implementations for Cachex, allowing the user
  to route commands between nodes in a cache cluster. This means that users
  can provide their own routing and rebalancing logic without having to depend
  on it being included in Cachex.
  """
  alias Cachex.Router
  alias Cachex.Services

  # add some service aliases
  alias Services.Informant
  alias Services.Overseer

  # import macro stuff
  import Cachex.Error
  import Cachex.Spec

  #############
  # Behaviour #
  #############

  @doc """
  Initialize a routing state for a cache.

  Please see all child implementations for supported options.
  """
  @callback init(cache :: Cachex.t(), options :: Keyword.t()) :: any

  @doc """
  Retrieve the list of nodes from a routing state.
  """
  @callback nodes(state :: any) :: [atom]

  @doc """
  Route a key to a node in a routing state.
  """
  @callback route(state :: any, key :: any) :: atom

  @doc """
  Create a child specification to back a routing state.
  """
  @callback children(cache :: Cachex.t(), options :: Keyword.t()) ::
              [Supervisor.child_spec()]

  ##################
  # Implementation #
  ##################

  @doc false
  defmacro __using__(_) do
    quote location: :keep, generated: true do
      @behaviour Cachex.Router

      @doc false
      def init(cache, options \\ []),
        do: nil

      @doc false
      def children(cache, options),
        do: []

      # state modifiers are overridable
      defoverridable init: 2, children: 2
    end
  end

  ##############
  # Public API #
  ##############

  @doc """
  Retrieve all currently connected nodes (including this one).
  """
  @spec connected() :: [atom]
  def connected(),
    do: [node() | :erlang.nodes(:connected)]

  @doc """
  Retrieve all routable nodes for a cache.
  """
  @spec nodes(cache :: Cachex.t()) :: {:ok, [atom]}
  def nodes(cache(router: router(module: module, state: state))),
    do: {:ok, module.nodes(state)}

  @doc """
  Executes a previously dispatched action..
  """
  # The first match short circuits local-only caches
  @spec route(Cachex.t(), atom, {atom, [any]}) :: any
  def route(cache(router: router(module: Router.Local)) = cache, module, call),
    do: route_local(cache, module, call)

  def route(cache() = cache, module, call),
    do: route_cluster(cache, module, call)

  @doc """
  Dispatches a call to an appropriate execution environment.

  This acts as a macro just to avoid the overhead of slicing up module
  names at runtime, when they can be guaranteed at compile time much
  more easily.
  """
  defmacro route(cache, {action, _arguments} = call) do
    # coveralls-ignore-start
    act_name =
      action
      |> Kernel.to_string()
      |> String.replace_trailing("?", "")
      |> Macro.camelize()

    act_join = :"Elixir.Cachex.Actions.#{act_name}"
    # coveralls-ignore-stop

    quote do
      Overseer.with(unquote(cache), fn cache ->
        call = unquote(call)
        module = unquote(act_join)

        Router.route(cache, module, call)
      end)
    end
  end

  ###############
  # Private API #
  ###############

  # Results merging for distributed cache results.
  #
  # Follows these rules:
  #
  # - Lists are always concatenated.
  # - Numbers are always summed.
  # - Booleans are always AND-ed.
  # - Maps are always merged (recursively).
  #
  # This has to be public due to scopes, but we hide the docs
  # because we don't really care for anybody else calling it.
  defp result_merge(left, right) when is_list(left),
    do: left ++ right

  defp result_merge(left, right) when is_number(left),
    do: left + right

  defp result_merge(left, right) when is_boolean(left),
    do: left && right

  # coveralls-ignore-start
  defp result_merge(left, right) when is_map(left) do
    Map.merge(left, right, fn _, left, right ->
      result_merge(left, right)
    end)
  end

  # coveralls-ignore-stop

  # Provides handling for local actions on this node.
  #
  # This will provide handling of notifications across hooks before and after
  # the execution of an action. This is taken from code formerly in the old
  # `Cachex.Actions` module, but has been moved here as it's more appropriate.
  #
  # If `notify` is set to false, notifications are disabled and the call is
  # simply executed as is. If `via` is provided, you can override the handle
  # passed to the hooks (useful for re-use of functions). An example of this
  # is `decr/4` which simply calls `incr/4` with `via: { :decr, arguments }`.
  defp route_local(cache, module, {_action, arguments} = call) do
    option = List.last(arguments)
    notify = Keyword.get(option, :notify, true)

    message =
      notify &&
        case option[:via] do
          msg when not is_tuple(msg) -> call
          msg -> msg
        end

    notify && Informant.broadcast(cache, message)
    result = apply(module, :execute, [cache | arguments])

    if notify do
      Informant.broadcast(
        cache,
        message,
        Keyword.get(option, :result, result)
      )
    end

    result
  end

  # actions based on a key
  @keyed_actions [
    :del,
    :exists?,
    :expire,
    :fetch,
    :get,
    :get_and_update,
    :incr,
    :invoke,
    :put,
    :refresh,
    :take,
    :touch,
    :ttl,
    :update
  ]

  # Provides handling to key-based actions distributed to remote nodes.
  #
  # The algorithm here is simple; hash the key and slot the value using JCH into
  # the total number of slots available (i.e. the count of the nodes). If it comes
  # out to the local node, just execute the local code, otherwise RPC the base call
  # to the remote node, and just assume that it'll correctly handle it.
  defp route_cluster(cache, module, {action, [key | _]} = call)
       when action in @keyed_actions do
    cache(router: router(module: router, state: nodes)) = cache
    route_node(cache, module, call, router.route(nodes, key))
  end

  # actions which merge outputs
  @merge_actions [
    :clear,
    :count,
    :empty?,
    :export,
    :import,
    :keys,
    :purge,
    :reset,
    :size
  ]

  # Provides handling of cross-node actions distributed over remote nodes.
  #
  # This will do an RPC call across all nodes to fetch their results and merge
  # them with the results on the local node. The hooks will only be notified
  # on the local node, due to an annoying recursion issue when handling the
  # same across all nodes - seems to provide better logic though.
  defp route_cluster(cache, module, {action, arguments} = call)
       when action in @merge_actions do
    # fetch the nodes from the cluster state
    cache(router: router(module: router, state: state)) = cache

    # all calls have options we can use
    options = List.last(arguments)

    # can force local node setting local: true
    results =
      case Keyword.get(options, :local) do
        true ->
          []

        _any ->
          # don't want to execute on the local node
          other_nodes =
            state
            |> router.nodes()
            |> List.delete(node())

          # execute the call on all other nodes
          {results, _} =
            :rpc.multicall(
              other_nodes,
              module,
              :execute,
              [cache | arguments]
            )

          results
      end

    # execution on the local node, using the local macros and then unpack
    {:ok, result} = route_local(cache, module, call)

    # results merge
    merge_result =
      results
      |> Enum.map(&elem(&1, 1))
      |> Enum.reduce(result, &result_merge/2)

    # return after merge
    {:ok, merge_result}
  end

  # actions which always run locally
  @local_actions [
    :inspect,
    :prune,
    :restore,
    :save,
    :stats,
    :warm
  ]

  # Provides handling of `:inspect` operations.
  #
  # These operations are guaranteed to run on the local nodes.
  defp route_cluster(cache, module, {action, _arguments} = call)
       when action in @local_actions,
       do: route_local(cache, module, call)

  # Provides handling of `:put_many` operations.
  #
  # These operations can only execute if their keys slot to the same remote nodes.
  defp route_cluster(cache, module, {:put_many, _arguments} = call),
    do: route_batch(cache, module, call, &elem(&1, 0))

  # Provides handling of `:transaction` operations.
  #
  # These operations can only execute if their keys slot to the same remote nodes.
  defp route_cluster(cache, module, {:transaction, [[] | _]} = call),
    do: route_local(cache, module, call)

  defp route_cluster(cache, module, {:transaction, [_keys | _]} = call),
    do: route_batch(cache, module, call, & &1)

  # Any other actions are only available with local: true in the call
  defp route_cluster(cache, module, {_action, arguments} = call) do
    # all calls have options we can use
    options = List.last(arguments)

    # can force local node setting local: true
    case Keyword.get(options, :local) do
      true -> route_local(cache, module, call)
      _any -> error(:non_distributed)
    end
  end

  # coveralls-ignore-start
  # Catch-all just in case we missed something...
  defp route_cluster(_cache, _module, _call),
    do: error(:non_distributed)

  # coveralls-ignore-stop

  # Calls a slot for the provided cache action if all keys slot to the same node.
  #
  # This is a delegate handler for `route_node/4`, but ensures that all keys slot to the
  # same node to avoid the case where we have to fork a call out internally.
  defp route_batch(cache, module, {_action, [keys | _]} = call, mapper) do
    # map all keys to a slot in the nodes list
    cache(router: router(module: router, state: state)) = cache
    slots = Enum.map(keys, &router.route(state, mapper.(&1)))

    # unique to avoid dups
    case Enum.uniq(slots) do
      # if there's a single slot it's safe to continue with the call to the remote
      [slot] ->
        route_node(cache, module, call, slot)

      # otherwise, cross_slot errors!
      _disable ->
        error(:cross_slot)
    end
  end

  # Calls a node for the provided cache action.
  #
  # This will determine a local slot and delegate locally if so, bypassing
  # any RPC calls in order to gain a slight bit of performance.
  defp route_node(cache, module, {action, arguments} = call, node) do
    current = node()
    cache(name: name) = cache

    case node do
      ^current ->
        route_local(cache, module, call)

      targeted ->
        result =
          :rpc.call(
            targeted,
            Cachex,
            action,
            [name | arguments]
          )

        with {:badrpc, reason} <- result do
          {:error, reason}
        end
    end
  end
end