Transactional pipeline

//
// Queen MQ by example, 3 of 3: the transactional handoff.
//
// A pipeline stage consumes from "orders" and, in a SINGLE transaction, both
// acknowledges the input and pushes the derived message to "invoices". Either
// both happen or neither does, so no crash can leave a stage that consumed an
// order without emitting its invoice.
//
// The last section shows the other half of the contract: a message acknowledged
// as failed is redelivered, not lost.
//
// Run it:
//   QUEEN_URL=http://localhost:6699 node 03-pipeline-transaction.mjs
//
// The program checks its own outcome and exits non-zero if any check fails.

import { Queen } from 'queen-mq'

const QUEEN_URL = process.env.QUEEN_URL || 'http://localhost:6699'

// Prefixed per language, suffixed per run, so runs and languages never collide.
const RUN_ID = Date.now().toString(36)
const Q_ORDERS = `ex-js-pipeline-orders-${RUN_ID}`
const Q_INVOICES = `ex-js-pipeline-invoices-${RUN_ID}`

// Each stage reads with its own consumer group, so each stage has its own cursor.
const G_INVOICER = 'ex-js-invoicer' // stage one: orders  -> invoices
const G_LEDGER = 'ex-js-ledger' // stage two: invoices -> here

const ORDERS = [
  { orderId: 'ORD-1', customer: 'acme', total: 120.5 },
  { orderId: 'ORD-2', customer: 'globex', total: 88.75 },
  { orderId: 'ORD-3', customer: 'initech', total: 310.0 },
  { orderId: 'ORD-4', customer: 'acme', total: 12.0 },
]

const VAT = 1.22

let checks = 0

function assert(condition, description) {
  if (!condition) throw new Error(description)
  checks++
  console.log(`  ok: ${description}`)
}

// The derived message: what stage one makes out of an order.
function invoiceFor(order) {
  return {
    invoiceFor: order.orderId,
    customer: order.customer,
    amountDue: Math.round(order.total * VAT * 100) / 100,
  }
}

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

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

  await queen.queue(Q_ORDERS).config({
    leaseTime: 30,
    // How many times a failed message is redelivered before it goes to the
    // dead letter queue. The failure section below spends one of these.
    retryLimit: 3,
  }).create()
  await queen.queue(Q_INVOICES).config({ leaseTime: 30, retryLimit: 3 }).create()
  console.log(`queues ${Q_ORDERS} and ${Q_INVOICES} created`)

  console.log('\npushing orders')
  for (const order of ORDERS) {
    // One partition per customer, so one customer's orders stay in order while
    // different customers are free to be processed in parallel.
    const [result] = await queen.queue(Q_ORDERS).partition(order.customer).push([{ data: order }])
    if (result.status !== 'queued') throw new Error(`push of ${order.orderId} answered ${result.status}`)
    console.log(`  ${order.orderId} -> partition ${order.customer}`)
  }

  // -------------------------------------------------- stage one: orders -> invoices

  console.log('\nstage one: ack the order and push the invoice in one transaction')
  let handled = 0
  while (handled < ORDERS.length) {
    // One message at a time keeps the unit of work equal to the unit of
    // transaction, which is the whole point of this stage.
    const batch = await queen.queue(Q_ORDERS)
      .group(G_INVOICER)
      .batch(1)
      .wait(true)
      .timeoutMillis(5000)
      .pop()

    // pop() returns an empty array both when the queue is empty and when the
    // call failed, so an empty result ends the loop and the assertion below
    // reports the shortfall instead of the loop spinning forever.
    if (batch.length === 0) break
    const order = batch[0]

    // The transaction: the ack of the input and the push of the output land in
    // the same PostgreSQL commit. The leaseId of the acked message travels with
    // it as a required lease, so if the lease had expired (another consumer
    // already took the order) the whole commit is refused, including the push.
    const result = await queen.transaction()
      .ack(order, 'completed', { consumerGroup: G_INVOICER })
      .queue(Q_INVOICES)
      .partition(order.partition)
      // Inside a transaction the broker mints the transactionId for the pushed
      // message; the client does not forward one of its own here.
      .push([{ data: invoiceFor(order.data) }])
      .commit()

    // commit() throws unless the broker answered success, so reaching this line
    // means both operations are durable.
    const kinds = result.results.map(r => `${r.type}:${r.success}`).join(' ')
    console.log(`  ${order.data.orderId} committed (${kinds})`)
    handled++
  }

  console.log('\nchecking stage one')
  assert(handled === ORDERS.length, `stage one handled all ${ORDERS.length} orders`)

  // Every ack in those transactions committed, so this group's cursor is at the
  // end of the queue and there is nothing left to hand off.
  const remaining = await queen.queue(Q_ORDERS).group(G_INVOICER).batch(10).wait(false).pop()
  assert(remaining.length === 0, 'no order is left unacknowledged')

  // -------------------------------------------- stage two: read what was derived

  console.log('\nstage two: consuming invoices')
  const invoices = []
  await queen.queue(Q_INVOICES)
    .group(G_LEDGER)
    .each()
    .batch(ORDERS.length)
    .partitions(3) // the three customer partitions, claimed in one poll
    .autoAck(false)
    .limit(ORDERS.length)
    .idleMillis(5000)
    .consume(async (msg) => {
      console.log(`  invoice for ${msg.data.invoiceFor}: ${msg.data.amountDue}`)
      invoices.push(msg.data)
      const ack = await queen.ack(msg, true, { group: G_LEDGER })
      if (!ack.success) throw new Error(`ack rejected: ${ack.error}`)
    })

  console.log('\nchecking the derived messages')
  assert(invoices.length === ORDERS.length, `every order produced one invoice (${ORDERS.length})`)
  const invoicedOrders = invoices.map(i => i.invoiceFor)
  assert(new Set(invoicedOrders).size === invoicedOrders.length, 'no invoice was produced twice')
  assert(
    ORDERS.every(order => {
      const invoice = invoices.find(i => i.invoiceFor === order.orderId)
      return invoice && invoice.amountDue === invoiceFor(order).amountDue
    }),
    'every invoice carries the amount derived from its order'
  )

  // ------------------------------------------------- what a failure would do

  console.log('\nfailure path: acknowledging one order as failed')
  const late = { orderId: 'ORD-5', customer: 'acme', total: 55.0 }
  const [pushed] = await queen.queue(Q_ORDERS).partition(late.customer).push([{ data: late }])

  const firstTry = await queen.queue(Q_ORDERS).group(G_INVOICER).batch(1).wait(true).timeoutMillis(5000).pop()
  assert(firstTry.length === 1, 'the new order was delivered once')
  assert(firstTry[0].transactionId === pushed.transaction_id, 'it is the order that was just pushed')

  // status false means `failed`. The broker releases the lease and clamps this
  // group's cursor at the failed message, so it is queued for redelivery rather
  // than being dropped. The error text is kept and lands on the dead letter row
  // if the retry budget ever runs out.
  const nack = await queen.ack(firstTry[0], false, {
    group: G_INVOICER,
    error: 'ledger unreachable, will retry',
  })
  assert(nack.success, 'the broker accepted the failed acknowledgement')

  const secondTry = await queen.queue(Q_ORDERS).group(G_INVOICER).batch(1).wait(true).timeoutMillis(5000).pop()
  assert(secondTry.length === 1, 'the failed order came back')
  assert(secondTry[0].transactionId === pushed.transaction_id, 'it is the same message, redelivered rather than lost')
  assert(secondTry[0].data.orderId === late.orderId, 'with its payload intact')

  // The retry succeeds this time, through the same transactional handoff.
  await queen.transaction()
    .ack(secondTry[0], 'completed', { consumerGroup: G_INVOICER })
    .queue(Q_INVOICES)
    .partition(secondTry[0].partition)
    .push([{ data: invoiceFor(late) }])
    .commit()

  const recovered = await queen.queue(Q_INVOICES).group(G_LEDGER).batch(5).wait(true).timeoutMillis(5000).pop()
  assert(recovered.length === 1, 'the retry produced exactly one invoice')
  assert(recovered[0].data.invoiceFor === late.orderId, `it is the invoice for ${late.orderId}`)
  const finalAck = await queen.ack(recovered, true, { group: G_LEDGER })
  assert(finalAck.success, 'and it acknowledges cleanly')

  // Clean up. Only on success, so a failed run leaves both queues on the broker.
  await queen.queue(Q_ORDERS).delete()
  await queen.queue(Q_INVOICES).delete()

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

"""Queen in Python: a transactional handoff between two queues.

Stage one reads from "orders" and, in a single transaction, acknowledges the
order it read and pushes the invoice it derived from it. Either both land or
neither does, so the pipeline cannot lose an order or invoice one twice. Stage
two reads "invoices" and checks that all of them are there, exactly once.

The last section shows the failure path: a message acknowledged as failed is
redelivered rather than lost.

Run it with the client source on the path:

    cd /path/to/queen
    PYTHONPATH=clients/client-py QUEEN_URL=http://localhost:6699 \
        python3 examples/full/py/03_pipeline_transaction.py

Exits 0 when every check passes, 1 on the first failure.
"""

import asyncio
import os
import sys
import uuid

from queen import Queen

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

# A fresh pair of queue names per run keeps this run independent of every
# previous one.
RUN_ID = uuid.uuid4().hex[:8]
ORDERS = f"ex-py-pipeline-orders-{RUN_ID}"
INVOICES = f"ex-py-pipeline-invoices-{RUN_ID}"

# One group per stage. Each stage has its own cursor over its own queue.
BILLING = "billing"  # stage one, reading orders
LEDGER = "ledger"  # stage two, reading invoices

ORDER_BATCH = [
    {"orderId": "A-1", "customer": "acme", "amount": 120},
    {"orderId": "A-2", "customer": "acme", "amount": 40},
    {"orderId": "B-1", "customer": "globex", "amount": 250},
    {"orderId": "C-1", "customer": "initech", "amount": 999},
]

VAT = 0.22


def check(condition, description):
    """Assert one claim and say so out loud. A failed check exits non-zero."""
    if not condition:
        print(f"FAIL  {description}")
        sys.exit(1)
    print(f"ok    {description}")


async def main():
    async with Queen(BROKER_URL) as client:
        print(f"broker {BROKER_URL}")
        print(f"queues {ORDERS} -> {INVOICES}\n")

        for name in (ORDERS, INVOICES):
            await (
                client.queue(name)
                .config({"lease_time": 60, "retry_limit": 3, "dedup_window_seconds": 300})
                .create()
            )
        print(f"      created {ORDERS} and {INVOICES}\n")

        pushed = await client.queue(ORDERS).push([{"data": o} for o in ORDER_BATCH])
        check(
            all(r["status"] == "queued" for r in pushed),
            f"{len(ORDER_BATCH)} orders are on {ORDERS}",
        )

        # ------------------------------------------------------------------
        # Stage one: read an order, derive an invoice, commit both together.
        # ------------------------------------------------------------------
        # pop() claims messages without committing anything. Nothing is settled
        # until the ack, which is why the ack can travel with the push in one
        # transaction: the broker applies both inside one PostgreSQL transaction
        # or neither of them.
        handled = 0
        while handled < len(ORDER_BATCH):
            batch = await (
                client.queue(ORDERS).group(BILLING).batch(2).wait(True).timeout_millis(3000).pop()
            )
            if not batch:
                break

            transaction = client.transaction()
            for message in batch:
                order = message["data"]
                invoice = {
                    "invoiceFor": order["orderId"],
                    "customer": order["customer"],
                    "total": round(order["amount"] * (1 + VAT), 2),
                }
                transaction = transaction.queue(INVOICES).push([{"data": invoice}])
                print(f"      order {order['orderId']} -> invoice {invoice['total']}")

            # The acks carry each message's leaseId, so a commit whose lease has
            # already expired (another consumer took over) fails instead of
            # writing invoices for work someone else is redoing.
            transaction = transaction.ack(batch, "completed", {"consumer_group": BILLING})

            result = await transaction.commit()
            check(
                result.get("success") is True,
                f"committed {len(batch)} acks and {len(batch)} pushes as one transaction",
            )
            handled += len(batch)

        check(handled == len(ORDER_BATCH), f"stage one handled all {len(ORDER_BATCH)} orders")

        drained = await (
            client.queue(ORDERS).group(BILLING).batch(10).wait(True).timeout_millis(2000).pop()
        )
        check(drained == [], f"{ORDERS} is drained for group {BILLING}")

        # ------------------------------------------------------------------
        # Stage two: the derived messages are all there, exactly once.
        # ------------------------------------------------------------------
        invoices = []

        async def handle(message):
            invoices.append(message["data"])
            print(f"      invoice for {message['data']['invoiceFor']} read by {LEDGER}")

        await (
            client.queue(INVOICES)
            .group(LEDGER)
            .batch(1)
            .limit(len(ORDER_BATCH))
            .idle_millis(5000)
            .timeout_millis(2000)
            .consume(handle)
        )

        billed = sorted(i["invoiceFor"] for i in invoices)
        expected = sorted(o["orderId"] for o in ORDER_BATCH)
        check(len(invoices) == len(ORDER_BATCH), f"{len(invoices)} invoices, expected {len(ORDER_BATCH)}")
        check(len(set(billed)) == len(billed), "no order was invoiced twice")
        check(billed == expected, "every order produced its invoice")

        totals = {i["invoiceFor"]: i["total"] for i in invoices}
        check(
            all(totals[o["orderId"]] == round(o["amount"] * (1 + VAT), 2) for o in ORDER_BATCH),
            "every invoice carries the total derived from its order",
        )

        # ------------------------------------------------------------------
        # The failure path: a failed ack is a redelivery, not a loss.
        # ------------------------------------------------------------------
        print()
        late = {"orderId": "D-1", "customer": "hooli", "amount": 500}
        await client.queue(ORDERS).push({"data": late})

        first_try = await (
            client.queue(ORDERS).group(BILLING).batch(1).wait(True).timeout_millis(3000).pop()
        )
        check(len(first_try) == 1, f"picked up order {late['orderId']}")

        # status False means failed. Nothing downstream was written, so the ack
        # says so: the broker releases the lease and rewinds this group's cursor
        # to the failed message instead of moving past it. After retry_limit
        # failures the message goes to the dead letter queue rather than looping.
        nacked = await client.ack(
            first_try[0], False, {"group": BILLING, "error": "invoice service refused the call"}
        )
        check(nacked.get("success") is True, "acknowledged the order as failed")
        print(f"      failed: {first_try[0]['transactionId']}")

        redelivered = await (
            client.queue(ORDERS).group(BILLING).batch(1).wait(True).timeout_millis(5000).pop()
        )
        check(len(redelivered) == 1, "the failed order was redelivered rather than dropped")
        check(
            redelivered[0]["transactionId"] == first_try[0]["transactionId"],
            "and it is the very same message, not a copy",
        )

        settled = await client.ack(redelivered[0], True, {"group": BILLING})
        check(settled.get("success") is True, "the retry succeeded and was acknowledged")

        gone = await (
            client.queue(ORDERS).group(BILLING).batch(10).wait(True).timeout_millis(2000).pop()
        )
        check(gone == [], "once acknowledged as completed it is not redelivered again")

        for name in (ORDERS, INVOICES):
            deleted = await client.queue(name).delete()
            check(deleted.get("deleted") is True, f"queue {name} deleted")

        print("\nPASS 03_pipeline_transaction")


asyncio.run(main())

// A transactional pipeline: consume from one queue and produce to another
// without ever losing or duplicating work.
//
// Stage one reads an order, and in a SINGLE broker transaction acknowledges
// that order and pushes the invoice derived from it. Stage two reads the
// invoices back and checks that each one is there exactly once. The last part
// acknowledges an order as failed and shows it being redelivered rather than
// dropped.
//
// Run it with:
//
//	QUEEN_URL=http://localhost:6699 go run ./pipeline-transaction
package main

import (
	"context"
	"fmt"
	"os"
	"time"

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

const (
	// Each stage of a pipeline reads with its own consumer group, so the two
	// cursors advance independently.
	ordersGroup   = "ex-go-invoicing"
	invoicesGroup = "ex-go-ledger"

	orderCount = 3

	// What stage one adds to every order it turns into an invoice.
	shippingFee = 5
)

// A per-run suffix gives every run queues that have never existed before, so a
// second run behaves exactly like the first.
var runID = time.Now().UnixMilli()

var (
	ordersQueue   = fmt.Sprintf("ex-go-pipeline-orders-%d", runID)
	invoicesQueue = fmt.Sprintf("ex-go-pipeline-invoices-%d", runID)
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
	defer cancel()

	if err := run(ctx); err != nil {
		fmt.Fprintf(os.Stderr, "FAILED: %v\n", err)
		os.Exit(1)
	}
	fmt.Println("OK: every order produced exactly one invoice, and the failed order came back")
}

func run(ctx context.Context) error {
	brokerURL := os.Getenv("QUEEN_URL")
	if brokerURL == "" {
		brokerURL = "http://localhost:6699"
	}

	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)

	for _, name := range []string{ordersQueue, invoicesQueue} {
		if _, err := client.Queue(name).Config(queen.QueueConfig{
			LeaseTime:  60, // seconds the handler has before the message is offered to someone else
			RetryLimit: 3,  // failed deliveries tolerated before the message is dead lettered
		}).Create().Execute(ctx); err != nil {
			return fmt.Errorf("create queue %s: %w", name, err)
		}
	}
	fmt.Printf("queues %q and %q ready\n", ordersQueue, invoicesQueue)

	for i := 1; i <= orderCount; i++ {
		if err := pushOrder(ctx, client, fmt.Sprintf("order-%d", i), 100*i); err != nil {
			return err
		}
	}

	if err := stageOne(ctx, client, orderCount); err != nil {
		return err
	}
	if err := stageTwo(ctx, client); err != nil {
		return err
	}
	if err := showFailureIsRedelivered(ctx, client); err != nil {
		return err
	}

	// Housekeeping, not part of the lesson: drop the queues this run created.
	for _, name := range []string{ordersQueue, invoicesQueue} {
		if _, err := client.Queue(name).Delete().Execute(ctx); err != nil {
			return fmt.Errorf("delete queue %s: %w", name, err)
		}
	}
	return nil
}

func pushOrder(ctx context.Context, client *queen.Queen, orderID string, amount int) error {
	responses, err := client.Queue(ordersQueue).
		Push(map[string]interface{}{"orderId": orderID, "amount": amount}).
		Execute(ctx)
	if err != nil {
		return fmt.Errorf("push %s: %w", orderID, err)
	}
	if responses[0].Status != "queued" {
		return fmt.Errorf("push %s: expected status queued, got %q", orderID, responses[0].Status)
	}
	fmt.Printf("pushed %s\n", orderID)
	return nil
}

// stageOne reads orders and hands each one to the invoices queue inside a
// single broker transaction.
func stageOne(ctx context.Context, client *queen.Queen, count int) error {
	for handled := 0; handled < count; handled++ {
		msg, err := popOne(ctx, client, ordersQueue, ordersGroup)
		if err != nil {
			return err
		}
		if msg == nil {
			return fmt.Errorf("only %d of %d orders were delivered", handled, count)
		}

		orderID, amount, err := readOrder(msg)
		if err != nil {
			return err
		}
		invoice := map[string]interface{}{
			"invoiceFor": orderID,
			"total":      amount + shippingFee,
		}

		// The handoff. The acknowledgement of the input and the push of the
		// derived message run in one PostgreSQL transaction, so they either
		// both happen or neither does. Ack first and crash, and the invoice is
		// lost; push first and crash, and the order is invoiced twice on
		// redelivery. This is the shape that has neither failure.
		//
		// The message's leaseId travels with the ack as a required lease, so if
		// the lease expired while this handler was working the commit fails
		// instead of invoicing an order another consumer already owns.
		resp, err := client.Transaction().
			Ack(msg, queen.AckStatusCompleted, queen.AckOptions{ConsumerGroup: ordersGroup}).
			Queue(invoicesQueue).Push(invoice).
			Commit(ctx)
		if err != nil {
			return fmt.Errorf("handoff for %s: %w", orderID, err)
		}
		// Commit reports a rolled back transaction in the body, not as an
		// error, so check Success as well.
		if !resp.Success {
			return fmt.Errorf("handoff for %s rolled back: %s", orderID, resp.Error)
		}
		fmt.Printf("handed off %s as an invoice of %.0f in one transaction\n", orderID, amount+shippingFee)
	}
	return nil
}

// stageTwo reads the invoices back and checks that stage one produced exactly
// one for each order.
func stageTwo(ctx context.Context, client *queen.Queen) error {
	seen := make(map[string]int, orderCount)

	collected := 0
	deadline := time.Now().Add(30 * time.Second)
	for collected < orderCount && time.Now().Before(deadline) {
		msgs, err := client.Queue(invoicesQueue).
			Group(invoicesGroup).
			Batch(orderCount).
			Wait(true).
			TimeoutMillis(5000).
			Pop(ctx)
		if err != nil {
			return fmt.Errorf("pop invoices: %w", err)
		}
		if len(msgs) == 0 {
			continue
		}

		for _, msg := range msgs {
			invoiceFor, ok := msg.Data["invoiceFor"].(string)
			if !ok {
				return fmt.Errorf("invoice %s has an unexpected payload: %v", msg.TransactionID, msg.Data)
			}
			seen[invoiceFor]++
			collected++
			fmt.Printf("invoice for %s: total %v\n", invoiceFor, msg.Data["total"])
		}

		if err := ackAll(ctx, client, msgs, invoicesGroup); err != nil {
			return err
		}
	}

	if collected != orderCount {
		return fmt.Errorf("expected %d invoices, got %d", orderCount, collected)
	}
	for i := 1; i <= orderCount; i++ {
		orderID := fmt.Sprintf("order-%d", i)
		if seen[orderID] != 1 {
			return fmt.Errorf("expected exactly 1 invoice for %s, got %d", orderID, seen[orderID])
		}
	}
	return nil
}

// showFailureIsRedelivered acknowledges an order as failed and checks that the
// broker hands it back instead of dropping it.
func showFailureIsRedelivered(ctx context.Context, client *queen.Queen) error {
	const orderID = "order-4"
	if err := pushOrder(ctx, client, orderID, 400); err != nil {
		return err
	}

	first, err := popOne(ctx, client, ordersQueue, ordersGroup)
	if err != nil {
		return err
	}
	if first == nil {
		return fmt.Errorf("%s was never delivered", orderID)
	}

	// This is what a handler that could not do its job reports. Nothing was
	// pushed to the invoices queue, because the transaction was never
	// committed, and the failure counts against the queue's retry limit.
	acks, err := client.Ack(ctx, first, false, queen.AckOptions{
		ConsumerGroup: ordersGroup,
		Error:         "payment gateway timed out",
	})
	if err != nil {
		return fmt.Errorf("failed ack: %w", err)
	}
	if !acks[0].Success {
		return fmt.Errorf("failed ack rejected: %s", acks[0].Error)
	}
	fmt.Printf("acked %s as failed\n", orderID)

	// The consumer group cursor stays clamped at the failed message, so the
	// very next pop gets it again. A nack costs a redelivery, never the message.
	second, err := popOne(ctx, client, ordersQueue, ordersGroup)
	if err != nil {
		return err
	}
	if second == nil {
		return fmt.Errorf("%s was not redelivered after the failed ack", orderID)
	}
	if second.TransactionID != first.TransactionID {
		return fmt.Errorf("expected %s back, got a different message %s",
			first.TransactionID, second.TransactionID)
	}
	fmt.Printf("redelivered %s (%s)\n", orderID, second.TransactionID)

	// The retry succeeds, so it takes the same transactional handoff as any
	// other order.
	_, amount, err := readOrder(second)
	if err != nil {
		return err
	}
	resp, err := client.Transaction().
		Ack(second, queen.AckStatusCompleted, queen.AckOptions{ConsumerGroup: ordersGroup}).
		Queue(invoicesQueue).Push(map[string]interface{}{
		"invoiceFor": orderID,
		"total":      amount + shippingFee,
	}).
		Commit(ctx)
	if err != nil {
		return fmt.Errorf("handoff for %s: %w", orderID, err)
	}
	if !resp.Success {
		return fmt.Errorf("handoff for %s rolled back: %s", orderID, resp.Error)
	}

	// One invoice, not two: the failed attempt pushed nothing at all.
	invoice, err := popOne(ctx, client, invoicesQueue, invoicesGroup)
	if err != nil {
		return err
	}
	if invoice == nil {
		return fmt.Errorf("the retry of %s produced no invoice", orderID)
	}
	if invoice.Data["invoiceFor"] != orderID {
		return fmt.Errorf("expected an invoice for %s, got %v", orderID, invoice.Data["invoiceFor"])
	}
	if err := ackAll(ctx, client, []*queen.Message{invoice}, invoicesGroup); err != nil {
		return err
	}
	fmt.Printf("invoice for %s: total %v\n", orderID, invoice.Data["total"])

	extra, err := client.Queue(invoicesQueue).
		Group(invoicesGroup).
		Batch(10).
		Wait(false).
		Pop(ctx)
	if err != nil {
		return fmt.Errorf("invoice drain check: %w", err)
	}
	if len(extra) != 0 {
		return fmt.Errorf("expected 1 invoice for %s, found %d more", orderID, len(extra))
	}
	return nil
}

// readOrder pulls the fields out of an order payload. JSON numbers arrive as
// float64 in the untyped payload map.
func readOrder(msg *queen.Message) (string, float64, error) {
	orderID, okID := msg.Data["orderId"].(string)
	amount, okAmount := msg.Data["amount"].(float64)
	if !okID || !okAmount {
		return "", 0, fmt.Errorf("order %s has an unexpected payload: %v", msg.TransactionID, msg.Data)
	}
	return orderID, amount, nil
}

// popOne long polls for a single message and gives up after its own deadline.
// Wait(true) is what makes it a long poll: Go's Pop returns immediately with
// whatever is already there unless you ask it to wait.
func popOne(ctx context.Context, client *queen.Queen, queueName, group string) (*queen.Message, error) {
	deadline := time.Now().Add(20 * time.Second)
	for time.Now().Before(deadline) {
		msgs, err := client.Queue(queueName).
			Group(group).
			Batch(1).
			Wait(true).
			TimeoutMillis(5000).
			Pop(ctx)
		if err != nil {
			return nil, fmt.Errorf("pop %s: %w", queueName, err)
		}
		if len(msgs) > 0 {
			return msgs[0], nil
		}
	}
	return nil, nil
}

// ackAll acknowledges a batch and checks every per-item verdict, because a
// rejected ack still arrives as HTTP 200.
func ackAll(ctx context.Context, client *queen.Queen, msgs []*queen.Message, group string) error {
	acks, err := client.Ack(ctx, msgs, true, queen.AckOptions{ConsumerGroup: group})
	if err != nil {
		return fmt.Errorf("ack: %w", err)
	}
	for i, ack := range acks {
		if !ack.Success {
			return fmt.Errorf("ack rejected for %s: %s", msgs[i].TransactionID, ack.Error)
		}
	}
	return nil
}