feat: speed up messages replication by avoiding re-encoding JSON

We can use the already JSON encoded payload and pass a Jason.Fragment
instead of decoding to then re-encode when the message gets sent to a
websocket

Author: edgurgel

lib/realtime/tenants/replication_connection.ex

@@ -159,6 +159,7 @@ defmodule Realtime.Tenants.ReplicationConnection do
 
   @impl true
   def init(%__MODULE__{tenant_id: tenant_id, monitored_pid: monitored_pid} = state) do
+    Process.flag(:fullsweep_after, 20)
     Logger.metadata(external_id: tenant_id, project: tenant_id)
     Process.monitor(monitored_pid)
 
@@ -315,7 +316,7 @@ defmodule Realtime.Tenants.ReplicationConnection do
            topic: topic,
            event: event,
            private: private,
-           payload: Map.put_new(payload, "id", id)
+           payload: Jason.Fragment.new(payload)
          },
          :ok <- BatchBroadcast.broadcast(nil, tenant, %{messages: [broadcast_message]}, true) do
       inserted_at = NaiveDateTime.from_iso8601!(inserted_at)
@@ -381,7 +382,6 @@ defmodule Realtime.Tenants.ReplicationConnection do
     |> Enum.zip(columns)
     |> Map.new(fn
       {nil, %{name: name}} -> {name, nil}
-      {value, %{name: name, type: "jsonb"}} -> {name, Jason.decode!(value)}
       {value, %{name: name, type: "bool"}} -> {name, value == "t"}
       {value, %{name: name}} -> {name, value}
     end)

test/realtime/tenants/replication_connection_test.exs

@@ -100,7 +100,6 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
           "event" => "INSERT",
           "meta" => %{"id" => row.id},
           "payload" => %{
-            "id" => row.id,
             "value" => value
           },
           "type" => "broadcast"
@@ -142,7 +141,6 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
                        "event" => "INSERT",
                        "meta" => %{"id" => id},
                        "payload" => %{
-                         "id" => id,
                          "value" => ^value
                        }
                      },
@@ -231,83 +229,6 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
       assert logs =~ "UnableToBroadcastChanges: %{messages: [%{payload: [\"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()}},
-        restart: :transient
-      )
-
-      topic = random_string()
-      tenant_topic = Tenants.tenant_topic(tenant.external_id, topic, false)
-      subscribe(tenant_topic, topic)
-
-      fixture =
-        message_fixture(tenant, %{
-          "topic" => topic,
-          "private" => true,
-          "event" => "INSERT",
-          "payload" => %{"value" => "something"}
-        })
-
-      fixture_id = fixture.id
-
-      assert_receive {:socket_push, :text, data}, 500
-      message = data |> IO.iodata_to_binary() |> Jason.decode!()
-
-      assert %{
-               "event" => "broadcast",
-               "payload" => %{
-                 "event" => "INSERT",
-                 "meta" => %{"id" => ^fixture_id},
-                 "payload" => payload,
-                 "type" => "broadcast"
-               },
-               "ref" => nil,
-               "topic" => ^topic
-             } = message
-
-      assert payload == %{
-               "value" => "something",
-               "id" => fixture_id
-             }
-    end
-
-    test "payload including id", %{tenant: tenant} do
-      start_link_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)
-      subscribe(tenant_topic, topic)
-
-      payload = %{"value" => "something", "id" => "123456"}
-
-      %{id: fixture_id} =
-        message_fixture(tenant, %{
-          "topic" => topic,
-          "private" => true,
-          "event" => "INSERT",
-          "payload" => payload
-        })
-
-      assert_receive {:socket_push, :text, data}, 500
-      message = data |> IO.iodata_to_binary() |> Jason.decode!()
-
-      assert %{
-               "event" => "broadcast",
-               "payload" => %{
-                 "meta" => %{"id" => ^fixture_id},
-                 "event" => "INSERT",
-                 "payload" => ^payload,
-                 "type" => "broadcast"
-               },
-               "ref" => nil,
-               "topic" => ^topic
-             } = message
-    end
-
     test "fails on existing replication slot", %{tenant: tenant} do
       {:ok, db_conn} = Database.connect(tenant, "realtime_test", :stop)
       name = "supabase_realtime_messages_replication_slot_test"