Reminder notifications that send exactly once: cron, Postgres due_at, queue

A Node reminder pipeline where the minute cron, the publish and the worker each refuse to duplicate: SKIP LOCKED claims, idempotency keys, a sent_at guard.

A reminder that arrives twice is a bug users report; one that never arrives is a bug they leave over. Getting both right in Node comes down to a due_at column, a cron that fires every minute, and a queue worker — and Infrai gives you the cron and the queue on the same key, so the only piece you own is the Postgres table. This is the version we’d actually run, with the duplicate-suppression written at all three layers rather than hoped for at one.

The short answer to “where does the schedule live”: in Postgres. The queue is transport, not memory.

Three places a reminder gets duplicated

Each layer of the pipeline retries, and each retry is a chance to send twice.

The cron fires your endpoint over HTTP and retries on failure — default 3 attempts. If your sweep already claimed rows before it timed out, attempt two claims them again. Then the publish itself can be retried by your own HTTP client after a socket timeout, even though the first request was accepted. And finally the queue is at-least-once by design, so a worker that crashes after sending but before acking will see the message again when the 300-second visibility timeout expires.

Fix all three or you’ve fixed none.

The rows that make it decidable

CREATE TABLE reminders (
  id            bigserial PRIMARY KEY,
  user_id       bigint      NOT NULL,
  channel       text        NOT NULL,
  body          text        NOT NULL,
  due_at        timestamptz NOT NULL,
  queued_at     timestamptz,
  sent_at       timestamptz,
  attempts      int         NOT NULL DEFAULT 0
);

CREATE INDEX reminders_due ON reminders (due_at) WHERE sent_at IS NULL AND queued_at IS NULL;

The claim is a single statement, and FOR UPDATE SKIP LOCKED is what lets two sweeps run at once without fighting over the same rows — the pattern PostgreSQL documents for exactly this shape of work queue.

UPDATE reminders SET queued_at = now()
WHERE id IN (
  SELECT id FROM reminders
  WHERE sent_at IS NULL AND queued_at IS NULL AND due_at <= now()
  ORDER BY due_at
  LIMIT 500
  FOR UPDATE SKIP LOCKED
)
RETURNING id, user_id, channel, body, due_at;

Setting queued_at inside the claim is the point. A row can only be handed to the queue once, and a crash between claim and publish leaves a visible orphan you can re-open with a second, much rarer query.

One cron, every minute

curl -sS -X POST https://api.infrai.cc/v1/cron/create \
  -H "Authorization: Bearer ${INFRAI_API_KEY:-your_infrai_api_key}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "reminder-sweep",
    "cron_expr": "* * * * *",
    "task": "https://api.example.com/internal/reminders/sweep",
    "timezone": "UTC",
    "overlap_policy": "skip",
    "timeout_seconds": 60,
    "retry": 1,
    "secret": "replace-with-a-32-byte-random-string"
  }'

The delivery target field is task, not task_url — that one costs people an afternoon. overlap_policy: "skip" means a sweep that runs long won’t have a second copy started on top of it, which removes an entire class of duplicate. And retry: 1 is deliberate: with an every-minute schedule, a failed run is cheaper to skip than to retry, because the next tick is 60 seconds away and will pick up the same rows.

Keep timeout_seconds under the tick interval. The maximum is 900, but a minute cron that takes 400 seconds is telling you the sweep should be publishing to the queue rather than doing the work.

The sweep handler

import express from "express";
import { timingSafeEqual, createHmac } from "node:crypto";
import pg from "pg";

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

const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
const app = express();
app.use(express.raw({ type: "*/*" }));

const CLAIM = `UPDATE reminders SET queued_at = now()
  WHERE id IN (SELECT id FROM reminders
    WHERE sent_at IS NULL AND queued_at IS NULL AND due_at <= now()
    ORDER BY due_at LIMIT 500 FOR UPDATE SKIP LOCKED)
  RETURNING id, user_id, channel, body`;

app.post("/internal/reminders/sweep", async (req, res) => {
  const expected = createHmac("sha256", SECRET).update(req.body).digest("hex");
  const got = Buffer.from(String(req.get("x-infrai-signature") ?? ""), "utf8");
  const want = Buffer.from(expected, "utf8");
  if (got.length !== want.length || !timingSafeEqual(got, want)) return res.status(401).end();

  const { rows } = await pool.query(CLAIM);
  let published = 0;
  for (const r of rows) {
    const out = await fetch("https://api.infrai.cc/v1/queue/publish", {
      method: "POST",
      headers: { authorization: `Bearer ${KEY}`, "content-type": "application/json" },
      body: JSON.stringify({
        queue: "due-reminders-sweep",
        payload: { reminder_id: r.id, user_id: r.user_id, channel: r.channel, body: r.body },
        idempotency_key: `reminder:${r.id}`,
      }),
    });
    if (out.ok) published++;
    else console.error(`publish failed for reminder ${r.id}: ${out.status} ${await out.text()}`);
  }
  res.json({ claimed: rows.length, published });
});

app.listen(3000, () => console.log("sweep endpoint on :3000"));

idempotency_key: "reminder:<id>" is the second layer. Replay the same publish after a timeout and you get the original message back rather than a twin — the response metadata even flags idempotent_replay.

The sender re-checks before it sends

At-least-once means the worker will occasionally see a message it has already handled. So the guard belongs in the write, not in an if:

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

const KEY = process.env.INFRAI_API_KEY;
if (!KEY) throw new Error("set INFRAI_API_KEY");
const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
const QUEUE = "due-reminders-sweep";

const post = async (path, body) => {
  const res = await fetch(`https://api.infrai.cc${path}`, {
    method: "POST",
    headers: { authorization: `Bearer ${KEY}`, "content-type": "application/json" },
    body: JSON.stringify(body),
  });
  if (!res.ok) throw new Error(`${path} -> ${res.status} ${await res.text()}`);
  return (await res.json()).data;
};

async function deliver(payload) {
  const claim = await pool.query(
    "UPDATE reminders SET sent_at = now(), attempts = attempts + 1 WHERE id = $1 AND sent_at IS NULL RETURNING id",
    [payload.reminder_id],
  );
  if (claim.rowCount === 0) return "already-sent";
  // real channel call goes here; throwing rolls nothing back, so compensate below
  return "sent";
}

for (;;) {
  const { items } = await post("/v1/queue/consume", { queue: QUEUE, max_messages: 10 });
  if (!items.length) { await sleep(3000); continue; }
  for (const msg of items) {
    try {
      const outcome = await deliver(msg.payload);
      await post("/v1/queue/ack", { queue: QUEUE, message_id: msg.message_id });
      if (outcome === "already-sent") console.log(`duplicate suppressed for ${msg.payload.reminder_id}`);
    } catch (err) {
      console.error(`reminder ${msg.payload.reminder_id} failed:`, err.message);
      await pool.query("UPDATE reminders SET sent_at = NULL WHERE id = $1", [msg.payload.reminder_id]);
      await post("/v1/queue/nack", { queue: QUEUE, message_id: msg.message_id, requeue: true });
    }
  }
}

Note the WHERE ... AND sent_at IS NULL RETURNING id. If rowCount is zero, another delivery already won the race and this copy exits quietly. Two workers, one send.

Ack quietly lies when it has nothing to ack — an unknown or already-expired message_id still returns HTTP 200, with {"acked": false} in the body. If your metrics are built on status codes you’ll never see a lease expiry.

Watching one minute of the sweep

curl -sS https://api.infrai.cc/v1/queue/stats/due-reminders-sweep \
  -H "Authorization: Bearer ${INFRAI_API_KEY:-your_infrai_api_key}"

available_count should spike each minute and fall to zero within seconds; a rising in_flight_count means workers are leasing and dying. dlq_count above zero means a reminder failed three times — read those by consuming due-reminders-sweep.dlq directly, since the dedicated DLQ listing route returned empty for us while the counter said otherwise.

What it costs to run for a month

Cron creation, cron runs, consume, ack, nack and stats are free rate-limited calls. Only queue.publish is metered, at $0.00002 per message (verified 2026-07-26). A product sending 50,000 reminders a month therefore pays about $1 for the queue, and the notification channel — email, SMS, push — is where the real money goes.

curl -sS https://api.infrai.cc/v1/discovery \
  -H "Authorization: Bearer ${INFRAI_API_KEY:-your_infrai_api_key}" \
  | jq '.capabilities[] | select(.id == "queue.publish" or .id == "cron.create") | {id, billable: .billing.is_billable, price: .billing.price_usd}'

curl -sS https://api.infrai.cc/v1/account/usage \
  -H "Authorization: Bearer ${INFRAI_API_KEY:-your_infrai_api_key}"

Prices here trend down and campaigns run, so the live figures may be lower than the ones printed above. The durable part is the shape: scheduling is free, transport is cheap and metered per message, and the same credential covers the send channel and the error capture, so per-tenant cost is one usage query instead of three invoices to reconcile.

When to reach for something else

If every reminder needs its own durable timer months out, and you want the retry policy expressed in code rather than in a table, Temporal is genuinely the better tool and you’d be better off adopting it than emulating it. BullMQ’s deterministic job ids give you the same de-duplication with less moving glue when you already run Redis. The limitation on the Infrai side is that the queue has no per-message scheduler beyond seven days and no repeatable-job concept at all — the minute cron plus due_at is the pattern, not a workaround. For picking the local wall-clock time each row should carry, see the companion piece at https://docs.infrai.cc/en/guides/queue/answers/timezone-aware-recurring-user-reminders-nodejs-cron-exa/.

References

• Infrai queue API reference
• Idempotent jobs — BullMQ
• PostgreSQL SELECT: FOR UPDATE SKIP LOCKED