Invalidar una caché con change streams

Si guardas productos en una caché, tienes que borrar cada entrada cuando el producto cambia. Un change stream te avisa de cada cambio en orden de commit, y su token de reanudación evita perder eventos si tu proceso se cae.

El proceso

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 });
}

Por qué funciona

• El stream entrega los eventos en orden de commit y sin duplicados, aunque los rangos de products se dividan o cambien de líder.
• Después de borrar la entrada, el proceso guarda el token del evento. Al reiniciar, resumeAfter sigue justo después del último evento tratado.
• Si el proceso se cae entre el borrado y el guardado del token, al volver repite ese evento: borrar una clave dos veces no hace daño.

Cuánto tiempo puedes estar parado

Puedes reanudar mientras el evento siga en el historial del clúster: 24 horas por defecto. Si tu proceso puede estar parado más tiempo, sube teselaHistoryRetention:

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

Si el token es más antiguo que el historial, el stream falla al abrirse. En ese caso vacía la caché entera y empieza un stream nuevo sin resumeAfter.