fix: properly handle match error

.github/workflows/tests.yml

@@ -3,6 +3,9 @@ on:
   pull_request:
     branches:
       - main
+  push:
+    branches:
+      - main
 jobs:
   format:
     name: Tests

lib/realtime_web/channels/realtime_channel.ex

@@ -420,6 +420,9 @@ defmodule RealtimeWeb.RealtimeChannel do
       {:error, :expected_claims_map} ->
         shutdown_response(socket, "Token claims must be a map")
 
+      {:error, :unable_to_set_policies, _msg} ->
+        shutdown_response(socket, "Realtime was unable to connect to the project database")
+
       {:error, error} ->
         shutdown_response(socket, inspect(error))
     end

mix.exs

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

test/integration/rt_channel_test.exs

@@ -6,6 +6,7 @@
   use RealtimeWeb.ConnCase, async: false
   import ExUnit.CaptureLog
   import Generators
+  import Mock
 
   require Logger
 
@@ -19,6 +20,7 @@
   alias Realtime.Integration.WebsocketClient
   alias Realtime.Repo
   alias Realtime.Tenants
+  alias Realtime.Tenants.Authorization
   alias Realtime.Tenants.Migrations
 
   @moduletag :capture_log
@@ -909,7 +911,6 @@
 
       # token becomes a string in between joins so it needs to be handled by the channel and not the socket
       Process.sleep(1000)
-      realtime_topic = "realtime:#{topic}"
       WebsocketClient.join(socket, realtime_topic, %{config: config, access_token: "potato"})
 
       assert_receive %Message{
@@ -924,6 +925,58 @@
 
       assert_receive %Message{event: "phx_close"}
     end
+
+    @tag policies: [
+           :authenticated_read_broadcast_and_presence,
+           :authenticated_write_broadcast_and_presence
+         ]
+    test "handles RPC error on token refreshed", %{topic: topic} do
+      with_mocks [
+        {Authorization, [:passthrough], build_authorization_params: &passthrough([&1])},
+        {Authorization, [:passthrough],
+         get_read_authorizations: [
+           in_series([:_, :_, :_], [&passthrough([&1, &2, &3]), {:error, "RPC Error"}])
+         ]}
+      ] do
+        {socket, access_token} = get_connection("authenticated")
+        config = %{broadcast: %{self: true}, private: true}
+        realtime_topic = "realtime:#{topic}"
+
+        WebsocketClient.join(socket, realtime_topic, %{config: config, access_token: access_token})
+
+        assert_receive %Phoenix.Socket.Message{event: "phx_reply"}, 500
+        assert_receive %Phoenix.Socket.Message{event: "presence_state"}, 500
+
+        # Update token to force update
+        {:ok, access_token} =
+          generate_token(%{:exp => System.system_time(:second) + 1000, role: "authenticated"})
+
+        log =
+          capture_log(fn ->
+            WebsocketClient.send_event(socket, realtime_topic, "access_token", %{
+              "access_token" => access_token
+            })
+
+            assert_receive %Phoenix.Socket.Message{
+                             event: "system",
+                             payload: %{
+                               "status" => "error",
+                               "extension" => "system",
+                               "message" =>
+                                 "Realtime was unable to connect to the project database"
+                             },
+                             topic: ^realtime_topic,
+                             join_ref: nil,
+                             ref: nil
+                           },
+                           500
+
+            assert_receive %Phoenix.Socket.Message{event: "phx_close", topic: ^realtime_topic}
+          end)
+
+        assert log =~ "Realtime was unable to connect to the project database"
+      end
+    end
   end
 
   describe "handle broadcast changes" do
@@ -1175,7 +1228,7 @@
     } do
       change_tenant_configuration(:private_only, true)
 
       Realtime.Tenants.Cache.invalidate_tenant_cache(@external_id)
 
       Process.sleep(100)
 
@@ -1513,10 +1566,10 @@
   end
 
   def setup_trigger(%{tenant: tenant, topic: topic} = context) do
     Realtime.Tenants.Connect.shutdown(@external_id)
     Process.sleep(500)
 
     {:ok, db_conn} = Realtime.Tenants.Connect.connect(@external_id)
 
     random_name = String.downcase("test_#{random_string()}")
     query = "CREATE TABLE #{random_name} (id serial primary key, details text)"
@@ -1553,7 +1606,7 @@
       {:ok, db_conn} = Database.connect(tenant, "realtime_test", :stop)
       query = "DROP TABLE #{random_name} CASCADE"
       Postgrex.query!(db_conn, query, [])
       Realtime.Tenants.Connect.shutdown(db_conn)
 
       Process.sleep(500)
     end)
@@ -1566,9 +1619,9 @@
   defp change_tenant_configuration(limit, value) do
     @external_id
     |> Realtime.Tenants.get_tenant_by_external_id()
     |> Realtime.Api.Tenant.changeset(%{limit => value})
     |> Realtime.Repo.update!()
 
     Realtime.Tenants.Cache.invalidate_tenant_cache(@external_id)
   end
 end

test/realtime_web/channels/realtime_channel_test.exs

@@ -6,9 +6,10 @@ defmodule RealtimeWeb.RealtimeChannelTest do
   import ExUnit.CaptureLog
 
   alias Phoenix.Socket
+  alias Realtime.Tenants.Authorization
   alias RealtimeWeb.ChannelsAuthorization
-  alias RealtimeWeb.UserSocket
   alias RealtimeWeb.Joken.CurrentTime
+  alias RealtimeWeb.UserSocket
 
   @tenant_external_id "dev_tenant"