# How to Build an AI-Ready Web Data Pipeline Using Bright Data and Node.js

> Source: <https://dev.to/codewithshahan/how-to-build-an-ai-ready-web-data-pipeline-using-bright-data-and-nodejs-gc3>
> Published: 2026-09-21 11:19:22+00:00

A scraper can succeed and still give you bad data.

A website can change its HTML, a selector can stop matching, or a scraper can return a partial record without throwing an obvious error. Your scraper may still report success while your application quietly stores [incorrect data.](https://medium.com/how-to-bypass-anti-bot-walls-for-production-ready-apps-7de5bf0891b7)

That’s the problem this guide solves.

Concretely, we’ll build a Node.js pipeline that uses Bright Data Scraper Studio for web extraction while our application handles normalization, validation, deduplication, snapshots, and change detection.

*🔎 Build your first prompt-based scraper in minutes using* *Scraper Studio AI Agent for Free.*

Then we’ll point our pipeline at a real Texas real-estate domain, track listings across multiple runs, deliberately break the scraper, and repair it [without changing the pipeline itself](https://www.reddit.com/r/webdev/comments/1b8umsq/why_are_devs_obsessed_with_separation_of_concerns/).

🔗 The complete source code is available in my [GitHub repository](https://github.com/codewithshahan/self-healing-ai-data-pipeline). You can also download [**The Web Data Pipeline**](https://codewithshahan.gumroad.com/l/self-healing-web-data-pipeline) handbook for free:

Our data extraction pipeline divides responsibilities into two layers:

For this guide specifically, the final outline looks like this:

Therefore, by following this [*Separation of Concerns (SoC)*](https://en.wikipedia.org/wiki/Separation_of_concerns) principle, *the same*[***pipeline can be reused***](https://www.reddit.com/r/dataengineering/comments/1ko8u01/best_practices_for_reusing_data_pipelines_across/) for e-commerce, crypto rates, travel listings, job boards, and competitor/financial data monitoring; the list goes on.

In a scalable data pipeline, *neither* layer should depend on the other’s internal implementation details.

We achieve that with two rules.

Our Node.js application should NEVER depend on the target website's HTML or DOM structure.

Instead, the extraction layer produces data that conforms to a stable contract. The extraction adapter knows about the external source; our application knows about the contract.

For example, our application can work with:

```
{
  "address": "string",
  "price": 305000,
  "bedrooms": 4
}
```

It doesn't need to know whether those values came from a CSS selector, an AI-generated scraper, or another extraction system.

This is the practical idea behind the [**Dependency Inversion Principle (DIP)**](https://codewithshahan.gumroad.com/l/cleancode-zero-to-one) and the [** Ports and Adapters (Hexagonal)**](https://scotthannen.org/blog/2024/07/29/hexagonal-ports-and-adapters-architecture-i.html) approach: keep volatile external systems away from the core application logic.

*Grab my book* *Clean Code Zero to One**to master those skills.*

Later, we'll implement this boundary with an `adapter` that maps source-specific data into our stable contract.

A common mistake is coupling the processing pipeline to one specific dataset. For example, hardcoding the entire pipeline around:

```
{
  "address": "string",
  "price": "number"
}
```

works for real estate, but becomes a problem when the same system needs to process products, jobs, news, or financial data.

The pipeline should therefore remain independent of *specific* payload keys. Its job is to handle common operations such as:

```
ingest → normalize → validate → deduplicate → store → compare
```

Source-specific rules belong in the `adapter` and `validation` level. That way, changing the data source does not require cloning the entire pipeline.

When an AI industry or quantitative hedge fund needs to track market trends or competitor headcount shifts across platforms, they aren’t looking for *static web pages*; rather, they are hunting for massive, continuous datasets to feed predictive analytics systems, RAG engines, or LLM applications.

However, modern *Web Application Firewalls (WAFs)* like [Cloudflare or Akamai](https://www.cloudflare.com/cloudflare-vs-akamai/) frequently issue “soft bans", returning fake HTTP `200 OK` responses that contain CAPTCHAs instead of data:

Furthermore, the traditional scraper is a victim of three primary forces: ***fragile selectors, silent failures, and aggressive anti-bot systems*.** When a dev writes a Playwright or Puppeteer script, they often rely on specific DOM paths. If the target site updates its UI, that selector vanishes.

Worst of all are the silent failures. The scraper doesn’t crash; it simply returns `_null_` for the price while successfully extracting the address. Your database is now being poisoned with partial records, a far more dangerous outcome than a hard crash.

Anti-bot defenses can create another failure mode. Even when your headers look clean, making *too many requests* per second or scraping too aggressively from a single IP can trigger a site’s edge defenses. Instead of allowing the request to reach the site’s internal systems, the CDN can reject it with responses such as `503 Slow Down` or `503 Service Unavailable:`

A managed scraper API, such as the [Web Unlocker](https://get.brightdata.com/bypass-antibot-walls) or the [Scraping Browser API](https://get.brightdata.com/codewithshahan-scraping-browser), can auto-fix the `503` error. But getting past the security layer is only the ***first half of the problem***.

Even when a handcrafted scraper successfully retrieves the page, the output may still be a messy, unstructured collection of nested strings. That output is difficult to reuse reliably across applications, AI agents, vector databases, and analytics systems.

You do not rise to the level of your application goals; you fall to the level of your data collection infrastructure.

The goal is therefore changing:

We are moving away from treating a scraper as a ***“static script”*** that we write and toward treating it as a ***“managed infrastructure”*** that we describe.

This change allows us to stop worrying about how the data is pulled and start focusing on what the data represents to our application logic.

To see this transformation in practice, consider tracking the US real estate market using a [Zillow Seattle](https://www.zillow.com/seattle-wa/) search page:

This raw page contains property cards, filtering dropdowns, and pagination elements. Our backend application cannot consume this visual presentation directly. It requires a structured payload closer to this:

```
{
  "price": 1450000,
  "bathrooms": 2,
  "square_feet": 2230,
  "address": "2714 10th Avenue W",
  "city": "Seattle",
  "zip_code": "98119",
  "status": "Active"
}
```

That transformation is the first data engineering problem: turning a human-facing webpage into a reliable data payload.

To solve this, a traditional Playwright scraper might locate an element like this:

``` js
const price = await page
.locator('[data-test="property-card-price"]')
.textContent();
```

Or perhaps the developer finds a class in the current HTML:

``` js
const price = await page
.locator('.property-card-price')
.textContent();
```

Both approaches may work today. But they tightly couple the scraper to the site’s current HTML structure. When that structure changes, the selectors stop matching. Even worse, the scraper might not throw an obvious runtime error; it will simply pass silent `null` values down your pipeline:

```
{
  "price": null,
  "bedrooms": 4,
  "bathrooms": 2
}
```

Fortunately, **that is manageable for *one* page**. It becomes much harder when a traditional scraper needs to collect thousands of records.

One property page is not the real challenge. A real estate or e-commerce application needs thousands of listings across cities, ZIP codes, and neighborhoods. Now the scraper has to deal with pagination, filtering, changing page structures, and thousands of individual records.

At this point, the HTML document is no longer the product. **The data is the product.**

And that data still isn’t useful until we decide how the application will consume it. Instead of keeping raw HTML, we may want:

The important question becomes:

*How can I reliably turn changing web pages into structured data that our software can use?*

This is where the responsibilities of the scraper and the application need to separate:

Once we start collecting useful data, a *completely* different set of engineering questions appears. These questions determine whether our system is simply collecting data or actually producing useful application logic.

For instance:

Eventually, a user might ask:

*“Find 3 bedroom houses under my budget that recently dropped in price and explain which ones look like the best opportunities.”*

A scraper cannot answer that question. It only extracts today’s price and doesn’t know what that price was yesterday or whether the change is significant.

But our application does.

A scraper, even an advanced extraction system such as Scraper Studio, is very good at turning changing web pages into structured data, but they are *inherently* stateless.

This distinction becomes much clearer with this example. Our pipeline could determine:

`_“This product lost 31.27% of its value compared with yesterday and crossed a critical risk threshold.”_`

To achieve this, our application needs to remember yesterday’s price, compare the two values, calculate the change, and decide what that change means.

Ultimately, the real business value is *rarely* the raw extraction itself. It comes from the intelligence built around the data.

At this point, you might wonder why we don’t simply clone this existing repo and move on.

Because the goal isn’t just to make one scraper work. I want you to understand how the mechanism works under the hood and build each part yourself.

If you rely entirely on AI automation loops without understanding what happens underneath, you’ll eventually be lost in the ocean. You will stop working on the actual application and start debugging the extraction infrastructure again.

This creates a **maintenance loop**: an expensive cycle where developers spend hours debugging proxy routing, browser behavior, selectors, and DOM mutations instead of shipping application features.

To break the “maintenance loop", let’s compare the ***three most common approaches*** developers use today:

`npx skills add brightdata/skills`.
The important difference is *how much of the extraction infrastructure you have to build and maintain yourself.*

Bright data has a unique one-click Self‑Healing feature for updating an existing scraper when its extraction logic becomes outdated. Plus, it’s AI-assisted scraper generation takes few minutes to crawl.

*Try generating your first**prompt-based scraper using Scraper Studio here**.*

For example, without writing Playwright project and manually designing selectors, you can describe a task like

`“Extract the property address, price, bedrooms, bathrooms, square footage, city, ZIP code, and listing status from this real-estate website.”`

The AI agent will generate the scraper and [schema](https://docs.brightdata.com/products/scraper-studio/input-and-output-schema), which we can test and refine before running it. Scraper Studio can then be triggered through its interface, API, or scheduled runs.

If you don't like IDE, you can also run [Bright Data CLI](https://docs.brightdata.com/products/scraper-studio/build-with-the-cli) locally. We’ll use that later when we deliberately break the scraper and repair it.

When a target site is redesigned and a scraper begins returning `null` or missing fields, you fix it in place using `bdata scraper heal` or using the built-in IDE's “Self-Healing" feature like this:

```
bdata scraper heal <COLLECTOR_ID> \
  "The price field is returning null after the page redesign. \
   Extract the current price and currency from the new page structure."
```

Because this process preserves the stable Collector ID, every API trigger, schedule, and integration keeps working without a single line of code being changed in our application layer.

This converts the ***“Maintenance Loop”*** from a hours-long engineering hurdle into a five-minute terminal command.

At the beginning of this guide, we established two rules:

Scraper Studio fits underneath those rules.

**We are not giving up control of our application**. Our Node.js pipeline still owns the stable data contract. The extraction layer can change while the application layer remains stable.

That’s exactly the separation we designed in Section 1.

Cost also depends on what the scraper properly loads. Bright Data currently lists **5,000 page loads per month** in the [free Scraper Studio tier](https://get.brightdata.com/ai-web-scraper-studio) and ***$1.50 per 1,000 page loads on pay-as-you-go*** plan with unlimited concurrency.

Keep in mind that a page load is not necessarily the same as one output record. A single page can produce multiple records, so pagination and navigation behavior matter when estimating usage and cost.

We’ll see why this matters when we build the Texas real-estate tracker.

The main thing is, offloading these complexities means we don’t have to build, manage, or scale proxy pools and headless browsers ourselves. That gives us the architecture we really want.

Now let’s build it.

Before building the pipeline logic, let’s create a small Node.js application that will eventually consume our scraped data.

We’ll keep the project modular so the same structure can be reused with other data sources later.

Open your terminal and create the project:

```
mkdir self-healing-ai-data-pipeline
cd self-healing-ai-data-pipeline
git init
npm init -y
```

Install the dependencies:

```
npm install express dotenv
npm install -D nodemon
```

Then open the project in VS Code:

```
code .
```

We’ll build the application incrementally instead of creating a large codebase upfront.

`package.json`
Open `package.json` and configure the scripts and ECMAScript Module support:

```
{
  "type": "module",
  "scripts": {
    "start": "node src/server.js",
    "dev": "nodemon src/server.js"
  }
}
```

The `"type": "module"` setting allows us to use modern `import` and `export` syntax throughout the project.

Create a `.env` file in the project root:

```
PORT=3000
BRIGHTDATA_API_TOKEN=your_api_token
BRIGHTDATA_COLLECTOR_ID=your_collector_id
```

We’ll use the Bright Data credentials later when we connect the application to Scraper Studio.

Also create a `.gitignore` file:

```
node_modules/
.env
.DS_Store
data/*.tmp
```

This keeps local dependencies, environment variables, and temporary files out of Git.

Now create the following file: `src/server.js`

Add the initial server:

``` python
import express from "express";
import dotenv from "dotenv";
dotenv.config();
const app = express();
const PORT = process.env.PORT || 3000;
app.get("/", (req, res) => {
  res.json({
    message: "Self-healing AI Data Pipeline is Activated!"
  });
});
app.listen(PORT, () => {
  console.log(`Pipeline running on port ${PORT}`);
});
```

Start the development server:

```
npm run dev
```

You should see:

```
Pipeline running on port 3000
```

Open `http://localhost:3000` in your browser.

You should get:

```
{
  "message": "Self-healing AI Data Pipeline is Activated!"
}
```

We are now ready to step away from the text editor and construct our remote infrastructure producer. No! Not yet; we need to build our house first: the pipeline prototype.

Now that the Node.js application is running, let’s build the processing layer that sits behind it.

The goal is to keep this layer independent of any specific website or dataset. Instead of hardcoding fields such as `bedrooms`, `symbol`, or `salary`, we'll translate different source formats into one small internal contract:

```
id
value
label
```

The source-specific code will live in an `adapter`. The pipeline itself will only process this common structure.

Create the required directories and modules:

**macOS/Linux:**

```
mkdir -p src/api src/collectors src/pipeline src/storage/snapshots
mv src/server.js src/api/server.js
touch src/pipeline/normalize.js src/pipeline/validate.js src/pipeline/deduplicate.js src/pipeline/index.js src/collectors/mock.js
```

**Windows PowerShell:**

```
New-Item -ItemType Directory -Force src/api, src/collectors, src/pipeline, src/storage/snapshots
Move-Item -Path src/server.js -Destination src/api/server.js -Force
'normalize.js', 'validate.js', 'deduplicate.js', 'index.js' | ForEach-Object {
  New-Item -ItemType File -Force "src/pipeline/$_"
}
New-Item -ItemType File -Force src/collectors/mock.js
```

Because the server has moved, update the development script in `package.json`:

```
"dev": "nodemon src/api/server.js"
```

Our project now looks like this:

```
self-healing-ai-data-pipeline/
├── src/
│   ├── api/
│   │   └── server.js
│   ├── collectors/
│   │   └── mock.js
│   ├── pipeline/
│   │   ├── normalize.js
│   │   ├── validate.js
│   │   ├── deduplicate.js
│   │   └── index.js
│   └── storage/
│       └── snapshots/
├── .env
├── .gitignore
└── package.json
```

The normalizer converts messy source values into the types expected by our internal contract.

For example, a scraper might return `$502,000` as a string. Our application should receive `502000` as a number.

Open:

`src/pipeline/normalize.js`

Add:

```
/**
 * Strips formatting artifacts and typecasts values to strict numeric floats.
 */
export function toNumber(value) {
  if (value === null || value === undefined) return null;
  const cleaned = String(value).replace(/,/g, "").replace(/[^0-9.]/g, "");
  if (!cleaned) return null;
  const number = Number(cleaned);
  return Number.isFinite(number) ? number : null;
}
/**
 * Normalizes an individual raw payload token into our unified application envelope.
 */
export function normalizeRecord(record, index) {
  return {
    id: record.id ?? `entity-${index + 1}`,
    value: toNumber(record.value),
    label: typeof record.label === "string" ? record.label.replace(/\s+/g, " ").trim() : "Untitled Entry",
    captured_at: new Date().toISOString()
  };
}
export function normalizeCollection(records) {
  if (!Array.isArray(records)) return [];
  return records.map((record, index) => normalizeRecord(record, index));
}
```

The key part is that the rest of the application no longer needs to understand how the original value was formatted.

Normalization gives us a consistent shape, but consistent shape does not necessarily mean valid data.

The validator acts as the integrity gate before records continue through the pipeline.

`src/pipeline/validate.js`

```
/**
 * Evaluates core schema boundaries for an individual normalized record.
*/
export function validateRecord(record) {
  const infractions = [];
  if (!record.id || record.id.startsWith("entity-")) {
    infractions.push("Contract Failure: Missing primary unique identification parameter.");
  }
  if (record.value === null || record.value <= 0) {
    infractions.push("Contract Failure: Invalid or zero numeric metric value.");
  }
  return {
    valid: infractions.length === 0,
    errors: infractions
  };
}
export function validateCollection(records) {
  if (!Array.isArray(records)) return [];
  return records.map((record) => ({
    record,
    ...validateRecord(record)
  }));
}
```

Invalid records are not allowed to silently continue. Instead, the validation result tells us which records passed and which need attention.

Next, we need to prevent the same entity from appearing multiple times in a collection.

Open: `src/pipeline/deduplicate.js`

```
/**
 * Filters out duplicate records from the collection using our unique identifier.
*/
export function deduplicateCollection(records) {
  if (!Array.isArray(records)) return [];
  const uniqueKeys = new Set();
  return records.filter((record) => {
    if (uniqueKeys.has(record.id)) return false;
    uniqueKeys.add(record.id);
    return true;
  });
}
```

The `Set` gives us a simple in-memory lookup for identifiers we've already seen.

Now combine the individual processing stages into one entry point.

Open: `src/pipeline/index.js`

``` js
import { normalizeCollection } from "./normalize.js";
import { deduplicateCollection } from "./deduplicate.js";
import { validateCollection } from "./validate.js";
/**
 * Runs raw extraction arrays through our universal processing middle tier.
 */
export function processRawIngestion(rawData) {
  const normalized = normalizeCollection(rawData);
  const unique = deduplicateCollection(normalized);
  const validationReport = validateCollection(unique);
  const cleanRecords = validationReport.filter(item => item.valid).map(item => item.record);
  const quarantineRecords = validationReport.filter(item => !item.valid);
  return {
    pipeline_state: quarantineRecords.length > 0 ? "WARNING" : "HEALTHY",
    telemetry: {
      raw_processed: normalized.length,
      purged_duplicates: normalized.length - unique.length,
      passed_compliance: cleanRecords.length,
      failed_compliance: quarantineRecords.length
    },
    data: cleanRecords,
    quarantine: quarantineRecords
  };
}
```

Our processing flow is now:

```
Raw Data
   ↓
Normalize
   ↓
Deduplicate
   ↓
Validate
   ↓
Clean Records + Quarantine
```

This is deliberately small. Each stage has one responsibility, and the pipeline does not know where the data originally came from.

Now let’s prove that idea. Create three small mock datasets in: `src/collectors/mock.js`

Each source uses completely different field names:

``` js
// Test Datasets mimicking HAR, Yahoo Finance, and RemoteOK raw fields
const rawHarPayload = [{ "id": "har-992", "price": "$502,000", "address": "2623 Pomeran Dr" }];
const rawYahooPayload = [{ "symbol": "AAPL", "price": "$234.12", "company_name": "Apple Inc." }];
const rawRemoteOkPayload = [{ "job_id": "remote-77", "salary": "$185,000", "position": "Software Engineer" }];
/**
 * Universal Source Adapter Layer translating source-specific 
fields to our pipeline interface
 */
export function getAdaptedMockSource(source) {
  if (source === "har") {
    return rawHarPayload.map(item => ({ id: item.id, value: item.price, label: item.address }));
  }
  if (source === "yahoo") {
    return rawYahooPayload.map(item => ({ id: item.symbol, value: item.price, label: item.company_name }));
  }
  if (source === "remoteok") {
    return rawRemoteOkPayload.map(item => ({ id: item.job_id, value: item.salary, label: item.position }));
  }
  return null;
}
```

Notice what happened here.

HAR uses `price` and `address`.

Yahoo Finance uses `symbol` and `company_name`.

RemoteOK uses `salary` and `position`.

The pipeline doesn’t care.

The adapter translates each source into:

```
{
  "id": "...",
  "value": "...",
  "label": "..."
}
```

That is the Adapter Strategy we designed earlier in action.

Now expose the pipeline through a small test endpoint.

Open: `src/api/server.js`

Replace the initial server with:

``` python
import express from "express";
import dotenv from "dotenv";
import { getAdaptedMockSource } from "../collectors/mock.js";
import { processRawIngestion } from "../pipeline/index.js";
dotenv.config();
const app = express();
app.use(express.json());
const PORT = process.env.PORT || 3000;
app.get("/api/pipeline/test-mock", (req, res) => {
  const source = req.query.source; // har, yahoo, or remoteok
  const adaptedPayload = getAdaptedMockSource(source);
  if (!adaptedPayload) {
    return res.status(400).json({ error: "Invalid source flag configuration." });
  }
  try {
    const processingResult = processRawIngestion(adaptedPayload);
    res.json({
      status: "success",
      pipeline_telemetry: processingResult.telemetry,
      results: processingResult.data
    });
  } catch (err) {
    res.status(500).json({ error: err.message });
  }
});
app.listen(PORT, () => console.log(`Pipeline server online on port: ${PORT}`));
```

Start the server:

```
npm run dev
```

Then test each source:

```
curl "http://localhost:3000/api/pipeline/test-mock?source=har"
curl "http://localhost:3000/api/pipeline/test-mock?source=yahoo"
curl "http://localhost:3000/api/pipeline/test-mock?source=remoteok"
```

Each request passes through the same pipeline.

For example, the “Remoteok” request produces a normalized record similar to:

```
{
  "status": "success",
  "pipeline_telemetry": {
    "raw_processed": 1,
    "purged_duplicates": 0,
    "passed_compliance": 1,
    "failed_compliance": 0
  },
  "results": [    {
      "id": "remote-77",
      "value": 18500,
      "label": "Software engineer"
    }
  ]
}
```

We’ve now proven the core architectural idea: **different source payloads can enter the same processing pipeline without changing the pipeline code.**

The adapter handles source-specific structure. The pipeline handles application-level processing.

That separation will become vital when we replace these mock collectors with real web data.

The prototype is working, so commit the current state:

```
git add .
git commit -m "Add reusable pipeline prototype with mock sources"
```

We now have a working processing core.

Next, we’ll replace the mock source with the real web-data extraction layer.

Our business goal is not just to collect the current state of a website. We want to track how that state changes over time.

Look at a simple example:

**Yesterday:** House A — $500,000

**Today:** House A — $450,000

How does the scraper know the price dropped?

**It doesn’t.**

A scraper normally sees the current page. Remembering what happened during previous runs is the responsibility of our application.

As we discussed this concept earlier, a scraper sees the current state of a page:

```
House A → $450,000
```

The previous crawl is not automatically available to the next run.

This is where our pipeline needs to add state. We’ll store the result of the previous run as a local snapshot. When a new run completes, we’ll compare the new dataset against that snapshot.

The basic flow looks like this:

```
Current Run
     ↓
Processed Records
     ↓
latest_snapshot.json
     ↑
Previous Run
     ↓
Compare
     ↓
Created / Updated / Removed
```

For these comparisons, we’ll use JavaScript `Map` objects later in the pipeline. This gives us linear O(N) comparison work rather than repeatedly scanning the entire dataset for every record.

We don’t need a database server for this prototype. A local JSON snapshot is enough to demonstrate the state-management layer before introducing more infrastructure.

The storage directory already exists from our project structure. Now create the snapshot module:

```
mkdir -p src/storage/snapshots
touch src/pipeline/snapshot.js
```

Open `src/pipeline/snapshot.js`.

We’ll use Node.js’s native asynchronous `fs/promises` API to read and write the snapshot file.

``` python
import fs from "node:fs/promises";
import path from "node:path";
const snapshotDir = path.join(
  process.cwd(),
  "src",
  "storage",
  "snapshots"
);
const snapshotFile = path.join(
  snapshotDir,
  "latest_snapshot.json"
);
/**
 * Saves the processed dataset to local disk.
 */
export async function saveSnapshot(records) {
  try {
    await fs.mkdir(snapshotDir, { recursive: true });
    await fs.writeFile(
      snapshotFile,
      JSON.stringify(records, null, 2),
      "utf8"
    );
    return true;
  } catch (err) {
    console.error(
      `[Snapshot Write Failure] Disk IO stalled: ${err.message}`
    );
    return false;
  }
}
/**
 * Loads the previous snapshot into memory.
 */
export async function loadSnapshot() {
  try {
    const rawBuffer = await fs.readFile(
      snapshotFile,
      "utf8"
    );
    return JSON.parse(rawBuffer);
  } catch (notFoundError) {
    // Return an empty array when no previous snapshot exists.
    return [];
  }
}
```

There are only two duties here:

`saveSnapshot()` writes the latest processed records to disk.

`loadSnapshot()` retrieves the previous snapshot when the next pipeline run starts.

On the first run, there is no previous snapshot, so `loadSnapshot()` simply returns an empty array.

Before using snapshots for change detection, let’s verify that our storage layer can actually write and read data.

`src/api/server.js`

For this isolated storage test, use:

``` python
import express from "express";
import dotenv from "dotenv";
import { saveSnapshot } from "../pipeline/snapshot.js";
dotenv.config();
const app = express();
app.use(express.json());
const PORT = process.env.PORT || 3000;
app.get("/health", (req, res) => res.json({ status: "ok" }));
// Snapshot Persistence Verification Route
app.post("/api/save", async (req, res) => {
  const records = req.body.records;
  if (!Array.isArray(records)) {
    return res.status(400).json({
      error: "Invalid Payload: Input must be a 'records' array."
    });
  }
  // Basic inline mapping to standardize test elements instantly inside server context
  const standardized = records.map((item, idx) => ({
    id: item.id ?? `id-${idx + 1}`,
    value: Number(
      String(item.value ?? item.price ?? "")
        .replace(/[^0-9.]/g, "")
    ),
    label: item.label ?? item.address ?? "Untitled Entry"
  }));
  const successfullySaved = await saveSnapshot(standardized);
  if (!successfullySaved) {
    return res.status(500).json({
      error: "Failed to persist snapshot container to disk."
    });
  }
  res.json({
    message: "Snapshot saved successfully.",
    record_count: standardized.length
  });
});
app.listen(PORT, () =>
  console.log(`Pipeline server listening on port: ${PORT}`)
);
```

Reload the program:

```
npm run dev
```

Open a second PowerShell window and create a test payload:

``` php
$TestData = @{
  records = @(
    @{
      id = "123"
      price = "$450,000"
      address = "2623 Pomeran Dr"
    }
  )
} | ConvertTo-Json -Depth 10 -Compress
```

Send it to the local endpoint:

```
Invoke-RestMethod `
  -Uri "http://127.0.0.1:3000/api/save" `
  -Method POST `
  -Headers @{"Content-Type"="application/json"} `
  -Body $TestData
```

You should receive a successful response confirming that the snapshot was saved.

Now verify the actual file on disk:

```
Get-Content .\src\storage\snapshots\latest_snapshot.json
```

You should see the stored canonical record:

🧠😎 Our pipeline now has memory.

The scraper can continue to focus on extraction. The application owns the historical state.

And that gives us the missing piece for the next step: **comparing two snapshots to determine exactly what changed.**

Now that our pipeline can remember the previous run, we can FINALLY answer the question we started with:

**What changed since the last crawl?**

Consider these two snapshots:

**Yesterday:**

```
House A → $500,000
House B → $350,000
```

**Today:**

```
House A → $450,000
House B → $350,000
House C → $600,000
```

A human can immediately spot three different states:

Our pipeline needs to detect those states programmatically.

We’ll represent them with three arrays:

`created[]` — records that exist today but not in the previous snapshot.`updated[]` — records that exist in both snapshots but whose tracked value changed.`removed[]` — records that existed previously but are missing from today's dataset.
Create:

`src/pipeline/compare.js`

We could compare every incoming record against every historical record using nested loops. That approach can reach O(N²) time complexity as the dataset grows.

Instead, we’ll index both datasets by their unique IDs using JavaScript `Map`.

Building the maps takes O(N), and each lookup is approximately O(1), allowing the overall comparison to remain O(N).

Add the following to `src/pipeline/compare.js`:

```
/**
 * Executes a delta analysis between two historical runs.
 *
 * @param {Array<Object>} yesterdayBaseline - Previous snapshot.
 * @param {Array<Object>} todayIncoming - Current dataset.
 * @returns {Object} Created, updated, and removed records.
 */
export function computeHistoricalDelta(yesterdayBaseline, todayIncoming) {
  // Guard against malformed parameters to keep processing loops safe
  const baselineArray = Array.isArray(yesterdayBaseline) ? yesterdayBaseline : [];
  const incomingArray = Array.isArray(todayIncoming) ? todayIncoming : [];
  // Pre-index collections into highly efficient O(1) maps to avoid O(N^2) array scan stalls
  const baselineMap = new Map(baselineArray.map(item => [item.id, item]));
  const incomingMap = new Map(incomingArray.map(item => [item.id, item]));
  const created = [];
  const updated = [];
  const removed = [];
  // Stage 1: Isolate additions and modifications by scanning today's fresh input keys
  for (const [id, incomingItem] of incomingMap.entries()) {
    const historicalItem = baselineMap.get(id);
    // If item doesn't exist in baseline, it's a new lifecycle record
    if (!historicalItem) {
      created.push(incomingItem);
      continue;
    }
    // Identify value metric variance since the previous system sweep
    if (incomingItem.value !== historicalItem.value) {
      const numericDelta = incomingItem.value - historicalItem.value;
      // Senior Practice: Guard against division-by-zero crashes if historical value is 0
      const baseDivisor = historicalItem.value === 0 ? 1 : historicalItem.value;
      const percentageShift = (numericDelta / baseDivisor) * 100;
      updated.push({
        id,
        label: incomingItem.label,
        previous_value: historicalItem.value,
        current_value: incomingItem.value,
        delta: numericDelta,
        // Safely invoke toFixed now that non-finite Infinity properties are blocked
        percent_change: Number(percentageShift.toFixed(2))
      });
    }
  }
  // Stage 2: Isolate deletions by verifying which historical items are missing today
  for (const [id, historicalItem] of baselineMap.entries()) {
    if (!incomingMap.has(id)) {
      removed.push(historicalItem);
    }
  }
  return { created, updated, removed };
}
```

The algorithm has two passes. The first scans today’s records to find **created** and **updated** entries.

The second scans yesterday’s records to find **removed** entries.

For our real estate tracker, this gives us a clean delta such as:

```
{
  "created": ["House C"],
  "updated": ["House A"],
  "removed": []
}
```

The actual `updated` records also contain the previous value, current value, absolute delta, and percentage change. That information will become useful when we classify significant changes.

Let’s connect the comparison engine to a temporary Express endpoint and test it against two simulated crawl runs.

WRite:

``` python
import express from "express";
import dotenv from "dotenv";
import { saveSnapshot, loadSnapshot } from "../pipeline/snapshot.js";
import { computeHistoricalDelta } from "../pipeline/compare.js";
dotenv.config();
const app = express();
app.use(express.json());
const PORT = process.env.PORT || 3000;
app.get("/health", (req, res) => res.json({ status: "ok" }));
// Delta Analysis Local Verification Route
app.post("/api/analyze", async (req, res) => {
  const incomingRecords = req.body.records;
  if (!Array.isArray(incomingRecords)) {
    return res.status(400).json({ error: "Invalid Payload: Must pass an array of 'records'." });
  }
  try {
    // 1. Fetch yesterday's snapshot reference state from local disk storage
    const historicalBaseline = await loadSnapshot();
    // 2. Compute mutations across lifecycle boundaries via our comparison engine
    const deltaAnalysis = computeHistoricalDelta(historicalBaseline, incomingRecords);
    // 3. Update local snapshot file to save today's records as tomorrow's baseline
    await saveSnapshot(incomingRecords);
    res.json({
      created: deltaAnalysis.created.length,
      updated: deltaAnalysis.updated.length,
      removed: deltaAnalysis.removed.length,
      raw_deltas: deltaAnalysis
    });
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
});
app.listen(PORT, () => console.log(`Pipeline running on port: ${PORT}`));
```

Reload the server:

```
npm run dev
```

We’ll use PowerShell to mimic two separate crawl executions.

First, seed the snapshot with yesterday’s data:

``` php
$BaselineData = @{
  records = @(
    @{ id = "A"; value = 500000; label = "House A" },
    @{ id = "B"; value = 350000; label = "House B" }
  )
} | ConvertTo-Json -Depth 5 -Compress
```

Send it to the analysis endpoint:

```
Invoke-RestMethod `
  -Uri "http://127.0.0.1:3000/api/analyze" `
  -Method POST `
  -Headers @{"Content-Type"="application/json"} `
  -Body $BaselineData
```

This creates our baseline snapshot.

Now recreate today’s crawl:

``` php
$TodayData = @{
  records = @(
    @{ id = "A"; value = 450000; label = "House A" },
    @{ id = "B"; value = 350000; label = "House B" },
    @{ id = "C"; value = 600000; label = "House C" }
  )
} | ConvertTo-Json -Depth 5 -Compress
```

Send it through the same endpoint:

```
Invoke-RestMethod `
  -Uri "http://127.0.0.1:3000/api/analyze" `
  -Method POST `
  -Headers @{"Content-Type"="application/json"} `
  -Body $TodayData
```

The backend loads the previous snapshot, compares it with today’s records, and then saves today’s dataset as the new baseline.

You should see counts similar to:

```
created : 1
updated : 1
removed : 0
```

*Note:**I run the seeding scripts two times during local testing; that's why the terminal is showing “*`_created: 2_`*"*

The detailed response should identify House C as created and House A as updated.

There’s an important detail here.

After the request finishes, today’s dataset becomes `latest_snapshot.json`.

That means the next execution will compare against **today**, not the original baseline.

This is what turns a simple file into a rolling state mechanism:

```
Run 1
   ↓
Snapshot A
Run 2
   ↓
Compare against Snapshot A
   ↓
Save Snapshot B
Run 3
   ↓
Compare against Snapshot B
   ↓
Save Snapshot C
```

Our pipeline can now detect historical changes.

But not every change deserves the same response.

A $5,000 price adjustment and a 90% price collapse are both technically `updated` events, but they shouldn't necessarily receive the same treatment.

That’s where event classification comes in.

For this prototype, we’ll classify value changes using configurable thresholds.

These numbers are our example thresholds, not universal rules:

`**MINOR**``** MAJOR**``** CRITICAL**`
A large change could be a genuine business event, but it could also indicate an upstream extraction problem. The classification layer doesn’t decide which one it is. It simply makes the change visible and machine-readable.

Go to: `src/pipeline/classify.js`

```
/**
 * Classifies calculated deltas into structured system events.
 *
 * @param {Object} deltas - Created, updated, and removed records.
 * @returns {Array<Object>} Structured event timeline.
 */
export function classifySystemEvents(deltas) {
  const alertsTimeline = [];
// Process value changes.
  for (const change of deltas.updated) {
    let severity = "MINOR";
    let type = "VALUE_MUTATED";
    const absoluteShift = Math.abs(
      change.percent_change
    );
    if (absoluteShift >= 15 && absoluteShift < 30) {
      severity = "MAJOR";
      type = "SIGNIFICANT_VALUE_SHIFT";
    }
    if (change.percent_change <= -30) {
      severity = "CRITICAL";
      type = "ANOMALOUS_VALUE_COLLAPSE";
    }
    alertsTimeline.push({
      id: change.id,
      type,
      severity,
      message: `${change.label} (ID: ${change.id}) shifted by ${change.percent_change}%.`,
      details: change
    });
  }
  // Process newly created records.
  for (const record of deltas.created) {
    alertsTimeline.push({
      id: record.id,
      type: "RECORD_CREATED",
      severity: "MINOR",
      message: `New entry mapped to tracking logs: ${record.label}`,
      details: record
    });
  }
  // Process removed records.
  for (const record of deltas.removed) {
    alertsTimeline.push({
      id: record.id,
      type: "RECORD_DELETED",
      severity: "MINOR",
      message: `Entry removed from upstream dataset: ${record.label}`,
      details: record
    });
  }
  return alertsTimeline;
}
```

Now the pipeline doesn’t just say:

```
House A changed.
```

It can produce a structured event:

```
{
  "id": "A",
  "type": "ANOMALOUS_VALUE_COLLAPSE",
  "severity": "CRITICAL"
}
```

That distinction becomes valuable when another system needs to consume these events.

Now integrate the classifier into our Express endpoint.

Write:

``` python
import express from "express";
import dotenv from "dotenv";
import { saveSnapshot, loadSnapshot } from "../pipeline/snapshot.js";
import { computeHistoricalDelta } from "../pipeline/compare.js";
import { classifySystemEvents } from "../pipeline/classify.js";
dotenv.config();
const app = express();
app.use(express.json());
const PORT = process.env.PORT || 3000;
app.post("/api/events", async (req, res) => {
  const incomingRecords = req.body.records;
  if (!Array.isArray(incomingRecords)) {
    return res.status(400).json({
      error: "Invalid Payload."
    });
  }
  try {
    const historicalBaseline = await loadSnapshot();
    const deltaAnalysis = computeHistoricalDelta(
      historicalBaseline,
      incomingRecords
    );
    const structuredAlertsTimeline =
      classifySystemEvents(deltaAnalysis);
    await saveSnapshot(incomingRecords);
    const criticalAnomaly =
      structuredAlertsTimeline.find(
        (event) => event.severity === "CRITICAL"
      );
    if (criticalAnomaly) {
      return res.json({
        severity: "CRITICAL",
        message: `${criticalAnomaly.details.label} lost ${Math.abs(criticalAnomaly.details.percent_change)}% of its value.`,
        timeline: structuredAlertsTimeline
      });
    }
    res.json({
      severity: "HEALTHY",
      message:
        "Ingestion finalized with no critical anomalies registered.",
      timeline: structuredAlertsTimeline
    });
  } catch (error) {
    res.status(500).json({
      error: error.message
    });
  }
});
app.listen(PORT, () =>
  console.log(`Event Classification server active on port: ${PORT}`)
);
```

Let’s deliberately create a large negative change and see whether our classifier catches it.

`src/storage/snapshots/latest_snapshot.json`

Replace its contents with:

```
[  {
    "id": "A",
    "value": 5000000,
    "label": "House A"
  }
]
```

Now create the test payload in PowerShell:

``` php
$AnomalyTestData = @{
  records = @(
    @{ id = "A"; value = 325000; label = "House A" }
  )
} | ConvertTo-Json -Depth 5 -Compress
```

Send it to `/api/events`:

```
Invoke-RestMethod `
  -Uri "http://127.0.0.1:3000/api/events" `
  -Method POST `
  -Headers @{"Content-Type"="application/json"} `
  -Body $AnomalyTestData
```

The value changes from $5,000,000 to $325,000, which is a **-93.5% change**.

Because that is below our -30% threshold, the classifier should return a `CRITICAL` event. Have a look:

Now run the same request again. This time, the result should no longer be critical.

**Why?**

Because the previous request already saved `$325,000` as the latest snapshot. The second request is therefore comparing `$325,000` against `$325,000`, producing no value change.

That’s the rolling state mechanism working as intended. Have a look:

We now have three important abilities:

```
Extract
  ↓
Remember
  ↓
Compare
  ↓
Classify
  ↓
Machine-readable Events
```

The pipeline can now detect when its data changes and distinguish ordinary updates from unusually large shifts.

The next challenge is more interesting: **what happens when the scraper itself breaks?**

Let’s deliberately break the extraction layer and see whether the rest of our pipeline can detect the failure.

So far, our pipeline has been working with local datasets.

That was intentional. We first built and tested the processing layer independently from external extraction services. Now we can connect that pipeline to a real web data provider without mixing provider-specific logic into our application.

Our pipeline already knows how to:

The next step is to make the extraction layer replaceable.

If we change scraping providers later, the processing pipeline should not need to change.

A common mistake is to put provider-specific code directly inside the main pipeline.

That creates a dependency like this:

```
Node.js Pipeline
      ↓
Bright Data-specific code
      ↓
Bright Data API
```

If we later switch providers, we have to modify the pipeline itself.

Instead, we’ll put an abstraction between the pipeline and the extraction provider:

```
                  ┌─────────────────────┐
                  │   Node.js Pipeline  │
                  └──────────┬──────────┘
                             │
                    Collector Interface
                             │
              ┌──────────────┴──────────────┐
              │                             │
   ScraperStudioCollector          FirecrawlCollector
              │                             │
        Bright Data                    Firecrawl
```

The pipeline only needs one contract: **Give the collector a target and receive structured records.**

The implementation behind that contract can change.

Our collector directory will look like this:

```
src/
└── collectors/
    ├── BaseCollector.js
    ├── ScraperStudioCollector.js
    ├── FirecrawlCollector.js
     ...other collectors
    └── CollectorFactory.js
```

This is the practical side of the [**Port and Adapter**](https://angular.love/ports-and-adapters-vs-hexagonal-architecture-is-it-the-same-pattern) pattern we introduced earlier: the application depends on an interface, while individual providers implement that interface.

`src/collectors/BaseCollector.js`

This class defines the contract that every collector must follow.

```
/**
 * Abstract Base Collector Interface.
 */
export class BaseCollector {
  /**
   * Programmatically retrieves raw, unsanitized extraction payload
   * arrays from a source target.
   *
   * @param {string} url - Target public website URL.
   * @returns {Promise<Array<Object>>} - Unnormalized data array.
   */
  async collect(url) {
    throw new Error(
      "Architecture Violation: Method 'collect(url)' must be implemented by subclass."
    );
  }
}
```

The base class does not know anything about Bright Data, Firecrawl, Playwright, or any other extraction system.

It only defines the expected behavior.

Every concrete collector must implement `collect(url)`.

Our primary provider will be Bright Data Scraper Studio.

`src/collectors/ScraperStudioCollector.js`

``` js
import { BaseCollector } from "./BaseCollector.js";
/**
 * Our primary provider (infrastructure layer).
 */
export class ScraperStudioCollector extends BaseCollector {
  async collect(url) {
    // Provider-specific implementation will be added here.
    throw new Error(
      "Bright Data Infrastructure Pipeline Placeholder. Activate via configuration parameters."
    );
  }
}
```

For now, this is intentionally a placeholder.

We are testing the architecture before adding the external API implementation. This makes it easier to verify that the rest of the application does not depend on Bright Data-specific details.

Now let’s add a second collector to prove that the abstraction is not tied to one provider.

`src/collectors/FirecrawlCollector.js`

``` js
import { BaseCollector } from "./BaseCollector.js";
/**
 * Alternative web crawler provider used to demonstrate
 * multi-provider collector architecture.
 */
export class FirecrawlCollector extends BaseCollector {
  async collect(url) {
    // Provider-specific implementation will be added here.
    throw new Error(
      "Firecrawl Infrastructure Pipeline Placeholder. Activate via configuration parameters."
    );
  }
}
```

Again, this is only an architectural placeholder.

The important part is that both providers expose the same `collect(url)` method.

The pipeline does not need to know which provider is behind it.

Now we need a single place that decides which provider to instantiate.

`src/collectors/CollectorFactory.js`

``` js
import { ScraperStudioCollector } from "./ScraperStudioCollector.js";
import { FirecrawlCollector } from "./FirecrawlCollector.js";
export class CollectorFactory {
  /**
   * Instantiates a concrete data provider dynamically
   * based on the configured provider type.
   *
   * @param {string} providerType - Collector identifier.
   */
  static create(providerType) {
    const type = String(providerType).toUpperCase();
    if (type === "SCRAPER_STUDIO") {
      return new ScraperStudioCollector();
    }
    if (type === "FIRECRAWL") {
      return new FirecrawlCollector();
    }
    throw new Error(
      `Factory Exception: Unsupported collector provider type [${providerType}]`
    );
  }
}
```

Now the application can request a collector without directly importing or constructing a provider.

For example:

``` js
const collector = CollectorFactory.create("SCRAPER_STUDIO");
```

Later, changing the configuration to:

``` js
const collector = CollectorFactory.create("FIRECRAWL");
```

changes the concrete implementation while keeping the pipeline interface the same.

That is the main benefit of the factory: provider selection is centralized instead of scattered throughout the application.

Before connecting real external services, let’s verify that the abstraction works at runtime.

`src/testCollector.js`

``` python
import dotenv from "dotenv";
import { CollectorFactory } from "./collectors/CollectorFactory.js";
dotenv.config();
async function runSanityCheck() {
  console.log("======================================================================");
  console.log("           COLLECTOR ABSTRACTION ARCHITECTURE SANITY CHECK");
  console.log("======================================================================");
  try {
    // Test 1: Verify factory error mechanics on unsupported providers
    try {
      console.log("[Factory Test] Attempting to create an invalid provider...");
      CollectorFactory.create("UNKNOWN_PROVIDER");
    } catch (e) {
      console.log(
        `✅ Success: Factory correctly caught boundary exception -> "${e.message}"`
      );
    }
    // Test 2: Verify dependency inversion polymorphism
    console.log(
      "\n[Factory Test] Creating concrete Scraper Studio class polymorphically..."
    );
    const collector = CollectorFactory.create("SCRAPER_STUDIO");
    console.log(
      `✅ Success: Instance verified. Class Name: ${collector.constructor.name}`
    );
    console.log("----------------------------------------------------------------------");
    console.log("Result Status: ARCHITECTURE CHECK PASSED.");
    console.log("======================================================================");
  } catch (err) {
    console.error(`❌ Unexpected Architecture Error: ${err.message}`);
  }
}
runSanityCheck();
```

Save all modules. Run the verification script from your terminal:

```
node src/testCollector.js
```

**Output:**

The test verifies two important behaviors.

First, an unsupported provider is rejected at the factory boundary.

Second, the factory can create a concrete collector through the common abstraction.

We haven’t connected a real external provider yet. That’s deliberate.

The architecture is now ready for that integration without forcing provider-specific code into the processing pipeline.

Commit the current changes if you want:

```
git add .
git commit -m "add polymorphic collector factory layer + change detection"
```

The next step is where this abstraction becomes useful: we’ll replace the Scraper Studio placeholder with the **real Bright Data integration** and start feeding live web data into the pipeline.

Our processing pipeline is ready. We now need to connect it to a real extraction source.

For this project, we’ll use Bright Data Scraper Studio's built-in Web IDE AI agent to generate the scraper and expose the resulting data to our Node.js pipeline.

Our target is a live Texas real-estate listing page on [HAR.com](https://www.har.com/houston/realestate/for_sale):

The goal isn’t to build a HAR.com-specific application.

We’re using real estate as the case study because it gives us a useful combination of structured fields, detail pages, pagination, and changing values such as property prices.

The processing pipeline remains independent of the source.

[Sign in to your Bright Data dashboard](https://get.brightdata.com/scraping-browser-signup) and open **Scraper Studio** from the workspace.

You’ll be prompted to enter a target URL.

For this project, enter: [https://www.har.com/houston/realestate/for_sale](https://www.har.com/houston/realestate/for_sale)

Below the URL field, you’ll find an optional **Add additional instructions** field.

The AI agent can inspect the target and ask questions during the configuration process, but providing the extraction requirements upfront gives it a clearer starting point.

For our real-estate tracker, use this custom prompt:

```
"Collect property information from the listings.
Follow links from the search results page into individual property pages.
Extract:
- Price
- Address
- Bedrooms
- Full bathrooms
- Half bathrooms
- Square footage
- Lot size
- Property type
- MLS number
- Listing status
Handle pagination across multiple result pages.
Exclude advertisements, sponsored placements, and sidebar links.
Return a structured JSON array."
```

After processing the target, Scraper Studio generates a proposed data schema as below:

Review the proposed fields before approving them.

*The Generated Schema Is Temporary:*

The schema generated by Scraper Studio describes the **external source**. It is not the same thing as the internal data contract we designed earlier.

Our Node.js pipeline should not care whether the scraper gets a value from a CSS selector, an AI-generated extraction rule, or a completely different provider. The external extraction layer can change. The internal contract should remain stable.This separation is what allows us to replace the extraction provider later without rewriting the processing pipeline.

If you don’t provide enough information in the additional instructions, Scraper Studio can guide you through an interactive configuration process.

Depending on the target, you may be asked whether you want to:

For our use case, select “**Extract data from individual property pages”**.

The listing page gives us links to properties, while the individual pages contain richer property information.

Next, Scraper Studio can ask whether pagination should be handled. Select **Yes**. This tells the scraper that the collection should account for additional result pages rather than stopping at the first page.

After the page analysis finishes, Scraper Studio presents the proposed extraction fields as below:

Review the fields and click **Approve**.

Now, we have our first functional scraper without writing the extraction logic ourselves. But we’re not going to treat the generated scraper as a black box.

The exciting part starts when we open the generated code.

After approval, Scraper Studio opens the scraper inside its workspace editor.

The generated JavaScript is divided into two important parts:

Understanding this distinction is useful because these two layers solve different problems.

The interaction layer decides *where the browser goes.* The parser decides *what data to extract from the page.*

**1. Auditing the Interaction Code**

The interaction code controls browser navigation, pagination, and the movement between crawling stages. Let's demonstrate key ideas behind this:

FunctionResponsibility`navigate()` Opens the target URL in the browser session.`parse()` Passes the current page to the parser.`rerun_stage()` Runs another instance of a stage, useful for distributing pagination work.`next_stage()` Passes discovered URLs or data into another crawling stage.

The scraper isn’t necessarily limited to one long browser session walking through every page sequentially. Instead, the generated workflow can split work into separate stages.

**2. Auditing the Parser Code**

When the interaction layer calls `parse()`, Scraper Studio passes the current page to the parser.

The parser is responsible for turning page content into structured records.

One useful pattern you’ll see in generated parser code is converting matching elements into an array before mapping them into structured results.

For example, using `.toArray().map()` keeps the transformation in a single expression and produces a normal JavaScript array that can be passed into the next stage.

More importantly, notice what the parser is doing in our workflow. It doesn’t try to extract every property field from the search-result card.

Instead, it first discovers the property URLs. Those URLs can then become inputs to the next crawling stage.

This is precisely the kind of multi-stage workflow we want for our tracker. But our Node.js application doesn’t know how to start this scraper yet.

So the next step is to connect the Scraper Studio collector to our **Collector abstraction** through its API by replacing the placeholder implementation from the previous section with a real cloud collection request.

Unlike traditional scraping frameworks, Scraper Studio wraps the browser automation layer behind two HTTP endpoints:

`POST /dca/trigger` — Start a new collection job

`GET /dca/dataset` — Download the completed dataset

Open your published scraper in the Scraper Studio workspace and open the **Initiate by API** tab:

You need [two values](https://docs.brightdata.com/products/scraper-studio/quickstart):

`c_` inside your scraper settings).
Now open your local `.env` file and add the values:

```
BRIGHTDATA_API_TOKEN=""
BRIGHTDATA_COLLECTOR_ID=""
```

Keep these values private. Never commit your `.env` file to Git.

Create an isolated new file for testing scraper studio API; we will later connect with baseCollector:

```
src/collectors/scraperStudio.js
```

We will use Node.js’s built-in `fetch()` to send the request to Scraper Studio.

The function below starts a new collection job and returns the response from Bright Data:

``` python
import dotenv from "dotenv";
dotenv.config();
/**
 * Dispatches an asynchronous trigger call to Scraper Studio's cloud queue.
 * @param {string} targetUrl - The target URL to scrape.
 * @returns {Promise<Object>} The asynchronous trigger response.
 */
export async function triggerRemoteScraper(targetUrl) {
  const token = process.env.BRIGHTDATA_API_TOKEN;
  const collectorId = process.env.BRIGHTDATA_COLLECTOR_ID;
  if (!token || !collectorId) {
    throw new Error(
      "Adapter Error: Missing valid Bright Data workspace credentials inside .env file."
    );
  }
  const triggerEndpoint = `https://api.brightdata.com/dca/trigger?collector=${collectorId}&queue_next=1`;
  const parameterPayload = JSON.stringify([    {
      url: targetUrl,
      max_page: 2 // Bound pagination limits during sandbox evaluation passes
    }
  ]);
  console.log(
    `[Cloud Post] Triggering remote automation collection for target: ${targetUrl}`
  );
  const response = await fetch(triggerEndpoint, {
    method: "POST",
    headers: {
      Authorization: `Bearer ${token}`,
      "Content-Type": "application/json"
    },
    body: parameterPayload
  });
  if (!response.ok) {
    throw new Error(
      `Cloud Trigger Rejection: HTTP status code [${response.status}]`
    );
  }
  return await response.json();
}
```

Notice that we are not launching a browser from Node.js. Our application simply sends the target URL to Scraper Studio. The cloud service takes care of the browser automation and collection job.

The `max_page` value is also intentionally limited here. During development and testing, this prevents an accidental request from crawling hundreds or thousands of pages.

Next, connect the scraper trigger to our Express server.

```
src/api/server.js
python
import express from "express";
import dotenv from "dotenv";
import { triggerRemoteScraper } from "../collectors/scraperStudio.js";
import { processRawIngestion } from "../pipeline/index.js";
dotenv.config();
const app = express();
app.use(express.json());
const PORT = process.env.PORT || 3000;
app.get("/health", (req, res) => res.json({ status: "ok" }));
// Asynchronous Trigger Endpoint Route
app.get("/api/collect", async (req, res) => {
  const baselineUrl =
    "https://www.har.com/houston/realestate/for_sale";
  const targetUrl = req.query.url || baselineUrl;
  try {
    const triggerReceipt = await triggerRemoteScraper(targetUrl);
    res.json(triggerReceipt);
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
});
app.listen(PORT, () =>
  console.log(`Pipeline Server actively running on port: ${PORT}`)
);
```

The new `/api/collect` route gives us a simple way to trigger the cloud scraper from our local application. If no URL is provided, it uses our Houston real-estate page as the default.

You could also pass another URL through the query string.

```
npm run dev
```

You should see your server’s startup message in the terminal.

Now open a second PowerShell or terminal window and run:

```
curl "http://127.0.0.1:3000/api/collect"
```

Your local Express server receives the request and calls the Scraper Studio API. If everything is configured correctly, Bright Data returns a successful trigger response as below:

This confirms that our Node.js application can communicate with Scraper Studio and start a real cloud collection job.

We have now moved from a local mock input to a real remote scraper.

After triggering the collector, open your Scraper Studio workspace and select the **Runs** tab.

You should see the collection job running there:

The Runs page lets you monitor the collection and access the resulting dataset. Scraper Studio provides export formats such as:

`JSON`
`CSV`
`XLSX`
`NDJSON`
At this point, our Node.js application can successfully start a remote scraping job.

But we still have a problem.

We do not want a human to open Scraper Studio every time, wait for a run to finish, and manually download a file.

The next step is to bring the completed dataset back into our Node.js pipeline programmatically. That will let our application **fetch, save, process, and analyze the data dynamically**.

Every chapter up to this point built one isolated piece of the system: a stable contract, a pipeline that normalizes and validates anything shaped like `{ id, value, label }`, a snapshot engine that gives the pipeline memory, a comparison engine that turns two snapshots into `created / updated / removed`, an event classifier that separates market noise from real anomalies, a polymorphic collector factory, and a live Scraper Studio trigger endpoint.

None of that is worth much until it survives contact with a real website.

This chapter is the *payoff*.

We will build a Texas real estate price tracker for Houston and Dallas cities. It will pull live listings from HAR.com through Scraper Studio and send them through the same pipeline we already built and tested.

The person running the tracker never needs to see a HAR.com URL, paste a property URL, or use the Collection API manually. They simply choose a city from a menu and watch real listings move through the pipeline.

There is no new architecture here. Only the pieces that genuinely need to change for real-world data will change.

Before connecting everything from Node.js, we must customize the scraper’s **pagination logic** in the real-world extraction workflow. This matters because a real website produces gigantic data than a small mock dataset. You will waste hours/reducing your valuable free credits for testing a single function if you don’t know how to handle pagination.

Let's look at our generated har.com scraper’s interaction code on scraper studio IDE:

**So where’s the actual risk for a repeatable real-world testing workflow?** It’s in what the “cap” is actually capping:

``` js
const max_page = Math.min(input.max_page || 10, 84); 
// HAR.com has 84 pages max (by default):
```

At first glance, `84` looks like a safety limit. It is not.

It is simply the current maximum number of pages for this HAR.com search. `Math.min(x, 84)` prevents a request from asking for page 85, but it still allows a request to crawl all 84 pages.

There is another problem. If `max_page` is not supplied, the scraper falls back to `10`. That sounds small until we look at what happens next.

Every property URL found on a search page becomes its own `next_stage()` call. In other words, every listing can create another browser job.

If a search page contains roughly 20 listings, the numbers quickly grow:

Here is how much time it takes for [har.com](http://har.com) product page after I limit it to 242 pages. You have to wait 3–8 minutes for testing each request as below:

None of this is a runaway bug. The scraper is doing exactly what we asked it to do. But it is a bad default for a tutorial where we may run the scraper repeatedly while testing snapshots, comparisons, and anomaly detection.

I have also seen much larger crawls in testing. One request produced more than 1,000 pages, while another test with [CoinMarketCap](http://coinmarketcap.com) reached more than 8,000 pages. That is a lot of browser work for a simple test.

Bright Data’s batch scraper concurrency can reach up to 100 concurrent jobs per scraper. Once that limit is exceeded, the API can return:

```
Maximum limit of 100 jobs per scraper has been exceeded. Please reduce the number of parallel jobs.
```

So the issue is not that our scraper is broken. The issue is that the default is too large for a repeatable demo.

**Put an explicit cap on the scraper:**

To fix that long-waiting scraping job for each test, we want a small and predictable run for local testing.

We will:

Therefore, rewrite the “**interaction code”** in the Scraper Studio IDE as follows:

``` js
const url = new URL(input.url);
// Guard: If this is an isolated property detail page, navigate, parse, and exit immediately.
// This prevents detail page crawls from recursively triggering search directory loops.
if (input.is_detail) {
    navigate(url.href);
    collect(parse());
    return;
}
// Otherwise, we are on the main search directory page
navigate(url.href);
const { property_urls } = parse();
// 💡 EXPLICIT PAGINATION CAP: Limit listing discovery to exactly 2 pages max
if (!input.is_rerun) {
    const max_page = 2; // Hard cap listing crawl page depth

    for (let page = 2; page <= max_page; page++) {
        const next_page_url = new URL(input.url);
        next_page_url.searchParams.set('page', page.toString());

        rerun_stage({
            url: next_page_url.href,
            is_rerun: true
        });
    }
}
// 💡 EXPLICIT CRAWL CAP: Limit deep property-detail crawls to exactly 5 listings per page
const target_listings = (property_urls || []).slice(0, 5);
for (let property_url of target_listings) {
    next_stage({
        url: property_url,
        is_detail: true // Flag to activate our detail-page guard in the next stage
    });
}
```

With that customized interaction code, you’ll see each API request takes only 30 seconds on average in IDE “Recent Runs” tab as below:

You may still see more pages than expected in some runs. In my case, the scraper returned 12 pages even though the interaction code was capped at `2`. The parent worker spawned the two intended pages, but some sub-workers also discovered pagination links inside the site’s footer.

This is a useful reminder: a crawl limit in one part of a scraper does not automatically prevent a website’s own links from creating additional work.

After changing the interaction code, save it and click **Save to production**, then click **Continue**.

*Note:**Scraper Studio keeps the draft and production/development versions separate. Our Node.js specifically trigger runs the published production version, not an unsaved draft.*

Always sync the production version after manually changing the scraper code. You can verify the production update from the **Changelog** tab in the top-right corner:

Open **Delivery preferences** in the scraper dashboard.

You will see several choices:

For this tracker, we want the complete dataset after the collection finishes.

So we use:

That configuration is necessary because it determines how our Node.js collector gets the data.

Our flow is:

```
POST /dca/trigger
        ↓
{ collection_id }
        ↓
GET /dca/dataset?id=...
        ↓
[ records ]
```

The dataset becomes available after the batch job finishes. If you choose a webhook instead, the architecture changes. Scraper Studio would push the results to your application instead of your application polling for them.

For this guide, we will keep **API download**.

Our trigger endpoint from Chapter 10 starts a collection and returns its receipt. That proved our connection worked.

But a real tracker needs to wait for the completed dataset.

Bright Data’s Collection API returns a status object while the dataset is still being built and returns the completed records when the job is finished. That means our collector needs to poll the dataset endpoint until the result is ready.

There are also two different timeout problems we need to keep separate:

A single network request should fail relatively quickly. A real browser crawl, however, may legitimately take several minutes.

So a slow or broken network request should be retried, while a collection that never finishes should eventually stop.

Open: `src/collectors/ScraperStudioCollector.js`

and update it with this below:

``` js
// src/collectors/ScraperStudioCollector.js
import { BaseCollector } from "./BaseCollector.js";
const TRIGGER_ENDPOINT = "https://api.brightdata.com/dca/trigger";
const DATASET_ENDPOINT = "https://api.brightdata.com/dca/dataset";
const POLL_INTERVAL_MS = 5000;              // how often we check in
const REQUEST_TIMEOUT_MS = 10000;           // how long ONE http call may hang
const MAX_COLLECTION_WAIT_MS = 20 * 60 * 1000; // how long the WHOLE job may take
function sleep(ms) {
  return new Promise((resolve) => setTimeout(resolve, ms));
}
export class ScraperStudioCollector extends BaseCollector {
  constructor() {
    super();
    this.token = process.env.BRIGHTDATA_API_TOKEN;
    this.collectorId = process.env.BRIGHTDATA_COLLECTOR_ID;
    if (!this.token || !this.collectorId) {
      throw new Error("Missing BRIGHTDATA_API_TOKEN or BRIGHTDATA_COLLECTOR_ID in your .env file.");
    }
  }
  /**
   * Fires one HTTP request with its own short timeout. This protects us
   * against a dead socket — it has nothing to do with how long the
   * overall scraping job is allowed to run.
   */
  async #request(url, options = {}) {
    const controller = new AbortController();
    const timer = setTimeout(() => controller.abort(), REQUEST_TIMEOUT_MS);
    try {
      return await fetch(url, { ...options, signal: controller.signal });
    } finally {
      clearTimeout(timer);
    }
  }
  async collect(url) {
    const collectionId = await this.#trigger(url);
    return this.#waitForDataset(collectionId);
  }
  async #trigger(targetUrl) {
    const triggerUrl = `TRIGGERENDPOINT?collector={this.collectorId}&queue_next=1`;
    const response = await this.#request(triggerUrl, {
      method: "POST",
      headers: {
        Authorization: `Bearer ${this.token}`,
        "Content-Type": "application/json"
      },
      body: JSON.stringify([{ url: targetUrl, max_page: 2 }])
    });
    if ([401, 403, 407].includes(response.status)) {
      throw new Error(`Scraper Studio rejected the request credentials [HTTP ${response.status}]. Check BRIGHTDATA_API_TOKEN.`);
    }
    if (!response.ok) {
      throw new Error(`Scraper Studio trigger failed [HTTP ${response.status}]`);
    }
    const body = await response.json();
    const collectionId = body.collection_id || body.id;
    if (!collectionId) {
      throw new Error("Trigger response did not include a collection id.");
    }
    console.log(`[Scraper Studio] Job triggered. Collection ID: ${collectionId}`);
    return collectionId;
  }
  async #waitForDataset(collectionId) {
    const datasetUrl = `${DATASET_ENDPOINT}?id=${collectionId}`;
    const startedAt = Date.now();
    let lastStatus = "INITIALIZING";
    while (Date.now() - startedAt < MAX_COLLECTION_WAIT_MS) {
      const elapsedSec = Math.floor((Date.now() - startedAt) / 1000);
      process.stdout.write(`\r[Scraper Studio] ${lastStatus.padEnd(12)} elapsed: ${elapsedSec}s   `);
      let response;
      try {
        response = await this.#request(datasetUrl, {
          headers: { Authorization: `Bearer ${this.token}` }
        });
      } catch (requestError) {
        // A single request timing out or dropping is a network hiccup,
        // not a job failure — try again on the next polling pass.
        await sleep(POLL_INTERVAL_MS);
        continue;
      }
      // Auth problems are a request problem, not a "still running" state.
      // Fail immediately instead of burning the 20-minute budget on a
      // token that will never work.
      if ([401, 403, 407].includes(response.status)) {
        process.stdout.write("\n");
        throw new Error(`Scraper Studio rejected the dataset request [HTTP ${response.status}]. Check BRIGHTDATA_API_TOKEN.`);
      }
      // A brief 404 right after triggering (before the snapshot id has
      // fully registered) is normal and self-resolves within a poll or
      // two. Any other 4xx means something about the request is wrong
      // and won't fix itself by waiting, so we fail fast on those.
      if (response.status === 404 && elapsedSec < 30) {
        lastStatus = "INITIALIZING";
        await sleep(POLL_INTERVAL_MS);
        continue;
      }
      if (response.status >= 400 && response.status < 500) {
        process.stdout.write("\n");
        throw new Error(`Scraper Studio rejected the dataset request [HTTP ${response.status}]`);
      }
      // 5xx and anything else non-2xx is treated as transient — the job
      // itself is still fine, only this one poll failed.
      if (!response.ok) {
        lastStatus = "RETRYING";
        await sleep(POLL_INTERVAL_MS);
        continue;
      }
      const body = await response.json();
      // Bright Data's documented contract: a plain JSON array means the
      // snapshot is finished. Anything else — e.g. { status: "building" }
      // — means the job is still running.
      if (Array.isArray(body)) {
        process.stdout.write("\n");
        console.log(`[Scraper Studio] Collection complete. Records received: ${body.length}`);
        return body;
      }
      lastStatus = body.status ? String(body.status).toUpperCase() : "BUILDING";
      await sleep(POLL_INTERVAL_MS);
    }
    process.stdout.write("\n");
    throw new Error(`Scraper Studio job ${collectionId} did not finish within the 20-minute wait budget.`);
  }
}
```

**Here is what each part protects us from:**

Our scraper implements a fail-fast polling architecture designed to handle diverse network conditions. It uses `AbortController` timeouts and strict HTTP error handling to **fail immediately on unrecoverable 4xx client errors, while safely retrying transient 5xx server errors** and tolerating brief 404 windows for delayed datasets. Finally, strict ceilings like `MAX_COLLECTION_WAIT_MS` prevent infinite hangs, while a live status line ensures full operational visibility.

The core distinction is that one request timing out does not fail the collection. A collection that never finishes does.

Bright Data’s [quickstart](https://docs.brightdata.com/products/scraper-studio/quickstart) uses a shorter overall wait for simpler jobs. We use 20 minutes because we may scrape thousands of pages using our pipeline in a production app in the future, and the scraper will be doing more work.

Now we can build the part that is specific to our Texas real estate model.

The collector gives us raw HAR.com records. Our pipeline, however, expects the stable contract:

```
id
value
label
...metadata
```

So we need an adapter between the two.

First, let’s look at a live JSON record returned by Scraper Studio:

```
{
  "price": { "value": 305000, "currency": "USD", "symbol": "$" },
  "address": "5238 Kylie Springs Ln, Houston, TX 77066",
  "bedrooms": 4,
  "bathrooms_full": 2,
  "bathrooms_half": 1,
  "square_feet": 2479,
  "lot_size": 10724,
  "property_type": "Single-Family",
  "mls_number": "53228136",
  "listing_status": "For Sale",
  "product_page_url": "https://www.har.com/homedetail/5238-kylie-springs-ln-houston-tx-77066/3690464",
  "input": { "url": "https://www.har.com/houston/realestate/for_sale", "max_page": 2 }
}
```

Two fields worth noticing before we map anything:

**1.** `**price**` **is an object**

The price is not simply: `305000`

Instead, Scraper Studio returns:

```
price.value
price.currency
price.symbol
```

We only need `price.value` for the main pipeline value.

**2.** `**input**` **is request metadata**

The `input` object tells us which URL and settings produced the record.

That is useful while debugging. But it is not part of the property itself.

Our adapter will ignore it.

```
src/pipeline/adapters.js
```

and replace its contents:

```
// src/pipeline/adapters.js
/**
 * Texas city registry. Each entry pairs a simple selection key with the
 * real HAR.com search URL our published Scraper Studio scraper targets.
 * The person running the tracker never sees or types this URL — they
 * just pick a city from a menu.
 */
export const cityRegistry = {
  1: { key: "houston", name: "Houston", url: "https://www.har.com/houston/realestate/for_sale" },
  2: { key: "dallas", name: "Dallas", url: "https://www.har.com/dallas/realestate/for_sale" }
};
/**
 * Universal Dataset Adapter.
 * Converts HAR.com's raw field names into our stable pipeline contract:
 * id, value, label, metadata. This is the ONLY file that knows HAR.com's
 * field names — normalize.js, validate.js, and compare.js never do.
 * Note that `input` (the echoed request) is deliberately never read here.
 */
export class DatasetAdapter {
  /**
   * @param {Array<Object>} rawData - Raw records from the Scraper Studio collector.
   * @param {string} city - The city key ("houston" | "dallas") for tagging.
   */
  static transform(rawData, city) {
    if (!Array.isArray(rawData)) return [];
    return rawData.map((item) => ({
      id: item.mls_number ? String(item.mls_number) : null,
      value: item.price ? item.price.value : null,
      label: item.address ? String(item.address).trim() : "Untitled Listing",
      metadata: {
        city,
        currency: item.price ? item.price.currency : "USD",
        currency_symbol: item.price ? item.price.symbol : "$",
        bedrooms: toInt(item.bedrooms),
        bathrooms_full: toInt(item.bathrooms_full),
        bathrooms_half: toInt(item.bathrooms_half),
        square_feet: toInt(item.square_feet),
        lot_size: toInt(item.lot_size),
        property_type: item.property_type ?? null,
        listing_status: item.listing_status ?? null,
        product_page_url: item.product_page_url ?? null
      }
    }));
  }
}
function toInt(value) {
  const number = parseInt(String(value ?? "").replace(/[^0-9]/g, ""), 10);
  return Number.isFinite(number) ? number : null;
}
```

The adapter isolates HAR.com from the rest of our system.

The rest of the pipeline does not need to know that HAR.com calls the identifier `mls_number`.

It only sees:

```
id
value
label
metadata
```

**Choosing a stable ID:**

Our `id` comes from the MLS number rather than the address. That addresses can change formatting between crawls. An address might appear slightly differently while still referring to the same listing.

The MLS number gives us a much more stable identity for comparison.

If our identity key changes between runs, the comparison engine could treat one listing as two different records: one removed and another created.

**Keeping website-specific fields in metadata:**

Everything specific to HAR.com goes inside `metadata`. Bedrooms, bathrooms, square footage, property type, and listing status do not become part of the core pipeline contract.

That means `normalize.js`, `validate.js`, and `compare.js` remain completely independent of HAR.com.

This is the abstraction we designed earlier finally working with real data.

**Preserving metadata during normalization**

There is one small generic change we need in the pipeline. Our current `normalize.js` carries the core fields forward, but we also want to preserve the optional metadata bag.

```
src/pipeline/normalize.js
```

and add the `metadata` field to `normalizeRecord`:

```
export function normalizeRecord(record, index) {
  return {
    id: record.id ?? `entity-${index + 1}`,
    value: toNumber(record.value),
    label: typeof record.label === "string" ? record.label.replace(/\s+/g, " ").trim() : "Untitled Entry",
    metadata: record.metadata && typeof record.metadata === "object" ? record.metadata : {},
    captured_at: new Date().toISOString()
  };
}
```

That’s the only change to the pipeline core in this chapter.

The change is generic. Any future adapter can attach metadata without teaching `normalize.js` anything about the source.

**Filtering by bedrooms:**

The bedroom filter belongs to the application, not the pipeline. So it runs **after** normalization, validation, and deduplication.

Add this helper to `src/pipeline/adapters.js`:

```
/**
 * Application-level filter. Runs strictly after the pipeline core has
 * already normalized/validated/deduplicated the records.
 */
export function filterByBedrooms(records, bedrooms) {
  if (!bedrooms || bedrooms === "any") return records;
  const target = Number(bedrooms);
  return records.filter((record) => record.metadata.bedrooms === target);
}
```

Now the pipeline answers:

*Is this data valid?*

The application answers:

*Which valid records do I want to see?*

Those are different responsibilities.

**City-specific snapshots**

There is one more problem to solve before we build the CLI. Our snapshot system currently stores one file:

```
latest_snapshot.json
```

That works for one dataset. It does not work for two cities. If we run Dallas after Houston and both cities use the same snapshot, the Dallas listings would be compared against Houston’s previous data.

Every Dallas listing could appear to be `CREATED`, while Houston listings could appear to be `REMOVED`.

We need one snapshot per city.

```
src/pipeline/snapshot.js
```

and replace it with:

``` python
// src/pipeline/snapshot.js
import fs from "node:fs/promises";
import path from "node:path";
const snapshotDir = path.join(process.cwd(), "src", "storage", "snapshots");
function snapshotFileFor(city) {
  return path.join(snapshotDir, `snapshot_${city}.json`);
}
export async function saveSnapshot(city, records) {
  try {
    await fs.mkdir(snapshotDir, { recursive: true });
    await fs.writeFile(snapshotFileFor(city), JSON.stringify(records, null, 2), "utf8");
    return true;
  } catch (err) {
    console.error(`[Snapshot Write Failure] ${city}: ${err.message}`);
    return false;
  }
}
export async function loadSnapshot(city) {
  try {
    const rawBuffer = await fs.readFile(snapshotFileFor(city), "utf8");
    return JSON.parse(rawBuffer);
  } catch (notFoundError) {
    // First run for this city — start from an empty baseline.
    return [];
  }
}
```

The comparison engine still does not know anything about cities. Only the snapshot layer and the caller know which city is being processed.

Now Houston gets:

```
snapshot_houston.json
```

and Dallas gets:

```
snapshot_dallas.json
```

Each city now has its own memory.

```
src/cli/trackTexas.js
```

This is our terminal client. Notice how little logic it owns. It asks the user for a city and bedroom filter, then connects the pieces we already built:

```
CLI
 ↓
Collector
 ↓
Adapter
 ↓
Pipeline
 ↓
Filter
 ↓
Snapshot
 ↓
Compare
 ↓
Classify
```

The person running the tracker never needs to enter a HAR.com URL or know the collector ID. They simply choose a city and a bedroom filter:

``` python
// src/cli/trackTexas.js
import readline from "node:readline/promises";
import dotenv from "dotenv";
import { CollectorFactory } from "../collectors/CollectorFactory.js";
import { cityRegistry, DatasetAdapter, filterByBedrooms } from "../pipeline/adapters.js";
import { processRawIngestion } from "../pipeline/index.js";
import { loadSnapshot, saveSnapshot } from "../pipeline/snapshot.js";
import { computeHistoricalDelta } from "../pipeline/compare.js";
import { classifySystemEvents } from "../pipeline/classify.js";
dotenv.config();
const rl = readline.createInterface({ input: process.stdin, output: process.stdout });
function formatMoney(amount) {
  return `$${Number(amount).toLocaleString()}`;
}
function movementIcon(percentChange) {
  if (percentChange > 0) return "📈";
  if (percentChange < 0) return "📉";
  return "➖";
}
async function main() {
  console.log("======================================================================");
  console.log("                     TEXAS REAL ESTATE PRICE COMPARISON  TRACKER               ");
  console.log("======================================================================");
  console.log("1) Houston");
  console.log("2) Dallas");
  const cityChoice = await rl.question("Select a city (1-2): ");
  const city = cityRegistry[cityChoice.trim()];
  if (!city) {
    console.error("Invalid selection.");
    rl.close();
    process.exit(1);
  }
  console.log("\nBedroom filter:");
  console.log("1) 2 bedrooms");
  console.log("2) 3 bedrooms");
  console.log("3) Any");
  const bedroomChoice = await rl.question("Select a filter (1-3): ");
  const bedrooms = { 1: 2, 2: 3, 3: "any" }[bedroomChoice.trim()] ?? "any";
  console.log(`\n[Tracker] City: ${city.name}`);
  console.log(`[Tracker] Bedroom filter: ${bedrooms}`);
  try {
    const baseline = await loadSnapshot(city.key);
    console.log(`[Tracker] Loaded ${baseline.length} record(s) from ${city.key}'s previous snapshot.`);
    const provider = process.env.COLLECTOR_PROVIDER || "SCRAPER_STUDIO";
    const collector = CollectorFactory.create(provider);
    const rawData = await collector.collect(city.url);
    console.log(`[Tracker] Raw records received: ${rawData.length}`);
    const standardized = DatasetAdapter.transform(rawData, city.key);
    const processed = processRawIngestion(standardized);
    console.log("[Tracker] Pipeline telemetry:", processed.telemetry);
    const filtered = filterByBedrooms(processed.data, bedrooms);
    console.log(`[Tracker] Records after bedroom filter: ${filtered.length}`);
    if (filtered.length === 0) {
      console.warn("[Tracker] Nothing left after filtering. Nothing to compare or save.");
      return;
    }
    const deltas = computeHistoricalDelta(baseline, filtered);
    const events = classifySystemEvents(deltas);
    console.log("\n----------------------------------------------------------------------");
    console.log(`   ${city.name.toUpperCase()} — FIRST ${Math.min(10, filtered.length)} RECORDS`);
    console.log("----------------------------------------------------------------------");
    for (const record of filtered.slice(0, 10)) {
      const previous = baseline.find((item) => item.id === record.id);
      console.log(`  [${record.id}] ${record.label}`);
      console.log(`    Price: ${formatMoney(record.value)}  |  Beds: ${record.metadata.bedrooms ?? "?"}  |  Status: ${record.metadata.listing_status ?? "unknown"}`);
      if (previous) {
        const change = record.value - previous.value;
        const percent = previous.value ? (change / previous.value) * 100 : 0;
        console.log(`    Previous: ${formatMoney(previous.value)}  ${movementIcon(percent)} change>=0?"+":""{formatMoney(change)} (${percent.toFixed(2)}%)`);
      } else {
        console.log("    Previous: [no baseline yet]");
      }
    }
    console.log("\n----------------------------------------------------------------------");
    console.log(`   CREATED: ${deltas.created.length}  |  UPDATED: ${deltas.updated.length}  |  REMOVED: ${deltas.removed.length}`);
    console.log("----------------------------------------------------------------------");
    for (const event of events) {
      console.log(`  [${event.severity}] ${event.type}: ${event.message}`);
    }
    const merged = new Map(baseline.map((item) => [item.id, item]));
    for (const record of filtered) {
      merged.set(record.id, record);
    }
    await saveSnapshot(city.key, [...merged.values()]);
    console.log(`\n[Tracker] Snapshot saved: snapshot_${city.key}.json`);
  } catch (error) {
    console.error(`\n[Tracker Error] ${error.message}`);
  } finally {
    rl.close();
  }
}
main();
```

Add the CLI command to `package.json`:

```
"scripts": {
  "track": "node src/cli/trackTexas.js"
}
```

Now run:

```
npm run track
```

The terminal itself is not the interesting part. It is only a thin client that connects the components we already built.

The same sequence could later run behind:

The underlying pipeline would not need to change.

Now we can test the complete system with real data.

**Test 1: Real Data Baseline**

Run:

```
npm run track
```

From the terminal user query box: Select 1) Houston

And then in the next query for bedroom filter, I have chosen 3 (Any) to watch the real flow happen end to end.

For a normal repeatable test, keep the scraper bounded to the two-page configuration from Section 11.1. If you want a larger real-world dataset like m and are willing to wait longer, you can increase the crawl size for that specific run (e.g., `max_page = 10`).

The first run has no previous snapshot. That means every valid listing should show:

```
Previous: [no baseline yet]
```

The validator may also quarantine records that do not contain the required fields, such as a usable price or MLS number.

That is exactly what we want. The same validation rules that worked against our mock data are now working against real listings.

After the run, the tracker writes:

```
snapshot_houston.json
```

Scraper Studio will also show the corresponding collection in its **Runs** tab:

**Test 2: Real Historical Price Comparison**

Now we can test whether the comparison engine detects a price change.

Open `src/storage/snapshots/snapshot_houston.json` and locate a live listing to use as our test case. For instance, I have chosen **7942 Hammerly Blvd, Houston, TX 77055,** which was recently stored by our pipeline’s latest scraper run:

`$439900`
`$539900`
Because real estate prices change slowly over weeks or months, we can’t just wait around for a real-world market drop to test our code. Instead, we will manually edit our local snapshot data to replicate a live price swing:

By artificially inflating our historical data, we force a mismatch. When the scraper runs next, it will perceive the actual market price as a sudden drop, allowing us to instantly verify that our comparison engine catches real-world fluctuations correctly.

Run the tracker again against Houston:

```
npm run track
```

The live value is lower than our edited historical value, so the comparison engine detects a price drop.

In my test, the result was `-18.52% price drop`:

That is below the `30%` critical anomaly threshold while still exceeding the `15%` major threshold discussed in earlier chapter, so it is classified according to the rules we already defined there.

See? we did not create a second comparison system for real estate. The exact same `compare.js` and `classify.js` code is now analyzing a real listing.

**Test 3: New and Removed Listings**

Edit `snapshot_houston.json` again.

This time:

`"81"`.
Now Run: `npm run track`

The listing you deleted from the snapshot shows up as `CREATED` (it’s new relative to your edited baseline), and the fake listing you added shows up as `DELETED` (it’s absent from the live crawl). This is the exact `compare.js` logic from Chapter 8, it has no idea it’s looking at real estate, it’s still just matching `id` keys through two `Map` objects.

**Test 4: Bad Data Protection**

Open the snapshot again. Give one listing an obviously misleading value, such as:

```
"value": 1
```

This mimics a scraper accidentally extracting a stray number instead of the actual property price. Run the tracker again.

You should see an anomaly similar to:

```
[CRITICAL] ANOMALOUS_VALUE_COLLAPSE: 5238 Kylie Springs Ln, Houston, TX 77066 (ID: 53228136) shifted by -99.99%.
```

The value is technically a valid number, so it can pass basic validation. But the comparison engine sees an extreme shift and the classifier flags it as *critical*. This is the reason why validation and anomaly detection are separate steps.

A number can be **valid** without being **trustworthy**.

**Test 5: Multiple Cities**

Run the tracker again and select: *2) Dallas*

Complete the sync. Now check:

```
src/storage/snapshots/
```

You should have two separate files:

```
snapshot_houston.json
snapshot_dallas.json
```

Open both files. Houston and Dallas should remain separate. Running Dallas should not cause Houston listings to appear as `REMOVED`. Running Houston should not cause Dallas listings to appear as `CREATED`.

That is the small change we made to `snapshot.js` doing its job.

Each city now has its own memory. The terminal was the visible part of these five tests, but it was never the important part. The important part was everything underneath it: **the contract, the pipeline, the memory, the comparison engine, and the event classifier.** We built those pieces before we had any real estate data.

This chapter proved that the architecture survives contact with a real website. Let’s save this checkpoint:

```
git add . && git commit -m "feat: Texas real estate valuation delta tracker, bounded pagination, city-scoped snapshots memory"
```

Next, we will look at what happens when HAR.com changes its layout underneath us.

More importantly, we will see how Scraper Studio’s **Self-Healing** can repair a broken extraction without us touching `normalize.js`, `compare.js`, or the rest of our pipeline core.

If you ran the Texas real-estate tracker in the previous chapter and all records looked healthy, great. The scraper is working against the live HAR.com site. But websites change.

A developer might rename a CSS class or change the page structure. When that happens, a traditional scraper can fail silently. Instead of throwing an error, it may return incomplete values.

We cannot wait for HAR.com to change its website just to test this. So we will break our scraper ourselves.

In this chapter, we will:

The Bright Data CLI is available as the `[@brightdata/cli](https://docs.brightdata.com/products/cli/installation)` npm package.

You can install it globally:

```
npm install -g @brightdata/cli
```

You can also run it without a permanent installation using `npx.` Plus its fastest way to run the Bright Data CLI, which runs the latest version with no global install:

```
npx -p @brightdata/cli brightdata --version
```

For this chapter, I will use the `bdata` command after installing the CLI globally. Check that it works:

```
bdata --version
```

**Logging In:** Authenticate the CLI with:

```
bdata login
```

This opens a browser for authentication. After login, the CLI stores the credentials locally and checks for the required Bright Data zones. If the `cli_unlocker` and `cli_browser` zones do not exist, the CLI can create them automatically.

You do not need to copy an API key into your project just to use the CLI.

Once the login is complete, the terminal is ready to manage your Bright Data scrapers.

Now let’s imitate a website redesign. A live website might change a selector from:

```
h2.font_size--large_extra_extra.color_carbon
```

to a completely different one. Our scraper could then keep running while returning missing values.

*Step 1: Open the Web IDE*

Open your Bright Data dashboard and go to our previous har.com scraper dashboard. In the left sidebar, open **Parser Code**. Find the line where the scraper extracts the price, address, or bedrooms. Let’s corrupt the `price` selector:

*Step 2: Break the Selector*

Change the working selector:

``` js
const price = extractPrice(
  'h2.font_size--large_extra_extra.color_carbon'
);
```

to a selector that does not exist:

``` js
const price = extractPrice(
  'div.completely-broken-price-class-that-does-not-exist'
);
```

*Step 3: Publish the Broken Version and Test*

Click the dropdown next to **Save** and choose **Save to production**. The active scraper now contains our broken selector. Delete all our local snapshot files that we generated earlier.

Run the tracker again: `npm run track`

Because the website elements no longer match our corrupted selector, the scraper returns `null` for bedrooms. Watch your terminal:

This is where our pipeline protects us. Our `validate.js` validation layer sees that required values are missing and prevents those records from entering the local data storage (`failed_compliance: 10`) and quarantined the dirty data safely.

The pipeline successfully protected our database from corruption.

Now we need to repair the extraction layer.

You have two choices to heal a broken scraper: you can use the web console “Self-Healing”, or you can use their terminal CLI.

While the web console works well for edits, the CLI is the superior tool for AI coding agents (Claude Code, Cursor, or Codex). I will intentionally use CLI to help you better understand both IDE (UI) and terminal self-healing workflow.

Run the healing command with your Collector ID and a clear description of the problem:

```
bdata scraper heal c_mssne7uc11iej44z4s \
  "The price selector is broken and returns null. Re-capture the price from the property details section." \
  --url https://www.har.com
```

The CLI sends the healing request to Bright Data’s scraper infrastructure. The AI agent analyzes the scraper and the target page, then prepares a new version of the scraper.

**Review the Proposed Fix:**

The healing process stops at an approval gate. You should see a status similar to:

```
Status: awaiting_approval
```

Refresh your Scraper Studio web IDE and you’ll see the this popup window:

Check that the price is being extracted from the correct part of the page. Do not approve a fix just because the command succeeded; make sure it returns the **right** value.

**Approve or Reject:**

If the proposed change looks correct, approve it using the CLI workflow.

```
bdata scraper approve c_mssne7uc11iej44z4s
```

If you decide that the proposed fix is wrong, reject it instead:

```
bdata scraper approve c_mssne7uc11iej44z4s --reject
```

Rejecting the fix leaves the existing scraper unchanged. You can then run `heal` again with clearer instructions.

One useful detail is that the **Collector ID stays the same** after a successful healing workflow. Your Node.js application therefore does not need a new collector ID just because the scraper implementation changed.

**Verifying the Fix:**

The scraper is repaired. Now run the tracker again: `npm run track`

The same Node.js pipeline receives data from the same collector. But this time, the required fields are populated again:

Our `ScraperStudioCollector.js` calls the same endpoints and Collector ID as before. The scraper now outputs clean, populated data, the *compliance failure* count ***drops back to zero***, and the new real estate listings flow into your local snapshots seamlessly:

That is the main benefit of the architecture we built throughout this guide. The scraper can be repaired at the infrastructure layer while the Node.js pipeline continues working with the same stable data contract.

Our pipeline is now a modular Node.js application. Moving it to a cloud server is the next step. The deployment steps depend on your hosting provider, but the process is usually:

Once deployed, your Express application can expose the same API routes we tested locally. An AI agent or app can then call the API without knowing implementation details.

Ultimately, Scraper Studio handles the website. Node.js pipeline handles the data. Snapshots store historical state. Delta comparison detects changes. The API makes the result reusable.

That is the final framework we set out to build.

To be honest, it is incredibly tempting to keep writing to set up managed databases, implement RAG workflows, configure cloud queues, or build a frontend. But doing so would violate the very principles of clean software design we set out to learn. Our **two-layer** pipline **is already complete.**

We don’t need to build all of that to prove the design works.

Instead, let’s look at ***three production pitfalls you should think*** about when moving this example toward production.

Some hosting environments use temporary filesystems. A restart or new deployment can remove files created by the running application.

If the snapshot disappears, the pipeline loses its historical baseline.

For a simple single-instance deployment, your hosting provider may offer persistent disk or volume storage. You can then point the snapshot directory to that location.

``` js
const snapshotDir =
  process.env.SNAPSHOT_DIR ||
  path.join(process.cwd(), "src", "storage", "snapshots");
```

For a larger application with multiple instances or concurrent writers, a database becomes a better choice. You could replace the file-based implementation behind:

```
loadSnapshot()
saveSnapshot()
```

with PostgreSQL, Redis, or another persistent store.

The rest of the pipeline does not need to know where the snapshot is stored. That is one of the benefits of keeping storage behind a small interface.

Our snapshot keeps the latest version of each listing. But a listing that disappears from the source can remain in the snapshot if we never remove it.

Over many runs, that can leave old records in the dataset. One simple solution is *time-based pruning.*

Our normalized records already contain `captured_at`, so we can use that timestamp to remove records that have not been seen recently.

``` js
export function pruneStaleRecords(records, maxAgeDays = 30) {
  const cutoff =
    Date.now() -
    maxAgeDays * 24 * 60 * 60 * 1000;
return records.filter((record) => {
    const seenAt = new Date(record.captured_at).getTime();
    return Number.isFinite(seenAt) && seenAt >= cutoff;
  });
}
```

The `30`-day value is only an example. Your application might use 7 days, 30 days, 90 days, or another rule depending on the data.

There is also an important rule here: *Prune the complete validated dataset, not a filtered view.*

For instance, if the user runs the tracker with a two-bedroom filter, a three-bedroom listing may not appear in that run. Historical storage should therefore operate on the full dataset before application-level filters are applied.

A large crawl may take longer than the HTTP timeout allowed by your hosting platform. The exact timeout depends on the platform and the type of service you are using.

That means a request such as:

```
POST /api/tracker/houston/sync
```

should not always be responsible for waiting until a large collection finishes. A better production design is to separate **starting a job** from **checking its result**.

The application can:

This is commonly called **asynchronous job processing** or **polling**.

It becomes especially useful when your scraper needs to process multiple pages. For the small project in this guide, a synchronous flow is easier to understand. For larger crawls, asynchronous processing is the natural next step.

We kept this pipeline small on purpose. Because it is reusable and reliable, you can easily plug it into large-scale production workflows.

Here are two high-demand popular use cases you can build right now:

Static documents make AI real estate analysts obsolete fast. You can use this pipeline to feed an LLM real-time data instead.

`/api/track/houston` endpoint.
Running web crawlers inside autonomous AI loops is financially risky. A single runaway loop on a dynamic billing platform can cost thousands in surprise server fees.

We started with a basic problem: A scraper can succeed and still give you bad data. Therefore, we built a data pipeline around it. The scraper extracts data. The pipeline makes that data useful.

We first created a stable contract. The pipeline core works with that contract instead of knowing the fields of a specific website.

Then we built the core processing stages: we added snapshots so the application could remember previous data. Then we added comparison and classification. Next, we created the collector factory so the application could work with different extraction providers.

Finally, we tested everything against live Texas real-estate listings from HAR.com. We did not put HAR.com-specific field names into the pipeline core. Instead, the adapter translated HAR.com’s response into our stable contract.

For testing our pipeline, we intentionally broke the live scraper. Our validation layer detected the bad data. Bright Data’s self-healing workflow repaired the scraper. We reviewed the proposed change and deployed it without changing the Node.js pipeline.

That is the main idea of this guide. The terminal tracker itself is not the most notable part. It is ONLY one client of the application. The reusable part is the architecture underneath.

Because doing that for every source creates a maintenance problem.

Without a stable contract, every new source introduces different field names and different business logic. With the contract, each source needs an adapter.

The rest of the pipeline can remain the same.

There is no *special* requirement to use Node.js. We use it here because Scraper Studio’s interaction and parser code also use JavaScript, so the entire example stays in one language.

The core ideas are not tied to Node.js. The normalization, validation, comparison, and classification logic could be implemented in Python, Go, or another language.

For an initial MVP ([Minimum Viable Product](https://www.atlassian.com/agile/product-management/minimum-viable-product)), using a JSON is the smartest, fastest choice. But as soon as the project goes live for hundreds of users or needs long-term history, you must transition that historical snapshot into a real database.

The examples and workflows in this guide are about publicly accessible web data. [Bright Data does not utilize private accounts](https://brightdata.com/trustcenter), bypass authentication mechanisms, or collect sensitive personal information (PII).

It is important to note that public visibility does not automatically grant a blanket right to collect or reuse data. The exact legal requirements heavily depend on the target website, the nature of the data, your specific jurisdiction, and your intended end-use.

Before deploying any scraper to production systems, developers must thoroughly review the target site’s terms of service, applicable local and international laws, [strict data-protection regulations (GDPR or CCPA)](https://www.cookieyes.com/blog/ccpa-vs-gdpr/), and their own internal organizational policies.

Yes. That is one of the main reasons we created the `collector factory` and `adapter layer`.

For a new source, you need:

`{ id, value, label, metadata }`.
The generic pipeline does not need to know whether the source contains houses, products, jobs, or another type of data.

The pipeline does not need to know who started it.

Instead of calling it from the interactive CLI, you can trigger the same sequence from a scheduled job:

```
CollectorFactory.create()
        ↓
DatasetAdapter.transform()
        ↓
processRawIngestion()
        ↓
computeHistoricalDelta()
        ↓
classifySystemEvents()
        ↓
saveSnapshot()
```

A cron job, scheduled cloud function, webhook, or another job scheduler can become the trigger. The pipeline itself does not need to change.

**Related article:** [How to Build an Unlocked AI Agent for Browser Automation with Node.js, Bright Data, Gemini, and Playwright](https://dev.to/codewithshahan/how-to-build-a-production-ready-browser-automation-ai-agent-with-nodejs-bright-data-gemini-and-39ii)