Große Datenmengen in Node.js streamen mit Backpressure

// ❌ Loads the entire table into a JS array before the first byte is sent
export async function GET(req: Request) {
  const users = await db.query<User>(
    "SELECT * FROM users WHERE tenant_id = $1",
    [req.tenantId],
  );
  return Response.json(users); // serializes all rows at once — heap spikes with result size
}
 
// ✅ Streams rows as they arrive — heap usage stays flat regardless of row count
export async function GET(req: Request) {
  const rows = db.queryStream<User>(
    "SELECT * FROM users WHERE tenant_id = $1",
    [req.tenantId],
  );
  return new Response(serializeNDJSON(rows), {
    headers: { "Content-Type": "application/x-ndjson" },
  });
}

import { pipeline, Transform } from "node:stream";
import { promisify } from "node:util";
 
const pipelineAsync = promisify(pipeline);
 
async function exportUsersToResponse(
  queryStream: NodeJS.ReadableStream,
  res: NodeJS.WritableStream,
): Promise<void> {
  const ndJsonTransform = new Transform({
    objectMode: true,
    transform(row: unknown, _encoding, callback) {
      try {
        this.push(JSON.stringify(row) + "\n");
        callback();
      } catch (err) {
        callback(err as Error);
      }
    },
  });
 
  // All three streams are destroyed together on error or completion
  await pipelineAsync(queryStream, ndJsonTransform, res);
}

// ❌ Ignores backpressure — the writable's internal buffer grows without bound
readable.on("data", (chunk) => {
  writable.write(chunk); // return value silently discarded
});
 
// ✅ Respects backpressure — pauses the source when the consumer's buffer is full
readable.on("data", (chunk) => {
  const canContinue = writable.write(chunk);
  if (!canContinue) {
    readable.pause();
    writable.once("drain", () => readable.resume());
  }
});

async function* transformRows<T, R>(
  source: AsyncIterable<T>,
  transform: (row: T) => R | Promise<R>,
): AsyncGenerator<R> {
  for await (const row of source) {
    yield await transform(row);
  }
}
 
async function* serializeToNDJSON<T>(source: AsyncIterable<T>): AsyncGenerator<string> {
  for await (const item of source) {
    yield JSON.stringify(item) + "\n";
  }
}

async function streamExport(tenantId: string, res: ServerResponse): Promise<void> {
  const rows = db.queryStream<RawUser>(
    "SELECT id, email, name, created_at FROM users WHERE tenant_id = $1 ORDER BY created_at",
    [tenantId],
  );
 
  const normalized = transformRows(rows, normalizeUser);
  const serialized = serializeToNDJSON(normalized);
 
  try {
    for await (const chunk of serialized) {
      const ok = res.write(chunk);
      if (!ok) {
        // Await drain before pulling more from the generator
        await new Promise<void>((resolve) => res.once("drain", resolve));
      }
    }
  } finally {
    res.end();
  }
}

import { Transform, pipeline } from "node:stream";
import { promisify } from "node:util";
 
const pipelineAsync = promisify(pipeline);
 
class CsvTransform extends Transform {
  private headerWritten = false;
 
  constructor(private readonly headers: string[]) {
    super({ objectMode: true });
  }
 
  override _transform(
    row: Record<string, unknown>,
    _encoding: BufferEncoding,
    callback: (err?: Error | null) => void,
  ): void {
    try {
      if (!this.headerWritten) {
        this.push(this.headers.join(",") + "\r\n");
        this.headerWritten = true;
      }
 
      const values = this.headers.map((h) => {
        const raw = String(row[h] ?? "").replace(/"/g, '""');
        return raw.includes(",") || raw.includes('"') || raw.includes("\n")
          ? `"${raw}"`
          : raw;
      });
 
      this.push(values.join(",") + "\r\n");
      callback();
    } catch (err) {
      callback(err as Error);
    }
  }
}
 
export async function handleCsvExport(
  tenantId: string,
  res: NodeJS.WritableStream,
): Promise<void> {
  const queryStream = db.queryStream<Record<string, unknown>>(
    "SELECT id, email, name, created_at FROM users WHERE tenant_id = $1",
    [tenantId],
  );
 
  const csvTransform = new CsvTransform(["id", "email", "name", "created_at"]);
 
  // The database cursor, transform, and HTTP response are all destroyed together on error
  await pipelineAsync(queryStream, csvTransform, res);
}