libcluster_uplink

defmodule Cluster.Strategy.Uplink do
  use GenServer
  use Cluster.Strategy

  alias Cluster.Strategy.State

  @default_polling_interval 5_000
  @service_discovery_key ~s(http:///1.0/config/user.service_discovery_endpoint)

  def start_link(args), do: GenServer.start_link(__MODULE__, args)

  @impl true
  def init([%State{meta: nil} = state]) do
    init([%State{state | :meta => MapSet.new()}])
  end

  def init([%State{} = state]) do
    {:ok, load(state)}
  end

  @impl true
  def handle_info(:timeout, state) do
    handle_info(:load, state)
  end

  def handle_info(:load, state) do
    {:noreply, load(state)}
  end

  def handle_info(_, state) do
    {:noreply, state}
  end

  defp load(
         %State{
           topology: topology,
           connect: connect,
           disconnect: disconnect,
           list_nodes: list_nodes
         } = state
       ) do
    case get_nodes(state) do
      {:ok, new_nodes} ->
        removed = MapSet.difference(state.meta, new_nodes)

        new_nodes =
          disconnect_nodes(
            topology,
            disconnect,
            list_nodes,
            MapSet.to_list(removed),
            new_nodes
          )

        new_nodes =
          connect_nodes(
            topology,
            connect,
            list_nodes,
            MapSet.to_list(new_nodes),
            new_nodes
          )

        Process.send_after(
          self(),
          :load,
          Keyword.get(
            state.config,
            :polling_interval,
            @default_polling_interval
          )
        )

        %State{state | :meta => new_nodes}

      _ ->
        Process.send_after(
          self(),
          :load,
          Keyword.get(
            state.config,
            :polling_interval,
            @default_polling_interval
          )
        )

        state
    end
  end

  defp get_nodes(%State{config: config}) do
    app_name = Keyword.fetch!(config, :app_name)
    service_discovery_endpoint = Keyword.get(config, :service_discovery_endpoint)

    endpoint =
      if service_discovery_endpoint do
        service_discovery_endpoint
      else
        %{body: instances_url} = Req.get!(@service_discovery_key, unix_socket: "/dev/lxd/sock")
        instances_url
      end

    Req.get!(endpoint)
    |> case do
      %{status: 200, body: %{"data" => %{"attributes" => %{"instances" => nodes}}}} ->
        nodes =
          nodes
          |> Enum.map(fn node_slug ->
            :"#{app_name}@#{node_slug}"
          end)

        {:ok, MapSet.new(nodes)}

      _ ->
        {:error, []}
    end
  end

  defp disconnect_nodes(
         topology,
         disconnect,
         list_nodes,
         to_be_removed,
         new_nodes
       ) do
    case Cluster.Strategy.disconnect_nodes(
           topology,
           disconnect,
           list_nodes,
           to_be_removed
         ) do
      :ok ->
        new_nodes

      {:error, bad_nodes} ->
        Enum.reduce(bad_nodes, new_nodes, fn {n, _}, acc ->
          MapSet.put(acc, n)
        end)
    end
  end

  defp connect_nodes(topology, connect, list_nodes, to_be_connected, new_nodes) do
    case Cluster.Strategy.connect_nodes(
           topology,
           connect,
           list_nodes,
           to_be_connected
         ) do
      :ok ->
        new_nodes

      {:error, bad_nodes} ->
        Enum.reduce(bad_nodes, new_nodes, fn {n, _}, acc ->
          MapSet.delete(acc, n)
        end)
    end
  end
end