Workflow pattern: one step, one transaction

await queen.queue('order-events')
  .group('workflow')
  .subscriptionMode('all')
  .autoAck(false)
  .each()
  .consume(async (msg) => {
    const { orderId } = msg.data
    const order = await queen.kv.get('orders', orderId)            // { status: 'paid' }
    if (order.value.status !== 'paid') {                   // already moved: just ack
      return queen.ack(msg, 'completed', { group: 'workflow' })
    }

    // Do the work of this step here: charge the card, book the courier, ...

    const result = await queen.transaction()
      .ack(msg, 'completed', { consumerGroup: 'workflow' })
      .kv.put('orders', orderId, { status: 'shipping' },
        { ttl: '30d', expect: order.version, required: true })  // only from what we read
      .queue('shipping').partition(orderId)
      .push([{ transactionId: `ship-${orderId}`, data: { orderId } }])
      .commit()
    if (result.success === false) return  // moved by someone else: nothing written
  })

async def step(msg):
    order_id = msg["data"]["orderId"]
    order = await queen.kv.get("orders", order_id)          # {'status': 'paid'}
    if order["value"]["status"] != "paid":                   # already moved: just ack
        return await queen.ack(msg, "completed", {"group": "workflow"})

    # Do the work of this step here: charge the card, book the courier, ...

    result = await (
        queen.transaction()
        .ack(msg, "completed", {"consumer_group": "workflow"})
        .kv.put("orders", order_id, {"status": "shipping"},
                ttl=timedelta(days=30), expect=order["version"],
                required=True)                               # only from what we read
        .queue("shipping").partition(order_id)
        .push([{"transactionId": f"ship-{order_id}", "data": {"orderId": order_id}}])
        .commit()
    )
    if not result:                    # moved by someone else: nothing written
        return

await (
    queen.queue("order-events")
    .group("workflow")
    .subscription_mode("all")
    .auto_ack(False)
    .each()
    .consume(step)
)

err = client.Queue("order-events").
	Group("workflow").
	SubscriptionMode(queen.SubscriptionModeAll).
	AutoAck(false).
	Each().
	Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
		opts := queen.AckOptions{ConsumerGroup: "workflow"}
		orderID := msg.Data["orderId"].(string)
		order, err := client.KV().Get(ctx, "orders", orderID) // Value {"status":"paid"}
		if err != nil {
			return err
		}
		var state struct{ Status string }
		if err := json.Unmarshal(order.Value, &state); err != nil {
			return err
		}
		if state.Status != "paid" { // already moved: just ack
			_, err := client.Ack(ctx, msg, true, opts)
			return err
		}

		// Do the work of this step here: charge the card, book the courier, ...

		next := map[string]interface{}{"status": "shipping"}
		result, err := client.Transaction().
			Ack(msg, queen.AckStatusCompleted, opts).
			KV(queen.KVPutOp("orders", orderID, next, queen.TTL(30*24*time.Hour),
				queen.KVWriteOptions{
					Expect:   queen.Expect(order.Version), // only from what we read
					Required: true,
				})).
			Queue("shipping").Partition(orderID).
			Push(queen.PushItem{TransactionID: "ship-" + orderID,
				Payload: map[string]interface{}{"orderId": orderID}}).
			Commit(ctx)
		if err != nil {
			return err
		}
		if result.IsKVPrecondition() {
			fmt.Println(orderID, "moved by someone else: nothing written")
		}
		return nil
	}).
	Execute(ctx)
if err != nil {
	log.Fatal(err)
}

let client = queen.clone();
queen
    .queue("order-events")
    .group("workflow")
    .subscription_mode(SubscriptionMode::All)
    .auto_ack(false)
    .consume(move |msg: Message| {
        let queen = client.clone();
        async move {
            let order_id = msg.data["orderId"].as_str().unwrap_or_default().to_string();
            let order = queen.kv().get("orders", &order_id).await?; // {"status":"paid"}
            if order.value.as_ref().and_then(|v| v["status"].as_str()) != Some("paid") {
                queen.ack(&msg).await?; // already moved: just ack
                return Ok(());
            }

            // Do the work of this step here: charge the card, book the courier, ...

            let thirty_days = Expiry::seconds(30 * 86400);
            let shipping = queen
                .kv()
                .put("orders", &order_id, json!({ "status": "shipping" }), thirty_days)
                .expect(order.version.unwrap_or(0)) // only from what we read
                .required()
                .operation()?;
            let result = queen
                .transaction()
                .ack(&msg)
                .kv(shipping)
                .push_item(TxnPushItem {
                    queue: "shipping".into(),
                    partition: Some(order_id.clone()),
                    payload: json!({ "orderId": order_id }),
                    transaction_id: Some(format!("ship-{order_id}")),
                    trace_id: None,
                })?
                .commit()
                .await?;
            if !result.success {
                return Ok(()); // moved by someone else: nothing written
            }
            Ok::<_, queen_mq::Error>(())
        }
    })
    .await?;

$queen->queue('order-events')
    ->group('workflow')
    ->subscriptionMode('all')
    ->autoAck(false)
    ->each()
    ->consume(function (array $msg) use ($queen) {
        $orderId = $msg['data']['orderId'];
        $order = $queen->kv()->get('orders', $orderId);       // value: [status => paid]
        if ($order['value']['status'] !== 'paid') {           // already moved: just ack
            $queen->ack($msg, 'completed', ['group' => 'workflow']);
            return;
        }

        // Do the work of this step here: charge the card, book the courier, ...

        $result = $queen->transaction()
            ->ack($msg, 'completed', ['consumerGroup' => 'workflow'])
            ->kv('orders')->put($orderId, ['status' => 'shipping'], [
                'ttlSeconds' => 30 * 86400,
                'expect' => $order['version'],                // only from what we read
                'required' => true,
            ])
            ->queue('shipping')->partition($orderId)->push([
                ['transactionId' => "ship-{$orderId}", 'data' => ['orderId' => $orderId]],
            ])
            ->commit();
        if ($result['success'] === false) {
            return;                        // moved by someone else: nothing written
        }
    })
    ->execute();

client.queue("order-events")
    .group("workflow")
    .subscription_mode("all")
    .auto_ack(false)
    .each()
    .consume([&](const json& msg) {
        const std::string order_id = msg["data"]["orderId"];
        json order = client.kv("orders").get(order_id);  // value: {"status":"paid"}
        if (order["value"]["status"] != "paid") {        // already moved: just ack
            client.ack(msg, true, {{"group", "workflow"}});
            return;
        }

        // Do the work of this step here: charge the card, book the courier, ...

        queen::KvWriteOptions from_what_we_read;         // only from what we read
        from_what_we_read.expect = order["version"].get<long long>();
        from_what_we_read.required = true;
        // client.transaction() cannot give its push a partition, so the step goes
        // to the broker's transaction route through the client's own HTTP layer.
        json step = {
            {"operations", json::array({
                {{"type", "ack"}, {"status", "completed"},
                 {"consumerGroup", "workflow"}, {"leaseId", msg["leaseId"]},
                 {"transactionId", msg["transactionId"]},
                 {"partitionId", msg["partitionId"]}},
                {{"type", "push"}, {"items", json::array({{
                    {"queue", "shipping"}, {"partition", order_id},
                    {"transactionId", "ship-" + order_id},
                    {"payload", {{"orderId", order_id}}},
                }})}},
            })},
            {"requiredLeases", json::array({msg["leaseId"]})},
            {"kv", json::array({queen::wire::kv_write_op("put", "orders", order_id,
                {{"status", "shipping"}}, queen::KvTtl::seconds(30 * 86400),
                from_what_we_read)})},
        };
        json result = client.get_http_client()->post("/api/v1/transaction", step);
        bool committed = result.value("success", false);
        if (!committed) return;  // moved by someone else: nothing written
    });