Webhooks desde un change stream

Un change stream te da cada pedido nuevo en orden de commit. Si guardas su token después de cada aviso, el proceso puede pararse y seguir donde lo dejó, sin perder pedidos.

Lo que vas a construir

Un proceso de Node.js que vigila la colección orders y manda un POST a una URL por cada pedido insertado con status: "paid". Cada aviso lleva la cabecera idempotency-key con el _id del pedido.

Requisitos

• Node.js 18 o posterior.
• Docker y mongosh, para arrancar TeselaDB en tu máquina y mirar los datos.

Prepara el entorno

docker run -d --name teseladb -p 27017:27017 viuzi/teseladb:latest \
  start --dbpath /data --insecure --insecure-allow-remote --bind_ip 0.0.0.0
export TESELADB_URI="mongodb://127.0.0.1:27017/shop"
mongosh "$TESELADB_URI" --quiet --eval 'db.runCommand({ ping: 1 }).ok'
mkdir webhooks && cd webhooks && npm init -y && npm install mongodb
node -e "require('node:http').createServer((req, res) => { req.pipe(process.stdout); res.end() }).listen(8080)" &
export WEBHOOK_URL="http://127.0.0.1:8080/"

El contenedor arranca un nodo de TeselaDB sin TLS ni usuarios, que basta para seguir la guía en tu máquina. Cuando mongosh responde 1, el nodo ya acepta conexiones. TESELADB_URI apunta a ese nodo y a la base shop; en un clúster de producción la cadena lleva TLS y usuario, como cuenta Conectar tu aplicación. Las tres últimas líneas crean la carpeta del programa y arrancan en el puerto 8080 un receptor que imprime cada aviso, con su dirección en WEBHOOK_URL; en tu sistema pondrás la URL de tu servicio.

1. El proceso

Guárdalo como webhooks.mjs. El filtro del stream deja pasar solo las inserciones de pedidos pagados:

import { MongoClient } from "mongodb";

const client = new MongoClient(process.env.TESELADB_URI);
const db = client.db("shop");
const state = db.collection("stream_state");
const saved = await state.findOne({ _id: "order-webhooks" });

// New paid orders only, from where the last run stopped.
const stream = db
  .collection("orders")
  .watch(
    [{ $match: { operationType: "insert", "fullDocument.status": "paid" } }],
    saved ? { resumeAfter: saved.token } : {},
  );

for await (const change of stream) {
  const response = await fetch(process.env.WEBHOOK_URL, {
    method: "POST",
    headers: { "content-type": "application/json", "idempotency-key": String(change.documentKey._id) },
    body: JSON.stringify({ type: "order.paid", order: change.fullDocument }),
  });
  if (!response.ok) throw new Error(`webhook answered ${response.status}`);
  await state.updateOne({ _id: "order-webhooks" }, { $set: { token: change._id } }, { upsert: true });
}

2. Avisos que pueden repetirse

El proceso guarda el token después de que el servicio conteste. Si se para entre el aviso y el guardado, al volver repite ese aviso. El servicio que recibe los avisos usa la cabecera idempotency-key para no tratar dos veces el mismo pedido. Si el servicio contesta con un error, el proceso se para sin guardar el token y lo reintenta al volver a arrancar.

Probarlo

Arranca el proceso con node webhooks.mjs y, desde mongosh, inserta un pedido pagado y otro pendiente en la misma llamada, para ver que el filtro del stream solo deja pasar el primero de esos dos pedidos:

db.getSiblingDB("shop").orders.insertMany([
  { _id: 1, status: "paid", total: 30 },
  { _id: 2, status: "pending", total: 12 },
]);

El servicio recibe un solo aviso, el del pedido 1. Para el proceso, inserta otro pedido pagado y arráncalo de nuevo: el aviso de ese pedido llega al volver, porque el stream sigue desde el token guardado.

Limpiar

db.getSiblingDB("shop").stream_state.deleteOne({ _id: "order-webhooks" });

Para el proceso con Ctrl+C y ejecuta esta orden en mongosh para borrar el token guardado, así una prueba nueva empieza desde cero. Al final borra también el contenedor con la última orden: el nodo guarda los datos dentro, sin volumen, así que con él desaparece todo lo que escribiste durante la guía y el puerto 27017 queda libre.

kill %1
docker rm -f teseladb

Siguientes pasos

• Filtrar eventos, con los filtros que acepta watch.
• Invalidar una caché con change streams, con el mismo patrón para Redis.