postgleam

-module(postgleam@stream).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/postgleam/stream.gleam").
-export([fetch_chunk/5, stream_query/6]).

-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.

-file("src/postgleam/stream.gleam", 36).
?DOC(" Fetch a single chunk of rows from an already-bound portal.\n").
-spec fetch_chunk(
    postgleam@connection:connection_state(),
    postgleam@connection:prepared_statement(),
    gleam@dict:dict(integer(), postgleam@codec:codec()),
    integer(),
    integer()
) -> {ok,
        {postgleam@connection:stream_chunk(),
            postgleam@connection:connection_state()}} |
    {error, postgleam@error:error()}.
fetch_chunk(State, Prepared, Registry, Max_rows, Timeout) ->
    postgleam@connection:execute_portal(
        State,
        Prepared,
        Registry,
        Max_rows,
        Timeout
    ).

-file("src/postgleam/stream.gleam", 99).
-spec list_append(list(JTF), list(JTF)) -> list(JTF).
list_append(A, B) ->
    case A of
        [] ->
            B;

        [X | Rest] ->
            [X | list_append(Rest, B)]
    end.

-file("src/postgleam/stream.gleam", 89).
-spec append_rows(
    list(list(gleam@option:option(postgleam@value:value()))),
    list(list(gleam@option:option(postgleam@value:value())))
) -> list(list(gleam@option:option(postgleam@value:value()))).
append_rows(Acc, New) ->
    case New of
        [] ->
            Acc;

        _ ->
            list_append(Acc, New)
    end.

-file("src/postgleam/stream.gleam", 106).
-spec ok(
    {ok, JTJ} | {error, postgleam@error:error()},
    fun((JTJ) -> {ok, JTM} | {error, postgleam@error:error()})
) -> {ok, JTM} | {error, postgleam@error:error()}.
ok(Result, Next) ->
    case Result of
        {ok, V} ->
            Next(V);

        {error, E} ->
            {error, E}
    end.

-file("src/postgleam/stream.gleam", 48).
-spec collect_chunks(
    postgleam@connection:connection_state(),
    postgleam@connection:prepared_statement(),
    gleam@dict:dict(integer(), postgleam@codec:codec()),
    integer(),
    integer(),
    postgleam@connection:stream_chunk(),
    list(list(gleam@option:option(postgleam@value:value())))
) -> {ok,
        {postgleam@connection:extended_query_result(),
            postgleam@connection:connection_state()}} |
    {error, postgleam@error:error()}.
collect_chunks(State, Prepared, Registry, Max_rows, Timeout, Chunk, Acc) ->
    case Chunk of
        {stream_more, Rows} ->
            New_acc = append_rows(Acc, Rows),
            ok(
                postgleam@connection:execute_portal(
                    State,
                    Prepared,
                    Registry,
                    Max_rows,
                    Timeout
                ),
                fun(_use0) ->
                    {Next_chunk, State@1} = _use0,
                    collect_chunks(
                        State@1,
                        Prepared,
                        Registry,
                        Max_rows,
                        Timeout,
                        Next_chunk,
                        New_acc
                    )
                end
            );

        {stream_done, Tag, Rows@1} ->
            All_rows = append_rows(Acc, Rows@1),
            ok(
                postgleam@connection:sync_portal(State, Timeout),
                fun(State@2) ->
                    {ok,
                        {{extended_query_result,
                                Tag,
                                erlang:element(5, Prepared),
                                All_rows},
                            State@2}}
                end
            )
    end.

-file("src/postgleam/stream.gleam", 15).
?DOC(
    " Fetch all rows from a query in chunks of max_rows.\n"
    " Returns the complete result after all chunks have been collected.\n"
).
-spec stream_query(
    postgleam@connection:connection_state(),
    postgleam@connection:prepared_statement(),
    list(gleam@option:option(postgleam@value:value())),
    gleam@dict:dict(integer(), postgleam@codec:codec()),
    integer(),
    integer()
) -> {ok,
        {postgleam@connection:extended_query_result(),
            postgleam@connection:connection_state()}} |
    {error, postgleam@error:error()}.
stream_query(State, Prepared, Params, Registry, Max_rows, Timeout) ->
    ok(
        postgleam@connection:bind_and_execute_portal(
            State,
            Prepared,
            Params,
            Registry,
            Max_rows,
            Timeout
        ),
        fun(_use0) ->
            {Chunk, State@1} = _use0,
            collect_chunks(
                State@1,
                Prepared,
                Registry,
                Max_rows,
                Timeout,
                Chunk,
                []
            )
        end
    ).