% 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(couch_index).
-behaviour(gen_server).

-vsn(2).

%% API
-export([start_link/1, stop/1, get_state/2, get_info/1]).
-export([trigger_update/2]).
-export([compact/1, compact/2, get_compactor_pid/1]).
-export([config_change/3]).

%% gen_server callbacks
-export([init/1, terminate/2, code_change/3]).
-export([handle_call/3, handle_cast/2, handle_info/2]).


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


-define(CHECK_INTERVAL, 600000). % 10 minutes
-define(RELISTEN_DELAY, 5000).

-record(st, {
    mod,
    idx_state,
    updater,
    compactor,
    waiters=[],
    commit_delay,
    committed=true,
    shutdown=false
}).


start_link({Module0, IdxState0}) ->
    [Module, IdxState] = couch_index_plugin:before_open(Module0, IdxState0),
    proc_lib:start_link(?MODULE, init, [{Module, IdxState}]).


stop(Pid) ->
    gen_server:cast(Pid, stop).


get_state(Pid, RequestSeq) ->
    gen_server:call(Pid, {get_state, RequestSeq}, infinity).


get_info(Pid) ->
    gen_server:call(Pid, get_info).


trigger_update(Pid, UpdateSeq) ->
    gen_server:cast(Pid, {trigger_update, UpdateSeq}).


compact(Pid) ->
    compact(Pid, []).


compact(Pid, Options) ->
    {ok, CPid} = gen_server:call(Pid, compact),
    case lists:member(monitor, Options) of
        true -> {ok, erlang:monitor(process, CPid)};
        false -> ok
    end.


get_compactor_pid(Pid) ->
    gen_server:call(Pid, get_compactor_pid).

config_change("query_server_config", "commit_freq", NewValue) ->
    ok = gen_server:cast(?MODULE, {config_update, NewValue}).


init({Mod, IdxState}) ->
    ok = config:subscribe_for_changes([{"query_server_config", "commit_freq"}]),
    DbName = Mod:get(db_name, IdxState),
    erlang:send_after(?CHECK_INTERVAL, self(), maybe_close),
    Resp = couch_util:with_db(DbName, fun(Db) ->
        case Mod:open(Db, IdxState) of
            {ok, IdxSt} ->
                couch_db:monitor(Db),
                {ok, IdxSt};
            Error ->
                Error
        end
    end),
    case Resp of
        {ok, NewIdxState} ->
            {ok, UPid} = couch_index_updater:start_link(self(), Mod),
            {ok, CPid} = couch_index_compactor:start_link(self(), Mod),
            Delay = config:get("query_server_config", "commit_freq", "5"),
            MsDelay = 1000 * list_to_integer(Delay),
            State = #st{
                mod=Mod,
                idx_state=NewIdxState,
                updater=UPid,
                compactor=CPid,
                commit_delay=MsDelay
            },
            Args = [
                Mod:get(db_name, IdxState),
                Mod:get(idx_name, IdxState),
                couch_index_util:hexsig(Mod:get(signature, IdxState))
            ],
            couch_log:info("Opening index for db: ~s idx: ~s sig: ~p", Args),
            proc_lib:init_ack({ok, self()}),
            gen_server:enter_loop(?MODULE, [], State);
        Other ->
            proc_lib:init_ack(Other)
    end.


terminate(Reason, State) ->
    #st{mod=Mod, idx_state=IdxState}=State,
    Mod:close(IdxState),
    send_all(State#st.waiters, Reason),
    couch_util:shutdown_sync(State#st.updater),
    couch_util:shutdown_sync(State#st.compactor),
    Args = [
        Mod:get(db_name, IdxState),
        Mod:get(idx_name, IdxState),
        couch_index_util:hexsig(Mod:get(signature, IdxState)),
        Reason
    ],
    couch_log:info("Closing index for db: ~s idx: ~s sig: ~p because ~r", Args),
    ok.


handle_call({get_state, ReqSeq}, From, State) ->
    #st{
        mod=Mod,
        idx_state=IdxState,
        waiters=Waiters
    } = State,
    IdxSeq = Mod:get(update_seq, IdxState),
    case ReqSeq =< IdxSeq of
        true ->
            {reply, {ok, IdxState}, State};
        _ -> % View update required
            couch_index_updater:run(State#st.updater, IdxState),
            Waiters2 = [{From, ReqSeq} | Waiters],
            {noreply, State#st{waiters=Waiters2}, infinity}
    end;
handle_call(get_info, _From, State) ->
    #st{mod=Mod} = State,
    IdxState = State#st.idx_state,
    {ok, Info0} = Mod:get(info, IdxState),
    IsUpdating = couch_index_updater:is_running(State#st.updater),
    IsCompacting = couch_index_compactor:is_running(State#st.compactor),
    IdxSeq = Mod:get(update_seq, IdxState),
    GetCommSeq = fun(Db) -> couch_db:get_committed_update_seq(Db) end,
    DbName = Mod:get(db_name, IdxState),
    CommittedSeq = couch_util:with_db(DbName, GetCommSeq),
    Info = Info0 ++ [
        {updater_running, IsUpdating},
        {compact_running, IsCompacting},
        {waiting_commit, State#st.committed == false},
        {waiting_clients, length(State#st.waiters)},
        {pending_updates, max(CommittedSeq - IdxSeq, 0)}
    ],
    {reply, {ok, Info}, State};
handle_call(reset, _From, State) ->
    #st{
        mod=Mod,
        idx_state=IdxState
    } = State,
    {ok, NewIdxState} = Mod:reset(IdxState),
    {reply, {ok, NewIdxState}, State#st{idx_state=NewIdxState}};
handle_call(compact, _From, State) ->
    Resp = couch_index_compactor:run(State#st.compactor, State#st.idx_state),
    {reply, Resp, State};
handle_call(get_compactor_pid, _From, State) ->
    {reply, {ok, State#st.compactor}, State};
handle_call({compacted, NewIdxState}, _From, State) ->
    #st{
        mod=Mod,
        idx_state=OldIdxState
    } = State,
    assert_signature_match(Mod, OldIdxState, NewIdxState),
    NewSeq = Mod:get(update_seq, NewIdxState),
    OldSeq = Mod:get(update_seq, OldIdxState),
    % For indices that require swapping files, we have to make sure we're
    % up to date with the current index. Otherwise indexes could roll back
    % (perhaps considerably) to previous points in history.
    case is_recompaction_enabled(NewIdxState, State) of
        true ->
            case NewSeq >= OldSeq of
                true -> {reply, ok, commit_compacted(NewIdxState, State)};
                false -> {reply, recompact, State}
            end;
        false ->
            {reply, ok, commit_compacted(NewIdxState, State)}
    end.

handle_cast({config_change, NewDelay}, State) ->
    MsDelay = 1000 * list_to_integer(NewDelay),
    {noreply, State#st{commit_delay=MsDelay}};
handle_cast({trigger_update, UpdateSeq}, State) ->
    #st{
        mod=Mod,
        idx_state=IdxState
    } = State,
    case UpdateSeq =< Mod:get(update_seq, IdxState) of
        true ->
            {noreply, State};
        false ->
            couch_index_updater:run(State#st.updater, IdxState),
            {noreply, State}
    end;
handle_cast({updated, NewIdxState}, State) ->
    {noreply, NewState} = handle_cast({new_state, NewIdxState}, State),
    case NewState#st.shutdown andalso (NewState#st.waiters =:= []) of
        true ->
            {stop, normal, NewState};
        false ->
            maybe_restart_updater(NewState),
            {noreply, NewState}
    end;
handle_cast({new_state, NewIdxState}, State) ->
    #st{
        mod=Mod,
        idx_state=OldIdxState,
        commit_delay=Delay
    } = State,
    assert_signature_match(Mod, OldIdxState, NewIdxState),
    CurrSeq = Mod:get(update_seq, NewIdxState),
    Args = [
        Mod:get(db_name, NewIdxState),
        Mod:get(idx_name, NewIdxState),
        CurrSeq
    ],
    couch_log:debug("Updated index for db: ~s idx: ~s seq: ~B", Args),
    Rest = send_replies(State#st.waiters, CurrSeq, NewIdxState),
    case State#st.committed of
        true -> erlang:send_after(Delay, self(), commit);
        false -> ok
    end,
    {noreply, State#st{
        idx_state=NewIdxState,
        waiters=Rest,
        committed=false
    }};
handle_cast({update_error, Error}, State) ->
    send_all(State#st.waiters, Error),
    {noreply, State#st{waiters=[]}};
handle_cast(stop, State) ->
    {stop, normal, State};
handle_cast(delete, State) ->
    #st{mod=Mod, idx_state=IdxState} = State,
    ok = Mod:delete(IdxState),
    {stop, normal, State};
handle_cast(ddoc_updated, State) ->
    #st{mod = Mod, idx_state = IdxState, waiters = Waiters} = State,
    DbName = Mod:get(db_name, IdxState),
    DDocId = Mod:get(idx_name, IdxState),
    Shutdown = couch_util:with_db(DbName, fun(Db) ->
        case couch_db:open_doc(Db, DDocId, [ejson_body, ?ADMIN_CTX]) of
            {not_found, deleted} ->
                true;
            {ok, DDoc} ->
                {ok, NewIdxState} = Mod:init(Db, DDoc),
                Mod:get(signature, NewIdxState) =/= Mod:get(signature, IdxState)
        end
    end),
    case Shutdown of
        true ->
            case Waiters of
                [] ->
                    {stop, normal, State};
                _ ->
                    {noreply, State#st{shutdown = true}}
            end;
        false ->
            {noreply, State#st{shutdown = false}}
    end;
handle_cast(_Mesg, State) ->
    {stop, unhandled_cast, State}.

handle_info(commit, #st{committed=true}=State) ->
    {noreply, State};
handle_info(commit, State) ->
    #st{mod=Mod, idx_state=IdxState, commit_delay=Delay} = State,
    DbName = Mod:get(db_name, IdxState),
    IdxName = Mod:get(idx_name, IdxState),
    GetCommSeq = fun(Db) -> couch_db:get_committed_update_seq(Db) end,
    CommittedSeq = couch_util:with_db(DbName, GetCommSeq),
    case CommittedSeq >= Mod:get(update_seq, IdxState) of
        true ->
            % Commit the updates
            ok = Mod:commit(IdxState),
            couch_event:notify(DbName, {index_commit, IdxName}),
            {noreply, State#st{committed=true}};
        _ ->
            % We can't commit the header because the database seq that's
            % fully committed to disk is still behind us. If we committed
            % now and the database lost those changes our view could be
            % forever out of sync with the database. But a crash before we
            % commit these changes, no big deal, we only lose incremental
            % changes since last committal.
            erlang:send_after(Delay, self(), commit),
            {noreply, State}
    end;
handle_info(maybe_close, State) ->
    % We need to periodically check if our index file still
    % exists on disk because index cleanups don't notify
    % the couch_index process when a file has been deleted. If
    % we don't check for this condition then the index can
    % remain open indefinitely wasting disk space.
    %
    % We make sure that we're idle before closing by looking
    % to see if we have any clients waiting for an update.
    Mod = State#st.mod,
    case State#st.waiters of
        [] ->
            case Mod:index_file_exists(State#st.idx_state) of
                true ->
                    erlang:send_after(?CHECK_INTERVAL, self(), maybe_close),
                    {noreply, State};
                false ->
                    {stop, normal, State}
            end;
        _ ->
            erlang:send_after(?CHECK_INTERVAL, self, maybe_close),
            {noreply, State}
    end;
handle_info({'DOWN', _, _, _Pid, _}, #st{mod=Mod, idx_state=IdxState}=State) ->
    Args = [Mod:get(db_name, IdxState), Mod:get(idx_name, IdxState)],
    couch_log:info("Index shutdown by monitor notice for db: ~s idx: ~s", Args),
    catch send_all(State#st.waiters, shutdown),
    {stop, normal, State#st{waiters=[]}};
handle_info({config_change, "query_server_config", "commit_freq", NewDelay, _}, State) ->
    handle_cast({config_change, NewDelay}, State);
handle_info({gen_event_EXIT, _Handler, _Reason}, State) ->
    erlang:send_after(?RELISTEN_DELAY, self(), restart_config_listener),
    {noreply, State};
handle_info(restart_config_listener, State) ->
    ok = config:subscribe_for_changes([{"query_server_config", "commit_freq"}]),
    {noreply, State}.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

maybe_restart_updater(#st{waiters=[]}) ->
    ok;
maybe_restart_updater(#st{mod=Mod, idx_state=IdxState}=State) ->
    couch_util:with_db(Mod:get(db_name, IdxState), fun(Db) ->
        UpdateSeq = couch_db:get_update_seq(Db),
        CommittedSeq = couch_db:get_committed_update_seq(Db),
        CanUpdate = UpdateSeq > CommittedSeq,
        UOpts = Mod:get(update_options, IdxState),
        case CanUpdate and lists:member(committed_only, UOpts) of
            true -> couch_db:ensure_full_commit(Db);
            false -> ok
        end
    end),
    couch_index_updater:run(State#st.updater, IdxState).


send_all(Waiters, Reply) ->
    [gen_server:reply(From, Reply) || {From, _} <- Waiters].


send_replies(Waiters, UpdateSeq, IdxState) ->
    Pred = fun({_, S}) -> S =< UpdateSeq end,
    {ToSend, Remaining} = lists:partition(Pred, Waiters),
    [gen_server:reply(From, {ok, IdxState}) || {From, _} <- ToSend],
    Remaining.

assert_signature_match(Mod, OldIdxState, NewIdxState) ->
    case {Mod:get(signature, OldIdxState), Mod:get(signature, NewIdxState)} of
        {Sig, Sig} -> ok;
        _ -> erlang:error(signature_mismatch)
    end.

commit_compacted(NewIdxState, State) ->
    #st{
        mod=Mod,
        idx_state=OldIdxState,
        updater=Updater,
        commit_delay=Delay
    } = State,
    {ok, NewIdxState1} = Mod:swap_compacted(OldIdxState, NewIdxState),
    % Restart the indexer if it's running.
    case couch_index_updater:is_running(Updater) of
        true -> ok = couch_index_updater:restart(Updater, NewIdxState1);
        false -> ok
    end,
    case State#st.committed of
        true -> erlang:send_after(Delay, self(), commit);
        false -> ok
    end,
    State#st{
        idx_state=NewIdxState1,
        committed=false
     }.

is_recompaction_enabled(IdxState, #st{mod = Mod}) ->
    DbName = binary_to_list(Mod:get(db_name, IdxState)),
    IdxName = binary_to_list(Mod:get(idx_name, IdxState)),
    IdxKey = DbName ++ ":" ++ IdxName,

    IdxSignature = couch_index_util:hexsig((Mod:get(signature, IdxState))),

    Global = get_value("view_compaction", "enabled_recompaction"),
    PerSignature = get_value("view_compaction.recompaction", IdxSignature),
    PerIdx = get_value("view_compaction.recompaction", IdxKey),
    PerDb = get_value("view_compaction.recompaction", DbName),

    find_most_specific([Global, PerDb, PerIdx, PerSignature], true).

find_most_specific(Settings, Default) ->
    Reversed = lists:reverse([Default | Settings]),
    [Value | _] = lists:dropwhile(fun(A) -> A =:= undefined end, Reversed),
    Value.

get_value(Section, Key) ->
    case config:get(Section, Key) of
        "enabled" -> true;
        "disabled" -> false;
        "true" -> true;
        "false" -> false;
        undefined -> undefined
    end.

-ifdef(TEST).
-include_lib("couch/include/couch_eunit.hrl").

get(db_name, _, _) ->
    <<"db_name">>;
get(idx_name, _, _) ->
    <<"idx_name">>;
get(signature, _, _) ->
    <<61,237,157,230,136,93,96,201,204,17,137,186,50,249,44,135>>.

setup(Settings) ->
    ok = meck:new([config], [passthrough]),
    ok = meck:new([test_index], [non_strict]),
    ok = meck:expect(config, get, fun(Section, Key) ->
        configure(Section, Key, Settings)
    end),
    ok = meck:expect(test_index, get, fun get/3),
    {undefined, #st{mod = {test_index}}}.

teardown(_, _) ->
    (catch meck:unload(config)),
    (catch meck:unload(test_index)),
    ok.

configure("view_compaction", "enabled_recompaction", [Global, _Db, _Index]) ->
    Global;
configure("view_compaction.recompaction", "db_name", [_Global, Db, _Index]) ->
    Db;
configure("view_compaction.recompaction", "db_name:" ++ _, [_, _, Index]) ->
    Index;
configure(Section, Key, _) ->
    meck:passthrough([Section, Key]).

recompaction_configuration_test_() ->
    {
        "Compaction tests",
        {
            setup,
            fun test_util:start_couch/0, fun test_util:stop_couch/1,
            {
                foreachx,
                fun setup/1, fun teardown/2,
                recompaction_configuration_tests()
            }
        }
    }.

recompaction_configuration_tests() ->
    AllCases = couch_tests_combinatorics:product([
        [undefined, "true", "false"],
        [undefined, "enabled", "disabled"],
        [undefined, "enabled", "disabled"]
    ]),

    EnabledCases = [
        [undefined, undefined, undefined],

        [undefined, undefined,"enabled"],
        [undefined, "enabled", undefined],
        [undefined, "disabled", "enabled"],
        [undefined, "enabled", "enabled"],

        ["true", undefined, undefined],
        ["true", undefined, "enabled"],
        ["true", "disabled", "enabled"],
        ["true", "enabled", undefined],
        ["true", "enabled", "enabled"],

        ["false", undefined, "enabled"],
        ["false", "enabled", undefined],
        ["false", "disabled", "enabled"],
        ["false", "enabled", "enabled"]
    ],

    DisabledCases = [
        [undefined, undefined, "disabled"],
        [undefined, "disabled", undefined],
        [undefined, "disabled", "disabled"],
        [undefined, "enabled", "disabled"],

        ["true", undefined, "disabled"],
        ["true", "disabled", undefined],
        ["true", "disabled", "disabled"],
        ["true", "enabled", "disabled"],

        ["false", undefined, undefined],
        ["false", undefined, "disabled"],
        ["false", "disabled", undefined],
        ["false", "disabled", "disabled"],
        ["false", "enabled", "disabled"]
    ],

    ?assertEqual([], AllCases -- (EnabledCases ++ DisabledCases)),

    [{Settings, fun should_not_call_recompact/2} || Settings <- DisabledCases]
    ++
    [{Settings, fun should_call_recompact/2} || Settings <- EnabledCases].

should_call_recompact(Settings, {IdxState, State}) ->
    {test_id(Settings), ?_test(begin
        ?assert(is_recompaction_enabled(IdxState, State)),
        ok
    end)}.

should_not_call_recompact(Settings, {IdxState, State}) ->
    {test_id(Settings), ?_test(begin
        ?assertNot(is_recompaction_enabled(IdxState, State)),
        ok
    end)}.

to_string(undefined) -> "undefined";
to_string(Value) -> Value.

test_id(Settings0) ->
    Settings1 = [to_string(Value) || Value <- Settings0],
    "[ " ++ lists:flatten(string:join(Settings1, " , ")) ++ " ]".

-endif.