Real-Time Data with ormDB Change Streams

Built-In Change Data Capture

ormDB is a relational database engine written in Rust that replaces PostgreSQL, MySQL, and SQLite. Change streams are a native capability of the engine, providing real-time change data capture (CDC) without external tools, plugins, or replication configuration.

When a record is created, updated, or deleted, ormDB emits a change event. Your application subscribes to these events at the entity level and receives them in real time. Each event includes a version token that enables reliable resume after disconnection.

Opening a Change Stream

Subscribe to changes on an entity:

import { OrmDB } from '@ormdb/client'

const db = new OrmDB('ormdb://localhost:5555/mydb')

const stream = db.changeStream('Post', {
  // Optional: filter which changes to receive
  where: { status: 'published' },
  // Optional: select specific fields
  fields: ['id', 'title', 'updated_at'],
})

stream.on('change', (event) => {
  console.log(event.type)       // 'insert' | 'update' | 'delete'
  console.log(event.entity)     // the full or partial entity state
  console.log(event.versionToken) // opaque cursor for resume
})

stream.on('error', (err) => {
  console.error('Stream error:', err)
})

Version Tokens and Reliable Resume

Every change event carries a version token. This token is an opaque cursor that represents the exact position in the change stream. Store the latest version token and pass it when reconnecting:

// Resume from where you left off
const stream = db.changeStream('Post', {
  resumeAfter: savedVersionToken,
})

ormDB guarantees that no changes are missed between the saved token and the current state. Events are delivered in commit order. If multiple changes occur within a single transaction, they appear as a batch with the same transaction ID.

Change Event Structure

Each change event contains:

{
  type: 'insert' | 'update' | 'delete',
  entity: 'Post',
  record: {
    id: 'post_abc123',
    title: 'Updated Title',
    status: 'published',
    updated_at: '2026-04-13T10:30:00Z',
  },
  previousRecord: {  // present on update and delete
    id: 'post_abc123',
    title: 'Original Title',
    status: 'draft',
    updated_at: '2026-04-12T08:00:00Z',
  },
  versionToken: 'vt_a1b2c3d4e5',
  transactionId: 'tx_f6g7h8i9j0',
  timestamp: '2026-04-13T10:30:00.123Z',
}

For updates, both the current and previous record state are included, enabling your application to react to specific field changes without querying the database.

Use Cases

Real-time UIs: Push database changes to connected clients via WebSockets. When a post is updated, subscribers see the change immediately.

Cache invalidation: Invalidate or update cached entities the moment they change in the database, without polling or TTLs.

Audit logs: Write every change to an append-only audit store with full before/after snapshots. Combine with row-level security for comprehensive data governance.

Cross-service sync: Propagate changes from ormDB to search indexes, analytics pipelines, or external services without building a separate ETL process. This is especially valuable in microservices architectures.

Event sourcing: Use change streams as the event log for event-driven architectures, with version tokens providing exactly-once processing semantics.

Filtering and Field Selection

Reduce bandwidth by subscribing only to the changes you need:

// Only receive changes where amount > 1000
const highValueOrders = db.changeStream('Order', {
  where: { amount: { gt: 1000 } },
  fields: ['id', 'amount', 'customer_id', 'status'],
})

Change streams are a zero-configuration feature. Unlike PostgreSQL’s logical replication, there is no replication slot to create, no publication to define, and no subscriber to register. Open a stream and start receiving changes.