DrinkMeAlice.Ws (drink_me_alice v0.1.0)

Passive, bounded WebSocket transport over Mint 1.10.1 / mint_web_socket 1.0.6 — production adaptation of the accepted BF-02A SurrealTransport.Ws probe.

HTTP/1 only (protocols: [:http1]), mode: :passive on connect and on Mint.WebSocket.new/5; inbound data arrives only through bounded Mint.WebSocket.recv/3 calls. The BEAM mailbox is never an unbounded buffer for socket data.
No WebSocket extensions are proposed, so permessage-deflate cannot be negotiated — compression is disabled by construction.
DrinkMeAlice.FrameGuard walks raw bytes ahead of Mint.WebSocket.decode/2 and refuses declared or aggregate payloads over :max_message (1 MiB); Mint-side buffer+fragment retention is separately tripwired at :max_retained.
Caller-driven ping/pong: pings are counted, bounded per batch, and answered with one batched send.
All waits are bounded: connect/upgrade deadlines, send_timeout, max_header_list_size on the upgrade response, per-recv_frames/2 frame/byte/ping bounds.
Error tuples are {:error, t(), reason} / {:closed, t(), frames} / {:timeout, t()}; reasons are sanitized — never Mint structs, frame payloads, headers, or auth material.
Fatal input paths close the owned socket at the recv boundary and return conn: nil, so close/2, recv_frames/2 and send_frame/2 stay safe and idempotent.

Text payloads are decoded through DrinkMeAlice.Json (the bounded Surreal-dialect profile) by classify/1 or by the owning DrinkMeAlice.Conn — never by unbounded :json.decode. Request correlation is owned by Conn; this module only moves frames.

Summary

Types

classified()

t()

Functions

close(state, timeout \\ 1000)

Graceful close: send a close frame at most once, await the peer's close under a finite deadline, then close the transport on every path. Always releases the owned socket. Returns :ok.

connect(host, port, path, opts \\ [])

Connect over cleartext HTTP/1 and upgrade path to a WebSocket.

decode_text(text)

Decode an inbound text payload through the bounded Surreal JSON profile. Notifications are already separated by feed/2; this is for RPC replies.

feed(state, bytes)

Feed raw inbound bytes through the framing guard, then Mint decode.

guard_message_bytes(ws)

Payload bytes accumulated in the guard's open fragmented message.

recv_frames(state, timeout \\ 2000)

One bounded socket read + guard + decode. Returns {:ok, state, frames}, {:timeout, state}, {:closed, state, frames} or {:error, state, r}.

retained_bytes(ws)

Bytes currently retained inside Mint.WebSocket (buffer + fragment).

send_frame(state, frame)

Encode and send one frame.

socket(ws)

Underlying TCP socket (observation only); nil after a fatal error.

Types

classified()

@type classified() ::
  {:text, binary()}
  | {:binary, binary()}
  | {:notification, map(), non_neg_integer()}
  | {:ping, binary()}
  | {:pong, binary()}
  | {:close, non_neg_integer() | nil, binary() | nil}
  | {:malformed, term()}

t()

@type t() :: %DrinkMeAlice.Ws{
  auto_pong: term(),
  close_attempted: term(),
  conn: term(),
  guard: term(),
  max_message: term(),
  max_retained: term(),
  ref: term(),
  retained_peak: term(),
  ws: term()
}

Functions

close(state, timeout \\ 1000)

@spec close(t(), timeout()) :: :ok

Graceful close: send a close frame at most once, await the peer's close under a finite deadline, then close the transport on every path. Always releases the owned socket. Returns :ok.

connect(host, port, path, opts \\ [])

@spec connect(
  :inet.hostname() | :inet.ip4_address(),
  :inet.port_number(),
  String.t(),
  keyword()
) :: {:ok, t(), [classified()]} | {:error, term()}

Connect over cleartext HTTP/1 and upgrade path to a WebSocket.

Returns {:ok, state, early_frames} — complete frames whose bytes arrived coalesced with the 101 response. They pass through the same guard/ping/close handling as later traffic.

Options: :connect_timeout, :upgrade_timeout, :send_timeout, :max_message, :protocols (default ["json"]), :headers, :auto_pong.

decode_text(text)

@spec decode_text(binary()) ::
  {:ok, DrinkMeAlice.Json.json()} | {:error, DrinkMeAlice.Error.t()}

Decode an inbound text payload through the bounded Surreal JSON profile. Notifications are already separated by feed/2; this is for RPC replies.

feed(state, bytes)

@spec feed(t(), binary()) :: {:ok, t(), [classified()]} | {:error, t(), term()}

Feed raw inbound bytes through the framing guard, then Mint decode.

Frames in the returned list may include {:notification, body, raw_byte_size} (SurrealDB live-query notifications classified inline — decoded through the bounded Json profile, never stored as terms beyond the raw byte size) and {:malformed, reason} entries.

guard_message_bytes(ws)

@spec guard_message_bytes(t()) :: non_neg_integer()

Payload bytes accumulated in the guard's open fragmented message.

recv_frames(state, timeout \\ 2000)

@spec recv_frames(t(), timeout()) ::
  {:ok, t(), [classified()]}
  | {:timeout, t()}
  | {:closed, t(), [classified()]}
  | {:error, t(), term()}

One bounded socket read + guard + decode. Returns {:ok, state, frames}, {:timeout, state}, {:closed, state, frames} or {:error, state, r}.

Every decoded frame is counted against a per-call cap before any ping is answered; pings beyond the batch cap fail the call. Server pings are answered with one batched send. A peer close frame is echoed at most once.

retained_bytes(ws)

@spec retained_bytes(t()) :: non_neg_integer()

Bytes currently retained inside Mint.WebSocket (buffer + fragment).

send_frame(state, frame)

@spec send_frame(t(), Mint.WebSocket.frame() | Mint.WebSocket.shorthand_frame()) ::
  {:ok, t()} | {:error, t(), term()}

Encode and send one frame.

socket(ws)

Underlying TCP socket (observation only); nil after a fatal error.