export async function sampleFirehose() { const cborx = await import("https://deno.land/x/cbor@v1.5.2/index.js"); const multiformats = await import("npm:multiformats"); const uint8arrays = await import("npm:uint8arrays"); const { z } = await import("https://deno.land/x/zod@v3.21.4/mod.ts"); const xrpc = await import("npm:@atproto/xrpc"); const cborCodec = await import( "https://cdn.jsdelivr.net/npm/@ipld/dag-cbor/+esm" ); const cborEncode = cborCodec.encode; enum FrameType { Message = 1, Error = -1, } const messageFrameHeader = z.object({ op: z.literal(FrameType.Message), t: z.string().optional(), // Message body type discriminator }); type MessageFrameHeader = z.infer; const errorFrameHeader = z.object({ op: z.literal(FrameType.Error), }); const errorFrameBody = z.object({ error: z.string(), message: z.string().optional(), // Error message }); type ErrorFrameHeader = z.infer; type ErrorFrameBody = { error: T; } & z.infer; const frameHeader = z.union([messageFrameHeader, errorFrameHeader]); type FrameHeader = z.infer; abstract class Frame { header: FrameHeader; body: unknown; get op(): FrameType { return this.header.op; } toBytes(): Uint8Array { return uint8arrays.concat([ cborEncode(this.header), cborEncode(this.body), ]); } isMessage(): this is MessageFrame { return this.op === FrameType.Message; } isError(): this is ErrorFrame { return this.op === FrameType.Error; } static fromBytes(bytes: Uint8Array) { const decoded = cborDecodeMulti(bytes); if (decoded.length > 2) { throw new Error("Too many CBOR data items in frame"); } const header = decoded[0]; let body: unknown = kUnset; if (decoded.length > 1) { body = decoded[1]; } const parsedHeader = frameHeader.safeParse(header); if (!parsedHeader.success) { throw new Error(`Invalid frame header: ${parsedHeader.error.message}`); } if (body === kUnset) { throw new Error("Missing frame body"); } const frameOp = parsedHeader.data.op; if (frameOp === FrameType.Message) { return new MessageFrame(body, { type: parsedHeader.data.t, }); } else if (frameOp === FrameType.Error) { const parsedBody = errorFrameBody.safeParse(body); if (!parsedBody.success) { throw new Error( `Invalid error frame body: ${parsedBody.error.message}`, ); } return new ErrorFrame(parsedBody.data); } else { const exhaustiveCheck: never = frameOp; throw new Error(`Unknown frame op: ${exhaustiveCheck}`); } } } class MessageFrame> extends Frame { header: MessageFrameHeader; body: T; constructor(body: T, opts?: { type?: string; }) { super(); this.header = opts?.type !== undefined ? { op: FrameType.Message, t: opts?.type } : { op: FrameType.Message }; this.body = body; } get type() { return this.header.t; } } class ErrorFrame extends Frame { header: ErrorFrameHeader; body: ErrorFrameBody; constructor(body: ErrorFrameBody) { super(); this.header = { op: FrameType.Error }; this.body = body; } get code() { return this.body.error; } get message() { return this.body.message; } } const kUnset = Symbol("unset"); const { CID } = multiformats; // add extension for decoding CIDs // decoding code taken from $ipld/dag-cbor // does not support encoding cids cborx.addExtension({ Class: CID, tag: 42, encode: () => { throw new Error("cannot encode cids"); }, decode: (bytes: Uint8Array) => { if (bytes[0] !== 0) { throw new Error("Invalid CID for CBOR tag 42; expected leading 0x00"); } return CID.decode(bytes.subarray(1)); // ignore leading 0x00 }, }); const cborDecodeMulti = (encoded: Uint8Array): unknown[] => { const decoded: unknown[] = []; cborx.decodeMultiple(encoded, (value) => { decoded.push(value); }); return decoded; }; async function* byFrame(ws: WebSocketStream) { const { readable } = await ws.connection; const reader = readable.getReader(); while (true) { const { value, done } = await reader.read(); if (done) { break; } if (value instanceof Uint8Array) { yield Frame.fromBytes(value); } } } async function* byMessage(ws: WebSocketStream) { for await (const frame of byFrame(ws)) { if (frame.isMessage()) { yield frame; } else if (frame.isError()) { throw new xrpc.XRPCError(-1, frame.code, frame.message); } else { throw new xrpc.XRPCError( xrpc.ResponseType.Unknown, undefined, "Unknown frame type", ); } } } const ws = new WebSocketStream( "wss://bsky.social/xrpc/com.atproto.sync.subscribeRepos", ); return { ws, byMessage }; }

sampleFirehose

Version 0