fix: improve crud remote calls

This approach prevents sending more information than needed reducing the amount of data send via gen_rpc. We also remove some unnecessary calls to primary in the delete controller operation

Author: filipecabaco

lib/realtime/api.ex

@@ -115,9 +115,13 @@ defmodule Realtime.Api do
     Logger.debug("create_tenant #{inspect(attrs, pretty: true)}")
     tenant_id = Map.get(attrs, :external_id) || Map.get(attrs, "external_id")
 
-    %Tenant{}
-    |> Tenant.changeset(attrs)
-    |> region_aware_write(:insert, tenant_id)
+    if master_region?() do
+      %Tenant{}
+      |> Tenant.changeset(attrs)
+      |> Repo.insert()
+    else
+      call(:create_tenant, [attrs], tenant_id)
+    end
   end
 
   @doc """
@@ -132,9 +136,17 @@ defmodule Realtime.Api do
       {:error, %Ecto.Changeset{}}
 
   """
-  def update_tenant(%Tenant{} = tenant, attrs) do
+  def update_tenant(tenant_id, attrs) when is_binary(tenant_id) do
+    replica = Replica.replica()
+    tenant = replica.get_by(Tenant, external_id: tenant_id)
+    update_tenant(tenant, attrs)
+  end
+
+  def update_tenant(%Tenant{external_id: external_id} = tenant, attrs) do
     changeset = Tenant.changeset(tenant, attrs)
-    updated = region_aware_write(changeset, :update, tenant.external_id)
+
+    updated =
+      if master_region?(), do: Repo.update(changeset), else: call(:update_tenant, [external_id, attrs], external_id)
 
     case updated do
       {:ok, tenant} ->
@@ -150,38 +162,23 @@ defmodule Realtime.Api do
     updated
   end
 
-  @doc """
-  Deletes a tenant.
-
-  ## Examples
-
-      iex> delete_tenant(tenant)
-      {:ok, %Tenant{}}
-
-      iex> delete_tenant(tenant)
-      {:error, %Ecto.Changeset{}}
-
-  """
-  def delete_tenant(%Tenant{} = tenant), do: Repo.delete(tenant)
-
   @spec delete_tenant_by_external_id(String.t()) :: boolean()
   def delete_tenant_by_external_id(id) do
-    from(t in Tenant, where: t.external_id == ^id)
-    |> region_aware_write(:delete_all, id)
-    |> case do
-      {num, _} when num > 0 -> true
-      _ -> false
+    if master_region?() do
+      from(t in Tenant, where: t.external_id == ^id)
+      |> Repo.delete_all()
+      |> case do
+        {num, _} when num > 0 -> true
+        _ -> false
+      end
+    else
+      call(:delete_tenant_by_external_id, [id], id)
     end
   end
 
-  @spec get_tenant_by_external_id(String.t(), atom()) :: Tenant.t() | nil
-  def get_tenant_by_external_id(external_id, repo \\ :replica)
-      when repo in [:primary, :replica] do
-    repo =
-      case repo do
-        :primary -> Repo
-        :replica -> Replica.replica()
-      end
+  @spec get_tenant_by_external_id(String.t()) :: Tenant.t() | nil
+  def get_tenant_by_external_id(external_id) do
+    repo = Replica.replica()
 
     Tenant
     |> repo.get_by(external_id: external_id)
@@ -195,26 +192,36 @@ defmodule Realtime.Api do
   end
 
   def rename_settings_field(from, to) do
-    for extension <- list_extensions("postgres_cdc_rls") do
-      {value, settings} = Map.pop(extension.settings, from)
-      new_settings = Map.put(settings, to, value)
-
-      extension
-      |> Changeset.cast(%{settings: new_settings}, [:settings])
-      |> region_aware_write(:update!, extension.external_id)
+    if master_region?() do
+      for extension <- list_extensions("postgres_cdc_rls") do
+        {value, settings} = Map.pop(extension.settings, from)
+        new_settings = Map.put(settings, to, value)
+
+        extension
+        |> Changeset.cast(%{settings: new_settings}, [:settings])
+        |> Repo.update()
+      end
+    else
+      call(:rename_settings_field, [from, to], from)
     end
   end
 
+  @spec preload_counters(nil | Realtime.Api.Tenant.t(), any()) :: nil | Realtime.Api.Tenant.t()
   @doc """
   Updates the migrations_ran field for a tenant.
   """
   @spec update_migrations_ran(binary(), integer()) :: {:ok, Tenant.t()} | {:error, term()}
   def update_migrations_ran(external_id, count) do
-    external_id
-    |> Cache.get_tenant_by_external_id()
-    |> Tenant.changeset(%{migrations_ran: count})
-    |> region_aware_write(:update!, external_id)
-    |> tap(fn _ -> Cache.distributed_invalidate_tenant_cache(external_id) end)
+    if master_region?() do
+      tenant = Cache.get_tenant_by_external_id(external_id)
+
+      tenant
+      |> Tenant.changeset(%{migrations_ran: count})
+      |> Repo.update()
+      |> tap(fn _ -> Cache.distributed_invalidate_tenant_cache(external_id) end)
+    else
+      call(:update_migrations_ran, [external_id, count], external_id)
+    end
   end
 
   def preload_counters(nil), do: nil
@@ -257,21 +264,13 @@ defmodule Realtime.Api do
 
   defp maybe_restart_db_connection(_changeset), do: nil
 
-  defp local_call? do
+  defp master_region? do
     region = Application.get_env(:realtime, :region)
     master_region = Application.get_env(:realtime, :master_region) || region
     region == master_region
   end
 
-  defp region_aware_write(%struct{} = argument, operation, tenant_id) when struct in [Changeset, Ecto.Query] do
-    if local_call?(),
-      do: local_call(operation, [argument], tenant_id),
-      else: remote_call(operation, [argument], tenant_id)
-  end
-
-  defp local_call(operation, args, _tenant_id), do: apply(Realtime.Repo, operation, args)
-
-  defp remote_call(operation, args, tenant_id) do
+  defp call(operation, args, tenant_id) do
     master_region = Application.get_env(:realtime, :master_region)
 
     with {:ok, master_node} <- Nodes.node_from_region(master_region, self()),
@@ -281,7 +280,7 @@ defmodule Realtime.Api do
   end
 
   defp wrapped_call(master_node, operation, args, tenant_id) do
-    case GenRpc.call(master_node, Realtime.Repo, operation, args, tenant_id: tenant_id) do
+    case GenRpc.call(master_node, __MODULE__, operation, args, tenant_id: tenant_id) do
       {:error, :rpc_error, reason} -> {:error, reason}
       {:error, reason} -> {:error, reason}
       result -> {:ok, result}

lib/realtime/repo_replica.ex

@@ -51,11 +51,9 @@ defmodule Realtime.Repo.Replica do
     # Do not create module if replica isn't set or configuration is not present
     cond do
       is_nil(replica) ->
-        Logger.info("Replica region not found, defaulting to Realtime.Repo")
         Realtime.Repo
 
       is_nil(replica_conf) ->
-        Logger.info("Replica config not found for #{region} region")
         Realtime.Repo
 
       true ->

lib/realtime_web/controllers/tenant_controller.ex

@@ -192,7 +192,7 @@ defmodule RealtimeWeb.TenantController do
   def delete(conn, %{"tenant_id" => tenant_id}) do
     stop_all_timeout = Enum.count(PostgresCdc.available_drivers()) * 1_000
 
-    with %Tenant{} = tenant <- Api.get_tenant_by_external_id(tenant_id, :primary),
+    with %Tenant{} = tenant <- Api.get_tenant_by_external_id(tenant_id),
          _ <- Tenants.suspend_tenant_by_external_id(tenant_id),
          true <- Api.delete_tenant_by_external_id(tenant_id),
          true <- Cache.distributed_invalidate_tenant_cache(tenant_id),

mix.exs

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

test/integration/region_aware_routing_test.exs

@@ -49,8 +49,8 @@ defmodule Realtime.Integration.RegionAwareRoutingTest do
 
     Mimic.expect(Realtime.GenRpc, :call, fn node, mod, func, args, opts ->
       assert node == master_node
-      assert mod == Realtime.Repo
-      assert func == :insert
+      assert mod == Realtime.Api
+      assert func == :create_tenant
       assert opts[:tenant_id] == external_id
 
       call_original(GenRpc, :call, [node, mod, func, args, opts])
@@ -80,16 +80,16 @@ defmodule Realtime.Integration.RegionAwareRoutingTest do
     Realtime.GenRpc
     |> Mimic.expect(:call, fn node, mod, func, args, opts ->
       assert node == master_node
-      assert mod == Realtime.Repo
-      assert func == :insert
+      assert mod == Realtime.Api
+      assert func == :create_tenant
       assert opts[:tenant_id] == tenant_attrs["external_id"]
 
       call_original(GenRpc, :call, [node, mod, func, args, opts])
     end)
     |> Mimic.expect(:call, fn node, mod, func, args, opts ->
       assert node == master_node
-      assert mod == Realtime.Repo
-      assert func == :update
+      assert mod == Realtime.Api
+      assert func == :update_tenant
       assert opts[:tenant_id] == tenant_attrs["external_id"]
 
       call_original(GenRpc, :call, [node, mod, func, args, opts])
@@ -123,16 +123,16 @@ defmodule Realtime.Integration.RegionAwareRoutingTest do
     Realtime.GenRpc
     |> Mimic.expect(:call, fn node, mod, func, args, opts ->
       assert node == master_node
-      assert mod == Realtime.Repo
-      assert func == :insert
+      assert mod == Realtime.Api
+      assert func == :create_tenant
       assert opts[:tenant_id] == tenant_attrs["external_id"]
 
       call_original(GenRpc, :call, [node, mod, func, args, opts])
     end)
     |> Mimic.expect(:call, fn node, mod, func, args, opts ->
       assert node == master_node
-      assert mod == Realtime.Repo
-      assert func == :delete_all
+      assert mod == Realtime.Api
+      assert func == :delete_tenant_by_external_id
       assert opts[:tenant_id] == tenant_attrs["external_id"]
 
       call_original(GenRpc, :call, [node, mod, func, args, opts])
@@ -164,16 +164,16 @@ defmodule Realtime.Integration.RegionAwareRoutingTest do
     Realtime.GenRpc
     |> Mimic.expect(:call, fn node, mod, func, args, opts ->
       assert node == master_node
-      assert mod == Realtime.Repo
-      assert func == :insert
+      assert mod == Realtime.Api
+      assert func == :create_tenant
       assert opts[:tenant_id] == tenant_attrs["external_id"]
 
       call_original(GenRpc, :call, [node, mod, func, args, opts])
     end)
     |> Mimic.expect(:call, fn node, mod, func, args, opts ->
       assert node == master_node
-      assert mod == Realtime.Repo
-      assert func == :update!
+      assert mod == Realtime.Api
+      assert func == :update_migrations_ran
       assert opts[:tenant_id] == tenant_attrs["external_id"]
 
       call_original(GenRpc, :call, [node, mod, func, args, opts])
@@ -184,7 +184,7 @@ defmodule Realtime.Integration.RegionAwareRoutingTest do
     new_migrations_ran = 5
     result = Api.update_migrations_ran(tenant.external_id, new_migrations_ran)
 
-    assert %Tenant{} = updated = result
+    assert {:ok, updated} = result
     assert updated.migrations_ran == new_migrations_ran
 
     reloaded = Realtime.Repo.get(Tenant, tenant.id)

test/realtime/api_test.exs

@@ -232,14 +232,6 @@ defmodule Realtime.ApiTest do
     end
   end
 
-  describe "delete_tenant/1" do
-    test "deletes the tenant" do
-      tenant = tenant_fixture()
-      assert {:ok, %Tenant{}} = Api.delete_tenant(tenant)
-      assert_raise Ecto.NoResultsError, fn -> Api.get_tenant!(tenant.id) end
-    end
-  end
-
   describe "delete_tenant_by_external_id/1" do
     test "deletes the tenant" do
       tenant = tenant_fixture()

test/test_helper.exs

@@ -15,7 +15,7 @@ end
 
 {:ok, _pid} = Containers.start_link(max_cases)
 
-for tenant <- Api.list_tenants(), do: Api.delete_tenant(tenant)
+for tenant <- Api.list_tenants(), do: Api.delete_tenant_by_external_id(tenant.external_id)
 
 tenant_name = "dev_tenant"
 tenant = Containers.initialize(tenant_name)
@@ -53,6 +53,7 @@ Mimic.copy(Realtime.Database)
 Mimic.copy(Realtime.GenCounter)
 Mimic.copy(Realtime.GenRpc)
 Mimic.copy(Realtime.Nodes)
+Mimic.copy(Realtime.Repo.Replica)
 Mimic.copy(Realtime.RateCounter)
 Mimic.copy(Realtime.Tenants.Authorization)
 Mimic.copy(Realtime.Tenants.Cache)