meshx_node

defmodule MeshxNode.Dist.Controller do
  @moduledoc false
  require Logger

  def call_controller(dist_ctrl, message) do
    ref = :erlang.monitor(:process, dist_ctrl)
    send(dist_ctrl, {ref, self(), message})

    receive do
      {^ref, result} ->
        Process.demonitor(ref, [:flush])
        result

      {'DOWN', ^ref, :process, ^dist_ctrl, reason} ->
        exit({:dist_controller_exit, reason})
    end
  end

  def flush_controller(pid, socket) do
    receive do
      {:tcp, ^socket, data} ->
        send(pid, {:tcp, socket, data})
        flush_controller(pid, socket)

      {:tcp_closed, ^socket} ->
        send(pid, {:tcp_closed, socket})
        flush_controller(pid, socket)
    after
      0 ->
        :ok
    end
  end

  @dist_controller_common_spawn_opts [message_queue_data: :off_heap, fullsweep_after: 0]

  def spawn_dist_controller(socket) do
    Process.spawn(
      fn -> dist_controller_setup(socket) end,
      [priority: :max] ++ @dist_controller_common_spawn_opts
    )
  end

  defp dist_controller_setup(socket) do
    tick_handler =
      Process.spawn(
        fn -> tick_handler(socket) end,
        [:link, priority: :max] ++ @dist_controller_common_spawn_opts
      )

    dist_controller_setup_loop(socket, tick_handler, :undefined)
  end

  defp dist_controller_setup_loop(socket, tick_handler, sup) do
    receive do
      {:tcp_closed, ^socket} ->
        exit(:connection_closed)

      {ref, from, {:supervisor, pid}} ->
        res = Process.link(pid)
        send(from, {ref, res})
        dist_controller_setup_loop(socket, tick_handler, pid)

      {ref, from, :tick_handler} ->
        send(from, {ref, tick_handler})
        dist_controller_setup_loop(socket, tick_handler, sup)

      {ref, from, :socket} ->
        send(from, {ref, socket})
        dist_controller_setup_loop(socket, tick_handler, sup)

      {ref, from, {:send, packet}} ->
        res = :gen_tcp.send(socket, packet)
        send(from, {ref, res})
        dist_controller_setup_loop(socket, tick_handler, sup)

      {ref, from, {:recv, length, timeout}} ->
        res = :gen_tcp.recv(socket, length, timeout)

        case res do
          {:ok, "snot_allowed"} -> Logger.warn("** Connection attempt to disallowed node **")
          _ -> :ok
        end

        send(from, {ref, res})
        dist_controller_setup_loop(socket, tick_handler, sup)

      {ref, from, :getll} ->
        send(from, {ref, {:ok, self()}})
        dist_controller_setup_loop(socket, tick_handler, sup)

      {ref, from, :pre_nodeup} ->
        res = :inet.setopts(socket, active: false, packet: 4)
        send(from, {ref, res})
        dist_controller_setup_loop(socket, tick_handler, sup)

      {ref, from, :post_nodeup} ->
        res = :inet.setopts(socket, active: false, packet: 4)
        send(from, {ref, res})
        dist_controller_setup_loop(socket, tick_handler, sup)

      {ref, from, {:handshake_complete, _node, d_handle}} ->
        send(from, {ref, :ok})

        input_handler =
          Process.spawn(
            fn -> dist_controller_input_handler(d_handle, socket, sup) end,
            [:link] ++ @dist_controller_common_spawn_opts
          )

        flush_controller(input_handler, socket)
        :gen_tcp.controlling_process(socket, input_handler)
        flush_controller(input_handler, socket)
        :erlang.dist_ctrl_input_handler(d_handle, input_handler)
        send(input_handler, d_handle)
        Process.flag(:priority, :normal)
        :erlang.dist_ctrl_get_data_notification(d_handle)
        dist_controller_output_handler(d_handle, socket)
    end
  end

  @active_input 10
  defp dist_controller_input_handler(d_handle, socket, sup) do
    Process.link(sup)

    receive do
      ^d_handle ->
        dist_controller_input_loop(d_handle, socket, 0)
    end
  end

  defp dist_controller_input_loop(d_handle, socket, n) when n <= @active_input / 2 do
    :inet.setopts(socket, active: @active_input - n)
    dist_controller_input_loop(d_handle, socket, @active_input)
  end

  defp dist_controller_input_loop(d_handle, socket, n) do
    receive do
      {:tcp, socket, data} ->
        try do
          :erlang.dist_ctrl_put_data(d_handle, data)
        catch
          _, _ -> death_row()
        end

        dist_controller_input_loop(d_handle, socket, n - 1)

      {:tcp_closed, ^socket} ->
        exit(:connection_closed)

      _ ->
        dist_controller_input_loop(d_handle, socket, n)
    end
  end

  defp dist_controller_output_handler(d_handle, socket) do
    receive do
      :dist_data ->
        try do
          dist_controller_send_data(d_handle, socket)
        catch
          _, _ -> death_row()
        end

        dist_controller_output_handler(d_handle, socket)

      _ ->
        dist_controller_output_handler(d_handle, socket)
    end
  end

  defp dist_controller_send_data(d_handle, socket) do
    case :erlang.dist_ctrl_get_data(d_handle) do
      :none ->
        :erlang.dist_ctrl_get_data_notification(d_handle)

      data ->
        socket_send(socket, data)
        dist_controller_send_data(d_handle, socket)
    end
  end

  defp tick_handler(socket) do
    receive do
      :tick ->
        socket_send(socket, '')

      _ ->
        :ok
    end

    tick_handler(socket)
  end

  defp socket_send(socket, data) do
    try do
      :gen_tcp.send(socket, data)
    catch
      {type, reason} -> death_row({:send_error, {type, reason}})
    else
      :ok ->
        :ok

      {:error, reason} ->
        death_row({:send_error, reason})
    end
  end

  defp death_row(), do: death_row(:connection_closed)

  defp death_row(reason) do
    receive do
      any -> any
    after
      5000 ->
        exit(reason)
    end
  end
end