Event-Driven CQRS на C# с паттерном Outbox

public async Task HandleAsync(OrderPlacedEvent @event)
{
    // Проверка, не обработали ли мы уже это событие
    if (await _processedEventsRepository.IsProcessedAsync(@event.Id))
        return;
        
    // Логика обновления read-модели
    await _orderReadModelRepository.CreateAsync(
        new OrderReadModel(@event.OrderId, @event.Total));
        
    // Отметка события как обработанного
    await _processedEventsRepository.MarkAsProcessedAsync(@event.Id);
}

// Команда
public record PlaceOrderCommand(string CustomerId, List<OrderItem> Items);
 
// Обработчик команды
public class PlaceOrderCommandHandler : ICommandHandler<PlaceOrderCommand>
{
    public async Task HandleAsync(PlaceOrderCommand command)
    {
        // Бизнес-логика и валидация
        var order = new Order(Guid.NewGuid().ToString(), command.CustomerId, command.Items);
        
        // Сохранение в БД и публикация события в рамках одной транзакции
        await _orderRepository.SaveAsync(order);
        
        // Событие автоматически публикуется через outbox
    }
}

public class OrderPlacedEventHandler : IEventHandler<OrderPlacedEvent>
{
    public async Task HandleAsync(OrderPlacedEvent @event)
    {
        // Обновление read-модели
        // Отправка уведомлений
        // Другие действия
    }
}

public async Task PlaceOrderAsync(Order order)
{
    using var conn = new SqlConnection(_dbConn);
    await conn.OpenAsync();
    using var tx = conn.BeginTransaction();
    try
    {
        // Вставка данных заказа
        await conn.ExecuteAsync(
            "INSERT INTO Orders (Id, Total) VALUES (@Id, @Total)",
            new { order.Id, order.Total }, 
            tx);
 
        // Подготовка и вставка события
        var evt = order.ToEvent();
        await conn.ExecuteAsync(
            "INSERT INTO Outbox (Id, Type, Payload) VALUES (@Id, @Type, @Payload)",
            new {
                Id = Guid.NewGuid(),
                Type = evt.GetType().Name,
                Payload = JsonSerializer.Serialize(evt)
            }, 
            tx);
 
        // Фиксация транзакции
        tx.Commit();
    }
    catch (Exception ex)
    {
        // Откат транзакции в случае ошибки
        tx.Rollback();
        throw;
    }
}

public async Task ProcessOutboxAsync()
{
    using var conn = new SqlConnection(_dbConn);
    var events = await conn.QueryAsync<OutboxRecord>(
        "SELECT TOP 100 * FROM Outbox WHERE Processed = 0");
        
    foreach (var record in events)
    {
        var evt = DeserializeEvent(record.Type, record.Payload);
        await _publisher.PublishAsync(evt);
 
        await conn.ExecuteAsync(
            "UPDATE Outbox SET Processed = 1 WHERE Id = @Id", 
            new { record.Id });
    }
}

public async Task HandleOrderPlacedEvent(OrderPlacedEvent @event)
{
    // Проверяем, не обрабатывали ли мы уже это событие
    if (await _processedEventsStore.HasBeenProcessed(@event.EventId))
        return;
        
    // Выполняем действия с событием
    await _orderService.ProcessNewOrder(@event.OrderId, @event.Items);
    
    // Сохраняем ID события как обработанного
    await _processedEventsStore.MarkAsProcessed(@event.EventId);
}

public async Task CleanupOutboxAsync()
{
    using var conn = new SqlConnection(_dbConn);
    await conn.ExecuteAsync(
        "DELETE FROM Outbox WHERE Processed = 1 AND CreatedAt < @threshold",
        new { threshold = DateTime.UtcNow.AddDays(-7) });
}

public async Task ProcessOutboxAsync()
{
    using var conn = new SqlConnection(_dbConn);
    await conn.OpenAsync();
    using var transaction = conn.BeginTransaction();
    
    // Получаем и блокируем записи для обработки
    var events = await conn.QueryAsync<OutboxRecord>(
        @"SELECT TOP(20) * FROM Outbox WITH (UPDLOCK, READPAST)
          WHERE Processed = 0
          ORDER BY CreatedAt", 
        transaction: transaction);
    
    foreach (var record in events)
    {
        try
        {
            var evt = DeserializeEvent(record.EventType, record.EventData);
            await _publisher.PublishAsync(evt);
            
            await conn.ExecuteAsync(
                @"UPDATE Outbox 
                  SET Processed = 1, ProcessedAt = @now 
                  WHERE Id = @Id",
                new { now = DateTime.UtcNow, record.Id },
                transaction);
        }
        catch (Exception ex)
        {
            await conn.ExecuteAsync(
                @"UPDATE Outbox 
                  SET RetryCount = RetryCount + 1, Error = @error 
                  WHERE Id = @Id",
                new { error = ex.ToString(), record.Id },
                transaction);
        }
    }
    
    await transaction.CommitAsync();
}

TimeSpan GetNextRetryDelay(int retryCount)
{
    return TimeSpan.FromSeconds(Math.Pow(2, Math.Min(retryCount, 8)));
}

// Публикация пакета событий
await _publisher.PublishBatchAsync(events);
 
// Обновление статуса для всего пакета
await conn.ExecuteAsync(
    @"UPDATE Outbox 
      SET Processed = 1, ProcessedAt = @now 
      WHERE Id IN @Ids",
    new { now = DateTime.UtcNow, Ids = events.Select(e => e.Id).ToList() },
    transaction);

public static IServiceCollection AddMediatrServices(this IServiceCollection services)
{
    return services
        .AddMediatR(cfg => cfg.RegisterServicesFromAssembly(Assembly.GetExecutingAssembly()))
        .AddTransient(typeof(IPipelineBehavior<,>), typeof(ValidationBehavior<,>))
        .AddTransient(typeof(IPipelineBehavior<,>), typeof(LoggingBehavior<,>));
}

public class Order
{
    public string Id { get; }
    public decimal Total { get; }
    public List<OrderItem> Items { get; }
    public string CustomerId { get; }
    public OrderStatus Status { get; private set; }
 
    public Order(string id, decimal total, List<OrderItem> items, string customerId)
    {
        Id = id;
        Total = total;
        Items = items;
        CustomerId = customerId;
        Status = OrderStatus.Created;
    }
 
    public void MarkAsPaid() => Status = OrderStatus.Paid;
 
    public OrderPlacedEvent ToPlacedEvent() => new(Id, Total, CustomerId, Items.Select(i => i.ToDto()).ToList());
    public OrderPaidEvent ToPaidEvent() => new(Id, Total, CustomerId);
}
 
public record OrderItem(string ProductId, int Quantity, decimal Price)
{
    public OrderItemDto ToDto() => new(ProductId, Quantity, Price);
}
 
public record OrderItemDto(string ProductId, int Quantity, decimal Price);
public record OrderPlacedEvent(string OrderId, decimal Total, string CustomerId, List<OrderItemDto> Items);
public record OrderPaidEvent(string OrderId, decimal Total, string CustomerId);

public class AppDbContext : DbContext
{
    public DbSet<Order> Orders { get; set; }
    public DbSet<OutboxMessage> OutboxMessages { get; set; }
 
    public AppDbContext(DbContextOptions<AppDbContext> options) : base(options) { }
 
    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        modelBuilder.Entity<Order>(entity =>
        {
            entity.HasKey(e => e.Id);
            entity.Property(e => e.Total).HasColumnType("decimal(18,2)");
            entity.Property(e => e.Status).HasConversion<string>();
            entity.HasMany<OrderItem>().WithOne().HasForeignKey("OrderId");
        });
 
        modelBuilder.Entity<OrderItem>(entity =>
        {
            entity.HasKey(e => new { e.OrderId, e.ProductId });
            entity.Property(e => e.Price).HasColumnType("decimal(18,2)");
        });
 
        modelBuilder.Entity<OutboxMessage>(entity =>
        {
            entity.HasKey(e => e.Id);
            entity.Property(e => e.EventType).IsRequired();
            entity.Property(e => e.EventData).IsRequired();
            entity.Property(e => e.CreatedAt).HasDefaultValueSql("GETUTCDATE()");
            entity.HasIndex(e => e.Processed);
        });
    }
}
 
public class OutboxMessage
{
    public Guid Id { get; set; }
    public string EventType { get; set; } = null!;
    public string EventData { get; set; } = null!;
    public DateTime CreatedAt { get; set; }
    public bool Processed { get; set; }
    public DateTime? ProcessedAt { get; set; }
    public string? Error { get; set; }
    public int RetryCount { get; set; }
}

public class OrderService : IOrderService
{
    private readonly AppDbContext _dbContext;
    private readonly IEventSerializer _eventSerializer;
 
    public OrderService(AppDbContext dbContext, IEventSerializer eventSerializer)
    {
        _dbContext = dbContext;
        _eventSerializer = eventSerializer;
    }
 
    public async Task<string> PlaceOrderAsync(PlaceOrderCommand command)
    {
        // Создаем новый заказ
        var order = new Order(
            Guid.NewGuid().ToString(),
            command.Items.Sum(i => i.Price * i.Quantity),
            command.Items.Select(i => new OrderItem(i.ProductId, i.Quantity, i.Price)).ToList(),
            command.CustomerId
        );
 
        // Начинаем транзакцию
        using var transaction = await _dbContext.Database.BeginTransactionAsync();
        try
        {
            // Сохраняем заказ
            _dbContext.Orders.Add(order);
            
            // Создаем событие и сохраняем его в outbox
            var @event = order.ToPlacedEvent();
            _dbContext.OutboxMessages.Add(new OutboxMessage
            {
                Id = Guid.NewGuid(),
                EventType = @event.GetType().Name,
                EventData = _eventSerializer.Serialize(@event),
                CreatedAt = DateTime.UtcNow,
                Processed = false
            });
            
            await _dbContext.SaveChangesAsync();
            
            // Фиксируем транзакцию
            await transaction.CommitAsync();
            
            return order.Id;
        }
        catch
        {
            // В случае ошибки откатываем транзакцию
            await transaction.RollbackAsync();
            throw;
        }
    }
}

public class OutboxProcessor : BackgroundService
{
    private readonly IServiceProvider _serviceProvider;
    private readonly ILogger<OutboxProcessor> _logger;
    private readonly IEventSerializer _eventSerializer;
    private readonly IEventPublisher _eventPublisher;
    private readonly TimeSpan _pollingInterval = TimeSpan.FromSeconds(10);
 
    public OutboxProcessor(
        IServiceProvider serviceProvider,
        ILogger<OutboxProcessor> logger,
        IEventSerializer eventSerializer,
        IEventPublisher eventPublisher)
    {
        _serviceProvider = serviceProvider;
        _logger = logger;
        _eventSerializer = eventSerializer;
        _eventPublisher = eventPublisher;
    }
 
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("Outbox processor is starting.");
 
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await ProcessOutboxMessagesAsync(stoppingToken);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error occurred while processing outbox messages.");
            }
 
            await Task.Delay(_pollingInterval, stoppingToken);
        }
 
        _logger.LogInformation("Outbox processor is stopping.");
    }
 
    private async Task ProcessOutboxMessagesAsync(CancellationToken stoppingToken)
    {
        using var scope = _serviceProvider.CreateScope();
        var dbContext = scope.ServiceProvider.GetRequiredService<AppDbContext>();
 
        var messages = await dbContext.OutboxMessages
            .Where(m => !m.Processed)
            .OrderBy(m => m.CreatedAt)
            .Take(20)
            .ToListAsync(stoppingToken);
 
        if (!messages.Any())
            return;
 
        _logger.LogInformation("Found {Count} messages to process", messages.Count);
 
        foreach (var message in messages)
        {
            try
            {
                var eventType = Type.GetType($"YourNamespace.{message.EventType}");
                if (eventType == null)
                {
                    _logger.LogWarning("Unknown event type: {EventType}", message.EventType);
                    message.Processed = true;
                    message.ProcessedAt = DateTime.UtcNow;
                    message.Error = "Unknown event type";
                    continue;
                }
 
                var @event = _eventSerializer.Deserialize(message.EventData, eventType);
                await _eventPublisher.PublishAsync(@event, stoppingToken);
 
                message.Processed = true;
                message.ProcessedAt = DateTime.UtcNow;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error processing outbox message {MessageId}", message.Id);
                message.RetryCount++;
                message.Error = ex.ToString();
 
                // Если превышено количество попыток, помечаем как обработаное с ошибкой
                if (message.RetryCount >= 5)
                {
                    message.Processed = true;
                    message.ProcessedAt = DateTime.UtcNow;
                }
            }
        }
 
        await dbContext.SaveChangesAsync(stoppingToken);
    }
}

public interface IEventPublisher
{
    Task PublishAsync<TEvent>(TEvent @event, CancellationToken cancellationToken = default) 
        where TEvent : class;
    Task PublishBatchAsync<TEvent>(IEnumerable<TEvent> events, CancellationToken cancellationToken = default) 
        where TEvent : class;
}

public class AzureServiceBusPublisher : IEventPublisher
{
    private readonly ServiceBusClient _client;
    private readonly ILogger<AzureServiceBusPublisher> _logger;
    private readonly Dictionary<Type, ServiceBusSender> _senders = new();
 
    public AzureServiceBusPublisher(string connectionString, ILogger<AzureServiceBusPublisher> logger)
    {
        _client = new ServiceBusClient(connectionString);
        _logger = logger;
    }
 
    public async Task PublishAsync<TEvent>(TEvent @event, CancellationToken cancellationToken = default) 
        where TEvent : class
    {
        var sender = GetOrCreateSender<TEvent>();
        var eventType = typeof(TEvent).Name;
        
        var message = new ServiceBusMessage(JsonSerializer.SerializeToUtf8Bytes(@event))
        {
            MessageId = Guid.NewGuid().ToString(),
            Subject = eventType,
            ApplicationProperties =
            {
                ["EventType"] = eventType
            }
        };
 
        await sender.SendMessageAsync(message, cancellationToken);
        _logger.LogInformation("Published event {EventType} with ID {MessageId}", 
            eventType, message.MessageId);
    }
 
    public async Task PublishBatchAsync<TEvent>(IEnumerable<TEvent> events, CancellationToken cancellationToken = default) 
        where TEvent : class
    {
        var sender = GetOrCreateSender<TEvent>();
        var eventType = typeof(TEvent).Name;
        
        var messages = events.Select(e => new ServiceBusMessage(JsonSerializer.SerializeToUtf8Bytes(e))
        {
            MessageId = Guid.NewGuid().ToString(),
            Subject = eventType,
            ApplicationProperties =
            {
                ["EventType"] = eventType
            }
        }).ToList();
 
        await sender.SendMessagesAsync(messages, cancellationToken);
        _logger.LogInformation("Published {Count} events of type {EventType}", 
            messages.Count, eventType);
    }
 
    private ServiceBusSender GetOrCreateSender<TEvent>() where TEvent : class
    {
        var eventType = typeof(TEvent);
        if (!_senders.TryGetValue(eventType, out var sender))
        {
            var queueName = eventType.Name.ToLowerInvariant();
            sender = _client.CreateSender(queueName);
            _senders[eventType] = sender;
        }
        return sender;
    }
 
    public async ValueTask DisposeAsync()
    {
        foreach (var sender in _senders.Values)
        {
            await sender.DisposeAsync();
        }
        await _client.DisposeAsync();
    }
}

public interface IEventSerializer
{
    string Serialize(object @event);
    object Deserialize(string data, Type eventType);
    TEvent Deserialize<TEvent>(string data) where TEvent : class;
}
 
public class JsonEventSerializer : IEventSerializer
{
    private readonly JsonSerializerOptions _options;
 
    public JsonEventSerializer()
    {
        _options = new JsonSerializerOptions
        {
            PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
            WriteIndented = false
        };
    }
 
    public string Serialize(object @event) => 
        JsonSerializer.Serialize(@event, @event.GetType(), _options);
 
    public object Deserialize(string data, Type eventType) => 
        JsonSerializer.Deserialize(data, eventType, _options) 
            ?? throw new InvalidOperationException($"Failed to deserialize {eventType.Name}");
 
    public TEvent Deserialize<TEvent>(string data) where TEvent : class => 
        JsonSerializer.Deserialize<TEvent>(data, _options) 
            ?? throw new InvalidOperationException($"Failed to deserialize {typeof(TEvent).Name}");
}

public class OrderCreatedEventHandler : INotificationHandler<OrderPlacedEvent>
{
    private readonly IReadModelRepository<OrderReadModel> _repository;
    private readonly ILogger<OrderCreatedEventHandler> _logger;
    private readonly IProcessedEventsRepository _processedEvents;
 
    public OrderCreatedEventHandler(
        IReadModelRepository<OrderReadModel> repository,
        IProcessedEventsRepository processedEvents,
        ILogger<OrderCreatedEventHandler> logger)
    {
        _repository = repository;
        _processedEvents = processedEvents;
        _logger = logger;
    }
 
    public async Task Handle(OrderPlacedEvent notification, CancellationToken cancellationToken)
    {
        var eventId = notification.GetType().Name + "_" + notification.OrderId;
        
        // Проверяем, не обрабатывали ли мы уже это событие
        if (await _processedEvents.HasBeenProcessedAsync(eventId, cancellationToken))
        {
            _logger.LogInformation("Event {EventId} already processed", eventId);
            return;
        }
 
        try
        {
            // Создаем read-модель на основе события
            var readModel = new OrderReadModel
            {
                Id = notification.OrderId,
                CustomerId = notification.CustomerId,
                Total = notification.Total,
                Status = "Created",
                Items = notification.Items.Select(i => new OrderItemReadModel
                {
                    ProductId = i.ProductId,
                    Quantity = i.Quantity,
                    Price = i.Price
                }).ToList()
            };
 
            await _repository.CreateAsync(readModel, cancellationToken);
            
            // Отмечаем событие как обработанное
            await _processedEvents.MarkAsProcessedAsync(eventId, cancellationToken);
            
            _logger.LogInformation("Created read model for order {OrderId}", notification.OrderId);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error processing OrderPlacedEvent for order {OrderId}", 
                notification.OrderId);
            throw; // Позволяем механизму повторных попыток сработать
        }
    }
}

public interface IProcessedEventsRepository
{
    Task<bool> HasBeenProcessedAsync(string eventId, CancellationToken cancellationToken = default);
    Task MarkAsProcessedAsync(string eventId, CancellationToken cancellationToken = default);
}
 
public class ProcessedEventsRepository : IProcessedEventsRepository
{
    private readonly AppDbContext _dbContext;
 
    public ProcessedEventsRepository(AppDbContext dbContext)
    {
        _dbContext = dbContext;
    }
 
    public async Task<bool> HasBeenProcessedAsync(string eventId, CancellationToken cancellationToken = default)
    {
        return await _dbContext.ProcessedEvents
            .AnyAsync(e => e.EventId == eventId, cancellationToken);
    }
 
    public async Task MarkAsProcessedAsync(string eventId, CancellationToken cancellationToken = default)
    {
        _dbContext.ProcessedEvents.Add(new ProcessedEvent
        {
            EventId = eventId,
            ProcessedAt = DateTime.UtcNow
        });
        
        await _dbContext.SaveChangesAsync(cancellationToken);
    }
}

public interface IReadModelRepository<T> where T : class
{
    Task<T?> GetByIdAsync(string id, CancellationToken cancellationToken = default);
    Task<IEnumerable<T>> GetAllAsync(CancellationToken cancellationToken = default);
    Task CreateAsync(T entity, CancellationToken cancellationToken = default);
    Task UpdateAsync(T entity, CancellationToken cancellationToken = default);
}
 
public class OrderReadModelRepository : IReadModelRepository<OrderReadModel>
{
    private readonly ReadDbContext _dbContext;
 
    public OrderReadModelRepository(ReadDbContext dbContext)
    {
        _dbContext = dbContext;
    }
 
    public async Task<OrderReadModel?> GetByIdAsync(string id, CancellationToken cancellationToken = default)
    {
        return await _dbContext.OrderReadModels
            .Include(o => o.Items)
            .FirstOrDefaultAsync(o => o.Id == id, cancellationToken);
    }
 
    public async Task<IEnumerable<OrderReadModel>> GetAllAsync(CancellationToken cancellationToken = default)
    {
        return await _dbContext.OrderReadModels
            .Include(o => o.Items)
            .ToListAsync(cancellationToken);
    }
 
    public async Task CreateAsync(OrderReadModel entity, CancellationToken cancellationToken = default)
    {
        _dbContext.OrderReadModels.Add(entity);
        await _dbContext.SaveChangesAsync(cancellationToken);
    }
 
    public async Task UpdateAsync(OrderReadModel entity, CancellationToken cancellationToken = default)
    {
        _dbContext.OrderReadModels.Update(entity);
        await _dbContext.SaveChangesAsync(cancellationToken);
    }
}

var builder = WebApplication.CreateBuilder(args);
 
// Регистрация контекстов баз данных
builder.Services.AddDbContext<AppDbContext>(options =>
    options.UseSqlServer(builder.Configuration.GetConnectionString("WriteDatabase")));
    
builder.Services.AddDbContext<ReadDbContext>(options =>
    options.UseSqlServer(builder.Configuration.GetConnectionString("ReadDatabase")));
 
// Регистрация сервисов
builder.Services
    .AddScoped<IOrderService, OrderService>()
    .AddScoped<IEventSerializer, JsonEventSerializer>()
    .AddSingleton<IEventPublisher>(sp => 
        new AzureServiceBusPublisher(
            builder.Configuration.GetConnectionString("ServiceBus")!,
            sp.GetRequiredService<ILogger<AzureServiceBusPublisher>>()))
    .AddScoped<IProcessedEventsRepository, ProcessedEventsRepository>()
    .AddScoped<IReadModelRepository<OrderReadModel>, OrderReadModelRepository>();
 
// Регистрация MediatR
builder.Services.AddMediatrServices();
 
// Регистрация фоновых сервисов
builder.Services.AddHostedService<OutboxProcessor>();
 
var app = builder.Build();

public class SnapshotService<T> where T : class
{
    private readonly ISnapshotRepository<T> _repository;
    private readonly ILogger<SnapshotService<T>> _logger;
 
    public SnapshotService(ISnapshotRepository<T> repository, ILogger<SnapshotService<T>> logger)
    {
        _repository = repository;
        _logger = logger;
    }
 
    public async Task<T?> GetLatestSnapshotAsync(string aggregateId, CancellationToken cancellationToken = default)
    {
        return await _repository.GetLatestAsync(aggregateId, cancellationToken);
    }
 
    public async Task SaveSnapshotAsync(string aggregateId, T snapshot, long version, CancellationToken cancellationToken = default)
    {
        await _repository.SaveAsync(aggregateId, snapshot, version, cancellationToken);
        _logger.LogInformation("Saved snapshot for aggregate {AggregateId} at version {Version}", 
            aggregateId, version);
    }
}

[Fact]
public async Task PlaceOrder_ShouldCreateOrderAndStoreEvent_WhenCommandIsValid()
{
    // Arrange
    var mockDbContext = new Mock<AppDbContext>();
    var mockEventSerializer = new Mock<IEventSerializer>();
    
    var service = new OrderService(mockDbContext.Object, mockEventSerializer.Object);
    var command = new PlaceOrderCommand("customer123", new List<OrderItemDto> 
    {
        new("product1", 2, 10.0m)
    });
    
    // Act
    await service.PlaceOrderAsync(command);
    
    // Assert
    mockDbContext.Verify(x => x.Orders.Add(It.IsAny<Order>()), Times.Once);
    mockDbContext.Verify(x => x.OutboxMessages.Add(It.IsAny<OutboxMessage>()), Times.Once);
    mockDbContext.Verify(x => x.SaveChangesAsync(It.IsAny<CancellationToken>()), Times.Once);
}

public record OrderPlacedEventV2(
    string OrderId, 
    decimal Total, 
    string CustomerId, 
    List<OrderItemDto> Items,
    string? PromotionCode = null  // Новое поле в версии 2
);

object DeserializeEvent(string type, string payload) =>
    type switch
    {
        "OrderPlacedEvent" => JsonSerializer.Deserialize<OrderPlacedEvent>(payload)!,
        "OrderPlacedEventV2" => JsonSerializer.Deserialize<OrderPlacedEventV2>(payload)!,
        _ => throw new InvalidOperationException($"Unknown event type {type}")
    };

builder.Services.AddOpenTelemetry()
    .WithTracing(tracerProviderBuilder =>
        tracerProviderBuilder
            .AddSource("MyCompany.OrderService")
            .AddSqlClientInstrumentation()
            .AddAspNetCoreInstrumentation()
            .AddJaegerExporter());

public async Task CleanupOutboxAsync()
{
    const int batchSize = 1000;
    var threshold = DateTime.UtcNow.AddDays(-30); // Храним события 30 дней
    
    int deleted;
    do
    {
        using var conn = new SqlConnection(_dbConn);
        deleted = await conn.ExecuteAsync(
            @"WITH OldMessages AS (
                SELECT TOP(@batchSize) Id
                FROM Outbox
                WHERE Processed = 1 AND ProcessedAt < @threshold
            )
            DELETE FROM Outbox WHERE Id IN (SELECT Id FROM OldMessages)",
            new { batchSize, threshold });
            
        await Task.Delay(1000); // Пауза между пакетами
    } 
    while (deleted >= batchSize);
}

OrderSystem/
  ├─ OrderSystem.Domain/           # Доменная модель и события
  ├─ OrderSystem.Infrastructure/   # Реализация репозиториев, Outbox
  ├─ OrderSystem.WriteApi/         # API для обработки команд
  ├─ OrderSystem.ReadApi/          # API для запросов
  ├─ OrderSystem.OutboxProcessor/  # Сервис обработки Outbox
  ├─ OrderSystem.Common/           # Общие интерфейсы и утилиты
  └─ OrderSystem.Tests/            # Юнит-тесты и интеграционные тесты

// OrderSystem.Domain/Entities/Order.cs
public class Order
{
    public string Id { get; }
    public string CustomerId { get; }
    public decimal Total { get; private set; }
    public List<OrderItem> Items { get; }
    public OrderStatus Status { get; private set; }
    
    // Конструктор и методы, создающие события
    
    public OrderPlacedEvent ToPlacedEvent() => 
        new(Id, CustomerId, Total, Items.Select(i => i.ToDto()).ToList());
        
    public void UpdateStatus(OrderStatus status)
    {
        // Валидация перехода между статусами
        Status = status;
    }
}
 
// События
public record OrderPlacedEvent(string OrderId, string CustomerId, 
    decimal Total, List<OrderItemDto> Items);

// OrderSystem.Infrastructure/Services/OrderService.cs
public class OrderService : IOrderService
{
    private readonly AppDbContext _dbContext;
    
    public async Task<string> PlaceOrderAsync(PlaceOrderCommand command)
    {
        var order = new Order(Guid.NewGuid().ToString(), command.CustomerId, 
            command.Items.Select(i => new OrderItem(i.ProductId, i.Quantity, i.Price)).ToList());
            
        using var transaction = await _dbContext.Database.BeginTransactionAsync();
        try
        {
            _dbContext.Orders.Add(order);
            
            // Сохраняем событие в Outbox
            var @event = order.ToPlacedEvent();
            _dbContext.OutboxMessages.Add(new OutboxMessage
            {
                Id = Guid.NewGuid(),
                EventType = @event.GetType().Name,
                EventData = JsonSerializer.Serialize(@event),
                CreatedAt = DateTime.UtcNow
            });
            
            await _dbContext.SaveChangesAsync();
            await transaction.CommitAsync();
            
            return order.Id;
        }
        catch
        {
            await transaction.RollbackAsync();
            throw;
        }
    }
}

// Улучшеная версия OutboxMessage
public class OutboxMessage
{
    // ... обычные поля
    public string EventHash { get; set; } = null!;
    
    public static string CalculateHash(string eventData)
    {
        using var sha = SHA256.Create();
        var hash = sha.ComputeHash(Encoding.UTF8.GetBytes(eventData));
        return Convert.ToBase64String(hash);
    }
}

// OrderSystem.ReadApi/Controllers/OrdersController.cs
[ApiController]
[Route("api/orders")]
public class OrdersController : ControllerBase
{
    private readonly IOrderReadRepository _repository;
    
    [HttpGet("{id}")]
    public async Task<ActionResult<OrderReadDto>> GetOrder(string id)
    {
        var order = await _repository.GetByIdAsync(id);
        if (order == null)
            return NotFound();
            
        return new OrderReadDto(
            order.Id,
            order.CustomerId,
            order.Total,
            order.Status,
            order.Items.Select(i => new OrderItemReadDto(
                i.ProductId, i.Quantity, i.Price, i.Total)).ToList()
        );
    }
}

// OrderSystem.OutboxProcessor/Worker.cs
public class Worker : BackgroundService
{
    private readonly IServiceProvider _serviceProvider;
    private readonly ILogger<Worker> _logger;
    private readonly IConfiguration _configuration;
    private readonly TimeSpan _pollingInterval;
 
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            using var scope = _serviceProvider.CreateScope();
            var processor = scope.ServiceProvider.GetRequiredService<IOutboxProcessor>();
            
            try
            {
                var count = await processor.ProcessPendingMessagesAsync(stoppingToken);
                if (count > 0)
                    _logger.LogInformation("Processed {Count} outbox messages", count);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error processing outbox messages");
            }
            
            await Task.Delay(_pollingInterval, stoppingToken);
        }
    }
}

// OrderSystem.Infrastructure/Services/OutboxProcessor.cs
public async Task<int> ProcessPendingMessagesAsync(CancellationToken cancellationToken)
{
    var messages = await _dbContext.OutboxMessages
        .Where(m => !m.Processed && (m.RetryCount < _maxRetries) && 
               (m.NextRetryTime == null || m.NextRetryTime <= DateTime.UtcNow))
        .OrderBy(m => m.CreatedAt)
        .Take(20)
        .ToListAsync(cancellationToken);
        
    if (!messages.Any())
        return 0;
        
    int processed = 0;
    foreach (var message in messages)
    {
        try
        {
            // Публикация события
            await _eventPublisher.PublishAsync(message.EventType, message.EventData);
            
            message.Processed = true;
            message.ProcessedAt = DateTime.UtcNow;
            processed++;
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "Failed to process message {MessageId}", message.Id);
            message.Error = ex.Message;
            message.RetryCount++;
            message.NextRetryTime = DateTime.UtcNow.AddSeconds(Math.Pow(2, message.RetryCount));
        }
    }
    
    await _dbContext.SaveChangesAsync(cancellationToken);
    return processed;
}

public async Task ProcessBatchAsync(IReadOnlyCollection<OutboxMessage> messages)
{
    var tasks = messages.Select(async message =>
    {
        try
        {
            await _eventPublisher.PublishAsync(message.EventType, message.EventData);
            return (message, success: true, error: (string?)null);
        }
        catch (Exception ex)
        {
            return (message, success: false, error: ex.Message);
        }
    });
    
    var results = await Task.WhenAll(tasks);
    
    // Обновляем статусы сообщений в БД
    await UpdateMessageStatusesAsync(results);
}

public class OutboxMessage
{
    // ... другие поля
    public string? AggregateId { get; set; }
    public long AggregateVersion { get; set; }
}

// OrderSystem.OutboxProcessor/Metrics.cs
public class OutboxMetrics
{
    private readonly IMetricsRoot _metrics;
    private readonly Counter _processedCounter;
    private readonly Counter _failedCounter;
    private readonly Histogram _processingTime;
    
    public void RecordProcessed() => _processedCounter.Increment();
    public void RecordFailed() => _failedCounter.Increment();
    public void RecordProcessingTime(long milliseconds) => 
        _processingTime.Update(milliseconds);
}