Tracing Which Internal Consumer Read Each Webhook Event (One Registration, One Queue)

#webhooks #queue #node #architecture

Use one webhook registration, land the payload in a durable queue, and make every internal consumer acknowledge its own copy. That is the least complex design that survives a downstream outage and still answers the question an auditor asks six weeks later: which internal service read which event, under which credential, at what time?

Fan-out is the easy part.

The flow fits in one breath. The platform POSTs to a single HTTPS endpoint you registered once; the receiver verifies the signature, stores the raw body keyed by the provider's event id, and writes one pending delivery row per registered internal consumer; each consumer claims its own rows, does its work, and acknowledges them; that acknowledgement is what lands in the access log. Nothing downstream ever talks to the platform directly, and nothing in the path deletes the evidence that a read happened.

I build B2B SaaS features where "who touched this account event" is a question finance and security both ask, so the design axis here isn't throughput. It's attribution.

How do you fan out one webhook registration to many internal consumers through a queue?

Three mechanisms cover almost every case, and they differ mainly in how much the queue remembers about each consumer.

Mechanism	Per-consumer acknowledgement	Audit granularity	Main limit
One queue per consumer	Native ack per queue	Per consumer, per message	Registration list lives in broker config; adding a consumer means new infrastructure
Stream with consumer groups	Ack moves a per-group cursor	Per group, per entry	A group's cursor says how far it got, not which principal read the row
Outbox table in your own database	Row state transition you control	Per consumer, per event, with your own columns	You own the polling loop, the locking, and the retention

The outbox table is the boring one, and for an auditability-first system it's usually the right boring. Because acknowledgement is a row update rather than a broker-internal counter, you can hang whatever the audit needs off it — the consumer name, the credential that did the work, the attempt count, the timestamp. A broker ack is a fast, opaque "done". A row update is a fact you can query a year later.

Everything below assumes at-least-once delivery, because that's what you get from every real platform and every real queue. Consumers must be idempotent. There's no negotiating with that.

A Node.js receiver that records every internal read

The receiver does four things and nothing else: verify, persist, fan out, return 204. No consumer logic runs inside the request.

import { createServer } from "node:http";
import { createHmac, timingSafeEqual } from "node:crypto";
import { Pool } from "pg";

const pool = new Pool({ connectionString: process.env.DATABASE_URL });
const secret = process.env.WEBHOOK_SECRET as string;
const CONSUMERS = ["billing-sync", "seat-provisioner", "usage-ledger"];

function verified(raw: Buffer, header: string): boolean {
  const expected = createHmac("sha256", secret).update(raw).digest();
  const got = Buffer.from(header.replace(/^sha256=/, ""), "hex");
  return got.length === expected.length && timingSafeEqual(got, expected);
}

createServer((req, res) => {
  if (req.method !== "POST" || req.url !== "/hooks/platform") {
    res.writeHead(404).end();
    return;
  }
  const chunks: Buffer[] = [];
  req.on("data", (c: Buffer) => chunks.push(c));
  req.on("end", async () => {
    const raw = Buffer.concat(chunks);
    if (!verified(raw, String(req.headers["x-signature"] ?? ""))) {
      res.writeHead(401).end();
      return;
    }
    const event = JSON.parse(raw.toString("utf8"));
    const db = await pool.connect();
    try {
      await db.query("begin");
      // The platform's own id is the idempotency key: a redelivered POST inserts nothing.
      const ins = await db.query(
        `insert into events (event_id, kind, payload, received_at)
         values ($1, $2, $3, now())
         on conflict (event_id) do nothing
         returning id`,
        [event.id, event.type, raw.toString("utf8")],
      );
      if (ins.rows.length) {
        await db.query(
          `insert into deliveries (event_ref, consumer) select $1, unnest($2::text[])`,
          [ins.rows[0].id, CONSUMERS],
        );
      }
      await db.query("commit");
      res.writeHead(204).end();
    } catch (err) {
      await db.query("rollback");
      console.error("ingest rejected", err);
      res.writeHead(503).end();   // unacknowledged: the platform retries on its own schedule
    } finally {
      db.release();
    }
  });
}).listen(8080);

Each consumer then drains its own rows. for update of d skip locked lets several workers of the same consumer run concurrently without handing the same event to two of them, which matters once you scale a slow consumer horizontally.

import { Pool } from "pg";
import { handle } from "./billing-sync.ts";

const pool = new Pool({ connectionString: process.env.DATABASE_URL });
const consumer = "billing-sync";
const principal = process.env.CONSUMER_PRINCIPAL as string;  // this consumer's own credential

export async function drain(batch = 20): Promise<number> {
  const db = await pool.connect();
  try {
    await db.query("begin");
    const { rows } = await db.query(
      `select d.id, e.event_id, e.payload
         from deliveries d
         join events e on e.id = d.event_ref
        where d.consumer = $1 and d.acked_at is null
        order by d.id
        limit $2
        for update of d skip locked`,
      [consumer, batch],
    );
    for (const row of rows) {
      await handle(JSON.parse(row.payload));
      await db.query(
        `update deliveries set acked_at = now(), attempts = attempts + 1 where id = $1`,
        [row.id],
      );
      await db.query(
        `insert into access_log (event_id, consumer, principal, read_at)
         values ($1, $2, $3, now())`,
        [row.event_id, consumer, principal],
      );
    }
    await db.query("commit");
    return rows.length;
  } catch (err) {
    await db.query("rollback");
    console.error(`${consumer} batch rolled back`, err);
    return 0;
  } finally {
    db.release();
  }
}

Two details in there carry the whole audit story. The acknowledgement and the access-log insert share a transaction with the work, so a crash mid-batch leaves the row unacknowledged and the log honest. And principal comes from that consumer's own credential rather than one shared internal token — if three services authenticate as the same identity, your access log records a shrug.

Acknowledgement is the audit record

Most implementations treat ack as garbage collection: processed, so delete. Then a customer disputes a seat change, and the only thing you can prove is that a row is gone.

Keep the transition instead of the deletion. A delivery row that moves from null to an acked_at timestamp, with attempts and principal alongside, gives you three separate answers — was the event received, was it claimed, was it completed — and those are genuinely different questions during an incident review. Cursor-based streams collapse the second and third into one number per group.

Secrets are the other half of attribution, and they're easy to get wrong in exactly this shape of system. The signing secret for the registration, the per-consumer credentials, and the database password have different rotation needs and different blast radii; OWASP's secrets management guidance is blunt about not sharing one credential across components precisely because it destroys traceability. Load them from the environment or a secrets store at boot, rotate the webhook secret by accepting two valid secrets during an overlap window, and keep the rotation event itself in the same audit stream.

Verify before you log. An unverified POST should leave no trace in events at all, or your access log becomes a place where anyone on the internet can write rows.

What survives an outage, and what it costs to keep

Outages come in two flavors here, and only one of them is yours to absorb. If a consumer is down, its delivery rows pile up, the other consumers keep going, and lag per consumer is a single count(*) where acked_at is null per name. That's the payoff for per-consumer acknowledgement: one broken service doesn't block the other two, and you don't need the platform to redeliver anything.

If your receiver is down, the platform retries on whatever schedule it documents — typically an exponential sequence over some hours — and then gives up. Never treat that window as your safety net. Storing raw bodies means replay is yours: re-insert delivery rows for one consumer and one event range, and the rest of the system doesn't notice. Replaying from your own store also keeps the audit trail coherent, because the replay is itself an event you can log.

Retention is where the cost conversation lands, and as a solo founder I care: raw JSON payloads at a few KB each are cheap until a chatty platform sends millions a month, at which point keeping full bodies for a year quietly becomes your third-largest storage line. A reasonable split is 30 days of full payloads for debugging and replay, then trim to the hash plus the audit columns, which are the part compliance actually wants. I'm not sure there's a universal number — it depends on how long your disputes take to arrive.

When a single shared queue is the wrong call

If you have one internal consumer and no audit requirement, skip all of this. Verify the signature, do the work, return 204. A queue you added for a consumer you don't have yet is just an extra thing to page you about.

Postgres as the fan-out substrate has a real ceiling. Polling costs you latency floor and wasted queries, and past roughly a few thousand events per second the locking and vacuum pressure stop being free — that's where a broker with consumer groups earns its operational weight, with the trade-off that per-principal attribution moves into your application logs instead of the queue. If you need months of replay across many independent readers, a log-structured broker is the better fit and a database table isn't; stick with your broker's retention and put the audit rows beside it rather than inside it. Hosted webhook gateways solve the receiver-reliability problem well, though the delivery record then lives in a third party's system, which may or may not satisfy the auditor who asks for it.

The catch with the outbox approach is that you own code nobody else maintains: the claim query, the backoff, the DLQ semantics when a consumer exceeds its attempt budget.

Before you call it done, check a short list against a staging replay. Every consumer has its own credential and its own row in the access log. The receiver answers in under a second, because the platform's timeout is short and slow verification looks like an outage from the outside. Unacknowledged counts are graphed per consumer with an alert on age rather than depth, since ten minutes of lag on seat provisioning and ten minutes on a usage ledger are not the same incident. A poisoned event lands in a dead-letter state with its attempt history intact instead of being retried forever. Secret rotation is rehearsed, not theorized. And replay for one consumer over one hour is a command someone has actually run, with the resulting access-log rows eyeballed once — because the first time you try it during a real incident is the worst possible time to learn that it double-charges somebody.

DEV Community