Session 16: Event-Driven Architecture & Messaging

Mục tiêu: Master event types, message broker patterns, MassTransit framework, và idempotency.

1. Event Types

Domain Events:
  - Internal trong 1 bounded context
  - Raised bởi Aggregate Root
  - Dispatch trong cùng transaction (MediatR)
  - VD: OrderItemAdded, PriceChanged

Integration Events:
  - Cross bounded context / cross service
  - Publish qua message broker (RabbitMQ, Service Bus)
  - Eventually consistent
  - VD: OrderSubmittedIntegrationEvent

Event Notifications:
  - Lightweight, chỉ chứa ID
  - Consumer tự query lại nếu cần details
  - VD: { type: "OrderUpdated", orderId: "xxx" }

Domain vs Integration Events

csharp
// Domain Event: in-process, đồng bộ
public record OrderItemAdded(Guid OrderId, Guid ProductId, int Quantity) : IDomainEvent;

// Integration Event: cross-service, bất đồng bộ
public record OrderSubmittedIntegrationEvent : IIntegrationEvent
{
    public Guid OrderId { get; init; }
    public Guid CustomerId { get; init; }
    public decimal TotalAmount { get; init; }
    public List<OrderItemDto> Items { get; init; } = [];
    public DateTime OccurredOn { get; init; } = DateTime.UtcNow;
    public Guid EventId { get; init; } = Guid.NewGuid();
}

// Pattern: Domain Event → Integration Event
public class OrderSubmittedDomainEventHandler(IEventBus eventBus)
    : INotificationHandler<OrderSubmittedEvent>
{
    public async Task Handle(OrderSubmittedEvent domainEvent, CancellationToken ct)
    {
        // Convert domain event → integration event
        var integrationEvent = new OrderSubmittedIntegrationEvent
        {
            OrderId = domainEvent.OrderId,
            TotalAmount = domainEvent.Total.Amount,
            // ... map fields
        };

        await eventBus.PublishAsync(integrationEvent, ct);
    }
}

2. Message Broker Patterns

RabbitMQ vs Azure Service Bus vs Kafka

3. MassTransit Framework Deep Dive

Setup

csharp
// Install: MassTransit, MassTransit.RabbitMQ (hoặc .Azure.ServiceBus)

builder.Services.AddMassTransit(x =>
{
    // Auto-discover all consumers in assembly
    x.AddConsumers(typeof(Program).Assembly);

    // RabbitMQ transport
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("rabbitmq://localhost", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        // Auto-configure endpoints from registered consumers
        cfg.ConfigureEndpoints(context);
    });

    // HOẶC Azure Service Bus transport
    // x.UsingAzureServiceBus((context, cfg) =>
    // {
    //     cfg.Host(connectionString);
    //     cfg.ConfigureEndpoints(context);
    // });
});

Publishing Events

csharp
// Publisher
public class OrderService(IPublishEndpoint publishEndpoint)
{
    public async Task SubmitOrderAsync(Order order)
    {
        // Save to DB first...

        // Publish integration event
        await publishEndpoint.Publish(new OrderSubmitted
        {
            OrderId = order.Id,
            CustomerId = order.CustomerId,
            TotalAmount = order.Total.Amount,
            Items = order.Items.Select(i => new OrderItemMessage
            {
                ProductId = i.ProductId,
                Quantity = i.Quantity,
                UnitPrice = i.UnitPrice
            }).ToList()
        });
    }
}

Consuming Events

csharp
// Consumer
public class OrderSubmittedConsumer(
    ILogger<OrderSubmittedConsumer> logger,
    ShippingDbContext db)
    : IConsumer<OrderSubmitted>
{
    public async Task Consume(ConsumeContext<OrderSubmitted> context)
    {
        var message = context.Message;
        logger.LogInformation("Processing order {OrderId}", message.OrderId);

        var shipment = new Shipment
        {
            OrderId = message.OrderId,
            Status = ShipmentStatus.Pending,
            CreatedAt = DateTime.UtcNow
        };

        db.Shipments.Add(shipment);
        await db.SaveChangesAsync();

        // Optionally publish response event
        await context.Publish(new ShipmentCreated
        {
            ShipmentId = shipment.Id,
            OrderId = message.OrderId
        });
    }
}

// Consumer Definition (advanced config)
public class OrderSubmittedConsumerDefinition : ConsumerDefinition<OrderSubmittedConsumer>
{
    protected override void ConfigureConsumer(
        IReceiveEndpointConfigurator endpointConfigurator,
        IConsumerConfigurator<OrderSubmittedConsumer> consumerConfigurator,
        IRegistrationContext context)
    {
        endpointConfigurator.UseMessageRetry(r => r.Intervals(
            TimeSpan.FromSeconds(1),
            TimeSpan.FromSeconds(5),
            TimeSpan.FromSeconds(15)));

        endpointConfigurator.UseInMemoryOutbox(context); // Outbox pattern
        endpointConfigurator.ConcurrentMessageLimit = 10;
    }
}

Request/Response Pattern (RPC over message broker)

csharp
// Server: Consumer responds
public class CheckStockConsumer : IConsumer<CheckStockRequest>
{
    public async Task Consume(ConsumeContext<CheckStockRequest> context)
    {
        var available = await CheckInventory(context.Message.ProductId, context.Message.Quantity);

        await context.RespondAsync(new CheckStockResponse
        {
            IsAvailable = available,
            ProductId = context.Message.ProductId
        });
    }
}

// Client: Request and wait for response
public class OrderService(IRequestClient<CheckStockRequest> stockClient)
{
    public async Task<bool> CanFulfillOrder(Guid productId, int quantity)
    {
        var response = await stockClient.GetResponse<CheckStockResponse>(
            new CheckStockRequest { ProductId = productId, Quantity = quantity },
            timeout: TimeSpan.FromSeconds(10));

        return response.Message.IsAvailable;
    }
}

4. Idempotency & Message Deduplication

csharp
// Problem: message có thể được deliver > 1 lần (at-least-once delivery)
// Solution: Idempotent consumers

// Approach 1: Check if already processed
public class OrderSubmittedConsumer(AppDbContext db) : IConsumer<OrderSubmitted>
{
    public async Task Consume(ConsumeContext<OrderSubmitted> context)
    {
        var messageId = context.MessageId ?? Guid.NewGuid();

        // Check if already processed
        var exists = await db.ProcessedMessages
            .AnyAsync(m => m.MessageId == messageId);

        if (exists)
        {
            // Already processed, skip
            return;
        }

        // Process...
        var shipment = CreateShipment(context.Message);
        db.Shipments.Add(shipment);

        // Record processed message
        db.ProcessedMessages.Add(new ProcessedMessage
        {
            MessageId = messageId,
            ProcessedAt = DateTime.UtcNow,
            ConsumerType = nameof(OrderSubmittedConsumer)
        });

        await db.SaveChangesAsync(); // Atomic: shipment + processed record
    }
}

// Approach 2: Natural idempotency (design operations to be idempotent)
// ❌ Non-idempotent: balance += 100 (2 lần = 200!)
// ✅ Idempotent: SET balance = 1100 WHERE version = 5
// ✅ Idempotent: INSERT IF NOT EXISTS

5. Event Schema Evolution & Versioning

csharp
// Version 1
public record OrderSubmitted
{
    public Guid OrderId { get; init; }
    public decimal TotalAmount { get; init; }
}

// Version 2: thêm field (backward compatible)
public record OrderSubmitted
{
    public Guid OrderId { get; init; }
    public decimal TotalAmount { get; init; }
    public string? Currency { get; init; } // Nullable = backward compatible
    public string? CustomerEmail { get; init; }
}

// Version 3: breaking change → new event type
public record OrderSubmittedV3
{
    public Guid OrderId { get; init; }
    public MoneyDto Total { get; init; } // Changed from decimal to object
}

// Strategies:
// 1. Always add optional fields (nullable) → backward compatible
// 2. Never remove/rename fields
// 3. Breaking changes → new event type + version in name
// 4. Consumers handle both old and new versions during transition

Tổng Kết Session 16

Bài tập: Setup MassTransit với RabbitMQ (Docker), publish OrderSubmitted event từ Orders service, consume trong Shipping service. Implement idempotent consumer.