Event-sourced cart

Define event types

Cart events are modelled as a Rust enum with #[serde(tag = "type")] for tagged JSON serialization. Each variant carries a cart_id used as the record key. A helper method extracts the cart ID regardless of variant.

#[derive(Serialize, Deserialize)]
#[serde(tag = "type")]
enum CartEvent {
    AddItem {
        cart_id: String,
        item: String,
        qty: u32,
        price: f64,
    },
    RemoveItem {
        cart_id: String,
        item: String,
    },
    Checkout {
        cart_id: String,
    },
}

impl CartEvent {
    fn cart_id(&self) -> &str {
        match self {
            CartEvent::AddItem { cart_id, .. } => cart_id,
            CartEvent::RemoveItem { cart_id, .. } => cart_id,
            CartEvent::Checkout { cart_id, .. } => cart_id,
        }
    }
}

An AddItem event serializes to JSON like:

Produce cart events

30 events are produced across 3 carts. Carts 1 and 2 go through a full lifecycle: add items, remove some, checkout. Cart 3 has items added but hasn't checked out yet. The cart ID is used as the record key.

let producer = Broker::producer(&broker);

let events: Vec<CartEvent> = vec![
    // Cart 1: add items, remove one, checkout
    CartEvent::AddItem { cart_id: "cart-1".into(),
        item: "Widget".into(), qty: 2, price: 9.99 },
    CartEvent::AddItem { cart_id: "cart-1".into(),
        item: "Gadget".into(), qty: 1, price: 24.99 },
    // ... more events ...
    CartEvent::Checkout { cart_id: "cart-1".into() },
    // Cart 2: similar lifecycle
    // Cart 3: items added, no checkout
];

for event in &events {
    let value = serde_json::to_string(event).unwrap();
    producer.send(&ProducerRecord::new(
        "cart-events",
        Some(event.cart_id().to_string()),
        value,
    )).unwrap();
}

Projection 1 — Cart state

The “cart-state” consumer group replays all 30 events and builds a HashMap<cart_id, Vec<LineItem>>. The apply_cart_event function handles each variant: AddItem inserts or updates quantities, RemoveItem drops the item, Checkout is a no-op for this projection.

let mut cart_state: HashMap<String, Vec<LineItem>> = HashMap::new();

let mut consumer = Broker::consumer(&broker, ConsumerConfig {
    group_id: "cart-state".into(),
    auto_commit: false,
    offset_reset: OffsetReset::Earliest,
});
consumer.subscribe(&["cart-events"]).unwrap();
let records = consumer.poll(Duration::from_millis(100)).unwrap();

for record in &records {
    let event: CartEvent = serde_json::from_str(&record.value).unwrap();
    apply_cart_event(&mut cart_state, &event);
}
consumer.commit_sync().unwrap();

Projection 2 — Revenue report

A second consumer group (“revenue”) reads the same 30 events but builds a different view. It tracks which carts have checked out and only counts their line item totals toward revenue. Cart 3 hasn't checked out, so it's excluded from the total.

let mut checked_out: HashMap<String, bool> = HashMap::new();
let mut revenue_items: HashMap<String, Vec<LineItem>> = HashMap::new();

let mut consumer = Broker::consumer(&broker, ConsumerConfig {
    group_id: "revenue".into(),
    auto_commit: false,
    offset_reset: OffsetReset::Earliest,
});
consumer.subscribe(&["cart-events"]).unwrap();
let records = consumer.poll(Duration::from_millis(100)).unwrap();

for record in &records {
    let event: CartEvent = serde_json::from_str(&record.value).unwrap();
    match &event {
        CartEvent::Checkout { cart_id } => {
            checked_out.insert(cart_id.clone(), true);
        }
        _ => apply_cart_event(&mut revenue_items, &event),
    }
}
consumer.commit_sync().unwrap();

Produce incremental events

5 more events complete cart-3: add two new items, remove one existing item, add another, and finally checkout. These events are appended to the same topic.

let new_events = vec![
    CartEvent::AddItem { cart_id: "cart-3".into(),
        item: "Clamp".into(), qty: 5, price: 2.50 },
    CartEvent::AddItem { cart_id: "cart-3".into(),
        item: "Hinge".into(), qty: 8, price: 1.75 },
    CartEvent::RemoveItem { cart_id: "cart-3".into(),
        item: "Anchor".into() },
    CartEvent::AddItem { cart_id: "cart-3".into(),
        item: "Latch".into(), qty: 3, price: 4.00 },
    CartEvent::Checkout { cart_id: "cart-3".into() },
];

Incremental projection updates

Both consumer groups poll again. Because they committed their offsets after step 4/5, they each receive exactly the 5 new events—not the full 35. Each projection incrementally updates its view without replaying the entire history.

// Cart state consumer picks up from offset 30
let records = cart_consumer.poll(Duration::from_millis(100)).unwrap();
assert_eq!(records.len(), 5);
for record in &records {
    let event: CartEvent = serde_json::from_str(&record.value).unwrap();
    apply_cart_event(&mut cart_state, &event);
}

// Revenue consumer also picks up from offset 30
let records = revenue_consumer.poll(Duration::from_millis(100)).unwrap();
assert_eq!(records.len(), 5);

Updated projections

After processing the 5 new events, both projections reflect the updated state. Cart-3 now has 8 items totalling $76.00 (with Anchor removed, Clamp/Hinge/Latch added), and the revenue report now includes cart-3 since it checked out.