Event-Driven Architecture Patterns 2025

public class Order : AggregateRoot
{
    public Guid OrderId { get; private set; }
    public OrderStatus Status { get; private set; }
    public decimal Total { get; private set; }
    private List<OrderItem> _items = new();

    // Command handler
    public void PlaceOrder(Guid customerId, List<OrderItem> items)
    {
        if (Status != OrderStatus.Draft)
            throw new InvalidOperationException("Order already placed");

        var @event = new OrderPlacedEvent
        {
            OrderId = Guid.NewGuid(),
            CustomerId = customerId,
            Items = items,
            Total = items.Sum(i => i.Price * i.Quantity),
            Timestamp = DateTime.UtcNow
        };

        ApplyEvent(@event); // Apply & store event
    }

    // Event handler - mutates state
    private void Apply(OrderPlacedEvent @event)
    {
        OrderId = @event.OrderId;
        _items = @event.Items;
        Total = @event.Total;
        Status = OrderStatus.Placed;
    }

    // Replay from event stream
    public static Order FromEvents(IEnumerable<DomainEvent> events)
    {
        var order = new Order();
        foreach (var @event in events)
        {
            order.ApplyEvent(@event, isNew: false);
        }
        return order;
    }
}

-- Event Store table
CREATE TABLE EventStore (
    EventId UNIQUEIDENTIFIER PRIMARY KEY,
    AggregateId UNIQUEIDENTIFIER NOT NULL,
    AggregateType NVARCHAR(255) NOT NULL,
    EventType NVARCHAR(255) NOT NULL,
    EventData NVARCHAR(MAX) NOT NULL, -- JSON
    Version INT NOT NULL,
    Timestamp DATETIME2 NOT NULL,
    Metadata NVARCHAR(MAX),

    CONSTRAINT UQ_Aggregate_Version
        UNIQUE (AggregateId, Version)
);

CREATE INDEX IX_AggregateId
    ON EventStore(AggregateId, Version);

-- Sample query: Load aggregate
SELECT EventType, EventData, Version
FROM EventStore
WHERE AggregateId = 'order_789'
ORDER BY Version ASC;

public class SnapshotStore
{
    // Save snapshot every N events (e.g., every 100)
    public async Task SaveSnapshot(
        Guid aggregateId,
        object state,
        int version)
    {
        await _db.Snapshots.AddAsync(new Snapshot
        {
            AggregateId = aggregateId,
            State = JsonSerializer.Serialize(state),
            Version = version,
            Timestamp = DateTime.UtcNow
        });
    }

    // Load: snapshot + events after snapshot
    public async Task<Order> LoadAggregate(Guid orderId)
    {
        // 1. Load latest snapshot
        var snapshot = await _db.Snapshots
            .Where(s => s.AggregateId == orderId)
            .OrderByDescending(s => s.Version)
            .FirstOrDefaultAsync();

        Order order;
        int fromVersion;

        if (snapshot != null)
        {
            order = JsonSerializer
                .Deserialize<Order>(snapshot.State);
            fromVersion = snapshot.Version + 1;
        }
        else
        {
            order = new Order();
            fromVersion = 1;
        }

        // 2. Replay events after snapshot
        var events = await _eventStore
            .GetEvents(orderId, fromVersion);

        foreach (var @event in events)
        {
            order.ApplyEvent(@event);
        }

        return order;
    }
}

// Result: 10,000 events → 100 events replay
// (snapshot at 9,900 + 100 new events)

// Command
public record PlaceOrderCommand(
    Guid CustomerId,
    List<OrderItem> Items
) : ICommand;

// Command Handler
public class PlaceOrderCommandHandler
    : ICommandHandler<PlaceOrderCommand>
{
    private readonly IEventStore _eventStore;
    private readonly IEventBus _eventBus;

    public async Task<Result> Handle(
        PlaceOrderCommand command)
    {
        // 1. Validate
        if (!command.Items.Any())
            return Result.Failure("No items");

        // 2. Create aggregate
        var order = new Order();
        order.PlaceOrder(
            command.CustomerId,
            command.Items
        );

        // 3. Save events
        var events = order.GetUncommittedEvents();
        await _eventStore.SaveEvents(
            order.OrderId,
            events
        );

        // 4. Publish events
        foreach (var @event in events)
        {
            await _eventBus.Publish(@event);
        }

        return Result.Success(order.OrderId);
    }
}

// Projection: OrderSummaryReadModel
public class OrderSummaryProjection
{
    private readonly IReadModelStore _readStore;

    // Event handler - builds read model
    public async Task Handle(OrderPlacedEvent @event)
    {
        var summary = new OrderSummaryReadModel
        {
            OrderId = @event.OrderId,
            CustomerId = @event.CustomerId,
            Total = @event.Total,
            Status = "Placed",
            PlacedAt = @event.Timestamp,
            ItemCount = @event.Items.Count
        };

        await _readStore.Save(summary);
    }

    public async Task Handle(OrderShippedEvent @event)
    {
        var summary = await _readStore
            .Get<OrderSummaryReadModel>(@event.OrderId);

        summary.Status = "Shipped";
        summary.ShippedAt = @event.Timestamp;

        await _readStore.Update(summary);
    }
}

// Read Model - denormalized, optimized for queries
public class OrderSummaryReadModel
{
    public Guid OrderId { get; set; }
    public Guid CustomerId { get; set; }
    public string CustomerName { get; set; } // Denormalized
    public decimal Total { get; set; }
    public string Status { get; set; }
    public int ItemCount { get; set; }
    public DateTime PlacedAt { get; set; }
    public DateTime? ShippedAt { get; set; }
}

// Query - fast, no joins needed
public class OrderQueryService
{
    private readonly IReadModelStore _readStore;

    public async Task<OrderSummaryReadModel>
        GetOrderSummary(Guid orderId)
    {
        // Direct lookup, no aggregate replay
        return await _readStore
            .Get<OrderSummaryReadModel>(orderId);
    }

    public async Task<List<OrderSummaryReadModel>>
        GetCustomerOrders(Guid customerId)
    {
        // Optimized query, indexed read model
        return await _readStore
            .Query<OrderSummaryReadModel>()
            .Where(o => o.CustomerId == customerId)
            .OrderByDescending(o => o.PlacedAt)
            .ToListAsync();
    }
}

public class OrderSagaOrchestrator
{
    private readonly IEventBus _eventBus;
    private readonly ISagaStateStore _stateStore;

    public async Task StartSaga(OrderPlacedEvent @event)
    {
        var sagaState = new OrderSagaState
        {
            SagaId = Guid.NewGuid(),
            OrderId = @event.OrderId,
            CurrentStep = SagaStep.Started,
            StartedAt = DateTime.UtcNow
        };

        await _stateStore.Save(sagaState);

        // Step 1: Process Payment
        await _eventBus.Send(new ProcessPaymentCommand
        {
            OrderId = @event.OrderId,
            Amount = @event.Total,
            SagaId = sagaState.SagaId
        });
    }

    // Payment successful
    public async Task Handle(PaymentProcessedEvent @event)
    {
        var saga = await _stateStore.Get(@event.SagaId);
        saga.CurrentStep = SagaStep.PaymentProcessed;
        saga.PaymentId = @event.PaymentId;
        await _stateStore.Update(saga);

        // Step 2: Reserve Stock
        await _eventBus.Send(new ReserveStockCommand
        {
            OrderId = saga.OrderId,
            SagaId = saga.SagaId
        });
    }

    // Stock reserved
    public async Task Handle(StockReservedEvent @event)
    {
        var saga = await _stateStore.Get(@event.SagaId);
        saga.CurrentStep = SagaStep.StockReserved;
        await _stateStore.Update(saga);

        // Step 3: Ship Order
        await _eventBus.Send(new ShipOrderCommand
        {
            OrderId = saga.OrderId,
            SagaId = saga.SagaId
        });
    }

    // Order shipped - saga complete
    public async Task Handle(OrderShippedEvent @event)
    {
        var saga = await _stateStore.Get(@event.SagaId);
        saga.CurrentStep = SagaStep.Completed;
        saga.CompletedAt = DateTime.UtcNow;
        await _stateStore.Update(saga);
    }

    // FAILURE HANDLING - Compensating Transactions

    // Payment failed
    public async Task Handle(PaymentFailedEvent @event)
    {
        var saga = await _stateStore.Get(@event.SagaId);
        saga.CurrentStep = SagaStep.Failed;
        saga.FailureReason = "Payment failed";
        await _stateStore.Update(saga);

        // Cancel order (no compensation needed - nothing done yet)
        await _eventBus.Publish(new OrderCancelledEvent
        {
            OrderId = saga.OrderId,
            Reason = "Payment failed"
        });
    }

    // Stock reservation failed
    public async Task Handle(StockReservationFailedEvent @event)
    {
        var saga = await _stateStore.Get(@event.SagaId);
        saga.CurrentStep = SagaStep.Compensating;
        await _stateStore.Update(saga);

        // Compensating: Refund payment
        await _eventBus.Send(new RefundPaymentCommand
        {
            PaymentId = saga.PaymentId,
            Amount = saga.OrderTotal,
            Reason = "Stock unavailable"
        });
    }

    // Shipping failed
    public async Task Handle(ShippingFailedEvent @event)
    {
        var saga = await _stateStore.Get(@event.SagaId);
        saga.CurrentStep = SagaStep.Compensating;
        await _stateStore.Update(saga);

        // Compensating Step 1: Release stock
        await _eventBus.Send(new ReleaseStockCommand
        {
            OrderId = saga.OrderId
        });

        // Compensating Step 2: Refund payment
        await _eventBus.Send(new RefundPaymentCommand
        {
            PaymentId = saga.PaymentId,
            Amount = saga.OrderTotal,
            Reason = "Shipping failed"
        });
    }
}

// Saga State
public class OrderSagaState
{
    public Guid SagaId { get; set; }
    public Guid OrderId { get; set; }
    public SagaStep CurrentStep { get; set; }
    public Guid? PaymentId { get; set; }
    public decimal OrderTotal { get; set; }
    public DateTime StartedAt { get; set; }
    public DateTime? CompletedAt { get; set; }
    public string? FailureReason { get; set; }
}

public enum SagaStep
{
    Started,
    PaymentProcessed,
    StockReserved,
    Completed,
    Compensating,
    Failed
}