Packages
reckon_db
2.2.2
5.11.0
5.10.4
5.10.3
5.10.1
5.10.0
5.9.1
5.9.0
5.8.3
5.8.2
5.8.1
5.8.0
5.7.0
5.6.1
5.6.0
5.5.5
5.5.4
5.5.3
5.5.2
5.5.1
5.5.0
5.4.0
5.2.2
5.2.1
5.2.0
5.1.0
5.0.0
4.0.0
3.1.2
3.1.1
3.0.0
2.3.7
2.3.6
2.3.5
2.3.4
2.3.3
2.3.2
2.3.1
2.3.0
2.2.2
2.2.0
2.1.4
2.1.3
2.1.2
2.1.1
2.1.0
2.0.0
1.7.7
1.7.6
1.7.5
1.7.4
1.7.3
1.7.2
1.7.1
1.7.0
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.3
1.3.2
1.3.1
1.3.0
1.2.7
1.2.6
1.2.5
1.2.4
1.2.3
1.2.2
1.2.1
1.2.0
1.1.1
1.1.0
1.0.3
1.0.2
1.0.1
1.0.0
BEAM-native Event Store built on Khepri/Ra with Raft consensus. Event sourcing, persistent subscriptions, snapshots, and automatic cluster formation via UDP multicast discovery. Ships embedded Rust NIFs for 3-15x acceleration of crypto, hashing, compression, aggregation, filter matching, and grap...
Current section
Files
Jump to
Current section
Files
src/reckon_db_archive_backend.erl
%% @doc Archive backend behaviour for reckon-db
%%
%% Defines the interface for archive storage backends.
%% Implementations can store archived events in various formats:
%% - Local files
%% - S3/object storage
%% - Network storage
%%
%% @author rgfaber
-module(reckon_db_archive_backend).
%% API
-export([make_key/4, parse_key/1]).
%% Type alias for #event{} from reckon_db.hrl — kept local so
%% archive backend implementations don't need to pull in the full
%% record definition just to satisfy the callback signatures. Mirrors
%% the pattern in reckon_db_log_backend.
-type event() :: term().
-export_type([event/0]).
%%====================================================================
%% Callbacks
%%====================================================================
%% Initialize the archive backend with options.
-callback init(Opts :: map()) ->
{ok, State :: term()} | {error, Reason :: term()}.
%% Archive a list of events.
%% The archive key is typically based on store_id, stream_id, and timestamp range.
-callback archive(State :: term(), ArchiveKey :: binary(), Events :: [event()]) ->
{ok, NewState :: term()} | {error, Reason :: term()}.
%% Read archived events by key.
-callback read(State :: term(), ArchiveKey :: binary()) ->
{ok, [event()], NewState :: term()} | {error, Reason :: term()}.
%% List all archive keys for a stream.
-callback list(State :: term(), StoreId :: atom(), StreamId :: binary()) ->
{ok, [binary()], NewState :: term()} | {error, Reason :: term()}.
%% Delete an archive.
-callback delete(State :: term(), ArchiveKey :: binary()) ->
{ok, NewState :: term()} | {error, Reason :: term()}.
%% Check if an archive exists.
-callback exists(State :: term(), ArchiveKey :: binary()) ->
{boolean(), NewState :: term()}.
%%====================================================================
%% Utility functions
%%====================================================================
%% @doc Generate a standard archive key.
-spec make_key(atom(), binary(), integer(), integer()) -> binary().
make_key(StoreId, StreamId, FromVersion, ToVersion) ->
iolist_to_binary([
atom_to_binary(StoreId, utf8), <<"/">>,
StreamId, <<"/">>,
integer_to_binary(FromVersion), <<"-">>,
integer_to_binary(ToVersion), <<".archive">>
]).
%% @doc Parse an archive key to extract metadata.
-spec parse_key(binary()) -> {ok, map()} | {error, invalid_key}.
parse_key(Key) ->
case binary:split(Key, <<"/">>, [global]) of
[StoreIdBin, StreamId, VersionRange] ->
parse_key_with_range(StoreIdBin, StreamId, VersionRange);
_ ->
{error, invalid_key}
end.
parse_key_with_range(StoreIdBin, StreamId, VersionRange) ->
case parse_version_range(VersionRange) of
{ok, FromVersion, ToVersion} ->
{ok, #{
store_id => binary_to_atom(StoreIdBin, utf8),
stream_id => StreamId,
from_version => FromVersion,
to_version => ToVersion
}};
error ->
{error, invalid_key}
end.
%% @private Parse version range from "N-M.archive" format
-spec parse_version_range(binary()) -> {ok, integer(), integer()} | error.
parse_version_range(VersionRange) ->
case binary:split(VersionRange, <<".archive">>) of
[Range, <<>>] ->
parse_range_pair(Range);
_ ->
error
end.
%% @private
parse_range_pair(Range) ->
case binary:split(Range, <<"-">>) of
[FromBin, ToBin] ->
safe_to_integers(FromBin, ToBin);
_ ->
error
end.
%% @private
safe_to_integers(FromBin, ToBin) ->
try
{ok, binary_to_integer(FromBin), binary_to_integer(ToBin)}
catch
_:_ -> error
end.