feat: caching internal db results

config/runtime.exs

@@ -154,7 +154,9 @@ if config_env() != :test do
     prom_poll_rate: System.get_env("PROM_POLL_RATE", "15000") |> String.to_integer(),
     global_upstream_ca: upstream_ca,
     global_downstream_cert: downstream_cert,
-    global_downstream_key: downstream_key
+    global_downstream_key: downstream_key,
+    cache_publ_name: System.get_env("CACHE_PUBL_NAME", "supavisor_cache"),
+    cache_repl_slot: System.get_env("CACHE_REPL_SLOT", "supavisor_cache")
 
   config :supavisor, Supavisor.Repo,
     url: System.get_env("DATABASE_URL", "ecto://postgres:postgres@localhost:6432/postgres"),

lib/supavisor.ex

@@ -134,7 +134,7 @@ defmodule Supavisor do
     {method, secrets} = auth_secrets
     Logger.debug("Starting pool for #{inspect({tenant, user_alias, method})}")
 
-    case Tenants.get_pool_config(tenant, user_alias) do
+    case Tenants.get_pool_config(:cached, tenant, user_alias) do
       %Tenant{} = tenant_record ->
         %{
           db_host: db_host,

lib/supavisor/application.ex

@@ -5,6 +5,8 @@ defmodule Supavisor.Application do
 
   use Application
   require Logger
+  import Cachex.Spec
+
   alias Supavisor.Monitoring.PromEx
 
   @impl true
@@ -72,8 +74,13 @@ defmodule Supavisor.Application do
         child_spec: DynamicSupervisor, strategy: :one_for_one, name: Supavisor.DynamicSupervisor
       },
       Supavisor.Vault,
-      {Cachex, name: Supavisor.Cache},
+      {Cachex,
+       [
+         name: Supavisor.Cache,
+         limit: limit(size: 50_000, policy: Cachex.Policy.LRW, reclaim: 0.5)
+       ]},
       Supavisor.TenantsMetrics,
+      Supavisor.CacheBuster,
       # Start the Endpoint (http/https)
       SupavisorWeb.Endpoint
     ]

lib/supavisor/cache_buster.ex

@@ -0,0 +1,166 @@
+defmodule Supavisor.CacheBuster do
+  @moduledoc false
+
+  use Postgrex.ReplicationConnection
+  require Logger
+  alias Supavisor.Protocol.WalDecoder.Decoder
+
+  alias Decoder.Messages.{
+    Relation,
+    Update,
+    Delete
+  }
+
+  def start_link(args) do
+    opts =
+      Application.get_env(:supavisor, Supavisor.Repo)[:url]
+      |> Ecto.Repo.Supervisor.parse_url()
+
+    Postgrex.ReplicationConnection.start_link(__MODULE__, args, opts)
+  end
+
+  @spec stop(pid) :: :ok
+  def stop(pid) do
+    GenServer.stop(pid)
+  end
+
+  @impl true
+  def init(_args) do
+    slot_name = Application.get_env(:supavisor, :cache_repl_slot)
+
+    state = %{
+      publication: Application.get_env(:supavisor, :cache_publ_name),
+      slot: slot_name <> "#{:erlang.phash2(node())}",
+      rels: %{},
+      step: nil
+    }
+
+    {:ok, state}
+  end
+
+  @impl true
+  def handle_connect(state) do
+    query = "CREATE_REPLICATION_SLOT #{state.slot} TEMPORARY LOGICAL pgoutput NOEXPORT_SNAPSHOT"
+    {:query, query, %{state | step: :create_slot}}
+  end
+
+  @impl true
+  def handle_result(results, %{step: :create_slot} = state)
+      when is_list(results) do
+    Logger.info("Replication slot created: #{inspect(results)}")
+
+    query =
+      "START_REPLICATION SLOT #{state.slot} LOGICAL 0/0 (proto_version '1', publication_names '#{state.publication}')"
+
+    {:stream, query, [], %{state | step: :streaming}}
+  end
+
+  def handle_result(results, state) do
+    Logger.error("Replication is failed: #{inspect(results)}")
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_data(<<?w, _header::192, msg::binary>>, state) do
+    new_state =
+      Decoder.decode_message(msg)
+      |> process_message(state)
+
+    {:noreply, new_state}
+  end
+
+  # keepalive
+  def handle_data(<<?k, _::128, 0>>, state) do
+    {:noreply, [], state}
+  end
+
+  def handle_data(<<?k, wal_end::64, _::64, 1>>, state) do
+    messages = [
+      <<?r, wal_end + 1::64, wal_end + 1::64, wal_end + 1::64, current_time()::64, 0>>
+    ]
+
+    {:noreply, messages, state}
+  end
+
+  def handle_data(data, state) do
+    Logger.error("Unknown data: #{inspect(data)}")
+    {:noreply, state}
+  end
+
+  defp process_message(%Relation{id: id, columns: columns, namespace: schema, name: table}, state) do
+    columns =
+      Enum.map(columns, fn %{name: name, type: type} ->
+        %{name: name, type: type}
+      end)
+
+    put_in(state, [:rels, id], {columns, schema, table})
+  end
+
+  defp process_message(%Update{} = msg, state) do
+    drop_cache(msg, state.rels[msg.relation_id])
+    state
+  end
+
+  defp process_message(%Delete{} = msg, state) do
+    drop_cache(msg, state.rels[msg.relation_id])
+    state
+  end
+
+  defp process_message(_msg, state) do
+    state
+  end
+
+  def drop_cache(msg, relation) do
+    Logger.debug("Got message: #{inspect(msg)}")
+    {columns, _schema, table} = relation
+
+    record = data_tuple_to_map(columns, msg.old_tuple_data)
+
+    tenant =
+      if table == "tenants" do
+        record["external_id"]
+      else
+        record["tenant_external_id"]
+      end
+
+    Logger.warning("Got update for tenant: #{inspect(tenant)}")
+
+    Supavisor.dirty_terminate(tenant)
+  end
+
+  ## Internal functions
+
+  defp data_tuple_to_map(column, tuple_data) do
+    column
+    |> Enum.with_index()
+    |> Enum.reduce_while(%{}, fn {column_map, index}, acc ->
+      case column_map do
+        %{name: column_name, type: column_type}
+        when is_binary(column_name) and is_binary(column_type) ->
+          try do
+            {:ok, elem(tuple_data, index)}
+          rescue
+            ArgumentError -> :error
+          end
+          |> case do
+            {:ok, record} ->
+              {:cont,
+               Map.put(
+                 acc,
+                 column_name,
+                 record
+               )}
+
+            :error ->
+              {:halt, acc}
+          end
+
+        _ ->
+          {:cont, acc}
+      end
+    end)
+  end
+
+  @epoch DateTime.to_unix(~U[2000-01-01 00:00:00Z], :microsecond)
+  defp current_time(), do: System.os_time(:microsecond) - @epoch
+end

lib/supavisor/client_handler.ex

@@ -112,7 +112,7 @@ defmodule Supavisor.ClientHandler do
 
     sni_hostname = try_get_sni(sock)
 
-    case Tenants.get_user(user, external_id, sni_hostname) do
+    case Tenants.get_user(:cached, user, external_id, sni_hostname) do
       {:ok, info} ->
         if info.tenant.enforce_ssl and !data.ssl do
           Logger.error("Tenant is not allowed to connect without SSL, user #{user}")
@@ -522,16 +522,8 @@ defmodule Supavisor.ClientHandler do
 
   def auth_secrets(%{user: user, tenant: tenant} = info, db_user) do
     cache_key = {:secrets, tenant.external_id, user}
-
-    case Cachex.fetch(Supavisor.Cache, cache_key, fn _key ->
-           {:commit, {:cached, get_secrets(info, db_user)}, ttl: 15_000}
-         end) do
-      {_, {:cached, value}} ->
-        value
-
-      {_, {:cached, value}, _} ->
-        value
-    end
+    fun = fn -> get_secrets(info, db_user) end
+    H.cached(cache_key, fun, 60_000)
   end
 
   @spec get_secrets(map, String.t()) :: {:ok, {:auth_query, fun()}} | {:error, term()}

lib/supavisor/helpers.ex

@@ -1,5 +1,6 @@
 defmodule Supavisor.Helpers do
   @moduledoc false
+  require Logger
 
   @spec check_creds_get_ver(map) :: {:ok, String.t()} | {:error, String.t()}
 
@@ -51,7 +52,7 @@
         Postgrex.query(conn, "select version()", [])
         |> case do
           {:ok, %{rows: [[version]]}} ->
             if !params["require_user"] do
               case get_user_secret(conn, params["auth_query"], user["db_user"]) do
                 {:ok, _} ->
                   {:halt, {:ok, version}}
@@ -303,4 +304,21 @@
   def parse_server_first(message, nonce) do
     :pgo_scram.parse_server_first(message, nonce) |> Map.new()
   end
+
+  def cached(key, fun, ttl \\ 0) do
+    Logger.debug("Fetching #{inspect(key)} from cache")
+
+    opts = if ttl > 0, do: [ttl: ttl], else: []
+
+    Cachex.fetch(Supavisor.Cache, key, fn _key ->
+      {:commit, {:cached, fun.()}, opts}
+    end)
+    |> case do
+      {_, {:cached, value}} ->
+        value
+
+      {_, {:cached, value}, _} ->
+        value
+    end
+  end
 end