glats

-module(glats@jetstream@consumer).
-compile([no_auto_import, nowarn_unused_vars]).

-export([request_next_message/2, info/3, create/3, subscribe/4, delete/3, names/2]).
-export_type([subscription_option/0, subscription/0, consumer_option/0, ack_policy/0, deliver_policy/0, replay_policy/0, consumer_info/0, consumer_config/0, sequence_info/0, request_message_option/0]).

-type subscription_option() :: {bind, binary(), binary()} |
    {bind_stream, binary()} |
    {with, consumer_option()}.

-opaque subscription() :: {pull_subscription,
        gleam@erlang@process:subject(glats:connection_message()),
        integer(),
        binary(),
        binary(),
        binary()} |
    {push_subscription,
        gleam@erlang@process:subject(glats:connection_message()),
        integer()}.

-type consumer_option() :: {durable_name, binary()} |
    {description, binary()} |
    {filter_subject, binary()} |
    {ack_policy, ack_policy()} |
    {ack_wait, integer()} |
    {deliver_policy, deliver_policy()} |
    {inactive_threshold, integer()} |
    {max_ack_pending, integer()} |
    {max_deliver, integer()} |
    {replay_policy, replay_policy()} |
    {num_replicas, integer()} |
    {sample_frequency, binary()} |
    {deliver_subject, binary()} |
    {deliver_group, binary()} |
    headers_only.

-type ack_policy() :: ack_explicit | ack_none | ack_all.

-type deliver_policy() :: deliver_all |
    deliver_last |
    deliver_last_per_subject |
    deliver_new |
    {deliver_by_start_sequence, integer()} |
    {deliver_by_start_time, binary()}.

-type replay_policy() :: replay_instant | replay_original.

-type consumer_info() :: {consumer_info,
        binary(),
        binary(),
        binary(),
        consumer_config(),
        sequence_info(),
        sequence_info(),
        integer(),
        integer(),
        integer(),
        integer()}.

-type consumer_config() :: {consumer_config,
        gleam@option:option(binary()),
        gleam@option:option(binary()),
        gleam@option:option(binary()),
        ack_policy(),
        gleam@option:option(integer()),
        deliver_policy(),
        gleam@option:option(integer()),
        gleam@option:option(integer()),
        gleam@option:option(integer()),
        replay_policy(),
        gleam@option:option(integer()),
        gleam@option:option(binary()),
        gleam@option:option(binary()),
        gleam@option:option(binary()),
        gleam@option:option(boolean())}.

-type sequence_info() :: {sequence_info, integer(), integer()}.

-type request_message_option() :: {batch, integer()} |
    no_wait |
    {expires, integer()}.

-spec apply_req_opt(
    gleam@map:map_(binary(), gleam@json:json()),
    request_message_option()
) -> gleam@map:map_(binary(), gleam@json:json()).
apply_req_opt(Prev, Opt) ->
    case Opt of
        {batch, Size} ->
            gleam@map:insert(Prev, <<"batch"/utf8>>, gleam@json:int(Size));

        {expires, Time} ->
            gleam@map:insert(Prev, <<"expires"/utf8>>, gleam@json:int(Time));

        no_wait ->
            gleam@map:insert(Prev, <<"no_wait"/utf8>>, gleam@json:bool(true))
    end.

-spec make_req_body(list(request_message_option())) -> binary().
make_req_body(Opts) ->
    _pipe = [{<<"batch"/utf8>>, gleam@json:int(1)}],
    _pipe@1 = gleam@map:from_list(_pipe),
    _pipe@2 = gleam@list:fold(Opts, _pipe@1, fun apply_req_opt/2),
    _pipe@3 = gleam@map:to_list(_pipe@2),
    _pipe@4 = gleam@json:object(_pipe@3),
    gleam@json:to_string(_pipe@4).

-spec do_req_next_msg(
    gleam@erlang@process:subject(glats:connection_message()),
    binary(),
    binary(),
    binary(),
    list(request_message_option())
) -> {ok, nil} | {error, glats@jetstream:jetstream_error()}.
do_req_next_msg(Conn, Stream, Consumer, Inbox, Opts) ->
    Subject = <<<<<<<<"$JS.API.CONSUMER"/utf8, ".MSG.NEXT."/utf8>>/binary,
                Stream/binary>>/binary,
            "."/utf8>>/binary,
        Consumer/binary>>,
    Reply_to = <<<<Inbox/binary, "."/utf8>>/binary,
        (glats@internal@util:random_string(6))/binary>>,
    _pipe = glats:publish_message(
        Conn,
        {message,
            Subject,
            gleam@map:new(),
            {some, Reply_to},
            make_req_body(Opts)}
    ),
    gleam@result:map_error(_pipe, fun(Err) -> {unknown, -1, Err} end).

-spec request_next_message(subscription(), list(request_message_option())) -> {ok,
        nil} |
    {error, glats@jetstream:jetstream_error()}.
request_next_message(Sub, Opts) ->
    case Sub of
        {pull_subscription, Conn, _, Stream, Consumer, Inbox} ->
            do_req_next_msg(Conn, Stream, Consumer, Inbox, Opts);

        _ ->
            {error,
                {pull_consumer_required,
                    <<"request_next_message only works on pull consumers"/utf8>>}}
    end.

-spec pull_subscribe(
    gleam@erlang@process:subject(glats:connection_message()),
    gleam@erlang@process:subject(glats:subscription_message()),
    binary(),
    binary()
) -> {ok, subscription()} | {error, glats@jetstream:jetstream_error()}.
pull_subscribe(Conn, Subscriber, Stream, Consumer) ->
    Inbox = glats@internal@util:random_inbox(<<""/utf8>>),
    _pipe = glats:subscribe(Conn, Subscriber, <<Inbox/binary, ".*"/utf8>>),
    _pipe@1 = gleam@result:map(
        _pipe,
        fun(_capture) ->
            {pull_subscription, Conn, _capture, Stream, Consumer, Inbox}
        end
    ),
    gleam@result:map_error(
        _pipe@1,
        fun(_) -> {unknown, -1, <<"unknown subscription error"/utf8>>} end
    ).

-spec push_subscribe(
    gleam@erlang@process:subject(glats:connection_message()),
    gleam@erlang@process:subject(glats:subscription_message()),
    binary(),
    gleam@option:option(binary())
) -> {ok, subscription()} | {error, glats@jetstream:jetstream_error()}.
push_subscribe(Conn, Subscriber, Subject, Group) ->
    _pipe = case Group of
        {some, Group@1} ->
            glats:queue_subscribe(Conn, Subscriber, Subject, Group@1);

        none ->
            glats:subscribe(Conn, Subscriber, Subject)
    end,
    _pipe@1 = gleam@result:map(
        _pipe,
        fun(_capture) -> {push_subscription, Conn, _capture} end
    ),
    gleam@result:map_error(
        _pipe@1,
        fun(_) -> {unknown, -1, <<"unknown subscription error"/utf8>>} end
    ).

-spec find_stream(
    gleam@erlang@process:subject(glats:connection_message()),
    binary(),
    list(subscription_option())
) -> {ok, binary()} | {error, glats@jetstream:jetstream_error()}.
find_stream(Conn, Subject, Opts) ->
    Stream@2 = gleam@list:find_map(Opts, fun(Opt) -> case Opt of
                {bind, Stream, _} ->
                    {ok, Stream};

                {bind_stream, Stream@1} ->
                    {ok, Stream@1};

                _ ->
                    {error, nil}
            end end),
    case Stream@2 of
        {ok, Stream@3} ->
            {ok, Stream@3};

        {error, nil} ->
            glats@jetstream@stream:find_stream_name_by_subject(Conn, Subject)
    end.

-spec ack_pol_to_string(ack_policy()) -> binary().
ack_pol_to_string(Pol) ->
    case Pol of
        ack_explicit ->
            <<"explicit"/utf8>>;

        ack_none ->
            <<"none"/utf8>>;

        ack_all ->
            <<"all"/utf8>>
    end.

-spec deliver_pol_to_list(deliver_policy()) -> list({binary(),
    gleam@json:json()}).
deliver_pol_to_list(Pol) ->
    case Pol of
        deliver_all ->
            [{<<"deliver_policy"/utf8>>, gleam@json:string(<<"all"/utf8>>)}];

        deliver_last ->
            [{<<"deliver_policy"/utf8>>, gleam@json:string(<<"last"/utf8>>)}];

        deliver_last_per_subject ->
            [{<<"deliver_policy"/utf8>>,
                    gleam@json:string(<<"last_per_subject"/utf8>>)}];

        deliver_new ->
            [{<<"deliver_policy"/utf8>>, gleam@json:string(<<"new"/utf8>>)}];

        {deliver_by_start_sequence, Seq} ->
            [{<<"deliver_policy"/utf8>>,
                    gleam@json:string(<<"by_start_sequence"/utf8>>)},
                {<<"opt_start_seq"/utf8>>, gleam@json:int(Seq)}];

        {deliver_by_start_time, Time} ->
            [{<<"deliver_policy"/utf8>>,
                    gleam@json:string(<<"by_start_time"/utf8>>)},
                {<<"opt_start_time"/utf8>>, gleam@json:string(Time)}]
    end.

-spec replay_pol_to_string(replay_policy()) -> binary().
replay_pol_to_string(Pol) ->
    case Pol of
        replay_instant ->
            <<"instant"/utf8>>;

        replay_original ->
            <<"original"/utf8>>
    end.

-spec apply_consumer_option(
    list({binary(), gleam@json:json()}),
    consumer_option()
) -> list({binary(), gleam@json:json()}).
apply_consumer_option(Prev, Opt) ->
    case Opt of
        {deliver_policy, Pol} ->
            _pipe = deliver_pol_to_list(Pol),
            gleam@list:append(Prev, _pipe);

        {ack_policy, Pol@1} ->
            _pipe@2 = {<<"ack_policy"/utf8>>,
                begin
                    _pipe@1 = ack_pol_to_string(Pol@1),
                    gleam@json:string(_pipe@1)
                end},
            gleam@list:prepend(Prev, _pipe@2);

        {replay_policy, Pol@2} ->
            _pipe@4 = {<<"replay_policy"/utf8>>,
                begin
                    _pipe@3 = replay_pol_to_string(Pol@2),
                    gleam@json:string(_pipe@3)
                end},
            gleam@list:prepend(Prev, _pipe@4);

        {durable_name, Name} ->
            _pipe@5 = {<<"durable_name"/utf8>>, gleam@json:string(Name)},
            gleam@list:prepend(Prev, _pipe@5);

        {description, Desc} ->
            _pipe@6 = {<<"description"/utf8>>, gleam@json:string(Desc)},
            gleam@list:prepend(Prev, _pipe@6);

        {filter_subject, Subj} ->
            _pipe@7 = {<<"filter_subject"/utf8>>, gleam@json:string(Subj)},
            gleam@list:prepend(Prev, _pipe@7);

        {ack_wait, Num} ->
            _pipe@8 = {<<"ack_wait"/utf8>>, gleam@json:int(Num)},
            gleam@list:prepend(Prev, _pipe@8);

        {inactive_threshold, Num@1} ->
            _pipe@9 = {<<"inactive_threshold"/utf8>>, gleam@json:int(Num@1)},
            gleam@list:prepend(Prev, _pipe@9);

        {max_ack_pending, Num@2} ->
            _pipe@10 = {<<"max_ack_pending"/utf8>>, gleam@json:int(Num@2)},
            gleam@list:prepend(Prev, _pipe@10);

        {max_deliver, Num@3} ->
            _pipe@11 = {<<"max_deliver"/utf8>>, gleam@json:int(Num@3)},
            gleam@list:prepend(Prev, _pipe@11);

        {num_replicas, Num@4} ->
            _pipe@12 = {<<"num_replicas"/utf8>>, gleam@json:int(Num@4)},
            gleam@list:prepend(Prev, _pipe@12);

        {sample_frequency, Freq} ->
            _pipe@13 = {<<"sample_freq"/utf8>>, gleam@json:string(Freq)},
            gleam@list:prepend(Prev, _pipe@13);

        {deliver_subject, Subj@1} ->
            _pipe@14 = {<<"deliver_subject"/utf8>>, gleam@json:string(Subj@1)},
            gleam@list:prepend(Prev, _pipe@14);

        {deliver_group, Group} ->
            _pipe@15 = {<<"deliver_group"/utf8>>, gleam@json:string(Group)},
            gleam@list:prepend(Prev, _pipe@15);

        headers_only ->
            _pipe@16 = {<<"headers_only"/utf8>>, gleam@json:bool(true)},
            gleam@list:prepend(Prev, _pipe@16)
    end.

-spec apply_consumer_options(
    list({binary(), gleam@json:json()}),
    list(consumer_option())
) -> list({binary(), gleam@json:json()}).
apply_consumer_options(Prev, Opts) ->
    gleam@list:fold(Opts, Prev, fun apply_consumer_option/2).

-spec consumer_options_to_json(binary(), list(consumer_option())) -> binary().
consumer_options_to_json(Stream, Opts) ->
    _pipe@1 = [{<<"stream_name"/utf8>>, gleam@json:string(Stream)},
        {<<"config"/utf8>>,
            begin
                _pipe = apply_consumer_options([], Opts),
                gleam@json:object(_pipe)
            end}],
    _pipe@2 = gleam@json:object(_pipe@1),
    gleam@json:to_string(_pipe@2).

-spec decode_info(binary()) -> {ok, consumer_info()} |
    {error, glats@jetstream:jetstream_error()}.
decode_info(Body) ->
    Decoder = gleam@dynamic:map(
        fun gleam@dynamic:string/1,
        fun gleam@dynamic:dynamic/1
    ),
    _pipe = gleam@json:decode(Body, Decoder),
    _pipe@1 = gleam@result:map(
        _pipe,
        fun 'Elixir.Glats.Jetstream':decode_consumer_info_data/1
    ),
    _pipe@2 = gleam@result:map_error(
        _pipe@1,
        fun(_) -> {-1, <<"decode error"/utf8>>} end
    ),
    _pipe@3 = gleam@result:flatten(_pipe@2),
    gleam@result:map_error(_pipe@3, fun glats@internal@js:map_code_to_error/1).

-spec info(
    gleam@erlang@process:subject(glats:connection_message()),
    binary(),
    binary()
) -> {ok, consumer_info()} | {error, glats@jetstream:jetstream_error()}.
info(Conn, Stream, Name) ->
    Subject = <<<<<<<<"$JS.API.CONSUMER"/utf8, ".INFO."/utf8>>/binary,
                Stream/binary>>/binary,
            "."/utf8>>/binary,
        Name/binary>>,
    case glats:request(Conn, Subject, <<""/utf8>>, 1000) of
        {ok, Msg} ->
            decode_info(erlang:element(5, Msg));

        {error, _} ->
            {error, {consumer_not_found, <<""/utf8>>}}
    end.

-spec create(
    gleam@erlang@process:subject(glats:connection_message()),
    binary(),
    list(consumer_option())
) -> {ok, consumer_info()} | {error, glats@jetstream:jetstream_error()}.
create(Conn, Stream, Opts) ->
    Durable_name = gleam@list:find(Opts, fun(O) -> case O of
                {durable_name, _} ->
                    true;

                _ ->
                    false
            end end),
    Subject = case Durable_name of
        {ok, {durable_name, Name}} ->
            <<<<<<<<"$JS.API.CONSUMER"/utf8, ".CREATE."/utf8>>/binary,
                        Stream/binary>>/binary,
                    "."/utf8>>/binary,
                Name/binary>>;

        {error, nil} ->
            <<<<"$JS.API.CONSUMER"/utf8, ".CREATE."/utf8>>/binary,
                Stream/binary>>
    end,
    Body = consumer_options_to_json(Stream, Opts),
    case glats:request(Conn, Subject, Body, 1000) of
        {ok, Msg} ->
            decode_info(erlang:element(5, Msg));

        {error, _} ->
            {error, {consumer_not_found, <<""/utf8>>}}
    end.

-spec ensure_consumer(
    gleam@erlang@process:subject(glats:connection_message()),
    binary(),
    binary(),
    list(consumer_option())
) -> {ok, consumer_info()} | {error, glats@jetstream:jetstream_error()}.
ensure_consumer(Conn, Stream, Subject, Opts) ->
    Opts@1 = case gleam@list:find(Opts, fun(Opt) -> case Opt of
                {filter_subject, _} ->
                    true;

                _ ->
                    false
            end end) of
        {ok, _} ->
            Opts;

        {error, nil} ->
            _pipe = Opts,
            gleam@list:prepend(_pipe, {filter_subject, Subject})
    end,
    create(Conn, Stream, Opts@1).

-spec find_consumer(
    gleam@erlang@process:subject(glats:connection_message()),
    binary(),
    binary(),
    list(subscription_option())
) -> {ok, consumer_info()} | {error, glats@jetstream:jetstream_error()}.
find_consumer(Conn, Stream, Subject, Opts) ->
    Consumer@1 = gleam@list:find_map(Opts, fun(Opt) -> case Opt of
                {bind, _, Consumer} ->
                    {ok, Consumer};

                _ ->
                    {error, nil}
            end end),
    case Consumer@1 of
        {ok, Consumer@2} ->
            info(Conn, Stream, Consumer@2);

        {error, nil} ->
            _pipe = gleam@list:filter_map(Opts, fun(Opt@1) -> case Opt@1 of
                        {with, O} ->
                            {ok, O};

                        _ ->
                            {error, nil}
                    end end),
            ensure_consumer(Conn, Stream, Subject, _pipe)
    end.

-spec subscribe(
    gleam@erlang@process:subject(glats:connection_message()),
    gleam@erlang@process:subject(glats:subscription_message()),
    binary(),
    list(subscription_option())
) -> {ok, subscription()} | {error, glats@jetstream:jetstream_error()}.
subscribe(Conn, Subscriber, Subject, Opts) ->
    gleam@result:then(
        find_stream(Conn, Subject, Opts),
        fun(Stream) ->
            gleam@result:then(
                find_consumer(Conn, Stream, Subject, Opts),
                fun(Consumer) ->
                    case erlang:element(14, erlang:element(5, Consumer)) of
                        none ->
                            pull_subscribe(
                                Conn,
                                Subscriber,
                                Stream,
                                erlang:element(3, Consumer)
                            );

                        {some, Subj} ->
                            push_subscribe(
                                Conn,
                                Subscriber,
                                Subj,
                                erlang:element(15, erlang:element(5, Consumer))
                            )
                    end
                end
            )
        end
    ).

-spec decode_delete(binary()) -> {ok, nil} |
    {error, glats@jetstream:jetstream_error()}.
decode_delete(Body) ->
    Decoder = gleam@dynamic:map(
        fun gleam@dynamic:string/1,
        fun gleam@dynamic:dynamic/1
    ),
    _pipe = gleam@json:decode(Body, Decoder),
    _pipe@1 = gleam@result:map(
        _pipe,
        fun 'Elixir.Glats.Jetstream':decode_delete_data/1
    ),
    _pipe@2 = gleam@result:map_error(
        _pipe@1,
        fun(_) -> {-1, <<"decode error"/utf8>>} end
    ),
    _pipe@3 = gleam@result:flatten(_pipe@2),
    gleam@result:map_error(_pipe@3, fun glats@internal@js:map_code_to_error/1).

-spec delete(
    gleam@erlang@process:subject(glats:connection_message()),
    binary(),
    binary()
) -> {ok, nil} | {error, glats@jetstream:jetstream_error()}.
delete(Conn, Stream, Name) ->
    Subject = <<<<<<<<"$JS.API.CONSUMER"/utf8, ".DELETE."/utf8>>/binary,
                Stream/binary>>/binary,
            "."/utf8>>/binary,
        Name/binary>>,
    case glats:request(Conn, Subject, <<""/utf8>>, 1000) of
        {ok, Msg} ->
            decode_delete(erlang:element(5, Msg));

        {error, _} ->
            {error, {stream_not_found, <<""/utf8>>}}
    end.

-spec decode_names(binary()) -> {ok, nil} |
    {error, glats@jetstream:jetstream_error()}.
decode_names(Body) ->
    Decoder = gleam@dynamic:map(
        fun gleam@dynamic:string/1,
        fun gleam@dynamic:dynamic/1
    ),
    _pipe = gleam@json:decode(Body, Decoder),
    _pipe@1 = gleam@result:map(
        _pipe,
        fun 'Elixir.Glats.Jetstream':decode_consumer_names_data/1
    ),
    _pipe@2 = gleam@result:map_error(
        _pipe@1,
        fun(_) -> {-1, <<"decode error"/utf8>>} end
    ),
    _pipe@3 = gleam@result:flatten(_pipe@2),
    gleam@result:map_error(_pipe@3, fun glats@internal@js:map_code_to_error/1).

-spec names(gleam@erlang@process:subject(glats:connection_message()), binary()) -> {ok,
        nil} |
    {error, glats@jetstream:jetstream_error()}.
names(Conn, Stream) ->
    Subject = <<<<"$JS.API.CONSUMER"/utf8, ".NAMES."/utf8>>/binary,
        Stream/binary>>,
    case glats:request(Conn, Subject, <<""/utf8>>, 1000) of
        {ok, Msg} ->
            decode_names(erlang:element(5, Msg));

        {error, _} ->
            {error, {stream_not_found, <<""/utf8>>}}
    end.