defmodule ActiveMemory.ActiveRepo do @moduledoc """ # The ActiveRepo An `ActiveRepo` manages multiple `ActiveMemory.Table`s from a single process. It is the multi-table counterpart to `ActiveMemory.Store` (which manages a single table), giving you one supervised entry point and a unified API over many tables. > It is named `ActiveRepo` rather than `Repo` so it does not collide with an > application's `Ecto.Repo` while keeping the familiar "repo" terminology. ```elixir defmodule MyApp.ActiveRepo do use ActiveMemory.ActiveRepo, tables: [ MyApp.People.Person, {MyApp.Dogs.Dog, seed_file: Path.expand("dog_seeds.exs", __DIR__), before_init: [{:warm, []}]} ] end ``` Add the `ActiveRepo` to your supervision tree like any other process: ```elixir children = [MyApp.ActiveRepo] ``` Tables may freely mix `:ets` and `:mnesia` adapters; each operation dispatches to the adapter configured on the given table. ## ActiveRepo API Every operation an `ActiveMemory.Store` offers is available here, with the same behavior; only the arities differ. Reads and `withdraw` take the table module as the first argument, while `write` and `delete` infer the table from the struct (or from a changeset's data). - `ActiveRepo.all/2` Get all records stored in a table, optionally ordered and paged - `ActiveRepo.count/2` Count the records in a table, without reading them - `ActiveRepo.delete/1` Delete the record provided, matched in full (see [Deleting a record](#module-deleting-a-record)) - `ActiveRepo.delete_all/1` Delete all records stored in a table - `ActiveRepo.exists?/3` Whether any record in a table matches an attributes search or `match` query - `ActiveRepo.get/2` Get the record with the given primary key, or `{:error, :not_found}` - `ActiveRepo.get!/2` Like `get/2` but raises `ActiveMemory.NotFoundError` - `ActiveRepo.get_by/2` Get the single record in a table matching an attributes search - `ActiveRepo.get_by!/2` Like `get_by/2` but raises `ActiveMemory.NotFoundError` - `ActiveRepo.one/2` Get one record from a table matching either an attributes search or `match` query. Raises `ActiveMemory.MultipleResultsError` when several match - `ActiveRepo.one!/2` Like `one/2` but raises `ActiveMemory.NotFoundError` - `ActiveRepo.reload/1` Re-read a record by its primary key, inferring the table - `ActiveRepo.reload!/1` Like `reload/1` but raises `ActiveMemory.NotFoundError` - `ActiveRepo.select/3` Get all records from a table matching either an attributes search or `match` query, optionally ordered and paged - `ActiveRepo.withdraw/2` **Atomically** get one record from a table matching either an attributes search or `match` query, delete the record and return it — exactly one concurrent caller wins, making it safe for take-once workloads - `ActiveRepo.write/1` Write a record into its table, from a struct or an `Ecto.Changeset`. An invalid changeset is returned as `{:error, changeset}` with its `action` set to `:insert`, exactly like `c:Ecto.Repo.insert/2` An operation for a struct or table that is not part of the `ActiveRepo` returns `{:error, :unknown_table}`. ## Reading, counting, ordering These behave exactly as they do on a `ActiveMemory.Store`, which documents them in full: [reading a single record](`ActiveMemory.Store`), counting, and ordering with `:order_by`/`:limit`/`:offset`. `get/2` reads by the table's primary key (its first field), `count/2` is O(1) and takes `sweep: true` on a `ttl` table, and `reload/1` infers its table from the struct as `write/1` and `delete/1` do. ## Deleting a record `delete/1` removes an **exact** record match: the struct you pass is compared field for field against what is stored. A struct that has diverged from the stored copy — a stale read, or one modified in memory — removes nothing and still returns `:ok`, the same answer given for a record that was never there. That is the only correct behavior for a `:bag` table, and on a `:set` table it keeps a delete from clobbering a newer version of the record. When you hold an identifier rather than a record you know is current, use `withdraw/2`: it matches on a query, is atomic, and reports whether anything was removed with `{:ok, record}` or `{:error, :not_found}`. ## Concurrency Like a `Store`, an `ActiveRepo` is a `GenServer`, but the data functions above are **not** routed through that process and are **not** serialized by it. They run in the **caller's** process and delegate straight to each table's adapter, so reads and writes execute with ETS/Mnesia concurrency — the single `GenServer` is not a bottleneck. Only lifecycle and metadata operations (`init`, `state/0`, `reload_seeds/1`) use the `GenServer`. These functions live on the `GenServer` module purely for **organization**: the `ActiveRepo` is the single place responsible for how the application talks to its tables, following the Single Responsibility Principle. See the [S.T.O.N.E principles](https://www.hpt-consulting.org/blog/stone-principles) for the broader design philosophy. ## Tables and per-table options Each entry of `tables:` is either a table module or a `{table, opts}` tuple. The supported per-table options mirror the single-table `ActiveMemory.Store`: - `seed_file` a path to a seed file auto loaded when the table is first created - `before_init` methods (defined on the `ActiveRepo`) run during the table's setup ## Expiry (TTL) Any table whose `ActiveMemory.Table` declares a `ttl` expires its records automatically: reads never return an expired record, and the `ActiveRepo` periodically sweeps expired records from every `ttl` table it owns to reclaim memory. The sweep cadence defaults to one minute and can be set with the `sweep_interval` option (milliseconds). Tables without a `ttl` are left untouched, and the sweep is only scheduled when at least one table uses a `ttl`. ## Initial State Like a `Store`, an `ActiveRepo` is a `GenServer` with state. The default state is: ```elixir %{started_at: "date time when first started", tables: [MyApp.People.Person, ...]} ``` Supply a `{method, args}` tuple to the `initial_state` keyword to override it; the method must return `{:ok, new_state}`. ## Resilience ETS tables created by an `ActiveRepo` get the same `ActiveMemory.TableHeir` protection as a `Store`: they survive an `ActiveRepo` crash and are reclaimed on restart, and seed files are not re-run on recovery. See `ActiveMemory.Store` for the `before_init` recovery caveat, which applies here as well. """ @default_sweep_interval :timer.seconds(60) defmacro __using__(opts) do quote do use GenServer alias ActiveMemory.Operations opts = unquote(Macro.expand(opts, __CALLER__)) @repo_tables Enum.map(Keyword.fetch!(opts, :tables), fn {table, table_opts} -> {table, table_opts} table -> {table, []} end) @tables Enum.map(@repo_tables, fn {table, _opts} -> table end) @initial_state Keyword.get(opts, :initial_state, :default) @sweep_interval Keyword.get(opts, :sweep_interval, unquote(@default_sweep_interval)) def start_link(options \\ []) do GenServer.start_link(__MODULE__, options, name: __MODULE__) end @impl true def init(_) do with :ok <- __setup_tables__(), {:ok, initial_state} <- __initial_state__() do __schedule_sweep__() {:ok, initial_state} end end @spec all(atom(), keyword()) :: list(map()) | {:error, :unknown_table} def all(table, opts \\ []) def all(table, opts) when table in @tables, do: Operations.all(table, opts) def all(_table, _opts), do: {:error, :unknown_table} @spec count(atom(), keyword()) :: non_neg_integer() | {:error, :unknown_table} def count(table, opts \\ []) def count(table, opts) when table in @tables, do: Operations.count(table, opts) def count(_table, _opts), do: {:error, :unknown_table} @spec delete(map()) :: :ok | {:error, any()} def delete(%{__struct__: table} = struct) when table in @tables do Operations.delete(struct, table) end def delete(_struct), do: {:error, :unknown_table} @spec delete_all(atom()) :: :ok | {:error, any()} def delete_all(table) when table in @tables, do: Operations.delete_all(table) def delete_all(_table), do: {:error, :unknown_table} @spec exists?(atom(), map() | tuple(), keyword()) :: boolean() | {:error, :unknown_table} def exists?(table, query, opts \\ []) def exists?(table, query, opts) when table in @tables, do: Operations.exists?(query, table, opts) def exists?(_table, _query, _opts), do: {:error, :unknown_table} @spec get(atom(), any()) :: {:ok, map()} | {:error, any()} def get(table, key) when table in @tables, do: Operations.get(key, table) def get(_table, _key), do: {:error, :unknown_table} @spec get!(atom(), any()) :: map() def get!(table, key) when table in @tables, do: Operations.get!(key, table) @spec get_by(atom(), map()) :: {:ok, map()} | {:error, any()} def get_by(table, query) when table in @tables, do: Operations.get_by(query, table) def get_by(_table, _query), do: {:error, :unknown_table} @spec get_by!(atom(), map()) :: map() def get_by!(table, query) when table in @tables, do: Operations.get_by!(query, table) @spec one(atom(), map() | tuple()) :: {:ok, map()} | {:error, any()} def one(table, query) when table in @tables, do: Operations.one(query, table) def one(_table, _query), do: {:error, :unknown_table} @spec one!(atom(), map() | tuple()) :: map() def one!(table, query) when table in @tables, do: Operations.one!(query, table) @spec reload(map()) :: {:ok, map()} | {:error, any()} def reload(%{__struct__: table} = struct) when table in @tables do Operations.reload(struct, table) end def reload(_struct), do: {:error, :unknown_table} @spec reload!(map()) :: map() def reload!(%{__struct__: table} = struct) when table in @tables do Operations.reload!(struct, table) end def reload_seeds(table) when table in @tables do GenServer.call(__MODULE__, {:reload_seeds, table}) end def reload_seeds(_table), do: {:error, :unknown_table} @spec select(atom(), map() | tuple(), keyword()) :: {:ok, list(map())} | {:error, any()} def select(table, query, opts \\ []) def select(table, query, opts) when table in @tables, do: Operations.select(query, table, opts) def select(_table, _query, _opts), do: {:error, :unknown_table} def state do GenServer.call(__MODULE__, :state) end @spec withdraw(atom(), map() | tuple()) :: {:ok, map()} | {:error, any()} def withdraw(table, query) when table in @tables, do: Operations.withdraw(query, table) def withdraw(_table, _query), do: {:error, :unknown_table} @spec write(map() | Ecto.Changeset.t()) :: {:ok, map()} | {:error, Ecto.Changeset.t() | any()} def write(%Ecto.Changeset{data: %{__struct__: table}} = changeset) when table in @tables do Operations.write(changeset, table) end def write(%{__struct__: table} = struct) when table in @tables do Operations.write(struct, table) end def write(_struct), do: {:error, :unknown_table} @impl true def handle_call({:reload_seeds, table}, _from, state) do {:reply, Operations.seed(__seed_file__(table), table), state} end @impl true def handle_call(:state, _from, state), do: {:reply, state, state} # Sent by `ActiveMemory.TableHeir` when it hands a recovered ETS table back to # this repo on restart. Ownership transfers with the message; nothing further # is required here. @impl true def handle_info({:"ETS-TRANSFER", _table_ref, _from, _data}, state) do {:noreply, state} end # Periodically reclaims memory from expired records across every `ttl` table. def handle_info(:sweep, state) do Operations.sweep_expired(@tables, System.system_time(:millisecond)) __schedule_sweep__() {:noreply, state} end def handle_info(_message, state), do: {:noreply, state} # Only the clause matching the compile-time option is generated so the # Elixir 1.19+ type checker never sees an unreachable clause. if @initial_state == :default do defp __initial_state__ do {:ok, %{ started_at: DateTime.utc_now(), tables: @tables }} end else defp __initial_state__ do {method, args} = @initial_state :erlang.apply(__MODULE__, method, args) end end defp __maybe_seed__(:recovered, _seed_file, _table), do: {:ok, :seed_success} defp __maybe_seed__(:created, seed_file, table), do: Operations.seed(seed_file, table) # Schedules the next expiry sweep only when at least one table uses a `ttl`. # The `ttl`s are read at runtime by `Operations`, never here: inspecting the # table modules while this one compiles would make them compile time # dependencies, which breaks tooling that loads this file on its own. defp __schedule_sweep__, do: Operations.schedule_sweep(@tables, @sweep_interval) defp __seed_file__(table) do {_table, table_opts} = List.keyfind(@repo_tables, table, 0) Keyword.get(table_opts, :seed_file) end defp __setup_table__(table, table_opts) do with {:ok, table_status} <- Operations.create_table(table), {:ok, :seed_success} <- __maybe_seed__(table_status, Keyword.get(table_opts, :seed_file), table), {:ok, _result} <- Operations.before_init(Keyword.get(table_opts, :before_init, :default), __MODULE__) do :ok end end defp __setup_tables__ do Enum.reduce_while(@repo_tables, :ok, fn {table, table_opts}, _acc -> case __setup_table__(table, table_opts) do :ok -> {:cont, :ok} {:error, _reason} = error -> {:halt, error} end end) end end end end