Update to use new purge API

COUCHDB-3326

8 files changed, 1165 insertions, 17 deletions

diff --git a/src/clouseau_rpc.erl b/src/clouseau_rpc.erl
index d9100764f..345b499e6 100644
--- a/src/clouseau_rpc.erl
+++ b/src/clouseau_rpc.erl
@@ -22,12 +22,15 @@
 -export([group1/7, group2/8, group2/2]).
 -export([delete/2, update/3, cleanup/1, cleanup/2, rename/1]).
 -export([analyze/2, version/0, disk_size/1]).
+-export([set_purge_seq/2, get_purge_seq/1, get_root_dir/0]).
 
 open_index(Peer, Path, Analyzer) ->
     rpc({main, clouseau()}, {open, Peer, Path, Analyzer}).
 
 disk_size(Path) ->
     rpc({main, clouseau()}, {disk_size, Path}).
+get_root_dir() ->
+    rpc({main, clouseau()}, {get_root_dir}).
 
 await(Ref, MinSeq) ->
     rpc(Ref, {await, MinSeq}).
@@ -41,6 +44,12 @@ info(Ref) ->
 get_update_seq(Ref) ->
     rpc(Ref, get_update_seq).
 
+set_purge_seq(Ref, Seq) ->
+    rpc(Ref, {set_purge_seq, Seq}).
+
+get_purge_seq(Ref) ->
+    rpc(Ref, get_purge_seq).
+
 %% @deprecated
 search(Ref, Query, Limit, Refresh, Bookmark, Sort) ->
     rpc(Ref, {search, Query, Limit, Refresh, Bookmark, Sort}).
diff --git a/src/dreyfus_epi.erl b/src/dreyfus_epi.erl
index 3cda975ea..cb07f8a34 100644
--- a/src/dreyfus_epi.erl
+++ b/src/dreyfus_epi.erl
@@ -19,7 +19,8 @@ app() ->
 
 providers() ->
     [
-         {chttpd_handlers, dreyfus_httpd_handlers}
+        {couch_db, dreyfus_plugin_couch_db},
+        {chttpd_handlers, dreyfus_httpd_handlers}
     ].
 
 
diff --git a/src/dreyfus_fabric_cleanup.erl b/src/dreyfus_fabric_cleanup.erl
index 501fdadd7..b5e030db0 100644
--- a/src/dreyfus_fabric_cleanup.erl
+++ b/src/dreyfus_fabric_cleanup.erl
@@ -25,6 +25,7 @@ go(DbName) ->
     {ok, DesignDocs} = fabric:design_docs(DbName),
     ActiveSigs = lists:usort(lists:flatmap(fun active_sigs/1,
         [couch_doc:from_json_obj(DD) || DD <- DesignDocs])),
+    cleanup_local_purge_doc(DbName, ActiveSigs),
     clouseau_rpc:cleanup(DbName, ActiveSigs),
     ok.
 
@@ -35,3 +36,39 @@ active_sigs(#doc{body={Fields}}=Doc) ->
          {ok, Index} = dreyfus_index:design_doc_to_index(Doc, IndexName),
          Index#index.sig
      end || IndexName <- IndexNames].
+
+cleanup_local_purge_doc(DbName, ActiveSigs) ->
+    {ok, BaseDir} = clouseau_rpc:get_root_dir(),
+    DbNamePattern = <<DbName/binary, ".*">>,
+    Pattern0 = filename:join([BaseDir, "shards", "*", DbNamePattern, "*"]),
+    Pattern = binary_to_list(iolist_to_binary(Pattern0)),
+    DirListStrs = filelib:wildcard(Pattern),
+    DirList = [iolist_to_binary(DL) || DL <- DirListStrs],
+    LocalShards = mem3:local_shards(DbName),
+    ActiveDirs = lists:foldl(fun(LS, AccOuter) ->
+        lists:foldl(fun(Sig, AccInner) ->
+            DirName = filename:join([BaseDir, LS#shard.name, Sig]),
+            [DirName | AccInner]
+        end, AccOuter, ActiveSigs)
+    end, [], LocalShards),
+
+    DeadDirs = DirList -- ActiveDirs,
+    lists:foldl(fun(IdxDir) ->
+        Sig = dreyfus_util:get_signature_from_idxdir(IdxDir),
+        case Sig of undefined -> ok; _ ->
+            DocId = dreyfus_util:get_local_purge_doc_id(Sig),
+            LocalShards = mem3:local_shards(DbName),
+            lists:foldl(fun(LS, _AccOuter) ->
+                ShardDbName = LS#shard.name,
+                {ok, ShardDb} = couch_db:open_int(ShardDbName, []),
+                case couch_db:open_doc(ShardDb, DocId, []) of
+                    {ok, LocalPurgeDoc} ->
+                        couch_db:update_doc(ShardDb,
+                            LocalPurgeDoc#doc{deleted=true}, [?ADMIN_CTX]);
+                    {not_found, _} ->
+                        ok
+                end,
+                couch_db:close(ShardDb)
+            end, [], LocalShards)
+        end
+    end, [], DeadDirs).
diff --git a/src/dreyfus_index.erl b/src/dreyfus_index.erl
index 93b25acf7..c6d4d856a 100644
--- a/src/dreyfus_index.erl
+++ b/src/dreyfus_index.erl
@@ -109,6 +109,7 @@ init({DbName, Index}) ->
             case couch_db:open_int(DbName, []) of
                 {ok, Db} ->
                     try couch_db:monitor(Db) after couch_db:close(Db) end,
+                    dreyfus_util:maybe_create_local_purge_doc(Db, Pid, Index),
                     proc_lib:init_ack({ok, self()}),
                     gen_server:enter_loop(?MODULE, [], State);
                 Error ->
diff --git a/src/dreyfus_index_updater.erl b/src/dreyfus_index_updater.erl
index dc2067be9..0c3b3b7eb 100644
--- a/src/dreyfus_index_updater.erl
+++ b/src/dreyfus_index_updater.erl
@@ -31,8 +31,9 @@ update(IndexPid, Index) ->
     erlang:put(io_priority, {view_update, DbName, IndexName}),
     {ok, Db} = couch_db:open_int(DbName, []),
     try
-        %% compute on all docs modified since we last computed.
-        TotalChanges = couch_db:count_changes_since(Db, CurSeq),
+        TotalUpdateChanges = couch_db:count_changes_since(Db, CurSeq),
+        TotalPurgeChanges = count_pending_purged_docs_since(Db, IndexPid),
+        TotalChanges = TotalUpdateChanges + TotalPurgeChanges,
 
         couch_task_status:add_task([
             {type, search_indexer},
@@ -47,14 +48,19 @@ update(IndexPid, Index) ->
         %% update status every half second
         couch_task_status:set_update_frequency(500),
 
+        %ExcludeIdRevs is [{Id1, Rev1}, {Id2, Rev2}, ...]
+        %The Rev is the final Rev, not purged Rev.
+        {ok, ExcludeIdRevs} = purge_index(Db, IndexPid, Index),
+        %% compute on all docs modified since we last computed.
+
         NewCurSeq = couch_db:get_update_seq(Db),
         Proc = get_os_process(Index#index.def_lang),
         try
             true = proc_prompt(Proc, [<<"add_fun">>, Index#index.def]),
             EnumFun = fun ?MODULE:load_docs/2,
-            Acc0 = {0, IndexPid, Db, Proc, TotalChanges, now()},
-
-            {ok, _} = couch_db:fold_changes(Db, CurSeq, EnumFun, Acc0),
+            [Changes] = couch_task_status:get([changes_done]),
+            Acc0 = {Changes, IndexPid, Db, Proc, TotalChanges, now(), ExcludeIdRevs},
+            {ok, _} = couch_db:fold_changes(Db, CurSeq, EnumFun, Acc0, []),
             ok = clouseau_rpc:commit(IndexPid, NewCurSeq)
         after
             ret_os_process(Proc)
@@ -64,10 +70,64 @@ update(IndexPid, Index) ->
         couch_db:close(Db)
     end.
 
-load_docs(FDI, {I, IndexPid, Db, Proc, Total, LastCommitTime}=Acc) ->
+load_docs(FDI, {I, IndexPid, Db, Proc, Total, LastCommitTime, ExcludeIdRevs}=Acc) ->
     couch_task_status:update([{changes_done, I}, {progress, (I * 100) div Total}]),
     DI = couch_doc:to_doc_info(FDI),
-    #doc_info{id=Id, high_seq=Seq, revs=[#rev_info{deleted=Del}|_]} = DI,
+    #doc_info{id=Id, high_seq=Seq, revs=[#rev_info{rev=Rev}|_]} = DI,
+    %check if it is processed in purge_index to avoid update the index again.
+    case lists:member({Id, Rev}, ExcludeIdRevs) of
+        true -> ok;
+        false -> update_or_delete_index(IndexPid, Db, DI, Proc)
+    end,
+    %% Force a commit every minute
+    case timer:now_diff(Now = now(), LastCommitTime) >= 60000000 of
+        true ->
+            ok = clouseau_rpc:commit(IndexPid, Seq),
+            {ok, {I+1, IndexPid, Db, Proc, Total, Now}};
+        false ->
+            {ok, setelement(1, Acc, I+1)}
+    end.
+
+purge_index(Db, IndexPid, Index) ->
+    {ok, IdxPurgeSeq} = clouseau_rpc:get_purge_seq(IndexPid),
+    Proc = get_os_process(Index#index.def_lang),
+    try
+        true = proc_prompt(Proc, [<<"add_fun">>, Index#index.def]),
+        FoldFun = fun({PurgeSeq, _UUID, Id, _Revs}, {Acc, _}) ->
+            Acc0 = case couch_db:get_full_doc_info(Db, Id) of
+                not_found ->
+                    ok = clouseau_rpc:delete(IndexPid, Id),
+                    Acc;
+                FDI ->
+                    DI = couch_doc:to_doc_info(FDI),
+                    #doc_info{id=Id, revs=[#rev_info{rev=Rev}|_]} = DI,
+                    case lists:member({Id, Rev}, Acc) of
+                        true -> Acc;
+                        false ->
+                            update_or_delete_index(IndexPid, Db, DI, Proc),
+                            [{Id, Rev} | Acc]
+                    end
+            end,
+            update_task(1),
+            {ok, {Acc0, PurgeSeq}}
+        end,
+
+        {ok, {ExcludeList, NewPurgeSeq}} = couch_db:fold_purge_infos(
+            Db, IdxPurgeSeq, FoldFun, {[], 0}, []),
+        clouseau_rpc:set_purge_seq(IndexPid, NewPurgeSeq),
+        update_local_doc(Db, Index, NewPurgeSeq),
+        {ok, ExcludeList}
+    after
+        ret_os_process(Proc)
+    end.
+
+count_pending_purged_docs_since(Db, IndexPid) ->
+    DbPurgeSeq = couch_db:get_purge_seq(Db),
+    {ok, IdxPurgeSeq} = clouseau_rpc:get_purge_seq(IndexPid),
+    DbPurgeSeq - IdxPurgeSeq.
+
+update_or_delete_index(IndexPid, Db, DI, Proc) ->
+    #doc_info{id=Id, revs=[#rev_info{deleted=Del}|_]} = DI,
     case Del of
         true ->
             ok = clouseau_rpc:delete(IndexPid, Id);
@@ -80,12 +140,20 @@ load_docs(FDI, {I, IndexPid, Db, Proc, Total, LastCommitTime}=Acc) ->
                 [] -> ok = clouseau_rpc:delete(IndexPid, Id);
                 _  -> ok = clouseau_rpc:update(IndexPid, Id, Fields1)
             end
-    end,
-    %% Force a commit every minute
-    case timer:now_diff(Now = now(), LastCommitTime) >= 60000000 of
-        true ->
-            ok = clouseau_rpc:commit(IndexPid, Seq),
-            {ok, {I+1, IndexPid, Db, Proc, Total, Now}};
-        false ->
-            {ok, setelement(1, Acc, I+1)}
     end.
+
+update_local_doc(Db, Index, PurgeSeq) ->
+    DocId = dreyfus_util:get_local_purge_doc_id(Index#index.sig),
+    DocContent = dreyfus_util:get_local_purge_doc_body(Db, DocId, PurgeSeq, Index),
+    couch_db:update_doc(Db, DocContent, []).
+
+update_task(NumChanges) ->
+    [Changes, Total] = couch_task_status:get([changes_done, total_changes]),
+    Changes2 = Changes + NumChanges,
+    Progress = case Total of
+        0 ->
+            0;
+        _ ->
+            (Changes2 * 100) div Total
+    end,
+    couch_task_status:update([{progress, Progress}, {changes_done, Changes2}]).
diff --git a/src/dreyfus_plugin_couch_db.erl b/src/dreyfus_plugin_couch_db.erl
new file mode 100644
index 000000000..b9f48ba74
--- /dev/null
+++ b/src/dreyfus_plugin_couch_db.erl
@@ -0,0 +1,26 @@
+% 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(dreyfus_plugin_couch_db).
+
+-export([
+    is_valid_purge_client/2,
+    on_compact/2
+]).
+
+
+is_valid_purge_client(DbName, Props) ->
+    dreyfus_util:verify_index_exists(DbName, Props).
+
+
+on_compact(DbName, DDocs) ->
+    dreyfus_util:ensure_local_purge_docs(DbName, DDocs).
diff --git a/src/dreyfus_util.erl b/src/dreyfus_util.erl
index 82e4292b4..e3c26999b 100644
--- a/src/dreyfus_util.erl
+++ b/src/dreyfus_util.erl
@@ -22,6 +22,16 @@
 -export([get_shards/2, sort/2, upgrade/1, export/1, time/2]).
 -export([in_black_list/1, in_black_list/3, maybe_deny_index/3]).
 -export([get_design_docid/1]).
+-export([
+    ensure_local_purge_docs/2,
+    get_value_from_options/2,
+    get_local_purge_doc_id/1,
+    get_local_purge_doc_body/4,
+    maybe_create_local_purge_doc/2,
+    maybe_create_local_purge_doc/3,
+    get_signature_from_idxdir/1,
+    verify_index_exists/2
+]).
 
 get_shards(DbName, #index_query_args{stale=ok}) ->
     mem3:ushards(DbName);
@@ -196,7 +206,136 @@ maybe_deny_index(DbName, GroupId, IndexName) ->
     end.
 
 get_design_docid(#doc{id = <<"_design/", DesignName/binary>>}) ->
-  DesignName.
+    DesignName.
+
+get_value_from_options(Key, Options) ->
+    case couch_util:get_value(Key, Options) of
+        undefined ->
+            Reason = binary_to_list(Key) ++ " must exist in Options.",
+            throw({bad_request, Reason});
+        Value -> Value
+    end.
+
+ensure_local_purge_docs(DbName, DDocs) ->
+    couch_util:with_db(DbName, fun(Db) ->
+        lists:foreach(fun(DDoc) ->
+            #doc{body = {Props}} = DDoc,
+            case couch_util:get_value(<<"indexes">>, Props) of
+                undefined -> false;
+                _ ->
+                    try dreyfus_index:design_doc_to_indexes(DDoc) of
+                        SIndexes -> ensure_local_purge_doc(Db, SIndexes)
+                    catch _:_ ->
+                        ok
+                    end
+            end
+        end, DDocs)
+    end).
+
+ensure_local_purge_doc(Db, SIndexes) ->
+    if SIndexes =/= [] ->
+        lists:map(fun(SIndex) ->
+            maybe_create_local_purge_doc(Db, SIndex)
+        end, SIndexes);
+    true -> ok end.
+
+maybe_create_local_purge_doc(Db, Index) ->
+    DocId = dreyfus_util:get_local_purge_doc_id(Index#index.sig),
+    case couch_db:open_doc(Db, DocId) of
+        {not_found, _} ->
+            DbPurgeSeq = couch_db:get_purge_seq(Db),
+            DocContent = dreyfus_util:get_local_purge_doc_body(
+                Db, DocId, DbPurgeSeq, Index),
+            couch_db:update_doc(Db, DocContent, []);
+        _ ->
+            ok
+    end.
+
+maybe_create_local_purge_doc(Db, IndexPid, Index) ->
+    DocId = dreyfus_util:get_local_purge_doc_id(Index#index.sig),
+    case couch_db:open_doc(Db, DocId) of
+        {not_found, _} ->
+            DbPurgeSeq = couch_db:get_purge_seq(Db),
+            clouseau_rpc:set_purge_seq(IndexPid, DbPurgeSeq),
+            DocContent = dreyfus_util:get_local_purge_doc_body(
+                Db, DocId, DbPurgeSeq, Index),
+            couch_db:update_doc(Db, DocContent, []);
+        _ ->
+            ok
+    end.
+
+get_local_purge_doc_id(Sig) ->
+    ?l2b(?LOCAL_DOC_PREFIX ++ "purge-" ++ "dreyfus-" ++ Sig).
+
+get_signature_from_idxdir(IdxDir) ->
+    IdxDirList = filename:split(IdxDir),
+    Sig = lists:last(IdxDirList),
+    case [Ch || Ch <- Sig, not (((Ch >= $0) and (Ch =< $9))
+        orelse ((Ch >= $a) and (Ch =< $f))
+        orelse ((Ch >= $A) and (Ch =< $F)))] == [] of
+        true -> Sig;
+        false -> undefined
+    end.
+
+get_local_purge_doc_body(Db, LocalDocId, PurgeSeq, Index) ->
+    #index{
+        name = IdxName,
+        ddoc_id = DDocId,
+        sig = Sig
+    } = Index,
+    {Mega, Secs, _} = os:timestamp(),
+    NowSecs = Mega * 1000000 + Secs,
+    JsonList = {[
+        {<<"_id">>, LocalDocId},
+        {<<"purge_seq">>, PurgeSeq},
+        {<<"timestamp_utc">>, NowSecs},
+        {<<"indexname">>, IdxName},
+        {<<"ddoc_id">>, DDocId},
+        {<<"signature">>, Sig},
+        {<<"type">>, <<"dreyfus">>}
+    ]},
+    couch_doc:from_json_obj(JsonList).
+
+verify_index_exists(DbName, Props) ->
+    try
+        Type = couch_util:get_value(<<"type">>, Props),
+        if Type =/= <<"dreyfus">> -> false; true ->
+            DDocId = couch_util:get_value(<<"ddoc_id">>, Props),
+            IndexName = couch_util:get_value(<<"indexname">>, Props),
+            Sig = couch_util:get_value(<<"signature">>, Props),
+            couch_util:with_db(DbName, fun(Db) ->
+                {ok, DesignDocs} = couch_db:get_design_docs(Db),
+                case get_ddoc(DbName, DesignDocs, DDocId) of
+                    #doc{} = DDoc ->
+                        {ok, IdxState} = dreyfus_index:design_doc_to_index(
+                            DDoc, IndexName),
+                        IdxState#index.sig == Sig;
+                    not_found ->
+                        false
+                end
+            end)
+        end
+    catch _:_ ->
+      false
+    end.
+
+get_ddoc(<<"shards/", _/binary>> = _DbName, DesignDocs, DDocId) ->
+    DDocs = [couch_doc:from_json_obj(DD) || DD <- DesignDocs],
+    case lists:keyfind(DDocId, #doc.id, DDocs) of
+        #doc{} = DDoc -> DDoc;
+        false -> not_found
+    end;
+get_ddoc(DbName, DesignDocs, DDocId) ->
+    couch_util:with_db(DbName, fun(Db) ->
+        case lists:keyfind(DDocId, #full_doc_info.id, DesignDocs) of
+            #full_doc_info{} = DDocInfo ->
+                {ok, DDoc} = couch_db:open_doc_int(
+                    Db, DDocInfo, [ejson_body]),
+                 DDoc;
+            false ->
+                not_found
+        end
+    end).
 
 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").
diff --git a/test/dreyfus_purge_test.erl b/test/dreyfus_purge_test.erl
new file mode 100644
index 000000000..a40e8b1ae
--- /dev/null
+++ b/test/dreyfus_purge_test.erl
@@ -0,0 +1,867 @@
+% 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(dreyfus_purge_test).
+
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("dreyfus/include/dreyfus.hrl").
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("mem3/include/mem3.hrl").
+
+
+-export([test_purge_single/0, test_purge_multiple/0, test_purge_multiple2/0,
+        test_purge_conflict/0, test_purge_conflict2/0, test_purge_conflict3/0, test_purge_conflict4/0,
+        test_purge_update/0, test_purge_update2/0,
+        test_delete/0, test_delete_purge_conflict/0, test_delete_conflict/0,
+        test_all/0]).
+-export([test_verify_index_exists1/0, test_verify_index_exists2/0, test_verify_index_exists_failed/0,
+        test_local_doc/0, test_delete_local_doc/0, test_purge_search/0]).
+
+-compile(export_all).
+
+test_all() ->
+    test_purge_single(),
+    test_purge_multiple(),
+    test_purge_multiple2(),
+    test_purge_conflict(),
+    test_purge_conflict2(),
+    test_purge_conflict3(),
+    test_purge_conflict4(),
+    test_purge_update(),
+    test_purge_update2(),
+    test_delete(),
+    test_delete_purge_conflict(),
+    test_delete_conflict(),
+    test_verify_index_exists1(),
+    test_verify_index_exists2(),
+    test_verify_index_exists_failed(),
+    test_delete_local_doc(),
+    test_local_doc(),
+    test_purge_search(),
+    ok.
+
+test_purge_single() ->
+    DbName = db_name(),
+    create_db_docs(DbName),
+    {ok, _, HitCount1, _, _, _} = dreyfus_search(DbName, <<"apple">>),
+    ?assertEqual(HitCount1, 1),
+    purge_docs(DbName, [<<"apple">>]),
+    {ok, _, HitCount2, _, _, _} = dreyfus_search(DbName, <<"apple">>),
+    ?assertEqual(HitCount2, 0),
+    delete_db(DbName),
+    ok.
+
+test_purge_multiple() ->
+    Query = <<"color:red">>,
+
+    %create the db and docs
+    DbName = db_name(),
+    create_db_docs(DbName),
+
+    %first search request
+    {ok, _, HitCount1, _, _, _} = dreyfus_search(DbName, Query),
+
+    ?assertEqual(HitCount1, 5),
+
+    %purge 5 docs
+    purge_docs(DbName, [<<"apple">>, <<"tomato">>, <<"cherry">>, <<"haw">>,
+        <<"strawberry">>]),
+
+    %second search request
+    {ok, _, HitCount2, _, _, _} = dreyfus_search(DbName, Query),
+
+    ?assertEqual(HitCount2, 0),
+
+    %delete the db
+    delete_db(DbName),
+    ok.
+
+test_purge_multiple2() ->
+    %create the db and docs
+    DbName = db_name(),
+    create_db_docs(DbName),
+
+    Query = <<"color:red">>,
+
+    %first search request
+    {ok, _, HitCount1, _, _, _} = dreyfus_search(DbName, Query),
+
+    ?assertEqual(HitCount1, 5),
+
+    %purge 2 docs
+    purge_docs(DbName, [<<"apple">>, <<"tomato">>]),
+
+    %second search request
+    {ok, _, HitCount2, _, _, _} = dreyfus_search(DbName, Query),
+
+    ?assertEqual(HitCount2, 3),
+
+    %purge 2 docs
+    purge_docs(DbName, [<<"cherry">>, <<"haw">>]),
+
+    %third search request
+    {ok, _, HitCount3, _, _, _} = dreyfus_search(DbName, Query),
+
+    ?assertEqual(HitCount3, 1),
+
+    %delete the db
+    delete_db(DbName),
+    ok.
+
+test_purge_conflict() ->
+    %create dbs and docs
+    SourceDbName = db_name(),
+    timer:sleep(2000),
+    TargetDbName = db_name(),
+
+    create_db_docs(SourceDbName),
+    create_db_docs(TargetDbName, <<"green">>),
+
+    %first search
+    {ok, _, RedHitCount1, _RedHits1, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount1, _GreenHits1, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(5, RedHitCount1 + GreenHitCount1),
+
+    %do replicate and make conflicted docs
+    {ok, _} = fabric:update_doc(<<"_replicator">>, make_replicate_doc(
+        SourceDbName, TargetDbName), [?ADMIN_CTX]),
+
+    %%check doc version
+    wait_for_replicate(TargetDbName, [<<"apple">>, <<"tomato">>, <<"cherry">>,
+        <<"haw">>, <<"strawberry">>], 2, 5),
+
+    %second search
+    {ok, _, RedHitCount2, _RedHits2, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount2, _GreenHits2, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(5, RedHitCount2 + GreenHitCount2),
+
+    purge_docs(TargetDbName, [<<"apple">>, <<"tomato">>, <<"cherry">>, <<"haw">>,
+        <<"strawberry">>]),
+
+    %third search
+    {ok, _, RedHitCount3, _RedHits3, _, _} = dreyfus_search(TargetDbName,
+        <<"color:red">>),
+    {ok, _, GreenHitCount3, _GreenHits3, _, _} = dreyfus_search(TargetDbName,
+        <<"color:green">>),
+
+    ?assertEqual(5, RedHitCount3 + GreenHitCount3),
+    ?assertEqual(RedHitCount2, GreenHitCount3),
+    ?assertEqual(GreenHitCount2, RedHitCount3),
+
+    delete_db(SourceDbName),
+    delete_db(TargetDbName),
+    ok.
+
+test_purge_conflict2() ->
+    %create dbs and docs
+    SourceDbName = db_name(),
+    timer:sleep(2000),
+    TargetDbName = db_name(),
+
+    create_db_docs(SourceDbName),
+    create_db_docs(TargetDbName, <<"green">>),
+
+    %first search
+    {ok, _, RedHitCount1, _RedHits1, _, _} = dreyfus_search(TargetDbName,
+        <<"color:red">>),
+    {ok, _, GreenHitCount1, _GreenHits1, _, _} = dreyfus_search(TargetDbName,
+        <<"color:green">>),
+
+    ?assertEqual(5, RedHitCount1 + GreenHitCount1),
+
+    %do replicate and make conflicted docs
+    {ok, _} = fabric:update_doc(<<"_replicator">>, make_replicate_doc(
+        SourceDbName, TargetDbName), [?ADMIN_CTX]),
+
+    wait_for_replicate(TargetDbName, [<<"apple">>, <<"tomato">>, <<"cherry">>,
+        <<"haw">>, <<"strawberry">>], 2, 5),
+
+    %second search
+    {ok, _, RedHitCount2, _RedHits2, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount2, _GreenHits2, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(5, RedHitCount2 + GreenHitCount2),
+
+    purge_docs(TargetDbName, [<<"apple">>, <<"tomato">>, <<"cherry">>,
+        <<"haw">>, <<"strawberry">>]),
+    purge_docs(TargetDbName, [<<"apple">>, <<"tomato">>, <<"cherry">>,
+        <<"haw">>, <<"strawberry">>]),
+
+    %third search
+    {ok, _, RedHitCount3, _RedHits3, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount3, _GreenHits3, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(0, RedHitCount3 + GreenHitCount3),
+
+    delete_db(SourceDbName),
+    delete_db(TargetDbName),
+    ok.
+
+
+test_purge_conflict3() ->
+    %create dbs and docs
+    SourceDbName = db_name(),
+    timer:sleep(2000),
+    TargetDbName = db_name(),
+
+    create_db_docs(SourceDbName),
+    create_db_docs(TargetDbName, <<"green">>),
+
+    %first search
+    {ok, _, RedHitCount1, _RedHits1, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount1, _GreenHits1, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(5, RedHitCount1 + GreenHitCount1),
+
+    %do replicate and make conflicted docs
+    {ok, _} = fabric:update_doc(<<"_replicator">>, make_replicate_doc(
+        SourceDbName, TargetDbName), [?ADMIN_CTX]),
+
+    %%check doc version
+    wait_for_replicate(TargetDbName, [<<"apple">>, <<"tomato">>, <<"cherry">>,
+        <<"haw">>, <<"strawberry">>], 2, 5),
+
+    %second search
+    {ok, _, RedHitCount2, _RedHits2, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount2, _GreenHits2, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(5, RedHitCount2 + GreenHitCount2),
+
+    purge_docs(TargetDbName, [<<"apple">>, <<"tomato">>, <<"cherry">>,
+        <<"haw">>, <<"strawberry">>]),
+
+    %third search
+    {ok, _, RedHitCount3, _RedHits3, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount3, _GreenHits3, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(5, RedHitCount3 + GreenHitCount3),
+    ?assertEqual(RedHitCount2, GreenHitCount3),
+    ?assertEqual(GreenHitCount2, RedHitCount3),
+
+    purge_docs(TargetDbName, [<<"apple">>, <<"tomato">>, <<"cherry">>,
+        <<"haw">>, <<"strawberry">>]),
+    {ok, _, RedHitCount4, _, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount4, _, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(0, RedHitCount4 + GreenHitCount4),
+
+    delete_db(SourceDbName),
+    delete_db(TargetDbName),
+    ok.
+
+test_purge_conflict4() ->
+    %create dbs and docs
+    SourceDbName = db_name(),
+    timer:sleep(2000),
+    TargetDbName = db_name(),
+
+    create_db_docs(SourceDbName, <<"green">>),
+    create_db_docs(TargetDbName, <<"red">>),
+
+    %first search
+    {ok, _, RedHitCount1, _RedHits1, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount1, _GreenHits1, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(5, RedHitCount1 + GreenHitCount1),
+
+    %do replicate and make conflicted docs
+    {ok, _} = fabric:update_doc(<<"_replicator">>, make_replicate_doc(
+        SourceDbName, TargetDbName), [?ADMIN_CTX]),
+
+    %%check doc version
+    wait_for_replicate(TargetDbName, [<<"apple">>, <<"tomato">>, <<"cherry">>,
+        <<"haw">>, <<"strawberry">>], 2, 5),
+
+    %second search
+    {ok, _, RedHitCount2, _RedHits2, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount2, _GreenHits2, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(5, RedHitCount2 + GreenHitCount2),
+
+    purge_docs_with_all_revs(TargetDbName, [<<"apple">>, <<"tomato">>,
+        <<"cherry">>, <<"haw">>, <<"strawberry">>]),
+
+    %third search
+    {ok, _, RedHitCount3, _RedHits3, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount3, _GreenHits3, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(0, RedHitCount3 + GreenHitCount3),
+
+    delete_db(SourceDbName),
+    delete_db(TargetDbName),
+    ok.
+
+test_purge_update() ->
+    %create the db and docs
+    DbName = db_name(),
+    create_db_docs(DbName),
+
+    QueryRed = <<"color:red">>,
+    QueryGreen = <<"color:green">>,
+
+    %first search request
+    {ok, _, HitCount1, _, _, _} = dreyfus_search(DbName, QueryRed),
+
+    ?assertEqual(HitCount1, 5),
+
+    %update doc
+    Rev = get_rev(DbName, <<"apple">>),
+    Doc = couch_doc:from_json_obj({[
+        {<<"_id">>, <<"apple">>},
+        {<<"_rev">>, couch_doc:rev_to_str(Rev)},
+        {<<"color">>, <<"green">>},
+        {<<"size">>, 8}
+    ]}),
+    {ok, _} = fabric:update_docs(DbName, [Doc], [?ADMIN_CTX]),
+
+    %second search request
+    {ok, _, HitCount2, _, _, _} = dreyfus_search(DbName, QueryRed),
+    {ok, _, HitCount3, _, _, _} = dreyfus_search(DbName, QueryGreen),
+
+    % 4 red and 1 green
+    ?assertEqual(HitCount2, 4),
+    ?assertEqual(HitCount3, 1),
+
+    % purge 2 docs, 1 red and 1 green
+    purge_docs(DbName, [<<"apple">>, <<"tomato">>]),
+
+    % third search request
+    {ok, _, HitCount4, _, _, _} = dreyfus_search(DbName, QueryRed),
+    {ok, _, HitCount5, _, _, _} = dreyfus_search(DbName, QueryGreen),
+
+    % 3 red and 0 green
+    ?assertEqual(HitCount4, 3),
+    ?assertEqual(HitCount5, 0),
+
+    delete_db(DbName),
+    ok.
+
+test_purge_update2() ->
+    %create the db and docs
+    DbName = db_name(),
+    create_db_docs(DbName),
+
+    Query1 = <<"size:1">>,
+    Query1000 = <<"size:1000">>,
+
+    %first search request
+    {ok, _, HitCount1, _, _, _} = dreyfus_search(DbName, Query1),
+    {ok, _, HitCount2, _, _, _} = dreyfus_search(DbName, Query1000),
+
+    ?assertEqual(HitCount1, 5),
+    ?assertEqual(HitCount2, 0),
+
+    %update doc 999 times, it will take about 30 seconds.
+    update_doc(DbName, <<"apple">>, 999),
+
+    %second search request
+    {ok, _, HitCount3, _, _, _} = dreyfus_search(DbName, Query1),
+    {ok, _, HitCount4, _, _, _} = dreyfus_search(DbName, Query1000),
+
+    % 4 value(1) and 1 value(1000)
+    ?assertEqual(HitCount3, 4),
+    ?assertEqual(HitCount4, 1),
+
+    % purge doc
+    purge_docs(DbName, [<<"apple">>]),
+
+    % third search request
+    {ok, _, HitCount5, _, _, _} = dreyfus_search(DbName, Query1),
+    {ok, _, HitCount6, _, _, _} = dreyfus_search(DbName, Query1000),
+
+    % 4 value(1) and 0 value(1000)
+    ?assertEqual(HitCount5, 4),
+    ?assertEqual(HitCount6, 0),
+
+    delete_db(DbName),
+    ok.
+
+test_delete() ->
+    DbName = db_name(),
+    create_db_docs(DbName),
+    {ok, _, HitCount1, _, _, _} = dreyfus_search(DbName, <<"apple">>),
+    ?assertEqual(HitCount1, 1),
+    ok = delete_docs(DbName, [<<"apple">>]),
+    {ok, _, HitCount2, _, _, _} = dreyfus_search(DbName, <<"apple">>),
+    ?assertEqual(HitCount2, 0),
+    delete_db(DbName),
+    ok.
+
+test_delete_conflict() ->
+    %create dbs and docs
+    SourceDbName = db_name(),
+    timer:sleep(2000),
+    TargetDbName = db_name(),
+
+    create_db_docs(SourceDbName),
+    create_db_docs(TargetDbName, <<"green">>),
+
+    %first search
+    {ok, _, RedHitCount1, _RedHits1, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount1, _GreenHits1, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(5, RedHitCount1 + GreenHitCount1),
+
+    %do replicate and make conflicted docs
+    {ok, _} = fabric:update_doc(<<"_replicator">>, make_replicate_doc(
+        SourceDbName, TargetDbName), [?ADMIN_CTX]),
+
+    wait_for_replicate(TargetDbName, [<<"apple">>, <<"tomato">>, <<"cherry">>,
+        <<"haw">>, <<"strawberry">>], 2, 5),
+
+    %second search
+    {ok, _, RedHitCount2, _RedHits2, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount2, _GreenHits2, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(5, RedHitCount2 + GreenHitCount2),
+
+    %delete docs
+    delete_docs(TargetDbName, [<<"apple">>, <<"tomato">>, <<"cherry">>,
+        <<"haw">>, <<"strawberry">>]),
+
+    %third search
+    {ok, _, RedHitCount3, _RedHits3, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount3, _GreenHits3, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(5, RedHitCount3 + GreenHitCount3),
+    ?assertEqual(RedHitCount2, GreenHitCount3),
+    ?assertEqual(GreenHitCount2, RedHitCount3),
+
+    delete_db(SourceDbName),
+    delete_db(TargetDbName),
+    ok.
+
+test_delete_purge_conflict() ->
+    %create dbs and docs
+    SourceDbName = db_name(),
+    timer:sleep(2000),
+    TargetDbName = db_name(),
+
+    create_db_docs(SourceDbName),
+    create_db_docs(TargetDbName, <<"green">>),
+
+    %first search
+    {ok, _, RedHitCount1, _RedHits1, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount1, _GreenHits1, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(5, RedHitCount1 + GreenHitCount1),
+
+    %do replicate and make conflicted docs
+    {ok, _} = fabric:update_doc(<<"_replicator">>, make_replicate_doc(
+        SourceDbName, TargetDbName), [?ADMIN_CTX]),
+
+    wait_for_replicate(TargetDbName, [<<"apple">>, <<"tomato">>, <<"cherry">>,
+        <<"haw">>, <<"strawberry">>], 2, 5),
+
+    %second search
+    {ok, _, RedHitCount2, _RedHits2, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount2, _GreenHits2, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(5, RedHitCount2 + GreenHitCount2),
+
+    %purge docs
+    purge_docs(TargetDbName, [<<"apple">>, <<"tomato">>, <<"cherry">>,
+        <<"haw">>, <<"strawberry">>]),
+
+    %delete docs
+    delete_docs(TargetDbName, [<<"apple">>, <<"tomato">>, <<"cherry">>,
+        <<"haw">>, <<"strawberry">>]),
+
+    %third search
+    {ok, _, RedHitCount3, _RedHits3, _, _} = dreyfus_search(
+        TargetDbName, <<"color:red">>),
+    {ok, _, GreenHitCount3, _GreenHits3, _, _} = dreyfus_search(
+        TargetDbName, <<"color:green">>),
+
+    ?assertEqual(RedHitCount3, 0),
+    ?assertEqual(GreenHitCount3, 0),
+    ?assertEqual(GreenHitCount3, 0),
+    ?assertEqual(RedHitCount3, 0),
+
+    delete_db(SourceDbName),
+    delete_db(TargetDbName),
+    ok.
+
+test_local_doc() ->
+    DbName = db_name(),
+    create_db_docs(DbName),
+
+    {ok, _, HitCount1, _, _, _} = dreyfus_search(DbName, <<"apple">>),
+    ?assertEqual(HitCount1, 1),
+    purge_docs(DbName, [<<"apple">>, <<"tomato">>, <<"cherry">>,
+        <<"strawberry">>]),
+    {ok, _, HitCount2, _, _, _} = dreyfus_search(DbName, <<"apple">>),
+    ?assertEqual(HitCount2, 0),
+
+    %get local doc
+    [Sig|_] = get_sigs(DbName),
+    LocalId = dreyfus_util:get_local_purge_doc_id(Sig),
+    LocalShards = mem3:local_shards(DbName),
+    PurgeSeqs = lists:map(fun(Shard) ->
+        {ok, Db} = couch_db:open_int(Shard#shard.name, [?ADMIN_CTX]),
+        {ok, LDoc} = couch_db:open_doc(Db, LocalId, []),
+        {Props} = couch_doc:to_json_obj(LDoc, []),
+        dreyfus_util:get_value_from_options(<<"timestamp_utc">>, Props),
+        PurgeSeq = dreyfus_util:get_value_from_options(<<"purge_seq">>, Props),
+        Type = dreyfus_util:get_value_from_options(<<"type">>, Props),
+        ?assertEqual(<<"dreyfus">>, Type),
+        couch_db:close(Db),
+        PurgeSeq
+    end, LocalShards),
+    ?assertEqual(lists:sum(PurgeSeqs), 4),
+
+    delete_db(DbName),
+    ok.
+
+test_verify_index_exists1() ->
+    DbName = db_name(),
+    create_db_docs(DbName),
+
+    {ok, _, HitCount1, _, _, _} = dreyfus_search(DbName, <<"apple">>),
+    ?assertEqual(HitCount1, 1),
+
+    ok = purge_docs(DbName, [<<"apple">>]),
+
+    {ok, _, HitCount2, _, _, _} = dreyfus_search(DbName, <<"apple">>),
+    ?assertEqual(HitCount2, 0),
+
+    ShardNames = [Sh || #shard{name = Sh} <- mem3:local_shards(DbName)],
+    [ShardDbName | _Rest ] = ShardNames,
+    {ok, Db} = couch_db:open(ShardDbName, [?ADMIN_CTX]),
+    {ok, LDoc} = couch_db:open_doc(Db,
+        dreyfus_util:get_local_purge_doc_id(
+            <<"49e82c2a910b1046b55cc45ad058a7ee">>), []
+    ),
+    #doc{body = {Props}} = LDoc,
+    ?assertEqual(true, dreyfus_util:verify_index_exists(ShardDbName, Props)),
+    delete_db(DbName),
+    ok.
+
+test_verify_index_exists2() ->
+    DbName = db_name(),
+    create_db_docs(DbName),
+
+    {ok, _, HitCount1, _, _, _} = dreyfus_search(DbName, <<"apple">>),
+    ?assertEqual(HitCount1, 1),
+
+    ShardNames = [Sh || #shard{name = Sh} <- mem3:local_shards(DbName)],
+    [ShardDbName | _Rest ] = ShardNames,
+    {ok, Db} = couch_db:open(ShardDbName, [?ADMIN_CTX]),
+    {ok, LDoc} = couch_db:open_doc(Db,
+        dreyfus_util:get_local_purge_doc_id(
+            <<"49e82c2a910b1046b55cc45ad058a7ee">>), []
+    ),
+    #doc{body = {Props}} = LDoc,
+    ?assertEqual(true, dreyfus_util:verify_index_exists(ShardDbName, Props)),
+
+    delete_db(DbName),
+    ok.
+
+test_verify_index_exists_failed() ->
+    DbName = db_name(),
+    create_db_docs(DbName),
+
+    {ok, _, HitCount1, _, _, _} = dreyfus_search(DbName, <<"apple">>),
+    ?assertEqual(HitCount1, 1),
+
+    ShardNames = [Sh || #shard{name = Sh} <- mem3:local_shards(DbName)],
+    [ShardDbName | _Rest ] = ShardNames,
+    {ok, Db} = couch_db:open(ShardDbName, [?ADMIN_CTX]),
+    {ok, LDoc} = couch_db:open_doc(Db,
+        dreyfus_util:get_local_purge_doc_id(
+            <<"49e82c2a910b1046b55cc45ad058a7ee">>), []
+    ),
+    #doc{body = {Options}} = LDoc,
+    OptionsDbErr = [
+        {<<"indexname">>,
+            dreyfus_util:get_value_from_options(<<"indexname">>, Options)},
+        {<<"ddoc_id">>,
+            dreyfus_util:get_value_from_options(<<"ddoc_id">>, Options)},
+        {<<"signature">>,
+            dreyfus_util:get_value_from_options(<<"signature">>, Options)}
+    ],
+    ?assertEqual(false, dreyfus_util:verify_index_exists(
+        ShardDbName, OptionsDbErr)),
+
+    OptionsIdxErr = [
+        {<<"indexname">>, <<"someindex">>},
+        {<<"ddoc_id">>,
+            dreyfus_util:get_value_from_options(<<"ddoc_id">>, Options)},
+        {<<"signature">>,
+            dreyfus_util:get_value_from_options(<<"signature">>, Options)}
+    ],
+    ?assertEqual(false, dreyfus_util:verify_index_exists(
+        ShardDbName, OptionsIdxErr)),
+
+    OptionsDDocErr = [
+        {<<"indexname">>,
+            dreyfus_util:get_value_from_options(<<"indexname">>, Options)},
+        {<<"ddoc_id">>,
+            <<"somedesigndoc">>},
+        {<<"signature">>,
+            dreyfus_util:get_value_from_options(<<"signature">>, Options)}
+    ],
+    ?assertEqual(false, dreyfus_util:verify_index_exists(
+        ShardDbName, OptionsDDocErr)),
+
+    OptionsSigErr = [
+        {<<"indexname">>,
+            dreyfus_util:get_value_from_options(<<"indexname">>, Options)},
+        {<<"ddoc_id">>,
+            dreyfus_util:get_value_from_options(<<"ddoc_id">>, Options)},
+        {<<"signature">>,
+            <<"12345678901234567890123456789012">>}
+    ],
+    ?assertEqual(false, dreyfus_util:verify_index_exists(
+        ShardDbName, OptionsSigErr)),
+
+    delete_db(DbName),
+    ok.
+
+test_delete_local_doc() ->
+    DbName = db_name(),
+    create_db_docs(DbName),
+
+    {ok, _, HitCount1, _, _, _} = dreyfus_search(DbName, <<"apple">>),
+    ?assertEqual(HitCount1, 1),
+
+    ok = purge_docs(DbName, [<<"apple">>]),
+
+    {ok, _, HitCount2, _, _, _} = dreyfus_search(DbName, <<"apple">>),
+    ?assertEqual(HitCount2, 0),
+
+    LDocId = dreyfus_util:get_local_purge_doc_id(
+        <<"49e82c2a910b1046b55cc45ad058a7ee">>),
+    ShardNames = [Sh || #shard{name = Sh} <- mem3:local_shards(DbName)],
+    [ShardDbName | _Rest ] = ShardNames,
+    {ok, Db} = couch_db:open(ShardDbName, [?ADMIN_CTX]),
+    {ok, _} = couch_db:open_doc(Db, LDocId, []),
+
+    delete_docs(DbName, [<<"_design/search">>]),
+    io:format("DbName ~p~n", [DbName]),
+    ?debugFmt("Converting ... ~n~p~n", [DbName]),
+
+
+    dreyfus_fabric_cleanup:go(DbName),
+    {ok, Db2} = couch_db:open(ShardDbName, [?ADMIN_CTX]),
+    {not_found, _} = couch_db:open_doc(Db2, LDocId, []),
+
+    delete_db(DbName),
+    ok.
+
+test_purge_search() ->
+    DbName = db_name(),
+    create_db_docs(DbName),
+    purge_docs(DbName, [<<"apple">>, <<"tomato">>, <<"haw">>]),
+    {ok, _, HitCount, _, _, _} = dreyfus_search(DbName, <<"color:red">>),
+    ?assertEqual(HitCount, 2),
+    delete_db(DbName),
+    ok.
+
+%private API
+db_name() ->
+    Nums = tuple_to_list(erlang:now()),
+    Prefix = "test-db",
+    Suffix = lists:concat([integer_to_list(Num) || Num <- Nums]),
+    list_to_binary(Prefix ++ "-" ++ Suffix).
+
+purge_docs(DBName, DocIds) ->
+    IdsRevs = [{DocId, [get_rev(DBName, DocId)]} || DocId <- DocIds],
+    {ok, _} = fabric:purge_docs(DBName, IdsRevs, []),
+    ok.
+
+purge_docs_with_all_revs(DBName, DocIds) ->
+    IdsRevs = [{DocId, get_revs(DBName, DocId)} || DocId <- DocIds],
+    {ok, _} = fabric:purge_docs(DBName, IdsRevs, []),
+    ok.
+
+dreyfus_search(DbName, KeyWord) ->
+    QueryArgs = #index_query_args{q = KeyWord},
+    {ok, DDoc} = fabric:open_doc(DbName, <<"_design/search">>, []),
+    dreyfus_fabric_search:go(DbName, DDoc, <<"index">>, QueryArgs).
+
+create_db_docs(DbName) ->
+    create_db(DbName),
+    create_docs(DbName, 5, <<"red">>).
+
+create_db_docs(DbName, Color) ->
+    create_db(DbName),
+    create_docs(DbName, 5, Color).
+
+create_docs(DbName, Count, Color) ->
+    {ok, _} = fabric:update_docs(DbName, make_docs(Count, Color), [?ADMIN_CTX]),
+    {ok, _} = fabric:update_doc(DbName, make_design_doc(dreyfus), [?ADMIN_CTX]).
+
+create_db(DbName) ->
+    ok = fabric:create_db(DbName, [?ADMIN_CTX, {q, 1}]).
+
+delete_db(DbName) ->
+    ok = fabric:delete_db(DbName, [?ADMIN_CTX]).
+
+make_docs(Count, Color) ->
+    [make_doc(I, Color) || I <- lists:seq(1, Count)].
+
+make_doc(Id, Color) ->
+    couch_doc:from_json_obj({[
+        {<<"_id">>, get_value(Id)},
+        {<<"color">>, Color},
+        {<<"size">>, 1}
+    ]}).
+
+get_value(Key) ->
+    case Key of
+        1 -> <<"apple">>;
+        2 -> <<"tomato">>;
+        3 -> <<"cherry">>;
+        4 -> <<"strawberry">>;
+        5 -> <<"haw">>;
+        6 -> <<"carrot">>;
+        7 -> <<"pitaya">>;
+        8 -> <<"grape">>;
+        9 -> <<"date">>;
+        10 -> <<"watermelon">>
+    end.
+
+make_design_doc(dreyfus) ->
+    couch_doc:from_json_obj({[
+        {<<"_id">>, <<"_design/search">>},
+        {<<"language">>, <<"javascript">>},
+        {<<"indexes">>, {[
+            {<<"index">>, {[
+                {<<"analyzer">>, <<"standard">>},
+                {<<"index">>, <<
+                    "function (doc) { \n"
+                    "  index(\"default\", doc._id);\n"
+                    "  if(doc.color) {\n"
+                    "    index(\"color\", doc.color);\n"
+                    "  }\n"
+                    "  if(doc.size) {\n"
+                    "    index(\"size\", doc.size);\n"
+                    "  }\n"
+                    "}"
+                >>}
+            ]}}
+        ]}}
+    ]}).
+
+make_replicate_doc(SourceDbName, TargetDbName) ->
+    couch_doc:from_json_obj({[
+        {<<"_id">>, list_to_binary("replicate_fm_" ++
+            binary_to_list(SourceDbName) ++ "_to_" ++ binary_to_list(TargetDbName))},
+        {<<"source">>, list_to_binary("http://localhost:15984/" ++ SourceDbName)},
+        {<<"target">>, list_to_binary("http://localhost:15984/" ++ TargetDbName)}
+    ]}).
+
+get_rev(DbName, DocId) ->
+    FDI = fabric:get_full_doc_info(DbName, DocId, []),
+    #doc_info{revs = [#rev_info{} = PrevRev | _]} = couch_doc:to_doc_info(FDI),
+    PrevRev#rev_info.rev.
+
+get_revs(DbName, DocId) ->
+    FDI = fabric:get_full_doc_info(DbName, DocId, []),
+    #doc_info{ revs = Revs } = couch_doc:to_doc_info(FDI),
+    [Rev#rev_info.rev || Rev <- Revs].
+
+update_doc(_, _, 0) ->
+    ok;
+update_doc(DbName, DocId, Times) ->
+    Rev = get_rev(DbName, DocId),
+    Doc = couch_doc:from_json_obj({[
+        {<<"_id">>, <<"apple">>},
+        {<<"_rev">>, couch_doc:rev_to_str(Rev)},
+        {<<"size">>, 1001 - Times}
+    ]}),
+    {ok, _} = fabric:update_docs(DbName, [Doc], [?ADMIN_CTX]),
+    update_doc(DbName, DocId, Times-1).
+
+delete_docs(DbName, DocIds) ->
+    lists:foreach(
+        fun(DocId) -> ok = delete_doc(DbName, DocId) end,
+        DocIds
+    ).
+
+delete_doc(DbName, DocId) ->
+    Rev = get_rev(DbName, DocId),
+    DDoc = couch_doc:from_json_obj({[
+        {<<"_id">>, DocId},
+        {<<"_rev">>, couch_doc:rev_to_str(Rev)},
+        {<<"_deleted">>, true}
+    ]}),
+    {ok, _} = fabric:update_doc(DbName, DDoc, [?ADMIN_CTX]),
+    ok.
+
+wait_for_replicate(_, _, _, 0) ->
+    couch_log:notice("[~p] wait time out", [?MODULE]),
+    ok;
+wait_for_replicate(DbName, DocIds, ExpectRevCount ,TimeOut)
+    when is_list(DocIds) ->
+    [wait_for_replicate(DbName, DocId, ExpectRevCount ,TimeOut) || DocId <- DocIds];
+wait_for_replicate(DbName, DocId, ExpectRevCount ,TimeOut) ->
+    FDI = fabric:get_full_doc_info(DbName, DocId, []),
+    #doc_info{ revs = Revs } = couch_doc:to_doc_info(FDI),
+    case erlang:length(Revs) of
+        ExpectRevCount ->
+            couch_log:notice("[~p] wait end by expect, time used:~p, DocId:~p",
+                [?MODULE, 5-TimeOut, DocId]),
+            ok;
+        true ->
+            timer:sleep(1000),
+            wait_for_replicate(DbName, DocId, ExpectRevCount ,TimeOut-1)
+    end,
+    ok.
+
+get_sigs(DbName) ->
+    {ok, DesignDocs} = fabric:design_docs(DbName),
+    lists:usort(lists:flatmap(fun active_sigs/1,
+        [couch_doc:from_json_obj(DD) || DD <- DesignDocs])).
+
+active_sigs(#doc{body={Fields}}=Doc) ->
+    {RawIndexes} = couch_util:get_value(<<"indexes">>, Fields, {[]}),
+    {IndexNames, _} = lists:unzip(RawIndexes),
+    [begin
+         {ok, Index} = dreyfus_index:design_doc_to_index(Doc, IndexName),
+         Index#index.sig
+     end || IndexName <- IndexNames].