Koto.Messaging.Wolverine

Реализация messaging-абстракций Koto на Wolverine + Kafka.

Что внутри

Тип	Назначение
`WolverineIntegrationEventPublisher`	`IIntegrationEventPublisher` → `IMessageBus.PublishAsync`
`WolverineIntegrationCommandDispatcher`	`IIntegrationCommandDispatcher` → fire-and-forget + request/reply
`IntegrationEventConsumerBase<TEvent>`	Идемпотентная база Kafka-консьюмера с роутингом в DLQ
`IntegrationCommandConsumerBase<TCommand>`	База Kafka-консьюмера команд со структурированным логированием
`CorrelationIdMiddleware`	Прокидывает `Envelope.CorrelationId` Wolverine через `CorrelationContext`
`IdempotencyMiddleware`	Дедупликация на уровне policy (альтернатива проверке в базе консьюмера)
`IProcessedMessageStore`	Абстракция хранилища идемпотентности (in-memory по умолчанию, для продакшена подключается PostgreSQL)

Настройка

// Program.cs
builder.Services.AddKotoWolverine(opts =>
{
    opts.RequestReplyTimeout = TimeSpan.FromSeconds(10);
    opts.IdempotencyWindow = TimeSpan.FromHours(24);
});

builder.Host.UseWolverine(opts =>
{
    opts.UseKafka("localhost:9092").AutoProvisionTopics();

    // Роутинг исходящих событий в топики
    opts.PublishMessage<OrderPlacedEvent>().ToKafkaTopic("orders.order-placed");

    // Слушаем входящие события
    opts.ListenToKafkaTopic("payments.payment-processed").ProcessInline();

    // Прокидываем correlation ID во всех хендлерах
    opts.Policies.AddMiddleware<CorrelationIdMiddleware>();

    // Автоматическая доставка доменных событий из агрегатов EF Core в outbox
    opts.PublishDomainEventsFromEntityFrameworkCore<IHasDomainEvents, IDomainEvent>(
        e => e.DomainEvents);
});

Конвенция имён топиков Kafka

Топики событий: {service}.{event-type} → orders.order-placed
Consumer-группы: {consuming-service}.{event-type}-consumer

Реализация консьюмера

public sealed class PaymentProcessedConsumer
    : IntegrationEventConsumerBase<PaymentProcessedEvent>
{
    public PaymentProcessedConsumer(
        IProcessedMessageStore store,
        ILogger<IntegrationEventConsumerBase<PaymentProcessedEvent>> logger)
        : base(store, logger) { }

    protected override async Task ConsumeAsync(PaymentProcessedEvent @event, CancellationToken ct)
    {
        // доменная логика здесь
    }
}

Продакшен-хранилище идемпотентности

Замените InMemoryProcessedMessageStore на устойчивое хранилище:

builder.Services.AddScoped<IProcessedMessageStore, PostgresProcessedMessageStore>();