Packages
Last-writer-wins and observed-remove maps for Gleam — conflict-free replicated map types with generic value support
Current section
Files
Jump to
Current section
Files
src/lattice_maps@or_map.erl
-module(lattice_maps@or_map).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/lattice_maps/or_map.gleam").
-export([new/2, update/3, get/2, remove_with_delta/2, remove/2, keys/1, values/1, merge/2, prune/2, internal_value_count/1, to_json/1, from_json/1, empty_delta/1, update_with_delta/3, apply_delta/2, merge_deltas/2, delta_to_json/1, delta_from_json/1]).
-export_type([o_r_map/0, o_r_map_delta/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(
" An observed-remove map (OR-Map) CRDT.\n"
"\n"
" Keys are tracked using an OR-Set with add-wins semantics: concurrent update\n"
" and remove of the same key resolves in favor of the update. Each value is\n"
" itself a CRDT (specified by `CrdtSpec` at construction), enabling nested\n"
" convergent data structures.\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" import lattice_maps/crdt\n"
" import lattice_core/replica_id\n"
" import lattice_counters/g_counter\n"
" import lattice_maps/or_map\n"
"\n"
" let map = or_map.new(replica_id.new(\"node-a\"), crdt.GCounterSpec)\n"
" |> or_map.update(\"score\", fn(c) {\n"
" let assert crdt.CrdtGCounter(gc) = c\n"
" crdt.CrdtGCounter(g_counter.increment(gc, 10))\n"
" })\n"
" ```\n"
).
-opaque o_r_map() :: {o_r_map,
lattice_core@replica_id:replica_id(),
lattice_maps@crdt:crdt_spec(),
lattice_sets@or_set:o_r_set(binary()),
gleam@dict:dict(binary(), lattice_maps@crdt:crdt()),
gleam@dict:dict(binary(), lattice_core@version_vector:version_vector())}.
-opaque o_r_map_delta() :: {o_r_map_delta,
lattice_core@replica_id:replica_id(),
lattice_maps@crdt:crdt_spec(),
lattice_sets@or_set:o_r_set(binary()),
gleam@dict:dict(binary(), lattice_maps@crdt:crdt()),
gleam@dict:dict(binary(), lattice_core@version_vector:version_vector())}.
-file("src/lattice_maps/or_map.gleam", 54).
-spec spec_to_string(lattice_maps@crdt:crdt_spec()) -> binary().
spec_to_string(Spec) ->
case Spec of
g_counter_spec ->
<<"g_counter"/utf8>>;
pn_counter_spec ->
<<"pn_counter"/utf8>>;
lww_register_spec ->
<<"lww_register"/utf8>>;
mv_register_spec ->
<<"mv_register"/utf8>>;
g_set_spec ->
<<"g_set"/utf8>>;
two_p_set_spec ->
<<"two_p_set"/utf8>>;
or_set_spec ->
<<"or_set"/utf8>>
end.
-file("src/lattice_maps/or_map.gleam", 66).
-spec string_to_spec(binary()) -> {ok, lattice_maps@crdt:crdt_spec()} |
{error, nil}.
string_to_spec(S) ->
case S of
<<"g_counter"/utf8>> ->
{ok, g_counter_spec};
<<"pn_counter"/utf8>> ->
{ok, pn_counter_spec};
<<"lww_register"/utf8>> ->
{ok, lww_register_spec};
<<"mv_register"/utf8>> ->
{ok, mv_register_spec};
<<"g_set"/utf8>> ->
{ok, g_set_spec};
<<"two_p_set"/utf8>> ->
{ok, two_p_set_spec};
<<"or_set"/utf8>> ->
{ok, or_set_spec};
_ ->
{error, nil}
end.
-file("src/lattice_maps/or_map.gleam", 83).
?DOC(
" Create a new empty OR-Map for the given replica with the specified CRDT type.\n"
"\n"
" The `crdt_spec` determines what type of CRDT is auto-created when `update`\n"
" is called on a key that does not yet exist in the map.\n"
).
-spec new(lattice_core@replica_id:replica_id(), lattice_maps@crdt:crdt_spec()) -> o_r_map().
new(Replica_id, Crdt_spec) ->
{o_r_map,
Replica_id,
Crdt_spec,
lattice_sets@or_set:new(Replica_id),
maps:new(),
maps:new()}.
-file("src/lattice_maps/or_map.gleam", 610).
-spec put_value(o_r_map(), binary(), lattice_maps@crdt:crdt()) -> {o_r_map(),
lattice_sets@or_set:o_r_set(binary())}.
put_value(Map, Key, Value) ->
{Updated_key_set, Key_set_delta} = lattice_sets@or_set:add_with_delta(
erlang:element(4, Map),
Key
),
{{o_r_map,
erlang:element(2, Map),
erlang:element(3, Map),
Updated_key_set,
gleam@dict:insert(erlang:element(5, Map), Key, Value),
gleam@dict:delete(erlang:element(6, Map), Key)},
Key_set_delta}.
-file("src/lattice_maps/or_map.gleam", 475).
-spec matches_spec(lattice_maps@crdt:crdt(), lattice_maps@crdt:crdt_spec()) -> boolean().
matches_spec(Value, Spec) ->
case {Value, Spec} of
{{crdt_g_counter, _}, g_counter_spec} ->
true;
{{crdt_pn_counter, _}, pn_counter_spec} ->
true;
{{crdt_lww_register, _}, lww_register_spec} ->
true;
{{crdt_mv_register, _}, mv_register_spec} ->
true;
{{crdt_g_set, _}, g_set_spec} ->
true;
{{crdt_two_p_set, _}, two_p_set_spec} ->
true;
{{crdt_or_set, _}, or_set_spec} ->
true;
{_, _} ->
false
end.
-file("src/lattice_maps/or_map.gleam", 603).
-spec current_value(o_r_map(), binary()) -> lattice_maps@crdt:crdt().
current_value(Map, Key) ->
case {lattice_sets@or_set:contains(erlang:element(4, Map), Key),
gleam_stdlib:map_get(erlang:element(5, Map), Key)} of
{true, {ok, Crdt_val}} ->
Crdt_val;
{_, _} ->
lattice_maps@crdt:default_crdt(
erlang:element(3, Map),
erlang:element(2, Map)
)
end.
-file("src/lattice_maps/or_map.gleam", 101).
?DOC(
" Apply a function to the CRDT value at `key`, auto-creating it if absent.\n"
"\n"
" If the key does not exist, a default value is created from `crdt_spec`\n"
" and passed to `f`. The key is added to the OR-Set, marking it active.\n"
" The return value of `f` replaces (or sets) the value for that key.\n"
"\n"
" See `update_with_delta` for the delta-state variant that also returns a\n"
" small payload suitable for incremental sync (e.g. over websockets).\n"
).
-spec update(
o_r_map(),
binary(),
fun((lattice_maps@crdt:crdt()) -> lattice_maps@crdt:crdt())
) -> o_r_map().
update(Map, Key, F) ->
Current = current_value(Map, Key),
New_value = F(Current),
Safe_value = case matches_spec(New_value, erlang:element(3, Map)) of
true ->
New_value;
false ->
Current
end,
{Updated, _} = put_value(Map, Key, Safe_value),
Updated.
-file("src/lattice_maps/or_map.gleam", 117).
?DOC(
" Get the CRDT value at `key`.\n"
"\n"
" Returns `Ok(crdt)` if the key is active in the OR-Set.\n"
" Returns `Error(Nil)` if the key has never been added, or has been removed\n"
" and not re-added.\n"
).
-spec get(o_r_map(), binary()) -> {ok, lattice_maps@crdt:crdt()} | {error, nil}.
get(Map, Key) ->
case lattice_sets@or_set:contains(erlang:element(4, Map), Key) of
true ->
case gleam_stdlib:map_get(erlang:element(5, Map), Key) of
{ok, Val} ->
{ok, Val};
{error, _} ->
{error, nil}
end;
false ->
{error, nil}
end.
-file("src/lattice_maps/or_map.gleam", 631).
?DOC(
" Remove a key and return both the new map and a delta capturing the\n"
" remove.\n"
"\n"
" The delta carries the OR-Set tombstones from `or_set.remove_with_delta`\n"
" and the recorded causal bound, which together let any remote replica\n"
" retract the key while preserving add-wins semantics for concurrent adds.\n"
).
-spec remove_with_delta(o_r_map(), binary()) -> {o_r_map(), o_r_map_delta()}.
remove_with_delta(Map, Key) ->
{Updated_key_set, Key_set_delta} = lattice_sets@or_set:remove_with_delta(
erlang:element(4, Map),
Key
),
{_, Bound} = lattice_sets@or_set:remove_with_bound(
erlang:element(4, Map),
Key
),
{Updated_bounds, Bounds_delta} = case lattice_core@version_vector:is_empty(
Bound
) of
true ->
{erlang:element(6, Map), maps:new()};
false ->
{gleam@dict:insert(erlang:element(6, Map), Key, Bound),
maps:from_list([{Key, Bound}])}
end,
Updated = {o_r_map,
erlang:element(2, Map),
erlang:element(3, Map),
Updated_key_set,
erlang:element(5, Map),
Updated_bounds},
Delta = {o_r_map_delta,
erlang:element(2, Map),
erlang:element(3, Map),
Key_set_delta,
maps:new(),
Bounds_delta},
{Updated, Delta}.
-file("src/lattice_maps/or_map.gleam", 139).
?DOC(
" Remove a key from the OR-Map.\n"
"\n"
" Removes the key from the OR-Set (marking it inactive). The underlying\n"
" CRDT value is retained until prune determines the removal is causally\n"
" stable. A causal bound is recorded so prune can later decide when it is\n"
" safe to discard the value.\n"
"\n"
" State-only convenience wrapper around `remove_with_delta` — the produced\n"
" delta is discarded. See `remove_with_delta` for the delta-state variant\n"
" that also returns a small payload suitable for incremental sync (e.g.\n"
" over websockets).\n"
).
-spec remove(o_r_map(), binary()) -> o_r_map().
remove(Map, Key) ->
{Updated, _} = remove_with_delta(Map, Key),
Updated.
-file("src/lattice_maps/or_map.gleam", 147).
?DOC(
" Return the list of all active keys (those present in the OR-Set).\n"
"\n"
" Order is not guaranteed.\n"
).
-spec keys(o_r_map()) -> list(binary()).
keys(Map) ->
gleam@set:to_list(lattice_sets@or_set:value(erlang:element(4, Map))).
-file("src/lattice_maps/or_map.gleam", 154).
?DOC(
" Return the CRDT values for all active keys.\n"
"\n"
" Order is not guaranteed and does not correspond to the order of `keys`.\n"
).
-spec values(o_r_map()) -> list(lattice_maps@crdt:crdt()).
values(Map) ->
Active_keys = lattice_sets@or_set:value(erlang:element(4, Map)),
gleam@dict:fold(
erlang:element(5, Map),
[],
fun(Acc, Key, Val) -> case gleam@set:contains(Active_keys, Key) of
true ->
[Val | Acc];
false ->
Acc
end end
).
-file("src/lattice_maps/or_map.gleam", 488).
-spec valid_value(o_r_map(), binary()) -> {ok, lattice_maps@crdt:crdt()} |
{error, nil}.
valid_value(Map, Key) ->
case gleam_stdlib:map_get(erlang:element(5, Map), Key) of
{ok, Value} ->
case matches_spec(Value, erlang:element(3, Map)) of
true ->
{ok, Value};
false ->
{ok,
lattice_maps@crdt:default_crdt(
erlang:element(3, Map),
erlang:element(2, Map)
)}
end;
{error, _} ->
{error, nil}
end.
-file("src/lattice_maps/or_map.gleam", 173).
?DOC(
" Merge two OR-Maps.\n"
"\n"
" The OR-Set key trackers are merged with add-wins semantics: if a key was\n"
" concurrently updated on one replica and removed on another, the key\n"
" survives in the merged result. CRDT values are merged per-key using\n"
" `crdt.merge` for type-specific convergence.\n"
"\n"
" Returns `Error(TypeMismatch(...))` if the two maps have different\n"
" `crdt_spec` values (e.g., one holds counters and the other holds sets).\n"
).
-spec merge(o_r_map(), o_r_map()) -> {ok, o_r_map()} |
{error, lattice_maps@crdt:merge_error()}.
merge(A, B) ->
case erlang:element(3, A) =:= erlang:element(3, B) of
false ->
{error,
{type_mismatch,
spec_to_string(erlang:element(3, A)),
spec_to_string(erlang:element(3, B))}};
true ->
Merged_key_set = lattice_sets@or_set:merge(
erlang:element(4, A),
erlang:element(4, B)
),
Active_keys = lattice_sets@or_set:value(Merged_key_set),
All_value_keys = gleam@set:to_list(
gleam@set:union(
gleam@set:from_list(maps:keys(erlang:element(5, A))),
gleam@set:from_list(maps:keys(erlang:element(5, B)))
)
),
Merged_values = gleam@list:fold(
All_value_keys,
maps:new(),
fun(Acc, Key) ->
Merged_crdt = case {valid_value(A, Key),
valid_value(B, Key)} of
{{ok, Ca}, {ok, Cb}} ->
case lattice_maps@crdt:merge(Ca, Cb) of
{ok, Merged} ->
Merged;
{error, _} ->
lattice_maps@crdt:default_crdt(
erlang:element(3, A),
erlang:element(2, A)
)
end;
{{ok, Ca@1}, {error, _}} ->
Ca@1;
{{error, _}, {ok, Cb@1}} ->
Cb@1;
{{error, _}, {error, _}} ->
erlang:error(#{gleam_error => panic,
message => <<"unreachable: key must exist in at least one map"/utf8>>,
file => <<?FILEPATH/utf8>>,
module => <<"lattice_maps/or_map"/utf8>>,
function => <<"merge"/utf8>>,
line => 199})
end,
gleam@dict:insert(Acc, Key, Merged_crdt)
end
),
All_bound_keys = gleam@set:to_list(
gleam@set:union(
gleam@set:from_list(maps:keys(erlang:element(6, A))),
gleam@set:from_list(maps:keys(erlang:element(6, B)))
)
),
Merged_bounds = gleam@list:fold(
All_bound_keys,
maps:new(),
fun(Acc@1, Key@1) ->
case gleam@set:contains(Active_keys, Key@1) of
true ->
Acc@1;
false ->
case {gleam_stdlib:map_get(
erlang:element(6, A),
Key@1
),
gleam_stdlib:map_get(
erlang:element(6, B),
Key@1
)} of
{{ok, Ba}, {ok, Bb}} ->
gleam@dict:insert(
Acc@1,
Key@1,
lattice_core@version_vector:merge(
Ba,
Bb
)
);
{{ok, Ba@1}, {error, _}} ->
gleam@dict:insert(Acc@1, Key@1, Ba@1);
{{error, _}, {ok, Bb@1}} ->
gleam@dict:insert(Acc@1, Key@1, Bb@1);
{{error, _}, {error, _}} ->
Acc@1
end
end
end
),
{ok,
{o_r_map,
erlang:element(2, A),
erlang:element(3, A),
Merged_key_set,
Merged_values,
Merged_bounds}}
end.
-file("src/lattice_maps/or_map.gleam", 251).
?DOC(
" Prune tombstones for keys and compact removed values whose removal is\n"
" causally stable.\n"
"\n"
" Delegates to `or_set.prune` to remove tombstones from the internal key\n"
" tracker. Then, for each removed key that has a recorded causal bound, if\n"
" the pruned version vector dominates that bound, the key's CRDT value is\n"
" discarded (the removal is stable and no concurrent re-add can reference\n"
" the old value).\n"
"\n"
" Only call this with a version vector representing events that have been\n"
" seen by all replicas (causally stable), otherwise zombie updates might be\n"
" incorrectly ignored.\n"
).
-spec prune(o_r_map(), lattice_core@version_vector:version_vector()) -> o_r_map().
prune(Map, Stable_vv) ->
Pruned_key_set = lattice_sets@or_set:prune(
erlang:element(4, Map),
Stable_vv
),
Pruned_vv = lattice_sets@or_set:pruned_vv(Pruned_key_set),
Active_keys = lattice_sets@or_set:value(Pruned_key_set),
{Compacted_values, Compacted_bounds} = gleam@dict:fold(
erlang:element(5, Map),
{maps:new(), erlang:element(6, Map)},
fun(Acc, Key, Val) ->
{Vals, Bounds} = Acc,
case gleam@set:contains(Active_keys, Key) of
true ->
{gleam@dict:insert(Vals, Key, Val), Bounds};
false ->
case gleam_stdlib:map_get(erlang:element(6, Map), Key) of
{ok, Bound} ->
case lattice_core@version_vector:dominates(
Pruned_vv,
Bound
) of
true ->
{Vals, gleam@dict:delete(Bounds, Key)};
false ->
{gleam@dict:insert(Vals, Key, Val), Bounds}
end;
{error, _} ->
{gleam@dict:insert(Vals, Key, Val), Bounds}
end
end
end
),
{o_r_map,
erlang:element(2, Map),
erlang:element(3, Map),
Pruned_key_set,
Compacted_values,
Compacted_bounds}.
-file("src/lattice_maps/or_map.gleam", 286).
?DOC(false).
-spec internal_value_count(o_r_map()) -> integer().
internal_value_count(Map) ->
maps:size(erlang:element(5, Map)).
-file("src/lattice_maps/or_map.gleam", 298).
?DOC(
" Encode an `ORMap` as a self-describing JSON value.\n"
"\n"
" The nested OR-Set (`key_set`) and CRDT values are double-encoded as JSON\n"
" strings so they can be decoded using the existing `from_json` APIs.\n"
"\n"
" Format: `{\"type\": \"or_map\", \"v\": 2, \"state\": {\"replica_id\": \"...\", \"crdt_spec\": \"...\", \"key_set\": \"...\", \"values\": [...], \"remove_bounds\": {...}}}`\n"
"\n"
" The encoded value can be restored with `from_json`.\n"
).
-spec to_json(o_r_map()) -> gleam@json:json().
to_json(Map) ->
{o_r_map, Rid, Crdt_spec, Key_set, Values, Remove_bounds} = Map,
Values_json = gleam@json:array(
maps:to_list(Values),
fun(Pair) ->
{Key, Crdt_val} = Pair,
gleam@json:object(
[{<<"key"/utf8>>, gleam@json:string(Key)},
{<<"crdt"/utf8>>,
gleam@json:string(
gleam@json:to_string(
lattice_maps@crdt:to_json(Crdt_val)
)
)}]
)
end
),
Bounds_json = gleam@json:dict(
Remove_bounds,
fun(K) -> K end,
fun(Vv) -> lattice_core@version_vector:to_json(Vv) end
),
gleam@json:object(
[{<<"type"/utf8>>, gleam@json:string(<<"or_map"/utf8>>)},
{<<"v"/utf8>>, gleam@json:int(2)},
{<<"state"/utf8>>,
gleam@json:object(
[{<<"replica_id"/utf8>>,
gleam@json:string(
lattice_core@replica_id:to_string(Rid)
)},
{<<"crdt_spec"/utf8>>,
gleam@json:string(spec_to_string(Crdt_spec))},
{<<"key_set"/utf8>>,
gleam@json:string(
gleam@json:to_string(
lattice_sets@or_set:to_json(Key_set)
)
)},
{<<"values"/utf8>>, Values_json},
{<<"remove_bounds"/utf8>>, Bounds_json}]
)}]
).
-file("src/lattice_maps/or_map.gleam", 499).
-spec crdt_name(lattice_maps@crdt:crdt()) -> binary().
crdt_name(Value) ->
case Value of
{crdt_g_counter, _} ->
<<"g_counter"/utf8>>;
{crdt_pn_counter, _} ->
<<"pn_counter"/utf8>>;
{crdt_lww_register, _} ->
<<"lww_register"/utf8>>;
{crdt_mv_register, _} ->
<<"mv_register"/utf8>>;
{crdt_g_set, _} ->
<<"g_set"/utf8>>;
{crdt_two_p_set, _} ->
<<"two_p_set"/utf8>>;
{crdt_or_set, _} ->
<<"or_set"/utf8>>;
{crdt_version_vector, _} ->
<<"version_vector"/utf8>>
end.
-file("src/lattice_maps/or_map.gleam", 416).
-spec decode_or_map_state(
binary(),
binary(),
binary(),
list({binary(), binary()}),
gleam@dict:dict(binary(), lattice_core@version_vector:version_vector())
) -> {ok, o_r_map()} | {error, gleam@json:decode_error()}.
decode_or_map_state(
Replica_id_str,
Crdt_spec_str,
Key_set_str,
Values_list,
Remove_bounds
) ->
case string_to_spec(Crdt_spec_str) of
{error, _} ->
{error,
{unable_to_decode,
[{decode_error,
<<"known CrdtSpec"/utf8>>,
Crdt_spec_str,
[<<"state"/utf8>>, <<"crdt_spec"/utf8>>]}]}};
{ok, Crdt_spec} ->
case lattice_sets@or_set:from_json(Key_set_str) of
{error, E} ->
{error, E};
{ok, Key_set} ->
Values_result = gleam@list:try_map(
Values_list,
fun(Pair) ->
{Key, Crdt_str} = Pair,
case lattice_maps@crdt:from_json(Crdt_str) of
{ok, C} ->
case matches_spec(C, Crdt_spec) of
true ->
{ok, {Key, C}};
false ->
{error,
{unable_to_decode,
[{decode_error,
spec_to_string(
Crdt_spec
),
crdt_name(C),
[<<"state"/utf8>>,
<<"values"/utf8>>]}]}}
end;
{error, E@1} ->
{error, E@1}
end
end
),
case Values_result of
{error, E@2} ->
{error, E@2};
{ok, Pairs} ->
{ok,
{o_r_map,
lattice_core@replica_id:new(Replica_id_str),
Crdt_spec,
Key_set,
maps:from_list(Pairs),
Remove_bounds}}
end
end
end.
-file("src/lattice_maps/or_map.gleam", 334).
?DOC(
" Decode an `ORMap` from a JSON string produced by `to_json`.\n"
"\n"
" Supports both v1 (no remove_bounds) and v2 (with remove_bounds) formats.\n"
" v1 maps are decoded with empty remove_bounds, meaning no value compaction\n"
" is possible until new removes are performed.\n"
"\n"
" Returns `Error` if the string is not valid JSON, does not match the\n"
" expected format, or contains an unknown `crdt_spec` string.\n"
).
-spec from_json(binary()) -> {ok, o_r_map()} |
{error, gleam@json:decode_error()}.
from_json(Json_string) ->
Value_pair_decoder = begin
gleam@dynamic@decode:field(
<<"key"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Key) ->
gleam@dynamic@decode:field(
<<"crdt"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Crdt_str) ->
gleam@dynamic@decode:success({Key, Crdt_str})
end
)
end
)
end,
Bounds_decoder = gleam@dynamic@decode:dict(
{decoder, fun gleam@dynamic@decode:decode_string/1},
lattice_core@version_vector:decoder()
),
State_decoder = begin
gleam@dynamic@decode:field(
<<"state"/utf8>>,
begin
gleam@dynamic@decode:field(
<<"replica_id"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Replica_id_str) ->
gleam@dynamic@decode:field(
<<"crdt_spec"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Crdt_spec_str) ->
gleam@dynamic@decode:field(
<<"key_set"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Key_set_str) ->
gleam@dynamic@decode:field(
<<"values"/utf8>>,
gleam@dynamic@decode:list(
Value_pair_decoder
),
fun(Values_list) ->
gleam@dynamic@decode:optional_field(
<<"remove_bounds"/utf8>>,
maps:new(),
Bounds_decoder,
fun(Remove_bounds) ->
gleam@dynamic@decode:success(
{Replica_id_str,
Crdt_spec_str,
Key_set_str,
Values_list,
Remove_bounds}
)
end
)
end
)
end
)
end
)
end
)
end,
fun(State) -> gleam@dynamic@decode:success(State) 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 =:= <<"or_map"/utf8>> of
false ->
{error,
{unable_to_decode,
[{decode_error,
<<"type=or_map"/utf8>>,
Type_tag@1,
[]}]}};
true ->
case Version@1 of
1 ->
case gleam@json:parse(Json_string, State_decoder) of
{error, E@1} ->
{error, E@1};
{ok,
{Replica_id_str@1,
Crdt_spec_str@1,
Key_set_str@1,
Values_list@1,
Remove_bounds@1}} ->
decode_or_map_state(
Replica_id_str@1,
Crdt_spec_str@1,
Key_set_str@1,
Values_list@1,
Remove_bounds@1
)
end;
2 ->
case gleam@json:parse(Json_string, State_decoder) of
{error, E@1} ->
{error, E@1};
{ok,
{Replica_id_str@1,
Crdt_spec_str@1,
Key_set_str@1,
Values_list@1,
Remove_bounds@1}} ->
decode_or_map_state(
Replica_id_str@1,
Crdt_spec_str@1,
Key_set_str@1,
Values_list@1,
Remove_bounds@1
)
end;
_ ->
{error,
{unable_to_decode,
[{decode_error,
<<"v=1 or v=2"/utf8>>,
erlang:integer_to_binary(Version@1),
[<<"v"/utf8>>]}]}}
end
end
end.
-file("src/lattice_maps/or_map.gleam", 549).
?DOC(
" Return the empty/identity delta for a map.\n"
"\n"
" `apply_delta(m, empty_delta(m))` returns `m` unchanged. Useful as the\n"
" starting accumulator when folding multiple delta-producing operations.\n"
).
-spec empty_delta(o_r_map()) -> o_r_map_delta().
empty_delta(Map) ->
{o_r_map_delta,
erlang:element(2, Map),
erlang:element(3, Map),
lattice_sets@or_set:new(erlang:element(2, Map)),
maps:new(),
maps:new()}.
-file("src/lattice_maps/or_map.gleam", 575).
?DOC(
" Apply a function to the value at `key` and return both the new map and a\n"
" delta capturing the touched key.\n"
"\n"
" The callback `f` receives the current value (or a default if absent) and\n"
" returns the new value. The delta contains that per-key CRDT value, which is\n"
" sparse at the map level and safe to apply to any replica of the same spec.\n"
"\n"
" ```gleam\n"
" or_map.update_with_delta(map, \"score\", fn(c) {\n"
" let assert crdt.CrdtGCounter(gc) = c\n"
" crdt.CrdtGCounter(g_counter.increment(gc, 5))\n"
" })\n"
" ```\n"
"\n"
" Returns `Error(TypeMismatch(...))` if the returned value does not match the\n"
" map's `crdt_spec`.\n"
).
-spec update_with_delta(
o_r_map(),
binary(),
fun((lattice_maps@crdt:crdt()) -> lattice_maps@crdt:crdt())
) -> {ok, {o_r_map(), o_r_map_delta()}} |
{error, lattice_maps@crdt:merge_error()}.
update_with_delta(Map, Key, F) ->
Current = current_value(Map, Key),
New_value = F(Current),
case matches_spec(New_value, erlang:element(3, Map)) of
false ->
{error,
{type_mismatch,
spec_to_string(erlang:element(3, Map)),
crdt_name(New_value)}};
true ->
{Updated, Key_set_delta} = put_value(Map, Key, New_value),
Delta = {o_r_map_delta,
erlang:element(2, Map),
erlang:element(3, Map),
Key_set_delta,
maps:from_list([{Key, New_value}]),
maps:new()},
{ok, {Updated, Delta}}
end.
-file("src/lattice_maps/or_map.gleam", 706).
-spec delete_bounds_for(
gleam@dict:dict(binary(), lattice_core@version_vector:version_vector()),
list(binary())
) -> gleam@dict:dict(binary(), lattice_core@version_vector:version_vector()).
delete_bounds_for(Bounds, Keys) ->
gleam@list:fold(
Keys,
Bounds,
fun(Acc, Key) -> gleam@dict:delete(Acc, Key) end
).
-file("src/lattice_maps/or_map.gleam", 677).
-spec apply_delta_unchecked(o_r_map(), o_r_map_delta()) -> {ok, o_r_map()} |
{error, lattice_maps@crdt:merge_error()}.
apply_delta_unchecked(Map, Delta) ->
Merged_key_set = lattice_sets@or_set:merge(
erlang:element(4, Map),
erlang:element(4, Delta)
),
Merged_values = gleam@dict:combine(
erlang:element(5, Map),
erlang:element(5, Delta),
fun(Local, Incoming) -> case lattice_maps@crdt:merge(Local, Incoming) of
{ok, Merged} ->
Merged;
{error, _} ->
lattice_maps@crdt:default_crdt(
erlang:element(3, Map),
erlang:element(2, Map)
)
end end
),
Merged_bounds = begin
_pipe = gleam@dict:combine(
erlang:element(6, Map),
erlang:element(6, Delta),
fun lattice_core@version_vector:merge/2
),
delete_bounds_for(_pipe, maps:keys(erlang:element(5, Delta)))
end,
{ok,
{o_r_map,
erlang:element(2, Map),
erlang:element(3, Map),
Merged_key_set,
Merged_values,
Merged_bounds}}.
-file("src/lattice_maps/or_map.gleam", 663).
?DOC(
" Apply a delta to a map, producing a new map.\n"
"\n"
" Returns `Error(TypeMismatch(...))` if the delta and map have different\n"
" `crdt_spec` values. Otherwise reconstructs the delta as a sparse\n"
" sparse state, so only the keys carried by the delta need value/bound work.\n"
"\n"
" Idempotent and commutative: applying the same delta twice, or applying\n"
" deltas in any order, yields the same final state.\n"
).
-spec apply_delta(o_r_map(), o_r_map_delta()) -> {ok, o_r_map()} |
{error, lattice_maps@crdt:merge_error()}.
apply_delta(Map, Delta) ->
case erlang:element(3, Map) =:= erlang:element(3, Delta) of
false ->
{error,
{type_mismatch,
spec_to_string(erlang:element(3, Map)),
spec_to_string(erlang:element(3, Delta))}};
true ->
apply_delta_unchecked(Map, Delta)
end.
-file("src/lattice_maps/or_map.gleam", 722).
?DOC(
" Combine two deltas into a single delta.\n"
"\n"
" Useful for transport layers that batch unacked deltas before\n"
" transmission: instead of sending N small deltas, merge them once and\n"
" send the join. Equivalent under `apply_delta` to applying both deltas\n"
" individually in either order.\n"
"\n"
" Returns `Error(TypeMismatch(...))` if the two deltas have different\n"
" `crdt_spec` values.\n"
).
-spec merge_deltas(o_r_map_delta(), o_r_map_delta()) -> {ok, o_r_map_delta()} |
{error, lattice_maps@crdt:merge_error()}.
merge_deltas(A, B) ->
case erlang:element(3, A) =:= erlang:element(3, B) of
false ->
{error,
{type_mismatch,
spec_to_string(erlang:element(3, A)),
spec_to_string(erlang:element(3, B))}};
true ->
Merged_key_set = lattice_sets@or_set:merge(
erlang:element(4, A),
erlang:element(4, B)
),
Merged_values = gleam@dict:combine(
erlang:element(5, A),
erlang:element(5, B),
fun(Va, Vb) -> case lattice_maps@crdt:merge(Va, Vb) of
{ok, M} ->
M;
{error, _} ->
lattice_maps@crdt:default_delta(
erlang:element(3, A),
erlang:element(2, A)
)
end end
),
Merged_bounds = gleam@dict:combine(
erlang:element(6, A),
erlang:element(6, B),
fun lattice_core@version_vector:merge/2
),
{ok,
{o_r_map_delta,
erlang:element(2, A),
erlang:element(3, A),
Merged_key_set,
Merged_values,
Merged_bounds}}
end.
-file("src/lattice_maps/or_map.gleam", 762).
?DOC(
" Encode an `ORMapDelta` as a self-describing JSON value.\n"
"\n"
" Format mirrors `to_json` for ORMap but uses `\"type\": \"or_map_delta\"`.\n"
" Use `delta_from_json` to decode.\n"
).
-spec delta_to_json(o_r_map_delta()) -> gleam@json:json().
delta_to_json(Delta) ->
{o_r_map_delta, Rid, Crdt_spec, Key_set_delta, Value_deltas, Bounds_delta} = Delta,
Values_json = gleam@json:array(
maps:to_list(Value_deltas),
fun(Pair) ->
{Key, Crdt_val} = Pair,
gleam@json:object(
[{<<"key"/utf8>>, gleam@json:string(Key)},
{<<"crdt"/utf8>>,
gleam@json:string(
gleam@json:to_string(
lattice_maps@crdt:to_json(Crdt_val)
)
)}]
)
end
),
Bounds_json = gleam@json:dict(
Bounds_delta,
fun(K) -> K end,
fun(Vv) -> lattice_core@version_vector:to_json(Vv) end
),
gleam@json:object(
[{<<"type"/utf8>>, gleam@json:string(<<"or_map_delta"/utf8>>)},
{<<"v"/utf8>>, gleam@json:int(1)},
{<<"state"/utf8>>,
gleam@json:object(
[{<<"replica_id"/utf8>>,
gleam@json:string(
lattice_core@replica_id:to_string(Rid)
)},
{<<"crdt_spec"/utf8>>,
gleam@json:string(spec_to_string(Crdt_spec))},
{<<"key_set_delta"/utf8>>,
gleam@json:string(
gleam@json:to_string(
lattice_sets@or_set:to_json(Key_set_delta)
)
)},
{<<"value_deltas"/utf8>>, Values_json},
{<<"remove_bounds_delta"/utf8>>, Bounds_json}]
)}]
).
-file("src/lattice_maps/or_map.gleam", 876).
-spec decode_or_map_delta_state(
binary(),
binary(),
binary(),
list({binary(), binary()}),
gleam@dict:dict(binary(), lattice_core@version_vector:version_vector())
) -> {ok, o_r_map_delta()} | {error, gleam@json:decode_error()}.
decode_or_map_delta_state(
Replica_id_str,
Crdt_spec_str,
Key_set_str,
Values_list,
Bounds
) ->
case string_to_spec(Crdt_spec_str) of
{error, _} ->
{error,
{unable_to_decode,
[{decode_error,
<<"known CrdtSpec"/utf8>>,
Crdt_spec_str,
[<<"state"/utf8>>, <<"crdt_spec"/utf8>>]}]}};
{ok, Crdt_spec} ->
case lattice_sets@or_set:from_json(Key_set_str) of
{error, E} ->
{error, E};
{ok, Key_set_delta} ->
Values_result = gleam@list:try_map(
Values_list,
fun(Pair) ->
{Key, Crdt_str} = Pair,
case lattice_maps@crdt:from_json(Crdt_str) of
{ok, C} ->
case matches_spec(C, Crdt_spec) of
true ->
{ok, {Key, C}};
false ->
{error,
{unable_to_decode,
[{decode_error,
spec_to_string(
Crdt_spec
),
crdt_name(C),
[<<"state"/utf8>>,
<<"value_deltas"/utf8>>]}]}}
end;
{error, E@1} ->
{error, E@1}
end
end
),
case Values_result of
{error, E@2} ->
{error, E@2};
{ok, Pairs} ->
{ok,
{o_r_map_delta,
lattice_core@replica_id:new(Replica_id_str),
Crdt_spec,
Key_set_delta,
maps:from_list(Pairs),
Bounds}}
end
end
end.
-file("src/lattice_maps/or_map.gleam", 795).
?DOC(" Decode an `ORMapDelta` from a JSON string produced by `delta_to_json`.\n").
-spec delta_from_json(binary()) -> {ok, o_r_map_delta()} |
{error, gleam@json:decode_error()}.
delta_from_json(Json_string) ->
Value_pair_decoder = begin
gleam@dynamic@decode:field(
<<"key"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Key) ->
gleam@dynamic@decode:field(
<<"crdt"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Crdt_str) ->
gleam@dynamic@decode:success({Key, Crdt_str})
end
)
end
)
end,
Bounds_decoder = gleam@dynamic@decode:dict(
{decoder, fun gleam@dynamic@decode:decode_string/1},
lattice_core@version_vector:decoder()
),
State_decoder = begin
gleam@dynamic@decode:field(
<<"state"/utf8>>,
begin
gleam@dynamic@decode:field(
<<"replica_id"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Replica_id_str) ->
gleam@dynamic@decode:field(
<<"crdt_spec"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Crdt_spec_str) ->
gleam@dynamic@decode:field(
<<"key_set_delta"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Key_set_str) ->
gleam@dynamic@decode:field(
<<"value_deltas"/utf8>>,
gleam@dynamic@decode:list(
Value_pair_decoder
),
fun(Values_list) ->
gleam@dynamic@decode:optional_field(
<<"remove_bounds_delta"/utf8>>,
maps:new(),
Bounds_decoder,
fun(Bounds) ->
gleam@dynamic@decode:success(
{Replica_id_str,
Crdt_spec_str,
Key_set_str,
Values_list,
Bounds}
)
end
)
end
)
end
)
end
)
end
)
end,
fun(State) -> gleam@dynamic@decode:success(State) 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 =:= <<"or_map_delta"/utf8>> of
false ->
{error,
{unable_to_decode,
[{decode_error,
<<"type=or_map_delta"/utf8>>,
Type_tag@1,
[]}]}};
true ->
case Version@1 of
1 ->
case gleam@json:parse(Json_string, State_decoder) of
{error, E@1} ->
{error, E@1};
{ok,
{Rid_str,
Spec_str,
Key_set_str@1,
Values_list@1,
Bounds@1}} ->
decode_or_map_delta_state(
Rid_str,
Spec_str,
Key_set_str@1,
Values_list@1,
Bounds@1
)
end;
_ ->
{error,
{unable_to_decode,
[{decode_error,
<<"v=1"/utf8>>,
erlang:integer_to_binary(Version@1),
[<<"v"/utf8>>]}]}}
end
end
end.