-module(kura_cache). -moduledoc """ Application-side read cache for kura projects. Sending one chat message, serving one authenticated request, executing one webhook — hot paths read the same few rows over and over, and none of them change often. This module keeps those values in an ETS table owned by a supervised process and invalidates them on write. The shape: - one public ETS table (`read_concurrency`), owned by the `kura_cache` gen_server so it outlives the request processes that fill it; - `fetch/3` is the only read API: a hit returns the cached value, a miss runs the loader and caches whatever it returns — except `undefined` and `{error, _}`, which pass through uncached (caching a transient failure turns a blip into a bug); - entries carry a per-kind TTL as a backstop: an eviction the application forgot to write costs staleness for seconds, not forever. A kind with no configured TTL is not cached at all — adding a kind is a deliberate act, so a stale read can always be traced to the configuration; - `evict/2` after every write to the underlying data. In a cluster the eviction must also reach the other nodes: `kura_cache` does not pick a transport, it calls the configured `broadcast` fun and exports `local_evict/2` / `local_flush/1` for the receiving side. Wire it to syn, pg, or nothing (single node). The cache is not a source of truth and never serves writes — anything that must be correct under concurrency goes straight to the database. ## Row-cache sugar `fetch_row/3` / `evict_row/2` cache rows of a kura schema keyed by primary key: the kind is the schema module, the key is normalized from the schema's key fields (a bare value for single-column keys, a `#{field => value}` map for composite ones — the same spec convention as `kura_repo_worker:get/3`). The discipline is cache-aside, see `docs/cache-aside-guide.md`: only read the cache by primary key, write the database first, evict (never re-put) after writes. ## Configuration ```erlang kura_cache:start_link(#{ kinds => #{ my_schema => 60_000, %% TTL in ms; kind = schema module or any atom fanout => 30_000 }, broadcast => fun my_app:broadcast_eviction/1, %% optional sweep_interval => 60_000 %% optional, default 60s }). ``` ## Telemetry `[kura_cache, hit]`, `[kura_cache, miss]` and `[kura_cache, evict]` are executed with `#{count => 1}` and `#{kind => Kind}` metadata. """. -behaviour(gen_server). -export([ start_link/1, fetch/3, evict/2, flush/1, clear/0, stats/0 ]). -export([ fetch_row/3, evict_row/2, flush_rows/1 ]). -export([ local_evict/2, local_flush/1 ]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]). -define(TABLE, kura_cache). -define(PT_KINDS, {kura_cache, kinds}). -define(PT_BROADCAST, {kura_cache, broadcast}). -type kind() :: atom(). -type broadcast_event() :: {evict, kind(), term()} | {flush, kind()}. -export_type([kind/0, broadcast_event/0]). %%---------------------------------------------------------------------- %% API %%---------------------------------------------------------------------- -doc """ Start the cache. `kinds` maps a kind to its TTL in milliseconds; `broadcast`, when given, is called with a `t:broadcast_event/0` after every `evict/2` and `flush/1` — deliver it to the other nodes and call `local_evict/2` / `local_flush/1` there. """. -spec start_link(map()) -> {ok, pid()} | {error, term()}. start_link(Opts) when is_map(Opts) -> gen_server:start_link({local, ?MODULE}, ?MODULE, Opts, []). -doc """ Cached read. On a miss `LoadFun` is run and its value cached; `undefined` and `{error, _}` values are returned but never cached. A kind without a configured TTL is never cached, and a cache that is not running behaves as all-miss — callers need no guard. """. -spec fetch(kind(), term(), fun(() -> term())) -> term(). fetch(Kind, Key, LoadFun) -> case lookup(Kind, Key) of {ok, Value} -> telemetry:execute([kura_cache, hit], #{count => 1}, #{kind => Kind}), Value; miss -> telemetry:execute([kura_cache, miss], #{count => 1}, #{kind => Kind}), Value = LoadFun(), _ = put_value(Kind, Key, Value), Value end. -doc "Drop one entry locally and hand the event to the broadcast hook.". -spec evict(kind(), term()) -> ok. evict(Kind, Key) -> local_evict(Kind, Key), broadcast({evict, Kind, Key}). -doc "Drop every entry of a kind, locally and via the broadcast hook.". -spec flush(kind()) -> ok. flush(Kind) -> local_flush(Kind), broadcast({flush, Kind}). -doc "Drop everything, locally only.". -spec clear() -> ok. clear() -> try ets:delete_all_objects(?TABLE), ok catch error:badarg -> ok end. -spec stats() -> #{size => non_neg_integer(), memory_words => non_neg_integer()}. stats() -> #{ size => info_or_zero(size), memory_words => info_or_zero(memory) }. %% ets:info/2 on a table that does not exist returns `undefined`, and a %% stats reader wants a number either way. info_or_zero(Item) -> case ets:info(?TABLE, Item) of undefined -> 0; N -> N end. %%---------------------------------------------------------------------- %% Row-cache sugar %%---------------------------------------------------------------------- -doc """ Cached row read, keyed by the schema's primary key. The kind is the schema module (configure its TTL under that name); `Spec` is a bare value for single-column keys or a `#{field => value}` map for composite ones. `LoadFun` typically wraps the system-of-record read: ```erlang kura_cache:fetch_row(user_schema, Id, fun() -> kura_repo_worker:get(db_repo, user_schema, Id) end) ``` """. -spec fetch_row(module(), term(), fun(() -> {ok, map()} | {error, term()})) -> {ok, map()} | {error, term()}. fetch_row(Schema, Spec, LoadFun) -> fetch(Schema, row_key(Schema, Spec), LoadFun). -doc "Evict the row addressed by the primary-key `Spec`.". -spec evict_row(module(), term()) -> ok. evict_row(Schema, Spec) -> evict(Schema, row_key(Schema, Spec)). -doc """ Drop every cached row of `Schema` — the backstop after `update_all`, `delete_all` or raw SQL, where the affected keys are unknown. """. -spec flush_rows(module()) -> ok. flush_rows(Schema) -> flush(Schema). %%---------------------------------------------------------------------- %% Receiving side of the broadcast hook %%---------------------------------------------------------------------- -doc "Local-only eviction: what a broadcast receiver calls.". -spec local_evict(kind(), term()) -> ok. local_evict(Kind, Key) -> telemetry:execute([kura_cache, evict], #{count => 1}, #{kind => Kind}), try ets:delete(?TABLE, {Kind, Key}), ok catch error:badarg -> ok end. -doc "Local-only flush of a kind: what a broadcast receiver calls.". -spec local_flush(kind()) -> ok. local_flush(Kind) -> try ets:match_delete(?TABLE, {{Kind, '_'}, '_', '_'}), ok catch error:badarg -> ok end. %%---------------------------------------------------------------------- %% gen_server %%---------------------------------------------------------------------- init(Opts) -> _ = ets:new(?TABLE, [ set, public, named_table, {read_concurrency, true}, {write_concurrency, true} ]), persistent_term:put(?PT_KINDS, maps:get(kinds, Opts, #{})), persistent_term:put(?PT_BROADCAST, normalize_broadcast(maps:get(broadcast, Opts, undefined))), Sweep = maps:get(sweep_interval, Opts, 60_000), {ok, _} = timer:send_interval(Sweep, sweep), {ok, #{}}. handle_call(_Req, _From, State) -> {reply, {error, unknown_call}, State}. handle_cast(_Msg, State) -> {noreply, State}. handle_info(sweep, State) -> sweep(), {noreply, State}; handle_info(_Msg, State) -> {noreply, State}. terminate(_Reason, _State) -> _ = persistent_term:erase(?PT_KINDS), _ = persistent_term:erase(?PT_BROADCAST), ok. %%---------------------------------------------------------------------- %% Internal %%---------------------------------------------------------------------- lookup(Kind, Key) -> try ets:lookup(?TABLE, {Kind, Key}) of [{_, Value, Expiry}] -> case erlang:monotonic_time(millisecond) < Expiry of %% lazily expired; the sweep reaps the memory true -> {ok, Value}; false -> miss end; [] -> miss catch %% cache not running (tests, boot): behave as all-miss error:badarg -> miss end. put_value(_Kind, _Key, undefined) -> ok; put_value(_Kind, _Key, {error, _}) -> ok; put_value(Kind, Key, Value) -> case maps:get(Kind, persistent_term:get(?PT_KINDS, #{}), undefined) of undefined -> ok; Ttl -> Expiry = erlang:monotonic_time(millisecond) + Ttl, try ets:insert(?TABLE, {{Kind, Key}, Value, Expiry}), ok catch error:badarg -> ok end end. broadcast(Event) -> case persistent_term:get(?PT_BROADCAST, undefined) of undefined -> ok; Fun -> %% The hook must never take the write path down with it. try _ = Fun(Event), ok catch _:_ -> ok end end. normalize_broadcast(undefined) -> undefined; normalize_broadcast(Fun) when is_function(Fun, 1) -> Fun; normalize_broadcast({M, F}) -> fun(Event) -> M:F(Event) end. %% Same key-spec convention as kura_repo_worker:get/3: a bare value is %% sugar for a single-column key; a map addresses each key column. The %% normalized key is the tuple of key values in schema key order, so a %% map spec and an equivalent future bare spec cannot disagree. row_key(Schema, Spec) when is_map(Spec) -> list_to_tuple([fetch_key_field(F, Spec, Schema) || F <- kura_schema:key(Schema)]); row_key(Schema, Value) -> case kura_schema:key(Schema) of [_] -> {Value}; Key -> error({key_spec_required, Schema, Key}) end. fetch_key_field(F, Spec, Schema) -> case Spec of #{F := V} -> V; #{} -> error({incomplete_key, Schema, F}) end. %% Expired entries are skipped on read; this only reclaims the memory %% of keys nobody asks for any more. sweep() -> Now = erlang:monotonic_time(millisecond), try Spec = [{{'_', '_', '$1'}, [{'<', '$1', Now}], [true]}], _ = ets:select_delete(?TABLE, Spec), ok catch error:badarg -> ok end.