---
title: "Webhook delivery system"
description: "Ordered delivery per subscriber, a bounded retry budget, and a dead-letter queue a support engineer can read."
---

> Queen MQ documentation, for AI agents
> Complete self-contained summary of Queen MQ: https://queenmq.com/llms-brief.txt
> Fetch that first when the question is about the product rather than about this page.
> Index of all pages: https://queenmq.com/llms.txt

# Webhook delivery system

Every SaaS product writes this one, and it is harder than it looks. Deliveries to one customer's
endpoint must arrive in order. A customer whose endpoint is down must not slow anybody else down.
Failures must be retried a bounded number of times, and what never succeeds has to end up
somewhere a human can look at.

One ordered lane per destination gives the first two properties for free: a dead endpoint backs up
its own lane and no other. The last two are the broker's retry budget and its dead-letter table,
which means they survive the sender process dying mid-flight, and a loop inside your handler would
not.

The program queues deliveries for three subscribers, one of which answers 500 to everything, and
checks that the healthy two receive every event in order while the dead one exhausts its budget
and lands in the dead-letter queue with its error attached.

### JavaScript

```js title="examples/apps/js/webhooks.mjs"
//
// A webhook delivery system.
//
// Every SaaS product ends up writing this one, and it is harder than it looks:
// deliveries to one customer's endpoint must arrive in order, a customer whose
// endpoint is down must not slow down anybody else's, failures must be retried
// a bounded number of times, and what never succeeds has to end up somewhere a
// human can look at.
//
// The shape here is one ordered lane per destination, created by the first
// delivery to it. A dead endpoint backs up its own lane and no other; retries
// are the broker's retry budget rather than a loop in your code; and what
// exhausts the budget lands in the dead-letter queue with the error attached.
//
//   webhook-deliveries (one partition per destination)
//     └── group "sender"  posts each delivery, fails on a dead endpoint
//           └── retryLimit exhausted -> dead-letter queue
//
// Run it:
//   QUEEN_URL=http://localhost:6632 node webhooks.mjs

import { Queen } from 'queen-mq'

const QUEEN_URL = process.env.QUEEN_URL || 'http://localhost:6632'
const RUN = Date.now().toString(36)
const DELIVERIES = `app-js-webhooks-${RUN}`
const GROUP = 'sender'

// Three subscribers. One of them has let its certificate expire, which is the
// most common way a webhook endpoint dies: it answers, but it answers 500.
const ENDPOINTS = {
  'acme.example': { healthy: true },
  'globex.example': { healthy: true },
  'initech.example': { healthy: false },
}
const EVENTS_PER_ENDPOINT = 3
const RETRY_LIMIT = 2

let checks = 0
const assert = (condition, description) => {
  if (!condition) throw new Error(description)
  checks++
  console.log(`  ok: ${description}`)
}

// Stands in for the HTTP POST to the subscriber. A real sender would use fetch
// and treat any non-2xx as a failure, which is exactly what throwing does here.
const postToEndpoint = async (endpoint, event) => {
  if (!ENDPOINTS[endpoint].healthy) {
    throw new Error(`${endpoint} answered 500`)
  }
  return { status: 200 }
}

const queen = new Queen({ url: QUEEN_URL, handleSignals: false })

try {
  console.log(`broker ${QUEEN_URL}`)

  // retryLimit is the delivery budget, and dlqAfterMaxRetries is what happens
  // when it runs out. Without the second flag an exhausted message is simply
  // marked failed and stays put; with it, the broker moves it to the
  // dead-letter table with the last error on the row.
  //
  // leaseTime is the other half of the contract: it is how long the broker
  // waits for a sender that took a delivery and never came back before handing
  // that delivery to someone else.
  await queen.queue(DELIVERIES).config({
    leaseTime: 30,
    retryLimit: RETRY_LIMIT,
    dlqAfterMaxRetries: true,
  }).create()

  // ------------------------------------------------------------------ queuing
  //
  // The application emits events. Each one goes into the partition of the
  // endpoint it is destined for, which is what makes "in order per subscriber"
  // a property of the storage rather than of the sender.
  console.log('\nqueuing deliveries')
  for (let seq = 1; seq <= EVENTS_PER_ENDPOINT; seq++) {
    for (const endpoint of Object.keys(ENDPOINTS)) {
      await queen.queue(DELIVERIES).partition(endpoint).push({
        // The event id makes the enqueue idempotent: an application that
        // retries its own emit does not create a second delivery.
        transactionId: `${endpoint}-evt-${seq}`,
        data: { endpoint, seq, type: 'invoice.paid', invoiceId: `INV-${seq}` },
      })
    }
  }
  console.log(`  ${EVENTS_PER_ENDPOINT * Object.keys(ENDPOINTS).length} deliveries queued`)

  // ------------------------------------------------------------------ sending
  //
  // The sender pool. autoAck is on, so a handler that returns acknowledges the
  // delivery and a handler that throws nacks it: the broker then redelivers it
  // until the retry budget is gone. That is the whole retry mechanism, and it
  // survives the sender process dying mid-flight, which a loop inside the
  // handler would not.
  console.log('\nsending')
  const deliveredTo = new Map()
  const attempts = new Map()

  await queen
    .queue(DELIVERIES)
    .group(GROUP)
    .subscriptionMode('all')
    .concurrency(3)
    .each()
    // Enough turns for every good delivery plus every attempt at the bad ones.
    .limit(EVENTS_PER_ENDPOINT * 2 + EVENTS_PER_ENDPOINT * (RETRY_LIMIT + 1))
    .idleMillis(6000)
    .consume(async (msg) => {
      const { endpoint, seq } = msg.data
      attempts.set(endpoint, (attempts.get(endpoint) ?? 0) + 1)

      // There is no attempt counter to read here: the broker's pop response
      // carries none, so the message does not either. The budget is the
      // broker's, and throwing is how one attempt of it is spent. A sender that
      // recognises a permanent error can skip the attempts it has left by
      // dead-lettering the message itself: autoAck(false), then
      // `await queen.ack(msg, 'dlq', { group: GROUP, error: reason })`, which
      // writes the same dead-letter row the checks below read.
      await postToEndpoint(endpoint, msg.data)

      const seen = deliveredTo.get(endpoint) ?? []
      seen.push(seq)
      deliveredTo.set(endpoint, seen)
      console.log(`  ${endpoint} <- event ${seq}`)
    })

  // ------------------------------------------------------------------ checking
  console.log('\nchecking')

  for (const [endpoint, meta] of Object.entries(ENDPOINTS)) {
    if (!meta.healthy) continue
    const seqs = deliveredTo.get(endpoint) ?? []
    assert(seqs.length === EVENTS_PER_ENDPOINT, `${endpoint} received all ${EVENTS_PER_ENDPOINT} events`)
    assert(
      JSON.stringify(seqs) === JSON.stringify([1, 2, 3]),
      `${endpoint} received them in the order they happened`
    )
  }

  assert(
    (deliveredTo.get('initech.example') ?? []).length === 0,
    'the dead endpoint received nothing, as it should'
  )
  assert(
    attempts.get('initech.example') > EVENTS_PER_ENDPOINT,
    'the dead endpoint was retried rather than dropped on the first failure'
  )

  // The dead-letter queue is a table you can read, not a log line. Each row
  // carries the payload, the endpoint it was for, and the last error, which is
  // what a support engineer needs to answer "why did this customer not get it".
  const dlq = await queen.queue(DELIVERIES).dlq().limit(50).get()
  const dead = dlq.messages.filter(m => m.data.endpoint === 'initech.example')

  assert(dead.length === EVENTS_PER_ENDPOINT, `all ${EVENTS_PER_ENDPOINT} dead deliveries are in the dead-letter queue`)
  assert(
    dead.every(m => (m.errorMessage ?? '').includes('answered 500')),
    'each dead-letter row carries the error that killed it'
  )
  assert(
    dlq.messages.every(m => m.data.endpoint === 'initech.example'),
    'no healthy endpoint put anything in the dead-letter queue'
  )

  console.log(`\n  dead letters: ${dead.map(m => `${m.data.endpoint}/${m.data.invoiceId}`).join(', ')}`)

  await queen.queue(DELIVERIES).delete()
  console.log(`\nPASS: ${checks} checks`)
} catch (err) {
  console.error(`\nFAIL: ${err.message}`)
  process.exitCode = 1
} finally {
  await queen.close()
}
```
### Python

```python title="examples/apps/py/webhooks.py"
#
# A webhook delivery system.
#
# Every SaaS product ends up writing this one, and it is harder than it looks:
# deliveries to one customer's endpoint must arrive in order, a customer whose
# endpoint is down must not slow down anybody else's, failures must be retried
# a bounded number of times, and what never succeeds has to end up somewhere a
# human can look at.
#
# The shape here is one ordered lane per destination, created by the first
# delivery to it. A dead endpoint backs up its own lane and no other; retries
# are the broker's retry budget rather than a loop in your code; and what
# exhausts the budget lands in the dead-letter queue with the error attached.
#
#   webhook-deliveries (one partition per destination)
#     `-- group "sender"  posts each delivery, fails on a dead endpoint
#           `-- retry_limit exhausted -> dead-letter queue
#
# Run it:
#   QUEEN_URL=http://localhost:6632 python3 webhooks.py

import asyncio
import os
import sys
import time

from queen import Queen

QUEEN_URL = os.environ.get("QUEEN_URL", "http://localhost:6632")

# The name is prefixed per language and suffixed per run, so every application
# in every language can share one broker and no run inherits state from another.
RUN = f"{int(time.time() * 1000):x}"
DELIVERIES = f"app-py-webhooks-{RUN}"
GROUP = "sender"

# Three subscribers. One of them has let its certificate expire, which is the
# most common way a webhook endpoint dies: it answers, but it answers 500.
ENDPOINTS = {
    "acme.example": {"healthy": True},
    "globex.example": {"healthy": True},
    "initech.example": {"healthy": False},
}
EVENTS_PER_ENDPOINT = 3
RETRY_LIMIT = 2

CHECKS = 0


def check(condition: bool, description: str) -> None:
    """Record one verified fact, or abort the run.

    This raises instead of using the `assert` statement, because `python3 -O`
    removes `assert` and the checks are the whole point of the program.
    """
    global CHECKS
    if not condition:
        raise AssertionError(description)
    CHECKS += 1
    print(f"  ok: {description}")


async def post_to_endpoint(endpoint: str, event: dict) -> dict:
    """Stand in for the HTTP POST to the subscriber.

    A real sender would use httpx and treat any non-2xx as a failure, which is
    exactly what raising does here.
    """
    if not ENDPOINTS[endpoint]["healthy"]:
        raise RuntimeError(f"{endpoint} answered 500")
    return {"status": 200}


async def main() -> int:
    # The whole client is async: every call below is awaited, and this is the
    # one event loop they all run on. Unlike the JavaScript client there is no
    # handleSignals switch, so SIGINT and SIGTERM are always handled for you.
    queen = Queen(url=QUEEN_URL)
    verdict, failed = "", False

    try:
        print(f"broker {QUEEN_URL}")

        # retry_limit is the delivery budget, and dlq_after_max_retries is what
        # happens when it runs out. Without the second flag an exhausted message
        # is simply marked failed and stays put; with it, the broker moves it to
        # the dead-letter table with the last error on the row.
        #
        # lease_time is the other half of the contract: it is how long the
        # broker waits for a sender that took a delivery and never came back
        # before handing that delivery to someone else. The config keys are
        # snake_case in Python and the client converts them to the camelCase the
        # broker expects.
        await queen.queue(DELIVERIES).config(
            {"lease_time": 30, "retry_limit": RETRY_LIMIT, "dlq_after_max_retries": True}
        ).create()

        # ------------------------------------------------------------ queuing
        #
        # The application emits events. Each one goes into the partition of the
        # endpoint it is destined for, which is what makes "in order per
        # subscriber" a property of the storage rather than of the sender.
        print("\nqueuing deliveries")
        for seq in range(1, EVENTS_PER_ENDPOINT + 1):
            for endpoint in ENDPOINTS:
                # The event id makes the enqueue idempotent: an application that
                # retries its own emit does not create a second delivery. The
                # item key stays camelCase here, because it is the wire name
                # rather than a client option.
                await queen.queue(DELIVERIES).partition(endpoint).push(
                    {
                        "transactionId": f"{endpoint}-evt-{seq}",
                        "data": {
                            "endpoint": endpoint,
                            "seq": seq,
                            "type": "invoice.paid",
                            "invoiceId": f"INV-{seq}",
                        },
                    }
                )
        print(f"  {EVENTS_PER_ENDPOINT * len(ENDPOINTS)} deliveries queued")

        # ------------------------------------------------------------ sending
        #
        # The sender pool. auto_ack is on by default, so a handler that returns
        # acknowledges the delivery and a handler that raises nacks it: the
        # broker then redelivers it until the retry budget is gone. That is the
        # whole retry mechanism, and it survives the sender process dying
        # mid-flight, which a loop inside the handler would not.
        print("\nsending")
        delivered_to: dict = {}
        attempts: dict = {}

        async def send(msg) -> None:
            endpoint = msg["data"]["endpoint"]
            seq = msg["data"]["seq"]
            attempts[endpoint] = attempts.get(endpoint, 0) + 1

            # There is no attempt counter to read here: the broker's pop
            # response carries none, so the message dict has none either. The
            # budget is the broker's, and raising is how one attempt of it is
            # spent. A sender that recognises a permanent error can skip the
            # attempts it has left by dead-lettering the message itself --
            # auto_ack(False), then
            # `await queen.ack(msg, "dlq", {"group": GROUP, "error": reason})`
            # -- which writes the same dead-letter row the checks below read.
            await post_to_endpoint(endpoint, msg["data"])

            delivered_to.setdefault(endpoint, []).append(seq)
            print(f"  {endpoint} <- event {seq}")

        await (
            queen.queue(DELIVERIES)
            .group(GROUP)
            # A group created after the messages were pushed starts at the tail,
            # so without this it would see nothing.
            .subscription_mode("all")
            .concurrency(3)
            .each()
            # Enough turns for every good delivery plus every attempt at the bad
            # ones.
            .limit(EVENTS_PER_ENDPOINT * 2 + EVENTS_PER_ENDPOINT * (RETRY_LIMIT + 1))
            # Stop after 6s of silence, so a stuck delivery fails the run instead
            # of hanging it.
            .idle_millis(6000)
            .consume(send)
        )

        # ----------------------------------------------------------- checking
        print("\nchecking")

        for endpoint, meta in ENDPOINTS.items():
            if not meta["healthy"]:
                continue
            seqs = delivered_to.get(endpoint, [])
            check(
                len(seqs) == EVENTS_PER_ENDPOINT,
                f"{endpoint} received all {EVENTS_PER_ENDPOINT} events",
            )
            check(seqs == [1, 2, 3], f"{endpoint} received them in the order they happened")

        check(
            len(delivered_to.get("initech.example", [])) == 0,
            "the dead endpoint received nothing, as it should",
        )
        check(
            attempts.get("initech.example", 0) > EVENTS_PER_ENDPOINT,
            "the dead endpoint was retried rather than dropped on the first failure",
        )

        # The dead-letter queue is a table you can read, not a log line. Each row
        # carries the payload, the endpoint it was for, and the last error, which
        # is what a support engineer needs to answer "why did this customer not
        # get it". The rows come back as plain dicts, with the payload under
        # "data" and the error under the broker's own wire name, "errorMessage".
        dlq = await queen.queue(DELIVERIES).dlq().limit(50).get()
        dead = [m for m in dlq["messages"] if m["data"]["endpoint"] == "initech.example"]

        check(
            len(dead) == EVENTS_PER_ENDPOINT,
            f"all {EVENTS_PER_ENDPOINT} dead deliveries are in the dead-letter queue",
        )
        check(
            all("answered 500" in (m.get("errorMessage") or "") for m in dead),
            "each dead-letter row carries the error that killed it",
        )
        check(
            all(m["data"]["endpoint"] == "initech.example" for m in dlq["messages"]),
            "no healthy endpoint put anything in the dead-letter queue",
        )

        listing = ", ".join(f"{m['data']['endpoint']}/{m['data']['invoiceId']}" for m in dead)
        print(f"\n  dead letters: {listing}")

        # Clean up on success only: a failed run leaves the queue on the broker
        # to be looked at.
        await queen.queue(DELIVERIES).delete()

        verdict = f"\nPASS: {CHECKS} checks"
    except Exception as err:
        verdict, failed = f"\nFAIL: {err}", True
    finally:
        # close() flushes the client-side buffers and closes the HTTP pool. It
        # narrates its own shutdown on stdout, which is why the verdict is
        # printed after it rather than before: PASS or FAIL stays the last line
        # of a run.
        await queen.close()

    # A failure goes to stderr, like the rest of the set. Flush stdout first so
    # the verdict still lands last when the two are piped into one file.
    sys.stdout.flush()
    print(verdict, file=sys.stderr if failed else sys.stdout)
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))
```
### Go

```go title="examples/apps/go/webhooks/main.go"
//
// A webhook delivery system.
//
// Every SaaS product ends up writing this one, and it is harder than it looks:
// deliveries to one customer's endpoint must arrive in order, a customer whose
// endpoint is down must not slow down anybody else's, failures must be retried
// a bounded number of times, and what never succeeds has to end up somewhere a
// human can look at.
//
// The shape here is one ordered lane per destination, created by the first
// delivery to it. A dead endpoint backs up its own lane and no other; retries
// are the broker's retry budget rather than a loop in your code; and what
// exhausts the budget lands in the dead-letter queue with the error attached.
//
//	webhook-deliveries (one partition per destination)
//	  |-- group "sender"  posts each delivery, fails on a dead endpoint
//	        |-- retryLimit exhausted -> dead-letter queue
//
// Run it:
//
//	QUEEN_URL=http://localhost:6632 GOWORK=off go run ./webhooks
package main

import (
	"context"
	"fmt"
	"os"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"

	queen "github.com/smartpricing/queen/clients/client-go"
)

var runID = strconv.FormatInt(time.Now().UnixMilli(), 36)

var deliveriesQueue = "app-go-webhooks-" + runID

const group = "sender"

// Three subscribers. One of them has let its certificate expire, which is the
// most common way a webhook endpoint dies: it answers, but it answers 500. The
// list is a slice rather than a map so the queuing order below is the same on
// every run, which Go map iteration would not give.
type endpoint struct {
	host    string
	healthy bool
}

var endpoints = []endpoint{
	{host: "acme.example", healthy: true},
	{host: "globex.example", healthy: true},
	{host: "initech.example", healthy: false},
}

const (
	eventsPerEndpoint = 3
	retryLimit        = 2
	deadEndpoint      = "initech.example"
)

func isHealthy(host string) bool {
	for _, e := range endpoints {
		if e.host == host {
			return e.healthy
		}
	}
	return false
}

// postToEndpoint stands in for the HTTP POST to the subscriber. A real sender
// would use net/http and treat any non-2xx as a failure, which is exactly what
// returning an error does here: with auto-ack on, an error from the handler is
// what the client turns into a negative acknowledgement.
func postToEndpoint(host string, event map[string]interface{}) error {
	if !isHealthy(host) {
		return fmt.Errorf("%s answered 500", host)
	}
	return nil
}

var checks int

func assert(condition bool, description string) error {
	if !condition {
		return fmt.Errorf("%s", description)
	}
	checks++
	fmt.Printf("  ok: %s\n", description)
	return nil
}

func main() {
	if err := run(); err != nil {
		fmt.Fprintf(os.Stderr, "\nFAIL: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("\nPASS: %d checks\n", checks)
}

func run() error {
	brokerURL := os.Getenv("QUEEN_URL")
	if brokerURL == "" {
		brokerURL = "http://localhost:6632"
	}

	// One context bounds the whole program: a broker that stops answering ends
	// the run instead of wedging it.
	ctx, cancel := context.WithTimeout(context.Background(), 180*time.Second)
	defer cancel()

	client, err := queen.New(brokerURL)
	if err != nil {
		return fmt.Errorf("create client: %w", err)
	}
	defer client.Close(context.Background())

	fmt.Printf("broker %s\n", brokerURL)

	// RetryLimit is the delivery budget, and DlqAfterMaxRetries is what happens
	// when it runs out. Without the second flag an exhausted message is simply
	// marked failed and stays put; with it, the broker moves it to the
	// dead-letter table with the last error on the row.
	//
	// LeaseTime is the other half of the contract: it is how long the broker
	// waits for a sender that took a delivery and never came back before
	// handing that delivery to someone else.
	if _, err := client.Queue(deliveriesQueue).
		Config(queen.QueueConfig{
			LeaseTime:          30,
			RetryLimit:         retryLimit,
			DlqAfterMaxRetries: true,
		}).
		Create().Execute(ctx); err != nil {
		return fmt.Errorf("create %s: %w", deliveriesQueue, err)
	}

	// ------------------------------------------------------------------ queuing
	//
	// The application emits events. Each one goes into the partition of the
	// endpoint it is destined for, which is what makes "in order per subscriber"
	// a property of the storage rather than of the sender.
	fmt.Println("\nqueuing deliveries")
	for seq := 1; seq <= eventsPerEndpoint; seq++ {
		for _, e := range endpoints {
			// The event id makes the enqueue idempotent: an application that
			// retries its own emit does not create a second delivery. This
			// client carries it on the push builder, not in the payload.
			if _, err := client.Queue(deliveriesQueue).
				Partition(e.host).
				Push(map[string]interface{}{
					"endpoint":  e.host,
					"seq":       seq,
					"type":      "invoice.paid",
					"invoiceId": fmt.Sprintf("INV-%d", seq),
				}).
				TransactionID(fmt.Sprintf("%s-evt-%d", e.host, seq)).
				Execute(ctx); err != nil {
				return fmt.Errorf("queue %s/%d: %w", e.host, seq, err)
			}
		}
	}
	fmt.Printf("  %d deliveries queued\n", eventsPerEndpoint*len(endpoints))

	// ------------------------------------------------------------------ sending
	//
	// The sender pool. Auto-ack is the default here, so a handler that returns
	// nil acknowledges the delivery and one that returns an error nacks it: the
	// broker then redelivers it until the retry budget is gone. That is the
	// whole retry mechanism, and it survives the sender process dying
	// mid-flight, which a loop inside the handler would not.
	fmt.Println("\nsending")
	var mu sync.Mutex
	deliveredTo := map[string][]int{}
	attempts := map[string]int{}

	err = client.Queue(deliveriesQueue).
		Group(group).
		SubscriptionMode(queen.SubscriptionModeAll).
		Concurrency(3).
		Each().
		// Enough turns for every good delivery plus every attempt at the bad
		// ones. Limit is per worker rather than a budget shared by the pool,
		// so it is a ceiling on a runaway goroutine; what actually
		// ends the pool is the idle bound, once the healthy lanes are drained
		// and the dead one has burnt its budget into the dead-letter queue.
		// TimeoutMillis caps each poll at a second so that bound is noticed
		// promptly instead of inside a 30 s long poll.
		Limit(eventsPerEndpoint*2+eventsPerEndpoint*(retryLimit+1)).
		IdleMillis(6000).
		TimeoutMillis(1000).
		Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
			host, _ := msg.Data["endpoint"].(string)
			seq, ok := msg.Data["seq"].(float64)
			if !ok {
				return fmt.Errorf("delivery %s has no numeric seq", msg.TransactionID)
			}

			// Three poll loops means three goroutines in this handler, so the
			// bookkeeping is behind a mutex. The JavaScript version needs no
			// lock because it has no threads.
			mu.Lock()
			attempts[host]++
			mu.Unlock()

			// msg.RetryCount exists on the struct but stays zero here: the
			// broker's pop response carries no attempt counter, and the field is
			// only filled on a dead-letter read. The budget is the broker's, and
			// returning an error is how one attempt of it is spent. A sender
			// that recognises a permanent error can skip the attempts it has
			// left only through the HTTP ack route, which takes a status string:
			// this client's Ack takes a bool, completed or failed, so unlike the
			// JavaScript one it cannot mark a delivery dead on the spot.
			if err := postToEndpoint(host, msg.Data); err != nil {
				return err
			}

			mu.Lock()
			deliveredTo[host] = append(deliveredTo[host], int(seq))
			mu.Unlock()
			fmt.Printf("  %s <- event %d\n", host, int(seq))
			return nil
		}).
		Execute(ctx)
	if err != nil {
		return fmt.Errorf("sending: %w", err)
	}

	// ------------------------------------------------------------------ checking
	fmt.Println("\nchecking")

	for _, e := range endpoints {
		if !e.healthy {
			continue
		}
		seqs := deliveredTo[e.host]
		if err := assert(
			len(seqs) == eventsPerEndpoint,
			fmt.Sprintf("%s received all %d events", e.host, eventsPerEndpoint),
		); err != nil {
			return err
		}
		if err := assert(
			slices.Equal(seqs, []int{1, 2, 3}),
			fmt.Sprintf("%s received them in the order they happened", e.host),
		); err != nil {
			return err
		}
	}

	if err := assert(
		len(deliveredTo[deadEndpoint]) == 0,
		"the dead endpoint received nothing, as it should",
	); err != nil {
		return err
	}
	if err := assert(
		attempts[deadEndpoint] > eventsPerEndpoint,
		"the dead endpoint was retried rather than dropped on the first failure",
	); err != nil {
		return err
	}

	// The dead-letter queue is a table you can read, not a log line. Each row
	// carries the payload, the endpoint it was for, and the last error, which is
	// what a support engineer needs to answer "why did this customer not get
	// it". DLQ takes a consumer group to filter by; empty means every group on
	// this queue, which is what the check below wants.
	dlq, err := client.Queue(deliveriesQueue).DLQ("").Limit(50).Get(ctx)
	if err != nil {
		return fmt.Errorf("read dead letters: %w", err)
	}

	var dead []queen.Message
	for _, m := range dlq.Messages {
		if host, _ := m.Data["endpoint"].(string); host == deadEndpoint {
			dead = append(dead, m)
		}
	}

	if err := assert(
		len(dead) == eventsPerEndpoint,
		fmt.Sprintf("all %d dead deliveries are in the dead-letter queue", eventsPerEndpoint),
	); err != nil {
		return err
	}

	carriesError := true
	for _, m := range dead {
		if !strings.Contains(m.ErrorMessage, "answered 500") {
			carriesError = false
		}
	}
	if err := assert(carriesError, "each dead-letter row carries the error that killed it"); err != nil {
		return err
	}

	onlyDead := true
	for _, m := range dlq.Messages {
		if host, _ := m.Data["endpoint"].(string); host != deadEndpoint {
			onlyDead = false
		}
	}
	if err := assert(onlyDead, "no healthy endpoint put anything in the dead-letter queue"); err != nil {
		return err
	}

	names := make([]string, 0, len(dead))
	for _, m := range dead {
		host, _ := m.Data["endpoint"].(string)
		invoice, _ := m.Data["invoiceId"].(string)
		names = append(names, host+"/"+invoice)
	}
	fmt.Printf("\n  dead letters: %s\n", strings.Join(names, ", "))

	// Clean up on success only: a failed run leaves the queue and its dead
	// letters on the broker to be looked at.
	if _, err := client.Queue(deliveriesQueue).Delete().Execute(ctx); err != nil {
		return fmt.Errorf("delete %s: %w", deliveriesQueue, err)
	}

	return nil
}
```
### Rust

```rust title="examples/apps/rust/src/bin/webhooks.rs"
//
// A webhook delivery system.
//
// Every SaaS product ends up writing this one, and it is harder than it looks:
// deliveries to one customer's endpoint must arrive in order, a customer whose
// endpoint is down must not slow down anybody else's, failures must be retried
// a bounded number of times, and what never succeeds has to end up somewhere a
// human can look at.
//
// The shape here is one ordered lane per destination, created by the first
// delivery to it. A dead endpoint backs up its own lane and no other; retries
// are the broker's retry budget rather than a loop in your code; and what
// exhausts the budget lands in the dead-letter queue with the error attached.
//
//   webhook-deliveries (one partition per destination)
//     └── group "sender"  posts each delivery, fails on a dead endpoint
//           └── retry_limit exhausted -> dead-letter queue
//
// Run it:
//   QUEEN_URL=http://localhost:6632 cargo run --bin webhooks

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use queen_mq::{Config, Message, PushItem, Queen, QueueOptions, SubscriptionMode};
use serde_json::json;

const GROUP: &str = "sender";

// Three subscribers. One of them has let its certificate expire, which is the
// most common way a webhook endpoint dies: it answers, but it answers 500.
//
// (endpoint, healthy)
const ENDPOINTS: [(&str, bool); 3] = [
    ("acme.example", true),
    ("globex.example", true),
    ("initech.example", false),
];
const EVENTS_PER_ENDPOINT: i64 = 3;
const RETRY_LIMIT: i32 = 2;

fn healthy(endpoint: &str) -> bool {
    ENDPOINTS
        .iter()
        .find(|(name, _)| *name == endpoint)
        .map(|(_, ok)| *ok)
        .unwrap_or(false)
}

/// Stands in for the HTTP POST to the subscriber. A real sender would use an
/// HTTP client and treat any non-2xx as a failure — which is exactly what
/// returning Err does here, since this client's consume loop nacks on Err and
/// records the reason.
async fn post_to_endpoint(endpoint: &str) -> Result<(), String> {
    if !healthy(endpoint) {
        return Err(format!("{endpoint} answered 500"));
    }
    Ok(())
}

struct Checks(usize);

impl Checks {
    fn assert(&mut self, condition: bool, description: &str) -> Result<(), String> {
        if !condition {
            return Err(description.to_string());
        }
        self.0 += 1;
        println!("  ok: {description}");
        Ok(())
    }
}

// Rust has no exceptions, so the shape the JavaScript gets from try/catch comes
// from `run` returning a Result: every `?` on the way down is a failed check or
// a failed call, and main turns it into FAIL and a non-zero exit.
#[tokio::main]
async fn main() {
    match run().await {
        Ok(checks) => println!("\nPASS: {checks} checks"),
        Err(e) => {
            eprintln!("\nFAIL: {e}");
            std::process::exit(1);
        }
    }
}

async fn run() -> Result<usize, String> {
    let url = std::env::var("QUEEN_URL").unwrap_or_else(|_| "http://localhost:6632".into());
    let run_id = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_millis();
    let deliveries = format!("app-rust-webhooks-{run_id}");

    let mut checks = Checks(0);
    println!("broker {url}");

    let queen = Queen::connect(Config::new(&url)).map_err(|e| e.to_string())?;

    // retry_limit is the delivery budget, and dlq_after_max_retries is what
    // happens when it runs out. Without the second flag an exhausted message is
    // simply marked failed and stays put; with it, the broker moves it to the
    // dead-letter table with the last error on the row.
    //
    // lease_time is the other half of the contract: it is how long the broker
    // waits for a sender that took a delivery and never came back before handing
    // that delivery to someone else.
    queen
        .queue(&deliveries)
        .configure(QueueOptions {
            lease_time: Some(30),
            retry_limit: Some(RETRY_LIMIT),
            dlq_after_max_retries: Some(true),
            ..Default::default()
        })
        .await
        .map_err(|e| e.to_string())?;

    // ------------------------------------------------------------------ queuing
    //
    // The application emits events. Each one goes into the partition of the
    // endpoint it is destined for, which is what makes "in order per subscriber"
    // a property of the storage rather than of the sender.
    println!("\nqueuing deliveries");
    for seq in 1..=EVENTS_PER_ENDPOINT {
        for (endpoint, _) in ENDPOINTS {
            // push() would mint a UUIDv7 transaction id; here the event id is
            // the idempotency key, so the item is built by hand and pushed
            // through push_items(). An application that retries its own emit
            // does not create a second delivery.
            queen
                .queue(&deliveries)
                .partition(endpoint)
                .push_items(vec![PushItem::new(
                    &deliveries,
                    json!({
                        "endpoint": endpoint,
                        "seq": seq,
                        "type": "invoice.paid",
                        "invoiceId": format!("INV-{seq}"),
                    }),
                )
                .partition(endpoint)
                .transaction_id(format!("{endpoint}-evt-{seq}"))])
                .await
                .map_err(|e| e.to_string())?;
        }
    }
    println!(
        "  {} deliveries queued",
        EVENTS_PER_ENDPOINT as usize * ENDPOINTS.len()
    );

    // ------------------------------------------------------------------ sending
    //
    // The sender pool. auto_ack is on by default, so a handler that returns Ok
    // acknowledges the delivery and a handler that returns Err nacks it with the
    // error as the reason: the broker then redelivers it until the retry budget
    // is gone. That is the whole retry mechanism, and it survives the sender
    // process dying mid-flight, which a loop inside the handler would not.
    println!("\nsending");
    let delivered_to: Arc<Mutex<HashMap<String, Vec<i64>>>> = Arc::new(Mutex::new(HashMap::new()));
    let attempts: Arc<Mutex<HashMap<String, usize>>> = Arc::new(Mutex::new(HashMap::new()));

    {
        let delivered_to = Arc::clone(&delivered_to);
        let attempts = Arc::clone(&attempts);
        queen
            .queue(&deliveries)
            .group(GROUP)
            .subscription_mode(SubscriptionMode::All)
            .concurrency(3)
            // Enough turns for every good delivery plus every attempt at the bad
            // ones. `limit` counts across all three workers, not per worker.
            .limit(
                (EVENTS_PER_ENDPOINT * 2 + EVENTS_PER_ENDPOINT * (RETRY_LIMIT as i64 + 1)) as u64,
            )
            // The idle timeout is checked between polls, so the poll window
            // bounds how promptly it fires: a redelivery that never arrives ends
            // the run rather than hanging it.
            .poll_timeout(Duration::from_secs(1))
            .idle(Duration::from_secs(6))
            .consume(move |msg: Message| {
                let delivered_to = Arc::clone(&delivered_to);
                let attempts = Arc::clone(&attempts);
                async move {
                    let endpoint = msg.data["endpoint"]
                        .as_str()
                        .unwrap_or_default()
                        .to_string();
                    let seq = msg.data["seq"].as_i64().unwrap_or(0);
                    *attempts
                        .lock()
                        .unwrap()
                        .entry(endpoint.clone())
                        .or_insert(0) += 1;

                    // A failed POST returns Err, and the loop turns that into a
                    // nack carrying this string — which is what ends up on the
                    // dead-letter row when the budget runs out.
                    //
                    // A delivered Message carries no attempt counter to back off
                    // on, here or in any other client: the broker's pop response
                    // has no such field. The broker counts the attempts itself
                    // and publishes the total only on the dead-letter row, as
                    // `retryCount` — on a DlqMessage that lands in `rest`, the
                    // map holding every key the struct does not name. So the
                    // budget is the broker's, and a sender that wants to give up
                    // early on an error it knows is permanent says so by
                    // dead-lettering the message itself (ack_with(&msg,
                    // AckStatus::Dlq, ...)) rather than by counting attempts.
                    post_to_endpoint(&endpoint).await?;

                    delivered_to
                        .lock()
                        .unwrap()
                        .entry(endpoint.clone())
                        .or_default()
                        .push(seq);
                    println!("  {endpoint} <- event {seq}");
                    Ok::<_, String>(())
                }
            })
            .await
            .map_err(|e| e.to_string())?;
    }

    // ------------------------------------------------------------------ checking
    println!("\nchecking");

    let delivered_to = delivered_to.lock().unwrap().clone();
    let attempts = attempts.lock().unwrap().clone();

    for (endpoint, ok) in ENDPOINTS {
        if !ok {
            continue;
        }
        let seqs = delivered_to.get(endpoint).cloned().unwrap_or_default();
        checks.assert(
            seqs.len() == EVENTS_PER_ENDPOINT as usize,
            &format!("{endpoint} received all {EVENTS_PER_ENDPOINT} events"),
        )?;
        checks.assert(
            seqs == [1, 2, 3],
            &format!("{endpoint} received them in the order they happened"),
        )?;
    }

    checks.assert(
        delivered_to
            .get("initech.example")
            .map(|s| s.len())
            .unwrap_or(0)
            == 0,
        "the dead endpoint received nothing, as it should",
    )?;
    checks.assert(
        attempts.get("initech.example").copied().unwrap_or(0) > EVENTS_PER_ENDPOINT as usize,
        "the dead endpoint was retried rather than dropped on the first failure",
    )?;

    // The dead-letter queue is a table you can read, not a log line. Each row
    // carries the payload, the endpoint it was for, and the last error, which is
    // what a support engineer needs to answer "why did this customer not get
    // it". This client takes the page size as an argument rather than as another
    // builder step, because those are the only two filters the broker honours
    // besides the queue and the group.
    let dlq = queen
        .queue(&deliveries)
        .dlq(Some(50), None)
        .await
        .map_err(|e| e.to_string())?;
    let dead: Vec<_> = dlq
        .messages
        .iter()
        .filter(|m| m.data["endpoint"] == json!("initech.example"))
        .collect();

    checks.assert(
        dead.len() == EVENTS_PER_ENDPOINT as usize,
        &format!("all {EVENTS_PER_ENDPOINT} dead deliveries are in the dead-letter queue"),
    )?;
    checks.assert(
        dead.iter()
            .all(|m| m.error.as_deref().unwrap_or("").contains("answered 500")),
        "each dead-letter row carries the error that killed it",
    )?;
    checks.assert(
        dlq.messages
            .iter()
            .all(|m| m.data["endpoint"] == json!("initech.example")),
        "no healthy endpoint put anything in the dead-letter queue",
    )?;

    let listed: Vec<String> = dead
        .iter()
        .map(|m| {
            format!(
                "{}/{}",
                m.data["endpoint"].as_str().unwrap_or("?"),
                m.data["invoiceId"].as_str().unwrap_or("?")
            )
        })
        .collect();
    println!("\n  dead letters: {}", listed.join(", "));

    // Clean up on success only: a failed run leaves the queue, and its DLQ rows,
    // on the broker to be looked at.
    queen
        .queue(&deliveries)
        .delete()
        .await
        .map_err(|e| e.to_string())?;

    queen.close().await.map_err(|e| e.to_string())?;

    Ok(checks.0)
}
```
### PHP

```php title="examples/apps/php/webhooks.php"
//
// A webhook delivery system.
//
// Every SaaS product ends up writing this one, and it is harder than it looks:
// deliveries to one customer's endpoint must arrive in order, a customer whose
// endpoint is down must not slow down anybody else's, failures must be retried
// a bounded number of times, and what never succeeds has to end up somewhere a
// human can look at.
//
// The shape here is one ordered lane per destination, created by the first
// delivery to it. A dead endpoint backs up its own lane and no other; retries
// are the broker's retry budget rather than a loop in your code; and what
// exhausts the budget lands in the dead-letter queue with the error attached.
//
//   webhook-deliveries (one partition per destination)
//     └── group "sender"  posts each delivery, fails on a dead endpoint
//           └── retryLimit exhausted -> dead-letter queue
//
// Run it:
//   QUEEN_URL=http://localhost:6632 php webhooks.php

require __DIR__ . '/vendor/autoload.php';

use Queen\Queen;

$QUEEN_URL = getenv('QUEEN_URL') ?: 'http://localhost:6632';
$RUN = base_convert((string) (int) (microtime(true) * 1000), 10, 36);
$DELIVERIES = "app-php-webhooks-{$RUN}";
$GROUP = 'sender';

// Three subscribers. One of them has let its certificate expire, which is the
// most common way a webhook endpoint dies: it answers, but it answers 500.
$ENDPOINTS = [
    'acme.example' => ['healthy' => true],
    'globex.example' => ['healthy' => true],
    'initech.example' => ['healthy' => false],
];
$EVENTS_PER_ENDPOINT = 3;
$RETRY_LIMIT = 2;

$checks = 0;
$assert = function (bool $condition, string $description) use (&$checks): void {
    if (!$condition) {
        throw new RuntimeException($description);
    }
    $checks++;
    echo "  ok: {$description}\n";
};

// Stands in for the HTTP POST to the subscriber. A real sender would use Guzzle
// and treat any non-2xx as a failure, which is exactly what throwing does here.
$postToEndpoint = function (string $endpoint, array $event) use ($ENDPOINTS): array {
    if (!$ENDPOINTS[$endpoint]['healthy']) {
        throw new RuntimeException("{$endpoint} answered 500");
    }
    return ['status' => 200];
};

$queen = new Queen($QUEEN_URL);
$exitCode = 0;

try {
    echo "broker {$QUEEN_URL}\n";

    // retryLimit is the delivery budget, and dlqAfterMaxRetries is what happens
    // when it runs out. Without the second flag an exhausted message is simply
    // marked failed and stays put; with it, the broker moves it to the
    // dead-letter table with the last error on the row.
    //
    // leaseTime is the other half of the contract: it is how long the broker
    // waits for a sender that took a delivery and never came back before handing
    // that delivery to someone else.
    //
    // config() fills in the queue defaults around whatever you name here, so the
    // three keys below are the whole configuration decision.
    $queen->queue($DELIVERIES)->config([
        'leaseTime' => 30,
        'retryLimit' => $RETRY_LIMIT,
        'dlqAfterMaxRetries' => true,
    ])->create()->execute();

    // ------------------------------------------------------------------ queuing
    //
    // The application emits events. Each one goes into the partition of the
    // endpoint it is destined for, which is what makes "in order per subscriber"
    // a property of the storage rather than of the sender.
    echo "\nqueuing deliveries\n";
    for ($seq = 1; $seq <= $EVENTS_PER_ENDPOINT; $seq++) {
        foreach (array_keys($ENDPOINTS) as $endpoint) {
            $queen->queue($DELIVERIES)->partition($endpoint)->push([[
                // The event id makes the enqueue idempotent: an application that
                // retries its own emit does not create a second delivery.
                'transactionId' => "{$endpoint}-evt-{$seq}",
                'data' => [
                    'endpoint' => $endpoint,
                    'seq' => $seq,
                    'type' => 'invoice.paid',
                    'invoiceId' => "INV-{$seq}",
                ],
            ]])->execute();
        }
    }
    echo '  ' . ($EVENTS_PER_ENDPOINT * count($ENDPOINTS)) . " deliveries queued\n";

    // ------------------------------------------------------------------ sending
    //
    // The sender pool. concurrency(3) is three long polls in flight at once on
    // one cURL multi-handle rather than three threads, and each poll claims a
    // partition of its own, so the three destinations are drained side by side.
    //
    // autoAck is off, and that is a deliberate choice rather than a formality.
    // The automatic path in this client nacks a throwing handler with a status
    // and nothing else: the delivery would be retried and eventually dead-
    // lettered exactly the same way, but the row a support engineer opens would
    // have an empty error. Acknowledging by hand is what puts the reason on it.
    // Everything else about retrying is unchanged: the budget is the broker's,
    // and it survives this process dying mid-flight, which a loop inside the
    // handler would not.
    //
    // timeoutMillis(1000) caps how long one poll parks on the broker. A round
    // ends only when every worker's poll has come back, so with the 30 s default
    // the last round of a drained queue would sit there for half a minute before
    // the idle bound could fire.
    echo "\nsending\n";
    $deliveredTo = [];
    $attempts = [];

    $queen
        ->queue($DELIVERIES)
        ->group($GROUP)
        ->subscriptionMode('all')
        ->concurrency(3)
        ->each()
        ->autoAck(false)
        // Enough turns for every good delivery plus every attempt at the bad
        // ones. This client spreads the bound over the pool, so each of the
        // three workers stops after a third of it, and the thirds add up to the
        // same total: no delivery is left without a worker allowed to take it.
        ->limit($EVENTS_PER_ENDPOINT * 2 + $EVENTS_PER_ENDPOINT * ($RETRY_LIMIT + 1))
        ->idleMillis(6000)
        ->timeoutMillis(1000)
        ->consume(function (array $msg) use ($queen, $postToEndpoint, $GROUP, &$deliveredTo, &$attempts): void {
            $endpoint = $msg['data']['endpoint'];
            $seq = $msg['data']['seq'];

            // A popped message here carries its payload, its ids and its lease,
            // and no attempt counter, so a sender that wants to back off, or to
            // give up early on an error it knows is permanent, counts its own
            // attempts. That is what this map is.
            $attempts[$endpoint] = ($attempts[$endpoint] ?? 0) + 1;

            try {
                $postToEndpoint($endpoint, $msg['data']);
            } catch (Throwable $failure) {
                // The nack spends one unit of the retry budget, and the error
                // travels with it: it is what the broker writes on the row when
                // the budget finally runs out.
                $nack = $queen->ack($msg, 'failed', ['group' => $GROUP, 'error' => $failure->getMessage()]);
                if (($nack[0]['success'] ?? false) !== true) {
                    throw new RuntimeException("the broker refused the nack for {$endpoint}/{$seq}");
                }
                echo "  {$endpoint} <- event {$seq} failed: {$failure->getMessage()}\n";
                return;
            }

            // Delivered. The ack names the consumer group explicitly: an ack sent
            // without it commits the queue's own cursor instead of this group's.
            // The reply arrives under an envelope whose outer success flag is set
            // before the broker is even read, so the numbered row beneath it is
            // the only proof the acknowledgement was taken.
            $ack = $queen->ack($msg, 'completed', ['group' => $GROUP]);
            if (($ack[0]['success'] ?? false) !== true) {
                throw new RuntimeException("the broker refused the ack for {$endpoint}/{$seq}");
            }

            $deliveredTo[$endpoint][] = $seq;
            echo "  {$endpoint} <- event {$seq}\n";
        })
        ->execute();

    // ------------------------------------------------------------------ checking
    echo "\nchecking\n";

    foreach ($ENDPOINTS as $endpoint => $meta) {
        if (!$meta['healthy']) {
            continue;
        }
        $seqs = $deliveredTo[$endpoint] ?? [];
        $assert(count($seqs) === $EVENTS_PER_ENDPOINT, "{$endpoint} received all {$EVENTS_PER_ENDPOINT} events");
        $assert($seqs === [1, 2, 3], "{$endpoint} received them in the order they happened");
    }

    $assert(
        count($deliveredTo['initech.example'] ?? []) === 0,
        'the dead endpoint received nothing, as it should'
    );
    $assert(
        ($attempts['initech.example'] ?? 0) > $EVENTS_PER_ENDPOINT,
        'the dead endpoint was retried rather than dropped on the first failure'
    );

    // The dead-letter queue is a table you can read, not a log line. Each row
    // carries the payload, the endpoint it was for, and the last error, which is
    // what a support engineer needs to answer "why did this customer not get it".
    $dlq = $queen->queue($DELIVERIES)->dlq()->limit(50)->get();
    $messages = $dlq['messages'] ?? [];
    $dead = array_values(array_filter($messages, fn(array $m): bool => $m['data']['endpoint'] === 'initech.example'));

    $assert(count($dead) === $EVENTS_PER_ENDPOINT, "all {$EVENTS_PER_ENDPOINT} dead deliveries are in the dead-letter queue");
    $assert(
        count(array_filter($dead, fn(array $m): bool => str_contains($m['errorMessage'] ?? '', 'answered 500'))) === count($dead),
        'each dead-letter row carries the error that killed it'
    );
    $assert(
        count(array_filter($messages, fn(array $m): bool => $m['data']['endpoint'] === 'initech.example')) === count($messages),
        'no healthy endpoint put anything in the dead-letter queue'
    );

    echo "\n  dead letters: " . implode(', ', array_map(
        fn(array $m): string => "{$m['data']['endpoint']}/{$m['data']['invoiceId']}",
        $dead
    )) . "\n";

    $queen->queue($DELIVERIES)->delete()->execute();

    echo "\nPASS: {$checks} checks\n";
} catch (Throwable $error) {
    fwrite(STDERR, "\nFAIL: " . $error->getMessage() . "\n");
    $exitCode = 1;
} finally {
    $queen->close();
}

exit($exitCode);
```
### C++

```cpp title="examples/apps/cpp/webhooks.cpp"
//
// A webhook delivery system.
//
// Every SaaS product ends up writing this one, and it is harder than it looks:
// deliveries to one customer's endpoint must arrive in order, a customer whose
// endpoint is down must not slow down anybody else's, failures must be retried
// a bounded number of times, and what never succeeds has to end up somewhere a
// human can look at.
//
// The shape here is one ordered lane per destination, created by the first
// delivery to it. A dead endpoint backs up its own lane and no other; retries
// are the broker's retry budget rather than a loop in your code; and what
// exhausts the budget lands in the dead-letter queue with the error attached.
//
//   webhook-deliveries (one partition per destination)
//     `-- group "sender"  posts each delivery, fails on a dead endpoint
//           `-- retryLimit exhausted -> dead-letter queue
//
// Build it (see examples/tutorials/cpp/01-hello-world.cpp for the headers
// queen_client.hpp expects but this repository does not vendor -- json.hpp
// under clients/server/vendor, threadpool.hpp under clients/server/include --
// and for why -lssl -lcrypto is required even over plain http):
//   mkdir -p build
//   c++ -std=c++17 -O1 -pthread \
//       -I../../../clients/client-cpp -I../../../clients/server/vendor \
//       -I/opt/homebrew/include -I"$(brew --prefix openssl)/include" \
//       webhooks.cpp -o build/webhooks \
//       -L"$(brew --prefix openssl)/lib" -lssl -lcrypto -lpthread
//
// Run it:
//   QUEEN_URL=http://localhost:6632 ./build/webhooks

#include "queen_client.hpp"

#include <atomic>
#include <chrono>
#include <cstdlib>
#include <exception>
#include <iostream>
#include <map>
#include <mutex>
#include <sstream>
#include <string>
#include <vector>

using queen::QueenClient;
using json = nlohmann::json;

static std::string run_id() {
    auto millis = std::chrono::duration_cast<std::chrono::milliseconds>(
                      std::chrono::system_clock::now().time_since_epoch())
                      .count();
    std::string out;
    const char* digits = "0123456789abcdefghijklmnopqrstuvwxyz";
    while (millis > 0) {
        out.insert(out.begin(), digits[millis % 36]);
        millis /= 36;
    }
    return out;
}

struct Endpoint {
    std::string host;
    bool healthy;
};

// Three subscribers. One of them has let its certificate expire, which is the
// most common way a webhook endpoint dies: it answers, but it answers 500.
static const std::vector<Endpoint> ENDPOINTS = {
    {"acme.example", true},
    {"globex.example", true},
    {"initech.example", false},
};
static const int EVENTS_PER_ENDPOINT = 3;
static const int RETRY_LIMIT = 2;
static const std::string GROUP = "sender";

// What a green run looks like: every healthy delivery succeeds once, and each
// dead one is attempted RETRY_LIMIT + 1 times before the budget is gone.
static const int HEALTHY_DELIVERIES = EVENTS_PER_ENDPOINT * 2;
static const int DEAD_DELIVERIES = EVENTS_PER_ENDPOINT;
static const int EXPECTED_ATTEMPTS =
    HEALTHY_DELIVERIES + EVENTS_PER_ENDPOINT * (RETRY_LIMIT + 1);

static int checks = 0;

// A throwing check: the failure travels to main() as an exception, which is
// what turns it into "FAIL: <reason>" and a non-zero exit.
static void check(bool condition, const std::string& description) {
    if (!condition) throw std::runtime_error(description);
    ++checks;
    std::cout << "  ok: " << description << std::endl;
}

static bool is_healthy(const std::string& host) {
    for (const Endpoint& endpoint : ENDPOINTS) {
        if (endpoint.host == host) return endpoint.healthy;
    }
    throw std::runtime_error("unknown endpoint " + host);
}

// Stands in for the HTTP POST to the subscriber. A real sender would use an
// HTTP client and treat any non-2xx as a failure, which is exactly what
// throwing does here.
static void post_to_endpoint(const std::string& host) {
    if (!is_healthy(host)) {
        throw std::runtime_error(host + " answered 500");
    }
}

static std::string join(const std::vector<int>& values) {
    std::ostringstream out;
    for (size_t i = 0; i < values.size(); ++i) {
        if (i) out << ", ";
        out << values[i];
    }
    return out.str();
}

int main() {
    const char* env_url = std::getenv("QUEEN_URL");
    const std::string QUEEN_URL = env_url ? env_url : "http://localhost:6632";
    const std::string DELIVERIES = "app-cpp-webhooks-" + run_id();

    QueenClient client(QUEEN_URL);

    std::string verdict;
    bool failed = false;

    try {
        std::cout << "broker " << QUEEN_URL << std::endl;

        // retryLimit is the delivery budget, and dlqAfterMaxRetries is what
        // happens when it runs out: the broker moves the exhausted message to
        // the dead-letter table with the last error on the row. The broker
        // defaults that flag to true, so it is set here to say what this
        // program depends on rather than to switch anything on; turning it off
        // is what would leave an exhausted delivery merely marked failed and
        // sitting in its lane.
        //
        // leaseTime is the other half of the contract: it is how long the
        // broker waits for a sender that took a delivery and never came back
        // before handing that delivery to someone else.
        //
        // The C++ QueueConfig struct has no dlqAfterMaxRetries field, so the
        // options object is assembled by hand and posted through the client's
        // own HTTP transport, which keeps the base URL, the retry and the 429
        // backoff policy that every other call in this file uses. Writing the
        // flag out is worth the detour because this program's last three checks
        // are about the dead-letter rows.
        json configured = client.get_http_client()->post(
            "/api/v1/configure",
            json{{"queue", DELIVERIES},
                 {"options", {{"leaseTime", 30},
                              {"retryLimit", RETRY_LIMIT},
                              {"dlqAfterMaxRetries", true}}}});
        if (!configured.value("configured", false)) {
            throw std::runtime_error("the broker did not configure the queue");
        }

        // -------------------------------------------------------------- queuing
        //
        // The application emits events. Each one goes into the partition of the
        // endpoint it is destined for, which is what makes "in order per
        // subscriber" a property of the storage rather than of the sender.
        std::cout << "\nqueuing deliveries" << std::endl;
        for (int seq = 1; seq <= EVENTS_PER_ENDPOINT; ++seq) {
            for (const Endpoint& endpoint : ENDPOINTS) {
                // The event id makes the enqueue idempotent: an application
                // that retries its own emit does not create a second delivery.
                client.queue(DELIVERIES).partition(endpoint.host).push({
                    json{{"transactionId",
                          endpoint.host + "-evt-" + std::to_string(seq)},
                         {"data", {{"endpoint", endpoint.host},
                                   {"seq", seq},
                                   {"type", "invoice.paid"},
                                   {"invoiceId", "INV-" + std::to_string(seq)}}}}
                });
            }
        }
        std::cout << "  " << EVENTS_PER_ENDPOINT * ENDPOINTS.size()
                  << " deliveries queued" << std::endl;

        // -------------------------------------------------------------- sending
        //
        // The sender pool. Here this client parts company with the JavaScript:
        // its automatic acknowledgement turns a thrown handler into a nack with
        // a null error, so the dead-letter rows would arrive with no reason on
        // them and a support engineer would have nothing to read. auto_ack
        // (false) hands the acknowledgement back to the handler, which sends
        // the failure with the message that caused it.
        //
        // That is the only difference. The retries are still the broker's: a
        // failed acknowledgement puts the delivery back in its lane until the
        // retry budget is gone, and that survives the sender process dying
        // mid-flight, which a loop inside the handler would not.
        //
        // concurrency(3) runs three poll loops and each pop claims one
        // partition, so the three destinations are drained in parallel and the
        // dead one backs up alone.
        //
        //   wait(false)     keeps the idle clock meaningful: it is only
        //                   consulted between polls, so a long-polling pop
        //                   would stretch the 6 second budget to the length of
        //                   one server-side park.
        //   idle_millis     the deadline. A delivery that never comes back
        //                   fails this run instead of hanging it.
        //   limit()         counts per worker, here and in the JavaScript
        //                   alike: each worker keeps its own tally, so with
        //                   three workers it is a backstop rather than the
        //                   thing that ends the run. The stop flag below is
        //                   what actually ends the loop, and it is raised on
        //                   the outcome -- every healthy delivery sent and
        //                   every dead one dead-lettered -- rather than on an
        //                   attempt count.
        std::cout << "\nsending" << std::endl;
        std::mutex lock;
        std::map<std::string, std::vector<int>> delivered_to;
        std::map<std::string, int> attempts;
        int dead_lettered = 0;
        int delivered_total = 0;
        std::atomic<bool> stop{false};
        std::exception_ptr handler_error;

        client.queue(DELIVERIES)
            .group(GROUP)
            .subscription_mode("all")
            .concurrency(3)
            .each()
            .auto_ack(false)
            .limit(EXPECTED_ATTEMPTS)
            .wait(false)
            .idle_millis(6000)
            .consume([&](const json& msg) {
                try {
                    const std::string host = msg["data"]["endpoint"].get<std::string>();
                    const int seq = msg["data"]["seq"].get<int>();

                    // The popped message carries no attempt counter, so a
                    // sender that wants to back off or give up early on an
                    // error it knows is permanent counts the attempts itself.
                    // That is all this map is.
                    {
                        std::lock_guard<std::mutex> guard(lock);
                        attempts[host] += 1;
                    }

                    std::string error;
                    try {
                        post_to_endpoint(host);
                    } catch (const std::exception& e) {
                        error = e.what();
                    }

                    // The acknowledgement names the consumer group explicitly:
                    // the broker does not read it off the message, and an ack
                    // sent without it commits the wrong cursor.
                    json context = {{"group", GROUP}};
                    if (!error.empty()) context["error"] = error;
                    json ack = client.ack(msg, error.empty(), context);

                    // A rejected acknowledgement still arrives as HTTP 200 with
                    // success: false on the item, so the per-item flag is the
                    // only proof the broker took it. The outer "success" only
                    // says the call did not throw.
                    if (!ack.value("success", false) || !ack["result"].is_array() ||
                        ack["result"].empty() ||
                        !ack["result"][0].value("success", false)) {
                        throw std::runtime_error("the broker rejected an acknowledgement");
                    }

                    std::lock_guard<std::mutex> guard(lock);
                    if (error.empty()) {
                        delivered_to[host].push_back(seq);
                        ++delivered_total;
                        std::cout << "  " << host << " <- event " << seq << std::endl;
                    } else if (ack["result"][0].value("dlq", false)) {
                        // The broker says this failure spent the last of the
                        // retry budget and the delivery is now a dead letter.
                        ++dead_lettered;
                    }

                    if (delivered_total >= HEALTHY_DELIVERIES &&
                        dead_lettered >= DEAD_DELIVERIES) {
                        stop = true;
                    }
                } catch (...) {
                    // auto_ack is off, so nothing acknowledges behind this
                    // handler's back: an escaped exception leaves the delivery
                    // leased until it expires. Carry the error out and stop.
                    std::lock_guard<std::mutex> guard(lock);
                    if (!handler_error) handler_error = std::current_exception();
                    stop = true;
                }
            }, &stop);
        if (handler_error) std::rethrow_exception(handler_error);

        // ------------------------------------------------------------- checking
        std::cout << "\nchecking" << std::endl;

        std::vector<int> in_order;
        for (int seq = 1; seq <= EVENTS_PER_ENDPOINT; ++seq) in_order.push_back(seq);

        for (const Endpoint& endpoint : ENDPOINTS) {
            if (!endpoint.healthy) continue;
            const std::vector<int>& seqs = delivered_to[endpoint.host];
            check(seqs.size() == static_cast<size_t>(EVENTS_PER_ENDPOINT),
                  endpoint.host + " received all " +
                      std::to_string(EVENTS_PER_ENDPOINT) + " events");
            check(seqs == in_order,
                  endpoint.host + " received them in the order they happened");
        }

        check(delivered_to["initech.example"].empty(),
              "the dead endpoint received nothing, as it should");
        check(attempts["initech.example"] > EVENTS_PER_ENDPOINT,
              "the dead endpoint was retried rather than dropped on the first "
              "failure");

        // The dead-letter queue is a table you can read, not a log line. Each
        // row carries the payload, the endpoint it was for, and the last error,
        // which is what a support engineer needs to answer "why did this
        // customer not get it".
        //
        // dlq().get() swallows a transport failure and answers with an empty
        // page rather than throwing, so an unreachable broker shows up as the
        // next three checks failing rather than as an exception.
        json dlq = client.queue(DELIVERIES).dlq().limit(50).get();
        const json& rows = dlq["messages"];

        std::vector<json> dead;
        for (const json& row : rows) {
            if (row["data"]["endpoint"] == "initech.example") dead.push_back(row);
        }

        check(dead.size() == static_cast<size_t>(EVENTS_PER_ENDPOINT),
              "all " + std::to_string(EVENTS_PER_ENDPOINT) +
                  " dead deliveries are in the dead-letter queue");

        bool every_row_explains_itself = true;
        for (const json& row : dead) {
            const std::string message = row.value("errorMessage", std::string());
            if (message.find("answered 500") == std::string::npos) {
                every_row_explains_itself = false;
            }
        }
        check(every_row_explains_itself,
              "each dead-letter row carries the error that killed it");

        bool only_the_dead_endpoint = true;
        for (const json& row : rows) {
            if (row["data"]["endpoint"] != "initech.example") only_the_dead_endpoint = false;
        }
        check(only_the_dead_endpoint,
              "no healthy endpoint put anything in the dead-letter queue");

        std::ostringstream letters;
        for (size_t i = 0; i < dead.size(); ++i) {
            if (i) letters << ", ";
            letters << dead[i]["data"]["endpoint"].get<std::string>() << "/"
                    << dead[i]["data"]["invoiceId"].get<std::string>();
        }
        std::cout << "\n  dead letters: " << letters.str() << std::endl;
        std::cout << "  attempts on the dead endpoint: "
                  << attempts["initech.example"] << ", on "
                  << ENDPOINTS[0].host << ": " << attempts[ENDPOINTS[0].host]
                  << " (" << join(delivered_to[ENDPOINTS[0].host]) << ")"
                  << std::endl;

        // Clean up on success only: a failed run leaves the queue and its
        // dead-letter rows on the broker to be looked at.
        client.queue(DELIVERIES).del();

        verdict = "\nPASS: " + std::to_string(checks) + " checks";
    } catch (const std::exception& err) {
        verdict = std::string("\nFAIL: ") + err.what();
        failed = true;
    }

    client.close();

    (failed ? std::cerr : std::cout) << verdict << std::endl;
    return failed ? 1 : 0;
}
```
### HTTP

```bash title="examples/apps/http/webhooks.sh"
#!/usr/bin/env bash
#
# A webhook delivery system, with nothing but curl.
#
# Every SaaS product ends up writing this one, and it is harder than it looks:
# deliveries to one customer's endpoint must arrive in order, a customer whose
# endpoint is down must not slow down anybody else's, failures must be retried a
# bounded number of times, and what never succeeds has to end up somewhere a
# human can look at.
#
# The shape here is one ordered lane per destination, created by the first
# delivery to it. A dead endpoint backs up its own lane and no other; retries are
# the broker's retry budget rather than a loop in your code; and what exhausts
# the budget lands in the dead-letter queue with the error attached.
#
#   webhook-deliveries (one partition per destination)
#     └── group "sender"  posts each delivery, fails on a dead endpoint
#           └── retryLimit exhausted -> dead-letter queue
#
# One sender runs per destination, as a background subshell on the
# partition-scoped pop route, so the lane isolation is real here rather than
# described: the dead endpoint spends the whole run failing and retrying while
# the two healthy ones are already finished.
#
# Run it:
#   QUEEN_URL=http://localhost:6632 bash webhooks.sh

set -euo pipefail

QUEEN_URL="${QUEEN_URL:-http://localhost:6632}"

# The name carries the language and a per-run suffix, so every application in
# every language can share one broker and no run inherits another's state.
RUN="$(date +%s)-$$"
DELIVERIES="app-http-webhooks-$RUN"
GROUP=app-http-sender

# Three subscribers: endpoint and whether it is healthy. One of them has let its
# certificate expire, which is the most common way a webhook endpoint dies: it
# answers, but it answers 500.
ENDPOINTS='acme.example yes
globex.example yes
initech.example no'
EVENTS_PER_ENDPOINT=3
RETRY_LIMIT=2

# 1,2,3: what a healthy subscriber must receive, in that order.
EXPECTED_SEQS="$(seq 1 "$EVENTS_PER_ENDPOINT" | paste -sd, -)"

# Every pop long-polls for this many milliseconds and no longer, so a sender
# re-checks its own progress rather than parking until the run is over.
POLL_MS=1000

# The bound that keeps a stall from becoming a hang. A sender that has not
# finished its lane by then stops, and the checks that follow report what is
# missing. Never wait for silence; wait for a total, with a deadline.
SEND_MS=30000

command -v jq >/dev/null 2>&1 || { echo "FAIL: jq is not installed"; exit 1; }

CHECKS=0
TMP="$(mktemp -d)"

# One exit path for everything. A failed check calls fail(), which records the
# reason and exits 1; any other command that fails under `set -e` arrives here
# too, with its own status. FAIL is printed exactly once, and only on failure.
cleanup() {
  local status=$?
  rm -rf "$TMP"
  if [ "$status" -ne 0 ]; then
    echo
    echo "FAIL: ${FAILURE:-a command exited with status $status}"
  fi
  exit "$status"
}
trap cleanup EXIT

fail() { FAILURE="$*"; exit 1; }

# check <actual> <expected> <description>
check() {
  [ "$1" = "$2" ] || fail "$3 (expected [$2], got [$1])"
  CHECKS=$((CHECKS + 1))
  echo "  ok: $3"
}

# ok <description>: records a check whose condition was already tested. check()
# compares two values, and one assertion below is an inequality.
ok() {
  CHECKS=$((CHECKS + 1))
  echo "  ok: $1"
}

# A millisecond clock, for the deadline only. GNU date spells it %3N; BSD date
# (macOS) has no %N and leaves the unconverted tail in the output, so a probe for
# anything that is not a digit tells the two apart, and perl, whose Time::HiRes
# is core, is the fallback.
if [ -z "$(date +%s%3N 2>/dev/null | tr -d '0-9')" ]; then
  now_ms() { date +%s%3N; }
else
  command -v perl >/dev/null 2>&1 \
    || { echo "FAIL: need GNU date or perl for a millisecond clock"; exit 1; }
  now_ms() { perl -MTime::HiRes -e 'printf "%d", Time::HiRes::time() * 1000'; }
fi

# Sets $STATUS to the HTTP status code and writes the response body to $OUT.
#
# $OUT is per-process: each sender below runs as a background subshell and points
# it at its own file, so three concurrent pops never overwrite each other's
# response. There is no --fail, because Queen reports outcomes in the body and
# several of the interesting ones arrive as 200.
OUT="$TMP/body"
request() {
  local method="$1" path="$2" body="${3:-}"
  if [ -n "$body" ]; then
    STATUS="$(curl -sS -o "$OUT" -w '%{http_code}' \
      -X "$method" "$QUEEN_URL$path" \
      -H 'content-type: application/json' -d "$body")"
  else
    STATUS="$(curl -sS -o "$OUT" -w '%{http_code}' -X "$method" "$QUEEN_URL$path")"
  fi
}

# healthy <endpoint>: prints yes or no. The shell this has to run on has no
# associative arrays, so the subscriber table is a few lines of text and awk is
# the lookup.
healthy() {
  printf '%s\n' "$ENDPOINTS" | awk -v e="$1" '$1 == e { print $2 }'
}

# Stands in for the HTTP POST to the subscriber. A real sender would use curl and
# treat any non-2xx as a failure, which is exactly what returning non-zero does
# here.
post_to_endpoint() {
  [ "$(healthy "$1")" = yes ] || return 1
  return 0
}

# lines <file>: how many rows a tally file holds, 0 when it does not exist yet.
lines() {
  [ -f "$1" ] || { echo 0; return; }
  wc -l < "$1" | tr -d ' '
}

echo "broker $QUEEN_URL"

# ---------------------------------------------------------------------------
# retryLimit is the delivery budget, and dlqAfterMaxRetries is what happens when
# it runs out. Without the second flag an exhausted message would simply be
# marked failed and stay put; with it, the broker moves it to the dead-letter
# table with the last error on the row. Both flags are sent explicitly because
# /configure is a full replace: what you leave out is reset to its default, not
# left as it was.
#
# leaseTime is the other half of the contract: it is how long the broker waits
# for a sender that took a delivery and never came back before handing that
# delivery to someone else.
# ---------------------------------------------------------------------------
configure_body="$(jq -n --arg queue "$DELIVERIES" --argjson retry "$RETRY_LIMIT" \
  '{queue: $queue,
    options: {leaseTime: 30, retryLimit: $retry,
              deadLetterQueue: true, dlqAfterMaxRetries: true}}')"
request POST /api/v1/configure "$configure_body"
[ "$STATUS" = 200 ] || fail "configure returned HTTP $STATUS"
check "$(jq -r .configured "$OUT")" true \
  "the queue was created with a delivery budget of $RETRY_LIMIT retries"

# ------------------------------------------------------------------------ queuing
#
# The application emits events. Each one goes into the partition of the endpoint
# it is destined for, which is what makes "in order per subscriber" a property of
# the storage rather than of the sender. Nothing was declared for a subscriber in
# advance: the partition comes into existence with the first delivery to it.
echo
echo "queuing deliveries"
seq_no=1
while [ "$seq_no" -le "$EVENTS_PER_ENDPOINT" ]; do
  while read -r endpoint is_healthy; do
    # The event id makes the enqueue idempotent: an application that retries its
    # own emit does not create a second delivery. The wire field for the body is
    # "payload"; "data" is what a pop calls it on the way back.
    body="$(jq -n --arg queue "$DELIVERIES" --arg endpoint "$endpoint" \
      --argjson seq "$seq_no" \
      '{items: [{
         queue:     $queue,
         partition: $endpoint,
         transactionId: ($endpoint + "-evt-" + ($seq | tostring)),
         payload: {endpoint: $endpoint, seq: $seq, type: "invoice.paid",
                   invoiceId: ("INV-" + ($seq | tostring))}
       }]}')"
    request POST /api/v1/push "$body"
    [ "$STATUS" = 201 ] || fail "push of $endpoint/$seq_no returned HTTP $STATUS"
    # HTTP 201 is not proof the message was stored: "buffered" and "failed" also
    # come back 201. The per-item status is the only answer.
    [ "$(jq -r '.[0].status' "$OUT")" = queued ] \
      || fail "push of $endpoint/$seq_no came back $(jq -r '.[0].status' "$OUT")"
  done <<EOF
$ENDPOINTS
EOF
  seq_no=$((seq_no + 1))
done
echo "  $((EVENTS_PER_ENDPOINT * 3)) deliveries queued"

# ------------------------------------------------------------------------ sending
#
# die() is a sender's fail(): a sender is a subshell, so its variables die with
# it and the parent would never see FAILURE. It leaves the reason in a file the
# parent reads after wait().
die() { printf '%s\n' "$*" > "$TMP/sender-error"; exit 1; }

# send_lane <endpoint>
#
# One sender, one destination. It pops that endpoint's partition by name, posts
# what it gets, and reports the outcome back to the broker; the loop is what an
# SDK's consume() does, and the ack status is what an SDK's autoAck derives from
# a handler that returned or threw.
#
# It stops when the lane is resolved, either because every event was delivered or
# because every event was dead-lettered, and in any case at the deadline.
send_lane() {
  local endpoint="$1"
  local deadline popfile delivered_file attempts_file dead_file
  local txn partition_id lease event_seq aborted ack_body

  OUT="$TMP/$endpoint-body"
  popfile="$TMP/$endpoint-pop"
  delivered_file="$TMP/delivered-$endpoint"
  attempts_file="$TMP/attempts-$endpoint"
  dead_file="$TMP/dead-$endpoint"
  : > "$delivered_file"
  : > "$attempts_file"
  : > "$dead_file"
  deadline=$(( $(now_ms) + SEND_MS ))

  while [ "$(( $(lines "$delivered_file") + $(lines "$dead_file") ))" \
          -lt "$EVENTS_PER_ENDPOINT" ]; do
    [ "$(now_ms)" -lt "$deadline" ] || break

    # The partition-scoped pop route claims exactly the lane you name. The
    # queue-scoped one lets the broker pick, and with `partitions` at its default
    # of 1 it would claim a single lane per call anyway; naming the partition is
    # what makes this sender belong to one subscriber.
    #
    # subscriptionMode=all is what makes a group created now read what was pushed
    # before it existed: a new cursor is seeded at the TAIL unless you say
    # otherwise. It seeds a cursor that does not exist yet, and is ignored on
    # every later pop.
    request GET "/api/v1/pop/queue/$DELIVERIES/partition/$endpoint?consumerGroup=$GROUP&subscriptionMode=all&batch=10&wait=true&timeout=$POLL_MS"
    # 204 is an empty pop, with no body at all. Here it means the lane is quiet
    # for the moment; the loop condition decides whether that is the end.
    [ "$STATUS" != 204 ] || continue
    [ "$STATUS" = 200 ] || die "pop on $endpoint returned HTTP $STATUS"
    cp "$OUT" "$popfile"

    lease="$(jq -r .leaseId "$popfile")"
    jq -r '.messages[] | [.transactionId, .partitionId, (.data.seq | tostring)] | @tsv' \
      "$popfile" > "$TMP/$endpoint-batch"

    aborted=0
    while IFS=$'\t' read -r txn partition_id event_seq; do
      echo "$event_seq" >> "$attempts_file"

      if ! post_to_endpoint "$endpoint"; then
        # -------------------------------------------------------------------
        # The delivery failed. A `failed` ack is the nack: it clamps the cursor
        # just below this message, so this one and everything after it in the
        # batch redelivers, and it charges the retry budget once. There is no
        # loop in this sender and no sleep: the redelivery is the retry, and it
        # survives this process dying mid-flight, which a loop would not.
        #
        # Note what does NOT charge the budget: a lease that merely expires.
        # Only an explicit `failed` ack does, so a crash-looping sender never
        # exhausts a message's life by crashing.
        #
        # `error` is the reason. It is not stored for a plain nack; it is what
        # goes on the dead-letter row if this is the nack that dead-letters the
        # message, which is what a support engineer reads later.
        #
        # The rest of the batch is deliberately left alone: the cursor is
        # already clamped here, so those messages are coming back regardless.
        # -------------------------------------------------------------------
        ack_body="$(jq -n --arg txn "$txn" --arg partitionId "$partition_id" \
          --arg group "$GROUP" --arg lease "$lease" \
          --arg error "$endpoint answered 500" \
          '{transactionId: $txn, partitionId: $partitionId, consumerGroup: $group,
            leaseId: $lease, status: "failed", error: $error}')"
        request POST /api/v1/ack "$ack_body"
        [ "$STATUS" = 200 ] || die "ack for $endpoint returned HTTP $STATUS"
        [ "$(jq -r '.[0].success' "$OUT")" = true ] \
          || die "ack refused for $endpoint: $(jq -r '.[0].error' "$OUT")"

        # dlq on the ack result is how a sender on raw HTTP learns the budget ran
        # out: true means this nack is the one that filed the dead-letter row and
        # moved the cursor past the poison delivery. An SDK hides this behind its
        # own retry accounting; here it is on the wire.
        if [ "$(jq -r '.[0].dlq' "$OUT")" = true ]; then
          echo "$event_seq" >> "$dead_file"
          echo "  $endpoint dead-lettered event $event_seq, its retry budget is gone"
        fi
        aborted=1
        break
      fi

      echo "$event_seq" >> "$delivered_file"
      echo "  $endpoint <- event $event_seq"
    done < "$TMP/$endpoint-batch"

    # Everything in the batch was posted, so commit the batch with one ack of its
    # LAST message: an ack is a cursor commit, so that completes every earlier
    # message of this partition for this group too. consumerGroup is mandatory,
    # here as everywhere: omit it and the commit lands on __QUEUE_MODE__, a cursor
    # this sender never read from, and the batch redelivers forever.
    if [ "$aborted" = 0 ]; then
      ack_body="$(jq -c --arg group "$GROUP" '{
        transactionId: .messages[-1].transactionId,
        partitionId:   .messages[-1].partitionId,
        consumerGroup: $group,
        leaseId:       .leaseId,
        status:        "completed"
      }' "$popfile")"
      request POST /api/v1/ack "$ack_body"
      [ "$STATUS" = 200 ] || die "ack for $endpoint returned HTTP $STATUS"
      [ "$(jq -r '.[0].success' "$OUT")" = true ] \
        || die "ack refused for $endpoint: $(jq -r '.[0].error' "$OUT")"
    fi
  done
}

echo
echo "sending"
rm -f "$TMP/sender-error"
PIDS=""
while read -r endpoint is_healthy; do
  send_lane "$endpoint" &
  PIDS="$PIDS $!"
done <<EOF
$ENDPOINTS
EOF
for pid in $PIDS; do
  wait "$pid" \
    || fail "a sender stopped: $(cat "$TMP/sender-error" 2>/dev/null || echo 'no reason recorded')"
done

# ----------------------------------------------------------------------- checking
echo
echo "checking"

while read -r endpoint is_healthy; do
  [ "$is_healthy" = yes ] || continue
  check "$(lines "$TMP/delivered-$endpoint")" "$EVENTS_PER_ENDPOINT" \
    "$endpoint received all $EVENTS_PER_ENDPOINT events"
  check "$(paste -sd, - < "$TMP/delivered-$endpoint")" "$EXPECTED_SEQS" \
    "$endpoint received them in the order they happened"
done <<EOF
$ENDPOINTS
EOF

check "$(lines "$TMP/delivered-initech.example")" 0 \
  'the dead endpoint received nothing, as it should'

DEAD_ATTEMPTS="$(lines "$TMP/attempts-initech.example")"
[ "$DEAD_ATTEMPTS" -gt "$EVENTS_PER_ENDPOINT" ] \
  || fail "the dead endpoint was tried only $DEAD_ATTEMPTS times, so it was not retried"
ok "the dead endpoint was retried rather than dropped on the first failure ($DEAD_ATTEMPTS attempts)"

check "$(lines "$TMP/dead-initech.example")" "$EVENTS_PER_ENDPOINT" \
  'the broker reported a dead-letter on the ack that exhausted each budget'

# ---------------------------------------------------------------------------
# The dead-letter queue is a table you can read, not a log line. Each row carries
# the payload snapshot, the endpoint it was for, and the last error, which is
# what a support engineer needs to answer "why did this customer not get it".
#
# Retention never purges these rows: a dead-lettered message stays until you
# delete it, replay it with POST /api/v1/messages/:partitionId/:transactionId/retry,
# or delete the queue.
# ---------------------------------------------------------------------------
request GET "/api/v1/dlq?queue=$DELIVERIES&limit=50"
[ "$STATUS" = 200 ] || fail "the dead-letter listing returned HTTP $STATUS"
cp "$OUT" "$TMP/dlq"

check "$(jq '[.messages[] | select(.data.endpoint == "initech.example")] | length' "$TMP/dlq")" \
  "$EVENTS_PER_ENDPOINT" \
  "all $EVENTS_PER_ENDPOINT dead deliveries are in the dead-letter queue"
check "$(jq '[.messages[] | select((.errorMessage // "") | contains("answered 500"))] | length' "$TMP/dlq")" \
  "$EVENTS_PER_ENDPOINT" 'each dead-letter row carries the error that killed it'
check "$(jq '[.messages[] | select(.data.endpoint != "initech.example")] | length' "$TMP/dlq")" 0 \
  'no healthy endpoint put anything in the dead-letter queue'

echo
echo "  dead letters: $(jq -r '[.messages[] | .data.endpoint + "/" + .data.invoiceId] | sort | join(", ")' "$TMP/dlq")"

# Clean up on success only: a failed run leaves the queue, and its dead letters,
# on the broker to be looked at. Deleting a queue that does not exist is also a
# 200, so check "deleted" rather than the status code.
request DELETE "/api/v1/resources/queues/$DELIVERIES"
[ "$(jq -r .deleted "$OUT")" = true ] || fail 'the queue was not deleted'

echo
echo "PASS: $CHECKS checks"
```

## Run it

Against a broker from [the quickstart](/start/quickstart):

```bash
QUEEN_URL=http://localhost:6632 examples/apps/run.sh
```

Every program on this page asserts the property it exists to demonstrate and exits
non-zero if it does not hold. The runner takes a language name to run one of them on its
own, for example `examples/apps/run.sh js`.

Source: https://queenmq.com/use/full-examples/webhooks/index.mdx