#Change Stream Sync

With an empty checkpoint store, startup first persists a versioned operation-time envelope containing a canonical EJSON BSON Timestamp, then opens the Change Stream. Custom stores must round-trip this opaque JSON value. Existing raw resume tokens remain supported; successful events continue to save raw tokens. Strict persistence failures, malformed envelopes and expired history stop progress without clearing the checkpoint. Explicit best-effort settings still permit non-durable progress. Target names must be exactly unique, with idempotency enabled or disabled.

Before downgrading, pause sync and preserve its store. Older readers cannot use the initial envelope as a resume token. A successfully persisted raw event token restores format compatibility, subject to the older version's behavior and MongoDB history retention. Do not clear the envelope to make downgrade start from now; an intentional resync requires a separate data reconciliation decision.

Per-target idempotency keys are opaque JSON tuples: ["v2", keyPrefix, targetName, rawKey]. Colons, quotes and Unicode in these components cannot merge distinct identities. keyBuilder(event, targetName) still receives the original inputs and returns the raw key; null, undefined or an empty string disables that event key. TTL and markMode retain their semantics.

Upgrade with writers stopped and queued work drained. Preserve the checkpoint, v1 and v2 marker stores, and a recovery snapshot. Do not run mixed key-format writers. V1 concatenated keys are ambiguous and are not trusted as unconditional v2 duplicate markers: without a finite, provable mapping, use target-owned business idempotence and allow replay. A generic get/set/del store cannot enumerate or reverse these legacy keys automatically.

For rollback, stop and drain again, preserve both marker generations, and verify a compatible raw resume token plus the target recovery point. The initial-checkpoint downgrade restriction above still applies. Do not clear either store or restart from now as a recovery shortcut. Failed active or staged target closes retain cleanup ownership; later stop() retries them and start() must finish cleanup before opening replacements. Other targets still receive a close attempt.

#Overview

Change Stream sync uses MongoDB's native Change Stream mechanism to dispatch accepted data-change events to one or more targets. It is useful for backup databases, denormalized projections, cache invalidation callbacks, and other asynchronous CDC workflows.

#Core Features

• ✅ Asynchronous CDC: Based on MongoDB Change Stream; latency depends on MongoDB, network, and target workload
• ✅ Decoupled design: Primary database writes do not wait for target apply callbacks
• ✅ Resume breakpoint: Resume Token is saved after all matching targets succeed, so restarts can resume from the latest persisted token
• ✅ Multi-target support: Sync to multiple backup databases at the same time
• ✅ Data Filtering: Custom filtering logic
• ✅ Data conversion: Support desensitization and field conversion
• ✅ Observable lifecycle: Sync stats expose running state and error counters for monitoring/restart workflows
• ✅ Health Check: Target health-check configuration for managed targets

#⚠️ Prerequisites

#Must satisfy

1. MongoDB Replica Set 🔴

   # Check whether it is a Replica Set
   rs.status()

2. MongoDB version >= 4.0 🔴

3. User Permissions 🔴

   // Requires changeStream permission
   {
       resource: { db: "dbName", collection: "" },
       actions: ["changeStream", "find"]
   }

#🚀 Quick Start

#Basic configuration

import MonSQLize from 'monsqlize';

const msq = new MonSQLize({
    type: 'mongodb',
    config: {
        uri: 'mongodb://localhost:27017/main',
        replicaSet: 'rs0'  // Required
    },

    // Synchronization configuration
    sync: {
        enabled: true,
        targets: [
            {
                name: 'backup-main',
                uri: 'mongodb://backup:27017/backup',
                collections: ['users', 'orders']
            }
        ]
    }
});

await msq.connect();

// Normal use, automatic synchronization
await msq.collection('users').insertOne({ name: 'Alice' });
// Automatically sync to backup-main

await msq.close();

#📖 Configuration options

#sync configuration

transform is a manager-level conversion hook. It runs once for each accepted change event before the event is delivered to all matching targets, so all targets receive the same transformed document. Delete events usually do not include fullDocument; custom targets should handle an undefined document and use event.documentKey for delete handling.

#targets[].Configuration

#resumeToken configuration

Resume token persistence is strict by default. File storage writes to a same-directory temporary file, fsyncs it, keeps the previous token as <path>.bak, and then atomically renames the temporary file into place. Startup also validates the stored token: a corrupted token fails fast when strictLoad is true, instead of silently starting without a resume token. After all eligible targets apply an event successfully, monSQLize saves the event resume token; syncedCount advances only after that save succeeds. If token persistence fails after the configured retries, or any eligible target fails to apply the event, the manager records the error, closes the live stream, marks isRunning: false, and does not process later queued events. This is an at-least-once contract, not exactly-once: a crash after target apply and before token save can replay the same event. Set strictSave: false and strictLoad: false only for legacy best-effort token-storage behavior, where a restart may replay already-applied events or start without the previously stored token. Built-in MongoDB targets are idempotent (replaceOne(..., { upsert: true }) / deleteOne()); custom apply targets should still deduplicate by change event _id.

#idempotency configuration

sync.idempotency is optional and disabled by default. When enabled, the manager builds a per-target idempotency key from the change event _id unless keyBuilder is provided. If the key is already present, that target is skipped and the event can still advance its resume token after all eligible targets are accounted for. Use a durable store for restart protection; the built-in memory fallback only protects repeated delivery in the same process. markMode: 'success' records after apply resolves. markMode: 'start' records before apply; it narrows unknown-success duplicate risk, but a failure after the marker is written can make runtime replay skip that target, so use it only when the target has its own durable idempotency and recovery path.

#💡 Usage example

#Example 1: Multiple backup targets

{
    sync: {
        enabled: true,
        targets: [
            {
                name: 'backup-asia',
                uri: 'mongodb://asia:27017/backup',
                collections: ['*']
            },
            {
                name: 'backup-us',
                uri: 'mongodb://us:27017/backup',
                collections: ['*']
            }
        ]
    }
}

#Example 2: Data filtering

{
    sync: {
        enabled: true,
        targets: [...],

        // Only sync active users
        filter: (event) => {
            if (event.ns?.coll === 'users') {
                return event.fullDocument?.status === 'active';
            }
            return true;
        }
    }
}

#Example 3: Data desensitization

{
    sync: {
        enabled: true,
        targets: [...],

        // Remove sensitive fields
        transform: (doc) => {
            delete doc.password;
            delete doc.ssn;
            return doc;
        }
    }
}

#Example 4: Redis Resume Token

const Redis = require('ioredis');
const redis = new Redis();

{
    sync: {
        enabled: true,
        targets: [...],
        resumeToken: {
            storage: 'redis',
            redis: redis,
            strictSave: true,
            saveRetries: 3,
            saveRetryDelayMs: 100
        }
    }
}

#📊 Performance impact

#🔧 API

#Get statistics

const stats = msq.getSyncStats();
console.log(stats);
// {
//   isRunning: true,
//   eventCount: 1234,
//   syncedCount: 1230,
//   errorCount: 4,
//   startTime: 2026-01-17T...,
//   lastEventTime: 2026-01-17T...,
//   lastError: null,
//   tokenSaveErrorCount: 0,
//   lastTokenSaveError: null,
//   targets: [...]
// }

#Manually stop synchronization

await msq.stopSync();

#Manually start synchronization

await msq.startSync();

#❓ FAQ

#Q1: Prompt "Change Stream is not available"

Reason: MongoDB is not a Replica Set

Solution:

# 1. Check the topology
rs.status()

# 2. If it is a single node, switch to Replica Set
rs.initiate()

#Q2: Is there a delay in synchronization?

Cause: Network delay, backup database performance

Solution:

1. Check network delay: ping backup-host
2. Check backup database performance: db.serverStatus()
3. Reduce the number of synchronized collections

#Q3: What should I do if the Resume Token is lost?

Impact: After restarting, synchronization starts from the current time and intermediate data is lost.

Solution:

1. Use Redis to store Resume Token
2. Back up the Resume Token file regularly
3. Manually synchronize all data once

#Q4: How to deal with synchronization failure?

Automatic processing:

• Targets that already applied the event are not rolled back, but any target failure stops the manager before the resume token advances
• A resume-token save failure stops the sync manager before the token can advance
• Change Stream driver errors and unexpected stream closes are logged and reflected in stats; monitor isRunning, errorCount, and lastError, then restart the manager or runtime from your supervisor when needed

Manual processing:

// View error statistics
const stats = msq.getSyncStats();
console.log(stats.errorCount);
console.log(stats.lastError);
console.log(stats.targets[0].lastError);

#🛡️ Best Practices

#1. Production environment configuration

{
    sync: {
        enabled: true,
        targets: [
            {
                name: 'backup-main',
                uri: 'mongodb://backup:27017/backup',
                collections: ['*'],
                healthCheck: {
                    enabled: true,
                    interval: 30000,  //30 seconds
                    timeout: 5000,
                    retries: 3
                }
            }
        ],
        resumeToken: {
            storage: 'redis',  //Use Redis
            redis: redisInstance
        }
    }
}

#2. Monitoring and Alarming

//Check statistics regularly
setInterval(() => {
    const stats = msq.getSyncStats();

    if (stats.errorCount > 100) {
        //Send alert
        sendAlert('Too many sync errors');
    }

    if (!stats.isRunning) {
        //Send alert
        sendAlert('Sync stopped');
    }
}, 60000);

#3. Graceful shutdown

process.on('SIGTERM', async () => {
    console.log('Received SIGTERM, closed gracefully...');
    await msq.close();
    process.exit(0);
});

#More resources

• Sample code
• MongoDB Change Streams official documentation
• Multi-pool configuration documentation