fix: ensure message id doesn't raise on non-map payloads (#1534)

Author: edgurgel

lib/realtime_web/channels/realtime_channel/message_dispatcher.ex

@@ -25,7 +25,7 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcher do
     # This reduce caches the serialization and bypasses the channel process going straight to the
     # transport process
 
-    message_id = msg.payload["meta"]["id"]
+    message_id = message_id(msg.payload)
 
     # Credo doesn't like that we don't use the result aggregation
     _ =
@@ -62,6 +62,9 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcher do
     :ok
   end
 
+  defp message_id(%{"meta" => %{"id" => id}}), do: id
+  defp message_id(_), do: nil
+
   defp already_replayed?(nil, _replayed_message_ids), do: false
   defp already_replayed?(message_id, replayed_message_ids), do: MapSet.member?(replayed_message_ids, message_id)
 

mix.exs

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

test/realtime_web/channels/realtime_channel/message_dispatcher_test.exs

@@ -112,6 +112,49 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcherTest do
       refute_receive _any
     end
 
+    test "payload is not a map" do
+      parent = self()
+
+      subscriber_pid =
+        spawn(fn ->
+          loop = fn loop ->
+            receive do
+              msg ->
+                send(parent, {:subscriber, msg})
+                loop.(loop)
+            end
+          end
+
+          loop.(loop)
+        end)
+
+      from_pid = :erlang.list_to_pid(~c'<0.2.1>')
+
+      subscribers = [
+        {subscriber_pid, {:rc_fastlane, self(), TestSerializer, "realtime:topic", {:log, "tenant123"}, MapSet.new()}},
+        {subscriber_pid, {:rc_fastlane, self(), TestSerializer, "realtime:topic", MapSet.new()}}
+      ]
+
+      msg = %Broadcast{topic: "some:other:topic", event: "event", payload: "not a map"}
+
+      log =
+        capture_log(fn ->
+          assert MessageDispatcher.dispatch(subscribers, from_pid, msg) == :ok
+        end)
+
+      assert log =~ "Received message on realtime:topic with payload: #{inspect(msg, pretty: true)}"
+
+      assert_receive {:encoded, %Broadcast{event: "event", payload: "not a map", topic: "realtime:topic"}}
+      assert_receive {:encoded, %Broadcast{event: "event", payload: "not a map", topic: "realtime:topic"}}
+
+      assert Agent.get(TestSerializer, & &1) == 1
+
+      assert_receive {:subscriber, :update_rate_counter}
+      assert_receive {:subscriber, :update_rate_counter}
+
+      refute_receive _any
+    end
+
     test "dispatches messages to non fastlane subscribers" do
       from_pid = :erlang.list_to_pid(~c'<0.2.1>')
 

test/realtime_web/channels/realtime_channel_test.exs

@@ -31,6 +31,46 @@ defmodule RealtimeWeb.RealtimeChannelTest do
   describe "broadcast" do
     @describetag policies: [:authenticated_all_topic_read]
 
+    test "broadcast map payload", %{tenant: tenant} do
+      jwt = Generators.generate_jwt_token(tenant)
+      {:ok, %Socket{} = socket} = connect(UserSocket, %{}, conn_opts(tenant, jwt))
+
+      config = %{
+        "presence" => %{"enabled" => false},
+        "broadcast" => %{"self" => true}
+      }
+
+      assert {:ok, _, socket} = subscribe_and_join(socket, "realtime:test", %{"config" => config})
+
+      push(socket, "broadcast", %{"event" => "my_event", "payload" => %{"hello" => "world"}})
+
+      assert_receive %Phoenix.Socket.Message{
+        topic: "realtime:test",
+        event: "broadcast",
+        payload: %{"event" => "my_event", "payload" => %{"hello" => "world"}}
+      }
+    end
+
+    test "broadcast non-map payload", %{tenant: tenant} do
+      jwt = Generators.generate_jwt_token(tenant)
+      {:ok, %Socket{} = socket} = connect(UserSocket, %{}, conn_opts(tenant, jwt))
+
+      config = %{
+        "presence" => %{"enabled" => false},
+        "broadcast" => %{"self" => true}
+      }
+
+      assert {:ok, _, socket} = subscribe_and_join(socket, "realtime:test", %{"config" => config})
+
+      push(socket, "broadcast", "not a map")
+
+      assert_receive %Phoenix.Socket.Message{
+        topic: "realtime:test",
+        event: "broadcast",
+        payload: "not a map"
+      }
+    end
+
     test "wrong replay params", %{tenant: tenant} do
       jwt = Generators.generate_jwt_token(tenant)
       {:ok, %Socket{} = socket} = connect(UserSocket, %{"log_level" => "warning"}, conn_opts(tenant, jwt))