% 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(fabric_db_create).
-export([go/2]).

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

%% @doc Create a new database, and all its partition files across the cluster
%%      Options is proplist with user_ctx, n, q, validate_name
go(DbName, Options) ->
    case validate_dbname(DbName, Options) of
        ok ->
            couch_partition:validate_dbname(DbName, Options),
            case db_exists(DbName) of
                true ->
                    {error, file_exists};
                false ->
                    {Shards, Doc} = generate_shard_map(DbName, Options),
                    CreateShardResult = create_shard_files(Shards, Options),
                    case CreateShardResult of
                        enametoolong ->
                            {error, {database_name_too_long, DbName}};
                        _ ->
                            case {CreateShardResult, create_shard_db_doc(Doc)} of
                                {ok, {ok, Status}} ->
                                    Status;
                                {ok, {error, conflict} = ShardDocError} ->
                                    % Check if it is just a race to create the shard doc
                                    case db_exists(DbName) of
                                        true -> {error, file_exists};
                                        false -> ShardDocError
                                    end;
                                {file_exists, {ok, _}} ->
                                    {error, file_exists};
                                {_, Error} ->
                                    Error
                            end
                    end
            end;
        Error ->
            Error
    end.

validate_dbname(DbName, Options) ->
    case couch_util:get_value(validate_name, Options, true) of
        false ->
            ok;
        true ->
            couch_db:validate_dbname(DbName)
    end.

generate_shard_map(DbName, Options) ->
    Suffix = mem3:generate_shard_suffix(),
    Shards = mem3:choose_shards(DbName, [{shard_suffix, Suffix} | Options]),
    case mem3_util:open_db_doc(DbName) of
        {ok, Doc} ->
            % the DB already exists, and may have a different Suffix
            ok;
        {not_found, _} ->
            Doc = make_document(Shards, Suffix, Options)
    end,
    {Shards, Doc}.

create_shard_files(Shards, Options) ->
    Workers = fabric_util:submit_jobs(Shards, create_db, [Options]),
    RexiMon = fabric_util:create_monitors(Shards),
    try fabric_util:recv(Workers, #shard.ref, fun handle_message/3, Workers) of
        {error, file_exists} ->
            file_exists;
        {error, enametoolong} ->
            enametoolong;
        {timeout, DefunctWorkers} ->
            fabric_util:log_timeout(DefunctWorkers, "create_db"),
            {error, timeout};
        _ ->
            ok
    after
        rexi_monitor:stop(RexiMon)
    end.

handle_message({error, enametoolong}, _, _) ->
    {error, enametoolong};
handle_message(file_exists, _, _) ->
    {error, file_exists};
handle_message({rexi_DOWN, _, {_, Node}, _}, _, Workers) ->
    case lists:filter(fun(S) -> S#shard.node =/= Node end, Workers) of
        [] ->
            {stop, ok};
        RemainingWorkers ->
            {ok, RemainingWorkers}
    end;
handle_message(_, Worker, Workers) ->
    case lists:delete(Worker, Workers) of
        [] ->
            {stop, ok};
        RemainingWorkers ->
            {ok, RemainingWorkers}
    end.

create_shard_db_doc(Doc) ->
    Shards = [#shard{node = N} || N <- mem3:nodes()],
    RexiMon = fabric_util:create_monitors(Shards),
    Workers = fabric_util:submit_jobs(Shards, create_shard_db_doc, [Doc]),
    Acc0 = {length(Shards), fabric_dict:init(Workers, nil)},
    try fabric_util:recv(Workers, #shard.ref, fun handle_db_update/3, Acc0) of
        {timeout, {_, WorkersDict}} ->
            DefunctWorkers = fabric_util:remove_done_workers(WorkersDict, nil),
            fabric_util:log_timeout(
                DefunctWorkers,
                "create_shard_db_doc"
            ),
            {error, timeout};
        Else ->
            Else
    after
        rexi_monitor:stop(RexiMon)
    end.

handle_db_update({rexi_DOWN, _, {_, Node}, _}, _Worker, {W, Counters}) ->
    New = fabric_dict:filter(fun(S, _) -> S#shard.node =/= Node end, Counters),
    maybe_stop(W, New);
handle_db_update({rexi_EXIT, _Reason}, Worker, {W, Counters}) ->
    maybe_stop(W, fabric_dict:erase(Worker, Counters));
handle_db_update(conflict, _, _) ->
    % just fail when we get any conflicts
    {error, conflict};
handle_db_update(Msg, Worker, {W, Counters}) ->
    maybe_stop(W, fabric_dict:store(Worker, Msg, Counters)).

maybe_stop(W, Counters) ->
    case fabric_dict:any(nil, Counters) of
        true ->
            {ok, {W, Counters}};
        false ->
            case lists:sum([1 || {_, ok} <- Counters]) of
                NumOk when NumOk >= (W div 2 + 1) ->
                    {stop, ok};
                NumOk when NumOk > 0 ->
                    {stop, accepted};
                _ ->
                    {error, internal_server_error}
            end
    end.

make_document([#shard{dbname = DbName} | _] = Shards, Suffix, Options) ->
    {RawOut, ByNodeOut, ByRangeOut} =
        lists:foldl(
            fun(#shard{node = N, range = [B, E]}, {Raw, ByNode, ByRange}) ->
                Range = ?l2b([
                    couch_util:to_hex(<<B:32/integer>>),
                    "-",
                    couch_util:to_hex(<<E:32/integer>>)
                ]),
                Node = couch_util:to_binary(N),
                {
                    [[<<"add">>, Range, Node] | Raw],
                    orddict:append(Node, Range, ByNode),
                    orddict:append(Range, Node, ByRange)
                }
            end,
            {[], [], []},
            Shards
        ),
    EngineProp =
        case couch_util:get_value(engine, Options) of
            E when is_binary(E) -> [{<<"engine">>, E}];
            _ -> []
        end,
    DbProps =
        case couch_util:get_value(props, Options) of
            Props when is_list(Props) -> [{<<"props">>, {Props}}];
            _ -> []
        end,
    #doc{
        id = DbName,
        body = {
            [
                {<<"shard_suffix">>, Suffix},
                {<<"changelog">>, lists:sort(RawOut)},
                {<<"by_node">>, {[{K, lists:sort(V)} || {K, V} <- ByNodeOut]}},
                {<<"by_range">>, {[{K, lists:sort(V)} || {K, V} <- ByRangeOut]}}
            ] ++ EngineProp ++ DbProps
        }
    }.

db_exists(DbName) -> is_list(catch mem3:shards(DbName)).

-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").

db_exists_test_() ->
    {
        setup,
        fun setup_all/0,
        fun teardown_all/1,
        [
            fun db_exists_for_existing_db/0,
            fun db_exists_for_missing_db/0
        ]
    }.

setup_all() ->
    meck:new(mem3).

teardown_all(_) ->
    meck:unload().

db_exists_for_existing_db() ->
    Mock = fun(DbName) when is_binary(DbName) ->
        [#shard{dbname = DbName, range = [0, 100]}]
    end,
    ok = meck:expect(mem3, shards, Mock),
    ?assertEqual(true, db_exists(<<"foobar">>)),
    ?assertEqual(true, meck:validate(mem3)).

db_exists_for_missing_db() ->
    Mock = fun(DbName) ->
        erlang:error(database_does_not_exist, [DbName])
    end,
    ok = meck:expect(mem3, shards, Mock),
    ?assertEqual(false, db_exists(<<"foobar">>)),
    ?assertEqual(false, meck:validate(mem3)).

-endif.