kura_cache

-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.