How do I type Node.js streams in checked JavaScript?

The short answer

Let readline split a file into lines (each one a string), and put the processing step in pipeline() as an async generator whose @param says what it receives:

import { createReadStream, createWriteStream } from "node:fs";
import { createInterface } from "node:readline";
import { pipeline } from "node:stream/promises";

/**
 * @param {string} path
 * @returns {Promise<number>}
 */
export async function countOrders(path) {
  const lines = createInterface({ input: createReadStream(path, "utf8"), crlfDelay: Infinity });
  let count = 0;
  for await (const line of lines) {
    if (line.startsWith("A-")) count += 1;
  }
  return count;
}

/**
 * @param {AsyncIterable<string>} source
 * @returns {AsyncGenerator<string>}
 */
async function* upperCase(source) {
  for await (const chunk of source) yield chunk.toUpperCase();
}

await pipeline(createReadStream("orders.csv", "utf8"), upperCase, createWriteStream("orders-upper.csv"));
console.log(await countOrders("orders.csv"));

import { createReadStream, createWriteStream } from "node:fs";
import { createInterface } from "node:readline";
import { pipeline } from "node:stream/promises";

export async function countOrders(path: string): Promise<number> {
  const lines = createInterface({ input: createReadStream(path, "utf8"), crlfDelay: Infinity });
  let count = 0;
  for await (const line of lines) {
    if (line.startsWith("A-")) count += 1;
  }
  return count;
}

async function* upperCase(source: AsyncIterable<string>): AsyncGenerator<string> {
  for await (const chunk of source) yield chunk.toUpperCase();
}

await pipeline(createReadStream("orders.csv", "utf8"), upperCase, createWriteStream("orders-upper.csv"));
console.log(await countOrders("orders.csv"));

With the "utf8" encoding the file stream yields strings, so upperCase declares AsyncIterable<string> and the pipeline is checked end to end.

Why it works

readline.createInterface() returns an object that is async-iterable, and @types/node types each line as a string. pipeline() from node:stream/promises accepts a source stream, any number of steps, and a destination; an async generator function is a valid step, typed by its own JSDoc. Because the generator is a plain function, the same async generator typing applies.

The same thing in TypeScript

The TypeScript equivalent tab annotates upperCase(source: AsyncIterable<string>): AsyncGenerator<string>. The calls to node:fs, node:readline and node:stream/promises are the same in both files.

What goes wrong

Iterating a stream directly gives any chunks, and an any value hides mistakes. Narrowing it with Buffer.isBuffer shows what the type really could be: here a Buffer can reach a variable declared as a string (TS2322):

import { createReadStream } from "node:fs";

/** @param {string} path */
export async function firstLine(path) {
  for await (const chunk of createReadStream(path)) {
    /** @type {string} */
    const text = Buffer.isBuffer(chunk) ? chunk : String(chunk);error TS2322: Type 'string | Buffer<ArrayBufferLike>' is not assignable to type 'string'.
  Type 'Buffer<ArrayBufferLike>' is not assignable to type 'string'.
    return text.split("\n")[0];
  }
}

Pass an encoding when you create the stream (createReadStream(path, "utf8")) and convert explicitly where you need text.

Gotchas

• crlfDelay: Infinity matters for Windows files. It makes readline treat \r\n as one line break; the types do not change.
• Errors reject the pipeline. await pipeline(…) rejects if any stream fails, so one try/catch around it handles every stage.
• Object-mode streams are untyped too. A stream of parsed rows still yields any; give the consuming function a @param with the row type.