-module(mongo@aggregation).
-compile(no_auto_import).

-export([aggregate/1, stages/2, match/2, lookup/5, project/2, add_fields/2, sort/2, group/2, skip/2, limit/2, exec/1]).
-export_type([pipeline/0]).

-opaque pipeline() :: {pipeline,
        mongo@client:collection(),
        gleam@queue:queue(bson@types:value())}.

-spec aggregate(mongo@client:collection()) -> pipeline().
aggregate(Collection) ->
    {pipeline, Collection, gleam@queue:new()}.

-spec stages(pipeline(), list(bson@types:value())) -> pipeline().
stages(Pipeline, Docs) ->
    gleam@list:fold(
        Docs,
        Pipeline,
        fun(New_pipeline, Current) -> append_stage(New_pipeline, Current) end
    ).

-spec match(pipeline(), bson@types:value()) -> pipeline().
match(Pipeline, Doc) ->
    append_stage(Pipeline, {document, [{<<"$match"/utf8>>, Doc}]}).

-spec lookup(pipeline(), binary(), binary(), binary(), binary()) -> pipeline().
lookup(Pipeline, From, Local_field, Foreign_field, Alias) ->
    append_stage(
        Pipeline,
        {document,
            [{<<"$lookup"/utf8>>,
                    {document,
                        [{<<"from"/utf8>>, {str, From}},
                            {<<"localField"/utf8>>, {str, Local_field}},
                            {<<"foreignField"/utf8>>, {str, Foreign_field}},
                            {<<"as"/utf8>>, {str, Alias}}]}}]}
    ).

-spec project(pipeline(), bson@types:value()) -> pipeline().
project(Pipeline, Doc) ->
    append_stage(Pipeline, {document, [{<<"$project"/utf8>>, Doc}]}).

-spec add_fields(pipeline(), bson@types:value()) -> pipeline().
add_fields(Pipeline, Doc) ->
    append_stage(Pipeline, {document, [{<<"$addFields"/utf8>>, Doc}]}).

-spec sort(pipeline(), bson@types:value()) -> pipeline().
sort(Pipeline, Doc) ->
    append_stage(Pipeline, {document, [{<<"$sort"/utf8>>, Doc}]}).

-spec group(pipeline(), bson@types:value()) -> pipeline().
group(Pipeline, Doc) ->
    append_stage(Pipeline, {document, [{<<"$group"/utf8>>, Doc}]}).

-spec skip(pipeline(), integer()) -> pipeline().
skip(Pipeline, Count) ->
    append_stage(Pipeline, {document, [{<<"$skip"/utf8>>, {integer, Count}}]}).

-spec limit(pipeline(), integer()) -> pipeline().
limit(Pipeline, Count) ->
    append_stage(Pipeline, {document, [{<<"$limit"/utf8>>, {integer, Count}}]}).

-spec exec(pipeline()) -> {ok, list(bson@types:value())} |
    {error, mongo@utils:mongo_error()}.
exec(Pipeline) ->
    case mongo@client:execute(
        erlang:element(2, Pipeline),
        {document,
            [{<<"aggregate"/utf8>>,
                    {str, erlang:element(3, erlang:element(2, Pipeline))}},
                {<<"cursor"/utf8>>, {document, []}},
                {<<"pipeline"/utf8>>,
                    begin
                        _pipe = erlang:element(3, Pipeline),
                        _pipe@1 = gleam@queue:to_list(_pipe),
                        {array, _pipe@1}
                    end}]}
    ) of
        {ok, Result} ->
            [{<<"cursor"/utf8>>, {document, Result@1}}, {<<"ok"/utf8>>, Ok}] = Result,
            [{<<"firstBatch"/utf8>>, {array, Docs}},
                {<<"id"/utf8>>, _@1},
                {<<"ns"/utf8>>, _@2}] = Result@1,
            case Ok of
                {double, 1.0} ->
                    {ok, Docs};

                _@3 ->
                    {error, {mongo_error, -16, <<""/utf8>>, null}}
            end;

        {error, {Code, Msg}} ->
            {error, {mongo_error, Code, Msg, null}}
    end.

-spec append_stage(pipeline(), bson@types:value()) -> pipeline().
append_stage(Pipeline, Stage) ->
    {pipeline,
        erlang:element(2, Pipeline),
        begin
            _pipe = erlang:element(3, Pipeline),
            gleam@queue:push_back(_pipe, Stage)
        end}.