Consuming events

[ConsumeEvent] is the single, unified way to subscribe — you never pick a bus or a plane on the consumer. The same attribute consumes either a domain or an integration event; the event type's marker (IDomainEvent / IIntegrationEvent) decides which plane delivers it. Every consumer is a fan-out subscriber: the event bus is pub/sub-only and never returns a value to the publisher. So a consumer reads identically whether the event runs inline or after commit. (For the two-plane model itself, see the events overview.) For request/reply — one typed response from a single target — don't use the event bus; call the handler directly (see calling a handler from other code).

There are two ways to write the consumer. Prefer the handler form — it makes the consumer a first-class unit of business logic with the full decorator pipeline; the method form is a lightweight alternative for a small side effect on a service you already have.

Handler form (preferred)

A consumer is a class implementing IHandler<TEvent> (the sugar over IHandler<TEvent, Result<Unit>>) — its request type is the event — annotated with a class-level [ConsumeEvent]. The sugar returns the non-generic Result:

using Elarion.Abstractions;
using Elarion.Abstractions.Messaging;

[ConsumeEvent]
public sealed class SendInvoiceEmail(IEmailSender email) : IHandler<InvoiceCreated> {
    public async ValueTask<Result> HandleAsync(InvoiceCreated e, CancellationToken ct) {
        await email.SendAsync(e.ClientEmail, ct);
        return Result.Success();
    }
}

Because it is a handler, it runs through the full handler decorator pipeline — tracing, resilience, validation, cache-invalidation — exactly like a command or query handler, and it is discovered, registered, and feature-gated by its module like every other handler. The event bus simply becomes one more trigger for the same unit of business logic. Use [ConsumeEvent(Order = n)] to order fan-out subscribers; Order ascends, and equal-order consumers run in a stable, generator-determined sequence.

Every consumer is a fan-out subscriber

The event bus is pub/sub-only, so a handler-form consumer always returns Result<Unit> — use the IHandler<TEvent> sugar, or the explicit IHandler<TEvent, Result<Unit>>. Every matching consumer runs — any number of handler-form consumers may subscribe to one event (each registers keyed by its own identity, so they never shadow one another), interleaved with method-form consumers by Order. A failed Result has no return channel to the publisher, so it surfaces as an EventConsumerFailedException, which each backend handles per its plane (the in-memory domain bus aggregates and rethrows, failing the command; the in-memory integration pump logs and isolates; the outbox dispatcher lets it propagate to trigger a retry).

A consumer that returns a non-Unit Result<T> is rejected (ELEVT005) on either plane — the bus has no response channel. When you need one typed response from a single target, that is request/reply, not pub/sub: call the target handler directly instead of publishing an event (see calling a handler from other code).

The clean way to get this right is a transaction decorator with an AppliesTo predicate that matches commands and integration-event handlers but not domain-event handlers — the generator then never attaches it to a domain-event handler, and the integration handler keeps it because its request is an IIntegrationEvent. One decorator, attached precisely at compile time, covers commands, queries, domain events, and integration events with no per-handler pipeline tags. (If you prefer to be explicit, a named read-only pipeline attribute that omits the transaction and resilience decorators works too, since pipeline attributes are most-specific-wins — see decorator pipelines.)

// Domain-event handler: shares the publisher's DbContext and transaction, so it must NOT
// open a nested transaction. With an AppliesTo predicate it needs no pipeline annotation at all.
[ConsumeEvent]
public sealed class RecalculateTotals(AppDbContext db) : IHandler<InvoiceLineAdded> {
    public async ValueTask<Result> HandleAsync(InvoiceLineAdded e, CancellationToken ct) {
        // Shares the publisher's DbContext; this write commits atomically with the command.
        await db.Invoices
            .Where(i => i.Id == e.InvoiceId)
            .ExecuteUpdateAsync(s => s.SetProperty(i => i.Total, i => i.Total + e.Amount), ct);
        return Result.Success();
    }
}

Method form (alternative)

When you just need a side effect on a service that already exists, annotate an instance method on a [Service] class instead — no dedicated handler class, but also no decorator pipeline. The consumed event type is the method's message parameter, and the plane comes from that type's marker:

[Service]
public sealed class InvoiceNotifications(ILogger<InvoiceNotifications> logger) {
    [ConsumeEvent]
    public async ValueTask OnInvoiceCreated(InvoiceCreated e, CancellationToken ct) {
        await NotifyAsync(e.ClientEmail, ct);
    }
}

Here too the method is a fan-out subscriber — it returns void / Task / ValueTask (throw to fail) or the non-generic Result / Task<Result> / ValueTask<Result> (a failed Result → EventConsumerFailedException, the same failure channel as the handler form). Every matching consumer runs. The event bus is pub/sub-only, so a method that returns a Result<TResponse> with a value is rejected (ELEVT002); when you need one typed response, call the target handler directly instead of consuming an event (see calling a handler from other code).

Beyond the message parameter, a method-form consumer may optionally declare — in any order — an IEventContext (or IEventContext<TEvent>) for the correlation id, the durable message id (MessageId, the dedup key on the integration plane), and the message, plus a CancellationToken. Both are supplied by the runtime; omit what you don't need, and use [ConsumeEvent(Order = n)] to order fan-out subscribers. (The handler form takes the event and an optional CancellationToken.)

The durable consumer identity is {service}.{method}({event type}). Runtime-supplied context and CancellationToken parameters are intentionally excluded, so adding or removing either does not orphan pending outbox deliveries or inbox claims. Two overloads on the same service with the same method name and event type would share that identity and are rejected by ELEVT006; rename or split one consumer.

A class-level [ConsumeEvent] on a non-handler class, or a method-form consumer not on a [Service], is reported (ELEVT001/ELEVT005); an invalid signature is ELEVT002.

Handling duplicates

Integration delivery is at-least-once: a consumer can see the same message more than once — a worker crash after the consumer ran but before its delivery was finalized, or a lease that expired under multiple instances. Outbox retries are independent per consumer, so a failed sibling no longer re-runs completed siblings. There is no exactly-once delivery; exactly-once effect comes from deduplication on the consumer side. Every redelivery carries the same IEventContext.MessageId (the durable message identity — for the outbox, the row's id), so that id is the deduplication key. (CorrelationId is a tracing identifier — don't key dedup on it.)

Handler-form consumers get this for free: the inbox is on by default. Every handler-form consumer whose event is an IIntegrationEvent is wrapped in an inbox decorator that claims the pair (consumer, message id) in the consumer's own transaction — replacing the plain transaction decorator — so the claim commits atomically with the consumer's business writes. A redelivery finds the committed claim and is acknowledged as already-done without re-running the consumer; a failed Result rolls the claim back with the writes, so the message retries. When two workers race the same message, the loser waits on the winner's uncommitted claim and then replays it — only a committed claim is ever acknowledged.

Why this must live in your transaction rather than the outbox row: the outbox row has the wrong grain. It is one row per event, finalized by the delivery worker per message, on the worker's own connection, after your consumer's transaction has already committed — and only once every fan-out consumer has run. Your effect is a fact per (message, consumer), and only a write inside your consumer's own transaction can atomically couple "already processed" to the effect itself. The inbox is exactly that write, done for you.

Two things to know about the default:

• Durability follows the idempotency store. The delivery tiers register the in-memory store automatically (process-local dedup); pair the outbox with AddElarionIdempotencyEntityFrameworkCore so inbox claims survive restarts — the outbox is durable, so its inbox should be too.
• Retention outlives retries. Inbox rows expire like idempotency keys after 24 h — comfortably above the outbox's maximum retry window (≈43 minutes at the defaults). The margin matters: a still-retrying message must never re-run because its claim was purged.

The inbox dedups the consumer's transactional effect. A foreign side effect inside the consumer — an email, a payment API call — re-runs only in the narrow window where the process dies between the foreign call and the commit. To close that last window, pass the message id as the downstream's idempotency key and let it collapse duplicates too. A method-form consumer reads it from its IEventContext parameter; a handler-form consumer (which takes only the event) injects IIdempotencyKeyAccessor — in a delivery scope its key is the seeded message id:

[ConsumeEvent]
public sealed class ChargeInvoice(IPaymentGateway payments, IIdempotencyKeyAccessor messageId)
    : IHandler<InvoiceCreated> {
    public async ValueTask<Result> HandleAsync(InvoiceCreated e, CancellationToken ct) {
        // The same id the inbox claims — stable across redeliveries, unique per message.
        messageId.TryGetKey(out var key);
        await payments.ChargeAsync(e.InvoiceId, idempotencyKey: key!, ct);
        return Result.Success();
    }
}

Opting out. [AllowDuplicates] declares that redelivery is harmless to this consumer and removes the inbox (the plain transaction decorator returns) — the consumer-side mirror of [AllowAnonymous] switching off a default guard. Declare it when the effect is naturally idempotent, or when the only effect is a call to a downstream keyed on the message id:

[ConsumeEvent]
[AllowDuplicates] // "mark paid only if not already paid" converges by itself
public sealed class MarkInvoicePaid(AppDbContext db) : IHandler<InvoicePaid> {
    public async ValueTask<Result> HandleAsync(InvoicePaid e, CancellationToken ct) {
        await db.Invoices
            .Where(i => i.Id == e.InvoiceId && i.Status != InvoiceStatus.Paid)
            .ExecuteUpdateAsync(s => s.SetProperty(i => i.Status, InvoiceStatus.Paid), ct);
        return Result.Success();
    }
}

Which one does my consumer need?

The default is already correct — [AllowDuplicates] is an optimization you may claim when you can say out loud why re-running is harmless, never something correctness requires. By effect shape:

A worked example: invoice creation, end to end

The classic CRUD-with-side-effects shape — create a row, generate a PDF, send an email, keep another module's statistics current — maps onto the pieces like this:

// The command — natural key first: is there a unique constraint that makes a second insert impossible?
// Here there isn't (nothing makes two invoices "the same"), so escalate to [Idempotent]: a client retry
// must return the FIRST invoice, not create a second. The client sends an Idempotency-Key.
// Bonus: a replayed command never re-runs, so InvoiceCreated is never published twice — the
// entire downstream chain is protected at its source.
[Idempotent]
public sealed class CreateInvoiceHandler(AppDbContext db, IIntegrationEventBus events)
    : IHandler<CreateInvoiceCommand, Result<InvoiceResponse>> { /* insert + PublishAsync + save */ }

// Default inbox (do nothing): the PDF blob, its link on the invoice, and the inbox claim commit
// atomically in the consumer's transaction. A redelivery replays instead of generating a second PDF.
[ConsumeEvent]
public sealed class GenerateInvoicePdf(AppDbContext db, IBlobStore blobs) : IHandler<InvoiceCreated> { … }

// Default inbox (do nothing): SMTP takes no dedup key, so keep the inbox — duplicates narrow to the
// crash window between the send and the commit, an accepted residual for email. With a keyed provider
// (Resend, Brevo, …) instead: declare [AllowDuplicates] and pass the message id as the provider's
// idempotency key — the recipient dedups better than the inbox can.
[ConsumeEvent]
public sealed class SendInvoiceEmail(IEmailSender email) : IHandler<InvoiceCreated> { … }

// [AllowDuplicates]: recomputes from the source table, so re-running converges on the same numbers.
// If this were incremental ("stats.Total += e.Amount") it would NOT converge — keep the default inbox.
[ConsumeEvent]
[AllowDuplicates]
public sealed class RecalculateStatistics(StatsDbContext db) : IHandler<InvoiceCreated> { … }

The rule of thumb that falls out: defaults need no justification; deviations must name their reason. Write the consumer and ship it — the default inbox is safe; declare [AllowDuplicates] only when you can say why re-running is harmless. On the command side, model a natural key + unique constraint first, and escalate to [Idempotent] only for a named reason — no natural key, replay-beats-rejection, money, or increments (see when does a command need it).

Method-form consumers have no pipeline, so no inbox. Convert to the handler form when dedup matters, or dedup by hand keyed on (consumer, context.MessageId) — a unique-constrained insert in the same transaction as the side effect, which is precisely what the inbox automates:

[ConsumeEvent]
public async ValueTask OnInvoiceCreated(
    InvoiceCreated e, IEventContext context, CancellationToken ct) {
    db.ProcessedEvents.Add(new ProcessedEvent {
        Consumer = nameof(InvoiceNotifications),
        MessageId = context.MessageId!.Value,   // unique index on (Consumer, MessageId)
    });
    DoTheSideEffect(e);
    try {
        await db.SaveChangesAsync(ct);
    }
    catch (DbUpdateException) {
        // The unique constraint rejected a duplicate — already processed, so let the message finalize.
    }
}

Subscribing without the generator (advanced)

[ConsumeEvent] is the normal path, but the runtime ultimately subscribes to whatever EventSubscriptionDescriptors are registered in DI — the generator simply emits those registrations. The descriptor type is public (Elarion.Abstractions.Messaging), so you can register one by hand for tests or advanced host wiring. It is the same unified shape for both planes: Plane selects domain vs integration, and InvokeAsync is the fan-out subscriber callback (the bus is pub/sub-only, so every descriptor is a fan-out subscriber):

using Elarion.Abstractions.Messaging;
using Microsoft.Extensions.DependencyInjection;

builder.Services.AddScoped<InvoiceNotifications>();
builder.Services.AddSingleton(new EventSubscriptionDescriptor {
    EventType   = typeof(InvoiceCreated),
    Plane       = EventPlane.Integration,          // or EventPlane.Domain — same descriptor, either plane
    ServiceType = typeof(InvoiceNotifications),
    InvokeAsync = static (sp, evt, ctx, ct) =>
        sp.GetRequiredService<InvoiceNotifications>()
          .OnInvoiceCreated((InvoiceCreated)evt, ct),
});

This is composition-time registration. The EventSubscriptionRegistry is built once from the registered descriptors and then frozen, so the set of consumers is fixed when the container is built. A hand-registered descriptor is also always active — module feature-gating happens at the generated Add{Module}EventConsumers call site, not in the descriptor itself.