# Production change stream consumer
```js
const pipeline = [
{ $match: { operationType: { $in: ["insert", "update", "replace"] } } }
];
async function runConsumer(resumeToken) {
const opts = {
fullDocument: "updateLookup", // updates carry the current full doc
...(resumeToken ? { resumeAfter: resumeToken } : {}),
};
const stream = collection.watch(pipeline, opts);
for await (const event of stream) {
await handleEvent(event);
await saveResumeToken(event._id); // persist AFTER handling
}
}
```
## Rules
- Filter in the pipeline (`$match` on `operationType`, namespace) so the server only sends what you need. Unfiltered streams on busy collections drown consumers.
- `fullDocument: "updateLookup"` gives you the post-update document for updates. Without it you only get the delta.
- Persist the resume token (`event._id`) after each successfully handled event, in the same transaction as your side effects if you need exactly-once-ish semantics. On restart, pass it as `resumeAfter`.
- Handle `invalidate` (collection dropped/renamed): the stream dies and `resumeAfter` will not save you. Alert and re-seed.
- Handle resume failure (token older than the oplog window): fall back to `startAtOperationTime` at now plus a backfill, not an infinite retry loop.
- Atlas runs replica sets, so change streams work on every tier including Free. No special cluster config needed.
## Verify
Kill and restart the consumer mid-stream: it resumes from the token with no missed or double-handled events (check your idempotency on the handler side).