Build a Web Content Pipeline: RSS to Enriched Data in Node.js

Every piece of this pipeline has appeared in an earlier lesson. The robots check and the per-host pacing came from lesson 1, the language gate and enrichment from lesson 2, the fingerprinting from lesson 3. What is new is the part that only shows up when a pipeline runs unattended, every hour, for months: it gets killed halfway through, a call fails at 3 a.m., a feed republishes an old article with a new tracking parameter, and someone asks what last month's credits were spent on. A pipeline that handles those is a product. One that doesn't is a script you babysit.

Here is the shape:

1
2

RSS feed / sitemap ──► discover ──► dedupe ──► extract ──► detect language ──► enrich ──► store
                         (10)        (free)      (25)            (2)            (20 + 10)

The numbers are credits per item, from each API's catalog page. They will matter in the last section.

Discover: feeds first, sitemaps second

The cheapest way to find new pages is to let the site tell you. Most publishers and many documentation sites expose an RSS or Atom feed listing their recent items. RSS to JSON turns one into structured JSON for 10 credits, however many items it holds.

Try it liveRSS to JSON10 credits

POSThttps://api.apiverve.com/v1/rsstojson

curl -X POST "https://api.apiverve.com/v1/rsstojson" \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"url":"https://blog.cloudflare.com/rss/"}'

url

feed is an array of items with title, link, pubDate and description. articles is the count; maxReached says whether the list was cut.

Each item in feed carries title, link, pubDate, description and website. When we called that feed it held 20 items and maxReached was false. If maxReached is ever true, the feed had more items than were returned, so poll more often or you may miss items between runs.

When a site has no feed, its sitemap is the next best source. A sitemap is a plain XML list of URLs, usually at /sitemap.xml or named in robots.txt, often with a <lastmod> date per URL. Fetch it yourself: it is a single file meant for crawlers, and the lastmod filter is what makes it usable. A large site's sitemap can list a hundred thousand URLs. You want the dozen that changed since your last run.

sources.js

import { verve } from './verve.js';

export async function fromFeed(feedUrl) {
  const { data } = await verve('rsstojson', { url: feedUrl });
  return (data.feed ?? []).map((item) => ({
    url: item.link,
    title: item.title,
    published: item.pubDate ? new Date(item.pubDate).toISOString() : null,
  }));
}

export async function fromSitemap(sitemapUrl, since) {
  const res = await fetch(sitemapUrl, { signal: AbortSignal.timeout(15000) });
  if (!res.ok) throw new Error(`sitemap HTTP ${res.status}`);
  const xml = await res.text();
  const items = [];
  for (const [, block] of xml.matchAll(/<url>([\s\S]*?)<\/url>/g)) {
    const loc = block.match(/<loc>\s*([^<\s]+)\s*<\/loc>/)?.[1];
    const lastmod = block.match(/<lastmod>\s*([^<\s]+)\s*<\/lastmod>/)?.[1];
    if (!loc) continue;
    if (since && lastmod && new Date(lastmod) <= since) continue;
    items.push({ url: loc.replace(/&amp;/g, '&'), title: null, published: lastmod ?? null });
  }
  return items;
}

A sitemap whose root is <sitemapindex> lists other sitemaps rather than pages; follow its <loc> entries the same way. And lastmod is the publisher's claim, not a guarantee. Some sites stamp every URL with today's date on each deploy, which makes the filter useless. If the first run with a since date still returns everything, that is what happened, and the content fingerprint in the next section is your backstop.

Idempotency: the same item, processed once

The single most important property of a scheduled pipeline is that running it twice does the same as running it once. Without it, every retry, overlapping cron run or manual re-run pays for the same extraction again and writes duplicate rows.

Idempotency starts with a stable identity for each item, and URLs are worse at that than they look. The same article can arrive as https://Example.com/post?utm_source=rss, https://example.com/post and https://example.com/post#comments. Canonicalise before you hash: lowercase the host, drop the fragment, strip tracking parameters, sort what remains.

identity.js

import { createHash } from 'node:crypto';

const TRACKING = /^(utm_\w+|fbclid|gclid|mc_cid|mc_eid|ref|ref_src)$/i;

export function canonical(raw) {
  const u = new URL(raw);
  u.hash = '';
  u.hostname = u.hostname.toLowerCase();
  for (const key of [...u.searchParams.keys()]) if (TRACKING.test(key)) u.searchParams.delete(key);
  u.searchParams.sort();
  if (u.pathname.length > 1 && u.pathname.endsWith('/')) u.pathname = u.pathname.slice(0, -1);
  return u.toString();
}

export const itemId = (url) => createHash('sha256').update(canonical(url)).digest('hex').slice(0, 20);
export const contentHash = (text) => createHash('sha256').update(text, 'utf8').digest('hex');

The second half of idempotency is checkpointing by stage. Each item gets its own record, and each stage writes its result into that record before the next stage starts. If the process dies after extraction but before enrichment, the next run sees the extracted text on disk and resumes at enrichment. It never pays 25 credits to extract the same page twice.

The content hash closes the last gap. When the same URL comes back with a newer lastmod, re-extract it, but compare the hash of the new text with the stored one. If they match, the page did not really change, and you skip the 30 credits of enrichment.

Check yourself

Your pipeline crashed after extracting 40 of 100 articles. On the re-run, what should happen to those 40?

Retries that know when to stop

Retries belong at the level of a single call. A timeout, a 5xx or a rate-limit 429 may well succeed on a second attempt; back off and try again, honouring retry-after when the server sends it. Everything else fails the same way every time. A 400 means your input is wrong. A 401 means your key is missing or unrecognised. A 429 with x-api-remaining-credits at 0 means the month's credits are spent, and that one should stop the whole run, not just the item, because every following call will fail too.

This client replaces the minimal verve.js from lesson 1: it returns { data, remaining } instead of bare data, carrying the remaining-credit header along with the answer. The pipeline uses it to reconcile its own cost accounting at the end of the run.

verve.js

const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

export class VerveError extends Error {
  constructor(message, { status = 0, retryable = false, fatal = false } = {}) {
    super(message);
    Object.assign(this, { status, retryable, fatal });
  }
}

async function once(api, body) {
  let res;
  try {
    res = await fetch(`https://api.apiverve.com/v1/${api}`, {
      method: 'POST',
      headers: { 'x-api-key': process.env.APIVERVE_API_KEY, 'Content-Type': 'application/json' },
      body: JSON.stringify(body),
      signal: AbortSignal.timeout(30000),
    });
  } catch (err) {
    throw new VerveError(`${api}: ${err.message}`, { retryable: true });
  }
  const remainingHeader = res.headers.get('x-api-remaining-credits');
  const remaining = remainingHeader === null ? null : Number(remainingHeader);
  const json = await res.json().catch(() => null);
  if (res.ok && json?.status === 'ok') return { data: json.data, remaining };

  const message = `${api}: ${json?.error || `HTTP ${res.status}`}`;
  if (res.status === 429 && remaining === 0) throw new VerveError(message, { status: 429, fatal: true });
  if (res.status === 401) throw new VerveError(message, { status: 401, fatal: true });
  const retryable = res.status === 429 || res.status >= 500;
  const err = new VerveError(message, { status: res.status, retryable });
  err.retryAfter = Number(res.headers.get('retry-after')) || 0;
  throw err;
}

export async function verve(api, body, { attempts = 3 } = {}) {
  for (let i = 1; ; i++) {
    try {
      return await once(api, body);
    } catch (err) {
      if (!err.retryable || i >= attempts) throw err;
      const backoff = err.retryAfter ? err.retryAfter * 1000 : 500 * 2 ** i + Math.random() * 250;
      await sleep(backoff);
    }
  }
}

The fatal flag is what lets the pipeline stop early. Without it, a run that exhausts credits on item 3 would grind through 97 more items, each failing after three attempts, and log 291 identical errors.

Check yourself

Halfway through a run, a call returns 429 and the x-api-remaining-credits header is 0. What should the pipeline do?

Answer the check above to continue.

The pipeline, with cost accounting

Each item lives in store/<id>.json, written atomically. Every call goes through step(), which records the API, its credit cost (only on success, since failed calls are not billed) and how long it took in a run-wide ledger. A budget cap stops the run from starting new items once spending reaches it. And at the end, the ledger's own total is compared with the difference in x-api-remaining-credits between the first and last call.

Node 18+, ESM. It reuses robots.js and queue.js from lesson 1 and cutAtSentence logic from lesson 2, inlined here so the file stands alone.

pipeline.js

import { mkdir, readFile, writeFile, rename } from 'node:fs/promises';
import { verve } from './verve.js';
import { fromFeed } from './sources.js';
import { canonical, itemId, contentHash } from './identity.js';
import { robotsFor, isAllowed } from './robots.js';
import { createQueue } from './queue.js';

// Credits per successful call, from each API's catalog page (the API definitions' tokenUse).
const CREDITS = { rsstojson: 10, websitetotext: 25, languagedetector: 2, textsummarizer: 20, keywordextractor: 10 };
const BUDGET = Number(process.env.RUN_BUDGET ?? 2000);
const STORE = './store';

const ledger = [];
let firstRemaining = null;
let lastRemaining = null;
const spent = () => ledger.reduce((sum, e) => sum + e.credits, 0);

async function step(api, body) {
  const started = Date.now();
  try {
    const { data, remaining } = await verve(api, body);
    if (remaining !== null) {
      // parallel calls can settle out of order, so keep the extremes rather than the latest
      firstRemaining = Math.max(firstRemaining ?? 0, remaining + CREDITS[api]);
      lastRemaining = Math.min(lastRemaining ?? Infinity, remaining);
    }
    ledger.push({ api, ok: true, credits: CREDITS[api], ms: Date.now() - started });
    return data;
  } catch (err) {
    ledger.push({ api, ok: false, credits: 0, ms: Date.now() - started, error: err.message });
    throw err;
  }
}

async function load(id) {
  try {
    return JSON.parse(await readFile(`${STORE}/${id}.json`, 'utf8'));
  } catch {
    return null;
  }
}

async function save(record) {
  const file = `${STORE}/${record.id}.json`;
  await writeFile(file + '.tmp', JSON.stringify(record, null, 2));
  await rename(file + '.tmp', file);
}

function cutAtSentence(text, max = 5000) {
  if (text.length <= max) return text;
  const cut = text.slice(0, max);
  const end = cut.lastIndexOf('. ');
  return end > max * 0.5 ? cut.slice(0, end + 1) : cut;
}

async function processItem(item, queue) {
  const url = canonical(item.url);
  const id = itemId(url);
  const record = (await load(id)) ?? { id, url, title: item.title, published: item.published, stages: {} };
  if (record.stages.enrich || record.stages.skipped) return 'done';

  if (!record.stages.extract) {
    const robots = await robotsFor(url);
    if (!isAllowed(robots, url)) {
      record.stages.skipped = 'robots.txt';
      await save(record);
      return 'skipped';
    }
    const gap = Math.max(2000, robots.delay * 1000);
    const page = await queue.run(new URL(url).host, gap, () => step('websitetotext', { url }));
    const text = (page.text ?? '').replace(/\s+/g, ' ').trim();
    record.stages.extract = { text, hash: contentHash(text), declaredLanguage: page.language ?? null };
    await save(record);
  }

  if (!record.stages.language) {
    const { text, declaredLanguage } = record.stages.extract;
    const sample = text.slice(0, 1000);
    let language = declaredLanguage ? declaredLanguage.slice(0, 2).toLowerCase() : null; // 'en-US' -> 'en'
    if (sample.length >= 20) {
      const d = await step('languagedetector', { text: sample });
      const [a, b] = d.detectedLanguages ?? [];
      if (d.primaryCode && (a?.confidence ?? 0) - (b?.confidence ?? 0) >= 0.03) language = d.primaryCode;
    }
    record.stages.language = { language };
    await save(record);
  }

  if (record.stages.language.language !== 'en') {
    record.stages.skipped = `language ${record.stages.language.language ?? 'unknown'}`;
    await save(record);
    return 'skipped';
  }

  const { text } = record.stages.extract;
  const [sum, kw] = await Promise.all([
    step('textsummarizer', { text: cutAtSentence(text), sentences: 3 }),
    step('keywordextractor', { text: text.slice(0, 50000) }),
  ]);
  record.stages.enrich = {
    summary: sum.summary,
    keywords: (kw.topKeywords ?? []).map((k) => k.keyword),
    enrichedAt: new Date().toISOString(),
  };
  await save(record);
  return 'enriched';
}

async function main() {
  await mkdir(STORE, { recursive: true });
  const feedUrl = process.argv[2];
  if (!feedUrl) throw new Error('usage: node pipeline.js <feed-url>');

  const items = await fromFeed(feedUrl);
  ledger.push({ api: 'rsstojson', ok: true, credits: CREDITS.rsstojson, ms: 0 });
  const queue = createQueue({ concurrency: 3 });
  const perItem = CREDITS.websitetotext + CREDITS.languagedetector + CREDITS.textsummarizer + CREDITS.keywordextractor;
  const outcomes = {};
  let stopped = null;

  for (const item of items) {
    if (stopped) break;
    if (spent() + perItem > BUDGET) {
      stopped = `budget of ${BUDGET} credits reached`;
      break;
    }
    try {
      const result = await processItem(item, queue);
      outcomes[result] = (outcomes[result] ?? 0) + 1;
    } catch (err) {
      outcomes.failed = (outcomes.failed ?? 0) + 1;
      console.error(`${item.url}: ${err.message}`);
      if (err.fatal) stopped = err.message;
    }
  }

  const byApi = {};
  for (const e of ledger) byApi[e.api] = (byApi[e.api] ?? 0) + e.credits;
  console.log({ outcomes, creditsByApi: byApi, creditsSpent: spent(), stopped });
  if (firstRemaining !== null && lastRemaining !== null) {
    console.log(`Balance moved by ${firstRemaining - lastRemaining} credits during the run (ledger: ${spent() - CREDITS.rsstojson} after the feed call)`);
  }
}

main().catch((err) => {
  console.error(err.message);
  process.exit(1);
});

export APIVERVE_API_KEY=YOUR_API_KEY
export RUN_BUDGET=500
node pipeline.js https://blog.cloudflare.com/rss/
node pipeline.js https://blog.cloudflare.com/rss/   # second run: only the 10-credit feed call
ls store

Items run one at a time here, which keeps the budget check exact and the output readable. The queue still enforces the per-host gap, which matters because every item in a single feed is usually on the same host. If you parallelise across items later, reserve each item's projected cost before starting it, or several items can pass the budget check at once and overshoot it together.

fromFeed calls verve directly rather than step, so the feed call is added to the ledger by hand; that is also why it is left out of the balance reconciliation, which starts from the first step call.

Reading the ledger

The ledger answers the questions people actually ask about a pipeline. A typical first run over a 20-item feed of English articles costs 10 for the feed plus 57 per item: 25 to extract, 2 to detect language, 20 to summarise and 10 for keywords, so about 1,150 credits. A second run an hour later costs 10, because every item is already done. A non-English item costs 27 and stops at the language gate. A failed call costs nothing.

The reconciliation line is the check that keeps the ledger honest. If the balance moved by more than the ledger recorded, something else is spending on the same key, or a credit figure in CREDITS is out of date. API prices can change; the catalog page is the source of truth. Note that the header shows the account's balance, so another process using the same key during the run will show up as a mismatch. That is a useful signal in itself: one key per pipeline makes cost attribution trivial.

Where to take it next

What you have is a complete, restartable pipeline with an audit trail. Natural extensions, each a few lines on top of this structure:

Store in a database instead of JSON files: one row per item, a column per stage, the item id as the primary key. An upsert on that key is your idempotency.
Re-check changed pages: on a later lastmod, clear the extract stage, re-extract, and compare the content hash before clearing enrich.
Watch the feed itself with the monitor from lesson 3, so a feed that silently stops publishing raises an alert instead of a quiet week.
Drop personal data at the extract stage, as lesson 1 described, before it is ever written to store/.

That completes the course, and with it the three courses of the Web Data Developer track. When you are ready, take the exam on the APIVerve Certified Developer: Web Data certification page. It covers extraction, site analysis and everything in this course, including hands-on questions you answer by calling the APIs yourself.

Finished the lesson?Mark it complete to track your progress toward the certification.

Next: Exam APV-210 →

Build a content pipeline