ProtoRune.Jetstream (proto_rune v0.5.1)

Copy Markdown

Consumer for the AT Protocol Jetstream, the filtered JSON re-encode of the repo firehose.

Where ProtoRune.Firehose delivers every commit on the network and decodes every CAR block locally, Jetstream applies wantedCollections and wantedDids filters server-side and ships plain JSON text frames: a consumer that only cares about a handful of collections receives (and pays to decode) only those events. Add it to your supervision tree:

children = [
  {ProtoRune.Jetstream,
   handler: MyConsumer, wanted_collections: ["place.quintal.feed.prosa"]}
]

or start it directly with start_link/1.

Handlers

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

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

Filtering and cursors

  • :wanted_collections - AT-URI collections to receive commit events for (repeatable server-side filter). Identity and account events are not collection-scoped and always flow through.
  • :wanted_dids - restrict events to these repository DIDs.
  • :cursor - a time_us timestamp (microseconds) to resume from. Jetstream keeps a rolling playback window; resuming may re-deliver a handful of events around the cursor, so consumers must stay idempotent. When the connection drops, the consumer reconnects resuming from the last delivered time_us. cursor/1 returns the current value, e.g. to persist it for a later restart.

Remaining options

  • :relay - the Jetstream instance base URL (default: "wss://jetstream2.us-east.bsky.network").
  • :auto_reconnect - reconnect automatically on connection loss (default: true). When false, the process stops with reason {:jetstream_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). Jetstream speaks the same WebSocket transport as the firehose, only with text frames.
  • :transport_opts - options passed to the transport.
  • :name - registers the process under the given name.

Summary

Types

Where decoded events are delivered to.

Functions

Returns a specification to start this module under a supervisor.

Returns the time_us of the last delivered event, or nil when no event has been delivered yet.

Starts a Jetstream consumer process.

Types

handler()

@type handler() ::
  pid() | (ProtoRune.Jetstream.Event.t() -> term()) | {module(), atom()}

Where decoded events are delivered to.

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 time_us of the last delivered event, or nil when no event has been delivered yet.

get_schema(atom)

options_t(data)

options_t!(data)

options_t_changeset(data)

start_link(opts)

@spec start_link(keyword() | map()) :: {:ok, pid()} | {:error, term()}

Starts a Jetstream consumer process.

See the module documentation for the available options.