Event-Driven Architecture Pro Playbook

# SYNC: checkout is only as available as its slowest dependency
def checkout(order):
    payment.charge(order)      # blocks
    fraud.check(order)         # blocks
    email.send(order)          # blocks → 3 ways to fail the sale

# ASYNC: emit once, return; consumers react independently
def checkout(order):
    db.save(order, status="placed")
    broker.publish("OrderPlaced", order)   # fire and continue
    return "ok"                # payment/fraud/email run downstream

# QUEUE — distribute work, each task done once
queue.send("resize-image", {"id": 42})     # exactly one worker resizes it
job = queue.receive(); process(job); job.ack()   # ack → removed

# STREAM — publish a fact many services consume & can replay
stream.append("orders", {"type":"OrderPlaced","id":42})
# billing, search, analytics each read from their OWN offset:
for event in stream.consume("orders", group="billing", from_offset="last"):
    handle(event)         # audit group can start from 0 to rebuild state

# producer — fire one event to a topic
broker.publish("OrderPlaced", {"order_id": 42, "total": 90})

# each consumer subscribes independently (own group = own copy)
broker.subscribe("OrderPlaced", group="billing",   handler=charge)
broker.subscribe("OrderPlaced", group="inventory", handler=reserve)
broker.subscribe("OrderPlaced", group="search",    handler=index)
# tomorrow: add group="loyalty" → zero changes to checkout

# producer: key by customer so their events stay ordered
producer.send("orders", key=str(customer_id), value=event)
#   same customer_id → same partition → strict order for that customer

# consumer group: N consumers share N+ partitions
consumer.subscribe(["orders"], group_id="billing")
for msg in consumer:
    handle(msg.value)
    consumer.commit()        # advance offset ONLY after success (topic 5)
# add partitions to scale; add consumers up to #partitions

for event in consumer:
    if seen.contains(event.id):        # idempotent guard (topic 6)
        consumer.commit(); continue    # duplicate → skip, then ack
    process(event)                     # do the work
    seen.add(event.id)
    consumer.commit()                  # ACK only AFTER success
# crash before commit → event redelivered → guard makes it a no-op
# this is "effectively-once" — the practical form of exactly-once

def handle(event):
    with db.transaction():                       # ONE atomic unit
        try:
            db.insert("processed_events", id=event.id)   # UNIQUE column
        except UniqueViolation:
            return                               # duplicate → no-op ✓
        apply_effect(event)                      # do the work in the SAME tx
# the unique constraint + single transaction close the race two
# concurrent duplicates would otherwise slip through (topic 5)
# even better: naturally idempotent ops — "set status=shipped" not "ship again"

# ✓ per-account ordering: key by account_id
producer.send("account-events", key=str(account_id), value=event)
#   Opened → Deposit → Closed for acct-7 ALL land on one partition, in order

# ✗ keying by something else (or none) scatters an account's events →
#   Closed can be handled before Opened → invalid state
# rule: partition key = the entity whose event order you must preserve
# need strict GLOBAL order? one partition — but that caps throughput (topic 4)

# CHOREOGRAPHY — each service reacts & emits the next event
on("OrderPlaced",  lambda e: (charge(e), emit("Paid", e)))
on("Paid",         lambda e: (reserve(e), emit("Reserved", e)))
on("Reserved",     lambda e: (ship(e), emit("Shipped", e)))
# no central view; flow = the sum of these subscriptions

# ORCHESTRATION — a coordinator drives & sees the whole flow
def fulfill(order):
    charge(order); reserve(order); ship(order); notify(order)
    # one place to read, reorder, add compensation (topic 11)