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 con el paquete mongodb.
• Un clúster de TeselaDB y un usuario con readWrite sobre la base shop.
• Las variables TESELADB_URI y WEBHOOK_URL, la URL del servicio que recibe los avisos.

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.

Siguientes pasos

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