Chat backend

//
// A chat backend: one ordered partition per conversation.
//
// Queen started as the broker of a hotel messaging product. Some conversations
// need a translation or an agent before their next message can be handled, and
// on a hashed Kafka topic one slow conversation held up every conversation that
// shared its partition. Here every conversation is a partition of its own,
// created by the first message sent to it, so a slow conversation waits on
// itself and on nothing else.
//
//   chat-messages (one partition per conversation)
//     ├── group "delivery"    marks each message delivered, fast
//     ├── group "enrichment"  translates the Japanese conversation, slow
//     └── group "sentiment"   added later, reads the whole history
//
// The program checks what the design promises: every message reaches each
// group once and in the order of its conversation, and the English
// conversations finish while the Japanese one is still being translated.
//
// 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'
// A fresh queue per run, so two runs never read each other's messages.
const RUN = Date.now().toString(36)
const MESSAGES = `app-js-chat-${RUN}`

// Three conversations. The Japanese one needs a translation pass, 400 ms a
// message against 10 ms for the others. It is listed first, so its messages are
// the oldest in the queue and its partition is usually handed out first: a
// consumer that let one conversation hold up another would fail the timing
// check below.
const CONVERSATIONS = {
  'conv-jp-1': { locale: 'jp', needsTranslation: true },
  'conv-en-1': { locale: 'en', needsTranslation: false },
  'conv-en-2': { locale: 'en', needsTranslation: false },
}
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 })

// Three workers in one consumer group. partitions(1) makes every pop take ONE
// conversation: by default a pop may sweep up several ready conversations, and a
// worker handles the messages of one pop in order, so a slow conversation would
// delay the others that came with it. Long polls end after a second and a
// worker stops after two quiet seconds, which is what lets this program finish;
// a service calls consume() without those two lines and runs until stopped.
const workers = (group) => queen
  .queue(MESSAGES)
  .group(group)
  .subscriptionMode('all') // a group created after the messages starts at the tail otherwise
  .concurrency(3)
  .partitions(1)
  .each()
  .timeoutMillis(1000)
  .idleMillis(2000)

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

  // A crashed worker's messages come back when its lease expires, and
  // retryLimit bounds how often a failing message is retried before it goes to
  // the dead-letter queue.
  await queen.queue(MESSAGES).config({ leaseTime: 60, retryLimit: 3 }).create()

  // ---------------------------------------------------------------- sending
  //
  // Sending a message is one push into the conversation's partition. Nothing
  // was declared for the conversation beforehand, and nothing has to be cleaned
  // up when it goes quiet.
  console.log('\nsending')
  let sent = 0
  for (let seq = 1; seq <= MESSAGES_PER_CONVERSATION; seq++) {
    for (const [conversationId, meta] of Object.entries(CONVERSATIONS)) {
      await queen.queue(MESSAGES).partition(conversationId).push({
        // The phone's own id for the message. A phone that retries a send it
        // never saw answered writes nothing the second time.
        transactionId: `${conversationId}-${seq}`,
        data: { conversationId, seq, locale: meta.locale, body: `message ${seq} in ${conversationId}` },
      })
      sent++
    }
  }
  console.log(`  ${sent} messages across ${Object.keys(CONVERSATIONS).length} conversations`)

  // The phone resends message 1 because the answer got lost on a bad network.
  const [resent] = 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(resent.status === 'duplicate', 'a resent message was recognised and not stored twice')

  // ------------------------------------------------------------- delivering
  //
  // Marking messages delivered is fast work and must never wait behind slow
  // work, so it is a consumer group of its own, with its own cursor.
  console.log('\ndelivering')
  const delivered = new Map()
  await workers('delivery').consume(async (msg) => {
    await sleep(10)
    const seqs = delivered.get(msg.data.conversationId) ?? []
    seqs.push(msg.data.seq)
    delivered.set(msg.data.conversationId, seqs)
  })

  const deliveredCount = [...delivered.values()].reduce((n, seqs) => n + seqs.length, 0)
  assert(deliveredCount === sent, `delivery saw all ${sent} messages once (got ${deliveredCount})`)
  for (const [conversationId, seqs] of delivered) {
    assert(
      seqs.every((seq, i) => seq === i + 1),
      `${conversationId} was delivered in order: ${seqs.join(',')}`
    )
  }

  // ------------------------------------------------------------- enrichment
  //
  // The slow group reads the same messages through its own cursor. On a topic
  // with a few hashed partitions, the Japanese conversation would sit in a
  // partition shared with English ones and hold them up. Here each worker holds
  // one conversation at a time, so the English conversations finish while the
  // Japanese one is still being translated.
  console.log('\nenriching')
  const finishedAt = new Map()
  const started = Date.now()
  await workers('enrichment').consume(async (msg) => {
    const { needsTranslation } = CONVERSATIONS[msg.data.conversationId]
    await sleep(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 English conversations finished while the Japanese one was still being translated')
  assert(slow >= MESSAGES_PER_CONVERSATION * 400, 'the Japanese conversation really took its six translations')

  // --------------------------------------------------------------- backfill
  //
  // A feature added later, sentiment scoring, wants every message ever sent. It
  // is one more consumer group starting from the oldest message: no producer
  // change, and no second copy of the data.
  console.log('\nbackfilling a new group')
  let scored = 0
  await workers('sentiment').consume(async () => { scored++ })
  assert(scored === sent, `a group created now read the whole history (${scored} messages)`)

  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 backend: one ordered partition per conversation.
#
# Queen started as the broker of a hotel messaging product. Some conversations
# need a translation or an agent before their next message can be handled, and
# on a hashed Kafka topic one slow conversation held up every conversation that
# shared its partition. Here every conversation is a partition of its own,
# created by the first message sent to it, so a slow conversation waits on
# itself and on nothing else.
#
#   chat-messages (one partition per conversation)
#     |-- group "delivery"    marks each message delivered, fast
#     |-- group "enrichment"  translates the Japanese conversation, slow
#     `-- group "sentiment"   added later, reads the whole history
#
# The program checks what the design promises: every message reaches each
# group once and in the order of its conversation, and the English
# conversations finish while the Japanese one is still being translated.
#
# 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 two runs never read each other's
# messages.
RUN = f"{int(time.time() * 1000):x}"
MESSAGES = f"app-py-chat-{RUN}"

# Three conversations. The Japanese one needs a translation pass, 400 ms a
# message against 10 ms for the others. It is listed first, so its messages are
# the oldest in the queue and its partition is usually handed out first: a
# consumer that let one conversation hold up another would fail the timing
# check below.
CONVERSATIONS = {
    "conv-jp-1": {"locale": "jp", "needs_translation": True},
    "conv-en-1": {"locale": "en", "needs_translation": False},
    "conv-en-2": {"locale": "en", "needs_translation": False},
}
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

    # Three workers in one consumer group: concurrency(3) runs three poll loops
    # on this event loop. partitions(1) makes every pop take ONE conversation:
    # by default a pop may sweep up several ready conversations, and a worker
    # handles the messages of one pop in order, so a slow conversation would
    # delay the others that came with it. Long polls end after a second and a
    # worker stops after two quiet seconds, which is what lets this program
    # finish; a service calls consume() without those two lines and runs until
    # stopped.
    def workers(group: str):
        return (
            queen.queue(MESSAGES)
            .group(group)
            # A group created after the messages were pushed starts at the tail,
            # so without this it would see nothing.
            .subscription_mode("all")
            .concurrency(3)
            .partitions(1)
            .each()
            .timeout_millis(1000)
            .idle_millis(2000)
        )

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

        # A crashed worker's messages come back when its lease expires, and
        # retry_limit bounds how often a failing message is retried before it
        # goes to the dead-letter queue. 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()

        # ------------------------------------------------------------ sending
        #
        # Sending a message is one push into the conversation's partition.
        # Nothing was declared for the conversation beforehand, and nothing has
        # to be cleaned up when it goes quiet.
        print("\nsending")
        sent = 0
        for seq in range(1, MESSAGES_PER_CONVERSATION + 1):
            for conversation_id, meta in CONVERSATIONS.items():
                await queen.queue(MESSAGES).partition(conversation_id).push(
                    {
                        # The phone's own id for the message. A phone that
                        # retries a send it never saw answered writes nothing
                        # the second time. The key is camelCase because it is
                        # the broker's wire name.
                        "transactionId": f"{conversation_id}-{seq}",
                        "data": {
                            "conversationId": conversation_id,
                            "seq": seq,
                            "locale": meta["locale"],
                            "body": f"message {seq} in {conversation_id}",
                        },
                    }
                )
                sent += 1
        print(f"  {sent} messages across {len(CONVERSATIONS)} conversations")

        # The phone resends message 1 because the answer got lost on a bad
        # network. 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 recognised and not stored twice",
        )

        # --------------------------------------------------------- delivering
        #
        # Marking messages delivered is fast work and must never wait behind
        # slow work, so it is a consumer group of its own, with its own cursor.
        # The handler is an async def taking one message: consume() awaits it
        # for every message and acknowledges the message when it returns.
        print("\ndelivering")
        delivered: dict = {}

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

        await workers("delivery").consume(deliver)

        delivered_count = sum(len(seqs) for seqs in delivered.values())
        check(
            delivered_count == sent,
            f"delivery saw all {sent} messages once (got {delivered_count})",
        )
        for conversation_id, seqs in delivered.items():
            check(
                seqs == list(range(1, len(seqs) + 1)),
                f"{conversation_id} was delivered in order: {','.join(map(str, seqs))}",
            )

        # --------------------------------------------------------- enrichment
        #
        # The slow group reads the same messages through its own cursor. On a
        # topic with a few hashed partitions, the Japanese conversation would
        # sit in a partition shared with English ones and hold them up. Here
        # each worker holds one conversation at a time, so the English
        # conversations finish while the Japanese one is still being
        # translated.
        print("\nenriching")
        finished_at: dict = {}
        # time.monotonic(), because these are durations: a wall clock that
        # steps mid-run could 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 workers("enrichment").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 English conversations finished while the Japanese one was still being translated",
        )
        check(
            slow >= MESSAGES_PER_CONVERSATION * 400,
            "the Japanese conversation really took its six translations",
        )

        # ----------------------------------------------------------- backfill
        #
        # A feature added later, sentiment scoring, wants every message ever
        # sent. It is one more consumer group starting from the oldest message:
        # no producer change, and no second copy of the data.
        print("\nbackfilling a new group")
        scored = 0

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

        await workers("sentiment").consume(score)
        check(scored == sent, f"a group created now read the whole history ({scored} messages)")

        # 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: 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 backend: one ordered partition per conversation.
//
// Queen started as the broker of a hotel messaging product. Some conversations
// need a translation or an agent before their next message can be handled, and
// on a hashed Kafka topic one slow conversation held up every conversation that
// shared its partition. Here every conversation is a partition of its own,
// created by the first message sent to it, so a slow conversation waits on
// itself and on nothing else.
//
//	chat-messages (one partition per conversation)
//	  |-- group "delivery"    marks each message delivered, fast
//	  |-- group "enrichment"  translates the Japanese conversation, slow
//	  `-- group "sentiment"   added later, reads the whole history
//
// The program checks what the design promises: every message reaches each
// group once and in the order of its conversation, and the English
// conversations finish while the Japanese one is still being translated.
//
// Run it:
//
//	QUEEN_URL=http://localhost:6632 GOWORK=off go run ./chat
package main

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

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

// A fresh queue per run, so two runs never read each other's messages.
var runID = strconv.FormatInt(time.Now().UnixMilli(), 36)

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

// Three conversations. The Japanese one needs a translation pass, 400 ms a
// message against 10 ms for the others. It is listed first, so its messages are
// the oldest in the queue and its partition is usually handed out first: a
// consumer that let one conversation hold up another would fail the timing
// check below. The list is a slice because Go randomises map iteration, and
// the send order has to be the same on every run.
type conversation struct {
	id               string
	locale           string
	needsTranslation bool
}

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

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 fails the run when it expires.
	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())

	// Three workers in one consumer group. Partitions(1) makes every pop take
	// ONE conversation: by default a pop may sweep up several ready
	// conversations, and a worker handles the messages of one pop in order, so
	// a slow conversation would delay the others that came with it. Long polls
	// end after a second and a worker stops after two quiet seconds, which is
	// what lets this program finish; a service consumes without those two
	// settings and runs until its context is cancelled. (Limit cannot end a
	// phase here: this client counts it per worker.)
	//
	// Each worker is a goroutine, so the handlers below run concurrently and
	// everything they touch is behind a mutex.
	workers := func(group string) *queen.QueueBuilder {
		return client.Queue(messagesQueue).
			Group(group).
			// A group created after the messages starts at the tail otherwise.
			SubscriptionMode(queen.SubscriptionModeAll).
			Concurrency(3).
			Partitions(1).
			Each().
			TimeoutMillis(1000).
			IdleMillis(2000)
	}

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

	// A crashed worker's messages come back when its lease expires, and
	// RetryLimit bounds how often a failing message is retried before it goes
	// to the dead-letter queue.
	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)
	}

	// ------------------------------------------------------------------ sending
	//
	// Sending a message is one push into the conversation's partition. Nothing
	// was declared for the conversation beforehand, 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 {
			// The transaction id is the phone's own id for the message. A
			// phone that retries a send it never saw answered writes nothing
			// the second time. This client takes it on the push builder.
			if _, err := client.Queue(messagesQueue).
				Partition(c.id).
				Push(map[string]interface{}{
					"conversationId": c.id,
					"seq":            seq,
					"locale":         c.locale,
					"body":           fmt.Sprintf("message %d in %s", seq, c.id),
				}).
				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))

	// The phone resends message 1 because the answer got lost on a bad network.
	resent, 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(resent[0].Status == "duplicate", "a resent message was recognised and not stored twice"); err != nil {
		return err
	}

	// --------------------------------------------------------------- delivering
	//
	// Marking messages delivered is fast work and must never wait behind slow
	// work, so it is a consumer group of its own, with its own cursor.
	fmt.Println("\ndelivering")
	var mu sync.Mutex
	delivered := map[string][]int{}

	err = workers("delivery").
		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)
	}

	deliveredCount := 0
	for _, seqs := range delivered {
		deliveredCount += len(seqs)
	}
	if err := assert(
		deliveredCount == sent,
		fmt.Sprintf("delivery saw all %d messages once (got %d)", sent, deliveredCount),
	); err != nil {
		return err
	}
	for _, c := range conversations {
		seqs := delivered[c.id]
		if err := assert(
			inSequence(seqs),
			fmt.Sprintf("%s was delivered in order: %s", c.id, joinInts(seqs)),
		); err != nil {
			return err
		}
	}

	// --------------------------------------------------------------- enrichment
	//
	// The slow group reads the same messages through its own cursor. On a topic
	// with a few hashed partitions, the Japanese conversation would sit in a
	// partition shared with English ones and hold them up. Here each worker
	// holds one conversation at a time, so the English conversations finish
	// while the Japanese one is still being translated.
	fmt.Println("\nenriching")
	finishedAt := map[string]time.Duration{}
	started := time.Now()

	err = workers("enrichment").
		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 conversation 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 English conversations finished while the Japanese one was still being translated",
	); err != nil {
		return err
	}
	if err := assert(
		slow >= messagesPerConversation*400*time.Millisecond,
		"the Japanese conversation really took its six translations",
	); err != nil {
		return err
	}

	// ----------------------------------------------------------------- backfill
	//
	// A feature added later, sentiment scoring, wants every message ever sent.
	// It is one more consumer group starting from the oldest message: no
	// producer change, and no second copy of the data.
	fmt.Println("\nbackfilling a new group")
	scored := 0

	err = workers("sentiment").
		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,
		fmt.Sprintf("a group created now read the whole history (%d messages)", scored),
	); 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
}

// inSequence reports whether seqs runs 1, 2, 3 and so on, without a gap, a
// repeat or a swap.
func inSequence(seqs []int) bool {
	for i, seq := range seqs {
		if seq != i+1 {
			return false
		}
	}
	return true
}

func joinInts(values []int) string {
	parts := make([]string, len(values))
	for i, v := range values {
		parts[i] = strconv.Itoa(v)
	}
	return strings.Join(parts, ",")
}

//
// A chat backend: one ordered partition per conversation.
//
// Queen started as the broker of a hotel messaging product. Some conversations
// need a translation or an agent before their next message can be handled, and
// on a hashed Kafka topic one slow conversation held up every conversation that
// shared its partition. Here every conversation is a partition of its own,
// created by the first message sent to it, so a slow conversation waits on
// itself and on nothing else.
//
//   chat-messages (one partition per conversation)
//     ├── group "delivery"    marks each message delivered, fast
//     ├── group "enrichment"  translates the Japanese conversation, slow
//     └── group "sentiment"   added later, reads the whole history
//
// The program checks what the design promises: every message reaches each
// group once and in the order of its conversation, and the English
// conversations finish while the Japanese one is still being translated.
//
// 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 Japanese one needs a translation pass, 400 ms a
// message against 10 ms for the others. It is listed first, so its messages are
// the oldest in the queue and its partition is usually handed out first: a
// consumer that let one conversation hold up another would fail the timing
// check below.
//
// (conversationId, locale, needsTranslation)
const CONVERSATIONS: [(&str, &str, bool); 3] = [
    ("conv-jp-1", "jp", true),
    ("conv-en-1", "en", false),
    ("conv-en-2", "en", false),
];
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());
    // A fresh queue per run, so two runs never read each other's messages.
    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
    // shutdown, through close() at the bottom.
    let queen = Queen::connect(Config::new(&url)).map_err(|e| e.to_string())?;

    // A crashed worker's messages come back when its lease expires, and
    // retry_limit bounds how often a failing message is retried before it goes
    // to the dead-letter queue. configure() merges, so an option left out keeps
    // the value the queue already has; this queue is new, so those are the
    // broker's defaults.
    queen
        .queue(&messages)
        .configure(QueueOptions {
            lease_time: Some(60),
            retry_limit: Some(3),
            ..Default::default()
        })
        .await
        .map_err(|e| e.to_string())?;

    // ------------------------------------------------------------------ sending
    //
    // Sending a message is one push into the conversation's partition. Nothing
    // was declared for the conversation beforehand, 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 payload = json!({
                "conversationId": conversation_id,
                "seq": seq,
                "locale": locale,
                "body": format!("message {seq} in {conversation_id}"),
            });

            // The transaction id is the phone's own id for the message. A
            // phone that retries a send it never saw answered writes nothing
            // the second time. push() would mint a UUIDv7 id of its own, so
            // the item is built by hand and sent with 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()
    );

    // The phone resends message 1 because the answer got lost on a bad network.
    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 resent = resent
        .first()
        .ok_or("the broker answered the resend with no result")?;
    checks.assert(
        resent.status == PushStatus::Duplicate,
        "a resent message was recognised and not stored twice",
    )?;

    // Three workers in one consumer group. partitions(1) makes every pop take
    // ONE conversation: by default a pop may sweep up several ready
    // conversations, and a worker handles the messages of one pop in order, so
    // a slow conversation would delay the others that came with it.
    //
    // In this client limit() counts across the three workers, so a phase ends
    // as soon as every message has been handled. idle() is the deadline behind
    // that count, for a message that never comes, and it is pool-wide too: the
    // first worker that has waited that long stops all three, even one that is
    // halfway through a conversation, so it has to outlast the 2.4 s the
    // Japanese conversation keeps one worker busy. Long polls end after a
    // second, so both are noticed promptly. A service sets none of the three
    // and runs until it is cancelled.
    //
    // The handlers run on three tasks at once, so what they record is behind a
    // mutex.
    let workers = |group: &str| {
        queen
            .queue(&messages)
            .group(group)
            // A group created after the messages starts at the tail otherwise.
            .subscription_mode(SubscriptionMode::All)
            .concurrency(3)
            .partitions(1)
            .limit(sent as u64)
            .poll_timeout(Duration::from_secs(1))
            .idle(Duration::from_secs(5))
    };

    // --------------------------------------------------------------- delivering
    //
    // Marking messages delivered is fast work and must never wait behind slow
    // work, so it is a consumer group of its own, with its own cursor. A
    // handler that returns Ok acks the message, and one that returns Err nacks
    // it.
    println!("\ndelivering");
    let delivered: Arc<Mutex<HashMap<String, Vec<i64>>>> = Arc::new(Mutex::new(HashMap::new()));
    {
        let sink = Arc::clone(&delivered);
        workers("delivery")
            .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();
    let delivered_count: usize = delivered.values().map(Vec::len).sum();
    checks.assert(
        delivered_count == sent,
        &format!("delivery saw all {sent} messages once (got {delivered_count})"),
    )?;

    // Checked in the order of CONVERSATIONS, because a HashMap has no iteration
    // order of its own and a passing run should print the same lines each time.
    for (conversation_id, _, _) in CONVERSATIONS {
        let seqs = delivered.get(conversation_id).cloned().unwrap_or_default();
        let listed: Vec<String> = seqs.iter().map(i64::to_string).collect();
        checks.assert(
            seqs.iter().enumerate().all(|(i, &seq)| seq == i as i64 + 1),
            &format!(
                "{conversation_id} was delivered in order: {}",
                listed.join(",")
            ),
        )?;
    }

    // --------------------------------------------------------------- enrichment
    //
    // The slow group reads the same messages through its own cursor. On a topic
    // with a few hashed partitions, the Japanese conversation would sit in a
    // partition shared with English ones and hold them up. Here each worker
    // holds one conversation at a time, so the English conversations finish
    // while the Japanese one is still being translated.
    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);
        workers("enrichment")
            .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;
                    // One conversation is handled by one worker at a time, so
                    // the last write for a conversation is when it finished.
                    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 English conversations finished while the Japanese one was still being translated",
    )?;
    checks.assert(
        slow >= (MESSAGES_PER_CONVERSATION as u128) * 400,
        "the Japanese conversation really took its six translations",
    )?;

    // ----------------------------------------------------------------- backfill
    //
    // A feature added later, sentiment scoring, wants every message ever sent.
    // It is one more consumer group starting from the oldest message: no
    // producer change, and no second copy of the data.
    println!("\nbackfilling a new group");
    let scored = Arc::new(Mutex::new(0usize));
    {
        let counter = Arc::clone(&scored);
        workers("sentiment")
            .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,
        &format!("a group created now read the whole history ({scored} messages)"),
    )?;

    // 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 backend: one ordered partition per conversation.
//
// Queen started as the broker of a hotel messaging product. Some conversations
// need a translation or an agent before their next message can be handled, and
// on a hashed Kafka topic one slow conversation held up every conversation that
// shared its partition. Here every conversation is a partition of its own,
// created by the first message sent to it, so a slow conversation waits on
// itself and on nothing else.
//
//   chat-messages (one partition per conversation)
//     ├── group "delivery"    marks each message delivered, fast
//     ├── group "enrichment"  translates the Japanese conversation, slow
//     └── group "sentiment"   added later, reads the whole history
//
// The program checks what the design promises: every message reaches each
// group once and in the order of its conversation, and the English
// conversations finish while the Japanese one is still being translated.
//
// 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';
// A fresh queue per run, so two runs never read each other's messages.
$RUN = base_convert((string) (int) (microtime(true) * 1000), 10, 36);
$MESSAGES = "app-php-chat-{$RUN}";

// Three conversations. The Japanese one needs a translation pass, 400 ms a
// message against 10 ms for the others. It is listed first, so its messages are
// the oldest in the queue and its partition is usually handed out first: a
// consumer that let one conversation hold up another would fail the timing
// check below.
$CONVERSATIONS = [
    'conv-jp-1' => ['locale' => 'jp', 'needsTranslation' => true],
    'conv-en-1' => ['locale' => 'en', 'needsTranslation' => false],
    'conv-en-2' => ['locale' => 'en', 'needsTranslation' => false],
];
$MESSAGES_PER_CONVERSATION = 6;
$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 fails here
    // and leaves no queue behind.
    if (!function_exists('pcntl_fork')) {
        throw new RuntimeException('this example forks its worker pool and needs the pcntl extension');
    }

    // A crashed worker's messages come back when its lease expires, and
    // retryLimit bounds how often a failing message is retried before it goes to
    // the dead-letter queue.
    $queen->queue($MESSAGES)->config(['leaseTime' => 60, 'retryLimit' => 3])->create()->execute();

    // ------------------------------------------------------------------ sending
    //
    // Sending a message is one push into the conversation's partition. Nothing
    // was declared for the conversation beforehand, 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 phone's own id for the message. A phone that retries a send
            // it never saw answered writes nothing the second time, and gets
            // the first message's id back.
            $queen->queue($MESSAGES)->partition($conversationId)->push([[
                'transactionId' => "{$conversationId}-{$seq}",
                'data' => $message,
            ]])->execute();
            $sent++;
        }
    }
    echo "  {$sent} messages across " . count($CONVERSATIONS) . " conversations\n";

    // The phone resends message 1 because the answer got lost on a bad network.
    // execute() hands back one result per item pushed, and the result carries
    // the verdict: 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 recognised and not stored twice');

    // --------------------------------------------------------------- delivering
    //
    // Marking messages delivered is fast work and must never wait behind slow
    // work, so it is a consumer group of its own, with its own cursor.
    //
    // concurrency(3) on this client is three long polls in flight at once on
    // one cURL multi-handle. The polls overlap, and the handlers run one after
    // another in this process. partitions(1) makes every pop take ONE
    // conversation: by default a pop may sweep up several ready conversations,
    // and a worker handles the messages of one pop in order, so a slow
    // conversation would delay the others that came with it. Long polls end
    // after a second and a worker stops after two quiet seconds, which is what
    // lets this program finish; a service calls consume() without those two
    // lines and runs until stopped.
    echo "\ndelivering\n";
    $delivered = [];
    $queen
        ->queue($MESSAGES)
        ->group('delivery')
        ->subscriptionMode('all') // a group created after the messages starts at the tail otherwise
        ->concurrency(3)
        ->partitions(1)
        ->each()
        ->timeoutMillis(1000)
        ->idleMillis(2000)
        ->consume(function (array $msg) use (&$delivered, $sleepMillis): void {
            $sleepMillis(10);
            $delivered[$msg['data']['conversationId']][] = $msg['data']['seq'];
        })
        ->execute();

    $deliveredCount = array_sum(array_map('count', $delivered));
    $assert($deliveredCount === $sent, "delivery saw all {$sent} messages once (got {$deliveredCount})");
    foreach ($delivered as $conversationId => $seqs) {
        $assert(
            $seqs === range(1, count($seqs)),
            "{$conversationId} was delivered in order: " . implode(',', $seqs)
        );
    }

    // --------------------------------------------------------------- enrichment
    //
    // The slow group reads the same messages through its own cursor. On a topic
    // with a few hashed partitions, the Japanese conversation would sit in a
    // partition shared with English ones and hold them up. Here each worker
    // holds one conversation at a time, so the English conversations finish
    // while the Japanese one is still being translated.
    //
    // 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 partitions. 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')
                    // One conversation per pop, as above. Without it the first
                    // process to poll can take all three conversations and
                    // translate the Japanese one before the English ones.
                    ->partitions(1)
                    ->each()
                    ->timeoutMillis(1000)
                    ->idleMillis(2000)
                    ->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, and never 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 three processes enriched all {$sent} messages once (got {$enriched})");

    $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 English conversations finished while the Japanese one was still being translated'
    );
    $assert(
        $slow >= $MESSAGES_PER_CONVERSATION * 400,
        'the Japanese conversation really took its six translations'
    );

    // ----------------------------------------------------------------- backfill
    //
    // A feature added later, sentiment scoring, wants every message ever sent.
    // It is one more consumer group starting from the oldest message: no
    // producer change, and no second copy of the data.
    echo "\nbackfilling a new group\n";
    $scored = 0;
    $queen
        ->queue($MESSAGES)
        ->group('sentiment')
        ->subscriptionMode('all')
        ->concurrency(3)
        ->partitions(1)
        ->each()
        ->timeoutMillis(1000)
        ->idleMillis(2000)
        ->consume(function (array $msg) use (&$scored): void {
            $scored++;
        })
        ->execute();

    $assert($scored === $sent, "a group created now read the whole history ({$scored} messages)");

    $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 backend: one ordered partition per conversation.
//
// Queen started as the broker of a hotel messaging product. Some conversations
// need a translation or an agent before their next message can be handled, and
// on a hashed Kafka topic one slow conversation held up every conversation that
// shared its partition. Here every conversation is a partition of its own,
// created by the first message sent to it, so a slow conversation waits on
// itself and on nothing else.
//
//   chat-messages (one partition per conversation)
//     |-- group "delivery"    marks each message delivered, fast
//     |-- group "enrichment"  translates the Japanese conversation, slow
//     `-- group "sentiment"   added later, reads the whole history
//
// The program checks what the design promises: every message reaches each
// group once and in the order of its conversation, and the English
// conversations finish while the Japanese one is still being translated.
//
// Build it. queen_client.hpp needs json.hpp and threadpool.hpp, which this
// repository carries under clients/server, and cpp-httplib, which it does not
// (Homebrew puts httplib.h under /opt/homebrew/include). The client switches on
// cpp-httplib's OpenSSL support, so -lssl -lcrypto are needed 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 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 Japanese one needs a translation pass, 400 ms a
// message against 10 ms for the others. It is listed first, so its messages are
// the oldest in the queue and its partition is usually handed out first: a
// consumer that let one conversation hold up another would fail the timing
// check below.
static const std::vector<Conversation> CONVERSATIONS = {
    {"conv-jp-1", "jp", true},
    {"conv-en-1", "en", false},
    {"conv-en-2", "en", false},
};
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();

    // There is no signal-handling option to turn off on this client:
    // QueenClient always installs its own SIGINT and SIGTERM handlers, and its
    // SIGINT handler ends the process on the spot, with exit status 0.
    QueenClient client(QUEEN_URL);

    std::string verdict;
    bool failed = false;

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

        // A crashed worker's messages come back when its lease expires, and
        // retry_limit bounds how often a failing message is retried before it
        // goes to the dead-letter queue. The C++ QueueConfig is a struct, and an
        // option it has no field for keeps the broker's default.
        queen::QueueConfig config;
        config.lease_time = 60;
        config.retry_limit = 3;
        client.queue(MESSAGES).config(config).create();

        // -------------------------------------------------------------- sending
        //
        // Sending a message is one push into the conversation's partition.
        // Nothing was declared for the conversation beforehand, 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 phone's own id for the message. A
                // phone that retries a send it never saw answered writes nothing
                // the second time, and gets the first message's id back.
                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;

        // The phone resends message 1 because the answer got lost on a bad
        // network. 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 recognised and not stored twice");

        // ----------------------------------------------------------- delivering
        //
        // Marking messages delivered is fast work and must never wait behind
        // slow work, so it is a consumer group of its own, with its own cursor.
        //
        // concurrency(3) runs three workers, each a thread with a poll loop of
        // its own. partitions(1) makes every pop take ONE conversation: by
        // default a pop may sweep up several ready conversations, and a worker
        // handles the messages of one pop in order, so a slow conversation
        // would delay the others that came with it.
        //
        // consume() blocks this thread until every worker has stopped, and
        // three settings bound it:
        //
        //   wait(false)     every pop answers at once, and a worker that found
        //                   nothing sleeps briefly before the next one. This
        //                   client has no setter for the long-poll timeout,
        //                   which is 30 seconds, and the stop flag and the idle
        //                   deadline are only consulted between pops.
        //   idle_millis     the deadline: a lost message fails this run instead
        //                   of hanging it.
        //   limit()         counts per worker: each worker keeps its own tally,
        //                   so three workers sharing 18 messages never see one
        //                   worker reach 18. It is only a backstop.
        //
        // What ends each phase is a shared counter, which raises the stop flag
        // the moment every message has been handled.
        //
        // consume() 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)
            .partitions(1)
            .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 all " + std::to_string(sent) + " messages once (got " +
                  std::to_string(delivered_total) + ")");

        // A conversation is leased to one worker at a time, and the lease moves
        // on only when that worker has acknowledged what it took, 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: " + join(delivered[conv.id]));
        }

        // ----------------------------------------------------------- enrichment
        //
        // The slow group reads the same messages through its own cursor. On a
        // topic with a few hashed partitions, the Japanese conversation would
        // sit in a partition shared with English ones and hold them up. Here
        // each worker holds one conversation at a time, so the English
        // conversations finish while the Japanese one is still being
        // translated.
        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)
            .partitions(1)
            .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 English conversations finished while the Japanese one was "
              "still being translated");
        check(slow >= MESSAGES_PER_CONVERSATION * 400,
              "the Japanese conversation really took its six translations");

        // ------------------------------------------------------------- backfill
        //
        // A feature added later, sentiment scoring, wants every message ever
        // sent. It is one more consumer group starting from the oldest message:
        // no producer change, and no second copy of the data.
        //
        // subscription_mode("all") is what points the new cursor at the oldest
        // message. A new group starts at the tail by default, so without it this
        // group would wait for the next message, and the phase would end on the
        // idle deadline with nothing scored.
        std::cout << "\nbackfilling a new group" << std::endl;
        std::atomic<int> scored{0};
        handled = 0;
        stop = false;

        client.queue(MESSAGES)
            .group("sentiment")
            .subscription_mode("all")
            .concurrency(3)
            .partitions(1)
            .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 created now read the whole history (" +
                  std::to_string(scored.load()) + " messages)");

        // Clean up on success only: a failed run leaves the queue on the broker
        // to be looked at. The method is del() because delete is a keyword.
        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: PASS or
    // FAIL stays the last line of a run.
    client.close();

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