Table of Contents

Jetstream live tail

Jetstream streams AT Protocol activity as decoded events over a WebSocket. With AtProtoJetstream, you can filter at the server and process events with await foreach; the live v2 stream reconnects after transient disconnections and resumes from the last yielded sequence. You do not need an archive API key to tail live events.

This guide covers the live tail only. For a bounded archive snapshot or an archive-to-live stream that can recover beyond the live server's lookback window, see network replay and snapshots. If you are upgrading an existing event-driven client or a v1 connection, see the Jetstream migration guide.

Warning

Treat data received from Bluesky, AT Protocol services, and streams such as Jetstream and the firehose as untrusted. This includes post and message text, profile and list metadata, labels, error details such as AtErrorDetail, and raw JSON. Before displaying it, encode it for the output context (for example, HTML-encode it for a web page). Before using a URL in a link or other active context, check its scheme against an allow-list of the schemes your application expects, such as https; a well-formed URL can still use an active scheme such as javascript:, which can lead to cross-site scripting. Validate other values before using them in active contexts.

The sample code on this page is for illustration. It does not sanitize or otherwise validate the input it uses; apply the validation and encoding appropriate for your application.

Quickstart: the live tail

Create a v2 client (the default), select a collection, and iterate over its events:

using idunno.AtProto.Jetstream;

await using var jetstream = new AtProtoJetstream(
    collections: ["app.bsky.feed.post"]);

await foreach (JetstreamEvent evt in jetstream.StreamAsync())
{
    if (evt is JetstreamCommitEvent { Commit.Operation: JetstreamCommitOperation.Create } post)
    {
        Console.WriteLine($"{post.Did} posted {post.Commit.Record}");
    }
}

StreamAsync opens the connection when iteration starts. Each JetstreamEvent has a Did, a Kind, a nullable Sequence (populated by v2), and a WitnessedAt timestamp. A commit carries the parsed record as JsonElement?, not a typed Bluesky record; deletes have no record. Check the kind and operation before accessing a commit's payload, and validate external records before deserializing them as a particular type.

The default v2 host is wss://jetstream.us-west.bsky.network. You can pass a different v2 host as uri to the constructor; AtProtoJetstream appends the subscription path. StreamAsync does not support JetstreamProtocolVersion.V1. See migration to retain an event-driven v1 connection.

Warning

Jetstream v2 delivery is at least once: a reconnect cursor is inclusive, so an event can be received again. StreamAsync suppresses events at or below the last yielded sequence within one enumeration, but this is not exactly-once processing across application restarts or failures. Persist the last successfully processed sequence with the state it represents, and make processing idempotent. Treat sequence values as server-provided data, not proof of event authenticity.

Filtering

Supply collections and DIDs to the constructor. A collection may be an exact NSID or a namespace wildcard such as app.bsky.feed.*:

await using var jetstream = new AtProtoJetstream(
    collections: ["app.bsky.feed.post", "app.bsky.feed.like"],
    dids: ["did:plc:ec72yg6n2sydzjvtovvdlxrk"]);

await foreach (JetstreamEvent evt in jetstream.StreamAsync())
{
    Console.WriteLine($"{evt.Sequence}: {evt.Kind} from {evt.Did}");
}

The collection filter applies to commits, not identity, account or sync events. The DID filter applies to all kinds. To ask the server for commits only, set KindFilter before starting the stream:

await using var jetstream = new AtProtoJetstream(collections: ["app.bsky.feed.post"]);
jetstream.KindFilter = [JetStreamEventKind.Commit];

await foreach (JetstreamEvent evt in jetstream.StreamAsync())
{
    if (evt is JetstreamCommitEvent commit)
    {
        Console.WriteLine($"{commit.Commit.Operation}: {commit.Commit.RKey}");
    }
}

Only omit identity, account and sync events if you do not need to act on account changes or repo resynchronization. A collection filter cannot be combined with a nonempty kind filter that excludes commits. V2 servers accept at most 100 collections and 10,000 DIDs. Changing filters while streaming reconnects with the new filters; choose stable filters when persisting a cursor for that stream.

Reacting to events

Match the decoded event type to process record changes, handles, account status, and sync markers:

await using var jetstream = new AtProtoJetstream(
    collections: ["app.bsky.feed.post"]);

await foreach (JetstreamEvent evt in jetstream.StreamAsync())
{
    switch (evt)
    {
        case JetstreamCommitEvent commit:
            if (commit.Commit.Operation == JetstreamCommitOperation.Delete)
            {
                Console.WriteLine($"Delete at://{commit.Did}/{commit.Commit.Collection}/{commit.Commit.RKey}");
            }
            else
            {
                Console.WriteLine($"Put {commit.Commit.Collection}/{commit.Commit.RKey}: {commit.Commit.Record}");
            }
            break;

        case JetstreamIdentityEvent identity:
            Console.WriteLine($"{identity.Did} changed handle to {identity.Identity.Handle}");
            break;

        case JetstreamAccountEvent account:
            Console.WriteLine($"{account.Did} active: {account.Account.Active}");
            break;

        case JetstreamSyncEvent sync:
            Console.WriteLine($"{sync.Did} needs resync from revision {sync.Sync.Rev}");
            break;
    }
}

A sync marker means a repository's commit history can no longer be followed; consumers maintaining a mirror should resynchronize that repository. Unknown future kinds can arrive as a plain JetstreamEvent with Kind == JetStreamEventKind.Unknown.

Showing handles

Events identify accounts by DID, and the handle in an identity event is not verified. To show verified handles, use a DidHandleCache. Set JetstreamOptions.DidHandleResolver, or call AtProtoJetstreamBuilder.WithDidHandleResolver(), and the Jetstream invalidates a cached handle when it receives an identity event for its DID:

using var didHandleCache = new DidHandleCache();

await using var jetstream = new AtProtoJetstream(
    options: new JetstreamOptions
    {
        DidHandleResolver = didHandleCache
    });

await foreach (JetstreamEvent evt in jetstream.StreamAsync())
{
    if (evt is JetstreamAccountEvent account)
    {
        Handle handle = await didHandleCache.ResolveHandleAsync(account.Did);
        Console.WriteLine($"{account.Did} ({handle}) active: {account.Account.Active}");
    }
}

Use TryGetCachedHandle(), which never makes a network request, for frequent events such as commits.

Stopping and shutdown

Breaking out of await foreach disposes the enumerator and closes its connection. You can also pass a cancellation token for shutdown initiated elsewhere:

using var shutdown = new CancellationTokenSource();
await using var jetstream = new AtProtoJetstream();

try
{
    await foreach (JetstreamEvent evt in jetstream.StreamAsync(cancellationToken: shutdown.Token))
    {
        Console.WriteLine(evt.Sequence);
        // Call shutdown.Cancel() from your shutdown handler.
    }
}
catch (OperationCanceledException) when (shutdown.IsCancellationRequested)
{
    // Normal shutdown.
}

One AtProtoJetstream instance supports only one active enumerator/connection. While enumerating, you cannot subscribe to RecordReceived, MessageReceived, ConnectionStateChanged, FaultRaised or InfoReceived, or call ConnectAsync. Remove existing handlers and close an event-driven connection before switching to enumeration. The migration guide shows both forms.

Resuming where you left off

V2 sequences are cursors scoped to the issuing server. StreamAsync() retries transient connection failures and suppresses the repeated event from its inclusive cursor within a single enumeration. By default retries are unlimited. Pass maximumReconnectAttempts: 5 to stop after five consecutive reconnections; yielding an event resets the count. An exhausted limit throws IOException. To resume after a process restart, supply the last successfully processed sequence:

long? savedSequence = LoadSavedSequence(); // Your own durable cursor store.
await using var jetstream = new AtProtoJetstream();

await foreach (JetstreamEvent evt in jetstream.StreamAsync(cursor: savedSequence))
{
    await ProcessAsync(evt);
    if (evt.Sequence is long sequence)
    {
        SaveSequence(sequence); // Persist only after processing succeeds.
    }
}

LoadSavedSequence, ProcessAsync and SaveSequence represent your application code. The server may send the saved cursor again after a restart, so processing should be idempotent (for example, key writes by at://{did}/{collection}/{rkey}). Do not persist LastSequence as a substitute for your processed cursor: it tracks received events, which can be ahead of completed work.

If the server refuses an expired cursor, StreamAsync throws JetstreamConnectionException (for example, ErrorDetail?.Error == "CursorTooOld"). It does not skip to the current tip; recover missing events with archive-to-live replay or explicitly choose a new live starting point. An OutdatedCursor notice also terminates enumeration instead of silently losing the gap.

See also

Edit this page