Multitenancy support

Makefile

@@ -1,23 +1,23 @@
 .PHONY: dev
 dev:
-	MIX_ENV=dev ERL_AFLAGS="-kernel shell_history enabled" iex -S mix phx.server
+	MIX_ENV=dev DB_ENC_KEY="1234567890123456" API_JWT_SECRET=dev ERL_AFLAGS="-kernel shell_history enabled" iex -S mix phx.server
 
 db_start:
 	docker-compose -f ./dev/docker-compose.db.yml up
 
 db_stop:
-	docker-compose -f ./dev/docker-compose.db.yml down  --remove-orphans
+	docker-compose -f ./dev/docker-compose.db.yml down --remove-orphans
 
 db_rebuild:
 	make db_stop
 	docker-compose -f ./dev/docker-compose.db.yml build
-	docker-compose -f ./dev/docker-compose.db.yml up --force-recreate --build
+	make db_start
 
 pgbench_init:
 	PGPASSWORD=postgres pgbench -i -h 127.0.0.1 -p 6432 -U postgres -d postgres
 
 pgbench_short:
-	PGPASSWORD=postgres pgbench -M extended --transactions 5 --jobs 4 --client 1 -h localhost -p 7654 -U postgres postgres
+	PGPASSWORD=postgres pgbench -M extended --transactions 5 --jobs 4 --client 1 -h localhost -p 7654 -U postgres#localhost postgres
 
 pgbench_long:
-	PGPASSWORD=postgres pgbench -M extended --transactions 500 --jobs 5 --client 100 -h localhost -p 7654 -U postgres postgres
+	PGPASSWORD=postgres pgbench -M extended --transactions 500 --jobs 5 --client 100 -h localhost -p 7654 -U postgres#localhost postgres

config/config.exs

@@ -8,8 +8,9 @@
 import Config
 
 config :pg_edge,
-  proxy_port: 7654
-  # ecto_repos: [PgEdge.Repo]
+  proxy_port: 7654,
+  ecto_repos: [PgEdge.Repo],
+  version: Mix.Project.config()[:version]
 
 # Configures the endpoint
 config :pg_edge, PgEdgeWeb.Endpoint,

config/runtime.exs

@@ -1,32 +1,13 @@
 import Config
 
 if config_env() == :prod do
-  # database_url =
-  #   System.get_env("DATABASE_URL") ||
-  #     raise """
-  #     environment variable DATABASE_URL is missing.
-  #     For example: ecto://USER:PASS@HOST/DATABASE
-  #     """
-
-  maybe_ipv6 = if System.get_env("ECTO_IPV6"), do: [:inet6], else: []
-
-  # config :pg_edge, PgEdge.Repo,
-  #   # ssl: true,
-  #   url: database_url,
-  #   pool_size: String.to_integer(System.get_env("POOL_SIZE") || "10"),
-  #   socket_options: maybe_ipv6
-
-  # The secret key base is used to sign/encrypt cookies and other secrets.
-  # A default value is used in config/dev.exs and config/test.exs but you
-  # want to use a different value for prod and you most likely don't want
-  # to check this value into version control, so we use an environment
-  # variable instead.
   secret_key_base =
     System.get_env("SECRET_KEY_BASE") ||
       raise """
       environment variable SECRET_KEY_BASE is missing.
       You can generate one by calling: mix phx.gen.secret
       """
+
   app_name =
     System.get_env("FLY_APP_NAME") ||
       raise "APP_NAME not available"
@@ -47,11 +28,29 @@ if config_env() == :prod do
 end
 
 if config_env() != :test do
+  config :pg_edge,
+    jwt_claim_validators: System.get_env("JWT_CLAIM_VALIDATORS", "{}") |> Jason.decode!(),
+    api_jwt_secret: System.get_env("API_JWT_SECRET"),
+    db_enc_key: System.get_env("DB_ENC_KEY")
+
+  config :pg_edge, PgEdge.Repo,
+    # priv: "priv/repo/pgedge",
+    # migration_default_prefix: "pgedge",
+    hostname: System.get_env("DB_HOST", "localhost"),
+    username: System.get_env("DB_USER", "postgres"),
+    password: System.get_env("DB_PASSWORD", "postgres"),
+    database: System.get_env("DB_NAME", "postgres"),
+    port: System.get_env("DB_PORT", "6432"),
+    pool_size: System.get_env("DB_POOL_SIZE", "5") |> String.to_integer(),
+    parameters: [
+      application_name: "pg_edge_meta"
+    ]
+
   config :pg_edge, PgEdge.DevTenant,
     db_host: System.get_env("TENANT_DB_HOST", "127.0.0.1"),
     db_port: System.get_env("TENANT_DB_PORT", "6432") |> String.to_integer(),
     db_name: System.get_env("TENANT_DB_NAME", "postgres"),
-    db_user:  System.get_env("TENANT_DB_USER", "postgres"),
+    db_user: System.get_env("TENANT_DB_USER", "postgres"),
     db_password: System.get_env("TENANT_DB_PASSWORD", "postgres"),
     connect_timeout: 5000,
     application_name: "pg_edge",

dev/postgres/00-setup.sql → dev/dev/postgres/00-setup.sql

@@ -8,4 +8,4 @@ alter default privileges in schema public grant all on tables    to anon, authen
 alter default privileges in schema public grant all on functions to anon, authenticated, service_role;
 alter default privileges in schema public grant all on sequences to anon, authenticated, service_role;
 
-create schema if not exists pg_edge;
+create schema if not exists pgedge;

lib/pg_edge.ex

@@ -1,9 +1,37 @@
 defmodule PgEdge do
-  @moduledoc """
-  PgEdge keeps the contexts that define your domain
-  and business logic.
+  @moduledoc false
+  require Logger
+  alias PgEdge.{Tenants, Tenants.Tenant}
 
-  Contexts are also responsible for managing your data, regardless
-  if it comes from the database, an external API or others.
-  """
+  @spec start_pool(String.t()) :: {:ok, pid} | {:error, any}
+  def start_pool(external_id) do
+    Logger.debug("Starting pool for #{external_id}")
+
+    case Tenants.get_tenant_by_external_id(external_id) do
+      %Tenant{} = tenant ->
+        pool_spec = [
+          name: pool_name(external_id),
+          worker_module: PgEdge.DbHandler,
+          size: tenant.pool_size,
+          max_overflow: 0
+        ]
+
+        DynamicSupervisor.start_child(
+          {:via, PartitionSupervisor, {PgEdge.DynamicSupervisor, self()}},
+          :poolboy.child_spec(:worker, pool_spec, tenant)
+        )
+
+      _ ->
+        Logger.error("Can't find tenant with external_id #{external_id}")
+        {:error, :tenant_not_found}
+    end
+  end
+
+  # TODO: implement stop_pool
+  def stop_pool(_), do: :not_implemented
+
+  @spec pool_name(any) :: {:via, Registry, {PgEdge.Registry.DbPool, any}}
+  def pool_name(external_id) do
+    {:via, Registry, {PgEdge.Registry.DbPool, external_id}}
+  end
 end

lib/pg_edge/application.ex

@@ -16,16 +16,24 @@ defmodule PgEdge.Application do
       []
     )
 
+    Registry.start_link(
+      keys: :unique,
+      name: PgEdge.Registry.DbPool
+    )
+
     children = [
       # Start the Ecto repository
-      # PgEdge.Repo,
+      PgEdge.Repo,
       # Start the Telemetry supervisor
       PgEdgeWeb.Telemetry,
       # Start the PubSub system
       {Phoenix.PubSub, name: PgEdge.PubSub},
       # Start the Endpoint (http/https)
       PgEdgeWeb.Endpoint,
-      :poolboy.child_spec(:worker, dev_pool())
+      {
+        PartitionSupervisor,
+        child_spec: DynamicSupervisor, strategy: :one_for_one, name: PgEdge.DynamicSupervisor
+      }
     ]
 
     # See https://hexdocs.pm/elixir/Supervisor.html
@@ -41,13 +49,4 @@ defmodule PgEdge.Application do
     PgEdgeWeb.Endpoint.config_change(changed, removed)
     :ok
   end
-
-  defp dev_pool do
-    [
-      name: {:local, :db_sess},
-      worker_module: PgEdge.DbHandler,
-      size: Application.get_env(:pg_edge, :pool_size),
-      max_overflow: 0
-    ]
-  end
 end

lib/pg_edge/client_handler.ex

@@ -36,6 +36,7 @@ defmodule PgEdge.ClientHandler do
         connected: false,
         buffer: "",
         db_pid: nil,
+        tenant: nil,
         state: :wait_startup_packet
       }
     )
@@ -56,8 +57,26 @@ defmodule PgEdge.ClientHandler do
     else
       hello = Client.decode_startup_packet(bin)
       Logger.debug("Client startup message: #{inspect(hello)}")
-      trans.send(socket, authentication_ok())
-      {:noreply, %{state | state: :idle}}
+
+      # TODO: rewrite case -> with
+      case hello.payload do
+        [["user", user_value] | _] ->
+          case String.split(user_value, "#") do
+            [_user, external_id] ->
+              # TODO: check the response
+              PgEdge.start_pool(external_id)
+              trans.send(socket, authentication_ok())
+              {:noreply, %{state | state: :idle, tenant: external_id}}
+
+            _ ->
+              Logger.error("Can't find external_id in #{inspect(user_value)}")
+              {:stop, :normal, state}
+          end
+
+        _ ->
+          Logger.error("Can't find user in #{inspect(hello)}")
+          {:stop, :normal, state}
+      end
     end
   end
 
@@ -73,12 +92,13 @@ defmodule PgEdge.ClientHandler do
     {:noreply, state}
   end
 
-  def handle_info({:tcp, _, bin}, %{buffer: buf, db_pid: db_pid} = state) do
+  def handle_info({:tcp, _, bin}, %{buffer: buf, db_pid: db_pid, tenant: tenant} = state) do
     db_pid =
       if db_pid do
         db_pid
       else
-        :poolboy.checkout(:db_sess, true, 60000)
+        PgEdge.pool_name(tenant)
+        |> :poolboy.checkout(true, 60000)
       end
 
     data = buf <> bin
@@ -121,11 +141,13 @@ defmodule PgEdge.ClientHandler do
   def handle_call(
         {:client_call, bin, ready?},
         _,
-        %{socket: socket, trans: trans, db_pid: db_pid} = state
+        %{socket: socket, trans: trans, db_pid: db_pid, tenant: tenant} = state
       ) do
     db_pid1 =
       if ready? do
-        :poolboy.checkin(:db_sess, db_pid)
+        PgEdge.pool_name(tenant)
+        |> :poolboy.checkin(db_pid)
+
         nil
       else
         db_pid
@@ -144,7 +166,7 @@ defmodule PgEdge.ClientHandler do
       <<0, 0, 0, 0>>,
       # parameter_status,<<"application_name">>,<<"nonode@nohost">>
       <<83, 0, 0, 0, 35>>,
-      <<"application_name", 0, "node@nohost", 0>>,
+      <<"application_name", 0, "nonode@nohost", 0>>,
       # parameter_status,<<"client_encoding">>,<<"UTF8">>
       <<83, 0, 0, 0, 25>>,
       <<99, 108, 105, 101, 110, 116, 95, 101, 110, 99, 111, 100, 105, 110, 103, 0, 85, 84, 70, 56,

lib/pg_edge/db_handler.ex

@@ -4,7 +4,7 @@ defmodule PgEdge.DbHandler do
   alias PgEdge.Protocol.Server
   alias PgEdge.ClientHandler, as: Client
 
-  def start_link(config) when is_list(config) do
+  def start_link(config) do
     GenServer.start_link(__MODULE__, config)
   end
 
@@ -13,23 +13,27 @@ defmodule PgEdge.DbHandler do
   end
 
   @impl true
-  def init(_) do
+  def init(args) do
     # IP
     # {:ok, host} =
     #   Application.get_env(:pg_edge, :db_host)
     #   |> String.to_charlist()
     #   |> :inet.parse_address()
 
-    host =
-      Application.get_env(:pg_edge, PgEdge.DevTenant)[:db_host]
-      |> String.to_charlist()
+    %{
+      db_host: db_host,
+      db_port: db_port,
+      db_user: db_user,
+      db_database: db_database,
+      db_password: db_password
+    } = args
 
     auth = %{
-      host: host,
-      port: Application.get_env(:pg_edge, PgEdge.DevTenant)[:db_port],
-      user: Application.get_env(:pg_edge, PgEdge.DevTenant)[:db_user],
-      database: Application.get_env(:pg_edge, PgEdge.DevTenant)[:db_name],
-      password: Application.get_env(:pg_edge, PgEdge.DevTenant)[:db_password],
+      host: String.to_charlist(db_host),
+      port: db_port,
+      user: db_user,
+      database: db_database,
+      password: fn -> decrypt_password(db_password) end,
       application_name: "pg_edge"
     }
 
@@ -58,7 +62,7 @@ defmodule PgEdge.DbHandler do
     Logger.debug("<-- <-- bin #{inspect(byte_size(bin))} bytes, caller: #{inspect(socket)}")
 
     messages =
-      if socket do
+      if socket && state.state == :idle do
         :gen_tcp.send(socket, bin)
         ""
       else
@@ -147,7 +151,7 @@ defmodule PgEdge.DbHandler do
               server_first_parts,
               nonce,
               state.auth.user,
-              state.auth.password
+              state.auth.password.()
             )
 
           bin = :pgo_protocol.encode_scram_response_message(client_final_message)
@@ -263,4 +267,9 @@ defmodule PgEdge.DbHandler do
   def active_once(socket) do
     :inet.setopts(socket, [{:active, :once}])
   end
+
+  defp decrypt_password(password) do
+    Application.get_env(:pg_edge, :db_enc_key)
+    |> PgEdge.Helpers.decrypt!(password)
+  end
 end

lib/pg_edge/helpers.ex

@@ -0,0 +1,31 @@
+defmodule PgEdge.Helpers do
+  @moduledoc """
+  This module includes helper functions for different contexts that can't be union in one module.
+  """
+
+  @spec encrypt!(<<_::128>>, binary()) :: binary()
+  def encrypt!(secret_key, text) do
+    :aes_128_ecb
+    |> :crypto.crypto_one_time(secret_key, pad(text), true)
+    |> Base.encode64()
+  end
+
+  @spec decrypt!(<<_::128>>, binary()) :: binary()
+  def decrypt!(secret_key, base64_text) do
+    crypto_text = Base.decode64!(base64_text)
+
+    :aes_128_ecb
+    |> :crypto.crypto_one_time(secret_key, crypto_text, false)
+    |> unpad()
+  end
+
+  defp pad(data) do
+    to_add = 16 - rem(byte_size(data), 16)
+    data <> :binary.copy(<<to_add>>, to_add)
+  end
+
+  defp unpad(data) do
+    to_remove = :binary.last(data)
+    :binary.part(data, 0, byte_size(data) - to_remove)
+  end
+end