Generic WebSocket consumer for the Exosphere.ATProto firehose using WebSockex.
This module connects to a relay's com.atproto.sync.subscribeRepos endpoint,
decodes frames into structured messages, and dispatches those messages via an
:on_event callback.
See the Firehose guide for a full walkthrough — message types, record extraction, verification, cursors, and production tips.
The :on_event callback receives (message, state) and must return an
updated state. It must not raise — the consumer does not catch exceptions
from the callback. If your callback can fail, wrap the failing work in a
Task (or your own supervised process) and return the original state.
The consumer reconnects automatically on disconnect or connection error.
A reconnect re-subscribes at the cursor the consumer has tracked in state
(the seq of the last message dispatched), so it resumes where the stream
left off rather than replaying from the original starting cursor. Reconnect
attempts are spaced out with linear backoff plus jitter, capped at a few
seconds. The tracked cursor is in-memory only — to resume a stream after a
restart, persist msg.seq from your callback and pass it back via the
:cursor option on next start.
Summary
Functions
Start the firehose consumer.
Types
@type stats() :: %{ frames: non_neg_integer(), messages: non_neg_integer(), errors: non_neg_integer(), started_at: integer() }
Functions
@spec start_link(keyword()) :: GenServer.on_start()
Start the firehose consumer.
Options
:relay_url- Relay WebSocket URL (default:"wss://bsky.network"):cursor- Starting cursor for resumption (optional):on_event- Callback invoked with each decoded message (required):name- Process name (optional)