Ordering and deduplication

//
// Queen MQ by example, 2 of 3: ordering and deduplication.
//
// Part one: one partition per customer. Messages for different customers are
// pushed interleaved, but each customer's messages come back in the order they
// were pushed, because a partition is the unit of ordering.
//
// Part two: pushing the same transactionId twice. The broker answers `duplicate`
// for the second push and stores nothing new.
//
// Run it:
//   QUEEN_URL=http://localhost:6699 node 02-ordering-and-dedup.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 so the three languages never collide, suffixed per run
// so a second run starts from a genuinely empty queue.
const RUN_ID = Date.now().toString(36)
const QUEUE = `ex-js-ordering-dedup-${RUN_ID}`
const GROUP = 'ex-js-auditor'

// Three customers, three partitions. Ordering in Queen is per partition, never
// global, so the partition key is the identity whose history must stay in order.
const CUSTOMERS = ['acme', 'globex', 'initech']
const EVENTS_PER_CUSTOMER = 4

// The dedup demonstration gets its own partition so it can be counted on its own.
const DEDUP_PARTITION = 'dedup-demo'

let checks = 0

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

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

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

  await queen.queue(QUEUE).config({
    leaseTime: 30,
    retryLimit: 3,
    // How long a transactionId is remembered for deduplication, in seconds.
    dedupWindowSeconds: 600,
  }).create()
  console.log(`queue ${QUEUE} created`)

  // ---------------------------------------------------------------- ordering

  // Build a global sequence that interleaves the customers: acme#0, globex#0,
  // initech#0, acme#1, ... Nothing in Queen preserves this global order, and the
  // point of the example is that nothing has to.
  const interleaved = []
  for (let seq = 0; seq < EVENTS_PER_CUSTOMER; seq++) {
    for (const customer of CUSTOMERS) {
      interleaved.push({ customer, seq, event: `${customer}-event-${seq}` })
    }
  }

  console.log('\npushing interleaved')
  const pushedPerCustomer = new Map(CUSTOMERS.map(c => [c, []]))
  for (const item of interleaved) {
    // partition(customer) puts this message in that customer's lane. Messages
    // in one lane are delivered in push order; two lanes have no order between
    // them and can be consumed in parallel.
    const [result] = await queen.queue(QUEUE).partition(item.customer).push([{ data: item }])
    if (result.status !== 'queued') throw new Error(`push of ${item.event} answered ${result.status}`)
    pushedPerCustomer.get(item.customer).push(item.event)
    console.log(`  ${item.event} -> partition ${item.customer}`)
  }

  console.log('\nconsuming')
  const arrivedPerCustomer = new Map(CUSTOMERS.map(c => [c, []]))
  const globalArrival = []
  await queen.queue(QUEUE)
    .group(GROUP)
    .each()
    .batch(interleaved.length)
    // Claim up to three partitions per poll. All of them share one lease, and
    // messages still arrive in order within each partition.
    .partitions(CUSTOMERS.length)
    .autoAck(false)
    .limit(interleaved.length)
    .idleMillis(5000)
    .consume(async (msg) => {
      arrivedPerCustomer.get(msg.data.customer).push(msg.data.event)
      globalArrival.push(msg.data.event)
      const ack = await queen.ack(msg, true, { group: GROUP })
      if (!ack.success) throw new Error(`ack rejected: ${ack.error}`)
    })

  console.log(`  push order:    ${interleaved.map(i => i.event).join(' ')}`)
  console.log(`  arrival order: ${globalArrival.join(' ')}`)

  console.log('\nchecking ordering')
  assert(globalArrival.length === interleaved.length, `consumed all ${interleaved.length} messages`)
  for (const customer of CUSTOMERS) {
    const pushed = pushedPerCustomer.get(customer).join(' ')
    const arrived = arrivedPerCustomer.get(customer).join(' ')
    assert(pushed === arrived, `partition ${customer} kept its order: ${arrived}`)
  }
  // The global sequence is allowed to differ, and here it does: the consumer
  // drains a partition at a time. Ordering was promised per partition only.
  assert(
    globalArrival.join(' ') !== interleaved.map(i => i.event).join(' '),
    'the global order differs from the push order, as documented'
  )

  // ------------------------------------------------------------------- dedup

  console.log('\npushing the same transactionId twice')

  // A transactionId is the idempotency key. Supplying your own means a retry
  // after a lost response cannot create a second message. It only has to be
  // unique within this queue's dedup window, so a business id is the usual pick.
  const invoiceId = 'ex-js-invoice-7001'
  const payload = { data: { invoice: invoiceId, amount: 42.5 }, transactionId: invoiceId }

  const [first] = await queen.queue(QUEUE).partition(DEDUP_PARTITION).push([payload])
  console.log(`  first push:  status=${first.status} message_id=${first.message_id}`)

  // The identical push again, as an at-least-once producer would send it.
  const [second] = await queen.queue(QUEUE).partition(DEDUP_PARTITION).push([payload])
  console.log(`  second push: status=${second.status} message_id=${second.message_id}`)

  console.log('\nchecking dedup')
  assert(first.status === 'queued', 'the first push was queued')
  assert(second.status === 'duplicate', 'the second push was answered as duplicate')
  // The broker echoes the id of the message it already holds rather than
  // minting a new one, so the producer can still resolve what it pushed.
  assert(second.message_id === first.message_id, 'the duplicate echoes the original message id')

  // Nothing new was stored: the partition holds exactly one message.
  const stored = await queen.queue(QUEUE)
    .partition(DEDUP_PARTITION)
    .group(GROUP)
    .batch(10)
    .wait(true)
    .timeoutMillis(3000)
    .pop()
  assert(stored.length === 1, 'the partition holds exactly one message, not two')
  assert(stored[0].transactionId === invoiceId, 'that message is the one that was pushed first')
  const ack = await queen.ack(stored, true, { group: GROUP })
  assert(ack.success, 'the surviving message acknowledges cleanly')

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

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

"""Queen in Python: per-entity ordering, and dedup by transaction id.

One partition per customer. Messages for a customer come back in the order they
were pushed, even though the three customers are interleaved on the wire. Then
the same message is pushed twice under one transaction id, and the broker
answers "duplicate" without storing anything new.

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/02_ordering_and_dedup.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 queue name per run keeps this run independent of every previous one.
RUN_ID = uuid.uuid4().hex[:8]
QUEUE = f"ex-py-ordering-dedup-{RUN_ID}"
GROUP = "audit"

# A partition is Queen's ordering lane. Order is guaranteed inside a partition
# and nowhere else, so the partition key is the entity whose history must stay
# in sequence: here, one customer.
CUSTOMERS = ["cust-acme", "cust-globex", "cust-initech"]
STEPS = ["created", "paid", "packed", "shipped"]


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"queue  {QUEUE}\n")

        await (
            client.queue(QUEUE)
            .config(
                {
                    "lease_time": 60,
                    "retry_limit": 3,
                    # The window the broker keeps transaction ids in for the
                    # duplicate check. Outside it, the same id is new work again.
                    "dedup_window_seconds": 300,
                }
            )
            .create()
        )
        print(f"      created {QUEUE} with a 300s dedup window\n")

        # ------------------------------------------------------------------
        # 1. Push, interleaved across the three customers.
        # ------------------------------------------------------------------
        # Step by step rather than customer by customer, so no customer's
        # messages are contiguous on the wire.
        push_order = []
        statuses = []
        for step in STEPS:
            for customer in CUSTOMERS:
                event = {"customer": customer, "step": step}
                # .partition(customer) puts this message in that customer's lane.
                result = await client.queue(QUEUE).partition(customer).push({"data": event})
                statuses.append(result[0]["status"])
                push_order.append(f"{customer}:{step}")

        print(f"      push order: {' '.join(push_order)}")
        check(len(push_order) == len(CUSTOMERS) * len(STEPS), f"pushed {len(push_order)} messages")
        check(all(s == "queued" for s in statuses), "every push was accepted as new work")

        # ------------------------------------------------------------------
        # 2. Consume all of them.
        # ------------------------------------------------------------------
        arrival = []

        async def handle(messages):
            # batch(n) with n > 1 hands the Python handler a list. partitions(3)
            # lets one pop claim up to three lanes at once, so a single call can
            # drain all three customers; the messages stay grouped by lane.
            for message in messages:
                arrival.append(message)
            lanes = sorted({m["partition"] for m in messages})
            print(f"      received {len(messages)} messages from lanes {lanes}")

        await (
            client.queue(QUEUE)
            .group(GROUP)
            .batch(len(push_order))
            .partitions(len(CUSTOMERS))
            .limit(len(push_order))
            .idle_millis(5000)
            .timeout_millis(2000)
            .consume(handle)
        )

        arrival_order = [f"{m['partition']}:{m['data']['step']}" for m in arrival]
        print(f"      arrival order: {' '.join(arrival_order)}\n")

        # ------------------------------------------------------------------
        # 3. Verify the ordering guarantee, per lane.
        # ------------------------------------------------------------------
        check(len(arrival) == len(push_order), f"all {len(push_order)} messages came back")

        for customer in CUSTOMERS:
            seen = [m["data"]["step"] for m in arrival if m["partition"] == customer]
            check(seen == STEPS, f"{customer} arrived in push order: {' -> '.join(seen)}")

        # The global sequence is free to differ from the push sequence (it is
        # grouped by lane above), and that is exactly the trade: ordering is a
        # per-partition promise, which is what lets partitions run in parallel.

        # ------------------------------------------------------------------
        # 4. Push the same message twice under one transaction id.
        # ------------------------------------------------------------------
        # The transaction id is the message's identity for the duplicate check.
        # Supply the id your business already has, and a retried push (a timeout,
        # a redelivered webhook, a restarted job) costs nothing and creates
        # nothing. Left unset, the client mints a fresh UUIDv7 per push, which
        # can never be a duplicate.
        payment_id = f"payment-{RUN_ID}-7781"
        payment = {"customer": "cust-acme", "step": "refunded", "paymentId": payment_id}

        first = await (
            client.queue(QUEUE)
            .partition("cust-acme")
            .push({"transactionId": payment_id, "data": payment})
        )
        print(f"      first push of {payment_id}: {first[0]['status']}")
        check(first[0]["status"] == "queued", "the first push of a transaction id is queued")

        # Byte-for-byte the same push, as a retry would send it.
        second = await (
            client.queue(QUEUE)
            .partition("cust-acme")
            .push({"transactionId": payment_id, "data": payment})
        )
        print(f"      second push of {payment_id}: {second[0]['status']}")
        check(second[0]["status"] == "duplicate", "the second push is rejected as a duplicate")
        check(
            second[0]["message_id"] == first[0]["message_id"],
            "the duplicate points at the original message rather than a new one",
        )

        # ------------------------------------------------------------------
        # 5. Verify the duplicate stored nothing.
        # ------------------------------------------------------------------
        # If the second push had been stored, this pop would return two messages.
        tail = await (
            client.queue(QUEUE).group(GROUP).batch(10).wait(True).timeout_millis(2000).pop()
        )
        check(len(tail) == 1, f"exactly one new message exists, not two (got {len(tail)})")
        check(tail[0]["transactionId"] == payment_id, "and it is the one pushed first")

        # Acking is a manual call when you pop rather than consume. partitionId
        # comes off the message and is mandatory: it is what makes the ack refer
        # to one message and no other.
        acked = await client.ack(tail[0], True, {"group": GROUP})
        check(acked.get("success") is True, "the refund message was acknowledged")

        deleted = await client.queue(QUEUE).delete()
        check(deleted.get("deleted") is True, f"queue {QUEUE} deleted")

        print("\nPASS 02_ordering_and_dedup")


asyncio.run(main())

// Ordering and deduplication: the two properties people come to Queen for.
//
// Part one gives every customer its own partition, pushes their events
// interleaved, and shows that each customer's events come back in the order
// they were pushed even though the global stream is interleaved.
//
// Part two pushes the same transactionId twice and shows the broker answering
// "duplicate" and storing nothing new.
//
// Run it with:
//
//	QUEEN_URL=http://localhost:6699 go run ./ordering-and-dedup
package main

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

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

const (
	consumerGroup = "ex-go-audit"

	eventsPerCustomer = 4

	// The dedup demonstration gets its own partition so that popping it does
	// not disturb the ordering assertions above.
	dedupPartition = "payments"
)

// A per-run suffix gives every run a queue that has never existed before: no
// messages, no consumer group cursor and, for the second half of this example,
// no deduplication history.
var runID = time.Now().UnixMilli()

var queueName = fmt.Sprintf("ex-go-ordering-dedup-%d", runID)

// A transactionId you choose yourself is what makes a push idempotent: the
// broker rejects a second push carrying an id it has already accepted inside
// the queue's dedup window. Derive it from your own data (here, the payment it
// stands for) so that a retry recomputes the same id.
var paymentTxID = fmt.Sprintf("ex-go-payment-%d", runID)

// One partition per customer. A partition is the unit of ordering, so this is
// the choice that buys per-customer FIFO and cross-customer parallelism at the
// same time.
var customers = []string{"customer-a", "customer-b", "customer-c"}

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

	if err := run(ctx); err != nil {
		fmt.Fprintf(os.Stderr, "FAILED: %v\n", err)
		os.Exit(1)
	}
	fmt.Println("OK: per partition order held and the duplicate push was rejected")
}

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)

	if _, err := client.Queue(queueName).Config(queen.QueueConfig{
		LeaseTime:  60,
		RetryLimit: 3,
	}).Create().Execute(ctx); err != nil {
		return fmt.Errorf("create queue: %w", err)
	}
	fmt.Printf("queue %q ready\n", queueName)

	if err := showOrdering(ctx, client); err != nil {
		return err
	}
	if err := showDedup(ctx, client); err != nil {
		return err
	}

	// Housekeeping, not part of the lesson: drop the queue this run created.
	if _, err := client.Queue(queueName).Delete().Execute(ctx); err != nil {
		return fmt.Errorf("delete queue: %w", err)
	}
	return nil
}

// showOrdering pushes interleaved events for three customers and checks that
// each customer's events come back in push order.
func showOrdering(ctx context.Context, client *queen.Queen) error {
	total := len(customers) * eventsPerCustomer

	// Push round by round, so the queue as a whole sees a-1, b-1, c-1, a-2, ...
	var pushOrder []string
	for seq := 1; seq <= eventsPerCustomer; seq++ {
		for _, customer := range customers {
			responses, err := client.Queue(queueName).
				Partition(customer).
				Push(map[string]interface{}{"customer": customer, "seq": seq}).
				Execute(ctx)
			if err != nil {
				return fmt.Errorf("push %s#%d: %w", customer, seq, err)
			}
			if responses[0].Status != "queued" {
				return fmt.Errorf("push %s#%d: expected status queued, got %q",
					customer, seq, responses[0].Status)
			}
			pushOrder = append(pushOrder, fmt.Sprintf("%s#%d", customer, seq))
		}
	}
	fmt.Printf("pushed:   %s\n", strings.Join(pushOrder, " "))

	delivered := make(map[string][]int, len(customers))
	var deliveryOrder []string

	consumed := 0
	deadline := time.Now().Add(30 * time.Second)
	for consumed < total && time.Now().Before(deadline) {
		// Partitions(3) claims up to three partitions in one call and Batch is
		// the global budget across them, so a single round trip can carry
		// several customers. They share one lease.
		msgs, err := client.Queue(queueName).
			Group(consumerGroup).
			Partitions(len(customers)).
			Batch(total).
			Wait(true).
			TimeoutMillis(5000).
			Pop(ctx)
		if err != nil {
			return fmt.Errorf("pop: %w", err)
		}
		if len(msgs) == 0 {
			continue
		}

		for _, msg := range msgs {
			// JSON numbers arrive as float64 in the untyped payload map.
			seq, ok := msg.Data["seq"].(float64)
			if !ok {
				return fmt.Errorf("message %s has no numeric seq", msg.TransactionID)
			}
			if msg.Partition == dedupPartition {
				return fmt.Errorf("unexpected message from the %s partition", dedupPartition)
			}
			delivered[msg.Partition] = append(delivered[msg.Partition], int(seq))
			deliveryOrder = append(deliveryOrder, fmt.Sprintf("%s#%d", msg.Partition, int(seq)))
			consumed++
		}

		acks, err := client.Ack(ctx, msgs, true, queen.AckOptions{ConsumerGroup: consumerGroup})
		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)
			}
		}
	}
	fmt.Printf("consumed: %s\n", strings.Join(deliveryOrder, " "))

	if consumed != total {
		return fmt.Errorf("expected %d messages, consumed %d", total, consumed)
	}

	// The guarantee is per partition, not global: the interleaving above may
	// differ from the push order, but each customer's own sequence may not.
	for _, customer := range customers {
		seqs := delivered[customer]
		if len(seqs) != eventsPerCustomer {
			return fmt.Errorf("%s: expected %d events, got %d", customer, eventsPerCustomer, len(seqs))
		}
		for i, seq := range seqs {
			if seq != i+1 {
				return fmt.Errorf("%s: out of order, expected seq %d at position %d, got %d",
					customer, i+1, i, seq)
			}
		}
		fmt.Printf("%s: in order %v\n", customer, seqs)
	}

	return nil
}

// showDedup pushes the same transactionId twice and checks that the second
// push is rejected as a duplicate without storing anything.
func showDedup(ctx context.Context, client *queen.Queen) error {
	first, err := client.Queue(queueName).
		Partition(dedupPartition).
		Push(map[string]interface{}{"paymentId": "payment-42", "amount": 100}).
		TransactionID(paymentTxID).
		Execute(ctx)
	if err != nil {
		return fmt.Errorf("first payment push: %w", err)
	}
	if first[0].Status != "queued" {
		return fmt.Errorf("first payment push: expected status queued, got %q", first[0].Status)
	}
	fmt.Printf("first push of %s: %s\n", paymentTxID, first[0].Status)

	// Same id, different payload: this is the retry of a producer that never
	// saw the first response. The broker must not charge the customer twice.
	second, err := client.Queue(queueName).
		Partition(dedupPartition).
		Push(map[string]interface{}{"paymentId": "payment-42", "amount": 999}).
		TransactionID(paymentTxID).
		Execute(ctx)
	if err != nil {
		return fmt.Errorf("second payment push: %w", err)
	}
	if second[0].Status != "duplicate" {
		return fmt.Errorf("second payment push: expected status duplicate, got %q", second[0].Status)
	}
	fmt.Printf("second push of %s: %s\n", paymentTxID, second[0].Status)

	// "Duplicate" has to mean nothing was stored, so read the partition back:
	// exactly one message, carrying the first payload rather than the second.
	msgs, err := popPayments(ctx, client)
	if err != nil {
		return err
	}
	if len(msgs) != 1 {
		return fmt.Errorf("expected exactly 1 payment message, got %d", len(msgs))
	}
	if msgs[0].TransactionID != paymentTxID {
		return fmt.Errorf("expected transactionId %s, got %s", paymentTxID, msgs[0].TransactionID)
	}
	amount, ok := msgs[0].Data["amount"].(float64)
	if !ok || amount != 100 {
		return fmt.Errorf("expected the first payload (amount 100), got %v", msgs[0].Data["amount"])
	}
	fmt.Printf("stored payment: amount %v\n", msgs[0].Data["amount"])

	acks, err := client.Ack(ctx, msgs[0], true, queen.AckOptions{ConsumerGroup: consumerGroup})
	if err != nil {
		return fmt.Errorf("ack payment: %w", err)
	}
	if !acks[0].Success {
		return fmt.Errorf("ack rejected for %s: %s", msgs[0].TransactionID, acks[0].Error)
	}

	// And nothing else is hiding behind it.
	leftovers, err := client.Queue(queueName).
		Partition(dedupPartition).
		Group(consumerGroup).
		Batch(10).
		Wait(false).
		Pop(ctx)
	if err != nil {
		return fmt.Errorf("payment drain check: %w", err)
	}
	if len(leftovers) != 0 {
		return fmt.Errorf("expected the payments partition to be empty, got %d message(s)", len(leftovers))
	}

	return nil
}

// popPayments reads the payments partition. Naming a partition other than the
// default switches the pop to the partition scoped route, so only that
// customer's lane is touched.
func popPayments(ctx context.Context, client *queen.Queen) ([]*queen.Message, error) {
	deadline := time.Now().Add(20 * time.Second)
	for time.Now().Before(deadline) {
		msgs, err := client.Queue(queueName).
			Partition(dedupPartition).
			Group(consumerGroup).
			Batch(10).
			Wait(true).
			TimeoutMillis(5000).
			Pop(ctx)
		if err != nil {
			return nil, fmt.Errorf("pop payments: %w", err)
		}
		if len(msgs) > 0 {
			return msgs, nil
		}
	}
	return nil, fmt.Errorf("no payment message was delivered")
}