Implement _design_docs and _local_docsprototype/fdb-layer-add-design-all-docs-namespacing

`_design_docs` reuses `_all_docs` logic and adjusts `start_key` and
`end_key` to be within the `_design/` prefix range.

Namespace setting was simplified to never have an `undefined`
value. This way it doesn't need extra case statements to handle it
further down in the FDB code.

4 files changed, 243 insertions, 36 deletions

diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl
index 183758512..04da99c19 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -859,22 +859,21 @@ all_docs_view(Req, Db, Keys, OP) ->
 
 send_all_docs(Db, #mrargs{keys = undefined} = Args, UserCtx, VAcc0) ->
     Opts = all_docs_view_opts(Args, UserCtx),
-    Acc = {iter, Db, Args, VAcc0},
+    NS = couch_util:get_value(namespace, Opts),
+    FoldFun = case NS of
+        <<"_all_docs">> -> fold_docs;
+        <<"_design">> -> fold_design_docs;
+        <<"_local">> -> fold_local_docs
+    end,
     ViewCb = fun view_cb/2,
-    {ok, {iter, _, _, VAcc1}} = fabric2_db:fold_docs(Db, ViewCb, Acc, Opts),
+    Acc = {iter, Db, Args, VAcc0},
+    {ok, {iter, _, _, VAcc1}} = fabric2_db:FoldFun(Db, ViewCb, Acc, Opts),
     VAcc1.
 
 
 send_all_docs_keys(Db, #mrargs{} = Args, UserCtx, VAcc0) ->
     Keys = apply_args_to_keylist(Args, Args#mrargs.keys),
-    %% namespace can be _set_ to `undefined`, so we
-    %% want simulate enum here
-    NS = case couch_util:get_value(namespace, Args#mrargs.extra) of
-        <<"_all_docs">> -> <<"_all_docs">>;
-        <<"_design">> -> <<"_design">>;
-        <<"_local">> -> <<"_local">>;
-        _ -> <<"_all_docs">>
-    end,
+    NS = couch_util:get_value(namespace, Args#mrargs.extra),
     TotalRows = fabric2_db:get_doc_count(Db, NS),
     Meta = case Args#mrargs.update_seq of
         true ->
@@ -889,7 +888,7 @@ send_all_docs_keys(Db, #mrargs{} = Args, UserCtx, VAcc0) ->
         _ -> Args#mrargs.doc_options
     end ++ [{user_ctx, UserCtx}],
     IncludeDocs = Args#mrargs.include_docs,
-    VAcc2 = lists:foldl(fun(DocId, Acc) ->
+    lists:foldl(fun(DocId, Acc) ->
         OpenOpts = [deleted | DocOpts],
         Row0 = case fabric2_db:open_doc(Db, DocId, OpenOpts) of
             {not_found, missing} ->
@@ -931,6 +930,7 @@ send_all_docs_keys(Db, #mrargs{} = Args, UserCtx, VAcc0) ->
 
 
 all_docs_view_opts(Args, UserCtx) ->
+    NS = couch_util:get_value(namespace, Args#mrargs.extra),
     StartKey = case Args#mrargs.start_key of
         undefined -> Args#mrargs.start_key_docid;
         SKey -> SKey
@@ -939,18 +939,23 @@ all_docs_view_opts(Args, UserCtx) ->
         undefined -> Args#mrargs.end_key_docid;
         EKey -> EKey
     end,
+    StartKeyOpts = case StartKey of
+        <<_/binary>> -> [{start_key, StartKey}];
+        undefined -> []
+    end,
     EndKeyOpts = case {EndKey, Args#mrargs.inclusive_end} of
         {<<_/binary>>, false} -> [{end_key_gt, EndKey}];
-        {_, _} -> [{end_key, EndKey}]
+        {<<_/binary>>, true} -> [{end_key, EndKey}];
+        {undefined, _} -> []
     end,
     [
         {user_ctx, UserCtx},
         {dir, Args#mrargs.direction},
-        {start_key, StartKey},
         {limit, Args#mrargs.limit},
         {skip, Args#mrargs.skip},
-        {update_seq, Args#mrargs.update_seq}
-    ] ++ EndKeyOpts.
+        {update_seq, Args#mrargs.update_seq},
+        {namespace, NS}
+    ] ++ StartKeyOpts ++ EndKeyOpts.
 
 
 apply_args_to_keylist(Args, Keys0) ->
@@ -2015,8 +2020,7 @@ monitor_attachments(Att) ->
 demonitor_refs(Refs) when is_list(Refs) ->
     [demonitor(Ref) || Ref <- Refs].
 
-set_namespace(<<"_all_docs">>, Args) ->
-    set_namespace(undefined, Args);
+
 set_namespace(<<"_local_docs">>, Args) ->
     set_namespace(<<"_local">>, Args);
 set_namespace(<<"_design_docs">>, Args) ->
diff --git a/src/fabric/src/fabric2_db.erl b/src/fabric/src/fabric2_db.erl
index 2afb780fa..9deb1930a 100644
--- a/src/fabric/src/fabric2_db.erl
+++ b/src/fabric/src/fabric2_db.erl
@@ -102,8 +102,8 @@
 
     fold_docs/3,
     fold_docs/4,
-    %% fold_local_docs/4,
-    %% fold_design_docs/4,
+    fold_design_docs/4,
+    fold_local_docs/4,
     fold_changes/4,
     fold_changes/5,
     %% count_changes_since/2,
@@ -136,6 +136,8 @@
     "(\\.[0-9]{10,})?$" % but allow an optional shard timestamp at the end
 ).
 
+-define(FIRST_DDOC_KEY, <<"_design/">>).
+-define(LAST_DDOC_KEY, <<"_design0">>).
 
 -define(RETURN(Term), throw({?MODULE, Term})).
 
@@ -314,6 +316,9 @@ 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">>);
 
@@ -729,8 +734,8 @@ fold_docs(Db, UserFun, UserAcc0, Options) ->
             } = TxDb,
 
             Prefix = erlfdb_tuple:pack({?DB_ALL_DOCS}, DbPrefix),
-            DocCount = get_doc_count(TxDb),
-
+            NS = couch_util:get_value(namespace, Options),
+            DocCount = get_doc_count(TxDb, NS),
             Meta = case lists:keyfind(update_seq, 1, Options) of
                 {_, true} ->
                     UpdateSeq = fabric2_db:get_update_seq(TxDb),
@@ -758,6 +763,44 @@ fold_docs(Db, UserFun, UserAcc0, Options) ->
     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, Options) ->
+     fabric2_fdb:transactional(Db, fun(TxDb) ->
+        try
+            #{
+                db_prefix := DbPrefix
+            } = TxDb,
+
+            Prefix = erlfdb_tuple:pack({?DB_LOCAL_DOCS}, DbPrefix),
+            DocCount = get_doc_count(TxDb, <<"doc_local_count">>),
+            Meta = [{total, DocCount}, {offset, null}],
+
+            UserAcc1 = maybe_stop(UserFun({meta, Meta}, UserAcc0)),
+
+            UserAcc2 = fabric2_fdb:fold_range(TxDb, Prefix, fun({K, V}, Acc) ->
+                {DocId} = erlfdb_tuple:unpack(K, Prefix),
+                LDoc = fabric2_fdb:fdb_to_local_doc(DocId, V),
+                #doc{revs = {Pos, [Rev]}} = LDoc,
+                maybe_stop(UserFun({row, [
+                    {id, DocId},
+                    {key, DocId},
+                    {value, {[{rev, couch_doc:rev_to_str({Pos, 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, []).
 
@@ -1615,3 +1658,43 @@ 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.
diff --git a/src/fabric/src/fabric2_fdb.erl b/src/fabric/src/fabric2_fdb.erl
index df3709673..772e5290b 100644
--- a/src/fabric/src/fabric2_fdb.erl
+++ b/src/fabric/src/fabric2_fdb.erl
@@ -43,6 +43,7 @@
     get_doc_body_future/3,
     get_doc_body_wait/4,
     get_local_doc/2,
+    fdb_to_local_doc/2,
 
     write_doc/6,
     write_local_doc/2,
@@ -487,15 +488,27 @@ get_doc_body_wait(#{} = Db0, DocId, RevInfo, Future) ->
     fdb_to_doc(Db, DocId, RevPos, [Rev | RevPath], BodyRows).
 
 
-get_local_doc(#{} = Db0, <<?LOCAL_DOC_PREFIX, _/binary>> = DocId) ->
+get_local_doc(#{} = Db, <<?LOCAL_DOC_PREFIX, _/binary>> = DocId) ->
     #{
         tx := Tx,
         db_prefix := DbPrefix
-    } = Db = ensure_current(Db0),
+    } = ensure_current(Db),
 
     Key = erlfdb_tuple:pack({?DB_LOCAL_DOCS, DocId}, DbPrefix),
     Val = erlfdb:wait(erlfdb:get(Tx, Key)),
-    fdb_to_local_doc(Db, DocId, Val).
+    fdb_to_local_doc(DocId, Val).
+
+
+fdb_to_local_doc(DocId, Bin) when is_binary(Bin) ->
+    {Rev, Body} = binary_to_term(Bin, [safe]),
+    #doc{
+        id = DocId,
+        revs = {0, [Rev]},
+        deleted = false,
+        body = Body
+    };
+fdb_to_local_doc(_DocId, not_found) ->
+    {not_found, missing}.
 
 
 write_doc(#{} = Db0, Doc, NewWinner0, OldWinner, ToUpdate, ToRemove) ->
@@ -993,18 +1006,6 @@ local_doc_to_fdb(Db, #doc{} = Doc) ->
     {Key, term_to_binary(Val, [{minor_version, 1}])}.
 
 
-fdb_to_local_doc(_Db, DocId, Bin) when is_binary(Bin) ->
-    {Rev, Body} = binary_to_term(Bin, [safe]),
-    #doc{
-        id = DocId,
-        revs = {0, [Rev]},
-        deleted = false,
-        body = Body
-    };
-fdb_to_local_doc(_Db, _DocId, not_found) ->
-    {not_found, missing}.
-
-
 chunkify_binary(Data) ->
     case Data of
         <<>> ->
diff --git a/test/elixir/test/basics_test.exs b/test/elixir/test/basics_test.exs
index 70dd6e84c..736cdbbec 100644
--- a/test/elixir/test/basics_test.exs
+++ b/test/elixir/test/basics_test.exs
@@ -356,4 +356,123 @@ defmodule BasicsTest do
     assert Map.has_key?(val, "rev")
   end
 
+  @tag :with_db
+  test "_design_docs works", context do
+    db_name = context[:db_name]
+    body = %{:a => 1}
+
+    resp = Couch.get("/#{db_name}/_design_docs")
+    assert resp.status_code == 200
+    assert resp.body == %{"offset" => :null, "rows" => [], "total_rows" => 0}
+
+    assert Couch.put("/#{db_name}/doc1", body: body).body["ok"]
+
+    # Make sure regular documents didn't get picked up
+    resp = Couch.get("/#{db_name}/_design_docs")
+    assert resp.status_code == 200
+    assert resp.body == %{"offset" => :null, "rows" => [], "total_rows" => 0}
+
+    # Add _design/doc1
+    assert Couch.put("/#{db_name}/_design/doc1", body: body).body["ok"]
+    resp = Couch.get("/#{db_name}/_design_docs")
+    assert resp.status_code == 200
+    assert resp.body["total_rows"] == 1
+    [row] = resp.body["rows"]
+
+    assert row["id"] == "_design/doc1"
+    assert row["key"] == "_design/doc1"
+
+    val = row["value"]
+    assert Map.has_key?(val, "rev")
+
+    # Add _design/doc5
+    assert Couch.put("/#{db_name}/_design/doc5", body: body).body["ok"]
+    resp = Couch.get("/#{db_name}/_design_docs")
+    assert resp.status_code == 200
+    [row1, row2] = resp.body["rows"]
+    assert row1["id"] == "_design/doc1"
+    assert row2["id"] == "_design/doc5"
+
+    # descending=true
+    resp = Couch.get("/#{db_name}/_design_docs?descending=true")
+    assert resp.status_code == 200
+    [row1, row2] = resp.body["rows"]
+    assert row1["id"] == "_design/doc5"
+    assert row2["id"] == "_design/doc1"
+
+    # start_key=doc2
+    resp = Couch.get("/#{db_name}/_design_docs?start_key=\"_design/doc2\"")
+    assert resp.status_code == 200
+    [row] = resp.body["rows"]
+    assert row["id"] == "_design/doc5"
+
+    # end_key=doc2
+    resp = Couch.get("/#{db_name}/_design_docs?end_key=\"_design/doc2\"")
+    assert resp.status_code == 200
+    [row] = resp.body["rows"]
+    assert row["id"] == "_design/doc1"
+
+    # inclusive_end=false
+    qstr = "start_key=\"_design/doc2\"&end_key=\"_design/doc5\"&inclusive_end=false"
+    resp = Couch.get("/#{db_name}/_design_docs?" <> qstr)
+    assert resp.status_code == 200
+    assert resp.body == %{"offset" => :null, "rows" => [], "total_rows" => 2}
+  end
+
+  @tag :with_db
+  test "_local_docs works", context do
+    db_name = context[:db_name]
+    body = %{:a => 1}
+
+    resp = Couch.get("/#{db_name}/_local_docs")
+    assert resp.status_code == 200
+    assert resp.body == %{"offset" => :null, "rows" => [], "total_rows" => 0}
+
+    # Add _local/doc1
+    assert Couch.put("/#{db_name}/_local/doc1", body: body).body["ok"]
+    resp = Couch.get("/#{db_name}/_local_docs")
+    assert resp.status_code == 200
+    assert resp.body["total_rows"] == 1
+    [row] = resp.body["rows"]
+
+    assert row["id"] == "_local/doc1"
+    assert row["key"] == "_local/doc1"
+
+    val = row["value"]
+    assert Map.has_key?(val, "rev")
+
+    # Add _local/doc5
+    assert Couch.put("/#{db_name}/_local/doc5", body: body).body["ok"]
+    resp = Couch.get("/#{db_name}/_local_docs")
+    assert resp.status_code == 200
+    [row1, row2] = resp.body["rows"]
+    assert row1["id"] == "_local/doc1"
+    assert row2["id"] == "_local/doc5"
+
+    # descending=true
+    resp = Couch.get("/#{db_name}/_local_docs?descending=true")
+    assert resp.status_code == 200
+    [row1, row2] = resp.body["rows"]
+    assert row1["id"] == "_local/doc5"
+    assert row2["id"] == "_local/doc1"
+
+    # start_key=doc2
+    resp = Couch.get("/#{db_name}/_local_docs?start_key=\"_local/doc2\"")
+    assert resp.status_code == 200
+    [row] = resp.body["rows"]
+    assert row["id"] == "_local/doc5"
+
+    # end_key=doc2
+    resp = Couch.get("/#{db_name}/_local_docs?end_key=\"_local/doc2\"")
+    assert resp.status_code == 200
+    [row] = resp.body["rows"]
+    assert row["id"] == "_local/doc1"
+
+    # inclusive_end=false
+    qstr = "start_key=\"_local/doc2\"&end_key=\"_local/doc5\"&inclusive_end=false"
+    resp = Couch.get("/#{db_name}/_local_docs?" <> qstr)
+    assert resp.status_code == 200
+    assert resp.body == %{"offset" => :null, "rows" => [], "total_rows" => 2}
+  end
+
 end