Add tests for fold_changes

4 files changed, 180 insertions, 29 deletions

diff --git a/src/fabric/src/fabric2_db.erl b/src/fabric/src/fabric2_db.erl
index 9971bbc8f..c276ae20f 100644
--- a/src/fabric/src/fabric2_db.erl
+++ b/src/fabric/src/fabric2_db.erl
@@ -307,11 +307,7 @@ get_security(#{security_doc := SecurityDoc}) ->
 
 get_update_seq(#{} = Db) ->
     fabric2_fdb:transactional(Db, fun(TxDb) ->
-        Opts = [{limit, 1}, {reverse, true}],
-        case fabric2_fdb:get_changes(TxDb, Opts) of
-            [] -> fabric2_util:to_hex(fabric2_util:seq_zero());
-            [{Seq, _}] -> Seq
-        end
+        fabric2_fdb:get_last_change(TxDb)
     end).
 
 
diff --git a/src/fabric/src/fabric2_fdb.erl b/src/fabric/src/fabric2_fdb.erl
index 19b19bb4b..769a5a838 100644
--- a/src/fabric/src/fabric2_fdb.erl
+++ b/src/fabric/src/fabric2_fdb.erl
@@ -45,8 +45,7 @@
 
     fold_docs/4,
     fold_changes/5,
-
-    get_changes/2,
+    get_last_change/1,
 
     debug_cluster/0,
     debug_cluster/2
@@ -251,7 +250,7 @@ get_info(#{} = Db) ->
             fabric2_util:seq_zero();
         [{SeqKey, _}] ->
             {?DB_CHANGES, SeqVS} = erlfdb_tuple:unpack(SeqKey, DbPrefix),
-            <<51:8, SeqBin:13/binary>> = erlfdb_tuple:pack({SeqVS}),
+            <<51:8, SeqBin:12/binary>> = erlfdb_tuple:pack({SeqVS}),
             SeqBin
     end,
     CProp = {update_seq, fabric2_util:to_hex(RawSeq)},
@@ -589,8 +588,7 @@ fold_changes(#{} = Db, SinceSeq0, UserFun, UserAcc0, Options) ->
         db_prefix := DbPrefix
     } = ensure_current(Db),
 
-    SinceSeq1 = fabric2_util:from_hex(SinceSeq0),
-    SinceSeq2 = <<51:8, SinceSeq1/binary>>,
+    SinceSeq1 = get_since_seq(SinceSeq0),
 
     Reverse = case fabric2_util:get_value(dir, Options, fwd) of
         fwd -> false;
@@ -598,17 +596,22 @@ fold_changes(#{} = Db, SinceSeq0, UserFun, UserAcc0, Options) ->
     end,
 
     {Start0, End0} = case Reverse of
-        false ->
-            {{?DB_CHANGES, SinceSeq2}, {?DB_CHANGES, <<16#FF>>}};
-        true ->
-            {{?DB_CHANGES}, {?DB_CHANGES, SinceSeq2}}
+        false -> {SinceSeq1, fabric2_util:seq_max_vs()};
+        true -> {fabric2_util:seq_zero_vs(), SinceSeq1}
     end,
 
-    Start = erlfdb_tuple:pack(Start0, DbPrefix),
-    End = erlfdb_tuple:pack(End0, DbPrefix),
+    Start1 = erlfdb_tuple:pack({?DB_CHANGES, Start0}, DbPrefix),
+    End1 = erlfdb_tuple:pack({?DB_CHANGES, End0}, DbPrefix),
+
+    {Start, End} = case Reverse of
+        false -> {erlfdb_key:first_greater_than(Start1), End1};
+        true -> {Start1, erlfdb_key:first_greater_than(End1)}
+    end,
 
     try
-        put('$last_changes_seq', SinceSeq0),
+        % We have to track this to return last_seq
+        <<51:8, FirstSeq:12/binary>> = erlfdb_tuple:pack({SinceSeq1}),
+        put('$last_changes_seq', fabric2_util:to_hex(FirstSeq)),
 
         UserAcc1 = maybe_stop(UserFun(start, UserAcc0)),
 
@@ -618,7 +621,7 @@ fold_changes(#{} = Db, SinceSeq0, UserFun, UserAcc0, Options) ->
 
             % This comes back as a versionstamp so we have
             % to pack it to get a binary.
-            <<51:8, SeqBin:13/binary>> = erlfdb_tuple:pack({UpdateSeq}),
+            <<51:8, SeqBin:12/binary>> = erlfdb_tuple:pack({UpdateSeq}),
             SeqHex = fabric2_util:to_hex(SeqBin),
             put('$last_changes_seq', SeqHex),
 
@@ -641,19 +644,22 @@ fold_changes(#{} = Db, SinceSeq0, UserFun, UserAcc0, Options) ->
     end.
 
 
-get_changes(#{} = Db, Options) ->
+get_last_change(#{} = Db) ->
     #{
         tx := Tx,
         db_prefix := DbPrefix
     } = ensure_current(Db),
 
-    {CStart, CEnd} = erlfdb_tuple:range({?DB_CHANGES}, DbPrefix),
-    Future = erlfdb:get_range(Tx, CStart, CEnd, Options),
-    lists:map(fun({Key, Val}) ->
-        {?DB_CHANGES, SeqVS} = erlfdb_tuple:unpack(Key, DbPrefix),
-        <<51:8, SeqBin:13/binary>> = erlfdb_tuple:pack({SeqVS}),
-        {fabric2_util:to_hex(SeqBin), Val}
-    end, erlfdb:wait(Future)).
+    {Start, End} = erlfdb_tuple:range({?DB_CHANGES}, DbPrefix),
+    Options = [{limit, 1}, {reverse, true}],
+    case erlfdb:get_range(Tx, Start, End, Options) of
+        [] ->
+            fabric2_util:to_hex(fabric2_util:seq_zero());
+        [{K, _V}] ->
+            {?DB_CHANGES, SeqVS} = erlfdb_tuple:unpack(K, DbPrefix),
+            <<51:8, SeqBin:12/binary>> = erlfdb_tuple:pack({SeqVS}),
+            fabric2_util:to_hex(SeqBin)
+    end.
 
 
 maybe_stop({ok, Acc}) ->
@@ -875,6 +881,20 @@ get_dir_and_bounds(DbPrefix, Options) ->
     {Reverse, StartKey4, EndKey4}.
 
 
+get_since_seq(Seq) when Seq == 0; Seq == <<"0">> ->
+    fabric2_util:seq_zero_vs();
+
+get_since_seq(Seq) when Seq == now; Seq == <<"now">> ->
+    fabric2_util:seq_max_vs();
+
+get_since_seq(Seq) when is_binary(Seq), size(Seq) == 24 ->
+    Seq1 = fabric2_util:from_hex(Seq),
+    Seq2 = <<51:8, Seq1/binary>>,
+    {SeqVS} = erlfdb_tuple:unpack(Seq2),
+    SeqVS.
+
+
+
 get_db_handle() ->
     case get(?PDICT_DB_KEY) of
         undefined ->
@@ -960,9 +980,8 @@ get_transaction_id(Tx) ->
 new_versionstamp(_Tx) ->
     % Eventually we'll have a erlfdb:get_next_tx_id(Tx)
     % that will return a monotonically incrementing
-    % integer. For now we just hardcode 0 since we're
-    % not doing multiple docs per batch.
-    % Various utility macros
+    % integer from 0 to 65535. For now we just hardcode
+    % 0 since we're not doing multiple docs per batch.
     TxId = 0,
     {versionstamp, 16#FFFFFFFFFFFFFFFF, 16#FFFF, TxId}.
 
diff --git a/src/fabric/src/fabric2_util.erl b/src/fabric/src/fabric2_util.erl
index d95fd160f..1696f06a6 100644
--- a/src/fabric/src/fabric2_util.erl
+++ b/src/fabric/src/fabric2_util.erl
@@ -16,7 +16,10 @@
 -export([
     revinfo_to_path/1,
     sort_revinfos/1,
+
     seq_zero/0,
+    seq_zero_vs/0,
+    seq_max_vs/0,
 
     user_ctx_to_json/1,
 
@@ -67,6 +70,14 @@ seq_zero() ->
     <<0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0>>.
 
 
+seq_zero_vs() ->
+    {versionstamp, 0, 0, 0}.
+
+
+seq_max_vs() ->
+    {versionstamp, 18446744073709551615, 65535, 65535}.
+
+
 user_ctx_to_json(Db) ->
     UserCtx = fabric2_db:get_user_ctx(Db),
     {[
diff --git a/src/fabric/test/fabric2_changes_fold_tests.erl b/src/fabric/test/fabric2_changes_fold_tests.erl
new file mode 100644
index 000000000..20f57a4f8
--- /dev/null
+++ b/src/fabric/test/fabric2_changes_fold_tests.erl
@@ -0,0 +1,125 @@
+% 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_changes_fold_tests).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("eunit/include/eunit.hrl").
+
+
+-define(DOC_COUNT, 25).
+
+
+changes_fold_test_() ->
+    {
+        "Test changes fold operations",
+        {
+            setup,
+            fun setup/0,
+            fun cleanup/1,
+            {with, [
+                fun fold_changes_basic/1,
+                fun fold_changes_since_now/1,
+                fun fold_changes_since_seq/1,
+                fun fold_changes_basic_rev/1,
+                fun fold_changes_since_now_rev/1,
+                fun fold_changes_since_seq_rev/1
+            ]}
+        }
+    }.
+
+
+setup() ->
+    Ctx = test_util:start_couch([fabric]),
+    {ok, Db} = fabric2_db:create(?tempdb(), [{user_ctx, ?ADMIN_USER}]),
+    Rows = lists:map(fun(Val) ->
+        DocId = fabric2_util:uuid(),
+        Doc = #doc{
+            id = DocId,
+            body = {[{<<"value">>, Val}]}
+        },
+        {ok, Rev} = fabric2_db:update_doc(Db, Doc, []),
+        UpdateSeq = fabric2_db:get_update_seq(Db),
+        #{
+            id => DocId,
+            seq => UpdateSeq,
+            deleted => false,
+            rev => couch_doc:rev_to_str(Rev)
+        }
+    end, lists:seq(1, ?DOC_COUNT)),
+    {Db, Rows, Ctx}.
+
+
+cleanup({Db, _DocIdRevs, Ctx}) ->
+    ok = fabric2_db:delete(fabric2_db:name(Db), []),
+    test_util:stop_couch(Ctx).
+
+
+fold_changes_basic({Db, DocRows, _}) ->
+    {ok, Rows} = fabric2_db:fold_changes(Db, 0, fun fold_fun/2, []),
+    ?assertEqual(lists:reverse(DocRows), Rows).
+
+
+fold_changes_since_now({Db, _, _}) ->
+    {ok, Rows} = fabric2_db:fold_changes(Db, now, fun fold_fun/2, []),
+    ?assertEqual([], Rows).
+
+
+fold_changes_since_seq({_, [], _}) ->
+    ok;
+
+fold_changes_since_seq({Db, [Row | RestRows], _}) ->
+    #{seq := Since} = Row,
+    {ok, Rows} = fabric2_db:fold_changes(Db, Since, fun fold_fun/2, []),
+    ?assertEqual(lists:reverse(RestRows), Rows),
+    fold_changes_since_seq({Db, RestRows, nil}).
+
+
+fold_changes_basic_rev({Db, _, _}) ->
+    Opts = [{dir, rev}],
+    {ok, Rows} = fabric2_db:fold_changes(Db, 0, fun fold_fun/2, [], Opts),
+    ?assertEqual([], Rows).
+
+
+fold_changes_since_now_rev({Db, DocRows, _}) ->
+    Opts = [{dir, rev}],
+    {ok, Rows} = fabric2_db:fold_changes(Db, now, fun fold_fun/2, [], Opts),
+    ?assertEqual(DocRows, Rows).
+
+
+fold_changes_since_seq_rev({_, [], _}) ->
+    ok;
+
+fold_changes_since_seq_rev({Db, DocRows, _}) ->
+    #{seq := Since} = lists:last(DocRows),
+    Opts = [{dir, rev}],
+    {ok, Rows} = fabric2_db:fold_changes(Db, Since, fun fold_fun/2, [], Opts),
+    ?assertEqual(DocRows, Rows),
+    RestRows = lists:sublist(DocRows, length(DocRows) - 1),
+    fold_changes_since_seq_rev({Db, RestRows, nil}).
+
+
+fold_fun(start, Acc) ->
+    {ok, Acc};
+fold_fun({change, {Props}}, Acc) ->
+    [{[{rev, Rev}]}] = fabric2_util:get_value(changes, Props),
+    Row = #{
+        id => fabric2_util:get_value(id, Props),
+        seq => fabric2_util:get_value(seq, Props),
+        deleted => fabric2_util:get_value(deleted, Props, false),
+        rev => Rev
+    },
+    {ok, [Row | Acc]};
+fold_fun({stop, _LastSeq, null}, Acc) ->
+    {ok, Acc}.