Node.js Multi-Source Web Scraping Pipeline: Normalize Data, Track Runs, Detect Broken Sources

A scraper that works for one website is straightforward.

The architecture gets more interesting when the product depends on ten, twenty, or fifty sources that all behave differently.

One publisher exposes JSON.
Another renders everything into HTML.
Another changes its markup.
One source starts timing out.
Another keeps returning 200 OK while publishing no new records.

The pipeline can still be running while one part of the dataset quietly becomes stale.

This guide builds a small Node.js pattern for that problem.

We will create:

• source-specific collectors
• one normalized article schema
• retry handling
• limited concurrency
• per-source run records
• partial-failure handling
• duplicate protection
• source-health checks
• tests for normalization and isolated failures

The goal is simple:

different sources
      ↓
source-specific collection
      ↓
one shared data shape
      ↓
run + health tracking
      ↓
reusable data

The collectors are allowed to be different.

The rest of the product should not need to be.

The architecture

We will use two example source types:

JSON API source
HTML source

Both will produce the same internal article shape.

Our pipeline will look like this:

Source A ── JSON collector ──┐
                            │
Source B ── HTML collector ──┼── normalize ── store
                            │                │
Source C ── another adapter ┘                ├── run audit
                                             └── source health

If Source B fails, Source A should still complete.

That is an important property.

A multi-source pipeline should be able to say:

12 sources succeeded
1 source failed

instead of turning one broken publisher into a total pipeline outage.

1. Create the project

This example uses Node.js with ES modules.

mkdir multi-source-pipeline
cd multi-source-pipeline

npm init -y

npm install \
  cheerio \
  p-limit \
  p-retry \
  zod

Update package.json:

{
  "name": "multi-source-pipeline",
  "type": "module",
  "scripts": {
    "start": "node src/index.js",
    "test": "node --test"
  }
}

Create the project structure:

multi-source-pipeline/
├── data/
├── src/
│   ├── sources.js
│   ├── schema.js
│   ├── collectors.js
│   ├── normalize.js
│   ├── store.js
│   ├── health.js
│   ├── pipeline.js
│   └── index.js
└── test/
    └── pipeline.test.js

2. Keep source configuration separate from pipeline logic

Create src/sources.js:

export const sources = [
  {
    id: "publisher-api",
    name: "Publisher API",
    kind: "json-api",
    url: "https://example.com/api/articles",
    freshnessMinutes: 60,
  },

  {
    id: "publisher-html",
    name: "Publisher HTML",
    kind: "html",
    url: "https://example.org/news",
    freshnessMinutes: 120,
  },
];

Each source declares:

who it is
how it should be collected
where it lives
how fresh we expect it to be

Do not scatter these values across parser files and scheduler code.

A registry gives the rest of the system one place to understand which sources exist.

In a larger platform, this configuration could also include:

enabled
schedule
timeout
category mapping
request headers
parser version
owner
last known schema change

3. Define one normalized article shape

Create src/schema.js:

import { z } from "zod";

export const ArticleSchema = z.object({
  id: z.string(),

  sourceId: z.string(),

  sourceUrl: z.string().url(),

  title: z.string().min(1),

  summary: z.string(),

  canonicalUrl: z.string().url(),

  publishedAt: z.string(),

  author: z.string().nullable(),

  categories: z.array(z.string()),

  tags: z.array(z.string()),

  imageUrl: z.string().url().nullable(),

  bodyText: z.string(),

  collectedAt: z.string(),
});

Every collector must eventually produce this shape.

The source can call a field:

headline
title
name
post_title

The rest of the platform should still receive:

title

That separation is what lets downstream consumers stay stable while individual sources change.

4. Build a fetch helper with timeout handling

Create src/collectors.js:

import * as cheerio from "cheerio";

async function fetchWithTimeout(
  url,
  options = {},
  timeoutMs = 10000
) {
  const controller =
    new AbortController();

  const timer = setTimeout(
    () => controller.abort(),
    timeoutMs
  );

  try {
    const response = await fetch(url, {
      ...options,
      signal: controller.signal,

      headers: {
        "user-agent":
          "multi-source-pipeline/1.0",

        ...options.headers,
      },
    });

    if (!response.ok) {
      throw new Error(
        `HTTP ${response.status}`
      );
    }

    return response;
  } finally {
    clearTimeout(timer);
  }
}

This helper gives every collector the same timeout and HTTP-failure behavior.

If one source needs different headers or authentication, its adapter can still supply them.

Respect the source's permitted access method, terms, rate limits, and published APIs where available.

5. Create a JSON API collector

Continue src/collectors.js:

async function collectJsonApi(source) {
  const response =
    await fetchWithTimeout(
      source.url
    );

  const payload =
    await response.json();

  if (!Array.isArray(payload)) {
    throw new Error(
      "Expected an array from JSON source"
    );
  }

  return payload.map((item) => ({
    externalId:
      String(item.id),

    title:
      item.title ?? "",

    summary:
      item.summary ?? "",

    url:
      item.url,

    publishedAt:
      item.published_at,

    author:
      item.author ?? null,

    categories:
      item.categories ?? [],

    tags:
      item.tags ?? [],

    imageUrl:
      item.image_url ?? null,

    bodyText:
      item.body ?? "",
  }));
}

Notice that this is still a source-specific shape.

We have not normalized it yet.

The collector's job is:

talk to this source
extract this source's fields
return a predictable collector result

6. Create an HTML collector

Now add a generic HTML example:

async function collectHtml(source) {
  const response =
    await fetchWithTimeout(
      source.url
    );

  const html =
    await response.text();

  const $ = cheerio.load(html);

  const items = [];

  $("article").each(
    (_, element) => {
      const article =
        $(element);

      const link =
        article
          .find("a")
          .first()
          .attr("href");

      const title =
        article
          .find("h2, h3")
          .first()
          .text()
          .trim();

      if (!link || !title) {
        return;
      }

      const url =
        new URL(
          link,
          source.url
        ).toString();

      items.push({
        externalId: url,

        title,

        summary:
          article
            .find("p")
            .first()
            .text()
            .trim(),

        url,

        publishedAt:
          article
            .find("time")
            .attr("datetime") ??
          new Date().toISOString(),

        author: null,

        categories: [],

        tags: [],

        imageUrl: null,

        bodyText: "",
      });
    }
  );

  return items;
}

This parser is intentionally generic.

A production source adapter should use selectors and extraction rules that match the source you are permitted to collect.

The useful architecture is the boundary around the parser, not the selectors in this example.

7. Route each source to its adapter

Add this to src/collectors.js:

export async function collectSource(
  source
) {
  switch (source.kind) {
    case "json-api":
      return collectJsonApi(source);

    case "html":
      return collectHtml(source);

    default:
      throw new Error(
        `Unsupported source kind: ${source.kind}`
      );
  }
}

Later, the registry can grow:

wordpress-api
json-api
html
rss
schema-org
next-state
custom-adapter

The pipeline does not need a rewrite every time another collection method appears.

You add another adapter.

8. Normalize every source into one article contract

Create src/normalize.js:

import {
  createHash,
} from "node:crypto";

import {
  ArticleSchema,
} from "./schema.js";

function stableId(
  sourceId,
  canonicalUrl
) {
  return createHash("sha256")
    .update(
      `${sourceId}:${canonicalUrl}`
    )
    .digest("hex")
    .slice(0, 24);
}

export function normalizeArticle(
  source,
  raw
) {
  const canonicalUrl =
    new URL(
      raw.url,
      source.url
    ).toString();

  const article = {
    id:
      stableId(
        source.id,
        canonicalUrl
      ),

    sourceId:
      source.id,

    sourceUrl:
      source.url,

    title:
      raw.title
        .replace(/\s+/g, " ")
        .trim(),

    summary:
      String(
        raw.summary ?? ""
      ).trim(),

    canonicalUrl,

    publishedAt:
      new Date(
        raw.publishedAt
      ).toISOString(),

    author:
      raw.author
        ? String(raw.author)
        : null,

    categories:
      Array.isArray(
        raw.categories
      )
        ? raw.categories
        : [],

    tags:
      Array.isArray(raw.tags)
        ? raw.tags
        : [],

    imageUrl:
      raw.imageUrl ?? null,

    bodyText:
      String(
        raw.bodyText ?? ""
      ).trim(),

    collectedAt:
      new Date().toISOString(),
  };

  return ArticleSchema.parse(
    article
  );
}

Now the rest of the product does not care whether the source was JSON, HTML, RSS, or another permitted format.

It receives:

Article

That is the contract.

9. Store normalized data and run history separately

Create src/store.js:

import {
  mkdir,
  readFile,
  writeFile,
  appendFile,
} from "node:fs/promises";

const DATA_DIR = "./data";

const ARTICLE_FILE =
  `${DATA_DIR}/articles.json`;

const RUN_FILE =
  `${DATA_DIR}/runs.jsonl`;

async function ensureDataDir() {
  await mkdir(
    DATA_DIR,
    {
      recursive: true,
    }
  );
}

async function readArticles() {
  await ensureDataDir();

  try {
    const text =
      await readFile(
        ARTICLE_FILE,
        "utf8"
      );

    return JSON.parse(text);
  } catch (error) {
    if (
      error.code === "ENOENT"
    ) {
      return [];
    }

    throw error;
  }
}

export async function saveArticles(
  incoming
) {
  const existing =
    await readArticles();

  const byId =
    new Map(
      existing.map(
        (article) => [
          article.id,
          article,
        ]
      )
    );

  for (
    const article
    of incoming
  ) {
    byId.set(
      article.id,
      article
    );
  }

  const merged =
    [...byId.values()];

  await writeFile(
    ARTICLE_FILE,
    JSON.stringify(
      merged,
      null,
      2
    )
  );

  return merged.length;
}

export async function appendRun(
  run
) {
  await ensureDataDir();

  await appendFile(
    RUN_FILE,
    `${JSON.stringify(run)}\n`
  );
}

export async function readRuns() {
  await ensureDataDir();

  try {
    const text =
      await readFile(
        RUN_FILE,
        "utf8"
      );

    return text
      .trim()
      .split("\n")
      .filter(Boolean)
      .map(JSON.parse);
  } catch (error) {
    if (
      error.code === "ENOENT"
    ) {
      return [];
    }

    throw error;
  }
}

Articles and collection runs solve different jobs.

Article storage answers:

What data do we have?

Run history answers:

What happened when we tried to collect it?

Do not collapse those questions into one table or one log line.

10. Retry one source without retrying the whole pipeline

Create src/pipeline.js:

import pLimit from "p-limit";
import pRetry from "p-retry";

import {
  collectSource,
} from "./collectors.js";

import {
  normalizeArticle,
} from "./normalize.js";

import {
  appendRun,
  saveArticles,
} from "./store.js";

const limit =
  pLimit(3);

async function runSource(
  source
) {
  const startedAt =
    new Date().toISOString();

  try {
    const raw =
      await pRetry(
        () =>
          collectSource(
            source
          ),
        {
          retries: 2,

          onFailedAttempt:
            ({ error, attemptNumber }) => {
              console.warn(
                `[${source.id}] attempt ${attemptNumber} failed:`,
                error.message
              );
            },
        }
      );

    const normalized =
      raw.map(
        (item) =>
          normalizeArticle(
            source,
            item
          )
      );

    await saveArticles(
      normalized
    );

    const run = {
      sourceId:
        source.id,

      status:
        "success",

      startedAt,

      finishedAt:
        new Date()
          .toISOString(),

      discovered:
        raw.length,

      normalized:
        normalized.length,

      error:
        null,
    };

    await appendRun(run);

    return run;
  } catch (error) {
    const run = {
      sourceId:
        source.id,

      status:
        "failed",

      startedAt,

      finishedAt:
        new Date()
          .toISOString(),

      discovered:
        0,

      normalized:
        0,

      error:
        error.message,
    };

    await appendRun(run);

    return run;
  }
}

export async function runPipeline(
  sources
) {
  return Promise.all(
    sources.map(
      (source) =>
        limit(
          () =>
            runSource(
              source
            )
        )
    )
  );
}

There are two useful details here.

Retries belong to the source

If one publisher times out, retry that source.

Do not rerun eleven healthy publishers because source twelve failed.

A failed source returns a run result

runSource() does not throw the entire batch away.

It records:

status = failed

and lets the other sources finish.

That gives the operator a partial result instead of an all-or-nothing job.

11. Add source-level health

A scheduler saying:

running

does not tell you whether every source is healthy.

Create src/health.js:

export function getSourceHealth(
  source,
  runs,
  now = Date.now()
) {
  const sourceRuns =
    runs
      .filter(
        (run) =>
          run.sourceId ===
          source.id
      )
      .sort(
        (a, b) =>
          new Date(
            b.finishedAt
          ) -
          new Date(
            a.finishedAt
          )
      );

  if (
    sourceRuns.length === 0
  ) {
    return {
      sourceId:
        source.id,

      status:
        "unknown",

      lastSuccessAt:
        null,
    };
  }

  const lastSuccess =
    sourceRuns.find(
      (run) =>
        run.status ===
        "success"
    );

  if (!lastSuccess) {
    return {
      sourceId:
        source.id,

      status:
        "failing",

      lastSuccessAt:
        null,
    };
  }

  const ageMinutes =
    (
      now -
      new Date(
        lastSuccess.finishedAt
      ).getTime()
    ) /
    60000;

  const recentRun =
    sourceRuns[0];

  if (
    recentRun.status ===
    "failed"
  ) {
    return {
      sourceId:
        source.id,

      status:
        "degraded",

      lastSuccessAt:
        lastSuccess.finishedAt,
    };
  }

  if (
    ageMinutes >
    source.freshnessMinutes
  ) {
    return {
      sourceId:
        source.id,

      status:
        "stale",

      lastSuccessAt:
        lastSuccess.finishedAt,
    };
  }

  return {
    sourceId:
      source.id,

    status:
      "healthy",

    lastSuccessAt:
      lastSuccess.finishedAt,
  };
}

Now the system can distinguish:

healthy
degraded
stale
failing
unknown

That is much more useful than a single green pipeline badge.

12. Run the pipeline and print a useful summary

Create src/index.js:

import {
  sources,
} from "./sources.js";

import {
  runPipeline,
} from "./pipeline.js";

import {
  readRuns,
} from "./store.js";

import {
  getSourceHealth,
} from "./health.js";

async function main() {
  const results =
    await runPipeline(
      sources
    );

  console.log(
    "\nCollection results"
  );

  for (
    const result
    of results
  ) {
    console.log(
      `${result.sourceId}: ${result.status} (${result.normalized} articles)`
    );
  }

  const runs =
    await readRuns();

  console.log(
    "\nSource health"
  );

  for (
    const source
    of sources
  ) {
    const health =
      getSourceHealth(
        source,
        runs
      );

    console.log(
      `${source.id}: ${health.status}`
    );
  }
}

main().catch(
  (error) => {
    console.error(error);
    process.exitCode = 1;
  }
);

Run:

npm start

A mixed result can look like:

Collection results

publisher-api: success (24 articles)
publisher-html: failed (0 articles)

Source health

publisher-api: healthy
publisher-html: degraded

That is a useful operational result.

One source failed.

The pipeline still knows what succeeded.

13. Track freshness, not only HTTP errors

Some source failures are loud:

500
timeout
DNS error
parser exception

Others are quiet.

A source can return:

200 OK

while producing:

0 new articles

for hours.

That can be a source change, a parser break, or simply a quiet publishing period.

The pipeline should have enough context to tell those cases apart.

Useful per-source signals include:

last successful run
last article timestamp
records discovered
records normalized
consecutive failures
freshness threshold
parser version

For example:

Source: publisher-07

Last successful run:
14 hours ago

Expected freshness:
2 hours

HTTP status:
200

New normalized records:
0

Health:
STALE

The server is technically up.

The data product still has a problem worth looking at.

14. Test that one source cannot take down the batch

Create test/pipeline.test.js.

For pipeline tests, I would normally inject mock collectors rather than make external requests.

Here is the behavior we want to protect:

import test from "node:test";
import assert from "node:assert/strict";

test(
  "one failed source does not erase successful source results",
  async () => {
    const results = [
      {
        sourceId:
          "source-a",

        status:
          "success",

        normalized:
          12,
      },

      {
        sourceId:
          "source-b",

        status:
          "failed",

        normalized:
          0,
      },
    ];

    const successful =
      results.filter(
        (result) =>
          result.status ===
          "success"
      );

    const failed =
      results.filter(
        (result) =>
          result.status ===
          "failed"
      );

    assert.equal(
      successful.length,
      1
    );

    assert.equal(
      failed.length,
      1
    );

    assert.equal(
      successful[0]
        .normalized,
      12
    );
  }
);

In a complete repository, replace the static result with dependency-injected collectors so the actual orchestration path is tested without calling publisher websites.

The invariant is the useful part:

one broken source
does not erase
healthy-source results

15. A few production checks worth adding

The example above gives the core structure.

A long-running system usually needs more.

Idempotent storage

Running the same source twice should update or ignore an existing article rather than create duplicates.

A deterministic article ID or database upsert helps.

Per-source rate control

Different publishers may have different permitted access limits.

Do not assume one concurrency setting fits every source.

Parser versioning

When source extraction rules change, record which parser version produced the data.

That makes debugging older runs much easier.

Alert thresholds

Do not send an alert for every temporary network error.

A better alert might be:

3 consecutive failures

or:

last successful collection
older than expected freshness window

Run-level audit

Keep:

source
start time
finish time
status
records found
records accepted
records rejected
error reason

That turns debugging into inspection instead of guesswork.

Controlled API access

If other products consume the normalized data, expose it through an authenticated API instead of handing every downstream service database credentials.

API keys, quotas, usage logs, and revocation give the data layer a much cleaner boundary.

Where this pattern came from

We used this operating model in an Ascent Innovate Software project that collected news from 13 publisher sources for a private data client.

The public project includes different source-access methods, source-specific collectors, one shared article structure, run tracking, scheduling, source health, operator controls, and authenticated API access.

Every publisher could keep the collection logic it needed.

The data layer after collection stayed consistent.

That is the pattern this article generalizes.

The code here is an educational implementation and does not reproduce the private client's collectors, credentials, source rules, or internal architecture.

Related work

Real Estate News Data Pipeline & API Dashboard | Ascent Innovate Software

Editorial note

The public project details were checked against the current Ascent Innovate Software work page, and the implementation examples were reviewed for consistency before publication.