fix: multiple valid subscriptions (#1596)

Author: edgurgel

lib/extensions/postgres_cdc_rls/cdc_rls.ex

@@ -43,7 +43,7 @@ defmodule Extensions.PostgresCdcRls do
   end
 
   defp subscription_list(params_list) do
-    Enum.reduce_while(params_list, [], fn params, acc ->
+    Enum.reduce_while(params_list, {:ok, []}, fn params, {:ok, acc} ->
       case Subscriptions.parse_subscription_params(params[:params]) do
         {:ok, subscription_params} ->
           {:cont, {:ok, [%{id: params.id, claims: params.claims, subscription_params: subscription_params} | acc]}}

lib/extensions/postgres_cdc_rls/subscriptions.ex

@@ -75,7 +75,7 @@ defmodule Extensions.PostgresCdcRls.Subscriptions do
       end)
     end)
   rescue
-    e -> {:error, e}
+    e in DBConnection.ConnectionError -> {:error, e}
   catch
     :exit, reason -> {:error, {:exit, reason}}
   end

mix.exs

@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
   def project do
     [
       app: :realtime,
-      version: "2.57.2",
+      version: "2.57.3",
       elixir: "~> 1.18",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,

test/realtime_web/channels/realtime_channel_test.exs

@@ -93,6 +93,58 @@ defmodule RealtimeWeb.RealtimeChannelTest do
       refute_receive _any
     end
 
+    test "multiple subscriptions", %{tenant: tenant} do
+      jwt = Generators.generate_jwt_token(tenant)
+      {:ok, %Socket{} = socket} = connect(UserSocket, %{}, conn_opts(tenant, jwt))
+
+      config = %{
+        "presence" => %{"enabled" => false},
+        "postgres_changes" => [
+          %{"event" => "INSERT", "schema" => "public", "table" => "test"},
+          %{"event" => "DELETE", "schema" => "public", "table" => "test"}
+        ]
+      }
+
+      assert {:ok, reply, _socket} = subscribe_and_join(socket, "realtime:test", %{"config" => config})
+
+      assert %{
+               postgres_changes: [
+                 %{:id => sub_id, "event" => "INSERT", "schema" => "public", "table" => "test"},
+                 %{
+                   :id => 4_845_530,
+                   "event" => "DELETE",
+                   "schema" => "public",
+                   "table" => "test"
+                 }
+               ]
+             } =
+               reply
+
+      assert_push "system",
+                  %{message: "Subscribed to PostgreSQL", status: "ok", extension: "postgres_changes", channel: "test"},
+                  5000
+
+      {:ok, conn} = Connect.lookup_or_start_connection(tenant.external_id)
+      %{rows: [[id]]} = Postgrex.query!(conn, "insert into test (details) values ('test') returning id", [])
+
+      assert_push "postgres_changes",
+                  %{
+                    data: %Realtime.Adapters.Changes.NewRecord{
+                      table: "test",
+                      type: "INSERT",
+                      record: %{"details" => "test", "id" => ^id},
+                      columns: [%{"name" => "id", "type" => "int4"}, %{"name" => "details", "type" => "text"}],
+                      errors: nil,
+                      schema: "public",
+                      commit_timestamp: _
+                    },
+                    ids: [4_845_530, ^sub_id]
+                  },
+                  500
+
+      refute_receive _any
+    end
+
     test "malformed subscription params", %{tenant: tenant} do
       jwt = Generators.generate_jwt_token(tenant)
       {:ok, %Socket{} = socket} = connect(UserSocket, %{}, conn_opts(tenant, jwt))