B2B Lead Sourcing Module

A production-grade TypeScript/Node.js pipeline that sources B2B leads from the WeWorkRemotely RSS feed, cleanses and enriches the data, validates it against a strict schema, stores it in SQLite, and POSTs standardized payloads to a webhook endpoint.

---

Table of Contents

• Overview
• Architecture
• Project Structure
• Installation
• Configuration
• Usage
  • Run the Pipeline
  • Start Mock Webhook Server
  • Inspect the Database
  • End-to-End Manual Test
• Modules
  • 1. Types & Schema (types.ts)
  • 2. RSS Fetcher (fetcher.ts)
  • 3. Cleaner & Enricher (cleaner.ts)
  • 4. Webhook Sender (sender.ts)
  • 5. Database Layer (database.ts)
  • 6. Orchestrator (index.ts)
  • 7. Mock Server (mock-server.ts)
• Database Schema
• Data Flow
• Error Handling Strategy
• Testing
• Scripts Reference
• Extending the Module

---

Overview

This module is designed for B2B outbound sales teams that need a steady stream of enriched company leads. It:

1. Discovers companies hiring remotely (a strong intent signal for B2B outreach)
2. Cleanses company names by stripping legal suffixes
3. Enriches each lead with:
   • Cleaned domain name (extracted from URLs or inferred from company name)
   • Tech stack keywords detected in the job description
   • Contact email addresses found in the posting
   • Inferred industry category
4. Persists all leads, quarantine records, and pipeline run history to a local SQLite database
5. Validates every lead against a Zod schema — invalid entries go to a quarantine log without crashing the pipeline
6. Delivers clean leads to any webhook endpoint with automatic retry and exponential backoff

---

Architecture

  WeWorkRemotely RSS Feed
          │
          ▼
  ┌─────────────────┐
  │   fetcher.ts    │  Axios GET + rss-parser, retry/backoff
  └────────┬────────┘
           │ Raw RSS Items
           ▼
  ┌─────────────────┐
  │   cleaner.ts    │  Name cleanup, domain extraction, email/tech parsing,
  │                 │  industry inference, Zod validation, quarantine
  └────────┬────────┘
           │ Validated Lead[]
           ├────────────────────────────┐
           ▼                            ▼
  ┌─────────────────┐       ┌──────────────────────┐
  │   sender.ts     │       │   database.ts        │
  │  (webhook POST) │       │  (SQLite persistence)│
  └────────┬────────┘       └──────────────────────┘
           │                           │
           ▼                  ┌─────────────────┐
  ┌─────────────────────┐     │   leads.db      │
  │  Webhook Endpoint   │     │  (SQLite file)  │
  └─────────────────────┘     └─────────────────┘

---

Project Structure

.
├── package.json              # Dependencies & scripts
├── tsconfig.json             # TypeScript configuration
├── jest.config.js            # Jest configuration
├── .env                      # Runtime environment variables
├── .env.example              # Environment variable template
├── README.md                 # This file
├── data/
│   └── leads.db              # SQLite database (auto-created)
├── src/
│   ├── index.ts              # Pipeline orchestrator (CLI entry point)
│   ├── mock-server.ts        # Express mock webhook receiver
│   └── sourcing/
│       ├── types.ts          # Zod schemas & TypeScript interfaces
│       ├── fetcher.ts        # RSS feed fetching with retry
│       ├── cleaner.ts        # Data sanitization & enrichment
│       ├── sender.ts         # Webhook POST with retry logic
│       └── database.ts       # SQLite persistence layer
└── tests/
    └── sourcing.test.ts      # 45 unit and integration tests

---

Installation

Prerequisites: Node.js 18+ and npm.

# Clone or enter the project directory
cd b2b-lead-sourcing

# Install dependencies
npm install

# Verify TypeScript compilation
npm run build

---

Configuration

All configuration is via environment variables (.env file):

Variable	Default	Description
WWR_RSS_URL	https://weworkremotely.com/categories/remote-programming-jobs.rss	Source RSS feed URL
WEBHOOK_URL	http://localhost:3000/webhook	Destination webhook endpoint
BATCH_SIZE	10	Number of leads sent per batch
MAX_RETRIES	3	Maximum HTTP attempts per request
RETRY_DELAY_MS	1000	Base delay in ms for exponential backoff
DB_PATH	./data/leads.db	SQLite database file path
NODE_ENV	development	Runtime environment
DEBUG	false	Enable raw item logging to stdout

Copy .env.example to .env and adjust:

cp .env.example .env

---

Usage

Run the Pipeline

npm run start

This executes the full pipeline:

1. Fetches and parses the RSS feed
2. Cleans and enriches each item
3. Stores valid leads and quarantine records in SQLite
4. Validates against the Zod schema (invalid items are quarantined)
5. POSTs valid leads to the webhook URL
6. Prints a summary with counts of fetched, valid, quarantined, sent, and errors

Start Mock Webhook Server

npm run mock-server

Starts an Express server on http://localhost:3000 with these endpoints:

Method	Path	Description
POST	/webhook	Receives and validates incoming leads; logs them to console
GET	/webhook/inspect	Returns all received leads as JSON

Database inspection endpoints (mock server):

Method	Path	Description
GET	/db/stats	Database statistics (total leads, runs, quarantines, sent)
GET	/db/leads	All leads from SQLite with latest pipeline run info
GET	/db/quarantines	Full quarantine log from database
GET	/db/pipeline-runs	Pipeline run history with counts

Inspect the Database

You can inspect the SQLite database directly with any SQLite client:

# Using sqlite3 CLI
sqlite3 data/leads.db

# Example queries:
SELECT * FROM pipeline_runs ORDER BY id DESC;
SELECT company_name, job_title, inferred_industry FROM leads;
SELECT * FROM quarantine_log;

End-to-End Manual Test

Terminal 1 — Start the mock receiver:

npm run mock-server
# Output: Mock webhook server running on http://localhost:3000

Terminal 2 — Run the pipeline:

npm run start

Inspect the mock server console to see enriched leads arrive in real time. Then check the database:

curl http://localhost:3000/db/stats
curl http://localhost:3000/db/leads

---

Modules

1. Types & Schema (types.ts)

src/sourcing/types.ts defines the core data contracts using Zod:

const LeadSchema = z.object({
  companyName:     z.string().min(1),        // Cleaned company name
  companyDomain:   z.string().min(1),        // e.g. "acme.com"
  jobTitle:        z.string().min(1),        // Original job title
  description:     z.string().default(''),   // Full job description (truncated to 2000 chars)
  sourceUrl:       z.string().url(),         // Original listing URL
  techStack:       z.array(z.string()).default([]),  // e.g. ["React", "AWS"]
  contactEmails:   z.array(z.string().email()).default([]),
  inferredIndustry: z.string().default('Technology'),  // Industry classification
  sourcedAt:       z.string().datetime(),    // ISO 8601 timestamp
});

Other exported types:

• RawRssItem — Raw RSS feed entry before processing
• QuarantinedItem — Records of items that failed validation
• FetchResult — Result of RSS feed fetch
• PipelineResult — Combined result of the entire pipeline run

2. RSS Fetcher (fetcher.ts)

src/sourcing/fetcher.ts is responsible for downloading and parsing the RSS feed.

Features:

• HTTP GET via Axios with configurable timeout (default 15s)
• XML parsing via rss-parser library
• Retry with exponential backoff: retries on network errors and 5xx responses (up to maxRetries times)
• No retry on 4xx: client errors throw immediately
• User-Agent header set to B2B-Lead-Sourcing/1.0
• Returns a FetchResult containing parsed items, feed title, and fetch timestamp

Error handling:

• Connection timeouts bubble up after exhausting retries
• Malformed XML is caught by rss-parser and rethrown
• Non-retryable HTTP errors are thrown immediately

3. Cleaner & Enricher (cleaner.ts)

src/sourcing/cleaner.ts transforms raw RSS items into validated Lead objects.

Company Name Normalization:

• Extracts company name from the title using separators ( is hiring, -, |, etc.)
• Strips legal entity suffixes: Inc, LLC, Ltd, Corp, GmbH, SA, PLC, and 15+ others
• Falls back to the raw title if no separator is found

Domain Extraction:

• Scans job description text for URLs, extracts clean hostname
• Falls back to {companyname}.com using the cleaned company name
• Final fallback: unknown-company.com

Tech Stack Keyword Matching:

• Scans description against a dictionary of 60+ technology keywords:
  • Frontend: React, Angular, Vue, Svelte, Next.js
  • Backend: Node.js, Python, Django, Go, Rust, Java, Spring Boot
  • Databases: PostgreSQL, MySQL, MongoDB, Redis, Elasticsearch
  • Cloud/DevOps: AWS, Azure, GCP, Docker, Kubernetes, Terraform
  • Mobile: React Native, Flutter, Swift
  • AI/ML: Machine Learning, NLP, LLM
  • Data: Spark, Kafka, Airflow, Databricks
• Uses word-boundary regex for accurate matching
• Deduplicates results

Email Extraction:

• Regex-based email detection in job description text
• Case normalization and deduplication

Industry Inference:

• Rule-based classification using keyword patterns in the description:
  • HealthTech — health, medic, biotech, pharma
  • FinTech — finance, banking, fintech, pay, insurance
  • EdTech — edu, learn, training
  • E-Commerce — ecommerce, shop, retail, marketplace
  • SaaS — saas, cloud, platform
  • DevTools — devtools, developer, api, infrastructure
  • Cybersecurity — cyber, security, secure, privacy
  • Gaming — gaming
  • RealEstate — real estate, property, proptech
  • Logistics — logistics, supply chain, shipping
  • Media — media, content, publish, news
  • Default: Technology

Validation & Quarantine:

• Each candidate lead is validated against LeadSchema using safeParse()
• Items that fail validation or have missing critical fields (title) are recorded in the quarantine log
• Quarantine records include the item index, reason for failure, and raw data for debugging
• The pipeline continues processing remaining items — no single bad item crashes the batch

4. Webhook Sender (sender.ts)

src/sourcing/sender.ts delivers validated leads to the configured webhook endpoint.

Features:

• Batch processing — leads are divided into batches of configurable size
• Per-lead POST — each lead is sent as an individual JSON payload
• Retry with exponential backoff — transient failures (network, 5xx) are retried
• Fast-fail on 4xx — 400 and 422 responses are treated as unrecoverable client errors
• Graceful degradation — if one lead fails, the remaining leads continue sending
• Timeout per request: 5 seconds

Retry formula:

delay = RETRY_DELAY_MS * 2^(attempt - 1)

5. Database Layer (database.ts)

src/sourcing/database.ts provides a full SQLite persistence layer using better-sqlite3.

Class: LeadDatabase

Method	Description
createPipelineRun(feedUrl)	Creates a new pipeline run record, returns the run object
updatePipelineRun(id, fields)	Updates status, counts, and timestamps on a run
insertLead(runId, lead)	Inserts a single lead, returns the row ID
insertLeads(runId, leads[])	Transactional bulk insert of multiple leads
insertQuarantine(runId, item)	Inserts a single quarantine record
insertQuarantines(runId, items[])	Transactional bulk insert of quarantines
markLeadSent(id)	Marks a single lead as sent with timestamp
markAllLeadsSent(runId)	Marks all unsent leads in a run as sent
getLeads(runId?)	Retrieves leads, optionally filtered by pipeline run
getLeadById(id)	Retrieves a single lead by ID
getPipelineRun(id)	Retrieves a single pipeline run
getLatestPipelineRun()	Returns the most recent pipeline run
getAllPipelineRuns()	Returns all pipeline runs ordered by recency
getQuarantines(runId?)	Retrieves quarantine records
getStats()	Returns aggregate statistics across all tables

Singleton access:

import { getDatabase } from './sourcing/database';
const db = getDatabase(); // defaults to DB_PATH from env

6. Orchestrator (index.ts)

src/index.ts is the CLI entry point that coordinates the three-stage pipeline:

1. Fetch — calls fetchFeed() with config from environment
2. Clean — calls cleanItems() and separates valid leads from quarantined
3. Store — persists leads and quarantine records to SQLite
4. Send — calls sendLeads() with valid leads and marks them sent on success
5. Report — prints a detailed summary with counts of all outcomes

The pipeline exits with code 1 if any unrecoverable error occurs.

7. Mock Server (mock-server.ts)

src/mock-server.ts is a lightweight Express server for development and testing.

• Validates incoming payloads against LeadSchema
• Logs each received lead's company, domain, job, tech stack, emails, and industry
• Returns 422 with validation details for invalid payloads
• Persists received leads to the SQLite database
• Provides REST endpoints for database inspection

Endpoints:

Method	Path	Description
POST	/webhook	Receive and persist a lead
GET	/webhook/inspect	All received leads from DB
GET	/db/stats	Database statistics
GET	/db/leads	All leads with latest pipeline run
GET	/db/quarantines	Full quarantine log
GET	/db/pipeline-runs	Pipeline run history

---

Database Schema

The SQLite database uses three related tables with proper indexing:

-- Tracks each pipeline execution
CREATE TABLE pipeline_runs (
  id                INTEGER PRIMARY KEY AUTOINCREMENT,
  status            TEXT    NOT NULL DEFAULT 'running',
  feed_url          TEXT    NOT NULL,
  total_raw         INTEGER NOT NULL DEFAULT 0,
  valid_count       INTEGER NOT NULL DEFAULT 0,
  quarantined_count INTEGER NOT NULL DEFAULT 0,
  sent_count        INTEGER NOT NULL DEFAULT 0,
  error_count       INTEGER NOT NULL DEFAULT 0,
  started_at        TEXT    NOT NULL,
  finished_at       TEXT
);

-- Validated, enriched leads
CREATE TABLE leads (
  id                INTEGER PRIMARY KEY AUTOINCREMENT,
  pipeline_run_id   INTEGER NOT NULL,
  company_name      TEXT    NOT NULL,
  company_domain    TEXT    NOT NULL,
  job_title         TEXT    NOT NULL,
  description       TEXT    DEFAULT '',
  source_url        TEXT    NOT NULL,
  tech_stack        TEXT    DEFAULT '[]',       -- JSON array
  contact_emails    TEXT    DEFAULT '[]',       -- JSON array
  inferred_industry TEXT    DEFAULT 'Technology',
  sourced_at        TEXT    NOT NULL,
  sent              INTEGER NOT NULL DEFAULT 0,
  sent_at           TEXT,
  created_at        TEXT    NOT NULL DEFAULT (datetime('now')),
  FOREIGN KEY (pipeline_run_id) REFERENCES pipeline_runs(id)
);

-- Items that failed validation
CREATE TABLE quarantine_log (
  id                INTEGER PRIMARY KEY AUTOINCREMENT,
  pipeline_run_id   INTEGER NOT NULL,
  item_index        INTEGER NOT NULL,
  reason            TEXT    NOT NULL,
  raw_data          TEXT    DEFAULT '{}',       -- JSON object
  created_at        TEXT    NOT NULL DEFAULT (datetime('now')),
  FOREIGN KEY (pipeline_run_id) REFERENCES pipeline_runs(id)
);

-- Indexes for efficient queries
CREATE INDEX idx_leads_pipeline ON leads(pipeline_run_id);
CREATE INDEX idx_leads_company  ON leads(company_name);
CREATE INDEX idx_leads_domain   ON leads(company_domain);
CREATE INDEX idx_leads_industry ON leads(inferred_industry);
CREATE INDEX idx_quarantine_pipeline ON quarantine_log(pipeline_run_id);

---

Data Flow

RSS XML
  │
  ▼
fetcher.ts ──► RawRssItem[] ──► cleaner.ts
                                    │
                                    ├── cleanCompanyName()      "Acme LLC" → "Acme"
                                    ├── attemptDomainExtraction() → "acme.com"
                                    ├── extractEmails()          → ["hr@acme.com"]
                                    ├── extractTechStack()       → ["React", "AWS"]
                                    ├── inferIndustry()          → "SaaS"
                                    │
                                    ▼
                              Zod LeadSchema
                              /            \
                         valid              invalid
                           │                  │
                     ┌─────┴─────┐            │
                     ▼           ▼            ▼
              database.ts    sender.ts   Quarantine Log
              (SQLite)       (webhook)       │
                                │            ▼
                                ▼       database.ts
                          Webhook       (SQLite)
                          POST

---

Error Handling Strategy

Layer	Error Type	Handling
Fetcher	Network timeout	Retry with exponential backoff up to MAX_RETRIES
Fetcher	4xx HTTP error	Throws immediately (no retry)
Fetcher	Malformed XML	Throws immediately
Fetcher	5xx HTTP error	Retry with exponential backoff
Cleaner	Missing title	Quarantined (pipeline continues)
Cleaner	Zod validation failure	Quarantined with detailed error message
Cleaner	Unexpected exception	Caught and quarantined
Database	SQL constraint violation	Thrown; logged as pipeline error
Sender	Network error / 5xx	Retry per-lead with backoff
Sender	400 / 422 response	Failed immediately (no retry)
Sender	One lead failure	Other leads continue sending
Pipeline	Fatal fetch/clean error	Pipeline marks run as failed in DB and exits

---

Testing

The test suite contains 45 tests covering all modules and edge cases.

npm run test

Test coverage includes:

• Schema Validation (5 tests): valid leads, missing fields, invalid URLs, invalid emails, default values
• Name & Domain Cleaning (7 tests): suffix removal for LLC, Inc., Ltd, Corp; domain from URL; name-based domain fallback; title separator parsing
• Tech Stack Extraction (4 tests): single keyword, multiple keywords, no keywords, duplicate deduplication
• Email Extraction (3 tests): multiple emails, deduplication, no emails
• Industry Inference (3 tests): FinTech, HealthTech, default Technology
• Validation & Quarantine (2 tests): missing title isolation, mixed good/bad items
• Webhook Sender (7 tests): success, retry-then-success, retry exhaustion, 400 fast-fail, timeout, multi-lead, partial failure
• Feed Fetcher (2 tests): malformed XML, connection timeout
• Database Layer (12 tests): pipeline run CRUD, lead insert/retrieve, bulk transactions, sent marking, quarantine insert, stats aggregation, missing record handling

---

Scripts Reference

Script	Command	Description
npm run start	ts-node src/index.ts	Run the lead sourcing pipeline
npm run mock-server	ts-node src/mock-server.ts	Start mock webhook receiver with DB inspection
npm run test	jest --verbose	Run all 45 tests with verbose output
npm run build	tsc	Compile TypeScript to dist/
npm run lint	tsc --noEmit	Type-check without emitting files

---

Extending the Module

Adding Tech Stack Keywords

Edit the TECH_KEYWORDS array in src/sourcing/cleaner.ts:

const TECH_KEYWORDS = [
  // ... existing keywords
  'NewFramework',
];

Adding Industry Classifications

Add a new regex pattern in the inferIndustry function in src/sourcing/cleaner.ts:

if (/\b(legal|law|compliance)\b/.test(lower)) return 'LegalTech';

Adding More RSS Sources

Extend fetcher.ts to accept multiple URLs or create a dispatcher in index.ts that processes multiple feeds.

Changing the Database Schema

Modify the SCHEMA constant in src/sourcing/database.ts and add corresponding methods to the LeadDatabase class. The database auto-creates tables on first connection.

Changing the Webhook Format

Modify the Lead interface in src/sourcing/types.ts and update the LeadSchema to match your target API contract. The database schema columns will also need to be updated to match.