fix: include payload id if not defined on realtime.send

Author: edgurgel

lib/realtime/tenants/migrations.ex

@@ -75,7 +75,8 @@ defmodule Realtime.Tenants.Migrations do
     SubscriptionIndexBridgingDisabled,
     RunSubscriptionIndexBridgingDisabled,
     BroadcastSendErrorLogging,
-    CreateMessagesReplayIndex
+    CreateMessagesReplayIndex,
+    BroadcastSendIncludePayloadId
   }
 
   @migrations [
@@ -142,7 +143,8 @@ defmodule Realtime.Tenants.Migrations do
     {20_250_506_224_012, SubscriptionIndexBridgingDisabled},
     {20_250_523_164_012, RunSubscriptionIndexBridgingDisabled},
     {20_250_714_121_412, BroadcastSendErrorLogging},
-    {20_250_905_041_441, CreateMessagesReplayIndex}
+    {20_250_905_041_441, CreateMessagesReplayIndex},
+    {20_251_103_001_201, BroadcastSendIncludePayloadId}
   ]
 
   defstruct [:tenant_external_id, :settings]

lib/realtime/tenants/repo/migrations/20251103001201_broadcast_send_include_payload_id.ex

@@ -0,0 +1,41 @@
+defmodule Realtime.Tenants.Migrations.BroadcastSendIncludePayloadId do
+  @moduledoc false
+  use Ecto.Migration
+
+  # Include ID in the payload if not defined
+  def change do
+    execute("""
+    CREATE OR REPLACE FUNCTION realtime.send(payload jsonb, event text, topic text, private boolean DEFAULT true ) RETURNS void
+    AS $$
+    DECLARE
+      generated_id uuid;
+      final_payload jsonb;
+    BEGIN
+      BEGIN
+        -- Generate a new UUID for the id
+        generated_id := gen_random_uuid();
+
+        -- Check if payload has an 'id' key, if not, add the generated UUID
+        IF payload ? 'id' THEN
+          final_payload := payload;
+        ELSE
+          final_payload := jsonb_set(payload, '{id}', to_jsonb(generated_id));
+        END IF;
+
+        -- Set the topic configuration
+        EXECUTE format('SET LOCAL realtime.topic TO %L', topic);
+
+        -- Attempt to insert the message
+        INSERT INTO realtime.messages (id, payload, event, topic, private, extension)
+        VALUES (generated_id, final_payload, event, topic, private, 'broadcast');
+      EXCEPTION
+        WHEN OTHERS THEN
+          -- Capture and notify the error
+          RAISE WARNING 'ErrorSendingBroadcastMessage: %', SQLERRM;
+      END;
+    END;
+    $$
+    LANGUAGE plpgsql;
+    """)
+  end
+end

test/realtime/tenants/replication_connection_test.exs

@@ -22,9 +22,8 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
     {:ok, db_conn} = Database.connect(tenant, "realtime_test", :stop)
     name = "supabase_realtime_messages_replication_slot_test"
     Postgrex.query(db_conn, "SELECT pg_drop_replication_slot($1)", [name])
-    Process.exit(db_conn, :normal)
 
-    %{tenant: tenant}
+    %{tenant: tenant, db_conn: db_conn}
   end
 
   describe "temporary process" do
@@ -70,7 +69,7 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
       assert {:error, _} = ReplicationConnection.start(tenant, self())
     end
 
-    test "starts a handler for the tenant and broadcasts", %{tenant: tenant} do
+    test "starts a handler for the tenant and broadcasts", %{tenant: tenant, db_conn: db_conn} do
       start_link_supervised!(
         {ReplicationConnection, %ReplicationConnection{tenant_id: tenant.external_id, monitored_pid: self()}},
         restart: :transient
@@ -121,7 +120,6 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
           })
         end
 
-      {:ok, db_conn} = Database.connect(tenant, "realtime_test", :stop)
       {:ok, _} = Realtime.Repo.insert_all_entries(db_conn, messages, Message)
 
       messages_received =
@@ -139,7 +137,7 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
                      "event" => "broadcast",
                      "payload" => %{
                        "event" => "INSERT",
-                       "meta" => %{"id" => id},
+                       "meta" => %{"id" => _id},
                        "payload" => %{
                          "value" => ^value
                        }
@@ -229,6 +227,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"