中文
devexamples

MongoDB Change Stream Typed Watch

React to collection changes with a typed MongoDB change stream, filtering operations, resuming by token after restart, and cleaning up on close.

TypescriptNodeData Fetching

Code

snippet
import { MongoClient, ChangeStreamDocument } from 'mongodb';

interface OrderDoc {
  _id: object;
  status: 'pending' | 'shipped' | 'delivered';
  total: number;
}

const client = new MongoClient(process.env.MONGO_URI ?? 'mongodb://127.0.0.1:27017');
await client.connect();

const orders = client.db('shop').collection<OrderDoc>('orders');

let resumeToken: string | null = null;

const stream = orders.watch(
  [{ $match: { operationType: { $in: ['insert', 'update'] } } }],
  { fullDocument: 'updateLookup' },
);

stream.on('change', (event: ChangeStreamDocument<OrderDoc>) => {
  if (event.operationType === 'insert' || event.operationType === 'update') {
    const doc = event.fullDocument;
    console.log('order changed:', doc?._id, 'status', doc?.status);
  }
  // 断点续传：记录最新令牌，重启后从该位置继续
  resumeToken = JSON.stringify(stream.resumeToken);
});

stream.on('error', async (err) => {
  console.error('stream failed:', err.message);
  await stream.close().catch(() => {});
});

await orders.insertOne({ status: 'pending', total: 1200 } as OrderDoc);

setTimeout(() => stream.close().then(() => client.close()), 2000);

Related Examples