Read status updates from Mastodon streaming API

stream.ts

type StreamEvent =
  | { event: "update"; status: Status }
  | { event: "status.update"; status: Status }
  | { event: "delete"; id: string; status?: never };
  
/**
 * Use like this:
 *   const { events } = await stream(`https://${instance}/api/v1/streaming/public`);
 *   for await (const { id, status } of events) {
 *     if (status) … do something …
 *     else if (id) … deleted …
 *   }
 */
async function stream(url: string): Promise<{
  events: AsyncGenerator<StreamEvent, void, unknown>;
  close: () => void;
}> {
  const response = await fetch(url);
  if (!response.ok) throw new Error(`Stream failed: ${response.statusText}`);

  const reader = response.body!.getReader();
  const utf8decoder = new TextDecoder();

  function close() {
    reader.cancel();
    clearTimeout(timeout);
  }

  // Server sends heartbeat every 15 seconds
  let timeout = setTimeout(close, 20_000);
  async function* iterator(): AsyncGenerator<StreamEvent, void, unknown> {
    try {
      let buffer = "";
      while (true) {
        const { done, value } = await reader.read();
        if (done) throw new Error("Stream closed");

        const chunk = utf8decoder.decode(value);
        buffer += chunk;
        let message;
        while ((message = buffer.match(/^(.|\n)*\n{2}/m)?.[0])) {
          yield* yieldEvent(message);

          clearTimeout(timeout);
          timeout = setTimeout(close, 20_000);
          buffer = buffer.slice(message.length);
        }
      }
    } finally {
      clearTimeout(timeout);
    }
  }

  return { events: iterator(), close };
}

const serializer = new SerializerNativeImpl();

function* yieldEvent(message: string): Generator<StreamEvent, void, unknown> {
  // Message may or may not contain event, eg :thump is just a heartbeat
  // message.
  const event = message.match(/^event: (.*)$/m)?.[1];
  const data = message.match(/^data: (.*)$/m)?.[1];
  if ((event === "update" || event === "status.update") && data) {
    const status = serializer.deserialize<Status>("application/json", data);
    yield { event, status };
  } else if (event === "delete" && data) yield { event, id: JSON.parse(data) };
}