anthropic_gleam

-module(anthropic@streaming@handler).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/anthropic/streaming/handler.gleam").
-export([new_streaming_state/0, process_chunk/2, finalize_stream/1, is_stream_complete/1, has_stream_error/1, get_stream_error/1, get_accumulated_events/1, build_stream_result/1, parse_sse_body/1, parse_streaming_response/1, parse_sse_chunk/2, stream_message/2, stream_message_with_callback/3, get_event_text/1, get_text_deltas/1, get_full_text/1, get_message_id/1, get_model/1, is_complete/1, has_error/1, get_error/1]).
-export_type([streaming_state/0, stream_result/0, stream_error/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.

?MODULEDOC(
    " Streaming handler for Anthropic API\n"
    "\n"
    " This module provides sans-io functions for parsing Server-Sent Events (SSE)\n"
    " from the Anthropic streaming API. The design follows the sans-io pattern,\n"
    " allowing you to use any HTTP client that supports streaming.\n"
    "\n"
    " ## Recommended: Use `api.chat_stream`\n"
    "\n"
    " For most use cases, use `api.chat_stream` from the main API module:\n"
    "\n"
    " ```gleam\n"
    " import anthropic/api\n"
    " import anthropic/client\n"
    " import anthropic/types/request\n"
    " import anthropic/types/message.{user_message}\n"
    "\n"
    " let assert Ok(client) = client.init()\n"
    " let req = request.new(\"claude-sonnet-4-20250514\", [user_message(\"Hello!\")], 1024)\n"
    "\n"
    " case api.chat_stream(client, req) {\n"
    "   Ok(result) -> io.println(api.stream_text(result))\n"
    "   Error(err) -> handle_error(err)\n"
    " }\n"
    " ```\n"
    "\n"
    " ## True Real-Time Streaming (Sans-IO)\n"
    "\n"
    " For true real-time streaming where you process events as they arrive,\n"
    " use the incremental parsing functions with your own streaming HTTP client:\n"
    "\n"
    " ```gleam\n"
    " import anthropic/http\n"
    " import anthropic/streaming/handler.{\n"
    "   new_streaming_state, process_chunk, finalize_stream\n"
    " }\n"
    "\n"
    " // 1. Build the streaming request\n"
    " let http_request = http.build_streaming_request(api_key, base_url, request)\n"
    "\n"
    " // 2. Start streaming with your HTTP client that supports chunked responses\n"
    " // (e.g., using Erlang's httpc with stream option, or any other client)\n"
    "\n"
    " // 3. Initialize streaming state\n"
    " let state = new_streaming_state()\n"
    "\n"
    " // 4. As each chunk arrives from your HTTP client, process it:\n"
    " let #(events, new_state) = process_chunk(state, chunk)\n"
    "\n"
    " // 5. Handle events in real-time as they arrive\n"
    " list.each(events, fn(event) {\n"
    "   case handler.get_event_text(event) {\n"
    "     Ok(text) -> io.print(text)  // Print immediately!\n"
    "     Error(_) -> Nil\n"
    "   }\n"
    " })\n"
    "\n"
    " // 6. Continue with new_state for next chunk...\n"
    "\n"
    " // 7. When stream ends, finalize to get any remaining events\n"
    " let final_events = finalize_stream(state)\n"
    " ```\n"
    "\n"
    " ## Batch Mode (Deprecated)\n"
    "\n"
    " The batch functions `stream_message` and `stream_message_with_callback` in\n"
    " this module are deprecated. Use `api.chat_stream` and\n"
    " `api.chat_stream_with_callback` instead for a unified API.\n"
    "\n"
    " **Note**: Batch mode waits for the complete response before returning.\n"
    " Use sans-io incremental parsing for true real-time streaming.\n"
).

-type streaming_state() :: {streaming_state,
        anthropic@internal@sse:sse_parser_state(),
        list(anthropic@streaming:stream_event()),
        boolean(),
        gleam@option:option(stream_error())}.

-type stream_result() :: {stream_result,
        list(anthropic@streaming:stream_event())}.

-type stream_error() :: {http_error, anthropic@error:anthropic_error()} |
    {sse_parse_error, binary()} |
    {event_decode_error, binary()} |
    {api_error, integer(), binary()}.

-file("src/anthropic/streaming/handler.gleam", 151).
?DOC(
    " Create a new streaming state for incremental parsing\n"
    "\n"
    " Use this when implementing true real-time streaming with your own HTTP client.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " let state = new_streaming_state()\n"
    "\n"
    " // As chunks arrive from your streaming HTTP client:\n"
    " let #(events, new_state) = process_chunk(state, chunk)\n"
    " list.each(events, handle_event_immediately)\n"
    " ```\n"
).
-spec new_streaming_state() -> streaming_state().
new_streaming_state() ->
    {streaming_state,
        anthropic@internal@sse:new_parser_state(),
        [],
        false,
        none}.

-file("src/anthropic/streaming/handler.gleam", 187).
?DOC(
    " Process a chunk of SSE data and return parsed events\n"
    "\n"
    " This is the core function for real-time streaming. Call it each time\n"
    " you receive a chunk from your streaming HTTP client.\n"
    "\n"
    " Returns a tuple of:\n"
    " - List of events parsed from this chunk (process these immediately!)\n"
    " - Updated state to use for the next chunk\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " // In your streaming HTTP callback:\n"
    " fn on_chunk(state: StreamingState, chunk: String) -> StreamingState {\n"
    "   let #(events, new_state) = process_chunk(state, chunk)\n"
    "\n"
    "   // Process events immediately as they arrive\n"
    "   list.each(events, fn(event) {\n"
    "     case get_event_text(event) {\n"
    "       Ok(text) -> io.print(text)  // Real-time output!\n"
    "       Error(_) -> Nil\n"
    "     }\n"
    "   })\n"
    "\n"
    "   new_state\n"
    " }\n"
    " ```\n"
).
-spec process_chunk(streaming_state(), binary()) -> {list(anthropic@streaming:stream_event()),
    streaming_state()}.
process_chunk(State, Chunk) ->
    Parse_result = anthropic@internal@sse:parse_chunk(
        erlang:element(2, State),
        Chunk
    ),
    Events = begin
        _pipe = erlang:element(2, Parse_result),
        gleam@list:filter_map(
            _pipe,
            fun(Sse_event) ->
                case anthropic@streaming@decoder:decode_event(Sse_event) of
                    {ok, Event} ->
                        {ok, Event};

                    {error, _} ->
                        {error, nil}
                end
            end
        )
    end,
    Completed = gleam@list:any(Events, fun(Event@1) -> case Event@1 of
                message_stop_event ->
                    true;

                _ ->
                    false
            end end),
    Error = begin
        _pipe@1 = gleam@list:find_map(Events, fun(Event@2) -> case Event@2 of
                    {error_event, Err} ->
                        {ok,
                            {sse_parse_error,
                                <<<<(erlang:element(2, Err))/binary, ": "/utf8>>/binary,
                                    (erlang:element(3, Err))/binary>>}};

                    _ ->
                        {error, nil}
                end end),
        gleam@option:from_result(_pipe@1)
    end,
    New_state = {streaming_state,
        erlang:element(3, Parse_result),
        lists:append(erlang:element(3, State), Events),
        Completed,
        Error},
    {Events, New_state}.

-file("src/anthropic/streaming/handler.gleam", 237).
?DOC(
    " Finalize the stream and get any remaining events\n"
    "\n"
    " Call this when your HTTP client signals the stream has ended.\n"
    " Returns any events that were buffered but not yet emitted.\n"
).
-spec finalize_stream(streaming_state()) -> list(anthropic@streaming:stream_event()).
finalize_stream(State) ->
    case anthropic@internal@sse:flush(erlang:element(2, State)) of
        {ok, Sse_event} ->
            case anthropic@streaming@decoder:decode_event(Sse_event) of
                {ok, Event} ->
                    [Event];

                {error, _} ->
                    []
            end;

        {error, _} ->
            []
    end.

-file("src/anthropic/streaming/handler.gleam", 250).
?DOC(" Check if the stream has completed successfully\n").
-spec is_stream_complete(streaming_state()) -> boolean().
is_stream_complete(State) ->
    erlang:element(4, State).

-file("src/anthropic/streaming/handler.gleam", 255).
?DOC(" Check if the stream encountered an error\n").
-spec has_stream_error(streaming_state()) -> boolean().
has_stream_error(State) ->
    gleam@option:is_some(erlang:element(5, State)).

-file("src/anthropic/streaming/handler.gleam", 260).
?DOC(" Get the error from the stream if one occurred\n").
-spec get_stream_error(streaming_state()) -> gleam@option:option(stream_error()).
get_stream_error(State) ->
    erlang:element(5, State).

-file("src/anthropic/streaming/handler.gleam", 265).
?DOC(" Get all accumulated events from the streaming state\n").
-spec get_accumulated_events(streaming_state()) -> list(anthropic@streaming:stream_event()).
get_accumulated_events(State) ->
    erlang:element(3, State).

-file("src/anthropic/streaming/handler.gleam", 270).
?DOC(" Build a StreamResult from the final streaming state\n").
-spec build_stream_result(streaming_state()) -> stream_result().
build_stream_result(State) ->
    {stream_result, erlang:element(3, State)}.

-file("src/anthropic/streaming/handler.gleam", 305).
?DOC(
    " Parse SSE body text into streaming events (batch mode)\n"
    "\n"
    " Use this for batch parsing when you have the complete SSE text.\n"
    " For real-time streaming, use `process_chunk` instead.\n"
).
-spec parse_sse_body(binary()) -> {ok, stream_result()} |
    {error, stream_error()}.
parse_sse_body(Body) ->
    State = new_streaming_state(),
    {Events, Final_state} = process_chunk(State, Body),
    Remaining = finalize_stream(Final_state),
    {ok, {stream_result, lists:append(Events, Remaining)}}.

-file("src/anthropic/streaming/handler.gleam", 292).
?DOC(
    " Parse an HTTP response body containing SSE events into StreamResult\n"
    "\n"
    " This is for batch parsing of a complete response. For real-time streaming,\n"
    " use `new_streaming_state` + `process_chunk` instead.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " let http_response = HttpResponse(status: 200, headers: [], body: sse_body)\n"
    " case parse_streaming_response(http_response) {\n"
    "   Ok(result) -> get_full_text(result.events)\n"
    "   Error(err) -> handle_error(err)\n"
    " }\n"
    " ```\n"
).
-spec parse_streaming_response(anthropic@http:http_response()) -> {ok,
        stream_result()} |
    {error, stream_error()}.
parse_streaming_response(Response) ->
    case erlang:element(2, Response) of
        200 ->
            parse_sse_body(erlang:element(4, Response));

        Status ->
            {error, {api_error, Status, erlang:element(4, Response)}}
    end.

-file("src/anthropic/streaming/handler.gleam", 316).
?DOC(
    " Parse a single SSE chunk and return events plus updated SSE state\n"
    "\n"
    " Lower-level function that works directly with SSE parser state.\n"
    " Consider using `process_chunk` with `StreamingState` for a higher-level API.\n"
).
-spec parse_sse_chunk(anthropic@internal@sse:sse_parser_state(), binary()) -> {list(anthropic@streaming:stream_event()),
    anthropic@internal@sse:sse_parser_state()}.
parse_sse_chunk(State, Chunk) ->
    Parse_result = anthropic@internal@sse:parse_chunk(State, Chunk),
    Events = begin
        _pipe = erlang:element(2, Parse_result),
        gleam@list:filter_map(
            _pipe,
            fun(Sse_event) ->
                case anthropic@streaming@decoder:decode_event(Sse_event) of
                    {ok, Event} ->
                        {ok, Event};

                    {error, _} ->
                        {error, nil}
                end
            end
        )
    end,
    {Events, erlang:element(3, Parse_result)}.

-file("src/anthropic/streaming/handler.gleam", 469).
?DOC(" Convert ConnectError to string\n").
-spec connect_error_to_string(gleam@httpc:connect_error()) -> binary().
connect_error_to_string(Err) ->
    case Err of
        {posix, Code} ->
            <<"POSIX error: "/utf8, Code/binary>>;

        {tls_alert, Code@1, Detail} ->
            <<<<<<"TLS alert "/utf8, Code@1/binary>>/binary, ": "/utf8>>/binary,
                Detail/binary>>
    end.

-file("src/anthropic/streaming/handler.gleam", 453).
?DOC(" Convert httpc error to AnthropicError\n").
-spec http_error_to_anthropic_error(gleam@httpc:http_error()) -> anthropic@error:anthropic_error().
http_error_to_anthropic_error(Err) ->
    case Err of
        invalid_utf8_response ->
            anthropic@error:http_error(<<"Invalid UTF-8 in response"/utf8>>);

        {failed_to_connect, Ip4, Ip6} ->
            anthropic@error:network_error(
                <<<<<<<<"Failed to connect to server (IPv4: "/utf8,
                                (connect_error_to_string(Ip4))/binary>>/binary,
                            ", IPv6: "/utf8>>/binary,
                        (connect_error_to_string(Ip6))/binary>>/binary,
                    ")"/utf8>>
            );

        response_timeout ->
            anthropic@error:timeout_error(0)
    end.

-file("src/anthropic/streaming/handler.gleam", 417).
?DOC(" Make a streaming HTTP request using gleam_httpc\n").
-spec make_streaming_request(anthropic@client:client(), binary()) -> {ok,
        gleam@http@response:response(binary())} |
    {error, anthropic@error:anthropic_error()}.
make_streaming_request(Api_client, Body) ->
    Base_url = erlang:element(3, erlang:element(2, Api_client)),
    Full_url = <<Base_url/binary, (<<"/v1/messages"/utf8>>)/binary>>,
    gleam@result:'try'(
        begin
            _pipe = gleam@http@request:to(Full_url),
            gleam@result:map_error(
                _pipe,
                fun(_) ->
                    anthropic@error:config_error(
                        <<"Invalid URL: "/utf8, Full_url/binary>>
                    )
                end
            )
        end,
        fun(Req) ->
            Req@1 = begin
                _pipe@1 = Req,
                _pipe@2 = gleam@http@request:set_method(_pipe@1, post),
                _pipe@3 = gleam@http@request:set_header(
                    _pipe@2,
                    <<"content-type"/utf8>>,
                    <<"application/json"/utf8>>
                ),
                _pipe@4 = gleam@http@request:set_header(
                    _pipe@3,
                    <<"x-api-key"/utf8>>,
                    anthropic@config:api_key_to_string(
                        erlang:element(2, erlang:element(2, Api_client))
                    )
                ),
                _pipe@5 = gleam@http@request:set_header(
                    _pipe@4,
                    <<"anthropic-version"/utf8>>,
                    <<"2023-06-01"/utf8>>
                ),
                _pipe@6 = gleam@http@request:set_header(
                    _pipe@5,
                    <<"accept"/utf8>>,
                    <<"text/event-stream"/utf8>>
                ),
                gleam@http@request:set_body(_pipe@6, Body)
            end,
            _pipe@7 = gleam@httpc:configure(),
            _pipe@8 = gleam@httpc:timeout(
                _pipe@7,
                erlang:element(5, erlang:element(2, Api_client))
            ),
            _pipe@9 = gleam@httpc:dispatch(_pipe@8, Req@1),
            gleam@result:map_error(
                _pipe@9,
                fun(Err) -> http_error_to_anthropic_error(Err) end
            )
        end
    ).

-file("src/anthropic/streaming/handler.gleam", 356).
?DOC(
    " Stream a message request and return all events (batch mode)\n"
    "\n"
    " @deprecated Use `api.chat_stream` instead for a unified streaming API\n"
    "\n"
    " **Note**: This function collects ALL events before returning. It does NOT\n"
    " provide true real-time streaming. For real-time streaming, use the sans-io\n"
    " functions (`new_streaming_state`, `process_chunk`) with a streaming HTTP client.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " // Prefer using api.chat_stream instead:\n"
    " case api.chat_stream(client, request) {\n"
    "   Ok(result) -> api.stream_text(result)\n"
    "   Error(err) -> handle_error(err)\n"
    " }\n"
    " ```\n"
).
-spec stream_message(
    anthropic@client:client(),
    anthropic@request:create_message_request()
) -> {ok, stream_result()} | {error, stream_error()}.
stream_message(Api_client, Message_request) ->
    Streaming_request = anthropic@request:with_stream(Message_request, true),
    Body = anthropic@request:request_to_json_string(Streaming_request),
    gleam@result:'try'(
        begin
            _pipe = make_streaming_request(Api_client, Body),
            gleam@result:map_error(_pipe, fun(Err) -> {http_error, Err} end)
        end,
        fun(Http_response) -> case erlang:element(2, Http_response) of
                200 ->
                    parse_sse_body(erlang:element(4, Http_response));

                Status ->
                    {error,
                        {api_error, Status, erlang:element(4, Http_response)}}
            end end
    ).

-file("src/anthropic/streaming/handler.gleam", 399).
?DOC(
    " Stream a message request with a callback for each event (batch mode)\n"
    "\n"
    " @deprecated Use `api.chat_stream_with_callback` instead for a unified streaming API\n"
    "\n"
    " **Note**: Despite the callback, this function collects ALL events before\n"
    " calling the callbacks. It does NOT provide true real-time streaming.\n"
    " For real-time streaming, use the sans-io functions with a streaming HTTP client.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " // Prefer using api.chat_stream_with_callback instead:\n"
    " api.chat_stream_with_callback(client, request, fn(event) {\n"
    "   case api.event_text(event) {\n"
    "     Ok(text) -> io.print(text)\n"
    "     Error(_) -> Nil\n"
    "   }\n"
    " })\n"
    " ```\n"
).
-spec stream_message_with_callback(
    anthropic@client:client(),
    anthropic@request:create_message_request(),
    fun((anthropic@streaming:stream_event()) -> nil)
) -> {ok, stream_result()} | {error, stream_error()}.
stream_message_with_callback(Api_client, Message_request, Callback) ->
    gleam@result:'try'(
        stream_message(Api_client, Message_request),
        fun(Stream_result) ->
            gleam@list:each(erlang:element(2, Stream_result), Callback),
            {ok, Stream_result}
        end
    ).

-file("src/anthropic/streaming/handler.gleam", 494).
?DOC(
    " Extract text from a single event if it contains text content\n"
    "\n"
    " Useful for real-time streaming to print text as it arrives.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " list.each(events, fn(event) {\n"
    "   case get_event_text(event) {\n"
    "     Ok(text) -> io.print(text)\n"
    "     Error(_) -> Nil\n"
    "   }\n"
    " })\n"
    " ```\n"
).
-spec get_event_text(anthropic@streaming:stream_event()) -> {ok, binary()} |
    {error, nil}.
get_event_text(Event) ->
    case Event of
        {content_block_delta_event_variant, Delta_event} ->
            case erlang:element(3, Delta_event) of
                {text_content_delta, Text_delta} ->
                    {ok, erlang:element(2, Text_delta)};

                _ ->
                    {error, nil}
            end;

        _ ->
            {error, nil}
    end.

-file("src/anthropic/streaming/handler.gleam", 507).
?DOC(" Filter events to only text deltas\n").
-spec get_text_deltas(list(anthropic@streaming:stream_event())) -> list(binary()).
get_text_deltas(Events) ->
    _pipe = Events,
    gleam@list:filter_map(_pipe, fun get_event_text/1).

-file("src/anthropic/streaming/handler.gleam", 513).
?DOC(" Get the full text from a stream of events\n").
-spec get_full_text(list(anthropic@streaming:stream_event())) -> binary().
get_full_text(Events) ->
    _pipe = get_text_deltas(Events),
    gleam@string:join(_pipe, <<""/utf8>>).

-file("src/anthropic/streaming/handler.gleam", 519).
?DOC(" Get the message ID from events\n").
-spec get_message_id(list(anthropic@streaming:stream_event())) -> {ok, binary()} |
    {error, nil}.
get_message_id(Events) ->
    _pipe = Events,
    gleam@list:find_map(_pipe, fun(Event) -> case Event of
                {message_start_event, Msg} ->
                    {ok, erlang:element(2, Msg)};

                _ ->
                    {error, nil}
            end end).

-file("src/anthropic/streaming/handler.gleam", 530).
?DOC(" Get the model from events\n").
-spec get_model(list(anthropic@streaming:stream_event())) -> {ok, binary()} |
    {error, nil}.
get_model(Events) ->
    _pipe = Events,
    gleam@list:find_map(_pipe, fun(Event) -> case Event of
                {message_start_event, Msg} ->
                    {ok, erlang:element(5, Msg)};

                _ ->
                    {error, nil}
            end end).

-file("src/anthropic/streaming/handler.gleam", 541).
?DOC(" Check if stream completed successfully (from event list)\n").
-spec is_complete(list(anthropic@streaming:stream_event())) -> boolean().
is_complete(Events) ->
    gleam@list:any(Events, fun(Event) -> case Event of
                message_stop_event ->
                    true;

                _ ->
                    false
            end end).

-file("src/anthropic/streaming/handler.gleam", 551).
?DOC(" Check if stream ended with an error (from event list)\n").
-spec has_error(list(anthropic@streaming:stream_event())) -> boolean().
has_error(Events) ->
    gleam@list:any(Events, fun(Event) -> case Event of
                {error_event, _} ->
                    true;

                _ ->
                    false
            end end).

-file("src/anthropic/streaming/handler.gleam", 561).
?DOC(" Get error from events if present\n").
-spec get_error(list(anthropic@streaming:stream_event())) -> {ok,
        anthropic@streaming:stream_error()} |
    {error, nil}.
get_error(Events) ->
    _pipe = Events,
    gleam@list:find_map(_pipe, fun(Event) -> case Event of
                {error_event, Err} ->
                    {ok, Err};

                _ ->
                    {error, nil}
            end end).