// Initially forked from @lukas.sampleFirehose export let firehoseReadMessagesExample = (async () => { 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; }; // byFrame removed, logic moved below // byMessage removed, logic moved below const socket = new WebSocket( "wss://bsky.social/xrpc/com.atproto.sync.subscribeRepos", ); socket.addEventListener("open", (event) => { console.log("socket open", event); }); socket.addEventListener("error", (event) => { throw `socket error ${event}`; }); // Read up to X messages const messageLimit = 20; // Only read for X milliseconds const readForXms = 5000; const messages = []; await new Promise((res, rej) => { setTimeout(() => res(messages), readForXms); socket.addEventListener("message", (event) => { const reader = new FileReader(); reader.onload = function () { const arrayBuffer = reader.result; const uint8Array = new Uint8Array(arrayBuffer); const frame = Frame.fromBytes(uint8Array); if (frame.isMessage()) { messages.push(frame); } else if (frame.isError()) { rej(xrpc.XRPCError(-1, frame.code, frame.message)); } else { rej( new xrpc.XRPCError( xrpc.ResponseType.Unknown, undefined, "Unknown frame type", ), ); } if (messages.length === messageLimit) { res(messages); } }; reader.readAsArrayBuffer(event.data); }); }); console.log("number of messages", messages.length); console.log(messages); })();