defmodule Plausible.Cache do @moduledoc """ Caching interface specific for Plausible. Usage: use Plausible.Cache # - Implement the callbacks required # - Optionally override `unwrap_cache_keys/1` # - Populate the cache with `Plausible.Cache.Warmer` Serves as a wrapper around `Plausible.Cache.Adapter`, where the underlying implementation can be transparently swapped. Even though normally the relevant Adapter processes are started, cache access is disabled during tests via the `:plausible, #{__MODULE__}, enabled: bool()` application env key. This can be overridden on case by case basis, using the child specs options. The `base_db_query/0` callback is used to generate the base query that is executed on every cache refresh. There are two modes of refresh operation: `:all` and `:updated_recently`; the former will invoke the query as is and clear all the existing entries, while the latter will attempt to limit the query to only the records that have been updated in the last 15 minutes and try to merge the new results with existing cache entries. Both refresh modes are normally executed periodically from within a warmer process; see: `Plausible.Cache.Warmer`. The reason for two modes is that the latter is lighter on the database and can be executed more frequently. When Cache is disabled via application env, the `get/1` function falls back to pure database lookups (implemented via `get_from_source/1` callback. This should help with introducing cached lookups in existing code, so that no existing tests should break. Refreshing the cache emits telemetry event defined as per `telemetry_event_refresh/2`. """ @doc "Unique cache name, used by underlying implementation" @callback name() :: atom() @doc "Supervisor child id, must be unique within the supervision tree" @callback child_id() :: atom() @doc "Optional repo to use. Defaults to Plausible.Repo" @callback repo() :: Ecto.Repo.t() @doc "Counts all items at the source, an aggregate query most likely" @callback count_all() :: integer() @doc "Optionally unwraps the keys of the cache items, in case one item is stored under multiple keys" @callback unwrap_cache_keys([any()]) :: [{any(), any()}] @doc "Returns the base Ecto query used to refresh the cache" @callback base_db_query() :: Ecto.Query.t() @doc "Retrieves the item from the source, in case the cache is disabled" @callback get_from_source(any()) :: any() @doc "Looks for application env value at `:plausible, #{__MODULE__}, enabled: bool()`" def enabled?() do Application.fetch_env!(:plausible, __MODULE__)[:enabled] == true end # credo:disable-for-this-file Credo.Check.Refactor.LongQuoteBlocks defmacro __using__(_opts) do quote do @behaviour Plausible.Cache @modes [:all, :updated_recently] alias Plausible.Cache.Adapter require Logger @spec put(any(), any(), Keyword.t()) :: any() def put(key, value, opts \\ []) do cache_name = Keyword.get(opts, :cache_name, name()) Adapter.put(cache_name, key, value) end @spec broadcast_put(any(), any(), Keyword.t()) :: :ok def broadcast_put(key, value, opts \\ []) do cache_name = Keyword.get(opts, :cache_name, name()) multicall_timeout = Keyword.get(opts, :multicall_timeout, :timer.seconds(5)) {:ok, _} = Task.start(fn -> :rpc.multicall(Adapter, :put, [cache_name, key, value, opts], multicall_timeout) end) :ok end @spec get(any(), Keyword.t()) :: any() | nil def get(key, opts \\ []) def get(key, opts) when is_list(opts) do cache_name = Keyword.get(opts, :cache_name, name()) force? = Keyword.get(opts, :force?, false) if Plausible.Cache.enabled?() or force? do Adapter.get(cache_name, key) else get_from_source(key) end end defoverridable get: 2 @spec get_or_store(any(), (-> any()), Keyword.t()) :: any() | nil def get_or_store(key, fallback_fn, opts \\ []) when is_function(fallback_fn, 0) and is_list(opts) do cache_name = Keyword.get(opts, :cache_name, name()) force? = Keyword.get(opts, :force?, false) if Plausible.Cache.enabled?() or force? do Adapter.get(cache_name, key, fallback_fn) else get_from_source(key) || fallback_fn.() end end def unwrap_cache_keys(items), do: items defoverridable unwrap_cache_keys: 1 def repo(), do: Plausible.Repo defoverridable repo: 0 @spec refresh_all(Keyword.t()) :: :ok def refresh_all(opts \\ []) do refresh( :all, base_db_query(), Keyword.put(opts, :delete_stale_items?, true) ) end @spec refresh_updated_recently(Keyword.t()) :: :ok def refresh_updated_recently(opts \\ []) def refresh_updated_recently(opts) do recently_updated_query = from([s, ...] in base_db_query(), order_by: [asc: s.updated_at], where: s.updated_at > ago(^15, "minute") ) refresh( :updated_recently, recently_updated_query, Keyword.put(opts, :delete_stale_items?, false) ) end defoverridable refresh_updated_recently: 1 @spec merge_items(new_items :: [any()], opts :: Keyword.t()) :: :ok def merge_items(new_items, opts \\ []) def merge_items([], _), do: :ok def merge_items(new_items, opts) do new_items = unwrap_cache_keys(new_items) cache_name = Keyword.get(opts, :cache_name, name()) :ok = Adapter.put_many(cache_name, new_items) if Keyword.get(opts, :delete_stale_items?, true) do old_keys = Adapter.keys(cache_name) new = MapSet.new(Enum.into(new_items, [], fn {k, _} -> k end)) old = MapSet.new(old_keys) old |> MapSet.difference(new) |> Enum.each(fn k -> Adapter.delete(cache_name, k) end) end :ok end @spec child_spec(Keyword.t()) :: Supervisor.child_spec() def child_spec(opts) do cache_name = Keyword.get(opts, :cache_name, name()) child_id = Keyword.get(opts, :child_id, child_id()) Adapter.child_spec(cache_name, child_id, opts) end @doc """ Ensures the cache has non-zero size unless no items exist. Useful for orchestrating app startup to prevent the service going up asynchronously with an empty cache. """ @spec ready?(atom()) :: boolean def ready?(cache_name \\ name()) do case size(cache_name) do n when is_integer(n) and n > 0 -> true 0 -> count_all() == 0 _ -> false end end defdelegate size(cache_name \\ name()), to: Plausible.Cache.Adapter @spec telemetry_event_refresh(atom(), atom()) :: list(atom()) def telemetry_event_refresh(cache_name \\ name(), mode) when mode in @modes do [:plausible, :cache, cache_name, :refresh, mode] end defp refresh(mode, query, opts) when mode in @modes do cache_name = Keyword.get(opts, :cache_name, name()) Plausible.PromEx.Plugins.PlausibleMetrics.measure_duration( telemetry_event_refresh(cache_name, mode), fn -> try do items = repo().all(query) :ok = merge_items(items, opts) catch _, e -> Logger.error("Error refreshing '#{cache_name}' - #{inspect(e)}") end end ) :ok end end end end