MongoDB change streams, typed.
Open a change stream on any collection. Basalt flattens documents into typed columns and keeps your views current.
Overview
How it works.
Basalt opens a change stream with fullDocument set to updateLookup. It snapshots each collection first, then resumes from the stored resume token. Nested fields are flattened into columns, and new fields are added as they appear.
Setup
Set it up in three steps.
01
Add the source
Pass a connection string for a replica set or Atlas cluster. The user needs read and changeStream on the collections you pick.
# stream users and sessions from Atlas basalt source add accounts --mongodb atlas-prod --collections users,sessions
✓ change stream opened · updateLookup
✓ snapshot 2.1M documents · resume token stored
02
Shape it with a view
Write plain SQL over the source. Basalt keeps the view current with every record and tracks lineage back to the topic, partition and offset.
CREATE MATERIALIZED VIEW revenue_by_min AS SELECT date_trunc('minute', ts) AS minute, currency, sum(amount) AS revenue, count(*) AS payments FROM payments WHERE status = 'captured' GROUP BY 1, 2;
03
Read it from anywhere
Every view is an endpoint the moment it exists. Call it over REST, connect with any Postgres client, or use the SDK from your app.
curl https://api.basalt.dev/v1/views/revenue_by_min?limit=5 \ -H "Authorization: Bearer $BASALT_TOKEN"
✓ 200 OK · 5 rows · served from eu-north-1 in 3.1 ms
Included
What you get.
01
Resume tokens
Tokens are stored with each write, so a restart picks up at the exact change.
02
Document flattening
Nested fields become typed columns. Arrays can be kept as JSON or unnested.
03
Atlas ready
Works with Atlas M10 and above and with self-managed replica sets.
04
Document pre-images
On MongoDB 6.0+, deletes and updates can carry the document before the change.
Related
