Streams, events and uploads

A handler may return a stream instead of a value: a Node Readable, a web ReadableStream or any async iterable. It goes out piece by piece, as it comes.

Server-sent events have a contract like any other answer, one schema per event:

import { sse, t } from "@vxnsin/inkan";

app.get(
  "/clock",
  {
    response: { 200: t.events({ tick: t.object({ n: t.int() }) }) },
    examples: [{ name: "three ticks", expect: [{ data: { n: 1 } }, { data: { n: 2 } }, { data: { n: 3 } }] }],
  },
  () =>
    sse(async function* (signal) {
      for (let n = 1; !signal.aborted; n++) {
        yield { event: "tick", data: { n } };
        await sleep(1000);
      }
    }),
);

The signal fires when the client goes away, so the loop stops. Quiet streams get a keep-alive comment every 15 seconds. In development, an event that breaks its schema ends the stream with an error event and a line on the console. inkan check and the docs page read as many events as an example expects, so a stream that never ends can still be an example.

Uploads are a t.file() in the body. A body with a file is read as multipart/form-data, and OpenAPI says so:

import { fileExample, t } from "@vxnsin/inkan";

app.post(
  "/avatars",
  {
    body: t.object({ user: t.int(), image: t.file().max(200_000).accept("image/png", "image/jpeg") }),
    examples: [{ name: "a tiny png", body: { user: 1, image: fileExample("me.png", "…", "image/png") } }],
  },
  ({ body }) => save(body.user, body.image.data), // name, type, size and data: a Buffer
);

Fields next to the file are coerced like a query ("1" becomes 1). A file that is too big or of the wrong type is a 400 that says which, and bodyLimit still caps the whole upload.

A file as the body itself, as fetch(url, { body: file }) sends it, is a t.binary(): the bytes, unparsed, with a limit and media types of their own. Its max is that route's body limit, so it can be larger than the app's:

app.post("/pictures", { body: t.binary().max(25_000_000).accept("image/*") }, ({ body, headers }) =>
  store(body, headers["content-type"]), // body: a Buffer
);

Larger than memory, a t.stream() hands the body to the handler unread, chunk by chunk as it arrives. Past its max the stream throws a 413 problem; an answer before the body is read, a 401 say, still reaches the client.

import { pipeline } from "node:stream/promises";

app.post("/backups", { body: t.stream().max(2 * 1024 ** 3) }, async ({ body }) => {
  await pipeline(body, createWriteStream(path)); // never all in memory
  return reply(201, { stored: true });
});

Any route can also say bodyLimit: n for itself. The typed client sends a Blob, bytes or a ReadableStream as they are, and the docs list the media types.