Current section

Files

Jump to
reckon_db src reckon_db_archive_file.erl
Raw

src/reckon_db_archive_file.erl

%% @doc File-based archive backend for reckon-db
%%
%% Stores archived events as Erlang term files on the local filesystem.
%% Archives are organized in directories by store and stream.
%%
%% Directory structure:
%% ```
%% {base_dir}/
%% {store_id}/
%% {stream_id}/
%% 0-99.archive
%% 100-199.archive
%% '''
%%
%% @author rgfaber
-module(reckon_db_archive_file).
-behaviour(reckon_db_archive_backend).
-include("reckon_db.hrl").
%% Behaviour callbacks
-export([
init/1,
archive/3,
read/2,
list/3,
delete/2,
exists/2
]).
-record(state, {
base_dir :: string()
}).
%%====================================================================
%% Behaviour callbacks
%%====================================================================
%% @doc Initialize the file archive backend.
%%
%% Options:
%% - base_dir: Directory where archives are stored (required)
-spec init(map()) -> {ok, #state{}} | {error, term()}.
init(Opts) ->
BaseDir = maps:get(base_dir, Opts),
case filelib:ensure_dir(BaseDir ++ "/") of
ok ->
{ok, #state{base_dir = BaseDir}};
{error, Reason} ->
{error, {failed_to_create_dir, Reason}}
end.
%% @doc Archive events to a file.
-spec archive(#state{}, binary(), [event()]) -> {ok, #state{}} | {error, term()}.
archive(#state{base_dir = BaseDir} = State, ArchiveKey, Events) ->
FilePath = key_to_path(BaseDir, ArchiveKey),
case filelib:ensure_dir(FilePath) of
ok ->
write_archive(State, FilePath, Events);
{error, Reason} ->
{error, {failed_to_create_dir, Reason}}
end.
write_archive(State, FilePath, Events) ->
Data = #{
version => 1,
created_at => erlang:system_time(millisecond),
event_count => length(Events),
events => Events
},
case file:write_file(FilePath, term_to_binary(Data, [compressed])) of
ok ->
{ok, State};
{error, Reason} ->
{error, {write_failed, Reason}}
end.
%% @doc Read events from an archive file.
-spec read(#state{}, binary()) -> {ok, [event()], #state{}} | {error, term()}.
read(#state{base_dir = BaseDir} = State, ArchiveKey) ->
FilePath = key_to_path(BaseDir, ArchiveKey),
case file:read_file(FilePath) of
{ok, Binary} ->
decode_archive(Binary, State, ArchiveKey);
{error, enoent} ->
{error, {archive_not_found, ArchiveKey}};
{error, Reason} ->
{error, {read_failed, Reason}}
end.
decode_archive(Binary, State, ArchiveKey) ->
try
#{events := Events} = binary_to_term(Binary),
{ok, Events, State}
catch
_:_ ->
{error, {corrupted_archive, ArchiveKey}}
end.
%% @doc List all archive keys for a stream.
-spec list(#state{}, atom(), binary()) -> {ok, [binary()], #state{}} | {error, term()}.
list(#state{base_dir = BaseDir} = State, StoreId, StreamId) ->
StreamDir = filename:join([BaseDir, atom_to_list(StoreId), binary_to_list(StreamId)]),
case file:list_dir(StreamDir) of
{ok, Files} ->
%% Filter for .archive files and convert to keys
ArchiveFiles = [F || F <- Files, filename:extension(F) =:= ".archive"],
Keys = [path_to_key(StoreId, StreamId, F) || F <- ArchiveFiles],
{ok, Keys, State};
{error, enoent} ->
{ok, [], State};
{error, Reason} ->
{error, {list_failed, Reason}}
end.
%% @doc Delete an archive file.
-spec delete(#state{}, binary()) -> {ok, #state{}} | {error, term()}.
delete(#state{base_dir = BaseDir} = State, ArchiveKey) ->
FilePath = key_to_path(BaseDir, ArchiveKey),
case file:delete(FilePath) of
ok ->
{ok, State};
{error, enoent} ->
{ok, State}; %% Already deleted
{error, Reason} ->
{error, {delete_failed, Reason}}
end.
%% @doc Check if an archive exists.
-spec exists(#state{}, binary()) -> {boolean(), #state{}}.
exists(#state{base_dir = BaseDir} = State, ArchiveKey) ->
FilePath = key_to_path(BaseDir, ArchiveKey),
{filelib:is_regular(FilePath), State}.
%%====================================================================
%% Internal functions
%%====================================================================
%% @private Convert archive key to file path
-spec key_to_path(string(), binary()) -> string().
key_to_path(BaseDir, ArchiveKey) ->
%% Key format: "store_id/stream_id/N-M.archive"
filename:join(BaseDir, binary_to_list(ArchiveKey)).
%% @private Convert file name to archive key
-spec path_to_key(atom(), binary(), string()) -> binary().
path_to_key(StoreId, StreamId, FileName) ->
iolist_to_binary([
atom_to_binary(StoreId, utf8), <<"/">>,
StreamId, <<"/">>,
list_to_binary(FileName)
]).