Nightly pending-webhook sweep: one cron tick, batched publish, one worker

A scheduled job that scans for pending webhooks and enqueues them in batches, with the Node worker that drains the backlog and the delay behaviour to watch out for.

The pattern that holds up at scale has three separate pieces: a schedule that decides when, a publish step that records what, and a worker that owns how fast. Squash any two together and the third one breaks first. On Infrai the middle piece is POST /v1/queue/publish_batch, which takes an array of messages in one round trip and hands the pacing problem to a consumer you control.

Concretely: a nightly job selects rows from your pending_webhooks table, publishes them 25 at a time, and exits in a few seconds. Nothing is sent yet. Delivery is the worker’s business, and it can run for an hour without anyone’s request timer caring.

Publishing is not delivering

That distinction is the whole design, and it’s what a plain for loop over pending rows inside the cron process gets wrong.

The third row isn’t wrong — it’s just a different trade-off. Per-event publishing gives lower latency and a smoother spend curve; the nightly sweep gives you one place to look when something didn’t go out.

Selecting the batch

Keyset pagination, not OFFSET. A pending table that grows during the sweep will happily serve you the same rows twice under OFFSET.

SELECT id, endpoint_url, event_type, payload
FROM pending_webhooks
WHERE status = 'pending'
  AND id > $1
  AND created_at < $2
ORDER BY id
LIMIT 500;

The created_at < $2 bound matters more than it looks. Pin it to the moment the sweep started and a re-run picks up exactly the same set — rows created mid-sweep wait for tomorrow instead of being half-processed twice.

Enqueue 25 at a time

export INFRAI_API_KEY="your_infrai_api_key"

curl -sS -X POST "https://api.infrai.cc/v1/queue/publish_batch" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{"queue":"webhooks-nightly","messages":[{"payload":{"webhook_id":8801,"event":"invoice.finalized"}},{"payload":{"webhook_id":8802,"event":"invoice.finalized"}}]}'

{
  "ok": true,
  "data": {
    "items": [
      { "message_id": "qmsg_JrBUYayJG3sgzSqSR3BvOmKb", "queue": "webhooks-nightly", "status": "available", "delivery_count": 0, "available_at": "2026-07-26T00:34:30Z", "priority": 0 },
      { "message_id": "qmsg_RTEy0gSCY80i2OredgxUZgwG", "queue": "webhooks-nightly", "status": "available", "delivery_count": 0, "available_at": "2026-07-26T00:34:30Z", "priority": 0 }
    ],
    "next_cursor": null
  }
}

Each element wraps its own payload object — a bare object in the array gets you INVALID_ARGUMENT: messages[0].payload must be an object, which is a five-minute mistake to make and a two-second one to fix. Messages cap at 256 KB, so send identifiers and let the worker fetch the body.

The sweep itself, in Node 22:

import process from "node:process";
import pg from "pg";

const KEY = process.env.INFRAI_API_KEY;
const DSN = process.env.DATABASE_URL;
if (!KEY || !DSN) throw new Error("set INFRAI_API_KEY and DATABASE_URL");

const QUEUE = "webhooks-nightly";
const CHUNK = 25;

async function api(path, payload) {
  const res = await fetch(`https://api.infrai.cc${path}`, {
    method: "POST",
    headers: { Authorization: `Bearer ${KEY}`, "Content-Type": "application/json" },
    body: JSON.stringify(payload),
  });
  const { ok, data, error } = await res.json();
  if (!ok) throw new Error(`${path}: ${error.code} ${error.message}`);
  return data;
}

const db = new pg.Client({ connectionString: DSN });
await db.connect();
const startedAt = new Date();
let lastId = 0;
let published = 0;

try {
  for (;;) {
    const { rows } = await db.query(
      "SELECT id, endpoint_url, event_type FROM pending_webhooks WHERE status = 'pending' AND id > $1 AND created_at < $2 ORDER BY id LIMIT 500",
      [lastId, startedAt],
    );
    if (rows.length === 0) break;
    for (let i = 0; i < rows.length; i += CHUNK) {
      const slice = rows.slice(i, i + CHUNK);
      const { items } = await api("/v1/queue/publish_batch", {
        queue: QUEUE,
        messages: slice.map((r) => ({ payload: { webhook_id: r.id, url: r.endpoint_url, event: r.event_type } })),
      });
      published += items.length;
    }
    lastId = rows[rows.length - 1].id;
  }
  console.log(`swept ${published} pending webhooks in ${Date.now() - startedAt.getTime()}ms`);
} finally {
  await db.end();
}

The worker owns the rate

It reads ten at a time, delivers, and acks. Nothing about it knows the batch came from a nightly sweep — which is the point, because tomorrow you’ll want to feed the same queue from a live event stream.

import process from "node:process";
import { setTimeout as sleep } from "node:timers/promises";

const KEY = process.env.INFRAI_API_KEY;
if (!KEY) throw new Error("INFRAI_API_KEY is not set");
const H = { Authorization: `Bearer ${KEY}`, "Content-Type": "application/json" };

async function api(path, payload) {
  const res = await fetch(`https://api.infrai.cc${path}`, { method: "POST", headers: H, body: JSON.stringify(payload) });
  const { ok, data, error } = await res.json();
  if (!ok) throw new Error(`${path}: ${error.code} ${error.message}`);
  return data;
}

let idle = 0;
for (;;) {
  const { items } = await api("/v1/queue/consume", { queue: "webhooks-nightly", max_messages: 10 });
  if (items.length === 0) {
    if (++idle > 20) break; // backlog cleared; exit so the supervisor can restart tomorrow
    await sleep(3000);
    continue;
  }
  idle = 0;
  for (const msg of items) {
    const { webhook_id, url, event } = msg.payload;
    try {
      const res = await fetch(url, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ event, webhook_id }) });
      if (!res.ok) throw new Error(`endpoint returned ${res.status}`);
      await api("/v1/queue/ack", { queue: "webhooks-nightly", receipt_handle: msg.message_id });
    } catch (e) {
      console.warn(`webhook ${webhook_id} attempt ${msg.delivery_count} failed: ${e.message}`);
    }
    await sleep(200); // 5 rps against the destination
  }
}

A failed delivery is simply not acked. The message reappears when its visibility window expires and comes back with delivery_count incremented; after three deliveries it lands in the queue’s dead-letter lane instead of cycling forever.

Check the sweep landed before you go to bed:

curl -sS "https://api.infrai.cc/v1/queue/stats/webhooks-nightly" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}"

available_count should equal what the sweep reported publishing, in_flight_count tracks whatever the worker is holding right now, and a rising dlq_count is your signal that a customer endpoint has been down all night.

Delayed delivery: schedule the tick, not the message

Here’s a limitation to design around rather than discover at 3am. Setting delivery_delay_seconds on a queue had no effect in our testing on 2026-07-26 — the value came back as 0 after the update and published messages were immediately available. If you need “deliver this in six hours”, the reliable shapes are a later cron tick, or a second queue that a separate worker drains on its own schedule. Don’t build a product feature on a per-message delay you haven’t verified.

What a 50,000-webhook night costs

curl -sS "https://api.infrai.cc/v1/discovery" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  | jq '.capabilities[] | select(.id | startswith("queue.publish")) | {id, unit: .billing.unit, price_usd: .billing.price_usd}'

Publishing is the metered part at $0.00002 per message, verified 2026-07-26, so 50,000 webhooks a night is $1 — and consume, ack and the dead-letter routes are free, meaning a retry storm doesn’t move the number. Batched publishes billed at the same per-message rate in our testing, though the catalogue lists publish_batch with a per-call unit, so check GET /v1/account/usage after your first real run rather than assuming either reading. Prices here trend downward and campaigns run, so today’s figure may be lower than this page.

When another tool is the better call

If your infrastructure is already AWS-shaped and IAM is doing your authorisation, sqs plus EventBridge is fewer moving parts than anything you’d add. If what you want is literally “POST this URL at 4am and retry on failure” with no worker process at all, qstash is built for that and you’d be better off using it. This pattern earns its keep when the queue is one capability among several on the same credential — the delivery worker that also stores an artefact, emails a summary, and attributes the run’s cost to a tenant without a second vendor account.

References

• Infrai queue API reference
• QUEUE_MESSAGE_TOO_LARGE error reference
• Webhook Scheduler: cron job alternatives for delayed webhooks
• DEV: adding queue and cron handlers to a Cloudflare Worker