メインコンテンツまでスキップ

Consuming the firehose

This page picks up from the Relay overview and walks through actually consuming the firehose: opening the connection, decoding events, and processing them in order. We use Bluesky's relay (relay1.us-east.bsky.network) in the examples, but the same code works against any relay.

Want JSON instead?

The firehose is a binary stream: efficient, but more work to decode. If you'd rather receive filtered JSON and skip the CBOR/CAR decoding entirely, use Jetstream.

Open the connection​

The firehose is served from the com.atproto.sync.subscribeRepos endpoint over a WebSocket. No authentication is required.

uri := "wss://relay1.us-east.bsky.network/xrpc/com.atproto.sync.subscribeRepos"
con, _, err := websocket.DefaultDialer.Dial(uri, http.Header{})

Each message that arrives is binary-encoded CBOR. Commit events carry a CAR slice of the repository holding the changed records. Most SDKs provide a wrapper that handles the framing and decoding for you. Our Go SDK is currently the most feature-complete for working with the firehose directly.

Handle each event​

Once the bytes are decoded, you work with a stream of repository operations: things like "create post", "create like", or "delete follow". The example below sets up a handler that prints each operation as it arrives:

rsc := &events.RepoStreamCallbacks{
  RepoCommit: func(evt *atproto.SyncSubscribeRepos_Commit) error {
    fmt.Println("Event from ", evt.Repo)
    for _, op := range evt.Ops {
      fmt.Printf(" - %s record %s\n", op.Action, op.Path)
    }
    return nil
  },
}

sched := sequential.NewScheduler("myfirehose", rsc.EventHandler)
events.HandleRepoStream(context.Background(), con, sched)

Here we set up a sequential scheduler, which handles events one at a time, in order. Other schedulers run handlers in parallel, or do limited concurrency keyed on the event's repo. This is useful once a single-threaded handler can't keep up with the network's volume, while still preserving per-repo ordering.

HandleRepoStream does the actual decoding of the data coming over the WebSocket and dispatches each event to the handler you wrote.

Keep your place in the stream​

The firehose is sequenced: every event carries a monotonically increasing cursor. If your consumer disconnects, you can reconnect and pass the last cursor you processed back as a cursor query parameter to resume roughly where you left off, rather than missing everything that happened while you were away.

wss://relay1.us-east.bsky.network/xrpc/com.atproto.sync.subscribeRepos?cursor=<seq>

Relays only buffer a limited backlog, so if you're offline long enough to fall outside that window, resuming by cursor alone isn't enough; you need to catch up from history. How you do that comes down to whether your position in the stream is stateless or stateful.

A stateless catch-up derives the gap on demand and forgets it afterward: Jetstream's network replay streams history and cuts back over to live in one pass, but tracking what you've processed is on you.

To persist your position across restarts, you need a stateful cursor. Tap is the recommended tool for long-lived cursor management. It sits in front of the firehose and keeps a durable, per-repo cursor: you subscribe to the repos you care about, and Tap handles the connection, verification, backfill, and buffering — delivering historical events first, then cutting over to the live tail. As you acknowledge events, Tap advances your cursor, so a restart resumes exactly where you left off.

go install github.com/bluesky-social/indigo/cmd/tap@latest

# Start Tap and open its event channel
tap run
websocat ws://localhost:2480/channel

# Subscribe to a repo — Tap backfills it, then streams live
curl -X POST http://localhost:2480/repos/add \
  -H "Content-Type: application/json" \
  -d '{"dids": ["did:plc:ewvi7nxzyoun6zhxrhs64oiz"]}'

Tap builds on the same repository backfill primitives described in atproto's Backfilling guide.

Either approach becomes more robust against a Sync 1.1 endpoint: the prevData field on each commit lets you detect gaps and verify you haven't missed an update for a given repo.

See also​