Atlas: build a production change stream consumer

TL;DR: Change streams power event-driven pipelines. The production pattern is a filtered pipeline, updateLookup, persisted resume tokens, and clean resume handling. 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).

Production change stream consumer

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

When to use this

• This covers exactly what the title says: Atlas.
• You are setting this up for the first time, or auditing an existing setup.
• You want the key gotchas in one place before you start.

When not to use this

• You are doing a different workflow with Atlas; these steps are specific to the title above.
• You need the full reference docs; this is the short path, not the manual.

Compatibility

• Not pinned to a specific version; follows current Atlas behavior.

Change stream options (pipeline filtering, fullDocument updateLookup, resumeAfter, invalidate handling) per https://www.mongodb.com/docs/manual/changeStreams/, verified 2026-09-26.