ProtoRune.Bot.Firehose (proto_rune v0.3.0)

Copy Markdown

A GenServer module that streams real-time events from the ATProto firehose and dispatches them to the bot server.

The Firehose strategy is an alternative to ProtoRune.Bot.Poller: instead of periodically fetching the bot's notifications, it keeps a WebSocket connection to a relay's com.atproto.sync.subscribeRepos endpoint and dispatches every decoded event to the bot server through the same {:handle_event, event, payload} message used by the poller.

The WebSocket connection, CBOR decoding, reconnection and backoff are delegated to ProtoRune.Firehose; this module only translates ProtoRune.Firehose.Event structs into bot events.

Dispatched events

  • :commit - one event per repository operation of a commit. The payload contains :repo, :rev, :seq, :time, :action (:create, :update or :delete), :path, :cid and :record (the decoded record, nil for deletions).
  • :identity, :account, :handle, :migrate, :tombstone, :info, :error, :unknown - one event per firehose message of that type. The payload contains :repo, :seq, :time and the raw decoded frame under :payload.

Options

  • :name (required) - The name of the GenServer instance.
  • :server_pid (required) - The bot server process that receives the events.
  • :relay - The relay base URL (default: "wss://bsky.network").
  • :cursor - The sequence number to start the stream from, for backfilling events missed while disconnected. Accepts an integer, a numeric string or "latest" (default: "latest", meaning live events only).
  • :auto_reconnect - Reconnect automatically when the connection drops (default: true).
  • :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 / :transport_opts - The ProtoRune.Firehose.Transport implementation to use and its options (default: ProtoRune.Firehose.Transport.Gun).

Example

ProtoRune.Bot.Firehose.start_link([
  name: :my_bot_firehose,
  server_pid: self(),
  cursor: 32_625_482_169
])

Summary

Types

kwargs()
option()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

cursor(server)

Returns the sequence number of the last delivered event, or nil when no event has been delivered yet. Feed it back as the :cursor option to backfill from this point after a restart.

start_link(opts)

Types

kwargs()

@type kwargs() :: [option(), ...]

option()

@type option() ::
  {:name, atom()}
  | {:relay, String.t()}
  | {:cursor, String.t() | non_neg_integer()}
  | {:auto_reconnect, boolean()}
  | {:backoff_initial, pos_integer()}
  | {:backoff_max, pos_integer()}
  | {:transport, module()}
  | {:transport_opts, keyword()}
  | {:server_pid, pid()}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

cursor(server)

@spec cursor(GenServer.server()) :: non_neg_integer() | nil

Returns the sequence number of the last delivered event, or nil when no event has been delivered yet. Feed it back as the :cursor option to backfill from this point after a restart.

start_link(opts)

@spec start_link(kwargs()) :: GenServer.on_start()