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.
That’s the problem this guide solves.
🏗️ What we’re building
- A Texas real-estate tracker that collects live HAR.com listings, filters them by city and bedrooms, and detects price changes over time.
- A reusable pipeline that separates web extraction from our application logic.
- A self-healing workflow that lets us repair the scraper without rewriting the Node.js pipeline when the target site’s structure changes.
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.
🔗 The complete source code is available in my GitHub repository. You can also download The Web Data Pipeline handbook for free:
1. The Data Extraction Pipeline Structure
Our data extraction pipeline divides responsibilities into two layers:
- The Infrastructure Layer. It handles proxies, browser fingerprints, CAPTCHAs, and heavy JavaScript rendering to turn unpredictable HTML into stateless structured data.
- The Pipeline Layer (Node.js App). It owns the application logic, validation, data state, and historical comparisons.
For this guide specifically, the final outline looks like this:
Therefore, by following this Separation of Concerns (SoC) principle, the same pipeline can be reused for e-commerce, crypto rates, travel listings, job boards, and competitor/financial data monitoring; the list goes on.
2. The Two Non-negotiable Data Pipeline Rules
In a scalable data pipeline, neither layer should depend on the other’s internal implementation details.
We achieve that with two rules.
RULE 1: The stable data contract is the boundary**
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) and the Ports and Adapters (Hexagonal) 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.
RULE 2: The pipeline must be payload-agnostic**
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.
3. Why Traditional Web Scrapers Break
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 frequently issue “soft bans", returning fake HTTP 200 OKresponses 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.
3.1. The extraction problem
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 or the Scraping Browser API, 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.
3.2. Turning a web page into structured data
To see this transformation in practice, consider tracking the US real estate market using a Zillow Seattle 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:
const price = await page
.locator('[data-test="property-card-price"]')
.textContent();
Or perhaps the developer finds a class in the current HTML:
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.
3.3. Scaling beyond a single page
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:
- JSON
- CSV
- database records
- Data API responses
- Webhook events
- object storage
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:
3.4. From extracted data to application logic
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:
- Which records are new?
- Which records changed?
- What changed between two runs?
- Can we reuse the same pipeline for another target or application?
- How do we identify meaningful changes in historical data?
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.
3.5. The maintenance loop
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.
4. The Rise of Self-Healing AI-Native Scrapers
To break the “maintenance loop", let’s compare the three most common approaches developers use today:
- DIY (Playwright / Scrapy). Take Weeks. You write the crawler, wire up proxies, run headless browsers, handle queues, retries, and anti-bot.
- Apify. Take days to weeks. You build your own actor, meaning writing the crawler, integrating a proxy, handling retries, and packaging it as an actor. Using a third-party actor from the Apify Store is faster, but you do not own the code.
- Scraper Studio. Take minutes per site. Describe the data you need in plain English, and its AI Agent can generate a scraper and output schema. You can then test and modify the generated JavaScript in the built-in IDE, or work with Bright Data through its CLI and coding-agent (Claude/Cursor) integrations by installing Bright Data skills via
npx skills add brightdata/skills.
The important difference is how much of the extraction infrastructure you have to build and maintain yourself.
4.1. The Self-Healing Advantage
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, 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 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.
4.2. Why Scraper Studio Fits Our Architecture
At the beginning of this guide, we established two rules:
- The application trusts a stable data contract.
- The pipeline remains payload-agnostic.
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.
4.3. What About Cost, Scale, and Concurrency
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 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.
5. Project Setup
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.
6.1 Configure 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.
6.2 Add Environment Variables
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.
5.3 Create the Node.js Server
Now create the following file: src/server.js
Add the initial server:
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.
6. Building 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.
6.1 Design the Prototype 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
6.2 Normalize the Data
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.
6.3 Validate the Contract
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.
Open:
src/pipeline/validate.js
Add:
/**
* 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.
6.4 Remove Duplicate Records
Next, we need to prevent the same entity from appearing multiple times in a collection.
Open: src/pipeline/deduplicate.js
Add:
/**
* 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.
6.5 Compose the Pipeline
Now combine the individual processing stages into one entry point.
Open: src/pipeline/index.js
Add:
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.
6.6 Test Multiple Data Sources
Now let’s prove that idea. Create three small mock datasets in: src/collectors/mock.js
Each source uses completely different field names:
// 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.
6.7 Connect the Pipeline to Express
Now expose the pipeline through a small test endpoint.
Open: src/api/server.js
Replace the initial server with:
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}`));
6.8 Test the Prototype
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.
6.9 Commit the Prototype
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.
7. Teaching Our Pipeline to Remember
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.
7.1. Why a Scraper Run Is Stateless
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
- Created: A new listing appeared.
- Updated: An existing listing changed, such as a price drop.
- Removed: A previously tracked listing is no longer present.
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.
7.2. Create Local Snapshot Storage
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.
Add:
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.
7.3. Test Snapshot Persistence
Before using snapshots for change detection, let’s verify that our storage layer can actually write and read data.
Open:
src/api/server.js
For this isolated storage test, use:
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:
$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.
8. Detecting Historical Changes
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:
- House A: The price changed from $500,000 to $450,000.
- House B: The value stayed unchanged.
- House C: A new listing appeared.
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.
8.1 Building the Dataset Comparison Engine
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.
8.2 Testing the Comparison Engine
Let’s connect the comparison engine to a temporary Express endpoint and test it against two simulated crawl runs.
Open: src/api/server.js
WRite:
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:
$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:
$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
Success Result
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.
8.3 Classifying System Events
For this prototype, we’ll classify value changes using configurable thresholds.
These numbers are our example thresholds, not universal rules:
- Under 15% —
**MINOR**: Small change that may not require immediate attention. - 15% to under 30% —
**MAJOR**: Significant change worth monitoring. - 30% or more downward —
**CRITICAL**: Large negative change that should be investigated.
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
Add:
/**
* 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.
8.4 Connect Event Classification
Now integrate the classifier into our Express endpoint.
Open: src/api/server.js
Write:
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}`)
);
8.5 Test Event Classification
Let’s deliberately create a large negative change and see whether our classifier catches it.
Open:
src/storage/snapshots/latest_snapshot.json
Replace its contents with:
[ {
"id": "A",
"value": 5000000,
"label": "House A"
}
]
Now create the test payload in PowerShell:
$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.
9. Building a Reusable Collector Factory
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:
- Normalize inconsistent values
- Validate incoming records
- Remove duplicates
- Store historical snapshots
- Compare datasets over time
- Classify notable market changes
The next step is to make the extraction layer replaceable.
If we change scraping providers later, the processing pipeline should not need to change.
9.1 Designing the Collector Abstraction
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 pattern we introduced earlier: the application depends on an interface, while individual providers implement that interface.
9.2 Building the Base Collector
Create:
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).
9.3 Defining the Primary Provider
Our primary provider will be Bright Data Scraper Studio.
Create:
src/collectors/ScraperStudioCollector.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.
9.4 Adding a Secondary Provider
Now let’s add a second collector to prove that the abstraction is not tied to one provider.
Create:
src/collectors/FirecrawlCollector.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.
9.5 Implementing the Collector Factory
Now we need a single place that decides which provider to instantiate.
Create:
src/collectors/CollectorFactory.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:
const collector = CollectorFactory.create("SCRAPER_STUDIO");
Later, changing the configuration to:
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.
9.6 Verifying the Architecture
Before connecting real external services, let’s verify that the abstraction works at runtime.
Create:
src/testCollector.js
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.
10. Generate the First Scraper with Scraper Studio AI Agent
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:
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.
Step 1: Initialize the Target URL
Sign in to your Bright Data dashboard 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
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.
Step 2: Configure the Extraction Workflow
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:
- Extract data from individual property pages
- Extract data only from the listing page
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.
Step 3: Understanding the Scraper Studio Generated Workflow
After approval, Scraper Studio opens the scraper inside its workspace editor.
The generated JavaScript is divided into two important parts:
- Interaction code—controls navigation and crawling behavior
- Parser code—extracts structured information from the current page
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:
FunctionResponsibilitynavigate()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.
11. Connecting Scraper Studio API to Node.js
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
11.1 Initiate by API
Open your published scraper in the Scraper Studio workspace and open the Initiate by API tab:
You need two values:
- API Token: Your Bright Data API token from Account Settings.
- Collector ID: The ID of your published Scraper Studio collector. You get this from your Scraper Studio Dashboard after creating the scraper (it will be an ID starting with
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.
11.2 Triggering Scraper Studio API from Node.js
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:
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.
11.3 Mounting the Express Route
Next, connect the scraper trigger to our Express server.
Open:
src/api/server.js
Replace its contents with:
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.
11.4 Verification Pass
Start the development server:
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.
11.5 Open the Runs Tab
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.
12. Building a Texas Real Estate Delta Tracker CLI
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.
12.1. How to Handle Pagination when Scraping
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:
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:
- Unset (Default): 10 pages = 200 results
- Set to 50: 50 pages = 1,000 results
- Set to 84: 84 pages = 1,680 results
Here is how much time it takes for 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 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:
- Limit listing pagination to 2 pages so it runs in seconds
- Limit property-detail crawling to 5 listings per page
- Prevent property-detail pages from starting another pagination loop
Therefore, rewrite the “interaction code” in the Scraper Studio IDE as follows:
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:
12.2. Delivery Preferences and Calling the Collection API
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:
- On a job completion (batch)
- JSON
- API download
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.
12.3. Waiting for the Job Without Timing Out
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:
- HTTP request timeout: How long one network request is allowed to hang.
- Collection timeout: How long the entire scraping job is allowed to run.
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:
// 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 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.
12.4. Building the Texas Tracker’s Data Layer
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.
Open:
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.
Open:
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.
Open:
src/pipeline/snapshot.js
and replace it with:
// 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.
12.5. Assembling an Interactive Tracker
Create:
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:
// 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:
- An Express API
- A scheduled job
- A background worker
- An AI agent tool call
The underlying pipeline would not need to change.
12.6. Final Testing Workflow
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:
- Live price right now:
$439900 - Local snapshot (manually edited for testing):
$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:
- Delete a few real listings from the snapshot.
- Add a fake listing with an ID that will not appear in the live crawl, such as
"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.
13. Scraper Studio Self-Healing in Production
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:
- Set up the Bright Data CLI.
- Intentionally break a scraper selector in Scraper Studio.
- Run our Node.js tracker and let validation catch the bad data.
- Use the Bright Data CLI to heal the scraper.
- Review the proposed fix before deploying it.
- Run the tracker again locally and verify that the data is healthy.
13.1. Installing the Bright Data CLI
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.
13.2. Deliberately Breaking the Scraper
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:
const price = extractPrice(
'h2.font_size--large_extra_extra.color_carbon'
);
to a selector that does not exist:
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.
13.3. Ask Bright Data to Heal the Scraper
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.
14. Deployment
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:
- Push your project to a GitHub repository.
- Connect the repository to a Node.js hosting platform such as Render, Railway, or AWS.
- Add your environment variables through the platform’s environment-variable settings.
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.
15. Scaling and Production
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.
1: The Temporary Cloud Filesystem
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.
For example:
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.
2: Time-Based Data Decay
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.
For example:
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.
3: Long-Running HTTP Requests
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:
- Start the collection job.
- Receive a job or snapshot identifier.
- Return control to the caller.
- Check the job status later.
- Process the final dataset when the collection is complete.
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.
16. What You Can Build Next
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:
1. A Real-Estate RAG Pipeline for AI Agents
Static documents make AI real estate analysts obsolete fast. You can use this pipeline to feed an LLM real-time data instead.
- How to build it: Set up a daily cron job (like Render Cron) to hit your
/api/track/houstonendpoint. - The Flow: The pipeline pulls live listings, normalizes the data, and flags major changes (like a 15%+ price drop). The server converts these updates into vector embeddings and saves them to Pinecone or Chroma.
- The Result: Your RAG system gets verified, daily market shifts. Your AI analyst can spot live investment opportunities without any manual data entry.
2. Capped-Budget Market Analyzers
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.
- How to build it: Run this pipeline on a flat-rate model (like Scraper Studio’s $1.50 per 1,000 pages). Cap each run at 2 directory pages.
- The Math: Every daily sync costs around $0.003.
- The Advantage: Strict pagination boundaries give you a hard safety ceiling. You can calculate and lock down your maximum monthly data budget with absolute certainty.
Conclusion
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.
FAQ
Why build a custom pipeline instead of parsing whatever the scraper returns?
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.
Why Node.js instead of Python?
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.
Why use JSON files instead of a database?
For an initial MVP (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.
Is web scraping legal?
The examples and workflows in this guide are about publicly accessible web data. Bright Data does not utilize private accounts, 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), and their own internal organizational policies.
Can I use this pipeline other than a single data source?
Yes. That is one of the main reasons we created the collector factory and adapter layer.
For a new source, you need:
- A scraper or collector for the source.
- An adapter that maps its fields to
{ id, value, label, metadata }. - A registration in the collector factory.
The generic pipeline does not need to know whether the source contains houses, products, jobs, or another type of data.
How do I run the pipeline on a schedule?
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
Top comments (0)