Fan out a stream to multiple Iceberg tables

Fan out a stream to multiple Iceberg tables

Consume the Bluesky Jetstream firehose in a Durable Object, ingest it into a single Basin Pipelines stream, and use one pipeline with multiple SQL statements to route events into separate Basin Catalog tables by type.

Last updated Oct 12, 2026 In this example, you will consume the public Bluesky Jetstream ↗︎ firehose, a live WebSocket stream of every post, like, repost, follow, and block on the network, and land it in Basin Catalog as queryable Apache Iceberg tables.

You will learn a core Basin Pipelines pattern: send every event to one stream, then use a single pipeline with multiple SQL statements to route ("fan out") that stream into several destination tables, one per event type, without running a separate pipeline for each.

flowchart TD
    A[Bluesky Jetstream WebSocket] --> B[Durable Object]
    B -->|send| C[bsky_events_stream]
    C --> D[bsky_pipeline]
    D --> E[bsky_post]
    D --> F[bsky_like]
    D --> G[bsky_repost]
    D --> H[bsky_follow]
    D --> I[bsky_block]

Prerequisites

Sign up for a Cloudflare account ↗︎.
Install Node.js ↗︎.

Node.js version manager

</summary>

Use <a href="https://viteplus.dev/guide/env">Vite+ (<code>vp env</code>) ↗︎</a> to install and manage Node.js versions.

For supported Node.js versions, refer to the <a href="https://developers.cloudflare.com/workers/wrangler/install-and-update/#install-wrangler">Wrangler system requirements</a>.

</details>

You will also need an R2 API token with Admin Read & Write permissions, which includes Basin Catalog and Basin SQL access. You will pass this token to each sink. No Bluesky account or API key is required, as Jetstream is public and unauthenticated.

1. Create a new Worker project

Create a new Worker project by running the following command:

npmyarnpnpm

npm create cloudflare@latest -- bluesky-pipeline

yarn create cloudflare bluesky-pipeline

pnpm create cloudflare@latest bluesky-pipeline

For setup, select the following options:

For What would you like to start with?, choose Hello World example.
For Which template would you like to use?, choose Worker only.
For Which language do you want to use?, choose TypeScript.
For Do you want to use git for version control?, choose Yes.
For Do you want to add an AGENTS.md file to help AI coding tools understand Cloudflare APIs?, choose Yes.
For Do you want to deploy your application?, choose No (we will be making some changes before deploying).

Change into your new project directory:

cd bluesky-pipeline

The pipelines commands used later require Wrangler v4 or later. If your project was scaffolded with an older version, update it now:

npmyarnpnpmbun

npm i -D wrangler@latest

yarn add -D wrangler@latest

pnpm add -D wrangler@latest

bun add -d wrangler@latest

2. Define the stream schema

A stream has a single schema. Because every event type flows through the same stream, the schema is the union of the fields you want across all types. Each routing statement later selects only the columns relevant to its destination.

Create a schema.json file in the root of your project:

{
    "fields": [
        { "name": "event_id", "type": "string", "required": true },
        { "name": "event_type", "type": "string", "required": true },
        { "name": "did", "type": "string", "required": false },
        { "name": "operation", "type": "string", "required": false },
        { "name": "event_time", "type": "timestamp", "required": false },
        { "name": "created_at", "type": "string", "required": false },
        { "name": "text", "type": "string", "required": false },
        { "name": "langs", "type": "string", "required": false },
        { "name": "subject_uri", "type": "string", "required": false },
        { "name": "subject_did", "type": "string", "required": false }
    ]
}

3. Create an R2 bucket and enable Basin Catalog

Your sinks write to Iceberg tables in Basin Catalog, so you need a bucket with the catalog enabled.

Create an R2 bucket named bluesky-pipeline:

npmyarnpnpm

npx wrangler r2 bucket create bluesky-pipeline

yarn wrangler r2 bucket create bluesky-pipeline

pnpm wrangler r2 bucket create bluesky-pipeline

Enable Basin Catalog on the bucket:

npmyarnpnpm

npx wrangler basin catalog enable bluesky-pipeline

yarn wrangler basin catalog enable bluesky-pipeline

pnpm wrangler basin catalog enable bluesky-pipeline

When you run this command, note the Warehouse name. You will need it to query your data with Basin SQL.

4. Create the stream, sinks, and pipeline

First, create the stream from your schema file:

npmyarnpnpm

npx wrangler basin pipelines streams create bsky_events_stream --schema-file schema.json

yarn wrangler basin pipelines streams create bsky_events_stream --schema-file schema.json

pnpm wrangler basin pipelines streams create bsky_events_stream --schema-file schema.json

Note the stream ID in the output. You will use it to configure the Worker binding in the next step.

Next, create one sink per destination table. Each sink writes to its own Iceberg table in Basin Catalog. Replace YOUR_CATALOG_TOKEN with your R2 API token.

npmyarnpnpm

npx wrangler basin pipelines sinks create bsky_post_sink --type basin-catalog --bucket bluesky-pipeline --namespace bluesky --table bsky_post --catalog-token YOUR_CATALOG_TOKEN --roll-interval 60

yarn wrangler basin pipelines sinks create bsky_post_sink --type basin-catalog --bucket bluesky-pipeline --namespace bluesky --table bsky_post --catalog-token YOUR_CATALOG_TOKEN --roll-interval 60

pnpm wrangler basin pipelines sinks create bsky_post_sink --type basin-catalog --bucket bluesky-pipeline --namespace bluesky --table bsky_post --catalog-token YOUR_CATALOG_TOKEN --roll-interval 60

npmyarnpnpm

npx wrangler basin pipelines sinks create bsky_like_sink --type basin-catalog --bucket bluesky-pipeline --namespace bluesky --table bsky_like --catalog-token YOUR_CATALOG_TOKEN --roll-interval 60

yarn wrangler basin pipelines sinks create bsky_like_sink --type basin-catalog --bucket bluesky-pipeline --namespace bluesky --table bsky_like --catalog-token YOUR_CATALOG_TOKEN --roll-interval 60

pnpm wrangler basin pipelines sinks create bsky_like_sink --type basin-catalog --bucket bluesky-pipeline --namespace bluesky --table bsky_like --catalog-token YOUR_CATALOG_TOKEN --roll-interval 60

npmyarnpnpm

npx wrangler basin pipelines sinks create bsky_repost_sink --type basin-catalog --bucket bluesky-pipeline --namespace bluesky --table bsky_repost --catalog-token YOUR_CATALOG_TOKEN --roll-interval 60

yarn wrangler basin pipelines sinks create bsky_repost_sink --type basin-catalog --bucket bluesky-pipeline --namespace bluesky --table bsky_repost --catalog-token YOUR_CATALOG_TOKEN --roll-interval 60

pnpm wrangler basin pipelines sinks create bsky_repost_sink --type basin-catalog --bucket bluesky-pipeline --namespace bluesky --table bsky_repost --catalog-token YOUR_CATALOG_TOKEN --roll-interval 60

npmyarnpnpm

npx wrangler basin pipelines sinks create bsky_follow_sink --type basin-catalog --bucket bluesky-pipeline --namespace bluesky --table bsky_follow --catalog-token YOUR_CATALOG_TOKEN --roll-interval 60

yarn wrangler basin pipelines sinks create bsky_follow_sink --type basin-catalog --bucket bluesky-pipeline --namespace bluesky --table bsky_follow --catalog-token YOUR_CATALOG_TOKEN --roll-interval 60

pnpm wrangler basin pipelines sinks create bsky_follow_sink --type basin-catalog --bucket bluesky-pipeline --namespace bluesky --table bsky_follow --catalog-token YOUR_CATALOG_TOKEN --roll-interval 60

npmyarnpnpm

npx wrangler basin pipelines sinks create bsky_block_sink --type basin-catalog --bucket bluesky-pipeline --namespace bluesky --table bsky_block --catalog-token YOUR_CATALOG_TOKEN --roll-interval 60

yarn wrangler basin pipelines sinks create bsky_block_sink --type basin-catalog --bucket bluesky-pipeline --namespace bluesky --table bsky_block --catalog-token YOUR_CATALOG_TOKEN --roll-interval 60

pnpm wrangler basin pipelines sinks create bsky_block_sink --type basin-catalog --bucket bluesky-pipeline --namespace bluesky --table bsky_block --catalog-token YOUR_CATALOG_TOKEN --roll-interval 60

Now create one pipeline whose SQL contains multiple INSERT statements, one per route. Each statement filters the stream by event_type and projects only the columns that matter for its table.

Create a fanout.sql file:

INSERT INTO bsky_post_sink
  SELECT event_id, did, operation, event_time, created_at, text, langs
  FROM bsky_events_stream WHERE event_type = 'post';

INSERT INTO bsky_like_sink
  SELECT event_id, did, operation, event_time, created_at, subject_uri
  FROM bsky_events_stream WHERE event_type = 'like';

INSERT INTO bsky_repost_sink
  SELECT event_id, did, operation, event_time, created_at, subject_uri
  FROM bsky_events_stream WHERE event_type = 'repost';

INSERT INTO bsky_follow_sink
  SELECT event_id, did, operation, event_time, created_at, subject_did
  FROM bsky_events_stream WHERE event_type = 'follow';

INSERT INTO bsky_block_sink
  SELECT event_id, did, operation, event_time, created_at, subject_did
  FROM bsky_events_stream WHERE event_type = 'block';

Create the pipeline from the file:

npmyarnpnpm

npx wrangler basin pipelines create bsky_pipeline --sql-file fanout.sql

yarn wrangler basin pipelines create bsky_pipeline --sql-file fanout.sql

pnpm wrangler basin pipelines create bsky_pipeline --sql-file fanout.sql

One pipeline writes to five tables. To add a new event type later, add one sink and one INSERT statement. Pipeline SQL cannot be modified after creation, so you delete and recreate the pipeline to change it. To learn more, refer to Route one stream to multiple tables.

5. Bind the stream to your Worker

Add the stream binding, a Durable Object to hold the WebSocket connection, and a cron trigger to keep the consumer alive. Replace <STREAM_ID> with the stream ID from step 4.

{
  "$schema": "./node_modules/wrangler/config-schema.json",
  "name": "bluesky-pipeline",
  "main": "src/index.ts",
  "upload_source_maps": true,
  // Set this to today's date
  "compatibility_date": "2026-10-27",
  "pipelines": [
    {
      "binding": "BSKY_STREAM",
      "stream": "<STREAM_ID>"
    }
  ],
  "durable_objects": {
    "bindings": [
      {
        "name": "JETSTREAM",
        "class_name": "JetstreamConsumer"
      }
    ]
  },
  "migrations": [
    {
      "tag": "v1",
      "new_sqlite_classes": [
        "JetstreamConsumer"
      ]
    }
  ],
  "triggers": {
    "crons": [
      "*/2 * * * *"
    ]
  }
}

name = "bluesky-pipeline"
main = "src/index.ts"
upload_source_maps = true
# Set this to today's date
compatibility_date = "2026-10-27"

[[pipelines]]
binding = "BSKY_STREAM"
stream = "<STREAM_ID>"

[[durable_objects.bindings]]
name = "JETSTREAM"
class_name = "JetstreamConsumer"

[[migrations]]
tag = "v1"
new_sqlite_classes = ["JetstreamConsumer"]

[triggers]
crons = ["*/2 * * * *"]

6. Consume the firehose in a Durable Object

A Durable Object is the right home for a long-lived WebSocket. It stays resident while the socket is open, and an alarm reconnects it if the connection drops. Buffer incoming events and send() them to the stream in batches to stay under the 5 MB per request limit. Persist the Jetstream time_us cursor so a reconnect resumes without gaps.

Replace the contents of src/index.ts with the following:

src/index.jsjs

```js import { DurableObject } from "cloudflare:workers";

// Jetstream collection -> our short event_type. Only these are kept. const COLLECTION_TO_TYPE = { "app.bsky.feed.post": "post", "app.bsky.feed.like": "like", "app.bsky.feed.repost": "repost", "app.bsky.graph.follow": "follow", "app.bsky.graph.block": "block", }; const WANTED = Object.keys(COLLECTION_TO_TYPE); const JETSTREAM_URL = "https://jetstream2.us-east.bsky.network/subscribe"; const FLUSH_MAX = 500; // rows per send() const FLUSH_MS = 1000; // flush at least once per second const RECONNECT_MS = 15000;

// Flatten one Jetstream message into a unified stream row, or null to skip. function toRow(ev) { if (ev?.kind !== "commit" || !ev.commit) return null; const c = ev.commit; const event_type = COLLECTION_TO_TYPE[c.collection]; if (!event_type) return null; const r = c.record ?? {}; const subject = r.subject; return { event_id: ${ev.did}/${c.collection}/${c.rkey}, event_type, did: ev.did ?? null, operation: c.operation ?? null, event_time: typeof ev.time_us === "number" ? new Date(ev.time_us / 1000).toISOString() : null, created_at: typeof r.createdAt === "string" ? r.createdAt : null, text: event_type === "post" && typeof r.text === "string" ? r.text : null, langs: event_type === "post" && Array.isArray(r.langs) ? r.langs.join(",") : null, subject_uri: typeof subject === "object" ? (subject?.uri ?? null) : null, subject_did: typeof subject === "string" ? subject : null, }; }