defmodule Nebulex.Adapters.Multilevel do @moduledoc """ Adapter module for the multi-level cache topology. The Multi-level adapter is a simple layer that works on top of a local or distributed cache implementation, enabling a cache hierarchy by levels. Multi-level caches generally operate by checking the fastest, level 1 (L1) cache first; if it hits, the adapter proceeds at high speed. If that first cache misses, the next fastest cache (level 2, L2) is checked, and so on, before accessing external memory (that can be handled by a `cacheable` decorator). For write functions, the "Write Through" policy is applied by default; this policy ensures that the data is stored safely as it is written throughout the hierarchy. However, it is possible to force the write operation in a specific level (although it is not recommended) via `:level` option, where the value is a positive integer greater than 0. ## How Multi-Level Caches Work A multi-level cache is organized in a hierarchy of cache layers, where each layer is typically faster but smaller than the next. The most common pattern is a **near-cache topology** with two levels: * **L1 (Level 1)**: Local cache - Fast, in-process, limited size. Examples: `Nebulex.Adapters.Local`. * **L2 (Level 2)**: Distributed/remote cache - Slower, shared across nodes, larger capacity. Examples: `Nebulex.Adapters.Partitioned`, Redis, etc. Multi-level caches provide **performance improvement** by reducing latency for frequently accessed data (served from L1) while maintaining large capacity through the L2 layer. ```asciidoc ┌──────────────┐ │ Client │ └──────┬───────┘ │ ┌────────┴─────────┐ │ L1 (Local) │◄── Fast, limited size └────────┬─────────┘ │ ┌────────┴─────────┐ │ L2 (Distributed) │◄── Slower, larger capacity └──────────────────┘ Read: L1 → L2 (replicate back on hit, inclusive mode) Write: L1 → L2 (write-through to all levels) ``` ### Cache Lookup (Read Operations) When you perform a **read operation** (e.g., `get`, `fetch`), the adapter checks the cache levels in order (L1 → L2 → L3, etc.) and returns the value from the **first level that contains it**: 1. Check L1 (fast, local cache). 2. If found → Return immediately. 3. If not found → Check L2. 4. If found in L2 → Return and replicate to L1 (if inclusive mode). 5. If not found → Check L3, L4, etc. 6. If not found in any level → Return error or miss. This **single-hop lookup** pattern ensures you always get data from the fastest available level while maintaining a fallback to slower (but larger) levels. **Example:** - Request for key "user:123" - L1 miss (not in local cache) - L2 hit (found in distributed cache) - Value returned to client - In inclusive mode: value is replicated back to L1 for future requests ### Write-Through Policy The multi-level adapter uses a **Write-Through** policy for write operations. This means data is written to **all levels synchronously** before returning to the caller: 1. Write to L1 (local cache). 2. Write to L2 (distributed cache). 3. Write to L3, L4, etc. (if any) 4. Return to caller. This ensures: * **Consistency**: Data is immediately available in all levels. * **Safety**: Data survives L1 failures (protected in L2). * **Simplicity**: No complex cache invalidation logic needed. **Trade-off**: Write operations are **slower** because they must complete on all levels before returning. However, this is typically acceptable when: - Read operations are usually more frequent. - Distributed writes are less latency-sensitive than reads. - Consistency guarantees are worth the cost. **Failure Handling**: If a write fails at any level, the operation stops (fail-fast). Partially written data at earlier levels is **not rolled back** (atomic writes are not guaranteed across levels). ### Replication in Inclusive Mode When using the **`:inclusive` policy** (default), data can exist in multiple levels simultaneously. The adapter automatically **replicates data backward** (from slower to faster levels) during reads: * **Inclusive mode**: Same key can exist in L1, L2, and L3 simultaneously. * **Exclusive mode**: Same key can exist in only one level at a time. #### Inclusive Mode Workflow 1. **Read (inclusive mode)**: - Check L1, L2, L3 in order. - If found in L2 (not in L1) → Replicate to L1. - Future requests hit L1 (faster). 2. **Write (inclusive mode)**: - Write to L1, L2, L3 (all levels). - Future reads hit L1 (fastest). 3. **Eviction (inclusive mode)**: - L1 evicts a key (e.g., due to size limit). - Key still exists in L2. - Next read finds it in L2 and replicates it back to L1. #### Exclusive Mode Workflow 1. **Read (exclusive mode)**: - Find key in L2. - Return value (do NOT replicate to L1). 2. **Write (exclusive mode)**: - Write to L1, L2, L3 (write-through still applies). - Data exists in all levels. 3. **Eviction (exclusive mode)**: - L1 evicts a key. - L2 still has it. - Next read must go to L2 (no replication). #### Choosing the Right Mode * **Inclusive (default)**: Good for hot-spot data patterns. - Pro: Faster subsequent reads (hot data stays in L1). - Con: More memory usage (duplicated in multiple levels). - Con: `get_all` is slower (requires per-entry replication). * **Exclusive**: Good for large data sets or strict memory limits. - Pro: Less memory usage (data exists once). - Con: Slower reads after eviction (must fetch from L2). - Con: More complex consistency management. ### TTL (Time-To-Live) Handling TTL values are **per-level and independent**: * Each level has its own TTL for the same key. * When writing with TTL, the same TTL is applied to all levels. * Each level independently expires the entry. * In inclusive mode: If L1 expires but L2 still has it, next read replicates from L2 back to L1 with remaining TTL. #### Why TTL is Per-Level Each cache level uses a **different adapter with its own eviction policy**. For example: * **L1 (`Nebulex.Adapters.Local`)**: TTL is evaluated **on-demand** during reads and via a **garbage collector** that periodically removes expired entries and creates new generations. Behavior is controlled by `gc_interval` option. * **L2 (`Nebulex.Adapters.Partitioned` or Redis)**: TTL is handled by the underlying distributed cache. Redis, for example, uses its own TTL expiration mechanism which may differ from the local adapter's approach. Because each adapter manages TTL differently, the **expiration timing and behavior can vary significantly across levels**: - L1 may expire entries within seconds (controlled by GC interval). - L2 may expire entries at slightly different times (depends on Redis TTL). - A key can exist in L2 but be expired in L1. - The remaining TTL is automatically synchronized when data is replicated during reads. This independence is **intentional and necessary** because: - Different adapters have different TTL mechanisms. - Forcing synchronized TTL across levels would require complex coordination. - Allowing independent TTL gives each adapter control over its eviction. - The cost is acceptable because reads automatically sync TTL through replication. This means: - Data can be available in L2 even after L1 expires it. - Evictions are independent per level and controlled by each adapter. - TTL synchronization happens automatically during reads (in inclusive mode). We can define a multi-level cache as follows: defmodule MyApp.Multilevel do use Nebulex.Cache, otp_app: :nebulex, adapter: Nebulex.Adapters.Multilevel defmodule L1 do use Nebulex.Cache, otp_app: :nebulex, adapter: Nebulex.Adapters.Local end defmodule L2 do use Nebulex.Cache, otp_app: :nebulex, adapter: Nebulex.Adapters.Partitioned end end Where the configuration for the cache and its levels must be in your application environment, usually defined in your `config/config.exs`: config :my_app, MyApp.Multilevel, inclusion_policy: :inclusive, levels: [ { MyApp.Multilevel.L1, gc_interval: :timer.hours(12) }, { MyApp.Multilevel.L2, primary: [ gc_interval: :timer.hours(12) ] } ] If your application was generated with a supervisor (by passing `--sup` to `mix new`) you will have a `lib/my_app/application.ex` file containing the application start callback that defines and starts your supervisor. You just need to edit the `start/2` function to start the cache as a supervisor on your application's supervisor: def start(_type, _args) do children = [ {MyApp.Multilevel, []}, ... ] See `Nebulex.Cache` for more information. ## Options This adapter supports the following options and all of them can be given via the cache configuration: #{Nebulex.Adapters.Multilevel.Options.start_options_docs()} ## Shared options Almost all of the cache functions outlined in `Nebulex.Cache` module accept the following options: #{Nebulex.Adapters.Multilevel.Options.common_runtime_options_docs()} ### Queryable API options The following options apply to `get_all`, `count_all`, `delete_all`, and `stream` commands: #{Nebulex.Adapters.Multilevel.Options.queryable_options_docs()} ## Telemetry events The multi-level adapter emits Telemetry events for itself and for each configured cache level. By default, each level gets a unique telemetry prefix derived from its module name, making events naturally distinguishable. ### Default Behavior Each cache level is a separate module (e.g., `MyApp.Multilevel.L1`, `MyApp.Multilevel.L2`), so each gets a unique telemetry prefix by default: defmodule MyApp.Multilevel do use Nebulex.Cache, otp_app: :my_app, adapter: Nebulex.Adapters.Multilevel defmodule L1 do use Nebulex.Cache, otp_app: :my_app, adapter: Nebulex.Adapters.Local end defmodule L2 do use Nebulex.Cache, otp_app: :my_app, adapter: Nebulex.Adapters.Partitioned end end With this setup, events are naturally separated by module: * Multilevel adapter: `[:my_app, :multilevel, :command, :start]` * L1 cache (Local): `[:my_app, :multilevel, :l1, :command, :start]` * L2 cache (Partitioned): `[:my_app, :multilevel, :l2, :command, :start]` * L2 primary storage: `[:my_app, :multilevel, :l2, :primary, :command, :start]` This default behavior is already good for distinguishing events based on which cache level emitted them. ### Custom Telemetry Prefixes (Optional) If you want to override the default prefix for a level, use the optional `:telemetry_prefix` option: config :my_app, MyApp.Multilevel, levels: [ {MyApp.Multilevel.L1, telemetry_prefix: [:my_app, :cache, :l1]}, {MyApp.Multilevel.L2, telemetry_prefix: [:my_app, :cache, :l2]} ] This is useful if you want: * A different naming convention for your telemetry events. * To aggregate events from multiple caches under a common prefix. * To match an existing telemetry structure in your application. ### Events for Distributed L2 Adapters If L2 uses a distributed adapter like `Nebulex.Adapters.Partitioned`, you also get events from its primary storage: MyApp.Multilevel.get(key) # Emits (with default prefixes): # [:my_app, :multilevel, :command, :start] # Multilevel wrapper # [:my_app, :multilevel, :l1, :command, :start] # L1 # [:my_app, :multilevel, :l2, :command, :start] # L2 wrapper # [:my_app, :multilevel, :l2, :primary, :command, :start] # L2 primary Refer to the [Telemetry guide](https://hexdocs.pm/nebulex/telemetry.html) for complete information on Nebulex Telemetry events and how to attach handlers. ## Info API As explained above, the multi-level adapter uses the configured cache levels. Therefore, the information provided by the `info` command will depend on the adapters configured for each level. The Nebulex built-in adapters support the recommended keys `:server`, `:memory`, and `:stats`. Additionally, the multi-level adapter supports: * `:levels_info` - A list with the info map for each cache level. For example, the info for `MyApp.Multilevel` may look like this: iex> MyApp.Multilevel.info!() %{ memory: %{total: nil, used: 206760}, server: %{ cache_module: MyApp.Multilevel, cache_name: :multilevel_inclusive, cache_adapter: Nebulex.Adapters.Multilevel, cache_pid: #PID<0.998.0>, nbx_version: "3.0.0" }, stats: %{ hits: 0, misses: 0, writes: 0, evictions: 0, expirations: 0, deletions: 0, updates: 0 }, levels_info: [ %{ memory: %{total: nil, used: 68920}, server: %{ cache_module: MyApp.Multilevel.L1, cache_name: MyApp.Multilevel.L1, cache_adapter: Nebulex.Adapters.Local, cache_pid: #PID<0.1000.0>, nbx_version: "3.0.0" }, stats: %{ hits: 0, misses: 0, writes: 0, evictions: 0, expirations: 0, deletions: 0, updates: 0 } }, %{ memory: %{total: nil, used: 68920}, nodes: [:"node1@127.0.0.1"], server: %{ cache_module: MyApp.Multilevel.L2, cache_name: MyApp.Multilevel.L2, cache_adapter: Nebulex.Adapters.Partitioned, cache_pid: #PID<0.1015.0>, nbx_version: "3.0.0" }, stats: %{ hits: 0, misses: 0, writes: 0, evictions: 0, expirations: 0, deletions: 0, updates: 0 }, nodes_info: %{ "node1@127.0.0.1": %{ memory: %{total: nil, used: 68920}, server: %{ cache_module: MyApp.Multilevel.L2.Primary, cache_name: MyApp.Multilevel.L2.Primary, cache_adapter: Nebulex.Adapters.Local, cache_pid: #PID<0.1017.0>, nbx_version: "3.0.0" }, stats: %{ hits: 0, misses: 0, writes: 0, evictions: 0, expirations: 0, deletions: 0, updates: 0 } } } } ] } ## Extended API This adapter provides some additional convenience functions to the `Nebulex.Cache` API. ### `inclusion_policy/0,1` Returns the inclusion policy of the cache. iex> MyCache.inclusion_policy() :inclusive ## Near cache topology example The multi-level adapter can be used to implement a **near-cache topology** with different types of cache backends. The most common pattern is L1 (local, fast) + L2 (distributed, larger capacity). ### L1 (Local) + L2 (Redis or External Cache) Instead of using `Nebulex.Adapters.Partitioned` for L2, you can use any external cache system via its adapter. For example, with Redis: defmodule MyApp.NearCache do use Nebulex.Cache, otp_app: :my_app, adapter: Nebulex.Adapters.Multilevel defmodule L1 do use Nebulex.Cache, otp_app: :my_app, adapter: Nebulex.Adapters.Local end defmodule L2 do use Nebulex.Cache, otp_app: :my_app, adapter: Nebulex.Adapters.Redis end end Configuration: config :my_app, MyApp.NearCache, levels: [ {MyApp.NearCache.L1, gc_interval: :timer.hours(1), max_size: 10_000}, {MyApp.NearCache.L2, conn_opts: [host: "localhost", port: 6379]} ] This topology provides: * **L1 (Local)**: In-process cache with fast micro-second latency. * **L2 (Redis)**: Shared across nodes, larger capacity, multi-millisecond latency. **Benefits:** - Hot data is served from L1 (very fast). - Cold data fetches from Redis L2, then cached in L1 on next access. - Data survives node restarts (in Redis). - Works across multiple nodes with a shared Redis instance. **Use case**: Web applications where you want blazing-fast L1 performance for frequently accessed data while using Redis for distributed, shared storage. ## Transactions The multilevel adapter supports distributed transactions via `Nebulex.Distributed.Transaction`, which uses Erlang's `:global` module for cluster-wide lock coordination. **Automatic Node Detection**: When using transactions with multilevel caches, the adapter automatically detects distributed levels (e.g., partitioned or replicated caches) and ensures locks are acquired across all relevant cluster nodes. This means you don't need to manually specify nodes when using distributed levels. **Example**: # Multilevel cache with local L1 and partitioned L2 MyApp.Multilevel.transaction(fn -> # Locks are automatically set across all partitioned cache nodes counter = MyApp.Multilevel.get!(:counter, default: 0) MyApp.Multilevel.put!(:counter, counter + 1) end, keys: [:counter]) **Recommendation**: Always specify the `:keys` option to enable fine-grained locking and maximize concurrency. Without keys, a global lock is used which serializes all transactions. See `Nebulex.Distributed.Transaction` for more information about transaction options, behavior, and performance considerations. ## CAVEATS Because this adapter reuses other existing/configured adapters, it inherits all their limitations too. Therefore, it is highly recommended to check the documentation of the adapters to use. """ # Provide Cache Implementation @behaviour Nebulex.Adapter @behaviour Nebulex.Adapter.KV @behaviour Nebulex.Adapter.Queryable @behaviour Nebulex.Adapter.Info # Inherit distributed transaction implementation using `:global` use Nebulex.Distributed.Transaction # Inherit default observable implementation use Nebulex.Adapter.Observable import Nebulex.Adapter import Nebulex.Utils alias __MODULE__.Options alias Nebulex.Adapters.Common.Info, as: I alias Nebulex.Distributed.Cluster alias Nebulex.Distributed.Helpers, as: H ## Nebulex.Adapter @impl true defmacro __before_compile__(_env) do quote do @doc """ A convenience function to get the cache inclusion policy. """ def inclusion_policy(name \\ __MODULE__) do name |> lookup_meta() |> Map.fetch!(:inclusion_policy) end end end @impl true def init(opts) do # Common options {telemetry_prefix, opts} = Keyword.pop!(opts, :telemetry_prefix) {telemetry, opts} = Keyword.pop!(opts, :telemetry) {cache, opts} = Keyword.pop!(opts, :cache) # Validate options opts = Options.validate_start_opts!(opts) # Get adapter options name = opts[:name] || cache levels = Keyword.fetch!(opts, :levels) inclusion_policy = Keyword.fetch!(opts, :inclusion_policy) # Build multi-level specs {children, meta_list} = children(levels, telemetry_prefix, telemetry) # Build adapter spec child_spec = Supervisor.child_spec( {Nebulex.Adapters.Multilevel.Supervisor, {camelize_and_concat([name, Supervisor]), children}}, id: {__MODULE__, name} ) adapter_meta = %{ telemetry_prefix: telemetry_prefix, telemetry: telemetry, name: name, levels: meta_list, inclusion_policy: inclusion_policy, started_at: DateTime.utc_now() } {:ok, child_spec, adapter_meta} end defp children(levels, telemetry_prefix, telemetry) do levels |> Enum.reverse() |> Enum.reduce({[], []}, fn {l_cache, l_opts}, {child_acc, meta_acc} -> l_opts = Keyword.merge([telemetry_prefix: telemetry_prefix, telemetry: telemetry], l_opts) meta = %{cache: l_cache, name: l_opts[:name]} {[{l_cache, l_opts} | child_acc], [meta | meta_acc]} end) end ## Nebulex.Adapter.KV @impl true def fetch(%{levels: levels, inclusion_policy: policy}, key, opts) do {ml_opts, opts} = validate_ml_opts!(opts) levels |> levels(ml_opts) |> Enum.reduce_while({nil, []}, fn level, {_, prev} -> case with_dynamic_cache(level, :fetch, [key, opts]) do {:error, %Nebulex.KeyError{key: ^key}} = error -> {:cont, {error, [level | prev]}} other -> {:halt, {other, [level | prev]}} end end) |> maybe_replicate(key, policy) end @impl true def put(%{levels: levels}, key, value, on_write, ttl, keep_ttl?, opts) do opts = Keyword.merge(opts, ttl: ttl, keep_ttl: keep_ttl?) case on_write do :put -> while_error(levels, :put, [key, value], {:ok, true}, opts) :put_new -> while_error(levels, :put_new, [key, value], {:ok, true}, opts) :replace -> while_error(levels, :replace, [key, value], {:ok, true}, opts) end end @impl true def put_all(%{levels: levels}, entries, on_write, ttl, opts) do {ml_opts, opts} = validate_ml_opts!(opts) opts = Keyword.put(opts, :ttl, ttl) action = if on_write == :put_new, do: :put_new_all, else: :put_all reducer = fn level, {_, level_acc} -> case with_dynamic_cache(level, action, [entries, opts]) do :ok -> {:cont, {{:ok, true}, [level | level_acc]}} {:ok, true} -> {:cont, {{:ok, true}, [level | level_acc]}} other -> _ = delete_from_levels(level_acc, entries) {:halt, {other, level_acc}} end end levels |> levels(ml_opts) |> Enum.reduce_while({{:ok, true}, []}, reducer) |> elem(0) end @impl true def delete(%{levels: levels}, key, opts) do while_error(levels, :delete, [key], :ok, Keyword.put(opts, :reverse, true)) end @impl true def take(%{levels: levels}, key, opts) do init = wrap_error Nebulex.KeyError, key: key, reason: :not_found levels |> levels(opts) |> do_take(init, key, opts) end defp do_take([], result, _key, _opts) do result end defp do_take([l_meta | rest], {:error, %Nebulex.KeyError{}}, key, opts) do result = with_dynamic_cache(l_meta, :take, [key, opts]) do_take(rest, result, key, opts) end defp do_take(levels, result, key, _opts) do _ = while_error(levels, :delete, [key], :ok, reverse: true) result end @impl true def has_key?(%{levels: levels}, key, opts) do while_ok(levels, :has_key?, [key], {:ok, false}, opts) end @impl true def ttl(%{levels: levels}, key, opts) do init = wrap_error Nebulex.KeyError, key: key, reason: :not_found while_ok(levels, :ttl, [key], init, opts) end @impl true def expire(%{levels: levels}, key, ttl, opts) do with_bool(levels, :expire, [key, ttl], {:ok, false}, opts) end @impl true def touch(%{levels: levels}, key, opts) do with_bool(levels, :touch, [key], {:ok, false}, opts) end @impl true def update_counter(%{levels: levels}, key, amount, default, ttl, opts) do while_error(levels, :incr, [key, amount], nil, [ttl: ttl, default: default] ++ opts) end ## Nebulex.Adapter.Queryable @impl true def execute(adapter_meta, query_spec, opts) do do_execute( adapter_meta, query_spec, validate_ml_opts!(opts, &Options.validate_queryable_opts!/1, [:replicate, :on_error]) ) end defp do_execute(_adapter_meta, %{op: :get_all, query: {:in, []}}, _opts) do {:ok, []} end defp do_execute(_adapter_meta, %{op: op, query: {:in, []}}, _opts) when op in [:count_all, :delete_all] do {:ok, 0} end defp do_execute( %{inclusion_policy: :inclusive} = adapter_meta, %{op: :get_all, query: {:in, keys}, select: select} = query, {ml_opts, opts} ) do replicate? = Keyword.fetch!(ml_opts, :replicate) level = Keyword.get(ml_opts, :level) if replicate? do fetch_keys(adapter_meta, keys, select, [level: level] ++ opts) else do_execute(%{adapter_meta | inclusion_policy: :exclusive}, query, {ml_opts, opts}) end end defp do_execute( %{levels: levels}, %{op: :get_all, query: {:in, keys}, select: select} = query, {ml_opts, opts} ) do reducer = query_reducer(:in) levels |> levels(ml_opts) |> Enum.reduce_while({{:ok, %{}}, keys}, fn level, {_, k_acc} = acc -> query = build_query(%{query | query: {:in, k_acc}, select: {:key, :value}}) level |> with_dynamic_cache(:get_all, [query, opts]) |> reducer.(acc) end) |> elem(0) |> select(select) end defp do_execute(%{levels: levels}, %{op: op, select: select} = query, {ml_opts, opts}) do ml_opts = if op == :delete_all, do: Keyword.put(ml_opts, :reverse, true), else: ml_opts acc_in = if op == :get_all, do: %{}, else: 0 query = build_query(%{query | select: {:key, :value}}) reducer = query_reducer(:q) levels |> levels(ml_opts) |> Enum.reduce_while({:ok, acc_in}, fn level, acc -> level |> with_dynamic_cache(op, [query, opts]) |> reducer.(acc) end) |> select(select) end @impl true def stream(adapter_meta, query, opts) do {ml_opts, opts} = validate_ml_opts!(opts, &Options.validate_queryable_opts!/1, [:replicate, :on_error]) # The multi-level adapter is a wrapper adapter, it doesn't implement any # cache storage, it depends on other cache adapters to do so. There are no # entries to stream from the multi-level adapter itself. Therefore, this # is a workaround to create a stream to trigger the multi-level evaluation # lazily. stream = fn _, _ -> on_error = Keyword.fetch!(ml_opts, :on_error) case do_execute(adapter_meta, %{query | op: :get_all}, {ml_opts, opts}) do {:ok, results} -> {:halt, [results]} {:error, _} when on_error == :nothing -> {:halt, []} {:error, reason} when on_error == :raise -> stacktrace = Process.info(self(), :current_stacktrace) |> elem(1) |> tl() reraise reason, stacktrace end end {:ok, Stream.flat_map(stream, & &1)} end ## Nebulex.Adapter.Transaction # Distributed adapters @distributed_adapters [ Nebulex.Adapters.Partitioned, Nebulex.Adapters.Replicated ] @impl true def transaction(%{levels: levels} = adapter_meta, fun, opts) do {ml_opts, opts} = validate_ml_opts!(opts) # Perhaps one of the levels is a distributed adapter, # then ensure the lock is set on the cluster nodes. nodes = levels |> levels(ml_opts) |> Enum.reduce([node()], fn %{name: name, cache: cache}, acc -> if cache.__adapter__() in @distributed_adapters do Cluster.pg_nodes(name || cache) ++ acc else acc end end) |> Enum.uniq() super(adapter_meta, fun, Keyword.put(opts, :nodes, nodes)) end @impl true def in_transaction?(adapter_meta, opts) do super(adapter_meta, opts) end ## Nebulex.Adapter.Info @impl true def info(adapter_meta, spec, opts) do {ml_opts, opts} = validate_ml_opts!(opts) info(adapter_meta, spec, opts, ml_opts) end defp info(adapter_meta, :all, opts, ml_opts) do with {:ok, levels_info} <- fetch_levels_info(adapter_meta, :all, opts, ml_opts) do levels_info |> info_agg() |> Map.take([:server, :memory, :stats, :levels_info]) |> Map.merge(%{server: I.info(adapter_meta, :server), levels_info: levels_info}) |> wrap_ok() end end defp info(adapter_meta, :server, _opts, _ml_opts) do {:ok, I.info(adapter_meta, :server)} end defp info(adapter_meta, :levels_info, opts, ml_opts) do fetch_levels_info(adapter_meta, :all, opts, ml_opts) end defp info(_adapter_meta, [], _opts, _ml_opts) do {:ok, %{}} end defp info(adapter_meta, spec, opts, ml_opts) when is_list(spec) do server = if Enum.member?(spec, :server) do %{server: I.info(adapter_meta, :server)} else %{} end with {:ok, levels_info} <- fetch_levels_info( adapter_meta, Enum.filter(spec, &(&1 != :levels_info)), opts, ml_opts ) do info = if Enum.member?(spec, :levels_info) do %{levels_info: levels_info} else %{} end levels_info |> info_agg() |> Map.merge(info) |> Map.merge(server) |> wrap_ok() end end defp info(adapter_meta, spec, opts, ml_opts) do with {:ok, levels_info} <- fetch_levels_info(adapter_meta, spec, opts, ml_opts) do {:ok, info_agg(levels_info)} end end defp fetch_levels_info(%{levels: levels}, spec, opts, ml_opts) do levels |> levels(ml_opts) |> Enum.reverse() |> Enum.reduce_while({:ok, []}, fn level_meta, {:ok, acc} -> case with_dynamic_cache(level_meta, :info, [spec, opts]) do {:ok, info} -> {:cont, {:ok, [info | acc]}} {:error, _} = error -> {:halt, error} end end) end defp info_agg(info) do Enum.reduce(info, %{}, &H.merge_info_maps(&2, Map.delete(&1, :server))) end ## Private functions defp with_dynamic_cache(%{cache: cache, name: nil}, action, args) do apply(cache, action, args) end defp with_dynamic_cache(%{cache: cache, name: name}, action, args) do cache.with_dynamic_cache(name, fn -> apply(cache, action, args) end) end defp validate_ml_opts!(opts, fun \\ &Options.validate_common_runtime_opts!/1, keys \\ []) do opts |> fun.() |> Keyword.split([:level, :reverse | keys]) end defp levels(levels, opts) do levels = if level = Keyword.get(opts, :level) do [Enum.at(levels, level - 1)] else levels end if Keyword.get(opts, :reverse) do Enum.reverse(levels) else levels end end defp while_error(levels, fun, args, acc, opts) do {ml_opts, opts} = validate_ml_opts!(opts) levels |> levels(ml_opts) |> do_while_error(fun, args ++ [opts], acc) end defp do_while_error([], _fun, _args, acc) do acc end defp do_while_error([l | levels], fun, args, acc) do case {with_dynamic_cache(l, fun, args), acc} do {:ok, value} -> do_while_error(levels, fun, args, value) {{:ok, bool}, {:ok, acc_bool}} when is_boolean(bool) -> do_while_error(levels, fun, args, {:ok, bool and acc_bool}) {{:ok, value}, nil} -> do_while_error(levels, fun, args, {:ok, value}) {{:ok, _}, {:ok, _} = acc} -> do_while_error(levels, fun, args, acc) {{:error, _} = error, _acc} -> error end end defp while_ok(levels, fun, args, init, opts) do {ml_opts, opts} = validate_ml_opts!(opts) args = args ++ [opts] levels |> levels(ml_opts) |> Enum.reduce_while(init, fn level_meta, acc -> case with_dynamic_cache(level_meta, fun, args) do {:error, %Nebulex.KeyError{}} -> {:cont, acc} {:ok, false} -> {:cont, acc} return -> {:halt, return} end end) end defp with_bool(levels, fun, args, acc, opts) do {ml_opts, opts} = validate_ml_opts!(opts) levels |> levels(ml_opts) |> with_bool(fun, args ++ [opts], acc) end defp with_bool([], _fun, _args, acc) do acc end defp with_bool([l | levels], fun, args, {:ok, acc}) do with {:ok, value} <- with_dynamic_cache(l, fun, args) do with_bool(levels, fun, args, {:ok, value or acc}) end end defp delete_from_levels(levels, entries) do for level_meta <- levels, {key, _} <- entries do with_dynamic_cache(level_meta, :delete, [key, []]) end end defp maybe_replicate({{:ok, value}, [level_meta | [_ | _] = levels]}, key, :inclusive) do case with_dynamic_cache(level_meta, :ttl, [key]) do {:ok, ttl} -> :ok = replicate_to_levels(levels, key, value, ttl) {:ok, value} {:error, %Nebulex.KeyError{key: ^key}} -> # The cache entry expired between the `fetch` and `ttl` calls, # don't replicate the entry {:ok, value} {:error, _} = error -> error end catch {:return, result} -> result end defp maybe_replicate({value, _levels}, _key, _inclusion_policy) do value end defp replicate_to_levels(levels, key, value, ttl) do Enum.each(levels, fn l -> with {:error, _} = error <- with_dynamic_cache(l, :put, [key, value, [ttl: ttl]]) do throw({:return, error}) end end) end defp build_query(%{select: select, query: query}) do query = with {:q, q} <- query, do: {:query, q} [query, select: select] end defp select({:ok, map}, select) when is_map(map) do case select do :key -> Map.keys(map) :value -> Map.values(map) _else -> Map.to_list(map) end |> wrap_ok() end defp select(other, _select) do other end defp fetch_keys(adapter_meta, keys, select, opts) do Enum.reduce_while(keys, {:ok, []}, fn k, {:ok, acc} -> case {fetch(adapter_meta, k, opts), select} do {{:ok, _v}, :key} -> {:cont, {:ok, [k | acc]}} {{:ok, v}, :value} -> {:cont, {:ok, [v | acc]}} {{:ok, v}, _} -> {:cont, {:ok, [{k, v} | acc]}} {{:error, %Nebulex.KeyError{}}, _} -> {:cont, {:ok, acc}} {error, _} -> {:halt, error} end end) end defp query_reducer(:in) do fn {:ok, res}, {{:ok, acc}, keys_acc} -> # Ensure no duplicates {acc, res_keys} = Enum.reduce(res, {acc, []}, fn {k, v}, {acc, k_acc} -> {Map.put_new(acc, k, v), [k | k_acc]} end) case keys_acc -- res_keys do [] -> {:halt, {{:ok, acc}, []}} keys_acc -> {:cont, {{:ok, acc}, keys_acc}} end {:error, _} = error, _ -> {:halt, {error, []}} end end defp query_reducer(:q) do fn {:ok, res}, {:ok, acc} when is_list(res) -> # Ensure no duplicates acc = Enum.reduce(res, acc, &Map.put_new(&2, elem(&1, 0), elem(&1, 1))) {:cont, {:ok, acc}} {:ok, res}, {:ok, acc} when is_integer(res) -> {:cont, {:ok, acc + res}} {:error, _} = error, _ -> {:halt, error} end end end