ovsdb_ex

defmodule OVSDB.Framer do
  @moduledoc """
  Incremental JSON message framer for the OVSDB wire protocol.

  OVSDB sends bare JSON objects back-to-back on the stream with
  **no length prefix and no delimiter** between messages — per
  [RFC 7047 §4][rfc-wire] (which defers to JSON-RPC 1.0 for framing,
  which itself says nothing). Receivers find message boundaries by
  brace-counting at the JSON structural level.

  [rfc-wire]: https://www.rfc-editor.org/rfc/rfc7047#section-4

  ## What this module does

  Accumulates bytes from the wire and extracts complete JSON objects
  as they become available. The state machine tracks:

    * Depth of `{` vs `}` — incremented and decremented only outside
      of strings, so `"{"` inside a string literal doesn't affect
      nesting.
    * Whether we're currently inside a JSON string.
    * Whether the previous character was `\\\\` (escape), so that
      `"\\\""` inside a string doesn't terminate the string.

  When depth returns to zero after incrementing, one complete
  message has been framed and is returned as a binary slice of the
  buffer.

  ## Why hand-written and not Jason

  Jason doesn't expose an incremental decoder. The alternatives
  (re-parse the entire buffer on every chunk and catch partial-input
  errors) are quadratic in message size and have ambiguous
  "incomplete vs malformed" signaling. The hand-written state
  machine is ~50 lines, zero-dep, and matches what every other
  OVSDB implementation does.

  ## UTF-8 safety

  RFC 8259 (JSON) mandates UTF-8 source and restricts structural
  characters (`{`, `}`, `"`, `[`, `]`, `,`, `:`) to their ASCII
  code points. Multi-byte UTF-8 sequences within strings never
  contain ASCII bytes, so byte-level brace counting is safe.

  ## Usage

      framer = Framer.new()

      # Feed bytes as they arrive from :gen_tcp
      {framer, messages} = Framer.feed(framer, chunk_1)
      {framer, more}     = Framer.feed(framer, chunk_2)

      # `messages` is a list of complete JSON binaries, ready for Jason.decode/1
      for msg <- messages, do: handle(Jason.decode!(msg))

  The framer never drops or reorders bytes; partial messages are
  buffered until completion. Malformed input (e.g. unbalanced
  braces at top level, bytes before the first `{`) is detected and
  returned as an error.
  """

  @enforce_keys []
  defstruct buffer: "", depth: 0, in_string: false, escape: false, start: 0, offset: 0

  @type t :: %__MODULE__{
          buffer: binary(),
          depth: non_neg_integer(),
          in_string: boolean(),
          escape: boolean(),
          start: non_neg_integer(),
          offset: non_neg_integer()
        }

  @typedoc """
  A freshly-created framer. Narrower than `t()` — all fields are at
  their initial values.
  """
  @type empty :: %__MODULE__{
          buffer: <<>>,
          depth: 0,
          in_string: false,
          escape: false,
          start: 0,
          offset: 0
        }

  @type framer_error :: :unexpected_byte | :buffer_overflow

  @doc """
  Creates a new empty framer.

      iex> framer = OVSDB.Framer.new()
      iex> framer.depth
      0
      iex> framer.buffer
      ""
  """
  @spec new() :: empty()
  def new, do: %__MODULE__{}

  @doc """
  Feeds a chunk of bytes into the framer.

  Returns `{framer, messages}` where `messages` is a list of
  complete JSON object binaries (in order), or
  `{:error, framer_error(), partial_framer}` on malformed input.

  Incomplete messages are held in the framer's internal buffer and
  emitted on subsequent `feed/2` calls once their closing `}`
  arrives.

      iex> {f, msgs} = OVSDB.Framer.feed(OVSDB.Framer.new(), ~s({"a":1}))
      iex> msgs
      [~s({"a":1})]
      iex> f.buffer
      ""

      iex> {f, msgs} = OVSDB.Framer.feed(OVSDB.Framer.new(), ~s({"a":1}{"b":2}))
      iex> msgs
      [~s({"a":1}), ~s({"b":2})]
      iex> f.buffer
      ""

      iex> f = OVSDB.Framer.new()
      iex> {f, msgs} = OVSDB.Framer.feed(f, ~s({"a":))
      iex> msgs
      []
      iex> {_f, msgs} = OVSDB.Framer.feed(f, ~s(1}))
      iex> msgs
      [~s({"a":1})]
  """
  @spec feed(t(), binary()) ::
          {t(), [binary()]} | {:error, framer_error(), t()}
  def feed(%__MODULE__{} = framer, chunk) when is_binary(chunk) do
    new_buffer = framer.buffer <> chunk
    do_feed(%{framer | buffer: new_buffer}, [])
  end

  # ---------------------------------------------------------------------------
  # Private — the state machine
  # ---------------------------------------------------------------------------

  defp do_feed(%__MODULE__{buffer: buf, offset: off} = framer, acc)
       when byte_size(buf) == off do
    # All bytes consumed; compact the buffer by discarding prefix up
    # to `start`. In the steady state `start == off` after a
    # complete-message emission, so this is a cheap slice.
    remaining =
      if framer.start == byte_size(buf) do
        ""
      else
        binary_part(buf, framer.start, byte_size(buf) - framer.start)
      end

    new_offset = framer.offset - framer.start
    new_start = 0

    {%{framer | buffer: remaining, offset: new_offset, start: new_start}, Enum.reverse(acc)}
  end

  defp do_feed(%__MODULE__{} = framer, acc) do
    <<_::binary-size(framer.offset), byte, _::binary>> = framer.buffer
    handle_byte(framer, byte, acc)
  end

  # Inside a string, previous byte was a backslash → consume
  # unconditionally.
  defp handle_byte(%__MODULE__{in_string: true, escape: true} = framer, _byte, acc) do
    do_feed(%{framer | escape: false, offset: framer.offset + 1}, acc)
  end

  # Inside a string. Closing quote or escape start?
  defp handle_byte(%__MODULE__{in_string: true} = framer, byte, acc) do
    case byte do
      ?\\ -> do_feed(%{framer | escape: true, offset: framer.offset + 1}, acc)
      ?" -> do_feed(%{framer | in_string: false, offset: framer.offset + 1}, acc)
      _ -> do_feed(%{framer | offset: framer.offset + 1}, acc)
    end
  end

  # Outside a string at depth 0, waiting for a new message. Skip
  # interstitial whitespace; anything other than `{` or whitespace
  # is malformed.
  defp handle_byte(%__MODULE__{depth: 0} = framer, byte, acc) do
    cond do
      byte == ?{ ->
        framer = %{framer | depth: 1, start: framer.offset, offset: framer.offset + 1}
        do_feed(framer, acc)

      byte in [?\s, ?\t, ?\n, ?\r] ->
        framer = %{framer | offset: framer.offset + 1, start: framer.offset + 1}
        do_feed(framer, acc)

      true ->
        {:error, :unexpected_byte, framer}
    end
  end

  # Outside a string, inside a message (depth > 0). Track braces
  # and string openings.
  defp handle_byte(%__MODULE__{} = framer, byte, acc) do
    case byte do
      ?{ ->
        do_feed(%{framer | depth: framer.depth + 1, offset: framer.offset + 1}, acc)

      ?} ->
        new_depth = framer.depth - 1
        new_offset = framer.offset + 1

        if new_depth == 0 do
          # Message complete — slice [start, new_offset) out of buffer.
          len = new_offset - framer.start
          message = binary_part(framer.buffer, framer.start, len)

          framer = %{
            framer
            | depth: 0,
              offset: new_offset,
              start: new_offset
          }

          do_feed(framer, [message | acc])
        else
          do_feed(%{framer | depth: new_depth, offset: new_offset}, acc)
        end

      ?" ->
        do_feed(%{framer | in_string: true, offset: framer.offset + 1}, acc)

      _ ->
        do_feed(%{framer | offset: framer.offset + 1}, acc)
    end
  end
end