The workflow in the previous lesson answered one question, once, while a user waited. Enrichment is the other half of most multi-API systems: a table of records (leads, accounts, suppliers, domains) that you annotate with facts from several APIs, keep up to date, and query later. The CRM shows the company's tech stack. The risk team filters accounts by domain age. The support tool shows a customer's local time from their phone number.
The difference in time scale changes the design. A request-path workflow lives for two seconds and can afford to start over. An enrichment pipeline lives for years, sees millions of records, survives deploys and outages mid-run, and is asked six months later "where did this value come from, and is it still true?" This lesson is about building one that can answer.
From loop to queue
The first version of every enrichment job is a loop: for each record, call the APIs, write the results. It works for a thousand records and fails at a hundred thousand, for reasons that have nothing to do with the APIs:
- A crash at record 61,000 means either starting over (paying again for 60,999 records) or hand-editing a resume point.
- One slow API holds every record behind it, even records that only need the fast ones.
- A rate-limit pause or exhausted credits stops everything, including stages that would have succeeded.
- There is nowhere to put a record that will never succeed, so it is retried on every run, forever.
A queue fixes the structure. Each unit of work is a job for one record and one stage, where a stage is one API applied to one input derived from the record:
| Stage | API | Input from the record | Typical re-check interval |
|---|---|---|---|
phone | phonenumbervalidator | phone, country | 180 days |
whois | whoislookup | domain of website | 30 days |
stack | techstackdetector | website | 30 days |
The intervals are design choices you make from how fast the underlying facts change, not values any API prescribes. Phone number metadata barely moves. Registrations change hands. Websites change their stack on a quarter's timescale.
Workers pull jobs, run one stage, write the result, and acknowledge the job. Stages for the same record run independently, so a WHOIS outage delays only WHOIS. Failed jobs go back on the queue with a delay. Jobs that can never succeed go to a dead-letter queue where a person can look at them. You can use a managed queue, a database table with a locked_until column, or a library; the design below does not depend on which.
Make every stage idempotent
Almost every queue delivers at least once. A worker that crashes after calling the API but before acknowledging the job will see that job again. Your stages must therefore be safe to run twice: the second run must leave the same state as the first, and ideally should not pay twice.
Two mechanisms give you that.
Key each result by what was asked, not by when. The result of a stage is a pure function of the API and its normalised parameters, so use those as the storage key. A second run computes the same key and overwrites the same row instead of appending a duplicate.
Check before calling. If a fresh result for that key already exists, skip the call. This turns a re-delivered job into a cheap read, and it also deduplicates across records: two leads at the same company share one WHOIS result, one call, one charge.
import { createHash } from 'node:crypto';
export function normalise(stage, params) {
const out = {};
for (const k of Object.keys(params).sort()) {
let v = String(params[k]).trim();
if (k === 'domain') v = v.toLowerCase().replace(/^www\./, '');
if (k === 'country') v = v.toUpperCase();
out[k] = v;
}
return out;
}
export function stageKey(stage, api, params) {
const canonical = JSON.stringify([stage, api, normalise(stage, params)]);
return createHash('sha256').update(canonical).digest('hex').slice(0, 32);
}Normalise only what the API genuinely treats as equivalent. Lower-casing a domain is safe. Stripping www. is a judgement you make for your use case: it is right for WHOIS, where the registration belongs to the bare domain, and wrong for a tech-stack scan, where www might be served by a different system. That is why normalise receives the stage.
Store provenance with every fact
A value without a source is a rumour. Every row your stages write should answer five questions on its own: which API produced it, with what input, when, from which version of the endpoint, and at what cost. Store the outcome too, including failures, so that "we tried and the domain is unregistered" is distinguishable from "we never tried".
A table shape that works in any SQL database:
CREATE TABLE enrichment (
key TEXT PRIMARY KEY, -- stageKey(stage, api, params)
stage TEXT NOT NULL,
api TEXT NOT NULL,
params TEXT NOT NULL, -- normalised JSON
outcome TEXT NOT NULL, -- 'ok' | 'not_found' | 'invalid' | 'error'
data TEXT, -- the API's data object, verbatim
http_status INTEGER,
api_version TEXT, -- x-api-version response header
shape TEXT, -- schema fingerprint, see below
credits INTEGER, -- catalog cost when billed, else 0
fetched_at TEXT NOT NULL,
expires_at TEXT NOT NULL
);
CREATE TABLE record_enrichment (record_id TEXT, stage TEXT, key TEXT, PRIMARY KEY (record_id, stage));The second table links records to results. Many records can point at one result, which is how deduplication shows up in the data model. Store the API's data object verbatim, and derive the columns your product reads from it in a separate step. When you later realise you needed a field you did not extract, it is already there and costs nothing to backfill from storage.
Each APIVerve response carries x-api-version, the version of the endpoint that served it. It is one of the headers exposed to browsers as well as servers, so any client can read it. Store it with the result. If the response you are processing does not carry it, store null rather than a guess.
Here is the worker that ties those pieces together:
import { stageKey, normalise } from './stage-key.mjs';
import { shapeOf } from './shape.mjs';
const CREDITS = { phonenumbervalidator: 10, whoislookup: 8, techstackdetector: 30 };
const TTL_DAYS = { phone: 180, whois: 30, stack: 30 };
export class Retry extends Error {}
export class PauseQueue extends Error {}
export async function runStage(job, db) {
const params = normalise(job.stage, job.params);
const key = stageKey(job.stage, job.api, params);
const existing = await db.getResult(key);
if (existing && new Date(existing.expires_at) > new Date()) {
await db.link(job.recordId, job.stage, key); // re-delivery or duplicate: no call
return existing;
}
const url = new URL(`https://api.apiverve.com/v1/${job.api}`);
for (const [k, v] of Object.entries(params)) url.searchParams.set(k, v);
let res;
try {
res = await fetch(url, { headers: { 'x-api-key': process.env.APIVERVE_API_KEY }, signal: AbortSignal.timeout(15_000) });
} catch {
throw new Retry(`${job.api}: network or timeout`);
}
const body = await res.json().catch(() => null);
if (res.status === 429) {
if (res.headers.get('x-api-remaining-credits') === '0') throw new PauseQueue('credits exhausted');
throw new Retry('rate limited');
}
if (res.status === 401 || res.status === 403) throw new PauseQueue(`${job.api}: HTTP ${res.status}, check the key`);
if (res.status >= 500 || (res.ok && body?.status !== 'ok')) throw new Retry(`${job.api}: HTTP ${res.status}`);
const outcome = res.ok ? 'ok' : res.status === 404 ? 'not_found' : res.status === 400 ? 'invalid' : 'error';
if (outcome === 'error') throw new Retry(`${job.api}: HTTP ${res.status}`);
const row = {
key, stage: job.stage, api: job.api, params: JSON.stringify(params), outcome,
data: outcome === 'ok' ? JSON.stringify(body.data) : null,
http_status: res.status,
api_version: res.headers.get('x-api-version'),
shape: outcome === 'ok' ? shapeOf(body.data) : null,
credits: outcome === 'ok' ? CREDITS[job.api] : 0,
fetched_at: new Date().toISOString(),
expires_at: new Date(Date.now() + TTL_DAYS[job.stage] * 864e5).toISOString(),
};
await db.upsertResult(row); // same key, same row: idempotent
await db.link(job.recordId, job.stage, key);
return row;
}The error classification is the part most pipelines get wrong:
400and404are answers. Bad input will be bad on every retry, and "nothing found" is a fact about the input. Both are stored as outcomes with an expiry, so the record is not retried on every run, and a person can query them.- Rate limits and
5xxare transient. ThrowingRetrysends the job back with a delay; your queue's backoff does the rest. - Exhausted credits stop the queue, not the job. Every job on the key will fail the same way.
PauseQueueshould stop all workers until the cycle renews or someone tops up. The check reads the header as a string, so a missing header (null) can never be mistaken for zero. 401and403are configuration problems: a wrong or revoked key, or a key scoped away from this API. Retrying will not fix them and the data is not at fault, so they pause the queue and should alert a person, just as exhausted credits do.- Failures never overwrite good data. The
upsertResultonly runs for real outcomes. A transient error leaves last month's valid result in place until a fresh one replaces it.
A worker calls whoislookup, writes the result row, then crashes before acknowledging the job. The queue redelivers it. With the design above, what happens?
Detect schema drift
Over years, responses change. Fields are added. A field you relied on starts arriving as null because your plan changed and it is a premium field. An input class you never saw before produces a response without a field that every previous response had. None of these is an error from the API's point of view, and none of them throws in your code. They show up as quietly wrong data.
The defence is to record the shape of each response, the set of paths with their JSON types, and compare it with what you expect.
import { createHash } from 'node:crypto';
export function paths(value, prefix = '', out = new Map()) {
const type = value === null ? 'null' : Array.isArray(value) ? 'array' : typeof value;
if (prefix) out.set(prefix, type);
if (type === 'object') for (const [k, v] of Object.entries(value)) paths(v, prefix ? `${prefix}.${k}` : k, out);
if (type === 'array' && value.length) paths(value[0], `${prefix}[]`, out);
return out;
}
export function shapeOf(data) {
const sig = [...paths(data)].sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)).map(([p, t]) => `${p}:${t}`).join('|');
return createHash('sha256').update(sig).digest('hex').slice(0, 16);
}Storing the hash per row makes drift queryable: SELECT stage, shape, count(*) ... GROUP BY 1, 2 shows when a new shape appears and how many rows have it. When one does, diff its paths against the previous shape and decide which of three kinds it is:
- Additive: new paths appeared. Usually harmless, and often useful. Your extraction ignores them until you choose to read them, which is why you store
dataverbatim. - Nullable: a path that was a string is now
nullon some rows. Check whether the field is premium and whether the key's plan changed, or whether these inputs simply lack the fact. - Breaking: a path disappeared or changed type. Stop extracting that field and look before trusting the column built from it.
Types deserve attention even when nothing changes. Phone Number Validator returns numberNational as a JSON number and nationalSignificantNumber as a string. When the national number starts with a zero, as Italian landlines do, a JSON number cannot carry it. Looking up +390669881022 returned numberNational: 669881022 and nationalSignificantNumber: "0669881022". Read the string field, or formatted.e164, for anything you will store or dial.
curl "https://api.apiverve.com/v1/phonenumbervalidator?number=%2B390669881022" \
-H "x-api-key: YOUR_API_KEY"Compare numberNational with nationalSignificantNumber: the number type dropped the leading zero. formatted.e164 is the safe canonical key.
Absent and null are different shapes too. A WHOIS lookup for an unregistered domain comes back 200 with createdDate missing altogether, while a premium field you lack comes back present and null. Your extraction should treat both as "unknown" for the product, and your shape tracking should keep them distinct, because they have different causes.
If your extraction runs in Python, the same fingerprint is a few lines:
import hashlib
def paths(value, prefix="", out=None):
out = {} if out is None else out
t = "null" if value is None else "array" if isinstance(value, list) else \
"object" if isinstance(value, dict) else "boolean" if isinstance(value, bool) else \
"number" if isinstance(value, (int, float)) else "string"
if prefix:
out[prefix] = t
if t == "object":
for k, v in value.items():
paths(v, f"{prefix}.{k}" if prefix else k, out)
if t == "array" and value:
paths(value[0], f"{prefix}[]", out)
return out
def shape_of(data):
sig = "|".join(f"{p}:{t}" for p, t in sorted(paths(data).items()))
return hashlib.sha256(sig.encode()).hexdigest()[:16]The bool check comes before the number check on purpose: in Python, True is an instance of int, and without that ordering every boolean would fingerprint as a number. Both versions sort paths by plain code-point order, so a JavaScript writer and a Python reader produce the same hash for the same response.
Staleness, re-enrichment and backfills
Every row has an expires_at. Re-enrichment is a scheduled query for rows whose expiry has passed and that some live record still references, turned into jobs. Three refinements make it affordable.
Refresh what is used. Records nobody has looked at in a year do not need fresh WHOIS data. Join against an activity signal, such as accounts still in an open pipeline or customers still active, and let the rest stay stale until someone opens them.
Use the provider's own freshness signals. Some responses tell you how current their underlying data is. Website Tech Stack Detector returns fingerprintsUpdatedAt, the time its fingerprint database was refreshed. When that date moves past the fetched_at of your stored results, a rescan may detect technologies the old fingerprints missed, even though the site itself has not changed.
curl "https://api.apiverve.com/v1/techstackdetector?url=https%3A%2F%2Fwordpress.org" \
-H "x-api-key: YOUR_API_KEY"Compare fingerprintsUpdatedAt with your stored fetched_at. Detections depend on the fingerprint set, not just on the site.
Backfill in batches, inside a budget. A backfill is re-enrichment at scale: a new stage added to the pipeline, a field you now need for every historical record, a TTL you shortened. It can be hundreds of thousands of calls, so plan it like a migration:
- Collect and deduplicate keys first. Ten thousand records at four hundred companies need four hundred WHOIS calls, not ten thousand. Duplicates in a batch are billed each time.
- Check the budget before starting. Multiply unique keys by the stage's credits and compare with the remaining balance, keeping a reserve for production traffic.
- Send up to 200 inputs per
POST /v1/<api>/batch. Each item inresultsis an ordinary envelope at the same index as its request. Items past your remaining credits come back individually as402; items past this minute's rate window come back as per-item429withretryAfter, inside an HTTP200. Write every successful item, requeue the429s, and stop on the first402. - Make it resumable. Process keys in a stable order and record the last completed chunk. A backfill that can resume after a crash never needs to be babysat.
- Give it its own key, so its usage shows up separately in analytics and a runaway backfill cannot be confused with production traffic.
Your backfill sends a 200-item batch to whoislookup. The HTTP status is 200, results 0-149 are ok, 150-179 have code 429 with retryAfter, and 180-199 have code 402. What should the backfill do?
What the pipeline gives you
Put together, the pipeline has properties a loop never has. A crash costs nothing but time, because every stage is idempotent and every result is keyed by its question. Every value in your product can be traced to an API, an input, a timestamp and an endpoint version. Drift shows up as a new shape in a GROUP BY, not as a customer complaint. Stale data is refreshed where it matters and left alone where it does not, and a backfill is a planned, budgeted, resumable job.
All of this assumes the facts come from someone else's API. The last lesson of this course asks when that is the wrong assumption, and how to decide what to build yourself: Build versus buy.