% Licensed under the Apache License, Version 2.0 (the "License"); you may not
% use this file except in compliance with the License. You may obtain a copy of
% the License at
%
%   http://www.apache.org/licenses/LICENSE-2.0
%
% Unless required by applicable law or agreed to in writing, software
% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
% License for the specific language governing permissions and limitations under
% the License.

-module(erlfdb_directory).

-define(DOCATTRS, ?OTP_RELEASE >= 27).

-if(?DOCATTRS).
-moduledoc """
[Directory Layer](https://apple.github.io/foundationdb/developer-guide.html#directories)
""".
-endif.

-export([
    root/0,
    root/1,

    create_or_open/3,
    create_or_open/4,
    create/3,
    create/4,
    open/3,
    open/4,

    list/2,
    list/3,

    exists/2,
    exists/3,

    move/4,
    move_to/3,

    remove/2,
    remove/3,
    remove_if_exists/2,
    remove_if_exists/3,

    get_id/1,
    get_name/1,
    get_root/1,
    get_root_for_path/2,
    get_node_prefix/1,
    get_path/1,
    get_layer/1,
    get_subspace/1,

    subspace/2,
    key/1,
    pack/2,
    pack_vs/2,
    unpack/2,
    range/1,
    range/2,
    contains/2,

    debug_nodes/2
]).

-include("erlfdb.hrl").

-define(LAYER_VERSION, {1, 0, 0}).
-define(DEFAULT_NODE_PREFIX, <<16#FE>>).
-define(SUBDIRS, 0).

root() ->
    init_root([]).

root(Options) ->
    init_root(Options).

create_or_open(TxObj, Node, Path) ->
    create_or_open(TxObj, Node, Path, <<>>).

create_or_open(TxObj, Node, PathIn, Layer) ->
    {Root, Path} = adj_path(Node, PathIn),
    if
        Path /= [] -> ok;
        true -> ?ERLFDB_ERROR({open_error, cannot_open_root})
    end,
    case create_or_open_int(TxObj, Root, Path, Layer) of
        #{is_absolute_root := true} ->
            ?ERLFDB_ERROR({open_error, cannot_open_root});
        Else ->
            Else
    end.

create(TxObj, Node, Path) ->
    create(TxObj, Node, Path, []).

create(TxObj, Node, PathIn, Options) ->
    {Root, Path} = adj_path(Node, PathIn),
    check_manual_node_name(Root, Options),
    erlfdb:transactional(TxObj, fun(Tx) ->
        Layer = erlfdb_util:get(Options, layer, <<>>),
        NodeName = erlfdb_util:get(Options, node_name, undefined),
        create_int(Tx, Root, Path, Layer, NodeName)
    end).

open(TxObj, Node, Path) ->
    open(TxObj, Node, Path, []).

open(TxObj, Node, PathIn, Options) ->
    {Root, Path} = adj_path(Node, PathIn),
    if
        Path /= [] -> ok;
        true -> ?ERLFDB_ERROR({open_error, cannot_open_root})
    end,
    erlfdb:transactional(TxObj, fun(Tx) ->
        Layer = erlfdb_util:get(Options, layer, <<>>),
        open_int(Tx, Root, Path, Layer)
    end).

list(TxObj, Node) ->
    list(TxObj, Node, {}).

list(TxObj, Node, PathIn) ->
    {Root, Path} = adj_path(Node, PathIn),
    erlfdb:transactional(TxObj, fun(Tx) ->
        check_version(Tx, Root, read),
        case find(Tx, Root, Path) of
            not_found ->
                ?ERLFDB_ERROR({list_error, missing_path, Path});
            ListNode ->
                Subdirs = ?ERLFDB_EXTEND(get_id(ListNode), ?SUBDIRS),
                SDLen = size(Subdirs),
                SDStart = <<Subdirs:SDLen/binary, 16#00>>,
                SDEnd = <<Subdirs:SDLen/binary, 16#FF>>,
                SubDirKVs = erlfdb:wait(erlfdb:get_range(Tx, SDStart, SDEnd)),
                lists:map(
                    fun({Key, NodeName}) ->
                        {DName} = ?ERLFDB_EXTRACT(Subdirs, Key),
                        ChildNode = init_node(Tx, ListNode, NodeName, DName),
                        {DName, ChildNode}
                    end,
                    SubDirKVs
                )
        end
    end).

exists(TxObj, Node) ->
    exists(TxObj, Node, {}).

exists(TxObj, Node, PathIn) ->
    %Root = get_root(Node),
    Root = get_root_for_path(Node, PathIn),
    {Root, Path} = adj_path(Root, Node, PathIn),
    erlfdb:transactional(TxObj, fun(Tx) ->
        check_version(Tx, Root, read),
        case find(Tx, Root, Path) of
            not_found ->
                false;
            _ChildNode ->
                true
        end
    end).

move(TxObj, Node, OldPathIn, NewPathIn) ->
    {Root, OldPath} = adj_path(Node, OldPathIn),
    {Root, NewPath} = adj_path(Node, NewPathIn),
    erlfdb:transactional(TxObj, fun(Tx) ->
        check_version(Tx, Root, write),
        check_not_subpath(OldPath, NewPath),

        OldNode = find(Tx, Root, OldPath),
        NewNode = find(Tx, Root, NewPath),

        if
            OldNode /= not_found -> ok;
            true -> ?ERLFDB_ERROR({move_error, missing_source, OldPath})
        end,

        if
            NewNode == not_found -> ok;
            true -> ?ERLFDB_ERROR({move_error, target_exists, NewPath})
        end,

        {NewParentPath, [NewName]} = lists:split(length(NewPath) - 1, NewPath),
        case find(Tx, Root, NewParentPath) of
            not_found ->
                ?ERLFDB_ERROR({move_error, missing_parent_node, NewParentPath});
            NewParentNode ->
                check_same_partition(OldNode, NewParentNode),
                ParentId = get_id(NewParentNode),
                NodeEntryId = ?ERLFDB_PACK(ParentId, {?SUBDIRS, NewName}),
                erlfdb:set(Tx, NodeEntryId, get_name(OldNode)),
                remove_from_parent(Tx, OldNode),
                OldNode#{path := get_path(Root) ++ NewPath}
        end
    end).

move_to(_TxObj, #{is_absolute_root := true}, _NewPath) ->
    ?ERLFDB_ERROR({move_error, root_cannot_be_moved});
move_to(TxObj, Node, NewAbsPathIn) ->
    Root = get_root_for_path(Node, []),
    RootPath = get_path(Root),
    RootPathLen = length(RootPath),
    NewAbsPath = path_init(NewAbsPathIn),
    IsPrefix = lists:prefix(RootPath, NewAbsPath),
    if
        IsPrefix -> ok;
        true -> ?ERLFDB_ERROR({move_error, partition_mismatch, RootPath, NewAbsPath})
    end,
    NodePath = get_path(Node),
    SrcPath = lists:nthtail(RootPathLen, NodePath),
    TgtPath = lists:nthtail(RootPathLen, NewAbsPath),
    move(TxObj, Root, SrcPath, TgtPath).

remove(TxObj, Node) ->
    remove_int(TxObj, Node, {}, false).

remove(TxObj, Node, Path) ->
    remove_int(TxObj, Node, Path, false).

remove_if_exists(TxObj, Node) ->
    remove_int(TxObj, Node, {}, true).

remove_if_exists(TxObj, Node, Path) ->
    remove_int(TxObj, Node, Path, true).

get_id(Node) ->
    invoke(Node, get_id, []).

get_name(Node) ->
    invoke(Node, get_name, []).

get_root(Node) ->
    invoke(Node, get_root, []).

get_root_for_path(Node, Path) ->
    invoke(Node, get_root_for_path, [Path]).

get_partition(Node) ->
    invoke(Node, get_partition, []).

get_node_prefix(Node) ->
    invoke(Node, get_node_prefix, []).

get_path(Node) ->
    invoke(Node, get_path, []).

get_layer(Node) ->
    invoke(Node, get_layer, []).

get_subspace(Node) ->
    invoke(Node, get_subspace, []).

subspace(Node, Tuple) ->
    erlfdb_subspace:create(get_subspace(Node), Tuple).

key(Node) ->
    erlfdb_subspace:key(get_subspace(Node)).

pack(Node, Tuple) ->
    erlfdb_subspace:pack(get_subspace(Node), Tuple).

pack_vs(Node, Tuple) ->
    erlfdb_subspace:pack_vs(get_subspace(Node), Tuple).

unpack(Node, Key) ->
    erlfdb_subspace:unpack(get_subspace(Node), Key).

range(Node) ->
    range(Node, {}).

range(Node, Tuple) ->
    erlfdb_subspace:range(get_subspace(Node), Tuple).

contains(Node, Key) ->
    erlfdb_subspace:contains(get_subspace(Node), Key).

debug_nodes(TxObj, _Node) ->
    erlfdb:fold_range(
        TxObj,
        <<16#02>>,
        <<16#FF>>,
        fun({K, V}, _Acc) ->
            io:format(standard_error, "~s => ~s~n", [
                erlfdb_util:repr(K),
                erlfdb_util:repr(V)
            ])
        end,
        nil
    ).

invoke(not_found, _, _) ->
    erlang:error(broken);
invoke(Node, FunName, Args) ->
    case Node of
        #{FunName := Fun} ->
            erlang:apply(Fun, [Node | Args]);
        #{} ->
            ?ERLFDB_ERROR({op_not_supported, FunName, Node})
    end.

init_root(Options) ->
    DefNodePref = ?DEFAULT_NODE_PREFIX,
    NodePrefix = erlfdb_util:get(Options, node_prefix, DefNodePref),
    RootNodeId = ?ERLFDB_EXTEND(NodePrefix, NodePrefix),
    ContentPrefix = erlfdb_util:get(Options, content_prefix, <<>>),
    AllowManual = erlfdb_util:get(Options, allow_manual_names, false),
    Allocator = erlfdb_hca:create(?ERLFDB_EXTEND(RootNodeId, <<"hca">>)),
    #{
        id => ?ERLFDB_EXTEND(NodePrefix, NodePrefix),
        node_prefix => NodePrefix,
        content_prefix => ContentPrefix,
        allocator => Allocator,
        allow_manual_names => AllowManual,
        is_absolute_root => true,

        get_id => fun(Self) -> maps:get(id, Self) end,
        get_root => fun(Self) -> Self end,
        get_root_for_path => fun(Self, _Path) -> Self end,
        get_partition => fun(Self) -> Self end,
        get_node_prefix => fun(Self) -> maps:get(node_prefix, Self) end,
        get_path => fun(_Self) -> [] end,
        get_layer => fun(_Self) -> <<>> end,
        get_subspace => fun(_Self) ->
            ?ERLFDB_ERROR({subspace_error, subspace_unsupported_for_root})
        end
    }.

init_node(Tx, Node, NodeName, PathName) ->
    NodePrefix = get_node_prefix(Node),
    NodeLayerId = ?ERLFDB_PACK(NodePrefix, {NodeName, <<"layer">>}),
    Layer =
        case erlfdb:wait(erlfdb:get(Tx, NodeLayerId)) of
            not_found ->
                ?ERLFDB_ERROR({internal_error, missing_node_layer, NodeLayerId});
            LName ->
                LName
        end,
    case Layer of
        <<"partition">> ->
            init_partition(Node, NodeName, PathName);
        _ ->
            init_directory(Node, NodeName, PathName, Layer)
    end.

init_partition(ParentNode, NodeName, PathName) ->
    NodeNameLen = size(NodeName),
    NodePrefix = <<NodeName:NodeNameLen/binary, 16#FE>>,
    RootNodeId = ?ERLFDB_EXTEND(NodePrefix, NodePrefix),
    Allocator = erlfdb_hca:create(?ERLFDB_EXTEND(RootNodeId, <<"hca">>)),
    #{
        id => RootNodeId,
        name => NodeName,
        root => get_root(ParentNode),
        node_prefix => NodePrefix,
        content_prefix => NodeName,
        allocator => Allocator,
        allow_manual_names => false,
        path => path_append(get_path(ParentNode), PathName),
        is_partition => true,

        get_id => fun(Self) -> maps:get(id, Self) end,
        get_name => fun(Self) -> maps:get(name, Self) end,
        get_root => fun(Self) -> Self end,
        get_root_for_path => fun(Self, PathIn) ->
            case PathIn of
                {} -> maps:get(root, Self);
                [] -> maps:get(root, Self);
                _ -> Self
            end
        end,
        get_partition => fun(Self) -> maps:get(root, Self) end,
        get_node_prefix => fun(Self) -> maps:get(node_prefix, Self) end,
        get_path => fun(Self) -> maps:get(path, Self) end,
        get_layer => fun(_Self) -> <<"partition">> end,
        get_subspace => fun(_Self) ->
            ?ERLFDB_ERROR({subspace_error, subspace_unsupported_for_partition})
        end
    }.

init_directory(ParentNode, NodeName, PathName, Layer) ->
    NodePrefix = get_node_prefix(ParentNode),
    ParentPath = get_path(ParentNode),
    #{
        id => ?ERLFDB_EXTEND(NodePrefix, NodeName),
        name => NodeName,
        root => get_root(ParentNode),
        path => path_append(ParentPath, PathName),
        layer => Layer,

        get_id => fun(Self) -> maps:get(id, Self) end,
        get_name => fun(Self) -> maps:get(name, Self) end,
        get_root => fun(Self) -> maps:get(root, Self) end,
        get_root_for_path => fun(Self, Path) ->
            NewPath = maps:get(path, Self) ++ path_init(Path),
            get_root_for_path(maps:get(root, Self), NewPath)
        end,
        get_partition => fun(Self) -> maps:get(root, Self) end,
        get_node_prefix => fun(Self) ->
            Root = maps:get(root, Self),
            get_node_prefix(Root)
        end,
        get_path => fun(Self) -> maps:get(path, Self) end,
        get_layer => fun(Self) -> maps:get(layer, Self) end,
        get_subspace => fun(Self) ->
            erlfdb_subspace:create({}, maps:get(name, Self))
        end
    }.

find(_Tx, Node, []) ->
    Node;
find(Tx, Node, [PathName | RestPath]) ->
    NodeEntryId = ?ERLFDB_PACK(get_id(Node), {?SUBDIRS, PathName}),
    case erlfdb:wait(erlfdb:get(Tx, NodeEntryId)) of
        not_found ->
            not_found;
        ChildNodeName ->
            ChildNode = init_node(Tx, Node, ChildNodeName, PathName),
            find(Tx, ChildNode, RestPath)
    end.

find_deepest(_Tx, Node, []) ->
    Node;
find_deepest(Tx, Node, [PathName | RestPath]) ->
    NodeEntryId = ?ERLFDB_PACK(get_id(Node), {?SUBDIRS, PathName}),
    case erlfdb:wait(erlfdb:get(Tx, NodeEntryId)) of
        not_found ->
            Node;
        ChildNodeName ->
            ChildNode = init_node(Tx, Node, ChildNodeName, PathName),
            find_deepest(Tx, ChildNode, RestPath)
    end.

create_or_open_int(_TxObj, Node, [], LayerIn) ->
    Layer =
        case LayerIn of
            <<>> -> <<>>;
            null -> <<>>;
            undefined -> <<>>;
            Else when is_binary(Else) -> Else
        end,
    NodeLayer = get_layer(Node),
    if
        Layer == <<>> orelse Layer == NodeLayer -> ok;
        true -> ?ERLFDB_ERROR({open_error, layer_mismatch, Layer, NodeLayer})
    end,
    Node;
create_or_open_int(TxObj, Node, PathIn, Layer) ->
    {Root, Path} = adj_path(Node, PathIn),
    erlfdb:transactional(TxObj, fun(Tx) ->
        {ParentPath, [PathName]} = lists:split(length(Path) - 1, Path),

        Parent = lists:foldl(
            fun(Name, CurrNode) ->
                try
                    open_int(Tx, CurrNode, Name, <<>>)
                catch
                    error:{?MODULE, {open_error, path_missing, _}} ->
                        create_int(Tx, CurrNode, Name, <<>>, undefined)
                end
            end,
            Root,
            ParentPath
        ),

        try
            open_int(Tx, Parent, PathName, Layer)
        catch
            error:{?MODULE, {open_error, path_missing, _}} ->
                create_int(Tx, Parent, PathName, Layer, undefined)
        end
    end).

create_int(Tx, Node, PathIn, Layer, NodeNameIn) ->
    Path = path_init(PathIn),
    try
        open_int(Tx, Node, Path, <<>>),
        ?ERLFDB_ERROR({create_error, path_exists, Path})
    catch
        error:{?MODULE, {open_error, path_missing, _}} ->
            Deepest = find_deepest(Tx, Node, Path),
            NodeName = create_node_name(Tx, Deepest, NodeNameIn),
            {ParentPath, [PathName]} = lists:split(length(Path) - 1, Path),
            case create_or_open_int(Tx, Node, ParentPath, <<>>) of
                not_found ->
                    ?ERLFDB_ERROR({create_error, missing_parent, ParentPath});
                Parent ->
                    check_version(Tx, Parent, write),
                    create_node(Tx, Parent, PathName, NodeName, Layer),
                    R = find(Tx, Parent, [PathName]),
                    if
                        R /= not_found -> R;
                        true -> erlang:error(broken)
                    end
            end
    end.

create_node(Tx, Parent, PathName, NodeName, LayerIn) ->
    NodeEntryId = ?ERLFDB_PACK(get_id(Parent), {?SUBDIRS, PathName}),
    erlfdb:set(Tx, NodeEntryId, NodeName),

    NodePrefix = get_node_prefix(Parent),
    NodeLayerId = ?ERLFDB_PACK(NodePrefix, {NodeName, <<"layer">>}),
    Layer =
        if
            LayerIn == undefined -> <<>>;
            true -> LayerIn
        end,
    erlfdb:set(Tx, NodeLayerId, Layer).

open_int(Tx, Node, PathIn, Layer) ->
    check_version(Tx, Node, read),
    Path = path_init(PathIn),
    case find(Tx, Node, Path) of
        not_found ->
            ?ERLFDB_ERROR({open_error, path_missing, Path});
        #{is_absolute_root := true} ->
            ?ERLFDB_ERROR({open_error, cannot_open_root});
        Opened ->
            NodeLayer = get_layer(Opened),
            if
                Layer == <<>> orelse Layer == NodeLayer -> ok;
                true -> ?ERLFDB_ERROR({open_error, layer_mismatch, Layer, NodeLayer})
            end,
            Opened
    end.

remove_int(TxObj, Node, PathIn, IgnoreMissing) ->
    Root = get_root_for_path(Node, PathIn),
    {Root, Path} = adj_path(Root, Node, PathIn),
    erlfdb:transactional(TxObj, fun(Tx) ->
        check_version(Tx, Root, write),
        case find(Tx, Root, Path) of
            not_found when IgnoreMissing ->
                ok;
            not_found ->
                ?ERLFDB_ERROR({remove_error, path_missing, Path});
            #{is_absolute_root := true} ->
                ?ERLFDB_ERROR({remove_error, cannot_remove_root});
            ToRem ->
                remove_recursive(Tx, ToRem),
                remove_from_parent(Tx, ToRem)
        end
    end).

remove_recursive(Tx, Node) ->
    % Remove all subdirectories
    lists:foreach(
        fun({_DirName, ChildNode}) ->
            remove_recursive(Tx, ChildNode)
        end,
        list(Tx, Node)
    ),

    % Delete all content for the node.
    ContentSS = erlfdb_subspace:create({}, get_name(Node)),
    {ContentStart, ContentEnd} = erlfdb_subspace:range(ContentSS),
    erlfdb:clear_range(Tx, ContentStart, ContentEnd),

    % Delete this node from the tree hierarchy
    NodeSubspace = erlfdb_subspace:create({}, get_id(Node)),
    {NodeStart, NodeEnd} = erlfdb_subspace:range(NodeSubspace),
    erlfdb:clear_range(Tx, NodeStart, NodeEnd).

remove_from_parent(Tx, Node) ->
    {Root, Path} = adj_path(get_root_for_path(Node, []), Node, []),
    {ParentPath, [PathName]} = lists:split(length(Path) - 1, Path),
    Parent = find(Tx, Root, ParentPath),

    NodeEntryId = ?ERLFDB_PACK(get_id(Parent), {?SUBDIRS, PathName}),
    erlfdb:clear(Tx, NodeEntryId).

check_manual_node_name(Root, Options) ->
    AllowManual = maps:get(allow_manual_names, Root),
    IsManual = lists:keyfind(node_name, 1, Options) /= false,
    if
        not (IsManual and not AllowManual) -> ok;
        true -> ?ERLFDB_ERROR({create_error, manual_node_names_prohibited})
    end.

create_node_name(Tx, Parent, NameIn) ->
    #{
        content_prefix := ContentPrefix,
        allow_manual_names := AllowManual,
        allocator := Allocator
    } = get_root(Parent),
    Name =
        case NameIn of
            null -> undefined;
            undefined -> undefined;
            _ when is_binary(NameIn) -> NameIn
        end,
    case Name of
        _ when Name == undefined ->
            BaseId = erlfdb_hca:allocate(Allocator, Tx),
            CPLen = size(ContentPrefix),
            NewName = <<ContentPrefix:CPLen/binary, BaseId/binary>>,

            KeysExist = erlfdb:wait(erlfdb:get_range_startswith(Tx, NewName, [{limit, 1}])),
            if
                KeysExist == [] ->
                    ok;
                true ->
                    ?ERLFDB_ERROR({
                        create_error,
                        keys_exist_for_allocated_name,
                        NewName
                    })
            end,

            IsFree = is_prefix_free(erlfdb:snapshot(Tx), Parent, NewName),
            if
                IsFree ->
                    ok;
                true ->
                    ?ERLFDB_ERROR({
                        create_error,
                        manual_names_conflict_with_allocated_name,
                        NewName
                    })
            end,

            NewName;
        _ when AllowManual andalso is_binary(Name) ->
            case is_prefix_free(Tx, Parent, NameIn) of
                true ->
                    ok;
                false ->
                    ?ERLFDB_ERROR({create_error, node_name_in_use, NameIn})
            end,
            NameIn;
        _ ->
            ?ERLFDB_ERROR({create_error, manual_node_names_prohibited})
    end.

is_prefix_free(Tx, Parent, NodeName) ->
    % We have to make sure that NodeName does not interact with
    % anything that currently exists in the tree. This means that
    % it must not be a prefix of any existing node id and also
    % that no existing node id is a prefix of this NodeName.
    %
    % A motivating example for why is that deletion of nodes
    % in the tree would end up deleting unrelated portions
    % of the tree when node ids overlapped. There would also
    % be other badness if keys overlapped with the layer
    % or ?SUBDIRS spaces.

    try
        % An empty name would obviously be kind of bonkers.
        if
            NodeName /= <<>> -> ok;
            true -> throw(false)
        end,

        Root = get_root(Parent),
        RootId = get_id(Root),
        NodePrefix = get_node_prefix(Root),
        NPLen = size(NodePrefix),

        % First check that the special case of the root node
        case bin_startswith(NodeName, RootId) of
            true -> throw(false);
            false -> ok
        end,

        % Check if any node id is a prefix of NodeName
        Start1 = <<NodePrefix:NPLen/binary, 16#00>>,
        End1 = ?ERLFDB_PACK(NodePrefix, {NodeName, null}),
        Opts1 = [{reverse, true}, {limit, 1}, {streaming_mode, exact}],
        Subspace = erlfdb_subspace:create({}, get_node_prefix(Parent)),
        erlfdb:fold_range(
            Tx,
            Start1,
            End1,
            fun({Key, _} = _E, _) ->
                KeyNodeId = element(1, erlfdb_subspace:unpack(Subspace, Key)),
                case bin_startswith(NodeName, KeyNodeId) of
                    true -> throw(false);
                    false -> ok
                end
            end,
            nil,
            Opts1
        ),

        % Check if NodeName is a prefix of any existing key
        Start2 = ?ERLFDB_EXTEND(NodePrefix, NodeName),
        End2 = ?ERLFDB_EXTEND(NodePrefix, erlfdb_key:strinc(NodeName)),
        Opts2 = [{limit, 1}, {streaming_mode, exact}],
        case erlfdb:wait(erlfdb:get_range(Tx, Start2, End2, Opts2)) of
            [_E | _] -> throw(false);
            [] -> ok
        end,

        true
    catch
        throw:false ->
            false
    end.

bin_startswith(Subject, Prefix) ->
    PrefixLen = size(Prefix),
    case Subject of
        <<Prefix:PrefixLen/binary, _/binary>> -> true;
        _ -> false
    end.

check_version(Tx, Node, PermLevel) ->
    Root = get_root(Node),
    VsnKey = ?ERLFDB_EXTEND(get_id(Root), <<"version">>),
    {LV1, LV2, _LV3} = ?LAYER_VERSION,
    {Major, Minor, Patch} =
        case erlfdb:wait(erlfdb:get(Tx, VsnKey)) of
            not_found when PermLevel == write ->
                initialize_directory(Tx, VsnKey);
            not_found ->
                ?LAYER_VERSION;
            VsnBin ->
                <<
                    V1:32/little-unsigned,
                    V2:32/little-unsigned,
                    V3:32/little-unsigned
                >> = VsnBin,
                {V1, V2, V3}
        end,

    Path = get_path(Node),

    if
        Major =< LV1 -> ok;
        true -> ?ERLFDB_ERROR({version_error, unreadable, Path, {Major, Minor, Patch}})
    end,

    if
        not (Minor > LV2 andalso PermLevel /= read) -> ok;
        true -> ?ERLFDB_ERROR({version_error, unwritable, Path, {Major, Minor, Patch}})
    end.

initialize_directory(Tx, VsnKey) ->
    {V1, V2, V3} = ?LAYER_VERSION,
    Packed = <<
        V1:32/little-unsigned,
        V2:32/little-unsigned,
        V3:32/little-unsigned
    >>,
    erlfdb:set(Tx, VsnKey, Packed),
    ?LAYER_VERSION.

check_same_partition(OldNode, NewParentNode) ->
    OldRoot = get_partition(OldNode),
    NewRoot = get_root(NewParentNode),
    if
        NewRoot == OldRoot -> ok;
        true -> ?ERLFDB_ERROR({move_error, partition_mismatch, OldRoot, NewRoot})
    end.

adj_path(Node, PathIn) ->
    adj_path(get_root(Node), Node, PathIn).

adj_path(Root, Node, PathIn) ->
    RootPathLen = length(get_path(Root)),
    NodePath = get_path(Node),
    NodeRelPath = lists:nthtail(RootPathLen, NodePath),
    Path = NodeRelPath ++ path_init(PathIn),
    {Root, Path}.

path_init(<<_/binary>> = Bin) ->
    check_utf8(0, Bin),
    [{utf8, Bin}];
path_init({utf8, <<_/binary>> = Bin} = Path) ->
    check_utf8(0, Bin),
    [Path];
path_init(Path) when is_list(Path) ->
    lists:flatmap(
        fun(Part) ->
            path_init(Part)
        end,
        Path
    );
path_init(Path) when is_tuple(Path) ->
    path_init(tuple_to_list(Path));
path_init(Else) ->
    ?ERLFDB_ERROR({path_error, invalid_path_component, Else}).

check_utf8(Offset, Binary) ->
    case Binary of
        <<_:Offset/binary>> ->
            true;
        <<_:Offset/binary, _/utf8, Rest/binary>> ->
            % Recalculating offset as a subtraction here is
            % slightly odd but this is to avoid having to
            % re-encode the utf8 code point and adding the
            % size of that new binary.
            check_utf8(size(Binary) - size(Rest), Binary);
        <<_:Offset/binary, _/binary>> ->
            ?ERLFDB_ERROR({path_error, invalid_utf8, Binary})
    end.

path_append(Path, Part) ->
    Path ++ path_init(Part).

check_not_subpath(OldPath, NewPath) ->
    case lists:prefix(OldPath, NewPath) of
        true ->
            ?ERLFDB_ERROR({
                move_error,
                target_is_subdirectory,
                OldPath,
                NewPath
            });
        false ->
            ok
    end.