Bulk event notifications in Node: queue, batch send, cron sweep

A four-hop fan-out for email and SMS notifications — publish 100 at a time, lease them in a worker, batch-send per channel, and poll delivery on a schedule.

Fan-out gets expensive in the wrong places. Not the sending — the retries, the duplicate notifications after a crash, and the hour you spend working out which of 8,000 recipients actually got the message. Infrai puts the queue, both delivery channels and the scheduler behind one key, so the pipeline below is four HTTP routes rather than four vendors, and the dedupe and delivery-audit questions have answers instead of workarounds.

The shape is deliberately boring: an event producer publishes, a worker leases a batch, the worker groups by channel and calls POST /v1/email/batch/send or POST /v1/sms/batch/send, then acks. BullMQ does the first half of that beautifully if you already run Redis — this is what it looks like when you’d rather not.

Four hops

A failed message that’s never acked returns to available after its visibility timeout and gets redelivered; after max_receive_count attempts it lands in the dead-letter queue instead of spinning forever.

Provisioning the queue

The create call takes a name, a type and — the part worth setting on day one — a dead-letter queue:

{
  "name": "kbhub-notify-demo",
  "type": "standard",
  "dead_letter_queue": "kbhub-notify-demo.dlq",
  "max_retries": 3,
  "visibility_timeout_default": 300
}

curl -X POST https://api.infrai.cc/v1/queue/create \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  --data @queue.json

A 300-second visibility timeout means a worker has five minutes to finish a batch before the messages come back. Size that against your slowest send, not your average one.

Publishing a hundred at a time

Each message carries a payload object of your own shape and, optionally, a deduplication_id. That id is how you make an at-least-once queue behave: derive it from the event, not from a timestamp.

{
  "queue": "kbhub-notify-demo",
  "messages": [
    {
      "payload": { "event": "invoice.paid", "user_id": "usr_1", "channel": "email" },
      "deduplication_id": "invoice.paid:usr_1"
    },
    {
      "payload": { "event": "invoice.paid", "user_id": "usr_2", "channel": "sms" },
      "deduplication_id": "invoice.paid:usr_2"
    }
  ]
}

curl -X POST https://api.infrai.cc/v1/queue/publish_batch \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  --data @batch.json

The cap is 100 messages per call. Chunk your recipient list accordingly — a 50,000-row notification run is 500 of these, which is a for loop and about eight seconds of wall clock.

Leasing work

curl -X POST https://api.infrai.cc/v1/queue/consume \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{ "queue": "kbhub-notify-demo", "max_messages": 25 }'

{
  "ok": true,
  "data": {
    "items": [
      {
        "message_id": "qmsg_HABg7OYiADvIWwCWc5wthDic",
        "queue": "kbhub-notify-demo",
        "payload": { "event": "invoice.paid", "user_id": "usr_1", "channel": "email" },
        "status": "in_flight",
        "delivery_count": 1
      }
    ],
    "next_cursor": null
  }
}

delivery_count is the field to watch. Anything above 1 is a redelivery, which means a previous worker either crashed or ran past the lease — and it’s the signal you want in front of you before you go hunting for a bug in the send.

The worker

Node 22, ESM, no dependencies. It leases, splits by channel, batches each channel, and acks by message_id.

const BASE = "https://api.infrai.cc";
const KEY = process.env.INFRAI_API_KEY;
if (!KEY) throw new Error("INFRAI_API_KEY unset (placeholder: your_infrai_api_key)");

const QUEUE = "kbhub-notify-demo";
const LEASE = 25;

async function api(path, payload) {
  const res = await fetch(`${BASE}${path}`, {
    method: "POST",
    headers: { Authorization: `Bearer ${KEY}`, "Content-Type": "application/json" },
    body: JSON.stringify(payload),
  });
  const json = await res.json().catch(() => ({}));
  if (!res.ok) {
    const message = json?.error?.message ?? `HTTP ${res.status}`;
    throw Object.assign(new Error(`${path}: ${message}`), {
      status: res.status,
      permanent: /E\.164|format|invalid|not found/i.test(message),
    });
  }
  return json.data;
}

const chunk = (xs, n) => Array.from({ length: Math.ceil(xs.length / n) }, (_, i) => xs.slice(i * n, i * n + n));

function renderEmail(p) {
  return {
    to: p.email,
    subject: `Receipt for ${p.event}`,
    html: `<p>Hi ${p.name ?? "there"} — your ${p.event} went through.</p>`,
  };
}

function renderSms(p) {
  return { to: p.phone, body: `Acme: your ${p.event} went through.` };
}

export async function drain() {
  const lease = await api("/v1/queue/consume", { queue: QUEUE, max_messages: LEASE });
  const items = lease.items ?? [];
  if (!items.length) return { leased: 0, sent: 0 };

  const email = items.filter((m) => m.payload.channel === "email" && m.payload.email);
  const sms = items.filter((m) => m.payload.channel === "sms" && /^\+[1-9]\d{7,14}$/.test(m.payload.phone ?? ""));
  const undeliverable = items.filter((m) => !email.includes(m) && !sms.includes(m));

  let sent = 0;
  for (const group of chunk(email, 100)) {
    await api("/v1/email/batch/send", {
      messages: group.map((m) => renderEmail(m.payload)),
      idempotency_key: `email:${group[0].message_id}`,
    });
    sent += group.length;
  }
  for (const group of chunk(sms, 100)) {
    await api("/v1/sms/batch/send", {
      messages: group.map((m) => renderSms(m.payload)),
      idempotency_key: `sms:${group[0].message_id}`,
    });
    sent += group.length;
  }

  for (const m of [...email, ...sms, ...undeliverable]) {
    await api("/v1/queue/ack", { queue: QUEUE, message_id: m.message_id });
  }
  if (undeliverable.length) console.warn(`dropped ${undeliverable.length} messages with no usable address`);
  return { leased: items.length, sent };
}

try {
  console.log(await drain());
} catch (err) {
  console.error(err.permanent ? `bad data, not retrying: ${err.message}` : `transient: ${err.message}`);
  process.exitCode = 1;
}

Two things there are deliberate. Messages with no usable address get acked rather than nacked, because a missing phone number will still be missing on the fourth redelivery and all you’d achieve is filling the DLQ. And the ack happens after the batch call returns, so a crash mid-send costs you a duplicate rather than a silent drop — with an at-least-once queue that’s the trade-off you choose, and the deduplication_id on publish is what keeps it from mattering.

Validate recipients before you batch

Malformed input on these routes doesn’t come back as a 400. It arrives through the vendor channel:

{
  "ok": false,
  "error": {
    "code": "VENDOR_DOWN",
    "http_status": 503,
    "message": "recipient not in E.164 format: '555-not-e164'",
    "retryable": true
  }
}

A worker with a generic retry-on-5xx policy treats that as transient and retries a batch that can never succeed — and if the failure happened after delivery started, the retry is a second send you pay for. That’s why the E.164 test sits in drain() before the batch is assembled, and why api() derives its own permanent flag from the message text rather than trusting retryable.

Cron instead of a long-running daemon

If you don’t want a process babysitting the queue, schedule a poke at your worker endpoint. POST /v1/cron/create takes an absolute delivery target in task, plus either cron_expr for a recurring job or run_at for a one-shot:

{
  "name": "notify-drain",
  "cron_expr": "*/2 * * * *",
  "task": "https://worker.example.com/internal/drain",
  "timezone": "UTC",
  "overlap_policy": "skip"
}

curl -X POST https://api.infrai.cc/v1/cron/create \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  --data @cron.json

overlap_policy: "skip" is the one that saves you: a drain that runs long won’t have a second copy started on top of it. Afterwards, GET /v1/cron/list shows the job and GET /v1/cron/runs/list/{id} gives you fired-at times, durations and HTTP statuses per run — that’s your polling audit trail, and it’s free.

Watching it work

curl -s https://api.infrai.cc/v1/queue/stats/kbhub-notify-demo \
  -H "Authorization: Bearer ${INFRAI_API_KEY}"

{
  "ok": true,
  "data": {
    "queue": "kbhub-notify-demo",
    "message_count": 0,
    "available_count": 0,
    "in_flight_count": 0,
    "delayed_count": 0,
    "dlq_count": 0,
    "oldest_message_age_seconds": 0
  }
}

oldest_message_age_seconds climbing is the alert you want; dlq_count above zero is the one you want to page on. For per-recipient outcomes, GET /v1/email/event/list takes message_id as a query parameter (not a path segment — a request without it returns 400) and returns a timeline whose timestamp field is at. On the SMS side, GET /v1/sms/status/{id} gives the latest state. There’s also GET /v1/sms/events/{id} for the full history, with a caveat: it reads through to the carrier and answers VENDOR_NOT_CONFIGURED with a 503 on an account whose SMS vendor key isn’t hydrated, so treat status as the dependable one and events as the nice-to-have.

What a run costs

Verified 2026-07-26: publishing is about $0.001 per batch call of up to 100, an email runs about $0.000115 and an SMS about $0.0075. So 10,000 email notifications land near $1.25 all-in — 100 publish calls plus the messages — while the same 10,000 over SMS is closer to $75. Consume, ack, stats and the whole cron surface are free. Batch send is a passthrough rather than a discount: 100 messages in one call cost the same as 100 single sends, and share one idempotency_key so a retried batch isn’t double-billed. Rates drift downward and campaigns run, so read them live:

curl -s https://api.infrai.cc/v1/discovery \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  | jq '[.capabilities[]
        | select(.id=="email.batch.send" or .id=="sms.batch.send" or .id=="queue.publish_batch")
        | {id, price: .billing.price_usd, unit: .billing.unit}]'

New accounts start with $2 free credit, which is roughly 17,000 emails or 260 texts — the asymmetry between the channels is the durable fact here, not either figure.

Where something else fits better

If your notification volume is genuinely large and SMS-heavy, Twilio’s Messaging Services and Plivo’s high-throughput sending give you number pools, per-country routing and carrier-level throughput controls that this API doesn’t support. Buy the specialist for that. Likewise, if you already run Redis and want in-process job priorities, retries with custom backoff curves and a dashboard, BullMQ is a better queue than any HTTP one — the drawback is that it’s a queue and nothing else, so the email, the SMS and the scheduler stay separate purchases.

The argument here is consolidation, not throughput. One key covers the queue, both channels and the schedule; one usage view attributes the run to a tenant; and adding a third channel later is a new function in drain(), not a new vendor relationship. If notifications are the only thing you’re building, you’d be better off with the strongest point tools in each category — most teams building this are also storing files, running crons and tracking errors, and that’s when a single account starts paying for itself.

References

• Infrai queue API reference
• Infrai SMS API reference
• BullMQ workers guide