A pure Effect TypeScript client library for Bluesky Jetstream - a simplified JSON event stream for the AT Protocol.
bun add effect-jetstream
This package depends on effect and @effect/platform; Bun will install them automatically. If you already use those packages, keep versions aligned.
Jetstream.live uses the global WebSocket (works in Bun and browsers). If your runtime does not provide one, use Jetstream.layer and supply Socket.WebSocketConstructor from @effect/platform/Socket.
import { Effect, Stream } from "effect"
import { Jetstream, JetstreamConfig } from "effect-jetstream"
const config = JetstreamConfig.JetstreamConfig.make({
wantedCollections: ["app.bsky.feed.post", "app.bsky.feed.like"]
})
const program = Effect.gen(function* () {
const jetstream = yield* Jetstream.Jetstream
yield* jetstream.stream.pipe(
Stream.filter((msg) => msg._tag === "CommitCreate"),
Stream.tap((msg) => Effect.log(`New ${msg.commit.collection}: ${msg.commit.rkey}`)),
Stream.runDrain
)
})
program.pipe(
Effect.provide(Jetstream.live(config)),
Effect.runPromise
)
import { Effect, Layer } from "effect"
import { Jetstream, JetstreamClient, JetstreamConfig } from "effect-jetstream"
const config = JetstreamConfig.JetstreamConfig.make({
wantedCollections: ["app.bsky.feed.post", "app.bsky.feed.like"]
})
const program = Effect.gen(function* () {
const client = yield* JetstreamClient.JetstreamClient
yield* client.onCreate("app.bsky.feed.post", (event) =>
Effect.log(`New post from ${event.did}: ${event.commit.record.text}`)
)
yield* client.onCreate("app.bsky.feed.like", (event) =>
Effect.log(`${event.did} liked ${event.commit.record.subject.uri}`)
)
yield* client.run
})
const MainLayer = JetstreamClient.layer.pipe(
Layer.provide(Jetstream.live(config))
)
program.pipe(Effect.provide(MainLayer), Effect.runPromise)
JetstreamConfig.JetstreamConfig.make({
// WebSocket endpoint (default: wss://jetstream1.us-east.bsky.network/subscribe)
endpoint: "wss://jetstream2.us-east.bsky.network/subscribe",
// Collections to subscribe to (supports wildcards like "app.bsky.feed.*")
wantedCollections: ["app.bsky.feed.post"],
// DIDs to filter by
wantedDids: ["did:plc:..."],
// Unix microseconds cursor for replay
cursor: 1725911162329308,
// Max message size (0 = no limit)
maxMessageSizeBytes: 1000000,
// Enable zstd compression (Jetstream uses a custom dictionary)
compress: false,
// Optional decoder for compressed payloads
// If omitted, Bun.zstdDecompress is used without a dictionary and a warning is logged.
// decoder: (data) => Effect.tryPromise(() => Bun.zstdDecompress(data))
})
When compress is true, provide a decoder that understands Jetstream's dictionary. If you omit it, the client falls back to Bun.zstdDecompress without a dictionary and logs a warning.
- Malformed messages are logged and dropped; the stream continues.
- Record decode failures in
JetstreamClientare logged and dropped. - Handler failures are logged and do not stop processing.
- Outbound messages are buffered;
sendwaits for a ready socket and resends after reconnect.
CommitCreate- New record createdCommitUpdate- Record updatedCommitDelete- Record deletedIdentityEvent- Handle or DID document changesAccountEvent- Account status changes
Post- app.bsky.feed.postLike- app.bsky.feed.likeRepost- app.bsky.feed.repostFollow- app.bsky.graph.followBlock- app.bsky.graph.blockProfile- app.bsky.actor.profile
interface Jetstream {
// Stream of parsed messages
readonly stream: Stream<JetstreamMessage, JetstreamError>
// Send a message to the server
readonly send: (message: SubscriberSourcedMessage) => Effect<void, JetstreamError>
// Update subscription options
readonly updateOptions: (options: OptionsUpdate) => Effect<void, JetstreamError>
// Gracefully close the connection and end the stream
readonly shutdown: Effect<void>
}
interface JetstreamClient {
// Register handler for new records in a collection
readonly onCreate: <C extends Collection>(
collection: C,
handler: (event: CommitCreate & { readonly commit: { readonly record: RecordFor<C> } }) => Effect<void>
) => Effect<void>
// Register handler for updated records
readonly onUpdate: <C extends Collection>(
collection: C,
handler: (event: CommitUpdate & { readonly commit: { readonly record: RecordFor<C> } }) => Effect<void>
) => Effect<void>
// Register handler for deleted records
readonly onDelete: <C extends Collection>(
collection: C,
handler: (event: CommitDelete) => Effect<void>
) => Effect<void>
// Register handler for event kinds (commit, identity, account)
readonly on: <K extends EventKind>(
kind: K,
handler: (event: EventFor<K>) => Effect<void>
) => Effect<void>
// Run the client (blocks forever)
readonly run: Effect<never, JetstreamError>
}
MIT