Streaming

The FSharp.ATProto.Streaming package provides real-time event streams over WebSocket. Two protocols are supported: Jetstream (JSON, simpler) and Firehose (CBOR, full fidelity). Both return IAsyncEnumerable<Result<Event, StreamError>> for incremental consumption.

open FSharp.ATProto.Streaming
open FSharp.ATProto.Syntax
open System.Threading

Jetstream

Jetstream is a JSON-based relay that provides a filtered, lightweight view of the AT Protocol event stream. It is the recommended starting point for most applications.

Key Types

Functions

Subscribing to Posts

let cts = new CancellationTokenSource()

let options =
    { Jetstream.defaultOptions with
        WantedCollections = [ "app.bsky.feed.post" ] }

let events = Jetstream.subscribe options cts.Token

task {
    let enumerator = events.GetAsyncEnumerator(cts.Token)

    while! enumerator.MoveNextAsync() do
        match enumerator.Current with
        | Ok (CommitEvent (did, _timeUs, info)) when info.Operation = Create ->
            printfn "New post from %s (rkey: %s)" (Did.value did) info.Rkey
        | Ok (IdentityEvent (did, _timeUs, identity)) ->
            printfn "Identity update: %s" (Did.value identity.Did)
        | Ok (AccountEvent (did, _timeUs, account)) ->
            printfn "Account %s active=%b" (Did.value did) account.Active
        | Error StreamError.Closed ->
            printfn "Stream closed"
        | Error e ->
            printfn "Error: %A" e
        | _ -> ()

    // Cancel to disconnect
    cts.Cancel()
}

Filtering by DID

To receive events only for specific accounts, set WantedDids:

    { Jetstream.defaultOptions with
        WantedDids = [ "did:plc:abc123"; "did:plc:def456" ] }

Resuming from a Cursor

Jetstream cursors are microsecond timestamps. To resume from where you left off:

    { Jetstream.defaultOptions with
        Cursor = Some 1709312400000000L }

Firehose

The Firehose provides the full-fidelity CBOR event stream from com.atproto.sync.subscribeRepos. It includes raw CAR blocks and is suitable for indexers and relay operators.

Key Types

Functions

Subscribing to the Firehose

let firehoseOptions =
    { Firehose.defaultOptions with
        Cursor = None }

let firehoseEvents = Firehose.subscribe firehoseOptions cts2.Token

task {
    let enumerator = firehoseEvents.GetAsyncEnumerator(cts2.Token)

    while! enumerator.MoveNextAsync() do
        match enumerator.Current with
        | Ok (FirehoseCommitEvent commit) ->
            for op in commit.Ops do
                printfn "[%d] %s %A %s" commit.Seq (Did.value commit.Repo) op.Action op.Path
        | Ok (FirehoseInfoEvent info) ->
            printfn "Info: %s" info.Name
        | Ok (FirehoseErrorEvent (error, message)) ->
            printfn "Error: %s %A" error message
        | Error e ->
            printfn "Stream error: %A" e
        | _ -> ()

    cts2.Cancel()
}

Graceful Shutdown

Both subscribe functions respect CancellationToken. Cancel the token to close the WebSocket connection cleanly. After cancellation, MoveNextAsync returns false and the enumerator disposes the underlying WebSocket.

// Run for 30 seconds then stop
cts.CancelAfter(30_000)