phoenix_pubsub

defmodule Phoenix.PubSub do
  @moduledoc """
  Realtime Publisher/Subscriber service.

  ## Getting started

  You start Phoenix.PubSub directly in your supervision
  tree:

      {Phoenix.PubSub, name: :my_pubsub}

  You can now use the functions in this module to subscribe
  and broadcast messages:

      iex> alias Phoenix.PubSub
      iex> PubSub.subscribe(:my_pubsub, "user:123")
      :ok
      iex> Process.info(self(), :messages)
      {:messages, []}
      iex> PubSub.broadcast(:my_pubsub, "user:123", {:user_update, %{id: 123, name: "Shane"}})
      :ok
      iex> Process.info(self(), :messages)
      {:messages, [{:user_update, %{id: 123, name: "Shane"}}]}

  ## Adapters

  Phoenix PubSub was designed to be flexible and support
  multiple backends. There are two officially supported
  backends:

    * `Phoenix.PubSub.PG2` - the default adapter that ships
      as part of Phoenix.PubSub. It uses Distributed Elixir,
      directly exchanging notifications between servers.
      It supports a `:pool_size` option to be given alongside
      the name, defaults to `1`. Note the `:pool_size` must
      be the same throughout the cluster, therefore don't
      configure the pool size based on `System.schedulers_online/0`,
      especially if you are using machines with different specs.

    * `Phoenix.PubSub.Redis` - uses Redis to exchange data between
      servers. It requires the `:phoenix_pubsub_redis` dependency.

  See `Phoenix.PubSub.Adapter` to implement a custom adapter.

  ## Safe pool size migration (when using `Phoenix.PubSub.PG2` adapter)

  When you need to change the pool size in a running cluster,
  you can use the `broadcast_pool_size` option to ensure no
  messages are lost during deployment. This is particularly
  important when increasing the pool size.

  Here's how to safely increase the pool size from 1 to 2:

  1. Initial state - Current configuration with `pool_size: 1`:
  ```
  {Phoenix.PubSub, name: :my_pubsub, pool_size: 1}
  ```

  ```mermaid
  graph TD
      subgraph "Initial State"
          subgraph "Node 1"
              A1[Shard 1<br/>Broadcast & Receive]
          end
          subgraph "Node 2"
              B1[Shard 1<br/>Broadcast & Receive]
          end
          A1 <--> B1
      end
  ```

  2. First deployment - Set the new pool size but keep broadcasting on the old size:
  ```
  {Phoenix.PubSub, name: :my_pubsub, pool_size: 2, broadcast_pool_size: 1}
  ```

  ```mermaid
  graph TD
      subgraph "First Deployment"
          subgraph "Node 1"
              A1[Shard 1<br/>Broadcast & Receive]
              A2[Shard 2<br/>Broadcast & Receive]
          end
          subgraph "Node 2"
              B1[Shard 1<br/>Broadcast & Receive]
              B2[Shard 2<br/>Receive Only]
          end
          A1 <--> B1
          A2 --> B2
      end
  ```

  3. Final deployment - All nodes running with new pool size:
  ```
  {Phoenix.PubSub, name: :my_pubsub, pool_size: 2}
  ```

  ```mermaid
  graph TD
      subgraph "Final State"
          subgraph "Node 1"
              A1[Shard 1<br/>Broadcast & Receive]
              A2[Shard 2<br/>Broadcast & Receive]
          end
          subgraph "Node 2"
              B1[Shard 1<br/>Broadcast & Receive]
              B2[Shard 2<br/>Broadcast & Receive]
          end
          A1 <--> B1
          A2 <--> B2
      end
  ```

  This two-step process ensures that:
  - All nodes can receive messages from both old and new pool sizes
  - No messages are lost during the transition
  - The cluster remains fully functional throughout the deployment

  To decrease the pool size, follow the same process in reverse order.

  """

  @type node_name :: atom | binary
  @type t :: atom
  @type topic :: binary
  @type message :: term
  @type dispatcher :: module

  defmodule BroadcastError do
    defexception [:message]

    def exception(msg) do
      %BroadcastError{message: "broadcast failed with #{inspect(msg)}"}
    end
  end

  @doc """
  Returns a child specification for pubsub with the given `options`.

  The `:name` is required as part of `options`. The remaining options
  are described below.

  ## Options

    * `:name` - the name of the pubsub to be started
    * `:adapter` - the adapter to use (defaults to `Phoenix.PubSub.PG2`)
    * `:pool_size` - number of pubsub partitions to launch
      (defaults to one partition for every 4 cores)
    * `:registry_size` - number of `Registry` partitions to launch
      (defaults to `:pool_size`). This controls the number of Registry partitions
      used for storing subscriptions and can be tuned independently from `:pool_size`
      for better performance characteristics.
    * `:broadcast_pool_size` - number of pubsub partitions used for broadcasting messages
      (defaults to `:pool_size`). This option is used during pool size migrations to ensure
      no messages are lost. See the "Safe Pool Size Migration" section in the module documentation.
    * `:dispatcher` - the default dispatcher module for broadcasts
      (defaults to `Phoenix.PubSub`). Can be overridden per-call by
      passing a dispatcher to `broadcast/4` and friends.
    * `:group_by` - controls how the underlying `Registry` partitions
      subscriptions, either `:pid` or `:key` (defaults to `:pid`). With
      `:pid`, entries are grouped by subscriber pid — best when topics
      have many subscribers each. With `:key`, entries are grouped by
      topic so key-based lookups touch a single partition — best when
      there are many topics with few subscribers each. `:key` requires
      Elixir v1.19 or later. See `Registry.start_link/1` for the
      underlying trade-offs.

  """
  @spec child_spec(keyword) :: Supervisor.child_spec()
  defdelegate child_spec(options), to: Phoenix.PubSub.Supervisor

  @doc """
  Subscribes the caller to the PubSub adapter's topic.

    * `pubsub` - The name of the pubsub system
    * `topic` - The topic to subscribe to, for example: `"users:123"`
    * `opts` - The optional list of options. See below.

  ## Duplicate Subscriptions

  Callers should only subscribe to a given topic a single time.
  Duplicate subscriptions for a Pid/topic pair are allowed and
  will cause duplicate events to be sent; however, when using
  `Phoenix.PubSub.unsubscribe/2`, all duplicate subscriptions
  will be dropped.

  ## Options

    * `:sender` - sets a custom `{module, metadata}` tuple that
      is called when broadcasting a message. See `Phoenix.PubSub.Sender`
      for more information

    * `:metadata` - DEPRECATED. Use `:sender` instead.
      provides metadata to be attached to this
      subscription. The metadata can be used by custom
      dispatching mechanisms. Metadata in the shape
      of `[atom | term]` is reserved for senders and raises

  """
  @spec subscribe(t, topic, keyword) :: :ok | {:error, term}
  def subscribe(pubsub, topic, opts \\ [])
      when is_atom(pubsub) and is_binary(topic) and is_list(opts) do
    meta =
      case opts[:sender] do
        {mod, term} when is_atom(mod) ->
          [mod | term]

        _ ->
          # TODO: Deprecate me
          case opts[:metadata] do
            [mod | _] when is_atom(mod) ->
              raise "passing metadata in the shape of [module | term] to subscribe is unsupported"

            other ->
              other
          end
      end

    case Registry.register(pubsub, topic, meta) do
      {:ok, _} -> :ok
      {:error, _} = error -> error
    end
  end

  @doc """
  Unsubscribes the caller from the PubSub adapter's topic.
  """
  @spec unsubscribe(t, topic) :: :ok
  def unsubscribe(pubsub, topic) when is_atom(pubsub) and is_binary(topic) do
    Registry.unregister(pubsub, topic)
  end

  @doc """
  Unsubscribes the caller from the PubSub adapter's topic taking the metadata into consideration.

  Unlike `unsubscribe/2`, this function matches on the metadata provided as an option when subscribed.
  This is useful when you have multiple subscriptions for the same topic with different metadata.

  ## Example

      iex> PubSub.subscribe(:my_pubsub, "users:123", metadata: :fast)
      :ok
      iex> PubSub.subscribe(:my_pubsub, "users:123", metadata: :slow)
      :ok
      iex> PubSub.unsubscribe_match(:my_pubsub, "users:123", :fast)
      :ok
      # Only the :fast subscription is removed, :slow remains active

  """
  @doc deprecated:
         "Using custom metadata when subscribing is deprecated, use the `:sender` option of `subscribe/3` instead. See `Phoenix.PubSub.Sender`."
  @spec unsubscribe_match(t, topic, term) :: :ok
  def unsubscribe_match(pubsub, topic, metadata) when is_atom(pubsub) and is_binary(topic) do
    Registry.unregister_match(pubsub, topic, metadata)
  end

  @doc """
  Unsubscribes the caller from the PubSub adapter's topic taking the sender into consideration.

  Accepts a `{module, metadata}` tuple, as passed to the `:sender` option of
  `subscribe/3`. Removes all of the caller's subscriptions for the topic whose
  sender module and metadata match, preserving other subscriptions.

  The metadata follows the same pattern-matching rules as `unsubscribe_match/3`.
  For example, `{MySender, :_}` matches any metadata for `MySender`.

  ## Example

      iex> PubSub.subscribe(:my_pubsub, "users:123", sender: {MySender, :fast})
      :ok
      iex> PubSub.subscribe(:my_pubsub, "users:123", sender: {MySender, :slow})
      :ok
      iex> PubSub.unsubscribe_sender(:my_pubsub, "users:123", {MySender, :fast})
      :ok
      # Only the :fast subscription is removed, :slow remains active

  """
  @spec unsubscribe_sender(t, topic, {module(), term()}) :: :ok
  def unsubscribe_sender(pubsub, topic, {module, metadata})
      when is_atom(pubsub) and is_binary(topic) and is_atom(module) do
    unsubscribe_match(pubsub, topic, [module | metadata])
  end

  @doc """
  Broadcasts message on given topic across the whole cluster.

    * `pubsub` - The name of the pubsub system
    * `topic` - The topic to broadcast to, ie: `"users:123"`
    * `message` - The payload of the broadcast

  """
  @spec broadcast(t, topic, message) :: :ok | {:error, term}
  def broadcast(pubsub, topic, message)
      when is_atom(pubsub) and is_binary(topic) do
    {:ok, {adapter, name, dispatcher}} = Registry.meta(pubsub, :pubsub)

    with :ok <- adapter.broadcast(name, topic, message, dispatcher) do
      dispatch(pubsub, :none, topic, message, dispatcher)
    end
  end

  @doc """
  Broadcasts message on given topic across the whole cluster using a custom dispatcher.
  """
  @doc deprecated:
         "Custom dispatchers are deprecated, use the `:sender` option of `subscribe/3` instead. See `Phoenix.PubSub.Sender`."
  @spec broadcast(t, topic, message, dispatcher) :: :ok | {:error, term}
  def broadcast(pubsub, topic, message, dispatcher)
      when is_atom(pubsub) and is_binary(topic) and is_atom(dispatcher) do
    {:ok, {adapter, name, default_dispatcher}} = Registry.meta(pubsub, :pubsub)
    dispatcher = dispatcher || default_dispatcher

    with :ok <- adapter.broadcast(name, topic, message, dispatcher) do
      dispatch(pubsub, :none, topic, message, dispatcher)
    end
  end

  @doc """
  Broadcasts message on given topic from the given process across the whole cluster.

    * `pubsub` - The name of the pubsub system
    * `from` - The pid that will send the message
    * `topic` - The topic to broadcast to, ie: `"users:123"`
    * `message` - The payload of the broadcast

  The default dispatcher will broadcast the message to all subscribers except for the
  process that initiated the broadcast.
  """
  @spec broadcast_from(t, pid, topic, message) :: :ok | {:error, term}
  def broadcast_from(pubsub, from, topic, message)
      when is_atom(pubsub) and is_pid(from) and is_binary(topic) do
    {:ok, {adapter, name, dispatcher}} = Registry.meta(pubsub, :pubsub)

    with :ok <- adapter.broadcast(name, topic, message, dispatcher) do
      dispatch(pubsub, from, topic, message, dispatcher)
    end
  end

  @doc """
  Broadcasts message on given topic from the given process across the whole cluster
  using a custom dispatcher.
  """
  @doc deprecated:
         "Custom dispatchers are deprecated, use the `:sender` option of `subscribe/3` instead. See `Phoenix.PubSub.Sender`."
  @spec broadcast_from(t, pid, topic, message, dispatcher) :: :ok | {:error, term}
  def broadcast_from(pubsub, from, topic, message, dispatcher)
      when is_atom(pubsub) and is_pid(from) and is_binary(topic) and is_atom(dispatcher) do
    {:ok, {adapter, name, default_dispatcher}} = Registry.meta(pubsub, :pubsub)
    dispatcher = dispatcher || default_dispatcher

    with :ok <- adapter.broadcast(name, topic, message, dispatcher) do
      dispatch(pubsub, from, topic, message, dispatcher)
    end
  end

  @doc """
  Broadcasts message on given topic only for the current node.

    * `pubsub` - The name of the pubsub system
    * `topic` - The topic to broadcast to, ie: `"users:123"`
    * `message` - The payload of the broadcast

  """
  @spec local_broadcast(t, topic, message) :: :ok
  def local_broadcast(pubsub, topic, message)
      when is_atom(pubsub) and is_binary(topic) do
    {:ok, {_adapter, _name, dispatcher}} = Registry.meta(pubsub, :pubsub)
    dispatch(pubsub, :none, topic, message, dispatcher)
  end

  @doc """
  Broadcasts message on given topic only for the current node
  with a custom dispatcher.
  """
  @doc deprecated:
         "Custom dispatchers are deprecated, use the `:sender` option of `subscribe/3` instead. See `Phoenix.PubSub.Sender`."
  @spec local_broadcast(t, topic, message, dispatcher) :: :ok
  def local_broadcast(pubsub, topic, message, dispatcher)
      when is_atom(pubsub) and is_binary(topic) and is_atom(dispatcher) do
    {:ok, {_adapter, _name, default_dispatcher}} = Registry.meta(pubsub, :pubsub)
    dispatch(pubsub, :none, topic, message, dispatcher || default_dispatcher)
  end

  @doc """
  Broadcasts message on given topic from a given process only for the current node.

    * `pubsub` - The name of the pubsub system
    * `from` - The pid that will send the message
    * `topic` - The topic to broadcast to, ie: `"users:123"`
    * `message` - The payload of the broadcast

  The default dispatcher will broadcast the message to all subscribers except for the
  process that initiated the broadcast.
  """
  @spec local_broadcast_from(t, pid, topic, message) :: :ok
  def local_broadcast_from(pubsub, from, topic, message)
      when is_atom(pubsub) and is_pid(from) and is_binary(topic) do
    {:ok, {_adapter, _name, dispatcher}} = Registry.meta(pubsub, :pubsub)
    dispatch(pubsub, from, topic, message, dispatcher)
  end

  @doc """
  Broadcasts message on given topic from a given process only for the current node
  using a custom dispatcher.
  """
  @doc deprecated:
         "Custom dispatchers are deprecated, use the `:sender` option of `subscribe/3` instead. See `Phoenix.PubSub.Sender`."
  @spec local_broadcast_from(t, pid, topic, message, dispatcher) :: :ok
  def local_broadcast_from(pubsub, from, topic, message, dispatcher)
      when is_atom(pubsub) and is_pid(from) and is_binary(topic) and is_atom(dispatcher) do
    {:ok, {_adapter, _name, default_dispatcher}} = Registry.meta(pubsub, :pubsub)
    dispatch(pubsub, from, topic, message, dispatcher || default_dispatcher)
  end

  @doc """
  Broadcasts message on given topic to a given node.

    * `node_name` - The target node name
    * `pubsub` - The name of the pubsub system
    * `topic` - The topic to broadcast to, ie: `"users:123"`
    * `message` - The payload of the broadcast

  **DO NOT** use this function if you wish to broadcast to the current
  node, as it is always serialized, use `local_broadcast/3` instead.
  """
  @spec direct_broadcast(node_name, t, topic, message) :: :ok | {:error, term}
  def direct_broadcast(node_name, pubsub, topic, message)
      when is_atom(pubsub) and is_binary(topic) do
    {:ok, {adapter, name, dispatcher}} = Registry.meta(pubsub, :pubsub)
    adapter.direct_broadcast(name, node_name, topic, message, dispatcher)
  end

  @doc """
  Broadcasts message on given topic to a given node using a custom dispatcher.
  """
  @doc deprecated:
         "Custom dispatchers are deprecated, use the `:sender` option of `subscribe/3` instead. See `Phoenix.PubSub.Sender`."
  @spec direct_broadcast(node_name, t, topic, message, dispatcher) :: :ok | {:error, term}
  def direct_broadcast(node_name, pubsub, topic, message, dispatcher)
      when is_atom(pubsub) and is_binary(topic) and is_atom(dispatcher) do
    {:ok, {adapter, name, default_dispatcher}} = Registry.meta(pubsub, :pubsub)
    adapter.direct_broadcast(name, node_name, topic, message, dispatcher || default_dispatcher)
  end

  @doc """
  Raising version of `broadcast/3`.
  """
  @spec broadcast!(t, topic, message) :: :ok
  def broadcast!(pubsub, topic, message) do
    case broadcast(pubsub, topic, message) do
      :ok -> :ok
      {:error, error} -> raise BroadcastError, "broadcast failed: #{inspect(error)}"
    end
  end

  @doc """
  Raising version of `broadcast/4`.
  """
  @doc deprecated:
         "Custom dispatchers are deprecated, use the `:sender` option of `subscribe/3` instead. See `Phoenix.PubSub.Sender`."
  @spec broadcast!(t, topic, message, dispatcher) :: :ok
  def broadcast!(pubsub, topic, message, dispatcher) do
    case broadcast(pubsub, topic, message, dispatcher) do
      :ok -> :ok
      {:error, error} -> raise BroadcastError, "broadcast failed: #{inspect(error)}"
    end
  end

  @doc """
  Raising version of `broadcast_from/4`.
  """
  @spec broadcast_from!(t, pid, topic, message) :: :ok
  def broadcast_from!(pubsub, from, topic, message) do
    case broadcast_from(pubsub, from, topic, message) do
      :ok -> :ok
      {:error, error} -> raise BroadcastError, "broadcast failed: #{inspect(error)}"
    end
  end

  @doc """
  Raising version of `broadcast_from/5`.
  """
  @doc deprecated:
         "Custom dispatchers are deprecated, use the `:sender` option of `subscribe/3` instead. See `Phoenix.PubSub.Sender`."
  @spec broadcast_from!(t, pid, topic, message, dispatcher) :: :ok
  def broadcast_from!(pubsub, from, topic, message, dispatcher) do
    case broadcast_from(pubsub, from, topic, message, dispatcher) do
      :ok -> :ok
      {:error, error} -> raise BroadcastError, "broadcast failed: #{inspect(error)}"
    end
  end

  @doc """
  Raising version of `direct_broadcast/4`.
  """
  @spec direct_broadcast!(node_name, t, topic, message) :: :ok
  def direct_broadcast!(node_name, pubsub, topic, message) do
    case direct_broadcast(node_name, pubsub, topic, message) do
      :ok -> :ok
      {:error, error} -> raise BroadcastError, "broadcast failed: #{inspect(error)}"
    end
  end

  @doc """
  Raising version of `direct_broadcast/5`.
  """
  @doc deprecated:
         "Custom dispatchers are deprecated, use the `:sender` option of `subscribe/3` instead. See `Phoenix.PubSub.Sender`."
  @spec direct_broadcast!(node_name, t, topic, message, dispatcher) :: :ok
  def direct_broadcast!(node_name, pubsub, topic, message, dispatcher) do
    case direct_broadcast(node_name, pubsub, topic, message, dispatcher) do
      :ok -> :ok
      {:error, error} -> raise BroadcastError, "broadcast failed: #{inspect(error)}"
    end
  end

  @doc """
  Returns the node name of the PubSub server.
  """
  @spec node_name(t) :: node_name
  def node_name(pubsub) do
    {:ok, {adapter, name, _dispatcher}} = Registry.meta(pubsub, :pubsub)
    adapter.node_name(name)
  end

  ## Dispatch callback

  @doc false
  def dispatch(entries, :none, message) do
    states =
      for {pid, metadata} <- entries, reduce: %{} do
        acc ->
          dispatch_with_optional_sender(pid, metadata, message, acc)
      end

    Enum.each(states, &finalize_sender/1)
  end

  def dispatch(entries, from, message) do
    states =
      for {pid, metadata} <- entries, pid != from, reduce: %{} do
        acc ->
          dispatch_with_optional_sender(pid, metadata, message, acc)
      end

    Enum.each(states, &finalize_sender/1)
  end

  defp dispatch(pubsub, from, topic, message, dispatcher) do
    Registry.dispatch(pubsub, topic, {dispatcher, :dispatch, [from, message]})
    :ok
  end

  defp dispatch_with_optional_sender(pid, [module | meta], message, acc) when is_atom(module) do
    state = Map.get(acc, module, nil)
    new_state = module.send(pid, meta, message, state)
    Map.put(acc, module, new_state)
  end

  defp dispatch_with_optional_sender(pid, _metadata, message, acc) do
    send(pid, message)
    acc
  end

  defp finalize_sender({module, state}) do
    if function_exported?(module, :finalize, 1), do: module.finalize(state)
  end
end