feat: presence enabled accepted as a valid to connect (#2071)

Author: filipecabaco

lib/realtime_web/channels/realtime_channel.ex

@@ -89,35 +89,13 @@ defmodule RealtimeWeb.RealtimeChannel do
          socket = assign_authorization_context(socket, sub_topic, claims),
          {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant_id),
          {:ok, socket} <- maybe_assign_policies(sub_topic, db_conn, socket),
+         :ok <- can_replay?(params["config"], sub_topic, socket),
          {:ok, replayed_message_ids} <-
            maybe_replay_messages(params["config"], sub_topic, db_conn, tenant_id, socket.assigns.private?) do
       tenant_topic = Tenants.tenant_topic(tenant_id, sub_topic, !socket.assigns.private?)
 
-      # presence.read gate carried in the fastlane metadata so the dispatcher can withhold
-      # presence_diff from members denied presence.read:
-      #   * public channel (no policies) -> true (no presence authorization, always receive diffs)
-      #   * private + presence enabled at join -> the authorized presence.read value (true/false)
-      #   * private + presence not enabled -> nil (read not evaluated yet). The dispatcher routes
-      #     these diffs to the channel process (handle_info) instead of fastlaning, where presence.read
-      #     is consulted at delivery time (it is authorized on-demand when presence is auto-enabled via
-      #     a track message - see PresenceHandler).
-      presence_read? =
-        case socket.assigns.policies do
-          nil -> true
-          %Policies{presence: %{read: read}} -> read
-        end
-
       # fastlane subscription
-      metadata =
-        MessageDispatcher.fastlane_metadata(
-          transport_pid,
-          serializer,
-          topic,
-          log_level,
-          tenant_id,
-          replayed_message_ids,
-          presence_read?
-        )
+      metadata = fastlane_metadata(socket, replayed_message_ids)
 
       RealtimeWeb.Endpoint.subscribe(tenant_topic, metadata: metadata)
       RealtimeWeb.Endpoint.subscribe("realtime:operations:" <> tenant_id, metadata: metadata)
@@ -152,7 +130,9 @@ defmodule RealtimeWeb.RealtimeChannel do
         self_broadcast: Join.self_broadcast?(join),
         tenant_topic: tenant_topic,
         channel_name: sub_topic,
-        presence_enabled?: presence_enabled?
+        presence_enabled?: presence_enabled?,
+        fastlane_metadata: metadata,
+        replayed_message_ids: replayed_message_ids
       }
 
       assigns =
@@ -350,7 +330,10 @@ defmodule RealtimeWeb.RealtimeChannel do
     end
   end
 
-  def handle_info(_msg, %{assigns: %{policies: %Policies{broadcast: %BroadcastPolicies{read: false}}}} = socket) do
+  def handle_info(
+        %{event: "broadcast"},
+        %{assigns: %{policies: %Policies{broadcast: %BroadcastPolicies{read: false}}}} = socket
+      ) do
     Logger.warning("Broadcast message ignored")
     {:noreply, socket}
   end
@@ -572,6 +555,8 @@ defmodule RealtimeWeb.RealtimeChannel do
          {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant_id),
          {:ok, socket} <- maybe_assign_policies(channel_name, db_conn, socket),
          :ok <- check_read_permissions_revoked(previous_policies, socket.assigns.policies) do
+      socket = maybe_resubscribe_fastlane(socket)
+
       Helpers.cancel_timer(pg_sub_ref)
       pg_change_params = Enum.map(pg_change_params, &Map.put(&1, :claims, claims))
 
@@ -989,9 +974,11 @@ defmodule RealtimeWeb.RealtimeChannel do
            ) do
       socket = assign(socket, :policies, policies)
 
-      if match?(%Policies{broadcast: %BroadcastPolicies{read: false}}, socket.assigns.policies),
-        do: {:error, :unauthorized, "You do not have permissions to read from this Channel topic: #{topic}"},
-        else: {:ok, socket}
+      %Policies{broadcast: %{read: broadcast_read?}, presence: %{read: presence_read?}} = socket.assigns.policies
+
+      if broadcast_read? || presence_read?,
+        do: {:ok, socket},
+        else: {:error, :unauthorized, "You do not have permissions to read from this Channel topic: #{topic}"}
     else
       {:error, :increase_connection_pool} ->
         {:error, :increase_connection_pool}
@@ -1011,6 +998,53 @@ defmodule RealtimeWeb.RealtimeChannel do
 
   defp maybe_assign_policies(_, _, socket), do: {:ok, assign(socket, policies: nil)}
 
+  # presence.read gate carried in the fastlane metadata so the dispatcher can withhold
+  # presence_diff from members denied presence.read:
+  #   * public channel (no policies) -> true (no presence authorization, always receive diffs)
+  #   * private + presence enabled at join -> the authorized presence.read value (true/false)
+  #   * private + presence not enabled -> nil (read not evaluated yet). The dispatcher routes
+  #     these diffs to the channel process (handle_info) instead of fastlaning, where presence.read
+  #     is consulted at delivery time (it is authorized on-demand when presence is auto-enabled via
+  #     a track message - see PresenceHandler).
+  defp fastlane_metadata(socket, replayed_message_ids) do
+    %{assigns: %{tenant: tenant_id, log_level: log_level, policies: policies}} = socket
+
+    MessageDispatcher.fastlane_metadata(
+      socket.transport_pid,
+      socket.serializer,
+      socket.topic,
+      log_level,
+      tenant_id,
+      replayed_message_ids,
+      if(policies, do: policies.presence.read, else: true),
+      if(policies, do: policies.broadcast.read, else: true)
+    )
+  end
+
+  defp maybe_resubscribe_fastlane(socket) do
+    %{assigns: %{fastlane_metadata: current, tenant: tenant_id, tenant_topic: tenant_topic}} = socket
+
+    case fastlane_metadata(socket, socket.assigns.replayed_message_ids) do
+      ^current ->
+        socket
+
+      updated ->
+        for pubsub_topic <- [tenant_topic, "realtime:operations:" <> tenant_id] do
+          RealtimeWeb.Endpoint.unsubscribe(pubsub_topic)
+          RealtimeWeb.Endpoint.subscribe(pubsub_topic, metadata: updated)
+        end
+
+        assign(socket, :fastlane_metadata, updated)
+    end
+  end
+
+  defp can_replay?(%{"broadcast" => %{"replay" => _}}, topic, %{
+         assigns: %{policies: %Policies{broadcast: %BroadcastPolicies{read: false}}}
+       }),
+       do: {:error, :unauthorized, "You do not have permissions to read from this Channel topic: #{topic}"}
+
+  defp can_replay?(_config, _topic, _socket), do: :ok
+
   # Detects read permissions that were granted under the previous token but are no longer allowed
   # after re-evaluating the policies with the new token. When that happens we disconnect the channel.
   defp check_read_permissions_revoked(%Policies{} = previous, %Policies{} = current) do

lib/realtime_web/channels/realtime_channel/message_dispatcher.ex

@@ -14,12 +14,15 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcher do
         log_level,
         tenant_id,
         replayed_message_ids \\ MapSet.new(),
-        presence_read? \\ true
+        presence_read? \\ true,
+        broadcast_read? \\ true
       ) do
-    {:rc_fastlane, fastlane_pid, serializer, topic, log_level, tenant_id, replayed_message_ids, presence_read?}
+    {:rc_fastlane, fastlane_pid, serializer, topic, log_level, tenant_id, replayed_message_ids, presence_read?,
+     broadcast_read?}
   end
 
   @presence_diff "presence_diff"
+  @broadcast "broadcast"
 
   @doc """
   This dispatch function caches encoded messages if fastlane is used
@@ -46,22 +49,24 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcher do
         {pid, _}, {cache, count} when pid == from ->
           {cache, count}
 
-        # Subscriber is authorized for broadcast.read but denied presence.read: withhold the
-        # presence_diff. Mirrors the can_read_presence?/1 gate on the presence_state push.
-        {_pid, {:rc_fastlane, _fastlane_pid, _serializer, _join_topic, _log_level, _tenant_id, _replayed, false}},
+        # Subscriber is denied presence.read: withhold the presence_diff. Mirrors the
+        # can_read_presence?/1 gate on the presence_state push.
+        {_pid,
+         {:rc_fastlane, _fastlane_pid, _serializer, _join_topic, _log_level, _tenant_id, _replayed, false, _bcast}},
         {cache, count} ->
           {cache, count}
 
         # presence.read not yet authorized (presence was not enabled at join): route to the channel
         # process so it can consult presence.read at delivery time. Wrapped in a
         # tuple so it is handled by RealtimeChannel.handle_info rather than intercepted and pushed by
         # Phoenix.Channel.Server's built-in %Broadcast{} handling.
-        {pid, {:rc_fastlane, _fastlane_pid, _serializer, _join_topic, _log_level, _tenant_id, _replayed, nil}},
+        {pid, {:rc_fastlane, _fastlane_pid, _serializer, _join_topic, _log_level, _tenant_id, _replayed, nil, _bcast}},
         {cache, count} ->
           send(pid, {:authorize_presence_diff, msg})
           {cache, count}
 
-        {_pid, {:rc_fastlane, fastlane_pid, serializer, join_topic, log_level, tenant_id, _replayed_message_ids, true}},
+        {_pid,
+         {:rc_fastlane, fastlane_pid, serializer, join_topic, log_level, tenant_id, _replayed_message_ids, true, _bcast}},
         {cache, count} ->
           maybe_log(log_level, join_topic, msg, tenant_id)
 
@@ -81,6 +86,7 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcher do
 
   def dispatch(subscribers, from, msg) do
     message_id = message_id(msg)
+    broadcast? = broadcast?(msg)
 
     _ =
       Enum.reduce(subscribers, %{}, fn
@@ -89,10 +95,9 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcher do
 
         {pid,
          {:rc_fastlane, fastlane_pid, serializer, join_topic, log_level, tenant_id, replayed_message_ids,
-          _presence_read?}},
+          _presence_read?, broadcast_read?}},
         cache ->
-          if already_replayed?(message_id, replayed_message_ids) do
-            # skip already replayed message
+          if (broadcast? and broadcast_read? != true) or already_replayed?(message_id, replayed_message_ids) do
             cache
           else
             send(pid, :update_rate_counter)
@@ -110,6 +115,10 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcher do
     :ok
   end
 
+  defp broadcast?(%UserBroadcast{}), do: true
+  defp broadcast?(%Broadcast{event: @broadcast}), do: true
+  defp broadcast?(_msg), do: false
+
   defp maybe_log(:info, join_topic, msg, tenant_id) when is_struct(msg) do
     log = "Received message on #{join_topic} with payload: #{inspect(msg, pretty: true)}"
     Logger.info(log, external_id: tenant_id, project: tenant_id)
@@ -160,7 +169,7 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcher do
 
   defp fastlane!(serializer, msg), do: {:ok, serializer.fastlane!(msg)}
 
-  defp tenant_id([{_pid, {:rc_fastlane, _, _, _, _, tenant_id, _, _}} | _]), do: tenant_id
+  defp tenant_id([{_pid, {:rc_fastlane, _, _, _, _, tenant_id, _, _, _}} | _]), do: tenant_id
   defp tenant_id(_), do: nil
 
   defp increment_presence_counter(tenant_id, "presence_diff", count) when is_binary(tenant_id) do