lattice_maps

-module(lattice_maps@lww_map).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/lattice_maps/lww_map.gleam").
-export([new/0, pruned_timestamp/1, set/4, get/2, remove/3, keys/1, tombstone_count/1, prune/2, values/1, merge/2, to_json/1, from_json/1]).
-export_type([l_w_w_map/0]).

-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.

?MODULEDOC(
    " A last-writer-wins map (LWW-Map) CRDT.\n"
    "\n"
    " Each key maps to a value and a timestamp. On conflict, the entry with the\n"
    " higher timestamp wins. Removal is timestamp-based (tombstone): a remove at\n"
    " timestamp T beats any set at timestamp < T. Keys are strings; values are\n"
    " strings.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " import lattice_maps/lww_map\n"
    "\n"
    " let a = lww_map.new() |> lww_map.set(\"name\", \"Alice\", 1)\n"
    " let b = lww_map.new() |> lww_map.set(\"name\", \"Bob\", 2)\n"
    " let merged = lww_map.merge(a, b)\n"
    " lww_map.get(merged, \"name\")  // -> Ok(\"Bob\")\n"
    " ```\n"
    "\n"
    " ## Tombstone Management\n"
    "\n"
    " Removing a key creates a tombstone that persists until pruned. Use\n"
    " `tombstone_count` to monitor growth and `prune` to reclaim space once all\n"
    " replicas have synced past a stable timestamp. The embedded\n"
    " `pruned_timestamp` ensures that stale entries from unsynced replicas are\n"
    " automatically rejected during merge (zombie prevention).\n"
).

-opaque l_w_w_map() :: {l_w_w_map,
        gleam@dict:dict(binary(), {gleam@option:option(binary()), integer()}),
        integer()}.

-file("src/lattice_maps/lww_map.gleam", 55).
?DOC(" Create a new empty LWW-Map.\n").
-spec new() -> l_w_w_map().
new() ->
    {l_w_w_map, maps:new(), 0}.

-file("src/lattice_maps/lww_map.gleam", 64).
?DOC(
    " Return the pruned timestamp.\n"
    "\n"
    " This is the highest stable timestamp passed to `prune`. Entries from remote\n"
    " replicas with timestamps at or below this value are treated as zombies during\n"
    " merge and discarded. A value of `0` means no pruning has occurred.\n"
).
-spec pruned_timestamp(l_w_w_map()) -> integer().
pruned_timestamp(Map) ->
    erlang:element(3, Map).

-file("src/lattice_maps/lww_map.gleam", 72).
?DOC(
    " Set a key to a value at the given timestamp.\n"
    "\n"
    " If the key already has an entry with an equal or higher timestamp, the\n"
    " existing entry is kept (LWW semantics: strictly greater timestamp wins).\n"
).
-spec set(l_w_w_map(), binary(), binary(), integer()) -> l_w_w_map().
set(Map, Key, Value, Timestamp) ->
    Should_update = case gleam_stdlib:map_get(erlang:element(2, Map), Key) of
        {error, _} ->
            true;

        {ok, {_, Existing_ts}} ->
            Timestamp > Existing_ts
    end,
    case Should_update of
        true ->
            {l_w_w_map,
                gleam@dict:insert(
                    erlang:element(2, Map),
                    Key,
                    {{some, Value}, Timestamp}
                ),
                erlang:element(3, Map)};

        false ->
            Map
    end.

-file("src/lattice_maps/lww_map.gleam", 91).
?DOC(
    " Get the value for a key.\n"
    "\n"
    " Returns `Ok(value)` if the key exists and is not tombstoned.\n"
    " Returns `Error(Nil)` if the key is missing or has been removed.\n"
).
-spec get(l_w_w_map(), binary()) -> {ok, binary()} | {error, nil}.
get(Map, Key) ->
    case gleam_stdlib:map_get(erlang:element(2, Map), Key) of
        {ok, {{some, Value}, _}} ->
            {ok, Value};

        _ ->
            {error, nil}
    end.

-file("src/lattice_maps/lww_map.gleam", 106).
?DOC(
    " Remove a key at the given timestamp by inserting a tombstone.\n"
    "\n"
    " If the key already has an entry with an equal or higher timestamp, the\n"
    " remove is rejected and the existing entry wins.\n"
    "\n"
    " Note: This operation creates a tombstone. Use `tombstone_count` to monitor\n"
    " growth and `prune` with a stable timestamp to remove tombstones once all\n"
    " replicas have observed them.\n"
).
-spec remove(l_w_w_map(), binary(), integer()) -> l_w_w_map().
remove(Map, Key, Timestamp) ->
    Should_remove = case gleam_stdlib:map_get(erlang:element(2, Map), Key) of
        {error, _} ->
            true;

        {ok, {_, Existing_ts}} ->
            Timestamp > Existing_ts
    end,
    case Should_remove of
        true ->
            {l_w_w_map,
                gleam@dict:insert(
                    erlang:element(2, Map),
                    Key,
                    {none, Timestamp}
                ),
                erlang:element(3, Map)};

        false ->
            Map
    end.

-file("src/lattice_maps/lww_map.gleam", 121).
?DOC(
    " Return all active (non-tombstoned) keys in the map.\n"
    "\n"
    " Order is not guaranteed.\n"
).
-spec keys(l_w_w_map()) -> list(binary()).
keys(Map) ->
    gleam@dict:fold(
        erlang:element(2, Map),
        [],
        fun(Acc, Key, Entry) -> case Entry of
                {{some, _}, _} ->
                    [Key | Acc];

                {none, _} ->
                    Acc
            end end
    ).

-file("src/lattice_maps/lww_map.gleam", 143).
?DOC(
    " Return the number of tombstoned (removed) entries in the map.\n"
    "\n"
    " Useful for monitoring tombstone growth and deciding when to call `prune`.\n"
    "\n"
    " ## Examples\n"
    "\n"
    " ```gleam\n"
    " let m = lww_map.new()\n"
    "   |> lww_map.set(\"a\", \"1\", 1)\n"
    "   |> lww_map.set(\"b\", \"2\", 1)\n"
    "   |> lww_map.remove(\"a\", 10)\n"
    " lww_map.tombstone_count(m)  // -> 1\n"
    " ```\n"
).
-spec tombstone_count(l_w_w_map()) -> integer().
tombstone_count(Map) ->
    gleam@dict:fold(
        erlang:element(2, Map),
        0,
        fun(Acc, _, Entry) -> case Entry of
                {none, _} ->
                    Acc + 1;

                {{some, _}, _} ->
                    Acc
            end end
    ).

-file("src/lattice_maps/lww_map.gleam", 178).
?DOC(
    " Prune tombstones at or below the given stable timestamp.\n"
    "\n"
    " Removes all tombstone entries (keys with `None` value) whose timestamp is\n"
    " less than or equal to `stable_timestamp`. Active entries are never removed.\n"
    " The `pruned_timestamp` is updated monotonically to `max(current, stable_timestamp)`.\n"
    "\n"
    " After pruning, `merge` automatically detects zombie entries — entries from\n"
    " remote replicas whose timestamps fall at or below the pruned threshold are\n"
    " discarded, preventing deleted keys from resurrecting.\n"
    "\n"
    " The `stable_timestamp` should ideally represent a point that all replicas\n"
    " have synced past. Using a conservative (older) value is always safe; using\n"
    " a value ahead of some replica means that replica's stale writes will be\n"
    " silently dropped on merge (which is the correct behavior for pruned history).\n"
    "\n"
    " ## Examples\n"
    "\n"
    " ```gleam\n"
    " let m = lww_map.new()\n"
    "   |> lww_map.set(\"a\", \"alive\", 1)\n"
    "   |> lww_map.remove(\"b\", 5)\n"
    "   |> lww_map.remove(\"c\", 15)\n"
    " let pruned = lww_map.prune(m, 10)\n"
    " lww_map.tombstone_count(pruned)  // -> 1 (only \"c\" at ts=15 remains)\n"
    " lww_map.get(pruned, \"a\")         // -> Ok(\"alive\")\n"
    " ```\n"
).
-spec prune(l_w_w_map(), integer()) -> l_w_w_map().
prune(Map, Stable_timestamp) ->
    New_pruned = gleam@int:max(erlang:element(3, Map), Stable_timestamp),
    {l_w_w_map,
        gleam@dict:filter(erlang:element(2, Map), fun(_, Entry) -> case Entry of
                    {none, Ts} ->
                        Ts > New_pruned;

                    {{some, _}, _} ->
                        true
                end end),
        New_pruned}.

-file("src/lattice_maps/lww_map.gleam", 194).
?DOC(
    " Return all active (non-tombstoned) values in the map.\n"
    "\n"
    " Order is not guaranteed and does not correspond to the order of `keys`.\n"
).
-spec values(l_w_w_map()) -> list(binary()).
values(Map) ->
    gleam@dict:fold(
        erlang:element(2, Map),
        [],
        fun(Acc, _, Entry) -> case Entry of
                {{some, Value}, _} ->
                    [Value | Acc];

                {none, _} ->
                    Acc
            end end
    ).

-file("src/lattice_maps/lww_map.gleam", 236).
?DOC(
    " An entry only present on one side is a zombie if its timestamp is at or\n"
    " below the other side's pruned_timestamp — meaning the other side already\n"
    " processed and garbage-collected the tombstone that would have suppressed it.\n"
).
-spec keep_if_not_zombie({gleam@option:option(binary()), integer()}, integer()) -> {ok,
        {gleam@option:option(binary()), integer()}} |
    {error, nil}.
keep_if_not_zombie(Entry, Other_pruned) ->
    {_, Ts} = Entry,
    case Ts =< Other_pruned of
        true ->
            {error, nil};

        false ->
            {ok, Entry}
    end.

-file("src/lattice_maps/lww_map.gleam", 247).
-spec choose_winner(
    {gleam@option:option(binary()), integer()},
    {gleam@option:option(binary()), integer()}
) -> {gleam@option:option(binary()), integer()}.
choose_winner(A, B) ->
    {_, Ts_a} = A,
    {_, Ts_b} = B,
    case Ts_a > Ts_b of
        true ->
            A;

        false ->
            case Ts_b > Ts_a of
                true ->
                    B;

                false ->
                    case {A, B} of
                        {{none, _}, {{some, _}, _}} ->
                            A;

                        {{{some, _}, _}, {none, _}} ->
                            B;

                        {{none, _}, {none, _}} ->
                            A;

                        {{{some, A_val}, _}, {{some, B_val}, _}} ->
                            case gleam@string:compare(A_val, B_val) of
                                gt ->
                                    A;

                                eq ->
                                    A;

                                lt ->
                                    B
                            end
                    end
            end
    end.

-file("src/lattice_maps/lww_map.gleam", 212).
?DOC(
    " Merge two LWW-Maps by resolving each key using the highest timestamp.\n"
    "\n"
    " Tombstones participate in merge: if a tombstone has a higher timestamp\n"
    " than the active entry for a key, the key remains removed after merging.\n"
    " On equal timestamps, a deterministic tie-break is used:\n"
    " tombstones win over active values, otherwise the lexicographically greater\n"
    " value wins.\n"
    "\n"
    " Merge is commutative, associative, and idempotent (a valid CRDT join).\n"
).
-spec merge(l_w_w_map(), l_w_w_map()) -> l_w_w_map().
merge(A, B) ->
    Merged_pruned = gleam@int:max(erlang:element(3, A), erlang:element(3, B)),
    All_keys = gleam@list:unique(
        lists:append(
            maps:keys(erlang:element(2, A)),
            maps:keys(erlang:element(2, B))
        )
    ),
    Merged = gleam@list:fold(
        All_keys,
        maps:new(),
        fun(Acc, Key) ->
            Entry = case {gleam_stdlib:map_get(erlang:element(2, A), Key),
                gleam_stdlib:map_get(erlang:element(2, B), Key)} of
                {{ok, Ea}, {ok, Eb}} ->
                    {ok, choose_winner(Ea, Eb)};

                {{ok, Ea@1}, {error, _}} ->
                    keep_if_not_zombie(Ea@1, erlang:element(3, B));

                {{error, _}, {ok, Eb@1}} ->
                    keep_if_not_zombie(Eb@1, erlang:element(3, A));

                {{error, _}, {error, _}} ->
                    erlang:error(#{gleam_error => panic,
                            message => <<"unreachable: key in all_keys but not in either dict"/utf8>>,
                            file => <<?FILEPATH/utf8>>,
                            module => <<"lattice_maps/lww_map"/utf8>>,
                            function => <<"merge"/utf8>>,
                            line => 223})
            end,
            case Entry of
                {ok, Winner} ->
                    gleam@dict:insert(Acc, Key, Winner);

                {error, nil} ->
                    Acc
            end
        end
    ),
    {l_w_w_map, Merged, Merged_pruned}.

-file("src/lattice_maps/lww_map.gleam", 284).
?DOC(
    " Encode a `LWWMap` as a self-describing JSON value.\n"
    "\n"
    " Entries are encoded as a JSON array where each element has `key`, `value`\n"
    " (nullable string for tombstones), and `timestamp` fields. The\n"
    " `pruned_timestamp` field records the highest stable timestamp passed to\n"
    " `prune`, enabling zombie detection after deserialization.\n"
    "\n"
    " Format: `{\"type\": \"lww_map\", \"v\": 2, \"state\": {\"entries\": [...], \"pruned_timestamp\": N}}`\n"
    "\n"
    " The encoded value can be restored with `from_json`.\n"
).
-spec to_json(l_w_w_map()) -> gleam@json:json().
to_json(Map) ->
    Entries_json = gleam@json:array(
        maps:to_list(erlang:element(2, Map)),
        fun(Pair) ->
            {Key, {Opt_value, Timestamp}} = Pair,
            gleam@json:object(
                [{<<"key"/utf8>>, gleam@json:string(Key)},
                    {<<"value"/utf8>>, case Opt_value of
                            {some, V} ->
                                gleam@json:string(V);

                            none ->
                                gleam@json:null()
                        end},
                    {<<"timestamp"/utf8>>, gleam@json:int(Timestamp)}]
            )
        end
    ),
    gleam@json:object(
        [{<<"type"/utf8>>, gleam@json:string(<<"lww_map"/utf8>>)},
            {<<"v"/utf8>>, gleam@json:int(2)},
            {<<"state"/utf8>>,
                gleam@json:object(
                    [{<<"entries"/utf8>>, Entries_json},
                        {<<"pruned_timestamp"/utf8>>,
                            gleam@json:int(erlang:element(3, Map))}]
                )}]
    ).

-file("src/lattice_maps/lww_map.gleam", 315).
?DOC(
    " Decode a `LWWMap` from a JSON string produced by `to_json`.\n"
    "\n"
    " Supports both v1 (no `pruned_timestamp`, defaults to 0) and v2 formats.\n"
    " Returns `Error` if the string is not valid JSON or does not match the\n"
    " expected format.\n"
).
-spec from_json(binary()) -> {ok, l_w_w_map()} |
    {error, gleam@json:decode_error()}.
from_json(Json_string) ->
    Entry_decoder = begin
        gleam@dynamic@decode:field(
            <<"key"/utf8>>,
            {decoder, fun gleam@dynamic@decode:decode_string/1},
            fun(Key) ->
                gleam@dynamic@decode:field(
                    <<"value"/utf8>>,
                    gleam@dynamic@decode:optional(
                        {decoder, fun gleam@dynamic@decode:decode_string/1}
                    ),
                    fun(Opt_value) ->
                        gleam@dynamic@decode:field(
                            <<"timestamp"/utf8>>,
                            {decoder, fun gleam@dynamic@decode:decode_int/1},
                            fun(Timestamp) ->
                                gleam@dynamic@decode:success(
                                    {Key, {Opt_value, Timestamp}}
                                )
                            end
                        )
                    end
                )
            end
        )
    end,
    V1_state_decoder = begin
        gleam@dynamic@decode:field(
            <<"state"/utf8>>,
            begin
                gleam@dynamic@decode:field(
                    <<"entries"/utf8>>,
                    gleam@dynamic@decode:list(Entry_decoder),
                    fun(Entries_list) ->
                        gleam@dynamic@decode:success(
                            {l_w_w_map, maps:from_list(Entries_list), 0}
                        )
                    end
                )
            end,
            fun(State) -> gleam@dynamic@decode:success(State) end
        )
    end,
    V2_state_decoder = begin
        gleam@dynamic@decode:field(
            <<"state"/utf8>>,
            begin
                gleam@dynamic@decode:field(
                    <<"entries"/utf8>>,
                    gleam@dynamic@decode:list(Entry_decoder),
                    fun(Entries_list@1) ->
                        gleam@dynamic@decode:field(
                            <<"pruned_timestamp"/utf8>>,
                            {decoder, fun gleam@dynamic@decode:decode_int/1},
                            fun(Pruned_ts) ->
                                gleam@dynamic@decode:success(
                                    {l_w_w_map,
                                        maps:from_list(Entries_list@1),
                                        Pruned_ts}
                                )
                            end
                        )
                    end
                )
            end,
            fun(State@1) -> gleam@dynamic@decode:success(State@1) end
        )
    end,
    Envelope_decoder = begin
        gleam@dynamic@decode:field(
            <<"type"/utf8>>,
            {decoder, fun gleam@dynamic@decode:decode_string/1},
            fun(Type_tag) ->
                gleam@dynamic@decode:field(
                    <<"v"/utf8>>,
                    {decoder, fun gleam@dynamic@decode:decode_int/1},
                    fun(Version) ->
                        gleam@dynamic@decode:success({Type_tag, Version})
                    end
                )
            end
        )
    end,
    case gleam@json:parse(Json_string, Envelope_decoder) of
        {error, E} ->
            {error, E};

        {ok, {Type_tag@1, Version@1}} ->
            case {Type_tag@1 =:= <<"lww_map"/utf8>>, Version@1} of
                {true, 1} ->
                    gleam@json:parse(Json_string, V1_state_decoder);

                {true, 2} ->
                    gleam@json:parse(Json_string, V2_state_decoder);

                {_, _} ->
                    {error,
                        {unable_to_decode,
                            [{decode_error,
                                    <<"type=lww_map and v=1 or v=2"/utf8>>,
                                    <<<<Type_tag@1/binary, " v="/utf8>>/binary,
                                        (erlang:integer_to_binary(Version@1))/binary>>,
                                    []}]}}
            end
    end.