bump to latest version; fix tests

Author: filipecabaco

lib/realtime/tenants/replication_connection.ex

@@ -15,22 +15,6 @@ defmodule Realtime.Tenants.ReplicationConnection do
   * `handler_module` - The module that will handle the data received from the replication stream.
   * `metadata` - The metadata to pass to the handler module.
 
-  ## How did we understood the data tuple format when pgoutput uses binary mode?
-  We checked the documentation with the help of AI and understood that pgoutput does the following logic:
-  * When `pgoutput` uses binary mode goes to https://doxygen.postgresql.org/proto_8c_source.html#l00767
-  * We end up in `OidSendFunctionCall` which gets the write function to call and transform into a binary payload based on OID - https://doxygen.postgresql.org/fmgr_8c.html#ad028dec5dc28e883308be264689f4559
-  * The functions are referenced in https://doxygen.postgresql.org/dir_19ba549c088f18020d0eadfcd07e87be.html
-  * It does `<data_tupe>.c` and calls `<data_type>_send` as the function to use
-    * Jsonb - https://doxygen.postgresql.org/jsonb_8c.html#afa4708e251f3078084e9986a46f228e9
-        * First byte is JSONB version (1), remainder is the text value
-    * Timestamp - https://doxygen.postgresql.org/backend_2utils_2adt_2timestamp_8c.html#ab0f80b3ae44bdcc50d0c0abb0c1d7939
-        * 64 bytes integer with microseconds since `2000-01-01`
-    * Boolean - https://doxygen.postgresql.org/bool_8c.html#a71cfb4f6c28259924b231dae4781b3fc
-        * 1 is true, 0 is false
-    * UUID - https://doxygen.postgresql.org/uuid_8c.html#a3cd205c61c108aaf08d6746c1d321e81
-        * 16 byte binary ( https://doxygen.postgresql.org/uuid_8h.html#a9692a0205a857ed2cc29558470c2ed77 )
-    * Text - https://doxygen.postgresql.org/varlena_8c.html#a0ea0ddd71c614e3fbd3571e70c53ed77
-        * Sends binary data
   """
   use Postgrex.ReplicationConnection
   use Realtime.Logs
@@ -73,7 +57,7 @@ defmodule Realtime.Tenants.ReplicationConnection do
             publication_name: nil,
             replication_slot_name: nil,
             output_plugin: "pgoutput",
-            proto_version: 1,
+            proto_version: 3,
             relations: %{},
             buffer: [],
             monitored_pid: nil,
@@ -332,7 +316,7 @@ defmodule Realtime.Tenants.ReplicationConnection do
            topic: topic,
            event: event,
            private: private,
-           payload: payload
+           payload: Jason.Fragment.new(payload)
          },
          :ok <- BatchBroadcast.broadcast(nil, tenant, %{messages: [broadcast_message]}, true) do
       latency_inserted_at = NaiveDateTime.utc_now(:microsecond) |> NaiveDateTime.diff(inserted_at, :microsecond)
@@ -397,8 +381,7 @@ defmodule Realtime.Tenants.ReplicationConnection do
     |> Enum.zip(columns)
     |> Map.new(fn
       {nil, %{name: name}} -> {name, nil}
-      {value, %{name: name, type: "bool"}} -> {name, value == "t"}
-      {value, %{name: name, type: "jsonb"}} -> {name, Jason.decode!(value)}
+      {value, %{name: name, type: "bool"}} -> {name, value}
       {value, %{name: name}} -> {name, value}
     end)
   end

test/realtime/tenants/replication_connection_test.exs

@@ -151,6 +151,88 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
       end
     end
 
+    test "starts a handler for the tenant and broadcasts to public channel", %{tenant: tenant, db_conn: db_conn} 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, true)
+      subscribe(tenant_topic, topic)
+
+      total_messages = 5
+      # Works with one insert per transaction
+      for _ <- 1..total_messages do
+        value = random_string()
+
+        row =
+          message_fixture(tenant, %{
+            "topic" => topic,
+            "private" => false,
+            "event" => "INSERT",
+            "payload" => %{"value" => value}
+          })
+
+        assert_receive {:socket_push, :text, data}
+        message = data |> IO.iodata_to_binary() |> Jason.decode!()
+
+        payload = %{
+          "event" => "INSERT",
+          "meta" => %{"id" => row.id},
+          "payload" => %{
+            "value" => value
+          },
+          "type" => "broadcast"
+        }
+
+        assert message == %{"event" => "broadcast", "payload" => payload, "ref" => nil, "topic" => topic}
+      end
+
+      Process.sleep(500)
+      # Works with batch inserts
+      messages =
+        for _ <- 1..total_messages do
+          Message.changeset(%Message{}, %{
+            "topic" => topic,
+            "private" => false,
+            "event" => "INSERT",
+            "extension" => "broadcast",
+            "payload" => %{"value" => random_string()}
+          })
+        end
+
+      {:ok, _} = Realtime.Repo.insert_all_entries(db_conn, messages, Message)
+
+      messages_received =
+        for _ <- 1..total_messages, into: [] do
+          assert_receive {:socket_push, :text, data}
+          data |> IO.iodata_to_binary() |> Jason.decode!()
+        end
+
+      for row <- messages do
+        assert Enum.count(messages_received, fn message_received ->
+                 value = row |> Map.from_struct() |> get_in([:changes, :payload, "value"])
+
+                 match?(
+                   %{
+                     "event" => "broadcast",
+                     "payload" => %{
+                       "event" => "INSERT",
+                       "meta" => %{"id" => _id},
+                       "payload" => %{
+                         "value" => ^value
+                       }
+                     },
+                     "ref" => nil,
+                     "topic" => ^topic
+                   },
+                   message_received
+                 )
+               end) == 1
+      end
+    end
+
     test "monitored pid stopping brings down ReplicationConnection ", %{tenant: tenant} do
       monitored_pid =
         spawn(fn ->
@@ -227,6 +309,86 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
       assert logs =~ "UnableToBroadcastChanges: %{messages: [%{payload: [\"Payload size exceeds tenant limit\"]}]}"
     end
 
+    test "payload without id", %{tenant: tenant, db_conn: db_conn} 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)
+
+      value = "something"
+      event = "INSERT"
+
+      Postgrex.query!(
+        db_conn,
+        "SELECT realtime.send (json_build_object ('value', $1 :: text)::jsonb, $2 :: text, $3 :: text, TRUE::bool);",
+        [value, event, topic]
+      )
+
+      {:ok, [%{id: id}]} = Repo.all(db_conn, from(m in Message), Message)
+
+      assert_receive {:socket_push, :text, data}, 500
+      message = data |> IO.iodata_to_binary() |> Jason.decode!()
+
+      assert %{
+               "event" => "broadcast",
+               "payload" => %{
+                 "event" => "INSERT",
+                 "meta" => %{"id" => ^id},
+                 "payload" => payload,
+                 "type" => "broadcast"
+               },
+               "ref" => nil,
+               "topic" => ^topic
+             } = message
+
+      assert payload == %{
+               "value" => "something",
+               "id" => id
+             }
+    end
+
+    test "payload including id", %{tenant: tenant, db_conn: db_conn} 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)
+
+      id = "123456"
+      value = "something"
+      event = "INSERT"
+
+      Postgrex.query!(
+        db_conn,
+        "SELECT realtime.send (json_build_object ('value', $1 :: text, 'id', $2 :: text)::jsonb, $3 :: text, $4 :: text, TRUE::bool);",
+        [value, id, event, topic]
+      )
+
+      {:ok, [%{id: message_id}]} = Repo.all(db_conn, from(m in Message), Message)
+
+      assert_receive {:socket_push, :text, data}, 500
+      message = data |> IO.iodata_to_binary() |> Jason.decode!()
+
+      assert %{
+               "event" => "broadcast",
+               "payload" => %{
+                 "meta" => %{"id" => ^message_id},
+                 "event" => "INSERT",
+                 "payload" => %{"value" => "something", "id" => ^id},
+                 "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"
@@ -355,33 +517,17 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
         "payload" => %{"value" => random_string()}
       })
 
-      # Telemetry event may arrive before socket_push, so we need to handle both
-      receive do
-        {:socket_push, :text, data} ->
-          message = data |> IO.iodata_to_binary() |> Jason.decode!()
-          assert %{"event" => "broadcast", "payload" => _, "ref" => nil, "topic" => ^topic} = message
-
-          assert_receive {[:realtime, :tenants, :broadcast_from_database],
-                          %{latency_committed_at: latency_committed_at, latency_inserted_at: latency_inserted_at},
-                          %{tenant: ^external_id}},
-                         1000
-
-          assert latency_committed_at
-          assert latency_inserted_at != 0
-
-        {[:realtime, :tenants, :broadcast_from_database],
-         %{latency_committed_at: latency_committed_at, latency_inserted_at: latency_inserted_at},
-         %{tenant: ^external_id}} ->
-          assert latency_committed_at
-          assert latency_inserted_at != 0
-
-          assert_receive {:socket_push, :text, data}, 2000
-          message = data |> IO.iodata_to_binary() |> Jason.decode!()
-          assert %{"event" => "broadcast", "payload" => _, "ref" => nil, "topic" => ^topic} = message
-      after
-        3000 ->
-          flunk("Expected to receive either socket_push or telemetry event")
-      end
+      assert_receive {:socket_push, :text, data}, 500
+      message = data |> IO.iodata_to_binary() |> Jason.decode!()
+
+      assert %{"event" => "broadcast", "payload" => _, "ref" => nil, "topic" => ^topic} = message
+
+      assert_receive {[:realtime, :tenants, :broadcast_from_database],
+                      %{latency_committed_at: latency_committed_at, latency_inserted_at: latency_inserted_at},
+                      %{tenant: ^external_id}}
+
+      assert latency_committed_at > 0
+      assert latency_inserted_at > 0
     end
   end