Chat messaging system

//
// A chat messaging system.
//
// This is the application Queen was written for. A hotel messaging product ran
// on Kafka and kept stalling: some conversations need a translation or an agent
// reply before the next message can be handled, and on a shared partition one
// slow conversation holds up every conversation behind it.
//
// The fix is structural rather than operational: one ordered lane per
// conversation, created by the first message sent to it. A conversation that
// takes ten seconds delays itself and nothing else.
//
// What this program builds:
//
//   chat-messages (one partition per conversation)
//     ├── group "delivery"    fast, marks each message as delivered
//     └── group "enrichment"  slow on conversations that need translation
//
// And what it proves: every message reaches both groups exactly once, in the
// order it was sent inside its own conversation, and the conversations that
// need no translation finish while the slow one is still working.
//
// Run it:
//   QUEEN_URL=http://localhost:6632 node chat.mjs

import { Queen } from 'queen-mq'

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

// Three conversations. The one in Japanese needs a translation pass, which is
// the slow work: 400 ms a message against 10 ms for the rest.
const CONVERSATIONS = {
  'conv-en-1': { locale: 'en', needsTranslation: false },
  'conv-en-2': { locale: 'en', needsTranslation: false },
  'conv-jp-1': { locale: 'jp', needsTranslation: true },
}
const MESSAGES_PER_CONVERSATION = 6

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

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

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

  // Leases are what make a crashed worker safe: a message whose handler dies is
  // redelivered once the lease expires. retryLimit bounds how many times that
  // can happen before the message is dead-lettered instead.
  await queen.queue(MESSAGES).config({ leaseTime: 60, retryLimit: 3 }).create()

  // ---------------------------------------------------------------- producing
  //
  // A chat client sends a message: one push, into the partition named after the
  // conversation. Nothing was declared for this conversation in advance, and
  // nothing has to be cleaned up when it goes quiet.
  console.log('\nsending')
  const sent = []
  for (let seq = 1; seq <= MESSAGES_PER_CONVERSATION; seq++) {
    for (const [conversationId, meta] of Object.entries(CONVERSATIONS)) {
      const message = {
        conversationId,
        seq,
        locale: meta.locale,
        body: `message ${seq} in ${conversationId}`,
        // The transaction id is the client's own idempotency key: a retry of
        // this send, from a phone on a flaky network, writes nothing the second
        // time and answers with the first message's id.
        sentAt: Date.now(),
      }
      await queen.queue(MESSAGES).partition(conversationId).push({
        transactionId: `${conversationId}-${seq}`,
        data: message,
      })
      sent.push(message)
    }
  }
  console.log(`  ${sent.length} messages across ${Object.keys(CONVERSATIONS).length} conversations`)

  // A resend of the same message: the client retried because it never saw the
  // first answer. The broker recognises the transaction id and stores nothing.
  const [duplicate] = await queen.queue(MESSAGES).partition('conv-en-1').push({
    transactionId: 'conv-en-1-1',
    data: { conversationId: 'conv-en-1', seq: 1, body: 'resent by the phone' },
  })
  assert(duplicate.status === 'duplicate', 'a resent message was deduplicated, not stored twice')

  // --------------------------------------------------------------- delivering
  //
  // The delivery worker is what marks a message as delivered to the recipients.
  // It is fast and must never fall behind, which is why it is its own consumer
  // group: it shares no cursor with the slow work below.
  //
  // concurrency(3) runs three poll loops, and each pop claims a partition, so
  // the three conversations are drained in parallel by three workers.
  console.log('\ndelivering')
  const delivered = new Map()
  await queen
    .queue(MESSAGES)
    .group('delivery')
    .subscriptionMode('all')
    .concurrency(3)
    .each()
    .limit(sent.length)
    .idleMillis(10000)
    .consume(async (msg) => {
      await sleep(10)
      const seen = delivered.get(msg.data.conversationId) ?? []
      seen.push(msg.data.seq)
      delivered.set(msg.data.conversationId, seen)
    })

  assert(
    [...delivered.values()].reduce((n, seqs) => n + seqs.length, 0) === sent.length,
    'delivery saw every message exactly once'
  )
  for (const [conversationId, seqs] of delivered) {
    assert(
      JSON.stringify(seqs) === JSON.stringify([...seqs].sort((a, b) => a - b)),
      `${conversationId} was delivered in order`
    )
  }

  // -------------------------------------------------------------- enrichment
  //
  // The slow group. It reads the same messages through its own cursor, and the
  // Japanese conversation costs 400 ms a message because it has to be
  // translated before it can be answered.
  //
  // This is where a shared partition would hurt: on a hashed topic these
  // messages would sit in the same lane as the English ones and hold them up.
  // Here each conversation has its own lane, so the English conversations
  // finish while the Japanese one is still being translated. The timings below
  // are the proof.
  console.log('\nenriching')
  const finishedAt = new Map()
  const started = Date.now()
  await queen
    .queue(MESSAGES)
    .group('enrichment')
    .subscriptionMode('all')
    .concurrency(3)
    .each()
    .limit(sent.length)
    .idleMillis(15000)
    .consume(async (msg) => {
      const meta = CONVERSATIONS[msg.data.conversationId]
      await sleep(meta.needsTranslation ? 400 : 10)
      finishedAt.set(msg.data.conversationId, Date.now() - started)
    })

  const slow = finishedAt.get('conv-jp-1')
  const fast = Math.max(finishedAt.get('conv-en-1'), finishedAt.get('conv-en-2'))
  console.log(`  english done after ${fast} ms, japanese after ${slow} ms`)

  assert(
    fast < slow,
    'the conversations needing no translation finished first, in the same worker pool'
  )
  assert(
    slow > MESSAGES_PER_CONVERSATION * 300,
    'the slow conversation really was slow, so the comparison means something'
  )

  // ------------------------------------------------------------------- replay
  //
  // A new feature needs the history: sentiment scoring over everything ever
  // said. It is a new consumer group reading from the beginning, and it costs
  // no producer change and no second copy of the data.
  console.log('\nbackfilling a new consumer')
  let scored = 0
  await queen
    .queue(MESSAGES)
    .group('sentiment')
    .subscriptionMode('all')
    .concurrency(3)
    .each()
    .limit(sent.length)
    .idleMillis(10000)
    .consume(async () => { scored++ })

  assert(scored === sent.length, 'a group added today read the whole history')

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

#
# A chat messaging system.
#
# This is the application Queen was written for. A hotel messaging product ran
# on Kafka and kept stalling: some conversations need a translation or an agent
# reply before the next message can be handled, and on a shared partition one
# slow conversation holds up every conversation behind it.
#
# The fix is structural rather than operational: one ordered lane per
# conversation, created by the first message sent to it. A conversation that
# takes ten seconds delays itself and nothing else.
#
# What this program builds:
#
#   chat-messages (one partition per conversation)
#     |-- group "delivery"    fast, marks each message as delivered
#     `-- group "enrichment"  slow on conversations that need translation
#
# And what it proves: every message reaches both groups exactly once, in the
# order it was sent inside its own conversation, and the conversations that
# need no translation finish while the slow one is still working.
#
# Run it:
#   QUEEN_URL=http://localhost:6632 python3 chat.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}"
MESSAGES = f"app-py-chat-{RUN}"

# Three conversations. The one in Japanese needs a translation pass, which is
# the slow work: 400 ms a message against 10 ms for the rest.
CONVERSATIONS = {
    "conv-en-1": {"locale": "en", "needs_translation": False},
    "conv-en-2": {"locale": "en", "needs_translation": False},
    "conv-jp-1": {"locale": "jp", "needs_translation": True},
}
MESSAGES_PER_CONVERSATION = 6

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 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;
    # the orderly shutdown of a run that ends normally is close(), at the bottom.
    queen = Queen(url=QUEEN_URL)
    verdict, failed = "", False

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

        # Leases are what make a crashed worker safe: a message whose handler
        # dies is redelivered once the lease expires. retry_limit bounds how
        # many times that can happen before the message is dead-lettered
        # instead. The config keys are snake_case in Python and the client
        # converts them to the camelCase the broker expects.
        await queen.queue(MESSAGES).config({"lease_time": 60, "retry_limit": 3}).create()

        # ---------------------------------------------------------- producing
        #
        # A chat client sends a message: one push, into the partition named
        # after the conversation. Nothing was declared for this conversation in
        # advance, and nothing has to be cleaned up when it goes quiet.
        print("\nsending")
        sent = []
        for seq in range(1, MESSAGES_PER_CONVERSATION + 1):
            for conversation_id, meta in CONVERSATIONS.items():
                message = {
                    "conversationId": conversation_id,
                    "seq": seq,
                    "locale": meta["locale"],
                    "body": f"message {seq} in {conversation_id}",
                    "sentAt": int(time.time() * 1000),
                }
                # The transaction id is the client's own idempotency key: a
                # retry of this send, from a phone on a flaky network, writes
                # nothing the second time and answers with the first message's
                # id. The item key stays camelCase here, because it is the wire
                # name rather than a client option.
                await queen.queue(MESSAGES).partition(conversation_id).push(
                    {"transactionId": f"{conversation_id}-{seq}", "data": message}
                )
                sent.append(message)
        print(f"  {len(sent)} messages across {len(CONVERSATIONS)} conversations")

        # A resend of the same message: the client retried because it never saw
        # the first answer. The broker recognises the transaction id and stores
        # nothing. What comes back is the broker's own reply, one entry per
        # item, with the broker's own key names.
        results = await queen.queue(MESSAGES).partition("conv-en-1").push(
            {
                "transactionId": "conv-en-1-1",
                "data": {"conversationId": "conv-en-1", "seq": 1, "body": "resent by the phone"},
            }
        )
        check(
            results[0]["status"] == "duplicate",
            "a resent message was deduplicated, not stored twice",
        )

        # --------------------------------------------------------- delivering
        #
        # The delivery worker is what marks a message as delivered to the
        # recipients. It is fast and must never fall behind, which is why it is
        # its own consumer group: it shares no cursor with the slow work below.
        #
        # concurrency(3) runs three poll loops, and each pop claims a partition,
        # so the three conversations are drained in parallel by three workers.
        # The handler is an async def taking one message: consume() awaits it
        # for every message and acknowledges on return.
        print("\ndelivering")
        delivered: dict = {}

        async def deliver(msg) -> None:
            await asyncio.sleep(0.01)
            delivered.setdefault(msg["data"]["conversationId"], []).append(msg["data"]["seq"])

        await (
            queen.queue(MESSAGES)
            .group("delivery")
            # 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()
            .limit(len(sent))
            # Stop after 10s of silence, so a lost message fails the run instead
            # of hanging it.
            .idle_millis(10000)
            .consume(deliver)
        )

        check(
            sum(len(seqs) for seqs in delivered.values()) == len(sent),
            "delivery saw every message exactly once",
        )
        for conversation_id, seqs in delivered.items():
            check(seqs == sorted(seqs), f"{conversation_id} was delivered in order")

        # --------------------------------------------------------- enrichment
        #
        # The slow group. It reads the same messages through its own cursor, and
        # the Japanese conversation costs 400 ms a message because it has to be
        # translated before it can be answered.
        #
        # This is where a shared partition would hurt: on a hashed topic these
        # messages would sit in the same lane as the English ones and hold them
        # up. Here each conversation has its own lane, so the English
        # conversations finish while the Japanese one is still being translated.
        # The timings below are the proof.
        print("\nenriching")
        finished_at: dict = {}
        # monotonic() rather than time(): these are durations, and a clock that
        # steps sideways mid-run must not turn a real ordering into a fake one.
        started = time.monotonic()

        async def enrich(msg) -> None:
            meta = CONVERSATIONS[msg["data"]["conversationId"]]
            await asyncio.sleep(0.4 if meta["needs_translation"] else 0.01)
            finished_at[msg["data"]["conversationId"]] = int((time.monotonic() - started) * 1000)

        await (
            queen.queue(MESSAGES)
            .group("enrichment")
            .subscription_mode("all")
            .concurrency(3)
            .each()
            .limit(len(sent))
            .idle_millis(15000)
            .consume(enrich)
        )

        slow = finished_at["conv-jp-1"]
        fast = max(finished_at["conv-en-1"], finished_at["conv-en-2"])
        print(f"  english done after {fast} ms, japanese after {slow} ms")

        check(
            fast < slow,
            "the conversations needing no translation finished first, in the same worker pool",
        )
        check(
            slow > MESSAGES_PER_CONVERSATION * 300,
            "the slow conversation really was slow, so the comparison means something",
        )

        # -------------------------------------------------------------- replay
        #
        # A new feature needs the history: sentiment scoring over everything ever
        # said. It is a new consumer group reading from the beginning, and it
        # costs no producer change and no second copy of the data.
        print("\nbackfilling a new consumer")
        scored = 0

        async def score(msg) -> None:
            nonlocal scored
            scored += 1

        await (
            queen.queue(MESSAGES)
            .group("sentiment")
            .subscription_mode("all")
            .concurrency(3)
            .each()
            .limit(len(sent))
            .idle_millis(10000)
            .consume(score)
        )

        check(scored == len(sent), "a group added today read the whole history")

        # Clean up on success only: a failed run leaves the queue on the broker
        # to be looked at.
        await queen.queue(MESSAGES).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()))

//
// A chat messaging system.
//
// This is the application Queen was written for. A hotel messaging product ran
// on Kafka and kept stalling: some conversations need a translation or an agent
// reply before the next message can be handled, and on a shared partition one
// slow conversation holds up every conversation behind it.
//
// The fix is structural rather than operational: one ordered lane per
// conversation, created by the first message sent to it. A conversation that
// takes ten seconds delays itself and nothing else.
//
// What this program builds:
//
//	chat-messages (one partition per conversation)
//	  |-- group "delivery"    fast, marks each message as delivered
//	  |-- group "enrichment"  slow on conversations that need translation
//
// And what it proves: every message reaches both groups exactly once, in the
// order it was sent inside its own conversation, and the conversations that
// need no translation finish while the slow one is still working.
//
// Run it:
//
//	QUEEN_URL=http://localhost:6632 GOWORK=off go run ./chat
package main

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

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

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

var messagesQueue = "app-go-chat-" + runID

// Three conversations. The one in Japanese needs a translation pass, which is
// the slow work: 400 ms a message against 10 ms for the rest. The list is a
// slice rather than a map because Go randomises map iteration and the send
// order below has to be the same on every run.
type conversation struct {
	id               string
	locale           string
	needsTranslation bool
}

var conversations = []conversation{
	{id: "conv-en-1", locale: "en", needsTranslation: false},
	{id: "conv-en-2", locale: "en", needsTranslation: false},
	{id: "conv-jp-1", locale: "jp", needsTranslation: true},
}

const messagesPerConversation = 6

func conversationByID(id string) (conversation, bool) {
	for _, c := range conversations {
		if c.id == id {
			return c, true
		}
	}
	return conversation{}, false
}

var checks int

// assert is the whole test framework here. Go has no exceptions, so a failed
// check is an error that unwinds run() and is printed once, at the bottom.
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"
	}

	// Every call in the Go client takes a context, and it is the only deadline
	// there is. This one bounds the whole program, so a broker that stops
	// answering ends the run instead of wedging it.
	ctx, cancel := context.WithTimeout(context.Background(), 180*time.Second)
	defer cancel()

	// queen.New installs no signal handlers: this program owns its shutdown
	// through the Close below.
	client, err := queen.New(brokerURL)
	if err != nil {
		return fmt.Errorf("create client: %w", err)
	}
	// Close gets a fresh context, because the one above may already be done by
	// the time the deferred call runs.
	defer client.Close(context.Background())

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

	// Leases are what make a crashed worker safe: a message whose handler dies
	// is redelivered once the lease expires. RetryLimit bounds how many times
	// that can happen before the message is dead-lettered instead.
	if _, err := client.Queue(messagesQueue).
		Config(queen.QueueConfig{LeaseTime: 60, RetryLimit: 3}).
		Create().Execute(ctx); err != nil {
		return fmt.Errorf("create %s: %w", messagesQueue, err)
	}

	// ---------------------------------------------------------------- producing
	//
	// A chat client sends a message: one push, into the partition named after
	// the conversation. Nothing was declared for this conversation in advance,
	// and nothing has to be cleaned up when it goes quiet.
	fmt.Println("\nsending")
	sent := 0
	for seq := 1; seq <= messagesPerConversation; seq++ {
		for _, c := range conversations {
			message := map[string]interface{}{
				"conversationId": c.id,
				"seq":            seq,
				"locale":         c.locale,
				"body":           fmt.Sprintf("message %d in %s", seq, c.id),
				"sentAt":         time.Now().UnixMilli(),
			}
			// The transaction id is the client's own idempotency key: a retry
			// of this send, from a phone on a flaky network, writes nothing the
			// second time and answers with the first message's id. In this
			// client it rides on the push builder rather than on the payload.
			if _, err := client.Queue(messagesQueue).
				Partition(c.id).
				Push(message).
				TransactionID(fmt.Sprintf("%s-%d", c.id, seq)).
				Execute(ctx); err != nil {
				return fmt.Errorf("push %s/%d: %w", c.id, seq, err)
			}
			sent++
		}
	}
	fmt.Printf("  %d messages across %d conversations\n", sent, len(conversations))

	// A resend of the same message: the client retried because it never saw the
	// first answer. The broker recognises the transaction id and stores nothing.
	duplicate, err := client.Queue(messagesQueue).
		Partition("conv-en-1").
		Push(map[string]interface{}{
			"conversationId": "conv-en-1",
			"seq":            1,
			"body":           "resent by the phone",
		}).
		TransactionID("conv-en-1-1").
		Execute(ctx)
	if err != nil {
		return fmt.Errorf("resend: %w", err)
	}
	if err := assert(duplicate[0].Status == "duplicate", "a resent message was deduplicated, not stored twice"); err != nil {
		return err
	}

	// --------------------------------------------------------------- delivering
	//
	// The delivery worker is what marks a message as delivered to the
	// recipients. It is fast and must never fall behind, which is why it is its
	// own consumer group: it shares no cursor with the slow work below.
	//
	// Concurrency(3) runs three poll loops, and each pop claims a partition, so
	// the three conversations are drained in parallel by three goroutines. The
	// handler therefore runs concurrently and everything it touches is behind a
	// mutex; the JavaScript version needs no lock because it has no threads.
	fmt.Println("\ndelivering")
	var mu sync.Mutex
	delivered := map[string][]int{}

	err = client.Queue(messagesQueue).
		Group("delivery").
		SubscriptionMode(queen.SubscriptionModeAll).
		Concurrency(3).
		Each().
		// Limit is per worker, not a budget shared by the pool, so it is a
		// ceiling on a runaway goroutine rather than the way the
		// pool ends: what ends it is the idle bound below. Four seconds of
		// silence is hundreds of times the 10 ms a message costs here, and a
		// message that never arrives fails the count check instead of hanging
		// the run. TimeoutMillis caps each poll at a second so that deadline is
		// noticed promptly rather than inside a 30 s long poll.
		Limit(sent).
		IdleMillis(4000).
		TimeoutMillis(1000).
		Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
			time.Sleep(10 * time.Millisecond)
			conversationID, _ := msg.Data["conversationId"].(string)
			// JSON numbers decode to float64 whatever they looked like when
			// they were pushed.
			seq, ok := msg.Data["seq"].(float64)
			if !ok {
				return fmt.Errorf("message %s has no numeric seq", msg.TransactionID)
			}

			mu.Lock()
			defer mu.Unlock()
			delivered[conversationID] = append(delivered[conversationID], int(seq))
			return nil
		}).
		Execute(ctx)
	if err != nil {
		return fmt.Errorf("delivery: %w", err)
	}

	total := 0
	for _, seqs := range delivered {
		total += len(seqs)
	}
	if err := assert(total == sent, "delivery saw every message exactly once"); err != nil {
		return err
	}
	for _, c := range conversations {
		if err := assert(
			slices.IsSorted(delivered[c.id]),
			fmt.Sprintf("%s was delivered in order", c.id),
		); err != nil {
			return err
		}
	}

	// -------------------------------------------------------------- enrichment
	//
	// The slow group. It reads the same messages through its own cursor, and
	// the Japanese conversation costs 400 ms a message because it has to be
	// translated before it can be answered.
	//
	// This is where a shared partition would hurt: on a hashed topic these
	// messages would sit in the same lane as the English ones and hold them up.
	// Here each conversation has its own lane, so the English conversations
	// finish while the Japanese one is still being translated. The timings
	// below are the proof.
	fmt.Println("\nenriching")
	finishedAt := map[string]time.Duration{}
	started := time.Now()

	err = client.Queue(messagesQueue).
		Group("enrichment").
		SubscriptionMode(queen.SubscriptionModeAll).
		Concurrency(3).
		Each().
		Limit(sent).
		// Longer than the delivery bound because the Japanese lane occupies one
		// worker for about 2.4 s, and a worker that has drained its own lane
		// must not leave before the run is over.
		IdleMillis(6000).
		TimeoutMillis(1000).
		Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
			conversationID, _ := msg.Data["conversationId"].(string)
			meta, ok := conversationByID(conversationID)
			if !ok {
				return fmt.Errorf("unknown conversation %q", conversationID)
			}
			if meta.needsTranslation {
				time.Sleep(400 * time.Millisecond)
			} else {
				time.Sleep(10 * time.Millisecond)
			}

			// One lane is handled by one worker at a time, so the last write
			// for a conversation is when that conversation finished.
			mu.Lock()
			defer mu.Unlock()
			finishedAt[conversationID] = time.Since(started)
			return nil
		}).
		Execute(ctx)
	if err != nil {
		return fmt.Errorf("enrichment: %w", err)
	}

	slow := finishedAt["conv-jp-1"]
	fast := max(finishedAt["conv-en-1"], finishedAt["conv-en-2"])
	fmt.Printf("  english done after %d ms, japanese after %d ms\n",
		fast.Milliseconds(), slow.Milliseconds())

	if err := assert(
		fast < slow,
		"the conversations needing no translation finished first, in the same worker pool",
	); err != nil {
		return err
	}
	if err := assert(
		slow > messagesPerConversation*300*time.Millisecond,
		"the slow conversation really was slow, so the comparison means something",
	); err != nil {
		return err
	}

	// ------------------------------------------------------------------- replay
	//
	// A new feature needs the history: sentiment scoring over everything ever
	// said. It is a new consumer group reading from the beginning, and it costs
	// no producer change and no second copy of the data. SubscriptionMode("all")
	// is what points the new cursor at the start: a group created today would
	// otherwise begin at the tail and score nothing.
	fmt.Println("\nbackfilling a new consumer")
	scored := 0

	err = client.Queue(messagesQueue).
		Group("sentiment").
		SubscriptionMode(queen.SubscriptionModeAll).
		Concurrency(3).
		Each().
		Limit(sent).
		IdleMillis(4000).
		TimeoutMillis(1000).
		Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
			mu.Lock()
			defer mu.Unlock()
			scored++
			return nil
		}).
		Execute(ctx)
	if err != nil {
		return fmt.Errorf("sentiment: %w", err)
	}

	if err := assert(scored == sent, "a group added today read the whole history"); err != nil {
		return err
	}

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

	return nil
}

//
// A chat messaging system.
//
// This is the application Queen was written for. A hotel messaging product ran
// on Kafka and kept stalling: some conversations need a translation or an agent
// reply before the next message can be handled, and on a shared partition one
// slow conversation holds up every conversation behind it.
//
// The fix is structural rather than operational: one ordered lane per
// conversation, created by the first message sent to it. A conversation that
// takes ten seconds delays itself and nothing else.
//
// What this program builds:
//
//   chat-messages (one partition per conversation)
//     ├── group "delivery"    fast, marks each message as delivered
//     └── group "enrichment"  slow on conversations that need translation
//
// And what it proves: every message reaches both groups exactly once, in the
// order it was sent inside its own conversation, and the conversations that
// need no translation finish while the slow one is still working.
//
// Run it:
//   QUEEN_URL=http://localhost:6632 cargo run --bin chat

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

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

// Three conversations. The one in Japanese needs a translation pass, which is
// the slow work: 400 ms a message against 10 ms for the rest.
//
// (conversationId, locale, needsTranslation)
const CONVERSATIONS: [(&str, &str, bool); 3] = [
    ("conv-en-1", "en", false),
    ("conv-en-2", "en", false),
    ("conv-jp-1", "jp", true),
];
const MESSAGES_PER_CONVERSATION: i64 = 6;

fn needs_translation(conversation_id: &str) -> bool {
    CONVERSATIONS
        .iter()
        .find(|(id, _, _)| *id == conversation_id)
        .map(|(_, _, slow)| *slow)
        .unwrap_or(false)
}

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 messages = format!("app-rust-chat-{run_id}");

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

    // Signal handlers are opt-in in this client — they sit behind the `signals`
    // feature — so nothing process-wide is installed and this program owns its
    // own shutdown, through close() at the bottom.
    let queen = Queen::connect(Config::new(&url)).map_err(|e| e.to_string())?;

    // Leases are what make a crashed worker safe: a message whose handler dies
    // is redelivered once the lease expires. retry_limit bounds how many times
    // that can happen before the message is dead-lettered instead. configure()
    // is a full replace, so every key left out goes back to the broker's own
    // default rather than keeping a previous value.
    queen
        .queue(&messages)
        .configure(QueueOptions {
            lease_time: Some(60),
            retry_limit: Some(3),
            ..Default::default()
        })
        .await
        .map_err(|e| e.to_string())?;

    // ---------------------------------------------------------------- producing
    //
    // A chat client sends a message: one push, into the partition named after
    // the conversation. Nothing was declared for this conversation in advance,
    // and nothing has to be cleaned up when it goes quiet.
    println!("\nsending");
    let mut sent = 0usize;
    for seq in 1..=MESSAGES_PER_CONVERSATION {
        for (conversation_id, locale, _) in CONVERSATIONS {
            let sent_at = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_millis() as i64;
            let payload = json!({
                "conversationId": conversation_id,
                "seq": seq,
                "locale": locale,
                "body": format!("message {seq} in {conversation_id}"),
                "sentAt": sent_at,
            });

            // push() mints a UUIDv7 transaction id for you. Here the id has to
            // be the client's own idempotency key — a retry of this send, from
            // a phone on a flaky network, must write nothing the second time —
            // so the item is built by hand and pushed through push_items().
            queen
                .queue(&messages)
                .partition(conversation_id)
                .push_items(vec![PushItem::new(&messages, payload)
                    .partition(conversation_id)
                    .transaction_id(format!("{conversation_id}-{seq}"))])
                .await
                .map_err(|e| e.to_string())?;
            sent += 1;
        }
    }
    println!(
        "  {sent} messages across {} conversations",
        CONVERSATIONS.len()
    );

    // A resend of the same message: the client retried because it never saw the
    // first answer. The broker recognises the transaction id and stores nothing.
    let resent = queen
        .queue(&messages)
        .partition("conv-en-1")
        .push_items(vec![PushItem::new(
            &messages,
            json!({ "conversationId": "conv-en-1", "seq": 1, "body": "resent by the phone" }),
        )
        .partition("conv-en-1")
        .transaction_id("conv-en-1-1")])
        .await
        .map_err(|e| e.to_string())?;
    let duplicate = resent
        .first()
        .ok_or("the broker answered the resend with no result")?;
    checks.assert(
        duplicate.status == PushStatus::Duplicate,
        "a resent message was deduplicated, not stored twice",
    )?;

    // --------------------------------------------------------------- delivering
    //
    // The delivery worker is what marks a message as delivered to the
    // recipients. It is fast and must never fall behind, which is why it is its
    // own consumer group: it shares no cursor with the slow work below.
    //
    // concurrency(3) runs three poll loops, and each pop claims a partition, so
    // the three conversations are drained in parallel by three workers. The
    // handler is a plain async closure; returning Ok acks the message, returning
    // Err nacks it. `limit` counts across all three workers, not per worker.
    //
    // idle() stops the loop after a stretch of silence, so a lost message fails
    // the run instead of hanging it. It is checked between polls, so
    // poll_timeout bounds how promptly it fires: with the default 30-second poll
    // window a 10-second silence would be noticed 30 seconds late.
    println!("\ndelivering");
    let delivered: Arc<Mutex<HashMap<String, Vec<i64>>>> = Arc::new(Mutex::new(HashMap::new()));
    {
        let sink = Arc::clone(&delivered);
        queen
            .queue(&messages)
            .group("delivery")
            .subscription_mode(SubscriptionMode::All)
            .concurrency(3)
            .limit(sent as u64)
            .poll_timeout(Duration::from_secs(1))
            .idle(Duration::from_secs(10))
            .consume(move |msg: Message| {
                let sink = Arc::clone(&sink);
                async move {
                    tokio::time::sleep(Duration::from_millis(10)).await;
                    let conversation_id = msg.data["conversationId"]
                        .as_str()
                        .unwrap_or_default()
                        .to_string();
                    let seq = msg.data["seq"].as_i64().unwrap_or(0);
                    sink.lock()
                        .unwrap()
                        .entry(conversation_id)
                        .or_default()
                        .push(seq);
                    Ok::<_, String>(())
                }
            })
            .await
            .map_err(|e| e.to_string())?;
    }

    let delivered = delivered.lock().unwrap().clone();
    checks.assert(
        delivered.values().map(|seqs| seqs.len()).sum::<usize>() == sent,
        "delivery saw every message exactly once",
    )?;

    // A HashMap has no iteration order, so the lanes are checked by name: the
    // output of a passing run should not depend on how the entries happened to
    // land in the table.
    let mut lanes: Vec<&String> = delivered.keys().collect();
    lanes.sort_unstable();
    for conversation_id in lanes {
        let seqs = &delivered[conversation_id];
        let mut in_order = seqs.clone();
        in_order.sort_unstable();
        checks.assert(
            *seqs == in_order,
            &format!("{conversation_id} was delivered in order"),
        )?;
    }

    // -------------------------------------------------------------- enrichment
    //
    // The slow group. It reads the same messages through its own cursor, and the
    // Japanese conversation costs 400 ms a message because it has to be
    // translated before it can be answered.
    //
    // This is where a shared partition would hurt: on a hashed topic these
    // messages would sit in the same lane as the English ones and hold them up.
    // Here each conversation has its own lane, so the English conversations
    // finish while the Japanese one is still being translated. The timings below
    // are the proof.
    println!("\nenriching");
    let finished_at: Arc<Mutex<HashMap<String, u128>>> = Arc::new(Mutex::new(HashMap::new()));
    let started = Instant::now();
    {
        let sink = Arc::clone(&finished_at);
        queen
            .queue(&messages)
            .group("enrichment")
            .subscription_mode(SubscriptionMode::All)
            .concurrency(3)
            .limit(sent as u64)
            .poll_timeout(Duration::from_secs(1))
            .idle(Duration::from_secs(15))
            .consume(move |msg: Message| {
                let sink = Arc::clone(&sink);
                async move {
                    let conversation_id = msg.data["conversationId"]
                        .as_str()
                        .unwrap_or_default()
                        .to_string();
                    let cost = if needs_translation(&conversation_id) {
                        400
                    } else {
                        10
                    };
                    tokio::time::sleep(Duration::from_millis(cost)).await;
                    sink.lock()
                        .unwrap()
                        .insert(conversation_id, started.elapsed().as_millis());
                    Ok::<_, String>(())
                }
            })
            .await
            .map_err(|e| e.to_string())?;
    }

    let finished_at = finished_at.lock().unwrap().clone();
    let at = |conversation_id: &str| -> Result<u128, String> {
        finished_at
            .get(conversation_id)
            .copied()
            .ok_or_else(|| format!("enrichment never finished {conversation_id}"))
    };
    let slow = at("conv-jp-1")?;
    let fast = at("conv-en-1")?.max(at("conv-en-2")?);
    println!("  english done after {fast} ms, japanese after {slow} ms");

    checks.assert(
        fast < slow,
        "the conversations needing no translation finished first, in the same worker pool",
    )?;
    checks.assert(
        slow > (MESSAGES_PER_CONVERSATION as u128) * 300,
        "the slow conversation really was slow, so the comparison means something",
    )?;

    // ------------------------------------------------------------------- replay
    //
    // A new feature needs the history: sentiment scoring over everything ever
    // said. It is a new consumer group reading from the beginning, and it costs
    // no producer change and no second copy of the data. SubscriptionMode::All
    // is what points the new cursor at the beginning — the default for a new
    // group is the tail, so without it this group would sit idle.
    println!("\nbackfilling a new consumer");
    let scored = Arc::new(Mutex::new(0usize));
    {
        let counter = Arc::clone(&scored);
        queen
            .queue(&messages)
            .group("sentiment")
            .subscription_mode(SubscriptionMode::All)
            .concurrency(3)
            .limit(sent as u64)
            .poll_timeout(Duration::from_secs(1))
            .idle(Duration::from_secs(10))
            .consume(move |_msg: Message| {
                let counter = Arc::clone(&counter);
                async move {
                    *counter.lock().unwrap() += 1;
                    Ok::<_, String>(())
                }
            })
            .await
            .map_err(|e| e.to_string())?;
    }

    let scored = *scored.lock().unwrap();
    checks.assert(scored == sent, "a group added today read the whole history")?;

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

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

    Ok(checks.0)
}

//
// A chat messaging system.
//
// This is the application Queen was written for. A hotel messaging product ran
// on Kafka and kept stalling: some conversations need a translation or an agent
// reply before the next message can be handled, and on a shared partition one
// slow conversation holds up every conversation behind it.
//
// The fix is structural rather than operational: one ordered lane per
// conversation, created by the first message sent to it. A conversation that
// takes ten seconds delays itself and nothing else.
//
// What this program builds:
//
//   chat-messages (one partition per conversation)
//     ├── group "delivery"    fast, marks each message as delivered
//     └── group "enrichment"  slow on conversations that need translation
//
// And what it proves: every message reaches both groups exactly once, in the
// order it was sent inside its own conversation, and the conversations that
// need no translation finish while the slow one is still working.
//
// Run it:
//   QUEEN_URL=http://localhost:6632 php chat.php
//
// Needs pcntl, which php-cli ships with on Linux and macOS: the enrichment
// pool below is three processes, because that is what a PHP worker pool is.

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);
$MESSAGES = "app-php-chat-{$RUN}";

// Three conversations. The one in Japanese needs a translation pass, which is
// the slow work: 400 ms a message against 10 ms for the rest.
$CONVERSATIONS = [
    'conv-en-1' => ['locale' => 'en', 'needsTranslation' => false],
    'conv-en-2' => ['locale' => 'en', 'needsTranslation' => false],
    'conv-jp-1' => ['locale' => 'jp', 'needsTranslation' => true],
];
$MESSAGES_PER_CONVERSATION = 6;
$TOTAL = $MESSAGES_PER_CONVERSATION * count($CONVERSATIONS);
$WORKERS = 3;

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

// PHP sleeps in microseconds, and a handler that sleeps blocks its whole
// process: that is the one fact the enrichment section below is built around.
$sleepMillis = fn(int $millis) => usleep($millis * 1000);

// There is no signal-handling option to turn off on this client: it installs
// SIGINT and SIGTERM handlers only for the duration of a consume loop and
// restores the previous ones when the loop returns, so this script keeps
// control of its own shutdown.
$queen = new Queen($QUEEN_URL);
$exitCode = 0;

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

    // Checked before anything is created, so a build without pcntl says so
    // instead of leaving a queue behind halfway through.
    if (!function_exists('pcntl_fork')) {
        throw new RuntimeException('this example forks its worker pool and needs the pcntl extension');
    }

    // Leases are what make a crashed worker safe: a message whose handler dies is
    // redelivered once the lease expires. retryLimit bounds how many times that
    // can happen before the message is dead-lettered instead.
    $queen->queue($MESSAGES)->config(['leaseTime' => 60, 'retryLimit' => 3])->create()->execute();

    // ---------------------------------------------------------------- producing
    //
    // A chat client sends a message: one push, into the partition named after the
    // conversation. Nothing was declared for this conversation in advance, and
    // nothing has to be cleaned up when it goes quiet.
    echo "\nsending\n";
    $sent = 0;
    for ($seq = 1; $seq <= $MESSAGES_PER_CONVERSATION; $seq++) {
        foreach ($CONVERSATIONS as $conversationId => $meta) {
            $message = [
                'conversationId' => $conversationId,
                'seq' => $seq,
                'locale' => $meta['locale'],
                'body' => "message {$seq} in {$conversationId}",
                'sentAt' => (int) (microtime(true) * 1000),
            ];
            // The transaction id is the client's own idempotency key: a retry of
            // this send, from a phone on a flaky network, writes nothing the second
            // time and answers with the first message's id.
            $queen->queue($MESSAGES)->partition($conversationId)->push([[
                'transactionId' => "{$conversationId}-{$seq}",
                'data' => $message,
            ]])->execute();
            $sent++;
        }
    }
    echo "  {$sent} messages across " . count($CONVERSATIONS) . " conversations\n";

    // A resend of the same message: the client retried because it never saw the
    // first answer. The broker recognises the transaction id and stores nothing.
    // execute() hands back one row per item pushed, and the row is where the
    // verdict is: this client does not throw on a duplicate.
    $resend = $queen->queue($MESSAGES)->partition('conv-en-1')->push([[
        'transactionId' => 'conv-en-1-1',
        'data' => ['conversationId' => 'conv-en-1', 'seq' => 1, 'body' => 'resent by the phone'],
    ]])->execute();
    $assert($resend[0]['status'] === 'duplicate', 'a resent message was deduplicated, not stored twice');

    // --------------------------------------------------------------- delivering
    //
    // The delivery worker is what marks a message as delivered to the recipients.
    // It is fast and must never fall behind, which is why it is its own consumer
    // group: it shares no cursor with the slow work below.
    //
    // concurrency(3) on this client is three long polls in flight at once on one
    // cURL multi-handle, not three threads: the polls overlap, the handlers still
    // run one after another. Each poll claims a partition of its own, so the
    // three conversations are drained side by side. That is all this section
    // needs: exactly-once and in-order are properties of the claim, not of how
    // many handlers run at the same instant. The timing proof further down needs
    // more than overlapping polls, and gets it.
    //
    // timeoutMillis(1000) caps how long one poll parks on the broker. The default
    // is 30 s, and a round here ends only when every worker's poll has come back,
    // so the last round of a drained queue would otherwise sit there for half a
    // minute before the idle bound could fire.
    echo "\ndelivering\n";
    $delivered = [];
    $queen
        ->queue($MESSAGES)
        ->group('delivery')
        ->subscriptionMode('all')
        ->concurrency(3)
        ->each()
        ->limit($TOTAL)
        ->idleMillis(10000)
        ->timeoutMillis(1000)
        ->consume(function (array $msg) use (&$delivered, $sleepMillis): void {
            $sleepMillis(10);
            $delivered[$msg['data']['conversationId']][] = $msg['data']['seq'];
        })
        ->execute();

    $assert(
        array_sum(array_map('count', $delivered)) === $sent,
        'delivery saw every message exactly once'
    );
    foreach ($delivered as $conversationId => $seqs) {
        $sorted = $seqs;
        sort($sorted);
        $assert($seqs === $sorted, "{$conversationId} was delivered in order");
    }

    // -------------------------------------------------------------- enrichment
    //
    // The slow group. It reads the same messages through its own cursor, and the
    // Japanese conversation costs 400 ms a message because it has to be
    // translated before it can be answered.
    //
    // This is where a shared partition would hurt: on a hashed topic these
    // messages would sit in the same lane as the English ones and hold them up.
    // Here each conversation has its own lane, so the English conversations
    // finish while the Japanese one is still being translated. The timings below
    // are the proof.
    //
    // A worker pool in PHP is processes. There is no event loop to interleave a
    // sleeping handler with a running one, so three overlapping polls in one
    // process would still translate and deliver strictly one after another, and
    // the clock would say nothing about lanes. Forking is what a PHP deployment
    // actually does, a queue:work fleet of three, and it is what makes the
    // measurement mean something.
    echo "\nenriching\n";
    $startedAt = microtime(true);
    $reportPaths = [];
    $children = [];

    for ($slot = 0; $slot < $WORKERS; $slot++) {
        $reportPath = sys_get_temp_dir() . "/app-php-chat-{$RUN}-{$slot}.json";
        $reportPaths[$slot] = $reportPath;

        $pid = pcntl_fork();
        if ($pid === -1) {
            throw new RuntimeException('could not fork an enrichment worker');
        }

        if ($pid === 0) {
            // Child. It builds its own client: the parent's cURL handles and
            // sockets are shared across the fork, and two processes taking turns
            // on one connection corrupt each other's replies.
            $childCode = 0;
            try {
                $worker = new Queen($QUEEN_URL);
                $finishedAt = [];
                $enriched = 0;

                $worker
                    ->queue($MESSAGES)
                    ->group('enrichment')
                    ->subscriptionMode('all')
                    ->each()
                    // The bound is the whole run, not this worker's share: which
                    // lane a worker claims is the broker's decision, so a worker
                    // that is handed two of them must be allowed to finish both.
                    // What ends a worker is the idle bound, once the lanes it can
                    // still claim have gone quiet.
                    ->limit($TOTAL)
                    ->idleMillis(3000)
                    ->timeoutMillis(1000)
                    ->consume(function (array $msg) use (
                        $CONVERSATIONS, $startedAt, $sleepMillis, &$finishedAt, &$enriched
                    ): void {
                        $conversationId = $msg['data']['conversationId'];
                        $sleepMillis($CONVERSATIONS[$conversationId]['needsTranslation'] ? 400 : 10);
                        $finishedAt[$conversationId] = (int) round((microtime(true) - $startedAt) * 1000);
                        $enriched++;
                    })
                    ->execute();

                file_put_contents($reportPath, json_encode([
                    'finishedAt' => $finishedAt,
                    'enriched' => $enriched,
                ]));
                $worker->close();
            } catch (Throwable $error) {
                file_put_contents($reportPath, json_encode(['error' => $error->getMessage()]));
                $childCode = 1;
            }
            // exit, not return: a child that fell through would run the parent's
            // remaining checks and delete the queue underneath it.
            exit($childCode);
        }

        $children[] = $pid;
    }

    // The parent only waits. Every worker reports the elapsed milliseconds at
    // which it last touched each conversation, and the merge keeps the latest of
    // those, which is when that conversation was finished with.
    $finishedAt = [];
    $enriched = 0;
    foreach ($children as $slot => $pid) {
        pcntl_waitpid($pid, $status);
        $report = json_decode((string) @file_get_contents($reportPaths[$slot]), true) ?: [];
        @unlink($reportPaths[$slot]);
        if (isset($report['error'])) {
            throw new RuntimeException("enrichment worker {$slot} failed: {$report['error']}");
        }
        if (!pcntl_wifexited($status) || pcntl_wexitstatus($status) !== 0) {
            throw new RuntimeException("enrichment worker {$slot} did not exit cleanly");
        }
        foreach ($report['finishedAt'] ?? [] as $conversationId => $millis) {
            $finishedAt[$conversationId] = max($finishedAt[$conversationId] ?? 0, $millis);
        }
        $enriched += $report['enriched'] ?? 0;
    }

    $assert($enriched === $sent, 'the pool enriched every message exactly once, across three processes');

    $slow = $finishedAt['conv-jp-1'] ?? 0;
    $fast = max($finishedAt['conv-en-1'] ?? 0, $finishedAt['conv-en-2'] ?? 0);
    echo "  english done after {$fast} ms, japanese after {$slow} ms\n";

    $assert(
        $fast < $slow,
        'the conversations needing no translation finished first, in the same worker pool'
    );
    $assert(
        $slow > $MESSAGES_PER_CONVERSATION * 300,
        'the slow conversation really was slow, so the comparison means something'
    );

    // ------------------------------------------------------------------- replay
    //
    // A new feature needs the history: sentiment scoring over everything ever
    // said. It is a new consumer group reading from the beginning, and it costs
    // no producer change and no second copy of the data.
    echo "\nbackfilling a new consumer\n";
    $scored = 0;
    $queen
        ->queue($MESSAGES)
        ->group('sentiment')
        ->subscriptionMode('all')
        ->concurrency(3)
        ->each()
        ->limit($TOTAL)
        ->idleMillis(10000)
        ->timeoutMillis(1000)
        ->consume(function (array $msg) use (&$scored): void {
            $scored++;
        })
        ->execute();

    $assert($scored === $sent, 'a group added today read the whole history');

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

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

exit($exitCode);

//
// A chat messaging system.
//
// This is the application Queen was written for. A hotel messaging product ran
// on Kafka and kept stalling: some conversations need a translation or an agent
// reply before the next message can be handled, and on a shared partition one
// slow conversation holds up every conversation behind it.
//
// The fix is structural rather than operational: one ordered lane per
// conversation, created by the first message sent to it. A conversation that
// takes ten seconds delays itself and nothing else.
//
// What this program builds:
//
//   chat-messages (one partition per conversation)
//     |-- group "delivery"    fast, marks each message as delivered
//     `-- group "enrichment"  slow on conversations that need translation
//
// And what it proves: every message reaches both groups exactly once, in the
// order it was sent inside its own conversation, and the conversations that
// need no translation finish while the slow one is still working.
//
// 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" \
//       chat.cpp -o build/chat \
//       -L"$(brew --prefix openssl)/lib" -lssl -lcrypto -lpthread
//
// Run it:
//   QUEEN_URL=http://localhost:6632 ./build/chat

#include "queen_client.hpp"

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

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

// The queue 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.
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 Conversation {
    std::string id;
    std::string locale;
    bool needs_translation;
};

// Three conversations. The one in Japanese needs a translation pass, which is
// the slow work: 400 ms a message against 10 ms for the rest.
static const std::vector<Conversation> CONVERSATIONS = {
    {"conv-en-1", "en", false},
    {"conv-en-2", "en", false},
    {"conv-jp-1", "jp", true},
};
static const int MESSAGES_PER_CONVERSATION = 6;
static const int TOTAL_MESSAGES =
    MESSAGES_PER_CONVERSATION * static_cast<int>(CONVERSATIONS.size());

static int checks = 0;

// C++ has no assert that survives -DNDEBUG and carries a message, so this is 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 const Conversation& conversation_by_id(const std::string& id) {
    for (const Conversation& c : CONVERSATIONS) {
        if (c.id == id) return c;
    }
    throw std::runtime_error("unknown conversation " + id);
}

static void sleep_millis(int millis) {
    std::this_thread::sleep_for(std::chrono::milliseconds(millis));
}

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();
}

static long long millis_since(std::chrono::steady_clock::time_point start) {
    return std::chrono::duration_cast<std::chrono::milliseconds>(
               std::chrono::steady_clock::now() - start)
        .count();
}

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 MESSAGES = "app-cpp-chat-" + run_id();

    // Unlike the JavaScript client there is no handleSignals switch:
    // QueenClient always installs its own SIGINT/SIGTERM handlers, so a Ctrl-C
    // during a run is the client's exit, not yours.
    QueenClient client(QUEEN_URL);

    std::string verdict;
    bool failed = false;

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

        // Leases are what make a crashed worker safe: a message whose handler
        // dies is redelivered once the lease expires. retry_limit bounds how
        // many times that can happen before the message is dead-lettered
        // instead. The C++ QueueConfig is a struct rather than an option bag,
        // and every field it does not carry keeps the broker's own default.
        queen::QueueConfig config;
        config.lease_time = 60;
        config.retry_limit = 3;
        client.queue(MESSAGES).config(config).create();

        // ------------------------------------------------------------ producing
        //
        // A chat client sends a message: one push, into the partition named
        // after the conversation. Nothing was declared for this conversation in
        // advance, and nothing has to be cleaned up when it goes quiet.
        std::cout << "\nsending" << std::endl;
        int sent = 0;
        for (int seq = 1; seq <= MESSAGES_PER_CONVERSATION; ++seq) {
            for (const Conversation& conv : CONVERSATIONS) {
                // push() takes a vector of items because one call can carry a
                // batch; the broker answers with one result per item, in order.
                //
                // The transaction id is the client's own idempotency key: a
                // retry of this send, from a phone on a flaky network, writes
                // nothing the second time and answers with the first message's
                // id.
                client.queue(MESSAGES).partition(conv.id).push({
                    json{{"transactionId", conv.id + "-" + std::to_string(seq)},
                         {"data", {{"conversationId", conv.id},
                                   {"seq", seq},
                                   {"locale", conv.locale},
                                   {"body", "message " + std::to_string(seq) +
                                                " in " + conv.id}}}}
                });
                ++sent;
            }
        }
        std::cout << "  " << sent << " messages across " << CONVERSATIONS.size()
                  << " conversations" << std::endl;

        // A resend of the same message: the client retried because it never saw
        // the first answer. The broker recognises the transaction id and stores
        // nothing. push() hands back the broker's reply unwrapped, so the
        // per-item verdict is read straight off the array.
        json duplicate = client.queue(MESSAGES).partition("conv-en-1").push({
            json{{"transactionId", "conv-en-1-1"},
                 {"data", {{"conversationId", "conv-en-1"},
                           {"seq", 1},
                           {"body", "resent by the phone"}}}}
        });
        check(duplicate.is_array() && duplicate.size() == 1 &&
                  duplicate[0]["status"] == "duplicate",
              "a resent message was deduplicated, not stored twice");

        // ----------------------------------------------------------- delivering
        //
        // The delivery worker is what marks a message as delivered to the
        // recipients. It is fast and must never fall behind, which is why it is
        // its own consumer group: it shares no cursor with the slow work below.
        //
        // concurrency(3) runs three poll loops, and each pop claims a
        // partition, so the three conversations are drained in parallel by
        // three workers. Nothing raises partitions() here: one lane per worker
        // is exactly the point.
        //
        // Three settings keep that loop honest:
        //
        //   wait(false)     a long-polling pop parks server-side for up to 30
        //                   seconds and the idle clock is only consulted
        //                   between polls, so a blocking pop would stretch a 10
        //                   second idle budget into half a minute.
        //   idle_millis     the deadline: a lost message fails this run instead
        //                   of hanging it.
        //   limit()         counts per worker, here and in the JavaScript
        //                   alike: each worker keeps its own tally, so three
        //                   workers sharing 18 messages never see one worker
        //                   reach 18. It is a backstop, not the thing that ends
        //                   the run.
        //
        // What ends the run is the shared counter below, which raises the stop
        // flag the moment every message has been handled. Without it every
        // consume in this file would sit out its idle deadline after the last
        // message, which is exactly what the JavaScript pays.
        //
        // consume() is synchronous in C++: it blocks this thread and runs
        // concurrency() workers in a pool until the stop flag, the idle
        // deadline or a worker's own limit ends every one of them. It also
        // catches whatever the handler throws and turns it into a
        // negative acknowledgement, so an exception raised in there never
        // reaches main() on its own: record it, raise the stop flag, and
        // rethrow once consume() has returned.
        std::cout << "\ndelivering" << std::endl;
        std::mutex lock;
        std::map<std::string, std::vector<int>> delivered;
        std::atomic<int> handled{0};
        std::atomic<bool> stop{false};
        std::exception_ptr handler_error;

        client.queue(MESSAGES)
            .group("delivery")
            .subscription_mode("all")
            .concurrency(3)
            .each()
            .limit(TOTAL_MESSAGES)
            .wait(false)
            .idle_millis(10000)
            .consume([&](const json& msg) {
                try {
                    sleep_millis(10);
                    {
                        std::lock_guard<std::mutex> guard(lock);
                        delivered[msg["data"]["conversationId"].get<std::string>()]
                            .push_back(msg["data"]["seq"].get<int>());
                    }
                    // The acknowledgement of this message happens after the
                    // handler returns, so the stop flag raised here still lets
                    // the last message commit.
                    if (++handled >= TOTAL_MESSAGES) stop = true;
                } catch (...) {
                    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);

        int delivered_total = 0;
        for (const auto& entry : delivered) delivered_total += entry.second.size();
        check(delivered_total == sent, "delivery saw every message exactly once");

        // One lane is only ever held by one worker at a time -- the lease is
        // released by the acknowledgement -- so the sequence numbers inside a
        // conversation come back in the order they were sent, however many
        // workers are running.
        std::vector<int> in_order;
        for (int seq = 1; seq <= MESSAGES_PER_CONVERSATION; ++seq) in_order.push_back(seq);
        for (const Conversation& conv : CONVERSATIONS) {
            check(delivered[conv.id] == in_order, conv.id + " was delivered in order");
        }

        // ----------------------------------------------------------- enrichment
        //
        // The slow group. It reads the same messages through its own cursor,
        // and the Japanese conversation costs 400 ms a message because it has
        // to be translated before it can be answered.
        //
        // This is where a shared partition would hurt: on a hashed topic these
        // messages would sit in the same lane as the English ones and hold them
        // up. Here each conversation has its own lane, so the English
        // conversations finish while the Japanese one is still being
        // translated. The timings below are the proof.
        std::cout << "\nenriching" << std::endl;
        std::map<std::string, long long> finished_at;
        handled = 0;
        stop = false;
        auto started = std::chrono::steady_clock::now();

        client.queue(MESSAGES)
            .group("enrichment")
            .subscription_mode("all")
            .concurrency(3)
            .each()
            .limit(TOTAL_MESSAGES)
            .wait(false)
            .idle_millis(15000)
            .consume([&](const json& msg) {
                try {
                    const std::string id =
                        msg["data"]["conversationId"].get<std::string>();
                    sleep_millis(conversation_by_id(id).needs_translation ? 400 : 10);
                    {
                        std::lock_guard<std::mutex> guard(lock);
                        finished_at[id] = millis_since(started);
                    }
                    if (++handled >= TOTAL_MESSAGES) stop = true;
                } catch (...) {
                    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);

        const long long slow = finished_at["conv-jp-1"];
        const long long fast =
            std::max(finished_at["conv-en-1"], finished_at["conv-en-2"]);
        std::cout << "  english done after " << fast << " ms, japanese after "
                  << slow << " ms" << std::endl;

        check(fast < slow,
              "the conversations needing no translation finished first, in the "
              "same worker pool");
        check(slow > MESSAGES_PER_CONVERSATION * 300,
              "the slow conversation really was slow, so the comparison means "
              "something");

        // --------------------------------------------------------------- replay
        //
        // A new feature needs the history: sentiment scoring over everything
        // ever said. It is a new consumer group reading from the beginning, and
        // it costs no producer change and no second copy of the data.
        //
        // subscription_mode("all") is what points the new cursor at the
        // beginning: the default for a new group is the tail, so without it
        // this group would sit idle waiting for the next message and the run
        // would end on the idle deadline with nothing scored.
        std::cout << "\nbackfilling a new consumer" << std::endl;
        std::atomic<int> scored{0};
        handled = 0;
        stop = false;

        client.queue(MESSAGES)
            .group("sentiment")
            .subscription_mode("all")
            .concurrency(3)
            .each()
            .limit(TOTAL_MESSAGES)
            .wait(false)
            .idle_millis(10000)
            .consume([&](const json&) {
                ++scored;
                if (++handled >= TOTAL_MESSAGES) stop = true;
            }, &stop);

        check(scored.load() == sent, "a group added today read the whole history");

        std::cout << "\n  delivery order in conv-jp-1: "
                  << join(delivered["conv-jp-1"]) << std::endl;

        // Clean up on success only: a failed run leaves the queue on the broker
        // to be looked at. del(), not delete: the word is taken.
        client.queue(MESSAGES).del();

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

    // close() flushes anything still sitting in the client-side push buffers
    // and drops them along with their timer threads. 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.
    client.close();

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