schoolbus

defmodule Schoolbus do
  @moduledoc """
  Manage multiple PubSub instances
  """

  @type bus_id :: atom
  @type topic :: atom | {atom, atom}
  @type topic_match :: topic | {atom, String.t()}
  @type subscription :: {topic, pid}
  @type topics_opt :: {:topics, [topic]}
  @type partitions_opt :: {:partitions, pos_integer}
  @type bus_id_opt :: {:id, bus_id}
  @type option :: topics_opt | partitions_opt | bus_id_opt

  @default_id :default

  @spec child_spec(opts :: term) :: Supervisor.child_spec()
  def child_spec(opts) do
    id = get_id(opts)

    %{
      id: :"#{__MODULE__}_#{id}",
      start: {__MODULE__, :start_link, [opts]},
      type: :supervisor,
      restart: :permanent,
      shutdown: 500
    }
  end

  @spec start_link([option]) :: {:ok, pid} | {:error, term}
  def start_link(opts) do
    partitions = Keyword.get(opts, :partitions, System.schedulers_online())
    topics = Keyword.get(opts, :topics, [])
    id = get_id(opts)
    name = to_name(id)

    Registry.start_link(
      keys: :duplicate,
      name: name,
      partitions: partitions,
      meta: [topics: topics]
    )
  end

  @spec to_name(bus_id) :: atom
  def to_name(id), do: :"#{__MODULE__}_#{id}"

  @spec topics :: [topic]
  @spec topics(bus_id) :: [topic]
  def topics(id \\ @default_id) do
    {:ok, topics} = id |> to_name |> Registry.meta(:topics)
    topics
  end

  @spec register(topic) :: :ok
  @spec register(topic, bus_id) :: :ok
  def register(topic, id \\ @default_id) do
    id
    |> to_name
    |> Registry.put_meta(:topics, [topic | topics(id)])
  end

  @spec subscribers :: [subscription]
  @spec subscribers(bus_id) :: [subscription]
  def subscribers(id \\ @default_id) do
    id
    |> to_name
    |> Registry.select([
      {
        {:"$1", :"$2", :"$3"},
        [],
        [{{:"$1", :"$2"}}]
      }
    ])
  end

  @spec subscribe(topic_match) :: {:ok, [topic]}
  @spec subscribe(topic_match, bus_id) :: {:ok, [topic]}
  def subscribe(topic), do: subscribe(topic, @default_id)

  def subscribe({subscribe_namespace, "*"}, bus_id) do
    bus_id
    |> registered_topics()
    |> Enum.reduce(
      {:ok, []},
      fn rt, {:ok, subscribed_topics} = acc ->
        registered_namespace = topic_namespace(rt)

        if subscribe_namespace == registered_namespace do
          {:ok, _} = bus_id |> to_name |> Registry.register(rt, [])
          {:ok, [rt | subscribed_topics]}
        else
          acc
        end
      end
    )
  end

  def subscribe(topic, bus_id) do
    bus_id
    |> registered_topics()
    |> Enum.member?(topic)
    |> case do
      true ->
        {:ok, _} = bus_id |> to_name |> Registry.register(topic, [])
        {:ok, [topic]}

      false ->
        {:ok, []}
    end
  end

  @spec unsubscribe(topic) :: :ok
  @spec unsubscribe(topic, bus_id) :: :ok
  def unsubscribe(topic, id \\ @default_id) do
    id
    |> to_name
    |> Registry.unregister(topic)
  end

  @spec broadcast(topic, term) :: :ok
  @spec broadcast(topic, term, bus_id) :: :ok
  def broadcast(topic, message, id \\ @default_id) do
    id
    |> to_name
    |> Registry.dispatch(
      topic,
      fn entries ->
        for {pid, _} <- entries, do: send(pid, message)
      end
    )
  end

  defp get_id(opts), do: Keyword.get(opts, :id, @default_id)

  defp topic_namespace({namespace, _}), do: namespace
  defp topic_namespace(namespace), do: namespace

  defp registered_topics(bus_id) do
    {:ok, topics} = bus_id |> to_name |> Registry.meta(:topics)
    topics
  end
end