defmodule ZenMonitor.Local.Connector do
  @moduledoc """
  `ZenMonitor.Local.Connector` performs a variety of duties.  For every remote that a the local
  is interested in monitoring processes on there will be a dedicated `ZenMonitor.Local.Connector`.
  This collection of Connectors are managed by a `GenRegistry` registered under the
  `ZenMonitor.Local.Connector` atom.

  # Connecting and Monitoring the remote `ZenMonitor.Proxy`

  Connectors, as their name suggests, connect to the `ZenMonitor.Proxy` on the remote node that they
  are responsible for.  They do this using standard ERTS Distribution, by invoking the remote
  Proxy's ping command.  A Remote is considered compatible if the ping command returns the :pong
  atom, otherwise it will be marked incompatible.

  Connectors manage their remote node's status in the global node status cache, and provide
  facilities for efficient querying of remote status, see `compatibility/1` and
  `cached_compatibility/1`

  # Batching and Updating the remote `ZenMonitor.Proxy`

  When a local process wishes to monitor a remote process, the Connector will be informed of this
  fact with a call to `monitor/3`.  The Connector is responsible for maintaining a local record of
  this monitor for future fan-out and for efficiently batching up these requests to be delivered
  to the remote ZenMonitor.Proxy.

  # Fan-out of Dead Summaries

  Periodically, the `ZenMonitor.Proxy` (technically the `ZenMonitor.Proxy.Batcher`) on the remote
  node will send a "Dead Summary".  This is a message from the remote that informs the Connector
  of all the processes the Connector has monitored that have gone down since the last summary.

  The Connector uses it's local records to generate a batch of _down dispatches_.  These are
  messages that look identical to the messages provided by `Process.monitor/1` when a process goes
  down.  It is sometimes necessary for the original monitoring process to be able to discern
  whether the `:DOWN` message originated from ERTS or from ZenMonitor, to aid this, ZenMonitor
  will wrap the original reason in a tuple of `{:zen_monitor, original_reason}`.

  The fan-out messages are sent to `ZenMonitor.Local` for eventual delivery via
  `ZenMonitor.Local.Dispatcher`, see those modules for more information.

  # Fan-out of nodedown / ZenMonitor.Proxy down

  The Connector is also responsible for monitoring the remote node and dealing with nodedown (or
  the node becoming incompatible, either due to the `ZenMonitor.Proxy` crashing or a code change).

  If the Connector detects that the remote it is responsible for is down or no longer compatible,
  it will fire every established monitor with `{:zen_monitor, :nodedown}`.  It uses the same
  mechanism as for Dead Summaries, see `ZenMonitor.Local` and `ZenMonitor.Local.Dispatcher` for
  more information.
  """
  use GenServer
  use Instruments.CustomFunctions, prefix: "zen_monitor.local.connector"

  alias ZenMonitor.Local
  alias ZenMonitor.Local.Tables

  @base_penalty 1_000
  @maximum_penalty 60_000
  @max_attempt :math.ceil(:math.log2(@maximum_penalty))
  @chunk_size 5000
  @sweep_interval 100

  @type t :: __MODULE__
  @type compatibility :: :compatible | :incompatible
  @type cached_compatibility :: compatibility | :miss | {:expired, integer} | :unavailable
  @type death_certificate :: {pid, reason :: any}
  @type down_dispatch :: {pid, {:DOWN, reference, :process, pid, {:zen_monitor, any}}}

  defmodule State do
    @moduledoc """
    Maintains the internal state for the Connector

      - `monitors` is an ETS table for keeping track of monitors for the purpose of fan-out.
      - `remote_node_monitored` is a flag used to track whether or not the remote node has been
        monitored
      - `remote_proxy_ref` is the monitoring reference of the remote node's ZenMonitor.Proxy
      - `remote` is the remote node for which the Connector is responsible.
      - `batch` is the queue of instructions pending until the next sweep.
      - `length` is the current length of the batch queue (calculating queue length is an O(n)
        operation, it is simple to track it as elements are added / removed)
    """
    @type t :: %__MODULE__{
            monitors: :ets.tab(),
            remote_node_monitored: boolean(),
            remote_proxy_ref: reference() | nil,
            remote: node(),
            length: integer(),
            batch: :queue.queue()
          }

    defstruct [
      :monitors,
      :remote,
      :remote_proxy_ref,
      remote_node_monitored: false,
      length: 0,
      batch: :queue.new()
    ]
  end

  ## Client

  def start_link(remote) do
    GenServer.start_link(__MODULE__, remote)
  end

  @doc """
  Get a connector from the registry by destination
  """
  @spec get(target :: ZenMonitor.destination()) :: pid()
  def get(target) do
    target
    |> ZenMonitor.find_node()
    |> get_for_node()
  end

  @doc """
  Get a connector from the registry by remote node
  """
  @spec get_for_node(remote :: node()) :: pid()
  def get_for_node(remote) when is_atom(remote) do
    case GenRegistry.lookup(__MODULE__, remote) do
      {:ok, connector} ->
        connector

      {:error, :not_found} ->
        {:ok, connector} = GenRegistry.lookup_or_start(__MODULE__, remote, [remote])
        connector
    end
  end

  @doc """
  Asynchronously monitors a pid.
  """
  @spec monitor(target :: ZenMonitor.destination(), ref :: reference(), subscriber :: pid()) ::
          :ok
  def monitor(target, ref, subscriber) do
    target
    |> get()
    |> GenServer.cast({:monitor, target, ref, subscriber})
  end

  @doc """
  Retrieves all the monitors established between the target and the subscriber
  """
  @spec monitors(target :: ZenMonitor.destination(), subscriber :: pid()) :: [reference()]
  def monitors(target, subscriber) do
    target
    |> get()
    |> GenServer.call({:monitors, target, subscriber})
  end

  @doc """
  Asynchronously demonitors a pid.
  """
  @spec demonitor(target :: ZenMonitor.destination(), ref :: reference()) :: :ok
  def demonitor(target, ref) do
    target
    |> get()
    |> GenServer.cast({:demonitor, target, ref})
  end

  @doc """
  Determine the effective compatibility of a remote node

  This will attempt a fast client-side lookup in the ETS table.  Only a positive `:compatible`
  record will result in `:compatible`, otherwise the effective compatibility is `:incompatible`
  """
  @spec compatibility(remote :: node()) :: compatibility
  def compatibility(remote) do
    case cached_compatibility(remote) do
      :compatible ->
        :compatible

      _ ->
        :incompatible
    end
  end

  @doc """
  Check the cached compatibility status for a remote node

  This will only perform a fast client-side lookup in the ETS table.  If an authoritative entry is
  found it will be returned (either `:compatible`, `:incompatible`, or `:unavailable`).  If no
  entry is found then `:miss` is returned.  If an expired entry is found then
  `{:expired, attempts}` is returned.
  """
  @spec cached_compatibility(remote :: node()) :: cached_compatibility
  def cached_compatibility(remote) do
    case :ets.lookup(Tables.nodes(), remote) do
      [] ->
        :miss

      [{^remote, :compatible}] ->
        :compatible

      [{^remote, {:incompatible, enforce_until, attempt}}] ->
        if enforce_until < ZenMonitor.now() do
          {:expired, attempt}
        else
          :incompatible
        end

      [{^remote, :unavailable}] ->
        :unavailable
    end
  end

  @doc """
  Connect to the provided remote

  This function will not consult the cache before calling into the GenServer, the GenServer will
  consult with the cache before attempting to connect, this allows for many callers to connect
  with the server guaranteeing that only one attempt will actually perform network work.

  If the compatibility of a remote host is needed instead, callers should use the
  `compatibility/1` or `cached_compatibility/1` functions.  `compatibility/1` will provide the
  effective compatibility, `cached_compatibility/1` is mainly used internally but can provide more
  detailed information about the cache status of the remote.  Neither of these methods,
  `compatibility/1` nor `cached_compatibility/1`, will perform network work or call into the
  GenServer.
  """
  @spec connect(remote :: node()) :: compatibility
  def connect(remote) do
    remote
    |> get_for_node()
    |> GenServer.call(:connect)
  end

  @doc """
  Gets the sweep interval from the Application Environment

  The sweep interval is the number of milliseconds to wait between sweeps, see
  ZenMonitor.Local.Connector's @sweep_interval for the default value

  This can be controlled at boot and runtime with the {:zen_monitor, :connector_sweep_interval}
  setting, see `ZenMonitor.Local.Connector.sweep_interval/1` for runtime convenience
  functionality.
  """
  @spec sweep_interval() :: integer
  def sweep_interval do
    Application.get_env(:zen_monitor, :connector_sweep_interval, @sweep_interval)
  end

  @doc """
  Puts the sweep interval into the Application Environment

  This is a simple convenience function for overwriting the
  {:zen_monitor, :connector_sweep_interval} setting at runtime.
  """
  @spec sweep_interval(value :: integer) :: :ok
  def sweep_interval(value) do
    Application.put_env(:zen_monitor, :connector_sweep_interval, value)
  end

  @doc """
  Gets the chunk size from the Application Environment

  The chunk size is the maximum number of subscriptions that will be sent during each sweep, see
  ZenMonitor.Local.Connector's @chunk_size for the default value

  This can be controlled at boot and runtime with the {:zen_monitor, :connector_chunk_size}
  setting, see `ZenMonitor.Local.Connector.chunk_size/1` for runtime convenience functionality.
  """
  @spec chunk_size() :: integer
  def chunk_size do
    Application.get_env(:zen_monitor, :connector_chunk_size, @chunk_size)
  end

  @doc """
  Puts the chunk size into the Application Environment

  This is a simple convenience function for overwriting the {:zen_monitor, :connector_chunk_size}
  setting at runtime.
  """
  @spec chunk_size(value :: integer) :: :ok
  def chunk_size(value) do
    Application.put_env(:zen_monitor, :connector_chunk_size, value)
  end

  ## Server

  def init(remote) do
    schedule_sweep()
    monitors = :ets.new(:monitors, [:private, :ordered_set])
    {:ok, %State{remote: remote, monitors: monitors}}
  end

  # Synchronous connect handler
  # Attempts to connect to the remote, this handler does check the cache before connecting to avoid
  # a thundering herd.
  def handle_call(:connect, _from, %State{} = state) do
    {result, state} = do_compatibility(state)
    {:reply, result, state}
  end

  # Returns all the monitors between a target and a subscriber
  def handle_call({:monitors, target, subscriber}, _from, %State{} = state) do
    size = :ets.info(state.monitors, :size)

    monitors =
      if size == 0 do
        # Don't bother doing the match on an empty table
        []
      else
        case :ets.match(state.monitors, {{target, :"$1"}, subscriber}, size) do
          :"$end_of_table" ->
            # Match failed
            []

          {monitors, _} ->
            # Unwrap the references
            List.flatten(monitors)
        end
      end

    {:reply, monitors, state}
  end

  # Handles establishing a new monitor
  # 1.  Records the monitor into the internal ETS table
  # 2.  If this is the first monitor for the pid, adds it to the queue for subsequent dispatch to
  #     the ZenMonitor.Proxy during the next sweep.
  def handle_cast(
        {:monitor, target, ref, subscriber},
        %State{batch: batch, length: length, monitors: monitors} = state
      ) do
    # Check if we should subscribe to this target (this check has to happen before we insert the
    # new monitor otherwise the new monitor will always be found and we will never enqueue
    # anything)
    should_subscribe? = unknown_target?(monitors, target)

    # Always add it to the monitor table
    :ets.insert(monitors, {{target, ref}, subscriber})

    # Enqueue the subscribe instruction if it isn't already monitored
    new_state =
      if should_subscribe? do
        increment("enqueue", 1, tags: ["op:subscribe"])
        %State{state | batch: :queue.in({:subscribe, target}, batch), length: length + 1}
      else
        state
      end

    {:noreply, new_state}
  end

  # Handles demonitoring a reference for a given pid
  # Cleans up the internal ETS record if it exists
  def handle_cast(
        {:demonitor, target, ref},
        %State{batch: batch, length: length, monitors: monitors} = state
      ) do
    # Remove it from the monitors table
    :ets.delete(monitors, {target, ref})

    # If that was the last monitor for the target, we should unsubscribe.  Unlike monitor we have
    # to perform this check after the delete or else the row we are deleting will always make the
    # target known.
    should_unsubscribe? = unknown_target?(monitors, target)

    # Enqueue the unsubscribe instruction if the target no longer exists
    state =
      if should_unsubscribe? do
        increment("enqueue", 1, tags: ["op:unsubscribe"])
        %State{state | batch: :queue.in({:unsubscribe, target}, batch), length: length + 1}
      else
        state
      end

    {:noreply, state}
  end

  # Handles nodedown for the Connector's remote
  # When the remote node goes down, every monitor maintained by the Connector should fire
  def handle_info({:nodedown, remote}, %State{remote: remote} = state) do
    # Mark this node as unavailable
    {:incompatible, state} = do_mark_unavailable(state)

    # Mark the remote node as unmonitored (any monitors that existed were just consumed)
    state = %State{state | remote_node_monitored: false}

    # Dispatch down to everyone
    {:noreply, do_down(state)}
  end

  # Handles when the proxy crashes because of noconnection
  # This reason indicates that we have lost connection with the remote node, mark it as unavailable.
  def handle_info({:DOWN, ref, :process, _, :noconnection}, %State{remote_proxy_ref: ref} = state) do
    # Mark this node as unavailable
    {:incompatible, state} = do_mark_unavailable(state)

    # Clear the remote_proxy_ref
    state = %State{state | remote_proxy_ref: nil}

    # Dispatch down to everyone
    {:noreply, do_down(state)}
  end

  # Handles when the proxy crashes for any other reason
  # Penalize the remote as incompatible and let the normal remote recovery take care of it.
  def handle_info({:DOWN, ref, :process, _, _}, %State{remote_proxy_ref: ref} = state) do
    # Mark this node as incompatible
    {:incompatible, state} = do_mark_incompatible(state, 1)

    # Clear the remote_proxy_ref
    state = %State{state | remote_proxy_ref: nil}

    # Dispatch down to everyone
    {:noreply, do_down(state)}
  end

  # Handle the dead summary from the remote
  # Periodically the remote node will send us a summary of everything that has died that we have
  # monitored.
  # Connector will find and consume all the matching monitors and enqueue the appropriate messages
  # for each monitor with ZenMonitor.Local
  def handle_info(
        {:dead, remote, death_certificates},
        %State{remote: remote, monitors: monitors} = state
      ) do
    death_certificates
    |> messages_for_death_certificates(monitors)
    |> Local.enqueue()

    {:noreply, state}
  end

  # Handle the periodic sweep
  # If the remote is compatible this will create a subscription summary up to chunk_size of all the
  # pids that need monitoring since the last sweep.  This will be sent to the remote for monitoring.
  # If the remote is incompatible, all pids since the last sweep will have their monitors fire with
  # `{:zen_monitor, :nodedown}`
  def handle_info(:sweep, %State{} = state) do
    new_state =
      case do_compatibility(state) do
        {:compatible, state} ->
          do_sweep(state)

        {:incompatible, state} ->
          do_down(state)
      end

    schedule_sweep()
    {:noreply, new_state}
  end

  def handle_info(_, %State{} = state) do
    increment("unhandled_info")
    {:noreply, state}
  end

  ## Private

  @spec do_compatibility(state :: State.t()) :: {compatibility, State.t()}
  defp do_compatibility(%State{remote: remote} = state) do
    case cached_compatibility(remote) do
      :miss ->
        do_connect(state, 1)

      {:expired, attempt} ->
        do_connect(state, attempt + 1)

      :unavailable ->
        do_connect(state, 1)

      hit ->
        {hit, state}
    end
  end

  @spec do_connect(State.t(), attempt :: integer) :: {compatibility, State.t()}
  defp do_connect(%State{remote: remote} = state, attempt) do
    try do
      with {:known_node, true} <- {:known_node, known_node?(remote)},
           {:ping, :pong} <-
             {:ping, ZenMonitor.gen_module().call({ZenMonitor.Proxy, remote}, :ping)} do
        do_mark_compatible(state)
      else
        {:known_node, false} ->
          do_mark_unavailable(state)

        {:ping, _} ->
          do_mark_incompatible(state, attempt)
      end
    catch
      :exit, {{:nodedown, _node}, _} ->
        do_mark_unavailable(state)

      :exit, _ ->
        do_mark_incompatible(state, attempt)
    end
  end

  @spec do_sweep(state :: State.t()) :: State.t()
  defp do_sweep(%State{batch: batch, length: length} = state) do
    {summary, overflow, new_length} = chunk(batch, length)
    increment("sweep", length - new_length)
    do_subscribe(state, summary)
    %State{state | batch: overflow, length: new_length}
  end

  @spec chunk(batch :: :queue.queue(), length :: integer) :: {[pid], :queue.queue(), integer}
  defp chunk(batch, length) do
    size = chunk_size()

    if length <= size do
      {:queue.to_list(batch), :queue.new(), 0}
    else
      {summary, overflow} = :queue.split(size, batch)
      {:queue.to_list(summary), overflow, length - size}
    end
  end

  @spec do_subscribe(state :: State.t(), summary :: []) :: :ok
  defp do_subscribe(%State{}, []), do: :ok

  defp do_subscribe(%State{remote: remote}, summary) do
    ZenMonitor.gen_module().cast({ZenMonitor.Proxy, remote}, {:process, self(), summary})
  end

  @spec do_down(state :: State.t()) :: State.t()
  defp do_down(%State{monitors: monitors} = state) do
    # Generate messages for every monitor
    messages =
      for [{{pid, ref}, subscriber}] <- :ets.match(monitors, :"$1") do
        {subscriber, {:DOWN, ref, :process, pid, {:zen_monitor, :nodedown}}}
      end

    # Clear the monitors table
    :ets.delete_all_objects(monitors)

    # Enqueue the messages with ZenMonitor.Local
    Local.enqueue(messages)

    # Return a new empty state
    %State{state | batch: :queue.new(), length: 0}
  end

  @spec do_mark_compatible(State.t()) :: {:compatible, State.t()}
  defp do_mark_compatible(%State{remote: remote} = state) do
    state =
      state
      |> monitor_remote_node()
      |> monitor_remote_proxy()

    :ets.insert(Tables.nodes(), {remote, :compatible})
    {:compatible, state}
  end

  @spec do_mark_incompatible(State.t(), attempt :: integer) :: {:incompatible, State.t()}
  defp do_mark_incompatible(%State{remote: remote} = state, attempt) do
    state = monitor_remote_node(state)

    :ets.insert(
      Tables.nodes(),
      {remote, {:incompatible, ZenMonitor.now() + penalty(attempt), attempt}}
    )

    {:incompatible, state}
  end

  @spec do_mark_unavailable(State.t()) :: {:incompatible, State.t()}
  defp do_mark_unavailable(%State{remote: remote} = state) do
    :ets.insert(Tables.nodes(), {remote, :unavailable})
    {:incompatible, state}
  end

  @spec monitor_remote_node(State.t()) :: State.t()
  defp monitor_remote_node(%State{remote_node_monitored: true} = state), do: state

  defp monitor_remote_node(%State{remote_node_monitored: false, remote: remote} = state) do
    Node.monitor(remote, true)
    %State{state | remote_node_monitored: true}
  end

  @spec monitor_remote_proxy(State.t()) :: State.t()
  defp monitor_remote_proxy(%State{remote_proxy_ref: nil, remote: remote} = state) do
    %State{state | remote_proxy_ref: Process.monitor({ZenMonitor.Proxy, remote})}
  end

  defp monitor_remote_proxy(%State{} = state), do: state

  @spec messages_for_death_certificates(
          death_certificates :: [death_certificate],
          monitors :: :ets.tab()
        ) :: [down_dispatch]
  defp messages_for_death_certificates(death_certificates, monitors) do
    do_messages_for_death_certificates(death_certificates, monitors, [])
  end

  @spec do_messages_for_death_certificates(
          death_certificates :: [death_certificate],
          monitors :: :ets.tab(),
          acc :: [down_dispatch]
        ) :: [down_dispatch]
  defp do_messages_for_death_certificates([], _monitors, acc), do: Enum.reverse(acc)

  defp do_messages_for_death_certificates([{pid, reason} | rest], monitors, acc) do
    acc =
      monitors
      |> :ets.match({{pid, :"$1"}, :"$2"})
      |> Enum.reduce(acc, fn [ref, subscriber], acc ->
        # Consume the monitor
        :ets.delete(monitors, {pid, ref})

        # Add the new message into the accumulator
        [{subscriber, {:DOWN, ref, :process, pid, {:zen_monitor, reason}}} | acc]
      end)

    do_messages_for_death_certificates(rest, monitors, acc)
  end

  @spec known_node?(remote :: node()) :: boolean()
  defp known_node?(remote) do
    remote == Node.self() or remote in Node.list()
  end

  @spec penalty(attempt :: integer) :: integer
  defp penalty(attempt) do
    min(@maximum_penalty, @base_penalty * round(:math.pow(2, min(attempt, @max_attempt))))
  end

  @spec unknown_target?(monitors :: :ets.tid(), target :: pid) :: boolean
  defp unknown_target?(monitors, target) do
    # ETS does not make for the most readable code, here's what the following line does.
    # Perform a match on the internal monitors table looking for keys that start with
    # {target, ...}
    # Since we are just interested to see if there are any, but don't care about the content, we
    # set the other fields to :_ to ignore them.
    # The target is known if there are _any_ results, so we apply a limit to the match of just 1
    # result.
    # This means that we either get back a tuple of {[[]]], continuation} or :"$end_of_table"
    # :"$end_of_table" implies that the match for a single item found nothing, therefore the
    # target does not exist and is unknown
    :ets.match(monitors, {{target, :_}, :_}, 1) == :"$end_of_table"
  end

  @spec schedule_sweep() :: reference
  defp schedule_sweep do
    Process.send_after(self(), :sweep, sweep_interval())
  end
end
