defmodule Ravix.Connection.RequestExecutor.Supervisor do @moduledoc false use DynamicSupervisor require Logger alias Ravix.Connection.{RequestExecutor, ServerNode, Topology, NodeSelector} def init(init_arg) do DynamicSupervisor.init( strategy: :one_for_one, extra_arguments: [init_arg] ) end @spec start_link(any) :: :ignore | {:error, any} | {:ok, pid} def start_link(store) do DynamicSupervisor.start_link(__MODULE__, %{}, name: supervisor_name(store)) end @doc """ Register a new RavenDB Database node for the informed store ## Parameters - store: the store module. E.g: Ravix.Test.Store - node: the node to be registered """ @spec register_node_pool(any, ServerNode.t()) :: :ignore | {:error, any} | {:ok, pid} | {:ok, pid, any} def register_node_pool(store, %ServerNode{} = node) do Logger.debug( "[RAVIX] Registering the connection pool with the node '#{node.url}' for the store '#{inspect(store)}'" ) node = %ServerNode{node | store: store} DynamicSupervisor.start_child(supervisor_name(store), {RequestExecutor, node}) end @spec remove_node_pool(atom(), pid) :: :ok | {:error, :not_found} def remove_node_pool(store, pid) do Logger.debug( "[RAVIX] Removing cluster node '#{inspect(pid)}' for the store '#{inspect(store)}'" ) DynamicSupervisor.terminate_child(supervisor_name(store), pid) end @doc """ Fetch the Node pools for the informed store ## Parameters - store: the store module: E.g: Ravix.Test.Store """ @spec fetch_node_pools(atom()) :: list({pid(), binary(), ServerNode.t()}) def fetch_node_pools(store) do Registry.select(:request_executor_pools, [ {{:"$1", :"$2", :"$3"}, [], [{{:"$1", :"$2", :"$3"}}]} ]) |> Enum.filter(fn {_pool_name, _pid, node} -> node.store == store end) |> Enum.map(fn {pool_name, pid, node} -> {pid, pool_name, node} end) end @doc """ Fetch all the executors for the informed Store ## Parameters - store: the store module: E.g: Ravix.Test.Store """ @spec fetch_executors(atom()) :: list(pid()) def fetch_executors(store) do Registry.select(:request_executors, [{{:"$1", :"$2", :"$3"}, [], [{{:"$1", :"$2"}}]}]) |> Enum.filter(fn {executor_store, _pid} -> executor_store == store end) |> Enum.map(fn {_executor_store, pid} -> pid end) end @doc """ Triggers a topology update for all nodes of a specific store ## Parameters - store: the store module. E.g: Ravix.Test.Store - topology: The `Ravix.Connection.Topology` to be used ## Returns - List of nodes `[new_nodes: list(Ravix.Connection.ServerNode), updated_nodes: list(Ravix.Connection.ServerNode)]` """ @spec update_topology(any, Ravix.Connection.Topology.t()) :: [ {:new_nodes, list} | {:updated_nodes, list} ] def update_topology(store, %Topology{} = topology) do Logger.info("[RAVIX] A topology update was requested by the server, updating nodes...") current_nodes = fetch_node_pools(store) remaining_nodes = remove_old_nodes(store, current_nodes, topology) updated_nodes = update_existing_nodes(remaining_nodes, topology) new_nodes = add_new_nodes(store, remaining_nodes, topology) [updated_nodes: updated_nodes, new_nodes: new_nodes] end defp remove_old_nodes(store, current_nodes, %Topology{} = topology) do new_nodes_ids = Enum.map(topology.nodes, &NodeSelector.node_id(&1)) nodes_to_delete = current_nodes |> Enum.reject(fn {_pid, pool_id, _node} -> Enum.member?(new_nodes_ids, pool_id) end) nodes_to_delete |> Enum.each(fn {pid, _, _} -> RequestExecutor.Supervisor.remove_node_pool(store, pid) end) Logger.info("[RAVIX] Unregistering the nodes '#{inspect(nodes_to_delete)}'") current_nodes -- nodes_to_delete end defp update_existing_nodes(nodes, %Topology{} = topology) do nodes |> Enum.map(fn {pid, _, node} -> {pid, %ServerNode{ node | cluster_tag: Topology.cluster_tag_for_node(topology, node.url) }} end) |> Enum.map(fn {pid, node} -> Logger.info( "[RAVIX] Updating node '#{inspect(node.url)}' for the database '#{inspect(node.database)}' with cluster tag '#{inspect(node.cluster_tag)}'" ) RequestExecutor.Supervisor.remove_node_pool(node.store, pid) RequestExecutor.Supervisor.register_node_pool(node.store, node) end) end defp add_new_nodes(store, existing_nodes, %Topology{} = topology) do existing_nodes = Enum.map(existing_nodes, fn {_pid, pool_id, _node} -> pool_id end) topology.nodes |> Enum.reject(&Enum.member?(existing_nodes, NodeSelector.node_id(&1))) |> Enum.map(fn new_node -> Logger.info( "[RAVIX] Registering new node '#{inspect(new_node)}' for the store '#{inspect(store)}'" ) RequestExecutor.Supervisor.register_node_pool(store, new_node) end) end defp supervisor_name(store) do :"#{store}.Executor" end end