Scenarios: File Upload, Download & Streaming

import { createPresignedPost } from "@aws-sdk/s3-presigned-post";

const { url, fields } = await createPresignedPost(s3, {
  Bucket: "uploads-quarantine",
  Key: `u/${userId}/${crypto.randomUUID()}`,
  Conditions: [
    ["content-length-range", 1, 100 * 1024 * 1024],
    ["starts-with", "$Content-Type", "video/"],
  ],
  Fields: { "Content-Type": declaredType },
  Expires: 300, // seconds
});
// browser: FormData(fields..., file LAST) → POST url

{
  "Rules": [
    { "ID": "abort-mpu", "Status": "Enabled", "Filter": {},
      "AbortIncompleteMultipartUpload": { "DaysAfterInitiation": 3 } },
    { "ID": "expire-pending", "Status": "Enabled",
      "Filter": { "Tag": { "Key": "status", "Value": "pending" } },
      "Expiration": { "Days": 1 } }
  ]
}

CREATE TABLE upload (
  id uuid PRIMARY KEY,
  tenant_id uuid NOT NULL,
  owner_id uuid NOT NULL,
  s3_key text NOT NULL UNIQUE,
  declared_size bigint NOT NULL,
  declared_type text NOT NULL,
  status text NOT NULL CHECK (status IN ('pending','uploaded','scanning','ready','rejected','expired')),
  sha256 text,
  created_at timestamptz NOT NULL DEFAULT now()
);
UPDATE upload SET status = 'uploaded' WHERE id = $1 AND owner_id = $2 AND status = 'pending';

import AWS from "aws-sdk";
const s3 = new AWS.S3();

export async function getUploadUrl(fileName: string) {
  return s3.getSignedUrlPromise("putObject", {
    Bucket: "uploads",
    Key: `videos/${fileName}`,
    ContentType: "video/mp4",
    Expires: 3600,
  });
}

[
  {
    "AllowedOrigins": ["https://app.example.com"],
    "AllowedMethods": ["PUT", "POST", "GET"],
    "AllowedHeaders": ["*"],
    "MaxAgeSeconds": 3000
  }
]

@Post("upload")
@UseInterceptors(FileInterceptor("file", { limits: { fileSize: 150 * 1024 * 1024 } }))
upload(@UploadedFile() file: Express.Multer.File) {
  return s3.upload({ Bucket, Key: `v/${file.originalname}`, Body: file.stream }).promise();
}

import busboy from "busboy";
import { Upload } from "@aws-sdk/lib-storage";
import { PassThrough } from "node:stream";

app.post("/upload", (req, res) => {
  const bb = busboy({ headers: req.headers, limits: { files: 1, fileSize: 150 * 1024 * 1024 } });
  bb.on("file", (_name, file, info) => {
    const body = new PassThrough();
    const upload = new Upload({
      client: s3,
      params: { Bucket, Key: `q/${crypto.randomUUID()}`, Body: body, ContentType: "application/octet-stream" },
      partSize: 8 * 1024 * 1024,
      queueSize: 2,
    });
    file.on("limit", () => { upload.abort(); res.status(413).end(); });
    req.on("close", () => { if (!res.writableEnded) upload.abort(); });
    file.pipe(body);
    upload.done().then(() => res.status(201).json({ ok: true })).catch(() => { if (!res.headersSent) res.status(500).end(); });
  });
  req.pipe(bb);
});

const queue = new PQueue({ concurrency: 5 });
app.post("/upload", upload.single("file"), (req, res) => {
  queue.add(() => uploadToS3(req.file!.buffer)).then(() => res.sendStatus(201));
});

import { fileTypeFromBuffer } from "file-type"; // ESM-only

const head = await readFirstBytes(s3Stream, 4100);
const ft = await fileTypeFromBuffer(head);
if (!ft || !["image/png", "image/jpeg"].includes(ft.mime)) throw new RejectedUpload("type");

import path from "node:path";
import fs from "node:fs";

bb.on("file", (_field, file, info) => {
  const target = path.join("/data/uploads", req.user.tenantId, info.filename);
  file.pipe(fs.createWriteStream(target));
});

app.get("/exports/:id/download", async (req, res) => {
  const obj = await s3.send(new GetObjectCommand({ Bucket, Key: keyFor(req.params.id) }));
  const bytes = await obj.Body!.transformToByteArray();
  res.setHeader("Content-Type", "application/zip");
  res.send(Buffer.from(bytes));
});

app.get("/orders/export", (req, res) => {
  const stream = db.streamOrders();
  res.write("[");
  stream.on("data", (row) => {
    res.write(JSON.stringify(row));
  });
  res.end("]");
});

retry: 3000
id: 1842
event: order-updated
data: {"orderId":"o_1","status":"shipped"}

: ping

import { Transform } from "node:stream";
import { createHash } from "node:crypto";

export function inspect(maxBytes: number) {
  const hash = createHash("sha256");
  let seen = 0; let head = Buffer.alloc(0); let checked = false;
  const t = new Transform({
    transform(chunk: Buffer, _enc, cb) {
      seen += chunk.length;
      if (seen > maxBytes) return cb(new Error("too_large"));
      if (!checked) {
        head = Buffer.concat([head, chunk]).subarray(0, 5);
        if (head.length >= 5) {
          checked = true;
          if (head.toString("latin1") !== "%PDF-") return cb(new Error("not_pdf"));
        }
      }
      hash.update(chunk);
      cb(null, chunk);
    },
    flush(cb) { t.emit("digest", hash.digest("hex"), seen); cb(); },
  });
  return t;
}

app.get("/folders/:id/zip", async (req, res) => {
  const keys = await listKeys(req.params.id); // 8,000 keys
  const archive = archiver("zip");
  archive.pipe(res);
  for (const key of keys) {
    const obj = await s3.send(new GetObjectCommand({ Bucket, Key: key }));
    archive.append(obj.Body as Readable, { name: key });
  }
  archive.finalize();
});

const bb = busboy({ headers: req.headers, limits: { fileSize: 50 * 1024 * 1024 } });
bb.on("file", (_n, file) => {
  const upload = new Upload({ client: s3, params: { Bucket, Key: key, Body: file } });
  upload.done().then(() => markReady(key)).then(() => res.sendStatus(201));
});
req.pipe(bb);

const toDb = new Transform({
  objectMode: true,
  transform(row, _enc, cb) {
    db.insert("items", row); // returns a Promise
    cb();
  },
});
await pipeline(fs.createReadStream(file), csvParser(), toDb, new Writable({ objectMode: true, write(_c, _e, cb) { cb(); } }));

import { Readable } from "node:stream";
export async function GET(req: Request) {
  const obj = await s3.send(new GetObjectCommand({ Bucket, Key }), { abortSignal: req.signal });
  return new Response(Readable.toWeb(obj.Body as Readable) as ReadableStream, {
    headers: { "Content-Type": obj.ContentType ?? "application/octet-stream", "Content-Length": String(obj.ContentLength) },
  });
}

app.get("/media/:id", async (req, res) => {
  const file = pathFor(req.params.id);
  const { size } = await fs.promises.stat(file);
  const [s, e] = (req.headers.range ?? "bytes=0-").replace("bytes=", "").split("-");
  const start = Number(s);
  const end = e ? Number(e) : size;
  res.writeHead(206, {
    "Content-Range": `bytes ${start}-${end}/${size}`,
    "Content-Length": end - start,
    "Content-Type": "video/mp4",
  });
  fs.createReadStream(file, { start, end }).pipe(res);
});

import { to as copyTo } from "pg-copy-streams";
import { pipeline } from "node:stream/promises";
import { createGzip } from "node:zlib";

if (!/^[0-9a-f-]{36}$/i.test(tenantId)) throw new Error("bad tenant id"); // COPY has no bind params
const client = await replicaPool.connect();
try {
  await client.query("SET statement_timeout = '10min'");
  const src = client.query(copyTo(
    "COPY (SELECT id, booked_at, amount, currency FROM txn WHERE tenant_id = '" + tenantId + "') TO STDOUT WITH (FORMAT csv, HEADER)"
  ));
  res.setHeader("Content-Type", "text/csv");
  res.setHeader("Content-Encoding", "gzip");
  res.setHeader("Content-Disposition", 'attachment; filename="transactions.csv"');
  await pipeline(src, createGzip(), res);
} finally { client.release(); }
// NOTE: COPY does not accept bind parameters — tenantId must be a validated UUID (or use a view/temp function).