Streaming NDJSON Validated Parse

import { z } from 'zod';
import { Transform } from 'node:stream';
import { createReadStream } from 'node:fs';
import { pipeline } from 'node:stream/promises';

const RecordSchema = z.object({
  userId: z.number().int().positive(),
  event: z.string().min(1),
  at: z.string().datetime(),
});

class ValidatingLines extends Transform {
  private buffer = '';
  public valid = 0;
  public quarantined = 0;

  constructor() {
    super({ objectMode: true, readableObjectMode: false });
  }

  _transform(chunk: Buffer, _enc: string, callback: (err?: Error | null, data?: never) => void) {
    this.buffer += chunk.toString('utf8');
    let newline: number;

    while ((newline = this.buffer.indexOf('\n')) !== -1) {
      const line = this.buffer.slice(0, newline).trim();
      this.buffer = this.buffer.slice(newline + 1);
      if (!line) continue;

      try {
        const parsed = RecordSchema.parse(JSON.parse(line));
        this.valid += 1;
        this.push(JSON.stringify(parsed) + '\n');
      } catch {
        this.quarantined += 1;
      }
    }
    callback();
  }
}

const validator = new ValidatingLines();

await pipeline(
  createReadStream('events.ndjson'),
  validator,
  async function* (records) {
    for await (const line of records) {
      if (validator.valid % 100 === 0) console.log('streamed', validator.valid);
    }
  },
);

console.log('valid:', validator.valid, 'quarantined:', validator.quarantined);