Fan out a stream to multiple Iceberg tab·续

Fan out a stream to multiple Iceberg tab·续

export class JetstreamConsumer extends DurableObject { ws = null; buf = []; lastFlush = 0; cursor = null; flushing = false;

// Arm the reconnect watchdog first, then connect (idempotent). async start() { await this.ctx.storage.setAlarm(Date.now() + RECONNECT_MS); await this.ensureConnected(); return { connected: this.ws !== null }; }

async ensureConnected() { if (this.ws) return; this.cursor ??= (await this.ctx.storage.get("cursor")) ?? null;

const params = new URLSearchParams(); for (const c of WANTED) params.append("wantedCollections", c); if (this.cursor) params.set("cursor", String(this.cursor));

ws.addEventListener("message", (e) => this.onMessage(e)); ws.addEventListener("close", () => (this.ws = null)); ws.addEventListener("error", () => (this.ws = null)); }

onMessage(e) { let ev; try { ev = JSON.parse(e.data); } catch { return; } if (typeof ev.time_us === "number") this.cursor = ev.time_us; const row = toRow(ev); if (row) this.buf.push(row); if ( this.buf.length >= FLUSH_MAX || Date.now() - this.lastFlush >= FLUSH_MS ) { void this.flush(); } }

// Serialize sends: flush one batch at a time, advancing the cursor on success. async flush() { if (this.flushing) return; this.flushing = true; try { while (this.buf.length > 0) { this.lastFlush = Date.now(); const batch = this.buf.splice(0, this.buf.length); const batchCursor = this.cursor; try { await this.env.BSKY_STREAM.send(batch); await this.ctx.storage.put("cursor", batchCursor); } catch (err) { this.buf.unshift(...batch); console.error("send failed, will retry", err); return; } } } finally { this.flushing = false; } }

// Watchdog: reconnect if dropped, flush stragglers, always reschedule. async alarm() { try { await this.ensureConnected(); await this.flush(); } catch (err) { console.error("alarm error", err); } finally { await this.ctx.storage.setAlarm(Date.now() + RECONNECT_MS); } } }

export default { async fetch(_req, env) { const stub = env.JETSTREAM.get(env.JETSTREAM.idFromName("singleton")); return Response.json(await stub.start()); }, async scheduled(_event, env) { const stub = env.JETSTREAM.get(env.JETSTREAM.idFromName("singleton")); await stub.start(); }, };


*src/index.tsts*

```ts
import { DurableObject } from "cloudflare:workers";
import type { Pipeline } from "cloudflare:pipelines";

interface Env {
    BSKY_STREAM: Pipeline;
    JETSTREAM: DurableObjectNamespace&lt;JetstreamConsumer>;
}

// Jetstream collection -> our short event_type. Only these are kept.
const COLLECTION_TO_TYPE: Record&lt;string, string> = {
    "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: any) {
    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,
    };
}

export class JetstreamConsumer extends DurableObject&lt;Env> {
    private ws: WebSocket | null = null;
    private buf: Record&lt;string, unknown>[] = [];
    private lastFlush = 0;
    private cursor: number | null = null;
    private flushing = false;

// Arm the reconnect watchdog first, then connect (idempotent).
    async start() {
        await this.ctx.storage.setAlarm(Date.now() + RECONNECT_MS);
        await this.ensureConnected();
        return { connected: this.ws !== null };
    }

private async ensureConnected() {
        if (this.ws) return;
        this.cursor ??= (await this.ctx.storage.get&lt;number>("cursor")) ?? null;

const params = new URLSearchParams();
        for (const c of WANTED) params.append("wantedCollections", c);
        if (this.cursor) params.set("cursor", String(this.cursor));

const resp = await fetch(`${JETSTREAM_URL}?${params}`, {
            headers: { Upgrade: "websocket" },
        });
        const ws = resp.webSocket;
        if (!ws) throw new Error(`Jetstream handshake failed: ${resp.status}`);
        ws.accept();
        this.ws = ws;

ws.addEventListener("message", (e) => this.onMessage(e));
        ws.addEventListener("close", () => (this.ws = null));
        ws.addEventListener("error", () => (this.ws = null));
    }

private onMessage(e: MessageEvent) {
        let ev: any;
        try {
            ev = JSON.parse(e.data as string);
        } catch {
            return;
        }
        if (typeof ev.time_us === "number") this.cursor = ev.time_us;
        const row = toRow(ev);
        if (row) this.buf.push(row);
        if (
            this.buf.length >= FLUSH_MAX ||
            Date.now() - this.lastFlush >= FLUSH_MS
        ) {
            void this.flush();
        }
    }

// Serialize sends: flush one batch at a time, advancing the cursor on success.
    private async flush() {
        if (this.flushing) return;
        this.flushing = true;
        try {
            while (this.buf.length > 0) {
                this.lastFlush = Date.now();
                const batch = this.buf.splice(0, this.buf.length);
                const batchCursor = this.cursor;
                try {
                    await this.env.BSKY_STREAM.send(batch);
                    await this.ctx.storage.put("cursor", batchCursor);
                } catch (err) {
                    this.buf.unshift(...batch);
                    console.error("send failed, will retry", err);
                    return;
                }
            }
        } finally {
            this.flushing = false;
        }
    }

// Watchdog: reconnect if dropped, flush stragglers, always reschedule.
    async alarm() {
        try {
            await this.ensureConnected();
            await this.flush();
        } catch (err) {
            console.error("alarm error", err);
        } finally {
            await this.ctx.storage.setAlarm(Date.now() + RECONNECT_MS);
        }
    }

}

export default {
    async fetch(_req, env): Promise&lt;Response> {
        const stub = env.JETSTREAM.get(env.JETSTREAM.idFromName("singleton"));
        return Response.json(await stub.start());
    },
    async scheduled(_event, env): Promise&lt;void> {
        const stub = env.JETSTREAM.get(env.JETSTREAM.idFromName("singleton"));
        await stub.start();
    },
} satisfies ExportedHandler&lt;Env>;

Generate types for your bindings:

npmyarnpnpm

npx wrangler types

yarn wrangler types

pnpm wrangler types

7. Deploy and start the consumer

Deploy the Worker:

npmyarnpnpm

npx wrangler deploy

yarn wrangler deploy

pnpm wrangler deploy

Open the Worker URL once to start the firehose. The cron trigger keeps it running:

curl https://bluesky-pipeline.YOUR_SUBDOMAIN.workers.dev

The command returns:

{ "connected": true }

Tail the logs to watch it work:

npmyarnpnpm

npx wrangler tail

yarn wrangler tail

pnpm wrangler tail

8. Query the tables with Basin SQL

The first data lands a few minutes after the first events arrive, while the pipeline warms up.

Set your Basin SQL token, then query each table. Replace YOUR_WAREHOUSE_NAME with the warehouse name you noted in step 3.

export WRANGLER_BASIN_SQL_AUTH_TOKEN=YOUR_CATALOG_TOKEN

Count the rows in the likes table:

npmyarnpnpm

npx wrangler basin sql query "YOUR_WAREHOUSE_NAME" "SELECT COUNT(*) FROM bluesky.bsky_like"

yarn wrangler basin sql query "YOUR_WAREHOUSE_NAME" "SELECT COUNT(*) FROM bluesky.bsky_like"

pnpm wrangler basin sql query "YOUR_WAREHOUSE_NAME" "SELECT COUNT(*) FROM bluesky.bsky_like"

Query English-language posts:

npmyarnpnpm

npx wrangler basin sql query "YOUR_WAREHOUSE_NAME" "SELECT text, langs FROM bluesky.bsky_post WHERE langs LIKE '%en%' LIMIT 10"

yarn wrangler basin sql query "YOUR_WAREHOUSE_NAME" "SELECT text, langs FROM bluesky.bsky_post WHERE langs LIKE '%en%' LIMIT 10"

pnpm wrangler basin sql query "YOUR_WAREHOUSE_NAME" "SELECT text, langs FROM bluesky.bsky_post WHERE langs LIKE '%en%' LIMIT 10"

Each table contains only its event type, projected to the relevant columns. The single pipeline did all the routing.

Conclusion

You consumed a high-velocity public WebSocket firehose with a Durable Object, ingested it into one Basin Pipelines stream, and used a single pipeline with multiple SQL statements to fan the stream out into five Iceberg tables by event type.

This one-stream-to-many-tables pattern generalizes to any tagged event source: clickstreams (by event_type), logs (by service or status), or IoT telemetry (by device_class). To extend it, add a sink and a matching INSERT ... WHERE statement.

To learn more about the SQL used here, refer to SELECT statements and Manage pipelines.

Was this helpful?

YesNo

{"@context":"https://schema.org","@type":"TechArticle","@id":"https://developers.cloudflare.com/basin-pipelines/examples/bluesky-firehose-fanout/#page","headline":"Fan out a stream to multiple Iceberg tables","description":"Route Bluesky Jetstream events into multiple Basin Catalog tables using a single Basin Pipelines stream and multiple SQL statements.","url":"https://developers.cloudflare.com/basin-pipelines/examples/bluesky-firehose-fanout/","inLanguage":"en","image":"https://developers.cloudflare.com/basin-pipelines/examples/bluesky-firehose-fanout/og.png?v=447a006739af8be5","dateModified":"2026-10-12","publisher":{"@type":"Organization","name":"Cloudflare","description":"One platform for your apps, agents, and workforce. Build, secure, and scale without managing infrastructure","url":"https://www.cloudflare.com/","sameAs":["https://github.com/cloudflare","https://www.linkedin.com/company/cloudflare","https://x.com/cloudflare"],"logo":{"@type":"ImageObject","url":"https://developers.cloudflare.com/logo.svg"},"address":{"@type":"PostalAddress","streetAddress":"101 Townsend St","addressLocality":"San Francisco","addressRegion":"CA","postalCode":"94107","addressCountry":"US"},"contactPoint":[{"@type":"ContactPoint","contactType":"Customer Support","url":"https://support.cloudflare.com/","availableLanguage":["English"]},{"@type":"ContactPoint","contactType":"Sales","url":"https://www.cloudflare.com/contact/","availableLanguage":["English"]}]},"isPartOf":{"@type":"WebSite","@id":"https://developers.cloudflare.com/#website","name":"Cloudflare Docs","url":"https://developers.cloudflare.com/"},"keywords":["TypeScript"]}

description: Migration guide for pipelines created before September 2025 to the new Basin Pipelines architecture. title: Legacy pipelines image: https://developers.cloudflare.com/basin-pipelines/reference/legacy-pipelines/og.png?v=dca9a67a548c1471

Documentation Index
Fetch the complete documentation index at: https://developers.cloudflare.com/basin-pipelines/llms.txt
Use this file to discover all available pages before exploring further.

Fan out a stream to multiple Iceberg tab·续

Fan out a stream to multiple Iceberg tab·续

7. Deploy and start the consumer

8. Query the tables with Basin SQL

Conclusion

On this page