anthropic_community

defmodule Anthropic.HTTPTransport.SSE do
  @moduledoc """
  Minimal line-buffering Server-Sent-Events parser. Consumes raw byte chunks (as delivered
  by `Finch.stream/5`'s `:data` callback) and emits complete `{event_name, data}` frames,
  buffering partial lines and partial frames across chunk boundaries.
  """

  defstruct buffer: "", pending_event: nil, pending_data: []

  @type frame :: {event :: String.t() | nil, data :: String.t()}
  @type t :: %__MODULE__{
          buffer: String.t(),
          pending_event: String.t() | nil,
          pending_data: list(String.t())
        }

  @spec new() :: t()
  def new, do: %__MODULE__{}

  @doc "Feeds a raw byte chunk in; returns the complete frames it produced and the updated parser state."
  @spec feed(t(), binary()) :: {list(frame()), t()}
  def feed(%__MODULE__{} = parser, chunk) when is_binary(chunk) do
    lines = String.split(parser.buffer <> chunk, "\n", trim: false)
    {complete_lines, [incomplete]} = Enum.split(lines, -1)

    process_lines(complete_lines, %{parser | buffer: incomplete})
  end

  defp process_lines(lines, parser) do
    Enum.reduce(lines, {[], parser}, fn line, {frames, parser} ->
      # The SSE spec allows CRLF line endings; strip a trailing \r left over from splitting
      # on "\n" alone so a "\r"-terminated field name/value doesn't leak into pending_event
      # or pending_data.
      case parse_line(String.trim_trailing(line, "\r"), parser) do
        {:frame, frame, parser} -> {frames ++ [frame], parser}
        {:continue, parser} -> {frames, parser}
      end
    end)
  end

  defp parse_line("", %{pending_data: []} = parser), do: {:continue, parser}

  defp parse_line("", parser) do
    data = parser.pending_data |> Enum.reverse() |> Enum.join("\n")
    {:frame, {parser.pending_event, data}, %{parser | pending_event: nil, pending_data: []}}
  end

  defp parse_line("event: " <> name, parser), do: {:continue, %{parser | pending_event: name}}

  defp parse_line("data: " <> data, parser),
    do: {:continue, %{parser | pending_data: [data | parser.pending_data]}}

  defp parse_line("data:" <> data, parser),
    do: {:continue, %{parser | pending_data: [String.trim_leading(data) | parser.pending_data]}}

  defp parse_line(":" <> _comment, parser), do: {:continue, parser}
  defp parse_line(_other, parser), do: {:continue, parser}
end