Example

Define your compile time options and implement the available callbacks to get a Jetstream V2 compatible subscriber.

defmodule MyApp.Posts do
  use Kite,
    endpoints: ["wss://jetstream.us-east.bsky.network"],
    collections: ["app.bsky.feed.post"],
    kinds: [:commit]

  def handle_event(payload, _context), do: IO.inspect(payload)
end

Add your newly defined subscriber to your application.ex and you're ready to go.

@impl true
def start(_type, _args) do

  children = [
    {MyApp.Posts, []} # <- Add your new module to the list of children.
    ...
  ]

  opts = [strategy: :one_for_one, name: MyApp.Supervisor]
  Supervisor.start_link(children, opts)
end

Or, for some more ad-hoc experimenting, try MyApp.Posts.start_link([]) in IEx.

Callbacks

handle_event/2 is required. It is recommended to handle_event minimal, offloading heavy processing or IO, to avoid the subscriber falling behind the jetstream. The second argument of handle_event is context, which is user defined data passed when starting the subscriber. This can be used to make metadata available to the handler where you run multiple instances of the same implementation.

If you don't define get_cursor/0 and put_cursor/1 the subscriber always starts from the latest event available. For most practical use cases you want to implement get_cursor/0 and put_cursor/1, in order to gracefully handle restarts or intermittent issues and being able to continue where you left off.

Options

The supported options are:

• endpoints - required - a list of websocket URLs for jetstreams, like "wss://jetstream.us-east.bsky.network" or "wss://jetstream.us-west.bsky.network". If you pass multiple endpoints, Kite will automatically failover and rotate through the options when failing to connect.
• collections - a list of NSID strings representing the resources you're interested in, like "app.bsky.feed.post" or "site.standard.document".
• kinds - a list of atoms corresponding to the types of events you want to see, normally just [:commit].
• context - use this to pass anything you want to the handle_event callback, it will be passed as the second argument.
• enabled - default true, set to false to prevent the subscriber from connecting to the jetstream, for example in tests.
• flush_interval - default 5s, the frequency with which to write the latest cursor state with put_cursor/1. Write more frequently to reduce the window of potential repeated events on a crash.
• rewind - number of seconds to rewind the cursor by when failing over, defaults to 10s.
• max_attempts - number of attempts, with backoff, before failing over when multiple endpoints are defined and we're failing to connect.

Anything set in the use Kite options can be overriden when starting the subscriber from your Application children list, eg:

children = [
  {MyApp.Posts, endpoints: [System.fetch_env!("JETSTREAM_URL")]}
]

Payloads

Jetstream outputs events in JSON, sending batches over the websocket, and Kite exposes the raw data for you. Jetstream emits 4 different "kinds" of events: #identity, #account, #commit, and #sync. For a full reference, check out the docs. For now we'll focus on #commit which represents changes to atproto records, including create, update, and delete. Note that records can have defined schemas, aka lexicons, but the data in the jetstream is not guaranteed to be validated and correct, so you must do your own validation.

Here's an example Bluesky post (#commit kind), in the full handler payload format:

%{
  "$type" => "network.bsky.jetstream.subscribeEvents#commit",
  "cid" => "bafyreigkx5h...",
  "collection" => "app.bsky.feed.post",
  "did" => "did:plc:user",
  "operation" => "create",
  "record" => %{
    "$type" => "app.bsky.feed.post",
    "createdAt" => "2026-09-16T18:13:33.933Z",
    "langs" => ["nl"],
    "reply" => %{
      "parent" => %{
        "cid" => "bafyreigkx5h...",
        "uri" => "at://did:plc:user/app.bsky.feed.post/rkey"
      },
      "root" => %{
        "cid" => "bafyreigkx5h...",
        "uri" => "at://did:plc:user/app.bsky.feed.post/rkey"
      }
    },
    "text" => "Voeg hier even vvd aan toe."
  },
  "rev" => "3mvn...",
  "rkey" => "3mvn...",
  "seq" => 25947795773,
  "time" => "2026-09-16T18:13:35.185499Z"
}

And to demonstrate how to write a handler for it, let's say we want to just print the text of the posts.

def handle_event(%{"operation" => "create"} = commit, _context) do
  %{"record" => record, "did" => did} = commit

  if text = record["text"] do
    IO.inspect(text)
  end
end

def handle_event(_commit, _context), do: :ok

Delivery guarantees

Event delivery on Jetstream V2 is best effort. Under normal circumstances you should expect at least once delivery, but there are situations where this is not possible. Kite uses timestamp cursors, as recommended by the Bluesky team, and implements automatic failover when multiple endpoints are configured. When failing over, Kite automatically rewinds the cursor a small amount to account for clock skew between instances. This is likely to lead to some replay of events. Cursors are periodically flushed with the put_cursor/1 callback, which means that a crashed subscriber can also lead to replay of events.

If your subscriber falls too far behind you can lose events, because the jetstream instances only keep a window of data matching about 1-2 days. To avoid losing events ensure that your handle_event/2 callback is fast enough to keep up with peak traffic, and keep heavy processing or IO done async outside of the subscriber, for example by using Task or Oban.

The jetstream instances themselves do not promise perfect uptime and have had several significant outages. If accuracy of data is important to you, ensure that you have a way of backfilling missing data.