tackle rejoin

Author: filipecabaco

lib/realtime_web/channels/realtime_channel.ex

@@ -94,38 +94,8 @@ defmodule RealtimeWeb.RealtimeChannel do
            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
-
-      broadcast_read? =
-        case socket.assigns.policies do
-          nil -> true
-          %Policies{broadcast: %{read: read}} -> read
-        end
-
       # fastlane subscription
-      metadata =
-        MessageDispatcher.fastlane_metadata(
-          transport_pid,
-          serializer,
-          topic,
-          log_level,
-          tenant_id,
-          replayed_message_ids,
-          presence_read?,
-          broadcast_read?
-        )
+      metadata = fastlane_metadata(socket, replayed_message_ids)
 
       RealtimeWeb.Endpoint.subscribe(tenant_topic, metadata: metadata)
       RealtimeWeb.Endpoint.subscribe("realtime:operations:" <> tenant_id, metadata: metadata)
@@ -160,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 =
@@ -583,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))
 
@@ -1024,6 +998,46 @@ 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}}}
        }),

test/integration/rt_channel/presence_test.exs

@@ -134,7 +134,7 @@ defmodule Realtime.Integration.RtChannel.PresenceTest do
       {service_role_socket, _} = get_connection(tenant, serializer, role: "service_role")
       topic = "realtime:#{topic}"
 
-      WebsocketClient.join(socket, topic, %{config: %{presence: %{key: "", enabled: true}, private: true}})
+      WebsocketClient.join(socket, topic, %{config: %{presence: %{key: "authenticated", enabled: true}, private: true}})
       assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}, topic: ^topic}, 500
       assert_receive %Message{event: "presence_state", payload: %{}, topic: ^topic}, 500
 
@@ -145,7 +145,7 @@ defmodule Realtime.Integration.RtChannel.PresenceTest do
       WebsocketClient.send_event(socket, topic, "presence", %{type: "presence", event: "TRACK", payload: payload})
 
       assert_receive %Message{event: "presence_diff", payload: %{"joins" => joins, "leaves" => %{}}, topic: ^topic}, 500
-      join_payload = joins |> Map.values() |> hd() |> get_in(["metas"]) |> hd()
+      join_payload = joins |> get_in(["authenticated", "metas"]) |> hd()
       assert get_in(join_payload, ["name"]) == payload.name
 
       broadcast = %{"event" => "TEST", "payload" => %{"msg" => 1}, "type" => "broadcast"}
@@ -490,6 +490,55 @@ defmodule Realtime.Integration.RtChannel.PresenceTest do
 
       refute_receive %Message{event: "phx_close", topic: ^realtime_topic}, 500
     end
+
+    @tag policies: [
+           :authenticated_read_presence,
+           :authenticated_write_presence,
+           :authenticated_read_broadcast_based_on_claim
+         ]
+    test "starts receiving broadcasts when broadcast read permission is granted on new access_token",
+         %{tenant: tenant, topic: topic, serializer: serializer} do
+      parent = self()
+      publisher_inbox = spawn_link(fn -> forward_frames(parent, :publisher) end)
+      {:ok, publisher_token} = token_valid(tenant, "service_role")
+
+      {:ok, publisher} =
+        WebsocketClient.connect(publisher_inbox, uri(tenant, serializer), serializer, [
+          {"x-api-key", publisher_token}
+        ])
+
+      # Token whose claims do not satisfy the broadcast read policy, so the join is allowed on presence alone
+      {socket, _} = get_connection(tenant, serializer, role: "authenticated", claims: %{broadcast_read: false})
+      realtime_topic = "realtime:#{topic}"
+
+      WebsocketClient.join(socket, realtime_topic, %{
+        config: %{presence: %{key: "authenticated", enabled: true}, private: true}
+      })
+
+      assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}, topic: ^realtime_topic}, 500
+      assert_receive %Message{event: "presence_state", topic: ^realtime_topic}, 500
+
+      WebsocketClient.join(publisher, realtime_topic, %{config: %{private: true}})
+      assert_receive {:publisher, %Message{event: "phx_reply", payload: %{"status" => "ok"}}}, 500
+
+      broadcast = %{"event" => "TEST", "payload" => %{"msg" => 1}, "type" => "broadcast"}
+      WebsocketClient.send_event(publisher, realtime_topic, "broadcast", broadcast)
+      refute_receive %Message{event: "broadcast", topic: ^realtime_topic}, 500
+
+      # New token whose claims satisfy the broadcast read policy
+      {:ok, new_token} =
+        generate_token(tenant, %{
+          exp: System.system_time(:second) + 1000,
+          role: "authenticated",
+          broadcast_read: true
+        })
+
+      WebsocketClient.send_event(socket, realtime_topic, "access_token", %{"access_token" => new_token})
+      refute_receive %Message{event: "phx_close", topic: ^realtime_topic}, 500
+
+      WebsocketClient.send_event(publisher, realtime_topic, "broadcast", broadcast)
+      assert_receive %Message{event: "broadcast", payload: ^broadcast, topic: ^realtime_topic}, 500
+    end
   end
 
   describe "database connection errors" do

test/support/generators.ex

@@ -272,6 +272,19 @@ defmodule Generators do
     """
   end
 
+  def policy_query(:authenticated_read_broadcast_based_on_claim, %{topic: name}) do
+    """
+    CREATE POLICY "authenticated_read_broadcast_claim_#{name}"
+    ON realtime.messages FOR SELECT
+    TO authenticated
+    USING (
+      realtime.topic() = '#{name}'
+      AND realtime.messages.extension = 'broadcast'
+      AND coalesce(((current_setting('request.jwt.claims', true))::jsonb ->> 'broadcast_read')::boolean, false)
+    );
+    """
+  end
+
   def policy_query(:broken_read_presence, _) do
     """
     CREATE POLICY "authenticated_read_presence"