diff --git a/firehose-worker/src/firehose-consumer.ts b/firehose-worker/src/firehose-consumer.ts index 3b50910..f60467b 100644 --- a/firehose-worker/src/firehose-consumer.ts +++ b/firehose-worker/src/firehose-consumer.ts @@ -5,13 +5,19 @@ import { applyResignRecord, applyActionRecordAction, } from '../../src/lib/server/firehose-records'; +import { parseCommit } from './jetstream-event'; export interface Env { DB: D1Database; } -/** Jetstream v1 relays records for unofficial collections without registration. */ -const JETSTREAM_URL = 'wss://jetstream2.us-east.bsky.network/subscribe'; +/** + * Jetstream v2, which filters by event kind. The v1 endpoint we used before + * cannot: a `wantedCollections` filter there constrains commits only, and the + * network's identity and account events keep arriving regardless. + */ +const JETSTREAM_URL = + 'wss://jetstream.us-east.bsky.network/xrpc/network.bsky.jetstream.subscribeEvents'; const COLLECTIONS = [ 'boo.sky.go.game', @@ -31,18 +37,6 @@ const KEEPALIVE_MS = 10 * 60 * 1000; const CURSOR_KEY = 'jetstream:cursor'; -interface CommitMessage { - kind: string; - did: string; - time_us?: number; - commit?: { - operation: string; - collection: string; - rkey: string; - record: Record; - }; -} - export interface FirehoseStatus { connected: boolean; cursor: number | null; @@ -112,8 +106,13 @@ export class FirehoseConsumer extends DurableObject { const url = new URL(JETSTREAM_URL); for (const collection of COLLECTIONS) { - url.searchParams.append('wantedCollections', collection); + url.searchParams.append('collections', collection); } + // Commits only. A collections filter constrains commit events alone, so + // without this the socket also carries every identity and account event on + // the network — measured at ~0.8/s, and every message is a Durable Object + // request, which is enough on its own to burn the free tier's daily quota. + url.searchParams.set('kinds', 'commit'); // Resume where the previous run stopped; Jetstream retains ~36h of history. const cursor = await this.ctx.storage.get(CURSOR_KEY); @@ -159,25 +158,22 @@ export class FirehoseConsumer extends DurableObject { } async handle(raw: string): Promise { - let message: CommitMessage; - try { - message = JSON.parse(raw); - } catch { - return; - } + const commit = parseCommit(raw); + if (!commit) return; - if (message.kind !== 'commit' || !message.commit) return; - const { operation, collection, rkey, record } = message.commit; - if (operation !== 'create' && operation !== 'update') return; + // Record the position before doing the work: v2 delivers at-least-once after + // a reconnect, and replaying an event is harmless, while losing our place is + // not. + if (commit.seq !== null) await this.ctx.storage.put(CURSOR_KEY, commit.seq); - if (typeof message.time_us === 'number') { - await this.ctx.storage.put(CURSOR_KEY, message.time_us); - } + // Deletes carry no record, and the index only tracks records that exist. + const { collection, record } = commit; + if (!record) return; const platform = { env: this.env }; if (collection === 'boo.sky.go.game') { - await applyGameRecord(platform, message.did, rkey, record); + await applyGameRecord(platform, commit.did, commit.rkey, record); } else if (collection === 'boo.sky.go.move') { await applyActionRecord(platform, record, 'move'); } else if (collection === 'boo.sky.go.pass') { diff --git a/firehose-worker/src/jetstream-event.ts b/firehose-worker/src/jetstream-event.ts new file mode 100644 index 0000000..a5544b0 --- /dev/null +++ b/firehose-worker/src/jetstream-event.ts @@ -0,0 +1,59 @@ +/** + * Jetstream v2 events. + * + * v2 wraps each event in a `message` envelope, with the event itself under + * `payload` and tagged by `$type`, its fields flat inside it. The subscription + * filters are `collections` (repeatable), `dids` and `kinds`. + */ +export interface JetstreamCommit { + did: string; + operation: 'create' | 'update' | 'delete'; + collection: string; + rkey: string; + /** Null for deletes, which carry no record. */ + record: Record | null; + /** Sequence number of the event, for resuming after a reconnect. */ + seq: number | null; +} + +interface Envelope { + payload?: { + $type?: string; + did?: string; + seq?: number; + operation?: string; + collection?: string; + rkey?: string; + record?: Record; + }; +} + +/** + * A commit event, or null for anything else — other event kinds, unparseable + * frames, and commits missing the fields the indexers need. + */ +export function parseCommit(raw: string): JetstreamCommit | null { + let envelope: Envelope; + try { + envelope = JSON.parse(raw) as Envelope; + } catch { + return null; + } + + const payload = envelope?.payload; + if (!payload || typeof payload.$type !== 'string') return null; + if (!payload.$type.endsWith('#commit')) return null; + if (!payload.did || !payload.collection || !payload.rkey) return null; + + const { operation } = payload; + if (operation !== 'create' && operation !== 'update' && operation !== 'delete') return null; + + return { + did: payload.did, + operation, + collection: payload.collection, + rkey: payload.rkey, + record: payload.record ?? null, + seq: typeof payload.seq === 'number' ? payload.seq : null, + }; +}