fix: tests

Author: edgurgel

lib/extensions/postgres_cdc_rls/subscriptions.ex

@@ -8,7 +8,7 @@ defmodule Extensions.PostgresCdcRls.Subscriptions do
 
   @type conn() :: Postgrex.conn()
   @type filter :: {binary, binary, binary}
-  @type subscription_params :: {binary, binary, binary, [filter]}
+  @type subscription_params :: {action_filter :: binary, schema :: binary, table :: binary, [filter]}
   @type subscription_list :: [%{id: binary, claims: map, subscription_params: subscription_params}]
 
   @filter_types ["eq", "neq", "lt", "lte", "gt", "gte", "in"]

test/integration/rt_channel_test.exs

@@ -34,9 +34,10 @@ defmodule Realtime.Integration.RtChannelTest do
       {:ok, conn} = Database.connect(tenant, "realtime_test")
 
       # Let's drop the publication to cause an error
-      Database.transaction(conn, fn db_conn ->
-        Postgrex.query!(db_conn, "drop publication if exists supabase_realtime_test")
-      end)
+      {:ok, _} =
+        Database.transaction(conn, fn db_conn ->
+          Postgrex.query!(db_conn, "drop publication if exists supabase_realtime_test")
+        end)
 
       {socket, _} = get_connection(tenant, serializer)
       topic = "realtime:any"
@@ -52,7 +53,7 @@ defmodule Realtime.Integration.RtChannelTest do
                              "channel" => "any",
                              "extension" => "postgres_changes",
                              "message" =>
-                               "Unable to subscribe to changes with given parameters. Please check Realtime is enabled for the given connect parameters: [schema: public, table: *, filters: []]",
+                               "Unable to subscribe to changes with given parameters. Please check Realtime is enabled for the given connect parameters: [event: INSERT, schema: public, table: *, filters: []]",
                              "status" => "error"
                            },
                            ref: nil,
@@ -2300,7 +2301,7 @@ defmodule Realtime.Integration.RtChannelTest do
       # Does it recover?
       assert Connect.ready?(tenant.external_id)
       {:ok, db_conn} = Connect.lookup_or_start_connection(tenant.external_id)
-      Process.sleep(1000)
+      Process.sleep(5000)
       %{rows: [[new_db_pid]]} = Postgrex.query!(db_conn, active_slot_query, [])
 
       assert new_db_pid != original_db_pid

test/realtime/extensions/cdc_rls/subscription_manager_test.exs

@@ -147,7 +147,7 @@ defmodule Realtime.Extensions.CdcRls.SubscriptionManagerTest do
 
     pg_change_params = %{
       id: uuid,
-      subscription_params: {"public", "*", []},
+      subscription_params: {"*", "public", "*", []},
       claims: %{
         "exp" => System.system_time(:second) + 100_000,
         "iat" => 0,

test/realtime_web/channels/realtime_channel_test.exs

@@ -107,9 +107,9 @@ defmodule RealtimeWeb.RealtimeChannelTest do
 
       assert %{
                postgres_changes: [
-                 %{:id => sub_id, "event" => "INSERT", "schema" => "public", "table" => "test"},
+                 %{:id => insert_sub_id, "event" => "INSERT", "schema" => "public", "table" => "test"},
                  %{
-                   :id => 4_845_530,
+                   :id => delete_sub_id,
                    "event" => "DELETE",
                    "schema" => "public",
                    "table" => "test"
@@ -123,9 +123,12 @@ defmodule RealtimeWeb.RealtimeChannelTest do
                   5000
 
       {:ok, conn} = Connect.lookup_or_start_connection(tenant.external_id)
+      # Insert, update and delete but update should not be received
       %{rows: [[id]]} = Postgrex.query!(conn, "insert into test (details) values ('test') returning id", [])
+      Postgrex.query!(conn, "update test set details = 'test' where id = $1", [id])
+      Postgrex.query!(conn, "delete from test where id = $1", [id])
 
-      assert_push "postgres_changes", %{data: data, ids: [4_845_530, ^sub_id]}, 500
+      assert_push "postgres_changes", %{data: data, ids: [^insert_sub_id]}, 500
 
       # we encode and decode because the data is a Jason.Fragment
       assert %{
@@ -138,6 +141,19 @@ defmodule RealtimeWeb.RealtimeChannelTest do
                "commit_timestamp" => _
              } = Jason.encode!(data) |> Jason.decode!()
 
+      assert_push "postgres_changes", %{data: data, ids: [^delete_sub_id]}, 500
+
+      # we encode and decode because the data is a Jason.Fragment
+      assert %{
+               "table" => "test",
+               "type" => "DELETE",
+               "old_record" => %{"id" => ^id},
+               "columns" => [%{"name" => "id", "type" => "int4"}, %{"name" => "details", "type" => "text"}],
+               "errors" => nil,
+               "schema" => "public",
+               "commit_timestamp" => _
+             } = Jason.encode!(data) |> Jason.decode!()
+
       refute_receive _any
     end
 
@@ -185,7 +201,7 @@ defmodule RealtimeWeb.RealtimeChannelTest do
       assert_push "system",
                   %{
                     message:
-                      "Unable to subscribe to changes with given parameters. Please check Realtime is enabled for the given connect parameters: [schema: public, table: doesnotexist, filters: []]",
+                      "Unable to subscribe to changes with given parameters. Please check Realtime is enabled for the given connect parameters: [event: *, schema: public, table: doesnotexist, filters: []]",
                     status: "error",
                     extension: "postgres_changes",
                     channel: "test"
@@ -216,7 +232,7 @@ defmodule RealtimeWeb.RealtimeChannelTest do
       assert_push "system",
                   %{
                     message:
-                      "Unable to subscribe to changes with given parameters. An exception happened so please check your connect parameters: [schema: public, table: test, filters: [{\"notacolumn\", \"eq\", \"123\"}]]. Exception: ERROR P0001 (raise_exception) invalid column for filter notacolumn",
+                      "Unable to subscribe to changes with given parameters. An exception happened so please check your connect parameters: [event: *, schema: public, table: test, filters: [{\"notacolumn\", \"eq\", \"123\"}]]. Exception: ERROR P0001 (raise_exception) invalid column for filter notacolumn",
                     status: "error",
                     extension: "postgres_changes",
                     channel: "test"