glats

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

-export([info/3, create/3, delete/3, names/2, request_batch/2, subscribe/4]).
-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_batch_option/0]).

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

-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()} |
    {max_waiting, integer()} |
    {max_request_expires, integer()} |
    {max_request_batch, integer()} |
    {max_request_max_bytes, integer()} |
    {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_batch_option() :: {batch, integer()} |
    no_wait |
    {expires, integer()}.

-file("src/glats/jetstream/consumer.gleam", 142).
-spec ack_policy_decoder() -> gleam@dynamic@decode:decoder(ack_policy()).
ack_policy_decoder() ->
    gleam@dynamic@decode:map(
        {decoder, fun gleam@dynamic@decode:decode_string/1},
        fun(Val) -> case Val of
                <<"all"/utf8>> ->
                    ack_all;

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

                <<"explicit"/utf8>> ->
                    ack_explicit;

                _ ->
                    ack_explicit
            end end
    ).

-file("src/glats/jetstream/consumer.gleam", 177).
-spec deliver_policy_decoder(
    gleam@option:option(integer()),
    gleam@option:option(binary())
) -> gleam@dynamic@decode:decoder(deliver_policy()).
deliver_policy_decoder(Opt_start_seq, Opt_start_time) ->
    Default = deliver_all,
    gleam@dynamic@decode:map(
        {decoder, fun gleam@dynamic@decode:decode_string/1},
        fun(Val) -> case Val of
                <<"all"/utf8>> ->
                    deliver_all;

                <<"last"/utf8>> ->
                    deliver_last;

                <<"last_per_subject"/utf8>> ->
                    deliver_last_per_subject;

                <<"new"/utf8>> ->
                    deliver_new;

                <<"by_start_sequence"/utf8>> ->
                    case Opt_start_seq of
                        {some, Seq} ->
                            {deliver_by_start_sequence, Seq};

                        none ->
                            Default
                    end;

                <<"by_start_time"/utf8>> ->
                    case Opt_start_time of
                        {some, Time} ->
                            {deliver_by_start_time, Time};

                        none ->
                            Default
                    end;

                _ ->
                    Default
            end end
    ).

-file("src/glats/jetstream/consumer.gleam", 216).
-spec replay_policy_decoder() -> gleam@dynamic@decode:decoder(replay_policy()).
replay_policy_decoder() ->
    gleam@dynamic@decode:map(
        {decoder, fun gleam@dynamic@decode:decode_string/1},
        fun(Val) -> case Val of
                <<"instant"/utf8>> ->
                    replay_instant;

                <<"original"/utf8>> ->
                    replay_original;

                _ ->
                    replay_instant
            end end
    ).

-file("src/glats/jetstream/consumer.gleam", 362).
-spec sequence_info_decoder() -> gleam@dynamic@decode:decoder(sequence_info()).
sequence_info_decoder() ->
    gleam@dynamic@decode:field(
        <<"consumer_seq"/utf8>>,
        {decoder, fun gleam@dynamic@decode:decode_int/1},
        fun(Consumer_seq) ->
            gleam@dynamic@decode:field(
                <<"stream_seq"/utf8>>,
                {decoder, fun gleam@dynamic@decode:decode_int/1},
                fun(Stream_seq) ->
                    gleam@dynamic@decode:success(
                        {sequence_info, Consumer_seq, Stream_seq}
                    )
                end
            )
        end
    ).

-file("src/glats/jetstream/consumer.gleam", 518).
-spec apply_req_opt(
    gleam@dict:dict(binary(), gleam@json:json()),
    request_batch_option()
) -> gleam@dict:dict(binary(), gleam@json:json()).
apply_req_opt(Prev, Opt) ->
    case Opt of
        {batch, Size} ->
            gleam@dict:insert(Prev, <<"batch"/utf8>>, gleam@json:int(Size));

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

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

-file("src/glats/jetstream/consumer.gleam", 509).
-spec make_req_body(list(request_batch_option())) -> binary().
make_req_body(Opts) ->
    _pipe = [{<<"batch"/utf8>>, gleam@json:int(1)}],
    _pipe@1 = maps:from_list(_pipe),
    _pipe@2 = gleam@list:fold(Opts, _pipe@1, fun apply_req_opt/2),
    _pipe@3 = maps:to_list(_pipe@2),
    _pipe@4 = gleam@json:object(_pipe@3),
    gleam@json:to_string(_pipe@4).

-file("src/glats/jetstream/consumer.gleam", 561).
-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
    ).

-file("src/glats/jetstream/consumer.gleam", 578).
-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, Topic, Group) ->
    _pipe@2 = glats:subscribe(
        Conn,
        Subscriber,
        Topic,
        begin
            _pipe = Group,
            _pipe@1 = gleam@option:map(
                _pipe,
                fun(Gr) -> [{queue_group, Gr}] end
            ),
            gleam@option:unwrap(_pipe@1, [])
        end
    ),
    _pipe@3 = gleam@result:map(
        _pipe@2,
        fun(_capture) -> {push_subscription, Conn, _capture} end
    ),
    gleam@result:map_error(
        _pipe@3,
        fun(_) -> {unknown, -1, <<"unknown subscription error"/utf8>>} end
    ).

-file("src/glats/jetstream/consumer.gleam", 599).
-spec find_stream(
    gleam@erlang@process:subject(glats:connection_message()),
    binary(),
    list(subscription_option())
) -> {ok, binary()} | {error, glats@jetstream:jetstream_error()}.
find_stream(Conn, Topic, 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, Topic)
    end.

-file("src/glats/jetstream/consumer.gleam", 763).
-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.

-file("src/glats/jetstream/consumer.gleam", 771).
-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.

-file("src/glats/jetstream/consumer.gleam", 790).
-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.

-file("src/glats/jetstream/consumer.gleam", 689).
-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),
            lists: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);

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

        {max_request_expires, Num@6} ->
            _pipe@15 = {<<"max_expires"/utf8>>, gleam@json:int(Num@6)},
            gleam@list:prepend(Prev, _pipe@15);

        {max_request_batch, Num@7} ->
            _pipe@16 = {<<"max_batch"/utf8>>, gleam@json:int(Num@7)},
            gleam@list:prepend(Prev, _pipe@16);

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

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

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

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

-file("src/glats/jetstream/consumer.gleam", 682).
-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).

-file("src/glats/jetstream/consumer.gleam", 669).
-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).

-file("src/glats/jetstream/consumer.gleam", 801).
-spec decode_optional_field(
    binary(),
    gleam@dynamic@decode:decoder(HTI),
    fun((gleam@option:option(HTI)) -> gleam@dynamic@decode:decoder(HTL))
) -> gleam@dynamic@decode:decoder(HTL).
decode_optional_field(Key, Field_decoder, Next) ->
    gleam@dynamic@decode:optional_field(
        Key,
        none,
        gleam@dynamic@decode:optional(Field_decoder),
        Next
    ).

-file("src/glats/jetstream/consumer.gleam", 318).
-spec consumer_config_decoder() -> gleam@dynamic@decode:decoder(consumer_config()).
consumer_config_decoder() ->
    gleam@dynamic@decode:field(
        <<"ack_policy"/utf8>>,
        ack_policy_decoder(),
        fun(Ack_policy) ->
            gleam@dynamic@decode:field(
                <<"replay_policy"/utf8>>,
                replay_policy_decoder(),
                fun(Replay_policy) ->
                    decode_optional_field(
                        <<"opt_start_seq"/utf8>>,
                        {decoder, fun gleam@dynamic@decode:decode_int/1},
                        fun(Opt_start_seq) ->
                            decode_optional_field(
                                <<"opt_start_time"/utf8>>,
                                {decoder,
                                    fun gleam@dynamic@decode:decode_string/1},
                                fun(Opt_start_time) ->
                                    gleam@dynamic@decode:field(
                                        <<"deliver_policy"/utf8>>,
                                        deliver_policy_decoder(
                                            Opt_start_seq,
                                            Opt_start_time
                                        ),
                                        fun(Deliver_policy) ->
                                            decode_optional_field(
                                                <<"durable_name"/utf8>>,
                                                {decoder,
                                                    fun gleam@dynamic@decode:decode_string/1},
                                                fun(Durable_name) ->
                                                    decode_optional_field(
                                                        <<"description"/utf8>>,
                                                        {decoder,
                                                            fun gleam@dynamic@decode:decode_string/1},
                                                        fun(Description) ->
                                                            decode_optional_field(
                                                                <<"filter_subject"/utf8>>,
                                                                {decoder,
                                                                    fun gleam@dynamic@decode:decode_string/1},
                                                                fun(
                                                                    Filter_subject
                                                                ) ->
                                                                    decode_optional_field(
                                                                        <<"ack_wait"/utf8>>,
                                                                        {decoder,
                                                                            fun gleam@dynamic@decode:decode_int/1},
                                                                        fun(
                                                                            Ack_wait
                                                                        ) ->
                                                                            decode_optional_field(
                                                                                <<"inactive_threshold"/utf8>>,
                                                                                {decoder,
                                                                                    fun gleam@dynamic@decode:decode_int/1},
                                                                                fun(
                                                                                    Inactive_threshold
                                                                                ) ->
                                                                                    decode_optional_field(
                                                                                        <<"max_ack_pending"/utf8>>,
                                                                                        {decoder,
                                                                                            fun gleam@dynamic@decode:decode_int/1},
                                                                                        fun(
                                                                                            Max_ack_pending
                                                                                        ) ->
                                                                                            decode_optional_field(
                                                                                                <<"max_deliver"/utf8>>,
                                                                                                {decoder,
                                                                                                    fun gleam@dynamic@decode:decode_int/1},
                                                                                                fun(
                                                                                                    Max_deliver
                                                                                                ) ->
                                                                                                    decode_optional_field(
                                                                                                        <<"num_replicas"/utf8>>,
                                                                                                        {decoder,
                                                                                                            fun gleam@dynamic@decode:decode_int/1},
                                                                                                        fun(
                                                                                                            Num_replicas
                                                                                                        ) ->
                                                                                                            decode_optional_field(
                                                                                                                <<"sample_freq"/utf8>>,
                                                                                                                {decoder,
                                                                                                                    fun gleam@dynamic@decode:decode_string/1},
                                                                                                                fun(
                                                                                                                    Sample_freq
                                                                                                                ) ->
                                                                                                                    decode_optional_field(
                                                                                                                        <<"deliver_subject"/utf8>>,
                                                                                                                        {decoder,
                                                                                                                            fun gleam@dynamic@decode:decode_string/1},
                                                                                                                        fun(
                                                                                                                            Deliver_subject
                                                                                                                        ) ->
                                                                                                                            decode_optional_field(
                                                                                                                                <<"deliver_group"/utf8>>,
                                                                                                                                {decoder,
                                                                                                                                    fun gleam@dynamic@decode:decode_string/1},
                                                                                                                                fun(
                                                                                                                                    Deliver_group
                                                                                                                                ) ->
                                                                                                                                    decode_optional_field(
                                                                                                                                        <<"headers_only"/utf8>>,
                                                                                                                                        {decoder,
                                                                                                                                            fun gleam@dynamic@decode:decode_bool/1},
                                                                                                                                        fun(
                                                                                                                                            Headers_only
                                                                                                                                        ) ->
                                                                                                                                            gleam@dynamic@decode:success(
                                                                                                                                                {consumer_config,
                                                                                                                                                    Durable_name,
                                                                                                                                                    Description,
                                                                                                                                                    Filter_subject,
                                                                                                                                                    Ack_policy,
                                                                                                                                                    Ack_wait,
                                                                                                                                                    Deliver_policy,
                                                                                                                                                    Inactive_threshold,
                                                                                                                                                    Max_ack_pending,
                                                                                                                                                    Max_deliver,
                                                                                                                                                    Replay_policy,
                                                                                                                                                    Num_replicas,
                                                                                                                                                    Sample_freq,
                                                                                                                                                    Deliver_subject,
                                                                                                                                                    Deliver_group,
                                                                                                                                                    Headers_only}
                                                                                                                                            )
                                                                                                                                        end
                                                                                                                                    )
                                                                                                                                end
                                                                                                                            )
                                                                                                                        end
                                                                                                                    )
                                                                                                                end
                                                                                                            )
                                                                                                        end
                                                                                                    )
                                                                                                end
                                                                                            )
                                                                                        end
                                                                                    )
                                                                                end
                                                                            )
                                                                        end
                                                                    )
                                                                end
                                                            )
                                                        end
                                                    )
                                                end
                                            )
                                        end
                                    )
                                end
                            )
                        end
                    )
                end
            )
        end
    ).

-file("src/glats/jetstream/consumer.gleam", 809).
-spec json_parse_jetstream_errorable(
    binary(),
    gleam@dynamic@decode:decoder(HTO)
) -> {ok, HTO} | {error, glats@jetstream:jetstream_error()}.
json_parse_jetstream_errorable(Body, Decoder) ->
    _pipe@1 = gleam@json:parse(
        Body,
        gleam@dynamic@decode:one_of(
            begin
                _pipe = Decoder,
                gleam@dynamic@decode:map(
                    _pipe,
                    fun(Field@0) -> {ok, Field@0} end
                )
            end,
            [begin
                    gleam@dynamic@decode:subfield(
                        [<<"error"/utf8>>, <<"err_code"/utf8>>],
                        {decoder, fun gleam@dynamic@decode:decode_int/1},
                        fun(Err_code) ->
                            gleam@dynamic@decode:subfield(
                                [<<"error"/utf8>>, <<"description"/utf8>>],
                                {decoder,
                                    fun gleam@dynamic@decode:decode_string/1},
                                fun(Description) ->
                                    gleam@dynamic@decode:success(
                                        {error, {Err_code, Description}}
                                    )
                                end
                            )
                        end
                    )
                end]
        )
    ),
    _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).

-file("src/glats/jetstream/consumer.gleam", 287).
-spec decode_info(binary()) -> {ok, consumer_info()} |
    {error, glats@jetstream:jetstream_error()}.
decode_info(Body) ->
    Decoder = begin
        gleam@dynamic@decode:field(
            <<"stream_name"/utf8>>,
            {decoder, fun gleam@dynamic@decode:decode_string/1},
            fun(Stream) ->
                gleam@dynamic@decode:field(
                    <<"name"/utf8>>,
                    {decoder, fun gleam@dynamic@decode:decode_string/1},
                    fun(Name) ->
                        gleam@dynamic@decode:field(
                            <<"created"/utf8>>,
                            {decoder, fun gleam@dynamic@decode:decode_string/1},
                            fun(Created) ->
                                gleam@dynamic@decode:field(
                                    <<"config"/utf8>>,
                                    consumer_config_decoder(),
                                    fun(Config) ->
                                        gleam@dynamic@decode:field(
                                            <<"delivered"/utf8>>,
                                            sequence_info_decoder(),
                                            fun(Delivered) ->
                                                gleam@dynamic@decode:field(
                                                    <<"ack_floor"/utf8>>,
                                                    sequence_info_decoder(),
                                                    fun(Ack_floor) ->
                                                        gleam@dynamic@decode:field(
                                                            <<"num_ack_pending"/utf8>>,
                                                            {decoder,
                                                                fun gleam@dynamic@decode:decode_int/1},
                                                            fun(Num_ack_pending) ->
                                                                gleam@dynamic@decode:field(
                                                                    <<"num_redelivered"/utf8>>,
                                                                    {decoder,
                                                                        fun gleam@dynamic@decode:decode_int/1},
                                                                    fun(
                                                                        Num_redelivered
                                                                    ) ->
                                                                        gleam@dynamic@decode:field(
                                                                            <<"num_waiting"/utf8>>,
                                                                            {decoder,
                                                                                fun gleam@dynamic@decode:decode_int/1},
                                                                            fun(
                                                                                Num_waiting
                                                                            ) ->
                                                                                gleam@dynamic@decode:field(
                                                                                    <<"num_pending"/utf8>>,
                                                                                    {decoder,
                                                                                        fun gleam@dynamic@decode:decode_int/1},
                                                                                    fun(
                                                                                        Num_pending
                                                                                    ) ->
                                                                                        gleam@dynamic@decode:success(
                                                                                            {consumer_info,
                                                                                                Stream,
                                                                                                Name,
                                                                                                Created,
                                                                                                Config,
                                                                                                Delivered,
                                                                                                Ack_floor,
                                                                                                Num_ack_pending,
                                                                                                Num_redelivered,
                                                                                                Num_waiting,
                                                                                                Num_pending}
                                                                                        )
                                                                                    end
                                                                                )
                                                                            end
                                                                        )
                                                                    end
                                                                )
                                                            end
                                                        )
                                                    end
                                                )
                                            end
                                        )
                                    end
                                )
                            end
                        )
                    end
                )
            end
        )
    end,
    _pipe = Body,
    json_parse_jetstream_errorable(_pipe, Decoder).

-file("src/glats/jetstream/consumer.gleam", 420).
-spec decode_delete(binary()) -> {ok, nil} |
    {error, glats@jetstream:jetstream_error()}.
decode_delete(Body) ->
    _pipe = Body,
    json_parse_jetstream_errorable(
        _pipe,
        begin
            gleam@dynamic@decode:field(
                <<"success"/utf8>>,
                {decoder, fun gleam@dynamic@decode:decode_bool/1},
                fun(_) -> gleam@dynamic@decode:success(nil) end
            )
        end
    ).

-file("src/glats/jetstream/consumer.gleam", 445).
-spec decode_names(binary()) -> {ok, list(binary())} |
    {error, glats@jetstream:jetstream_error()}.
decode_names(Body) ->
    _pipe = Body,
    json_parse_jetstream_errorable(
        _pipe,
        begin
            gleam@dynamic@decode:field(
                <<"consumers"/utf8>>,
                gleam@dynamic@decode:optional(
                    gleam@dynamic@decode:list(
                        {decoder, fun gleam@dynamic@decode:decode_string/1}
                    )
                ),
                fun(Consumers) ->
                    gleam@dynamic@decode:success(
                        begin
                            _pipe@1 = Consumers,
                            gleam@option:unwrap(_pipe@1, [])
                        end
                    )
                end
            )
        end
    ).

-file("src/glats/jetstream/consumer.gleam", 273).
?DOC(" Get info about a consumer by stream and name.\n").
-spec info(
    gleam@erlang@process:subject(glats:connection_message()),
    binary(),
    binary()
) -> {ok, consumer_info()} | {error, glats@jetstream:jetstream_error()}.
info(Conn, Stream, Name) ->
    Topic = <<<<<<<<"$JS.API.CONSUMER"/utf8, ".INFO."/utf8>>/binary,
                Stream/binary>>/binary,
            "."/utf8>>/binary,
        Name/binary>>,
    case glats:request(Conn, Topic, <<""/utf8>>, [], 1000) of
        {ok, Msg} ->
            decode_info(erlang:element(5, Msg));

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

-file("src/glats/jetstream/consumer.gleam", 375).
?DOC(" Creates a new consumer\n").
-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),
    Topic = 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>>;

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

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

-file("src/glats/jetstream/consumer.gleam", 410).
?DOC(" Deletes a consumer\n").
-spec delete(
    gleam@erlang@process:subject(glats:connection_message()),
    binary(),
    binary()
) -> {ok, nil} | {error, glats@jetstream:jetstream_error()}.
delete(Conn, Stream, Name) ->
    Topic = <<<<<<<<"$JS.API.CONSUMER"/utf8, ".DELETE."/utf8>>/binary,
                Stream/binary>>/binary,
            "."/utf8>>/binary,
        Name/binary>>,
    case glats:request(Conn, Topic, <<""/utf8>>, [], 1000) of
        {ok, Msg} ->
            decode_delete(erlang:element(5, Msg));

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

-file("src/glats/jetstream/consumer.gleam", 435).
?DOC(" Get list of consumer names in a stream.\n").
-spec names(gleam@erlang@process:subject(glats:connection_message()), binary()) -> {ok,
        list(binary())} |
    {error, glats@jetstream:jetstream_error()}.
names(Conn, Stream) ->
    Topic = <<<<"$JS.API.CONSUMER"/utf8, ".NAMES."/utf8>>/binary,
        Stream/binary>>,
    case glats:request(Conn, Topic, <<""/utf8>>, [], 1000) of
        {ok, Msg} ->
            decode_names(erlang:element(5, Msg));

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

-file("src/glats/jetstream/consumer.gleam", 485).
-spec do_req_next_msg(
    gleam@erlang@process:subject(glats:connection_message()),
    binary(),
    binary(),
    binary(),
    list(request_batch_option())
) -> {ok, nil} | {error, glats@jetstream:jetstream_error()}.
do_req_next_msg(Conn, Stream, Consumer, Inbox, Opts) ->
    Topic = <<<<<<<<"$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, Topic, maps:new(), {some, Reply_to}, make_req_body(Opts)}
    ),
    gleam@result:map_error(
        _pipe,
        fun(Err) -> {unknown, -1, gleam@string:inspect(Err)} end
    ).

-file("src/glats/jetstream/consumer.gleam", 474).
?DOC(" Request a batch of messages for a pull subscription.\n").
-spec request_batch(subscription(), list(request_batch_option())) -> {ok, nil} |
    {error, glats@jetstream:jetstream_error()}.
request_batch(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.

-file("src/glats/jetstream/consumer.gleam", 646).
-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, Topic, 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, Topic})
    end,
    create(Conn, Stream, Opts@1).

-file("src/glats/jetstream/consumer.gleam", 619).
-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, Topic, 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, Topic, _pipe)
    end.

-file("src/glats/jetstream/consumer.gleam", 545).
?DOC(
    " Subscribe to a topic in a stream.\n"
    "\n"
    " - If no option is provided it will attempt to look up a stream\n"
    "   by the topic and create an ephemeral consumer for the\n"
    "   subscription.\n"
    " - If `Bind(\"stream\", \"consumer\")` is provided it will subsribe\n"
    "   to the stream and existing consumer, failing if either do not\n"
    "   exist.\n"
    " - If `BindStream(\"stream\")` is provided it will not attempt to\n"
    "   lookup the stream by topic but creates an ephemeral consumer\n"
    "   for the subscription.\n"
    "\n"
    " In the cases where an ephemeral consumer will be created\n"
    " `With(ConsumerOption)` can be provided to configure it.\n"
).
-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, Topic, Opts) ->
    gleam@result:then(
        find_stream(Conn, Topic, Opts),
        fun(Stream) ->
            gleam@result:then(
                find_consumer(Conn, Stream, Topic, 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
    ).