Transactional outbox relay.

Writing the order to the database and then publishing OrderCreated to Kafka is a dual write: a crash between the two loses the event (Day 2, chapter 1). The transactional outbox writes the order and the event to the database in one transaction; a separate relay publishes outbox rows and marks them published (section 2.2).

The relay can still crash after publishing and before marking. On restart it publishes the same event again, so the pipeline is at-least-once, and consumers deduplicate by event id (section 3.2, solution 1). At-least-once delivery plus an idempotent consumer is what "exactly-once" means in practice: every event takes effect once.

You get a fake database with transactions and a fake broker (see Given). Build three pieces:

1. create_order(db, order_id, user_id, amount): inside one with db.transaction(): block, db.insert_order(...) and db.insert_outbox(aggregate_type="Order", aggregate_id=order_id, event_type="OrderCreated", payload={"order_id": order_id, "user_id": user_id, "amount": amount}). Return the outbox event id. If either write fails, neither may persist.

2. OutboxRelay(db, broker, clock, batch_size=100) with run_once() -> int:

• fetch up to batch_size unpublished rows, oldest id first (db.fetch_unpublished(limit))
• publish each to topic f"{aggregate_type.lower()}.events" (so order.events), key aggregate_id, value {"event_id": id, "event_type": ..., "payload": ...}
• after each successful publish, db.mark_published(id, clock.now())
• if broker.publish raises BrokerUnavailable, stop the batch there: the failed event and everything after it stay unpublished, so order is kept. Return how many were published
• a Crash is the process dying: let it propagate (it is a BaseException, so a plain except Exception won't swallow it)

3. IdempotentConsumer(handler) with receive(message) -> bool: call handler(message) the first time an event_id arrives and return True; return False for a duplicate. Remember an event id only after the handler succeeds, so a handler that raised gets the redelivery.

In JavaScript and TypeScript the names are camelCase: createOrder(db, orderId, userId, amount), new OutboxRelay(db, broker, clock, batchSize = 100) with runOnce(), and IdempotentConsumer with receive(message). The transaction is a callback: db.transaction(() => { ... }) rolls back every write if the callback throws and returns what it returns. Rows and messages use camelCase fields too: rows are { id, aggregateType, aggregateId, eventType, payload, publishedAt }, the message value is { eventId, eventType, payload }, and the payload is { orderId, userId, amount }. JavaScript has no BaseException, so Crash is an ordinary error: catch only BrokerUnavailable and let anything else propagate.