Invalidate a cache with change streams

If you keep products in a cache, you have to delete each entry when the product changes. A change stream tells you about every change in commit order, and its resume token keeps you from losing events if your process goes down.

The process

import { MongoClient } from "mongodb";
import { createClient } from "redis";

const mongo = new MongoClient(process.env.TESELADB_URI);
const redis = createClient({ url: process.env.REDIS_URL });
await Promise.all([mongo.connect(), redis.connect()]);

const db = mongo.db("shop");
const state = db.collection("stream_state");
const saved = await state.findOne({ _id: "product-cache" });

const stream = db
  .collection("products")
  .watch(
    [{ $match: { operationType: { $in: ["update", "replace", "delete"] } } }],
    saved ? { resumeAfter: saved.token } : {},
  );

for await (const change of stream) {
  await redis.del(`product:${change.documentKey._id}`);
  await state.updateOne({ _id: "product-cache" }, { $set: { token: change._id } }, { upsert: true });
}

Why it works

• The stream delivers events in commit order and without duplicates, even if the ranges of products split or change leader.
• After deleting the entry, the process saves the event's token. On restart, resumeAfter carries on right after the last event handled.
• If the process goes down between the delete and saving the token, it repeats that event on return: deleting a key twice does no harm.

How long you can be stopped

You can resume while the event is still in the cluster's history: 24 hours by default. If your process can be stopped for longer, raise teselaHistoryRetention:

db.adminCommand({ setClusterParameter: { teselaHistoryRetention: { seconds: 259200 } } });

If the token is older than the history, the stream fails to open. In that case empty the whole cache and start a new stream without resumeAfter.