fix: create containers per tenant

.github/workflows/tests.yml

@@ -61,6 +61,6 @@ jobs:
       - name: Start epmd
         run: epmd -daemon
       - name: Run tests
-        run: MIX_ENV=test mix coveralls.github --trace
+        run: MIX_ENV=test MAX_CASES=2 mix coveralls.github --trace
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

config/config.exs

@@ -13,7 +13,7 @@ config :realtime,
 
 # Configures the endpoint
 config :realtime, RealtimeWeb.Endpoint,
-  url: [host: "localhost"],
+  url: [host: "127.0.0.1"],
   secret_key_base: "ktyW57usZxrivYdvLo9os7UGcUUZYKchOMHT3tzndmnHuxD09k+fQnPUmxlPMUI3",
   render_errors: [view: RealtimeWeb.ErrorView, accepts: ~w(html json), layout: false],
   pubsub_server: Realtime.PubSub,

config/runtime.exs

@@ -4,7 +4,7 @@ config :logflare_logger_backend,
   url: System.get_env("LOGFLARE_LOGGER_BACKEND_URL", "https://api.logflare.app")
 
 app_name = System.get_env("APP_NAME", "")
-default_db_host = System.get_env("DB_HOST", "localhost")
+default_db_host = System.get_env("DB_HOST", "127.0.0.1")
 username = System.get_env("DB_USER", "postgres")
 password = System.get_env("DB_PASSWORD", "postgres")
 database = System.get_env("DB_NAME", "postgres")

config/test.exs

@@ -21,7 +21,7 @@ for repo <- [
     username: "postgres",
     password: "postgres",
     database: "realtime_test",
-    hostname: "localhost",
+    hostname: "127.0.0.1",
     pool: Ecto.Adapters.SQL.Sandbox
 end
 
@@ -43,7 +43,9 @@ config :joken,
   current_time_adapter: RealtimeWeb.Joken.CurrentTime.Mock
 
 # Print only errors during test
-config :logger, level: :warning
+config :logger,
+  compile_time_purge_matching: [[module: Postgrex], [module: DBConnection]],
+  level: :warning
 
 # Configures Elixir's Logger
 config :logger, :console,

lib/realtime/context_cache.ex

@@ -9,14 +9,9 @@ defmodule Realtime.ContextCache do
     cache = cache_name(context)
     cache_key = {{fun, arity}, args}
 
-    case Cachex.fetch(cache, cache_key, fn {{_fun, _arity}, args} ->
-           {:commit, {:cached, apply(context, fun, args)}}
-         end) do
-      {:commit, {:cached, value}} ->
-        value
-
-      {:ok, {:cached, value}} ->
-        value
+    case Cachex.fetch(cache, cache_key, fn {{_fun, _arity}, args} -> {:commit, {:cached, apply(context, fun, args)}} end) do
+      {:commit, {:cached, value}} -> value
+      {:ok, {:cached, value}} -> value
     end
   end
 

lib/realtime/tenants.ex

@@ -257,7 +257,10 @@ defmodule Realtime.Tenants do
     |> Cache.get_tenant_by_external_id()
     |> Tenant.changeset(%{suspend: true})
     |> Repo.update!()
-    |> tap(fn _ -> broadcast_operation_event(:suspend_tenant, external_id) end)
+    |> tap(fn _ ->
+      Cache.invalidate_tenant_cache(external_id)
+      broadcast_operation_event(:suspend_tenant, external_id)
+    end)
   end
 
   @doc """
@@ -269,7 +272,10 @@ defmodule Realtime.Tenants do
     |> Cache.get_tenant_by_external_id()
     |> Tenant.changeset(%{suspend: false})
     |> Repo.update!()
-    |> tap(fn _ -> broadcast_operation_event(:unsuspend_tenant, external_id) end)
+    |> tap(fn _ ->
+      Cache.invalidate_tenant_cache(external_id)
+      broadcast_operation_event(:unsuspend_tenant, external_id)
+    end)
   end
 
   defp broadcast_operation_event(action, external_id) do

lib/realtime/tenants/connect.ex

@@ -25,12 +25,6 @@ defmodule Realtime.Tenants.Connect do
   alias Realtime.Tenants.Migrations
   alias Realtime.UsersCounter
 
-  @pipes [
-    GetTenant,
-    CheckConnection,
-    StartCounters,
-    RegisterProcess
-  ]
   @rpc_timeout_default 30_000
   @check_connected_user_interval_default 50_000
   @connected_users_bucket_shutdown [0, 0, 0, 0, 0, 0]
@@ -48,7 +42,7 @@ defmodule Realtime.Tenants.Connect do
   """
   @spec lookup_or_start_connection(binary(), keyword()) ::
           {:ok, pid()} | {:error, term()}
-  def lookup_or_start_connection(tenant_id, opts \\ []) do
+  def lookup_or_start_connection(tenant_id, opts \\ []) when is_binary(tenant_id) do
     case get_status(tenant_id) do
       {:ok, conn} ->
         {:ok, conn}
@@ -171,7 +165,14 @@ defmodule Realtime.Tenants.Connect do
   def init(%{tenant_id: tenant_id} = state) do
     Logger.metadata(external_id: tenant_id, project: tenant_id)
 
-    case Piper.run(@pipes, state) do
+    pipes = [
+      GetTenant,
+      CheckConnection,
+      StartCounters,
+      RegisterProcess
+    ]
+
+    case Piper.run(pipes, state) do
       {:ok, acc} ->
         {:ok, acc, {:continue, :run_migrations}}
 

lib/realtime/tenants/connect/check_connection.ex

@@ -10,11 +10,8 @@ defmodule Realtime.Tenants.Connect.CheckConnection do
     %{tenant: tenant} = acc
 
     case Database.check_tenant_connection(tenant) do
-      {:ok, conn} ->
-        {:ok, %{acc | db_conn_pid: conn, db_conn_reference: Process.monitor(conn)}}
-
-      {:error, error} ->
-        {:error, error}
+      {:ok, conn} -> {:ok, %{acc | db_conn_pid: conn, db_conn_reference: Process.monitor(conn)}}
+      {:error, error} -> {:error, error}
     end
   end
 end

lib/realtime/tenants/janitor.ex

@@ -60,9 +60,7 @@ defmodule Realtime.Tenants.Janitor do
   @table_name Realtime.Tenants.Connect
   @syn_table :"syn_registry_by_name_Elixir.Realtime.Tenants.Connect"
   @matchspec [{{:"$1"}, [], [:"$1"]}]
-  @syn_matchspec [
-    {{:"$1", :"$2", :"$3", :"$4", :"$5", Node.self()}, [], [:"$1"]}
-  ]
+  @syn_matchspec [{{:"$1", :"$2", :"$3", :"$4", :"$5", Node.self()}, [], [:"$1"]}]
 
   @impl true
   def handle_info(:delete_old_messages, state) do

mix.exs

@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
   def project do
     [
       app: :realtime,
-      version: "2.34.24",
+      version: "2.34.25",
       elixir: "~> 1.17.3",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,
@@ -101,7 +101,6 @@ defmodule Realtime.MixProject do
         "ecto.create --quiet",
         "run priv/repo/seeds_before_migration.exs",
         "ecto.migrate --migrations-path=priv/repo/migrations",
-        "run priv/repo/seeds_after_migration.exs",
         "test"
       ],
       "assets.deploy": ["esbuild default --minify", "tailwind default --minify", "phx.digest"]

priv/repo/seeds.exs

@@ -4,7 +4,7 @@ import Ecto.Adapters.SQL, only: [query: 3]
 
 tenant_name = System.get_env("SELF_HOST_TENANT_NAME", "realtime-dev")
 env = if :ets.whereis(Mix.State) != :undefined, do: Mix.env(), else: :prod
-default_db_host = if env in [:dev, :test], do: "localhost", else: "host.docker.internal"
+default_db_host = if env in [:dev, :test], do: "127.0.0.1", else: "host.docker.internal"
 
 Repo.transaction(fn ->
   case Repo.get_by(Tenant, external_id: tenant_name) do

rel/overlays/config.example.yml

@@ -1,6 +1,6 @@
 endpoint_port: 4000
 db_repo:
-  - hostname: "localhost"
+  - hostname: "127.0.0.1"
     username: "postgres"
     password: "postgres"
     database: "postgres"

test/api_jwt_secret_test.exs

@@ -1,5 +1,6 @@
 defmodule RealtimeWeb.ApiJwtSecretTest do
-  use RealtimeWeb.ConnCase
+  # async: false due to usage of mock
+  use RealtimeWeb.ConnCase, async: false
   import Mock
   alias RealtimeWeb.JwtVerification
 

test/integration/rt_channel_test.exs

@@ -1,8 +1,7 @@
 Code.require_file("../support/websocket_client.exs", __DIR__)
 
 defmodule Realtime.Integration.RtChannelTest do
-  # async: false due to the fact that multiple operations against the database will use the same connection
-
+  # async: false due to the fact that multiple operations against the same tenant and usage of mocks
   use RealtimeWeb.ConnCase, async: false
   import ExUnit.CaptureLog
   import Generators
@@ -14,16 +13,14 @@ defmodule Realtime.Integration.RtChannelTest do
   alias Phoenix.Socket.Message
   alias Phoenix.Socket.V1
   alias Postgrex
-  alias Realtime.Api.Tenant
   alias Realtime.Database
   alias Realtime.Integration.RtChannelTest.Endpoint
   alias Realtime.Integration.WebsocketClient
   alias Realtime.RateCounter
-  alias Realtime.Repo
   alias Realtime.Tenants
   alias Realtime.Tenants.Authorization
   alias Realtime.Tenants.Cache
-  alias Realtime.Tenants.Migrations
+
   @moduletag :capture_log
   @port 4002
   @serializer V1.JSONSerializer
@@ -78,8 +75,9 @@ defmodule Realtime.Integration.RtChannelTest do
     RateCounter.stop(@external_id)
     Cache.invalidate_tenant_cache(@external_id)
     Process.sleep(500)
-    [tenant] = Tenant |> Repo.all() |> Repo.preload(:extensions)
-    :ok = Migrations.run_migrations(tenant)
+
+    tenant = Tenants.get_tenant_by_external_id(@external_id)
+
     %{tenant: tenant}
   end
 
@@ -1265,7 +1263,7 @@ defmodule Realtime.Integration.RtChannelTest do
           get_connection("authenticated", %{:exp => System.system_time(:second) - 1000})
         end)
 
-      assert log =~ "InvalidJWTToken: Token as expired 1000 seconds ago"
+      assert log =~ "InvalidJWTToken: Token as expired"
     end
   end
 
@@ -1466,7 +1464,7 @@ defmodule Realtime.Integration.RtChannelTest do
     claims =
       Map.merge(
         %{
-          ref: "localhost",
+          ref: "127.0.0.1",
           iat: System.system_time(:second),
           exp: System.system_time(:second) + 604_800
         },