only log in case of error

Author: filipecabaco

lib/realtime_web/channels/realtime_channel.ex

@@ -5,8 +5,8 @@ defmodule RealtimeWeb.RealtimeChannel do
   use RealtimeWeb, :channel
   use RealtimeWeb.RealtimeChannel.Logging
 
+  alias RealtimeWeb.SocketDisconnect
   alias DBConnection.Backoff
-  alias Phoenix.Socket
 
   alias Realtime.Crypto
   alias Realtime.GenCounter
@@ -22,14 +22,11 @@ defmodule RealtimeWeb.RealtimeChannel do
   alias Realtime.Tenants.Connect
 
   alias RealtimeWeb.Channels.Payloads.Join
-  alias RealtimeWeb.Channels.Payloads.Config
-  alias RealtimeWeb.Channels.Payloads.PostgresChange
   alias RealtimeWeb.ChannelsAuthorization
   alias RealtimeWeb.RealtimeChannel.BroadcastHandler
   alias RealtimeWeb.RealtimeChannel.MessageDispatcher
   alias RealtimeWeb.RealtimeChannel.PresenceHandler
   alias RealtimeWeb.RealtimeChannel.Tracker
-  alias RealtimeWeb.SocketDisconnect
 
   @confirm_token_ms_interval :timer.minutes(5)
 
@@ -50,11 +47,20 @@ defmodule RealtimeWeb.RealtimeChannel do
     Logger.metadata(external_id: tenant_id, project: tenant_id)
     Logger.put_process_level(self(), log_level)
 
-    # We always need to assign the access token so we can get the logs metadata working as expected
-    socket = assign_access_token(socket, params)
+    socket =
+      socket
+      |> assign_access_token(params)
+      |> assign_counter()
+      |> assign_presence_counter()
+      |> assign(:private?, !!params["config"]["private"])
+      |> assign(:policies, nil)
+
+    case Join.validate(params) do
+      {:ok, _join} -> nil
+      {:error, :invalid_join_payload, errors} -> log_error(socket, "InvalidJoinPayload", errors)
+    end
 
-    with {:ok, %Socket{} = socket, %Join{} = configuration} <- configure_socket(socket, params),
-         :ok <- SignalHandler.shutdown_in_progress?(),
+    with :ok <- SignalHandler.shutdown_in_progress?(),
          :ok <- only_private?(tenant_id, socket),
          :ok <- limit_joins(socket),
          :ok <- limit_channels(socket),
@@ -64,6 +70,7 @@ defmodule RealtimeWeb.RealtimeChannel do
          {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant_id),
          {:ok, socket} <- maybe_assign_policies(sub_topic, db_conn, socket) do
       tenant_topic = Tenants.tenant_topic(tenant_id, sub_topic, !socket.assigns.private?)
+
       # fastlane subscription
       metadata =
         MessageDispatcher.fastlane_metadata(transport_pid, serializer, topic, socket.assigns.log_level, tenant_id)
@@ -72,11 +79,15 @@ defmodule RealtimeWeb.RealtimeChannel do
 
       Phoenix.PubSub.subscribe(Realtime.PubSub, "realtime:operations:" <> tenant_id)
 
-      is_new_api = new_api?(configuration)
-
-      presence_enabled? = Join.presence_enabled?(configuration)
+      is_new_api = new_api?(params)
+      # TODO: Default will be moved to false in the future
+      presence_enabled? =
+        case get_in(params, ["config", "presence", "enabled"]) do
+          enabled when is_boolean(enabled) -> enabled
+          _ -> true
+        end
 
-      pg_change_params = pg_change_params(is_new_api, configuration, channel_pid, claims, sub_topic)
+      pg_change_params = pg_change_params(is_new_api, params, channel_pid, claims, sub_topic)
 
       opts = %{
         is_new_api: is_new_api,
@@ -93,13 +104,13 @@ defmodule RealtimeWeb.RealtimeChannel do
       state = %{postgres_changes: add_id_to_postgres_changes(pg_change_params)}
 
       assigns = %{
-        ack_broadcast: Join.ack_broadcast?(configuration),
+        ack_broadcast: !!params["config"]["broadcast"]["ack"],
         confirm_token_ref: confirm_token_ref,
         is_new_api: is_new_api,
         pg_sub_ref: nil,
         pg_change_params: pg_change_params,
-        presence_key: Join.presence_key(configuration),
-        self_broadcast: Join.self_broadcast?(configuration),
+        presence_key: presence_key(params),
+        self_broadcast: !!params["config"]["broadcast"]["self"],
         tenant_topic: tenant_topic,
         channel_name: sub_topic,
         presence_enabled?: presence_enabled?
@@ -113,9 +124,6 @@ defmodule RealtimeWeb.RealtimeChannel do
 
       {:ok, state, assign(socket, assigns)}
     else
-      {:error, :invalid_join_payload, errors, socket} ->
-        log_error(socket, "InvalidJoinPayload", errors)
-
       {:error, :expired_token, msg} ->
         maybe_log_warning(socket, "InvalidJWTToken", msg)
 
@@ -192,23 +200,6 @@ defmodule RealtimeWeb.RealtimeChannel do
     end
   end
 
-  defp configure_socket(socket, params) do
-    case Join.validate(params) do
-      {:ok, configuration} ->
-        socket =
-          socket
-          |> assign_counter()
-          |> assign_presence_counter()
-          |> assign(:private?, Join.private?(configuration))
-          |> assign(:policies, nil)
-
-        {:ok, socket, configuration}
-
-      {:error, :invalid_join_payload, errors} ->
-        {:error, :invalid_join_payload, errors, socket}
-    end
-  end
-
   @impl true
   def handle_info(:update_rate_counter, %{assigns: %{limits: %{max_events_per_second: max}}} = socket) do
     count(socket)
@@ -546,24 +537,40 @@ defmodule RealtimeWeb.RealtimeChannel do
 
   defp count(%{assigns: %{rate_counter: counter}}), do: GenCounter.add(counter.id)
 
-  defp assign_access_token(socket, params) do
-    %{assigns: %{tenant_token: tenant_token, headers: headers}} = socket
+  defp presence_key(params) do
+    case params["config"]["presence"]["key"] do
+      key when is_binary(key) and key != "" -> key
+      _ -> UUID.uuid1()
+    end
+  end
+
+  defp assign_access_token(%{assigns: %{headers: headers}} = socket, params) do
+    access_token = Map.get(params, "access_token") || Map.get(params, "user_token")
     {_, header} = Enum.find(headers, {nil, nil}, fn {k, _} -> k == "x-api-key" end)
 
-    access_token = Map.get(params, "access_token")
-    user_token = Map.get(params, "user_token")
+    case access_token do
+      nil -> assign(socket, :access_token, header)
+      "sb_" <> _ -> assign(socket, :access_token, header)
+      _ -> handle_access_token(socket, params)
+    end
+  end
 
-    access_token =
-      cond do
-        is_binary(access_token) and !String.starts_with?(access_token, "sb_") -> access_token
-        is_binary(user_token) and !String.starts_with?(user_token, "sb_") -> user_token
-        is_binary(tenant_token) and !String.starts_with?(tenant_token, "sb_") -> tenant_token
-        true -> header
-      end
+  defp assign_access_token(socket, params), do: handle_access_token(socket, params)
+
+  defp handle_access_token(%{assigns: %{tenant_token: _tenant_token}} = socket, %{"user_token" => user_token})
+       when is_binary(user_token) do
+    assign(socket, :access_token, user_token)
+  end
 
+  defp handle_access_token(%{assigns: %{tenant_token: _tenant_token}} = socket, %{"access_token" => access_token})
+       when is_binary(access_token) do
     assign(socket, :access_token, access_token)
   end
 
+  defp handle_access_token(%{assigns: %{tenant_token: tenant_token}} = socket, _params) when is_binary(tenant_token) do
+    assign(socket, :access_token, tenant_token)
+  end
+
   defp confirm_token(%{assigns: assigns}) do
     %{jwt_secret: jwt_secret, access_token: access_token} = assigns
 
@@ -630,30 +637,28 @@ defmodule RealtimeWeb.RealtimeChannel do
     })
   end
 
-  defp new_api?(%Join{config: config}) when not is_nil(config), do: true
+  defp new_api?(%{"config" => _}), do: true
   defp new_api?(_), do: false
 
-  defp pg_change_params(true, %Join{config: %Config{postgres_changes: postgres_changes}}, channel_pid, claims, _)
-       when not is_nil(postgres_changes) do
-    postgres_changes
-    |> Enum.reject(&is_nil/1)
-    |> Enum.map(fn %PostgresChange{table: table, event: event, schema: schema, filter: filter} ->
-      params =
-        %{"table" => table, "filter" => filter, "schema" => schema, "event" => event}
-        |> Enum.reject(fn {_, v} -> is_nil(v) end)
-        |> Map.new()
-
-      %{
-        id: UUID.uuid1(),
-        channel_pid: channel_pid,
-        claims: claims,
-        params: params
-      }
-    end)
+  defp pg_change_params(true, params, channel_pid, claims, _) do
+    case get_in(params, ["config", "postgres_changes"]) do
+      [_ | _] = params_list ->
+        params_list
+        |> Enum.reject(&is_nil/1)
+        |> Enum.map(fn params ->
+          %{
+            id: UUID.uuid1(),
+            channel_pid: channel_pid,
+            claims: claims,
+            params: params
+          }
+        end)
+
+      _ ->
+        []
+    end
   end
 
-  defp pg_change_params(true, _, _, _, _), do: []
-
   defp pg_change_params(false, _, channel_pid, claims, sub_topic) do
     params =
       case String.split(sub_topic, ":", parts: 3) do

test/integration/rt_channel_test.exs

@@ -420,18 +420,28 @@ defmodule Realtime.Integration.RtChannelTest do
                      500
     end
 
-    test "nil postgres changes params identified as error", %{tenant: tenant} do
+    test "handle nil postgres changes params as empty param changes", %{tenant: tenant} do
       {socket, _} = get_connection(tenant)
       topic = "realtime:any"
       config = %{postgres_changes: [nil]}
 
-      log =
-        capture_log(fn ->
-          WebsocketClient.join(socket, topic, %{config: config})
-          Process.sleep(500)
-        end)
+      WebsocketClient.join(socket, topic, %{config: config})
 
-      assert log =~ "InvalidJoinPayload"
+      assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}, topic: ^topic}, 200
+      assert_receive %Phoenix.Socket.Message{event: "presence_state", payload: %{}, topic: ^topic}, 500
+
+      refute_receive %Message{
+                       event: "system",
+                       payload: %{
+                         "channel" => "any",
+                         "extension" => "postgres_changes",
+                         "message" => "Subscribed to PostgreSQL",
+                         "status" => "ok"
+                       },
+                       ref: nil,
+                       topic: ^topic
+                     },
+                     1000
     end
   end
 

test/realtime_web/channels/realtime_channel_test.exs

@@ -503,9 +503,12 @@ defmodule RealtimeWeb.RealtimeChannelTest do
 
     test "expired jwt returns a error with sub data if available log_level=warning", %{tenant: tenant} do
       sub = random_string()
+
       api_key = Generators.generate_jwt_token(tenant)
-      claims = %{role: "authenticated", exp: System.system_time(:second) - 1, sub: sub}
-      jwt = Generators.generate_jwt_token(tenant, claims)
+
+      jwt =
+        Generators.generate_jwt_token(tenant, %{role: "authenticated", exp: System.system_time(:second) - 1, sub: sub})
+
       assert {:ok, socket} = connect(UserSocket, %{"log_level" => "warning"}, conn_opts(tenant, api_key))
 
       log =
@@ -672,44 +675,6 @@ defmodule RealtimeWeb.RealtimeChannelTest do
     end
   end
 
-  describe "join payload validations" do
-    test "valid payload allows join", %{tenant: tenant} do
-      jwt = Generators.generate_jwt_token(tenant)
-      {:ok, %Socket{} = socket} = connect(UserSocket, %{"log_level" => "warning"}, conn_opts(tenant, jwt))
-
-      config = %{
-        "config" => %{
-          "private" => false,
-          "broadcast" => %{"ack" => false, "self" => false},
-          "presence" => %{"enabled" => true, "key" => "potato"},
-          "postgres_changes" => [
-            %{"event" => "INSERT", "schema" => "public", "table" => "users", "filter" => "id=eq.1"},
-            %{"event" => "DELETE", "schema" => "public", "table" => "users", "filter" => "id=eq.2"},
-            %{"event" => "UPDATE", "schema" => "public", "table" => "users", "filter" => "id=eq.3"}
-          ]
-        },
-        "access_token" => jwt
-      }
-
-      assert {:ok, _, %Socket{}} = subscribe_and_join(socket, "realtime:test", config)
-    end
-
-    test "invalid payload returns error", %{tenant: tenant} do
-      jwt = Generators.generate_jwt_token(tenant)
-      {:ok, %Socket{} = socket} = connect(UserSocket, %{"log_level" => "warning"}, conn_opts(tenant, jwt))
-
-      log =
-        capture_log(fn ->
-          assert {:error, %{reason: reason}} =
-                   subscribe_and_join(socket, "realtime:test", %{"config" => "potato"})
-
-          assert reason =~ "unable to parse, expected a map"
-        end)
-
-      assert log =~ "InvalidJoinPayload"
-    end
-  end
-
   test "registers transport pid and channel pid per tenant", %{tenant: tenant} do
     jwt = Generators.generate_jwt_token(tenant)
     {:ok, %Socket{} = socket} = connect(UserSocket, %{"log_level" => "warning"}, conn_opts(tenant, jwt))