% 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_httpd_changes).

-export([handle_changes_req/2,
         handle_changes/3,
         handle_view_changes/3]).

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

handle_changes_req(#httpd{method='POST'}=Req, Db) ->
    couch_httpd:validate_ctype(Req, "application/json"),
    handle_changes_req1(Req, Db);
handle_changes_req(#httpd{method='GET'}=Req, Db) ->
    handle_changes_req1(Req, Db);
handle_changes_req(#httpd{path_parts=[_,<<"_changes">>]}=Req, _Db) ->
    couch_httpd:send_method_not_allowed(Req, "GET,HEAD,POST").

handle_changes_req1(Req, #db{name=DbName}=Db) ->
    AuthDbName = ?l2b(couch_config:get("couch_httpd_auth", "authentication_db")),
    case AuthDbName of
    DbName ->
        % in the authentication database, _changes is admin-only.
        ok = couch_db:check_is_admin(Db);
    _Else ->
        % on other databases, _changes is free for all.
        ok
    end,

    MakeCallback = fun(Resp) ->
        fun({change, {ChangeProp}=Change, _}, "eventsource") ->
            Seq = proplists:get_value(<<"seq">>, ChangeProp),
            couch_httpd:send_chunk(Resp, ["data: ", ?JSON_ENCODE(Change),
                              "\n", "id: ", ?JSON_ENCODE(Seq),
                              "\n\n"]);
        ({change, Change, _}, "continuous") ->
            couch_httpd:send_chunk(Resp, [?JSON_ENCODE(Change) | "\n"]);
        ({change, Change, Prepend}, _) ->
            couch_httpd:send_chunk(Resp, [Prepend, ?JSON_ENCODE(Change)]);
        (start, "eventsource") ->
            ok;
        (start, "continuous") ->
            ok;
        (start, _) ->
            couch_httpd:send_chunk(Resp, "{\"results\":[\n");
        ({stop, _EndSeq}, "eventsource") ->
            couch_httpd:end_json_response(Resp);
        ({stop, EndSeq}, "continuous") ->
            couch_httpd:send_chunk(
                Resp,
                [?JSON_ENCODE({[{<<"last_seq">>, EndSeq}]}) | "\n"]
            ),
            couch_httpd:end_json_response(Resp);
        ({stop, EndSeq}, _) ->
            couch_httpd:send_chunk(
                Resp,
                io_lib:format("\n],\n\"last_seq\":~w}\n", [EndSeq])
            ),
            couch_httpd:end_json_response(Resp);
        (timeout, _) ->
            couch_httpd:send_chunk(Resp, "\n")
        end
    end,
    ChangesArgs = parse_changes_query(Req, Db),
    ChangesFun = handle_changes(ChangesArgs, Req, Db),
    WrapperFun = case ChangesArgs#changes_args.feed of
    "normal" ->
        {ok, Info} = couch_db:get_db_info(Db),
        CurrentEtag = couch_httpd:make_etag(Info),
        fun(FeedChangesFun) ->
            couch_httpd:etag_respond(
                Req,
                CurrentEtag,
                fun() ->
                    {ok, Resp} = couch_httpd:start_json_response(
                         Req, 200, [{"ETag", CurrentEtag}]
                    ),
                    FeedChangesFun(MakeCallback(Resp))
                end
            )
        end;
    "eventsource" ->
        Headers = [
            {"Content-Type", "text/event-stream"},
            {"Cache-Control", "no-cache"}
        ],
        {ok, Resp} = couch_httpd:start_chunked_response(Req, 200, Headers),
        fun(FeedChangesFun) ->
            FeedChangesFun(MakeCallback(Resp))
        end;
    _ ->
        % "longpoll" or "continuous"
        {ok, Resp} = couch_httpd:start_json_response(Req, 200),
        fun(FeedChangesFun) ->
            FeedChangesFun(MakeCallback(Resp))
        end
    end,
    couch_stats_collector:increment(
        {httpd, clients_requesting_changes}
    ),
    try
        WrapperFun(ChangesFun)
    after
    couch_stats_collector:decrement(
        {httpd, clients_requesting_changes}
    )
    end.


handle_changes(ChangesArgs, Req, Db) ->

    case ChangesArgs#changes_args.filter of
        "_view" ->
            handle_view_changes(ChangesArgs, Req, Db);
        _ ->
            couch_changes:handle_changes(ChangesArgs, Req, Db)
    end.

%% wrapper around couch_mrview_changes.
%% This wrapper mimic couch_changes:handle_changes/3 and return a
%% Changefun that can be used by the handle_changes_req function. Also
%% while couch_mrview_changes:handle_changes/6 is returning tha view
%% changes this function return docs corresponding to the changes
%% instead so it can be used to replace the _view filter.
handle_view_changes(ChangesArgs, #httpd{method=Method}=Req, Db) ->
    %% parse view parameter
    {DDocId, VName} = parse_view_param(Req),

    %% get view options
    {Query, Queries, NoIndex, JsonReq} = case Req of
        {json_req, {Props}} ->
            {Q} = couch_util:get_value(<<"query">>, Props, {[]}),
            Queries1 = couch_util:get_value(<<"queries">>, Props,
                                            undefined),
            NoIndex1 = (couch_util:get_value(<<"use_index">>, Q,
                                            <<"yes">>) =:= <<"no">>),
            {Q, Queries1, NoIndex1, true};
        #httpd{}=Req ->
            Q = couch_httpd:qs(Req),
            Queries1 = case Method of
                'POST' ->
                    {Props} = couch_httpd:json_body_obj(Req),
                    couch_util:get_value(<<"queries">>, Props, undefined);
                _ -> undefined
            end,
            NoIndex1 = couch_httpd:qs_value(Req, "use_index", "yes") =:= "no",
            {Q, Queries1, NoIndex1, false}
    end,
    ViewOptions = parse_view_options(Query, JsonReq, []),


    QueriesOptions = case Queries of
        undefined -> undefined;
        _ ->
            lists:foldl(fun({Options}, Acc) ->
                    ViewOpts = parse_view_options(Options, true, []),
                    [ViewOpts | Acc]
            end, [], lists:reverse(Queries))
    end,

    {ok, Infos} = couch_mrview:get_info(Db, DDocId),
    IsIndexed = lists:member(<<"seq_indexed">>,
                             proplists:get_value(update_options, Infos,
                                                 [])),

    case {IsIndexed, NoIndex} of
        {true, false} ->
            handle_view_changes(Db, DDocId, VName, ViewOptions,
                                QueriesOptions, ChangesArgs, Req);
        {true, true} when ViewOptions /= [] orelse QueriesOptions /= undefined ->
            ?LOG_ERROR("Tried to filter a non sequence indexed view~n",[]),
            throw({bad_request, seqs_not_indexed});
        {false, _} when ViewOptions /= [] orelse QueriesOptions /= undefined ->
            ?LOG_ERROR("Tried to filter a non sequence indexed view~n",[]),
            throw({bad_request, seqs_not_indexed});
        {_, _} ->
            %% old method we are getting changes using the btree instead
            %% which is not efficient, log it
            ?LOG_WARN("Filter without using a seq_indexed view.~n", []),
            couch_changes:handle_changes(ChangesArgs, Req, Db)
    end.

handle_view_changes(#db{name=DbName}=Db0, DDocId, VName, ViewOptions,
                    QueriesOptions, ChangesArgs, Req) ->
    #changes_args{
        feed = ResponseType,
        since = Since,
        db_open_options = DbOptions,
        heartbeat=Heartbeat,
        timeout=Timeout} = ChangesArgs,

    Refresh = refresh_option(Req),

    Options0 = [{since, Since},
                {view_options, ViewOptions},
                {queries, QueriesOptions},
                {refresh, Refresh},
                {heartbeat, Heartbeat},
                {timeout, Timeout}],

    Options = case ResponseType of
        "continuous" -> [stream | Options0];
        "eventsource" -> [stream | Options0];
        "longpoll" -> [{stream, once} | Options0];
        _ -> Options0
    end,

    %% reopen the db with the db options given to the changes args
    couch_db:close(Db0),
    DbOptions1 = [{user_ctx, Db0#db.user_ctx} | DbOptions],
    {ok, Db} = couch_db:open(DbName, DbOptions1),


    %% initialise the changes fun
    ChangesFun = fun(Callback) ->
            Callback(start, ResponseType),

            Acc0 = {"", 0, Db, Callback, ChangesArgs},
            couch_mrview_changes:handle_changes(DbName, DDocId, VName,
                                               fun view_changes_cb/2,
                                               Acc0, Options)
    end,
    ChangesFun.


view_changes_cb(stop, {LastSeq, {_, _, _, Callback, Args}}) ->
    Callback({stop, LastSeq}, Args#changes_args.feed);

view_changes_cb(heartbeat, {_, _, _, Callback, Args}=Acc) ->
    Callback(timeout, Args#changes_args.feed),
    {ok, Acc};
view_changes_cb({{Seq, _Key, DocId}, Val},
                {Prepend, OldLimit, Db0, Callback, Args}=Acc) ->

    %% is the key removed from the index?
    Removed = case Val of
        {[{<<"_removed">>, true}]} -> true;
        _ -> false
    end,

    #changes_args{
        feed = ResponseType,
        limit = Limit} = Args,

    %% if the doc sequence is > to the one in the db record, reopen the
    %% database since it means we don't have the latest db value.
    Db = case Db0#db.update_seq >= Seq of
        true -> Db0;
        false ->
            {ok, Db1} = couch_db:reopen(Db0),
            Db1
    end,

    case couch_db:get_doc_info(Db, DocId) of
        {ok, DocInfo} ->
            %% get change row
            ChangeRow = view_change_row(Db, DocInfo, Args, Removed),

            %% emit change row
            Callback({change, ChangeRow, Prepend}, ResponseType),

            %% if we achieved the limit, stop here, else continue.
            NewLimit = OldLimit + 1,
            if Limit > NewLimit ->
                    {ok, {<<",\n">>, NewLimit, Db, Callback, Args}};
                true ->
                    {stop, {<<"">>, NewLimit, Db, Callback, Args}}
            end;
        {error, not_found} ->
            %% doc not found, continue
            {ok, Acc};
        Error ->
            throw(Error)
    end.


view_change_row(Db, DocInfo, Args, Removed) ->
    #doc_info{id = Id, high_seq = Seq, revs = Revs} = DocInfo,
    [#rev_info{rev=Rev, deleted=Del0} | _] = Revs,

    #changes_args{style=Style,
                  include_docs=InDoc,
                  doc_options = DocOpts,
                  conflicts=Conflicts}=Args,

    Changes = case Style of
        main_only ->
            [{[{<<"rev">>, couch_doc:rev_to_str(Rev)}]}];
        all_docs ->
            [{[{<<"rev">>, couch_doc:rev_to_str(R)}]}
                || #rev_info{rev=R} <- Revs]
    end,

    Del = case {Del0, Removed} of
        {true, _} -> deleted;
        {false, true} -> removed;
        _ -> false
    end,

    {[{<<"seq">>, Seq}, {<<"id">>, Id}, {<<"changes">>, Changes}] ++
     deleted_item(Del) ++ case InDoc of
            true ->
                Opts = case Conflicts of
                    true -> [deleted, conflicts];
                    false -> [deleted]
                end,
                Doc = couch_index_util:load_doc(Db, DocInfo, Opts),
                case Doc of
                    null ->
                        [{doc, null}];
                    _ ->
                        [{doc, couch_doc:to_json_obj(Doc, DocOpts)}]
                end;
            false ->
                []
    end}.

parse_changes_query(Req, Db) ->
    ChangesArgs = lists:foldl(fun({Key, Value}, Args) ->
        case {string:to_lower(Key), Value} of
        {"feed", _} ->
            Args#changes_args{feed=Value};
        {"descending", "true"} ->
            Args#changes_args{dir=rev};
        {"since", "now"} ->
            UpdateSeq = couch_util:with_db(Db#db.name, fun(WDb) ->
                                        couch_db:get_update_seq(WDb)
                                end),
            Args#changes_args{since=UpdateSeq};
        {"since", _} ->
            Args#changes_args{since=list_to_integer(Value)};
        {"last-event-id", _} ->
            Args#changes_args{since=list_to_integer(Value)};
        {"limit", _} ->
            Args#changes_args{limit=list_to_integer(Value)};
        {"style", _} ->
            Args#changes_args{style=list_to_existing_atom(Value)};
        {"heartbeat", "true"} ->
            Args#changes_args{heartbeat=true};
        {"heartbeat", _} ->
            Args#changes_args{heartbeat=list_to_integer(Value)};
        {"timeout", _} ->
            Args#changes_args{timeout=list_to_integer(Value)};
        {"include_docs", "true"} ->
            Args#changes_args{include_docs=true};
        {"attachments", "true"} ->
            Opts = Args#changes_args.doc_options,
            Args#changes_args{doc_options=[attachments|Opts]};
        {"att_encoding_info", "true"} ->
            Opts = Args#changes_args.doc_options,
            Args#changes_args{doc_options=[att_encoding_info|Opts]};
        {"conflicts", "true"} ->
            Args#changes_args{conflicts=true};
        {"filter", _} ->
            Args#changes_args{filter=Value};
        _Else -> % unknown key value pair, ignore.
            Args
        end
    end, #changes_args{}, couch_httpd:qs(Req)),
    %% if it's an EventSource request with a Last-event-ID header
    %% that should override the `since` query string, since it's
    %% probably the browser reconnecting.
    case ChangesArgs#changes_args.feed of
        "eventsource" ->
            case couch_httpd:header_value(Req, "last-event-id") of
                undefined ->
                    ChangesArgs;
                Value ->
                    ChangesArgs#changes_args{since=list_to_integer(Value)}
            end;
        _ ->
            ChangesArgs
    end.

parse_view_param({json_req, {Props}}) ->
    {Query} = couch_util:get_value(<<"query">>, Props),
    parse_view_param1(couch_util:get_value(<<"view">>, Query, <<"">>));
parse_view_param(Req) ->
    parse_view_param1(list_to_binary(couch_httpd:qs_value(Req, "view", ""))).

parse_view_param1(ViewParam) ->
    case re:split(ViewParam, <<"/">>) of
        [DName, ViewName] ->
            {<< "_design/", DName/binary >>, ViewName};
        _ ->
            throw({bad_request, "Invalid `view` parameter."})
    end.

parse_view_options([], _JsonReq, Acc) ->
    Acc;
parse_view_options([{K, V} | Rest], JsonReq, Acc) ->
    Acc1 = case couch_util:to_binary(K) of
        <<"reduce">> ->
            [{reduce, couch_mrview_http:parse_boolean(V)}];
        <<"key">> ->
            V1 = parse_json(V, JsonReq),
            [{start_key, V1}, {end_key, V1} | Acc];
        <<"keys">> ->
            [{keys, parse_json(V, JsonReq)} | Acc];
        <<"startkey">> ->
            [{start_key, parse_json(V, JsonReq)} | Acc];
        <<"start_key">> ->
            [{start_key, parse_json(V, JsonReq)} | Acc];
        <<"startkey_docid">> ->
            [{start_key_docid, couch_util:to_binary(V)} | Acc];
        <<"start_key_docid">> ->
            [{start_key_docid, couch_util:to_binary(V)} | Acc];
        <<"endkey">> ->
            [{end_key, parse_json(V, JsonReq)} | Acc];
        <<"end_key">> ->
            [{end_key, parse_json(V, JsonReq)} | Acc];
        <<"endkey_docid">> ->
            [{start_key_docid, couch_util:to_binary(V)} | Acc];
        <<"end_key_docid">> ->
            [{start_key_docid, couch_util:to_binary(V)} | Acc];
        <<"limit">> ->
            [{limit, couch_mrview_http:parse_pos_int(V)} | Acc];
        <<"count">> ->
            throw({query_parse_error, <<"QS param `count` is not `limit`">>});
        <<"stale">> when V =:= <<"ok">> orelse V =:= "ok" ->
            [{stale, ok} | Acc];
        <<"stale">> when V =:= <<"update_after">> orelse V =:= "update_after" ->
            [{stale, update_after} | Acc];
        <<"stale">> ->
            throw({query_parse_error, <<"Invalid value for `stale`.">>});
        <<"descending">> ->
            case couch_mrview_http:parse_boolean(V) of
                true ->
                    [{direction, rev} | Acc];
                _ ->
                    [{direction, fwd} | Acc]
            end;
        <<"skip">> ->
            [{skip, couch_mrview_http:parse_pos_int(V)} | Acc];
        <<"group">> ->
            case couch_mrview_http:parse_booolean(V) of
                true ->
                    [{group_level, exact} | Acc];
                _ ->
                    [{group_level, 0} | Acc]
            end;
        <<"group_level">> ->
            [{group_level, couch_mrview_http:parse_pos_int(V)} | Acc];
        <<"inclusive_end">> ->
            [{inclusive_end, couch_mrview_http:parse_boolean(V)}];
        _ ->
            Acc
    end,
    parse_view_options(Rest, JsonReq, Acc1).

refresh_option({json_req, {Props}}) ->
    {Query} = couch_util:get_value(<<"query">>, Props),
    couch_util:get_value(<<"refresh">>, Query, true);
refresh_option(Req) ->
    case couch_httpd:qs_value(Req, "refresh", "true") of
        "false" -> false;
        _ -> true
    end.

parse_json(V, true) when is_binary(V) ->
    ?JSON_DECODE(V);
parse_json(V, false) when is_list(V) ->
    ?JSON_DECODE(V);
parse_json(V, _) ->
    V.

deleted_item(deleted) -> [{<<"deleted">>, true}];
deleted_item(removed) -> [{<<"removed">>, true}];
deleted_item(_) -> [].