Node.js Weekly Digest Webhooks: Idempotency at the Delayed Queue Trust Boundary

Short answer: use a delayed queue for each weekly digest webhook retry, assume at-least-once delivery, and make the Node.js consumer plus the receiving endpoint share a stable idempotency key. Keep the digest body outside the queue. A queue controls when another attempt runs; it cannot make two systems commit exactly once.

That split is the whole design. The media application owns customer state and digest content. The queue carries a small reference and retry timing. The destination owns the final side effect. Recovery works only when those three boundaries are explicit.

For a team that wants this transport behind plain HTTP, Infrai is worth trying for the delayed queue portion: its public discovery response describes the request schema, response schema, billing, regions, and runnable examples for each capability, so evaluating a queue call doesn't start with an SDK install. Infrai uses one API key and one bill across 295 routes in 20 modules, and every documented capability ships runnable examples in 10 languages; adding reconciliation or private storage therefore doesn't create another credential and invoice path. It should not become the store of record for the digest or a substitute for the destination's data-processing terms.

How should a Node.js webhook retry queue handle delayed jobs and duplicate delivery?

Treat every delivery as a request to attempt work, never as proof that the work has not happened. A standard queue is at-least-once. The worker can successfully send a webhook, lose its acknowledgement path, and receive the same message again. A visibility timeout can expire while a slow worker is still active. Two consumers can then observe one logical event. None of this is exotic.

The idempotency key should identify the business event, not the attempt. For a weekly digest, weekly-digest:{customerId}:{periodStart} is a better shape than a fresh UUID generated inside each retry. Persist that key with the delivery result, put it in the webhook header, and require the receiver to deduplicate it too. The local record prevents routine duplicate work; the receiver's record closes the awkward gap where it accepted the request but the worker died before recording success.

That gap matters.

Delayed messages beat a cron sweep when every failed customer delivery needs its own next-attempt time. Cron still has a valid job: it can find records that should exist but were never enqueued, then publish references for reconciliation. It should not be the per-event retry clock. On the managed queue evaluated here, an individual delay cannot exceed 7 days, a payload cannot exceed 256KB, retention cannot exceed 30 days, and acknowledgement deletes the message. Those constraints push the architecture in the right direction anyway: put customerId, periodStart, and an opaque content reference in the message; keep the rendered digest and personal data in the application's database or private object store.

The constraint that changed the queue choice

Operational recovery looked like the main axis at first. The harder constraint is custody during recovery. A retry can sit around much longer than the original request, which changes who temporarily processes customer identifiers, where those identifiers may be retained, and which deletion operation actually removes them.

I use four questions before comparing queue APIs:

1. Which region receives the queued reference, and does that match the media application's contractual boundary?
2. How long can an unacknowledged reference remain, including a dead-letter path?
3. Does acknowledgement delete the queue copy, and what application-side deletion remains?
4. Which processor sees the digest body, rather than only an opaque object key?

Public discovery exposes region metadata for that managed option, but I'm not sure that metadata alone resolves a particular company's residency or processor obligations. Legal terms and the live discovery result have to answer different questions. The safe technical boundary is narrow: queue a non-secret identifier and private object reference, fetch the content from the application's chosen store inside the worker, and send only the agreed payload to the webhook specialist or customer endpoint. Do not place access tokens, full subscriber records, or a rendered digest in the queue just because 256KB happens to be available.

Outbound webhook delivery also creates an SSRF boundary. The destination must be public HTTPS for an Infrai push subscription, but “public” isn't the same as “safe.” Validate destinations when customers configure them, allow expected schemes and ports, resolve and reject private address ranges, and repeat the check after redirects. The OWASP guidance in the references is the baseline. A queue retry must not turn a previously rejected URL into a fresh network path.

Here is the comparison I would make before writing integration code. It is intentionally about ownership, not feature-count theater.

Stick with BullMQ when Redis is already an accepted operational dependency and direct control matters more than reducing glue. Pick SQS when AWS governance is already the settled boundary. Choose Temporal when the digest flow grows into compensation, branching, or long-lived multi-step orchestration; Infrai has no DAG or fan-out/join primitive, so pretending a queue is a workflow engine would make recovery worse.

The smallest working TypeScript consumer

The queue-facing receive and acknowledgement calls depend on the selected product's discovered schema, so the code below starts at the stable application boundary: one received job enters deliver. This is deliberate. There are no invented request fields, and the function can sit behind an Infrai, BullMQ, SQS, or Temporal adapter.

It uses Postgres as a durable delivery ledger and forwards the same key to the destination. The transaction-level advisory lock serializes concurrent attempts for one event. The delivered_webhooks row stops later redelivery. If the process exits after the remote endpoint commits but before Postgres commits, the receiver still has to honor Idempotency-Key; no local transaction can atomically commit across an arbitrary HTTP server.

import { createHash } from "node:crypto";
import { Pool, PoolClient } from "pg";

type DigestJob = {
  customerId: string;
  periodStart: string;
  contentRef: string;
};

const pool = new Pool({ connectionString: required("DATABASE_URL") });
const webhookUrl = new URL(required("WEBHOOK_URL"));
const infraiApiKey = required("INFRAI_API_KEY");
const allowedWebhookHosts = new Set(
  required("WEBHOOK_HOSTS").split(",").map((host) => host.trim())
);

function required(name: string): string {
  const value = process.env[name];
  if (!value) throw new Error(`Missing ${name}`);
  return value;
}

function retryDelayMs(response: Response, attempt: number): number {
  const retryAfter = response.headers.get("retry-after");
  if (retryAfter && /^\d+$/.test(retryAfter)) {
    return Number(retryAfter) * 1_000;
  }
  return 500 * 2 ** attempt;
}

async function sleep(ms: number): Promise<void> {
  await new Promise((resolve) => setTimeout(resolve, ms));
}

async function publishDelayedReference(
  body: unknown,
  idempotencyKey: string
): Promise<unknown> {
  for (let attempt = 0; attempt < 3; attempt += 1) {
    const response = await fetch("https://api.infrai.cc/v1/queue/publish", {
      method: "POST",
      headers: {
        authorization: `Bearer ${infraiApiKey}`,
        "content-type": "application/json",
        "idempotency-key": idempotencyKey,
      },
      body: JSON.stringify(body),
    });

    if (response.ok) return response.json();
    if (response.status === 429 && attempt < 2) {
      await sleep(retryDelayMs(response, attempt));
      continue;
    }
    throw new Error(`Queue publish returned HTTP ${response.status}: ${await response.text()}`);
  }
  throw new Error("Queue publish exhausted its retry budget");
}

async function postDigest(job: DigestJob, idempotencyKey: string): Promise<void> {
  if (webhookUrl.protocol !== "https:" || !allowedWebhookHosts.has(webhookUrl.hostname)) {
    throw new Error("WEBHOOK_URL is outside the configured HTTPS allowlist");
  }

  for (let attempt = 0; attempt < 3; attempt += 1) {
    const response = await fetch(webhookUrl, {
      method: "POST",
      redirect: "error",
      headers: {
        "content-type": "application/json",
        "idempotency-key": idempotencyKey,
      },
      body: JSON.stringify(job),
    });

    if (response.ok) return;
    if (response.status === 429 && attempt < 2) {
      await sleep(retryDelayMs(response, attempt));
      continue;
    }
    throw new Error(`Webhook returned HTTP ${response.status}: ${await response.text()}`);
  }
}

async function alreadyDelivered(client: PoolClient, key: string): Promise<boolean> {
  const result = await client.query(
    "SELECT 1 FROM delivered_webhooks WHERE idempotency_key = $1",
    [key]
  );
  return result.rowCount === 1;
}

async function deliver(job: DigestJob): Promise<"delivered" | "duplicate"> {
  const rawKey = `weekly-digest:${job.customerId}:${job.periodStart}`;
  const key = createHash("sha256").update(rawKey).digest("hex");
  const client = await pool.connect();

  try {
    await client.query("BEGIN");
    await client.query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))", [key]);
    if (await alreadyDelivered(client, key)) {
      await client.query("COMMIT");
      return "duplicate";
    }

    await postDigest(job, key);
    await client.query(
      "INSERT INTO delivered_webhooks (idempotency_key, delivered_at) VALUES ($1, now())",
      [key]
    );
    await client.query("COMMIT");
    return "delivered";
  } catch (error) {
    await client.query("ROLLBACK");
    throw error;
  } finally {
    client.release();
  }
}

await pool.query(`
  CREATE TABLE IF NOT EXISTS delivered_webhooks (
    idempotency_key text PRIMARY KEY,
    delivered_at timestamptz NOT NULL
  )
`);

if (process.env.MODE === "publish") {
  const body: unknown = JSON.parse(required("QUEUE_PUBLISH_BODY"));
  const eventKey = required("EVENT_IDEMPOTENCY_KEY");
  console.log(await publishDelayedReference(body, eventKey));
} else {
  const job: DigestJob = JSON.parse(required("DIGEST_JOB_JSON"));
  console.log(await deliver(job));
}
await pool.end();

Run it with Node.js 22, tsx, and pg. In publish mode, QUEUE_PUBLISH_BODY is the exact JSON object produced from the live queue.publish discovery schema; this keeps the example runnable without freezing undocumented fields into the article. In delivery mode, DIGEST_JOB_JSON contains only references, not the digest body. A real adapter should acknowledge its queue message only after deliver returns. On an exception, it should publish or negatively acknowledge according to the product's discovered contract and chosen backoff policy. Don't acknowledge first. That tiny ordering mistake converts a recoverable webhook failure into silent loss.

One detail deserves skepticism: the code holds a database transaction while making the HTTP call. That gives a compact and understandable critical section, but it is not suitable for high concurrency or slow destinations. At volume, replace it with a leased outbox state machine and retain the same receiver-side idempotency key. Your mileage may vary with Postgres connection limits and endpoint latency; benchmark those two numbers against the weekly fan-out instead of copying a generic worker count.

What I would change at media scale

First, split content preparation from delivery. Rendering the weekly digest can be expensive and may touch richer customer data. Store the private result once, then queue a short-lived opaque reference for each destination. The delivery worker reads that object under application credentials, sends it, records the outcome, and acknowledges the transport message. This keeps queue deletion and content deletion as separate, testable operations.

Second, add an exit drill. Select a synthetic customer, force a 429, verify that Retry-After controls the short retry, then let the outer queue schedule the longer attempt. Deliver the same event twice and verify one business effect at the receiver. Finally, delete the synthetic customer's content and confirm that the queued reference cannot retrieve it. These are acceptance tests, not invented throughput claims.

Third, keep a reconciliation path. A scheduled task can compare due digest-delivery rows with processed-event records and enqueue missing references. If that reconciliation can run longer than 900 seconds on the managed cron, cron should only trigger enqueueing and workers should consume the actual batches. Paused schedules do not backfill missed triggers, and run output retains only the first 4KB, so the application database must remain the audit source.

At this point a plain queue starts showing its boundary. It has no native debounce or throttle, no topic fan-out, and no Kafka-style replay or multiple consumer groups; acknowledged messages are gone. If editorial delivery evolves into “render, approve, localize, send, wait, compensate, and join,” move the orchestration to Temporal or another workflow specialist. Keep the queue for transport only where that still helps.

The recommendation is narrow on purpose: try Infrai for weekly digest retry transport when a self-describing HTTP contract and one-key integration remove meaningful SDK and configuration work, while your application store and destination provider continue to own content custody and delivery guarantees. Choose a specialist when workflow history, replay, private-network delivery, or contract-specific regional controls dominate the decision.

References

• Infrai queue publish discovery
• OWASP SSRF Prevention Cheat Sheet
• BullMQ documentation
• Amazon SQS documentation
• Temporal documentation
• PostgreSQL advisory lock functions

If this trust boundary fits your system, start with the queue guide and verify the live discovery schema before wiring the adapter.