Packages
Typed distributed messaging for Gleam on the BEAM.
Retired package: Deprecated - The project needs to be redesigned around a much smaller and clearer core.
Current section
Files
Jump to
Current section
Files
src/distribute@codec.erl
-module(distribute@codec).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/distribute/codec.gleam").
-export([encode/2, decode/2, decode_sized/2, to_decoder/1, string_encoder/0, string_sized_decoder/0, string_decoder/0, int_encoder/0, int_sized_decoder/0, int_decoder/0, float_encoder/0, float_sized_decoder/0, float_decoder/0, bool_encoder/0, bool_sized_decoder/0, bool_decoder/0, bitarray_encoder/0, bitarray_sized_decoder/0, bitarray_decoder/0, wrap_envelope/3, unwrap_envelope/1, receive_with_decoder/4, new_schema/4, schema_encode/2, schema_decode/2, schema_matches_tag/2, peek_tag/1, peek_envelope/1, versioned_decoder/2, versioned_decoder_from_schemas/1, list_encoder/1, list_sized_decoder/1, list_decoder/1, list_decoder_fixed/1, dynamic_encoder/0, dynamic_decoder/0, any_encoder/0, pid_encoder/0, pid_decoder/0, subject_encoder/0, subject_decoder/0, option_encoder/1, option_sized_decoder/1, option_decoder/1, result_encoder/2, result_sized_decoder/2, result_decoder/2, tuple2_encoder/2, tuple2_sized_decoder/2, tuple2_decoder/2, tuple3_encoder/3, tuple3_sized_decoder/3, tuple3_decoder/3, decode_error_to_string/1, build_migration_chain/1, schema_decode_with_migrations/2, encode_error_to_string/1, build_migration_graph/1]).
-export_type([encode_error/0, decode_error/0, schema/1]).
-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.
-type encode_error() :: {invalid_value, binary()} |
{encode_failed, binary()} |
{value_too_large, binary()}.
-type decode_error() :: {invalid_binary, binary()} |
{type_mismatch, binary()} |
{decode_failed, binary()} |
{insufficient_data, binary()} |
decode_timeout |
{tag_mismatch, binary(), binary()} |
{version_mismatch, integer(), integer()} |
{migration_missing, integer()} |
{migration_failed, binary()}.
-type schema(FLV) :: {schema,
binary(),
integer(),
fun((FLV) -> {ok, bitstring()} | {error, encode_error()}),
fun((bitstring()) -> {ok, FLV} | {error, decode_error()})}.
-file("src/distribute/codec.gleam", 75).
?DOC(" Encode a value using the given encoder.\n").
-spec encode(fun((FMJ) -> {ok, bitstring()} | {error, encode_error()}), FMJ) -> {ok,
bitstring()} |
{error, encode_error()}.
encode(Encoder, Value) ->
Encoder(Value).
-file("src/distribute/codec.gleam", 89).
?DOC(" Decode binary data using the given decoder.\n").
-spec decode(
fun((bitstring()) -> {ok, FMN} | {error, decode_error()}),
bitstring()
) -> {ok, FMN} | {error, decode_error()}.
decode(Decoder, Data) ->
Decoder(Data).
-file("src/distribute/codec.gleam", 100).
?DOC(" Decode binary data using a sized decoder, returning value and remaining bytes.\n").
-spec decode_sized(
fun((bitstring()) -> {ok, {FMR, bitstring()}} | {error, decode_error()}),
bitstring()
) -> {ok, {FMR, bitstring()}} | {error, decode_error()}.
decode_sized(Decoder, Data) ->
Decoder(Data).
-file("src/distribute/codec.gleam", 108).
?DOC(" Convert a SizedDecoder to a simple Decoder (discards remaining bytes).\n").
-spec to_decoder(
fun((bitstring()) -> {ok, {FMV, bitstring()}} | {error, decode_error()})
) -> fun((bitstring()) -> {ok, FMV} | {error, decode_error()}).
to_decoder(Sized) ->
fun(Data) -> case Sized(Data) of
{ok, {Value, _}} ->
{ok, Value};
{error, E} ->
{error, E}
end end.
-file("src/distribute/codec.gleam", 122).
?DOC(" String encoder: UTF-8 encoding with 16-bit length prefix.\n").
-spec string_encoder() -> fun((binary()) -> {ok, bitstring()} |
{error, encode_error()}).
string_encoder() ->
fun(S) ->
Bytes = gleam_stdlib:identity(S),
Len = erlang:byte_size(Bytes),
case Len > 65535 of
true ->
{error,
{value_too_large,
<<"string too long: "/utf8,
(erlang:integer_to_binary(Len))/binary>>}};
false ->
Len_bytes = <<Len:16>>,
{ok, gleam@bit_array:append(Len_bytes, Bytes)}
end
end.
-file("src/distribute/codec.gleam", 137).
?DOC(" String sized decoder: returns decoded string and remaining bytes.\n").
-spec string_sized_decoder() -> fun((bitstring()) -> {ok,
{binary(), bitstring()}} |
{error, decode_error()}).
string_sized_decoder() ->
fun(Data) -> case Data of
<<Len:16, Rest/binary>> ->
Rest_size = erlang:byte_size(Rest),
case Rest_size >= Len of
true ->
case gleam_stdlib:bit_array_slice(Rest, 0, Len) of
{ok, Str_bytes} ->
case gleam@bit_array:to_string(Str_bytes) of
{ok, S} ->
case gleam_stdlib:bit_array_slice(
Rest,
Len,
Rest_size - Len
) of
{ok, Remaining} ->
{ok, {S, Remaining}};
{error, _} ->
{ok, {S, <<>>}}
end;
{error, _} ->
{error,
{invalid_binary,
<<"invalid UTF-8 sequence"/utf8>>}}
end;
{error, _} ->
{error,
{insufficient_data, <<"slice failed"/utf8>>}}
end;
false ->
{error,
{insufficient_data,
<<"incomplete string data"/utf8>>}}
end;
_ ->
{error, {invalid_binary, <<"missing length prefix"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 168).
?DOC(" String decoder: simple decoder that discards remaining bytes.\n").
-spec string_decoder() -> fun((bitstring()) -> {ok, binary()} |
{error, decode_error()}).
string_decoder() ->
to_decoder(string_sized_decoder()).
-file("src/distribute/codec.gleam", 177).
?DOC(" Integer encoder: 64-bit big-endian encoding (8 bytes).\n").
-spec int_encoder() -> fun((integer()) -> {ok, bitstring()} |
{error, encode_error()}).
int_encoder() ->
fun(I) -> {ok, <<I:64>>} end.
-file("src/distribute/codec.gleam", 182).
?DOC(" Integer sized decoder: returns decoded int and remaining bytes.\n").
-spec int_sized_decoder() -> fun((bitstring()) -> {ok, {integer(), bitstring()}} |
{error, decode_error()}).
int_sized_decoder() ->
fun(Data) -> case Data of
<<I:64, Rest/binary>> ->
{ok, {I, Rest}};
_ ->
{error,
{invalid_binary, <<"insufficient data for int64"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 192).
?DOC(" Integer decoder: simple decoder.\n").
-spec int_decoder() -> fun((bitstring()) -> {ok, integer()} |
{error, decode_error()}).
int_decoder() ->
to_decoder(int_sized_decoder()).
-file("src/distribute/codec.gleam", 201).
?DOC(" Float encoder: 64-bit IEEE 754 encoding (8 bytes).\n").
-spec float_encoder() -> fun((float()) -> {ok, bitstring()} |
{error, encode_error()}).
float_encoder() ->
fun(F) -> {ok, <<F/float>>} end.
-file("src/distribute/codec.gleam", 206).
?DOC(" Float sized decoder: returns decoded float and remaining bytes.\n").
-spec float_sized_decoder() -> fun((bitstring()) -> {ok, {float(), bitstring()}} |
{error, decode_error()}).
float_sized_decoder() ->
fun(Data) -> case Data of
<<F/float, Rest/binary>> ->
{ok, {F, Rest}};
_ ->
{error,
{invalid_binary, <<"insufficient data for float64"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 216).
?DOC(" Float decoder: simple decoder.\n").
-spec float_decoder() -> fun((bitstring()) -> {ok, float()} |
{error, decode_error()}).
float_decoder() ->
to_decoder(float_sized_decoder()).
-file("src/distribute/codec.gleam", 225).
?DOC(" Boolean encoder: single byte (0 or 1).\n").
-spec bool_encoder() -> fun((boolean()) -> {ok, bitstring()} |
{error, encode_error()}).
bool_encoder() ->
fun(B) -> case B of
true ->
{ok, <<1>>};
false ->
{ok, <<0>>}
end end.
-file("src/distribute/codec.gleam", 235).
?DOC(" Boolean sized decoder: returns decoded bool and remaining bytes.\n").
-spec bool_sized_decoder() -> fun((bitstring()) -> {ok,
{boolean(), bitstring()}} |
{error, decode_error()}).
bool_sized_decoder() ->
fun(Data) -> case Data of
<<1, Rest/binary>> ->
{ok, {true, Rest}};
<<0, Rest@1/binary>> ->
{ok, {false, Rest@1}};
_ ->
{error, {invalid_binary, <<"invalid boolean value"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 246).
?DOC(" Boolean decoder: simple decoder.\n").
-spec bool_decoder() -> fun((bitstring()) -> {ok, boolean()} |
{error, decode_error()}).
bool_decoder() ->
to_decoder(bool_sized_decoder()).
-file("src/distribute/codec.gleam", 255).
?DOC(" BitArray encoder: 32-bit length prefix followed by raw bytes.\n").
-spec bitarray_encoder() -> fun((bitstring()) -> {ok, bitstring()} |
{error, encode_error()}).
bitarray_encoder() ->
fun(Bytes) ->
Len = erlang:byte_size(Bytes),
{ok, gleam@bit_array:append(<<Len:32>>, Bytes)}
end.
-file("src/distribute/codec.gleam", 263).
?DOC(" BitArray sized decoder: returns decoded bytes and remaining.\n").
-spec bitarray_sized_decoder() -> fun((bitstring()) -> {ok,
{bitstring(), bitstring()}} |
{error, decode_error()}).
bitarray_sized_decoder() ->
fun(Data) -> case Data of
<<Len:32, Rest/binary>> ->
Rest_size = erlang:byte_size(Rest),
case Rest_size >= Len of
true ->
case gleam_stdlib:bit_array_slice(Rest, 0, Len) of
{ok, Bytes} ->
case gleam_stdlib:bit_array_slice(
Rest,
Len,
Rest_size - Len
) of
{ok, Remaining} ->
{ok, {Bytes, Remaining}};
{error, _} ->
{ok, {Bytes, <<>>}}
end;
{error, _} ->
{error,
{insufficient_data,
<<"bitarray slice failed"/utf8>>}}
end;
false ->
{error,
{insufficient_data,
<<"incomplete bitarray data"/utf8>>}}
end;
_ ->
{error,
{invalid_binary, <<"missing bitarray length prefix"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 289).
?DOC(" BitArray decoder: simple decoder.\n").
-spec bitarray_decoder() -> fun((bitstring()) -> {ok, bitstring()} |
{error, decode_error()}).
bitarray_decoder() ->
to_decoder(bitarray_sized_decoder()).
-file("src/distribute/codec.gleam", 354).
-spec wrap_envelope(binary(), integer(), bitstring()) -> bitstring().
wrap_envelope(Tag, Version, Payload) ->
Tag_bytes = gleam_stdlib:identity(Tag),
Tag_len = erlang:byte_size(Tag_bytes),
Tag_len_bytes = <<Tag_len:16>>,
Version_bytes = <<Version:32>>,
gleam@bit_array:append(
gleam@bit_array:append(Tag_len_bytes, Tag_bytes),
gleam@bit_array:append(Version_bytes, Payload)
).
-file("src/distribute/codec.gleam", 366).
-spec unwrap_envelope(bitstring()) -> {ok, {binary(), integer(), bitstring()}} |
{error, decode_error()}.
unwrap_envelope(Data) ->
case Data of
<<Tag_len:16, Rest/binary>> ->
Rest_size = erlang:byte_size(Rest),
case Rest_size >= (Tag_len + 4) of
false ->
{error, {insufficient_data, <<"envelope too short"/utf8>>}};
true ->
case gleam_stdlib:bit_array_slice(Rest, 0, Tag_len) of
{ok, Tag_bytes} ->
case gleam@bit_array:to_string(Tag_bytes) of
{ok, Tag_str} ->
case gleam_stdlib:bit_array_slice(
Rest,
Tag_len,
Rest_size - Tag_len
) of
{ok, After_tag} ->
case After_tag of
<<Version:32, Payload/binary>> ->
{ok,
{Tag_str,
Version,
Payload}};
_ ->
{error,
{invalid_binary,
<<"missing version/payload in envelope"/utf8>>}}
end;
{error, _} ->
{error,
{insufficient_data,
<<"failed to slice after tag"/utf8>>}}
end;
{error, _} ->
{error,
{invalid_binary,
<<"invalid UTF-8 in tag"/utf8>>}}
end;
{error, _} ->
{error,
{insufficient_data,
<<"failed to slice tag bytes"/utf8>>}}
end
end;
_ ->
{error, {invalid_binary, <<"missing envelope length prefix"/utf8>>}}
end.
-file("src/distribute/codec.gleam", 410).
-spec receive_with_decoder(
fun((bitstring()) -> {ok, FOF} | {error, decode_error()}),
binary(),
integer(),
bitstring()
) -> {ok, FOF} | {error, decode_error()}.
receive_with_decoder(Decoder, Expected_tag, Expected_version, Data) ->
case unwrap_envelope(Data) of
{ok, {Tag, Version, Payload}} ->
case Tag =:= Expected_tag of
false ->
{error, {type_mismatch, <<"tag mismatch"/utf8>>}};
true ->
case Version =:= Expected_version of
false ->
{error,
{type_mismatch, <<"version mismatch"/utf8>>}};
true ->
decode(Decoder, Payload)
end
end;
{error, E} ->
{error, E}
end.
-file("src/distribute/codec.gleam", 441).
?DOC(" Create a new schema with the given tag, version, encoder, and decoder.\n").
-spec new_schema(
binary(),
integer(),
fun((FOJ) -> {ok, bitstring()} | {error, encode_error()}),
fun((bitstring()) -> {ok, FOJ} | {error, decode_error()})
) -> schema(FOJ).
new_schema(Tag, Version, Encoder, Decoder) ->
{schema, Tag, Version, Encoder, Decoder}.
-file("src/distribute/codec.gleam", 451).
?DOC(" Encode a value using a schema. Automatically wraps in envelope with tag+version.\n").
-spec schema_encode(schema(FON), FON) -> {ok, bitstring()} |
{error, encode_error()}.
schema_encode(Schema, Value) ->
case (erlang:element(4, Schema))(Value) of
{ok, Payload} ->
{ok,
wrap_envelope(
erlang:element(2, Schema),
erlang:element(3, Schema),
Payload
)};
{error, E} ->
{error, E}
end.
-file("src/distribute/codec.gleam", 462).
?DOC(" Decode a value using a schema. Validates tag and version before decoding.\n").
-spec schema_decode(schema(FOR), bitstring()) -> {ok, FOR} |
{error, decode_error()}.
schema_decode(Schema, Data) ->
case unwrap_envelope(Data) of
{ok, {Tag, Version, Payload}} ->
case Tag =:= erlang:element(2, Schema) of
false ->
{error, {tag_mismatch, erlang:element(2, Schema), Tag}};
true ->
case Version =:= erlang:element(3, Schema) of
false ->
{error,
{version_mismatch,
erlang:element(3, Schema),
Version}};
true ->
(erlang:element(5, Schema))(Payload)
end
end;
{error, E} ->
{error, E}
end.
-file("src/distribute/codec.gleam", 484).
?DOC(
" Check if binary data matches a schema's tag (without full decode).\n"
" Useful for routing messages to the correct handler.\n"
).
-spec schema_matches_tag(schema(any()), bitstring()) -> boolean().
schema_matches_tag(Schema, Data) ->
case unwrap_envelope(Data) of
{ok, {Tag, _, _}} ->
Tag =:= erlang:element(2, Schema);
{error, _} ->
false
end.
-file("src/distribute/codec.gleam", 492).
?DOC(" Get the tag from a binary envelope without full decode.\n").
-spec peek_tag(bitstring()) -> {ok, binary()} | {error, decode_error()}.
peek_tag(Data) ->
case unwrap_envelope(Data) of
{ok, {Tag, _, _}} ->
{ok, Tag};
{error, E} ->
{error, E}
end.
-file("src/distribute/codec.gleam", 500).
?DOC(" Get tag and version from a binary envelope without decoding payload.\n").
-spec peek_envelope(bitstring()) -> {ok, {binary(), integer()}} |
{error, decode_error()}.
peek_envelope(Data) ->
case unwrap_envelope(Data) of
{ok, {Tag, Version, _}} ->
{ok, {Tag, Version}};
{error, E} ->
{error, E}
end.
-file("src/distribute/codec.gleam", 509).
?DOC(
" Create a versioned migration path between schema versions.\n"
" Returns a decoder that can handle multiple versions.\n"
).
-spec versioned_decoder(
binary(),
list({integer(), fun((bitstring()) -> {ok, FPB} | {error, decode_error()})})
) -> fun((bitstring()) -> {ok, FPB} | {error, decode_error()}).
versioned_decoder(Tag, Handlers) ->
fun(Data) -> case unwrap_envelope(Data) of
{ok, {Actual_tag, Version, Payload}} ->
case Actual_tag =:= Tag of
false ->
{error, {tag_mismatch, Tag, Actual_tag}};
true ->
case gleam@list:find(
Handlers,
fun(H) -> erlang:element(1, H) =:= Version end
) of
{ok, {_, Decoder}} ->
Decoder(Payload);
{error, _} ->
{error, {version_mismatch, -1, Version}}
end
end;
{error, E} ->
{error, E}
end end.
-file("src/distribute/codec.gleam", 539).
?DOC(
" Build a versioned decoder from a list of `Schema(a)`.\n"
" All schemas must share the same tag. Returns a decoder that picks the\n"
" appropriate schema by version and decodes the payload.\n"
).
-spec versioned_decoder_from_schemas(list(schema(FPF))) -> fun((bitstring()) -> {ok,
FPF} |
{error, decode_error()}).
versioned_decoder_from_schemas(Schemas) ->
fun(Data) -> case Schemas of
[] ->
{error, {invalid_binary, <<"no schemas provided"/utf8>>}};
[{schema, Tag, _, _, _} | _] ->
Handlers = gleam@list:map(
Schemas,
fun(S) -> {erlang:element(3, S), erlang:element(5, S)} end
),
(versioned_decoder(Tag, Handlers))(Data)
end end.
-file("src/distribute/codec.gleam", 745).
-spec find_edge(
list({integer(),
integer(),
fun((bitstring()) -> {ok, bitstring()} | {error, decode_error()})}),
integer(),
integer()
) -> {ok, fun((bitstring()) -> {ok, bitstring()} | {error, decode_error()})} |
{error, decode_error()}.
find_edge(Edges, From, To) ->
case gleam@list:find(
Edges,
fun(E) ->
(erlang:element(1, E) =:= From) andalso (erlang:element(2, E) =:= To)
end
) of
{ok, {_, _, Migr}} ->
{ok, Migr};
{error, _} ->
{error, {migration_missing, From}}
end.
-file("src/distribute/codec.gleam", 761).
?DOC(" Helper for result chaining in binary operations.\n").
-spec result_then(
{ok, FQS} | {error, FQT},
fun((FQS) -> {ok, FQW} | {error, FQT})
) -> {ok, FQW} | {error, FQT}.
result_then(Result, F) ->
case Result of
{ok, Value} ->
F(Value);
{error, E} ->
{error, E}
end.
-file("src/distribute/codec.gleam", 769).
?DOC(" Helper to encode list elements recursively.\n").
-spec encode_list_elements(
list(FRB),
fun((FRB) -> {ok, bitstring()} | {error, encode_error()}),
bitstring()
) -> {ok, bitstring()} | {error, encode_error()}.
encode_list_elements(Lst, Encoder, Acc) ->
case Lst of
[] ->
{ok, Acc};
[Head | Tail] ->
result_then(
encode(Encoder, Head),
fun(Encoded_head) ->
New_acc = gleam@bit_array:append(Acc, Encoded_head),
encode_list_elements(Tail, Encoder, New_acc)
end
)
end.
-file("src/distribute/codec.gleam", 298).
?DOC(" List encoder: 16-bit length prefix followed by encoded elements.\n").
-spec list_encoder(fun((FNN) -> {ok, bitstring()} | {error, encode_error()})) -> fun((list(FNN)) -> {ok,
bitstring()} |
{error, encode_error()}).
list_encoder(Element_encoder) ->
fun(Lst) ->
Len = erlang:length(Lst),
case Len > 65535 of
true ->
{error,
{value_too_large,
<<"list too long: "/utf8,
(erlang:integer_to_binary(Len))/binary>>}};
false ->
Len_bytes = <<Len:16>>,
result_then(
encode_list_elements(Lst, Element_encoder, <<>>),
fun(Encoded_elements) ->
{ok,
gleam@bit_array:append(Len_bytes, Encoded_elements)}
end
)
end
end.
-file("src/distribute/codec.gleam", 785).
?DOC(" Helper to decode list elements using SizedDecoder (proper implementation).\n").
-spec decode_list_elements_sized(
bitstring(),
fun((bitstring()) -> {ok, {FRG, bitstring()}} | {error, decode_error()}),
integer(),
list(FRG)
) -> {ok, {list(FRG), bitstring()}} | {error, decode_error()}.
decode_list_elements_sized(Data, Decoder, Remaining, Acc) ->
case Remaining of
0 ->
{ok, {lists:reverse(Acc), Data}};
_ ->
case Decoder(Data) of
{ok, {Element, Rest}} ->
decode_list_elements_sized(
Rest,
Decoder,
Remaining - 1,
[Element | Acc]
);
{error, E} ->
{error, E}
end
end.
-file("src/distribute/codec.gleam", 316).
?DOC(
" List sized decoder using a SizedDecoder for elements.\n"
" This properly tracks byte consumption for each element.\n"
).
-spec list_sized_decoder(
fun((bitstring()) -> {ok, {FNR, bitstring()}} | {error, decode_error()})
) -> fun((bitstring()) -> {ok, {list(FNR), bitstring()}} |
{error, decode_error()}).
list_sized_decoder(Element_decoder) ->
fun(Data) -> case Data of
<<Len:16, Rest/binary>> ->
decode_list_elements_sized(Rest, Element_decoder, Len, []);
_ ->
{error, {invalid_binary, <<"missing list length"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 329).
?DOC(" List decoder using a SizedDecoder for elements.\n").
-spec list_decoder(
fun((bitstring()) -> {ok, {FNV, bitstring()}} | {error, decode_error()})
) -> fun((bitstring()) -> {ok, list(FNV)} | {error, decode_error()}).
list_decoder(Element_decoder) ->
to_decoder(list_sized_decoder(Element_decoder)).
-file("src/distribute/codec.gleam", 807).
?DOC(" Legacy helper for fixed-size decoders (deprecated).\n").
-spec decode_list_elements_legacy(
bitstring(),
fun((bitstring()) -> {ok, FRM} | {error, decode_error()}),
integer(),
list(FRM)
) -> {ok, list(FRM)} | {error, decode_error()}.
decode_list_elements_legacy(Data, Decoder, Remaining, Acc) ->
case Remaining of
0 ->
{ok, lists:reverse(Acc)};
_ ->
result_then(
decode(Decoder, Data),
fun(Element) ->
decode_list_elements_legacy(
Data,
Decoder,
Remaining - 1,
[Element | Acc]
)
end
)
end.
-file("src/distribute/codec.gleam", 339).
?DOC(
" List decoder using a simple Decoder - ONLY for fixed-size elements.\n"
" \n"
" ⚠️ **WARNING**: This function only works correctly with fixed-size decoders\n"
" (e.g., int_decoder, float_decoder, bool_decoder). For variable-size elements\n"
" (strings, nested lists, custom types), use `list_decoder` with a SizedDecoder.\n"
).
-spec list_decoder_fixed(
fun((bitstring()) -> {ok, FNZ} | {error, decode_error()})
) -> fun((bitstring()) -> {ok, list(FNZ)} | {error, decode_error()}).
list_decoder_fixed(Element_decoder) ->
fun(Data) -> case Data of
<<Len:16, Rest/binary>> ->
decode_list_elements_legacy(Rest, Element_decoder, Len, []);
_ ->
{error, {invalid_binary, <<"missing list length"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 863).
?DOC(
" Encoder for any Gleam/Erlang term using `term_to_binary`.\n"
"\n"
" ⚠️ **ESCAPE HATCH** — This encoder bypasses Gleam's type system.\n"
" The binary format is opaque and may not be portable across:\n"
" - Different Erlang/OTP versions\n"
" - Different Gleam versions\n"
" - Type definition changes in your code\n"
"\n"
" Use typed codecs (string_encoder, int_encoder, etc.) whenever possible.\n"
" This is suitable only for temporary in-cluster communication or prototyping.\n"
).
-spec dynamic_encoder() -> fun((gleam@dynamic:dynamic_()) -> {ok, bitstring()} |
{error, encode_error()}).
dynamic_encoder() ->
fun(D) -> {ok, erlang:term_to_binary(D)} end.
-file("src/distribute/codec.gleam", 875).
?DOC(
" Decoder for any Gleam/Erlang term using `binary_to_term`.\n"
"\n"
" ⚠️ **ESCAPE HATCH** — This decoder bypasses Gleam's type system.\n"
" The returned `Dynamic` must be validated using `gleam/dynamic` decoders.\n"
" Binary data from untrusted sources should be treated with extreme caution.\n"
"\n"
" Uses `binary_to_term([safe])` to prevent atom table attacks, but the\n"
" decoded value still requires runtime type checking.\n"
).
-spec dynamic_decoder() -> fun((bitstring()) -> {ok, gleam@dynamic:dynamic_()} |
{error, decode_error()}).
dynamic_decoder() ->
fun(B) -> case codec_ffi:safe_binary_to_term(B) of
{ok, D} ->
{ok, D};
{error, _} ->
{error, {invalid_binary, <<"invalid erlang term"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 888).
?DOC(
" Encode any Gleam value to binary using Erlang term serialization.\n"
"\n"
" ⚠️ **ESCAPE HATCH** — Bypasses type safety. See `dynamic_encoder` warnings.\n"
" This is a convenience function for quick prototyping.\n"
).
-spec any_encoder() -> fun((any()) -> {ok, bitstring()} |
{error, encode_error()}).
any_encoder() ->
fun(Value) -> {ok, erlang:term_to_binary(Value)} end.
-file("src/distribute/codec.gleam", 899).
?DOC(
" Encoder for Pids.\n"
"\n"
" Note: Pids are inherently untyped in Erlang. This encoder uses\n"
" `term_to_binary` which is safe for same-cluster communication.\n"
).
-spec pid_encoder() -> fun((gleam@erlang@process:pid_()) -> {ok, bitstring()} |
{error, encode_error()}).
pid_encoder() ->
fun(P) -> {ok, erlang:term_to_binary(P)} end.
-file("src/distribute/codec.gleam", 906).
?DOC(
" Decoder for Pids.\n"
"\n"
" Note: The decoded Pid is validated as a proper Erlang pid internally.\n"
).
-spec pid_decoder() -> fun((bitstring()) -> {ok, gleam@erlang@process:pid_()} |
{error, decode_error()}).
pid_decoder() ->
fun(B) -> case codec_ffi:safe_binary_to_term(B) of
{ok, D} ->
{ok, codec_ffi:unsafe_coerce(D)};
{error, _} ->
{error, {invalid_binary, <<"invalid pid binary"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 919).
?DOC(
" Encoder for Subjects.\n"
"\n"
" Note: Subject serialization preserves the Pid but the type parameter\n"
" is erased. The receiving side must know the expected message type.\n"
).
-spec subject_encoder() -> fun((gleam@erlang@process:subject(any())) -> {ok,
bitstring()} |
{error, encode_error()}).
subject_encoder() ->
fun(S) -> {ok, erlang:term_to_binary(S)} end.
-file("src/distribute/codec.gleam", 928).
?DOC(
" Decoder for Subjects.\n"
"\n"
" ⚠️ **Type parameter is not validated** — The returned Subject(a) will\n"
" accept any type parameter at compile time. Ensure the type matches\n"
" what was encoded, or use Schema-based messaging for safety.\n"
).
-spec subject_decoder() -> fun((bitstring()) -> {ok,
gleam@erlang@process:subject(any())} |
{error, decode_error()}).
subject_decoder() ->
fun(B) -> case codec_ffi:safe_binary_to_term(B) of
{ok, D} ->
{ok, codec_ffi:unsafe_coerce(D)};
{error, _} ->
{error, {invalid_binary, <<"invalid subject binary"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 942).
?DOC(" Codec for Option(a).\n").
-spec option_encoder(fun((FSI) -> {ok, bitstring()} | {error, encode_error()})) -> fun((gleam@option:option(FSI)) -> {ok,
bitstring()} |
{error, encode_error()}).
option_encoder(Inner) ->
fun(Opt) -> case Opt of
none ->
{ok, <<0>>};
{some, Value} ->
gleam@result:'try'(
Inner(Value),
fun(Encoded) ->
{ok, gleam@bit_array:append(<<1>>, Encoded)}
end
)
end end.
-file("src/distribute/codec.gleam", 955).
?DOC(" Option sized decoder.\n").
-spec option_sized_decoder(
fun((bitstring()) -> {ok, {FSM, bitstring()}} | {error, decode_error()})
) -> fun((bitstring()) -> {ok, {gleam@option:option(FSM), bitstring()}} |
{error, decode_error()}).
option_sized_decoder(Inner) ->
fun(Data) -> case Data of
<<0, Rest/binary>> ->
{ok, {none, Rest}};
<<1, Rest@1/binary>> ->
case Inner(Rest@1) of
{ok, {Value, Remaining}} ->
{ok, {{some, Value}, Remaining}};
{error, E} ->
{error, E}
end;
_ ->
{error, {invalid_binary, <<"invalid option tag"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 971).
?DOC(" Option decoder (simple).\n").
-spec option_decoder(fun((bitstring()) -> {ok, FSQ} | {error, decode_error()})) -> fun((bitstring()) -> {ok,
gleam@option:option(FSQ)} |
{error, decode_error()}).
option_decoder(Inner) ->
fun(Data) -> case Data of
<<0, _/binary>> ->
{ok, none};
<<1, Rest/binary>> ->
gleam@result:'try'(
Inner(Rest),
fun(Value) -> {ok, {some, Value}} end
);
_ ->
{error, {invalid_binary, <<"invalid option tag"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 985).
?DOC(" Codec for Result(a, e).\n").
-spec result_encoder(
fun((FSU) -> {ok, bitstring()} | {error, encode_error()}),
fun((FSW) -> {ok, bitstring()} | {error, encode_error()})
) -> fun(({ok, FSU} | {error, FSW}) -> {ok, bitstring()} |
{error, encode_error()}).
result_encoder(Ok_encoder, Error_encoder) ->
fun(Res) -> case Res of
{ok, Value} ->
gleam@result:'try'(
Ok_encoder(Value),
fun(Encoded) ->
{ok, gleam@bit_array:append(<<0>>, Encoded)}
end
);
{error, Err} ->
gleam@result:'try'(
Error_encoder(Err),
fun(Encoded@1) ->
{ok, gleam@bit_array:append(<<1>>, Encoded@1)}
end
)
end end.
-file("src/distribute/codec.gleam", 1004).
?DOC(" Result sized decoder.\n").
-spec result_sized_decoder(
fun((bitstring()) -> {ok, {FTB, bitstring()}} | {error, decode_error()}),
fun((bitstring()) -> {ok, {FTD, bitstring()}} | {error, decode_error()})
) -> fun((bitstring()) -> {ok, {{ok, FTB} | {error, FTD}, bitstring()}} |
{error, decode_error()}).
result_sized_decoder(Ok_decoder, Error_decoder) ->
fun(Data) -> case Data of
<<0, Rest/binary>> ->
case Ok_decoder(Rest) of
{ok, {Value, Remaining}} ->
{ok, {{ok, Value}, Remaining}};
{error, E} ->
{error, E}
end;
<<1, Rest@1/binary>> ->
case Error_decoder(Rest@1) of
{ok, {Err, Remaining@1}} ->
{ok, {{error, Err}, Remaining@1}};
{error, E@1} ->
{error, E@1}
end;
_ ->
{error, {invalid_binary, <<"invalid result tag"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 1028).
?DOC(" Result decoder (simple).\n").
-spec result_decoder(
fun((bitstring()) -> {ok, FTI} | {error, decode_error()}),
fun((bitstring()) -> {ok, FTK} | {error, decode_error()})
) -> fun((bitstring()) -> {ok, {ok, FTI} | {error, FTK}} |
{error, decode_error()}).
result_decoder(Ok_decoder, Error_decoder) ->
fun(Data) -> case Data of
<<0, Rest/binary>> ->
gleam@result:'try'(
Ok_decoder(Rest),
fun(Value) -> {ok, {ok, Value}} end
);
<<1, Rest@1/binary>> ->
gleam@result:'try'(
Error_decoder(Rest@1),
fun(Err) -> {ok, {error, Err}} end
);
_ ->
{error, {invalid_binary, <<"invalid result tag"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 1052).
?DOC(" Tuple2 encoder with length prefixes for proper boundary tracking.\n").
-spec tuple2_encoder(
fun((FTP) -> {ok, bitstring()} | {error, encode_error()}),
fun((FTR) -> {ok, bitstring()} | {error, encode_error()})
) -> fun(({FTP, FTR}) -> {ok, bitstring()} | {error, encode_error()}).
tuple2_encoder(First, Second) ->
fun(Tuple) ->
{A, B} = Tuple,
gleam@result:'try'(
First(A),
fun(Encoded_a) ->
gleam@result:'try'(
Second(B),
fun(Encoded_b) ->
Len_a = erlang:byte_size(Encoded_a),
{ok,
gleam_stdlib:bit_array_concat(
[<<Len_a:32>>, Encoded_a, Encoded_b]
)}
end
)
end
)
end.
-file("src/distribute/codec.gleam", 1064).
?DOC(" Tuple2 sized decoder.\n").
-spec tuple2_sized_decoder(
fun((bitstring()) -> {ok, {FTU, bitstring()}} | {error, decode_error()}),
fun((bitstring()) -> {ok, {FTW, bitstring()}} | {error, decode_error()})
) -> fun((bitstring()) -> {ok, {{FTU, FTW}, bitstring()}} |
{error, decode_error()}).
tuple2_sized_decoder(First, Second) ->
fun(Data) -> case Data of
<<Len_a:32, Rest/binary>> ->
case gleam_stdlib:bit_array_slice(Rest, 0, Len_a) of
{ok, First_data} ->
case First(First_data) of
{ok, {A, _}} ->
Rest_size = erlang:byte_size(Rest),
case gleam_stdlib:bit_array_slice(
Rest,
Len_a,
Rest_size - Len_a
) of
{ok, Second_data} ->
case Second(Second_data) of
{ok, {B, Remaining}} ->
{ok, {{A, B}, Remaining}};
{error, E} ->
{error, E}
end;
{error, _} ->
{error,
{insufficient_data,
<<"tuple2 second element"/utf8>>}}
end;
{error, E@1} ->
{error, E@1}
end;
{error, _} ->
{error,
{insufficient_data, <<"tuple2 first element"/utf8>>}}
end;
_ ->
{error,
{invalid_binary, <<"missing tuple2 length prefix"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 1098).
?DOC(" Tuple2 decoder (simple).\n").
-spec tuple2_decoder(
fun((bitstring()) -> {ok, {FTZ, bitstring()}} | {error, decode_error()}),
fun((bitstring()) -> {ok, {FUB, bitstring()}} | {error, decode_error()})
) -> fun((bitstring()) -> {ok, {FTZ, FUB}} | {error, decode_error()}).
tuple2_decoder(First, Second) ->
to_decoder(tuple2_sized_decoder(First, Second)).
-file("src/distribute/codec.gleam", 1106).
?DOC(" Tuple3 encoder.\n").
-spec tuple3_encoder(
fun((FUE) -> {ok, bitstring()} | {error, encode_error()}),
fun((FUG) -> {ok, bitstring()} | {error, encode_error()}),
fun((FUI) -> {ok, bitstring()} | {error, encode_error()})
) -> fun(({FUE, FUG, FUI}) -> {ok, bitstring()} | {error, encode_error()}).
tuple3_encoder(First, Second, Third) ->
fun(Tuple) ->
{A, B, C} = Tuple,
gleam@result:'try'(
First(A),
fun(Encoded_a) ->
gleam@result:'try'(
Second(B),
fun(Encoded_b) ->
gleam@result:'try'(
Third(C),
fun(Encoded_c) ->
Len_a = erlang:byte_size(Encoded_a),
Len_b = erlang:byte_size(Encoded_b),
{ok,
gleam_stdlib:bit_array_concat(
[<<Len_a:32>>,
Encoded_a,
<<Len_b:32>>,
Encoded_b,
Encoded_c]
)}
end
)
end
)
end
)
end.
-file("src/distribute/codec.gleam", 1131).
?DOC(" Tuple3 sized decoder.\n").
-spec tuple3_sized_decoder(
fun((bitstring()) -> {ok, {FUL, bitstring()}} | {error, decode_error()}),
fun((bitstring()) -> {ok, {FUN, bitstring()}} | {error, decode_error()}),
fun((bitstring()) -> {ok, {FUP, bitstring()}} | {error, decode_error()})
) -> fun((bitstring()) -> {ok, {{FUL, FUN, FUP}, bitstring()}} |
{error, decode_error()}).
tuple3_sized_decoder(First, Second, Third) ->
fun(Data) -> case Data of
<<Len_a:32, Rest/binary>> ->
Rest_size = erlang:byte_size(Rest),
case gleam_stdlib:bit_array_slice(Rest, 0, Len_a) of
{ok, First_data} ->
case First(First_data) of
{ok, {A, _}} ->
case gleam_stdlib:bit_array_slice(
Rest,
Len_a,
Rest_size - Len_a
) of
{ok, After_a} ->
case After_a of
<<Len_b:32, Rest2/binary>> ->
Rest2_size = erlang:byte_size(
Rest2
),
case gleam_stdlib:bit_array_slice(
Rest2,
0,
Len_b
) of
{ok, Second_data} ->
case Second(Second_data) of
{ok, {B, _}} ->
case gleam_stdlib:bit_array_slice(
Rest2,
Len_b,
Rest2_size - Len_b
) of
{ok,
Third_data} ->
case Third(
Third_data
) of
{ok,
{C,
Remaining}} ->
{ok,
{{A,
B,
C},
Remaining}};
{error,
E} ->
{error,
E}
end;
{error, _} ->
{error,
{insufficient_data,
<<"tuple3 third"/utf8>>}}
end;
{error, E@1} ->
{error, E@1}
end;
{error, _} ->
{error,
{insufficient_data,
<<"tuple3 second slice"/utf8>>}}
end;
_ ->
{error,
{invalid_binary,
<<"tuple3 missing second length"/utf8>>}}
end;
{error, _} ->
{error,
{insufficient_data,
<<"tuple3 after first"/utf8>>}}
end;
{error, E@2} ->
{error, E@2}
end;
{error, _} ->
{error, {insufficient_data, <<"tuple3 first"/utf8>>}}
end;
_ ->
{error, {invalid_binary, <<"tuple3 missing length"/utf8>>}}
end end.
-file("src/distribute/codec.gleam", 1196).
?DOC(" Tuple3 decoder (simple).\n").
-spec tuple3_decoder(
fun((bitstring()) -> {ok, {FUS, bitstring()}} | {error, decode_error()}),
fun((bitstring()) -> {ok, {FUU, bitstring()}} | {error, decode_error()}),
fun((bitstring()) -> {ok, {FUW, bitstring()}} | {error, decode_error()})
) -> fun((bitstring()) -> {ok, {FUS, FUU, FUW}} | {error, decode_error()}).
tuple3_decoder(First, Second, Third) ->
to_decoder(tuple3_sized_decoder(First, Second, Third)).
-file("src/distribute/codec.gleam", 1209).
?DOC(" Format a DecodeError as a human-readable string.\n").
-spec decode_error_to_string(decode_error()) -> binary().
decode_error_to_string(Error) ->
case Error of
{invalid_binary, Msg} ->
<<"Invalid binary: "/utf8, Msg/binary>>;
{type_mismatch, Msg@1} ->
<<"Type mismatch: "/utf8, Msg@1/binary>>;
{decode_failed, Msg@2} ->
<<"Decode failed: "/utf8, Msg@2/binary>>;
{insufficient_data, Msg@3} ->
<<"Insufficient data: "/utf8, Msg@3/binary>>;
decode_timeout ->
<<"Decode timeout"/utf8>>;
{tag_mismatch, Expected, Got} ->
<<<<<<<<"Tag mismatch: expected '"/utf8, Expected/binary>>/binary,
"', got '"/utf8>>/binary,
Got/binary>>/binary,
"'"/utf8>>;
{version_mismatch, Expected@1, Got@1} ->
<<<<<<"Version mismatch: expected "/utf8,
(erlang:integer_to_binary(Expected@1))/binary>>/binary,
", got "/utf8>>/binary,
(erlang:integer_to_binary(Got@1))/binary>>;
{migration_missing, Step} ->
<<"Migration missing for step: "/utf8,
(erlang:integer_to_binary(Step))/binary>>;
{migration_failed, Msg@4} ->
<<"Migration failed: "/utf8, Msg@4/binary>>
end.
-file("src/distribute/codec.gleam", 627).
-spec apply_steps(
integer(),
integer(),
bitstring(),
list({integer(),
fun((bitstring()) -> {ok, bitstring()} | {error, decode_error()})})
) -> {ok, bitstring()} | {error, decode_error()}.
apply_steps(Current, Target, Payload, Migrations) ->
case Current =:= Target of
true ->
{ok, Payload};
false ->
case gleam@list:find(
Migrations,
fun(M) -> erlang:element(1, M) =:= Current end
) of
{ok, {_, Migr}} ->
case Migr(Payload) of
{ok, Next_payload} ->
apply_steps(
Current + 1,
Target,
Next_payload,
Migrations
);
{error, E} ->
{error,
{migration_failed, decode_error_to_string(E)}}
end;
{error, _} ->
{error, {migration_missing, Current}}
end
end.
-file("src/distribute/codec.gleam", 612).
?DOC(
" Build a migration chain from single-step migrations.\n"
"\n"
" The returned function has signature: fn(from_version, to_version, payload)\n"
" and applies the sequence of migrations (from->from+1, ... -> to) if possible.\n"
" If a step is missing, returns `MigrationMissing(step)`.\n"
).
-spec build_migration_chain(
list({integer(),
fun((bitstring()) -> {ok, bitstring()} | {error, decode_error()})})
) -> fun((integer(), integer(), bitstring()) -> {ok, bitstring()} |
{error, decode_error()}).
build_migration_chain(Migrations) ->
fun(From, To, Payload) -> case From =:= To of
true ->
{ok, Payload};
false ->
case From > To of
true ->
{error,
{migration_failed,
<<"downgrade not supported"/utf8>>}};
false ->
apply_steps(From, To, Payload, Migrations)
end
end end.
-file("src/distribute/codec.gleam", 560).
?DOC(
" Decode a schema applying migrations when necessary.\n"
"\n"
" - `target_schema` is the desired schema to decode into (its decoder is used).\n"
" - `migrations` is a list of pairs `(version, migr_fn)` where `migr_fn`\n"
" transforms an older payload into the payload expected by the target schema.\n"
"\n"
" If a payload arrives with an older version, the corresponding migration\n"
" is applied before decoding. If no migration is found the decoder returns\n"
" `VersionMismatch`.\n"
).
-spec schema_decode_with_migrations(
schema(FPJ),
list({integer(),
fun((bitstring()) -> {ok, bitstring()} | {error, decode_error()})})
) -> fun((bitstring()) -> {ok, FPJ} | {error, decode_error()}).
schema_decode_with_migrations(Target_schema, Migrations) ->
fun(Data) -> case unwrap_envelope(Data) of
{ok, {Tag, Version, Payload}} ->
case Tag =:= erlang:element(2, Target_schema) of
false ->
{error,
{tag_mismatch,
erlang:element(2, Target_schema),
Tag}};
true ->
case Version =:= erlang:element(3, Target_schema) of
true ->
(erlang:element(5, Target_schema))(Payload);
false ->
case gleam@list:find(
Migrations,
fun(M) ->
erlang:element(1, M) =:= Version
end
) of
{ok, {_, Migr}} ->
case Migr(Payload) of
{ok, New_payload} ->
(erlang:element(
5,
Target_schema
))(New_payload);
{error, E} ->
{error, E}
end;
{error, _} ->
Chain = build_migration_chain(
Migrations
),
case Chain(
Version,
erlang:element(3, Target_schema),
Payload
) of
{ok, New_payload@1} ->
(erlang:element(
5,
Target_schema
))(New_payload@1);
{error, E@1} ->
{error, E@1}
end
end
end
end;
{error, E@2} ->
{error, E@2}
end end.
-file("src/distribute/codec.gleam", 725).
-spec apply_path_edges(
list({integer(),
integer(),
fun((bitstring()) -> {ok, bitstring()} | {error, decode_error()})}),
list(integer()),
bitstring()
) -> {ok, bitstring()} | {error, decode_error()}.
apply_path_edges(Edges, Versions, Payload) ->
case Versions of
[] ->
{ok, Payload};
[_] ->
{ok, Payload};
[From, To | Rest] ->
case find_edge(Edges, From, To) of
{ok, Migr} ->
case Migr(Payload) of
{ok, Next} ->
apply_path_edges(Edges, [To | Rest], Next);
{error, E} ->
{error,
{migration_failed, decode_error_to_string(E)}}
end;
{error, _} ->
{error, {migration_missing, From}}
end
end.
-file("src/distribute/codec.gleam", 1230).
?DOC(" Format an EncodeError as a human-readable string.\n").
-spec encode_error_to_string(encode_error()) -> binary().
encode_error_to_string(Error) ->
case Error of
{invalid_value, Msg} ->
<<"Invalid value: "/utf8, Msg/binary>>;
{encode_failed, Msg@1} ->
<<"Encode failed: "/utf8, Msg@1/binary>>;
{value_too_large, Msg@2} ->
<<"Value too large: "/utf8, Msg@2/binary>>
end.
-file("src/distribute/codec.gleam", 702).
-spec find_path_neighbors(
list({integer(),
integer(),
fun((bitstring()) -> {ok, bitstring()} | {error, decode_error()})}),
list(integer()),
list(integer()),
integer()
) -> {ok, list(integer())} | {error, decode_error()}.
find_path_neighbors(Edges, Path, Neighbors, Goal) ->
case Neighbors of
[] ->
{error, {migration_missing, Goal}};
[N | Rest] ->
case gleam@list:contains(Path, N) of
true ->
find_path_neighbors(Edges, Path, Rest, Goal);
false ->
case find_path(Edges, [N | Path], Goal) of
{ok, Found_path} ->
{ok, Found_path};
{error, _} ->
find_path_neighbors(Edges, Path, Rest, Goal)
end
end
end.
-file("src/distribute/codec.gleam", 673).
-spec find_path(
list({integer(),
integer(),
fun((bitstring()) -> {ok, bitstring()} | {error, decode_error()})}),
list(integer()),
integer()
) -> {ok, list(integer())} | {error, decode_error()}.
find_path(Edges, Path, Goal) ->
case Path of
[] ->
{error, {migration_missing, Goal}};
[Last | _] ->
case Last =:= Goal of
true ->
{ok, Path};
false ->
find_path_neighbors(
Edges,
Path,
gleam@list:fold(Edges, [], fun(Acc, E) -> case E of
{From, To, _} ->
case From =:= Last of
true ->
[To | Acc];
false ->
Acc
end
end end),
Goal
)
end
end.
-file("src/distribute/codec.gleam", 660).
?DOC(
" Build a migration graph from a list of migration edges.\n"
" Each edge is a single migration from `from` -> `to` (not necessarily +1).\n"
" The returned function will find a path from `from` to `to` (if any) and\n"
" apply each edge's migration in sequence. If no path exists, returns\n"
" `MigrationMissing(step)` where `step` is the first missing intermediate version.\n"
).
-spec build_migration_graph(
list({integer(),
integer(),
fun((bitstring()) -> {ok, bitstring()} | {error, decode_error()})})
) -> fun((integer(), integer(), bitstring()) -> {ok, bitstring()} |
{error, decode_error()}).
build_migration_graph(Edges) ->
fun(From, To, Payload) -> case find_path(Edges, [From], To) of
{ok, Path} ->
apply_path_edges(Edges, lists:reverse(Path), Payload);
{error, _} ->
{error, {migration_missing, From}}
end end.