pr feedback

Author: filipecabaco

lib/realtime/tenants.ex

@@ -511,8 +511,9 @@ defmodule Realtime.Tenants do
     |> validate_payload_size(payload)
   end
 
+  @payload_size_padding 500
   def validate_payload_size(%Tenant{max_payload_size_in_kb: max_payload_size_in_kb}, payload) do
-    max_payload_size = max_payload_size_in_kb * 1000
+    max_payload_size = max_payload_size_in_kb * 1000 + @payload_size_padding
     payload_size = :erlang.external_size(payload)
     if payload_size > max_payload_size, do: {:error, :payload_size_exceeded}, else: :ok
   end

lib/realtime/tenants/batch_broadcast.ex

@@ -33,7 +33,7 @@ defmodule Realtime.Tenants.BatchBroadcast do
             messages: list(%{id: String.t(), topic: String.t(), payload: map(), event: String.t(), private: boolean()})
           },
           super_user :: boolean()
-        ) :: :ok | {:error, atom()}
+        ) :: :ok | {:error, atom() | Ecto.Changeset.t()}
   def broadcast(auth_params, tenant, messages, super_user \\ false)
 
   def broadcast(%Plug.Conn{} = conn, %Tenant{} = tenant, messages, super_user) do
@@ -71,15 +71,11 @@ defmodule Realtime.Tenants.BatchBroadcast do
       |> Enum.group_by(fn event -> Map.get(event, :topic) end)
       |> Enum.each(fn {topic, events} ->
         if super_user do
-          Enum.each(events, fn message ->
-            send_message_and_count(tenant, events_per_second_rate, message, false)
-          end)
+          Enum.each(events, fn message -> send_message_and_count(tenant, events_per_second_rate, message, false) end)
         else
           case permissions_for_message(tenant, auth_params, topic) do
             %Policies{broadcast: %BroadcastPolicies{write: true}} ->
-              Enum.each(events, fn message ->
-                send_message_and_count(tenant, events_per_second_rate, message, false)
-              end)
+              Enum.each(events, fn message -> send_message_and_count(tenant, events_per_second_rate, message, false) end)
 
             _ ->
               nil
@@ -88,6 +84,9 @@ defmodule Realtime.Tenants.BatchBroadcast do
       end)
 
       :ok
+    else
+      %Ecto.Changeset{valid?: false} = changeset -> {:error, changeset}
+      error -> error
     end
   end
 

lib/realtime/tenants/replication_connection.ex

@@ -329,11 +329,12 @@ defmodule Realtime.Tenants.ReplicationConnection do
 
       {:noreply, state}
     else
-      {:error, error} ->
+      {:error, %Ecto.Changeset{valid?: false} = changeset} ->
+        error = Ecto.Changeset.traverse_errors(changeset, &elem(&1, 0))
         log_error("UnableToBroadcastChanges", error)
         {:noreply, state}
 
-      %Ecto.Changeset{valid?: false, changes: %{messages: [%{errors: [payload: {error, _}]} | _]}} ->
+      {:error, error} ->
         log_error("UnableToBroadcastChanges", error)
         {:noreply, state}
 

lib/realtime_web/channels/realtime_channel/presence_handler.ex

@@ -161,13 +161,5 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandler do
     end
   end
 
-  # Added due to the fact that JSON decoding adds some overhead and erlang term will be slighly larger
-  @payload_size_padding 500
-  defp validate_payload_size(tenant, payload) do
-    if :erlang.external_size(payload) > tenant.max_payload_size_in_kb * 1000 + @payload_size_padding do
-      {:error, :payload_size_exceeded}
-    else
-      :ok
-    end
-  end
+  defp validate_payload_size(tenant, payload), do: Tenants.validate_payload_size(tenant, payload)
 end

lib/realtime_web/controllers/fallback_controller.ex

@@ -45,7 +45,7 @@ defmodule RealtimeWeb.FallbackController do
     |> render("error.json", message: message)
   end
 
-  def call(conn, %Ecto.Changeset{valid?: true} = changeset) do
+  def call(conn, {:error, %Ecto.Changeset{valid?: false} = changeset}) do
     log_error(
       "UnprocessableEntity",
       Ecto.Changeset.traverse_errors(changeset, &translate_error/1)

test/realtime/tenants/batch_broadcast_test.exs

@@ -372,7 +372,7 @@ defmodule Realtime.Tenants.BatchBroadcastTest do
       reject(&TenantBroadcaster.pubsub_broadcast/5)
 
       result = BatchBroadcast.broadcast(nil, tenant, messages, false)
-      assert %Ecto.Changeset{valid?: false} = result
+      assert {:error, %Ecto.Changeset{valid?: false}} = result
     end
 
     test "returns changeset error when payload is missing", %{tenant: tenant} do
@@ -382,7 +382,7 @@ defmodule Realtime.Tenants.BatchBroadcastTest do
       reject(&TenantBroadcaster.pubsub_broadcast/5)
 
       result = BatchBroadcast.broadcast(nil, tenant, messages, false)
-      assert %Ecto.Changeset{valid?: false} = result
+      assert {:error, %Ecto.Changeset{valid?: false}} = result
     end
 
     test "returns changeset error when event is missing", %{tenant: tenant} do
@@ -391,14 +391,14 @@ defmodule Realtime.Tenants.BatchBroadcastTest do
 
       reject(&TenantBroadcaster.pubsub_broadcast/5)
       result = BatchBroadcast.broadcast(nil, tenant, messages, false)
-      assert %Ecto.Changeset{valid?: false} = result
+      assert {:error, %Ecto.Changeset{valid?: false}} = result
     end
 
     test "returns changeset error when messages array is empty", %{tenant: tenant} do
       messages = %{messages: []}
       reject(&TenantBroadcaster.pubsub_broadcast/5)
       result = BatchBroadcast.broadcast(nil, tenant, messages, false)
-      assert %Ecto.Changeset{valid?: false} = result
+      assert {:error, %Ecto.Changeset{valid?: false}} = result
     end
   end
 
@@ -410,9 +410,7 @@ defmodule Realtime.Tenants.BatchBroadcastTest do
 
       RateCounter
       |> stub(:new, fn _ -> {:ok, nil} end)
-      |> stub(:get, fn ^events_per_second_rate ->
-        {:ok, %RateCounter{avg: tenant.max_events_per_second + 1}}
-      end)
+      |> stub(:get, fn ^events_per_second_rate -> {:ok, %RateCounter{avg: tenant.max_events_per_second + 1}} end)
 
       reject(&TenantBroadcaster.pubsub_broadcast/5)
 
@@ -482,10 +480,11 @@ defmodule Realtime.Tenants.BatchBroadcastTest do
 
       result = BatchBroadcast.broadcast(nil, tenant, messages, false)
 
-      assert %Ecto.Changeset{
-               valid?: false,
-               changes: %{messages: [%{errors: [payload: {"Payload size exceeds tenant limit", []}]}]}
-             } = result
+      assert {:error,
+              %Ecto.Changeset{
+                valid?: false,
+                changes: %{messages: [%{errors: [payload: {"Payload size exceeds tenant limit", []}]}]}
+              }} = result
     end
   end
 

test/realtime/tenants/replication_connection_test.exs

@@ -228,7 +228,7 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
           refute_receive %Phoenix.Socket.Broadcast{}, 500
         end)
 
-      assert logs =~ "UnableToBroadcastChanges: Payload size exceeds tenant limit"
+      assert logs =~ "UnableToBroadcastChanges: %{messages: [%{payload: [\"Payload size exceeds tenant limit\"]}]}"
     end
 
     test "payload without id", %{tenant: tenant} do

test/realtime_web/channels/realtime_channel/broadcast_handler_test.exs

@@ -27,15 +27,15 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandlerTest do
           socket
       end
 
-      Process.sleep(120)
-
       for _ <- 1..100 do
         topic = "realtime:#{topic}"
         assert_receive {:socket_push, :text, data}
         message = data |> IO.iodata_to_binary() |> Jason.decode!()
         assert message == %{"event" => "broadcast", "payload" => %{"a" => "b"}, "ref" => nil, "topic" => topic}
       end
 
+      Process.sleep(120)
+
       {:ok, %{avg: avg, bucket: buckets}} = RateCounter.get(Tenants.events_per_second_rate(tenant))
       assert Enum.sum(buckets) == 100
       assert avg > 0
@@ -50,8 +50,6 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandlerTest do
           socket
       end
 
-      Process.sleep(120)
-
       refute_received _any
 
       {:ok, %{avg: avg}} = RateCounter.get(Tenants.events_per_second_rate(tenant))
@@ -68,15 +66,14 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandlerTest do
           socket
       end
 
-      Process.sleep(120)
-
       for _ <- 1..100 do
         topic = "realtime:#{topic}"
         assert_received {:socket_push, :text, data}
         message = data |> IO.iodata_to_binary() |> Jason.decode!()
         assert message == %{"event" => "broadcast", "payload" => %{"a" => "b"}, "ref" => nil, "topic" => topic}
       end
 
+      Process.sleep(120)
       {:ok, %{avg: avg, bucket: buckets}} = RateCounter.get(Tenants.events_per_second_rate(tenant))
       assert Enum.sum(buckets) == 100
       assert avg > 0.0
@@ -96,8 +93,6 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandlerTest do
           socket
       end
 
-      Process.sleep(120)
-
       for _ <- 1..100 do
         topic = "realtime:#{topic}"
         assert_received {:socket_push, :text, data}
@@ -120,9 +115,7 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandlerTest do
           socket
       end
 
-      Process.sleep(120)
-
-      refute_received {:socket_push, :text, _}
+      refute_received {:socket_push, :text, _}, 120
     end
 
     @tag policies: [:read_matching_user_role, :write_matching_user_role], role: "anon"
@@ -139,8 +132,6 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandlerTest do
           socket
       end
 
-      Process.sleep(120)
-
       for _ <- 1..100 do
         topic = "realtime:#{topic}"
         assert_received {:socket_push, :text, data}
@@ -163,9 +154,7 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandlerTest do
           socket
       end
 
-      Process.sleep(120)
-
-      refute_received {:socket_push, :text, _}
+      refute_received {:socket_push, :text, _}, 120
     end
 
     test "with nil policy and invalid user, won't send message", %{topic: topic, tenant: tenant, db_conn: db_conn} do
@@ -177,8 +166,6 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandlerTest do
           socket
       end
 
-      Process.sleep(120)
-
       refute_received _any
 
       {:ok, %{avg: avg}} = RateCounter.get(Tenants.events_per_second_rate(tenant))
@@ -259,15 +246,14 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandlerTest do
           socket
       end
 
-      Process.sleep(120)
-
       for _ <- 1..100 do
         topic = "realtime:#{topic}"
         assert_received {:socket_push, :text, data}
         message = data |> IO.iodata_to_binary() |> Jason.decode!()
         assert message == %{"event" => "broadcast", "payload" => %{"a" => "b"}, "ref" => nil, "topic" => topic}
       end
 
+      Process.sleep(120)
       {:ok, %{avg: avg, bucket: buckets}} = RateCounter.get(Tenants.events_per_second_rate(tenant))
       assert Enum.sum(buckets) == 100
       assert avg > 0.0
@@ -290,6 +276,7 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandlerTest do
       end
 
       Process.sleep(120)
+
       {:ok, %{avg: avg, bucket: buckets}} = RateCounter.get(Tenants.events_per_second_rate(tenant))
       assert Enum.sum(buckets) == 100
       assert avg > 0.0
@@ -327,9 +314,7 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandlerTest do
                  socket
                )
 
-      Process.sleep(120)
-
-      refute_received {:socket_push, :text, _}
+      refute_received {:socket_push, :text, _}, 120
     end
 
     test "handle payload size excedding limits in public channels", %{topic: topic, tenant: tenant, db_conn: db_conn} do
@@ -342,9 +327,7 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandlerTest do
                  socket
                )
 
-      Process.sleep(120)
-
-      refute_received {:socket_push, :text, _}
+      refute_received {:socket_push, :text, _}, 120
     end
 
     test "handle payload size excedding limits in private channel and if ack it will receive error", %{
@@ -365,9 +348,7 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandlerTest do
                  socket
                )
 
-      Process.sleep(120)
-
-      refute_received {:socket_push, :text, _}
+      refute_received {:socket_push, :text, _}, 120
     end
 
     test "handle payload size excedding limits in public channels and if ack it will receive error", %{
@@ -384,9 +365,7 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandlerTest do
                  socket
                )
 
-      Process.sleep(120)
-
-      refute_received {:socket_push, :text, _}
+      refute_received {:socket_push, :text, _}, 120
     end
   end
 

test/realtime_web/controllers/broadcast_controller_test.exs

@@ -158,8 +158,6 @@ defmodule RealtimeWeb.BroadcastControllerTest do
     } 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)}