Node.js Internals

setTimeout(() => console.log("timeout"));
setImmediate(() => console.log("immediate"));
Promise.resolve().then(() => console.log("promise"));
process.nextTick(() => console.log("nextTick"));
// nextTick, promise, then timeout/immediate in non-deterministic order (main module)

const Env = z.object({
  PORT: z.coerce.number().default(3000),
  DATABASE_URL: z.string().url(),
  FEATURE_X: z.enum(["true", "false"]).transform((v) => v === "true"),
});
export const env = Env.parse(process.env); // crash at boot, not at 3am

// A: main module
setTimeout(() => console.log("timeout"), 0);
setImmediate(() => console.log("immediate"));

// B: inside an I/O callback
import { readFile } from "node:fs";
readFile(import.meta.filename, () => {
  setTimeout(() => console.log("timeout"), 0);
  setImmediate(() => console.log("immediate"));
});

setTimeout(() => {
  console.log("t1");
  Promise.resolve().then(() => console.log("p1"));
  process.nextTick(() => console.log("n1"));
}, 0);
setTimeout(() => {
  console.log("t2");
}, 0);

import { once } from "node:events";
for (const row of rows) {
  if (!out.write(format(row))) await once(out, "drain");
}
out.end();

import { pipeline } from "node:stream/promises";
import { createGzip } from "node:zlib";
await pipeline(
  fs.createReadStream("report.csv"),
  async function* (src) { for await (const chunk of src) yield redact(chunk); },
  createGzip(),
  res,
);

import { AsyncLocalStorage } from "node:async_hooks";
export const ctx = new AsyncLocalStorage<{ requestId: string; tenantId?: string }>();
app.use((req, _res, next) => {
  ctx.run({ requestId: req.get("x-request-id") ?? crypto.randomUUID() }, next);
});
logger.info = (msg) => base.info({ ...ctx.getStore(), msg });

app.get("/search", async (req, res) => {
  const ac = new AbortController();
  res.on("close", () => { if (!res.writableFinished) ac.abort(); });
  const signal = AbortSignal.any([ac.signal, AbortSignal.timeout(2000)]);
  const r = await fetch(searchUrl(req.query), { signal });
  res.json(await r.json());
});

const server = app.listen(3000);
server.keepAliveTimeout = 65_000; // > ALB idle timeout (60s)
server.headersTimeout = 66_000;   // must exceed keepAliveTimeout

import { monitorEventLoopDelay, performance } from "node:perf_hooks";
const h = monitorEventLoopDelay({ resolution: 20 }); h.enable();
let last = performance.eventLoopUtilization();
setInterval(() => {
  const now = performance.eventLoopUtilization(last); last = performance.eventLoopUtilization();
  metrics.gauge("elu", now.utilization);
  metrics.gauge("loop_delay_p99_ms", h.percentile(99) / 1e6); h.reset();
}, 10_000).unref();

import { setImmediate as yieldToLoop } from "node:timers/promises";
for (let i = 0; i < rows.length; i += 1000) {
  processBatch(rows.slice(i, i + 1000));
  await yieldToLoop(); // let I/O and timers run
}

await pipeline(
  fs.createReadStream(path),
  parse({ columns: true, bom: true }),       // csv-parse: Transform in objectMode
  async function* (rows) {
    let batch: Row[] = [];
    for await (const r of rows) {
      batch.push(validate(r));
      if (batch.length === 1000) { yield batch; batch = []; }
    }
    if (batch.length) yield batch;
  },
  new Writable({ objectMode: true, async write(batch, _e, cb) {
    try { await repo.bulkInsert(batch); cb(); } catch (e) { cb(e as Error); }
  } }),
);

function merge(t: any, s: any) {
  for (const k in s) {
    if (typeof s[k] === "object") merge((t[k] ??= {}), s[k]);
    else t[k] = s[k];
  }
}
merge({}, JSON.parse('{"__proto__": {"isAdmin": true}}'));
({} as any).isAdmin; // true — every object is now "admin"

const EMAIL = /^([a-zA-Z0-9]+)*@example\.com$/;
EMAIL.test("a".repeat(35) + "!"); // exponential backtracking, blocks the loop

import { exec } from "node:child_process";
app.post("/convert", (req, res) => {
  const name = req.body.filename; // e.g. "report.docx"
  exec(`libreoffice --headless --convert-to pdf /uploads/${name}`, (err, out) => {
    if (err) return res.status(500).end();
    res.json({ ok: true });
  });
});

import { EventEmitter } from "node:events";
export const priceBus = new EventEmitter(); // module-level singleton

app.get("/prices/stream", (req, res) => {
  res.setHeader("Content-Type", "text/event-stream");
  priceBus.on("price", (p) => res.write(`data: ${JSON.stringify(p)}\n\n`));
});

app.post("/import", upload.single("file"), async (req, res) => {
  const text = await fs.promises.readFile(req.file!.path, "utf8");
  const rows = text.split("\n").map((l) => l.split(","));
  await Promise.all(rows.map((r) => db.insert("products", toProduct(r))));
  res.json({ imported: rows.length });
});

const out = fs.createWriteStream("/tmp/export.csv");
const cursor = db.query("SELECT * FROM orders").stream();
cursor.on("data", (row) => {
  out.write(toCsvLine(row)); // return value ignored
});
cursor.on("end", () => out.end());

// login
const hash = await new Promise<Buffer>((ok, ko) =>
  crypto.pbkdf2(password, salt, 600_000, 64, "sha512", (e, k) => (e ? ko(e) : ok(k))));
// other endpoints
const tpl = await fs.promises.readFile("templates/invoice.html", "utf8");
const r = await fetch("https://inventory.internal/stock"); // dns.lookup under the hood

let shuttingDown = false;
async function shutdown(signal: string) {
  if (shuttingDown) return; shuttingDown = true;
  logger.info({ signal }, "shutting down");
  health.ready = false;
  await sleep(5_000);                        // let the LB stop routing
  const force = setTimeout(() => { server.closeAllConnections(); process.exit(1); }, 20_000).unref();
  const closed = new Promise<void>((r) => server.close(() => r()));
  const sweep = setInterval(() => server.closeIdleConnections(), 500); // sockets that go idle after close()
  await closed; clearInterval(sweep);
  await Promise.allSettled([consumer.disconnect(), db.end(), redis.quit()]);
  clearTimeout(force);
  process.exit(0);
}
process.on("SIGTERM", shutdown).on("SIGINT", shutdown);

process.on("SIGTERM", () => {
  server.close();
  db.end();
  process.exit(0);
});

import { EventEmitter } from "node:events";
const orders = new EventEmitter();
orders.on("created", async (order) => {
  await inventory.reserve(order); // rejects when out of stock
});
try {
  orders.emit("created", order);
} catch (e) {
  logger.error(e); // never reached
}

app.get("/files/:id", (req, res) => {
  const src = fs.createReadStream(pathFor(req.params.id));
  src.pipe(createGzip()).pipe(res);
});