# 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).