The Complete Production SOP: How to Deploy and Run MongoDB Change Streams & Real-Time Sync in Autonomous AI Agents

---
name: "mongodb-change-streams-reactive"
description: "Stream real-time insert, update, and delete mutations from MongoDB replica set oplogs."
metadata:
  author: "MongoDB"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "listen to mongodb change streams"
  - "resume mongodb change stream"
  - "stream oplog mutations"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

# MongoDB Change Streams & Real-Time Sync Rules
1. Always verify the MongoDB instance is a replica set or sharded cluster before opening watch().
2. Set fullDocument: 'updateLookup' on collection.watch() to capture the current state for downstream workers.
3. Cache changeStream resume token (`changeEvent._id`) to persistent storage atomically upon message commit.
4. Initialize reconnect streams using `resumeAfter` or `startAfter` when recovering from closed cursors.

import { MongoClient, ChangeStreamDocument, ResumeToken } from 'mongodb';
import Redis from 'ioredis';

const redis = new Redis(process.env.REDIS_URL || 'redis://localhost:6379');
const RESUME_TOKEN_KEY = 'mongo:change_stream:resume_token:orders';

export async function initOrderStream(client: MongoClient) {
  const db = client.db('production');
  const collection = db.collection('orders');
  
  const savedToken = await redis.get(RESUME_TOKEN_KEY);
  const resumeOptions = savedToken ? { resumeAfter: JSON.parse(savedToken) } : {};

  const pipeline = [
    { $match: { operationType: { $in: ['insert', 'update', 'replace'] } } }
  ];

  const stream = collection.watch(pipeline, {
    fullDocument: 'updateLookup',
    ...resumeOptions
  });

  stream.on('change', async (change: ChangeStreamDocument) => {
    await processOrderEvent(change);
    if (change._id) {
      await redis.set(RESUME_TOKEN_KEY, JSON.stringify(change._id));
    }
  });

  return stream;
}

async function processOrderEvent(change: ChangeStreamDocument) {
  // Downstream sync processing logic
}

import { describe, it, expect, vi } from 'vitest';
import { initOrderStream } from '../src/stream/change-stream-client';

describe('MongoDB Change Stream Initialization', () => {
  it('configures updateLookup and match filter correctly', async () => {
    const mockWatch = vi.fn().mockReturnValue({ on: vi.fn() });
    const mockDb = {
      collection: vi.fn().mockReturnValue({ watch: mockWatch })
    };
    const mockClient = { db: vi.fn().mockReturnValue(mockDb) } as any;

    await initOrderStream(mockClient);
    expect(mockWatch).toHaveBeenCalledWith(
      expect.arrayContaining([
        expect.objectContaining({ $match: expect.any(Object) })
      ]),
      expect.objectContaining({ fullDocument: 'updateLookup' })
    );
  });
});