Async Patterns

// Error-first callback convention (errback)
fs.readFile("file.txt", "utf8", (err, data) => {
  if (err) return callback(err);  // always propagate
  callback(null, data);           // null for "no error"
});

// Writing an errback function
function readJson(path, cb) {
  fs.readFile(path, "utf8", (err, text) => {
    if (err) return cb(err);
    try {
      cb(null, JSON.parse(text));
    } catch (parseErr) {
      cb(parseErr);
    }
  });
}

import { promisify } from "node:util";
import fs from "node:fs";

// Convert a single errback function to a promise-returning one
const readFile = promisify(fs.readFile);
const data = await readFile("file.txt", "utf8");

// Promisify manually
function delay(ms) {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

// util.promisify respects the [util.promisify.custom] symbol
myLib[util.promisify.custom] = (arg) => new Promise((res, rej) => { ... });

// Create
const p = new Promise((resolve, reject) => {
  setTimeout(() => resolve("done"), 1000);
});

// Consume
p.then((val) => console.log(val))
 .catch((err) => console.error(err))
 .finally(() => console.log("always"));

// Promise static methods
Promise.resolve(value);           // already-resolved
Promise.reject(new Error("x"));  // already-rejected

// Concurrency combinators
const [a, b] = await Promise.all([fetchA(), fetchB()]);  // all or nothing

const results = await Promise.allSettled([p1, p2, p3]);
// [{ status: "fulfilled", value }, { status: "rejected", reason }]

const settled   = await Promise.race([p1, p2]);      // first to settle
const fulfilled = await Promise.any([p1, p2, p3]);   // first to FULFILL
// throws AggregateError if all reject

async function fetchUser(id) {
  const res  = await fetch(`/users/${id}`);
  if (!res.ok) throw new Error(`HTTP ${res.status}`);
  return res.json();
}

// Error handling
async function main() {
  try {
    const user = await fetchUser(1);
    console.log(user);
  } catch (err) {
    console.error(err);
  }
}

// Parallel execution
const [users, posts] = await Promise.all([
  fetchUsers(),
  fetchPosts(),
]);

// Sequential when order matters
for (const id of ids) {
  await processOne(id);   // one at a time
}

// Parallel with concurrency limit
import { default as pLimit } from "p-limit";
const limit = pLimit(5);
await Promise.all(ids.map(id => limit(() => processOne(id))));

// Top-level await (ESM only)
const config = await fs.promises.readFile("config.json", "utf8");
export default JSON.parse(config);

// Async generator — source of async values
async function* paginate(url) {
  let cursor = null;
  do {
    const res  = await fetch(url + (cursor ? `?cursor=${cursor}` : ""));
    const page = await res.json();
    yield* page.items;
    cursor = page.nextCursor;
  } while (cursor);
}

// Consume with for-await-of
for await (const item of paginate("https://api.example.com/items")) {
  console.log(item);
}

// Async iterable from array
async function* fromArray(arr) {
  for (const item of arr) yield item;
}

// Collect async iterable
async function collect(iter) {
  const items = [];
  for await (const item of iter) items.push(item);
  return items;
}

// Pipe two async generators
async function* transform(source) {
  for await (const item of source) {
    yield item.toUpperCase();
  }
}

const ac = new AbortController();
const { signal } = ac;

// Cancel after timeout
const tid = setTimeout(() => ac.abort(new Error("timed out")), 5000);

try {
  const res = await fetch(url, { signal });
  clearTimeout(tid);
} catch (err) {
  if (err.name === "AbortError") console.log("cancelled");
  else throw err;
}

// AbortSignal.timeout (Node 17.3+) — no manual cleanup needed
const res = await fetch(url, { signal: AbortSignal.timeout(5000) });

// AbortSignal.any (Node 20.3+) — abort when ANY signal fires
const signal = AbortSignal.any([signal1, signal2]);

// Listen for abort in your own async operation
signal.addEventListener("abort", () => cleanup());
if (signal.aborted) throw signal.reason;   // early check

import { setTimeout as sleep, setInterval as tick, setImmediate as immediate }
  from "node:timers/promises";

await sleep(1000);                      // delay 1 s
await sleep(500, "value");             // resolves with "value"
await sleep(500, undefined, { signal }); // cancellable

for await (const _ of tick(200)) {
  doWork();
}

await immediate();                      // next event loop tick

// Result type pattern (no-throw)
async function safeRun(fn) {
  try {
    return { ok: true, value: await fn() };
  } catch (err) {
    return { ok: false, error: err };
  }
}
const { ok, value, error } = await safeRun(() => fetchUser(1));

// Retry with exponential backoff
async function retry(fn, { attempts = 3, base = 100 } = {}) {
  for (let i = 0; i < attempts; i++) {
    try {
      return await fn();
    } catch (err) {
      if (i === attempts - 1) throw err;
      await sleep(base * 2 ** i);
    }
  }
}

// Timeout any promise
function withTimeout(promise, ms) {
  return Promise.race([
    promise,
    new Promise((_, reject) =>
      setTimeout(() => reject(new Error("Timeout")), ms)
    ),
  ]);
}

// Phase order (simplified):
// 1. nextTick queue
// 2. microtask queue (Promises, queueMicrotask)
// 3. Timers  (setTimeout, setInterval callbacks whose time has passed)
// 4. Pending callbacks (I/O errors deferred to next loop)
// 5. Idle / Prepare
// 6. Poll  (retrieve I/O events; block here if nothing pending)
// 7. Check (setImmediate)
// 8. Close callbacks (socket close, etc.)

setTimeout(() => console.log("1 timer"),    0);
setImmediate(() => console.log("2 immediate"));
process.nextTick(() => console.log("3 nextTick"));
Promise.resolve().then(() => console.log("4 promise"));
queueMicrotask(() => console.log("5 microtask"));

// Output: 3 nextTick → 4 promise → 5 microtask → 1 timer → 2 immediate
// (timer vs immediate order can vary when NOT inside an I/O callback)

// Fan-out: run N tasks at once
const results = await Promise.all(items.map(processItem));

// Fan-out with max concurrency (manual semaphore)
async function mapWithConcurrency(items, fn, limit) {
  const results = [];
  const executing = new Set();
  for (const item of items) {
    const p = Promise.resolve().then(() => fn(item));
    results.push(p);
    executing.add(p);
    p.finally(() => executing.delete(p));
    if (executing.size >= limit) await Promise.race(executing);
  }
  return Promise.all(results);
}

// Producer-consumer with async queue
async function* producer() {
  for (let i = 0; i < 100; i++) {
    await sleep(10);
    yield i;
  }
}

async function consumer(iter, workerId) {
  for await (const item of iter) {
    await processItem(item, workerId);
  }
}

// Barrier: wait for N things to happen
function createBarrier(n) {
  let resolve;
  let count = 0;
  const promise = new Promise(r => resolve = r);
  return {
    signal() { if (++count >= n) resolve(); },
    wait() { return promise; },
  };
}

import { AsyncLocalStorage } from "node:async_hooks";

const requestContext = new AsyncLocalStorage();

// Set context at request entry
app.use((req, res, next) => {
  requestContext.run({ requestId: uuid() }, next);
});

// Read anywhere in the async call tree
function logSomething(msg) {
  const ctx = requestContext.getStore();
  console.log(`[${ctx?.requestId}] ${msg}`);
}