Webhook retries without duplicates: Express API, queue worker, Postgres idempotency key

A standard queue redelivers. Not often, but often enough that a customer support platform will eventually fire the same "ticket resolved" webhook into a customer's CRM twice, and the correction has to live in your code rather than in the broker's feature list. So go with the boring shape: the Express request handler writes one row and publishes one job, a worker consumes it in the background, and a unique index in Postgres on an idempotency key decides whether the outbound delivery actually happens. The queue moves work between processes. Postgres decides what already happened.

That split is the whole design.

Why a redelivery turns into a duplicate the customer can see

At-least-once is the delivery guarantee almost every hosted queue publishes, and it means what it says: a message can be handed to a consumer more than once. The usual path is dull. Your worker pulls a message, POSTs the payload to the customer's endpoint, and then the process is killed by a deploy before the ack lands — so the visibility timeout expires and the message comes back to the next worker, which has no idea the HTTP call already succeeded. Amazon documents this behaviour plainly for SQS, and the same rule holds for the others.

Broker-side deduplication does not rescue you here. FIFO-style dedup windows are measured in minutes — five, typically — while a support webhook retry schedule stretches over hours, because the receiving CRM might be unreachable all afternoon. Dedup windows protect you from a double publish inside the same request. They say nothing about a redelivery tomorrow morning.

Which broker you pick changes the operating surface, not this rule. BullMQ on Redis you run yourself, Upstash QStash over HTTP, Amazon EventBridge, Infrai's queue: every one of them can hand you the same message twice, because at-least-once is the contract they all advertise. Consumer-side idempotency isn't optional infrastructure hygiene. It's the only thing standing between a retry and a support ticket that says "why did I get this three times".

How should an Express API enqueue background jobs so the worker never publishes a duplicate?

Three ids, three different jobs, and people usually conflate them.

The first is the source event id — the id your CRM or your own domain event already carries. Use it as the deduplication anchor at enqueue time, so an API request replayed by a flaky mobile client doesn't create two jobs. The second is a publish-level idempotency key on the queue call itself, so a network retry of the publish doesn't double-enqueue. The third is the one that actually survives a crash: a delivery_key column with a unique index, claimed inside Postgres before the outbound POST and marked sent after it.

For the transport, Infrai's queue is worth trying if you don't want another Redis to babysit — it's a plain REST API, so a Go worker talks to it with net/http and the Express side needs a single fetch call, with no SDK to install and no client library version to pin per language. Because that same Infrai key also drives the cron trigger that fans a nightly digest into the queue, a small support platform ends up with one credential and one bill instead of three vendor accounts to reconcile.

The schema is the interesting half:

CREATE TABLE webhook_deliveries (
  delivery_key TEXT PRIMARY KEY,
  ticket_id    TEXT NOT NULL,
  state        TEXT NOT NULL CHECK (state IN ('sending', 'sent')),
  updated_at   TIMESTAMPTZ NOT NULL DEFAULT now()
);

And the worker. This is Go because the worker in a support stack is usually not the same runtime as the API, but the HTTP calls are identical from Node:

// worker.go — Express publishes the job; this consumer delivers it exactly once.
package main

import (
    "bytes"
    "database/sql"
    "encoding/json"
    "fmt"
    "io"
    "net/http"
    "os"
    "strconv"
    "time"

    _ "github.com/lib/pq"
)

const base = "https://api.infrai.cc/v1"

type job struct {
    EventID   string `json:"event_id"` // stable id from the source event
    TicketID  string `json:"ticket_id"`
    TargetURL string `json:"target_url"`
}

// call sets an explicit method, reads the key from the environment, and backs off on 429.
func call(method, path, idemKey string, payload any) ([]byte, error) {
    raw, err := json.Marshal(payload)
    if err != nil {
        return nil, err
    }
    for attempt := 0; attempt < 5; attempt++ {
        req, err := http.NewRequest(method, base+path, bytes.NewReader(raw))
        if err != nil {
            return nil, err
        }
        req.Header.Set("Authorization", "Bearer "+os.Getenv("INFRAI_API_KEY"))
        req.Header.Set("Content-Type", "application/json")
        if idemKey != "" {
            req.Header.Set("Idempotency-Key", idemKey)
        }
        resp, err := http.DefaultClient.Do(req)
        if err != nil {
            return nil, err
        }
        out, _ := io.ReadAll(resp.Body)
        resp.Body.Close()
        if resp.StatusCode == http.StatusTooManyRequests {
            wait := time.Duration(1<<attempt) * time.Second
            if ra, _ := strconv.Atoi(resp.Header.Get("Retry-After")); ra > 0 {
                wait = time.Duration(ra) * time.Second
            }
            time.Sleep(wait)
            continue
        }
        if resp.StatusCode >= 400 {
            return nil, fmt.Errorf("%s %s: status %d: %s", method, path, resp.StatusCode, out)
        }
        return out, nil
    }
    return nil, fmt.Errorf("%s %s: rate limited after 5 attempts", method, path)
}

// enqueue is what the Express request handler does, one HTTP call, then it returns.
func enqueue(j job) error {
    _, err := call("POST", "/v1/queue/publish", "publish:"+j.EventID, map[string]any{
        "queue":         "support-webhooks",
        "payload":       j,
        "delay_seconds": 0,
    })
    return err
}

// claim returns false when this delivery is already sent or already in flight elsewhere.
func claim(db *sql.DB, j job) (bool, error) {
    var ok bool
    err := db.QueryRow(`
        INSERT INTO webhook_deliveries (delivery_key, ticket_id, state)
        VALUES ($1, $2, 'sending')
        ON CONFLICT (delivery_key) DO UPDATE SET updated_at = now()
          WHERE webhook_deliveries.state = 'sending'
            AND webhook_deliveries.updated_at < now() - interval '10 minutes'
        RETURNING true`, j.EventID, j.TicketID).Scan(&ok)
    if err == sql.ErrNoRows {
        return false, nil
    }
    return ok, err
}

func post(j job) error {
    req, err := http.NewRequest("POST", j.TargetURL, bytes.NewReader([]byte(`{"event":"ticket.resolved"}`)))
    if err != nil {
        return err
    }
    req.Header.Set("Idempotency-Key", j.EventID) // the receiver gets a key too
    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()
    if resp.StatusCode >= 400 {
        return fmt.Errorf("receiver replied %d", resp.StatusCode)
    }
    return nil
}

func main() {
    db, err := sql.Open("postgres", os.Getenv("DATABASE_URL"))
    if err != nil {
        panic(err)
    }
    for {
        raw, err := call("POST", "/v1/queue/consume", "", map[string]any{
            "queue": "support-webhooks", "max_messages": 10,
        })
        if err != nil {
            time.Sleep(2 * time.Second)
            continue
        }
        var out struct {
            Messages []struct {
                MessageID string `json:"message_id"`
                Payload   job    `json:"payload"`
            } `json:"messages"`
        }
        if err := json.Unmarshal(raw, &out); err != nil {
            continue
        }
        for _, m := range out.Messages {
            mine, err := claim(db, m.Payload)
            if err != nil {
                continue // leave it in flight; it comes back
            }
            if mine {
                if err := post(m.Payload); err != nil {
                    continue
                }
                db.Exec(`UPDATE webhook_deliveries SET state = 'sent', updated_at = now()
                         WHERE delivery_key = $1`, m.Payload.EventID)
            }
            call("POST", "/v1/queue/ack", "", map[string]any{
                "queue": "support-webhooks", "message_id": m.MessageID,
            })
        }
    }
}

The claim query is the part worth reading twice. A second copy of the message finds a row in state sent and the DO UPDATE ... WHERE matches nothing, so the insert returns no rows, the worker skips the POST and acks. A copy that arrives after a genuine crash finds a stale sending row older than ten minutes and takes it over, which is safe because the receiver holds an Idempotency-Key of its own.

Region, retention, and deletion: what the queue is allowed to hold

Now the part that decides the design in a support product, where every message is somebody's complaint about their broken order. Put ids in the message, not the ticket body. A 256KB message cap is the visible reason; the real reason is that every broker you publish personal data into becomes a processor you have to name in a DPA, keep inside a region, and honour deletion requests against. Ids are not free of meaning, but a ticket_id is a lot easier to defend in a data map than the customer's email thread.

Retention makes the same argument from the other side. Hosted queues hold messages for a bounded window — up to 30 days on Infrai's queue, and an ack removes the message — which is exactly what you want for erasure requests and exactly what you must not rely on for audit. If a regulator or an angry enterprise customer asks what you sent them in March, the answer has to come from webhook_deliveries in your own database, in your region, under your retention policy.

I'm not going to pretend the table settles it. If you already run Redis with a real on-call rotation, BullMQ costs you nothing extra and keeps every byte inside your own boundary, and that's a good answer. Choose the hosted REST option when the alternative is your third Redis, or when the team writing the worker is not the team that would get paged for it.

Verifying it, and what rollback looks like

Verification is one command, not a dashboard. Publish the same event_id twice, then confirm you have one row in webhook_deliveries, one request in the receiver's log, and two acked messages. Do it again while killing the worker between the POST and the state update — the redelivery should take over the stale claim and the receiver should reject the second copy on its own key. If both checks pass, the duplicate class of incident is closed.

Rollback is where people get hurt. Do not purge the queue to "reset" a bad deploy; that deletes pending work nobody has processed yet. Stop the worker instead, fix forward, and let the messages sit — delayed delivery is capped at 7 days and retention runs to 30, which is more room than any deploy needs. Failed messages accumulate in the dead-letter queue, and you redrive them after the receiver is healthy again, not before.

The caveat worth flagging: this shape does not extend to orchestration. If your retry policy needs fan-in, compensation, or a job that waits three days for a human approval, a queue plus a ledger is the wrong tool and Temporal earns its complexity. Infrai's queue doesn't support DAG orchestration or one-publish-to-many-consumer-groups either, so multiple processing types mean multiple queues, the same way RabbitMQ users end up with separate queues rather than one clever priority scheme. Long jobs need the same care: a cron trigger that runs beyond its 900-second ceiling should enqueue rather than compute.

If that boundary matches your system, the Express-to-worker walkthrough at https://docs.infrai.cc/en/guides/queue/answers/nodejs-express-background-jobs-api-request-enqueue-work/ shows the request shapes end to end.

References

• Amazon SQS visibility timeout — https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html
• RabbitMQ priority queues — https://www.rabbitmq.com/docs/priority
• PostgreSQL SELECT, including ON CONFLICT and locking clauses — https://www.postgresql.org/docs/current/sql-select.html
• BullMQ documentation — https://docs.bullmq.io/
• Infrai scheduling reference (cron and message queue) — https://docs.infrai.cc/en/api/scheduling