% 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(fabric2_fdb).


-export([
    transactional/1,
    transactional/3,
    transactional/2,

    with_snapshot/2,

    create/2,
    open/2,
    ensure_current/1,
    delete/1,
    undelete/3,
    remove_deleted_db/2,
    exists/1,

    get_dir/1,

    list_dbs/4,
    list_dbs_info/4,
    list_deleted_dbs_info/4,

    get_info/1,
    get_info_future/2,
    get_info_wait/1,
    set_config/3,

    get_stat/2,
    incr_stat/3,
    incr_stat/4,

    get_all_revs/2,
    get_all_revs_future/2,
    get_winning_revs/3,
    get_winning_revs_future/3,
    get_revs_wait/2,
    get_non_deleted_rev/3,

    get_doc_body/3,
    get_doc_body_future/3,
    get_doc_body_wait/4,

    get_local_doc_rev_future/2,
    get_local_doc_rev_wait/1,
    get_local_doc_body_future/3,
    get_local_doc_body_wait/4,
    get_local_doc/2,
    get_local_doc_rev/3,

    write_doc/6,
    write_local_doc/2,

    read_attachment/3,
    write_attachment/4,

    get_last_change/1,

    fold_range/5,

    vs_to_seq/1,
    seq_to_vs/1,
    next_vs/1,

    new_versionstamp/1,

    get_approximate_tx_size/1,

    chunkify_binary/1,
    chunkify_binary/2,

    debug_cluster/0,
    debug_cluster/2
]).


-include_lib("couch/include/couch_db.hrl").
-include("fabric2.hrl").


-define(MAX_FOLD_RANGE_RETRIES, 3).


-record(fold_acc, {
    db,
    restart_tx,
    start_key,
    end_key,
    limit,
    skip,
    retries,
    base_opts,
    user_fun,
    user_acc
}).

-record(info_future, {
    tx,
    db_prefix,
    changes_future,
    meta_future,
    uuid_future,
    retries = 0
}).


transactional(Fun) ->
    do_transaction(Fun, undefined).


transactional(DbName, Options, Fun) when is_binary(DbName) ->
    with_span(Fun, #{'db.name' => DbName}, fun() ->
        transactional(fun(Tx) ->
            Fun(init_db(Tx, DbName, Options))
        end)
    end).


transactional(#{tx := undefined} = Db, Fun) ->
    DbName = maps:get(name, Db, undefined),
    try
        Db1 = refresh(Db),
        Reopen = maps:get(reopen, Db1, false),
        Db2 = maps:remove(reopen, Db1),
        LayerPrefix = case Reopen of
            true -> undefined;
            false -> maps:get(layer_prefix, Db2)
        end,
        with_span(Fun, #{'db.name' => DbName}, fun() ->
            do_transaction(fun(Tx) ->
                case Reopen of
                    true -> Fun(reopen(Db2#{tx => Tx}));
                    false -> Fun(Db2#{tx => Tx})
                end
            end, LayerPrefix)
        end)
    catch throw:{?MODULE, reopen} ->
        with_span('db.reopen', #{'db.name' => DbName}, fun() ->
            transactional(Db#{reopen => true}, Fun)
        end)
    end;
transactional(#{tx := {erlfdb_snapshot, _}} = Db, Fun) ->
    DbName = maps:get(name, Db, undefined),
    with_span(Fun, #{'db.name' => DbName}, fun() ->
        Fun(Db)
    end);

transactional(#{tx := {erlfdb_transaction, _}} = Db, Fun) ->
    DbName = maps:get(name, Db, undefined),
    with_span(Fun, #{'db.name' => DbName}, fun() ->
        Fun(Db)
    end).


do_transaction(Fun, LayerPrefix) when is_function(Fun, 1) ->
    Db = get_db_handle(),
    try
        erlfdb:transactional(Db, fun(Tx) ->
            case get(erlfdb_trace) of
                Name when is_binary(Name) ->
                    UId = erlang:unique_integer([positive]),
                    UIdBin = integer_to_binary(UId, 36),
                    TxId = <<Name/binary, "_", UIdBin/binary>>,
                    erlfdb:set_option(Tx, transaction_logging_enable, TxId);
                _ ->
                    ok
            end,
            case is_transaction_applied(Tx) of
                true ->
                    get_previous_transaction_result();
                false ->
                    execute_transaction(Tx, Fun, LayerPrefix)
            end
        end)
    after
        clear_transaction()
    end.


with_snapshot(#{tx := {erlfdb_transaction, _} = Tx} = TxDb, Fun) ->
    SSDb = TxDb#{tx := erlfdb:snapshot(Tx)},
    Fun(SSDb);

with_snapshot(#{tx := {erlfdb_snapshot, _}} = SSDb, Fun) ->
    Fun(SSDb).


create(#{} = Db0, Options) ->
    #{
        name := DbName,
        tx := Tx,
        layer_prefix := LayerPrefix
    } = Db1 = ensure_current(Db0, false),

    DbKey = erlfdb_tuple:pack({?ALL_DBS, DbName}, LayerPrefix),
    HCA = erlfdb_hca:create(erlfdb_tuple:pack({?DB_HCA}, LayerPrefix)),
    AllocPrefix = erlfdb_hca:allocate(HCA, Tx),
    DbPrefix = erlfdb_tuple:pack({?DBS, AllocPrefix}, LayerPrefix),
    erlfdb:set(Tx, DbKey, DbPrefix),

    % This key is responsible for telling us when something in
    % the database cache (i.e., fabric2_server's ets table) has
    % changed and requires re-loading. This currently includes
    % revs_limit and validate_doc_update functions. There's
    % no order to versioning here. Its just a value that changes
    % that is used in the ensure_current check.
    DbVersionKey = erlfdb_tuple:pack({?DB_VERSION}, DbPrefix),
    DbVersion = fabric2_util:uuid(),
    erlfdb:set(Tx, DbVersionKey, DbVersion),

    UUID = fabric2_util:uuid(),

    Defaults = [
        {?DB_CONFIG, <<"uuid">>, UUID},
        {?DB_CONFIG, <<"revs_limit">>, ?uint2bin(1000)},
        {?DB_CONFIG, <<"security_doc">>, <<"{}">>},
        {?DB_STATS, <<"doc_count">>, ?uint2bin(0)},
        {?DB_STATS, <<"doc_del_count">>, ?uint2bin(0)},
        {?DB_STATS, <<"doc_design_count">>, ?uint2bin(0)},
        {?DB_STATS, <<"doc_local_count">>, ?uint2bin(0)},
        {?DB_STATS, <<"sizes">>, <<"external">>, ?uint2bin(2)},
        {?DB_STATS, <<"sizes">>, <<"views">>, ?uint2bin(0)}
    ],
    lists:foreach(fun
        ({P, K, V}) ->
            Key = erlfdb_tuple:pack({P, K}, DbPrefix),
            erlfdb:set(Tx, Key, V);
        ({P, S, K, V}) ->
            Key = erlfdb_tuple:pack({P, S, K}, DbPrefix),
            erlfdb:set(Tx, Key, V)
    end, Defaults),

    UserCtx = fabric2_util:get_value(user_ctx, Options, #user_ctx{}),
    Options1 = lists:keydelete(user_ctx, 1, Options),

    Db2 = Db1#{
        uuid => UUID,
        db_prefix => DbPrefix,
        db_version => DbVersion,

        revs_limit => 1000,
        security_doc => {[]},
        user_ctx => UserCtx,
        check_current_ts => erlang:monotonic_time(millisecond),

        validate_doc_update_funs => [],
        before_doc_update => undefined,
        after_doc_read => undefined,
        % All other db things as we add features,

        db_options => Options1,
        interactive => false
    },
    aegis:init_db(Db2, Options).


open(#{} = Db0, Options) ->
    #{
        name := DbName,
        tx := Tx,
        layer_prefix := LayerPrefix
    } = Db1 = ensure_current(Db0, false),

    DbKey = erlfdb_tuple:pack({?ALL_DBS, DbName}, LayerPrefix),
    DbPrefix = case erlfdb:wait(erlfdb:get(Tx, DbKey)) of
        Bin when is_binary(Bin) -> Bin;
        not_found -> erlang:error(database_does_not_exist)
    end,

    DbVersionKey = erlfdb_tuple:pack({?DB_VERSION}, DbPrefix),
    DbVersion = erlfdb:wait(erlfdb:get(Tx, DbVersionKey)),

    UserCtx = fabric2_util:get_value(user_ctx, Options, #user_ctx{}),
    Options1 = lists:keydelete(user_ctx, 1, Options),

    UUID = fabric2_util:get_value(uuid, Options1),
    Options2 = lists:keydelete(uuid, 1, Options1),

    Interactive = fabric2_util:get_value(interactive, Options2, false),
    Options3 = lists:keydelete(interactive, 1, Options2),

    Db2 = Db1#{
        db_prefix => DbPrefix,
        db_version => DbVersion,

        uuid => <<>>,
        revs_limit => 1000,
        security_doc => {[]},

        user_ctx => UserCtx,
        check_current_ts => erlang:monotonic_time(millisecond),

        % Place holders until we implement these
        % bits.
        validate_doc_update_funs => [],
        before_doc_update => undefined,
        after_doc_read => undefined,

        db_options => Options3,
        interactive => Interactive
    },

    Db3 = load_config(Db2),
    Db4 = aegis:open_db(Db3),

    case {UUID, Db4} of
        {undefined, _} -> ok;
        {<<_/binary>>, #{uuid := UUID}} -> ok;
        {<<_/binary>>, #{uuid := _}} -> erlang:error(database_does_not_exist)
    end,

    load_validate_doc_funs(Db4).


% Match on `name` in the function head since some non-fabric2 db
% objects might not have names and so they don't get cached
refresh(#{tx := undefined, name := DbName} = Db) ->
    #{
        uuid := UUID,
        md_version := OldVer
    } = Db,

    case fabric2_server:fetch(DbName, UUID) of
        % Relying on these assumptions about the `md_version` value:
        %  - It is bumped every time `db_version` is bumped
        %  - Is a versionstamp, so we can check which one is newer
        %  - If it is `not_found`, it would sort less than a binary value
        #{md_version := Ver} = Db1 when Ver > OldVer ->
            Db1#{
                user_ctx := maps:get(user_ctx, Db),
                security_fun := maps:get(security_fun, Db),
                interactive := maps:get(interactive, Db)
            };
        _ ->
            Db
    end;

refresh(#{} = Db) ->
    Db.


reopen(#{} = OldDb) ->
    require_transaction(OldDb),
    #{
        tx := Tx,
        name := DbName,
        uuid := UUID,
        db_options := Options,
        user_ctx := UserCtx,
        security_fun := SecurityFun,
        interactive := Interactive
    } = OldDb,
    Options1 = lists:keystore(user_ctx, 1, Options, {user_ctx, UserCtx}),
    NewDb = open(init_db(Tx, DbName, Options1), Options1),

    % Check if database was re-created
    case {Interactive, maps:get(uuid, NewDb)} of
        {true, _} -> ok;
        {false, UUID} -> ok;
        {false, _OtherUUID} -> error(database_does_not_exist)
    end,

    NewDb#{security_fun := SecurityFun, interactive := Interactive}.


delete(#{} = Db) ->
    DoRecovery = fabric2_util:do_recovery(),
    case DoRecovery of
        true -> soft_delete_db(Db);
        false -> hard_delete_db(Db)
    end.


undelete(#{} = Db0, TgtDbName, TimeStamp) ->
    #{
        name := DbName,
        tx := Tx,
        layer_prefix := LayerPrefix
    } = ensure_current(Db0, false),
    DbKey = erlfdb_tuple:pack({?ALL_DBS, TgtDbName}, LayerPrefix),
    case erlfdb:wait(erlfdb:get(Tx, DbKey)) of
        Bin when is_binary(Bin) ->
            file_exists;
        not_found ->
            DeletedDbTupleKey = {
                ?DELETED_DBS,
                DbName,
                TimeStamp
            },
            DeleteDbKey = erlfdb_tuple:pack(DeletedDbTupleKey, LayerPrefix),
            case erlfdb:wait(erlfdb:get(Tx, DeleteDbKey)) of
                not_found ->
                    not_found;
                DbPrefix ->
                    erlfdb:set(Tx, DbKey, DbPrefix),
                    erlfdb:clear(Tx, DeleteDbKey),
                    bump_db_version(#{
                        tx => Tx,
                        db_prefix => DbPrefix
                    }),
                    ok
            end
    end.


remove_deleted_db(#{} = Db0, TimeStamp) ->
    #{
        name := DbName,
        tx := Tx,
        layer_prefix := LayerPrefix
    } = ensure_current(Db0, false),

    DeletedDbTupleKey = {
        ?DELETED_DBS,
        DbName,
        TimeStamp
    },
    DeletedDbKey = erlfdb_tuple:pack(DeletedDbTupleKey, LayerPrefix),
    case erlfdb:wait(erlfdb:get(Tx, DeletedDbKey)) of
        not_found ->
            not_found;
        DbPrefix ->
            erlfdb:clear(Tx, DeletedDbKey),
            erlfdb:clear_range_startswith(Tx, DbPrefix),
            bump_db_version(#{
                tx => Tx,
                db_prefix => DbPrefix
            }),
            ok
    end.


exists(#{name := DbName} = Db) when is_binary(DbName) ->
    #{
        tx := Tx,
        layer_prefix := LayerPrefix
    } = ensure_current(Db, false),

    DbKey = erlfdb_tuple:pack({?ALL_DBS, DbName}, LayerPrefix),
    case erlfdb:wait(erlfdb:get(Tx, DbKey)) of
        Bin when is_binary(Bin) -> true;
        not_found -> false
    end.


get_dir(Tx) ->
    Root = erlfdb_directory:root(),
    Dir = fabric2_server:fdb_directory(),
    CouchDB = erlfdb_directory:create_or_open(Tx, Root, Dir),
    erlfdb_directory:get_name(CouchDB).


list_dbs(Tx, Callback, AccIn, Options0) ->
    Options = case fabric2_util:get_value(restart_tx, Options0) of
        undefined -> [{restart_tx, true} | Options0];
        _AlreadySet -> Options0
    end,
    LayerPrefix = get_dir(Tx),
    Prefix = erlfdb_tuple:pack({?ALL_DBS}, LayerPrefix),
    fold_range({tx, Tx}, Prefix, fun({K, _V}, Acc) ->
        {DbName} = erlfdb_tuple:unpack(K, Prefix),
        Callback(DbName, Acc)
    end, AccIn, Options).


list_dbs_info(Tx, Callback, AccIn, Options0) ->
    Options = case fabric2_util:get_value(restart_tx, Options0) of
        undefined -> [{restart_tx, true} | Options0];
        _AlreadySet -> Options0
    end,
    LayerPrefix = get_dir(Tx),
    Prefix = erlfdb_tuple:pack({?ALL_DBS}, LayerPrefix),
    fold_range({tx, Tx}, Prefix, fun({DbNameKey, DbPrefix}, Acc) ->
        {DbName} = erlfdb_tuple:unpack(DbNameKey, Prefix),
        InfoFuture = get_info_future(Tx, DbPrefix),
        Callback(DbName, InfoFuture, Acc)
    end, AccIn, Options).


list_deleted_dbs_info(Tx, Callback, AccIn, Options0) ->
    Options = case fabric2_util:get_value(restart_tx, Options0) of
        undefined -> [{restart_tx, true} | Options0];
        _AlreadySet -> Options0
    end,
    LayerPrefix = get_dir(Tx),
    Prefix = erlfdb_tuple:pack({?DELETED_DBS}, LayerPrefix),
    fold_range({tx, Tx}, Prefix, fun({DbKey, DbPrefix}, Acc) ->
        {DbName, TimeStamp} = erlfdb_tuple:unpack(DbKey, Prefix),
        InfoFuture = get_info_future(Tx, DbPrefix),
        Callback(DbName, TimeStamp, InfoFuture, Acc)
    end, AccIn, Options).


get_info(#{} = Db) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = ensure_current(Db),
    DbInfo = get_info_wait(get_info_future(Tx, DbPrefix)),
    AegisProps = aegis:get_db_info(Db),
    [{encryption, {AegisProps}} | DbInfo].


get_info_future(Tx, DbPrefix) ->
    {CStart, CEnd} = erlfdb_tuple:range({?DB_CHANGES}, DbPrefix),
    ChangesFuture = erlfdb:get_range(Tx, CStart, CEnd, [
        {streaming_mode, exact},
        {limit, 1},
        {reverse, true}
    ]),

    UUIDKey = erlfdb_tuple:pack({?DB_CONFIG, <<"uuid">>}, DbPrefix),
    UUIDFuture = erlfdb:get(Tx, UUIDKey),

    StatsPrefix = erlfdb_tuple:pack({?DB_STATS}, DbPrefix),
    MetaFuture = erlfdb:get_range_startswith(Tx, StatsPrefix),

    % Save the tx object only if it's read-only as we might retry to get the
    % future again after the tx was reset
    SaveTx = case erlfdb:get_writes_allowed(Tx) of
        true -> undefined;
        false -> Tx
    end,

    #info_future{
        tx = SaveTx,
        db_prefix = DbPrefix,
        changes_future = ChangesFuture,
        meta_future = MetaFuture,
        uuid_future = UUIDFuture
    }.


get_info_wait(#info_future{tx = Tx, retries = Retries} = Future)
        when Tx =:= undefined orelse Retries >= 2 ->
    get_info_wait_int(Future);

get_info_wait(#info_future{tx = Tx, retries = Retries} = Future) ->
    try
        get_info_wait_int(Future)
    catch
        error:{erlfdb_error, ?TRANSACTION_CANCELLED} ->
            Future1 = get_info_future(Tx, Future#info_future.db_prefix),
            get_info_wait(Future1#info_future{retries = Retries + 1});
        error:{erlfdb_error, ?TRANSACTION_TOO_OLD} ->
            ok = erlfdb:reset(Tx),
            Future1 = get_info_future(Tx, Future#info_future.db_prefix),
            get_info_wait(Future1#info_future{retries = Retries + 1})
    end.


load_config(#{} = Db) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = Db,

    {Start, End} = erlfdb_tuple:range({?DB_CONFIG}, DbPrefix),
    Future = erlfdb:get_range(Tx, Start, End),

    lists:foldl(fun({K, V}, DbAcc) ->
        {?DB_CONFIG, Key} = erlfdb_tuple:unpack(K, DbPrefix),
        case Key of
            <<"uuid">> ->  DbAcc#{uuid := V};
            <<"revs_limit">> -> DbAcc#{revs_limit := ?bin2uint(V)};
            <<"security_doc">> -> DbAcc#{security_doc := ?JSON_DECODE(V)}
        end
    end, Db, erlfdb:wait(Future)).


set_config(#{} = Db0, Key, Val) when is_atom(Key) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = Db = ensure_current(Db0),
    {BinKey, BinVal} = case Key of
        uuid -> {<<"uuid">>, Val};
        revs_limit -> {<<"revs_limit">>, ?uint2bin(max(1, Val))};
        security_doc -> {<<"security_doc">>, ?JSON_ENCODE(Val)}
    end,
    DbKey = erlfdb_tuple:pack({?DB_CONFIG, BinKey}, DbPrefix),
    erlfdb:set(Tx, DbKey, BinVal),
    {ok, DbVersion} = bump_db_version(Db),
    {ok, Db#{db_version := DbVersion, Key := Val}}.


get_stat(#{} = Db, StatKey) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = ensure_current(Db),

    Key = erlfdb_tuple:pack({?DB_STATS, StatKey}, DbPrefix),

    % Might need to figure out some sort of type
    % system here. Uints are because stats are all
    % atomic op adds for the moment.
    ?bin2uint(erlfdb:wait(erlfdb:get(Tx, Key))).


incr_stat(_Db, _StatKey, 0) ->
    ok;

incr_stat(#{} = Db, StatKey, Increment) when is_integer(Increment) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = ensure_current(Db),

    Key = erlfdb_tuple:pack({?DB_STATS, StatKey}, DbPrefix),
    erlfdb:add(Tx, Key, Increment).


incr_stat(_Db, _Section, _Key, 0) ->
    ok;

incr_stat(#{} = Db, Section, Key, Increment) when is_integer(Increment) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = ensure_current(Db),

    BinKey = erlfdb_tuple:pack({?DB_STATS, Section, Key}, DbPrefix),
    erlfdb:add(Tx, BinKey, Increment).


get_all_revs(#{} = Db, DocId) ->
    DbName = maps:get(name, Db, undefined),
    with_span('db.get_all_revs', #{'db.name' => DbName, 'doc.id' => DocId}, fun() ->
        Future = get_all_revs_future(Db, DocId),
        get_revs_wait(Db, Future)
    end).


get_all_revs_future(#{} = Db, DocId) ->
    Options = [{streaming_mode, want_all}],
    get_revs_future(Db, DocId, Options).


get_winning_revs(Db, DocId, NumRevs) ->
    DbName = maps:get(name, Db, undefined),
    with_span('db.get_winning_revs', #{'db.name' => DbName, 'doc.id' => DocId}, fun() ->
        Future = get_winning_revs_future(Db, DocId, NumRevs),
        get_revs_wait(Db, Future)
    end).


get_winning_revs_future(#{} = Db, DocId, NumRevs) ->
    Options = [{reverse, true}, {limit, NumRevs}],
    get_revs_future(Db, DocId, Options).


get_revs_future(#{} = Db, DocId, Options) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = ensure_current(Db),

    {StartKey, EndKey} = erlfdb_tuple:range({?DB_REVS, DocId}, DbPrefix),
    erlfdb:fold_range_future(Tx, StartKey, EndKey, Options).


get_revs_wait(#{} = Db, RangeFuture) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = ensure_current(Db),

    RevRows = erlfdb:fold_range_wait(Tx, RangeFuture, fun({K, V}, Acc) ->
        Key = erlfdb_tuple:unpack(K, DbPrefix),
        Val = erlfdb_tuple:unpack(V),
        [fdb_to_revinfo(Key, Val) | Acc]
    end, []),
    lists:reverse(RevRows).


get_non_deleted_rev(#{} = Db, DocId, RevId) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = ensure_current(Db),

    {RevPos, Rev} = RevId,

    BaseKey = {?DB_REVS, DocId, true, RevPos, Rev},
    Key = erlfdb_tuple:pack(BaseKey, DbPrefix),
    case erlfdb:wait(erlfdb:get(Tx, Key)) of
        not_found ->
            not_found;
        Val ->
            fdb_to_revinfo(BaseKey, erlfdb_tuple:unpack(Val))
    end.


get_doc_body(Db, DocId, RevInfo) ->
    DbName = maps:get(name, Db, undefined),
    with_span('db.get_doc_body', #{'db.name' => DbName, 'doc.id' => DocId}, fun() ->
        Future = get_doc_body_future(Db, DocId, RevInfo),
        get_doc_body_wait(Db, DocId, RevInfo, Future)
    end).


get_doc_body_future(#{} = Db, DocId, RevInfo) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = ensure_current(Db),

    #{
        rev_id := {RevPos, Rev}
    } = RevInfo,

    Key = {?DB_DOCS, DocId, RevPos, Rev},
    {StartKey, EndKey} = erlfdb_tuple:range(Key, DbPrefix),
    erlfdb:fold_range_future(Tx, StartKey, EndKey, []).


get_doc_body_wait(#{} = Db0, DocId, RevInfo, Future) ->
    #{
        tx := Tx
    } = Db = ensure_current(Db0),

    #{
        rev_id := {RevPos, Rev},
        rev_path := RevPath
    } = RevInfo,

    FoldFun = aegis:wrap_fold_fun(Db, fun({_K, V}, Acc) ->
        [V | Acc]
    end),
    RevBodyRows = erlfdb:fold_range_wait(Tx, Future, FoldFun, []),
    BodyRows = lists:reverse(RevBodyRows),

    fdb_to_doc(Db, DocId, RevPos, [Rev | RevPath], BodyRows).


get_local_doc_rev_future(Db, DocId) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = ensure_current(Db),

    Key = erlfdb_tuple:pack({?DB_LOCAL_DOCS, DocId}, DbPrefix),
    erlfdb:get(Tx, Key).


get_local_doc_rev_wait(Future) ->
    erlfdb:wait(Future).


get_local_doc_body_future(#{} = Db, DocId, _Rev) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = ensure_current(Db),

    Prefix = erlfdb_tuple:pack({?DB_LOCAL_DOC_BODIES, DocId}, DbPrefix),
    erlfdb:get_range_startswith(Tx, Prefix).


get_local_doc_body_wait(#{} = Db0, DocId, Rev, Future) ->
    Db = ensure_current(Db0),

    {_, Chunks} = lists:unzip(aegis:decrypt(Db, erlfdb:wait(Future))),
    fdb_to_local_doc(Db, DocId, Rev, Chunks).


get_local_doc(#{} = Db, <<?LOCAL_DOC_PREFIX, _/binary>> = DocId) ->
    RevFuture = get_local_doc_rev_future(Db, DocId),
    Rev = get_local_doc_rev_wait(RevFuture),

    BodyFuture = get_local_doc_body_future(Db, DocId, Rev),
    get_local_doc_body_wait(Db, DocId, Rev, BodyFuture).


get_local_doc_rev(_Db0, <<?LOCAL_DOC_PREFIX, _/binary>> = DocId, Val) ->
    case Val of
        <<255, RevBin/binary>> ->
            % Versioned local docs
            try
                case erlfdb_tuple:unpack(RevBin) of
                    {?CURR_LDOC_FORMAT, Rev, _Size} -> Rev
                end
            catch _:_ ->
                erlang:error({invalid_local_doc_rev, DocId, Val})
            end;
        <<131, _/binary>> ->
            % Compatibility clause for an older encoding format
            try binary_to_term(Val, [safe]) of
                {Rev, _} -> Rev;
                _ -> erlang:error({invalid_local_doc_rev, DocId, Val})
            catch
                error:badarg ->
                    erlang:error({invalid_local_doc_rev, DocId, Val})
            end;
        <<_/binary>> ->
            try binary_to_integer(Val) of
                IntVal when IntVal >= 0 ->
                    Val;
                _ ->
                    erlang:error({invalid_local_doc_rev, DocId, Val})
            catch
                error:badarg ->
                    erlang:error({invalid_local_doc_rev, DocId, Val})
            end
    end.


write_doc(#{} = Db0, Doc, NewWinner0, OldWinner, ToUpdate, ToRemove) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = Db = ensure_current(Db0),

    #doc{
        id = DocId,
        deleted = Deleted,
        atts = Atts
    } = Doc,

    % Doc body

    ok = write_doc_body(Db, Doc),

    % Attachment bookkeeping

    % If a document's attachments have changed we have to scan
    % for any attachments that may need to be deleted. The check
    % for `>= 2` is a bit subtle. The important point is that
    % one of the revisions will be from the new document so we
    % have to find at least one more beyond that to assert that
    % the attachments have not changed.
    AttHash = fabric2_util:hash_atts(Atts),
    RevsToCheck = [NewWinner0] ++ ToUpdate ++ ToRemove,
    AttHashCount = lists:foldl(fun(Att, Count) ->
        #{att_hash := RevAttHash} = Att,
        case RevAttHash == AttHash of
            true -> Count + 1;
            false -> Count
        end
    end, 0, RevsToCheck),
    if
        AttHashCount == length(RevsToCheck) ->
            ok;
        AttHashCount >= 2 ->
            ok;
        true ->
            cleanup_attachments(Db, DocId, Doc, ToRemove)
    end,

    % Revision tree

    NewWinner = NewWinner0#{
        winner := true
    },
    NewRevId = maps:get(rev_id, NewWinner),

    {WKey, WVal, WinnerVS} = revinfo_to_fdb(Tx, DbPrefix, DocId, NewWinner),
    ok = erlfdb:set_versionstamped_value(Tx, WKey, WVal),

    lists:foreach(fun(RI0) ->
        RI = RI0#{winner := false},
        {K, V, undefined} = revinfo_to_fdb(Tx, DbPrefix, DocId, RI),
        ok = erlfdb:set(Tx, K, V)
    end, ToUpdate),

    lists:foreach(fun(RI0) ->
        RI = RI0#{winner := false},
        {K, _, undefined} = revinfo_to_fdb(Tx, DbPrefix, DocId, RI),
        ok = erlfdb:clear(Tx, K),
        ok = clear_doc_body(Db, DocId, RI0)
    end, ToRemove),

    % _all_docs

    UpdateStatus = case {OldWinner, NewWinner} of
        {not_found, #{deleted := false}} ->
            created;
        {not_found, #{deleted := true}} ->
            replicate_deleted;
        {#{deleted := true}, #{deleted := false}} ->
            recreated;
        {#{deleted := false}, #{deleted := false}} ->
            updated;
        {#{deleted := false}, #{deleted := true}} ->
            deleted;
        {#{deleted := true}, #{deleted := true}} ->
            ignore
    end,

    case UpdateStatus of
        replicate_deleted ->
            ok;
        ignore ->
            ok;
        deleted ->
            ADKey = erlfdb_tuple:pack({?DB_ALL_DOCS, DocId}, DbPrefix),
            ok = erlfdb:clear(Tx, ADKey);
        _ ->
            ADKey = erlfdb_tuple:pack({?DB_ALL_DOCS, DocId}, DbPrefix),
            ADVal = erlfdb_tuple:pack(NewRevId),
            ok = erlfdb:set(Tx, ADKey, ADVal)
    end,

    % _changes

    if OldWinner == not_found -> ok; true ->
        OldSeq = maps:get(sequence, OldWinner),
        OldSeqKey = erlfdb_tuple:pack({?DB_CHANGES, OldSeq}, DbPrefix),
        erlfdb:clear(Tx, OldSeqKey)
    end,

    NewSeqKey = erlfdb_tuple:pack_vs({?DB_CHANGES, WinnerVS}, DbPrefix),
    NewSeqVal = erlfdb_tuple:pack({DocId, Deleted, NewRevId}),
    erlfdb:set_versionstamped_key(Tx, NewSeqKey, NewSeqVal),

    % Bump db version on design doc changes

    IsDDoc = case Doc#doc.id of
        <<?DESIGN_DOC_PREFIX, _/binary>> -> true;
        _ -> false
    end,

    if not IsDDoc -> ok; true ->
        bump_db_version(Db)
    end,

    % Update our document counts

    case UpdateStatus of
        created ->
            if not IsDDoc -> ok; true ->
                incr_stat(Db, <<"doc_design_count">>, 1)
            end,
            incr_stat(Db, <<"doc_count">>, 1);
        recreated ->
            if not IsDDoc -> ok; true ->
                incr_stat(Db, <<"doc_design_count">>, 1)
            end,
            incr_stat(Db, <<"doc_count">>, 1),
            incr_stat(Db, <<"doc_del_count">>, -1);
        replicate_deleted ->
            incr_stat(Db, <<"doc_del_count">>, 1);
        ignore ->
            ok;
        deleted ->
            if not IsDDoc -> ok; true ->
                incr_stat(Db, <<"doc_design_count">>, -1)
            end,
            incr_stat(Db, <<"doc_count">>, -1),
            incr_stat(Db, <<"doc_del_count">>, 1);
        updated ->
            ok
    end,

    fabric2_db_plugin:after_doc_write(Db, Doc, NewWinner, OldWinner,
        NewRevId, WinnerVS),

    % Update database size
    AddSize = sum_add_rev_sizes([NewWinner | ToUpdate]),
    RemSize = sum_rem_rev_sizes(ToRemove),
    incr_stat(Db, <<"sizes">>, <<"external">>, AddSize - RemSize),

    ok.


write_local_doc(#{} = Db0, Doc) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = Db = ensure_current(Db0),

    Id = Doc#doc.id,

    {LDocKey, LDocVal, NewSize, Rows} = local_doc_to_fdb(Db, Doc),

    {WasDeleted, PrevSize} = case erlfdb:wait(erlfdb:get(Tx, LDocKey)) of
        <<255, RevBin/binary>> ->
            case erlfdb_tuple:unpack(RevBin) of
                {?CURR_LDOC_FORMAT, _Rev, Size} ->
                    {false, Size}
            end;
        <<_/binary>> ->
            {false, 0};
        not_found ->
            {true, 0}
    end,

    BPrefix = erlfdb_tuple:pack({?DB_LOCAL_DOC_BODIES, Id}, DbPrefix),

    case Doc#doc.deleted of
        true ->
            erlfdb:clear(Tx, LDocKey),
            erlfdb:clear_range_startswith(Tx, BPrefix);
        false ->
            erlfdb:set(Tx, LDocKey, LDocVal),
            % Make sure to clear the whole range, in case there was a larger
            % document body there before.
            erlfdb:clear_range_startswith(Tx, BPrefix),
            lists:foreach(fun({K, V}) ->
                erlfdb:set(Tx, K, aegis:encrypt(Db, K, V))
            end, Rows)
    end,

    case {WasDeleted, Doc#doc.deleted} of
        {true, false} ->
            incr_stat(Db, <<"doc_local_count">>, 1);
        {false, true} ->
            incr_stat(Db, <<"doc_local_count">>, -1);
        _ ->
            ok
    end,

    incr_stat(Db, <<"sizes">>, <<"external">>, NewSize - PrevSize),

    ok.


read_attachment(#{} = Db, DocId, AttId) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = ensure_current(Db),

    AttKey = erlfdb_tuple:pack({?DB_ATTS, DocId, AttId}, DbPrefix),
    Data = case erlfdb:wait(erlfdb:get_range_startswith(Tx, AttKey)) of
        not_found ->
            throw({not_found, missing});
        KVs ->
            {_, Chunks} = lists:unzip(aegis:decrypt(Db, KVs)),
            iolist_to_binary(Chunks)
    end,

    IdKey = erlfdb_tuple:pack({?DB_ATT_NAMES, DocId, AttId}, DbPrefix),
    case erlfdb:wait(erlfdb:get(Tx, IdKey)) of
        <<>> ->
            Data; % Old format, before CURR_ATT_STORAGE_VER = 0
        <<_/binary>> = InfoBin ->
            {?CURR_ATT_STORAGE_VER, Compressed} = erlfdb_tuple:unpack(InfoBin),
            case Compressed of
                true -> binary_to_term(Data, [safe]);
                false -> Data
            end
    end.


write_attachment(#{} = Db, DocId, Data, Encoding)
        when is_binary(Data), is_atom(Encoding) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = ensure_current(Db),

    AttId = fabric2_util:uuid(),

    {Data1, Compressed} = case Encoding of
        gzip ->
            {Data, false};
        _ ->
            Opts = [{minor_version, 1}, {compressed, 6}],
            CompressedData = term_to_binary(Data, Opts),
            case size(CompressedData) < Data of
                true -> {CompressedData, true};
                false -> {Data, false}
            end
    end,

    IdKey = erlfdb_tuple:pack({?DB_ATT_NAMES, DocId, AttId}, DbPrefix),
    InfoVal = erlfdb_tuple:pack({?CURR_ATT_STORAGE_VER, Compressed}),
    ok = erlfdb:set(Tx, IdKey, InfoVal),

    Chunks = chunkify_binary(Data1),

    lists:foldl(fun(Chunk, ChunkId) ->
        AttKey = erlfdb_tuple:pack({?DB_ATTS, DocId, AttId, ChunkId}, DbPrefix),
        ok = erlfdb:set(Tx, AttKey, aegis:encrypt(Db, AttKey, Chunk)),
        ChunkId + 1
    end, 0, Chunks),
    {ok, AttId}.


get_last_change(#{} = Db) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = ensure_current(Db),

    {Start, End} = erlfdb_tuple:range({?DB_CHANGES}, DbPrefix),
    Options = [{limit, 1}, {reverse, true}],
    case erlfdb:get_range(Tx, Start, End, Options) of
        [] ->
            vs_to_seq(fabric2_util:seq_zero_vs());
        [{K, _V}] ->
            {?DB_CHANGES, SeqVS} = erlfdb_tuple:unpack(K, DbPrefix),
            vs_to_seq(SeqVS)
    end.


fold_range(TxOrDb, RangePrefix, UserFun, UserAcc, Options) ->
    {Db, Tx} = case TxOrDb of
        {tx, TxObj} ->
            {undefined, TxObj};
        #{} = DbObj ->
            DbObj1 = #{tx := TxObj} = ensure_current(DbObj),
            {DbObj1, TxObj}
    end,
    % FoundationDB treats a limit 0 of as unlimited so we guard against it
    case fabric2_util:get_value(limit, Options) of 0 -> UserAcc; _ ->
        FAcc = get_fold_acc(Db, RangePrefix, UserFun, UserAcc, Options),
        try
            fold_range(Tx, FAcc)
        after
            erase(?PDICT_FOLD_ACC_STATE)
        end
    end.


fold_range(Tx, FAcc) ->
    #fold_acc{
        start_key = Start,
        end_key = End,
        limit = Limit,
        base_opts = BaseOpts,
        restart_tx = DoRestart
    } = FAcc,
    case DoRestart of false -> ok; true ->
        ok = erlfdb:set_option(Tx, disallow_writes)
    end,
    Opts = [{limit, Limit} | BaseOpts],
    Callback = fun fold_range_cb/2,
    try
        #fold_acc{
            user_acc = FinalUserAcc
        } = erlfdb:fold_range(Tx, Start, End, Callback, FAcc, Opts),
        FinalUserAcc
    catch error:{erlfdb_error, ?TRANSACTION_TOO_OLD} when DoRestart ->
        % Possibly handle cluster_version_changed and future_version as well to
        % continue iteration instead fallback to transactional and retrying
        % from the beginning which is bound to fail when streaming data out to a
        % socket.
        fold_range(Tx, restart_fold(Tx, FAcc))
    end.


vs_to_seq(VS) when is_tuple(VS) ->
    % 51 is the versionstamp type tag
    <<51:8, SeqBin:12/binary>> = erlfdb_tuple:pack({VS}),
    fabric2_util:to_hex(SeqBin).


seq_to_vs(Seq) when is_binary(Seq) ->
    Seq1 = fabric2_util:from_hex(Seq),
    % 51 is the versionstamp type tag
    Seq2 = <<51:8, Seq1/binary>>,
    {VS} = erlfdb_tuple:unpack(Seq2),
    VS.


next_vs({versionstamp, VS, Batch, TxId}) ->
    {V, B, T} = case TxId =< 65535 of
        true ->
            {VS, Batch, TxId + 1};
        false ->
            case Batch =< 65535 of
                true ->
                    {VS, Batch + 1, 0};
                false ->
                    {VS + 1, 0, 0}
            end
    end,
    {versionstamp, V, B, T}.


new_versionstamp(Tx) ->
    TxId = erlfdb:get_next_tx_id(Tx),
    {versionstamp, 16#FFFFFFFFFFFFFFFF, 16#FFFF, TxId}.


get_approximate_tx_size(#{} = TxDb) ->
    require_transaction(TxDb),
    #{tx := Tx} = TxDb,
    erlfdb:wait(erlfdb:get_approximate_size(Tx)).


chunkify_binary(Data) ->
    chunkify_binary(Data, binary_chunk_size()).


chunkify_binary(Data, Size) ->
    case Data of
        <<>> ->
            [];
        <<Head:Size/binary, Rest/binary>> ->
            [Head | chunkify_binary(Rest, Size)];
        <<_/binary>> when size(Data) < Size ->
            [Data]
    end.


debug_cluster() ->
    debug_cluster(<<>>, <<16#FE, 16#FF, 16#FF>>).


debug_cluster(Start, End) ->
    transactional(fun(Tx) ->
        lists:foreach(fun({Key, Val}) ->
            io:format(standard_error, "~s => ~s~n", [
                    string:pad(erlfdb_util:repr(Key), 60),
                    erlfdb_util:repr(Val)
                ])
        end, erlfdb:get_range(Tx, Start, End))
    end).


init_db(Tx, DbName, Options) ->
    Prefix = get_dir(Tx),
    Version = erlfdb:wait(erlfdb:get(Tx, ?METADATA_VERSION_KEY)),
    #{
        name => DbName,
        tx => Tx,
        layer_prefix => Prefix,
        md_version => Version,

        security_fun => undefined,
        db_options => Options
    }.


load_validate_doc_funs(#{} = Db) ->
    FoldFun = fun
        ({row, Row}, Acc) ->
            DDocInfo = #{id => fabric2_util:get_value(id, Row)},
            {ok, [DDocInfo | Acc]};
        (_, Acc) ->
            {ok, Acc}
    end,

    Options = [
        {start_key, <<"_design/">>},
        {end_key, <<"_design0">>}
    ],

    {ok, Infos1} = fabric2_db:fold_docs(Db, FoldFun, [], Options),

    Infos2 = lists:map(fun(Info) ->
        #{
            id := DDocId = <<"_design/", _/binary>>
        } = Info,
        Info#{
            rev_info => get_winning_revs_future(Db, DDocId, 1)
        }
    end, Infos1),

    Infos3 = lists:flatmap(fun(Info) ->
        #{
            id := DDocId,
            rev_info := RevInfoFuture
        } = Info,
        [RevInfo] = get_revs_wait(Db, RevInfoFuture),
        #{deleted := Deleted} = RevInfo,
        if Deleted -> []; true ->
            [Info#{
                rev_info := RevInfo,
                body => get_doc_body_future(Db, DDocId, RevInfo)
            }]
        end
    end, Infos2),

    VDUs = lists:flatmap(fun(Info) ->
        #{
            id := DDocId,
            rev_info := RevInfo,
            body := BodyFuture
        } = Info,
        #doc{} = Doc = get_doc_body_wait(Db, DDocId, RevInfo, BodyFuture),
        case couch_doc:get_validate_doc_fun(Doc) of
            nil -> [];
            Fun -> [Fun]
        end
    end, Infos3),

    Db#{
        validate_doc_update_funs := VDUs
    }.


bump_metadata_version(Tx) ->
    % The 14 zero bytes is pulled from the PR for adding the
    % metadata version key. Not sure why 14 bytes when version
    % stamps are only 80, but whatever for now.
    erlfdb:set_versionstamped_value(Tx, ?METADATA_VERSION_KEY, <<0:112>>).


check_metadata_version(#{} = Db) ->
    #{
        tx := Tx,
        md_version := Version
    } = Db,

    AlreadyChecked = get(?PDICT_CHECKED_MD_IS_CURRENT),
    if AlreadyChecked == true -> {current, Db}; true ->
        case erlfdb:wait(erlfdb:get_ss(Tx, ?METADATA_VERSION_KEY)) of
            Version ->
                put(?PDICT_CHECKED_MD_IS_CURRENT, true),
                % We want to set a read conflict on the db version as we'd want
                % to conflict with any writes to this particular db. However
                % during db creation db prefix might not exist yet so we don't
                % add a read-conflict on it then.
                case maps:get(db_prefix, Db, not_found) of
                    not_found ->
                        ok;
                    <<_/binary>> = DbPrefix ->
                        DbVerKey = erlfdb_tuple:pack({?DB_VERSION}, DbPrefix),
                        erlfdb:add_read_conflict_key(Tx, DbVerKey)
                end,
                {current, Db};
            NewVersion ->
                {stale, Db#{md_version := NewVersion}}
        end
    end.


bump_db_version(#{} = Db) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = Db,

    DbVersionKey = erlfdb_tuple:pack({?DB_VERSION}, DbPrefix),
    DbVersion = fabric2_util:uuid(),
    ok = erlfdb:set(Tx, DbVersionKey, DbVersion),
    ok = bump_metadata_version(Tx),
    {ok, DbVersion}.


check_db_version(#{} = Db, CheckDbVersion) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix,
        db_version := DbVersion
    } = Db,

    AlreadyChecked = get(?PDICT_CHECKED_DB_IS_CURRENT),
    if not CheckDbVersion orelse AlreadyChecked == true -> current; true ->
        DbVersionKey = erlfdb_tuple:pack({?DB_VERSION}, DbPrefix),
        case erlfdb:wait(erlfdb:get(Tx, DbVersionKey)) of
            DbVersion ->
                put(?PDICT_CHECKED_DB_IS_CURRENT, true),
                current;
            _NewDBVersion ->
                stale
        end
    end.


soft_delete_db(Db) ->
    #{
        name := DbName,
        tx := Tx,
        layer_prefix := LayerPrefix,
        db_prefix := DbPrefix
    } = ensure_current(Db),

    DbKey = erlfdb_tuple:pack({?ALL_DBS, DbName}, LayerPrefix),
    Timestamp = list_to_binary(fabric2_util:iso8601_timestamp()),
    DeletedDbKeyTuple = {?DELETED_DBS, DbName, Timestamp},
    DeletedDbKey = erlfdb_tuple:pack(DeletedDbKeyTuple, LayerPrefix),
    case erlfdb:wait(erlfdb:get(Tx, DeletedDbKey)) of
        not_found ->
            erlfdb:set(Tx, DeletedDbKey, DbPrefix),
            erlfdb:clear(Tx, DbKey),
            bump_db_version(Db),
            ok;
        _Val ->
            {deletion_frequency_exceeded, DbName}
    end.


hard_delete_db(Db) ->
    #{
        name := DbName,
        tx := Tx,
        layer_prefix := LayerPrefix,
        db_prefix := DbPrefix
    } = ensure_current(Db),

    DbKey = erlfdb_tuple:pack({?ALL_DBS, DbName}, LayerPrefix),

    erlfdb:clear(Tx, DbKey),
    erlfdb:clear_range_startswith(Tx, DbPrefix),
    bump_metadata_version(Tx),
    ok.


write_doc_body(#{} = Db0, #doc{} = Doc) ->
    #{
        tx := Tx
    } = Db = ensure_current(Db0),

    Rows = doc_to_fdb(Db, Doc),
    lists:foreach(fun({Key, Value}) ->
        ok = erlfdb:set(Tx, Key, aegis:encrypt(Db, Key, Value))
    end, Rows).


clear_doc_body(_Db, _DocId, not_found) ->
    % No old body to clear
    ok;

clear_doc_body(#{} = Db, DocId, #{} = RevInfo) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = Db,

    #{
        rev_id := {RevPos, Rev}
    } = RevInfo,

    BaseKey = {?DB_DOCS, DocId, RevPos, Rev},
    {StartKey, EndKey} = erlfdb_tuple:range(BaseKey, DbPrefix),
    ok = erlfdb:clear_range(Tx, StartKey, EndKey).


cleanup_attachments(Db, DocId, NewDoc, ToRemove) ->
    #{
        tx := Tx,
        db_prefix := DbPrefix
    } = Db,

    RemoveRevs = lists:map(fun(#{rev_id := RevId}) -> RevId end, ToRemove),

    % Gather all known document revisions
    {ok, DiskDocs} = fabric2_db:open_doc_revs(Db, DocId, all, []),
    AllDocs = [{ok, NewDoc} | DiskDocs],

    % Get referenced attachment ids
    ActiveIdSet = lists:foldl(fun({ok, Doc}, Acc) ->
        #doc{
            revs = {Pos, [Rev | _]}
        } = Doc,
        case lists:member({Pos, Rev}, RemoveRevs) of
            true ->
                Acc;
            false ->
                lists:foldl(fun(Att, InnerAcc) ->
                    {loc, _Db, _DocId, AttId} = couch_att:fetch(data, Att),
                    sets:add_element(AttId, InnerAcc)
                end, Acc, Doc#doc.atts)
        end
    end, sets:new(), AllDocs),

    AttPrefix = erlfdb_tuple:pack({?DB_ATT_NAMES, DocId}, DbPrefix),
    Options = [{streaming_mode, want_all}],
    Future = erlfdb:get_range_startswith(Tx, AttPrefix, Options),

    ExistingIdSet = lists:foldl(fun({K, _}, Acc) ->
        {?DB_ATT_NAMES, DocId, AttId} = erlfdb_tuple:unpack(K, DbPrefix),
        sets:add_element(AttId, Acc)
    end, sets:new(), erlfdb:wait(Future)),

    AttsToRemove = sets:subtract(ExistingIdSet, ActiveIdSet),

    lists:foreach(fun(AttId) ->
        IdKey = erlfdb_tuple:pack({?DB_ATT_NAMES, DocId, AttId}, DbPrefix),
        erlfdb:clear(Tx, IdKey),

        ChunkKey = erlfdb_tuple:pack({?DB_ATTS, DocId, AttId}, DbPrefix),
        erlfdb:clear_range_startswith(Tx, ChunkKey)
    end, sets:to_list(AttsToRemove)).


revinfo_to_fdb(Tx, DbPrefix, DocId, #{winner := true} = RevId) ->
    #{
        deleted := Deleted,
        rev_id := {RevPos, Rev},
        rev_path := RevPath,
        branch_count := BranchCount,
        att_hash := AttHash,
        rev_size := RevSize
    } = RevId,
    VS = new_versionstamp(Tx),
    Key = {?DB_REVS, DocId, not Deleted, RevPos, Rev},
    Val = {
        ?CURR_REV_FORMAT,
        VS,
        BranchCount,
        list_to_tuple(RevPath),
        AttHash,
        RevSize
    },
    KBin = erlfdb_tuple:pack(Key, DbPrefix),
    VBin = erlfdb_tuple:pack_vs(Val),
    {KBin, VBin, VS};

revinfo_to_fdb(_Tx, DbPrefix, DocId, #{} = RevId) ->
    #{
        deleted := Deleted,
        rev_id := {RevPos, Rev},
        rev_path := RevPath,
        att_hash := AttHash,
        rev_size := RevSize
    } = RevId,
    Key = {?DB_REVS, DocId, not Deleted, RevPos, Rev},
    Val = {?CURR_REV_FORMAT, list_to_tuple(RevPath), AttHash, RevSize},
    KBin = erlfdb_tuple:pack(Key, DbPrefix),
    VBin = erlfdb_tuple:pack(Val),
    {KBin, VBin, undefined}.


fdb_to_revinfo(Key, {?CURR_REV_FORMAT, _, _, _, _, _} = Val) ->
    {?DB_REVS, _DocId, NotDeleted, RevPos, Rev} = Key,
    {_RevFormat, Sequence, BranchCount, RevPath, AttHash, RevSize} = Val,
    #{
        winner => true,
        exists => true,
        deleted => not NotDeleted,
        rev_id => {RevPos, Rev},
        rev_path => tuple_to_list(RevPath),
        sequence => Sequence,
        branch_count => BranchCount,
        att_hash => AttHash,
        rev_size => RevSize
    };

fdb_to_revinfo(Key, {?CURR_REV_FORMAT, _, _, _} = Val)  ->
    {?DB_REVS, _DocId, NotDeleted, RevPos, Rev} = Key,
    {_RevFormat, RevPath, AttHash, RevSize} = Val,
    #{
        winner => false,
        exists => true,
        deleted => not NotDeleted,
        rev_id => {RevPos, Rev},
        rev_path => tuple_to_list(RevPath),
        sequence => undefined,
        branch_count => undefined,
        att_hash => AttHash,
        rev_size => RevSize
    };

fdb_to_revinfo(Key, {0, Seq, BCount, RPath}) ->
    Val = {1, Seq, BCount, RPath, <<>>},
    fdb_to_revinfo(Key, Val);

fdb_to_revinfo(Key, {0, RPath}) ->
    Val = {1, RPath, <<>>},
    fdb_to_revinfo(Key, Val);

fdb_to_revinfo(Key, {1, Seq, BCount, RPath, AttHash}) ->
    % Don't forget to change ?CURR_REV_FORMAT to 2 here when it increments
    Val = {?CURR_REV_FORMAT, Seq, BCount, RPath, AttHash, 0},
    fdb_to_revinfo(Key, Val);

fdb_to_revinfo(Key, {1, RPath, AttHash}) ->
    % Don't forget to change ?CURR_REV_FORMAT to 2 here when it increments
    Val = {?CURR_REV_FORMAT, RPath, AttHash, 0},
    fdb_to_revinfo(Key, Val).


doc_to_fdb(Db, #doc{} = Doc) ->
    #{
        db_prefix := DbPrefix
    } = Db,

    #doc{
        id = Id,
        revs = {Start, [Rev | _]},
        body = Body,
        atts = Atts,
        deleted = Deleted
    } = Doc,

    DiskAtts = lists:map(fun couch_att:to_disk_term/1, Atts),

    Opts = [{minor_version, 1}, {compressed, 6}],
    Value = term_to_binary({Body, DiskAtts, Deleted}, Opts),
    Chunks = chunkify_binary(Value),

    {Rows, _} = lists:mapfoldl(fun(Chunk, ChunkId) ->
        Key = erlfdb_tuple:pack({?DB_DOCS, Id, Start, Rev, ChunkId}, DbPrefix),
        {{Key, Chunk}, ChunkId + 1}
    end, 0, Chunks),

    Rows.


fdb_to_doc(_Db, _DocId, _Pos, _Path, []) ->
    {not_found, missing};

fdb_to_doc(Db, DocId, Pos, Path, BinRows) when is_list(BinRows) ->
    Bin = iolist_to_binary(BinRows),
    {Body, DiskAtts, Deleted} = binary_to_term(Bin, [safe]),
    Atts = lists:map(fun(Att) ->
        couch_att:from_disk_term(Db, DocId, Att)
    end, DiskAtts),
    Doc0 = #doc{
        id = DocId,
        revs = {Pos, Path},
        body = Body,
        atts = Atts,
        deleted = Deleted
    },

    case Db of
        #{after_doc_read := undefined} -> Doc0;
        #{after_doc_read := ADR} -> ADR(Doc0, Db)
    end.


local_doc_to_fdb(Db, #doc{} = Doc) ->
    #{
        db_prefix := DbPrefix
    } = Db,

    #doc{
        id = Id,
        revs = {0, [Rev]},
        body = Body
    } = Doc,

    Key = erlfdb_tuple:pack({?DB_LOCAL_DOCS, Id}, DbPrefix),

    StoreRev = case Rev of
        _ when is_integer(Rev) -> integer_to_binary(Rev);
        _ when is_binary(Rev) -> Rev
    end,

    BVal = term_to_binary(Body, [{minor_version, 1}, {compressed, 6}]),
    {Rows, _} = lists:mapfoldl(fun(Chunk, ChunkId) ->
        K = erlfdb_tuple:pack({?DB_LOCAL_DOC_BODIES, Id, ChunkId}, DbPrefix),
        {{K, Chunk}, ChunkId + 1}
    end, 0, chunkify_binary(BVal)),

    NewSize = fabric2_util:ldoc_size(Doc),
    RawValue = erlfdb_tuple:pack({?CURR_LDOC_FORMAT, StoreRev, NewSize}),

    % Prefix our tuple encoding to make upgrades easier
    Value = <<255, RawValue/binary>>,

    {Key, Value, NewSize, Rows}.


fdb_to_local_doc(_Db, _DocId, not_found, []) ->
    {not_found, missing};

fdb_to_local_doc(_Db, DocId, <<131, _/binary>> = Val, []) ->
    % This is an upgrade clause for the old encoding. We allow reading the old
    % value and will perform an upgrade of the storage format on an update.
    {Rev, Body} = binary_to_term(Val, [safe]),
    #doc{
        id = DocId,
        revs = {0, [Rev]},
        deleted = false,
        body = Body
    };

fdb_to_local_doc(_Db, DocId, <<255, RevBin/binary>>, Rows) when is_list(Rows) ->
    Rev = case erlfdb_tuple:unpack(RevBin) of
        {?CURR_LDOC_FORMAT, Rev0, _Size} -> Rev0
    end,

    BodyBin = iolist_to_binary(Rows),
    Body = binary_to_term(BodyBin, [safe]),

    #doc{
        id = DocId,
        revs = {0, [Rev]},
        deleted = false,
        body = Body
    };

fdb_to_local_doc(Db, DocId, RawRev, Rows) ->
    BaseRev = erlfdb_tuple:pack({?CURR_LDOC_FORMAT, RawRev, 0}),
    Rev = <<255, BaseRev/binary>>,
    fdb_to_local_doc(Db, DocId, Rev, Rows).


sum_add_rev_sizes(RevInfos) ->
    lists:foldl(fun(RI, Acc) ->
        #{
            exists := Exists,
            rev_size := Size
        } = RI,
        case Exists of
            true -> Acc;
            false -> Size + Acc
        end
    end, 0, RevInfos).


sum_rem_rev_sizes(RevInfos) ->
    lists:foldl(fun(RI, Acc) ->
        #{
            exists := true,
            rev_size := Size
        } = RI,
        Size + Acc
    end, 0, RevInfos).


get_fold_acc(Db, RangePrefix, UserCallback, UserAcc, Options)
        when is_map(Db) orelse Db =:= undefined ->

    Reverse = case fabric2_util:get_value(dir, Options) of
        rev -> true;
        _ -> false
    end,

    StartKey0 = fabric2_util:get_value(start_key, Options),
    EndKeyGt = fabric2_util:get_value(end_key_gt, Options),
    EndKey0 = fabric2_util:get_value(end_key, Options, EndKeyGt),
    InclusiveEnd = EndKeyGt == undefined,
    WrapKeys = fabric2_util:get_value(wrap_keys, Options) /= false,

    % CouchDB swaps the key meanings based on the direction
    % of the fold. FoundationDB does not so we have to
    % swap back here.
    {StartKey1, EndKey1} = case Reverse of
        false -> {StartKey0, EndKey0};
        true -> {EndKey0, StartKey0}
    end,

    % Set the maximum bounds for the start and endkey
    StartKey2 = case StartKey1 of
        undefined ->
            <<RangePrefix/binary, 16#00>>;
        SK2 when not WrapKeys ->
            erlfdb_tuple:pack(SK2, RangePrefix);
        SK2 ->
            erlfdb_tuple:pack({SK2}, RangePrefix)
    end,

    EndKey2 = case EndKey1 of
        undefined ->
            <<RangePrefix/binary, 16#FF>>;
        EK2 when Reverse andalso not WrapKeys ->
            PackedEK = erlfdb_tuple:pack(EK2, RangePrefix),
            <<PackedEK/binary, 16#FF>>;
        EK2 when Reverse ->
            PackedEK = erlfdb_tuple:pack({EK2}, RangePrefix),
            <<PackedEK/binary, 16#FF>>;
        EK2 when not WrapKeys ->
            erlfdb_tuple:pack(EK2, RangePrefix);
        EK2 ->
            erlfdb_tuple:pack({EK2}, RangePrefix)
    end,

    % FoundationDB ranges are applied as SK <= key < EK
    % By default, CouchDB is SK <= key <= EK with the
    % optional inclusive_end=false option changing that
    % to SK <= key < EK. Also, remember that CouchDB
    % swaps the meaning of SK and EK based on direction.
    %
    % Thus we have this wonderful bit of logic to account
    % for all of those combinations.

    StartKey3 = case {Reverse, InclusiveEnd} of
        {true, false} ->
            erlfdb_key:first_greater_than(StartKey2);
        _ ->
            StartKey2
    end,

    EndKey3 = case {Reverse, InclusiveEnd} of
        {false, true} when EndKey0 /= undefined ->
            erlfdb_key:first_greater_than(EndKey2);
        {true, _} ->
            erlfdb_key:first_greater_than(EndKey2);
        _ ->
            EndKey2
    end,

    Skip = case fabric2_util:get_value(skip, Options) of
        S when is_integer(S), S >= 0 -> S;
        _ -> 0
    end,

    Limit = case fabric2_util:get_value(limit, Options) of
        L when is_integer(L), L >= 0 -> L + Skip;
        undefined -> 0
    end,

    TargetBytes = case fabric2_util:get_value(target_bytes, Options) of
        T when is_integer(T), T >= 0 -> [{target_bytes, T}];
        undefined -> []
    end,

    StreamingMode = case fabric2_util:get_value(streaming_mode, Options) of
        undefined -> [];
        Name when is_atom(Name) -> [{streaming_mode, Name}]
    end,

    Snapshot = case fabric2_util:get_value(snapshot, Options) of
        undefined -> [];
        B when is_boolean(B) -> [{snapshot, B}]
    end,

    BaseOpts = [{reverse, Reverse}]
            ++ TargetBytes
            ++ StreamingMode
            ++ Snapshot,

    RestartTx = fabric2_util:get_value(restart_tx, Options, false),

    #fold_acc{
        db = Db,
        start_key = StartKey3,
        end_key = EndKey3,
        skip = Skip,
        limit = Limit,
        retries = 0,
        base_opts = BaseOpts,
        restart_tx = RestartTx,
        user_fun = UserCallback,
        user_acc = UserAcc
    }.


fold_range_cb({K, V}, #fold_acc{} = Acc) ->
    #fold_acc{
        skip = Skip,
        limit = Limit,
        user_fun = UserFun,
        user_acc = UserAcc,
        base_opts = Opts
    } = Acc,
    Acc1 = case Skip =:= 0 of
        true ->
            UserAcc1 = UserFun({K, V}, UserAcc),
            Acc#fold_acc{limit = max(0, Limit - 1), user_acc = UserAcc1};
        false ->
            Acc#fold_acc{skip = Skip - 1, limit = Limit - 1}
    end,
    Acc2 = case fabric2_util:get_value(reverse, Opts, false) of
        true -> Acc1#fold_acc{end_key = erlfdb_key:last_less_or_equal(K)};
        false -> Acc1#fold_acc{start_key = erlfdb_key:first_greater_than(K)}
    end,
    put(?PDICT_FOLD_ACC_STATE, Acc2),
    Acc2.


restart_fold(Tx, #fold_acc{} = Acc) ->
    erase(?PDICT_CHECKED_MD_IS_CURRENT),

    ok = erlfdb:reset(Tx),

    case {erase(?PDICT_FOLD_ACC_STATE), Acc#fold_acc.retries} of
        {#fold_acc{db = Db} = Acc1, _} ->
            Acc1#fold_acc{db = check_db_instance(Db), retries = 0};
        {undefined, Retries} when Retries < ?MAX_FOLD_RANGE_RETRIES ->
            Db = check_db_instance(Acc#fold_acc.db),
            Acc#fold_acc{db = Db, retries = Retries + 1};
        {undefined, _} ->
            error(fold_range_not_progressing)
    end.


get_db_handle() ->
    case get(?PDICT_DB_KEY) of
        undefined ->
            {ok, Db} = application:get_env(fabric, db),
            put(?PDICT_DB_KEY, Db),
            Db;
        Db ->
            Db
    end.


require_transaction(#{tx := {erlfdb_snapshot, _}} = _Db) ->
    ok;
require_transaction(#{tx := {erlfdb_transaction, _}} = _Db) ->
    ok;
require_transaction(#{} = _Db) ->
    erlang:error(transaction_required).


ensure_current(Db) ->
    ensure_current(Db, true).


ensure_current(#{} = Db0, CheckDbVersion) ->
    require_transaction(Db0),
    Db3 = case check_metadata_version(Db0) of
        {current, Db1} ->
            Db1;
        {stale, Db1} ->
            case check_db_version(Db1, CheckDbVersion) of
                current ->
                    % If db version is current, update cache with the latest
                    % metadata so other requests can immediately see the
                    % refreshed db handle.
                    Now = erlang:monotonic_time(millisecond),
                    Db2 = Db1#{check_current_ts := Now},
                    fabric2_server:maybe_update(Db2),
                    Db2;
                stale ->
                    fabric2_server:maybe_remove(Db1),
                    throw({?MODULE, reopen})
            end
    end,
    case maps:get(security_fun, Db3) of
        SecurityFun when is_function(SecurityFun, 2) ->
            #{security_doc := SecDoc} = Db3,
            ok = SecurityFun(Db3, SecDoc),
            Db3#{security_fun := undefined};
        undefined ->
            Db3
    end.


check_db_instance(undefined) ->
    undefined;

check_db_instance(#{} = Db) ->
    require_transaction(Db),
    case check_metadata_version(Db) of
        {current, Db1} ->
            Db1;
        {stale, Db1} ->
            #{
                tx := Tx,
                uuid := UUID,
                db_prefix := DbPrefix
            } = Db1,
            UUIDKey = erlfdb_tuple:pack({?DB_CONFIG, <<"uuid">>}, DbPrefix),
            case erlfdb:wait(erlfdb:get(Tx, UUIDKey)) of
                UUID -> Db1;
                _ -> error(database_does_not_exist)
            end
    end.


is_transaction_applied(Tx) ->
    is_commit_unknown_result()
        andalso has_transaction_id()
        andalso transaction_id_exists(Tx).


get_previous_transaction_result() ->
    get(?PDICT_TX_RES_KEY).


execute_transaction(Tx, Fun, LayerPrefix) ->
    put(?PDICT_CHECKED_MD_IS_CURRENT, false),
    put(?PDICT_CHECKED_DB_IS_CURRENT, false),
    Result = Fun(Tx),
    case erlfdb:is_read_only(Tx) of
        true ->
            ok;
        false ->
            erlfdb:set(Tx, get_transaction_id(Tx, LayerPrefix), <<>>),
            put(?PDICT_TX_RES_KEY, Result)
    end,
    Result.


clear_transaction() ->
    fabric2_txids:remove(get(?PDICT_TX_ID_KEY)),
    erase(?PDICT_CHECKED_DB_IS_CURRENT),
    erase(?PDICT_CHECKED_MD_IS_CURRENT),
    erase(?PDICT_TX_ID_KEY),
    erase(?PDICT_TX_RES_KEY).


is_commit_unknown_result() ->
    erlfdb:get_last_error() == ?COMMIT_UNKNOWN_RESULT.


has_transaction_id() ->
    is_binary(get(?PDICT_TX_ID_KEY)).


transaction_id_exists(Tx) ->
    erlfdb:wait(erlfdb:get(Tx, get(?PDICT_TX_ID_KEY))) == <<>>.


get_transaction_id(Tx, LayerPrefix) ->
    case get(?PDICT_TX_ID_KEY) of
        undefined ->
            TxId = fabric2_txids:create(Tx, LayerPrefix),
            put(?PDICT_TX_ID_KEY, TxId),
            TxId;
        TxId when is_binary(TxId) ->
            TxId
    end.


with_span(Operation, ExtraTags, Fun) ->
    case ctrace:has_span() of
        true ->
            Tags = maps:merge(#{
                'span.kind' => <<"client">>,
                component => <<"couchdb.fabric">>,
                'db.instance' => fabric2_server:fdb_cluster(),
                'db.namespace' => fabric2_server:fdb_directory(),
                'db.type' => <<"fdb">>,
                nonce => get(nonce),
                pid => self()
            }, ExtraTags),
            ctrace:with_span(Operation, Tags, Fun);
        false ->
            Fun()
    end.


get_info_wait_int(#info_future{} = InfoFuture) ->
    #info_future{
        db_prefix = DbPrefix,
        changes_future = ChangesFuture,
        uuid_future = UUIDFuture,
        meta_future = MetaFuture
    } = InfoFuture,

    RawSeq = case erlfdb:wait(ChangesFuture) of
        [] ->
            vs_to_seq(fabric2_util:seq_zero_vs());
        [{SeqKey, _}] ->
            {?DB_CHANGES, SeqVS} = erlfdb_tuple:unpack(SeqKey, DbPrefix),
            vs_to_seq(SeqVS)
    end,
    CProp = {update_seq, RawSeq},

    UUIDProp = {uuid, erlfdb:wait(UUIDFuture)},

    MProps = lists:foldl(fun({K, V}, Acc) ->
        case erlfdb_tuple:unpack(K, DbPrefix) of
            {?DB_STATS, <<"doc_count">>} ->
                [{doc_count, ?bin2uint(V)} | Acc];
            {?DB_STATS, <<"doc_del_count">>} ->
                [{doc_del_count, ?bin2uint(V)} | Acc];
            {?DB_STATS, <<"sizes">>, Name} ->
                Val = ?bin2uint(V),
                {_, {Sizes}} = lists:keyfind(sizes, 1, Acc),
                NewSizes = lists:keystore(Name, 1, Sizes, {Name, Val}),
                lists:keystore(sizes, 1, Acc, {sizes, {NewSizes}});
            {?DB_STATS, _} ->
                Acc
        end
    end, [{sizes, {[]}}], erlfdb:wait(MetaFuture)),

    [CProp, UUIDProp | MProps].


binary_chunk_size() ->
    config:get_integer(
        "fabric", "binary_chunk_size", ?DEFAULT_BINARY_CHUNK_SIZE).


-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").

fdb_to_revinfo_version_compatibility_test() ->
    DocId = <<"doc_id">>,
    FirstRevFormat = 0,
    RevPos = 1,
    Rev = <<60,84,174,140,210,120,192,18,100,148,9,181,129,165,248,92>>,
    RevPath = {},
    NotDeleted = true,
    Sequence = {versionstamp, 10873034897377, 0, 0},
    BranchCount = 1,

    KeyWinner = {?DB_REVS, DocId, NotDeleted, RevPos, Rev},
    ValWinner = {FirstRevFormat, Sequence, BranchCount, RevPath},
    ExpectedWinner = expected(
        true, BranchCount, NotDeleted, RevPos, Rev, RevPath, Sequence),
    ?assertEqual(ExpectedWinner, fdb_to_revinfo(KeyWinner, ValWinner)),

    KeyLoser = {?DB_REVS, DocId, NotDeleted, RevPos, Rev},
    ValLoser = {FirstRevFormat, RevPath},
    ExpectedLoser = expected(
        false, undefined, NotDeleted, RevPos, Rev, RevPath, undefined),
    ?assertEqual(ExpectedLoser, fdb_to_revinfo(KeyLoser, ValLoser)),
    ok.


expected(Winner, BranchCount, NotDeleted, RevPos, Rev, RevPath, Sequence) ->
    #{
        att_hash => <<>>,
        branch_count => BranchCount,
        deleted => not NotDeleted,
        exists => true,
        rev_id => {RevPos, Rev},
        rev_path => tuple_to_list(RevPath),
        rev_size => 0,
        sequence => Sequence,
        winner => Winner
    }.


-endif.