feat: use tenants from external source

integration-tests/tests/multi-tenancy.lux

@@ -118,74 +118,6 @@
   ?[\d+]
   ??}},{"headers":{"control":"up-to-date"}}]$PS1
 
-# Disable fail pattern for Electric as we are going to kill it
-[shell electric]
-  -
-
-## kill Electric
-[shell orchestrator]
-  !kill $(lsof -ti:3000)
-  ??$PS1
-
-## restart Electric
-[shell electric]
-  ??$PS1
-  # Re-enable fail pattern for Electric
-  -$fail_pattern
-  [invoke setup_multi_tenant_electric]
-  ???Reloading tenant tenant1 from storage
-  ???Reloading tenant tenant2 from storage
-  ???[info] Running Electric.Plug.Router with Bandit 1.5.5 at 0.0.0.0:3000 (http)
-
-## Make a query to check that they still see their data
-[shell tenant1]
-  # Query the shape
-  !curl -i -X GET "http://localhost:3000/v1/shape?table=items&offset=${offset}&handle=${shape_id}&database_id=tenant1"
-  ???[{"headers":{"control":"up-to-date"}}]
-  ??$PS1
-
-[shell tenant2]
-  # Query the shape
-  !curl -i -X GET "http://localhost:3000/v1/shape?table=items&offset=${offset}&handle=${shape_id}&database_id=tenant2"
-  ???[{"headers":{"control":"up-to-date"}}]
-  ??$PS1
-
-## Make a live query on both and check that it still works
-[shell tenant1]
-  !curl -i -X GET "localhost:3000/v1/shape?table=items&offset=$offset&handle="$shape_id"&database_id=tenant1&live"
-
-[shell tenant2]
-  !curl -i -X GET "localhost:3000/v1/shape?table=items&offset=$offset&handle="$shape_id"&database_id=tenant2&live"
-
-## Insert some data in tenant 1
-[shell tenant1_psql]
-  !INSERT INTO items (id, val) VALUES (3, 'tenant 1');
-  ??INSERT 0 1
-
-## Insert some data in tenant 2
-[shell tenant2_psql]
-  !INSERT INTO items (id, val) VALUES (3, 'tenant 2');
-  ??INSERT 0 1
-
-## Check that tenant 1 sees the new data
-[shell tenant1]
-  # give some time for the data to sync
-  [sleep 1]
-  ?\e\[1melectric-offset\e\[0m: ([\d_]+)
-  [local offset=$1]
-  ??[{"offset":"$offset","value":{"id":"3","val":"tenant 1"},"key":"\"public\".\"items\"/\"3\"","headers":{"relation":["public","items"],"operation":"insert","txid":
-  ?[\d+]
-  ??}},{"headers":{"control":"up-to-date"}}]$PS1
-
-## Check that tenant 2 sees the new data
-[shell tenant2]
-  [sleep 1]
-  ?\e\[1melectric-offset\e\[0m: ([\d_]+)
-  [local offset=$1]
-  ??[{"offset":"$offset","value":{"id":"3","val":"tenant 2"},"key":"\"public\".\"items\"/\"3\"","headers":{"relation":["public","items"],"operation":"insert","txid":
-  ?[\d+]
-  ??}},{"headers":{"control":"up-to-date"}}]$PS1
-
 [shell electric]
   # disable fail pattern because deleting a tenant will stop the tenant processes
   # which will output some error messages because of the shutdown
@@ -203,20 +135,6 @@
   # Verify that tenant 1 still exists
   [invoke check_tenant_status "tenant1" "active" 3000]
 
-## kill Electric
-[shell orchestrator]
-  !kill $(lsof -ti:3000)
-  ??$PS1
-
-## restart Electric and check that only tenant 1 is reloaded and not tenant 2
-[shell electric]
-  ??$PS1
-  # Set fail pattern to fail if tenant 2 is reloaded
-  -Reloading tenant tenant2 from storage
-  !PORT=3000 ../scripts/electric_dev.sh
-  ???Reloading tenant tenant1 from storage
-  ???[info] Running Electric.Plug.Router with Bandit 1.5.5 at 0.0.0.0:3000 (http)
-
 [cleanup]
   [invoke teardown]
   # Also tear down the first tenant

packages/sync-service/config/runtime.exs

@@ -106,8 +106,7 @@ case {database_url, default_tenant} do
 
     config :electric, default_connection_opts: Electric.Utils.obfuscate_password(connection_opts)
 
-    # if `default_tenant` is nil, generate a random UUID for it
-    tenant_id = default_tenant || Electric.Utils.uuid4()
+    tenant_id = default_tenant || "00000000-0000-0000-0000-000000000000"
     config :electric, default_tenant: tenant_id
 end
 
@@ -209,4 +208,6 @@ config :electric,
   prometheus_port: prometheus_port,
   storage: storage,
   persistent_kv: persistent_kv,
-  listen_on_ipv6?: env!("ELECTRIC_LISTEN_ON_IPV6", :boolean, false)
+  control_plane: env!("ELECTRIC_CONTROL_PLANE", &Electric.ControlPlane.parse_config/1, nil),
+  listen_on_ipv6?: env!("ELECTRIC_LISTEN_ON_IPV6", :boolean, false),
+  tenant_tables_name: :tenant_tables

packages/sync-service/lib/electric/application.ex

@@ -105,7 +105,8 @@ defmodule Electric.Application do
       },
       pool_opts: %{
         size: Application.fetch_env!(:electric, :db_pool_size)
-      }
+      },
+      control_plane: Application.get_env(:electric, :control_plane, nil)
     }
 
     Electric.Application.Configuration.save(config)

packages/sync-service/lib/electric/application/configuration.ex

@@ -9,6 +9,7 @@ defmodule Electric.Application.Configuration do
     persistent_kv
     replication_opts
     pool_opts
+    control_plane
   ]a
 
   @type t :: %__MODULE__{}

packages/sync-service/lib/electric/control_plane.ex

@@ -0,0 +1,136 @@
+defmodule Electric.ControlPlane do
+  @moduledoc """
+  Functions that interact with the control plane that exists outside Electric.
+  """
+  require Logger
+
+  defstruct [
+    :base_url,
+    :auth,
+    paths: %{
+      "tenant_shape" => %{
+        "url" => "/v1/shape",
+        "params" => %{
+          "offset" => "-1",
+          "table" => "databases",
+          "where" => "electric_url ILIKE '%%{instance_id}'",
+          "select" => "id,connection_url"
+        }
+      }
+    }
+  ]
+
+  @type t() :: %__MODULE__{
+          base_url: String.t(),
+          auth: nil | String.t() | {:basic, String.t()} | {:bearer, String.t()},
+          paths: %{optional(String.t()) => map()}
+        }
+
+  def parse_config(""), do: nil
+
+  def parse_config(config_string) do
+    result = Jason.decode!(config_string)
+
+    %__MODULE__{
+      base_url: Map.fetch!(result, "base_url"),
+      auth: Map.get(result, "auth", nil),
+      paths: Map.get(result, "paths", %__MODULE__{}.paths)
+    }
+  end
+
+  @spec list_tenants(t(), keyword()) ::
+          {:ok, included :: list(map()), deleted :: list(map())} | {:error, :unreachable}
+  def list_tenants(%__MODULE__{} = plane, opts) do
+    %{electric_instance_id: instance_id} =
+      Keyword.get_lazy(opts, :app_config, fn -> Electric.Application.Configuration.get() end)
+
+    plane
+    |> build_req("tenant_shape", instance_id)
+    |> read_electric_api_until_done()
+    |> case do
+      {:ok, result} when is_list(result) ->
+        # We expect the control plane to fulfill the Electric API, so we can decode it here from the complete response
+        {ins_acc, del_acc} =
+          result
+          |> Stream.reject(&get_in(&1, ["headers", "control"]))
+          |> Stream.map(
+            &{get_in(&1, ["headers", "operation"]), Map.fetch!(&1, "key"),
+             Map.fetch!(&1, "value")}
+          )
+          |> collect_ops()
+
+        {:ok, Map.values(ins_acc), Map.values(del_acc)}
+
+      {:ok, %Req.Response{status: status, body: body}} ->
+        Logger.error(
+          "Could not reach the control plane while trying to list tenants. Latest response has status #{status} and body #{inspect(body)}"
+        )
+
+        {:error, :unreachable}
+
+      {:error, error} ->
+        Logger.error(
+          "Could not reach the control plane while trying to list tenants. Latest response was #{inspect(error)}"
+        )
+
+        {:error, :unreachable}
+    end
+  end
+
+  # We need to read the Electric stream until complete
+  defp read_electric_api_until_done(req, agg \\ []) do
+    with {:ok, %Req.Response{status: 200} = resp} <-
+           Req.get(req, max_retries: 4, retry_delay: 1_000) do
+      if Req.Response.get_header(resp, "electric-up-to-date") != [] do
+        {:ok, agg ++ resp.body}
+      else
+        [electric_handle] = Req.Response.get_header(resp, "electric-handle")
+        [electric_offset] = Req.Response.get_header(resp, "electric-offset")
+
+        req
+        |> Req.merge(params: [handle: electric_handle, offset: electric_offset])
+        |> read_electric_api_until_done(agg ++ resp.body)
+      end
+    end
+  end
+
+  defp build_req(%__MODULE__{} = plane, path_name, instance_id) do
+    %{"url" => url} = path_spec = Map.fetch!(plane.paths, path_name)
+    url = insert_instance_id(url, instance_id)
+
+    params =
+      path_spec
+      |> Map.get("params", [])
+      |> Enum.map(fn {k, v} -> {k, insert_instance_id(v, instance_id)} end)
+
+    headers =
+      path_spec
+      |> Map.get("headers", [])
+      |> Enum.map(fn {k, v} -> {k, insert_instance_id(v, instance_id)} end)
+
+    Req.new(
+      base_url: plane.base_url,
+      url: url,
+      params: params,
+      auth: plane.auth,
+      headers: headers
+    )
+  end
+
+  defp insert_instance_id(string, instance_id),
+    do: String.replace(string, "%{instance_id}", to_string(instance_id))
+
+  @spec collect_ops(Enumerable.t()) :: {map(), map()}
+  defp collect_ops(ops) do
+    Enum.reduce(ops, {%{}, %{}}, fn
+      {"insert", key, value}, {ins_acc, del_acc} ->
+        {Map.put(ins_acc, key, value), del_acc}
+
+      {"update", key, value}, {ins_acc, del_acc} ->
+        {Map.update!(ins_acc, key, &Map.merge(&1, value)), del_acc}
+
+      {"delete", key, value}, {ins_acc, del_acc} ->
+        {Map.delete(ins_acc, key), Map.put(del_acc, key, value)}
+    end)
+  end
+end

packages/sync-service/lib/electric/plug/remove_database_plug.ex

@@ -16,7 +16,7 @@ defmodule Electric.Plug.RemoveDatabasePlug do
         |> send_resp(200, Jason.encode_to_iodata!(tenant_id))
         |> halt()
 
-      :not_found ->
+      {:error, :not_found} ->
         conn
         |> send_resp(404, Jason.encode_to_iodata!("Database #{tenant_id} not found."))
         |> halt()