Author: Daniel Gerlag, Principal Software Engineer
In architectural discussions surrounding modern data systems, we previously examined how pivoting from an event-driven to a change-driven paradigm introduces a highly efficient, low-cost model for dependency inversion. Rather than requiring downstream systems to reconstruct context from a deluge of raw events, this approach leverages Drasi, an open-source data change processing platform, to evaluate complex data variations across disparate sources and trigger instantaneous reactions. By utilizing @drasi/lib, developers can embed this continuous-query engine directly in-process within Node.js applications, achieving sophisticated change detection without provisioning additional infrastructure.
Every Node.js developer has written this function. Maybe you called it sync, or poll, or tick. It looked something like this:
setInterval(async () => {
const rows = await db.query('SELECT * FROM orders WHERE status = $1', ['open']);
const added = rows.filter((r) => !lastRows.some((p) => p.id === r.id));
const removed = lastRows.filter((p) => !rows.some((r) => r.id === p.id));
const changed = rows.filter((r) => {
const prev = lastRows.find((p) => p.id === r.id);
return prev && prev.total !== r.total;
});
if (added.length || removed.length || changed.length) notify({ added, removed, changed });
lastRows = rows;
}, 5000); It works. It also hammers the database with a query whose answer is usually identical to the last one, and it reacts up to five seconds late. Changes that happen and un-happen inside the same window never show up at all. The diffing logic grows a new branch every time someone asks a slightly different question, and when you need a second query, you copy the whole thing.
None of this code is about your product. It is bookkeeping: the machinery of working out what changed, written by hand, over and over.
The usual escape hatch is infrastructure. Kafka and a stream processor, Flink, ksqlDB, a managed streaming database. Those are excellent tools. They are also a cluster to run and a budget to defend, and plenty of Node.js applications will never justify either, so the polling loop stays.
@drasi/lib offers a third option: npm install, and no servers at all.
Event-driven, meet change-driven
Node developers already think in events. Something fires, a handler runs. EventEmitter, webhooks and message queues all work that way.
Drasi is change-driven:
Event-driven tells you something happened. Change-driven tells you the answer to your question is now different, and how.
An event stream hands you raw occurrences and leaves you to reconstruct meaning from them. With a change-driven system you declare the meaning once, up front, as a query. The engine keeps the answer to that query continuously correct and tells you which results appeared, changed or disappeared.
Drasi is an open-source engine built around this model, and it has three moving parts:
Sources → Continuous Queries → Reactions
- Sources feed changes in.
- Continuous Queries declare what changes matter.
- Reactions do something when the answer moves.
Drasi ships in a few forms, including a standalone server. @drasi/lib is the one Node developers should care about: the same engine, compiled as a native Node addon and running inside your process. No server, no broker, no network hop. Just a dependency in your package.json.
Hello, @drasi/lib
Here is a complete program. It runs as written.
import { Drasi } from '@drasi/lib';
const drasi = await Drasi.create('my-app');
await drasi.start();
// 1. A source you push changes into from your own code.
await drasi.addJsSource('orders');
// 2. A continuous query that declares what you care about.
await drasi.addQuery(
'open-orders',
`MATCH
(o:Order)
WHERE
o.status = 'open'
RETURN
o.id AS id,
o.total AS total`,
['orders'],
);
// 3. A reaction: a plain JavaScript callback.
await drasi.addJsReaction('print', ['open-orders'], (event) => {
console.log('open-orders changed:', event.results);
});
// Now push some changes.
await drasi.pushChange('orders', {
op: 'insert', id: 'o1', labels: ['Order'],
properties: { id: 'o1', status: 'open', total: 42 },
});
await drasi.pushChange('orders', {
op: 'insert', id: 'o2', labels: ['Order'],
properties: { id: 'o2', status: 'open', total: 17 },
});
// Close o1. It drops out of the query, and the reaction fires.
await drasi.pushChange('orders', {
op: 'update', id: 'o1', labels: ['Order'],
properties: { id: 'o1', status: 'closed', total: 42 },
}); Ten lines of setup and no infrastructure. The query is declared once, its result set stays correct on its own, and your callback hears about every movement.
Nothing here keeps a previous copy of the rows, and nothing computes a diff.
Cypher, without the graph database
The query language is Cypher, and GQL is also supported. That trips people up, so let's deal with it now: you are not running a graph database. Nothing is being stored in Neo4j. Drasi models incoming data as a property graph internally, because a graph is what lets a single query span several unrelated sources.
In practice you need three ideas: MATCH a pattern of labelled nodes, WHERE to filter, RETURN to shape the output. Anyone comfortable with SQL will pick it up quickly.
You get diffs, not snapshots
When your reaction fires, it does not receive the whole result set. It receives the change to the result set:
{
query_id: 'open-orders',
sequence: 2,
timestamp: '2026-08-04T20:32:46.115998Z',
metadata: { processed_by: 'drasi-core', result_count: 1, source_id: 'orders' },
results: [
{ type: 'DELETE', data: { id: 'o1', total: 42 }, row_signature: 18083160937290286091n }
],
} Each entry carries a type: ADD, UPDATE or DELETE for row-level changes, and aggregation when a grouped value moves. Updates carry both before and after, so you can see exactly what moved:
await drasi.addJsReaction('audit', ['open-orders'], (event) => {
for (const diff of event.results) {
if (diff.type === 'ADD') console.log('new open order', diff.data);
if (diff.type === 'DELETE') console.log('order closed', diff.data);
if (diff.type === 'UPDATE') console.log('total changed', diff.before, '→', diff.after);
}
}); That is what separates a continuous query from a query you happen to re-run on a timer. The engine maintains the result set incrementally: when a change arrives, it computes the minimal set of result rows affected. It does not recompute the query, and it does not hand you a snapshot and leave you to work out the delta. All the work in that opening setInterval now happens inside the engine.
You can still read current state with await drasi.getQueryResults('open-orders'), which is what you want for an initial page render.
Three things that get hard without it
You could hand-roll the open-orders example. The next three are where hand-rolling stops being reasonable.
1. Joining data across systems that know nothing about each other
Real applications have data in several places: orders in PostgreSQL, live location updates arriving over HTTP, session state in Redis. Answering one question across all of them normally means ETL, a sync job, or a pile of application-level joins that go stale.
Drasi lets a query span sources and declare the relationship itself. This is a synthetic join, built from matching property values, with no foreign key and no shared schema:
await drasi.addQuery(
'messages-with-location',
`MATCH
(m:Message)-[:FROM_USER]->(u:UserLocation)
RETURN
m.MessageId AS Id,
m.Message AS Message,
m.From AS Sender,
u.location AS Location,
u.status AS Status`,
['messages', 'location-tracker'],
'cypher',
[{
id: 'FROM_USER',
keys: [
{ label: 'Message', property: 'From' },
{ label: 'UserLocation', property: 'name' },
],
}],
); messages is a PostgreSQL table streaming via logical replication. location-tracker is an HTTP endpoint receiving JSON posts. They have never heard of one another. The FROM_USER relationship exists only inside the query, declared by joining Message.From to UserLocation.name.
The result set stays live across both: a new row in Postgres updates it, and so does a POST to the HTTP source. Nothing had to be copied into a warehouse first.
2. Reacting to things that don't happen
Some of the most valuable conditions in software are non-events: a heartbeat that stopped, an order that was placed but never picked up, a sensor that has gone quiet, a payment that has been pending too long. Nothing fires when nothing happens, so an event-driven system cannot see any of it. You end up writing a reaper job on a timer to scan for staleness, which is another polling loop with its own bugs.
A continuous query states the condition directly:
await drasi.addQuery(
'stuck-orders',
`MATCH
(o:Order)
WHERE
drasi.trueFor(o.status = 'pending', duration({ seconds: 30 }))
RETURN
o.id AS id,
o.customer AS customer`,
['orders'],
); drasi.trueFor() requires the condition to hold continuously for the given duration. The engine knows when o.status became pending, so it schedules its own re-evaluation for thirty seconds after that moment. Ship the order at twenty-nine seconds and that evaluation is cancelled, so the row never appears. Leave it sitting there and the row appears the instant the threshold is crossed, with no incoming data to trigger it. Your reaction fires because nothing happened.
Updating something unrelated on the order does not restart the clock, because the timer is anchored to when the condition became true rather than to the last write.
If your deadline is an absolute timestamp rather than an elapsed duration, drasi.trueLater(condition, instant) does the same job against a specific future moment.
Doing this by hand means a timer, a table of last-seen timestamps, and a plan for what happens after a restart.
3. Aggregates that stay correct
await drasi.addQuery(
'revenue-by-region',
`MATCH
(o:Order)
WHERE
o.status = 'complete'
RETURN
o.region AS Region,
count(o) AS Orders,
sum(o.total) - sum(o.discount) AS NetRevenue,
avg(o.total) AS AverageOrder,
max(o.total) AS LargestOrder`,
['orders'],
); A live GROUP BY with a formula on top. NetRevenue is derived from two separate running totals, and every value here is maintained incrementally: one completed order touches only its own region's row, and your reaction receives the before and after for that row.
It works in reverse too. Refund an order and it leaves the group, so that region's count, net revenue, average and maximum all correct themselves in a single update. Nothing rescans the table, and there is no cache to invalidate.
Where the data comes from
Two options, and you can mix them freely.
JavaScript sources are for data your application already has. Call addJsSource, then pushChange from wherever it makes sense: a webhook handler, a Kafka consumer, a WebSocket message, an ORM hook, a CDC stream you already run.
app.post('/webhook', async (req, res) => {
await drasi.pushChange('events', {
op: 'insert', id: req.body.id, labels: ['Event'], properties: req.body,
});
res.sendStatus(202);
}); Plugin sources connect to external systems directly. They are self-contained native plugins, the same ones the Drasi server uses, pulled from a registry at runtime:
await drasi.installPlugin('source/postgres', pluginsDir);
await drasi.loadPlugins(pluginsDir);
await drasi.addSource('postgres', 'messages', PG_CONFIG, true, {
kind: 'postgres', config: PG_CONFIG,
}); That gives you PostgreSQL change data capture in three lines. Because the catalogue is shared with the server, the same set is available to you: sources for PostgreSQL, MySQL, SQL Server, Oracle, HTTP and gRPC, and reactions for things like Server-Sent Events, so you can stream query changes straight to a browser.
Questions you're probably about to ask
Isn't this just RxJS with extra steps? Rx composes streams; it has no persistent state of its own. It cannot join today's event against a record it saw last week, keep an aggregate correct across restarts, or fire when nothing happens. Drasi maintains query state. That is the difference.
Why not PostgreSQL LISTEN/NOTIFY? Fine for one database, and useful. But it hands you a notification, not a query result. There are no cross-source joins, no aggregates and no absence detection, and it does nothing for data that isn't in Postgres.
Is this a replacement for Kafka or Flink? No. If you're processing millions of events per second across a fleet, use the fleet-scale tools. @drasi/lib targets the very large space below that line, where the problem is real but a cluster is absurd.
Do I have to learn a graph database? No. As above: the property graph is an internal model, not a storage engine you operate.
Try it
The fastest path is the Getting Started tutorial: a single-file console app that connects to a live PostgreSQL table and runs five continuous queries (change detection, a filter, an aggregation, a time-based query and a cross-source join), printing every change as it happens. It runs in a dev container, so you are one command away from a working system.
npm install @drasi/lib- Project:https://drasi.io
Then go and find your setInterval.
Microsoft