goose

-module(goose).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/goose.gleam").
-export([default_config/0, build_url/1, parse_event/1, start_consumer/2]).
-export_type([jetstream_event/0, commit_data/0, identity_data/0, account_data/0, jetstream_config/0, connection_state/0, decompressor/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 jetstream_event() :: {commit_event, binary(), integer(), commit_data()} |
    {identity_event, binary(), integer(), identity_data()} |
    {account_event, binary(), integer(), account_data()} |
    {unknown_event, binary()}.

-type commit_data() :: {commit_data,
        binary(),
        binary(),
        binary(),
        binary(),
        gleam@option:option(gleam@dynamic:dynamic_()),
        gleam@option:option(binary())}.

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

-type account_data() :: {account_data, boolean(), binary(), integer(), binary()}.

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

-type connection_state() :: {connection_state,
        jetstream_config(),
        fun((binary()) -> nil),
        gleam@option:option(decompressor())}.

-type decompressor() :: {decompressor,
        goose@internal@zstd:decompression_context(),
        goose@internal@zstd:decompression_dict()}.

-file("src/goose.gleam", 74).
?DOC(
    " Create a default configuration for US East endpoint\n"
    " Includes automatic retry with exponential backoff (1s, 2s, 4s, 8s, 16s, 32s, capped at 60s)\n"
).
-spec default_config() -> jetstream_config().
default_config() ->
    {jetstream_config,
        <<"wss://jetstream2.us-east.bsky.network/subscribe"/utf8>>,
        [],
        [],
        none,
        none,
        false,
        false}.

-file("src/goose.gleam", 87).
?DOC(" Build the WebSocket URL with query parameters\n").
-spec build_url(jetstream_config()) -> binary().
build_url(Config) ->
    Base = erlang:element(2, Config),
    Mut_params = [],
    Mut_params@1 = case erlang:element(3, Config) of
        [] ->
            Mut_params;

        Collections ->
            Collection_params = gleam@list:map(
                Collections,
                fun(Col) -> <<"wantedCollections="/utf8, Col/binary>> end
            ),
            lists:append(Collection_params, Mut_params)
    end,
    Mut_params@2 = case erlang:element(4, Config) of
        [] ->
            Mut_params@1;

        Dids ->
            Did_params = gleam@list:map(
                Dids,
                fun(Did) -> <<"wantedDids="/utf8, Did/binary>> end
            ),
            lists:append(Did_params, Mut_params@1)
    end,
    Mut_params@3 = case erlang:element(5, Config) of
        none ->
            Mut_params@2;

        {some, Cursor_val} ->
            lists:append(
                [<<"cursor="/utf8, (gleam@string:inspect(Cursor_val))/binary>>],
                Mut_params@2
            )
    end,
    Mut_params@4 = case erlang:element(6, Config) of
        none ->
            Mut_params@3;

        {some, Size_val} ->
            lists:append(
                [<<"maxMessageSizeBytes="/utf8,
                        (gleam@string:inspect(Size_val))/binary>>],
                Mut_params@3
            )
    end,
    Mut_params@5 = case erlang:element(7, Config) of
        false ->
            lists:append([<<"compress=false"/utf8>>], Mut_params@4);

        true ->
            lists:append([<<"compress=true"/utf8>>], Mut_params@4)
    end,
    Mut_params@6 = case erlang:element(8, Config) of
        false ->
            lists:append([<<"requireHello=false"/utf8>>], Mut_params@5);

        true ->
            lists:append([<<"requireHello=true"/utf8>>], Mut_params@5)
    end,
    case Mut_params@6 of
        [] ->
            Base;

        Params ->
            <<<<Base/binary, "?"/utf8>>/binary,
                (gleam@string:join(lists:reverse(Params), <<"&"/utf8>>))/binary>>
    end.

-file("src/goose.gleam", 146).
?DOC(" Load the zstd decompression dictionary\n").
-spec load_decompressor() -> {ok, decompressor()} | {error, binary()}.
load_decompressor() ->
    Priv_dir = goose_ffi:priv_dir(),
    Dict_path = <<Priv_dir/binary, "/zstd_dictionary"/utf8>>,
    gleam@result:'try'(
        begin
            _pipe = simplifile_erl:read_bits(Dict_path),
            gleam@result:map_error(
                _pipe,
                fun(_) ->
                    <<"Failed to load zstd dictionary from "/utf8,
                        Dict_path/binary>>
                end
            )
        end,
        fun(Dict_data) ->
            Dctx = ezstd:create_decompression_context(1024 * 1024),
            Ddict = ezstd:create_ddict(Dict_data),
            gleam@result:'try'(
                goose@internal@zstd:select_ddict(Dctx, Ddict),
                fun(_) -> {ok, {decompressor, Dctx, Ddict}} end
            )
        end
    ).

-file("src/goose.gleam", 174).
?DOC(" Decompress zstd-compressed data\n").
-spec decompress_data(bitstring(), decompressor()) -> {ok, binary()} |
    {error, binary()}.
decompress_data(Data, Decompressor) ->
    case goose_ffi:decompress_using_ddict_safe(
        Data,
        erlang:element(3, Decompressor)
    ) of
        {ok, Decompressed} ->
            _pipe = gleam@bit_array:to_string(Decompressed),
            gleam@result:replace_error(
                _pipe,
                <<"Failed to decode decompressed data as UTF-8"/utf8>>
            );

        {error, Msg} ->
            case gleam_stdlib:contains_string(
                Msg,
                <<"ZSTD_CONTENTSIZE_UNKNOWN"/utf8>>
            ) of
                true ->
                    case goose_ffi:decompress_streaming_safe(
                        erlang:element(2, Decompressor),
                        Data
                    ) of
                        {ok, Decompressed@1} ->
                            _pipe@1 = gleam@bit_array:to_string(Decompressed@1),
                            gleam@result:replace_error(
                                _pipe@1,
                                <<"Failed to decode streaming decompressed data as UTF-8"/utf8>>
                            );

                        {error, _} ->
                            {error, <<"Streaming decompression failed"/utf8>>}
                    end;

                false ->
                    {error, <<"Decompression failed"/utf8>>}
            end
    end.

-file("src/goose.gleam", 292).
?DOC(" Handle incoming WebSocket messages\n").
-spec handle_message(
    connection_state(),
    goose@stratus:message(nil),
    goose@stratus:connection()
) -> goose@stratus:next(connection_state(), nil).
handle_message(State, Msg, _) ->
    case Msg of
        {text, Text} ->
            (erlang:element(3, State))(Text),
            goose@stratus:continue(State);

        {binary, Data} ->
            case erlang:element(4, State) of
                {some, Decompressor} ->
                    case decompress_data(Data, Decompressor) of
                        {ok, Text@1} ->
                            (erlang:element(3, State))(Text@1),
                            goose@stratus:continue(State);

                        {error, _} ->
                            goose@stratus:continue(State)
                    end;

                none ->
                    goose@stratus:continue(State)
            end;

        {user, _} ->
            goose@stratus:continue(State)
    end.

-file("src/goose.gleam", 337).
?DOC(" Calculate exponential backoff capped at 60 seconds\n").
-spec calculate_backoff(integer()) -> integer().
calculate_backoff(Retry_count) ->
    case Retry_count of
        0 ->
            1;

        1 ->
            2;

        2 ->
            4;

        3 ->
            8;

        4 ->
            16;

        5 ->
            32;

        _ ->
            60
    end.

-file("src/goose.gleam", 388).
?DOC(" Decoder for commit with record (create/update operations)\n").
-spec commit_with_record_decoder() -> gleam@dynamic@decode:decoder(commit_data()).
commit_with_record_decoder() ->
    gleam@dynamic@decode:field(
        <<"rev"/utf8>>,
        {decoder, fun gleam@dynamic@decode:decode_string/1},
        fun(Rev) ->
            gleam@dynamic@decode:field(
                <<"operation"/utf8>>,
                {decoder, fun gleam@dynamic@decode:decode_string/1},
                fun(Operation) ->
                    gleam@dynamic@decode:field(
                        <<"collection"/utf8>>,
                        {decoder, fun gleam@dynamic@decode:decode_string/1},
                        fun(Collection) ->
                            gleam@dynamic@decode:field(
                                <<"rkey"/utf8>>,
                                {decoder,
                                    fun gleam@dynamic@decode:decode_string/1},
                                fun(Rkey) ->
                                    gleam@dynamic@decode:field(
                                        <<"record"/utf8>>,
                                        {decoder,
                                            fun gleam@dynamic@decode:decode_dynamic/1},
                                        fun(Record) ->
                                            gleam@dynamic@decode:field(
                                                <<"cid"/utf8>>,
                                                {decoder,
                                                    fun gleam@dynamic@decode:decode_string/1},
                                                fun(Cid) ->
                                                    gleam@dynamic@decode:success(
                                                        {commit_data,
                                                            Rev,
                                                            Operation,
                                                            Collection,
                                                            Rkey,
                                                            {some, Record},
                                                            {some, Cid}}
                                                    )
                                                end
                                            )
                                        end
                                    )
                                end
                            )
                        end
                    )
                end
            )
        end
    ).

-file("src/goose.gleam", 406).
?DOC(" Decoder for commit without record (delete operations)\n").
-spec commit_without_record_decoder() -> gleam@dynamic@decode:decoder(commit_data()).
commit_without_record_decoder() ->
    gleam@dynamic@decode:field(
        <<"rev"/utf8>>,
        {decoder, fun gleam@dynamic@decode:decode_string/1},
        fun(Rev) ->
            gleam@dynamic@decode:field(
                <<"operation"/utf8>>,
                {decoder, fun gleam@dynamic@decode:decode_string/1},
                fun(Operation) ->
                    gleam@dynamic@decode:field(
                        <<"collection"/utf8>>,
                        {decoder, fun gleam@dynamic@decode:decode_string/1},
                        fun(Collection) ->
                            gleam@dynamic@decode:field(
                                <<"rkey"/utf8>>,
                                {decoder,
                                    fun gleam@dynamic@decode:decode_string/1},
                                fun(Rkey) ->
                                    gleam@dynamic@decode:success(
                                        {commit_data,
                                            Rev,
                                            Operation,
                                            Collection,
                                            Rkey,
                                            none,
                                            none}
                                    )
                                end
                            )
                        end
                    )
                end
            )
        end
    ).

-file("src/goose.gleam", 379).
?DOC(" Decoder for commit data - handles both create/update (with record) and delete (without)\n").
-spec commit_data_decoder() -> gleam@dynamic@decode:decoder(commit_data()).
commit_data_decoder() ->
    gleam@dynamic@decode:one_of(
        commit_with_record_decoder(),
        [commit_without_record_decoder()]
    ).

-file("src/goose.gleam", 371).
?DOC(" Decoder for commit events\n").
-spec commit_event_decoder() -> gleam@dynamic@decode:decoder(jetstream_event()).
commit_event_decoder() ->
    gleam@dynamic@decode:field(
        <<"did"/utf8>>,
        {decoder, fun gleam@dynamic@decode:decode_string/1},
        fun(Did) ->
            gleam@dynamic@decode:field(
                <<"time_us"/utf8>>,
                {decoder, fun gleam@dynamic@decode:decode_int/1},
                fun(Time_us) ->
                    gleam@dynamic@decode:field(
                        <<"commit"/utf8>>,
                        commit_data_decoder(),
                        fun(Commit) ->
                            gleam@dynamic@decode:success(
                                {commit_event, Did, Time_us, Commit}
                            )
                        end
                    )
                end
            )
        end
    ).

-file("src/goose.gleam", 430).
?DOC(" Decoder for identity data\n").
-spec identity_data_decoder() -> gleam@dynamic@decode:decoder(identity_data()).
identity_data_decoder() ->
    gleam@dynamic@decode:field(
        <<"did"/utf8>>,
        {decoder, fun gleam@dynamic@decode:decode_string/1},
        fun(Did) ->
            gleam@dynamic@decode:field(
                <<"handle"/utf8>>,
                {decoder, fun gleam@dynamic@decode:decode_string/1},
                fun(Handle) ->
                    gleam@dynamic@decode:field(
                        <<"seq"/utf8>>,
                        {decoder, fun gleam@dynamic@decode:decode_int/1},
                        fun(Seq) ->
                            gleam@dynamic@decode:field(
                                <<"time"/utf8>>,
                                {decoder,
                                    fun gleam@dynamic@decode:decode_string/1},
                                fun(Time) ->
                                    gleam@dynamic@decode:success(
                                        {identity_data, Did, Handle, Seq, Time}
                                    )
                                end
                            )
                        end
                    )
                end
            )
        end
    ).

-file("src/goose.gleam", 422).
?DOC(" Decoder for identity events\n").
-spec identity_event_decoder() -> gleam@dynamic@decode:decoder(jetstream_event()).
identity_event_decoder() ->
    gleam@dynamic@decode:field(
        <<"did"/utf8>>,
        {decoder, fun gleam@dynamic@decode:decode_string/1},
        fun(Did) ->
            gleam@dynamic@decode:field(
                <<"time_us"/utf8>>,
                {decoder, fun gleam@dynamic@decode:decode_int/1},
                fun(Time_us) ->
                    gleam@dynamic@decode:field(
                        <<"identity"/utf8>>,
                        identity_data_decoder(),
                        fun(Identity) ->
                            gleam@dynamic@decode:success(
                                {identity_event, Did, Time_us, Identity}
                            )
                        end
                    )
                end
            )
        end
    ).

-file("src/goose.gleam", 447).
?DOC(" Decoder for account data\n").
-spec account_data_decoder() -> gleam@dynamic@decode:decoder(account_data()).
account_data_decoder() ->
    gleam@dynamic@decode:field(
        <<"active"/utf8>>,
        {decoder, fun gleam@dynamic@decode:decode_bool/1},
        fun(Active) ->
            gleam@dynamic@decode:field(
                <<"did"/utf8>>,
                {decoder, fun gleam@dynamic@decode:decode_string/1},
                fun(Did) ->
                    gleam@dynamic@decode:field(
                        <<"seq"/utf8>>,
                        {decoder, fun gleam@dynamic@decode:decode_int/1},
                        fun(Seq) ->
                            gleam@dynamic@decode:field(
                                <<"time"/utf8>>,
                                {decoder,
                                    fun gleam@dynamic@decode:decode_string/1},
                                fun(Time) ->
                                    gleam@dynamic@decode:success(
                                        {account_data, Active, Did, Seq, Time}
                                    )
                                end
                            )
                        end
                    )
                end
            )
        end
    ).

-file("src/goose.gleam", 439).
?DOC(" Decoder for account events\n").
-spec account_event_decoder() -> gleam@dynamic@decode:decoder(jetstream_event()).
account_event_decoder() ->
    gleam@dynamic@decode:field(
        <<"did"/utf8>>,
        {decoder, fun gleam@dynamic@decode:decode_string/1},
        fun(Did) ->
            gleam@dynamic@decode:field(
                <<"time_us"/utf8>>,
                {decoder, fun gleam@dynamic@decode:decode_int/1},
                fun(Time_us) ->
                    gleam@dynamic@decode:field(
                        <<"account"/utf8>>,
                        account_data_decoder(),
                        fun(Account) ->
                            gleam@dynamic@decode:success(
                                {account_event, Did, Time_us, Account}
                            )
                        end
                    )
                end
            )
        end
    ).

-file("src/goose.gleam", 350).
?DOC(" Parse a JSON event string into a JetstreamEvent\n").
-spec parse_event(binary()) -> jetstream_event().
parse_event(Json_string) ->
    case gleam@json:parse(Json_string, commit_event_decoder()) of
        {ok, Event} ->
            Event;

        {error, _} ->
            case gleam@json:parse(Json_string, identity_event_decoder()) of
                {ok, Event@1} ->
                    Event@1;

                {error, _} ->
                    case gleam@json:parse(Json_string, account_event_decoder()) of
                        {ok, Event@2} ->
                            Event@2;

                        {error, _} ->
                            {unknown_event, Json_string}
                    end
            end
    end.

-file("src/goose.gleam", 331).
?DOC(" Handle WebSocket connection close\n").
-spec handle_close(connection_state()) -> nil.
handle_close(State) ->
    start_with_retry_internal(
        erlang:element(2, State),
        erlang:element(3, State),
        0
    ).

-file("src/goose.gleam", 228).
?DOC(" Internal function to handle connection with retry\n").
-spec start_with_retry_internal(
    jetstream_config(),
    fun((binary()) -> nil),
    integer()
) -> nil.
start_with_retry_internal(Config, On_event, Retry_count) ->
    Url = build_url(Config),
    Http_url = begin
        _pipe = Url,
        _pipe@1 = gleam@string:replace(
            _pipe,
            <<"wss://"/utf8>>,
            <<"https://"/utf8>>
        ),
        gleam@string:replace(_pipe@1, <<"ws://"/utf8>>, <<"http://"/utf8>>)
    end,
    Req@1 = case gleam@http@request:to(Http_url) of
        {ok, Req} -> Req;
        _assert_fail ->
            erlang:error(#{gleam_error => let_assert,
                        message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
                        file => <<?FILEPATH/utf8>>,
                        module => <<"goose"/utf8>>,
                        function => <<"start_with_retry_internal"/utf8>>,
                        line => 242,
                        value => _assert_fail,
                        start => 6891,
                        'end' => 6932,
                        pattern_start => 6902,
                        pattern_end => 6909})
    end,
    Decompressor = case erlang:element(7, Config) of
        true ->
            case load_decompressor() of
                {ok, Dec} ->
                    {some, Dec};

                {error, _} ->
                    none
            end;

        false ->
            none
    end,
    State = {connection_state, Config, On_event, Decompressor},
    Result = begin
        _pipe@2 = goose@stratus:new_with_initialiser(
            Req@1,
            fun() -> {ok, goose@stratus:initialised(State)} end
        ),
        _pipe@3 = goose@stratus:on_message(_pipe@2, fun handle_message/3),
        _pipe@4 = goose@stratus:on_close(_pipe@3, fun handle_close/1),
        _pipe@5 = goose@stratus:with_connect_timeout(_pipe@4, 30000),
        goose@stratus:start(_pipe@5)
    end,
    case Result of
        {ok, _} ->
            gleam_erlang_ffi:sleep_forever();

        {error, _} ->
            Backoff_seconds = calculate_backoff(Retry_count),
            gleam_erlang_ffi:sleep(Backoff_seconds * 1000),
            start_with_retry_internal(Config, On_event, Retry_count + 1)
    end.

-file("src/goose.gleam", 220).
?DOC(
    " Start consuming the Jetstream feed with automatic retry logic\n"
    "\n"
    " Handles connection failures gracefully with exponential backoff and automatic reconnection.\n"
    " The retry behavior is configured through the JetstreamConfig fields:\n"
    " - max_backoff_seconds: Maximum wait time between retries\n"
    " - log_connection_events: Log connects/disconnects\n"
    " - log_retry_attempts: Log retry attempts and errors\n"
    "\n"
    " Example:\n"
    " ```gleam\n"
    " let config = goose.default_config()\n"
    "\n"
    " goose.start_consumer(config, fn(event_json) {\n"
    "   // Handle event\n"
    "   io.println(event_json)\n"
    " })\n"
    " ```\n"
).
-spec start_consumer(jetstream_config(), fun((binary()) -> nil)) -> nil.
start_consumer(Config, On_event) ->
    start_with_retry_internal(Config, On_event, 0).