const { Pool } = require('pg');
const Cursor = require('pg-cursor');
const pool = new Pool();
async function* streamUsers({ since, batchSize = 500 }) {
const client = await pool.connect();
const cursor = client.query(
new Cursor(
'SELECT id, email, created_at FROM users WHERE created_at >= $1 ORDER BY id',
[since]
)
);
try {
for (;;) {
const rows = await cursor.read(batchSize);
if (rows.length === 0) break;
for (const row of rows) yield row;
}
} finally {
await cursor.close();
client.release();
}
}
module.exports = { streamUsers };
const { Transform } = require('stream');
class JsonArrayStream extends Transform {
constructor(options = {}) {
super({ ...options, writableObjectMode: true });
this.first = true;
}
_transform(record, _encoding, callback) {
try {
const prefix = this.first ? '[' : ',';
this.first = false;
this.push(prefix + JSON.stringify(record));
callback();
} catch (err) {
callback(err);
}
}
_flush(callback) {
// Emit a valid empty array if no records ever arrived.
this.push(this.first ? '[]' : ']');
callback();
}
}
module.exports = { JsonArrayStream };
const { Readable, pipeline } = require('stream');
const { JsonArrayStream } = require('./jsonArrayStream');
const { streamUsers } = require('./usersCursor');
function exportUsers(req, res) {
const since = req.query.since || '1970-01-01';
res.setHeader('Content-Type', 'application/json; charset=utf-8');
res.setHeader('Cache-Control', 'no-store');
const source = Readable.from(streamUsers({ since }), { objectMode: true });
pipeline(source, new JsonArrayStream(), res, (err) => {
if (!err) return;
// Headers are already flushed, so a status code can't be changed here.
req.log.error({ err }, 'user export stream failed');
res.destroy(err);
});
}
module.exports = { exportUsers };
Serializing a huge database result set into a JSON array with JSON.stringify forces the entire dataset into memory before a single byte reaches the client. For thousands or millions of rows that means unbounded heap growth and a slow time-to-first-byte. The pattern shown here streams rows straight from a database cursor through a Transform stream, emitting valid JSON incrementally while honoring Node's backpressure so memory stays flat regardless of result size.
The jsonArrayStream tab implements the core Transform in object mode on the readable side. It buffers nothing beyond the current chunk: on the first _transform call it prefixes [, and every subsequent object is prefixed with a comma. The trick is the first flag — it lets the stream emit valid JSON separators without knowing how many records will arrive. JSON.stringify runs per row, so only one object is serialized at a time. _flush closes the array with ], and crucially still writes [] when the source was empty, so the output is always parseable. Because it extends Transform, push respecting the return value gives cooperative backpressure for free.
The usersCursor tab exposes an async generator backed by a pg cursor via pg-query-stream. Fetching in batches (batchSize) keeps the round-trips reasonable while never materializing the full table. Using a generator means the consumer pulls rows on demand, which naturally pairs with a stream that slows down when the socket is congested.
The exportController tab wires the two together with stream.pipeline, which is the safe way to connect stages: it propagates errors, destroys every stream on failure, and prevents socket leaks. Setting Content-Type: application/json up front commits to a streamed body, and Readable.from adapts the async generator into a proper readable source.
A key edge case is error handling after headers are sent — once bytes flow, an HTTP error status can no longer be set, so the handler logs and destroys the response instead. This approach trades the simplicity of a single res.json for constant memory and early first-byte, which is exactly the right call for large exports, reporting endpoints, and any response whose size scales with data volume.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
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)
class AddSettingsToAccounts < ActiveRecord::Migration[7.1]
disable_ddl_transaction!
def change
add_column :accounts, :settings, :jsonb, null: false, default: {}Postgres JSONB Partial Index for Feature Flags
Share this code
Here's the card — post it anywhere.
