Code snippets — Event Driven Architecture

snippet-2.go

func ReplayOrder(events []Event) Order {
    var order Order
    for _, event := range events {
        switch e := event.(type) {
        case OrderPlaced:
            order.ID = e.OrderID
            order.Status = "pending"
            order.Items = e.Items
            order.Total = e.Total
        case PaymentTaken:
            order.Status = "paid"
            order.PaidAt = e.Timestamp
        case OrderShipped:
            order.Status = "shipped"
            order.TrackingCode = e.TrackingCode
        }
    }
    return order
}

snippet-2.txt

Order Service → emits "OrderPlaced" event
                          ↓
              Payment Service (subscribes) → processes payment → emits "PaymentCompleted"
              Inventory Service (subscribes) → reserves stock
              Email Service (subscribes) → sends confirmation
              Analytics Service (subscribes) → records sale

snippet-3.go

// WRONG: Race condition — DB succeeds but event publish fails
func PlaceOrder(order Order) error {
    db.Save(order)                    // DB commit
    eventBus.Publish("OrderPlaced")   // What if this fails?
    return nil
}

snippet-3.txt

Traditional:
┌─────────────────────────────┐
│ Order Table                 │
│ id: 42, status: "shipped"   │
│ total: 99.99, items: [...]  │
└─────────────────────────────┘

Event Sourced:
┌──────────────────────────────────────────┐
│ Order Events Table                       │
│ 1. OrderPlaced   (2026-01-10 10:00)     │
│ 2. PaymentTaken  (2026-01-10 10:01)     │
│ 3. ItemsReserved (2026-01-10 10:02)     │
│ 4. OrderShipped  (2026-01-10 14:30)     │
└──────────────────────────────────────────┘

snippet-4.go

func PlaceOrder(ctx context.Context, order Order) error {
    return db.Transaction(func(tx *sql.Tx) error {
        // 1. Save the order
        if err := tx.Exec(`INSERT INTO orders ...`, order); err != nil {
            return err
        }

        // 2. Write event to outbox IN THE SAME TRANSACTION
        event := OrderPlacedEvent{OrderID: order.ID, ...}
        eventJSON, _ := json.Marshal(event)
        tx.Exec(`INSERT INTO outbox (event_type, payload, status)
                 VALUES (?, ?, 'pending')`, "OrderPlaced", eventJSON)

        return nil
    })
}

// Separate outbox processor (runs periodically)
func ProcessOutbox(ctx context.Context) {
    for {
        events, _ := db.Query(`SELECT id, event_type, payload
                                FROM outbox WHERE status = 'pending'
                                ORDER BY created_at LIMIT 100`)
        for _, event := range events {
            if err := eventBus.Publish(event.Type, event.Payload); err != nil {
                continue // Will retry next cycle
            }
            db.Exec(`UPDATE outbox SET status = 'published' WHERE id = ?`, event.ID)
        }
        time.Sleep(100 * time.Millisecond)
    }
}

snippet-4.txt

Write Side (Commands):
User → PlaceOrder command → Order Service → stores events → publishes OrderPlaced

Read Side (Queries):
OrderPlaced event → Projection Service → updates read model (denormalized SQL/Redis)
User → GetOrderStatus query → reads from read model (fast!)

snippet-5.go

func HandleOrderPlaced(ctx context.Context, event OrderPlacedEvent) error {
    // Check if already processed
    if processed, _ := processedEvents.Has(event.EventID); processed {
        return nil  // Already handled, skip
    }

    // Process the event
    if err := paymentService.Charge(ctx, event); err != nil {
        return err
    }

    // Mark as processed
    processedEvents.Set(event.EventID, 24*time.Hour)
    return nil
}

snippet-6.go

// Too granular — causes chatty event storms
UserFirstNameChanged
UserLastNameChanged
UserEmailChanged

// Better — meaningful business event
UserProfileUpdated { Changes: [{field: "email", newValue: "..."}] }

snippet-7.go

// After N retries, send to DLQ instead of dropping
func handleWithRetry(event Event, maxRetries int) {
    for attempt := 0; attempt < maxRetries; attempt++ {
        if err := processEvent(event); err == nil {
            return
        }
        time.Sleep(backoff(attempt))
    }
    dlq.Publish(event)  // Park for manual inspection
}

snippet.go

// Something happened in the Order domain
type OrderPlaced struct {
    OrderID    string    `json:"order_id"`
    CustomerID string    `json:"customer_id"`
    Items      []Item    `json:"items"`
    Total      float64   `json:"total"`
    PlacedAt   time.Time `json:"placed_at"`
}

type OrderCancelled struct {
    OrderID    string    `json:"order_id"`
    Reason     string    `json:"reason"`
    CancelledAt time.Time `json:"cancelled_at"`
}

snippet.txt

Order Service → [HTTP POST] → Payment Service → [HTTP POST] → Inventory Service