Webhooks from a change stream

A change stream gives you each new order in commit order. If you save its token after each notification, the process can stop and carry on where it left off, without losing orders.

What you will build

A Node.js process that watches the orders collection and sends a POST to a URL for each order inserted with status: "paid". Each notification carries the idempotency-key header with the order's _id.

Requirements

• Node.js 18 or later with the mongodb package.
• A TeselaDB cluster and a user with readWrite on the shop database.
• The TESELADB_URI and WEBHOOK_URL variables, the URL of the service that receives the notifications.

1. The process

Save it as webhooks.mjs. The stream's filter lets through only the inserts of paid orders:

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. Notifications that can repeat

The process saves the token after the service answers. If it stops between the notification and the save, it repeats that notification when it comes back. The receiving service uses the idempotency-key header so it does not handle the same order twice. If the service answers with an error, the process stops without saving the token and retries on the next start.

Try it

Start the process with node webhooks.mjs and, from mongosh, insert one paid order and one pending order in the same call, so you can see that the filter of the stream lets through only the first of the two:

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

The service gets a single notification, the one for order 1. Stop the process, insert another paid order and start it again: that order's notification arrives on the restart, because the stream continues from the saved token.

Clean up

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

Stop the process with Ctrl+C and run this command in mongosh to delete the saved token, so a new test starts from scratch.

Next steps

• Filter events, with the filters watch accepts.
• Invalidate a cache with change streams, with the same pattern for Redis.