grains

defmodule Grains.GenGrain do
  use GenServer

  alias Grains.Bread

  @bread_key :bread
  @successors_key :successors
  @routes_key :routes
  @predecessors_key :predecessors
  @process_map_key :process_map
  @own_name_key :own_name

  require Logger

  @type from :: atom()
  @type state :: term()
  @type return :: timeout() | :hibernate | {:continue, term()}
  @type reply :: term()
  @type tag :: term()

  @doc """
  Handle a pull by a grain.

  When pulled by `pull/0,1`, this callback is triggered. The grain directly reply with
  a `:reply` tuple, ignore the pull with `:noreply`, or use `push/1,2` and `:noreply`.

  See `c:handle_pull/3` for handling `pull_with_tag/1,2`.
  """
  @callback handle_pull(from(), state()) ::
              {:reply, reply(), state()}
              | {:reply, reply(), state(), return()}
              | {:noreply, state()}
              | {:noreply, state(), return()}

  @doc """
  Handle a tagged pull by a grain.

  When pulled by `pull_with_tag/1,2`, this callback is triggered. The grain
  directly reply with a `:reply` tuple, ignore the pull with `:noreply`, or use
  `push/1,2` and `:noreply`.

  See `c:handle_pull/2` for handling `pull/0,1`.
  """
  @callback handle_pull(from(), tag(), state()) ::
              {:reply, reply(), state()}
              | {:reply, reply(), state(), return()}
              | {:noreply, state()}
              | {:noreply, state(), return()}

  @doc """
  Handle incoming data.

  This callback is triggered when any grain pushes data to this grain, using either
  `push/1` or `push/2`.
  """
  @callback handle_push(term(), from(), state()) ::
              {:noreply, state()} | {:noreply, state(), return()}

  @optional_callbacks [handle_pull: 2, handle_pull: 3, handle_push: 3]

  defmodule State do
    @moduledoc false
    @enforce_keys [:substate, :mod]
    defstruct @enforce_keys
  end

  defmacro __using__(_args) do
    quote do
      use GenServer

      @behaviour Grains.GenGrain

      defdelegate push(short_name, msg), to: Grains.GenGrain
      defdelegate push(msg), to: Grains.GenGrain
      defdelegate pull(from), to: Grains.GenGrain
      defdelegate pull_with_tag(from, tag), to: Grains.GenGrain
      defdelegate pull(), to: Grains.GenGrain
      defdelegate pull_with_tag(tag), to: Grains.GenGrain
      defdelegate own_name(), to: Grains.GenGrain
      defdelegate own_full_name(), to: Grains.GenGrain
    end
  end

  def start_link(mod, bread, name, args, opts) do
    GenServer.start_link(__MODULE__, {mod, bread, name, args}, opts)
  end

  def init({mod, bread, short_name, args}) do
    store(short_name, bread)

    case mod.init(args) do
      {:ok, substate} ->
        {:ok, %State{substate: substate, mod: mod}}

      {:ok, substate, ret} ->
        {:ok, %State{substate: substate, mod: mod}, ret}

      other ->
        other
    end
  end

  def handle_call({:debug, :reply_chain, sender, reply_chain, message}, _from, state) do
    {:reply, debug_reply_chain_forward(reply_chain, sender, message), state}
  end

  def handle_call({:debug, :automatic_reply_chain, last_process, sender, message}, _from, state) do
    name = Process.get(@own_name_key)
    bread = Process.get(@bread_key)
    paths = Bread.paths_between(bread, name, last_process)
    number_of_paths = length(paths)

    for path <- paths do
      :ok = debug_reply_chain_forward(path, sender, message)
    end

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

  def handle_call(msg, from, state = %State{mod: mod, substate: sub}) do
    mod.handle_call(msg, from, sub) |> handle_return(state)
  end

  def handle_cast(msg, state = %State{mod: mod, substate: sub}) do
    mod.handle_cast(msg, sub) |> handle_return(state)
  end

  def handle_info({:debug, :reply_chain, sender, reply_chain, message}, state) do
    debug_reply_chain_forward(reply_chain, sender, message)

    {:noreply, state}
  end

  def handle_info({:push, short_from, msg}, state = %State{mod: mod, substate: sub}) do
    case mod.handle_push(msg, short_from, sub) do
      {:noreply, sub} ->
        {:noreply, %State{state | substate: sub}}

      {:noreply, sub, ret} ->
        {:noreply, %State{state | substate: sub}, ret}
    end
  end

  def handle_info({:pull, short_from}, state = %State{mod: mod, substate: sub}) do
    mod.handle_pull(short_from, sub)
    |> handle_return(state)
    |> handle_pull_return(short_from)
  end

  def handle_info({:pull, tag, short_from}, state = %State{mod: mod, substate: sub}) do
    mod.handle_pull(short_from, tag, sub)
    |> handle_return(state)
    |> handle_pull_return(short_from)
  end

  def handle_info(msg, state = %State{mod: mod, substate: sub}) do
    mod.handle_info(msg, sub) |> handle_return(state)
  end

  defp debug_reply_chain_forward([last], sender, message) do
    case debug_reply_chain_check_current([last]) do
      :ok ->
        send(sender, message)

      e = {:error, _} ->
        Logger.error("Debug reply chain: #{inspect(e)}")
        e
    end
  end

  defp debug_reply_chain_forward(reply_chain = [_current, next | rest], sender, message) do
    with :ok <- debug_reply_chain_check_current(reply_chain),
         :ok <- send_with_error(next, {:debug, :reply_chain, sender, [next | rest], message}) do
      :ok
    else
      e = {:error, _} ->
        Logger.error("Debug reply chain: #{inspect(e)}")
    end
  end

  defp debug_reply_chain_check_current([current | _]) do
    own_name = own_name()
    own_full_name = own_full_name()

    if current in [own_name, own_full_name] do
      :ok
    else
      {:error, [reply_chain_current: current, expected_one_of: [own_name, own_full_name]]}
    end
  end

  def handle_continue(term, state = %State{mod: mod, substate: sub}) do
    mod.handle_continue(term, sub) |> handle_return(state)
  end

  def code_change(old_vsn, state = %State{mod: mod, substate: sub}, extra) do
    if Kernel.function_exported?(mod, :code_change, 3) do
      case mod.code_change(old_vsn, sub, extra) do
        {:ok, new_sub} ->
          {:ok, %State{state | substate: new_sub}}

        e = {:error, _} ->
          e
      end
    else
      {:ok, state}
    end
  end

  def format_status(reason, [pdict, %State{mod: mod, substate: sub}]) do
    if Kernel.function_exported?(mod, :format_status, 2) do
      mod.format_status(reason, [pdict, sub])
    else
      [{:data, [{'State', sub}]}]
    end
  end

  def terminate(reason, %State{mod: mod, substate: sub}) do
    if Kernel.function_exported?(mod, :terminate, 2) do
      mod.terminate(reason, sub)
    else
      :ok
    end
  end

  defp handle_return(return, state) do
    case return do
      {:noreply, sub} ->
        {:noreply, %State{state | substate: sub}}

      {:noreply, sub, ret} ->
        {:noreply, %State{state | substate: sub}, ret}

      {:reply, reply, sub} ->
        {:reply, reply, %State{state | substate: sub}}

      {:reply, reply, sub, ret} ->
        {:reply, reply, %State{state | substate: sub}, ret}
    end
  end

  defp handle_pull_return(return, short_from) do
    case return do
      {:reply, reply, state} ->
        push(short_from, reply)
        {:noreply, state}

      {:reply, reply, state, ret} ->
        push(short_from, reply)
        {:noreply, state, ret}

      other ->
        other
    end
  end

  def pull(pred) do
    case predecessors() do
      {:ok, predecessors} ->
        if Enum.member?(predecessors, pred) do
          send_pull(pred, own_name())
          :ok
        else
          {:error, {:push_to_unknown, pred}}
        end

      e = {:error, _} ->
        e
    end
  end

  def pull() do
    case predecessors() do
      {:ok, []} ->
        {:error, :no_predecessors}

      {:ok, preds} ->
        name = own_name()
        Enum.map(preds, fn s -> send_pull(s, name) end)
        :ok

      e = {:error, _} ->
        e
    end
  end

  def pull_with_tag(pred, tag) do
    case predecessors() do
      {:ok, predecessors} ->
        if Enum.member?(predecessors, pred) do
          send_pull(pred, own_name(), tag)
          :ok
        else
          {:error, {:push_to_unknown, pred}}
        end

      e = {:error, _} ->
        e
    end
  end

  def pull_with_tag(tag) do
    case predecessors() do
      {:ok, []} ->
        {:error, :no_predecessors}

      {:ok, preds} ->
        name = own_name()
        Enum.map(preds, fn s -> send_pull(s, name, tag) end)
        :ok

      e = {:error, _} ->
        e
    end
  end

  def push(short_name, msg) do
    case get_valid_successors(msg) do
      {:ok, successors} ->
        if Enum.member?(successors, short_name) do
          send_pub(short_name, msg)
          :ok
        else
          {:error, {:push_to_unknown_no_route, short_name}}
        end

      e = {:error, _} ->
        e
    end
  end

  def push(msg) do
    case get_valid_successors(msg) do
      {:ok, []} ->
        {:error, :no_successors}

      {:ok, succs} ->
        Enum.map(succs, fn s ->
          send_pub(s, msg)
        end)

        :ok

      e = {:error, _} ->
        e
    end
  end

  defp get_valid_successors(msg) do
    with {:routes, routes} when is_list(routes) <- {:routes, routes()},
         {:successors, {:ok, successors}} when is_list(successors) <- {:successors, successors()} do
      valid_successors =
        successors
        |> Enum.filter(fn succ ->
          local_routes =
            routes
            |> Enum.filter(fn {grain, _routes} -> grain == succ end)
            |> Enum.map(fn {_grain, routes} -> routes end)

          # no routes -> send message
          local_routes == [] ||
            local_routes
            # this is the OR of routes: A => [route(:foo, B), route(:bar, B)]
            |> Enum.any?(
              # this is the AND: nested routes
              &Enum.all?(&1, fn r -> r.(msg) end)
            )
        end)
        |> Enum.uniq()

      {:ok, valid_successors}
    else
      {:routes, _} ->
        {:error, :no_routes}

      {:successors, _} ->
        {:error, :no_successors}
    end
  end

  defp send_pub(grain, msg) do
    maybe_send(grain, {:push, own_name(), msg})
  end

  defp send_pull(grain, name) do
    maybe_send(grain, {:pull, name})
  end

  defp send_pull(grain, name, tag) do
    maybe_send(grain, {:pull, tag, name})
  end

  defp maybe_send(grain, msg) do
    case Process.whereis(full_name(grain)) do
      nil ->
        :ok

      pid ->
        send(pid, msg)
    end
  end

  defp send_with_error(grain, msg) do
    with {:ok, pid} <- resolve(grain),
         :ok <- check_successor_or_predecessor(grain) do
      send(pid, msg)

      :ok
    end
  end

  defp resolve(grain) do
    case Process.whereis(full_name(grain)) do
      nil ->
        {:error, {:unknown_grain, grain}}

      pid ->
        {:ok, pid}
    end
  end

  defp check_successor_or_predecessor(grain) do
    {:ok, predecessors} = predecessors()
    {:ok, successors} = successors()

    cond do
      grain in predecessors ->
        :ok

      grain in successors ->
        :ok

      true ->
        {:error,
         {:grain_not_linked,
          own_name: own_name(), grain: grain, predecessors: predecessors, successors: successors}}
    end
  end

  defp store(short_name, bread) do
    Process.put(@bread_key, bread)
    Process.put(@process_map_key, bread.process_map)
    Process.put(@own_name_key, short_name)

    Process.put(@successors_key, Bread.successors(bread, short_name))
    Process.put(@routes_key, Bread.routes(bread, short_name))
    Process.put(@predecessors_key, Bread.predecessors(bread, short_name))
  end

  def full_name(short_name) do
    process_map() |> Map.fetch!(short_name)
  end

  def own_name() do
    Process.get(@own_name_key)
  end

  def own_full_name do
    own_name() |> full_name()
  end

  defp process_map() do
    Process.get(@process_map_key)
  end

  @doc """
  Get a list of the grain's successors.

  Note that this can only be called from a grain process.
  """
  def successors() do
    if successors = Process.get(@successors_key) do
      {:ok, successors}
    else
      {:error, :no_successors}
    end
  end

  defp routes() do
    Process.get(@routes_key)
  end

  @doc """
  Get a list of the grain's predecessors.

  Note that this can only be called from a grain process.
  """
  def predecessors() do
    if predecessors = Process.get(@predecessors_key) do
      {:ok, predecessors}
    else
      {:error, :no_predecessors}
    end
  end
end