glats

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

-export([info/2, create/4, update/3, delete/2, purge/2, find_stream_name_by_subject/2, get_message/3]).
-export_type([retention_policy/0, discard_policy/0, stream_option/0, stream_info/0, stream_config/0, stream_state/0, stream_message/0, access_method/0, raw_stream_message/0]).

-type retention_policy() :: limits_policy | interest_policy | work_queue_policy.

-type discard_policy() :: discard_old | discard_new.

-type stream_option() :: {description, binary()} |
    {retention, retention_policy()} |
    {max_consumers, integer()} |
    {max_messages, integer()} |
    {max_bytes, integer()} |
    {max_age, integer()} |
    {max_messages_per_subject, integer()} |
    {max_message_size, integer()} |
    {discard, discard_policy()} |
    {storage, glats@jetstream:storage_type()} |
    {num_replicas, integer()} |
    {duplicate_window, integer()} |
    {allow_direct, boolean()} |
    {mirror_direct, boolean()} |
    {deny_delete, boolean()} |
    {deny_purge, boolean()} |
    {allow_rollup, boolean()} |
    {discard_new_per_subject, boolean()}.

-type stream_info() :: {stream_info, binary(), stream_config(), stream_state()}.

-type stream_config() :: {stream_config,
        binary(),
        list(binary()),
        gleam@option:option(binary()),
        gleam@option:option(integer()),
        gleam@option:option(integer()),
        gleam@option:option(integer()),
        gleam@option:option(integer()),
        gleam@option:option(integer()),
        gleam@option:option(integer()),
        gleam@option:option(binary()),
        gleam@option:option(binary()),
        gleam@option:option(integer()),
        gleam@option:option(integer()),
        gleam@option:option(boolean()),
        gleam@option:option(boolean()),
        gleam@option:option(boolean()),
        gleam@option:option(boolean()),
        gleam@option:option(boolean()),
        gleam@option:option(boolean())}.

-type stream_state() :: {stream_state,
        integer(),
        integer(),
        integer(),
        binary(),
        integer(),
        binary(),
        integer()}.

-type stream_message() :: {stream_message, integer(), binary(), glats:message()}.

-type access_method() :: {sequence_id, integer()} | {last_by_subject, binary()}.

-type raw_stream_message() :: {raw_stream_message,
        binary(),
        integer(),
        gleam@option:option(binary()),
        binary(),
        binary()}.

-spec encode_get_message_body(access_method()) -> binary().
encode_get_message_body(Method) ->
    _pipe = case Method of
        {sequence_id, Seq} ->
            [{<<"seq"/utf8>>, gleam@json:int(Seq)}];

        {last_by_subject, Subj} ->
            [{<<"last_by_subj"/utf8>>, gleam@json:string(Subj)}]
    end,
    _pipe@1 = gleam@json:object(_pipe),
    gleam@json:to_string(_pipe@1).

-spec raw_to_stream_message(raw_stream_message()) -> {ok, stream_message()} |
    {error, nil}.
raw_to_stream_message(Msg) ->
    gleam@result:then(
        begin
            _pipe = gleam@base:decode64(erlang:element(5, Msg)),
            gleam@result:map(_pipe, fun gleam@bit_string:to_string/1)
        end,
        fun(Body) ->
            Hdrs = case erlang:element(4, Msg) of
                {some, Data} ->
                    _pipe@1 = glats@internal@js:decode_b64_headers(Data),
                    gleam@result:unwrap(_pipe@1, gleam@map:new());

                none ->
                    gleam@map:new()
            end,
            {ok,
                {stream_message,
                    erlang:element(3, Msg),
                    erlang:element(6, Msg),
                    {message,
                        erlang:element(2, Msg),
                        Hdrs,
                        none,
                        begin
                            _pipe@2 = Body,
                            gleam@result:unwrap(_pipe@2, <<""/utf8>>)
                        end}}}
        end
    ).

-spec ret_pol_to_string(retention_policy()) -> binary().
ret_pol_to_string(Pol) ->
    case Pol of
        limits_policy ->
            <<"limits"/utf8>>;

        interest_policy ->
            <<"interest"/utf8>>;

        work_queue_policy ->
            <<"workqueue"/utf8>>
    end.

-spec dis_pol_to_string(discard_policy()) -> binary().
dis_pol_to_string(Pol) ->
    case Pol of
        discard_old ->
            <<"old"/utf8>>;

        discard_new ->
            <<"new"/utf8>>
    end.

-spec storage_to_string(glats@jetstream:storage_type()) -> binary().
storage_to_string(Storage) ->
    case Storage of
        file_storage ->
            <<"file"/utf8>>;

        memory_storage ->
            <<"memory"/utf8>>
    end.

-spec apply_stream_option(list({binary(), gleam@json:json()}), stream_option()) -> list({binary(),
    gleam@json:json()}).
apply_stream_option(Prev, Opt) ->
    Pair = case Opt of
        {description, Desc} ->
            {<<"description"/utf8>>, gleam@json:string(Desc)};

        {retention, Pol} ->
            {<<"retention"/utf8>>,
                begin
                    _pipe = ret_pol_to_string(Pol),
                    gleam@json:string(_pipe)
                end};

        {discard, Pol@1} ->
            {<<"discard"/utf8>>,
                begin
                    _pipe@1 = dis_pol_to_string(Pol@1),
                    gleam@json:string(_pipe@1)
                end};

        {storage, Storage} ->
            {<<"storage"/utf8>>,
                begin
                    _pipe@2 = storage_to_string(Storage),
                    gleam@json:string(_pipe@2)
                end};

        {max_consumers, Num} ->
            {<<"max_consumers"/utf8>>, gleam@json:int(Num)};

        {max_messages, Num@1} ->
            {<<"max_msgs"/utf8>>, gleam@json:int(Num@1)};

        {max_bytes, Num@2} ->
            {<<"max_bytes"/utf8>>, gleam@json:int(Num@2)};

        {max_age, Num@3} ->
            {<<"max_age"/utf8>>, gleam@json:int(Num@3)};

        {max_messages_per_subject, Num@4} ->
            {<<"max_msgs_per_subject"/utf8>>, gleam@json:int(Num@4)};

        {max_message_size, Num@5} ->
            {<<"max_msg_size"/utf8>>, gleam@json:int(Num@5)};

        {num_replicas, Num@6} ->
            {<<"num_replicas"/utf8>>, gleam@json:int(Num@6)};

        {duplicate_window, Num@7} ->
            {<<"duplicate_window"/utf8>>, gleam@json:int(Num@7)};

        {allow_direct, Val} ->
            {<<"allow_direct"/utf8>>, gleam@json:bool(Val)};

        {mirror_direct, Val@1} ->
            {<<"mirror_direct"/utf8>>, gleam@json:bool(Val@1)};

        {deny_delete, Val@2} ->
            {<<"deny_delete"/utf8>>, gleam@json:bool(Val@2)};

        {deny_purge, Val@3} ->
            {<<"deny_purge"/utf8>>, gleam@json:bool(Val@3)};

        {allow_rollup, Val@4} ->
            {<<"allow_rollup_hdrs"/utf8>>, gleam@json:bool(Val@4)};

        {discard_new_per_subject, Val@5} ->
            {<<"discard_new_per_subject"/utf8>>, gleam@json:bool(Val@5)}
    end,
    gleam@list:prepend(Prev, Pair).

-spec apply_stream_options(
    list({binary(), gleam@json:json()}),
    list(stream_option())
) -> list({binary(), gleam@json:json()}).
apply_stream_options(Prev, Opts) ->
    gleam@list:fold(Opts, Prev, fun apply_stream_option/2).

-spec stream_options_to_json(
    list({binary(), gleam@json:json()}),
    list(stream_option())
) -> binary().
stream_options_to_json(Prev, Opts) ->
    _pipe = Prev,
    _pipe@1 = apply_stream_options(_pipe, Opts),
    _pipe@2 = gleam@json:object(_pipe@1),
    gleam@json:to_string(_pipe@2).

-spec decode_info(binary()) -> {ok, stream_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_stream_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()) -> {ok,
        stream_info()} |
    {error, glats@jetstream:jetstream_error()}.
info(Conn, Name) ->
    Topic = <<<<"$JS.API.STREAM"/utf8, ".INFO."/utf8>>/binary, Name/binary>>,
    case glats:request(Conn, Topic, <<""/utf8>>, [], 1000) of
        {ok, Msg} ->
            decode_info(erlang:element(5, Msg));

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

-spec create(
    gleam@erlang@process:subject(glats:connection_message()),
    binary(),
    list(binary()),
    list(stream_option())
) -> {ok, stream_info()} | {error, glats@jetstream:jetstream_error()}.
create(Conn, Name, Subjects, Opts) ->
    Topic = <<<<"$JS.API.STREAM"/utf8, ".CREATE."/utf8>>/binary, Name/binary>>,
    Body = begin
        _pipe = [{<<"name"/utf8>>, gleam@json:string(Name)},
            {<<"subjects"/utf8>>,
                gleam@json:array(Subjects, fun gleam@json:string/1)}],
        stream_options_to_json(_pipe, Opts)
    end,
    case glats:request(Conn, Topic, Body, [], 1000) of
        {ok, Msg} ->
            decode_info(erlang:element(5, Msg));

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

-spec update(
    gleam@erlang@process:subject(glats:connection_message()),
    binary(),
    list(stream_option())
) -> {ok, stream_info()} | {error, glats@jetstream:jetstream_error()}.
update(Conn, Name, Opts) ->
    Topic = <<<<"$JS.API.STREAM"/utf8, ".UPDATE."/utf8>>/binary, Name/binary>>,
    Body = begin
        _pipe = [{<<"name"/utf8>>, gleam@json:string(Name)}],
        stream_options_to_json(_pipe, Opts)
    end,
    case glats:request(Conn, Topic, Body, [], 1000) of
        {ok, Msg} ->
            decode_info(erlang:element(5, Msg));

        {error, _} ->
            {error, {stream_not_found, <<""/utf8>>}}
    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_stream_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()) -> {ok,
        nil} |
    {error, glats@jetstream:jetstream_error()}.
delete(Conn, Name) ->
    Topic = <<<<"$JS.API.STREAM"/utf8, ".DELETE."/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.

-spec decode_purge(binary()) -> {ok, integer()} |
    {error, glats@jetstream:jetstream_error()}.
decode_purge(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_stream_purge_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 purge(gleam@erlang@process:subject(glats:connection_message()), binary()) -> {ok,
        integer()} |
    {error, glats@jetstream:jetstream_error()}.
purge(Conn, Name) ->
    Topic = <<<<"$JS.API.STREAM"/utf8, ".PURGE."/utf8>>/binary, Name/binary>>,
    case glats:request(Conn, Topic, <<""/utf8>>, [], 1000) of
        {ok, Msg} ->
            decode_purge(erlang:element(5, Msg));

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

-spec decode_names(binary()) -> {ok, list(binary())} |
    {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_stream_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 find_stream_name_by_subject(
    gleam@erlang@process:subject(glats:connection_message()),
    binary()
) -> {ok, binary()} | {error, glats@jetstream:jetstream_error()}.
find_stream_name_by_subject(Conn, Subject) ->
    Topic = <<"$JS.API.STREAM"/utf8, ".NAMES"/utf8>>,
    Body = begin
        _pipe = [{<<"subject"/utf8>>, gleam@json:string(Subject)}],
        _pipe@1 = gleam@json:object(_pipe),
        gleam@json:to_string(_pipe@1)
    end,
    case glats:request(Conn, Topic, Body, [], 1000) of
        {ok, Msg} ->
            case decode_names(erlang:element(5, Msg)) of
                {ok, Names} ->
                    _pipe@2 = gleam@list:first(Names),
                    gleam@result:map_error(
                        _pipe@2,
                        fun(_) ->
                            {stream_not_found, <<"stream not found"/utf8>>}
                        end
                    );

                {error, Err} ->
                    {error, Err}
            end;

        {error, Err@1} ->
            _pipe@3 = glats@internal@js:map_glats_error_to_jetstream(Err@1),
            {error, _pipe@3}
    end.

-spec decode_raw_message(binary()) -> {ok, raw_stream_message()} |
    {error, glats@jetstream:jetstream_error()}.
decode_raw_message(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_raw_stream_message_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 get_message(
    gleam@erlang@process:subject(glats:connection_message()),
    binary(),
    access_method()
) -> {ok, stream_message()} | {error, glats@jetstream:jetstream_error()}.
get_message(Conn, Stream, Method) ->
    Topic = <<<<"$JS.API.STREAM"/utf8, ".MSG.GET."/utf8>>/binary,
        Stream/binary>>,
    Body = encode_get_message_body(Method),
    case glats:request(Conn, Topic, Body, [], 1000) of
        {ok, Msg} ->
            _pipe = decode_raw_message(erlang:element(5, Msg)),
            gleam@result:then(
                _pipe,
                fun(D) -> _pipe@1 = raw_to_stream_message(D),
                    gleam@result:map_error(
                        _pipe@1,
                        fun(_) -> {decode_error, erlang:element(5, Msg)} end
                    ) end
            );

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