Real-time event stream client for the ATProto firehose.

Connects to com.atproto.sync.subscribeRepos over WebSocket, decodes the CBOR frames and delivers each event as a ProtoRune.Firehose.Event to a consumer-provided handler.

ProtoRune is a library and starts no processes on its own: add the firehose to your own supervision tree:

children = [
  {ProtoRune.Firehose, handler: self()}
]

or start it directly with start_link/1.

Handlers

The required :handler option tells the client where events go:

• a pid, which receives {:firehose, %ProtoRune.Firehose.Event{}} messages
• a one-arity function, called with the event
• a {module, function} tuple, called as function.(event)

Backfill and cursors

Every event carries a sequence number in event.seq. Pass a previously seen sequence number as the :cursor option to backfill events missed while disconnected (relays keep a rolling history window). When the connection drops, the client reconnects automatically resuming from the last delivered sequence number. cursor/1 returns the current sequence number, e.g. to persist it for a later restart.

Options

• :handler (required) - where events are delivered (see above).
• :relay - the relay base URL (default: "wss://bsky.network").
• :cursor - sequence number to start the stream from (default: live).
• :auto_reconnect - reconnect automatically on connection loss (default: true). When false, the process stops with reason {:firehose_disconnected, reason} instead.
• :backoff_initial / :backoff_max - reconnect backoff bounds in milliseconds (default: 1_000 / 30_000). The delay doubles after each failed attempt and resets once connected.
• :transport - the ProtoRune.Firehose.Transport implementation to use (default: ProtoRune.Firehose.Transport.Gun).
• :transport_opts - options passed to the transport, e.g. [timeout: 10_000].
• :name - registers the process under the given name.

Examples

defmodule MyConsumer do
  use GenServer

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def init(_opts) do
    {:ok, firehose} = ProtoRune.Firehose.start_link(handler: self())
    {:ok, %{firehose: firehose}}
  end

  def handle_info({:firehose, %ProtoRune.Firehose.Event{type: :commit} = event}, state) do
    Enum.each(event.ops, &process_op(event, &1))
    {:noreply, state}
  end

  def handle_info({:firehose, _event}, state), do: {:noreply, state}
end