improvement: manage tenant databases, so strategy :context needs no tenant_repo of its own

lib/data_layer.ex

@@ -2290,8 +2290,8 @@ defmodule AshSqlite.DataLayer do
   defp with_tenant_repo(resource, tenant, type, fun) do
     case AshSqlite.DataLayer.Info.tenant_repo(resource) do
       # No tenant repo and a tenant means multitenancy this data layer does not
-      # resolve to a connection, such as `strategy :attribute`. `VerifyTenantRepo`
-      # covers `strategy :context` at compile time.
+      # resolve to a connection, such as `strategy :attribute`. `strategy :context`
+      # always has one, by default `AshSqlite.MultiTenancy.TenantRepo`.
       nil ->
         fun.()
 

lib/data_layer/info.ex

@@ -69,12 +69,22 @@ defmodule AshSqlite.DataLayer.Info do
   @doc """
   The tenant repo for a resource as `{module, opts}`, or nil.
 
-  Only set for resources with context multitenancy. A function given to the
+  Only set for resources with context multitenancy, where it defaults to
+  `AshSqlite.MultiTenancy.TenantRepo` so that database-per-tenant works without
+  the application supplying a runtime of its own. A function given to the
   `tenant_repo` option is returned wrapped in a module implementing
   `AshSqlite.TenantRepo`.
   """
   def tenant_repo(resource) do
-    Extension.get_opt(resource, [:sqlite], :tenant_repo, nil, true)
+    case Extension.get_opt(resource, [:sqlite], :tenant_repo, nil, true) do
+      nil ->
+        if Ash.Resource.Info.multitenancy_strategy(resource) == :context do
+          {AshSqlite.MultiTenancy.TenantRepo, []}
+        end
+
+      tenant_repo ->
+        tenant_repo
+    end
   end
 
   @doc "The configured table for a resource"

lib/multi_tenancy.ex

@@ -0,0 +1,196 @@
+# SPDX-FileCopyrightText: 2023 ash_sqlite contributors <https://github.com/ash-project/ash_sqlite/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshSqlite.MultiTenancy do
+  @moduledoc """
+  One SQLite database per tenant: choosing the file, opening it, migrating it, and closing it again.
+  """
+  use Supervisor
+
+  alias AshSqlite.MultiTenancy.Binds
+  alias AshSqlite.MultiTenancy.Connection
+  alias AshSqlite.MultiTenancy.Manager
+  alias AshSqlite.MultiTenancy.Registry, as: TenantRegistry
+
+  @doc false
+  def child_spec(opts) do
+    %{
+      id: {__MODULE__, Keyword.fetch!(opts, :repo)},
+      start: {__MODULE__, :start_link, [opts]},
+      type: :supervisor
+    }
+  end
+
+  def start_link(opts) do
+    repo = Keyword.fetch!(opts, :repo)
+    Supervisor.start_link(__MODULE__, opts, name: Module.concat(repo, MultiTenancy))
+  end
+
+  @impl true
+  def init(opts) do
+    repo = Keyword.fetch!(opts, :repo)
+    _dir = Keyword.fetch!(opts, :dir)
+    verify_migrations_path!(Keyword.get(opts, :migrations_path))
+
+    children = [
+      TenantRegistry.child_spec(repo),
+      Binds.child_spec(repo),
+      AshSqlite.MultiTenancy.ConnectionSupervisor.child_spec(repo),
+      Manager.child_spec(opts)
+    ]
+
+    Supervisor.init(children, strategy: :one_for_one)
+  end
+
+  @doc "The repo instance serving `tenant`, starting it if it is not resident."
+  @spec connection_for(module(), String.t()) :: {:ok, pid()} | {:error, term()}
+  def connection_for(repo, tenant) do
+    do_connection_for(repo, tenant)
+  rescue
+    # `Registry.lookup/2` raises on a registry never started, naming an internal
+    # registry rather than the omission.
+    exception in ArgumentError ->
+      if started?(repo), do: reraise(exception, __STACKTRACE__), else: not_started!(repo)
+  catch
+    :exit, {:noproc, _} = reason ->
+      if started?(repo), do: exit(reason), else: not_started!(repo)
+  end
+
+  defp do_connection_for(repo, tenant) do
+    case TenantRegistry.lookup(repo, tenant) do
+      {:ok, _connection, repo_pid} when is_pid(repo_pid) ->
+        {:ok, repo_pid}
+
+      # Registered, but its `init/1` has not published a repo yet, so this caller
+      # raced the activation. Asking the connection directly waits for it.
+      {:ok, connection, nil} ->
+        case Connection.repo_pid(connection) do
+          repo_pid when is_pid(repo_pid) -> {:ok, repo_pid}
+          _ -> Manager.activate(repo, tenant)
+        end
+
+      :error ->
+        Manager.activate(repo, tenant)
+    end
+  end
+
+  defp started?(repo), do: is_pid(Process.whereis(Module.concat(repo, MultiTenancy)))
+
+  # Raised rather than left as `unknown registry: ...`, which names an
+  # implementation detail instead of the omission.
+  defp not_started!(repo) do
+    raise """
+    #{inspect(repo)} has no tenant fleet running, so there is no database to select \
+    for this tenant.
+
+    A resource with `strategy :context` gets `AshSqlite.MultiTenancy.TenantRepo` as \
+    its tenant repo, which asks `AshSqlite.MultiTenancy` for a connection. Add it to \
+    your supervision tree, after the repo:
+
+        children = [
+          #{inspect(repo)},
+          {AshSqlite.MultiTenancy,
+           repo: #{inspect(repo)},
+           dir: "priv/tenants",
+           migrations_path: "priv/repo/tenant_migrations"}
+        ]
+
+    If this repo's tenants are managed elsewhere -- Turso, Litestream, a router of \
+    your own -- name it as the resource's `tenant_repo` instead, and this \
+    one is not needed.
+    """
+  end
+
+  @doc "Runs `fun` with `tenant`'s database bound to the calling process."
+  @spec with_tenant(module(), String.t(), (-> result)) :: result when result: var
+  def with_tenant(repo, tenant, fun) when is_function(fun, 0) do
+    case connection_for(repo, tenant) do
+      {:ok, repo_pid} ->
+        bound(repo, tenant, repo_pid, fun)
+
+      # An unbound statement here would run against another tenant's data.
+      {:error, reason} ->
+        raise AshSqlite.MultiTenancy.UnavailableError, tenant: tenant, reason: reason
+    end
+  end
+
+  defp bound(repo, tenant, repo_pid, fun) do
+    case Binds.bound(repo, tenant) do
+      :ok ->
+        previous = repo.put_dynamic_repo(repo_pid)
+
+        try do
+          fun.()
+        after
+          repo.put_dynamic_repo(previous)
+          Binds.released(repo, tenant)
+        end
+
+      # Being closed, so wait for it to land and bind whatever replaces it.
+      :closing ->
+        Process.sleep(1)
+        with_tenant(repo, tenant, fun)
+    end
+  end
+
+  @doc """
+  Closes a tenant's database, leaving the file on disk.
+
+  Waits `:grace_ms` (1000 by default) for statements in flight, and reports
+  `{:error, :busy}` rather than closing under one. `force: true` closes regardless.
+  """
+  defdelegate close(repo, tenant, opts \\ []), to: Manager
+
+  @doc "Closes a tenant's database and deletes it, WAL sidecars included."
+  defdelegate delete(repo, tenant), to: Manager
+
+  @doc "Moves a tenant's database to another tenant's name, sidecars included."
+  defdelegate rename(repo, from, to), to: Manager
+
+  @doc "Tenants holding an open connection right now."
+  @spec resident(module()) :: [String.t()]
+  defdelegate resident(repo), to: TenantRegistry
+
+  @doc "Every tenant with a database on disk, plus any that are resident."
+  defdelegate all_tenants(repo), to: Manager
+
+  @doc "Tenants that failed to activate, with the reason."
+  defdelegate quarantined(repo), to: Manager
+
+  @doc "Clears a tenant's quarantine, so the next request tries again."
+  defdelegate release(repo, tenant), to: Manager
+
+  @doc "Stops this node taking on new tenants."
+  defdelegate seal(repo), to: Manager
+
+  @doc "Lets this node accept tenants again."
+  defdelegate unseal(repo), to: Manager
+
+  @doc "Activates and migrates each tenant in turn, reporting what happened."
+  defdelegate migrate_all(repo, tenants \\ :all, opts \\ []), to: Manager
+
+  @doc "Where a tenant's database is, whether or not it exists yet."
+  defdelegate path_for(repo, tenant), to: Manager
+
+  @doc "The fleet configuration."
+  defdelegate config(repo), to: Manager
+
+  # Checked at boot, not activation: a typo would otherwise surface as `no such
+  # table` whenever the first tenant woke up.
+  defp verify_migrations_path!(nil), do: :ok
+
+  defp verify_migrations_path!(path) do
+    if File.dir?(path) do
+      :ok
+    else
+      raise ArgumentError, """
+      :migrations_path #{inspect(path)} is not a directory.
+
+      Tenant databases are migrated from it, and `Ecto.Migrator` treats a missing \
+      directory as one containing no migrations -- so every tenant would open an \
+      empty database and fail on its first query instead of here.
+      """
+    end
+  end
+end

lib/multi_tenancy/binds.ex

@@ -0,0 +1,140 @@
+# SPDX-FileCopyrightText: 2023 ash_sqlite contributors <https://github.com/ash-project/ash_sqlite/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshSqlite.MultiTenancy.Binds do
+  @moduledoc """
+  Counts who is using each tenant, so that closing one can be safe.
+  """
+  use GenServer
+
+  @doc false
+  def child_spec(repo) do
+    %{id: {__MODULE__, repo}, start: {__MODULE__, :start_link, [repo]}}
+  end
+
+  @doc false
+  def start_link(repo), do: GenServer.start_link(__MODULE__, repo, name: name(repo))
+
+  @doc "The table serving `repo`."
+  @spec name(module()) :: module()
+  def name(repo), do: Module.concat(repo, TenantBinds)
+
+  @doc """
+  Records that the calling process has bound `tenant`.
+
+  Increments *before* checking the closing mark, and backs the increment out if the
+  tenant turns out to be closing. The two are separate ETS objects, so no single
+  operation covers both -- but this order is still safe, because the increment is
+  published before the check. A closer can therefore never read a count of zero for
+  a bind that goes on to proceed, which is the direction that loses data. The other
+  direction only costs a closer one more pass around its grace period.
+  """
+  @spec bound(module(), String.t()) :: :ok | :closing
+  def bound(repo, tenant) do
+    :ets.update_counter(name(repo), {:binds, tenant}, {2, 1}, {{:binds, tenant}, 0})
+
+    if closing?(repo, tenant) do
+      :ets.update_counter(name(repo), {:binds, tenant}, {2, -1, 0, 0})
+      :closing
+    else
+      :ok
+    end
+  end
+
+  @doc "Records that the calling process has released `tenant`, and marks it used."
+  @spec released(module(), String.t()) :: :ok
+  def released(repo, tenant) do
+    # Clamped at zero: an unmatched release would make a busy tenant look evictable.
+    :ets.update_counter(name(repo), {:binds, tenant}, {2, -1, 0, 0}, {{:binds, tenant}, 0})
+    touch(repo, tenant)
+  end
+
+  @doc "How many processes are bound to `tenant`."
+  @spec count(module(), String.t()) :: non_neg_integer()
+  def count(repo, tenant) do
+    case :ets.lookup(name(repo), {:binds, tenant}) do
+      [{_, count}] -> count
+      [] -> 0
+    end
+  end
+
+  @doc "Marks `tenant` as used now."
+  @spec touch(module(), String.t()) :: :ok
+  def touch(repo, tenant) do
+    :ets.insert(name(repo), {{:used, tenant}, System.monotonic_time()})
+    :ok
+  end
+
+  @doc "When `tenant` was last used, in `System.monotonic_time/0` units."
+  @spec last_used(module(), String.t()) :: integer() | nil
+  def last_used(repo, tenant) do
+    case :ets.lookup(name(repo), {:used, tenant}) do
+      [{_, at}] -> at
+      [] -> nil
+    end
+  end
+
+  @doc "The tenant among `candidates` that should be evicted, if any."
+  @spec least_recently_used(module(), [String.t()]) :: String.t() | nil
+  def least_recently_used(repo, candidates) do
+    candidates
+    |> Enum.reject(&(count(repo, &1) > 0 or closing?(repo, &1)))
+    |> Enum.min_by(&sort_key(repo, &1), fn -> nil end)
+  end
+
+  # Never-used sorts first, and cannot be spelled as a sentinel timestamp:
+  # `System.monotonic_time/0` is normally negative, so 0 would read as recent.
+  defp sort_key(repo, tenant) do
+    case last_used(repo, tenant) do
+      nil -> {0, 0}
+      at -> {1, at}
+    end
+  end
+
+  @doc "Marks `tenant` as closing, so that `bound/2` refuses it."
+  @spec begin_closing(module(), String.t()) :: :ok
+  def begin_closing(repo, tenant) do
+    :ets.insert(name(repo), {{:closing, tenant}, true})
+    :ok
+  end
+
+  @doc "Clears the closing mark for `tenant`."
+  @spec end_closing(module(), String.t()) :: :ok
+  def end_closing(repo, tenant) do
+    :ets.delete(name(repo), {:closing, tenant})
+    :ok
+  end
+
+  @doc "Whether `tenant` is being closed right now."
+  @spec closing?(module(), String.t()) :: boolean()
+  def closing?(repo, tenant), do: :ets.member(name(repo), {:closing, tenant})
+
+  @doc """
+  Drops what is recorded about how `tenant` has been used.
+
+  Deliberately not the closing mark. That is lifecycle state owned by whoever took
+  it, and `close/3` is called *inside* `rename/3` and `delete/2` while they hold one
+  across a file operation -- clearing it here would reopen the window they took it
+  to close.
+  """
+  @spec forget(module(), String.t()) :: :ok
+  def forget(repo, tenant) do
+    table = name(repo)
+    Enum.each([:binds, :used], &:ets.delete(table, {&1, tenant}))
+    :ok
+  end
+
+  @impl true
+  def init(repo) do
+    :ets.new(name(repo), [
+      :named_table,
+      :public,
+      :set,
+      read_concurrency: true,
+      write_concurrency: true
+    ])
+
+    {:ok, repo}
+  end
+end