prism

defmodule Prism do
  @moduledoc """
  Prism is a simple local message broker.

  With this module, you can subscribe handlers to topics. When an event is published
  to the broker - any subscriber of that topic will have its handler function called.
  This permits simple event-driven mechanisms to be implemented within applications.

  Prism avoids the use of processes when publishing events to subscribers. In that
  sense - handlers are all invoked in the publishing process. If you wish something
  to be done "out of band" from the publisher - you will be resposibsle for the
  implementation.

  ## Example

      defmodule MyBroker do
        use Prism

        def start_link(subscribers \\\\ []) do
          Prism.start_link(name: __MODULE__, subscribers: subscribers)
        end

        def subscribe(handler_id, topic_or_topics, handler) do
          Prism.subscribe(__MODULE__, handler_id, topic_or_topics, handler)
        end

        def publish(topic, event) do
          Prism.publish(__MODULE__, topic, event)
        end
      end

      defmodule Callback do
        def call(topic, event) do
          IO.inspect(topic)
          IO.inspect(event)
          :ok
        end
      end

      # Start the broker
      MyBroker.start_link()

      # Subscribe a handler
      MyBroker.subscribe("my_handler", [:my, :topic], &Callback.call/2)

      # Publish to a topic
      MyBroker.publish([:my, :topic], "hello")

  """

  @typedoc "The unique broker identifier"
  @type broker_id :: atom()

  @typedoc "The name of a topic"
  @type topic :: [atom(), ...]

  @typedoc "A list of topics"
  @type topics :: [topic()]

  @typedoc "The unique event handler identifier"
  @type handler_id :: term()

  @typedoc "The custom payload sent to event handlers"
  @type event :: term()

  @typedoc "A function invoked when publishing events"
  @type handler :: (topic(), event() -> any())

  @typedoc "A tuple used to describe a subscriber"
  @type subscriber :: {handler_id(), topics(), handler()}

  @typedoc "A list of subscribers"
  @type subscribers :: [subscriber()]

  @typedoc "The result of a publish for a single event handler"
  @type result :: {handler_id(), term()}

  @typedoc "The result of a publish"
  @type results :: [result()]

  @typedoc "Option values used by `start_link/1`"
  @type option :: {:name, broker_id()} | {:subscribers, subscribers()}

  @typedoc "Options used by `start_link/1`"
  @type options :: [option()]

  alias Prism.Server

  @doc false
  defmacro __using__(opts) do
    quote location: :keep, bind_quoted: [opts: opts] do
      def child_spec(init_arg) do
        default = %{
          id: __MODULE__,
          start: {__MODULE__, :start_link, [init_arg]}
        }

        Supervisor.child_spec(default, unquote(Macro.escape(opts)))
      end
    end
  end

  ################################
  # Public API
  ################################

  @doc """
  Starts a message broker process.

  ## Options

    * `:name` - The name of the broker process.
    * `:subscribers` - A list of subscribers to start the broker with.

  """
  @spec start_link(options()) :: GenServer.on_start()
  def start_link(opts) when is_list(opts) do
    {broker_id, subscribers} = build_options(opts)
    Server.start_link(broker_id, subscribers)
  end

  @doc """
  Subscribes a handler to a topic or topics within the broker.

  `handler_id` must be unique. If another handler with the same ID already exists
  then `{:error, :already_exists}` is returned. `topic_or_topics` is a list or
  list of lists containing atoms that describe an event occurance. `handler` is
  a 2-arity function that is called when an event is published to a subscribed
  topic.

  Due to how anonymous functions are implemented in the Erlang VM, it is best to
  use function captures (i.e. `&Mod.fun/2`) as handlers to achieve maximum performance.
  In other words, avoid using literal anonymous functions (`fn ... -> ... end`) or
  local function captures (`&handle_event/2`) as handlers.

  Please see `publish/3` for information on publishing to subscribers.

  ## Examples

      iex> Prism.subscribe(:my_broker, "my_handler", [:foo, :bar], &Callback.call/2)
      :ok

      iex> Prism.subscribe(:my_broker, "my_handler", [:foo, :bar], &Callback.call/2)
      {:error, :already_exists}

      iex> Prism.subscribe(:my_broker, "my_other_handler", [[:foo, :bar], [:bar, :baz]], &Callback.call/2)
      :ok

  """
  @spec subscribe(broker_id(), handler_id(), topic() | topics(), handler(), timeout()) ::
          :ok | {:error, reason :: atom()}
  def subscribe(broker_id, handler_id, topic_or_topics, handler, timeout \\ 5_000)

  def subscribe(broker_id, handler_id, [topic_part | _] = topic, handler, timeout)
      when is_atom(topic_part) do
    subscribe(broker_id, handler_id, [topic], handler, timeout)
  end

  def subscribe(broker_id, handler_id, topics, handler, timeout)
      when is_atom(broker_id) and is_list(topics) and is_function(handler, 2) do
    assert_topics(topics)
    GenServer.call(broker_id, {:subscribe, handler_id, topics, handler}, timeout)
  end

  @doc """
  Deletes a handler from the broker.

  Handlers that are deleted will no longer recieve events that are published to
  the broker.

  ## Examples

      iex> Prism.subscribe(:my_broker, "my_handler", [:foo, :bar], &Callback.call/2)
      :ok

      iex> Prism.delete(:my_broker, "my_handler")
      :ok

  """
  @spec delete(broker_id(), handler_id(), timeout()) :: :ok | {:error, term()}
  def delete(broker_id, handler_id, timeout \\ 5_000)
      when is_atom(broker_id) do
    GenServer.call(broker_id, {:delete, handler_id}, timeout)
  end

  @doc """
  Publishes an event to subscribers of the given topic.

  Publishing is done in a syncronous manner within the calling process. If you
  need work to be done "out of bound" of the publisher - its your responsibility
  to implement it in that way. Any exceptions raised by a handler will also propagate
  to the publisher.

  This will return a list of results in the format `{handler_id, result}`, where
  `result` is what is returned by the handler.

  ## Examples

      iex> Prism.subscribe(:my_broker, "my_handler", [:foo, :bar], fn topic, event -> {topic, event} end)
      :ok

      iex> Prism.publish(:my_broker, [:foo, :bar], "hello")
      [{"my_handler", {[:foo, :bar], "hello"}}]

  """
  @spec publish(broker_id(), topic(), event()) :: results()
  def publish(broker_id, topic, event) do
    subscribers = :ets.lookup(broker_id, topic)
    do_publish(subscribers, event, [])
  end

  ################################
  # Private API
  ################################

  defp do_publish([], _payload, results), do: results

  defp do_publish([subscriber | subscribers], event, results) do
    results = do_publish(subscriber, event, results)
    do_publish(subscribers, event, results)
  end

  defp do_publish({topic, {handler_id, handler}}, event, results) do
    result = handler.(topic, event)
    [{handler_id, result} | results]
  end

  defp build_options(opts) do
    broker_id = Keyword.get(opts, :name)
    assert_name(broker_id)
    subscribers = Keyword.get(opts, :subscribers, [])
    assert_subscribers(subscribers)
    {broker_id, subscribers}
  end

  defp assert_name(broker_id) when is_atom(broker_id), do: :ok

  defp assert_name(broker_id) do
    raise ArgumentError, "expected name to be an atom, got: #{inspect(broker_id)}"
  end

  defp assert_subscribers([]), do: :ok

  defp assert_subscribers([{_handler_id, topics, handler} | subscribers])
       when is_list(topics) and is_function(handler, 2) do
    assert_topics(topics)
    assert_subscribers(subscribers)
  end

  defp assert_subscribers([subscriber | _]) do
    raise ArgumentError,
          "expected subscriber to be a tuple in the format {handler_id, topics, handler}, got: #{
            inspect(subscriber)
          }"
  end

  defp assert_topics([]), do: :ok

  defp assert_topics([topic | topics]) do
    assert_topic(topic)
    assert_topics(topics)
  end

  defp assert_topic(topic) do
    unless is_list(topic) and length(topic) > 0 and Enum.all?(topic, &is_atom/1) do
      raise ArgumentError,
            "expected subscriber topic to be a list of atoms, got: #{inspect(topic)}"
    end
  end
end