feat: parse presence enabled from join params

lib/realtime_web/channels/realtime_channel.ex

@@ -68,6 +68,13 @@ defmodule RealtimeWeb.RealtimeChannel do
       Phoenix.PubSub.subscribe(Realtime.PubSub, "realtime:operations:" <> tenant_id)
 
       is_new_api = new_api?(params)
+      # TODO: Default will be moved to false in the future
+      presence_enabled? =
+        case get_in(params, ["config", "presence", "enabled"]) do
+          enabled when is_boolean(enabled) -> enabled
+          _ -> true
+        end
+
       pg_change_params = pg_change_params(is_new_api, params, channel_pid, claims, sub_topic)
 
       opts = %{
@@ -93,11 +100,13 @@ defmodule RealtimeWeb.RealtimeChannel do
         presence_key: presence_key(params),
         self_broadcast: !!params["config"]["broadcast"]["self"],
         tenant_topic: tenant_topic,
-        channel_name: sub_topic
+        channel_name: sub_topic,
+        presence_enabled?: presence_enabled?
       }
 
-      # Start presence and add user
-      send(self(), :sync_presence)
+      # Start presence and add user if presence is enabled
+      if presence_enabled?, do: send(self(), :sync_presence)
+
       Realtime.UsersCounter.add(transport_pid, tenant_id)
       SocketDisconnect.add(tenant_id, socket)
 
@@ -318,7 +327,9 @@ defmodule RealtimeWeb.RealtimeChannel do
     {:stop, :shutdown, socket}
   end
 
-  def handle_info(:sync_presence, socket), do: PresenceHandler.sync(socket)
+  def handle_info(:sync_presence, %{assigns: %{presence_enabled?: true}} = socket), do: PresenceHandler.sync(socket)
+  def handle_info(:sync_presence, socket), do: {:noreply, socket}
+
   def handle_info(:unsuspend_tenant, socket), do: {:noreply, socket}
 
   def handle_info(msg, socket) do

lib/realtime_web/channels/realtime_channel/presence_handler.ex

@@ -18,9 +18,12 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandler do
   alias RealtimeWeb.RealtimeChannel.Logging
 
   @spec handle(map(), Socket.t()) :: {:reply, :error | :ok, Socket.t()}
+  def handle(_, %{assigns: %{presence_enabled?: false}} = socket), do: {:reply, :ok, socket}
   def handle(payload, %{assigns: %{private?: false}} = socket), do: handle(payload, nil, socket)
 
   @spec handle(map(), pid() | nil, Socket.t()) :: {:reply, :error | :ok, Socket.t()}
+  def handle(_, _, %{assigns: %{presence_enabled?: false}} = socket), do: {:reply, :ok, socket}
+
   def handle(%{"event" => event} = payload, db_conn, socket) do
     event = String.downcase(event, :ascii)
 
@@ -36,6 +39,8 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandler do
   Sends presence state to connected clients
   """
   @spec sync(Socket.t()) :: {:noreply, Socket.t()}
+  def sync(%{assigns: %{presence_enabled?: false}} = socket), do: {:noreply, socket}
+
   def sync(%{assigns: %{private?: false}} = socket) do
     %{assigns: %{tenant_topic: topic}} = socket
     socket = count(socket)

mix.exs

@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
   def project do
     [
       app: :realtime,
-      version: "2.57.11",
+      version: "2.58.0",
       elixir: "~> 1.17.3",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,

test/integration/rt_channel_test.exs

@@ -661,7 +661,7 @@ defmodule Realtime.Integration.RtChannelTest do
 
     test "public presence", %{tenant: tenant} do
       {socket, _} = get_connection(tenant)
-      config = %{presence: %{key: ""}, private: false}
+      config = %{presence: %{key: "", enabled: true}, private: false}
       topic = "realtime:any"
 
       WebsocketClient.join(socket, topic, %{config: config})
@@ -689,7 +689,7 @@ defmodule Realtime.Integration.RtChannelTest do
     test "private presence with read and write permissions will be able to track and receive presence changes",
          %{tenant: tenant, topic: topic} do
       {socket, _} = get_connection(tenant, "authenticated")
-      config = %{presence: %{key: ""}, private: true}
+      config = %{presence: %{key: "", enabled: true}, private: true}
       topic = "realtime:#{topic}"
 
       WebsocketClient.join(socket, topic, %{config: config})
@@ -714,7 +714,7 @@ defmodule Realtime.Integration.RtChannelTest do
     test "private presence with read and write permissions will be able to track and receive presence changes using a remote node",
          %{tenant: tenant, topic: topic} do
       {socket, _} = get_connection(tenant, "authenticated")
-      config = %{presence: %{key: ""}, private: true}
+      config = %{presence: %{key: "", enabled: true}, private: true}
       topic = "realtime:#{topic}"
 
       WebsocketClient.join(socket, topic, %{config: config})
@@ -739,7 +739,7 @@ defmodule Realtime.Integration.RtChannelTest do
          %{tenant: tenant, topic: topic} do
       {socket, _} = get_connection(tenant, "authenticated")
       {secondary_socket, _} = get_connection(tenant, "service_role")
-      config = fn key -> %{presence: %{key: key}, private: true} end
+      config = fn key -> %{presence: %{key: key, enabled: true}, private: true} end
       topic = "realtime:#{topic}"
 
       WebsocketClient.join(socket, topic, %{config: config.("authenticated")})
@@ -791,7 +791,7 @@ defmodule Realtime.Integration.RtChannelTest do
     test "handles lack of connection to database error on private channels", %{tenant: tenant, topic: topic} do
       topic = "realtime:#{topic}"
       {socket, _} = get_connection(tenant, "authenticated")
-      WebsocketClient.join(socket, topic, %{config: %{private: true}})
+      WebsocketClient.join(socket, topic, %{config: %{private: true, presence: %{enabled: true}}})
       assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}, topic: ^topic}, 300
       assert_receive %Message{event: "presence_state"}
 
@@ -812,7 +812,7 @@ defmodule Realtime.Integration.RtChannelTest do
     test "lack of connection to database error does not impact public channels", %{tenant: tenant, topic: topic} do
       topic = "realtime:#{topic}"
       {socket, _} = get_connection(tenant, "authenticated")
-      WebsocketClient.join(socket, topic, %{config: %{private: false}})
+      WebsocketClient.join(socket, topic, %{config: %{private: false, presence: %{enabled: true}}})
       assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}, topic: ^topic}, 300
       assert_receive %Message{event: "presence_state"}
 
@@ -828,6 +828,58 @@ defmodule Realtime.Integration.RtChannelTest do
 
       refute log =~ "UnableToHandlePresence"
     end
+
+    @tag policies: [:authenticated_read_broadcast_and_presence, :authenticated_write_broadcast_and_presence]
+
+    test "presence enabled if param enabled is set in configuration for private channels", %{
+      tenant: tenant,
+      topic: topic
+    } do
+      {socket, _} = get_connection(tenant, "authenticated")
+      topic = "realtime:#{topic}"
+
+      WebsocketClient.join(socket, topic, %{config: %{private: true, presence: %{enabled: true}}})
+      assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}}, 500
+      assert_receive %Message{event: "presence_state"}, 500
+    end
+
+    @tag policies: [:authenticated_read_broadcast_and_presence, :authenticated_write_broadcast_and_presence]
+
+    test "presence disabled if param 'enabled' is set to false in configuration for private channels", %{
+      tenant: tenant,
+      topic: topic
+    } do
+      {socket, _} = get_connection(tenant, "authenticated")
+      topic = "realtime:#{topic}"
+
+      WebsocketClient.join(socket, topic, %{config: %{private: true, presence: %{enabled: false}}})
+      assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}}, 500
+      refute_receive %Message{event: "presence_state"}, 500
+    end
+
+    test "presence enabled if param enabled is set in configuration for public channels", %{
+      tenant: tenant,
+      topic: topic
+    } do
+      {socket, _} = get_connection(tenant, "authenticated")
+      topic = "realtime:#{topic}"
+
+      WebsocketClient.join(socket, topic, %{config: %{private: false, presence: %{enabled: true}}})
+      assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}}, 500
+      assert_receive %Message{event: "presence_state"}, 500
+    end
+
+    test "presence disabled if param 'enabled' is set to false in configuration for public channels", %{
+      tenant: tenant,
+      topic: topic
+    } do
+      {socket, _} = get_connection(tenant, "authenticated")
+      topic = "realtime:#{topic}"
+
+      WebsocketClient.join(socket, topic, %{config: %{private: false, presence: %{enabled: false}}})
+      assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}}, 500
+      refute_receive %Message{event: "presence_state"}, 500
+    end
   end
 
   describe "token handling" do

test/realtime_web/channels/realtime_channel/presence_handler_test.exs

@@ -180,6 +180,33 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
 
       assert log =~ "UnknownPresenceEvent"
     end
+
+    test "socket with presence enabled false will ignore presence events in public channel", %{
+      tenant: tenant,
+      topic: topic
+    } do
+      key = random_string()
+      policies = %Policies{presence: %PresencePolicies{read: true, write: true}}
+      socket = socket_fixture(tenant, topic, key, policies, false, false)
+
+      assert {:reply, :ok, _socket} = PresenceHandler.handle(%{"event" => "track"}, socket)
+      topic = "realtime:#{topic}"
+      refute_receive %Broadcast{topic: ^topic, event: "presence_diff"}
+    end
+
+    test "socket with presence enabled false will ignore presence events in private channel", %{
+      tenant: tenant,
+      topic: topic,
+      db_conn: db_conn
+    } do
+      key = random_string()
+      policies = %Policies{presence: %PresencePolicies{read: true, write: true}}
+      socket = socket_fixture(tenant, topic, key, policies, false, false)
+
+      assert {:reply, :ok, _socket} = PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
+      topic = "realtime:#{topic}"
+      refute_receive %Broadcast{topic: ^topic, event: "presence_diff"}
+    end
   end
 
   defp initiate_tenant(context) do
@@ -213,7 +240,8 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
            broadcast: %BroadcastPolicies{read: true},
            presence: %PresencePolicies{read: true, write: nil}
          },
-         private? \\ true
+         private? \\ true,
+         enabled? \\ true
        ) do
     claims = %{sub: random_string(), role: "authenticated", exp: Joken.current_time() + 1_000}
     signer = Joken.Signer.create("HS256", "secret")
@@ -248,7 +276,8 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
         authorization_context: authorization_context,
         rate_counter: rate_counter,
         private?: private?,
-        presence_key: presence_key
+        presence_key: presence_key,
+        presence_enabled?: enabled?
       }
     }
   end