import { Readable } from "node:stream";
import { S3Client } from "@aws-sdk/client-s3";
import { Upload } from "@aws-sdk/lib-storage";
const s3 = new S3Client({ region: process.env.AWS_REGION });
const BUCKET = process.env.UPLOAD_BUCKET!;
export interface UploadResult {
key: string;
location: string;
bytes: number;
}
export async function streamToS3(
key: string,
body: Readable,
contentType: string,
): Promise<UploadResult> {
let bytes = 0;
body.on("data", (chunk: Buffer) => {
bytes += chunk.length;
});
const upload = new Upload({
client: s3,
params: { Bucket: BUCKET, Key: key, Body: body, ContentType: contentType },
partSize: 8 * 1024 * 1024, // 8 MB parts
queueSize: 4, // upload up to 4 parts concurrently
leavePartsOnError: false,
});
const out = await upload.done();
return { key, location: out.Location ?? "", bytes };
}
import busboy from "busboy";
import { randomUUID } from "node:crypto";
import type { Request } from "express";
import { streamToS3, UploadResult } from "./s3-upload";
const MAX_FILE_SIZE = 512 * 1024 * 1024; // 512 MB
export function receiveUploads(req: Request): Promise<UploadResult[]> {
return new Promise((resolve, reject) => {
const bb = busboy({
headers: req.headers,
limits: { fileSize: MAX_FILE_SIZE, files: 5 },
});
const pending: Promise<UploadResult>[] = [];
bb.on("file", (_field, stream, info) => {
const key = `uploads/${randomUUID()}-${info.filename}`;
stream.on("limit", () => {
stream.destroy();
reject(new Error(`File ${info.filename} exceeds size limit`));
});
pending.push(streamToS3(key, stream, info.mimeType));
});
bb.on("filesLimit", () => reject(new Error("Too many files")));
bb.on("error", (err) => reject(err));
bb.on("finish", () => {
Promise.all(pending).then(resolve).catch(reject);
});
req.pipe(bb);
});
}
import { Router, Request, Response } from "express";
import { receiveUploads } from "./upload.middleware";
export const uploadRouter = Router();
uploadRouter.post("/uploads", async (req: Request, res: Response) => {
const contentType = req.headers["content-type"] ?? "";
if (!contentType.startsWith("multipart/form-data")) {
return res.status(415).json({ error: "Expected multipart/form-data" });
}
try {
const results = await receiveUploads(req);
return res.status(201).json({
uploaded: results.map((r) => ({ key: r.key, bytes: r.bytes })),
});
} catch (err) {
// Ensure the socket is drained so the connection is not left hanging.
req.unpipe();
req.resume();
const message = err instanceof Error ? err.message : "Upload failed";
return res.status(400).json({ error: message });
}
});
This snippet shows how to accept large file uploads in an Express service without buffering entire files into memory or writing them to temporary disk first. The core idea is that busboy parses a multipart/form-data request as a stream of events, handing back a readable stream per file the moment its headers arrive. That stream can be piped straight into the AWS SDK's Upload helper, so bytes flow from the socket into S3 in fixed-size parts while the client is still uploading.
In s3-upload.ts, the streamToS3 function wraps @aws-sdk/lib-storage's Upload, which internally performs a multipart upload: it reads from the source stream, splits it into partSize chunks, and uploads parts concurrently up to queueSize. Passing the file's own readable stream as Body means the SDK applies backpressure — if S3 is slow, the SDK stops pulling, the file stream pauses, and busboy in turn pauses the request socket. This is why memory stays flat regardless of file size, which is the whole point of streaming over multer-style disk staging.
In upload.middleware.ts, createUploadHandler builds the busboy instance from the request headers and applies limits for file size and count so a malicious client cannot exhaust resources. Each file event yields the field name, the readable stream, and metadata like filename and mimeType. The handler collects a promise per file and, critically, listens for the limits event on each file stream: when a file exceeds fileSize, busboy emits limit and the stream must be drained or destroyed, otherwise the request hangs. The filesLimit and error events are forwarded to reject the aggregate.
In routes.ts, the route awaits Promise.all of all file uploads only after busboy emits finish, ensuring every part has completed before responding. A pitfall worth noting: responding early or throwing without consuming the stream leaves the socket half-read, so errors destroy the busboy stream to unblock the connection. This pattern suits APIs handling video, backups, or datasets where files routinely exceed available RAM, trading the simplicity of buffered parsing for constant memory and true streaming throughput.
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
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 axios, { AxiosError } from 'axios'
import { v4 as uuidv4 } from 'uuid'
const api = axios.create({
baseURL: import.meta.env.VITE_API_URL || 'http://localhost:3000/api/v1',
timeout: 15000,Axios API client with interceptors
import { create } from 'zustand'
import { persist } from 'zustand/middleware'
interface FilterState {
search: string
category: string | nullZustand for lightweight state management
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)
Share this code
Here's the card — post it anywhere.
