% 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(cpse_test_fold_changes).
-compile(export_all).
-compile(nowarn_export_all).


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


-define(NUM_DOCS, 25).


setup_each() ->
    {ok, Db} = cpse_util:create_db(),
    Db.


teardown_each(Db) ->
    ok = couch_server:delete(couch_db:name(Db), []).


cpse_empty_changes(Db) ->
    ?assertEqual(0, couch_db_engine:count_changes_since(Db, 0)),
    ?assertEqual({ok, []},
            couch_db_engine:fold_changes(Db, 0, fun fold_fun/2, [], [])).


cpse_single_change(Db1) ->
    Actions = [{create, {<<"a">>, {[]}}}],
    {ok, Db2} = cpse_util:apply_actions(Db1, Actions),

    ?assertEqual(1, couch_db_engine:count_changes_since(Db2, 0)),
    ?assertEqual({ok, [{<<"a">>, 1}]},
            couch_db_engine:fold_changes(Db2, 0, fun fold_fun/2, [], [])).


cpse_two_changes(Db1) ->
    Actions = [
        {create, {<<"a">>, {[]}}},
        {create, {<<"b">>, {[]}}}
    ],
    {ok, Db2} = cpse_util:apply_actions(Db1, Actions),

    ?assertEqual(2, couch_db_engine:count_changes_since(Db2, 0)),
    {ok, Changes} =
            couch_db_engine:fold_changes(Db2, 0, fun fold_fun/2, [], []),
    ?assertEqual([{<<"a">>, 1}, {<<"b">>, 2}], lists:reverse(Changes)).


cpse_two_changes_batch(Db1) ->
    Actions = [
        {batch, [
            {create, {<<"a">>, {[]}}},
            {create, {<<"b">>, {[]}}}
        ]}
    ],
    {ok, Db2} = cpse_util:apply_actions(Db1, Actions),

    ?assertEqual(2, couch_db_engine:count_changes_since(Db2, 0)),
    {ok, Changes} =
            couch_db_engine:fold_changes(Db2, 0, fun fold_fun/2, [], []),
    ?assertEqual([{<<"a">>, 1}, {<<"b">>, 2}], lists:reverse(Changes)).


cpse_two_changes_batch_sorted(Db1) ->
    Actions = [
        {batch, [
            {create, {<<"b">>, {[]}}},
            {create, {<<"a">>, {[]}}}
        ]}
    ],
    {ok, Db2} = cpse_util:apply_actions(Db1, Actions),

    ?assertEqual(2, couch_db_engine:count_changes_since(Db2, 0)),
    {ok, Changes} =
            couch_db_engine:fold_changes(Db2, 0, fun fold_fun/2, [], []),
    ?assertEqual([{<<"a">>, 1}, {<<"b">>, 2}], lists:reverse(Changes)).


cpse_update_one(Db1) ->
    Actions = [
        {create, {<<"a">>, {[]}}},
        {update, {<<"a">>, {[]}}}
    ],
    {ok, Db2} = cpse_util:apply_actions(Db1, Actions),

    ?assertEqual(1, couch_db_engine:count_changes_since(Db2, 0)),
    ?assertEqual({ok, [{<<"a">>, 2}]},
            couch_db_engine:fold_changes(Db2, 0, fun fold_fun/2, [], [])).


cpse_update_first_of_two(Db1) ->
    Actions = [
        {create, {<<"a">>, {[]}}},
        {create, {<<"b">>, {[]}}},
        {update, {<<"a">>, {[]}}}
    ],
    {ok, Db2} = cpse_util:apply_actions(Db1, Actions),

    ?assertEqual(2, couch_db_engine:count_changes_since(Db2, 0)),
    {ok, Changes} =
            couch_db_engine:fold_changes(Db2, 0, fun fold_fun/2, [], []),
    ?assertEqual([{<<"b">>, 2}, {<<"a">>, 3}], lists:reverse(Changes)).


cpse_update_second_of_two(Db1) ->
    Actions = [
        {create, {<<"a">>, {[]}}},
        {create, {<<"b">>, {[]}}},
        {update, {<<"b">>, {[]}}}
    ],
    {ok, Db2} = cpse_util:apply_actions(Db1, Actions),

    ?assertEqual(2, couch_db_engine:count_changes_since(Db2, 0)),
    {ok, Changes} =
            couch_db_engine:fold_changes(Db2, 0, fun fold_fun/2, [], []),
    ?assertEqual([{<<"a">>, 1}, {<<"b">>, 3}], lists:reverse(Changes)).


cpse_check_mutation_ordering(Db1) ->
    Actions = shuffle(lists:map(fun(Seq) ->
        {create, {docid(Seq), {[]}}}
    end, lists:seq(1, ?NUM_DOCS))),

    DocIdOrder = [DocId || {_, {DocId, _}} <- Actions],
    DocSeqs = lists:zip(DocIdOrder, lists:seq(1, ?NUM_DOCS)),

    {ok, Db2} = cpse_util:apply_actions(Db1, Actions),

    % First lets see that we can get the correct
    % suffix/prefix starting at every update sequence
    lists:foreach(fun(Seq) ->
        {ok, Suffix} =
                couch_db_engine:fold_changes(Db2, Seq, fun fold_fun/2, [], []),
        ?assertEqual(lists:nthtail(Seq, DocSeqs), lists:reverse(Suffix)),

        {ok, Prefix} = couch_db_engine:fold_changes(
                Db2, Seq, fun fold_fun/2, [], [{dir, rev}]),
        ?assertEqual(lists:sublist(DocSeqs, Seq + 1), Prefix)
    end, lists:seq(0, ?NUM_DOCS)),

    ok = do_mutation_ordering(Db2, ?NUM_DOCS + 1, DocSeqs, []).


do_mutation_ordering(Db, _Seq, [], FinalDocSeqs) ->
    {ok, RevOrder} = couch_db_engine:fold_changes(Db, 0, fun fold_fun/2, [], []),
    ?assertEqual(FinalDocSeqs, lists:reverse(RevOrder)),
    ok;

do_mutation_ordering(Db, Seq, [{DocId, _OldSeq} | Rest], DocSeqAcc) ->
    Actions = [{update, {DocId, {[]}}}],
    {ok, NewDb} = cpse_util:apply_actions(Db, Actions),
    NewAcc = DocSeqAcc ++ [{DocId, Seq}],
    Expected = Rest ++ NewAcc,
    {ok, RevOrder} =
            couch_db_engine:fold_changes(NewDb, 0, fun fold_fun/2, [], []),
    ?assertEqual(Expected, lists:reverse(RevOrder)),
    do_mutation_ordering(NewDb, Seq + 1, Rest, NewAcc).


shuffle(List) ->
    Paired = [{couch_rand:uniform(), I} || I <- List],
    Sorted = lists:sort(Paired),
    [I || {_, I} <- Sorted].


fold_fun(#full_doc_info{id=Id, update_seq=Seq}, Acc) ->
    {ok, [{Id, Seq} | Acc]}.


docid(I) ->
    Str = io_lib:format("~4..0b", [I]),
    iolist_to_binary(Str).