Node/More/Examples/

Web Scraper Pipeline

A polite, rate-limited scraper: retries, named queues, middleware logging, and a fan-in workflow that aggregates results.

Scraping is bursty, failure-prone, and must respect the target's limits. This example fetches a set of pages under a strict rate limit, retries transient failures with backoff, isolates network work on its own queue, and joins the results with a workflow.

Project structure

code

scraper/
  tasks.ts     # fetch / extract / store / aggregate, plus logging middleware
  run.ts       # submit a chord: fetch-all → aggregate
  worker.ts    # worker + periodic cleanup

tasks.ts

fetchPage is rate-limited to 30 requests/minute and retries with exponential backoff. Network tasks run on a dedicated network queue so a backlog there never starves CPU-bound extraction.

import { Queue } from "@byteveda/flexiq";

export const queue = new Queue({ dbPath: "scraper.db" });

queue.task("fetchPage", async (url: string) => {
  const res = await fetch(url);
  if (!res.ok) {
    throw new Error(`HTTP ${res.status} for ${url}`);
  }
  return res.text();
}, {
  rateLimit: "30/m",
  maxRetries: 4,
  retryBackoff: { baseMs: 1_000, maxMs: 60_000 },
});

queue.task("extractLinks", (html: string) => {
  return [...html.matchAll(/href="(https?:\/\/[^"]+)"/g)].map((m) => m[1]);
});

// Source step: supplies the URL list the fan-out expands over.
queue.task("listUrls", (urls: string[]) => urls);

// Fan-in combiner: receives one entry per fetched page (each child's result).
queue.task("aggregate", (pages: string[]) => {
  const links = pages.flatMap((html) =>
    [...html.matchAll(/href="(https?:\/\/[^"]+)"/g)].map((m) => m[1]),
  );
  return { pages: pages.length, links: links.length, unique: new Set(links).size };
});

queue.task("cleanupCache", async (olderThanMs: number) => {
  await queue.purgeCompleted(olderThanMs);
});

// Middleware: one log line per execution and per failure.
queue.use({
  before(ctx) {
    console.log(`→ ${ctx.taskName} ${ctx.jobId}`);
  },
  after(ctx) {
    console.log(`✓ ${ctx.taskName} ${ctx.jobId}`);
  },
  onError(ctx, err) {
    console.warn(`✗ ${ctx.taskName} ${ctx.jobId}: ${String(err)}`);
  },
});

run.ts

A fan-out / fan-in expands the URL list into one fetchPage child per URL, then aggregate receives every fetched page as [html, …]. The fetches land on the network queue; the join runs on the default.

import { queue } from "./tasks";

const urls = ["https://a.example", "https://b.example", "https://c.example"];

const run = queue.workflows
  .define("scrape", 1)
  .step("list", "listUrls", { args: [urls] })
  .fanOut("fetch", { after: "list", task: "fetchPage", itemsFrom: "list", queue: "network" })
  .fanIn("aggregate", { after: "fetch", task: "aggregate" })
  .submit();

const final = await run.wait({ timeoutMs: 120_000 });
console.log(run.runId, final.state);

Fan-in passes the children's results to its task as [childResult, …], so aggregate sees every fetched page. A chord would instead join on completion and call the callback with its own args — use it when the join doesn't need the children's results.

worker.ts

The worker consumes both queues and sweeps completed jobs hourly.

import { queue } from "./tasks";

queue.registerPeriodic("cache-sweep", "cleanupCache", "0 * * * *", {
  args: [24 * 60 * 60 * 1000],
});

const worker = queue.runWorker({ queues: ["default", "network"] });
process.on("SIGINT", () => {
  worker.stop();
  process.exit(0);
});

Running it

bash

node worker.ts

bash

node run.ts

Key patterns

Pattern	Where
Politeness via rate limit	`task(..., { rateLimit: "30/m" })`
Transient-failure retries	`retryBackoff`
Workload isolation	`network` queue + `runWorker({ queues })`
Cross-cutting logging	`queue.use({ before, after, onError })`
Parallel-then-join over results	workflow `fanOut` / `fanIn`
Scheduled maintenance	`registerPeriodic` + `purgeCompleted`

tasks.ts

fetchPage is rate-limited to 30 requests/minute and retries with exponential backoff. Network tasks run on a dedicated network queue so a backlog there never starves CPU-bound extraction.

import { Queue } from "@byteveda/flexiq";

export const queue = new Queue({ dbPath: "scraper.db" });

queue.task("fetchPage", async (url: string) => {
  const res = await fetch(url);
  if (!res.ok) {
    throw new Error(`HTTP ${res.status} for ${url}`);
  }
  return res.text();
}, {
  rateLimit: "30/m",
  maxRetries: 4,
  retryBackoff: { baseMs: 1_000, maxMs: 60_000 },
});

queue.task("extractLinks", (html: string) => {
  return [...html.matchAll(/href="(https?:\/\/[^"]+)"/g)].map((m) => m[1]);
});

// Source step: supplies the URL list the fan-out expands over.
queue.task("listUrls", (urls: string[]) => urls);

// Fan-in combiner: receives one entry per fetched page (each child's result).
queue.task("aggregate", (pages: string[]) => {
  const links = pages.flatMap((html) =>
    [...html.matchAll(/href="(https?:\/\/[^"]+)"/g)].map((m) => m[1]),
  );
  return { pages: pages.length, links: links.length, unique: new Set(links).size };
});

queue.task("cleanupCache", async (olderThanMs: number) => {
  await queue.purgeCompleted(olderThanMs);
});

// Middleware: one log line per execution and per failure.
queue.use({
  before(ctx) {
    console.log(`→ ${ctx.taskName} ${ctx.jobId}`);
  },
  after(ctx) {
    console.log(`✓ ${ctx.taskName} ${ctx.jobId}`);
  },
  onError(ctx, err) {
    console.warn(`✗ ${ctx.taskName} ${ctx.jobId}: ${String(err)}`);
  },
});

run.ts

import { queue } from "./tasks";

const urls = ["https://a.example", "https://b.example", "https://c.example"];

const run = queue.workflows
  .define("scrape", 1)
  .step("list", "listUrls", { args: [urls] })
  .fanOut("fetch", { after: "list", task: "fetchPage", itemsFrom: "list", queue: "network" })
  .fanIn("aggregate", { after: "fetch", task: "aggregate" })
  .submit();

const final = await run.wait({ timeoutMs: 120_000 });
console.log(run.runId, final.state);

worker.ts

The worker consumes both queues and sweeps completed jobs hourly.

import { queue } from "./tasks";

queue.registerPeriodic("cache-sweep", "cleanupCache", "0 * * * *", {
  args: [24 * 60 * 60 * 1000],
});

const worker = queue.runWorker({ queues: ["default", "network"] });
process.on("SIGINT", () => {
  worker.stop();
  process.exit(0);
});

Key patterns

Pattern	Where
Politeness via rate limit	`task(..., { rateLimit: "30/m" })`
Transient-failure retries	`retryBackoff`
Workload isolation	`network` queue + `runWorker({ queues })`
Cross-cutting logging	`queue.use({ before, after, onError })`
Parallel-then-join over results	workflow `fanOut` / `fanIn`
Scheduled maintenance	`registerPeriodic` + `purgeCompleted`