Node.js readable streams for large file processing

import { createReadStream } from 'fs';
import { createInterface } from 'readline';
import { pipeline } from 'stream/promises';
import { Transform } from 'stream';

// Process CSV line by line (memory-efficient)
async function processCsvFile(filePath: string): Promise<void> {
  const fileStream = createReadStream(filePath, { encoding: 'utf8' });
  const rl = createInterface({ input: fileStream, crlfDelay: Infinity });

  let lineNumber = 0;
  const errors: string[] = [];

  for await (const line of rl) {
    lineNumber++;
    if (lineNumber === 1) continue; // skip header

    try {
      const fields = line.split(',');
      await processRecord({ id: fields[0], name: fields[1], value: Number(fields[2]) });
    } catch (err) {
      errors.push(`Line ${lineNumber}: ${err instanceof Error ? err.message : 'unknown error'}`);
    }
  }

  console.log(`Processed ${lineNumber - 1} records, ${errors.length} errors`);
}

// Transform stream for JSON Lines format (one JSON object per line)
class JsonLineParser extends Transform {
  constructor() { super({ objectMode: true }); }

  _transform(chunk: Buffer, _encoding: string, callback: () => void) {
    const lines = chunk.toString().split('\n').filter(l => l.trim());
    for (const line of lines) {
      try { this.push(JSON.parse(line)); } catch { /* skip invalid */ }
    }
    callback();
  }
}

// Batch records before inserting into DB
class BatchProcessor extends Transform {
  private batch: object[] = [];
  private readonly batchSize = 100;

  constructor() { super({ objectMode: true }); }

  async _transform(record: object, _encoding: string, callback: () => void) {
    this.batch.push(record);
    if (this.batch.length >= this.batchSize) {
      await this.flushBatch();
    }
    callback();
  }

  async _flush(callback: () => void) {
    await this.flushBatch();
    callback();
  }

  private async flushBatch() {
    if (this.batch.length > 0) {
      await db.bulkInsert(this.batch);
      this.batch = [];
    }
  }
}