add new tests and handling in replication connection and broadcast enpoint

Author: filipecabaco

lib/realtime/tenants/batch_broadcast.ex

@@ -130,11 +130,9 @@ defmodule Realtime.Tenants.BatchBroadcast do
     payload = %{"payload" => message.payload, "event" => message.event, "type" => "broadcast"}
 
     payload =
-      if message[:id] do
-        Map.put(payload, "meta", %{"id" => message.id})
-      else
-        payload
-      end
+      if message[:id],
+        do: Map.put(payload, "meta", %{"id" => message.id}),
+        else: payload
 
     broadcast = %Phoenix.Socket.Broadcast{topic: message.topic, event: @event_type, payload: payload}
 

lib/realtime/tenants/replication_connection.ex

@@ -333,6 +333,10 @@ defmodule Realtime.Tenants.ReplicationConnection do
         log_error("UnableToBroadcastChanges", error)
         {:noreply, state}
 
+      %Ecto.Changeset{valid?: false, changes: %{messages: [%{errors: [payload: {error, _}]} | _]}} ->
+        log_error("UnableToBroadcastChanges", error)
+        {:noreply, state}
+
       _ ->
         {:noreply, state}
     end

test/realtime/tenants/replication_connection_test.exs

@@ -206,6 +206,31 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
       assert logs =~ "UnableToBroadcastChanges"
     end
 
+    test "message that exceeds payload size logs error", %{tenant: tenant} do
+      logs =
+        capture_log(fn ->
+          start_supervised!(
+            {ReplicationConnection, %ReplicationConnection{tenant_id: tenant.external_id, monitored_pid: self()}},
+            restart: :transient
+          )
+
+          topic = random_string()
+          tenant_topic = Tenants.tenant_topic(tenant.external_id, topic, false)
+          assert :ok = Endpoint.subscribe(tenant_topic)
+
+          message_fixture(tenant, %{
+            "event" => random_string(),
+            "topic" => random_string(),
+            "private" => true,
+            "payload" => %{"data" => random_string(tenant.max_payload_size_in_kb * 1000 + 1)}
+          })
+
+          refute_receive %Phoenix.Socket.Broadcast{}, 500
+        end)
+
+      assert logs =~ "UnableToBroadcastChanges: Payload size exceeds tenant limit"
+    end
+
     test "payload without id", %{tenant: tenant} do
       start_link_supervised!(
         {ReplicationConnection, %ReplicationConnection{tenant_id: tenant.external_id, monitored_pid: self()}},

test/realtime_web/controllers/broadcast_controller_test.exs

@@ -151,6 +151,32 @@ defmodule RealtimeWeb.BroadcastControllerTest do
 
       refute_receive {:socket_push, _, _}
     end
+
+    test "returns 422 when batch of messages includes a message that exceeds the tenant payload size", %{
+      conn: conn,
+      tenant: tenant
+    } do
+      sub_topic_1 = "sub_topic_1"
+      sub_topic_2 = "sub_topic_2"
+      topic_1 = Tenants.tenant_topic(tenant, sub_topic_1)
+      topic_2 = Tenants.tenant_topic(tenant, sub_topic_2)
+
+      payload_1 = %{"data" => "data"}
+      payload_2 = %{"data" => random_string(tenant.max_payload_size_in_kb * 1000 + 100)}
+      event_1 = "event_1"
+      event_2 = "event_2"
+
+      conn =
+        post(conn, Routes.broadcast_path(conn, :broadcast), %{
+          "messages" => [
+            %{"topic" => sub_topic_1, "payload" => payload_1, "event" => event_1},
+            %{"topic" => sub_topic_1, "payload" => payload_1, "event" => event_1},
+            %{"topic" => sub_topic_2, "payload" => payload_2, "event" => event_2}
+          ]
+        })
+
+      assert conn.status == 422
+    end
   end
 
   describe "too many requests" do