-module(shelf_ffi). -export([ open_no_load/3, close/3, cleanup/3, insert/3, insert_list/3, insert_new/3, lookup_set/2, lookup_bag/2, member/2, delete_key/2, delete_object/3, delete_all/1, to_list/1, fold/3, size/1, save/2, sync_dets/1, sync_dets/2, update_counter/3, dets_fold_into_ets_strict/3, dets_insert/2, dets_insert_list/2, dets_delete_key/2, dets_delete_object/3, dets_delete_all/1, reload_atomic/3, validate_path/2, normalize_path/1 ]). %% ── DETS atom registry ────────────────────────────────────────────────── %% DETS requires atom names. To avoid unbounded atom creation from %% user-provided paths, we maintain a registry ETS table that maps %% path binaries to deterministic atoms from a bounded pool. %% %% The registry ETS table is owned by a dedicated long-lived process %% (shelf_dets_registry_owner) so it survives the death of any individual %% caller process. -define(REGISTRY, shelf_dets_registry). -define(POOL_SIZE, 65536). -define(MAX_COLLISION_ATTEMPTS, 100). ensure_registry() -> case ets:whereis(?REGISTRY) of undefined -> start_registry_owner(); _ -> ok end. start_registry_owner() -> Pid = spawn(fun registry_loop/0), try register(shelf_dets_registry_owner, Pid) of true -> Pid ! {create_registry, self()}, receive {registry_created, Pid} -> ok after 5000 -> error(registry_timeout) end catch error:badarg -> %% Another process won the race to register. Stop the orphan. Pid ! stop, wait_for_registry(20) end. registry_loop() -> receive {create_registry, From} -> try ets:new(?REGISTRY, [set, public, named_table, {keypos, 1}, {read_concurrency, true}]), From ! {registry_created, self()} catch _:badarg -> %% Table already exists (edge case), still ack. From ! {registry_created, self()} end, registry_loop(); stop -> ok; _ -> registry_loop() end. wait_for_registry(0) -> error(registry_not_available); wait_for_registry(N) -> case ets:whereis(?REGISTRY) of undefined -> timer:sleep(50), wait_for_registry(N - 1); _ -> ok end. %% Map a path binary to a bounded atom. Uses erlang:phash2 to hash %% the path into a fixed pool of atoms (shelf_dets_0 .. shelf_dets_N). %% Collisions are capped at MAX_COLLISION_ATTEMPTS to keep atom creation bounded. path_to_dets_name(Path) -> ensure_registry(), case ets:lookup(?REGISTRY, Path) of [{Path, Name}] -> Name; [] -> Hash = erlang:phash2(Path, ?POOL_SIZE), Name = find_available_name(Path, Hash, 0), %% Use insert_new to handle races — if another process %% registered this path first, use their name. case ets:insert_new(?REGISTRY, {Path, Name}) of true -> Name; false -> case ets:lookup(?REGISTRY, Path) of [{Path, ExistingName}] -> ExistingName; [] -> path_to_dets_name(Path) end end end. find_available_name(_Path, _Hash, Attempt) when Attempt >= ?MAX_COLLISION_ATTEMPTS -> error(atom_pool_exhausted); find_available_name(Path, Hash, Attempt) -> Candidate = list_to_atom("shelf_dets_" ++ integer_to_list(Hash) ++ "_" ++ integer_to_list(Attempt)), %% Check if this atom is already used by a different path case ets:match_object(?REGISTRY, {'_', Candidate}) of [] -> Candidate; [{Path, Candidate}] -> Candidate; %% Same path, reuse _ -> find_available_name(Path, Hash, Attempt + 1) %% Collision, try next end. %% Remove a path from the registry (called on close). unregister_dets_name(Path) -> ensure_registry(), ets:delete(?REGISTRY, Path), ok. %% ── Open (no load) ────────────────────────────────────────────────────── %% Creates an ETS table + opens a DETS file but does NOT load DETS into ETS. %% Used by the validated loading path where Gleam decodes entries before insertion. open_no_load(_Name, Path, TypeBin) -> Type = binary_to_atom(TypeBin, utf8), DetsName = path_to_dets_name(Path), try {ok, Dets} = dets:open_file(DetsName, [ {file, binary_to_list(Path)}, {type, Type}, {repair, true} ]), try Ets = ets:new(shelf_ets, [Type, protected, {keypos, 1}, {read_concurrency, true}]), %% Spawn a guardian to close DETS if the owning process dies. %% Safe to call erlang:monitor inside the spawned process: if %% OwnerPid is already dead when monitor/2 runs, it delivers %% an immediate 'DOWN' rather than silently dropping it. OwnerPid = self(), Guardian = spawn(fun() -> erlang:monitor(process, OwnerPid), receive {'DOWN', _, process, OwnerPid, _} -> _ = dets:close(Dets), unregister_dets_name(Path); stop -> ok end end), {ok, {Ets, Dets, Guardian}} catch _:badarg -> _ = dets:close(Dets), unregister_dets_name(Path), {error, {erlang_error, <<"Failed to create table">>}} end catch _:{badmatch, {error, Reason}} -> unregister_dets_name(Path), {error, translate_error(Reason)}; _:Reason -> unregister_dets_name(Path), {error, translate_error(Reason)} end. %% ── Streaming DETS → ETS loaders ──────────────────────────────────────── %% Validate and insert entries one at a time using dets:foldl, avoiding %% materializing the entire DETS contents into a Gleam list. %% To avoid row-by-row ETS boundary crossing, we batch entries. -define(LOAD_BATCH_SIZE, 5000). flush_batch(_Ets, []) -> ok; %% Reverse restores DETS traversal order before bulk insert. %% ETS bag tables preserve insertion order, so this matters for %% callers that expect values under a key to stay in DETS order. flush_batch(Ets, Batch) -> ets:insert(Ets, lists:reverse(Batch)). %% Abort on first decode failure using throw. %% DecoderFun takes a raw entry and returns {ok, Pair} or {error, Errors}. dets_fold_into_ets_strict(Dets, Ets, DecoderFun) -> try Result = dets:foldl( fun(Entry, {Count, Batch}) -> case DecoderFun(Entry) of {ok, Pair} -> NewBatch = [Pair | Batch], case Count + 1 of ?LOAD_BATCH_SIZE -> flush_batch(Ets, NewBatch), {0, []}; NewCount -> {NewCount, NewBatch} end; {error, Errors} -> throw({type_mismatch, Errors}) end end, {0, []}, Dets ), case Result of {error, Reason} -> {error, translate_error(Reason)}; {_, FinalBatch} -> flush_batch(Ets, FinalBatch), {ok, nil} end catch throw:{type_mismatch, Errors} -> {error, {type_mismatch, Errors}}; _:CatchReason -> {error, translate_error(CatchReason)} end. %% ── Atomic reload ─────────────────────────────────────────────────────── %% Load DETS entries into a scratch ETS table first. Only replace the %% live table contents after the full validation succeeds; on failure the %% live ETS is untouched. reload_atomic(Ets, Dets, DecoderFun) -> case check_owner(Ets) of {error, _} = Err -> Err; ok -> reload_atomic_validated(Ets, Dets, DecoderFun) end. reload_atomic_validated(Ets, Dets, DecoderFun) -> Type = ets:info(Ets, type), Scratch = ets:new(shelf_reload_scratch, [Type, protected]), try case dets_fold_into_ets_strict(Dets, Scratch, DecoderFun) of {ok, nil} -> %% Validation passed — swap contents. ets:delete_all_objects(Ets), ets:insert(Ets, ets:tab2list(Scratch)), ets:delete(Scratch), {ok, nil}; {error, _} = Err -> ets:delete(Scratch), Err end catch _:Reason -> (catch ets:delete(Scratch)), {error, translate_error(Reason)} end. %% ── Guardian ──────────────────────────────────────────────────────────── stop_guardian(Guardian) -> Guardian ! stop, ok. %% ── Cleanup ───────────────────────────────────────────────────────────── %% Delete ETS table and close DETS without saving. Used on validation failure. cleanup(Ets, Dets, Guardian) -> stop_guardian(Guardian), Path = try dets_to_path(Dets) catch _:_ -> undefined end, DetsResult = (catch dets:close(Dets)), _ = (catch ets:delete(Ets)), case Path of undefined -> ok; _ -> unregister_dets_name(Path) end, case DetsResult of ok -> {ok, nil}; {error, Reason} -> {error, translate_error(Reason)}; _ -> {ok, nil} end. %% ── Close ─────────────────────────────────────────────────────────────── %% Atomic save ETS→DETS via temp file, close DETS, delete ETS. %% On save failure, everything is left intact so the caller can retry. close(Ets, Dets, Guardian) -> case check_owner(Ets) of {error, _} = Err -> Err; ok -> Path = dets_to_path(Dets), case attempt_close_save(Ets, Dets) of ok -> finalize_close(Ets, Dets, Guardian, Path); {error, Reason} = Err -> case preserve_table_after_close_error(Path, Reason) of true -> Err; false -> teardown_resources(Ets, Dets, Guardian, Path), Err end end end. attempt_close_save(Ets, Dets) -> try save(Ets, Dets) of {ok, nil} -> ok; {error, Reason} -> {error, Reason} catch _:Reason -> {error, translate_error(Reason)} end. preserve_table_after_close_error(_Path, table_closed) -> false; preserve_table_after_close_error(undefined, _Reason) -> false; preserve_table_after_close_error(_Path, _Reason) -> true. finalize_close(Ets, Dets, Guardian, Path) -> stop_guardian(Guardian), CloseResult = close_dets_handle(Dets), _ = (catch ets:delete(Ets)), case Path of undefined -> ok; _ -> unregister_dets_name(Path) end, case CloseResult of ok -> {ok, nil}; {error, Reason} -> {error, Reason} end. teardown_resources(Ets, Dets, Guardian, Path) -> stop_guardian(Guardian), _ = close_dets_handle(Dets), _ = (catch ets:delete(Ets)), case Path of undefined -> ok; _ -> unregister_dets_name(Path) end, ok. close_dets_handle(Dets) -> try dets:close(Dets) of ok -> ok; {error, Reason} -> {error, translate_error(Reason)} catch _:Reason -> {error, translate_error(Reason)} end. %% Get the file path from a DETS reference as a binary. dets_to_path(Dets) -> case dets:info(Dets, filename) of undefined -> undefined; Filename -> list_to_binary(Filename) end. %% ── Insert ────────────────────────────────────────────────────────────── insert(Ets, _Dets, Object) -> try ets:insert(Ets, Object) of true -> {ok, nil} catch _:Reason -> {error, classify_ets_error(Ets, Reason)} end. insert_list(Ets, _Dets, Objects) -> try ets:insert(Ets, Objects) of true -> {ok, nil} catch _:Reason -> {error, classify_ets_error(Ets, Reason)} end. insert_new(Ets, _Dets, Object) -> %% _Dets is unused here — ETS insert_new only checks ETS. %% The Gleam caller handles DETS persistence for WriteThrough mode. try ets:insert_new(Ets, Object) of true -> {ok, nil}; false -> {error, key_already_present} catch _:Reason -> {error, classify_ets_error(Ets, Reason)} end. %% ── Lookup ────────────────────────────────────────────────────────────── %% Always reads from ETS (fast path). lookup_set(Ets, Key) -> try ets:lookup(Ets, Key) of [] -> {error, not_found}; [{_, Value} | _] -> {ok, Value} catch _:Reason -> {error, translate_error(Reason)} end. lookup_bag(Ets, Key) -> try ets:lookup(Ets, Key) of Results when is_list(Results) -> Values = [V || {_, V} <- Results], case Values of [] -> {error, not_found}; _ -> {ok, Values} end catch _:Reason -> {error, translate_error(Reason)} end. member(Ets, Key) -> try {ok, ets:member(Ets, Key)} catch _:Reason -> {error, translate_error(Reason)} end. %% ── Delete ────────────────────────────────────────────────────────────── delete_key(Ets, Key) -> try ets:delete(Ets, Key) of true -> {ok, nil} catch _:Reason -> {error, classify_ets_error(Ets, Reason)} end. delete_object(Ets, Key, Value) -> try ets:delete_object(Ets, {Key, Value}) of true -> {ok, nil} catch _:Reason -> {error, classify_ets_error(Ets, Reason)} end. delete_all(Ets) -> try ets:delete_all_objects(Ets) of true -> {ok, nil} catch _:Reason -> {error, classify_ets_error(Ets, Reason)} end. %% ── Query ─────────────────────────────────────────────────────────────── to_list(Ets) -> try {ok, ets:tab2list(Ets)} catch _:Reason -> {error, translate_error(Reason)} end. fold(Ets, Fun, Acc0) -> try ets:foldl(Fun, Acc0, Ets) of Result -> {ok, Result} catch _:Reason -> {error, translate_error(Reason)} end. size(Ets) -> try case ets:info(Ets, size) of undefined -> {error, table_closed}; Size -> {ok, Size} end catch _:Reason -> {error, translate_error(Reason)} end. %% ── Persistence ───────────────────────────────────────────────────────── %% Check that the caller is the ETS table owner. %% Returns ok | {error, not_owner | table_closed}. check_owner(Ets) -> case ets:info(Ets, owner) of undefined -> {error, table_closed}; Pid when Pid =:= self() -> ok; _ -> {error, not_owner} end. %% Atomic save: snapshot ETS to a temp DETS file, then rename over the original. %% This prevents data loss if the process is killed mid-save. save(Ets, Dets) -> case check_owner(Ets) of {error, _} = Err -> Err; ok -> try OrigPath = dets_to_path(Dets), case OrigPath of undefined -> {error, table_closed}; _ -> TmpPath = <>, Type = dets:info(Dets, type), safe_save_impl(Ets, Dets, OrigPath, TmpPath, Type) end catch _:Reason -> {error, translate_error(Reason)} end end. safe_save_impl(Ets, Dets, OrigPath, TmpPath, Type) -> TmpPathList = binary_to_list(TmpPath), OrigPathList = binary_to_list(OrigPath), TmpName = {shelf_tmp, make_ref()}, try %% 1. Open temp DETS {ok, TmpDets} = dets:open_file(TmpName, [ {file, TmpPathList}, {type, Type}, {repair, false} ]), %% 2. Snapshot ETS into temp DETS TmpDets = ets:to_dets(Ets, TmpDets), %% 3. Close temp (flushes to disk) ok = dets:close(TmpDets), %% 4. Close original DETS ok = dets:close(Dets), %% 5. Atomic rename (POSIX guarantees this is atomic) ok = file:rename(TmpPathList, OrigPathList), %% 6. Reopen DETS at original path with the same atom name DetsName = Dets, {ok, _} = dets:open_file(DetsName, [ {file, OrigPathList}, {type, Type}, {repair, true} ]), {ok, nil} catch _:Error -> %% Clean up temp file on failure _ = (catch dets:close(TmpName)), _ = (catch file:delete(TmpPathList)), %% Try to reopen original if it was closed _ = (catch dets:open_file(Dets, [ {file, OrigPathList}, {type, Type}, {repair, true} ])), {error, translate_error(Error)} end. %% Flush DETS write buffer to OS. sync_dets(Dets) -> try dets:sync(Dets) of ok -> {ok, nil}; {error, Reason} -> {error, translate_error(Reason)} catch _:Reason -> {error, translate_error(Reason)} end. %% Owner-guarded sync: only the ETS owner may flush DETS. sync_dets(Ets, Dets) -> case check_owner(Ets) of {error, _} = Err -> Err; ok -> sync_dets(Dets) end. %% ── Counters ──────────────────────────────────────────────────────────── update_counter(Ets, Key, Increment) -> try {ok, ets:update_counter(Ets, Key, Increment)} catch error:badarg -> case ets:info(Ets, owner) of undefined -> {error, table_closed}; OwnerPid when OwnerPid =:= self() -> %% Owner, so badarg is a data problem case ets:lookup(Ets, Key) of [] -> {error, not_found}; _ -> {error, {erlang_error, <<"update_counter failed: value is not an integer">>}} end; _OtherPid -> {error, not_owner} end; _:Reason -> {error, classify_ets_error(Ets, Reason)} end. %% ── Targeted DETS operations (for WriteThrough mode) ──────────────────── dets_insert(Dets, Object) -> try dets:insert(Dets, Object) of ok -> {ok, nil} catch _:Reason -> {error, translate_error(Reason)} end. dets_insert_list(Dets, Objects) -> try dets:insert(Dets, Objects) of ok -> {ok, nil} catch _:Reason -> {error, translate_error(Reason)} end. dets_delete_key(Dets, Key) -> try dets:delete(Dets, Key) of ok -> {ok, nil} catch _:Reason -> {error, translate_error(Reason)} end. dets_delete_object(Dets, Key, Value) -> try dets:delete_object(Dets, {Key, Value}) of ok -> {ok, nil} catch _:Reason -> {error, translate_error(Reason)} end. dets_delete_all(Dets) -> try dets:delete_all_objects(Dets) of ok -> {ok, nil} catch _:Reason -> {error, translate_error(Reason)} end. %% ── Error translation ────────────────────────────────────────────────── translate_error(not_found) -> not_found; translate_error(key_already_present) -> key_already_present; translate_error(name_conflict) -> name_conflict; translate_error(not_owner) -> not_owner; translate_error(type_mismatch) -> type_mismatch; translate_error({type_mismatch, Errors}) -> {type_mismatch, Errors}; translate_error({invalid_path, Msg}) -> {invalid_path, Msg}; translate_error(table_closed) -> table_closed; translate_error(badarg) -> table_closed; translate_error({file_error, _, enoent}) -> {file_error, <<"File not found">>}; translate_error({file_error, _, eacces}) -> {file_error, <<"Permission denied">>}; translate_error({file_error, _, enospc}) -> file_size_limit_exceeded; translate_error({file_error, _, Reason}) -> {file_error, list_to_binary(io_lib:format("~p", [Reason]))}; translate_error({error, Reason}) -> translate_error(Reason); translate_error(Reason) -> {erlang_error, list_to_binary(io_lib:format("~p", [Reason]))}. %% Classify an ETS badarg error: is the table closed, or is the caller %% not the owner? For `protected` tables, writes from non-owners raise %% badarg just like a deleted table does. classify_ets_error(Ets, badarg) -> case ets:info(Ets, owner) of undefined -> %% Table doesn't exist — genuinely closed / deleted table_closed; OwnerPid when OwnerPid =:= self() -> %% We are the owner but still got badarg — bad arguments {erlang_error, <<"ETS badarg: invalid arguments">>}; _OtherPid -> %% Table exists but we aren't the owner not_owner end; classify_ets_error(_Ets, Reason) -> translate_error(Reason). %% ── Path validation ──────────────────────────────────────────────────── validate_path(Path, BaseDirectory) -> BaseAbs = filename:absname(binary_to_list(BaseDirectory)), Resolved = filename:absname(binary_to_list(Path), BaseAbs), %% Normalize by splitting and rejoining (resolves . and ..) Normalized = normalize_path(Resolved), NormalizedBase = normalize_path(BaseAbs), %% Ensure the resolved path is inside the base directory. %% We must check for a directory boundary to prevent sibling directory %% bypass (e.g. base="/app/data" must not match "/app/data_sibling/x"). case Normalized =:= NormalizedBase orelse lists:prefix(NormalizedBase ++ "/", Normalized) of true -> {ok, list_to_binary(Normalized)}; false -> {error, {invalid_path, <<"Path escapes base directory">>}} end. normalize_path(Path) -> Parts = filename:split(Path), NormalizedParts = normalize_parts(Parts, []), filename:join(NormalizedParts). normalize_parts([], Acc) -> lists:reverse(Acc); normalize_parts(["." | Rest], Acc) -> normalize_parts(Rest, Acc); normalize_parts([".." | Rest], [_ | Acc]) -> normalize_parts(Rest, Acc); normalize_parts([".." | Rest], []) -> %% Already at root, ignore normalize_parts(Rest, []); normalize_parts([Part | Rest], Acc) -> normalize_parts(Rest, [Part | Acc]).