% 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_db).

-export([
    create/2,
    open/2,
    delete/2,
    undelete/4,

    list_dbs/0,
    list_dbs/1,
    list_dbs/3,

    list_dbs_info/0,
    list_dbs_info/1,
    list_dbs_info/3,

    list_deleted_dbs_info/0,
    list_deleted_dbs_info/1,
    list_deleted_dbs_info/3,

    check_is_admin/1,
    check_is_member/1,

    name/1,
    get_after_doc_read_fun/1,
    get_before_doc_update_fun/1,
    get_committed_update_seq/1,
    get_compacted_seq/1,
    get_compactor_pid/1,
    get_db_info/1,
    %% get_partition_info/2,
    get_del_doc_count/1,
    get_doc_count/1,
    get_doc_count/2,
    %% get_epochs/1,
    %% get_filepath/1,
    get_instance_start_time/1,
    get_pid/1,
    get_revs_limit/1,
    get_revs_limit/2,
    get_security/1,
    get_security/2,
    get_update_seq/1,
    get_user_ctx/1,
    get_uuid/1,
    %% get_purge_seq/1,
    %% get_oldest_purge_seq/1,
    %% get_purge_infos_limit/1,

    is_clustered/1,
    is_db/1,
    is_partitioned/1,
    is_system_db/1,
    is_system_db_name/1,
    is_replicator_db/1,
    is_users_db/1,

    set_revs_limit/2,
    %% set_purge_infos_limit/2,
    set_security/2,
    set_user_ctx/2,

    ensure_full_commit/1,
    ensure_full_commit/2,

    %% load_validation_funs/1,
    %% reload_validation_funs/1,

    open_doc/2,
    open_doc/3,
    open_doc_revs/4,
    %% open_doc_int/3,
    get_doc_info/2,
    get_full_doc_info/2,
    get_full_doc_infos/2,
    get_missing_revs/2,
    get_design_docs/1,
    %% get_purge_infos/2,

    %% get_minimum_purge_seq/1,
    %% purge_client_exists/3,

    validate_docid/1,
    %% doc_from_json_obj_validate/2,

    update_doc/2,
    update_doc/3,
    update_docs/2,
    update_docs/3,
    %% delete_doc/3,

    %% purge_docs/2,
    %% purge_docs/3,

    read_attachment/3,
    write_attachment/3,

    fold_docs/3,
    fold_docs/4,
    fold_docs/5,
    fold_design_docs/4,
    fold_local_docs/4,
    fold_changes/4,
    fold_changes/5,
    %% count_changes_since/2,
    %% fold_purge_infos/4,
    %% fold_purge_infos/5,

    %% calculate_start_seq/3,
    %% owner_of/2,

    %% start_compact/1,
    %% cancel_compact/1,
    %% wait_for_compaction/1,
    %% wait_for_compaction/2,

    dbname_suffix/1,
    validate_dbname/1,

    %% make_doc/5,
    new_revid/2,

    apply_open_doc_opts/3
]).

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

% Default max database name length is based on CouchDb < 4.x compatibility. See
% default.ini entry for additional information.
-define(DEFAULT_MAX_DATABASE_NAME_LENGTH, 238).

-define(DBNAME_REGEX,
    % use the stock CouchDB regex
    "^[a-z][a-z0-9\\_\\$()\\+\\-\\/]*"
    % but allow an optional shard timestamp at the end
    "(\\.[0-9]{10,})?$"
).

-define(FIRST_DDOC_KEY, <<"_design/">>).
-define(LAST_DDOC_KEY, <<"_design0">>).

-define(RETURN(Term), throw({?MODULE, Term})).

-define(DEFAULT_UPDATE_DOCS_BATCH_SIZE, 2500000).

-record(bacc, {
    db,
    docs,
    batch_size,
    options,
    rev_futures,
    seen,
    results
}).

create(DbName, Options) ->
    case validate_dbname(DbName) of
        ok ->
            Result = fabric2_fdb:transactional(DbName, fun(TxDb) ->
                case fabric2_fdb:exists(TxDb) of
                    true ->
                        {error, file_exists};
                    false ->
                        fabric2_fdb:create(TxDb, Options)
                end
            end),
            % We cache outside of the transaction so that we're sure
            % that the transaction was committed.
            case Result of
                #{} = Db0 ->
                    Db1 = maybe_add_sys_db_callbacks(Db0),
                    ok = fabric2_server:store(Db1),
                    fabric2_db_plugin:after_db_create(DbName, get_uuid(Db1)),
                    {ok, Db1#{tx := undefined}};
                Error ->
                    Error
            end;
        Error ->
            Error
    end.

open(DbName, Options) ->
    UUID = fabric2_util:get_value(uuid, Options),
    case fabric2_server:fetch(DbName, UUID) of
        #{} = Db ->
            Db1 = maybe_set_user_ctx(Db, Options),
            Db2 = maybe_set_interactive(Db1, Options),
            {ok, require_member_check(Db2)};
        undefined ->
            Result = fabric2_fdb:transactional(DbName, fun(TxDb) ->
                fabric2_fdb:open(TxDb, Options)
            end),
            % Cache outside the transaction retry loop
            case Result of
                #{} = Db0 ->
                    Db1 = maybe_add_sys_db_callbacks(Db0),
                    ok = fabric2_server:store(Db1),
                    Db2 = Db1#{tx := undefined},
                    {ok, require_member_check(Db2)};
                Error ->
                    Error
            end
    end.

delete(DbName, Options) ->
    % Delete doesn't check user_ctx, that's done at the HTTP API level
    % here we just care to get the `database_does_not_exist` error thrown
    Options1 = lists:keystore(user_ctx, 1, Options, ?ADMIN_CTX),
    case lists:keyfind(deleted_at, 1, Options1) of
        {deleted_at, TimeStamp} ->
            fabric2_fdb:transactional(DbName, fun(TxDb) ->
                fabric2_fdb:remove_deleted_db(TxDb, TimeStamp)
            end);
        false ->
            {ok, Db} = open(DbName, Options1),
            Resp = fabric2_fdb:transactional(Db, fun(TxDb) ->
                fabric2_fdb:delete(TxDb)
            end),
            if
                Resp /= ok ->
                    Resp;
                true ->
                    fabric2_db_plugin:after_db_delete(DbName, get_uuid(Db)),
                    fabric2_server:remove(DbName)
            end
    end.

undelete(DbName, TgtDbName, TimeStamp, Options) ->
    case validate_dbname(TgtDbName) of
        ok ->
            Resp = fabric2_fdb:transactional(
                DbName,
                fun(TxDb) ->
                    fabric2_fdb:undelete(TxDb, TgtDbName, TimeStamp)
                end
            ),
            if
                Resp /= ok ->
                    ok;
                true ->
                    {ok, Db} = open(TgtDbName, Options),
                    fabric2_db_plugin:after_db_create(TgtDbName, get_uuid(Db))
            end,
            Resp;
        Error ->
            Error
    end.

list_dbs() ->
    list_dbs([]).

list_dbs(Options) ->
    Callback = fun(DbName, Acc) -> [DbName | Acc] end,
    DbNames = fabric2_fdb:transactional(fun(Tx) ->
        fabric2_fdb:list_dbs(Tx, Callback, [], Options)
    end),
    lists:reverse(DbNames).

list_dbs(UserFun, UserAcc0, Options) ->
    FoldFun = fun(DbName, Acc) -> maybe_stop(UserFun({row, [{id, DbName}]}, Acc)) end,
    try
        UserAcc1 = maybe_stop(UserFun({meta, []}, UserAcc0)),
        UserAcc2 = fabric2_fdb:transactional(fun(Tx) ->
            fabric2_fdb:list_dbs(Tx, FoldFun, UserAcc1, Options)
        end),
        {ok, maybe_stop(UserFun(complete, UserAcc2))}
    catch
        throw:{stop, FinalUserAcc} ->
            {ok, FinalUserAcc}
    end.

list_dbs_info() ->
    list_dbs_info([]).

list_dbs_info(Options) ->
    Callback = fun(Value, Acc) ->
        NewAcc =
            case Value of
                {meta, _} -> Acc;
                {row, DbInfo} -> [DbInfo | Acc];
                complete -> Acc
            end,
        {ok, NewAcc}
    end,
    {ok, DbInfos} = list_dbs_info(Callback, [], Options),
    {ok, lists:reverse(DbInfos)}.

list_dbs_info(UserFun, UserAcc0, Options) ->
    FoldFun = fun(DbName, InfoFuture, {FutureQ, Count, Acc}) ->
        NewFutureQ = queue:in({DbName, InfoFuture}, FutureQ),
        drain_info_futures(NewFutureQ, Count + 1, UserFun, Acc)
    end,
    try
        UserAcc1 = maybe_stop(UserFun({meta, []}, UserAcc0)),
        InitAcc = {queue:new(), 0, UserAcc1},
        UserAcc3 = fabric2_fdb:transactional(fun(Tx) ->
            {FinalFutureQ, _, UserAcc2} = fabric2_fdb:list_dbs_info(
                Tx,
                FoldFun,
                InitAcc,
                Options
            ),
            drain_all_info_futures(FinalFutureQ, UserFun, UserAcc2)
        end),
        {ok, maybe_stop(UserFun(complete, UserAcc3))}
    catch
        throw:{stop, FinalUserAcc} ->
            {ok, FinalUserAcc}
    end.

list_deleted_dbs_info() ->
    list_deleted_dbs_info([]).

list_deleted_dbs_info(Options) ->
    Callback = fun(Value, Acc) ->
        NewAcc =
            case Value of
                {meta, _} -> Acc;
                {row, DbInfo} -> [DbInfo | Acc];
                complete -> Acc
            end,
        {ok, NewAcc}
    end,
    {ok, DbInfos} = list_deleted_dbs_info(Callback, [], Options),
    {ok, lists:reverse(DbInfos)}.

list_deleted_dbs_info(UserFun, UserAcc0, Options0) ->
    Dir = fabric2_util:get_value(dir, Options0, fwd),
    StartKey0 = fabric2_util:get_value(start_key, Options0),
    EndKey0 = fabric2_util:get_value(end_key, Options0),

    {FirstBinary, LastBinary} =
        case Dir of
            fwd -> {<<>>, <<255>>};
            rev -> {<<255>>, <<>>}
        end,

    StartKey1 =
        case StartKey0 of
            undefined ->
                {FirstBinary};
            DbName0 when is_binary(DbName0) ->
                {DbName0, FirstBinary};
            [DbName0, TimeStamp0] when is_binary(DbName0), is_binary(TimeStamp0) ->
                {DbName0, TimeStamp0};
            BadStartKey ->
                erlang:error({invalid_start_key, BadStartKey})
        end,
    EndKey1 =
        case EndKey0 of
            undefined ->
                {LastBinary};
            DbName1 when is_binary(DbName1) ->
                {DbName1, LastBinary};
            [DbName1, TimeStamp1] when is_binary(DbName1), is_binary(TimeStamp1) ->
                {DbName1, TimeStamp1};
            BadEndKey ->
                erlang:error({invalid_end_key, BadEndKey})
        end,

    Options1 = Options0 -- [{start_key, StartKey0}, {end_key, EndKey0}],
    Options2 =
        [
            {start_key, StartKey1},
            {end_key, EndKey1},
            {wrap_keys, false}
        ] ++ Options1,

    FoldFun = fun(DbName, TimeStamp, InfoFuture, {FutureQ, Count, Acc}) ->
        NewFutureQ = queue:in({DbName, TimeStamp, InfoFuture}, FutureQ),
        drain_deleted_info_futures(NewFutureQ, Count + 1, UserFun, Acc)
    end,
    try
        UserAcc1 = maybe_stop(UserFun({meta, []}, UserAcc0)),
        InitAcc = {queue:new(), 0, UserAcc1},
        UserAcc3 = fabric2_fdb:transactional(fun(Tx) ->
            {FinalFutureQ, _, UserAcc2} = fabric2_fdb:list_deleted_dbs_info(
                Tx,
                FoldFun,
                InitAcc,
                Options2
            ),
            drain_all_deleted_info_futures(FinalFutureQ, UserFun, UserAcc2)
        end),
        {ok, maybe_stop(UserFun(complete, UserAcc3))}
    catch
        throw:{stop, FinalUserAcc} ->
            {ok, FinalUserAcc}
    end.

is_admin(Db, {SecProps}) when is_list(SecProps) ->
    case fabric2_db_plugin:check_is_admin(Db) of
        true ->
            true;
        false ->
            UserCtx = get_user_ctx(Db),
            {Admins} = get_admins(SecProps),
            is_authorized(Admins, UserCtx)
    end.

check_is_admin(Db) ->
    check_is_admin(Db, get_security(Db)).

check_is_admin(Db, SecDoc) ->
    case is_admin(Db, SecDoc) of
        true ->
            ok;
        false ->
            UserCtx = get_user_ctx(Db),
            Reason = <<"You are not a db or server admin.">>,
            throw_security_error(UserCtx, Reason)
    end.

check_is_member(Db) ->
    check_is_member(Db, get_security(Db)).

check_is_member(Db, SecDoc) ->
    case is_member(Db, SecDoc) of
        true ->
            ok;
        false ->
            UserCtx = get_user_ctx(Db),
            throw_security_error(UserCtx)
    end.

require_admin_check(#{} = Db) ->
    Db#{security_fun := fun check_is_admin/2}.

require_member_check(#{} = Db) ->
    Db#{security_fun := fun check_is_member/2}.

name(#{name := DbName}) ->
    DbName.

get_after_doc_read_fun(#{after_doc_read := AfterDocRead}) ->
    AfterDocRead.

get_before_doc_update_fun(#{before_doc_update := BeforeDocUpdate}) ->
    BeforeDocUpdate.

get_committed_update_seq(#{} = Db) ->
    get_update_seq(Db).

get_compacted_seq(#{} = Db) ->
    get_update_seq(Db).

get_compactor_pid(#{} = _Db) ->
    nil.

get_db_info(#{} = Db) ->
    DbProps = fabric2_fdb:transactional(Db, fun(TxDb) ->
        fabric2_fdb:get_info(TxDb)
    end),
    {ok, make_db_info(name(Db), DbProps)}.

get_del_doc_count(#{} = Db) ->
    get_doc_count(Db, <<"doc_del_count">>).

get_doc_count(Db) ->
    get_doc_count(Db, <<"doc_count">>).

get_doc_count(Db, undefined) ->
    get_doc_count(Db, <<"doc_count">>);
get_doc_count(Db, <<"_all_docs">>) ->
    get_doc_count(Db, <<"doc_count">>);
get_doc_count(DbName, <<"_design">>) ->
    get_doc_count(DbName, <<"doc_design_count">>);
get_doc_count(DbName, <<"_local">>) ->
    get_doc_count(DbName, <<"doc_local_count">>);
get_doc_count(Db, Key) ->
    fabric2_fdb:transactional(Db, fun(TxDb) ->
        fabric2_fdb:get_stat(TxDb, Key)
    end).

get_instance_start_time(#{}) ->
    0.

get_pid(#{}) ->
    nil.

get_revs_limit(#{} = Db) ->
    get_revs_limit(Db, []).

get_revs_limit(#{} = Db, Opts) ->
    CurrentDb = get_cached_db(Db, Opts),
    maps:get(revs_limit, CurrentDb).

get_security(#{} = Db) ->
    get_security(Db, []).

get_security(#{} = Db, Opts) ->
    CurrentDb = get_cached_db(Db, Opts),
    maps:get(security_doc, CurrentDb).

get_update_seq(#{} = Db) ->
    fabric2_fdb:transactional(Db, fun(TxDb) ->
        fabric2_fdb:get_last_change(TxDb)
    end).

get_user_ctx(#{user_ctx := UserCtx}) ->
    UserCtx.

get_uuid(#{uuid := UUID}) ->
    UUID.

is_clustered(#{}) ->
    false.

is_db(#{name := _}) ->
    true;
is_db(_) ->
    false.

is_partitioned(#{}) ->
    false.

is_system_db(#{name := DbName}) ->
    is_system_db_name(DbName).

is_system_db_name(DbName) when is_list(DbName) ->
    is_system_db_name(?l2b(DbName));
is_system_db_name(DbName) when is_binary(DbName) ->
    Suffix = filename:basename(DbName),
    case {filename:dirname(DbName), lists:member(Suffix, ?SYSTEM_DATABASES)} of
        {<<".">>, Result} ->
            Result;
        {_Prefix, false} ->
            false;
        {Prefix, true} ->
            ReOpts = [{capture, none}, dollar_endonly],
            re:run(Prefix, ?DBNAME_REGEX, ReOpts) == match
    end.

is_replicator_db(#{name := DbName}) ->
    is_replicator_db(DbName);
is_replicator_db(DbName) when is_binary(DbName) ->
    fabric2_util:dbname_ends_with(DbName, <<"_replicator">>).

is_users_db(#{name := DbName}) ->
    is_users_db(DbName);
is_users_db(DbName) when is_binary(DbName) ->
    AuthenticationDb = config:get("chttpd_auth", "authentication_db"),
    CfgUsersSuffix = config:get("couchdb", "users_db_suffix", "_users"),

    IsAuthCache =
        if
            AuthenticationDb == undefined -> false;
            true -> DbName == ?l2b(AuthenticationDb)
        end,
    IsCfgUsersDb = fabric2_util:dbname_ends_with(DbName, ?l2b(CfgUsersSuffix)),
    IsGlobalUsersDb = fabric2_util:dbname_ends_with(DbName, <<"_users">>),

    IsAuthCache orelse IsCfgUsersDb orelse IsGlobalUsersDb.

set_revs_limit(#{} = Db0, RevsLimit) when is_integer(RevsLimit) ->
    Db1 = require_admin_check(Db0),
    Resp = fabric2_fdb:transactional(Db1, fun(TxDb) ->
        fabric2_fdb:set_config(TxDb, revs_limit, RevsLimit)
    end),
    case Resp of
        {ok, #{} = Db2} -> fabric2_server:store(Db2);
        Err -> Err
    end.

set_security(#{} = Db0, Security) ->
    Db1 = require_admin_check(Db0),
    ok = fabric2_util:validate_security_object(Security),
    Resp = fabric2_fdb:transactional(Db1, fun(TxDb) ->
        fabric2_fdb:set_config(TxDb, security_doc, Security)
    end),
    case Resp of
        {ok, #{} = Db2} -> fabric2_server:store(Db2);
        Err -> Err
    end.

set_user_ctx(#{} = Db, UserCtx) ->
    Db#{user_ctx := UserCtx}.

ensure_full_commit(#{}) ->
    {ok, 0}.

ensure_full_commit(#{}, _Timeout) ->
    {ok, 0}.

open_doc(#{} = Db, DocId) ->
    open_doc(Db, DocId, []).

open_doc(#{} = Db, <<?LOCAL_DOC_PREFIX, _/binary>> = DocId, _Options) ->
    fabric2_fdb:transactional(Db, fun(TxDb) ->
        case fabric2_fdb:get_local_doc(TxDb, DocId) of
            #doc{} = Doc -> {ok, Doc};
            Else -> Else
        end
    end);
open_doc(#{} = Db, DocId, Options) ->
    NeedsTreeOpts = [revs_info, conflicts, deleted_conflicts],
    NeedsTree = (Options -- NeedsTreeOpts /= Options),
    OpenDeleted = lists:member(deleted, Options),
    fabric2_fdb:transactional(Db, fun(TxDb) ->
        Revs =
            case NeedsTree of
                true -> fabric2_fdb:get_all_revs(TxDb, DocId);
                false -> fabric2_fdb:get_winning_revs(TxDb, DocId, 1)
            end,
        if
            Revs == [] ->
                {not_found, missing};
            true ->
                #{winner := true} = RI = lists:last(Revs),
                case fabric2_fdb:get_doc_body(TxDb, DocId, RI) of
                    #doc{deleted = true} when not OpenDeleted ->
                        {not_found, deleted};
                    #doc{} = Doc ->
                        apply_open_doc_opts(Doc, Revs, Options);
                    Else ->
                        Else
                end
        end
    end).

open_doc_revs(Db, DocId, Revs, Options) ->
    Latest = lists:member(latest, Options),
    fabric2_fdb:transactional(Db, fun(TxDb) ->
        AllRevInfos = fabric2_fdb:get_all_revs(TxDb, DocId),
        RevTree = lists:foldl(
            fun(RI, TreeAcc) ->
                RIPath = fabric2_util:revinfo_to_path(RI),
                {Merged, _} = couch_key_tree:merge(TreeAcc, RIPath),
                Merged
            end,
            [],
            AllRevInfos
        ),
        {Found, Missing} =
            case Revs of
                all ->
                    {couch_key_tree:get_all_leafs(RevTree), []};
                _ when Latest ->
                    couch_key_tree:get_key_leafs(RevTree, Revs);
                _ ->
                    couch_key_tree:get(RevTree, Revs)
            end,
        Docs = lists:map(
            fun({Value, {Pos, [Rev | RevPath]}}) ->
                case Value of
                    ?REV_MISSING ->
                        % We have the rev in our list but know nothing about it
                        {{not_found, missing}, {Pos, Rev}};
                    _ ->
                        RevInfo = #{
                            rev_id => {Pos, Rev},
                            rev_path => RevPath
                        },
                        case fabric2_fdb:get_doc_body(TxDb, DocId, RevInfo) of
                            #doc{} = Doc ->
                                apply_open_doc_opts(Doc, AllRevInfos, Options);
                            Else ->
                                {Else, {Pos, Rev}}
                        end
                end
            end,
            Found
        ),
        MissingDocs = [{{not_found, missing}, MRev} || MRev <- Missing],
        {ok, Docs ++ MissingDocs}
    end).

get_doc_info(Db, DocId) ->
    case get_full_doc_info(Db, DocId) of
        not_found -> not_found;
        FDI -> couch_doc:to_doc_info(FDI)
    end.

get_full_doc_info(Db, DocId) ->
    RevInfos = fabric2_fdb:transactional(Db, fun(TxDb) ->
        fabric2_fdb:get_all_revs(TxDb, DocId)
    end),
    if
        RevInfos == [] ->
            not_found;
        true ->
            #{winner := true} = Winner = lists:last(RevInfos),
            RevTree = lists:foldl(
                fun(RI, TreeAcc) ->
                    RIPath = fabric2_util:revinfo_to_path(RI),
                    {Merged, _} = couch_key_tree:merge(TreeAcc, RIPath),
                    Merged
                end,
                [],
                RevInfos
            ),
            #full_doc_info{
                id = DocId,
                update_seq = fabric2_fdb:vs_to_seq(maps:get(sequence, Winner)),
                deleted = maps:get(deleted, Winner),
                rev_tree = RevTree
            }
    end.

get_full_doc_infos(Db, DocIds) ->
    fabric2_fdb:transactional(Db, fun(TxDb) ->
        lists:map(
            fun(DocId) ->
                get_full_doc_info(TxDb, DocId)
            end,
            DocIds
        )
    end).

get_missing_revs(Db, JsonIdRevs) ->
    IdRevs = [idrevs(IdR) || IdR <- JsonIdRevs],
    AllRevInfos = fabric2_fdb:transactional(Db, fun(TxDb) ->
        lists:foldl(
            fun({Id, _Revs}, Acc) ->
                case maps:is_key(Id, Acc) of
                    true ->
                        Acc;
                    false ->
                        RevInfos = fabric2_fdb:get_all_revs(TxDb, Id),
                        Acc#{Id => RevInfos}
                end
            end,
            #{},
            IdRevs
        )
    end),
    AllMissing = lists:flatmap(
        fun({Id, Revs}) ->
            #{Id := RevInfos} = AllRevInfos,
            Missing =
                try
                    lists:foldl(
                        fun(RevInfo, RevAcc) ->
                            if
                                RevAcc /= [] -> ok;
                                true -> throw(all_found)
                            end,
                            filter_found_revs(RevInfo, RevAcc)
                        end,
                        Revs,
                        RevInfos
                    )
                catch
                    throw:all_found ->
                        []
                end,
            if
                Missing == [] ->
                    [];
                true ->
                    PossibleAncestors = find_possible_ancestors(RevInfos, Missing),
                    [{Id, Missing, PossibleAncestors}]
            end
        end,
        IdRevs
    ),
    {ok, AllMissing}.

get_design_docs(Db) ->
    fabric2_fdb:transactional(Db, fun(TxDb) ->
        #{
            db_prefix := DbPrefix
        } = TxDb,

        Prefix = erlfdb_tuple:pack({?DB_ALL_DOCS}, DbPrefix),
        Options = set_design_doc_keys([]),
        FoldFun = fun({Key, Val}, Acc) ->
            {DocId} = erlfdb_tuple:unpack(Key, Prefix),
            RevId = erlfdb_tuple:unpack(Val),
            Rev = #{
                rev_id => RevId,
                rev_path => []
            },
            Future = fabric2_fdb:get_doc_body_future(TxDb, DocId, Rev),
            [{DocId, Rev, Future} | Acc]
        end,
        Futures = fabric2_fdb:fold_range(TxDb, Prefix, FoldFun, [], Options),

        % Using foldl instead of map means that the design
        % docs come out in sorted order.
        lists:foldl(
            fun({DocId, Rev, Future}, Acc) ->
                [fabric2_fdb:get_doc_body_wait(TxDb, DocId, Rev, Future) | Acc]
            end,
            [],
            Futures
        )
    end).

validate_docid(<<"">>) ->
    throw({illegal_docid, <<"Document id must not be empty">>});
validate_docid(<<"_design/">>) ->
    throw({illegal_docid, <<"Illegal document id `_design/`">>});
validate_docid(<<"_local/">>) ->
    throw({illegal_docid, <<"Illegal document id `_local/`">>});
validate_docid(Id) when is_binary(Id) ->
    ConfigDocIdLength = config:get_integer(
        "couchdb", "max_document_id_length", ?DOC_ID_LIMIT_BYTES
    ),
    DocIdLength = min(ConfigDocIdLength, ?DOC_ID_LIMIT_BYTES),
    case DocIdLength > 0 andalso byte_size(Id) > DocIdLength of
        true -> throw({illegal_docid, <<"Document id is too long">>});
        false -> ok
    end,
    case couch_util:validate_utf8(Id) of
        false -> throw({illegal_docid, <<"Document id must be valid UTF-8">>});
        true -> ok
    end,
    case Id of
        <<?DESIGN_DOC_PREFIX, _/binary>> ->
            ok;
        <<?LOCAL_DOC_PREFIX, _/binary>> ->
            ok;
        <<"_", _/binary>> ->
            case fabric2_db_plugin:validate_docid(Id) of
                true ->
                    ok;
                false ->
                    throw(
                        {illegal_docid, <<"Only reserved document ids may start with underscore.">>}
                    )
            end;
        _Else ->
            ok
    end;
validate_docid(Id) ->
    ?LOG_DEBUG(#{what => illegal_docid, docid => Id}),
    couch_log:debug("Document id is not a string: ~p", [Id]),
    throw({illegal_docid, <<"Document id must be a string">>}).

update_doc(Db, Doc) ->
    update_doc(Db, Doc, []).

update_doc(Db, Doc, Options) ->
    case update_docs(Db, [Doc], Options) of
        {ok, [{ok, NewRev}]} ->
            {ok, NewRev};
        {ok, [{{_Id, _Rev}, Error}]} ->
            throw(Error);
        {error, [{{_Id, _Rev}, Error}]} ->
            throw(Error);
        {error, [Error]} ->
            throw(Error);
        {ok, []} ->
            % replication success
            {Pos, [RevId | _]} = Doc#doc.revs,
            {ok, {Pos, RevId}}
    end.

update_docs(Db, Docs) ->
    update_docs(Db, Docs, []).

update_docs(Db, Docs0, Options) ->
    Docs1 = apply_before_doc_update(Db, Docs0, Options),
    try
        validate_atomic_update(Docs0, lists:member(all_or_nothing, Options)),

        Resps0 = batch_update_docs(Db, Docs1, Options),

        % Notify index builder
        fabric2_index:db_updated(name(Db)),

        % Convert errors
        Resps1 = lists:map(
            fun(Resp) ->
                case Resp of
                    {#doc{} = Doc, Error} ->
                        #doc{
                            id = DocId,
                            revs = Revs
                        } = Doc,
                        RevId =
                            case Revs of
                                {RevPos, [Rev | _]} -> {RevPos, Rev};
                                {0, []} -> {0, <<>>};
                                Else -> Else
                            end,
                        {{DocId, RevId}, Error};
                    Else ->
                        Else
                end
            end,
            Resps0
        ),
        case is_replicated(Options) of
            true ->
                {ok,
                    lists:flatmap(
                        fun(R) ->
                            case R of
                                {ok, []} -> [];
                                {{_, _}, {ok, []}} -> [];
                                Else -> [Else]
                            end
                        end,
                        Resps1
                    )};
            false ->
                {ok, Resps1}
        end
    catch
        throw:{aborted, Errors} ->
            {aborted, Errors}
    end.

read_attachment(Db, DocId, AttId) ->
    fabric2_fdb:transactional(Db, fun(TxDb) ->
        fabric2_fdb:read_attachment(TxDb, DocId, AttId)
    end).

write_attachment(Db, DocId, Att) ->
    Data = couch_att:fetch(data, Att),
    Encoding = couch_att:fetch(encoding, Att),
    {ok, AttId} = fabric2_fdb:write_attachment(Db, DocId, Data, Encoding),
    couch_att:store(data, {loc, Db, DocId, AttId}, Att).

fold_docs(Db, UserFun, UserAcc) ->
    fold_docs(Db, UserFun, UserAcc, []).

fold_docs(Db, UserFun, UserAcc0, Options) ->
    fabric2_fdb:transactional(Db, fun(TxDb) ->
        try
            #{
                db_prefix := DbPrefix
            } = TxDb,

            Prefix = erlfdb_tuple:pack({?DB_ALL_DOCS}, DbPrefix),
            Meta = get_all_docs_meta(TxDb, Options),

            UserAcc1 = maybe_stop(UserFun({meta, Meta}, UserAcc0)),

            UserAcc2 = fabric2_fdb:fold_range(
                TxDb,
                Prefix,
                fun({K, V}, Acc) ->
                    {DocId} = erlfdb_tuple:unpack(K, Prefix),
                    RevId = erlfdb_tuple:unpack(V),
                    Row0 = [
                        {id, DocId},
                        {key, DocId},
                        {value, {[{rev, couch_doc:rev_to_str(RevId)}]}}
                    ],

                    DocOpts = couch_util:get_value(doc_opts, Options, []),
                    OpenOpts = [deleted | DocOpts],

                    Row1 =
                        case lists:keyfind(include_docs, 1, Options) of
                            {include_docs, true} ->
                                Row0 ++ open_json_doc(TxDb, DocId, OpenOpts, DocOpts);
                            _ ->
                                Row0
                        end,

                    maybe_stop(UserFun({row, Row1}, Acc))
                end,
                UserAcc1,
                Options
            ),

            {ok, maybe_stop(UserFun(complete, UserAcc2))}
        catch
            throw:{stop, FinalUserAcc} ->
                {ok, FinalUserAcc}
        end
    end).

fold_docs(Db, DocIds, UserFun, UserAcc0, Options) ->
    fabric2_fdb:transactional(Db, fun(TxDb) ->
        try
            NeedsTreeOpts = [revs_info, conflicts, deleted_conflicts],
            NeedsTree = (Options -- NeedsTreeOpts /= Options),

            InitAcc = #{
                revs_q => queue:new(),
                revs_count => 0,
                body_q => queue:new(),
                body_count => 0,
                doc_opts => Options,
                user_acc => UserAcc0,
                user_fun => UserFun
            },

            FinalAcc1 = lists:foldl(
                fun(DocId, Acc) ->
                    #{
                        revs_q := RevsQ,
                        revs_count := RevsCount
                    } = Acc,
                    Future = fold_docs_get_revs(TxDb, DocId, NeedsTree),
                    NewAcc = Acc#{
                        revs_q := queue:in({DocId, Future}, RevsQ),
                        revs_count := RevsCount + 1
                    },
                    drain_fold_docs_revs_futures(TxDb, NewAcc)
                end,
                InitAcc,
                DocIds
            ),

            FinalAcc2 = drain_all_fold_docs_revs_futures(TxDb, FinalAcc1),
            FinalAcc3 = drain_all_fold_docs_body_futures(TxDb, FinalAcc2),

            #{
                user_acc := FinalUserAcc
            } = FinalAcc3,
            {ok, FinalUserAcc}
        catch
            throw:{stop, StopUserAcc} ->
                {ok, StopUserAcc}
        end
    end).

fold_design_docs(Db, UserFun, UserAcc0, Options1) ->
    Options2 = set_design_doc_keys(Options1),
    fold_docs(Db, UserFun, UserAcc0, Options2).

fold_local_docs(Db, UserFun, UserAcc0, Options0) ->
    % This is mostly for testing and sanity checking. When calling from a test
    % namespace will be automatically set. We also assert when called from the
    % API the correct namespace was set
    Options =
        case lists:keyfind(namespace, 1, Options0) of
            {namespace, <<"_local">>} -> Options0;
            false -> [{namespace, <<"_local">>} | Options0]
        end,
    fabric2_fdb:transactional(Db, fun(TxDb) ->
        try
            #{
                db_prefix := DbPrefix
            } = TxDb,

            Prefix = erlfdb_tuple:pack({?DB_LOCAL_DOCS}, DbPrefix),
            Meta = get_all_docs_meta(TxDb, Options),

            UserAcc1 = maybe_stop(UserFun({meta, Meta}, UserAcc0)),

            UserAcc2 = fabric2_fdb:fold_range(
                TxDb,
                Prefix,
                fun({K, V}, Acc) ->
                    {DocId} = erlfdb_tuple:unpack(K, Prefix),
                    Rev = fabric2_fdb:get_local_doc_rev(TxDb, DocId, V),
                    maybe_stop(
                        UserFun(
                            {row, [
                                {id, DocId},
                                {key, DocId},
                                {value, {[{rev, couch_doc:rev_to_str({0, Rev})}]}}
                            ]},
                            Acc
                        )
                    )
                end,
                UserAcc1,
                Options
            ),

            {ok, maybe_stop(UserFun(complete, UserAcc2))}
        catch
            throw:{stop, FinalUserAcc} ->
                {ok, FinalUserAcc}
        end
    end).

fold_changes(Db, SinceSeq, UserFun, UserAcc) ->
    fold_changes(Db, SinceSeq, UserFun, UserAcc, []).

fold_changes(Db, SinceSeq, UserFun, UserAcc, Options) ->
    fabric2_fdb:transactional(Db, fun(TxDb) ->
        try
            #{
                db_prefix := DbPrefix
            } = TxDb,

            Prefix = erlfdb_tuple:pack({?DB_CHANGES}, DbPrefix),

            Dir =
                case fabric2_util:get_value(dir, Options, fwd) of
                    rev -> rev;
                    _ -> fwd
                end,

            RestartTx =
                case fabric2_util:get_value(restart_tx, Options) of
                    undefined -> [{restart_tx, true}];
                    _AlreadySet -> []
                end,

            StartKey = get_since_seq(TxDb, Dir, SinceSeq),
            EndKey =
                case fabric2_util:get_value(end_key, Options) of
                    undefined when Dir == rev ->
                        fabric2_util:seq_zero_vs();
                    undefined ->
                        fabric2_util:seq_max_vs();
                    EK when is_binary(EK) ->
                        fabric2_fdb:seq_to_vs(EK);
                    EK when is_tuple(EK), element(1, EK) == versionstamp ->
                        EK
                end,
            BaseOpts = [{start_key, StartKey}] ++ RestartTx ++ Options,
            FoldOpts = lists:keystore(end_key, 1, BaseOpts, {end_key, EndKey}),

            {ok,
                fabric2_fdb:fold_range(
                    TxDb,
                    Prefix,
                    fun({K, V}, Acc) ->
                        {SeqVS} = erlfdb_tuple:unpack(K, Prefix),
                        {DocId, Deleted, RevId} = erlfdb_tuple:unpack(V),

                        Change = #{
                            id => DocId,
                            sequence => fabric2_fdb:vs_to_seq(SeqVS),
                            rev_id => RevId,
                            deleted => Deleted
                        },

                        maybe_stop(UserFun(Change, Acc))
                    end,
                    UserAcc,
                    FoldOpts
                )}
        catch
            throw:{stop, FinalUserAcc} ->
                {ok, FinalUserAcc}
        end
    end).

dbname_suffix(DbName) ->
    filename:basename(DbName).

validate_dbname(DbName) when is_list(DbName) ->
    validate_dbname(?l2b(DbName));
validate_dbname(DbName) when is_binary(DbName) ->
    fabric2_db_plugin:validate_dbname(
        DbName, DbName, fun validate_dbname_int/2
    ).

validate_dbname_int(DbName, DbName) when is_binary(DbName) ->
    case validate_dbname_length(DbName) of
        ok -> validate_dbname_pat(DbName);
        {error, _} = Error -> Error
    end.

validate_dbname_length(DbName) ->
    MaxLength = config:get_integer(
        "couchdb",
        "max_database_name_length",
        ?DEFAULT_MAX_DATABASE_NAME_LENGTH
    ),
    case byte_size(DbName) =< MaxLength of
        true -> ok;
        false -> {error, {database_name_too_long, DbName}}
    end.

validate_dbname_pat(DbName) ->
    case re:run(DbName, ?DBNAME_REGEX, [{capture, none}, dollar_endonly]) of
        match ->
            ok;
        nomatch ->
            case is_system_db_name(DbName) of
                true -> ok;
                false -> {error, {illegal_database_name, DbName}}
            end
    end.

maybe_add_sys_db_callbacks(Db) ->
    IsReplicatorDb = is_replicator_db(Db),
    IsUsersDb = is_users_db(Db),

    {BDU, ADR} =
        if
            IsReplicatorDb ->
                {
                    fun couch_replicator_docs:before_doc_update/3,
                    fun couch_replicator_docs:after_doc_read/2
                };
            IsUsersDb ->
                {
                    fun fabric2_users_db:before_doc_update/3,
                    fun fabric2_users_db:after_doc_read/2
                };
            true ->
                {undefined, undefined}
        end,

    Db#{
        before_doc_update := BDU,
        after_doc_read := ADR
    }.

make_db_info(DbName, Props) ->
    BaseProps = [
        {cluster, {[{n, 0}, {q, 0}, {r, 0}, {w, 0}]}},
        {compact_running, false},
        {data_size, 0},
        {db_name, DbName},
        {disk_format_version, 0},
        {disk_size, 0},
        {instance_start_time, <<"0">>},
        {purge_seq, 0}
    ],

    lists:foldl(
        fun({Key, Val}, Acc) ->
            lists:keystore(Key, 1, Acc, {Key, Val})
        end,
        BaseProps,
        Props
    ).

drain_info_futures(FutureQ, Count, _UserFun, Acc) when Count < 100 ->
    {FutureQ, Count, Acc};
drain_info_futures(FutureQ, Count, UserFun, Acc) when Count >= 100 ->
    {{value, {DbName, Future}}, RestQ} = queue:out(FutureQ),
    InfoProps = fabric2_fdb:get_info_wait(Future),
    DbInfo = make_db_info(DbName, InfoProps),
    NewAcc = maybe_stop(UserFun({row, DbInfo}, Acc)),
    {RestQ, Count - 1, NewAcc}.

drain_all_info_futures(FutureQ, UserFun, Acc) ->
    case queue:out(FutureQ) of
        {{value, {DbName, Future}}, RestQ} ->
            InfoProps = fabric2_fdb:get_info_wait(Future),
            DbInfo = make_db_info(DbName, InfoProps),
            NewAcc = maybe_stop(UserFun({row, DbInfo}, Acc)),
            drain_all_info_futures(RestQ, UserFun, NewAcc);
        {empty, _} ->
            Acc
    end.

drain_deleted_info_futures(FutureQ, Count, _UserFun, Acc) when Count < 100 ->
    {FutureQ, Count, Acc};
drain_deleted_info_futures(FutureQ, Count, UserFun, Acc) when Count >= 100 ->
    {{value, {DbName, TimeStamp, Future}}, RestQ} = queue:out(FutureQ),
    BaseProps = fabric2_fdb:get_info_wait(Future),
    DeletedProps =
        BaseProps ++
            [
                {deleted, true},
                {timestamp, TimeStamp}
            ],
    DbInfo = make_db_info(DbName, DeletedProps),
    NewAcc = maybe_stop(UserFun({row, DbInfo}, Acc)),
    {RestQ, Count - 1, NewAcc}.

drain_all_deleted_info_futures(FutureQ, UserFun, Acc) ->
    case queue:out(FutureQ) of
        {{value, {DbName, TimeStamp, Future}}, RestQ} ->
            BaseProps = fabric2_fdb:get_info_wait(Future),
            DeletedProps =
                BaseProps ++
                    [
                        {deleted, true},
                        {timestamp, TimeStamp}
                    ],
            DbInfo = make_db_info(DbName, DeletedProps),
            NewAcc = maybe_stop(UserFun({row, DbInfo}, Acc)),
            drain_all_deleted_info_futures(RestQ, UserFun, NewAcc);
        {empty, _} ->
            Acc
    end.

fold_docs_get_revs(Db, <<?LOCAL_DOC_PREFIX, _/binary>> = DocId, _) ->
    fabric2_fdb:get_local_doc_rev_future(Db, DocId);
fold_docs_get_revs(Db, DocId, true) ->
    fabric2_fdb:get_all_revs_future(Db, DocId);
fold_docs_get_revs(Db, DocId, false) ->
    fabric2_fdb:get_winning_revs_future(Db, DocId, 1).

fold_docs_get_revs_wait(_Db, <<?LOCAL_DOC_PREFIX, _/binary>>, RevsFuture) ->
    Rev = fabric2_fdb:get_local_doc_rev_wait(RevsFuture),
    [Rev];
fold_docs_get_revs_wait(Db, _DocId, RevsFuture) ->
    fabric2_fdb:get_revs_wait(Db, RevsFuture).

fold_docs_get_doc_body_future(
    Db,
    <<?LOCAL_DOC_PREFIX, _/binary>> = DocId,
    [Rev]
) ->
    fabric2_fdb:get_local_doc_body_future(Db, DocId, Rev);
fold_docs_get_doc_body_future(Db, DocId, Revs) ->
    Winner = get_rev_winner(Revs),
    fabric2_fdb:get_doc_body_future(Db, DocId, Winner).

fold_docs_get_doc_body_wait(
    Db,
    <<?LOCAL_DOC_PREFIX, _/binary>> = DocId,
    [Rev],
    _DocOpts,
    BodyFuture
) ->
    case fabric2_fdb:get_local_doc_body_wait(Db, DocId, Rev, BodyFuture) of
        {not_found, missing} -> {not_found, missing};
        Doc -> {ok, Doc}
    end;
fold_docs_get_doc_body_wait(Db, DocId, Revs, DocOpts, BodyFuture) ->
    RevInfo = get_rev_winner(Revs),
    Base = fabric2_fdb:get_doc_body_wait(
        Db,
        DocId,
        RevInfo,
        BodyFuture
    ),
    apply_open_doc_opts(Base, Revs, DocOpts).

drain_fold_docs_revs_futures(_TxDb, #{revs_count := C} = Acc) when C < 100 ->
    Acc;
drain_fold_docs_revs_futures(TxDb, Acc) ->
    drain_one_fold_docs_revs_future(TxDb, Acc).

drain_all_fold_docs_revs_futures(_TxDb, #{revs_count := C} = Acc) when C =< 0 ->
    Acc;
drain_all_fold_docs_revs_futures(TxDb, #{revs_count := C} = Acc) when C > 0 ->
    NewAcc = drain_one_fold_docs_revs_future(TxDb, Acc),
    drain_all_fold_docs_revs_futures(TxDb, NewAcc).

drain_one_fold_docs_revs_future(TxDb, Acc) ->
    #{
        revs_q := RevsQ,
        revs_count := RevsCount,
        body_q := BodyQ,
        body_count := BodyCount
    } = Acc,
    {{value, {DocId, RevsFuture}}, RestRevsQ} = queue:out(RevsQ),

    Revs = fold_docs_get_revs_wait(TxDb, DocId, RevsFuture),
    DocFuture =
        case Revs of
            [] ->
                {DocId, [], not_found};
            [_ | _] ->
                BodyFuture = fold_docs_get_doc_body_future(TxDb, DocId, Revs),
                {DocId, Revs, BodyFuture}
        end,
    NewAcc = Acc#{
        revs_q := RestRevsQ,
        revs_count := RevsCount - 1,
        body_q := queue:in(DocFuture, BodyQ),
        body_count := BodyCount + 1
    },
    drain_fold_docs_body_futures(TxDb, NewAcc).

drain_fold_docs_body_futures(_TxDb, #{body_count := C} = Acc) when C < 100 ->
    Acc;
drain_fold_docs_body_futures(TxDb, Acc) ->
    drain_one_fold_docs_body_future(TxDb, Acc).

drain_all_fold_docs_body_futures(_TxDb, #{body_count := C} = Acc) when C =< 0 ->
    Acc;
drain_all_fold_docs_body_futures(TxDb, #{body_count := C} = Acc) when C > 0 ->
    NewAcc = drain_one_fold_docs_body_future(TxDb, Acc),
    drain_all_fold_docs_body_futures(TxDb, NewAcc).

drain_one_fold_docs_body_future(TxDb, Acc) ->
    #{
        body_q := BodyQ,
        body_count := BodyCount,
        doc_opts := DocOpts,
        user_fun := UserFun,
        user_acc := UserAcc
    } = Acc,
    {{value, {DocId, Revs, BodyFuture}}, RestBodyQ} = queue:out(BodyQ),
    Doc =
        case BodyFuture of
            not_found ->
                {not_found, missing};
            _ ->
                fold_docs_get_doc_body_wait(TxDb, DocId, Revs, DocOpts, BodyFuture)
        end,
    NewUserAcc = maybe_stop(UserFun(DocId, Doc, UserAcc)),
    Acc#{
        body_q := RestBodyQ,
        body_count := BodyCount - 1,
        user_acc := NewUserAcc
    }.

get_rev_winner(Revs) ->
    [Winner] = lists:filter(
        fun(Rev) ->
            maps:get(winner, Rev)
        end,
        Revs
    ),
    Winner.

new_revid(Db, Doc) ->
    #doc{
        id = DocId,
        body = Body,
        revs = {OldStart, OldRevs},
        atts = Atts,
        deleted = Deleted
    } = Doc,

    {NewAtts, AttSigInfo} = lists:mapfoldl(
        fun(Att, Acc) ->
            [Name, Type, Data, Md5] = couch_att:fetch([name, type, data, md5], Att),
            case Data of
                {loc, _, _, _} ->
                    {Att, [{Name, Type, Md5} | Acc]};
                _ ->
                    Att1 = couch_att:flush(Db, DocId, Att),
                    Att2 = couch_att:store(revpos, OldStart + 1, Att1),
                    {Att2, [{Name, Type, couch_att:fetch(md5, Att2)} | Acc]}
            end
        end,
        [],
        Atts
    ),

    Rev =
        case length(Atts) == length(AttSigInfo) of
            true ->
                OldRev =
                    case OldRevs of
                        [] -> 0;
                        [OldRev0 | _] -> OldRev0
                    end,
                SigTerm = [Deleted, OldStart, OldRev, Body, AttSigInfo],
                couch_hash:md5_hash(term_to_binary(SigTerm, [{minor_version, 1}]));
            false ->
                erlang:error(missing_att_info)
        end,

    Doc#doc{
        revs = {OldStart + 1, [Rev | OldRevs]},
        atts = NewAtts
    }.

get_all_docs_meta(TxDb, Options) ->
    NS = couch_util:get_value(namespace, Options),
    DocCount = get_doc_count(TxDb, NS),
    case lists:keyfind(update_seq, 1, Options) of
        {_, true} ->
            UpdateSeq = fabric2_db:get_update_seq(TxDb),
            [{update_seq, UpdateSeq}];
        _ ->
            []
    end ++ [{total, DocCount}, {offset, null}].

maybe_set_interactive(#{} = Db, Options) ->
    Interactive = fabric2_util:get_value(interactive, Options, false),
    Db#{interactive := Interactive}.

maybe_set_user_ctx(Db, Options) ->
    case fabric2_util:get_value(user_ctx, Options) of
        #user_ctx{} = UserCtx ->
            set_user_ctx(Db, UserCtx);
        undefined ->
            Db
    end.

is_member(Db, {SecProps}) when is_list(SecProps) ->
    case is_admin(Db, {SecProps}) of
        true ->
            true;
        false ->
            case is_public_db(SecProps) of
                true ->
                    true;
                false ->
                    {Members} = get_members(SecProps),
                    UserCtx = get_user_ctx(Db),
                    is_authorized(Members, UserCtx)
            end
    end.

is_authorized(Group, UserCtx) ->
    #user_ctx{
        name = UserName,
        roles = UserRoles
    } = UserCtx,
    Names = fabric2_util:get_value(<<"names">>, Group, []),
    Roles = fabric2_util:get_value(<<"roles">>, Group, []),
    case check_security(roles, UserRoles, [<<"_admin">> | Roles]) of
        true ->
            true;
        false ->
            check_security(names, UserName, Names)
    end.

check_security(roles, [], _) ->
    false;
check_security(roles, UserRoles, Roles) ->
    UserRolesSet = ordsets:from_list(UserRoles),
    RolesSet = ordsets:from_list(Roles),
    not ordsets:is_disjoint(UserRolesSet, RolesSet);
check_security(names, _, []) ->
    false;
check_security(names, null, _) ->
    false;
check_security(names, UserName, Names) ->
    lists:member(UserName, Names).

throw_security_error(#user_ctx{name = null} = UserCtx) ->
    Reason = <<"You are not authorized to access this db.">>,
    throw_security_error(UserCtx, Reason);
throw_security_error(#user_ctx{name = _} = UserCtx) ->
    Reason = <<"You are not allowed to access this db.">>,
    throw_security_error(UserCtx, Reason).

throw_security_error(#user_ctx{} = UserCtx, Reason) ->
    Error = security_error_type(UserCtx),
    throw({Error, Reason}).

security_error_type(#user_ctx{name = null}) ->
    unauthorized;
security_error_type(#user_ctx{name = _}) ->
    forbidden.

is_public_db(SecProps) ->
    {Members} = get_members(SecProps),
    Names = fabric2_util:get_value(<<"names">>, Members, []),
    Roles = fabric2_util:get_value(<<"roles">>, Members, []),
    Names =:= [] andalso Roles =:= [].

get_admins(SecProps) ->
    fabric2_util:get_value(<<"admins">>, SecProps, {[]}).

get_members(SecProps) ->
    % we fallback to readers here for backwards compatibility
    case fabric2_util:get_value(<<"members">>, SecProps) of
        undefined ->
            fabric2_util:get_value(<<"readers">>, SecProps, {[]});
        Members ->
            Members
    end.

apply_open_doc_opts(Doc0, Revs, Options) ->
    IncludeRevsInfo = lists:member(revs_info, Options),
    IncludeConflicts = lists:member(conflicts, Options),
    IncludeDelConflicts = lists:member(deleted_conflicts, Options),
    IncludeLocalSeq = lists:member(local_seq, Options),

    % This revs_info becomes fairly useless now that we're
    % not keeping old document bodies around...
    Meta1 =
        if
            not IncludeRevsInfo ->
                [];
            true ->
                {Pos, [Rev | RevPath]} = Doc0#doc.revs,
                RevPathMissing = lists:map(fun(R) -> {R, missing} end, RevPath),
                [{revs_info, Pos, [{Rev, available} | RevPathMissing]}]
        end,

    Meta2 =
        if
            not IncludeConflicts ->
                [];
            true ->
                Conflicts = [RI || RI = #{winner := false, deleted := false} <- Revs],
                if
                    Conflicts == [] ->
                        [];
                    true ->
                        ConflictRevs = [maps:get(rev_id, RI) || RI <- Conflicts],
                        [{conflicts, ConflictRevs}]
                end
        end,

    Meta3 =
        if
            not IncludeDelConflicts ->
                [];
            true ->
                DelConflicts = [RI || RI = #{winner := false, deleted := true} <- Revs],
                if
                    DelConflicts == [] ->
                        [];
                    true ->
                        DelConflictRevs = [maps:get(rev_id, RI) || RI <- DelConflicts],
                        [{deleted_conflicts, DelConflictRevs}]
                end
        end,

    Meta4 =
        if
            not IncludeLocalSeq ->
                [];
            true ->
                #{winner := true, sequence := SeqVS} = lists:last(Revs),
                [{local_seq, fabric2_fdb:vs_to_seq(SeqVS)}]
        end,

    Doc1 =
        case lists:keyfind(atts_since, 1, Options) of
            {_, PossibleAncestors} ->
                #doc{
                    revs = DocRevs,
                    atts = Atts0
                } = Doc0,
                RevPos = find_ancestor_rev_pos(DocRevs, PossibleAncestors),
                Atts1 = lists:map(
                    fun(Att) ->
                        [AttPos, Data] = couch_att:fetch([revpos, data], Att),
                        if
                            AttPos > RevPos -> couch_att:store(data, Data, Att);
                            true -> couch_att:store(data, stub, Att)
                        end
                    end,
                    Atts0
                ),
                Doc0#doc{atts = Atts1};
            false ->
                Doc0
        end,

    {ok, Doc1#doc{meta = Meta1 ++ Meta2 ++ Meta3 ++ Meta4}}.

find_ancestor_rev_pos({_, []}, _PossibleAncestors) ->
    0;
find_ancestor_rev_pos(_DocRevs, []) ->
    0;
find_ancestor_rev_pos({RevPos, [RevId | Rest]}, AttsSinceRevs) ->
    case lists:member({RevPos, RevId}, AttsSinceRevs) of
        true -> RevPos;
        false -> find_ancestor_rev_pos({RevPos - 1, Rest}, AttsSinceRevs)
    end.

filter_found_revs(RevInfo, Revs) ->
    #{
        rev_id := {Pos, Rev},
        rev_path := RevPath
    } = RevInfo,
    FullRevPath = [Rev | RevPath],
    lists:flatmap(
        fun({FindPos, FindRev} = RevIdToFind) ->
            if
                FindPos > Pos ->
                    [RevIdToFind];
                true ->
                    % Add 1 because lists:nth is 1 based
                    Idx = Pos - FindPos + 1,
                    case Idx > length(FullRevPath) of
                        true ->
                            [RevIdToFind];
                        false ->
                            case lists:nth(Idx, FullRevPath) == FindRev of
                                true -> [];
                                false -> [RevIdToFind]
                            end
                    end
            end
        end,
        Revs
    ).

find_possible_ancestors(RevInfos, MissingRevs) ->
    % Find any revinfos that are possible ancestors
    % of the missing revs. A possible ancestor is
    % any rev that has a start position less than
    % any missing revision. Stated alternatively,
    % find any revinfo that could theoretically
    % extended to be one or more of the missing
    % revisions.
    %
    % Since we are looking at any missing revision
    % we can just compare against the maximum missing
    % start position.
    MaxMissingPos =
        case MissingRevs of
            [] -> 0;
            [_ | _] -> lists:max([Start || {Start, _Rev} <- MissingRevs])
        end,
    lists:flatmap(
        fun(RevInfo) ->
            #{rev_id := {RevPos, _} = RevId} = RevInfo,
            case RevPos < MaxMissingPos of
                true -> [RevId];
                false -> []
            end
        end,
        RevInfos
    ).

apply_before_doc_update(Db, Docs, Options) ->
    UpdateType =
        case lists:member(replicated_changes, Options) of
            true -> replicated_changes;
            false -> interactive_edit
        end,
    lists:map(
        fun(Doc) ->
            fabric2_db_plugin:before_doc_update(Db, Doc, UpdateType)
        end,
        Docs
    ).

update_doc_int(#{} = Db, #doc{} = Doc, Options) ->
    IsLocal =
        case Doc#doc.id of
            <<?LOCAL_DOC_PREFIX, _/binary>> -> true;
            _ -> false
        end,
    try
        case {IsLocal, is_replicated(Options)} of
            {false, false} -> update_doc_interactive(Db, Doc, Options);
            {false, true} -> update_doc_replicated(Db, Doc, Options);
            {true, _} -> update_local_doc(Db, Doc, Options)
        end
    catch
        throw:{?MODULE, Return} ->
            Return
    end.

batch_update_docs(Db, Docs, Options) ->
    BAcc = #bacc{
        db = Db,
        docs = Docs,
        batch_size = get_batch_size(Options),
        options = Options,
        rev_futures = #{},
        seen = [],
        results = []
    },
    #bacc{results = Res} = batch_update_docs(BAcc),
    lists:reverse(Res).

batch_update_docs(#bacc{docs = []} = BAcc) ->
    BAcc;
batch_update_docs(#bacc{db = Db} = BAcc) ->
    #bacc{
        db = Db,
        docs = Docs,
        options = Options
    } = BAcc,

    BAccTx2 = fabric2_fdb:transactional(Db, fun(TxDb) ->
        BAccTx = BAcc#bacc{db = TxDb},
        case is_replicated(Options) of
            false ->
                Tagged = tag_docs(Docs),
                RevFutures = get_winning_rev_futures(TxDb, Tagged),
                BAccTx1 = BAccTx#bacc{
                    docs = Tagged,
                    rev_futures = RevFutures
                },
                batch_update_interactive_tx(BAccTx1);
            true ->
                BAccTx1 = batch_update_replicated_tx(BAccTx),
                % For replicated updates reset `seen` after every transaction
                BAccTx1#bacc{seen = []}
        end
    end),

    % Clean up after the transaction ends so we can recurse with a clean state
    maps:map(
        fun(Tag, RangeFuture) when is_reference(Tag) ->
            ok = erlfdb:cancel(RangeFuture, [flush])
        end,
        BAccTx2#bacc.rev_futures
    ),

    BAcc1 = BAccTx2#bacc{
        db = Db,
        rev_futures = #{}
    },

    batch_update_docs(BAcc1).

batch_update_interactive_tx(#bacc{docs = []} = BAcc) ->
    BAcc;
batch_update_interactive_tx(#bacc{} = BAcc) ->
    #bacc{
        db = TxDb,
        docs = [Doc | Docs],
        options = Options,
        batch_size = MaxSize,
        rev_futures = RevFutures,
        seen = Seen,
        results = Results
    } = BAcc,
    {Res, Seen1} =
        try
            update_docs_interactive(TxDb, Doc, Options, RevFutures, Seen)
        catch
            throw:{?MODULE, Return} ->
                {Return, Seen}
        end,
    BAcc1 = BAcc#bacc{
        docs = Docs,
        results = [Res | Results],
        seen = Seen1
    },
    case fabric2_fdb:get_approximate_tx_size(TxDb) > MaxSize of
        true -> BAcc1;
        false -> batch_update_interactive_tx(BAcc1)
    end.

batch_update_replicated_tx(#bacc{docs = []} = BAcc) ->
    BAcc;
batch_update_replicated_tx(#bacc{} = BAcc) ->
    #bacc{
        db = TxDb,
        docs = [Doc | Docs],
        options = Options,
        batch_size = MaxSize,
        seen = Seen,
        results = Results
    } = BAcc,
    case lists:member(Doc#doc.id, Seen) of
        true ->
            % If we already updated this doc in the current transaction, wait
            % till the next transaction to update it again.
            BAcc;
        false ->
            Res = update_doc_int(TxDb, Doc, Options),
            BAcc1 = BAcc#bacc{
                docs = Docs,
                results = [Res | Results],
                seen = [Doc#doc.id | Seen]
            },
            case fabric2_fdb:get_approximate_tx_size(TxDb) > MaxSize of
                true -> BAcc1;
                false -> batch_update_replicated_tx(BAcc1)
            end
    end.

update_docs_interactive(
    Db,
    #doc{id = <<?LOCAL_DOC_PREFIX, _/binary>>} = Doc,
    Options,
    _Futures,
    SeenIds
) ->
    {update_local_doc(Db, Doc, Options), SeenIds};
update_docs_interactive(Db, Doc, Options, Futures, SeenIds) ->
    case lists:member(Doc#doc.id, SeenIds) of
        true ->
            {conflict, SeenIds};
        false ->
            Future = maps:get(doc_tag(Doc), Futures),
            case update_doc_interactive(Db, Doc, Future, Options) of
                {ok, _} = Resp ->
                    {Resp, [Doc#doc.id | SeenIds]};
                _ = Resp ->
                    {Resp, SeenIds}
            end
    end.

update_doc_interactive(Db, Doc0, Options) ->
    % Get the current winning revision. This is needed
    % regardless of which branch we're updating. The extra
    % revision we're grabbing is an optimization to
    % save us a round trip if we end up deleting
    % the winning revision branch.
    NumRevs =
        if
            Doc0#doc.deleted -> 2;
            true -> 1
        end,
    Future = fabric2_fdb:get_winning_revs_future(Db, Doc0#doc.id, NumRevs),
    update_doc_interactive(Db, Doc0, Future, Options).

update_doc_interactive(Db, Doc0, Future, _Options) ->
    RevInfos = fabric2_fdb:get_revs_wait(Db, Future),
    {Winner, SecondPlace} =
        case RevInfos of
            [] -> {not_found, not_found};
            [WRI] -> {WRI, not_found};
            [WRI, SPRI] -> {WRI, SPRI}
        end,
    WinnerRevId =
        case Winner of
            not_found ->
                {0, <<>>};
            _ ->
                case maps:get(deleted, Winner) of
                    true -> {0, <<>>};
                    false -> maps:get(rev_id, Winner)
                end
        end,

    % Check that a revision was specified if required
    Doc0RevId = doc_to_revid(Doc0),
    HasRev = Doc0RevId =/= {0, <<>>},
    if
        HasRev orelse WinnerRevId == {0, <<>>} -> ok;
        true -> ?RETURN({Doc0, conflict})
    end,

    % Allow inserting new deleted documents. Only works when the document has
    % never existed to match CouchDB 3.x
    case not HasRev andalso Doc0#doc.deleted andalso is_map(Winner) of
        true -> ?RETURN({Doc0, conflict});
        false -> ok
    end,

    % Get the target revision to update
    Target =
        case Doc0RevId == WinnerRevId of
            true ->
                Winner;
            false ->
                case fabric2_fdb:get_non_deleted_rev(Db, Doc0#doc.id, Doc0RevId) of
                    #{deleted := false} = Target0 ->
                        Target0;
                    not_found ->
                        % Either a missing revision or a deleted
                        % revision. Either way a conflict. Note
                        % that we get not_found for a deleted revision
                        % because we only check for the non-deleted
                        % key in fdb
                        ?RETURN({Doc0, conflict})
                end
        end,

    Doc1 =
        case Winner of
            #{deleted := true} when not Doc0#doc.deleted ->
                % When recreating a deleted document we want to extend
                % the winning revision branch rather than create a
                % new branch. If we did not do this we could be
                % recreating into a state that previously existed.
                Doc0#doc{revs = fabric2_util:revinfo_to_revs(Winner)};
            #{} ->
                % Otherwise we're extending the target's revision
                % history with this update
                Doc0#doc{revs = fabric2_util:revinfo_to_revs(Target)};
            not_found ->
                % Creating a new doc means our revs start empty
                Doc0
        end,

    % Validate the doc update and create the
    % new revinfo map
    Doc2 = prep_and_validate(Db, Doc1, Target),

    Doc3 = new_revid(Db, Doc2),

    #doc{
        deleted = NewDeleted,
        revs = {NewRevPos, [NewRev | NewRevPath]},
        atts = Atts
    } = Doc4 = stem_revisions(Db, Doc3),

    NewRevInfo = #{
        winner => undefined,
        exists => false,
        deleted => NewDeleted,
        rev_id => {NewRevPos, NewRev},
        rev_path => NewRevPath,
        sequence => undefined,
        branch_count => undefined,
        att_hash => fabric2_util:hash_atts(Atts),
        rev_size => fabric2_util:rev_size(Doc4)
    },

    % Gather the list of possible winnig revisions
    Possible =
        case Target == Winner of
            true when not Doc4#doc.deleted ->
                [NewRevInfo];
            true when Doc4#doc.deleted ->
                case SecondPlace of
                    #{} -> [NewRevInfo, SecondPlace];
                    not_found -> [NewRevInfo]
                end;
            false ->
                [NewRevInfo, Winner]
        end,

    % Sort the rev infos such that the winner is first
    {NewWinner0, NonWinner} =
        case fabric2_util:sort_revinfos(Possible) of
            [W] -> {W, not_found};
            [W, NW] -> {W, NW}
        end,

    BranchCount =
        case Winner of
            not_found -> 1;
            #{branch_count := BC} -> BC
        end,
    NewWinner = NewWinner0#{branch_count := BranchCount},
    ToUpdate =
        if
            NonWinner == not_found -> [];
            true -> [NonWinner]
        end,
    ToRemove =
        if
            Target == not_found -> [];
            true -> [Target]
        end,

    ok = fabric2_fdb:write_doc(
        Db,
        Doc4,
        NewWinner,
        Winner,
        ToUpdate,
        ToRemove
    ),

    {ok, {NewRevPos, NewRev}}.

update_doc_replicated(Db, Doc0, _Options) ->
    #doc{
        id = DocId,
        deleted = Deleted,
        revs = {RevPos, [Rev | RevPath]}
    } = Doc0,

    DocRevInfo0 = #{
        winner => undefined,
        exists => false,
        deleted => Deleted,
        rev_id => {RevPos, Rev},
        rev_path => RevPath,
        sequence => undefined,
        branch_count => undefined,
        att_hash => <<>>,
        rev_size => null
    },

    AllRevInfos = fabric2_fdb:get_all_revs(Db, DocId),

    RevTree = lists:foldl(
        fun(RI, TreeAcc) ->
            RIPath = fabric2_util:revinfo_to_path(RI),
            {Merged, _} = couch_key_tree:merge(TreeAcc, RIPath),
            Merged
        end,
        [],
        AllRevInfos
    ),

    DocRevPath = fabric2_util:revinfo_to_path(DocRevInfo0),

    {NewTree, Status} = couch_key_tree:merge(RevTree, DocRevPath),
    if
        Status /= internal_node ->
            ok;
        true ->
            % We already know this revision so nothing
            % left to do.
            ?RETURN({Doc0, {ok, []}})
    end,

    % Its possible to have a replication with fewer than $revs_limit
    % revisions which extends an existing branch. To avoid
    % losing revision history we extract the new node from the
    % tree and use the combined path after stemming.
    {[{_, {RevPos, UnstemmedRevs}}], []} =
        couch_key_tree:get(NewTree, [{RevPos, Rev}]),

    Doc1 = stem_revisions(Db, Doc0#doc{revs = {RevPos, UnstemmedRevs}}),

    {RevPos, [Rev | NewRevPath]} = Doc1#doc.revs,
    DocRevInfo1 = DocRevInfo0#{rev_path := NewRevPath},

    % Find any previous revision we knew about for
    % validation and attachment handling.
    AllLeafsFull = couch_key_tree:get_all_leafs_full(NewTree),
    LeafPath = get_leaf_path(RevPos, Rev, AllLeafsFull),
    PrevRevInfo = find_prev_revinfo(RevPos, LeafPath),
    Doc2 = prep_and_validate(Db, Doc1, PrevRevInfo),
    Doc3 = flush_doc_atts(Db, Doc2),
    DocRevInfo2 = DocRevInfo1#{
        atts_hash => fabric2_util:hash_atts(Doc3#doc.atts),
        rev_size => fabric2_util:rev_size(Doc3)
    },

    % Possible winners are the previous winner and
    % the new DocRevInfo
    Winner =
        case fabric2_util:sort_revinfos(AllRevInfos) of
            [#{winner := true} = WRI | _] -> WRI;
            [] -> not_found
        end,
    {NewWinner0, NonWinner} =
        case Winner == PrevRevInfo of
            true ->
                {DocRevInfo2, not_found};
            false ->
                [W, NW] = fabric2_util:sort_revinfos([Winner, DocRevInfo2]),
                {W, NW}
        end,

    NewWinner = NewWinner0#{branch_count := length(AllLeafsFull)},
    ToUpdate =
        if
            NonWinner == not_found -> [];
            true -> [NonWinner]
        end,
    ToRemove =
        if
            PrevRevInfo == not_found -> [];
            true -> [PrevRevInfo]
        end,

    ok = fabric2_fdb:write_doc(
        Db,
        Doc3,
        NewWinner,
        Winner,
        ToUpdate,
        ToRemove
    ),

    {ok, []}.

update_local_doc(Db, Doc0, _Options) ->
    Doc1 =
        case increment_local_doc_rev(Doc0) of
            {ok, Updated} -> Updated;
            {error, Error} -> ?RETURN({Doc0, Error})
        end,

    ok = fabric2_fdb:write_local_doc(Db, Doc1),

    #doc{revs = {0, [Rev]}} = Doc1,
    {ok, {0, integer_to_binary(Rev)}}.

flush_doc_atts(Db, Doc) ->
    #doc{
        id = DocId,
        atts = Atts
    } = Doc,
    NewAtts = lists:map(
        fun(Att) ->
            case couch_att:fetch(data, Att) of
                {loc, _, _, _} ->
                    Att;
                _ ->
                    couch_att:flush(Db, DocId, Att)
            end
        end,
        Atts
    ),
    Doc#doc{atts = NewAtts}.

get_winning_rev_futures(Db, Docs) ->
    lists:foldl(
        fun(Doc, Acc) ->
            #doc{
                id = DocId,
                deleted = Deleted
            } = Doc,
            IsLocal =
                case DocId of
                    <<?LOCAL_DOC_PREFIX, _/binary>> -> true;
                    _ -> false
                end,
            if
                IsLocal ->
                    Acc;
                true ->
                    NumRevs =
                        if
                            Deleted -> 2;
                            true -> 1
                        end,
                    Future = fabric2_fdb:get_winning_revs_future(Db, DocId, NumRevs),
                    DocTag = doc_tag(Doc),
                    Acc#{DocTag => Future}
            end
        end,
        #{},
        Docs
    ).

prep_and_validate(Db, NewDoc, PrevRevInfo) ->
    HasStubs = couch_doc:has_stubs(NewDoc),
    HasVDUs = [] /= maps:get(validate_doc_update_funs, Db),
    IsDDoc =
        case NewDoc#doc.id of
            <<?DESIGN_DOC_PREFIX, _/binary>> -> true;
            _ -> false
        end,

    WasDeleted =
        case PrevRevInfo of
            not_found -> false;
            #{deleted := D} -> D
        end,

    PrevDoc =
        case HasStubs orelse (HasVDUs and not IsDDoc) of
            true when PrevRevInfo /= not_found, not WasDeleted ->
                case fabric2_fdb:get_doc_body(Db, NewDoc#doc.id, PrevRevInfo) of
                    #doc{} = PDoc -> PDoc;
                    {not_found, _} -> nil
                end;
            _ ->
                nil
        end,

    MergedDoc =
        if
            not HasStubs ->
                NewDoc;
            true ->
                % This will throw an error if we have any
                % attachment stubs missing data
                couch_doc:merge_stubs(NewDoc, PrevDoc)
        end,
    check_duplicate_attachments(MergedDoc),
    validate_doc_update(Db, MergedDoc, PrevDoc),
    MergedDoc.

validate_doc_update(Db, #doc{id = <<"_design/", _/binary>>} = Doc, _) ->
    case catch check_is_admin(Db) of
        ok -> validate_ddoc(Db, Doc);
        Error -> ?RETURN({Doc, Error})
    end;
validate_doc_update(Db, Doc, PrevDoc) ->
    #{
        security_doc := Security,
        validate_doc_update_funs := VDUs
    } = Db,
    Fun = fun() ->
        JsonCtx = fabric2_util:user_ctx_to_json(Db),
        lists:map(
            fun(VDU) ->
                try
                    case VDU(Doc, PrevDoc, JsonCtx, Security) of
                        ok -> ok;
                        Error1 -> throw(Error1)
                    end
                catch
                    throw:Error2 ->
                        ?RETURN({Doc, Error2})
                end
            end,
            VDUs
        )
    end,
    Stat = [couchdb, query_server, vdu_process_time],
    if
        VDUs == [] -> ok;
        true -> couch_stats:update_histogram(Stat, Fun)
    end.

validate_ddoc(Db, DDoc) ->
    try
        ok = couch_views_validate:validate_ddoc(Db, DDoc)
    catch
        throw:{invalid_design_doc, Reason} ->
            throw({bad_request, invalid_design_doc, Reason});
        throw:{compilation_error, Reason} ->
            throw({bad_request, compilation_error, Reason});
        throw:Error ->
            ?RETURN({DDoc, Error})
    end.

validate_atomic_update(_, false) ->
    ok;
validate_atomic_update(AllDocs, true) ->
    % TODO actually perform the validation.  This requires some hackery, we need
    % to basically extract the prep_and_validate_updates function from couch_db
    % and only run that, without actually writing in case of a success.
    Error = {not_implemented, <<"all_or_nothing is not supported">>},
    PreCommitFailures = lists:map(
        fun(#doc{id = Id, revs = {Pos, Revs}}) ->
            case Revs of
                [] -> RevId = <<>>;
                [RevId | _] -> ok
            end,
            {{Id, {Pos, RevId}}, Error}
        end,
        AllDocs
    ),
    throw({aborted, PreCommitFailures}).

check_duplicate_attachments(#doc{atts = Atts}) ->
    lists:foldl(
        fun(Att, Names) ->
            Name = couch_att:fetch(name, Att),
            case ordsets:is_element(Name, Names) of
                true -> throw({bad_request, <<"Duplicate attachments">>});
                false -> ordsets:add_element(Name, Names)
            end
        end,
        ordsets:new(),
        Atts
    ).

get_since_seq(Db, rev, <<>>) ->
    get_since_seq(Db, rev, now);
get_since_seq(_Db, _Dir, Seq) when Seq == <<>>; Seq == <<"0">>; Seq == 0 ->
    fabric2_util:seq_zero_vs();
get_since_seq(Db, Dir, Seq) when Seq == now; Seq == <<"now">> ->
    CurrSeq = fabric2_fdb:get_last_change(Db),
    get_since_seq(Db, Dir, CurrSeq);
get_since_seq(_Db, _Dir, Seq) when is_binary(Seq), size(Seq) == 24 ->
    fabric2_fdb:next_vs(fabric2_fdb:seq_to_vs(Seq));
get_since_seq(Db, Dir, List) when is_list(List) ->
    get_since_seq(Db, Dir, list_to_binary(List));
get_since_seq(_Db, _Dir, Seq) ->
    erlang:error({invalid_since_seq, Seq}).

get_leaf_path(Pos, Rev, [{Pos, [{Rev, _RevInfo} | LeafPath]} | _]) ->
    LeafPath;
get_leaf_path(Pos, Rev, [_WrongLeaf | RestLeafs]) ->
    get_leaf_path(Pos, Rev, RestLeafs).

find_prev_revinfo(_Pos, []) ->
    not_found;
find_prev_revinfo(Pos, [{_Rev, ?REV_MISSING} | RestPath]) ->
    find_prev_revinfo(Pos - 1, RestPath);
find_prev_revinfo(_Pos, [{_Rev, #{} = RevInfo} | _]) ->
    RevInfo.

increment_local_doc_rev(#doc{deleted = true} = Doc) ->
    {ok, Doc#doc{revs = {0, [0]}}};
increment_local_doc_rev(#doc{revs = {0, []}} = Doc) ->
    {ok, Doc#doc{revs = {0, [1]}}};
increment_local_doc_rev(#doc{revs = {0, [RevStr | _]}} = Doc) ->
    try
        PrevRev = binary_to_integer(RevStr),
        {ok, Doc#doc{revs = {0, [PrevRev + 1]}}}
    catch
        error:badarg ->
            {error, <<"Invalid rev format">>}
    end;
increment_local_doc_rev(#doc{}) ->
    {error, <<"Invalid rev format">>}.

doc_to_revid(#doc{revs = Revs}) ->
    case Revs of
        {0, []} -> {0, <<>>};
        {RevPos, [Rev | _]} -> {RevPos, Rev}
    end.

tag_docs([]) ->
    [];
tag_docs([#doc{meta = Meta} = Doc | Rest]) ->
    Meta1 = lists:keystore(ref, 1, Meta, {ref, make_ref()}),
    NewDoc = Doc#doc{meta = Meta1},
    [NewDoc | tag_docs(Rest)].

doc_tag(#doc{meta = Meta}) ->
    fabric2_util:get_value(ref, Meta).

idrevs({Id, Revs}) when is_list(Revs) ->
    {docid(Id), [rev(R) || R <- Revs]}.

docid(DocId) when is_list(DocId) ->
    list_to_binary(DocId);
docid(DocId) ->
    DocId.

rev(Rev) when is_list(Rev); is_binary(Rev) ->
    couch_doc:parse_rev(Rev);
rev({Seq, Hash} = Rev) when is_integer(Seq), is_binary(Hash) ->
    Rev.

maybe_stop({ok, Acc}) ->
    Acc;
maybe_stop({stop, Acc}) ->
    throw({stop, Acc}).

set_design_doc_keys(Options1) ->
    Dir = couch_util:get_value(dir, Options1, fwd),
    Options2 = set_design_doc_start_key(Options1, Dir),
    set_design_doc_end_key(Options2, Dir).

set_design_doc_start_key(Options, fwd) ->
    Key1 = couch_util:get_value(start_key, Options, ?FIRST_DDOC_KEY),
    Key2 = max(Key1, ?FIRST_DDOC_KEY),
    lists:keystore(start_key, 1, Options, {start_key, Key2});
set_design_doc_start_key(Options, rev) ->
    Key1 = couch_util:get_value(start_key, Options, ?LAST_DDOC_KEY),
    Key2 = min(Key1, ?LAST_DDOC_KEY),
    lists:keystore(start_key, 1, Options, {start_key, Key2}).

set_design_doc_end_key(Options, fwd) ->
    case couch_util:get_value(end_key_gt, Options) of
        undefined ->
            Key1 = couch_util:get_value(end_key, Options, ?LAST_DDOC_KEY),
            Key2 = min(Key1, ?LAST_DDOC_KEY),
            lists:keystore(end_key, 1, Options, {end_key, Key2});
        EKeyGT ->
            Key2 = min(EKeyGT, ?LAST_DDOC_KEY),
            lists:keystore(end_key_gt, 1, Options, {end_key_gt, Key2})
    end;
set_design_doc_end_key(Options, rev) ->
    case couch_util:get_value(end_key_gt, Options) of
        undefined ->
            Key1 = couch_util:get_value(end_key, Options, ?FIRST_DDOC_KEY),
            Key2 = max(Key1, ?FIRST_DDOC_KEY),
            lists:keystore(end_key, 1, Options, {end_key, Key2});
        EKeyGT ->
            Key2 = max(EKeyGT, ?FIRST_DDOC_KEY),
            lists:keystore(end_key_gt, 1, Options, {end_key_gt, Key2})
    end.

stem_revisions(#{} = Db, #doc{} = Doc) ->
    #{revs_limit := RevsLimit} = Db,
    #doc{revs = {RevPos, Revs}} = Doc,
    case RevPos >= RevsLimit of
        true -> Doc#doc{revs = {RevPos, lists:sublist(Revs, RevsLimit)}};
        false -> Doc
    end.

open_json_doc(Db, DocId, OpenOpts, DocOpts) ->
    case fabric2_db:open_doc(Db, DocId, OpenOpts) of
        {not_found, missing} ->
            [];
        {ok, #doc{deleted = true}} ->
            [{doc, null}];
        {ok, #doc{} = Doc} ->
            [{doc, couch_doc:to_json_obj(Doc, DocOpts)}]
    end.

get_cached_db(#{} = Db, Opts) when is_list(Opts) ->
    MaxAge = fabric2_util:get_value(max_age, Opts, 0),
    Now = erlang:monotonic_time(millisecond),
    Age = Now - maps:get(check_current_ts, Db),
    case Age < MaxAge of
        true ->
            Db;
        false ->
            fabric2_fdb:transactional(Db, fun(TxDb) ->
                fabric2_fdb:ensure_current(TxDb)
            end)
    end.

is_replicated(Options) when is_list(Options) ->
    lists:member(replicated_changes, Options).

get_batch_size(Options) ->
    case fabric2_util:get_value(batch_size, Options) of
        undefined ->
            config:get_integer(
                "fabric",
                "update_docs_batch_size",
                ?DEFAULT_UPDATE_DOCS_BATCH_SIZE
            );
        Val when is_integer(Val) ->
            Val
    end.