How to Build an Event-Driven Lead Scoring Pipeline with Node.js and PostgreSQL

Community Article
Community articles are authored by SitePoint Premium contributors. Content is screened before publication, and SitePoint reserves the right to moderate or remove articles that violate our guidelines. Views expressed are those of the authors and do not necessarily reflect those of SitePoint.

How to Build an Event-Driven Lead Scoring Pipeline with Node.js and PostgreSQL

VPVasyl PopovychPublished inNode.js·
September 17, 2026
·Updated:September 17, 2026

The AI briefing for Developers

Stay up to date with AI tools, model releases, and developer workflows that matter.

Weekly. Free. One click to leave.

SitePoint Premium
Stay Relevant and Grow Your Career in Tech

• Premium Results
• Publish articles on SitePoint
• Daily curated jobs
• Learning Paths
• Discounts to dev tools

7 Day Free Trial. Cancel Anytime.

A trial signup, a first project, and a teammate invitation tell you different things about a potential customer. Yet many applications send those actions to separate systems, leaving someone to reconcile the activity before deciding who needs a sales conversation.

An event-driven lead scoring pipeline makes that decision repeatable. Your application reports meaningful actions, a scoring service applies explicit rules, and a database records both the result and the evidence behind it.

The arithmetic is simple. The harder part is processing the same webhook twice without awarding extra points, handling two events arriving together, and preserving a sales notification when a downstream service fails.

In this tutorial, we will build a small Node.js service that handles those cases with PostgreSQL transactions, database constraints, and a transactional outbox. You will have a signed event endpoint, an auditable scoring history, and a durable record of leads that need sales review.

Define what the score means

We will separate customer fit from product behavior. Fit describes whether a lead belongs to the audience the business can serve. Behavior describes what that lead has done in the product.

For this example, a lead receives 20 fit points if its company matches both the target industry and company-size range. Otherwise, it receives zero. A trusted application or CRM assigns this value before sending behavioral events. The webhook cannot change it.

Our behavioral model uses four milestones:

A lead enters the review queue when it has 20 fit points and at least 40 behavioral points. A poor-fit lead cannot qualify simply by accumulating more activity. These weights and thresholds are teaching examples, not validated predictors of revenue.

Each milestone earns points once per lead for the lifetime of this demo model. Creating ten projects will not produce ten scoring increments. The maximum behavioral score is 75, and the maximum combined score is 95.

Crossing the threshold requests sales review. It does not establish budget, buying authority, or sales qualification. In a real workflow, a direct demo request should also have its own response path so a scoring rule cannot silently suppress it.

Set the boundaries of the example

This service accepts events from one trusted backend for one business. Lead records already exist, event IDs are globally unique, and retries preserve the original event body. Anonymous identity resolution and account-level aggregation are outside the example.

1. A backend creates and signs an event after a product action succeeds.
2. The scoring API verifies the signature and validates the event.
3. One database transaction records the event, updates the lead, and, when appropriate, creates an outbox message.
4. A separate dispatcher later delivers that message to a CRM or another internal service.

An outbox is a database table of messages waiting to be delivered. Keeping the message in the same transaction as the score prevents a committed scoring decision from losing its corresponding handoff request.

The tutorial implements steps one through three and defines the delivery contract for step four. A CRM-specific dispatcher is a separate integration. No external sales messages are sent by the sample code.

Prepare the project

The example targets Node.js 24, PostgreSQL 17, Express 5, and version 8 of the pg package. You will need Docker for the database commands below, along with basic JavaScript and SQL knowledge. Use a fresh local project and database.

mkdir lead-scoring-democd lead-scoring-demonpm init -ynpm install express@5 pg@8

Use .mjs filenames for the JavaScript examples so Node treats them as ES modules. Keep the generated package lockfile when sharing or deploying the project.

Start an isolated development database:

docker run --name lead-scoring-postgres -e POSTGRES_USER=scorer -e POSTGRES_PASSWORD=local-demo-only -e POSTGRES_DB=lead_scoring -p 127.0.0.1:5433:5432 -d postgres:17

Create a local .env file containing the following values. The database password is for this local demo only; do not reuse it in a deployed service.

DATABASE_URL=postgres://scorer:local-demo-only@127.0.0.1:5433/lead_scoringWEBHOOK_SECRET=replace-with-your-own-random-secret

Generate a signing secret with the following command and paste its output into .env:

node -e "console.log(require('node:crypto').randomBytes(32).toString('hex'))"

Keep .env out of version control and never expose this secret to a browser. A frontend can report an action to your application, but a trusted backend must verify the action and associate it with the authenticated lead before signing it.

Save the following as schema.sql:

CREATETABLEleads(id uuid PRIMARYKEY,fit_score integer NOTNULLCHECK(fit_score IN(0,20)),behavior_score integer NOTNULLDEFAULT0CHECK(behavior_score BETWEEN0AND75),review_requested_at timestamptz);CREATETABLElead_events(id uuid PRIMARYKEY,lead_id uuid NOTNULLREFERENCESleads(id),event_type text NOTNULLCHECK(event_type IN('trial_started','project_created','teammate_invited','demo_requested')),occurred_at timestamptz NOTNULL,received_at timestamptz NOTNULLDEFAULTnow(),points integer NOTNULLCHECK(points BETWEEN0AND30),model_version text NOTNULL);CREATEUNIQUEINDEX one_credit_per_milestoneONlead_events(lead_id, event_type)WHERE points >0;CREATEINDEX lead_event_history ONlead_events(lead_id, received_at);CREATETABLEoutbox(id uuid PRIMARYKEYDEFAULTgen_random_uuid(),lead_id uuid NOTNULLUNIQUEREFERENCESleads(id),payload jsonb NOTNULL,created_at timestamptz NOTNULLDEFAULTnow(),delivered_at timestamptz);INSERTINTOleads(id, fit_score)VALUES('11111111-1111-4111-8111-111111111111',20);

The leads table holds the current score and the first review-request timestamp. lead_events explains where the score came from, including events that were accepted but earned no additional points. outbox records the handoff request.

Two different uniqueness rules matter here. The event ID prevents a delivery retry from being processed again. The partial unique index permits at most one positive-point event for a particular lead and milestone, even when a sender uses a new event ID. The outbox’s unique lead ID limits this example to one review request per lead.

Wait for PostgreSQL to be ready, then run the schema once against the fresh database:

docker exec lead-scoring-postgres pg_isready -U scorerdocker exec -i lead-scoring-postgres psql -U scorer -d lead_scoring -v ON_ERROR_STOP=1< schema.sql

For an existing application, make the schema a versioned migration. The seeded UUID gives us a known, suitable lead for the walkthrough.

Keep scoring rules on the server

Save the rules and event validation as model.mjs:

exportconstMODEL_VERSION='v1';exportconstWEIGHTS=Object.freeze({trial_started:10,project_created:20,teammate_invited:15,demo_requested:30,});exportfunctionreadyForReview(fit, behavior){return fit ===20&& behavior >=40;}exportfunctionvalidateEvent(value){const uuid =newRegExp('^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}'+'-[89ab][0-9a-f]{3}-[0-9a-f]{12}$',);const fields =['id','leadId','type','occurredAt'];if(!value ||typeof value !=='object'||Array.isArray(value)||Object.keys(value).length!== fields.length||!fields.every(key=>typeof value[key]==='string')){thrownewError('Expected exactly four string fields');}if(!uuid.test(value.id)||!uuid.test(value.leadId)||!Object.hasOwn(WEIGHTS, value.type)){thrownewError('Invalid ID or event type');}const time =newDate(value.occurredAt);if(!Number.isFinite(time.getTime())|| time.toISOString()!== value.occurredAt|| time.getTime()>Date.now()+300_000){thrownewError('Use a valid UTC ISO timestamp');}return value;}

The incoming event contains exactly four fields: an event ID, a lead ID, an event type, and its occurrence time. It contains no points or fit score. Allowing callers to supply scoring values would make the model depend on data they control.

This validator intentionally accepts a narrow timestamp format: the UTC ISO string returned by Date.toISOString(), including milliseconds and the final Z. It also rejects timestamps more than five minutes in the future. It accepts old events because these are lifetime milestones; we will revisit that limitation later.

The model version is stored with each event. Do not change the weights in a live deployment without a migration or a deliberate replay plan: historical points and the materialized lead score must remain consistent.

Authenticate the original request bytes

A webhook should prove that it came from the trusted producer. We will sign the delivery timestamp followed by a period and the exact request body using HMAC-SHA256.

Save this code as signatures.mjs:

import{ createHmac, timingSafeEqual }from'node:crypto';exportfunctionsign(body, timestamp, secret){returncreateHmac('sha256', secret).update(`${timestamp}.`).update(body).digest('hex');}exportfunctionverify(body, timestamp, signature, secret){if(!Buffer.isBuffer(body)||!/^[0-9]{10}$/.test(timestamp ??'')||!/^[0-9a-f]{64}$/.test(signature ??''))returnfalse;const age =Math.abs(Date.now()/1000-Number(timestamp));if(age >300)returnfalse;const expected =Buffer.from(sign(body, timestamp, secret),'hex');returntimingSafeEqual(expected,Buffer.from(signature,'hex'));}

Verification rejects malformed signatures and deliveries outside a five-minute window before comparing the digests. The length check matters because Node’s timingSafeEqual requires buffers of equal length. See the official Node.js crypto documentation for its behavior and timing caveats.

A timestamp limits replay age; the database event ID handles replays within the accepted window. A legitimate retry after that window needs a new delivery timestamp and signature, but the same event ID, occurrence time, and body. Keep both machines’ clocks synchronized.

This is a custom backend-to-backend signing contract. It is not a substitute for a third-party provider’s documented webhook-verification scheme. Use HTTPS in deployment: a signature authenticates the message but does not encrypt its contents.

Apply the event in one transaction

Save the database operation as score.mjs:

import{WEIGHTS,MODEL_VERSION, readyForReview }from'./model.mjs';functionhttpError(status, message){returnObject.assign(newError(message),{ status });}exportasyncfunctionapplyEvent(pool, event){const client =await pool.connect();try{await client.query('BEGIN ISOLATION LEVEL READ COMMITTED');await client.query("SET LOCAL lock_timeout = '2s'");await client.query("SET LOCAL statement_timeout = '5s'");const{rows: leads }=await client.query('SELECT * FROM leads WHERE id = $1 FOR UPDATE',[event.leadId],);if(!leads.length)throwhttpError(404,'Unknown lead');const lead = leads[0];const{rows: prior }=await client.query(`SELECT 1 FROM lead_eventsWHERE lead_id = $1 AND event_type = $2 AND points > 0`,[event.leadId, event.type],);const points = prior.length?0:WEIGHTS[event.type];const inserted =await client.query(`INSERT INTO lead_events(id, lead_id, event_type, occurred_at, points, model_version)VALUES ($1, $2, $3, $4, $5, $6)ON CONFLICT (id) DO NOTHING RETURNING id`,[event.id, event.leadId, event.type, event.occurredAt,points,MODEL_VERSION],);if(!inserted.rowCount){const{rows:[saved]}=await client.query('SELECT * FROM lead_events WHERE id = $1',[event.id],);if(saved.lead_id!== event.leadId|| saved.event_type!== event.type|| saved.occurred_at.toISOString()!== event.occurredAt){throwhttpError(409,'Event ID reused with different data');}await client.query('COMMIT');return{duplicate:true};}const behavior = lead.behavior_score+ points;const total = lead.fit_score+ behavior;const requestReview =!lead.review_requested_at&&readyForReview(lead.fit_score, behavior);await client.query(`UPDATE leads SET behavior_score = $2,review_requested_at = CASE WHEN $3 THEN now()ELSE review_requested_at ENDWHERE id = $1`,[event.leadId, behavior, requestReview],);if(requestReview){await client.query('INSERT INTO outbox (lead_id, payload) VALUES ($1, $2)',[event.leadId,JSON.stringify({type:'lead.review_requested',leadId: event.leadId,score: total,modelVersion:MODEL_VERSION,})],);}await client.query('COMMIT');return{duplicate:false, points, total, requestReview };}catch(error){await client.query('ROLLBACK');throw error;}finally{client.release();}}

There are several deliberate choices in this transaction.

First, it locks the lead before inserting an event. Every writer that changes scoring state should follow this order. Concurrent events for the same lead wait their turn, while other leads can continue processing. PostgreSQL holds the row lock until the transaction ends. Its row-lock documentation describes how FOR UPDATE blocks conflicting writers.

Second, all statements use one checked-out client. Using separate pool.query() calls could send statements to different connections and break the transaction. This is an explicit requirement in the node-postgres transaction guide.

Third, the transaction uses Read Committed isolation. After waiting for another writer, its subsequent queries can see that writer’s committed milestone. That is important for the query that decides whether to award points. See PostgreSQL transaction isolation for the snapshot behavior.

Fourth, ON CONFLICT (id) DO NOTHING makes retries harmless, but we also compare the saved event with the submitted data. Reusing an ID for a different event produces 409 Conflict, rather than silently hiding a producer bug. PostgreSQL’s INSERT documentation explains the conflict and RETURNING clauses.

Finally, the score, review timestamp, and outbox message commit together. If any statement fails, none of those changes is retained. A database timeout becomes a retryable API failure. The producer must retry with the same event ID and bounded exponential backoff with jitter.

SQL values use placeholders rather than string concatenation, following the parameterized query pattern. The local lock and statement timeouts also prevent a request from waiting indefinitely for a blocked statement.

Expose the event endpoint

importexpressfrom'express';importpgfrom'pg';import{ validateEvent }from'./model.mjs';import{ verify }from'./signatures.mjs';import{ applyEvent }from'./score.mjs';const{DATABASE_URL,WEBHOOK_SECRET}= process.env;if(!DATABASE_URL||!WEBHOOK_SECRET){thrownewError('Set DATABASE_URL and WEBHOOK_SECRET');}const pool =newpg.Pool({connectionString:DATABASE_URL,max:10,connectionTimeoutMillis:3000,});pool.on('error',error=>console.error('db_pool', error.code));const app =express();app.post('/events', express.raw({type:'application/json',limit:'16kb',inflate:false,}),async(req, res)=>{if(!verify(req.body, req.get('x-timestamp'),req.get('x-signature'),WEBHOOK_SECRET)){return res.status(401).json({error:'Invalid signature'});}let event;try{event =validateEvent(JSON.parse(req.body.toString('utf8')));}catch{return res.status(400).json({error:'Invalid event'});}try{return res.json(awaitapplyEvent(pool, event));}catch(error){console.error('event_failed',{id: event.id,code: error.code});return res.status(error.status??503).json({error: error.status? error.message:'Retry later',});}});app.use((error, req, res, next)=>{res.status(error.status??500).json({error:'Request rejected'});});app.listen(3000,'127.0.0.1',()=>console.log('Listening on 3000'));

The route uses a raw body parser so verification sees the original bytes. Parsing JSON and serializing it again can change whitespace or key order and invalidate the signature. Do not put a global JSON parser in front of this route. Express documents raw body parsing and its size limits in its API reference.

The service responds successfully only after the transaction commits. It returns 400 for an invalid event, 401 for a failed signature, 404 for an unknown lead, and 409 for a conflicting event ID. Unexpected processing failures return 503, allowing the producer to retry.

node --env-file=.env server.mjs

It binds to localhost for the demonstration. A deployed service also needs TLS termination, rate limiting, monitoring, graceful shutdown, and restricted database credentials. The development database user is not an appropriate production application role.

Send events and verify the result

import{ randomUUID }from'node:crypto';import{ sign }from'./signatures.mjs';const secret = process.env.WEBHOOK_SECRET;if(!secret)thrownewError('Set WEBHOOK_SECRET');for(const type of['trial_started','project_created','teammate_invited',]){const body =Buffer.from(JSON.stringify({id:randomUUID(),leadId:'11111111-1111-4111-8111-111111111111',type,occurredAt:newDate().toISOString(),}));for(let attempt =0; attempt <2; attempt++){const timestamp =String(Math.floor(Date.now()/1000));const response =awaitfetch('http://127.0.0.1:3000/events',{method:'POST',headers:{'content-type':'application/json','x-timestamp': timestamp,'x-signature':sign(body, timestamp, secret),},body,});console.log(response.status,await response.json());}}

node --env-file=.env send.mjs

Against the freshly seeded database, the expected response sequence is:

200{duplicate:false,points:10,total:30,requestReview:false}200{duplicate:true}200{duplicate:false,points:20,total:50,requestReview:false}200{duplicate:true}200{duplicate:false,points:15,total:65,requestReview:true}200{duplicate:true}

The total includes the lead’s 20 fit points. Three distinct milestones produce 45 behavioral points, so the lead receives one review request.

docker exec lead-scoring-postgres psql -U scorer -d lead_scoring -c 'SELECT fit_score, behavior_score, review_requested_at FROM leads;'docker exec lead-scoring-postgres psql -U scorer -d lead_scoring -c 'SELECT event_type, points, model_version FROM lead_events;'docker exec lead-scoring-postgres psql -U scorer -d lead_scoring -c 'SELECT payload, delivered_at FROM outbox;'

Expect three event rows and one undelivered outbox row. Running the sender again creates new event IDs, so those events are not delivery duplicates. They should still earn zero points because their milestones have already been credited. The outbox should remain at one row.

Deliver the handoff independently

The outbox records an obligation to send a message; it does not prove that the CRM has received it. A dispatcher should read pending rows, submit each payload, and mark it delivered only after a successful response.

For a single-worker prototype, select a pending row under a transaction lock:

SELECT id, payloadFROM outboxWHERE delivered_at ISNULLORDERBY created_at, idLIMIT1FORUPDATESKIPLOCKED;

Send the row’s UUID as a stable idempotency key to a receiver that explicitly supports it. After a confirmed success, set delivered_at and commit. If delivery fails, roll back and retry later. PostgreSQL documents SKIP LOCKED as useful for queue-like access in its SELECT reference.

An HTTP timeout leaves a difficult case: the receiver may have completed the action even though the dispatcher never received the response. A retry can therefore deliver the same notification again. The receiver must deduplicate the outbox ID; adding an arbitrary HTTP header does not create that guarantee. This is at-least-once delivery, not end-to-end exactly-once processing.

For production, avoid holding database locks during slow network calls. Add a short claim transaction, a lease-expiry field, an ownership token, attempt counts, and a next-attempt timestamp. A worker can then claim a job, commit, send it, and acknowledge it only if it still owns the lease. Expired leases allow recovery after a crash. Cap retries, monitor overdue messages, and retain failed jobs for investigation.

The event producer needs durability too. If the product action commits but the backend crashes before sending its event, this scoring service cannot recover something it never received. Use an outbox in the originating application or another durable event-delivery mechanism.

Test failure cases before connecting sales tools

The unit tests should cover the fit gate, threshold boundaries, unknown event types, malformed IDs, future timestamps, signature tampering, and stale deliveries. Node provides a built-in test runner, so the rule tests need no test framework.

For example, save this short test as rules.test.mjs:

importtestfrom'node:test';importassertfrom'node:assert/strict';import{ readyForReview }from'./model.mjs';test('sales review requires fit and sufficient behavior',()=>{assert.equal(readyForReview(20,39),false);assert.equal(readyForReview(20,40),true);assert.equal(readyForReview(0,75),false);});

node --test rules.test.mjs

Unit tests alone cannot establish transaction correctness. Run the following integration checks against a disposable PostgreSQL database before publishing or deploying your adaptation:

Exercise the actual HTTP endpoint for signature tests and the actual database for concurrency tests. Mocking a row lock does not test PostgreSQL’s locking behavior.

Decide what to add before production

This first version uses permanent milestone credit. Old activity does not expire, so it describes historical engagement rather than current buying intent. Delayed events can also trigger a review long after the activity occurred. For time-sensitive scoring, define an event-age policy and add time-based recalculation.

A rolling-window model needs more than subtracting a fixed amount from every lead. Recompute each lead’s behavior from qualifying events in the chosen window, apply repeat-event caps consistently, and schedule recalculation for inactive leads as well as those generating new events. Use the same lead-locking discipline as the ingestion path. Define whether a previously reviewed lead can qualify again, and change the outbox uniqueness rule to identify a qualification cycle if necessary.

Other extensions depend on the application:

1. For multiple tenants or event producers, include authenticated tenant and source identifiers in keys and access checks.
   Do not trust a tenant ID merely because it appears in JSON.
2. Treat fit changes as explicit, auditable updates that acquire the lead lock and re-evaluate the review rule. Changing fit_score directly will not trigger this code by itself.
3. Add a way to reverse invalid activity. This minimal model has no event corrections, cancellations, or negative scores.
4. Set a retention policy, document access rights, and avoid copying unnecessary personal information into event payloads or logs. If you prune raw history, retain the minimal deduplication and milestone-credit records needed to prevent an old event from earning points again.
5. Compare score bands with later outcomes such as accepted opportunities and paid conversion. Tune weights using evidence from the product rather than treating the example numbers as benchmarks.

Make the score explainable

A useful score should be easy to inspect. For a lead with a total of 65, this model can show the stored fit score, three credited milestones, their timestamps, and the model version that assigned the points. An operator can also see whether the review request is still waiting in the outbox.

That visibility is the value of building the pipeline carefully. Stable IDs make retries safe, row locks protect concurrent updates, and a shared transaction keeps the scoring decision connected to its handoff. Start with a few defensible signals, verify the failure cases, and expand the model only when customer outcomes justify the extra complexity.