const express = require('express');
const multer = require('multer');
const os = require('os');
const fs = require('fs/promises');
const { parseCsvStream } = require('./csvStreamParser');
const upload = multer({
storage: multer.diskStorage({ destination: os.tmpdir() }),
limits: { fileSize: 2 * 1024 * 1024 * 1024 } // 2GB
});
const router = express.Router();
router.post('/imports/contacts', upload.single('file'), async (req, res) => {
if (!req.file) {
return res.status(400).json({ error: 'file field is required' });
}
try {
const result = await parseCsvStream(req.file.path, req.app.locals.db);
res.status(result.errors.length ? 207 : 200).json(result);
} catch (err) {
res.status(422).json({ error: err.message });
} finally {
await fs.unlink(req.file.path).catch(() => {});
}
});
module.exports = router;
const fs = require('fs');
const { pipeline } = require('stream/promises');
const { parse } = require('csv-parse');
const { BatchInserter } = require('./BatchInserter');
async function parseCsvStream(filePath, db) {
const source = fs.createReadStream(filePath, { encoding: 'utf8' });
const parser = parse({
columns: true,
trim: true,
skip_empty_lines: true,
relax_column_count: true
});
const inserter = new BatchInserter(db, { batchSize: 500 });
await pipeline(source, parser, inserter);
return {
inserted: inserter.insertedCount,
skipped: inserter.errors.length,
errors: inserter.errors.slice(0, 100)
};
}
module.exports = { parseCsvStream };
const { Writable } = require('stream');
class BatchInserter extends Writable {
constructor(db, { batchSize = 500 } = {}) {
super({ objectMode: true });
this.db = db;
this.batchSize = batchSize;
this.buffer = [];
this.errors = [];
this.insertedCount = 0;
this.rowNum = 0;
}
_validate(row) {
if (!row.email || !/.+@.+\..+/.test(row.email)) {
return 'invalid or missing email';
}
if (!row.name) return 'missing name';
return null;
}
async _flush() {
if (this.buffer.length === 0) return;
const batch = this.buffer;
this.buffer = [];
const values = batch.map((r) => [r.email.toLowerCase(), r.name]);
await this.db.query(
'INSERT INTO contacts (email, name) VALUES ' +
values.map((_, i) => `($${i * 2 + 1}, $${i * 2 + 2})`).join(',') +
' ON CONFLICT (email) DO NOTHING',
values.flat()
);
this.insertedCount += batch.length;
}
_write(row, _enc, callback) {
this.rowNum += 1;
const problem = this._validate(row);
if (problem) {
this.errors.push({ row: this.rowNum, reason: problem });
return callback();
}
this.buffer.push(row);
if (this.buffer.length < this.batchSize) return callback();
this._flush().then(() => callback()).catch(callback);
}
_final(callback) {
this._flush().then(() => callback()).catch(callback);
}
}
module.exports = { BatchInserter };
Parsing a multi-gigabyte CSV upload by buffering it into memory is a classic way to crash a Node process. The approach shown here treats the upload as a pipeline of streams so that at any moment only a few rows and one insert batch are held in memory, regardless of file size. The uploadRoutes tab wires an Express endpoint with multer configured for disk storage rather than memory storage, which is the key first decision: multer.diskStorage writes the incoming request body to a temp file, so the HTTP body never has to fit in RAM.
Once the file is on disk, parseCsvStream in the csvStreamParser tab builds the real work. It opens a fs.createReadStream, pipes it through csv-parse configured with columns: true so each record arrives as an object keyed by header, and wraps the whole thing in pipeline from stream/promises. Using pipeline matters because it propagates errors and destroys every stream in the chain if any stage fails, avoiding leaked file descriptors on a malformed file.
The heart of the pattern is the Writable stream BatchInserter in the BatchInserter stream tab. Each parsed row flows into its _write method, which validates the row, pushes valid ones into an internal this.buffer, and only touches the database once the buffer reaches batchSize. Crucially, _write's callback is not invoked until the batch flush resolves. This is how backpressure works: the parser cannot push the next record until the writable signals readiness, so a slow database naturally throttles the fast file reader instead of letting rows pile up in memory.
Invalid rows are collected in this.errors rather than aborting the whole import, which suits bulk ingestion where partial success is acceptable. _final flushes the last partial batch so no rows are lost at end-of-file. The controller reports counts and always removes the temp file in a finally block. A subtle trade-off: this design commits batches incrementally, so a mid-stream failure leaves earlier batches persisted; when full atomicity is required, wrapping the pipeline in a single transaction or staging table is the alternative. This structure scales to arbitrarily large files with flat, predictable memory use.
Related snips
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
class SignupForm
include ActiveModel::Model
include ActiveModel::Attributes
attribute :account_name, :string
attribute :email, :stringShallow Controller, Deep Params: Form Object Pattern
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
import { randomBytes, createHash } from "crypto";
import jwt from "jsonwebtoken";
import { RefreshTokenStore } from "./store";
const ACCESS_SECRET = process.env.ACCESS_SECRET!;
const ACCESS_TTL = "15m";JWT access + refresh token rotation (conceptual)
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Form Validation Example</title>
<style>HTML forms with validation and accessibility
package pools
import (
"bytes"
"sync"
)sync.Pool for bytes.Buffer to reduce allocations in hot paths
Share this code
Here's the card — post it anywhere.
