improvement: open one database per tenant under a supervisor the repo starts

lib/multi_tenancy.ex

@@ -0,0 +1,141 @@
+# SPDX-FileCopyrightText: 2023 ash_sqlite contributors <https://github.com/ash-project/ash_sqlite/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshSqlite.MultiTenancy do
+  @moduledoc """
+  A `c:AshSqlite.Repo.tenant_repo/2` implementation that gives each tenant its own database file.
+
+  A tenant's file is opened and migrated on first use, and the repo's `init/2` sees the tenant as `config[:tenant]`.
+  Only `dir` is required. `max_children` defaults to `:infinity` and `open_timeout`, in milliseconds, to 15000.
+
+  ```elixir
+  config :my_app, MyApp.Repo,
+    tenants: [
+      dir: "priv/tenants",
+      migrations_path: "priv/repo/tenant_migrations",
+      max_children: 1_000,
+      open_timeout: 15_000
+    ]
+
+  defmodule MyApp.Repo do
+    use AshSqlite.Repo, otp_app: :my_app
+
+    def tenant_repo(tenant, context),
+      do: AshSqlite.MultiTenancy.tenant_repo(__MODULE__, tenant, context)
+  end
+  ```
+  """
+
+  alias AshSqlite.MultiTenancy.{Supervisor, Tenant}
+
+  @unreserved ~c"abcdefghijklmnopqrstuvwxyz0123456789-_."
+
+  # 255 is the usual per-name limit. `-journal` is the longest suffix SQLite adds to
+  # a database's name, and `.db` is ours.
+  @max_encoded_bytes 255 - byte_size(".db") - byte_size("-journal")
+
+  @attempts 3
+
+  @doc "The repo instance serving `tenant`, opening its database if it is not open."
+  @spec tenant_repo(module(), term(), map()) :: pid()
+  def tenant_repo(repo, tenant, _context) do
+    registry = Supervisor.registry(repo)
+
+    if is_nil(Process.whereis(registry)) do
+      raise ArgumentError, """
+      #{inspect(repo)} has no tenant supervisor running, so tenant #{inspect(tenant)} \
+      has no database to open. Configure the `tenants:` key and start the repo:
+
+          config :my_app, #{inspect(repo)}, tenants: [dir: "priv/tenants"]
+      """
+    end
+
+    {:ok, dir} = Registry.meta(registry, :dir)
+    {:ok, open_timeout} = Registry.meta(registry, :open_timeout)
+    path = Path.join(dir, encode(tenant) <> ".db")
+
+    instance(repo, tenant, path, open_timeout, @attempts)
+  end
+
+  defp instance(_repo, tenant, _path, _open_timeout, 0) do
+    raise "tenant #{inspect(tenant)} kept stopping while being opened"
+  end
+
+  defp instance(repo, tenant, path, open_timeout, attempts) do
+    pid = start_tenant(repo, tenant, path)
+
+    case Tenant.instance(pid, open_timeout) do
+      {:ok, instance} ->
+        instance
+
+      # `:stopped` arrives only after that tenant has exited, and registering a `:via`
+      # name replaces a dead pid's entry, so the next start registers a new tenant.
+      :stopped ->
+        instance(repo, tenant, path, open_timeout, attempts - 1)
+
+      {:timeout, pid} ->
+        DynamicSupervisor.terminate_child(Supervisor.tenants(repo), pid)
+
+        raise "the database of tenant #{inspect(tenant)} did not open within #{open_timeout}ms. \
+        Raise `tenants: [open_timeout: ...]` if opening it legitimately takes longer."
+
+      {:error, reason} ->
+        raise "could not open the database of tenant #{inspect(tenant)}: #{Exception.format_exit(reason)}"
+    end
+  end
+
+  defp start_tenant(repo, tenant, path) do
+    case DynamicSupervisor.start_child(Supervisor.tenants(repo), {Tenant, {repo, tenant, path}}) do
+      {:ok, pid} ->
+        pid
+
+      # May be a tenant that is stopping; `instance/5` waits for it and retries.
+      {:error, {:already_started, pid}} ->
+        pid
+
+      {:error, :max_children} ->
+        raise "could not open tenant #{inspect(tenant)}: #{inspect(repo)} already has the \
+        most tenants open that `tenants: [max_children: ...]` allows."
+
+      {:error, reason} ->
+        raise "could not start tenant #{inspect(tenant)}: #{inspect(reason)}"
+    end
+  end
+
+  # Escapes rather than sanitises, so no two tenants share a file and no tenant names a
+  # path outside `dir`.
+  defp encode(tenant) when is_binary(tenant) and byte_size(tenant) > 0 do
+    encoded = tenant |> :binary.bin_to_list() |> Enum.map_join(&encode_byte/1)
+
+    if byte_size(encoded) > @max_encoded_bytes do
+      raise ArgumentError, """
+      tenant #{inspect(tenant)} encodes to #{byte_size(encoded)} bytes, and a \
+      database name can be at most #{@max_encoded_bytes}. Use a shorter identifier, \
+      such as a slug or a UUID, as the tenant.
+      """
+    end
+
+    encoded
+  end
+
+  defp encode(tenant) do
+    raise ArgumentError, """
+    expected a non-empty binary tenant, got: #{inspect(tenant)}
+
+    A tenant names a database file, and stringifying another term here would let two \
+    tenants that Ash tells apart name one file. Convert it where you set the tenant.
+    """
+  end
+
+  defp encode_byte(byte) when byte in @unreserved, do: <<byte>>
+
+  defp encode_byte(byte) do
+    <<hi::4, lo::4>> = <<byte>>
+    <<?~, hex(hi), hex(lo)>>
+  end
+
+  # Lowercase, so two encodings never differ only by case.
+  defp hex(nibble) when nibble < 10, do: ?0 + nibble
+  defp hex(nibble), do: ?a + nibble - 10
+end

lib/multi_tenancy/supervisor.ex

@@ -0,0 +1,101 @@
+# SPDX-FileCopyrightText: 2023 ash_sqlite contributors <https://github.com/ash-project/ash_sqlite/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshSqlite.MultiTenancy.Supervisor do
+  @moduledoc false
+  use Supervisor
+
+  def child_spec(repo, config, ecto_child_spec) do
+    case Keyword.fetch(config, :tenants) do
+      :error ->
+        ecto_child_spec
+
+      {:ok, tenants} ->
+        if !AshSqlite.DataLayer.tenant_repo?(repo) do
+          raise ArgumentError, """
+          #{inspect(repo)} is configured with `tenants:`, but does not implement \
+          `c:AshSqlite.Repo.tenant_repo/2`. Add this to the repo:
+
+              def tenant_repo(tenant, context), do: AshSqlite.MultiTenancy.tenant_repo(__MODULE__, tenant, context)
+          """
+        end
+
+        base_repo = if config[:database], do: [ecto_child_spec], else: []
+
+        %{
+          id: repo,
+          start: {__MODULE__, :start_link, [{repo, tenants, base_repo}]},
+          type: :supervisor
+        }
+    end
+  end
+
+  def start_link({repo, _tenants, _base_repo} = arg) do
+    Supervisor.start_link(__MODULE__, arg, name: Module.concat(repo, TenantSupervisor))
+  end
+
+  def registry(repo), do: Module.concat(repo, TenantRegistry)
+  def tenants(repo), do: Module.concat(repo, Tenants)
+
+  @impl true
+  def init({repo, tenants, base_repo}) do
+    migrations = compile_migrations(tenants[:migrations_path])
+
+    children =
+      [
+        {Registry,
+         keys: :unique,
+         name: registry(repo),
+         meta: [
+           migrations: migrations,
+           dir: Keyword.fetch!(tenants, :dir),
+           open_timeout: Keyword.get(tenants, :open_timeout, 15_000)
+         ]},
+        {DynamicSupervisor,
+         name: tenants(repo), max_children: Keyword.get(tenants, :max_children, :infinity)}
+      ] ++ base_repo
+
+    # A restarted Registry has forgotten every tenant, so the tenants started after it
+    # must go too, or a second process could open the same file.
+    Supervisor.init(children, strategy: :rest_for_one)
+  end
+
+  defp compile_migrations(nil), do: []
+
+  # Mirrors `Ecto.Migrator`'s naming rules: `<version>_<name>.exs`.
+  defp compile_migrations(path) do
+    [path, "**", "*.exs"]
+    |> Path.join()
+    |> Path.wildcard()
+    |> Enum.flat_map(fn file ->
+      case Integer.parse(Path.basename(file)) do
+        {version, "_" <> _} -> [{version, migration_module!(file)}]
+        _ -> []
+      end
+    end)
+    |> Enum.sort()
+  end
+
+  # A restart, or a second repo sharing the path, reuses the modules already loaded
+  # rather than redefining code a running migration may be executing.
+  defp migration_module!(file) do
+    modules =
+      case Code.require_file(file) do
+        nil -> modules_compiled_from(file)
+        compiled -> Enum.map(compiled, &elem(&1, 0))
+      end
+
+    Enum.find(modules, &function_exported?(&1, :__migration__, 0)) ||
+      raise Ecto.MigrationError,
+            "file #{Path.relative_to_cwd(file)} does not define an Ecto.Migration"
+  end
+
+  defp modules_compiled_from(file) do
+    source = file |> Path.expand() |> String.to_charlist()
+
+    for {module, _} <- :code.all_loaded(),
+        module.module_info(:compile)[:source] == source,
+        do: module
+  end
+end

lib/multi_tenancy/tenant.ex

@@ -0,0 +1,84 @@
+# SPDX-FileCopyrightText: 2023 ash_sqlite contributors <https://github.com/ash-project/ash_sqlite/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshSqlite.MultiTenancy.Tenant do
+  @moduledoc """
+  The process that owns one tenant's database: it opens and migrates the file, and its repo instance lives and dies with it.
+  """
+  use GenServer, restart: :temporary, shutdown: 30_000
+
+  def start_link({repo, tenant, path}) do
+    GenServer.start_link(__MODULE__, {repo, tenant, path},
+      name: {:via, Registry, {AshSqlite.MultiTenancy.Supervisor.registry(repo), tenant}}
+    )
+  end
+
+  @doc """
+  The repo instance of the tenant process `pid`, waiting up to `timeout` for its database to open.
+
+  `:stopped` means the process has exited.
+  """
+  @spec instance(pid(), timeout()) ::
+          {:ok, pid()} | :stopped | {:timeout, pid()} | {:error, term()}
+  def instance(pid, timeout) do
+    {:ok, GenServer.call(pid, :instance, timeout)}
+  catch
+    :exit, {reason, {GenServer, :call, _}}
+    when reason in [:noproc, :normal, :shutdown, :killed] ->
+      :stopped
+
+    :exit, {{:shutdown, _}, {GenServer, :call, _}} ->
+      :stopped
+
+    :exit, {:timeout, {GenServer, :call, _}} ->
+      {:timeout, pid}
+
+    :exit, reason ->
+      {:error, reason}
+  end
+
+  @impl true
+  # `DynamicSupervisor.start_child/2` runs `init/1` inside the supervisor, so opening
+  # here would queue every tenant behind every other tenant's open.
+  def init({repo, tenant, path}) do
+    # Trapped so `terminate/2` can stop the instance before this process leaves the
+    # Registry; otherwise a successor could open the file while the old pool still holds it.
+    Process.flag(:trap_exit, true)
+    {:ok, %{repo: repo, tenant: tenant, path: path, instance: nil}, {:continue, :open}}
+  end
+
+  @impl true
+  def handle_continue(:open, state) do
+    registry = AshSqlite.MultiTenancy.Supervisor.registry(state.repo)
+    {:ok, migrations} = Registry.meta(registry, :migrations)
+    File.mkdir_p!(Path.dirname(state.path))
+
+    {:ok, instance} = state.repo.start_link(name: nil, database: state.path, tenant: state.tenant)
+    state = %{state | instance: instance}
+
+    Ecto.Migrator.run(state.repo, migrations, :up, all: true, dynamic_repo: instance, log: false)
+    Ecto.Adapters.SQL.query!(instance, "SELECT 1", [])
+
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_call(:instance, _from, state), do: {:reply, state.instance, state}
+
+  @impl true
+  def handle_info({:EXIT, instance, reason}, %{instance: instance} = state) do
+    {:stop, reason, %{state | instance: nil}}
+  end
+
+  def handle_info({:EXIT, _pid, _reason}, state), do: {:noreply, state}
+
+  @impl true
+  def terminate(_reason, %{instance: instance}) when is_pid(instance) do
+    Supervisor.stop(instance, :shutdown)
+  catch
+    :exit, _ -> :ok
+  end
+
+  def terminate(_reason, _state), do: :ok
+end

lib/repo.ex

@@ -20,6 +20,10 @@ defmodule AshSqlite.Repo do
 
   - `:tenant_migrations_path` - The path where your tenant migrations are stored (only relevant for a multitenant implementation)
   - `:snapshots_path` - The path where the resource snapshots for the migration generator are stored.
+
+  In the repo's application config, as it is read before `init/2` runs:
+
+  - `:tenants` - Opens one database file per tenant. See `AshSqlite.MultiTenancy` for its options.
   """
 
   @doc "Use this to inform the data layer about what extensions are installed"
@@ -41,6 +45,9 @@ defmodule AshSqlite.Repo do
   instance. Every resource on a tenant repo must use `strategy :context`, so put
   tables shared across tenants on a repo module of their own.
 
+  `AshSqlite.MultiTenancy.tenant_repo/3` is a default implementation that opens
+  one database file per tenant.
+
   `context.type` is `:read` for queries and aggregates, and `:mutate` for writes and
   for opening a transaction.
   """
@@ -65,11 +72,22 @@ defmodule AshSqlite.Repo do
         adapter: adapter,
         otp_app: otp_app
 
+      defoverridable child_spec: 1
+
       @behaviour AshSqlite.Repo
+      @ash_sqlite_otp_app otp_app
       @before_compile AshSqlite.Repo.BeforeCompile
 
       defoverridable insert: 2, insert: 1, insert!: 2, insert!: 1
 
+      def child_spec(opts) do
+        AshSqlite.MultiTenancy.Supervisor.child_spec(
+          __MODULE__,
+          Keyword.merge(Application.get_env(@ash_sqlite_otp_app, __MODULE__, []), opts),
+          super(Keyword.delete(opts, :tenants))
+        )
+      end
+
       def installed_extensions, do: []
       def migrations_path, do: nil
       def write_transactions?, do: false
@@ -78,6 +96,7 @@ defmodule AshSqlite.Repo do
       def init(_, config) do
         new_config =
           config
+          |> Keyword.drop([:tenant, :tenants])
           |> Keyword.put(:installed_extensions, installed_extensions())
           |> Keyword.put(:migrations_path, migrations_path())
           |> Keyword.put(:case_sensitive_like, :on)

test/multi_tenancy/migrations/20260101000000_create_posts.exs

@@ -0,0 +1,14 @@
+# SPDX-FileCopyrightText: 2023 ash_sqlite contributors <https://github.com/ash-project/ash_sqlite/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshSqlite.Test.MultiTenancyMigrations.CreatePosts do
+  use Ecto.Migration
+
+  def change do
+    create table(:posts, primary_key: false) do
+      add(:id, :binary_id, primary_key: true)
+      add(:title, :text)
+    end
+  end
+end