Exosphere.ATProto.Firehose.Consumer (Exosphere v0.2.0)

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.

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, replaying the original subscription URL (including its starting cursor). The cursor tracked in state 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

Types

stats()

t()

Functions

start_link(opts)

Start the firehose consumer.

Types

stats()

@type stats() :: %{
  frames: non_neg_integer(),
  messages: non_neg_integer(),
  errors: non_neg_integer(),
  started_at: integer()
}

t()

@type t() :: %Exosphere.ATProto.Firehose.Consumer{
  cursor: integer() | nil,
  on_event: (map(), t() -> t()),
  stats: stats()
}

Functions

start_link(opts)

@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)